简介这是一套面向量化入门者与Python爱好者的轻量级AI炒股系统实战代码聚焦解决个人投资者在择时决策与复盘分析中的核心痛点无未来函数依赖、兼顾涨跌判断与涨幅预测、模块化可扩展。资源共11个文件含4个核心Python脚本如app.py、main.py、train.py、2个预训练模型.pkl格式分别对应600025.SH的分类与回归任务、2个示例数据data.csv、trades.csv及依赖说明等整体仅598KB普通笔记本即可运行调试。已有423人学习下载适合快速上手多Agent架构设计思想——选股、风控、择时、复盘四大功能解耦清晰配套Streamlit交互界面免前端开发支持单票深度择时推演并自动生成结构化周度复盘报告所有逻辑均基于当日及历史数据具备实盘参考价值。1. 项目概述一个实战派的AI量化交易系统最近几年AI炒股或者说量化交易已经从机构专属的神坛逐渐走向了个人开发者和交易爱好者的视野。但说实话很多开源项目要么是“玩具级”的只能跑个回测离实盘十万八千里要么就是“黑盒级”的代码复杂得像天书你根本不知道它为什么买、为什么卖。我自己在金融科技领域摸爬滚打了十几年从传统策略写到机器学习踩过的坑不计其数。所以我一直想动手做一个东西它得是从零开始、结构清晰、每一步都有据可循的同时还得是多模型、多任务协同的不能只靠一个模型打天下。于是就有了这个项目一个基于多Agent架构和LGBM双模型的AI炒股系统。它的核心目标很明确第一实现有效的择时分析告诉你什么时候该进场、什么时候该离场第二完成深度的周度复盘不仅仅是看盈亏更要分析策略为什么有效或失效为下一周的调整提供依据。这不仅仅是把数据喂给模型然后输出买卖信号那么简单它更像是一个由多个“智能体”分工协作的虚拟交易团队有的负责选股有的负责择时有的负责风控还有的负责事后复盘总结。这个系统适合谁呢首先它适合有一定Python编程基础对机器学习尤其是树模型和金融市场有基本了解的朋友你想知道一个可用的量化系统到底是怎么搭建起来的。其次它也适合那些已经写过一些简单策略但苦于策略不稳定、无法系统化评估和迭代的交易者。通过这个项目你能获得的不只是一套源代码更是一套构建稳健AI交易系统的工程化思维和方法论。我会把整个从数据获取、处理、特征工程、模型训练、多Agent调度到回测、模拟盘乃至实盘衔接的完整链条掰开揉碎了讲清楚。我们不用那些玄而又玄的“黑科技”就用扎实的工程实践和可解释性强的模型搭建一个你自己能完全掌控、能持续迭代的“AI交易员”。2. 核心架构设计为什么是多Agent与LGBM双模型在动手写代码之前我们必须把架构想清楚。为什么选择“多Agent架构”和“LGBM双模型”这个组合这背后是对于量化交易复杂性的深度考量而不是简单的技术堆砌。2.1 多Agent架构模拟一个专业的交易团队传统的量化策略往往是一个“单体巨无霸”函数里面混杂了数据获取、清洗、特征计算、信号生成、风险控制等所有逻辑。这种代码的维护和迭代是一场噩梦一个地方的修改可能会引发意想不到的连锁反应。更重要的是它不符合真实的投资决策流程。在真实的投资机构中决策是分工协作的研究员提供股票池和基本面观点量化分析师负责建模和信号计算交易员负责执行和盘口观察风控官时刻盯着仓位和风险指标。多Agent架构正是为了模拟这一专业化分工。在我们的系统中设计了以下几个核心AgentDataFetcher Agent数据获取代理它的职责单一而明确就是定时、稳定地从各种数据源如财经API、本地数据库获取原始的行情数据、基本面数据和宏观数据。它需要处理网络异常、数据格式不统一、数据缺失等问题并为下游Agent提供干净、统一的数据接口。它的存在将易变的数据获取逻辑与核心的策略逻辑解耦。FeatureEngineer Agent特征工程代理这是策略的“厨房”。它接收原始数据加工出模型所需的“食材”——即特征。这包括计算各种技术指标如MACD, RSI, 布林带、价量特征、波动率特征以及进行必要的标准化、归一化处理。这个Agent的优化空间极大特征的好坏直接决定了模型性能的天花板。LGBM_Predictor Agent预测代理这是系统的“大脑”之一。它装载着我们训练好的LightGBM模型。它的任务是根据FeatureEngineer提供的特征预测未来一段时间例如未来5天的股价涨跌概率或收益率。我们采用双模型设计一个模型专注于短期波动捕捉例如未来1-3天的涨跌另一个模型专注于中期趋势判断例如未来5-10天的方向。双模型可以相互验证降低单一模型误判的风险。SignalFusion Agent信号融合代理接收到来自双模型的预测结果后这个Agent负责做最终的决策。它不是简单地对预测结果取平均而是根据市场状态如波动率、趋势强度、模型近期表现如预测置信度进行加权融合。例如在震荡市中可能更信任短期模型在趋势市中则加大中期模型的权重。它输出的是统一的、带有置信度的交易信号如强烈买入、买入、观望、卖出。PortfolioManager Agent组合管理代理信号有了但怎么买买多少这就是它的工作。它根据信号强度、账户总资金、个股风险如波动率、以及相关性计算具体的仓位分配。它实现了简单的风险平价或均值-方差优化思想确保投资组合不会过度集中在某一只股票或某一个行业上。Backtest/Review Agent回测与复盘代理这是系统学习和进化的关键。它不仅在历史数据上模拟交易更重要的是在每周结束后进行周度复盘。它会分析本周哪些交易赚了钱、哪些亏了钱原因是什么是模型失效还是遇到了极端行情它会生成详细的复盘报告包括收益曲线、最大回撤、夏普比率、交易胜率等并且会标注出需要重点关注的问题为下一周的特征调整或模型迭代提供方向。注意Agent之间通过清晰定义的接口如Python类方法、消息队列进行通信每个Agent内部状态独立。这种设计让系统具备了极强的可扩展性。未来如果你想加入一个基于Transformer的Agent或者一个专门处理新闻情感的Agent只需要让它遵循同样的通信规范就能轻松接入系统而不需要重写其他部分。2.2 LGBM双模型稳健与可解释性的平衡为什么选择LightGBMLGBM而不是更“时髦”的深度学习模型如LSTM、Transformer效率与性能在表格数据金融数据本质上是时间序列表格数据上梯度提升树模型如XGBoost, LightGBM, CatBoost长期以来都是性能霸主。它们训练速度快对特征缺失不敏感且通常不需要像深度学习那样复杂的调参和大量的数据。可解释性这是关键。炒股最怕“黑盒”。LGBM提供了特征重要性排序我们可以清楚地知道是哪些指标例如“成交量放大率”、“过去20日波动率”对模型的决策影响最大。这在进行周度复盘时至关重要如果本周策略亏损我们可以通过分析特征重要性判断是市场逻辑变了原来重要的特征失效了还是出现了新的驱动因素。双模型设计的考量模型A短期模型输入是近期高频特征如分钟级价量变化、盘口失衡目标是预测未来1-3日的涨跌。它更敏感试图捕捉市场短期的情绪和资金流动。模型B中期模型输入是中长期特征如周度均线排列、行业资金流向、宏观指标变化目标是预测未来5-10日的趋势方向。它更稳健试图把握基本面的动量。分工与协作短期模型可能发出许多交易信号但通过中期模型的“趋势过滤器”可以过滤掉那些逆中期趋势的“噪音信号”从而提高整体胜率。例如短期模型看涨但中期模型显示下行趋势那么SignalFusion Agent可能会降低该信号的权重甚至将其忽略。3. 从零搭建环境准备与核心模块实现理论讲完了我们开始动手。假设你已经有Python基础并且安装了Anaconda或Miniconda。我们从最干净的环境开始。3.1 项目环境与依赖管理首先创建一个独立的虚拟环境是专业开发的第一步它能避免包版本冲突。# 创建名为 ai_trader 的虚拟环境 conda create -n ai_trader python3.9 -y conda activate ai_trader接着初始化项目目录并使用requirements.txt严格管理依赖。这是项目可复现性的生命线。my_ai_trading_system/ ├── agents/ # 存放所有Agent的类定义 ├── data/ # 存放原始和加工后的数据 ├── models/ # 存放训练好的LGBM模型文件 ├── utils/ # 存放工具函数数据获取、指标计算等 ├── config.py # 全局配置文件参数、路径 ├── main.py # 系统主入口 ├── train.py # 模型训练脚本 ├── backtest.py # 回测与复盘脚本 └── requirements.txt # 项目依赖requirements.txt文件内容如下我注释了每个核心库的用途# 数据处理与分析 pandas1.4.0 numpy1.22.0 # 机器学习模型 lightgbm3.3.0 # 核心模型库 scikit-learn1.0.0 # 用于数据划分、评估指标 # 数据获取 (这里以Tushare为例你可以替换为其他数据源) tushare1.2.89 # 消息队列用于高级的Agent异步通信初期可用简单方式替代 # pika1.2.0 # 回测引擎我们也可以自己实现但使用成熟库更稳定 backtrader1.9.76.123 # 强大的回测框架 # 可视化 matplotlib3.5.0 seaborn0.11.0 # 配置文件管理 pyyaml6.0 # 进度条 tqdm4.64.0使用pip install -r requirements.txt一键安装所有依赖。3.2 核心Agent的代码实现以FeatureEngineer和LGBM_Predictor为例我们深入两个最核心的Agent看看代码具体如何组织。FeatureEngineer Agent它的核心是一个FeatureEngineer类包含各种特征计算方法。# agents/feature_engineer.py import pandas as pd import numpy as np from typing import Dict, Optional class FeatureEngineer: def __init__(self, config: Dict): self.config config # 可以在这里初始化一些中间状态比如移动平均线的缓存 def calculate_technical_indicators(self, df: pd.DataFrame) - pd.DataFrame: 计算一系列技术指标特征。 df 需包含 columns: [open, high, low, close, volume] # 避免修改原数据 df df.copy() # 1. 价格衍生特征 df[returns] df[close].pct_change() df[log_returns] np.log(df[close] / df[close].shift(1)) # 2. 移动平均线及相关特征 df[ma5] df[close].rolling(window5).mean() df[ma20] df[close].rolling(window20).mean() df[ma60] df[close].rolling(window60).mean() df[price_vs_ma20] df[close] / df[ma20] - 1 # 价格偏离20日均线的百分比 # 3. 布林带 df[rolling_std] df[close].rolling(window20).std() df[bb_upper] df[ma20] 2 * df[rolling_std] df[bb_lower] df[ma20] - 2 * df[rolling_std] df[bb_width] (df[bb_upper] - df[bb_lower]) / df[ma20] # 布林带宽度衡量波动率 df[bb_position] (df[close] - df[bb_lower]) / (df[bb_upper] - df[bb_lower]) # 价格在布林带中的位置 # 4. RSI (相对强弱指数) delta df[close].diff() gain (delta.where(delta 0, 0)).rolling(window14).mean() loss (-delta.where(delta 0, 0)).rolling(window14).mean() rs gain / loss df[rsi] 100 - (100 / (1 rs)) # 5. 成交量相关特征 df[volume_ma5] df[volume].rolling(window5).mean() df[volume_ratio] df[volume] / df[volume_ma5] # 成交量相对变化 # 6. 波动率特征 df[volatility_20d] df[returns].rolling(window20).std() * np.sqrt(252) # 年化波动率 # 删除因滚动计算产生的NaN行 df.dropna(inplaceTrue) return df def run(self, raw_data: pd.DataFrame) - pd.DataFrame: 主运行方法对外接口 featured_data self.calculate_technical_indicators(raw_data) # 这里可以链式调用其他特征计算方法如基本面特征、宏观特征等 return featured_data实操心得特征工程是永无止境的。上述只是最基础的示例。在实际项目中你需要不断尝试和挖掘有效的特征例如横截面特征个股指标在全市场或同行业中的分位数排名。动量与反转特征过去N日的累计收益率、特定形态识别如金叉死叉。资金流特征基于Level-2数据的主动买入/卖出金额。注意避免未来函数这是量化的大忌。任何特征的计算只能使用当前及之前的历史数据绝不能用到未来的数据。rolling().mean()这类操作是安全的因为它只用到窗口内的历史数据。LGBM_Predictor Agent这个Agent负责加载模型并进行预测。# agents/lgbm_predictor.py import joblib import lightgbm as lgb import pandas as pd import numpy as np from typing import Tuple class LGBMPredictor: def __init__(self, model_short_path: str, model_mid_path: str): 初始化加载短期和中期两个模型。 try: self.model_short joblib.load(model_short_path) # 或 lgb.Booster(model_file...) self.model_mid joblib.load(model_mid_path) print(f模型加载成功: {model_short_path}, {model_mid_path}) except FileNotFoundError as e: print(f模型文件未找到: {e}) # 可以在这里触发模型训练流程 self.model_short None self.model_mid None def predict(self, features_df: pd.DataFrame) - Tuple[np.ndarray, np.ndarray]: 使用双模型进行预测。 返回: (短期模型预测结果, 中期模型预测结果) 预测结果可以是概率值分类任务或具体数值回归任务。 if self.model_short is None or self.model_mid is None: raise ValueError(模型未加载无法进行预测。) # 确保特征列的顺序与模型训练时完全一致这是线上预测最常见的坑。 # 通常需要在训练时保存一个特征列名列表在这里进行对齐。 # 假设我们有一个 self.feature_columns 属性保存了列顺序 # features_df features_df[self.feature_columns] X features_df.values # 转换为numpy数组 pred_short self.model_short.predict(X) pred_mid self.model_mid.predict(X) return pred_short, pred_mid def get_feature_importance(self, model_type: str short) - pd.DataFrame: 获取模型特征重要性用于复盘分析 model self.model_short if model_type short else self.model_mid if isinstance(model, lgb.Booster): importance model.feature_importance(importance_typegain) feature_names model.feature_name() else: # 如果是sklearn API包装的模型 importance model.feature_importances_ feature_names model.feature_name_ # 假设训练时传入了feature_name return pd.DataFrame({ feature: feature_names, importance: importance }).sort_values(importance, ascendingFalse)4. 模型训练与特征工程实战有了Agent的骨架我们需要用数据来填充血肉。模型训练是离线的、但至关重要的环节。4.1 数据准备与标注我们以A股日线数据为例使用Tushare Pro获取数据。目标是预测未来N日的涨跌分类问题或收益率回归问题。# utils/data_fetcher.py (简化版) import tushare as ts import pandas as pd from datetime import datetime, timedelta class DataFetcher: def __init__(self, token): ts.set_token(token) self.pro ts.pro_api() def get_daily_data(self, ts_code, start_date, end_date): 获取日线行情数据 df self.pro.daily(ts_codets_code, start_datestart_date, end_dateend_date) df[trade_date] pd.to_datetime(df[trade_date]) df.set_index(trade_date, inplaceTrue) df.sort_index(inplaceTrue) # 重命名列以符合我们的特征计算函数 df.rename(columns{open:open, high:high, low:low, close:close, vol:volume}, inplaceTrue) return df[[open, high, low, close, volume]] # 假设我们获取了多只股票的数据并合并 # all_data pd.concat({code: df for code, df in stock_data_dict.items()}, names[ts_code, trade_date])接下来是关键步骤创建标签。我们以“未来5日收益率是否超过阈值”作为二分类标签。def create_label(df, future_days5, threshold_pct0.02): 创建分类标签。 df: 包含close列的DataFrame 未来future_days日的收益率 threshold_pct 则标记为1上涨否则为0。 # 计算未来N日的收益率 future_return df[close].shift(-future_days) / df[close] - 1 # 创建标签 label (future_return threshold_pct).astype(int) # 由于使用了未来数据最后future_days行的标签是NaN需要删除 df[label] label df.dropna(subset[label], inplaceTrue) return df4.2 训练流程与参数调优训练脚本train.py负责组织整个流程数据加载 - 特征工程 - 划分训练/验证集 - 训练LGBM模型 - 评估 - 保存。# train.py 核心部分 import pandas as pd from sklearn.model_selection import train_test_split, TimeSeriesSplit from sklearn.metrics import accuracy_score, f1_score, roc_auc_score import lightgbm as lgb import joblib from agents.feature_engineer import FeatureEngineer from utils.data_fetcher import DataFetcher, create_label def train_model(data_df, label_collabel, model_nameshort_term): 训练一个LGBM模型 # 1. 分离特征和标签 X data_df.drop(columns[label_col]) y data_df[label_col] # 2. 划分数据集 - 对于时间序列必须按时间划分不能随机打乱 split_idx int(len(X) * 0.8) # 80%训练20%测试 X_train, X_test X.iloc[:split_idx], X.iloc[split_idx:] y_train, y_test y.iloc[:split_idx], y.iloc[split_idx:] # 3. 定义模型参数 params { objective: binary, # 二分类 metric: auc, # 评估指标 boosting_type: gbdt, num_leaves: 31, learning_rate: 0.05, feature_fraction: 0.9, # 防止过拟合 bagging_fraction: 0.8, bagging_freq: 5, verbose: -1, seed: 42 } # 4. 创建数据集 lgb_train lgb.Dataset(X_train, y_train) lgb_eval lgb.Dataset(X_test, y_test, referencelgb_train) # 5. 训练 print(f开始训练 {model_name} 模型...) gbm lgb.train(params, lgb_train, num_boost_round1000, valid_sets[lgb_eval], callbacks[lgb.early_stopping(stopping_rounds50), lgb.log_evaluation(50)]) # 6. 评估 y_pred_prob gbm.predict(X_test, num_iterationgbm.best_iteration) y_pred (y_pred_prob 0.5).astype(int) auc roc_auc_score(y_test, y_pred_prob) accuracy accuracy_score(y_test, y_pred) print(f{model_name} 模型测试集 AUC: {auc:.4f}, Accuracy: {accuracy:.4f}) # 7. 保存模型和特征列非常重要 joblib.dump(gbm, fmodels/{model_name}_model.pkl) # 保存训练时的特征列顺序 feature_columns X_train.columns.tolist() joblib.dump(feature_columns, fmodels/{model_name}_features.pkl) print(f模型和特征列已保存至 models/ 目录) return gbm # 主函数 if __name__ __main__: # 获取数据、特征工程、创建标签... # df ... # 分别用不同参数创建短期和中期标签训练两个模型 # df_short create_label(df.copy(), future_days3, threshold_pct0.015) # df_mid create_label(df.copy(), future_days10, threshold_pct0.03) # model_short train_model(df_short, model_namemodel_short) # model_mid train_model(df_mid, model_namemodel_mid)注意事项时间序列泄露绝对不能用未来的信息预测过去。划分数据集必须按时间顺序。使用TimeSeriesSplit进行交叉验证更严谨。样本不均衡股市中大涨大跌的日子是少数。如果正样本标签为1远少于负样本需要在params中设置is_unbalance: True或使用scale_pos_weight参数。特征重要性分析训练后务必查看gbm.feature_importance()。如果某些特征重要性极低可以考虑剔除简化模型。过拟合通过num_leaves、min_data_in_leaf、feature_fraction、bagging_fraction等参数控制模型复杂度并依靠验证集早停。5. 系统整合与回测复盘当各个Agent和模型都准备好后我们需要一个“指挥官”来调度它们这就是main.py。同时一个严谨的回测与复盘流程是检验系统成败的唯一标准。5.1 主流程调度与信号生成main.py模拟了每个交易日的决策流程。# main.py (核心逻辑) import pandas as pd from datetime import datetime import schedule import time from agents.data_fetcher_agent import DataFetcherAgent from agents.feature_engineer_agent import FeatureEngineerAgent from agents.lgbm_predictor_agent import LGBMPredictorAgent from agents.signal_fusion_agent import SignalFusionAgent from agents.portfolio_manager_agent import PortfolioManagerAgent from config import Config def daily_trading_job(): 每日收盘后执行的任务 print(f\n 执行交易日任务 {datetime.now().date()} ) # 1. 初始化所有Agent config Config() data_agent DataFetcherAgent(config) feature_agent FeatureEngineerAgent(config) predictor_agent LGBMPredictorAgent(config) signal_agent SignalFusionAgent(config) portfolio_agent PortfolioManagerAgent(config) # 2. 数据获取 print(步骤1: 获取市场数据...) raw_data_dict data_agent.fetch_market_data() # 返回一个字典key为股票代码value为DataFrame all_signals [] for ts_code, raw_df in raw_data_dict.items(): # 3. 特征工程 print(f 处理 {ts_code}: 特征工程...) feature_df feature_agent.run(raw_df) # 4. 模型预测 (只取最新一行的特征进行预测) latest_features feature_df.iloc[[-1]] # 获取最新一天的特征 pred_short, pred_mid predictor_agent.predict(latest_features) # 5. 信号融合 signal_strength, action signal_agent.fuse(pred_short[0], pred_mid[0], ts_code, feature_df) all_signals.append({ ts_code: ts_code, date: latest_features.index[0], pred_short: pred_short[0], pred_mid: pred_mid[0], signal: signal_strength, action: action # BUY, SELL, HOLD }) # 6. 组合管理 print(步骤5: 生成投资组合建议...) portfolio portfolio_agent.generate_portfolio(all_signals) print(f今日建议组合:\n{portfolio.to_string()}) # 7. 记录日志 (为复盘做准备) log_today_decision(all_signals, portfolio) print( 交易日任务完成 \n) def log_today_decision(signals, portfolio): 将当日决策记录到CSV或数据库 log_df pd.DataFrame(signals) log_df[portfolio_weight] log_df[ts_code].map(portfolio.set_index(ts_code)[weight]) log_df.to_csv(flogs/decision_{datetime.now().date()}.csv, modea, headerFalse) if __name__ __main__: # 模拟定时任务实际部署可用cron或APScheduler print(AI交易系统启动...) # 立即运行一次 daily_trading_job() # 设定每天下午4点收盘后运行 schedule.every().day.at(16:00).do(daily_trading_job) while True: schedule.run_pending() time.sleep(60)5.2 回测引擎实现与绩效分析回测不是简单的“信号出现就买卖”它需要模拟真实的交易场景考虑手续费、滑点、涨停跌停无法买卖等限制。我们使用成熟的backtrader框架来构建严谨的回测。# backtest.py import backtrader as bt import pandas as pd from agents.signal_fusion_agent import SignalFusionAgent class AITraderStrategy(bt.Strategy): params ( (signal_agent, None), # 传入我们自定义的信号Agent (portfolio_agent, None), # 传入组合管理Agent ) def __init__(self): # 跟踪订单和持仓 self.orders {} self.signals {} def next(self): 在每个Bar如每天结束时调用 current_date self.datas[0].datetime.date(0) # 1. 获取所有股票池的最新数据生成特征这里简化实际应从feature_agent获取 # 2. 调用signal_agent和portfolio_agent生成当日信号和仓位 # 3. 根据目标仓位调整现有仓位 for data in self.datas: symbol data._name target_weight ... # 从portfolio_agent获取该股票的目标权重 current_value self.getposition(data).size * data.close[0] total_value self.broker.getvalue() target_value total_value * target_weight # 计算需要交易的数量 size_to_trade (target_value - current_value) / data.close[0] if abs(size_to_trade) 0: self.order_target_percent(data, targettarget_weight) def notify_order(self, order): 订单状态回调用于处理成交、拒绝等情况 if order.status in [order.Completed]: print(f{order.data._name} 订单成交, 价格: {order.executed.price:.2f}, 数量: {order.executed.size}) def run_backtest(data_feed_dict, signal_agent, portfolio_agent, initial_cash100000.0): 运行回测 cerebro bt.Cerebro() cerebro.broker.setcash(initial_cash) cerebro.broker.setcommission(commission0.0003) # 设置手续费万三 cerebro.addsizer(bt.sizers.PercentSizer, percents95) # 每次使用95%的可用资金留有余地 # 添加数据 for symbol, df in data_feed_dict.items(): data bt.feeds.PandasData(datanamedf, namesymbol) cerebro.adddata(data) # 添加策略 cerebro.addstrategy(AITraderStrategy, signal_agentsignal_agent, portfolio_agentportfolio_agent) print(初始资金: %.2f % cerebro.broker.getvalue()) cerebro.run() print(最终资金: %.2f % cerebro.broker.getvalue()) # 绘制图表 cerebro.plot(stylecandlestick)5.3 周度复盘超越盈亏的深度分析回测结束后生成资金曲线只是第一步。周度复盘才是系统进化的核心。每周你应该运行一个复盘脚本至少分析以下几点绩效归因收益来源本周收益主要来自哪几只股票是Alpha超额收益还是Beta市场波动亏损分析亏损的交易是否符合策略逻辑是模型预测错误还是遇到了黑天鹅事件如个股暴雷模型在亏损交易发生前的预测置信度如何代码实现计算夏普比率、最大回撤、卡玛比率、胜率、盈亏比等关键指标并与上周、上月对比。模型与信号诊断特征重要性变化对比本周和上周的特征重要性排名是否有显著变化如果有可能意味着市场主导逻辑发生了切换。信号有效性统计“强烈买入”信号后续的实际上涨概率“卖出”信号后续的实际下跌概率。信号是否依然有效双模型一致性分析短期模型和中期模型发出冲突信号的案例最终市场走向如何这有助于优化SignalFusion Agent的融合规则。风险检查持仓集中度是否单只股票或单个行业仓位过重策略容量当前策略管理的资金量是否接近其有效容量模拟加大资金量后滑点成本是否显著增加极端行情表现在市场大涨大跌如涨跌幅超过2%的日子里策略表现如何是跟上了还是反向生成复盘报告 将上述分析自动化生成一份结构化的Markdown或PDF报告。报告模板可以包括## 第XX周交易复盘报告 ### 一、整体绩效概览 - 周收益率X.XX% - 基准如沪深300收益率X.XX% - 超额收益率X.XX% - 本周最大回撤X.XX% - 交易次数XX次胜率XX% ### 二、关键交易分析 #### 最佳交易Top 3 1. 股票A买入理由模型信号持有X天收益率X% 2. ... #### 最差交易Bottom 3 1. 股票B买入理由亏损原因分析模型误判/市场系统性风险/个股事件 2. ... ### 三、模型健康度检查 - 特征重要性Top 5[特征1, 特征2...] - 与上周对比变化[特征X上升特征Y下降] - 信号准确率买入信号准确率XX%卖出信号准确率XX% ### 四、风险提示与下周计划 - 当前主要风险... - 下周调整计划考虑调整特征X的权重/尝试加入新的特征Y/微调模型融合参数至...6. 常见问题、避坑指南与进阶思考在实际搭建和运行过程中你会遇到无数个坑。这里记录一些最具代表性的问题和我的解决方案。6.1 数据与特征工程相关Q1: 数据源不稳定或API限制怎么办A1: 这是第一个拦路虎。务必实现数据本地化缓存。每次从API获取数据后立即保存到本地数据库如SQLite或Parquet文件。下次请求时先检查本地是否有数据只获取增量部分。同时为DataFetcher Agent设计重试机制和优雅降级策略例如主API失败后切换备用数据源。Q2: 如何避免“未来函数”A2: 这是量化策略失效的元凶之一。必须建立严格的代码审查清单任何特征计算只允许使用.shift()、.rolling()、.expanding()等面向过去窗口的函数。绝对禁止使用.shift(-N)未来数据或任何需要未来信息才能计算的指标例如用到当日最高最低价的指标必须在收盘后、下一日开盘前计算。在特征工程模块中对每个特征函数进行“未来函数”标注和单元测试。Q3: 特征太多导致过拟合怎么办A3: 遵循“由简入繁”的原则。初阶从少数几个经典且逻辑清晰的技术指标开始如MA, RSI, 布林带。中阶使用LGBM自带的特征重要性进行筛选剔除重要性接近零的特征。可以尝试使用SelectFromModel进行自动化特征选择。高阶引入递归特征消除RFE或使用PCA等降维方法处理高度相关的特征组。但要注意降维后的特征可解释性会变差。6.2 模型训练与预测相关Q4: 模型在训练集上表现很好但在回测或实盘上很差过拟合。A4: 这是机器学习在量化中最常见的挑战。数据划分必须使用时间序列交叉验证TimeSeriesSplit而不是随机划分。参数调优使用GridSearchCV或Optuna等工具进行超参数优化时验证集也必须按时间顺序划分。正则化加大LGBM的min_data_in_leaf、min_gain_to_split降低num_leaves增加feature_fraction和bagging_fraction。简化模型有时候一个参数更少、更简单的模型其泛化能力反而更强。Q5: 市场风格切换模型突然失效。A5: 没有永远有效的模型。我们的多Agent和周度复盘机制就是为了应对这一点。动态再训练设定一个规则例如当模型在最近一个月的滚动回测中夏普比率低于某个阈值时自动触发模型的重新训练。LGBM_Predictor Agent需要具备模型热更新的能力。集成学习除了双模型可以训练多个在不同时间区间或不同股票池上表现良好的模型让SignalFusion Agent根据当前市场波动率、趋势等宏观指标动态选择或加权集成这些模型。6.3 系统与实盘相关Q6: 回测结果完美但模拟盘/实盘效果不佳。A6: 回测与实盘的差距主要来自滑点与手续费回测中假设按收盘价成交实盘中大单交易会有冲击成本。在回测中必须设置更保守的滑点例如0.1%和手续费。涨停跌停回测中必须加入“涨停不能买入跌停不能卖出”的逻辑。数据延迟与异步实盘数据是流式的且不同数据源可能有微小延迟。确保你的系统在收到所有必要数据后才触发决策逻辑并考虑使用事件驱动架构替代简单的定时轮询。Q7: 如何将系统部署到生产环境A7: 从本地脚本到稳定运行的系统需要工程化升级。容器化使用Docker将整个系统及其依赖打包确保环境一致性。任务调度使用专业的调度系统如Apache Airflow或Celery替代简单的schedule库管理复杂的依赖任务如数据更新-特征计算-模型预测-下单。监控与告警为系统添加关键指标监控如数据更新是否成功、模型预测耗时、Agent心跳并设置告警如邮件、钉钉机器人。Backtest/Review Agent可以定期自动运行并将复盘报告发送给你。Q8: 这个系统能保证赚钱吗A8:绝对不能。这是一个需要你深刻理解的核心理念。本系统提供的是一套严谨、可解释、可迭代的量化研究框架和工具。它帮助你系统性地验证想法减少情绪化交易但它无法预测未来。市场的有效性意味着任何公开的、简单的规律都会迅速消失。你的核心竞争力将来自于1对市场独特的理解转化为独特的特征2更快的迭代速度通过自动化复盘和再训练3更严格的风险管理通过PortfolioManager Agent实现。把这个系统看作是你的“AI研究员”和“纪律执行者”而真正的战略决策和进化方向依然依赖于你作为“基金经理”的洞察力。从历史数据中发现的规律在未来可能会失效持续的学习、验证和调整才是你在市场中长期生存的根本。本文还有配套的精品资源点击获取 
长河君 · 2026/8/30 19:52:06 · 阅读 1.2k 从零构建多Agent量化交易系统：基于LGBM双模型与周度复盘实战
简介这是一套面向量化入门者与Python爱好者的轻量级AI炒股系统实战代码聚焦解决个人投资者在择时决策与复盘分析中的核心痛点无未来函数依赖、兼顾涨跌判断与涨幅预测、模块化可扩展。资源共11个文件含4个核心Python脚本如app.py、main.py、train.py、2个预训练模型.pkl格式分别对应600025.SH的分类与回归任务、2个示例数据data.csv、trades.csv及依赖说明等整体仅598KB普通笔记本即可运行调试。已有423人学习下载适合快速上手多Agent架构设计思想——选股、风控、择时、复盘四大功能解耦清晰配套Streamlit交互界面免前端开发支持单票深度择时推演并自动生成结构化周度复盘报告所有逻辑均基于当日及历史数据具备实盘参考价值。1. 项目概述一个实战派的AI量化交易系统最近几年AI炒股或者说量化交易已经从机构专属的神坛逐渐走向了个人开发者和交易爱好者的视野。但说实话很多开源项目要么是“玩具级”的只能跑个回测离实盘十万八千里要么就是“黑盒级”的代码复杂得像天书你根本不知道它为什么买、为什么卖。我自己在金融科技领域摸爬滚打了十几年从传统策略写到机器学习踩过的坑不计其数。所以我一直想动手做一个东西它得是从零开始、结构清晰、每一步都有据可循的同时还得是多模型、多任务协同的不能只靠一个模型打天下。于是就有了这个项目一个基于多Agent架构和LGBM双模型的AI炒股系统。它的核心目标很明确第一实现有效的择时分析告诉你什么时候该进场、什么时候该离场第二完成深度的周度复盘不仅仅是看盈亏更要分析策略为什么有效或失效为下一周的调整提供依据。这不仅仅是把数据喂给模型然后输出买卖信号那么简单它更像是一个由多个“智能体”分工协作的虚拟交易团队有的负责选股有的负责择时有的负责风控还有的负责事后复盘总结。这个系统适合谁呢首先它适合有一定Python编程基础对机器学习尤其是树模型和金融市场有基本了解的朋友你想知道一个可用的量化系统到底是怎么搭建起来的。其次它也适合那些已经写过一些简单策略但苦于策略不稳定、无法系统化评估和迭代的交易者。通过这个项目你能获得的不只是一套源代码更是一套构建稳健AI交易系统的工程化思维和方法论。我会把整个从数据获取、处理、特征工程、模型训练、多Agent调度到回测、模拟盘乃至实盘衔接的完整链条掰开揉碎了讲清楚。我们不用那些玄而又玄的“黑科技”就用扎实的工程实践和可解释性强的模型搭建一个你自己能完全掌控、能持续迭代的“AI交易员”。2. 核心架构设计为什么是多Agent与LGBM双模型在动手写代码之前我们必须把架构想清楚。为什么选择“多Agent架构”和“LGBM双模型”这个组合这背后是对于量化交易复杂性的深度考量而不是简单的技术堆砌。2.1 多Agent架构模拟一个专业的交易团队传统的量化策略往往是一个“单体巨无霸”函数里面混杂了数据获取、清洗、特征计算、信号生成、风险控制等所有逻辑。这种代码的维护和迭代是一场噩梦一个地方的修改可能会引发意想不到的连锁反应。更重要的是它不符合真实的投资决策流程。在真实的投资机构中决策是分工协作的研究员提供股票池和基本面观点量化分析师负责建模和信号计算交易员负责执行和盘口观察风控官时刻盯着仓位和风险指标。多Agent架构正是为了模拟这一专业化分工。在我们的系统中设计了以下几个核心AgentDataFetcher Agent数据获取代理它的职责单一而明确就是定时、稳定地从各种数据源如财经API、本地数据库获取原始的行情数据、基本面数据和宏观数据。它需要处理网络异常、数据格式不统一、数据缺失等问题并为下游Agent提供干净、统一的数据接口。它的存在将易变的数据获取逻辑与核心的策略逻辑解耦。FeatureEngineer Agent特征工程代理这是策略的“厨房”。它接收原始数据加工出模型所需的“食材”——即特征。这包括计算各种技术指标如MACD, RSI, 布林带、价量特征、波动率特征以及进行必要的标准化、归一化处理。这个Agent的优化空间极大特征的好坏直接决定了模型性能的天花板。LGBM_Predictor Agent预测代理这是系统的“大脑”之一。它装载着我们训练好的LightGBM模型。它的任务是根据FeatureEngineer提供的特征预测未来一段时间例如未来5天的股价涨跌概率或收益率。我们采用双模型设计一个模型专注于短期波动捕捉例如未来1-3天的涨跌另一个模型专注于中期趋势判断例如未来5-10天的方向。双模型可以相互验证降低单一模型误判的风险。SignalFusion Agent信号融合代理接收到来自双模型的预测结果后这个Agent负责做最终的决策。它不是简单地对预测结果取平均而是根据市场状态如波动率、趋势强度、模型近期表现如预测置信度进行加权融合。例如在震荡市中可能更信任短期模型在趋势市中则加大中期模型的权重。它输出的是统一的、带有置信度的交易信号如强烈买入、买入、观望、卖出。PortfolioManager Agent组合管理代理信号有了但怎么买买多少这就是它的工作。它根据信号强度、账户总资金、个股风险如波动率、以及相关性计算具体的仓位分配。它实现了简单的风险平价或均值-方差优化思想确保投资组合不会过度集中在某一只股票或某一个行业上。Backtest/Review Agent回测与复盘代理这是系统学习和进化的关键。它不仅在历史数据上模拟交易更重要的是在每周结束后进行周度复盘。它会分析本周哪些交易赚了钱、哪些亏了钱原因是什么是模型失效还是遇到了极端行情它会生成详细的复盘报告包括收益曲线、最大回撤、夏普比率、交易胜率等并且会标注出需要重点关注的问题为下一周的特征调整或模型迭代提供方向。注意Agent之间通过清晰定义的接口如Python类方法、消息队列进行通信每个Agent内部状态独立。这种设计让系统具备了极强的可扩展性。未来如果你想加入一个基于Transformer的Agent或者一个专门处理新闻情感的Agent只需要让它遵循同样的通信规范就能轻松接入系统而不需要重写其他部分。2.2 LGBM双模型稳健与可解释性的平衡为什么选择LightGBMLGBM而不是更“时髦”的深度学习模型如LSTM、Transformer效率与性能在表格数据金融数据本质上是时间序列表格数据上梯度提升树模型如XGBoost, LightGBM, CatBoost长期以来都是性能霸主。它们训练速度快对特征缺失不敏感且通常不需要像深度学习那样复杂的调参和大量的数据。可解释性这是关键。炒股最怕“黑盒”。LGBM提供了特征重要性排序我们可以清楚地知道是哪些指标例如“成交量放大率”、“过去20日波动率”对模型的决策影响最大。这在进行周度复盘时至关重要如果本周策略亏损我们可以通过分析特征重要性判断是市场逻辑变了原来重要的特征失效了还是出现了新的驱动因素。双模型设计的考量模型A短期模型输入是近期高频特征如分钟级价量变化、盘口失衡目标是预测未来1-3日的涨跌。它更敏感试图捕捉市场短期的情绪和资金流动。模型B中期模型输入是中长期特征如周度均线排列、行业资金流向、宏观指标变化目标是预测未来5-10日的趋势方向。它更稳健试图把握基本面的动量。分工与协作短期模型可能发出许多交易信号但通过中期模型的“趋势过滤器”可以过滤掉那些逆中期趋势的“噪音信号”从而提高整体胜率。例如短期模型看涨但中期模型显示下行趋势那么SignalFusion Agent可能会降低该信号的权重甚至将其忽略。3. 从零搭建环境准备与核心模块实现理论讲完了我们开始动手。假设你已经有Python基础并且安装了Anaconda或Miniconda。我们从最干净的环境开始。3.1 项目环境与依赖管理首先创建一个独立的虚拟环境是专业开发的第一步它能避免包版本冲突。# 创建名为 ai_trader 的虚拟环境 conda create -n ai_trader python3.9 -y conda activate ai_trader接着初始化项目目录并使用requirements.txt严格管理依赖。这是项目可复现性的生命线。my_ai_trading_system/ ├── agents/ # 存放所有Agent的类定义 ├── data/ # 存放原始和加工后的数据 ├── models/ # 存放训练好的LGBM模型文件 ├── utils/ # 存放工具函数数据获取、指标计算等 ├── config.py # 全局配置文件参数、路径 ├── main.py # 系统主入口 ├── train.py # 模型训练脚本 ├── backtest.py # 回测与复盘脚本 └── requirements.txt # 项目依赖requirements.txt文件内容如下我注释了每个核心库的用途# 数据处理与分析 pandas1.4.0 numpy1.22.0 # 机器学习模型 lightgbm3.3.0 # 核心模型库 scikit-learn1.0.0 # 用于数据划分、评估指标 # 数据获取 (这里以Tushare为例你可以替换为其他数据源) tushare1.2.89 # 消息队列用于高级的Agent异步通信初期可用简单方式替代 # pika1.2.0 # 回测引擎我们也可以自己实现但使用成熟库更稳定 backtrader1.9.76.123 # 强大的回测框架 # 可视化 matplotlib3.5.0 seaborn0.11.0 # 配置文件管理 pyyaml6.0 # 进度条 tqdm4.64.0使用pip install -r requirements.txt一键安装所有依赖。3.2 核心Agent的代码实现以FeatureEngineer和LGBM_Predictor为例我们深入两个最核心的Agent看看代码具体如何组织。FeatureEngineer Agent它的核心是一个FeatureEngineer类包含各种特征计算方法。# agents/feature_engineer.py import pandas as pd import numpy as np from typing import Dict, Optional class FeatureEngineer: def __init__(self, config: Dict): self.config config # 可以在这里初始化一些中间状态比如移动平均线的缓存 def calculate_technical_indicators(self, df: pd.DataFrame) - pd.DataFrame: 计算一系列技术指标特征。 df 需包含 columns: [open, high, low, close, volume] # 避免修改原数据 df df.copy() # 1. 价格衍生特征 df[returns] df[close].pct_change() df[log_returns] np.log(df[close] / df[close].shift(1)) # 2. 移动平均线及相关特征 df[ma5] df[close].rolling(window5).mean() df[ma20] df[close].rolling(window20).mean() df[ma60] df[close].rolling(window60).mean() df[price_vs_ma20] df[close] / df[ma20] - 1 # 价格偏离20日均线的百分比 # 3. 布林带 df[rolling_std] df[close].rolling(window20).std() df[bb_upper] df[ma20] 2 * df[rolling_std] df[bb_lower] df[ma20] - 2 * df[rolling_std] df[bb_width] (df[bb_upper] - df[bb_lower]) / df[ma20] # 布林带宽度衡量波动率 df[bb_position] (df[close] - df[bb_lower]) / (df[bb_upper] - df[bb_lower]) # 价格在布林带中的位置 # 4. RSI (相对强弱指数) delta df[close].diff() gain (delta.where(delta 0, 0)).rolling(window14).mean() loss (-delta.where(delta 0, 0)).rolling(window14).mean() rs gain / loss df[rsi] 100 - (100 / (1 rs)) # 5. 成交量相关特征 df[volume_ma5] df[volume].rolling(window5).mean() df[volume_ratio] df[volume] / df[volume_ma5] # 成交量相对变化 # 6. 波动率特征 df[volatility_20d] df[returns].rolling(window20).std() * np.sqrt(252) # 年化波动率 # 删除因滚动计算产生的NaN行 df.dropna(inplaceTrue) return df def run(self, raw_data: pd.DataFrame) - pd.DataFrame: 主运行方法对外接口 featured_data self.calculate_technical_indicators(raw_data) # 这里可以链式调用其他特征计算方法如基本面特征、宏观特征等 return featured_data实操心得特征工程是永无止境的。上述只是最基础的示例。在实际项目中你需要不断尝试和挖掘有效的特征例如横截面特征个股指标在全市场或同行业中的分位数排名。动量与反转特征过去N日的累计收益率、特定形态识别如金叉死叉。资金流特征基于Level-2数据的主动买入/卖出金额。注意避免未来函数这是量化的大忌。任何特征的计算只能使用当前及之前的历史数据绝不能用到未来的数据。rolling().mean()这类操作是安全的因为它只用到窗口内的历史数据。LGBM_Predictor Agent这个Agent负责加载模型并进行预测。# agents/lgbm_predictor.py import joblib import lightgbm as lgb import pandas as pd import numpy as np from typing import Tuple class LGBMPredictor: def __init__(self, model_short_path: str, model_mid_path: str): 初始化加载短期和中期两个模型。 try: self.model_short joblib.load(model_short_path) # 或 lgb.Booster(model_file...) self.model_mid joblib.load(model_mid_path) print(f模型加载成功: {model_short_path}, {model_mid_path}) except FileNotFoundError as e: print(f模型文件未找到: {e}) # 可以在这里触发模型训练流程 self.model_short None self.model_mid None def predict(self, features_df: pd.DataFrame) - Tuple[np.ndarray, np.ndarray]: 使用双模型进行预测。 返回: (短期模型预测结果, 中期模型预测结果) 预测结果可以是概率值分类任务或具体数值回归任务。 if self.model_short is None or self.model_mid is None: raise ValueError(模型未加载无法进行预测。) # 确保特征列的顺序与模型训练时完全一致这是线上预测最常见的坑。 # 通常需要在训练时保存一个特征列名列表在这里进行对齐。 # 假设我们有一个 self.feature_columns 属性保存了列顺序 # features_df features_df[self.feature_columns] X features_df.values # 转换为numpy数组 pred_short self.model_short.predict(X) pred_mid self.model_mid.predict(X) return pred_short, pred_mid def get_feature_importance(self, model_type: str short) - pd.DataFrame: 获取模型特征重要性用于复盘分析 model self.model_short if model_type short else self.model_mid if isinstance(model, lgb.Booster): importance model.feature_importance(importance_typegain) feature_names model.feature_name() else: # 如果是sklearn API包装的模型 importance model.feature_importances_ feature_names model.feature_name_ # 假设训练时传入了feature_name return pd.DataFrame({ feature: feature_names, importance: importance }).sort_values(importance, ascendingFalse)4. 模型训练与特征工程实战有了Agent的骨架我们需要用数据来填充血肉。模型训练是离线的、但至关重要的环节。4.1 数据准备与标注我们以A股日线数据为例使用Tushare Pro获取数据。目标是预测未来N日的涨跌分类问题或收益率回归问题。# utils/data_fetcher.py (简化版) import tushare as ts import pandas as pd from datetime import datetime, timedelta class DataFetcher: def __init__(self, token): ts.set_token(token) self.pro ts.pro_api() def get_daily_data(self, ts_code, start_date, end_date): 获取日线行情数据 df self.pro.daily(ts_codets_code, start_datestart_date, end_dateend_date) df[trade_date] pd.to_datetime(df[trade_date]) df.set_index(trade_date, inplaceTrue) df.sort_index(inplaceTrue) # 重命名列以符合我们的特征计算函数 df.rename(columns{open:open, high:high, low:low, close:close, vol:volume}, inplaceTrue) return df[[open, high, low, close, volume]] # 假设我们获取了多只股票的数据并合并 # all_data pd.concat({code: df for code, df in stock_data_dict.items()}, names[ts_code, trade_date])接下来是关键步骤创建标签。我们以“未来5日收益率是否超过阈值”作为二分类标签。def create_label(df, future_days5, threshold_pct0.02): 创建分类标签。 df: 包含close列的DataFrame 未来future_days日的收益率 threshold_pct 则标记为1上涨否则为0。 # 计算未来N日的收益率 future_return df[close].shift(-future_days) / df[close] - 1 # 创建标签 label (future_return threshold_pct).astype(int) # 由于使用了未来数据最后future_days行的标签是NaN需要删除 df[label] label df.dropna(subset[label], inplaceTrue) return df4.2 训练流程与参数调优训练脚本train.py负责组织整个流程数据加载 - 特征工程 - 划分训练/验证集 - 训练LGBM模型 - 评估 - 保存。# train.py 核心部分 import pandas as pd from sklearn.model_selection import train_test_split, TimeSeriesSplit from sklearn.metrics import accuracy_score, f1_score, roc_auc_score import lightgbm as lgb import joblib from agents.feature_engineer import FeatureEngineer from utils.data_fetcher import DataFetcher, create_label def train_model(data_df, label_collabel, model_nameshort_term): 训练一个LGBM模型 # 1. 分离特征和标签 X data_df.drop(columns[label_col]) y data_df[label_col] # 2. 划分数据集 - 对于时间序列必须按时间划分不能随机打乱 split_idx int(len(X) * 0.8) # 80%训练20%测试 X_train, X_test X.iloc[:split_idx], X.iloc[split_idx:] y_train, y_test y.iloc[:split_idx], y.iloc[split_idx:] # 3. 定义模型参数 params { objective: binary, # 二分类 metric: auc, # 评估指标 boosting_type: gbdt, num_leaves: 31, learning_rate: 0.05, feature_fraction: 0.9, # 防止过拟合 bagging_fraction: 0.8, bagging_freq: 5, verbose: -1, seed: 42 } # 4. 创建数据集 lgb_train lgb.Dataset(X_train, y_train) lgb_eval lgb.Dataset(X_test, y_test, referencelgb_train) # 5. 训练 print(f开始训练 {model_name} 模型...) gbm lgb.train(params, lgb_train, num_boost_round1000, valid_sets[lgb_eval], callbacks[lgb.early_stopping(stopping_rounds50), lgb.log_evaluation(50)]) # 6. 评估 y_pred_prob gbm.predict(X_test, num_iterationgbm.best_iteration) y_pred (y_pred_prob 0.5).astype(int) auc roc_auc_score(y_test, y_pred_prob) accuracy accuracy_score(y_test, y_pred) print(f{model_name} 模型测试集 AUC: {auc:.4f}, Accuracy: {accuracy:.4f}) # 7. 保存模型和特征列非常重要 joblib.dump(gbm, fmodels/{model_name}_model.pkl) # 保存训练时的特征列顺序 feature_columns X_train.columns.tolist() joblib.dump(feature_columns, fmodels/{model_name}_features.pkl) print(f模型和特征列已保存至 models/ 目录) return gbm # 主函数 if __name__ __main__: # 获取数据、特征工程、创建标签... # df ... # 分别用不同参数创建短期和中期标签训练两个模型 # df_short create_label(df.copy(), future_days3, threshold_pct0.015) # df_mid create_label(df.copy(), future_days10, threshold_pct0.03) # model_short train_model(df_short, model_namemodel_short) # model_mid train_model(df_mid, model_namemodel_mid)注意事项时间序列泄露绝对不能用未来的信息预测过去。划分数据集必须按时间顺序。使用TimeSeriesSplit进行交叉验证更严谨。样本不均衡股市中大涨大跌的日子是少数。如果正样本标签为1远少于负样本需要在params中设置is_unbalance: True或使用scale_pos_weight参数。特征重要性分析训练后务必查看gbm.feature_importance()。如果某些特征重要性极低可以考虑剔除简化模型。过拟合通过num_leaves、min_data_in_leaf、feature_fraction、bagging_fraction等参数控制模型复杂度并依靠验证集早停。5. 系统整合与回测复盘当各个Agent和模型都准备好后我们需要一个“指挥官”来调度它们这就是main.py。同时一个严谨的回测与复盘流程是检验系统成败的唯一标准。5.1 主流程调度与信号生成main.py模拟了每个交易日的决策流程。# main.py (核心逻辑) import pandas as pd from datetime import datetime import schedule import time from agents.data_fetcher_agent import DataFetcherAgent from agents.feature_engineer_agent import FeatureEngineerAgent from agents.lgbm_predictor_agent import LGBMPredictorAgent from agents.signal_fusion_agent import SignalFusionAgent from agents.portfolio_manager_agent import PortfolioManagerAgent from config import Config def daily_trading_job(): 每日收盘后执行的任务 print(f\n 执行交易日任务 {datetime.now().date()} ) # 1. 初始化所有Agent config Config() data_agent DataFetcherAgent(config) feature_agent FeatureEngineerAgent(config) predictor_agent LGBMPredictorAgent(config) signal_agent SignalFusionAgent(config) portfolio_agent PortfolioManagerAgent(config) # 2. 数据获取 print(步骤1: 获取市场数据...) raw_data_dict data_agent.fetch_market_data() # 返回一个字典key为股票代码value为DataFrame all_signals [] for ts_code, raw_df in raw_data_dict.items(): # 3. 特征工程 print(f 处理 {ts_code}: 特征工程...) feature_df feature_agent.run(raw_df) # 4. 模型预测 (只取最新一行的特征进行预测) latest_features feature_df.iloc[[-1]] # 获取最新一天的特征 pred_short, pred_mid predictor_agent.predict(latest_features) # 5. 信号融合 signal_strength, action signal_agent.fuse(pred_short[0], pred_mid[0], ts_code, feature_df) all_signals.append({ ts_code: ts_code, date: latest_features.index[0], pred_short: pred_short[0], pred_mid: pred_mid[0], signal: signal_strength, action: action # BUY, SELL, HOLD }) # 6. 组合管理 print(步骤5: 生成投资组合建议...) portfolio portfolio_agent.generate_portfolio(all_signals) print(f今日建议组合:\n{portfolio.to_string()}) # 7. 记录日志 (为复盘做准备) log_today_decision(all_signals, portfolio) print( 交易日任务完成 \n) def log_today_decision(signals, portfolio): 将当日决策记录到CSV或数据库 log_df pd.DataFrame(signals) log_df[portfolio_weight] log_df[ts_code].map(portfolio.set_index(ts_code)[weight]) log_df.to_csv(flogs/decision_{datetime.now().date()}.csv, modea, headerFalse) if __name__ __main__: # 模拟定时任务实际部署可用cron或APScheduler print(AI交易系统启动...) # 立即运行一次 daily_trading_job() # 设定每天下午4点收盘后运行 schedule.every().day.at(16:00).do(daily_trading_job) while True: schedule.run_pending() time.sleep(60)5.2 回测引擎实现与绩效分析回测不是简单的“信号出现就买卖”它需要模拟真实的交易场景考虑手续费、滑点、涨停跌停无法买卖等限制。我们使用成熟的backtrader框架来构建严谨的回测。# backtest.py import backtrader as bt import pandas as pd from agents.signal_fusion_agent import SignalFusionAgent class AITraderStrategy(bt.Strategy): params ( (signal_agent, None), # 传入我们自定义的信号Agent (portfolio_agent, None), # 传入组合管理Agent ) def __init__(self): # 跟踪订单和持仓 self.orders {} self.signals {} def next(self): 在每个Bar如每天结束时调用 current_date self.datas[0].datetime.date(0) # 1. 获取所有股票池的最新数据生成特征这里简化实际应从feature_agent获取 # 2. 调用signal_agent和portfolio_agent生成当日信号和仓位 # 3. 根据目标仓位调整现有仓位 for data in self.datas: symbol data._name target_weight ... # 从portfolio_agent获取该股票的目标权重 current_value self.getposition(data).size * data.close[0] total_value self.broker.getvalue() target_value total_value * target_weight # 计算需要交易的数量 size_to_trade (target_value - current_value) / data.close[0] if abs(size_to_trade) 0: self.order_target_percent(data, targettarget_weight) def notify_order(self, order): 订单状态回调用于处理成交、拒绝等情况 if order.status in [order.Completed]: print(f{order.data._name} 订单成交, 价格: {order.executed.price:.2f}, 数量: {order.executed.size}) def run_backtest(data_feed_dict, signal_agent, portfolio_agent, initial_cash100000.0): 运行回测 cerebro bt.Cerebro() cerebro.broker.setcash(initial_cash) cerebro.broker.setcommission(commission0.0003) # 设置手续费万三 cerebro.addsizer(bt.sizers.PercentSizer, percents95) # 每次使用95%的可用资金留有余地 # 添加数据 for symbol, df in data_feed_dict.items(): data bt.feeds.PandasData(datanamedf, namesymbol) cerebro.adddata(data) # 添加策略 cerebro.addstrategy(AITraderStrategy, signal_agentsignal_agent, portfolio_agentportfolio_agent) print(初始资金: %.2f % cerebro.broker.getvalue()) cerebro.run() print(最终资金: %.2f % cerebro.broker.getvalue()) # 绘制图表 cerebro.plot(stylecandlestick)5.3 周度复盘超越盈亏的深度分析回测结束后生成资金曲线只是第一步。周度复盘才是系统进化的核心。每周你应该运行一个复盘脚本至少分析以下几点绩效归因收益来源本周收益主要来自哪几只股票是Alpha超额收益还是Beta市场波动亏损分析亏损的交易是否符合策略逻辑是模型预测错误还是遇到了黑天鹅事件如个股暴雷模型在亏损交易发生前的预测置信度如何代码实现计算夏普比率、最大回撤、卡玛比率、胜率、盈亏比等关键指标并与上周、上月对比。模型与信号诊断特征重要性变化对比本周和上周的特征重要性排名是否有显著变化如果有可能意味着市场主导逻辑发生了切换。信号有效性统计“强烈买入”信号后续的实际上涨概率“卖出”信号后续的实际下跌概率。信号是否依然有效双模型一致性分析短期模型和中期模型发出冲突信号的案例最终市场走向如何这有助于优化SignalFusion Agent的融合规则。风险检查持仓集中度是否单只股票或单个行业仓位过重策略容量当前策略管理的资金量是否接近其有效容量模拟加大资金量后滑点成本是否显著增加极端行情表现在市场大涨大跌如涨跌幅超过2%的日子里策略表现如何是跟上了还是反向生成复盘报告 将上述分析自动化生成一份结构化的Markdown或PDF报告。报告模板可以包括## 第XX周交易复盘报告 ### 一、整体绩效概览 - 周收益率X.XX% - 基准如沪深300收益率X.XX% - 超额收益率X.XX% - 本周最大回撤X.XX% - 交易次数XX次胜率XX% ### 二、关键交易分析 #### 最佳交易Top 3 1. 股票A买入理由模型信号持有X天收益率X% 2. ... #### 最差交易Bottom 3 1. 股票B买入理由亏损原因分析模型误判/市场系统性风险/个股事件 2. ... ### 三、模型健康度检查 - 特征重要性Top 5[特征1, 特征2...] - 与上周对比变化[特征X上升特征Y下降] - 信号准确率买入信号准确率XX%卖出信号准确率XX% ### 四、风险提示与下周计划 - 当前主要风险... - 下周调整计划考虑调整特征X的权重/尝试加入新的特征Y/微调模型融合参数至...6. 常见问题、避坑指南与进阶思考在实际搭建和运行过程中你会遇到无数个坑。这里记录一些最具代表性的问题和我的解决方案。6.1 数据与特征工程相关Q1: 数据源不稳定或API限制怎么办A1: 这是第一个拦路虎。务必实现数据本地化缓存。每次从API获取数据后立即保存到本地数据库如SQLite或Parquet文件。下次请求时先检查本地是否有数据只获取增量部分。同时为DataFetcher Agent设计重试机制和优雅降级策略例如主API失败后切换备用数据源。Q2: 如何避免“未来函数”A2: 这是量化策略失效的元凶之一。必须建立严格的代码审查清单任何特征计算只允许使用.shift()、.rolling()、.expanding()等面向过去窗口的函数。绝对禁止使用.shift(-N)未来数据或任何需要未来信息才能计算的指标例如用到当日最高最低价的指标必须在收盘后、下一日开盘前计算。在特征工程模块中对每个特征函数进行“未来函数”标注和单元测试。Q3: 特征太多导致过拟合怎么办A3: 遵循“由简入繁”的原则。初阶从少数几个经典且逻辑清晰的技术指标开始如MA, RSI, 布林带。中阶使用LGBM自带的特征重要性进行筛选剔除重要性接近零的特征。可以尝试使用SelectFromModel进行自动化特征选择。高阶引入递归特征消除RFE或使用PCA等降维方法处理高度相关的特征组。但要注意降维后的特征可解释性会变差。6.2 模型训练与预测相关Q4: 模型在训练集上表现很好但在回测或实盘上很差过拟合。A4: 这是机器学习在量化中最常见的挑战。数据划分必须使用时间序列交叉验证TimeSeriesSplit而不是随机划分。参数调优使用GridSearchCV或Optuna等工具进行超参数优化时验证集也必须按时间顺序划分。正则化加大LGBM的min_data_in_leaf、min_gain_to_split降低num_leaves增加feature_fraction和bagging_fraction。简化模型有时候一个参数更少、更简单的模型其泛化能力反而更强。Q5: 市场风格切换模型突然失效。A5: 没有永远有效的模型。我们的多Agent和周度复盘机制就是为了应对这一点。动态再训练设定一个规则例如当模型在最近一个月的滚动回测中夏普比率低于某个阈值时自动触发模型的重新训练。LGBM_Predictor Agent需要具备模型热更新的能力。集成学习除了双模型可以训练多个在不同时间区间或不同股票池上表现良好的模型让SignalFusion Agent根据当前市场波动率、趋势等宏观指标动态选择或加权集成这些模型。6.3 系统与实盘相关Q6: 回测结果完美但模拟盘/实盘效果不佳。A6: 回测与实盘的差距主要来自滑点与手续费回测中假设按收盘价成交实盘中大单交易会有冲击成本。在回测中必须设置更保守的滑点例如0.1%和手续费。涨停跌停回测中必须加入“涨停不能买入跌停不能卖出”的逻辑。数据延迟与异步实盘数据是流式的且不同数据源可能有微小延迟。确保你的系统在收到所有必要数据后才触发决策逻辑并考虑使用事件驱动架构替代简单的定时轮询。Q7: 如何将系统部署到生产环境A7: 从本地脚本到稳定运行的系统需要工程化升级。容器化使用Docker将整个系统及其依赖打包确保环境一致性。任务调度使用专业的调度系统如Apache Airflow或Celery替代简单的schedule库管理复杂的依赖任务如数据更新-特征计算-模型预测-下单。监控与告警为系统添加关键指标监控如数据更新是否成功、模型预测耗时、Agent心跳并设置告警如邮件、钉钉机器人。Backtest/Review Agent可以定期自动运行并将复盘报告发送给你。Q8: 这个系统能保证赚钱吗A8:绝对不能。这是一个需要你深刻理解的核心理念。本系统提供的是一套严谨、可解释、可迭代的量化研究框架和工具。它帮助你系统性地验证想法减少情绪化交易但它无法预测未来。市场的有效性意味着任何公开的、简单的规律都会迅速消失。你的核心竞争力将来自于1对市场独特的理解转化为独特的特征2更快的迭代速度通过自动化复盘和再训练3更严格的风险管理通过PortfolioManager Agent实现。把这个系统看作是你的“AI研究员”和“纪律执行者”而真正的战略决策和进化方向依然依赖于你作为“基金经理”的洞察力。从历史数据中发现的规律在未来可能会失效持续的学习、验证和调整才是你在市场中长期生存的根本。本文还有配套的精品资源点击获取