from __future__ import annotations

import argparse
import json
import os
import time
from collections import deque
from datetime import datetime, timezone
from decimal import Decimal
from pathlib import Path
from typing import Any

import websocket

WS_URL = "wss://fstream.binance.com/ws/!forceOrder@arr"
DEFAULT_OUTPUT = Path("runtime/research/binance_liquidations_latest.json")
DEFAULT_JOURNAL = Path("runtime/research/binance_liquidations.jsonl")
DEFAULT_PID = Path("runtime/research/binance_liquidation_collector.pid")
WATCHED = {"BTCUSDT": "BTC", "ETHUSDT": "ETH", "SOLUSDT": "SOL", "LINKUSDT": "LINK"}


def _d(value: Any) -> Decimal:
    try:
        return Decimal(str(value))
    except Exception:
        return Decimal("0")


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def normalize_event(payload: dict[str, Any]) -> dict[str, Any] | None:
    raw_order = payload.get("o")
    order: dict[str, Any] = raw_order if isinstance(raw_order, dict) else payload
    symbol = str(order.get("s") or "").upper()
    coin = WATCHED.get(symbol)
    if not coin:
        return None
    price = _d(order.get("ap") or order.get("p"))
    quantity = _d(order.get("z") or order.get("q"))
    side = str(order.get("S") or "").upper()
    return {"timestamp": _now(), "event_time_ms": int(payload.get("E") or order.get("T") or time.time() * 1000), "coin": coin, "symbol": symbol, "forced_order_side": side, "liquidated_position_side": "long" if side == "SELL" else "short" if side == "BUY" else "unknown", "price": str(price), "quantity": str(quantity), "notional_usd": str(price * quantity), "source_id": "binance_force_order_websocket", "research_only": True}


def aggregate(events: deque[dict[str, Any]], *, window_seconds: int) -> dict[str, Any]:
    cutoff_ms = int((time.time() - window_seconds) * 1000)
    while events and int(events[0].get("event_time_ms") or 0) < cutoff_ms:
        events.popleft()
    coins: dict[str, dict[str, Decimal | int]] = {}
    for event in events:
        row = coins.setdefault(str(event["coin"]), {"liquidation_notional_usd": Decimal("0"), "long_liquidated_usd": Decimal("0"), "short_liquidated_usd": Decimal("0"), "event_count": 0})
        notional = _d(event.get("notional_usd"))
        row["liquidation_notional_usd"] = _d(row["liquidation_notional_usd"]) + notional
        key = "long_liquidated_usd" if event.get("liquidated_position_side") == "long" else "short_liquidated_usd"
        row[key] = _d(row[key]) + notional
        row["event_count"] = int(row["event_count"]) + 1
    return {"schema": "binance_liquidations.v1", "generated_at": _now(), "window_seconds": window_seconds, "status": "ok", "source_id": "binance_force_order_websocket", "research_only": True, "live_order_allowed": False, "mainnet_signed_action": False, "coins": {coin: {key: str(value) if isinstance(value, Decimal) else value for key, value in row.items()} for coin, row in coins.items()}}


def save(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
    tmp.replace(path)


def run(*, output: Path, journal: Path, duration_seconds: int, window_seconds: int) -> int:
    events: deque[dict[str, Any]] = deque()
    started = time.monotonic()
    last_heartbeat = 0.0
    backoff = 1
    while duration_seconds <= 0 or time.monotonic() - started < duration_seconds:
        try:
            ws = websocket.create_connection(WS_URL, timeout=20)
            ws.settimeout(5)
            backoff = 1
            while duration_seconds <= 0 or time.monotonic() - started < duration_seconds:
                try:
                    raw = ws.recv()
                    payload = json.loads(raw)
                    rows = payload if isinstance(payload, list) else [payload]
                    for item in rows:
                        if not isinstance(item, dict):
                            continue
                        event = normalize_event(item)
                        if event:
                            events.append(event)
                            journal.parent.mkdir(parents=True, exist_ok=True)
                            journal.open("a", encoding="utf-8").write(json.dumps(event, sort_keys=True) + "\n")
                except websocket.WebSocketTimeoutException:
                    pass
                if time.monotonic() - last_heartbeat >= 30:
                    save(output, aggregate(events, window_seconds=window_seconds))
                    last_heartbeat = time.monotonic()
            ws.close()
        except Exception as exc:
            save(output, {"schema": "binance_liquidations.v1", "generated_at": _now(), "status": "degraded", "source_id": "binance_force_order_websocket", "error_type": type(exc).__name__, "research_only": True, "live_order_allowed": False, "mainnet_signed_action": False, "coins": {}})
            if duration_seconds > 0 and time.monotonic() - started >= duration_seconds:
                break
            time.sleep(min(backoff, 30))
            backoff = min(backoff * 2, 30)
    save(output, aggregate(events, window_seconds=window_seconds))
    return 0


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Read-only Binance public liquidation websocket collector.")
    parser.add_argument("--output", default=str(DEFAULT_OUTPUT))
    parser.add_argument("--journal", default=str(DEFAULT_JOURNAL))
    parser.add_argument("--pid-file", default=str(DEFAULT_PID))
    parser.add_argument("--duration-seconds", type=int, default=0)
    parser.add_argument("--window-seconds", type=int, default=900)
    args = parser.parse_args(argv)
    pid = Path(args.pid_file)
    pid.parent.mkdir(parents=True, exist_ok=True)
    pid.write_text(str(os.getpid()), encoding="utf-8")
    return run(output=Path(args.output), journal=Path(args.journal), duration_seconds=args.duration_seconds, window_seconds=args.window_seconds)


if __name__ == "__main__":
    raise SystemExit(main())