利用VNPY回測引擎分析實盤交易，并用excel和pdf輸出分析結果

算是之前一篇文章的后續

http://blog.itpub.net/22259926/viewspace-2564798/

之前寫了一篇文章，在實盤交易時候，用數據庫記錄交易信息，其實就是把交易信息類型VtTradeData的實例保存的到數據庫中。

之前盤后分析主要是數據庫把collection導出到csv文檔，然后下載本地用excel分析，拼湊出sharpe指標，趨勢圖一些；費時費力還不準確；只能說近似正確。

一直想找個什么分析工具，后來一想，其實VNPY回測引擎就很不錯，簡單易用；就是搗鼓搗鼓，利用VNPY回測引擎分析實盤交易，并用excel和pdf輸出分析結果。

這里所有代碼還是針對VNPY 1.92的，因為現在我的實盤還是這個版本，如果VNPY2的版本，其實應該改改也可以用。

完整代碼如下，只要保存到一個本地路徑，執行就可以。

整個流程是：

1. 使用方法load_tradedata，傳入交易記錄數據庫信息，和collection名稱，讀取交易信息，保存為OrderDict格式

2. 使用initEngine4Deal，初始化一個回測引擎，傳入品種交易參數，比如手續費一類，進行更完善計算，滑點可以為0，因為是真實成交數據； 這里還要提供歷史品種行情數據，為按日結算提供參考；同時使用真實歷史交易信息Dict，替換本來是回測生成的交易信息tradeDict，這個也是核心步驟。

3.然后用回測引擎的按比分析方法，和按日分析方法分析；把分析結果用Dict保存；轉為excel輸出；圖像plt用pdf輸出。這里有個中文亂碼問題，比較討厭，可以搜索看看解決方法。

完成后截圖下如，

Excel做了行列轉換，方便查看

大部分注釋都解釋了。有幾點要介紹下，代碼如下:

傳入的是用list包Dict的格式，大部分都注釋解釋了。

# encoding: UTF-8
"""
從DEAL數據庫中讀取交易記錄，利用回測分析引擎分析，并用excel和pdf輸出分析結果
"""
from __future__ import division
from __future__ import print_function
from vnpy.trader.app.ctaStrategy.ctaBacktesting import BacktestingEngine, MINUTE_DB_NAME
from vnpy.trader.vtObject import VtTickData, VtBarData
from vnpy.trader.vtGateway import VtTradeData
from vnpy.trader.vtGlobal import globalSetting
import pymongo
from collections import OrderedDict
from datetime import datetime,date
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from matplotlib.backends.backend_pdf import PdfPages
def load_tradedata(dbName, collectionName, startDate):
   """
   讀取交易歷史記錄，返回用OrderedDict保存的VtTradeData
   :param dbName:
   :param collectionName:
   :return: {ID:VtTradeData}
   """
   dbClient = pymongo.MongoClient(globalSetting['mongoHost'], globalSetting['mongoPort'])
   collection = dbClient[dbName][collectionName]
   # 載入初始化需要用的數據
   flt = {}
   initCursor = collection.find(flt).sort('tradeTime')
   tradeDict = OrderedDict() # 交易記錄字典
   tradeDictID = 0 # 交易編號
   for d in initCursor:
      trade = VtTradeData()
      trade.__dict__ = d
      trade.tradeTime = datetime.strptime(trade.tradeTime, '%Y-%m-%d %H:%M:%S')
      trade.dt = trade.tradeTime
      tradeDictID += 1  # 成交編號自增1
      tradeDict[str(tradeDictID)] = trade
   return tradeDict
def initEngine4Deal(dealCollection,historyCollection,startDate,tradeRate,tradeSize,tradePriceTick,tradeSlipe = 0,
                       tradeCapital = 50000):
   """
   傳入參數，返回回測引擎，保護實際交易數據，用來分析交易情況，。
   :param dealCollection:
   :param historyCollection:
   :param startDate:
   :param tradeSlipe:
   :param tradeRate:
   :param tradeSize:
   :param tradePriceTick:
   :param tradeCapital:
   :param exportPDF:
   :return:
   """
   # 創建回測引擎對象
   engine = BacktestingEngine()
   # 設置回測使用的數據
   engine.setBacktestingMode(engine.BAR_MODE)  # 設置引擎的回測模式為K線
   engine.setDatabase(MINUTE_DB_NAME, historyCollection)  # 設置使用的歷史數據庫
   engine.setStartDate(startDate)  # 設置回測用的數據起始日期
   # 配置回測引擎參數
   engine.setSlippage(tradeSlipe)  # 設置滑點為1跳
   engine.setRate(tradeRate)  # 設置手續費萬1
   engine.setSize(tradeSize)  # 設置合約大小
   engine.setPriceTick(tradePriceTick)  # 設置最小價格變動
   engine.setCapital(tradeCapital)  # 設置回測本金
   engine.loadHistoryData()
   bar = None
   for d in engine.dbCursor:
      data = VtBarData()
      data.__dict__ = d
      engine.updateDailyClose(data.datetime, data.close)
      bar = data
   # # 構建每日收盤價
   # for bar in engine.BackTestData:
   #  engine.updateDailyClose(bar.datetime, bar.close)
   # 是回測數據后一個bar，用于未結束持倉計算收益
   engine.bar = bar
   #  讀取歷史交易數據，塞入回測引擎
   tradeDict = load_tradedata("VnTrader_DEAL_Db",dealCollection,startDate)
   engine.tradeDict = tradeDict
   # 顯示成交記錄
   # for i in range(len(tradeDict)):
   #  d = engine.tradeDict[str(i + 1)].__dict__
   #  print('TradeID: %s, Time: %s, Direction: %s, Price: %s, Volume: %s' % (
   #  d['tradeID'], d['dt'], d['direction'], d['price'], d['volume']))
   return engine
def tradeResultAnalysis(engine,dealCollection):
   """
   傳入回測引擎，和deal名稱，按deal進行回測分析，返回回測分析結果DICT格式，和plt圖標用于pdf輸出
   :param engine:
   :param dealCollection:
   :return:
   """
   TradeResult = {}
   d = engine.calculateBacktestingResult()
   TradeResult[u"第一筆交易/FirstDeal"] =  d['timeList'][0]
   TradeResult[u"最后一筆交易/LastDeal"] = d['timeList'][-1]
   TradeResult[u"總交易次數/DealNumber"] = (d['totalResult'])
   TradeResult[u"總盈虧/DealPnL"] = (d['capital'])
   TradeResult[u"最大回撤/MaxDrawdown"] = (min(d['drawdownList']))
   TradeResult[u"平均每筆盈利/AveragePnL"] = (d['capital'] / d['totalResult'])
   TradeResult[u"平均每筆滑點/AverageSlip"] = (d['totalSlippage'] / d['totalResult'])
   TradeResult[u"平均每筆傭金/AverageCommisson"] = (d['totalCommission'] / d['totalResult'])
   TradeResult[u"勝率/WinRate%"] =(d['winningRate'])
   TradeResult[u"盈利交易平均值/AverageProfit"] = (d['averageWinning'])
   TradeResult[u"虧損交易平均值/AverageLoss"] = (d['averageLosing'])
   TradeResult[u"盈虧比/ProfitLossRatio"] = (d['profitLossRatio'])
   plt = tradeResult2Plt(d,TradeResult,dealCollection)
   return TradeResult, plt
def dailyResultAnalysis(engine, dealCollection):
   """
   傳入回測引擎，和deal名稱，按每日進行回測分析，返回回測分析結果DICT格式，和plt圖標用于pdf輸出
   :param engine:
   :param dealCollection:
   :return:
   """
   engine.calculateDailyResult()
   dx, dailyResult = engine.calculateDailyStatistics()
   plt = dailyResult2Plt(dx, dailyResult, dealCollection)
   return dailyResult,plt
def tradeResult2Plt(d,TradeResult,dealCollection):
   # 輸出按交易統計 plt 圖標
   # 繪圖
   fig = plt.figure(figsize=(10, 18))
   pText = plt.subplot(5, 1, 1)
   pText.set_title("TradeResultAnalysis " + dealCollection)
   pText.text(0,0.1,str(TradeResult),fontsize=12, bbox={'facecolor':'white'},wrap=True)
   pCapital = plt.subplot(5, 1, 2)
   pCapital.set_ylabel("capital")
   pCapital.plot(d['capitalList'], color='r', lw=0.8)
   pDD = plt.subplot(5, 1, 3)
   pDD.set_ylabel("DD")
   pDD.bar(range(len(d['drawdownList'])), d['drawdownList'], color='g')
   pPnl = plt.subplot(5, 1, 4)
   pPnl.set_ylabel("pnl")
   pPnl.hist(d['pnlList'], bins=50, color='c')
   pPos = plt.subplot(5, 1, 5)
   pPos.set_ylabel("Position")
   if d['posList'][-1] == 0:
      del d['posList'][-1]
   tradeTimeIndex = [item.strftime("%m/%d %H:%M:%S") for item in d['tradeTimeList']]
   xindex = np.arange(0, len(tradeTimeIndex), np.int(len(tradeTimeIndex) / 10))
   tradeTimeIndex = list(map(lambda i: tradeTimeIndex[i], xindex))
   pPos.plot(d['posList'], color='k', drawstyle='steps-pre')
   pPos.set_ylim(-1.2, 1.2)
   plt.sca(pPos)
   # plt.rcParams['font.sans-serif'] = ['SimSun']  # 用來正常顯示中文標簽
   plt.rcParams['axes.unicode_minus'] = False  # 用來正常顯示負號
   plt.tight_layout()
   plt.xticks(xindex, tradeTimeIndex, rotation=30)  # 旋轉15
   return plt
def dailyResult2Plt(d, dailyResult,dealCollection):
   # 繪圖
   fig = plt.figure(figsize=(10, 18))
   pText = plt.subplot(5, 1, 1)
   pText.set_title("DailyResultAnalysis " + dealCollection)
   pText.text(0,0.1,str(dailyResult),fontsize=12 , bbox={'facecolor':'white'},wrap=True)
   pBalance = plt.subplot(5, 1, 2)
   pBalance.set_title('Balance')
   plt.plot(d['date'], d['balance'])
   pDrawdown = plt.subplot(5, 1, 3)
   pDrawdown.set_title('Drawdown')
   pDrawdown.fill_between(range(len(d['drawdown'])), d['drawdown'])
   pPnl = plt.subplot(5, 1, 4)
   pPnl.set_title('Daily Pnl')
   plt.bar(range(len(d['drawdown'])), d['netPnl'])
   pKDE = plt.subplot(5, 1, 5)
   pKDE.set_title('Daily Pnl Distribution')
   plt.hist(d['netPnl'], bins=50)
   return plt
def toExcel(resultlist, path="C:\data\datframe.xlsx"):
   # 按照輸入統計數據隊列和路徑，輸出excel，這里不提供新增模式，如果想，可以改
   # dft.to_csv(path,index=False,header=True, mode = 'a')
   summayKey = resultlist[0].keys()
   df = pd.DataFrame(columns=summayKey)
   for result in resultlist:
      new = pd.DataFrame(result, index=["0"])
      df = df.append(new, ignore_index=True,sort=True)
   dft = pd.DataFrame(df.values.T, index=df.columns, columns=df["DealCollection"])
   dft.to_excel(path, index=True, header=True)
   print("回測統計結果輸出到" + path)
if __name__ == "__main__":
   DealCollectionList = [
      {
         "historyCollection": "rb9999",
         "dealCollection": "ATRStrategy RB",
         "StartDate": "20200301",
         "Size": 10,
         "Rate": 0.0001,
         "PriceTick":1
      },
      {
         "historyCollection": "fu8888",
         "dealCollection": "CCIStrategy fu",
         "StartDate": "20200301",
         "Size": 10,
         "Rate": 0.0001,
         "PriceTick": 1
      },
   ]
   resultList = []
   pdf = PdfPages("DealResultAnalysisPDF" + str(date.today())+ "v1.pdf")
   for dealCollction in DealCollectionList:
      ResultDict = {}
      # 加入交易集名稱
      dealCollctionName = dealCollction["dealCollection"]
      ResultDict["DealCollection"] = dealCollctionName
      # 初始回測引擎
      engine = initEngine4Deal(
         dealCollection = dealCollction["dealCollection"],
         historyCollection = dealCollction["historyCollection"],
         startDate = dealCollction["StartDate"],
         tradeSize = dealCollction["Size"],
         tradeRate =  dealCollction["Rate"],
         tradePriceTick = dealCollction["PriceTick"]
      )
      # 按deal進行分析，傳入resultDict，和pdf
      tradeResult,plt=tradeResultAnalysis(engine,dealCollctionName)
      ResultDict.update(tradeResult)
      pdf.savefig()
      plt.close()
      # 按每日進行分析，傳入resultDict，和pdf
      dailyResult,plt= dailyResultAnalysis(engine,dealCollctionName)
      ResultDict.update(dailyResult)
      pdf.savefig()
      plt.close()
      resultList.append(ResultDict)
   pdf.close()
   path = "DealResultAnalysisExcel" + str(date.today()) + "v2.xls"
   toExcel(resultList, path)