Comptabiliser le funding des contrats à terme perpétuels en backtest
Résumé
Ce document décrit un registre qui applique les flux de trésorerie de funding aux positions sur contrats à terme perpétuels lors d’un backtest. À chaque horodatage de funding, il utilise la quantité signée de la position, le prix de référence actuel, tout multiplicateur de contrat et le taux de funding pour calculer l’ajustement de trésorerie. Un taux de funding positif a donc des effets opposés sur les positions longues et courtes.
Le registre normalise les horodatages en UTC, rejette les clés de règlement manquantes ou en double ainsi que les taux non finis, et empêche l’application deux fois d’un même horodatage. Il enveloppe la mise à jour temporelle du courtier afin que le règlement suive la mise à jour du prix de référence, et indique le P&L cumulé du funding, le nombre d’événements et le nombre de règlements. Le calcul des métriques exige que chaque clé de taux de funding fournie ait atteint la chronologie du moteur. Cette approche suppose que des prix de référence valides soient disponibles au règlement pour les positions ouvertes ; elle déclenche une erreur si l’un d’eux manque. Il s’agit d’une infrastructure comptable pour les backtests, et non d’une preuve de rentabilité d’une stratégie de trading.
Idées clés
- Le flux de trésorerie du funding est calculé à partir de la taille signée de la position, de sa valeur au prix de référence, du multiplicateur et du taux de funding.
- Le règlement est appliqué après la mise à jour des prix de référence par le courtier pour l’horodatage concerné.
- La normalisation UTC et la protection contre les doublons aident à éviter les règlements manqués ou répétés.
- La vérification des métriques exige que tous les horodatages de règlement fournis figurent sur la chronologie du moteur.
- Une position ouverte sans prix de référence actuel entraîne l’échec du règlement.
Étiquettes
Texte intégral
# 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),
}
```Reproduit dans son intégralité avec attribution, conformément à la licence de la source. Licence: MIT
Ce résumé a été rédigé par l’agent de recherche de Stratmill à partir de la source originale ; il n’en est pas une copie.