简介这是一套面向量化入门者与Python爱好者的轻量级AI炒股系统实战代码聚焦股票择时决策与周度复盘两大核心需求解决初学者难以将机器学习模型落地到真实交易场景的痛点。资源共11个文件含4个核心Python模块如app.py交互入口、main.py主流程、train.py训练逻辑、utils.py工具函数、2个预训练模型文件pkl格式分别对应600025.SH的涨跌分类与涨幅回归任务、2个示例数据文件csv格式含行情与交易记录以及依赖说明与缓存文件整体仅598KB普通笔记本即可快速运行与调试。已有423人学习下载配套Streamlit界面无需前端基础点选即运行多Agent架构清晰分离选股、风控、择时与复盘模块便于理解系统设计思想并开展二次开发所有模型严格规避未来函数确保策略可回测、可实盘演进。1. 从零构建AI炒股系统的核心挑战与架构选型最近几年AI在量化投资领域的应用已经从实验室走向了实战。很多朋友私信问我有没有一个从零开始、能跑通、有逻辑、可解释的AI炒股系统搭建方案今天我就结合自己过去几年踩过的坑和趟出来的路分享一套基于多Agent协作架构和LGBM双模型的实战系统。这套系统的核心目标不是追求“圣杯”而是提供一个稳定、可迭代、支持择时与复盘的分析框架让你能真正理解市场信号而不是当一个“黑盒”的奴隶。为什么选择多Agent和LGBM这个组合这源于我对传统量化策略的反思。过去我们写一个庞大的策略脚本所有逻辑糅在一起特征工程、信号生成、风险控制互相耦合一旦市场风格切换调整起来异常痛苦就像试图修改一栋大楼的地基。而多Agent架构将复杂的投资决策过程拆解成多个专注的“智能体”比如一个负责数据清洗一个负责特征提取一个负责择时判断它们各司其职通过标准化的“语言”消息协作。这样做的好处是模块化、易维护、可解释性强你可以清晰地看到是哪个环节的判断导致了最终的交易信号。至于模型为什么是LGBMLightGBM在金融时序预测这个领域我们面对的数据往往是高维、稀疏、存在大量非线性关系的。树模型特别是梯度提升树GBDT家族在处理这类数据上有天然优势对缺失值不敏感、能自动进行特征组合、不需要复杂的标准化预处理。相比它的兄弟XGBoostLGBM在训练速度和内存消耗上优势明显这对于需要快速迭代、回测大量特征的量化场景至关重要。我选择构建“双模型”并非简单的模型融合而是让它们承担不同的任务一个模型专注于中短期趋势的捕捉择时模型另一个模型则着眼于更长周期的结构判断和复盘验证复盘模型形成决策上的交叉验证。这套系统的输出最终会落实到两个核心动作上择时分析与周度复盘。择时告诉你“现在该不该动”复盘告诉你“过去为什么那么动”以及“未来该怎么优化”。接下来我将手把手带你搭建这个系统的每一个模块并分享其中那些文档里不会写的细节和教训。2. 系统基石数据管道与特征工程Agent的实现细节任何AI系统的上限都取决于数据质量。我们的第一个Agent就是数据管道与特征工程Agent。它的使命是从原始、混乱的市场数据中提炼出干净、有效、具有预测能力的特征。2.1 数据源的选取与实时/离线管道设计数据源是第一步也是最容易埋坑的地方。对于A股市场我建议的入门组合是Tushare Pro获取基础日线/分钟线、财务数据 AkShare获取更丰富的宏观、板块、资金流数据。这里有个关键点不要将所有数据源的API调用混在一个线程里。我们应该为每个数据源设计独立的子Agent或模块它们并行获取数据然后统一交给一个“数据聚合器”处理。这样做的好处是当某一个数据源临时宕机或限流时不会导致整个数据管道崩溃。数据管道需要区分离线训练管道和在线推理管道。离线管道用于模型训练。它会一次性拉取长达数年的历史数据进行复杂的特征计算和标签生成。这个过程计算量大可以放在夜间定时任务如Apache Airflow中跑。在线管道用于实时的择时判断。它只关心最新一天的数据并基于离线管道已经计算好的特征公式如20日均线快速计算出今日的特征值。这里必须引入缓存机制比如用Redis存储前一天计算好的中间结果避免重复计算。一个典型的离线数据获取与清洗的代码骨架如下import pandas as pd import tushare as ts from datetime import datetime, timedelta class DataFetcherAgent: def __init__(self, token): ts.set_token(token) self.pro ts.pro_api() def fetch_daily_data(self, ts_code, start_date, end_date): 获取复权日线数据 # 使用pro.daily接口并指定复权因子 df self.pro.daily(ts_codets_code, start_datestart_date, end_dateend_date, adjqfq) # 关键检查数据是否连续是否有停牌导致的缺失 expected_dates pd.date_range(startstart_date, endend_date, freqB) df[trade_date] pd.to_datetime(df[trade_date]) df df.set_index(trade_date).reindex(expected_dates) # 对于停牌日采用前向填充OHLC但成交量设为0 df[[open, high, low, close]] df[[open, high, low, close]].ffill() df[vol] df[vol].fillna(0) df[ts_code] ts_code return df.reset_index()2.2 特征构造超越技术指标的阿尔法挖掘特征工程是量化模型的灵魂。除了常见的移动平均线MA、布林带Bollinger Bands、相对强弱指数RSI等技术指标我们必须挖掘更具信息量的“阿尔法特征”。价量关系特征这是核心。例如volume_ratio当日成交量 / 过去20日平均成交量。放量上涨和放量下跌的意义截然不同。price_volume_corr过去N日价格变化与成交量变化的滚动相关系数。用来判断价量是否背离。市场情绪特征利用另类数据。从AkShare获取“沪深港通资金流向”构造north_money_net_inflow北向净流入的滚动占比。计算市场宽度如当日上涨股票数量与下跌股票数量之比。时序统计特征return_skewness,return_kurtosis过去20日收益率的偏度和峰度描述收益分布形态。volatility_ratio短期波动率如5日与长期波动率如20日之比捕捉波动率突变。横截面特征这是提升模型鲁棒性的关键。将个股特征与所在行业或全市场股票进行对比。rank_close个股收盘价在过去20日内的百分位排名。zscore_volume个股成交量相对于同行业其他股票的标准分数。注意特征构造的最大陷阱是“未来函数”。确保任何在时间t使用的特征其计算仅依赖于t时刻及之前的信息。在代码中务必使用.shift(1)来将特征对齐到下一期的标签。我强烈建议将特征计算函数封装成类并内置未来函数检查。2.3 标签定义如何教会模型识别“好时机”对于择时模型标签y值的定义直接决定了模型学习的目标。二分类涨/跌简单但信息损失大。我采用的是三分类标签并结合了未来波动率调整。类别1买入信号未来N日例如5日的收益率 阈值A如1.5%且未来N日最大回撤 阈值B如2%。这不仅是要求涨还要求涨得“稳”。类别-1卖出信号未来N日收益率 阈值C如-1.5%。类别0持有信号介于两者之间。这样定义的标签迫使模型不仅要预测方向还要对未来的波动性有所考量。标签生成的代码如下def create_labels(price_series, lookahead_days5, up_threshold0.015, down_threshold-0.015, drawdown_limit0.02): future_return price_series.pct_change(lookahead_days).shift(-lookahead_days) future_max_drawdown price_series.rolling(lookahead_days).apply(lambda x: (x.max() - x.iloc[-1]) / x.max()).shift(-lookahead_days) labels pd.Series(0, indexprice_series.index) # 默认持有 buy_condition (future_return up_threshold) (future_max_drawdown drawdown_limit) sell_condition (future_return down_threshold) labels[buy_condition] 1 labels[sell_condition] -1 # 删除最后lookahead_days个因为无法计算未来数据的样本 labels labels.iloc[:-lookahead_days] return labels数据Agent的最后会输出一个标准的DataFrame索引为时间列包括所有清洗后的特征和标签并通过消息队列如Redis Pub/Sub或直接内存共享传递给下一个Agent。3. 核心引擎LGBM双模型的设计、训练与调优数据准备好后就进入了模型环节。我们设计两个LGBM模型它们共享特征输入但任务不同共同构成系统的“大脑”。3.1 择时模型聚焦短期信号捕捉择时模型Timing Model的目标是判断未来短期内如未来1-5个交易日的走势输出买入、持有或卖出的概率。它的训练有以下几个关键点样本权重并非所有样本都同等重要。市场有明显趋势单边上涨或下跌时期的样本其规律性更强应赋予更高权重。我们可以用过去一段时间的波动率或趋势强度如ADX指标来动态分配样本权重。时间序列交叉验证这是金融数据验证的黄金准则。绝对不能用随机打乱的K-Fold必须使用TimeSeriesSplit确保验证集的时间永远在训练集之后防止信息泄露。特征重要性分析LGBM训练后输出特征重要性图。定期审视剔除长期重要性很低的特征防止过拟合。你会发现价量相关特征和横截面排名特征往往是最重要的。择时模型的训练代码框架import lightgbm as lgb from sklearn.model_selection import TimeSeriesSplit import numpy as np class TimingModelAgent: def __init__(self): self.model None self.feature_names None def train(self, X, y, sample_weightNone): self.feature_names X.columns.tolist() tscv TimeSeriesSplit(n_splits5) params { objective: multiclass, # 三分类 num_class: 3, metric: multi_logloss, boosting_type: gbdt, num_leaves: 31, learning_rate: 0.05, feature_fraction: 0.8, # 每次迭代随机选80%特征增加鲁棒性 bagging_fraction: 0.8, # 类似随机森林的行采样 bagging_freq: 5, verbose: -1, seed: 42 } cv_results [] 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] w_train sample_weight.iloc[train_idx] if sample_weight is not None else None lgb_train lgb.Dataset(X_train, y_train, weightw_train, feature_nameself.feature_names) lgb_val lgb.Dataset(X_val, y_val, referencelgb_train, feature_nameself.feature_names) gbm lgb.train(params, lgb_train, num_boost_round1000, valid_sets[lgb_val], callbacks[lgb.early_stopping(50), lgb.log_evaluation(100)]) cv_results.append(gbm.best_score[valid_0][multi_logloss]) print(fCV平均LogLoss: {np.mean(cv_results):.4f}) # 用全量数据训练最终模型 final_train_data lgb.Dataset(X, y, weightsample_weight, feature_nameself.feature_names) self.model lgb.train(params, final_train_data, num_boost_round500)3.2 复盘模型着眼于周期验证与模式诊断复盘模型Review Model的目标不同。它使用更长的未来窗口例如20个交易日来评估一段时期的市场状态并用于周度复盘。它的标签可以定义为1未来一段时间出现显著上涨趋势。0震荡市。-1未来一段时间出现显著下跌趋势。这个模型的作用不是直接交易而是验证择时信号当择时模型发出买入信号时复盘模型可以判断当前是否处于一个中长期的上涨环境中提高信号的可信度。复盘分析在每周复盘时输入当前的特征复盘模型会输出对接下来中期走势的“看法”与市场实际走势对比用于诊断当前市场风格是否与模型认知匹配。两个模型的关系是协作而非主从。有时择时模型看多但复盘模型看空这可能意味着是短期反弹而非反转此时可以降低仓位。这种交叉验证机制能有效过滤掉一些假信号。3.3 超参数调优与模型更新策略LGBM的超参数不少手动调优效率低。我使用Optuna库进行贝叶斯优化。需要优化的核心参数包括num_leaves控制模型复杂度、learning_rate、feature_fraction、bagging_fraction等。目标函数就是时间序列交叉验证的平均LogLoss。实操心得模型不是一劳永逸的。市场在变模型的效力会衰减。我设定的规则是每日监控记录模型在最近一个月滚动窗口内的预测准确率或AUC。当准确率连续一周低于阈值如55%触发警报。月度重训无论表现如何每月用截至上月末的最新数据重新训练一次模型保持模型对最新市场信息的敏感度。季度大更新每季度重新进行一轮完整的特征筛选和超参数优化。4. 多Agent协作框架的工程实现与消息流前面提到了多Agent现在我们来具体实现这个协作框架。我们主要需要四个Agent它们通过一个中央消息总线Message Bus进行通信。这里为了简化我们用Python的multiprocessing模块和Queue来实现进程间通信生产环境可以考虑Celery或Ray。4.1 Agent角色定义与消息协议DataAgent数据Agent如前所述负责定时获取、清洗、计算特征。它完成后会向消息总线发布一条消息主题为data.ready消息体包含数据存储的路径或唯一标识。FeatureAgent特征Agent监听data.ready消息。它负责从原始数据中提取出模型所需的最终特征向量。它可能会进行一些在线特征计算如最新的横截面排名。完成后发布features.ready消息。ModelAgent模型Agent这是一个复合Agent内部包含择时模型和复盘模型两个实例。它监听features.ready消息。收到后加载最新的特征数据分别用两个模型进行推理。它将两个模型的预测结果例如择时模型预测的买入概率、复盘模型预测的趋势强度组合成一个综合信号。然后发布prediction.ready消息。DecisionAgent决策Agent监听prediction.ready消息。它是风控和执行的最终关卡。它根据综合信号、当前账户持仓、预设的风控规则如单日最大亏损、总仓位上限来生成最终的交易指令买/卖/调仓。它不直接执行交易而是将指令发布到order.command主题由下游的执行系统处理。消息格式可以统一使用JSON例如{ topic: features.ready, timestamp: 2023-10-27 15:00:00, payload: { data_id: feature_set_20231027, features: {ts_code: 000001.SZ, ma5: 12.5, volume_ratio: 1.2, ...} } }4.2 核心协作流程代码示例以下是利用multiprocessing模拟的简化版流程import multiprocessing as mp import time import json def data_agent(data_queue, feature_queue): 模拟数据Agent while True: # 模拟数据准备过程 time.sleep(2) mock_data {data_id: fdata_{int(time.time())}, raw: processed_data} data_queue.put(json.dumps({topic: data.ready, payload: mock_data})) print(DataAgent: 数据已就绪) def feature_agent(data_queue, feature_queue, model_queue): 模拟特征Agent while True: msg data_queue.get() msg_dict json.loads(msg) if msg_dict[topic] data.ready: # 模拟特征计算 time.sleep(1) features {ts_code: 000001.SZ, signal_strength: 0.75} feature_queue.put(json.dumps({topic: features.ready, payload: features})) print(FeatureAgent: 特征已就绪) def model_agent(feature_queue, decision_queue): 模拟模型Agent while True: msg feature_queue.get() msg_dict json.loads(msg) if msg_dict[topic] features.ready: # 模拟模型预测 time.sleep(0.5) timing_pred 0.6 # 买入概率 review_pred 0.8 # 看多强度 combined_signal timing_pred * 0.7 review_pred * 0.3 # 加权综合信号 decision_queue.put(json.dumps({ topic: prediction.ready, payload: {combined_signal: combined_signal, timestamp: time.time()} })) print(fModelAgent: 预测完成综合信号强度 {combined_signal:.2f}) def decision_agent(decision_queue): 模拟决策Agent while True: msg decision_queue.get() msg_dict json.loads(msg) if msg_dict[topic] prediction.ready: signal msg_dict[payload][combined_signal] # 简单决策规则 if signal 0.6: action BUY elif signal 0.4: action SELL else: action HOLD print(fDecisionAgent: 生成指令 [{action}] 基于信号 {signal:.2f}) if __name__ __main__: data_queue mp.Queue() feature_queue mp.Queue() decision_queue mp.Queue() processes [ mp.Process(targetdata_agent, args(data_queue, feature_queue)), mp.Process(targetfeature_agent, args(data_queue, feature_queue, decision_queue)), mp.Process(targetmodel_agent, args(feature_queue, decision_queue)), mp.Process(targetdecision_agent, args(decision_queue,)) ] for p in processes: p.start() for p in processes: p.join()这种架构的扩展性极强。如果你想增加一个监控市场情绪的Agent只需要让它监听data.ready消息生成情绪指标然后发布sentiment.ready消息让ModelAgent同时监听features.ready和sentiment.ready等两者都到达后再进行预测即可。5. 择时信号生成、回测与周度复盘闭环系统跑起来后我们每天会得到择时信号。但信号如何转化为可执行的策略历史表现如何这就需要回测和复盘。5.1 从模型概率到交易信号阈值动态调整模型输出的是属于各个类别的概率如[0.1, 0.8, 0.1]分别对应跌、平、涨的概率。我们需要一个阈值来将概率转化为行动。一个常见的错误是使用固定阈值如买入概率0.7就买。更好的方法是动态阈值。我们可以根据模型在最近一段时间如过去100个交易日的预测结果计算出一条概率-正确率曲线。然后选择一个能使样本外夏普比率或Calmar比率最大的阈值作为当前交易阈值。例如通过回测发现当买入概率大于0.65时后续上涨的确定性最高那么这个0.65就是当前的动态买入阈值。这个阈值可以每周或每月更新一次。5.2 回测系统搭建的关键细节回测不是简单的“信号出现就买卖”。必须考虑以下细节否则回测结果会严重失真“过拟合”的另一种形式交易成本必须包含佣金如万分之三和印花税卖出时千分之一。对于小资金滑点假设0.1%的影响也很大。仓位管理是满仓进出还是固定比例如每次20%我推荐使用凯利公式或固定分数法的动态仓位管理。根据模型的预测概率和历史胜率、盈亏比来计算每次投入的本金比例。交易频率限制避免过于频繁的交易。可以设置最小持仓周期如至少持有3天来过滤信号。基准对比回测结果一定要和同期的大盘指数如沪深300做对比看是否有超额收益Alpha。一个简单的回测引擎核心逻辑class BacktestEngine: def __init__(self, initial_capital100000, commission_rate0.0003, tax_rate0.001): self.capital initial_capital self.position 0 # 持有股数 self.commission_rate commission_rate self.tax_rate tax_rate self.trade_log [] def execute_trade(self, date, price, signal, prob): signal: 1 买入, -1 卖出, 0 持有 if signal 1 and self.position 0: # 买入逻辑考虑动态仓位 max_position_value self.capital * self.position_sizing(prob) # 仓位管理函数 can_buy_shares int(max_position_value / (price * (1 self.commission_rate))) if can_buy_shares 0: cost can_buy_shares * price * (1 self.commission_rate) self.capital - cost self.position can_buy_shares self.trade_log.append({date:date, action:BUY, price:price, shares:can_buy_shares, capital:self.capital}) elif signal -1 and self.position 0: # 卖出逻辑 sell_value self.position * price tax sell_value * self.tax_rate commission sell_value * self.commission_rate net_proceeds sell_value - tax - commission self.capital net_proceeds self.trade_log.append({date:date, action:SELL, price:price, shares:self.position, capital:self.capital}) self.position 05.3 周度复盘系统优化的核心反馈环周度复盘不是看盈亏而是诊断系统。我每周日晚上会运行复盘脚本主要做以下几件事信号有效性分析统计本周发出的所有买卖信号计算信号发出后N日如3日、5日的平均收益率、胜率。对比择时模型和复盘模型的预测看是否出现重大分歧分歧点在哪里。特征贡献度分析使用SHAPSHapley Additive exPlanations值分析本周重要交易日的预测结果是哪些特征主导了模型的判断这些特征的变化是否符合逻辑例如如果发现“北向资金净流入”这个特征在下跌日给出了很高的正向SHAP值那就要去检查数据源是否出了问题或者市场逻辑是否发生了变化。市场状态匹配度将复盘模型对本周市场状态的判断上涨/震荡/下跌与实际走势对比。如果不匹配需要分析是模型问题还是市场出现了未学习过的极端情况如政策突发。生成复盘报告自动生成一份Markdown报告内容包括本周盈亏概况、信号统计、重要特征分析、模型信心指数、以及下周需要关注的重点例如“模型显示波动率特征重要性上升下周需警惕高波动风险”。这个复盘闭环是系统持续进化的动力。基于复盘发现的问题你可以去调整特征、重新训练模型、或者修改决策Agent的风控参数。6. 生产环境部署、监控与持续迭代一个能跑在回测里的系统和一个能7x24小时稳定运行的生产系统是两回事。以下是部署和运维的关键点。6.1 系统部署与调度推荐使用Docker容器化部署每个Agent。这保证了环境的一致性也便于扩展。使用docker-compose来编排所有服务。调度方面DataAgent需要定时触发如每天收盘后。可以用Linux的cron或者更专业的Apache Airflow来定义整个工作流DAG有向无环图。一个简单的docker-compose.yml示例version: 3 services: redis: image: redis:alpine container_name: ai_trading_bus ports: - 6379:6379 data_agent: build: ./data_agent depends_on: - redis environment: - REDIS_HOSTredis # 可以通过cron或Airflow从外部触发 model_agent: build: ./model_agent depends_on: - redis environment: - REDIS_HOSTredis restart: unless-stopped # 设置自动重启6.2 监控与告警系统无人值守监控必须到位。数据质量监控DataAgent在获取数据后应检查数据完整性有无缺失日期、异常值涨跌幅超过10%是否合理。发现问题立即发送告警如通过钉钉/企业微信机器人。模型性能监控每天记录模型对最新数据的预测概率并跟踪预测后的实际涨跌。计算滚动窗口内的准确率、AUC等指标。当指标持续下滑时告警。Agent健康监控每个Agent定期向一个监控中心发送“心跳”。如果某个Agent失联监控系统应能重启容器或通知管理员。交易指令监控DecisionAgent发出的每一个交易指令都必须被详细日志记录并最好有二次确认机制例如将指令发送到手机App确认后再执行。6.3 持续迭代没有完美的系统只有不断进化的系统这个系统搭建完成后只是一个起点。市场在变你的认知也在变系统必须迭代。月度小迭代每月末用过去几个月的新数据重新训练模型增量训练或全量训练更新动态阈值。季度中迭代每季度重新审视特征池。根据SHAP分析和市场新逻辑加入可能的新特征例如新的宏观指标剔除失效的老特征。年度大迭代每年对系统架构进行回顾。是否需要引入新的Agent如一个专门做舆情分析的NLP Agent消息总线是否需要从Redis换成Kafka以应对更大数据量模型是否需要尝试新的架构如Transformer时序模型与LGBM结合记住这个多Agent的LGBM双模型系统最大的优势不是它当前多厉害而是它的模块化和可扩展性。每一个环节都可以被单独替换、升级、优化而不会牵一发而动全身。这让你能在一个稳固的框架内持续探索AI与市场博弈的奥秘。从0到1搭建的过程固然充满挑战但当你看到系统开始自主地分析、判断、并给出有逻辑的信号时那种成就感是无与伦比的。希望这份超详细的指南能帮你少走弯路更快地构建起属于自己的AI投资分析框架。本文还有配套的精品资源点击获取