siloqy/prod/bingx/journal.py

from __future__ import annotations

import json
import logging
import urllib.parse
import urllib.request
from dataclasses import dataclass
from datetime import datetime, timezone
from hashlib import sha256
from typing import Any

from prod.ch_writer import ch_put
from prod.ch_writer import ch_put_green
from prod.ch_writer import ch_put_prodgreen
from prod.ch_writer import ch_put_violet
from prod.ch_writer import ch_put_pink

# ─── Account event rate control (§10.2) ──────────────────────────────────────
import os
import time as _time

_ACCOUNT_EVENT_RATE_CAP = int(os.environ.get("PINK_ACCOUNT_EVENT_RATE_CAP", "5"))


class _AccountEventRateLimiter:
    """Token-bucket rate limiter for account events (PINK data volume control)."""
    def __init__(self, max_per_sec: int = 5):
        self._max = max(max_per_sec, 1)
        self._tokens = float(self._max)
        self._last = _time.monotonic()

    def allow(self) -> bool:
        now = _time.monotonic()
        self._tokens = min(self._max, self._tokens + (now - self._last) * self._max)
        self._last = now
        if self._tokens >= 1.0:
            self._tokens -= 1.0
            return True
        return False
from prod.ch_writer import ts_us
from prod.bingx.leverage import LEVERAGE_MAPPING_RULE


CH_URL = "http://localhost:8123"
CH_USER = "dolphin"
CH_PASS = "dolphin_ch_2026"
CH_DB = "dolphin"
JOURNAL_EVENT_TYPE = "BINGX_SNAPSHOT"
LOGGER = logging.getLogger(__name__)


def _json_safe(value: Any) -> Any:
    if isinstance(value, dict):
        return {str(key): _json_safe(val) for key, val in value.items()}
    if isinstance(value, list):
        return [_json_safe(item) for item in value]
    if isinstance(value, tuple):
        return [_json_safe(item) for item in value]
    if hasattr(value, "isoformat"):
        try:
            return value.isoformat()
        except Exception:
            pass
    if hasattr(value, "as_decimal"):
        try:
            return str(value.as_decimal())
        except Exception:
            pass
    if hasattr(value, "__dict__"):
        return _json_safe(dict(vars(value)))
    return value


def _capital_from_balances(balances: Any) -> float:
    if not isinstance(balances, list):
        LOGGER.warning("BingX journal account snapshot balances payload is not a list")
        return 0.0
    found = 0.0
    for row in balances:
        if not isinstance(row, dict):
            LOGGER.warning("BingX journal account snapshot skipped malformed balance row: %r", row)
            continue
        capital = 0.0
        for key in ("total", "balance", "equity", "availableMargin", "availableBalance", "walletBalance", "free"):
            try:
                capital = float(row.get(key, 0.0) or 0.0)
            except Exception:
                continue
            if capital > 0 and capital == capital:
                found = capital
                return capital
        if capital > 0 and capital == capital:
            found = capital
            return capital
    if balances:
        LOGGER.error("BingX journal account snapshot contained no usable balance rows")
    return found


def _open_notional_from_positions(positions: Any) -> float:
    if not isinstance(positions, dict):
        LOGGER.warning("BingX journal positions payload is not a dict")
        return 0.0
    total = 0.0
    for row in positions.values():
        if not isinstance(row, dict):
            LOGGER.warning("BingX journal skipped malformed position row: %r", row)
            continue
        try:
            qty = abs(
                float(
                    row.get("positionAmt")
                    or row.get("positionQty")
                    or row.get("positionSize")
                    or row.get("quantity")
                    or row.get("pa")
                    or 0.0
                )
            )
            if qty <= 0.0:
                continue
            notional = row.get("positionValue") or row.get("notional") or row.get("openNotional")
            if notional is not None:
                total += abs(float(notional or 0.0))
                continue
            entry = (
                row.get("entryPrice")
                or row.get("avgPrice")
                or row.get("markPrice")
                or row.get("avgEntryPrice")
                or row.get("ep")
                or row.get("ap")
                or 0.0
            )
            total += qty * abs(float(entry or 0.0))
        except Exception:
            LOGGER.warning("BingX journal skipped unreadable position row: %r", row)
            continue
    return total


def _filled_order_count_from_fills(fills: Any) -> int:
    if not isinstance(fills, list):
        return 0
    seen: set[str] = set()
    count = 0
    for snapshot in fills:
        if not isinstance(snapshot, dict):
            continue
        row = snapshot.get("row") if isinstance(snapshot.get("row"), dict) else snapshot
        if not isinstance(row, dict):
            continue
        status = str(row.get("status") or "").upper()
        if status and status not in {"FILLED", "CLOSED"}:
            continue
        trade_key = str(snapshot.get("_trade_key") or "").strip()
        if trade_key:
            base_key = trade_key.split(":", 1)[0]
        else:
            base_key = str(
                row.get("orderId")
                or row.get("orderID")
                or row.get("clientOrderId")
                or row.get("clientOrderID")
                or ""
            ).strip()
        if not base_key or base_key in seen:
            continue
        seen.add(base_key)
        count += 1
    return count


_STRATEGY_DB_MAP: dict[str, str] = {
    "blue": "dolphin",
    "green": "dolphin_green",
    "prodgreen": "dolphin_prodgreen",
    "pink": "dolphin_pink",
    "violet": "dolphin_violet",
}

_STRATEGY_SINK_MAP: dict[str, Any] = {
    "blue": ch_put,
    "green": ch_put_green,
    "prodgreen": ch_put_prodgreen,
    "pink": ch_put_pink,
    "violet": ch_put_violet,
}

_STRATEGY_SINK_NAME_MAP: dict[str, str] = {
    "blue": "ch_put",
    "green": "ch_put_green",
    "prodgreen": "ch_put_prodgreen",
    "pink": "ch_put_pink",
    "violet": "ch_put_violet",
}


def _db_for_strategy(strategy: str) -> str:
    name = str(strategy or "").lower()
    return _STRATEGY_DB_MAP.get(name, "dolphin_prodgreen" if name.startswith("prod") else CH_DB)


def _sink_for_strategy(strategy: str):
    strategy_lower = str(strategy or "").lower()
    sink = _STRATEGY_SINK_MAP.get(strategy_lower)
    if callable(sink):
        return sink
    sink_name = _STRATEGY_SINK_NAME_MAP.get(strategy_lower)
    if sink_name:
        sink = globals().get(sink_name)
        if callable(sink):
            return sink
    return ch_put_prodgreen if strategy_lower.startswith("prod") else ch_put


@dataclass(frozen=True)
class BingxJournalSnapshot:
    ts: int
    strategy: str
    account_id: str
    ledger_authority: str
    payload: dict[str, Any]
    fingerprint: str
    reason: str = ""


def build_snapshot(
    *,
    strategy: str,
    account_id: str,
    ledger_authority: str,
    payload: dict[str, Any],
    reason: str = "",
) -> BingxJournalSnapshot:
    payload_json = json.dumps(_json_safe(payload), sort_keys=True, separators=(",", ":"))
    fingerprint = sha256(payload_json.encode("utf-8")).hexdigest()
    return BingxJournalSnapshot(
        ts=ts_us(),
        strategy=strategy,
        account_id=account_id,
        ledger_authority=ledger_authority,
        payload=payload,
        fingerprint=fingerprint,
        reason=reason,
    )


def write_snapshot(snapshot: BingxJournalSnapshot) -> None:
    account = snapshot.payload.get("account", {})
    balances = account.get("balances", [])
    capital = _capital_from_balances(balances)
    peak_capital = capital
    drawdown_pct = 0.0
    if capital <= 0.0:
        LOGGER.error(
            "BingX journal snapshot has no usable capital for strategy=%s account_id=%s reason=%s",
            snapshot.strategy,
            snapshot.account_id,
            snapshot.reason or JOURNAL_EVENT_TYPE,
        )
    positions = snapshot.payload.get("positions", {})
    open_positions = len(positions) if isinstance(positions, dict) else 0
    current_open_notional = _open_notional_from_positions(positions)
    current_account_leverage = current_open_notional / capital if capital > 0 else 0.0
    configured = snapshot.payload.get("configured_leverage", {})
    exchange_leverage = 0
    if isinstance(configured, dict) and configured:
        try:
            exchange_leverage = max(int(v) for v in configured.values() if int(v) > 0)
        except Exception:
            exchange_leverage = 0
    fills = snapshot.payload.get("fills", [])
    fills_today = len(fills) if isinstance(fills, list) else 0
    trades_today = _filled_order_count_from_fills(fills)
    sink = _sink_for_strategy(snapshot.strategy)
    sink(
        "account_events",
        {
            "ts": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f"),
            "event_type": snapshot.reason or JOURNAL_EVENT_TYPE,
            "strategy": snapshot.strategy,
            "posture": "N/A",
            "capital": capital,
            "peak_capital": peak_capital,
            "drawdown_pct": drawdown_pct,
            "pnl_today": 0.0,
            "trades_today": trades_today,
            "open_positions": open_positions,
            "boost": 1.0,
            "beta": 1.0,
            "current_open_notional": current_open_notional,
            "current_account_leverage": current_account_leverage,
            "exchange_leverage": exchange_leverage,
            "exchange_leverage_mode": "mapped_conservative_integer",
            "leverage_mapping_rule": LEVERAGE_MAPPING_RULE,
            "notes": json.dumps(
                {
                    "account_id": snapshot.account_id,
                    "ledger_authority": snapshot.ledger_authority,
                    "fingerprint": snapshot.fingerprint,
                    "fills_today": fills_today,
                    "filled_orders_today": trades_today,
                    "payload": _json_safe(snapshot.payload),
                },
                sort_keys=True,
                separators=(",", ":"),
            ),
        },
    )


def load_latest_snapshot(strategy: str, account_id: str | None = None) -> dict[str, Any] | None:
    record = load_latest_record(strategy, account_id=account_id)
    if record is None:
        return None
    return record.get("payload")


def load_latest_record(strategy: str, account_id: str | None = None) -> dict[str, Any] | None:
    clauses = [f"strategy = {json.dumps(strategy)}"]
    if account_id:
        clauses.append(f"JSONExtractString(notes, 'account_id') = {json.dumps(account_id)}")
    where = " AND ".join(clauses)
    sql = (
        "SELECT ts, event_type, strategy, notes "
        f"FROM account_events WHERE {where} ORDER BY ts DESC LIMIT 1 FORMAT JSONEachRow"
    )
    url = f"{CH_URL}/?database={_db_for_strategy(strategy)}&query={urllib.parse.quote(sql)}"
    req = urllib.request.Request(url)
    req.add_header("X-ClickHouse-User", CH_USER)
    req.add_header("X-ClickHouse-Key", CH_PASS)
    try:
        with urllib.request.urlopen(req, timeout=5) as resp:
            body = resp.read().decode("utf-8").strip()
        if not body:
            return None
        row = json.loads(body)
        notes = row.get("notes")
        if not notes:
            return None
        parsed = json.loads(notes)
        return {
            "ts": row.get("ts"),
            "event_type": row.get("event_type"),
            "strategy": row.get("strategy"),
            "notes": parsed,
            "payload": parsed.get("payload"),
        }
    except Exception:
        return None


def resolve_account_event_rate_cap() -> int:
    """Return the configured account event rate cap (rows/sec) per §10.2."""
    raw = os.environ.get("PINK_ACCOUNT_EVENT_RATE_CAP", "")
    try:
        val = int(raw)
        return max(val, 1)
    except (TypeError, ValueError):
        return _ACCOUNT_EVENT_RATE_CAP