165 lines
8.6 KiB
Python
165 lines
8.6 KiB
Python
"""
|
||
Options Lab — "was this option well priced between two dates?" A fine-grained pricing audit
|
||
for a single contract on an instrument, reusing the same `as_of` historical reconstruction
|
||
built for the Portfolio retrospective comparison (services.strategy_comparison /
|
||
services.option_chain's as_of param — see project memory).
|
||
|
||
The strike is chosen WITH HINDSIGHT: the one closest to where the underlying actually ended
|
||
up by `date_b` ("as if we'd guessed the strike correctly"). That's deliberate — a contract
|
||
near-the-money-at-the-outcome is the one whose value is most sensitive to the realized move,
|
||
which makes it the most revealing lens on whether the volatility priced in at `date_a` was
|
||
actually justified by what happened, rather than picking an arbitrary strike that stayed
|
||
deep OTM/ITM the whole time and would show almost nothing either way.
|
||
|
||
The price move for each leg is decomposed via its own real Greeks at date_a — Delta×Δspot +
|
||
Theta×elapsed_days + Vega×ΔIV — the "explained" move; whatever's left over ("residual") is
|
||
what a pure Black-Scholes/Greeks story doesn't account for (gamma curvature, skew shift,
|
||
liquidity/spread noise, or a genuine pricing anomaly).
|
||
"""
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
|
||
def _realized_vol(yf_ticker: str, date_a: str, date_b: str) -> Optional[float]:
|
||
"""Annualized realized vol of the underlying's own daily closes over [date_a, date_b] —
|
||
compared against the option's implied vol at date_a to answer "was IV a good forecast
|
||
of what actually happened," the classic IV-vs-RV question."""
|
||
import numpy as np
|
||
from services.data_fetcher import get_historical
|
||
|
||
hist = get_historical(yf_ticker, start=date_a, end=date_b, interval="1d")
|
||
closes = [h["close"] for h in hist if h.get("close")]
|
||
if len(closes) < 3:
|
||
return None
|
||
log_returns = np.diff(np.log(closes))
|
||
if len(log_returns) < 2:
|
||
return None
|
||
return float(np.std(log_returns, ddof=1) * np.sqrt(252))
|
||
|
||
|
||
def _leg_quote(rows: List[Dict[str, Any]], expiry_date: str, strike: float, option_type: str) -> Optional[Dict[str, Any]]:
|
||
for r in rows:
|
||
if r.get("expiry_date") == expiry_date and abs((r.get("strike") or -1e9) - strike) < 1e-6 and r.get("option_type") == option_type:
|
||
bid, ask = r.get("bid") or 0.0, r.get("ask") or 0.0
|
||
mid = r.get("mid") or (round((bid + ask) / 2, 6) if (bid > 0 and ask > 0) else 0.0)
|
||
vol_pct = r.get("volatility_pct")
|
||
return {"bid": bid, "ask": ask, "mid": mid, "iv": (float(vol_pct) / 100.0) if vol_pct is not None else None}
|
||
return None
|
||
|
||
|
||
def analyze_option_pricing(ticker: str, date_a: str, date_b: str, target_dte: Optional[int] = None) -> Dict[str, Any]:
|
||
from datetime import datetime
|
||
from services.database import get_saxo_option_symbol_for_ticker, get_snapshot_rows_asof
|
||
from services.option_chain import get_chain_slice
|
||
from services.options_pricer import black_scholes
|
||
|
||
saxo_symbol = get_saxo_option_symbol_for_ticker(ticker)
|
||
if not saxo_symbol:
|
||
return {"available": False, "reason": f"'{ticker}' n'est pas lié à un chain Saxo (Config → Instruments Watchlist)."}
|
||
if date_b <= date_a:
|
||
return {"available": False, "reason": "La date de fin doit être postérieure à la date de départ."}
|
||
|
||
# Same hindsight principle as the strike selection below: by default, target the expiry
|
||
# closest to date_b (not an arbitrary fixed DTE) — so the contract is still evaluated
|
||
# right around the moment we actually care about, rather than risking one that's been
|
||
# expired for weeks by date_b just because target_dte was picked independently of it.
|
||
# An explicit target_dte still overrides this, e.g. to deliberately look at a
|
||
# longer-dated contract than the comparison window itself.
|
||
if target_dte is None:
|
||
target_dte = (
|
||
datetime.strptime(date_b[:10], "%Y-%m-%d").date() - datetime.strptime(date_a[:10], "%Y-%m-%d").date()
|
||
).days
|
||
|
||
try:
|
||
chain_a = get_chain_slice(saxo_symbol, target_days=target_dte, n_expiries=1, as_of=date_a)
|
||
except ValueError as e:
|
||
return {"available": False, "reason": f"Pas d'historique Saxo au {date_a} : {e}"}
|
||
|
||
expiry = chain_a["expiries"][0]
|
||
expiry_date = expiry["expiry_date"]
|
||
spot_a = chain_a["spot"]
|
||
strikes_a = sorted({row["strike"] for row in expiry["calls"]} | {row["strike"] for row in expiry["puts"]})
|
||
if not strikes_a or spot_a is None:
|
||
return {"available": False, "reason": "Aucun strike/spot exploitable dans le chain à cette date."}
|
||
|
||
rows_b = get_snapshot_rows_asof(saxo_symbol, date_b)
|
||
if not rows_b:
|
||
return {"available": False, "reason": f"Pas d'historique Saxo au {date_b}."}
|
||
spot_b = next((r["spot"] for r in rows_b if r.get("spot") is not None), None)
|
||
if spot_b is None:
|
||
return {"available": False, "reason": "Spot manquant dans l'historique Saxo à la date de fin."}
|
||
|
||
# "As if we'd guessed the strike" — closest to where the underlying actually ended up.
|
||
chosen_strike = min(strikes_a, key=lambda k: abs(k - spot_b))
|
||
|
||
rows_a = get_snapshot_rows_asof(saxo_symbol, date_a)
|
||
d_a = datetime.strptime(date_a[:10], "%Y-%m-%d").date()
|
||
d_b = datetime.strptime(date_b[:10], "%Y-%m-%d").date()
|
||
d_exp = datetime.strptime(expiry_date[:10], "%Y-%m-%d").date()
|
||
elapsed_days = (d_b - d_a).days
|
||
days_to_expiry_a = (d_exp - d_a).days
|
||
days_to_expiry_b = (d_exp - d_b).days
|
||
expired_by_b = days_to_expiry_b <= 0
|
||
|
||
r = 0.05
|
||
legs_out: Dict[str, Any] = {}
|
||
for opt_type in ("call", "put"):
|
||
q_a = _leg_quote(rows_a, expiry_date, chosen_strike, opt_type)
|
||
if not q_a or q_a["mid"] <= 0 or q_a["iv"] is None:
|
||
legs_out[opt_type] = {"available": False}
|
||
continue
|
||
|
||
greeks_a = black_scholes(spot_a, chosen_strike, max(days_to_expiry_a, 1) / 365, r, q_a["iv"], opt_type)
|
||
intrinsic_a = max(0.0, spot_a - chosen_strike) if opt_type == "call" else max(0.0, chosen_strike - spot_a)
|
||
time_value_a = q_a["mid"] - intrinsic_a
|
||
|
||
if expired_by_b:
|
||
price_b = max(0.0, spot_b - chosen_strike) if opt_type == "call" else max(0.0, chosen_strike - spot_b)
|
||
iv_b, intrinsic_b, time_value_b = None, price_b, 0.0
|
||
else:
|
||
q_b = _leg_quote(rows_b, expiry_date, chosen_strike, opt_type)
|
||
if not q_b or q_b["mid"] <= 0:
|
||
legs_out[opt_type] = {"available": False}
|
||
continue
|
||
price_b, iv_b = q_b["mid"], q_b["iv"]
|
||
intrinsic_b = max(0.0, spot_b - chosen_strike) if opt_type == "call" else max(0.0, chosen_strike - spot_b)
|
||
time_value_b = price_b - intrinsic_b
|
||
|
||
actual_change = price_b - q_a["mid"]
|
||
delta_pnl = greeks_a["delta"] * (spot_b - spot_a)
|
||
theta_pnl = greeks_a["theta"] * elapsed_days
|
||
vega_pnl = greeks_a["vega"] * ((iv_b - q_a["iv"]) * 100) if iv_b is not None else 0.0
|
||
explained = delta_pnl + theta_pnl + vega_pnl
|
||
|
||
legs_out[opt_type] = {
|
||
"available": True,
|
||
"price_a": round(q_a["mid"], 4), "price_b": round(price_b, 4), "actual_change": round(actual_change, 4),
|
||
"iv_a": round(q_a["iv"], 4), "iv_b": round(iv_b, 4) if iv_b is not None else None,
|
||
"intrinsic_a": round(intrinsic_a, 4), "time_value_a": round(time_value_a, 4),
|
||
"intrinsic_b": round(intrinsic_b, 4), "time_value_b": round(time_value_b, 4),
|
||
"greeks_a": {k: greeks_a[k] for k in ("delta", "gamma", "theta", "vega")},
|
||
"attribution": {
|
||
"delta_pnl": round(delta_pnl, 4), "theta_pnl": round(theta_pnl, 4), "vega_pnl": round(vega_pnl, 4),
|
||
"explained": round(explained, 4), "residual": round(actual_change - explained, 4),
|
||
},
|
||
}
|
||
|
||
realized_vol = _realized_vol(ticker, date_a, date_b)
|
||
iv_a_ref = next((legs_out[t]["iv_a"] for t in ("call", "put") if legs_out.get(t, {}).get("available")), None)
|
||
|
||
return {
|
||
"available": True,
|
||
"ticker": ticker, "saxo_symbol": saxo_symbol,
|
||
"date_a": date_a, "date_b": date_b, "elapsed_days": elapsed_days,
|
||
"target_dte_used": target_dte,
|
||
"expiry_date": expiry_date, "expired_by_date_b": expired_by_b,
|
||
"spot_a": round(spot_a, 6), "spot_b": round(spot_b, 6),
|
||
"spot_change_pct": round((spot_b - spot_a) / spot_a * 100, 2) if spot_a else None,
|
||
"chosen_strike": chosen_strike,
|
||
"iv_a": iv_a_ref,
|
||
"realized_vol": round(realized_vol, 4) if realized_vol is not None else None,
|
||
"vol_risk_premium": (
|
||
round(iv_a_ref - realized_vol, 4) if (iv_a_ref is not None and realized_vol is not None) else None
|
||
),
|
||
"legs": legs_out,
|
||
}
|