vnpy/examples/CtaBacktesting/util_branch_testing.py

# encoding: UTF-8

"""
批量测试相关方法
# 华富资产 李来佳
"""

import sys, os, platform, gc, copy,multiprocessing
from datetime import datetime
from time import sleep
vnpy_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))

sys.path.append(vnpy_root)

import numpy as np
import pandas as pd
import talib as ta                  # 科学计算库
import statsmodels.api as sm        # 统计库
import matplotlib
import matplotlib.pyplot as plt
import math                         # 数学计算相关
matplotlib.rcParams['figure.figsize'] = (20.0, 10.0)
import  traceback
from vnpy.trader.setup_logger import *
from vnpy.trader.app.ctaStrategy.ctaBacktesting import BacktestingEngine, OptimizationSetting, MINUTE_DB_NAME


# 合并回测结果
def combine_results(results_list):
    result_df = None

    # 判断结果集是否有数据
    if len(results_list) < 1:
        print('no records')
        return None

    effected_results = 0

    for dict in results_list:
        # 测试项目
        test_item = dict['test_item']
        # 测试csv文件
        file_name = dict['result_file']

        if not os.path.isfile(file_name):
            continue

        effected_results += 1

        # 读取测试文件
        df = pd.read_csv(file_name)
        # 修正索引为时间日期索引
        df = df.set_index(pd.DatetimeIndex(df['date']))

        if result_df is None:
            # 首个测试结果，将净值字段设置为周期
            result_df = df['rate'].to_frame(name=test_item)
            # 汇总净值
            result_df['rate'] = result_df[test_item]
        else:
            # 增加新的测试结果数据
            result_df[test_item] = df['rate']

            # 汇总净值
            result_df['rate'] = result_df['rate'] + result_df[test_item]

        # 释放内存
        l = [df]
        del df
        del l

    if effected_results > 0:
        # 净值平均
        result_df['avg_rate'] = result_df['rate'] / effected_results
        # 组合净值累加（仍然按照1个策略的总资金，累加各策略的收益）
        result_df['group_rate'] = result_df['rate'] - effected_results + 1

        # 删除累加的rate
        result_df.drop('rate', axis=1, inplace=True)

    return result_df


# 计算最大回撤，单日最大回撤
def calculate_result(result_df, rate_column):
    max_rate = 0
    max_loss = 0
    max_rate_date = None
    max_loss_date = None
    max_loss_info = '-'

    for idx in result_df.index:
        # 当前日净值
        cur_rate = result_df[rate_column].loc[idx]

        if cur_rate > max_rate:
            max_rate = cur_rate
            max_rate_date = idx.strftime('%Y-%m-%d')

        cur_loss = max_rate - cur_rate

        if cur_loss > max_loss:
            max_loss = cur_loss
            max_loss_date = idx.strftime('%Y-%m-%d')
            max_loss_percent = max_loss / max_rate
            max_loss_info = u'{} from {} to {},rate {}=>{},max loss rate {}'.format(rate_column, max_rate_date,
                                                                                    max_loss_date, max_rate, cur_rate,
                                                                                    max_loss_percent)
    return max_loss_info

def single_strategy_test(test_settings):
    """
    根据设置参数，执行回测品种，
    :param strategyClass: 策略类
    :param test_settings: 策略参数设置
            test_settings['bar_file']: 回测的bar csv文件路径
            test_settings['bar_interval']: 回测的Bar csv周期
            test_settings['report_file']: 净值输出报告的保存路径
    :return: 
    """

    from vnpy.trader.vtEvent import EventEngine2
    eventEngine = EventEngine2()
    eventEngine.start()

    # 创建回测引擎
    engine = BacktestingEngine(eventEngine=eventEngine)
    # 设置回测的策略类
    engine.setStrategyName(test_settings['name'])
    # 创建日志
    if 'debug' in test_settings:
        engine.createLogger(debug=test_settings['debug'])
    else:
        engine.createLogger()

        # 设置引擎的回测模式
        if test_settings['mode'] == 'tick':
            engine.setBacktestingMode(engine.TICK_MODE)
        else:
            engine.setBacktestingMode(engine.BAR_MODE)

    strategy_settings = copy.copy(test_settings)
    # 设置回测的策略类
    if 'filenamePrefix' in strategy_settings:
        engine.setStrategyName(strategy_settings["filenamePrefix"])
    else:
        engine.setStrategyName(test_settings['name'])

    if 'is_7x24' in test_settings:
        engine.is_7x24 = test_settings['is_7x24']

    # del strategy_settings['size']
    #del strategy_settings['margin_rate']
    del strategy_settings['initCapital']

    if 'report_file' in strategy_settings:
        del strategy_settings['report_file']

    # 设置回测用的数据起始日期
    if 'start_date' in strategy_settings:
        engine.setStartDate(test_settings['start_date'], initDays = strategy_settings.get('initDays', 10))
        del strategy_settings['start_date']
    else:
        engine.setStartDate('20110101',initDays = strategy_settings.get('initDays',10))

    # 设置回测用的数据结束日期
    if 'end_date' in test_settings:
        engine.setEndDate(test_settings['end_date'])
        del strategy_settings['end_date']
    else:
        engine.setEndDate('20171201')

    # engine.connectMysql()
    engine.setDatabase(dbName=MINUTE_DB_NAME, symbol=test_settings['vtSymbol'])

    # 设置产品相关参数
    if 'slippage' in test_settings and test_settings['slippage'] > 0:
        engine.setSlippage(test_settings['slippage'])
    else:
        engine.setSlippage(0)
    engine.setRate(test_settings['rate'] if 'rate' in test_settings else float(0.0001))  # 万1
    engine.setSize(test_settings['size'])  # 合约大小
    engine.setMinDiff(test_settings['minDiff']) # 合约价格最小跳动
    engine.setMarginRate(test_settings['margin_rate']) # 合约保证金率
    if 'fixCommission' in test_settings:
        engine.fixCommission = float(test_settings['fixCommission'])  # 固定交易费用（每次开平仓收费）

    # 删除本地json文件
    data_path = os.path.abspath(os.path.join(os.getcwd(),'data'))
    if not os.path.exists(data_path):
        os.mkdir(data_path)

    logs_path = os.path.abspath(os.path.join(os.getcwd(), 'logs'))
    if not os.path.exists(logs_path):
        os.mkdir(logs_path)

    up_grid_json_file = os.path.abspath(os.path.join(data_path,'{0}_upGrids.json'.format(test_settings['name'])))
    dn_grid_json_file = os.path.abspath(os.path.join(data_path,'{0}_dnGrids.json'.format(test_settings['name'])))
    grid_json_file = os.path.abspath(os.path.join(data_path,'{0}_Grids.json'.format(test_settings['name'])))
    policy_json_file = os.path.abspath(os.path.join(data_path, '{0}_Policy.json'.format(test_settings['name'])))

    if os.path.isfile(up_grid_json_file):
        print(u'{0} exist,remove it'.format(up_grid_json_file))
        try:
            os.remove(up_grid_json_file)
        except Exception as ex:
            print(u'{0}：{1}'.format(Exception, ex))
            return False

    if os.path.isfile(dn_grid_json_file):
        print(u'{0}exist,remove it'.format(dn_grid_json_file))
        try:
            os.remove(dn_grid_json_file)
        except Exception as ex:
            print(u'{0}：{1}'.format(Exception, ex))
            return False

    if os.path.isfile(grid_json_file):
        print(u'{0}exist,remove it'.format(grid_json_file))
        try:
            os.remove(grid_json_file)
        except Exception as ex:
            print(u'{0}：{1}'.format(Exception, ex))
            return False

    if os.path.isfile(policy_json_file):
        print(u'{0}exist,remove it'.format(policy_json_file))
        try:
            os.remove(policy_json_file)
        except Exception as ex:
            print(u'{0}：{1}'.format(Exception, ex))
            return False

    # 在引擎中创建策略对象
    print(u'run {} using:{}'.format(test_settings['strategy'],strategy_settings))
    engine.initStrategy(test_settings['strategy'], setting=strategy_settings)

    # 设置每日净值的报告文档存储路径
    daily_report_file = 'DailyList.csv' if 'report_file' not in test_settings else test_settings['report_file']
    engine.setDailyReportName(daily_report_file)

    # 使用简单复利模式计算
    engine.usageCompounding = False  # True时，只针对FINAL_MODE有效

    # 启用实时计算净值模式REALTIME_MODE / FINAL_MODE 回测结束时统一计算模式
    engine.calculateMode = engine.REALTIME_MODE
    engine.capital = test_settings['initCapital']  # 设置期初资金
    engine.initCapital = test_settings['initCapital']  # 设置期初资金
    engine.avaliable = test_settings['initCapital']  # 设置期初资金
    engine.netCapital = test_settings['initCapital']
    engine.maxCapital = test_settings['initCapital']  # 设置期初资金
    engine.maxNetCapital = test_settings['initCapital']  # 设置期初资金
    engine.percentLimit = test_settings['percentLimit']  # 设置资金使用上限比例(%)
    engine.barTimeInterval = 60 * test_settings['bar_interval']  # 回测文件中，bar的周期秒数，用于csv文件自动减时间

    try:
        # 前置动作(无参数)
        pre_functions = test_settings.get('pre_functions',[])
        for fun_name in pre_functions:
            try:
                if not isinstance(fun_name,str):
                    continue
                if hasattr(engine.strategy,fun_name):
                    fun = getattr(engine.strategy,fun_name)
                    if fun is not None:
                        fun()
            except Exception as ex:
                print(u'调用前置动作异常:{},{}'.format(str(ex),traceback.format_exc()),file=sys.stderr)

        # 开始跑回测
        if 'bar_file' in test_settings:
            engine.runBackTestingWithBarFile(test_settings['bar_file'])
        else:
            engine.runBackTestingWithDataSource()

        print('{}finished loop bars'.format(test_settings['name']))
        # 显示回测结果
        engine.showBacktestingResult()

        # 保存策略得内部数据
        engine.saveStrategyData()

        print('{} finished'.format(test_settings['name']))

        return True
    except Exception as ex:
        print(u'single_strategy_test exception:{}'.format(str(ex)))
        traceback.print_exc()
        return False

def multi_period_test(gid, group_setting):
    """
    多周期回测品种组合
    1、对group_settings进行分解，分解出各个运行周期的参数设置
    2、逐一周期运行测试
    3、添加回测结果
    :param gid: 测试组名
    :param group_setting：dict，包含参数，多周期清单
     如果多周期，则对每一周期执行回测，并汇总结果。
     :return 回测的每日净值统计文件
    """

    # 回测的分钟周期
    minutes_interval_list = group_setting['minute_list']  # 回测分钟队列（3，5，10等）

    strategyClass = group_setting['strategy']

    # 测试批次时间
    test_dt = datetime.now().strftime('%Y%m%d_%H%M')

    # 回测结果队列，对应测试分钟队列
    daily_results = []
    return_results = []

    # 启动多进程
    pool = multiprocessing.Pool(multiprocessing.cpu_count())
    l = []

    # 逐一分钟级别进行回测
    for m_i in minutes_interval_list:
        settings = copy.copy(group_setting)
        del settings['minute_list']

        settings['name'] = '{}_{}_{}_M{}'.format(gid, strategyClass.className, group_setting['symbol'], m_i)

        # 资金占用比例，根据组合内周期数量，进行平均分配
        settings['percentLimit'] = group_setting['percentLimit']/ len(minutes_interval_list)
        settings['vtSymbol'] = group_setting['symbol']
        settings['MinInterval'] = m_i

        settings['mode'] = 'bar'
        settings['backtesting'] = True
        settings['bar_interval'] = group_setting['bar_interval'] if 'bar_interval' in group_setting else 1
        settings['strategy'] = strategyClass
        # 回测报告文件保存路径: 组合，测试实例名称,测试时间
        daily_report_file = os.path.abspath(os.path.join(group_setting['report_folder'], u'{}_daily_{}.csv'
                                                         .format(settings['name'], test_dt)))

        settings['report_file'] = daily_report_file

        #if rt:
            # 回测报告集登记
        daily_results.append({'test_item': 'M{}'.format(m_i), 'result_file': daily_report_file})

        l.append(pool.apply_async(single_strategy_test, (settings,)))
        #rt = single_strategy_test(test_settings=settings)

        # 执行内存回收
        gc.collect()
        sleep(10)

    result_list = [res.get() for res in l]

    for idx, rt in enumerate(result_list):
        if rt:
            return_results.append(daily_results[idx])

    pool.close()
    pool.join()
    return return_results

def run_multiperiod_test(gid, group_settings):
    """
    运行多周期的组合测试    
    :param gid: 组合名称
    :param group_settings: 
    :return: 
    """
    m = '_'.join(str(e) for e in group_settings['minute_list'])
    if 'report_folder' in group_settings:
        final_file = os.path.abspath(os.path.join(group_settings['report_folder'], '{}_{}_Report_{}.csv'.format(gid, group_settings['symbol'], m)))
    else:
        # 报告所在目录
        report_folder = os.path.abspath(os.path.join(os.getcwd(), 'logs', gid))
        if not os.path.exists(report_folder):
            os.mkdir(report_folder)
        # 汇总报告文件
        final_file = os.path.abspath(
            os.path.join(report_folder, '{}_{}_Report_{}.csv'.format(gid, group_settings['symbol'], m)))
        group_settings['report_folder'] = report_folder

    if not os.path.exists(group_settings['report_folder']):
        os.makedirs(group_settings['report_folder'])

    # 运行回测方法,统计结果
    daily_results = multi_period_test(gid, group_settings)

    # 统计结果
    backtest_df = combine_results(daily_results)

    # 保存汇总记录到文件
    backtest_df.to_csv(final_file)

    # 显示资金曲线汇总
    fig, ax1 = plt.subplots()

    period_columns = [item['test_item'] for item in daily_results]

    fig.patch.set_facecolor('white')
    ax1.plot(backtest_df[period_columns])
    ax1.legend()

    # 释放内存
    l = [backtest_df]
    del backtest_df
    del l

    # 释放内存
    gc.collect()
    print( 'finished run_multiparameter_test')

def multi_parameter_test(gid, settings_list):
    """
    不同参数的回测组合
    :param gid: 组合ID
    :param settings_list: 参数列表 
    :return: 
    """
    # 回测结果队列，对应测试参数队列
    daily_results = []
    return_results = []

    # 每个测试的参数名称
    para_list = [i['paraName'] for i in settings_list]

    if len(para_list) == 0:
        return daily_results

    # 启动多进程
    pool = multiprocessing.Pool(multiprocessing.cpu_count())
    #pool = multiprocessing.Pool(2)
    l = []

    print('multi_parameter_test,total:{}'.format(len(settings_list)))

    for idx, strategy_settings in enumerate(settings_list):
        settings = copy.copy(strategy_settings)
        # 测试时间
        test_dt = datetime.now().strftime('%Y%m%d_%H%M')

        del settings['log_file']
        if 'minute_list' in settings:
            del settings['minute_list']
        settings['vtSymbol'] = settings['symbol']
        settings['mode'] = 'bar'
        settings['backtesting'] = True
        if 'bar_interval' not in settings:
            settings['bar_interval'] = 1

        # 回测报告文件保存路径: 组合，测试实例名称,测试时间
        daily_report_file = os.path.abspath(os.path.join(settings['report_folder'], u'{}_daily_{}.csv'.format(settings['name'], test_dt)))

        settings['report_file'] = daily_report_file

        l.append(pool.apply_async(single_strategy_test, (settings,)))

        # 回测报告集登记
        daily_results.append({'test_item': settings['paraName'], 'result_file': daily_report_file})

        # 执行内存回收
        gc.collect()
        sleep(10)

    result_list = [res.get() for res in l]

    # 返回结果是正确的，才添加到返回列表中
    for idx, rt in enumerate(result_list):
        if rt:
            return_results.append(daily_results[idx])

    pool.close()
    pool.join()

    return return_results

def run_multiparameter_test(gid, settings_list):
    """
    多策略组测试
    :param gid: 
    :param settings_list: 
    :return:     """
    if len(settings_list) == 0:
        raise ReferenceError('Zero settings')

    first_setting = settings_list[0]

    paraNames = '_'.join(i['paraName'] for i in settings_list)
    if 'report_folder' in first_setting:
        report_folder = first_setting['report_folder']
        final_file = os.path.abspath(os.path.join(first_setting['report_folder'],
                                                  '{}_{}_Report_{}.csv'.format(gid, first_setting['symbol'],
                                                                               paraNames)))
    else:
        # 报告所在目录
        report_folder = os.path.abspath(os.path.join(os.getcwd(), 'logs'))
        final_file = os.path.abspath(
            os.path.join(report_folder, '{}_{}_Report_{}.csv'.format(gid, first_setting['symbol'], paraNames)))

    if not os.path.exists(report_folder):
        os.makedirs(report_folder)

    for settings in settings_list:
        settings['report_folder'] = report_folder

    # 运行回测方法,统计结果
    daily_results = multi_parameter_test(gid, settings_list)

    # 统计结果
    #backtest_df = combine_results(daily_results)

    # 保存汇总记录到文件
    #backtest_df.to_csv(final_file)

    # 显示资金曲线汇总
   ##fig, ax = plt.subplots()

    #period_columns = [item['test_item'] for item in daily_results]

    #fig.patch.set_facecolor('white')

    #for column in period_columns:
    #    ax.plot(backtest_df[column], label=column)

    #ax.legend()

    #title = u'{} {}'.format(gid, paraNames)
    #plt.title(title)

    #fig.savefig(u'{}/rate.png'.format(report_folder))

    ## 释放内存
    gc.collect()

    print( 'finished run_multiparameter_test')

def single_func(para):
    logger=setup_logger('MyLog', name='my{}'.format(para))
    if para > 5:
        print( u'more than 5')
        logger.info('More than 5')
        return True
    else:
        print ('less')
        logger.info('Less than 5')
        return False

def multi_func():

    import logging
    # 启动多进程
    pool = multiprocessing.Pool(multiprocessing.cpu_count())

    logger = setup_logger('MyLog')

    logger.info('main process')
    l = []

    for i in range(0,10):
        l.append(pool.apply_async(single_func,(i,)))

    results = [res.get() for res in l]

    pool.close()
    pool.join()