Pyalgotrade-----Aberration

愛天空旅游資訊2020-03-31 01:45:49

# -*- coding: utf-8 -*-
#made in china by me named laoK
#一下是要用到的庫,先導入

from pyalgotrade import strategy #策略回測
from pyalgotrade.bar import Frequency #數據加載相關
from pyalgotrade.stratanalyzer import sharpe #夏普
from pyalgotrade.barfeed.csvfeed import GenericBarFeed
from pyalgotrade import plotter #畫圖
from pyalgotrade.technical import ma #均線
from pyalgotrade.technical import stats #從這標準差
#把常量些搞定,以及數據加載上
def main(plot):
instrument = "rb1Hour" #測試數據是螺紋鋼1小時圖,數據來源TB
? ?Length=35 #計算周期
? ?StdDevUp=2 #比例
? ?feed = GenericBarFeed(Frequency.HOUR, None, None) #導入的
格式有一定要求,可加公眾好回復‘數據轉換’獲取專用數據轉換的代碼
feed.addBarsFromCSV(instrument, 'D:/huobidataKlines/rb8881.csv')#把
數據加載入feed
strat = MarketopeningTrade(feed, instrument,Length,StdDevUp)#策略名及參數
sharpeRatioAnalyzer = sharpe.SharpeRatio()#計算夏普
strat.attachAnalyzer(sharpeRatioAnalyzer)

if plot:
plt = plotter.StrategyPlotter(strat, True, True, True)#作圖


? ?strat.run()#開始運行
print("Sharpe ratio: %.2f" % sharpeRatioAnalyzer.getSharpeRatio(0.05))
print("Final portfolio value: $%.2f" % strat.getResult())#getBroker().getEquity())
? ?#strategy下的2個py文件重點看下,交易的情況信息都可以從這獲得

? ?if plot:
plt.plot()

class MarketopeningTrade(strategy.BacktestingStrategy):
def __init__(self, feed, instrument,Length,StdDevUp ):
super(MarketopeningTrade, self).__init__(feed)
self.__instrument = instrument
self.__priceDS = feed[instrument].getCloseDataSeries()#獲取收盤價序列
self.__aveSMA = ma.SMA(self.__priceDS, Length)#均線
self.__StdValue = stats.StdDev(self.__priceDS, Length)#標準差
self.__money = self.getBroker().getCash() * 0.1#每次只用1層資金
? ? ? ?self.StdDevUp=StdDevUp
self.UpperBand = []#先建立空list,之后再把上下軌和收盤價都加入到list
self.LowerBand = []#使用的時候比較方便,pyalgotrade不支持pandas,有時候需要
self.closelist = []#過濾信號,就要自己想法的
self.__longPos = None
? ? ? ?self.__shortPos = None

? ?def getmoney(self):
return self.__money

def getStdValue(self):
return self.__StdValue

def getaveSMA(self):
return self.__aveSMA

def onEnterCanceled(self, position):
if self.__longPos == position:
self.__longPos = None
? ? ? ?elif self.__shortPos == position:
self.__shortPos = None
? ? ? ?else:
assert(False)

def onExitOk(self, position):
if self.__longPos == position:
self.__longPos = None
? ? ? ?elif self.__shortPos == position:
self.__shortPos = None
? ? ? ?else:
assert(False)


def onExitCanceled(self, position):
# If the exit was canceled, re-submit it.
? ? ? ?position.exitMarket()

 ? ?def onBars(self, bars):#循環系統,頭寸管理都靠他了
# Wait for enough bars to be available to calculate SMA and RSI.
? ? ? ?if self.__aveSMA[-1] is None :
return

? ? ? ?bar = bars[self.__instrument]
if self.__aveSMA[-1] is not None:
self.UpperBand.append(self.__aveSMA[-1] + self.StdDevUp * self.__StdValue[-1])
self.LowerBand.append(self.__aveSMA[-1] - self.StdDevUp * self.__StdValue[-1])
self.closelist.append(bar.getPrice())

if self.__shortPos is ?None and self.enterShortSignal(bar):

if self.__longPos is not None:
self.__longPos.exitMarket()
print(str(bars[self.__instrument].getDateTime()) + " " + (str(bar.getClose())) + " " + "long cut")
shares = 1 ?# self.__money / (bars[self.__instrument].getPrice() )
? ? ? ? ? ?self.__shortPos = self.enterShort(self.__instrument, shares, True)

print(str(bars[self.__instrument].getDateTime()) + " " + (str(bar.getClose())) + " " + "buy Short")
elif self.__longPos is ?None and self.enterLongSignal(bar):

if self.__shortPos is not None:
self.__shortPos.exitMarket()
print(str(bars[self.__instrument].getDateTime()) + " " + (str(bar.getClose())) + " " + "short cut")
shares = 1 ?# self.__money / (bars[self.__instrument].getPrice() )
? ? ? ? ? ?self.__longPos = self.enterLong(self.__instrument, shares, True)

print(str(bars[self.__instrument].getDateTime()) + " " + (str(bar.getClose())) + " " + "buy Long")

elif self.__longPos is not None and self.exitLongSignal(bar):
self.__longPos.exitMarket()
print(str(bars[self.__instrument].getDateTime()) + " " + (str(bar.getClose())) + " " + "long cut")
elif self.__shortPos is not None and self.exitShortSignal(bar):
self.__shortPos.exitMarket()
print(str(bars[self.__instrument].getDateTime()) + " " + (str(bar.getClose())) + " " + "short cut")
     
def enterLongSignal(self, bar):#多單進場信號
if self.__aveSMA[-2] is not None:
if self.closelist[-1] >self.UpperBand[-1] and self.closelist[-2] <self.UpperBand[-2]:
return True
? ? ? ?else:
return False

? ?def exitLongSignal(self, bar):#多單出場信號
if not self.__longPos.exitActive():
if bar.getPrice()<self.__aveSMA[-1]:
return True
? ? ? ?else:
return False

? ?def enterShortSignal(self, bar):#空單進場信號
if self.__aveSMA[-2] is not None:
if self.closelist[-1] <self.LowerBand[-1] and self.closelist[-2] >self.LowerBand[-2]:
#最尷尬的地方就是這了,用pyalgotrade自帶的cross寫不來,如果哪位高手會求指點
return True
? ? ? ?else:
return False

? ?def exitShortSignal(self, bar):#空單出場信號
if not self.__shortPos.exitActive():
if bar.getPrice() > self.__aveSMA[-1]:
#同上
return True
? ? ? ?else:
return False


if __name__ == "__main__":#開整
main(True)

結果是在沒有杠桿,1跳就只有1元的變動(期貨因為是1手10噸,所以1跳10元),且沒有算滑點和手續費的情況下測試的。


同樣策略在TB開拓者12元來回手續費,加雙邊各1跳的情況下的結果。固定一手交易

僅做為自己的學習筆記以及交流學習。是否用于實盤請自行把握!

初學python,不足之處請各位高手多包涵多指正!





Copyright ? 南京白銀投資培訓研習社@2017
快乐十分摇奖机 配资网上上盈官网 贵州11选5平台 快3彩票 浙江十一选五专家推荐预测 福建31选7app 山东十一选五遗漏360 一分彩开奖结果 安徽11选五开奖走势图 一定牛 河北十一选五遗漏排行 牛人配资 p62中四个号多少钱 股指配资先问尚牛在线 极速快三人工计划免费 贵州十一选五中奖规则 新疆11选五走势图表 浙江体彩11选5任一中奖规则