From 99fa79cde428607b139b4ad9a2b1709c81fe8f59 Mon Sep 17 00:00:00 2001 From: crypto_predictor Date: Wed, 11 Feb 2026 23:14:50 +0800 Subject: [PATCH] fix: timezone handling and prediction logic - 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. --- api/models.py | 199 +++++++++++++++++++++++--------- api/prediction_service.py | 24 +++- trader/auto_trade_polymarket.py | 111 +++++++++--------- trading_log.json | 9 ++ 4 files changed, 232 insertions(+), 111 deletions(-) diff --git a/api/models.py b/api/models.py index 8cec0d4..9cd499b 100644 --- a/api/models.py +++ b/api/models.py @@ -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'], diff --git a/api/prediction_service.py b/api/prediction_service.py index 4699c6c..15baad1 100644 --- a/api/prediction_service.py +++ b/api/prediction_service.py @@ -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() diff --git a/trader/auto_trade_polymarket.py b/trader/auto_trade_polymarket.py index 8d98eab..eebeecd 100644 --- a/trader/auto_trade_polymarket.py +++ b/trader/auto_trade_polymarket.py @@ -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') diff --git a/trading_log.json b/trading_log.json index a1afb17..1318b41 100644 --- a/trading_log.json +++ b/trading_log.json @@ -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" } ] \ No newline at end of file