vn.py量化社区
By Traders, For Traders.
Member
avatar
加入于:
帖子: 54
声望: 3

如题,如果我的策略是在4h上确定方向,在15分钟上确定买卖点。我想问一下写法.
是否应该这么写?简单来说就是bg,am都按照数据源为minute还是hour定义2套?
这样的话,具体数据输入的时候是怎么循环?

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_15bar,
                window=15, 
                on_window_bar=self.on_15min_bar, 
                interval='1m') 

    self.bg4H = BarGenerator(self.on_4bar,
                window=4, 
                on_window_bar=self.on_4H_bar, 
                interval='1h') 

#定义2个周期的am
    self.am15min = ArrayManager()
    self.am4H = ArrayManager()
"""以下省略"""


#定义2个周期的 on_bar 函数
def on_15_bar(self,bar:BarData):
    """k线更新"""
    self.bg15min.update_bar(bar)#将1分钟K线合成15分钟K线

def on_4_bar(self,bar:BarData):#执行4H的K线
    self.bg4H.update_bar(bar)#将1小时K线合成4小时K线
    am = self.am

#定义2个周期的on_hour或者min_bar函数
def on_15min_bar(self,bar:BarData):#执行15分钟的K线
    am = self.am15min

    am.update_bar(bar)
    if not am.inited:
        return
    fast_ma = am.sma(self.fast_window, array= True)

def on_4H_bar(self,bar:BarData):#执行4H的K线
    am = self.am4H

    am.update_bar(bar)
    if not am.inited:
        return
    fast_ma = am.sma(self.fast_window, array= True)
Administrator
avatar
加入于:
帖子: 3566
声望: 189

注意interval要使用常量Interval:

from vnpy.trader.constant import Interval

Interval.MINUTE或者Interval.HOUR

然后直接在1分钟K线回调on_bar函数里:

def on_bar(self, bar: BarData):
    self.bg15min.update_bar(bar)
    self.bg4H.update_bar(bar)
Member
avatar
加入于:
帖子: 54
声望: 3

用Python的交易员 wrote:

注意interval要使用常量Interval:

from vnpy.trader.constant import Interval

Interval.MINUTE或者Interval.HOUR

然后直接在1分钟K线回调on_bar函数里:

def on_bar(self, bar: BarData):
    self.bg15min.update_bar(bar)
    self.bg4H.update_bar(bar)

抱歉,没看懂您的解答,您的意思仍旧基于1m对K线进行合成,所以应该是从1分钟合成240min线(4h)和15min吧?用不到Interval.HOUR吧?

另外,from vnpy.trader.constant import Interval 用在哪里?在bg和am的定义中我没看到需要用Interval.MINUTE的地方。
bg的定义是否如下,这样在on_bar中就可以引用了,对吗?谢谢啦!

    self.bg15min = BarGenerator(self.on_bar,
                window=15, 
                on_window_bar=self.on_15min_bar, 
                interval=Interval.MINUTE) 

    self.bg4H = BarGenerator(self.on_bar,
                window=240, 
                on_window_bar=self.on_4H_bar, 
                interval=Interval.MINUTE) 
Administrator
avatar
加入于:
帖子: 3566
声望: 189
    self.bg4H = BarGenerator(self.on_bar,
                window=4, 
                on_window_bar=self.on_4H_bar, 
                interval=Interval.HOUR)

分钟线合成不允许window超过60,4小时线应该选择Interval.HOUR,同时window为4。

Member
avatar
加入于:
帖子: 54
声望: 3

趁着假期有空,写了小时和分钟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()
Member
avatar
加入于:
帖子: 36
声望: 1

好例子,正在找跨周期的写法,赞!

© 2015-2019 上海韦纳软件科技有限公司
备案服务号:沪ICP备18006526号-3