subuhui/crpto_predictor
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add CLAUDE.md guide and preprocessing comparison script

Browse Source
This commit is contained in:
crypto_predictor committed 2026-02-10 23:26:44 +08:00
1 parent 84288a687a
commit d34b67f522
3 files changed
+469 -1
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.claude/settings.local.json
+6 -1
View File
@@ -12,7 +12,12 @@
"Bash(git push:*)",
"Bash(conda activate:*)",
"WebSearch",
"Bash(pip install:*)"
"Bash(pip install:*)",
"Bash(git status:*)",
"Bash(git -C /Users/xxz/PycharmProjects/crypto_predictor check-ignore -v test_dlinear_jan_2026_create_labels.py)",
"Bash(git -C /Users/xxz/PycharmProjects/crypto_predictor status)",
"Bash(git -C /Users/xxz/PycharmProjects/crypto_predictor diff)",
"Bash(git -C /Users/xxz/PycharmProjects/crypto_predictor log --oneline -5)"
]
}
}
CLAUDE.md
+83
View File
@@ -0,0 +1,83 @@
# CLAUDE.md
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
## 项目概述
基于深度学习的 BTC 小时级涨跌预测系统，包含模型训练、FastAPI 后端、Vue 3 看板、Polymarket 自动交易。当前最佳模型 DLinear 准确率 53.10%。
## 环境与构建
```bash
# Python 环境（必须使用 Conda，禁止 venv/poetry）
conda activate crypto_pred_312
# 安装依赖（PyTorch 需根据平台单独安装，requirements.txt 不含 torch）
pip install -r requirements.txt
# 数据库：仅支持 MySQL，禁止 SQLite
# 配置：复制 .env.example 为 .env 并填写 MySQL 连接信息
```
## 常用命令
```bash
# 启动完整系统（API + 自动交易）
python start_system.py
# 单独启动 API 服务
python -m uvicorn api.main:app --host 0.0.0.0 --port 8000
# 单独启动自动交易
python auto_trade_polymarket.py
# 训练 DLinear 分类模型
python train_classification_full.py
# 测试模型（2026年数据）
python test_models_feb_2026.py
# 集成预测
python ensemble_predict.py
# 测试与检查
pytest
flake8 .
```
## 架构
```
数据流: Binance API → data/fetcher.py → data/preprocessor.py (技术指标) → data/dataset.py → 模型推理
交易流: 模型预测 → auto_trade_polymarket.py → Polymarket 下单 → MySQL 记录
服务流: FastAPI (api/main.py) → 前端看板 (api/static/) → http://localhost:8000
```
### 核心模块
- **`data/`** — 数据层：`fetcher.py`(Binance数据获取)、`preprocessor.py`(特征工程: RSI+MACD)、`dataset.py`(PyTorch数据集)、`database.py`(市场数据MySQL)、`trading_db.py`(交易历史MySQL)
- **`models/`** — 模型层：`dlinear/model.py`(主力模型，时序分解+线性层)、`nbeats/`、`base.py`(抽象基类)
- **`models/checkpoints/dlinear_classification_2026/`** — 训练产物：`best_model.pt`、`preprocessor.pkl`、`test_2026_results.json`
- **`training/`** — 训练层：`trainer.py`(训练循环，支持 MPS/CUDA/CPU)、`config.py`(超参数)、`evaluator.py`(评估)
- **`api/`** — 服务层：`main.py`(FastAPI入口)、`models.py`(DB操作+时区转换)、`prediction_service.py`(预测服务)、`polymarket_client.py`(交易客户端)
- **`start_system.py`** — 统一启动入口（同时启动 API 和自动交易）
- **`auto_trade_polymarket.py`** — 自动交易主程序（每小时执行预测+下单）
### DLinear 模型结构
输入 (168h × 9特征) → MovingAvg 时序分解 → 季节性/趋势分量各经线性层 → MLP 聚合 → logits 输出 (BCEWithLogitsLoss)
## 开发约束
- **数据库**：仅 MySQL（`pymysql`），禁止 SQLite。连接使用 `data/database.py` 的 `DatabaseManager` 或 `api/models.py` 的 `get_db_connection`
- **时区**：数据库存 UTC/ET，前端展示 CST（北京时间），转换用 `api.models.convert_to_cst()`
- **代码注释**：中文，解释"为什么"
- **命名**：变量/函数 `snake_case`，类 `PascalCase`
- **代理访问 Binance**：需在 `.env` 配置 `SOCKS5_PROXY` 和 `USE_PROXY=true`
- **日志**：交易日志在 `logs/polymarket_trading.log`（10MB 轮转）
## 禁止事项
- 禁止引入 SQLite 或恢复已删除的 SQLite 代码
- 禁止在未确认的情况下升级核心库版本（torch, pandas 等）
- 禁止删除看似无用但可能被外部脚本调用的代码
compare_preprocess.py
+380
View File
@@ -0,0 +1,380 @@
"""对比 test 旧逻辑 vs auto_trade 逻辑的数据处理差异
两种预处理方式：
- test旧逻辑：整个df一次性预处理，无create_labels，按索引遍历窗口
- auto_trade逻辑：截取到当前K线的df，含create_labels，取X[-1]预测
对比内容：
1. 预处理后的df行数差异
2. 同一时间点的特征值差异（RSI、MACD等）
3. scaler transform后的值差异
4. 最终输入窗口的数值差异
5. 模型输出logits/概率/预测结果差异
"""
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import torch
import numpy as np
import pandas as pd
import pickle
import pandas_ta as ta
from datetime import timedelta
from sklearn.preprocessing import StandardScaler
from data.fetcher import BinanceDataFetcher
# ==================== pickle反序列化需要的占位类 ====================
class ClassificationPreprocessor:
"""pkl文件中保存的类，pickle.load时需要能找到这个类名"""
def __init__(self, lookback=168, forecast_horizon=1):
self.lookback = lookback
self.forecast_horizon = forecast_horizon
self.scaler = StandardScaler()
self.is_fitted = False
# ==================== 两种预处理器 ====================
class TestPreprocessor:
"""test旧逻辑：无create_labels"""
def __init__(self, lookback=168):
self.lookback = lookback
self.scaler = StandardScaler()
def preprocess(self, df, fit=True):
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]
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])
else:
scaled_features = self.scaler.transform(df[feature_cols])
X = []
valid_indices = []
for i in range(len(scaled_features) - self.lookback):
X.append(scaled_features[i:i + self.lookback])
valid_indices.append(i + self.lookback - 1)
return np.array(X), df, valid_indices, scaled_features
class AutoTradePreprocessor:
"""auto_trade逻辑：含create_labels"""
def __init__(self, lookback=168, forecast_horizon=1):
self.lookback = lookback
self.forecast_horizon = forecast_horizon
self.scaler = StandardScaler()
def preprocess(self, df, fit=True):
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]
# create_labels: shift + 砍最后一行
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)
df = df[:-self.forecast_horizon]
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])
else:
scaled_features = self.scaler.transform(df[feature_cols])
X = []
for i in range(len(scaled_features) - self.lookback):
X.append(scaled_features[i:i + self.lookback])
return np.array(X), df, scaled_features
def utc_to_beijing(ts):
return ts + timedelta(hours=8)
def main():
print("=" * 100)
print("数据处理对比：test旧逻辑 vs auto_trade逻辑")
print("=" * 100)
device = torch.device('mps' if torch.backends.mps.is_available() else 'cpu')
# 加载模型和训练时的scaler
from models.dlinear.model import DLinear
checkpoint = torch.load('./models/checkpoints/dlinear_classification_2026/best_model.pt',
map_location=device, weights_only=False)
model = DLinear(checkpoint['config']).to(device).eval()
with open('./models/checkpoints/dlinear_classification_2026/preprocessor.pkl', 'rb') as f:
trained_preprocessor = pickle.load(f)
# 提取训练时的scaler参数
trained_scaler = trained_preprocessor.scaler
print(f"训练时scaler均值: {trained_scaler.mean_}")
print(f"训练时scaler方差: {trained_scaler.scale_}")
# 获取数据
print("\n获取数据...")
fetcher = BinanceDataFetcher()
df_full = fetcher.fetch_ohlcv('BTC', '1h', start_date='2025-12-01', end_date='2026-03-01', use_cache=False)
print(f"总行数: {len(df_full)}")
# 选几个2月10日（北京时间）的时间点做对比
# 北京时间02-10 00:00 = UTC 02-09 16:00
test_targets_utc = [
'2026-02-09 16:00:00', # 北京 02-10 00:00
'2026-02-09 20:00:00', # 北京 02-10 04:00
'2026-02-10 00:00:00', # 北京 02-10 08:00
'2026-02-10 04:00:00', # 北京 02-10 12:00
'2026-02-10 08:00:00', # 北京 02-10 16:00
'2026-02-10 12:00:00', # 北京 02-10 20:00
]
feature_cols = ['open', 'high', 'low', 'close', 'volume',
'rsi', 'macd', 'macd_hist', 'macd_signal']
for target_utc_str in test_targets_utc:
target_utc = pd.Timestamp(target_utc_str)
target_bj = utc_to_beijing(target_utc)
# 找到目标K线在df中的索引
target_mask = df_full['timestamp'] == target_utc
if not target_mask.any():
print(f"\n⚠ 找不到 {target_utc_str}，跳过")
continue
target_idx = df_full[target_mask].index[0]
input_idx = target_idx - 1 # 输入的最后一根K线
input_row = df_full.iloc[input_idx]
target_row = df_full.iloc[target_idx]
print(f"\n{'=' * 100}")
print(f"目标K线: 北京{target_bj.strftime('%m-%d %H:%M')} (UTC {target_utc_str})")
print(f"输入末根: UTC {input_row['timestamp']} O:{input_row['open']:.2f} H:{input_row['high']:.2f} L:{input_row['low']:.2f} C:{input_row['close']:.2f}")
print(f"目标K线: UTC {target_row['timestamp']} 收盘:{target_row['close']:.2f}")
print("-" * 100)
# ===== 方式A: test旧逻辑 =====
# 对整个df一次性预处理，按索引找到对应窗口
test_prep = TestPreprocessor(lookback=168)
test_prep.scaler = trained_scaler # 使用训练时的scaler
X_test, df_test, idx_test, scaled_test = test_prep.preprocess(df_full, fit=False)
# 找到input_idx对应的窗口
# idx_test[i] = i + 168 - 1，所以 input_idx 在 df_test 中的位置需要查找
# df_test 是 dropna 后 reset_index 的，timestamp 可能对应不同的行号
test_input_mask = df_test['timestamp'] == input_row['timestamp']
if test_input_mask.any():
test_df_idx = df_test[test_input_mask].index[0]
# 对应的窗口序号
test_window_i = test_df_idx - 168 + 1
if 0 <= test_window_i < len(X_test):
X_test_window = X_test[test_window_i]
else:
X_test_window = None
print(" [test] 窗口索引越界")
else:
test_df_idx = None
X_test_window = None
print(" [test] 在df_test中找不到对应时间")
# ===== 方式B: auto_trade逻辑 =====
# 截取到input_idx为止，含create_labels
df_up_to_t = df_full.iloc[:input_idx + 1]
auto_prep = AutoTradePreprocessor(lookback=168)
auto_prep.scaler = trained_scaler # 使用训练时的scaler
X_auto, df_auto, scaled_auto = auto_prep.preprocess(df_up_to_t, fit=False)
if len(X_auto) > 0:
X_auto_window = X_auto[-1] # 取最后一个窗口
else:
X_auto_window = None
print(" [auto] X为空")
# ===== 对比 =====
# 1. df行数对比
print(f"\n[1] 预处理后df行数")
print(f" test: {len(df_test)} 行 (整个df, 无create_labels)")
print(f" auto: {len(df_auto)} 行 (截取到input, 含create_labels砍尾)")
# 2. 最后一行时间对比
print(f"\n[2] 预处理后df最后一行")
print(f" test最后一行: {df_test.iloc[-1]['timestamp']} close={df_test.iloc[-1]['close']:.2f}")
print(f" auto最后一行: {df_auto.iloc[-1]['timestamp']} close={df_auto.iloc[-1]['close']:.2f}")
# 3. 输入末根K线的原始特征对比
if test_df_idx is not None:
test_features = df_test.loc[test_df_idx, feature_cols]
auto_last_row = df_auto.iloc[-1]
auto_features = auto_last_row[feature_cols]
print(f"\n[3] 输入末根K线原始特征值")
print(f" {'特征':<15} {'test值':>15} {'auto值':>15} {'差异':>15}")
print(f" {'-'*60}")
for col in feature_cols:
tv = test_features[col]
av = auto_features[col]
diff = tv - av
marker = " ⚠" if abs(diff) > 1e-6 else ""
print(f" {col:<15} {tv:>15.6f} {av:>15.6f} {diff:>15.8f}{marker}")
# 4. scaler transform 后的值对比（输入末根K线）
if test_df_idx is not None and X_test_window is not None and X_auto_window is not None:
test_scaled_last = scaled_test[test_df_idx]
auto_scaled_last = scaled_auto[-1]
print(f"\n[4] scaler transform 后的值（输入末根K线）")
print(f" {'特征':<15} {'test值':>15} {'auto值':>15} {'差异':>15}")
print(f" {'-'*60}")
for j, col in enumerate(feature_cols):
tv = test_scaled_last[j]
av = auto_scaled_last[j]
diff = tv - av
marker = " ⚠" if abs(diff) > 1e-6 else ""
print(f" {col:<15} {tv:>15.6f} {av:>15.6f} {diff:>15.8f}{marker}")
# 5. 输入窗口对比（整个168步窗口）
if X_test_window is not None and X_auto_window is not None:
window_diff = X_test_window - X_auto_window
max_diff = np.max(np.abs(window_diff))
mean_diff = np.mean(np.abs(window_diff))
nonzero_count = np.count_nonzero(np.abs(window_diff) > 1e-6)
print(f"\n[5] 输入窗口对比 (168步 × 9特征 = 1512个值)")
print(f" 最大绝对差异: {max_diff:.10f}")
print(f" 平均绝对差异: {mean_diff:.10f}")
print(f" 差异>1e-6的值: {nonzero_count} / 1512")
if nonzero_count > 0:
# 找到差异最大的位置
max_pos = np.unravel_index(np.argmax(np.abs(window_diff)), window_diff.shape)
print(f" 最大差异位置: 步={max_pos[0]} 特征={feature_cols[max_pos[1]]}")
print(f" test值: {X_test_window[max_pos]:.10f}")
print(f" auto值: {X_auto_window[max_pos]:.10f}")
# 6. 模型推理对比
if X_test_window is not None and X_auto_window is not None:
with torch.no_grad():
test_tensor = torch.FloatTensor(X_test_window[np.newaxis]).to(device)
auto_tensor = torch.FloatTensor(X_auto_window[np.newaxis]).to(device)
test_logits = model(test_tensor).cpu().numpy().item()
auto_logits = model(auto_tensor).cpu().numpy().item()
test_prob = 1 / (1 + np.exp(-test_logits))
auto_prob = 1 / (1 + np.exp(-auto_logits))
test_pred = "涨" if test_prob > 0.5 else "跌"
auto_pred = "涨" if auto_prob > 0.5 else "跌"
match = "✓ 一致" if test_pred == auto_pred else "✗ 不一致"
print(f"\n[6] 模型推理结果")
print(f" {'':15} {'test':>15} {'auto':>15} {'差异':>15}")
print(f" {'-'*60}")
print(f" {'logits':<15} {test_logits:>15.8f} {auto_logits:>15.8f} {test_logits-auto_logits:>15.8f}")
print(f" {'概率':<15} {test_prob:>15.8f} {auto_prob:>15.8f} {test_prob-auto_prob:>15.8f}")
print(f" {'预测':<15} {test_pred:>15} {auto_pred:>15} {match:>15}")
# ===== 全量对比：2月所有小时 =====
print(f"\n\n{'=' * 100}")
print("全量对比：2月份所有小时预测结果差异统计")
print("=" * 100)
# test方式：一次性预处理
test_prep_full = TestPreprocessor(lookback=168)
test_prep_full.scaler = trained_scaler
X_test_full, df_test_full, idx_test_full, _ = test_prep_full.preprocess(df_full, fit=False)
# 构建 timestamp -> test窗口索引 的映射
test_ts_map = {}
for i, idx in enumerate(idx_test_full):
ts = df_test_full.iloc[idx]['timestamp']
test_ts_map[ts] = i
total = 0
same = 0
diff_list = []
for target_idx in range(1, len(df_full)):
target_time_utc = df_full.iloc[target_idx]['timestamp']
target_time_bj = utc_to_beijing(target_time_utc)
if not (target_time_bj.year == 2026 and target_time_bj.month == 2):
continue
input_idx = target_idx - 1
input_ts = df_full.iloc[input_idx]['timestamp']
# test方式
test_i = test_ts_map.get(input_ts)
if test_i is None or test_i >= len(X_test_full):
continue
# auto方式
df_up_to_t = df_full.iloc[:input_idx + 1]
auto_prep_tmp = AutoTradePreprocessor(lookback=168)
auto_prep_tmp.scaler = trained_scaler
X_auto_tmp, _, _ = auto_prep_tmp.preprocess(df_up_to_t, fit=False)
if len(X_auto_tmp) == 0:
continue
with torch.no_grad():
test_logits = model(torch.FloatTensor(X_test_full[test_i:test_i+1]).to(device)).cpu().numpy().item()
auto_logits = model(torch.FloatTensor(X_auto_tmp[-1:]).to(device)).cpu().numpy().item()
test_prob = 1 / (1 + np.exp(-test_logits))
auto_prob = 1 / (1 + np.exp(-auto_logits))
test_pred = 1 if test_prob > 0.5 else 0
auto_pred = 1 if auto_prob > 0.5 else 0
total += 1
if test_pred == auto_pred:
same += 1
else:
diff_list.append({
'time_bj': target_time_bj.strftime('%m-%d %H:%M'),
'test_prob': test_prob,
'auto_prob': auto_prob,
'test_pred': '涨' if test_pred == 1 else '跌',
'auto_pred': '涨' if auto_pred == 1 else '跌',
})
print(f"总预测数: {total}")
print(f"结果一致: {same} ({same/total:.1%})")
print(f"结果不同: {len(diff_list)} ({len(diff_list)/total:.1%})")
if diff_list:
print(f"\n不一致的时间点:")
print(f" {'北京时间':<15} {'test概率':>10} {'test预测':>8} {'auto概率':>10} {'auto预测':>8}")
print(f" {'-'*55}")
for d in diff_list:
print(f" {d['time_bj']:<15} {d['test_prob']:>10.6f} {d['test_pred']:>8} {d['auto_prob']:>10.6f} {d['auto_pred']:>8}")
print("=" * 100)
if __name__ == "__main__":
main()