""" Thin Saxo OpenAPI REST client for options chain snapshots. The options-chain subscription response shape (confirmed against a real payload): Snapshot.Expiries[].Strikes[].{Call,Put}, each with Bid/Ask/Uic directly and Delta/Gamma/Theta/Vega/MidVolatility nested under .Greeks. There is no Mid field (computed as (bid+ask)/2) and no underlying spot field at all — MidStrikePrice on the nearest expiry is used as the spot proxy. Only strikes within Saxo's active quoting window carry a Call/Put payload; the rest are bare {Index, Strike}. """ import logging import time import uuid from datetime import date from typing import Any, Dict, List, Optional import httpx from services.options_pricer import black_scholes from services.saxo_auth import SAXO_API_BASE_URL, get_valid_access_token from services.vol_surface import Surface logger = logging.getLogger(__name__) _OPTION_ASSET_TYPES = "StockOption,StockIndexOption,FuturesOption,FxVanillaOption" # Applied around a Black-Scholes theoretical price when Saxo returns no live Bid/Ask (FX # options in particular go quiet outside FX market hours — closed over the weekend) but # still supplies MidVolatility/Greeks from its own model, so IV/spot/strike are usable. # # Calibrated against a real Saxo EURUSD chain: the quoted spread there sits at a near- # constant ~9-10 pips (~0.0009-0.0010) in ABSOLUTE terms across every strike/expiry — # 40%+ of the premium for a cheap OTM contract, ~8% for an expensive one. A pure # percentage-of-premium spread badly underprices the OTM end, so the wider of the two # (percentage floor vs. a spot-scaled absolute floor) is used. _SYNTHETIC_SPREAD_PCT = 0.05 _SYNTHETIC_SPREAD_FLOOR_PCT_OF_SPOT = 0.0008 _SYNTHETIC_RATE = 0.02 def _synthesize_quote( spot: Optional[float], strike: Optional[float], expiry_date: Optional[str], snapshot_date: str, iv_pct: Optional[float], option_type: str, ) -> tuple: if not spot or not strike or not expiry_date or not iv_pct or iv_pct <= 0: return None, None, None try: days_to_expiry = (date.fromisoformat(expiry_date) - date.fromisoformat(snapshot_date)).days except ValueError: return None, None, None T = max(days_to_expiry, 1) / 365 theo = float(black_scholes(spot, strike, T, _SYNTHETIC_RATE, iv_pct / 100.0, option_type)["price"]) if theo <= 0: return None, None, None full_spread = max(theo * _SYNTHETIC_SPREAD_PCT, spot * _SYNTHETIC_SPREAD_FLOOR_PCT_OF_SPOT) half_spread = full_spread / 2 bid = round(max(0.0, theo - half_spread), 6) ask = round(theo + half_spread, 6) return bid, ask, round((bid + ask) / 2, 6) # Bounded, stable catalogs worth fully caching in our own DB (StockOption/StockIndexOption # are far too large to bulk-fetch — those stay resolved on demand via Keywords search). CATALOG_ASSET_TYPES = ["FuturesOption", "FxVanillaOption"] # symbol -> resolved instrument details, cheap in-process cache (roots don't change within a session) _root_uic_cache: Dict[str, Dict[str, Any]] = {} class SaxoNotConnected(Exception): pass class SaxoApiError(Exception): """Wraps an httpx.HTTPStatusError with Saxo's actual JSON error body (ErrorCode/Message), since the bare status code is useless for debugging a 400 on a subscription body.""" def __init__(self, resp: httpx.Response): detail = resp.text try: detail = resp.json() except Exception: pass super().__init__(f"Saxo API {resp.status_code} on {resp.request.method} {resp.request.url}: {detail}") def _raise_for_status(resp: httpx.Response): if resp.is_error: raise SaxoApiError(resp) _account_key_cache: Optional[str] = None def _headers() -> Dict[str, str]: token = get_valid_access_token() if not token: raise SaxoNotConnected("Saxo n'est pas connecté — complétez l'OAuth login d'abord.") return {"Authorization": f"Bearer {token}"} def _get(path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: resp = httpx.get(f"{SAXO_API_BASE_URL}{path}", headers=_headers(), params=params or {}, timeout=15) _raise_for_status(resp) return resp.json() def list_instruments(asset_types: str, keywords: Optional[str] = None, max_pages: int = 20) -> List[Dict[str, Any]]: """Paginated dump of /ref/v1/instruments for a given AssetTypes filter — used to build our own local catalog for bounded, stable asset classes (FuturesOption/FxVanillaOption).""" all_items: List[Dict[str, Any]] = [] top = 1000 skip = 0 for _ in range(max_pages): params: Dict[str, Any] = {"AssetTypes": asset_types, "$top": top, "$skip": skip} if keywords: params["Keywords"] = keywords data = _get("/ref/v1/instruments", params) items = data.get("Data") or data.get("data") or [] all_items.extend(items) if len(items) < top: break skip += top return all_items def get_default_account_key() -> str: """Most /trade/v1 subscriptions (including the options chain) require AccountKey in Arguments — /port/v1/accounts/me is the standard way to get the user's default account.""" global _account_key_cache if _account_key_cache: return _account_key_cache data = _get("/port/v1/accounts/me") items = data.get("Data") or data.get("data") or [] if not items: raise ValueError(f"Aucun compte trouvé dans /port/v1/accounts/me: {data}") account_key = _first(items[0], "AccountKey", "accountKey") if not account_key: raise ValueError(f"Champ AccountKey introuvable pour le compte: {items[0]}") _account_key_cache = account_key return account_key def _first(d: Dict[str, Any], *keys: str) -> Any: for k in keys: if k in d: return d[k] return None def resolve_instrument(symbol: str, asset_types: str = _OPTION_ASSET_TYPES) -> Dict[str, Any]: """Full matched instrument (Uic + Symbol/Description/AssetType) — surfaced by /validate so a wrong-ticker guess is visibly distinguishable from an account-entitlement error.""" cache_key = f"{symbol}|{asset_types}" if cache_key in _root_uic_cache: return _root_uic_cache[cache_key] # Fast path: exchange-suffixed catalog symbols (e.g. "EUU:XCME", "OG:xcme") are Saxo # identifiers, not natural-language search terms — Saxo's free-text Keywords search # isn't reliable for them. If we already cached this exact symbol via a catalog # refresh, use its known Uic directly instead of re-searching live. from services.database import get_saxo_catalog_by_symbol cached_entry = get_saxo_catalog_by_symbol(symbol) if cached_entry: result = { "uic": cached_entry["uic"], "symbol": cached_entry["symbol"], "description": cached_entry["description"], "asset_type": cached_entry["asset_type"], } _root_uic_cache[cache_key] = result return result data = _get("/ref/v1/instruments", {"Keywords": symbol, "AssetTypes": asset_types}) items = data.get("Data") or data.get("data") or [] if not items: raise ValueError(f"Aucun instrument Saxo trouvé pour '{symbol}' (AssetTypes={asset_types})") # An exact symbol match always wins over Saxo's own search ranking — e.g. searching # "EURUSD" must resolve to the plain EURUSD root, not one of the "EURUSD Weekly (2)..." # futures-option variants that also match the keyword and could rank first. exact = next((i for i in items if str(_first(i, "Symbol", "symbol") or "").upper() == symbol.upper()), None) item = exact or items[0] uic = _first(item, "Identifier", "Uic", "uic") if uic is None: raise ValueError(f"Champ Uic introuvable dans la réponse instruments pour '{symbol}': {item}") result = { "uic": int(uic), "symbol": _first(item, "Symbol", "symbol"), "description": _first(item, "Description", "description"), "asset_type": _first(item, "AssetType", "assetType"), } _root_uic_cache[cache_key] = result return result def get_price_quote(symbol: str, asset_type: str = "FxSpot", amount: int = 100000) -> Dict[str, Any]: """ Basic (non-options) price lookup via /trade/v1/infoprices/list — confirmed working shape (AccountKey, Uics, AssetType, Amount, FieldGroups) straight from a real Saxo Explorer response. Used to test plain market-data access independently of the options chain (which returns NoDataAccess even though basic FxSpot/Stock prices work fine). """ instrument = resolve_instrument(symbol, asset_types=asset_type) data = _get("/trade/v1/infoprices/list", { "AccountKey": get_default_account_key(), "Uics": instrument["uic"], "AssetType": asset_type, "Amount": amount, "FieldGroups": "DisplayAndFormat,Quote", }) items = data.get("Data") or [] if not items: raise ValueError(f"Aucune cotation renvoyée pour '{symbol}' ({asset_type})") row = items[0] quote = row.get("Quote", {}) display = row.get("DisplayAndFormat", {}) return { "symbol": instrument["symbol"], "description": display.get("Description"), "asset_type": asset_type, "uic": instrument["uic"], "bid": quote.get("Bid"), "ask": quote.get("Ask"), "last_updated": row.get("LastUpdated"), "price_source": row.get("PriceSource"), } def resolve_option_root_uic(symbol: str) -> int: return resolve_instrument(symbol)["uic"] if not legs: raise ValueError(f"Impossible d'extraire des contrats depuis contractoptionspaces (clés reçues: {list(space.keys())})") return legs def _snapshot_via_subscription(uic: int, asset_type: str = "StockOption") -> Dict[str, Any]: """POST creates a subscription and returns an initial snapshot in the same response; DELETE immediately after — no websocket needs to be kept alive for a one-off read.""" context_id = f"of-{uuid.uuid4().hex[:12]}" reference_id = "chain" token_headers = _headers() body = { "ContextId": context_id, "ReferenceId": reference_id, "Arguments": {"Identifier": uic, "AssetType": asset_type, "AccountKey": get_default_account_key()}, "RefreshRate": 5000, } resp = httpx.post( f"{SAXO_API_BASE_URL}/trade/v1/optionschain/subscriptions", headers={**token_headers, "Content-Type": "application/json"}, json=body, timeout=15, ) _raise_for_status(resp) snapshot = resp.json() try: httpx.delete( f"{SAXO_API_BASE_URL}/trade/v1/optionschain/subscriptions/{context_id}/{reference_id}", headers=token_headers, timeout=10, ) except Exception as e: logger.warning(f"[Saxo] Failed to clean up options-chain subscription {context_id}: {e}") return snapshot def debug_chain_raw(symbol: str) -> Dict[str, Any]: """Returns the raw (unparsed) options-chain subscription response for one symbol — diagnostic only, used to pin down exactly where Mid/Greeks/Spot live in Saxo's real payload instead of guessing field names blind.""" instrument = resolve_instrument(symbol) subscription = _snapshot_via_subscription(instrument["uic"], asset_type=instrument["asset_type"] or "StockOption") return {"instrument": instrument, "raw": subscription} def snapshot_options_chain(symbol: str, target_days: int = 30) -> List[Dict[str, Any]]: """ Returns normalized rows ready for services/database.save_saxo_snapshot_rows: {symbol, snapshot_date, spot, expiry_date, strike, option_type, bid, ask, mid, volatility_pct, delta, gamma, theta, vega, is_synthetic}. bid/ask/mid are Black-Scholes-synthesized (is_synthetic=True) whenever Saxo returns no live Bid/Ask for that contract (e.g. FX options outside market hours) — using that contract's own IV when Saxo quoted it, or otherwise an IV borrowed from a smile built across whatever strikes/expiries in this same snapshot DID carry a live MidVolatility (Saxo's "active quoting window" is often just the near-the-money strikes on the nearest expiry; the rest of the chain has no Greeks/MidVolatility at all, not just no Bid/Ask). """ instrument = resolve_instrument(symbol) root_uic = instrument["uic"] subscription = _snapshot_via_subscription(root_uic, asset_type=instrument["asset_type"] or "StockOption") # The POST response is the streaming-subscription envelope (ContextId/ReferenceId/Format/ # RefreshRate/InactivityTimeout/State) — the actual chain payload is nested under "Snapshot". snapshot = subscription.get("Snapshot") or subscription snapshot_date = date.today().isoformat() expiry_blocks = _first(snapshot, "Expiries", "OptionsChain") or [] # No spot/underlying price field exists anywhere in this response (confirmed against a # real payload) — MidStrikePrice on the nearest expiry is the best available proxy. spot = next((eb.get("MidStrikePrice") for eb in expiry_blocks if eb.get("MidStrikePrice") is not None), None) # First pass: take exactly what Saxo quoted, no synthesis yet. raw: List[Dict[str, Any]] = [] for expiry_block in expiry_blocks: expiry_date = (expiry_block.get("Expiry") or "")[:10] or None try: days_to_expiry = (date.fromisoformat(expiry_date) - date.fromisoformat(snapshot_date)).days if expiry_date else None except ValueError: days_to_expiry = None for strike_block in (expiry_block.get("Strikes") or []): strike = strike_block.get("Strike") for side_key in ("Call", "Put"): side = strike_block.get(side_key) if not side: continue greeks = side.get("Greeks") or {} bid, ask = side.get("Bid"), side.get("Ask") mid_vol = greeks.get("MidVolatility") raw.append({ "symbol": symbol.upper(), "snapshot_date": snapshot_date, "spot": float(spot) if spot is not None else None, "expiry_date": expiry_date, "days_to_expiry": days_to_expiry, "strike": float(strike) if strike is not None else None, "option_type": "put" if side_key == "Put" else "call", "bid": bid, "ask": ask, "mid": round((bid + ask) / 2, 6) if (bid is not None and ask is not None) else None, # MidVolatility comes back as a decimal fraction (0.05 = 5%) — store as an # actual percentage to match the volatility_pct column's name/convention. "volatility_pct": round(mid_vol * 100, 4) if mid_vol is not None else None, "delta": greeks.get("Delta"), "gamma": greeks.get("Gamma"), "theta": greeks.get("Theta"), "vega": greeks.get("Vega"), }) if not raw: raise ValueError(f"Snapshot Saxo vide pour '{symbol}' (clés reçues: {list(snapshot.keys())})") fallback_surface = _build_fallback_surface(spot, raw) rows: List[Dict[str, Any]] = [] for r in raw: bid, ask, mid, vol_pct = r["bid"], r["ask"], r["mid"], r["volatility_pct"] is_synthetic = False if not bid and not ask: iv_for_synth = vol_pct if iv_for_synth is None and fallback_surface is not None and r["strike"] and r["days_to_expiry"]: iv_for_synth = round(fallback_surface.iv_at(r["strike"], max(r["days_to_expiry"], 1)) * 100, 4) syn_bid, syn_ask, syn_mid = _synthesize_quote( r["spot"], r["strike"], r["expiry_date"], snapshot_date, iv_for_synth, r["option_type"], ) if syn_bid is not None: bid, ask, mid, is_synthetic = syn_bid, syn_ask, syn_mid, True if vol_pct is None: vol_pct = iv_for_synth rows.append({ **{k: v for k, v in r.items() if k != "days_to_expiry"}, "bid": bid, "ask": ask, "mid": mid, "volatility_pct": vol_pct, "is_synthetic": is_synthetic, }) return rows def _build_fallback_surface(spot: Optional[float], raw_rows: List[Dict[str, Any]]) -> Optional[Surface]: """A smile built only from strikes/expiries that carried a live MidVolatility in this same snapshot — used to borrow a plausible IV for contracts Saxo didn't quote at all.""" if not spot: return None by_days: Dict[float, Dict[str, Any]] = {} for r in raw_rows: if r["volatility_pct"] is None or r["days_to_expiry"] is None or r["strike"] is None: continue exp = by_days.setdefault(r["days_to_expiry"], {"days_to_expiry": r["days_to_expiry"], "calls": [], "puts": []}) entry = {"strike": r["strike"], "iv": r["volatility_pct"] / 100.0} (exp["calls"] if r["option_type"] == "call" else exp["puts"]).append(entry) if not by_days: return None return Surface(spot, list(by_days.values()))