dekun
b733e551a0
Add scripts/normalize_ambiguous_unicode.py; fix corrupted patch_instance_theme_templates.py. Preserves curly quotes in string literals; removes Git homoglyph warnings on .env.example. Co-authored-by: Cursor <cursoragent@cursor.com>
118 lines
4.4 KiB
Python
118 lines
4.4 KiB
Python
"""中控资金统计:期权 USDC/USDT 按 1:1 计入 USDT 合计."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Optional
|
|
|
|
|
|
def _safe_float(value: Any) -> Optional[float]:
|
|
try:
|
|
if value is None or value == "":
|
|
return None
|
|
v = float(value)
|
|
return v if v >= 0 else None
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _account_total_usdt(funding: Any, trading: Any) -> Optional[float]:
|
|
fu = _safe_float(funding)
|
|
tu = _safe_float(trading)
|
|
if fu is None or tu is None:
|
|
return None
|
|
return round(fu + tu, 4)
|
|
|
|
|
|
def stablecoin_usdt_equiv(value: Any) -> Optional[float]:
|
|
"""USDC / USDT 按 1:1 折算为 USDT 统计口径."""
|
|
return _safe_float(value)
|
|
|
|
|
|
def _sum_optional(*values: Any) -> Optional[float]:
|
|
parts = [_safe_float(v) for v in values]
|
|
present = [p for p in parts if p is not None]
|
|
if not present:
|
|
return None
|
|
return round(sum(present), 4)
|
|
|
|
|
|
def options_balances_usdt_equiv(options_snap: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""从期权 snapshot 提取资金户/交易户 USDT 等价余额."""
|
|
snap = options_snap if isinstance(options_snap, dict) else {}
|
|
if snap.get("enabled") is False:
|
|
return {"ok": False, "funding_usdt": None, "trading_usdt": None}
|
|
if snap.get("ok") is False:
|
|
return {"ok": False, "funding_usdt": None, "trading_usdt": None}
|
|
bal = snap.get("balances") if isinstance(snap.get("balances"), dict) else snap
|
|
funding = _sum_optional(bal.get("funding_usdt"), bal.get("funding_usdc"))
|
|
trading = _sum_optional(bal.get("trading_usdt"), bal.get("trading_usdc"))
|
|
ok = funding is not None and trading is not None
|
|
return {"ok": ok, "funding_usdt": funding, "trading_usdt": trading}
|
|
|
|
|
|
def options_float_pnl_usdt(options_snap: dict[str, Any] | None) -> Optional[float]:
|
|
snap = options_snap if isinstance(options_snap, dict) else {}
|
|
if snap.get("enabled") is False or snap.get("ok") is False:
|
|
return None
|
|
upl = snap.get("upl_total_usdc")
|
|
if upl is None:
|
|
return None
|
|
try:
|
|
return round(float(upl), 4)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def options_open_position_count(options_snap: dict[str, Any] | None) -> int:
|
|
snap = options_snap if isinstance(options_snap, dict) else {}
|
|
if snap.get("enabled") is False or snap.get("ok") is False:
|
|
return 0
|
|
if snap.get("position_count") is not None:
|
|
try:
|
|
return max(0, int(snap.get("position_count")))
|
|
except (TypeError, ValueError):
|
|
pass
|
|
pos = snap.get("positions")
|
|
return len(pos) if isinstance(pos, list) else 0
|
|
|
|
|
|
def merge_perp_options_balances(
|
|
perpetual_funding_usdt: Any,
|
|
perpetual_trading_usdt: Any,
|
|
options_snap: dict[str, Any] | None,
|
|
) -> dict[str, Any]:
|
|
"""永续 + 期权余额合并为中控 USDT 统计口径."""
|
|
opt = options_balances_usdt_equiv(options_snap)
|
|
funding = _sum_optional(perpetual_funding_usdt, opt.get("funding_usdt"))
|
|
trading = _sum_optional(perpetual_trading_usdt, opt.get("trading_usdt"))
|
|
total = _account_total_usdt(funding, trading)
|
|
perp_total = _account_total_usdt(perpetual_funding_usdt, perpetual_trading_usdt)
|
|
opt_total = _account_total_usdt(opt.get("funding_usdt"), opt.get("trading_usdt"))
|
|
data_ok = total is not None
|
|
return {
|
|
"perpetual_funding_usdt": _safe_float(perpetual_funding_usdt),
|
|
"perpetual_trading_usdt": _safe_float(perpetual_trading_usdt),
|
|
"options_funding_usdt": opt.get("funding_usdt"),
|
|
"options_trading_usdt": opt.get("trading_usdt"),
|
|
"options_ok": bool(opt.get("ok")),
|
|
"funding_usdt": funding,
|
|
"trading_usdt": trading,
|
|
"total_usdt": total,
|
|
"perpetual_total_usdt": perp_total,
|
|
"options_total_usdt": opt_total,
|
|
"data_ok": data_ok,
|
|
}
|
|
|
|
|
|
def merge_board_row_balances(row: dict[str, Any]) -> dict[str, Any]:
|
|
"""监控板行 → 含期权的资金统计."""
|
|
caps = row.get("capabilities") or []
|
|
options_snap = row.get("options") if "options" in caps else None
|
|
merged = merge_perp_options_balances(
|
|
row.get("funding_usdt") if row.get("account_ok") else None,
|
|
row.get("trading_usdt") if row.get("account_ok") else None,
|
|
options_snap,
|
|
)
|
|
merged["options_float_pnl_u"] = options_float_pnl_usdt(options_snap)
|
|
merged["options_open_position_count"] = options_open_position_count(options_snap)
|
|
return merged
|