简介：基于vnpy的二次开发项目，围绕选股、回测与机器学习三大主题，提供完整工程源码、详细文档与实验数据，适合金融量化方向的在校学生、研究人员以及希望深入vnpy框架的Python开发者使用。压缩包共1659个文件，包含C/C++头文件与实现、Python策略脚本、动态链接库、Jupyter Notebook实验笔记、npy数据文件等类型，覆盖底层交易网关封装、策略逻辑编写、数据预处理与模型推理等环节；Notebook与npy文件可直接配合复现机器学习选股流程，整体包体约59.1MB。项目为个人高分毕设，经导师指导并通过答辩，代码均经运行验证，功能完整，可部署运行。已有188人学习下载。资料内详细文档与源码目录结构清晰，既适用于课程设计、毕业设计或初期项目演示，也便于在此基础上扩展自定义策略；内含针对不同交易系统的接口适配代码，可辅助理解vnpy如何接入券商柜台，从而快速掌握二次开发与量化研究全流程。
1. 这个vnpy二次开发资源到底值得拆吗：选股、回测、机器学习一条线走通
有人拿vnpy跑了半年，还在单标的CTA里打转；有人的vnpy体验却是另一回事——每天早上跑完选股脚本，A股股票池只剩下几十只值得盯的标的，策略一挂就是一个月。我拆这份vnpy二次开发资料包最直接的感受是：它把选股、回测、机器学习三个经常被分开玩的东西，串成了一根流水线，而且每一步都有能直接跑的代码和按文件编号排好的文档。适合谁？适合那种已经用vnpy下过单、却不知道怎么把一只股票的策略扩展到一篮子股票的人，也适合课程设计想拿“量化+机器学习”凑一套完整亮点的学生。你不需要精通C++，Python能写明白函数调用就能跟完。
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 # axis=1 表示按行做排名，pct=True 输出0~1的分位值 rps = ret.rank(axis=1, pct=True) return rps def stock_picker(close_df: pd.DataFrame, score_weights: dict) -> pd.Series: total_score = pd.Series(0, index=close_df.columns) for win, weight in score_weights.items(): rps = calc_rps(close_df, window=win) total_score = total_score + rps.iloc[-1] * weight return total_score.sort_values(ascending=False)参数上我建议至少算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 数据层的坑：时区、复权与停牌
坑1：vnpy的bar时间戳永远是UTC+8的datetime，但你从外部数据源拿到的日K可能是0点或16点。
现象：策略回测收益正常，实盘信号晚一天触发。
原因：外部数据源把日期存成了2024-01-01 00:00:00，而vnpy的BarData.datetime是按交易日收盘时间生成的。
解决：数据入库前统一对datetime做一次replace(hour=0, minute=0, second=0)，或者直接按日期字符串匹配，不要依赖时间戳相等判断。
坑2：前复权数据用错导致选股指标失真。
现象：RPS排名前十的股票里混进一堆刚除权的低价股。
原因：直接用未复权价格算20日涨幅，除权当天的价格跳空会把涨幅算成暴跌。
解决：日K一律用前复权，且算因子时前复权的价格序列必须是从上市日起连续复权的，不能用片段复权。
坑3：停牌股在回测里被当成可交易标的。
现象：换仓日选中的股票里有停牌股，回测时仓位照常买入，实盘根本挂不上单。
原因：回测器没有过滤当日status字段，把停牌也当成有bar数据。
解决：在信号生成时就把当日成交量小于一个阈值的标的剔除，成交量极低时即使没停牌也基本成交不了。
5.2 回测层的坑：未来函数与成交价
坑4：用当天收盘价计算信号，再用当天收盘价成交。
现象：回测胜率超过70%，实盘一月就打回原形。
原因：信号计算和成交用了同一个bar的close，等于你在收盘瞬间既知道结果又下了单。
解决：信号用第T根bar计算，成交价要么用第T+1根bar的开盘价，要么强制在信号中引入shift(1)，让成交永远落后信号一个bar。
坑5：滑点参数设成0还能自我安慰。
现象：回测资金曲线平滑上涨，实盘手续费和滑点一加就亏损。
原因：股票回测的滑点不是按价格固定百分比算的，而是按买一卖一价差幅度。
解决：把滑点从固定百分比改成“按当日振幅的0.1倍”估算，至少保留0.5%的缓冲，同时把手续费按卖出的万五来算，A股印花税只在卖出时收取，很多人漏掉这一项。
5.3 机器学习层的坑：标签泄漏与样本重叠
坑6：训练集和验证集按行随机切分，准确率虚高。
现象：模型在验证集上AUC 0.85，实盘策略一塌糊涂。
原因：同一只股票的相邻交易日样本高度相关，随机切分把同一段时间的数据同时放进了训练集和验证集。
解决：按连续时间段切分，直接用某一天作为切分线，之前的全部做训练，之后的全部做验证。
坑7：因子数据没做截面归一化，导致模型学到的是绝对价格水平。
现象：训练集里高价股占比高，模型推理时对高价股打分虚高。
原因：成交量、换手率这些因子在不同股票间的量纲差异太大，模型学到了绝对数量特征而不是相对强弱。
解决：每个交易日对所有股票的因子做一次横截面rank或zscore，让输入变成相对排名，这也是ML选股和传统因子选股最本质的一个区别。
6. 把ML信号接进vnpy：walk-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_estimators=200, max_depth=6) 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_years=3这个参数很关键，A股风格周期通常在2~3年左右，超过5年的数据反而会成为噪声。AUC阈值我一般设在0.55以上才让策略上线，低于这个数的模型即使胜率看着高，也大概率是过拟合了。
6.2 每日信号快照与模型版本核对
模型上线后最怕的是静默失效。我现在的日常流程是：每天收盘后把当日的模型输出分数存一份CSV快照，隔一周回来对比分数排名和实际收益的差异，同时记录模型文件名哈希值，防止不小心覆盖了旧模型文件。这份资料包里的check_result.py脚本做的就是这件事，它会把当日的信号快照和3日后的实际收益做一次相关性检验，我用它替代了每天肉眼盯资金曲线。
从那以后，我每次改策略都强制走一遍这个流程：先walk-forward验证，再对比信号快照和实际收益，最后才决定是否上实盘。这套习惯救了我很多次，希望帮到你。
本文还有配套的精品资源，点击获取
