Author SHA1 Message Date
crypto_predictor 5ca1779d7d fix: 精简 requirements.txt,移除未使用的依赖
移除的依赖:
- gluonts: 项目中未使用
- pytorch-lightning: 非必需
- torchmetrics: 非必需
- onnx/onnxruntime: 非核心功能
- sqlalchemy: 使用 pymysql 直连
- jupyter/matplotlib/seaborn/pytest: 开发工具,非运行必需

保留的核心依赖:
- pandas, numpy, scikit-learn: 数据处理
- granite-tsfm: TTM-R2 模型
- pandas-ta: 技术指标
- fastapi, uvicorn: API 服务
- httpx, python-binance: 数据获取
- schedule: 任务调度
- pymysql, cryptography: 数据库

注意:PyTorch 需要单独安装(Windows 用 CPU 版本)
2026-02-07 22:28:46 +08:00
crypto_predictor 9320b79730 refactor: 重新整理 requirements.txt
主要更改:
- 修正 TTM 模型包名: tsfm_public -> granite-tsfm
- 添加缺失依赖: httpx, schedule, pytorch-lightning, torchmetrics
- 移除过时依赖: darts, u8darts
- 按功能分类组织依赖
- 添加 Windows/macOS 安装说明注释
- 更新安装脚本和文档中的包名
2026-02-07 22:26:07 +08:00
crypto_predictor 06800ddb43 fix: 修正 tsfm-public 包名,添加一键安装脚本
- 添加 install_windows.bat 一键安装脚本
- 修正文档中 tsfm_public -> tsfm-public (PyPI 包名用连字符)
- 添加 GitHub 备用安装方式
2026-02-07 22:02:56 +08:00
crypto_predictor 340c11d4ab docs: 添加 Windows 环境安装指南
包含完整的安装流程:
- Miniconda 安装
- Conda 环境创建
- PyTorch CPU 版本安装
- 项目依赖安装
- 数据库配置
- 环境变量配置
- 运行和验证步骤
- 常见问题解答
2026-02-07 21:55:56 +08:00
crypto_predictor d0696a42ba feat: 添加 Windows 启动脚本
- start_system.bat: CMD 批处理脚本,双击即可运行
- start_system.ps1: PowerShell 脚本,支持进程监控和优雅退出

两个脚本都会:
1. 激活 crypto_pred_312 conda 环境
2. 启动 API 服务 (端口 8000)
3. 启动自动交易程序
2026-02-07 21:54:46 +08:00
crypto_predictor 233f0c5d10 feat: 添加 Windows 兼容性支持,优化设备选择逻辑
- start_system.py: 跨平台进程管理
  - Windows 使用 CREATE_NEW_PROCESS_GROUP 和 terminate()
  - Unix 使用 setsid 和 killpg
  - 自动检测并启用 Windows ANSI 颜色支持

- 设备选择逻辑优化:
  - macOS: 优先使用 MPS,回退到 CPU
  - Windows/Linux: 强制使用 CPU(无 CUDA 依赖)

- 影响文件:
  - api/prediction_service.py
  - auto_trade_polymarket.py
  - training/config.py
2026-02-07 21:48:06 +08:00
crypto_predictor 8cdfb2a8e2 fix: 统一时区处理,准确率统计使用CST时间分组
- 增强 convert_to_cst() 函数,新增 source_tz 参数支持 UTC/ET/CST 输入
- 修改 get_monthly_stats() 使用 CONVERT_TZ 按 CST 月份统计
- 修改 get_daily_stats() 使用 CONVERT_TZ 按 CST 日期分组
- 修改 get_hourly_details() 按 CST 日期筛选和小时排序
- 交易记录使用 source_tz='CST' 避免重复转换
- 前端表头添加'(北京)'时区标注
2026-02-07 18:14:37 +08:00
crypto_predictor 6182630203 docs: 更新 AGENTS.md 开发指南,反映最新的架构变更 2026-02-07 16:58:44 +08:00
crypto_predictor a12e035c1c feat: 添加系统启动脚本 start_system.py 2026-02-07 16:50:58 +08:00
crypto_predictor 55453d3739 feat(frontend): 添加交易看板,统一时区为CST
refactor(db): 移除所有SQLite相关代码,迁移至MySQL
2026-02-07 16:48:14 +08:00
crypto_predictor 5e0d6b18ef chore: Secure secrets and add MySQL deps 2026-02-06 17:24:33 +08:00
crypto_predictor 8de6c54887 feat: Add MySQL support and migration script 2026-02-06 16:35:08 +08:00
crypto_predictor 006ccb2a1d chore: Remove environment.yml 2026-02-06 16:28:54 +08:00
crypto_predictor 4bd5cb44e4 data: Add market_data.db 2026-02-06 15:50:34 +08:00
crypto_predictor bd722242ad docs: Add AGENTS.md and initial trading database 2026-02-06 15:49:43 +08:00
crypto_predictor db937fdcf9 feat: Add DB logging, price monitoring, and duplicate trade prevention 2026-02-06 15:48:45 +08:00
crypto_predictor fcf64f4a3d 添加数据库文件 2026-02-04 18:38:09 +08:00
crypto_predictor 59fcb11ffa 接入数据库 2026-02-04 18:35:10 +08:00
27 changed files with 4514 additions and 230 deletions

No files matched your search

-32
View File
@@ -1,32 +0,0 @@
# Binance API (public data - no keys needed)
BINANCE_API_URL=https://api.binance.com
# 代理配置 (SOCKS5)
SOCKS5_PROXY=socks5://127.0.0.1:7890
USE_PROXY=true
# Data Configuration
DATA_CACHE_DIR=./data/cache
LOOKBACK_1H=168 # 1 week for 1h data
LOOKBACK_15M=672 # 1 week for 15m data
FORECAST_HORIZON=24 # Predict 24 periods ahead
# Training Configuration
BATCH_SIZE=64
LEARNING_RATE=0.001
MAX_EPOCHS=100
EARLY_STOPPING_PATIENCE=10
TRAIN_SPLIT=0.7
VAL_SPLIT=0.15
TEST_SPLIT=0.15
# Model Paths
MODEL_DIR=./models/checkpoints
ONNX_DIR=./models/onnx
# API Configuration
API_PORT=5000
API_HOST=0.0.0.0
# Classification Threshold
PRICE_CHANGE_THRESHOLD=0.015 # 1.5% threshold for binary labels
+7
View File
@@ -30,3 +30,10 @@ API_HOST=0.0.0.0
# Classification Threshold
PRICE_CHANGE_THRESHOLD=0.015 # 1.5% threshold for binary labels
# MySQL Configuration
MYSQL_HOST=localhost
MYSQL_PORT=3306
MYSQL_USER=user
MYSQL_PASSWORD=password
MYSQL_DB=crypto_predictor
+4 -1
View File
@@ -9,9 +9,11 @@ venv/
env/
ENV/
# Sensitive
.env
# 数据和缓存
data/cache/*.csv
*.db
# 模型检查点(保留最佳模型,忽略训练过程文件)
models/checkpoints/*/epoch_*.pt
@@ -38,3 +40,4 @@ logs/
# 临时文件
*.tmp
.cache/
test_*.py
+73
View File
@@ -0,0 +1,73 @@
# AGENTS.md - 智能代理开发指南
本文件定义了在该项目中进行开发的 AI 代理(Agent)的行为准则、技术规范和操作流程。请严格遵守。
## 1. 核心行为准则 (Core Mandates)
**最高优先级原则:**
* **语言要求**:所有回答必须使用**中文**。
* **称呼要求**:每次回答必须加上**“主人”**。
* **最小改动 (Minimal Change)**:只修改完成任务所必须的代码。**禁止**为了“最佳实践”或“代码美观”而进行未被请求的重构。
* **环境约束**:本项目严格使用 **Conda** 环境 (`crypto_pred_312`)。禁止使用 venv/poetry/pipenv。
* **复用优先**:在编写新功能前,必须检查 `data/`, `models/`, `api/` 下是否已有类似实现。
## 2. 环境与构建 (Environment & Build)
* **Python 环境**: `conda activate crypto_pred_312`
* **依赖安装**: `pip install -r requirements.txt`
* **数据库**:
* **仅支持 MySQL**。禁止引入 SQLite 代码。
* 配置:通过 `.env` 文件配置 `MYSQL_HOST`, `MYSQL_USER` 等参数。
* 连接管理:使用 `data/database.py` 中的 `DatabaseManager` 或 `api/models.py` 中的 `get_db_connection`。
## 3. 项目结构说明 (Project Structure)
* **`api/`**: 核心服务层
* `main.py`: FastAPI 主应用入口。
* `models.py`: 数据库模型与操作(MySQL),包含时区转换逻辑。
* `prediction_service.py`: 预测业务逻辑。
* `static/`: Vue 3 前端静态资源(Dashboard)。
* **`data/`**: 数据层
* `database.py`: 市场数据 (OHLCV) 数据库管理。
* `trading_db.py`: 交易历史数据库管理。
* **`models/`**: 机器学习模型定义 (DLinear, TTM-R2 等) 及训练产物。
* **`scripts/`**: 辅助工具脚本。
* **根目录**:
* `start_system.py`: **系统统一启动脚本**(同时启动 API 和自动交易)。
* `auto_trade_polymarket.py`: 自动交易主程序。
## 4. 开发规范 (Development Standards)
### 4.1 时区处理 (Timezone)
* **存储**: 数据库中时间戳通常为 **UTC** 或 **ET (美东时间)**。
* **展示**: 前端及日志输出必须转换为 **CST (北京时间)**。
* **工具**: 使用 `api.models.convert_to_cst` 进行转换。
### 4.2 代码风格
* **注释**: 必须使用**中文**解释“为什么这样做”。
* **命名**: 变量/函数 `snake_case`,类名 `PascalCase`。
* **类型提示**: 推荐在关键接口使用 Type Hints。
### 4.3 数据库操作
* **禁止**使用 SQLite。
* 使用 `pymysql` 或 `sqlalchemy` 进行操作。
* 确保所有数据库连接都使用上下文管理器 (`with` 语句) 以自动关闭。
## 5. 测试与验证 (Testing & QA)
* **运行测试**: `pytest`
* **代码检查**: `flake8 .`
* **本地验证**:
* 修改 API 或前端后,运行 `./start_system.py` 并访问 `http://localhost:8000` 验证。
* 修改模型后,运行 `python ensemble_predict.py` 验证预测逻辑。
## 6. 禁止事项 (Constraints)
* ❌ **禁止**猜测业务逻辑。
* ❌ **禁止**删除看似无用但可能被外部脚本调用的代码。
* ❌ **禁止**在未确认的情况下升级核心库版本 (如 torch, pandas)。
* ❌ **禁止**输出英语回复 (代码变量名除外)。
* ❌ **禁止**恢复已删除的 SQLite 相关代码。
---
*遵循此指南,确保高效、安全地协助主人完成任务。*
+1
View File
@@ -0,0 +1 @@
# api模块初始化
+191
View File
@@ -0,0 +1,191 @@
"""
FastAPI 主应用 - 交易看板 API
"""
import os
import sys
from datetime import datetime
from typing import Optional
from zoneinfo import ZoneInfo
from fastapi import FastAPI, Query, HTTPException
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
from fastapi.middleware.cors import CORSMiddleware
# 添加项目路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from api.models import TradeModel, PredictionModel
from api.prediction_service import get_prediction_service
# 创建 FastAPI 应用
app = FastAPI(
title="Crypto Predictor Dashboard API",
description="BTC 预测交易看板 API",
version="1.0.0"
)
# 启用 CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 静态文件目录
STATIC_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'static')
# 挂载静态文件
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
# ==================== 页面路由 ====================
@app.get("/")
async def index():
"""返回主页"""
return FileResponse(os.path.join(STATIC_DIR, 'index.html'))
# ==================== 交易 API ====================
@app.get("/api/trades/current")
async def get_current_trade():
"""获取当前/最新交易"""
try:
trade = TradeModel.get_current_trade()
if trade:
return {"success": True, "data": trade}
return {"success": True, "data": None, "message": "暂无交易记录"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/trades/history")
async def get_trade_history(
page: int = Query(1, ge=1, description="页码"),
limit: int = Query(10, ge=1, le=100, description="每页条数"),
month: Optional[str] = Query(None, description="月份筛选 (YYYY-MM)"),
day: Optional[str] = Query(None, description="日期筛选 (YYYY-MM-DD)")
):
"""获取历史交易记录"""
try:
result = TradeModel.get_history(page=page, limit=limit, month=month, day=day)
return {"success": True, **result}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# ==================== 准确率 API ====================
@app.get("/api/accuracy/current")
async def get_current_prediction():
"""获取当前小时的预测"""
try:
service = get_prediction_service()
prediction = service.predict_current_hour()
return {"success": True, "data": prediction}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/accuracy/monthly")
async def get_monthly_accuracy(
month: Optional[str] = Query(None, description="月份 (YYYY-MM),默认当月")
):
"""获取月度准确率统计"""
try:
# 默认当月
if not month:
et_tz = ZoneInfo('America/New_York')
month = datetime.now(et_tz).strftime('%Y-%m')
# 确保有数据
service = get_prediction_service()
service.ensure_predictions_for_month(month)
# 获取统计
stats = PredictionModel.get_monthly_stats(month)
return {
"success": True,
"month": month,
"data": stats
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/accuracy/daily")
async def get_daily_accuracy(
month: Optional[str] = Query(None, description="月份 (YYYY-MM),默认当月"),
page: int = Query(1, ge=1, description="页码"),
limit: int = Query(10, ge=1, le=100, description="每页条数")
):
"""获取每日准确率统计"""
try:
# 默认当月
if not month:
et_tz = ZoneInfo('America/New_York')
month = datetime.now(et_tz).strftime('%Y-%m')
# 确保有数据
service = get_prediction_service()
service.ensure_predictions_for_month(month)
# 获取统计
result = PredictionModel.get_daily_stats(month=month, page=page, limit=limit)
return {
"success": True,
"month": month,
**result
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/accuracy/hourly/{date}")
async def get_hourly_accuracy(date: str):
"""获取指定日期的每小时准确率详情"""
try:
# 确保有数据
service = get_prediction_service()
service.ensure_predictions_for_date(date)
# 获取详情
details = PredictionModel.get_hourly_details(date)
return {
"success": True,
"date": date,
"data": details
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# ==================== 状态 API ====================
@app.get("/api/status")
async def get_status():
"""获取服务状态"""
et_tz = ZoneInfo('America/New_York')
now_et = datetime.now(et_tz)
return {
"success": True,
"status": "running",
"server_time": datetime.now().isoformat(),
"server_time_et": now_et.isoformat(),
"version": "1.0.0"
}
# ==================== 启动入口 ====================
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
+475
View File
@@ -0,0 +1,475 @@
"""
数据库模型和操作类 - MySQL
"""
import pymysql
import os
from typing import Optional, List, Dict
from datetime import datetime
from contextlib import contextmanager
from zoneinfo import ZoneInfo
from dotenv import load_dotenv
load_dotenv()
# 时区定义
TZ_UTC = ZoneInfo('UTC')
TZ_ET = ZoneInfo('America/New_York') # 美东时间 (Polymarket)
TZ_CST = ZoneInfo('Asia/Shanghai') # 中国标准时间 (前端显示)
def convert_to_cst(timestamp_val, source_tz: str = 'UTC') -> str:
"""
将时间戳转换为 CST 时间字符串
Args:
timestamp_val: datetime 对象或 ISO 格式时间戳字符串
source_tz: 输入时区 - 'UTC'(默认,币安数据), 'ET'(Polymarket), 'CST'(已是北京时间)
Returns:
CST 时间字符串 (YYYY-MM-DD HH:MM:SS)
"""
if not timestamp_val:
return ""
try:
# 如果是 datetime 对象 (MySQL 返回)
if isinstance(timestamp_val, datetime):
dt = timestamp_val
else:
# 字符串格式
dt = datetime.fromisoformat(str(timestamp_val).replace('Z', '+00:00'))
# 根据来源时区设置时区信息
if dt.tzinfo is None:
if source_tz == 'ET':
dt = dt.replace(tzinfo=TZ_ET)
elif source_tz == 'CST':
# 已经是 CST,直接格式化返回
return dt.strftime('%Y-%m-%d %H:%M:%S')
else: # 默认 UTC
dt = dt.replace(tzinfo=TZ_UTC)
# 如果来源已经是 CST,直接格式化
if source_tz == 'CST':
return dt.strftime('%Y-%m-%d %H:%M:%S')
# 转换为 CST
dt_cst = dt.astimezone(TZ_CST)
return dt_cst.strftime('%Y-%m-%d %H:%M:%S')
except Exception:
return str(timestamp_val)
@contextmanager
def get_db_connection():
"""获取 MySQL 数据库连接"""
conn = pymysql.connect(
host=os.getenv('MYSQL_HOST'),
port=int(os.getenv('MYSQL_PORT', 3306)),
user=os.getenv('MYSQL_USER'),
password=os.getenv('MYSQL_PASSWORD'),
database=os.getenv('MYSQL_DB'),
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor
)
try:
yield conn
finally:
conn.close()
def _dict_from_row(row):
"""将数据库行转换为字典"""
if row is None:
return None
if isinstance(row, dict):
return row
return dict(row)
def init_model_predictions_table():
"""初始化 model_predictions 表"""
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS model_predictions (
id INT AUTO_INCREMENT PRIMARY KEY,
timestamp DATETIME NOT NULL,
timestamp_et VARCHAR(50),
date VARCHAR(10),
hour INT,
model_name VARCHAR(50) NOT NULL,
prediction INT,
probability DOUBLE,
actual_result INT,
is_correct INT,
open_price DOUBLE,
close_price DOUBLE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE KEY uk_date_hour_model (date, hour, model_name),
INDEX idx_date (date),
INDEX idx_model (model_name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
''')
conn.commit()
# 初始化表
init_model_predictions_table()
class TradeModel:
"""交易记录模型"""
@staticmethod
def get_current_trade() -> Optional[Dict]:
"""获取最新的交易记录"""
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT * FROM trades
ORDER BY timestamp DESC
LIMIT 1
""")
row = cursor.fetchone()
result = _dict_from_row(row)
if result and result.get('timestamp'):
# 交易记录时间已经是 CST,直接格式化
result['timestamp_cst'] = convert_to_cst(result['timestamp'], source_tz='CST')
return result
@staticmethod
def get_history(page: int = 1, limit: int = 10,
month: Optional[str] = None,
day: Optional[str] = None) -> Dict:
"""
获取历史交易记录
Args:
page: 页码
limit: 每页条数
month: 月份筛选 (YYYY-MM)
day: 日期筛选 (YYYY-MM-DD)
"""
offset = (page - 1) * limit
with get_db_connection() as conn:
cursor = conn.cursor()
# 构建查询条件
where_clauses = []
params = []
if month:
where_clauses.append("DATE_FORMAT(timestamp, '%%Y-%%m') = %s")
params.append(month)
if day:
where_clauses.append("DATE(timestamp) = %s")
params.append(day)
where_sql = " AND ".join(where_clauses) if where_clauses else "1=1"
# 查询数据
sql = f"SELECT * FROM trades WHERE {where_sql} ORDER BY timestamp DESC LIMIT %s OFFSET %s"
cursor.execute(sql, params + [limit, offset])
rows = cursor.fetchall()
# 查询总数
count_sql = f"SELECT COUNT(*) as cnt FROM trades WHERE {where_sql}"
cursor.execute(count_sql, params)
result = cursor.fetchone()
total = result['cnt'] if result else 0
# 转换每条记录的时间戳
data_list = []
for row in rows:
item = _dict_from_row(row)
if item and item.get('timestamp'):
# 交易记录时间已经是 CST,直接格式化
item['timestamp_cst'] = convert_to_cst(item['timestamp'], source_tz='CST')
data_list.append(item)
return {
"data": data_list,
"page": page,
"limit": limit,
"total": total,
"total_pages": (total + limit - 1) // limit if limit > 0 else 0
}
class PredictionModel:
"""预测记录模型"""
@staticmethod
def get_prediction(date: str, hour: int, model_name: str) -> Optional[Dict]:
"""获取指定时间和模型的预测记录"""
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT * FROM model_predictions
WHERE date = %s AND hour = %s AND model_name = %s
""", (date, hour, model_name))
row = cursor.fetchone()
return _dict_from_row(row)
@staticmethod
def save_prediction(data: Dict) -> int:
"""
保存预测记录
Args:
data: 预测数据字典
Returns:
插入的记录ID
"""
with get_db_connection() as conn:
cursor = conn.cursor()
sql = """
INSERT INTO model_predictions
(timestamp, timestamp_et, date, hour, model_name,
prediction, probability, actual_result, is_correct,
open_price, close_price)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
prediction = VALUES(prediction),
probability = VALUES(probability),
actual_result = VALUES(actual_result),
is_correct = VALUES(is_correct),
open_price = VALUES(open_price),
close_price = VALUES(close_price)
"""
cursor.execute(sql, (
data.get('timestamp'),
data.get('timestamp_et'),
data.get('date'),
data.get('hour'),
data.get('model_name'),
data.get('prediction'),
data.get('probability'),
data.get('actual_result'),
data.get('is_correct'),
data.get('open_price'),
data.get('close_price')
))
conn.commit()
return cursor.lastrowid or 0
@staticmethod
def update_actual_result(date: str, hour: int, model_name: str,
actual_result: int, close_price: float) -> bool:
"""更新实际结果"""
with get_db_connection() as conn:
cursor = conn.cursor()
# 先获取预测值
cursor.execute("""
SELECT prediction FROM model_predictions
WHERE date = %s AND hour = %s AND model_name = %s
""", (date, hour, model_name))
row = cursor.fetchone()
if not row:
return False
prediction = row['prediction']
is_correct = 1 if prediction == actual_result else 0
cursor.execute("""
UPDATE model_predictions
SET actual_result = %s, is_correct = %s, close_price = %s
WHERE date = %s AND hour = %s AND model_name = %s
""", (actual_result, is_correct, close_price, date, hour, model_name))
conn.commit()
return True
@staticmethod
def get_monthly_stats(month: str) -> Dict:
"""
获取月度统计
Args:
month: 月份 (YYYY-MM)
Returns:
各模型的统计数据
"""
with get_db_connection() as conn:
cursor = conn.cursor()
sql = """
SELECT
model_name,
COUNT(*) as total,
SUM(CASE WHEN is_correct = 1 THEN 1 ELSE 0 END) as correct,
SUM(CASE WHEN is_correct = 0 THEN 1 ELSE 0 END) as incorrect
FROM model_predictions
WHERE DATE_FORMAT(CONVERT_TZ(timestamp, '+00:00', '+08:00'), '%%Y-%%m') = %s
AND actual_result IS NOT NULL
GROUP BY model_name
"""
cursor.execute(sql, (month,))
rows = cursor.fetchall()
result = {}
for row in rows:
row_dict = _dict_from_row(row)
model = row_dict['model_name']
total = row_dict['total']
correct = row_dict['correct'] or 0
incorrect = row_dict['incorrect'] or 0
accuracy = (correct / total * 100) if total > 0 else 0
result[model] = {
"total": total,
"correct": correct,
"incorrect": incorrect,
"accuracy": round(accuracy, 2)
}
return result
@staticmethod
def get_daily_stats(month: str, page: int = 1, limit: int = 10) -> Dict:
"""
获取每日统计
Args:
month: 月份 (YYYY-MM)
page: 页码
limit: 每页条数
"""
with get_db_connection() as conn:
cursor = conn.cursor()
sql = """
SELECT
DATE(CONVERT_TZ(timestamp, '+00:00', '+08:00')) as date_cst,
model_name,
COUNT(*) as total,
SUM(CASE WHEN is_correct = 1 THEN 1 ELSE 0 END) as correct,
SUM(CASE WHEN is_correct = 0 THEN 1 ELSE 0 END) as incorrect
FROM model_predictions
WHERE DATE_FORMAT(CONVERT_TZ(timestamp, '+00:00', '+08:00'), '%%Y-%%m') = %s
AND actual_result IS NOT NULL
GROUP BY date_cst, model_name
ORDER BY date_cst DESC
"""
cursor.execute(sql, (month,))
rows = cursor.fetchall()
# 重组数据: 按日期合并不同模型的统计
daily_data = {}
for row in rows:
row_dict = _dict_from_row(row)
date = str(row_dict['date_cst']) # CST 日期
if date not in daily_data:
daily_data[date] = {"date": date}
model = row_dict['model_name']
total = row_dict['total']
correct = row_dict['correct'] or 0
incorrect = row_dict['incorrect'] or 0
accuracy = (correct / total * 100) if total > 0 else 0
daily_data[date][model] = {
"total": total,
"correct": correct,
"incorrect": incorrect,
"accuracy": round(accuracy, 2)
}
# 排序并分页
sorted_dates = sorted(daily_data.keys(), reverse=True)
total_count = len(sorted_dates)
offset = (page - 1) * limit
paginated_dates = sorted_dates[offset:offset + limit]
data = [daily_data[d] for d in paginated_dates]
return {
"data": data,
"page": page,
"limit": limit,
"total": total_count,
"total_pages": (total_count + limit - 1) // limit if limit > 0 else 0
}
@staticmethod
def get_hourly_details(date: str) -> List[Dict]:
"""
获取指定日期的每小时详情
Args:
date: 日期 (YYYY-MM-DD, CST 北京时间)
"""
with get_db_connection() as conn:
cursor = conn.cursor()
# 按 CST 日期筛选,并获取 CST 小时
cursor.execute("""
SELECT
*,
HOUR(CONVERT_TZ(timestamp, '+00:00', '+08:00')) as hour_cst,
CONVERT_TZ(timestamp, '+00:00', '+08:00') as timestamp_cst_dt
FROM model_predictions
WHERE DATE(CONVERT_TZ(timestamp, '+00:00', '+08:00')) = %s
ORDER BY hour_cst, model_name
""", (date,))
rows = cursor.fetchall()
# 按 CST 小时重组数据
hourly_data = {}
for row in rows:
row_dict = _dict_from_row(row)
hour_cst = row_dict['hour_cst'] # 使用 CST 小时
if hour_cst not in hourly_data:
# 将 timestamp 转换为 CST 时间字符串
timestamp_cst = convert_to_cst(row_dict.get('timestamp', ''))
hourly_data[hour_cst] = {
"hour": hour_cst, # 返回 CST 小时
"timestamp_et": row_dict['timestamp_et'],
"timestamp_cst": timestamp_cst,
"open_price": row_dict['open_price'],
"close_price": row_dict['close_price'],
"actual_result": row_dict['actual_result'],
"models": {}
}
model = row_dict['model_name']
hourly_data[hour_cst]["models"][model] = {
"prediction": row_dict['prediction'],
"probability": row_dict['probability'],
"is_correct": row_dict['is_correct']
}
# 按 CST 小时排序返回
return [hourly_data[h] for h in sorted(hourly_data.keys())]
@staticmethod
def get_missing_hours(date: str, model_name: str) -> List[int]:
"""
获取指定日期缺少预测的小时列表
Args:
date: 日期 (YYYY-MM-DD)
model_name: 模型名称
Returns:
缺少预测的小时列表
"""
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT hour FROM model_predictions
WHERE date = %s AND model_name = %s
""", (date, model_name))
rows = cursor.fetchall()
existing_hours = {row['hour'] for row in rows}
all_hours = set(range(24))
return sorted(list(all_hours - existing_hours))
+390
View File
@@ -0,0 +1,390 @@
"""
Polymarket API 客户端
用于与Polymarket CLOB API交互,支持:
- 搜索BTC相关市场
- 获取订单簿和价格
- 下单交易
"""
import os
import httpx
from typing import Optional, Dict, List, Any
from dataclasses import dataclass
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
@dataclass
class MarketInfo:
"""市场信息"""
condition_id: str
question: str
outcomes: List[str]
outcome_prices: List[float]
tokens: List[Dict]
volume: float
liquidity: float
end_date: str
class PolymarketClient:
"""Polymarket API客户端"""
GAMMA_API = "https://gamma-api.polymarket.com"
CLOB_API = "https://clob.polymarket.com"
def __init__(self, private_key: Optional[str] = None):
"""
初始化客户端
Args:
private_key: 钱包私钥(用于交易),如不提供则只能读取数据
"""
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
# HTTP客户端 - 禁用压缩避免brotli问题
self.http = httpx.Client(
timeout=30.0,
proxy=proxy,
headers={'Accept-Encoding': 'identity'} # 禁用压缩
)
# 如果有私钥,初始化CLOB客户端
self.clob_client = None
if self.private_key:
self._init_clob_client()
def _init_clob_client(self):
"""初始化CLOB交易客户端"""
try:
from py_clob_client.client import ClobClient
from py_clob_client.clob_types import ApiCreds
# Polygon链ID
chain_id = 137
# 初始化客户端
self.clob_client = ClobClient(
host=self.CLOB_API,
key=self.private_key,
chain_id=chain_id,
)
# 获取或创建API凭证
self.clob_client.set_api_creds(self.clob_client.create_or_derive_api_creds())
print("✓ CLOB客户端初始化成功")
except Exception as e:
print(f"⚠️ CLOB客户端初始化失败: {e}")
self.clob_client = None
def search_markets(self, query: str = "BTC", limit: int = 20) -> List[Dict]:
"""
搜索市场
Args:
query: 搜索关键词
limit: 返回数量限制
Returns:
市场列表
"""
try:
response = self.http.get(
f"{self.GAMMA_API}/markets",
params={
"active": "true",
"closed": "false",
"limit": limit,
}
)
response.raise_for_status()
markets = response.json()
# 过滤包含关键词的市场
filtered = [m for m in markets if query.lower() in m.get("question", "").lower()]
return filtered
except Exception as e:
print(f"搜索市场失败: {e}")
return []
def search_btc_1h_markets(self) -> List[Dict]:
"""
搜索BTC 1小时价格预测市场
Returns:
BTC 1H相关市场列表
"""
try:
response = self.http.get(
f"{self.GAMMA_API}/markets",
params={
"active": "true",
"closed": "false",
"limit": 100,
}
)
response.raise_for_status()
markets = response.json()
# 过滤BTC 1小时相关市场
btc_1h_markets = []
keywords = ["btc", "bitcoin"]
time_keywords = ["1h", "1 hour", "hourly", "hour"]
for m in markets:
question = m.get("question", "").lower()
# 检查是否包含BTC关键词
has_btc = any(kw in question for kw in keywords)
# 检查是否包含1小时关键词
has_1h = any(kw in question for kw in time_keywords)
if has_btc and has_1h:
btc_1h_markets.append(m)
return btc_1h_markets
except Exception as e:
print(f"搜索BTC 1H市场失败: {e}")
return []
def get_market_info(self, condition_id: str) -> Optional[MarketInfo]:
"""
获取市场详细信息
Args:
condition_id: 市场条件ID
Returns:
市场信息对象
"""
try:
response = self.http.get(f"{self.GAMMA_API}/markets/{condition_id}")
response.raise_for_status()
data = response.json()
return MarketInfo(
condition_id=data.get("conditionId", ""),
question=data.get("question", ""),
outcomes=data.get("outcomes", []),
outcome_prices=data.get("outcomePrices", []),
tokens=data.get("tokens", []),
volume=float(data.get("volume", 0)),
liquidity=float(data.get("liquidity", 0)),
end_date=data.get("endDate", ""),
)
except Exception as e:
print(f"获取市场信息失败: {e}")
return None
def get_orderbook(self, token_id: str) -> Dict:
"""
获取订单簿
Args:
token_id: 代币ID
Returns:
订单簿数据
"""
try:
response = self.http.get(
f"{self.CLOB_API}/book",
params={"token_id": token_id}
)
response.raise_for_status()
return response.json()
except Exception as e:
print(f"获取订单簿失败: {e}")
return {}
def get_price(self, token_id: str) -> Optional[float]:
"""
获取代币当前价格
Args:
token_id: 代币ID
Returns:
当前价格(0-1之间)
"""
try:
response = self.http.get(
f"{self.CLOB_API}/price",
params={"token_id": token_id, "side": "buy"}
)
response.raise_for_status()
data = response.json()
return float(data.get("price", 0))
except Exception as e:
print(f"获取价格失败: {e}")
return None
def place_market_order(
self,
token_id: str,
side: str, # "BUY" or "SELL"
amount: float, # USDC金额
) -> Optional[Dict]:
"""
下市价单
Args:
token_id: 代币ID
side: 买卖方向 "BUY" 或 "SELL"
amount: 下注金额(USDC)
Returns:
订单响应
"""
if not self.clob_client:
print("❌ CLOB客户端未初始化,无法下单")
return None
try:
from py_clob_client.order_builder.constants import BUY, SELL
order_side = BUY if side.upper() == "BUY" else SELL
# 创建市价单
order = self.clob_client.create_market_order(
token_id=token_id,
amount=amount,
side=order_side,
)
# 提交订单
response = self.clob_client.post_order(order)
print(f"✓ 订单已提交: {response}")
return response
except Exception as e:
print(f"❌ 下单失败: {e}")
return None
def place_limit_order(
self,
token_id: str,
side: str, # "BUY" or "SELL"
price: float, # 价格(0-1)
size: float, # 数量
) -> Optional[Dict]:
"""
下限价单
Args:
token_id: 代币ID
side: 买卖方向 "BUY" 或 "SELL"
price: 价格(0-1之间)
size: 购买数量
Returns:
订单响应
"""
if not self.clob_client:
print("❌ CLOB客户端未初始化,无法下单")
return None
try:
from py_clob_client.order_builder.constants import BUY, SELL
order_side = BUY if side.upper() == "BUY" else SELL
# 创建限价单
order = self.clob_client.create_limit_order(
token_id=token_id,
price=price,
size=size,
side=order_side,
)
# 提交订单
response = self.clob_client.post_order(order)
print(f"✓ 限价单已提交: {response}")
return response
except Exception as e:
print(f"❌ 下单失败: {e}")
return None
def get_positions(self) -> List[Dict]:
"""
获取当前仓位
Returns:
仓位列表
"""
if not self.clob_client:
print("❌ CLOB客户端未初始化")
return []
try:
positions = self.clob_client.get_positions()
return positions
except Exception as e:
print(f"获取仓位失败: {e}")
return []
def cancel_order(self, order_id: str) -> bool:
"""
取消订单
Args:
order_id: 订单ID
Returns:
是否成功
"""
if not self.clob_client:
print("❌ CLOB客户端未初始化")
return False
try:
self.clob_client.cancel(order_id)
print(f"✓ 订单已取消: {order_id}")
return True
except Exception as e:
print(f"❌ 取消订单失败: {e}")
return False
def close(self):
"""关闭客户端"""
self.http.close()
# 测试函数
def test_search_markets():
"""测试搜索市场功能"""
client = PolymarketClient()
print("搜索BTC相关市场...")
markets = client.search_markets("BTC", limit=10)
if markets:
print(f"\n找到 {len(markets)} 个BTC相关市场:\n")
for m in markets[:5]:
print(f" 问题: {m.get('question', 'N/A')}")
print(f" 条件ID: {m.get('conditionId', 'N/A')}")
print(f" 结果: {m.get('outcomes', [])}")
print(f" 价格: {m.get('outcomePrices', [])}")
print("-" * 50)
else:
print("未找到BTC相关市场")
print("\n搜索BTC 1H市场...")
btc_1h = client.search_btc_1h_markets()
if btc_1h:
print(f"找到 {len(btc_1h)} 个BTC 1H市场")
for m in btc_1h:
print(f" {m.get('question', 'N/A')}")
else:
print("未找到BTC 1H市场")
client.close()
if __name__ == "__main__":
test_search_markets()
+483
View File
@@ -0,0 +1,483 @@
"""
预测服务 - DLinear 和 TTM-R2 模型预测
"""
import os
import sys
import pickle
from datetime import datetime, timedelta
from typing import Optional, List, Dict, Tuple
from zoneinfo import ZoneInfo
import torch
import numpy as np
import pandas as pd
import pandas_ta as ta
from sklearn.preprocessing import StandardScaler
# 添加项目路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from data.fetcher import BinanceDataFetcher
from api.models import PredictionModel
class ClassificationPreprocessor:
"""DLinear分类预处理器"""
def __init__(self, lookback: int = 168, forecast_horizon: int = 1):
self.lookback = lookback
self.forecast_horizon = forecast_horizon
self.scaler = StandardScaler()
self.is_fitted = False
def add_features(self, df: pd.DataFrame) -> pd.DataFrame:
df = df.copy()
df['rsi'] = ta.rsi(df['close'], length=14)
macd = ta.macd(df['close'])
if macd is not None and not macd.empty:
df['macd'] = macd.iloc[:, 0]
df['macd_hist'] = macd.iloc[:, 1]
df['macd_signal'] = macd.iloc[:, 2]
return df
def create_labels(self, df: pd.DataFrame) -> pd.DataFrame:
df = df.copy()
df['future_price'] = df['close'].shift(-self.forecast_horizon)
df['future_return'] = (df['future_price'] - df['close']) / df['close']
df['label'] = (df['future_return'] > 0).astype(int)
return df[:-self.forecast_horizon]
def preprocess(self, df: pd.DataFrame, fit: bool = True):
df = self.add_features(df)
df = self.create_labels(df)
df = df.dropna().reset_index(drop=True)
feature_cols = ['open', 'high', 'low', 'close', 'volume',
'rsi', 'macd', 'macd_hist', 'macd_signal']
if fit:
scaled_features = self.scaler.fit_transform(df[feature_cols])
self.is_fitted = True
self.feature_columns = feature_cols
else:
scaled_features = self.scaler.transform(df[feature_cols])
X, y = [], []
for i in range(len(scaled_features) - self.lookback):
X.append(scaled_features[i:i + self.lookback])
y.append(df['label'].iloc[i + self.lookback])
return np.array(X), np.array(y), df
class CustomUnpickler(pickle.Unpickler):
"""自定义反序列化器,用于处理 __main__ 模块的类"""
def find_class(self, module, name):
if module == "__main__":
if name == "ClassificationPreprocessor":
return ClassificationPreprocessor
return super().find_class(module, name)
class PredictionService:
"""预测服务 - 管理 DLinear 和 TTM-R2 模型"""
# 类级别的锁,确保多个请求不会同时生成预测
_prediction_lock = None
def __init__(self):
import threading
import platform
# 初始化锁
if PredictionService._prediction_lock is None:
PredictionService._prediction_lock = threading.Lock()
# 设备选择:macOS 用 MPS,Windows/Linux 用 CPU
if platform.system() == 'Darwin': # macOS
self.device = torch.device('mps' if torch.backends.mps.is_available() else 'cpu')
else: # Windows / Linux
self.device = torch.device('cpu')
print(f"使用设备: {self.device}")
self.et_tz = ZoneInfo('America/New_York')
self.fetcher = BinanceDataFetcher()
# 模型和预处理器
self.dlinear_model = None
self.dlinear_preprocessor = None
self.ttm_model = None
# 记录正在处理的日期和月份,避免重复处理
self._processing_dates = set()
self._processing_months = set()
self._load_models()
def _load_models(self):
"""加载模型"""
print("正在加载预测模型...")
# 加载 DLinear
from models.dlinear.model import DLinear
dlinear_path = './models/checkpoints/dlinear_classification_2026/best_model.pt'
preprocessor_path = './models/checkpoints/dlinear_classification_2026/preprocessor.pkl'
checkpoint = torch.load(dlinear_path, map_location=self.device, weights_only=False)
self.dlinear_model = DLinear(checkpoint['config'])
self.dlinear_model.load_state_dict(checkpoint['model_state_dict'])
self.dlinear_model = self.dlinear_model.to(self.device).eval()
with open(preprocessor_path, 'rb') as f:
try:
self.dlinear_preprocessor = pickle.load(f)
except AttributeError:
f.seek(0)
self.dlinear_preprocessor = CustomUnpickler(f).load()
print(f" ✓ DLinear 已加载")
# 加载 TTM-R2
from tsfm_public.models.tinytimemixer import TinyTimeMixerForPrediction
self.ttm_model = TinyTimeMixerForPrediction.from_pretrained(
"ibm-granite/granite-timeseries-ttm-r2",
trust_remote_code=True
)
self.ttm_model = self.ttm_model.to(self.device).eval()
print(f" ✓ TTM-R2 已加载")
print(f" 设备: {self.device}")
def _get_btc_data(self, days: int = 30) -> pd.DataFrame:
"""获取BTC历史数据"""
end_date = (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d')
start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
df = self.fetcher.fetch_ohlcv(
symbol='BTC',
timeframe='1h',
start_date=start_date,
end_date=end_date,
use_cache=False
)
return df
def predict_dlinear(self, df: pd.DataFrame) -> Tuple[int, float]:
"""
使用 DLinear 预测下一小时的涨跌
输入: df 包含历史K线数据 (最后一行是最新的已完成K线)
输出: 预测下一小时相对于 df 最后一行收盘价的涨跌
Returns:
(prediction, probability): 预测结果(0=跌, 1=涨)和概率
"""
# 预处理: 使用全部数据,不去掉最后一行
X, _, _ = self.dlinear_preprocessor.preprocess(df, fit=False)
if len(X) == 0:
return 0, 0.5
# 取最后一个窗口进行预测
X_last = torch.FloatTensor(X[-1:]).to(self.device)
with torch.no_grad():
logits = self.dlinear_model(X_last).cpu().numpy()
prob = 1 / (1 + np.exp(-logits))
prediction = int(prob.item() > 0.5)
probability = float(prob.item())
return prediction, probability
def predict_ttm(self, df: pd.DataFrame) -> Tuple[int, float]:
"""
使用 TTM-R2 预测下一小时的涨跌
输入: df 包含历史K线数据 (最后一行是最新的已完成K线)
输出: 预测下一小时相对于 df 最后一行收盘价的涨跌
Returns:
(prediction, probability): 预测结果(0=跌, 1=涨)和概率
"""
context_length = 512
all_prices = df['close'].values
# 使用最后 context_length 个价格
if len(all_prices) >= context_length:
context_data = all_prices[-context_length:].reshape(1, -1, 1)
else:
# 数据不足,用最后一个价格填充
padded = np.full(context_length, all_prices[-1])
padded[-len(all_prices):] = all_prices
context_data = padded.reshape(1, -1, 1)
# 基准价格: 最后一行的收盘价 (即预测目标小时的前一小时收盘价)
base_price = float(df['close'].iloc[-1])
context_tensor = torch.FloatTensor(context_data).to(self.device)
with torch.no_grad():
output = self.ttm_model(context_tensor)
pred_price = output.prediction_outputs.cpu().numpy()[0, 0, 0]
# 判断涨跌: 预测价格 vs 基准价格
prediction = 1 if pred_price > base_price else 0
# 计算伪概率: 基于价格变化百分比
price_pct = (pred_price - base_price) / base_price
# 将价格变化百分比映射到概率 (sigmoid)
probability = 1 / (1 + np.exp(-price_pct * 100))
return prediction, float(probability)
def predict_current_hour(self) -> Dict:
"""
预测当前小时的涨跌
逻辑:
- 获取最新的K线数据 (最后一行是最新已完成的K线)
- 预测下一小时相对于最后一行收盘价的涨跌
Returns:
预测结果字典
"""
df = self._get_btc_data()
# 获取当前时间 (美东)
now_et = datetime.now(self.et_tz)
current_date = now_et.strftime('%Y-%m-%d')
current_hour = now_et.hour
# 最后一个已完成的K线信息 (作为预测基准)
last_timestamp = df['timestamp'].iloc[-1]
last_close = float(df['close'].iloc[-1])
# 当前正在形成的K线信息
current_open = float(df['open'].iloc[-1])
current_price = float(df['close'].iloc[-1]) # 当前价格
results = {
"date": current_date,
"hour": current_hour,
"timestamp": str(last_timestamp),
"base_price": last_close, # 预测基准价格
"current_price": current_price,
"models": {}
}
# DLinear 预测 (预测下一小时相对于 last_close 的涨跌)
dlinear_pred, dlinear_prob = self.predict_dlinear(df)
results["models"]["DLinear"] = {
"prediction": dlinear_pred,
"probability": round(dlinear_prob, 4),
"direction": "UP" if dlinear_pred == 1 else "DOWN"
}
# TTM-R2 预测
ttm_pred, ttm_prob = self.predict_ttm(df)
results["models"]["TTM-R2"] = {
"prediction": ttm_pred,
"probability": round(ttm_prob, 4),
"direction": "UP" if ttm_pred == 1 else "DOWN"
}
return results
def ensure_predictions_for_date(self, date: str) -> None:
"""
确保指定日期有预测数据,如果没有则生成
Args:
date: 日期 (YYYY-MM-DD)
"""
# 使用锁避免并发冲突
with PredictionService._prediction_lock:
# 检查是否正在处理该日期
if date in self._processing_dates:
return
# 标记正在处理
self._processing_dates.add(date)
try:
# 检查是否有缺失的小时
for model_name in ["DLinear", "TTM-R2"]:
missing_hours = PredictionModel.get_missing_hours(date, model_name)
if not missing_hours:
continue
print(f"为 {date} 生成 {model_name} 预测 (缺失 {len(missing_hours)} 小时)...")
# 获取足够的历史数据
target_date = datetime.strptime(date, '%Y-%m-%d')
df = self._get_historical_data(target_date)
if df is None or len(df) < 200:
print(f" 警告: 数据不足,无法为 {date} 生成预测")
continue
# 为每个缺失的小时生成预测
for hour in missing_hours:
self._generate_prediction_for_hour(df, date, hour, model_name)
finally:
# 移除处理标记
with PredictionService._prediction_lock:
self._processing_dates.discard(date)
def _get_historical_data(self, target_date: datetime) -> Optional[pd.DataFrame]:
"""获取指定日期的历史数据"""
try:
start_date = (target_date - timedelta(days=30)).strftime('%Y-%m-%d')
end_date = (target_date + timedelta(days=2)).strftime('%Y-%m-%d')
df = self.fetcher.fetch_ohlcv(
symbol='BTC',
timeframe='1h',
start_date=start_date,
end_date=end_date,
use_cache=True
)
return df
except Exception as e:
print(f"获取历史数据失败: {e}")
return None
def _generate_prediction_for_hour(self, df: pd.DataFrame, date: str,
hour: int, model_name: str) -> None:
"""
为指定小时生成预测
预测逻辑:
- 预测目标: {date} {hour}:00 的K线收盘价涨跌
- 输入数据: {hour-1}:00 及以前的K线数据 (不包含目标小时)
- 涨跌判断: 预测价格 vs 前一小时收盘价 (即 {hour-1}:00 的 close)
- 正确率验证: 目标小时的 close vs 前一小时的 close
Args:
df: 历史K线数据 (UTC时间戳)
date: 目标日期 (美东时间 YYYY-MM-DD)
hour: 目标小时 (美东时间 0-23)
model_name: 模型名称
"""
try:
# 1. 构建目标时间 (美东时间 -> UTC)
target_time_et = pd.Timestamp(f"{date} {hour:02d}:00:00", tz='America/New_York')
target_time_utc = target_time_et.tz_convert('UTC').tz_localize(None)
# 2. 确保 timestamp 列是 datetime 类型
if 'timestamp_dt' not in df.columns:
df = df.copy()
df['timestamp_dt'] = pd.to_datetime(df['timestamp'])
# 3. 找到目标小时的K线 (用于验证)
target_mask = df['timestamp_dt'] == target_time_utc
if not target_mask.any():
# 目标小时数据不存在,跳过
return
target_row = df[target_mask].iloc[0]
target_open = float(target_row['open'])
target_close = float(target_row['close'])
# 4. 找到前一小时的K线 (用于预测输入和基准价格)
prev_time_utc = target_time_utc - pd.Timedelta(hours=1)
prev_mask = df['timestamp_dt'] == prev_time_utc
if not prev_mask.any():
# 前一小时数据不存在,跳过
return
prev_row = df[prev_mask].iloc[0]
prev_close = float(prev_row['close'])
# 5. 获取预测输入数据: 前一小时及以前的所有数据
input_mask = df['timestamp_dt'] <= prev_time_utc
df_input = df[input_mask].copy()
if len(df_input) < 200:
# 数据不足,跳过
return
# 6. 调用模型预测
if model_name == "DLinear":
prediction, probability = self.predict_dlinear(df_input)
else: # TTM-R2
prediction, probability = self.predict_ttm(df_input)
# 7. 判断实际涨跌 (目标小时收盘价 vs 前一小时收盘价)
actual_result = 1 if target_close > prev_close else 0
# 8. 判断预测是否正确
is_correct = 1 if prediction == actual_result else 0
# 9. 保存预测结果
prediction_data = {
"timestamp": target_time_utc.isoformat(),
"timestamp_et": target_time_et.isoformat(),
"date": date,
"hour": hour,
"model_name": model_name,
"prediction": prediction,
"probability": probability,
"actual_result": actual_result,
"is_correct": is_correct,
"open_price": target_open,
"close_price": target_close
}
PredictionModel.save_prediction(prediction_data)
except Exception as e:
print(f"生成预测失败 ({date} {hour}:00 {model_name}): {e}")
def ensure_predictions_for_month(self, month: str) -> None:
"""
确保指定月份有预测数据
Args:
month: 月份 (YYYY-MM)
"""
# 使用锁避免并发冲突
with PredictionService._prediction_lock:
# 检查是否正在处理该月份
if month in self._processing_months:
return
# 标记正在处理
self._processing_months.add(month)
try:
# 解析月份
year, mon = map(int, month.split('-'))
# 计算该月的日期范围
from calendar import monthrange
_, last_day = monthrange(year, mon)
# 对于过去的日期,确保有预测数据
today = datetime.now(self.et_tz).date()
for day in range(1, last_day + 1):
date = f"{year}-{mon:02d}-{day:02d}"
date_obj = datetime.strptime(date, '%Y-%m-%d').date()
# 只处理过去和今天的日期
if date_obj <= today:
self.ensure_predictions_for_date(date)
finally:
# 移除处理标记
with PredictionService._prediction_lock:
self._processing_months.discard(month)
# 全局单例
_prediction_service: Optional[PredictionService] = None
def get_prediction_service() -> PredictionService:
"""获取预测服务单例"""
global _prediction_service
if _prediction_service is None:
_prediction_service = PredictionService()
return _prediction_service
+252
View File
@@ -0,0 +1,252 @@
/**
* BTC 预测交易看板 - Vue 3 应用
*/
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: this.getCurrentMonth()
},
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, timestamp_cst) {
// 优先使用后端返回的 CST 时间
if (timestamp_cst) {
return timestamp_cst;
}
if (!timestamp) return '-';
const date = new Date(timestamp);
return date.toLocaleString('zh-CN', {
month: '2-digit',
day: '2-digit',
hour: '2-digit',
minute: '2-digit'
});
},
formatPercent(value) {
if (value === null || value === undefined) return '-';
return `${(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';
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';
},
// ==================== API 调用 ====================
async fetchAPI(url) {
try {
const response = await fetch(url);
const data = await response.json();
if (!data.success) {
console.error('API Error:', data);
}
return data;
} catch (error) {
console.error('Fetch Error:', error);
return { success: false, error: error.message };
}
},
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');
+414
View File
@@ -0,0 +1,414 @@
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>BTC 预测交易看板</title>
<!-- Bootstrap 5 CSS -->
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.2/dist/css/bootstrap.min.css" rel="stylesheet">
<!-- Bootstrap Icons -->
<link href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.11.1/font/bootstrap-icons.css" rel="stylesheet">
<!-- Custom CSS -->
<link href="/static/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 预测交易看板
</a>
<span class="navbar-text text-light">
<i class="bi bi-clock"></i> {{ serverTime }}
</span>
</div>
</nav>
<!-- 主容器 -->
<div class="container mt-4">
<!-- Tab 导航 -->
<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>
<!-- 交易详情 Tab -->
<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">${{ currentTrade.price?.toFixed(3) || '-' }}</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>
</tr>
</thead>
<tbody>
<tr v-for="trade in tradeHistory.data" :key="trade.id">
<td>{{ formatTime(trade.timestamp, trade.timestamp_cst) }}</td>
<td class="text-truncate" style="max-width: 200px;">
{{ trade.market_title }}
</td>
<td :class="getPredictionClass(trade.prediction)">
<i :class="trade.prediction === 'UP' ? 'bi-arrow-up' : 'bi-arrow-down'"></i>
{{ trade.prediction }}
</td>
<td>{{ formatPercent(trade.probability) }}</td>
<td>${{ trade.price?.toFixed(3) }}</td>
<td :class="getStatusClass(trade.status)">
{{ getStatusText(trade.status) }}
</td>
</tr>
<tr v-if="!tradeHistory.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">共 {{ 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>
<!-- 模型准确率 Tab -->
<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">
<!-- DLinear 卡片 -->
<div class="col-md-6">
<div class="card h-100 border-primary">
<div class="card-header bg-primary text-white">
<i class="bi bi-graph-up-arrow"></i> DLinear 模型
</div>
<div class="card-body">
<div class="row text-center" v-if="monthlyStats.DLinear">
<div class="col-3">
<div class="stat-number">{{ monthlyStats.DLinear.total }}</div>
<div class="stat-label">预测次数</div>
</div>
<div class="col-3">
<div class="stat-number text-success">{{ monthlyStats.DLinear.correct }}</div>
<div class="stat-label">正确</div>
</div>
<div class="col-3">
<div class="stat-number text-danger">{{ monthlyStats.DLinear.incorrect }}</div>
<div class="stat-label">错误</div>
</div>
<div class="col-3">
<div class="stat-number text-primary">{{ monthlyStats.DLinear.accuracy }}%</div>
<div class="stat-label">准确率</div>
</div>
</div>
<div v-else class="text-center text-muted py-3">
暂无数据
</div>
</div>
</div>
</div>
<!-- TTM-R2 卡片 -->
<div class="col-md-6">
<div class="card h-100 border-success">
<div class="card-header bg-success text-white">
<i class="bi bi-cpu"></i> TTM-R2 模型
</div>
<div class="card-body">
<div class="row text-center" v-if="monthlyStats['TTM-R2']">
<div class="col-3">
<div class="stat-number">{{ monthlyStats['TTM-R2'].total }}</div>
<div class="stat-label">预测次数</div>
</div>
<div class="col-3">
<div class="stat-number text-success">{{ monthlyStats['TTM-R2'].correct }}</div>
<div class="stat-label">正确</div>
</div>
<div class="col-3">
<div class="stat-number text-danger">{{ monthlyStats['TTM-R2'].incorrect }}</div>
<div class="stat-label">错误</div>
</div>
<div class="col-3">
<div class="stat-number text-primary">{{ monthlyStats['TTM-R2'].accuracy }}%</div>
<div class="stat-label">准确率</div>
</div>
</div>
<div v-else class="text-center text-muted py-3">
暂无数据
</div>
</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">
<thead class="table-light">
<tr>
<th>日期 (北京)</th>
<th>DLinear 预测</th>
<th>DLinear 准确率</th>
<th>TTM-R2 预测</th>
<th>TTM-R2 准确率</th>
<th>操作</th>
</tr>
</thead>
<tbody>
<tr v-for="day in dailyStats.data" :key="day.date">
<td>{{ day.date }}</td>
<td>
<span v-if="day.DLinear">
{{ day.DLinear.correct }}/{{ day.DLinear.total }}
</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['TTM-R2']">
{{ day['TTM-R2'].correct }}/{{ day['TTM-R2'].total }}
</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-lg 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">
<thead class="table-light sticky-top">
<tr>
<th>时间 (北京)</th>
<th>实际涨跌</th>
<th>DLinear</th>
<th>TTM-R2</th>
</tr>
</thead>
<tbody>
<tr v-for="hour in hourlyModal.data" :key="hour.hour">
<td>{{ hour.timestamp_cst || (hour.hour + ':00') }}</td>
<td :class="hour.actual_result === 1 ? 'text-success' : 'text-danger'">
<i :class="hour.actual_result === 1 ? 'bi-arrow-up' : 'bi-arrow-down'"></i>
{{ hour.actual_result === 1 ? 'UP' : 'DOWN' }}
</td>
<td>
<span v-if="hour.models?.DLinear"
:class="hour.models.DLinear.is_correct ? 'text-success' : 'text-danger'">
<i :class="hour.models.DLinear.prediction === 1 ? 'bi-arrow-up' : 'bi-arrow-down'"></i>
{{ hour.models.DLinear.prediction === 1 ? 'UP' : 'DOWN' }}
<i :class="hour.models.DLinear.is_correct ? 'bi-check-circle' : 'bi-x-circle'"></i>
</span>
<span v-else class="text-muted">-</span>
</td>
<td>
<span v-if="hour.models?.['TTM-R2']"
:class="hour.models['TTM-R2'].is_correct ? 'text-success' : 'text-danger'">
<i :class="hour.models['TTM-R2'].prediction === 1 ? 'bi-arrow-up' : 'bi-arrow-down'"></i>
{{ hour.models['TTM-R2'].prediction === 1 ? 'UP' : 'DOWN' }}
<i :class="hour.models['TTM-R2'].is_correct ? 'bi-check-circle' : 'bi-x-circle'"></i>
</span>
<span v-else class="text-muted">-</span>
</td>
</tr>
<tr v-if="!hourlyModal.data?.length">
<td colspan="4" 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>
<!-- Loading 遮罩 -->
<div v-if="loading" class="loading-overlay">
<div class="spinner-border text-primary" role="status">
<span class="visually-hidden">加载中...</span>
</div>
</div>
</div>
<!-- Bootstrap 5 JS -->
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.2/dist/js/bootstrap.bundle.min.js"></script>
<!-- Vue 3 -->
<script src="https://cdn.jsdelivr.net/npm/vue@3.4.15/dist/vue.global.prod.js"></script>
<!-- App JS -->
<script src="/static/app.js"></script>
</body>
</html>
+202
View File
@@ -0,0 +1,202 @@
/**
* BTC 预测交易看板 - 自定义样式
*/
/* 全局样式 */
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;
}
/* Tab 样式 */
.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;
}
.stat-number {
font-size: 1.75rem;
font-weight: 700;
}
/* 表格样式 */
.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);
}
/* 涨跌颜色 */
.text-success {
color: #198754 !important;
}
.text-danger {
color: #dc3545 !important;
}
/* 分页 */
.pagination {
margin-bottom: 0;
}
.page-link {
color: #0d6efd;
}
.page-item.disabled .page-link {
color: #6c757d;
}
/* Loading 遮罩 */
.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;
}
}
/* 准确率颜色 */
.accuracy-high {
color: #198754;
font-weight: 600;
}
.accuracy-medium {
color: #fd7e14;
font-weight: 600;
}
.accuracy-low {
color: #dc3545;
font-weight: 600;
}
/* 边框颜色增强 */
.border-primary {
border-color: #0d6efd !important;
border-width: 2px !important;
}
.border-success {
border-color: #198754 !important;
border-width: 2px !important;
}
/* 按钮样式 */
.btn-outline-primary:hover {
color: #fff;
}
/* 截断文本 */
.text-truncate {
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
/* 间距调整 */
.gap-2 {
gap: 0.5rem !important;
}
+944
View File
@@ -0,0 +1,944 @@
"""
BTC Hourly 自动交易脚本
基于DLinear模型预测,在Polymarket自动下单BTC小时涨跌
支持:
- 每小时自动寻找对应市场
- 实时价格刷新
- 定时下单
- 详细日志记录
"""
import os
import sys
import argparse
import json
import time
import logging
from logging.handlers import RotatingFileHandler
import schedule
from datetime import datetime, timedelta
from typing import Optional, Tuple, Dict
import pickle
from zoneinfo import ZoneInfo
import torch
import numpy as np
import pandas as pd
import pandas_ta as ta
from sklearn.preprocessing import StandardScaler
from dotenv import load_dotenv
import httpx
# 添加项目路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from data.fetcher import BinanceDataFetcher
from data.trading_db import TradingDatabase
load_dotenv()
def setup_logging(log_dir: str = "logs", log_level: int = logging.INFO) -> logging.Logger:
"""
配置日志系统
Args:
log_dir: 日志目录
log_level: 日志级别
Returns:
配置好的logger
"""
# 创建日志目录
os.makedirs(log_dir, exist_ok=True)
# 创建logger
logger = logging.getLogger("polymarket_trader")
logger.setLevel(log_level)
# 避免重复添加handler
if logger.handlers:
return logger
# 日志格式
formatter = logging.Formatter(
"%(asctime)s | %(levelname)-8s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
# 文件handler (按大小轮转, 最大10MB, 保留5个备份)
log_file = os.path.join(log_dir, "polymarket_trading.log")
file_handler = RotatingFileHandler(
log_file,
maxBytes=10*1024*1024, # 10MB
backupCount=5,
encoding='utf-8'
)
file_handler.setLevel(log_level)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
# 控制台handler
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(log_level)
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
return logger
# 初始化全局logger
logger = setup_logging()
class ClassificationPreprocessor:
"""DLinear分类预处理器"""
def __init__(self, lookback: int = 168, forecast_horizon: int = 1):
self.lookback = lookback
self.forecast_horizon = forecast_horizon
self.scaler = StandardScaler()
self.is_fitted = False
def add_features(self, df: pd.DataFrame) -> pd.DataFrame:
df = df.copy()
df['rsi'] = ta.rsi(df['close'], length=14)
macd = ta.macd(df['close'])
if macd is not None and not macd.empty:
df['macd'] = macd.iloc[:, 0]
df['macd_hist'] = macd.iloc[:, 1]
df['macd_signal'] = macd.iloc[:, 2]
return df
def create_labels(self, df: pd.DataFrame) -> pd.DataFrame:
df = df.copy()
df['future_price'] = df['close'].shift(-self.forecast_horizon)
df['future_return'] = (df['future_price'] - df['close']) / df['close']
df['label'] = (df['future_return'] > 0).astype(int)
return df[:-self.forecast_horizon]
def preprocess(self, df: pd.DataFrame, fit: bool = True):
df = self.add_features(df)
df = self.create_labels(df)
df = df.dropna().reset_index(drop=True)
feature_cols = ['open', 'high', 'low', 'close', 'volume',
'rsi', 'macd', 'macd_hist', 'macd_signal']
if fit:
scaled_features = self.scaler.fit_transform(df[feature_cols])
self.is_fitted = True
self.feature_columns = feature_cols
else:
scaled_features = self.scaler.transform(df[feature_cols])
X, y = [], []
for i in range(len(scaled_features) - self.lookback):
X.append(scaled_features[i:i + self.lookback])
y.append(df['label'].iloc[i + self.lookback])
return np.array(X), np.array(y), df
class PolymarketBTCTrader:
"""Polymarket BTC Hourly交易器"""
GAMMA_API = "https://gamma-api.polymarket.com"
CLOB_API = "https://clob.polymarket.com"
# 月份名称映射
MONTH_NAMES = {
1: 'january', 2: 'february', 3: 'march', 4: 'april',
5: 'may', 6: 'june', 7: 'july', 8: 'august',
9: 'september', 10: 'october', 11: 'november', 12: 'december'
}
def __init__(
self,
private_key: Optional[str] = None,
max_bet: float = 1.0,
max_price: float = 0.50,
dry_run: bool = True,
):
self.private_key = private_key or os.getenv("POLYMARKET_PRIVATE_KEY")
self.max_bet = max_bet
self.max_price = max_price
self.dry_run = dry_run
# HTTP客户端
proxy = os.getenv("SOCKS5_PROXY") if os.getenv("USE_PROXY", "false").lower() == "true" else None
self.http = httpx.Client(
timeout=30.0,
proxy=proxy,
headers={'Accept-Encoding': 'identity'}
)
# 设备选择:macOS 用 MPS,Windows/Linux 用 CPU
import platform
if platform.system() == 'Darwin': # macOS
self.device = torch.device('mps' if torch.backends.mps.is_available() else 'cpu')
else: # Windows / Linux
self.device = torch.device('cpu')
print(f"使用设备: {self.device}")
# 加载模型
self.model = None
self.preprocessor = None
self._load_model()
# 数据获取器
self.fetcher = BinanceDataFetcher()
# 数据库
self.db = TradingDatabase()
# CLOB客户端(用于下单)
self.clob_client = None
if self.private_key and not self.dry_run:
self._init_clob_client()
# 当前市场缓存
self.current_market = None
self.last_price_update = None
def _load_model(self):
"""加载DLinear模型"""
logger.info("加载DLinear模型...")
from models.dlinear.model import DLinear
checkpoint_path = './models/checkpoints/dlinear_classification_2026/best_model.pt'
preprocessor_path = './models/checkpoints/dlinear_classification_2026/preprocessor.pkl'
checkpoint = torch.load(checkpoint_path, map_location=self.device, weights_only=False)
self.model = DLinear(checkpoint['config'])
self.model.load_state_dict(checkpoint['model_state_dict'])
self.model = self.model.to(self.device).eval()
with open(preprocessor_path, 'rb') as f:
self.preprocessor = pickle.load(f)
logger.info(f"✓ 模型已加载 (设备: {self.device})")
def _init_clob_client(self):
"""初始化CLOB交易客户端"""
try:
from py_clob_client.client import ClobClient
self.clob_client = ClobClient(
host=self.CLOB_API,
key=self.private_key,
chain_id=137, # Polygon
)
self.clob_client.set_api_creds(self.clob_client.create_or_derive_api_creds())
logger.info("✓ CLOB客户端初始化成功")
except Exception as e:
logger.warning(f"⚠️ CLOB客户端初始化失败: {e}")
self.clob_client = None
def _log_to_db(self, result: Dict, market: Optional[Dict]):
"""记录交易结果到数据库"""
try:
# 确定状态
status = 'failure'
if self.dry_run:
status = 'test'
elif result.get('success'):
status = 'success'
elif result.get('error'):
status = 'failure'
elif result.get('prediction'): # 有预测但未交易(如市场未找到)
status = 'in_progress' # 或者 failure? 未找到市场算 failure
if result.get('error'):
status = 'failure'
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': 'DLinear',
'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'),
'raw_data': result
}
self.db.log_trade(log_data)
logger.info(f"💾 交易记录已保存到数据库")
except Exception as e:
logger.error(f"❌ 保存数据库失败: {e}")
def generate_market_slug(self, target_time: Optional[datetime] = None) -> str:
"""
生成市场slug
格式: bitcoin-up-or-down-{month}-{day}-{hour}am/pm-et
例如: bitcoin-up-or-down-february-3-8am-et
Args:
target_time: 目标时间(ET时区)
for_current_hour: 如果True,使用当前整点;否则使用下一个整点
Returns:
市场slug
"""
# 获取美东时间
et_tz = ZoneInfo('America/New_York')
if target_time is None:
now_et = datetime.now(et_tz)
target_time = now_et.replace(minute=0, second=0, microsecond=0)
month_name = self.MONTH_NAMES[target_time.month]
day = target_time.day
# 转换为12小时制
hour_24 = target_time.hour
if hour_24 == 0:
hour_12 = 12
am_pm = 'am'
elif hour_24 < 12:
hour_12 = hour_24
am_pm = 'am'
elif hour_24 == 12:
hour_12 = 12
am_pm = 'pm'
else:
hour_12 = hour_24 - 12
am_pm = 'pm'
slug = f"bitcoin-up-or-down-{month_name}-{day}-{hour_12}{am_pm}-et"
return slug
def get_market_by_slug(self, slug: str) -> Optional[Dict]:
"""
通过slug获取市场信息
Args:
slug: 市场slug
Returns:
市场信息
"""
try:
response = self.http.get(
f"{self.GAMMA_API}/events",
params={'slug': slug}
)
response.raise_for_status()
events = response.json()
if not events:
return None
event = events[0]
markets = event.get('markets', [])
if not markets:
return None
market = markets[0]
# outcomePrices可能是JSON字符串或列表
outcome_prices_raw = market.get('outcomePrices', ['0', '0'])
if isinstance(outcome_prices_raw, str):
import json as _json
try:
outcome_prices_raw = _json.loads(outcome_prices_raw)
except:
outcome_prices_raw = ['0', '0']
# clobTokenIds也可能是JSON字符串
clob_tokens_raw = market.get('clobTokenIds', [])
if isinstance(clob_tokens_raw, str):
import json as _json
try:
clob_tokens_raw = _json.loads(clob_tokens_raw)
except:
clob_tokens_raw = []
return {
'title': event.get('title', ''),
'slug': slug,
'question': market.get('question', ''),
'condition_id': market.get('conditionId', ''),
'outcomes': market.get('outcomes', []),
'outcome_prices': [float(p) for p in outcome_prices_raw],
'tokens': clob_tokens_raw,
'end_date': event.get('endDate', ''),
'closed': market.get('closed', False),
}
except Exception as e:
logger.error(f"获取市场失败: {e}")
return None
def get_current_hourly_market(self) -> Optional[Dict]:
"""
获取BTC市场
Args:
for_current_hour: 如果True,获取当前整点市场;否则获取下一个整点市场
Returns:
市场信息
"""
slug = self.generate_market_slug()
logger.info(f"搜索市场: {slug}")
market = self.get_market_by_slug(slug)
if market:
self.current_market = market
self.last_price_update = datetime.now()
logger.info(f"✓ 找到市场: {market['title']}")
logger.info(f" Up价格: ${market['outcome_prices'][0]:.3f}")
logger.info(f" Down价格: ${market['outcome_prices'][1]:.3f}")
else:
logger.warning(f"❌ 未找到市场: {slug}")
return market
def refresh_prices(self) -> Optional[Dict]:
"""
刷新当前市场价格
Returns:
更新后的市场信息
"""
if not self.current_market:
self.get_current_hourly_market()
# 使用CLOB API获取实时价格
tokens = self.current_market.get('tokens', [])
if len(tokens) >= 2:
realtime_prices = self.get_realtime_prices(tokens[0], tokens[1])
if realtime_prices:
old_prices = self.current_market['outcome_prices']
new_prices = realtime_prices
self.current_market['outcome_prices'] = new_prices
self.last_price_update = datetime.now()
# 显示价格变化
up_change = new_prices[0] - old_prices[0]
down_change = new_prices[1] - old_prices[1]
logger.info(f"📊 实时价格 [{datetime.now().strftime('%H:%M:%S')}]")
logger.info(f" Up: ${new_prices[0]:.3f} ({up_change:+.3f})")
logger.info(f" Down: ${new_prices[1]:.3f} ({down_change:+.3f})")
return self.current_market
# 备用: 使用Gamma API
slug = self.current_market['slug']
market = self.get_market_by_slug(slug)
if market:
old_prices = self.current_market['outcome_prices']
new_prices = market['outcome_prices']
self.current_market = market
self.last_price_update = datetime.now()
# 显示价格变化
up_change = new_prices[0] - old_prices[0]
down_change = new_prices[1] - old_prices[1]
logger.info(f"📊 价格更新(Gamma) [{datetime.now().strftime('%H:%M:%S')}]")
logger.info(f" Up: ${new_prices[0]:.3f} ({up_change:+.3f})")
logger.info(f" Down: ${new_prices[1]:.3f} ({down_change:+.3f})")
return market
def get_realtime_prices(self, up_token_id: str, down_token_id: str) -> Optional[list]:
"""
从CLOB API获取实时买入价格
使用/price?side=BUY端点获取真实买入价格 (与页面显示一致)
Args:
up_token_id: Up代币ID
down_token_id: Down代币ID
Returns:
[up_price, down_price] 或 None
"""
try:
# 使用/price端点获取买入价格
up_response = self.http.get(
f"{self.CLOB_API}/price",
params={'token_id': up_token_id, 'side': 'BUY'}
)
up_response.raise_for_status()
up_data = up_response.json()
down_response = self.http.get(
f"{self.CLOB_API}/price",
params={'token_id': down_token_id, 'side': 'BUY'}
)
down_response.raise_for_status()
down_data = down_response.json()
# 解析买入价格
up_price = float(up_data.get('price', 0.5))
down_price = float(down_data.get('price', 0.5))
return [up_price, down_price]
except Exception as e:
logger.debug(f"获取实时价格失败: {e}")
return None
def execute_trade(self, prediction: int, market: Dict) -> Dict:
"""
执行交易
Args:
prediction: 预测结果(0=跌, 1=涨)
market: 市场信息
Returns:
交易结果
"""
result = {
'timestamp': datetime.now().isoformat(),
'market': market['title'],
'prediction': 'UP' if prediction == 1 else 'DOWN',
'success': False,
'order': None,
'error': None,
}
if prediction == 1: # 预测涨 -> 买UP
token_id = market['tokens'][0]
price = market['outcome_prices'][0]
side = "UP"
else: # 预测跌 -> 买DOWN
token_id = market['tokens'][1]
price = market['outcome_prices'][1]
side = "DOWN"
result['side'] = side
result['price'] = price
# 检查价格限制
if price > self.max_price:
result['error'] = f"价格${price:.2f}超过限制${self.max_price:.2f}"
logger.error(f"❌ {result['error']}")
return result
# 计算数量
size = self.max_bet / price
result['size'] = size
result['amount'] = self.max_bet
logger.info("📋 交易详情:")
logger.info(f" 方向: {side}")
logger.info(f" 价格: ${price:.3f}")
logger.info(f" 金额: ${self.max_bet:.2f} USDC")
logger.info(f" 数量: {size:.2f} 股")
if self.dry_run:
logger.info("🔸 [模拟模式] 未实际下单")
result['success'] = True
result['order'] = 'DRY_RUN'
return result
# 实际下单
if not self.clob_client:
result['error'] = "CLOB客户端未初始化"
logger.error(f"❌ {result['error']}")
return result
try:
from py_clob_client.order_builder.constants import BUY
order = self.clob_client.create_limit_order(
token_id=token_id,
price=price,
size=size,
side=BUY,
)
response = self.clob_client.post_order(order)
logger.info(f"✓ 订单已提交: {response}")
result['success'] = True
result['order'] = response
except Exception as e:
result['error'] = str(e)
logger.error(f"❌ 下单失败: {e}")
return result
def get_prediction_with_delay(self, delay_seconds: float = 2.0) -> Tuple[int, float]:
"""
延迟后获取K线并预测当前小时的涨跌
注意: 使用上一小时收盘的K线(倒数第二根)来预测当前小时
例如: 22:00整点时,用21:00收盘的K线预测22:00这个小时的涨跌
Args:
delay_seconds: 延迟秒数(等待K线更新)
Returns:
(prediction, probability): 预测结果(0=跌, 1=涨)和概率
"""
logger.info(f"⏳ 等待{delay_seconds}秒获取最新K线...")
time.sleep(delay_seconds)
logger.info("获取最新BTC数据...")
end_date = (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d')
start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')
df = self.fetcher.fetch_ohlcv(
symbol='BTC',
timeframe='1h',
start_date=start_date,
end_date=end_date,
use_cache=False # 不使用缓存,获取最新数据
)
# 使用倒数第二根K线(上一小时收盘)进行预测
# 例如22:00时,最新K线是14:00 UTC(22:00北京),我们用13:00 UTC(21:00北京)的数据
prev_time = df['timestamp'].iloc[-2]
prev_close = df['close'].iloc[-2]
current_time = df['timestamp'].iloc[-1]
current_close = df['close'].iloc[-1]
logger.info(f" 上一K线时间: {prev_time} (用于预测)")
logger.info(f" 上一收盘价: ${prev_close:,.2f}")
logger.info(f" 当前K线时间: {current_time} (预测目标)")
logger.info(f" 当前收盘价: ${current_close:,.2f} (实际结果)")
# 用除最后一根外的数据进行预测
df_for_prediction = df.iloc[:-1]
X, _, _ = self.preprocessor.preprocess(df_for_prediction, fit=False)
X_last = torch.FloatTensor(X[-1:]).to(self.device)
with torch.no_grad():
logits = self.model(X_last).cpu().numpy()
prob = 1 / (1 + np.exp(-logits))
prediction = int(prob.item() > 0.5)
probability = float(prob.item())
direction = "📈 UP (涨)" if prediction == 1 else "📉 DOWN (跌)"
confidence = probability if prediction == 1 else 1 - probability
logger.info(f"🤖 DLinear预测: {direction}")
logger.info(f" 置信度: {confidence:.2%}")
# 显示预测是否正确 (用于验证)
actual_up = current_close > prev_close
if prediction == 1:
correct = actual_up
else:
correct = not actual_up
logger.info(f" 实际涨跌: {'📈 涨' if actual_up else '📉 跌'} ({'✅ 正确' if correct else '❌ 错误'})")
return prediction, probability
def run_once(self) -> Dict:
"""运行一次交易流程(不等待整点)"""
logger.info("运行单次交易...")
return self.run_at_hour_start(delay_seconds=0)
def run_at_hour_start(self, delay_seconds: float = 2.0, timeout_minutes: int = 50) -> Dict:
"""
整点时执行的交易流程
1. 延迟delay_seconds秒等待K线更新
2. 获取最新K线
3. DLinear预测当前整点的收盘涨跌
4. 获取对应的Polymarket市场
5. 循环检查价格:
- 如果价格 <= max_price: 下单并结束
- 如果价格 > max_price: 等待并重试(直到超时)
Args:
delay_seconds: 整点后延迟秒数
timeout_minutes: 价格等待超时时间(分钟)
Returns:
执行结果
"""
logger.info("=" * 60)
logger.info(f"⏰ 整点交易触发 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
logger.info("=" * 60)
logger.info(f" 模式: {'🔸 模拟' if self.dry_run else '💰 真实交易'}")
logger.info(f" 最大下注: ${self.max_bet:.2f} USDC")
logger.info(f" 最高价格: ${self.max_price:.2f}")
logger.info(f" 等待超时: {timeout_minutes}分钟")
result = {
'timestamp': datetime.now().isoformat(),
'success': False,
'error': None,
}
# 1. 延迟后获取预测
prediction, probability = self.get_prediction_with_delay(delay_seconds)
result['prediction'] = 'UP' if prediction == 1 else 'DOWN'
result['probability'] = probability if prediction == 1 else 1 - probability
# 2. 获取初始市场
market = self.get_current_hourly_market()
if not market:
result['error'] = '未找到市场'
logger.error(f"❌ {result['error']}")
self._log_to_db(result, None)
return result
result['market'] = market['title']
result['market_slug'] = market['slug']
if market['closed']:
result['error'] = '市场已关闭'
logger.error(f"❌ {result['error']}")
self._log_to_db(result, market)
return result
# 检查是否已交易过该市场
check_status = 'test' if self.dry_run else 'success'
if self.db.check_existing_trade(market['slug'], status=check_status):
result['error'] = '该市场已存在成功交易记录'
logger.warning(f"⚠️ {result['error']}, 跳过交易")
return result
# 3. 循环检查价格并交易
start_time = datetime.now()
end_time = start_time + timedelta(minutes=timeout_minutes)
logger.info(f"🔄 开始价格监控 (预测: {'UP' if prediction == 1 else 'DOWN'})...")
while datetime.now() < end_time:
# 刷新价格
market = self.refresh_prices()
if not market:
logger.warning("无法获取市场价格,重试中...")
time.sleep(10)
continue
# 检查价格
price_idx = 0 if prediction == 1 else 1
current_price = market['outcome_prices'][price_idx]
logger.info(f" 当前价格: ${current_price:.3f} (目标: <= ${self.max_price:.2f})")
if current_price <= self.max_price:
# 价格符合,执行交易
logger.info("✅ 价格满足条件,正在下单...")
trade_result = self.execute_trade(prediction, market)
result.update(trade_result)
# 记录数据库
self._log_to_db(result, market)
logger.info("=" * 60)
return result
else:
# 价格过高,等待
wait_time = 30 # 等待30秒
elapsed = int((datetime.now() - start_time).total_seconds())
logger.info(f"⏳ 价格过高,继续等待... (已耗时: {elapsed}s)")
time.sleep(wait_time)
# 超时
result['error'] = f'等待价格超时 ({timeout_minutes}分钟)'
result['price'] = market['outcome_prices'][0] if prediction == 1 else market['outcome_prices'][1]
result['side'] = 'UP' if prediction == 1 else 'DOWN'
logger.error(f"❌ {result['error']}")
self._log_to_db(result, market)
logger.info("=" * 60)
return result
def _show_initial_prediction(self):
"""
启动时显示当前预测(不下单)
"""
logger.info("\n📊 当前预测状态:")
end_date = (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d')
start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')
df = self.fetcher.fetch_ohlcv(
symbol='BTC',
timeframe='1h',
start_date=start_date,
end_date=end_date,
use_cache=False
)
# 显示K线信息
prev_time = df['timestamp'].iloc[-2]
prev_close = df['close'].iloc[-2]
current_time = df['timestamp'].iloc[-1]
current_close = df['close'].iloc[-1]
logger.info(f" 上一K线: {prev_time} 收盘价: ${prev_close:,.2f}")
logger.info(f" 当前K线: {current_time} 收盘价: ${current_close:,.2f}")
# 进行预测
df_for_prediction = df.iloc[:-1]
X, _, _ = self.preprocessor.preprocess(df_for_prediction, fit=False)
X_last = torch.FloatTensor(X[-1:]).to(self.device)
with torch.no_grad():
logits = self.model(X_last).cpu().numpy()
prob = 1 / (1 + np.exp(-logits))
prediction = int(prob.item() > 0.5)
probability = float(prob.item())
direction = "📈 UP (涨)" if prediction == 1 else "📉 DOWN (跌)"
confidence = probability if prediction == 1 else 1 - probability
logger.info(f" 🤖 预测: {direction} (置信度: {confidence:.2%})")
# 显示市场价格
if self.current_market:
up_price = self.current_market['outcome_prices'][0]
down_price = self.current_market['outcome_prices'][1]
if prediction == 1:
logger.info(f" 💰 买入Up价格: ${up_price:.3f} {'✅ 可下单' if up_price <= self.max_price else '❌ 超过限价'}")
else:
logger.info(f" 💰 买入Down价格: ${down_price:.3f} {'✅ 可下单' if down_price <= self.max_price else '❌ 超过限价'}")
# 计算距下一整点时间
now = datetime.now()
next_hour = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)
wait_minutes = int((next_hour - now).total_seconds() / 60)
logger.info(f" ⏰ 距离下一整点: {wait_minutes}分钟")
logger.info("")
def wait_for_next_hour(self) -> datetime:
"""
等待到下一个整点
Returns:
下一个整点时间
"""
now = datetime.now()
next_hour = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)
wait_seconds = (next_hour - now).total_seconds()
logger.info(f"⏳ 等待下一个整点: {next_hour.strftime('%H:%M:%S')}")
logger.info(f" 剩余等待: {int(wait_seconds)}秒")
return next_hour
def run_scheduler(self, interval_minutes: int = 5, delay_seconds: float = 2.0):
"""
定时运行交易器
工作流程:
1. 每个整点+delay_seconds秒触发交易
2. 期间每interval_minutes分钟刷新价格
Args:
interval_minutes: 价格刷新间隔(分钟)
delay_seconds: 整点后延迟秒数
"""
logger.info("=" * 60)
logger.info("🤖 BTC Hourly自动交易已启动")
logger.info("=" * 60)
logger.info(f" 模式: {'🔸 模拟' if self.dry_run else '💰 真实交易'}")
logger.info(f" 下单时机: 每个整点后延迟{delay_seconds}秒")
logger.info(f" 价格刷新: 每{interval_minutes}分钟")
logger.info(f" 最大下注: ${self.max_bet:.2f} USDC")
logger.info(f" 最高价格: ${self.max_price:.2f}")
logger.info(" 按 Ctrl+C 停止")
logger.info("=" * 60)
# 立即获取整点的市场
self.get_current_hourly_market()
# 启动时进行一次预测(不下单,仅显示)
self._show_initial_prediction()
# 每小时整点后执行交易
def hourly_trade():
self.run_at_hour_start(delay_seconds=delay_seconds)
schedule.every().hour.at(":00").do(lambda: time.sleep(delay_seconds) or hourly_trade())
try:
while True:
schedule.run_pending()
time.sleep(1)
except KeyboardInterrupt:
logger.info("✓ 交易器已停止")
def close(self):
"""关闭资源"""
self.http.close()
def main():
parser = argparse.ArgumentParser(description='BTC Hourly自动交易 - 基于DLinear模型')
parser.add_argument('--dry-run', action='store_true', default=True,
help='模拟模式(不实际下单)')
parser.add_argument('--real', action='store_true',
help='真实交易模式')
parser.add_argument('--max-bet', type=float, default=1.0,
help='最大下注金额(USDC)')
parser.add_argument('--max-price', type=float, default=0.50,
help='最高可接受价格')
parser.add_argument('--once', action='store_true',
help='只执行一次(不启动定时器)')
parser.add_argument('--interval', type=int, default=1,
help='价格刷新间隔(分钟)')
parser.add_argument('--check-market', action='store_true',
help='只检查当前市场信息')
args = parser.parse_args()
dry_run = not args.real
# 真实交易确认
if not dry_run:
logger.warning("⚠️ 警告: 即将进行真实交易!")
logger.warning(f" 最大下注: ${args.max_bet:.2f} USDC")
confirm = input("确认继续? (yes/no): ")
if confirm.lower() != 'yes':
logger.info("已取消")
return
trader = PolymarketBTCTrader(
max_bet=args.max_bet,
max_price=args.max_price,
dry_run=dry_run,
)
try:
if args.check_market:
# 只检查市场
trader.get_current_hourly_market()
elif args.once:
# 执行一次
result = trader.run_once()
# 保存日志
log_file = 'trading_log.json'
logs = []
if os.path.exists(log_file):
with open(log_file, 'r') as f:
logs = json.load(f)
logs.append(result)
with open(log_file, 'w') as f:
json.dump(logs, f, indent=2, ensure_ascii=False)
logger.info(f"结果已保存到 {log_file}")
else:
# 定时运行
trader.run_scheduler(interval_minutes=args.interval)
finally:
trader.close()
if __name__ == "__main__":
main()
+157
View File
@@ -0,0 +1,157 @@
import pymysql
import pandas as pd
from typing import Optional
from contextlib import contextmanager
import os
from dotenv import load_dotenv
from sqlalchemy import create_engine, text
load_dotenv()
class DatabaseManager:
"""管理 OHLCV 数据的 MySQL 数据库"""
def __init__(self, db_path: str = None):
"""初始化数据库管理器 (db_path 参数已废弃,仅为兼容性保留)"""
self._engine = None
self.initialize_db()
def _get_sqlalchemy_engine(self):
"""获取 SQLAlchemy 引擎(用于 pandas read_sql)"""
if self._engine is None:
host = os.getenv('MYSQL_HOST')
port = os.getenv('MYSQL_PORT', 3306)
user = os.getenv('MYSQL_USER')
password = os.getenv('MYSQL_PASSWORD')
database = os.getenv('MYSQL_DB')
self._engine = create_engine(
f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}?charset=utf8mb4"
)
return self._engine
@contextmanager
def get_connection(self):
"""获取 MySQL 数据库连接"""
conn = pymysql.connect(
host=os.getenv('MYSQL_HOST'),
port=int(os.getenv('MYSQL_PORT', 3306)),
user=os.getenv('MYSQL_USER'),
password=os.getenv('MYSQL_PASSWORD'),
database=os.getenv('MYSQL_DB'),
charset='utf8mb4',
)
try:
yield conn
finally:
conn.close()
def initialize_db(self):
"""创建表(如果不存在)"""
with self.get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS ohlcv (
timestamp DATETIME,
symbol VARCHAR(20),
timeframe VARCHAR(10),
open DOUBLE,
high DOUBLE,
low DOUBLE,
close DOUBLE,
volume DOUBLE,
PRIMARY KEY (symbol, timeframe, timestamp),
INDEX idx_ohlcv_time (symbol, timeframe, timestamp)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
""")
conn.commit()
def save_ohlcv(self, df: pd.DataFrame, symbol: str, timeframe: str):
"""保存 OHLCV 数据到数据库 (Upsert)"""
if df.empty:
return
# 确保 timestamp 是 datetime 类型
df = df.copy()
if not pd.api.types.is_datetime64_any_dtype(df['timestamp']):
df['timestamp'] = pd.to_datetime(df['timestamp'])
# 转换为字符串格式
df['timestamp'] = df['timestamp'].dt.strftime('%Y-%m-%d %H:%M:%S')
# 添加元数据列
if 'symbol' not in df.columns:
df['symbol'] = symbol
if 'timeframe' not in df.columns:
df['timeframe'] = timeframe
cols = ['timestamp', 'symbol', 'timeframe', 'open', 'high', 'low', 'close', 'volume']
data = [tuple(row) for row in df[cols].values]
with self.get_connection() as conn:
cursor = conn.cursor()
sql = """
REPLACE INTO ohlcv
(timestamp, symbol, timeframe, open, high, low, close, volume)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
"""
cursor.executemany(sql, data)
conn.commit()
def get_ohlcv(self, symbol: str, timeframe: str, start_date: Optional[str] = None, end_date: Optional[str] = None) -> pd.DataFrame:
"""从数据库获取 OHLCV 数据"""
query = """
SELECT timestamp, open, high, low, close, volume
FROM ohlcv
WHERE symbol = :p0 AND timeframe = :p1
"""
params = {'p0': symbol, 'p1': timeframe}
param_idx = 2
if start_date:
query += f" AND timestamp >= :p{param_idx}"
params[f'p{param_idx}'] = start_date
param_idx += 1
if end_date:
# 处理 'YYYY-MM-DD' 格式
if len(str(end_date)) == 10:
query += f" AND timestamp < DATE_ADD(:p{param_idx}, INTERVAL 1 DAY)"
else:
query += f" AND timestamp <= :p{param_idx}"
params[f'p{param_idx}'] = end_date
query += " ORDER BY timestamp ASC"
engine = self._get_sqlalchemy_engine()
with engine.connect() as conn:
df = pd.read_sql_query(text(query), conn, params=params, parse_dates=['timestamp'])
return df
def get_latest_timestamp(self, symbol: str, timeframe: str) -> Optional[pd.Timestamp]:
"""获取指定 symbol/timeframe 的最新时间戳"""
with self.get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT MAX(timestamp)
FROM ohlcv
WHERE symbol = %s AND timeframe = %s
""", (symbol, timeframe))
result = cursor.fetchone()
return pd.to_datetime(result[0]) if result and result[0] else None
def get_earliest_timestamp(self, symbol: str, timeframe: str) -> Optional[pd.Timestamp]:
"""获取指定 symbol/timeframe 的最早时间戳"""
with self.get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT MIN(timestamp)
FROM ohlcv
WHERE symbol = %s AND timeframe = %s
""", (symbol, timeframe))
result = cursor.fetchone()
return pd.to_datetime(result[0]) if result and result[0] else None
+66 -53
View File
@@ -10,6 +10,8 @@ from binance.client import Client
import pytz
from dotenv import load_dotenv
from data.database import DatabaseManager
# 加载环境变量
load_dotenv()
@@ -23,10 +25,10 @@ class BinanceDataFetcher:
}
def __init__(self, cache_dir: str = "./data/cache"):
"""初始化Binance客户端和缓存目录
"""初始化Binance客户端和数据库管理器
Args:
cache_dir: 缓存下载数据的目录
cache_dir: 缓存/数据库目录
"""
# 配置代理
proxies = None
@@ -58,8 +60,8 @@ class BinanceDataFetcher:
print(" 3. 或者使用已缓存的数据")
self.client = None
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
# 初始化数据库管理器
self.db_manager = DatabaseManager(os.path.join(cache_dir, "market_data.db"))
def fetch_ohlcv(
self,
@@ -81,21 +83,7 @@ class BinanceDataFetcher:
Returns:
包含以下列的DataFrame: timestamp, open, high, low, close, volume
"""
# 生成缓存文件名
cache_file = self._get_cache_filename(symbol, timeframe, start_date, end_date)
# 检查缓存
if use_cache and cache_file.exists():
print(f"从缓存加载数据:{cache_file}")
return pd.read_csv(cache_file, parse_dates=['timestamp'])
# 检查客户端是否可用
if self.client is None:
raise ConnectionError(
"Binance客户端不可用。请使用VPN或使用已缓存的数据。"
)
# 设置默认日期
# 设置日期范围
if end_date is None:
end_dt = datetime.now(pytz.UTC)
else:
@@ -106,6 +94,34 @@ class BinanceDataFetcher:
else:
start_dt = datetime.strptime(start_date, '%Y-%m-%d').replace(tzinfo=pytz.UTC)
start_str = start_dt.strftime('%Y-%m-%d')
end_str = end_dt.strftime('%Y-%m-%d')
# 1. 尝试从数据库获取数据
df = pd.DataFrame()
if use_cache:
df = self.db_manager.get_ohlcv(symbol, timeframe, start_str, end_str)
# 检查数据是否足够
if not df.empty:
min_ts = df['timestamp'].min().replace(tzinfo=pytz.UTC)
max_ts = df['timestamp'].max().replace(tzinfo=pytz.UTC)
# 简单判断覆盖范围 (容忍少量缺失)
if min_ts <= start_dt + timedelta(days=2) and max_ts >= end_dt - timedelta(days=2):
print(f"从数据库加载数据:{symbol} {timeframe} ({len(df)} rows)")
return df
# 2. 从API获取数据
if self.client is None:
if not df.empty:
print(f"⚠ API不可用,返回数据库现有数据 ({len(df)} rows)")
return df
raise ConnectionError(
"Binance客户端不可用。请使用VPN或确保数据库有数据。"
)
# Binance使用USDT交易对
pair = f"{symbol}USDT"
interval = self.INTERVALS[timeframe]
@@ -113,53 +129,47 @@ class BinanceDataFetcher:
print(f"获取 {symbol} {timeframe} 数据,时间范围:{start_dt.date()} 至 {end_dt.date()}...")
# 获取K线数据
klines = self.client.get_historical_klines(
symbol=pair,
interval=interval,
start_str=str(int(start_dt.timestamp() * 1000)),
end_str=str(int(end_dt.timestamp() * 1000))
)
try:
klines = self.client.get_historical_klines(
symbol=pair,
interval=interval,
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}),返回数据库现有数据")
return df
raise e
if not klines:
raise ValueError(f"未获取到 {symbol} {timeframe} 的数据")
if not df.empty:
return df
raise ValueError(f"未获取到 {symbol} {timeframe} 的数据")
# 转换为DataFrame
df = pd.DataFrame(klines, columns=[
new_df = pd.DataFrame(klines, columns=[
'timestamp', 'open', 'high', 'low', 'close', 'volume',
'close_time', 'quote_volume', 'trades', 'taker_buy_base',
'taker_buy_quote', 'ignore'
])
# 选择并转换列
df = df[['timestamp', 'open', 'high', 'low', 'close', 'volume']].copy()
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
new_df = new_df[['timestamp', 'open', 'high', 'low', 'close', 'volume']].copy()
new_df['timestamp'] = pd.to_datetime(new_df['timestamp'], unit='ms')
for col in ['open', 'high', 'low', 'close', 'volume']:
df[col] = df[col].astype(float)
new_df[col] = new_df[col].astype(float)
# 按时间排序
df = df.sort_values('timestamp').reset_index(drop=True)
new_df = new_df.sort_values('timestamp').reset_index(drop=True)
print(f"已获取 {len(df)} 行数据,时间范围:{df['timestamp'].min()} 至 {df['timestamp'].max()}")
print(f"已获取 {len(new_df)} 行数据")
# 保存到缓存
df.to_csv(cache_file, index=False)
print(f"已缓存数据至:{cache_file}")
return df
def _get_cache_filename(
self,
symbol: str,
timeframe: str,
start_date: Optional[str],
end_date: Optional[str]
) -> Path:
"""根据参数生成缓存文件名"""
start = start_date or "1y"
end = end_date or "now"
filename = f"{symbol}_{timeframe}_{start}_{end}.csv"
return self.cache_dir / filename
# 保存到数据库
self.db_manager.save_ohlcv(new_df, symbol, timeframe)
return new_df
def get_latest_price(self, symbol: Literal['BTC', 'ETH', 'SOL']) -> float:
"""获取符号的当前价格
@@ -187,5 +197,8 @@ if __name__ == "__main__":
print(f"日期范围:{df['timestamp'].min()} 至 {df['timestamp'].max()}")
# 获取当前价格
current_price = fetcher.get_latest_price('BTC')
print(f"\n当前BTC价格:${current_price:,.2f}")
try:
current_price = fetcher.get_latest_price('BTC')
print(f"\n当前BTC价格:${current_price:,.2f}")
except:
print("\n无法获取当前价格")
+110
View File
@@ -0,0 +1,110 @@
import pymysql
from datetime import datetime
import json
import os
from contextlib import contextmanager
from dotenv import load_dotenv
load_dotenv()
class TradingDatabase:
"""交易数据库管理类 - 用于记录交易历史和预测结果 (MySQL)"""
def __init__(self, db_path=None):
"""初始化 (db_path 参数已废弃,仅为兼容性保留)"""
self._init_db()
@contextmanager
def get_connection(self):
"""获取 MySQL 数据库连接"""
conn = pymysql.connect(
host=os.getenv('MYSQL_HOST'),
port=int(os.getenv('MYSQL_PORT', 3306)),
user=os.getenv('MYSQL_USER'),
password=os.getenv('MYSQL_PASSWORD'),
database=os.getenv('MYSQL_DB'),
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor
)
try:
yield conn
finally:
conn.close()
def _init_db(self):
"""初始化数据库表结构"""
with self.get_connection() as conn:
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS trades (
id INT AUTO_INCREMENT PRIMARY KEY,
timestamp DATETIME,
market_title VARCHAR(255),
market_slug VARCHAR(255),
model_name VARCHAR(50),
prediction VARCHAR(10),
probability DOUBLE,
action VARCHAR(10),
price DOUBLE,
amount DOUBLE,
size DOUBLE,
status VARCHAR(20),
error_message TEXT,
raw_data JSON,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_market_slug (market_slug),
INDEX idx_status (status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
''')
conn.commit()
def log_trade(self, trade_data: dict):
"""记录交易信息"""
try:
with self.get_connection() as conn:
cursor = conn.cursor()
timestamp = trade_data.get('timestamp', datetime.now().isoformat())
raw_data_json = json.dumps(trade_data.get('raw_data', {}), ensure_ascii=False)
sql = '''
INSERT INTO trades (
timestamp, market_title, market_slug, model_name,
prediction, probability, action, price, amount, size,
status, error_message, raw_data
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
'''
cursor.execute(sql, (
timestamp,
trade_data.get('market_title'),
trade_data.get('market_slug'),
trade_data.get('model_name'),
trade_data.get('prediction'),
trade_data.get('probability'),
trade_data.get('action'),
trade_data.get('price'),
trade_data.get('amount'),
trade_data.get('size'),
trade_data.get('status'),
trade_data.get('error_message'),
raw_data_json
))
conn.commit()
except Exception as e:
print(f"写入数据库失败: {e}")
def check_existing_trade(self, market_slug: str, status: str = 'success') -> bool:
"""检查指定市场是否已有成功交易记录"""
try:
with self.get_connection() as conn:
cursor = conn.cursor()
sql = 'SELECT COUNT(*) as cnt FROM trades WHERE market_slug = %s AND status = %s'
cursor.execute(sql, (market_slug, status))
result = cursor.fetchone()
count = result['cnt'] if result else 0
return count > 0
except Exception as e:
print(f"检查已存在交易失败: {e}")
return False
+295
View File
@@ -0,0 +1,295 @@
# Windows 环境安装指南
本指南帮助您在 Windows 系统上部署加密货币预测系统。
---
## 一、前置要求
- Windows 10/11 (64位)
- 至少 8GB 内存
- 至少 10GB 可用磁盘空间
- 网络连接(需访问 GitHub、PyPI、Binance API)
---
## 二、安装 Miniconda
### 1. 下载 Miniconda
访问 [Miniconda 官网](https://docs.conda.io/en/latest/miniconda.html) 下载 Windows 64-bit 安装包,或直接下载:
```
https://repo.anaconda.com/miniconda/Miniconda3-latest-Windows-x86_64.exe
```
### 2. 安装
1. 双击运行安装程序
2. 选择 "Just Me" (仅为当前用户安装)
3. 安装路径建议使用默认值或 `C:\Miniconda3`
4. **重要**:勾选 "Add Miniconda3 to my PATH environment variable"
5. 完成安装
### 3. 验证安装
打开 **命令提示符 (CMD)** 或 **PowerShell**:
```powershell
conda --version
# 应显示类似: conda 24.x.x
```
---
## 三、创建 Conda 环境
### 1. 创建 Python 3.12 环境
```powershell
conda create -n crypto_pred_312 python=3.12 -y
```
### 2. 激活环境
```powershell
conda activate crypto_pred_312
```
激活后,命令行前缀应显示 `(crypto_pred_312)`
---
## 四、克隆项目
### 1. 安装 Git (如未安装)
下载并安装 [Git for Windows](https://git-scm.com/download/win)
### 2. 克隆仓库
```powershell
cd C:\Projects # 或您喜欢的目录
git clone http://192.168.50.24:3000/subuhui/crpto_predictor.git
cd crpto_predictor
```
---
## 五、安装依赖
### 1. 安装 PyTorch (CPU 版本)
```powershell
# CPU 版本 PyTorch(Windows 不使用 GPU)
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu
```
### 2. 安装项目依赖
```powershell
pip install -r requirements.txt
```
### 3. 安装 IBM 时序模型 (TTM-R2)
```powershell
# PyPI 包名是 granite-tsfm,导入时用 tsfm_public
pip install granite-tsfm
```
验证安装:
```powershell
python -c "from tsfm_public.models.tinytimemixer import TinyTimeMixerForPrediction; print('TTM-R2 安装成功')"
```
---
## 六、配置环境变量
### 1. 复制环境变量模板
```powershell
copy .env.example .env
```
### 2. 编辑 .env 文件
使用记事本或 VSCode 编辑 `.env` 文件:
```ini
# MySQL 数据库配置
MYSQL_HOST=your_mysql_host
MYSQL_PORT=3306
MYSQL_USER=your_username
MYSQL_PASSWORD=your_password
MYSQL_DB=crypto_predictor
# Binance API (可选,用于实时数据)
BINANCE_API_KEY=your_api_key
BINANCE_API_SECRET=your_api_secret
# Polymarket 配置 (可选,用于自动交易)
POLYMARKET_PRIVATE_KEY=your_private_key
DRY_RUN=true # 设为 true 进行模拟交易
# 代理配置 (可选)
USE_PROXY=false
SOCKS5_PROXY=socks5://127.0.0.1:1080
```
---
## 七、数据库配置
### 1. 安装 MySQL
下载并安装 [MySQL Community Server](https://dev.mysql.com/downloads/mysql/)
或使用 Docker:
```powershell
docker run -d --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=yourpassword mysql:8.0
```
### 2. 创建数据库
```sql
CREATE DATABASE crypto_predictor CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
```
### 3. 初始化表结构
首次运行时会自动创建表,或手动执行:
```powershell
python -c "from api.models import init_model_predictions_table; from data.trading_db import TradingDatabase; TradingDatabase()"
```
---
## 八、验证安装
### 1. 测试 PyTorch
```powershell
python -c "import torch; print(f'PyTorch: {torch.__version__}'); print(f'Device: cpu')"
```
### 2. 测试模型加载
```powershell
python -c "from training.config import TrainingConfig; c = TrainingConfig(); print(f'Device: {c.get_device()}')"
# 应输出: Device: cpu
```
### 3. 测试数据库连接
```powershell
python -c "from api.models import get_db_connection; conn = get_db_connection(); print('数据库连接成功!')"
```
---
## 九、运行系统
### 方式一:双击批处理脚本 (推荐)
双击项目根目录的 `start_system.bat`
### 方式二:PowerShell 脚本
```powershell
.\start_system.ps1
```
### 方式三:Python 脚本
```powershell
python start_system.py
```
### 方式四:分别启动
```powershell
# 终端 1: API 服务
python -m uvicorn api.main:app --host 0.0.0.0 --port 8000
# 终端 2: 自动交易
python auto_trade_polymarket.py
```
---
## 十、访问系统
- **前端看板**: http://localhost:8000
- **API 文档**: http://localhost:8000/docs
---
## 常见问题
### Q1: `conda activate` 不生效
在 PowerShell 中需要先初始化 conda:
```powershell
conda init powershell
# 重启 PowerShell
```
### Q2: pip 安装超时
使用国内镜像:
```powershell
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
```
### Q3: PyTorch 安装失败
确保使用 CPU 版本的安装命令,不要安装 CUDA 版本。
### Q4: 数据库连接失败
1. 检查 MySQL 服务是否运行
2. 检查 `.env` 中的数据库配置
3. 检查防火墙是否允许 3306 端口
### Q5: 模型文件缺失
确保 `models/checkpoints/` 目录包含预训练模型文件。如缺失,联系管理员获取。
---
## 目录结构
```
crpto_predictor/
├── api/ # API 服务
│ ├── main.py # FastAPI 入口
│ ├── models.py # 数据库模型
│ ├── prediction_service.py # 预测服务
│ └── static/ # 前端静态文件
├── models/ # 模型定义和权重
│ ├── checkpoints/ # 模型权重文件
│ └── dlinear/ # DLinear 模型
├── data/ # 数据层
├── training/ # 训练配置
├── start_system.py # 跨平台启动脚本
├── start_system.bat # Windows CMD 脚本
├── start_system.ps1 # Windows PowerShell 脚本
├── auto_trade_polymarket.py # 自动交易程序
├── requirements.txt # 依赖列表
└── .env # 环境变量 (需自行创建)
```
---
## 更新项目
```powershell
git pull origin main
pip install -r requirements.txt
```
---
*如有问题,请联系管理员。*
+26 -42
View File
@@ -205,12 +205,13 @@ def main():
# 还原真实价格
nbeats_pred_prices = nbeats_preprocessor.inverse_transform_price(nbeats_pred_scaled)
# 【修改】获取当前K线的open价格(未来可交易价格)用于判断方向
# 第i个样本对应的是第168+i根K线,其open价格是真正可以交易的价格
current_open_prices = df_nbeats['open'].iloc[168:168+len(nbeats_pred_prices)].values
nbeats_preds = (nbeats_pred_prices > current_open_prices).astype(int) # 1=涨, 0=跌
# 【修复】使用当前K线的close价格判断涨跌(与标签定义一致)
# 第i个样本对应的是第168+i根K线,其close价格是当前收盘价
# 标签定义: label = (future_close > current_close)
current_close_prices = df_nbeats['close'].iloc[168:168+len(nbeats_pred_prices)].values
nbeats_preds = (nbeats_pred_prices > current_close_prices).astype(int) # 1=涨, 0=跌
print(f"✓ N-BEATS预测完成 (使用当前K线open价格判断)")
print(f"✓ N-BEATS预测完成 (使用当前K线close价格判断)")
print(f" 预测上涨: {(nbeats_preds==1).sum()} ({(nbeats_preds==1).sum()/len(nbeats_preds)*100:.1f}%)")
print(f" 预测下跌: {(nbeats_preds==0).sum()} ({(nbeats_preds==0).sum()/len(nbeats_preds)*100:.1f}%)")
@@ -223,7 +224,6 @@ def main():
# 从df_2026获取价格数据
all_prices = df_2026['close'].values
all_open_prices = df_2026['open'].values
# 为每个样本创建预测
for i in range(min_samples):
@@ -235,12 +235,12 @@ def main():
if end_idx > len(all_prices):
# 如果超出范围,使用最后512个价格
context_data = all_prices[-context_length:].reshape(1, -1, 1)
# 【修改】使用对应K线的open价格作为当前可交易价格
current_open_price = all_open_prices[min(168 + i, len(all_open_prices) - 1)]
# 【修复】使用对应K线的close价格作为当前收盘价
current_close_price = all_prices[min(168 + i, len(all_prices) - 1)]
else:
context_data = all_prices[start_idx:end_idx].reshape(1, -1, 1)
# 【修改】使用第end_idx根K线的open价格(即当前K线的开盘价)
current_open_price = all_open_prices[end_idx] if end_idx < len(all_open_prices) else all_open_prices[-1]
# 【修复】使用第168+i根K线的close价格(即当前K线的收盘价)
current_close_price = all_prices[168 + i] if 168 + i < len(all_prices) else all_prices[-1]
context_tensor = torch.FloatTensor(context_data).to(device)
@@ -249,12 +249,12 @@ def main():
# 预测第一个时间步的值
pred_price = output.prediction_outputs.cpu().numpy()[0, 0, 0]
# 【修改】判断方向: 预测价格 vs 当前K线open价格
ttm_preds_list.append(1 if pred_price > current_open_price else 0)
# 【修复】判断方向: 预测价格 vs 当前K线close价格(与标签定义一致)
ttm_preds_list.append(1 if pred_price > current_close_price else 0)
ttm_preds = np.array(ttm_preds_list)
print(f"✓ TTM-R2预测完成 (使用当前K线open价格判断)")
print(f"✓ TTM-R2预测完成 (使用当前K线close价格判断)")
print(f" 预测上涨: {(ttm_preds==1).sum()} ({(ttm_preds==1).sum()/len(ttm_preds)*100:.1f}%)")
print(f" 预测下跌: {(ttm_preds==0).sum()} ({(ttm_preds==0).sum()/len(ttm_preds)*100:.1f}%)")
@@ -358,38 +358,22 @@ def main():
print(f"\n集成策略性能:")
print(f" 双模型交集 (DLinear+NBEATS): {two_model_acc*100:.2f}% (覆盖率: {two_model_valid.sum()/len(dlinear_preds)*100:.1f}%)")
print(f" 三模型多数投票: {ensemble_accuracy*100:.2f}% (覆盖率: 100%)")
if len(valid_preds) > 0:
print(f" 三模型交集 (ALL): {ensemble_accuracy*100:.2f}% (覆盖率: {valid_mask.sum()/len(dlinear_preds)*100:.1f}%)")
# 性能评估
print("\n" + "="*80)
print("🎯 策略评估")
print("="*80)
# 性能评估
print("\n" + "="*80)
print("🎯 策略评估")
print("="*80)
best_single = max(dlinear_acc, nbeats_acc, ttm_acc)
best_single = max(dlinear_acc, nbeats_acc, ttm_acc)
if ensemble_accuracy > best_single:
improvement = ensemble_accuracy - best_single
print(f"\n ✅ 三模型交集优于所有单模型!")
print(f" 准确率提升: {improvement*100:.2f}个百分点")
print(f" 代价: 弃权{(~valid_mask).sum()}次预测 ({(~valid_mask).sum()/len(dlinear_preds)*100:.1f}%)")
elif ensemble_accuracy > two_model_acc:
improvement = ensemble_accuracy - two_model_acc
print(f"\n ✅ 三模型交集优于双模型交集!")
print(f" 准确率提升: {improvement*100:.2f}个百分点")
print(f" 但覆盖率下降: {(two_model_valid.sum() - valid_mask.sum())/len(dlinear_preds)*100:.1f}%")
else:
print(f"\n ⚠️ 三模型交集未能提升准确率")
print(f" 可能原因: TTM-R2表现太差({ttm_acc*100:.2f}%),拖累了整体性能")
print(f" 建议: 使用双模型交集(DLinear+NBEATS)")
# 实际收益分析
print(f"\n💰 实际收益分析:")
print(f" 如果只在三个模型都同意时交易:")
print(f" - 交易次数: {valid_mask.sum()} ({valid_mask.sum()/len(dlinear_preds)*100:.1f}%)")
print(f" - 准确率: {ensemble_accuracy*100:.2f}%")
print(f" - 避免交易: {(~valid_mask).sum()}次 (规避不确定性)")
if ensemble_accuracy > best_single:
improvement = ensemble_accuracy - best_single
print(f"\n ✅ 三模型多数投票优于所有单模型!")
print(f" 准确率提升: {improvement*100:.2f}个百分点")
else:
print(f"\n ⚠️ 三模型多数投票未能超越最佳单模型")
print(f" 可能原因: 某个模型表现太差,拖累整体性能")
print("\n" + "="*80)
print("✓ 三模型集成预测分析完成!")
-63
View File
@@ -1,63 +0,0 @@
name: crypto_pred
channels:
- pytorch
- conda-forge
- defaults
dependencies:
# Python
- python=3.10
# PyTorch (Mac M1 支持)
- pytorch
- torchvision
- torchaudio
# 数据处理
- pandas>=2.0.0
- numpy>=1.24.0
- scikit-learn>=1.3.0
# 开发工具
- jupyter
- ipython
- tqdm
# 可视化
- matplotlib
- seaborn
# 通过pip安装的包
- pip
- pip:
# Binance API
- python-binance>=1.0.19
# 环境变量
- python-dotenv>=1.0.0
# 时区处理
- pytz
# ONNX
- onnx>=1.14.0
- onnxruntime>=1.15.0
# 技术指标
- pandas-ta>=0.3.14b
# 模型库
- darts>=0.27.0
- u8darts[torch]>=0.27.0
- tsfm_public>=0.2.0
- gluonts[torch]>=0.14.0
# FastAPI
- fastapi>=0.104.0
- uvicorn[standard]>=0.24.0
- pydantic>=2.0.0
- python-multipart>=0.0.6
# 测试
- pytest>=7.4.0
- black>=23.0.0
- flake8>=6.0.0
+58
View File
@@ -0,0 +1,58 @@
@echo off
chcp 65001 >nul
REM ============================================
REM Windows 环境安装脚本
REM 安装所有依赖和模型
REM ============================================
echo ==========================================
echo 加密货币预测系统 - Windows 安装脚本
echo ==========================================
echo.
REM 检查 conda
where conda >nul 2>&1
if %errorlevel% neq 0 (
echo [错误] 未找到 conda
echo 请先安装 Miniconda: https://docs.conda.io/en/latest/miniconda.html
pause
exit /b 1
)
echo [1/6] 创建 conda 环境...
call conda create -n crypto_pred_312 python=3.12 -y
if %errorlevel% neq 0 (
echo [警告] 环境可能已存在,继续...
)
echo.
echo [2/6] 激活环境...
call conda activate crypto_pred_312
echo.
echo [3/6] 安装 PyTorch (CPU 版本)...
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu
echo.
echo [4/5] 安装项目依赖...
pip install -r requirements.txt
echo.
echo [5/5] 验证安装...
python -c "import torch; print(f'PyTorch: {torch.__version__}')"
python -c "from tsfm_public.models.tinytimemixer import TinyTimeMixerForPrediction; print('TTM-R2: 已安装')"
python -c "import pandas_ta; print('pandas-ta: 已安装')"
python -c "from training.config import TrainingConfig; c=TrainingConfig(); print(f'设备: {c.get_device()}')"
echo.
echo ==========================================
echo 安装完成!
echo ==========================================
echo.
echo 下一步:
echo 1. 复制 .env.example 为 .env 并配置数据库
echo 2. 双击 start_system.bat 启动系统
echo 3. 访问 http://localhost:8000
echo.
pause
+54 -25
View File
@@ -1,40 +1,69 @@
# Core dependencies
torch>=2.0.0
torchvision
torchaudio
python-binance>=1.0.19
# ==============================================
# 加密货币预测系统依赖
# Python 3.12 | 支持 macOS (MPS) 和 Windows (CPU)
# ==============================================
# ----------------------------------------------
# PyTorch (根据平台单独安装)
# ----------------------------------------------
# macOS: pip install torch torchvision torchaudio
# Windows CPU: pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu
# 注意:requirements.txt 中不包含 torch,需要手动安装
# ----------------------------------------------
# 数据处理
# ----------------------------------------------
pandas>=2.0.0
numpy>=1.24.0
scikit-learn>=1.3.0
pytz
python-dotenv>=1.0.0
# ONNX
onnx>=1.14.0
onnxruntime>=1.15.0
# ----------------------------------------------
# 时序模型
# ----------------------------------------------
# IBM Granite TTM-R2 (导入名: tsfm_public)
granite-tsfm>=0.3.0
# Model-specific dependencies
darts>=0.27.0
u8darts[torch]>=0.27.0
tsfm_public>=0.2.0
gluonts[torch]>=0.14.0
# ----------------------------------------------
# 技术指标
# ----------------------------------------------
pandas-ta>=0.3.14b
# API dependencies
# ----------------------------------------------
# API 服务
# ----------------------------------------------
fastapi>=0.104.0
uvicorn[standard]>=0.24.0
pydantic>=2.0.0
python-multipart>=0.0.6
# Technical indicators
pandas-ta>=0.3.14b
# ----------------------------------------------
# HTTP 客户端
# ----------------------------------------------
httpx>=0.25.0
# Development tools
jupyter>=1.0.0
matplotlib>=3.7.0
seaborn>=0.12.0
pytest>=7.4.0
black>=23.0.0
flake8>=6.0.0
# ----------------------------------------------
# 数据源
# ----------------------------------------------
python-binance>=1.0.19
# Utilities
# ----------------------------------------------
# 任务调度
# ----------------------------------------------
schedule>=1.2.0
# ----------------------------------------------
# 数据库
# ----------------------------------------------
pymysql>=1.1.0
cryptography>=41.0.0
# ----------------------------------------------
# 环境变量
# ----------------------------------------------
python-dotenv>=1.0.0
# ----------------------------------------------
# 进度条
# ----------------------------------------------
tqdm>=4.66.0
+14 -8
View File
@@ -1,10 +1,17 @@
"""生成测试数据以便在无法连接Binance时使用"""
import sys
import os
from pathlib import Path
# 添加项目根目录 to path
sys.path.append(str(Path(__file__).resolve().parent.parent))
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from pathlib import Path
from data.database import DatabaseManager
def generate_sample_data(
symbol: str = 'BTC',
@@ -76,14 +83,13 @@ def generate_sample_data(
df = pd.DataFrame(data)
# 保存到缓存
cache_dir = Path('./data/cache')
cache_dir.mkdir(parents=True, exist_ok=True)
# 保存到数据库
# 注意:我们这里假设DatabaseManager的默认路径在脚本运行时是相对于CWD的
# 如果从根目录运行,./data/cache/market_data.db 是正确的
db_manager = DatabaseManager()
db_manager.save_ohlcv(df, symbol, timeframe)
cache_file = cache_dir / f"{symbol}_{timeframe}_1y_now.csv"
df.to_csv(cache_file, index=False)
print(f"✓ 已生成 {len(df)} 行模拟数据并保存到:{cache_file}")
print(f"✓ 已生成 {len(df)} 行模拟数据并保存到数据库")
return df
+47
View File
@@ -0,0 +1,47 @@
@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
+90
View File
@@ -0,0 +1,90 @@
# ============================================
# 系统启动脚本 (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"
}
+137
View File
@@ -0,0 +1,137 @@
#!/usr/bin/env python3
"""
系统启动脚本
同时启动:
1. FastAPI 后端服务 (提供前端看板 API)
2. Polymarket 自动交易程序
支持平台:macOS, Linux, Windows
"""
import subprocess
import sys
import time
import signal
import os
import platform
from datetime import datetime
# 检测操作系统
IS_WINDOWS = platform.system() == 'Windows'
# 颜色代码 (Windows CMD 默认不支持 ANSI,但现代 Windows Terminal 支持)
if IS_WINDOWS:
# 尝试启用 Windows ANSI 支持
try:
import ctypes
kernel32 = ctypes.windll.kernel32
kernel32.SetConsoleMode(kernel32.GetStdHandle(-11), 7)
ANSI_SUPPORTED = True
except Exception:
ANSI_SUPPORTED = False
else:
ANSI_SUPPORTED = True
if ANSI_SUPPORTED:
GREEN = '\033[92m'
YELLOW = '\033[93m'
RED = '\033[91m'
RESET = '\033[0m'
else:
GREEN = YELLOW = RED = RESET = ''
def log(message, color=RESET):
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(f"{color}[{timestamp}] {message}{RESET}")
def start_process(command, name):
"""启动子进程(跨平台兼容)"""
log(f"正在启动 {name}...", YELLOW)
try:
if IS_WINDOWS:
# Windows: 不使用 preexec_fn,使用 CREATE_NEW_PROCESS_GROUP
process = subprocess.Popen(
command,
shell=True,
stdout=sys.stdout,
stderr=sys.stderr,
creationflags=subprocess.CREATE_NEW_PROCESS_GROUP
)
else:
# macOS / Linux: 使用 setsid 创建新进程组
process = subprocess.Popen(
command,
shell=True,
stdout=sys.stdout,
stderr=sys.stderr,
preexec_fn=os.setsid
)
log(f"{name} 已启动 (PID: {process.pid})", GREEN)
return process
except Exception as e:
log(f"启动 {name} 失败: {e}", RED)
return None
def terminate_process(process, name):
"""终止进程(跨平台兼容)"""
if process is None:
return
try:
if IS_WINDOWS:
# Windows: 使用 terminate() 或发送 CTRL_BREAK_EVENT
process.terminate()
else:
# macOS / Linux: 终止整个进程组
os.killpg(os.getpgid(process.pid), signal.SIGTERM)
log(f"{name} 已停止", GREEN)
except Exception as e:
log(f"停止 {name} 时出错: {e}", RED)
# 强制杀死
try:
process.kill()
except Exception:
pass
def main():
# 激活 Conda 环境
# 注意:在子进程中直接使用当前已激活的 python 解释器路径,避免 conda activate 问题
python_exec = sys.executable
# 1. 启动 API 服务 (Uvicorn)
# 使用 --host 0.0.0.0 允许局域网访问
api_cmd = f"{python_exec} -m uvicorn api.main:app --host 0.0.0.0 --port 8000"
api_process = start_process(api_cmd, "API 服务 & 前端看板")
# 等待几秒确保 API 启动
time.sleep(3)
# 2. 启动自动交易程序
# 使用 -u 参数禁用输出缓冲
trade_cmd = f"{python_exec} -u auto_trade_polymarket.py"
trade_process = start_process(trade_cmd, "Polymarket 自动交易程序")
log("系统启动完成!", GREEN)
log("前端访问地址: http://localhost:8000", GREEN)
log("按 Ctrl+C 停止所有服务...", YELLOW)
try:
# 监控子进程状态
while True:
if api_process and api_process.poll() is not None:
log("API 服务意外退出", RED)
break
if trade_process and trade_process.poll() is not None:
log("自动交易程序意外退出", RED)
break
time.sleep(1)
except KeyboardInterrupt:
log("\n正在停止服务...", YELLOW)
finally:
# 优雅退出
terminate_process(api_process, "API 服务")
terminate_process(trade_process, "自动交易程序")
log("所有服务已停止", GREEN)
if __name__ == "__main__":
main()
+13
View File
@@ -0,0 +1,13 @@
[
{
"timestamp": "2026-02-03T21:18:33.773774",
"market": "Bitcoin Up or Down - February 3, 9AM ET",
"prediction": "DOWN",
"success": false,
"order": null,
"error": "价格$0.51超过限制$0.50",
"side": "DOWN",
"price": 0.505,
"probability": 0.5563886165618896
}
]
+11 -6
View File
@@ -38,18 +38,23 @@ class TrainingConfig:
Returns:
Device string ('mps', 'cuda', or 'cpu')
Note:
macOS 使用 MPS,Windows/Linux 使用 CPU
"""
import torch
import platform
if self.device != 'auto':
return self.device
# Auto-detect best available device
if torch.backends.mps.is_available():
return 'mps'
elif torch.cuda.is_available():
return 'cuda'
else:
# 根据操作系统选择设备:macOS 用 MPS,Windows/Linux 用 CPU
if platform.system() == 'Darwin': # macOS
if torch.backends.mps.is_available():
return 'mps'
else:
return 'cpu'
else: # Windows / Linux
return 'cpu'