博主用AI+手工修复，实现了一个基于目标持仓信号的期货交易回测框架，采用先进先出（FIFO）原则逐笔管理多空持仓明细，同时对代码生成结果逐一测试，修复完成后写个博客记录一下。

回测逻辑

1. 交易执行逻辑

依据前一交易日收盘生成的目标持仓信号，在次日开盘价执行交易，支持开仓、平仓、加仓、减仓及多空双向操作。

2. 持仓管理

利用栈结构存储每笔开仓的手数和价格，并按 FIFO 顺序（优先平掉最早开立的仓位）处理平仓。

3. 盈亏计算

• 平仓时按开盘价与持仓结算价之差计算已实现盈亏。
• 每日收盘后按结算价计算未实现浮动盈亏。
• 逐日累计总权益，并输出每日收益与累计收益。
• 费用处理：可分别设置开仓、平仓的固定手续费（按手数收取）。

4. 输出结果

• result_df：每日统计表，包含交易信号、持仓调整量、当日持仓、当日收益、累计收益等字段。
• trade_df：逐笔成交明细表，记录每笔交易的方向、手数、价格、手续费及备注（如开多、平空、减多等）。

完整代码

下面的代码可直接运行：

from copy import deepcopy

import pandas as pd


def futures_backtest_fifo(
        trade_dataframe,
        open_col,
        settle_col,
        signal_col,
        multiplier=10000,
        open_comm=0.0,
        close_comm=0.0,
        round_precision=5,
):
    """
    基于目标持仓信号的期货回测框架（FIFO 持仓明细 + 交易明细）5

    Parameters
    ----------
    trade_dataframe : pd.DataFrame
        必须包含日期索引或列、开盘价、结算价、目标持仓信号列。
    open_col : str
        用于执行交易的开盘价列名。
    settle_col : str
        用于逐日盯市的结算价列名。
    signal_col : str
        目标持仓列，当日收盘决定，次日开盘执行。
    multiplier : float
        合约乘数
    open_comm : float, default 0.0
        开仓每手固定手续费。
    close_comm : float, default 0.0
        平仓每手固定手续费。
    round_precision: float, default 5.0
        计算精度
    Returns
    -------
    result_df : pd.DataFrame
        每日统计表，包含列：日期、真实目标值、预测目标值、交易信号、
        持仓调整量、当日持仓、当日收益、累计收益。
    trade_df : pd.DataFrame
        逐笔成交明细表，包含列：日期、方向、手数、价格、手续费、备注。
    """
    trade_dataframe = deepcopy(trade_dataframe)
    # 处理合约乘数
    if open_col != settle_col:
        trade_dataframe[open_col] = trade_dataframe[open_col] * multiplier
    trade_dataframe[settle_col] = trade_dataframe[settle_col] * multiplier

    # 持仓明细栈：存储 (手数, 开仓价)，手数正为多，负为空
    position_stack = []
    total_position = 0
    prev_equity = 0.0
    prev_settle = None

    result_rows = []
    trades = []  # 记录每一笔成交

    for i in range(len(trade_dataframe)):
        row = trade_dataframe.iloc[i]
        open_price = row[open_col]
        settle_price = row[settle_col]
        signal = row[signal_col]

        trade_volume = 0
        realized_pnl = 0.0
        fee = 0.0  # 费用计算

        # ---------- 执行前一日的目标持仓信号 ----------
        if i > 0:
            # prev_signal = df.iloc[i - 1][signal_col]
            today_target_vol = trade_dataframe.iloc[i - 1][signal_col]  # 今日的目标仓位，来源于昨日交易信号

            adjust = today_target_vol - total_position
            trade_volume = adjust

            # 修改当日结算
            for _i in range(len(position_stack)):
                position_stack[_i] = (position_stack[_i][0], prev_settle)

            if adjust != 0:
                # 情况1：平掉全部持仓（目标与持仓反向或目标为0）
                if total_position != 0 and today_target_vol * total_position <= 0:
                    while position_stack:
                        _trade_volume, _trade_cost = position_stack.pop(0)
                        if _trade_volume > 0:
                            pnl = round((open_price - prev_settle) * abs(_trade_volume), round_precision)  # 今日浮盈浮亏
                            direction = "卖出"
                            note = "平多"
                        else:
                            pnl = round((prev_settle - open_price) * abs(_trade_volume), round_precision)
                            direction = "买入"
                            note = "平空"
                        realized_pnl += pnl
                        fee += abs(_trade_volume) * close_comm
                        total_position -= _trade_volume
                        # 记录平仓明细
                        trades.append({
                            "日期": row.name if trade_dataframe.index.name is not None else i,
                            "方向": direction,
                            "手数": abs(_trade_volume),
                            "价格": open_price,
                            "手续费": abs(_trade_volume) * close_comm,
                            "备注": note
                        })
                    # 此时 total_position 应为 0

                # 如果目标不为 0，则开新仓至目标手数
                if today_target_vol != 0:
                    open_qty = today_target_vol - total_position  # 追加量
                    if open_qty > 0:
                        position_stack.append((open_qty, open_price))
                        direction = "买入"
                        note = "开多"
                    elif open_qty < 0:
                        position_stack.append((open_qty, open_price))
                        direction = "卖出"
                        note = "开空"
                    fee += abs(open_qty) * open_comm
                    total_position = deepcopy(today_target_vol)
                    # 记录开仓明细
                    trades.append({
                        "日期": row.name if trade_dataframe.index.name is not None else i,
                        "方向": direction,
                        "手数": abs(open_qty),
                        "价格": open_price,
                        "手续费": abs(open_qty) * open_comm,
                        "备注": note
                    })

                # 情况2：同向调整（加仓或减仓），且目标不为 0
                elif total_position != 0 and today_target_vol * total_position > 0:
                    diff = today_target_vol - total_position
                    if abs(today_target_vol) > abs(total_position):
                        # 加仓
                        add_qty = diff
                        position_stack.append((add_qty, open_price))
                        fee += abs(add_qty) * open_comm
                        total_position = today_target_vol
                        trades.append({
                            "日期": row.name if trade_dataframe.index.name is not None else i,
                            "方向": "买入" if add_qty > 0 else "卖出",
                            "手数": abs(add_qty),
                            "价格": open_price,
                            "手续费": abs(add_qty) * open_comm,
                            "备注": "加多" if add_qty > 0 else "加空"
                        })
                    elif abs(today_target_vol) < abs(total_position):
                        # 减仓：按 FIFO 平掉部分持仓
                        to_close = abs(diff)
                        remaining_to_close = to_close
                        while remaining_to_close > 0 and position_stack:
                            _trade_volume, _trade_cost = position_stack[0]
                            abs_lot = abs(_trade_volume)
                            if abs_lot <= remaining_to_close:
                                position_stack.pop(0)
                                if _trade_volume > 0:
                                    pnl = (open_price - prev_settle) * abs_lot
                                    direction = "卖出"
                                    note = "减多"
                                else:
                                    pnl = (prev_settle - open_price) * abs_lot
                                    direction = "买入"
                                    note = "减空"
                                realized_pnl += pnl
                                fee += abs_lot * close_comm
                                remaining_to_close -= abs_lot
                                total_position -= _trade_volume
                                trades.append({
                                    "日期": row.name if trade_dataframe.index.name is not None else i,
                                    "方向": direction,
                                    "手数": abs_lot,
                                    "价格": open_price,
                                    "手续费": abs_lot * close_comm,
                                    "备注": note
                                })
                            else:
                                if _trade_volume > 0:
                                    close_qty = remaining_to_close
                                    pnl = (open_price - prev_settle) * close_qty
                                    position_stack[0] = (_trade_volume - close_qty, _trade_cost)
                                    direction = "卖出"
                                    note = "减多"
                                else:
                                    close_qty = -remaining_to_close
                                    pnl = (prev_settle - open_price) * remaining_to_close
                                    position_stack[0] = (_trade_volume - close_qty, _trade_cost)
                                    direction = "买入"
                                    note = "减空"
                                realized_pnl += pnl
                                fee += remaining_to_close * close_comm
                                total_position -= close_qty
                                trades.append({
                                    "日期": row.name if trade_dataframe.index.name is not None else i,
                                    "方向": direction,
                                    "手数": remaining_to_close,
                                    "价格": open_price,
                                    "手续费": remaining_to_close * close_comm,
                                    "备注": note
                                })
                                remaining_to_close = 0

        # ---------- 计算当日浮动盈亏与权益 ----------
        if total_position != 0:
            # 以当日结算价统计全部的持仓标的
            total_cost = sum(lot * cost for lot, cost in position_stack)
            float_pnl = round((settle_price * total_position) - total_cost, round_precision)  # 今日盈亏
        else:
            float_pnl = 0.0

        # cumulative_pnl = prev_equity + realized_pnl - fee + (float_pnl - prev_float_pnl)
        cumulative_pnl = round(prev_equity + realized_pnl - fee + float_pnl, round_precision)
        # daily_pnl = equity - prev_equity

        # ---------- 记录每日统计 ----------
        result_rows.append({
            "日期": row.name if trade_dataframe.index.name is not None else i,
            "交易信号": signal,
            "持仓调整量": trade_volume,
            "当日持仓": total_position,
            "当日收益": deepcopy(round(cumulative_pnl - prev_equity, round_precision)),
            "累计收益": deepcopy(round(cumulative_pnl, round_precision)),
        })

        # ---------- 更新状态 ----------
        # prev_float_pnl = deepcopy(float_pnl)
        prev_settle = deepcopy(settle_price)
        prev_equity = deepcopy(cumulative_pnl)

    # 构造 DataFrame
    result_df = pd.DataFrame(result_rows)
    trade_df = pd.DataFrame(trades)

    return result_df, trade_df


def main():
    """回测逻辑
    当天出信号，第二天交易，target_signal表示
    """
    df = pd.DataFrame({
        "date": pd.date_range("2022-01-01", freq="D", periods=11),
        "open": [100.59, 100.48, 99.67, 99.86, 98.99, 99.51, 99.31, 100.17, 98.79, 99.03, 100.83],
        "settle": [100.3, 100.95, 100.04, 100.58, 100.5, 100.81, 100.3, 100.88, 100.13, 100.58, 100.25],
        "target_signal": [1, 2, 2, 2, 2, 0, -2, -1, 0, 2, 1],
    })
    df.set_index("date", inplace=True)
    df.index = pd.to_datetime(df.index)
    result, trade_df = futures_backtest_fifo(
        df, open_col="open", settle_col="settle", signal_col="target_signal", open_comm=2, close_comm=2, multiplier=10000)
    print(result)


if __name__ == '__main__':
    main()