from vnpy_ctastrategy import (
CtaTemplate,
StopOrder,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)
class demo3(CtaTemplate):
author="action华"
fixed_size=4
k1=0.5
k2=0.005
k3=2
range=[]
dev=0
intra_trade_high=0
intra_trade_low=0
long_stop=0
short_stop=0
parameters = ["k1", "k2", "fixed_size"]
variables = [ "dev","intra_trade_high","ntra_trade_low","long_stop", "short_stop"]
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(5)
self.bars = []
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.am.update_bar(bar)
if not self.inited:
return
w=abs(bar.close_price-bar.open_price)
self.range.append(w)
if len(self.range)<5:
return
else:
self.range.pop(0)
self.dev=(sum(self.range)-self.range[-1])/4
if self.range[-1]<self.k3*self.dev:
return
"""入场"""
if self.pos==0:
self.intra_trade_high=bar.high_price
self.intra_trade_low=bar.low_price
if 0<bar.open_price-bar.close_price<self.k1*self.range[-1]:
self.buy(bar.close_price,self.fixed_size, stop=True)
if 0<bar.close_price-bar.open_price<self.k1*self.range[-1]:
self.short(bar.close_price,self.fixed_size, stop=True)
elif self.pos>0:
self.intra_trade_high=max(self.intra_trade_high,bar.high_price)
self.intra_trade_low=bar.low_price
self.long_stop=self.intra_trade_high-self.k2*bar.high_price
self.sell(self.long_stop, abs(self.pos), True)
elif self.pos < 0:
self.intra_trade_high = bar.high_price
self.intra_trade_low = min(self.intra_trade_low, bar.low_price)
self.short_stop = self.intra_trade_low +self.k2*bar.high_price
self.cover(self.short_stop, abs(self.pos), True)
self.put_event()
def on_order(self, order: OrderData):
"""
Callback of new order data update.
"""
pass
def on_trade(self, trade: TradeData):
"""
Callback of new trade data update.
"""
print("是否交易")
self.put_event()
def on_stop_order(self, stop_order: StopOrder):
"""
Callback of stop order update.
"""
pass
