"""ParlayAPI WebSocket reference client (Python).

Demonstrates the correct way to consume `/v1/ws/odds-fast/{sport_key}`:
  * Reconnect with capped exponential backoff
  * Distinguish between three causes of silence:
      1. Connection died (no heartbeats arriving)
      2. Market is quiet (heartbeats arriving, upstream healthy)
      3. Upstream is degraded (heartbeats arriving, upstream worst_sla
         not "ok")
  * Honor the server's `quiet_seconds` and `upstream` fields
  * Send `subscribe` to filter broadcasts to one event_id
  * Send periodic `ping` and read the `pong` for application-level
    health when the heartbeat itself is too coarse

Use as a copy-paste starting point. The handler functions are the
spots you'd customize for your own pipeline (price storage, alerts,
trade decisions). Everything else is plumbing.

Requires: websockets >= 12
    pip install 'websockets>=12,<14'

Run:
    PARLAYAPI_KEY=your_key SPORT_KEY=baseball_mlb python ws_reference_client.py
"""
from __future__ import annotations

import asyncio
import json
import os
import sys
import time
from typing import Any

try:
    import websockets
    from websockets.exceptions import ConnectionClosed, InvalidStatus
except ImportError:
    print(
        "websockets not installed. Run: pip install 'websockets>=12,<14'",
        file=sys.stderr,
    )
    sys.exit(2)


# Connection knobs. Defaults are tuned for the live broadcast cadence;
# adjust if you have specific latency / cost targets.
API_HOST = os.environ.get("PARLAYAPI_HOST", "parlay-api.com")
API_KEY = os.environ.get("PARLAYAPI_KEY", "").strip()
SPORT_KEY = os.environ.get("SPORT_KEY", "baseball_mlb").strip()
PING_INTERVAL_S = float(os.environ.get("PING_INTERVAL_S", "30"))
# Silent watchdog: if we don't receive ANY message (incl. heartbeat)
# in this window, assume the connection is dead and force a reconnect.
# The server heartbeat fires every 5s so anything past 30s is real silence.
SILENT_WATCHDOG_S = float(os.environ.get("SILENT_WATCHDOG_S", "30"))
RECONNECT_BASE_S = 1.0
RECONNECT_CAP_S = 60.0


def on_odds_update(envelope: dict[str, Any]) -> None:
    """Customize this for your pipeline.

    Envelope shape (representative):
      {
        "type": "odds_update",
        "event_id": "...",
        "sport_key": "baseball_mlb",
        "home_team": "Yankees",
        "away_team": "Red Sox",
        "market": "h2h",
        "side": "home",
        "bookmaker": "draftkings",
        "price": -135,
        "implied_prob": 0.5745,
        "timestamp_ms": 1779100000000,
        ...
      }
    """
    print(
        f"odds_update {envelope.get('sport_key')} "
        f"{envelope.get('home_team')} v {envelope.get('away_team')} "
        f"{envelope.get('market')}/{envelope.get('side')} "
        f"@ {envelope.get('bookmaker')} = {envelope.get('price')}",
        flush=True,
    )


def on_heartbeat(envelope: dict[str, Any]) -> None:
    """Diagnostic helper for the heartbeat shape.

    Heartbeat shape:
      {
        "type": "heartbeat",
        "timestamp": 1779100000,
        "connections": 23,
        "quiet_seconds": 12.3,           # time since last push to YOU
        "upstream": {
          "worst_sla": "ok" | "degraded" | "breach" | "stale" | "missing",
          "counts": {"ok": 6, "degraded": 9, ...},
          "generated_at_ms": ...
        }
      }

    Two signals to act on:
      * quiet_seconds high + upstream.worst_sla == "ok" -> market is
        quiet, your connection is fine, don't reconnect.
      * quiet_seconds high + upstream.worst_sla in
        {"stale", "breach"} -> our upstream is degraded; reconnecting
        won't help, just wait or fall back to a different sport_key.
    """
    quiet = envelope.get("quiet_seconds")
    upstream = envelope.get("upstream") or {}
    worst = upstream.get("worst_sla", "?")
    if quiet is not None and quiet > 30 and worst != "ok":
        print(
            f"heartbeat: quiet for {quiet:.1f}s; upstream worst_sla={worst}",
            flush=True,
        )


def on_pong(envelope: dict[str, Any]) -> None:
    pass


def on_subscribed(envelope: dict[str, Any]) -> None:
    print(f"subscribed: event_id={envelope.get('event_id')}", flush=True)


HANDLERS = {
    "odds_update": on_odds_update,
    "heartbeat":   on_heartbeat,
    "pong":        on_pong,
    "subscribed":  on_subscribed,
}


async def run_one_session() -> None:
    """One WebSocket session. Returns when the connection ends for any
    reason. The outer loop reconnects."""
    if not API_KEY:
        print("PARLAYAPI_KEY not set", file=sys.stderr)
        raise SystemExit(2)

    url = (
        f"wss://{API_HOST}/v1/ws/odds-fast/{SPORT_KEY}"
        f"?apiKey={API_KEY}"
    )
    print(f"connecting to {url.split('?')[0]}", flush=True)

    last_msg_ts = time.time()
    last_ping_ts = time.time()

    async with websockets.connect(url, ping_interval=None) as ws:
        print("connected", flush=True)
        # Send an explicit application-level ping every PING_INTERVAL_S.
        # The protocol-level ping/pong (ping_interval) is disabled
        # because we want our own pong with quiet_seconds + upstream.
        async def pinger():
            nonlocal last_ping_ts
            while True:
                await asyncio.sleep(PING_INTERVAL_S)
                try:
                    await ws.send(json.dumps({"type": "ping"}))
                    last_ping_ts = time.time()
                except Exception:
                    return

        async def watchdog():
            while True:
                await asyncio.sleep(5)
                idle = time.time() - last_msg_ts
                if idle > SILENT_WATCHDOG_S:
                    print(
                        f"watchdog: no message in {idle:.1f}s, forcing reconnect",
                        flush=True,
                    )
                    try:
                        await ws.close(code=1000, reason="silent watchdog")
                    except Exception:
                        pass
                    return

        ping_task = asyncio.create_task(pinger())
        watchdog_task = asyncio.create_task(watchdog())

        try:
            async for raw in ws:
                last_msg_ts = time.time()
                try:
                    envelope = json.loads(raw)
                except json.JSONDecodeError:
                    continue
                t = envelope.get("type")
                handler = HANDLERS.get(t)
                if handler:
                    handler(envelope)
                else:
                    # Unknown frame type. Log so we notice if the
                    # server adds new types we should handle.
                    print(f"unknown frame type: {t}", flush=True)
        finally:
            for task in (ping_task, watchdog_task):
                task.cancel()
                try:
                    await task
                except Exception:
                    pass


async def main() -> int:
    """Outer reconnect loop with capped exponential backoff."""
    backoff = RECONNECT_BASE_S
    while True:
        try:
            await run_one_session()
            # Clean exit from the inner session: backoff resets.
            backoff = RECONNECT_BASE_S
            print("session ended cleanly, reconnecting in 1s", flush=True)
            await asyncio.sleep(RECONNECT_BASE_S)
        except (ConnectionClosed, InvalidStatus, OSError) as e:
            print(
                f"session ended: {type(e).__name__}: {str(e)[:120]}; "
                f"reconnect in {backoff:.1f}s",
                flush=True,
            )
            await asyncio.sleep(backoff)
            backoff = min(RECONNECT_CAP_S, backoff * 2)
        except KeyboardInterrupt:
            print("interrupted", flush=True)
            return 0
        except Exception as e:
            print(
                f"unexpected: {type(e).__name__}: {str(e)[:160]}; "
                f"reconnect in {backoff:.1f}s",
                flush=True,
            )
            await asyncio.sleep(backoff)
            backoff = min(RECONNECT_CAP_S, backoff * 2)


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))