381 lines
15 KiB
Python
381 lines
15 KiB
Python
"""对比 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()
|