对齐加密货币永续合约、溢价与链上数据
代码 《交易机器学习》
总结
本文档介绍加密货币永续期货及相关链上序列的数据访问和对齐约定。文中说明，溢价指数柱线的时间戳记录其开盘时间：八小时柱线记录该区间末尾资金费结算前的溢价，因此应与随后发生的结算配对。可将小时级永续合约 OHLCV 重采样为与资金费时间对齐的八小时柱线，分别取首个开盘价、最高价、最低价、最后收盘价和成交量总和，再按交易对和时间戳与溢价数据连接。
数据加载器还支持按交易对和日期筛选、限制交易对集合，以及读取历史 DeFi TVL 或每日币种价格和成交量序列。实际启示是，研究资金费策略时，时间戳含义很重要：柱线若与错误的结算时点配对，测得的溢价关系可能发生偏移。本文档介绍数据集格式和获取约定，并非交易策略或实证结果。数据是否可用取决于本地文件；文中说明 CoinGecko 数据源的免费层历史窗口有限。
核心观点
- 八小时溢价柱线的时间戳标记区间开盘时点，对应区间末尾的资金费结算。
- 分析资金费时，应将溢价柱线与时间戳之后一个区间的结算配对。
- 小时级 OHLCV 重采样时，取首个开盘价、最高价、最低价、最后收盘价和成交量总和。
- 如果永续合约柱线和溢价观测的交易对及开盘时间戳一致，就可以直接连接。
- 本文档提供数据加载约定，并未证明任何交易策略能够盈利。
标签
全文
# loader.py
```py
"""Crypto loaders: market (OHLCV, premium index) and on-chain (DefiLlama TVL, CoinGecko)."""
from typing import Literal
import polars as pl
from data.exceptions import DataNotFoundError
from utils import ML4T_DATA_PATH
from utils.data_quality import apply_max_symbols
def list_crypto_perps() -> list[str]:
"""List perpetual-futures symbols available in the local data store.
Returns:
Sorted list of Binance USDT perps (e.g., ``["AAVEUSDT", ..., "XRPUSDT"]``).
Raises:
DataNotFoundError: If ``crypto/perps_1h.parquet`` is missing.
Example:
>>> list_crypto_perps()[:3]
['AAVEUSDT', 'ADAUSDT', 'APTUSDT']
"""
path = ML4T_DATA_PATH / "crypto" / "market" / "perps_1h.parquet"
if not path.exists():
raise DataNotFoundError(
dataset_name="Crypto Perpetuals OHLCV",
path=path,
download_script="data/crypto/market/download.py",
readme="data/crypto/README.md",
)
return pl.scan_parquet(path).select("symbol").unique().collect().to_series().sort().to_list()
def load_crypto_premium(
frequency: Literal["1h", "8h"] = "8h",
symbols: list[str] | None = None,
start_date: str | None = None,
end_date: str | None = None,
max_symbols: int = 0,
) -> pl.DataFrame:
"""Load crypto premium index for funding rate arbitrage case study.
Args:
frequency: Data frequency. "8h" gives one bar per funding interval. Default is "8h".
symbols: Optional list of symbols to filter (e.g., ["BTCUSDT", "ETHUSDT"])
start_date: Optional start date (YYYY-MM-DD format)
end_date: Optional end date (YYYY-MM-DD format)
max_symbols: Limit to N random symbols (0 = all). Seed-deterministic.
Returns:
DataFrame with columns: timestamp, symbol, premium_index_open/high/low/close.
`timestamp` is the bar's opening time, as Binance stamps klines, so an 8h bar
stamped 00:00 spans 00:00-08:00 and its close is the premium entering the 08:00
funding settlement. Pair a bar with the settlement one interval later.
"""
filename = f"premium_index_{frequency}.parquet"
path = ML4T_DATA_PATH / "crypto" / "market" / filename
if not path.exists():
raise DataNotFoundError(
dataset_name="Crypto Premium Index",
path=path,
download_script="data/crypto/market/download.py --premium",
readme="data/crypto/README.md",
)
df = pl.read_parquet(path)
# Apply filters
if symbols:
df = df.filter(pl.col("symbol").is_in(symbols))
if start_date:
df = df.filter(pl.col("timestamp").dt.date() >= pl.lit(start_date).str.to_date())
if end_date:
df = df.filter(pl.col("timestamp").dt.date() <= pl.lit(end_date).str.to_date())
return apply_max_symbols(df, max_symbols)
def load_crypto_perps(
frequency: Literal["1h", "8h"] = "1h",
symbols: list[str] | None = None,
start_date: str | None = None,
end_date: str | None = None,
max_symbols: int = 0,
) -> pl.DataFrame:
"""Load crypto perpetual futures OHLCV data.
Args:
frequency: Data frequency. "1h" for raw hourly data, "8h" for funding-aligned
8-hour bars (00:00, 08:00, 16:00 UTC - standard funding settlement times).
symbols: Optional list of symbols to filter (e.g., ["BTCUSDT", "ETHUSDT"])
start_date: Optional start date (YYYY-MM-DD format)
end_date: Optional end date (YYYY-MM-DD format)
max_symbols: Limit to N random symbols (0 = all). Seed-deterministic.
Returns:
DataFrame with columns: timestamp, symbol, open, high, low, close, volume
"""
# Always load from 1h source
filename = "perps_1h.parquet"
path = ML4T_DATA_PATH / "crypto" / "market" / filename
if not path.exists():
raise DataNotFoundError(
dataset_name="Crypto Perpetuals OHLCV",
path=path,
download_script="data/crypto/market/download.py",
readme="data/crypto/README.md",
)
lf = pl.scan_parquet(path)
ts_type = lf.collect_schema()["timestamp"]
tz = getattr(ts_type, "time_zone", None)
def _ts_lit(d: str) -> pl.Expr:
e = pl.lit(d).str.to_datetime()
return e.dt.replace_time_zone(tz) if tz else e
# Apply filters before resampling (parquet pushdown via row-group pruning)
if symbols:
lf = lf.filter(pl.col("symbol").is_in(symbols))
if start_date:
lf = lf.filter(pl.col("timestamp") >= _ts_lit(start_date))
if end_date:
# Include the entire end_date for intraday
lf = lf.filter(pl.col("timestamp") < _ts_lit(end_date) + pl.duration(days=1))
df = lf.collect()
# Apply max_symbols before resampling
df = apply_max_symbols(df, max_symbols)
if frequency == "8h":
# Resample to 8H aligned with funding settlement times (00:00, 08:00, 16:00 UTC)
df = (
df.sort(["symbol", "timestamp"])
.group_by_dynamic(
"timestamp",
every="8h",
period="8h",
by="symbol",
closed="left",
label="left",
)
.agg(
pl.col("open").first(),
pl.col("high").max(),
pl.col("low").min(),
pl.col("close").last(),
pl.col("volume").sum(),
)
.sort(["symbol", "timestamp"])
)
# Join premium index data. Both sides are Binance klines stamped at the bar's
# opening time, so the timestamps line up directly.
premium_path = ML4T_DATA_PATH / "crypto" / "market" / "premium_index_8h.parquet"
if premium_path.exists():
premium = pl.read_parquet(premium_path)
if symbols:
premium = premium.filter(pl.col("symbol").is_in(df["symbol"].unique()))
df = df.join(premium, on=["symbol", "timestamp"], how="left")
return df
# --- On-chain / DeFi metrics ---
def load_defillama_chain_tvl(chain: str = "total") -> pl.DataFrame:
"""Load historical Total Value Locked (TVL) from DefiLlama.
Produced by `data/crypto/onchain/download.py`.
Args:
chain: "total" for aggregate DeFi TVL across all chains (default),
or a chain name like "Ethereum", "Solana", "BSC", "Arbitrum".
Matches the filename suffix (lowercased).
Returns:
DataFrame with `timestamp` (Date) and `tvl_usd` (float) columns.
"""
suffix = chain.lower()
path = ML4T_DATA_PATH / "crypto" / "onchain" / f"defillama_tvl_{suffix}.parquet"
if not path.exists():
chains_flag = "" if suffix == "total" else f" --chains {chain}"
raise DataNotFoundError(
dataset_name=f"DefiLlama TVL ({chain})",
path=path,
download_script=f"data/crypto/onchain/download.py --dataset defillama{chains_flag}",
readme="data/crypto/onchain/README.md",
)
return pl.read_parquet(path)
def load_coingecko_ohlcv(coin: str = "ethereum") -> pl.DataFrame:
"""Load daily prices/volume for one coin from CoinGecko.
Produced by `data/crypto/onchain/download.py --dataset coingecko`.
Free-tier window is 365 days; re-run the downloader to refresh.
Args:
coin: CoinGecko coin id (lowercase). Defaults to "ethereum".
Returns:
DataFrame with `timestamp` (Date), `price_usd`, `volume_usd`.
"""
path = ML4T_DATA_PATH / "crypto" / "onchain" / f"coingecko_{coin.lower()}.parquet"
if not path.exists():
raise DataNotFoundError(
dataset_name=f"CoinGecko OHLCV ({coin})",
path=path,
download_script=f"data/crypto/onchain/download.py --dataset coingecko --coins {coin}",
readme="data/crypto/onchain/README.md",
)
return pl.read_parquet(path)
```在遵守原作品许可的前提下，附作者信息全文展示。 许可协议: MIT
此摘要由 Stratmill 研究智能体根据原文撰写，并非原文副本。