拓冰网站定制 
简介一份面向Python爱好者与量化入门者的轻量级AI炒股系统完整工程核心解决本地轻量运行与二次开发需求。系统采用多Agent架构将选股、风控、择时、复盘解耦同时落地分类与回归双模型既能判断涨跌方向又可预测涨幅区间且预测严格遵循无未来函数原则仅使用当日及之前数据。自带Streamlit交互界面无需编写前端即可完成单票深度择时与周度复盘报告生成普通笔记本即可运行。压缩包共11个文件以py源代码为主辅以两个模型pkl权重文件、CSV交易数据、以及编译生成的pyc缓存整体大小598KB结构精简、便于按模块学习与改造。目前已有440人学习参考适合想快速搭建可解释、可扩展AI选股框架的读者入手可基于现有模型与界面逻辑直接进行策略回测、参数调优或数据源替换。1. 从0到1搭AI炒股系统为什么是“多AgentLGBM”而不是堆深度学习模型很多人一听到AI炒股系统第一反应是上深度学习或者强化学习模型。但摸过真实行情数据之后我的结论是日线级别的预测梯度提升树的可解释性和鲁棒性对表格特征更好用。标题里的多Agent架构本质上是把数据、特征、信号、复盘四条流水线拆开让每一步都可独立验证LGBM双模型则分别负责日频择时和周度复盘用两个视角去确认同一个交易决策。这套方案适合有Python基础、想把交易想法转成可回测、可运行的源代码系统的开发者。先把流程闭环跑通再慢慢优化策略比一开始就追求模型复杂度更实际。2. 先把架构立住多Agent怎么分工LGBM双模型又怎么配合2.1 多Agent不是聊天机器人四条流水线各管一段提到多Agent很多人会联想到大语言模型那套智能体编排但在这个场景里Agent更接近“职责独立的服务模块”。我一开始就把系统拆成四个AgentDataAgent负责行情获取与落库FeatureAgent负责因子计算与特征缓存SignalAgent负责加载模型并生成择时信号ReviewAgent负责周度复盘报告。它们之间不直接互相调用函数而是通过中间状态文件或数据库表传递数据。这样做的原因很直接任何一个环节崩了可以单独重跑不把整条链路带崩任何一个环节想换实现比如把数据源从tushare换成akshare只需要改一个模块。下面是最小可运行的多Agent骨架代码# agents.py — 多Agent最小骨架 # 设计原则每个Agent只暴露run()入口输入输出都是文件路径便于单独重试 from pathlib import Path import pandas as pd class DataAgent: 数据Agent拉取行情 - 简单清洗 - 落盘parquet def __init__(self, data_dir: str ./data): self.data_dir Path(data_dir) self.data_dir.mkdir(exist_okTrue) def run(self, symbol: str, start: str, end: str) - str: # 生产环境这里用akshare/tushare拉真实日线 # 下面生成一份确定性示例数据先把流程跑通 dates pd.date_range(start, end, freqB) df pd.DataFrame({ close: 10 (dates.dayofyear % 7) * 0.5, volume: 1_000_000 (dates.dayofyear % 5) * 100_000, }, indexdates) df[open] df[close] * 0.99 df[high] df[close] * 1.02 df[low] df[close] * 0.98 path self.data_dir / f{symbol}.parquet df.to_parquet(path) return str(path) class FeatureAgent: 特征Agent读行情算技术指标输出特征文件 def run(self, path: str) - str: df pd.read_parquet(path) ema12 df[close].ewm(span12).mean() ema26 df[close].ewm(span26).mean() df[macd_dif] ema12 - ema26 df[macd_dea] df[macd_dif].ewm(span9).mean() df[macd_hist] (df[macd_dif] - df[macd_dea]) * 2 # 简化RSI正式实现在第3章 df[rsi14] df[close].pct_change().rolling(14).mean() df[vol_ma20] df[volume].rolling(20).mean() df[vol_ratio] df[volume] / df[vol_ma20] feature_path path.replace(.parquet, _feat.parquet) df.to_parquet(feature_path) return feature_path这段代码的逻辑说明DataAgent的run方法入参是股票代码和起止日期出参是parquet文件路径FeatureAgent接住这个路径算出macd、rsi、量比等基础特征再写一份新文件。中间没有共享内存变量也没有全局状态所以任何一个Agent都能独立重放历史数据。参数说明DataAgent示例里没有体现freq参数真实工程里应该再收一个freq参数daily/min5用来切换日线和分钟线FeatureAgent里的ewm(span12)是MACD快线参数默认12日是美国标准A股有人改成10日这类参数应该后移到config.yaml而不是写死在代码里。2.2 LGBM双模型不是“两个一样的模型”择时与复盘的技术差异标题里的LGBM双模型我拆成两个TimingModel日频择时模型和ReviewModel周频复盘模型。很多人的直觉是“训练两个一样的模型再平均”但这两者的样本构造和标签定义完全不同。TimingModel的输入是每日特征向量标签是未来3日的收益是否超过阈值作用是在每个交易日收盘后给出次日仓位建议偏“战术”。ReviewModel的输入是过去5个交易日的聚合特征——周涨幅、周均量比、周内信号分布、胜率统计——标签是未来一周能否跑赢沪深300指数作用是判断该继续持有还是清仓偏“战略”。为什么不能共用一个模型日频模型面对的是高噪声、低信噪比的样本正负样本比例常常接近1:9周频模型拿到的样本更规整但样本量大幅缩小需要不同的采样策略和正则化强度。因此我在代码里把两个模型的训练入口分开train_timing.py和train_review.py配置文件里也使用两个独立的model节。2.3 最小工程骨架目录结构、配置文件和主流程我倾向于把整个项目按功能拆成下面这个结构任何人扫一眼目录就知道每个文件是干什么的ai_stock/ ├── agents/ │ ├── __init__.py │ ├── data_agent.py # 行情数据 │ ├── feature_agent.py # 特征计算 │ ├── signal_agent.py # 模型推理与信号生成 │ └── review_agent.py # 周度复盘 ├── config/ │ └── config.yaml # 参数集中管理 ├── data/ # parquet数据文件 ├── models/ # 训练好的模型产物 ├── reports/ # 周度复盘报告输出 ├── train_timing.py # 择时模型训练 ├── train_review.py # 周度复盘模型训练 └── run_pipeline.py # 每日主流程run_pipeline.py的任务非常简单就是把四个Agent串起来# run_pipeline.py — 每日主流程 # 参数统一从config读取不写死在业务代码里 import yaml from agents.data_agent import DataAgent from agents.feature_agent import FeatureAgent def main(): with open(config/config.yaml, r, encodingutf-8) as f: cfg yaml.safe_load(f) data_agent DataAgent(data_dircfg[data][dir]) feat_agent FeatureAgent() for symbol in cfg[universe][symbols]: raw_path data_agent.run( symbolsymbol, startcfg[data][start_date], endcfg[data][end_date] ) feat_path feat_agent.run(raw_path) # 后续交给SignalAgent和ReviewAgent if __name__ __main__: main()逻辑说明每个Agent都是同一个模式——构造函数负责依赖注入run方法负责执行。config.yaml控制所有易变参数股票池、训练日期、特征开关、模型路径等。参数说明cfg[universe][symbols]在真实环境里可能是一份沪深300成分股列表几百个标的跑下来DataAgent的单线程拉取会非常慢我通常在这里直接换用ThreadPoolExecutor并发拉取但要控制并发数在10以内避免被数据源限流。3. 数据流水线与特征工程行情原始数据怎么变成模型能吃的特征3.1 行情数据落库日线为主、分钟线做验证复权因子不能忘第一次写AI炒股系统的人很容易直接拿复权后的日线数据开练。但真正落地时我发现一个关键问题前复权数据会随着后续除权事件发生而整体变化如果你在历史回测和实盘之间共用同一份前复权数据模型会“看到”未来才出现的价格修正。所以我通常用后复权方式做训练交易信号生成后用不复权价做修正。下面是一段真实使用的取数逻辑# data_agent.py 里的真实取数片段 # 目标拿到后复权日线并落库保留原价列用于实盘修正 import akshare as ak import pandas as pd def fetch_daily_post_adjusted(symbol: str, start: str, end: str) - pd.DataFrame: # stock_zh_a_hist 返回未复权数据需要自己算复权因子 raw ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart, end_dateend, adjust ) raw.columns [date, open, close, high, low, volume, amount, amplitude, pct_change, change, turnover] # 基于收盘价序列构造累计复权因子简化版 # 更严谨的做法是从数据源拿hfq_factor字段 raw[adj_factor] (raw[close] / raw[close].iloc[0]).cummax() raw[close_hfq] raw[close] * raw[adj_factor] raw[open_hfq] raw[open] * raw[adj_factor] raw[high_hfq] raw[high] * raw[adj_factor] raw[low_hfq] raw[low] * raw[adj_factor] return raw逻辑说明这里的核心是adj_factor字段用cummax模拟累计复权因子真实生产环境应从数据源直接读取复权因子字段。参数说明perioddaily表示日线如果你要做盘中择时可以考虑period60之类的分钟线但分钟线数据量与日线不在一个量级存储和计算成本都会暴涨。我的建议是先用日线跑通闭环再根据回测结果决定是否需要分钟线。3.2 特征工程MACD、RSI、量比、布林带特征设计不需要太多花哨因子。先保证基础的技术指标正确MACD、RSI、布林带、量比、动量。这些指标是公开的但实现细节千差万别稍不留意就和其他人的结果对不上。# feature_agent.py 的核心特征计算 # 注意这里所有计算都基于后复权价格避免除权跳变污染指标 def add_indicators(df: pd.DataFrame) - pd.DataFrame: close df[close_hfq] high df[high_hfq] low df[low_hfq] volume df[volume] # ---- MACD ---- ema12 close.ewm(span12, adjustFalse).mean() ema26 close.ewm(span26, adjustFalse).mean() df[macd_dif] ema12 - ema26 df[macd_dea] df[macd_dif].ewm(span9, adjustFalse).mean() df[macd_hist] 2 * (df[macd_dif] - df[macd_dea]) # ---- RSI(14) ---- delta close.diff() gain delta.clip(lower0).rolling(14).mean() loss (-delta.clip(upper0)).rolling(14).mean() rs gain / (loss 1e-9) df[rsi14] 100 - 100 / (1 rs) # ---- 布林带(20, 2) ---- mid close.rolling(20).mean() std close.rolling(20).std() df[boll_upper] mid 2 * std df[boll_lower] mid - 2 * std df[boll_pos] (close - df[boll_lower]) / (df[boll_upper] - df[boll_lower] 1e-9) # ---- 量比当日成交量 / 过去5日均量---- df[vol_ratio5] volume / (volume.rolling(5).mean() 1e-9) # ---- 动量20日收益---- df[ret_20d] close.pct_change(20) return df逻辑说明MACD用adjustFalse的ewm和主流行情软件一致RSI用标准的SMA平滑版本而不是Wilder指数平滑两者数值有差异但这不影响模型训练只要全流程统一即可。参数说明boll的20日和2倍标准差是默认参数可以用config里的feature_params节覆盖。计算完成后我习惯把所有NaN行直接drop模型训练阶段宁可少样本也不要让缺失值污染分裂点。MACD双底这类形态特征后续可以做成窗口扫描特征基础因子先扛住再谈形态。3.3 标签构造用未来3日收益分档而不是简单的涨跌标签构造是整个系统里最影响上限的环节。不要把标签设成“明天涨/跌”这种二分类噪声太大。我的做法是构造未来3日的累计收益再按阈值分档。# label_engineering.py — 构造分类标签 # 核心思路用未来3日收益的分位数作为分档边界 def make_labels(df: pd.DataFrame, horizon: int 3) - pd.DataFrame: # 未来horizon日收益shift是核心保证只用历史信息 df[future_ret] df[close_hfq].shift(-horizon) / df[close_hfq] - 1.0 # 按整段区间的分位数分三档0下跌1震荡2上涨 low_q df[future_ret].quantile(0.3) high_q df[future_ret].quantile(0.7) df[label] 0 df.loc[df[future_ret] high_q, label] 2 df.loc[(df[future_ret] low_q) (df[future_ret] high_q), label] 1 # 最后几行未来收益为空直接删除避免NaN标签进入训练 df df.dropna(subset[future_ret, label]) return df逻辑说明shift(-horizon)是标准做法表示把未来第3天的收盘价移动到今天这一行和今天的收盘价比对得到未来收益。参数说明quantile(0.3)和quantile(0.7)决定了三个类别的样本比例如果你用的股票池是沪深300这两个分位点得到的样本分布与用中证500完全不同需要重新计算。这类参数我一样放进config里而不是每跑一次就改代码。3.4 数据质量检查停牌、缺失值、极端值的处理数据质量这块吃过的亏最多。停牌日没有成交数据如果你直接用ffill向前填充模型会以为这些天有交易产生无法交易的样本直接导致训练集泄漏。我通常的做法是# data_quality.py — 数据质量检查 def check_quality(df: pd.DataFrame) - dict: report {} # 1. 缺失比例 report[missing_ratio] df.isna().mean().to_dict() # 2. 零成交检查成交量为0的日期列表 zero_vol_days df[df[volume] 0].index.tolist() report[zero_volume_days] len(zero_vol_days) # 3. 极端值检查单日涨跌超过20%的记录 extreme df[df[pct_change].abs() 0.2] report[extreme_days] len(extreme) # 4. 连续停牌段检查 df[is_trading] df[volume] 0 df[group] (df[is_trading] ! df[is_trading].shift()).cumsum() pause_groups df[~df[is_trading]].groupby(group).size() report[longest_pause] int(pause_groups.max()) if len(pause_groups) 0 else 0 return report逻辑说明这个check_quality函数输出一份简单的质量报告让你在训练之前就知道这份数据能不能用。longest_pause字段特别重要如果一只股票连续停牌超过10天复牌后的跳空会直接影响模型对特征的判断——复牌第一天的pct_change往往异常大如果你没做截断处理模型会被这个极端值带偏。通常我的做法是把复牌后第一天的极端涨跌做winsorize比如限制在±10%或者干脆把复牌日样本从训练集中剔除。4. 训练LGBM双模型参数、验证和择时信号生成4.1 择时模型的训练样本权重、类别不平衡与早停前面我把标签设成了三分类下跌/震荡/上涨但震荡类样本在决策上没有可操作性。所以实际训练时我让模型只输出“上涨概率”这个连续值即把label2的样本视为正样本其余视为负样本。如果一定要多分类就用LightGBM的multi_logloss只是后续信号生成时要额外处理。# train_timing.py — 择时模型训练入口 import lightgbm as lgb import pandas as pd import numpy as np from sklearn.model_selection import TimeSeriesSplit def train_timing_model(feature_path: str, config: dict) - lgb.Booster: df pd.read_parquet(feature_path) # 前面的标签构造已经存在df[label]里0/1/2 # 这里把label变成二值0不涨原0和11上涨原2 df[target] (df[label] 2).astype(int) feature_cols config[timing_model][features] X df[feature_cols] y df[target] # 时序切分不用随机K折避免未来信息泄漏 tscv TimeSeriesSplit(n_splits5) params { objective: binary, metric: binary_logloss, learning_rate: 0.05, num_leaves: 31, min_data_in_leaf: 50, feature_fraction: 0.8, bagging_fraction: 0.8, bagging_freq: 1, verbose: -1, seed: 42, } best_iters [] for train_idx, val_idx in tscv.split(X): dtrain lgb.Dataset(X.iloc[train_idx], y.iloc[train_idx]) dvalid lgb.Dataset(X.iloc[val_idx], y.iloc[val_idx]) booster lgb.train( params, dtrain, num_boost_round2000, valid_sets[dvalid], callbacks[lgb.early_stopping(100), lgb.log_evaluation(200)], ) best_iters.append(booster.best_iteration) # 用全部数据以最佳迭代次数的中位数重新训练最终模型 best_iter int(np.median(best_iters)) final_dtrain lgb.Dataset(X, y) final_model lgb.train(params, final_dtrain, num_boost_roundbest_iter) final_model.save_model(models/timing_model.txt) return final_model逻辑说明TimeSeriesSplit是关键——它按时间顺序切训练集和验证集保证验证集永远在训练集之后避免“用未来预测过去”的伪高准确率。early_stopping的100轮是patience参数意思是验证集指标连续100轮不改善就提前终止。参数说明min_data_in_leaf50是防止过拟合的重要旋钮数据量小就调大到100feature_fraction和bagging_fraction同时启用相当于对特征和样本都做了随机采样是LightGBM里最常见的防过拟合组合。这里有一个常见的翻车点直接对全量数据集用train_test_split随机切分一旦测试集日期早于训练集回测结果就会虚假地好。4.2 复盘模型的训练周度聚合特征周度复盘模型不能拿每日特征直接怼进去因为它的任务和日频模型完全不同。把一周的特征做聚合累计收益、平均量比、信号分布、实际盈亏。这些聚合特征让模型能回答“这周表现与历史哪些状态最像”。# train_review.py — 周度复盘模型训练入口 def build_weekly_features(daily_df: pd.DataFrame) - pd.DataFrame: # 按周聚合周一为每周起点 weekly daily_df.resample(W-MON).agg({ ret_20d: last, rsi14: mean, vol_ratio5: mean, macd_hist: sum, close_hfq: lambda x: x.iloc[-1] / x.iloc[0] - 1, # 周收益 }).dropna() weekly.columns [momentum, avg_rsi, avg_vol_ratio, macd_power, weekly_ret] return weekly def train_review_model(weekly_df: pd.DataFrame, config: dict) - lgb.Booster: # 标签未来一周是否收益为正作为正样本 weekly_df[target] (weekly_df[weekly_ret].shift(-1) 0).astype(int) weekly_df weekly_df.dropna(subset[target]) # 周频样本量小用更保守的参数 params { objective: binary, metric: auc, learning_rate: 0.03, num_leaves: 15, min_data_in_leaf: 10, verbosity: -1, } X weekly_df[[momentum, avg_rsi, avg_vol_ratio, macd_power]] y weekly_df[target] model lgb.train(params, lgb.Dataset(X, y), num_boost_round200) model.save_model(models/review_model.txt) return model逻辑说明周频模型的样本量只有日频的五分之一所以参数要更保守num_leaves15、learning_rate0.03、num_boost_round200固定轮数不再做早停因为样本太少早停不稳定。参数说明resample(W-MON)表示把时间序列按每周一为起点重新分组这个窗口起点对周度复盘很重要——如果你从周四开始聚合得到的就是“跨周”的数据语义上很别扭。4.3 LightGBM必调参数清单我整理了一份实战中必调的参数表这些参数每一个都直接影响过回测收益曲线参数影响我的经验值调参方向num_leaves过拟合核心旋钮15-50数据量小就调小数据量大再加大min_data_in_leaf防止学到单样本规律50-100单样本预测值异常时增大learning_rate收敛速度与精度折中0.03-0.05小学习率配多轮数会更好feature_fraction特征采样比例0.7-0.9特征多、噪声大时调小bagging_fraction样本采样比例0.7-0.9配合bagging_freq1使用lambda_l2正则化1.0-5.0验证集和训练集差距大时增大min_split_gain最小分裂增益0.0-0.1分裂过碎时增大这些参数的调整顺序也有讲究。我一般先固定learning_rate0.05和num_boost_round2000然后调num_leaves和min_data_in_leaf确认验证集指标不再上涨后再调feature_fraction和bagging_fraction最后用早停确认最优轮数。反过来从正则化参数开始调往往很难定位问题。提示训练日志里如果看到“No further splits with positive gain”这类信息先别急着加特征多半是min_data_in_leaf或num_leaves太小让模型找不到合适的分裂点。4.4 信号合成双模型怎么联合出买卖决策模型训练完之后还需要一套信号合成逻辑。我的实现很简单每日收盘后TimingModel输出未来3日上涨概率p_t每周五收盘后ReviewModel输出下周继续向上的概率p_w。合成规则如下# signal_agent.py — 双模型信号合成 def generate_signal(timing_prob: float, review_prob: float) - str: # 规则先跑通再优化 if timing_prob 0.65 and review_prob 0.5: return buy elif timing_prob 0.35 and review_prob 0.5: return sell else: return hold逻辑说明0.65和0.35是两个概率阈值不是模型参数而是交易策略参数。为什么ReviewModel的阈值设在0.5而不是0.65因为周频样本更少、概率估计本身更保守不需要那么苛刻的阈值。参数说明预算足够的团队会用网格搜索去扫这两个阈值但在个人项目里建议先从固定阈值开始积累一段时间回测数据后再用验证集反馈调整。到这里信号合成和回测可以联动。简单回测# backtest.py — 最小回测逻辑只算日频收益 def simple_backtest(df_with_signal: pd.DataFrame, initial_cash: float 100_000): cash initial_cash position 0 # 持仓股数 for i in range(len(df_with_signal)): row df_with_signal.iloc[i] if row[signal] buy and position 0: position cash / row[close_hfq] cash 0 elif row[signal] sell and position 0: cash position * row[close_hfq] position 0 # 最终资产 现金 持仓市值 final cash position * df_with_signal[close_hfq].iloc[-1] return final / initial_cash - 1逻辑说明这份回测只做了最简单的“满仓/空仓”切换适合看模型能不能跑出方向感。真正的回测还应该加入手续费和滑点这部分放在第5章展开。5. 避坑与常见问题数据、模型、回测里最容易翻车的地方5.1 前视偏差复权因子用错回测曲线好得吓人现象回测曲线一路向上年化收益高得吓人实盘一跑就开始亏。 原因用了前复权数据训练而前复权价格序列会随着后续除权除息而整体变化模型相当于提前看到了“未来的复权因子”。前复权的本质是把历史价格按照当前价格水平做整体拉伸每一次除权都会改变全部历史价格训练和回测只要跨过除权日特征数值就包含了未来信息。 解决改用后复权数据训练信号生成时再映射回真实成交价。如果数据源只提供前复权那么每次除权事件发生后就重拉历史数据且保证训练窗口和回测窗口不能跨越同一个除权日太久。我自己的做法是数据文件里单独存一列adj_factor每次训练前用这列做一致性校验发现因子序列有跳变就报警。5.2 停牌数据泄漏模型学会了“停牌后必涨”现象回测中每次复牌后都触发买入信号而且这部分交易贡献了大部分利润。 原因停牌期间成交量是0pct_change也是0这些假样本被当成“横盘整理”输入模型模型学会了在低波动、低成交量的状态下重仓买入——这些样本在现实中根本没有可成交的价格复牌后的跳空却会计入收益。 解决训练前把非交易日停牌日全部剔除或者留一个is_trading列作为特征但训练时禁用它。更极端的情况是长时间停牌后的复牌首日直接剔除样本因为这类样本的量价数据在模型眼里完全失真。检查方法也很简单统计一下训练集里“volume0”的行数如果不为0说明数据清洗没做到位。5.3 滑点设置太理想回测盈利实盘打脸现象回测收益接近70%实盘换手一次就亏掉2%的价差。 原因回测按收盘价成交忽略了冲击成本和买卖价差。小盘低价股的流动性远不如沪深300成分股模型专挑这些票容易出现“信号出来但根本买卖不动”的情况尤其当信号集中在尾盘时盘口只有卖单没有买单。 解决在回测里加固定滑点比如万分之五和千分之一的交易手续费然后在信号过滤里限制一天只交易一单对于单日成交量小于某个阈值的信号直接丢弃。更严格的做法是模拟逐笔撮合但个人项目里先加固定滑点就够了等模型跑出真金白银的收益曲线再上复杂撮合。5.4 特征穿越用当日收盘数据预测当日涨跌现象模型在验证集上AUC高达0.93实盘预测却断崖下跌。 原因构造标签时没有使用shift(-horizon)或者在计算特征时用了当日收盘价又把当日涨跌当成未来涨跌——这种错误最常见的是把当日的pct_change同时作为特征和标签的原料。 解决代码里严格区分t时刻可用信息与t1时刻信息。一个简单的检查方法是取某一天的真实历史特征丢给模型推理输出概率再和你已知的第二天结果对比看模型是不是仅仅因为看了当天收盘价才给的信号。这个问题我再补充一个细节检查特征时用特征的最后一行去推理——如果这一天的收益已经体现在特征里了比如ret_20d包含了当天就不应该再用它预测当天两者时间戳一旦对齐就泄漏了。6. 周度复盘模块落地自动报告、版本管理与人工复核6.1 自动生成复盘报告ReviewAgent在每周五收盘后把本周全部信号和结果汇总生成一份Markdown报告。下面是最精简的生成逻辑# review_agent.py — 周度报告核心片段 def generate_weekly_report(signals, results, week_tag): hit (results[pct_change] 0).sum() lines [ f# 周度复盘 {week_tag}, f信号数 {len(results)}上涨命中 {hit}命中率 {hit/len(results):.1%}, ] Path(freports/weekly_{week_tag}.md).write_text( \n.join(lines), encodingutf-8 ) return freports/weekly_{week_tag}.md这篇报告的核心不是看历史好坏而是观察模型置信度在接下来一周是否会回落。我每周五收盘后都会先跑一遍报告再决定下周要不要降低仓位。6.2 模型版本管理给每次训练留后悔药新模型上线前我会把训练参数、指标、数据区间写到一个json文件def save_model_with_meta(model, metrics, tag): model.save_model(fmodels/{tag}.txt) meta { metrics: metrics, config: load_current_config(), created_at: time.strftime(%Y-%m-%d), } with open(fmodels/{tag}.json, w, encodingutf-8) as f: json.dump(meta, f, ensure_asciiFalse, indent2)tag我习惯用训练结束日期比如timing_model_20250115。加载模型时先看json里的config和数据区间就能知道这个模型当时是在什么环境下训练的旧模型也能回滚。6.3 人工复核清单与我的习惯我的复盘清单固定五条置信度在0.5到0.65之间的信号不操作双模型分歧超过30%的个股只观察不加仓命中率连续两周下滑先查特征分布漂移消息面出大事件的个股跳过模型判断新模型先幽灵跑两周再上岗。这套方法里最实用的验证技巧就是幽灵跟踪让新模型只记录信号不下单两周后对比旧模型的命中率。我每次训练完都靠这一步决定要不要替换模型也靠它把参数调整的翻车风险压到最低。希望这条从0到1搭建AI炒股系统的落地路径能帮到你少走我当年走过的弯路。本文还有配套的精品资源点击获取 多Agent+LGBM双模型：从0到1搭建AI炒股系统实战
简介一份面向Python爱好者与量化入门者的轻量级AI炒股系统完整工程核心解决本地轻量运行与二次开发需求。系统采用多Agent架构将选股、风控、择时、复盘解耦同时落地分类与回归双模型既能判断涨跌方向又可预测涨幅区间且预测严格遵循无未来函数原则仅使用当日及之前数据。自带Streamlit交互界面无需编写前端即可完成单票深度择时与周度复盘报告生成普通笔记本即可运行。压缩包共11个文件以py源代码为主辅以两个模型pkl权重文件、CSV交易数据、以及编译生成的pyc缓存整体大小598KB结构精简、便于按模块学习与改造。目前已有440人学习参考适合想快速搭建可解释、可扩展AI选股框架的读者入手可基于现有模型与界面逻辑直接进行策略回测、参数调优或数据源替换。1. 从0到1搭AI炒股系统为什么是“多AgentLGBM”而不是堆深度学习模型很多人一听到AI炒股系统第一反应是上深度学习或者强化学习模型。但摸过真实行情数据之后我的结论是日线级别的预测梯度提升树的可解释性和鲁棒性对表格特征更好用。标题里的多Agent架构本质上是把数据、特征、信号、复盘四条流水线拆开让每一步都可独立验证LGBM双模型则分别负责日频择时和周度复盘用两个视角去确认同一个交易决策。这套方案适合有Python基础、想把交易想法转成可回测、可运行的源代码系统的开发者。先把流程闭环跑通再慢慢优化策略比一开始就追求模型复杂度更实际。2. 先把架构立住多Agent怎么分工LGBM双模型又怎么配合2.1 多Agent不是聊天机器人四条流水线各管一段提到多Agent很多人会联想到大语言模型那套智能体编排但在这个场景里Agent更接近“职责独立的服务模块”。我一开始就把系统拆成四个AgentDataAgent负责行情获取与落库FeatureAgent负责因子计算与特征缓存SignalAgent负责加载模型并生成择时信号ReviewAgent负责周度复盘报告。它们之间不直接互相调用函数而是通过中间状态文件或数据库表传递数据。这样做的原因很直接任何一个环节崩了可以单独重跑不把整条链路带崩任何一个环节想换实现比如把数据源从tushare换成akshare只需要改一个模块。下面是最小可运行的多Agent骨架代码# agents.py — 多Agent最小骨架 # 设计原则每个Agent只暴露run()入口输入输出都是文件路径便于单独重试 from pathlib import Path import pandas as pd class DataAgent: 数据Agent拉取行情 - 简单清洗 - 落盘parquet def __init__(self, data_dir: str ./data): self.data_dir Path(data_dir) self.data_dir.mkdir(exist_okTrue) def run(self, symbol: str, start: str, end: str) - str: # 生产环境这里用akshare/tushare拉真实日线 # 下面生成一份确定性示例数据先把流程跑通 dates pd.date_range(start, end, freqB) df pd.DataFrame({ close: 10 (dates.dayofyear % 7) * 0.5, volume: 1_000_000 (dates.dayofyear % 5) * 100_000, }, indexdates) df[open] df[close] * 0.99 df[high] df[close] * 1.02 df[low] df[close] * 0.98 path self.data_dir / f{symbol}.parquet df.to_parquet(path) return str(path) class FeatureAgent: 特征Agent读行情算技术指标输出特征文件 def run(self, path: str) - str: df pd.read_parquet(path) ema12 df[close].ewm(span12).mean() ema26 df[close].ewm(span26).mean() df[macd_dif] ema12 - ema26 df[macd_dea] df[macd_dif].ewm(span9).mean() df[macd_hist] (df[macd_dif] - df[macd_dea]) * 2 # 简化RSI正式实现在第3章 df[rsi14] df[close].pct_change().rolling(14).mean() df[vol_ma20] df[volume].rolling(20).mean() df[vol_ratio] df[volume] / df[vol_ma20] feature_path path.replace(.parquet, _feat.parquet) df.to_parquet(feature_path) return feature_path这段代码的逻辑说明DataAgent的run方法入参是股票代码和起止日期出参是parquet文件路径FeatureAgent接住这个路径算出macd、rsi、量比等基础特征再写一份新文件。中间没有共享内存变量也没有全局状态所以任何一个Agent都能独立重放历史数据。参数说明DataAgent示例里没有体现freq参数真实工程里应该再收一个freq参数daily/min5用来切换日线和分钟线FeatureAgent里的ewm(span12)是MACD快线参数默认12日是美国标准A股有人改成10日这类参数应该后移到config.yaml而不是写死在代码里。2.2 LGBM双模型不是“两个一样的模型”择时与复盘的技术差异标题里的LGBM双模型我拆成两个TimingModel日频择时模型和ReviewModel周频复盘模型。很多人的直觉是“训练两个一样的模型再平均”但这两者的样本构造和标签定义完全不同。TimingModel的输入是每日特征向量标签是未来3日的收益是否超过阈值作用是在每个交易日收盘后给出次日仓位建议偏“战术”。ReviewModel的输入是过去5个交易日的聚合特征——周涨幅、周均量比、周内信号分布、胜率统计——标签是未来一周能否跑赢沪深300指数作用是判断该继续持有还是清仓偏“战略”。为什么不能共用一个模型日频模型面对的是高噪声、低信噪比的样本正负样本比例常常接近1:9周频模型拿到的样本更规整但样本量大幅缩小需要不同的采样策略和正则化强度。因此我在代码里把两个模型的训练入口分开train_timing.py和train_review.py配置文件里也使用两个独立的model节。2.3 最小工程骨架目录结构、配置文件和主流程我倾向于把整个项目按功能拆成下面这个结构任何人扫一眼目录就知道每个文件是干什么的ai_stock/ ├── agents/ │ ├── __init__.py │ ├── data_agent.py # 行情数据 │ ├── feature_agent.py # 特征计算 │ ├── signal_agent.py # 模型推理与信号生成 │ └── review_agent.py # 周度复盘 ├── config/ │ └── config.yaml # 参数集中管理 ├── data/ # parquet数据文件 ├── models/ # 训练好的模型产物 ├── reports/ # 周度复盘报告输出 ├── train_timing.py # 择时模型训练 ├── train_review.py # 周度复盘模型训练 └── run_pipeline.py # 每日主流程run_pipeline.py的任务非常简单就是把四个Agent串起来# run_pipeline.py — 每日主流程 # 参数统一从config读取不写死在业务代码里 import yaml from agents.data_agent import DataAgent from agents.feature_agent import FeatureAgent def main(): with open(config/config.yaml, r, encodingutf-8) as f: cfg yaml.safe_load(f) data_agent DataAgent(data_dircfg[data][dir]) feat_agent FeatureAgent() for symbol in cfg[universe][symbols]: raw_path data_agent.run( symbolsymbol, startcfg[data][start_date], endcfg[data][end_date] ) feat_path feat_agent.run(raw_path) # 后续交给SignalAgent和ReviewAgent if __name__ __main__: main()逻辑说明每个Agent都是同一个模式——构造函数负责依赖注入run方法负责执行。config.yaml控制所有易变参数股票池、训练日期、特征开关、模型路径等。参数说明cfg[universe][symbols]在真实环境里可能是一份沪深300成分股列表几百个标的跑下来DataAgent的单线程拉取会非常慢我通常在这里直接换用ThreadPoolExecutor并发拉取但要控制并发数在10以内避免被数据源限流。3. 数据流水线与特征工程行情原始数据怎么变成模型能吃的特征3.1 行情数据落库日线为主、分钟线做验证复权因子不能忘第一次写AI炒股系统的人很容易直接拿复权后的日线数据开练。但真正落地时我发现一个关键问题前复权数据会随着后续除权事件发生而整体变化如果你在历史回测和实盘之间共用同一份前复权数据模型会“看到”未来才出现的价格修正。所以我通常用后复权方式做训练交易信号生成后用不复权价做修正。下面是一段真实使用的取数逻辑# data_agent.py 里的真实取数片段 # 目标拿到后复权日线并落库保留原价列用于实盘修正 import akshare as ak import pandas as pd def fetch_daily_post_adjusted(symbol: str, start: str, end: str) - pd.DataFrame: # stock_zh_a_hist 返回未复权数据需要自己算复权因子 raw ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart, end_dateend, adjust ) raw.columns [date, open, close, high, low, volume, amount, amplitude, pct_change, change, turnover] # 基于收盘价序列构造累计复权因子简化版 # 更严谨的做法是从数据源拿hfq_factor字段 raw[adj_factor] (raw[close] / raw[close].iloc[0]).cummax() raw[close_hfq] raw[close] * raw[adj_factor] raw[open_hfq] raw[open] * raw[adj_factor] raw[high_hfq] raw[high] * raw[adj_factor] raw[low_hfq] raw[low] * raw[adj_factor] return raw逻辑说明这里的核心是adj_factor字段用cummax模拟累计复权因子真实生产环境应从数据源直接读取复权因子字段。参数说明perioddaily表示日线如果你要做盘中择时可以考虑period60之类的分钟线但分钟线数据量与日线不在一个量级存储和计算成本都会暴涨。我的建议是先用日线跑通闭环再根据回测结果决定是否需要分钟线。3.2 特征工程MACD、RSI、量比、布林带特征设计不需要太多花哨因子。先保证基础的技术指标正确MACD、RSI、布林带、量比、动量。这些指标是公开的但实现细节千差万别稍不留意就和其他人的结果对不上。# feature_agent.py 的核心特征计算 # 注意这里所有计算都基于后复权价格避免除权跳变污染指标 def add_indicators(df: pd.DataFrame) - pd.DataFrame: close df[close_hfq] high df[high_hfq] low df[low_hfq] volume df[volume] # ---- MACD ---- ema12 close.ewm(span12, adjustFalse).mean() ema26 close.ewm(span26, adjustFalse).mean() df[macd_dif] ema12 - ema26 df[macd_dea] df[macd_dif].ewm(span9, adjustFalse).mean() df[macd_hist] 2 * (df[macd_dif] - df[macd_dea]) # ---- RSI(14) ---- delta close.diff() gain delta.clip(lower0).rolling(14).mean() loss (-delta.clip(upper0)).rolling(14).mean() rs gain / (loss 1e-9) df[rsi14] 100 - 100 / (1 rs) # ---- 布林带(20, 2) ---- mid close.rolling(20).mean() std close.rolling(20).std() df[boll_upper] mid 2 * std df[boll_lower] mid - 2 * std df[boll_pos] (close - df[boll_lower]) / (df[boll_upper] - df[boll_lower] 1e-9) # ---- 量比当日成交量 / 过去5日均量---- df[vol_ratio5] volume / (volume.rolling(5).mean() 1e-9) # ---- 动量20日收益---- df[ret_20d] close.pct_change(20) return df逻辑说明MACD用adjustFalse的ewm和主流行情软件一致RSI用标准的SMA平滑版本而不是Wilder指数平滑两者数值有差异但这不影响模型训练只要全流程统一即可。参数说明boll的20日和2倍标准差是默认参数可以用config里的feature_params节覆盖。计算完成后我习惯把所有NaN行直接drop模型训练阶段宁可少样本也不要让缺失值污染分裂点。MACD双底这类形态特征后续可以做成窗口扫描特征基础因子先扛住再谈形态。3.3 标签构造用未来3日收益分档而不是简单的涨跌标签构造是整个系统里最影响上限的环节。不要把标签设成“明天涨/跌”这种二分类噪声太大。我的做法是构造未来3日的累计收益再按阈值分档。# label_engineering.py — 构造分类标签 # 核心思路用未来3日收益的分位数作为分档边界 def make_labels(df: pd.DataFrame, horizon: int 3) - pd.DataFrame: # 未来horizon日收益shift是核心保证只用历史信息 df[future_ret] df[close_hfq].shift(-horizon) / df[close_hfq] - 1.0 # 按整段区间的分位数分三档0下跌1震荡2上涨 low_q df[future_ret].quantile(0.3) high_q df[future_ret].quantile(0.7) df[label] 0 df.loc[df[future_ret] high_q, label] 2 df.loc[(df[future_ret] low_q) (df[future_ret] high_q), label] 1 # 最后几行未来收益为空直接删除避免NaN标签进入训练 df df.dropna(subset[future_ret, label]) return df逻辑说明shift(-horizon)是标准做法表示把未来第3天的收盘价移动到今天这一行和今天的收盘价比对得到未来收益。参数说明quantile(0.3)和quantile(0.7)决定了三个类别的样本比例如果你用的股票池是沪深300这两个分位点得到的样本分布与用中证500完全不同需要重新计算。这类参数我一样放进config里而不是每跑一次就改代码。3.4 数据质量检查停牌、缺失值、极端值的处理数据质量这块吃过的亏最多。停牌日没有成交数据如果你直接用ffill向前填充模型会以为这些天有交易产生无法交易的样本直接导致训练集泄漏。我通常的做法是# data_quality.py — 数据质量检查 def check_quality(df: pd.DataFrame) - dict: report {} # 1. 缺失比例 report[missing_ratio] df.isna().mean().to_dict() # 2. 零成交检查成交量为0的日期列表 zero_vol_days df[df[volume] 0].index.tolist() report[zero_volume_days] len(zero_vol_days) # 3. 极端值检查单日涨跌超过20%的记录 extreme df[df[pct_change].abs() 0.2] report[extreme_days] len(extreme) # 4. 连续停牌段检查 df[is_trading] df[volume] 0 df[group] (df[is_trading] ! df[is_trading].shift()).cumsum() pause_groups df[~df[is_trading]].groupby(group).size() report[longest_pause] int(pause_groups.max()) if len(pause_groups) 0 else 0 return report逻辑说明这个check_quality函数输出一份简单的质量报告让你在训练之前就知道这份数据能不能用。longest_pause字段特别重要如果一只股票连续停牌超过10天复牌后的跳空会直接影响模型对特征的判断——复牌第一天的pct_change往往异常大如果你没做截断处理模型会被这个极端值带偏。通常我的做法是把复牌后第一天的极端涨跌做winsorize比如限制在±10%或者干脆把复牌日样本从训练集中剔除。4. 训练LGBM双模型参数、验证和择时信号生成4.1 择时模型的训练样本权重、类别不平衡与早停前面我把标签设成了三分类下跌/震荡/上涨但震荡类样本在决策上没有可操作性。所以实际训练时我让模型只输出“上涨概率”这个连续值即把label2的样本视为正样本其余视为负样本。如果一定要多分类就用LightGBM的multi_logloss只是后续信号生成时要额外处理。# train_timing.py — 择时模型训练入口 import lightgbm as lgb import pandas as pd import numpy as np from sklearn.model_selection import TimeSeriesSplit def train_timing_model(feature_path: str, config: dict) - lgb.Booster: df pd.read_parquet(feature_path) # 前面的标签构造已经存在df[label]里0/1/2 # 这里把label变成二值0不涨原0和11上涨原2 df[target] (df[label] 2).astype(int) feature_cols config[timing_model][features] X df[feature_cols] y df[target] # 时序切分不用随机K折避免未来信息泄漏 tscv TimeSeriesSplit(n_splits5) params { objective: binary, metric: binary_logloss, learning_rate: 0.05, num_leaves: 31, min_data_in_leaf: 50, feature_fraction: 0.8, bagging_fraction: 0.8, bagging_freq: 1, verbose: -1, seed: 42, } best_iters [] for train_idx, val_idx in tscv.split(X): dtrain lgb.Dataset(X.iloc[train_idx], y.iloc[train_idx]) dvalid lgb.Dataset(X.iloc[val_idx], y.iloc[val_idx]) booster lgb.train( params, dtrain, num_boost_round2000, valid_sets[dvalid], callbacks[lgb.early_stopping(100), lgb.log_evaluation(200)], ) best_iters.append(booster.best_iteration) # 用全部数据以最佳迭代次数的中位数重新训练最终模型 best_iter int(np.median(best_iters)) final_dtrain lgb.Dataset(X, y) final_model lgb.train(params, final_dtrain, num_boost_roundbest_iter) final_model.save_model(models/timing_model.txt) return final_model逻辑说明TimeSeriesSplit是关键——它按时间顺序切训练集和验证集保证验证集永远在训练集之后避免“用未来预测过去”的伪高准确率。early_stopping的100轮是patience参数意思是验证集指标连续100轮不改善就提前终止。参数说明min_data_in_leaf50是防止过拟合的重要旋钮数据量小就调大到100feature_fraction和bagging_fraction同时启用相当于对特征和样本都做了随机采样是LightGBM里最常见的防过拟合组合。这里有一个常见的翻车点直接对全量数据集用train_test_split随机切分一旦测试集日期早于训练集回测结果就会虚假地好。4.2 复盘模型的训练周度聚合特征周度复盘模型不能拿每日特征直接怼进去因为它的任务和日频模型完全不同。把一周的特征做聚合累计收益、平均量比、信号分布、实际盈亏。这些聚合特征让模型能回答“这周表现与历史哪些状态最像”。# train_review.py — 周度复盘模型训练入口 def build_weekly_features(daily_df: pd.DataFrame) - pd.DataFrame: # 按周聚合周一为每周起点 weekly daily_df.resample(W-MON).agg({ ret_20d: last, rsi14: mean, vol_ratio5: mean, macd_hist: sum, close_hfq: lambda x: x.iloc[-1] / x.iloc[0] - 1, # 周收益 }).dropna() weekly.columns [momentum, avg_rsi, avg_vol_ratio, macd_power, weekly_ret] return weekly def train_review_model(weekly_df: pd.DataFrame, config: dict) - lgb.Booster: # 标签未来一周是否收益为正作为正样本 weekly_df[target] (weekly_df[weekly_ret].shift(-1) 0).astype(int) weekly_df weekly_df.dropna(subset[target]) # 周频样本量小用更保守的参数 params { objective: binary, metric: auc, learning_rate: 0.03, num_leaves: 15, min_data_in_leaf: 10, verbosity: -1, } X weekly_df[[momentum, avg_rsi, avg_vol_ratio, macd_power]] y weekly_df[target] model lgb.train(params, lgb.Dataset(X, y), num_boost_round200) model.save_model(models/review_model.txt) return model逻辑说明周频模型的样本量只有日频的五分之一所以参数要更保守num_leaves15、learning_rate0.03、num_boost_round200固定轮数不再做早停因为样本太少早停不稳定。参数说明resample(W-MON)表示把时间序列按每周一为起点重新分组这个窗口起点对周度复盘很重要——如果你从周四开始聚合得到的就是“跨周”的数据语义上很别扭。4.3 LightGBM必调参数清单我整理了一份实战中必调的参数表这些参数每一个都直接影响过回测收益曲线参数影响我的经验值调参方向num_leaves过拟合核心旋钮15-50数据量小就调小数据量大再加大min_data_in_leaf防止学到单样本规律50-100单样本预测值异常时增大learning_rate收敛速度与精度折中0.03-0.05小学习率配多轮数会更好feature_fraction特征采样比例0.7-0.9特征多、噪声大时调小bagging_fraction样本采样比例0.7-0.9配合bagging_freq1使用lambda_l2正则化1.0-5.0验证集和训练集差距大时增大min_split_gain最小分裂增益0.0-0.1分裂过碎时增大这些参数的调整顺序也有讲究。我一般先固定learning_rate0.05和num_boost_round2000然后调num_leaves和min_data_in_leaf确认验证集指标不再上涨后再调feature_fraction和bagging_fraction最后用早停确认最优轮数。反过来从正则化参数开始调往往很难定位问题。提示训练日志里如果看到“No further splits with positive gain”这类信息先别急着加特征多半是min_data_in_leaf或num_leaves太小让模型找不到合适的分裂点。4.4 信号合成双模型怎么联合出买卖决策模型训练完之后还需要一套信号合成逻辑。我的实现很简单每日收盘后TimingModel输出未来3日上涨概率p_t每周五收盘后ReviewModel输出下周继续向上的概率p_w。合成规则如下# signal_agent.py — 双模型信号合成 def generate_signal(timing_prob: float, review_prob: float) - str: # 规则先跑通再优化 if timing_prob 0.65 and review_prob 0.5: return buy elif timing_prob 0.35 and review_prob 0.5: return sell else: return hold逻辑说明0.65和0.35是两个概率阈值不是模型参数而是交易策略参数。为什么ReviewModel的阈值设在0.5而不是0.65因为周频样本更少、概率估计本身更保守不需要那么苛刻的阈值。参数说明预算足够的团队会用网格搜索去扫这两个阈值但在个人项目里建议先从固定阈值开始积累一段时间回测数据后再用验证集反馈调整。到这里信号合成和回测可以联动。简单回测# backtest.py — 最小回测逻辑只算日频收益 def simple_backtest(df_with_signal: pd.DataFrame, initial_cash: float 100_000): cash initial_cash position 0 # 持仓股数 for i in range(len(df_with_signal)): row df_with_signal.iloc[i] if row[signal] buy and position 0: position cash / row[close_hfq] cash 0 elif row[signal] sell and position 0: cash position * row[close_hfq] position 0 # 最终资产 现金 持仓市值 final cash position * df_with_signal[close_hfq].iloc[-1] return final / initial_cash - 1逻辑说明这份回测只做了最简单的“满仓/空仓”切换适合看模型能不能跑出方向感。真正的回测还应该加入手续费和滑点这部分放在第5章展开。5. 避坑与常见问题数据、模型、回测里最容易翻车的地方5.1 前视偏差复权因子用错回测曲线好得吓人现象回测曲线一路向上年化收益高得吓人实盘一跑就开始亏。 原因用了前复权数据训练而前复权价格序列会随着后续除权除息而整体变化模型相当于提前看到了“未来的复权因子”。前复权的本质是把历史价格按照当前价格水平做整体拉伸每一次除权都会改变全部历史价格训练和回测只要跨过除权日特征数值就包含了未来信息。 解决改用后复权数据训练信号生成时再映射回真实成交价。如果数据源只提供前复权那么每次除权事件发生后就重拉历史数据且保证训练窗口和回测窗口不能跨越同一个除权日太久。我自己的做法是数据文件里单独存一列adj_factor每次训练前用这列做一致性校验发现因子序列有跳变就报警。5.2 停牌数据泄漏模型学会了“停牌后必涨”现象回测中每次复牌后都触发买入信号而且这部分交易贡献了大部分利润。 原因停牌期间成交量是0pct_change也是0这些假样本被当成“横盘整理”输入模型模型学会了在低波动、低成交量的状态下重仓买入——这些样本在现实中根本没有可成交的价格复牌后的跳空却会计入收益。 解决训练前把非交易日停牌日全部剔除或者留一个is_trading列作为特征但训练时禁用它。更极端的情况是长时间停牌后的复牌首日直接剔除样本因为这类样本的量价数据在模型眼里完全失真。检查方法也很简单统计一下训练集里“volume0”的行数如果不为0说明数据清洗没做到位。5.3 滑点设置太理想回测盈利实盘打脸现象回测收益接近70%实盘换手一次就亏掉2%的价差。 原因回测按收盘价成交忽略了冲击成本和买卖价差。小盘低价股的流动性远不如沪深300成分股模型专挑这些票容易出现“信号出来但根本买卖不动”的情况尤其当信号集中在尾盘时盘口只有卖单没有买单。 解决在回测里加固定滑点比如万分之五和千分之一的交易手续费然后在信号过滤里限制一天只交易一单对于单日成交量小于某个阈值的信号直接丢弃。更严格的做法是模拟逐笔撮合但个人项目里先加固定滑点就够了等模型跑出真金白银的收益曲线再上复杂撮合。5.4 特征穿越用当日收盘数据预测当日涨跌现象模型在验证集上AUC高达0.93实盘预测却断崖下跌。 原因构造标签时没有使用shift(-horizon)或者在计算特征时用了当日收盘价又把当日涨跌当成未来涨跌——这种错误最常见的是把当日的pct_change同时作为特征和标签的原料。 解决代码里严格区分t时刻可用信息与t1时刻信息。一个简单的检查方法是取某一天的真实历史特征丢给模型推理输出概率再和你已知的第二天结果对比看模型是不是仅仅因为看了当天收盘价才给的信号。这个问题我再补充一个细节检查特征时用特征的最后一行去推理——如果这一天的收益已经体现在特征里了比如ret_20d包含了当天就不应该再用它预测当天两者时间戳一旦对齐就泄漏了。6. 周度复盘模块落地自动报告、版本管理与人工复核6.1 自动生成复盘报告ReviewAgent在每周五收盘后把本周全部信号和结果汇总生成一份Markdown报告。下面是最精简的生成逻辑# review_agent.py — 周度报告核心片段 def generate_weekly_report(signals, results, week_tag): hit (results[pct_change] 0).sum() lines [ f# 周度复盘 {week_tag}, f信号数 {len(results)}上涨命中 {hit}命中率 {hit/len(results):.1%}, ] Path(freports/weekly_{week_tag}.md).write_text( \n.join(lines), encodingutf-8 ) return freports/weekly_{week_tag}.md这篇报告的核心不是看历史好坏而是观察模型置信度在接下来一周是否会回落。我每周五收盘后都会先跑一遍报告再决定下周要不要降低仓位。6.2 模型版本管理给每次训练留后悔药新模型上线前我会把训练参数、指标、数据区间写到一个json文件def save_model_with_meta(model, metrics, tag): model.save_model(fmodels/{tag}.txt) meta { metrics: metrics, config: load_current_config(), created_at: time.strftime(%Y-%m-%d), } with open(fmodels/{tag}.json, w, encodingutf-8) as f: json.dump(meta, f, ensure_asciiFalse, indent2)tag我习惯用训练结束日期比如timing_model_20250115。加载模型时先看json里的config和数据区间就能知道这个模型当时是在什么环境下训练的旧模型也能回滚。6.3 人工复核清单与我的习惯我的复盘清单固定五条置信度在0.5到0.65之间的信号不操作双模型分歧超过30%的个股只观察不加仓命中率连续两周下滑先查特征分布漂移消息面出大事件的个股跳过模型判断新模型先幽灵跑两周再上岗。这套方法里最实用的验证技巧就是幽灵跟踪让新模型只记录信号不下单两周后对比旧模型的命中率。我每次训练完都靠这一步决定要不要替换模型也靠它把参数调整的翻车风险压到最低。希望这条从0到1搭建AI炒股系统的落地路径能帮到你少走我当年走过的弯路。本文还有配套的精品资源点击获取