简介本资源是一套基于Python实现的股票预测系统源码面向金融数据分析初学者、量化交易爱好者及高校相关专业学生旨在通过历史行情数据建模辅助投资决策。包内共77个文件含43个Python脚本涵盖数据预处理、策略生成、模型训练与结果可视化等核心模块、21个CSV历史行情数据如eth_usdt_5m_okxdata系列、BankOfChinaStrategy对应标的等、7个文本配置与日志文件、2个Excel标的参数表、2张PNG分析图表以及LICENSE、.gitignore和readme等工程规范文件压缩包大小为89.28MB。已有367人学习下载资源结构清晰策略模块化程度高——如btcStrategyDuo.py、ethStrategy5mRandom.py、ZiguangStrategyMA.py等文件分别封装了多周期、多标的、多算法的预测逻辑配合autoExportTrain.py、handExportTrain.py等训练调度脚本便于读者快速复现、对比与迭代优化模型。1. 这不是“预测股价”而是构建一个可验证、可迭代、能落地的股票时序建模闭环很多刚接触量化的朋友看到“股票预测系统”第一反应是能不能明天涨停——答案是否定的。真正有价值的Python股票预测系统本质是一个围绕金融时间序列特性构建的数据驱动建模闭环从原始行情数据清洗、特征工程设计如量价关系、波动率结构、多周期动量、模型选择LSTM/Transformer/XGBoost并非万能需匹配样本长度与噪声水平、回测框架嵌入避免未来信息泄露到预测结果的业务解释如方向概率、置信区间、风险阈值触发。它不承诺收益但能系统性降低主观判断偏差为交易策略提供可复现的信号生成模块。适合有Python基础、理解基本统计概念、愿意花时间调试数据管道的量化入门者与策略研究员不适合期待“一键暴富”的零基础用户。本系统设计强调可复现性所有依赖明确版本、可审计性每步输出中间结果、可替换性模型/特征/评估模块解耦源码结构遵循data → feature → model → backtest → serve五层分治逻辑而非堆砌算法黑箱。2. 用pandasTA-Lib构建抗噪行情数据管道从原始OHLCV到结构化特征矩阵2.1 原始数据获取与标准化统一时间戳、处理缺失与异常值股票预测的起点不是模型而是干净、对齐、带业务语义的时序数据。我们不依赖第三方API实时拉取易受限且不稳定而是以本地CSV或SQLite存储日线数据确保复现性。关键在于三步标准化时间对齐强制所有股票使用同一交易日历如pandas_market_calendars.get_calendar(XSHG)将非交易日填充为NaN而非前向填充避免引入虚假连续性价格一致性对复权因子做累积乘积校准确保close_adj close * cumprod(adj_factor)而非简单除权异常值过滤用滚动Z-score窗口20日识别单日涨跌幅8σ的点标记为is_outlier1后续特征工程中屏蔽该样本。import pandas as pd import numpy as np import sqlite3 def load_and_clean_data(ticker: str, db_path: str) - pd.DataFrame: conn sqlite3.connect(db_path) # SQL确保按时间升序且只取有效交易日 sql f SELECT trade_date, open, high, low, close, volume, adj_factor FROM stock_daily WHERE symbol ? AND trade_date 2018-01-01 ORDER BY trade_date ASC df pd.read_sql(sql, conn, params(ticker,), parse_dates[trade_date]) conn.close() # 复权处理关键 df[close_adj] df[close] * df[adj_factor].cumprod() df[open_adj] df[open] * df[adj_factor].shift(1).fillna(1).cumprod() # 计算滚动Z-score并标记异常 df[ret] df[close_adj].pct_change() df[zscore_ret] df[ret].rolling(20).apply( lambda x: (x[-1] - x.mean()) / x.std() if len(x) 5 else np.nan ) df[is_outlier] (df[zscore_ret].abs() 8).astype(int) return df.set_index(trade_date).sort_index() # 示例调用 df_raw load_and_clean_data(600519.SH, stock_data.db) print(f原始数据形状: {df_raw.shape}, 异常点占比: {df_raw[is_outlier].mean():.2%})提示adj_factor必须是逐日调整因子非累计否则复权结果错误。若数据源无复权字段需用akshare等库补全但务必验证复权后前复权价与后复权价逻辑一致。2.2 特征工程用TA-Lib生成技术指标 自定义量价结构特征TA-Lib是金融特征生成的工业级标准但直接调用ta.SMA()易忽略参数敏感性。我们封装为可配置的特征生成器重点解决两个痛点指标滞后性对齐与多周期特征融合。滞后对齐所有指标计算后用shift(-1)将当日指标值对齐到次日预测目标因预测明日收盘价特征需基于今日及历史数据结构化组合不单独用MACD而是构造macd_hist_ratio macd_hist / (high - low)捕捉波动率归一化的动量强度。import talib as ta def generate_features(df: pd.DataFrame) - pd.DataFrame: df_feat df.copy() # 基础价格特征已复权 df_feat[high_low_ratio] df_feat[high_adj] / df_feat[low_adj] df_feat[close_open_ratio] df_feat[close_adj] / df_feat[open_adj] # TA-Lib指标窗口参数显式声明避免默认值陷阱 df_feat[sma_5] ta.SMA(df_feat[close_adj], timeperiod5) df_feat[sma_20] ta.SMA(df_feat[close_adj], timeperiod20) df_feat[rsi_14] ta.RSI(df_feat[close_adj], timeperiod14) # MACD需三值分离再构造结构特征 macd, signal, hist ta.MACD(df_feat[close_adj], fastperiod12, slowperiod26, signalperiod9) df_feat[macd_hist] hist df_feat[macd_hist_ratio] hist / (df_feat[high_adj] - df_feat[low_adj]) # 波动率特征关键 df_feat[volatility_10] df_feat[close_adj].pct_change().rolling(10).std() df_feat[volume_ma_ratio] df_feat[volume] / df_feat[volume].rolling(20).mean() # 所有特征向后平移1位对齐预测目标 feat_cols [c for c in df_feat.columns if c not in [close_adj, is_outlier]] df_feat[feat_cols] df_feat[feat_cols].shift(1) return df_feat df_feat generate_features(df_raw) print(特征列数:, len([c for c in df_feat.columns if c.startswith((sma_, rsi_, macd_, volatility_))]))注意TA-Lib安装需提前编译pip install TA-Lib在Windows上常失败推荐用conda install -c conda-forge ta-lib。若无法安装可用pandas_ta替代但需验证pandas_ta.macd()与TA-Lib结果误差0.1%。2.3 特征矩阵构建处理缺失值、标准化与目标变量定义金融数据天然存在大量缺失新股上市初期、停牌直接删除会丢失关键样本。我们采用分层填充策略数值型特征用rolling median窗口30日填充比均值更抗异常值分类特征如有用mode填充目标变量定义为next_close_direction sign(close_t1 - close_t)即二分类方向预测比回归更鲁棒。def build_feature_matrix(df: pd.DataFrame, target_col: str close_adj) - tuple: # 定义特征列排除原始价格、索引、标记列 feature_cols [c for c in df.columns if c not in [open_adj, high_adj, low_adj, close_adj, volume, adj_factor, is_outlier, ret, zscore_ret]] X df[feature_cols].copy() y np.sign(df[target_col].shift(-1) - df[target_col]) # 次日方向 # 分层填充缺失值 for col in X.select_dtypes(include[np.number]).columns: X[col] X[col].fillna(X[col].rolling(30, min_periods5).median()) # 标准化仅对数值型保留原始分布形态 from sklearn.preprocessing import StandardScaler scaler StandardScaler() X_scaled pd.DataFrame( scaler.fit_transform(X.select_dtypes(include[np.number])), columnsX.select_dtypes(include[np.number]).columns, indexX.index ) # 对齐X与y去除首尾无效行 valid_idx X_scaled.dropna().index.intersection(y.dropna().index) X_final X_scaled.loc[valid_idx] y_final y.loc[valid_idx].astype(int) # -1, 0, 1 → 转为整数 return X_final, y_final, scaler X, y, scaler build_feature_matrix(df_feat) print(f最终特征矩阵形状: {X.shape}, 标签分布: {y.value_counts().to_dict()})3. 模型选型与训练XGBoost作为基线LSTM处理长周期依赖的实操对比3.1 XGBoost高解释性、低过拟合的强基线模型XGBoost在金融时序中表现稳定因其内置正则化lambda,alpha和列采样colsample_bytree天然抑制过拟合。关键参数设置原则n_estimators300足够学习复杂模式但不过度拟合短期噪声max_depth5限制树深度防止记忆局部波动learning_rate0.05小学习率配合早停提升泛化性eval_metriclogloss适配二分类任务比error更敏感。from xgboost import XGBClassifier from sklearn.model_selection import TimeSeriesSplit from sklearn.metrics import classification_report, confusion_matrix # 时间序列交叉验证避免未来信息泄露 tscv TimeSeriesSplit(n_splits5) model_xgb XGBClassifier( n_estimators300, max_depth5, learning_rate0.05, subsample0.8, colsample_bytree0.8, reg_alpha0.1, # L1正则 reg_lambda1.0, # L2正则 eval_metriclogloss, random_state42, use_label_encoderFalse ) # 按时间顺序划分训练/验证集 train_size int(len(X) * 0.7) X_train, X_val X.iloc[:train_size], X.iloc[train_size:] y_train, y_val y.iloc[:train_size], y.iloc[train_size:] # 训练并预测 model_xgb.fit(X_train, y_train, eval_set[(X_train, y_train), (X_val, y_val)], early_stopping_rounds50, verboseFalse) y_pred_xgb model_xgb.predict(X_val) print(XGBoost验证集报告:) print(classification_report(y_val, y_pred_xgb))提示TimeSeriesSplit比KFold更合理但需注意其分割方式——每次验证集都是前一次训练集之后的连续片段符合真实交易场景。3.2 LSTM捕获长周期时序依赖的PyTorch实现当需要建模跨月价格惯性如趋势延续性时LSTM优于XGBoost。但直接套用会失败金融序列信噪比低需叠加注意力机制与DropPath正则。我们采用轻量级架构输入滑动窗口window_size60覆盖季度周期隐藏层单层LSTM Attention Layer计算各时间步权重输出3分类涨/跌/横盘Softmax。import torch import torch.nn as nn from torch.utils.data import Dataset, DataLoader class StockDataset(Dataset): def __init__(self, X: pd.DataFrame, y: pd.Series, window_size: int 60): self.X torch.tensor(X.values, dtypetorch.float32) self.y torch.tensor(y.values, dtypetorch.long) self.window_size window_size def __len__(self): return len(self.X) - self.window_size def __getitem__(self, idx): x_window self.X[idx:idxself.window_size] y_label self.y[idxself.window_size] return x_window, y_label class LSTMWithAttention(nn.Module): def __init__(self, input_dim, hidden_dim, num_classes, dropout0.3): super().__init__() self.lstm nn.LSTM(input_dim, hidden_dim, batch_firstTrue, dropoutdropout) self.attention nn.Sequential( nn.Linear(hidden_dim, hidden_dim), nn.Tanh(), nn.Linear(hidden_dim, 1) ) self.classifier nn.Sequential( nn.Dropout(dropout), nn.Linear(hidden_dim, num_classes) ) def forward(self, x): lstm_out, _ self.lstm(x) # [batch, seq, hidden] # Attention权重 attn_weights torch.softmax(self.attention(lstm_out), dim1) context torch.sum(attn_weights * lstm_out, dim1) # [batch, hidden] return self.classifier(context) # 数据加载 dataset StockDataset(X, y, window_size60) train_size int(len(dataset) * 0.7) train_dataset, val_dataset torch.utils.data.random_split( dataset, [train_size, len(dataset)-train_size] ) train_loader DataLoader(train_dataset, batch_size32, shuffleFalse) # 时间序列不shuffle val_loader DataLoader(val_dataset, batch_size32, shuffleFalse) # 模型训练省略完整训练循环聚焦关键参数 model_lstm LSTMWithAttention( input_dimX.shape[1], hidden_dim64, num_classes3, dropout0.3 ) optimizer torch.optim.Adam(model_lstm.parameters(), lr0.001) criterion nn.CrossEntropyLoss() # 训练后验证代码略重点看验证集F1-score注意LSTM训练需GPU加速且batch_size不宜过大易过拟合hidden_dim64在60日窗口下已足够。若验证集准确率低于XGBoost说明当前特征未体现长周期依赖应检查window_size或增加波动率特征。4. 回测引擎嵌入用Backtrader实现信号-仓位-风控闭环验证4.1 将预测信号转化为可执行交易策略模型输出只是概率需映射为具体交易动作。我们定义三档信号规则signal 1预测上涨开多仓仓位min(0.5, prob_up * 0.8)概率加权上限50%signal -1预测下跌开空仓同理signal 0横盘平仓保持现金。关键约束单次交易手续费0.03%、滑点0.1%、最大回撤20%自动止损。import backtrader as bt class PredictionStrategy(bt.Strategy): params ( (commission_pct, 0.0003), (slippage_pct, 0.001), (max_drawdown, 0.2), ) def __init__(self): self.prediction self.datas[0].prediction # 从数据加载时注入预测值 self.order None def next(self): # 获取当前预测信号假设pred为-1,0,1 pred self.prediction[0] # 当前bar的预测 pos_size self.broker.getvalue() * 0.5 # 最大仓位50% # 平仓逻辑 if self.position and pred 0: self.close() return # 开仓逻辑 if not self.position: if pred 1: # 看涨 size int(pos_size / self.data.close[0]) self.buy(sizesize) elif pred -1: # 看跌 size int(pos_size / self.data.close[0]) self.sell(sizesize) def stop(self): # 记录最终资产 self.log(fFinal Value: {self.broker.getvalue():.2f}) # 加载预测结果到数据源 data bt.feeds.PandasData(datanamedf_feat.reset_index()) data.prediction y_pred_xgb # 注入XGBoost预测序列 cerebro bt.Cerebro() cerebro.adddata(data) cerebro.addstrategy(PredictionStrategy) cerebro.broker.setcash(100000.0) cerebro.broker.setcommission(commissionself.params.commission_pct) cerebro.addanalyzer(bt.analyzers.DrawDown, _namedrawdown) results cerebro.run()4.2 回测结果解析超越胜率的关键指标回测不能只看总收益需分析策略健壮性。核心指标解读夏普比率 1.0单位风险超额收益达标最大回撤 15%风控有效盈亏比Profit Factor 1.5盈利交易总收益/亏损交易总损失月度胜率稳定性连续3个月胜率40%需触发模型重训。# 提取分析器结果 strat results[0] drawdown strat.analyzers.drawdown.get_analysis() print(f总收益率: {(cerebro.broker.getvalue() / 100000 - 1)*100:.2f}%) print(f最大回撤: {drawdown.max.drawdown:.2f}%) print(f夏普比率: {strat.analyzers.sharperatio.get_analysis()[sharperatio]:.2f}) # 生成交易日志用于归因分析 trades strat.analyzers.trades.get_analysis() for trade in trades: print(f交易日期: {trade.open_datetime}, 类型: {trade.type}, f盈亏: {trade.pnlcomm:.2f}, 持仓天数: {trade.barlen})提示Backtrader默认不支持多空同时持仓若需对冲策略需继承bt.Strategy重写notify_order方法手动管理多空头寸净值。5. 模型服务化与持续监控用Flask暴露预测API并设置漂移告警5.1 构建轻量级预测API输入股票代码返回明日方向概率将训练好的XGBoost模型封装为REST API关键要求输入校验、缓存最近预测、异步更新特征。避免每次请求都重新计算TA-Lib指标。from flask import Flask, request, jsonify import joblib import pandas as pd app Flask(__name__) # 加载模型与标准化器 model joblib.load(xgb_model.pkl) scaler joblib.load(scaler.pkl) # 特征生成函数复用前述generate_features app.route(/predict, methods[POST]) def predict(): data request.json ticker data.get(ticker) if not ticker: return jsonify({error: Missing ticker}), 400 try: # 1. 加载最新数据从本地DB df load_and_clean_data(ticker, stock_data.db) # 2. 生成特征仅最新1条避免全量计算 latest_feat generate_features(df).iloc[[-1]] # 取最后一行 # 3. 标准化 X_new scaler.transform(latest_feat.select_dtypes(include[np.number])) # 4. 预测概率 prob model.predict_proba(X_new)[0] result { ticker: ticker, prob_up: float(prob[2]), # 假设类别0-1,10,21 prob_down: float(prob[0]), prob_flat: float(prob[1]), predicted_class: int(model.predict(X_new)[0]) } return jsonify(result) except Exception as e: return jsonify({error: str(e)}), 500 if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse) # 生产环境禁用debug5.2 数据漂移监控用KS检验检测特征分布偏移模型上线后市场风格切换会导致特征分布漂移如波动率骤升。我们每日定时任务检查对每个数值特征计算新数据 vs 训练数据的KS统计量若KS 0.15经验阈值触发告警并冻结预测服务。from scipy.stats import ks_2samp import schedule import time def check_drift(): # 加载最新100条特征数据 latest_X X.tail(100) drift_alerts [] for col in latest_X.select_dtypes(include[np.number]).columns: # KS检验新数据 vs 训练数据取同长度样本 train_sample X[col].sample(nlen(latest_X), random_state42) ks_stat, p_value ks_2samp(latest_X[col], train_sample) if ks_stat 0.15: drift_alerts.append({ feature: col, ks_stat: round(ks_stat, 3), p_value: round(p_value, 3) }) if drift_alerts: print(f检测到{len(drift_alerts)}个特征漂移:, drift_alerts) # 发送企业微信/钉钉告警此处省略集成代码 # 同时标记模型为driftedAPI返回维护状态 else: print(无显著漂移) # 每日9:30执行A股开盘后 schedule.every().day.at(09:30).do(check_drift) while True: schedule.run_pending() time.sleep(3600) # 每小时检查一次注意KS检验要求两样本独立因此train_sample必须随机抽样而非切片。若某特征持续漂移如volume_ma_ratio在牛市中系统性升高需重新设计该特征例如改用分位数归一化。5.3 模型热更新无需重启服务的在线重训机制当漂移告警触发或新标注数据积累如人工修正100条预测错误需无缝更新模型。核心是双模型实例原子切换新模型训练完成保存为xgb_model_v2.pklAPI服务检查model_version.txt文件若版本号变更则加载新模型并更新版本文件切换过程100ms用户无感知。# 在API中添加版本检查 current_version None current_model None def load_model_if_updated(): global current_version, current_model with open(model_version.txt, r) as f: version f.read().strip() if version ! current_version: current_model joblib.load(fxgb_model_v{version}.pkl) current_version version print(f模型已更新至版本: {version}) app.before_request def before_request(): load_model_if_updated()真正的股票预测系统不是终点而是你量化能力的校准器——每一次回测失败都在告诉你市场比代码更诚实而可复现的源码是你唯一能反复打磨的武器。本文还有配套的精品资源点击获取