subuhui/crpto_predictor
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: timezone handling and prediction logic

Browse Source 
- api/models.py: Fix timezone conversion in stats queries by calculating UTC ranges in Python instead of using SQL CONVERT_TZ. Ensure JSON serializable types.
- api/prediction_service.py:
    - Update ensure_predictions_for_month to check previous day (ET) to cover CST month start.
    - Update _generate_prediction_for_hour to skip saving unclosed candles (current hour).
- trader/auto_trade_polymarket.py: Update to use hourly_model_predictions (V2) table for storing live trading predictions.
This commit is contained in:
crypto_predictor committed 2026-02-11 23:14:50 +08:00
1 parent 2993014fa8
commit 99fa79cde4
4 files changed
+232 -111
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

api/models.py
+142 -57
View File
@@ -4,7 +4,7 @@
import pymysql
import os
from typing import Optional, List, Dict
from datetime import datetime
from datetime import datetime, timedelta
from contextlib import contextmanager
from zoneinfo import ZoneInfo
from dotenv import load_dotenv
@@ -340,7 +340,29 @@ class PredictionModel:
def get_monthly_stats_v2(month: str) -> Dict:
"""
获取月度统计 (V2 - 从宽表)
修改：不使用 SQL 进行时区转换和格式化，而是查询 UTC 时间范围，然后在 Python 中处理
"""
# 计算 UTC 时间范围
try:
year, mon = map(int, month.split('-'))
start_date_cst = datetime(year, mon, 1, tzinfo=TZ_CST)
# 下个月第一天
if mon == 12:
end_date_cst = datetime(year + 1, 1, 1, tzinfo=TZ_CST)
else:
end_date_cst = datetime(year, mon + 1, 1, tzinfo=TZ_CST)
# 转换为 UTC
start_dt_utc = start_date_cst.astimezone(TZ_UTC)
end_dt_utc = end_date_cst.astimezone(TZ_UTC)
start_str = start_dt_utc.strftime('%Y-%m-%d %H:%M:%S')
end_str = end_dt_utc.strftime('%Y-%m-%d %H:%M:%S')
except Exception as e:
print(f"日期解析错误: {e}")
return {}
with get_db_connection() as conn:
cursor = conn.cursor()
@@ -352,25 +374,26 @@ class PredictionModel:
col_pred = f"{model}_prediction"
col_correct = f"{model}_correct"
# 查询 UTC 时间范围内的所有数据
sql = f"""
SELECT
COUNT({col_pred}) as total,
SUM(CASE WHEN {col_correct} = 1 THEN 1 ELSE 0 END) as correct,
SUM(CASE WHEN {col_correct} = 0 THEN 1 ELSE 0 END) as incorrect
FROM hourly_model_predictions
WHERE DATE_FORMAT(CONVERT_TZ(timestamp, '+00:00', '+08:00'), '%%Y-%%m') = %s
WHERE timestamp >= %s AND timestamp < %s
AND actual_result IS NOT NULL
"""
cursor.execute(sql, (month,))
cursor.execute(sql, (start_str, end_str))
row = cursor.fetchone()
if row:
total = row['total']
correct = row['correct'] or 0
incorrect = row['incorrect'] or 0
total = int(row['total'])
correct = int(row['correct'] or 0)
incorrect = int(row['incorrect'] or 0)
accuracy = (correct / total * 100) if total > 0 else 0
# 格式化键名以匹配前端 (例如 dlinear -> DLinear)
# 格式化键名以匹配前端
model_key = {
'dlinear': 'DLinear',
'nbeats': 'N-BEATS',
@@ -391,55 +414,97 @@ class PredictionModel:
def get_daily_stats_v2(month: str, page: int = 1, limit: int = 10) -> Dict:
"""
获取每日统计 (V2 - 从宽表)
修改：不使用 SQL 进行时区转换，查询 UTC 范围，然后在 Python 中转换
"""
# 计算 UTC 时间范围
try:
year, mon = map(int, month.split('-'))
start_date_cst = datetime(year, mon, 1, tzinfo=TZ_CST)
if mon == 12:
end_date_cst = datetime(year + 1, 1, 1, tzinfo=TZ_CST)
else:
end_date_cst = datetime(year, mon + 1, 1, tzinfo=TZ_CST)
start_dt_utc = start_date_cst.astimezone(TZ_UTC)
end_dt_utc = end_date_cst.astimezone(TZ_UTC)
start_str = start_dt_utc.strftime('%Y-%m-%d %H:%M:%S')
end_str = end_dt_utc.strftime('%Y-%m-%d %H:%M:%S')
except Exception:
return {"data": [], "page": page, "limit": limit, "total": 0, "total_pages": 0}
with get_db_connection() as conn:
cursor = conn.cursor()
# 分页查询日期
offset = (page - 1) * limit
date_sql = """
SELECT DISTINCT DATE(CONVERT_TZ(timestamp, '+00:00', '+08:00')) as date_cst
# 获取范围内的所有原始数据
# 我们需要 timestamp, actual_result, 和各模型的 prediction/correct
# 为了按 CST 日期分组，我们需要取出每一行，然后在 Python 中转换时间
sql = """
SELECT
timestamp,
actual_result,
dlinear_prediction, dlinear_correct,
nbeats_prediction, nbeats_correct,
ttm_prediction, ttm_correct,
ensemble_prediction, ensemble_correct
FROM hourly_model_predictions
WHERE DATE_FORMAT(CONVERT_TZ(timestamp, '+00:00', '+08:00'), '%%Y-%%m') = %s
ORDER BY date_cst DESC
LIMIT %s OFFSET %s
WHERE timestamp >= %s AND timestamp < %s
AND actual_result IS NOT NULL
ORDER BY timestamp DESC
"""
cursor.execute(date_sql, (month, limit, offset))
dates = [row['date_cst'] for row in cursor.fetchall()]
cursor.execute(sql, (start_str, end_str))
rows = cursor.fetchall()
# 总天数
count_sql = """
SELECT COUNT(DISTINCT DATE(CONVERT_TZ(timestamp, '+00:00', '+08:00'))) as cnt
FROM hourly_model_predictions
WHERE DATE_FORMAT(CONVERT_TZ(timestamp, '+00:00', '+08:00'), '%%Y-%%m') = %s
"""
cursor.execute(count_sql, (month,))
total_count = cursor.fetchone()['cnt']
# 在 Python 中处理分组
daily_stats = {} # { "YYYY-MM-DD": { "date": "...", "dlinear": {...}, ... } }
daily_data = []
models = ['dlinear', 'nbeats', 'ttm', 'ensemble']
for date_obj in dates:
date_str = str(date_obj)
day_stats = {"date": date_str}
for row in rows:
# 转换 timestamp 到 CST
ts_utc = row['timestamp']
# 确保是带时区的 datetime
if isinstance(ts_utc, str):
ts_utc = datetime.fromisoformat(str(ts_utc).replace('Z', '+00:00'))
if ts_utc.tzinfo is None:
ts_utc = ts_utc.replace(tzinfo=TZ_UTC)
ts_cst = ts_utc.astimezone(TZ_CST)
date_str = ts_cst.strftime('%Y-%m-%d')
if date_str not in daily_stats:
daily_stats[date_str] = {
"date": date_str,
"raw_data": {m: {"total": 0, "correct": 0} for m in ['dlinear', 'nbeats', 'ttm', 'ensemble']}
}
# 累加统计
models = ['dlinear', 'nbeats', 'ttm', 'ensemble']
for model in models:
col_pred = f"{model}_prediction"
col_correct = f"{model}_correct"
pred = row[f"{model}_prediction"]
correct_flag = row[f"{model}_correct"]
sql = f"""
SELECT
COUNT({col_pred}) as total,
SUM(CASE WHEN {col_correct} = 1 THEN 1 ELSE 0 END) as correct
FROM hourly_model_predictions
WHERE DATE(CONVERT_TZ(timestamp, '+00:00', '+08:00')) = %s
AND actual_result IS NOT NULL
"""
cursor.execute(sql, (date_str,))
row = cursor.fetchone()
total = row['total']
correct = row['correct'] or 0
if pred is not None:
daily_stats[date_str]['raw_data'][model]['total'] += 1
if correct_flag == 1:
daily_stats[date_str]['raw_data'][model]['correct'] += 1
# 整理为最终格式
sorted_dates = sorted(daily_stats.keys(), reverse=True)
total_count = len(sorted_dates)
# 分页
offset = (page - 1) * limit
paginated_dates = sorted_dates[offset:offset + limit]
result_data = []
for date_key in paginated_dates:
day_item = daily_stats[date_key]
formatted_item = {"date": day_item["date"]}
for model in ['dlinear', 'nbeats', 'ttm', 'ensemble']:
raw = day_item['raw_data'][model]
total = int(raw['total'])
correct = int(raw['correct'])
accuracy = (correct / total * 100) if total > 0 else 0
model_key = {
@@ -449,16 +514,15 @@ class PredictionModel:
'ensemble': 'Ensemble'
}.get(model, model)
day_stats[model_key] = {
formatted_item[model_key] = {
"total": total,
"correct": correct,
"accuracy": round(accuracy, 2)
}
daily_data.append(day_stats)
result_data.append(formatted_item)
return {
"data": daily_data,
"data": result_data,
"page": page,
"limit": limit,
"total": total_count,
@@ -469,25 +533,46 @@ class PredictionModel:
def get_hourly_details_v2(date: str) -> List[Dict]:
"""
获取指定日期的每小时详情 (V2)
修改：根据传入的 CST 日期计算 UTC 范围，查询该范围内的数据
"""
try:
# date 是 'YYYY-MM-DD' (CST)
date_cst_start = datetime.strptime(date, '%Y-%m-%d').replace(tzinfo=TZ_CST)
date_cst_end = date_cst_start + timedelta(days=1)
start_dt_utc = date_cst_start.astimezone(TZ_UTC)
end_dt_utc = date_cst_end.astimezone(TZ_UTC)
start_str = start_dt_utc.strftime('%Y-%m-%d %H:%M:%S')
end_str = end_dt_utc.strftime('%Y-%m-%d %H:%M:%S')
except Exception as e:
print(f"日期处理错误: {e}")
return []
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT
*,
HOUR(CONVERT_TZ(timestamp, '+00:00', '+08:00')) as hour_cst,
CONVERT_TZ(timestamp, '+00:00', '+08:00') as timestamp_cst_dt
SELECT *
FROM hourly_model_predictions
WHERE DATE(CONVERT_TZ(timestamp, '+00:00', '+08:00')) = %s
ORDER BY hour_cst
""", (date,))
WHERE timestamp >= %s AND timestamp < %s
ORDER BY timestamp ASC
""", (start_str, end_str))
rows = cursor.fetchall()
result = []
for row in rows:
# 计算 CST 小时用于显示排序 (虽然 SQL 已经按 timestamp 排序了，但为了确保前端显示 hour 字段正确)
ts_utc = row['timestamp']
if isinstance(ts_utc, str):
ts_utc = datetime.fromisoformat(str(ts_utc).replace('Z', '+00:00'))
if ts_utc.tzinfo is None:
ts_utc = ts_utc.replace(tzinfo=TZ_UTC)
ts_cst = ts_utc.astimezone(TZ_CST)
hour_cst = ts_cst.hour
hour_data = {
"hour": row['hour_cst'],
"hour": hour_cst,
"timestamp_et": row['timestamp_et'],
"timestamp_cst": convert_to_cst(row.get('timestamp', '')),
"open_price": row['open_price'],
api/prediction_service.py
+23 -1
View File
@@ -226,6 +226,17 @@ class PredictionService:
preds = self.inference_manager.predict_all(df_input)
# 7. 判断实际涨跌
# Check if the candle is closed (current time > candle open time + 1 hour)
current_now_utc = datetime.now(ZoneInfo('UTC')).replace(tzinfo=None)
candle_close_time = target_time_utc + timedelta(hours=1)
is_closed = current_now_utc >= candle_close_time
if not is_closed:
# 仅保存已完成的K线预测，未完成的跳过 (等待完成后再生成)
# 实时交易逻辑在 auto_trade_polymarket.py 中单独处理
return
actual_result = 1 if target_close > prev_close else 0
# 8. 组装数据 (宽表格式)
@@ -266,7 +277,11 @@ class PredictionService:
def ensure_predictions_for_month(self, month: str) -> None:
"""
确保指定月份有预测数据
确保指定月份有预测数据 (CST)
需要检查:
1. 该月的所有日期 (ET)
2. 上个月的最后一天 (ET) -> 因为 ET 上个月最后一天 11:00-23:00 对应 CST 该月第一天 00:00-12:00
Args:
month: 月份 (YYYY-MM)
"""
@@ -281,8 +296,15 @@ class PredictionService:
from calendar import monthrange
_, last_day = monthrange(year, mon)
# 1. 确保上个月最后一天有数据 (为了 CST 的开头)
first_day_of_month = datetime(year, mon, 1).date()
prev_day = first_day_of_month - timedelta(days=1)
prev_day_str = prev_day.strftime('%Y-%m-%d')
self.ensure_predictions_for_date(prev_day_str)
today = datetime.now(self.et_tz).date()
# 2. 确保本月所有日期
for day in range(1, last_day + 1):
date = f"{year}-{mon:02d}-{day:02d}"
date_obj = datetime.strptime(date, '%Y-%m-%d').date()
trader/auto_trade_polymarket.py
+58 -53
View File
@@ -541,41 +541,43 @@ class PolymarketBTCTrader:
final_probability = all_preds['dlinear']['probability']
logger.warning("⚠️ 集成预测不可用，降级为 DLinear")
# 保存所有预测结果到 model_predictions 表
# 保存所有预测结果到 hourly_model_predictions 表 (V2)
try:
et_tz = ZoneInfo('America/New_York')
current_time_utc = pd.Timestamp(current_time)
current_time_et = current_time_utc.tz_localize('UTC').tz_convert(et_tz)
common_data = {
prediction_data = {
"timestamp": current_time_utc.isoformat(),
"timestamp_et": current_time_et.isoformat(),
"date": current_time_et.strftime('%Y-%m-%d'),
"hour": current_time_et.hour,
"actual_result": None,
"is_correct": None,
"open_price": float(df['open'].iloc[-1]),
"close_price": None,
"actual_result": None,
# DLinear
"dlinear_prediction": all_preds.get('dlinear', {}).get('prediction'),
"dlinear_prob": all_preds.get('dlinear', {}).get('probability'),
"dlinear_correct": None,
# N-BEATS
"nbeats_prediction": all_preds.get('nbeats', {}).get('prediction'),
"nbeats_prob": all_preds.get('nbeats', {}).get('probability'),
"nbeats_correct": None,
# TTM-R2
"ttm_prediction": all_preds.get('ttm', {}).get('prediction'),
"ttm_prob": all_preds.get('ttm', {}).get('probability'),
"ttm_correct": None,
# Ensemble
"ensemble_prediction": all_preds.get('ensemble', {}).get('prediction'),
"ensemble_votes": all_preds.get('ensemble', {}).get('votes'),
"ensemble_correct": None,
}
# 辅助函数
def save_single_pred(name, pred_res):
data = common_data.copy()
data['model_name'] = name
data['prediction'] = pred_res['prediction']
data['probability'] = pred_res.get('probability', 0.5)
PredictionModel.save_prediction(data)
if 'dlinear' in all_preds:
save_single_pred("DLinear", all_preds['dlinear'])
if 'nbeats' in all_preds:
save_single_pred("N-BEATS", all_preds['nbeats'])
if 'ttm' in all_preds:
save_single_pred("TTM-R2", all_preds['ttm'])
if 'ensemble' in all_preds and all_preds['ensemble']:
# 集成也保存
save_single_pred("Ensemble", all_preds['ensemble'])
PredictionModel.save_hourly_prediction(prediction_data)
logger.info(f"💾 所有模型预测记录已保存 (ET {current_time_et.strftime('%Y-%m-%d %H:%M')})")
except Exception as e:
logger.error(f"保存预测记录失败: {e}")
@@ -599,22 +601,17 @@ class PolymarketBTCTrader:
prev_close = float(prev_row['close'])
actual_result = 1 if prev_close > prev_open else 0
models_to_update = ["DLinear", "N-BEATS", "TTM-R2", "Ensemble"]
# 使用 V2 更新方法 (宽表)
updated = PredictionModel.update_actual_result_v2(
date=prev_date,
hour=prev_hour,
actual_result=actual_result,
close_price=prev_close
)
for model_name in models_to_update:
updated = PredictionModel.update_actual_result(
date=prev_date,
hour=prev_hour,
model_name=model_name,
actual_result=actual_result,
close_price=prev_close
)
if updated:
# 仅打印一条日志代表更新成功，避免刷屏
pass
logger.info(f"📝 回填上一小时结果: ET {prev_date} {prev_hour}:00, "
f"实际{'涨' if actual_result == 1 else '跌'} (O:{prev_open:.0f} C:{prev_close:.0f})")
if updated:
logger.info(f"📝 回填上一小时结果: ET {prev_date} {prev_hour}:00, "
f"实际{'涨' if actual_result == 1 else '跌'} (O:{prev_open:.0f} C:{prev_close:.0f})")
except Exception as e:
logger.error(f"回填上一小时结果失败: {e}")
@@ -801,30 +798,38 @@ class PolymarketBTCTrader:
current_time_utc = pd.Timestamp(current_time)
current_time_et = current_time_utc.tz_localize('UTC').tz_convert(et_tz)
# 1. 保存预测到 model_predictions
common_data = {
# 1. 保存预测到 hourly_model_predictions (V2)
prediction_data = {
"timestamp": current_time_utc.isoformat(),
"timestamp_et": current_time_et.isoformat(),
"date": current_time_et.strftime('%Y-%m-%d'),
"hour": current_time_et.hour,
"actual_result": None,
"is_correct": None,
"open_price": float(df['open'].iloc[-1]),
"close_price": None,
"actual_result": None,
# DLinear
"dlinear_prediction": all_preds.get('dlinear', {}).get('prediction'),
"dlinear_prob": all_preds.get('dlinear', {}).get('probability'),
"dlinear_correct": None,
# N-BEATS
"nbeats_prediction": all_preds.get('nbeats', {}).get('prediction'),
"nbeats_prob": all_preds.get('nbeats', {}).get('probability'),
"nbeats_correct": None,
# TTM-R2
"ttm_prediction": all_preds.get('ttm', {}).get('prediction'),
"ttm_prob": all_preds.get('ttm', {}).get('probability'),
"ttm_correct": None,
# Ensemble
"ensemble_prediction": all_preds.get('ensemble', {}).get('prediction'),
"ensemble_votes": all_preds.get('ensemble', {}).get('votes'),
"ensemble_correct": None,
}
def save_single_pred(name, pred_res):
data = common_data.copy()
data['model_name'] = name
data['prediction'] = pred_res['prediction']
data['probability'] = pred_res.get('probability', 0.5)
PredictionModel.save_prediction(data)
if 'dlinear' in all_preds: save_single_pred("DLinear", all_preds['dlinear'])
if 'nbeats' in all_preds: save_single_pred("N-BEATS", all_preds['nbeats'])
if 'ttm' in all_preds: save_single_pred("TTM-R2", all_preds['ttm'])
if 'ensemble' in all_preds and all_preds['ensemble']: save_single_pred("Ensemble", all_preds['ensemble'])
PredictionModel.save_hourly_prediction(prediction_data)
logger.info(f"💾 [Init] 初始预测已保存到数据库")
# 2. 记录当前监控状态到 trades (作为 'in_progress')
trading_log.json
+9
View File
@@ -9,5 +9,14 @@
"side": "DOWN",
"price": 0.505,
"probability": 0.5563886165618896
},
{
"timestamp": "2026-02-11T22:32:46.922211",
"success": false,
"error": "该市场已存在成功交易记录",
"prediction": "DOWN",
"probability": 0.0,
"market": "Bitcoin Up or Down - February 11, 9AM ET",
"market_slug": "bitcoin-up-or-down-february-11-9am-et"
}
]