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