我在no ui模式下运行回测,发现注释掉如下代码,也可正常运行,是不是这些代码只是与界面有关?
# parameters = ["fast_window", "slow_window"]
# variables = ["fast_ma0", "fast_ma1", "slow_ma0", "slow_ma1"]
完整代码如下:
from vnpy.app.cta_strategy.backtesting import BacktestingEngine, OptimizationSetting
from vnpy.app.cta_strategy.strategies.atr_rsi_strategy import (
AtrRsiStrategy,
)
# from vnpy.app.cta_strategy.strategies.double_ma_strategy import (
# DoubleMaStrategy,
# )
from vnpy.app.cta_strategy import (
CtaTemplate,
StopOrder,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)
# 策略
class DoubleMaStrategy(CtaTemplate):
author = "用Python的交易员"
fast_window = 10
slow_window = 20
# fast_ma0 = 0.0
# fast_ma1 = 0.0
# slow_ma0 = 0.0
# slow_ma1 = 0.0
# parameters = ["fast_window", "slow_window"]
# variables = ["fast_ma0", "fast_ma1", "slow_ma0", "slow_ma1"]
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(size=30)
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.load_bar(30)
def on_start(self):
"""
Callback when strategy is started.
"""
self.write_log("策略启动")
# self.put_event()
def on_stop(self):
"""
Callback when strategy is stopped.
"""
self.write_log("策略停止")
# self.put_event()
# 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.
"""
# print(bar.datetime,bar.open_price)
am = self.am
am.update_bar(bar)
if not am.inited:
return
fast_ma = am.sma(self.fast_window, array=True)
# self.fast_ma0 = fast_ma[-1] # 今日
# self.fast_ma1 = fast_ma[-2]
slow_ma = am.sma(self.slow_window, array=True)
# self.slow_ma0 = slow_ma[-1]
# self.slow_ma1 = slow_ma[-2]
cross_over = fast_ma[-1] > slow_ma[-1] and fast_ma[-2] < slow_ma[-2]
cross_below = fast_ma[-1] < slow_ma[-1] and fast_ma[-2] > slow_ma[-2]
if cross_over:
if self.pos == 0:
print('buy creat0',bar.vt_symbol,bar.datetime,bar.close_price)
self.buy(bar.close_price, 1)
elif self.pos < 0:
print('cover creat',bar.datetime,bar.close_price, 'pos',self.pos)
self.cover(bar.close_price, 1)
print('buy creat',bar.vt_symbol,bar.datetime,bar.close_price)
self.buy(bar.close_price, 1)
elif cross_below:
if self.pos == 0:
print('short creat0',bar.symbol,bar.vt_symbol,bar.datetime,bar.close_price)
self.short(bar.close_price, 1)
elif self.pos > 0:
print('sell creat',bar.datetime,bar.close_price)
self.sell(bar.close_price, 1)
print('short creat',bar.datetime,bar.close_price)
self.short(bar.close_price, 1)
# self.put_event()
def on_order(self, order: OrderData):
"""
Callback of new order data update.
"""
print('on order',order.datetime,'orderid',order.orderid,order.direction,order.status, 'price',order.price, 'vol',order.volume,'trade',order.traded)
def on_trade(self, trade: TradeData):
"""
Callback of new trade data update.
"""
print('on trade===>',trade.datetime,'tradeid',trade.tradeid,trade.direction,'price',trade.price,'vol',trade.volume,'orderid',trade.orderid)
# self.put_event()
def on_stop_order(self, stop_order: StopOrder):
"""
Callback of stop order update.
"""
pass
################
# 主程序开始
#####################
from datetime import datetime
engine = BacktestingEngine()
engine.set_parameters(
vt_symbol="600000.SSE",
interval="d",
start=datetime(2000, 1, 4),
end=datetime(2001, 4, 30),
rate=0.3/10000,
slippage=0.00,
size=300,
pricetick=0.0001, # 下单的时候,vnpy会把成交价格处理成与pricetick的小数位数相同,且是pricetick的整数倍。
capital=1_000_000,
)
engine.add_strategy(DoubleMaStrategy, {})
engine.load_data()
engine.run_backtesting()
df = engine.calculate_result()
engine.calculate_statistics()
# engine.show_chart()