admin/OpenFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: saxo price

Browse Source
This commit is contained in:
OpenSquared
2026-07-29 18:07:26 +02:00
parent 61e120a8c5
commit e514e9799b
1 changed files with 100 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

123
backend/services/saxo_client.py
View File

@@ -356,18 +356,13 @@ 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."""
def _open_subscription(uic: int, asset_type: str, headers: Dict[str, str]) -> tuple:
"""POST creates a subscription and returns an initial snapshot in the same response.
Caller owns the subscription until _close_subscription is called — this lets
snapshot_options_chain scroll it via PATCH before tearing it down, instead of the
old POST-then-immediately-DELETE pattern (still used as-is by debug_chain_raw)."""
context_id = f"of-{uuid.uuid4().hex[:12]}"
reference_id = "chain"
token_headers = _headers()
body = {
"ContextId": context_id,
"ReferenceId": reference_id,
@@ -376,20 +371,49 @@ def _snapshot_via_subscription(uic: int, asset_type: str = "StockOption") -> Dic
}
resp = httpx.post(
f"{SAXO_API_BASE_URL}/trade/v1/optionschain/subscriptions",
headers={**token_headers, "Content-Type": "application/json"},
headers={**headers, "Content-Type": "application/json"},
json=body, timeout=15,
)
_raise_for_status(resp)
snapshot = resp.json()
return context_id, reference_id, resp.json()
def _close_subscription(context_id: str, reference_id: str, headers: Dict[str, str]) -> None:
try:
httpx.delete(
f"{SAXO_API_BASE_URL}/trade/v1/optionschain/subscriptions/{context_id}/{reference_id}",
headers=token_headers, timeout=10,
headers=headers, timeout=10,
)
except Exception as e:
logger.warning(f"[Saxo] Failed to clean up options-chain subscription {context_id}: {e}")
def _patch_subscription_window(
context_id: str, reference_id: str, indices: List[int], max_strikes_per_expiry: int, headers: Dict[str, str],
) -> Dict[str, Any]:
"""Scrolls an already-open options-chain subscription to a different set of expiry
indices (0 = nearest expiry, 1 = next, ...) — the mechanism Saxo's own option board
uses to page through a chain. https://www.developer.saxo/openapi/learn/options-chain
documents Arguments.Expiries[].Index / MaxStrikesPerExpiry for this, with a hard cap
of (#indices requested * max_strikes_per_expiry) <= 100 per call. Response shape is
lower-confidence (not directly confirmed against a real payload the way the POST
snapshot shape was) — assumed to mirror the same Expiries[]/Strikes[] structure,
positionally indexed the same way, with untouched positions left empty/unchanged."""
resp = httpx.patch(
f"{SAXO_API_BASE_URL}/trade/v1/optionschain/subscriptions/{context_id}/{reference_id}",
headers={**headers, "Content-Type": "application/json"},
json={"Expiries": [{"Index": i} for i in indices], "MaxStrikesPerExpiry": max_strikes_per_expiry},
timeout=15,
)
_raise_for_status(resp)
return resp.json()
def _snapshot_via_subscription(uic: int, asset_type: str = "StockOption") -> Dict[str, Any]:
"""One-off snapshot: open then immediately close, no scrolling — used by debug_chain_raw."""
headers = _headers()
context_id, reference_id, snapshot = _open_subscription(uic, asset_type, headers)
_close_subscription(context_id, reference_id, headers)
return snapshot
@@ -402,7 +426,27 @@ def debug_chain_raw(symbol: str) -> Dict[str, Any]:
return {"instrument": instrument, "raw": subscription}
def snapshot_options_chain(symbol: str, target_days: int = 30) -> List[Dict[str, Any]]:
# Saxo's options-chain subscription only actively quotes a narrow default window (near-the-
# money strikes on the ~3 nearest expiry indices) even though the chain itself extends much
# further — the initial snapshot already lists every expiry index up to ExpiryCount with its
# date, just with an empty Strikes array beyond that default window. These constants govern
# scrolling the subscription via PATCH to backfill the rest, respecting Saxo's fixed
# (#expiries_requested * MaxStrikesPerExpiry) <= 100 cap per call.
_MAX_STRIKES_PER_SAXO_REQUEST = 100
_STRIKES_PER_EXPIRY_WINDOW = 20
_EXPIRIES_PER_PATCH_BATCH = _MAX_STRIKES_PER_SAXO_REQUEST // _STRIKES_PER_EXPIRY_WINDOW
def _expiry_days_out(expiry: Optional[str], snapshot_date: str) -> Optional[int]:
if not expiry:
return None
try:
return (date.fromisoformat(expiry[:10]) - date.fromisoformat(snapshot_date)).days
except ValueError:
return None
def snapshot_options_chain(symbol: str, max_days: int = 120) -> 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,
@@ -410,20 +454,53 @@ def snapshot_options_chain(symbol: str, target_days: int = 30) -> List[Dict[str,
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).
strikes/expiries in this same snapshot DID carry a live MidVolatility.
Covers every expiry within `max_days` (not just Saxo's default near-dated window) by
opening the subscription, scrolling it via PATCH in batches of _EXPIRIES_PER_PATCH_BATCH
expiry indices, and closing it — see _patch_subscription_window. A batch that fails
(network hiccup, entitlement gap) is logged and skipped rather than aborting the whole
symbol, so a partial chain is still better than none.
"""
instrument = resolve_instrument(symbol)
root_uic = instrument["uic"]
asset_type = instrument["asset_type"] or "StockOption"
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()
headers = _headers()
context_id, reference_id, opened = _open_subscription(root_uic, asset_type, headers)
try:
# The POST response is the streaming-subscription envelope (ContextId/ReferenceId/
# Format/RefreshRate/InactivityTimeout/State) — the actual chain payload is nested
# under "Snapshot".
snapshot = opened.get("Snapshot") or opened
snapshot_date = date.today().isoformat()
expiry_blocks = _first(snapshot, "Expiries", "OptionsChain") or []
by_index: Dict[int, Dict[str, Any]] = {i: eb for i, eb in enumerate(expiry_blocks)}
wanted_indices = []
for i, eb in by_index.items():
dte = _expiry_days_out(eb.get("Expiry"), snapshot_date)
if dte is not None and dte <= max_days:
wanted_indices.append(i)
wanted_indices.sort()
for batch_start in range(0, len(wanted_indices), _EXPIRIES_PER_PATCH_BATCH):
batch = wanted_indices[batch_start:batch_start + _EXPIRIES_PER_PATCH_BATCH]
try:
patched = _patch_subscription_window(context_id, reference_id, batch, _STRIKES_PER_EXPIRY_WINDOW, headers)
except Exception as e:
logger.warning(f"[Saxo] Options-chain scroll failed for {symbol} expiry indices {batch}: {e}")
continue
patched_snapshot = patched.get("Snapshot") or patched
for i, eb in enumerate(_first(patched_snapshot, "Expiries", "OptionsChain") or []):
if i in batch and (eb.get("Strikes") or []):
by_index[i] = eb
expiry_blocks = [by_index[i] for i in sorted(by_index)]
finally:
_close_subscription(context_id, reference_id, headers)
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)