feat: sexo history

This commit is contained in:
OpenSquared
2026-07-18 18:54:12 +02:00
parent ec5cf5abf4
commit 212bdf6678
4 changed files with 34 additions and 12 deletions

View File

@@ -37,10 +37,14 @@ def update_watchlist(req: WatchlistRequest):
def _validate_symbol(symbol: str) -> dict:
from services.saxo_client import resolve_option_root_uic, SaxoNotConnected
from services.saxo_client import resolve_instrument, SaxoNotConnected
try:
uic = resolve_option_root_uic(symbol)
return {"symbol": symbol.upper(), "valid": True, "uic": uic, "error": None}
info = resolve_instrument(symbol)
return {
"symbol": symbol.upper(), "valid": True, "error": None,
"uic": info["uic"], "matched_symbol": info["symbol"],
"description": info["description"], "asset_type": info["asset_type"],
}
except SaxoNotConnected as e:
return {"symbol": symbol.upper(), "valid": False, "uic": None, "error": str(e)}
except Exception as e:

View File

@@ -26,8 +26,8 @@ logger = logging.getLogger(__name__)
_OPTION_ASSET_TYPES = "StockOption,StockIndexOption,FuturesOption,ContractFutures"
# symbol -> option root Uic, cheap in-process cache (roots don't change within a session)
_root_uic_cache: Dict[str, int] = {}
# 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):
@@ -91,7 +91,9 @@ def _first(d: Dict[str, Any], *keys: str) -> Any:
return None
def resolve_option_root_uic(symbol: str) -> int:
def resolve_instrument(symbol: str) -> 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."""
if symbol in _root_uic_cache:
return _root_uic_cache[symbol]
@@ -100,12 +102,23 @@ def resolve_option_root_uic(symbol: str) -> int:
if not items:
raise ValueError(f"Aucun option root Saxo trouvé pour '{symbol}' (réponse: {list(data.keys())})")
uic = _first(items[0], "Uic", "uic", "Identifier")
item = items[0]
uic = _first(item, "Uic", "uic", "Identifier")
if uic is None:
raise ValueError(f"Champ Uic introuvable dans la réponse instruments pour '{symbol}': {items[0]}")
raise ValueError(f"Champ Uic introuvable dans la réponse instruments pour '{symbol}': {item}")
_root_uic_cache[symbol] = int(uic)
return int(uic)
result = {
"uic": int(uic),
"symbol": _first(item, "Symbol", "symbol"),
"description": _first(item, "Description", "description"),
"asset_type": _first(item, "AssetType", "assetType"),
}
_root_uic_cache[symbol] = result
return result
def resolve_option_root_uic(symbol: str) -> int:
return resolve_instrument(symbol)["uic"]
def get_option_space(root_uic: int) -> Dict[str, Any]: