趁着假期有空，写了小时和分钟2个时间框架的多周期策略代码示例。也算填补了学习中的一个坑。
用的最简单的双均线策略来做多时间框架策略，在2小时线上金叉后，在15分钟线上的金叉买进，死叉卖出。但2小时线上死叉，即使15分钟金叉也不买进。反之亦然。
首先写策略,保存在 demo_strategy_MultiTimeframe.py：
from typing import Any
from vnpy.trader.object import(
BarData,
TickData
)
from vnpy.app.cta_strategy import(
CtaTemplate,
BarGenerator, #k线合成器
ArrayManager #K线时间序列容器，将K线数据放到容器中去
)
from vnpy.trader.constant import Interval #注意，在这加载的interval
from typing import Callable
class MultiTimeFrameStrategy(CtaTemplate):
""" """
author = "Smart Trader"
#定义参数
maTrend_fast_window = 5
maTrend_slow_window = 20
fast_window = 10
slow_window = 20
#定义变量
#长周期策略
maTrend_fast_ma0 = 0.0
maTrend_fast_ma1 = 0.0
maTrend_slow_ma0 = 0.0
maTrend_slow_ma1 = 0.0
maTrend = 0# 判断长周期是多还是空
#短周期策略
fast_ma0 = 0.0
fast_ma1 = 0.0
slow_ma0 = 0.0
slow_ma1 = 0.0
parameters = [
'maTrend_fast_window',
'maTrend_slow_window',
'fast_window',
'slow_window'
]
variables = [
'maTrend_fast_ma0',
'maTrend_fast_ma1',
'maTrend_slow_ma0',
'maTrend_slow_ma1',
'maTrend',
'fast_ma0',
'fast_ma1',
'slow_ma0',
'slow_ma1',
]
def __init__(
self,
cta_engine: Any,
strategy_name: str,
vt_symbol: str,
setting: dict,
):
super().__init__(cta_engine,strategy_name,vt_symbol,setting)
#定义2个周期的bg函数
self.bg15min = BarGenerator(self.on_bar,
window=15,
on_window_bar=self.on_15min_bar,
interval=Interval.MINUTE)
self.bg2H = BarGenerator(self.on_bar,
window=2,
on_window_bar=self.on_2H_bar,
interval=Interval.HOUR)
#定义2个周期的am
self.am15min = ArrayManager()
self.am2H = ArrayManager()
def on_init(self):
"""策略初始化"""
#注册
self.write_log('策略初始化')
self.load_bar(10)
def on_start(self):
"""启动"""
self.write_log('策略启动')
def on_stop(self):
"""停止"""
self.write_log('策略停止')
def on_tick(self,tick:TickData):
"""Tick更新"""
self.bg2H.update_tick(tick)
self.bg15min.update_tick(tick)#bg会自动判断当前分钟是否已经走完，合成1分钟线
#定义2个周期的 on_bar 函数
def on_bar(self,bar:BarData):
"""k线更新,切记先更新长周期，再更新短周期"""
self.bg2H.update_bar(bar)#将1小时K线合成1小时K线
self.bg15min.update_bar(bar)#将1分钟K线合成15分钟K线
#定义2个周期的on_hour或者min_bar函数
def on_2H_bar(self,bar:BarData):#执行2H的K线，只做判断，不做买卖
am = self.am2H
am.update_bar(bar)
if not am.inited:
return
fast_ma = am.sma(self.maTrend_fast_window, array= True)
self.maTrend_fast_ma0= fast_ma[-1]
self.maTrend_fast_ma1= fast_ma[-2]
slow_ma = am.sma(self.maTrend_slow_window, array= True)
self.maTrend_slow_ma0= slow_ma[-1]
self.maTrend_slow_ma1= slow_ma[-2]
#判断均线定义
cross_over = (self.maTrend_fast_ma0>= self.maTrend_slow_ma0 and
self.maTrend_fast_ma1< self.maTrend_slow_ma1)
cross_below = (self.maTrend_fast_ma0<= self.maTrend_slow_ma0 and
self.maTrend_fast_ma1 > self.maTrend_slow_ma1)
if cross_over:
self.maTrend=1
elif cross_below:
self.maTrend=-1
def on_15min_bar(self,bar:BarData):#执行15分钟的K线 做出判断并且发出买卖信号
self.cancel_all
am = self.am15min
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 = (self.fast_ma0>= self.slow_ma0 and
self.fast_ma1< self.slow_ma1)
cross_below = (self.fast_ma0<= self.slow_ma0 and
self.fast_ma1 > self.slow_ma1)
if cross_over :
price = bar .close_price + 5
if not self.pos and self.maTrend>0: #如果金叉，并且大周期金叉
self.buy(price,1)
elif self.pos < 0:
self.cover(price,1)
if self.maTrend>0: #如果金叉，并且大周期金叉
self.buy(price,1)
elif cross_below:
price = bar .close_price - 5
if not self.pos and self.maTrend<0:
self.short(price,1)
elif self.pos > 0:
self.sell(price,1)
if self.maTrend<0:
self.short(price,1)
#更新图形界面
self.put_event()
然后写回测文件，保存在demo_backtesting_MutiTimeframe.py:
from vnpy.app.cta_strategy.backtesting import BacktestingEngine, OptimizationSetting
from demo_strategy_MultiTimeframe import MultiTimeFrameStrategy
from datetime import datetime
if __name__ == "__main__":
engine = BacktestingEngine()
engine.set_parameters(
vt_symbol="IF88.CFFEX",
interval="1m",
start=datetime(2010, 1, 1),
end=datetime(2019, 4, 30),
rate=0.3/10000,
slippage=0.2,
size=300,
pricetick=0.2,
capital=1_000_000,
)
engine.add_strategy(MultiTimeFrameStrategy, { #注意下面赋值的写法，每个变量都要带引号
'maTrend_fast_window' : 5,
'maTrend_slow_window' : 20,
'fast_window' : 5,
'slow_window' : 20 })
engine.load_data()
engine.run_backtesting()
df = engine.calculate_result()
engine.calculate_statistics()
engine.show_chart()
