简介基于vnpy的二次开发综合实践包面向量化交易开发者、金融科技从业者及有Python基础的投资者重点解决自动化选股、策略回测与机器学习落地难题。包内共1656个文件解压后约59MB其中299个Python脚本覆盖数据获取、策略编写与模型调用217个C源文件与639个头文件用于底层逻辑和高性能计算407个头文件支撑多类交易接口49个动态库便于直接链接调用另有少量模型、配置文件与说明目录结构清晰便于二次开发。目前已有596人学习。内容涉及CTP、IB等行情接入与数据库管理回测引擎自带收益曲线、夏普比率、最大回撤等统计指标机器学习部分展示线性回归、SVM、随机森林及多模型集成选股方法并以C扩展加速矩阵运算可据此构建完整自动化交易系统。1. 基于vnpy的二次开发选股、回测和机器学习到底在做什么拿到这套基于vnpy的二次开发包等于把量化研究的三个环节焊成了一条流水线先用因子打分或机器学习模型把全市场几千只股票压缩成几十只的股票池再拿vnpy的事件驱动回测引擎验证策略参数和历史绩效最后把训练好的模型预测值接回vnpy的CtaTemplate生成交易信号。这个方向解决的核心痛点是vnpy原生能力偏CTA单品种股票的批量选股、组合回测和机器学习信号接入都得靠自己二次开发补齐。适合有Python基础、熟悉pandas、想用机器学习做选股但又不想从零搭框架的团队或个人。下面按我实际落地的顺序把每一块的扩展点、关键参数和踩过的坑讲清楚。2. 二次开发的骨架vnpy留了哪些口子、工程目录怎么摆vnpy在国内开源交易框架里属于“重但完整”的那一类行情网关、事件引擎、策略模板、回测引擎都是现成的二次开发不是重写它而是选对扩展点接进去。选错口子的代价很大——我见过有人把机器学习模型训练直接塞进on_bar回调里回测一次跑四个小时实盘还卡顿。先讲清楚三个扩展点再给最小工程骨架。2.1 二次开发优先改的三个位置策略模板、数据服务、应用脚本vnpy的事件驱动链路可以简化为四步gateway收到行情 - 事件引擎分发 - 策略回调on_bar/on_tick - 策略调用buy/sell发单。这条链路决定二次开发只需要动三处。第一处是策略模板层。cta_strategy模块里的CtaTemplate是基类重写on_bar、on_tick、参数定义就是最常见的二次开发。股票选股模型产出的预测值最终也落到这里变成信号函数——策略逻辑本身不复杂复杂的是“这个预测值怎么算出来”。第二处是数据服务层。vnpy默认数据库是SQLite但股票量化需要行情、复权因子、财务指标多张表数据量远大于CTA。常见做法是保留vnpy的数据库接口同时旁路维护一个自己的数据库我一般直接开一个data/market.db回测和选股脚本都从这里读避免被vnpy的ORM层限制。注意vnpy 2.x到3.x的数据库配置方式变过升级后要重点检查vnpy.trader.database相关的import。第三处是应用脚本层。选股、因子计算、批量回测这类一次性任务不建议做成vnpy的app写成独立python脚本与vnpy通过数据库交互就够了。真正常跑的只有两件事vnpy主程序负责实盘或仿真独立脚本负责研究和回测。网关层不建议动。vnpy的gateway对接柜台协议稳定性要求极高除非要接私有行情否则改了后续升级全是坑。我处理过的二次开发项目里凡是动了gateway的无一例外在vnpy版本升级时翻车。2.2 最小工程骨架目录结构、依赖安装和连通性验证一个能复现的vnpy二次开发包目录应该按数据、因子、策略、回测、模型五块分而不是把所有代码堆在vnpy的example目录里。下面是我常用的一套结构project_root/ ├── config.py # 全局参数资金、费率、回测区间 ├── data/ │ ├── fetch_daily.py # 拉取日线行情入库 │ ├── fetch_finance.py # 拉取财务指标入库 │ └── market.db # 本地SQLite数据库 ├── factors/ │ └── factor_lib.py # 因子计算与标准化 ├── strategy/ │ └── ml_strategy.py # 继承CtaTemplate的策略模板 ├── backtest/ │ ├── run_backtest.py # 单股回测 │ └── run_portfolio.py # 多股组合回测 └── ml/ ├── feature_engine.py # 特征工程 └── train_model.py # 模型训练与滚动验证依赖安装建议用独立虚拟环境vnpy的依赖比较重和系统Python混装容易在NumPy版本上打架python -m venv .venv source .venv/bin/activate # Windows下用 .venv\Scripts\activate pip install vnpy vnpy_ctastrategy pandas numpy akshare scikit-learn lightgbm装完先做连通性验证不要急着跑策略python -c from vnpy.trader.constant import Interval; print(Interval.DAILY) python -c from vnpy_ctastrategy import CtaTemplate, BacktestingEngine; print(cta ok)第一行验证vnpy基础包能import第二行验证你二次开发最依赖的cta_strategy模块在。如果第二行报ImportError是vnpy版本差异导致的路径问题优先查vnpy_ctastrategy的包名和Python版本兼容表。我遇到过的情况是vnpy 2.x时代import自vnpy.app.cta_strategy3.x改成vnpy_ctastrategy代码里路径不换就全挂。注意Python版本建议3.8到3.11。vnpy 3.x在3.12上部分依赖包没有预编译wheel装起来很痛苦。3. 选股模块落地数据接入、因子计算和股票池输出选股模块是整个流水线的上游它的输出直接决定后面回测和机器学习吃什么数据。很多团队在这里犯的第一个错是数据量贪多——把能拉到的字段全存下来结果因子计算时被数据清洗耗掉大半时间。我的原则是行情表只存开高低收、成交量、换手率五个核心字段财务表只存ROE、毛利率、负债率等十来个因子原料字段其余随用随拉。3.1 行情与财务数据接入把数据灌进本地SQLitevnpy的数据库接口可以扩展但股票选股阶段我更喜欢直接操作SQLite理由很实际选股脚本和回测脚本都是离线任务不追求实时SQLite轻量、可移植、用pandas读写都方便。下面这段是我常用的日线拉取入库脚本# data/fetch_daily.py import akshare as ak import sqlite3 import pandas as pd def fetch_daily(symbol: str, start: str, end: str, adjust: str hfq) - pd.DataFrame: 拉取单只股票日线并写入SQLite避免每次回测都重复请求数据源。 adjust参数 qfq 前复权适合看盘 hfq 后复权适合做因子计算和历史回测 空串 不复权适合算真实涨跌幅。 df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart, end_dateend, adjustadjust, ) df.columns [date, open, close, high, low, volume, amount, amplitude, pct, change, turnover] df[symbol] symbol df[date] pd.to_datetime(df[date]) return df[[symbol, date, open, close, high, low, volume, turnover]] conn sqlite3.connect(data/market.db) for code in [600519, 000858, 601318, 000001]: df fetch_daily(code, 20180101, 20241231, adjusthfq) df.to_sql(daily_bar, conn, if_existsappend, indexFalse) conn.close()这段代码里最关键的参数是adjust。用后复权做因子计算是底线前复权价格会被后续除权除息事件整体修正意味着你在2018年这个历史时点用到了2019年才知道的除权信息这就是数据泄漏。后复权只做一次折算历史上是什么样就是什么样。回测时如果数据源本身给了复权因子字段比如tushare的adj_factor优先存因子而不是直接存复权价这样能随时切换口径。财务数据接口各大数据源差异较大akshare的ak.stock_financial_abstract按股票拉tushare的pro接口可以按报告期批量拉。字段我建议只留四个roe、gross_profit_margin、debt_to_assets、operating_cash_flow_per_share按报告期对齐到交易日期时注意用“公告日期”而不是“报告期”报告期数据披露有滞后用报告期对齐等于让历史回测提前知道了财报。3.2 因子计算与打分选股从全市场压到TopN因子不是越多越好这一点是我做机器学习选股后的血泪经验。两三个逻辑清楚的基础因子配上严格的样本外验证远好过堆三四十个相关性很高的因子在模型里做黑匣子。先给一个可以直接抄的因子清单因子名计算方式方向momentum_20过去20日收益率越高越好volatility_20过去20日收益标准差越低越好turnover_2020日平均换手率适中偏好valuationPE倒数或PB倒数越高越好价值roe_latest最近报告期ROE越高越好打分选股的逻辑是每个因子做z-score标准化方向不对的取负然后等权相加得总分取前N只进入股票池。代码如下# factors/factor_lib.py def compute_and_score(df: pd.DataFrame, top_n: int 30) - list: 计算因子、标准化、等权打分返回TopN股票代码。 # 按股票分组计算因子取每个股票最新一行 df df.sort_values([symbol, date]) g df.groupby(symbol, group_keysFalse) df[momentum_20] g[close].transform(lambda x: x.pct_change(20)) df[volatility_20] g[close].transform(lambda x: x.pct_change().rolling(20).std()) df[turnover_20] g[turnover].transform(lambda x: x.rolling(20).mean()) latest df.groupby(symbol).tail(1).copy() # z-score注意波动率因子方向是负向 latest[mom_z] (latest[momentum_20] - latest[momentum_20].mean()) / latest[momentum_20].std() latest[vol_z] -1 * (latest[volatility_20] - latest[volatility_20].mean()) / latest[volatility_20].std() latest[score] (latest[mom_z] latest[vol_z]) / 2 return latest.nlargest(top_n, score)[symbol].tolist()这段代码里有三个要点。其一是transform保证了因子对齐到每个交易日不会因为groupby打乱时间顺序其二是负向因子取负号再标准化让所有因子统一成“越大越好”其三是最终返回的股票池要落到数据库或CSV里回测脚本按调仓日读取不在回测过程中现算因子——回测阶段现算因子容易混入未来数据属于回测的大忌。到这里选股模块的核心路径已经跑通原始数据入库、因子计算、打分选股、输出池子。机器学习选股在这个框架里只是把“等权打分”替换成“模型预测值排序”特征工程直接复用上面的因子这个衔接放到第6章讲。4. 回测模块落地撮合参数、单股回测和多股组合净值选股选出来的是池子池子能不能赚钱取决于策略和回测设置。vnpy的回测引擎BacktestingEngine是基于事件驱动的和backtrader这类框架比优势是回测逻辑和实盘策略共用一套CtaTemplate代码劣势是原生只支持单合约回测。股票组合回测需要自己在外面包一层下面先讲单股回测的参数再讲组合合成。4.1 单股回测BacktestingEngine的关键参数怎么设vnpy回测的最小步骤是定义策略类 - 创建BacktestingEngine - set_parameters设置合约和成本 - add_strategy - load_data - run_backtesting - calculate_result。先看一个最小示例# backtest/run_backtest.py from vnpy.trader.constant import Interval from vnpy_ctastrategy import CtaTemplate, BacktestingEngine class MomentumStrategy(CtaTemplate): 演示策略突破20日高点买入跌破20日低点清仓。 def on_init(self): self.bars [] self.pos 0 def on_bar(self, bar): if not self.inited: return self.bars.append(bar) if len(self.bars) 20: return high_20 max(b.close_price for b in self.bars[-20:]) low_20 min(b.close_price for b in self.bars[-20:]) if bar.close_price high_20 and self.pos 0: self.buy(bar.close_price, 100) elif bar.close_price low_20 and self.pos 0: self.sell(bar.close_price, self.pos) engine BacktestingEngine() engine.set_parameters( vt_symbol600519.SSE, intervalInterval.DAILY, start20180101, end20241231, rate0.00025, # 手续费率股票默认按万2.5 slippage0.01, # 滑点单位是价格不是百分比 size1, # 股票合约乘数是1 pricetick0.01, # 最小价格变动单位 capital1_000_000, # 初始资金 ) engine.add_strategy(MomentumStrategy, {}) engine.load_data() engine.run_backtesting() df engine.calculate_result() # 返回含datetime和balance的净值序列 engine.calculate_statistics()参数说明里最容易搞错的是slippage和size。slippage在vnpy里是价格单位股票设0.01表示每笔成交价在信号价基础上多付一分钱别把它当成百分比去设否则回测成本会被放大一百倍。size对股票是1对期货是合约乘数比如螺纹钢是10股票用户直接填1即可。另外vt_symbol的格式是“代码.交易所”上交所是SSE深交所是SZSE写成600519.SSE和000001.SZSE写反了回测会报合约信息错误。回测跑完先看calculate_result的净值曲线再跑calculate_statistics拿统计指标。不要直接信任统计函数里的夏普数值股票日线策略用日频收益算夏普和CTA用小时线算的夏普不在一个量纲上横向比较时统一频率才有意义。4.2 多股组合回测逐股回测再合成组合净值vnpy原生只支持单合约回测做多股组合回测的常见做法是对股票池每只股票跑一遍单股回测得到每日收益序列再按日期对齐做等权组合。这个方案很直接缺点是每次调仓都要重新跑全部股票好在股票池一般只有30到50只跑一轮分钟级就能完成。# backtest/run_portfolio.py import sqlite3 import pandas as pd from vnpy.trader.constant import Interval from vnpy_ctastrategy import BacktestingEngine def backtest_one(symbol: str, exchange: str, cfg: dict) - pd.DataFrame: 单股回测返回以日期为索引的每日收益序列。 engine BacktestingEngine() engine.set_parameters( vt_symbolf{symbol}.{exchange}, intervalInterval.DAILY, startcfg[start], endcfg[end], ratecfg[rate], slippagecfg[slippage], size1, pricetick0.01, capitalcfg[capital], ) engine.add_strategy(MomentumStrategy, {}) engine.load_data() engine.run_backtesting() result engine.calculate_result() # 含datetime、balance result[daily_return] result[balance].pct_change().fillna(0) return result.set_index(datetime)[daily_return] cfg {start: 20180101, end: 20241231, rate: 0.00025, slippage: 0.01, capital: 1_000_000} conn sqlite3.connect(data/market.db) # 股票池表由第3章的选股脚本写入 pool pd.read_sql(select distinct symbol from stock_pool, conn)[symbol] nav 1.0 for code in pool: exchange SSE if code.startswith(6) else SZSE ret backtest_one(code, exchange, cfg) nav nav * (1 ret).fillna(1) # 等权合成先简化处理 portfolio_nav nav这里我做了简化等权合成的正确做法是每只股票的资金权重相同且每天重新平衡到等权上面的逐股累乘相当于“买入并持有每只股票”的组合如果策略本身是调仓型需要在每个调仓日对每只股票的收益序列做权重还原。更稳的做法是每只股票回测时记录每日持仓市值组合层按市值加权或等权重新平衡。组合回测的绩效指标要重点看五个年化收益率、夏普比率、最大回撤、卡玛比率年化收益除以最大回撤、年化换手率。其中换手率最容易被忽视一个高夏普但换手率极高的策略实盘里会被冲击成本和滑点吃掉大半收益。我一般给组合回测加一个换手率上限约束超过阈值直接放弃该参数组。5. 避坑vnpy二次开发和机器学习回测的5个高频翻车点前面三章落地之后下面5个问题最该反复看。我都在实际项目里遇到过前三个偏数据和框架层后两个偏策略和模型层按“现象-原因-解决”写清楚每个都能单独排掉。5.1 数据与框架层复权口径、vnpy版本和数据对齐坑一复权因子不一致同一策略回测结果差一倍。现象同一个双均线策略一套回测年化25%另一套只有8%代码几乎没差别。原因第一套用前复权数据第二套用后复权数据。前复权价格会被后续除权事件整体缩放导致历史K线的涨跌幅被改写策略信号的触发时间点都变了回测指标自然全变。解决全流程统一用后复权做因子和信号计算实盘下单时再把后复权价格换算成真实成交价。数据入库时把复权因子字段单独存一列不要只存复权后的价格这样切换口径只改一行代码。坑二升级vnpy之后二次开发代码大面积失效。现象原来跑得好好的策略pip install -U vnpy之后import直接报错策略类找不到基类。原因vnpy 2.x到3.x做了模块拆分原来vnpy.app.cta_strategy的路径改为独立的vnpy_ctastrategy包数据库ORM也换过一轮。文档里搜得到的旧示例大多是老版本直接照着抄就撞墙。解决用requirements.txt锁死vnpy和vnpy_ctastrategy版本升级前先在一个虚拟环境里跑最小回测验证。我现在的做法是项目根目录放一个requirements-lock.txt里面固定主版本号除非需要新功能否则不主动升级vnpy。坑三行情时间戳和财务公告日期错位。现象机器学习特征里ROE和行情数据合并后特征值数量对不上IC值忽高忽低。原因财务数据接口默认按报告期给数据比如年报是12月31日但实际披露往往是次年三四月份。按报告期对齐到行情日期等于让1月份的行情用上了4月才公布的财报数据泄漏。解决对齐财务数据时用公告日期ann_date而不是报告期end_date在特征工程里让“最新可得”的财务数据在披露后才进入样本。5.2 策略与模型层前视偏差、数据泄漏和幸存者偏差坑四on_bar里用收盘价发单等于偷看收盘价。现象回测净值曲线光滑上涨实盘一塌糊涂。逐笔对比发现策略在当天收盘价创新高时买入回测引擎也按当天收盘价成交——这在日线策略里就是典型的前视偏差。原因on_bar收到的bar是已经收盘的完整K线此时你看到收盘价已经是事实但vnpy回测引擎默认按当前bar价格撮合订单等于下了个“收盘后才提交的市价单”还能按收盘价成交实盘根本做不到。解决信号产生和成交错开一根bar常见做法是on_bar里用上一根bar的收盘价做判断当前bar开盘价成交或者让策略在on_bar里只使用开盘时可得的数据收盘数据只用于下一根bar的信号决策。关键是养成“上一根bar出信号、下一根bar成交”的习惯。坑五机器学习回测的幸存者偏差和随机划分。现象机器学习选股策略样本内IC达到0.08回测年化30%以上上线后大幅缩水。原因两个问题叠加。一是用当前时点沪深300成分股回测历史退市股被排除在外历史回测的股票池本身是“活下来的股票”二是训练时把历史数据随机打乱划分训练集和测试集时间序列的强自相关性让模型在“用未来验证过去”。解决股票池用历史时点成分股或者全市场规模股至少包含历史退市股训练和验证严格按时间切分训练集只包含验证集之前的样本不做随机shuffle。机器学习模型在量化里的应用流程最大的忌讳就是把图像识别那套“随机划分数据集”的习惯直接搬过来。6. 机器学习进阶滚动验证与把模型预测接进vnpy策略选股不只有打分法把机器学习模型的预测值作为选股信号是这个二次开发包里的核心进阶用法。机器学习在量化里的应用流程可以拆成四步特征工程、时间序列切分、模型训练、样本外验证。这里只讲最容易做歪的两步。6.1 滚动验证先过样本外这一关股票收益率不是独立同分布的随机打乱划分训练集和测试集在海量样本的CV任务里没问题在金融时序里就是数据泄漏。滚动验证的规则是训练集永远在验证集之前且按时间段前移重复训练。# ml/train_model.py import pandas as pd from lightgbm import LGBMRegressor def rolling_predict(features: pd.DataFrame, train_days: int 240, pred_days: int 20, target: str future_return): 滚动训练LightGBM输出各段样本外预测值用于计算IC和分组收益。 preds [] idx [] for start in range(train_days, len(features), pred_days): train features.iloc[start - train_days:start] test features.iloc[start:start pred_days] if len(test) 20: continue model LGBMRegressor(n_estimators100, learning_rate0.05, num_leaves15, verbose-1) model.fit(train.drop(columns[target]), train[target]) preds.append(model.predict(test.drop(columns[target]))) idx.extend(test.index) pred_series pd.Series( pd.concat([pd.Series(p) for p in preds]).values, indexidx) return pred_series滚动验证的窗口参数要按策略的调仓频率来定。日线策略我常用train_days240约一年、pred_days20约一个月滚动这样每次训练都在最近一年的数据上做。learning_rate和num_leaves的取值宁小勿大股票收益预测本身信噪比极低模型复杂度过高必过拟合。收敛标准是样本外IC稳定在正值且排名稳定如果只有一段特定市场里IC好看这个信号大概率是市场风格不是你模型的能力。6.2 把模型预测接进vnpy策略on_bar里只做推理不做训练模型训练是离线任务vnpy策略里只加载训练好的模型做推理。常见做法是feature_engine和model一起序列化策略on_bar里先算因子特征再调model.predict把预测值拼成一个信号# strategy/ml_strategy.py import pickle class MLStrategy(CtaTemplate): buy_threshold 0.02 # 预测未来20日收益大于2%买入 sell_threshold -0.02 def on_init(self): with open(ml/model.pkl, rb) as f: self.model pickle.load(f) with open(ml/feature_engine.pkl, rb) as f: self.feature_engine pickle.load(f) def on_bar(self, bar): if not self.inited: return feat self.feature_engine.calc(bar) # 只用当前bar已收盘可得的数据 pred self.model.predict(feat)[0] if pred self.buy_threshold and self.pos 0: self.buy(bar.close_price, 100) elif pred self.sell_threshold and self.pos 0: self.sell(bar.close_price, self.pos)注意两个细节其一feature_engine的输入只能是当前bar及之前数据能算出来的字段任何用到未来bar的统计量都会让模型训练时的特征分布和实盘不一致其二on_init里加载模型文件不要在on_bar里每次读盘回测和实盘都会因此变慢。我现在每上一套机器学习选股都先把滚动验证跑一遍确认样本外IC稳定才接进vnpy策略。回测里看到的高夏普多半是某个环节偷看了未来数据对齐和滚动验证这两步能省掉后面大部分返工。希望帮到你。本文还有配套的精品资源点击获取