永续期货资金费核算与回测

代码《交易机器学习》

总结

本文介绍一套账本，用于在回测中将资金现金流计入永续期货持仓。每个资金费时间戳，它都会使用持仓的带符号数量、当前标记价格、合约乘数（如有）以及资金费率计算现金调整。因此，正资金费率对多头和空头的影响相反。

账本会将时间戳标准化为 UTC，拒绝缺失或重复的结算键以及非有限费率，并防止同一时间戳被重复应用。它包装经纪商的时间更新，使结算在标记价格更新后进行，并报告累计资金费盈亏、事件数和结算数。要计算指标，所有提供的资金费率键都必须已到达引擎时间线。该方法依赖于未平仓头寸在结算时有有效标记价格；若缺少标记价格，就会报错。这是用于回测的会计基础设施，并非交易策略盈利的证据。

核心观点

资金现金流根据带符号的持仓规模、标记价值、乘数和资金费率计算。
经纪商更新该时间戳的标记价格后，再应用结算。
UTC 标准化和重复保护有助于避免遗漏或重复结算。
指标检查要求所有提供的结算时间戳都出现在引擎时间线上。
未平仓头寸缺少当前标记价格时，结算会失败。

# funding_backtest.py


```py
"""Funding-settlement accounting for perpetual-futures engine backtests."""

from __future__ import annotations

import math
from datetime import UTC, datetime
from functools import wraps
from typing import Any

import polars as pl


def _as_utc(value: datetime) -> datetime:
    return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)


class FundingSettlementLedger:
    """Apply position-signed funding during the engine's bar-time update."""

    def __init__(self, funding_rates: pl.DataFrame) -> None:
        required = {"symbol", "timestamp", "funding_rate"}
        missing = required - set(funding_rates.columns)
        if missing:
            raise ValueError(f"funding rates are missing columns: {sorted(missing)}")
        selected = funding_rates.select("symbol", "timestamp", "funding_rate")
        if selected.null_count().row(0) != (0, 0, 0):
            raise ValueError("funding settlements cannot contain null keys or rates")
        if selected.n_unique(["symbol", "timestamp"]) != selected.height:
            raise ValueError("funding settlement keys must be unique")

        self._rates: dict[datetime, dict[str, float]] = {}
        for row in selected.sort("timestamp", "symbol").iter_rows(named=True):
            rate = float(row["funding_rate"])
            if not math.isfinite(rate):
                raise ValueError("funding rates must be finite")
            self._rates.setdefault(_as_utc(row["timestamp"]), {})[str(row["symbol"])] = rate
        self._rate_count = selected.height
        self._settled_timestamps: set[datetime] = set()
        self._funding_pnl = 0.0
        self._funding_events = 0
        self._funding_settlements = 0
        self._installed = False

    def settle(self, timestamp: datetime, broker: Any) -> float:
        """Settle one timestamp exactly once against positions marked on that bar."""
        normalized = _as_utc(timestamp)
        if normalized in self._settled_timestamps:
            return 0.0
        rates = self._rates.get(normalized)
        if rates is None:
            return 0.0
        self._settled_timestamps.add(normalized)
        self._funding_settlements += len(rates)

        event_cash = 0.0
        for symbol, rate in rates.items():
            position = broker.positions.get(symbol)
            if position is None or float(position.quantity) == 0.0:
                continue
            mark = broker.get_mark_price(symbol, quantity=position.quantity)
            if mark is None:
                raise RuntimeError(f"funding settlement has no current mark for {symbol!r}")
            event_cash -= (
                float(position.quantity)
                * float(mark)
                * float(getattr(position, "multiplier", 1.0))
                * rate
            )
        if event_cash:
            broker.cash = float(broker.cash) + event_cash
            self._funding_pnl += event_cash
            self._funding_events += 1
        return event_cash

    def install(self, broker: Any) -> None:
        """Install settlement immediately after each engine mark update."""
        if self._installed:
            raise RuntimeError("funding settlement ledger is already installed")
        original_update_time = broker._update_time

        @wraps(original_update_time)
        def update_time_with_funding(timestamp, *args, **kwargs):
            result = original_update_time(timestamp, *args, **kwargs)
            self.settle(timestamp, broker)
            return result

        broker._update_time = update_time_with_funding
        self._installed = True

    def metrics(self) -> dict[str, float]:
        """Return cashflows actually presented to the engine timeline."""
        if self._funding_settlements != self._rate_count:
            raise RuntimeError(
                "funding settlement coverage is incomplete: "
                f"{self._funding_settlements}/{self._rate_count} keys reached the engine timeline"
            )
        return {
            "funding_pnl": self._funding_pnl,
            "funding_events": float(self._funding_events),
            "funding_settlements": float(self._funding_settlements),
        }

```

在遵守原作品许可的前提下，附作者信息全文展示。许可协议: MIT

此摘要由 Stratmill 研究智能体根据原文撰写，并非原文副本。

在回测中核算永续期货资金费

总结

核心观点

标签