个人 AI 量化这件事真正难的不是训练一个准确率更高的模型而是把行情数据、特征工程、模型推理、回测验证和交易执行组装成一条可以反复修改、排查和跟踪的流水线。Min 这个名字看起来是一个“Personal AI Quant”项目本质上它解决的是个人开发者做量化研究时最常见的痛点单点代码都有但系统整体不可复现、信号不可解释、结果不可验证。这篇博客会围绕 Min 的设计思路拆解一个个人 AI 量化助手从环境搭建、数据层、特征工程、模型信号到回测模拟盘的完整工程实现路径。读完以后你可以按同样的结构搭建一套属于自己研究用的量化实验框架重点理解每一层为什么这样设计以及实际落地时会踩到哪些数据和时间序列相关的坑。1. 先理解 Min 这类 Personal AI Quant 的定位和核心链路1.1 个人量化为什么需要 AI 而不是简单规则传统量化策略通常由一组可编程规则组成例如均线金叉、布林带突破、RSI 超卖反弹。这类策略的逻辑清晰、容易解释但缺点也很明显规则的阈值需要人工反复调整市场状态切换后旧规则容易失效而且多个规则组合在一起时权重很难人工优化。AI 量化把“规则生成”和“阈值选择”交给模型完成。模型从历史数据中学习特征与未来收益之间的关系输出一个信号或概率。这样做的好处是特征的组合方式可以更复杂坏处是模型成了黑箱出问题时很难定位是数据问题、特征问题还是模型参数问题。Min 这类工具的价值不是让你放弃规则而是用工程手段把“数据 - 特征 - 模型 - 信号 - 执行 - 反馈”这条链路标准化让 AI 信号可以被回测、被对比、被解释而不是成为一个无法验证的预测函数。1.2 Min 的模块划分数据、信号、执行、评估一个可落地的个人 AI 量化系统至少要有四个独立模块这也是 Min 的架构主线。模块职责关键输出常见风险数据层获取行情、财务、宏观等原始数据统一清洗和存储标准化 DataFrame、数据库表时区错乱、未来函数、复权不一致特征层将原始数据转换成模型可用的特征矩阵特征 DataFrame、特征血缘记录特征泄露、缺失值处理不一致信号层使用规则或模型生成交易信号信号序列、预测概率、置信度过拟合、信号抖动、不可解释执行层根据信号计算目标仓位对接模拟盘或实盘网关订单、持仓、成交记录滑点估计不足、风控缺失、接口异常评估层回测、绩效分析、归因和监控回测报告、绩效指标、告警幸存者偏差、前视偏差、指标口径错误这五个模块里最容易出问题的是数据层和评估层。模型选型反而相对简单因为个人量化初期用随机森林、XGBoost 这类可解释性较好的模型比一上来就上深度神经网络更合适。1.3 学习环境、研究环境和生产环境的边界个人量化项目刚起步时一定要区分三种环境。学习环境以跑通流程为主可以用日线数据、单股票、模拟盘代码可以写得粗糙目标是理解整条链路。研究环境要求数据可追溯、特征可复现、回测结果可对比这一阶段必须引入配置文件和实验记录。生产环境则需要考虑调度、日志、监控、异常恢复、权限和回滚甚至需要独立的数据库和服务器。很多个人开发者直接在本地写一个 Jupyter Notebook 完成所有步骤这对研究没问题但一旦要把策略跑起来就必须把代码重构成模块化项目。Min 的定位更接近“个人 AI 量化助手”它应该属于研究环境向生产环境过渡的形态所以项目结构从一开始就要为后续扩展留出空间。2. 环境、依赖与项目结构先搭出可复现的研究骨架2.1 Python 环境和依赖版本对齐个人量化项目以 Python 为主建议使用 Python 3.10 或 3.11。低于 3.9 会有类型注解兼容问题高于 3.12 部分数据分析库的二进制包可能还没完全适配。实际使用时需要根据自己的系统确认版本。下面是 Min 这类项目常见依赖清单这里只给出用途落地时以官方发布的最新稳定版本为准。依赖库用途需要注意的点pandas数据处理、时间序列处理版本差异会影响时间序列操作行为numpy数值计算不要用旧接口例如np.float已废弃scikit-learn分类模型、特征选择、交叉验证模型 API 稳定适合学习环境yfinance获取公开行情数据网络不稳定、接口限流需要缓存sqlalchemy数据库访问抽象用连接池管理数据库连接fastapi提供研究和监控接口适合把信号服务化apscheduler定时任务调度避免多个调度器重复触发matplotlib可视化回测结果和特征分布中文显示需要额外配置安装依赖时建议使用虚拟环境不推荐直接装在系统 Python 里。项目根目录下执行python -m venv .venv source .venv/bin/activate pip install -r requirements.txt这里使用虚拟环境的目的是让依赖版本可复现。如果换了机器或者过段时间回来继续开发只需要重新安装同一份requirements.txt不会因为系统环境里其他包版本冲突导致行为不一致。2.2 目录结构按数据边界分模块不按功能类型分目录很多入门项目会按utils、models、strategies这种“文件类型”组织目录但量化项目更适合按“数据边界”组织因为数据依赖关系决定了整个系统会不会被改乱。一个可参考的结构如下min-quant/ ├── config/ │ ├── settings.yaml │ └── symbols.txt ├── data/ │ ├── raw/ │ └── processed/ ├── min_quant/ │ ├── datasource/ │ │ ├── base.py │ │ ├── yahoo.py │ │ └── cache.py │ ├── features/ │ │ ├── builder.py │ │ └── registry.py │ ├── models/ │ │ ├── trainer.py │ │ └── inference.py │ ├── strategy/ │ │ ├── rule_based.py │ │ └── ml_signal.py │ ├── backtest/ │ │ ├── engine.py │ │ └── metrics.py │ ├── broker/ │ │ ├── base.py │ │ ├── paper.py │ │ └── risk.py │ └── pipeline.py ├── scripts/ │ ├── download_data.py │ ├── train_model.py │ ├── run_backtest.py │ └── run_paper_trading.py ├── tests/ └── requirements.txt直接按上面的名字照搬不是必须的但有几个原则值得保留第一datasource、features、strategy、broker按数据流切分避免模型代码直接调用外部接口。第二config和data放在顶层因为配置和原始数据是实验可复现的基础。第三scripts里只放可执行入口不要把调度逻辑写进库代码。这样的结构在后续扩展多数据源、多策略时改动范围会小很多。2.3 配置文件把变量从代码里抽出来不要把股票代码、数据库地址、模型参数写死在 Python 代码里。用 YAML 配置文件统一管理。data: source: yahoo symbols: - AAPL - MSFT start_date: 2020-01-01 end_date: 2025-01-01 interval: 1d cache_dir: data/raw features: window_short: 5 window_long: 20 volatility_window: 10 model: name: random_forest params: n_estimators: 200 max_depth: 4 min_samples_leaf: 20 backtest: initial_cash: 100000 commission_rate: 0.001 slippage_rate: 0.0005 start_date: 2023-01-01 end_date: 2025-01-01 risk: max_position_pct: 0.2 max_daily_loss_pct: 0.02配置文件的读取可以用pyyaml或pydantic。如果项目较大推荐pydantic-settings因为它能在加载配置时做类型校验配置写错时能提前暴露问题。这里要注意数据的时间范围、回测的时间范围和模型训练的时间范围应该分开配置。很多人把训练集和回测集混在一起导致模型在回测阶段看到了训练阶段的数据结果虚高。3. 数据层实现行情、财务和特征数据的统一管理3.1 数据源抽象不要把外部接口绑死在业务代码里数据源是整个量化系统最容易变的部分。今天用 Yahoo明天可能换成其他公开数据源后天可能接本地数据库。如果业务代码直接调用yfinance.download换数据源时就要改所有调用点。更合理的做法是定义一个统一的数据源接口所有数据获取都走这个接口。# min_quant/datasource/base.py from abc import ABC, abstractmethod import pandas as pd class DataSource(ABC): abstractmethod def load_price_data( self, symbol: str, start: str, end: str, interval: str 1d, ) - pd.DataFrame: 返回统一的 OHLCV 数据。 class CachedDataSource(DataSource): 带文件缓存的包装器避免每次运行都重新下载数据。 def __init__(self, source: DataSource, cache_dir: str): self._source source self._cache_dir cache_dir def load_price_data(self, symbol, start, end, interval1d): cache_path f{self._cache_dir}/{symbol}_{start}_{end}_{interval}.parquet try: return pd.read_parquet(cache_path) except FileNotFoundError: df self._source.load_price_data(symbol, start, end, interval) df.to_parquet(cache_path) return df返回的 DataFrame 应该有统一字段open、high、low、close、volume索引是带时区信息的DatetimeIndex。这样下游特征工程不需要关心数据来自哪个数据源。使用缓存的意义在于模型训练和回测会反复读取历史数据每次重新下载既慢又受限于接口频率。第一次下载后落到本地parquet后续运行直接读取速度快也方便多次实验对比。3.2 用 yfinance 实现一个具体数据源以公开数据源为例实现YahooDataSource# min_quant/datasource/yahoo.py import yfinance as yf import pandas as pd from min_quant.datasource.base import DataSource class YahooDataSource(DataSource): def load_price_data( self, symbol: str, start: str, end: str, interval: str 1d, ) - pd.DataFrame: df yf.download( symbol, startstart, endend, intervalinterval, auto_adjustTrue, progressFalse, ) if df.empty: raise ValueError(f{symbol} no data in {start} ~ {end}) df df.reset_index() df.columns [col.lower() for col in df.columns] # yfinance 多级列时保留 close/open/high/low/volume rename_map {adj close: close} df df.rename(columnsrename_map) required [open, high, low, close, volume] for col in required: if col not in df.columns: raise ValueError(fmissing column: {col}) df df.set_index(date) df.index pd.to_datetime(df.index, utcTrue) return df[required].sort_index()这个实现有几个关键点。auto_adjustTrue很重要它表示自动复权能让历史价格反映分红拆股影响。如果忽略复权回测里会出现价格跳空导致收益计算失真。数据索引使用datetime并转换为 UTC是为了避免不同市场时区混在一起。之后做日线聚合时再按本地时区转换。注意yfinance不是官方量化数据接口稳定性无法保障。个人研究可以用但如果要跑实盘策略需要找稳定、合规的数据供应商同时注意使用条款。3.3 数据质量检查和时区处理数据下载完成后不能直接进入特征工程。至少要检查这几项数据是否为空是否有重复索引是否有缺失值或 NaN时间索引是否严格递增是否有价格小于等于 0 的异常记录复权价格是否和原始价格差异过大一个简单的校验函数# min_quant/datasource/validation.py import pandas as pd def validate_price_data(df: pd.DataFrame, symbol: str) - None: if df.empty: raise ValueError(f{symbol} price data is empty) if not df.index.is_monotonic_increasing: raise ValueError(f{symbol} index is not monotonic) if df.index.has_duplicates: raise ValueError(f{symbol} index has duplicates) numeric_cols [open, high, low, close, volume] if df[numeric_cols].isna().any().any(): raise ValueError(f{symbol} contains NaN in price data) if (df[low] 0).any() or (df[close] 0).any(): raise ValueError(f{symbol} contains non-positive price)这里要注意真实数据偶尔会出现停牌日缺失、个别字段为 NaN 的情况。直接抛异常是最安全的策略因为下游特征计算遇到未知缺失值时错误会被隐藏直到最终结果异常才暴露。更好的做法是在数据下载脚本里先记录日志再决定是补全还是丢弃。注意任何数据校验逻辑都要先于特征计算执行。数据层的错误如果被放过后面看到的往往是“模型效果异常”而不是“数据缺失”排查成本会高很多。4. 特征工程与 AI 信号生成这是 Min 的判断核心4.1 特征设计先做能解释的特征再谈复杂模型对于个人量化项目特征不必一开始就堆很多。先把技术指标、波动率、成交量和收益分布这几类基础特征做出来让模型有足够信息可用再逐步增加外部数据。下面是一些适合个人日线策略的初始特征。特征名称计算方式含义注意事项ret_1当日收益率短期动量计算时要 shift避免用到当日未来信息ma_5/ma_20移动平均趋势方向窗口长度要和持有周期匹配rsi_14相对强弱指标超买超卖状态极端值可能意味着趋势延续而非反转volatility_1010 日收益率标准差波动风险波动率剧烈变化常伴随流动性风险volume_ratio当日成交量 / 过去均值放量特征放量需要结合涨跌方向理解close_position收盘价在区间中的位置日内强弱计算公式为 (close - low) / (high - low)特征计算的关键是不要引入未来函数。例如计算ma_20时第t行的特征只能使用t及之前的数据不能使用t1及之后的数据。pandas 的rolling默认包含当前值但如果你要表示“截至昨天”的信号就必须把窗口整体 shift。4.2 特征工程实现示例下面代码演示如何基于统一 OHLCV 数据生成特征矩阵。# min_quant/features/builder.py import numpy as np import pandas as pd def compute_features(df: pd.DataFrame) - pd.DataFrame: out pd.DataFrame(indexdf.index) close df[close] out[ret_1] close.pct_change() out[ma_5] close.rolling(5).mean() out[ma_20] close.rolling(20).mean() out[ma_ratio] out[ma_5] / out[ma_20] - 1 delta close.diff() up delta.clip(lower0) down -delta.clip(upper0) roll_up up.rolling(14).mean() roll_down down.rolling(14).mean() rs roll_up / roll_down.replace(0, np.nan) out[rsi_14] 100 - (100 / (1 rs)) out[volatility_10] close.pct_change().rolling(10).std() volume_mean df[volume].rolling(20).mean() out[volume_ratio] df[volume] / volume_mean.replace(0, np.nan) high df[high] low df[low] range_ (high - low).replace(0, np.nan) out[close_position] (close - low) / range_ out out.replace([np.inf, -np.inf], np.nan) return out特征生成之后一定要做一次对齐检查。比如ma_20前 19 行是 NaNrsi_14前若干行也是 NaN。在训练模型前需要删除包含 NaN 的行但要注意删除后索引是否还能对应到回测阶段的时间点。这里最容易犯的错误是训练时删除了前 20 行 NaN但回测时又从第 1 行开始推理导致模型面对未见过的 NaN 特征行为异常。正确做法是训练和推理使用完全一致的特征处理逻辑并且确保推理阶段也有足够的历史数据来计算滚动窗口。4.3 用随机森林训练一个可解释的分类模型AI 信号可以用分类模型生成。一个常见设定是根据未来 N 日收益是否大于某个阈值构造标签1/0然后让模型学习特征与标签之间的关系。标签构造是一个较敏感的设计因为标签本身代表了策略目标。例如定义未来 5 日收益大于 1% 为正样本市场环境不同时这个阈值可能完全不合适。建议先做历史收益分布分析再用中位数或分位数确定阈值。训练代码示例# min_quant/models/trainer.py import pandas as pd from sklearn.ensemble import RandomForestClassifier from sklearn.model_selection import train_test_split from sklearn.metrics import accuracy_score, precision_score def create_label(features: pd.DataFrame, close: pd.Series, horizon: int 5, threshold: float 0.01) - pd.Series: future_ret close.shift(-horizon) / close - 1 label (future_ret threshold).astype(int) label label.reindex(features.index) return label def train_model(feature_df: pd.DataFrame, label: pd.Series): data pd.concat([feature_df, label.rename(label)], axis1) data data.dropna() X data.drop(columns[label]) y data[label] X_train, X_test, y_train, y_test train_test_split( X, y, test_size0.2, shuffleFalse, stratifyNone ) model RandomForestClassifier( n_estimators200, max_depth4, min_samples_leaf20, random_state42, n_jobs-1, ) model.fit(X_train, y_train) y_pred model.predict(X_test) print(accuracy:, accuracy_score(y_test, y_pred)) print(precision:, precision_score(y_test, y_pred)) return model, X.columns.tolist()train_test_split在这种时间序列问题上使用shuffleFalse更合理因为要模拟未来数据。如果随机打乱训练集和测试集混在不同时间段里评估结果容易虚高。但shuffleFalse还不够严格因为前面样本和后面样本可能来自同一时间段的不同股票存在时间重叠。严格做法是按时间切分例如用前 80% 时间训练后 20% 时间测试。这个细节在个人项目里容易被忽略。4.4 规则信号与 AI 信号的融合AI 信号不应解释为“必须交易”而应该作为决策打分的一部分。典型的融合方式是先由规则信号过滤交易条件再由 AI 信号决定仓位大小。例如一个保守的组合逻辑# min_quant/strategy/ml_signal.py import pandas as pd def generate_signal(price_df: pd.DataFrame, feature_df: pd.DataFrame, model, feature_cols): current price_df.reindex(feature_df.index) feature_values feature_df[feature_cols].copy() prob model.predict_proba(feature_values)[:, 1] feature_df feature_df.copy() feature_df[score] prob # 规则过滤 ma_5 price_df[close].rolling(5).mean() ma_20 price_df[close].rolling(20).mean() trend_ok ma_5 ma_20 # 信号 模型概率 * 趋势过滤 feature_df[signal] feature_df[score] * trend_ok return feature_df[signal]这里的信号不是二元 0/1而是一个介于 0 和 1 之间的分数。执行层可以用这个分数来决定是否建仓、仓位多大。这样设计的好处是规则负责过滤明显不利的环境模型负责在可交易环境下给出排序两个环节职责清晰。4.5 模型可解释性AI 信号为什么不能是黑箱个人量化中模型一旦开始影响资金决策可解释性就不再是加分项而是必要项。至少要做到三件事第一能知道每个特征对预测结果的影响方向。随机森林可以直接用feature_importances_查看整体重要性但重要性只能说明预测时使用了哪些特征不能说明特征变大时预测结果往哪个方向变。更精细的做法是用 SHAP 分析单个样本的预测。第二能定位预测失效的时间段。模型往往只在部分市场状态下有效要把预测概率和真实收益放在同一张图里观察两者背离的区间。第三能对比规则策略和 AI 策略。如果 AI 模型在回测中没有明显优于简单规则就不值得承担它的复杂性和不可解释性。from sklearn.inspection import permutation_importance result permutation_importance(model, X_test, y_test, n_repeats10, random_state42) for i, col in enumerate(X_test.columns): print(f{col}: {result.importances_mean[i]:.4f} /- {result.importances_std[i]:.4f})permutation_importance通过打乱某个特征来判断它对预测结果的影响比默认的feature_importances_更直观。如果某个特征打乱后模型效果变化很大说明预测强依赖该特征。5. 回测、模拟盘与实盘网关验证信号而不是验证运气5.1 回测引擎的边界和注意事项回测的目的不是证明策略能赚钱而是验证信号的有效性、交易成本和执行方式是否合理。一个最小回测引擎可以基于逐行循环实现。# min_quant/backtest/engine.py import pandas as pd def run_backtest(signal: pd.Series, price: pd.Series, initial_cash100000.0, commission_rate0.001): cash initial_cash position 0.0 equity_curve [] current_signal 0.0 for ts, close in price.items(): sig signal.get(ts, 0.0) # 只在信号变化时交易 if sig ! current_signal: if position 0: cash position * close cash - cash * commission_rate position 0.0 if sig 0: position cash * sig / close cash - position * close cash - cash * commission_rate current_signal sig equity cash position * close equity_curve.append((ts, equity)) result pd.DataFrame(equity_curve, columns[date, equity]).set_index(date) result[return] result[equity].pct_change() return result这个引擎只用于学习不适用于真实研究。它缺少了滑点、持仓时间限制、涨跌停无法成交、停牌无法卖出等真实约束。真实回测引擎至少要考虑以下细节使用次日开盘价或收盘价成交避免用信号当日价格成交带来的前视偏差。对每笔交易加入滑点成本。考虑最小交易单位和整数股避免产生小数股。对涨跌停、停牌、流动性不足的情况做不可成交处理。注意回测引擎本身也是一个需要测试的程序。先用一个你手工能算清楚的简单场景验证回测引擎再让它计算复杂策略否则回测结果错误时你会误以为是策略问题。5.2 网关抽象统一模拟盘和实盘接口Min 这样的系统会把“交易执行”抽象成网关让上层策略代码不关心当前是模拟盘还是实盘。# min_quant/broker/base.py from abc import ABC, abstractmethod class BrokerGateway(ABC): abstractmethod def get_cash(self) - float: ... abstractmethod def get_position(self, symbol: str) - float: ... abstractmethod def submit_order(self, symbol: str, side: str, quantity: float) - str: ... abstractmethod def cancel_order(self, order_id: str) - bool: ...模拟盘实现可以很简单内部用一个 DataFrame 保存持仓和现金按收盘价成交并记录手续费。# min_quant/broker/paper.py class PaperBroker(BrokerGateway): def __init__(self, initial_cash: float, commission_rate: float): self.cash initial_cash self.positions {} self.commission_rate commission_rate self.trades [] def submit_order(self, symbol, side, quantity): price self._get_latest_price(symbol) cost quantity * price if side buy: if cost cost * self.commission_rate self.cash: raise ValueError(insufficient cash) self.cash - cost cost * self.commission_rate self.positions[symbol] self.positions.get(symbol, 0) quantity elif side sell: if self.positions.get(symbol, 0) quantity: raise ValueError(insufficient position) self.cash cost - cost * self.commission_rate self.positions[symbol] - quantity self.trades.append({symbol: symbol, side: side, quantity: quantity, price: price}) return fpaper_order_{len(self.trades)}模拟盘的目的是验证策略在接近真实环境下的执行效果。它仍然依赖_get_latest_price这个函数如何取价如果每次都取当前实时价而策略在收盘后运行价格就会对不上。建议在模拟盘里固定使用每日收盘价或次日开盘价并记录成交时使用的价格方便后续对账。5.3 风控模块仓位、止损和每日限额个人量化系统非常容易出现一次交易失误抹掉多日收益的情况。风控模块不是可选项至少要包含这几条单标的仓位上限。例如不超过总资产 20%。单日亏损上限。例如当日回撤超过 2%当天不再开新仓。信号置信度阈值。模型概率低于阈值时不开仓。最大持仓数量。避免资金分散到过多标的后难以管理。每次下单前校验可用资金和可用持仓防止重复提交订单。# min_quant/broker/risk.py class RiskManager: def __init__(self, max_position_pct0.2, max_daily_loss_pct0.02): self.max_position_pct max_position_pct self.max_daily_loss_pct max_daily_loss_pct self.initial_equity None self.day_start_equity None def check_order(self, symbol, side, quantity, estimated_price, current_equity): if self.initial_equity is None: self.initial_equity current_equity self.day_start_equity current_equity if side buy: cost quantity * estimated_price if cost current_equity * self.max_position_pct: return False, fposition too large: {cost} {current_equity * self.max_position_pct:.2f} if side sell: pass if current_equity self.day_start_equity * (1 - self.max_daily_loss_pct): return False, daily loss limit reached return True, ok风控逻辑要在模拟盘和实盘共用一套代码不要在模拟盘里关掉风控否则模拟盘跑出来的收益无法说明问题。6. 一次完整的运行流程训练、回测、观察和绩效评估6.1 数据更新和模型训练脚本整个系统通过脚本串联。先更新数据再训练模型然后回测最后进入模拟盘。# 更新原始行情数据 python scripts/download_data.py --config config/settings.yaml # 训练模型并打印特征重要性 python scripts/train_model.py --config config/settings.yaml # 运行回测 python scripts/run_backtest.py --config config/settings.yaml # 启动模拟盘定时任务 python scripts/run_paper_trading.py --config config/settings.yamldownload_data.py的核心逻辑是读取配置文件中的股票列表和时间范围调用CachedDataSource下载数据并写入data/raw。如果某只股票已经缓存且日期范围覆盖当前配置可以直接跳过下载减少重复请求。train_model.py输出模型文件和特征列清单。这里要记录训练时使用的数据版本、特征版本和参数否则后续很难复现实验结果。一个简单的做法是把训练配置和特征列名写入 JSON 文件{ model_path: models/rf_20250101.pkl, feature_cols: [ma_ratio, rsi_14, volatility_10, volume_ratio, close_position], train_start: 2020-01-01, train_end: 2024-01-01, n_estimators: 200, max_depth: 4 }这样每次实验都有迹可循。6.2 回测结果和绩效指标回测完成后的输出至少包括净值和交易记录绩效指标用一个函数集中计算。# min_quant/backtest/metrics.py import numpy as np import pandas as pd def compute_metrics(equity_curve: pd.DataFrame) - dict: equity equity_curve[equity] daily_return equity.pct_change().dropna() total_return equity.iloc[-1] / equity.iloc[0] - 1 annual_return (1 total_return) ** (252 / max(len(equity), 1)) - 1 volatility daily_return.std() * np.sqrt(252) sharpe annual_return / volatility if volatility 0 else 0.0 max_drawdown (equity / equity.cummax() - 1).min() return { total_return: total_return, annual_return: annual_return, sharpe: sharpe, volatility: volatility, max_drawdown: max_drawdown, trade_count: len(equity_curve), }需要注意这里用 252 作为年化交易天数适合日线策略。如果是小时级或分钟级数据年化系数要重新计算不能直接套用。6.3 绩效指标速查指标计算方式用途取值范围建议总收益率期末净值 / 期初净值 - 1整体盈利水平只能作为参考不能单独看年化收益率总收益按年化折算跨周期对比需要考虑无风险利率夏普比率(年化收益 - 无风险利率) / 年化波动风险调整后收益大于 1 属于可接受大于 2 需要警惕过拟合最大回撤净值从峰值回落的最大比例风险感受和个人心理承受能力相关建议控制在 20% 以内交易次数所有买入或卖出次数成本评估过少可能样本不足过多可能被手续费吞掉收益如果回测结果中夏普比率超过 3而且样本量不大大概率存在问题优先怀疑前视偏差、幸存者偏差或过拟合而不是真的找到了无敌策略。6.4 模拟盘运行的观察方式模拟盘不需要实时盯着但需要每天检查几个关键点日志里是否出现预期交易。信号是否按时生成。风控是否阻止了本应阻止的交易。模拟盘持仓和日志中的成交记录是否一致。模型概率分布是否发生明显漂移。建议把模拟盘的每日净值存到 SQLite 表里后续可以和回测净值画在同一个图表里对比。如果模拟盘表现和回测差距过大先不要急着优化模型先检查数据、成交价、手续费和风控是否和回测一致。7. 常见问题排查从数据到模型的典型异常链路7.1 回测收益很高但模拟盘持续亏损这是个人量化项目最常见的现象核心原因是回测环境比真实环境乐观。排查顺序如下检查回测是否用了信号当日的收盘价成交。如果是改成次日开盘价或次日收盘价后再看结果。检查是否加入了滑点和手续费。真实环境的手续费、印花税和滑点成本会明显吞噬高频交易收益。检查回测是否把停牌、涨跌停的股票按正常价格成交。真实环境中这些情况下单无法成交回测却会按理想价成交。检查是否存在未来函数。比如用未来 N 日数据计算特征或者用全量数据做归一化。这类问题排查时应该先输出回测中的每笔交易明细逐笔对比模拟盘里的近似场景。如果回测中某笔交易在模拟盘里根本触发不了问题就出在执行假设上。7.2 模型训练时准确率高但没有交易信号准确率高不一定是好事。当正负样本比例严重不均衡时模型可能把所有样本都预测为多数类准确率看似很高但对少数类的预测能力几乎没有。检查手段如下print(y.value_counts(normalizeTrue))如果负样本占比超过 90%就要考虑调整标签阈值或使用class_weightbalanced。另一个原因是特征缺失值过多模型把大量样本预测为低概率。要检查推理时传入的特征是否包含 NaN以及滚动窗口在早期是否因为没有足够历史而全部为 NaN。生产环境里信号中断的另一个常见原因是数据源停更、股票代码变更、网络超时。建议对每天的模型推理结果生成摘要例如当天有信号的股票数量、平均概率、缺失特征数量便于快速发现异常。7.3 模型过拟合和特征泄露过拟合表现为训练集效果远好于验证集或者回测结果对参数极其敏感。特征泄露则更加隐蔽常见情况包括用未来收益构造特征例如在特征计算中使用了shift(-1)。做数据归一化时用全量数据的均值和标准差导致测试集信息泄漏到训练集。用整段数据筛选特征后再做交叉验证选择过程本身引入了未来信息。标签和特征来自同一天的数据且特征包含了当天收盘到标签构造时间段内的信息。预防特征泄露的有效手段是把所有需要滚动计算的东西都用时间序列方式处理训练和推理走完全相同的代码路径并且对训练集和测试集按时间切分而不是随机切分。7.4 API 限流、数据空值和排错表外部数据接口经常会因为请求频率过高返回空数据或异常。不能用异常信息代替业务日志要在每次数据下载时记录请求地址、返回行数、耗时和错误信息。问题现象可能原因检查方式处理建议数据下载为空股票代码错误、日期范围超出上市时间、接口限流单独打印原始返回内容检查代码格式增加重试和缓存特征全是 NaN滚动窗口太大且数据长度不足打印每列非空数量增加数据长度或减小窗口回测收益异常高前视偏差、未计手续费、幸存者偏差逐笔检查交易时间和价格改用次日价格成交加入滑点手续费模拟盘下单失败现金不足、持仓不足、风控拦截查看风控日志和订单拒绝原因先处理订单状态再调整策略仓位模型预测概率长期不变特征退化、数据源停更、缓存未刷新对比最近 N 天特征分布检查数据更新任务和缓存策略这里要强调一个通用排错顺序先确认输入数据正确再确认特征正确再确认模型推理正确最后才怀疑信号和执行逻辑。很多问题表面出在模型层根因都在数据层。8. 生产化建议与后续扩展方向8.1 日志、监控和告警要先于收益考虑个人量化系统在模拟盘跑通后会想把策略放进更自动化的运行环境。此时最优先做的不是提升模型效果而是让系统可观测。每条关键路径都要有日志数据下载成功或失败记录数据行数、时间范围。特征计算完成记录特征维度、缺失值数量。模型推理完成记录当日信号数量和概率分布。订单提交、成交、取消、拒绝记录唯一订单号。监控指标可以做成一个简单的/metrics接口也可以用 SQLite 表记录每日快照。对个人项目来说最实用的告警是“信号缺失”和“策略运行失败”两类。比如连续两个交易日没有产生任何信号或者数据更新任务连续失败三次都应该触发提醒。8.2 部署与调度用定时任务而不是人肉执行个人量化系统通常只需要一台云服务器就能运行。配置不用太高但建议满足这些条件系统时间准确使用 UTC 存储时间展示时再转本地时间。使用systemd或cron调度定时任务保证数据更新、模型训练、信号生成按计划执行。数据库定期备份data/raw和models目录纳入版本管理。环境变量和敏感信息不要写入配置文件统一放在.env或密钥管理服务里。一个简单的调度方案是使用cron# 每个交易日 18:10 更新数据 10 18 * * 1-5 cd /opt/min-quant .venv/bin/python scripts/download_data.py --config config/settings.yaml logs/download.log 21注意cron里的%符号有特殊含义实际使用时需要转义或改用脚本封装。另外不要在同一台机器上跑多个重复的定时任务否则会出现重复下载、重复下单。8.3 从 AI 信号到可解释投资助手的扩展方向Min 作为个人 AI Quant 项目后续扩展方向可以从三条线展开。第一条线是数据扩展。在行情数据基础上加入财务数据、宏观经济指标、新闻情绪数据让模型的特征不再局限于价格和成交量。这个方向工作量在数据清洗和一致性处理上而非模型算法。第二条线是信号集成。不再只输出一个概率而是把规则信号、统计信号、基本面信号和模型信号放在一起做加权或排序形成一个多因子视角。这样即使某个信号失效整体系统也不会立刻失效。第三条线是交互和解释。把模型每次预测的关键特征、特征贡献、相似历史样本展示给使用者让 AI 信号从“黑箱打分”变成“可讨论的决策参考”。这也是个人 AI 量化助手和普通选股软件最大的区别。对个人开发者来说最值得投入的并不是写更多预测函数而是把数据、特征、模型和验证这几件事做成一个稳定闭环。Min 这个方向的价值也正在于此。下一步可以先从模拟盘开始用一个你觉得有解释能力的信号跑通全流程再逐步加入更多数据源和模型让系统在真实环境中慢慢暴露问题并迭代。