利用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) 当前标题：利用VNPY回测引擎分析实盘交易，并用excel和pdf输出分析结果-创新互联
浏览路径：http://shouzuofang.com/article/dichic.html