from __future__ import annotations

import asyncio
from dataclasses import dataclass
from datetime import datetime
from typing import Dict, Set, Tuple, Any

import httpx
import asyncio
import time
import random

from app.config import settings
from app.streaming.live_store import live_store

ALPHA_BASE = "https://www.alphavantage.co/query"


@dataclass(frozen=True)
class AVKey:
    symbol: str  # e.g., XAUUSD
    timeframe: str  # '1m' only for hub


class AlphaVantageHub:
    """
    Polls Alpha Vantage FX_INTRADAY for the latest bar per (symbol, 1m),
    fans out to subscribers via asyncio.Queue, and ingests into live_store.
    Ensures a single poller per (symbol,timeframe).
    """

    def __init__(self) -> None:
        self._subs: Dict[AVKey, Set[asyncio.Queue]] = {}
        self._tasks: Dict[AVKey, asyncio.Task] = {}
        self._lock = asyncio.Lock()
        # Conditional request caches
        self._etags: Dict[AVKey, str] = {}
        self._last_mod: Dict[AVKey, str] = {}
        # Global token-bucket (Alpha free tier ~5 req/min)
        self._rate_lock = asyncio.Lock()
        self._tokens: float = 5.0
        self._max_tokens: float = 5.0
        self._refill_rate_per_sec: float = 5.0 / 60.0
        self._last_refill_ts: float = time.time()

    async def _acquire_token(self) -> None:
        # Simple async token bucket
        while True:
            async with self._rate_lock:
                now = time.time()
                elapsed = now - self._last_refill_ts
                if elapsed > 0:
                    self._tokens = min(self._max_tokens, self._tokens + elapsed * self._refill_rate_per_sec)
                    self._last_refill_ts = now
                if self._tokens >= 1.0:
                    self._tokens -= 1.0
                    return
                # Not enough tokens, compute wait time for next token
                need = 1.0 - self._tokens
                wait = max(0.1, need / self._refill_rate_per_sec)
            await asyncio.sleep(min(wait, 5.0))

    def get_status(self) -> list[dict]:
        out: list[dict] = []
        for key, subs in self._subs.items():
            hist = live_store.get_history(key.symbol, key.timeframe)
            last_ts = hist[-1]["time"] if hist else None
            last_iso = None
            if isinstance(last_ts, (int, float)):
                try:
                    last_iso = datetime.utcfromtimestamp(int(last_ts)).isoformat() + "Z"
                except Exception:
                    last_iso = None
            out.append({
                "symbol": key.symbol,
                "timeframe": key.timeframe,
                "subscribers": len(subs),
                "last_event_time": last_iso,
            })
        return out

    async def subscribe(self, symbol: str, timeframe: str = "1m") -> Tuple[asyncio.Queue, Any]:
        if timeframe != "1m":
            raise ValueError("AlphaVantageHub currently supports timeframe '1m' only")
        key = AVKey(symbol=symbol.upper().replace("/", ""), timeframe=timeframe)
        q: asyncio.Queue = asyncio.Queue(maxsize=100)
        async with self._lock:
            subs = self._subs.get(key)
            if not subs:
                subs = set()
                self._subs[key] = subs
            subs.add(q)
            if key not in self._tasks:
                self._tasks[key] = asyncio.create_task(self._run_poller(key))

        async def _unsubscribe() -> None:
            async with self._lock:
                s = self._subs.get(key)
                if s and q in s:
                    s.remove(q)
                try:
                    q.put_nowait(None)
                except Exception:
                    pass
                if s is not None and len(s) == 0:
                    t = self._tasks.pop(key, None)
                    if t:
                        t.cancel()
                    self._subs.pop(key, None)
        return q, _unsubscribe

    async def _run_poller(self, key: AVKey) -> None:
        symbol = key.symbol
        from_symbol = symbol[:3]
        to_symbol = symbol[3:]
        apikey = settings.ALPHA_VANTAGE_API_KEY or "demo"
        last_ts: int | None = None
        poll_interval = 60  # seconds
        backoff_cap = 300    # max 5 min
        async with httpx.AsyncClient(timeout=30) as client:
            while True:
                try:
                    await self._acquire_token()
                    params = {
                        "function": "FX_INTRADAY",
                        "from_symbol": from_symbol,
                        "to_symbol": to_symbol,
                        "interval": "1min",
                        "outputsize": "compact",
                        "apikey": apikey,
                    }
                    headers = {}
                    et = self._etags.get(key)
                    lm = self._last_mod.get(key)
                    if et:
                        headers["If-None-Match"] = et
                    if lm:
                        headers["If-Modified-Since"] = lm
                    r = await client.get(ALPHA_BASE, params=params, headers=headers)
                    if r.status_code == 304:
                        # Not modified, keep interval
                        delay = poll_interval + random.uniform(0, 2)
                        await asyncio.sleep(delay)
                        continue
                    # Raise for other non-2xx
                    r.raise_for_status()
                    # Store caching headers for next time
                    etag = r.headers.get("ETag")
                    if etag:
                        self._etags[key] = etag
                    last_mod = r.headers.get("Last-Modified")
                    if last_mod:
                        self._last_mod[key] = last_mod
                    js = r.json()
                    series = js.get("Time Series FX (1min)") or {}
                    if series:
                        latest_ts_str = max(series.keys())
                        dt = datetime.fromisoformat(latest_ts_str)
                        tsec = int(dt.timestamp())
                        if last_ts is None or tsec > last_ts:
                            row = series[latest_ts_str]
                            evt = {
                                "symbol": symbol,
                                "timeframe": key.timeframe,
                                "open_time": dt.isoformat(),
                                "close_time": dt.isoformat(),
                                "open": float(row["1. open"]),
                                "high": float(row["2. high"]),
                                "low": float(row["3. low"]),
                                "close": float(row["4. close"]),
                                "volume": float(row.get("5. volume", 0.0)),
                                "is_closed": True,
                                "source": "alpha_vantage",
                            }
                            try:
                                live_store.ingest_bar(symbol=symbol, timeframe="1m", bar={
                                    "time": tsec,
                                    "open": evt["open"],
                                    "high": evt["high"],
                                    "low": evt["low"],
                                    "close": evt["close"],
                                    "volume": evt["volume"],
                                })
                            except Exception:
                                pass
                            subs = self._subs.get(key) or set()
                            for q in list(subs):
                                try:
                                    if q.full():
                                        q.get_nowait()
                                    q.put_nowait(evt)
                                except Exception:
                                    try:
                                        subs.remove(q)
                                    except Exception:
                                        pass
                            last_ts = tsec
                    # success -> reset interval
                    poll_interval = 60
                except httpx.HTTPStatusError as e:
                    status = e.response.status_code if e.response else None
                    # 429 or 5xx -> exponential backoff
                    if status == 429 or (status and 500 <= status < 600):
                        poll_interval = min(backoff_cap, max(60, int(poll_interval * 2)))
                    # else, keep interval
                except Exception:
                    # network or parse error
                    poll_interval = min(backoff_cap, max(60, int(poll_interval * 2)))
                # sleep with small jitter
                delay = poll_interval + random.uniform(0, 2)
                await asyncio.sleep(delay)


# Singleton hub
alpha_hub = AlphaVantageHub()
