"""Read-side queries for the market API (candles, ticker, book, trades, data quality, features)."""

from __future__ import annotations

import time
from typing import Any

import asyncpg
import numpy as np
import pandas as pd

from ktrade.db.pool import rows_to_dicts


async def list_pairs(pool: asyncpg.Pool, universe_only: bool = True, q: str | None = None) -> list[dict[str, Any]]:
    sql = "SELECT pair_id, altname, wsname, base, quote, status, ordermin, costmin, tick_size, pair_decimals, lot_decimals, cost_decimals, leverage_buy, leverage_sell, in_universe, updated_at FROM pairs"
    cond, args = [], []
    if universe_only:
        cond.append("in_universe")
    if q:
        args.append(f"%{q.upper()}%")
        cond.append(f"(wsname ILIKE ${len(args)} OR altname ILIKE ${len(args)})")
    if cond:
        sql += " WHERE " + " AND ".join(cond)
    sql += " ORDER BY in_universe DESC, wsname LIMIT 500"
    out = rows_to_dicts(await pool.fetch(sql, *args))
    for r in out:
        for k in ("ordermin", "costmin", "tick_size"):
            r[k] = float(r[k])
    return out


async def get_candles(pool: asyncpg.Pool, symbol: str, interval: int, limit: int = 1000, start: int | None = None, end: int | None = None, closed_only: bool = False) -> list[dict[str, Any]]:
    cond = ["symbol=$1", "interval_min=$2"]
    args: list[Any] = [symbol, interval]
    if start is not None:
        args.append(start)
        cond.append(f"ts >= ${len(args)}")
    if end is not None:
        args.append(end)
        cond.append(f"ts <= ${len(args)}")
    if closed_only:
        cond.append("closed")
    args.append(min(max(limit, 1), 20000))
    rows = await pool.fetch(f"SELECT ts, open, high, low, close, vwap, volume, count, closed FROM candles WHERE {' AND '.join(cond)} ORDER BY ts DESC LIMIT ${len(args)}", *args)
    return [dict(r) for r in reversed(rows)]


async def candles_frame(pool: asyncpg.Pool, symbol: str, interval: int, start: int | None = None, end: int | None = None) -> pd.DataFrame:
    rows = await get_candles(pool, symbol, interval, limit=20000, start=start, end=end, closed_only=True)
    if not rows:
        return pd.DataFrame(columns=["ts", "open", "high", "low", "close", "vwap", "volume", "count"])
    df = pd.DataFrame(rows).drop(columns=["closed"])
    return df.sort_values("ts").reset_index(drop=True)


async def ticker(pool: asyncpg.Pool, symbol: str | None = None) -> list[dict[str, Any]]:
    if symbol:
        rows = await pool.fetch("SELECT symbol, payload, received_at FROM ticker_latest WHERE symbol=$1", symbol)
    else:
        rows = await pool.fetch("SELECT symbol, payload, received_at FROM ticker_latest ORDER BY symbol")
    now = time.time()
    return [{**r["payload"], "received_at": r["received_at"], "age_s": round(now - r["received_at"], 1)} for r in rows]


async def book(pool: asyncpg.Pool, symbol: str) -> dict[str, Any] | None:
    r = await pool.fetchrow("SELECT payload, received_at, valid FROM book_latest WHERE symbol=$1", symbol)
    if not r:
        return None
    return {**r["payload"], "received_at": r["received_at"], "age_s": round(time.time() - r["received_at"], 1), "valid": r["valid"]}


async def recent_trades(pool: asyncpg.Pool, symbol: str, limit: int = 100) -> list[dict[str, Any]]:
    rows = await pool.fetch("SELECT trade_id, ts, price, volume, side, order_type FROM trades WHERE symbol=$1 ORDER BY ts DESC LIMIT $2", symbol, min(limit, 1000))
    return rows_to_dicts(rows)


async def book_samples(pool: asyncpg.Pool, symbol: str, seconds: int = 3600) -> list[dict[str, Any]]:
    since = int(time.time()) - seconds
    rows = await pool.fetch("SELECT ts, best_bid, best_ask, spread_bps, bid_depth, ask_depth, imbalance FROM book_samples WHERE symbol=$1 AND ts>=$2 ORDER BY ts", symbol, since)
    return rows_to_dicts(rows)


async def data_quality(pool: asyncpg.Pool, stale_after: int) -> list[dict[str, Any]]:
    rows = await pool.fetch(
        """SELECT c.symbol, c.interval_min, count(*) AS bars, min(ts) AS first_ts, max(ts) AS last_ts,
                  (SELECT count(*) FROM data_gaps g WHERE g.symbol=c.symbol AND g.interval_min=c.interval_min) AS gaps,
                  (SELECT coalesce(sum(missing_bars),0) FROM data_gaps g WHERE g.symbol=c.symbol AND g.interval_min=c.interval_min) AS missing_bars
           FROM candles c WHERE closed GROUP BY c.symbol, c.interval_min ORDER BY c.symbol, c.interval_min"""
    )
    now = int(time.time())
    out = []
    for r in rows:
        d = dict(r)
        iv = d["interval_min"]
        expected_last = (now // (iv * 60) - 1) * iv * 60
        d["lag_bars"] = max(0, (expected_last - d["last_ts"]) // (iv * 60))
        d["stale"] = d["lag_bars"] * iv * 60 > max(stale_after, iv * 60 * 2)
        d["coverage_days"] = round((d["last_ts"] - d["first_ts"]) / 86400, 2)
        out.append(d)
    return out


def realized_vol(df: pd.DataFrame, interval_min: int, window: int = 30) -> dict[str, float | None]:
    if len(df) < window + 2:
        return {"rv_window": None, "rv_annualized": None}
    r = np.log(df["close"]).diff().dropna().tail(window)
    per_year = 365 * 24 * 60 / interval_min
    rv = float(r.std(ddof=1))
    return {"rv_window": rv, "rv_annualized": rv * np.sqrt(per_year)}


async def overview(pool: asyncpg.Pool, universe: list[str], stale_after: int) -> dict[str, Any]:
    tick = {t["symbol"]: t for t in await ticker(pool)}
    out = []
    for s in universe:
        t = tick.get(s)
        df = await candles_frame(pool, s, 60)
        rv = realized_vol(df, 60, 24 * 7) if not df.empty else {"rv_window": None, "rv_annualized": None}
        out.append({
            "symbol": s, "last": t.get("last") if t else None, "change_pct": t.get("change_pct") if t else None,
            "bid": t.get("bid") if t else None, "ask": t.get("ask") if t else None, "volume": t.get("volume") if t else None,
            "spread_bps": ((t["ask"] - t["bid"]) / ((t["ask"] + t["bid"]) / 2) * 1e4) if t and t.get("bid") and t.get("ask") else None,
            "age_s": t.get("age_s") if t else None, "stale": (t is None) or (t["age_s"] > stale_after), "rv_7d_annualized": rv["rv_annualized"],
            "hourly_bars": int(len(df)),
        })
    return {"symbols": out}