from __future__ import annotations

import asyncio
import csv
import json
from collections.abc import AsyncIterator, Iterator
from dataclasses import dataclass, field
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any

import websockets

from .models import Candle


TESTNET_STREAM_BASE_URL = "wss://stream.testnet.binance.vision/ws"
BPS = Decimal("10000")


def _parse_int_kline_field(kline: dict, field: str) -> int:
    try:
        return int(kline[field])
    except (TypeError, ValueError) as exc:
        raise ValueError(f"invalid kline field: {field}") from exc


def _parse_decimal_kline_field(kline: dict, field: str) -> Decimal:
    try:
        return Decimal(kline[field])
    except (InvalidOperation, TypeError, ValueError) as exc:
        raise ValueError(f"invalid kline field: {field}") from exc


def kline_payload_to_candle(payload: dict) -> Candle | None:
    kline = payload.get("k", {})
    if kline.get("x") is not True:
        return None
    symbol = payload.get("s") or kline.get("s")
    if not symbol:
        raise ValueError("closed kline payload missing symbol")

    required_fields = ("t", "T", "o", "h", "l", "c", "v")
    for field in required_fields:
        if field not in kline:
            raise ValueError(f"closed kline payload missing kline field: {field}")

    return Candle(
        symbol=symbol,
        open_time=_parse_int_kline_field(kline, "t"),
        close_time=_parse_int_kline_field(kline, "T"),
        open=_parse_decimal_kline_field(kline, "o"),
        high=_parse_decimal_kline_field(kline, "h"),
        low=_parse_decimal_kline_field(kline, "l"),
        close=_parse_decimal_kline_field(kline, "c"),
        volume=_parse_decimal_kline_field(kline, "v"),
    )


def load_candles_csv(path: str | Path, symbol: str) -> Iterator[Candle]:
    with Path(path).open("r", encoding="utf-8", newline="") as handle:
        reader = csv.DictReader(handle)
        for row in reader:
            yield Candle(
                symbol=symbol,
                open_time=int(row["open_time"]),
                close_time=int(row["close_time"]),
                open=Decimal(row["open"]),
                high=Decimal(row["high"]),
                low=Decimal(row["low"]),
                close=Decimal(row["close"]),
                volume=Decimal(row["volume"]),
            )


def _decimal_text(value: Decimal | None) -> str | None:
    if value is None:
        return None
    return format(value.normalize(), "f")


def _ratio(numerator: Decimal, denominator: Decimal) -> Decimal | None:
    if denominator == 0:
        return None
    return numerator / denominator


@dataclass
class _RealtimeAccumulator:
    sample_seconds: int
    bookticker_updates: int = 0
    depth_updates: int = 0
    agg_trade_updates: int = 0
    spreads: list[Decimal] = field(default_factory=list)
    best_bid_quantities: list[Decimal] = field(default_factory=list)
    best_ask_quantities: list[Decimal] = field(default_factory=list)
    depth_bid_notionals: list[Decimal] = field(default_factory=list)
    depth_ask_notionals: list[Decimal] = field(default_factory=list)
    book_imbalances: list[Decimal] = field(default_factory=list)
    agg_trade_buy_quote_volume: Decimal = Decimal("0")
    agg_trade_sell_quote_volume: Decimal = Decimal("0")
    large_trade_count: int = 0
    large_trade_quote_sum: Decimal = Decimal("0")

    def apply_book_ticker(self, payload: dict[str, Any]) -> None:
        bid = Decimal(str(payload.get("b", "0")))
        ask = Decimal(str(payload.get("a", "0")))
        bid_qty = Decimal(str(payload.get("B", "0")))
        ask_qty = Decimal(str(payload.get("A", "0")))
        midpoint = (bid + ask) / Decimal("2")
        if midpoint:
            self.spreads.append(((ask - bid) / midpoint) * BPS)
        self.best_bid_quantities.append(bid_qty)
        self.best_ask_quantities.append(ask_qty)
        self.bookticker_updates += 1

    def apply_depth(self, payload: dict[str, Any]) -> None:
        bid_notional = _sum_book_side(payload.get("bids", []))
        ask_notional = _sum_book_side(payload.get("asks", []))
        total = bid_notional + ask_notional
        self.depth_bid_notionals.append(bid_notional)
        self.depth_ask_notionals.append(ask_notional)
        if total:
            self.book_imbalances.append((bid_notional - ask_notional) / total)
        self.depth_updates += 1

    def apply_agg_trade(self, payload: dict[str, Any]) -> None:
        price = Decimal(str(payload.get("p", "0")))
        quantity = Decimal(str(payload.get("q", "0")))
        quote = price * quantity
        if bool(payload.get("m", False)):
            self.agg_trade_sell_quote_volume += quote
        else:
            self.agg_trade_buy_quote_volume += quote
        if quote >= Decimal("50"):
            self.large_trade_count += 1
            self.large_trade_quote_sum += quote
        self.agg_trade_updates += 1

    def features(self, status: str = "ok") -> dict[str, Any]:
        return {
            "status": status,
            "sample_seconds": self.sample_seconds,
            "bookticker_updates": self.bookticker_updates,
            "depth_updates": self.depth_updates,
            "agg_trade_updates": self.agg_trade_updates,
            "spread_min_bps": _decimal_text(min(self.spreads) if self.spreads else None),
            "spread_max_bps": _decimal_text(max(self.spreads) if self.spreads else None),
            "spread_avg_bps": _decimal_text(_average(self.spreads)),
            "best_bid_qty_avg": _decimal_text(_average(self.best_bid_quantities)),
            "best_ask_qty_avg": _decimal_text(_average(self.best_ask_quantities)),
            "depth_bid_notional_avg": _decimal_text(_average(self.depth_bid_notionals)),
            "depth_ask_notional_avg": _decimal_text(_average(self.depth_ask_notionals)),
            "book_imbalance_avg": _decimal_text(_average(self.book_imbalances)),
            "agg_trade_buy_quote_volume": _decimal_text(self.agg_trade_buy_quote_volume),
            "agg_trade_sell_quote_volume": _decimal_text(self.agg_trade_sell_quote_volume),
            "agg_trade_buy_sell_ratio": _decimal_text(
                _ratio(self.agg_trade_buy_quote_volume, self.agg_trade_sell_quote_volume)
            ),
            "large_trade_count": self.large_trade_count,
            "large_trade_quote_sum": _decimal_text(self.large_trade_quote_sum),
        }


def _sum_book_side(rows: object) -> Decimal:
    if not isinstance(rows, list):
        return Decimal("0")
    total = Decimal("0")
    for row in rows:
        if not isinstance(row, list) or len(row) < 2:
            continue
        total += Decimal(str(row[0])) * Decimal(str(row[1]))
    return total


def _average(values: list[Decimal]) -> Decimal | None:
    if not values:
        return None
    return sum(values, Decimal("0")) / Decimal(len(values))


def _empty_realtime_features(sample_seconds: int, status: str) -> dict[str, Any]:
    return _RealtimeAccumulator(sample_seconds=sample_seconds).features(status=status)


def _combined_stream_url(symbols: list[str], stream_base_url: str) -> str:
    base = stream_base_url.rstrip("/")
    if base.endswith("/ws"):
        base = base[:-3]
    streams = []
    for symbol in symbols:
        lowered = symbol.lower()
        streams.extend([f"{lowered}@bookTicker", f"{lowered}@depth20@100ms", f"{lowered}@aggTrade"])
    return f"{base}/stream?streams={'/'.join(streams)}"


def _stream_symbol(stream: str, payload: dict[str, Any]) -> str | None:
    symbol = payload.get("s")
    if symbol:
        return str(symbol).upper()
    if "@" in stream:
        return stream.split("@", 1)[0].upper()
    return None


async def sample_realtime_market(
    symbols: list[str] | tuple[str, ...],
    *,
    stream_base_url: str = TESTNET_STREAM_BASE_URL,
    sample_seconds: int = 15,
) -> dict[str, dict[str, Any]]:
    normalized_symbols = [symbol.upper() for symbol in symbols]
    accumulators = {
        symbol: _RealtimeAccumulator(sample_seconds=sample_seconds)
        for symbol in normalized_symbols
    }
    try:
        async with websockets.connect(_combined_stream_url(normalized_symbols, stream_base_url)) as websocket:
            try:
                async with asyncio.timeout(sample_seconds):
                    async for raw_message in websocket:
                        _apply_realtime_message(raw_message, accumulators)
            except TimeoutError:
                pass
    except Exception:
        return {
            symbol: _empty_realtime_features(sample_seconds, "unavailable")
            for symbol in normalized_symbols
        }
    return {symbol: accumulator.features() for symbol, accumulator in accumulators.items()}


def _apply_realtime_message(raw_message: str, accumulators: dict[str, _RealtimeAccumulator]) -> None:
    try:
        message = json.loads(raw_message)
    except json.JSONDecodeError:
        return
    if not isinstance(message, dict):
        return
    stream = str(message.get("stream", ""))
    payload = message.get("data", message)
    if not isinstance(payload, dict):
        return
    symbol = _stream_symbol(stream, payload)
    if symbol not in accumulators:
        return
    accumulator = accumulators[symbol]
    try:
        if "bookTicker" in stream:
            accumulator.apply_book_ticker(payload)
        elif "depth" in stream:
            accumulator.apply_depth(payload)
        elif "aggTrade" in stream:
            accumulator.apply_agg_trade(payload)
    except (InvalidOperation, TypeError, ValueError):
        return


async def stream_closed_klines(
    symbol: str,
    interval: str = "1m",
    stream_base_url: str = TESTNET_STREAM_BASE_URL,
) -> AsyncIterator[Candle]:
    stream_name = f"{symbol.lower()}@kline_{interval}"
    url = f"{stream_base_url.rstrip('/')}/{stream_name}"
    async with websockets.connect(url) as websocket:
        async for raw_message in websocket:
            payload = json.loads(raw_message)
            candle = kline_payload_to_candle(payload)
            if candle is not None:
                yield candle