简介：这是一套面向个人投资者与量化爱好者的 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: # adjust='qfq' 前复权，回测必须用前复权，否则除权日会出现假跳空 df = ak.stock_zh_a_hist(symbol=symbol, period="daily", start_date=start, end_date=end, adjust="qfq") 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"第 {i+1} 次失败: {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_exists="append", index=False) conn.close()这段代码有三个参数值得说清楚。adjust="qfq"是前复权，回测场景下必须用，否则每次分红送股都会在 K 线上留一个假缺口，均线全乱。retry=3配合2 ** i的指数退避，是因为免费接口在批量拉取时经常返回空或者超时，硬重试会被限流更久。if_exists="append"意味着你要自己处理重复写入，实际项目里我会先建唯一索引(symbol, date)，再用INSERT OR REPLACE，避免同一只股票重复入库。
2.3 批量拉取时的并发与限流参数
单只股票拉十年日线大概 1 到 2 秒，全 A 股五千多只，串行要两三个小时。我一般用concurrent.futures.ThreadPoolExecutor，但并发数不能开太大——免费接口对同一 IP 的 QPS 有限制，实测max_workers=4比较稳，开到 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(lower=0).ewm(alpha=1/14, adjust=False).mean() loss = (-delta.clip(upper=0)).ewm(alpha=1/14, adjust=False).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(alpha=1/14, adjust=False)是 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], label=train["label"]) dvalid = lgb.Dataset(test[feature_cols], label=test["label"]) model = lgb.train( params, dtrain, num_boost_round=500, 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_leaf=50是关键参数。默认值是 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列可以直接推送到券商的交易接口，但记住加一个手动确认环节，别全自动，系统出错时你连后悔药都没得吃。
我自己跑这套系统两年多，最大的习惯是每次改完特征或参数，先把回测区间切成三段分别验证，任何一段表现异常就回去查数据。模型不是越复杂越好，能稳定跑、能解释、能复现，比多那两三个点收益重要得多。希望帮到你。
本文还有配套的精品资源，点击获取
