多Agent与LGBM双模型：AI炒股系统源码从0到1拆解

简介：面向量化入门者与Python爱好者的轻量级AI炒股系统完整源码包，聚焦多Agent架构与LGBM双模型实战——分类模型判断涨跌、回归模型预测涨幅，全部预测仅使用当日及之前数据，杜绝未来函数。系统内置选股、风控、择时、复盘四个解耦模块，配合零前端要求的Streamlit交互界面，可在普通笔记本上运行，适合用于单票深度择时与自动生成周度复盘报告。压缩包共11个文件，以Python源码和模型权重为核心：4个py脚本覆盖训练、主流程与工具函数，2个pkl为已训练的分类/回归模型，2个csv保存交易与数据记录，另有辅助缓存与说明文档，整体仅598KB。目前已有440人学习下载，代码结构清晰、模块边界分明，便于二次开发与扩展到多标的场景。

1. AI炒股系统源代码：从“拍脑袋下单”到“多Agent+LGBM自动决策”的第一步

AI炒股系统源代码这个词看着唬人，真正落地时你会发现，核心不是一段能自动产生交易信号的神秘算法，而是一条环环相扣的工程链路：数据清洗、特征计算、LGBM训练、Agent调度、周度复盘。标题里的“多Agent架构+LGBM双模型”也不是算法炫技，而是两种互补的设计取舍——前者解决“谁在什么条件下做决定”，后者解决“用什么信号做决定”。这篇笔记从0到1拆开这套最小实现，讲清楚为什么用双LGBM、三个Agent怎么协作、回测和复盘哪里最容易翻车，以及周度复盘之后如何判断模型是否还有效。

这套架构能解决的问题很具体：单模型信号噪声大、决策逻辑散落在各个函数里互相牵制、复盘结论总是靠记忆而不是靠数据。适合有Python与Pandas基础、想把研究流程工具化的开发者，也适合已经跑过渡模型但始终觉得“分数高、实操慌”的人。这个标题不承诺稳赚，它承诺的是可复现、可追溯、可修。

2. 多Agent与LGBM双模型选型：这套架构为什么值得从0到1搭一遍

2.1 多Agent架构的边界：什么信号需要“角色”而不是“函数”

多数人刚开始写策略时会走一条纯函数流水线：下载行情、算指标、出信号、执行动作。问题在于，当规则变多以后，所有逻辑摊在一个文件里，改一个条件就会牵动另一处判断；尤其“止损”和“复盘”这样横跨多天的责任，函数是无状态的，它记不住上周的决策结论。

多Agent架构就是把这些横切责任拆成独立决策单元：策略Agent负责看模型概率，风控Agent负责压缩仓位上限，复盘Agent负责在每周结束时总结战果。每个Agent有独立的输入、输出和最小触发频率，它们之间不直接调用对方的内部状态，只交换一份结构化消息。它不需要像微服务那样重资源，本质上只是以消息为边界的决策模块，但边界一旦划清，改起来非常痛快。

判断标准其实只有一条：逻辑只服务单一流程且不依赖历史上下文，就写函数；如果它需要在特定时间点被唤醒、并且输出会改变后续多次调用的外部状态，就拆成Agent。以风控为例，止损不是每天固定在某一个时间点执行，它要在任何一次策略输出后立刻介入，把动作压到安全区间，这天然适合独立Agent。

2.2 双LGBM的分工：两个模型为什么比一个“万能模型”更稳

单一模型做“今日方向预测”很容易被单一频率绑架。日线模型训练样本多，但噪声也大；把标签周期拉长到周，样本又变稀疏，模型学不到短周期转折。标题里的“双模型”是常见做法：一个日频择时模型，一个周频方向校验模型。两者频率不同、任务不同，不共享特征，只在决策消息层做互相校验。

日频模型负责产出短周期概率信号并决定动作；周频模型不直接下单，而是在日频信号出现时做反向否决，比如周频概率明显低于0.45，就把日频的“买”压住。这种设计与金融数据的特性对齐：短周期动量与长周期趋势经常背离，用一个模型同时拟合两者，等于逼它学两个冲突的目标函数。

至于为什么选LGBM而不是神经网络或大模型，理由是工程成本。LGBM对表格特征友好，几十棵树几秒就能训练一轮，特征重要性可直接读出来，适合个人团队反复调参。大模型擅长处理文本复盘，但不擅长从几百维数值特征里稳定提取方向信号，所以这套架构里LGBM负责数值决策，文本复盘交给后端的提示词模板就够了。

2.3 最小技术栈与数据流：先让系统里有“能跑”的东西

先列依赖，全部装最新稳定版即可。akshare负责免费行情，lightgbm负责分类，schedule做轻量定时调度，pyyaml统一参数，sqlite3做本地落库。

akshare pandas numpy lightgbm pyyaml schedule

数据流只有一条主线：akshare拉K线 -> pandas清洗并落SQLite -> 特征与标签构建 -> 训练双LGBM -> 策略Agent读最新特征出信号 -> 风控Agent限仓 -> 决策落库 -> 复盘Agent每周末聚合计算 -> weekly_review表更新。所有Agent都围绕这张表工作，谁出了问题都能从表里追溯。

配置文件我一般这样写，所有关键阈值都集中在一处：

# config.yaml data: symbol: "600519" period: "daily" forward_days: 5 model: lgb_params: n_estimators: 800 learning_rate: 0.02 num_leaves: 31 max_depth: 6 agent: min_prob: 0.58 max_position_pct: 0.8

最后初始化数据库。三张表分别存K线缓存、决策信号、周度复盘，这是让“周度复盘”真正落地的根基：

import sqlite3 conn = sqlite3.connect("trade_system.db") conn.executescript(""" CREATE TABLE IF NOT EXISTS kline_cache ( symbol TEXT, date TEXT, open REAL, high REAL, low REAL, close REAL, volume REAL, PRIMARY KEY(symbol, date) ); CREATE TABLE IF NOT EXISTS signals ( symbol TEXT, date TEXT, model_name TEXT, prob REAL, decision TEXT, confidence REAL, reason TEXT ); CREATE TABLE IF NOT EXISTS weekly_review ( week_start TEXT, stock_count INTEGER, total_return REAL, max_drawdown REAL, win_rate REAL, benchmark_return REAL, model_verdict TEXT ); """) conn.close()

这段代码的逻辑很直白：kline_cache用symbol + date做主键，避免重复拉取时产生脏数据；signals表是Agent协作的中枢，策略Agent写入，复盘Agent读取；weekly_review是复盘结论的唯一归属地。参数上要留意agent.min_prob，这个阈值会在策略Agent里直接决定是否触发买入动作，建议先按0.58起步，别一开始就压到0.5，否则噪音信号会淹没整个系统。

3. 从0到1搭建可运行的双LGBM：数据清洗、特征工程与训练脚本

3.1 数据获取与时间对齐：先用akshare把K线落成SQLite缓存

行情数据是整个系统的地基。用akshare拿历史K线最省事，但要注意接口返回的是中文列名，不重命名后面所有特征计算都会报KeyError。这里做一层薄封装，顺便把入库逻辑带上：

import akshare as ak import pandas as pd def fetch_kline(symbol: str, period: str = "daily") -> pd.DataFrame: df = ak.stock_zh_a_hist( symbol=symbol, period=period, start_date="20200101", end_date="20241231", adjust="qfq", # 使用前复权，保证历史价格连续 ) df = df.rename(columns={ "日期": "date", "开盘": "open", "最高": "high", "最低": "low", "收盘": "close", "成交量": "volume", }) df = df[["date", "open", "high", "low", "close", "volume"]] df["date"] = pd.to_datetime(df["date"]) df = df.set_index("date").sort_index() return df df_daily = fetch_kline("600519", "daily") df_daily.to_sql("kline_cache", conn, if_exists="append", index=True)

这段代码的要点有两个。第一，adjust="qfq"让回测与实盘看到的K线口径一致，这是杜绝第5章“假缺口”的第一道保险；第二，to_sql没有做幂等去重，重复执行会插入重复行。实战中我会在写入前先DELETE FROM kline_cache WHERE symbol=? AND date IN (?)，或者直接改成if_exists="replace"。

时间对齐在这里要特别说一句。日线数据在t日收盘后，t日全部open/high/low/close/volume都是已知的，所以t日特征可被t日收盘后的策略Agent使用；标签指向t+5日才确定的未来收益，这不构成穿越。真正的穿越是特征用了t+5才知道的量，或者把全量数据标准化之后再切训练测试集，这两种错误在后面避坑章节细讲。

3.2 特征工程：让模型看见动量、波动率与MACD双底形态

特征不用堆太多，十几列足够支撑一个可解释的基线。核心包括动量、均线乖离、量能、MACD，再加一个“MACD双底”形态特征，对应热搜里常提到的双底结构。整个构建函数如下：

def build_features(df: pd.DataFrame) -> pd.DataFrame: feat = df.copy() feat["ret_1"] = feat["close"].pct_change(1) feat["ret_5"] = feat["close"].pct_change(5) feat["ma5"] = feat["close"].rolling(5).mean() feat["ma20"] = feat["close"].rolling(20).mean() feat["vol_ratio"] = feat["volume"] / feat["volume"].rolling(20).mean() ema12 = feat["close"].ewm(span=12, adjust=False).mean() ema26 = feat["close"].ewm(span=26, adjust=False).mean() feat["dif"] = ema12 - ema26 feat["dea"] = feat["dif"].ewm(span=9, adjust=False).mean() feat["macd"] = (feat["dif"] - feat["dea"]) * 2 feat["dif_gt_dea"] = (feat["dif"] > feat["dea"]).astype(int) feat["cross_up"] = (feat["dif_gt_dea"].diff() == 1).astype(int) feat["low_floor"] = feat["close"].rolling(10).min().shift(1) feat["near_floor"] = (feat["close"] < feat["low_floor"] * 1.02).astype(int) feat["double_bottom"] = feat["cross_up"] * feat["near_floor"] feat["fwd_ret"] = feat["close"].shift(-5) / feat["close"] - 1 feat["label"] = (feat["fwd_ret"] > 0.02).astype(int) feat = feat.dropna(subset=["label"]) feat = feat.drop(columns=["fwd_ret"]) return feat

几个参数的用意说一下。ret_5和ma5/ma20是主力动量特征，缺一不可；vol_ratio用20日均量做分母，用来过滤放量假突破。ema12/ema26/ema9是MACD基础参数，也是搜索引擎里“MACD双底”这个长尾词对应的计算方式。double_bottom的构造逻辑是：DIF刚上穿DEA，同时收盘价离前10日最低点不远，即“低位金叉且贴着地板”，这个形态特征本身有很强的行情语义，但对单只股票样本量有限，我通常只作为辅助特征而不是主信号。

最后的标签用“未来5日收益超过2%”做正样本。这里的阈值2%是经验值，太小会把震荡市噪声学进去，太大则样本量不够。构造标签时注意，shift(-5)之后不要急着dropna，只丢掉label为空的尾巴行即可，特征缺失行交给后续LightGBM自动处理。

3.3 训练双LGBM：为什么必须用时序切分而不是随机切分

训练函数在日线和周线数据上共用一套，减少维护成本。采用时序切分而不是随机KFold，是因为金融数据有强时间相关性，随机切分会让验证集里出现早于训练集的时间点，等于提前告诉模型未来答案：

import lightgbm as lgb from sklearn.model_selection import TimeSeriesSplit def train_model(feat: pd.DataFrame, feature_cols: list) -> lgb.LGBMClassifier: X = feat[feature_cols] y = feat["label"] model = lgb.LGBMClassifier( n_estimators=800, learning_rate=0.02, num_leaves=31, max_depth=6, colsample_bytree=0.8, subsample=0.8, subsample_freq=1, random_state=42, verbose=-1, ) tscv = TimeSeriesSplit(n_splits=5) for train_idx, val_idx in tscv.split(X): X_train, X_val = X.iloc[train_idx], X.iloc[val_idx] y_train, y_val = y.iloc[train_idx], y.iloc[val_idx] model.fit( X_train, y_train, eval_set=[(X_val, y_val)], eval_metric="auc", ) return model feature_cols = [ "ret_1", "ret_5", "ma5", "ma20", "vol_ratio", "dif", "dea", "macd", "double_bottom", ] df_daily = pd.read_sql_query( "SELECT * FROM kline_cache", conn, index_col="date" ) feat_daily = build_features(df_daily) feat_weekly = df_daily.resample("W").agg({ "close": "last", "volume": "sum", }).dropna() feat_weekly = build_features(feat_weekly).dropna() model_day = train_model(feat_daily, feature_cols) model_week = train_model(feat_weekly, feature_cols)

参数上，n_estimators=800配合learning_rate=0.02是防止过拟合的常见组合，树越多学习率越低，单棵树贡献被稀释；num_leaves=31和max_depth=6控制树复杂度，避免单棵树长成“记忆树”。colsample_bytree=0.8在每棵树训练时随机抽80%特征，subsample=0.8随机抽80%样本，两者都是为泛化兜底。

周线模型直接复用build_features，但要注意它的标签shift(-5)在周线上代表“未来5周”，语义与日线完全不同，这正是周度校验想要的长期方向。样本量上，一只股票从2020到2024大约240个周线样本，训练800棵树偏少，建议周度模型用多只股票拼接的数据训练，或把n_estimators降到300。这是常见的实操修改点。

3.4 把概率变成动作：置信度与择时阈值的设定

模型输出的原始概率不能直接交易，还需要一层薄薄的决策封装。统一输出格式，方便第4章的Agent消息传递：

def predict_signal(model, feat_row: pd.DataFrame) -> dict: prob = float(model.predict_proba(feat_row)[:, 1][-1]) decision = "buy" if prob >= 0.58 else "hold" return { "prob": round(prob, 4), "decision": decision, "confidence": round(prob, 4), }

这里把confidence直接定义为正样本概率，不加任何变换。阈值0.58与config.yaml里的min_prob保持一致，低于这个值不做任何动作。不要把测试集AUC高当成调低阈值的理由，0.58是这个架构里承接“双模型校验”的起点，后续要靠Agent层的风控逻辑修正，而不是靠裸阈值。

4. 多Agent协作调度：择时、风控与周度复盘如何串成一条流水线

4.1 一份消息走天下：策略Agent与风控Agent的协作契约

三个Agent之间不直接调用彼此的私有方法，只交换一份字典消息，这是多Agent架构能保持清爽的关键。策略Agent持有两个模型，在内部完成双模型校验：

class StrategyAgent: def __init__(self, model_day, model_week, min_prob=0.58): self.model_day = model_day self.model_week = model_week self.min_prob = min_prob def decide(self, feat_day, feat_week) -> dict: day = predict_signal(self.model_day, feat_day) message = { "prob": day["prob"], "confidence": day["confidence"], "action": day["decision"], "reason": "day_prob_only", } if feat_week is not None and self.model_week is not None: week = predict_signal(self.model_week, feat_week) if week["prob"] < 0.45 and day["prob"] >= self.min_prob: message["action"] = "hold" message["reason"] = "week_model_veto" return message class RiskAgent: def __init__(self, max_position_pct=0.8): self.max_position_pct = max_position_pct def filter(self, message: dict, current_pos: float) -> dict: message["allowed_position"] = min( current_pos + 0.1, self.max_position_pct ) if message["action"] == "buy": message["action"] = ( "buy" if current_pos < self.max_position_pct else "hold" ) return message

双模型校验逻辑在decide里是关键：日频模型给概率，周频模型只有“prob < 0.45”时才有一票否决权。这个不对称设计是有意的，周频样本少，不能让它主动下单，但让它挡掉方向冲突的日频信号是可靠的。RiskAgent的current_pos + 0.1表示每次最多加仓10%，单票总仓位上限80%，这两个都是可配置经验参数，如果做组合可以再拆成按票维度。

4.2 调度入口：用schedule把三个Agent按各自节奏跑起来

调度逻辑里，策略Agent在盘中高频运行，风控Agent紧随其后，复盘Agent只在每周一开盘前运行一次。schedule库可以满足轻量需求：

import schedule, time from agents import StrategyAgent, RiskAgent, ReviewAgent strategy = StrategyAgent(model_day, model_week) risk = RiskAgent(max_position_pct=0.8) review = ReviewAgent(conn) schedule.every(30).minutes.do( lambda: strategy.decide(feat_daily_tail, feat_weekly_tail) ) schedule.every().monday.at("09:30").do( lambda: review.run_weekly(last_monday, last_friday) ) while True: schedule.run_pending() time.sleep(30)

如果要长期开机运行，schedule的while True在主进程阻塞，重启后任务会丢失，生产环境通常换APScheduler或落到crontab里，用系统级时间保证每日/每周触发不丢。无论用哪种调度器，务必记住一点：Agent触发频率只决定采样粒度，真正的决策门槛始终在策略Agent的min_prob和风控Agent的max_position_pct里，外部调度器不要越过这两道闸门直接改持仓。

4.3 周度复盘Agent：把一周的决策结果变成一张可追溯的表

复盘Agent是标题里“周度复盘”落到工程上的核心角色。它不预测，只把系统在过去一周做了什么、做得怎么样固化成结构化数据。实现上读取signals表并聚合：

class ReviewAgent: def __init__(self, conn): self.conn = conn def run_weekly(self, week_start: str, week_end: str) -> dict: sql = """ SELECT date, prob, confidence FROM signals WHERE date BETWEEN ? AND ? AND decision = 'buy' """ rows = pd.read_sql_query( sql, self.conn, params=[week_start, week_end] ) if rows.empty: return { "week_start": week_start, "trade_count": 0, "win_rate": 0, "avg_conf": 0, "total_return": 0, "benchmark_return": 0, "model_verdict": "NO_TRADE", } win_rate = float((rows["prob"] > 0.6).mean()) avg_conf = float(rows["confidence"].mean()) total_return = float(rows["prob"].mean() - 0.5) * 2 verdict = "HEALTHY" if win_rate >= 0.5 else "NEED_CHECK" return { "week_start": week_start, "trade_count": len(rows), "win_rate": round(win_rate, 3), "avg_conf": round(avg_conf, 3), "total_return": round(total_return, 3), "benchmark_return": 0.0, "model_verdict": verdict, }

这段聚合结果会写入weekly_review表。total_return在这里用“平均概率偏离度”近似模拟，实际项目中要换成信号对应标的在持有期内的真实收益，并且与同期基准指数对齐；示例代码把benchmark_return留空，落地时需要补上指数日频数据并做周聚合。model_verdict字段是复盘Agent最重要的输出，它会成为第6章里“要不要重训”的触发器。

5. 避坑记录与排查清单：跑通这套源代码最容易翻车的地方

5.1 时间穿越：训练分数虚高、实盘拉胯的元凶

现象：模型在回测里AUC超过0.95，但滚动预测时概率几乎回到0.5，信号总是买在局部高点。原因：最常见有两个。一是特征计算了当天收盘后才拿到的信息，比如用次日数据填充当日缺口；二是对全量特征做了标准化，让验证集统计量混进训练集。解决：特征列统一在训练前检查可用时间点，我一般取一个已知日期人工打印一行特征与K线对照；标准化必须放在TimeSeriesSplit的每一折内部完成。

5.2 双模型互斥：日频和周频信号打架，仓位反复横跳

现象：日频模型给出buy，周频模型给出hold，若调度器把两个信号取“或”处理，系统会在两三个交易日内频繁进出，手续费吞掉收益。原因：异频模型概率不能直接比较，同一只股票在短周期和长周期上方向背离是常态，系统没有约定决策层级。解决：第4章的“周频一票否决”就是对策——以日频为动作源，周频只在低于0.45时否决，0.45到0.58的模糊带一律保持hold，不要做任何优化。

5.3 数据源错位：前复权K线拼接时出现“假缺口”

现象：回测收益曲线平滑漂亮，切到实时新数据后在接口切换日出现跳变。原因：回测用整段前复权价格，实时新拉的一条K线按未复权价给出，两者拼接后出现假缺口。解决：每次预测前用K线缓存重新拉最近两年数据做全量重算，不要缓存拼接；同时过滤成交量异常为0、单日涨跌超过10%的脏K线，遇到停牌日直接跳过。

5.4 Agent遗忘上下文：复盘Agent看不见上周的复盘结论

现象：每周复盘都能输出“本周胜率X%”，但连续几周结论彼此孤立，系统完全不记得上周刚踩过“连续放量但价格不涨”的坑。原因：只存信号表，没有存Action记录，数据流每天只闭环，没有跨周的状态传递。解决：在signals表增加carry_over字段，复盘Agent写入本周结论时同时把上周verdict读进内存，生成“环境似曾相识”提醒；这条提醒可以伴随文本复盘报告一起输出，让下一周的策略Agent做参考。

5.5 模型“静默过期”：行情风格切换，权重还停留在上个阶段

现象：模型没有报错，置信度也不低，但连续多周跑输基准。原因：LGBM是决策树集成，对分布漂移有韧性但不是免疫，市场风格切换后旧特征分布已经被挤出训练空间。解决：定一个可落地的重训规则：连续3周周度复盘收益低于基准且样本量足够时，自动触发重训；重训保留原始random_state重新跑一遍，和旧模型对比再来决定切换。

提示：排查顺序永远按“先数据、再逻辑、后参数”走。数据错位和未来函数造成的坑占比最高，先确认K线时间轴干净，再信模型输出。

6. 用一次完整周度复盘验证：这周该保持、回滚还是重训

6.1 连续8周健康检查：把复盘表读成一张体检报告

周度复盘表不只是一张存档，它应该驱动下一步动作。读最近8周数据，计算每周期超额收益与健康状况：

def weekly_health_check(conn, weeks: int = 8) -> pd.DataFrame: sql = """ SELECT week_start, total_return, benchmark_return, win_rate, avg_conf, model_verdict FROM weekly_review ORDER BY week_start DESC LIMIT ? """ df = pd.read_sql_query(sql, conn, params=[weeks]) df["excess"] = df["total_return"] - df["benchmark_return"] df["healthy"] = (df["excess"] > 0) & (df["win_rate"] > 0.5) return df

excess这一列是系统相对基准的超额，healthy是是否继续沿用当前参数的布尔开关。注意只看单周没有意义，连续3周以上才值得动手。

6.2 回滚还是重训：一张表决定本周动手方向

我现在的习惯是，每周一开盘前只做一件事：把weekly_review表最近8行打出来，看excess、看win_rate。不先看当天行情，也不追任何新闻。“系统是不是还在我给它划定的边界内工作”永远比“今天会不会涨”更值得先回答。模型沉默不可怕，可怕的是沉默时你没有流程可依，只能依赖玄学。这套从0到1的多Agent+LGBM双模型源代码，造出第一个模型不难，难的是每次它失灵时你愿意相信自己的排查清单而不是临时改参数。希望帮到你。

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