Export a Vibe-Trading backtest strategy to a runnable vnpy CtaTemplate Python class — supports A-share equities, futures, and crypto via BarGenerator + ArrayManager.
Export a Vibe-Trading backtest strategy to a runnable vnpy CtaTemplate Python class — supports A-share equities, futures, and crypto via BarGenerator + ArrayManager.
category
tool
Overview
This skill translates a Vibe-Trading strategy into a vnpy CtaTemplate subclass.py file
that can be loaded directly into the vnpy CTA Strategy App for live trading or vnpy backtesting.
Output file: artifacts/vnpy_strategy/<StrategyName>Strategy.py (inside the run directory).
vnpy is the most widely-used open-source quant framework in mainland China (39k+ GitHub stars).
Use this skill when the user asks to export to vnpy, requests a /vnpy command, or wants to
run a Vibe-Trading strategy inside vnpy's CTA backtester or live trading engine.
Called once at startup; call load_bar(n) to warm up indicators
on_start
Called when strategy is started by user
on_stop
Called when strategy is stopped
on_tick
Receives live tick data; forward to BarGenerator
on_bar
Main logic — called once per bar by BarGenerator
on_order
Order status updates
on_trade
Fill notifications
on_stop_order
Stop-order status (if using stop orders)
Always callself.cancel_all() at the start of on_bar to avoid stale orders.
Always callself.put_event() at the end of on_bar to refresh the UI.
Full Template
See scripts/cta_template.py for a complete, runnable example (MA crossover).
The template below is the canonical skeleton — replace the # SIGNAL LOGIC section:
from vnpy_ctastrategy import (
CtaTemplate,
StopOrder,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)
class {{StrategyName}}Strategy(CtaTemplate):
"""
Vibe-Trading export — {{StrategyName}}
Generated from run: {{run_id}}
Instrument: {{vt_symbol}}
"""
author = "Vibe-Trading"# ── Parameters (editable in vnpy UI) ──────────────────────────────────
{{param_name}} = {{param_default}} # add one line per parameter
parameters = [{{param_list_as_strings}}]
# ── Variables (displayed in vnpy UI, reset on strategy restart) ────────
{{var_name}} = 0.0# add one line per runtime variable
variables = [{{var_list_as_strings}}]
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)
self.am = ArrayManager()
# initialise variable attributes to match class-level defaults# (vnpy requires instance attributes for variables declared above)defon_init(self):
self.write_log("Strategy initialised")
self.load_bar({{warmup_bars}}) # load enough bars to warm up all indicatorsdefon_start(self):
self.write_log("Strategy started")
self.put_event()
defon_stop(self):
self.write_log("Strategy stopped")
defon_tick(self, tick: TickData):
self.bg.update_tick(tick)
defon_bar(self, bar: BarData):
self.cancel_all()
am = self.am
am.update_bar(bar)
ifnot am.inited:
return# ── INDICATOR CALCULATIONS ──────────────────────────────────────────# translate indicators from signal_engine.py using the mapping table# ── SIGNAL LOGIC ───────────────────────────────────────────────────# set cross_over / cross_under (or long_signal / short_signal) here# ── ORDER EXECUTION ────────────────────────────────────────────────if cross_over:
ifself.pos == 0:
self.buy(bar.close_price, 1)
elifself.pos < 0:
self.cover(bar.close_price, 1)
self.buy(bar.close_price, 1)
elif cross_under:
ifself.pos == 0:
self.short(bar.close_price, 1)
elifself.pos > 0:
self.sell(bar.close_price, 1)
self.short(bar.close_price, 1)
self.put_event()
defon_order(self, order: OrderData):
passdefon_trade(self, trade: TradeData):
self.put_event()
defon_stop_order(self, stop_order: StopOrder):
pass
Python → ArrayManager Indicator Mapping
ArrayManager is vnpy's built-in vectorised indicator library. Always prefer it over pandas
when the equivalent method exists — it is faster and avoids look-ahead bias.
Python (Vibe-Trading / pandas / ta-lib)
vnpy ArrayManager
df['close'].rolling(n).mean()
am.sma(n)
df['close'].ewm(span=n).mean()
am.ema(n)
ta.RSI(close, n)
am.rsi(n)
ta.MACD(close, 12, 26, 9)
am.macd(12, 26, 9) → (macd, signal, hist)
Bollinger Bands
am.boll(n, dev) → (mid, upper, lower)
ATR
am.atr(n)
ADX
am.adx(n)
df['close'].rolling(n).std()
am.std(n)
Stochastic K, D
am.kd(n, m) → (k, d)
df['high'].rolling(n).max()
am.high_array[-n:].max()
df['low'].rolling(n).min()
am.low_array[-n:].min()
Donchian channel
am.donchian(n) → (upper, lower)
df['close'].shift(1) (previous bar)
am.close_array[-2]
Last N bars as array
am.sma(n, array=True) (returns full array)
Using arrays: pass array=True to get the full history array (e.g. for crossover detection):
