自己做量化交易軟體(6)通通量化歷史交易回測設計1
阿新 • • 發佈：2018-12-11
下面開始介紹回測功能的設計。我們前面軟體主要用的python函式設計，從現在開始使用python的類設計。 首先設計類class hpQuant(object)。初始化函式中存放一些交易變數。其中一些變數可以外部設定，例如總資金，交易手續費等等，我們按照A股交易規則，買賣有佣金，賣出需要印花稅。 所以關於交易回測的程式碼，我們存放到HP_sys.py檔案中。下面直接給出全部原始碼。讀者可複製到前面下載的通通專案目錄中。
import sys,os import numpy as np import pandas as pd from HP_global import * from HP_set import * class hpQuant(object): def __init__(self): self.order_df=None #下單流水 self.trade_df=None #交易流水 self.security_df=None #持倉清單 self.money2=1000000.00 #總資金 self.money=1000000.00 #資金 self.priceBuy=0.00 #最後一次買入價格 self.priceSell=999999.00 #最後一次賣出價格 self.amount=0.00 #證券數量 self.code="" #證券程式碼 self.stamp_duty=0.001 #印花稅 0.1% self.trading_Commission=0.0005 #交易佣金0.05% self.priceStopLoss=0.00 #止損價 self.position=False #持倉狀態 self.stop_loss_on=True #允許止損 self.stop_loss_num=0 #止損次數 self.stop_loss_max=50 #止損3次,就停止交易 self.stop_loss_range=0.05 #止損幅度 self.trade=True #允許交易 self.Init() def Init(self): self.order_df = pd.DataFrame(columns = ['date', 'time','mode','code','amount','price']) self.order_df =self.order_df.reset_index(level=None, drop=True ,col_level=0, col_fill='') self.trade_df = pd.DataFrame(columns = ['date', 'time','mode','code','amount','price','money']) self.trade_df =self.trade_df.reset_index(level=None, drop=True ,col_level=0, col_fill='') self.security_df = pd.DataFrame(columns = ['code','amount','price','money']) self.security_df =self.security_df.reset_index(level=None, drop=True ,col_level=0, col_fill='') def PrintOrder(self): print(self.order_df) def PrintTrade(self): print(self.trade_df) def PrintSecurity(self): print(self.security_df) def Order(self,date,time,mode,code,amount,price): ln=len(self.order_df) df_new = pd.DataFrame({'date':date,'time':time,'mode':mode,'code':code,'amount':amount,'price':price},index=[ln]) self.order_df=self.order_df.append( df_new,ignore_index=True) ln=len(self.trade_df) if mode==1: #買入 se=amount*price*(1+self.trading_Commission) self.money=self.money-se df_new = pd.DataFrame({'date':date,'time':time,'mode':mode,'code':code,'amount':amount,'price':price,'money':self.money},index=[ln]) self.trade_df=self.trade_df.append( df_new,ignore_index=True) if len(self.security_df[self.security_df.code==code])==0 : df_new = pd.DataFrame({'code':code,'amount':amount,'price':se/amount,'money':se},index=[ln]) self.security_df=self.security_df.append( df_new,ignore_index=True) else: self.security_df.index=self.security_df['code'] self.security_df.loc[code,'amount']=self.security_df.loc[code,'amount']+amount self.security_df=self.security_df.reset_index(level=None, drop=True ,col_level=0, col_fill='') ln=len(self.security_df[self.security_df.code==code]) if mode==2 and ln>0: #賣出 self.security_df.index=self.security_df['code'] am=self.security_df.loc[code,'amount'] if am<amount : amount=am se=amount*price*(1-self.trading_Commission-self.stamp_duty) self.money=self.money+se df_new = pd.DataFrame({'date':date,'time':time,'mode':mode,'code':code,'amount':amount,'price':price,'money':self.money},index=[ln]) self.trade_df=self.trade_df.append( df_new,ignore_index=True) am2=self.security_df.loc[code,'amount']-amount self.security_df.loc[code,'amount']=am2 if am2==0: self.security_df=self.security_df.drop(code, axis = 0) # 在行的維度上刪除行 self.security_df=self.security_df.reset_index(level=None, drop=True ,col_level=0, col_fill='') #####################回測功能######################### def Trade_testing(self,df,tp1,tp2,al=''): self.Init() self.trade=True #允許交易 myMoney=self.money2 if (al.strip()==''): na='property' else: na=al self.text=' ----開始回測-----\n' i = 0 ZB_l = [] while i < len(df): close=df.get_value(i, 'close') if (df.get_value(i, tp1) >0 and self.position==False and self.trade==True) : #買點 self.priceBuy=df.get_value(i, 'close') x=int(myMoney/(self.priceBuy*(1+self.trading_Commission))/100) self.amount=x*100.00 self.Order(df.get_value(i, 'date'),'14:45:01',1,self.code,self.amount,close) myMoney=myMoney-self.amount*self.priceBuy*(1.00+self.trading_Commission) self.position=True self.priceStopLoss=self.priceBuy*(1-self.stop_loss_range) self.text=self.text+'日期:'+df.get_value(i, 'date').strftime('%Y-%m-%d')+' 買入:'+str(round(self.amount,0))+'股, 價格:'+str(round(self.priceBuy,2))+'\n' if (df.get_value(i, tp2) >0 and self.position==True and self.trade==True) : #賣點 self.priceSell=df.get_value(i, 'close') myMoney=myMoney+self.amount*self.priceSell*(1.00-self.trading_Commission-self.stamp_duty) self.position=False self.Order(df.get_value(i, 'date'),'14:45:02',2,self.code,self.amount,df.get_value(i, 'close')) self.text=self.text+'日期:'+df.get_value(i, 'date').strftime('%Y-%m-%d')+' 賣出:'+str(round(self.amount,0))+'股, 價格:'+str(round(self.priceBuy,2))+'獲利:'+str(round((myMoney-self.money2)/self.money2*100,2))+'%\n' self.amount=0.00 if (close<=self.priceStopLoss and self.position==True and self.trade and self.stop_loss_on): #止損 self.priceSell=self.priceStopLoss-0.01 myMoney=myMoney+self.amount*self.priceSell*(1.00-self.trading_Commission-self.stamp_duty) self.position=False self.stop_loss_num=self.stop_loss_num+1 self.Order(df.get_value(i, 'date'),'14:45:02',2,self.code,self.amount,df.get_value(i, 'close')) self.text=self.text+'日期:'+df.get_value(i, 'date').strftime('%Y-%m-%d')+' 止損:'+str(round(self.amount,0))+'股, 價格:'+str(round(self.priceBuy,2))+'獲利:'+str(round((myMoney-self.money2)/self.money2*100,2))+'%\n' self.amount=0.00 if (self.stop_loss_num>=self.stop_loss_max): self.trade=False y= (myMoney+self.amount*close-self.money2)/self.money2 *100 ZB_l.append(y) i = i + 1 ZB_s = pd.Series(ZB_l) ZB = pd.Series(ZB_s, name = na) df = df.join(ZB) y= (myMoney+self.amount*close-self.money2)/self.money2 *100 self.text=self.text+'總投入:1000000.00,最終獲利幅度'+str(round(y,0))+'%\n' return df
使用方法：
import HP_lib as mylib from HP_sys import * ##回測 tt=hpQuant() ##初始化類 #下面是使用者可設定資訊。 # self.money2=1000000.00 #總資金 # self.code="" #證券程式碼 # self.stamp_duty=0.001 #印花稅 0.1% # self.trading_Commission=0.0005 #交易佣金0.05% # self.stop_loss_on=True #允許止損 # self.stop_loss_max=50 #止損3次,就停止交易 # self.stop_loss_range=0.05 #止損幅度 tt.code=code #證券程式碼，必須輸入 tt.stop_loss_on=False #關閉自動止損 df3=tt.Trade_testing(df2,'B1','S1','HL') #開始回測 print('\n列印交易過程') tt.PrintTrade() #列印交易過程 print('\n列印持倉資訊') tt.PrintSecurity() #列印持倉資訊 print('\n 列印內部交易記錄資訊') print(tt.text) #列印交易資訊
具體使用，我在下一篇介紹一個簡單的回測例子。