from vnpy.app.cta_strategy import (
CtaTemplate,
BarGenerator,
ArrayManager,
Direction
)
from vnpy.trader.object import (
BarData,
TickData,
TradeData,
OrderData,
)
from typing import Any
from vnpy.trader.constant import Interval
from datetime import time
import datetime
class NewAtrStrategy(CtaTemplate):
""""""
author = ""
atr_window = 30
atr_cs = 6
zs = 20
fix_size = 1
bars = []
up_line = 0.0
down_line = 0.0
mid_line = 0.0
atr_value = 0.0
short_entry = 0.0
long_entry = 0.0
long_stop = 0.0
short_stop = 0.0
parameters = [
'atr_window',
'atr_cs',
'zs',
'fix_size'
]
variables = [
'up_line',
'down_line',
'atr_value'
]
def __init__(
self,
cta_engine: Any,
strategy_name: str,
vt_symbol: str,
setting: dict,
):
super().__init__(cta_engine,strategy_name,vt_symbol,setting)
self.bg5 = BarGenerator(self.on_bar,
window=5,
on_window_bar=self.on_5minutes_bar,
interval=Interval.MINUTE)
self.am5 = ArrayManager()
self.bars = []
def on_init(self):
"""策略引擎初始化"""
self.write_log("策略初始化")
self.load_bar(20)
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):
self.bg5.update_tick(tick)
def on_bar(self,bar:BarData):
self.bg5.update_bar(bar)
def on_5minutes_bar(self,bar:BarData):
self.cancel_all
am5 = self.am5
am5.update_bar(bar)
self.bars.append(bar)
if len(self.bars) <= 2:
return
else:
self.bars.pop(0)
last_bar = self.bars[-2]
if not am5.inited:
return
if last_bar.datetime.date() != bar.datetime.date():
self.mid_line = last_bar.close_price
art_array = am5.atr(self.atr_window, array=True)
self.atr_value = art_array[-1]
self.up_line = self.mid_line + self.atr_cs * self.atr_value
self.down_line = self.mid_line - self.atr_cs * self.atr_value
if self.pos == 0:
self.long_entry = 0
self.long_stop = 0
self.short_entry = 0
self.short_stop = 0
#上穿up_line做多
if bar.close_price >= self.up_line and last_bar.close_price < self.up_line:
price = bar.close_price
self.buy(price, self.fix_size)
#下穿down_line做空
elif bar.close_price <= self.down_line and last_bar.close_price > self.down_line:
price = bar.close_price
self.short(price, self.fix_size)
elif self.pos > 0:
#固定止损，用停止单
if self.long_stop:
self.sell(self.long_stop,self.pos,stop=True)
if bar.close_price <= self.down_line:
self.sell(bar.close_price,abs(self.pos),stop=True)
elif self.pos < 0:
#固定止损
if self.short_stop:
self.cover(self.short_stop,abs(self.pos),stop=True)
if bar.close_price >= self.up_line:
self.cover(bar.close_price,abs(self.pos),stop=True)
self.put_event()
def on_trade(self,trade: TradeData):
if self.pos != 0:
if trade.direction == Direction.LONG:
self.long_entry = trade.price
self.long_stop = self.long_entry * (1000-self.zs)/1000
else:
self.short_entry = trade.price
self.short_stop = self.short_entry * (1000 + self.zs)/1000
self.put_event()
def on_order(self,order: OrderData):
pass
def on_stop_order(self,order: OrderData):
pass
回测的问题是：
一：会报错
二：交易量巨大
求帮助啊，谢谢！
