vnpy二次开发实战：选股、回测与机器学习信号集成

简介这份资源围绕vnpy量化框架展开二次开发覆盖选股、回测与机器学习三大方向面向计算机、人工智能、通信工程等专业的在校学生、教师及企业开发者可用于毕业设计、课程设计、项目立项演示或量化交易入门进阶。压缩包共1659个文件约59.1MB以C头文件与源码h、hpp、cpp构成交易接口底层Python脚本承担策略与回测逻辑另含Jupyter笔记本、npy数据、dll动态库及界面文件结构完整、层次分明。资源已获导师认可答辩评审95分代码经测试可正常运行。读者可从中获得完整的量化选股与回测实现思路、机器学习模型接入方式、多交易接口如CTP、SGIT等的对接范例以及可复用的工程目录组织与排错参考适合在此基础上修改扩展快速搭建自己的量化研究项目。目前已有188人学习关注。1. 从一份 vnpy 二次开发包说起选股、回测、机器学习到底怎么串起来很多人第一次接触 vnpy 是从 CTA 策略模板开始的写个双均线、跑个螺纹钢回测觉得量化交易不过如此。但真正把 vnpy 当成一套可扩展的量化研究底座来用问题就来了选股逻辑往哪塞机器学习模型怎么和回测引擎对接回测结果怎么验证不是过拟合这份「基于 vnpy 的二次开发选股、回测、机器学习」的资料包本质上解决的就是这三个环节的工程化串联问题。它适合已经跑通过 vnpy 基础回测、想往多因子选股和 ML 信号方向走的开发者也适合手里有选股公式但不知道怎么系统化验证的从业者。核心思路不复杂用 vnpy 的数据层和回测引擎做骨架把选股模块做成独立信号源把机器学习输出转成标准信号格式最后统一走事件驱动回测。下面按落地顺序拆开讲。2. vnpy 二次开发的骨架数据、信号、回测三层怎么改2.1 为什么直接改 vnpy 源码是下策vnpy 的架构是事件驱动加模块化核心对象是EventEngine、MainEngine、BaseStrategy和BacktestingEngine。很多人拿到资料包第一反应是去改vnpy/app/cta_strategy里的源码把选股逻辑硬塞进on_bar里。这样做短期能跑但后面加机器学习信号、换股票池、做多因子排序时代码会变成一团乱麻。常见做法是保持 vnpy 核心不动在vnpy/app下新建一个独立模块比如stock_selector让它继承BaseApp通过MainEngine注册进去。这样选股逻辑和 CTA 策略解耦回测时既可以单独跑选股模块也可以把选股结果作为信号源喂给策略。具体操作上先确认 vnpy 版本。资料包一般基于 vnpy 3.xPython 3.10 左右。安装完 vnpy 后在项目根目录建apps/stock_selector/里面至少放四个文件__init__.py、engine.py、template.py、ui.py如果不用 GUI 可以省。engine.py里定义StockSelectorEngine负责加载股票列表、拉取历史数据、计算因子、输出选股结果。template.py里定义StockSelectorTemplate作为策略和选股引擎之间的接口。关键点是选股引擎不直接下单只产出信号信号格式统一成SignalData对象包含symbol、datetime、direction、strength四个字段。这样后面接回测或实盘都方便。# apps/stock_selector/engine.py from vnpy.event import EventEngine, Event from vnpy.trader.engine import BaseEngine, MainEngine from vnpy.trader.object import BarData from vnpy.trader.constant import Interval, Exchange from typing import List, Dict import pandas as pd APP_NAME StockSelector class StockSelectorEngine(BaseEngine): 选股引擎负责数据加载、因子计算、信号输出 def __init__(self, main_engine: MainEngine, event_engine: EventEngine): super().__init__(main_engine, event_engine, APP_NAME) self.symbols: List[str] [] # 股票池 self.bars: Dict[str, pd.DataFrame] {} # 缓存历史数据 self.signals: List[dict] [] # 选股信号 def load_symbols(self, symbol_list: List[str]): 设置股票池常见做法是从 CSV 或数据库读取 self.symbols symbol_list def load_bars(self, start: str, end: str, interval: Interval Interval.DAILY): 通过 main_engine 的数据库接口拉取历史 K 线 for symbol in self.symbols: bars self.main_engine.get_bars( symbol, interval, start, end, outputlambda x: x ) if bars: df pd.DataFrame([{ datetime: b.datetime, open: b.open_price, high: b.high_price, low: b.low_price, close: b.close_price, volume: b.volume, } for b in bars]) self.bars[symbol] df.set_index(datetime) def compute_factors(self) - pd.DataFrame: 计算多因子返回 DataFrame索引为 symbol records [] for symbol, df in self.bars.items(): if len(df) 60: continue close df[close] volume df[volume] # 动量因子20 日收益率 momentum close.iloc[-1] / close.iloc[-20] - 1 # 波动率因子20 日收益率标准差 volatility close.pct_change().iloc[-20:].std() # 量比因子5 日均量 / 20 日均量 vol_ratio volume.iloc[-5:].mean() / volume.iloc[-20:].mean() records.append({ symbol: symbol, momentum: momentum, volatility: volatility, vol_ratio: vol_ratio, }) return pd.DataFrame(records).set_index(symbol) def select(self, top_n: int 10) - List[dict]: 按动量排序选前 N 只输出标准信号 factors self.compute_factors() if factors.empty: return [] ranked factors.sort_values(momentum, ascendingFalse).head(top_n) self.signals [{ symbol: sym, datetime: pd.Timestamp.now(), direction: long, strength: row[momentum], } for sym, row in ranked.iterrows()] return self.signals这段代码的逻辑说明load_bars通过main_engine.get_bars拉数据这是 vnpy 的标准数据接口底层走的是数据库SQLite 或 PostgreSQL。compute_factors里只写了三个因子实际资料包里通常有十几个包括反转、换手、市值等。select方法输出信号列表每个信号带strength后面回测时可以用它做仓位权重。参数说明top_n控制选股数量一般 5 到 20 只momentum的窗口 20 日可以改成 10 日或 60 日取决于持股周期。注意get_bars的output参数在 vnpy 3.x 里是回调函数这里用lambda x: x直接返回列表如果数据量大建议改成生成器避免内存爆掉。2.2 把选股信号接入回测引擎的两种方式选股信号有了怎么让 vnpy 的回测引擎认有两种常见做法。第一种是写一个StockSelectionStrategy继承BaseStrategy在on_init里调用选股引擎拿到信号然后在on_bar里根据信号决定是否开仓。这种方式适合信号频率和 K 线频率一致的场景比如日线选股、日线调仓。第二种是绕过 CTA 策略模板直接用BacktestingEngine的add_strategy方法但把选股逻辑写在策略的on_bars里利用 vnpy 的多标的回测能力。资料包里一般两种都会给但推荐第一种因为结构清晰容易调试。# apps/stock_selector/strategy.py from vnpy_ctastrategy import CtaTemplate from vnpy.trader.object import BarData, TickData from vnpy.trader.constant import Direction from apps.stock_selector.engine import StockSelectorEngine class StockSelectionStrategy(CtaTemplate): 选股策略根据选股引擎信号调仓 author quant_dev # 策略参数 top_n 10 rebalance_days 5 fixed_size 100 parameters [top_n, rebalance_days, fixed_size] variables [hold_days, selected_symbols] def __init__(self, cta_engine, strategy_name, vt_symbol, setting): super().__init__(cta_engine, strategy_name, vt_symbol, setting) self.hold_days 0 self.selected_symbols [] self.selector: StockSelectorEngine None def on_init(self): 策略初始化时获取选股引擎实例 self.write_log(选股策略初始化) self.selector self.cta_engine.main_engine.get_engine(StockSelector) if self.selector: self.selector.load_symbols([000001.SZSE, 600000.SSE, 000300.SSE]) self.selector.load_bars(2020-01-01, 2023-12-31) def on_bar(self, bar: BarData): 每根 K 线触发按调仓周期执行选股 self.hold_days 1 if self.hold_days self.rebalance_days: return self.hold_days 0 if not self.selector: return signals self.selector.select(top_nself.top_n) self.selected_symbols [s[symbol] for s in signals] # 先平掉不在新信号里的仓位 for symbol in list(self.pos_data.keys()): if symbol not in self.selected_symbols and self.pos_data[symbol] 0: self.sell(symbol, bar.close_price, abs(self.pos_data[symbol])) # 再开新仓 for sig in signals: symbol sig[symbol] if self.pos_data.get(symbol, 0) 0: self.buy(symbol, bar.close_price, self.fixed_size) def on_tick(self, tick: TickData): pass def on_stop(self): self.write_log(选股策略停止)逻辑说明on_init里通过main_engine.get_engine(StockSelector)拿到选股引擎实例这是 vnpy 模块间通信的标准方式。on_bar里用hold_days控制调仓频率避免每天换股导致交易成本过高。self.pos_data是 vnpy 策略基类维护的持仓字典键是vt_symbol值是持仓数量。参数说明rebalance_days设为 5 表示每周调仓一次日线回测下就是 5 根 K 线fixed_size是每只股票买入股数A 股一般 100 股一手所以设 100 的整数倍。注意self.sell和self.buy的第二个参数是价格回测时用bar.close_price会有未来函数嫌疑更严谨的做法是用下一根 K 线的开盘价但 vnpy 的 CTA 模板默认在当前 bar 收盘价成交需要自己在on_bar里做延迟处理。3. 机器学习信号怎么塞进 vnpy 回测从特征到下单3.1 特征工程和标签构造的工程细节机器学习选股的核心不是模型多复杂而是特征和标签对不对齐。资料包里通常会给一套完整的特征计算流程包括量价特征、财务特征、技术指标。但很多人翻车在标签构造上用未来数据做标签回测夏普很高实盘一塌糊涂。常见做法是标签用未来 N 日收益率但特征只能用当前及历史数据。在 vnpy 里做这件事建议把特征计算和标签生成放在独立的ml_pipeline.py里不要和策略代码混在一起。# apps/stock_selector/ml_pipeline.py import pandas as pd import numpy as np from sklearn.ensemble import GradientBoostingClassifier from sklearn.model_selection import TimeSeriesSplit from sklearn.metrics import accuracy_score import joblib def build_features(df: pd.DataFrame) - pd.DataFrame: 从 OHLCV 构造特征所有特征只使用历史数据 feat pd.DataFrame(indexdf.index) close df[close] volume df[volume] high df[high] low df[low] # 收益率特征 feat[ret_1] close.pct_change(1) feat[ret_5] close.pct_change(5) feat[ret_20] close.pct_change(20) # 波动率特征 feat[vol_20] close.pct_change().rolling(20).std() # 量价特征 feat[vol_ratio_5_20] volume.rolling(5).mean() / volume.rolling(20).mean() feat[price_pos_20] (close - low.rolling(20).min()) / ( high.rolling(20).max() - low.rolling(20).min() 1e-8 ) # 技术指标RSI delta close.diff() gain delta.where(delta 0, 0).rolling(14).mean() loss -delta.where(delta 0, 0).rolling(14).mean() rs gain / (loss 1e-8) feat[rsi_14] 100 - 100 / (1 rs) return feat.dropna() def build_labels(df: pd.DataFrame, forward_days: int 5, threshold: float 0.02) - pd.Series: 标签未来 forward_days 日收益率是否超过 threshold future_ret df[close].shift(-forward_days) / df[close] - 1 labels (future_ret threshold).astype(int) return labels def train_model(features: pd.DataFrame, labels: pd.Series): 时序交叉验证训练 GBDT 模型 tscv TimeSeriesSplit(n_splits5) model GradientBoostingClassifier( n_estimators100, max_depth3, learning_rate0.1, random_state42 ) scores [] for train_idx, val_idx in tscv.split(features): X_train, X_val features.iloc[train_idx], features.iloc[val_idx] y_train, y_val labels.iloc[train_idx], labels.iloc[val_idx] model.fit(X_train, y_train) pred model.predict(X_val) scores.append(accuracy_score(y_val, pred)) print(f时序 CV 准确率: {np.mean(scores):.4f}) model.fit(features, labels) # 全量重训 joblib.dump(model, ml_model.pkl) return model逻辑说明build_features里所有特征都是滚动计算rolling和pct_change不会引入未来数据。build_labels用shift(-forward_days)构造未来收益标签这是监督学习的标准做法但要注意回测时不能用全量数据训练再回测同一段必须做时序切分。train_model用TimeSeriesSplit做交叉验证避免随机切分导致数据泄露。参数说明forward_days设为 5 表示预测未来一周涨跌threshold设为 0.02 表示涨幅超过 2% 才算正样本这两个参数直接决定正负样本比例一般正样本占比 30% 到 50% 比较合理。GradientBoostingClassifier的max_depth设 3 是为了防止过拟合股票数据噪声大树太深容易记住噪声。3.2 把模型预测转成 vnpy 信号并回测模型训练完下一步是在回测中调用。注意不能在on_bar里实时训练模型那样太慢且容易泄露。正确做法是离线训练好模型回测时只做推理。在选股引擎里加一个predict方法加载ml_model.pkl对最新特征做预测输出概率作为信号强度。# apps/stock_selector/engine.py 中追加 import joblib class StockSelectorEngine(BaseEngine): # ... 前面的代码不变 ... def load_model(self, model_path: str ml_model.pkl): 加载离线训练好的模型 self.model joblib.load(model_path) def predict(self, top_n: int 10) - List[dict]: 用 ML 模型预测输出概率最高的 top_n if not hasattr(self, model): raise RuntimeError(模型未加载先调用 load_model) records [] for symbol, df in self.bars.items(): if len(df) 60: continue feat build_features(df) if feat.empty: continue latest feat.iloc[-1:] prob self.model.predict_proba(latest)[0][1] # 正类概率 records.append({symbol: symbol, prob: prob}) if not records: return [] result pd.DataFrame(records).sort_values(prob, ascendingFalse).head(top_n) self.signals [{ symbol: row[symbol], datetime: pd.Timestamp.now(), direction: long, strength: row[prob], } for _, row in result.iterrows()] return self.signals逻辑说明load_model在策略on_init里调用一次避免重复加载。predict遍历股票池对每只股票计算最新特征用predict_proba拿正类概率按概率排序取前 N。参数说明top_n和前面的动量选股一致但 ML 选股通常更集中建议 5 到 10 只。注意build_features需要至少 60 根 K 线才能算出 20 日波动率等指标所以股票池里上市时间太短的票要过滤掉。回测时把策略里的self.selector.select换成self.selector.predict即可其他逻辑不变。4. 回测结果怎么验证不是过拟合参数、样本、基准三件事4.1 样本内外划分和滚动回测回测跑出高夏普先别高兴。资料包里一般会强调样本内外划分但很多人只做一次切分比如 2020 到 2022 训练2023 回测。这样仍然可能过拟合因为 2023 的行情可能恰好适合模型。更稳的做法是滚动回测每年重新训练模型用下一年做测试滚动前进。在 vnpy 里实现滚动回测需要把回测引擎包一层循环每次改变数据区间和模型文件。# rolling_backtest.py from datetime import datetime from vnpy_ctastrategy.backtesting import BacktestingEngine from apps.stock_selector.strategy import StockSelectionStrategy def rolling_backtest(years: list): 滚动回测每年训练下一年测试 results [] for i in range(len(years) - 1): train_start f{years[i]}-01-01 train_end f{years[i]}-12-31 test_start f{years[i1]}-01-01 test_end f{years[i1]}-12-31 # 这里假设模型已经按年份训练好文件名为 ml_model_{year}.pkl engine BacktestingEngine() engine.set_parameters( vt_symbol000300.SSE, intervald, startdatetime.strptime(test_start, %Y-%m-%d), enddatetime.strptime(test_end, %Y-%m-%d), rate0.0003, slippage0.002, size1, pricetick0.01, capital1_000_000, ) engine.add_strategy(StockSelectionStrategy, { top_n: 10, rebalance_days: 5, fixed_size: 100, }) engine.load_data() engine.run_backtesting() df engine.calculate_result() stats engine.calculate_statistics() results.append({ test_year: years[i1], total_return: stats[total_return], sharpe_ratio: stats[sharpe_ratio], max_drawdown: stats[max_drawdown], }) return pd.DataFrame(results)逻辑说明rolling_backtest按年份循环每次用下一年的数据做测试。set_parameters里的rate是手续费率A 股一般万三slippage是滑点设 0.002 比较保守size是合约乘数股票设 1capital是初始资金。参数说明years列表比如[2020, 2021, 2022, 2023]会产出 2021、2022、2023 三年的测试结果。如果三年夏普都为正且差异不大说明模型有一定泛化能力如果只有某一年特别好其他年份亏损大概率是过拟合。注意每次回测前要确保模型文件对应训练年份不能用未来数据训练。4.2 基准对比和交易成本敏感性回测结果必须和基准比。选股策略的基准通常用沪深 300 指数如果策略跑不赢指数那不如直接买 ETF。在 vnpy 里可以加载指数数据作为基准计算超额收益。另外要做交易成本敏感性测试把手续费和滑点调高一倍看策略是否还能盈利。很多高频选股策略在成本翻倍后直接亏钱这种策略实盘基本没戏。指标策略值沪深300说明年化收益18.5%6.2%策略跑赢基准夏普比率1.350.45风险调整后收益更好最大回撤-22%-35%回撤控制更好换手率每月 80%—换手偏高成本敏感成本翻倍后年化9.8%—仍为正但优势缩小这张表是回测报告里必须有的。换手率每月 80% 意味着一年换手近 10 倍手续费和滑点会吃掉大量收益。如果成本翻倍后年化降到 9.8%虽然还是正的但实盘冲击成本可能更高需要谨慎。常见做法是设一个换手率上限比如每月不超过 50%在选股时加一个持仓周期约束避免频繁调仓。5. 避坑与排查vnpy 二次开发选股回测的五个血泪教训5.1 数据对齐没做好回测结果全是幻觉现象回测夏普 3.0实盘一跑就亏。原因选股用的日线数据和回测用的 K 线时间戳没对齐比如选股用了当天收盘后发布的财务数据但回测在当天开盘就成交了。解决所有特征数据必须滞后一天使用财务数据要按公告日期对齐不能用报告期。在 vnpy 里可以在on_bar里判断bar.datetime和特征日期的关系确保只用bar.datetime之前的数据。5.2 股票池包含退市股幸存者偏差拉高收益现象回测收益很高但股票池里全是现在还在交易的票。原因退市股没包含进去而退市股往往跌得很惨。解决股票池要用历史成分股每个调仓日只包含当时已上市且未退市的股票。资料包里一般会给一个get_index_components(date)函数按日期返回成分股列表。如果自己搭可以从聚宽、米筐等平台导出历史成分股存成 CSV 按日期索引。5.3 机器学习模型用了未来数据做特征标准化现象模型在验证集上准确率 70%实盘预测完全随机。原因特征标准化用了全量数据的均值和方差相当于把未来信息泄露给了训练集。解决标准化只能在训练集上拟合然后应用到验证集和测试集。在ml_pipeline.py里用StandardScaler时必须fit在训练集上transform在测试集上。滚动回测时每年重新fit一次。5.4 vnpy 回测引擎的成交价假设过于乐观现象回测成交价用收盘价实盘根本买不到。原因vnpy CTA 模板默认在当前 bar 收盘价成交但实盘中收盘价只是一瞬间的价格大单根本成交不了。解决在策略里做延迟成交比如信号在on_bar产生下一根 bar 开盘价成交。vnpy 支持在on_bar里用self.buy但指定stopFalse实际成交价由回测引擎的cross_limit_order决定可以改成用bar.open_price作为参考。5.5 多标的回测时资金分配不合理现象选股选出 10 只每只都买 100 股结果高价股占用大量资金低价股仓位过小。原因fixed_size是固定股数没有按资金等权分配。解决改成按资金比例下单每只股票分配capital / top_n的资金然后除以股价得到股数。在策略里用self.buy(symbol, price, int(capital / top_n / price / 100) * 100)注意 A 股必须 100 股整数倍。6. 进阶技巧用 vnpy 的 EventEngine 做实时选股信号推送回测跑通后下一步是把选股信号推到实盘或模拟盘。vnpy 的EventEngine本身就是为实时事件设计的可以自定义一个EVENT_SELECTOR_SIGNAL事件类型选股引擎算完信号后put到事件引擎策略或 UI 层register这个事件做响应。这样选股和交易解耦选股可以按分钟、小时或日频跑交易策略按自己的节奏处理信号。# apps/stock_selector/event.py from vnpy.event import Event from vnpy.trader.object import BaseData EVENT_SELECTOR_SIGNAL eSelectorSignal class SelectorSignalData(BaseData): 选股信号事件数据 def __init__(self, symbol: str, direction: str, strength: float, datetime): super().__init__() self.symbol symbol self.direction direction self.strength strength self.datetime datetime # 在 engine.py 的 select 或 predict 方法末尾 from apps.stock_selector.event import EVENT_SELECTOR_SIGNAL, SelectorSignalData from vnpy.event import Event def select(self, top_n: int 10): # ... 前面的计算逻辑 ... for sig in self.signals: event_data SelectorSignalData( symbolsig[symbol], directionsig[direction], strengthsig[strength], datetimesig[datetime], ) self.event_engine.put(Event(EVENT_SELECTOR_SIGNAL, event_data)) return self.signals逻辑说明SelectorSignalData继承BaseData方便统一处理。event_engine.put把信号推送到事件队列任何注册了EVENT_SELECTOR_SIGNAL的组件都会收到。实盘时可以在run.py里注册一个监听器收到信号后调用main_engine.send_order下单。参数说明strength可以用来做仓位权重比如按 strength 排序前 3 名各分配 20% 资金后面各 10%。注意事件引擎是异步的信号推送和订单执行之间可能有延迟高频场景要加时间戳校验。我自己做这类项目最大的教训是回测阶段偷的懒实盘都会加倍还回来。尤其是数据对齐和成本假设这两块宁可回测收益难看一点也要把假设做保守。另一个习惯是每加一个因子或模型先跑一遍滚动回测看三年以上的表现而不是只看最近一年。希望帮到你。本文还有配套的精品资源点击获取