from vnpy_ctastrategy import (
CtaTemplate,
StopOrder,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)
from .my_strategies_tool import NewBarGennerator
from vnpy.trader.constant import Interval
class DemoStrategy(CtaTemplate):
"""演示用的简单双均线"""
# 策略作者
author = "Smart Trader"
# 定义参数
fast_window = 10
slow_window = 20
fast_window1 = 10
slow_window1 = 20
k=5
# 定义变量
fast_ma0 = 0.0
fast_ma1 = 0.0
slow_ma0 = 0.0
slow_ma1 = 0.0
fast_ma01 = 0.0
fast_ma11 = 0.0
slow_ma01 = 0.0
slow_ma11 = 0.0
cross_over2=True
cross_below2=True
# 添加参数和变量名到对应的列表
parameters = ["fast_window", "slow_window","k","fast_window", "slow_window"]
variables = ["fast_ma0", "fast_ma1", "slow_ma0", "slow_ma1","fast_ma01", "fast_ma11", "slow_ma01", "slow_ma11","cross_over2","cross_below2"]
def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
""""""
super().__init__(cta_engine, strategy_name, vt_symbol, setting)
# K线合成器：从Tick合成分钟K线用xzsde
self.bg = BarGenerator(self.on_bar)
self.bg15=NewBarGennerator(self.on_bar, window=self.k, on_window_bar=self.on_5min_bar, interval=Interval.MINUTE)
# 时间序列容器：计算技术指标用
self.am = ArrayManager()
self.am2=ArrayManager()
def on_init(self):
"""
当策略被初始化时调用该函数。
"""
# 输出个日志信息，下同
self.write_log("策略初始化")
# 加载10天的历史数据用于初始化回放
self.load_bar(10)
def on_start(self):
"""
当策略被启动时调用该函数。
"""
self.write_log("策略启动")
# 通知图形界面更新（策略最新状态）
# 不调用该函数则界面不会变化
self.put_event()
def on_stop(self):
"""
当策略被停止时调用该函数。
"""
self.write_log("策略停止")
self.put_event()
def on_tick(self, tick: TickData):
"""
通过该函数收到Tick推送。
"""
self.bg.update_tick(tick)
self.bg15.update_tick(tick)
def on_bar(self, bar: BarData):
"""
通过该函数收到新的1分钟K线推送。
"""
self.bg.update_bar(bar)
self.bg15.update_bar(bar)
am = self.am
am.update_bar(bar)
if not am.inited:
return
fast_ma1 = am.sma(self.fast_window1, array=True)
self.fast_ma01 = fast_ma1[-1] # T时刻数值
self.fast_ma11 = fast_ma1[-2] # T-1时刻数值
# 计算慢速均线
slow_ma1 = am.sma(self.slow_window1, array=True)
self.slow_ma01 = slow_ma1[-1]
self.slow_ma11 = slow_ma1[-2]
# 判断是否金叉
self.cross_over2 = (self.fast_ma01 > self.slow_ma01 and
self.fast_ma11 < self.slow_ma11)
# 判断是否死叉
self.cross_below2 = (self.fast_ma01 < self.slow_ma01 and
self.fast_ma11 > self.slow_ma11)
def on_5min_bar(self, bar: BarData):
am2=self.am2
# 更新K线到时间序列容器中
am2.update_bar(bar)
# 若缓存的K线数量尚不够计算技术指标，则直接返回
if not am2.inited:
return
# 计算快速均线
fast_ma = am2.sma(self.fast_window, array=True)
self.fast_ma0 = fast_ma[-1] # T时刻数值
self.fast_ma1 = fast_ma[-2] # T-1时刻数值
# 计算慢速均线
slow_ma = am2.sma(self.slow_window, array=True)
self.slow_ma0 = slow_ma[-1]
self.slow_ma1 = slow_ma[-2]
# 判断是否金叉
cross_over = (self.fast_ma0 > self.slow_ma0 and
self.fast_ma1 < self.slow_ma1)
# 判断是否死叉
cross_below = (self.fast_ma0 < self.slow_ma0 and
self.fast_ma1 > self.slow_ma1)
# 如果发生了金叉
if cross_over and self.cross_over2:
# 为了保证成交，在K线收盘价上加5发出限价单
price = bar.close_price + 5
# 当前无仓位，则直接开多
if self.pos == 0:
self.buy(price, 1)
# 当前持有空头仓位，则先平空，再开多
elif self.pos < 0:
self.cover(price, 1)
self.buy(price, 1)
# 如果发生了死叉
elif cross_below and self.cross_below2:
price = bar.close_price - 5
# 当前无仓位，则直接开空
if self.pos == 0:
self.short(price, 1)
# 当前持有空头仓位，则先平多，再开空
elif self.pos > 0:
self.sell(price, 1)
self.short(price, 1)
self.put_event()
def on_order(self, order: OrderData):
"""
通过该函数收到委托状态更新推送。
"""
pass
def on_trade(self, trade: TradeData):
"""
通过该函数收到成交推送。
"""
# 成交后策略逻辑仓位发生变化，需要通知界面更新。
self.put_event()
def on_stop_order(self, stop_order: StopOrder):
"""
通过该函数收到本地停止单推送。
"""
pass
16:53:02 策略文件加载完成
16:53:18 ----------------------------------------
16:53:18 开始加载历史数据
16:53:18 加载进度：# [0%]
16:53:18 加载进度：# [10%]
16:53:19 加载进度：## [20%]
16:53:20 加载进度：### [30%]
16:53:21 加载进度：#### [39%]
16:53:21 加载进度：##### [49%]
16:53:22 加载进度：###### [59%]
16:53:23 加载进度：####### [69%]
16:53:23 加载进度：######## [79%]
16:53:23 加载进度：######### [89%]
16:53:23 加载进度：########## [99%]
16:53:23 历史数据加载完成，数据量：54945
16:53:23 触发异常，回测终止
16:53:23 Traceback (most recent call last):
File "C:\veighna_studio\lib\site-packages\vnpy_ctastrategy\backtesting.py", line 228, in run_backtesting
self.callback(data)
File "C:\Users\1\strategies\demo_strategy.py", line 98, in on_bar
self.bg.update_bar(bar)
File "C:\veighna_studio\lib\site-packages\vnpy\trader\utility.py", line 266, in update_bar
self.update_bar_minute_window(bar)
File "C:\veighna_studio\lib\site-packages\vnpy\trader\utility.py", line 302, in update_bar_minute_window
if not (bar.datetime.minute + 1) % self.window:
ZeroDivisionError: integer division or modulo by zero
16:53:23 开始计算逐日盯市盈亏
16:53:23 成交记录为空，无法计算
16:53:23 开始计算策略统计指标
16:53:23 策略统计指标计算完成
