策略案例 · 日内清仓与套利

本文给出两个常见的期货策略案例：日内清仓（预设时间段内平掉所有持仓，避免持仓过夜）和跨品种套利（两个相关合约的比值回归交易）。

案例一：日内清仓策略

场景

某些策略只做日内交易，需要在预设的时间段内对所有持仓进行平仓操作，以避免持仓过夜可能带来的风险，或根据特定的交易策略需求结束当日的交易活动。

策略逻辑

• 参数开关：day_clear_mode 控制是否启用日内清盘
• 时间窗口：day_clear_time（开始时间）到 day_over_time（结束时间）
• 每个 tick 检查当前时间是否落在窗口内，落在窗口内即调用清仓函数
• 遍历 get_position() 返回的全部持仓，扣除冻结数量后分别计算今仓、昨仓的可平数量
• 上期所（SHFE）/ 能源中心（INE）区分平今（comboffsetflag="3"）与平昨（comboffsetflag="1"）；其他交易所今昨仓合并按平昨处理
• 平仓单按持仓方向以涨停价 / 跌停价报出，尽量保证成交
• 不做逐笔回报确认：每次触发时重新获取持仓再分配，直到全部持仓分配结束

完整代码

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


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


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

    # 定义日内清盘启动选项，默认值为 False
    day_clear_mode: bool = Field(default=False, title="日内清盘")
    # 定义日内清盘开始时间，默认值为 “14:58”
    day_clear_time: str = Field(default="14:58", title="日内清盘开始时间")
    # 定义日内清盘结束时间，默认值为 “15:00”
    day_over_time: str = Field(default="15:00", title="日内清盘结束时间")


# 定义初始化函数，当程序初始化时会调用该函数
def on_init(context):
    # 打印初始化完成策略开始的提示信息
    print("demo start")


# 定义 tick 事件处理函数，当收到 tick 数据时会调用该函数
def on_tick(context):
    """获取账户信息"""
    # 调用日内清盘函数
    day_over_close(context)


# 定义一个函数，用于判断交易所 ID 是否为上海期货交易所（SHFE）或上海国际能源交易中心（INE）
def isSHFEorINE(context, exchangeid):
    # 判断 exchangeid 是否在字符串"SHFE INE"中
    return exchangeid in "SHFE INE"


# 定义一个函数，用于提交平仓订单，symbol 为合约代码，exchangid 为交易所 ID，direction 为买卖方向，volume 为下单数量，comboffsetflag 为开平标志
def close_order(context, symbol, exchangid, direction, volumn, comboffsetflag):
    try:
        # 获取指定合约的 tick 数据
        tick = get_tick(symbol)
        # 判断 tick 数据中是否包含 ActionDay 字段，如果不包含则表示 tick 获取失败
        if not tick.get("ActionDay"):
            # tick 获取失败，返回等待下一次运行
            return
        # 获取 tick 数据中的涨停价，并转换为浮点数类型
        upper_price = float(tick.get("UpperLimitPrice"))
        # 获取 tick 数据中的跌停价，并转换为浮点数类型
        lower_price = float(tick.get("LowerLimitPrice"))
        # 根据买卖方向选择下单价格，如果方向为 “0” 则使用涨停价，否则使用跌停价
        price = upper_price if direction == "0" else lower_price
        # 构建订单字典
        order = {
            "symbol": symbol,  # 合约代码
            "exchangeid": exchangid,  # 交易所ID
            "limitprice": price,  # 下单价格
            "direction": direction,  # 买卖方向
            "orderpricetype": "2",  # 订单价格类型
            "comboffsetflag": comboffsetflag,  # 开平标志
            "volumn": volumn,  # 下单数量
        }
        # 发送订单
        send_order(order)
    except Exception as e:
        # 打印开单出现错误的提示信息，包含错误信息
        print(f"开单出现错误=>请查看GridTrade.close_order[{e}]")
        # 打印具体的错误信息
        print(e)
        # 开单失败，返回 0
        return 0
    # 开单成功，返回 1
    return 1


# 定义一个函数，用于清空持仓
def clear_position(context):
    # 获取当前持仓信息
    position = get_position()

    # 定义平仓方向映射字典，将持仓方向映射为平仓方向
    CLOSE_DIRECTION_MAP = {"2": "1", "3": "0"}

    # 遍历持仓信息列表
    for item in position:
        # 获取持仓方向
        PosiDirection = item.get("PosiDirection")
        # 根据持仓方向获取平仓方向
        direction = CLOSE_DIRECTION_MAP[PosiDirection]
        # 获取昨仓冻结数量
        YdStrikeFrozen = item.get("YdStrikeFrozen")
        # 获取多仓冻结数量
        LongFrozen = item.get("LongFrozen")
        # 获取空仓冻结数量
        ShortFrozen = item.get("ShortFrozen")
        # 根据平仓方向选择冻结数量
        Frozen = LongFrozen if direction == "0" else ShortFrozen
        # 计算今仓冻结数量
        todayFrozen = Frozen - YdStrikeFrozen
        # 计算今仓可平数量
        today_volume = item.get("TodayPosition") - todayFrozen
        # 计算昨仓可平数量
        yesterday_volume = item.get("YdPosition") - YdStrikeFrozen
        # 获取交易所ID
        exchangeid = item.get("ExchangeID")
        # 获取合约代码
        InstrumentID = item.get("InstrumentID")
        # 初始化今仓平仓数量为 0
        today_position = 0
        # 判断交易所是否为上海期货交易所或上海国际能源交易中心
        if isSHFEorINE(context, exchangeid):
            # 如果是，则今仓平仓数量为今仓可平数量，昨仓平仓数量为昨仓可平数量
            today_position = today_volume
            yesterday_position = yesterday_volume
        else:
            # 如果不是，则昨仓平仓数量为今仓可平数量和昨仓可平数量之和
            yesterday_position = today_volume + yesterday_volume
        """暂定一组订单同时报两次"""
        """不做回复确认 => 重获取持仓时再做重新分配 直到分配结束"""
        # 如果今仓平仓数量大于 0，则提交今仓平仓订单
        if today_position > 0:
            close_order(context, InstrumentID, exchangeid, direction, today_position, "3")
        # 如果昨仓平仓数量大于 0，则提交昨仓平仓订单
        if yesterday_position > 0:
            close_order(context, InstrumentID, exchangeid, direction, yesterday_position, "1")


# 定义一个函数，用于在到达指定时间时执行平仓操作
def day_over_close(context):
    """到达时间执行平仓"""
    # 从datetime模块中导入 datetime 和 time 类，用于处理日期和时间
    from datetime import datetime, time

    # 判断日内清盘模式是否开启，如果未开启则直接返回
    if not context.day_clear_mode:
        return
    # 获取当前时间
    now = datetime.now().time()

    # 定义时间段
    # 将日内清盘开始时间字符串转换为 time 对象
    start_time = datetime.strptime(context.day_clear_time, "%H:%M").time()  # 开始时间为 14:58
    # 将日内清盘结束时间字符串转换为 time 对象
    end_time = datetime.strptime(context.day_over_time, "%H:%M").time()  # 结束时间为 15:00
    # 判断当前时间是否在指定的时间段内
    if start_time <= now < end_time:
        # 如果在时间段内，则调用清空持仓函数
        clear_position(context)


# 定义停止函数，当程序停止时会调用该函数
def on_stop(context):
    # 打印程序停止的提示信息
    print("demo停止")

关键要点

1. 时间窗口判断：用 datetime.strptime 把参数字符串解析成 time 对象，以 start_time <= now < end_time 判断是否处于清仓窗口；结束时间取开区间，避免越过窗口后仍反复触发
2. 今昨仓拆分：冻结数量按平仓方向取 LongFrozen / ShortFrozen，其中昨仓冻结为 YdStrikeFrozen，二者之差才是今仓冻结；可平数量须从 TodayPosition / YdPosition 中扣除对应冻结
3. 上期所特殊规则：SHFE / INE 必须区分平今（"3"）与平昨（"1"），其他交易所没有平今概念，今仓并入昨仓一并按平昨报出
4. 涨跌停价报单：平多按涨停价、平空按跌停价报限价单，用极端价格提高收盘前的成交概率
5. tick 获取保护：close_order 中先校验 tick 的 ActionDay，取不到行情直接跳过，等待下一个 tick 重试
6. 不做回报确认：策略不跟踪每笔平仓单的成交状态，而是在窗口内每个 tick 重新拉取持仓、重新计算可平数量并报单，直至持仓清零

---

案例二：跨品种套利策略

场景

基于 j2505（焦炭）和 jm2505（焦煤）的跨品种套利，监控两种合约比值的变化，动态计算其均值和标准差；当比值偏离均值超过一倍标准差时进行反向开仓，博弈比值回归到一倍标准差区间之内。

套利策略涉及多品种或周期的合约数据订阅，订阅方法参见多品种多周期相关章节，通过传入的合约列表解析出具体合约，即可进行后续开仓信号的计算。

策略逻辑

1. 通过 QMF_SUBSCRIBE_SYMBOLS 订阅两个合约（如 j2505:M5&jm2505:M5），在 on_init 中解析出两个合约 ID
2. 每个 K 线收盘时获取两个合约的收盘价，计算比值序列 ratio = close1 / close2
3. 用 Pandas 滚动计算比值在 calc_len 周期内的均值和标准差，得到上下轨（均值 ± deviation 倍标准差）
4. 开仓信号：比值由轨内上穿上轨 → 开反向套利仓（卖出基础合约、买入第二合约）；由轨内下穿下轨 → 开正向套利仓（买入基础合约、卖出第二合约）
5. 平仓信号：比值重新穿回轨道内侧时，按持仓方向平掉对应套利腿
6. on_trade 回调中维护本地持仓字典 context.positions；on_bar 中先处理平仓信号、再处理开仓信号

完整代码

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


# 导入NumPy库，用于高效的数值计算
import numpy as np

# 导入Pandas库，用于数据处理和分析
import pandas as pd

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


# 判断当前是否持有指定方向的仓位
def isHolding(positions, instrument_id, direction):
    # 检查指定的合约 ID 是否在持仓信息字典中
    if instrument_id in positions:
        # 从持仓信息字典中获取指定合约的持仓信息
        position = positions.get(instrument_id)
        # 从持仓信息中获取持仓方向
        position_type = position.get("Direction")
        # 判断持仓方向是否与指定方向一致
        if str(position_type) == str(direction):
            return True
    return False


# 获取 K 线数据
def get_k_line_data(context):
    # 调用g et_kline 函数获取基础合约的 K 线数据，并将结果存储在klines中
    context.klines = get_kline(context.base_instrument_id, context.base_period, context.calc_len)
    # 从 K 线数据中提取收盘价数据，并存储在 close_array 中
    context.close_array = context.klines.get("close")
    # 调用 get_kline 函数获取第二个合约的 K 线数据，并存储在 klines2 中
    context.klines2 = get_kline(context.instrument_id2, context.base_period, context.calc_len)
    # 从第二个合约的 K 线数据中提取收盘价数据，并存储在 close_array2 中
    context.close_array2 = context.klines2.get("close")


# 计算比例数据
def calc_indicator(context):
    # 检查基础合约或第二个合约的收盘价数据是否为空
    if (not context.close_array) or (not context.close_array2):
        # 若为空，返回 -1 表示数据不足
        return -1
    # 检查基础合约或第二个合约的收盘价数据长度是否小于统计周期
    if len(context.close_array) < context.calc_len or len(context.close_array2) < context.calc_len:
        # 若长度不足，直接返回
        return
    # 将基础合约和第二个合约的收盘价数据转换为 NumPy 数组，并计算比例数组
    ratio_array = np.asarray(context.close_array) / np.asarray(context.close_array2)
    # 将比例数组转换为 Pandas 的 Series 对象，方便进行滚动计算
    series = pd.Series(ratio_array)
    # 计算比例数组的滚动均值，并将结果转换为NumPy数组
    rolling_mean = series.rolling(window=context.calc_len).mean().values
    # 计算比例数组的滚动标准差，并将结果转换为NumPy数组
    rolling_std = series.rolling(window=context.calc_len).std().values
    # 将 context 中的比例数组、均值数组和标准差数组分别向后滚动一位
    context.ratio_array = np.roll(context.ratio_array, -1)
    context.mean_array = np.roll(context.mean_array, -1)
    context.std_array = np.roll(context.std_array, -1)
    # 将最新的比例值、均值和标准差分别赋值给数组的最后一个元素
    context.ratio_array[-1] = ratio_array[-1]
    context.mean_array[-1] = rolling_mean[-1]
    context.std_array[-1] = rolling_std[-1]


# 根据偏离度产生开仓信号
def get_open_signal(context):
    # 检查比例数组、均值数组或标准差数组的长度是否小于 3
    if len(context.ratio_array) < 3 or len(context.mean_array) < 3 or len(context.std_array) < 3:
        # 若长度不足，返回 0 表示数值不足，无法产生开仓信号
        return 0
    # 初始化开仓信号为 0，表示无开仓信号
    sig_open = 0
    # 判断是否满足开空仓的条件
    if (
        context.ratio_array[-3] <= context.mean_array[-3] + context.deviation * context.std_array[-3]
        and context.ratio_array[-2] > context.mean_array[-2] + context.deviation * context.std_array[-2]
    ):
        # 若满足条件，将开仓信号设置为 -1，表示开空仓
        sig_open = -1
    # 判断是否满足开多仓的条件
    elif (
        context.ratio_array[-3] >= context.mean_array[-3] - context.deviation * context.std_array[-3]
        and context.ratio_array[-2] < context.mean_array[-2] - context.deviation * context.std_array[-2]
    ):
        # 若满足条件，将开仓信号设置为 1，表示开多仓
        sig_open = 1
    # 计算上轨线的值
    upper = context.mean_array[-2] + context.deviation * context.std_array[-2]
    # 计算下轨线的值
    lower = context.mean_array[-2] - context.deviation * context.std_array[-2]
    return sig_open


# 根据指标计算平仓信号
def get_close_signal(context):
    # 检查比例数组、均值数组或标准差数组的长度是否小于 3
    if len(context.ratio_array) < 3 or len(context.mean_array) < 3 or len(context.std_array) < 3:
        # 若长度不足，返回 0 表示数值不足，无法产生平仓信号
        return 0
    # 初始化平仓信号为 0，表示无平仓信号
    sig_close = 0
    # 判断是否满足平空仓的条件
    if (
        context.ratio_array[-3] >= context.mean_array[-3] + context.deviation * context.std_array[-3]
        and context.ratio_array[-2] < context.mean_array[-2] + context.deviation * context.std_array[-2]
    ):
        # 若满足条件，将平仓信号设置为 1，表示平空仓
        sig_close = 1
    # 判断是否满足平多仓的条件
    elif (
        context.ratio_array[-3] <= context.mean_array[-3] - context.deviation * context.std_array[-3]
        and context.ratio_array[-2] > context.mean_array[-2] - context.deviation * context.std_array[-2]
    ):
        # 若满足条件，将平仓信号设置为 -1，表示平多仓
        sig_close = -1
    return sig_close


# 判断当前是否持有仓位
def isHoldingPosition(context):
    # 检查基础合约和第二个合约的多仓和空仓情况，只要有一个合约持有仓位，则返回 True
    if (
        isHolding(context.positions, context.base_instrument_id, "0")
        or isHolding(context.positions, context.base_instrument_id, "1")
        or isHolding(context.positions, context.instrument_id2, "0")
        or isHolding(context.positions, context.instrument_id2, "1")
    ):
        return True
    else:
        return False


# 开一对正向套利仓位
def open_long_position(context):
    # 以指定手数买入基础合约
    buy_open(context.volume, context.base_instrument_id)
    # 以指定手数卖出第二个合约
    sell_open(context.volume, context.instrument_id2)


# 开一对反向套利仓位
def open_short_position(context):
    # 以指定手数卖出基础合约
    sell_open(context.volume, context.base_instrument_id)
    # 以指定手数买入第二个合约
    buy_open(context.volume, context.instrument_id2)


# 根据信号开仓
def open_position_by_signal(context, sig_open):
    if sig_open == 1:  # 开多信号
        # 检查当前是否没有持仓
        if not isHoldingPosition(context):
            # 若没有持仓，则开一对正向套利仓位
            open_long_position(context)
    elif sig_open == -1:  # 开空信号
        # 检查当前是否没有持仓
        if not isHoldingPosition(context):
            # 若没有持仓，则开一对反向套利仓位
            open_short_position(context)


# 结束一对正向套利仓位
def close_long_position(context):
    # 检查是否持有基础合约的多仓
    if isHolding(context.positions, context.base_instrument_id, "0"):
        # 若持有多仓，则以指定手数卖出基础合约进行平仓
        sell_close(context.volume, context.base_instrument_id)
    # 检查是否持有第二个合约的空仓
    if isHolding(context.positions, context.instrument_id2, "1"):
        # 若持有空仓，则以指定手数买入第二个合约进行平仓
        buy_close(context.volume, context.instrument_id2)


# 结束一对反向套利仓位
def close_short_position(context):
    # 检查是否持有基础合约的空仓
    if isHolding(context.positions, context.base_instrument_id, "1"):
        # 若持有空仓，则以指定手数买入基础合约进行平仓
        buy_close(context.volume, context.base_instrument_id)
    # 检查是否持有第二个合约的多仓
    if isHolding(context.positions, context.instrument_id2, "0"):
        # 若持有多仓，则以指定手数卖出第二个合约进行平仓
        sell_close(context.volume, context.instrument_id2)


# 根据信号平仓
def close_position_by_signal(context, sig_close):
    if sig_close == 1:  # 平空信号
        # 若收到平空信号，则结束一对反向套利仓位
        close_short_position(context)
    elif sig_close == -1:  # 平多信号
        # 若收到平多信号，则结束一对正向套利仓位
        close_long_position(context)


#  K 线收盘运行逻辑
def on_bar(context):
    # 调用 get_k_line_data 函数获取 K 线数据
    get_k_line_data(context)
    # 调用 calc_indicator 函数计算比例数据
    calc_indicator(context)
    # 调用 get_open_signal 函数获取开仓信号
    sig_open = get_open_signal(context)
    # 调用 get_close_signal 函数获取平仓信号
    sig_close = get_close_signal(context)
    # 调用 close_position_by_signal 函数根据平仓信号进行平仓操作
    close_position_by_signal(context, sig_close)
    # 调用 open_position_by_signal 函数根据开仓信号进行开仓操作
    open_position_by_signal(context, sig_open)


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


# 定义一个参数映射模型类
class Params(BaseModel, validate_assignment=True):
    # 统计周期，默认为 100
    calc_len: int = Field(default=100, title="统计周期")
    # 触发交易的偏离度(标准差倍数)，默认为 1.0
    deviation: float = Field(default=1.0, title="触发交易的偏离度(标准差倍数)")
    # 手数，默认为1
    volume: int = Field(default=1, title="手数")
    # 订阅合约和周期，默认为 "j2505:M5&jm2505:M5"
    QMF_SUBSCRIBE_SYMBOLS: str = Field(default="j2505:M5&jm2505:M5", title="订阅合约和周期")


# 初始化一些参数
def on_init(context):
    # 将订阅合约和周期字符串按 "&" 分割成列表
    symbols = QMF_SUBSCRIBE_SYMBOLS.split("&")
    # 从分割后的列表中提取基础合约的 ID，并存储在 context.base_instrument_id 中
    context.base_instrument_id = symbols[0].split(":")[0]
    # 从分割后的列表中提取第二个合约的 ID，并存储在 context.instrument_id2 中
    context.instrument_id2 = symbols[1].split(":")[0]
    # 从 BASE_SETTING 中获取基础周期，并存储在 context.base_period 中
    context.base_period = BASE_SETTING.get("bPeriod")
    # 初始化持仓信息字典为空字典
    context.positions = {}
    # 初始化基础合约的收盘价数组为长度为统计周期的全零数组
    context.close_array = np.zeros(context.calc_len)
    # 初始化比例数组为长度为统计周期的全零数组
    context.ratio_array = np.zeros(context.calc_len)
    # 初始化均值数组为长度为统计周期的全零数组
    context.mean_array = np.zeros(context.calc_len)
    # 初始化标准差数组为长度为统计周期的全零数组
    context.std_array = np.zeros(context.calc_len)
    # 初始化第二个合约的收盘价数组为长度为统计周期的全一数组
    context.close_array2 = np.ones(context.calc_len)
    # 初始化第二个合约的比例数组为长度为统计周期的全零数组
    context.ratio_array2 = np.zeros(context.calc_len)
    # 初始化第二个合约的均值数组为长度为统计周期的全零数组
    context.mean_array2 = np.zeros(context.calc_len)
    # 初始化第二个合约的标准差数组为长度为统计周期的全零数组
    context.std_array2 = np.zeros(context.calc_len)


# 每次有行情进来，便会调用 on_tick 函数刷新一次
def on_tick(context):
    # 每产生新的 K 线时，调用 on_bar_run 函数执行 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)


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


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

关键要点

1. 比值而非价差：跨品种套利两合约价格量级不同，以收盘价比值 close1 / close2 作为套利标的，比直接用价差更具可比性
2. 滚动统计：借助 Pandas 的 rolling(window=calc_len).mean()/.std() 计算滚动均值与标准差，上下轨为 均值 ± deviation × 标准差
3. 穿越式信号：信号比较的是倒数第 3 根与倒数第 2 根 K 线（即已收盘的最近两根），要求比值从轨内穿越到轨外（或反向穿回）才触发，避免在轨道边缘反复报单
4. 双腿对称开平仓：正向套利为「买基础合约 + 卖第二合约」，反向套利反之；平仓时逐腿检查持仓方向，有哪条腿就平哪条腿
5. 本地持仓维护：on_trade 中按开平仓标志增删 context.positions，isHolding 据此判断各合约各方向是否有仓，开仓前先确认无持仓
6. 先平后开：on_bar 中先执行平仓逻辑再执行开仓逻辑，保证同一根 K 线上信号翻转时先释放旧仓位

---

相关阅读

• [核心API 实时行情数据]
• [核心API 历史K线数据]
• [核心API 账户持仓查询]
• [核心API 交易指令]
• [案例：网格策略]
• [案例：自动移仓与订单备份]