feat: saxo connector
This commit is contained in:
Binary file not shown.
@@ -1,11 +1,14 @@
|
||||
from typing import List, Optional
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Query
|
||||
from pydantic import BaseModel
|
||||
|
||||
from services import saxo_auth
|
||||
from services.saxo_scheduler import get_watchlist, set_watchlist
|
||||
from services.database import get_saxo_snapshots, get_saxo_snapshot_symbols
|
||||
from services.database import (
|
||||
get_saxo_snapshots, get_saxo_snapshot_symbols,
|
||||
get_saxo_catalog, get_saxo_catalog_summary, upsert_saxo_catalog_rows,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/api/saxo", tags=["saxo"])
|
||||
|
||||
@@ -90,3 +93,44 @@ def history(
|
||||
date_to: Optional[str] = Query(None),
|
||||
):
|
||||
return get_saxo_snapshots(symbol, date_from, date_to)
|
||||
|
||||
|
||||
class CatalogRefreshRequest(BaseModel):
|
||||
asset_types: Optional[List[str]] = None # default: CATALOG_ASSET_TYPES (FuturesOption, FxVanillaOption)
|
||||
|
||||
|
||||
@router.post("/catalog/refresh")
|
||||
def refresh_catalog(req: CatalogRefreshRequest):
|
||||
from services.saxo_client import list_instruments, CATALOG_ASSET_TYPES, SaxoNotConnected, SaxoApiError
|
||||
asset_types = req.asset_types or CATALOG_ASSET_TYPES
|
||||
counts: Dict[str, int] = {}
|
||||
try:
|
||||
for asset_type in asset_types:
|
||||
items = list_instruments(asset_type)
|
||||
rows = [{
|
||||
"uic": item.get("Identifier") or item.get("Uic"),
|
||||
"symbol": item.get("Symbol"),
|
||||
"asset_type": asset_type,
|
||||
"description": item.get("Description"),
|
||||
} for item in items if item.get("Identifier") or item.get("Uic")]
|
||||
upsert_saxo_catalog_rows(rows)
|
||||
counts[asset_type] = len(rows)
|
||||
except SaxoNotConnected as e:
|
||||
raise HTTPException(status_code=401, detail=str(e))
|
||||
except SaxoApiError as e:
|
||||
raise HTTPException(status_code=502, detail=str(e))
|
||||
return {"refreshed": counts}
|
||||
|
||||
|
||||
@router.get("/catalog")
|
||||
def catalog(
|
||||
asset_type: Optional[str] = Query(None),
|
||||
q: Optional[str] = Query(None),
|
||||
limit: int = Query(200),
|
||||
):
|
||||
return get_saxo_catalog(asset_type, q, limit)
|
||||
|
||||
|
||||
@router.get("/catalog/summary")
|
||||
def catalog_summary():
|
||||
return get_saxo_catalog_summary()
|
||||
|
||||
@@ -74,6 +74,16 @@ def init_db():
|
||||
)""")
|
||||
c.execute("CREATE INDEX IF NOT EXISTS idx_saxo_snap_symbol_date ON saxo_option_snapshots(symbol, snapshot_date)")
|
||||
|
||||
c.execute("""CREATE TABLE IF NOT EXISTS saxo_instrument_catalog (
|
||||
uic INTEGER PRIMARY KEY,
|
||||
symbol TEXT NOT NULL,
|
||||
asset_type TEXT NOT NULL,
|
||||
description TEXT,
|
||||
updated_at TEXT DEFAULT (datetime('now'))
|
||||
)""")
|
||||
c.execute("CREATE INDEX IF NOT EXISTS idx_saxo_catalog_type ON saxo_instrument_catalog(asset_type)")
|
||||
c.execute("CREATE INDEX IF NOT EXISTS idx_saxo_catalog_symbol ON saxo_instrument_catalog(symbol)")
|
||||
|
||||
c.execute("""CREATE TABLE IF NOT EXISTS strategy_scenarios (
|
||||
id TEXT PRIMARY KEY,
|
||||
symbol TEXT NOT NULL,
|
||||
@@ -6126,3 +6136,44 @@ def get_saxo_snapshot_symbols() -> List[Dict[str, Any]]:
|
||||
""").fetchall()
|
||||
conn.close()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def upsert_saxo_catalog_rows(rows: List[Dict[str, Any]]):
|
||||
if not rows:
|
||||
return
|
||||
conn = get_conn()
|
||||
conn.executemany("""INSERT INTO saxo_instrument_catalog (uic, symbol, asset_type, description, updated_at)
|
||||
VALUES (?,?,?,?, datetime('now'))
|
||||
ON CONFLICT(uic) DO UPDATE SET
|
||||
symbol=excluded.symbol, asset_type=excluded.asset_type,
|
||||
description=excluded.description, updated_at=datetime('now')""",
|
||||
[(r["uic"], r["symbol"], r["asset_type"], r.get("description")) for r in rows])
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
|
||||
def get_saxo_catalog(asset_type: Optional[str] = None, q: Optional[str] = None, limit: int = 200) -> List[Dict[str, Any]]:
|
||||
conn = get_conn()
|
||||
query = "SELECT * FROM saxo_instrument_catalog WHERE 1=1"
|
||||
params: List[Any] = []
|
||||
if asset_type:
|
||||
query += " AND asset_type=?"
|
||||
params.append(asset_type)
|
||||
if q:
|
||||
query += " AND (symbol LIKE ? OR description LIKE ?)"
|
||||
params.extend([f"%{q}%", f"%{q}%"])
|
||||
query += " ORDER BY symbol LIMIT ?"
|
||||
params.append(limit)
|
||||
rows = conn.execute(query, params).fetchall()
|
||||
conn.close()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def get_saxo_catalog_summary() -> List[Dict[str, Any]]:
|
||||
conn = get_conn()
|
||||
rows = conn.execute("""
|
||||
SELECT asset_type, COUNT(*) AS count, MAX(updated_at) AS last_refreshed
|
||||
FROM saxo_instrument_catalog GROUP BY asset_type ORDER BY asset_type
|
||||
""").fetchall()
|
||||
conn.close()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
@@ -24,7 +24,11 @@ from services.saxo_auth import SAXO_API_BASE_URL, get_valid_access_token
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_OPTION_ASSET_TYPES = "StockOption,StockIndexOption,FuturesOption"
|
||||
_OPTION_ASSET_TYPES = "StockOption,StockIndexOption,FuturesOption,FxVanillaOption"
|
||||
|
||||
# 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]] = {}
|
||||
@@ -67,6 +71,25 @@ def _get(path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||||
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."""
|
||||
@@ -103,7 +126,7 @@ def resolve_instrument(symbol: str) -> Dict[str, Any]:
|
||||
raise ValueError(f"Aucun option root Saxo trouvé pour '{symbol}' (réponse: {list(data.keys())})")
|
||||
|
||||
item = items[0]
|
||||
uic = _first(item, "Uic", "uic", "Identifier")
|
||||
uic = _first(item, "Identifier", "Uic", "uic")
|
||||
if uic is None:
|
||||
raise ValueError(f"Champ Uic introuvable dans la réponse instruments pour '{symbol}': {item}")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user