期魔方期魔方
注册
KB / ARTICLE

策略案例 · 自动移仓与订单备份

策略文档 · 期货版修订 R1发布于 2026-10-10

本文给出两个实用案例：主力合约映射切换时的自动移仓换月，以及策略重启后从本地磁盘恢复历史订单继续运行的方案。

案例一：自动移仓换月

场景

对于订阅主连合约进行交易的策略，当主力合约映射发生改变后，可以通过 get_current_main_symbol（获取当前主力合约）函数检测到这种改变，从而进行必要的处理。

需要注意：回测模式下主力连续合约作为一个独立合约存在，不存在移仓换月，不需处理；只在任务模式（实盘自动交易）下才需要这个处理。当基于主连合约（后缀 888）进行交易并发生主力映射改变时，下方案例会自动平掉原主力合约，并在新主力合约开出同样数量的仓位。

策略逻辑

  • 参数开关：auto_hot_change 控制是否启用自动移仓，默认关闭
  • 触发前提：当前合约 ID 中包含 "888"（主连合约）且开关打开
  • on_init 时通过 get_current_main_symbol 记录初始主力合约 context.hot_symbol
  • 每根 K 线收盘比较最新主力合约与 context.hot_symbol，不一致即判定发生换月
  • 移仓动作：先平掉 context.hot_symbol（旧主力）上的多 / 空仓位，再在新主力上开同等数量仓位
  • 移仓完成后把 context.hot_symbol 更新为最新主力；未换月时执行 custom_trade_on_k_line 中的正常交易逻辑
  • 持仓字典 context.positions 由 on_trade 回调按开 / 平仓标志实时维护

完整代码

"""
文件类型: 量化策略
帮助文档: https://jukequant.com/knowledge/category/strategy
期魔方，为您提供专业的量化服务
"""


# 从 pydantic 库中导入 BaseModel 和 Field 类，用于定义参数映射模型
from pydantic import BaseModel, Field


# 定义 K 线交易逻辑的函数，目前函数体为空，可根据实际需求实现具体逻辑
def custom_trade_on_k_line(context):
    # 此处可添加基于 K 线的交易逻辑代码
    pass


# 判断是否在指定合约上持有仓位的函数
def is_holding_on_symbol(context, direction, symbol):
    # 检查指定合约是否在持仓字典中
    if symbol in context.positions:
        # 从持仓字典中获取指定合约的持仓信息
        position = context.positions.get(symbol)
        # 从持仓信息中获取持仓方向
        position_type = position.get("Direction")
        # 比较持仓方向是否与传入的方向一致
        if str(position_type) == str(direction):
            # 如果一致，说明持有该方向的仓位，返回 True
            return True
    # 否则，返回 False，表示未持有该方向的仓位
    return False


# 判断是否需要进行移仓换月操作的函数
def need_deal_hot_change(context):
    # 检查合约 ID 中是否包含 "888" 且自动移仓换月开关是否打开
    if ("888" in context.base_instrument_id) and context.auto_hot_change == True:
        # 获取当前的主力合约代码
        cur_hot = get_current_main_symbol(context.base_instrument_id)
        # 比较当前记录的主力合约代码与最新的主力合约代码是否不同
        if context.hot_symbol != cur_hot:  # 基于主连合约交易，并且发生了主力换月
            # 如果不同，说明需要进行移仓换月操作，返回 True
            return True
    else:
        # 否则，返回 False，表示不需要进行移仓换月操作
        return False


# 关闭原主力合约仓位的函数
def close_position_on_prev_hot(context):
    # 检查原主力合约是否持有多头仓位
    if is_holding_on_symbol(context, "0", context.hot_symbol):
        # 如果持有多头仓位，执行卖出平仓操作
        sell_close(context.volume, context.hot_symbol)
    # 检查原主力合约是否持有空头仓位
    if is_holding_on_symbol(context, "1", context.hot_symbol):
        # 如果持有空头仓位，执行买入平仓操作
        buy_close(context.volume, context.hot_symbol)


# 在新主力合约建立仓位的函数
def open_position_on_new_hot(context):
    # 获取当前的主力合约代码
    cur_hot = get_current_main_symbol(context.base_instrument_id)
    # 检查新主力合约是否未持有多头仓位
    if not is_holding_on_symbol(context, "0", cur_hot):
        # 如果未持有多头仓位，执行买入开仓操作
        buy_open(context.volume, cur_hot)
    # 检查新主力合约是否未持有空头仓位
    if not is_holding_on_symbol(context, "1", cur_hot):
        # 如果未持有空头仓位，执行卖出开仓操作
        sell_open(context.volume, cur_hot)


# 如果原主力合约有仓位，将其移到新主力合约的函数
def move_position(context):
    # 检查原主力合约是否持有空头仓位
    holding_short = is_holding_on_symbol(context, "1", context.hot_symbol)
    # 检查原主力合约是否持有多头仓位
    holding_long = is_holding_on_symbol(context, "0", context.hot_symbol)
    # 如果原主力合约持有空头或多头仓位
    if holding_short or holding_long:
        # 关闭原主力合约的仓位
        close_position_on_prev_hot(context)
        # 在新主力合约建立仓位
        open_position_on_new_hot(context)


# K 线收盘时的运行逻辑函数
def on_bar(context):
    # 判断是否需要进行移仓换月操作
    if need_deal_hot_change(context):
        # 如果需要，执行移仓操作
        move_position(context)
        # 更新记录最新的主力合约映射
        context.hot_symbol = get_current_main_symbol(context.base_instrument_id)
    else:
        # 如果不需要，执行基于 K 线的交易逻辑
        custom_trade_on_k_line(context)


############################## 主程序部分 ##############################


# 定义一个参数映射模型类
class Params(BaseModel, validate_assignment=True):
    # 定义手数字段，默认值为 1，字段标题为 "手数"
    volume: int = Field(default=1, title="手数")
    # 定义是否自动移仓换月字段，默认值为 False，字段标题为 "是否自动移仓换月"
    auto_hot_change: bool = Field(default=False, title="是否自动移仓换月")


# 初始化一些参数的函数，在策略启动时调用
def on_init(context):
    # 从 BASE_SETTING 中获取基础合约 ID
    context.base_instrument_id = BASE_SETTING.get("Instrument")
    # 从 BASE_SETTING 中获取基础周期
    context.base_period = BASE_SETTING.get("bPeriod")
    # 初始化持仓字典为空
    context.positions = {}
    # 初始化获取当前的主力合约
    context.hot_symbol = get_current_main_symbol(context.base_instrument_id)
    # 初始化获取当前的主力合约
    context.cur_symbol = get_current_main_symbol(context.base_instrument_id)


# 每次有行情进来时调用的函数，用于刷新策略状态
def on_tick(context):
    # 注册 K 线处理回调函数，当 K 线收盘时调用 on_bar 函数
    on_bar_run(on_bar, context)


# 返回委托单信息的函数，注意此处的报单信息是未成交状态
def on_order(context, order):
    # 可根据实际需求处理委托单信息
    pass


# 返回成交单信息并更新仓位的函数
def on_trade(context, trade):
    # 从成交单中获取合约 ID
    InstrumentID = trade.get("InstrumentID")
    # 从成交单中获取开平标志
    OffsetFlag = trade.get("OffsetFlag")
    # 如果开平标志为 "0"，表示开仓操作
    if OffsetFlag == "0":
        # 将成交单信息存入持仓字典中
        context.positions[InstrumentID] = trade
    else:
        # 否则，表示平仓操作，从持仓字典中移除该合约的持仓信息
        context.positions.pop(InstrumentID)


# 返回报错信息的函数
def on_error(context, order):
    # 可根据实际需求处理报错信息
    pass


# 策略回测中途发生错误或回撤完毕时调用的函数
def on_stop(context):
    # 可根据实际需求处理策略停止时的操作
    pass

关键要点

  1. 主连合约前提：get_current_main_symbol 的入参是带 888 后缀的主连合约 ID（来自 BASE_SETTING），返回当前实际主力合约；need_deal_hot_change 同时要求合约含 888 且 auto_hot_change 打开
  2. 仅任务模式需要：回测模式下主连是独立合约、不存在映射切换，因此该逻辑只需在实盘任务模式启用
  3. 换月判定：on_init 记录的 context.hot_symbol 与每根 K 线查到的最新主力比较，不一致即发生换月；移仓后再把 context.hot_symbol 更新为新主力
  4. 平旧开新一次完成：move_position 中先平旧主力、随即开新主力，不等待平仓成交回报；是否有仓完全以 context.positions 为准，而该字典由 on_trade 回调维护
  5. 开仓防重复：open_position_on_new_hot 在新主力的多 / 空两个方向上分别先检查 is_holding_on_symbol，避免重复开仓
  6. 正常交易入口：未换月时 on_bar 走 custom_trade_on_k_line，实际策略只需在该函数中补充自己的 K 线交易逻辑

案例二：订单备份与恢复

场景

在跑自动交易途中，如果突然关了电脑或客户端，重启后不做处理的情况下，之前的订单和任务就丢了。希望接着之前跑的任务和持仓继续跑，其中一种解决方法是：把订单保存到本地一个指定目录，当策略启动时，从本地磁盘读取历史订单信息加载到内存中，方便基于历史订单继续执行之前的交易逻辑，或执行不同的交易逻辑。

策略逻辑

  1. 策略本体是双均线系统：用 TA-Lib 的 SMA 计算短周期（len1，默认 5）与长周期（len2，默认 20）两条均线
  2. 金叉（短均线上穿长均线）→ 平空并开多；死叉（短均线下穿长均线）→ 平多并开空
  3. 信号以倒数第 3、第 2 根 K 线的均线关系做穿越确认，避免在交叉点反复报单
  4. 每笔成交在 on_trade 中调用 record_order：先按 max_order_num 截断，再把成交单追加到 context.order_list，并以 JSON 全量覆盖写入本地文件
  5. on_init 时调用 restore_orders：文件存在且非空才读取，把历史订单恢复到 context.order_list
  6. on_bar 中先处理平仓信号、再处理开仓信号

完整代码

"""
文件类型: 量化策略
帮助文档: https://jukequant.com/knowledge/category/strategy
期魔方，为您提供专业的量化服务
"""


# 导入 NumPy 库，用于数组操作等
import numpy as np

# 导入 TA-Lib 库，用于技术分析指标的计算
import talib as ta

# 从 pydantic 库中导入 BaseModel 和 Field 类，用于定义参数映射模型
from pydantic import BaseModel, Field


# 判断当前是否持有指定方向的仓位
def isHolding(context, direction):
    # 检查当前交易品种是否存在于持仓信息中
    if context.base_instrument_id in context.positions:
        # 从持仓信息中获取当前交易品种的成交单信息，并赋值给变量 position
        position = context.positions.get(context.base_instrument_id)
        # 从成交单信息中获取仓位方向
        position_type = position.get("Direction")
        # 判断获取到的仓位方向是否与指定的方向一致
        if str(position_type) == str(direction):
            return True
    return False


# 获取 K 线数据
def get_k_line_data(context):
    # 从外置参数中获取均线 1 的周期
    len1 = context.len1
    # 从外置参数中获取均线 2 的周期
    len2 = context.len2
    # 调用 get_kline 函数获取指定交易品种、指定周期的 K 线数据，数据长度为均线周期的最大值
    context.klines = get_kline(context.base_instrument_id, context.base_period, max(len1, len2))
    # 从 K 线数据中获取日期时间数组
    context.dt_array = context.klines.get("datetime")
    # 从 K 线数据中获取收盘价数组
    context.close_array = context.klines.get("close")


# 计算均线指标数据
def calc_sma(context):
    # 从外置参数中获取均线 1 的周期
    len1 = context.len1
    # 从外置参数中获取均线 2 的周期
    len2 = context.len2
    # 从 K 线数据中获取收盘价数组
    context.close_array = context.klines.get("close")
    # 使用 TA-Lib 库的 SMA 函数计算收盘价数组的均线 1 数据
    ma1_array = ta.SMA(np.asarray(context.close_array), len1)
    # 使用 TA-Lib 库的 SMA 函数计算收盘价数组的均线 2 数据
    ma2_array = ta.SMA(np.asarray(context.close_array), len2)
    # 将均线 1 数组元素循环左移一位
    context.ma1_array = np.roll(context.ma1_array, -1)
    # 将均线 2 数组元素循环左移一位
    context.ma2_array = np.roll(context.ma2_array, -1)
    # 将最新计算的均线1值赋值给均线 1 数组的最后一个元素
    context.ma1_array[-1] = ma1_array[-1]
    # 将最新计算的均线2值赋值给均线 2 数组的最后一个元素
    context.ma2_array[-1] = ma2_array[-1]


# 根据 SMA 指标产生开仓信号
def get_open_signal_by_ma(context):
    # 检查均线 1 数组和均线 2 数组的长度是否小于 3，如果小于 3 则说明数据不足
    if len(context.ma2_array) < 3 or len(context.ma1_array) < 3:
        return 0  # 数值不足，返回信号值 0 表示无开仓信号
    # 初始化开仓信号值为 0，表示无开仓信号
    sig_open = 0
    # 判断均线 1 是否在倒数第三个数据点小于等于均线 2，且在倒数第二个数据点大于均线 2
    if context.ma1_array[-3] <= context.ma2_array[-3] and context.ma1_array[-2] > context.ma2_array[-2]:
        sig_open = 1  # 满足条件则表示产生开多信号，将信号值设为 1
    # 判断均线1是否在倒数第三个数据点大于等于均线 2，且在倒数第二个数据点小于均线 2
    elif context.ma1_array[-3] >= context.ma2_array[-3] and context.ma1_array[-2] < context.ma2_array[-2]:
        sig_open = -1  # 满足条件则表示产生开空信号，将信号值设为 -1
    return sig_open


# 根据指标计算平仓信号
def get_close_signal_by_ma(context):
    # 检查均线 1 数组和均线 2 数组的长度是否小于 3，如果小于 3 则说明数据不足
    if len(context.ma2_array) < 3 or len(context.ma1_array) < 3:
        return 0  # 数值不足，返回信号值 0 表示无平仓信号
    # 初始化平仓信号值为 0，表示无平仓信号
    sig_close = 0
    # 判断均线 1 是否在倒数第三个数据点小于等于均线 2，且在倒数第二个数据点大于均线 2
    if context.ma1_array[-3] <= context.ma2_array[-3] and context.ma1_array[-2] > context.ma2_array[-2]:
        sig_close = 1  # 满足条件则表示产生平空信号，将信号值设为 1
    # 判断均线 1 是否在倒数第三个数据点大于等于均线 2，且在倒数第二个数据点小于均线 2
    elif context.ma1_array[-3] >= context.ma2_array[-3] and context.ma1_array[-2] < context.ma2_array[-2]:
        sig_close = -1  # 满足条件则表示产生平多信号，将信号值设为 -1
    return sig_close


# 根据信号开仓
def open_position_by_signal(context, sig_open):
    if sig_open == 1:  # 开多信号
        # 检查是否没有持有多仓
        if not isHolding(context, "0"):
            # 调用 buy_open 函数进行开多操作
            ret = buy_open(context.volume, context.base_instrument_id)
    elif sig_open == -1:  # 开空信号
        # 检查是否没有持有空仓
        if not isHolding(context, "1"):
            # 调用 sell_open 函数进行开空操作
            sell_open(context.volume, context.base_instrument_id)


# 根据信号平仓
def close_position_by_signal(context, sig_close):
    if sig_close == 1:  # 平空信号
        # 检查是否持有空仓
        if isHolding(context, "1"):
            # 调用 buy_close 函数进行平空操作
            buy_close(context.volume, context.base_instrument_id)
    elif sig_close == -1:  # 平多信号
        # 检查是否持有多仓
        if isHolding(context, "0"):
            # 调用 sell_close 函数进行平多操作
            sell_close(context.volume, context.base_instrument_id)


#  K 线收盘运行逻辑
def on_bar(context):
    # 调用 get_k_line_data 函数获取 K 线数据
    get_k_line_data(context)
    # 调用 calc_sma 函数计算均线指标数据
    calc_sma(context)
    # 调用 get_open_signal_by_ma 函数根据 SMA 指标产生开仓信号
    sig_open = get_open_signal_by_ma(context)
    # 调用 get_close_signal_by_ma 函数根据 SMA 指标产生平仓信号
    sig_close = get_close_signal_by_ma(context)
    # 调用 close_position_by_signal 函数根据平仓信号进行平仓操作
    close_position_by_signal(context, sig_close)
    # 调用 open_position_by_signal 函数根据开仓信号进行开仓操作
    open_position_by_signal(context, sig_open)


# 记录订单到磁盘文件
def record_order(context, order):
    # 导入 json 库，用于处理 JSON 数据
    import json

    # 检查订单列表的长度是否达到最大保存订单个数
    if len(context.order_list) >= context.max_order_num:
        # 如果达到最大个数，则截取订单列表，只保留最近的最大个数的订单
        context.order_list = context.order_list[1 - context.max_order_num :]
    # 将新订单添加到订单列表中
    context.order_list.append(order)
    # 以写入模式打开订单保存文件
    with open(context.order_dir, "w") as f:
        # 将订单列表以 JSON 格式写入文件
        json.dump(context.order_list, f)


# 从磁盘文件读取订单到内存
def restore_orders(context):
    # 导入 json 库，用于处理 JSON 数据
    import json

    # 导入 os 库，用于操作系统相关功能，如文件路径操作等
    import os

    # 检查订单保存文件是否存在
    if os.path.exists(context.order_dir):
        # 以只读模式打开订单保存文件
        with open(context.order_dir) as f:
            # 检查文件大小是否不为 0，即文件是否有内容
            if os.path.getsize(context.order_dir) != 0:
                # 从文件中加载 JSON 数据，并将其转换为 Python 字典列表
                order_dicts = json.load(f)
                # 将加载的订单字典列表赋值给 order_list
                context.order_list = order_dicts
                # put_log(f'restored history orders : \n{context.order_list}', 'INFO')


############################## 主程序部分 ##############################


# 定义一个参数映射模型类
class Params(BaseModel, validate_assignment=True):
    """参数映射模型"""

    # 定义均线 1 的周期，默认值为 5
    len1: int = Field(default=5, title="均线 1 周期")
    # 定义均线 2 的周期，默认值为 20
    len2: int = Field(default=20, title="均线 2 周期")
    # 定义交易手数，默认值为1
    volume: int = Field(default=1, title="手数")
    # 定义订单保存地址，默认值为 r"E:\order.json"，注意这里要改成本机地址
    order_dir: str = Field(default=r"E:\order.json", title="订单保存地址")
    # 定义最大保存订单个数，默认值为 10
    max_order_num: int = Field(default=10, title="最大保存订单个数")


# 初始化一些参数
def on_init(context):
    # 从 BASE_SETTING 中获取交易品种 ID
    context.base_instrument_id = BASE_SETTING.get("Instrument")
    # 从 BASE_SETTING 中获取 K 线周期
    context.base_period = BASE_SETTING.get("bPeriod")
    # 初始化持仓信息为空字典
    context.positions = {}
    # 初始化收盘价数组，长度为均线2的周期，元素初始值都为 0
    context.close_array = np.zeros(context.len2)
    # 初始化均线1数组，长度为均线1的周期，元素初始值都为 0
    context.ma1_array = np.zeros(context.len1)
    # 初始化均线2数组，长度为均线2的周期，元素初始值都为 0
    context.ma2_array = np.zeros(context.len2)
    # 初始化订单列表为空列表
    context.order_list = []
    # 调用 restore_orders 函数从磁盘文件读取订单到内存
    restore_orders(context)


# 每次有行情进来，便会调用 on_tick 函数刷新一次
def on_tick(context):
    # 注册 K 线处理回调，当 K 线收盘时调用 on_bar 函数处理 K 线数据
    on_bar_run(on_bar, context)


# 返回委托单信息 注意此处的报单信息是未成交状态
def on_order(context, order):
    pass


# 返回成交单信息 记录仓位变化
def on_trade(context, trade):
    # 从成交单信息中获取交易品种ID
    InstrumentID = trade.get("InstrumentID")
    # 从成交单信息中获取开平标志
    OffsetFlag = trade.get("OffsetFlag")
    # 判断开平标志是否为开仓标志 "0"
    if OffsetFlag == "0":
        # 如果是开仓，则将成交单信息添加到持仓信息中
        context.positions[InstrumentID] = trade
    else:
        # 如果是平仓，则从持仓信息中移除该交易品种的持仓信息
        context.positions.pop(InstrumentID)
    # 调用 record_order 函数将成交单信息记录到磁盘文件
    record_order(context, trade)


# 返回报错
def on_error(context, order):
    pass


# 策略回测中途发生错误 or 回撤完毕时调用
def on_stop(context):
    pass

关键要点

  1. 双均线穿越信号：ta.SMA 计算两条均线，金叉开多 / 平空、死叉开空 / 平多；信号比较 [-3] 与 [-2] 两个已收盘位置，要求发生真实穿越才触发
  2. 成交即落盘：on_trade 在更新持仓字典后立即调用 record_order，把整份订单列表以 JSON 覆盖写入 context.order_dir，异常退出前最后一笔成交也不会丢
  3. 截断逻辑：订单数达到 max_order_num 时先截取 order_list[1 - max_order_num:]（保留最近 max_order_num - 1 条），再追加新订单，写入文件后总数恰好不超过上限
  4. 启动恢复：on_init 最后调用 restore_orders，仅当文件存在且大小非 0 时才 json.load，避免文件不存在或空文件导致解析异常
  5. 保存路径需修改：order_dir 默认 E:\order.json，实盘使用前必须改成本机真实可写路径
  6. 先平后开：on_bar 先执行平仓信号再执行开仓信号，同一根 K 线上金叉 / 死叉时先释放反向旧仓再建立新仓
  7. 本地持仓字典：context.positions 完全由 on_trade 按开 / 平仓标志增删，isHolding 据此判断当前品种是否持有指定方向仓位

相关阅读

  • [核心API 账户持仓查询]
  • [核心API 交易指令]
  • [核心API 辅助功能]
  • [案例：日内清仓与套利]
  • [案例：大单拆分]
已复制到剪贴板
策略文档 · 期货版修订 R1

返回策略文档 · 期货版板块

评论