2792 lines
113 KiB
Python
2792 lines
113 KiB
Python
# encoding: UTF-8
|
||
|
||
'''
|
||
本文件中包含的是CTA模块的回测引擎，回测引擎的API和CTA引擎一致，
|
||
可以使用和实盘相同的代码进行回测。
|
||
'''
|
||
from __future__ import division
|
||
|
||
from datetime import datetime, timedelta
|
||
from collections import OrderedDict
|
||
from itertools import product
|
||
import multiprocessing
|
||
import pymongo
|
||
|
||
from ctaBase import *
|
||
from vtConstant import *
|
||
from vtGateway import VtOrderData, VtTradeData
|
||
from vtFunction import loadMongoSetting
|
||
|
||
from eventEngine import *
|
||
|
||
import MySQLdb
|
||
import json
|
||
import os
|
||
import cPickle
|
||
import csv
|
||
import logging
|
||
import copy
|
||
import pandas as pd
|
||
import re
|
||
|
||
########################################################################
|
||
class BacktestingEngine(object):
|
||
"""
|
||
CTA回测引擎
|
||
函数接口和策略引擎保持一样，
|
||
从而实现同一套代码从回测到实盘。
|
||
# modified by IncenseLee：
|
||
1.增加Mysql数据库的支持；
|
||
2.修改装载数据为批量式后加载模式。
|
||
3.增加csv 读取bar的回测模式
|
||
4.增加csv 读取tick合并价差的回测模式
|
||
5.增加EventEngine，并对newBar增加发送OnBar事件，供外部的回测主体显示Bar线。
|
||
"""
|
||
|
||
TICK_MODE = 'tick' # 数据模式，逐Tick回测
|
||
BAR_MODE = 'bar' # 数据模式，逐Bar回测
|
||
|
||
REALTIME_MODE ='RealTime' # 逐笔交易计算资金，供策略获取资金容量，计算开仓数量
|
||
FINAL_MODE = 'Final' # 最后才统计交易，不适合按照百分比等开仓数量计算
|
||
|
||
#----------------------------------------------------------------------
|
||
def __init__(self, eventEngine = None):
|
||
"""Constructor"""
|
||
|
||
self.eventEngine = eventEngine
|
||
|
||
# 本地停止单编号计数
|
||
self.stopOrderCount = 0
|
||
# stopOrderID = STOPORDERPREFIX + str(stopOrderCount)
|
||
|
||
# 本地停止单字典
|
||
# key为stopOrderID，value为stopOrder对象
|
||
self.stopOrderDict = {} # 停止单撤销后不会从本字典中删除
|
||
self.workingStopOrderDict = {} # 停止单撤销后会从本字典中删除
|
||
|
||
# 引擎类型为回测
|
||
self.engineType = ENGINETYPE_BACKTESTING
|
||
|
||
# 回测相关
|
||
self.strategy = None # 回测策略
|
||
self.mode = self.BAR_MODE # 回测模式，默认为K线
|
||
|
||
self.startDate = ''
|
||
self.initDays = 0
|
||
self.endDate = ''
|
||
|
||
self.slippage = 0 # 回测时假设的滑点
|
||
self.rate = 0 # 回测时假设的佣金比例（适用于百分比佣金）
|
||
self.size = 1 # 合约大小，默认为1
|
||
self.priceTick = 0 # 价格最小变动
|
||
|
||
self.dbClient = None # 数据库客户端
|
||
self.dbCursor = None # 数据库指针
|
||
|
||
self.historyData = [] # 历史数据的列表，回测用
|
||
self.initData = [] # 初始化用的数据
|
||
self.backtestingData = [] # 回测用的数据
|
||
|
||
self.dbName = '' # 回测数据库名
|
||
self.symbol = '' # 回测集合名
|
||
self.margin_rate = 0.11 # 回测合约的保证金比率
|
||
|
||
self.dataStartDate = None # 回测数据开始日期，datetime对象
|
||
self.dataEndDate = None # 回测数据结束日期，datetime对象
|
||
self.strategyStartDate = None # 策略启动日期（即前面的数据用于初始化），datetime对象
|
||
|
||
self.limitOrderDict = OrderedDict() # 限价单字典
|
||
self.workingLimitOrderDict = OrderedDict() # 活动限价单字典，用于进行撮合用
|
||
self.limitOrderCount = 0 # 限价单编号
|
||
|
||
# 持仓缓存字典
|
||
# key为vtSymbol，value为PositionBuffer对象
|
||
self.posBufferDict = {}
|
||
|
||
self.tradeCount = 0 # 成交编号
|
||
self.tradeDict = OrderedDict() # 成交字典
|
||
|
||
self.logList = [] # 日志记录
|
||
|
||
# 当前最新数据，用于模拟成交用
|
||
self.tick = None
|
||
self.bar = None
|
||
self.dt = None # 最新的时间
|
||
self.gatewayName = u'BackTest'
|
||
|
||
# csvFile相关
|
||
self.barTimeInterval = 60 # csv文件，属于K线类型，K线的周期（秒数）,缺省是1分钟
|
||
|
||
# 费用情况
|
||
self.avaliable = EMPTY_FLOAT
|
||
self.percent = EMPTY_FLOAT
|
||
self.percentLimit = 30 # 投资仓位比例上限
|
||
|
||
# 回测计算相关
|
||
self.calculateMode = self.FINAL_MODE
|
||
self.usageCompounding = False # 是否使用简单复利 （只针对FINAL_MODE有效）
|
||
|
||
self.initCapital = 10000 # 期初资金
|
||
self.capital = self.initCapital # 资金 （相当于Balance）
|
||
self.maxCapital = self.initCapital # 资金最高净值
|
||
|
||
self.maxPnl = 0 # 最高盈利
|
||
self.minPnl = 0 # 最大亏损
|
||
self.maxVolume = 1 # 最大仓位数
|
||
self.winningResult = 0 # 盈利次数
|
||
self.losingResult = 0 # 亏损次数
|
||
|
||
self.totalResult = 0 # 总成交数量
|
||
self.totalWinning = 0 # 总盈利
|
||
self.totalLosing = 0 # 总亏损
|
||
self.totalTurnover = 0 # 总成交金额（合约面值）
|
||
self.totalCommission = 0 # 总手续费
|
||
self.totalSlippage = 0 # 总滑点
|
||
|
||
self.timeList = [] # 时间序列
|
||
self.pnlList = [] # 每笔盈亏序列
|
||
self.capitalList = [] # 盈亏汇总的时间序列
|
||
self.drawdownList = [] # 回撤的时间序列
|
||
self.drawdownRateList = [] # 最大回撤比例的时间序列
|
||
|
||
self.dailyList = []
|
||
self.exportTradeList = [] # 导出交易记录列表
|
||
|
||
self.fixCommission = EMPTY_FLOAT # 固定交易费用
|
||
|
||
def getAccountInfo(self):
|
||
"""返回账号的实时权益，可用资金，仓位比例,投资仓位比例上限"""
|
||
if self.capital == EMPTY_FLOAT:
|
||
self.percent = EMPTY_FLOAT
|
||
|
||
return self.capital, self.avaliable, self.percent, self.percentLimit
|
||
|
||
#----------------------------------------------------------------------
|
||
def setStartDate(self, startDate='20100416', initDays=10):
|
||
"""设置回测的启动日期"""
|
||
self.startDate = startDate
|
||
self.initDays = initDays
|
||
|
||
self.dataStartDate = datetime.strptime(startDate, '%Y%m%d')
|
||
|
||
# 初始化天数
|
||
initTimeDelta = timedelta(initDays)
|
||
|
||
self.strategyStartDate = self.dataStartDate + initTimeDelta
|
||
|
||
#----------------------------------------------------------------------
|
||
def setEndDate(self, endDate=''):
|
||
"""设置回测的结束日期"""
|
||
self.endDate = endDate
|
||
if endDate:
|
||
self.dataEndDate = datetime.strptime(endDate, '%Y%m%d')
|
||
# 若不修改时间则会导致不包含dataEndDate当天数据
|
||
self.dataEndDate.replace(hour=23, minute=59)
|
||
else:
|
||
self.dataEndDate = datetime.now()
|
||
|
||
def setMinDiff(self, minDiff):
|
||
"""设置回测品种的最小跳价，用于修正数据"""
|
||
self.minDiff = minDiff
|
||
self.priceTick = minDiff
|
||
|
||
#----------------------------------------------------------------------
|
||
def setBacktestingMode(self, mode):
|
||
"""设置回测模式"""
|
||
self.mode = mode
|
||
|
||
#----------------------------------------------------------------------
|
||
def setDatabase(self, dbName, symbol):
|
||
"""设置历史数据所用的数据库"""
|
||
self.dbName = dbName
|
||
self.symbol = symbol
|
||
|
||
def setMarginRate(self, margin_rate):
|
||
|
||
if margin_rate!= EMPTY_FLOAT:
|
||
self.margin_rate = margin_rate
|
||
|
||
# ----------------------------------------------------------------------
|
||
def setSlippage(self, slippage):
|
||
"""设置滑点点数"""
|
||
self.slippage = slippage
|
||
|
||
# ----------------------------------------------------------------------
|
||
def setSize(self, size):
|
||
"""设置合约大小"""
|
||
self.size = size
|
||
|
||
# ----------------------------------------------------------------------
|
||
def setRate(self, rate):
|
||
"""设置佣金比例"""
|
||
self.rate = rate
|
||
|
||
# ----------------------------------------------------------------------
|
||
def setPriceTick(self, priceTick):
|
||
"""设置价格最小变动"""
|
||
self.priceTick = priceTick
|
||
self.minDiff = priceTick
|
||
|
||
#----------------------------------------------------------------------
|
||
def loadHistoryDataFromMongo(self):
|
||
"""载入历史数据"""
|
||
host, port, log = loadMongoSetting()
|
||
|
||
self.dbClient = pymongo.MongoClient(host, port)
|
||
collection = self.dbClient[self.dbName][self.symbol]
|
||
|
||
self.output(u'开始载入数据')
|
||
|
||
# 首先根据回测模式，确认要使用的数据类
|
||
if self.mode == self.BAR_MODE:
|
||
dataClass = CtaBarData
|
||
func = self.newBar
|
||
else:
|
||
dataClass = CtaTickData
|
||
func = self.newTick
|
||
|
||
# 载入初始化需要用的数据
|
||
flt = {'datetime':{'$gte':self.dataStartDate,
|
||
'$lt':self.strategyStartDate}}
|
||
initCursor = collection.find(flt)
|
||
|
||
# 将数据从查询指针中读取出，并生成列表
|
||
self.initData = [] # 清空initData列表
|
||
for d in initCursor:
|
||
data = dataClass()
|
||
data.__dict__ = d
|
||
self.initData.append(data)
|
||
|
||
# 载入回测数据
|
||
if not self.dataEndDate:
|
||
flt = {'datetime':{'$gte':self.strategyStartDate}} # 数据过滤条件
|
||
else:
|
||
flt = {'datetime':{'$gte':self.strategyStartDate,
|
||
'$lte':self.dataEndDate}}
|
||
self.dbCursor = collection.find(flt)
|
||
|
||
self.output(u'载入完成，数据量：%s' %(initCursor.count() + self.dbCursor.count()))
|
||
|
||
#----------------------------------------------------------------------
|
||
def connectMysql(self):
|
||
"""连接MysqlDB"""
|
||
|
||
# 载入json文件
|
||
fileName = 'mysql_connect.json'
|
||
try:
|
||
f = file(fileName)
|
||
except IOError:
|
||
self.writeCtaLog(u'回测引擎读取Mysql_connect.json失败')
|
||
return
|
||
|
||
# 解析json文件
|
||
setting = json.load(f)
|
||
try:
|
||
mysql_host = str(setting['host'])
|
||
mysql_port = int(setting['port'])
|
||
mysql_user = str(setting['user'])
|
||
mysql_passwd = str(setting['passwd'])
|
||
mysql_db = str(setting['db'])
|
||
|
||
except IOError:
|
||
self.writeCtaLog(u'回测引擎读取Mysql_connect.json,连接配置缺少字段，请检查')
|
||
return
|
||
|
||
try:
|
||
self.__mysqlConnection = MySQLdb.connect(host=mysql_host, user=mysql_user,
|
||
passwd=mysql_passwd, db=mysql_db, port=mysql_port)
|
||
self.__mysqlConnected = True
|
||
self.writeCtaLog(u'回测引擎连接MysqlDB成功')
|
||
except Exception:
|
||
self.writeCtaLog(u'回测引擎连接MysqlDB失败')
|
||
|
||
#----------------------------------------------------------------------
|
||
def loadDataHistoryFromMysql(self, symbol, startDate, endDate):
|
||
"""载入历史TICK数据
|
||
如果加载过多数据会导致加载失败,间隔不要超过半年
|
||
"""
|
||
|
||
if not endDate:
|
||
endDate = datetime.today()
|
||
|
||
# 看本地缓存是否存在
|
||
if self.__loadDataHistoryFromLocalCache(symbol, startDate, endDate):
|
||
self.writeCtaLog(u'历史TICK数据从Cache载入')
|
||
return
|
||
|
||
# 每次获取日期周期
|
||
intervalDays = 10
|
||
|
||
for i in range (0,(endDate - startDate).days +1, intervalDays):
|
||
d1 = startDate + timedelta(days = i )
|
||
|
||
if (endDate - d1).days > 10:
|
||
d2 = startDate + timedelta(days = i + intervalDays -1 )
|
||
else:
|
||
d2 = endDate
|
||
|
||
# 从Mysql 提取数据
|
||
self.__qryDataHistoryFromMysql(symbol, d1, d2)
|
||
|
||
self.writeCtaLog(u'历史TICK数据共载入{0}条'.format(len(self.historyData)))
|
||
|
||
# 保存本地cache文件
|
||
self.__saveDataHistoryToLocalCache(symbol, startDate, endDate)
|
||
|
||
|
||
def __loadDataHistoryFromLocalCache(self, symbol, startDate, endDate):
|
||
"""看本地缓存是否存在
|
||
added by IncenseLee
|
||
"""
|
||
|
||
# 运行路径下cache子目录
|
||
cacheFolder = os.getcwd()+'/cache'
|
||
|
||
# cache文件
|
||
cacheFile = u'{0}/{1}_{2}_{3}.pickle'.\
|
||
format(cacheFolder, symbol, startDate.strftime('%Y-%m-%d'), endDate.strftime('%Y-%m-%d'))
|
||
|
||
if not os.path.isfile(cacheFile):
|
||
return False
|
||
|
||
else:
|
||
try:
|
||
# 从cache文件加载
|
||
cache = open(cacheFile,mode='r')
|
||
self.historyData = cPickle.load(cache)
|
||
cache.close()
|
||
return True
|
||
except Exception as e:
|
||
self.writeCtaLog(u'读取文件{0}失败'.format(cacheFile))
|
||
return False
|
||
|
||
def __saveDataHistoryToLocalCache(self, symbol, startDate, endDate):
|
||
"""保存本地缓存
|
||
added by IncenseLee
|
||
"""
|
||
|
||
# 运行路径下cache子目录
|
||
cacheFolder = os.getcwd()+'/cache'
|
||
|
||
# 创建cache子目录
|
||
if not os.path.isdir(cacheFolder):
|
||
os.mkdir(cacheFolder)
|
||
|
||
# cache 文件名
|
||
cacheFile = u'{0}/{1}_{2}_{3}.pickle'.\
|
||
format(cacheFolder, symbol, startDate.strftime('%Y-%m-%d'), endDate.strftime('%Y-%m-%d'))
|
||
|
||
# 重复存在 返回
|
||
if os.path.isfile(cacheFile):
|
||
return False
|
||
|
||
else:
|
||
# 写入cache文件
|
||
cache = open(cacheFile, mode='w')
|
||
cPickle.dump(self.historyData,cache)
|
||
cache.close()
|
||
return True
|
||
|
||
#----------------------------------------------------------------------
|
||
def __qryDataHistoryFromMysql(self, symbol, startDate, endDate):
|
||
"""从Mysql载入历史TICK数据
|
||
added by IncenseLee
|
||
"""
|
||
|
||
try:
|
||
self.connectMysql()
|
||
if self.__mysqlConnected:
|
||
|
||
# 获取指针
|
||
cur = self.__mysqlConnection.cursor(MySQLdb.cursors.DictCursor)
|
||
|
||
if endDate:
|
||
|
||
# 开始日期 ~ 结束日期
|
||
sqlstring = ' select \'{0}\' as InstrumentID, str_to_date(concat(ndate,\' \', ntime),' \
|
||
'\'%Y-%m-%d %H:%i:%s\') as UpdateTime,price as LastPrice,vol as Volume, day_vol as DayVolume,' \
|
||
'position_vol as OpenInterest,bid1_price as BidPrice1,bid1_vol as BidVolume1, ' \
|
||
'sell1_price as AskPrice1, sell1_vol as AskVolume1 from TB_{0}MI ' \
|
||
'where ndate between cast(\'{1}\' as date) and cast(\'{2}\' as date) order by UpdateTime'.\
|
||
format(symbol, startDate, endDate)
|
||
|
||
elif startDate:
|
||
|
||
# 开始日期 - 当前
|
||
sqlstring = ' select \'{0}\' as InstrumentID,str_to_date(concat(ndate,\' \', ntime),' \
|
||
'\'%Y-%m-%d %H:%i:%s\') as UpdateTime,price as LastPrice,vol as Volume, day_vol as DayVolume,' \
|
||
'position_vol as OpenInterest,bid1_price as BidPrice1,bid1_vol as BidVolume1, ' \
|
||
'sell1_price as AskPrice1, sell1_vol as AskVolume1 from TB__{0}MI ' \
|
||
'where ndate > cast(\'{1}\' as date) order by UpdateTime'.\
|
||
format( symbol, startDate)
|
||
|
||
else:
|
||
|
||
# 所有数据
|
||
sqlstring =' select \'{0}\' as InstrumentID,str_to_date(concat(ndate,\' \', ntime),' \
|
||
'\'%Y-%m-%d %H:%i:%s\') as UpdateTime,price as LastPrice,vol as Volume, day_vol as DayVolume,' \
|
||
'position_vol as OpenInterest,bid1_price as BidPrice1,bid1_vol as BidVolume1, ' \
|
||
'sell1_price as AskPrice1, sell1_vol as AskVolume1 from TB__{0}MI order by UpdateTime'.\
|
||
format(symbol)
|
||
|
||
self.writeCtaLog(sqlstring)
|
||
|
||
# 执行查询
|
||
count = cur.execute(sqlstring)
|
||
self.writeCtaLog(u'历史TICK数据共{0}条'.format(count))
|
||
|
||
|
||
# 分批次读取
|
||
fetch_counts = 0
|
||
fetch_size = 1000
|
||
|
||
while True:
|
||
results = cur.fetchmany(fetch_size)
|
||
|
||
if not results:
|
||
break
|
||
|
||
fetch_counts = fetch_counts + len(results)
|
||
|
||
if not self.historyData:
|
||
self.historyData =results
|
||
|
||
else:
|
||
self.historyData = self.historyData + results
|
||
|
||
self.writeCtaLog(u'{1}~{2}历史TICK数据载入共{0}条'.format(fetch_counts,startDate,endDate))
|
||
|
||
|
||
else:
|
||
self.writeCtaLog(u'MysqlDB未连接，请检查')
|
||
|
||
except MySQLdb.Error as e:
|
||
self.writeCtaLog(u'MysqlDB载入数据失败，请检查.Error {0}'.format(e))
|
||
|
||
def __dataToTick(self, data):
|
||
"""
|
||
数据库查询返回的data结构，转换为tick对象
|
||
added by IncenseLee """
|
||
|
||
tick = CtaTickData()
|
||
symbol = data['InstrumentID']
|
||
tick.symbol = symbol
|
||
|
||
# 创建TICK数据对象并更新数据
|
||
tick.vtSymbol = symbol
|
||
# tick.openPrice = data['OpenPrice']
|
||
# tick.highPrice = data['HighestPrice']
|
||
# tick.lowPrice = data['LowestPrice']
|
||
tick.lastPrice = float(data['LastPrice'])
|
||
|
||
# bug fix:
|
||
# ctp日常传送的volume数据，是交易日日内累加值。数据库的volume，是数据商自行计算整理的
|
||
# 因此，改为使用DayVolume，与CTP实盘一致
|
||
#tick.volume = data['Volume']
|
||
tick.volume = data['DayVolume']
|
||
tick.openInterest = data['OpenInterest']
|
||
|
||
# tick.upperLimit = data['UpperLimitPrice']
|
||
# tick.lowerLimit = data['LowerLimitPrice']
|
||
|
||
tick.datetime = data['UpdateTime']
|
||
tick.date = tick.datetime.strftime('%Y-%m-%d')
|
||
tick.time = tick.datetime.strftime('%H:%M:%S')
|
||
# 数据库中并没有tradingDay的数据，回测时，暂时按照date授予。
|
||
tick.tradingDay = tick.date
|
||
|
||
tick.bidPrice1 = float(data['BidPrice1'])
|
||
# tick.bidPrice2 = data['BidPrice2']
|
||
# tick.bidPrice3 = data['BidPrice3']
|
||
# tick.bidPrice4 = data['BidPrice4']
|
||
# tick.bidPrice5 = data['BidPrice5']
|
||
|
||
tick.askPrice1 = float(data['AskPrice1'])
|
||
# tick.askPrice2 = data['AskPrice2']
|
||
# tick.askPrice3 = data['AskPrice3']
|
||
# tick.askPrice4 = data['AskPrice4']
|
||
# tick.askPrice5 = data['AskPrice5']
|
||
|
||
tick.bidVolume1 = data['BidVolume1']
|
||
# tick.bidVolume2 = data['BidVolume2']
|
||
# tick.bidVolume3 = data['BidVolume3']
|
||
# tick.bidVolume4 = data['BidVolume4']
|
||
# tick.bidVolume5 = data['BidVolume5']
|
||
|
||
tick.askVolume1 = data['AskVolume1']
|
||
# tick.askVolume2 = data['AskVolume2']
|
||
# tick.askVolume3 = data['AskVolume3']
|
||
# tick.askVolume4 = data['AskVolume4']
|
||
# tick.askVolume5 = data['AskVolume5']
|
||
|
||
return tick
|
||
|
||
#----------------------------------------------------------------------
|
||
def getMysqlDeltaDate(self,symbol, startDate, decreaseDays):
|
||
"""从mysql库中获取交易日前若干天
|
||
added by IncenseLee
|
||
"""
|
||
try:
|
||
if self.__mysqlConnected:
|
||
|
||
# 获取mysql指针
|
||
cur = self.__mysqlConnection.cursor()
|
||
|
||
sqlstring='select distinct ndate from TB_{0}MI where ndate < ' \
|
||
'cast(\'{1}\' as date) order by ndate desc limit {2},1'.format(symbol, startDate, decreaseDays-1)
|
||
|
||
# self.writeCtaLog(sqlstring)
|
||
|
||
count = cur.execute(sqlstring)
|
||
|
||
if count > 0:
|
||
|
||
# 提取第一条记录
|
||
result = cur.fetchone()
|
||
|
||
return result[0]
|
||
|
||
else:
|
||
self.writeCtaLog(u'MysqlDB没有查询结果，请检查日期')
|
||
|
||
else:
|
||
self.writeCtaLog(u'MysqlDB未连接，请检查')
|
||
|
||
except MySQLdb.Error as e:
|
||
self.writeCtaLog(u'MysqlDB载入数据失败，请检查.Error {0}: {1}'.format(e.arg[0],e.arg[1]))
|
||
|
||
# 出错后缺省返回
|
||
return startDate-timedelta(days=3)
|
||
|
||
# ----------------------------------------------------------------------
|
||
def runBackTestingWithArbTickFile(self,mainPath, arbSymbol):
|
||
"""运行套利回测（使用本地tickcsv数据)
|
||
参数：套利代码 SP rb1610&rb1701
|
||
added by IncenseLee
|
||
原始的tick，分别存放在白天目录1和夜盘目录2中，每天都有各个合约的数据
|
||
Z:\ticks\SHFE\201606\RB\0601\
|
||
RB1610.txt
|
||
RB1701.txt
|
||
....
|
||
Z:\ticks\SHFE_night\201606\RB\0601
|
||
RB1610.txt
|
||
RB1701.txt
|
||
....
|
||
|
||
夜盘目录为自然日，不是交易日。
|
||
|
||
按照回测的开始日期，到结束日期，循环每一天。
|
||
每天优先读取日盘数据，再读取夜盘数据。
|
||
读取eg1（如RB1610），读取Leg2（如RB701），合并成价差tick，灌输到策略的onTick中。
|
||
"""
|
||
self.capital = self.initCapital # 更新设置期初资金
|
||
|
||
if len(arbSymbol) < 1:
|
||
self.writeCtaLog(u'套利合约为空')
|
||
return
|
||
|
||
if not (arbSymbol.upper().index("SP") == 0 and arbSymbol.index(" ") > 0 and arbSymbol.index("&") > 0):
|
||
self.writeCtaLog(u'套利合约格式不符合')
|
||
return
|
||
|
||
# 获得Leg1，leg2
|
||
legs = arbSymbol[arbSymbol.index(" "):]
|
||
leg1 = legs[1:legs.index("&")]
|
||
leg2 = legs[legs.index("&") + 1:]
|
||
self.writeCtaLog(u'Leg1:{0},Leg2:{1}'.format(leg1, leg2))
|
||
|
||
if not self.dataStartDate:
|
||
self.writeCtaLog(u'回测开始日期未设置。')
|
||
return
|
||
# RB
|
||
if len(self.symbol)<1:
|
||
self.writeCtaLog(u'回测对象未设置。')
|
||
return
|
||
|
||
if not self.dataEndDate:
|
||
self.dataEndDate = datetime.today()
|
||
|
||
#首先根据回测模式，确认要使用的数据类
|
||
if self.mode == self.BAR_MODE:
|
||
self.writeCtaLog(u'本回测仅支持tick模式')
|
||
return
|
||
|
||
testdays = (self.dataEndDate - self.dataStartDate).days
|
||
|
||
if testdays < 1:
|
||
self.writeCtaLog(u'回测时间不足')
|
||
return
|
||
|
||
for i in range(0, testdays):
|
||
|
||
testday = self.dataStartDate + timedelta(days = i)
|
||
|
||
self.output(u'回测日期:{0}'.format(testday))
|
||
|
||
# 白天数据
|
||
self.__loadArbTicks(mainPath,testday,leg1,leg2)
|
||
|
||
# 夜盘数据
|
||
self.__loadArbTicks(mainPath+'_night', testday, leg1, leg2)
|
||
|
||
|
||
def __loadArbTicks(self,mainPath,testday,leg1,leg2):
|
||
|
||
self.writeCtaLog(u'加载回测日期:{0}\{1}的价差tick'.format(mainPath, testday))
|
||
|
||
cachefilename = u'{0}_{1}_{2}_{3}_{4}'.format(self.symbol,leg1,leg2, mainPath, testday.strftime('%Y%m%d'))
|
||
|
||
arbTicks = self.__loadArbTicksFromLocalCache(cachefilename)
|
||
|
||
dt = None
|
||
|
||
if len(arbTicks) < 1:
|
||
|
||
leg1File = u'z:\\ticks\\{0}\\{1}\\{2}\\{3}\\{4}.txt' \
|
||
.format(mainPath, testday.strftime('%Y%m'), self.symbol, testday.strftime('%m%d'), leg1)
|
||
if not os.path.isfile(leg1File):
|
||
self.writeCtaLog(u'{0}文件不存在'.format(leg1File))
|
||
return
|
||
|
||
leg2File = u'z:\\ticks\\{0}\\{1}\\{2}\\{3}\\{4}.txt' \
|
||
.format(mainPath, testday.strftime('%Y%m'), self.symbol, testday.strftime('%m%d'), leg2)
|
||
if not os.path.isfile(leg2File):
|
||
self.writeCtaLog(u'{0}文件不存在'.format(leg2File))
|
||
return
|
||
|
||
# 先读取leg2的数据到目录，以日期时间为key
|
||
leg2Ticks = {}
|
||
|
||
leg2CsvReadFile = file(leg2File, 'rb')
|
||
#reader = csv.DictReader((line.replace('\0',' ') for line in leg2CsvReadFile), delimiter=",")
|
||
reader = csv.DictReader(leg2CsvReadFile, delimiter=",")
|
||
self.writeCtaLog(u'加载{0}'.format(leg2File))
|
||
for row in reader:
|
||
tick = CtaTickData()
|
||
|
||
tick.vtSymbol = self.symbol
|
||
tick.symbol = self.symbol
|
||
|
||
tick.date = testday.strftime('%Y%m%d')
|
||
tick.tradingDay = tick.date
|
||
tick.time = row['Time']
|
||
|
||
try:
|
||
tick.datetime = datetime.strptime(tick.date + ' ' + tick.time, '%Y%m%d %H:%M:%S.%f')
|
||
except Exception as ex:
|
||
self.writeCtaError(u'日期转换错误:{0},{1}:{2}'.format(tick.date + ' ' + tick.time, Exception, ex))
|
||
continue
|
||
|
||
# 修正毫秒
|
||
if tick.datetime.replace(microsecond = 0) == dt:
|
||
# 与上一个tick的时间（去除毫秒后）相同,修改为500毫秒
|
||
tick.datetime=tick.datetime.replace(microsecond = 500)
|
||
tick.time = tick.datetime.strftime('%H:%M:%S.%f')
|
||
|
||
else:
|
||
tick.datetime = tick.datetime.replace(microsecond=0)
|
||
tick.time = tick.datetime.strftime('%H:%M:%S.%f')
|
||
|
||
dt = tick.datetime
|
||
|
||
tick.lastPrice = float(row['LastPrice'])
|
||
tick.volume = int(float(row['LVolume']))
|
||
tick.bidPrice1 = float(row['BidPrice']) # 叫买价（价格低）
|
||
tick.bidVolume1 = int(float(row['BidVolume']))
|
||
tick.askPrice1 = float(row['AskPrice']) # 叫卖价（价格高）
|
||
tick.askVolume1 = int(float(row['AskVolume']))
|
||
|
||
# 排除涨停/跌停的数据
|
||
if (tick.bidPrice1 == float('1.79769E308') and tick.bidVolume1 == 0) \
|
||
or (tick.askPrice1 == float('1.79769E308') and tick.askVolume1 == 0):
|
||
continue
|
||
|
||
dtStr = tick.date + ' ' + tick.time
|
||
if dtStr in leg2Ticks:
|
||
self.writeCtaError(u'日内数据重复，异常,数据时间为:{0}'.format(dtStr))
|
||
else:
|
||
leg2Ticks[dtStr] = tick
|
||
|
||
leg1CsvReadFile = file(leg1File, 'rb')
|
||
#reader = csv.DictReader((line.replace('\0',' ') for line in leg1CsvReadFile), delimiter=",")
|
||
reader = csv.DictReader(leg1CsvReadFile, delimiter=",")
|
||
self.writeCtaLog(u'加载{0}'.format(leg1File))
|
||
|
||
dt = None
|
||
for row in reader:
|
||
|
||
arbTick = CtaTickData()
|
||
|
||
arbTick.date = testday.strftime('%Y%m%d')
|
||
arbTick.time = row['Time']
|
||
try:
|
||
arbTick.datetime = datetime.strptime(arbTick.date + ' ' + arbTick.time, '%Y%m%d %H:%M:%S.%f')
|
||
except Exception as ex:
|
||
self.writeCtaError(u'日期转换错误:{0},{1}:{2}'.format(arbTick.date + ' ' + arbTick.time, Exception, ex))
|
||
continue
|
||
|
||
# 修正毫秒
|
||
if arbTick.datetime.replace(microsecond=0) == dt:
|
||
# 与上一个tick的时间（去除毫秒后）相同,修改为500毫秒
|
||
arbTick.datetime = arbTick.datetime.replace(microsecond=500)
|
||
arbTick.time = arbTick.datetime.strftime('%H:%M:%S.%f')
|
||
|
||
else:
|
||
arbTick.datetime = arbTick.datetime.replace(microsecond=0)
|
||
arbTick.time = arbTick.datetime.strftime('%H:%M:%S.%f')
|
||
|
||
dt = arbTick.datetime
|
||
dtStr = ' '.join([arbTick.date, arbTick.time])
|
||
|
||
if dtStr in leg2Ticks:
|
||
leg2Tick = leg2Ticks[dtStr]
|
||
|
||
arbTick.vtSymbol = self.symbol
|
||
arbTick.symbol = self.symbol
|
||
|
||
arbTick.lastPrice = EMPTY_FLOAT
|
||
arbTick.volume = EMPTY_INT
|
||
|
||
leg1AskPrice1 = float(row['AskPrice'])
|
||
leg1AskVolume1 = int(float(row['AskVolume']))
|
||
|
||
leg1BidPrice1 = float(row['BidPrice'])
|
||
leg1BidVolume1 = int(float(row['BidVolume']))
|
||
|
||
# 排除涨停/跌停的数据
|
||
if ((leg1AskPrice1 == float('1.79769E308') or leg1AskPrice1 == 0) and leg1AskVolume1 == 0) \
|
||
or ((leg1BidPrice1 == float('1.79769E308') or leg1BidPrice1 == 0) and leg1BidVolume1 == 0):
|
||
continue
|
||
|
||
# 叫卖价差=leg1.askPrice1 - leg2.bidPrice1，volume为两者最小
|
||
arbTick.askPrice1 = leg1AskPrice1 - leg2Tick.bidPrice1
|
||
arbTick.askVolume1 = min(leg1AskVolume1, leg2Tick.bidVolume1)
|
||
|
||
# 叫买价差=leg1.bidPrice1 - leg2.askPrice1，volume为两者最小
|
||
arbTick.bidPrice1 = leg1BidPrice1 - leg2Tick.askPrice1
|
||
arbTick.bidVolume1 = min(leg1BidVolume1, leg2Tick.askVolume1)
|
||
|
||
arbTicks.append(arbTick)
|
||
|
||
del leg2Ticks[dtStr]
|
||
|
||
# 保存到历史目录
|
||
if len(arbTicks) > 0:
|
||
self.__saveArbTicksToLocalCache(cachefilename, arbTicks)
|
||
|
||
for t in arbTicks:
|
||
# 推送到策略中
|
||
self.newTick(t)
|
||
|
||
def __loadArbTicksFromLocalCache(self, filename):
|
||
"""从本地缓存中，加载数据"""
|
||
# 运行路径下cache子目录
|
||
cacheFolder = os.getcwd() + '/cache'
|
||
|
||
# cache文件
|
||
cacheFile = u'{0}/{1}.pickle'. \
|
||
format(cacheFolder, filename)
|
||
|
||
if not os.path.isfile(cacheFile):
|
||
return []
|
||
else:
|
||
# 从cache文件加载
|
||
cache = open(cacheFile, mode='r')
|
||
l = cPickle.load(cache)
|
||
cache.close()
|
||
return l
|
||
|
||
def __saveArbTicksToLocalCache(self, filename, arbticks):
|
||
"""保存价差tick到本地缓存目录"""
|
||
# 运行路径下cache子目录
|
||
cacheFolder = os.getcwd() + '/cache'
|
||
|
||
# 创建cache子目录
|
||
if not os.path.isdir(cacheFolder):
|
||
os.mkdir(cacheFolder)
|
||
|
||
# cache 文件名
|
||
cacheFile = u'{0}/{1}.pickle'. \
|
||
format(cacheFolder, filename)
|
||
|
||
# 重复存在 返回
|
||
if os.path.isfile(cacheFile):
|
||
return False
|
||
|
||
else:
|
||
# 写入cache文件
|
||
cache = open(cacheFile, mode='w')
|
||
cPickle.dump(arbticks, cache)
|
||
cache.close()
|
||
return True
|
||
|
||
def runBackTestingWithNonStrArbTickFile(self, leg1MainPath, leg2MainPath, leg1Symbol,leg2Symbol):
|
||
"""运行套利回测（使用本地tickcsv数据)
|
||
参数：
|
||
leg1MainPath： leg1合约所在的市场路径
|
||
leg2MainPath： leg2合约所在的市场路径
|
||
leg1Symbol： leg1合约
|
||
Leg2Symbol：leg2合约
|
||
added by IncenseLee
|
||
原始的tick，分别存放在白天目录1和夜盘目录2中，每天都有各个合约的数据
|
||
Z:\ticks\SHFE\201606\RB\0601\
|
||
RB1610.txt
|
||
RB1701.txt
|
||
....
|
||
Z:\ticks\SHFE_night\201606\RB\0601
|
||
RB1610.txt
|
||
RB1701.txt
|
||
....
|
||
|
||
夜盘目录为自然日，不是交易日。
|
||
|
||
按照回测的开始日期，到结束日期，循环每一天。
|
||
每天优先读取日盘数据，再读取夜盘数据。
|
||
读取eg1（如RB1610），读取Leg2（如RB701），根据两者tick的时间优先顺序，逐一tick灌输到策略的onTick中。
|
||
"""
|
||
self.capital = self.initCapital # 更新设置期初资金
|
||
|
||
if not self.dataStartDate:
|
||
self.writeCtaLog(u'回测开始日期未设置。')
|
||
return
|
||
# RB
|
||
if len(self.symbol)<1:
|
||
self.writeCtaLog(u'回测对象未设置。')
|
||
return
|
||
|
||
if not self.dataEndDate:
|
||
self.dataEndDate = datetime.today()
|
||
|
||
#首先根据回测模式，确认要使用的数据类
|
||
if self.mode == self.BAR_MODE:
|
||
self.writeCtaLog(u'本回测仅支持tick模式')
|
||
return
|
||
|
||
testdays = (self.dataEndDate - self.dataStartDate).days
|
||
|
||
if testdays < 1:
|
||
self.writeCtaLog(u'回测时间不足')
|
||
return
|
||
|
||
for i in range(0, testdays):
|
||
testday = self.dataStartDate + timedelta(days = i)
|
||
|
||
self.output(u'回测日期:{0}'.format(testday))
|
||
|
||
# 加载运行白天数据
|
||
self.__loadNotStdArbTicks(leg1MainPath, leg2MainPath, testday, leg1Symbol,leg2Symbol)
|
||
|
||
self.savingDailyData(testday, self.capital, self.maxCapital)
|
||
|
||
# 加载运行夜盘数据
|
||
self.__loadNotStdArbTicks(leg1MainPath+'_night', leg2MainPath+'_night', testday, leg1Symbol, leg2Symbol)
|
||
|
||
self.savingDailyData(self.dataEndDate, self.capital, self.maxCapital)
|
||
|
||
def __loadTicksFromFile(self, filepath, tickDate, vtSymbol):
|
||
"""从文件中读取tick"""
|
||
# 先读取数据到Dict，以日期时间为key
|
||
ticks = OrderedDict()
|
||
|
||
if not os.path.isfile(filepath):
|
||
self.writeCtaLog(u'{0}文件不存在'.format(filepath))
|
||
return ticks
|
||
dt = None
|
||
csvReadFile = file(filepath, 'rb')
|
||
|
||
reader = csv.DictReader(csvReadFile, delimiter=",")
|
||
self.writeCtaLog(u'加载{0}'.format(filepath))
|
||
for row in reader:
|
||
tick = CtaTickData()
|
||
|
||
tick.vtSymbol = vtSymbol
|
||
tick.symbol = vtSymbol
|
||
|
||
tick.date = tickDate.strftime('%Y%m%d')
|
||
tick.tradingDay = tick.date
|
||
tick.time = row['Time']
|
||
|
||
try:
|
||
tick.datetime = datetime.strptime(tick.date + ' ' + tick.time, '%Y%m%d %H:%M:%S.%f')
|
||
except Exception as ex:
|
||
self.writeCtaError(u'日期转换错误:{0},{1}:{2}'.format(tick.date + ' ' + tick.time, Exception, ex))
|
||
continue
|
||
|
||
# 修正毫秒
|
||
if tick.datetime.replace(microsecond=0) == dt:
|
||
# 与上一个tick的时间（去除毫秒后）相同,修改为500毫秒
|
||
tick.datetime = tick.datetime.replace(microsecond=500)
|
||
tick.time = tick.datetime.strftime('%H:%M:%S.%f')
|
||
|
||
else:
|
||
tick.datetime = tick.datetime.replace(microsecond=0)
|
||
tick.time = tick.datetime.strftime('%H:%M:%S.%f')
|
||
|
||
dt = tick.datetime
|
||
|
||
tick.lastPrice = float(row['LastPrice'])
|
||
tick.volume = int(float(row['LVolume']))
|
||
tick.bidPrice1 = float(row['BidPrice']) # 叫买价（价格低）
|
||
tick.bidVolume1 = int(float(row['BidVolume']))
|
||
tick.askPrice1 = float(row['AskPrice']) # 叫卖价（价格高）
|
||
tick.askVolume1 = int(float(row['AskVolume']))
|
||
|
||
# 排除涨停/跌停的数据
|
||
if (tick.bidPrice1 == float('1.79769E308') and tick.bidVolume1 == 0) \
|
||
or (tick.askPrice1 == float('1.79769E308') and tick.askVolume1 == 0):
|
||
continue
|
||
|
||
dtStr = tick.date + ' ' + tick.time
|
||
if dtStr in ticks:
|
||
self.writeCtaError(u'日内数据重复，异常,数据时间为:{0}'.format(dtStr))
|
||
else:
|
||
ticks[dtStr] = tick
|
||
|
||
return ticks
|
||
|
||
def __loadNotStdArbTicks(self, leg1MainPath,leg2MainPath, testday, leg1Symbol, leg2Symbol):
|
||
|
||
self.writeCtaLog(u'加载回测日期:{0}的价差tick'.format( testday))
|
||
p = re.compile(r"([A-Z]+)[0-9]+", re.I)
|
||
|
||
leg1_shortSymbol = p.match(leg1Symbol)
|
||
leg2_shortSymbol = p.match(leg2Symbol)
|
||
|
||
if leg1_shortSymbol is None or leg2_shortSymbol is None:
|
||
self.writeCtaLog(u'{0},{1}不能正则分解'.format(leg1Symbol, leg2Symbol))
|
||
return
|
||
|
||
leg1_shortSymbol = leg1_shortSymbol.group(1)
|
||
leg2_shortSymbol = leg2_shortSymbol.group(1)
|
||
|
||
# E:\Ticks\ZJ\2015\201505\TF
|
||
leg1File = u'{0}\\{1}\\{2}\\{3}\\{4}\\{5}.txt' \
|
||
.format(leg1MainPath, testday.strftime('%Y'),testday.strftime('%Y%m'), leg1_shortSymbol, testday.strftime('%m%d'), leg1Symbol)
|
||
if not os.path.isfile(leg1File):
|
||
self.writeCtaLog(u'{0}文件不存在'.format(leg1File))
|
||
return
|
||
|
||
leg2File = u'{0}\\{1}\\{2}\\{3}\\{4}\\{5}.txt' \
|
||
.format(leg2MainPath, testday.strftime('%Y'), testday.strftime('%Y%m'), leg2_shortSymbol, testday.strftime('%m%d'), leg2Symbol)
|
||
if not os.path.isfile(leg2File):
|
||
self.writeCtaLog(u'{0}文件不存在'.format(leg2File))
|
||
return
|
||
|
||
leg1Ticks = self.__loadTicksFromFile(filepath=leg1File,tickDate= testday, vtSymbol=leg1Symbol)
|
||
if len(leg1Ticks) == 0:
|
||
self.writeCtaLog(u'{0}读取tick数为空'.format(leg1File))
|
||
return
|
||
|
||
leg2Ticks = self.__loadTicksFromFile(filepath=leg2File, tickDate=testday, vtSymbol=leg2Symbol)
|
||
if len(leg2Ticks) == 0:
|
||
self.writeCtaLog(u'{0}读取tick数为空'.format(leg1File))
|
||
return
|
||
|
||
leg1_tick = None
|
||
leg2_tick = None
|
||
|
||
while not (len(leg1Ticks) == 0 or len(leg2Ticks) == 0):
|
||
if leg1_tick is None and len(leg1Ticks) > 0:
|
||
leg1_tick = leg1Ticks.popitem(last=False)
|
||
if leg2_tick is None and len(leg2Ticks) > 0:
|
||
leg2_tick = leg2Ticks.popitem(last=False)
|
||
|
||
if leg1_tick is None and leg2_tick is not None:
|
||
self.newTick(leg2_tick[1])
|
||
leg2_tick = None
|
||
elif leg1_tick is not None and leg2_tick is None:
|
||
self.newTick(leg1_tick[1])
|
||
leg1_tick = None
|
||
elif leg1_tick is not None and leg2_tick is not None:
|
||
leg1 = leg1_tick[1]
|
||
leg2 = leg2_tick[1]
|
||
if leg1.datetime <= leg2.datetime:
|
||
self.newTick(leg1)
|
||
leg1_tick = None
|
||
else:
|
||
self.newTick(leg2)
|
||
leg2_tick = None
|
||
|
||
def runBackTestingWithNonStrArbTickFile2(self, leg1MainPath, leg2MainPath, leg1Symbol, leg2Symbol):
|
||
"""运行套利回测（使用本地tickcsv数据，数据从taobao标普购买)
|
||
参数：
|
||
leg1MainPath： leg1合约所在的市场路径
|
||
leg2MainPath： leg2合约所在的市场路径
|
||
leg1Symbol： leg1合约
|
||
Leg2Symbol：leg2合约
|
||
added by IncenseLee
|
||
原始的tick，存放在相应市场下每天的目录中，目录包含市场各个合约的数据
|
||
E:\ticks\SQ\201606\20160601\
|
||
RB10.csv
|
||
RB01.csv
|
||
....
|
||
|
||
目录为交易日。
|
||
按照回测的开始日期，到结束日期，循环每一天。
|
||
|
||
读取eg1（如RB1610），读取Leg2（如RB701），根据两者tick的时间优先顺序，逐一tick灌输到策略的onTick中。
|
||
"""
|
||
self.capital = self.initCapital # 更新设置期初资金
|
||
|
||
if not self.dataStartDate:
|
||
self.writeCtaLog(u'回测开始日期未设置。')
|
||
return
|
||
# RB
|
||
if len(self.symbol) < 1:
|
||
self.writeCtaLog(u'回测对象未设置。')
|
||
return
|
||
|
||
if not self.dataEndDate:
|
||
self.dataEndDate = datetime.today()
|
||
|
||
# 首先根据回测模式，确认要使用的数据类
|
||
if self.mode == self.BAR_MODE:
|
||
self.writeCtaLog(u'本回测仅支持tick模式')
|
||
return
|
||
|
||
testdays = (self.dataEndDate - self.dataStartDate).days
|
||
|
||
if testdays < 1:
|
||
self.writeCtaLog(u'回测时间不足')
|
||
return
|
||
|
||
for i in range(0, testdays):
|
||
testday = self.dataStartDate + timedelta(days=i)
|
||
|
||
self.output(u'回测日期:{0}'.format(testday))
|
||
|
||
# 加载运行每天数据
|
||
self.__loadNotStdArbTicks2(leg1MainPath, leg2MainPath, testday, leg1Symbol, leg2Symbol)
|
||
|
||
self.savingDailyData(testday, self.capital, self.maxCapital)
|
||
|
||
|
||
def __loadTicksFromFile2(self, filepath, tickDate, vtSymbol):
|
||
"""从csv文件中UnicodeDictReader读取tick"""
|
||
# 先读取数据到Dict，以日期时间为key
|
||
ticks = OrderedDict()
|
||
|
||
if not os.path.isfile(filepath):
|
||
self.writeCtaLog(u'{0}文件不存在'.format(filepath))
|
||
return ticks
|
||
dt = None
|
||
csvReadFile = file(filepath, 'rb')
|
||
df = pd.read_csv(filepath, encoding='gbk')
|
||
df.columns = ['date', 'time', 'lastPrice', 'lastVolume', 'totalInterest', 'position',
|
||
'bidPrice1', 'bidVolume1', 'bidPrice2', 'bidVolume2', 'bidPrice3', 'bidVolume3',
|
||
'askPrice1', 'askVolume1', 'askPrice2', 'askVolume2', 'askPrice3', 'askVolume3','BS']
|
||
self.writeCtaLog(u'加载{0}'.format(filepath))
|
||
for i in range(0,len(df)):
|
||
#日期, 时间, 成交价, 成交量, 总量, 属性(持仓增减), B1价, B1量, B2价, B2量, B3价, B3量, S1价, S1量, S2价, S2量, S3价, S3量, BS
|
||
# 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18
|
||
row = df.iloc[i].to_dict()
|
||
|
||
tick = CtaTickData()
|
||
|
||
tick.vtSymbol = vtSymbol
|
||
tick.symbol = vtSymbol
|
||
|
||
tick.date = row['date']
|
||
tick.tradingDay = tickDate.strftime('%Y%m%d')
|
||
tick.time = row['time']
|
||
|
||
try:
|
||
tick.datetime = datetime.strptime(tick.date + ' ' + tick.time, '%Y-%m-%d %H:%M:%S')
|
||
except Exception as ex:
|
||
self.writeCtaError(u'日期转换错误:{0},{1}:{2}'.format(tick.date + ' ' + tick.time, Exception, ex))
|
||
continue
|
||
|
||
tick.date = tick.datetime.strftime('%Y%m%d')
|
||
# 修正毫秒
|
||
if tick.datetime.replace(microsecond=0) == dt:
|
||
# 与上一个tick的时间（去除毫秒后）相同,修改为500毫秒
|
||
tick.datetime = tick.datetime.replace(microsecond=500)
|
||
tick.time = tick.datetime.strftime('%H:%M:%S.%f')
|
||
|
||
else:
|
||
tick.datetime = tick.datetime.replace(microsecond=0)
|
||
tick.time = tick.datetime.strftime('%H:%M:%S.%f')
|
||
|
||
dt = tick.datetime
|
||
|
||
tick.lastPrice = float(row['lastPrice'])
|
||
tick.volume = int(float(row['lastVolume']))
|
||
tick.bidPrice1 = float(row['bidPrice1']) # 叫买价（价格低）
|
||
tick.bidVolume1 = int(float(row['bidVolume1']))
|
||
tick.askPrice1 = float(row['askPrice1']) # 叫卖价（价格高）
|
||
tick.askVolume1 = int(float(row['askVolume1']))
|
||
|
||
# 排除涨停/跌停的数据
|
||
if (tick.bidPrice1 == float('1.79769E308') and tick.bidVolume1 == 0) \
|
||
or (tick.askPrice1 == float('1.79769E308') and tick.askVolume1 == 0):
|
||
continue
|
||
|
||
dtStr = tick.date + ' ' + tick.time
|
||
if dtStr in ticks:
|
||
self.writeCtaError(u'日内数据重复，异常,数据时间为:{0}'.format(dtStr))
|
||
else:
|
||
ticks[dtStr] = tick
|
||
|
||
return ticks
|
||
|
||
def __loadNotStdArbTicks2(self, leg1MainPath, leg2MainPath, testday, leg1Symbol, leg2Symbol):
|
||
|
||
self.writeCtaLog(u'加载回测日期:{0}的价差tick'.format(testday))
|
||
p = re.compile(r"([A-Z]+)[0-9]+",re.I)
|
||
|
||
leg1_shortSymbol = p.match(leg1Symbol)
|
||
leg2_shortSymbol = p.match(leg2Symbol)
|
||
|
||
if leg1_shortSymbol is None or leg2_shortSymbol is None:
|
||
self.writeCtaLog(u'{0},{1}不能正则分解'.format(leg1Symbol,leg2Symbol))
|
||
return
|
||
|
||
leg1_shortSymbol = leg1_shortSymbol.group(1)
|
||
leg2_shortSymbol = leg2_shortSymbol.group(1)
|
||
|
||
# E:\Ticks\SQ\2014\201401\20140102\ag01_20140102.csv
|
||
leg1File = u'e:\\ticks\\{0}\\{1}\\{2}\\{3}\\{4}{5}_{3}.csv' \
|
||
.format(leg1MainPath, testday.strftime('%Y'), testday.strftime('%Y%m'), testday.strftime('%Y%m%d'), leg1_shortSymbol, leg1Symbol[-2:])
|
||
if not os.path.isfile(leg1File):
|
||
self.writeCtaLog(u'{0}文件不存在'.format(leg1File))
|
||
return
|
||
|
||
leg2File = u'e:\\ticks\\{0}\\{1}\\{2}\\{3}\\{4}{5}_{3}.csv' \
|
||
.format(leg2MainPath,testday.strftime('%Y'), testday.strftime('%Y%m'), testday.strftime('%Y%m%d'), leg2_shortSymbol, leg2Symbol[-2:])
|
||
if not os.path.isfile(leg2File):
|
||
self.writeCtaLog(u'{0}文件不存在'.format(leg2File))
|
||
return
|
||
|
||
leg1Ticks = self.__loadTicksFromFile2(filepath=leg1File, tickDate=testday, vtSymbol=leg1Symbol)
|
||
if len(leg1Ticks) == 0:
|
||
self.writeCtaLog(u'{0}读取tick数为空'.format(leg1File))
|
||
return
|
||
|
||
leg2Ticks = self.__loadTicksFromFile2(filepath=leg2File, tickDate=testday, vtSymbol=leg2Symbol)
|
||
if len(leg2Ticks) == 0:
|
||
self.writeCtaLog(u'{0}读取tick数为空'.format(leg1File))
|
||
return
|
||
|
||
leg1_tick = None
|
||
leg2_tick = None
|
||
|
||
while not (len(leg1Ticks) == 0 or len(leg2Ticks) == 0):
|
||
if leg1_tick is None and len(leg1Ticks) > 0:
|
||
leg1_tick = leg1Ticks.popitem(last=False)
|
||
if leg2_tick is None and len(leg2Ticks) > 0:
|
||
leg2_tick = leg2Ticks.popitem(last=False)
|
||
|
||
if leg1_tick is None and leg2_tick is not None:
|
||
self.newTick(leg2_tick[1])
|
||
leg2_tick = None
|
||
elif leg1_tick is not None and leg2_tick is None:
|
||
self.newTick(leg1_tick[1])
|
||
leg1_tick = None
|
||
elif leg1_tick is not None and leg2_tick is not None:
|
||
leg1 = leg1_tick[1]
|
||
leg2 = leg2_tick[1]
|
||
if leg1.datetime <= leg2.datetime:
|
||
self.newTick(leg1)
|
||
leg1_tick = None
|
||
else:
|
||
self.newTick(leg2)
|
||
leg2_tick = None
|
||
#----------------------------------------------------------------------
|
||
def runBackTestingWithBarFile(self, filename):
|
||
"""运行回测（使用本地csv数据)
|
||
added by IncenseLee
|
||
"""
|
||
self.capital = self.initCapital # 更新设置期初资金
|
||
if not filename:
|
||
self.writeCtaLog(u'请指定回测数据文件')
|
||
return
|
||
|
||
if not self.dataStartDate:
|
||
self.writeCtaLog(u'回测开始日期未设置。')
|
||
return
|
||
|
||
if not self.dataEndDate:
|
||
self.dataEndDate = datetime.today()
|
||
|
||
import os
|
||
if not os.path.isfile(filename):
|
||
self.writeCtaLog(u'{0}文件不存在'.format(filename))
|
||
|
||
if len(self.symbol) < 1:
|
||
self.writeCtaLog(u'回测对象未设置。')
|
||
return
|
||
|
||
# 首先根据回测模式，确认要使用的数据类
|
||
if not self.mode == self.BAR_MODE:
|
||
self.writeCtaLog(u'文件仅支持bar模式，若扩展tick模式，需要修改本方法')
|
||
return
|
||
|
||
self.output(u'开始回测')
|
||
|
||
#self.strategy.inited = True
|
||
self.strategy.onInit()
|
||
self.output(u'策略初始化完成')
|
||
|
||
self.strategy.trading = True
|
||
self.strategy.onStart()
|
||
self.output(u'策略启动完成')
|
||
|
||
self.output(u'开始回放数据')
|
||
|
||
import csv
|
||
|
||
csvfile = file(filename,'rb')
|
||
|
||
reader = csv.DictReader((line.replace('\0', '') for line in csvfile), delimiter=",")
|
||
|
||
for row in reader:
|
||
|
||
try:
|
||
|
||
bar = CtaBarData()
|
||
bar.symbol = self.symbol
|
||
bar.vtSymbol = self.symbol
|
||
|
||
# 从tb导出的csv文件
|
||
#bar.open = float(row['Open'])
|
||
#bar.high = float(row['High'])
|
||
#bar.low = float(row['Low'])
|
||
#bar.close = float(row['Close'])
|
||
#bar.volume = float(row['TotalVolume'])#
|
||
#barEndTime = datetime.strptime(row['Date']+' ' + row['Time'], '%Y/%m/%d %H:%M:%S')
|
||
|
||
# 从ricequant导出的csv文件
|
||
bar.open = float(row['open'])
|
||
bar.high = float(row['high'])
|
||
bar.low = float(row['low'])
|
||
bar.close = float(row['close'])
|
||
bar.volume = float(row['volume'])
|
||
barEndTime = datetime.strptime(row['index'], '%Y-%m-%d %H:%M:%S')
|
||
bar.tradingDay = row['trading_date']
|
||
|
||
# 使用Bar的开始时间作为datetime
|
||
bar.datetime = barEndTime - timedelta(seconds=self.barTimeInterval)
|
||
|
||
bar.date = bar.datetime.strftime('%Y-%m-%d')
|
||
bar.time = bar.datetime.strftime('%H:%M:%S')
|
||
|
||
if not (bar.datetime < self.dataStartDate or bar.datetime >= self.dataEndDate):
|
||
self.newBar(bar)
|
||
|
||
except Exception as ex:
|
||
self.writeCtaLog(u'{0}:{1}'.format(Exception, ex))
|
||
continue
|
||
|
||
#----------------------------------------------------------------------
|
||
def runBacktestingWithMysql(self):
|
||
"""运行回测(使用Mysql数据）
|
||
added by IncenseLee
|
||
"""
|
||
self.capital = self.initCapital # 更新设置期初资金
|
||
|
||
if not self.dataStartDate:
|
||
self.writeCtaLog(u'回测开始日期未设置。')
|
||
return
|
||
|
||
if not self.dataEndDate:
|
||
self.dataEndDate = datetime.today()
|
||
|
||
if len(self.symbol)<1:
|
||
self.writeCtaLog(u'回测对象未设置。')
|
||
return
|
||
|
||
# 首先根据回测模式，确认要使用的数据类
|
||
if self.mode == self.BAR_MODE:
|
||
dataClass = CtaBarData
|
||
func = self.newBar
|
||
else:
|
||
dataClass = CtaTickData
|
||
func = self.newTick
|
||
|
||
self.output(u'开始回测')
|
||
|
||
#self.strategy.inited = True
|
||
self.strategy.onInit()
|
||
self.output(u'策略初始化完成')
|
||
|
||
self.strategy.trading = True
|
||
self.strategy.onStart()
|
||
self.output(u'策略启动完成')
|
||
|
||
self.output(u'开始回放数据')
|
||
|
||
|
||
# 每次获取日期周期
|
||
intervalDays = 10
|
||
|
||
for i in range (0,(self.dataEndDate - self.dataStartDate).days +1, intervalDays):
|
||
d1 = self.dataStartDate + timedelta(days = i )
|
||
|
||
if (self.dataEndDate - d1).days > intervalDays:
|
||
d2 = self.dataStartDate + timedelta(days = i + intervalDays -1 )
|
||
else:
|
||
d2 = self.dataEndDate
|
||
|
||
# 提取历史数据
|
||
self.loadDataHistoryFromMysql(self.symbol, d1, d2)
|
||
|
||
self.output(u'数据日期:{0} => {1}'.format(d1,d2))
|
||
# 将逐笔数据推送
|
||
for data in self.historyData:
|
||
|
||
# 记录最新的TICK数据
|
||
self.tick = self.__dataToTick(data)
|
||
self.dt = self.tick.datetime
|
||
|
||
# 处理限价单
|
||
self.crossLimitOrder()
|
||
self.crossStopOrder()
|
||
|
||
# 推送到策略引擎中
|
||
self.strategy.onTick(self.tick)
|
||
|
||
# 清空历史数据
|
||
self.historyData = []
|
||
|
||
self.output(u'数据回放结束')
|
||
|
||
#----------------------------------------------------------------------
|
||
def runBacktesting(self):
|
||
"""运行回测"""
|
||
|
||
self.capital = self.initCapital # 更新设置期初资金
|
||
|
||
# 载入历史数据
|
||
#self.loadHistoryData()
|
||
self.loadHistoryDataFromMongo()
|
||
|
||
# 首先根据回测模式，确认要使用的数据类
|
||
if self.mode == self.BAR_MODE:
|
||
dataClass = CtaBarData
|
||
func = self.newBar
|
||
else:
|
||
dataClass = CtaTickData
|
||
func = self.newTick
|
||
|
||
self.output(u'开始回测')
|
||
|
||
self.strategy.inited = True
|
||
self.strategy.onInit()
|
||
self.output(u'策略初始化完成')
|
||
|
||
self.strategy.trading = True
|
||
self.strategy.onStart()
|
||
self.output(u'策略启动完成')
|
||
|
||
self.output(u'开始回放数据')
|
||
|
||
for d in self.dbCursor:
|
||
data = dataClass()
|
||
data.__dict__ = d
|
||
func(data)
|
||
|
||
self.output(u'数据回放结束')
|
||
|
||
def __sendOnBarEvent(self, bar):
|
||
"""发送Bar的事件"""
|
||
if self.eventEngine is not None:
|
||
eventType = EVENT_ON_BAR + '_' + self.symbol
|
||
event = Event(type_= eventType)
|
||
event.dict_['data'] = bar
|
||
self.eventEngine.put(event)
|
||
|
||
# ----------------------------------------------------------------------
|
||
def newBar(self, bar):
|
||
"""新的K线"""
|
||
self.bar = bar
|
||
self.dt = bar.datetime
|
||
self.crossLimitOrder() # 先撮合限价单
|
||
self.crossStopOrder() # 再撮合停止单
|
||
self.strategy.onBar(bar) # 推送K线到策略中
|
||
self.__sendOnBarEvent(bar) # 推送K线到事件
|
||
|
||
#----------------------------------------------------------------------
|
||
def newTick(self, tick):
|
||
"""新的Tick"""
|
||
self.tick = tick
|
||
self.dt = tick.datetime
|
||
self.crossLimitOrder()
|
||
self.crossStopOrder()
|
||
self.strategy.onTick(tick)
|
||
|
||
#----------------------------------------------------------------------
|
||
def initStrategy(self, strategyClass, setting=None):
|
||
"""
|
||
初始化策略
|
||
setting是策略的参数设置，如果使用类中写好的默认设置则可以不传该参数
|
||
"""
|
||
self.strategy = strategyClass(self, setting)
|
||
if not self.strategy.name:
|
||
self.strategy.name = self.strategy.className
|
||
|
||
self.strategy.onInit()
|
||
self.strategy.onStart()
|
||
|
||
#----------------------------------------------------------------------
|
||
def sendOrder(self, vtSymbol, orderType, price, volume, strategy):
|
||
"""发单"""
|
||
|
||
self.writeCtaLog(u'{0},{1},{2}@{3}'.format(vtSymbol, orderType, price, volume))
|
||
self.limitOrderCount += 1
|
||
orderID = str(self.limitOrderCount)
|
||
|
||
order = VtOrderData()
|
||
order.vtSymbol = vtSymbol
|
||
order.price = self.roundToPriceTick(price)
|
||
order.totalVolume = volume
|
||
order.status = STATUS_NOTTRADED # 刚提交尚未成交
|
||
order.orderID = orderID
|
||
order.vtOrderID = orderID
|
||
order.orderTime = str(self.dt)
|
||
|
||
# added by IncenseLee
|
||
order.gatewayName = self.gatewayName
|
||
|
||
# CTA委托类型映射
|
||
if orderType == CTAORDER_BUY:
|
||
order.direction = DIRECTION_LONG
|
||
order.offset = OFFSET_OPEN
|
||
elif orderType == CTAORDER_SELL:
|
||
order.direction = DIRECTION_SHORT
|
||
order.offset = OFFSET_CLOSE
|
||
elif orderType == CTAORDER_SHORT:
|
||
order.direction = DIRECTION_SHORT
|
||
order.offset = OFFSET_OPEN
|
||
elif orderType == CTAORDER_COVER:
|
||
order.direction = DIRECTION_LONG
|
||
order.offset = OFFSET_CLOSE
|
||
|
||
# modified by IncenseLee
|
||
key = u'{0}.{1}'.format(order.gatewayName, orderID)
|
||
# 保存到限价单字典中
|
||
self.workingLimitOrderDict[key] = order
|
||
self.limitOrderDict[key] = order
|
||
return key
|
||
|
||
#----------------------------------------------------------------------
|
||
def cancelOrder(self, vtOrderID):
|
||
"""撤单"""
|
||
if vtOrderID in self.workingLimitOrderDict:
|
||
order = self.workingLimitOrderDict[vtOrderID]
|
||
order.status = STATUS_CANCELLED
|
||
order.cancelTime = str(self.dt)
|
||
del self.workingLimitOrderDict[vtOrderID]
|
||
|
||
def cancelOrders(self, symbol, offset=EMPTY_STRING):
|
||
"""撤销所有单"""
|
||
# Symbol参数:指定合约的撤单；
|
||
# OFFSET参数:指定Offset的撤单,缺省不填写时，为所有
|
||
self.writeCtaLog(u'从所有订单中撤销{0}\{1}'.format(offset, symbol))
|
||
for vtOrderID in self.workingLimitOrderDict.keys():
|
||
order = self.workingLimitOrderDict[vtOrderID]
|
||
|
||
if offset == EMPTY_STRING:
|
||
offsetCond = True
|
||
else:
|
||
offsetCond = order.offset == offset
|
||
|
||
if order.symbol == symbol and offsetCond:
|
||
self.writeCtaLog(u'撤销订单:{0},{1} {2}@{3}'.format(vtOrderID, order.direction, order.price, order.totalVolume))
|
||
order.status = STATUS_CANCELLED
|
||
order.cancelTime = str(self.dt)
|
||
del self.workingLimitOrderDict[vtOrderID]
|
||
|
||
#----------------------------------------------------------------------
|
||
def sendStopOrder(self, vtSymbol, orderType, price, volume, strategy):
|
||
"""发停止单（本地实现）"""
|
||
|
||
self.stopOrderCount += 1
|
||
stopOrderID = STOPORDERPREFIX + str(self.stopOrderCount)
|
||
|
||
so = StopOrder()
|
||
so.vtSymbol = vtSymbol
|
||
so.price = self.roundToPriceTick(price)
|
||
so.volume = volume
|
||
so.strategy = strategy
|
||
so.stopOrderID = stopOrderID
|
||
so.status = STOPORDER_WAITING
|
||
|
||
if orderType == CTAORDER_BUY:
|
||
so.direction = DIRECTION_LONG
|
||
so.offset = OFFSET_OPEN
|
||
elif orderType == CTAORDER_SELL:
|
||
so.direction = DIRECTION_SHORT
|
||
so.offset = OFFSET_CLOSE
|
||
elif orderType == CTAORDER_SHORT:
|
||
so.direction = DIRECTION_SHORT
|
||
so.offset = OFFSET_OPEN
|
||
elif orderType == CTAORDER_COVER:
|
||
so.direction = DIRECTION_LONG
|
||
so.offset = OFFSET_CLOSE
|
||
|
||
# 保存stopOrder对象到字典中
|
||
self.stopOrderDict[stopOrderID] = so
|
||
self.workingStopOrderDict[stopOrderID] = so
|
||
|
||
return stopOrderID
|
||
|
||
#----------------------------------------------------------------------
|
||
def cancelStopOrder(self, stopOrderID):
|
||
"""撤销停止单"""
|
||
# 检查停止单是否存在
|
||
if stopOrderID in self.workingStopOrderDict:
|
||
so = self.workingStopOrderDict[stopOrderID]
|
||
so.status = STOPORDER_CANCELLED
|
||
del self.workingStopOrderDict[stopOrderID]
|
||
|
||
#----------------------------------------------------------------------
|
||
def crossLimitOrder(self):
|
||
"""基于最新数据撮合限价单"""
|
||
# 先确定会撮合成交的价格
|
||
if self.mode == self.BAR_MODE:
|
||
buyCrossPrice = self.bar.low # 若买入方向限价单价格高于该价格，则会成交
|
||
sellCrossPrice = self.bar.high # 若卖出方向限价单价格低于该价格，则会成交
|
||
buyBestCrossPrice = self.bar.open # 在当前时间点前发出的买入委托可能的最优成交价
|
||
sellBestCrossPrice = self.bar.open # 在当前时间点前发出的卖出委托可能的最优成交价
|
||
vtSymbol = self.bar.vtSymbol
|
||
else:
|
||
buyCrossPrice = self.tick.askPrice1
|
||
sellCrossPrice = self.tick.bidPrice1
|
||
buyBestCrossPrice = self.tick.askPrice1
|
||
sellBestCrossPrice = self.tick.bidPrice1
|
||
vtSymbol = self.tick.vtSymbol
|
||
|
||
# 遍历限价单字典中的所有限价单
|
||
for orderID, order in self.workingLimitOrderDict.items():
|
||
# 判断是否会成交
|
||
buyCross = order.direction == DIRECTION_LONG and order.price >= buyCrossPrice and vtSymbol == order.vtSymbol
|
||
sellCross = order.direction == DIRECTION_SHORT and order.price <= sellCrossPrice and vtSymbol == order.vtSymbol
|
||
|
||
# 如果发生了成交
|
||
if buyCross or sellCross:
|
||
# 推送成交数据
|
||
self.tradeCount += 1 # 成交编号自增1
|
||
|
||
tradeID = str(self.tradeCount)
|
||
trade = VtTradeData()
|
||
trade.vtSymbol = order.vtSymbol
|
||
trade.tradeID = tradeID
|
||
trade.vtTradeID = tradeID
|
||
trade.orderID = order.orderID
|
||
trade.vtOrderID = order.orderID
|
||
trade.direction = order.direction
|
||
trade.offset = order.offset
|
||
|
||
# 以买入为例：
|
||
# 1. 假设当根K线的OHLC分别为：100, 125, 90, 110
|
||
# 2. 假设在上一根K线结束(也是当前K线开始)的时刻，策略发出的委托为限价105
|
||
# 3. 则在实际中的成交价会是100而不是105，因为委托发出时市场的最优价格是100
|
||
if buyCross:
|
||
trade.price = min(order.price, buyBestCrossPrice)
|
||
self.strategy.pos += order.totalVolume
|
||
else:
|
||
trade.price = max(order.price, sellBestCrossPrice)
|
||
self.strategy.pos -= order.totalVolume
|
||
|
||
trade.volume = order.totalVolume
|
||
trade.tradeTime = str(self.dt)
|
||
trade.dt = self.dt
|
||
self.strategy.onTrade(trade)
|
||
|
||
self.tradeDict[tradeID] = trade
|
||
self.writeCtaLog(u'TradeId:{0}'.format(tradeID))
|
||
|
||
# 推送委托数据
|
||
order.tradedVolume = order.totalVolume
|
||
order.status = STATUS_ALLTRADED
|
||
|
||
self.strategy.onOrder(order)
|
||
|
||
# 从字典中删除该限价单
|
||
try:
|
||
del self.workingLimitOrderDict[orderID]
|
||
except Exception as ex:
|
||
self.writeCtaError(u'{0}:{1}'.format(Exception, ex))
|
||
|
||
# 实时计算模式
|
||
if self.calculateMode == self.REALTIME_MODE:
|
||
self.realtimeCalculate()
|
||
|
||
#----------------------------------------------------------------------
|
||
def crossStopOrder(self):
|
||
"""基于最新数据撮合停止单"""
|
||
# 先确定会撮合成交的价格，这里和限价单规则相反
|
||
if self.mode == self.BAR_MODE:
|
||
buyCrossPrice = self.bar.high # 若买入方向停止单价格低于该价格，则会成交
|
||
sellCrossPrice = self.bar.low # 若卖出方向限价单价格高于该价格，则会成交
|
||
bestCrossPrice = self.bar.open # 最优成交价，买入停止单不能低于，卖出停止单不能高于
|
||
vtSymbol = self.bar.vtSymbol
|
||
else:
|
||
buyCrossPrice = self.tick.lastPrice
|
||
sellCrossPrice = self.tick.lastPrice
|
||
bestCrossPrice = self.tick.lastPrice
|
||
vtSymbol = self.tick.vtSymbol
|
||
|
||
# 遍历停止单字典中的所有停止单
|
||
for stopOrderID, so in self.workingStopOrderDict.items():
|
||
# 判断是否会成交
|
||
buyCross = so.direction == DIRECTION_LONG and so.price <= buyCrossPrice and vtSymbol == so.vtSymbol
|
||
sellCross = so.direction == DIRECTION_SHORT and so.price >= sellCrossPrice and vtSymbol == so.vtSymbol
|
||
|
||
# 如果发生了成交
|
||
if buyCross or sellCross:
|
||
# 推送成交数据
|
||
self.tradeCount += 1 # 成交编号自增1
|
||
tradeID = str(self.tradeCount)
|
||
trade = VtTradeData()
|
||
trade.vtSymbol = so.vtSymbol
|
||
trade.tradeID = tradeID
|
||
trade.vtTradeID = tradeID
|
||
|
||
if buyCross:
|
||
self.strategy.pos += so.volume
|
||
trade.price = max(bestCrossPrice, so.price)
|
||
else:
|
||
self.strategy.pos -= so.volume
|
||
trade.price = min(bestCrossPrice, so.price)
|
||
|
||
self.limitOrderCount += 1
|
||
orderID = str(self.limitOrderCount)
|
||
trade.orderID = orderID
|
||
trade.vtOrderID = orderID
|
||
|
||
trade.direction = so.direction
|
||
trade.offset = so.offset
|
||
trade.volume = so.volume
|
||
trade.tradeTime = str(self.dt)
|
||
trade.dt = self.dt
|
||
self.strategy.onTrade(trade)
|
||
|
||
self.tradeDict[tradeID] = trade
|
||
|
||
# 推送委托数据
|
||
so.status = STOPORDER_TRIGGERED
|
||
|
||
order = VtOrderData()
|
||
order.vtSymbol = so.vtSymbol
|
||
order.symbol = so.vtSymbol
|
||
order.orderID = orderID
|
||
order.vtOrderID = orderID
|
||
order.direction = so.direction
|
||
order.offset = so.offset
|
||
order.price = so.price
|
||
order.totalVolume = so.volume
|
||
order.tradedVolume = so.volume
|
||
order.status = STATUS_ALLTRADED
|
||
order.orderTime = trade.tradeTime
|
||
self.strategy.onOrder(order)
|
||
|
||
self.limitOrderDict[orderID] = order
|
||
|
||
# 从字典中删除该限价单
|
||
if stopOrderID in self.workingStopOrderDict:
|
||
del self.workingStopOrderDict[stopOrderID]
|
||
|
||
# 若采用实时计算净值
|
||
if self.calculateMode == self.REALTIME_MODE:
|
||
self.realtimeCalculate()
|
||
|
||
|
||
#----------------------------------------------------------------------
|
||
def insertData(self, dbName, collectionName, data):
|
||
"""考虑到回测中不允许向数据库插入数据，防止实盘交易中的一些代码出错"""
|
||
pass
|
||
|
||
#----------------------------------------------------------------------
|
||
def loadBar(self, dbName, collectionName, startDate):
|
||
"""直接返回初始化数据列表中的Bar"""
|
||
return self.initData
|
||
|
||
#----------------------------------------------------------------------
|
||
def loadTick(self, dbName, collectionName, startDate):
|
||
"""直接返回初始化数据列表中的Tick"""
|
||
return self.initData
|
||
|
||
#----------------------------------------------------------------------
|
||
def writeCtaLog(self, content):
|
||
"""记录日志"""
|
||
#log = str(self.dt) + ' ' + content
|
||
#self.logList.append(log)
|
||
|
||
# 写入本地log日志
|
||
logging.info(content)
|
||
|
||
def writeCtaError(self, content):
|
||
"""记录异常"""
|
||
self.output(content)
|
||
self.writeCtaLog(content)
|
||
|
||
#----------------------------------------------------------------------
|
||
def output(self, content):
|
||
"""输出内容"""
|
||
print str(datetime.now()) + "\t" + content
|
||
|
||
def realtimeCalculate(self):
|
||
"""实时计算交易结果"""
|
||
|
||
resultDict = OrderedDict() # 交易结果记录
|
||
|
||
longTrade = [] # 未平仓的多头交易
|
||
shortTrade = [] # 未平仓的空头交易
|
||
longid = EMPTY_STRING
|
||
shortid = EMPTY_STRING
|
||
|
||
no_match_shortTrade = False
|
||
no_match_longTrade = False
|
||
|
||
# 对交易记录逐一处理
|
||
for tradeid in self.tradeDict.keys():
|
||
trade = self.tradeDict[tradeid]
|
||
# 多头交易
|
||
if trade.direction == DIRECTION_LONG:
|
||
# 存在空单
|
||
if len(shortTrade)>0:
|
||
# 检查是否存在与Symbol一致的空单
|
||
pop_indexs = [i for i, val in enumerate(shortTrade) if val.vtSymbol == trade.vtSymbol]
|
||
if len(pop_indexs) < 1:
|
||
#self.output(u'空头交易清单，没有{0}的空单'.format(trade.vtSymbol))
|
||
no_match_shortTrade = True
|
||
|
||
# 如果尚无空单
|
||
if len(shortTrade) == 0 or no_match_shortTrade:
|
||
#self.output(u'{0}多开:{1},{2}'.format(trade.vtSymbol, trade.volume, trade.price))
|
||
#self.writeCtaLog(u'{0}多开:{1},{2}'.format(trade.vtSymbol, trade.volume, trade.price))
|
||
longTrade.append(trade)
|
||
longid = tradeid
|
||
no_match_shortTrade = False
|
||
|
||
# 当前多头交易为平空
|
||
else:
|
||
gId = tradeid # 交易组（多个平仓数为一组）
|
||
gr = None # 组合的交易结果
|
||
|
||
coverVolume = trade.volume
|
||
|
||
while coverVolume > 0:
|
||
if len(shortTrade)==0:
|
||
self.writeCtaError(u'异常，没有开空仓的数据')
|
||
break
|
||
pop_indexs = [i for i, val in enumerate(shortTrade) if val.vtSymbol == trade.vtSymbol]
|
||
if len(pop_indexs) < 1:
|
||
self.writeCtaError(u'没有对应的symbol:{0}开空仓数据'.format(trade.vtSymbol))
|
||
break
|
||
pop_index = pop_indexs[0]
|
||
# 从未平仓的空头交易
|
||
entryTrade = shortTrade.pop(pop_index)
|
||
|
||
# 开空volume，不大于平仓volume
|
||
if coverVolume >= entryTrade.volume:
|
||
self.writeCtaLog(u'coverVolume:{0} >= entryTrade.volume:{1}'.format(coverVolume, entryTrade.volume))
|
||
coverVolume = coverVolume - entryTrade.volume
|
||
|
||
#self.output(u'{0}空平:{1},{2}'.format(entryTrade.vtSymbol, entryTrade.volume, trade.price))
|
||
#self.writeCtaLog(u'{0}空平:{1},{2}'.format(entryTrade.vtSymbol, entryTrade.volume, trade.price))
|
||
|
||
result = TradingResult(entryPrice=entryTrade.price,
|
||
entryDt=entryTrade.dt,
|
||
exitPrice=trade.price,
|
||
exitDt=trade.dt,
|
||
volume=-entryTrade.volume,
|
||
rate=self.rate,
|
||
slippage=self.slippage,
|
||
size=self.size,
|
||
groupId=gId,
|
||
fixcommission=self.fixCommission)
|
||
|
||
t = {}
|
||
t['vtSymbol'] = entryTrade.vtSymbol
|
||
t['OpenTime'] = entryTrade.tradeTime
|
||
t['OpenPrice'] = entryTrade.price
|
||
t['Direction'] = u'Short'
|
||
t['CloseTime'] = trade.tradeTime
|
||
t['ClosePrice'] = trade.price
|
||
t['Volume'] = entryTrade.volume
|
||
t['Profit'] = result.pnl
|
||
self.exportTradeList.append(t)
|
||
|
||
msg = u'Gid:{0} {1}[{2}:开空tid={3}:{4}]-[{5}.平空tid={6},{7},vol:{8}],净盈亏：{9}'\
|
||
.format(gId, entryTrade.vtSymbol, entryTrade.tradeTime, shortid, entryTrade.price,
|
||
trade.tradeTime, tradeid, trade.price,
|
||
entryTrade.volume, result.pnl)
|
||
self.output(msg)
|
||
|
||
self.writeCtaLog(msg)
|
||
|
||
if type(gr) == type(None):
|
||
if coverVolume > 0:
|
||
# 属于组合
|
||
gr = copy.deepcopy(result)
|
||
|
||
# 删除开空交易单
|
||
del self.tradeDict[entryTrade.tradeID]
|
||
|
||
else:
|
||
# 不属于组合
|
||
resultDict[entryTrade.dt] = result
|
||
|
||
# 删除平空交易单，
|
||
del self.tradeDict[trade.tradeID]
|
||
# 删除开空交易单
|
||
del self.tradeDict[entryTrade.tradeID]
|
||
|
||
else:
|
||
# 更新组合的数据
|
||
gr.turnover = gr.turnover + result.turnover
|
||
gr.commission = gr.commission + result.commission
|
||
gr.slippage = gr.slippage + result.slippage
|
||
gr.pnl = gr.pnl + result.pnl
|
||
|
||
# 删除开空交易单
|
||
del self.tradeDict[entryTrade.tradeID]
|
||
|
||
# 所有仓位平完
|
||
if coverVolume == 0:
|
||
gr.volume = trade.volume
|
||
resultDict[entryTrade.dt] = gr
|
||
# 删除平空交易单，
|
||
del self.tradeDict[trade.tradeID]
|
||
|
||
# 开空volume,大于平仓volume，需要更新减少tradeDict的数量。
|
||
else:
|
||
self.writeCtaLog(u'Short volume:{0} > Cover volume:{1}，需要更新减少tradeDict的数量。'.format(entryTrade.volume,coverVolume))
|
||
shortVolume = entryTrade.volume - coverVolume
|
||
|
||
result = TradingResult(entryPrice=entryTrade.price,
|
||
entryDt=entryTrade.dt,
|
||
exitPrice=trade.price,
|
||
exitDt=trade.dt,
|
||
volume=-coverVolume,
|
||
rate=self.rate,
|
||
slippage=self.slippage,
|
||
size=self.size,
|
||
groupId=gId,
|
||
fixcommission=self.fixCommission)
|
||
|
||
t = {}
|
||
t['vtSymbol'] = entryTrade.vtSymbol
|
||
t['OpenTime'] = entryTrade.tradeTime
|
||
t['OpenPrice'] = entryTrade.price
|
||
t['Direction'] = u'Short'
|
||
t['CloseTime'] = trade.tradeTime
|
||
t['ClosePrice'] = trade.price
|
||
t['Volume'] = coverVolume
|
||
t['Profit'] = result.pnl
|
||
self.exportTradeList.append(t)
|
||
|
||
msg = u'Gid:{0} {1}[{2}:开空tid={3}:{4}]-[{5}.平空tid={6},{7},vol:{8}],净盈亏：{9}'\
|
||
.format(gId, entryTrade.vtSymbol, entryTrade.tradeTime, shortid, entryTrade.price,
|
||
trade.tradeTime, tradeid, trade.price,
|
||
coverVolume, result.pnl)
|
||
self.output(msg)
|
||
self.writeCtaLog(msg)
|
||
|
||
# 更新（减少）开仓单的volume,重新推进开仓单列表中
|
||
entryTrade.volume = shortVolume
|
||
shortTrade.append(entryTrade)
|
||
|
||
coverVolume = 0
|
||
|
||
if type(gr) == type(None):
|
||
resultDict[entryTrade.dt] = result
|
||
|
||
else:
|
||
# 更新组合的数据
|
||
gr.turnover = gr.turnover + result.turnover
|
||
gr.commission = gr.commission + result.commission
|
||
gr.slippage = gr.slippage + result.slippage
|
||
gr.pnl = gr.pnl + result.pnl
|
||
gr.volume = trade.volume
|
||
resultDict[entryTrade.dt] = gr
|
||
|
||
# 删除平空交易单，
|
||
del self.tradeDict[trade.tradeID]
|
||
|
||
if type(gr) != type(None):
|
||
self.writeCtaLog(u'组合净盈亏:{0}'.format(gr.pnl))
|
||
|
||
self.writeCtaLog(u'-------------')
|
||
|
||
# 空头交易
|
||
else:
|
||
if len(longTrade) > 0:
|
||
pop_indexs = [i for i, val in enumerate(longTrade) if val.vtSymbol == trade.vtSymbol]
|
||
if len(pop_indexs) < 1:
|
||
#self.output(u'多头交易清单，没有{0}的多单'.format(trade.vtSymbol))
|
||
no_match_longTrade = True
|
||
|
||
# 如果尚无多单
|
||
if len(longTrade) == 0 or no_match_longTrade:
|
||
#self.output(u'{0}空开:{1},{2}'.format(trade.vtSymbol, trade.volume, trade.price))
|
||
#self.writeCtaLog(u'{0}空开:{1},{2}'.format(trade.vtSymbol, trade.volume, trade.price))
|
||
shortTrade.append(trade)
|
||
shortid = tradeid
|
||
no_match_longTrade = False
|
||
|
||
# 当前空头交易为平多
|
||
else:
|
||
gId = tradeid # 交易组（多个平仓数为一组） s
|
||
gr = None # 组合的交易结果
|
||
|
||
sellVolume = trade.volume
|
||
|
||
while sellVolume > 0:
|
||
if len(longTrade) == 0:
|
||
self.writeCtaError(u'异常，没有开多单')
|
||
break
|
||
|
||
pop_indexs = [i for i, val in enumerate(longTrade) if val.vtSymbol == trade.vtSymbol]
|
||
if len(pop_indexs) < 1:
|
||
self.writeCtaError(u'没有对应的symbol{0}开多仓数据,'.format(trade.vtSymbol))
|
||
break
|
||
|
||
pop_index = pop_indexs[0]
|
||
|
||
entryTrade = longTrade.pop(pop_index)
|
||
|
||
# 开多volume，不大于平仓volume
|
||
if sellVolume >= entryTrade.volume:
|
||
self.writeCtaLog(u'{0}Sell Volume:{1} >= Entry Volume:{2}'.format(entryTrade.vtSymbol, sellVolume, entryTrade.volume))
|
||
sellVolume = sellVolume - entryTrade.volume
|
||
#self.output(u'{0}多平:{1},{2}'.format(entryTrade.vtSymbol, entryTrade.volume, trade.price))
|
||
#self.writeCtaLog(u'{0}多平:{1},{2}'.format(entryTrade.vtSymbol, entryTrade.volume, trade.price))
|
||
|
||
|
||
result = TradingResult(entryPrice=entryTrade.price,
|
||
entryDt=entryTrade.dt,
|
||
exitPrice=trade.price,
|
||
exitDt=trade.dt,
|
||
volume=entryTrade.volume,
|
||
rate=self.rate,
|
||
slippage=self.slippage,
|
||
size=self.size,
|
||
groupId=gId,
|
||
fixcommission=self.fixCommission)
|
||
|
||
t = {}
|
||
t['vtSymbol'] = entryTrade.vtSymbol
|
||
t['OpenTime'] = entryTrade.tradeTime
|
||
t['OpenPrice'] = entryTrade.price
|
||
t['Direction'] = u'Long'
|
||
t['CloseTime'] = trade.tradeTime
|
||
t['ClosePrice'] = trade.price
|
||
t['Volume'] = entryTrade.volume
|
||
t['Profit'] = result.pnl
|
||
self.exportTradeList.append(t)
|
||
|
||
msg = u'Gid:{0} {1}[{2}:开多tid={3}:{4}]-[{5}.平多tid={6},{7},vol:{8}],净盈亏：{9}'\
|
||
.format(gId, entryTrade.vtSymbol,
|
||
entryTrade.tradeTime, longid, entryTrade.price,
|
||
trade.tradeTime, tradeid, trade.price,
|
||
entryTrade.volume, result.pnl)
|
||
self.output(msg)
|
||
self.writeCtaLog(msg)
|
||
|
||
if type(gr) == type(None):
|
||
if sellVolume > 0:
|
||
# 属于组合
|
||
gr = copy.deepcopy(result)
|
||
# 删除开多交易单
|
||
del self.tradeDict[entryTrade.tradeID]
|
||
|
||
else:
|
||
# 不属于组合
|
||
resultDict[entryTrade.dt] = result
|
||
|
||
# 删除平多交易单，
|
||
del self.tradeDict[trade.tradeID]
|
||
# 删除开多交易单
|
||
del self.tradeDict[entryTrade.tradeID]
|
||
|
||
else:
|
||
# 更新组合的数据
|
||
gr.turnover = gr.turnover + result.turnover
|
||
gr.commission = gr.commission + result.commission
|
||
gr.slippage = gr.slippage + result.slippage
|
||
gr.pnl = gr.pnl + result.pnl
|
||
|
||
# 删除开多交易单
|
||
del self.tradeDict[entryTrade.tradeID]
|
||
|
||
if sellVolume == 0:
|
||
gr.volume = trade.volume
|
||
resultDict[entryTrade.dt] = gr
|
||
# 删除平多交易单，
|
||
del self.tradeDict[trade.tradeID]
|
||
|
||
# 开多volume,大于平仓volume，需要更新减少tradeDict的数量。
|
||
else:
|
||
longVolume = entryTrade.volume -sellVolume
|
||
self.writeCtaLog(u'Long Volume:{0} > sell Volume:{1}'.format(entryTrade.volume,sellVolume))
|
||
|
||
result = TradingResult(entryPrice=entryTrade.price,
|
||
entryDt=entryTrade.dt,
|
||
exitPrice=trade.price,
|
||
exitDt=trade.dt,
|
||
volume=sellVolume,
|
||
rate=self.rate,
|
||
slippage=self.slippage,
|
||
size=self.size,
|
||
groupId=gId,
|
||
fixcommission=self.fixCommission)
|
||
|
||
t = {}
|
||
t['vtSymbol'] = entryTrade.vtSymbol
|
||
t['OpenTime'] = entryTrade.tradeTime
|
||
t['OpenPrice'] = entryTrade.price
|
||
t['Direction'] = u'Long'
|
||
t['CloseTime'] = trade.tradeTime
|
||
t['ClosePrice'] = trade.price
|
||
t['Volume'] = sellVolume
|
||
t['Profit'] = result.pnl
|
||
self.exportTradeList.append(t)
|
||
|
||
self.writeCtaLog(u'Gid:{0} {1}[{2}:开多tid={3}:{4}]-[{5}.平多tid={6},{7},vol:{8}],净盈亏：{9}'
|
||
.format(gId, entryTrade.vtSymbol,entryTrade.tradeTime, longid, entryTrade.price,
|
||
trade.tradeTime, tradeid, trade.price,
|
||
sellVolume, result.pnl))
|
||
|
||
# 减少开多volume,重新推进开多单列表中
|
||
entryTrade.volume = longVolume
|
||
longTrade.append(entryTrade)
|
||
|
||
sellVolume = 0
|
||
|
||
if type(gr) == type(None):
|
||
resultDict[entryTrade.dt] = result
|
||
|
||
else:
|
||
# 更新组合的数据
|
||
gr.turnover = gr.turnover + result.turnover
|
||
gr.commission = gr.commission + result.commission
|
||
gr.slippage = gr.slippage + result.slippage
|
||
gr.pnl = gr.pnl + result.pnl
|
||
gr.volume = trade.volume
|
||
resultDict[entryTrade.dt] = gr
|
||
|
||
# 删除平多交易单，
|
||
del self.tradeDict[trade.tradeID]
|
||
|
||
if type(gr) != type(None):
|
||
self.writeCtaLog(u'组合净盈亏:{0}'.format(gr.pnl))
|
||
|
||
self.writeCtaLog(u'-------------')
|
||
|
||
# 计算仓位比例
|
||
occupyMoney = EMPTY_FLOAT
|
||
occupyLongVolume = EMPTY_INT
|
||
occupyShortVolume = EMPTY_INT
|
||
if len(longTrade) > 0:
|
||
for t in longTrade:
|
||
occupyMoney += t.price * abs(t.volume) * self.size * self.margin_rate
|
||
occupyLongVolume += abs(t.volume)
|
||
if len(shortTrade) > 0:
|
||
for t in shortTrade:
|
||
occupyMoney += t.price * abs(t.volume) * self.size * self.margin_rate
|
||
occupyShortVolume += (t.volume)
|
||
|
||
self.output(u'occupyLongVolume:{0},occupyShortVolume:{1}'.format(occupyLongVolume,occupyShortVolume))
|
||
self.writeCtaLog(u'occupyLongVolume:{0},occupyShortVolume:{1}'.format(occupyLongVolume, occupyShortVolume))
|
||
# 最大持仓
|
||
self.maxVolume = max(self.maxVolume, occupyLongVolume + occupyShortVolume)
|
||
|
||
self.avaliable = self.capital - occupyMoney
|
||
self.percent = round(float(occupyMoney * 100 / self.capital), 2)
|
||
|
||
# 检查是否有平交易
|
||
if not resultDict:
|
||
|
||
msg = u''
|
||
if len(longTrade) > 0:
|
||
msg += u'持多仓{0},'.format(occupyLongVolume)
|
||
|
||
if len(shortTrade) > 0:
|
||
msg += u'持空仓{0},'.format(occupyShortVolume)
|
||
|
||
msg += u'资金占用:{0},仓位:{1}'.format(occupyMoney, self.percent)
|
||
self.output(msg)
|
||
self.writeCtaLog(msg)
|
||
return
|
||
|
||
# 对交易结果汇总统计
|
||
for time, result in resultDict.items():
|
||
|
||
if result.pnl > 0:
|
||
self.winningResult += 1
|
||
self.totalWinning += result.pnl
|
||
else:
|
||
self.losingResult += 1
|
||
self.totalLosing += result.pnl
|
||
self.capital += result.pnl
|
||
self.maxCapital = max(self.capital, self.maxCapital)
|
||
#self.maxVolume = max(self.maxVolume, result.volume)
|
||
drawdown = self.capital - self.maxCapital
|
||
drawdownRate = round(float(drawdown*100/self.maxCapital),4)
|
||
|
||
self.pnlList.append(result.pnl)
|
||
self.timeList.append(time)
|
||
self.capitalList.append(self.capital)
|
||
self.drawdownList.append(drawdown)
|
||
self.drawdownRateList.append(drawdownRate)
|
||
|
||
self.totalResult += 1
|
||
self.totalTurnover += result.turnover
|
||
self.totalCommission += result.commission
|
||
self.totalSlippage += result.slippage
|
||
|
||
self.output(u'[{5}],{6} Vol:{0},盈亏:{1},回撤:{2}/{3},权益:{4}'.
|
||
format(abs(result.volume), result.pnl, drawdown,
|
||
drawdownRate, self.capital, result.groupId, time))
|
||
|
||
# 重新计算一次avaliable
|
||
self.avaliable = self.capital - occupyMoney
|
||
self.percent = round(float(occupyMoney * 100 / self.capital), 2)
|
||
|
||
def savingDailyData(self, d, c, m):
|
||
"""保存每日数据"""
|
||
dict = {}
|
||
dict['date'] = d.strftime('%Y/%m/%d')
|
||
dict['capital'] = c
|
||
dict['maxCapital'] = m
|
||
dict['rate'] = c / self.initCapital
|
||
self.dailyList.append(dict)
|
||
|
||
# ----------------------------------------------------------------------
|
||
def calculateBacktestingResult(self):
|
||
"""
|
||
计算回测结果
|
||
Modified by Incense Lee
|
||
增加了支持逐步加仓的计算：
|
||
例如，前面共有6次开仓（1手开仓+5次加仓，每次1手），平仓只有1次（六手）。那么，交易次数是6次（开仓+平仓）。
|
||
暂不支持每次加仓数目不一致的核对（因为比较复杂）
|
||
|
||
增加组合的支持。（组合中，仍然按照1手逐步加仓和多手平仓的方法，即使启用了复利模式，也仍然按照这个规则，只是在计算收益时才乘以系数）
|
||
|
||
增加期初权益，每次交易后的权益，可用资金，仓位比例。
|
||
|
||
"""
|
||
self.output(u'计算回测结果')
|
||
|
||
# 首先基于回测后的成交记录，计算每笔交易的盈亏
|
||
resultDict = OrderedDict() # 交易结果记录
|
||
longTrade = [] # 未平仓的多头交易
|
||
shortTrade = [] # 未平仓的空头交易
|
||
|
||
i = 1
|
||
|
||
tradeUnit = 1
|
||
|
||
longid = EMPTY_STRING
|
||
shortid = EMPTY_STRING
|
||
|
||
for tradeid in self.tradeDict.keys():
|
||
|
||
trade = self.tradeDict[tradeid]
|
||
|
||
# 多头交易
|
||
if trade.direction == DIRECTION_LONG:
|
||
# 如果尚无空头交易
|
||
if not shortTrade:
|
||
longTrade.append(trade)
|
||
longid = tradeid
|
||
# 当前多头交易为平空
|
||
else:
|
||
gId = i # 交易组（多个平仓数为一组）
|
||
gt = 1 # 组合的交易次数
|
||
gr = None # 组合的交易结果
|
||
|
||
if trade.volume >tradeUnit:
|
||
self.writeCtaLog(u'平仓数{0},组合编号:{1}'.format(trade.volume,gId))
|
||
gt = int(trade.volume/tradeUnit)
|
||
|
||
for tv in range(gt):
|
||
|
||
entryTrade = shortTrade.pop(0)
|
||
|
||
result = TradingResult(entryPrice=entryTrade.price,
|
||
entryDt=entryTrade.dt,
|
||
exitPrice=trade.price,
|
||
exitDt=trade.dt,
|
||
volume=-tradeUnit,
|
||
rate=self.rate,
|
||
slippage=self.slippage,
|
||
size=self.size,
|
||
groupId=gId,
|
||
fixcommission=self.fixCommission)
|
||
|
||
|
||
if tv == 0:
|
||
if gt == 1:
|
||
resultDict[entryTrade.dt] = result
|
||
else:
|
||
gr = copy.deepcopy(result)
|
||
else:
|
||
gr.turnover = gr.turnover + result.turnover
|
||
gr.commission = gr.commission + result.commission
|
||
gr.slippage = gr.slippage + result.slippage
|
||
gr.pnl = gr.pnl + result.pnl
|
||
|
||
if tv == gt -1:
|
||
gr.volume = trade.volume
|
||
resultDict[entryTrade.dt] = gr
|
||
|
||
t = {}
|
||
t['OpenTime'] = entryTrade.tradeTime.strftime('%Y/%m/%d %H:%M:%S')
|
||
t['OpenPrice'] = entryTrade.price
|
||
t['Direction'] = u'Short'
|
||
t['CloseTime'] = trade.tradeTime.strftime('%Y/%m/%d %H:%M:%S')
|
||
t['ClosePrice'] = trade.price
|
||
t['Volume'] = tradeUnit
|
||
t['Profit'] = result.pnl
|
||
self.exportTradeList.append(t)
|
||
|
||
self.writeCtaLog(u'{9}@{6} [{7}:开空{0},short:{1}]-[{8}:平空{2},cover:{3},vol:{4}],净盈亏:{5}'
|
||
.format(entryTrade.tradeTime, entryTrade.price,
|
||
trade.tradeTime, trade.price, tradeUnit, result.pnl,
|
||
i, shortid, tradeid,gId))
|
||
i = i+1
|
||
|
||
if type(gr) != type(None):
|
||
self.writeCtaLog(u'组合净盈亏:{0}'.format(gr.pnl))
|
||
|
||
self.writeCtaLog(u'-------------')
|
||
|
||
# 空头交易
|
||
else:
|
||
# 如果尚无多头交易
|
||
if not longTrade:
|
||
shortTrade.append(trade)
|
||
shortid = tradeid
|
||
# 当前空头交易为平多
|
||
else:
|
||
gId = i # 交易组（多个平仓数为一组）
|
||
gt = 1 # 组合的交易次数
|
||
gr = None # 组合的交易结果
|
||
|
||
if trade.volume >tradeUnit:
|
||
self.writeCtaLog(u'平仓数{0},组合编号:{1}'.format(trade.volume,gId))
|
||
gt = int(trade.volume/tradeUnit)
|
||
|
||
for tv in range(gt):
|
||
|
||
entryTrade = longTrade.pop(0)
|
||
|
||
result = TradingResult(entryPrice=entryTrade.price,
|
||
entryDt=entryTrade.dt,
|
||
exitPrice=trade.price,
|
||
exitDt=trade.dt,
|
||
volume=tradeUnit,
|
||
rate=self.rate,
|
||
slippage=self.slippage,
|
||
size=self.size,
|
||
groupId=gId,
|
||
fixcommission=self.fixCommission)
|
||
if tv == 0:
|
||
if gt==1:
|
||
resultDict[entryTrade.dt] = result
|
||
else:
|
||
gr = copy.deepcopy(result)
|
||
else:
|
||
gr.turnover = gr.turnover + result.turnover
|
||
gr.commission = gr.commission + result.commission
|
||
gr.slippage = gr.slippage + result.slippage
|
||
gr.pnl = gr.pnl + result.pnl
|
||
|
||
if tv == gt -1:
|
||
gr.volume = trade.volume
|
||
resultDict[entryTrade.dt] = gr
|
||
|
||
t = {}
|
||
t['OpenTime'] = entryTrade.tradeTime.strftime('%Y/%m/%d %H:%M:%S')
|
||
t['OpenPrice'] = entryTrade.price
|
||
t['Direction'] = u'Long'
|
||
t['CloseTime'] = trade.tradeTime.strftime('%Y/%m/%d %H:%M:%S')
|
||
t['ClosePrice'] = trade.price
|
||
t['Volume'] = tradeUnit
|
||
t['Profit'] = result.pnl
|
||
self.exportTradeList.append(t)
|
||
|
||
|
||
self.writeCtaLog(u'{9}@{6} [{7}:开多{0},buy:{1}]-[{8}.平多{2},sell:{3},vol:{4}],净盈亏：{5}'
|
||
.format(entryTrade.tradeTime, entryTrade.price,
|
||
trade.tradeTime,trade.price, tradeUnit, result.pnl,
|
||
i, longid, tradeid, gId))
|
||
i = i+1
|
||
|
||
if type(gr) != type(None):
|
||
self.writeCtaLog(u'组合净盈亏:{0}'.format(gr.pnl))
|
||
|
||
self.writeCtaLog(u'-------------')
|
||
|
||
# 检查是否有交易
|
||
if not resultDict:
|
||
self.output(u'无交易结果')
|
||
return {}
|
||
|
||
# 然后基于每笔交易的结果，我们可以计算具体的盈亏曲线和最大回撤等
|
||
|
||
"""
|
||
initCapital = 40000 # 期初资金
|
||
capital = initCapital # 资金
|
||
maxCapital = initCapital # 资金最高净值
|
||
|
||
maxPnl = 0 # 最高盈利
|
||
minPnl = 0 # 最大亏损
|
||
maxVolume = 1 # 最大仓位数
|
||
|
||
wins = 0
|
||
|
||
totalResult = 0 # 总成交数量
|
||
totalTurnover = 0 # 总成交金额（合约面值）
|
||
totalCommission = 0 # 总手续费
|
||
totalSlippage = 0 # 总滑点
|
||
|
||
timeList = [] # 时间序列
|
||
pnlList = [] # 每笔盈亏序列
|
||
capitalList = [] # 盈亏汇总的时间序列
|
||
drawdownList = [] # 回撤的时间序列
|
||
drawdownRateList = [] # 最大回撤比例的时间序列
|
||
"""
|
||
drawdown = 0 # 回撤
|
||
compounding = 1 # 简单的复利基数（如果资金是期初资金的x倍，就扩大开仓比例,例如3w开1手，6w开2手，12w开4手)
|
||
|
||
for time, result in resultDict.items():
|
||
|
||
# 是否使用简单复利
|
||
if self.usageCompounding:
|
||
compounding = int(self.capital/self.initCapital)
|
||
|
||
if result.pnl > 0:
|
||
self.winningResult += 1
|
||
self.totalWinning += result.pnl
|
||
else:
|
||
self.losingResult += 1
|
||
self.totalLosing += result.pnl
|
||
|
||
self.capital += result.pnl*compounding
|
||
self.maxCapital = max(self.capital, self.maxCapital)
|
||
self.maxVolume = max(self.maxVolume, result.volume*compounding)
|
||
drawdown = self.capital - self.maxCapital
|
||
drawdownRate = round(float(drawdown*100/self.maxCapital),4)
|
||
|
||
self.pnlList.append(result.pnl*compounding)
|
||
self.timeList.append(time)
|
||
self.capitalList.append(self.capningital)
|
||
self.drawdownList.append(drawdown)
|
||
self.drawdownRateList.append(drawdownRate)
|
||
|
||
self.totalResult += 1
|
||
self.totalTurnover += result.turnover*compounding
|
||
self.totalCommission += result.commission*compounding
|
||
self.totalSlippage += result.slippage*compounding
|
||
|
||
# ---------------------------------------------------------------------
|
||
def exportTradeResult(self):
|
||
"""到处回测结果表"""
|
||
if not self.exportTradeList:
|
||
return
|
||
csvOutputFile = os.path.abspath(os.path.join(os.path.dirname(__file__), 'TestLogs',
|
||
'TradeList_{0}.csv'.format(datetime.now().strftime('%Y%m%d_%H%M'))))
|
||
|
||
import csv
|
||
csvWriteFile = file(csvOutputFile, 'wb')
|
||
fieldnames = ['vtSymbol','OpenTime', 'OpenPrice', 'Direction', 'CloseTime', 'ClosePrice', 'Volume', 'Profit']
|
||
writer = csv.DictWriter(f=csvWriteFile, fieldnames=fieldnames, dialect='excel')
|
||
writer.writeheader()
|
||
|
||
for row in self.exportTradeList:
|
||
writer.writerow(row)
|
||
|
||
if not self.dailyList:
|
||
return
|
||
|
||
csvOutputFile2 = os.path.abspath(os.path.join(os.path.dirname(__file__), 'TestLogs',
|
||
'DailyList_{0}.csv'.format(datetime.now().strftime('%Y%m%d_%H%M'))))
|
||
|
||
csvWriteFile2 = file(csvOutputFile2, 'wb')
|
||
fieldnames = ['date','capital', 'maxCapital','rate']
|
||
writer2 = csv.DictWriter(f=csvWriteFile2, fieldnames=fieldnames, dialect='excel')
|
||
writer2.writeheader()
|
||
|
||
for row in self.dailyList:
|
||
writer2.writerow(row)
|
||
|
||
|
||
def getResult(self):
|
||
# 返回回测结果
|
||
d = {}
|
||
d['initCapital'] = self.initCapital
|
||
d['capital'] = self.capital - self.initCapital
|
||
d['maxCapital'] = self.maxCapital
|
||
|
||
if len(self.pnlList) == 0:
|
||
return {}
|
||
|
||
d['maxPnl'] = max(self.pnlList)
|
||
d['minPnl'] = min(self.pnlList)
|
||
|
||
d['maxVolume'] = self.maxVolume
|
||
d['totalResult'] = self.totalResult
|
||
d['totalTurnover'] = self.totalTurnover
|
||
d['totalCommission'] = self.totalCommission
|
||
d['totalSlippage'] = self.totalSlippage
|
||
d['timeList'] = self.timeList
|
||
d['pnlList'] = self.pnlList
|
||
d['capitalList'] = self.capitalList
|
||
d['drawdownList'] = self.drawdownList
|
||
d['drawdownRateList'] = self.drawdownRateList
|
||
d['winningRate'] = round(100 * self.winningResult / len(self.pnlList), 4)
|
||
|
||
averageWinning = 0 # 这里把数据都初始化为0
|
||
averageLosing = 0
|
||
profitLossRatio = 0
|
||
|
||
if self.winningResult:
|
||
averageWinning = self.totalWinning / self.winningResult # 平均每笔盈利
|
||
if self.losingResult:
|
||
averageLosing = self.totalLosing / self.losingResult # 平均每笔亏损
|
||
if averageLosing:
|
||
profitLossRatio = -averageWinning / averageLosing # 盈亏比
|
||
|
||
d['averageWinning'] = averageWinning
|
||
d['averageLosing'] = averageLosing
|
||
d['profitLossRatio'] = profitLossRatio
|
||
|
||
return d
|
||
|
||
#----------------------------------------------------------------------
|
||
def showBacktestingResult(self):
|
||
"""显示回测结果"""
|
||
if self.calculateMode != self.REALTIME_MODE:
|
||
self.calculateBacktestingResult()
|
||
|
||
d = self.getResult()
|
||
|
||
if len(d)== 0:
|
||
self.output(u'无交易结果')
|
||
return
|
||
|
||
# 导出交易清单
|
||
self.exportTradeResult()
|
||
|
||
# 输出
|
||
self.output('-' * 30)
|
||
self.output(u'第一笔交易：\t%s' % d['timeList'][0])
|
||
self.output(u'最后一笔交易：\t%s' % d['timeList'][-1])
|
||
|
||
self.output(u'总交易次数：\t%s' % formatNumber(d['totalResult']))
|
||
self.output(u'期初资金：\t%s' % formatNumber(d['initCapital']))
|
||
self.output(u'总盈亏：\t%s' % formatNumber(d['capital']))
|
||
self.output(u'资金最高净值：\t%s' % formatNumber(d['maxCapital']))
|
||
|
||
self.output(u'每笔最大盈利：\t%s' % formatNumber(d['maxPnl']))
|
||
self.output(u'每笔最大亏损：\t%s' % formatNumber(d['minPnl']))
|
||
self.output(u'净值最大回撤: \t%s' % formatNumber(min(d['drawdownList'])))
|
||
self.output(u'净值最大回撤率: \t%s' % formatNumber(min(d['drawdownRateList'])))
|
||
self.output(u'胜率：\t%s' % formatNumber(d['winningRate']))
|
||
|
||
self.output(u'盈利交易平均值\t%s' % formatNumber(d['averageWinning']))
|
||
self.output(u'亏损交易平均值\t%s' % formatNumber(d['averageLosing']))
|
||
self.output(u'盈亏比：\t%s' % formatNumber(d['profitLossRatio']))
|
||
|
||
self.output(u'最大持仓：\t%s' % formatNumber(d['maxVolume']))
|
||
|
||
self.output(u'平均每笔盈利：\t%s' %formatNumber(d['capital']/d['totalResult']))
|
||
|
||
self.output(u'平均每笔滑点成本：\t%s' %formatNumber(d['totalSlippage']/d['totalResult']))
|
||
self.output(u'平均每笔佣金：\t%s' %formatNumber(d['totalCommission']/d['totalResult']))
|
||
|
||
# 绘图
|
||
import matplotlib.pyplot as plt
|
||
import numpy as np
|
||
|
||
try:
|
||
import seaborn as sns # 如果安装了seaborn则设置为白色风格
|
||
sns.set_style('whitegrid')
|
||
except ImportError:
|
||
pass
|
||
|
||
pCapital = plt.subplot(4, 1, 1)
|
||
pCapital.set_ylabel("capital")
|
||
pCapital.plot(d['capitalList'], color='r', lw=0.8)
|
||
|
||
pDD = plt.subplot(4, 1, 2)
|
||
pDD.set_ylabel("DD")
|
||
pDD.bar(range(len(d['drawdownList'])), d['drawdownList'], color='g')
|
||
|
||
pPnl = plt.subplot(4, 1, 3)
|
||
pPnl.set_ylabel("pnl")
|
||
pPnl.hist(d['pnlList'], bins=50, color='c')
|
||
|
||
"""
|
||
pPos = plt.subplot(4, 1, 4)
|
||
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 = 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.tight_layout()
|
||
#plt.xticks(xindex, tradeTimeIndex, rotation=30) # 旋转15
|
||
|
||
plt.show()
|
||
|
||
#----------------------------------------------------------------------
|
||
def putStrategyEvent(self, name):
|
||
"""发送策略更新事件，回测中忽略"""
|
||
pass
|
||
|
||
#----------------------------------------------------------------------
|
||
def runOptimization(self, strategyClass, optimizationSetting):
|
||
"""优化参数"""
|
||
# 获取优化设置
|
||
settingList = optimizationSetting.generateSetting()
|
||
targetName = optimizationSetting.optimizeTarget
|
||
|
||
# 检查参数设置问题
|
||
if not settingList or not targetName:
|
||
self.output(u'优化设置有问题，请检查')
|
||
|
||
# 遍历优化
|
||
resultList = []
|
||
for setting in settingList:
|
||
self.clearBacktestingResult()
|
||
self.output('-' * 30)
|
||
self.output('setting: %s' %str(setting))
|
||
self.initStrategy(strategyClass, setting)
|
||
self.runBacktesting()
|
||
d = self.calculateBacktestingResult()
|
||
try:
|
||
targetValue = d[targetName]
|
||
except KeyError:
|
||
targetValue = 0
|
||
resultList.append(([str(setting)], targetValue))
|
||
|
||
# 显示结果
|
||
resultList.sort(reverse=True, key=lambda result:result[1])
|
||
self.output('-' * 30)
|
||
self.output(u'优化结果：')
|
||
for result in resultList:
|
||
self.output(u'%s: %s' %(result[0], result[1]))
|
||
return result
|
||
|
||
#----------------------------------------------------------------------
|
||
def clearBacktestingResult(self):
|
||
"""清空之前回测的结果"""
|
||
# 清空限价单相关
|
||
self.limitOrderCount = 0
|
||
self.limitOrderDict.clear()
|
||
self.workingLimitOrderDict.clear()
|
||
|
||
# 清空停止单相关
|
||
self.stopOrderCount = 0
|
||
self.stopOrderDict.clear()
|
||
self.workingStopOrderDict.clear()
|
||
|
||
# 清空成交相关
|
||
self.tradeCount = 0
|
||
self.tradeDict.clear()
|
||
|
||
#----------------------------------------------------------------------
|
||
def runParallelOptimization(self, strategyClass, optimizationSetting):
|
||
"""并行优化参数"""
|
||
# 获取优化设置
|
||
settingList = optimizationSetting.generateSetting()
|
||
targetName = optimizationSetting.optimizeTarget
|
||
|
||
# 检查参数设置问题
|
||
if not settingList or not targetName:
|
||
self.output(u'优化设置有问题，请检查')
|
||
|
||
# 多进程优化，启动一个对应CPU核心数量的进程池
|
||
pool = multiprocessing.Pool(multiprocessing.cpu_count())
|
||
l = []
|
||
|
||
for setting in settingList:
|
||
l.append(pool.apply_async(optimize, (strategyClass, setting,
|
||
targetName, self.mode,
|
||
self.startDate, self.initDays, self.endDate,
|
||
self.slippage, self.rate, self.size,
|
||
self.dbName, self.symbol)))
|
||
pool.close()
|
||
pool.join()
|
||
|
||
# 显示结果
|
||
resultList = [res.get() for res in l]
|
||
resultList.sort(reverse=True, key=lambda result:result[1])
|
||
self.output('-' * 30)
|
||
self.output(u'优化结果：')
|
||
for result in resultList:
|
||
self.output(u'%s: %s' %(result[0], result[1]))
|
||
|
||
#----------------------------------------------------------------------
|
||
def roundToPriceTick(self, price):
|
||
"""取整价格到合约最小价格变动"""
|
||
if not self.priceTick:
|
||
return price
|
||
|
||
newPrice = round(price/self.priceTick, 0) * self.priceTick
|
||
return newPrice
|
||
|
||
|
||
|
||
########################################################################
|
||
class TradingResult(object):
|
||
"""每笔交易的结果"""
|
||
|
||
#----------------------------------------------------------------------
|
||
def __init__(self, entryPrice,entryDt, exitPrice,exitDt,volume, rate, slippage, size, groupId, fixcommission=EMPTY_FLOAT):
|
||
"""Constructor"""
|
||
self.entryPrice = entryPrice # 开仓价格
|
||
self.exitPrice = exitPrice # 平仓价格
|
||
|
||
self.entryDt = entryDt # 开仓时间datetime
|
||
self.exitDt = exitDt # 平仓时间
|
||
|
||
self.volume = volume # 交易数量（+/-代表方向）
|
||
self.groupId = groupId # 主交易ID（针对多手平仓）
|
||
|
||
self.turnover = (self.entryPrice + self.exitPrice) * size * abs(volume) # 成交金额
|
||
if fixcommission:
|
||
self.commission = fixcommission * self.volume
|
||
else:
|
||
self.commission = self.turnover * rate # 手续费成本
|
||
self.slippage = slippage * 2 * size * abs(volume) # 滑点成本
|
||
self.pnl = ((self.exitPrice - self.entryPrice) * volume * size
|
||
- self.commission - self.slippage) # 净盈亏
|
||
|
||
|
||
########################################################################
|
||
class OptimizationSetting(object):
|
||
"""优化设置"""
|
||
|
||
#----------------------------------------------------------------------
|
||
def __init__(self):
|
||
"""Constructor"""
|
||
self.paramDict = OrderedDict()
|
||
|
||
self.optimizeTarget = '' # 优化目标字段
|
||
|
||
#----------------------------------------------------------------------
|
||
def addParameter(self, name, start, end=None, step=None):
|
||
"""增加优化参数"""
|
||
if end is None and step is None:
|
||
self.paramDict[name] = [start]
|
||
return
|
||
|
||
if end < start:
|
||
print u'参数起始点必须不大于终止点'
|
||
return
|
||
|
||
if step <= 0:
|
||
print u'参数布进必须大于0'
|
||
return
|
||
|
||
l = []
|
||
param = start
|
||
|
||
while param <= end:
|
||
l.append(param)
|
||
param += step
|
||
|
||
self.paramDict[name] = l
|
||
|
||
#----------------------------------------------------------------------
|
||
def generateSetting(self):
|
||
"""生成优化参数组合"""
|
||
# 参数名的列表
|
||
nameList = self.paramDict.keys()
|
||
paramList = self.paramDict.values()
|
||
|
||
# 使用迭代工具生产参数对组合
|
||
productList = list(product(*paramList))
|
||
|
||
# 把参数对组合打包到一个个字典组成的列表中
|
||
settingList = []
|
||
for p in productList:
|
||
d = dict(zip(nameList, p))
|
||
settingList.append(d)
|
||
|
||
return settingList
|
||
|
||
#----------------------------------------------------------------------
|
||
def setOptimizeTarget(self, target):
|
||
"""设置优化目标字段"""
|
||
self.optimizeTarget = target
|
||
|
||
|
||
#----------------------------------------------------------------------
|
||
def formatNumber(n):
|
||
"""格式化数字到字符串"""
|
||
rn = round(n, 2) # 保留两位小数
|
||
return format(rn, ',') # 加上千分符
|
||
|
||
|
||
#----------------------------------------------------------------------
|
||
def optimize(strategyClass, setting, targetName,
|
||
mode, startDate, initDays, endDate,
|
||
slippage, rate, size,
|
||
dbName, symbol):
|
||
"""多进程优化时跑在每个进程中运行的函数"""
|
||
engine = BacktestingEngine()
|
||
engine.setBacktestingMode(mode)
|
||
engine.setStartDate(startDate, initDays)
|
||
engine.setEndDate(endDate)
|
||
engine.setSlippage(slippage)
|
||
engine.setRate(rate)
|
||
engine.setSize(size)
|
||
engine.setDatabase(dbName, symbol)
|
||
|
||
engine.initStrategy(strategyClass, setting)
|
||
engine.runBacktesting()
|
||
d = engine.calculateBacktestingResult()
|
||
try:
|
||
targetValue = d[targetName]
|
||
except KeyError:
|
||
targetValue = 0
|
||
return (str(setting), targetValue)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
# 以下内容是一段回测脚本的演示，用户可以根据自己的需求修改
|
||
# 建议使用ipython notebook或者spyder来做回测
|
||
# 同样可以在命令模式下进行回测（一行一行输入运行）
|
||
from strategy.strategyEmaDemo import *
|
||
|
||
# 创建回测引擎
|
||
engine = BacktestingEngine()
|
||
|
||
# 设置引擎的回测模式为K线
|
||
engine.setBacktestingMode(engine.BAR_MODE)
|
||
|
||
# 设置回测用的数据起始日期
|
||
engine.setStartDate('20110101')
|
||
|
||
# 载入历史数据到引擎中
|
||
engine.setDatabase(MINUTE_DB_NAME, 'IF0000')
|
||
|
||
# 设置产品相关参数
|
||
engine.setSlippage(0.2) # 股指1跳
|
||
engine.setRate(0.3/10000) # 万0.3
|
||
engine.setSize(300) # 股指合约大小
|
||
|
||
# 在引擎中创建策略对象
|
||
engine.initStrategy(EmaDemoStrategy, {})
|
||
|
||
# 开始跑回测
|
||
engine.runBacktesting()
|
||
|
||
# 显示回测结果
|
||
# spyder或者ipython notebook中运行时，会弹出盈亏曲线图
|
||
# 直接在cmd中回测则只会打印一些回测数值
|
||
engine.showBacktestingResult()
|
||
|
||
|