Aligning Crypto Perpetual, Premium, and On-Chain Data
Code Machine Learning for Trading
Summary
This document describes data access and alignment conventions for crypto perpetual futures and related on-chain series. It explains that premium-index bars are timestamped at their opening time: an eight-hour bar records the premium leading into the funding settlement at the end of that interval, so it should be paired with the subsequent settlement. Hourly perpetual OHLCV can be resampled into funding-aligned eight-hour bars by taking the first open, maximum high, minimum low, last close, and summed volume, then joined to premium data on symbol and timestamp.
The loaders also support filtering by symbol and date, limiting the symbol set, and reading historical DeFi TVL or daily coin price and volume series. The practical lesson is that timestamp semantics matter when studying funding strategies: pairing a bar with the wrong settlement can shift the measured premium relationship. The document describes dataset formats and retrieval conventions rather than a trading strategy or empirical result. Data availability depends on local files, and the CoinGecko source is described as having a limited free-tier history window.
Key ideas
- Eight-hour premium bars are stamped at interval opening and correspond to the funding settlement at the interval’s end.
- For funding analysis, pair a premium bar with the settlement one interval after its timestamp.
- Resampling hourly OHLCV uses first open, maximum high, minimum low, last close, and summed volume.
- Perpetual bars and premium observations can be joined directly when their symbol and opening timestamps align.
- The document provides data loading conventions, not evidence that any trading strategy is profitable.
Tags
Full text
# 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)
```Shown in full with attribution under the source's licence. Licence: MIT
This summary was written by Stratmill's research agent from the original; it is not a copy of the source.