企业官网建设流程全解析

简介：基于vnpy的二次开发综合实践包，面向量化交易开发者、金融科技从业者及有Python基础的投资者，重点解决自动化选股、策略回测与机器学习落地难题。包内共1656个文件，解压后约59MB，其中299个Python脚本覆盖数据获取、策略编写与模型调用，217个C++源文件与639个头文件用于底层逻辑和高性能计算，407个头文件支撑多类交易接口，49个动态库便于直接链接调用，另有少量模型、配置文件与说明，目录结构清晰，便于二次开发。目前已有596人学习。内容涉及CTP、IB等行情接入与数据库管理，回测引擎自带收益曲线、夏普比率、最大回撤等统计指标；机器学习部分展示线性回归、SVM、随机森林及多模型集成选股方法，并以C++扩展加速矩阵运算，可据此构建完整自动化交易系统。

1. 基于vnpy的二次开发：选股、回测和机器学习到底在做什么

拿到这套基于vnpy的二次开发包，等于把量化研究的三个环节焊成了一条流水线：先用因子打分或机器学习模型把全市场几千只股票压缩成几十只的股票池，再拿vnpy的事件驱动回测引擎验证策略参数和历史绩效，最后把训练好的模型预测值接回vnpy的CtaTemplate生成交易信号。这个方向解决的核心痛点是，vnpy原生能力偏CTA单品种，股票的批量选股、组合回测和机器学习信号接入都得靠自己二次开发补齐。适合有Python基础、熟悉pandas、想用机器学习做选股但又不想从零搭框架的团队或个人。下面按我实际落地的顺序，把每一块的扩展点、关键参数和踩过的坑讲清楚。

2. 二次开发的骨架：vnpy留了哪些口子、工程目录怎么摆

vnpy在国内开源交易框架里属于“重但完整”的那一类：行情网关、事件引擎、策略模板、回测引擎都是现成的，二次开发不是重写它，而是选对扩展点接进去。选错口子的代价很大——我见过有人把机器学习模型训练直接塞进on_bar回调里，回测一次跑四个小时，实盘还卡顿。先讲清楚三个扩展点，再给最小工程骨架。

2.1 二次开发优先改的三个位置：策略模板、数据服务、应用脚本

vnpy的事件驱动链路可以简化为四步：gateway收到行情 -> 事件引擎分发 -> 策略回调on_bar/on_tick -> 策略调用buy/sell发单。这条链路决定二次开发只需要动三处。

第一处是策略模板层。cta_strategy模块里的CtaTemplate是基类，重写on_bar、on_tick、参数定义就是最常见的二次开发。股票选股模型产出的预测值，最终也落到这里变成信号函数——策略逻辑本身不复杂，复杂的是“这个预测值怎么算出来”。

第二处是数据服务层。vnpy默认数据库是SQLite，但股票量化需要行情、复权因子、财务指标多张表，数据量远大于CTA。常见做法是保留vnpy的数据库接口，同时旁路维护一个自己的数据库（我一般直接开一个data/market.db），回测和选股脚本都从这里读，避免被vnpy的ORM层限制。注意vnpy 2.x到3.x的数据库配置方式变过，升级后要重点检查vnpy.trader.database相关的import。

第三处是应用脚本层。选股、因子计算、批量回测这类一次性任务，不建议做成vnpy的app，写成独立python脚本与vnpy通过数据库交互就够了。真正常跑的只有两件事：vnpy主程序负责实盘或仿真，独立脚本负责研究和回测。

网关层不建议动。vnpy的gateway对接柜台协议，稳定性要求极高，除非要接私有行情，否则改了后续升级全是坑。我处理过的二次开发项目里，凡是动了gateway的，无一例外在vnpy版本升级时翻车。

2.2 最小工程骨架：目录结构、依赖安装和连通性验证

一个能复现的vnpy二次开发包，目录应该按数据、因子、策略、回测、模型五块分，而不是把所有代码堆在vnpy的example目录里。下面是我常用的一套结构：

project_root/ ├── config.py # 全局参数：资金、费率、回测区间 ├── data/ │ ├── fetch_daily.py # 拉取日线行情入库 │ ├── fetch_finance.py # 拉取财务指标入库 │ └── market.db # 本地SQLite数据库 ├── factors/ │ └── factor_lib.py # 因子计算与标准化 ├── strategy/ │ └── ml_strategy.py # 继承CtaTemplate的策略模板 ├── backtest/ │ ├── run_backtest.py # 单股回测 │ └── run_portfolio.py # 多股组合回测 └── ml/ ├── feature_engine.py # 特征工程 └── train_model.py # 模型训练与滚动验证

依赖安装建议用独立虚拟环境，vnpy的依赖比较重，和系统Python混装容易在NumPy版本上打架：

python -m venv .venv source .venv/bin/activate # Windows下用 .venv\Scripts\activate pip install vnpy vnpy_ctastrategy pandas numpy akshare scikit-learn lightgbm

装完先做连通性验证，不要急着跑策略：

python -c "from vnpy.trader.constant import Interval; print(Interval.DAILY)" python -c "from vnpy_ctastrategy import CtaTemplate, BacktestingEngine; print('cta ok')"

第一行验证vnpy基础包能import，第二行验证你二次开发最依赖的cta_strategy模块在。如果第二行报ImportError，是vnpy版本差异导致的路径问题，优先查vnpy_ctastrategy的包名和Python版本兼容表。我遇到过的情况是vnpy 2.x时代import自vnpy.app.cta_strategy，3.x改成vnpy_ctastrategy，代码里路径不换就全挂。

注意：Python版本建议3.8到3.11。vnpy 3.x在3.12上部分依赖包没有预编译wheel，装起来很痛苦。

3. 选股模块落地：数据接入、因子计算和股票池输出

选股模块是整个流水线的上游，它的输出直接决定后面回测和机器学习吃什么数据。很多团队在这里犯的第一个错是数据量贪多——把能拉到的字段全存下来，结果因子计算时被数据清洗耗掉大半时间。我的原则是：行情表只存开高低收、成交量、换手率五个核心字段，财务表只存ROE、毛利率、负债率等十来个因子原料字段，其余随用随拉。

3.1 行情与财务数据接入：把数据灌进本地SQLite

vnpy的数据库接口可以扩展，但股票选股阶段我更喜欢直接操作SQLite，理由很实际：选股脚本和回测脚本都是离线任务，不追求实时，SQLite轻量、可移植、用pandas读写都方便。下面这段是我常用的日线拉取入库脚本：

# data/fetch_daily.py import akshare as ak import sqlite3 import pandas as pd def fetch_daily(symbol: str, start: str, end: str, adjust: str = "hfq") -> pd.DataFrame: """拉取单只股票日线并写入SQLite，避免每次回测都重复请求数据源。 adjust参数： qfq 前复权，适合看盘； hfq 后复权，适合做因子计算和历史回测； 空串 不复权，适合算真实涨跌幅。 """ df = ak.stock_zh_a_hist( symbol=symbol, period="daily", start_date=start, end_date=end, adjust=adjust, ) df.columns = ["date", "open", "close", "high", "low", "volume", "amount", "amplitude", "pct", "change", "turnover"] df["symbol"] = symbol df["date"] = pd.to_datetime(df["date"]) return df[["symbol", "date", "open", "close", "high", "low", "volume", "turnover"]] conn = sqlite3.connect("data/market.db") for code in ["600519", "000858", "601318", "000001"]: df = fetch_daily(code, "20180101", "20241231", adjust="hfq") df.to_sql("daily_bar", conn, if_exists="append", index=False) conn.close()

这段代码里最关键的参数是adjust。用后复权做因子计算是底线：前复权价格会被后续除权除息事件整体修正，意味着你在2018年这个历史时点用到了2019年才知道的除权信息，这就是数据泄漏。后复权只做一次折算，历史上是什么样就是什么样。回测时如果数据源本身给了复权因子字段（比如tushare的adj_factor），优先存因子而不是直接存复权价，这样能随时切换口径。

财务数据接口各大数据源差异较大，akshare的ak.stock_financial_abstract按股票拉，tushare的pro接口可以按报告期批量拉。字段我建议只留四个：roe、gross_profit_margin、debt_to_assets、operating_cash_flow_per_share，按报告期对齐到交易日期时注意用“公告日期”而不是“报告期”，报告期数据披露有滞后，用报告期对齐等于让历史回测提前知道了财报。

3.2 因子计算与打分选股：从全市场压到TopN

因子不是越多越好，这一点是我做机器学习选股后的血泪经验。两三个逻辑清楚的基础因子，配上严格的样本外验证，远好过堆三四十个相关性很高的因子在模型里做黑匣子。先给一个可以直接抄的因子清单：

打分选股的逻辑是：每个因子做z-score标准化，方向不对的取负，然后等权相加得总分，取前N只进入股票池。代码如下：

# factors/factor_lib.py def compute_and_score(df: pd.DataFrame, top_n: int = 30) -> list: """计算因子、标准化、等权打分，返回TopN股票代码。""" # 按股票分组计算因子，取每个股票最新一行 df = df.sort_values(["symbol", "date"]) g = df.groupby("symbol", group_keys=False) df["momentum_20"] = g["close"].transform(lambda x: x.pct_change(20)) df["volatility_20"] = g["close"].transform(lambda x: x.pct_change().rolling(20).std()) df["turnover_20"] = g["turnover"].transform(lambda x: x.rolling(20).mean()) latest = df.groupby("symbol").tail(1).copy() # z-score，注意波动率因子方向是负向 latest["mom_z"] = (latest["momentum_20"] - latest["momentum_20"].mean()) / latest["momentum_20"].std() latest["vol_z"] = -1 * (latest["volatility_20"] - latest["volatility_20"].mean()) / latest["volatility_20"].std() latest["score"] = (latest["mom_z"] + latest["vol_z"]) / 2 return latest.nlargest(top_n, "score")["symbol"].tolist()

这段代码里有三个要点。其一是transform保证了因子对齐到每个交易日，不会因为groupby打乱时间顺序；其二是负向因子取负号再标准化，让所有因子统一成“越大越好”；其三是最终返回的股票池要落到数据库或CSV里，回测脚本按调仓日读取，不在回测过程中现算因子——回测阶段现算因子容易混入未来数据，属于回测的大忌。

到这里，选股模块的核心路径已经跑通：原始数据入库、因子计算、打分选股、输出池子。机器学习选股在这个框架里只是把“等权打分”替换成“模型预测值排序”，特征工程直接复用上面的因子，这个衔接放到第6章讲。

4. 回测模块落地：撮合参数、单股回测和多股组合净值

选股选出来的是池子，池子能不能赚钱取决于策略和回测设置。vnpy的回测引擎BacktestingEngine是基于事件驱动的，和backtrader这类框架比，优势是回测逻辑和实盘策略共用一套CtaTemplate代码，劣势是原生只支持单合约回测。股票组合回测需要自己在外面包一层，下面先讲单股回测的参数，再讲组合合成。

4.1 单股回测：BacktestingEngine的关键参数怎么设

vnpy回测的最小步骤是：定义策略类 -> 创建BacktestingEngine -> set_parameters设置合约和成本 -> add_strategy -> load_data -> run_backtesting -> calculate_result。先看一个最小示例：

# backtest/run_backtest.py from vnpy.trader.constant import Interval from vnpy_ctastrategy import CtaTemplate, BacktestingEngine class MomentumStrategy(CtaTemplate): """演示策略：突破20日高点买入，跌破20日低点清仓。""" def on_init(self): self.bars = [] self.pos = 0 def on_bar(self, bar): if not self.inited: return self.bars.append(bar) if len(self.bars) < 20: return high_20 = max(b.close_price for b in self.bars[-20:]) low_20 = min(b.close_price for b in self.bars[-20:]) if bar.close_price > high_20 and self.pos == 0: self.buy(bar.close_price, 100) elif bar.close_price < low_20 and self.pos > 0: self.sell(bar.close_price, self.pos) engine = BacktestingEngine() engine.set_parameters( vt_symbol="600519.SSE", interval=Interval.DAILY, start="20180101", end="20241231", rate=0.00025, # 手续费率，股票默认按万2.5 slippage=0.01, # 滑点，单位是价格，不是百分比 size=1, # 股票合约乘数是1 pricetick=0.01, # 最小价格变动单位 capital=1_000_000, # 初始资金 ) engine.add_strategy(MomentumStrategy, {}) engine.load_data() engine.run_backtesting() df = engine.calculate_result() # 返回含datetime和balance的净值序列 engine.calculate_statistics()

参数说明里最容易搞错的是slippage和size。slippage在vnpy里是价格单位，股票设0.01表示每笔成交价在信号价基础上多付一分钱，别把它当成百分比去设，否则回测成本会被放大一百倍。size对股票是1，对期货是合约乘数（比如螺纹钢是10），股票用户直接填1即可。另外vt_symbol的格式是“代码.交易所”，上交所是SSE，深交所是SZSE，写成600519.SSE和000001.SZSE，写反了回测会报合约信息错误。

回测跑完先看calculate_result的净值曲线，再跑calculate_statistics拿统计指标。不要直接信任统计函数里的夏普数值，股票日线策略用日频收益算夏普和CTA用小时线算的夏普不在一个量纲上，横向比较时统一频率才有意义。

4.2 多股组合回测：逐股回测再合成组合净值

vnpy原生只支持单合约回测，做多股组合回测的常见做法是：对股票池每只股票跑一遍单股回测，得到每日收益序列，再按日期对齐做等权组合。这个方案很直接，缺点是每次调仓都要重新跑全部股票，好在股票池一般只有30到50只，跑一轮分钟级就能完成。

# backtest/run_portfolio.py import sqlite3 import pandas as pd from vnpy.trader.constant import Interval from vnpy_ctastrategy import BacktestingEngine def backtest_one(symbol: str, exchange: str, cfg: dict) -> pd.DataFrame: """单股回测，返回以日期为索引的每日收益序列。""" engine = BacktestingEngine() engine.set_parameters( vt_symbol=f"{symbol}.{exchange}", interval=Interval.DAILY, start=cfg["start"], end=cfg["end"], rate=cfg["rate"], slippage=cfg["slippage"], size=1, pricetick=0.01, capital=cfg["capital"], ) engine.add_strategy(MomentumStrategy, {}) engine.load_data() engine.run_backtesting() result = engine.calculate_result() # 含datetime、balance result["daily_return"] = result["balance"].pct_change().fillna(0) return result.set_index("datetime")["daily_return"] cfg = {"start": "20180101", "end": "20241231", "rate": 0.00025, "slippage": 0.01, "capital": 1_000_000} conn = sqlite3.connect("data/market.db") # 股票池表由第3章的选股脚本写入 pool = pd.read_sql("select distinct symbol from stock_pool", conn)["symbol"] nav = 1.0 for code in pool: exchange = "SSE" if code.startswith("6") else "SZSE" ret = backtest_one(code, exchange, cfg) nav = nav * (1 + ret).fillna(1) # 等权合成，先简化处理 portfolio_nav = nav

这里我做了简化：等权合成的正确做法是每只股票的资金权重相同，且每天重新平衡到等权；上面的逐股累乘相当于“买入并持有每只股票”的组合，如果策略本身是调仓型，需要在每个调仓日对每只股票的收益序列做权重还原。更稳的做法是每只股票回测时记录每日持仓市值，组合层按市值加权或等权重新平衡。

组合回测的绩效指标要重点看五个：年化收益率、夏普比率、最大回撤、卡玛比率（年化收益除以最大回撤）、年化换手率。其中换手率最容易被忽视，一个高夏普但换手率极高的策略，实盘里会被冲击成本和滑点吃掉大半收益。我一般给组合回测加一个换手率上限约束，超过阈值直接放弃该参数组。

5. 避坑：vnpy二次开发和机器学习回测的5个高频翻车点

前面三章落地之后，下面5个问题最该反复看。我都在实际项目里遇到过，前三个偏数据和框架层，后两个偏策略和模型层，按“现象-原因-解决”写清楚，每个都能单独排掉。

5.1 数据与框架层：复权口径、vnpy版本和数据对齐

坑一：复权因子不一致，同一策略回测结果差一倍。

现象：同一个双均线策略，一套回测年化25%，另一套只有8%，代码几乎没差别。

原因：第一套用前复权数据，第二套用后复权数据。前复权价格会被后续除权事件整体缩放，导致历史K线的涨跌幅被改写，策略信号的触发时间点都变了，回测指标自然全变。

解决：全流程统一用后复权做因子和信号计算，实盘下单时再把后复权价格换算成真实成交价。数据入库时把复权因子字段单独存一列，不要只存复权后的价格，这样切换口径只改一行代码。

坑二：升级vnpy之后二次开发代码大面积失效。

现象：原来跑得好好的策略，pip install -U vnpy之后import直接报错，策略类找不到基类。

原因：vnpy 2.x到3.x做了模块拆分，原来vnpy.app.cta_strategy的路径改为独立的vnpy_ctastrategy包，数据库ORM也换过一轮。文档里搜得到的旧示例大多是老版本，直接照着抄就撞墙。

解决：用requirements.txt锁死vnpy和vnpy_ctastrategy版本，升级前先在一个虚拟环境里跑最小回测验证。我现在的做法是项目根目录放一个requirements-lock.txt，里面固定主版本号，除非需要新功能，否则不主动升级vnpy。

坑三：行情时间戳和财务公告日期错位。

现象：机器学习特征里ROE和行情数据合并后，特征值数量对不上，IC值忽高忽低。

原因：财务数据接口默认按报告期给数据，比如年报是12月31日，但实际披露往往是次年三四月份。按报告期对齐到行情日期，等于让1月份的行情用上了4月才公布的财报，数据泄漏。

解决：对齐财务数据时用公告日期（ann_date）而不是报告期（end_date），在特征工程里让“最新可得”的财务数据在披露后才进入样本。

5.2 策略与模型层：前视偏差、数据泄漏和幸存者偏差

坑四：on_bar里用收盘价发单，等于偷看收盘价。

现象：回测净值曲线光滑上涨，实盘一塌糊涂。逐笔对比发现，策略在当天收盘价创新高时买入，回测引擎也按当天收盘价成交——这在日线策略里就是典型的前视偏差。

原因：on_bar收到的bar是已经收盘的完整K线，此时你看到收盘价已经是事实，但vnpy回测引擎默认按当前bar价格撮合订单，等于下了个“收盘后才提交的市价单”还能按收盘价成交，实盘根本做不到。

解决：信号产生和成交错开一根bar，常见做法是on_bar里用上一根bar的收盘价做判断，当前bar开盘价成交；或者让策略在on_bar里只使用开盘时可得的数据，收盘数据只用于下一根bar的信号决策。关键是养成“上一根bar出信号、下一根bar成交”的习惯。

坑五：机器学习回测的幸存者偏差和随机划分。

现象：机器学习选股策略样本内IC达到0.08，回测年化30%以上，上线后大幅缩水。

原因：两个问题叠加。一是用当前时点沪深300成分股回测历史，退市股被排除在外，历史回测的股票池本身是“活下来的股票”；二是训练时把历史数据随机打乱划分训练集和测试集，时间序列的强自相关性让模型在“用未来验证过去”。

解决：股票池用历史时点成分股或者全市场规模股，至少包含历史退市股；训练和验证严格按时间切分，训练集只包含验证集之前的样本，不做随机shuffle。机器学习模型在量化里的应用流程，最大的忌讳就是把图像识别那套“随机划分数据集”的习惯直接搬过来。

6. 机器学习进阶：滚动验证与把模型预测接进vnpy策略

选股不只有打分法，把机器学习模型的预测值作为选股信号，是这个二次开发包里的核心进阶用法。机器学习在量化里的应用流程可以拆成四步：特征工程、时间序列切分、模型训练、样本外验证。这里只讲最容易做歪的两步。

6.1 滚动验证：先过样本外这一关

股票收益率不是独立同分布的，随机打乱划分训练集和测试集在海量样本的CV任务里没问题，在金融时序里就是数据泄漏。滚动验证的规则是：训练集永远在验证集之前，且按时间段前移重复训练。

# ml/train_model.py import pandas as pd from lightgbm import LGBMRegressor def rolling_predict(features: pd.DataFrame, train_days: int = 240, pred_days: int = 20, target: str = "future_return"): """滚动训练LightGBM，输出各段样本外预测值，用于计算IC和分组收益。""" preds = [] idx = [] for start in range(train_days, len(features), pred_days): train = features.iloc[start - train_days:start] test = features.iloc[start:start + pred_days] if len(test) < 20: continue model = LGBMRegressor(n_estimators=100, learning_rate=0.05, num_leaves=15, verbose=-1) model.fit(train.drop(columns=[target]), train[target]) preds.append(model.predict(test.drop(columns=[target]))) idx.extend(test.index) pred_series = pd.Series( pd.concat([pd.Series(p) for p in preds]).values, index=idx) return pred_series

滚动验证的窗口参数要按策略的调仓频率来定。日线策略我常用train_days=240（约一年）、pred_days=20（约一个月）滚动，这样每次训练都在最近一年的数据上做。learning_rate和num_leaves的取值宁小勿大，股票收益预测本身信噪比极低，模型复杂度过高必过拟合。收敛标准是样本外IC稳定在正值且排名稳定，如果只有一段特定市场里IC好看，这个信号大概率是市场风格，不是你模型的能力。

6.2 把模型预测接进vnpy策略：on_bar里只做推理不做训练

模型训练是离线任务，vnpy策略里只加载训练好的模型做推理。常见做法是feature_engine和model一起序列化，策略on_bar里先算因子特征再调model.predict，把预测值拼成一个信号：

# strategy/ml_strategy.py import pickle class MLStrategy(CtaTemplate): buy_threshold = 0.02 # 预测未来20日收益大于2%买入 sell_threshold = -0.02 def on_init(self): with open("ml/model.pkl", "rb") as f: self.model = pickle.load(f) with open("ml/feature_engine.pkl", "rb") as f: self.feature_engine = pickle.load(f) def on_bar(self, bar): if not self.inited: return feat = self.feature_engine.calc(bar) # 只用当前bar已收盘可得的数据 pred = self.model.predict(feat)[0] if pred > self.buy_threshold and self.pos == 0: self.buy(bar.close_price, 100) elif pred < self.sell_threshold and self.pos > 0: self.sell(bar.close_price, self.pos)

注意两个细节：其一，feature_engine的输入只能是当前bar及之前数据能算出来的字段，任何用到未来bar的统计量都会让模型训练时的特征分布和实盘不一致；其二，on_init里加载模型文件，不要在on_bar里每次读盘，回测和实盘都会因此变慢。

我现在每上一套机器学习选股，都先把滚动验证跑一遍，确认样本外IC稳定才接进vnpy策略。回测里看到的高夏普，多半是某个环节偷看了未来，数据对齐和滚动验证这两步能省掉后面大部分返工。希望帮到你。

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