简介:本资源是一套面向本科生与初学者的股价预测综合实践项目,涵盖线性回归(LR)、长短期记忆网络(LSTM)、自回归积分滑动平均(ARIMA)及K近邻(KNN)等多种主流机器学习方法,专为毕业设计、期末大作业及课程设计打造。项目提供完整可运行代码、详细注释、可视化分析图表及配套文档,支持端到端数据预处理、模型训练、回测评估与结果展示,新手亦能快速上手并理解各模型原理与适用场景。压缩包共31个文件,含5个核心Python脚本(如forecast.py、backtest.py、models.py)、2个CSV行情数据、2个HTML交互图表、17张模型效果对比图(含LSTM、ARIMA、Prophet等拟合曲线与残差分析),以及README.md、requirements.txt等工程化支撑文件,整体仅1.45MB,轻量易部署。目前已有117人学习下载,项目经严格调试,功能完备、界面直观、结构清晰,附带个股持有分析、月度资金曲线与收益率可视化等实用模块,具备扎实的教学示范性与实际复现价值。
1. 股价预测不是玄学:为什么用 LR、LSTM、ARIMA、KNN 四种方法并行建模,反而比单模型更稳?
你手头有一组日频股票收盘价数据,想跑个预测——结果发现:用 LSTM 训出来曲线很“丝滑”,但一到跳空缺口就崩;ARIMA 在平稳段误差小,可遇到财报季突变直接失灵;KNN 看似鲁棒,却对最近三天的量价组合过度敏感,把噪声当信号;LR 更惨,连趋势方向都常判反。这不是你调参不行,而是股价本身不具备单一统计假设的刚性结构:它既含长期记忆(适合 LSTM),又带短期自相关(ARIMA 擅长),还存在局部相似模式(KNN 可捕获),同时线性成分不可忽略(LR 提供基线锚点)。本项目不追求“最优单模型”,而是构建一个四路并行、结果可比、误差可归因的预测框架:每个模型独立训练、统一评估、输出带置信区间的区间预测,最终用简单加权融合降低方差。适合刚接触金融时序的新手建立完整 pipeline 意识,也适合有经验者快速验证新特征或替换某一路模型。所有代码在本地 CPU 即可跑通,无需 GPU,数据预处理与模型封装已解耦,你改一行就能切模型、换股票、调窗口。
2. 数据准备与特征工程:从原始 OHLCV 到模型可吞食的张量
2.1 原始数据清洗:处理缺失、停牌、复权与异常值
股价数据最常踩的第一个坑,是直接拿交易所原始 CSV 开干。真实场景中,你需要先解决三类硬伤:
- 停牌日填充:不能简单前向填充(会伪造交易信号),应标记为
NaN并在后续滑动窗口中自动剔除; - 复权处理:必须使用后复权(post-adjusted)价格,否则分红送股导致的价格断层会彻底污染 ARIMA 的平稳性检验;
- 异常值过滤:单日涨跌幅 >15%(非ST/科创板)或成交量突增 5 倍以上,需用
scipy.stats.zscore检出并设为NaN,而非删除——因为 LSTM 需要时间连续性。
import pandas as pd import numpy as np from scipy import stats def load_and_clean_stock_data(file_path: str) -> pd.DataFrame: df = pd.read_csv(file_path, parse_dates=['date'], index_col='date') # 1. 强制后复权:若原始列含 'close_adj' 优先用,否则用 'close' + 复权因子列 price_col = 'close_adj' if 'close_adj' in df.columns else 'close' df['price'] = df[price_col].ffill() # 后复权价已处理断层,仅需前向填充极少数缺失 # 2. 标记停牌:成交量为 0 或 NaN 且价格未变,视为停牌 df['is_suspended'] = (df['volume'] == 0) | (df['volume'].isna()) df.loc[df['is_suspended'], ['price', 'volume']] = np.nan # 3. 异常值检测(Z-score > 4) z_scores = np.abs(stats.zscore(df[['price', 'volume']].dropna())) outliers = (z_scores > 4).any(axis=1) outlier_idx = df.dropna(subset=['price', 'volume']).index[outliers] df.loc[outlier_idx, ['price', 'volume']] = np.nan return df[['price', 'volume']].sort_index() # 示例调用 raw_df = load_and_clean_stock_data("stock_600519.csv") # 贵州茅台示例 print(f"原始记录数: {len(raw_df)}, 清洗后有效记录: {raw_df['price'].count()}")提示:
raw_df['price'].count()是关键指标——若清洗后有效数据 < 原始 85%,说明该股票噪音过大,建议换标的或检查数据源质量。不要强行补全。
2.2 构造多尺度特征:不只是 lag,而是「滞后+滚动+比率」三维组合
LR 和 KNN 对特征敏感度高,LSTM 虽能自动学习,但喂高质量特征仍显著提升收敛速度与泛化性。我们不堆砌上百个技术指标,只保留可解释、易复现、有经济含义的 12 维特征:
| 特征名 | 计算方式 | 用途说明 |
|---|---|---|
price_lag1 | shift(1) | 最基础的自回归信息 |
price_rtn | pct_change(1) | 日收益率,消除量纲 |
vol_ma5 | rolling(5).mean() | 5日均量,衡量流动性热度 |
vol_ratio | volume / vol_ma5 | 当前量能相对强度,突破信号前置指标 |
high_low_ratio | high / low(需原始OHLC) | 当日波动率代理,反映多空博弈烈度 |
price_ma20 | rolling(20).mean() | 月线支撑/压力位参考 |
price_dist_ma20 | (price - price_ma20) / price_ma20 | 价格偏离月线程度,超买超卖指示 |
rsi_14 | 100 - 100/(1 + avg_up/avg_down) | 经典RSI,但用ta库精确实现(见下文) |
# 安装 ta 库:pip install ta from ta.momentum import RSIIndicator from ta.volatility import BollingerBands def build_features(df: pd.DataFrame, window_size: int = 60) -> pd.DataFrame: feat = df.copy() # 基础滞后与收益率 feat['price_lag1'] = feat['price'].shift(1) feat['price_rtn'] = feat['price'].pct_change(1) # 滚动统计(避免未来信息泄露:全部用 closed='left') feat['vol_ma5'] = feat['volume'].rolling(5, closed='left').mean() feat['vol_ratio'] = feat['volume'] / feat['vol_ma5'] feat['price_ma20'] = feat['price'].rolling(20, closed='left').mean() feat['price_dist_ma20'] = (feat['price'] - feat['price_ma20']) / feat['price_ma20'] # 技术指标(需确保 ta 库版本 >= 0.10.2) rsi_ind = RSIIndicator(close=feat['price'], window=14, fillna=False) feat['rsi_14'] = rsi_ind.rsi() bb_ind = BollingerBands(close=feat['price'], window=20, window_dev=2, fillna=False) feat['bb_high'] = bb_ind.bollinger_hband() feat['bb_low'] = bb_ind.bollinger_lband() feat['bb_width'] = (feat['bb_high'] - feat['bb_low']) / feat['price_ma20'] # 布林带宽度归一化 # 目标变量:预测未来第 1 天收盘价(回归任务)或涨跌方向(分类任务) feat['target_price'] = feat['price'].shift(-1) # 预测明日收盘价 feat['target_class'] = (feat['target_price'] > feat['price']).astype(int) # 1=涨,0=跌 return feat feat_df = build_features(raw_df) print("特征维度:", feat_df.shape[1], "| 示例特征:", list(feat_df.columns[-5:]))注意:所有
rolling必须显式指定closed='left',否则 pandas 默认closed='both'会引入未来信息(即用当天数据计算包含当天的均值),这是金融时序建模最隐蔽的翻车点。
2.3 滑动窗口切片:为 LSTM 准备 (samples, timesteps, features),为其他模型准备 (samples, features)
LSTM 输入是三维张量,而 LR/KNN/ARIMA 接收二维表格。我们用统一窗口逻辑生成两套数据:
- LSTM 输入:取过去
timesteps=60天的全部特征(12维),预测第 61 天的target_price; - 传统模型输入:将同一 60 天窗口的最后一天所有特征(即
timesteps=1的切片)作为样本,目标仍是第 61 天价格。
def create_sequences( data: pd.DataFrame, target_col: str = 'target_price', timesteps: int = 60, feature_cols: list = None ) -> tuple: if feature_cols is None: feature_cols = [c for c in data.columns if c not in ['target_price', 'target_class']] X_lstm, y_lstm = [], [] X_trad, y_trad = [], [] # 确保目标列存在且无 NaN valid_mask = ~data[target_col].isna() data_valid = data[valid_mask].copy() for i in range(timesteps, len(data_valid)): # LSTM:取 [i-timesteps : i] 共 timesteps 行 seq_X = data_valid.iloc[i-timesteps:i][feature_cols].values seq_y = data_valid.iloc[i][target_col] X_lstm.append(seq_X) y_lstm.append(seq_y) # 传统模型:只取窗口最后一行的特征(即第 i-1 行,因目标是第 i 行) X_trad.append(data_valid.iloc[i-1][feature_cols].values) y_trad.append(seq_y) return ( np.array(X_lstm), np.array(y_lstm), np.array(X_trad), np.array(y_trad) ) X_lstm, y_lstm, X_trad, y_trad = create_sequences( feat_df, timesteps=60, feature_cols=['price_lag1', 'price_rtn', 'vol_ratio', 'price_dist_ma20', 'rsi_14', 'bb_width'] ) print(f"LSTM 输入形状: {X_lstm.shape} → {X_lstm.shape[0]} 个样本, 每个含 60 步×6 特征") print(f"传统模型输入形状: {X_trad.shape} → {X_trad.shape[0]} 个样本, 每个含 6 特征")血泪经验:
X_trad的构造必须用i-1行特征预测i行目标,而不是i行特征预测i行目标——后者是数据穿越(data leakage),模型会在训练集上表现完美,实盘必崩。
3. 四模型并行训练:LR、KNN、ARIMA、LSTM 的最小可行实现
3.1 线性回归(LR):用 sklearn 实现,但必须做标准化与残差诊断
LR 不是“玩具模型”。在股价预测中,它提供可解释的基线:若 LR 的 R² < 0.05,说明数据噪声主导,所有复杂模型都难有收益。关键步骤:
- 特征必须标准化(
StandardScaler),否则量纲差异(如price_dist_ma20在 ±0.1,rsi_14在 0~100)会让系数失真; - 训练后必须画残差图,若残差随预测值增大而扩散(heteroscedasticity),说明需对目标变量做对数变换。
from sklearn.linear_model import LinearRegression from sklearn.preprocessing import StandardScaler from sklearn.metrics import r2_score, mean_absolute_error import matplotlib.pyplot as plt # 划分训练/测试(按时间顺序,禁用 shuffle!) split_idx = int(0.8 * len(X_trad)) X_train_lr, X_test_lr = X_trad[:split_idx], X_trad[split_idx:] y_train_lr, y_test_lr = y_trad[:split_idx], y_trad[split_idx:] # 标准化 scaler_lr = StandardScaler() X_train_scaled = scaler_lr.fit_transform(X_train_lr) X_test_scaled = scaler_lr.transform(X_test_lr) # 训练 lr_model = LinearRegression() lr_model.fit(X_train_scaled, y_train_lr) # 预测 y_pred_lr = lr_model.predict(X_test_scaled) # 评估 r2 = r2_score(y_test_lr, y_pred_lr) mae = mean_absolute_error(y_test_lr, y_pred_lr) print(f"LR 测试集 R²: {r2:.4f}, MAE: {mae:.4f}") # 残差诊断图 residuals = y_test_lr - y_pred_lr plt.figure(figsize=(10, 4)) plt.subplot(1, 2, 1) plt.scatter(y_pred_lr, residuals, alpha=0.5) plt.axhline(y=0, color='r', linestyle='--') plt.xlabel('Predicted Price') plt.ylabel('Residual') plt.title('Residual vs Fitted') plt.subplot(1, 2, 2) plt.hist(residuals, bins=30, alpha=0.7) plt.xlabel('Residual') plt.ylabel('Frequency') plt.title('Residual Distribution') plt.tight_layout() plt.show()参数说明:
StandardScaler的fit_transform必须只在训练集上调用,测试集用transform—— 这是防止信息泄露的铁律。若残差图显示明显漏斗形,应在create_sequences中对target_price做np.log1p变换,并在预测后np.expm1还原。
3.2 KNN:用距离加权 + 特征缩放,避免“最近邻”变成“最近噪声”
KNN 在股价中有效,是因为市场存在局部相似性(如同样在年报前3天、量比>2、RSI<30 的形态,后续走势常趋同)。但直接KNeighborsRegressor(n_neighbors=5)会失败,原因:
- 未缩放特征导致
price_dist_ma20(±0.1)被rsi_14(0~100)完全淹没; - 固定
n_neighbors=5在不同市场阶段(牛市/熊市)鲁棒性差; - 未加权导致离得远的邻居和近的邻居投票权重一样。
from sklearn.neighbors import KNeighborsRegressor from sklearn.model_selection import TimeSeriesSplit, GridSearchCV # 时间序列交叉验证(非随机打乱!) tscv = TimeSeriesSplit(n_splits=3) # 网格搜索:在训练集上找最优 k 和权重 param_grid = { 'n_neighbors': [3, 5, 7, 10], 'weights': ['uniform', 'distance'], 'p': [1, 2] # 曼哈顿 or 欧氏距离 } knn_model = KNeighborsRegressor() grid_search = GridSearchCV( knn_model, param_grid, cv=tscv, scoring='neg_mean_absolute_error', n_jobs=-1 ) grid_search.fit(X_train_scaled, y_train_lr) # 注意:这里用 LR 的训练标签,因目标一致 print("KNN 最优参数:", grid_search.best_params_) best_knn = grid_search.best_estimator_ y_pred_knn = best_knn.predict(X_test_scaled) mae_knn = mean_absolute_error(y_test_lr, y_pred_knn) print(f"KNN 测试集 MAE: {mae_knn:.4f} (vs LR: {mae:.4f})")关键点:
TimeSeriesSplit是唯一合规的 CV 方式——它保证每次验证集都在训练集之后,模拟真实预测场景。若用KFold,模型会看到“未来”数据,MAE 虚高 30%+。
3.3 ARIMA:用 pmdarima 自动选参,但必须手动验证平稳性与残差白噪声
ARIMA 不是黑箱。pmdarima.auto_arima能省力,但若跳过三步检验,模型就是空中楼阁:
- ADF 检验:确认一阶差分后序列平稳(p-value < 0.05);
- ACF/PACF 图:目视判断
p,q初值; - Ljung-Box 检验:拟合后残差必须是白噪声(p-value > 0.05)。
import pmdarima as pm from statsmodels.tsa.stattools import adfuller from statsmodels.stats.diagnostic import acorr_ljungbox # 1. ADF 检验(用原始 price 序列) adf_result = adfuller(raw_df['price'].dropna()) print(f"ADF Statistic: {adf_result[0]:.4f}, p-value: {adf_result[1]:.4f}") # 若 p > 0.05,需差分:d=1 # 2. 自动拟合 ARIMA(限定搜索范围防过拟合) arima_model = pm.auto_arima( raw_df['price'].dropna(), start_p=1, max_p=3, start_q=1, max_q=3, d=1, # 强制一阶差分 seasonal=False, stepwise=True, suppress_warnings=True, error_action="ignore", n_jobs=-1 ) print("ARIMA 最优阶数:", arima_model.order) # 如 (2,1,1) # 3. 残差白噪声检验 residuals_arima = arima_model.resid() lb_test = acorr_ljungbox(residuals_arima, lags=[10], return_df=True) print("Ljung-Box p-value:", lb_test['lb_pvalue'].iloc[0]) # 4. 预测未来 1 步(对应 test 长度) y_pred_arima = arima_model.predict(n_periods=len(y_test_lr)) mae_arima = mean_absolute_error(y_test_lr, y_pred_arima) print(f"ARIMA 测试集 MAE: {mae_arima:.4f}")避坑提醒:
auto_arima默认m=0(非季节性),但 A 股存在明显月度效应(如每月初资金面宽松),若你研究的是月频数据,必须设seasonal=True, m=12并增加max_P,max_Q。
3.4 LSTM:用 Keras 构建双层堆叠,但 dropout 位置与早停策略决定成败
LSTM 易过拟合。我们的最小可靠结构:
- 输入层:
LSTM(50, return_sequences=True, dropout=0.2, recurrent_dropout=0.2) - 第二层:
LSTM(30, dropout=0.2, recurrent_dropout=0.2) - 输出层:
Dense(1) - 关键:
recurrent_dropout必须 >0,否则循环连接过拟合;early_stopping监控验证集 loss,patience=10。
import tensorflow as tf from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense, Dropout from tensorflow.keras.callbacks import EarlyStopping # 划分 LSTM 数据(同样按时间顺序) split_idx_lstm = int(0.8 * len(X_lstm)) X_train_lstm, X_test_lstm = X_lstm[:split_idx_lstm], X_lstm[split_idx_lstm:] y_train_lstm, y_test_lstm = y_lstm[:split_idx_lstm], y_lstm[split_idx_lstm:] # 构建模型 model_lstm = Sequential([ LSTM(50, return_sequences=True, dropout=0.2, recurrent_dropout=0.2, input_shape=(X_lstm.shape[1], X_lstm.shape[2])), LSTM(30, dropout=0.2, recurrent_dropout=0.2), Dense(1) ]) model_lstm.compile(optimizer='adam', loss='mse', metrics=['mae']) early_stopping = EarlyStopping( monitor='val_loss', patience=10, restore_best_weights=True ) history = model_lstm.fit( X_train_lstm, y_train_lstm, batch_size=32, epochs=100, validation_split=0.2, callbacks=[early_stopping], verbose=0 ) y_pred_lstm = model_lstm.predict(X_test_lstm).flatten() mae_lstm = mean_absolute_error(y_test_lstm, y_pred_lstm) print(f"LSTM 测试集 MAE: {mae_lstm:.4f}") # 绘制训练曲线 plt.figure(figsize=(8, 4)) plt.plot(history.history['loss'], label='Train Loss') plt.plot(history.history['val_loss'], label='Val Loss') plt.legend() plt.title('LSTM Training Loss') plt.show()玄学参数:
batch_size=32是经验值——太小收敛慢,太大易震荡;recurrent_dropout=0.2比dropout=0.5更有效,因它专门正则化循环连接,而非输入连接。
4. 避坑指南:四个模型在股价预测中必踩的 5 个真实坑
4.1 现象:LSTM 训练 loss 下降快,但测试 MAE 比 LR 还高
原因:输入数据未归一化。LSTM 对输入尺度极度敏感,若price在 1000+ 量级而rsi_14在 0~100,梯度更新会严重偏向大尺度特征,导致模型学不到rsi的微弱信号。
解决:在create_sequences后,对X_lstm沿feature维度做 MinMaxScaler(非 StandardScaler),因 LSTM 更适应 [0,1] 区间。代码:
from sklearn.preprocessing import MinMaxScaler scaler_lstm = MinMaxScaler() X_lstm_reshaped = X_lstm.reshape(-1, X_lstm.shape[-1]) X_lstm_scaled = scaler_lstm.fit_transform(X_lstm_reshaped) X_lstm = X_lstm_scaled.reshape(X_lstm.shape)4.2 现象:ARIMA 预测结果是一条直线
原因:auto_arima选了d=0(未差分),而股价原始序列非平稳,模型只能拟合常数项。
解决:强制d=1,并在auto_arima中设stationary=False,同时用adf_test验证差分后序列:
diff_series = raw_df['price'].diff().dropna() adf_diff = adfuller(diff_series) assert adf_diff[1] < 0.05, "一阶差分后仍不平稳,请尝试 d=2"4.3 现象:KNN 预测值集中在几个离散水平,像阶梯
原因:n_neighbors过小(如 k=1),导致模型退化为“复制最近一天价格”,缺乏平滑性。
解决:网格搜索时n_neighbors至少覆盖 [5, 15],且必须用weights='distance',让近邻权重更高。若仍阶梯化,说明特征维度不足,需增加price_ma60或vol_ma20等长周期特征。
4.4 现象:LR 的 R² 为负数
原因:测试集分布与训练集偏移(如训练用牛市数据,测试用熊市),或目标变量存在强异方差。
解决:
- 检查
y_train_lr与y_test_lr的均值、标准差比值,若 >1.5 倍,说明分布漂移,应重切分点(如按年份切:2020-2022 训练,2023 测试); - 对
y_train_lr做 Box-Cox 变换:from scipy import stats; y_train_box, _ = stats.boxcox(y_train_lr),预测后逆变换。
4.5 现象:四模型预测结果差异极大,融合后效果反而不如单模型
原因:未做预测区间校准。LSTM 输出点预测,ARIMA 可输出置信区间,但直接平均点预测会放大误差。
解决:统一用分位数回归思想——对每个模型,用其验证集残差的 10%/90% 分位数,构建预测区间[pred - q10, pred + q90],融合时只取区间交集的中点。代码逻辑:
# 假设 y_val_true, y_val_pred 为各模型在验证集上的真值与预测 residuals = y_val_true - y_val_pred q10, q90 = np.percentile(residuals, [10, 90]) interval_low = y_pred_test - q90 # 注意:q90 是正数,减去它得下界 interval_high = y_pred_test - q10 # q10 是负数,减去它得上界 ensemble_pred = (interval_low + interval_high) / 25. 模型融合与实盘推演:用加权平均 + 置信度门控,把四模型变成稳定信号源
5.1 为什么不用 stacking 或 voting?——金融场景的特殊约束
Stacking(用元模型学习各模型输出)在图像识别中有效,但在股价预测中是危险的:
- 元模型(如第二层 LR)会拟合各模型在历史上的“错误模式”,一旦市场风格切换(如从价值股切换到题材股),元模型会把旧错误当规律;
- Voting 要求模型输出类别(涨/跌),但股价预测本质是回归问题,强行分类损失关键信息(如“涨1%”和“涨5%”决策完全不同)。
我们采用轻量级、可解释、可审计的融合策略:
- 权重动态计算:每个模型在最近 30 天验证集上的 MAE 取倒数,归一化为权重;
- 置信度门控:若某模型预测区间宽度 > 近 30 天均值的 2 倍,则将其权重置 0(拒绝低置信度信号);
- 输出带区间:最终预测 = 加权平均点预测,区间 = 加权平均上下界。
def calculate_dynamic_weights( models: list, # [lr_model, knn_model, arima_model, lstm_model] X_val_list: list, # 对应各模型的验证集 X y_val: np.ndarray, # 真实验证标签 window_size: int = 30 ) -> np.ndarray: weights = [] intervals = [] for i, model in enumerate(models): if i == 2: # ARIMA 模型 y_pred_val = model.predict(n_periods=len(y_val)) elif i == 3: # LSTM 模型 y_pred_val = model.predict(X_val_list[i]).flatten() else: # LR/KNN y_pred_val = model.predict(X_val_list[i]) # 计算最近 window_size 天的 MAE recent_mae = mean_absolute_error( y_val[-window_size:], y_pred_val[-window_size:] ) weights.append(1 / (recent_mae + 1e-6)) # 防零 # 估算预测区间(用验证集残差分位数) residuals = y_val - y_pred_val q10, q90 = np.percentile(residuals[-window_size:], [10, 90]) intervals.append((q10, q90)) # 归一化权重 weights = np.array(weights) weights = weights / weights.sum() # 置信度门控:若区间宽度 > 均值2倍,权重归零 widths = np.array([q90 - q10 for q10, q90 in intervals]) mean_width = np.mean(widths) for i in range(len(weights)): if widths[i] > 2 * mean_width: weights[i] = 0 weights = weights / (weights.sum() + 1e-6) # 再次归一化 return weights, intervals # 示例:假设有四模型验证预测 X_val_lr = X_trad[split_idx:split_idx+100] # LR 验证集 X_val_knn = X_trad[split_idx:split_idx+100] # KNN 验证集 X_val_lstm = X_lstm[split_idx_lstm:split_idx_lstm+100] # LSTM 验证集 y_val = y_trad[split_idx:split_idx+100] # 假设已训练好四模型:lr_model, knn_model, arima_model, model_lstm models = [lr_model, knn_model, arima_model, model_lstm] X_val_list = [X_val_lr, X_val_knn, None, X_val_lstm] # ARIMA 不需要 X,填 None weights, intervals = calculate_dynamic_weights(models, X_val_list, y_val) print("动态权重:", [f"{w:.3f}" for w in weights]) print("各模型区间宽度:", [f"{q90-q10:.4f}" for q10, q90 in intervals])5.2 实盘推演:用滚动回测验证融合策略的稳定性
真正考验模型的,不是单次测试 MAE,而是在未知时间段上的持续表现。我们用滚动窗口回测(Rolling Walk-Forward Validation):
- 初始训练集:前 1000 天;
- 每次预测 1 天,然后将该天数据加入训练集,滑动窗口;
- 记录每天的融合预测、真实值、各模型权重、区间覆盖率(真实值落在预测区间内的比例)。
def rolling_backtest( raw_df: pd.DataFrame, models: list, feature_func: callable, window_size: int = 1000, test_days: int = 250 ) -> pd.DataFrame: results = [] feat_df = feature_func(raw_df) for day in range(window_size, window_size + test_days): # 构建当日训练数据 train_feat = feat_df.iloc[:day] X_train_lstm, y_train_lstm, X_train_trad, y_train_trad = create_sequences( train_feat, timesteps=60 ) # 重新训练模型(简化版:实际中 LR/KNN 可增量更新,ARIMA/LSTM 需重训) # ... (此处省略训练代码,聚焦逻辑) # 预测第 day 天 y_pred_fused, interval_low, interval_high = ensemble_predict( models, X_train_trad, X_train_lstm, feat_df.iloc[day-60:day] ) true_price = feat_df.iloc[day]['target_price'] in_interval = (true_price >= interval_low) and (true_price <= interval_high) results.append({ 'date': feat_df.index[day], 'pred': y_pred_fused, 'true': true_price, 'interval_low': interval_low, 'interval_high': interval_high, 'in_interval': in_interval, 'error': abs(true_price - y_pred_fused) }) return pd.DataFrame(results) # 执行回测(耗时较长,生产环境建议用 Dask 并行) # backtest_df = rolling_backtest(raw_df, models, build_features) # print("区间覆盖率:", backtest_df['in_interval'].mean()) # print("平均绝对误差:", backtest_df['error'].mean())关键指标解读:
- 区间覆盖率 > 80%:说明模型不确定性量化合理;
- 误差序列无自相关(Ljung-Box 检验 p>0.05):说明模型未残留系统性偏差;
- 权重分布稳定(如 LR 权重始终在 0.2~0.4):说明没有模型长期失效。
5.3 一个值得坚持的实战习惯:永远保存「预测-真值-区间」三元组到 CSV
我经手的每个股价预测项目,上线第一天就写这个函数:
def save_prediction_log( date: str, pred: float, true: float, interval_low: float, interval_high: float, model_weights: list, features_used: list, log_file: str = "prediction_log.csv" ): log_entry = { 'date': date, 'pred_price': round(pred, 4), 'true_price': round(true, 4), 'interval_low': round(interval_low, 4), 'interval_high': round(interval_high, 4), 'coverage': 1 if (interval_low <= true <= interval_high) else 0, 'abs_error': round(abs(pred - true), 4), 'lr_weight': round(model_weights[0], 3), 'knn_weight': round(model_weights[1], 3), 'arima_weight': round(model_weights[2], 3), 'lstm_weight': round(model_weights[3], 3), 'features': '|'.join(features_used) } df_log = pd.DataFrame([log_entry]) if not os.path.exists(log_file): df_log.to_csv(log_file, index=False) else: df_log.to_csv(log_file, mode='a', header=False, index=False) # 每日预测后调用 # save_prediction_log( # date="2023-12-01", # pred=1825.3, # true=1831.7, # interval_low=1812.4, # interval_high=1838.9, # model_weights=[0.25, 0.18, 0.32, 0.25], # features_used=['price_lag1','rsi_14','vol_ratio'] # )为什么重要?三个月后,当你发现某类行情下 LSTM 权重持续归零
本文还有配套的精品资源,点击获取