from vnpy.app.cta_strategy import (
CtaTemplate,
StopOrder,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)
from vnpy.trader.constant import Direction
class ChatGptDt(CtaTemplate):
""""""
author = "vn.py"
vt_symbol = "IF2305.CFFEX"
fixed_size = 1
atr_length = 22
atr_ma_length = 10
k1 = 0.5
k2 = 0.5
trading_window = 30
atr_value = 0
atr_ma_value = 0
range_up = 0
range_down = 0
long_entry = 0
long_exit = 0
short_entry = 0
short_exit = 0
trading_window_high = 0
trading_window_low = 0
exit_long = False
exit_short = False
parameters = [
"fixed_size",
"atr_length",
"atr_ma_length",
"k1",
"k2",
"trading_window",
]
variables = [
"atr_value",
"atr_ma_value",
"range_up",
"range_down",
"long_entry",
"long_exit",
"short_entry",
"short_exit",
"trading_window_high",
"trading_window_low",
"exit_long",
"exit_short",
]
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()
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.am.update_bar(bar)
if not self.am.inited:
return
self.calculate_atr()
self.calculate_range()
self.calculate_entry_exit()
self.calculate_trading_window()
self.calculate_exit()
if self.pos == 0:
self.entry_long()
self.entry_short()
elif self.pos > 0:
self.exit_long()
self.entry_short()
elif self.pos < 0:
self.exit_short()
self.entry_long()
self.put_variables_event()
def calculate_atr(self):
""""""
atr_array = self.am.atr(self.atr_length, array=True)
self.atr_value = atr_array[-1]
self.atr_ma_value = atr_array[-self.atr_ma_length:].mean()
def calculate_range(self):
""""""
self.range_up = self.k1 * self.atr_value +self.am.high_array[-self.trading_window:].max()
self.range_down = self.k2 * self.atr_value - self.am.low_array[-self.trading_window:].min()
def calculate_entry_exit(self):
""""""
self.long_entry = self.range_up
self.long_exit = self.atr_ma_value
self.short_entry = self.range_down
self.short_exit = self.atr_ma_value
def calculate_trading_window(self):
""""""
self.trading_window_high = self.am.high_array[
-self.trading_window:
].max()
self.trading_window_low = self.am.low_array[
-self.trading_window:
].min()
def calculate_exit(self):
""""""
self.exit_long = (
self.pos > 0
and self.am.low_array[-1] < self.trading_window_low
)
self.exit_short = (
self.pos < 0
and self.am.high_array[-1] > self.trading_window_high
)
def entry_long(self):
""""""
if self.close_price > self.long_entry:
self.buy(self.long_entry, self.fixed_size, True)
def exit_long(self):
""""""
if self.exit_long:
self.sell(self.long_exit, abs(self.pos), True)
def entry_short(self):
""""""
if self.close_price < self.short_entry:
self.short(self.short_entry, self.fixed_size, True)
def exit_short(self):
""""""
if self.exit_short:
self.cover(self.short_exit, abs(self.pos), True)