首页
完整示例 / 期货示例

完整示例

期货示例

四份可直接运行的期货回测策略：单合约均线多空、多合约目标组合、主力合约换月、限价挂单与超时撤单，附运行要求与参数说明。

以下每份代码都包含所需回调与辅助逻辑，复制整段到策略编辑器即可运行。合约、回测日期和数据覆盖要互相匹配：示例合约是历史代码，换期间时换成对应的真实合约。示例用于演示 API 用法，不承诺收益。

示例 演示内容 运行要求
单合约均线多空 用已结束 Bar 计算均线；先平反向仓，下一次回调确认后开新方向；查询、成交和拒单日志 日线或分钟；区间覆盖长窗口所需 Bar；单一真实合约
多合约目标组合 历史动量构造多空目标字典；调仓间隔；行情不齐时不清仓；调仓前检查未完成委托 日线或分钟；两个真实合约都有有效数据；策略专属账户
主力合约换月 主力映射预加载、取上一已知映射、先平旧仓再开新仓、缺行情时等待 日线或分钟；需主力映射及真实合约行情；独立的多头账户
限价挂单与撤单 GFD 限价单、价位对齐、订单号登记、等待若干 Bar 后撤单、回调清理 分钟回测；按所选合约设置 price_tick

参数约定：前三份的 hands 是目标手数；限价示例的 hands 是单次委托手数，cancel_after_bars 是等待的 Bar 数，offset_ticks 是距参考价的最小变动价位个数。调仓间隔、窗口参数为整数，合约列表为字符串列表。

一、单合约均线多空

python
"""单合约均线多空策略；在平台选择日线或分钟回测后运行。"""
from panda_backtest.api.api import (
    buy_open, sell_open, buy_close, sell_close, get_open_future_orders,
)


def initialize(context):
    context.account = context.run_info.future_account
    context.symbol = "A2501.DCE"
    context.hands = 1
    context.fast_window = 5
    context.slow_window = 20
    context.prices = []
    if not 0 < context.fast_window < context.slow_window:
        raise ValueError("均线窗口必须满足 0 < 短窗口 < 长窗口")
    if context.hands <= 0:
        raise ValueError("目标手数必须为正整数")
    context.sub_future_symbol([context.symbol])


def handle_data(context, data):
    bar = data[context.symbol]
    if bar is None or not (0 < bar.close < float("inf")):
        return
    # 先从已经结束的历史 bar 计算信号，再加入本次价格。
    signal = 0
    if len(context.prices) >= context.slow_window:
        fast = sum(context.prices[-context.fast_window:]) / context.fast_window
        slow = sum(context.prices[-context.slow_window:]) / context.slow_window
        if fast > slow:
            signal = 1
        elif fast < slow:
            signal = -1
    context.prices.append(float(bar.close))
    context.prices = context.prices[-context.slow_window:]
    if signal == 0 or get_open_future_orders(context.account, context.symbol):
        return
    account = context.future_account_dict.get(context.account)
    if account is None:
        raise ValueError("找不到期货账户")
    position = account.positions[context.symbol]
    # 反向仓位先平；下一次回调核实持仓后，才建立新方向。
    if signal > 0 and position.sell_quantity > 0:
        amount = int(position.closable_sell_quantity)
        if amount > 0:
            buy_close(context.account, context.symbol, amount)
        return
    if signal < 0 and position.buy_quantity > 0:
        amount = int(position.closable_buy_quantity)
        if amount > 0:
            sell_close(context.account, context.symbol, amount)
        return
    if signal > 0:
        difference = int(context.hands - position.buy_quantity)
        if difference > 0:
            buy_open(context.account, context.symbol, difference)
        elif difference < 0:
            amount = int(min(-difference, position.closable_buy_quantity))
            if amount > 0:
                sell_close(context.account, context.symbol, amount)
    else:
        difference = int(context.hands - position.sell_quantity)
        if difference > 0:
            sell_open(context.account, context.symbol, difference)
        elif difference < 0:
            amount = int(min(-difference, position.closable_sell_quantity))
            if amount > 0:
                buy_close(context.account, context.symbol, amount)


def on_future_trade_rtn(context, order):
    print(f"成交回报 {order.order_book_id} 订单={order.order_id} "
          f"本次={order.cur_filled_quantity} 累计={order.filled_quantity}手")


def future_order_cancel(context, order):
    print(f"撤单或拒单 {order.order_id} 状态={order.status} 原因={order.message}")


def after_trading(context):
    account = context.future_account_dict[context.account]
    print(f"{context.trade_date} 权益={account.total_value:.2f}元 "
          f"可用={account.cash:.2f}元 保证金={account.margin:.2f}元")

二、多合约目标组合

python
"""多个真实合约按历史动量调整目标手数；使用专属回测账户。"""
from panda_backtest.api.api import target_future_group_order, get_open_future_orders


def initialize(context):
    context.account = context.run_info.future_account
    context.symbols = ["A2501.DCE", "M2501.DCE"]
    context.hands = 1
    context.lookback = 5
    context.rebalance_bars = 5
    context.bar_count = 0
    context.last_rebalance = -1
    context.history = {symbol: [] for symbol in context.symbols}
    if context.hands <= 0 or context.lookback < 2 or context.rebalance_bars < 1:
        raise ValueError("手数、窗口及调仓间隔参数不合法")
    if len(context.symbols) != len(set(context.symbols)):
        raise ValueError("合约列表不能重复")
    context.sub_future_symbol(context.symbols)


def handle_data(context, data):
    # 缺少任意目标合约行情时，不以空目标字典触发清仓。
    current = {}
    for symbol in context.symbols:
        bar = data[symbol]
        if bar is None or not (0 < bar.close < float("inf")):
            return
        current[symbol] = float(bar.close)
    context.bar_count += 1
    ready = all(len(context.history[s]) >= context.lookback for s in context.symbols)
    long_targets = {}
    short_targets = {}
    if ready:
        for symbol in context.symbols:
            prices = context.history[symbol]
            if prices[-1] > prices[0]:
                long_targets[symbol] = int(context.hands)
            elif prices[-1] < prices[0]:
                short_targets[symbol] = int(context.hands)
    for symbol in context.symbols:
        context.history[symbol].append(current[symbol])
        context.history[symbol] = context.history[symbol][-context.lookback:]
    if not ready:
        return
    if context.last_rebalance >= 0:
        if context.bar_count - context.last_rebalance < context.rebalance_bars:
            return
    if get_open_future_orders(context.account):
        return
    account = context.future_account_dict[context.account]
    for symbol in account.positions.keys():
        bar = data[symbol]
        if bar is None or not (0 < bar.close < float("inf")):
            return
    # 字典描述整个账户的目标；缺席的方向会被平仓。
    target_future_group_order(context.account, long_targets, short_targets)
    context.last_rebalance = context.bar_count
    print(f"组合调仓：目标多头={long_targets}，目标空头={short_targets}")


def on_future_trade_rtn(context, order):
    print(f"组合成交 {order.order_book_id} 累计={order.filled_quantity}手")


def future_order_cancel(context, order):
    print(f"组合订单未完成 {order.order_id} 状态={order.status} 原因={order.message}")


def after_trading(context):
    account = context.future_account_dict[context.account]
    print(f"{context.trade_date} 组合权益={account.total_value:.2f}元")

三、主力合约换月

单品种主力换月：采用前一个有记录交易日的主力，确认旧仓归零后才开新仓，流程说明见主力换月与排查。这份示例使用独立账户、只管理一个品种的多头，不要直接挂到已有其他策略持仓的账户上。

python
"""单品种主力换月：采用前一个有记录交易日的主力，确认旧仓归零后开新仓。"""
import panda_data
from panda_backtest.api.api import (
    buy_open, sell_close, get_open_future_orders,
)


def initialize(context):
    context.account = context.run_info.future_account
    context.underlying = "A"
    context.hands = 1
    context.target_symbol = ""
    context.mapping = {}
    start = panda_data.get_prev_trade_date(context.run_info.start_date, n=1)
    if not start:
        raise ValueError("无法确定预热交易日")
    frame = panda_data.get_future_dominant(
        underlying_symbol=[context.underlying],
        start_date=start,
        end_date=context.run_info.end_date,
    )
    if frame is None or frame.empty:
        raise ValueError("主力合约数据为空")
    required = ["underlying_symbol", "date", "symbol"]
    if any(name not in frame.columns for name in required):
        raise ValueError("主力合约数据缺少必要列")
    for _, row in frame.iterrows():
        if str(row["underlying_symbol"]).upper() != context.underlying:
            continue
        date = str(row["date"]).replace("-", "")[:8]
        symbol = row["symbol"]
        if not isinstance(symbol, str) or "." not in symbol or "_" in symbol:
            raise ValueError("主力映射必须给出真实交割合约代码")
        if len(date) != 8 or not date.isdigit():
            raise ValueError("主力映射日期格式不正确")
        if date in context.mapping and context.mapping[date] != symbol:
            raise ValueError("同一天存在不同主力合约记录")
        context.mapping[date] = symbol
    context.mapping_dates = sorted(context.mapping)
    if not context.mapping_dates:
        raise ValueError("没有所选品种的主力映射")


def before_trading(context):
    # 只使用当前交易日之前的记录，避免读取当天收盘后才形成的主力判定。
    dates = [date for date in context.mapping_dates if date < context.trade_date]
    if not dates:
        raise ValueError("当前交易日之前没有可用主力记录")
    context.target_symbol = context.mapping[dates[-1]]
    context.sub_future_symbol([context.target_symbol])


def handle_data(context, data):
    if not context.target_symbol or get_open_future_orders(context.account):
        return
    account = context.future_account_dict[context.account]
    target_bar = data[context.target_symbol]
    if target_bar is None or not (0 < target_bar.close < float("inf")):
        return
    # 此例使用独立账户，只管理一个品种的多头，拒绝接管已有空头。
    for symbol in account.positions.keys():
        if account.positions[symbol].sell_quantity > 0:
            raise ValueError("换月示例账户含空头，请使用独立的多头回测账户")
    old_symbols = [s for s in account.positions.keys() if s != context.target_symbol]
    if old_symbols:
        for symbol in old_symbols:
            position = account.positions[symbol]
            bar = data[symbol]
            if bar is None or not (0 < bar.close < float("inf")):
                continue
            amount = int(position.closable_buy_quantity)
            if amount > 0:
                sell_close(context.account, symbol, amount)
        # 本次仅平旧仓。下一次回调重新读取持仓确认平仓结果。
        return
    position = account.positions[context.target_symbol]
    difference = int(context.hands - position.buy_quantity)
    if difference > 0:
        buy_open(context.account, context.target_symbol, difference)
    elif difference < 0:
        amount = int(min(-difference, position.closable_buy_quantity))
        if amount > 0:
            sell_close(context.account, context.target_symbol, amount)


def on_future_trade_rtn(context, order):
    print(f"换月成交 {order.order_book_id} 累计={order.filled_quantity}手")


def future_order_cancel(context, order):
    print(f"换月订单未完成 {order.order_id} 状态={order.status} 原因={order.message}")


def after_trading(context):
    account = context.future_account_dict[context.account]
    print(f"{context.trade_date} 主力目标={context.target_symbol} "
          f"实际合约={account.positions.keys()} 权益={account.total_value:.2f}元")

四、限价挂单与超时撤单

每个交易日最多发一笔开多委托，持仓后不继续加仓。默认使用历史豆一合约和 price_tick=1.0，更换合约时同步核实最小变动价位。

日线模式一天只有一次 handle_data，GFD 单可能在下一次调用前就已到期，所以观察超时撤单请用分钟回测。

python
"""限价挂单及超时撤单：每个交易日最多发一笔开多委托。"""
from panda_backtest.api.api import (
    buy_open, cancel_future_order, get_open_future_orders,
    LimitOrderStyle, TimeConditionType,
)


def initialize(context):
    context.account = context.run_info.future_account
    context.symbol = "A2501.DCE"
    context.hands = 1
    context.price_tick = 1.0
    context.offset_ticks = 2
    context.cancel_after_bars = 3
    context.bar_count = 0
    context.sent_today = False
    context.orders = {}
    context.finished_orders = set()
    if not (0 < context.price_tick < float("inf")) or context.cancel_after_bars < 1:
        raise ValueError("最小变动价位和撤单等待 bar 数必须为正数")
    context.sub_future_symbol([context.symbol])


def before_trading(context):
    context.sent_today = False


def handle_data(context, data):
    context.bar_count += 1
    open_orders = get_open_future_orders(context.account, context.symbol)
    # 只撤本示例登记的订单；查到其他订单时也不重复下单。
    open_ids = {order.order_id for order in open_orders}
    for order_id in list(context.orders):
        if order_id not in open_ids:
            context.orders.pop(order_id, None)
    for order in open_orders:
        if order.order_id in context.orders:
            if context.bar_count - context.orders[order.order_id] >= context.cancel_after_bars:
                cancel_future_order(context.account, order.order_id)
    if open_orders or context.sent_today:
        return
    position = context.future_account_dict[context.account].positions[context.symbol]
    if position.buy_quantity > 0 or position.sell_quantity > 0:
        return
    bar = data[context.symbol]
    if bar is None or not (0 < bar.close < float("inf")):
        return
    ticks = int(bar.close / context.price_tick) - int(context.offset_ticks)
    limit_price = round(ticks * context.price_tick, 8)
    if limit_price <= 0:
        return
    context.sent_today = True
    orders = buy_open(
        context.account, context.symbol, int(context.hands),
        style=LimitOrderStyle(limit_price, time_condition=TimeConditionType.GFD),
    )
    # 回测回调可能先于函数返回：仅登记返回时仍在途的订单。
    for order in orders or []:
        if order.status in (0, 1, 4, 6) and order.order_id not in context.finished_orders:
            context.orders[order.order_id] = context.bar_count
        elif order.status == -1:
            print(f"委托被拒绝：{order.message}")


def on_future_trade_rtn(context, order):
    if order.status in (-1, 2, 3, 5, 7):
        context.finished_orders.add(order.order_id)
        context.orders.pop(order.order_id, None)
    print(f"限价成交 {order.order_id} 本次={order.cur_filled_quantity}手 "
          f"累计={order.filled_quantity}手 剩余={order.unfilled_quantity}手")


def future_order_cancel(context, order):
    if order.status in (-1, 2, 3, 5, 7):
        context.finished_orders.add(order.order_id)
    context.orders.pop(order.order_id, None)
    print(f"委托结束 {order.order_id} 状态={order.status} 原因={order.message}")


def after_trading(context):
    account = context.future_account_dict[context.account]
    print(f"{context.trade_date} 权益={account.total_value:.2f}元")