feat: 添加 cloudflare_pages、mysql_d1_backup_scheduler，更新 start_system、ensemble_models 及配置文件

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

.claude/settings.local.json

@@ -17,7 +17,8 @@
       "Bash(git -C /Users/xxz/PycharmProjects/crypto_predictor check-ignore -v test_dlinear_jan_2026_create_labels.py)",
       "Bash(git -C /Users/xxz/PycharmProjects/crypto_predictor status)",
       "Bash(git -C /Users/xxz/PycharmProjects/crypto_predictor diff)",
-      "Bash(git -C /Users/xxz/PycharmProjects/crypto_predictor log --oneline -5)"
+      "Bash(git -C /Users/xxz/PycharmProjects/crypto_predictor log --oneline -5)",
+      "Bash(git checkout:*)"
     ]
   }
 }

.env.example

@@ -5,6 +5,10 @@ BINANCE_API_URL=https://api.binance.com
 SOCKS5_PROXY=socks5://127.0.0.1:7890
 USE_PROXY=true
 
+# send_msg 独立代理（可选，优先级高于 SOCKS5_PROXY）
+# SEND_MSG_PROXY=socks5://127.0.0.1:7891
+# SEND_MSG_PROXY_PORT=7891
+
 # Data Configuration
 DATA_CACHE_DIR=./data/cache
 LOOKBACK_1H=168      # 1 week for 1h data
@@ -37,3 +41,22 @@ MYSQL_PORT=3306
 MYSQL_USER=user
 MYSQL_PASSWORD=password
 MYSQL_DB=crypto_predictor
+
+# Cloudflare D1 Backup Configuration
+# 开关：true 开启 MySQL -> D1 定时备份（每小时 :30）
+D1_BACKUP_ENABLED=false
+CF_D1_API_TOKEN=your_cloudflare_api_token
+CF_D1_ACCOUNT_ID=your_cloudflare_account_id
+CF_D1_DATABASE_ID=your_d1_database_id
+# 备份批大小（单次 SQL 插入的最大行数）
+D1_BACKUP_BATCH_SIZE=500
+# 单条 SQL 最大参数数（D1 限制差异较大，默认保守）
+D1_MAX_SQL_PARAMS=90
+# 指定备份表（逗号分隔）；留空时自动备份 MySQL 全部表
+D1_BACKUP_TABLES=ohlcv,hourly_model_predictions,trades
+
+# 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

@@ -12,9 +12,13 @@ import httpx
 from typing import Optional, Dict, List, Any
 from dataclasses import dataclass
 from dotenv import load_dotenv
+from utils.network_proxy import apply_global_proxy_from_env, get_proxy_url
+from utils.project_logger import get_project_logger
 
 # 加载环境变量
 load_dotenv()
+apply_global_proxy_from_env()
+logger = get_project_logger("api.polymarket_client")
 
 
 @dataclass
@@ -46,7 +50,7 @@ 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 = get_proxy_url()
         
         # HTTP客户端 - 禁用压缩避免brotli问题
         self.http = httpx.Client(
@@ -79,9 +83,9 @@ class PolymarketClient:
             # 获取或创建API凭证
             self.clob_client.set_api_creds(self.clob_client.create_or_derive_api_creds())
             
-            print("✓ CLOB客户端初始化成功")
+            logger.info("CLOB客户端初始化成功")
         except Exception as e:
-            print(f"⚠️ CLOB客户端初始化失败: {e}")
+            logger.warning("CLOB客户端初始化失败: %s", e)
             self.clob_client = None
     
     def search_markets(self, query: str = "BTC", limit: int = 20) -> List[Dict]:
@@ -111,7 +115,7 @@ class PolymarketClient:
             filtered = [m for m in markets if query.lower() in m.get("question", "").lower()]
             return filtered
         except Exception as e:
-            print(f"搜索市场失败: {e}")
+            logger.error("搜索市场失败: %s", e)
             return []
     
     def search_btc_1h_markets(self) -> List[Dict]:
@@ -150,7 +154,7 @@ class PolymarketClient:
             
             return btc_1h_markets
         except Exception as e:
-            print(f"搜索BTC 1H市场失败: {e}")
+            logger.error("搜索BTC 1H市场失败: %s", e)
             return []
     
     def get_market_info(self, condition_id: str) -> Optional[MarketInfo]:
@@ -179,7 +183,7 @@ class PolymarketClient:
                 end_date=data.get("endDate", ""),
             )
         except Exception as e:
-            print(f"获取市场信息失败: {e}")
+            logger.error("获取市场信息失败: %s", e)
             return None
     
     def get_orderbook(self, token_id: str) -> Dict:
@@ -200,7 +204,7 @@ class PolymarketClient:
             response.raise_for_status()
             return response.json()
         except Exception as e:
-            print(f"获取订单簿失败: {e}")
+            logger.error("获取订单簿失败: %s", e)
             return {}
     
     def get_price(self, token_id: str) -> Optional[float]:
@@ -222,7 +226,7 @@ class PolymarketClient:
             data = response.json()
             return float(data.get("price", 0))
         except Exception as e:
-            print(f"获取价格失败: {e}")
+            logger.error("获取价格失败: %s", e)
             return None
     
     def place_market_order(
@@ -243,7 +247,7 @@ class PolymarketClient:
             订单响应
         """
         if not self.clob_client:
-            print("❌ CLOB客户端未初始化，无法下单")
+            logger.error("CLOB客户端未初始化，无法下单")
             return None
         
         try:
@@ -260,10 +264,10 @@ class PolymarketClient:
             
             # 提交订单
             response = self.clob_client.post_order(order)
-            print(f"✓ 订单已提交: {response}")
+            logger.info("订单已提交: %s", response)
             return response
         except Exception as e:
-            print(f"❌ 下单失败: {e}")
+            logger.error("下单失败: %s", e)
             return None
     
     def place_limit_order(
@@ -286,7 +290,7 @@ class PolymarketClient:
             订单响应
         """
         if not self.clob_client:
-            print("❌ CLOB客户端未初始化，无法下单")
+            logger.error("CLOB客户端未初始化，无法下单")
             return None
         
         try:
@@ -304,10 +308,10 @@ class PolymarketClient:
             
             # 提交订单
             response = self.clob_client.post_order(order)
-            print(f"✓ 限价单已提交: {response}")
+            logger.info("限价单已提交: %s", response)
             return response
         except Exception as e:
-            print(f"❌ 下单失败: {e}")
+            logger.error("下单失败: %s", e)
             return None
     
     def get_positions(self) -> List[Dict]:
@@ -318,14 +322,14 @@ class PolymarketClient:
             仓位列表
         """
         if not self.clob_client:
-            print("❌ CLOB客户端未初始化")
+            logger.error("CLOB客户端未初始化")
             return []
         
         try:
             positions = self.clob_client.get_positions()
             return positions
         except Exception as e:
-            print(f"获取仓位失败: {e}")
+            logger.error("获取仓位失败: %s", e)
             return []
     
     def cancel_order(self, order_id: str) -> bool:
@@ -339,15 +343,15 @@ class PolymarketClient:
             是否成功
         """
         if not self.clob_client:
-            print("❌ CLOB客户端未初始化")
+            logger.error("CLOB客户端未初始化")
             return False
         
         try:
             self.clob_client.cancel(order_id)
-            print(f"✓ 订单已取消: {order_id}")
+            logger.info("订单已取消: %s", order_id)
             return True
         except Exception as e:
-            print(f"❌ 取消订单失败: {e}")
+            logger.error("取消订单失败: %s", e)
             return False
     
     def close(self):

cloudflare_pages/.wrangler/cache/pages.json

@@ -0,0 +1,4 @@
+{
+  "account_id": "7ea8a511cad4d467490fe36aa7d6638c",
+  "project_name": "trades"
+}

cloudflare_pages/.wrangler/cache/wrangler-account.json

@@ -0,0 +1,6 @@
+{
+  "account": {
+    "id": "7ea8a511cad4d467490fe36aa7d6638c",
+    "name": "Subuhuinew@gmail.com's Account"
+  }
+}

cloudflare_pages/README.md

@@ -0,0 +1,41 @@
+# Cloudflare Pages 看板
+
+本目录提供一个可直接部署到 Cloudflare Pages 的页面，数据源为 D1（由 MySQL 备份任务同步而来）。
+
+## 目录结构
+
+- `public/`：静态页面资源
+- `functions/api/`：Pages Functions，直接查询 D1
+
+## 必要前置
+
+1. 已在 D1 中有备份表：
+- `backup_trades`
+- `backup_hourly_model_predictions`
+
+2. Pages 项目绑定 D1（绑定名必须是 `DB`）
+
+## 本地开发
+
+```bash
+cd cloudflare_pages
+npx wrangler pages dev public --d1 DB=<你的D1数据库ID>
+```
+
+## 生产部署
+
+1. 在 Cloudflare Pages 新建项目，根目录设置为 `cloudflare_pages`
+2. Build command 留空
+3. Build output directory 设置为 `public`
+4. 在 Pages 项目 Settings -> Functions -> D1 bindings 中添加：
+- Variable name: `DB`
+- D1 database: 选择你的数据库
+
+## 页面接口
+
+- `/api/status`
+- `/api/trades/current`
+- `/api/trades/history?page=1&limit=15&month=YYYY-MM&day=YYYY-MM-DD`
+- `/api/accuracy/monthly?month=YYYY-MM`
+- `/api/accuracy/daily?month=YYYY-MM&page=1&limit=10`
+- `/api/accuracy/hourly/YYYY-MM-DD`

cloudflare_pages/functions/api/accuracy/daily.js

@@ -0,0 +1,101 @@
+function json(data, status = 200) {
+  return new Response(JSON.stringify(data), {
+    status,
+    headers: { "content-type": "application/json; charset=UTF-8" },
+  });
+}
+
+function toInt(v, d) {
+  const n = Number(v);
+  return Number.isFinite(n) && n > 0 ? Math.floor(n) : d;
+}
+
+function ratio(correct, total) {
+  if (!total) return 0;
+  return Math.round((correct / total) * 10000) / 100;
+}
+
+function item(total, correct) {
+  const t = Number(total || 0);
+  const c = Number(correct || 0);
+  return {
+    total: t,
+    correct: c,
+    incorrect: Math.max(0, t - c),
+    accuracy: ratio(c, t),
+  };
+}
+
+export async function onRequestGet(context) {
+  const db = context.env.DB;
+  if (!db) {
+    return json({ success: false, error: "缺少 D1 绑定 DB" }, 500);
+  }
+
+  try {
+    const url = new URL(context.request.url);
+    const month = (url.searchParams.get("month") || new Date().toISOString().slice(0, 7)).trim();
+    const page = toInt(url.searchParams.get("page"), 1);
+    const limit = Math.min(100, toInt(url.searchParams.get("limit"), 10));
+    const offset = (page - 1) * limit;
+
+    const baseWhere = "date LIKE ?";
+    const monthArg = `${month}%`;
+
+    const dataSql = `
+      SELECT
+        date,
+        SUM(CASE WHEN actual_result IS NOT NULL AND ensemble_prediction IS NOT NULL THEN 1 ELSE 0 END) AS ens_total,
+        SUM(CASE WHEN ensemble_correct = 1 THEN 1 ELSE 0 END) AS ens_correct,
+
+        SUM(CASE WHEN actual_result IS NOT NULL AND dlinear_prediction IS NOT NULL THEN 1 ELSE 0 END) AS dl_total,
+        SUM(CASE WHEN dlinear_correct = 1 THEN 1 ELSE 0 END) AS dl_correct,
+
+        SUM(CASE WHEN actual_result IS NOT NULL AND nbeats_prediction IS NOT NULL THEN 1 ELSE 0 END) AS nb_total,
+        SUM(CASE WHEN nbeats_correct = 1 THEN 1 ELSE 0 END) AS nb_correct,
+
+        SUM(CASE WHEN actual_result IS NOT NULL AND ttm_prediction IS NOT NULL THEN 1 ELSE 0 END) AS ttm_total,
+        SUM(CASE WHEN ttm_correct = 1 THEN 1 ELSE 0 END) AS ttm_correct
+      FROM backup_hourly_model_predictions
+      WHERE ${baseWhere}
+      GROUP BY date
+      ORDER BY date DESC
+      LIMIT ? OFFSET ?
+    `;
+
+    const countSql = `
+      SELECT COUNT(*) AS cnt
+      FROM (
+        SELECT date
+        FROM backup_hourly_model_predictions
+        WHERE ${baseWhere}
+        GROUP BY date
+      ) x
+    `;
+
+    const dataResult = await db.prepare(dataSql).bind(monthArg, limit, offset).all();
+    const countResult = await db.prepare(countSql).bind(monthArg).first();
+
+    const rows = (dataResult.results || []).map((r) => ({
+      date: r.date,
+      Ensemble: item(r.ens_total, r.ens_correct),
+      DLinear: item(r.dl_total, r.dl_correct),
+      "N-BEATS": item(r.nb_total, r.nb_correct),
+      "TTM-R2": item(r.ttm_total, r.ttm_correct),
+    }));
+
+    const total = Number(countResult?.cnt || 0);
+    const totalPages = total > 0 ? Math.ceil(total / limit) : 1;
+
+    return json({
+      success: true,
+      data: rows,
+      page,
+      limit,
+      total,
+      total_pages: totalPages,
+    });
+  } catch (error) {
+    return json({ success: false, error: String(error) }, 500);
+  }
+}

cloudflare_pages/functions/api/accuracy/hourly/[date].js

@@ -0,0 +1,67 @@
+function json(data, status = 200) {
+  return new Response(JSON.stringify(data), {
+    status,
+    headers: { "content-type": "application/json; charset=UTF-8" },
+  });
+}
+
+function modelItem(prediction, probability, isCorrect) {
+  if (prediction === null || prediction === undefined) return null;
+  return {
+    prediction: Number(prediction),
+    probability: probability === null || probability === undefined ? null : Number(probability),
+    is_correct: isCorrect === null || isCorrect === undefined ? null : Number(isCorrect) === 1,
+  };
+}
+
+export async function onRequestGet(context) {
+  const db = context.env.DB;
+  if (!db) {
+    return json({ success: false, error: "缺少 D1 绑定 DB" }, 500);
+  }
+
+  try {
+    const date = context.params.date;
+    const sql = `
+      SELECT
+        timestamp,
+        hour,
+        open_price,
+        close_price,
+        actual_result,
+        dlinear_prediction,
+        dlinear_prob,
+        dlinear_correct,
+        nbeats_prediction,
+        nbeats_prob,
+        nbeats_correct,
+        ttm_prediction,
+        ttm_prob,
+        ttm_correct,
+        ensemble_prediction,
+        ensemble_correct
+      FROM backup_hourly_model_predictions
+      WHERE date = ?
+      ORDER BY hour ASC
+    `;
+
+    const result = await db.prepare(sql).bind(date).all();
+    const rows = (result.results || []).map((r) => ({
+      timestamp_cst: r.timestamp,
+      hour: r.hour,
+      open_price: r.open_price === null || r.open_price === undefined ? null : Number(r.open_price),
+      close_price: r.close_price === null || r.close_price === undefined ? null : Number(r.close_price),
+      actual_result: r.actual_result === null || r.actual_result === undefined ? null : Number(r.actual_result),
+      models: {
+        Ensemble: modelItem(r.ensemble_prediction, null, r.ensemble_correct),
+        DLinear: modelItem(r.dlinear_prediction, r.dlinear_prob, r.dlinear_correct),
+        "N-BEATS": modelItem(r.nbeats_prediction, r.nbeats_prob, r.nbeats_correct),
+        "TTM-R2": modelItem(r.ttm_prediction, r.ttm_prob, r.ttm_correct),
+      },
+    }));
+
+    return json({ success: true, date, data: rows });
+  } catch (error) {
+    return json({ success: false, error: String(error) }, 500);
+  }
+}

cloudflare_pages/functions/api/accuracy/monthly.js

@@ -0,0 +1,87 @@
+function json(data, status = 200) {
+  return new Response(JSON.stringify(data), {
+    status,
+    headers: { "content-type": "application/json; charset=UTF-8" },
+  });
+}
+
+function ratio(correct, total) {
+  if (!total) return 0;
+  return Math.round((correct / total) * 10000) / 100;
+}
+
+export async function onRequestGet(context) {
+  const db = context.env.DB;
+  if (!db) {
+    return json({ success: false, error: "缺少 D1 绑定 DB" }, 500);
+  }
+
+  try {
+    const url = new URL(context.request.url);
+    const month = (url.searchParams.get("month") || new Date().toISOString().slice(0, 7)).trim();
+
+    const sql = `
+      SELECT
+        SUM(CASE WHEN actual_result IS NOT NULL AND ensemble_prediction IS NOT NULL THEN 1 ELSE 0 END) AS ens_total,
+        SUM(CASE WHEN ensemble_correct = 1 THEN 1 ELSE 0 END) AS ens_correct,
+
+        SUM(CASE WHEN actual_result IS NOT NULL AND dlinear_prediction IS NOT NULL THEN 1 ELSE 0 END) AS dl_total,
+        SUM(CASE WHEN dlinear_correct = 1 THEN 1 ELSE 0 END) AS dl_correct,
+
+        SUM(CASE WHEN actual_result IS NOT NULL AND nbeats_prediction IS NOT NULL THEN 1 ELSE 0 END) AS nb_total,
+        SUM(CASE WHEN nbeats_correct = 1 THEN 1 ELSE 0 END) AS nb_correct,
+
+        SUM(CASE WHEN actual_result IS NOT NULL AND ttm_prediction IS NOT NULL THEN 1 ELSE 0 END) AS ttm_total,
+        SUM(CASE WHEN ttm_correct = 1 THEN 1 ELSE 0 END) AS ttm_correct
+      FROM backup_hourly_model_predictions
+      WHERE date LIKE ?
+    `;
+
+    const row = await db.prepare(sql).bind(`${month}%`).first();
+
+    const ensTotal = Number(row?.ens_total || 0);
+    const ensCorrect = Number(row?.ens_correct || 0);
+
+    const dlTotal = Number(row?.dl_total || 0);
+    const dlCorrect = Number(row?.dl_correct || 0);
+
+    const nbTotal = Number(row?.nb_total || 0);
+    const nbCorrect = Number(row?.nb_correct || 0);
+
+    const ttmTotal = Number(row?.ttm_total || 0);
+    const ttmCorrect = Number(row?.ttm_correct || 0);
+
+    return json({
+      success: true,
+      month,
+      data: {
+        Ensemble: {
+          total: ensTotal,
+          correct: ensCorrect,
+          incorrect: Math.max(0, ensTotal - ensCorrect),
+          accuracy: ratio(ensCorrect, ensTotal),
+        },
+        DLinear: {
+          total: dlTotal,
+          correct: dlCorrect,
+          incorrect: Math.max(0, dlTotal - dlCorrect),
+          accuracy: ratio(dlCorrect, dlTotal),
+        },
+        "N-BEATS": {
+          total: nbTotal,
+          correct: nbCorrect,
+          incorrect: Math.max(0, nbTotal - nbCorrect),
+          accuracy: ratio(nbCorrect, nbTotal),
+        },
+        "TTM-R2": {
+          total: ttmTotal,
+          correct: ttmCorrect,
+          incorrect: Math.max(0, ttmTotal - ttmCorrect),
+          accuracy: ratio(ttmCorrect, ttmTotal),
+        },
+      },
+    });
+  } catch (error) {
+    return json({ success: false, error: String(error) }, 500);
+  }
+}

cloudflare_pages/functions/api/status.js

@@ -0,0 +1,15 @@
+function json(data, status = 200) {
+  return new Response(JSON.stringify(data), {
+    status,
+    headers: { "content-type": "application/json; charset=UTF-8" },
+  });
+}
+
+export async function onRequestGet(context) {
+  return json({
+    success: true,
+    status: "running",
+    server_time: new Date().toISOString(),
+    version: "pages-1.0.0",
+  });
+}

cloudflare_pages/functions/api/trades/current.js

@@ -0,0 +1,23 @@
+function json(data, status = 200) {
+  return new Response(JSON.stringify(data), {
+    status,
+    headers: { "content-type": "application/json; charset=UTF-8" },
+  });
+}
+
+export async function onRequestGet(context) {
+  const db = context.env.DB;
+  if (!db) {
+    return json({ success: false, error: "缺少 D1 绑定 DB" }, 500);
+  }
+
+  try {
+    const row = await db
+      .prepare("SELECT * FROM backup_trades ORDER BY timestamp DESC LIMIT 1")
+      .first();
+
+    return json({ success: true, data: row || null });
+  } catch (error) {
+    return json({ success: false, error: String(error) }, 500);
+  }
+}

cloudflare_pages/functions/api/trades/history.js

@@ -0,0 +1,71 @@
+function json(data, status = 200) {
+  return new Response(JSON.stringify(data), {
+    status,
+    headers: { "content-type": "application/json; charset=UTF-8" },
+  });
+}
+
+function toInt(v, d) {
+  const n = Number(v);
+  return Number.isFinite(n) && n > 0 ? Math.floor(n) : d;
+}
+
+export async function onRequestGet(context) {
+  const db = context.env.DB;
+  if (!db) {
+    return json({ success: false, error: "缺少 D1 绑定 DB" }, 500);
+  }
+
+  try {
+    const url = new URL(context.request.url);
+    const page = toInt(url.searchParams.get("page"), 1);
+    const limit = Math.min(100, toInt(url.searchParams.get("limit"), 10));
+    const month = (url.searchParams.get("month") || "").trim();
+    const day = (url.searchParams.get("day") || "").trim();
+    const offset = (page - 1) * limit;
+
+    const where = ["1=1"];
+    const params = [];
+
+    // 时间字段是 YYYY-MM-DD HH:MM:SS 文本，直接前缀过滤可走索引/减少转换成本
+    if (month) {
+      where.push("timestamp LIKE ?");
+      params.push(`${month}%`);
+    }
+    if (day) {
+      where.push("timestamp LIKE ?");
+      params.push(`${day}%`);
+    }
+
+    const whereSql = where.join(" AND ");
+    const dataSql = `
+      SELECT *
+      FROM backup_trades
+      WHERE ${whereSql}
+      ORDER BY timestamp DESC
+      LIMIT ? OFFSET ?
+    `;
+    const countSql = `
+      SELECT COUNT(*) AS cnt
+      FROM backup_trades
+      WHERE ${whereSql}
+    `;
+
+    const dataResult = await db.prepare(dataSql).bind(...params, limit, offset).all();
+    const countResult = await db.prepare(countSql).bind(...params).first();
+
+    const total = Number(countResult?.cnt || 0);
+    const totalPages = total > 0 ? Math.ceil(total / limit) : 1;
+
+    return json({
+      success: true,
+      data: dataResult.results || [],
+      page,
+      limit,
+      total,
+      total_pages: totalPages,
+    });
+  } catch (error) {
+    return json({ success: false, error: String(error) }, 500);
+  }
+}

cloudflare_pages/public/app.js

@@ -0,0 +1,200 @@
+const { createApp } = Vue;
+
+const app = createApp({
+  data() {
+    return {
+      loading: false,
+      activeTab: 'trades',
+      serverTime: '',
+      currentTrade: null,
+      tradeHistory: {
+        data: [],
+        page: 1,
+        limit: 10,
+        total: 0,
+        total_pages: 0,
+      },
+      tradeFilter: {
+        month: '',
+        day: '',
+      },
+      accuracyFilter: {
+        month: '',
+      },
+      monthlyStats: {},
+      dailyStats: {
+        data: [],
+        page: 1,
+        limit: 10,
+        total: 0,
+        total_pages: 0,
+      },
+      hourlyModal: {
+        date: '',
+        data: [],
+      },
+    };
+  },
+
+  mounted() {
+    this.accuracyFilter.month = this.getCurrentMonth();
+    this.loadCurrentTrade();
+    this.loadTradeHistory(1);
+    this.updateServerTime();
+    setInterval(() => this.updateServerTime(), 60000);
+  },
+
+  methods: {
+    getCurrentMonth() {
+      const now = new Date();
+      return `${now.getFullYear()}-${String(now.getMonth() + 1).padStart(2, '0')}`;
+    },
+
+    updateServerTime() {
+      const now = new Date();
+      this.serverTime = now.toLocaleString('zh-CN', {
+        year: 'numeric',
+        month: '2-digit',
+        day: '2-digit',
+        hour: '2-digit',
+        minute: '2-digit',
+      });
+    },
+
+    formatTime(timestamp) {
+      if (!timestamp) return '-';
+      return String(timestamp).slice(0, 19);
+    },
+
+    formatPercent(value) {
+      if (value === null || value === undefined) return '-';
+      return `${(Number(value) * 100).toFixed(1)}%`;
+    },
+
+    getPredictionClass(prediction) {
+      if (prediction === 'UP') return 'text-success fw-bold';
+      if (prediction === 'DOWN') return 'text-danger fw-bold';
+      return '';
+    },
+
+    getStatusClass(status) {
+      switch (status) {
+        case 'success': return 'text-success';
+        case 'test': return 'text-info';
+        case 'failure': return 'text-danger';
+        case 'in_progress': return 'text-warning';
+        default: return 'text-muted';
+      }
+    },
+
+    getStatusText(status) {
+      switch (status) {
+        case 'success': return '成功';
+        case 'test': return '测试';
+        case 'failure': return '失败';
+        case 'in_progress': return '进行中';
+        default: return status || '-';
+      }
+    },
+
+    getAccuracyClass(accuracy) {
+      if (accuracy >= 60) return 'text-success fw-bold';
+      if (accuracy >= 50) return 'text-warning';
+      return 'text-danger';
+    },
+
+    async fetchAPI(url) {
+      try {
+        const response = await fetch(url);
+        return await response.json();
+      } catch (error) {
+        return { success: false, error: String(error) };
+      }
+    },
+
+    async loadCurrentTrade() {
+      const data = await this.fetchAPI('/api/trades/current');
+      if (data.success) {
+        this.currentTrade = data.data;
+      }
+    },
+
+    async loadTradeHistory(page) {
+      if (page < 1) return;
+      this.loading = true;
+
+      let url = `/api/trades/history?page=${page}&limit=${this.tradeHistory.limit}`;
+      if (this.tradeFilter.month) {
+        url += `&month=${this.tradeFilter.month}`;
+      }
+      if (this.tradeFilter.day) {
+        url += `&day=${this.tradeFilter.day}`;
+      }
+
+      const data = await this.fetchAPI(url);
+      if (data.success) {
+        this.tradeHistory = {
+          data: data.data,
+          page: data.page,
+          limit: data.limit,
+          total: data.total,
+          total_pages: data.total_pages,
+        };
+      }
+
+      this.loading = false;
+    },
+
+    clearTradeFilter() {
+      this.tradeFilter.month = '';
+      this.tradeFilter.day = '';
+      this.loadTradeHistory(1);
+    },
+
+    async loadAccuracyData() {
+      this.loading = true;
+      const monthlyData = await this.fetchAPI(`/api/accuracy/monthly?month=${this.accuracyFilter.month}`);
+      if (monthlyData.success) {
+        this.monthlyStats = monthlyData.data || {};
+      }
+      await this.loadDailyStats(1);
+      this.loading = false;
+    },
+
+    async loadDailyStats(page) {
+      if (page < 1) return;
+      const data = await this.fetchAPI(`/api/accuracy/daily?month=${this.accuracyFilter.month}&page=${page}&limit=10`);
+      if (data.success) {
+        this.dailyStats = {
+          data: data.data,
+          page: data.page,
+          limit: data.limit,
+          total: data.total,
+          total_pages: data.total_pages,
+        };
+      }
+    },
+
+    async showHourlyDetails(date) {
+      this.loading = true;
+      this.hourlyModal.date = date;
+      const data = await this.fetchAPI(`/api/accuracy/hourly/${date}`);
+      if (data.success) {
+        this.hourlyModal.data = data.data || [];
+      }
+      this.loading = false;
+      const modal = new bootstrap.Modal(document.getElementById('hourlyModal'));
+      modal.show();
+    },
+  },
+
+  watch: {
+    activeTab(newTab) {
+      if (newTab === 'accuracy' && !this.dailyStats.data.length) {
+        this.loadAccuracyData();
+      }
+    },
+  },
+});
+
+app.mount('#app');

cloudflare_pages/public/index.html

@@ -0,0 +1,251 @@
+<!DOCTYPE html>
+<html lang="zh-CN">
+<head>
+  <meta charset="UTF-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1.0">
+  <title>BTC 预测交易看板 (Pages + D1)</title>
+  <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.2/dist/css/bootstrap.min.css" rel="stylesheet">
+  <link href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.11.1/font/bootstrap-icons.css" rel="stylesheet">
+  <link href="/style.css" rel="stylesheet">
+</head>
+<body>
+  <div id="app">
+    <nav class="navbar navbar-expand-lg navbar-dark bg-dark">
+      <div class="container">
+        <a class="navbar-brand" href="#">
+          <i class="bi bi-currency-bitcoin"></i> BTC 预测交易看板 (Pages)
+        </a>
+        <span class="navbar-text text-light">
+          <i class="bi bi-clock"></i> {{ serverTime }}
+        </span>
+      </div>
+    </nav>
+
+    <div class="container mt-4">
+      <ul class="nav nav-tabs mb-4" role="tablist">
+        <li class="nav-item">
+          <button class="nav-link" :class="{ active: activeTab === 'trades' }" @click="activeTab = 'trades'">
+            <i class="bi bi-graph-up"></i> 交易详情
+          </button>
+        </li>
+        <li class="nav-item">
+          <button class="nav-link" :class="{ active: activeTab === 'accuracy' }" @click="activeTab = 'accuracy'">
+            <i class="bi bi-bullseye"></i> 模型准确率
+          </button>
+        </li>
+      </ul>
+
+      <div v-show="activeTab === 'trades'">
+        <div class="card mb-4">
+          <div class="card-header bg-primary text-white">
+            <i class="bi bi-lightning-charge"></i> 当前小时交易
+          </div>
+          <div class="card-body">
+            <div v-if="currentTrade" class="row">
+              <div class="col-md-3">
+                <div class="stat-card">
+                  <div class="stat-label">市场</div>
+                  <div class="stat-value small">{{ currentTrade.market_title || '-' }}</div>
+                </div>
+              </div>
+              <div class="col-md-2">
+                <div class="stat-card">
+                  <div class="stat-label">预测方向</div>
+                  <div class="stat-value" :class="getPredictionClass(currentTrade.prediction)">
+                    {{ currentTrade.prediction || '-' }}
+                  </div>
+                </div>
+              </div>
+              <div class="col-md-2">
+                <div class="stat-card">
+                  <div class="stat-label">概率</div>
+                  <div class="stat-value">{{ formatPercent(currentTrade.probability) }}</div>
+                </div>
+              </div>
+              <div class="col-md-2">
+                <div class="stat-card">
+                  <div class="stat-label">策略</div>
+                  <div class="stat-value small">{{ currentTrade.strategy || 'Ensemble' }}</div>
+                </div>
+              </div>
+              <div class="col-md-3">
+                <div class="stat-card">
+                  <div class="stat-label">状态</div>
+                  <div class="stat-value" :class="getStatusClass(currentTrade.status)">
+                    {{ getStatusText(currentTrade.status) }}
+                  </div>
+                </div>
+              </div>
+            </div>
+            <div v-else class="text-center text-muted py-4">
+              <i class="bi bi-inbox"></i> 暂无交易记录
+            </div>
+          </div>
+        </div>
+
+        <div class="card">
+          <div class="card-header d-flex justify-content-between align-items-center">
+            <span><i class="bi bi-clock-history"></i> 历史交易记录</span>
+            <div class="d-flex gap-2">
+              <input type="month" class="form-control form-control-sm" v-model="tradeFilter.month" @change="loadTradeHistory(1)" style="width: 150px;">
+              <input type="date" class="form-control form-control-sm" v-model="tradeFilter.day" @change="loadTradeHistory(1)" style="width: 150px;">
+              <button class="btn btn-sm btn-outline-secondary" @click="clearTradeFilter">
+                <i class="bi bi-x-circle"></i> 清除
+              </button>
+            </div>
+          </div>
+          <div class="card-body p-0">
+            <div class="table-responsive">
+              <table class="table table-hover table-striped mb-0">
+                <thead class="table-light">
+                  <tr>
+                    <th>时间</th>
+                    <th>市场</th>
+                    <th>策略</th>
+                    <th>预测</th>
+                    <th>概率</th>
+                    <th>价格</th>
+                    <th>状态</th>
+                  </tr>
+                </thead>
+                <tbody>
+                  <tr v-for="trade in tradeHistory.data" :key="trade.id">
+                    <td>{{ formatTime(trade.timestamp) }}</td>
+                    <td class="text-truncate" style="max-width: 200px;">{{ trade.market_title }}</td>
+                    <td><span class="badge bg-secondary">{{ trade.strategy || 'Ensemble' }}</span></td>
+                    <td :class="getPredictionClass(trade.prediction)">{{ trade.prediction }}</td>
+                    <td>{{ formatPercent(trade.probability) }}</td>
+                    <td>{{ trade.price ?? '-' }}</td>
+                    <td :class="getStatusClass(trade.status)">{{ getStatusText(trade.status) }}</td>
+                  </tr>
+                  <tr v-if="!tradeHistory.data?.length"><td colspan="7" class="text-center text-muted py-4">暂无数据</td></tr>
+                </tbody>
+              </table>
+            </div>
+          </div>
+          <div class="card-footer d-flex justify-content-between align-items-center">
+            <span class="text-muted">共 {{ tradeHistory.total }} 条</span>
+            <nav>
+              <ul class="pagination pagination-sm mb-0">
+                <li class="page-item" :class="{ disabled: tradeHistory.page <= 1 }">
+                  <a class="page-link" href="#" @click.prevent="loadTradeHistory(tradeHistory.page - 1)">上一页</a>
+                </li>
+                <li class="page-item disabled"><span class="page-link">{{ tradeHistory.page }} / {{ tradeHistory.total_pages }}</span></li>
+                <li class="page-item" :class="{ disabled: tradeHistory.page >= tradeHistory.total_pages }">
+                  <a class="page-link" href="#" @click.prevent="loadTradeHistory(tradeHistory.page + 1)">下一页</a>
+                </li>
+              </ul>
+            </nav>
+          </div>
+        </div>
+      </div>
+
+      <div v-show="activeTab === 'accuracy'">
+        <div class="mb-4">
+          <div class="d-flex gap-2 align-items-center">
+            <label class="form-label mb-0">选择月份:</label>
+            <input type="month" class="form-control" v-model="accuracyFilter.month" @change="loadAccuracyData" style="width: 180px;">
+          </div>
+        </div>
+
+        <div class="row mb-4">
+          <div class="col-md-3"><div class="card h-100 border-primary shadow-sm"><div class="card-header bg-primary text-white text-center">集成策略</div><div class="card-body text-center" v-if="monthlyStats.Ensemble"><h2 class="display-6 text-primary fw-bold">{{ monthlyStats.Ensemble.accuracy }}%</h2><div class="text-muted mb-2">准确率</div><div class="small text-muted">{{ monthlyStats.Ensemble.correct }}/{{ monthlyStats.Ensemble.total }}</div></div><div v-else class="text-center text-muted py-3">暂无数据</div></div></div>
+          <div class="col-md-3"><div class="card h-100 border-secondary"><div class="card-header bg-secondary text-white text-center">DLinear</div><div class="card-body text-center" v-if="monthlyStats.DLinear"><h3 class="text-dark">{{ monthlyStats.DLinear.accuracy }}%</h3><div class="small text-muted">{{ monthlyStats.DLinear.correct }}/{{ monthlyStats.DLinear.total }}</div></div><div v-else class="text-center text-muted py-3">暂无数据</div></div></div>
+          <div class="col-md-3"><div class="card h-100 border-secondary"><div class="card-header bg-secondary text-white text-center">N-BEATS</div><div class="card-body text-center" v-if="monthlyStats['N-BEATS']"><h3 class="text-dark">{{ monthlyStats['N-BEATS'].accuracy }}%</h3><div class="small text-muted">{{ monthlyStats['N-BEATS'].correct }}/{{ monthlyStats['N-BEATS'].total }}</div></div><div v-else class="text-center text-muted py-3">暂无数据</div></div></div>
+          <div class="col-md-3"><div class="card h-100 border-secondary"><div class="card-header bg-secondary text-white text-center">TTM-R2</div><div class="card-body text-center" v-if="monthlyStats['TTM-R2']"><h3 class="text-dark">{{ monthlyStats['TTM-R2'].accuracy }}%</h3><div class="small text-muted">{{ monthlyStats['TTM-R2'].correct }}/{{ monthlyStats['TTM-R2'].total }}</div></div><div v-else class="text-center text-muted py-3">暂无数据</div></div></div>
+        </div>
+
+        <div class="card">
+          <div class="card-header"><i class="bi bi-calendar3"></i> 每日准确率概览</div>
+          <div class="card-body p-0">
+            <div class="table-responsive">
+              <table class="table table-hover mb-0 text-center">
+                <thead class="table-light">
+                  <tr>
+                    <th class="text-start ps-4">日期</th>
+                    <th>集成策略</th>
+                    <th>DLinear</th>
+                    <th>N-BEATS</th>
+                    <th>TTM-R2</th>
+                    <th>操作</th>
+                  </tr>
+                </thead>
+                <tbody>
+                  <tr v-for="day in dailyStats.data" :key="day.date">
+                    <td class="text-start ps-4">{{ day.date }}</td>
+                    <td><span v-if="day.Ensemble" :class="getAccuracyClass(day.Ensemble.accuracy)">{{ day.Ensemble.accuracy }}%</span><span v-else class="text-muted">-</span></td>
+                    <td><span v-if="day.DLinear" :class="getAccuracyClass(day.DLinear.accuracy)">{{ day.DLinear.accuracy }}%</span><span v-else class="text-muted">-</span></td>
+                    <td><span v-if="day['N-BEATS']" :class="getAccuracyClass(day['N-BEATS'].accuracy)">{{ day['N-BEATS'].accuracy }}%</span><span v-else class="text-muted">-</span></td>
+                    <td><span v-if="day['TTM-R2']" :class="getAccuracyClass(day['TTM-R2'].accuracy)">{{ day['TTM-R2'].accuracy }}%</span><span v-else class="text-muted">-</span></td>
+                    <td><button class="btn btn-sm btn-outline-primary" @click="showHourlyDetails(day.date)"><i class="bi bi-eye"></i> 详情</button></td>
+                  </tr>
+                  <tr v-if="!dailyStats.data?.length"><td colspan="6" class="text-center text-muted py-4">暂无数据</td></tr>
+                </tbody>
+              </table>
+            </div>
+          </div>
+          <div class="card-footer d-flex justify-content-between align-items-center">
+            <span class="text-muted">共 {{ dailyStats.total }} 天</span>
+            <nav>
+              <ul class="pagination pagination-sm mb-0">
+                <li class="page-item" :class="{ disabled: dailyStats.page <= 1 }"><a class="page-link" href="#" @click.prevent="loadDailyStats(dailyStats.page - 1)">上一页</a></li>
+                <li class="page-item disabled"><span class="page-link">{{ dailyStats.page }} / {{ dailyStats.total_pages }}</span></li>
+                <li class="page-item" :class="{ disabled: dailyStats.page >= dailyStats.total_pages }"><a class="page-link" href="#" @click.prevent="loadDailyStats(dailyStats.page + 1)">下一页</a></li>
+              </ul>
+            </nav>
+          </div>
+        </div>
+      </div>
+    </div>
+
+    <div class="modal fade" id="hourlyModal" tabindex="-1">
+      <div class="modal-dialog modal-xl modal-dialog-scrollable">
+        <div class="modal-content">
+          <div class="modal-header">
+            <h5 class="modal-title"><i class="bi bi-clock"></i> {{ hourlyModal.date }} 每小时预测详情</h5>
+            <button type="button" class="btn-close" data-bs-dismiss="modal"></button>
+          </div>
+          <div class="modal-body p-0">
+            <table class="table table-sm table-hover mb-0 text-center align-middle">
+              <thead class="table-light sticky-top">
+                <tr>
+                  <th>时间</th>
+                  <th>开盘</th>
+                  <th>收盘</th>
+                  <th>实际</th>
+                  <th class="table-primary">集成策略</th>
+                  <th>DLinear</th>
+                  <th>N-BEATS</th>
+                  <th>TTM-R2</th>
+                </tr>
+              </thead>
+              <tbody>
+                <tr v-for="hour in hourlyModal.data" :key="hour.hour">
+                  <td>{{ hour.timestamp_cst ? hour.timestamp_cst.split(' ')[1] : (hour.hour + ':00') }}</td>
+                  <td>{{ hour.open_price ?? '-' }}</td>
+                  <td>{{ hour.close_price ?? '-' }}</td>
+                  <td :class="hour.actual_result === 1 ? 'text-success fw-bold' : (hour.actual_result === 0 ? 'text-danger fw-bold' : 'text-muted')">{{ hour.actual_result === 1 ? 'UP' : (hour.actual_result === 0 ? 'DOWN' : '-') }}</td>
+                  <td class="table-primary bg-opacity-10"><span v-if="hour.models?.Ensemble" :class="hour.models.Ensemble.is_correct ? 'text-success' : 'text-danger'">{{ hour.models.Ensemble.prediction === 1 ? 'UP' : 'DOWN' }}</span></td>
+                  <td><span v-if="hour.models?.DLinear" :class="hour.models.DLinear.is_correct ? 'text-success' : 'text-danger'">{{ hour.models.DLinear.prediction === 1 ? 'UP' : 'DOWN' }}</span></td>
+                  <td><span v-if="hour.models?.['N-BEATS']" :class="hour.models['N-BEATS'].is_correct ? 'text-success' : 'text-danger'">{{ hour.models['N-BEATS'].prediction === 1 ? 'UP' : 'DOWN' }}</span></td>
+                  <td><span v-if="hour.models?.['TTM-R2']" :class="hour.models['TTM-R2'].is_correct ? 'text-success' : 'text-danger'">{{ hour.models['TTM-R2'].prediction === 1 ? 'UP' : 'DOWN' }}</span></td>
+                </tr>
+                <tr v-if="!hourlyModal.data?.length"><td colspan="8" class="text-center text-muted py-4">暂无数据</td></tr>
+              </tbody>
+            </table>
+          </div>
+          <div class="modal-footer"><button type="button" class="btn btn-secondary" data-bs-dismiss="modal">关闭</button></div>
+        </div>
+      </div>
+    </div>
+
+    <div v-if="loading" class="loading-overlay">
+      <div class="spinner-border text-primary" role="status"><span class="visually-hidden">加载中...</span></div>
+    </div>
+  </div>
+
+  <script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.2/dist/js/bootstrap.bundle.min.js"></script>
+  <script src="https://cdn.jsdelivr.net/npm/vue@3.4.15/dist/vue.global.prod.js"></script>
+  <script src="/app.js"></script>
+</body>
+</html>

cloudflare_pages/public/style.css

@@ -0,0 +1,118 @@
+body {
+  background-color: #f8f9fa;
+  font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, 'Helvetica Neue', Arial, sans-serif;
+}
+
+.navbar-brand {
+  font-weight: 600;
+}
+
+.navbar-brand i {
+  color: #f7931a;
+}
+
+.nav-tabs .nav-link {
+  color: #6c757d;
+  border: none;
+  padding: 0.75rem 1.5rem;
+}
+
+.nav-tabs .nav-link.active {
+  color: #0d6efd;
+  border-bottom: 3px solid #0d6efd;
+  background: transparent;
+}
+
+.nav-tabs .nav-link:hover:not(.active) {
+  border-bottom: 3px solid #dee2e6;
+}
+
+.card {
+  box-shadow: 0 0.125rem 0.25rem rgba(0, 0, 0, 0.075);
+  border-radius: 0.5rem;
+}
+
+.card-header {
+  font-weight: 600;
+  border-radius: 0.5rem 0.5rem 0 0 !important;
+}
+
+.stat-card {
+  padding: 1rem;
+  text-align: center;
+  background: #f8f9fa;
+  border-radius: 0.5rem;
+  height: 100%;
+}
+
+.stat-label {
+  font-size: 0.75rem;
+  color: #6c757d;
+  text-transform: uppercase;
+  letter-spacing: 0.5px;
+  margin-bottom: 0.25rem;
+}
+
+.stat-value {
+  font-size: 1.25rem;
+  font-weight: 600;
+}
+
+.table {
+  margin-bottom: 0;
+}
+
+.table th {
+  font-weight: 600;
+  font-size: 0.875rem;
+  text-transform: uppercase;
+  letter-spacing: 0.5px;
+  color: #6c757d;
+  border-bottom: 2px solid #dee2e6;
+}
+
+.table td {
+  vertical-align: middle;
+}
+
+.table-hover tbody tr:hover {
+  background-color: rgba(13, 110, 253, 0.05);
+}
+
+.loading-overlay {
+  position: fixed;
+  top: 0;
+  left: 0;
+  width: 100%;
+  height: 100%;
+  background: rgba(255, 255, 255, 0.8);
+  display: flex;
+  align-items: center;
+  justify-content: center;
+  z-index: 9999;
+}
+
+.modal-content {
+  border-radius: 0.5rem;
+}
+
+.modal-header {
+  background: #f8f9fa;
+  border-radius: 0.5rem 0.5rem 0 0;
+}
+
+@media (max-width: 768px) {
+  .stat-card {
+    margin-bottom: 1rem;
+  }
+
+  .card-header .d-flex {
+    flex-direction: column;
+    gap: 0.5rem !important;
+  }
+
+  .nav-tabs .nav-link {
+    padding: 0.5rem 1rem;
+    font-size: 0.875rem;
+  }
+}

cloudflare_pages/wrangler.toml

@@ -0,0 +1,10 @@
+name = "trades"
+compatibility_date = "2026-03-10"
+
+[vars]
+APP_TITLE = "BTC 预测看板"
+
+[[d1_databases]]
+binding = "DB"
+database_name = "orders"
+database_id = "ae0cd6aa-10da-4459-9768-8e6ffad55a00"

cloudflare_pages/wrangler.toml.example

@@ -0,0 +1,11 @@
+name = "trades"
+compatibility_date = "2026-03-10"
+
+[vars]
+APP_TITLE = "BTC 预测看板"
+
+[[d1_databases]]
+binding = "DB"
+database_name = "orders"
+database_id = "ae0cd6aa-10da-4459-9768-8e6ffad55a00"
+

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,6 +1,7 @@
 """从Binance获取加密货币OHLCV数据"""
 
 import os
+import time
 from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Optional, Literal
@@ -11,9 +12,14 @@ import pytz
 from dotenv import load_dotenv
 
 from data.database import DatabaseManager
+from utils.network_proxy import apply_global_proxy_from_env, get_proxy_url
+from utils.project_logger import get_project_logger
+from utils.send_msg import lark_notifier
 
 # 加载环境变量
 load_dotenv()
+apply_global_proxy_from_env()
+logger = get_project_logger("data.fetcher")
 
 
 class BinanceDataFetcher:
@@ -30,17 +36,16 @@ class BinanceDataFetcher:
         Args:
             cache_dir: 缓存/数据库目录
         """
-        # 配置代理
+        # 统一读取代理配置，兼容全局代理环境变量。
         proxies = None
-        use_proxy = os.getenv('USE_PROXY', 'false').lower() == 'true'
+        proxy_url = get_proxy_url()
 
-        if use_proxy:
+        if proxy_url:
-            socks5_proxy = os.getenv('SOCKS5_PROXY', 'socks5://127.0.0.1:7890')
             proxies = {
-                'http': socks5_proxy,
+                'http': proxy_url,
-                'https': socks5_proxy
+                'https': proxy_url
             }
-            print(f"✓ 使用SOCKS5代理: {socks5_proxy}")
+            logger.info("使用代理: %s", proxy_url)
 
         try:
             # 初始化Binance客户端（带代理）
@@ -51,13 +56,10 @@ class BinanceDataFetcher:
 
             # 测试连接
             self.client.ping()
-            print("✓ 已成功连接到Binance API")
+            logger.info("已成功连接到Binance API")
         except Exception as e:
-            print(f"⚠ Binance API连接失败: {e}")
+            logger.warning("Binance API连接失败: %s", e)
-            print("提示：")
+            logger.warning("提示: 检查代理设置、代理服务状态，或使用缓存数据")
-            print("  1. 检查代理设置 (.env文件)")
-            print("  2. 确保代理服务正在运行")
-            print("  3. 或者使用已缓存的数据")
             self.client = None
 
         # 初始化数据库管理器
@@ -69,7 +71,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 +112,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("从数据库加载数据: %s %s (%s rows)", symbol, timeframe, len(df))
                     return df
 
         # 2. 从API获取数据
         if self.client is None:
             if not df.empty:
-                 print(f"⚠ API不可用，返回数据库现有数据 ({len(df)} rows)")
+                 logger.warning("API不可用，返回数据库现有数据 (%s rows)", len(df))
                  return df
             
             raise ConnectionError(
@@ -126,23 +129,35 @@ class BinanceDataFetcher:
         pair = f"{symbol}USDT"
         interval = self.INTERVALS[timeframe]
 
-        print(f"获取 {symbol} {timeframe} 数据，时间范围：{start_dt.date()} 至 {end_dt.date()}...")
+        logger.info("获取 %s %s 数据，时间范围: %s 至 %s", symbol, timeframe, start_dt.date(), end_dt.date())
 
-        # 获取K线数据
+        klines = []
-        try:
+        for attempt in range(max_retries):
-            klines = self.client.get_historical_klines(
+            try:
-                symbol=pair,
+                klines = self.client.get_historical_klines(
-                interval=interval,
+                    symbol=pair,
-                start_str=str(int(start_dt.timestamp() * 1000)),
+                    interval=interval,
-                end_str=str(int(end_dt.timestamp() * 1000))
+                    start_str=str(int(start_dt.timestamp() * 1000)),
-            )
+                    end_str=str(int(end_dt.timestamp() * 1000))
-        except Exception as e:
+                )
-            if not df.empty:
+                # 如果成功获取，跳出循环
-                print(f"⚠ API请求失败 ({e})，返回数据库现有数据")
+                break
-                return df
+            except Exception as e:
-            raise e
+                # 如果是最后一次尝试，且依然失败
+                if attempt == max_retries - 1:
+                    logger.error("第 %s 次尝试失败，已达到最大重试次数。错误: %s", attempt + 1, e)
+                    if not df.empty:
+                        logger.warning("API请求最终失败，返回数据库现有数据")
+                        return df
+                    raise e  # 如果没有缓存数据，抛出异常
+                
+                # 计算等待时间 (指数退避: 2s, 4s, 8s...)
+                wait_time = 2 * (2 ** attempt) 
+                logger.warning("请求失败(尝试 %s/%s)，%s 秒后重试，错误: %s", 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,7 +179,7 @@ class BinanceDataFetcher:
         # 按时间排序
         new_df = new_df.sort_values('timestamp').reset_index(drop=True)
 
-        print(f"已获取 {len(new_df)} 行数据")
+        logger.info("已获取 %s 行数据", len(new_df))
 
         # 保存到数据库
         self.db_manager.save_ohlcv(new_df, symbol, timeframe)

data/trading_db.py

@@ -53,11 +53,27 @@ 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):
         """记录交易信息"""
@@ -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
+                        status, strategy, error_message, raw_data, order_id
-                    ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+                    ) 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,488 @@
+"""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年3月 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年3月数据...")
+    fetcher = BinanceDataFetcher()
+    df = fetcher.fetch_ohlcv('BTC', '1h', start_date='2026-01-01', end_date='2026-04-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)}
+
+    # 找出3月（北京时间）的所有预测目标K线索引
+    # df 中每一行都是一根已收盘的K线，timestamp 是 UTC
+    jan_targets = []
+    for target_idx in range(1, len(df)):
+        # 同样在筛选样本时也将时间向后移动一个小时
+        target_time_utc = df.iloc[target_idx]['timestamp'] + timedelta(hours=1)
+        target_time_bj = utc_to_beijing(target_time_utc)
+        if target_time_bj.year == 2026 and target_time_bj.month == 3:
+            jan_targets.append(target_idx)
+
+    print(f"3月（北京时间）目标K线数: {len(jan_targets)}")
+
+    results = []
+
+    for target_idx in jan_targets:
+        # 原始timestamp表示k线的开始时间，因此预测的结果实际对应下一小时的k线。
+        # 所以展示时需要将时间向后移动1小时。
+        target_time_utc = df.iloc[target_idx]['timestamp'] + timedelta(hours=1)
+        target_time_bj = utc_to_beijing(target_time_utc)
+
+        # 现在我们将预测目标设为下一根k线（即target_idx+1），
+        # 输入使用它的前一根（等于原先的target_idx）。
+        predicted_idx = target_idx + 1
+        if predicted_idx >= len(df):
+            # 已经到达数据末尾，无法构建完整样本，跳过
+            continue
+        input_idx = target_idx
+        # 开盘/收盘对应预测k线
+        open_price = df.iloc[predicted_idx]['open']
+        close_price = df.iloc[predicted_idx]['close']
+        # 实际涨跌以预测k线的开盘为基准
+        actual_label = 1 if close_price > open_price 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,
+            'open': open_price,
+            'close': close_price,
+            '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年3月，北京时间)")
+    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['open']:>10.2f} {row['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,387 @@
+"""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
+from utils.network_proxy import apply_global_proxy_from_env
+
+apply_global_proxy_from_env()
+
+# ==================== 时区与美股判断 ====================
+
+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,18 +209,18 @@ class InferenceManager:
             print(f"DLinear 模型文件不存在: {config['model_path']}")
             return
 
-        checkpoint = torch.load(config['model_path'], map_location=self.device, weights_only=False)
+        try:
-        model = DLinear(checkpoint['config'])
+            self.models['dlinear'] = ort.InferenceSession(config['model_path'])
-        model.load_state_dict(checkpoint['model_state_dict'])
+            
-        self.models['dlinear'] = model.to(self.device).eval()
+            with open(config['preprocessor_path'], 'rb') as f:
-        
+                try:
-        with open(config['preprocessor_path'], 'rb') as f:
+                    self.preprocessors['dlinear'] = pickle.load(f)
-            try:
+                except AttributeError:
-                self.preprocessors['dlinear'] = pickle.load(f)
+                    f.seek(0)
-            except AttributeError:
+                    self.preprocessors['dlinear'] = CustomUnpickler(f).load()
-                f.seek(0)
+            print("  ✓ DLinear (ONNX) 已加载")
-                self.preprocessors['dlinear'] = CustomUnpickler(f).load()
+        except Exception as e:
-        print("  ✓ DLinear 已加载")
+            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)
+        try:
-        model = NBeats(checkpoint['config'])
+            self.models['nbeats'] = ort.InferenceSession(config['model_path'])
-        self.models['nbeats'] = model.to(self.device).eval()
+            with open(config['preprocessor_path'], 'rb') as f:
-        with open(config['preprocessor_path'], 'rb') as f:
+                 try:
-             try:
+                    self.preprocessors['nbeats'] = pickle.load(f)
-                self.preprocessors['nbeats'] = pickle.load(f)
+                 except AttributeError:
-             except AttributeError:
+                    f.seek(0)
-                f.seek(0)
+                    self.preprocessors['nbeats'] = CustomUnpickler(f).load()
-                self.preprocessors['nbeats'] = CustomUnpickler(f).load()
+            print("  ✓ N-BEATS (ONNX) 已加载")
-        print("  ✓ N-BEATS 已加载")
+        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(
+            self.models['ttm'] = ort.InferenceSession(config['model_path'])
-                config['model_name'],
+            print("  ✓ TTM-R2 (ONNX) 已加载")
-                trust_remote_code=True
-            )
-            self.models['ttm'] = model.to(self.device).eval()
-            print("  ✓ TTM-R2 已加载")
         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():
+        try:
-            logits = model(X_last).cpu().numpy()
+            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
+            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():
+        try:
-            output = model(X_last).cpu().numpy()
+            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]
@@ -327,7 +323,10 @@ class InferenceManager:
             # N-BEATS 回归模型没有直接概率
             probability = 0.6 if prediction == 1 else 0.4 # 占位
             
-        return prediction, probability
+            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,20 +352,30 @@ 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():
+        try:
-            output = model(context_tensor)
+            input_name = session.get_inputs()[0].name
-            pred_price = output.prediction_outputs.cpu().numpy()[0, 0, 0]
+            output_name = session.get_outputs()[0].name
             
-        prediction = 1 if pred_price > base_price else 0
+            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
-        price_pct = (pred_price - base_price) / base_price
+            # We want the first prediction step [0, 0, 0]
-        probability = 1 / (1 + np.exp(-price_pct * 100))
+            pred_price = output[0, 0, 0]
-        
+            
-        return prediction, float(probability)
+            prediction = 1 if pred_price > base_price else 0
+            
+            # 简单的概率模拟
+            price_pct = (pred_price - base_price) / base_price
+            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]:
         """

mysql_d1_backup_scheduler.py

@@ -0,0 +1,371 @@
+#!/usr/bin/env python3
+"""
+MySQL 数据定时备份到 Cloudflare D1
+- 定时：每小时 XX:30
+- 模式：按游标列增量同步（timestamp/created_at/updated_at/id）
+"""
+
+import json
+import os
+import threading
+import time
+from datetime import datetime
+from decimal import Decimal
+from typing import Any
+
+import httpx
+import schedule
+from dotenv import load_dotenv
+
+from api.models import get_db_connection
+from utils.project_logger import get_project_logger
+
+load_dotenv()
+logger = get_project_logger("mysql_d1_backup")
+
+
+class MySQLD1BackupService:
+    """MySQL -> D1 备份服务。"""
+
+    def __init__(self):
+        self.enabled = os.getenv("D1_BACKUP_ENABLED", "false").lower() == "true"
+        self.api_token = os.getenv("CF_D1_API_TOKEN", "")
+        self.account_id = os.getenv("CF_D1_ACCOUNT_ID", "")
+        self.database_id = os.getenv("CF_D1_DATABASE_ID", "")
+        self.batch_size = int(os.getenv("D1_BACKUP_BATCH_SIZE", "500"))
+        self.max_sql_params = int(os.getenv("D1_MAX_SQL_PARAMS", "90"))
+        self.tables_config = os.getenv("D1_BACKUP_TABLES", "").strip()
+        self._job_lock = threading.Lock()
+
+    def is_config_ready(self) -> bool:
+        """检查 D1 配置是否完整。"""
+        if not self.enabled:
+            return False
+        return bool(self.api_token and self.account_id and self.database_id)
+
+    def _d1_endpoint(self) -> str:
+        return (
+            f"https://api.cloudflare.com/client/v4/accounts/"
+            f"{self.account_id}/d1/database/{self.database_id}/query"
+        )
+
+    def _query_d1(self, sql: str, params: list[Any] | None = None) -> dict:
+        """执行 D1 SQL。"""
+        headers = {
+            "Authorization": f"Bearer {self.api_token}",
+            "Content-Type": "application/json",
+        }
+        statement: dict[str, Any] = {"sql": sql}
+        if params is not None:
+            statement["params"] = params
+
+        with httpx.Client(timeout=30) as client:
+            # Cloudflare D1 REST API 对 /query 更稳定的格式是语句数组
+            resp = client.post(self._d1_endpoint(), headers=headers, json=[statement])
+            if resp.status_code >= 400:
+                # 兼容部分环境下使用对象格式的情况，失败时回退重试一次
+                fallback_resp = client.post(self._d1_endpoint(), headers=headers, json=statement)
+                if fallback_resp.status_code >= 400:
+                    sql_preview = " ".join(sql.strip().split())[:180]
+                    raise RuntimeError(
+                        f"D1 HTTP {fallback_resp.status_code}，sql={sql_preview}，响应={fallback_resp.text[:500]}"
+                    )
+                data = fallback_resp.json()
+            else:
+                data = resp.json()
+
+        if not data.get("success", False):
+            raise RuntimeError(f"D1 SQL 执行失败: {data}")
+        return data
+
+    def _d1_results(self, response: dict) -> list[dict]:
+        """解析 D1 响应结果。"""
+        result = response.get("result", [])
+        if not result:
+            return []
+        first = result[0] if isinstance(result, list) else result
+        return first.get("results", []) if isinstance(first, dict) else []
+
+    def _normalize_value(self, value: Any) -> Any:
+        """把 MySQL 字段值转换为可写入 D1 的类型。"""
+        if value is None:
+            return None
+        if isinstance(value, (datetime,)):
+            return value.strftime("%Y-%m-%d %H:%M:%S")
+        if isinstance(value, Decimal):
+            return float(value)
+        if isinstance(value, bytes):
+            return value.hex()
+        if isinstance(value, (dict, list)):
+            return json.dumps(value, ensure_ascii=False)
+        return value
+
+    def _sqlite_type_from_mysql(self, mysql_type: str) -> str:
+        t = mysql_type.lower()
+        if any(x in t for x in ["int", "bit", "bool"]):
+            return "INTEGER"
+        if any(x in t for x in ["double", "float", "decimal", "numeric", "real"]):
+            return "REAL"
+        if any(x in t for x in ["blob", "binary", "varbinary"]):
+            return "BLOB"
+        return "TEXT"
+
+    def _get_mysql_tables(self) -> list[str]:
+        """获取待备份表。"""
+        if self.tables_config:
+            return [x.strip() for x in self.tables_config.split(",") if x.strip()]
+
+        with get_db_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute("SHOW TABLES")
+            rows = cursor.fetchall()
+
+        tables: list[str] = []
+        for row in rows:
+            # DictCursor 下 SHOW TABLES 返回单键字典
+            table_name = list(row.values())[0]
+            tables.append(str(table_name))
+        return tables
+
+    def _get_mysql_columns(self, table_name: str) -> list[dict]:
+        with get_db_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute(f"SHOW COLUMNS FROM `{table_name}`")
+            return list(cursor.fetchall())
+
+    def _guess_cursor_column(self, columns: list[dict]) -> str | None:
+        """选择增量同步游标列。"""
+        column_names = {c["Field"] for c in columns}
+        for candidate in ["timestamp", "created_at", "updated_at", "id"]:
+            if candidate in column_names:
+                return candidate
+        return None
+
+    def _ensure_d1_meta_tables(self):
+        """创建 D1 元数据表。"""
+        self._query_d1(
+            """
+            CREATE TABLE IF NOT EXISTS backup_sync_state (
+                table_name TEXT PRIMARY KEY,
+                cursor_col TEXT,
+                last_cursor TEXT,
+                updated_at TEXT
+            )
+            """
+        )
+        self._query_d1(
+            """
+            CREATE TABLE IF NOT EXISTS backup_run_log (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                run_at TEXT,
+                table_name TEXT,
+                synced_rows INTEGER,
+                status TEXT,
+                message TEXT
+            )
+            """
+        )
+
+    def _ensure_d1_table(self, table_name: str, columns: list[dict]):
+        """创建 D1 备份表。"""
+        table_sql_name = f"backup_{table_name}"
+        col_defs: list[str] = []
+        primary_keys: list[str] = []
+
+        for col in columns:
+            field = col["Field"]
+            col_type = self._sqlite_type_from_mysql(col["Type"])
+            col_defs.append(f'"{field}" {col_type}')
+            if col.get("Key") == "PRI":
+                primary_keys.append(field)
+
+        # SQLite/D1 对复合主键需要表级 PRIMARY KEY 约束，不能多列各自 PRIMARY KEY
+        if primary_keys:
+            pk_cols = ", ".join([f'"{k}"' for k in primary_keys])
+            col_defs.append(f"PRIMARY KEY ({pk_cols})")
+
+        create_sql = f"CREATE TABLE IF NOT EXISTS \"{table_sql_name}\" ({', '.join(col_defs)})"
+        self._query_d1(create_sql)
+
+    def _get_last_cursor(self, table_name: str) -> str | None:
+        """读取 D1 已同步游标。"""
+        resp = self._query_d1(
+            "SELECT last_cursor FROM backup_sync_state WHERE table_name = ?",
+            [table_name],
+        )
+        rows = self._d1_results(resp)
+        if not rows:
+            return None
+        return rows[0].get("last_cursor")
+
+    def _update_last_cursor(self, table_name: str, cursor_col: str, cursor_value: Any):
+        """更新 D1 已同步游标。"""
+        now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        self._query_d1(
+            """
+            INSERT INTO backup_sync_state(table_name, cursor_col, last_cursor, updated_at)
+            VALUES (?, ?, ?, ?)
+            ON CONFLICT(table_name) DO UPDATE SET
+              cursor_col = excluded.cursor_col,
+              last_cursor = excluded.last_cursor,
+              updated_at = excluded.updated_at
+            """,
+            [table_name, cursor_col, str(cursor_value), now_str],
+        )
+
+    def _log_run(self, table_name: str, synced_rows: int, status: str, message: str):
+        """记录备份执行日志。"""
+        run_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        self._query_d1(
+            "INSERT INTO backup_run_log(run_at, table_name, synced_rows, status, message) VALUES (?, ?, ?, ?, ?)",
+            [run_at, table_name, synced_rows, status, message[:1000]],
+        )
+
+    def _fetch_mysql_rows(
+        self,
+        table_name: str,
+        cursor_col: str | None,
+        last_cursor: str | None,
+    ) -> list[dict]:
+        """按游标从 MySQL 抽取增量数据。"""
+        with get_db_connection() as conn:
+            cursor = conn.cursor()
+            if cursor_col and last_cursor:
+                sql = (
+                    f"SELECT * FROM `{table_name}` WHERE `{cursor_col}` > %s "
+                    f"ORDER BY `{cursor_col}` ASC LIMIT %s"
+                )
+                cursor.execute(sql, (last_cursor, self.batch_size))
+            elif cursor_col:
+                sql = f"SELECT * FROM `{table_name}` ORDER BY `{cursor_col}` ASC LIMIT %s"
+                cursor.execute(sql, (self.batch_size,))
+            else:
+                sql = f"SELECT * FROM `{table_name}` LIMIT %s"
+                cursor.execute(sql, (self.batch_size,))
+            return list(cursor.fetchall())
+
+    def _upsert_rows_to_d1(self, table_name: str, columns: list[dict], rows: list[dict]):
+        """把数据批量 UPSERT 到 D1。"""
+        if not rows:
+            return
+
+        table_sql_name = f"backup_{table_name}"
+        col_names = [c["Field"] for c in columns]
+        col_sql = ", ".join([f'"{c}"' for c in col_names])
+
+        # 避免 D1 的 SQL 变量数量限制触发 400
+        # 通过环境变量 D1_MAX_SQL_PARAMS 控制，默认保守值 90
+        chunk_size = max(1, self.max_sql_params // max(1, len(col_names)))
+
+        for i in range(0, len(rows), chunk_size):
+            chunk_rows = rows[i:i + chunk_size]
+            values_clause: list[str] = []
+            params: list[Any] = []
+            for row in chunk_rows:
+                placeholders = ", ".join(["?" for _ in col_names])
+                values_clause.append(f"({placeholders})")
+                for col in col_names:
+                    params.append(self._normalize_value(row.get(col)))
+
+            sql = (
+                f"INSERT OR REPLACE INTO \"{table_sql_name}\" ({col_sql}) "
+                f"VALUES {', '.join(values_clause)}"
+            )
+            self._query_d1(sql, params)
+
+    def backup_table_once(self, table_name: str) -> int:
+        """单表执行一次增量备份，返回同步行数。"""
+        columns = self._get_mysql_columns(table_name)
+        if not columns:
+            return 0
+
+        cursor_col = self._guess_cursor_column(columns)
+        self._ensure_d1_table(table_name, columns)
+
+        total_synced = 0
+        last_cursor = self._get_last_cursor(table_name)
+
+        while True:
+            rows = self._fetch_mysql_rows(table_name, cursor_col, last_cursor)
+            if not rows:
+                break
+
+            self._upsert_rows_to_d1(table_name, columns, rows)
+            total_synced += len(rows)
+
+            if cursor_col:
+                last_cursor = self._normalize_value(rows[-1].get(cursor_col))
+                self._update_last_cursor(table_name, cursor_col, last_cursor)
+            else:
+                # 无游标列时仅做一批，避免重复全量
+                break
+
+            if len(rows) < self.batch_size:
+                break
+
+        return total_synced
+
+    def run_backup_once(self):
+        """执行一次全表备份任务。"""
+        if not self.is_config_ready():
+            logger.warning("D1 备份未启用或配置不完整，跳过本次任务")
+            return
+
+        if not self._job_lock.acquire(blocking=False):
+            logger.warning("上一次 D1 备份尚未完成，跳过本次任务")
+            return
+
+        try:
+            self._ensure_d1_meta_tables()
+            tables = self._get_mysql_tables()
+            logger.info("D1 备份任务开始，待同步表: %s", ", ".join(tables))
+
+            for table_name in tables:
+                try:
+                    synced = self.backup_table_once(table_name)
+                    logger.info("表 %s 同步完成，本次同步 %d 行", table_name, synced)
+                    self._log_run(table_name, synced, "success", "ok")
+                except Exception as table_error:
+                    logger.exception("表 %s 同步失败", table_name)
+                    self._log_run(table_name, 0, "failed", str(table_error))
+        except Exception:
+            logger.exception("D1 备份任务执行失败")
+        finally:
+            self._job_lock.release()
+
+
+def schedule_mysql_d1_backup():
+    """启动 D1 备份定时任务（每小时 :30）。"""
+    service = MySQLD1BackupService()
+    if not service.is_config_ready():
+        logger.warning("D1 备份调度未启动：请检查 D1_BACKUP_ENABLED 与 Cloudflare 参数")
+        return
+
+    # 启动后先执行一次，避免首次备份必须等待到下一个 :30
+    service.run_backup_once()
+
+    # 使用独立调度器，避免与项目内其他定时任务共享全局调度状态
+    local_scheduler = schedule.Scheduler()
+    local_scheduler.every().hour.at(":30").do(service.run_backup_once)
+    logger.info("MySQL -> D1 定时备份已启动（每小时 :30 执行）")
+
+    while True:
+        local_scheduler.run_pending()
+        time.sleep(1)
+
+
+def run_mysql_d1_backup_scheduler_background():
+    """后台线程启动 D1 备份调度器。"""
+    service = MySQLD1BackupService()
+    if not service.is_config_ready():
+        logger.warning("D1 备份配置不完整，后台线程不启动")
+        return None
+
+    scheduler_thread = threading.Thread(target=schedule_mysql_d1_backup, daemon=True)
+    scheduler_thread.start()
+    logger.info("D1 备份调度器已在后台启动")
+    return scheduler_thread
+
+
+if __name__ == "__main__":
+    schedule_mysql_d1_backup()

scripts/export_onnx.py

@@ -1,13 +1,9 @@
-"""导出 DLinear 模型为 ONNX 格式。
+"""导出 DLinear, N-BEATS 和 TTM-R2 模型为 ONNX 格式。"""
-
-用法:
-    python scripts/export_onnx.py
-
-输出:
-    models/checkpoints/dlinear_classification_2026/model.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 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}")
+        
+        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()
-    checkpoint_path = project_root / "models/checkpoints/dlinear_classification_2026/best_model.pt"
+    export_nbeats()
-    output_path = checkpoint_path.parent / "model.onnx"
+    #export_ttm_r2()
-
-    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}")
-
 
 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
 """
@@ -14,6 +15,28 @@ import signal
 import os
 import platform
 from datetime import datetime
+from dotenv import load_dotenv
+
+from utils.network_proxy import apply_global_proxy_from_env
+from utils.project_logger import get_project_logger
+
+try:
+    from daily_report_sender import run_scheduler_background
+except ImportError:
+    # 如果模块导入失败，定义一个空函数避免启动失败
+    def run_scheduler_background():
+        return None
+
+try:
+    from mysql_d1_backup_scheduler import run_mysql_d1_backup_scheduler_background
+except ImportError:
+    # 如果模块导入失败，定义一个空函数避免启动失败
+    def run_mysql_d1_backup_scheduler_background():
+        return None
+
+load_dotenv()
+apply_global_proxy_from_env()
+logger = get_project_logger("start_system")
 
 # 检测操作系统
 IS_WINDOWS = platform.system() == 'Windows'
@@ -42,6 +65,7 @@ else:
 def log(message, color=RESET):
     timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
     print(f"{color}[{timestamp}] {message}{RESET}")
+    logger.info(message)
 
 def start_process(command, name):
     """启动子进程（跨平台兼容）"""
@@ -91,6 +115,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 +150,29 @@ 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
+
+    # 4. 启动 MySQL -> Cloudflare D1 定时备份（每小时 :30）
+    try:
+        d1_backup_thread = run_mysql_d1_backup_scheduler_background()
+        if d1_backup_thread:
+            log("MySQL -> D1 定时备份程序已启动（每小时 :30）", GREEN)
+        else:
+            log("D1 定时备份未启动（可能未启用或配置不完整）", YELLOW)
+    except Exception as e:
+        log(f"⚠️  启动 D1 定时备份程序异常: {e}", YELLOW)
+        d1_backup_thread = None
+    
     log("系统启动完成！", GREEN)
     log("前端访问地址: http://localhost:8000", GREEN)
     log("按 Ctrl+C 停止所有服务...", YELLOW)
@@ -117,10 +181,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

@@ -34,10 +34,13 @@ 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.network_proxy import apply_global_proxy_from_env, get_proxy_url
+from utils.send_msg import lark_notifier
 
 load_dotenv()
+apply_global_proxy_from_env()
 
-
+    
 def setup_logging(log_dir: str = "logs", log_level: int = logging.INFO) -> logging.Logger:
     """
     配置日志系统
@@ -118,7 +121,7 @@ class PolymarketBTCTrader:
         self.dry_run = dry_run
         
         # HTTP客户端
-        proxy = os.getenv("SOCKS5_PROXY") if os.getenv("USE_PROXY", "false").lower() == "true" else None
+        proxy = get_proxy_url()
         self.http = httpx.Client(
             timeout=30.0,
             proxy=proxy,
@@ -150,18 +153,80 @@ class PolymarketBTCTrader:
         """初始化CLOB交易客户端"""
         try:
             from py_clob_client.client import ClobClient
+            from py_clob_client.clob_types import ApiCreds, OpenOrderParams
             
-            self.clob_client = ClobClient(
+            funder = os.getenv("FUNDER_ADDRESS")
-                host=self.CLOB_API,
+            nonce_str = os.getenv("POLYMARKET_NONCE")  # 可选，取决于你的钱包设置
-                key=self.private_key,
+            nonce = int(nonce_str) if nonce_str else None
-                chain_id=137,  # Polygon
+            if self.private_key and funder and nonce is not None:
-            )
+                try: 
-            self.clob_client.set_api_creds(self.clob_client.create_or_derive_api_creds())
+                    self.clob_client = ClobClient(
-            logger.info("✓ CLOB客户端初始化成功")
+                        host=self.CLOB_API,
+                        key=self.private_key,
+                        chain_id=137,  # Polygon
+                        signature_type=2,
+                        funder=funder
+                    )
+                    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 +256,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)
@@ -214,6 +280,7 @@ class PolymarketBTCTrader:
         
         month_name = self.MONTH_NAMES[target_time.month]
         day = target_time.day
+        year = target_time.year
         
         # 转换为12小时制
         hour_24 = target_time.hour
@@ -230,62 +297,67 @@ class PolymarketBTCTrader:
             hour_12 = hour_24 - 12
             am_pm = 'pm'
         
-        slug = f"bitcoin-up-or-down-{month_name}-{day}-{hour_12}{am_pm}-et"
+        slug = f"bitcoin-up-or-down-{month_name}-{day}-{year}-{hour_12}{am_pm}-et"
         return slug
     
     def get_market_by_slug(self, slug: str) -> Optional[Dict]:
         """通过slug获取市场信息"""
-        try:
+        for attempt in range(3):
-            response = self.http.get(
+            try:
-                f"{self.GAMMA_API}/events",
+                response = self.http.get(
-                params={'slug': slug}
+                    f"{self.GAMMA_API}/events",
-            )
+                    params={'slug': slug}
-            response.raise_for_status()
+                )
-            events = response.json()
+                response.raise_for_status()
+                events = response.json()
             
-            if not events:
+                if not events:
-                return None
+                    return None
             
-            event = events[0]
+                event = events[0]
-            markets = event.get('markets', [])
+                markets = event.get('markets', [])
             
-            if not markets:
+                if not markets:
-                return None
+                    return None
             
-            market = markets[0]
+                market = markets[0]
             
             # outcomePrices可能是JSON字符串或列表
-            outcome_prices_raw = market.get('outcomePrices', ['0', '0'])
+                outcome_prices_raw = market.get('outcomePrices', ['0', '0'])
-            if isinstance(outcome_prices_raw, str):
+                if isinstance(outcome_prices_raw, str):
-                import json as _json
+                    import json as _json
-                try:
+                    try:
-                    outcome_prices_raw = _json.loads(outcome_prices_raw)
+                        outcome_prices_raw = _json.loads(outcome_prices_raw)
-                except:
+                    except:
-                    outcome_prices_raw = ['0', '0']
+                        outcome_prices_raw = ['0', '0']
             
             # clobTokenIds也可能是JSON字符串
-            clob_tokens_raw = market.get('clobTokenIds', [])
+                clob_tokens_raw = market.get('clobTokenIds', [])
-            if isinstance(clob_tokens_raw, str):
+                if isinstance(clob_tokens_raw, str):
-                import json as _json
+                    import json as _json
-                try:
+                    try:
-                    clob_tokens_raw = _json.loads(clob_tokens_raw)
+                        clob_tokens_raw = _json.loads(clob_tokens_raw)
-                except:
+                    except:
-                    clob_tokens_raw = []
+                        clob_tokens_raw = []
             
-            return {
+                return {
-                'title': event.get('title', ''),
+                    'title': event.get('title', ''),
-                'slug': slug,
+                    'slug': slug,
-                'question': market.get('question', ''),
+                    'question': market.get('question', ''),
-                'condition_id': market.get('conditionId', ''),
+                    'condition_id': market.get('conditionId', ''),
-                'outcomes': market.get('outcomes', []),
+                    'outcomes': market.get('outcomes', []),
-                'outcome_prices': [float(p) for p in outcome_prices_raw],
+                    'outcome_prices': [float(p) for p in outcome_prices_raw],
-                'tokens': clob_tokens_raw,
+                    'tokens': clob_tokens_raw,
-                'end_date': event.get('endDate', ''),
+                    'end_date': event.get('endDate', ''),
-                'closed': market.get('closed', False),
+                    'closed': market.get('closed', False),
-            }
+                }
-        except Exception as e:
+            except Exception as e:
-            logger.error(f"获取市场失败: {e}")
+                logger.warning(f"尝试 {attempt+1}/3 获取市场失败: {e}")
-            return None
+                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]:
         """获取BTC市场"""
@@ -379,6 +451,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 +541,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:
+        # if price > self.max_price:
-            result['error'] = f"价格${price:.2f}超过限制${self.max_price:.2f}"
+        #     result['error'] = f"价格${price:.2f}超过限制${self.max_price:.2f}"
-            logger.error(f"❌ {result['error']}")
+        #     logger.error(f"❌ {result['error']}")
-            return result
+        #     return result
         
         # 计算数量
         size = self.max_bet / price
@@ -423,6 +578,7 @@ class PolymarketBTCTrader:
             logger.info("🔸 [模拟模式] 未实际下单")
             result['success'] = True
             result['order'] = 'DRY_RUN'
+            result['status'] = 'simulated'
             return result
         
         # 实际下单
@@ -433,21 +589,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(
+            # 2. 检查现有挂单
-                token_id=token_id,
+            order_id = None
-                price=price,
+            try:
-                size=size,
+                open_order_params = OpenOrderParams(asset_id=token_id)
-                side=BUY,
+                existing_orders = self.clob_client.get_orders(open_order_params)
-            )
+                if existing_orders:
-            
+                    existing_order = existing_orders[0]
-            response = self.clob_client.post_order(order)
+                    order_id = existing_order.get('orderID')
-            logger.info(f"✓ 订单已提交: {response}")
+                    logger.warning(f"⚠️ 检测到已有挂单: {order_id}，跳过下单")
-            result['success'] = True
+                    result['order_id'] = order_id
-            result['order'] = response
+                    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,
+                )
+                
+                # 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 +857,108 @@ class PolymarketBTCTrader:
             return result
             
         # 检查是否已交易过该市场
-        check_status = 'test' if self.dry_run else 'success'
+        existing_trade = self.db.get_existing_trade(market['slug'])
-        if self.db.check_existing_trade(market['slug'], status=check_status):
+        
-            result['error'] = '该市场已存在成功交易记录'
+        if existing_trade:
-            logger.warning(f"⚠️ {result['error']}, 跳过交易")
+            status = existing_trade.get('status')
-            return result
+            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()
+        # start_time = datetime.now()
-        end_time = start_time + timedelta(minutes=timeout_minutes)
+        # 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()
+            # market = self.refresh_prices()
-            if not market:
+            # if not market:
-                logger.warning("无法获取市场价格，重试中...")
+            #     logger.warning("无法获取市场价格，重试中...")
-                time.sleep(10)
+            #     time.sleep(10)
-                continue
+            #     continue
             
             # 检查价格
-            price_idx = 0 if prediction == 1 else 1
+            # price_idx = 0 if prediction == 1 else 1
-            current_price = market['outcome_prices'][price_idx]
+            # 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)
+        trade_result = self.execute_trade(prediction, market)
-                result.update(trade_result)
+        result.update(trade_result)
-                
+        
-                # 记录数据库
+        # 如果下单成功且有order_id，进入状态监控
-                self._log_to_db(result, market)
+        if result.get('success') and result.get('order_id'):
-                
+            final_status = self.monitor_order_status(result['order_id'])
-                logger.info("=" * 60)
+            result['status'] = final_status
-                return result
+        
-            else:
+        # 记录数据库
-                # 价格过高，等待
+        self._log_to_db(result, market)
-                wait_time = 30  # 等待30秒
+        
-                elapsed = int((datetime.now() - start_time).total_seconds())
+        logger.info("=" * 60)
-                logger.info(f"⏳ 价格过高，继续等待... (已耗时: {elapsed}s)")
+        return result
-                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['error'] = f'等待价格超时 ({timeout_minutes}分钟)'
-        result['price'] = market['outcome_prices'][0] if prediction == 1 else market['outcome_prices'][1]
+        # result['price'] = market['outcome_prices'][0] if prediction == 1 else market['outcome_prices'][1]
-        result['side'] = 'UP' if prediction == 1 else 'DOWN'
+        # result['side'] = 'UP' if prediction == 1 else 'DOWN'
-        logger.error(f"❌ {result['error']}")
+        # logger.error(f"❌ {result['error']}")
         
-        self._log_to_db(result, market)
+        # self._log_to_db(result, market)
-        logger.info("=" * 60)
+        # logger.info("=" * 60)
         
-        return result
+        # return result
     
     def _show_initial_prediction(self):
         """
@@ -790,9 +1033,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/network_proxy.py

@@ -0,0 +1,53 @@
+"""统一网络代理配置工具。"""
+
+import os
+from typing import Optional
+
+
+def get_proxy_url() -> Optional[str]:
+    """从环境变量读取代理地址，统一兼容不同变量名。"""
+    use_proxy = os.getenv("USE_PROXY", "false").lower() == "true"
+    if not use_proxy:
+        return None
+
+    return (
+        os.getenv("SOCKS5_PROXY")
+        or os.getenv("ALL_PROXY")
+        or os.getenv("HTTPS_PROXY")
+        or os.getenv("HTTP_PROXY")
+    )
+
+
+def get_send_msg_proxy_url() -> Optional[str]:
+    """读取 send_msg 独立代理，未配置时回退到全局代理。"""
+    send_msg_proxy = os.getenv("SEND_MSG_PROXY")
+    if send_msg_proxy:
+        return send_msg_proxy
+
+    send_msg_proxy_port = os.getenv("SEND_MSG_PROXY_PORT")
+    if send_msg_proxy_port:
+        return f"socks5://127.0.0.1:{send_msg_proxy_port}"
+
+    return get_proxy_url()
+
+
+def apply_global_proxy_from_env() -> Optional[str]:
+    """将代理写入全局环境变量，确保 requests/httpx/curl 等默认走同一代理。"""
+    proxy_url = get_proxy_url()
+    if not proxy_url:
+        return 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
+
+    # 保留本地直连，避免影响本机服务调用。
+    if not os.getenv("NO_PROXY"):
+        os.environ["NO_PROXY"] = "127.0.0.1,localhost"
+    if not os.getenv("no_proxy"):
+        os.environ["no_proxy"] = os.environ["NO_PROXY"]
+
+    return proxy_url

utils/project_logger.py

@@ -0,0 +1,40 @@
+"""项目统一日志工具。"""
+
+import logging
+import os
+from logging.handlers import RotatingFileHandler
+
+
+def get_project_logger(name: str) -> logging.Logger:
+    """返回统一配置的日志实例，避免重复添加处理器。"""
+    logger = logging.getLogger(name)
+    if logger.handlers:
+        return logger
+
+    os.makedirs("logs", exist_ok=True)
+
+    logger.setLevel(logging.INFO)
+    logger.propagate = False
+
+    formatter = logging.Formatter(
+        "%(asctime)s | %(name)s | %(levelname)s | %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+
+    file_handler = RotatingFileHandler(
+        "logs/project.log",
+        maxBytes=10 * 1024 * 1024,
+        backupCount=5,
+        encoding="utf-8",
+    )
+    file_handler.setFormatter(formatter)
+    file_handler.setLevel(logging.INFO)
+
+    console_handler = logging.StreamHandler()
+    console_handler.setFormatter(formatter)
+    console_handler.setLevel(logging.INFO)
+
+    logger.addHandler(file_handler)
+    logger.addHandler(console_handler)
+
+    return logger

utils/send_msg.py

@@ -0,0 +1,54 @@
+import requests
+from utils.network_proxy import apply_global_proxy_from_env, get_send_msg_proxy_url
+from utils.project_logger import get_project_logger
+
+
+apply_global_proxy_from_env()
+logger = get_project_logger("send_msg")
+
+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
+        
+        # 使用 Session 可以复用底层 TCP 连接，提升发送效率
+        self.session = requests.Session()
+
+        proxy_url = get_send_msg_proxy_url()
+        if proxy_url:
+            self.session.proxies.update({
+                "http": proxy_url,
+                "https": proxy_url,
+            })
+            logger.info("send_msg 使用独立代理: %s", proxy_url)
+        else:
+            logger.info("send_msg 未启用代理")
+
+    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:
+            # 使用 data=payload 发送 Form Data
+            response = self.session.post(self.url, data=payload, timeout=5)
+            response.raise_for_status()
+            logger.info("告警发送成功: %s", title)
+        except Exception as e:
+            # 告警系统自身的错误只记录日志，绝不抛出异常影响主程序
+            logger.error("发送飞书告警失败: %s", e)
+
+# 实例化一个全局单例供其他模块直接导入使用
+lark_notifier = LarkNotifier()