Python API 回测

一、系统概述

本文主要为对接闪策回测系统的终端厂商、策略开发人员以及个人或机构投资者提供帮助指引

• 支持文件单回测和策略回测

• 支持期货和证券的回测（以及支持期货与证券同时进行回测）

• 支持模拟撮合成交模式和概率成交模式

• 支持回测结果落地

二、接口开发说明

2.1 文件列表

用户应关注的文件内容如下：

backtestapi-python/
├── backtest_runtime
│   ├── conf
│   │   └── db_connection.json -- 回测数据库配置
│   ├── lib
│   │   ├── libscbacktest.so -- 闪策回测动态库
│   │   └── ...              -- 其他依赖项
│   ├── pybind11
│   │   ├── scbacktestpy.cpython-38-x86_64-linux-gnu.so
│   │   ├── scbacktestpy.cpython-39-x86_64-linux-gnu.so
│   │   └── ... -- 其他 Python 版本动态库
│   ├── scripts
│   │   ├── generateBacktestConfig.py -- 将用户配置文件转为回测配置文件
│   │   └── ...                       -- 其他脚本
│   └── shared/
├── conf
│   ├── initConfig.json -- 用户配置文件样例
├── src
│   └── Strategy.py -- 策略代码示例
├── README.md
└── run.sh - 运行脚本

2.2 配置文件说明

见 回测配置文件说明

2.3 Demo 示例

2.3.1 开发环境

Linux 开发环境：Python >= 3.8 ( API Demo 在 Ubuntu 18.04 / CentOS 7 环境中经过测试，使用 Python 3.8 运行 )

2.3.2 运行方法

# 以 Ubuntu 系统为例
cd /path/to/backtestapi-python
./run.sh # run.sh 脚本会在当前目录创建一个 workdir 子目录，运行产物均放在 workdir 目录下

2.4 接口描述

Python 接口的核心为 sc.Context 类，该类定义了所有用户行为相关接口。

import scbacktestpy as sc

context = sc.Context("config.json", callback_module)

2.4.1 回测上下文构造和初始化（sc.Context）

• 函数原型： sc.Context(config_path: str, callback_module: module)

• 输入参数:

参数名称	参数说明	用法
config_path	配置文件路径（为完整配置文件，非 initConfig.json）	输入
callback_module	包含回调函数的回调模块或对象	输入

• 返回： sc.Context 回测上下文对象

• 可能异常：

  • RuntimeError: 当 API 创建失败时

• 用法说明： 创建回测上下文对象。在构造时会自动创建底层 API 对象。此时尚未初始化，需要调用 Init() 方法进行初始化。

初始化流程示例：

import scbacktestpy as sc

# Define callback module
class CallbackHandler:
    def OnInit(self, context):
        print("Strategy initialized")

    def OnScSnapshotData(self, context, hdr, msg):
        print("Received market data")

    # ... other callbacks ...

# Create context
callback_module = CallbackHandler()
context = sc.Context("config.json", callback_module)

# Initialize
result = context.Init()
if result != sc.SCOMS_SUCC:  # Check error code
    print("Init failed")
    exit(1)

# Run backtest
context.run()

2.4.2 初始化（Init）

• 函数原型： context.Init() -> sc.SCOMSError

• 输入参数：

参数名称	参数说明	用法
无

• 返回： 返回错误码。sc.SCOMS_SUCC 表示成功，其他值表示失败

• 可能异常：

  • RuntimeError: 当初始化失败时

• 用法说明： 初始化回测系统。该方法会：

  1. 初始化底层 API

  2. 调用 callback_module 中的 OnInit(context) 回调（如果存在）

  3. 为 run() 做准备

必须在调用 run() 之前调用此方法。

2.4.3 启动回测（run）

• 函数原型： context.run() -> None

• 输入参数：

参数名称	参数说明	用法
无

• 返回： 无

• 用法说明： 启动回测主循环。此方法为阻塞式调用，会运行直到回测结束。运行完成后，会调用 callback_module 中的 OnStop(context) 回调（如果存在）。

2.4.4 订阅事件（SubscribeEvent）

• 函数原型： context.SubscribeEvent(messageType: sc.mtype_t) -> sc.SCOMSError

• 输入参数：

参数名称	参数说明	用法	样例
messageType	事件类型枚举值	输入	sc.MSG_OMS_STATUS_CHANGE、sc.MSG_SC_SNAPSHOT_DATA 等

• 返回： 返回错误码。sc.SCOMSError 表示成功，其他值表示失败

• 用法说明： 订阅指定类型的事件。被订阅的事件在回测收到时，会触发对应的回调函数。支持的事件类型定义在 sc.mtype_t 枚举中。

常见订阅如：

context.SubscribeEvent(sc.MSG_OMS_STATUS_CHANGE)   # 订阅订单状态变化
context.SubscribeEvent(sc.MSG_OMS_FILL_NEW)       # 订阅订单成交
context.SubscribeEvent(sc.MSG_SC_SNAPSHOT_DATA)   # 订阅快照行情

2.4.5 取消订阅事件（UnSubscribeEvent）

• 函数原型： context.UnSubscribeEvent(messageType: sc.mtype_t) -> None

• 输入参数：

参数名称	参数说明	用法
messageType	事件类型枚举值	输入

• 返回： 无

• 用法说明： 取消订阅指定类型的事件。取消订阅后，该类型事件的回调将不再被触发。

2.4.6 订阅定时器（SubscribeTimer）

• 函数原型： context.SubscribeTimer(interval: int, startTime: datetime.datetime = timeval_t.earliest, endTime: datetime.datetime = timeval_t.latest) -> int

• 输入参数：

参数名称	参数说明	用法
interval	定时器间隔，单位为毫秒	输入
startTime	定时器开始时间，默认为回测开始时间	输入
endTime	定时器结束时间，默认为回测结束时间	输入

• 返回： 返回定时器 ID（int）。该 ID 将在触发定时器时通过 OnTimer 回调传入

• 用法说明： 注册周期定时器。定时器从 startTime 开始，每 interval 毫秒触发一次，直到 endTime 结束。

timer_id = context.SubscribeTimer(60 * 1000)  # 每分钟触发一次

2.4.7 订阅指定时刻定时器（SubscribeSpecifiedTimer）

• 函数原型： context.SubscribeSpecifiedTimer(specifiedTime: datetime.datetime) -> int

• 输入参数：

参数名称	参数说明	用法
specifiedTime	指定的触发时刻	输入

• 返回： 返回定时器 ID（int）。该 ID 将在触发定时器时通过 OnTimer 回调传入

• 用法说明： 注册在指定时刻触发一次的定时器。

2.4.8 取消定时器（UnSubscribeTimer）

• 函数原型： context.UnSubscribeTimer(id: int) -> None

• 输入参数：

参数名称	参数说明	用法
id	定时器 ID	输入

• 返回： 无

• 用法说明： 取消之前通过 SubscribeTimer 或 SubscribeSpecifiedTimer 注册的定时器。

2.4.9 获取当前时间（GetCurrentTime）

• 函数原型： context.GetCurrentTime() -> datetime.datetime

• 输入参数：

参数名称	参数说明	用法
无

• 返回： 返回 datetime.datetime 对象，表示回测时间线中的当前时间

• 用法说明： 获取回测当前时间。该时间在回测过程中会不断更新。

2.4.10 发布订单（SendOrder）

• 函数原型： context.SendOrder(instr: str | sc.instrument_t, px: float, size: int, dir: int, orderid: int, tif: int = sc.TIF_DAY, oflags: int = sc.OFLAG_NONE, investor: str = "") -> sc.SCOMSError

• 输入参数：

参数名称	参数说明	用法	样例
instr	合约标识符（交易所格式字符串或 sc.instrument_t 对象）	输入	"au2505.SHFE" 或 sc.instrument_t 对象
px	报价（0 表示市价单）	输入	100.5 或 0
size	订单数量	输入	1
dir	订单买卖方向	输入	sc.BUY 或 sc.SELL
orderid	订单 ID	输入	用户自定义的唯一标识符
tif	有效期类型	输入	当前仅支持 sc.TIF_DAY（默认值为 sc.TIF_DAY）
oflags	开平方向	输入	sc.FUTURES_OPEN、sc.FUTURES_CLOSE 等（默认值为 sc.OFLAG_NONE 等于 sc.STOCK_OPEN_CLOSE）
investor	投资者标识符	输入	默认为空字符串，回测中暂未使用

• 返回： 无

• 用法说明： 发布订单。该方法支持两种重载：

  1. 使用交易所格式字符串标识合约：context.SendOrder("au2505.SHFE", px, size, dir, orderid, tif, oflags)

  2. 使用 sc.instrument_t 对象：context.SendOrder(instr_obj, px, size, dir, orderid, tif, oflags)

示例：

result = context.SendOrder(
    instr="au2505.SHFE",
    px=100.5,
    size=1,
    dir=sc.BUY,
    orderid=1,
    tif=sc.TIF_DAY,
    oflags=sc.FUTURES_OPEN,
)
if result != sc.SCOMS_SUCC:
    print(f"SendOrder failed with error code {result}")

2.4.11 撤单（CancelOrder）

• 函数原型： context.CancelOrder(orderid: int, investor: str = "") -> sc.SCOMSError

• 输入参数：

参数名称	参数说明	用法	样例
orderid	要撤销的订单 ID	输入
investor	投资者标识符	输入	默认为空字符串，回测中暂未使用

• 返回： 无

• 用法说明： 撤销之前发布的订单。orderid 应与 SendOrder 中使用的订单 ID 相对应。

result = context.CancelOrder(orderid=1)
if result != sc.SCOMS_SUCC:
    print(f"CancelOrder failed with error code {result}")

2.4.12 合约字符串转换

2.4.12.1 合约转为字符串（InstrToExchStr）

• 函数原型： context.InstrToExchStr(instr: sc.instrument_t) -> str

• 输入参数：

参数名称	参数说明	用法
instr	instrument_t 类型的合约对象	输入

• 返回： 返回交易所格式的合约字符串（如 "au2505.SHFE"）

• 用法说明： 该接口用于将 sc.instrument_t 对象转换为交易所格式的合约字符串标识。 在听取行情时，通常也可以直接读取行情的 instr_str 对象，其本身就是交易所格式的合约字符串，这样操作更为方便。

2.4.12.2 字符串转为合约（ExchIdToInstr）

• 函数原型： context.ExchIdToInstr(exchIdStr: str) -> sc.instrument_t

• 输入参数：

参数名称	参数说明	用法
exchIdStr	交易所格式的合约字符串	输入

• 返回： 返回对应的 sc.instrument_t 合约对象

• 用法说明： 该接口用于将交易所格式的合约字符串转换为 sc.instrument_t 对象。支持带交易所后缀（如 "au2505.SHFE"）或不带后缀（如 "au2505"）的合约字符串

2.4.13 回调函数

回调函数需要在 callback_module 中实现。支持的回调函数如下：

2.4.13.1 策略初始化回调（OnInit）

• 函数原型： def OnInit(context: sc.Context) -> None

• 回调参数：

参数名称	参数说明
context	回测上下文对象

• 用法说明： 在 context.Init() 执行结束时被调用。用户可以在此回调中完成以下操作：

  • 订阅事件

  • 注册定时器

  • 初始化策略状态

2.4.13.2 策略停止回调（OnStop）

• 函数原型： def OnStop(context: sc.Context) -> None

• 回调参数：

参数名称	参数说明
context	回测上下文对象

• 用法说明： 在 context.run() 完成后被调用。用户可以在此回调中执行清理工作。

2.4.13.3 订单状态变化回调（OnOmsStatusChange）

• 函数原型： def OnOmsStatusChange(context: sc.Context, hdr: sc.message_hdr_t, msg: sc.msg_oms_status_change) -> None

• 回调参数：

参数名称	参数说明
context	回测上下文对象
hdr	消息头（sc.message_hdr_t）
msg	订单状态变化消息体（sc.msg_oms_status_change）

• 用法说明： 当订单状态发生变化时（包括新订单确认、部分成交、完全成交、撤单、拒绝等）被触发。需要先调用 context.SubscribeEvent(sc.MSG_OMS_STATUS_CHANGE) 订阅此事件。

2.4.13.4 订单成交回调（OnOmsFillNew）

• 函数原型： def OnOmsFillNew(context: sc.Context, hdr: sc.message_hdr_t, msg: sc.msg_oms_fill_new) -> None

• 回调参数：

参数名称	参数说明
context	回测上下文对象
hdr	消息头（sc.message_hdr_t）
msg	成交消息体（sc.msg_oms_fill_new）

• 用法说明： 当订单产生成交时被触发。需要先调用 context.SubscribeEvent(sc.MSG_OMS_FILL_NEW) 订阅此事件。

2.4.13.5 撤单拒绝回调（OnOmsCancelReject）

• 函数原型： def OnOmsCancelReject(context: sc.Context, hdr: sc.message_hdr_t, msg: sc.msg_oms_cancel_reject) -> None

• 回调参数：

参数名称	参数说明
context	回测上下文对象
hdr	消息头（sc.message_hdr_t）
msg	撤单拒绝消息体（sc.msg_oms_cancel_reject）

• 用法说明： 当撤单被拒绝时被触发。需要先调用 context.SubscribeEvent(sc.MSG_OMS_CANCEL_REJECT) 订阅此事件。

2.4.13.6 快照行情回调（OnScSnapshotData）

• 函数原型： def OnScSnapshotData(context: sc.Context, hdr: sc.message_hdr_t, msg: sc.msg_sc_snapshot_data) -> None

• 回调参数：

参数名称	参数说明
context	回测上下文对象
hdr	消息头（sc.message_hdr_t）
msg	快照行情消息体（sc.msg_sc_snapshot_data）

• 用法说明： 当接收到快照行情时被触发。需要先调用 context.SubscribeEvent(sc.MSG_SC_SNAPSHOT_DATA) 订阅此事件。

2.4.13.7 快照委托队列行情回调（OnScSnapshotDataOlist）

• 函数原型： def OnScSnapshotDataOlist(context: sc.Context, hdr: sc.message_hdr_t, msg: sc.msg_sc_snapshot_data_olist) -> None

• 回调参数：

参数名称	参数说明
context	回测上下文对象
hdr	消息头（sc.message_hdr_t）
msg	快照委托队列消息体（sc.msg_sc_snapshot_data_olist）

• 用法说明： 当接收到快照委托队列行情时被触发。需要先调用 context.SubscribeEvent(sc.MSG_SC_SNAPSHOT_DATA_OLIST) 订阅此事件。

2.4.13.8 逐笔委托回调（OnScMarketTbt）

• 函数原型： def OnScMarketTbt(context: sc.Context, hdr: sc.message_hdr_t, msg: sc.msg_sc_market_tbt) -> None

• 回调参数：

参数名称	参数说明
context	回测上下文对象
hdr	消息头（sc.message_hdr_t）
msg	逐笔委托消息体（sc.msg_sc_market_tbt）

• 用法说明： 当接收到逐笔委托数据时被触发。需要先调用 context.SubscribeEvent(sc.MSG_SC_MARKET_TBT) 订阅此事件。

2.4.13.9 逐笔成交回调（OnScTickTbt）

• 函数原型： def OnScTickTbt(context: sc.Context, hdr: sc.message_hdr_t, msg: sc.msg_sc_tick_tbt) -> None

• 回调参数：

参数名称	参数说明
context	回测上下文对象
hdr	消息头（sc.message_hdr_t）
msg	逐笔成交消息体（sc.msg_sc_tick_tbt）

• 用法说明： 当接收到逐笔成交数据时被触发。需要先调用 context.SubscribeEvent(sc.MSG_SC_TICK_TBT) 订阅此事件。

2.4.13.10 定时器回调（OnTimer）

• 函数原型： def OnTimer(context: sc.Context, timer_id: int) -> None

• 回调参数：

参数名称	参数说明
context	回测上下文对象
timer_id	触发的定时器 ID

• 用法说明： 当定时器被触发时被调用。timer_id 对应 SubscribeTimer 或 SubscribeSpecifiedTimer 返回的定时器 ID。

def OnTimer(context, timer_id):
    if timer_id == 1:
        print("Timer 1 triggered")
    elif timer_id == 2:
        print("Timer 2 triggered")

2.5 数据类型和常量

Python API 通过 scbacktestpy 模块提供了所有必要的数据类型和常量定义，详见 数据结构说明。

2.5.1 全局状态存储（sc.g）

sc.g 是一个全局状态对象，用于在 Python 和底层 C++ 模块之间共享数据。这对于在不同回调函数之间传递状态信息或在回测过程中维护全局变量非常有用。

• 类型： sc.GlobalState

• 用途： 提供一个字典式的全局状态存储，允许用户在整个回测过程中存储和访问全局变量

• 特点：

  • 可以像字典一样存储任意 Python 对象

  • 支持点操作符和中括号访问

  • 跨越回调函数、定时器等不同上下文保持状态

使用示例：

import scbacktestpy as sc

class CallbackHandler:
    def OnInit(self, context):
        """初始化时设置全局变量"""
        sc.g.order_count = 0
        sc.g.trade_count = 0

        context.SubscribeEvent(sc.MSG_SC_SNAPSHOT_DATA)
        context.SubscribeEvent(sc.MSG_OMS_FILL_NEW)

    def OnScSnapshotData(self, context, hdr, msg):
        """行情回调中使用全局变量"""
        # 读取全局变量
        current_count = sc.g.order_count
        print(f"Current order count: {current_count}")

        # 修改全局变量
        sc.g.order_count += 1

    def OnOmsFillNew(self, context, hdr, msg):
        """成交回调中使用全局变量"""
        sc.g.trade_count += 1
        print(f"Total trades: {sc.g.trade_count}")

    def OnStop(self, context):
        """停止时访问全局变量"""
        print(f"Final order count: {sc.g.order_count}")
        print(f"Final trade count: {sc.g.trade_count}")

注意事项：

• 可以存储任何 Python 对象（数字、字符串、列表、字典等）

• 对于复杂的数据结构，建议使用字典或列表存储相关信息

2.6 编写策略

2.6.1 基本结构

编写策略首先需要创建一个回调模块（包含回调函数），然后创建Context对象并传入该回调模块：

import scbacktestpy as sc

# Define callback module
class CallbackHandler:
    def __init__(self):
        self.instrument_str = "au2505.SHFE"
        self.order_id = 1

    def OnInit(self, context):
        """初始化策略 - 订阅事件和设置定时器"""
        # 订阅订单状态变化
        context.SubscribeEvent(sc.MSG_OMS_STATUS_CHANGE)
        # 订阅成交推送
        context.SubscribeEvent(sc.MSG_OMS_FILL_NEW)
        # 订阅快照行情
        context.SubscribeEvent(sc.MSG_SC_SNAPSHOT_DATA)

        # 设置定时器
        trading_timer_id = context.SubscribeTimer(60 * 1000)  # 每分钟触发

    def OnScSnapshotData(self, context, hdr, msg):
        """处理快照行情"""
        # 行情中的 instr_str 字段内容通常为交易所格式的合约字符串，如：au2505.SHFE
        instr_str = msg.instr_str if hasattr(msg, 'instr_str') else ""
        if instr_str != self.instrument_str:
            return

        latest_price = msg.latest_price if hasattr(msg, 'latest_price') else 0.0

        # 实现行情处理逻辑
        print(f"Received market data: {instr_str}, latest price: {latest_price}")

    def OnOmsStatusChange(self, context, hdr, msg):
        """处理订单状态变化"""
        order_id = msg.orderid if hasattr(msg, 'orderid') else 0
        new_status = msg.newstatus if hasattr(msg, 'newstatus') else 0
        reason = msg.reason if hasattr(msg, 'reason') else 0

        # 查找对应的枚举值
        if new_status == sc.ostatus_t.STAT_DONE:
            if reason == sc.oreason_t.R_FILL:
                print(f"Order {order_id} fully filled")
            elif reason == sc.oreason_t.R_CANCEL:
                print(f"Order {order_id} cancelled")

    def OnOmsFillNew(self, context, hdr, msg):
        """处理订单成交"""
        trade_price = msg.px if hasattr(msg, 'px') else 0.0
        fill_size = msg.size if hasattr(msg, 'size') else 0
        print(f"Order filled at price {trade_price} with size {fill_size}")

    def OnTimer(self, context, timer_id):
        """处理定时器事件"""
        print(f"Timer {timer_id} triggered")

    def OnStop(self, context):
        """策略停止回调"""
        print("Backtest stopped")

# Main execution
if __name__ == "__main__":
    callback_handler = CallbackHandler()
    context = sc.Context("config.json", callback_handler)

    # Initialize backtest
    result = context.Init()
    if result != sc.SCOMS_SUCC:
        print(f"Init failed with error code {result}")
        exit(1)

    # Run backtest
    context.run()

2.6.2 发送订单

def send_order(self, context, direction, price=0, quantity=1):
    """发送订单"""
    try:
        result = context.SendOrder(
            instr=self.instrument_str,
            px=price,
            size=quantity,
            dir=direction,
            orderid=self.order_id,
            tif=sc.TIF_DAY,
            oflags=sc.FUTURES_OPEN,
        )
        if result != sc.SCOMS_SUCC:
            print(f"SendOrder failed with error code {result}")
            return False

        self.order_id += 1
        print(f"Order sent: direction={direction}, price={price}, quantity={quantity}")
        return True
    except Exception as e:
        print(f"Failed to send order: {e}")
        return False

2.6.3 发送撤单

def cancel_order(self, context, orderid):
    """发送撤单"""
    try:
        result = context.CancelOrder(orderid=orderid)
        if result != sc.SCOMS_SUCC:
            print(f"CancelOrder failed with error code {result}")
            return False

        print(f"Cancel request sent for order {orderid}")
        return True
    except Exception as e:
        print(f"Failed to cancel order: {e}")
        return False

2.6.4 获取当前时间

def get_current_time(self, context):
    """获取回测当前时间"""
    try:
        current_time = context.GetCurrentTime()
        print(f"Current backtest time: {current_time}")
        return current_time
    except Exception as e:
        print(f"Failed to get current time: {e}")
        return None

2.7 运行策略

2.7.1 命令行运行

# 切换到 backtestapi-python 目录
cd /path/to/backtestapi-python

# 设置环境变量
export PYTHONPATH=$PWD/backtest_runtime/pybind11:$PYTHONPATH
export LD_LIBRARY_PATH=$PWD/backtest_runtime/lib:$LD_LIBRARY_PATH

# 运行策略
python /path/to/your/strategy.py /path/to/your/config.json

2.7.2 使用运行脚本

# 使用提供的运行脚本（推荐）
./run.sh

该脚本会自动设置环境变量并在 workdir 目录中运行策略

三、注意事项

1. 回测系统内部时间管理

见 C++ API 回测系统内部时间管理

2. 异常处理

建议在策略代码中添加适当的异常处理，特别是在初始化、发送订单和处理行情数据时。Python API 可能抛出异常，或返回异常值：

3.1 Context 初始化异常

在创建和初始化 Context 时可能出现的异常：

try:
    callback_handler = CallbackHandler()
    context = sc.Context("config.json", callback_handler)
    result = context.Init()
    if result != sc.SCOMS_SUCC:
        raise RuntimeError(f"Init failed with error code {result}")

except RuntimeError as e:
    print(f"Failed to create context: {e}")
except FileNotFoundError as e:
    print(f"Config file not found: {e}")

常见原因：

• 配置文件路径不存在

• 配置文件格式错误

• 底层 API 初始化失败

• 验证配置文件中配置的相关文件是否真实存在，例如：reference_data.symbol_file_path

3.2 事件发送异常

在发送订单或撤单时可能出现的异常：

try:
    result = context.SendOrder(
        instr="au2505.SHFE",
        px=100.5,
        size=1,
        dir=sc.BUY,
        orderid=1,
        tif=sc.TIF_DAY,
        oflags=sc.FUTURES_OPEN,
    )
    if result != SCOMS_SUCC:
        raise RuntimeError(f"SendOrder failed with error code {result}")

except RuntimeError as e:
    print(f"Failed to publish order: {e}")


try:
    result = context.CancelOrder(orderid=1)
    if result != SCOMS_SUCC:
        raise RuntimeError(f"CancelOrder failed with error code {result}")

except RuntimeError as e:
    print(f"Failed to publish order: {e}")

常见原因：

• 合约代码无效（如 “au2505.SHFE” 中的合约不存在）

• 订单参数不符合业务规则（价格、数量等）

• 交易所不支持该操作

3.3 事件订阅异常

在订阅或取消订阅事件时可能出现的异常：

try:
    result = context.SubscribeEvent(sc.MSG_OMS_STATUS_CHANGE)
    if result != sc.SCOMS_SUCC:
        raise RuntimeError(f"Subscribe event failed with error code {result}")
except Exception as e:
    print(f"Failed to subscribe event: {e}")

常见原因：

• 无效的消息类型

• 系统内部错误

3.4 定时器注册异常

在订阅定时器时可能出现的异常：

try:
    timer_id = context.SubscribeTimer(60 * 1000)
    if timer_id <= 0:
        raise RuntimeError(f"Failed to subscribe timer, got invalid timer id: {timer_id}")
except Exception as e:
    print(f"Failed to subscribe timer: {e}")

常见原因：

• 定时器参数无效

• 系统内部错误

四、常见问题

Q1：为什么收不到某个合约的行情？

• 检查配置文件中的 instruments 列表是否包含该合约

• 确认已正确调用 context.SubscribeEvent() 订阅对应 sc.mtype_t 的行情

• 验证合约代码格式是否正确（如 “au2505.SHFE”）

• 检查对应行情的回调函数是否正确实现

Q2：订单被拒绝的常见原因？

• 合约代码错误或不存在

• 订单参数格式错误

• 合约不在配置的交易时间内

• 使用了不支持的订单类型或标志

Q3：如何调试策略？

• 在回调函数中使用 print() 输出调试信息

• 验证配置文件格式和参数

• 在 try-except 块中捕获异常信息

Q4：定时器没有触发的原因？

• 确认定时器的触发时间是否在回测时间范围内

• 检查定时器的开始和结束时间参数设置是否正确

• 验证定时器ID是否有效（> 0）

Q5：如何处理多合约交易？

可以在回调处理函数中检查消息的 instr_str 字段来判断合约，然后根据不同的合约执行相应的交易逻辑。例如：

def OnScSnapshotData(self, context, hdr, msg):
    instr_str = msg.instr_str if hasattr(msg, 'instr_str') else ""
    if instr_str == "au2505.SHFE":
        # Handle au2505
        pass
    elif instr_str == "ag2505.SHFE":
        # Handle ag2505
        pass

Q6：Context 初始化时出现异常怎么办？

• 检查配置文件路径是否正确

• 确认 SCSymbol.table 文件是否存在

• 验证配置文件 JSON 格式是否正确

• 检查回调模块中是否正确实现了必要的回调方法

• 查看返回的错误码：result = context.Init()，然后查看 SCOMSError 定义

Q7：发送订单时出现异常？

• 检查合约代码格式（如 “au2505.SHFE”）

• 验证订单参数是否正确（价格、数量、方向等）

• 确认枚举值的使用是否正确（如 sc.BUY、sc.TIF_DAY 等）

Q8：回调函数没有被触发？

• 确认已调用 context.SubscribeEvent() 订阅相应的消息类型

• 检查回调函数名称是否正确（如 OnOmsStatusChange、OnScSnapshotData 等）

• 验证回调函数签名是否正确：def CallbackName(self, context, hdr, msg) -> None

• 对于定时器，确认已调用 context.SubscribeTimer() 或 context.SubscribeSpecifiedTimer()

• 检查 OnInit() 是否正确实现，因为订阅操作通常在这里进行

Q9：如何在回测期间获取当前时间？

使用 context.GetCurrentTime() 方法获取回测时间线中的当前时间。该方法在回测过程中会不断更新，随着时间向前推进而改变。

Q10：如何正确传递 callback_module 参数？

callback_module 应该是一个包含回调函数的对象。通常是一个类实例，示例如下：

class MyCallbacks:
    def OnInit(self, context):
        pass

    def OnScSnapshotData(self, context, hdr, msg):
        pass

    # ... other callbacks ...

callbacks = MyCallbacks()
context = sc.Context("config.json", callbacks)
result = context.Init()
if result != sc.SCOMS_SUCC:
    print("Init failed")
    exit(1)
context.run()

或者是一个模块，示例如下：

# CallbackModule 包含需要的回调函数 OnInit OnScSnapshotData 等
import CallbackModule

context = sc.Context("config.json", CallbackModule)
result = context.Init()
if result != sc.SCOMS_SUCC:
    print("Init failed")
    exit(1)
context.run()