简介基于vnpy的二次开发实践资源面向量化交易开发者、金融分析师及机器学习爱好者聚焦选股、策略回测与AI辅助决策三大环节。压缩包内共1656个文件以Python源码、C头文件与实现文件为主包含299个py脚本、639个h、407个hpp及217个cpp另含少量dll动态库、ipynb示例及配置文件整体约59.07MB覆盖数据接口、回测引擎、策略库与C高性能扩展等模块。已有598人学习。通过该资源可掌握vnpy框架的二次开发方法理解如何接入实时行情、编写并回测复杂策略、集成Scikit-Learn或TensorFlow等机器学习库进行价格预测与多因子选股同时可参考C扩展技巧提升计算效率适用于希望构建自动化选股与交易系统的中高级开发者系统学习并积累实战经验。1. 从vnpy源码到自用平台选股、回测、机器学习这条链路值不值得做很多人第一次装好vnpy跑通自带示例策略的回测就以为“二次开发”已经完成大半。真要把选股、回测、机器学习串成一条能反复产出结果的流水线很快就会撞上三堵墙vnpy默认没有选股模块选出来的股票池不知道放哪里回测引擎的默认参数跟自己的交易逻辑对不上算出来的夏普完全不可信机器学习模型算出的信号也不知道该塞进策略的哪个位置。这三个问题单独看都不难连在一起就成了劝退点。标题里那个.zip暗示了这类交付物的真实形态核心价值往往不是压缩包里的某段代码而是把vnpy从“能跑示例”改造成“能跑自己的想法”的那套工程骨架。这篇文章我会按自己常用的改造路径展开选股产出股票池回测验证交易逻辑机器学习提供信号三者闭环。适合已经在用vnpy、但觉得默认功能不够用的从业者也适合准备用vnpy做全流程验证的初学者。看完你会知道每个模块从哪下手、参数怎么定以及哪些坑是必然要踩的。2. 先摸清vnpy的扩展边界事件驱动架构里哪些能改、哪些必须自己写2.1 vnpy的模块化结构与二次开发切入点vnpy本质上是一个事件驱动框架行情、委托、成交、日志都被包装成事件在内部的事件循环里流转。这个设计对实盘交易很友好但二次开发时容易迷失方向因为不是所有功能都适合塞进事件循环。真正需要关心的切入点只有四个gateway负责把外部行情和交易通道转成内部事件app是功能聚合层自带的cta_strategy、data_manager等模块都在这一层strategy是策略逻辑所在database负责K线、合约、订单的持久化。选股和机器学习在vnpy里没有现成位置常见做法是在app目录下新增自研模块或者干脆独立成一个进程只借用vnpy的数据库和回测引擎。判断一件事该不该在vnpy内部改我一般用一条标准它是否依赖实时事件流。实盘策略必须挂到事件循环里这是vnpy的主场选股是典型的批量截面计算跑在vnpy外面更合适回测是离线过程直接调vnpy的回测引擎机器学习训练需要全市场历史数据应当放独立进程。按这样划分改动量最小出问题时也最好定位。很多翻车的项目都是把选股和训练逻辑塞进Strategy里结果盘中卡顿、内存暴涨最后连回测结果都复现不出来。2.2 最小环境准备与目录规划我建议按“vnpy内核不动周边独立”的方式组织工程先装最小依赖集pip install vnpy vnpy_ctastrategy vnpy_sqlite pandas scikit-learn这个命令覆盖了跑通整套流程的最小集合vnpy核心负责事件引擎和主窗口vnpy_ctastrategy提供CtaTemplate策略模板和回测引擎vnpy_sqlite用来落股票池和模型信号scikit-learn先顶替机器学习库的位置后续可以按需换成其他库。版本不同时vnpy_ctastrategy的导入路径可能有差异本文按常见版本的import方式写遇到ModuleNotFoundError时优先检查包名和Python环境的对应关系。代码结构不要等写大了再拆一开始就按功能分开project/ ├── vnpy_app_ext/ # 自研app后续注册进vnpy主程序 ├── stock_picker/ # 选股脚本不依赖vnpy运行 ├── ml_signal/ # 特征工程、训练、推理 └── backtest/ # 回测配置和结果分析选股和机器学习独立成目录意味着你可以脱离vnpy的图形界面单独调试。等结果稳定后再把信号文件喂给vnpy回测。这一步很关键直接决定后面回测失败时是策略问题、数据问题还是模型问题不用一锅粥地查。2.3 第一个修改点让vnpy跑起来并加载自研策略先做一次最小验证写一个空策略在vnpy里加载成功再在回测引擎里跑通。这里的目标不是赚钱而是确认你的改造环境没有链路断层。from vnpy.app.cta_strategy import CtaTemplate class SignalFilterStrategy(CtaTemplate): 从外部信号文件读取买卖指令再交给vnpy撮合 def __init__(self, cta_engine, strategy_name, vt_symbol, parameter): super().__init__(cta_engine, strategy_name, vt_symbol, parameter) self.signal_file parameter.get(signal_file, ) self.target_pos 0 def on_init(self): self.write_log(策略初始化完成) self.load_bar(10) def on_bar(self, bar): # 非实盘环境里这里先保持空仓验证链路 self.pos 0注意parameter是dict类型vnpy会把外部传入的策略参数整体打包给你signal_file字段就是给后续机器学习信号留的接口。on_bar在vnpy里每根K线触发一次你后续所有交易判断都从这里进入。这个空策略能加载、能跑回测说明整个二次开发环境已经通了接下来逐个模块填功能。3. 选股模块在vnpy里扩展一个自己的股票池应用3.1 选股在vnpy里的位置为什么需要单独造一个应用选股本质是对全市场股票做截面打分依赖的是某一时刻所有股票的因子值而不是单只股票的时序行情。vnpy的事件驱动模型处理单标的的行情流和委托流是强项但对“全市场排序”这件事没有自然落点。如果试图在策略里遍历全市场股票事件循环会被阻塞盘中实时性完全丧失。常见做法是把选股做成独立应用定时产出股票池写入数据库实盘策略只订阅池子里的标的而不是让策略去全市场扫描。开始做之前先想清楚股票池的更新频率。日频选股只需要每个交易日收盘后算一次盘中没有任何新信息不需要实时计算。低频任务用系统定时器或者自己写循环都行没必要挂进vnpy事件引擎。这样选股模块就成了一个标准的数据批处理任务写起来简单测起来也稳定。3.2 因子计算与打分脚本选股模块的核心是因子计算。以下脚本从vnpy的sqlite数据库读日线数据计算动量、波动率、成交异常三个因子然后做百分位排名合成总分。行业中性化因为依赖行业分类数据这里先省略但真实使用时建议加上否则选出来的股票可能高度集中在某一两个行业里。import pandas as pd from sqlalchemy import create_engine engine create_engine(sqlite:///vnpy.db) def load_daily_bars(vt_symbol): df pd.read_sql_table(dbbardata, engine, columns[ datetime, symbol, exchange, close, volume ]) df df[df[symbol] vt_symbol.split(.)[0]] df[datetime] pd.to_datetime(df[datetime]) return df.sort_values(datetime).set_index(datetime) def compute_factors(symbol_list, lookback20): rows [] for symbol in symbol_list: df load_daily_bars(symbol) if len(df) lookback: continue df[ret] df[close].pct_change() momentum df[close].iloc[-1] / df[close].iloc[-lookback] - 1 volatility df[ret].tail(lookback).std() volume_ratio df[volume].iloc[-1] / df[volume].tail(lookback).mean() rows.append({symbol: symbol, momentum: momentum, volatility: volatility, volume_ratio: volume_ratio}) return pd.DataFrame(rows) factors compute_factors([600001.SSE, 600002.SSE]) factors[score] ( factors[momentum].rank(pctTrue) * 0.5 - factors[volatility].rank(pctTrue) * 0.3 factors[volume_ratio].rank(pctTrue) * 0.2 ) top_pool factors.nlargest(20, score)load_daily_bars从vnpy默认的sqlite表dbbardata读K线直接绕开vnpy的DataManager界面批量处理时快很多。vt_symbol传入的是“代码.交易所”格式读取后只保留symbol字段避免交易所后缀干扰匹配。compute_factors里每个symbol单独读一次表逻辑清晰但性能一般数据量大时可以改成全市场一次读入再分组效果相同。这里“按股票循环”是故意保留的为的是让你先看清每个因子的构造过程再去做性能优化。score权重0.5、0.3、0.2是经验值不要当成标准答案可以先跑一版看股票池的行业分布和换手率再回头调。rank(pctTrue)把因子值转成百分位排名消除了不同股票价格量纲的影响这是多因子打分里比较稳妥的标准化方式。3.3 股票池落库与定时调度算完因子后要把股票池存到vnpy能读的地方。最简单的方法是单独建一张表策略启动时只订阅表中当天的标的。我把评分一并存下来这样后续回测时能分析“高分组是不是真的跑赢低分组”而不是只留下一份股票池清单。from sqlalchemy import Column, String, Float, create_engine from sqlalchemy.orm import declarative_base, sessionmaker Base declarative_base() class StockPool(Base): __tablename__ stock_pool date Column(String, primary_keyTrue) symbol Column(String, primary_keyTrue) score Column(Float) engine create_engine(sqlite:///stock_pool.db) Base.metadata.create_all(engine) Session sessionmaker(bindengine) session Session() for _, row in top_pool.iterrows(): session.add(StockPool(date2024-01-05, symbolrow[symbol], scorerow[score])) session.commit()这里把日期设为主键之一是为了支持历史股票池回溯。以后做组合回测时可以按日期取当天的持仓而不是用当前股票池去回测过去那会造成严重的前视偏差。定时调度我用系统crontab每天收盘后跑一次选股脚本vnpy策略次日开盘前加载最新股票池。如果不想引入crontab也可以在vnpy里开一个定时器但盘中触发选股纯属浪费资源而且会在交易时段占用事件循环。4. 回测引擎二次开发把vnpy默认回测改成能接自研信号的形态4.1 回测引擎的工作流与可扩展参数vnpy的回测引擎是标准的事件驱动撮合流程按时间顺序喂入K线策略的on_bar逐根处理下单后引擎按设定撮合成交。二次开发的三个重点分别是数据怎么喂、撮合参数怎么设、结果指标怎么算。先看一个最基础的调用骨架from vnpy.app.cta_strategy.backtesting import BacktestingEngine from datetime import datetime engine BacktestingEngine() engine.set_parameters( vt_symbol600001.SSE, interval1d, startdatetime(2022, 1, 1), enddatetime(2024, 1, 1), rate0.0003, slippage0.02, size1, pricetick0.01, capital1_000_000, ) engine.add_strategy(MyStrategy, {}) engine.load_data() engine.run_backtesting() df engine.calculate_result() engine.show_chart()set_parameters里的rate是手续费率按成交金额双边收取slippage是滑点按每个tick的价格单位计算pricetick是最小变动价位限价单价格不在pricetick整数倍上会被引擎拦掉size表示每手的合约乘数股票一般填1capital是初始资金vnpy按总资金百分比下仓的逻辑依赖这个参数。这个骨架能跑通但离一个可用回测系统还差得远。真正的二次开发在于你的选股结果是给一篮子股票用的而vnpy默认的回测引擎一次只测一个合约。你需要在外面自己做一层容器循环遍历股票池把每只股票的回测收益汇总成组合曲线。这里的组合收益计算不能简单平均要考虑每只股票在池子里的权重、停牌日缺失收益的处理以及换仓日的交易成本。4.2 配置手续费、滑点与涨跌停约束股票回测里有几个参数特别容易误导人。手续费建议不要只用默认的万三要按你实际开户的佣金率、最低收费和卖出印花税一起折算进rate。滑点建议用固定价格单位而不是比例因为股票的最小变动价位是固定的比例滑点在小市值股票上会被放大失真。涨跌停约束是股票回测最容易踩的坑vnpy默认撮合不校验涨跌停意味着你会在实盘根本买不进的价位成交。vnpy没有内建的涨跌停判断我一般在策略自己的on_bar里加过滤def is_limit_up(self, bar): pre_close getattr(bar, pre_close, None) if pre_close is None: return False limit_ratio 0.1 return bar.close_price pre_close * (1 limit_ratio) - 1e-6 def is_limit_down(self, bar): pre_close getattr(bar, pre_close, None) if pre_close is None: return False limit_ratio 0.1 return bar.close_price pre_close * (1 - limit_ratio) 1e-6 def on_bar(self, bar): if self.is_limit_up(bar) or self.is_limit_down(bar): return # 正常信号处理这里的关键点是判断涨跌停用的必须是除权除息后的前收盘价。如果数据源没有复权pre_close会在除权日出现跳变涨停判断就会当天失效。1e-6的容差是处理浮点计算的误差避免价格正好在涨停价上被漏判。bar里如果没有pre_close字段getattr会返回None函数直接跳过过滤这种情况要回到数据源确认不能放着不管。回测参数汇总如下参数建议值说明rate按实际佣金折算含佣金、印花税、最低收费建议用单边统一费率slippage固定价格单位按pricetick的整数倍设置不要用比例pricetick0.01沪深A股最小变动价位涨跌停过滤自行在on_bar实现引擎不校验策略层必须处理capital等于实盘计划资金影响下单手数和仓位百分比4.3 扩展绩效指标把最大回撤和夏普换成自定义指标calculate_result返回的是逐日盈亏DataFrame你可以基于它计算任何自定义指标。比如我想看“选股打分前20名的股票每天换仓一次的累计净值”就必须自己聚合多品种的回测结果。def portfolio_net_value(per_stock_results, weightsNone): 把多只股票的逐日收益汇总成组合净值 all_returns [] for symbol, result in per_stock_results.items(): ret result[daily_balance].pct_change() all_returns.append(ret.rename(symbol)) df pd.concat(all_returns, axis1) if weights is None: weights pd.Series(1.0 / len(df.columns), indexdf.columns) df[portfolio] (df * weights).sum(axis1) df[nav] (1 df[portfolio]).cumprod() return df这段代码先把每只股票的账户净值转成日收益率再按权重合成组合收益最后累乘得到组合净值。weights可以传入你选股评分里的真实权重等权只是兜底方案。注意concat时一定要先对齐日期索引vnpy回测结果里周末没有数据不同股票的停牌日也会导致索引错位合并前要统一reindex到同一交易日历。这一步不做稳后面的夏普、最大回撤全是错的。5. 机器学习接入vnpy从特征工程到信号落库的完整链路5.1 特征与标签怎么构造对齐K线和收益机器学习接入vnpy难的不是训练模型而是让模型输出能对齐到vnpy的K线时间轴。标签构造必须避免前视偏差用t日的特征预测t1日之后一段时间的收益t日收盘后才知道的所有信息才能进特征。我常用的标签是未来5日收益方向特征则全部来自t日及之前的数据def build_samples(df, feature_cols, horizon5): df df.sort_values(datetime).reset_index(dropTrue) df[future_ret] df[close].shift(-horizon) / df[close] - 1 df[label] (df[future_ret] 0).astype(int) df df.dropna(subset[future_ret]) features df[feature_cols].shift(1) # 关键特征整体滞后一天 labels df[label] return features.iloc[:-horizon], labels.iloc[:-horizon]shift(1)是整条链路里最重要的操作它把所有特征整体滞后一天确保t日收盘后生成的因子和下一天的开盘决策对齐。这一行能拦住绝大多数前视偏差代价是少了一天样本。horizon决定预测周期日频策略一般取5周频可以取20。shift(-horizon)算的是未来收益必须放在dropna之前否则样本数会莫名减少。特征列的选择上动量、波动率、成交量变化率、价格偏离均线的幅度都够用。不建议一上来就堆几百个因子先跑十几个经典因子看AUC和特征重要性后面再加。特征数量远大于样本数量时任何模型都容易过拟合这在股票日线数据上尤其明显。5.2 训练一个模型并导出信号文件用随机森林做分类器理由是可解释性强、不需要过多调参、在中等样本量下不容易爆。下面这段代码从build_samples接过features和labels做时间序列切分后训练模型features, labels build_samples(all_bars, feature_cols, horizon5) from sklearn.ensemble import RandomForestClassifier from sklearn.model_selection import train_test_split X_train, X_test, y_train, y_test train_test_split( features, labels, test_size0.2, shuffleFalse ) model RandomForestClassifier( n_estimators200, max_depth5, min_samples_leaf50, n_jobs-1, random_state42, ) model.fit(X_train, y_train) pred_proba model.predict_proba(X_test)[:, 1]train_test_split用shuffleFalse做时间序列切分避免随机打散把未来数据混进训练集。min_samples_leaf50是控制过拟合的关键参数叶子节点样本太少模型会记住历史上的异常K线。n_estimators200在日线数据量下足够再多边际收益很小训练时间却直线上升。random_state固定下来保证回测结果可复现否则每次跑的结果都不一样你根本没法判断改动是有效还是噪声。训练完成后把预测概率写入信号文件供vnpy策略读取signal_df pd.DataFrame({ datetime: feature_index, symbol: symbol, score: pred_proba }) signal_df.to_csv(signal_600001.csv, indexFalse)时间戳格式是最容易忽略的坑。vnpy里BarData的datetime通常是纳秒精度而pandas默认存csv再读回来是微秒或字符串格式直接对不上。我一般把信号文件里的时间统一格式化成“YYYY-MM-DD HH:MM:SS”读入策略后再用pd.to_datetime统一转换这个习惯帮我避掉了大量莫名其妙的对齐错误。5.3 策略侧加载信号模型输出如何变成策略买卖条件拿到信号文件后在vnpy策略里按时间匹配信号。这里的关键是日频信号和vnpy的K线时间戳要能精确对齐否则每天第一根K线就会漏匹配。import pandas as pd class MLSignalStrategy(CtaTemplate): def __init__(self, cta_engine, strategy_name, vt_symbol, parameter): super().__init__(cta_engine, strategy_name, vt_symbol, parameter) self.signal_table {} signal_df pd.read_csv(parameter[signal_file]) signal_df[datetime] pd.to_datetime(signal_df[datetime]) for _, row in signal_df.iterrows(): self.signal_table[row[datetime]] row[score] def on_bar(self, bar): bar_time bar.datetime.replace(minute0, second0, microsecond0) score self.signal_table.get(bar_time, 0.0) if score 0.6 and self.pos 0: self.buy(bar.close_price, 1) elif score 0.4 and self.pos 0: self.sell(bar.close_price, self.pos)这里用signal_table存了时间到概率的映射on_bar每根K线查一次避免了每次读取csv文件的开销。score0.6这个阈值不是固定不变的我通常先看验证集上预测概率的分布取一个能保证交易次数不过少的阈值。bar_time做了分钟归零处理日频信号就把时分秒全部清掉再查表。如果不做这一步每天只有精确到纳秒的那根K线能匹配上其他时间全部查不到信号策略就会莫名空仓。6. vnpy二次开发避坑清单数据对齐、前视偏差与过拟合6.1 前视偏差回测很漂亮实盘就翻车现象模型回测年化50%实盘连亏三个月。原因特征或标签泄露了未来信息。解决把特征整体shift(1)用t日收盘后能拿到的数据预测t1日标签用未来收益时确保label对应的时间段完整走完才进入样本集。记住一条硬规则任何进模型的数字必须只依赖截止到t日收盘的信息。6.2 数据对齐停牌、复权和时间戳格式现象回测结果里某些股票出现连续极端净值跳变。原因停牌日被当成正常交易日计算收益或者前复权价格在除权日跳空。解决回测前过滤停牌日用前复权数据计算因子和收益信号文件里的时间戳格式化统一读入vnpy后再做一次datetime转换不同股票回测结果合并时先reindex到同一个交易日历。6.3 模型过拟合样本内测试不能说明任何问题现象训练集AUC 0.85验证集AUC只有0.52。原因模型容量太大或特征数远多于样本数。解决min_samples_leaf放大到50以上特征控制在30个以内用滚动训练验证模型稳定性。机器学习加进vnpy后最大的错觉就是“回测跑完就万事大吉”实际上回测只是把前两个坑的后果放大给你看。6.4 我的最终验证习惯滚动样本外测试与信号延迟模拟最后说一下我现在固定用的验证流程。每次训练完模型先做两次回测第一次用原始信号第二次把信号整体向后平移一个K线模拟实盘里信号产生和下单之间的延迟。两次回测的收益差距如果超过10%说明策略对执行时机太敏感我会直接毙掉这个模型不投入实盘。这个平移可以用5.3里的signal_table逻辑做把读取时的datetime统一加一个bar周期。我踩过最狠的一次是信号文件里score列的小数点精度出了问题概率0.65被读成6.5所有买入信号全部触发净值曲线当场拉成直线。后来我把信号文件纳入版本管理每次回测前用脚本做格式校验检查score是否在0到1之间、时间戳是否单调递增、符号是否匹配标的代码。这个习惯救了我很多次。希望这些经验帮到你。本文还有配套的精品资源点击获取