"""
"""
from datetime import time
from vnpy.app.cta_strategy import (
CtaTemplate,
StopOrder,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)
class DualThrust(CtaTemplate):
""""""
author = "wew20000@126.com"
#input parameters
k1 = 0.5
k2 = 0.5
fixed_size = 1
#bar datas
bars = []
day_open = 0
day_high = 0
day_low = 0
day_close = 0
exit_time = time(hour=14, minute=55)
#technical analysis indicators
DT_range = 0
long_entry = 0
short_entry = 0
#status control parameters
long_entered = False
short_entered = False
#other
parameters = ["k1", "k2", "fixed_size"]
variables = ["DT_range", "long_entry", "short_entry", "exit_time"]
def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
""""""
super(DualThrust, self).__init__(
cta_engine, strategy_name, vt_symbol, setting
)
self.bg = BarGenerator(self.on_bar)
self.am = ArrayManager()
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.cancel_all()
self.bars.append(bar)
if len(self.bars) <= 2:
return
else:
self.bars.pop(0)
last_bar = self.bars[-2]
# New Day
if last_bar.datetime.date() != bar.datetime.date():
if self.day_close:
self.DT_range = self.day_high - self.day_low
self.long_entry = bar.open_price + self.k1 * self.DT_range#第一分钟K线确定上轨
self.short_entry = bar.open_price - self.k2 * self.DT_range#第一分钟K线确定下轨
#每天第一根K线收盘价为当日K线OHLC
self.day_open = bar.open_price
self.day_high = bar.high_price
self.day_low = bar.close_price
self.day_close = bar.close_price
self.long_entered = False
self.short_entered = False
# Today
else:
self.day_high = max(self.day_high, bar.high_price)
self.day_low = min(self.day_low, bar.low_price)
self.day_close = bar.close_price
# Check Data
if self.DT_range == 0:
return
# Trading
if bar.datetime.time() < self.exit_time:
#no position
if self.pos == 0:
if bar.close_price > self.day_open:
if not self.long_entered:
self.buy(self.long_entry, self.fixed_size, stop=True)
else:
if not self.short_entered:
self.short(self.short_entry, self.fixed_size, stop=True)
#long position
elif self.pos > 0:
self.long_entered = True
self.sell(self.short_entry, self.fixed_size, stop=True)
if not self.short_entered:
self.short(self.short_entry, self.fixed_size, stop=True)
#short position
elif self.pos < 0:
self.short_entered = True
self.cover(self.long_entry, self.fixed_size, stop=True)
if not self.long_entered:
self.buy(self.long_entry, self.fixed_size, stop=True)
# Close existing position
else:
if self.pos > 0:
self.sell(bar.close_price * 0.99, abs(self.pos))
elif self.pos < 0:
self.cover(bar.close_price * 1.01, abs(self.pos))
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.
"""
self.put_event()
def on_stop_order(self, stop_order: StopOrder):
"""
Callback of stop order update.
"""
pass上面是我按教程自己写的dt策略，应该和默认的策略逻辑是一样的，但是回测结果不同。开仓点位有细微差别请问是不是代码哪里有问题
