94 lines
3.9 KiB
Python
94 lines
3.9 KiB
Python
"""
|
|
Option chain fetcher for the Strategy Builder — reads exclusively from our own
|
|
accumulated Saxo history (saxo_option_snapshots, refreshed every ~5 min by
|
|
services/saxo_scheduler.py). No live yfinance/Saxo call here: by the time the
|
|
Strategy Builder needs a chain, it's already been fetched and parsed correctly
|
|
by the periodic snapshot poller, so this is a fast, reliable DB read instead of
|
|
repeating the whole resolve/subscribe/parse dance per pricing request.
|
|
"""
|
|
from datetime import date, datetime
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
def get_chain_slice(symbol: str, target_days: int = 8, n_expiries: int = 3) -> Dict[str, Any]:
|
|
"""
|
|
Builds a chain slice from the latest accumulated Saxo snapshot rows for `symbol`
|
|
(services/database.get_latest_saxo_snapshot_rows). Returns the `n_expiries`
|
|
expirations closest to target_days, each with calls/puts rows shaped
|
|
{strike, bid, ask, mid, last, iv, open_interest, volume} — same shape regardless
|
|
of source, so vol_surface.py/strategy_engine.py need no changes.
|
|
"""
|
|
from services.database import get_latest_saxo_snapshot_rows
|
|
|
|
flat_rows = get_latest_saxo_snapshot_rows(symbol.upper())
|
|
if not flat_rows:
|
|
raise ValueError(
|
|
f"Aucun historique Saxo pour '{symbol}' — ajoutez-le à la watchlist "
|
|
f"(Config → Saxo) et attendez le prochain cycle de snapshot (~5 min)."
|
|
)
|
|
|
|
spot = next((r["spot"] for r in flat_rows if r.get("spot") is not None), None)
|
|
today = date.today()
|
|
|
|
by_expiry: Dict[str, List[Dict[str, Any]]] = {}
|
|
for r in flat_rows:
|
|
if r.get("expiry_date"):
|
|
by_expiry.setdefault(r["expiry_date"], []).append(r)
|
|
|
|
def _days_to(expiry_date: str) -> int:
|
|
return (datetime.strptime(expiry_date[:10], "%Y-%m-%d").date() - today).days
|
|
|
|
selected = sorted(by_expiry.keys(), key=lambda e: abs(_days_to(e) - target_days))[:max(1, n_expiries)]
|
|
|
|
def _row_shape(r: Dict[str, Any]) -> Dict[str, Any]:
|
|
bid = r.get("bid") or 0.0
|
|
ask = 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 {
|
|
"strike": float(r["strike"]),
|
|
"bid": float(bid),
|
|
"ask": float(ask),
|
|
"mid": float(mid),
|
|
"last": float(mid),
|
|
"iv": float(vol_pct) / 100.0 if vol_pct is not None else 0.0,
|
|
"open_interest": 0,
|
|
"volume": 0,
|
|
}
|
|
|
|
expiries_out = []
|
|
for expiry_date in sorted(selected, key=_days_to):
|
|
rows = by_expiry[expiry_date]
|
|
calls = sorted([_row_shape(r) for r in rows if r["option_type"] == "call"], key=lambda x: x["strike"])
|
|
puts = sorted([_row_shape(r) for r in rows if r["option_type"] == "put"], key=lambda x: x["strike"])
|
|
if not calls and not puts:
|
|
continue
|
|
expiries_out.append({
|
|
"expiry_date": expiry_date,
|
|
"days_to_expiry": _days_to(expiry_date),
|
|
"calls": calls,
|
|
"puts": puts,
|
|
})
|
|
|
|
if not expiries_out:
|
|
raise ValueError(f"Historique Saxo présent pour '{symbol}' mais aucune échéance exploitable (pas de cotation Call/Put dans la fenêtre active de Saxo).")
|
|
|
|
return {
|
|
"symbol": symbol.upper(),
|
|
"proxy": symbol.upper(),
|
|
"spot": round(float(spot), 6) if spot is not None else None,
|
|
"expiries": expiries_out,
|
|
}
|
|
|
|
|
|
def find_quote(chain_slice: Dict[str, Any], expiry_date: str, strike: float, option_type: str) -> Optional[Dict[str, Any]]:
|
|
"""Look up a single contract's quote row within a previously fetched chain slice."""
|
|
for exp in chain_slice["expiries"]:
|
|
if exp["expiry_date"] != expiry_date:
|
|
continue
|
|
rows = exp["calls"] if option_type == "call" else exp["puts"]
|
|
for row in rows:
|
|
if abs(row["strike"] - strike) < 1e-6:
|
|
return row
|
|
return None
|