在台式机上正常运行的策略，放到新买的笔记本电脑上运行没有成交记录。
在笔记本上运行的结果是这样的：
在台式机上正常运行的策略，放到新买的笔记本电脑上运行没有成交记录。
在笔记本上运行的结果是这样的：
可以在你的笔记本上jupyter跑跑试试看
xiaohe wrote:
可以在你的笔记本上jupyter跑跑试试看
你说的jupyter 是指非图形界面运行模式吗？
Python的jupyter notebook回测，代码参考这里：https://github.com/vnpy/vnpy/tree/master/examples/cta_backtesting
李春宝 wrote:
用Python的交易员 wrote:
Python的jupyter notebook回测，代码参考这里：https://github.com/vnpy/vnpy/tree/master/examples/cta_backtesting
还是没有成交记录，也不报错，还是没有查出原因。
from vnpy.app.cta_strategy import (
CtaTemplate,
BarGenerator,
ArrayManager,
TickData,
BarData,
OrderData,
TradeData,
StopOrder
)class cash_dispenser(CtaTemplate):
""""""#开发都姓名 author = "LI CHUNBAO" strategy_name = "均线交叉" #定义参数 fast_window = 10 slow_window = 50 # fixed_sl = 5 #移动止损浮动金额 fixed_size = 1 #交易手数 long_period = 30 medium_period = 10 short_period = 4#定义变量 ma_trend = 0 macd_trend = 0 long_entry = 0 #多头入场价格 long_sl = 0 #多头止损价格 short_entry = 0 #空头入场价格 short_sl = 0 #空头止损价格#添加参数和变量到对应的列表 parameters = [ "fast_window", "slow_window", "fixed_size", "long_period", "medium_period", "short_period" ] variables = [ "ma_trend", "macd_trend", "fixed_size", "long_entry", "long_sl", "short_entry", "short_sl" ] def __init__( self, cta_engine, strategy_name: str, vt_symbol: str, setting: dict, ): """""" super().__init__(cta_engine,strategy_name,vt_symbol,setting) self.bg_short_min = BarGenerator(self.on_bar,self.short_period,self.on_short_min_bar) self.am_short_min = ArrayManager() self.bg_medium_min = BarGenerator(self.on_bar,self.medium_period,self.on_medium_min_bar) self.am_medium_min = ArrayManager() self.bg_long_min = BarGenerator(self.on_bar,self.long_period,self.on_long_min_bar) self.am_long_min = 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.
"""
def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
self.bg_long_min.update_bar(bar)
self.bg_medium_min.update_bar(bar)
self.bg_short_min.update_bar(bar)
def on_short_min_bar(self,bar:BarData):
"""short period分钟回调函数"""
self.cancel_all()self.am_short_min.update_bar(bar) # 计算技术指标 self.fast_short_min_ma = self.am_short_min.sma(self.fast_window,array=True) self.fast_short_min_ma0 =self.fast_short_min_ma[-1] self.fast_short_min_ma1 =self.fast_short_min_ma[-2] self.slow_short_min_ma = self.am_short_min.sma(self.slow_window,array=True) self.slow_short_min_ma0 =self.slow_short_min_ma[-1] self.slow_short_min_ma1 =self.slow_short_min_ma[-2] # 判断均线交叉 self.cross_over_short_min = (self.fast_short_min_ma0 >= self.slow_short_min_ma0 and self.fast_short_min_ma1 < self.slow_short_min_ma1) self.cross_below_short_min = (self.fast_short_min_ma0 <= self.slow_short_min_ma0 and self.fast_short_min_ma1 > self.slow_short_min_ma1) if self.pos > 0: if self.cross_below_short_min: self.price = bar.close_price self.sell(self.price, abs(self.pos), stop = True) #卖出平仓 elif self.pos < 0: if self.cross_over_short_min: self.price = bar.open_price self.cover(self.price, abs(self.pos), stop = True) #买入平仓 self.put_event() def on_medium_min_bar(self,bar:BarData): """medium period分钟回调函数""" self.cancel_all() self.am_medium_min.update_bar(bar) if not self.am_medium_min.inited: return # 计算技术指标 self.fast_medium_min_ma = self.am_medium_min.sma(self.fast_window,array=True) self.fast_medium_min_ma0 =self.fast_medium_min_ma[-1] self.fast_medium_min_ma1 =self.fast_medium_min_ma[-2] self.slow_medium_min_ma = self.am_medium_min.sma(self.slow_window,array=True) self.slow_medium_min_ma0 =self.slow_medium_min_ma[-1] self.slow_medium_min_ma1 =self.slow_medium_min_ma[-2] # 判断均线交叉 self.cross_over_medium_min = (self.fast_medium_min_ma0 >= self.slow_medium_min_ma0 and self.fast_medium_min_ma1 < self.slow_medium_min_ma1) self.cross_below_medium_min = (self.fast_medium_min_ma0 <= self.slow_medium_min_ma0 and self.fast_medium_min_ma1 > self.slow_medium_min_ma1) # print(f"金叉:{self.cross_over}",f"死叉:{self.cross_below}") if self.pos == 0: if self.ma_trend > 0 and self.macd_trend > 0: if self.cross_over_medium_min: self.price = bar.close_price self.buy(self.price, self.fixed_size, stop = True) #买入开仓 elif self.ma_trend < 0 and self.macd_trend < 0: if self.cross_below_medium_min: self.price = bar.open_price self.short(self.price, self.fixed_size, stop = True) #卖出开仓 self.put_event() def on_long_min_bar(self,bar:BarData): """long_period回调函数""" self.am_long_min.update_bar(bar) if not self.am_long_min.inited: return self.slow_long_min_ma = self.am_long_min.sma(self.slow_window) if bar.low_price > self.slow_long_min_ma : self.ma_trend = 1 elif bar.low_price < self.slow_long_min_ma : self.ma_trend = -1 else: self.ma_trend = 0 # macd指标 self.diff,self.dea,self.macd = self.am_long_min.macd(12,26,9) if self.diff > 0 and self.dea > 0: self.macd_trend = 1 elif self.diff < 0 and self.dea < 0: self.macd_trend = -1 else: self.macd_trend = 0 self.put_event() def on_trade(self, trade: TradeData): """ Callback of new trade data update. """ # msg = f"新的成交, 策略{self.strategy_name}, 方向{trade.direction}, 开平{trade.offset}, 当前仓位{self.pos}" # self.send_email(msg) self.put_event()
def on_order(self, order: OrderData):
"""
Callback of new order data update.
"""
pass
def on_stop_order(self, stop_order: StopOrder):
"""
Callback of stop order update.
"""
pass
