345 lines
15 KiB
Python
345 lines
15 KiB
Python
"""
|
|
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
|
|
|
|
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 from IV (is_synthetic=True) whenever Saxo returns no live
|
|
Bid/Ask for that contract (e.g. FX options outside market hours).
|
|
"""
|
|
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)
|
|
|
|
rows: List[Dict[str, Any]] = []
|
|
for expiry_block in expiry_blocks:
|
|
expiry_date = (expiry_block.get("Expiry") or "")[:10] or None
|
|
for strike_block in (strike_block 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")
|
|
option_type = "put" if side_key == "Put" else "call"
|
|
vol_pct = round(mid_vol * 100, 4) if mid_vol is not None else None
|
|
mid = round((bid + ask) / 2, 6) if (bid is not None and ask is not None) else None
|
|
|
|
is_synthetic = False
|
|
if not bid and not ask:
|
|
syn_bid, syn_ask, syn_mid = _synthesize_quote(
|
|
spot, strike, expiry_date, snapshot_date, vol_pct, option_type,
|
|
)
|
|
if syn_bid is not None:
|
|
bid, ask, mid, is_synthetic = syn_bid, syn_ask, syn_mid, True
|
|
|
|
rows.append({
|
|
"symbol": symbol.upper(),
|
|
"snapshot_date": snapshot_date,
|
|
"spot": float(spot) if spot is not None else None,
|
|
"expiry_date": expiry_date,
|
|
"strike": float(strike) if strike is not None else None,
|
|
"option_type": option_type,
|
|
"bid": bid,
|
|
"ask": ask,
|
|
"mid": mid,
|
|
# 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": vol_pct,
|
|
"delta": greeks.get("Delta"),
|
|
"gamma": greeks.get("Gamma"),
|
|
"theta": greeks.get("Theta"),
|
|
"vega": greeks.get("Vega"),
|
|
"is_synthetic": is_synthetic,
|
|
})
|
|
|
|
if not rows:
|
|
raise ValueError(f"Snapshot Saxo vide pour '{symbol}' (clés reçues: {list(snapshot.keys())})")
|
|
return rows
|