vn.py（开源量化交易框架）

vn.py 是中国最受欢迎的开源量化交易框架，由社区驱动。支持CTA策略、价差交易、期权波动率交易等。连接CTP、飞马、恒生及20多家券商接口。

文档：https://www.vnpy.com/docs/cn/
GitHub：https://github.com/vnpy/vnpy

安装

bash

安装核心框架

pip install vnpy

安装CTP接口（期货）

pip install vnpy-ctp

安装常用组件

pip install vnpy-ctastrategy # CTA策略模块 pip install vnpy-spreadtrading # 价差交易模块 pip install vnpy-datamanager # 数据管理模块 pip install vnpy-sqlite # SQLite数据库 pip install vnpy-rqdata # RQData数据服务

架构概述

VeighNa Trader (GUI)
├── Gateway (Broker Interface)
│ ├── CTP (Futures)
│ ├── Femas (Futures)
│ ├── Hundsun UFT (Securities)
│ ├── EMT (Securities)
│ └── IB / Alpaca (International)
├── App (Application Modules)
│ ├── CtaStrategy (CTA Strategy)
│ ├── SpreadTrading (Spread Trading)
│ ├── OptionMaster (Options Trading)
│ ├── PortfolioStrategy (Portfolio Strategy)
│ └── AlgoTrading (Algorithmic Trading)
└── DataService
├── RQData
├── TuShare
└── Custom Data Sources

启动GUI

python
from vnpy.event import EventEngine
from vnpy.trader.engine import MainEngine
from vnpy.trader.ui import MainWindow, create_qapp

导入券商接口

from vnpy_ctp import CtpGateway

导入应用模块

from vnpy_ctastrategy import CtaStrategyApp from vnpy_spreadtrading import SpreadTradingApp from vnpy_datamanager import DataManagerApp

创建应用

qapp = create_qapp() event_engine = EventEngine() mainengine = MainEngine(eventengine)

添加券商接口

mainengine.addgateway(CtpGateway)

添加应用模块

mainengine.addapp(CtaStrategyApp) mainengine.addapp(SpreadTradingApp) mainengine.addapp(DataManagerApp)

创建并显示主窗口

mainwindow = MainWindow(mainengine, event_engine) main_window.showMaximized() qapp.exec()

---

CTA策略开发

CTA（商品交易顾问）策略是vn.py中的核心策略类型，适用于趋势跟踪、均值回归等。

策略模板

python
from vnpy_ctastrategy import (
CtaTemplate,
StopOrder,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)

class DoubleMaStrategy(CtaTemplate):
双均线CTA策略
author = 量化开发者

# 策略参数（可在GUI中编辑）
fast_window = 10 # 快速均线周期
slow_window = 20 # 慢速均线周期
fixed_size = 1 # 每次交易手数

# 策略变量（在GUI中显示）
fast_ma0 = 0.0
slow_ma0 = 0.0

# 参数和变量列表（用于GUI显示和持久化）
parameters = [fastwindow, slowwindow, fixed_size]
variables = [fastma0, slowma0]

def init(self, ctaengine, strategyname, vt_symbol, setting):
super().init(ctaengine, strategyname, vt_symbol, setting)

# K线生成器：将tick合成1分钟K线，再合成N分钟K线
self.bg = BarGenerator(self.on_bar)
# 数组管理器：存储K线数据并计算技术指标
self.am = ArrayManager(size=100) # 保留最近100根K线

def on_init(self):
策略初始化——加载历史数据
self.write_log(策略初始化中)
self.load_bar(10) # 加载最近10天的历史K线

def on_start(self):
策略启动
self.write_log(策略已启动)

def on_stop(self):
策略停止
self.write_log(策略已停止)

def on_tick(self, tick: TickData):
Tick数据回调——合成为K线
self.bg.update_tick(tick)

def on_bar(self, bar: BarData):
K线数据回调——主要交易逻辑
self.am.update_bar(bar)
if not self.am.inited:
return # 数据不足，等待

# 计算快慢均线
fastma = self.am.sma(self.fastwindow, array=False)
slowma = self.am.sma(self.slowwindow, array=False)

# 检测金叉/死叉
crossover = fastma > slow_ma # 金叉
crossbelow = fastma < slow_ma # 死叉

if cross_over:
if self.pos == 0:
self.buy(bar.closeprice, self.fixedsize) # 无持仓，开多
elif self.pos < 0:
self.cover(bar.close_price, abs(self.pos)) # 先平空
self.buy(bar.closeprice, self.fixedsize) # 再开多

elif cross_below:
if self.pos == 0:
self.short(bar.closeprice, self.fixedsize) # 无持仓，开空
elif self.pos > 0:
self.sell(bar.close_price, abs(self.pos)) # 先平多
self.short(bar.closeprice, self.fixedsize) # 再开空

# 更新GUI显示变量
self.fastma0 = fastma
self.slowma0 = slowma
self.put_event() # 触发GUI更新

def on_order(self, order: OrderData):
订单状态更新
pass

def on_trade(self, trade: TradeData):
成交回调
self.put_event()

def onstoporder(self, stop_order: StopOrder):
停止单状态更新
pass

---

交易函数参考

方法	描述
self.buy(price, volume)	买入开多仓
self.sell(price, volume)

卖出平多仓 |
| self.short(price, volume) | 卖出开空仓 |
| self.cover(price, volume) | 买入平空仓 |
| self.cancel_all() | 撤销所有挂单 |
| self.write_log(msg) | 写入日志信息 |
| self.put_event() | 触发GUI更新 |
| self.load_bar(days) | 加载N天历史K线 |
| self.load_tick(days) | 加载N天历史Tick |

ArrayManager技术指标方法

ArrayManager 提供内置的技术指标计算：

python
am = ArrayManager(size=100)

移动平均线

am.sma(n, array=False) # 简单移动平均线 am.ema(n, array=False) # 指数移动平均线 am.kama(n, array=False) # 考夫曼自适应移动平均线

波动率

am.std(n, array=False) # 标准差 am.atr(n, array=False) # 平均真实波幅

动量指标

am.rsi(n, array=False) # 相对强弱指标 am.cci(n, array=False) # 商品通道指数 am.macd(fast, slow, signal) # MACD（返回dif, dea, macd） am.adx(n, array=False) # 平均趋向指数

布林带

am.boll(n, dev, array=False) # 返回（上轨，下轨）通道

唐奇安通道

am.donchian(n, array=False) # 返回（上轨，下轨）通道

其他

am.aroon(n, array=False) # 阿隆指标 am.trix(n, array=False) # 三重指数移动平均线

---

BarGenerator — 多周期K线

python
from vnpy_ctastrategy import BarGenerator

class MyStrategy(CtaTemplate):
def init(self, ctaengine, strategyname, vt_symbol,