企业官网建设流程全解析

简介：一份面向Python爱好者与量化入门者的轻量级AI炒股系统完整工程，核心解决本地轻量运行与二次开发需求。系统采用多Agent架构，将选股、风控、择时、复盘解耦；同时落地分类与回归双模型，既能判断涨跌方向，又可预测涨幅区间，且预测严格遵循无未来函数原则，仅使用当日及之前数据。自带Streamlit交互界面，无需编写前端即可完成单票深度择时与周度复盘报告生成，普通笔记本即可运行。压缩包共11个文件，以py源代码为主，辅以两个模型pkl权重文件、CSV交易数据、以及编译生成的pyc缓存，整体大小598KB，结构精简、便于按模块学习与改造。目前已有440人学习参考，适合想快速搭建可解释、可扩展AI选股框架的读者入手，可基于现有模型与界面逻辑直接进行策略回测、参数调优或数据源替换。

1. 从0到1搭AI炒股系统：为什么是“多Agent+LGBM”，而不是堆深度学习模型

很多人一听到AI炒股系统，第一反应是上深度学习或者强化学习模型。但摸过真实行情数据之后，我的结论是：日线级别的预测，梯度提升树的可解释性和鲁棒性对表格特征更好用。标题里的多Agent架构，本质上是把数据、特征、信号、复盘四条流水线拆开，让每一步都可独立验证；LGBM双模型则分别负责日频择时和周度复盘，用两个视角去确认同一个交易决策。这套方案适合有Python基础、想把交易想法转成可回测、可运行的源代码系统的开发者。先把流程闭环跑通，再慢慢优化策略，比一开始就追求模型复杂度更实际。

2. 先把架构立住：多Agent怎么分工，LGBM双模型又怎么配合

2.1 多Agent不是聊天机器人：四条流水线各管一段

提到多Agent，很多人会联想到大语言模型那套智能体编排，但在这个场景里，Agent更接近“职责独立的服务模块”。我一开始就把系统拆成四个Agent：DataAgent负责行情获取与落库，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_ok=True) def run(self, symbol: str, start: str, end: str) -> str: # 生产环境这里用akshare/tushare拉真实日线 # 下面生成一份确定性示例数据，先把流程跑通 dates = pd.date_range(start, end, freq="B") df = pd.DataFrame({ "close": 10 + (dates.dayofyear % 7) * 0.5, "volume": 1_000_000 + (dates.dayofyear % 5) * 100_000, }, index=dates) 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(span=12).mean() ema26 = df["close"].ewm(span=26).mean() df["macd_dif"] = ema12 - ema26 df["macd_dea"] = df["macd_dif"].ewm(span=9).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(span=12)是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", encoding="utf-8") as f: cfg = yaml.safe_load(f) data_agent = DataAgent(data_dir=cfg["data"]["dir"]) feat_agent = FeatureAgent() for symbol in cfg["universe"]["symbols"]: raw_path = data_agent.run( symbol=symbol, start=cfg["data"]["start_date"], end=cfg["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( symbol=symbol, period="daily", start_date=start, end_date=end, 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模拟累计复权因子，真实生产环境应从数据源直接读取复权因子字段。参数说明：period="daily"表示日线，如果你要做盘中择时，可以考虑period="60"之类的分钟线，但分钟线数据量与日线不在一个量级，存储和计算成本都会暴涨。我的建议是先用日线跑通闭环，再根据回测结果决定是否需要分钟线。

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(span=12, adjust=False).mean() ema26 = close.ewm(span=26, adjust=False).mean() df["macd_dif"] = ema12 - ema26 df["macd_dea"] = df["macd_dif"].ewm(span=9, adjust=False).mean() df["macd_hist"] = 2 * (df["macd_dif"] - df["macd_dea"]) # ---- RSI(14) ---- delta = close.diff() gain = delta.clip(lower=0).rolling(14).mean() loss = (-delta.clip(upper=0)).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用adjust=False的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 择时模型的训练：样本权重、类别不平衡与早停

前面我把标签设成了三分类（下跌/震荡/上涨），但震荡类样本在决策上没有可操作性。所以实际训练时我让模型只输出“上涨概率”这个连续值，即把label=2的样本视为正样本，其余视为负样本。如果一定要多分类，就用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和1），1=上涨（原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_splits=5) 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_round=2000, 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_round=best_iter) final_model.save_model("models/timing_model.txt") return final_model

逻辑说明：TimeSeriesSplit是关键——它按时间顺序切训练集和验证集，保证验证集永远在训练集之后，避免“用未来预测过去”的伪高准确率。early_stopping的100轮是patience参数，意思是验证集指标连续100轮不改善就提前终止。参数说明：min_data_in_leaf=50是防止过拟合的重要旋钮，数据量小就调大到100；feature_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_round=200) model.save_model("models/review_model.txt") return model

逻辑说明：周频模型的样本量只有日频的五分之一，所以参数要更保守：num_leaves=15、learning_rate=0.03、num_boost_round=200固定轮数，不再做早停，因为样本太少早停不稳定。参数说明：resample("W-MON")表示把时间序列按每周一为起点重新分组，这个窗口起点对周度复盘很重要——如果你从周四开始聚合，得到的就是“跨周”的数据，语义上很别扭。

4.3 LightGBM必调参数清单

我整理了一份实战中必调的参数表，这些参数每一个都直接影响过回测收益曲线：

这些参数的调整顺序也有讲究。我一般先固定learning_rate=0.05和num_boost_round=2000，然后调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 停牌数据泄漏：模型学会了“停牌后必涨”

现象：回测中每次复牌后都触发买入信号，而且这部分交易贡献了大部分利润。 原因：停牌期间成交量是0，pct_change也是0，这些假样本被当成“横盘整理”输入模型，模型学会了在低波动、低成交量的状态下重仓买入——这些样本在现实中根本没有可成交的价格，复牌后的跳空却会计入收益。 解决：训练前把非交易日（停牌日）全部剔除，或者留一个is_trading列作为特征但训练时禁用它。更极端的情况是长时间停牌后的复牌首日直接剔除样本，因为这类样本的量价数据在模型眼里完全失真。检查方法也很简单：统计一下训练集里“volume==0”的行数，如果不为0，说明数据清洗没做到位。

5.3 滑点设置太理想：回测盈利，实盘打脸

现象：回测收益接近70%，实盘换手一次就亏掉2%的价差。 原因：回测按收盘价成交，忽略了冲击成本和买卖价差。小盘低价股的流动性远不如沪深300成分股，模型专挑这些票容易出现“信号出来但根本买卖不动”的情况，尤其当信号集中在尾盘时，盘口只有卖单没有买单。 解决：在回测里加固定滑点（比如万分之五）和千分之一的交易手续费，然后在信号过滤里限制一天只交易一单，对于单日成交量小于某个阈值的信号直接丢弃。更严格的做法是模拟逐笔撮合，但个人项目里先加固定滑点就够了，等模型跑出真金白银的收益曲线再上复杂撮合。

5.4 特征穿越：用当日收盘数据预测当日涨跌

现象：模型在验证集上AUC高达0.93，实盘预测却断崖下跌。 原因：构造标签时没有使用shift(-horizon)，或者在计算特征时用了当日收盘价，又把当日涨跌当成未来涨跌——这种错误最常见的是把当日的pct_change同时作为特征和标签的原料。 解决：代码里严格区分t时刻可用信息与t+1时刻信息。一个简单的检查方法是：取某一天的真实历史特征丢给模型推理，输出概率，再和你已知的第二天结果对比，看模型是不是仅仅因为看了当天收盘价才给的信号。这个问题我再补充一个细节：检查特征时用特征的最后一行去推理——如果这一天的收益已经体现在特征里了（比如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(f"reports/weekly_{week_tag}.md").write_text( "\n".join(lines), encoding="utf-8" ) return f"reports/weekly_{week_tag}.md"

这篇报告的核心不是看历史好坏，而是观察模型置信度在接下来一周是否会回落。我每周五收盘后都会先跑一遍报告，再决定下周要不要降低仓位。

6.2 模型版本管理：给每次训练留后悔药

新模型上线前，我会把训练参数、指标、数据区间写到一个json文件：

def save_model_with_meta(model, metrics, tag): model.save_model(f"models/{tag}.txt") meta = { "metrics": metrics, "config": load_current_config(), "created_at": time.strftime("%Y-%m-%d"), } with open(f"models/{tag}.json", "w", encoding="utf-8") as f: json.dump(meta, f, ensure_ascii=False, indent=2)

tag我习惯用训练结束日期，比如timing_model_20250115。加载模型时先看json里的config和数据区间，就能知道这个模型当时是在什么环境下训练的，旧模型也能回滚。

6.3 人工复核清单与我的习惯

我的复盘清单固定五条：置信度在0.5到0.65之间的信号不操作；双模型分歧超过30%的个股只观察不加仓；命中率连续两周下滑先查特征分布漂移；消息面出大事件的个股跳过模型判断；新模型先幽灵跑两周再上岗。

这套方法里最实用的验证技巧就是幽灵跟踪：让新模型只记录信号不下单，两周后对比旧模型的命中率。我每次训练完都靠这一步决定要不要替换模型，也靠它把参数调整的翻车风险压到最低。希望这条从0到1搭建AI炒股系统的落地路径能帮到你，少走我当年走过的弯路。

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