☰
股票价格预测源码实战:从数据管道到模型评估的完整拆解
2026/10/13 12:42:26 网站建设 项目流程

简介:这份资源是面向计算机、人工智能、大数据、数学及电子信息等专业学生与技术学习者的机器学习实战项目源码,聚焦股票市场价格的分析与预测场景,可作为课程设计、期末大作业或毕业设计的参考方案。压缩包共2个文件,以1个Python脚本和1个txt说明文件为主,整体约45KB,体量轻便,便于快速下载与本地调试。其中Python脚本承担数据读取、特征处理与模型训练预测的核心逻辑,txt文件则用于补充依赖或运行说明,配合requirements.txt可辅助还原运行环境。目前已有299人学习下载,说明该方向具备一定的关注度。读者可从中获取一套完整的股票价格预测实现思路,理解机器学习算法在金融时序数据上的落地方式,并在此基础上进行参数调整、模型替换或功能扩展,适合具备一定编程基础、希望将算法理论转化为实际项目经验的学习者参考使用。

1. 股票价格预测源码拆解:从一份 zip 到能跑通的基线模型

拿到「基于机器学习算法的股票市场股票价格的分析及预测源码.zip」这类压缩包,多数人的第一反应是解压、找requirements.txt、pip install、然后python main.py。但真正跑过股票预测项目的人都知道,这一步大概率会翻车——不是缺数据,就是模型输出全是 NaN,再不然就是预测曲线和真实价格完全对不上。问题不在代码本身,而在于股票价格预测这件事,从数据获取、特征构造到模型评估,每一步都有它自己的脾气。

这份源码对应的技术方向,是用机器学习做股票价格的回归预测。核心链路通常包括:历史行情数据获取、技术指标计算、特征工程、模型训练、回测评估。适合两类人:一是想拿一个完整项目练手机器学习流程的入门者,二是想快速搭一个基线模型再往上叠策略的量化从业者。下面按「数据怎么来、特征怎么造、模型怎么选、坑怎么避」的顺序,把这条链路拆开讲清楚。

2. 数据管道与特征工程:股票价格预测的原料从哪来

2.1 行情数据获取的三种常见方式

股票价格预测的第一步永远是拿到干净的历史数据。常见做法有三种:

第一种是用yfinance这类开源库直接拉取。优点是零成本、接口简单,缺点是 A 股数据覆盖不全,美股数据偶尔有拆股复权问题。第二种是用akshare或tushare拉 A 股数据,前者免费但接口稳定性一般,后者需要注册拿 token,积分门槛逐年提高。第三种是本地 CSV 文件,很多源码包里会附带一份data/stock.csv,格式通常是日期、开盘、最高、最低、收盘、成交量六列。

我一般会先检查源码包里有没有自带数据文件。如果有,先读进来看看字段名和日期范围;如果没有,再决定用哪个库拉。下面是一个用akshare拉取单只股票日线数据的示例:

import akshare as ak import pandas as pd # 拉取贵州茅台日线数据,前复权 df = ak.stock_zh_a_hist( symbol="600519", period="daily", start_date="20200101", end_date="20241231", adjust="qfq" # 前复权,避免除权除息导致价格跳空 ) # 统一列名,方便后续处理 df = df.rename(columns={ "日期": "date", "开盘": "open", "收盘": "close", "最高": "high", "最低": "low", "成交量": "volume" }) df["date"] = pd.to_datetime(df["date"]) df = df.sort_values("date").reset_index(drop=True) print(df.head())

这段代码的关键在adjust="qfq"这个参数。股票在除权除息日会出现价格跳空,如果不做复权处理,模型会把这种跳空当成真实的价格波动来学习,导致预测偏差。前复权是把历史价格按最新股本调整,适合做时序预测;后复权适合做收益率计算。很多源码包默认不复权,这是第一个要检查的点。

2.2 技术指标构造:把原始价格变成模型能吃的特征

原始的开高低收量五列,直接喂给模型效果通常很差。原因是价格序列是非平稳的,均值和方法随时间漂移,模型很难学到稳定模式。常见做法是构造技术指标和滞后特征。

最基础的几个特征包括:收益率(close.pct_change())、移动平均线(MA5、MA10、MA20)、波动率(滚动标准差)、成交量变化率。再进阶一点可以加 RSI、MACD、布林带位置。下面是一个特征构造的代码片段:

import numpy as np # 日收益率 df["ret"] = df["close"].pct_change() # 移动平均线 for window in [5, 10, 20]: df[f"ma_{window}"] = df["close"].rolling(window).mean() df[f"vol_{window}"] = df["ret"].rolling(window).std() # 成交量比率 df["vol_ratio"] = df["volume"] / df["volume"].rolling(5).mean() # 滞后特征:用前 N 天的收盘价预测下一天 for lag in [1, 2, 3, 5]: df[f"close_lag_{lag}"] = df["close"].shift(lag) # 预测目标:下一天的收盘价 df["target"] = df["close"].shift(-1) # 去掉因滚动和滞后产生的空值 df = df.dropna().reset_index(drop=True)

这里有几个参数需要根据数据频率调整。window的选择上,日线数据常用 5、10、20,分钟线数据要缩小到 3、5、10。lag的数量不宜过多,超过 5 个容易引入多重共线性,树模型对此不敏感,但线性模型会出问题。target用shift(-1)表示用今天的所有特征预测明天的收盘价,这是最常见的设定,但要注意最后一行会变成 NaN,必须 drop 掉。

注意:特征构造阶段最容易犯的错误是引入未来信息。比如用rolling().mean()时没有加shift(1),导致当天的均值包含了当天的收盘价,而当天收盘价又是预测目标的一部分。这种数据泄漏会让回测结果虚高,实盘一跑就亏。

2.3 训练集与测试集的切分方式

股票时序数据不能随机切分。常见做法是按时间顺序切,比如前 80% 做训练,后 20% 做测试。更严谨的做法是滚动窗口切分:用前 N 天训练,预测第 N+1 天,然后窗口往后滑一天。下面是一个时间序列切分的示例:

from sklearn.model_selection import TimeSeriesSplit # 按时间顺序切分,不 shuffle split_idx = int(len(df) * 0.8) train = df.iloc[:split_idx] test = df.iloc[split_idx:] # 特征列和目标列 feature_cols = [c for c in df.columns if c not in ["date", "target"]] X_train, y_train = train[feature_cols], train["target"] X_test, y_test = test[feature_cols], test["target"] print(f"训练集: {X_train.shape}, 测试集: {X_test.shape}")

如果源码包里用的是train_test_split(shuffle=True),那基本可以判定这个项目的回测结果不可信。时序数据的自相关性会让随机切分把未来信息泄漏到训练集,模型在测试集上的表现会好得离谱,但没有任何实盘价值。

3. 模型选型与训练:线性回归、树模型和 LSTM 的取舍

3.1 为什么基线模型优先选树模型而不是 LSTM

很多股票预测源码包一上来就用 LSTM,理由是「时序数据适合循环神经网络」。这个逻辑听起来对,但实操中 LSTM 在股票日线数据上的表现往往不如 XGBoost 或 LightGBM。原因有三个:第一,股票日线数据量通常只有几千条,LSTM 参数量大,容易过拟合;第二,LSTM 对超参数敏感,学习率、层数、隐藏单元数调起来很费时间;第三,树模型对特征尺度不敏感,不需要做归一化,工程上更省事。

我一般会先用 LightGBM 跑一个基线,确认特征和标签的构造没问题,再考虑要不要上深度学习。下面是一个 LightGBM 回归的示例:

import lightgbm as lgb from sklearn.metrics import mean_squared_error, mean_absolute_error # 构造 Dataset train_data = lgb.Dataset(X_train, label=y_train) valid_data = lgb.Dataset(X_test, label=y_test, reference=train_data) # 参数设置 params = { "objective": "regression", "metric": "rmse", "boosting_type": "gbdt", "num_leaves": 31, # 叶子数,数据量小就调小 "learning_rate": 0.05, # 学习率,配合 early_stopping "feature_fraction": 0.8, # 每棵树随机选 80% 特征 "bagging_fraction": 0.8, # 每轮随机选 80% 样本 "bagging_freq": 5, "verbose": -1 } # 训练 model = lgb.train( params, train_data, num_boost_round=1000, valid_sets=[valid_data], callbacks=[lgb.early_stopping(50), lgb.log_evaluation(100)] ) # 预测 y_pred = model.predict(X_test, num_iteration=model.best_iteration) rmse = mean_squared_error(y_test, y_pred, squared=False) mae = mean_absolute_error(y_test, y_pred) print(f"RMSE: {rmse:.4f}, MAE: {mae:.4f}")

num_leaves是最关键的参数。数据量在 5000 条以下时,建议设在 15 到 31 之间;超过 10000 条可以放到 63。learning_rate设 0.05 配合early_stopping(50)是比较稳的组合,如果发现模型收敛太慢,可以提到 0.1,但不要超过 0.2。feature_fraction和bagging_fraction用来控制过拟合,股票数据噪声大,这两个值建议设在 0.7 到 0.9 之间。

3.2 线性回归作为 sanity check 的价值

在跑复杂模型之前,我习惯先用线性回归跑一遍。不是为了追求精度,而是为了验证特征和目标之间是否存在线性关系。如果线性回归的 RMSE 和 LightGBM 差不太多,说明特征构造有问题,可能大部分特征都是噪声。如果线性回归的 RMSE 是 LightGBM 的两三倍,说明非线性模型确实抓到了额外模式,值得继续调。

from sklearn.linear_model import Ridge from sklearn.preprocessing import StandardScaler # 线性模型需要标准化 scaler = StandardScaler() X_train_scaled = scaler.fit_transform(X_train) X_test_scaled = scaler.transform(X_test) ridge = Ridge(alpha=1.0) ridge.fit(X_train_scaled, y_train) y_pred_ridge = ridge.predict(X_test_scaled) rmse_ridge = mean_squared_error(y_test, y_pred_ridge, squared=False) print(f"Ridge RMSE: {rmse_ridge:.4f}")

alpha是正则化强度,股票数据噪声大,建议从 1.0 起步,如果发现系数波动很大就调大,如果欠拟合就调小。标准化必须用训练集的均值和方差,不能把测试集一起 fit,否则又是数据泄漏。

3.3 LSTM 的正确打开方式

如果确实要上 LSTM,有几个参数必须调对。第一,输入序列长度(seq_len)建议设在 10 到 30 之间,对应两周到一个月的历史窗口。第二,LSTM 层数不要超过两层,隐藏单元数控制在 32 到 128 之间。第三,必须加 Dropout,比例设在 0.2 到 0.5。第四,输出层用线性激活,因为预测的是连续价格。

import torch import torch.nn as nn class LSTMModel(nn.Module): def __init__(self, input_size, hidden_size=64, num_layers=2, dropout=0.3): super().__init__() self.lstm = nn.LSTM( input_size, hidden_size, num_layers, batch_first=True, dropout=dropout ) self.fc = nn.Linear(hidden_size, 1) def forward(self, x): # x shape: (batch, seq_len, input_size) out, _ = self.lstm(x) # 取最后一个时间步的输出 out = self.fc(out[:, -1, :]) return out.squeeze() # 构造序列数据 def create_sequences(X, y, seq_len=20): xs, ys = [], [] for i in range(len(X) - seq_len): xs.append(X[i:i+seq_len]) ys.append(y[i+seq_len]) return np.array(xs), np.array(ys) X_seq, y_seq = create_sequences(X_train.values, y_train.values, seq_len=20) print(f"序列数据形状: {X_seq.shape}")

LSTM 的训练需要 GPU 才跑得快,如果只有 CPU,训练时间会很长。另外 LSTM 对学习率非常敏感,建议用1e-3起步,配合ReduceLROnPlateau调度器。如果发现 loss 震荡不收敛,先检查数据有没有做归一化,LSTM 对输入尺度比树模型敏感得多。

4. 回测与评估:为什么你的 RMSE 很低但实盘亏钱

4.1 价格预测和收益率预测的本质区别

很多源码包直接预测收盘价,然后用 RMSE 评估。问题在于,股票价格是非平稳序列,今天 100 块明天 101 块,后天 99 块,模型只要预测「和昨天差不多」就能拿到很低的 RMSE。但这种模型没有任何交易价值,因为它学到的只是价格的持续性,而不是方向性。

更合理的做法是预测收益率,或者把价格预测转成方向预测。下面是一个把价格预测转成方向准确率的评估代码:

# 用预测价格和真实价格算方向准确率 y_pred_price = y_pred y_true_price = y_test.values # 方向:今天预测价 vs 昨天真实价 pred_direction = np.sign(y_pred_price - df["close"].iloc[split_idx:split_idx+len(y_pred)].values) true_direction = np.sign(y_true_price - df["close"].iloc[split_idx:split_idx+len(y_pred)].values) direction_acc = (pred_direction == true_direction).mean() print(f"方向准确率: {direction_acc:.4f}")

方向准确率能到 55% 以上,说明模型有实用价值;低于 52% 基本就是随机猜测。很多源码包只报 RMSE 不报方向准确率,这是评估体系不完整。

4.2 回测中的交易成本与滑点

如果要把预测信号转成交易策略,必须考虑交易成本和滑点。A 股单边手续费大约万分之三,加上印花税千分之一(卖出时收),一次完整买卖的成本大约千分之一点六。滑点按千分之一估算。如果模型预测的日收益率只有千分之二,扣掉成本基本不赚钱。

# 简单回测:预测涨就买入,预测跌就空仓 initial_capital = 100000 capital = initial_capital position = 0 cost_rate = 0.0016 # 单次交易成本 for i in range(len(y_pred)): signal = 1 if y_pred[i] > y_true_price[i-1] else 0 if signal == 1 and position == 0: position = capital / y_true_price[i] capital = 0 capital -= position * y_true_price[i] * cost_rate elif signal == 0 and position > 0: capital = position * y_true_price[i] capital -= capital * cost_rate position = 0 final_value = capital + position * y_true_price[-1] print(f"最终资金: {final_value:.2f}, 收益率: {(final_value/initial_capital-1)*100:.2f}%")

这段回测代码很粗糙,但能说明一个问题:如果策略的收益率扣掉成本后还不如买入持有,那这个模型就没有实盘价值。很多源码包的回测曲线很漂亮,但仔细一看没有扣成本,或者用了未来函数。

4.3 滚动回测比单次切分更可信

单次训练集测试集切分只能看到一个时间段的表現。更严谨的做法是滚动回测:每次用过去 N 天训练,预测未来 M 天,然后窗口往前滑。下面是一个滚动回测的框架:

def walk_forward_validation(df, feature_cols, train_window=500, test_window=20): results = [] for start in range(0, len(df) - train_window - test_window, test_window): train = df.iloc[start:start+train_window] test = df.iloc[start+train_window:start+train_window+test_window] model = lgb.train( params, lgb.Dataset(train[feature_cols], label=train["target"]), num_boost_round=200 ) pred = model.predict(test[feature_cols]) rmse = mean_squared_error(test["target"], pred, squared=False) results.append({"start": start, "rmse": rmse}) return pd.DataFrame(results) results = walk_forward_validation(df, feature_cols) print(results["rmse"].describe())

滚动回测的 RMSE 分布比单次切分更有参考价值。如果 RMSE 的方差很大,说明模型在不同市场环境下表现不稳定,需要检查特征是否对市场状态敏感。

5. 避坑与排查:股票预测源码里最常见的五个翻车点

5.1 数据泄漏:回测收益虚高的头号原因

现象:回测曲线平滑向上,年化收益 50% 以上,但实盘一跑就亏。

原因:特征构造时用了未来信息。最常见的是rolling().mean()没有加shift(1),或者用了全量数据的均值和方差做标准化。

解决:所有滚动计算必须shift(1),标准化只能在训练集上 fit。检查方法是把特征和目标的相关性算出来,如果某个特征和目标的相关性超过 0.9,大概率有泄漏。

5.2 复权处理不当:价格跳空被当成真实波动

现象:模型在除权除息日附近预测误差突然变大。

原因:数据没有做前复权或后复权处理,除权日的价格跳空被模型当成真实下跌或上涨。

解决:拉数据时加adjust="qfq",或者用akshare的复权接口。如果源码包自带数据,先检查除权日附近的价格是否连续。

5.3 过拟合:训练集 RMSE 远低于测试集

现象:训练集 RMSE 0.5,测试集 RMSE 5.0,差距十倍以上。

原因:模型太复杂,或者特征太多但样本太少。LSTM 和深层树模型最容易出这个问题。

解决:先降模型复杂度,num_leaves调到 15 以下,LSTM 层数降到一层。再减特征,用feature_importance排序,只保留前 20 个。最后加正则化,lambda_l1和lambda_l2设到 0.1 到 1.0。

5.4 标签构造错误:预测目标和特征错位

现象:模型预测值永远比真实值滞后一天。

原因:target用shift(-1)构造,但特征里包含了当天的收盘价,而当天收盘价在预测时是不可知的。

解决:特征里不能包含预测目标当天的任何信息。如果预测明天收盘价,特征只能用到今天收盘为止的数据。检查方法是把target和close做相关性,如果相关性 0.99 以上,说明特征里混入了目标。

5.5 评估指标单一:只看 RMSE 不看方向准确率

现象:RMSE 很低,但按预测信号交易不赚钱。

原因:RMSE 衡量的是价格拟合程度,不是方向判断能力。价格序列自相关性强,模型只要预测「和昨天差不多」就能拿到低 RMSE。

解决:同时看 RMSE、MAE、方向准确率、扣费后收益率。方向准确率低于 52% 的模型直接放弃,不要浪费时间调参。

6. 从基线到可用:三个提升预测质量的实操技巧

6.1 用分类代替回归做方向预测

把「预测明天收盘价」改成「预测明天涨还是跌」,问题就从回归变成分类。分类模型对噪声的鲁棒性更强,评估指标也更直接。下面是一个用 LightGBM 做二分类的示例:

# 构造分类标签:明天涨为 1,跌为 0 df["label"] = (df["close"].shift(-1) > df["close"]).astype(int) df = df.dropna().reset_index(drop=True) # 切分 split_idx = int(len(df) * 0.8) train, test = df.iloc[:split_idx], df.iloc[split_idx:] feature_cols = [c for c in df.columns if c not in ["date", "label", "target"]] # 分类参数 clf_params = { "objective": "binary", "metric": "auc", "num_leaves": 15, "learning_rate": 0.05, "feature_fraction": 0.8, "bagging_fraction": 0.8, "bagging_freq": 5, "verbose": -1 } clf = lgb.train( clf_params, lgb.Dataset(train[feature_cols], label=train["label"]), num_boost_round=500, valid_sets=[lgb.Dataset(test[feature_cols], label=test["label"])], callbacks=[lgb.early_stopping(50)] ) # 评估 AUC 和准确率 from sklearn.metrics import roc_auc_score, accuracy_score pred_prob = clf.predict(test[feature_cols]) pred_label = (pred_prob > 0.5).astype(int) print(f"AUC: {roc_auc_score(test['label'], pred_prob):.4f}") print(f"准确率: {accuracy_score(test['label'], pred_label):.4f}")

AUC 能到 0.55 以上就说明模型有区分能力,0.6 以上算不错。分类模型的好处是可以直接输出概率,方便做仓位管理——概率 0.6 时半仓,概率 0.8 时满仓。

6.2 多模型融合降低单模型方差

单个模型的预测波动大,把 LightGBM、XGBoost、Ridge 的预测结果做加权平均,能显著降低方差。权重可以用验证集上的表现来定,表现好的权重大。

from sklearn.ensemble import VotingRegressor from xgboost import XGBRegressor # 三个基模型 lgb_model = lgb.LGBMRegressor(num_leaves=15, learning_rate=0.05, n_estimators=300) xgb_model = XGBRegressor(max_depth=4, learning_rate=0.05, n_estimators=300) ridge_model = Ridge(alpha=1.0) # 融合 ensemble = VotingRegressor([ ("lgb", lgb_model), ("xgb", xgb_model), ("ridge", ridge_model) ], weights=[0.5, 0.3, 0.2]) ensemble.fit(X_train, y_train) y_pred_ensemble = ensemble.predict(X_test) rmse_ensemble = mean_squared_error(y_test, y_pred_ensemble, squared=False) print(f"融合模型 RMSE: {rmse_ensemble:.4f}")

权重不是固定的,可以根据市场状态动态调整。比如波动率高的时期,树模型的权重调大;波动率低的时期,线性模型的权重调大。

6.3 用特征重要性做特征筛选

特征不是越多越好。股票预测里,很多技术指标之间高度相关,保留太多反而引入噪声。用feature_importance排序,只保留前 20 到 30 个特征,模型表现通常更稳定。

# 获取特征重要性 importance = pd.DataFrame({ "feature": feature_cols, "importance": clf.feature_importance(importance_type="gain") }).sort_values("importance", ascending=False) # 选前 20 个 top_features = importance.head(20)["feature"].tolist() print(importance.head(10)) # 用筛选后的特征重新训练 clf_top = lgb.train( clf_params, lgb.Dataset(train[top_features], label=train["label"]), num_boost_round=500 ) pred_top = clf_top.predict(test[top_features]) print(f"筛选后 AUC: {roc_auc_score(test['label'], pred_top):.4f}")

我一般会跑两轮:第一轮用全部特征,看重要性分布;第二轮只用前 20 个,对比 AUC 有没有下降。如果下降不超过 0.01,就用筛选后的特征,模型更轻量,推理更快。

最后说一个我自己的习惯:每次拿到新的股票预测源码,先不看模型部分,先看数据管道和评估代码。数据管道决定模型能学到什么,评估代码决定你能不能相信结果。这两块没问题,模型部分再差也差不到哪去;这两块有问题,模型再花哨也是空中楼阁。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询