siloqy/prod/bingx/sandbox_status.py

from __future__ import annotations

import json
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

DEFAULT_SANDBOX_STATUS_PATH = Path("/tmp/bingx_sandbox_status.json")


@dataclass(frozen=True)
class BingxSandboxStatus:
    """Small sidecar snapshot for BingX demo/testnet state.

    The snapshot is intentionally local-only so it can be used by tests and
    operators without writing into BLUE state, ClickHouse, or production logs.
    """

    ts: str
    environment: str
    balance: float
    equity: float
    available_margin: float
    unrealized_profit: float
    used_margin: float
    open_positions: int
    open_orders: int
    account_currency: str = "VST"
    clean: bool = False
    notes: dict[str, Any] | None = None

    def to_dict(self) -> dict[str, Any]:
        return {
            "ts": self.ts,
            "environment": self.environment,
            "account_currency": self.account_currency,
            "balance": self.balance,
            "equity": self.equity,
            "available_margin": self.available_margin,
            "unrealized_profit": self.unrealized_profit,
            "used_margin": self.used_margin,
            "open_positions": self.open_positions,
            "open_orders": self.open_orders,
            "clean": self.clean,
            "notes": self.notes or {},
        }


def _safe_float(value: Any, default: float = 0.0) -> float:
    try:
        out = float(value)
    except Exception:
        return default
    return out if out == out else default


def _count_positions(positions: Any) -> int:
    if isinstance(positions, list):
        return sum(1 for item in positions if isinstance(item, dict))
    return 0


def _count_orders(open_orders: Any) -> int:
    if isinstance(open_orders, dict):
        orders = open_orders.get("orders")
        if isinstance(orders, list):
            return sum(1 for item in orders if isinstance(item, dict))
    if isinstance(open_orders, list):
        return sum(1 for item in open_orders if isinstance(item, dict))
    return 0


def build_sandbox_status(
    *,
    balance_payload: dict[str, Any],
    positions_payload: Any,
    open_orders_payload: Any,
    environment: str = "VST",
    account_currency: str = "VST",
    notes: dict[str, Any] | None = None,
) -> BingxSandboxStatus:
    balance_row = balance_payload.get("balance", balance_payload) if isinstance(balance_payload, dict) else {}
    if not isinstance(balance_row, dict):
        balance_row = {}
    balance = _safe_float(balance_row.get("balance"), 0.0)
    equity = _safe_float(balance_row.get("equity"), balance)
    available_margin = _safe_float(balance_row.get("availableMargin"), 0.0)
    unrealized_profit = _safe_float(balance_row.get("unrealizedProfit"), 0.0)
    used_margin = _safe_float(balance_row.get("usedMargin"), 0.0)
    open_positions = _count_positions(positions_payload)
    open_orders = _count_orders(open_orders_payload)
    return BingxSandboxStatus(
        ts=datetime.now(timezone.utc).isoformat(),
        environment=str(environment),
        account_currency=str(account_currency),
        balance=balance,
        equity=equity,
        available_margin=available_margin,
        unrealized_profit=unrealized_profit,
        used_margin=used_margin,
        open_positions=open_positions,
        open_orders=open_orders,
        clean=(open_positions == 0 and open_orders == 0),
        notes=notes or {},
    )


def snapshot_path(path: str | Path | None = None) -> Path:
    return Path(path) if path is not None else DEFAULT_SANDBOX_STATUS_PATH


def write_sandbox_status(status: BingxSandboxStatus, path: str | Path | None = None) -> Path:
    target = snapshot_path(path)
    target.write_text(json.dumps(status.to_dict(), indent=2, sort_keys=True))
    return target


def load_sandbox_status(path: str | Path | None = None) -> dict[str, Any] | None:
    target = snapshot_path(path)
    if not target.exists():
        return None
    try:
        return json.loads(target.read_text())
    except Exception:
        return None