Saxo connector

backend/main.py

@@ -1,3 +1,6 @@
+from dotenv import load_dotenv
+load_dotenv()
+
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from routers import market_data, geopolitical, options, backtest, ai, portfolio, config, patterns, journal, cycle as cycle_router, profiles as profiles_router, reasoning as reasoning_router, knowledge as knowledge_router, options_vol as options_vol_router, analytics as analytics_router, risk as risk_router
@@ -21,6 +24,8 @@ from routers import instruments_watchlist as instruments_watchlist_router
 from routers import wavelet as wavelet_router
 from routers import ai_chat as ai_chat_router
 from routers import strategy_builder as strategy_builder_router
+from routers import saxo_oauth as saxo_oauth_router
+from routers import saxo as saxo_router
 from services.database import init_db, get_config, cleanup_stale_running_cycles
 import os
 import logging
@@ -134,6 +139,9 @@ def startup():
     # Start institutional reports weekly scheduler (COT Fridays, EIA Wednesdays)
     from services.institutional_scheduler import start_institutional_scheduler
     start_institutional_scheduler()
+    # Start Saxo OAuth token refresh + options-chain snapshot poller
+    from services.saxo_scheduler import start_saxo_scheduler
+    start_saxo_scheduler()
 
     # One-time migration: remove FXStreet-sourced rows (decimal format) — FF sync below will refill from FF
     try:
@@ -211,6 +219,8 @@ def shutdown():
     stop_pnl_scheduler()
     from services.institutional_scheduler import stop_institutional_scheduler
     stop_institutional_scheduler()
+    from services.saxo_scheduler import stop_saxo_scheduler
+    stop_saxo_scheduler()
 
 
 app.include_router(market_data.router)
@@ -249,6 +259,8 @@ app.include_router(instruments_watchlist_router.router)
 app.include_router(wavelet_router.router)
 app.include_router(ai_chat_router.router)
 app.include_router(strategy_builder_router.router)
+app.include_router(saxo_oauth_router.router)
+app.include_router(saxo_router.router)
 
 
 @app.get("/")

backend/routers/saxo.py

@@ -0,0 +1,59 @@
+from typing import 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
+
+router = APIRouter(prefix="/api/saxo", tags=["saxo"])
+
+
+class WatchlistRequest(BaseModel):
+    symbols: List[str]
+
+
+@router.get("/status")
+def status():
+    return saxo_auth.get_status()
+
+
+@router.post("/disconnect")
+def disconnect():
+    saxo_auth.disconnect()
+    return {"disconnected": True}
+
+
+@router.get("/watchlist")
+def watchlist():
+    return {"symbols": get_watchlist()}
+
+
+@router.put("/watchlist")
+def update_watchlist(req: WatchlistRequest):
+    set_watchlist(req.symbols)
+    return {"symbols": get_watchlist()}
+
+
+@router.post("/snapshot-now/{symbol}")
+def snapshot_now(symbol: str):
+    from services.saxo_client import snapshot_options_chain, SaxoNotConnected
+    from services.database import save_saxo_snapshot_rows
+    try:
+        rows = snapshot_options_chain(symbol)
+    except SaxoNotConnected as e:
+        raise HTTPException(status_code=401, detail=str(e))
+    except ValueError as e:
+        raise HTTPException(status_code=502, detail=str(e))
+    save_saxo_snapshot_rows(rows)
+    return {"symbol": symbol.upper(), "rows_saved": len(rows)}
+
+
+@router.get("/history")
+def history(
+    symbol: Optional[str] = Query(None),
+    date_from: Optional[str] = Query(None),
+    date_to: Optional[str] = Query(None),
+):
+    return get_saxo_snapshots(symbol, date_from, date_to)

backend/routers/saxo_oauth.py

@@ -0,0 +1,41 @@
+"""OAuth2 login/callback for Saxo — paths outside /api, matching the callback URL
+registered on the Saxo app dashboard (https://openfin.open-squared.tech/oauth/saxo/callback)."""
+import logging
+
+from fastapi import APIRouter, HTTPException, Query
+from fastapi.responses import RedirectResponse
+
+from services import saxo_auth
+
+logger = logging.getLogger(__name__)
+router = APIRouter(tags=["saxo-oauth"])
+
+FRONTEND_CONFIG_URL = "/config"
+
+
+@router.get("/oauth/saxo/login")
+def saxo_login():
+    if not saxo_auth.is_configured():
+        raise HTTPException(status_code=400, detail="SAXO_APP_KEY / SAXO_APP_SECRET manquants dans backend/.env")
+    state = saxo_auth.new_login_state()
+    return RedirectResponse(saxo_auth.build_authorize_url(state))
+
+
+@router.get("/oauth/saxo/callback")
+def saxo_callback(code: str = Query(...), state: str = Query(...)):
+    if not saxo_auth.consume_and_verify_state(state):
+        raise HTTPException(status_code=400, detail="État OAuth invalide (CSRF) — relancez la connexion depuis Config.")
+    try:
+        saxo_auth.exchange_code_for_tokens(code)
+    except Exception as e:
+        logger.error(f"[Saxo OAuth] Token exchange failed: {e}")
+        return RedirectResponse(f"{FRONTEND_CONFIG_URL}?saxo_error=1")
+    return RedirectResponse(f"{FRONTEND_CONFIG_URL}?saxo_connected=1")
+
+
+@router.post("/oauth/saxo/refresh")
+def saxo_manual_refresh():
+    result = saxo_auth.refresh_tokens()
+    if not result:
+        raise HTTPException(status_code=400, detail="Saxo n'est pas connecté")
+    return {"refreshed": True}

backend/services/database.py

@@ -44,6 +44,36 @@ def init_db():
         created_at TEXT DEFAULT (datetime('now'))
     )""")
 
+    c.execute("""CREATE TABLE IF NOT EXISTS saxo_auth (
+        id INTEGER PRIMARY KEY CHECK (id = 1),
+        access_token TEXT,
+        refresh_token TEXT,
+        expires_at TEXT,
+        refresh_expires_at TEXT,
+        environment TEXT DEFAULT 'sim',
+        updated_at TEXT DEFAULT (datetime('now'))
+    )""")
+
+    c.execute("""CREATE TABLE IF NOT EXISTS saxo_option_snapshots (
+        id TEXT PRIMARY KEY,
+        symbol TEXT NOT NULL,
+        snapshot_date TEXT NOT NULL,
+        spot REAL,
+        expiry_date TEXT NOT NULL,
+        strike REAL NOT NULL,
+        option_type TEXT NOT NULL,
+        bid REAL,
+        ask REAL,
+        mid REAL,
+        volatility_pct REAL,
+        delta REAL,
+        gamma REAL,
+        theta REAL,
+        vega REAL,
+        created_at TEXT DEFAULT (datetime('now'))
+    )""")
+    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 strategy_scenarios (
         id TEXT PRIMARY KEY,
         symbol TEXT NOT NULL,
@@ -6017,3 +6047,72 @@ def delete_saved_strategy(strategy_id: str) -> bool:
     deleted = cur.rowcount > 0
     conn.close()
     return deleted
+
+
+# ── Saxo OpenAPI: OAuth tokens & option snapshots ─────────────────────────────
+
+def save_saxo_tokens(access_token: str, refresh_token: str, expires_at: str, refresh_expires_at: str, environment: str = "sim"):
+    conn = get_conn()
+    conn.execute("""INSERT INTO saxo_auth (id, access_token, refresh_token, expires_at, refresh_expires_at, environment, updated_at)
+        VALUES (1, ?, ?, ?, ?, ?, datetime('now'))
+        ON CONFLICT(id) DO UPDATE SET
+            access_token=excluded.access_token, refresh_token=excluded.refresh_token,
+            expires_at=excluded.expires_at, refresh_expires_at=excluded.refresh_expires_at,
+            environment=excluded.environment, updated_at=datetime('now')""",
+        (access_token, refresh_token, expires_at, refresh_expires_at, environment))
+    conn.commit()
+    conn.close()
+
+
+def get_saxo_tokens() -> Optional[Dict[str, Any]]:
+    conn = get_conn()
+    row = conn.execute("SELECT * FROM saxo_auth WHERE id=1").fetchone()
+    conn.close()
+    return dict(row) if row else None
+
+
+def clear_saxo_tokens():
+    conn = get_conn()
+    conn.execute("DELETE FROM saxo_auth WHERE id=1")
+    conn.commit()
+    conn.close()
+
+
+def save_saxo_snapshot_rows(rows: List[Dict[str, Any]]):
+    if not rows:
+        return
+    import uuid
+    conn = get_conn()
+    conn.executemany("""INSERT INTO saxo_option_snapshots (
+        id, symbol, snapshot_date, spot, expiry_date, strike, option_type,
+        bid, ask, mid, volatility_pct, delta, gamma, theta, vega
+    ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", [
+        (
+            f"SNAP-{uuid.uuid4().hex[:12]}",
+            r["symbol"], r["snapshot_date"], r.get("spot"), r["expiry_date"], r["strike"], r["option_type"],
+            r.get("bid"), r.get("ask"), r.get("mid"), r.get("volatility_pct"),
+            r.get("delta"), r.get("gamma"), r.get("theta"), r.get("vega"),
+        )
+        for r in rows
+    ])
+    conn.commit()
+    conn.close()
+
+
+def get_saxo_snapshots(symbol: Optional[str] = None, date_from: Optional[str] = None, date_to: Optional[str] = None) -> List[Dict[str, Any]]:
+    conn = get_conn()
+    query = "SELECT * FROM saxo_option_snapshots WHERE 1=1"
+    params: List[Any] = []
+    if symbol:
+        query += " AND symbol=?"
+        params.append(symbol)
+    if date_from:
+        query += " AND snapshot_date>=?"
+        params.append(date_from)
+    if date_to:
+        query += " AND snapshot_date<=?"
+        params.append(date_to)
+    query += " ORDER BY snapshot_date DESC, expiry_date, strike LIMIT 5000"
+    rows = conn.execute(query, params).fetchall()
+    conn.close()
+    return [dict(r) for r in rows]

backend/services/saxo_auth.py

@@ -0,0 +1,144 @@
+"""
+Saxo OpenAPI OAuth2 Authorization Code Grant — SIM environment only.
+
+Access tokens last ~20 min, refresh tokens ~40 min (SIM), so a background scheduler
+(services/saxo_scheduler.py) must proactively refresh well before either expires —
+this module only exposes the primitives (build the login URL, exchange code, refresh,
+read a currently-valid token).
+"""
+import os
+import secrets
+import logging
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, Optional
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+SAXO_APP_KEY = os.environ.get("SAXO_APP_KEY", "")
+SAXO_APP_SECRET = os.environ.get("SAXO_APP_SECRET", "")
+SAXO_REDIRECT_URI = os.environ.get("SAXO_REDIRECT_URI", "https://openfin.open-squared.tech/oauth/saxo/callback")
+SAXO_AUTH_BASE_URL = os.environ.get("SAXO_AUTH_BASE_URL", "https://sim.logonvalidation.net")
+SAXO_API_BASE_URL = os.environ.get("SAXO_API_BASE_URL", "https://gateway.saxobank.com/sim/openapi")
+
+# In-memory CSRF state for the OAuth login->callback round trip (single-user desktop app).
+_pending_state: Optional[str] = None
+
+
+def is_configured() -> bool:
+    return bool(SAXO_APP_KEY and SAXO_APP_SECRET)
+
+
+def new_login_state() -> str:
+    global _pending_state
+    _pending_state = secrets.token_urlsafe(24)
+    return _pending_state
+
+
+def consume_and_verify_state(state: str) -> bool:
+    global _pending_state
+    ok = bool(_pending_state) and secrets.compare_digest(_pending_state, state or "")
+    _pending_state = None
+    return ok
+
+
+def build_authorize_url(state: str) -> str:
+    from urllib.parse import urlencode
+    params = {
+        "response_type": "code",
+        "client_id": SAXO_APP_KEY,
+        "redirect_uri": SAXO_REDIRECT_URI,
+        "state": state,
+    }
+    return f"{SAXO_AUTH_BASE_URL}/authorize?{urlencode(params)}"
+
+
+def _store_token_response(token_resp: Dict[str, Any]):
+    from services.database import save_saxo_tokens
+    now = datetime.now(timezone.utc)
+    expires_at = (now + timedelta(seconds=int(token_resp["expires_in"]))).isoformat()
+    refresh_expires_at = (now + timedelta(seconds=int(token_resp["refresh_token_expires_in"]))).isoformat()
+    save_saxo_tokens(
+        access_token=token_resp["access_token"],
+        refresh_token=token_resp["refresh_token"],
+        expires_at=expires_at,
+        refresh_expires_at=refresh_expires_at,
+        environment="sim",
+    )
+
+
+def exchange_code_for_tokens(code: str) -> Dict[str, Any]:
+    resp = httpx.post(
+        f"{SAXO_AUTH_BASE_URL}/token",
+        data={
+            "grant_type": "authorization_code",
+            "code": code,
+            "redirect_uri": SAXO_REDIRECT_URI,
+        },
+        auth=(SAXO_APP_KEY, SAXO_APP_SECRET),
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+        timeout=15,
+    )
+    resp.raise_for_status()
+    token_resp = resp.json()
+    _store_token_response(token_resp)
+    logger.info("[Saxo] Connected — tokens stored")
+    return token_resp
+
+
+def refresh_tokens() -> Optional[Dict[str, Any]]:
+    from services.database import get_saxo_tokens
+    stored = get_saxo_tokens()
+    if not stored or not stored.get("refresh_token"):
+        return None
+
+    resp = httpx.post(
+        f"{SAXO_AUTH_BASE_URL}/token",
+        data={
+            "grant_type": "refresh_token",
+            "refresh_token": stored["refresh_token"],
+            "redirect_uri": SAXO_REDIRECT_URI,
+        },
+        auth=(SAXO_APP_KEY, SAXO_APP_SECRET),
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+        timeout=15,
+    )
+    resp.raise_for_status()
+    token_resp = resp.json()
+    _store_token_response(token_resp)
+    logger.info("[Saxo] Token refreshed")
+    return token_resp
+
+
+def get_valid_access_token() -> Optional[str]:
+    from services.database import get_saxo_tokens
+    stored = get_saxo_tokens()
+    if not stored or not stored.get("access_token"):
+        return None
+
+    expires_at = datetime.fromisoformat(stored["expires_at"])
+    if datetime.now(timezone.utc) < expires_at - timedelta(seconds=60):
+        return stored["access_token"]
+
+    refreshed = refresh_tokens()
+    return refreshed["access_token"] if refreshed else None
+
+
+def get_status() -> Dict[str, Any]:
+    from services.database import get_saxo_tokens
+    stored = get_saxo_tokens()
+    if not stored or not stored.get("access_token"):
+        return {"connected": False, "configured": is_configured(), "environment": "sim"}
+    return {
+        "connected": True,
+        "configured": is_configured(),
+        "environment": stored.get("environment", "sim"),
+        "expires_at": stored.get("expires_at"),
+        "refresh_expires_at": stored.get("refresh_expires_at"),
+    }
+
+
+def disconnect():
+    from services.database import clear_saxo_tokens
+    clear_saxo_tokens()

backend/services/saxo_client.py

@@ -0,0 +1,182 @@
+"""
+Thin Saxo OpenAPI REST client for options chain snapshots.
+
+Two parts have high confidence (directly verified against Saxo's docs): the OAuth
+flow (services/saxo_auth.py) and the general shape of a streaming *subscription*
+call (ContextId/ReferenceId/Arguments, POST-then-DELETE for a one-off snapshot).
+
+One part has LOWER confidence and is written defensively on purpose: the exact
+field names returned by /ref/v1/instruments and /ref/v1/instruments/contractoptionspaces
+(Saxo's Swagger UI didn't expose the full response schema to static fetching). Every
+extraction below tries a couple of documented field-name variants and raises a
+clear error showing the raw keys received if none match — this is the one piece
+worth a quick calibration pass against a real response once SIM is connected.
+"""
+import logging
+import time
+import uuid
+from datetime import date
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+from services.saxo_auth import SAXO_API_BASE_URL, get_valid_access_token
+
+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] = {}
+
+
+class SaxoNotConnected(Exception):
+    pass
+
+
+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)
+    resp.raise_for_status()
+    return resp.json()
+
+
+def _first(d: Dict[str, Any], *keys: str) -> Any:
+    for k in keys:
+        if k in d:
+            return d[k]
+    return None
+
+
+def resolve_option_root_uic(symbol: str) -> int:
+    if symbol in _root_uic_cache:
+        return _root_uic_cache[symbol]
+
+    data = _get("/ref/v1/instruments", {"Keywords": symbol, "AssetTypes": _OPTION_ASSET_TYPES})
+    items = data.get("Data") or data.get("data") or []
+    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")
+    if uic is None:
+        raise ValueError(f"Champ Uic introuvable dans la réponse instruments pour '{symbol}': {items[0]}")
+
+    _root_uic_cache[symbol] = int(uic)
+    return int(uic)
+
+
+def get_option_space(root_uic: int) -> Dict[str, Any]:
+    return _get(f"/ref/v1/instruments/contractoptionspaces/{root_uic}")
+
+
+def _extract_option_legs(space: Dict[str, Any]) -> List[Dict[str, Any]]:
+    """Flatten the option space response into [{uic, expiry_date, strike, option_type}, ...]."""
+    expiries = _first(space, "OptionSpace", "SpecificOptions", "Expiries") or []
+    legs: List[Dict[str, Any]] = []
+
+    # Some Saxo response shapes nest strikes/sides under each expiry; others return a flat
+    # list of contracts directly. Handle both defensively.
+    for entry in expiries:
+        expiry_date = _first(entry, "Expiry", "ExpiryDate", "Date")
+        strikes = _first(entry, "SpecificOptions", "Strikes", "Options") or []
+        for opt in strikes:
+            uic = _first(opt, "Uic", "uic")
+            strike = _first(opt, "StrikePrice", "Strike")
+            side = _first(opt, "PutCall", "OptionType", "Side")
+            if uic is None or strike is None or side is None:
+                continue
+            legs.append({
+                "uic": int(uic),
+                "expiry_date": expiry_date,
+                "strike": float(strike),
+                "option_type": "put" if str(side).lower().startswith("p") else "call",
+            })
+
+    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": {"Uic": uic, "AssetType": asset_type},
+        "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,
+    )
+    resp.raise_for_status()
+    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 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}
+    """
+    root_uic = resolve_option_root_uic(symbol)
+    space = get_option_space(root_uic)
+    legs = _extract_option_legs(space)
+
+    snapshot = _snapshot_via_subscription(root_uic)
+    spot = _first(snapshot, "UnderlyingSpotPrice", "Spot", "UnderlyingPrice")
+    snapshot_date = date.today().isoformat()
+
+    by_uic = {leg["uic"]: leg for leg in legs}
+    rows: List[Dict[str, Any]] = []
+
+    for expiry_block in (_first(snapshot, "Expiries", "OptionsChain") or []):
+        for strike_block in (_first(expiry_block, "Strikes") or []):
+            for side_key in ("Call", "Put", "call", "put"):
+                side = strike_block.get(side_key)
+                if not side:
+                    continue
+                uic = _first(side, "Uic", "ContractId")
+                leg = by_uic.get(int(uic)) if uic is not None else None
+                rows.append({
+                    "symbol": symbol.upper(),
+                    "snapshot_date": snapshot_date,
+                    "spot": float(spot) if spot is not None else None,
+                    "expiry_date": _first(expiry_block, "Expiry", "ExpiryDate") or (leg["expiry_date"] if leg else None),
+                    "strike": float(_first(strike_block, "Strike", "StrikePrice") or (leg["strike"] if leg else 0)),
+                    "option_type": "put" if side_key.lower() == "put" else "call",
+                    "bid": _first(side, "Bid"),
+                    "ask": _first(side, "Ask"),
+                    "mid": _first(side, "Mid"),
+                    "volatility_pct": _first(strike_block, "MidVolatilityPct", "VolatilityPct"),
+                    "delta": _first(side, "DeltaPct", "Delta"),
+                    "gamma": _first(side, "Gamma"),
+                    "theta": _first(side, "Theta"),
+                    "vega": _first(side, "Vega"),
+                })
+
+    if not rows:
+        raise ValueError(f"Snapshot Saxo vide pour '{symbol}' (clés reçues: {list(snapshot.keys())})")
+    return rows

backend/services/saxo_scheduler.py

@@ -0,0 +1,97 @@
+"""
+Background scheduler for the Saxo connection — mirrors services/var_scheduler.py.
+
+Two independent loops:
+- Token refresh: proactively renews the access/refresh token well before the 40-min
+  refresh-token expiry (SIM), so the user never has to re-login as long as the
+  server keeps running.
+- Snapshot poller: periodically captures an options-chain snapshot for each symbol
+  in the configured watchlist and appends it to saxo_option_snapshots.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import threading
+
+logger = logging.getLogger(__name__)
+
+_refresh_thread: threading.Thread | None = None
+_snapshot_thread: threading.Thread | None = None
+_refresh_stop = threading.Event()
+_snapshot_stop = threading.Event()
+
+DEFAULT_WATCHLIST = ["SPY", "QQQ", "GLD", "SLV", "USO"]
+REFRESH_INTERVAL_SECONDS = 10 * 60  # well under the 40-min refresh-token lifetime
+
+
+def get_watchlist() -> list[str]:
+    from .database import get_config
+    raw = get_config("saxo_watchlist")
+    if not raw:
+        return DEFAULT_WATCHLIST
+    try:
+        return json.loads(raw)
+    except Exception:
+        return DEFAULT_WATCHLIST
+
+
+def set_watchlist(symbols: list[str]):
+    from .database import set_config
+    set_config("saxo_watchlist", json.dumps([s.upper() for s in symbols]))
+
+
+def _refresh_loop(stop: threading.Event):
+    while not stop.wait(0):
+        try:
+            from .saxo_auth import get_status, refresh_tokens
+            if get_status().get("connected"):
+                refresh_tokens()
+                logger.info("[Saxo Scheduler] Token refreshed proactively")
+        except Exception as e:
+            logger.error(f"[Saxo Scheduler] Refresh loop error: {e}")
+        stop.wait(timeout=REFRESH_INTERVAL_SECONDS)
+
+
+def _snapshot_loop(stop: threading.Event):
+    while not stop.wait(0):
+        try:
+            from .database import get_config
+            from .saxo_auth import get_status
+            minutes = float(get_config("saxo_snapshot_minutes") or "20")
+
+            if get_status().get("connected"):
+                from .saxo_client import snapshot_options_chain
+                from .database import save_saxo_snapshot_rows
+                for symbol in get_watchlist():
+                    try:
+                        rows = snapshot_options_chain(symbol)
+                        save_saxo_snapshot_rows(rows)
+                        logger.info(f"[Saxo Scheduler] Snapshot saved: {symbol} ({len(rows)} rows)")
+                    except Exception as e:
+                        logger.warning(f"[Saxo Scheduler] Snapshot failed for {symbol}: {e}")
+            else:
+                minutes = max(minutes, 5)  # don't busy-loop while disconnected
+        except Exception as e:
+            logger.error(f"[Saxo Scheduler] Snapshot loop error: {e}")
+            minutes = 20
+
+        stop.wait(timeout=minutes * 60)
+
+
+def start_saxo_scheduler():
+    global _refresh_thread, _snapshot_thread
+    _refresh_stop.clear()
+    _snapshot_stop.clear()
+    if not (_refresh_thread and _refresh_thread.is_alive()):
+        _refresh_thread = threading.Thread(target=_refresh_loop, args=(_refresh_stop,), name="saxo-refresh", daemon=True)
+        _refresh_thread.start()
+    if not (_snapshot_thread and _snapshot_thread.is_alive()):
+        _snapshot_thread = threading.Thread(target=_snapshot_loop, args=(_snapshot_stop,), name="saxo-snapshot", daemon=True)
+        _snapshot_thread.start()
+    logger.info("[Saxo Scheduler] Started")
+
+
+def stop_saxo_scheduler():
+    _refresh_stop.set()
+    _snapshot_stop.set()