Python源码拆解：AI股票智能分析系统从数据到实盘

简介这是一套面向个人投资者与量化爱好者的 LLM 驱动股票智能分析系统源码覆盖 A 股、港股、美股及主流指数解决盯盘耗时、选股效率低、决策依据分散等痛点。系统以 AI 决策仪表盘输出一句话核心结论、精确买卖点与操作清单并融合技术面、筹码分布、舆情情报与实时新闻多维分析内置 A 股三段式复盘与美股 Regime Strategy 策略引擎支持大盘复盘、AI 回测验证方向胜率与止盈止损命中率以及 11 种策略的多轮问股对话。推送侧打通企业微信、飞书、Telegram、邮件等渠道还可上传自选股截图由 Vision LLM 识别代码一键加入监控。资源包共 356 个文件以 150 个 py 源码、33 个 tsx 与 29 个 ts 前端组件、23 个 md 文档为主另含 yml/yaml 配置、json 数据与 png/gif 演示素材压缩包约 49.91MB目录按后端、前端、策略与部署脚本分层。已有 197 人学习适合想用 AI 降低盯盘压力、快速搭建自选股分析流水线的读者参考。1. 从一份 Python 源码说起AI 股票智能分析系统到底在分析什么很多人第一次听到「AI 股票智能分析系统」这个词脑子里浮现的是自动下单、稳赚不赔的黑匣子。真把一份 Python 源码摊开来看你会发现它做的事情其实很朴素把行情数据抓下来、算成指标、喂给模型、输出一个带概率的判断再画成图给你看。它不替你决策它替你省掉每天手动拉数据、算均线、翻财报的时间。这套东西适合两类人一类是会用 Python 但没做过金融数据的开发者想拿一个完整项目练手另一类是做了几年交易、想把自己的经验写成可回测规则的人。核心链路就四段——数据获取、特征计算、模型推理、可视化输出任何一份能跑的源码都绕不开这四段。下面我按自己搭这套系统的顺序把每一段拆开讲清楚包括参数怎么设、哪里容易翻车。2. 数据层怎么搭从 akshare 拉 A 股日线到本地缓存2.1 为什么数据层决定了整个系统的上限模型再花哨喂进去的数据有幸存者偏差或者复权错误输出全是垃圾。我见过太多人一上来就调模型结果回测收益曲线漂亮得离谱实盘一跑就崩问题九成出在数据层。常见的数据来源分三类免费接口akshare、tushare 基础版、券商 API、本地历史文件。做个人项目我一般用 akshare因为它不需要 token、覆盖 A 股/港股/美股、字段够用缺点是接口偶尔抽风必须自己做缓存和重试。选型上有个关键判断你要的是「能复现的回测」还是「实时盯盘」。前者对数据完整性要求高必须落库后者对延迟敏感可以接受内存缓存。绝大多数人做的是前者所以第一步就是把数据落到本地 SQLite 或 Parquet别每次跑都重新请求。2.2 用 akshare 拉日线并落库的最小代码import akshare as ak import pandas as pd import sqlite3 import time def fetch_daily(symbol: str, start: str, end: str, retry: int 3) - pd.DataFrame: 拉取单只股票日线带重试。symbol 格式如 600519 for i in range(retry): try: # adjustqfq 前复权回测必须用前复权否则除权日会出现假跳空 df ak.stock_zh_a_hist(symbolsymbol, perioddaily, start_datestart, end_dateend, adjustqfq) if df.empty: raise ValueError(f{symbol} 返回空数据) df df.rename(columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, 换手率: turnover }) df[symbol] symbol df[date] pd.to_datetime(df[date]) return df[[symbol, date, open, high, low, close, volume, amount, turnover]] except Exception as e: print(f第 {i1} 次失败: {e}) time.sleep(2 ** i) # 指数退避避免被限流 raise RuntimeError(f{symbol} 拉取失败) def save_to_sqlite(df: pd.DataFrame, db_path: str stock.db): conn sqlite3.connect(db_path) df.to_sql(daily, conn, if_existsappend, indexFalse) conn.close()这段代码有三个参数值得说清楚。adjustqfq是前复权回测场景下必须用否则每次分红送股都会在 K 线上留一个假缺口均线全乱。retry3配合2 ** i的指数退避是因为免费接口在批量拉取时经常返回空或者超时硬重试会被限流更久。if_existsappend意味着你要自己处理重复写入实际项目里我会先建唯一索引(symbol, date)再用INSERT OR REPLACE避免同一只股票重复入库。2.3 批量拉取时的并发与限流参数单只股票拉十年日线大概 1 到 2 秒全 A 股五千多只串行要两三个小时。我一般用concurrent.futures.ThreadPoolExecutor但并发数不能开太大——免费接口对同一 IP 的 QPS 有限制实测max_workers4比较稳开到 8 以上就开始大面积超时。配合每批之间time.sleep(0.5)五千只股票大概四十分钟能拉完。提示第一次全量拉取建议分行业或分板块跑跑完一个板块就落库一次中途断了不用从头再来。这是血泪经验我最早一次跑了两个小时断在最后全部重来。3. 特征工程把 OHLCV 变成模型能吃的因子3.1 技术指标不是越多越好先想清楚要预测什么很多人做这套系统的第一反应是把 MACD、KDJ、RSI、BOLL 全算一遍几十个特征一股脑塞进模型。结果模型过拟合训练集准确率 90%测试集 50%。问题不在模型在特征。你得先回答一个问题我要预测的是「明天涨还是跌」还是「未来五天收益率超过 3% 的概率」前者是分类后者是回归加阈值特征设计完全不同。我一般会分三组特征趋势类均线斜率、ADX、动量类RSI、ROC、量能类量比、换手率变化。每组挑两三个总共不超过十个先跑通再逐步加。特征太多模型学到的全是噪声。3.2 用 pandas 算均线、RSI 和量比import pandas as pd import numpy as np def add_features(df: pd.DataFrame) - pd.DataFrame: df df.sort_values(date).copy() close df[close] # 均线斜率用 5 日均线相对前 5 日的变化率比单纯均线值更能反映趋势 ma5 close.rolling(5).mean() df[ma5_slope] ma5.pct_change(5) # RSI 14 日标准 Wilder 平滑 delta close.diff() gain delta.clip(lower0).ewm(alpha1/14, adjustFalse).mean() loss (-delta.clip(upper0)).ewm(alpha1/14, adjustFalse).mean() rs gain / loss.replace(0, np.nan) df[rsi14] 100 - 100 / (1 rs) # 量比当日成交量 / 过去 5 日均量 df[vol_ratio] df[volume] / df[volume].rolling(5).mean() # 标签未来 5 日收益率是否超过 3% df[future_ret] close.shift(-5) / close - 1 df[label] (df[future_ret] 0.03).astype(int) return df.dropna()这里有几个容易翻车的点。ewm(alpha1/14, adjustFalse)是 Wilder 原始 RSI 的算法很多人用rolling(14).mean()算结果和行情软件对不上回测和实盘信号不一致。vol_ratio的分母用 5 日均量如果你用的是 20 日量比会钝化短线信号出不来。标签用shift(-5)意味着最后 5 行没有标签必须dropna否则训练时会把 NaN 当 0 处理模型直接学歪。3.3 特征标准化和未来函数排查标准化用StandardScaler没问题但必须只在训练集上 fit再 transform 测试集。我见过有人对整个数据集 fit这叫数据泄露测试集准确率虚高十几个点。未来函数更隐蔽任何用到shift(-n)的特征都会引入未来信息标签可以用特征绝对不行。排查方法很简单把特征列逐个和未来收益率算相关性如果某个特征相关性高得离谱比如 0.8 以上八成是泄露了。4. 模型选型与训练LightGBM 为什么比 LSTM 更适合这个场景4.1 表格数据上树模型几乎总是赢股票因子是典型的表格数据几百到几千行特征十几个。这种数据上LightGBM、XGBoost 这类梯度提升树的表现稳定超过 LSTM。原因有两个一是树模型对特征尺度不敏感不用纠结标准化二是小样本下树模型不容易过拟合LSTM 参数多几千行数据根本喂不饱。我一般先用 LightGBM 跑一个基线如果 AUC 能到 0.55 以上再考虑要不要上深度学习。注意金融预测 AUC 0.55 已经算可用别拿图像分类的 0.95 来要求它。4.2 LightGBM 训练与时间序列切分import lightgbm as lgb from sklearn.metrics import roc_auc_score def train_model(df: pd.DataFrame, feature_cols: list): df df.sort_values(date) # 时间序列切分前 80% 训练后 20% 测试绝不能随机切 split int(len(df) * 0.8) train, test df.iloc[:split], df.iloc[split:] params { objective: binary, metric: auc, learning_rate: 0.05, num_leaves: 31, min_data_in_leaf: 50, # 小样本必须调大防止叶子过拟合 feature_fraction: 0.8, bagging_fraction: 0.8, bagging_freq: 5, verbose: -1 } dtrain lgb.Dataset(train[feature_cols], labeltrain[label]) dvalid lgb.Dataset(test[feature_cols], labeltest[label]) model lgb.train( params, dtrain, num_boost_round500, valid_sets[dvalid], callbacks[lgb.early_stopping(50), lgb.log_evaluation(100)] ) pred model.predict(test[feature_cols]) print(f测试集 AUC: {roc_auc_score(test[label], pred):.4f}) return modelmin_data_in_leaf50是关键参数。默认值是 20在几千行数据上会让每个叶子只装二三十个样本模型把噪声全记住了。调到 50 到 100 之间AUC 通常能稳一点。early_stopping(50)是防止过拟合的后悔药验证集 50 轮不提升就停别硬跑满 500 轮。时间序列切分必须按日期切随机切分会让模型在训练时「看到」未来的数据测试 AUC 虚高到 0.7 以上都是假的。4.3 特征重要性和模型可解释性训练完打印model.feature_importance()如果某个特征重要性占了一半以上要警惕。要么这个特征确实是核心因子要么它泄露了未来信息。我一般会要求前三个特征的重要性加起来不超过 70%否则回去检查特征。另外LightGBM 可以输出 SHAP 值对单笔预测解释「为什么今天给出买入信号」这在实盘复盘时非常有用。5. 避坑与排查这套系统最容易翻车的五个地方5.1 回测收益高得离谱实盘完全对不上现象回测年化 80%实盘跑三个月亏 10%。原因九成是未来函数或者幸存者偏差。检查方法把回测的每一笔交易日期打印出来看是否有在涨停板买入、跌停板卖出的记录如果有说明用了当日收盘价成交实际根本买不进。解决成交价用次日开盘价涨停日跳过退市股票要保留在样本里。5.2 akshare 接口突然返回空数据现象昨天还能拉今天全部返回空 DataFrame。原因接口字段变更或者被限流。解决先单独请求一只股票看返回结构字段名可能从「收盘」变成了「收盘价」如果是限流降低并发到 2加长 sleep。我一般会在代码里加一个字段映射的兜底逻辑字段名对不上时打印原始列名方便快速定位。5.3 模型训练集 AUC 0.9测试集 0.5现象训练集表现完美测试集和抛硬币一样。原因过拟合加数据泄露。解决先砍特征从十个砍到五个再调min_data_in_leaf到 100最后检查标准化是不是在全量数据上 fit 的。三步做完还不行说明特征本身没有预测力回去重新设计因子。5.4 复权方式选错导致均线全乱现象某只股票在分红日 K 线出现巨大跳空均线跟着断掉。原因用了不复权数据。解决回测统一用前复权qfq实盘看盘可以用不复权但两套数据要分开存别混在一个表里。我一般建两张表daily_qfq和daily_raw用的时候明确指定。5.5 预测信号每天变无法执行现象模型今天说买明天说卖信号频繁翻转。原因特征里包含了太多短期噪声或者标签阈值设得太低。解决把标签从「明天涨跌」改成「未来五天收益超 3%」信号自然平滑或者在输出层加一个连续两天同向才触发的规则。这是实盘和回测最大的差别回测可以每天调仓实盘手续费和滑点会吃掉所有利润。6. 把系统跑起来之后信号过滤与实盘对接的一个技巧模型输出概率之后直接按 0.5 阈值买卖是最粗糙的做法。我一般会加一层信号过滤只有当预测概率连续两天都超过 0.6且当日成交量大于 5 日均量时才触发买入。这一层规则能把信号数量砍掉一半但胜率通常能提几个点。下面这个过滤函数可以直接接在模型输出后面。def filter_signal(df: pd.DataFrame, prob_col: str prob, threshold: float 0.6) - pd.DataFrame: 连续两天概率超阈值 放量才给出买入信号 df df.sort_values(date).copy() df[signal] 0 cond_prob (df[prob_col] threshold) (df[prob_col].shift(1) threshold) cond_vol df[volume] df[volume].rolling(5).mean() df.loc[cond_prob cond_vol, signal] 1 return df参数上threshold从 0.5 提到 0.6信号变少但更准连续两天的条件是为了过滤掉单日噪声。回测时对比一下加过滤前后的夏普比率如果夏普提升但总收益下降说明过滤有效——你牺牲了交易次数换来了稳定性。实盘对接时这个signal列可以直接推送到券商的交易接口但记住加一个手动确认环节别全自动系统出错时你连后悔药都没得吃。我自己跑这套系统两年多最大的习惯是每次改完特征或参数先把回测区间切成三段分别验证任何一段表现异常就回去查数据。模型不是越复杂越好能稳定跑、能解释、能复现比多那两三个点收益重要得多。希望帮到你。本文还有配套的精品资源点击获取