PINK Phase 0 and 1: VST WS confirmed plus AccountSnapshotV2 account core

prod/bingx/sandbox_status.py

@@ -1,126 +0,0 @@
-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