chore: remove session reuse and pin proxy to 7898 for lark notifier

- 新增 daily_report_sender.py 定时任务程序
- 北京时间每天 8 点推送昨天 24 小时预测统计
- 包含各模型准确率、逐小时预测详情
- 失败交易订单汇总
- start_system.py 集成启动报告推送服务

.env.example

@@ -37,3 +37,9 @@ MYSQL_PORT=3306
 MYSQL_USER=user
 MYSQL_PASSWORD=password
 MYSQL_DB=crypto_predictor
+
+# Polymarket Configuration
+POLYMARKET_PRIVATE_KEY=your_private_key_here
+POLYMARKET_API_KEY=your_api_key_here
+POLYMARKET_API_SECRET=your_api_secret_here
+POLYMARKET_API_PASSPHRASE=your_api_passphrase_here

api/polymarket_client.py

@@ -36,6 +36,16 @@ class PolymarketClient:
     GAMMA_API = "https://gamma-api.polymarket.com"
     CLOB_API = "https://clob.polymarket.com"
 
+    @staticmethod
+    def _apply_global_proxy(proxy_url: str) -> None:
+        """统一设置代理环境变量，确保第三方网络库也走代理。"""
+        os.environ["HTTP_PROXY"] = proxy_url
+        os.environ["HTTPS_PROXY"] = proxy_url
+        os.environ["ALL_PROXY"] = proxy_url
+        os.environ["http_proxy"] = proxy_url
+        os.environ["https_proxy"] = proxy_url
+        os.environ["all_proxy"] = proxy_url
+    
     def __init__(self, private_key: Optional[str] = None):
         """
         初始化客户端
@@ -46,7 +56,8 @@ class PolymarketClient:
         self.private_key = private_key or os.getenv("POLYMARKET_PRIVATE_KEY")
         
         # 配置代理
-        proxy = os.getenv("SOCKS5_PROXY") if os.getenv("USE_PROXY", "false").lower() == "true" else None
+        proxy = os.getenv("SOCKS5_PROXY", "socks5://127.0.0.1:7890")
+        self._apply_global_proxy(proxy)
         
         # HTTP客户端 - 禁用压缩避免brotli问题
         self.http = httpx.Client(

daily_report_sender.py

@@ -0,0 +1,379 @@
+#!/usr/bin/env python3
+"""
+每日数据报告定时推送程序
+- 每天北京时间 8 点推送昨天的 24 小时预测统计
+- 包含各模型准确率、逐小时预测结果详情
+"""
+
+import schedule
+import time
+import logging
+from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo
+from typing import Dict, List
+
+from utils.send_msg import lark_notifier
+from api.models import PredictionModel, convert_to_cst, get_db_connection, TZ_CST, TZ_UTC
+
+# 配置日志
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    handlers=[
+        logging.FileHandler('logs/daily_report.log'),
+        logging.StreamHandler()
+    ]
+)
+logger = logging.getLogger(__name__)
+
+
+class DailyReportSender:
+    """每日数据报告发送器"""
+    
+    def __init__(self):
+        self.model = PredictionModel()
+        self.lark = lark_notifier
+    
+    def get_yesterday_failed_trades(self) -> List[Dict]:
+        """
+        获取昨天的失败交易订单
+        
+        Returns:
+            [
+                {
+                    'timestamp': '2026-02-27 10:30:00',
+                    'market_title': 'BTC 价格',
+                    'action': 'BUY',
+                    'status': 'failed',
+                    'error_message': '余额不足'
+                },
+                ...
+            ]
+        """
+        try:
+            # 获取昨天的日期（CST）
+            now_cst = datetime.now(TZ_CST)
+            yesterday_cst = now_cst - timedelta(days=1)
+            yesterday_date = yesterday_cst.strftime('%Y-%m-%d')
+            
+            # 计算 CST 时间范围
+            start_cst = datetime.strptime(f"{yesterday_date} 00:00:00", '%Y-%m-%d %H:%M:%S').replace(tzinfo=TZ_CST)
+            end_cst = start_cst + timedelta(days=1)
+            
+            start_str = start_cst.strftime('%Y-%m-%d %H:%M:%S')
+            end_str = end_cst.strftime('%Y-%m-%d %H:%M:%S')
+            
+            # 从 trades 表查询失败订单
+            with get_db_connection() as conn:
+                cursor = conn.cursor()
+                sql = """
+                    SELECT 
+                        timestamp,
+                        market_title,
+                        action,
+                        status,
+                        error_message
+                    FROM trades
+                    WHERE timestamp >= %s AND timestamp < %s
+                      AND status != 'MATCHED'
+                    ORDER BY timestamp DESC
+                """
+                cursor.execute(sql, (start_str, end_str))
+                rows = cursor.fetchall()
+            
+            return rows if rows else []
+            
+        except Exception as e:
+            logger.error(f"获取昨天失败交易失败: {e}", exc_info=True)
+            return []
+    
+    def get_yesterday_statistics(self) -> Dict:
+        """
+        获取昨天的 24 小时预测统计和详细数据
+        
+        Returns:
+            {
+                'date': 'YYYY-MM-DD',
+                'models': {
+                    'DLinear': {'total': 24, 'correct': 14, 'accuracy': 58.3},
+                    'N-BEATS': {...},
+                    'TTM-R2': {...},
+                    'Ensemble': {...}
+                },
+                'hourly_details': [
+                    {
+                        'hour': '00:00-01:00',
+                        'actual': '涨',
+                        'dlinear': {'pred': '涨', 'correct': True},
+                        'nbeats': {...},
+                        'ttm': {...},
+                        'ensemble': {...}
+                    },
+                    ...
+                ]
+            }
+        """
+        try:
+            # 获取昨天的日期（CST）
+            now_cst = datetime.now(TZ_CST)
+            yesterday_cst = now_cst - timedelta(days=1)
+            yesterday_date = yesterday_cst.strftime('%Y-%m-%d')
+            
+            # 计算 UTC 时间范围
+            start_cst = datetime.strptime(f"{yesterday_date} 00:00:00", '%Y-%m-%d %H:%M:%S').replace(tzinfo=TZ_CST)
+            end_cst = start_cst + timedelta(days=1)
+            
+            start_utc = start_cst.astimezone(TZ_UTC)
+            end_utc = end_cst.astimezone(TZ_UTC)
+            
+            start_str = start_utc.strftime('%Y-%m-%d %H:%M:%S')
+            end_str = end_utc.strftime('%Y-%m-%d %H:%M:%S')
+            
+            # 查询数据库
+            with get_db_connection() as conn:
+                cursor = conn.cursor()
+                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 timestamp >= %s AND timestamp < %s
+                    ORDER BY timestamp ASC
+                """
+                cursor.execute(sql, (start_str, end_str))
+                rows = cursor.fetchall()
+            
+            # 整理统计数据
+            models = ['dlinear', 'nbeats', 'ttm', 'ensemble']
+            model_names = {
+                'dlinear': 'DLinear',
+                'nbeats': 'N-BEATS',
+                'ttm': 'TTM-R2',
+                'ensemble': 'Ensemble'
+            }
+            
+            # 初始化统计
+            stats = {
+                'date': yesterday_date,
+                'models': {},
+                'hourly_details': []
+            }
+            
+            for model in models:
+                stats['models'][model_names[model]] = {
+                    'total': 0,
+                    'correct': 0,
+                    'accuracy': 0.0
+                }
+            
+            # 处理每行数据
+            for idx, row in enumerate(rows):
+                if row.get('actual_result') is None:
+                    continue
+                
+                # 获取时间戳（转为 CST 显示）
+                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_start = ts_cst.strftime('%H:%M')
+                hour_end = (ts_cst + timedelta(hours=1)).strftime('%H:%M')
+                hour_label = f"{hour_start}-{hour_end}"
+                
+                # 实际结果
+                actual_result = row['actual_result']
+                actual_label = '涨' if actual_result == 1 else '跌'
+                
+                # 每小时预测详情
+                hourly_item = {
+                    'hour': hour_label,
+                    'actual': actual_label
+                }
+                
+                # 逐模型处理预测结果
+                for model in models:
+                    pred = row[f"{model}_prediction"]
+                    correct = row[f"{model}_correct"]
+                    
+                    # 更新统计
+                    if pred is not None:
+                        stats['models'][model_names[model]]['total'] += 1
+                        if correct == 1:
+                            stats['models'][model_names[model]]['correct'] += 1
+                    
+                    # 添加到小时详情
+                    pred_label = '涨' if pred == 1 else ('跌' if pred == 0 else '无')
+                    correct_flag = correct == 1 if correct is not None else None
+                    
+                    hourly_item[model] = {
+                        'pred': pred_label,
+                        'correct': '✓' if correct_flag else ('✗' if correct_flag is False else '-')
+                    }
+                
+                stats['hourly_details'].append(hourly_item)
+            
+            # 计算准确率百分比
+            for model_name in stats['models']:
+                total = stats['models'][model_name]['total']
+                correct = stats['models'][model_name]['correct']
+                if total > 0:
+                    stats['models'][model_name]['accuracy'] = round(correct / total * 100, 2)
+            
+            return stats
+            
+        except Exception as e:
+            logger.error(f"获取昨天统计数据失败: {e}", exc_info=True)
+            return None
+    
+    def format_report(self, stats: Dict, failed_trades: List[Dict] = None) -> str:
+        """
+        格式化报告为飞书 Markdown 格式
+        
+        Args:
+            stats: get_yesterday_statistics 返回的数据字典
+            failed_trades: 失败订单列表
+        
+        Returns:
+            Markdown 格式的报告文本
+        """
+        if not stats:
+            return "❌ 无法获取昨天的统计数据，请检查数据库连接"
+        
+        date = stats['date']
+        models = stats['models']
+        
+        # 构建报告头
+        report = f"**【BTC 日报 - {date}】**\n\n"
+        
+        # 模型准确率统计部分
+        report += "**📊 模型准确率统计（24小时）**\n"
+        for model_name in ['DLinear', 'N-BEATS', 'TTM-R2', 'Ensemble']:
+            data = models.get(model_name, {})
+            total = data.get('total', 0)
+            correct = data.get('correct', 0)
+            accuracy = data.get('accuracy', 0)
+            
+            # 进度条可视化
+            bar_length = 10
+            filled = int(accuracy / 100 * bar_length)
+            bar = '█' * filled + '░' * (bar_length - filled)
+            
+            report += f"• **{model_name}**: {correct}/{total} {bar} {accuracy}%\n"
+        
+        report += "\n"
+        
+        # 交易失败订单详情
+        report += "**⚠️ 交易失败订单**\n"
+        if failed_trades and len(failed_trades) > 0:
+            report += f"昨天共有 {len(failed_trades)} 个失败订单\n"
+            for idx, trade in enumerate(failed_trades[:10], 1):  # 仅显示前10个
+                timestamp = trade.get('timestamp', '未知')
+                market_title = trade.get('market_title', '未知市场')
+                action = trade.get('action', '未知')
+                status = trade.get('status', '未知')
+                error_msg = trade.get('error_message', '无详细信息')
+                
+                # 简化错误信息 - 仅取第一行
+                if error_msg:
+                    error_msg = error_msg.split('\n')[0][:50]  # 仅取前50字
+                else:
+                    error_msg = '无错误信息'
+                
+                report += f"{idx}. [{timestamp}] {market_title} ({action}) → {error_msg}\n"
+            if len(failed_trades) > 10:
+                report += f"... 还有 {len(failed_trades) - 10} 个失败订单\n"
+        else:
+            report += "昨日无失败订单，运营正常 ✓\n"
+        
+        report += "\n生成时间: " + datetime.now(TZ_CST).strftime('%Y-%m-%d %H:%M:%S CST')
+        
+        return report
+    
+    def send_daily_report(self):
+        """获取昨天统计并推送"""
+        try:
+            logger.info("开始推送每日报告...")
+            
+            stats = self.get_yesterday_statistics()
+            if not stats:
+                logger.warning("无法获取昨天统计数据")
+                self.lark.send_alert(
+                    title="⚠️ 每日报告生成失败",
+                    content="无法获取昨天的统计数据，请检查数据库连接"
+                )
+                return
+            
+            # 获取失败订单
+            failed_trades = self.get_yesterday_failed_trades()
+            
+            # 格式化报告
+            report = self.format_report(stats, failed_trades)
+            
+            # 推送到飞书
+            self.lark.send_alert(
+                title="📊 BTC 每日数据报告",
+                content=report
+            )
+            
+            logger.info(f"每日报告已送出: {stats['date']}")
+            
+        except Exception as e:
+            logger.error(f"推送每日报告异常: {e}", exc_info=True)
+            self.lark.send_alert(
+                title="❌ 每日报告推送异常",
+                content=f"错误详情: {str(e)}"
+            )
+
+
+def schedule_daily_report():
+    """
+    启动每日报告定时任务
+    北京时间每天 8 点执行
+    """
+    sender = DailyReportSender()
+    sender.send_daily_report()  # 启动时立即发送一次（可选）
+    # 使用 schedule 库在北京时间 8 点触发
+    # 由于 schedule 库基于本地时间，当前环境如果是 macOS 默认时间，
+    # 我们使用一个包装器来处理 CST 时间转换
+    def cst_8am_job():
+        """北京时间 8 点执行的任务"""
+        now_cst = datetime.now(TZ_CST)
+        if now_cst.hour == 8 and now_cst.minute < 1:
+            # 确保只在 08:00-08:01 间隔执行一次（避免重复）
+            sender.send_daily_report()
+    
+    # 每分钟检查一次是否需要执行 (确保能在 8 点执行)
+    schedule.every(1).minutes.do(cst_8am_job)
+    
+    logger.info("每日报告定时任务已启动（北京时间每天 8 点执行）")
+    
+    # 启动调度循环
+    while True:
+        schedule.run_pending()
+        time.sleep(1)
+
+
+def run_scheduler_background():
+    """
+    在后台线程中运行调度器（供 start_system.py 调用）
+    """
+    import threading
+    
+    scheduler_thread = threading.Thread(target=schedule_daily_report, daemon=True)
+    scheduler_thread.start()
+    
+    logger.info("每日报告调度器已在后台启动")
+    return scheduler_thread
+
+
+if __name__ == '__main__':
+    # 支持单独运行以及从其他模块导入
+    logger.info("启动每日报告系统...")
+    schedule_daily_report()

data/database.py

@@ -26,7 +26,9 @@ class DatabaseManager:
             password = os.getenv('MYSQL_PASSWORD')
             database = os.getenv('MYSQL_DB')
             self._engine = create_engine(
-                f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}?charset=utf8mb4"
+                f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}?charset=utf8mb4",
+                pool_pre_ping=True,
+                pool_recycle=3600
             )
         return self._engine
 

data/fetcher.py

@@ -1,21 +1,40 @@
 """从Binance获取加密货币OHLCV数据"""
 
 import os
+import json
+import time
 from datetime import datetime, timedelta
 from pathlib import Path
-from typing import Optional, Literal
+from typing import Optional, Literal, Any
+import logging
+from urllib.parse import urlparse
 
 import pandas as pd
 from binance.client import Client
 import pytz
 from dotenv import load_dotenv
+from websocket import create_connection
 
 from data.database import DatabaseManager
+from utils.send_msg import lark_notifier
 
 # 加载环境变量
 load_dotenv()
 
 
+logger = logging.getLogger("polymarket_trader")
+
+
+def _apply_global_proxy(proxy_url: str) -> None:
+    """统一设置代理环境变量，确保第三方网络库也走代理。"""
+    os.environ['HTTP_PROXY'] = proxy_url
+    os.environ['HTTPS_PROXY'] = proxy_url
+    os.environ['ALL_PROXY'] = proxy_url
+    os.environ['http_proxy'] = proxy_url
+    os.environ['https_proxy'] = proxy_url
+    os.environ['all_proxy'] = proxy_url
+
+
 class BinanceDataFetcher:
     """从Binance公开API获取和缓存OHLCV数据"""
 
@@ -24,6 +43,13 @@ class BinanceDataFetcher:
         '15m': Client.KLINE_INTERVAL_15MINUTE,
     }
 
+    INTERVAL_MS = {
+        '1h': 60 * 60 * 1000,
+        '15m': 15 * 60 * 1000,
+    }
+
+    WS_API_ENDPOINT = "wss://ws-api.binance.com/ws-api/v3"
+
     def __init__(self, cache_dir: str = "./data/cache"):
         """初始化Binance客户端和数据库管理器
         
@@ -31,33 +57,24 @@ class BinanceDataFetcher:
             cache_dir: 缓存/数据库目录
         """
         # 配置代理
-        proxies = None
-        use_proxy = os.getenv('USE_PROXY', 'false').lower() == 'true'
-
-        if use_proxy:
         socks5_proxy = os.getenv('SOCKS5_PROXY', 'socks5://127.0.0.1:7890')
+        _apply_global_proxy(socks5_proxy)
         proxies = {
             'http': socks5_proxy,
             'https': socks5_proxy
         }
-            print(f"✓ 使用SOCKS5代理: {socks5_proxy}")
+        logger.info(f"强制使用SOCKS5代理: {socks5_proxy}")
 
         try:
-            # 初始化Binance客户端（带代理）
-            if proxies:
+            # 初始化Binance客户端（强制走代理）
             self.client = Client(requests_params={'proxies': proxies})
-            else:
-                self.client = Client()
 
             # 测试连接
             self.client.ping()
-            print("✓ 已成功连接到Binance API")
+            logger.info("已成功连接到Binance API")
         except Exception as e:
-            print(f"⚠ Binance API连接失败: {e}")
-            print("提示：")
-            print("  1. 检查代理设置 (.env文件)")
-            print("  2. 确保代理服务正在运行")
-            print("  3. 或者使用已缓存的数据")
+            logger.warning(f"Binance API连接失败: {e}")
+            logger.warning("提示: 1) 检查代理设置 2) 确保代理服务运行 3) 或者使用缓存数据")
             self.client = None
 
         # 初始化数据库管理器
@@ -69,7 +86,8 @@ class BinanceDataFetcher:
         timeframe: Literal['1h', '15m'],
         start_date: Optional[str] = None,
         end_date: Optional[str] = None,
-        use_cache: bool = True
+        use_cache: bool = True,
+        max_retries: int = 3  # [新增] 最大重试次数参数
     ) -> pd.DataFrame:
         """获取加密货币对的OHLCV数据
 
@@ -109,13 +127,13 @@ class BinanceDataFetcher:
                 
                 # 简单判断覆盖范围 (容忍少量缺失)
                 if min_ts <= start_dt + timedelta(days=2) and max_ts >= end_dt - timedelta(days=2):
-                    print(f"从数据库加载数据：{symbol} {timeframe} ({len(df)} rows)")
+                    logger.info(f"从数据库加载数据: {symbol} {timeframe} ({len(df)} rows)")
                     return df
 
         # 2. 从API获取数据
         if self.client is None:
             if not df.empty:
-                 print(f"⚠ API不可用，返回数据库现有数据 ({len(df)} rows)")
+                 logger.warning(f"API不可用，返回数据库现有数据 ({len(df)} rows)")
                  return df
             
             raise ConnectionError(
@@ -125,24 +143,39 @@ class BinanceDataFetcher:
         # Binance使用USDT交易对
         pair = f"{symbol}USDT"
         interval = self.INTERVALS[timeframe]
+        interval_ms = self.INTERVAL_MS[timeframe]
 
-        print(f"获取 {symbol} {timeframe} 数据，时间范围：{start_dt.date()} 至 {end_dt.date()}...")
+        logger.info(f"获取 {symbol} {timeframe} 数据，时间范围: {start_dt.date()} 至 {end_dt.date()}")
 
-        # 获取K线数据
+        klines = []
+        for attempt in range(max_retries):
             try:
-            klines = self.client.get_historical_klines(
+                # 使用 Binance WebSocket API 的 klines 方法获取历史K线
+                klines = self._get_historical_klines_via_ws(
                     symbol=pair,
                     interval=interval,
-                start_str=str(int(start_dt.timestamp() * 1000)),
-                end_str=str(int(end_dt.timestamp() * 1000))
+                    start_ts_ms=int(start_dt.timestamp() * 1000),
+                    end_ts_ms=int(end_dt.timestamp() * 1000),
+                    interval_ms=interval_ms
                 )
+                # 如果成功获取，跳出循环
+                break
             except Exception as e:
+                # 如果是最后一次尝试，且依然失败
+                if attempt == max_retries - 1:
+                    logger.error(f"第 {attempt + 1} 次尝试失败，已达到最大重试次数。错误: {e}")
                     if not df.empty:
-                print(f"⚠ API请求失败 ({e})，返回数据库现有数据")
+                        logger.warning("WebSocket请求最终失败，返回数据库现有数据")
                         return df
-            raise e
+                    raise e  # 如果没有缓存数据，抛出异常
+                
+                # 计算等待时间 (指数退避: 2s, 4s, 8s...)
+                wait_time = 2 * (2 ** attempt) 
+                logger.warning(f"请求失败 (尝试 {attempt + 1}/{max_retries})，{wait_time}秒后重试。错误: {e}")
+                time.sleep(wait_time)
 
         if not klines:
+             lark_notifier.send_alert(f"数据获取失败: {symbol} {timeframe}", "连续3次请求或解析发生异常，已放弃重试")
              if not df.empty:
                  return df
              raise ValueError(f"未获取到 {symbol} {timeframe} 的数据")
@@ -164,13 +197,94 @@ class BinanceDataFetcher:
         # 按时间排序
         new_df = new_df.sort_values('timestamp').reset_index(drop=True)
 
-        print(f"已获取 {len(new_df)} 行数据")
+        logger.info(f"已获取 {len(new_df)} 行数据")
 
         # 保存到数据库
         self.db_manager.save_ohlcv(new_df, symbol, timeframe)
         
         return new_df
 
+    def _get_historical_klines_via_ws(
+        self,
+        symbol: str,
+        interval: str,
+        start_ts_ms: int,
+        end_ts_ms: int,
+        interval_ms: int,
+        limit: int = 1000
+    ):
+        """通过 Binance WebSocket API 分页拉取历史K线。"""
+        if end_ts_ms <= start_ts_ms:
+            return []
+
+        all_klines = []
+        request_id = 1
+        current_start = start_ts_ms
+
+        # WebSocket API 单次最多返回 1000 条，这里按窗口分页拉取。
+        while current_start < end_ts_ms:
+            max_window_end = current_start + (limit * interval_ms)
+            current_end = min(end_ts_ms, max_window_end)
+
+            payload = {
+                "id": request_id,
+                "method": "klines",
+                "params": {
+                    "symbol": symbol,
+                    "interval": interval,
+                    "startTime": current_start,
+                    "endTime": current_end,
+                    "limit": limit
+                }
+            }
+
+            response_data = self._ws_request(payload)
+            batch = response_data.get("result", [])
+            if not batch:
+                break
+
+            all_klines.extend(batch)
+            last_open_time = int(batch[-1][0])
+            next_start = last_open_time + interval_ms
+            if next_start <= current_start:
+                break
+            current_start = next_start
+            request_id += 1
+
+        return all_klines
+
+    def _ws_request(self, payload: dict) -> dict:
+        """发送单次 WebSocket API 请求并返回 JSON 响应。"""
+        ws_kwargs: dict[str, Any] = {
+            "timeout": 30
+        }
+
+        socks5_proxy = os.getenv('SOCKS5_PROXY', 'socks5://127.0.0.1:7890')
+        parsed = urlparse(socks5_proxy)
+        if parsed.hostname and parsed.port:
+            scheme = (parsed.scheme or 'socks5').lower()
+            if scheme not in {'socks4', 'socks5', 'http'}:
+                scheme = 'socks5'
+            if scheme == 'socks5h':
+                scheme = 'socks5'
+            ws_kwargs.update({
+                "http_proxy_host": parsed.hostname,
+                "http_proxy_port": parsed.port,
+                "proxy_type": scheme
+            })
+
+        ws = create_connection(self.WS_API_ENDPOINT, **ws_kwargs)
+        try:
+            ws.send(json.dumps(payload))
+            raw = ws.recv()
+            data = json.loads(raw)
+            status = data.get("status", 200)
+            if status != 200:
+                raise RuntimeError(f"Binance WS API返回异常状态: {data}")
+            return data
+        finally:
+            ws.close()
+
     def get_latest_price(self, symbol: Literal['BTC', 'ETH', 'SOL']) -> float:
         """获取符号的当前价格
 
@@ -180,6 +294,9 @@ class BinanceDataFetcher:
         Returns:
             当前USDT价格
         """
+        if self.client is None:
+            raise ConnectionError("Binance客户端不可用，无法获取最新价格")
+
         pair = f"{symbol}USDT"
         ticker = self.client.get_symbol_ticker(symbol=pair)
         return float(ticker['price'])

data/trading_db.py

@@ -53,12 +53,28 @@ class TradingDatabase:
                     error_message TEXT,
                     raw_data JSON,
                     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    order_id VARCHAR(100),
                     INDEX idx_market_slug (market_slug),
                     INDEX idx_status (status)
                 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
             ''')
             conn.commit()
             
+            # 检查并添加 order_id 列 (Migration)
+            self._ensure_column_exists(conn, 'trades', 'order_id', 'VARCHAR(100)')
+
+    def _ensure_column_exists(self, conn, table_name, column_name, column_type):
+        """确保表中存在指定列"""
+        try:
+            cursor = conn.cursor()
+            cursor.execute(f"SHOW COLUMNS FROM {table_name} LIKE '{column_name}'")
+            if not cursor.fetchone():
+                print(f"正在添加列 {column_name} 到表 {table_name}...")
+                cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_type}")
+                conn.commit()
+        except Exception as e:
+            print(f"检查/添加列失败: {e}")
+
     def log_trade(self, trade_data: dict):
         """记录交易信息"""
         try:
@@ -72,8 +88,8 @@ class TradingDatabase:
                     INSERT INTO trades (
                         timestamp, market_title, market_slug, model_name,
                         prediction, probability, action, price, amount, size,
-                        status, strategy, error_message, raw_data
-                    ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+                        status, strategy, error_message, raw_data, order_id
+                    ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                 '''
                 
                 cursor.execute(sql, (
@@ -90,7 +106,8 @@ class TradingDatabase:
                     trade_data.get('status'),
                     trade_data.get('strategy'),
                     trade_data.get('error_message'),
-                    raw_data_json
+                    raw_data_json,
+                    trade_data.get('order_id')
                 ))
                 conn.commit()
         except Exception as e:
@@ -109,3 +126,31 @@ class TradingDatabase:
         except Exception as e:
             print(f"检查已存在交易失败: {e}")
             return False
+
+    def get_existing_trade(self, market_slug: str) -> dict:
+        """获取指定市场的最近一次交易记录"""
+        try:
+            with self.get_connection() as conn:
+                cursor = conn.cursor()
+                # Get the latest trade for this market
+                sql = 'SELECT * FROM trades WHERE market_slug = %s ORDER BY id DESC LIMIT 1'
+                cursor.execute(sql, (market_slug,))
+                return cursor.fetchone()
+        except Exception as e:
+            print(f"获取交易记录失败: {e}")
+            return None
+
+    def update_trade_status(self, trade_id: int, status: str, error_message: str = None):
+        """更新交易状态"""
+        try:
+            with self.get_connection() as conn:
+                cursor = conn.cursor()
+                if error_message:
+                    sql = 'UPDATE trades SET status = %s, error_message = %s WHERE id = %s'
+                    cursor.execute(sql, (status, error_message, trade_id))
+                else:
+                    sql = 'UPDATE trades SET status = %s WHERE id = %s'
+                    cursor.execute(sql, (status, trade_id))
+                conn.commit()
+        except Exception as e:
+            print(f"更新交易状态失败: {e}")

ensemble_dlinear_jan_2026.py

@@ -0,0 +1,289 @@
+"""2026年1月BTC数据 - DLinear 准确率测试
+
+输入数据：预测K线的前一根（即用截止到t的数据预测t+1）
+价格对比：预测结果与预测K线(t+1)的收盘价做对比
+时间：币安数据UTC，输出转为北京时间(UTC+8)，按北京时间分天
+DLinear预测逻辑：整个df一次性预处理，按索引遍历窗口
+输出：每小时详细数据、每天准确率、总准确率、美股开盘标记
+"""
+
+import sys
+import os
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import torch
+import numpy as np
+import pandas as pd
+import pickle
+import pandas_ta as ta
+from datetime import timedelta
+from sklearn.preprocessing import StandardScaler
+from data.fetcher import BinanceDataFetcher
+
+# ==================== 时区与美股判断 ====================
+
+def utc_to_beijing(ts):
+    """UTC时间转北京时间(UTC+8)"""
+    return ts + timedelta(hours=8)
+
+def is_us_stock_open_by_utc(utc_ts):
+    """根据UTC时间判断美股当天是否开盘
+
+    美股交易日基于美东时间(ET)，UTC-5(冬令时)
+    2026年1月在冬令时范围内（夏令时3月第二个周日开始）
+    所以美东 = UTC - 5h
+    """
+    et_ts = utc_ts - timedelta(hours=5)
+    et_date = pd.Timestamp(et_ts.date())
+
+    us_holidays_jan_2026 = {
+        pd.Timestamp('2026-01-01'),  # 元旦 New Year's Day
+        pd.Timestamp('2026-01-19'),  # 马丁·路德·金日 MLK Day
+    }
+
+    if et_date.weekday() >= 5:
+        return False
+    if et_date in us_holidays_jan_2026:
+        return False
+    return True
+
+# ==================== pickle反序列化占位类 ====================
+
+class ClassificationPreprocessor:
+    """pickle.load时需要能找到这个类名"""
+    def __init__(self, lookback=168, forecast_horizon=1):
+        self.lookback = lookback
+        self.forecast_horizon = forecast_horizon
+        self.scaler = StandardScaler()
+        self.is_fitted = False
+
+# ==================== 工具函数 ====================
+
+def fmt_pred(val):
+    """格式化预测结果"""
+    if val is None or (isinstance(val, float) and np.isnan(val)):
+        return "N/A"
+    return "涨" if int(val) == 1 else "跌"
+
+def fmt_result(pred, actual):
+    """格式化对错"""
+    if pred is None or (isinstance(pred, float) and np.isnan(pred)):
+        return "  "
+    return "✓" if int(pred) == int(actual) else "✗"
+
+
+def main():
+    print("=" * 90)
+    print("2026年1月 BTC 1h 涨跌预测准确率测试 (DLinear)")
+    print("输入：预测K线的前一根 | 对比：预测结果 vs 预测K线收盘价 | 时间：北京时间(UTC+8)")
+    print("=" * 90)
+
+    device = torch.device('mps' if torch.backends.mps.is_available() else 'cpu')
+    print(f"使用设备: {device}")
+
+    # ========== 步骤1: 加载模型 ==========
+    print("\n[1/4] 加载模型...")
+
+    from models.dlinear.model import DLinear
+    checkpoint = torch.load('./models/checkpoints/dlinear_classification_2026/best_model.pt',
+                            map_location=device, weights_only=False)
+    model = DLinear(checkpoint['config'])
+    model.load_state_dict(checkpoint['model_state_dict'])
+    model = model.to(device).eval()
+    with open('./models/checkpoints/dlinear_classification_2026/preprocessor.pkl', 'rb') as f:
+        preprocessor = pickle.load(f)
+    print("  ✓ DLinear已加载")
+
+    # ========== 步骤2: 获取测试数据 ==========
+    print("\n[2/4] 获取数据...")
+    fetcher = BinanceDataFetcher()
+    df = fetcher.fetch_ohlcv('BTC', '1h', start_date='2025-12-01', end_date='2026-03-01', use_cache=False)
+
+    print(f"数据范围(UTC): {df['timestamp'].min()} -> {df['timestamp'].max()}")
+    print(f"总行数: {len(df)}")
+
+    # ========== 步骤3: 预处理与预测 ==========
+    print("\n[3/4] 运行预测...")
+
+    # DLinear 整体预处理（无create_labels，整个df一次性处理）
+    lookback = preprocessor.lookback
+    df_dl = df.copy()
+    df_dl['rsi'] = ta.rsi(df_dl['close'], length=14)
+    macd = ta.macd(df_dl['close'])
+    if macd is not None and not macd.empty:
+        df_dl['macd'] = macd.iloc[:, 0]
+        df_dl['macd_hist'] = macd.iloc[:, 1]
+        df_dl['macd_signal'] = macd.iloc[:, 2]
+    df_dl = df_dl.dropna().reset_index(drop=True)
+
+    feature_cols = ['open', 'high', 'low', 'close', 'volume',
+                   'rsi', 'macd', 'macd_hist', 'macd_signal']
+    scaled = preprocessor.scaler.transform(df_dl[feature_cols])
+
+    X_dl = []
+    idx_dl = []
+    for i in range(len(scaled) - lookback):
+        X_dl.append(scaled[i:i + lookback])
+        idx_dl.append(i + lookback - 1)
+    X_dl = np.array(X_dl)
+
+    # 构建 timestamp -> 窗口索引映射
+    dl_ts_map = {}
+    for i, idx in enumerate(idx_dl):
+        ts = df_dl.iloc[idx]['timestamp']
+        dl_ts_map[ts] = i
+
+    # 找出1月（北京时间）的所有预测目标K线索引
+    jan_targets = []
+    for target_idx in range(1, len(df)):
+        target_time_utc = df.iloc[target_idx]['timestamp']
+        target_time_bj = utc_to_beijing(target_time_utc)
+        if target_time_bj.year == 2026 and target_time_bj.month == 2:
+            jan_targets.append(target_idx)
+
+    print(f"1月（北京时间）目标K线数: {len(jan_targets)}")
+
+    results = []
+
+    for target_idx in jan_targets:
+        target_time_utc = df.iloc[target_idx]['timestamp']
+        target_time_bj = utc_to_beijing(target_time_utc)
+
+        # 输入K线索引（前一根）
+        input_idx = target_idx - 1
+        current_close = df.iloc[input_idx]['close']
+        target_close = df.iloc[target_idx]['close']
+        actual_label = 1 if target_close > current_close else 0
+
+        # --- DLinear 预测 ---
+        dl_pred = None
+        dl_prob = None
+        input_ts = df.iloc[input_idx]['timestamp']
+        dl_i = dl_ts_map.get(input_ts)
+        if dl_i is not None and dl_i < len(X_dl):
+            x_tensor = torch.FloatTensor(X_dl[dl_i:dl_i+1]).to(device)
+            with torch.no_grad():
+                logits = model(x_tensor).cpu().numpy()
+                prob = 1 / (1 + np.exp(-logits))
+                dl_pred = int(prob.item() > 0.5)
+                dl_prob = float(prob.item())
+
+        results.append({
+            'time_utc': target_time_utc,
+            'time_bj': target_time_bj,
+            'current_close': current_close,
+            'target_close': target_close,
+            'actual': actual_label,
+            'dlinear': dl_pred,
+            'dl_prob': dl_prob,
+        })
+
+    # ========== 步骤4: 输出结果 ==========
+    if not results:
+        print("没有找到符合条件的测试样本")
+        return
+
+    df_res = pd.DataFrame(results)
+    df_res['date_bj'] = df_res['time_bj'].dt.date
+    df_res['us_open'] = df_res['time_utc'].apply(is_us_stock_open_by_utc)
+
+    total_samples = len(df_res)
+
+    print(f"\n[4/4] 准确率统计 (2026年1月，北京时间)")
+    print(f"总样本数: {total_samples}")
+
+    # ========== 每小时详细数据 ==========
+    print("\n" + "=" * 90)
+    print("每小时详细数据")
+    print("=" * 90)
+
+    hourly_header = (
+        f"{'北京时间':<18} {'前收盘':>10} {'现收盘':>10} {'实际':>4} "
+        f"{'DL':>3} {'':>1} {'概率':>6}"
+    )
+
+    for date_bj, day_group in df_res.groupby('date_bj'):
+        us_open = day_group['us_open'].iloc[0]
+        us_str = "美股开盘" if us_open else "美股休市"
+
+        dl_day_correct = (day_group['dlinear'] == day_group['actual']).sum()
+        dl_day_total = day_group['dlinear'].notna().sum()
+        dl_day_acc = dl_day_correct / dl_day_total if dl_day_total > 0 else 0
+
+        print(f"\n--- {date_bj} ({us_str}) --- DL准确率:{dl_day_acc:.1%} ({dl_day_correct}/{dl_day_total})")
+        print(hourly_header)
+        print("-" * 90)
+
+        for _, row in day_group.iterrows():
+            time_str = row['time_bj'].strftime('%m-%d %H:%M')
+            actual_str = fmt_pred(row['actual'])
+            dl_str = fmt_pred(row['dlinear'])
+            dl_r = fmt_result(row['dlinear'], row['actual'])
+            prob_str = f"{row['dl_prob']:.3f}" if row['dl_prob'] is not None else "  N/A"
+
+            print(
+                f"{time_str:<18} {row['current_close']:>10.2f} {row['target_close']:>10.2f} {actual_str:>4} "
+                f"{dl_str:>3} {dl_r:>1} {prob_str:>6}"
+            )
+
+    # ========== 每日准确率汇总表 ==========
+    print("\n\n" + "=" * 90)
+    print("每日准确率汇总 (北京时间)")
+    print("=" * 90)
+
+    header = f"{'日期':<14} {'样本':>4} {'美股':>6}  {'DLinear':>8}"
+    print(header)
+    print("-" * 90)
+
+    for date_bj, group in df_res.groupby('date_bj'):
+        total = len(group)
+        us_open = group['us_open'].iloc[0]
+        us_str = "开盘" if us_open else "休市"
+
+        dl_valid = group.dropna(subset=['dlinear'])
+        dl_acc = (dl_valid['dlinear'] == dl_valid['actual']).mean() if len(dl_valid) > 0 else 0
+        dl_str = f"{dl_acc:.1%}"
+
+        print(f"{str(date_bj):<14} {total:>4} {us_str:>6}  {dl_str:>8}")
+
+    print("-" * 90)
+
+    # --- 总准确率 ---
+    dl_valid_all = df_res.dropna(subset=['dlinear'])
+    dl_total = (dl_valid_all['dlinear'] == dl_valid_all['actual']).mean() if len(dl_valid_all) > 0 else 0
+
+    print(f"{'总计':<14} {total_samples:>4} {'':>6}  {dl_total:.2%}")
+    print("=" * 90)
+
+    # --- 美股开盘日 vs 休市日对比 ---
+    print("\n[美股开盘日 vs 休市日 准确率对比]")
+    print("-" * 60)
+    for label, flag in [("开盘日", True), ("休市日", False)]:
+        subset = df_res[df_res['us_open'] == flag]
+        if len(subset) == 0:
+            continue
+        n = len(subset)
+        dl_s = subset.dropna(subset=['dlinear'])
+        dl_a = (dl_s['dlinear'] == dl_s['actual']).mean() if len(dl_s) > 0 else 0
+        print(f"  {label} ({n}样本): DLinear={dl_a:.2%}")
+
+    print("-" * 60)
+
+    # --- 预测分布统计 ---
+    print("\n[预测分布统计]")
+    print("-" * 60)
+    actual_up = (df_res['actual'] == 1).sum()
+    actual_down = (df_res['actual'] == 0).sum()
+    print(f"  实际: 上涨={actual_up} ({actual_up/total_samples:.1%})  下跌={actual_down} ({actual_down/total_samples:.1%})")
+
+    dl_valid = df_res['dlinear'].dropna()
+    if len(dl_valid) > 0:
+        up = (dl_valid == 1).sum()
+        down = (dl_valid == 0).sum()
+        print(f"  DLinear: 预测上涨={up} ({up/len(dl_valid):.1%})  预测下跌={down} ({down/len(dl_valid):.1%})")
+
+    print("=" * 90)
+
+
+if __name__ == "__main__":
+    main()

ensemble_models_feb_2026.py

@@ -0,0 +1,247 @@
+"""2026年2月BTC数据 - DLinear, N-BEATS, TTM-R2 准确率测试 (使用统一推理管理器)
+
+输入数据：预测K线的前一根（即用截止到t的数据预测t+1）
+价格对比：预测结果与预测K线(t+1)的收盘价做对比
+时间：币安数据UTC，输出转为北京时间(UTC+8)，按北京时间分天
+"""
+
+import sys
+import os
+import pandas as pd
+import numpy as np
+from datetime import timedelta
+from typing import Dict, List, Any
+
+# 添加项目路径
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from data.fetcher import BinanceDataFetcher
+from models.inference import get_inference_manager
+
+# ==================== 时区与美股判断 ====================
+
+def utc_to_beijing(ts):
+    """UTC时间转北京时间(UTC+8)"""
+    return ts + timedelta(hours=8)
+
+def is_us_stock_open_by_utc(utc_ts):
+    """根据UTC时间判断美股当天是否开盘"""
+    et_ts = utc_ts - timedelta(hours=5)
+    et_date = pd.Timestamp(et_ts.date())
+
+    us_holidays_feb_2026 = {
+        pd.Timestamp('2026-02-16'),  # 总统日 Presidents' Day
+    }
+
+    if et_date.weekday() >= 5:
+        return False
+    if et_date in us_holidays_feb_2026:
+        return False
+    return True
+
+def fmt_pred(val):
+    """格式化预测结果"""
+    if val is None or (isinstance(val, float) and np.isnan(val)):
+        return "N/A"
+    return "涨" if int(val) == 1 else "跌"
+
+def fmt_result(pred, actual):
+    """格式化对错"""
+    if pred is None or (isinstance(pred, float) and np.isnan(pred)):
+        return "  "
+    return "✓" if int(pred) == int(actual) else "✗"
+
+
+def main():
+    print("=" * 110)
+    print("2026年2月 BTC 1h 涨跌预测准确率测试 (DLinear, N-BEATS, TTM-R2)")
+    print("基于统一推理管理器 (models.inference)")
+    print("=" * 110)
+
+    # ========== 步骤1: 加载模型 ==========
+    print("\n[1/4] 加载模型...")
+    manager = get_inference_manager()
+    manager.load_all_models()
+    
+    # ========== 步骤2: 获取测试数据 ==========
+    print("\n[2/4] 获取2026年2月数据...")
+    fetcher = BinanceDataFetcher()
+    # 获取足够长的数据以供lookback
+    df = fetcher.fetch_ohlcv('BTC', '1h', start_date='2025-12-01', end_date='2026-03-01', use_cache=False)
+
+    print(f"数据范围(UTC): {df['timestamp'].min()} -> {df['timestamp'].max()}")
+    print(f"总行数: {len(df)}")
+
+    # ========== 步骤3: 运行预测 ==========
+    print("\n[3/4] 运行预测 (逐小时回测)...")
+    
+    # 找出2月（北京时间）的所有预测目标K线索引
+    jan_targets = []
+    for target_idx in range(200, len(df)): # 从200开始确保有足够lookback
+        target_time_utc = df.iloc[target_idx]['timestamp']
+        target_time_bj = utc_to_beijing(target_time_utc)
+        if target_time_bj.year == 2026 and target_time_bj.month == 2:
+            jan_targets.append(target_idx)
+            
+    print(f"2月（北京时间）目标K线数: {len(jan_targets)}")
+    
+    results = []
+    
+    for i, target_idx in enumerate(jan_targets):
+        if i % 50 == 0:
+            print(f"  处理进度: {i}/{len(jan_targets)}...")
+            
+        target_time_utc = df.iloc[target_idx]['timestamp']
+        target_time_bj = utc_to_beijing(target_time_utc)
+
+        # 输入数据：截止到 target_idx - 1
+        # 注意：predict_all 需要传入DataFrame，它会取最后一行作为 "当前"
+        # 所以我们需要传入 df.iloc[:target_idx] (不包含 target_idx)
+        # Python切片是 [start:end)，所以 [:target_idx] 包含 0 到 target_idx-1
+        df_input = df.iloc[:target_idx].copy()
+        
+        # 实际结果
+        current_close = df.iloc[target_idx-1]['close']
+        target_close = df.iloc[target_idx]['close']
+        actual_label = 1 if target_close > current_close else 0
+        
+        # 预测
+        preds = manager.predict_all(df_input)
+        
+        dl_pred = preds.get('dlinear', {}).get('prediction')
+        nb_pred = preds.get('nbeats', {}).get('prediction')
+        ttm_pred = preds.get('ttm', {}).get('prediction')
+        ens_pred = preds.get('ensemble', {}).get('prediction')
+        
+        results.append({
+            'time_utc': target_time_utc,
+            'time_bj': target_time_bj,
+            'current_close': current_close,
+            'target_close': target_close,
+            'actual': actual_label,
+            'dlinear': dl_pred,
+            'nbeats': nb_pred,
+            'ttm': ttm_pred,
+            'ensemble': ens_pred,
+            'us_open': is_us_stock_open_by_utc(target_time_utc)
+        })
+
+    # ========== 步骤4: 输出结果 ==========
+    if not results:
+        print("没有找到符合条件的测试样本")
+        return
+
+    df_res = pd.DataFrame(results)
+    df_res['date_bj'] = df_res['time_bj'].dt.date
+    total_samples = len(df_res)
+
+    print(f"\n[4/4] 准确率统计 (2026年2月，北京时间)")
+    print(f"总样本数: {total_samples}")
+
+    # ========== 每小时详细数据 ==========
+    print("\n" + "=" * 110)
+    print("每小时详细数据")
+    print("=" * 110)
+
+    hourly_header = (
+        f"{'北京时间':<18} {'前收盘':>10} {'现收盘':>10} {'实际':>4} "
+        f"{'DL':>3} {'':>1} {'NB':>3} {'':>1} {'TTM':>3} {'':>1} {'集成':>4} {'':<1}"
+    )
+
+    for date_bj, day_group in df_res.groupby('date_bj'):
+        us_open = day_group['us_open'].iloc[0]
+        us_str = "美股开盘" if us_open else "美股休市"
+
+        dl_day_acc = (day_group['dlinear'] == day_group['actual']).mean()
+        
+        nb_valid = day_group.dropna(subset=['nbeats'])
+        nb_day_acc = (nb_valid['nbeats'] == nb_valid['actual']).mean() if len(nb_valid) > 0 else None
+        
+        ttm_valid = day_group.dropna(subset=['ttm'])
+        ttm_day_acc = (ttm_valid['ttm'] == ttm_valid['actual']).mean() if len(ttm_valid) > 0 else None
+        
+        ens_valid = day_group.dropna(subset=['ensemble'])
+        ens_day_acc = (ens_valid['ensemble'] == ens_valid['actual']).mean() if len(ens_valid) > 0 else None
+
+        nb_acc_str = f"{nb_day_acc:.1%}" if nb_day_acc is not None else "N/A"
+        ttm_acc_str = f"{ttm_day_acc:.1%}" if ttm_day_acc is not None else "N/A"
+        ens_acc_str = f"{ens_day_acc:.1%}" if ens_day_acc is not None else "N/A"
+
+        print(f"\n--- {date_bj} ({us_str}) --- DL:{dl_day_acc:.1%} | NB:{nb_acc_str} | TTM:{ttm_acc_str} | 集成:{ens_acc_str}")
+        print(hourly_header)
+        print("-" * 110)
+
+        for _, row in day_group.iterrows():
+            time_str = row['time_bj'].strftime('%m-%d %H:%M')
+            actual_str = fmt_pred(row['actual'])
+
+            dl_str = fmt_pred(row['dlinear'])
+            dl_r = fmt_result(row['dlinear'], row['actual'])
+            nb_str = fmt_pred(row['nbeats'])
+            nb_r = fmt_result(row['nbeats'], row['actual'])
+            ttm_str = fmt_pred(row['ttm'])
+            ttm_r = fmt_result(row['ttm'], row['actual'])
+            ens_str = fmt_pred(row['ensemble'])
+            ens_r = fmt_result(row['ensemble'], row['actual'])
+
+            print(
+                f"{time_str:<18} {row['current_close']:>10.2f} {row['target_close']:>10.2f} {actual_str:>4} "
+                f"{dl_str:>3} {dl_r:>1} {nb_str:>3} {nb_r:>1} {ttm_str:>3} {ttm_r:>1} {ens_str:>4} {ens_r:<1}"
+            )
+
+    # ========== 每日准确率汇总表 ==========
+    print("\n\n" + "=" * 110)
+    print("每日准确率汇总 (北京时间)")
+    print("=" * 110)
+
+    header = f"{'日期':<14} {'样本':>4} {'美股':>6}  {'DLinear':>8} {'N-BEATS':>8} {'TTM-R2':>8} {'集成(投票)':>10}"
+    print(header)
+    print("-" * 110)
+
+    for date_bj, group in df_res.groupby('date_bj'):
+        total = len(group)
+        us_open = group['us_open'].iloc[0]
+        us_str = "开盘" if us_open else "休市"
+
+        dl_acc = (group['dlinear'] == group['actual']).mean()
+
+        nb_valid = group.dropna(subset=['nbeats'])
+        nb_acc = (nb_valid['nbeats'] == nb_valid['actual']).mean() if len(nb_valid) > 0 else None
+
+        ttm_valid = group.dropna(subset=['ttm'])
+        ttm_acc = (ttm_valid['ttm'] == ttm_valid['actual']).mean() if len(ttm_valid) > 0 else None
+
+        ens_valid = group.dropna(subset=['ensemble'])
+        ens_acc = (ens_valid['ensemble'] == ens_valid['actual']).mean() if len(ens_valid) > 0 else None
+
+        dl_str = f"{dl_acc:.1%}"
+        nb_str = f"{nb_acc:.1%}" if nb_acc is not None else "N/A"
+        ttm_str = f"{ttm_acc:.1%}" if ttm_acc is not None else "N/A"
+        ens_str = f"{ens_acc:.1%}" if ens_acc is not None else "N/A"
+
+        print(f"{str(date_bj):<14} {total:>4} {us_str:>6}  {dl_str:>8} {nb_str:>8} {ttm_str:>8} {ens_str:>10}")
+
+    print("-" * 110)
+
+    # --- 总准确率 ---
+    dl_total = (df_res['dlinear'] == df_res['actual']).mean()
+
+    nb_all = df_res.dropna(subset=['nbeats'])
+    nb_total = (nb_all['nbeats'] == nb_all['actual']).mean() if len(nb_all) > 0 else None
+
+    ttm_all = df_res.dropna(subset=['ttm'])
+    ttm_total = (ttm_all['ttm'] == ttm_all['actual']).mean() if len(ttm_all) > 0 else None
+
+    ens_all = df_res.dropna(subset=['ensemble'])
+    ens_total = (ens_all['ensemble'] == ens_all['actual']).mean() if len(ens_all) > 0 else None
+
+    nb_t_str = f"{nb_total:.2%}" if nb_total is not None else "N/A"
+    ttm_t_str = f"{ttm_total:.2%}" if ttm_total is not None else "N/A"
+    ens_t_str = f"{ens_total:.2%}" if ens_total is not None else "N/A"
+
+    print(f"{'总计':<14} {total_samples:>4} {'':>6}  {dl_total:.2%}  {nb_t_str:>8} {ttm_t_str:>8} {ens_t_str:>10}")
+    print("=" * 110)
+
+
+if __name__ == "__main__":
+    main()

ensemble_models_feb_2026_back.py

@@ -0,0 +1,427 @@
+"""2026年2月BTC数据 - DLinear, N-BEATS, TTM-R2 准确率测试
+
+输入数据：预测K线的前一根（即用截止到t的数据预测t+1）
+价格对比：预测结果与预测K线(t+1)的收盘价做对比
+时间：币安数据UTC，输出转为北京时间(UTC+8)，按北京时间分天
+DLinear预测逻辑：整个df一次性预处理，按索引遍历窗口
+输出：每小时详细数据、每天准确率、总准确率、三模型集成、美股开盘标记
+"""
+
+import sys
+import os
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import torch
+import numpy as np
+import pandas as pd
+import pickle
+import pandas_ta as ta
+from datetime import timedelta
+from sklearn.preprocessing import StandardScaler
+from data.fetcher import BinanceDataFetcher
+from tsfm_public.models.tinytimemixer import TinyTimeMixerForPrediction
+
+# ==================== 时区与美股判断 ====================
+
+def utc_to_beijing(ts):
+    """UTC时间转北京时间(UTC+8)"""
+    return ts + timedelta(hours=8)
+
+def is_us_stock_open_by_utc(utc_ts):
+    """根据UTC时间判断美股当天是否开盘
+
+    美股交易日基于美东时间(ET)，UTC-5(冬令时)
+    2026年2月在冬令时范围内（夏令时3月第二个周日开始）
+    所以美东 = UTC - 5h
+    """
+    et_ts = utc_ts - timedelta(hours=5)
+    et_date = pd.Timestamp(et_ts.date())
+
+    us_holidays_feb_2026 = {
+        pd.Timestamp('2026-02-16'),  # 总统日 Presidents' Day
+    }
+
+    if et_date.weekday() >= 5:
+        return False
+    if et_date in us_holidays_feb_2026:
+        return False
+    return True
+
+# ==================== 预处理器定义 ====================
+
+class ClassificationPreprocessor:
+    """DLinear分类预处理器 - pickle反序列化占位类"""
+    def __init__(self, lookback=168, forecast_horizon=1):
+        self.lookback = lookback
+        self.forecast_horizon = forecast_horizon
+        self.scaler = StandardScaler()
+        self.is_fitted = False
+
+class NBeatsPreprocessor:
+    """N-BEATS预处理器"""
+    def __init__(self, lookback=168, forecast_horizon=1):
+        self.lookback = lookback
+        self.forecast_horizon = forecast_horizon
+        self.price_scaler = StandardScaler()
+        self.is_fitted = False
+
+    def preprocess(self, df, fit=True):
+        df = df.copy()
+        prices = df['close'].values.reshape(-1, 1)
+
+        if fit:
+            scaled_prices = self.price_scaler.fit_transform(prices)
+            self.is_fitted = True
+        else:
+            scaled_prices = self.price_scaler.transform(prices)
+
+        X = []
+        valid_indices = []
+
+        for i in range(len(scaled_prices) - self.lookback):
+            X.append(scaled_prices[i:i + self.lookback, 0])
+            valid_indices.append(i + self.lookback - 1)
+
+        return np.array(X), df, valid_indices
+
+
+def fmt_pred(val):
+    """格式化预测结果"""
+    if val is None or (isinstance(val, float) and np.isnan(val)):
+        return "N/A"
+    return "涨" if int(val) == 1 else "跌"
+
+def fmt_result(pred, actual):
+    """格式化对错"""
+    if pred is None or (isinstance(pred, float) and np.isnan(pred)):
+        return "  "
+    return "✓" if int(pred) == int(actual) else "✗"
+
+
+def main():
+    print("=" * 110)
+    print("2026年2月 BTC 1h 涨跌预测准确率测试 (DLinear, N-BEATS, TTM-R2)")
+    print("输入：预测K线的前一根 | 对比：预测结果 vs 预测K线收盘价 | 时间：北京时间(UTC+8)")
+    print("DLinear预测逻辑：整个df一次性预处理，按索引遍历窗口")
+    print("=" * 110)
+
+    device = torch.device('mps' if torch.backends.mps.is_available() else 'cpu')
+    print(f"使用设备: {device}")
+
+    # ========== 步骤1: 加载模型 ==========
+    print("\n[1/4] 加载模型...")
+
+    from models.dlinear.model import DLinear
+    dlinear_checkpoint = torch.load('./models/checkpoints/dlinear_classification_2026/best_model.pt', map_location=device, weights_only=False)
+    dlinear_model = DLinear(dlinear_checkpoint['config'])
+    dlinear_model.load_state_dict(dlinear_checkpoint['model_state_dict'])
+    dlinear_model = dlinear_model.to(device).eval()
+    with open('./models/checkpoints/dlinear_classification_2026/preprocessor.pkl', 'rb') as f:
+        dlinear_preprocessor = pickle.load(f)
+    print("  ✓ DLinear已加载")
+
+    from models.nbeats.model import NBeats
+    nbeats_checkpoint = torch.load('./models/checkpoints/nbeats_btc_1h/best_model.pt', map_location=device, weights_only=False)
+    nbeats_model = NBeats(nbeats_checkpoint['config']).to(device).eval()
+    with open('./models/checkpoints/nbeats_btc_1h/preprocessor.pkl', 'rb') as f:
+        nbeats_preprocessor = pickle.load(f)
+    print("  ✓ N-BEATS已加载")
+
+    ttm_model = TinyTimeMixerForPrediction.from_pretrained(
+        "ibm-granite/granite-timeseries-ttm-r2",
+        trust_remote_code=True
+    ).to(device).eval()
+    print("  ✓ TTM-R2已加载")
+
+    # ========== 步骤2: 获取测试数据 ==========
+    print("\n[2/4] 获取2026年1月数据...")
+    fetcher = BinanceDataFetcher()
+    df = fetcher.fetch_ohlcv('BTC', '1h', start_date='2025-10-01', end_date='2026-02-01', use_cache=False)
+
+    print(f"数据范围(UTC): {df['timestamp'].min()} -> {df['timestamp'].max()}")
+    print(f"总行数: {len(df)}")
+
+    # ========== 步骤3: 预处理与预测 ==========
+    print("\n[3/4] 运行预测...")
+
+    # DLinear 整体预处理（无create_labels，整个df一次性处理）
+    dl_lookback = dlinear_preprocessor.lookback
+    df_dl = df.copy()
+    df_dl['rsi'] = ta.rsi(df_dl['close'], length=14)
+    macd = ta.macd(df_dl['close'])
+    if macd is not None and not macd.empty:
+        df_dl['macd'] = macd.iloc[:, 0]
+        df_dl['macd_hist'] = macd.iloc[:, 1]
+        df_dl['macd_signal'] = macd.iloc[:, 2]
+    df_dl = df_dl.dropna().reset_index(drop=True)
+
+    feature_cols = ['open', 'high', 'low', 'close', 'volume',
+                   'rsi', 'macd', 'macd_hist', 'macd_signal']
+    scaled_dl = dlinear_preprocessor.scaler.transform(df_dl[feature_cols])
+
+    X_dl = []
+    idx_dl = []
+    for i in range(len(scaled_dl) - dl_lookback):
+        X_dl.append(scaled_dl[i:i + dl_lookback])
+        idx_dl.append(i + dl_lookback - 1)
+    X_dl = np.array(X_dl)
+    # 构建 timestamp -> 窗口索引映射
+    dl_ts_map = {}
+    for i, idx in enumerate(idx_dl):
+        ts = df_dl.iloc[idx]['timestamp']
+        dl_ts_map[ts] = i
+
+    # N-BEATS 整体预处理（N-BEATS只用close，不受create_labels影响）
+    X_nb, df_nb, idx_nb = nbeats_preprocessor.preprocess(df, fit=False)
+
+    # TTM-R2 整体预处理
+    ttm_context_len = 512
+    scaler_ttm = StandardScaler()
+    scaled_close_ttm = scaler_ttm.fit_transform(df[['close']])
+    X_ttm = []
+    idx_ttm = []
+    for i in range(len(scaled_close_ttm) - ttm_context_len):
+        X_ttm.append(scaled_close_ttm[i:i+ttm_context_len])
+        idx_ttm.append(i + ttm_context_len - 1)
+    X_ttm = np.array(X_ttm)
+    idx_ttm_map = {v: k for k, v in enumerate(idx_ttm)}
+
+    # 找出2月（北京时间）的所有预测目标K线索引
+    # df 中每一行都是一根已收盘的K线，timestamp 是 UTC
+    jan_targets = []
+    for target_idx in range(1, len(df)):
+        target_time_utc = df.iloc[target_idx]['timestamp']
+        target_time_bj = utc_to_beijing(target_time_utc)
+        if target_time_bj.year == 2026 and target_time_bj.month == 1:
+            jan_targets.append(target_idx)
+
+    print(f"1月（北京时间）目标K线数: {len(jan_targets)}")
+
+    results = []
+
+    for target_idx in jan_targets:
+        target_time_utc = df.iloc[target_idx]['timestamp']
+        target_time_bj = utc_to_beijing(target_time_utc)
+
+        # 输入K线索引（前一根）
+        input_idx = target_idx - 1
+        current_close = df.iloc[input_idx]['close']
+        target_close = df.iloc[target_idx]['close']
+        actual_label = 1 if target_close > current_close else 0
+
+        # --- DLinear 预测（整个df一次性预处理，按索引查找窗口） ---
+        dl_pred = None
+        dl_prob = None
+        input_ts = df.iloc[input_idx]['timestamp']
+        dl_i = dl_ts_map.get(input_ts)
+        if dl_i is not None and dl_i < len(X_dl):
+            x_dl_tensor = torch.FloatTensor(X_dl[dl_i:dl_i+1]).to(device)
+            with torch.no_grad():
+                logits = dlinear_model(x_dl_tensor).cpu().numpy()
+                prob = 1 / (1 + np.exp(-logits))
+                dl_pred = int(prob.item() > 0.5)
+                dl_prob = float(prob.item())
+
+        # --- N-BEATS 预测 ---
+        nb_pred = None
+        if input_idx in idx_nb:
+            nb_i = idx_nb.index(input_idx)
+            x_nb_tensor = torch.FloatTensor(X_nb[nb_i:nb_i+1]).to(device)
+            with torch.no_grad():
+                nb_out = nbeats_model(x_nb_tensor).cpu().numpy()
+                if nb_out.ndim == 1:
+                    pred_price_scaled = nb_out[0]
+                else:
+                    pred_price_scaled = nb_out[0, 0]
+                current_price_scaled = X_nb[nb_i, -1]
+                nb_pred = 1 if pred_price_scaled > current_price_scaled else 0
+
+        # --- TTM-R2 预测 ---
+        ttm_pred = None
+        ttm_i = idx_ttm_map.get(input_idx)
+        if ttm_i is not None:
+            x_ttm_tensor = torch.FloatTensor(X_ttm[ttm_i:ttm_i+1]).to(device)
+            with torch.no_grad():
+                ttm_out = ttm_model(x_ttm_tensor)
+                ttm_pred_scaled = ttm_out.prediction_outputs[0, 0, 0].cpu().numpy()
+                ttm_current_scaled = X_ttm[ttm_i, -1, 0]
+                ttm_pred = 1 if ttm_pred_scaled > ttm_current_scaled else 0
+
+        # 集成预测（多数投票）
+        votes = [v for v in [dl_pred, nb_pred, ttm_pred] if v is not None]
+        ens_pred = (1 if sum(votes) >= 2 else 0) if len(votes) == 3 else None
+
+        results.append({
+            'time_utc': target_time_utc,
+            'time_bj': target_time_bj,
+            'current_close': current_close,
+            'target_close': target_close,
+            'actual': actual_label,
+            'dlinear': dl_pred,
+            'nbeats': nb_pred,
+            'ttm': ttm_pred,
+            'ensemble': ens_pred,
+        })
+
+    # ========== 步骤4: 输出结果 ==========
+    if not results:
+        print("没有找到符合条件的测试样本")
+        return
+
+    df_res = pd.DataFrame(results)
+    df_res['date_bj'] = df_res['time_bj'].dt.date
+    df_res['us_open'] = df_res['time_utc'].apply(is_us_stock_open_by_utc)
+
+    total_samples = len(df_res)
+
+    print(f"\n[4/4] 准确率统计 (2026年2月，北京时间)")
+    print(f"总样本数: {total_samples}")
+
+    # ========== 每小时详细数据 ==========
+    print("\n" + "=" * 110)
+    print("每小时详细数据")
+    print("=" * 110)
+
+    hourly_header = (
+        f"{'北京时间':<18} {'前收盘':>10} {'现收盘':>10} {'实际':>4} "
+        f"{'DL':>3} {'':>1} {'NB':>3} {'':>1} {'TTM':>3} {'':>1} {'集成':>4} {'':<1}"
+    )
+
+    for date_bj, day_group in df_res.groupby('date_bj'):
+        us_open = day_group['us_open'].iloc[0]
+        us_str = "美股开盘" if us_open else "美股休市"
+
+        dl_day_acc = (day_group['dlinear'] == day_group['actual']).mean()
+        nb_day_valid = day_group.dropna(subset=['nbeats'])
+        nb_day_acc = (nb_day_valid['nbeats'] == nb_day_valid['actual']).mean() if len(nb_day_valid) > 0 else None
+        ttm_day_valid = day_group.dropna(subset=['ttm'])
+        ttm_day_acc = (ttm_day_valid['ttm'] == ttm_day_valid['actual']).mean() if len(ttm_day_valid) > 0 else None
+        ens_day_valid = day_group.dropna(subset=['ensemble'])
+        ens_day_acc = (ens_day_valid['ensemble'] == ens_day_valid['actual']).mean() if len(ens_day_valid) > 0 else None
+
+        nb_acc_str = f"{nb_day_acc:.1%}" if nb_day_acc is not None else "N/A"
+        ttm_acc_str = f"{ttm_day_acc:.1%}" if ttm_day_acc is not None else "N/A"
+        ens_acc_str = f"{ens_day_acc:.1%}" if ens_day_acc is not None else "N/A"
+
+        print(f"\n--- {date_bj} ({us_str}) --- DL:{dl_day_acc:.1%} | NB:{nb_acc_str} | TTM:{ttm_acc_str} | 集成:{ens_acc_str}")
+        print(hourly_header)
+        print("-" * 110)
+
+        for _, row in day_group.iterrows():
+            time_str = row['time_bj'].strftime('%m-%d %H:%M')
+            actual_str = fmt_pred(row['actual'])
+
+            dl_str = fmt_pred(row['dlinear'])
+            dl_r = fmt_result(row['dlinear'], row['actual'])
+            nb_str = fmt_pred(row['nbeats'])
+            nb_r = fmt_result(row['nbeats'], row['actual'])
+            ttm_str = fmt_pred(row['ttm'])
+            ttm_r = fmt_result(row['ttm'], row['actual'])
+            ens_str = fmt_pred(row['ensemble'])
+            ens_r = fmt_result(row['ensemble'], row['actual'])
+
+            print(
+                f"{time_str:<18} {row['current_close']:>10.2f} {row['target_close']:>10.2f} {actual_str:>4} "
+                f"{dl_str:>3} {dl_r:>1} {nb_str:>3} {nb_r:>1} {ttm_str:>3} {ttm_r:>1} {ens_str:>4} {ens_r:<1}"
+            )
+
+    # ========== 每日准确率汇总表 ==========
+    print("\n\n" + "=" * 110)
+    print("每日准确率汇总 (北京时间)")
+    print("=" * 110)
+
+    header = f"{'日期':<14} {'样本':>4} {'美股':>6}  {'DLinear':>8} {'N-BEATS':>8} {'TTM-R2':>8} {'集成(投票)':>10}"
+    print(header)
+    print("-" * 110)
+
+    for date_bj, group in df_res.groupby('date_bj'):
+        total = len(group)
+        us_open = group['us_open'].iloc[0]
+        us_str = "开盘" if us_open else "休市"
+
+        dl_acc = (group['dlinear'] == group['actual']).mean()
+
+        nb_valid = group.dropna(subset=['nbeats'])
+        nb_acc = (nb_valid['nbeats'] == nb_valid['actual']).mean() if len(nb_valid) > 0 else None
+
+        ttm_valid = group.dropna(subset=['ttm'])
+        ttm_acc = (ttm_valid['ttm'] == ttm_valid['actual']).mean() if len(ttm_valid) > 0 else None
+
+        ens_valid = group.dropna(subset=['ensemble'])
+        ens_acc = (ens_valid['ensemble'] == ens_valid['actual']).mean() if len(ens_valid) > 0 else None
+
+        dl_str = f"{dl_acc:.1%}"
+        nb_str = f"{nb_acc:.1%}" if nb_acc is not None else "N/A"
+        ttm_str = f"{ttm_acc:.1%}" if ttm_acc is not None else "N/A"
+        ens_str = f"{ens_acc:.1%}" if ens_acc is not None else "N/A"
+
+        print(f"{str(date_bj):<14} {total:>4} {us_str:>6}  {dl_str:>8} {nb_str:>8} {ttm_str:>8} {ens_str:>10}")
+
+    print("-" * 110)
+
+    # --- 总准确率 ---
+    dl_total = (df_res['dlinear'] == df_res['actual']).mean()
+
+    nb_all = df_res.dropna(subset=['nbeats'])
+    nb_total = (nb_all['nbeats'] == nb_all['actual']).mean() if len(nb_all) > 0 else None
+
+    ttm_all = df_res.dropna(subset=['ttm'])
+    ttm_total = (ttm_all['ttm'] == ttm_all['actual']).mean() if len(ttm_all) > 0 else None
+
+    ens_all = df_res.dropna(subset=['ensemble'])
+    ens_total = (ens_all['ensemble'] == ens_all['actual']).mean() if len(ens_all) > 0 else None
+
+    nb_t_str = f"{nb_total:.2%}" if nb_total is not None else "N/A"
+    ttm_t_str = f"{ttm_total:.2%}" if ttm_total is not None else "N/A"
+    ens_t_str = f"{ens_total:.2%}" if ens_total is not None else "N/A"
+
+    print(f"{'总计':<14} {total_samples:>4} {'':>6}  {dl_total:.2%}  {nb_t_str:>8} {ttm_t_str:>8} {ens_t_str:>10}")
+    print("=" * 110)
+
+    # --- 美股开盘日 vs 休市日对比 ---
+    print("\n[美股开盘日 vs 休市日 准确率对比]")
+    print("-" * 80)
+    for label, flag in [("开盘日", True), ("休市日", False)]:
+        subset = df_res[df_res['us_open'] == flag]
+        if len(subset) == 0:
+            continue
+        n = len(subset)
+        dl_a = (subset['dlinear'] == subset['actual']).mean()
+
+        nb_s = subset.dropna(subset=['nbeats'])
+        nb_a = (nb_s['nbeats'] == nb_s['actual']).mean() if len(nb_s) > 0 else None
+
+        ttm_s = subset.dropna(subset=['ttm'])
+        ttm_a = (ttm_s['ttm'] == ttm_s['actual']).mean() if len(ttm_s) > 0 else None
+
+        ens_s = subset.dropna(subset=['ensemble'])
+        ens_a = (ens_s['ensemble'] == ens_s['actual']).mean() if len(ens_s) > 0 else None
+
+        nb_str = f"{nb_a:.2%}" if nb_a is not None else "N/A"
+        ttm_str = f"{ttm_a:.2%}" if ttm_a is not None else "N/A"
+        ens_str = f"{ens_a:.2%}" if ens_a is not None else "N/A"
+
+        print(f"  {label} ({n}样本): DLinear={dl_a:.2%}  N-BEATS={nb_str}  TTM-R2={ttm_str}  集成={ens_str}")
+
+    print("-" * 80)
+
+    # --- 预测分布统计 ---
+    print("\n[预测分布统计]")
+    print("-" * 80)
+    actual_up = (df_res['actual'] == 1).sum()
+    actual_down = (df_res['actual'] == 0).sum()
+    print(f"  实际: 上涨={actual_up} ({actual_up/total_samples:.1%})  下跌={actual_down} ({actual_down/total_samples:.1%})")
+
+    for name, col in [("DLinear", "dlinear"), ("N-BEATS", "nbeats"), ("TTM-R2", "ttm"), ("集成", "ensemble")]:
+        valid = df_res[col].dropna()
+        if len(valid) == 0:
+            continue
+        up = (valid == 1).sum()
+        down = (valid == 0).sum()
+        print(f"  {name}: 预测上涨={up} ({up/len(valid):.1%})  预测下跌={down} ({down/len(valid):.1%})")
+
+    print("=" * 110)
+
+
+if __name__ == "__main__":
+    main()

ensemble_models_feb_2026_back_onnx.py

@@ -0,0 +1,477 @@
+"""2026年2月BTC数据 - DLinear, N-BEATS, TTM-R2 准确率测试 (ONNX版)
+
+输入数据：预测K线的前一根（即用截止到t的数据预测t+1）
+价格对比：预测结果与预测K线(t+1)的收盘价做对比
+时间：币安数据UTC，输出转为北京时间(UTC+8)，按北京时间分天
+DLinear预测逻辑：整个df一次性预处理，按索引遍历窗口
+输出：每小时详细数据、每天准确率、总准确率、三模型集成、美股开盘标记
+"""
+
+import sys
+import os
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import numpy as np
+import pandas as pd
+import pickle
+import pandas_ta as ta
+import onnxruntime as ort
+from datetime import timedelta
+from sklearn.preprocessing import StandardScaler
+from data.fetcher import BinanceDataFetcher
+
+# ==================== 时区与美股判断 ====================
+
+def utc_to_beijing(ts):
+    """UTC时间转北京时间(UTC+8)"""
+    return ts + timedelta(hours=8)
+
+def is_us_stock_open_by_utc(utc_ts):
+    """根据UTC时间判断美股当天是否开盘
+
+    美股交易日基于美东时间(ET)，UTC-5(冬令时)
+    2026年2月在冬令时范围内（夏令时3月第二个周日开始）
+    所以美东 = UTC - 5h
+    """
+    et_ts = utc_ts - timedelta(hours=5)
+    et_date = pd.Timestamp(et_ts.date())
+
+    us_holidays_feb_2026 = {
+        pd.Timestamp('2026-02-16'),  # 总统日 Presidents' Day
+    }
+
+    if et_date.weekday() >= 5:
+        return False
+    if et_date in us_holidays_feb_2026:
+        return False
+    return True
+
+# ==================== 预处理器定义 ====================
+
+class ClassificationPreprocessor:
+    """DLinear分类预处理器 - pickle反序列化占位类"""
+    def __init__(self, lookback=168, forecast_horizon=1):
+        self.lookback = lookback
+        self.forecast_horizon = forecast_horizon
+        self.scaler = StandardScaler()
+        self.is_fitted = False
+
+class NBeatsPreprocessor:
+    """N-BEATS预处理器"""
+    def __init__(self, lookback=168, forecast_horizon=1):
+        self.lookback = lookback
+        self.forecast_horizon = forecast_horizon
+        self.price_scaler = StandardScaler()
+        self.is_fitted = False
+
+    def preprocess(self, df, fit=True):
+        df = df.copy()
+        prices = df['close'].values.reshape(-1, 1)
+
+        if fit:
+            scaled_prices = self.price_scaler.fit_transform(prices)
+            self.is_fitted = True
+        else:
+            scaled_prices = self.price_scaler.transform(prices)
+
+        X = []
+        valid_indices = []
+
+        for i in range(len(scaled_prices) - self.lookback):
+            X.append(scaled_prices[i:i + self.lookback, 0])
+            valid_indices.append(i + self.lookback - 1)
+
+        return np.array(X), df, valid_indices
+
+
+def fmt_pred(val):
+    """格式化预测结果"""
+    if val is None or (isinstance(val, float) and np.isnan(val)):
+        return "N/A"
+    return "涨" if int(val) == 1 else "跌"
+
+def fmt_result(pred, actual):
+    """格式化对错"""
+    if pred is None or (isinstance(pred, float) and np.isnan(pred)):
+        return "  "
+    return "✓" if int(pred) == int(actual) else "✗"
+
+
+def main():
+    print("=" * 110)
+    print("2026年2月 BTC 1h 涨跌预测准确率测试 (DLinear, N-BEATS, TTM-R2) - ONNX Version")
+    print("输入：预测K线的前一根 | 对比：预测结果 vs 预测K线收盘价 | 时间：北京时间(UTC+8)")
+    print("DLinear预测逻辑：整个df一次性预处理，按索引遍历窗口")
+    print("=" * 110)
+
+    # ========== 步骤1: 加载模型 ==========
+    print("\n[1/4] 加载 ONNX 模型...")
+
+    # Load DLinear ONNX
+    try:
+        dlinear_session = ort.InferenceSession('./models/checkpoints/dlinear_classification_2026/model.onnx')
+        print("  ✓ DLinear (ONNX) 已加载")
+        
+        with open('./models/checkpoints/dlinear_classification_2026/preprocessor.pkl', 'rb') as f:
+            dlinear_preprocessor = pickle.load(f)
+    except Exception as e:
+        print(f"  ✗ DLinear 加载失败: {e}")
+        return
+
+    # Load N-BEATS ONNX
+    try:
+        nbeats_session = ort.InferenceSession('./models/checkpoints/nbeats_btc_1h/model.onnx')
+        print("  ✓ N-BEATS (ONNX) 已加载")
+        
+        with open('./models/checkpoints/nbeats_btc_1h/preprocessor.pkl', 'rb') as f:
+            nbeats_preprocessor = pickle.load(f)
+    except Exception as e:
+        print(f"  ✗ N-BEATS 加载失败: {e}")
+        return
+
+    # Load TTM-R2 ONNX
+    ttm_session = None
+    try:
+        temp_session = ort.InferenceSession('./models/checkpoints/ttm_r2_onnx/ttm_r2_crypto_channel_mixing.onnx')
+        
+        # Test inference with dummy input
+        print("  Testing TTM-R2 ONNX inference...")
+        dummy_input = np.random.randn(1, 512, 1).astype(np.float32)
+        ort_inputs = {}
+        input_names = [i.name for i in temp_session.get_inputs()]
+        print(f"Model inputs: {input_names}")  
+        # 根据 session 中的名称映射输入
+        for input_meta in temp_session.get_inputs():
+            name = input_meta.name
+            if 'past_values' in name or name == 'input':
+                ort_inputs[name] = dummy_input
+            elif 'mask' in name:
+                # 创建全 1 mask
+                mask = np.ones_like(dummy_input)
+                # 检查期望的类型
+                if input_meta.type == 'tensor(int64)':
+                    mask = mask.astype(np.int64)
+                else:
+                    mask = mask.astype(np.float32)
+                ort_inputs[name] = mask
+        print(f"Session run input keys: {list(ort_inputs.keys())}")
+        for k, v in ort_inputs.items():
+            print(f"Input {k} shape: {v.shape}, dtype: {v.dtype}")
+                
+        temp_session.run(None, ort_inputs)
+        
+        ttm_session = temp_session
+        print("  ✓ TTM-R2 (ONNX) 已加载且可用 (Optimizations Disabled)")
+    except Exception as e:
+        print(f"  ⚠ TTM-R2 (ONNX) 加载或推理失败: {e}")
+        print("    将在本次测试中跳过 TTM-R2 模型")
+        ttm_session = None
+
+    # ========== 步骤2: 获取测试数据 ==========
+    print("\n[2/4] 获取2026年2月数据...")
+    fetcher = BinanceDataFetcher()
+    df = fetcher.fetch_ohlcv('BTC', '1h', start_date='2025-10-01', end_date='2026-03-01', use_cache=False)
+
+    print(f"数据范围(UTC): {df['timestamp'].min()} -> {df['timestamp'].max()}")
+    print(f"总行数: {len(df)}")
+
+    # ========== 步骤3: 预处理与预测 ==========
+    print("\n[3/4] 运行预测...")
+
+    # DLinear 整体预处理（无create_labels，整个df一次性处理）
+    dl_lookback = dlinear_preprocessor.lookback
+    df_dl = df.copy()
+    df_dl['rsi'] = ta.rsi(df_dl['close'], length=14)
+    macd = ta.macd(df_dl['close'])
+    if macd is not None and not macd.empty:
+        df_dl['macd'] = macd.iloc[:, 0]
+        df_dl['macd_hist'] = macd.iloc[:, 1]
+        df_dl['macd_signal'] = macd.iloc[:, 2]
+    df_dl = df_dl.dropna().reset_index(drop=True)
+
+    feature_cols = ['open', 'high', 'low', 'close', 'volume',
+                   'rsi', 'macd', 'macd_hist', 'macd_signal']
+    scaled_dl = dlinear_preprocessor.scaler.transform(df_dl[feature_cols])
+
+    X_dl = []
+    idx_dl = []
+    for i in range(len(scaled_dl) - dl_lookback):
+        X_dl.append(scaled_dl[i:i + dl_lookback])
+        idx_dl.append(i + dl_lookback - 1)
+    X_dl = np.array(X_dl, dtype=np.float32)
+    # 构建 timestamp -> 窗口索引映射
+    dl_ts_map = {}
+    for i, idx in enumerate(idx_dl):
+        ts = df_dl.iloc[idx]['timestamp']
+        dl_ts_map[ts] = i
+
+    # N-BEATS 整体预处理（N-BEATS只用close，不受create_labels影响）
+    X_nb, df_nb, idx_nb = nbeats_preprocessor.preprocess(df, fit=False)
+    X_nb = X_nb.astype(np.float32) # Ensure float32
+
+    # TTM-R2 整体预处理
+    ttm_context_len = 512
+    scaler_ttm = StandardScaler()
+    scaled_close_ttm = scaler_ttm.fit_transform(df[['close']])
+    X_ttm = []
+    idx_ttm = []
+    for i in range(len(scaled_close_ttm) - ttm_context_len):
+        X_ttm.append(scaled_close_ttm[i:i+ttm_context_len])
+        idx_ttm.append(i + ttm_context_len - 1)
+    X_ttm = np.array(X_ttm, dtype=np.float32)
+    idx_ttm_map = {v: k for k, v in enumerate(idx_ttm)}
+
+    # 找出2月（北京时间）的所有预测目标K线索引
+    # df 中每一行都是一根已收盘的K线，timestamp 是 UTC
+    jan_targets = []
+    for target_idx in range(1, len(df)):
+        target_time_utc = df.iloc[target_idx]['timestamp']
+        target_time_bj = utc_to_beijing(target_time_utc)
+        if target_time_bj.year == 2026 and target_time_bj.month == 2:
+            jan_targets.append(target_idx)
+
+    print(f"2月（北京时间）目标K线数: {len(jan_targets)}")
+
+    results = []
+
+    for target_idx in jan_targets:
+        target_time_utc = df.iloc[target_idx]['timestamp']
+        target_time_bj = utc_to_beijing(target_time_utc)
+
+        # 输入K线索引（前一根）
+        input_idx = target_idx - 1
+        current_close = df.iloc[input_idx]['close']
+        target_close = df.iloc[target_idx]['close']
+        actual_label = 1 if target_close > current_close else 0
+
+        # --- DLinear 预测 (ONNX) ---
+        dl_pred = None
+        dl_prob = None
+        input_ts = df.iloc[input_idx]['timestamp']
+        dl_i = dl_ts_map.get(input_ts)
+        if dl_i is not None and dl_i < len(X_dl):
+            x_dl_input = X_dl[dl_i:dl_i+1] # (1, 168, 9)
+            
+            # ONNX Inference
+            # Input name: 'input'
+            logits = dlinear_session.run(['output'], {'input': x_dl_input})[0]
+            
+            prob = 1 / (1 + np.exp(-logits))
+            dl_pred = int(prob.item() > 0.5)
+            dl_prob = float(prob.item())
+
+        # --- N-BEATS 预测 (ONNX) ---
+        nb_pred = None
+        if input_idx in idx_nb:
+            nb_i = idx_nb.index(input_idx)
+            x_nb_input = X_nb[nb_i:nb_i+1] # (1, 168)
+            
+            # ONNX Inference
+            # Input name: 'input'
+            nb_out = nbeats_session.run(['forecast'], {'input': x_nb_input})[0]
+            
+            if nb_out.ndim == 1:
+                pred_price_scaled = nb_out[0]
+            else:
+                pred_price_scaled = nb_out[0, 0]
+            current_price_scaled = X_nb[nb_i, -1]
+            nb_pred = 1 if pred_price_scaled > current_price_scaled else 0
+
+        # --- TTM-R2 预测 (ONNX) ---
+        ttm_pred = None
+        ttm_i = idx_ttm_map.get(input_idx)
+        if ttm_i is not None and ttm_session is not None:
+            x_ttm_input = X_ttm[ttm_i:ttm_i+1] # (1, 512, 1)
+            
+            # Debug shape
+            # print(f"TTM input shape: {x_ttm_input.shape}, dtype: {x_ttm_input.dtype}")
+
+            # ONNX Inference
+            # Input name: 'past_values'
+            # Try transposing to (1, 1, 512) just in case
+            # x_ttm_input = x_ttm_input.transpose(0, 2, 1)
+            
+            ttm_out = ttm_session.run(None, {'past_values': x_ttm_input})[0]
+            
+            ttm_pred_scaled = ttm_out[0, 0, 0]
+            ttm_current_scaled = X_ttm[ttm_i, -1, 0]
+            ttm_pred = 1 if ttm_pred_scaled > ttm_current_scaled else 0
+
+        # 集成预测（多数投票）
+        votes = [v for v in [dl_pred, nb_pred, ttm_pred] if v is not None]
+        ens_pred = (1 if sum(votes) >= 2 else 0) if len(votes) == 3 else None
+
+        results.append({
+            'time_utc': target_time_utc,
+            'time_bj': target_time_bj,
+            'current_close': current_close,
+            'target_close': target_close,
+            'actual': actual_label,
+            'dlinear': dl_pred,
+            'nbeats': nb_pred,
+            'ttm': ttm_pred,
+            'ensemble': ens_pred,
+        })
+
+    # ========== 步骤4: 输出结果 ==========
+    if not results:
+        print("没有找到符合条件的测试样本")
+        return
+
+    df_res = pd.DataFrame(results)
+    df_res['date_bj'] = df_res['time_bj'].dt.date
+    df_res['us_open'] = df_res['time_utc'].apply(is_us_stock_open_by_utc)
+
+    total_samples = len(df_res)
+
+    print(f"\n[4/4] 准确率统计 (2026年2月，北京时间)")
+    print(f"总样本数: {total_samples}")
+
+    # ========== 每小时详细数据 ==========
+    print("\n" + "=" * 110)
+    print("每小时详细数据")
+    print("=" * 110)
+
+    hourly_header = (
+        f"{'北京时间':<18} {'前收盘':>10} {'现收盘':>10} {'实际':>4} "
+        f"{'DL':>3} {'':>1} {'NB':>3} {'':>1} {'TTM':>3} {'':>1} {'集成':>4} {'':<1}"
+    )
+
+    for date_bj, day_group in df_res.groupby('date_bj'):
+        us_open = day_group['us_open'].iloc[0]
+        us_str = "美股开盘" if us_open else "美股休市"
+
+        dl_day_acc = (day_group['dlinear'] == day_group['actual']).mean()
+        nb_day_valid = day_group.dropna(subset=['nbeats'])
+        nb_day_acc = (nb_day_valid['nbeats'] == nb_day_valid['actual']).mean() if len(nb_day_valid) > 0 else None
+        ttm_day_valid = day_group.dropna(subset=['ttm'])
+        ttm_day_acc = (ttm_day_valid['ttm'] == ttm_day_valid['actual']).mean() if len(ttm_day_valid) > 0 else None
+        ens_day_valid = day_group.dropna(subset=['ensemble'])
+        ens_day_acc = (ens_day_valid['ensemble'] == ens_day_valid['actual']).mean() if len(ens_day_valid) > 0 else None
+
+        nb_acc_str = f"{nb_day_acc:.1%}" if nb_day_acc is not None else "N/A"
+        ttm_acc_str = f"{ttm_day_acc:.1%}" if ttm_day_acc is not None else "N/A"
+        ens_acc_str = f"{ens_day_acc:.1%}" if ens_day_acc is not None else "N/A"
+
+        print(f"\n--- {date_bj} ({us_str}) --- DL:{dl_day_acc:.1%} | NB:{nb_acc_str} | TTM:{ttm_acc_str} | 集成:{ens_acc_str}")
+        print(hourly_header)
+        print("-" * 110)
+
+        for _, row in day_group.iterrows():
+            time_str = row['time_bj'].strftime('%m-%d %H:%M')
+            actual_str = fmt_pred(row['actual'])
+
+            dl_str = fmt_pred(row['dlinear'])
+            dl_r = fmt_result(row['dlinear'], row['actual'])
+            nb_str = fmt_pred(row['nbeats'])
+            nb_r = fmt_result(row['nbeats'], row['actual'])
+            ttm_str = fmt_pred(row['ttm'])
+            ttm_r = fmt_result(row['ttm'], row['actual'])
+            ens_str = fmt_pred(row['ensemble'])
+            ens_r = fmt_result(row['ensemble'], row['actual'])
+
+            print(
+                f"{time_str:<18} {row['current_close']:>10.2f} {row['target_close']:>10.2f} {actual_str:>4} "
+                f"{dl_str:>3} {dl_r:>1} {nb_str:>3} {nb_r:>1} {ttm_str:>3} {ttm_r:>1} {ens_str:>4} {ens_r:<1}"
+            )
+
+    # ========== 每日准确率汇总表 ==========
+    print("\n\n" + "=" * 110)
+    print("每日准确率汇总 (北京时间)")
+    print("=" * 110)
+
+    header = f"{'日期':<14} {'样本':>4} {'美股':>6}  {'DLinear':>8} {'N-BEATS':>8} {'TTM-R2':>8} {'集成(投票)':>10}"
+    print(header)
+    print("-" * 110)
+
+    for date_bj, group in df_res.groupby('date_bj'):
+        total = len(group)
+        us_open = group['us_open'].iloc[0]
+        us_str = "开盘" if us_open else "休市"
+
+        dl_acc = (group['dlinear'] == group['actual']).mean()
+
+        nb_valid = group.dropna(subset=['nbeats'])
+        nb_acc = (nb_valid['nbeats'] == nb_valid['actual']).mean() if len(nb_valid) > 0 else None
+
+        ttm_valid = group.dropna(subset=['ttm'])
+        ttm_acc = (ttm_valid['ttm'] == ttm_valid['actual']).mean() if len(ttm_valid) > 0 else None
+
+        ens_valid = group.dropna(subset=['ensemble'])
+        ens_acc = (ens_valid['ensemble'] == ens_valid['actual']).mean() if len(ens_valid) > 0 else None
+
+        dl_str = f"{dl_acc:.1%}"
+        nb_str = f"{nb_acc:.1%}" if nb_acc is not None else "N/A"
+        ttm_str = f"{ttm_acc:.1%}" if ttm_acc is not None else "N/A"
+        ens_str = f"{ens_acc:.1%}" if ens_acc is not None else "N/A"
+
+        print(f"{str(date_bj):<14} {total:>4} {us_str:>6}  {dl_str:>8} {nb_str:>8} {ttm_str:>8} {ens_str:>10}")
+
+    print("-" * 110)
+
+    # --- 总准确率 ---
+    dl_total = (df_res['dlinear'] == df_res['actual']).mean()
+
+    nb_all = df_res.dropna(subset=['nbeats'])
+    nb_total = (nb_all['nbeats'] == nb_all['actual']).mean() if len(nb_all) > 0 else None
+
+    ttm_all = df_res.dropna(subset=['ttm'])
+    ttm_total = (ttm_all['ttm'] == ttm_all['actual']).mean() if len(ttm_all) > 0 else None
+
+    ens_all = df_res.dropna(subset=['ensemble'])
+    ens_total = (ens_all['ensemble'] == ens_all['actual']).mean() if len(ens_all) > 0 else None
+
+    nb_t_str = f"{nb_total:.2%}" if nb_total is not None else "N/A"
+    ttm_t_str = f"{ttm_total:.2%}" if ttm_total is not None else "N/A"
+    ens_t_str = f"{ens_total:.2%}" if ens_total is not None else "N/A"
+
+    print(f"{'总计':<14} {total_samples:>4} {'':>6}  {dl_total:.2%}  {nb_t_str:>8} {ttm_t_str:>8} {ens_t_str:>10}")
+    print("=" * 110)
+
+    # --- 美股开盘日 vs 休市日对比 ---
+    print("\n[美股开盘日 vs 休市日 准确率对比]")
+    print("-" * 80)
+    for label, flag in [("开盘日", True), ("休市日", False)]:
+        subset = df_res[df_res['us_open'] == flag]
+        if len(subset) == 0:
+            continue
+        n = len(subset)
+        dl_a = (subset['dlinear'] == subset['actual']).mean()
+
+        nb_s = subset.dropna(subset=['nbeats'])
+        nb_a = (nb_s['nbeats'] == nb_s['actual']).mean() if len(nb_s) > 0 else None
+
+        ttm_s = subset.dropna(subset=['ttm'])
+        ttm_a = (ttm_s['ttm'] == ttm_s['actual']).mean() if len(ttm_s) > 0 else None
+
+        ens_s = subset.dropna(subset=['ensemble'])
+        ens_a = (ens_s['ensemble'] == ens_s['actual']).mean() if len(ens_s) > 0 else None
+
+        nb_str = f"{nb_a:.2%}" if nb_a is not None else "N/A"
+        ttm_str = f"{ttm_a:.2%}" if ttm_a is not None else "N/A"
+        ens_str = f"{ens_a:.2%}" if ens_a is not None else "N/A"
+
+        print(f"  {label} ({n}样本): DLinear={dl_a:.2%}  N-BEATS={nb_str}  TTM-R2={ttm_str}  集成={ens_str}")
+
+    print("-" * 80)
+
+    # --- 预测分布统计 ---
+    print("\n[预测分布统计]")
+    print("-" * 80)
+    actual_up = (df_res['actual'] == 1).sum()
+    actual_down = (df_res['actual'] == 0).sum()
+    print(f"  实际: 上涨={actual_up} ({actual_up/total_samples:.1%})  下跌={actual_down} ({actual_down/total_samples:.1%})")
+
+    for name, col in [("DLinear", "dlinear"), ("N-BEATS", "nbeats"), ("TTM-R2", "ttm"), ("集成", "ensemble")]:
+        valid = df_res[col].dropna()
+        if len(valid) == 0:
+            continue
+        up = (valid == 1).sum()
+        down = (valid == 0).sum()
+        print(f"  {name}: 预测上涨={up} ({up/len(valid):.1%})  预测下跌={down} ({down/len(valid):.1%})")
+
+    print("=" * 110)
+
+
+if __name__ == "__main__":
+    main()

ensemble_models_feb_2026_deepseek.py

@@ -0,0 +1,384 @@
+"""2026年2月BTC数据 - DLinear, N-BEATS, TTM-R2, DeepSeek-R1(高并发版) 准确率测试
+
+输入数据：预测K线的前一根（即用截止到t的数据预测t+1）
+价格对比：预测结果与预测K线(t+1)的收盘价做对比
+时间：币安数据UTC，输出转为北京时间(UTC+8)，按北京时间分天
+输出：每小时详细数据、每天准确率、总准确率、四模型集成、美股开盘标记
+"""
+
+import sys
+import os
+import json
+import re
+import time
+import asyncio
+import aiohttp
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import torch
+import numpy as np
+import pandas as pd
+import pickle
+import pandas_ta as ta
+from datetime import timedelta
+from sklearn.preprocessing import StandardScaler
+from data.fetcher import BinanceDataFetcher
+from tsfm_public.models.tinytimemixer import TinyTimeMixerForPrediction
+
+# ==================== 时区与美股判断 ====================
+
+def utc_to_beijing(ts):
+    return ts + timedelta(hours=8)
+
+def is_us_stock_open_by_utc(utc_ts):
+    et_ts = utc_ts - timedelta(hours=5)
+    et_date = pd.Timestamp(et_ts.date())
+    us_holidays_feb_2026 = {pd.Timestamp('2026-02-16')}
+    if et_date.weekday() >= 5 or et_date in us_holidays_feb_2026:
+        return False
+    return True
+
+# ==================== 预处理器定义 ====================
+
+class ClassificationPreprocessor:
+    def __init__(self, lookback=168, forecast_horizon=1):
+        self.lookback = lookback
+        self.forecast_horizon = forecast_horizon
+        self.scaler = StandardScaler()
+        self.is_fitted = False
+
+class NBeatsPreprocessor:
+    def __init__(self, lookback=168, forecast_horizon=1):
+        self.lookback = lookback
+        self.forecast_horizon = forecast_horizon
+        self.price_scaler = StandardScaler()
+        self.is_fitted = False
+
+    def preprocess(self, df, fit=True):
+        df = df.copy()
+        prices = df['close'].values.reshape(-1, 1)
+        if fit:
+            scaled_prices = self.price_scaler.fit_transform(prices)
+            self.is_fitted = True
+        else:
+            scaled_prices = self.price_scaler.transform(prices)
+        X, valid_indices = [], []
+        for i in range(len(scaled_prices) - self.lookback):
+            X.append(scaled_prices[i:i + self.lookback, 0])
+            valid_indices.append(i + self.lookback - 1)
+        return np.array(X), df, valid_indices
+
+def fmt_pred(val):
+    if val is None or (isinstance(val, float) and np.isnan(val)): return "N/A"
+    return "涨" if int(val) == 1 else "跌"
+
+def fmt_result(pred, actual):
+    if pred is None or (isinstance(pred, float) and np.isnan(pred)): return "  "
+    return "✓" if int(pred) == int(actual) else "✗"
+
+
+from datetime import datetime
+import json
+import re
+import time
+import asyncio
+from openai import AsyncOpenAI
+import openai
+
+# ==================== DeepSeek 异步并发核心逻辑 (双重锁绝对防 429) ====================
+
+def get_log_time():
+    return datetime.now().strftime("%H:%M:%S")
+
+async def fetch_deepseek_single(client, messages, item_id, rate_lock, global_state, sem):
+    """
+    单次异步API调用：并发数控制 + 漏桶控速 + 指数退避重试
+    """
+    attempt = 0
+    while True:
+        attempt += 1
+        
+        # 0. 并发锁：保证同一时刻，最多只有 N 个请求在英伟达服务器上运行
+        async with sem:
+            # 1. 速率锁：保证就算允许并发，发包的瞬间也要间隔至少 2.1 秒 (约 28次/分钟，留足余量)
+            async with rate_lock:
+                now = time.time()
+                elapsed = now - global_state['last_call_time']
+                if elapsed < 2.1:
+                    await asyncio.sleep(2.1 - elapsed)
+                global_state['last_call_time'] = time.time()
+                
+            print(f"[{get_log_time()}] 🚀 [任务 {item_id}] 发起请求 (第 {attempt} 次尝试)...")
+                
+            try:
+                # 2. 调用 API (注意：请确保 model 是你刚才探测到的最新可用模型名)
+                completion = await client.chat.completions.create(
+                    model="deepseek-ai/deepseek-v3.2", # <--- 如果旧名字报错410，请务必替换为最新名字
+                    messages=messages,
+                    temperature=0.6,
+                    top_p=0.7,
+                    max_tokens=1024,
+                    timeout=60.0
+                )
+                
+                content = completion.choices[0].message.content
+                match = re.search(r'\{.*\}', content, re.DOTALL)
+                
+                if match:
+                    res = json.loads(match.group(0))
+                    pred = res.get('prediction')
+                    
+                    if pred in [0, 1]:
+                        pred_str = "涨" if pred == 1 else "跌"
+                        print(f"[{get_log_time()}] ✅ [任务 {item_id}] 成功获取预测: {pred_str}")
+                        return item_id, int(pred)
+                    else:
+                        print(f"[{get_log_time()}] ⚠️ [任务 {item_id}] 异常预测值: {pred}，重新请求...")
+                        continue
+                else:
+                    print(f"[{get_log_time()}] ⚠️ [任务 {item_id}] JSON 解析失败，重新请求...")
+                    continue
+
+            except openai.RateLimitError:
+                # 指数退避：第一次等 5 秒，第二次等 10 秒，第三次等 20 秒...
+                wait_time = 5 * (2 ** (attempt - 1))
+                wait_time = min(wait_time, 60) # 最多等 60 秒
+                print(f"[{get_log_time()}] 🛑 [任务 {item_id}] 触发限流 (429)，休眠 {wait_time} 秒后重试...")
+                await asyncio.sleep(wait_time)
+            except openai.APITimeoutError:
+                print(f"[{get_log_time()}] ⏳ [任务 {item_id}] 请求超时 (>60s)，重新请求...")
+                await asyncio.sleep(2)
+            except openai.APIStatusError as e:
+                if e.status_code in [400, 401, 403, 404, 410]:
+                    print(f"[{get_log_time()}] 🚨 [任务 {item_id}] 致命错误 (HTTP {e.status_code}): {e.message}。放弃。")
+                    return item_id, None
+                print(f"[{get_log_time()}] ❌ [任务 {item_id}] API 报错 (HTTP {e.status_code})，休眠 3 秒后重试...")
+                await asyncio.sleep(3)
+            except Exception as e:
+                print(f"[{get_log_time()}] 💥 [任务 {item_id}] 未知错误: {e}，休眠 3 秒后重新请求...")
+                await asyncio.sleep(3)
+
+
+async def run_all_deepseek_async(tasks_data, api_key):
+    """统筹所有的异步请求"""
+    total = len(tasks_data)
+    
+    # === 核心参数配置 ===
+    MAX_CONCURRENT = 8 # 绝对并发上限：同一时刻最多 2 个请求在处理
+    
+    print("\n" + "="*80)
+    print(f"[{get_log_time()}] 开始 DeepSeek 异步并发请求 (双重防 429 版)")
+    print(f"总任务数: {total} 个 | 严格并发数: {MAX_CONCURRENT} | 发包间隔: 2.1 秒")
+    print("="*80 + "\n")
+    
+    client = AsyncOpenAI(
+        base_url="https://integrate.api.nvidia.com/v1",
+        api_key=api_key
+    )
+    
+    sem = asyncio.Semaphore(MAX_CONCURRENT)
+    rate_lock = asyncio.Lock()
+    global_state = {'last_call_time': 0.0}
+    ds_results_dict = {}
+    
+    tasks = []
+    for item in tasks_data:
+        task = asyncio.create_task(
+            fetch_deepseek_single(client, item['messages'], item['id'], rate_lock, global_state, sem)
+        )
+        tasks.append(task)
+        
+    completed = 0
+    for coro in asyncio.as_completed(tasks):
+        item_id, pred = await coro
+        ds_results_dict[item_id] = pred
+        completed += 1
+        
+        if completed % 10 == 0 or completed == total:
+            print(f"\n[{get_log_time()}] 📊 【全局进度】 DeepSeek 已完成: {completed}/{total} ({(completed/total):.1%})\n")
+            
+    return ds_results_dict
+# ==================== 主程序 ====================
+
+def main():
+    print("=" * 115)
+    print("BTC 1h 涨跌预测准确率测试 (DLinear, N-BEATS, TTM-R2, DeepSeek-R1 异步版)")
+    print("=" * 115)
+
+    device = torch.device('mps' if torch.backends.mps.is_available() else 'cpu')
+    print(f"使用设备: {device}")
+
+    # ========== 步骤1: 加载模型 ==========
+    print("\n[1/4] 加载本地模型...")
+    from models.dlinear.model import DLinear
+    dlinear_checkpoint = torch.load('./models/checkpoints/dlinear_classification_2026/best_model.pt', map_location=device, weights_only=False)
+    dlinear_model = DLinear(dlinear_checkpoint['config']).to(device).eval()
+    dlinear_model.load_state_dict(dlinear_checkpoint['model_state_dict'])
+    with open('./models/checkpoints/dlinear_classification_2026/preprocessor.pkl', 'rb') as f:
+        dlinear_preprocessor = pickle.load(f)
+
+    from models.nbeats.model import NBeats
+    nbeats_checkpoint = torch.load('./models/checkpoints/nbeats_btc_1h/best_model.pt', map_location=device, weights_only=False)
+    nbeats_model = NBeats(nbeats_checkpoint['config']).to(device).eval()
+    with open('./models/checkpoints/nbeats_btc_1h/preprocessor.pkl', 'rb') as f:
+        nbeats_preprocessor = pickle.load(f)
+
+    ttm_model = TinyTimeMixerForPrediction.from_pretrained("ibm-granite/granite-timeseries-ttm-r2", trust_remote_code=True).to(device).eval()
+
+    ds_api_key = os.environ.get("NVIDIA_API_KEY", "")
+    if not ds_api_key:
+        print("  ⚠️ 未检测到 NVIDIA_API_KEY，跳过 DeepSeek 测试。")
+    else:
+        print("  ✓ DeepSeek-R1 (异步并发) 已就绪")
+
+    # ========== 步骤2: 获取历史数据 ==========
+    print("\n[2/4] 获取历史数据并计算指标...")
+    fetcher = BinanceDataFetcher()
+    df = fetcher.fetch_ohlcv('BTC', '1h', start_date='2025-10-01', end_date='2026-03-01', use_cache=False)
+    
+    df['rsi'] = ta.rsi(df['close'], length=14)
+    macd = ta.macd(df['close'])
+    if macd is not None and not macd.empty:
+        df['macd_hist'] = macd.iloc[:, 1]
+
+    # ========== 步骤3: 本地预处理与预测 ==========
+    print("\n[3/4] 运行本地预测 & 构建 API 任务队列...")
+    # DLinear预处理
+    df_dl = df.copy()
+    df_dl['rsi'] = ta.rsi(df_dl['close'], length=14)
+    macd_dl = ta.macd(df_dl['close'])
+    if macd_dl is not None and not macd_dl.empty:
+        df_dl['macd'], df_dl['macd_hist'], df_dl['macd_signal'] = macd_dl.iloc[:, 0], macd_dl.iloc[:, 1], macd_dl.iloc[:, 2]
+    df_dl = df_dl.dropna().reset_index(drop=True)
+    scaled_dl = dlinear_preprocessor.scaler.transform(df_dl[['open', 'high', 'low', 'close', 'volume', 'rsi', 'macd', 'macd_hist', 'macd_signal']])
+    X_dl, dl_ts_map = [], {}
+    for i in range(len(scaled_dl) - dlinear_preprocessor.lookback):
+        X_dl.append(scaled_dl[i:i + dlinear_preprocessor.lookback])
+        dl_ts_map[df_dl.iloc[i + dlinear_preprocessor.lookback - 1]['timestamp']] = i
+    X_dl = np.array(X_dl)
+
+    # N-BEATS预处理
+    X_nb, df_nb, idx_nb = nbeats_preprocessor.preprocess(df, fit=False)
+
+    # TTM预处理
+    scaler_ttm = StandardScaler()
+    scaled_close_ttm = scaler_ttm.fit_transform(df[['close']])
+    X_ttm, idx_ttm_map = [], {}
+    for i in range(len(scaled_close_ttm) - 512):
+        X_ttm.append(scaled_close_ttm[i:i+512])
+        idx_ttm_map[i + 511] = i
+    X_ttm = np.array(X_ttm)
+
+    target_targets = []
+    for target_idx in range(1, len(df)):
+        target_time_bj = utc_to_beijing(df.iloc[target_idx]['timestamp'])
+        if target_time_bj.year == 2026 and target_time_bj.month == 2:  # 锁定 2026年 2月 数据
+            target_targets.append(target_idx)
+
+    temp_results = []
+    deepseek_tasks_data = []
+
+    for target_idx in target_targets:
+        input_idx = target_idx - 1
+        current_close, target_close = df.iloc[input_idx]['close'], df.iloc[target_idx]['close']
+        actual_label = 1 if target_close > current_close else 0
+
+        # DL
+        dl_pred, input_ts = None, df.iloc[input_idx]['timestamp']
+        if (dl_i := dl_ts_map.get(input_ts)) is not None and dl_i < len(X_dl):
+            with torch.no_grad():
+                logits = dlinear_model(torch.FloatTensor(X_dl[dl_i:dl_i+1]).to(device)).cpu().numpy()
+                dl_pred = int((1 / (1 + np.exp(-logits))).item() > 0.5)
+
+        # NB
+        nb_pred = None
+        if input_idx in idx_nb:
+            nb_i = idx_nb.index(input_idx)
+            with torch.no_grad():
+                nb_out = nbeats_model(torch.FloatTensor(X_nb[nb_i:nb_i+1]).to(device)).cpu().numpy()
+                nb_pred = 1 if (nb_out[0] if nb_out.ndim == 1 else nb_out[0, 0]) > X_nb[nb_i, -1] else 0
+
+        # TTM
+        ttm_pred = None
+        if (ttm_i := idx_ttm_map.get(input_idx)) is not None:
+            with torch.no_grad():
+                ttm_pred_scaled = ttm_model(torch.FloatTensor(X_ttm[ttm_i:ttm_i+1]).to(device)).prediction_outputs[0, 0, 0].cpu().numpy()
+                ttm_pred = 1 if ttm_pred_scaled > X_ttm[ttm_i, -1, 0] else 0
+
+        # 构造 DeepSeek Payload
+        if ds_api_key:
+            recent_k = df.iloc[max(0, input_idx-14) : input_idx+1]
+            if len(recent_k) >= 5:
+                kline_text = "".join([f"时间: {row['timestamp']}, 开盘: {row['open']:.2f}, 收盘: {row['close']:.2f}, RSI: {row.get('rsi', 0):.2f}\n" for _, row in recent_k.iterrows()])
+                sys_prompt = '你是一位顶级量化分析师。请分析BTC 1小时K线，预测下一小时涨跌。必须输出合法的JSON格式：{"prediction": 1(看涨)或0(看跌), "reason": "理由"}'
+                messages = [
+                    {"role": "system", "content": sys_prompt}, 
+                    {"role": "user", "content": f"近15周期数据:\n{kline_text}\n输出JSON:"}
+                ]
+                deepseek_tasks_data.append({'id': target_idx, 'messages': messages})
+
+        temp_results.append({
+            'id': target_idx, 'time_utc': df.iloc[target_idx]['timestamp'], 'time_bj': utc_to_beijing(df.iloc[target_idx]['timestamp']),
+            'current_close': current_close, 'target_close': target_close, 'actual': actual_label,
+            'dlinear': dl_pred, 'nbeats': nb_pred, 'ttm': ttm_pred, 'deepseek': None
+        })
+
+    # ========== 异步执行 DeepSeek 并合入数据 ==========
+    if deepseek_tasks_data:
+        # 并发数设为 5，对 NVIDIA 免费/基础层级比较安全
+        ds_predictions = asyncio.run(run_all_deepseek_async(deepseek_tasks_data, ds_api_key))
+        for res in temp_results:
+            if res['id'] in ds_predictions:
+                res['deepseek'] = ds_predictions[res['id']]
+
+    # 结算集成模型结果 (四投票)
+    final_results = []
+    for res in temp_results:
+        votes = [v for v in [res['dlinear'], res['nbeats'], res['ttm'], res['deepseek']] if v is not None]
+        if votes:
+            vote_sum = sum(votes)
+            if vote_sum > len(votes) / 2: res['ensemble'] = 1
+            elif vote_sum < len(votes) / 2: res['ensemble'] = 0
+            else: res['ensemble'] = res['dlinear'] if res['dlinear'] is not None else 0
+        else:
+            res['ensemble'] = None
+        final_results.append(res)
+
+    # ========== 步骤4: 输出结果 ==========
+    if not final_results:
+        print("没有找到符合条件的测试样本")
+        return
+
+    df_res = pd.DataFrame(final_results)
+    df_res['date_bj'] = df_res['time_bj'].dt.date
+    df_res['us_open'] = df_res['time_utc'].apply(is_us_stock_open_by_utc)
+
+    # === (下方打印逻辑与原版一致，缩略优化版) ===
+    print(f"\n[4/4] 准确率统计 (北京时间)\n总样本数: {len(df_res)}")
+    print("\n" + "=" * 115 + "\n每小时详细数据\n" + "=" * 115)
+    hourly_h = f"{'北京时间':<18} {'前收盘':>10} {'现收盘':>10} {'实际':>4} {'DL':>3} {'':>1} {'NB':>3} {'':>1} {'TTM':>3} {'':>1} {'DS':>3} {'':>1} {'集成':>4} {'':<1}"
+
+    for date_bj, day_group in df_res.groupby('date_bj'):
+        us_str = "美股开盘" if day_group['us_open'].iloc[0] else "美股休市"
+        def s_acc(c): return (day_group.dropna(subset=[c])[c] == day_group.dropna(subset=[c])['actual']).mean() if len(day_group.dropna(subset=[c])) > 0 else None
+        print(f"\n--- {date_bj} ({us_str}) --- DL:{s_acc('dlinear'):.1%} | NB:{s_acc('nbeats'):.1%} | TTM:{s_acc('ttm'):.1%} | DS:{s_acc('deepseek'):.1%} | 集成:{s_acc('ensemble'):.1%}")
+        print(hourly_h + "\n" + "-" * 115)
+        for _, r in day_group.iterrows():
+            print(f"{r['time_bj'].strftime('%m-%d %H:%M'):<18} {r['current_close']:>10.2f} {r['target_close']:>10.2f} {fmt_pred(r['actual']):>4} "
+                  f"{fmt_pred(r['dlinear']):>3} {fmt_result(r['dlinear'],r['actual']):>1} {fmt_pred(r['nbeats']):>3} {fmt_result(r['nbeats'],r['actual']):>1} "
+                  f"{fmt_pred(r['ttm']):>3} {fmt_result(r['ttm'],r['actual']):>1} {fmt_pred(r['deepseek']):>3} {fmt_result(r['deepseek'],r['actual']):>1} "
+                  f"{fmt_pred(r['ensemble']):>4} {fmt_result(r['ensemble'],r['actual']):<1}")
+
+    print("\n\n" + "=" * 115 + "\n每日准确率汇总 (北京时间)\n" + "=" * 115)
+    print(f"{'日期':<14} {'样本':>4} {'美股':>6}  {'DLinear':>8} {'N-BEATS':>8} {'TTM-R2':>8} {'DeepSeek':>8} {'集成':>10}\n" + "-" * 115)
+    for date_bj, g in df_res.groupby('date_bj'):
+        def g_acc(c): return (g.dropna(subset=[c])[c] == g.dropna(subset=[c])['actual']).mean() if len(g.dropna(subset=[c])) > 0 else None
+        print(f"{str(date_bj):<14} {len(g):>4} {'开盘' if g['us_open'].iloc[0] else '休市':>6}  {g_acc('dlinear'):>8.1%} {g_acc('nbeats'):>8.1%} {g_acc('ttm'):>8.1%} {g_acc('deepseek'):>8.1%} {g_acc('ensemble'):>10.1%}".replace('nan%', 'N/A'))
+
+    def t_acc(c): return (df_res.dropna(subset=[c])[c] == df_res.dropna(subset=[c])['actual']).mean() if len(df_res.dropna(subset=[c])) > 0 else None
+    print("-" * 115)
+    print(f"{'总计':<14} {len(df_res):>4} {'':>6}  {t_acc('dlinear'):>8.2%} {t_acc('nbeats'):>8.2%} {t_acc('ttm'):>8.2%} {t_acc('deepseek'):>8.2%} {t_acc('ensemble'):>10.2%}".replace('nan%', 'N/A'))
+    print("=" * 115)
+
+if __name__ == "__main__":
+    main()

export_timesfm_onnx.py

@@ -1,295 +0,0 @@
-"""将Google TimesFM 2.5模型导出为ONNX格式 - 使用正确的API"""
-
-import sys
-import os
-sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
-
-import torch
-import numpy as np
-
-
-def export_timesfm_to_onnx():
-    print("="*80)
-    print("导出TimesFM 2.5 (200M) 模型为ONNX格式")
-    print("="*80)
-
-    # ========== 步骤1: 加载TimesFM模型 ==========
-    print("\n[1/6] 加载TimesFM 2.5模型...")
-
-    try:
-        import timesfm
-
-        # 使用正确的from_pretrained方法加载模型
-        print("  从HuggingFace加载预训练模型...")
-        print("  模型ID: google/timesfm-2.5-200m-pytorch")
-
-        model = timesfm.TimesFM_2p5_200M_torch.from_pretrained(
-            "google/timesfm-2.5-200m-pytorch",
-            torch_compile=False  # 导出ONNX时不使用torch.compile
-        )
-
-        print("✓ TimesFM 2.5模型已加载")
-
-    except ImportError as e:
-        print(f"✗ TimesFM未安装或导入失败: {e}")
-        print("  请运行: pip install git+https://github.com/google-research/timesfm.git")
-        return
-    except Exception as e:
-        print(f"✗ 加载模型失败: {e}")
-        import traceback
-        traceback.print_exc()
-        return
-
-    # ========== 步骤2: 编译模型配置 ==========
-    print("\n[2/6] 配置模型...")
-
-    try:
-        # 配置模型
-        model.compile(
-            timesfm.ForecastConfig(
-                max_context=512,        # BTC历史数据512小时
-                max_horizon=24,         # 预测未来24小时
-                normalize_inputs=True,
-                use_continuous_quantile_head=True,
-                force_flip_invariance=True,
-                infer_is_positive=True,
-                fix_quantile_crossing=True,
-            )
-        )
-
-        print("✓ 模型配置完成")
-        print(f"  最大上下文长度: 512")
-        print(f"  最大预测长度: 24")
-        print(f"  归一化输入: True")
-        print(f"  使用分位数预测: True")
-
-    except Exception as e:
-        print(f"✗ 配置失败: {e}")
-        import traceback
-        traceback.print_exc()
-        return
-
-    # ========== 步骤3: 测试模型 ==========
-    print("\n[3/6] 测试模型推理...")
-
-    try:
-        # 创建测试数据
-        test_input = [
-            np.linspace(85000, 90000, 512),  # 模拟BTC价格从85k到90k
-            np.sin(np.linspace(0, 20, 512)) * 1000 + 87000,  # 模拟震荡价格
-        ]
-
-        # 测试推理
-        point_forecast, quantile_forecast = model.forecast(
-            horizon=24,
-            inputs=test_input,
-        )
-
-        print("✓ 模型推理成功")
-        print(f"  点预测形状: {point_forecast.shape}")  # (2, 24)
-        print(f"  分位数预测形状: {quantile_forecast.shape}")  # (2, 24, 11)
-
-    except Exception as e:
-        print(f"⚠️  模型测试失败: {e}")
-        import traceback
-        traceback.print_exc()
-        print("  继续尝试导出...")
-
-    # ========== 步骤4: 提取核心模型并导出ONNX ==========
-    print("\n[4/6] 提取核心模型...")
-
-    try:
-        # TimesFM的核心模型在model.model中
-        core_model = model.model
-        core_model.eval()
-
-        print("✓ 核心模型已提取")
-        print(f"  模型类型: {type(core_model)}")
-
-        # 创建ONNX导出包装器
-        class TimesFMONNXWrapper(torch.nn.Module):
-            """TimesFM ONNX导出包装器"""
-            def __init__(self, timesfm_core_model):
-                super().__init__()
-                self.model = timesfm_core_model
-
-            def forward(self, inputs, masks=None):
-                """
-                Args:
-                    inputs: (batch_size, context_length) - 输入时间序列
-                    masks: (batch_size, context_length) - 可选mask，1=有效，0=无效
-                Returns:
-                    point_forecast: (batch_size, horizon_length) - 点预测
-                """
-                # 如果没有提供masks，创建全1的mask
-                if masks is None:
-                    masks = torch.ones_like(inputs)
-
-                # 调用TimesFM核心模型
-                # 注意：这里需要根据实际的forward签名调整
-                outputs, _ = self.model(inputs, masks, decode_caches=None)
-
-                # outputs包含: (input_embeddings, output_embeddings, output_ts, output_quantile_spread)
-                # output_ts是点预测
-                point_forecast = outputs[2]
-
-                return point_forecast
-
-        wrapped_model = TimesFMONNXWrapper(core_model)
-        wrapped_model.eval()
-
-        print("✓ ONNX包装器已创建")
-
-    except Exception as e:
-        print(f"✗ 提取核心模型失败: {e}")
-        import traceback
-        traceback.print_exc()
-        return
-
-    # ========== 步骤5: 导出为ONNX ==========
-    print("\n[5/6] 导出模型为ONNX格式...")
-
-    onnx_path = "./models/timesfm_2p5_200m.onnx"
-    os.makedirs("./models", exist_ok=True)
-
-    try:
-        # 准备示例输入
-        batch_size = 1
-        context_len = 512
-
-        dummy_input = torch.randn(batch_size, context_len, dtype=torch.float32)
-        dummy_mask = torch.ones(batch_size, context_len, dtype=torch.float32)
-
-        print(f"  示例输入形状: inputs={dummy_input.shape}, masks={dummy_mask.shape}")
-        print("  开始导出ONNX...")
-
-        # 导出为ONNX
-        torch.onnx.export(
-            wrapped_model,
-            (dummy_input, dummy_mask),
-            onnx_path,
-            export_params=True,
-            opset_version=14,
-            do_constant_folding=True,
-            input_names=['inputs', 'masks'],
-            output_names=['point_forecast'],
-            dynamic_axes={
-                'inputs': {0: 'batch_size', 1: 'context_length'},
-                'masks': {0: 'batch_size', 1: 'context_length'},
-                'point_forecast': {0: 'batch_size', 1: 'horizon_length'}
-            },
-            verbose=False
-        )
-
-        print(f"✓ ONNX模型已导出")
-        print(f"  保存路径: {onnx_path}")
-        print(f"  文件大小: {os.path.getsize(onnx_path) / (1024**2):.2f} MB")
-
-    except Exception as e:
-        print(f"✗ ONNX导出失败: {e}")
-        import traceback
-        traceback.print_exc()
-
-        # 尝试简化版本
-        print("\n尝试简化导出方案...")
-        try:
-            # 只导出单输入版本
-            torch.onnx.export(
-                wrapped_model,
-                dummy_input,
-                onnx_path,
-                export_params=True,
-                opset_version=14,
-                input_names=['inputs'],
-                output_names=['predictions'],
-            )
-            print(f"✓ 简化版ONNX模型已导出: {onnx_path}")
-        except Exception as e2:
-            print(f"✗ 简化导出也失败: {e2}")
-            return
-
-    # ========== 步骤6: 验证ONNX模型 ==========
-    print("\n[6/6] 验证ONNX模型...")
-
-    try:
-        import onnx
-        import onnxruntime as ort
-
-        # 加载并检查ONNX模型
-        onnx_model = onnx.load(onnx_path)
-        onnx.checker.check_model(onnx_model)
-        print("✓ ONNX模型结构验证通过")
-
-        # 显示模型信息
-        print(f"\n模型信息:")
-        print(f"  输入:")
-        for input_tensor in onnx_model.graph.input:
-            dims = [d.dim_value if d.dim_value > 0 else 'dynamic' for d in input_tensor.type.tensor_type.shape.dim]
-            print(f"    - {input_tensor.name}: {dims}")
-
-        print(f"  输出:")
-        for output_tensor in onnx_model.graph.output:
-            dims = [d.dim_value if d.dim_value > 0 else 'dynamic' for d in output_tensor.type.tensor_type.shape.dim]
-            print(f"    - {output_tensor.name}: {dims}")
-
-        # 创建ONNX Runtime推理会话
-        print("\n  创建ONNX Runtime推理会话...")
-        ort_session = ort.InferenceSession(onnx_path)
-
-        # 准备测试输入
-        test_input = np.random.randn(1, 512).astype(np.float32)
-        test_mask = np.ones((1, 512), dtype=np.float32)
-
-        # 运行推理
-        print("  运行推理测试...")
-        outputs = ort_session.run(
-            None,
-            {'inputs': test_input, 'masks': test_mask}
-        )
-
-        print(f"✓ ONNX推理成功")
-        print(f"  输出数量: {len(outputs)}")
-        for i, output in enumerate(outputs):
-            print(f"  输出{i}形状: {output.shape}")
-
-    except ImportError:
-        print("⚠️  未安装onnx或onnxruntime，跳过验证")
-        print("  安装命令: pip install onnx onnxruntime")
-    except Exception as e:
-        print(f"⚠️  ONNX验证失败: {e}")
-        import traceback
-        traceback.print_exc()
-        print("  但模型文件已保存，可以尝试手动验证")
-
-    # ========== 总结 ==========
-    print("\n" + "="*80)
-    print("✓ TimesFM 2.5模型导出完成!")
-    print("="*80)
-
-    if os.path.exists(onnx_path):
-        print(f"\n保存位置: {onnx_path}")
-        print(f"文件大小: {os.path.getsize(onnx_path) / (1024**2):.2f} MB")
-
-        print(f"\n使用方法:")
-        print(f"```python")
-        print(f"import onnxruntime as ort")
-        print(f"import numpy as np")
-        print(f"")
-        print(f"# 加载模型")
-        print(f"session = ort.InferenceSession('{onnx_path}')")
-        print(f"")
-        print(f"# 准备输入（512小时BTC历史价格）")
-        print(f"btc_prices = np.array([85000, 85100, ...])  # 512个价格点")
-        print(f"inputs = btc_prices.reshape(1, -1).astype(np.float32)")
-        print(f"masks = np.ones((1, 512), dtype=np.float32)")
-        print(f"")
-        print(f"# 推理")
-        print(f"outputs = session.run(None, {{'inputs': inputs, 'masks': masks}})")
-        print(f"predictions = outputs[0]  # 未来24小时价格预测")
-        print(f"```")
-
-    print("\n" + "="*80)
-
-
-if __name__ == "__main__":
-    export_timesfm_to_onnx()

models/inference.py

@@ -8,20 +8,13 @@ import torch
 import numpy as np
 import pandas as pd
 import pandas_ta as ta
+import onnxruntime as ort
 from typing import Dict, List, Optional, Tuple, Union, Any
 from sklearn.preprocessing import StandardScaler
 
 # 添加项目根目录到 sys.path
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from models.dlinear.model import DLinear
-from models.nbeats.model import NBeats
-try:
-    from tsfm_public.models.tinytimemixer import TinyTimeMixerForPrediction
-except (ImportError, NameError):
-    # 尝试作为可选依赖处理
-    TinyTimeMixerForPrediction = None
-
 
 
 class CustomUnpickler(pickle.Unpickler):
@@ -171,15 +164,15 @@ class InferenceManager:
         # 配置路径
         self.configs = {
             'dlinear': {
-                'model_path': './models/checkpoints/dlinear_classification_2026/best_model.pt',
+                'model_path': './models/checkpoints/dlinear_classification_2026/model.onnx',
                 'preprocessor_path': './models/checkpoints/dlinear_classification_2026/preprocessor.pkl'
             },
             'nbeats': {
-                'model_path': './models/checkpoints/nbeats_btc_1h/best_model.pt',
+                'model_path': './models/checkpoints/nbeats_btc_1h/model.onnx',
                 'preprocessor_path': './models/checkpoints/nbeats_btc_1h/preprocessor.pkl'
             },
             'ttm': {
-                'model_name': 'ibm-granite/granite-timeseries-ttm-r2'
+                'model_path': './models/checkpoints/ttm_r2_onnx/ttm_r2_crypto_channel_mixing.onnx'
             }
         }
         self.initialized = True
@@ -216,10 +209,8 @@ class InferenceManager:
             print(f"DLinear 模型文件不存在: {config['model_path']}")
             return
 
-        checkpoint = torch.load(config['model_path'], map_location=self.device, weights_only=False)
-        model = DLinear(checkpoint['config'])
-        model.load_state_dict(checkpoint['model_state_dict'])
-        self.models['dlinear'] = model.to(self.device).eval()
+        try:
+            self.models['dlinear'] = ort.InferenceSession(config['model_path'])
             
             with open(config['preprocessor_path'], 'rb') as f:
                 try:
@@ -227,7 +218,9 @@ class InferenceManager:
                 except AttributeError:
                     f.seek(0)
                     self.preprocessors['dlinear'] = CustomUnpickler(f).load()
-        print("  ✓ DLinear 已加载")
+            print("  ✓ DLinear (ONNX) 已加载")
+        except Exception as e:
+            print(f"DLinear 加载失败: {e}")
 
     def _load_nbeats(self):
         config = self.configs['nbeats']
@@ -235,31 +228,23 @@ class InferenceManager:
             print(f"N-BEATS 模型文件不存在: {config['model_path']}")
             return
 
-        checkpoint = torch.load(config['model_path'], map_location=self.device, weights_only=False)
-        model = NBeats(checkpoint['config'])
-        self.models['nbeats'] = model.to(self.device).eval()
+        try:
+            self.models['nbeats'] = ort.InferenceSession(config['model_path'])
             with open(config['preprocessor_path'], 'rb') as f:
                  try:
                     self.preprocessors['nbeats'] = pickle.load(f)
                  except AttributeError:
                     f.seek(0)
                     self.preprocessors['nbeats'] = CustomUnpickler(f).load()
-        print("  ✓ N-BEATS 已加载")
+            print("  ✓ N-BEATS (ONNX) 已加载")
+        except Exception as e:
+            print(f"N-BEATS 加载失败: {e}")
 
     def _load_ttm(self):
-        # 检查 TTM 类是否已导入
-        if TinyTimeMixerForPrediction is None:
-            print("TTM-R2 模型类未定义 (可能是依赖库缺失)")
-            return
-
         config = self.configs['ttm']
         try:
-            model = TinyTimeMixerForPrediction.from_pretrained(
-                config['model_name'],
-                trust_remote_code=True
-            )
-            self.models['ttm'] = model.to(self.device).eval()
-            print("  ✓ TTM-R2 已加载")
+            self.models['ttm'] = ort.InferenceSession(config['model_path'])
+            print("  ✓ TTM-R2 (ONNX) 已加载")
         except Exception as e:
             print(f"TTM 加载失败: {e}")
 
@@ -272,7 +257,7 @@ class InferenceManager:
             if 'dlinear' not in self.models:
                 return None, None
             
-        model = self.models['dlinear']
+        session = self.models['dlinear']
         preprocessor = self.preprocessors['dlinear']
         
         # 预处理
@@ -282,15 +267,22 @@ class InferenceManager:
             return None, None
             
         # 取最后一个窗口
-        X_last = torch.FloatTensor(X[-1:]).to(self.device)
+        # ONNX 需要 float32
+        X_last = X[-1:].astype(np.float32)
         
-        with torch.no_grad():
-            logits = model(X_last).cpu().numpy()
+        try:
+            input_name = session.get_inputs()[0].name
+            output_name = session.get_outputs()[0].name
+            
+            logits = session.run([output_name], {input_name: X_last})[0]
             prob = 1 / (1 + np.exp(-logits))
             prediction = int(prob.item() > 0.5)
             probability = float(prob.item())
             
             return prediction, probability
+        except Exception as e:
+            print(f"DLinear 预测出错: {e}")
+            return None, None
 
     def predict_nbeats(self, df: pd.DataFrame) -> Tuple[Optional[int], Optional[float]]:
         """
@@ -301,7 +293,7 @@ class InferenceManager:
             if 'nbeats' not in self.models:
                 return None, None
             
-        model = self.models['nbeats']
+        session = self.models['nbeats']
         preprocessor = self.preprocessors['nbeats']
         
         # 预处理
@@ -311,11 +303,15 @@ class InferenceManager:
             return None, None
             
         # 取最后一个窗口
-        X_last = torch.FloatTensor(X[-1:]).to(self.device)
+        X_last = X[-1:].astype(np.float32)
         current_price_scaled = X[-1, -1] # 窗口中最后一个点的缩放价格
         
-        with torch.no_grad():
-            output = model(X_last).cpu().numpy()
+        try:
+            input_name = session.get_inputs()[0].name
+            output_name = session.get_outputs()[0].name
+            
+            output = session.run([output_name], {input_name: X_last})[0]
+            
             # output shape: (batch, forecast_horizon) or (batch, 1)
             if output.ndim == 1:
                 pred_price_scaled = output[0]
@@ -328,6 +324,9 @@ class InferenceManager:
             probability = 0.6 if prediction == 1 else 0.4 # 占位
             
             return prediction, probability
+        except Exception as e:
+            print(f"N-BEATS 预测出错: {e}")
+            return None, None
 
     def predict_ttm(self, df: pd.DataFrame) -> Tuple[Optional[int], Optional[float]]:
         """
@@ -338,7 +337,7 @@ class InferenceManager:
             if 'ttm' not in self.models:
                 return None, None
             
-        model = self.models['ttm']
+        session = self.models['ttm']
         context_length = 512
         
         all_prices = df['close'].values
@@ -353,12 +352,19 @@ class InferenceManager:
             padded[-len(all_prices):] = all_prices
             context_data = padded.reshape(1, -1, 1)
             
-        context_tensor = torch.FloatTensor(context_data).to(self.device)
+        # ONNX input
+        context_data = context_data.astype(np.float32)
         base_price = float(df['close'].iloc[-1])
         
-        with torch.no_grad():
-            output = model(context_tensor)
-            pred_price = output.prediction_outputs.cpu().numpy()[0, 0, 0]
+        try:
+            input_name = session.get_inputs()[0].name
+            output_name = session.get_outputs()[0].name
+            
+            output = session.run([output_name], {input_name: context_data})[0]
+            # output shape might be (batch, prediction_length, features)
+            # prediction_length=96 usually for TTM zero-shot
+            # We want the first prediction step [0, 0, 0]
+            pred_price = output[0, 0, 0]
             
             prediction = 1 if pred_price > base_price else 0
             
@@ -367,6 +373,9 @@ class InferenceManager:
             probability = 1 / (1 + np.exp(-price_pct * 100))
             
             return prediction, float(probability)
+        except Exception as e:
+            print(f"TTM 预测出错: {e}")
+            return None, None
 
     def predict_all(self, df: pd.DataFrame) -> Dict[str, Any]:
         """

requirements.txt

@@ -46,6 +46,9 @@ httpx>=0.25.0
 # 数据源
 # ----------------------------------------------
 python-binance>=1.0.19
+websocket-client>=1.8.0
+python-socks>=2.5.0
+PySocks>=1.7.1
 
 # ----------------------------------------------
 # 任务调度

scripts/export_onnx.py

@@ -1,13 +1,9 @@
-"""导出 DLinear 模型为 ONNX 格式。
-
-用法:
-    python scripts/export_onnx.py
-
-输出:
-    models/checkpoints/dlinear_classification_2026/model.onnx
-"""
+"""导出 DLinear, N-BEATS 和 TTM-R2 模型为 ONNX 格式。"""
 
 import sys
+import os
+import torch
+import numpy as np
 from pathlib import Path
 
 # 添加项目根目录到路径
@@ -15,23 +11,124 @@ project_root = Path(__file__).parent.parent
 sys.path.insert(0, str(project_root))
 
 from models.dlinear.model import DLinear
+from models.nbeats.model import NBeats
+from tsfm_public.toolkit.get_model import get_model
 
 
-def main():
-    # 模型路径
+def export_dlinear():
+    print("\n" + "="*50)
+    print("导出 DLinear 模型...")
+    try:
         checkpoint_path = project_root / "models/checkpoints/dlinear_classification_2026/best_model.pt"
         output_path = checkpoint_path.parent / "model.onnx"
         
+        if not checkpoint_path.exists():
+            print(f"跳过: 找不到检查点 {checkpoint_path}")
+            return
+
         print(f"加载模型: {checkpoint_path}")
         model = DLinear.load(str(checkpoint_path), device='cpu')
         
         print(f"模型配置: lookback={model.config.lookback}, input_dim={model.config.input_dim}")
-    print(f"参数量: {model.count_parameters():,}")
         
-    # 导出为 ONNX
         model.to_onnx(str(output_path))
         print(f"✓ ONNX 模型已保存到: {output_path}")
+    except Exception as e:
+        print(f"导出 DLinear 失败: {e}")
 
+def export_nbeats():
+    print("\n" + "="*50)
+    print("导出 N-BEATS 模型...")
+    try:
+        checkpoint_path = project_root / "models/checkpoints/nbeats_btc_1h/best_model.pt"
+        output_path = checkpoint_path.parent / "model.onnx"
+        
+        if not checkpoint_path.exists():
+            print(f"跳过: 找不到检查点 {checkpoint_path}")
+            return
+
+        print(f"加载模型: {checkpoint_path}")
+        checkpoint = torch.load(str(checkpoint_path), map_location='cpu')
+        model = NBeats(checkpoint['config'])
+        model.eval()
+        
+        # N-BEATS input: (batch_size, input_size)
+        # input_size default is 168
+        input_size = checkpoint['config'].input_size
+        dummy_input = torch.randn(1, input_size)
+        
+        print(f"导出 ONNX (input_size={input_size})...")
+        torch.onnx.export(
+            model,
+            dummy_input,
+            str(output_path),
+            export_params=True,
+            opset_version=17,
+            do_constant_folding=True,
+            input_names=['input'],
+            output_names=['forecast'],
+            dynamic_axes={'input': {0: 'batch_size'}, 'forecast': {0: 'batch_size'}}
+        )
+        print(f"✓ ONNX 模型已保存到: {output_path}")
+    except Exception as e:
+        print(f"导出 N-BEATS 失败: {e}")
+        import traceback
+        traceback.print_exc()
+
+def export_ttm_r2():
+    print("\n" + "="*50)
+    print("导出 TTM-R2 模型...")
+    try:
+        model_name = "ibm-granite/granite-timeseries-ttm-r2"
+        output_dir = project_root / "models/checkpoints/ttm_r2_onnx"
+        output_path = output_dir / "model.onnx"
+        
+        output_dir.mkdir(parents=True, exist_ok=True)
+        
+        print(f"加载模型: {model_name}")
+        base_model = get_model(
+            model_path=model_name,
+            context_length=512,
+            prediction_length=96,
+        )
+        base_model.eval()
+        
+        print(f"TTM Config: {base_model.config}")
+        
+        # TTM input: (batch_size, context_length, num_features)
+        # Context length 512, features 1
+        dummy_input = torch.randn(1, 512, 1)
+        with torch.no_grad():
+            output = base_model(dummy_input)
+            if hasattr(output, 'prediction_outputs'):
+                test_output = output.prediction_outputs
+            else:
+                test_output = output
+            print(f"✓ 输出形状: {test_output.shape}")
+        print(f"导出 ONNX (固定 Batch Size=1)...")
+        torch.onnx.export(
+            base_model,
+            dummy_input,
+            str(output_path),
+            export_params=True,
+            opset_version=17,  # Revert to 14 as it is standard
+            do_constant_folding=True,
+            input_names=['past_values'],
+            output_names=['predictions'],
+            # Remove dynamic axes to fix potential shape inference issues
+            dynamic_axes={'past_values': {0: 'batch_size'}, 'predictions': {0: 'batch_size'}},
+            verbose=False
+        )
+        print(f"✓ ONNX 模型已保存到: {output_path}")
+    except Exception as e:
+        print(f"导出 TTM-R2 失败: {e}")
+        import traceback
+        traceback.print_exc()
+
+def main():
+    export_dlinear()
+    export_nbeats()
+    #export_ttm_r2()
 
 if __name__ == "__main__":
     main()

scripts/hourly_fetch_btc_1h.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+"""整点后5秒抓取 BTC 1h K线（至少512根，且包含最新时间）"""
+
+import os
+import sys
+import time
+from datetime import datetime, timedelta
+
+import pandas as pd
+import pytz
+
+# 直接运行脚本时，确保可以从项目根目录导入 data 包
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from data.fetcher import BinanceDataFetcher
+
+
+def get_next_run_time(delay_seconds: int = 5) -> datetime:
+    """计算下一次执行时间：下一个整点 + delay_seconds。"""
+    now_utc = datetime.now(pytz.UTC)
+    next_hour = (now_utc + timedelta(hours=1)).replace(minute=0, second=0, microsecond=0)
+    return next_hour + timedelta(seconds=delay_seconds)
+
+
+def fetch_min_klines_with_latest(
+    fetcher: BinanceDataFetcher,
+    min_klines: int = 512,
+    latest_check_retries: int = 8,
+    latest_check_interval: int = 2,
+) -> pd.DataFrame:
+    """
+    拉取 BTC 1h K线，确保：
+    1) 数量不少于 min_klines
+    2) 包含当前小时的最新时间（UTC 整点）
+    """
+    now_utc = datetime.now(pytz.UTC)
+    expected_latest_ts = now_utc.replace(minute=0, second=0, microsecond=0).replace(tzinfo=None)
+
+    lookback_days = 35
+    for attempt in range(latest_check_retries):
+        start_date = (datetime.now(pytz.UTC) - timedelta(days=lookback_days)).strftime("%Y-%m-%d")
+
+        # 使用 WebSocket 拉取，避免仅命中缓存导致最新K线不更新。
+        df = fetcher.fetch_ohlcv(
+            symbol="BTC",
+            timeframe="1h",
+            start_date=start_date,
+            end_date=None,
+            use_cache=False,
+            max_retries=3,
+        )
+
+        if len(df) < min_klines:
+            lookback_days += 7
+            print(
+                f"[WARN] 当前仅获取 {len(df)} 根K线，小于 {min_klines}，扩大回看窗口到 {lookback_days} 天后重试。"
+            )
+            continue
+
+        latest_ts = df["timestamp"].max().to_pydatetime().replace(tzinfo=None)
+        if latest_ts >= expected_latest_ts:
+            result = df.tail(min_klines).copy().reset_index(drop=True)
+            print(
+                f"[OK] 获取成功: {len(result)} 根 | 最新时间: {latest_ts} UTC | 期望最新时间: {expected_latest_ts} UTC"
+            )
+            return result
+
+        if attempt < latest_check_retries - 1:
+            print(
+                f"[WAIT] 最新K线未到位，当前最新: {latest_ts} UTC，期望: {expected_latest_ts} UTC，"
+                f"{latest_check_interval}秒后重试({attempt + 1}/{latest_check_retries})。"
+            )
+            time.sleep(latest_check_interval)
+
+    raise RuntimeError("多次重试后仍未拿到包含最新时间的BTC 1h K线数据")
+
+
+def run_once(min_klines: int = 512) -> None:
+    """执行一次抓取任务。"""
+    fetcher = BinanceDataFetcher()
+    data = fetch_min_klines_with_latest(fetcher=fetcher, min_klines=min_klines)
+
+    print("-" * 80)
+    print(f"抓取完成，返回K线数量: {len(data)}")
+    print(f"时间范围: {data['timestamp'].min()} -> {data['timestamp'].max()} (UTC)")
+    print(data.tail(3)[["timestamp", "open", "high", "low", "close", "volume"]])
+    print("-" * 80)
+
+
+def run_forever(min_klines: int = 512, delay_seconds: int = 5) -> None:
+    """常驻运行：每个整点延迟 delay_seconds 后执行抓取。"""
+    while True:
+        next_run = get_next_run_time(delay_seconds=delay_seconds)
+        sleep_seconds = max(0.0, (next_run - datetime.now(pytz.UTC)).total_seconds())
+
+        print(f"[SCHEDULE] 当前UTC: {datetime.now(pytz.UTC)}，下次执行: {next_run}，等待 {sleep_seconds:.1f} 秒")
+        time.sleep(sleep_seconds)
+
+        try:
+            run_once(min_klines=min_klines)
+        except Exception as exc:
+            print(f"[ERROR] 本轮抓取失败: {exc}")
+
+
+def main() -> None:
+    """脚本入口。"""
+    # 默认常驻运行，如需只执行一次，可把这里改成 run_once(min_klines=512)
+    run_forever(min_klines=512, delay_seconds=5)
+
+
+if __name__ == "__main__":
+    main()

start_system.bat

@@ -1,47 +0,0 @@
-@echo off
-chcp 65001 >nul
-REM ============================================
-REM  系统启动脚本 (Windows)
-REM  启动 API 服务和自动交易程序
-REM ============================================
-
-echo [%date% %time%] 正在启动系统...
-
-REM 检查 conda 环境
-where conda >nul 2>&1
-if %errorlevel% neq 0 (
-    echo [错误] 未找到 conda，请先安装 Anaconda 或 Miniconda
-    pause
-    exit /b 1
-)
-
-REM 激活 conda 环境
-echo [%date% %time%] 激活 conda 环境: crypto_pred_312
-call conda activate crypto_pred_312
-if %errorlevel% neq 0 (
-    echo [错误] 无法激活 conda 环境 crypto_pred_312
-    echo 请先创建环境: conda create -n crypto_pred_312 python=3.12
-    pause
-    exit /b 1
-)
-
-REM 启动 API 服务 (后台运行)
-echo [%date% %time%] 启动 API 服务...
-start "API服务" cmd /c "python -m uvicorn api.main:app --host 0.0.0.0 --port 8000"
-
-REM 等待 API 启动
-timeout /t 3 /nobreak >nul
-
-REM 启动自动交易程序 (后台运行)
-echo [%date% %time%] 启动自动交易程序...
-start "自动交易" cmd /c "python -u auto_trade_polymarket.py"
-
-echo.
-echo ============================================
-echo   系统启动完成!
-echo   前端访问: http://localhost:8000
-echo   关闭窗口可停止对应服务
-echo ============================================
-echo.
-
-pause

start_system.ps1

@@ -1,90 +0,0 @@
-# ============================================
-#  系统启动脚本 (Windows PowerShell)
-#  启动 API 服务和自动交易程序
-# ============================================
-
-$ErrorActionPreference = "Stop"
-
-function Write-Log {
-    param([string]$Message, [string]$Color = "White")
-    $timestamp = Get-Date -Format "yyyy-MM-dd HH:mm:ss"
-    Write-Host "[$timestamp] $Message" -ForegroundColor $Color
-}
-
-Write-Log "正在启动系统..." "Yellow"
-
-# 检查 conda
-$condaPath = Get-Command conda -ErrorAction SilentlyContinue
-if (-not $condaPath) {
-    Write-Log "错误: 未找到 conda，请先安装 Anaconda 或 Miniconda" "Red"
-    Read-Host "按回车键退出"
-    exit 1
-}
-
-# 激活 conda 环境
-Write-Log "激活 conda 环境: crypto_pred_312" "Yellow"
-try {
-    conda activate crypto_pred_312
-} catch {
-    Write-Log "错误: 无法激活 conda 环境 crypto_pred_312" "Red"
-    Write-Log "请先创建环境: conda create -n crypto_pred_312 python=3.12" "Yellow"
-    Read-Host "按回车键退出"
-    exit 1
-}
-
-# 启动 API 服务
-Write-Log "启动 API 服务..." "Yellow"
-$apiProcess = Start-Process -FilePath "python" `
-    -ArgumentList "-m uvicorn api.main:app --host 0.0.0.0 --port 8000" `
-    -PassThru -NoNewWindow
-
-Write-Log "API 服务已启动 (PID: $($apiProcess.Id))" "Green"
-
-# 等待 API 启动
-Start-Sleep -Seconds 3
-
-# 启动自动交易程序
-Write-Log "启动自动交易程序..." "Yellow"
-$tradeProcess = Start-Process -FilePath "python" `
-    -ArgumentList "-u auto_trade_polymarket.py" `
-    -PassThru -NoNewWindow
-
-Write-Log "自动交易程序已启动 (PID: $($tradeProcess.Id))" "Green"
-
-Write-Host ""
-Write-Host "============================================" -ForegroundColor Cyan
-Write-Host "  系统启动完成!" -ForegroundColor Green
-Write-Host "  前端访问: http://localhost:8000" -ForegroundColor Green
-Write-Host "  按 Ctrl+C 停止所有服务" -ForegroundColor Yellow
-Write-Host "============================================" -ForegroundColor Cyan
-Write-Host ""
-
-# 等待用户中断
-try {
-    while ($true) {
-        # 检查进程状态
-        if ($apiProcess.HasExited) {
-            Write-Log "API 服务意外退出" "Red"
-            break
-        }
-        if ($tradeProcess.HasExited) {
-            Write-Log "自动交易程序意外退出" "Red"
-            break
-        }
-        Start-Sleep -Seconds 1
-    }
-} finally {
-    Write-Log "正在停止服务..." "Yellow"
-    
-    if (-not $apiProcess.HasExited) {
-        Stop-Process -Id $apiProcess.Id -Force -ErrorAction SilentlyContinue
-        Write-Log "API 服务已停止" "Green"
-    }
-    
-    if (-not $tradeProcess.HasExited) {
-        Stop-Process -Id $tradeProcess.Id -Force -ErrorAction SilentlyContinue
-        Write-Log "自动交易程序已停止" "Green"
-    }
-    
-    Write-Log "所有服务已停止" "Green"
-}

start_system.py

@@ -4,6 +4,7 @@
 同时启动：
 1. FastAPI 后端服务 (提供前端看板 API)
 2. Polymarket 自动交易程序
+3. 每日数据报告推送程序 (北京时间每天 8 点)
 
 支持平台：macOS, Linux, Windows
 """
@@ -15,6 +16,13 @@ import os
 import platform
 from datetime import datetime
 
+try:
+    from daily_report_sender import run_scheduler_background
+except ImportError:
+    # 如果模块导入失败，定义一个空函数避免启动失败
+    def run_scheduler_background():
+        return None
+
 # 检测操作系统
 IS_WINDOWS = platform.system() == 'Windows'
 
@@ -91,6 +99,23 @@ def terminate_process(process, name):
         except Exception:
             pass
 
+def send_alert(content):
+    """发送告警消息 (使用 curl)"""
+    try:
+        url = "https://zhiyinzhe.xyz/sendMsg/api/send"
+        cmd = [
+            "curl", "-X", "POST", url,
+            "-F", "bot_name=weibo",
+            "-F", "receive_id=on_f972f5f8c4de6ac967574b97dc8c1d82",
+            "-F", "receive_id_type=union_id",
+            "-F", f"content={content}"
+        ]
+        # 使用 run 等待请求完成，确保告警发出
+        subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10)
+        log(f"已发送告警: {content}", GREEN)
+    except Exception as e:
+        log(f"发送告警失败: {e}", RED)
+
 def main():
     # 激活 Conda 环境
     # 注意：在子进程中直接使用当前已激活的 python 解释器路径，避免 conda activate 问题
@@ -109,6 +134,18 @@ def main():
     trade_cmd = f"{python_exec} -u trader/auto_trade_polymarket.py"
     trade_process = start_process(trade_cmd, "Polymarket 自动交易程序")
     
+    # 3. 启动每日数据报告推送程序（后台线程）
+    # 北京时间每天 8 点推送昨天的数据统计
+    try:
+        report_thread = run_scheduler_background()
+        if report_thread:
+            log("每日数据报告推送程序已启动", GREEN)
+        else:
+            log("⚠️  每日数据报告推送程序启动失败", YELLOW)
+    except Exception as e:
+        log(f"⚠️  启动每日报告程序异常: {e}", YELLOW)
+        report_thread = None
+    
     log("系统启动完成！", GREEN)
     log("前端访问地址: http://localhost:8000", GREEN)
     log("按 Ctrl+C 停止所有服务...", YELLOW)
@@ -117,10 +154,14 @@ def main():
         # 监控子进程状态
         while True:
             if api_process and api_process.poll() is not None:
-                log("API 服务意外退出", RED)
+                msg = "API 服务意外退出"
+                log(msg, RED)
+                send_alert(msg)
                 break
             if trade_process and trade_process.poll() is not None:
-                log("自动交易程序意外退出", RED)
+                msg = "自动交易程序意外退出"
+                log(msg, RED)
+                send_alert(msg)
                 break
             time.sleep(1)
             

trader/auto_trade_polymarket.py

@@ -15,7 +15,7 @@ import argparse
 import json
 import time
 import logging
-from logging.handlers import RotatingFileHandler
+from logging.handlers import TimedRotatingFileHandler
 import schedule
 from datetime import datetime, timedelta
 from typing import Optional, Tuple, Dict, List
@@ -34,6 +34,7 @@ from data.fetcher import BinanceDataFetcher
 from data.trading_db import TradingDatabase
 from api.models import PredictionModel
 from models.inference import get_inference_manager
+from utils.send_msg import lark_notifier
 
 load_dotenv()
 
@@ -67,24 +68,20 @@ def setup_logging(log_dir: str = "logs", log_level: int = logging.INFO) -> loggi
         datefmt="%Y-%m-%d %H:%M:%S"
     )
     
-    # 文件handler (按大小轮转, 最大10MB, 保留5个备份)
+    # 文件handler (按天轮转, 保留最近30天)
     log_file = os.path.join(log_dir, "polymarket_trading.log")
-    file_handler = RotatingFileHandler(
+    file_handler = TimedRotatingFileHandler(
         log_file,
-        maxBytes=10*1024*1024,  # 10MB
-        backupCount=5,
+        when='midnight',
+        interval=1,
+        backupCount=30,
         encoding='utf-8'
     )
+    file_handler.suffix = "%Y-%m-%d"
     file_handler.setLevel(log_level)
     file_handler.setFormatter(formatter)
     logger.addHandler(file_handler)
     
-    # 控制台handler
-    console_handler = logging.StreamHandler(sys.stdout)
-    console_handler.setLevel(log_level)
-    console_handler.setFormatter(formatter)
-    logger.addHandler(console_handler)
-    
     return logger
 
 
@@ -92,6 +89,16 @@ def setup_logging(log_dir: str = "logs", log_level: int = logging.INFO) -> loggi
 logger = setup_logging()
 
 
+def _apply_global_proxy(proxy_url: str) -> None:
+    """统一设置代理环境变量，确保第三方网络库也走代理。"""
+    os.environ["HTTP_PROXY"] = proxy_url
+    os.environ["HTTPS_PROXY"] = proxy_url
+    os.environ["ALL_PROXY"] = proxy_url
+    os.environ["http_proxy"] = proxy_url
+    os.environ["https_proxy"] = proxy_url
+    os.environ["all_proxy"] = proxy_url
+
+
 class PolymarketBTCTrader:
     """Polymarket BTC Hourly交易器 (多模型集成版)"""
     
@@ -118,7 +125,8 @@ class PolymarketBTCTrader:
         self.dry_run = dry_run
         
         # HTTP客户端
-        proxy = os.getenv("SOCKS5_PROXY") if os.getenv("USE_PROXY", "false").lower() == "true" else None
+        proxy = os.getenv("SOCKS5_PROXY", "socks5://127.0.0.1:7890")
+        _apply_global_proxy(proxy)
         self.http = httpx.Client(
             timeout=30.0,
             proxy=proxy,
@@ -150,18 +158,80 @@ class PolymarketBTCTrader:
         """初始化CLOB交易客户端"""
         try:
             from py_clob_client.client import ClobClient
+            from py_clob_client.clob_types import ApiCreds, OpenOrderParams
             
+            funder = os.getenv("FUNDER_ADDRESS")
+            nonce_str = os.getenv("POLYMARKET_NONCE")  # 可选，取决于你的钱包设置
+            nonce = int(nonce_str) if nonce_str else None
+            if self.private_key and funder and nonce is not None:
+                try: 
                     self.clob_client = ClobClient(
                         host=self.CLOB_API,
                         key=self.private_key,
                         chain_id=137,  # Polygon
+                        signature_type=2,
+                        funder=funder
                     )
-            self.clob_client.set_api_creds(self.clob_client.create_or_derive_api_creds())
-            logger.info("✓ CLOB客户端初始化成功")
+                    derived_creds = self.clob_client.create_or_derive_api_creds(nonce=nonce)
+                    self.clob_client.set_api_creds(derived_creds)
+                    logger.info(f"✓ 私钥派生 API Creds 成功! Derived Key: {derived_creds.api_key}")
+                    # 验证API Key是否有效
+                    self.clob_client.get_api_keys()
+                    # 打印账户信息
+                    self._print_account_info()
+                    
+                except Exception as e:
+                    logger.warning(f"⚠️ 环境变量API凭证鉴权失败: {e}")
+                    self._print_account_info()
+            else:
+                self._print_account_info()
+
         except Exception as e:
             logger.warning(f"⚠️ CLOB客户端初始化失败: {e}")
             self.clob_client = None
 
+    def _print_account_info(self):
+        """打印账户余额和Allowance信息"""
+        try:
+            logger.info("-" * 40)
+            logger.info("👤 账户信息检查:")
+            
+            # 1. 获取地址
+            address = self.clob_client.get_address()
+            logger.info(f"   钱包地址: {address}")
+            
+            # 2. 获取余额和Allowance
+            try:
+                from py_clob_client.clob_types import BalanceAllowanceParams, AssetType
+                
+                # 检查 USDC (Collateral) 的余额
+                # 在Polymarket中，USDC通常作为 COLLATERAL
+                params = BalanceAllowanceParams(asset_type=AssetType.COLLATERAL)
+                
+                balance_allowance = self.clob_client.get_balance_allowance(params)
+                logger.info(f"   USDC余额: ${float(balance_allowance.get('balance', 0)) / 10**6:.2f}") # USDC是6位小数
+                logger.info(f"   授权额度: {balance_allowance.get('allowance', 'N/A')}")
+                
+                # 检查是否足够下单
+                balance = float(balance_allowance.get('balance', 0)) / 10**6
+                if balance < self.max_bet:
+                    logger.warning(f"⚠️ 余额不足: ${balance:.2f} < 下单额 ${self.max_bet:.2f}")
+                    
+            except Exception as e:
+                logger.warning(f"   无法获取详细余额信息: {e}")
+                # 备用方法
+                try:
+                     logger.info(f"   尝试备用方法获取余额...")
+                     # 有些版本的 client 不需要 params
+                     ba = self.clob_client.get_balance_allowance()
+                     logger.info(f"   Balance Info: {ba}")
+                except:
+                     pass
+                
+            logger.info("-" * 40)
+        except Exception as e:
+            logger.error(f"❌ 获取账户信息失败: {e}")
+
     def _log_to_db(self, result: Dict, market: Optional[Dict]):
         """记录交易结果到数据库"""
         try:
@@ -191,6 +261,7 @@ class PolymarketBTCTrader:
                 'size': result.get('size'),
                 'status': status,
                 'error_message': result.get('error'),
+                'order_id': result.get('order_id'),
                 'raw_data': result
             }
             self.db.log_trade(log_data)
@@ -235,6 +306,7 @@ class PolymarketBTCTrader:
     
     def get_market_by_slug(self, slug: str) -> Optional[Dict]:
         """通过slug获取市场信息"""
+        for attempt in range(3):
             try:
                 response = self.http.get(
                     f"{self.GAMMA_API}/events",
@@ -284,7 +356,11 @@ class PolymarketBTCTrader:
                     'closed': market.get('closed', False),
                 }
             except Exception as e:
-            logger.error(f"获取市场失败: {e}")
+                logger.warning(f"尝试 {attempt+1}/3 获取市场失败: {e}")
+                if attempt < 2:
+                    time.sleep(5)  # 等待5秒后重试
+        
+        lark_notifier.send_alert(f"Slug: {slug}","连续3次请求或解析发生异常，已放弃重试")
         return None
     
     def get_current_hourly_market(self) -> Optional[Dict]:
@@ -379,6 +455,87 @@ class PolymarketBTCTrader:
             logger.debug(f"获取实时价格失败: {e}")
             return None
     
+    def monitor_order_status(self, order_id: str, timeout_seconds: Optional[int] = None) -> str:
+        """
+        监控订单状态直到成交或超时
+        timeout_seconds: 超时秒数。如果为None，默认监控直到当前小时结束前1分钟。
+        """
+        if timeout_seconds is None:
+             now = datetime.now()
+             next_hour = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)
+             end_time = next_hour - timedelta(minutes=1)
+             timeout_seconds = int((end_time - now).total_seconds())
+             if timeout_seconds < 10: 
+                 timeout_seconds = 10 # 至少监控10秒
+
+        logger.info(f"🕵️ 开始监控订单状态: {order_id} (超时: {timeout_seconds}s)")
+        start_monitor = datetime.now()
+        last_status = 'UNKNOWN'
+        
+        while (datetime.now() - start_monitor).total_seconds() < timeout_seconds:
+            try:
+                order_data = self.clob_client.get_order(order_id)
+                status = order_data.get('status')
+                last_status = status
+                
+                if status in ['MATCHED', 'FILLED']:
+                    logger.info(f"🎉 订单已成交! 状态: {status}")
+                    return status
+                elif status == 'CANCELED':
+                    logger.warning(f"🚫 订单已取消")
+                    return status
+                elif status == 'LIVE':
+                    logger.info(f"   订单挂单中 (LIVE)...")
+                    time.sleep(30)
+                else:
+                    logger.info(f"   订单状态: {status} (等待成交...)")
+                    time.sleep(30) # 轮询间隔
+            except Exception as e:
+                logger.warning(f"⚠️ 查询订单状态失败: {e}")
+                time.sleep(30)
+        
+        logger.info(f"⚠️ 监控超时，当前状态: {last_status}")
+        return last_status
+
+    def _log_to_db(self, result: Dict, market: Optional[Dict]):
+        """记录交易结果到数据库"""
+        try:
+            # 确定状态
+            status = 'UNKNOWN'
+            
+            if self.dry_run:
+                status = 'test'
+            else:
+                # 真实交易：优先使用 Poly 订单状态
+                # 如果没有 order_id，说明下单未成功发起，标记为 FAILED
+                if not result.get('order_id'):
+                    status = 'FAILED'
+                elif result.get('status'):
+                    status = result.get('status')
+                else:
+                    status = 'SUBMITTED' # 有ID但无状态，暂时标记为已提交
+            
+            log_data = {
+                'timestamp': result.get('timestamp'),
+                'market_title': result.get('market'),
+                'market_slug': market.get('slug') if market else result.get('market_slug'),
+                'model_name': 'Ensemble', # 记录为集成策略
+                'prediction': result.get('prediction'),
+                'probability': result.get('probability'),
+                'action': result.get('side'),
+                'price': result.get('price'),
+                'amount': result.get('amount'),
+                'size': result.get('size'),
+                'status': status,
+                'error_message': result.get('error'),
+                'order_id': result.get('order_id'),
+                'raw_data': result
+            }
+            self.db.log_trade(log_data)
+            logger.info(f"💾 交易记录已保存到数据库 (Status: {status})")
+        except Exception as e:
+            logger.error(f"❌ 保存数据库失败: {e}")
+
     def execute_trade(self, prediction: int, market: Dict) -> Dict:
         """执行交易"""
         result = {
@@ -388,25 +545,27 @@ class PolymarketBTCTrader:
             'success': False,
             'order': None,
             'error': None,
+            'order_id': None,
+            'status': None
         }
         
         if prediction == 1:  # 预测涨 -> 买UP
             token_id = market['tokens'][0]
-            price = market['outcome_prices'][0]
+            price = 0.5 # 暂时固定价格
             side = "UP"
         else:  # 预测跌 -> 买DOWN
             token_id = market['tokens'][1]
-            price = market['outcome_prices'][1]
+            price = 0.5 # 暂时固定价格
             side = "DOWN"
         
         result['side'] = side
         result['price'] = price
         
         # 检查价格限制
-        if price > self.max_price:
-            result['error'] = f"价格${price:.2f}超过限制${self.max_price:.2f}"
-            logger.error(f"❌ {result['error']}")
-            return result
+        # if price > self.max_price:
+        #     result['error'] = f"价格${price:.2f}超过限制${self.max_price:.2f}"
+        #     logger.error(f"❌ {result['error']}")
+        #     return result
         
         # 计算数量
         size = self.max_bet / price
@@ -423,6 +582,7 @@ class PolymarketBTCTrader:
             logger.info("🔸 [模拟模式] 未实际下单")
             result['success'] = True
             result['order'] = 'DRY_RUN'
+            result['status'] = 'simulated'
             return result
         
         # 实际下单
@@ -433,21 +593,60 @@ class PolymarketBTCTrader:
         
         try:
             from py_clob_client.order_builder.constants import BUY
+            from py_clob_client.clob_types import OrderArgs, OpenOrderParams
             
-            order = self.clob_client.create_limit_order(
-                token_id=token_id,
+            # 2. 检查现有挂单
+            order_id = None
+            try:
+                open_order_params = OpenOrderParams(asset_id=token_id)
+                existing_orders = self.clob_client.get_orders(open_order_params)
+                if existing_orders:
+                    existing_order = existing_orders[0]
+                    order_id = existing_order.get('orderID')
+                    logger.warning(f"⚠️ 检测到已有挂单: {order_id}，跳过下单")
+                    result['order_id'] = order_id
+                    result['success'] = True
+            except Exception as e:
+                logger.warning(f"⚠️ 检查挂单失败 (不影响下单): {e}")
+
+            if not order_id:
+                # 创建订单参数
+                order_args = OrderArgs(
                     price=price,
                     size=size,
                     side=BUY,
+                    token_id=token_id,
                 )
                 
-            response = self.clob_client.post_order(order)
-            logger.info(f"✓ 订单已提交: {response}")
+                # 3. 原子化下单 (Create & Post)
+                logger.info(f"正在下单... (Price: {price}, Size: {size})")
+                response = self.clob_client.create_and_post_order(order_args)
+                
+                if response and response.get('success'):
+                    order_id = response.get('orderID')
+                    result['order_id'] = order_id
                     result['success'] = True
                     result['order'] = response
+                    logger.info(f"✓ 订单已提交, Order ID: {order_id}")
+                else:
+                    error_msg = response.get('errorMsg') if response else "未知错误"
+                    result['error'] = f"下单失败: {error_msg}"
+                    logger.error(f"❌ {result['error']}")
+                    logger.error(f"   Response: {response}")
+
+            # 4. 立即查询状态 (无论是新单还是旧单)
+            if order_id:
+                try:
+                    order_status = self.clob_client.get_order(order_id)
+                    status = order_status.get('status')
+                    result['status'] = status
+                    logger.info(f"   订单状态: {status}")
+                except Exception as e:
+                    logger.warning(f"⚠️ 无法获取订单状态: {e}")
+
         except Exception as e:
             result['error'] = str(e)
-            logger.error(f"❌ 下单失败: {e}")
+            logger.error(f"❌ 下单异常: {e}")
         
         return result
     
@@ -662,60 +861,108 @@ class PolymarketBTCTrader:
             return result
             
         # 检查是否已交易过该市场
-        check_status = 'test' if self.dry_run else 'success'
-        if self.db.check_existing_trade(market['slug'], status=check_status):
+        existing_trade = self.db.get_existing_trade(market['slug'])
+        
+        if existing_trade:
+            status = existing_trade.get('status')
+            order_id = existing_trade.get('order_id')
+            
+            # 如果是成功状态或模拟测试，且不需要进一步同步，则直接跳过
+            if status in ['test', 'MATCHED', 'FILLED']:
+                # 但如果是 dry_run 模式，且 existing_trade 也是 test，则认为是已完成
+                if self.dry_run and status == 'test':
+                    result['error'] = '该市场已存在模拟交易记录'
+                    logger.warning(f"⚠️ {result['error']}, 跳过交易")
+                    return result
+                elif not self.dry_run and status in ['MATCHED', 'FILLED']:
                     result['error'] = '该市场已存在成功交易记录'
                     logger.warning(f"⚠️ {result['error']}, 跳过交易")
                     return result
             
+            # 如果存在未完成的订单 (且不是 dry_run)，尝试同步状态
+            # FAILED 和 CANCELED 可以重试
+            if not self.dry_run and order_id and status not in ['FAILED', 'CANCELED', 'test']:
+                logger.info(f"🔄 发现已有订单 ({status}), 正在同步状态... OrderID: {order_id}")
+                try:
+                    # 使用较短的超时时间进行一次快速检查
+                    new_status = self.monitor_order_status(order_id, timeout_seconds=10)
+                    
+                    if new_status != status:
+                        self.db.update_trade_status(existing_trade['id'], new_status)
+                        logger.info(f"📝 数据库状态已更新: {status} -> {new_status}")
+                    
+                    if new_status in ['MATCHED', 'FILLED']:
+                        result['status'] = new_status
+                        result['error'] = '该市场已存在成功交易记录 (已同步)'
+                        return result
+                    elif new_status == 'CANCELED':
+                        logger.warning("⚠️ 原订单已取消，将尝试重新下单...")
+                        # 允许后续逻辑继续执行，尝试重新下单
+                        pass
+                    else:
+                        logger.info(f"⏳ 订单状态仍为 {new_status}，继续等待或根据策略处理...")
+                        # 如果还在 OPEN/LIVE，可能需要取消重下或者继续等待？
+                        # 这里简单处理：如果还在挂单中，暂不重复下单
+                        if new_status in ['OPEN', 'LIVE', 'pending']:
+                             result['error'] = f'该市场已存在挂单 ({new_status})'
+                             result['order_id'] = order_id
+                             return result
+                except Exception as e:
+                    logger.warning(f"⚠️ 同步订单状态失败: {e}")
+
         # 3. 循环检查价格并交易
-        start_time = datetime.now()
-        end_time = start_time + timedelta(minutes=timeout_minutes)
+        # start_time = datetime.now()
+        # end_time = start_time + timedelta(minutes=timeout_minutes)
         
-        logger.info(f"🔄 开始价格监控 (预测: {'UP' if prediction == 1 else 'DOWN'})...")
+        # logger.info(f"🔄 开始价格监控 (预测: {'UP' if prediction == 1 else 'DOWN'})...")
         
-        while datetime.now() < end_time:
+        # while datetime.now() < end_time:
             # 刷新价格
-            market = self.refresh_prices()
-            if not market:
-                logger.warning("无法获取市场价格，重试中...")
-                time.sleep(10)
-                continue
+            # market = self.refresh_prices()
+            # if not market:
+            #     logger.warning("无法获取市场价格，重试中...")
+            #     time.sleep(10)
+            #     continue
             
             # 检查价格
-            price_idx = 0 if prediction == 1 else 1
-            current_price = market['outcome_prices'][price_idx]
+            # price_idx = 0 if prediction == 1 else 1
+            # current_price = market['outcome_prices'][price_idx]
             
-            logger.info(f"   当前价格: ${current_price:.3f} (目标: <= ${self.max_price:.2f})")
+            # logger.info(f"   当前价格: ${current_price:.3f} (目标: <= ${self.max_price:.2f})")
             
-            if current_price <= self.max_price:
+            # if current_price <= self.max_price:
                 # 价格符合，执行交易
-                logger.info("✅ 价格满足条件，正在下单...")
+        logger.info("✅ 预测完成，正在下单(限价 0.5)...")
         trade_result = self.execute_trade(prediction, market)
         result.update(trade_result)
         
+        # 如果下单成功且有order_id，进入状态监控
+        if result.get('success') and result.get('order_id'):
+            final_status = self.monitor_order_status(result['order_id'])
+            result['status'] = final_status
+        
         # 记录数据库
         self._log_to_db(result, market)
         
         logger.info("=" * 60)
         return result
-            else:
-                # 价格过高，等待
-                wait_time = 30  # 等待30秒
-                elapsed = int((datetime.now() - start_time).total_seconds())
-                logger.info(f"⏳ 价格过高，继续等待... (已耗时: {elapsed}s)")
-                time.sleep(wait_time)
+            # else:
+            #     # 价格过高，等待
+            #     wait_time = 30  # 等待30秒
+            #     elapsed = int((datetime.now() - start_time).total_seconds())
+            #     logger.info(f"⏳ 价格过高，继续等待... (已耗时: {elapsed}s)")
+            #     time.sleep(wait_time)
         
         # 超时
-        result['error'] = f'等待价格超时 ({timeout_minutes}分钟)'
-        result['price'] = market['outcome_prices'][0] if prediction == 1 else market['outcome_prices'][1]
-        result['side'] = 'UP' if prediction == 1 else 'DOWN'
-        logger.error(f"❌ {result['error']}")
+        # result['error'] = f'等待价格超时 ({timeout_minutes}分钟)'
+        # result['price'] = market['outcome_prices'][0] if prediction == 1 else market['outcome_prices'][1]
+        # result['side'] = 'UP' if prediction == 1 else 'DOWN'
+        # logger.error(f"❌ {result['error']}")
         
-        self._log_to_db(result, market)
-        logger.info("=" * 60)
+        # self._log_to_db(result, market)
+        # logger.info("=" * 60)
         
-        return result
+        # return result
     
     def _show_initial_prediction(self):
         """
@@ -790,9 +1037,9 @@ def main():
                        help='模拟模式(不实际下单)')
     parser.add_argument('--real', action='store_true',
                        help='真实交易模式')
-    parser.add_argument('--max-bet', type=float, default=1.0,
+    parser.add_argument('--max-bet', type=float, default=2.5,
                        help='最大下注金额(USDC)')
-    parser.add_argument('--max-price', type=float, default=0.50,
+    parser.add_argument('--max-price', type=float, default=0.5,
                        help='最高可接受价格')
     parser.add_argument('--once', action='store_true',
                        help='只执行一次(不启动定时器)')

utils/send_msg.py

@@ -0,0 +1,41 @@
+import requests
+
+class LarkNotifier:
+    """飞书消息发送工具类"""
+    
+    def __init__(self, bot_name="weibo", receive_id="on_f972f5f8c4de6ac967574b97dc8c1d82"):
+        # 将基础配置放在初始化中，方便后续修改或通过环境变量传入
+        self.url = "https://zhiyinzhe.xyz/sendMsg/api/send"
+        self.bot_name = bot_name
+        self.receive_id = receive_id
+
+        # 代理固定为本地 7898 端口
+        self.proxies = {
+            "http": "socks5h://127.0.0.1:7898",
+            "https": "socks5h://127.0.0.1:7898"
+        }
+
+    def send_alert(self, title: str, content: str):
+        """
+        发送通用文本告警
+        :param title: 告警标题 (如 "⚠️ 市场获取失败")
+        :param content: 告警具体内容 (如 "Slug: xxx \n 原因: xxx")
+        """
+        payload = {
+            "bot_name": self.bot_name,
+            "receive_id": self.receive_id,
+            "receive_id_type": "union_id",
+            "content": f"{title}\n{content}"
+        }
+        
+        try:
+            # 每次请求直接调用 requests.post，不复用 Session
+            response = requests.post(self.url, data=payload, timeout=5, proxies=self.proxies)
+            response.raise_for_status()
+            print(f"告警发送成功: {title}")
+        except Exception as e:
+            # 告警系统自身的错误只记录日志，绝不抛出异常影响主程序
+            print(f"发送飞书告警失败: {e}")
+
+# 实例化一个全局单例供其他模块直接导入使用
+lark_notifier = LarkNotifier()