基于统计的交易策略简易实现VNPY的示例分析

这篇文章将为大家详细讲解有关基于统计的交易策略简易实现VNPY的示例分析，文章内容质量较高，因此小编分享给大家做个参考，希望大家阅读完这篇文章后对相关知识有一定的了解。

交易思维是基于历史数据中，一组数据比如100天中，K线中最高点或者最低点相对于开始价位价差点差，再利用numpy的函数numpy.percentile(), 计算在比如95%机会，最高点或者最低点的点差数字。如果点差是5个点，就可以认为下一根K线也有95%概率有5个点受益。

尝试在VNPY实现。

思路整理：

1.入场：如果最近N（30）个D分钟k线，通过下面代码计算，分析对于概率prb比如90%，如果存在一个点差大于TickValueLimit一个值TickValue，说明过去N个分钟，有P的概率，bar开始下单，在bar中有最高点或者最低点获得TickValue。那么在下个bar开始时候，买入。

2.出场，如果到达持有价格POSprice +/- TickValue, 则卖出；重新进行入场分析。如果这个bar中间没到达目标价格，在bar结束时候分析是否还满足入场条件，如果继续满足则持有，否则平仓，如果是反向，则反向开单。

3.止损，如果在持有时候，下跌到反向POSPrice +/- Multiple * TickValue 价格时候，平仓。Multiple 随着时间增加逐渐减少。

示例代码如下，最后很遗憾，回测效果非常不好。

from __future__ import divisionfrom vnpy.trader.vtConstant import EMPTY_STRING, EMPTY_FLOAT, OFFSET_OPEN,OFFSET_CLOSEfrom vnpy.trader.app.ctaStrategy.ctaTemplate import (CtaTemplate,                                                     BarGenerator,                                                     ArrayManager)import numpy as npfrom datetime import datetime, time########################################################################class PercentileStrategy(CtaTemplate):    """MACD策略Demo"""    className = 'PercentileStrategy'    author = u'BillyZhang'    fixedSize = 1    # 策略参数    calWindow = 15    percentile = 95    tickValueLimit = 5    Multiple = 0.8    # 策略变量    p = 0    tickValue = 0    tradeSign = 0    tickValueHigh = 0    tickValueLow = 0    longStop = 0  # 多头止损    shortStop = 0  # 空头止损    margin = 0    lowerLimit = 0    upperLimit = 50000    # 时间    initDays = 0    DAY_START = time(9, 10)  # 日盘启动和停止时间    DAY_END = time(14, 55)    NIGHT_START = time(21, 10)  # 夜盘启动和停止时间    NIGHT_END = time(10, 55)    # 参数列表，保存了参数的名称    paramList = ['name',                 'className',                 'author',                 'vtSymbol',                 'initDays',                 'fixedSize',                 'calWindow',                 'percentile',                 'tickValueLimit',                 'Multiple'                 ]    # 变量列表，保存了变量的名称    varList = ['inited',               'trading',               'pos',               'longStop',               'shortStop',               'posPrice',               'lowerLimit',               'p',               'tickValue',               'tradeSign',               'tickValueHigh',               'tickValueLow'                ]    # 同步列表，保存了需要保存到数据库的变量名称    syncList = ['pos',                'posPrice',                'longStop',                'shortStop'                ]    # ----------------------------------------------------------------------    def __init__(self, ctaEngine, setting):        """Constructor"""        super(PercentileStrategy, self).__init__(ctaEngine, setting)        self.am = ArrayManager(size = self.calWindow)        # 注意策略类中的可变对象属性（通常是list和dict等），在策略初始化时需要重新创建，        # 否则会出现多个策略实例之间数据共享的情况，有可能导致潜在的策略逻辑错误风险，        # 策略类中的这些可变对象属性可以选择不写，全都放在__init__下面，写主要是为了阅读        # 策略时方便（更多是个编程习惯的选择）    # ----------------------------------------------------------------------    def onInit(self):        """初始化策略（必须由用户继承实现）"""        self.writeCtaLog(u'%s策略初始化' % self.name)        initData = self.loadBar(self.initDays)        for bar in initData:            self.onBar(bar)        self.putEvent()    # ----------------------------------------------------------------------    def onStart(self):        """启动策略（必须由用户继承实现）"""        if self.pos == 0:            self.writeCtaLog(u'%s策略启动' % self.name)        # 当前无仓位，发送开仓委托        # 持有多头仓位        self.putEvent()    # ----------------------------------------------------------------------    def onStop(self):        """停止策略（必须由用户继承实现）"""        self.writeCtaLog(u'%s策略停止' % self.name)        self.putEvent()    # ----------------------------------------------------------------------    def onTick(self, tick):        """收到行情TICK推送（必须由用户继承实现）"""        if self.lowerLimit == 0 or self.upperLimit == 0:            self.lowerLimit = tick.lowerLimit            self.upperLimit = tick.upperLimit        self.bg.updateTick(tick)    # ----------------------------------------------------------------------    def onBar(self, bar):        """收到Bar推送（必须由用户继承实现）"""        #如果是当然最后5分钟，略过        am = self.am        am.updateBar(bar)        if not am.inited:            return        # currentTime = datetime.now().time()        currentTime = time(9,20)        #计算p，和tickValue        MaxHigh = am.high / am.open        MaxLow = am.low / am.open        MaxClose = am.close / am.open        lpHigh = np.percentile(MaxHigh, 100 - self.percentile)        lpLow = np.percentile(MaxLow,  self.percentile)        self.tickValueHigh = abs(bar.open - bar.open*lpHigh)        self.tickValueLow = abs(bar.open - bar.open * lpLow)        if self.tickValueHigh > self.tickValueLow and self.tickValueHigh > self.tickValueLimit:            self.tradeSign = 1        elif self.tickValueHigh < self.tickValueLow and self.tickValueLow > self.tickValueLimit:            self.tradeSign = -1        else:            self.tradeSign = 0        # 平当日仓位, 如果当前时间是结束前日盘15点28分钟,或者夜盘10点58分钟，如果有持仓，平仓。        if ((currentTime >= self.DAY_START and currentTime <= self.DAY_END) or            (currentTime >= self.NIGHT_START and currentTime <= self.NIGHT_END)):            if self.pos == 0:                if self.tradeSign == 0:                    pass                elif self.tradeSign == 1 and bar.close > self.lowerLimit:                    self.buy(bar.close + 5,self.fixedSize,False)                elif self.tradeSign == -1 and bar.close < self.upperLimit:                    self.short(bar.close - 5,self.fixedSize,False)            elif self.pos > 0:                if self.tradeSign == 1 or self.tradeSign == 0:                    pass                elif self.tradeSign == -1:                    self.sell(bar.close-5, abs(self.pos), False)            elif self.pos < 0:                if self.tradeSign == -1 or self.tradeSign == 0:                    pass                elif self.tradeSign ==1:                    self.cover(bar.close+5, abs(self.pos), False)        else:            if self.pos > 0:                self.sell(bar.close-5, abs(self.pos), False)            elif self.pos < 0:                self.cover(bar.close+5, abs(self.pos), False)            elif self.pos == 0:                return    # ----------------------------------------------------------------------    def onOrder(self, order):        """收到委托变化推送（必须由用户继承实现）"""        # 对于无需做细粒度委托控制的策略，可以忽略onOrder        pass    # ----------------------------------------------------------------------    def onTrade(self, trade):        # 发出状态更新事件        """收到成交推送（必须由用户继承实现）"""        # 对于无需做细粒度委托控制的策略，可以忽略onOrder        if trade.offset == OFFSET_OPEN:            self.posPrice = trade.price            if self.tradeSign == 1:                self.sell(self.posPrice + self.tickValueHigh,abs(self.pos),False)                self.sell(self.posPrice - self.Multiple*self.tickValueHigh, abs(self.pos), True)            elif self.tradeSign == -1:                self.cover(self.posPrice - self.tickValueLow, abs(self.pos), False)                self.cover(self.posPrice + self.Multiple*self.tickValueLow, abs(self.pos),True)        elif trade.offset == OFFSET_CLOSE:            self.cancelAll()            self.tradeSign = 0            # 同步数据到数据库        self.saveSyncData()    # ----------------------------------------------------------------------    def onStopOrder(self, so):        """停止单推送"""        pass

关于基于统计的交易策略简易实现VNPY的示例分析就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。