Construção de retornos de opções do mesmo contrato e trajetórias de hedge delta
Código Machine Learning for Trading
Resumo
Este utilitário cria artefatos de rótulos para straddles de opções do S&P 500, usando o mesmo símbolo, preço de exercício e vencimento na entrada e na saída. Ele alinha as datas das características às sessões de mercado subsequentes, cria datas de saída para cinco e dez sessões e associa preços de opções de compra e venda para calcular os retornos do straddle. Também registra cotações de compra e venda e pode persistir o painel de preços de origem. O escopo do cache é armazenado junto aos artefatos para que execuções com símbolos reduzidos ou datas abreviadas não sejam confundidas com cobertura completa.
Uma trajetória de hedge separada acompanha os deltas das opções de compra e venda e os preços do ativo subjacente ao longo dos dias de manutenção. O P&L acumulado do hedge aplica o delta combinado do fechamento anterior à variação seguinte do ativo subjacente e conta os dias de hedge observados, para que trajetórias incompletas possam ser identificadas em vez de tratadas como totalmente protegidas. Os deslocamentos de sessão vêm do painel inteiro, mesmo quando o conjunto de símbolos é reduzido, preservando o horizonte pretendido. O trecho descreve a construção de rótulos e o tratamento dos dados, não uma estratégia de opções testada nem resultados de desempenho. Os rótulos podem estar indisponíveis quando faltam cotações ou sessões posteriores, e a trajetória de hedge trata explicitamente os dias de contrato não observados como incompletos.
Ideias principais
- Mantenha símbolo, preço de exercício e vencimento fixos ao associar preços de opções entre as datas de entrada e saída.
- Conte os horizontes de manutenção em sessões de mercado usando o calendário completo do painel, mesmo quando o universo de símbolos é reduzido.
- Calcule os retornos do straddle a partir dos preços médios combinados das opções de compra e venda na entrada e na saída.
- Acumule o P&L do hedge delta usando o delta combinado das opções no fechamento anterior e a variação seguinte do preço do ativo subjacente.
- Acompanhe a contagem de observações do hedge para que cotações de contrato ausentes não pareçam indicar um hedge completo.
Tags
Texto completo
# _label_artifacts.py
```py
"""Build same-contract label artifacts for the S&P 500 options case study."""
from __future__ import annotations
import json
from pathlib import Path
import polars as pl
from data import load_sp500_options_straddles, load_sp500_options_straddles_raw
from utils.data_quality import top_entities
from utils.paths import get_case_study_dir
HORIZONS = (5, 10)
MAX_HOLDING = max(HORIZONS)
JOIN_KEYS = ["symbol", "strike", "expiration"]
def ensure_label_artifacts(
*,
case_study_id: str = "sp500_options",
max_symbols: int = 0,
start_date: str | None = None,
force_rebuild: bool = False,
save_prices: bool = False,
) -> dict[str, Path]:
"""Ensure same-contract option artifacts exist for label construction."""
case_dir = get_case_study_dir(case_study_id)
labels_dir = case_dir / "labels"
labels_dir.mkdir(parents=True, exist_ok=True)
contract_returns_path = labels_dir / "contract_returns.parquet"
hedge_path_path = labels_dir / "hedge_path.parquet"
prices_path = labels_dir / "prices.parquet"
required = [contract_returns_path, hedge_path_path]
# A cached artifact was built over whatever window the run that wrote it saw, so the
# scope is recorded beside it and the cache is only reused for the same request.
# Without that, a narrower run returns the full panel and leaves the reduction
# silently unapplied, and the reduced files a narrower run writes are then accepted
# by the next default run as if they covered everything.
scope_path = labels_dir / "contract_returns.scope.json"
scope = {"max_symbols": max_symbols, "start_date": start_date}
cached_scope = json.loads(scope_path.read_text()) if scope_path.exists() else None
if not force_rebuild and cached_scope == scope and all(path.exists() for path in required):
return {
"contract_returns": contract_returns_path,
"hedge_path": hedge_path_path,
"prices": prices_path,
}
straddles = load_sp500_options_straddles()
if start_date is not None:
straddles = straddles.filter(pl.col("timestamp") >= pl.lit(start_date).str.to_date())
# Horizons are counted in market sessions, so the calendar comes from the whole panel
# and `max_symbols` thins only the entries it is applied to. Deriving the offsets from
# a symbol subset instead would drop every session none of those symbols was quoted
# on, and the exit dates would then be that many sessions further out than declared.
trading_dates = straddles["timestamp"].unique().sort().to_list()
entry_rows = straddles
if max_symbols > 0:
# `top_entities` and not a local sort: it breaks the tie on the symbol, and this
# reduction has to agree with the one the feature and modelling stages make on the
# same panel. A symbol the labels kept and the features dropped joins to nulls.
entry_rows = straddles.filter(pl.col("symbol").is_in(top_entities(straddles, max_symbols)))
# A signal date qualifies once the panel has a session to enter on; how far past
# that the panel has to run is a property of each horizon, so every offset column
# runs off the end of the panel as a null rather than shortening the frame. Sizing
# one frame for the longest horizon instead trims the five-session labels by the
# ten-session one, and drops signal dates from the hold-to-expiry label, which
# needs only an entry price and an expiration.
def _shifted(step: int) -> list[object]:
return trading_dates[step:] + [None] * step
offset_data = {"feature_date": trading_dates, "entry_date": _shifted(1)}
for horizon in HORIZONS:
offset_data[f"exit_{horizon}d_date"] = _shifted(1 + horizon)
for day in range(MAX_HOLDING + 1):
offset_data[f"path_date_{day}"] = _shifted(1 + day)
date_offsets = pl.DataFrame(offset_data).drop_nulls("entry_date")
entries = (
entry_rows.select(["timestamp", "symbol", "strike", "expiration"])
.join(date_offsets.rename({"feature_date": "timestamp"}), on="timestamp", how="inner")
.rename({"timestamp": "feature_date"})
)
contracts = entries.select("symbol", "strike", "expiration").unique()
raw_lookup = (
load_sp500_options_straddles_raw(lazy=True)
.join(contracts.lazy(), on=JOIN_KEYS, how="semi")
.filter(pl.col("bid") >= 0.01)
.select(
[
pl.col("timestamp").alias("date"),
"symbol",
"strike",
"expiration",
"call_put",
"mid_price",
"bid",
"ask",
"delta",
"underlying_price",
"days_to_maturity",
]
)
.collect()
)
raw_calls = raw_lookup.filter(pl.col("call_put") == "C")
raw_puts = raw_lookup.filter(pl.col("call_put") == "P")
result = entries
result = result.join(
_build_price_lookup(raw_calls, "entry_date", "entry_call"),
on=["entry_date"] + JOIN_KEYS,
how="left",
)
result = result.join(
_build_price_lookup(raw_puts, "entry_date", "entry_put"),
on=["entry_date"] + JOIN_KEYS,
how="left",
)
for horizon in HORIZONS:
date_col = f"exit_{horizon}d_date"
result = result.join(
_build_price_lookup(raw_calls, date_col, f"exit_call_{horizon}d"),
on=[date_col] + JOIN_KEYS,
how="left",
)
result = result.join(
_build_price_lookup(raw_puts, date_col, f"exit_put_{horizon}d"),
on=[date_col] + JOIN_KEYS,
how="left",
)
result = result.with_columns(
(pl.col("entry_call_mid") + pl.col("entry_put_mid")).alias("entry_straddle_mid"),
)
for horizon in HORIZONS:
exit_straddle = f"exit_straddle_mid_{horizon}d"
result = result.with_columns(
(pl.col(f"exit_call_{horizon}d_mid") + pl.col(f"exit_put_{horizon}d_mid")).alias(
exit_straddle
),
pl.col(f"exit_call_{horizon}d_mid").is_not_null().alias(f"exit_found_{horizon}d"),
)
result = result.with_columns(
(
(pl.col("entry_straddle_mid") - pl.col(exit_straddle))
/ pl.col("entry_straddle_mid")
).alias(f"fwd_ret_{horizon}d"),
)
hedge_needs = (
entries.select(
["feature_date", "symbol", "strike", "expiration"]
+ [f"path_date_{d}" for d in range(MAX_HOLDING + 1)]
)
.unpivot(
[f"path_date_{d}" for d in range(MAX_HOLDING + 1)],
index=["feature_date", "symbol", "strike", "expiration"],
variable_name="holding_day_str",
value_name="holding_date",
)
.with_columns(
pl.col("holding_day_str").str.extract(r"(\d+)").cast(pl.Int32).alias("holding_day")
)
.drop("holding_day_str")
)
hedge_call = raw_calls.select(
[
pl.col("date").alias("holding_date"),
"symbol",
"strike",
"expiration",
pl.col("delta").alias("call_delta"),
]
)
hedge_put = raw_puts.select(
[
pl.col("date").alias("holding_date"),
"symbol",
"strike",
"expiration",
pl.col("delta").alias("put_delta"),
]
)
underlying_prices = raw_calls.select(
[pl.col("date").alias("holding_date"), "symbol", "strike", "expiration", "underlying_price"]
).unique(subset=["holding_date", "symbol", "strike", "expiration"])
hedge_path = (
hedge_needs.join(hedge_call, on=["holding_date"] + JOIN_KEYS, how="left")
.join(hedge_put, on=["holding_date"] + JOIN_KEYS, how="left")
.join(underlying_prices, on=["holding_date"] + JOIN_KEYS, how="left")
.with_columns((pl.col("call_delta") + pl.col("put_delta")).alias("instr_delta"))
)
return_cols = (
["feature_date", "symbol", "strike", "expiration", "entry_date"]
+ [f"exit_{horizon}d_date" for horizon in HORIZONS]
+ ["entry_call_mid", "entry_put_mid", "entry_straddle_mid"]
+ ["entry_call_bid", "entry_call_ask", "entry_put_bid", "entry_put_ask"]
)
for horizon in HORIZONS:
return_cols += [
f"exit_call_{horizon}d_mid",
f"exit_put_{horizon}d_mid",
f"exit_straddle_mid_{horizon}d",
f"fwd_ret_{horizon}d",
f"exit_found_{horizon}d",
f"exit_call_{horizon}d_bid",
f"exit_call_{horizon}d_ask",
f"exit_put_{horizon}d_bid",
f"exit_put_{horizon}d_ask",
]
contract_returns = result.select(return_cols)
contract_returns.write_parquet(contract_returns_path)
hedge_path_out = hedge_path.select(
[
"feature_date",
"symbol",
"strike",
"expiration",
"holding_day",
"holding_date",
"call_delta",
"put_delta",
"instr_delta",
"underlying_price",
]
).sort(["symbol", "feature_date", "holding_day"])
hedge_path_out.write_parquet(hedge_path_path)
scope_path.write_text(json.dumps(scope, sort_keys=True) + "\n")
if save_prices:
straddles.write_parquet(prices_path)
return {
"contract_returns": contract_returns_path,
"hedge_path": hedge_path_path,
"prices": prices_path,
}
def accrued_hedge_pnl(
hedge_path: pl.DataFrame, horizons: tuple[int, ...] = HORIZONS
) -> pl.DataFrame:
"""Hedge P&L accrued over each horizon, with the number of days it was observed on.
The hedge is rebalanced at each close, so the P&L on holding day ``d`` is the delta
set on day ``d-1`` applied to that day's move in the underlying. A day the contract
was not quoted on contributes nothing and is *counted*: a sum over a path with holes
is a partial hedge, and the count is what lets the caller null the label rather than
present it as fully hedged.
Summed in day order. An unordered parallel sum re-associates the floating point, which
moves the label's last bit and its content digest from one run to the next.
"""
cohort = ["symbol", "feature_date"]
move = pl.col("underlying_price") - pl.col("underlying_price").shift(1).over(cohort)
daily = hedge_path.sort([*cohort, "holding_day"]).with_columns(
(pl.col("instr_delta").shift(1).over(cohort) * move).alias("daily_pnl")
)
accrued = daily.group_by(cohort).agg(
*[
expr
for horizon in horizons
for expr in (
pl.col("daily_pnl")
.filter(pl.col("holding_day").is_between(1, horizon))
.sort_by(pl.col("holding_day").filter(pl.col("holding_day").is_between(1, horizon)))
.sum()
.alias(f"hedge_pnl_{horizon}d"),
pl.col("daily_pnl")
.filter(pl.col("holding_day").is_between(1, horizon))
.is_not_null()
.sum()
.alias(f"hedge_days_{horizon}d"),
)
]
)
return accrued.rename({"feature_date": "timestamp"})
def summarize_label_artifacts(
*,
case_study_id: str = "sp500_options",
) -> dict[str, int]:
"""Return row counts for persisted same-contract artifacts."""
case_dir = get_case_study_dir(case_study_id)
labels_dir = case_dir / "labels"
summary: dict[str, int] = {}
for name in ("contract_returns", "hedge_path", "prices"):
path = labels_dir / f"{name}.parquet"
if path.exists():
summary[name] = int(pl.scan_parquet(path).select(pl.len()).collect().item())
return summary
def _build_price_lookup(raw_leg: pl.DataFrame, date_alias: str, prefix: str) -> pl.DataFrame:
return raw_leg.select(
[
pl.col("date").alias(date_alias),
"symbol",
"strike",
"expiration",
pl.col("mid_price").alias(f"{prefix}_mid"),
pl.col("bid").alias(f"{prefix}_bid"),
pl.col("ask").alias(f"{prefix}_ask"),
]
)
```Exibido na íntegra, com atribuição conforme a licença da fonte. Licença: MIT
Este resumo foi escrito pelo agente de pesquisa da Stratmill com base no original; não é uma cópia da fonte.