蜗牛博客VNPY学习记录：
VN.PY 2.0学习记录一（如何回测）
VN.PY 2.0学习记录二（策略开发）
Vn.py学习记录三（米筐教程)
VN.PY 2.0学习记录四（多线程、多进程）
Vn.py学习记录五–交易时间段及Widgets
Vn.py学习记录六（无界面模拟盘）
Vn.py学习记录七（V2.0.5版本）
Vnpy学习记录八（R-Breaker及pickle）
Vn.py学习记录九（事件驱动引擎）
VN.PY学习记录十（源码概述）
VNPY学习记录11（微信+Vscode）
VNPY学习记录12（父子进程、回调函数）
VNPY学习记录13（部署到云服务器，实现自动交易）
一、如何将成交信息发到微信
原理：利用vnpy的内置函数发送邮件到qq邮箱，微信绑定QQ邮箱，在微信收到信息。
1、设定邮箱，注意password是邮箱授权码。
2、发送邮件的代码，写在on_trade
def on_trade(self.trade:TradeData):
msg = f" 新的成交，策略{self.strategy_name}，方向{trade.direction}，开平{trade.offset}，当前仓位{self.pos}"
self.send_email(msg)
二、curtro策略
复习的知识要点：
1.布林的代码（包括参数、使用）
2.RSI的代码
3.双均线的代码
4.移动上操的代码
from vnpy.app.cta_strategy import (
CtaTemplate,
BarGenerator,
ArrayManager,
TickData,
BarData,
OrderData,
TradeData,
StopOrder
)
class CuatroStrategy(CtaTemplate):
""""""
author = "vn_trader"
boll_window = 20
boll_dev = 1.8
rsi_window = 14
rsi_signal = 19
fast_window = 4
slow_window = 26
trailing_long = 0.5
trailing_short = 0.3
fixed_size = 1
boll_up = 0
boll_down = 0
rsi_value = 0
rsi_long = 0
rsi_short = 0
fast_ma = 0
slow_ma = 0
ma_trend = 0
intra_trade_high = 0
intra_trade_low = 0
long_stop = 0
short_stop = 0
parameters = [
"boll_window",
"boll_dev",
"rsi_window",
"rsi_signal",
"fast_window",
"slow_window",
"trailing_long",
"trailing_short",
"fixed_size"
]
variables = [
"boll_up",
"boll_down",
"rsi_value",
"rsi_long",
"rsi_short",
"fast_ma",
"slow_ma",
"ma_trend",
"intra_trade_high",
"intra_trade_low",
"long_stop",
"short_stop"
]
def __init__(
self,
cta_engine,
strategy_name: str,
vt_symbol: str,
setting: dict,
):
""""""
super().__init__(cta_engine, strategy_name, vt_symbol, setting)
self.rsi_long = 50 + self.rsi_signal
self.rsi_short = 50 - self.rsi_signal
self.bg5 = BarGenerator(self.on_bar, 5, self.on_5min_bar) #BarGenerator，ArrayManager几乎是标配。
self.bg15 = BarGenerator(self.on_bar, 15, self.on_15min_bar)
self.am5 = ArrayManager()
self.am15 = ArrayManager()
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.load_bar(10) #这个不要漏掉，不然有时会报错。
def on_start(self):
"""
Callback when strategy is started.
"""
self.write_log("策略启动")
def on_stop(self):
"""
Callback when strategy is stopped.
"""
self.write_log("策略停止")
def on_tick(self, tick: TickData):
"""
Callback of new tick data update.
"""
self.bg5.update_tick(tick) #将tick推送给bargenerator,合成1分钟bar
def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
self.bg5.update_bar(bar)
self.bg15.update_bar(bar)
def on_5min_bar(self, bar: BarData):
""""""
self.cancel_all()
self.am5.update_bar(bar)
if not self.am5.inited or not self.am15.inited:
return
self.boll_up, self.boll_down = self.am5.boll(self.boll_window, self.boll_dev)
self.rsi_value = self.am5.rsi(self.rsi_window)
boll_width = self.boll_up - self.boll_down
# No position
if self.pos == 0:
self.intra_trade_high = bar.high_price
self.intra_trade_low = bar.low_price
self.long_stop = 0
self.short_stop = 0
if self.ma_trend > 0 and self.rsi_value >= self.rsi_long:
self.buy(self.boll_up, self.fixed_size, stop=True)
if self.ma_trend < 0 and self.rsi_value <= self.rsi_short:
self.short(self.boll_down, self.fixed_size, stop=True)
# Long position
elif self.pos > 0:
self.intra_trade_high = max(self.intra_trade_high, bar.high_price)
self.long_stop = (self.intra_trade_high - self.trailing_long * boll_width)
self.sell(self.long_stop, abs(self.pos), stop=True)
# Short position
else:
self.intra_trade_low = min(self.intra_trade_low, bar.low_price)
self.short_stop = (self.intra_trade_low + self.trailing_short * boll_width)
self.cover(self.short_stop, abs(self.pos), stop=True)
self.put_event()
def on_15min_bar(self, bar: BarData):
""""""
self.am15.update_bar(bar)
if not self.am15.inited:
return
self.fast_ma = self.am15.sma(self.fast_window)
self.slow_ma = self.am15.sma(self.slow_window)
if self.fast_ma > self.slow_ma:
self.ma_trend = 1
elif self.fast_ma < self.slow_ma:
self.ma_trend = -1
else:
self.ma_trend = 0
self.put_event()
def on_trade(self, trade: TradeData):
"""
Callback of new trade data update.
"""
self.put_event()
def on_order(self, order: OrderData):
"""
Callback of new order data update.
"""
pass
def on_stop_order(self, stop_order: StopOrder):
"""
Callback of stop order update.
"""
pass
三、cinco策略（动态仓位管理）
代码
from vnpy.app.cta_strategy import (
CtaTemplate,
BarGenerator,
ArrayManager,
TickData,
BarData,
OrderData,
TradeData,
StopOrder
)
class CincoStrategy(CtaTemplate):
""""""
author = "vnpy"
boll_window = 42
boll_dev = 2.2
trailing_long = 0.65
trailing_short = 0.65
atr_window = 4
risk_level = 300
boll_up = 0
boll_down = 0
trading_size = 0
intra_trade_high = 0
intra_trade_low = 0
long_stop = 0
short_stop = 0
atr_value = 0
parameters = [
"boll_window",
"boll_dev",
"trailing_long",
"trailing_short",
"atr_window",
"risk_level"
]
variables = [
"boll_up",
"boll_down",
"trading_size",
"intra_trade_high",
"intra_trade_low",
"long_stop",
"short_stop",
"atr_value"
]
def __init__(
self,
cta_engine,
strategy_name: str,
vt_symbol: str,
setting: dict,
):
""""""
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):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.load_bar(10)
def on_start(self):
"""
Callback when strategy is started.
"""
self.write_log("策略启动")
def on_stop(self):
"""
Callback when strategy is stopped.
"""
self.write_log("策略停止")
def on_tick(self, tick: TickData):
"""
Callback of new tick data update.
"""
self.bg.update_tick(tick)
def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
self.bg.update_bar(bar)
def on_15min_bar(self, bar: BarData):
""""""
self.cancel_all()
self.am.update_bar(bar)
if not self.inited:
return
self.boll_up, self.boll_down = self.am.boll(self.boll_window, self.boll_dev)
boll_width = self.boll_up - self.boll_down
if not self.pos:
self.atr_value = self.am.atr(self.atr_window)
self.trading_size = int(self.risk_level / self.atr_value)
self.intra_trade_high = bar.high_price
self.intra_trade_low = bar.low_price
self.long_stop = 0
self.short_stop = 0
self.buy(self.boll_up, self.trading_size, stop=True)
self.short(self.boll_down, self.trading_size, stop=True)
elif self.pos > 0:
self.intra_trade_high = max(self.intra_trade_high, bar.high_price)
self.long_stop = self.intra_trade_high - self.trailing_long * boll_width
self.sell(self.long_stop, abs(self.pos), stop=True)
else:
self.intra_trade_low = min(self.intra_trade_low, bar.low_price)
self.short_stop = self.intra_trade_low + self.trailing_short * boll_width
self.cover(self.short_stop, abs(self.pos), stop=True)
self.put_event()
def on_trade(self, trade: TradeData):
"""
Callback of new trade data update.
"""
self.put_event()
def on_order(self, order: OrderData):
"""
Callback of new order data update.
"""
pass
def on_stop_order(self, stop_order: StopOrder):
"""
Callback of stop order update.
"""
pass