企业官网建设流程全解析

简介：这是一份基于vnpy进行二次开发的量化交易实践资源，面向熟悉Python并希望深入金融量化领域的开发者、分析师与投资者，聚焦自动化选股、策略回测与机器学习模型集成。资源内容涵盖交易接口对接、数据管理、回测引擎扩展及性能优化等模块，结合C++扩展可满足高频计算场景需求，并提供了从数据获取到策略验证的完整路径。包内共1656个文件，以h与hpp头文件、cpp源文件、py脚本及ipynb笔记为主，辅以dll动态库和配置文件，压缩包整体约59.07MB，目录结构清晰，便于按模块定位与二次开发。机器学习部分演示了线性回归、支持向量机、随机森林等算法在价格预测与多因子选股中的应用，并涉及交叉验证、正则化防过拟合以及集成学习组合策略，有助于提升模型稳健性。资源已有598人学习，适合用于搭建自动化交易系统、扩展策略库并将AI技术落地到实际投资决策。

1. vnpy二次开发：选股、回测、机器学习三个模块怎么协同才不翻车

如果你已经用vnpy跑通过几套CTA策略，再看“基于vnpy的二次开发，选股、回测、机器学习”这个标题，大概率会有同一个困惑：vnpy原生不是已经带回测吗，为什么还要再开发一层？答案是，vnpy自带的CTA策略模板是单标的、单策略、信号即下单的模型，而股票选股天然是“先把几十只股票打分选出来，再统一分配资金”的组合决策。这个标题背后是一套以vnpy回测引擎为执行底座、上面挂因子选股和机器学习信号的综合方案，核心价值是让你不用换框架，就能把模型驱动的股票池决策接进成熟的回测环境。适合有Python基础、正从规则型CTA转向股票因子策略的量化从业者——新手能照步骤搭起来，熟手可以直接改参数和撮合逻辑。

2. 选股模块独立成池：因子打分脚本输出Top N候选池，再交给vnpy回测

2.1 为什么选股要独立在vnpy之外：数据口径与交易时点错位

vnpy的架构里，一个策略实例只绑定一个合约，它处理的是这个合约的K线事件，然后发单到网关。如果你把选股逻辑直接写进策略模板，每个标的行为例都会重复跑一遍全市场打分，数据量大时回测会明显变慢。更麻烦的是，选股经常要用的财务、市值数据和行情数据更新频率、复权口径都不一样，混在事件循环里处理，出的错往往很难查。

我一般把选股做成完全独立的任务：每天收盘后或每周跑一次，拉全市场日线加市值数据，算因子、打总分，输出一个Top N的候选池CSV。vnpy策略只负责“读池子、按白名单过滤、执行交易”。这样回测和实盘共用同一个池子文件，选股逻辑可以单独调参验证，完全不动vnpy核心。这个结构还有一个好处：后面换数据服务商或换因子，只改选股脚本，不影响交易策略。

2.2 因子打分选股的完整脚本：输入日线数据，输出Top N候选池

常见的做法是先把每只股票的历史日线落盘成CSV，再写一个打分脚本批量处理。下面的脚本就是候选池生成器的骨架，它按动量、波动率倒数、市值对数三个因子加权打分，输出得分最高的30只股票到candidate_pool.csv。

import pandas as pd import numpy as np from pathlib import Path # 假设 data/ 目录下每只股票一个 CSV， # 列：date, open, high, low, close, volume, amount, mv # mv 是流通市值，选股日必须用截至当天已经确认的数据 def load_stock(symbol: str) -> pd.DataFrame: df = pd.read_csv(f"data/{symbol}.csv", parse_dates=["date"]) return df.sort_values("date").reset_index(drop=True) def score_stock(df: pd.DataFrame, w_momentum: float, w_vol_inv: float, w_ln_mv: float) -> float: # 动量因子：过去20日涨跌幅，需要至少21根K线 momentum = df["close"].iloc[-1] / df["close"].iloc[-21] - 1 # 波动率倒数：过去20日日收益标准差越小，分数越高 daily_ret = df["close"].pct_change().tail(20) vol_inv = 1 / (daily_ret.std() + 1e-8) # 小市值偏好：市值取对数后参与打分 ln_mv = np.log(df["mv"].iloc[-1] + 1) return (w_momentum * momentum + w_vol_inv * vol_inv + w_ln_mv * ln_mv) symbols = [p.stem for p in Path("data").glob("*.csv")] rows = [] for sym in symbols: df = load_stock(sym) if len(df) < 60: # 数据量不足的股票排除 continue score = score_stock(df, 0.4, 0.3, 0.3) rows.append({"symbol": sym, "score": score}) pool = pd.DataFrame(rows).sort_values("score", ascending=False) pool.head(30).to_csv("candidate_pool.csv", index=False)

三个因子的权重0.4/0.3/0.3只是示例，含义分别是趋势、风险、市值偏好的强弱。动量因子捕捉上涨惯性，波动率倒数偏向稳定品种，市值对数则把注意力引向小盘。需要注意量纲问题：动量是百分比小数（如0.05），波动率倒数可能是几十甚至上百，市值对数是十几，直接加权会导致市值因子主导打分。正式环境里每个因子先做z-score标准化再加权，这个脚本只是跑通流程的最小版本。

2.3 候选池接入vnpy回测：白名单过滤与多标的循环

候选池生成后，回测阶段就是对池内每只股票逐个跑vnpy的BacktestingEngine。vnpy原生引擎是单标的的，组合回测就自己写循环，逐标的跑完再汇总统计。

from vnpy_ctastrategy.backtesting import BacktestingEngine # 取候选池前20只进入回测 candidates = pd.read_csv("candidate_pool.csv").head(20)["symbol"].tolist() total_returns = {} for sym in candidates: engine = BacktestingEngine() engine.set_parameters( vt_symbol=f"{sym}.XXXX", # 替换为真实股票代码和交易所后缀 interval=Interval.DAILY, # 日线即可，股票日频策略不需要分钟线 start=datetime(2023, 1, 1), end=datetime(2024, 12, 31), rate=0.0005, # 单边费用，含印花税分摊 slippage=0.02, # 保守滑点 size=100, # A股一手100股 pricetick=0.01, capital=1_000_000 # 单标的初始资金设为等额 ) engine.add_strategy(MyStrategy, {}) engine.load_data() engine.run_backtesting() df = engine.calculate_result() total_returns[sym] = df.loc["总收益率", "result"] # 等权组合收益 = 各标的收益的平均 print("组合年化收益约", sum(total_returns.values()) / len(total_returns))

注意engine要在循环里每次重新创建，不能复用同一个实例。vnpy的BacktestingEngine会在run_backtesting时累积内部状态，重复使用会把上次的持仓和信号带进下一只股票，结果完全失真。另外，set_parameters里的vt_symbol后缀要正确，股票回测一般用交易所后缀区分，写错会导致load_data拉不到数据而静默返回空K线，回测结果全是0。这类问题不会报错，只能靠检查净值曲线发现。

提示：回测前先确认数据服务商提供的区间覆盖start到end。区间不足时vnpy不会提醒，只会返回空数据。

3. 回测模块二次开发：日频组合回测的撮合循环与三个必调参数

3.1 vnpy回测引擎的撮合假设：价格成交逻辑在股票日频上的误差

vnpy CTA回测是事件驱动，每根K线推给策略的on_bar，策略发单后引擎按bar的价格加减滑点撮合。CTA常见的是分钟级或tick级数据，成交价格误差有限；但股票日频策略直接用日bar的收盘价撮合，等于“看到了收盘价再下单，却按收盘价成交”。这是最典型的前视偏差来源，回测收益会明显虚高，越短周期的因子虚高越严重。

所以股票日频回测要做一处本地改造：信号在T日收盘后生成，成交放到T+1日开盘价。vnpy自带的BacktestingEngine没有直接暴露这种“次日开盘撮合”的开关，二次开发就是绕开CTA引擎的默认撮合，写一个适合股票调仓频率的循环。这个循环不需要处理tick级小数，逻辑简单很多。

3.2 改成日频调仓回测：T日收盘信号、T+1开盘撮合的最小循环

下面是我常用的日频组合回测循环。它同时处理了三个关键点：信号只用到T日之前的数据、成交用次日开盘价、加了一字板和换手过滤。这个循环可以独立跑，也可以作为改造vnpy回测的参考骨架。

import pandas as pd from pathlib import Path def run_rotation(model, pool, data_dir: Path, fee_rate=0.0005, min_turnover=0.05): # 每只股票一个日线CSV，索引为date，含 open/high/low/close/amount daily = {sym: pd.read_csv(data_dir / f"{sym}.csv", index_col="date", parse_dates=True) for sym in pool} trade_dates = pd.date_range("20230101", "20241231", freq="B") cash = 1_000_000 qty = {sym: 0 for sym in pool} # 持仓股数 cur_weights = {sym: 0.0 for sym in pool} nav = pd.Series(dtype=float) for date in trade_dates: # 决策：周一收盘后，用截止到上周五的数据出信号 if date.weekday() == 0: signal_weights = dict(cur_weights) # 复制一份，避免影响当前持仓 for sym in pool: hist = daily[sym].loc[:date].iloc[:-1] # 剔除当天，防未来函数 if len(hist) < 21: continue x = build_features(hist) # 特征函数与训练时保持一致 prob = model.predict_proba(x)[0, 1] signal_weights[sym] = 1.0 if prob >= 0.6 else 0.0 # 换手过滤：目标权重和当前权重差距过小就不动 turnover = sum(abs(signal_weights.get(s, 0) - cur_weights.get(s, 0)) for s in set(signal_weights) | set(cur_weights)) if turnover < min_turnover: signal_weights = cur_weights # 撮合：用当日开盘价成交，一字板不成交 for sym in pool: target_w = signal_weights.get(sym, 0) if target_w <= 0: continue row = daily[sym].loc[date] if row["open"] == row["high"] == row["low"]: continue # 一字板实际买不进 target_qty = int(cash * target_w / row["open"] / 100) * 100 delta_qty = target_qty - qty[sym] if delta_qty > 0: cash -= delta_qty * row["open"] * (1 + fee_rate) qty[sym] += delta_qty elif delta_qty < 0: cash += (-delta_qty) * row["open"] * (1 - fee_rate) qty[sym] += delta_qty # 收盘后按收盘价记录组合净值 equity = cash + sum(qty[s] * daily[sym].loc[date, "close"] for s in pool if qty[s] > 0) nav.loc[date] = equity cur_weights = {s: qty[s] * daily[sym].loc[date, "close"] / equity for s in pool} return nav

撮合顺序是这个循环的核心：周一收盘后才有完整数据算出信号，周二开盘去成交，这和你手工交易的时间线完全一致。build_features必须是你在训练模型时用的同一套特征构造逻辑，否则推理输入错位，模型输出没有意义。一字板过滤用open == high == low做近似判断，停牌日则需要额外维护一个停牌名单，这里没展开。

3.3 三个必调参数：手续费率、滑点、最小换手过滤

股票日频策略一年换手几十次，手续费和滑点的设定直接决定回测结果靠不靠谱。我一般固定用下面这组起点参数，再根据实际品种微调。

换手过滤的代码很短，但很多新手会漏掉：

turnover = sum(abs(new_weights[s] - old_weights[s]) for s in pool) if turnover < min_turnover: new_weights = old_weights # 差异太小，干脆不操作

回测报告里的换手率也要重点看，不是只看收益。如果策略年化收益高但换手率几百倍，把费率调高后收益塌掉，说明策略本质是在赚交易频率的钱，实盘会被成本吞噬。

4. 机器学习决策接入策略：特征、标签与模型推理的完整链路

4.1 特征与标签怎么构造：未来N日收益当监督信号，时间对齐是核心

机器学习在选股里的角色不是预测股价，而是预测“未来一段时间是否上涨”。常见做法是把未来5日收益率大于某个阈值的样本标为正类，让模型学习特征到涨跌概率的映射。为什么用5日而不是1日？单日收益噪声太大，模型学到的大多是噪声；5日能过滤掉部分日内反转噪音，正样本比例也更稳定。

阈值取2%是一个相对保守的门槛，不同市场环境下正样本占比会浮动，训练前先打印一下label的均值确认类别平衡度。特征统一用公开日线就能算出来的：动量、波动率、成交金额均值和市值对数。好处是数据获取成本低，坏处是信号同质化严重——大家都在用这些因子，模型提升空间有限。先跑通链路，再逐步加独有数据源。

4.2 训练脚本：按时间切分样本，单股票内部算特征防泄漏

训练脚本要做两处强制约束：特征和标签必须在单只股票内部计算，防止pandas的shift和pct_change跨股票串行；训练集和验证集按时间顺序切分，绝不随机打乱。

import pandas as pd import numpy as np import joblib from lightgbm import LGBMClassifier HORIZON = 5 # 未来5日 THRESHOLD = 0.02 # 5日收益超2%标记为正样本 def add_features(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() df["ret_5"] = df["close"].pct_change(5) df["ret_20"] = df["close"].pct_change(20) df["vol_20"] = df["close"].pct_change().rolling(20).std() df["amount_ma5"] = df["amount"].rolling(5).mean() df["ln_mv"] = np.log(df["mv"] + 1) return df def add_label(df: pd.DataFrame) -> pd.DataFrame: # 标签必须在单只股票内部 shift，防止跨股票串数据 future_ret = df["close"].shift(-HORIZON) / df["close"] - 1 df["label"] = (future_ret > THRESHOLD).astype(int) return df frames = [] for sym in symbols: # symbols 由第2章脚本提供 raw = load_stock(sym) df = add_features(raw).dropna() df = add_label(df).dropna() df["symbol"] = sym frames.append(df) all_data = pd.concat(frames, ignore_index=True) all_data = all_data.sort_values("date", ignore_index=True) # 按时间排序 # 前70%时间段训练，后30%时间段验证 split_idx = int(len(all_data) * 0.7) train = all_data.iloc[:split_idx] test = all_data.iloc[split_idx:] feature_cols = ["ret_5", "ret_20", "vol_20", "amount_ma5", "ln_mv"] model = LGBMClassifier(n_estimators=300, learning_rate=0.05, num_leaves=15, random_state=42) model.fit(train[feature_cols], train["label"]) importances = pd.Series(model.feature_importances_, index=feature_cols).sort_values(ascending=False) print(importances) joblib.dump({"model": model, "feature_cols": feature_cols, "train_cutoff": all_data["date"].iloc[split_idx]}, "stock_model.pkl")

这里最关键的是concat之后重新按日期排序，再做前70%切分。如果不排序，concat顺序是“先股票A全时间段再股票B全时间段”，前70%行对应的不是早期时间，而是前一批股票的所有数据，验证集和训练集会混入同时间段的不同股票，回测时模型相当于偷偷见过验证期附近的横截面信息。n_estimators=300对日频数据偏高，通常先试100，观察验证集AUC不再上升就停。num_leaves=15是限制模型复杂度的手段，叶子太少欠拟合，太多就过拟合，15到31之间是常见搜索区间。

4.3 策略里加载模型做推理：决策日才算特征，概率阈值转目标仓位

训练完成后，模型文件放进vnpy策略目录。策略在on_init时一次性加载模型，不要把joblib.load写在on_bar里，否则每根K线都做一次磁盘IO，回测会慢到无法接受。下面是嵌入策略的骨架。

from collections import deque from vnpy_ctastrategy import StrategyTemplate class MLStockStrategy(StrategyTemplate): model_file = "stock_model.pkl" predict_threshold = 0.6 def on_init(self): pkl = joblib.load(self.model_file) self.model = pkl["model"] self.feature_cols = pkl["feature_cols"] self.history = deque(maxlen=40) # 40根K线足够算20日特征 def on_bar(self, bar: BarData): self.history.append(bar) if len(self.history) < 21: return # 只在周一出信号，其余交易日沿用上次仓位 if bar.datetime.weekday() != 0: return df = bars_to_df(self.history) # 自己实现的转换，保留 close/amount/mv x = build_features(df)[-1:][self.feature_cols] prob = self.model.predict_proba(x)[0, 1] target = self.predict_threshold if prob >= self.predict_threshold else 0.0 self.set_target_weight(target) def set_target_weight(self, w: float): # 简化版权重管理：把当前持仓调整到目标权重 w # 内部按总资金和目标价折算股数，再调用 buy/sell 发单 target_qty = int(self.capital * w / self.last_price / 100) * 100 diff = target_qty - self.pos if diff > 0: self.buy(self.last_price, diff) elif diff < 0: self.sell(self.last_price, -diff)

bars_to_df是自己写的一个小工具函数，把deque里的BarData对象转成DataFrame，只保留close、amount、mv等列。注意特征构造的输入窗口长度：deque最大40，但build_features里pct_change(20)至少要21根，留足裕量。行为上“周一推理，平时沿用”把推理次数压缩到每周一次，回测速度提升明显，也更接近实际操盘节奏——普通人不会每天都调仓。

5. 二次开发避坑：5个让回测漂亮实盘崩盘的问题及排查

回测做得漂亮、实盘就崩，几乎是股票策略开发里的默认剧本。下面五条是我排查过很多次后归纳出来的，按“现象→原因→解决”列出，方便直接对照检查。

5.1 未来函数：标签错位让回测年化虚高

现象：回测年化40%以上，最大回撤不到5%，资金曲线接近直线，实盘却完全对不上。

原因：最常见的错位是“当日收盘价”同时用于计算特征和作为成交价。你在T日收盘后算完信号，却按T日收盘价成交，等于用已知信息交易。另一个隐蔽来源是标签构造时用了shift(-5)，但没有保证shift是在单只股票内部完成的，导致跨股票串数据。

解决：统一时间对齐规则——特征严格用T日及以前的数据，信号在T日收盘后生成，成交用T+1开盘价。在回测循环里加一行调试输出，把“T日信号、T+1成交价、T+1成交数量”打印出来，核对三个时点是否错位，比看任何指标都直接。

5.2 复权口径不一致：除权日价格对不上账面

现象：某只股票回测明细里，除权除息日前后出现异常跳空，买卖价格和K线显示价格对不上。

原因：数据源默认给前复权，选股脚本用的CSV和后端数据接口用的可能是两套复权口径。除权日当天，前复权的历史价格被整体修正，而成交价如果取自另一套未复权数据，就会在断层上成交。

解决：全链路统一使用后复权（或统一前复权），选股、训练、回测读同一份数据文件。除权日附近强制不调仓，避免在价格断层上成交。每次更换数据源后，随机抽几只股票检查除权日的价格连续性。

5.3 涨跌停和停牌：理想撮合带来的虚假成交

现象：小盘股策略回测胜率极高，实盘却很平庸——涨停买不进、跌停卖不出、停牌根本无交易。

原因：vnpy默认撮合假设任何价格都能成交，实际上一字涨停时挂单排在队伍末尾根本成交不了，跌停时卖单同样出不去。

解决：在撮合循环里加一字板判断，开盘价、最高价、最低价相等时跳过当天调仓。停牌日需要单独维护名单，数据服务商一般会提供停复牌标记，拉取后存成字典，撮合前先查一遍。

5.4 模型推理拖慢回测：特征重复计算的黑匣子

现象：明明只有20只股票，日频策略，回测却跑了一晚上。

原因：特征在每个交易日对每只股票重复计算，模型也在每根K线里反复加载。很多人把joblib.load写进on_bar，回测性能直接崩掉，这就是个黑匣子，表面上看不出逻辑错误，就是慢。

解决：特征计算结果用字典缓存，key是“日期+股票代码”，同一天重复使用；模型在on_init里加载进内存；推理只放在决策日，非决策日沿用上次信号。以上三点做到，回测时间能快一到两个数量级。

5.5 过拟合：样本内惊艳、样本外打回原形

现象：训练集收益神级，验证集立刻衰退，怎么调参都救不回来。

原因：最常见的是随机切分数据集——随机洗牌后训练集和测试集包含重叠时间段，模型直接记住了时间段噪声。其次是不加约束的超参搜索，几百组参数里总能挑出一组样本内漂亮的。

解决：严格按时间顺序切分，验证集时间必须晚于训练集；滚动验证用连续多个时间段反复测试，只保留每个时间段都稳定的参数；超参搜索控制在3到5组，别追求样本内最优。

6. 进阶：滚动训练与特征漂移预警，让模型不是一次性玩具

6.1 滚动训练与模型版本自动加载

静态训练出的模型放到第二年会明显失效，滚动训练是让机器学习模块可长期运行的底线。常见做法是每月最后一个交易日，用最近12个月数据重训模型，文件名带训练日期（如stock_model_20250131.pkl），策略每天检查模型目录，发现新文件就重新加载。

latest_model = max(MODEL_DIR.glob("stock_model_*.pkl"), key=lambda p: p.stat().st_mtime) if latest_model != self.current_model: self.model = joblib.load(latest_model) self.current_model = latest_model

这个懒加载方式不用重启vnpy进程，训练脚本单独跑，策略自动热更新。不同训练日期的模型不要覆盖同一个文件，保留历史模型文件，方便回测时对比“用旧模型和用新模型的结果差异”。

6.2 特征漂移预警：模型该重新训练的信号

模型失效前通常有先兆：top特征分布和训练集越差越远。训练时把各特征的均值train_mean和标准差train_std存成json，策略实时计算当前特征相对训练集的z-score，最大值超过2就说明市场风格可能变了。

drift = (current_features - train_mean) / train_std if drift.abs().max() > 2: self.target_weight = 0.0 # 特征漂移过大，先降仓等重训

我自己的习惯是模型训练时间超过90天就强制停用信号，宁可空仓也不硬做。滚动训练、漂移预警、到期降仓这三件事配合起来，机器学习选股才算真正脱离了“回测黑匣子”阶段，变成一个可持续维护的生产组件。这个坑吃多了之后就明白：模型不重训，收益曲线迟早给你上一课。希望帮到你。

本文还有配套的精品资源，点击获取