In [1]:
#!/Tsan/bin/python
# -*- coding: utf-8 -*-
In [2]:
#"""此note book为股票回测框架的示例"""
In [3]:
# Libraries to use
from __future__ import division
import numpy as np
import pandas as pd
import statsmodels.api as sm
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime
In [4]:
# Import My own library for factor testing
from SingleFactorTest import factorFilterFunctions as ff
from BackTestingEngine import backTestingEngine as bte
#from config import *
In [5]:
path = ff.data_path
In [6]:
startTime = datetime.strptime('20120504', '%Y%m%d')
endTime = datetime.strptime('20170928', '%Y%m%d')
initstartTime = datetime.strptime('20120304', '%Y%m%d')
initendTime = datetime.strptime('20170928', '%Y%m%d')
In [7]:
filenameAdjOpen = 'OwnfactorAdjustedOpen.h5'
filenameAdjClose = 'OwnfactorAdjustedClose.h5'
filenameVolume = 'LZ_CN_STKA_QUOTE_TVOLUME.h5'
In [8]:
# 初始化
test = bte.StkBacktesting(path)
In [9]:
test.setInitCap(5000000)
test.availableCashNow
Out[9]:
In [10]:
# 添加数据
test.addData('adjOpen',filenameAdjOpen)
test.addData('adjClose',filenameAdjClose)
test.addData('volume',filenameVolume)
In [11]:
# 截取回测区间和初始化区间
test.setBackTestingPeriod(startTime,endTime)
test.setInitialPeriod(initstartTime,initendTime)
In [ ]:
In [12]:
# 双均线筛选买卖股
ma5 = test.dataDict['adjClose'].rolling(window=5, min_periods=5).mean()
ma20 = test.dataDict['adjClose'].rolling(window=20, min_periods=20).mean()
buyStk = ((ma5 > ma20) & ((ma5 < ma20).shift(1) )).shift(1).loc[test.backTestingDateList.tolist()]
sellStk = ((ma5 < ma20) & ((ma5 > ma20).shift(1) )).shift(1).loc[test.backTestingDateList]
In [13]:
endofMonth = sorted(list(set(buyStk .iloc[buyStk .resample('M').size().cumsum().sub(1)].index)))
In [ ]:
In [14]:
# 按天循环
for date in test.backTestingDateList:
if date in endofMonth:
#todayindex = test.backTestingDateList.tolist().index(date)
#yesterday = test.backTestingDateList[todayindex -1]
#print yesterday
# 切片确定当天的买卖股票
sellslice = sellStk.loc[date]
sellList = sellslice.loc[sellslice==True].index.tolist()
#print 'before',len(sellList)
buyslice = buyStk.loc[date]
buyList = buyslice.loc[buyslice==True].index.tolist()
# 买卖单
buyOrders = [test.makeOrder(date,stkID,100,20,1) for stkID in buyList]
# 区分有无持仓时，只有有持仓时才可以卖，并且顺序永远是先卖后买
if test.getCurrentPosition():
print date, 'Current position Num',len(test.getCurrentPosition())
sellList = list(set(sellList) & set(test.getCurrentPosition().keys()))
sellOrders = [test.makeOrder(date,stkID,test.getCurrentPosition()[stkID].volume,20,-1) for stkID in sellList]
totalOrders = sellOrders + buyOrders
#print 'buyNumber',len(buyOrders)
#print 'sellNumber', len(sellList)
else:
#print 'Current position is 0'
totalOrders = buyOrders
#print len(totalOrders)
if totalOrders:
test.allOrdersDict[date] = totalOrders # 保存order到字典里
test.crossOrder(date) # 撮合
test.updateHoldingInfo(date)
else: # 若当日无订单则直接更新持仓
test.updateHoldingInfo(date) # 不管当日是否交易此方法必须每次调用
#print date,'Wrong one', test.getAllPosition()[date].__len__()
#print date,'compare',len(test.getCurrentPosition())
#try:
# print yesterday,'yester',test.getAllPosition()[yesterday].__len__()
#except:
# pass
else:
# 每日更新市值
test.updateHoldingInfo(date)
In [15]:
#for i,j in test._allCurrentPositionDict.iteritems():
# print i,j.values
In [ ]:
In [16]:
positiondf = pd.DataFrame.from_dict(test.getAllPosition(), orient='index').reindex(test.backTestingDateList).fillna('ffill')
In [17]:
def converToVolume(holdingclass):
try:
volume = holdingclass.volume
except:
volume = np.NaN
return volume
In [18]:
positiondf.applymap(converToVolume).fillna(0).tail(10)
Out[18]:
In [19]:
pd.DataFrame.from_dict(test.tradingPnlDict, orient='index').tail()
Out[19]:
In [ ]:
In [ ]:
In [20]:
test.showBackTestingResult()
In [24]:
a =pd.Series(test.totalTradingAmountDict).reindex(test.backTestingDateList).fillna(0)
In [25]:
b =pd.Series(test.totalMarketValueDict).reindex(test.backTestingDateList).fillna(method ='ffill')
In [ ]: