简介这份资源围绕vnpy量化框架的二次开发展开面向具备一定Python基础、希望深入选股、回测与机器学习方向的在校学生、教师及企业开发者也可作为毕业设计、课程设计或项目立项的参考范例。压缩包共1659个文件约59.1MB以C头文件与源码h、hpp、cpp构成底层交易接口Python脚本负责策略与业务逻辑另含Jupyter笔记本、npy数据文件及dll动态库等覆盖从行情接入到策略执行的完整链路。内容预览可见ctp、sgit、xgj等多路交易接口的封装实现便于理解多柜台接入与二次开发思路。目前已有188人学习关注。项目源码经测试可运行并附详细文档读者可据此掌握选股因子构建、回测流程编排与机器学习模型融合的实践方法也能在现有代码上修改扩展快速迁移到自身课题或作业场景中。1. 从一份 vnpy 二次开发包说起选股、回测、机器学习到底怎么串起来很多人第一次接触量化都是被“选股公式胜率 98%”这类标题带进来的结果下载一堆指标源码装进通达信里跑两天就发现——信号是滞后的回测是过拟合的实盘是亏钱的。这份基于 vnpy 的二次开发资料包走的是另一条路它不给你一个“妖股选股公式源码”而是把选股、回测、机器学习三个环节用 Python 串成一条可复现的流水线。vnpy 本身是国内用得比较多的量化交易框架事件驱动引擎、数据容器、回测撮合都给你搭好了二次开发的重点就落在“怎么把自己的选股逻辑和机器学习模型塞进去”。这份资料适合两类人一是已经会写 Python、想从 backtrader 多股回测那种脚本式框架迁移到工程化框架的从业者二是做机器学习项目、想把模型落到金融时序数据上的学生或工程师。它解决的核心问题是——让选股信号不再停留在公式层面而是能进入一个带撮合、带滑点、带手续费的完整回测环境里验证。2. vnpy 二次开发的环境搭建与模块拆解2.1 为什么选 vnpy 而不是 backtraderbacktrader 多股回测写起来确实顺手一个 Cerebro 对象把数据、策略、分析器全装进去几十行就能跑出一个多标的回测。但它的定位偏向“研究工具”事件循环是同步的撮合逻辑相对简化滑点和手续费的建模粒度不够细。vnpy 的定位是“交易系统”核心是 EventEngine——一个基于队列的事件驱动引擎行情、订单、成交、持仓都是事件策略订阅事件后做出反应。这个差异在回测阶段就能体现出来vnpy 的 BacktestingEngine 会模拟限价单排队、部分成交、撤单重发这些细节回测出来的曲线更接近实盘。二次开发的空间也在这里。vnpy 的模块化程度高数据层用 BaseDatafeed 抽象策略层用 CtaTemplate 抽象你只需要继承对应的基类把选股逻辑和机器学习推理逻辑填进去就行。资料包里应该包含了完整的模块目录结构常见做法是分成 data、strategy、ml、backtest 四个子包每个子包各司其职。2.2 环境搭建的完整步骤先确认 Python 版本。vnpy 对 Python 版本有要求常见做法是用 3.10 或 3.11太新的版本某些 C 扩展可能还没适配。建议用 conda 建独立环境避免和系统里的其他包冲突。# 创建独立环境Python 版本按 vnpy 当前稳定版要求选 conda create -n vnpy_dev python3.10 conda activate vnpy_dev # 安装 vnpy 核心包用 pip 而不是 conda避免依赖解析差异 pip install vnpy # 安装回测和数据分析常用依赖 pip install pandas numpy ta-lib scikit-learn matplotlib # 如果 ta-lib 安装报错先装 C 库再装 Python 包 # macOS: brew install ta-lib # Ubuntu: sudo apt-get install ta-lib装完之后验证一下核心模块能不能导入# 验证 vnpy 核心模块和回测引擎是否可用 from vnpy.event import EventEngine from vnpy.trader.engine import MainEngine from vnpy_ctastrategy.backtesting import BacktestingEngine # 如果能走到这里不报错说明基础环境没问题 print(vnpy core modules loaded)参数说明EventEngine 是全局事件引擎MainEngine 是主引擎负责加载各功能模块BacktestingEngine 来自 vnpy_ctastrategy 包是专门做 CTA 策略回测的。注意 vnpy 从某个版本开始把策略模块拆成了独立包安装 vnpy 之后可能还需要单独 pip install vnpy_ctastrategy具体看资料包里的 requirements.txt。2.3 资料包的目录结构怎么读拿到一个二次开发包先别急着跑代码把目录结构看清楚能省很多时间。常见做法是目录作用先看哪个文件data/数据加载与清洗脚本loader.py 看数据格式strategy/选股与交易策略stock_selector.py 看选股逻辑ml/机器学习模型训练与推理train.py 看特征工程backtest/回测配置与执行入口run_backtest.py 看参数docs/详细文档先读 README 和配置说明重点看 strategy/ 下的选股逻辑是怎么和 vnpy 的 CtaTemplate 对接的。通常做法是选股模块先跑出一个股票池然后策略模块对池子里的标的逐个加载历史数据、跑回测。这个“先选后测”的流程和集合竞价选股 Python 脚本的思路类似但工程化程度高很多。3. 选股模块的二次开发从公式到可回测信号3.1 选股逻辑的代码化改造通达信选股公式胜率 98% 这种话做过实盘的人都知道是怎么回事——公式在历史数据上拟合出来的信号换一段行情就失效。把公式改成 Python 代码第一步不是翻译公式而是想清楚这个信号的经济含义。比如“涨停缩量回调选股”本质是在找强势股短期回调后的二次启动点那你要定义的是涨停怎么判定、缩量怎么度量、回调幅度和时间的边界在哪。import pandas as pd import numpy as np def select_pullback_stocks(df: pd.DataFrame, limit_up_pct: float 0.095, shrink_ratio: float 0.6, pullback_days: int 3) - list: 涨停缩量回调选股逻辑 df: 包含 code, date, close, volume, pct_chg 的 DataFrame limit_up_pct: 涨停判定阈值主板用 0.095创业板用 0.195 shrink_ratio: 缩量比例当日成交量 / 涨停日成交量 pullback_days: 回调观察天数 selected [] for code, group in df.groupby(code): group group.sort_values(date).reset_index(dropTrue) for i in range(len(group) - pullback_days): # 找到涨停日 if group.loc[i, pct_chg] limit_up_pct * 100: limit_vol group.loc[i, volume] # 检查后续几天是否缩量回调 pullback group.loc[i1:ipullback_days] if len(pullback) pullback_days: continue avg_vol pullback[volume].mean() price_drop (group.loc[i, close] - pullback[close].min()) / group.loc[i, close] if avg_vol / limit_vol shrink_ratio and 0.02 price_drop 0.12: selected.append(code) break return selected逻辑说明先按股票代码分组在每组里找涨停日然后检查涨停后若干天的量价关系。缩量比例 shrink_ratio 设 0.6 意味着回调期平均成交量不到涨停日的六成价格回调幅度控制在 2% 到 12% 之间。这两个参数是经验值不同市场阶段需要调整。注意涨停判定阈值要区分板块主板 10%、创业板和科创板 20%资料包里如果有参数配置文件优先读配置而不是硬编码。3.2 把选股结果接入 vnpy 数据层选股模块跑出来的是一堆股票代码接下来要让 vnpy 能加载这些标的的历史数据。vnpy 的数据层用 BaseDatafeed 抽象常见做法是继承这个基类实现 load_bar_data 方法从本地 CSV 或数据库里读数据。from vnpy.trader.datafeed import BaseDatafeed from vnpy.trader.object import BarData from vnpy.trader.constant import Exchange, Interval import pandas as pd class LocalCsvDatafeed(BaseDatafeed): 从本地 CSV 加载历史数据适配 vnpy 回测引擎 def __init__(self, data_dir: str): self.data_dir data_dir def load_bar_data(self, symbol, exchange, interval, start, end): # 按 symbol 读取对应 CSV列名对齐 vnpy 要求 path f{self.data_dir}/{symbol}.csv df pd.read_csv(path, parse_dates[datetime]) df df[(df[datetime] start) (df[datetime] end)] bars [] for _, row in df.iterrows(): bar BarData( symbolsymbol, exchangeexchange, datetimerow[datetime], intervalinterval, open_pricerow[open], high_pricerow[high], low_pricerow[low], close_pricerow[close], volumerow[volume], gateway_nameLOCAL ) bars.append(bar) return bars参数说明symbol 是股票代码exchange 用 Exchange.SSE 或 Exchange.SZSEinterval 回测日线用 Interval.DAILY。BarData 的字段名要和 vnpy 定义一致open_price 不是 open这个容易写错。数据目录结构建议按 symbol.csv 命名每个文件包含 datetime、open、high、low、close、volume 六列datetime 格式用 ISO 标准。3.3 选股与回测的衔接方式选股模块和回测模块的衔接有两种常见做法。一种是“静态池”选股脚本先跑一遍把股票池写进配置文件回测时只加载池子里的标的。另一种是“动态池”在策略的 on_bar 里实时判断是否满足选股条件满足就下单。静态池适合日线级别的中低频策略动态池适合日内或高频场景。资料包里大概率用的是静态池方案因为机器学习模型训练需要固定的标的集合动态池会导致训练集和回测集不一致。具体做法是在 run_backtest.py 里读一个 stock_pool.json然后循环加载每个标的的数据跑回测。这个流程和“选股魔法三步点金”那种黑盒公式的区别在于——每一步都是可检查、可修改、可复现的。4. 机器学习模型在 vnpy 回测中的落地4.1 特征工程从行情数据到模型输入机器学习中的数据处理是什么放到量化场景里就是把 OHLCV 这些原始行情数据转成模型能吃的特征矩阵。常见做法是构造三类特征动量类N 日收益率、均线偏离度、波动类ATR、布林带宽度、量能类换手率、量比。资料包里如果有 ml/feature.py重点看它用了哪些特征、窗口参数怎么设的。def build_features(df: pd.DataFrame, windows: list [5, 10, 20]) - pd.DataFrame: 构造机器学习特征矩阵 windows: 滚动窗口列表分别对应短中长周期 features pd.DataFrame(indexdf.index) for w in windows: # 动量特征过去 w 日收益率 features[fret_{w}d] df[close].pct_change(w) # 均线偏离收盘价 / w 日均线 - 1 features[fma_dev_{w}d] df[close] / df[close].rolling(w).mean() - 1 # 波动特征w 日收益率标准差 features[fvol_{w}d] df[close].pct_change().rolling(w).std() # 量能特征当日成交量 / w 日均量 features[fvol_ratio_{w}d] df[volume] / df[volume].rolling(w).mean() # 标签未来 5 日收益率是否为正 features[label] (df[close].shift(-5) / df[close] - 1 0).astype(int) return features.dropna()逻辑说明windows 参数控制特征的时间尺度5 日捕捉短期动量20 日捕捉中期趋势。标签构造用未来 5 日收益率方向这是一个二分类问题。注意 shift(-5) 会产生未来数据在实盘推理时不能用训练阶段可以但回测时要确保模型只用到当前时刻之前的数据。这个坑在机器学习项目里非常常见资料包里如果有 walk-forward 验证的代码重点看它怎么处理时间边界。4.2 模型训练与滚动预测金融时序数据不能用随机划分训练集和测试集必须按时间顺序切分。常见做法是 walk-forward用前 N 年数据训练预测第 N1 年然后窗口向前滚动。资料包里如果用了 sklearn 的 TimeSeriesSplit说明作者意识到了这个问题。from sklearn.ensemble import GradientBoostingClassifier from sklearn.metrics import accuracy_score import numpy as np def walk_forward_train(features: pd.DataFrame, train_years: int 3, retrain_freq: int 60): 滚动训练与预测 train_years: 训练窗口长度年 retrain_freq: 每隔多少个交易日重新训练一次 feature_cols [c for c in features.columns if c ! label] dates features.index.get_level_values(date).unique().sort_values() predictions [] for i in range(train_years * 252, len(dates), retrain_freq): train_end dates[i] train_start dates[max(0, i - train_years * 252)] train_mask (features.index.get_level_values(date) train_start) \ (features.index.get_level_values(date) train_end) test_mask (features.index.get_level_values(date) train_end) \ (features.index.get_level_values(date) dates[min(i retrain_freq, len(dates)-1)]) X_train features.loc[train_mask, feature_cols] y_train features.loc[train_mask, label] X_test features.loc[test_mask, feature_cols] if len(X_train) 100 or len(X_test) 0: continue model GradientBoostingClassifier(n_estimators100, max_depth3, random_state42) model.fit(X_train, y_train) pred model.predict_proba(X_test)[:, 1] for idx, p in zip(features.loc[test_mask].index, pred): predictions.append({date: idx[1], code: idx[0], prob: p}) return pd.DataFrame(predictions)参数说明train_years 设 3 是经验值太短模型学不到规律太长会引入过时模式。retrain_freq 设 60 意味着每季度重新训练一次平衡模型时效性和计算成本。GradientBoostingClassifier 的 max_depth 设 3 是防止过拟合金融数据信噪比低树太深容易记住噪声。注意 predict_proba 返回的是概率回测时可以用概率阈值过滤信号比如只交易 prob 0.6 的标的。4.3 把模型预测接入 vnpy 策略模型跑出来的是一张 date-code-prob 的表接下来要在 vnpy 策略的 on_bar 里读这张表根据概率决定是否下单。常见做法是把预测结果存成 CSV策略初始化时加载到内存on_bar 里按日期和代码查表。from vnpy_ctastrategy import CtaTemplate from vnpy.trader.object import BarData, TickData class MLStrategy(CtaTemplate): 基于机器学习预测概率的选股策略 prob_threshold 0.6 # 概率阈值高于此值才开仓 fixed_size 100 # 每次下单股数 def __init__(self, engine, strategy_name, vt_symbol, setting): super().__init__(engine, strategy_name, vt_symbol, setting) self.predictions {} # {date: {code: prob}} def on_init(self): # 加载模型预测结果 import pandas as pd df pd.read_csv(ml_predictions.csv, parse_dates[date]) for _, row in df.iterrows(): date_key row[date].strftime(%Y-%m-%d) if date_key not in self.predictions: self.predictions[date_key] {} self.predictions[date_key][row[code]] row[prob] self.write_log(ML predictions loaded) def on_bar(self, bar: BarData): date_key bar.datetime.strftime(%Y-%m-%d) code bar.symbol if date_key not in self.predictions: return prob self.predictions[date_key].get(code, 0) # 概率高于阈值且无持仓时开仓 if prob self.prob_threshold and self.pos 0: self.buy(bar.close_price, self.fixed_size) # 概率低于阈值且有多头持仓时平仓 elif prob 0.4 and self.pos 0: self.sell(bar.close_price, abs(self.pos))逻辑说明on_init 里加载预测结果on_bar 里按日期和代码查概率。prob_threshold 控制开仓门槛0.4 是平仓阈值形成不对称的进出场规则。fixed_size 是固定手数实盘时应该改成按资金比例动态计算。注意 vnpy 的 CtaTemplate 有 pos 属性记录当前持仓buy 和 sell 是内置的下单方法参数是价格和数量。5. 回测跑通之后避坑与常见问题排查5.1 回测曲线漂亮但实盘不成交现象回测年化收益 50% 以上实盘挂单经常不成交滑点吃掉大部分利润。原因回测时用了市价单逻辑或者限价单假设总能成交但实盘里流动性差的标的挂单可能排很久。解决在 BacktestingEngine 里设置合理的滑点和手续费参数限价单用排队模型不要用“成交价 收盘价”这种简化假设。资料包里如果有 backtest/setting.json检查 slippage 和 commission 字段。5.2 机器学习模型在回测中用了未来数据现象模型准确率 70% 以上回测收益曲线平滑向上但实盘一上就亏。原因特征工程里用了未来信息比如用全样本的均值和标准差做标准化或者标签构造时 shift 方向写反。解决所有滚动计算必须用 expanding 或 rolling 窗口标准化参数只能从训练集计算然后应用到测试集。walk-forward 验证是底线不能省。5.3 vnpy 数据加载报字段缺失现象回测启动时报 KeyError 或 AttributeError提示 BarData 缺少某个字段。原因本地 CSV 的列名和 vnpy 要求的字段名不一致比如用了 open 而不是 open_price或者 datetime 格式不是 ISO 标准。解决在数据加载脚本里做列名映射datetime 用 pd.to_datetime 统一转换。资料包里如果有 data/loader.py重点看它的列名映射逻辑。5.4 选股池太小导致回测过拟合现象选股模块筛出来的股票只有十几只回测收益很高但换一批数据就失效。原因选股条件太苛刻样本量不足统计上不显著。解决放宽选股阈值把股票池扩大到至少 50 只以上或者用滚动选股的方式动态调整池子。机器学习模型尤其需要足够的样本量几十个样本训出来的模型没有泛化能力。5.5 回测速度慢到无法接受现象一个包含 100 只股票、5 年日线数据的回测跑了几个小时还没结束。原因on_bar 里做了重复计算或者数据加载用了逐行循环而不是向量化。解决把特征计算和模型推理提前到回测之前回测时只查表数据加载用 pandas 向量化操作避免 iterrows。vnpy 的回测引擎本身是事件驱动的单标的回测速度不慢慢通常是策略逻辑写得低效。6. 进阶技巧用参数敏感性分析验证策略稳健性回测跑通只是第一步真正决定策略能不能上实盘的是参数敏感性。一个策略如果对参数极度敏感——比如均线周期从 20 改成 21 收益就腰斩——那它大概率是过拟合的。我一般会做三件事参数扫描、样本外验证、蒙特卡洛模拟。参数扫描用网格搜索但不要暴力穷举。以选股模块的 shrink_ratio 和 pullback_days 为例各取 3 到 5 个值组合起来跑回测看收益和回撤的分布。如果最优参数附近收益变化平缓说明策略稳健如果只有某一个点收益特别高周围全是亏损那就是拟合出来的。import itertools import json def param_sensitivity(base_config: dict, param_grid: dict): 参数敏感性扫描 base_config: 基础配置 param_grid: {参数名: [候选值列表]} keys param_grid.keys() results [] for values in itertools.product(*param_grid.values()): config base_config.copy() config.update(dict(zip(keys, values))) # 调用回测入口返回收益、回撤、夏普 metrics run_backtest(config) results.append({**config, **metrics}) df pd.DataFrame(results) # 按夏普比率排序看最优参数附近的分布 df df.sort_values(sharpe, ascendingFalse) return df # 使用示例 base {limit_up_pct: 0.095, shrink_ratio: 0.6, pullback_days: 3} grid {shrink_ratio: [0.4, 0.5, 0.6, 0.7, 0.8], pullback_days: [2, 3, 4, 5]} result_df param_sensitivity(base, grid) print(result_df.head(10))样本外验证的做法是留出最近一年的数据完全不参与训练和调参等策略在训练集上定型后拿这一年数据跑一遍。如果样本外收益和样本内差距在 30% 以内可以认为策略有一定泛化能力如果样本外直接亏损那就回炉重造。蒙特卡洛模拟则是把交易序列随机打乱看收益分布的置信区间这个用 numpy 的 random.permutation 就能做。还有一个容易被忽略的点回测时的资金曲线要按日频对齐不能只看总收益。vnpy 的 BacktestingEngine 会输出每日盈亏用 pandas 重采样成周频或月频看回撤持续时间和恢复周期。如果最大回撤持续了半年以上实盘时很少有人能扛住。从那以后我每次拿到一个新的选股逻辑或机器学习模型都强制走一遍参数敏感性加样本外验证哪怕回测曲线再漂亮也不跳过。希望帮到你。本文还有配套的精品资源点击获取