Python股票预测系统：从数据管道到Flask服务的完整工程实践

简介本资源是一套基于Python实现的股票预测系统源码面向金融数据分析初学者、量化交易爱好者及高校相关专业学生旨在通过历史行情数据建模辅助投资决策。包内共77个文件含43个Python脚本涵盖数据预处理、策略生成、模型训练与结果可视化等核心模块、21个CSV历史行情数据如ETH/USDT多周期K线、银行股日线等、7个文本配置与日志文件、2个Excel个股数据表、2张PNG分析图表以及LICENSE、.gitignore等工程规范文件压缩包大小89.28MB。已有367人学习下载资源结构清晰策略丰富——包含比特币双重策略、中国银行/紫光股份等个股专研模块以及5分钟/15分钟/小时级多粒度回测支持配套readme.txt和Markdown文档提供快速上手指引所有代码可直接运行调试是理解量化策略落地与时间序列预测实践的完整工程范例。1. 为什么用 Python 做股票预测不是写个“涨跌判断”就完事了很多人一看到“基于Python的股票预测系统设计源码”第一反应是又一个用 sklearn.LinearRegression 拟合收盘价、画条线就交差的课程作业。但真实场景里这类系统失败的根本原因从来不是模型不够深而是数据链路断裂、特征逻辑脱钩、回测机制缺失、部署路径模糊——你跑通了 train.py却不知道 predict.py 怎么接入实时行情接口也不知道 model.pkl 在生产环境里被 pickle 版本不兼容卡死三次。本文聚焦的是一个能从原始行情数据出发经清洗、特征工程、多模型训练与对比、滚动回测验证、再到轻量服务化封装的完整闭环。它不追求年化 100% 的幻觉指标但要求每一步输出可复现、参数可解释、异常可定位。适合有 Pandas/Numpy 基础、已跑过 LSTM 单例但卡在实盘衔接的中级开发者也适合作为毕业设计或内部工具原型的技术骨架。2. 从 Tushare 获取原始行情到结构化特征表数据管道必须可重放股票预测的起点不是模型而是带时间戳对齐、无前向泄露、含合理滞后窗口的特征矩阵。直接用 yfinance 或 akshare 拉日线不做字段校验和缺失填充后续所有模型输出都是空中楼阁。我们采用 Tusharev2 接口作为主数据源因其免费额度覆盖 A 股全市场、字段规范、支持复权处理且社区维护活跃——这比硬啃交易所 Level-2 原始报单更贴近工程落地场景。2.1 安装依赖与获取 Token 的最小可行配置pip install tushare pandas numpy scikit-learn joblib tqdmTushare 需注册获取 token写入环境变量避免硬编码export TUSHARE_TOKENyour_1234567890abcdefg_token_here提示Token 请勿提交至 Git。实际项目中建议用python-dotenv加载.env文件而非os.environ.get()直接读取。2.2 构建可重放的数据拉取函数核心约束有三时间范围可控支持指定起止日期避免每次全量拉取复权一致统一使用adjqfq前复权消除分红送股干扰字段精简只保留trade_date, open, high, low, close, vol, amount七列其余如change,pct_chg等由代码动态计算确保逻辑透明。# data_loader.py import tushare as ts import pandas as pd import os def fetch_stock_data(ts_code, start_date, end_date, freqD): 拉取单只股票日线行情前复权 :param ts_code: 如 000001.SZ :param start_date: 20200101 :param end_date: 20231231 :return: pd.DataFrame, 列名标准化为小写索引为 DatetimeIndex pro ts.pro_api(os.getenv(TUSHARE_TOKEN)) df pro.daily( ts_codets_code, trade_date, start_datestart_date, end_dateend_date, fieldstrade_date,open,high,low,close,vol,amount ) df[trade_date] pd.to_datetime(df[trade_date]) df df.sort_values(trade_date).set_index(trade_date) # 统一列名小写便于后续链式操作 df.columns [col.lower() for col in df.columns] return df # 示例拉取平安银行 2022 年数据 df_raw fetch_stock_data(000001.SZ, 20220101, 20221231) print(df_raw.shape) # (242, 7)该函数返回的df_raw是原始输入尚未做任何特征衍生。关键点在于它不调用pro.adj_factor单独查复权因子而是依赖pro.daily接口内置的adjqfq参数默认即启用避免因子表与行情表日期错位。2.3 特征工程构造无未来信息的滞后序列与技术指标预测明日收盘价不能用今日的 MACD 信号——因为 MACD 本身依赖未来 N 日数据。真正可部署的特征必须满足所有计算仅基于 t 时刻及之前的数据。我们定义三类特征类型示例计算逻辑是否滞后基础价格比率close/open,high/low当日内比值无滞后否滞后统计量close.shift(1).rolling(5).mean()5 日均价t 时刻值反映 t−1 至 t−5 数据是技术指标修正版ta.SMA(close, timeperiod10)使用 TA-Lib 的 SMA但传入close[:i]子序列是# feature_engineer.py import pandas as pd import talib as ta def generate_features(df): 输入fetch_stock_data 返回的 DataFrame索引为 DatetimeIndex 输出新增特征列的 DataFrame保留原始 price/vol 字段 df df.copy() # 1. 基础比率当日完成无泄露 df[cr] df[close] / df[open] # 收盘/开盘 df[hr] df[high] / df[low] # 最高/最低 # 2. 滞后统计显式 shift明确时间偏移 df[ma5] df[close].rolling(5).mean().shift(1) # t 时刻的 5 日均值 t−1 至 t−5 均值 df[vol_ma10] df[vol].rolling(10).mean().shift(1) df[price_std20] df[close].rolling(20).std().shift(1) # 3. TA-Lib 指标需截断历史序列避免跨期 # 注意ta.SMA 要求输入为 np.array且长度足够 close_arr df[close].values ma10_arr ta.SMA(close_arr, timeperiod10) # 将结果对齐ta.SMA 返回数组首 9 个为 nan故需 shift(1) 对齐 df[ta_ma10] pd.Series(ma10_arr, indexdf.index).shift(1) # 4. 目标变量明日收盘价回归任务或涨跌标签分类任务 df[target_close] df[close].shift(-1) # 预测 t1 收盘价 df[label] (df[target_close] df[close]).astype(int) # 1涨0跌 return df.dropna(subset[ma5, ta_ma10, target_close]) # 删除含 nan 行 df_feat generate_features(df_raw) print(f特征矩阵形状{df_feat.shape}) # 例如 (232, 12)比原始少 10 行因 rolling 和 shift注意dropna(subset[...])是硬性要求。若某行任一特征为 NaN则整行剔除不可用 fillna(0) 或 ffill——那会引入未来信息或扭曲分布。滚动窗口大小如 5、10、20需与业务周期匹配A 股周度波动常用 5 日月度趋势用 20 日。3. 多模型并行训练与滚动时序验证拒绝随机切分股票时间序列具有强自相关性与结构性突变如政策发布、财报季用train_test_split(random_state42)切分训练/测试集等同于把 2020 年数据当训练集、2021 年当测试集却让模型“看见”2021 年的全局统计量如整体均值。正确做法是滚动窗口Rolling Window 前向链Forward Chaining用前 N 天训练预测第 N1 天再滑动一天重复此过程。3.1 定义滚动验证器与特征-标签分离逻辑# model_trainer.py from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor from sklearn.linear_model import LinearRegression from sklearn.metrics import mean_absolute_error, mean_squared_error import numpy as np class RollingValidator: def __init__(self, window_size250, step10): :param window_size: 训练窗口长度交易日 :param step: 每次滑动步长交易日 self.window_size window_size self.step step def split(self, X, y, dates): 生成滚动切片索引 :return: generator of (train_idx, test_idx) n len(X) for i in range(self.window_size, n - 1, self.step): train_end i test_start train_end test_end min(train_end self.step, n) yield list(range(train_end - self.window_size, train_end)), list(range(test_start, test_end)) def prepare_xy(df, feature_cols, target_coltarget_close): 分离特征 X 与目标 y确保顺序与日期对齐 X df[feature_cols].values y df[target_col].values dates df.index return X, y, dates # 定义特征列排除原始价格与目标列 feature_cols [cr, hr, ma5, vol_ma10, price_std20, ta_ma10] X, y, dates prepare_xy(df_feat, feature_cols) # 初始化验证器 validator RollingValidator(window_size250, step20) # 用 250 天约一年训练每 20 天评估一次3.2 并行训练三种模型并记录时序性能我们选择 LinearRegression基线、RandomForest非线性拟合、GradientBoosting梯度提升作对比。关键点每次滚动切片独立训练模型不复用上一轮权重——这是模拟实盘冷启动的真实压力。models { Linear: LinearRegression(), RF: RandomForestRegressor(n_estimators100, random_state42), GBDT: GradientBoostingRegressor(n_estimators100, random_state42) } results {name: {mae: [], rmse: []} for name in models} for train_idx, test_idx in validator.split(X, y, dates): X_train, X_test X[train_idx], X[test_idx] y_train, y_test y[train_idx], y[test_idx] for name, model in models.items(): model.fit(X_train, y_train) y_pred model.predict(X_test) mae mean_absolute_error(y_test, y_pred) rmse np.sqrt(mean_squared_error(y_test, y_pred)) results[name][mae].append(mae) results[name][rmse].append(rmse) # 汇总各模型在全部滚动窗口上的平均误差 summary {} for name, metrics in results.items(): summary[name] { avg_mae: np.mean(metrics[mae]), avg_rmse: np.mean(metrics[rmse]), std_mae: np.std(metrics[mae]) } # 打印结果示例 Linear: avg_mae0.321, avg_rmse0.412, std_mae0.089 RF: avg_mae0.287, avg_rmse0.375, std_mae0.072 GBDT: avg_mae0.273, avg_rmse0.361, std_mae0.065 提示std_mae反映模型稳定性。若 GBDT 的 std_mae 显著高于 RF说明其在某些波动剧烈窗口如年报披露期过拟合严重此时应增加早停early stopping或降低树深度。滚动验证耗时较长建议用joblib.Parallel并行化外层循环但注意model.fit()本身已多线程需设n_jobs1避免嵌套并行冲突。3.3 保存最优模型与特征缩放器模型上线前必须固化预处理逻辑。此处使用StandardScaler对特征做 Z-score 标准化并与模型一同序列化from sklearn.preprocessing import StandardScaler import joblib # 用全部可用数据非滚动训练最终 scaler 和模型 scaler StandardScaler() X_scaled scaler.fit_transform(X) final_model GradientBoostingRegressor(n_estimators100, random_state42) final_model.fit(X_scaled, y) # 保存为 .pkl joblib.dump(scaler, scaler_000001.pkl) joblib.dump(final_model, gbdt_000001.pkl) # 验证加载是否正常 scaler_load joblib.load(scaler_000001.pkl) model_load joblib.load(gbdt_000001.pkl) test_input X_scaled[-1:].reshape(1, -1) pred model_load.predict(test_input) print(f加载模型预测{pred[0]:.3f}) # 应与原模型输出一致4. 构建轻量预测服务Flask API 实时行情注入训练完成的模型只是静态快照要变成可用系统必须解决两个问题如何接收新行情不能每天手动跑 batch如何返回结构化预测不是打印数字而是 JSON 响应含置信区间与解释性指标。我们选用 Flask非 FastAPI——因其零依赖、调试友好、适合内网部署且与现有 Python 生态无缝衔接。4.1 定义预测端点接收单日行情返回次日预测# app.py from flask import Flask, request, jsonify import joblib import pandas as pd import numpy as np app Flask(__name__) # 加载模型与 scaler全局单例 scaler joblib.load(scaler_000001.pkl) model joblib.load(gbdt_000001.pkl) feature_cols [cr, hr, ma5, vol_ma10, price_std20, ta_ma10] app.route(/predict, methods[POST]) def predict_next_close(): POST body 示例 { open: 14.25, high: 14.52, low: 14.18, close: 14.43, vol: 1250000, amount: 18050000 } try: data request.get_json() if not data: return jsonify({error: No JSON data provided}), 400 # 构造单行 DataFrame模拟当日收盘后输入 df_input pd.DataFrame([data]) # 计算基础比率 df_input[cr] df_input[close] / df_input[open] df_input[hr] df_input[high] / df_input[low] # 此处需外部提供滞后特征关键约束ma5 等必须由上游系统计算好传入 # 因为 Flask 无法访问历史数据库故假设调用方已计算好 required_lag [ma5, vol_ma10, price_std20, ta_ma10] for col in required_lag: if col not in data: return jsonify({error: fMissing required lag feature: {col}}), 400 # 组织特征向量 X_new df_input[feature_cols].values # 标准化 预测 X_scaled scaler.transform(X_new) pred_close model.predict(X_scaled)[0] # 返回结构化响应 return jsonify({ status: success, predicted_close: round(float(pred_close), 3), current_close: float(data[close]), predicted_change_pct: round((pred_close - data[close]) / data[close] * 100, 3), timestamp: pd.Timestamp.now().isoformat() }) except Exception as e: return jsonify({error: str(e)}), 500 if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse) # 生产环境禁用 debugTrue注意ma5,vol_ma10等滞后特征不能由 Flask 实时计算因为缺少历史序列。正确架构是上游调度任务如 Airflow每日拉取行情、计算特征、存入 Redis 或本地 CSV本服务只做最后一步推理。此处接口强制校验这些字段存在倒逼数据管道完整性。4.2 启动服务与 CURL 测试# 终端 1启动服务 python app.py # 终端 2发送测试请求替换为真实数值 curl -X POST http://localhost:5000/predict \ -H Content-Type: application/json \ -d { open: 14.25, high: 14.52, low: 14.18, close: 14.43, vol: 1250000, amount: 18050000, ma5: 14.32, vol_ma10: 1120000, price_std20: 0.18, ta_ma10: 14.29 }响应示例{ status: success, predicted_close: 14.512, current_close: 14.43, predicted_change_pct: 0.568, timestamp: 2024-06-15T09:23:41.123456 }4.3 部署加固进程守护与请求限流Flask 默认单线程生产需配gunicorn同时加入简单限流防刷pip install gunicorn gunicorn -w 4 -b 0.0.0.0:5000 --timeout 30 app:app在app.py中添加内存级限流每 IP 每分钟最多 10 次from collections import defaultdict, deque import time # 简单内存限流生产环境建议用 Redis ip_requests defaultdict(deque) def is_rate_limited(ip): now time.time() # 清理 60 秒前的请求 while ip_requests[ip] and ip_requests[ip][0] now - 60: ip_requests[ip].popleft() if len(ip_requests[ip]) 10: return True ip_requests[ip].append(now) return False app.before_request def limit_requests(): ip request.remote_addr if is_rate_limited(ip): return jsonify({error: Rate limit exceeded}), 4295. 关键参数调优表与三个必避坑点模型效果不等于系统可用。以下参数直接影响实盘鲁棒性必须在开发阶段就锁定而三个高频坑点90% 的初学者会在部署后第二天才发现。5.1 核心参数影响速查表参数推荐值调整依据过大风险过小风险rolling_window滚动训练天数2501 年A 股风格切换周期模型僵化无法适应新趋势过拟合短期噪音泛化差step滚动步长201 月平衡评估频次与计算开销频繁重训CPU 持续 100%评估滞后错过拐点n_estimators树模型100默认平衡精度与速度内存暴涨预测延迟 500ms拟合不足MAE 上升 15%timeperiodTA-Lib SMA10短期趋势敏感度噪声放大信号频繁翻转延迟过大错过启动点5.2 三个必须提前规避的硬伤坑点 1Pickle 版本锁死导致线上加载失败joblib.dump(model, model.pkl)生成的文件与 Python 版本、scikit-learn 版本强绑定。同一模型在 Python 3.9 sklearn 1.2 下保存在 3.11 sklearn 1.4 下加载会报ModuleNotFoundError。✅ 正解改用sklearn.externals.joblib已废弃统一用pickle 显式版本声明import pickle import sklearn with open(model_v1.2.pkl, wb) as f: pickle.dump({ model: model, sklearn_version: sklearn.__version__, python_version: ..join(map(str, sys.version_info[:2])) }, f)加载时校验版本不匹配则拒绝启动。坑点 2TA-Lib 编译失败引发 pip install 卡住pip install TA-Lib在 Windows/macOS 上常因编译器缺失失败且官方 wheel 仅支持特定 Python 版本。✅ 正解放弃 pip改用预编译二进制包Windows从 https://www.lfd.uci.edu/~gohlke/pythonlibs/#ta-lib 下载对应cp39-cp39-win_amd64.whl执行pip install TA_Lib‑0.4.28‑cp39‑cp39‑win_amd64.whlmacOSbrew install ta-lib后pip install TA-LibLinuxapt-get install build-essential后再 pip。坑点 3DataFrame 时区未统一导致日期错位Tushare 返回的trade_date是datetime64[ns]但若本地系统时区为Asia/Shanghai而 pandas 默认按 UTC 解析会导致2022-01-01被误判为2021-12-31 16:00:0000:00。✅ 正解所有时间列显式赋时区df[trade_date] pd.to_datetime(df[trade_date]).dt.tz_localize(Asia/Shanghai) df df.set_index(trade_date)并在generate_features()开头加断言assert df.index.tz Asia/Shanghai, Timezone mismatch detected!最后一行不总结只留一个可立即执行的动作运行python -m pytest tests/test_data_pipeline.py -v验证数据管道时区与缺失值处理逻辑是否符合预期。本文还有配套的精品资源点击获取