最近看了一些关于 Two Sigma 的公开访谈和工程分享里面关于“AI 如何真正落地到投资场景”的内容反复在讲几件事数据质量、特征工程、回测的坑以及模型失效时的风险控制。这些观点放到非金融的 AI 工程里也同样成立。这篇文章不打算逐字复述访谈而是把其中核心的技术思想提取出来再结合一个可运行的量化投资最小案例拆解 AI 投资系统的完整链路。适合的读者有三类一是刚开始接触量化投资的算法工程师想理解模型之外还有哪些关键环节二是做 AI 平台建设、想了解金融场景如何落地机器学习的技术人三是正在学习 Python 数据分析、想找一个综合性实战项目的开发者。读完本文你会理解 AI 投资的整体架构掌握数据、特征、模型、回测、风控五个核心环节并拿到一套可以直接运行的 Python 最小选股流水线。先说一句必要的提醒本文所有内容仅作为技术学习和工程思路参考不构成任何投资建议。1. Two Sigma 是谁它的 AI 投资思路为什么值得学1.1 Two Sigma 在量化投资行业的定位Two Sigma 是一家成立于 2001 年的量化投资管理公司创始人是 David Siegel 和 John Overdeck。业内对它的普遍印象是一家更像科技公司的投资机构。它很早就把机器学习、人工智能、大数据和分布式计算应用到投资研究里而不是靠传统的人工基本面分析为主。在 Two Sigma 的公开资料里经常出现几个关键词数据Data研究平台Research Platform机器学习Machine Learning严谨回测Rigorous Backtesting风险管理Risk Management这里值得注意的一点是Two Sigma 并不把自己定位成“纯 AI 公司”而是金融公司里最会做 AI 工程化的那一类。它强调的不是某个模型多厉害而是整套系统的稳定性和可扩展性。1.2 AI 赋能投资的本质不是“用模型替代人”关于“AI 赋能投资”很多人的第一反应是AI 是不是能自动选股、自动交易然后躺赚真实的量化投资远没有这么简单。Two Sigma 的公开观点更接近下面这个模型人类研究员负责提出假设、定义问题、分析异常 AI 系统负责处理海量数据、生成特征、训练预测模型 风控与工程师负责控制回撤、监控模型漂移、保障系统稳定也就是说AI 在投资中的作用是“放大器”和“加速器”它把研究员原本需要几周才能完成的验证过程压缩到几小时同时能够持续处理人类无法手工阅读的海量数据。但它并没有消除投资决策中的不确定性。在实际工程中这意味着我们要构建的不是一个“会预测涨跌的神奇模型”而是一条完整的流水线数据接入 - 数据清洗 - 特征计算 - 模型训练 - 回测验证 - 模拟交易 - 实盘执行 - 风险监控后面所有的技术内容都会围绕这条流水线展开。2. AI 投资系统的整体架构从工程视角看一套 AI 投资系统可以拆成五个层级。2.1 数据层数据是 AI 投资的起点。除了传统的量价数据开盘价、收盘价、成交量、成交额Two Sigma 这类机构还会使用大量另类数据比如卫星拍摄的停车场车辆数用于估计零售企业的经营情况信用卡交易流水用于推断消费趋势天气数据用于分析农业、能源等行业的短期波动新闻情绪、社交舆情数据用于事件驱动策略但是数据层最大的成本不是买数据而是清洗数据。真实数据里普遍存在缺失值、异常值、复权价格调整、财报口径变化、公司停牌等问题。任何一步处理不到位都会直接影响后续特征和模型的质量。2.2 特征层特征层是把原始数据加工成模型可用输入的地方。在量化投资里特征通常被称为因子Factor。常见的因子包括动量因子过去 N 天的累计收益率反转因子过去 N 天涨多了未来可能跌波动率因子过去 N 天的收益率标准差流动性因子成交量、换手率、买卖价差基本面因子市盈率、市净率、营收增速特征层的核心挑战是如何避免前视偏差Look-ahead Bias。简单说就是用 T 日的数据去预测 T1 日的收益特征里不能包含任何 T1 日才知道的信息。2.3 模型层模型层负责学习特征与未来收益之间的关系。传统量化里常用线性回归、时间序列模型现在的 AI 量化系统更多使用梯度提升树XGBoost、LightGBM适合表格型特征神经网络LSTM、Transformer适合序列数据和另类数据深度强化学习用于优化交易执行而不是直接预测涨跌Two Sigma 在公开分享中特别强调模型复杂度不是第一优先级稳定性和可解释性才是。一个简单模型如果特征扎实、数据干净往往比一个复杂模型在实盘中的表现更稳定。2.4 回测与执行层回测是用历史数据验证策略收益和风险的过程。一套合格的回测系统必须处理交易成本佣金、印花税、滑点涨跌停限制涨停时买不进跌停时卖不出最小交易单位A 股一手是 100 股停牌股票不可交易下单延迟与市场冲击如果回测系统过于简化结果往往非常漂亮但实盘表现会大打折扣。这也是 Two Sigma 反复提醒的“回测偏差”问题。2.5 风险与监控层金融场景中风险控制比收益更重要。模型上线后必须持续监控几个关键指标策略净值回撤最大回撤超过阈值时是否需要降仓行业暴露、风格暴露是否漂移模型预测分布是否发生了显著变化数据源是否出现缺失或延迟交易执行是否偏离预期一个优秀的 AI 投资系统不是追求每笔都赚而是确保长期风险可控、在极端行情下不会爆仓。3. 环境准备与项目结构下面进入实操环节。我们通过一个 Python 项目模拟从数据生成到模型回测的完整流程。3.1 运行环境操作系统Windows / macOS / Linux 均可Python 版本3.8 或更高推荐 3.10主要依赖库pandas numpy scikit-learn模型部分为了简单可视化也可以额外安装 matplotlib但本案例不强制使用。安装方式pip install pandas numpy scikit-learn matplotlib如果你的环境里同时存在 Python 2 的环境建议使用虚拟环境python -m venv quant_ai_venv source quant_ai_venv/bin/activate # Windows 下为 quant_ai_venv\Scripts\activate3.2 项目目录我们创建一个名为quant_ai_demo的目录quant_ai_demo/ ├── data_generator.py # 生成合成行情数据 ├── pipeline.py # 主流程特征、模型、回测 └── README.md实际项目中更合理的结构会拆得更细quant_research/ ├── data/ │ ├── raw_data/ # 原始数据 │ └── processed_data/ # 清洗后数据 ├── features/ │ ├── factor_calc.py # 因子计算 │ └── factor_store.py # 因子存储 ├── models/ │ ├── train.py # 训练脚本 │ ├── predict.py # 预测脚本 │ └── evaluate.py # 模型评估 ├── backtest/ │ ├── engine.py # 回测引擎 │ └── report.py # 回测报告 └── risk/ └── monitor.py # 风险监控案例为了便于学习和展示会简化成两个文件但逻辑分层保持清晰。4. 核心原理拆解AI 投资中的关键工程要点4.1 为什么特征比模型更关键Two Sigma 的公开访谈里反复提到他们花费大量时间在数据验证和特征工程上。原因很简单机器学习模型只能学习数据中已有的规律如果特征本身没有预测能力换任何复杂模型都很难产生稳定的 alpha。在时序金融数据上做特征工程要特别小心以下问题用未来数据这是最严重的错误属于标签泄漏用全样本做标准化会引入未来统计量在训练阶段就“偷看”了未来分布因子与标签之间的因果关系不成立比如用当天收盘后的网络舆情预测当天收益率因此在实际项目中特征计算通常是逐日滚动进行的每个特征只能使用“截至当日已经发生”的数据。4.2 滚动窗口训练避免时间穿越金融数据不是独立同分布的今天和明天之间存在序列相关性。如果把数据随机打乱再做交叉验证模型会学到未来信息导致回测结果虚高。正确做法是使用时序切分按时间排序数据。前 70% 的时间段作为训练集后 30% 作为测试集。如果数据量足够大可以使用滚动训练不断把最新数据加入训练集丢弃最旧数据让模型适应市场变化。本案例中我们会采用一次性的时间切分来保持代码简单但会说明滚动窗口的扩展方式。4.3 标签的定义决定模型上限在监督学习中标签决定了模型学习的方向。在 AI 选股场景里常见标签有分类标签未来 N 日收益率是否大于 0回归标签未来 N 日收益率的具体数值排序标签未来 N 日收益率的横截面排名不同的标签会让模型变成不同的任务。分类任务简单易用但丢失了收益率幅度信息回归任务能预测具体收益但噪声更大排序任务更适合投资场景因为它关注的是“哪只股票会更好”而不是“绝对收益率是多少”。本案例使用分类标签未来 1 日收益率是否大于 0。4.4 回测偏差一个隐蔽但致命的坑即使模型在训练集和测试集上都表现稳定回测结果仍然可能失真常见原因包括偏差类型说明常见来源前视偏差使用未来信息计算特征或信号财报数据未做滞后处理幸存者偏差只选择当前仍存在的股票忽略了退市股票股票池未包含历史退市样本过拟合偏差回测参数被反复调整到历史最优在回测集上不断试参数交易成本偏差回测忽略滑点、手续费、冲击成本回测引擎未建模成交限制避免回测偏差的关键不是消除某一个偏差而是建立一套完整的“白盒回测机制”每一次回测都详细记录数据版本、特征版本、模型参数、交易成本假设确保任何结果都可以复现。5. 完整实战搭建一个最小 AI 选股流水线下面我们实现一个完整可运行的 AI 选股最小案例。案例包含四个步骤生成合成行情数据构造特征与标签训练随机森林分类模型执行简版回测并输出绩效指标5.1 生成合成行情数据我们先创建data_generator.py用来生成模拟股票价格数据。这里使用几何随机游走并加入一个微弱的动量效应让数据不是完全随机。# 文件路径quant_ai_demo/data_generator.py import numpy as np import pandas as pd def generate_market_data(periods1200, n_stocks8, seed42): 生成合成行情数据。 参数 ---- periods : int 交易日数量默认 1200约等于 5 年。 n_stocks : int 股票数量默认 8。 seed : int 随机种子便于结果复现。 返回 ---- DataFrame行索引为日期列名为 stock_0 ~ stock_{n_stocks-1}。 np.random.seed(seed) dates pd.date_range(2020-01-01, periodsperiods, freqB) all_prices {} for i in range(n_stocks): # 每只股票有不同的漂移项模拟行业差异 drift np.random.normal(0.0002, 0.0005) # 随机收益率 returns np.random.normal(drift, 0.02, periods) # 加入微弱的动量如果前一天上涨今天更可能继续小幅上涨 for t in range(1, periods): if returns[t - 1] 0: returns[t] 0.0003 # 从 100 元起步累积价格 price 100 * np.exp(np.cumsum(returns)) all_prices[fstock_{i}] price df pd.DataFrame(all_prices, indexdates) return df这里有两个细节值得说明使用np.exp(np.cumsum(returns))生成价格序列可以避免价格变成负数。加入的微弱动量项会让模型有一定学习空间但不保证一定能获得显著超额收益这是刻意为之更接近真实世界的噪声环境。5.2 构造特征与标签创建pipeline.py先写特征与标签构建函数。# 文件路径quant_ai_demo/pipeline.py import numpy as np import pandas as pd from sklearn.ensemble import RandomForestClassifier from sklearn.metrics import accuracy_score from data_generator import generate_market_data def build_features_and_labels(df, feat_window10): 将原始行情数据转换为特征和标签。 参数 ---- df : DataFrame 每股一列的价格数据。 feat_window : int 回看窗口特征最多使用最近 N 个交易日数据。 返回 ---- DataFrame包含特征列、标签列和未来收益列。 frames [] for col in df.columns: tmp pd.DataFrame(indexdf.index) # 基础特征收益率 tmp[stock] col tmp[ret_1] df[col].pct_change(1) tmp[ret_5] df[col].pct_change(5) tmp[ret_10] df[col].pct_change(10) # 波动率特征 tmp[vol_5] tmp[ret_1].rolling(5).std() tmp[vol_10] tmp[ret_1].rolling(10).std() # 标签与未来收益 # 未来 1 日收益率 tmp[future_ret_1] df[col].pct_change(1).shift(-1) # 标签未来 1 日收益是否为正 tmp[label] (tmp[future_ret_1] 0).astype(int) frames.append(tmp) all_data pd.concat(frames) all_data all_data.dropna() return all_data这段代码的核心思想是每只股票独立计算特征因此特征列只依赖股票自身历史行情。shift(-1)表示我们把未来一期的收益率移动到当前行作为标签或预测目标。这里要注意真实项目中还会有市场整体因子、行业因子以及个股横截面相对强度等特征。本案例为了突出流程只取了最简单的时序特征。5.3 训练模型并生成预测继续在pipeline.py中加入模型训练函数。def train_and_predict(all_data, train_ratio0.7, n_estimators100): 切分训练集/测试集训练随机森林返回预测结果和模型。 feature_cols [ret_1, ret_5, ret_10, vol_5, vol_10] # 保证按时间排序 all_data all_data.sort_index() X all_data[feature_cols].values y all_data[label].values # 按时间顺序切分不能随机打乱 split_idx int(len(all_data) * train_ratio) X_train, X_test X[:split_idx], X[split_idx:] y_train, y_test y[:split_idx], y[split_idx:] model RandomForestClassifier( n_estimatorsn_estimators, max_depth5, random_state42, ) model.fit(X_train, y_train) prob_test model.predict_proba(X_test)[:, 1] pred_test (prob_test 0.5).astype(int) acc accuracy_score(y_test, pred_test) print(f测试集准确率: {acc:.4f}) # 把预测概率放回测试数据中便于后续回测 test_data all_data.iloc[split_idx:].copy() test_data[prob] prob_test return test_data, model, feature_cols随机森林的predict_proba可以输出预测为“上涨”的概率这个概率可以被当作股票评分。评分越高代表模型认为该股票未来上涨的可能性越大。5.4 执行简版回测并输出绩效回测的核心逻辑是每一天在当日可交易的股票中选出评分最高的 2 只。假设两只股票等权持有。持有一天计算组合当日的平均收益。累计起来得到组合净值曲线。计算年化收益、年化波动率、夏普比率、最大回撤。def run_backtest(test_data, top_k2): 极简回测每天持有评分最高的 top_k 只股票等权持仓。 注意这是一个教学级回测没有考虑手续费、滑点、停牌等细节。 test_data test_data.sort_index() dates test_data.index.unique() daily_returns [] for dt in dates: day_df test_data.loc[dt] # 如果某天股票数据不足跳过 if isinstance(day_df, pd.Series): day_df day_df.to_frame().T if len(day_df) top_k: continue # 选评分最高的 top_k 只股票 top_k_df day_df.nlargest(top_k, prob) # 等权组合收益简化版没考虑交易成本 daily_ret top_k_df[future_ret_1].mean() daily_returns.append({date: dt, daily_ret: daily_ret}) ret_df pd.DataFrame(daily_returns).set_index(date) ret_df[cum_ret] (1 ret_df[daily_ret]).cumprod() # 基准全部股票等权每天再平衡 bench_daily test_data.groupby(test_data.index)[future_ret_1].mean() bench_cum (1 bench_daily).cumprod() # 计算策略绩效指标 strategy_ann_ret ret_df[daily_ret].mean() * 252 strategy_ann_vol ret_df[daily_ret].std() * np.sqrt(252) sharpe strategy_ann_ret / strategy_ann_vol if strategy_ann_vol 0 else 0 # 最大回撤 cum_max ret_df[cum_ret].cummax() drawdown ret_df[cum_ret] / cum_max - 1 max_drawdown drawdown.min() benchmark_ann_ret bench_daily.mean() * 252 print( 策略绩效摘要教学示例 ) print(f策略年化收益: {strategy_ann_ret:.2%}) print(f策略年化波动: {strategy_ann_vol:.2%}) print(f夏普比率: {sharpe:.2f}) print(f最大回撤: {max_drawdown:.2%}) print(f基准年化收益: {benchmark_ann_ret:.2%}) return ret_df, bench_cum def main(): print(1. 生成合成行情数据) df generate_market_data(periods1200, n_stocks8, seed42) print(2. 构建特征与标签) all_data build_features_and_labels(df) print(3. 训练模型并预测) test_data, model, feature_cols train_and_predict(all_data) print(4. 执行简版回测) ret_df, bench_cum run_backtest(test_data, top_k2) if __name__ __main__: main()5.5 运行与预期结果在项目目录下执行python pipeline.py预期会输出类似下面的内容1. 生成合成行情数据 2. 构建特征与标签 3. 训练模型并预测 测试集准确率: 0.5xxx 4. 执行简版回测 策略绩效摘要教学示例 策略年化收益: X.XX% 策略年化波动: XX.XX% 夏普比率: X.XX 最大回撤: -XX.XX% 基准年化收益: X.XX%由于我们设置了随机种子输出是稳定可复现的。但需要理解这只是演示流程不代表真实市场中的策略水平。如果你希望看到特征重要性可以在train_and_predict中加一行print(dict(zip(feature_cols, model.feature_importances_)))真实场景里特征重要性分析能帮你发现哪些因子对预测贡献最大方便后续进一步迭代。6. 常见问题与排查思路在 AI 投资系统的开发过程中下面这些问题几乎是必踩的坑。问题现象常见原因解决思路模型准确率很高超过 90%标签泄漏特征包含未来信息检查是否使用shift()滞后检查特征时序回测表现极好实盘表现极差回测交易成本设置过低或模型过拟合增加手续费、滑点模型减少参数搜索次数模型上线后每月收益持续衰减市场风格切换因子失效建立因子监控体系定期重训练模型股票停牌导致回测买入失败回测引擎未处理停牌状态根据交易所停牌数据过滤不可交易日期预测概率长期集中在 0.5 附近特征本身区分度不高或任务难度大寻找更有效的因子考虑使用横截面标签训练集和测试集分布差异明显市场发生结构变化使用滚动训练减小单次训练窗口下面展开两个最典型的问题。6.1 模型“准确率”很高但实际没用如果你的分类模型在历史数据上准确率接近 100%大概率不是发现了圣杯而是发生了数据泄漏。最常见的一种泄漏是假设我们要用 T 日收盘后的数据预测 T 日收益但实际代码里用了 T 日的最终收益去算特征。这等于答案已经写进题目里了。排查方法把特征和标签按时间线画出来确认特征总是早于标签。检查所有shift()参数的位置。对任意一行样本确认它用到的所有历史窗口都不包含未来数据。6.2 回测盈利、实盘亏损这是量化投资里最经典的落差原因集中在四点回测没算交易成本。回测假设可以按收盘价无限成交。策略在小资金量下没问题但资金量变大后冲击成本上升。策略在历史数据上被反复调参产生了过拟合。解决方向不是追求更复杂的模型而是先从回测引擎下手把成交限制、成本模型、滑点、涨跌停、停牌全部加上让回测结果更接近真实。7. 从 Two Sigma 的访谈中总结的最佳实践Two Sigma 的公开分享虽然讲的是投资场景但背后的工程思想完全可以迁移到其他 AI 应用领域。这里总结五条我认为最值得记住的建议。7.1 数据质量 模型复杂度很多 AI 项目失败不是模型不够先进而是数据质量不过关。字段缺失、口径不一致、时间对齐错误任何一个问题都会让模型学到错误的模式。建议建立数据管道的自动校验机制比如检查数据是否有重复日期检查收益率是否超出合理范围检查数据缺失比例是否超过阈值记录每个数据文件的生成时间和版本号7.2 模型要上线先设计监控很多团队花三个月训练模型却只花一个下午做上线这是反过来的。模型监控不是上线后的事而是在设计阶段就要考虑的系统能力。对于投资场景至少要监控模型的预测概率分布是否发生漂移特征分布是否发生显著变化策略收益是否偏离预期区间实际成交价格与模型假设的成交价格差异7.3 可解释性不是非要线性模型Two Sigma 并不是要求所有模型都可解释而是强调“风险可解释”。也就是说我们不需要完全理解模型内部每一个参数但需要知道模型在哪些情况下容易失效模型的预测是否过度集中在某些行业模型是否在利用某个已经失效的历史规律这可以通过特征重要性分析、Shapley 值分析、行业暴露分析等方法实现。7.4 工程平台能力决定研究效率Two Sigma 的内部研究平台本质上是一个数据、计算、回测一体化的研发环境。研究员可以在平台上快速完成“提出假设 - 找数据 - 算特征 - 跑回测 - 看结果”的闭环。对中小团队来说即使没有自研平台也可以通过以下组合搭建类似能力使用统一的数据仓库存储原始数据和特征数据使用 Jupyter Notebook 做研究原型验证使用 Airflow 调度每日数据更新和特征计算使用 MLflow 记录每次实验的模型参数和指标关键是让每一次实验都可复现、可比较、可追溯。7.5 守好风险管理边界在金融场景里AI 模型输出的是“建议”而不是“最终决策”。风险约束应该在模型之外单独存在比如单只股票持仓上限单一行业暴露上限最大回撤触发降仓机制极端行情下的熔断开关这种“模型做预测、规则做约束”的架构即使模型某个阶段失效系统整体也不会失控。这个思路对其他 AI 应用同样适用比如风控模型、推荐系统的兜底策略。8. 总结与学习路线这篇文章从 Two Sigma 的公开访谈和工程分享出发梳理了 AI 赋能投资的完整链路数据层、特征层、模型层、回测层、风险监控层。核心结论是AI 在投资里不是玄学而是一套系统工程数据质量、特征设计、回测严谨性和风险约束往往比模型算法本身更决定成败。文中给出了一套最小可运行的 Python 选股流水线包含合成数据生成、特征计算、随机森林模型训练和简版回测。你可以把这份代码作为起点逐步加入以下内容继续练习使用真实历史行情数据替换合成数据。增加更多因子如成交量因子、基本面因子、行业因子。用 LightGBM 替换随机森林对比模型效果。在回测中加入手续费、滑点、涨跌停限制。用滚动训练替代一次性切分观察模型稳定性。添加风险监控模块模拟极端行情下的组合表现。如果遇到这篇文章里没有覆盖的报错或问题欢迎在评论区把报错信息和你的代码结构发出来一起讨论。也可以先收藏本文需要的时候直接对照流程排查。把这些关键步骤亲手跑通一遍你才会真正理解“AI 赋能投资”这几个字背后的工程分量。