收集并对齐永续期货资金费率结算数据

代码《交易机器学习》

总结

本文档介绍一个流程，用于下载 Binance 官方 USD-M 永续合约资金费率记录、进行标准化，并缓存到列式文件中。流程会并行获取指定标的和日期范围的月度存档，将记录转换为 UTC 时间戳，统一结算字段，按标的和时间去重，并在保存前检查缺失值或无效值。加载器支持按需筛选标的和日期。

回测使用的资金费率记录会进一步限定为价格面板中的准确标的和时间戳键。这一点很重要，因为资金费率成本按结算时持有的头寸收取；运行所用日期窗口不同时，相关结算记录也会不同。如果没有结算记录与价格匹配，该流程会报错。本文档说明数据处理和对齐方式，而非交易信号或实证策略；它不提供表现证据。数据源是某一家交易所的 USD-M 永续合约，因此所述记录不能证明覆盖其他交易场所或衍生品。

核心观点

按标的和月份收集资金费率存档，再将其标准化为 UTC 结算记录。
该流程会按标的和时间戳对记录去重，并剔除无效费率或结算间隔。
回测应将资金费率结算数据与价格面板覆盖的准确键对齐。
资金费率成本取决于结算时持有的头寸，因此适用记录会随回测窗口变化。
所述数据源涵盖 Binance USD-M 永续合约资金费率数据，并非所有交易场所或工具。

# funding_data.py


```py
"""Official Binance USD-M perpetual funding-rate data for the crypto case study."""

from __future__ import annotations

import io
import zipfile
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import UTC, date, datetime
from pathlib import Path

import httpx
import polars as pl
import yaml

from data.exceptions import DataNotFoundError
from utils import ML4T_DATA_PATH
from utils.paths import get_case_study_dir

BASE_URL = "https://data.binance.vision/data/futures/um/monthly/fundingRate"
FUNDING_PATH = ML4T_DATA_PATH / "crypto" / "market" / "funding_rate.parquet"


def _month_keys(start_date: str, end_date: str) -> list[str]:
    start = datetime.strptime(start_date, "%Y-%m-%d")
    end = datetime.strptime(end_date, "%Y-%m-%d")
    keys = []
    year, month = start.year, start.month
    while (year, month) <= (end.year, end.month):
        keys.append(f"{year}-{month:02d}")
        month += 1
        if month == 13:
            year += 1
            month = 1
    return keys


def _fetch_month(symbol: str, month: str) -> tuple[pl.DataFrame | None, bool]:
    url = f"{BASE_URL}/{symbol}/{symbol}-fundingRate-{month}.zip"
    response = httpx.get(url, timeout=30, follow_redirects=True)
    if response.status_code == 404:
        return None, True
    response.raise_for_status()
    with zipfile.ZipFile(io.BytesIO(response.content)) as archive:
        names = archive.namelist()
        if len(names) != 1:
            raise ValueError(f"Expected one CSV in {url}, found {names}")
        frame = pl.read_csv(io.BytesIO(archive.read(names[0]))).with_columns(
            pl.lit(symbol).alias("symbol")
        )
    required = {"calc_time", "funding_interval_hours", "last_funding_rate", "symbol"}
    missing = required - set(frame.columns)
    if missing:
        raise ValueError(f"Funding archive schema changed for {url}: missing {sorted(missing)}")
    return frame, False


def _normalize_funding(frames: list[pl.DataFrame]) -> pl.DataFrame:
    funding = (
        pl.concat(frames, how="diagonal_relaxed")
        .with_columns(
            pl.from_epoch("calc_time", time_unit="ms")
            .dt.replace_time_zone("UTC")
            .cast(pl.Datetime("ms", "UTC"))
            .dt.truncate("1h")
            .alias("timestamp"),
            pl.col("funding_interval_hours").cast(pl.Int16),
            pl.col("last_funding_rate").cast(pl.Float64).alias("funding_rate"),
        )
        .select("timestamp", "symbol", "funding_interval_hours", "funding_rate")
        .unique(["timestamp", "symbol"], maintain_order=False)
        .sort("timestamp", "symbol")
    )
    if funding.is_empty():
        raise ValueError("Funding archive produced no rows")
    if funding.select(pl.struct("timestamp", "symbol").is_duplicated().any()).item():
        raise ValueError("Funding archive contains duplicate timestamp-symbol keys")
    invalid = funding.filter(
        pl.col("funding_rate").is_nan()
        | pl.col("funding_rate").is_infinite()
        | (pl.col("funding_interval_hours") <= 0)
        | (pl.col("funding_interval_hours") > 8)
    )
    if invalid.height:
        raise ValueError(f"Funding archive contains {invalid.height} invalid rows")
    return funding


def download_funding_rates(
    symbols: list[str],
    start_date: str,
    end_date: str,
    *,
    output_path: Path = FUNDING_PATH,
    max_workers: int = 12,
) -> pl.DataFrame:
    """Download monthly official funding settlements and atomically cache them."""
    tasks = [(symbol, month) for symbol in symbols for month in _month_keys(start_date, end_date)]
    frames: list[pl.DataFrame] = []
    missing = 0
    print(f"Downloading {len(tasks):,} Binance funding-rate archives", flush=True)
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {
            pool.submit(_fetch_month, symbol, month): (symbol, month) for symbol, month in tasks
        }
        for index, future in enumerate(as_completed(futures), start=1):
            frame, not_listed = future.result()
            if frame is not None:
                frames.append(frame)
            elif not_listed:
                missing += 1
            if index % 100 == 0 or index == len(tasks):
                print(
                    f"  completed {index:,}/{len(tasks):,}; files={len(frames):,}; "
                    f"unavailable={missing:,}",
                    flush=True,
                )
    funding = _normalize_funding(frames)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    temporary_path = output_path.with_suffix(".tmp.parquet")
    funding.write_parquet(temporary_path)
    temporary_path.replace(output_path)
    print(
        f"Saved {len(funding):,} settlements for {funding['symbol'].n_unique()} symbols "
        f"to {output_path}",
        flush=True,
    )
    return funding


def load_funding_rates(
    *,
    symbols: list[str] | None = None,
    start_date: str | None = None,
    end_date: str | None = None,
    path: Path = FUNDING_PATH,
) -> pl.DataFrame:
    """Load official funding settlements with optional symbol and date filters."""
    if not path.exists():
        raise DataNotFoundError(
            dataset_name="Binance USD-M Perpetual Funding Rates",
            path=path,
            download_script="case_studies/crypto_perps_funding/funding_data.py",
            readme="case_studies/crypto_perps_funding/README.md",
        )
    funding = pl.read_parquet(path)
    if symbols:
        funding = funding.filter(pl.col("symbol").is_in(symbols))
    if start_date:
        funding = funding.filter(pl.col("timestamp").dt.date() >= pl.lit(start_date).str.to_date())
    if end_date:
        funding = funding.filter(pl.col("timestamp").dt.date() <= pl.lit(end_date).str.to_date())
    return funding.sort("timestamp", "symbol")


def funding_rates_for_prices(prices: pl.DataFrame) -> pl.DataFrame:
    """Official funding settlements restricted to the exact keys ``prices`` covers.

    The backtest charges funding against the position held at each settlement, so the
    settlements a run is identified by have to be the ones its own price panel spans. A
    holdout run slices a different window from the validation run it inherits its
    configuration from, which is why this is derived from the frame rather than carried
    across with the rest of the specification.
    """

    def _day(value: object) -> str:
        if isinstance(value, datetime):
            return value.date().isoformat()
        if isinstance(value, date):
            return value.isoformat()
        raise TypeError(f"expected a date-like timestamp, got {type(value).__name__}")

    timestamp_dtype = prices.schema["timestamp"]
    price_keys = prices.select("symbol", "timestamp").unique()
    symbols = price_keys.get_column("symbol").unique().to_list()
    funding = load_funding_rates(
        symbols=symbols,
        start_date=_day(price_keys.get_column("timestamp").min()),
        end_date=_day(price_keys.get_column("timestamp").max()),
    )
    funding = funding.with_columns(
        pl.col("symbol").cast(price_keys.schema["symbol"]),
        pl.col("timestamp").cast(timestamp_dtype),
    ).join(price_keys, on=["symbol", "timestamp"], how="semi")
    if funding.is_empty():
        raise ValueError("crypto backtest resolved no official funding settlements")
    return funding.sort("timestamp", "symbol")


def main() -> None:
    case_dir = get_case_study_dir("crypto_perps_funding")
    setup = yaml.safe_load((case_dir / "config" / "setup.yaml").read_text())
    download_funding_rates(
        list(setup["universe"]["symbols"]),
        "2020-01-01",
        "2025-12-31",
    )
    print(f"Completed at {datetime.now(UTC).isoformat()}", flush=True)


if __name__ == "__main__":
    main()

```

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

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

收集并对齐永续期货资金费率结算数据

总结

核心观点

标签