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

vn.py is the most popular open-source quantitative trading framework in China, community-driven. Supports CTA strategies, spread trading, options volatility trading, and more. Connects to CTP, Femas, Hundsun, and 20+ broker gateways.

Docs: https://www.vnpy.com/docs/cn/ GitHub: https://github.com/vnpy/vnpy

安装

# Install core framework
pip install vnpy

# Install CTP gateway (futures)
pip install vnpy-ctp

# Install common components
pip install vnpy-ctastrategy    # CTA strategy module
pip install vnpy-spreadtrading  # Spread trading module
pip install vnpy-datamanager    # Data management module
pip install vnpy-sqlite         # SQLite database
pip install vnpy-rqdata         # RQData data service

架构概述

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

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

# Import broker gateways
from vnpy_ctp import CtpGateway

# Import application modules
from vnpy_ctastrategy import CtaStrategyApp
from vnpy_spreadtrading import SpreadTradingApp
from vnpy_datamanager import DataManagerApp

# Create application
qapp = create_qapp()
event_engine = EventEngine()
main_engine = MainEngine(event_engine)

# Add broker gateways
main_engine.add_gateway(CtpGateway)

# Add application modules
main_engine.add_app(CtaStrategyApp)
main_engine.add_app(SpreadTradingApp)
main_engine.add_app(DataManagerApp)

# Create and show main window
main_window = MainWindow(main_engine, event_engine)
main_window.showMaximized()
qapp.exec()

---

CTA策略开发

CTA (Commodity Trading Advisor) strategies are the core strategy type in vn.py, suitable for trend following, mean reversion, etc.

策略模板

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

class DoubleMaStrategy(CtaTemplate):
    """Dual Moving Average CTA Strategy"""
    author = "Quant Developer"

    # Strategy parameters (editable in GUI)
    fast_window = 10       # 快速均线周期
    slow_window = 20       # 慢速均线周期
    fixed_size = 1         # 每次交易手数

    # Strategy variables (displayed in GUI)
    fast_ma0 = 0.0
    slow_ma0 = 0.0

    # Parameter and variable lists (for GUI display and persistence)
    parameters = ["fast_window", "slow_window", "fixed_size"]
    variables = ["fast_ma0", "slow_ma0"]

    def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
        super().__init__(cta_engine, strategy_name, vt_symbol, setting)

        # Bar generator: synthesize ticks into 1-min bars, then into N-min bars
        self.bg = BarGenerator(self.on_bar)
        # Array manager: store bar data and compute technical indicators
        self.am = ArrayManager(size=100)  # Keep last 100 bars

    def on_init(self):
        """Strategy initialization — load historical data"""
        self.write_log("Strategy initializing")
        self.load_bar(10)  # Load last 10 days of historical bars

    def on_start(self):
        """Strategy started"""
        self.write_log("Strategy started")

    def on_stop(self):
        """Strategy stopped"""
        self.write_log("Strategy stopped")

    def on_tick(self, tick: TickData):
        """Tick data callback — synthesize into bars"""
        self.bg.update_tick(tick)

    def on_bar(self, bar: BarData):
        """Bar data callback — main trading logic"""
        self.am.update_bar(bar)
        if not self.am.inited:
            return  # Not enough data, wait

        # Calculate fast and slow moving averages
        fast_ma = self.am.sma(self.fast_window, array=False)
        slow_ma = self.am.sma(self.slow_window, array=False)

        # Detect golden cross / death cross
        cross_over = fast_ma > slow_ma    # 金叉
        cross_below = fast_ma < slow_ma   # 死叉

        if cross_over:
            if self.pos == 0:
                self.buy(bar.close_price, self.fixed_size)    # No position, go long
            elif self.pos < 0:
                self.cover(bar.close_price, abs(self.pos))    # Close short first
                self.buy(bar.close_price, self.fixed_size)    # Then go long

        elif cross_below:
            if self.pos == 0:
                self.short(bar.close_price, self.fixed_size)  # No position, go short
            elif self.pos > 0:
                self.sell(bar.close_price, abs(self.pos))     # Close long first
                self.short(bar.close_price, self.fixed_size)  # Then go short

        # Update GUI display variables
        self.fast_ma0 = fast_ma
        self.slow_ma0 = slow_ma
        self.put_event()  # Trigger GUI update

    def on_order(self, order: OrderData):
        """Order status update"""
        pass

    def on_trade(self, trade: TradeData):
        """Trade execution callback"""
        self.put_event()

    def on_stop_order(self, stop_order: StopOrder):
        """Stop order status update"""
        pass

---

交易函数参考

Method	Description
self.buy(price, volume)	Buy to open long position
self.sell(price, volume)	Sell to close long position
self.short(price, volume)	Sell to open short position
self.cover(price, volume)	Buy to close short position
self.cancel_all()	Cancel all pending orders
self.write_log(msg)	Write log message
self.put_event()	Trigger GUI update
self.load_bar(days)	Load N days of historical bars
self.load_tick(days)	Load N days of historical ticks

ArrayManager技术指标方法

The ArrayManager provides built-in technical indicator calculations:

am = ArrayManager(size=100)

# Moving Averages
am.sma(n, array=False)          # Simple Moving Average
am.ema(n, array=False)          # Exponential Moving Average
am.kama(n, array=False)         # Kaufman Adaptive Moving Average

# Volatility
am.std(n, array=False)          # Standard Deviation
am.atr(n, array=False)          # Average True Range

# Momentum
am.rsi(n, array=False)          # Relative Strength Index
am.cci(n, array=False)          # Commodity Channel Index
am.macd(fast, slow, signal)     # MACD (returns dif, dea, macd)
am.adx(n, array=False)          # Average Directional Index

# Bollinger Bands
am.boll(n, dev, array=False)    # 收益率 (upper, lower) bands

# Donchian Channel
am.donchian(n, array=False)     # 收益率 (upper, lower) channel

# Other
am.aroon(n, array=False)        # Aroon indicator
am.trix(n, array=False)         # Triple EMA

---

BarGenerator — 多周期K线

from vnpy_ctastrategy import BarGenerator

class MyStrategy(CtaTemplate):
    def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
        super().__init__(cta_engine, strategy_name, vt_symbol, setting)

        # Synthesize ticks into 1-min bars, then 1-min bars into 15-min bars
        self.bg = BarGenerator(self.on_bar, 15, self.on_15min_bar)
        self.am = ArrayManager()

    def on_tick(self, tick: TickData):
        self.bg.update_tick(tick)

    def on_bar(self, bar: BarData):
        """1-minute bar callback — feed into 15-min generator"""
        self.bg.update_bar(bar)

    def on_15min_bar(self, bar: BarData):
        """15-minute bar callback — main trading logic"""
        self.am.update_bar(bar)
        if not self.am.inited:
            return
        # Trading logic here

Supported intervals for BarGenerator: - Minutes: 1, 3, 5, 15, 30, 60 - Hours: pass Interval.HOUR as the interval parameter - Custom: any integer N for N-minute bars

---

价差交易

from vnpy_spreadtrading import (
    SpreadStrategyTemplate,
    SpreadAlgoTemplate,
    SpreadData,
    LegData,
    BacktestingEngine,
)

class SpreadArbitrageStrategy(SpreadStrategyTemplate):
    """Simple spread arbitrage strategy"""
    author = "Quant Developer"

    buy_price = 0.0
    sell_price = 0.0
    max_pos = 10
    payup = 10

    parameters = ["buy_price", "sell_price", "max_pos", "payup"]

    def on_init(self):
        self.write_log("Strategy initializing")

    def on_start(self):
        self.write_log("Strategy started")

    def on_spread_data(self):
        """Spread data update callback"""
        spread = self.get_spread_tick()
        if not spread:
            return

        if self.spread_pos < self.max_pos:
            if spread.last_price <= self.buy_price:
                self.start_long_algo(
                    spread.last_price, self.max_pos - self.spread_pos,
                    payup=self.payup
                )

        if self.spread_pos > -self.max_pos:
            if spread.last_price >= self.sell_price:
                self.start_short_algo(
                    spread.last_price, self.max_pos + self.spread_pos,
                    payup=self.payup
                )

---

实盘交易配置

from vnpy.event import EventEngine
from vnpy.trader.engine import MainEngine
from vnpy_ctp import CtpGateway
from vnpy_ctastrategy import CtaStrategyApp

# Initialize engine
event_engine = EventEngine()
main_engine = MainEngine(event_engine)
main_engine.add_gateway(CtpGateway)
cta_engine = main_engine.add_app(CtaStrategyApp)

# Connect to CTP broker
ctp_setting = {
    "userid": "your_user_id",
    "password": "your_password",
    "brokerid": "9999",
    "td_address": "tcp://180.168.146.187:10201",
    "md_address": "tcp://180.168.146.187:10211",
    "appid": "simnow_client_test",
    "auth_code": "0000000000000000",
}
main_engine.connect(ctp_setting, "CTP")

# Initialize CTA engine and add strategy
cta_engine.init_engine()
cta_engine.add_strategy(
    DoubleMaStrategy,
    "double_ma_IF",
    "IF2401.CFFEX",
    {"fast_window": 10, "slow_window": 20}
)
cta_engine.init_strategy("double_ma_IF")
cta_engine.start_strategy("double_ma_IF")

---

数据管理

from vnpy_datamanager import DataManagerApp

# Add data manager to main engine
dm_engine = main_engine.add_app(DataManagerApp)

# Download historical data from RQData
dm_engine.download_bar_data(
    symbol="IF2401",
    exchange="CFFEX",
    interval="1m",
    start=datetime(2024, 1, 1)
)

---

进阶示例

RSI均值回归策略

from vnpy_ctastrategy import CtaTemplate, BarData, BarGenerator, ArrayManager

class RsiStrategy(CtaTemplate):
    """RSI mean reversion strategy — buy oversold, sell overbought"""
    author = "Quant Developer"

    rsi_period = 14
    rsi_buy = 30       # Oversold threshold
    rsi_sell = 70      # Overbought threshold
    fixed_size = 1

    parameters = ["rsi_period", "rsi_buy", "rsi_sell", "fixed_size"]
    variables = ["rsi_value"]

    rsi_value = 0.0

    def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
        super().__init__(cta_engine, strategy_name, vt_symbol, setting)
        self.bg = BarGenerator(self.on_bar, 15, self.on_15min_bar)
        self.am = ArrayManager()

    def on_init(self):
        self.write_log("Strategy initializing")
        self.load_bar(10)

    def on_start(self):
        self.write_log("Strategy started")

    def on_stop(self):
        self.write_log("Strategy stopped")

    def on_tick(self, tick):
        self.bg.update_tick(tick)

    def on_bar(self, bar):
        self.bg.update_bar(bar)

    def on_15min_bar(self, bar: BarData):
        self.am.update_bar(bar)
        if not self.am.inited:
            return

        self.rsi_value = self.am.rsi(self.rsi_period)

        if self.pos == 0:
            if self.rsi_value < self.rsi_buy:
                self.buy(bar.close_price, self.fixed_size)
            elif self.rsi_value > self.rsi_sell:
                self.short(bar.close_price, self.fixed_size)
        elif self.pos > 0:
            if self.rsi_value > self.rsi_sell:
                self.sell(bar.close_price, abs(self.pos))
                self.short(bar.close_price, self.fixed_size)
        elif self.pos < 0:
            if self.rsi_value < self.rsi_buy:
                self.cover(bar.close_price, abs(self.pos))
                self.buy(bar.close_price, self.fixed_size)

        self.put_event()

    def on_order(self, order):
        pass

    def on_trade(self, trade):
        self.put_event()

    def on_stop_order(self, stop_order):
        pass

布林带突破策略

from vnpy_ctastrategy import CtaTemplate, BarData, BarGenerator, ArrayManager

class BollBreakoutStrategy(CtaTemplate):
    """Bollinger Band breakout strategy with ATR-based stop loss"""
    author = "Quant Developer"

    boll_period = 20
    boll_dev = 2.0
    atr_period = 14
    atr_multiplier = 2.0
    fixed_size = 1

    parameters = ["boll_period", "boll_dev", "atr_period", "atr_multiplier", "fixed_size"]
    variables = ["boll_up", "boll_down", "atr_value"]

    boll_up = 0.0
    boll_down = 0.0
    atr_value = 0.0

    def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
        super().__init__(cta_engine, strategy_name, vt_symbol, setting)
        self.bg = BarGenerator(self.on_bar, 30, self.on_30min_bar)
        self.am = ArrayManager()

    def on_init(self):
        self.write_log("Strategy initializing")
        self.load_bar(10)

    def on_start(self):
        self.write_log("Strategy started")

    def on_stop(self):
        self.write_log("Strategy stopped")

    def on_tick(self, tick):
        self.bg.update_tick(tick)

    def on_bar(self, bar):
        self.bg.update_bar(bar)

    def on_30min_bar(self, bar: BarData):
        self.cancel_all()
        self.am.update_bar(bar)
        if not self.am.inited:
            return

        self.boll_up, self.boll_down = self.am.boll(self.boll_period, self.boll_dev)
        self.atr_value = self.am.atr(self.atr_period)

        if self.pos == 0:
            # Breakout above upper band — go long
            if bar.close_price > self.boll_up:
                self.buy(bar.close_price, self.fixed_size)
            # Breakout below lower band — go short
            elif bar.close_price < self.boll_down:
                self.short(bar.close_price, self.fixed_size)
        elif self.pos > 0:
            # Stop loss: ATR trailing stop
            stop_price = bar.close_price - self.atr_value * self.atr_multiplier
            self.sell(stop_price, abs(self.pos), stop=True)
        elif self.pos < 0:
            stop_price = bar.close_price + self.atr_value * self.atr_multiplier
            self.cover(stop_price, abs(self.pos), stop=True)

        self.put_event()

    def on_order(self, order):
        pass

    def on_trade(self, trade):
        self.put_event()

    def on_stop_order(self, stop_order):
        pass

---

支持的接口

Gateway	Market	Protocol
CTP	China Futures	CTP
Femas	China Futures	Femas
Hundsun UFT	China Securities	UFT
EMT	China Securities	EMT
XTP	China Securities	XTP
IB	International	TWS API
Alpaca	US Stocks	REST API
Binance	Crypto	REST/WebSocket

使用技巧

• vn.py is the go-to framework for live trading in Chinese futures and securities markets.
• Use BarGenerator to synthesize multi-timeframe bars from tick data.
• ArrayManager provides 20+ built-in technical indicators — no need for external libraries.
• For backtesting, use the built-in BacktestingEngine in vnpy_ctastrategy.
• CTP gateway requires a broker account with CTP access (e.g., SimNow for testing).
• The modular architecture allows mixing gateways and strategy types freely.
• Docs: https://www.vnpy.com/docs/cn/

---

社区与支持

由 大佬量化 (BossQuant) 维护 — 量化交易教学与策略研发团队。

微信客服: bossquant1 · Bilibili · 搜索 大佬量化 — 微信公众号 / Bilibili / 抖音