［策略］基于勝率的趨勢交易策略 · Python 量化交易教程

# ［策略］基于勝率的趨勢交易策略 > 來源：https://uqer.io/community/share/565bbfa4f9f06c6c8a91ae7a ## 策略說明簡單構建了一個基于勝率的趨勢交易策略。認為過去一段時間（N天）內勝率較高、信息比率較高的股票會在緊隨其后的幾天有較好的表現 1）先根據勝率要求篩選出過去N天勝率高的股票作為預選股票（benchmark可以是定義的確定閾值，或者是某個指數相應的收益率），用aprior算法進行快速篩選。第i只股票勝率的計算方式如下: ``` winRate(i) = sum([sign(ret(i,t)-ret(bm,t))==1]/N)|t~(t-N,t) ＊ret(i,t): i股票在第t天的收益率；＊ret（bm,t）: benchmark在第t天的收益率； ``` 2）從篩選出的股票中選擇過去N天信息比率（收益率／波動率）高的部分股票構建備選投資組合； 3）依據被選投資組合做買入操作，使用可用資金的50%～70%； 4）設定股票止損位在收益下跌至0.95，止損時將倉位調整至原倉位的40%～60%； 5）調倉頻率為5天，股票池為滬深300。 ```py import numpy as np from CAL.PyCAL import * ################################################################################ # Back Test Functions ################################################################################ def initialize(account): # 初始化虛擬賬戶狀態 pass ####init the univese of the choosen stock def universeInit(): stockComponent = DataAPI.MktTickRTSnapshotIndexGet(securityID=u"000300.XSHG",field=u"lastPrice,shortNM") stockCount = len(stockComponent) stockTicker = stockComponent['ticker'] stockExchgID = stockComponent['exchangeCD'] stockID = [] for index in range(stockCount): stockID.append(stockTicker[index] + '.' + stockExchgID[index]) return stockID ####deal with the trading signals def handle_data(account): # 每個交易日的買入賣出指令 ####Presettings histLength = 10 stockDataThres = 0 ####Dictionary of the return Rate closePrice = account.get_attribute_history('closePrice',histLength) retRate = {} for index in account.universe: retRate[index] = ((closePrice[index][1:] - closePrice[index][:-1])/closePrice[index][:-1]).tolist() ###ret list of the benchmark calendar = Calendar('China.SSE') startDate = calendar.advanceDate(account.current_date,'-'+str(histLength)+'B').toDateTime() benchmark = DataAPI.MktIdxdGet(ticker = "000300", field = "closeIndex", beginDate = startDate, endDate = account.current_date,pandas = '1') bmClose = benchmark['closeIndex'].tolist() bmRet = [] for index in range(len(bmClose)-1): bmRet.append((bmClose[1:][index]-bmClose[:-1][index])/bmClose[:-1][index]) ####List of transactions transactions = [] for index in range(histLength-1): tmpt = [] for stock in account.universe: if retRate[stock][index] > stockDataThres: # if retRate[stock][index] > bmRet[index]: tmpt.append(stock) transactions.append(tmpt) ####List of hot stocks hotStock = [] hotStockDict,hotStockList = apriori(transactions,0.95) for index in hotStockList: for stock in index: if stock not in hotStock: hotStock.append(stock) ####List of the portfolio retRate = {} fluctRate = {} sharpRate = {} for index in hotStock: retRate[index] = ((closePrice[index][-1] - closePrice[index][0])/closePrice[index][0]) fluctRate[index] = np.std(closePrice[index]) sharpRate[index] = retRate[index]/fluctRate[index] portfolio = [index[0] for index in sorted(sharpRate.items(),key = lambda sharpRate:sharpRate[1])[-len(sharpRate)/2:]] ####Stop loss at -0.05 validSecHist = account.get_attribute_history('closePrice',2) for index in account.valid_secpos: if (validSecHist[index][-1] - validSecHist[index][0])/validSecHist[index][0] < -0.05: order_to(index,0.45*account.valid_secpos[index]) ####Buy portfolio for index in portfolio: amount = 0.65*account.cash/len(hotStock)/account.referencePrice[index] order(index,amount) return ######################################################################################## # Aprior algorithm ######################################################################################## def elementsDet(datasets): if type(datasets) == list: elements = {} for index in datasets: for index1 in index: if elements.has_key(index1) == False: elements[index1] = 1 else: elements[index1] += 1 return elements if type(datasets) == dict: elements = {} for index in datasets: if type(index) == tuple: index = list(index) for index1 in index: if elements.has_key(index1) == False: elements[index1] = 0 else: elements[index] = 0 return elements pass def checkAssociation(subset,objset): for index in subset: if index not in objset: return False return True pass def support(subset,datasets): count = 0 for transaction in datasets: if checkAssociation(subset,transaction) == True: count += 1 return 1.0*count/len(datasets) pass def apriori(datasets,minsup): candidateIterator = [] electIterator = [] length = len(datasets) ##init part #the candidate elements = elementsDet(datasets) candidate = {} for index in elements: candidate[index] = 1.0*elements[index]/length candidateIterator.append(candidate) #the elect elect = {} for index in candidate: if candidate[index] > minsup: elect[index] = candidate[index] electIterator.append(elect) ##the update part itera = 1 while(len(electIterator[-1]) != 0): candidateOld = candidateIterator[-1] electOld = electIterator[-1] elementsOld = elementsDet(electOld) # print elementsOld candidate = {} ##the candidate for index in electOld: for index1 in elementsOld: if type(index) != list and type(index) != tuple: if index1 != index: tmp = [] tmp.append(index) tmp.append(index1) tmp.sort() if candidate.has_key(tuple(tmp)) == False: candidate[tuple(tmp)] = 0 if type(index) == tuple: tmp = list(index) if tmp.count(index1) == False: tmp1 = tmp tmp1.append(index1) tmp1.sort() if candidate.has_key(tuple(tmp1)) == False: candidate[tuple(tmp1)] = 0 candidateIterator.append(candidate) ##the elect elect = {} for index in candidate: candidate[index] = support(index,datasets) for index in candidate: if candidate[index] > minsup: elect[index] = candidate[index] electIterator.append(elect) # print 'iteartion ' + str(itera) + ' is finished!' itera += 1 ##the elected frequency sets dictionary: the value is the key's support electedDict = {} for index in electIterator: for index1 in index: electedDict[index1] = index[index1] ##the elected frequency sets lists electedList = [] for index in electIterator: tmp = [] for index1 in index: if type(index1) == tuple: tmp1 = [] for ele in index1: tmp1.append(ele) tmp.append(tmp1) else: tmp.append([str(index1)]) tmp.sort() for index1 in tmp: electedList.append(index1) return electedDict,electedList ################################################################################ # Back Test Presetting ################################################################################ start = '2011-01-01' # 回測起始時間 end = '2015-11-01' # 回測結束時間 benchmark = 'HS300' # 策略參考標準 universe = set_universe('HS300') # universe = universeInit() # 證券池，支持股票和基金 capital_base = 100000 # 起始資金 freq = 'd' # 策略類型，'d'表示日間策略使用日線回測，'m'表示日內策略使用分鐘線回測 refresh_rate = 5 # 調倉頻率，表示執行handle_data的時間間隔，若freq = 'd'時間間隔的單位為交易日，若freq = 'm'時間間隔為分鐘 ``` ![](https://box.kancloud.cn/2016-07-30_579cbda7e04b5.jpg) ## 策略表現 + 策略能產生一定的alpha； + 策略表現與起點強相關，sharpRatio不穩定； + 策略表現會受到自身參數設定的影響，例如勝率選擇周期、篩選閾值、調倉頻率、建倉頭寸、止損倉位等，需要依據表現對其進行優化； + 策略在2011年4月至12月、2015年6月到11月有相對好的表現，可見其相對較適用于趨勢下跌的市場環境。 ## 問題探討因子選股模型的流程應該是怎樣的？小編認為構建因子選股的模型需要有如下過程： 1. 大類配置：根據宏觀判斷市場，進行市場判斷（根據不同市場選擇不同因子）、資產配置（不同風險性證券的配比選擇??不同熱度的行業配比選擇）和策略選擇（市場中性、單邊做多等）； 2. 選股－alpha端：對選股因子進行有效性分析，包括單因子的預測性、因子間相關性，構建多因子模型使得選股有盡可能高的alpha； 3. 選股－風險端：對alpha端的多因子模型進行風險評估，根據風險因子優化模型，使模型盡可能達到有效邊界； 4. 擇時－買賣時點：對根據因子模型選出的股票進行擇時分析，進一步篩選投資組合中的股票及判斷作何操作；因子選股中比較basic的問題，歡迎社區的小伙伴們發表看法、評論和拍醒~