27 changed files with 1613 additions and 62 deletions

No files matched your search

+2 -1
View File
@@ -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:*)"
]
}
}
+17
View File
@@ -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
@@ -38,6 +42,19 @@ 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
+23 -19
View File
@@ -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):
+4
View File
@@ -0,0 +1,4 @@
{
"account_id": "7ea8a511cad4d467490fe36aa7d6638c",
"project_name": "trades"
}
@@ -0,0 +1,6 @@
{
"account": {
"id": "7ea8a511cad4d467490fe36aa7d6638c",
"name": "Subuhuinew@gmail.com's Account"
}
}
+41
View File
@@ -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`
@@ -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);
}
}
@@ -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);
}
}
@@ -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);
}
}
+15
View File
@@ -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",
});
}
@@ -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);
}
}
@@ -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);
}
}
+200
View File
@@ -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');
+251
View File
@@ -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>
+118
View File
@@ -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;
}
}
+10
View File
@@ -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"
+11
View File
@@ -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"
+20 -20
View File
@@ -12,10 +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:
@@ -32,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:
socks5_proxy = os.getenv('SOCKS5_PROXY', 'socks5://127.0.0.1:7890')
if proxy_url:
proxies = {
'http': socks5_proxy,
'https': socks5_proxy
'http': proxy_url,
'https': proxy_url
}
print(f"✓ 使用SOCKS5代理: {socks5_proxy}")
logger.info("使用代理: %s", proxy_url)
try:
# 初始化Binance客户端(带代理)
@@ -53,13 +56,10 @@ class BinanceDataFetcher:
# 测试连接
self.client.ping()
print("✓ 已成功连接到Binance API")
logger.info("已成功连接到Binance API")
except Exception as e:
print(f"⚠ Binance API连接失败: {e}")
print("提示:")
print(" 1. 检查代理设置 (.env文件)")
print(" 2. 确保代理服务正在运行")
print(" 3. 或者使用已缓存的数据")
logger.warning("Binance API连接失败: %s", e)
logger.warning("提示: 检查代理设置、代理服务状态,或使用缓存数据")
self.client = None
# 初始化数据库管理器
@@ -112,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(
@@ -129,7 +129,7 @@ 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())
klines = []
for attempt in range(max_retries):
@@ -145,15 +145,15 @@ class BinanceDataFetcher:
except Exception as e:
# 如果是最后一次尝试,且依然失败
if attempt == max_retries - 1:
print(f"❌ 第 {attempt + 1} 次尝试失败,已达到最大重试次数。错误: {e}")
logger.error("第 %s 次尝试失败,已达到最大重试次数。错误: %s", attempt + 1, e)
if not df.empty:
print(f"⚠ API请求最终失败,返回数据库现有数据")
logger.warning("API请求最终失败,返回数据库现有数据")
return df
raise e # 如果没有缓存数据,抛出异常
# 计算等待时间 (指数退避: 2s, 4s, 8s...)
wait_time = 2 * (2 ** attempt)
print(f"⚠ 请求失败 (尝试 {attempt + 1}/{max_retries}),{wait_time}秒后重试... 错误: {e}")
logger.warning("请求失败(尝试 %s/%s),%s 秒后重试,错误: %s", attempt + 1, max_retries, wait_time, e)
time.sleep(wait_time)
if not klines:
@@ -179,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)
+29 -18
View File
@@ -99,7 +99,7 @@ def fmt_result(pred, actual):
def main():
print("=" * 110)
print("2026年2月 BTC 1h 涨跌预测准确率测试 (DLinear, N-BEATS, TTM-R2) - ONNX Version")
print("2026年3月 BTC 1h 涨跌预测准确率测试 (DLinear, N-BEATS, TTM-R2) - ONNX Version")
print("输入:预测K线的前一根 | 对比:预测结果 vs 预测K线收盘价 | 时间:北京时间(UTC+8)")
print("DLinear预测逻辑:整个df一次性预处理,按索引遍历窗口")
print("=" * 110)
@@ -168,9 +168,9 @@ def main():
ttm_session = None
# ========== 步骤2: 获取测试数据 ==========
print("\n[2/4] 获取2026年2月数据...")
print("\n[2/4] 获取2026年3月数据...")
fetcher = BinanceDataFetcher()
df = fetcher.fetch_ohlcv('BTC', '1h', start_date='2025-10-01', end_date='2026-03-01', use_cache=False)
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)}")
@@ -221,28 +221,38 @@ def main():
X_ttm = np.array(X_ttm, dtype=np.float32)
idx_ttm_map = {v: k for k, v in enumerate(idx_ttm)}
# 找出2月(北京时间)的所有预测目标K线索引
# 找出3月(北京时间)的所有预测目标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_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 == 2:
if target_time_bj.year == 2026 and target_time_bj.month == 3:
jan_targets.append(target_idx)
print(f"2月(北京时间)目标K线数: {len(jan_targets)}")
print(f"3月(北京时间)目标K线数: {len(jan_targets)}")
results = []
for target_idx in jan_targets:
target_time_utc = df.iloc[target_idx]['timestamp']
# 原始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线索引(前一根)
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
# 现在我们将预测目标设为下一根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
@@ -304,8 +314,8 @@ def main():
results.append({
'time_utc': target_time_utc,
'time_bj': target_time_bj,
'current_close': current_close,
'target_close': target_close,
'open': open_price,
'close': close_price,
'actual': actual_label,
'dlinear': dl_pred,
'nbeats': nb_pred,
@@ -324,7 +334,7 @@ def main():
total_samples = len(df_res)
print(f"\n[4/4] 准确率统计 (2026年2月,北京时间)")
print(f"\n[4/4] 准确率统计 (2026年3月,北京时间)")
print(f"总样本数: {total_samples}")
# ========== 每小时详细数据 ==========
@@ -332,8 +342,9 @@ def main():
print("每小时详细数据")
print("=" * 110)
# 改为显示开盘/收盘
hourly_header = (
f"{'北京时间':<18} {'前收盘':>10} {'现收盘':>10} {'实际':>4} "
f"{'北京时间':<18} {'开盘':>10} {'收盘':>10} {'实际':>4} "
f"{'DL':>3} {'':>1} {'NB':>3} {'':>1} {'TTM':>3} {'':>1} {'集成':>4} {'':<1}"
)
@@ -371,7 +382,7 @@ def main():
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"{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}"
)
+3
View File
@@ -24,6 +24,9 @@ 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()
# ==================== 时区与美股判断 ====================
+371
View File
@@ -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()
+27
View File
@@ -15,6 +15,10 @@ 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
@@ -23,6 +27,17 @@ 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'
@@ -50,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):
"""启动子进程(跨平台兼容)"""
@@ -146,6 +162,17 @@ def main():
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)
+5 -2
View File
@@ -34,9 +34,11 @@ 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:
@@ -119,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,
@@ -278,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
@@ -294,7 +297,7 @@ 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]:
+53
View File
@@ -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
+40
View File
@@ -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
+18 -2
View File
@@ -1,4 +1,10 @@
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:
"""飞书消息发送工具类"""
@@ -12,6 +18,16 @@ class LarkNotifier:
# 使用 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):
"""
发送通用文本告警
@@ -29,10 +45,10 @@ class LarkNotifier:
# 使用 data=payload 发送 Form Data
response = self.session.post(self.url, data=payload, timeout=5)
response.raise_for_status()
print(f"告警发送成功: {title}")
logger.info("告警发送成功: %s", title)
except Exception as e:
# 告警系统自身的错误只记录日志,绝不抛出异常影响主程序
print(f"发送飞书告警失败: {e}")
logger.error("发送飞书告警失败: %s", e)
# 实例化一个全局单例供其他模块直接导入使用
lark_notifier = LarkNotifier()