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 随着时间增加逐渐减少。

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

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 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 = uBillyZhang 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) # 注意策略类中的云南idc服务商可变对象属性（通常是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