简介基于vnpy的二次开发项目围绕选股、回测与机器学习三大主题提供完整工程源码、详细文档与实验数据适合金融量化方向的在校学生、研究人员以及希望深入vnpy框架的Python开发者使用。压缩包共1659个文件包含C/C头文件与实现、Python策略脚本、动态链接库、Jupyter Notebook实验笔记、npy数据文件等类型覆盖底层交易网关封装、策略逻辑编写、数据预处理与模型推理等环节Notebook与npy文件可直接配合复现机器学习选股流程整体包体约59.1MB。项目为个人高分毕设经导师指导并通过答辩代码均经运行验证功能完整可部署运行。已有188人学习下载。资料内详细文档与源码目录结构清晰既适用于课程设计、毕业设计或初期项目演示也便于在此基础上扩展自定义策略内含针对不同交易系统的接口适配代码可辅助理解vnpy如何接入券商柜台从而快速掌握二次开发与量化研究全流程。1. 这个vnpy二次开发资源到底值得拆吗选股、回测、机器学习一条线走通有人拿vnpy跑了半年还在单标的CTA里打转有人的vnpy体验却是另一回事——每天早上跑完选股脚本A股股票池只剩下几十只值得盯的标的策略一挂就是一个月。我拆这份vnpy二次开发资料包最直接的感受是它把选股、回测、机器学习三个经常被分开玩的东西串成了一根流水线而且每一步都有能直接跑的代码和按文件编号排好的文档。适合谁适合那种已经用vnpy下过单、却不知道怎么把一只股票的策略扩展到一篮子股票的人也适合课程设计想拿“量化机器学习”凑一套完整亮点的学生。你不需要精通CPython能写明白函数调用就能跟完。2. 二次开发先找对入口engine、App与策略模板的挂载链路2.1 从vnpy的启动流程看懂“二次开发到底改哪”vnpy的启动逻辑其实很短先建EventEngine再建MainEngine然后把App一个个挂进去。很多人在二次开发时一上来就改vnpy内置的cta_strategy模块改到最后发现行情订阅、合约查询、数据库存储全都耦合在一起越改越乱。from vnpy.event import EventEngine from vnpy.trader.engine import MainEngine from vnpy.trader.ui import MainWindow, create_qapp # 这里替换成你自己的App类名 from my_app import MyStockPickerApp def main(): qapp create_qapp() event_engine EventEngine() main_engine MainEngine(event_engine) # add_app是vnpy识别新模块的唯一入口 main_engine.add_app(MyStockPickerApp) main_window MainWindow(main_engine, event_engine) main_window.showMaximized() qapp.exec() if __name__ __main__: main()这段代码的作用是把你的自定义App挂进vnpy主引擎。add_app执行之后vnpy会扫到App里的engine_class自动创建对应的Engine实例。所以二次开发的第一步不是改官方代码而是写一个自己的App。参数上只有一个地方需要注意如果你不想启动UI把MainWindow那两行去掉直接跑main_engine也可以但事件引擎必须保留否则策略的on_bar永远不会被触发。2.2 一个自定义App骨架把选股逻辑挂进事件驱动App的最小结构是三个文件__init__.py里声明App类engine.py里写引擎ui.py里写界面。引擎是整个二次开发的核心它不直接跑策略而是负责数据加载、选股结果推送、以及把信号转成vnpy能认识的BarData。from vnpy.trader.engine import BaseEngine, EngineType from vnpy.event import EventEngine, EVENT_TIMER class MyStockPickerEngine(BaseEngine): engine_type EngineType.LIVE def __init__(self, main_engine, event_engine): super().__init__(main_engine, event_engine, StockPicker) self.stock_pool [] self.event_engine.register(EVENT_TIMER, self.process_timer) def process_timer(self, event): # 定时任务的回调vnpy默认每1秒推送一次EVENT_TIMER # 不适合在这里做重计算只做状态检查 self.main_engine.write_log(选股引擎心跳正常) def start_picking(self, date: str): # 真正的选股逻辑放这里后续第3章会展开 self.stock_pool self.run_picker(date) self.main_engine.write_log(f{date} 选股完成标的数: {len(self.stock_pool)})这个骨架说明一件事vnpy的事件驱动不是靠你写while循环而是靠event_engine.register把回调函数挂到指定事件上。EVENT_TIMER适合做定时器行情事件EVENT_BAR则适合做盘中信号。我一般在process_timer里只做日期判断到点才触发run_picker避免每分钟都在跑全市场扫描。3. 选股模块怎么建RPS、因子打分与机器学习增强的落地顺序3.1 先算RPS和因子分选股池的压缩逻辑选股模块的第一步是把全市场几千只股票压缩到一两百只。RPS股价相对强度是欧奈尔体系里最常用的指标核心思想是看某只股票过去N日涨幅在全市场的分位排名。很多讲RPS的资料把它写得玄乎其实就是rank和percentile两个函数的事。import pandas as pd import numpy as np def calc_rps(close_df: pd.DataFrame, window: int 20) - pd.Series: close_df: 每列是一只股票每行是一个交易日值为收盘价 window: 计算过去多少日的涨幅常用20、60、120 # 用未来数据是选股大忌这里直接用shift保证只用已发生数据 ret close_df / close_df.shift(window) - 1 # axis1 表示按行做排名pctTrue 输出0~1的分位值 rps ret.rank(axis1, pctTrue) return rps def stock_picker(close_df: pd.DataFrame, score_weights: dict) - pd.Series: total_score pd.Series(0, indexclose_df.columns) for win, weight in score_weights.items(): rps calc_rps(close_df, windowwin) total_score total_score rps.iloc[-1] * weight return total_score.sort_values(ascendingFalse)参数上我建议至少算20、60、120三个窗口短周期负责捕捉启动长周期负责确认趋势。score_weights常见配比是{20: 0.3, 60: 0.3, 120: 0.4}。这里最容易翻车的地方是shift(window)如果你不shift排名里就会混进当天收益率实盘当天你根本拿不到这笔数据属于典型的未来函数。3.2 机器学习增强用日频因子训练一个二分类打分器纯RPS选股的问题在于因子太单一牛熊切换时容易追高。这个资料包里的做法是用机器学习做第二道筛选把RPS、波动率、换手率、量比这些日频因子堆进去训练一个二分类模型预测未来10日收益率能否跑赢中位数。模型不需要多复杂随机森林在这个场景下已经够稳重点是样本和标签的构造方式。from sklearn.ensemble import RandomForestClassifier from sklearn.metrics import roc_auc_score import pandas as pd def build_features(df: pd.DataFrame) - pd.DataFrame: df: 单只股票的日K按时间升序 df df.copy() df[rps_20] df[close].rolling(20).apply(lambda x: (x[-1] / x[0] - 1) * 100) df[volatility] df[close].pct_change().rolling(20).std() df[turnover_ratio] df[volume] / df[float_shares] df[volume_ratio] df[volume] / df[volume].rolling(5).mean() df[label] (df[close].shift(-10) / df[close] - 1 df[close].pct_change(10).median()).astype(int) return df.dropna()这步的标签构造是个高频翻车点shift(-10)表示用未来10天的收盘价作为标签但输入特征只截止到当前行。训练的时候把全部数据直接丢进RandomForestClassifier去看准确率的人大概率会得到一个虚高的线上指标原因就是训练集和测试集时间重叠。标准做法是按时间切分前80%训练、后20%验证且验证集从某一天开始连续截取不能随机抽样。3.3 选股结果落库SQLite表结构设计选股结果不能每次现算尤其是盘中策略需要秒级响应的时候。这个资料包用的是vnpy的database接口但选股结果建议单独建表毕竟它和行情数据、交易记录的冷热程度都不同。CREATE TABLE IF NOT EXISTS stock_pool ( id INTEGER PRIMARY KEY AUTOINCREMENT, trade_date TEXT NOT NULL, symbol TEXT NOT NULL, score REAL NOT NULL, source TEXT NOT NULL, UNIQUE(trade_date, symbol) );代码里用INSERT OR REPLACE写入这样同一交易日重复跑选股不会产生脏数据。读取的时候按trade_date过滤策略初始化时直接拉当日选股结果避免策略启动阶段去复算全市场因子。这个表设计上只有两个字段要注意source用来区分是纯RPS选出的还是ML增强后的方便事后归因symbol建议存成600519.SSE这种带交易所后缀的格式这样可以直接被vnpy的ContractData匹配不用再做字符串转换。4. 多股回测怎么做一颗引擎跑完整个股票池的取舍4.1 原生BacktestingEngine为什么只能单标的vnpy自带的BacktestingEngine是给CTA策略用的它的核心是单标的的单bar推进逻辑策略的on_bar收到的永远是一只股票的bar持仓、手续费、滑点也都是围绕这一个symbol算的。想拿它直接跑股票池第一关就过不去——它根本没有“组合市值”的概念也没有换仓逻辑。我在拆这份资料包时看到作者的处理方式很务实保留vnpy的数据加载和bar生成部分自己写了一个轻量组合回测器。原因是vnpy的BaseData、BarData定义本身很干净直接沿用能让数据格式保持一致。4.2 简化组合回测器的实现核心思想是按交易日循环每天先处理调仓信号再更新净资产曲线。信号来自第3章选的股票池调仓频率可以按天也可以按周。class PortfolioBacktester: def __init__(self, init_cash: float 1_000_000, commission: float 0.0003, slippage: float 0.002): self.init_cash init_cash self.commission commission self.slippage slippage self.positions {} self.cash init_cash def run(self, bars_by_date: dict, signals_by_date: dict): equity_curve [] for trade_date, bars in bars_by_date.items(): # 拿到当日调仓目标 target signals_by_date.get(trade_date, None) if target is not None: self.rebalance(target, bars) # 先用收盘价粗略估算当日净值 market_value sum(pos[volume] * bars[sym].close_price for sym, pos in self.positions.items()) equity_curve.append({date: trade_date, equity: self.cash market_value}) return equity_curve def rebalance(self, target_symbols: list, bars: dict): # 先清掉不在目标池中的持仓 for symbol in list(self.positions.keys()): if symbol not in target_symbols: self.cash self.positions[symbol][volume] * bars[symbol].close_price * (1 - self.slippage) - self.commission del self.positions[symbol] # 等权重买入新目标 target_volume self.init_cash / len(target_symbols) / 100 for symbol in target_symbols: if symbol not in self.positions: price bars[symbol].close_price * (1 self.slippage) self.positions[symbol] {volume: target_volume, price: price} self.cash - target_volume * price * (1 self.commission)参数上几个关键点slippage我默认设0.002这是A股散户打涨停板或开盘成交最常见的冲击成本量级commission按万三设target_volume除以100是为了凑整手不符合A股100股一手的规则会导致回测曲线和实盘偏差很大。这段代码舍掉了涨跌停无法成交、停牌无法卖出这两个细节但作为课程设计或策略初筛已经够用。4.3 回测结果与信号复盘的对照检查多股回测最容易出现“净值曲线很漂亮实盘却亏钱”的情况。我的习惯是回测完不看最终收益率先看三个东西最大回撤对应的日期区间、那段时间股票池的行业分布、以及调仓时间点是收盘前还是次日开盘。这个简化回测器里的信号默认在当日收盘后处理但实际下单要次日开盘才能成交所以如果你看到回测里换仓当天就产生了收益贡献基本可以判定是未来函数。参数上还有个小技巧回测区间至少跨越一次明显的风格切换。比如2021年是茅指数行情2023年是中小盘行情只跑单边行情会让组合回测器看起来无敌实盘一遇风格反转就露馅。5. 避坑专题vnpy二次开发里我翻过的七个车5.1 数据层的坑时区、复权与停牌坑1vnpy的bar时间戳永远是UTC8的datetime但你从外部数据源拿到的日K可能是0点或16点。现象策略回测收益正常实盘信号晚一天触发。原因外部数据源把日期存成了2024-01-01 00:00:00而vnpy的BarData.datetime是按交易日收盘时间生成的。解决数据入库前统一对datetime做一次replace(hour0, minute0, second0)或者直接按日期字符串匹配不要依赖时间戳相等判断。坑2前复权数据用错导致选股指标失真。现象RPS排名前十的股票里混进一堆刚除权的低价股。原因直接用未复权价格算20日涨幅除权当天的价格跳空会把涨幅算成暴跌。解决日K一律用前复权且算因子时前复权的价格序列必须是从上市日起连续复权的不能用片段复权。坑3停牌股在回测里被当成可交易标的。现象换仓日选中的股票里有停牌股回测时仓位照常买入实盘根本挂不上单。原因回测器没有过滤当日status字段把停牌也当成有bar数据。解决在信号生成时就把当日成交量小于一个阈值的标的剔除成交量极低时即使没停牌也基本成交不了。5.2 回测层的坑未来函数与成交价坑4用当天收盘价计算信号再用当天收盘价成交。现象回测胜率超过70%实盘一月就打回原形。原因信号计算和成交用了同一个bar的close等于你在收盘瞬间既知道结果又下了单。解决信号用第T根bar计算成交价要么用第T1根bar的开盘价要么强制在信号中引入shift(1)让成交永远落后信号一个bar。坑5滑点参数设成0还能自我安慰。现象回测资金曲线平滑上涨实盘手续费和滑点一加就亏损。原因股票回测的滑点不是按价格固定百分比算的而是按买一卖一价差幅度。解决把滑点从固定百分比改成“按当日振幅的0.1倍”估算至少保留0.5%的缓冲同时把手续费按卖出的万五来算A股印花税只在卖出时收取很多人漏掉这一项。5.3 机器学习层的坑标签泄漏与样本重叠坑6训练集和验证集按行随机切分准确率虚高。现象模型在验证集上AUC 0.85实盘策略一塌糊涂。原因同一只股票的相邻交易日样本高度相关随机切分把同一段时间的数据同时放进了训练集和验证集。解决按连续时间段切分直接用某一天作为切分线之前的全部做训练之后的全部做验证。坑7因子数据没做截面归一化导致模型学到的是绝对价格水平。现象训练集里高价股占比高模型推理时对高价股打分虚高。原因成交量、换手率这些因子在不同股票间的量纲差异太大模型学到了绝对数量特征而不是相对强弱。解决每个交易日对所有股票的因子做一次横截面rank或zscore让输入变成相对排名这也是ML选股和传统因子选股最本质的一个区别。6. 把ML信号接进vnpywalk-forward验证与一套能坚持的核对习惯6.1 滚动验证的代码骨架最后这一步是把机器学习模型真正接到策略里之后怎么确认它没有失效。我不用一次性回测而是做滚动验证每年重新训练一次模型然后用下一年数据做样本外测试评估指标不达标就停用。def walk_forward_validate(factor_df: pd.DataFrame, train_years: int 3): results [] for year in range(2019, 2024): train factor_df[factor_df[year] year] test factor_df[factor_df[year] year] # 只保留最近train_years年的数据做训练防止旧数据干扰 train train[train[year] year - train_years] model RandomForestClassifier(n_estimators200, max_depth6) model.fit(train[feature_cols], train[label]) test[score] model.predict_proba(test[feature_cols])[:, 1] auc roc_auc_score(test[label], test[score]) results.append({year: year, auc: auc}) return pd.DataFrame(results)这段代码的逻辑是2019年验证集用2016~2018年训练2020年验证集用2017~2019年训练以此类推。train_years3这个参数很关键A股风格周期通常在2~3年左右超过5年的数据反而会成为噪声。AUC阈值我一般设在0.55以上才让策略上线低于这个数的模型即使胜率看着高也大概率是过拟合了。6.2 每日信号快照与模型版本核对模型上线后最怕的是静默失效。我现在的日常流程是每天收盘后把当日的模型输出分数存一份CSV快照隔一周回来对比分数排名和实际收益的差异同时记录模型文件名哈希值防止不小心覆盖了旧模型文件。这份资料包里的check_result.py脚本做的就是这件事它会把当日的信号快照和3日后的实际收益做一次相关性检验我用它替代了每天肉眼盯资金曲线。从那以后我每次改策略都强制走一遍这个流程先walk-forward验证再对比信号快照和实际收益最后才决定是否上实盘。这套习惯救了我很多次希望帮到你。本文还有配套的精品资源点击获取