基于統(tǒng)計(jì)的交易策略簡(jiǎn)易實(shí)現(xiàn)VNPY的示例分析

這篇文章將為大家詳細(xì)講解有關(guān)基于統(tǒng)計(jì)的交易策略簡(jiǎn)易實(shí)現(xiàn)VNPY的示例分析，文章內(nèi)容質(zhì)量較高，因此小編分享給大家做個(gè)參考，希望大家閱讀完這篇文章后對(duì)相關(guān)知識(shí)有一定的了解。

交易思維是基于歷史數(shù)據(jù)中，一組數(shù)據(jù)比如100天中，K線中最高點(diǎn)或者最低點(diǎn)相對(duì)于開(kāi)始價(jià)位價(jià)差點(diǎn)差，再利用numpy的函數(shù)numpy.percentile(), 計(jì)算在比如95%機(jī)會(huì)，最高點(diǎn)或者最低點(diǎn)的點(diǎn)差數(shù)字。如果點(diǎn)差是5個(gè)點(diǎn)，就可以認(rèn)為下一根K線也有95%概率有5個(gè)點(diǎn)受益。

嘗試在VNPY實(shí)現(xiàn)。

思路整理：

1.入場(chǎng)：如果最近N（30）個(gè)D分鐘k線，通過(guò)下面代碼計(jì)算，分析對(duì)于概率prb比如90%，如果存在一個(gè)點(diǎn)差大于TickValueLimit一個(gè)值TickValue，說(shuō)明過(guò)去N個(gè)分鐘，有P的概率，bar開(kāi)始下單，在bar中有最高點(diǎn)或者最低點(diǎn)獲得TickValue。那么在下個(gè)bar開(kāi)始時(shí)候，買入。

2.出場(chǎng)，如果到達(dá)持有價(jià)格POSprice +/- TickValue, 則賣出；重新進(jìn)行入場(chǎng)分析。如果這個(gè)bar中間沒(méi)到達(dá)目標(biāo)價(jià)格，在bar結(jié)束時(shí)候分析是否還滿足入場(chǎng)條件，如果繼續(xù)滿足則持有，否則平倉(cāng)，如果是反向，則反向開(kāi)單。

3.止損，如果在持有時(shí)候，下跌到反向POSPrice +/- Multiple * TickValue 價(jià)格時(shí)候，平倉(cāng)。Multiple 隨著時(shí)間增加逐漸減少。

示例代碼如下，最后很遺憾，回測(cè)效果非常不好。

from __future__ import division
from vnpy.trader.vtConstant import EMPTY_STRING, EMPTY_FLOAT, OFFSET_OPEN,OFFSET_CLOSE
from vnpy.trader.app.ctaStrategy.ctaTemplate import (CtaTemplate,
                                                     BarGenerator,
                                                     ArrayManager)
import numpy as np
from datetime import datetime, time
########################################################################
class PercentileStrategy(CtaTemplate):
    """MACD策略Demo"""
    className = 'PercentileStrategy'
    author = u'BillyZhang'
    fixedSize = 1
    # 策略參數(shù)
    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
    # 時(shí)間
    initDays = 0
    DAY_START = time(9, 10)  # 日盤(pán)啟動(dòng)和停止時(shí)間
    DAY_END = time(14, 55)
    NIGHT_START = time(21, 10)  # 夜盤(pán)啟動(dòng)和停止時(shí)間
    NIGHT_END = time(10, 55)
    # 參數(shù)列表，保存了參數(shù)的名稱
    paramList = ['name',
                 'className',
                 'author',
                 'vtSymbol',
                 'initDays',
                 'fixedSize',
                 'calWindow',
                 'percentile',
                 'tickValueLimit',
                 'Multiple'
                 ]
    # 變量列表，保存了變量的名稱
    varList = ['inited',
               'trading',
               'pos',
               'longStop',
               'shortStop',
               'posPrice',
               'lowerLimit',
               'p',
               'tickValue',
               'tradeSign',
               'tickValueHigh',
               'tickValueLow'
                ]
    # 同步列表，保存了需要保存到數(shù)據(jù)庫(kù)的變量名稱
    syncList = ['pos',
                'posPrice',
                'longStop',
                'shortStop'
                ]
    # ----------------------------------------------------------------------
    def __init__(self, ctaEngine, setting):
        """Constructor"""
        super(PercentileStrategy, self).__init__(ctaEngine, setting)
        self.am = ArrayManager(size = self.calWindow)
        # 注意策略類中的可變對(duì)象屬性（通常是list和dict等），在策略初始化時(shí)需要重新創(chuàng)建，
        # 否則會(huì)出現(xiàn)多個(gè)策略實(shí)例之間數(shù)據(jù)共享的情況，有可能導(dǎo)致潛在的策略邏輯錯(cuò)誤風(fēng)險(xiǎn)，
        # 策略類中的這些可變對(duì)象屬性可以選擇不寫(xiě)，全都放在__init__下面，寫(xiě)主要是為了閱讀
        # 策略時(shí)方便（更多是個(gè)編程習(xí)慣的選擇）
    # ----------------------------------------------------------------------
    def onInit(self):
        """初始化策略（必須由用戶繼承實(shí)現(xiàn)）"""
        self.writeCtaLog(u'%s策略初始化' % self.name)
        initData = self.loadBar(self.initDays)
        for bar in initData:
            self.onBar(bar)
        self.putEvent()
    # ----------------------------------------------------------------------
    def onStart(self):
        """啟動(dòng)策略（必須由用戶繼承實(shí)現(xiàn)）"""
        if self.pos == 0:
            self.writeCtaLog(u'%s策略啟動(dòng)' % self.name)
        # 當(dāng)前無(wú)倉(cāng)位，發(fā)送開(kāi)倉(cāng)委托
        # 持有多頭倉(cāng)位
        self.putEvent()
    # ----------------------------------------------------------------------
    def onStop(self):
        """停止策略（必須由用戶繼承實(shí)現(xiàn)）"""
        self.writeCtaLog(u'%s策略停止' % self.name)
        self.putEvent()
    # ----------------------------------------------------------------------
    def onTick(self, tick):
        """收到行情T(mén)ICK推送（必須由用戶繼承實(shí)現(xiàn)）"""
        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推送（必須由用戶繼承實(shí)現(xiàn)）"""
        #如果是當(dāng)然最后5分鐘，略過(guò)
        am = self.am
        am.updateBar(bar)
        if not am.inited:
            return
        # currentTime = datetime.now().time()
        currentTime = time(9,20)
        #計(jì)算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
        # 平當(dāng)日倉(cāng)位, 如果當(dāng)前時(shí)間是結(jié)束前日盤(pán)15點(diǎn)28分鐘,或者夜盤(pán)10點(diǎn)58分鐘，如果有持倉(cāng)，平倉(cāng)。
        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):
        """收到委托變化推送（必須由用戶繼承實(shí)現(xiàn)）"""
        # 對(duì)于無(wú)需做細(xì)粒度委托控制的策略，可以忽略onOrder
        pass
    # ----------------------------------------------------------------------
    def onTrade(self, trade):
        # 發(fā)出狀態(tài)更新事件
        """收到成交推送（必須由用戶繼承實(shí)現(xiàn)）"""
        # 對(duì)于無(wú)需做細(xì)粒度委托控制的策略，可以忽略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
            # 同步數(shù)據(jù)到數(shù)據(jù)庫(kù)
        self.saveSyncData()
    # ----------------------------------------------------------------------
    def onStopOrder(self, so):
        """停止單推送"""
        pass

關(guān)于基于統(tǒng)計(jì)的交易策略簡(jiǎn)易實(shí)現(xiàn)VNPY的示例分析就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，可以學(xué)到更多知識(shí)。如果覺(jué)得文章不錯(cuò)，可以把它分享出去讓更多的人看到。