多品种回测数据汇总教程

一、背景

由于极智量化想要同一个策略尝试不同品种每次都得运行然后换不同的品种去跑，比较麻烦，干脆直接手搓一个工具来选取同一策略下最好的品种去跑实盘

二、使用教程

1. python包安装

1
2
# 页面上安装python的excel包工具
openpyxl

2. 准备Excel

合约编号定期会变更需要自定替换，只有状态为启用的品种才会跑，每次运行只需清空状态、盈利比率、胜率、盈利次数、亏损次数、交易次数、净利润内的表格数据即可

下载 Excel 表格

3. 准备ExcelUtils.py工具

将以下的ExcelUtils.py工具放到策略根目录，如果根目录有变动自行更改其中的file_path路径变量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
from openpyxl import load_workbook

file_path = "Quant/Strategy/用户策略/合约.xlsx"


class Contract:
    def __init__(self, name, code, is_enabled, status):
        self.name = name
        self.code = code
        self.is_enabled = is_enabled
        self.status = status


'''将数字转化为百分比并保留两位小数'''
def convert_to_percentage(num,include_percent=True):
    if include_percent:
        return "{:.2f}%".format(num * 100)
    else:
        return "{:.2f}".format(num)


'''初始化excel并读取数据 返回未完成的一条合约代码'''
def initExcel():
    workbook = load_workbook(file_path)  # 读取当前目录下的文件
    sheet = workbook.active  # 读取当前活动工作表
    total_rows = sheet.max_row - 1  # 获取总行数（除去表头）
    # 创建对象列表
    contracts = []
    first_contract = None
    for row_index in range(2, sheet.max_row + 1):  # 从第 2 行开始，假设第 1 行是表头
        name = sheet.cell(row_index, 2).value
        code = sheet.cell(row_index, 3).value
        is_enabled = sheet.cell(row_index, 4).value
        status = sheet.cell(row_index, 5).value
        if is_enabled == "启用" and status != "已完成":
            contract = Contract(name, code, is_enabled, status)
            contracts.append(contract)
    if len(contracts) > 0:  # 取出对象第一条数据
        first_contract = contracts[0]
    LogInfo(f"总合约数：{total_rows}，未完成合约数：{len(contracts)}")
    res = first_contract.code if first_contract else ""
    if res == "":
        raise ValueError("暂无可执行合约")
    return res


'''更新excel数据'''
def updateExcel(contract_code, new_data):
    workbook = load_workbook(file_path)
    sheet = workbook.active
    for row_index in range(2, sheet.max_row + 1):  # 忽略i表头进行遍历
        current_contract = sheet.cell(row_index, 3).value  # 合约编号在索引2
        if current_contract == contract_code:
            for col, data in new_data.items():
                sheet.cell(row_index, col).value = data
    workbook.save(file_path)


'''批量赋值计算'''
def bath_calculation():
    # 计算胜率
    Win_percentage = convert_to_percentage((PercentProfit() + NumWinTimes()) / NumAllTimes()) if NumAllTimes() != 0 else "0.00%"
    data = {  # data中未包含的数据列不做修改
        5: "已完成",
        6: convert_to_percentage(PercentProfit()),  # 盈利率
        7: Win_percentage,  # 胜率（保本次数+盈利次数）/总次数
        8: NumWinTimes(),  # 盈利次数
        9: NumLoseTimes(),  # 亏损次数
        10: NumWinTimes() + NumLoseTimes(),  # 总次数=盈利次数+亏损次数
        11: convert_to_percentage(GrossProfit(),False),  # 净利润(总利润-保留两位小数)
    }
    return data


'''回测结束后自动更新excel重启下一个合约'''
def autom_update_restart(contract_code):
    data = bath_calculation()
    updateExcel(contract_code, data)
    if contract_code != "":
        ReloadStrategy()

4. 引用

只需要在initialize()和hisover_callback()阶段里进行调用，handle_data()里正常写回测开平仓逻辑即可

1
2
3
4
5
6
7
8
9
10
11
12
13
import ExcelUtils # 引入excel工具

# 策略开始运行时执行该函数一次，并初始化excel工具
def initialize(context):  
    global contract_code
    SetOrderWay(2)  # 设置K线稳定后发单
    contract_code=ExcelUtils.initExcel() # 初始化中引用
    SetBarInterval(contract_code,'M',5,1000,200)
    
# 历史回测阶段结束时执行该函数一次
def hisover_callback(context):
    ExcelUtils.autom_update_restart(contract_code) # 回测结束时再次引入函数
    pass

5. 例子

以常规的双均线策略为例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
import talib
import ExcelUtils

p1 = 5  # 快周期
p2 = 20 # 慢周期
qty = 1
def initialize(context):  
    global contract_code
    SetOrderWay(2)  # 设置K线稳定后发单
    SetAFunUseForHis()# 设置账户函数可以在历史阶段使用
    # contract_code=Symbol() # 用界面设置的合约
    # contract_code="ZCE|F|SH|605" # 用代码固定的合约
    contract_code=ExcelUtils.initExcel() 
    SetBarInterval(contract_code,'M',5,1000,200)    # 以短周期为主
    
# 历史测回执行具体逻辑
def his_trigger(ma1, ma2):
    if ma1[-1] > ma2[-1] and MarketPosition() <= 0:
        Buy(qty, Close()[-1])
    elif ma1[-1] < ma2[-1] and MarketPosition() >= 0:
        SellShort(qty, Close()[-1])

def handle_data(context):
    if CurrentBar() < p2:
        return;
        
    ma1 = talib.MA(Close(), p1)
    ma2 = talib.MA(Close(), p2)  
    
    his = context.strategyStatus() == 'H'
    if his:
        his_trigger(ma1, ma2) 

# 历史回测阶段结束时执行该函数一次
def hisover_callback(context):
    LogInfo("版本:双均线策略")
    ExcelUtils.autom_update_restart(contract_code)
    pass

7. 效果展示

image-20260211174806606

三、总结

通常是用来快速寻找适合同一策略的不同品种，做数据对比然后用来优化跑实盘