我的策略是这样的计算3分钟和15分钟的两次ema，点击启动策略调用on_start发现通过 self.load_bar(10)数据计算的[ "double_ema_15_previous","double_ema_15_current", "double_ema_3_previous", "double_ema_3_current",]和在文华上面计算的几乎一模一样，但是随着实盘测试传入新的bar之后就开始偏移，停止策略点击开始调用on_start发现通过 self.load_bar(10)计算的又正确了。请问这个是什么原因？非常感谢。
`from vnpy_ctastrategy import (
CtaTemplate,
StopOrder,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)
from datetime import datetime
import talib
class DoubleEmaStrategy_3min_15min(CtaTemplate):
""""""
author = "用Python的交易员"
ema_period = 20
pos_size = 1
price_unit=1
double_ema_15_current = 0.0
double_ema_15_previous = 0.0
double_ema_3_current = 0.0
double_ema_3_previous = 0.0
last_double_ema_3_up = None # 用于保存上一个周期的3分钟双EMA趋势方向
last_double_ema_15_up = None # 用于保存上一个周期的15分钟双EMA趋势方向
parameters = ["ema_period", "pos_size","price_unit"]
variables = [ "double_ema_15_previous","double_ema_15_current", "double_ema_3_previous", "double_ema_3_current",]
def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
""""""
super().__init__(cta_engine, strategy_name, vt_symbol, setting)
self.bg15 = BarGenerator(self.on_bar, 15, self.on_15min_bar)
self.am15 = ArrayManager()
self.bg3 = BarGenerator(self.on_bar, 3, self.on_3min_bar)
self.am3 = 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.load_bar(10)
# 计算第一次EMA
ema_3_array = self.am3.ema(self.ema_period, array=True)
# 更新到am3以便计算第二次EMA
double_ema_3 = talib.EMA(ema_3_array, self.ema_period)
self.double_ema_3_previous = double_ema_3[-2]
self.double_ema_3_current = double_ema_3[-1]
# 计算第一次EMA
ema_15_array = self.am15.ema(self.ema_period, array=True)
# 更新到am15以便计算第二次EMA
double_ema_15 = talib.EMA(ema_15_array, self.ema_period)
self.double_ema_15_previous = double_ema_15[-2]
self.double_ema_15_current = double_ema_15[-1]
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.bg3.update_tick(tick)
self.bg15.update_tick(tick)
def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
self.bg3.update_bar(bar)
self.bg15.update_bar(bar)
self.put_event()
def on_3min_bar(self, bar: BarData):
""""""
self.cancel_all()
self.am3.update_bar(bar)
if not self.am3.inited:
return
self.update_ema_3(bar)
# 判断双EMA趋势
self.check_trend_and_execute(bar)
self.put_event()
def on_15min_bar(self, bar: BarData):
""""""
self.am15.update_bar(bar)
if not self.am15.inited:
return
self.update_ema_15(bar)
self.put_event()
def update_ema_3(self, bar: BarData):
"""
更新3分钟双EMA
"""
ema_3_array = self.am3.ema(self.ema_period, array=True)
double_ema_3 = talib.EMA(ema_3_array, self.ema_period)
self.double_ema_3_previous = double_ema_3[-2]
self.double_ema_3_current = double_ema_3[-1]
def update_ema_15(self, bar: BarData):
"""
更新15分钟双EMA
"""
ema_15_array = self.am15.ema(self.ema_period, array=True)
double_ema_15 = talib.EMA(ema_15_array, self.ema_period)
self.double_ema_15_previous = double_ema_15[-2]
self.double_ema_15_current = double_ema_15[-1]
def check_trend_and_execute(self, bar: BarData):
"""
根据双EMA趋势判断执行交易
"""
double_ema_3_up = self.double_ema_3_current > self.double_ema_3_previous
double_ema_3_down = self.double_ema_3_current < self.double_ema_3_previous
double_ema_15_up = self.double_ema_15_current > self.double_ema_15_previous
double_ema_15_down = self.double_ema_15_current < self.double_ema_15_previous
# 3分钟EMA方向变化时记录日志
if self.last_double_ema_3_up is not None and self.last_double_ema_3_up != double_ema_3_up:
if double_ema_3_up:
print(f"在{bar.datetime}, 3分钟EMA从下降变为上升")
else:
print(f"在{bar.datetime}, 3分钟EMA从上升变为下降")
# 15分钟EMA方向变化时记录日志
if self.last_double_ema_15_up is not None and self.last_double_ema_15_up != double_ema_15_up:
if double_ema_15_up:
print(f"在{bar.datetime}, 15分钟EMA从下降变为上升")
else:
print(f"在{bar.datetime}, 15分钟EMA从上升变为下降")
# 保存当前趋势方向用于下一个周期比较
self.last_double_ema_3_up = double_ema_3_up
self.last_double_ema_15_up = double_ema_15_up
# 执行交易逻辑
# 平空单
if double_ema_3_up and self.pos < 0:
self.cover(bar.close_price, abs(self.pos))
# 平多单
if double_ema_3_down and self.pos > 0:
self.sell(bar.close_price, abs(self.pos))
# 开多仓
if double_ema_3_up and double_ema_15_up:
if self.pos < 0:
self.cover(bar.close_price, abs(self.pos))
if self.pos == 0:
self.buy(bar.close_price - self.price_unit, self.pos_size,True)
print(f"多开{self.pos_size}")
# 开空仓
if double_ema_3_down and double_ema_15_down:
if self.pos > 0:
self.sell(bar.close_price, abs(self.pos))
if self.pos == 0:
self.short(bar.close_price + self.price_unit, self.pos_size,True)
print(f"空开{self.pos_size}")
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
`
