从零搭建AI量化信号引擎：特征、训练、回测与风控

AI对冲基金在近期的舆论中被推到了风口浪尖。模型跑得很快，收益曲线很诱人，但一旦遇到极端行情，高杠杆和黑盒策略会把前期盈利全部吐回去，甚至引来监管关注。对一个做量化系统的人而言，这件事最重要的启示不是“AI能不能赚钱”，而是：当系统用机器学习模型做交易决策时，工程上怎样保证它可解释、可回测、可审计、可恢复。本文会从零搭建一个最小可运行的AI量化信号引擎，包含特征计算、模型训练、在线推理、回测验证、风控校验和审计日志。读完你会得到一套能用于个人学习、也能引导生产系统设计的完整链路。

1. 为什么AI对冲基金需要先把“稳健性”作为第一目标

1.1 AI在交易系统中的真实位置：信号生成器，不是交易引擎

很多人把“AI量化交易”理解成：AI模型直接决定买卖，甚至自动下单。真实系统里，模型只是交易链路中的一个环节。它的作用是接收特征输入，输出一个预测或信号，比如未来一段时间价格会上涨的概率、波动率会升高多少、某个因子是否有效。真正的交易执行还需要订单管理、持仓计算、风控校验、成交回报、账务处理等一整套工程模块。

把AI模型当成交易引擎，是很多小团队早期容易犯的错误。模型训练准确率高，往往只代表历史样本内拟合得好，不代表实盘能赚钱。一个看起来“下一秒涨”的预测，如果成交价已经因为网络延迟偏离了几十个基点，这笔预测可能根本没有执行价值。如果模型只输出交易方向，而没有仓位和止损，一次预测错误就可能造成巨大回撤。

所以，AI在交易系统中的合理位置是“信号生成器”。它负责给出有信息量的判断，但最终下单前还必须经过仓位管理、风险校验、成本判断和人工或规则兜底。把AI定位成引擎的一部分，而不是全部，是后续所有工程设计的起点。

1.2 监管关注点：透明度、杠杆与极端风险

近期的监管审查让AI量化系统不得不重新审视自己的工程化程度。证券监管机构（比如美国SEC）关注自动交易系统，通常不是因为“AI模型本身违法”，而是担心几个现实问题：策略是否对投资者充分披露；高杠杆下是否隐藏了极端回撤风险；系统故障或模型误判时能否追溯责任；日志能否还原当时的决策过程。

对开发者来说，监管审查最直接的落点是“可解释性”和“可审计性”。可解释性不是要求你给每个预测都写一段人话，而是要求关键决策能够有依据，比如用了哪些特征、模型版本是什么、置信度是多少、为什么超过阈值才下单。可审计性要求系统能回答“某一天某笔单子是谁、哪个模块、哪个模型版本、在什么参数下发出的”。

这些问题如果等到出现问题才去补，通常已经来不及。模型推理服务日志没有存，特征数据过期被覆盖，风控规则在代码分支中无法确认是否生效。这些工程欠账，在普通业务系统里可能只是影响排障，在量化系统里就会变成合规风险。

1.3 一套可对抗风险的工程骨架：数据、模型、风控、审计

基于上面的判断，一个合格AI量化系统至少需要六层模块：

1. 数据接入层：负责行情、财务、舆情等数据的采集和清洗。
2. 特征计算层：把原始数据转换成模型需要的特征，并保证训练和推理口径一致。
3. 模型推理层：加载模型，输出预测结果、置信度和辅助解释信息。
4. 交易信号层：把模型输出转换成目标仓位、限价区间和止损条件。
5. 风控校验层：在下单前检查仓位、价格、收益回撤、频率等风险指标。
6. 审计日志层：记录模型版本、输入特征摘要、信号、订单、风控结果和配置变更。

本文后续就用这六层做一个最小实现。学习环境下可以单机运行，生产环境只需要把每一层替换成独立服务并增加消息队列和分布式存储。

2. 系统设计与技术选型：先定义数据流，再写模型

2.1 模块划分与数据流

开始写代码前，最好先把数据流画在纸上。整个链路从数据源开始，经过清洗、特征、模型、信号、风控，最后到达模拟或真实交易网关。每一步都可能产生日志，日志还要回到统一的审计存储。

最小系统的数据流可以描述为：

行情数据源(SQLite/CSV) -> 数据清洗模块 -> 特征数据集 -> 模型训练流程(MLflow注册) -> 模型推理服务(FastAPI) -> 信号生成模块 -> 风控校验模块 -> 模拟交易账户/纸面交易 -> 审计日志(PostgreSQL/JSON文件)

这里把模型训练与在线推理分开，是刻意为之。训练流程可以离线和批处理运行，推理服务必须满足低延迟和稳定返回。生产系统中这两个模块通常会部署在不同的容器或服务里，避免训练任务占用CPU或内存影响到实盘预测。

2.2 技术栈选择与版本建议

下面是一组适合个人学习和中小团队验证方案的组合：

这个组合不是唯一答案。如果你更熟悉Java，可以使用Spring AI做推理客户端，但模型训练和特征计算仍然建议放在Python侧。不要让技术栈分散，先跑通最小闭环是最重要的。

2.3 目录结构与配置外置化

一个可维护的量化项目，目录结构要让人一眼看懂递归关系。下面是一个标准目录：

ai-trading-system/ ├── app/ │ ├── data/ │ │ ├── loader.py │ │ └── cleaner.py │ ├── features/ │ │ └── engine.py │ ├── models/ │ │ ├── trainer.py │ │ └── inference.py │ ├── risk/ │ │ └── checker.py │ ├── audit/ │ │ └── logger.py │ ├── api/ │ │ └── predict.py │ ├── config.py │ └── main.py ├── tests/ ├── scripts/ ├── data/ ├── .env └── requirements.txt

配置外置化是生产环境的第一道基础。不要在代码里写死数据库地址或者第三方服务Token。使用pydantic-settings可以在启动时自动读取.env文件，并对缺失配置直接报错。

from pydantic_settings import BaseSettings class Settings(BaseSettings): model_path: str = "models/lgbm.txt" db_url: str = "postgresql://user:pass@localhost:5432/trading" redis_url: str = "redis://localhost:6379/0" log_dir: str = "logs" max_position_pct: float = 0.2 max_drawdown_pct: float = 0.1 api_key_env: str = "TRADING_API_KEY" class Config: env_file = ".env" settings = Settings()

这里的重点不是代码本身，而是约束：所有环境差异都必须通过配置体现，训练环境和实盘环境使用同一个配置模板，只是不同值。否则经常会出现“训练时用了A数据源，实盘接入了B数据源，特征口径不同步”的故障。

3. 实现AI信号引擎：特征、训练、预测与仓位计算

3.1 特征工程：避免未来函数

特征工程决定模型效果，但比效果更重要的是“特征是否在未来函数”。未来函数指的是在t时刻做预测时，使用了t时刻之后才能知道的数据。典型错误包括：用当日收盘后发布的财务数据预测当天盘中价格，用未来一段时间的收益率做标签时没有对齐时间窗口。

下面代码是一个常见的最小特征集合：

import pandas as pd import numpy as np def compute_features(df: pd.DataFrame) -> pd.DataFrame: data = df.copy() data["return_1"] = data["close"].pct_change(1) data["return_5"] = data["close"].pct_change(5) data["volatility_10"] = data["return_1"].rolling(10).std() data["volume_ratio"] = data["volume"] / data["volume"].rolling(20).mean() data["high_low_ratio"] = data["high"] / data["low"] - 1 data = data.dropna().reset_index(drop=True) return data

每一个特征都要能回答三个问题：这个时刻是否已知；是否经过滚动计算；缺失值如何填充。这里的pct_change(5)当天收盘后是已知的，但如果你用它预测当天“未来1天收益”，就要确保预测的输入发生在标签之前，否则就是数据泄漏。

3.2 训练模型并使用MLflow注册

训练流程的代码看起来简单，但生产环境必须加三样东西：数据切分、模型标准化、模型注册。下面是一个基于LightGBM的示例：

import lightgbm as lgb import mlflow import mlflow.lightgbm from sklearn.model_selection import train_test_split from sklearn.metrics import accuracy_score, roc_auc_score X = features[["return_1", "return_5", "volatility_10", "volume_ratio", "high_low_ratio"]] y = (features["future_return_5"] > 0).astype(int) X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, shuffle=False ) with mlflow.start_run(): model = lgb.LGBMClassifier( n_estimators=200, learning_rate=0.05, num_leaves=15, max_depth=5, ) model.fit( X_train, y_train, eval_set=[(X_test, y_test)], callbacks=[lgb.early_stopping(20, verbose=False)] ) auc = roc_auc_score(y_test, model.predict_proba(X_test)[:, 1]) mlflow.log_metric("test_auc", float(auc)) mlflow.lightgbm.log_model(model, artifact_path="model") mlflow.log_param("n_estimators", 200)

注意这里shuffle=False。时间序列数据不能随机打乱切分，否则模型会“偷看未来”。如果你用严格时间序列预测，还要考虑时序交叉验证，而不是普通K折。

3.3 信号生成与仓位计算

模型输出是一个概率值，不能直接把概率大于0.5当成买入信号。概率高只能说明“方向判断更有把握”，还要结合波动率决定仓位。一个常见的仓位计算方式是风险平价：预测上涨概率越高、当前波动率越低，就分配越高的仓位；反过来，波动率升高时减仓。

def target_position(prob_up: float, vol_10: float) -> float: if vol_10 is None or vol_10 <= 0: return 0.0 strength = (prob_up - 0.5) * 2 # 从0到1，表示方向置信度 vol_scaled = 0.05 / max(vol_10, 0.001) raw_target = strength * vol_scaled return float(np.clip(raw_target, 0.0, 1.0)) def make_signal(prob_up: float, vol_10: float, current_price: float) -> dict: pos = target_position(prob_up, vol_10) return { "direction": "long" if prob_up > 0.5 else "short", "target_weight": pos, "expected_fill_price": current_price, "stop_loss_price": current_price * 0.95 if prob_up > 0.5 else current_price * 1.05, "confidence": round(abs(prob_up - 0.5) * 2, 4), }

这里的target_position只是说明思路，实际项目还要考虑账户权益、最大持仓限制和交易成本。仓位计算的目标不是“抓住所有机会”，而是“让自己在判断错误时仍然可以继续交易”。

3.4 风控前置校验

信号生成之后，必须经过风控模块校验才能提交到交易网关。风控规则要放在最前端，不要在订单到达交易所时才判断。常见的检查项包括：

class RiskChecker: def __init__(self, max_position: float, max_daily_loss: float): self.max_position = max_position self.max_daily_loss = max_daily_loss def check(self, signal: dict, account_state: dict) -> tuple[bool, str]: if signal["target_weight"] > self.max_position: return False, "position_limit_exceeded" if account_state["daily_loss"] > self.max_daily_loss: return False, "daily_loss_limit_hit" if signal["expected_fill_price"] <= 0: return False, "invalid_price" return True, "ok"

风控规则最好由配置管理，不要写死在代码里。因为参数调整可能很频繁，每次改代码再发布，既慢又容易出错；通过配置中心或.env修改风控阈值后，至少要有告警和日志记录。

4. 在线推理与回测：验证“训练-实盘”一致性

4.1 回测框架的最小实现

回测不是把历史数据送给模型，让模型输出信号就行。回测还要模拟成交价格、手续费、滑点和资金变化。下面是一个极简逐bar回测循环：

def run_backtest(features, prices, model, initial_cash=100_000): cash = initial_cash position = 0.0 commission_rate = 0.0005 for i in range(len(features)): row = features.iloc[i:i + 1] prob_up = model.predict_proba(row)[:, 1][0] signal = make_signal(prob_up, row["volatility_10"].iloc[0], prices[i]) target_pos = signal["target_weight"] trade_value = (target_pos - position) * prices[i] * cash commission = abs(trade_value) * commission_rate cash -= commission if target_pos > position: buy_value = cash * (target_pos - position) cash -= buy_value position += (buy_value - commission) / prices[i] else: sell_value = position * prices[i] * (position - target_pos) cash += sell_value - commission position -= (position - target_pos) final_equity = cash + position * prices[-1] return final_equity

这个回测很粗糙，它假设每次都能按收盘价成交，没有考虑限价单的未成交和滑点。真实策略至少还要加入盘中价格序列、成交判定和风控触发逻辑。回测的意义不是证明“策略很赚钱”，而是确认“训练和推理链路没有明显bug”。

4.2 FastAPI在线推理服务

把训练好的模型封装成API，是连接策略与交易系统的核心。下面是一个最小推理服务：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel import joblib app = FastAPI() model = joblib.load("models/lgbm.joblib") class FeaturePayload(BaseModel): return_1: float return_5: float volatility_10: float volume_ratio: float high_low_ratio: float class SignalResponse(BaseModel): direction: str target_weight: float confidence: float model_version: str @app.post("/predict", response_model=SignalResponse) def predict(payload: FeaturePayload): import pandas as pd df = pd.DataFrame([payload.dict()]) prob_up = model.predict_proba(df)[:, 1][0] signal = make_signal(prob_up, payload.volatility_10, 0.0) return SignalResponse( direction=signal["direction"], target_weight=signal["target_weight"], confidence=signal["confidence"], model_version="v1.0.0", )

生产环境中，model_version不能写死。它应该来自模型注册表，并且每次推理都要把版本记录到日志。否则模型已经更新到v2，线上还在用v1，回测报告和实盘结果对不上，问题极难排查。

4.3 如何保证训练和在线推理使用同一特征口径

特征不一致是量化系统最常见的bug。训练时用pct_change(5)，在线推理时因为数据长度不够，可能在第一行产生NaN，然后填充方式不同，导致预测结果完全不同。

解决办法是把特征计算逻辑抽成一个公共模块，训练脚本和推理服务都引用同一个函数。上面的compute_features函数可以放到app/features/engine.py，训练时读原始数据后调用它，在线推理时也通过同一段代码计算并行特征。

此外，在线服务需要做数据完整性校验。例如，如果volatility_10为NaN，不能直接丢给模型，应该返回错误码并记录日志。这比默默预测一个异常值安全得多。

5. 审计日志、监控与监管合规检查清单

5.1 为什么审计日志比盈利曲线更重要

盈利曲线只展示结果，审计日志才能解释结果是怎么来的。监管机构追问一笔亏损交易时，系统必须能回答：在什么时间、基于什么特征、使用什么模型版本、生成了什么信号、风控是否通过、订单是否发送、成交价格是多少。如果这些信息残缺，系统再赚钱也无法通过审查。

从工程角度看，审计日志对排查问题同样关键。模型发生异常预测、订单重复发送、风控规则失效，都需要日志定位责任模块。没有日志的系统，等于每次故障都从零开始猜测。

5.2 需要记录的事件与数据结构

最小系统至少需要四类日志：

日志建议使用JSON格式，便于采集到Elasticsearch或ClickHouse中。下面是一个推理日志示例：

{ "event_type": "inference", "timestamp": "2025-01-15T10:30:00.123Z", "model_version": "v1.0.0", "prob_up": 0.68, "feature_hash": "a3f9c2d5e8b1b0f8", "request_id": "req_123456" }

feature_hash是用来快速定位请求的，即使不保存完整特征，也可以通过hash判断输入是否有异常变化。生产环境建议保存一份特征快照，防止依赖的原始行情已删除。

5.3 监管合规检查清单

下面是一份面向AI交易系统的上线检查清单，可以直接作为发布门禁：

这里的数据保留年限是参考常见金融业务要求，具体周期要根据实际平台和监管规定确认。不要自行承诺或省略该环节。

5.4 面对监管质询时的准备动作

如果系统需要向监管提交材料，应该提前准备好：模型说明文档、训练数据描述、风控参数与变更记录、特定时间段的日志导出工具、复盘报告模板。平时可以每季度做一次“模拟质询”，随机挑选一天，要求负责人回答当天所有交易信号为什么被生成，风控为什么放行或拒绝。这个过程能发现大部分日志盲区。

6. 常见问题与排查路径

6.1 预测结果与回测差异巨大

现象：回测里同一段历史数据预测准确率很高，但实盘或模拟盘预测结果明显偏离。

先按顺序排查：

1. 输入特征是否一致，包括字段名、方向、单位。
2. 是否使用同一版本模型。
3. 数据是否存在前视偏差，比如特征计算用到了未来价格。
4. 在线推理是否对NaN做了不同于训练时的填充。
5. 模型是否被并发请求踩到了全局变量。

最隐蔽的原因是训练时用整个数据集做特征标准化，推理时却用新的均值和方差。处理办法是把状态化特征算子（标准化、滚动均值）固化保存，并在推理前加载。

6.2 风控未生效导致超仓

现象：一笔交易目标仓位超过风控阈值，仍然下单成功。

可能原因包括：风控逻辑只写在模型推理端，没有在下单端重复校验；或者信号和订单是两个服务，竞态条件下风控通过后仓位已经变化。解决方式是采用“事务性风控”，订单发送前再次获取最新账户状态，并且对订单执行幂等控制。任何绕过风控的旁路下单接口，都要在网关层直接拦截。

6.3 日志缺失导致无法回溯

现象：某笔异常交易发生后，查询日志发现推理日志和订单日志的request_id对不上。

常见原因是日志由不同服务分别打印，没有统一链路追踪。解决方式是为每个信号生成一个全局唯一的signal_id，在推理、信号、风控、订单全链路传递。日志写入失败时不能静默忽略，至少要有错误计数和告警。

6.4 数据偏移导致模型失效

现象：模型上线初期表现不错，几周后预测准确率下降，回撤变大。

监控特征分布是最直接的预警方式。可以每天保存特征均值、方差和分位数，如果某一天特征分布与训练集偏移超过阈值，就触发重训练提醒。对于LightGBM这类树模型，还可以监控特征重要度和叶子输出分布的变化。不要等到回撤发生再重训练，而要在指标劣化前切换模型。

7. 从学习环境到生产环境：最佳实践与扩展方向

7.1 上线前最佳实践清单

经过完整的开发、回测、验证后，如果要把系统部署到生产环境，至少应该遵守下面这些实践：

• 模型文件打入带版本号的镜像，不在容器启动时从外部下载模型。
• 推理服务无状态化，水平扩展后不会出现状态冲突。
• 每个请求都生成唯一ID，并贯穿日志、监控和订单链路。
• 风控参数从配置中心读取，参数变更必须生成审计记录。
• 所有的交易接口都做幂等控制，重复请求不会导致重复下单。
• 初始上线时使用纸面交易账号，观察至少一个完整交易周期再切换小资金。
• 数据源不可用时，推理服务应该返回错误而非默认真实值。
• 定期执行故障演练：模拟模型不输出、数据库宕机、交易所连接断开，验证系统能否安全降级。
• 代码和模型分开版本控制，训练代码变化后重新回测，不能只看预测精度。
• 每次部署都准备回滚方案，包括模型文件、配置和数据库迁移脚本。

7.2 与大模型和AI Agent结合的扩展方向

大语言模型在量化系统中的应用正在增加，但不要直接把大模型输出接在下单链路上。比较稳妥的方向有两个：一个是把大模型用于舆情分析、财报解读和策略解释，将分析结果作为特征输入到传统模型；另一个是让AI Agent承担“监控和分析”职能，比如读取告警、生成复盘报告、给出参数调整建议，但所有变更仍然需要人工确认。

这里可以借鉴AI Agent的工程原则：Agent可以提方案，不能直接执行外部影响性动作。交易系统里，下单是最高风险动作，必须由确定性风控模块把关。大模型和Agent适合做辅助层，不负责最终资金操作。

7.3 最终建议

AI对冲基金的技术重点从来不是训练一个更复杂的模型。相反，先把工程链路做扎实，让每一次预测、每一笔订单、每一次风控触发都有据可查，比短期收益提升更重要。从本文的最小系统出发，后续可以逐步加入更多数据源、多因子模型、组合优化和实时计算引擎。但无论扩展到多复杂，都要保留那条核心基线：模型输出只是信号，风险校验必须前置，审计日志必须完整。只要这三条不倒，系统遇到极端行情时至少能说清楚发生了什么，也能给出下一步修复方向。