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>
128 lines
3.9 KiB
Python
128 lines
3.9 KiB
Python
"""实例页:持仓未实现盈亏(实时盈亏)汇总."""
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
|
|
from lib.hub.hub_position_metrics import parse_position_unrealized_pnl
|
|
|
|
|
|
def position_row_contracts(pos: dict[str, Any]) -> float:
|
|
"""持仓张数:与三所 app 内 _position_row_effective_contracts 规则一致."""
|
|
if not isinstance(pos, dict):
|
|
return 0.0
|
|
info = pos.get("info") or {}
|
|
if not isinstance(info, dict):
|
|
info = {}
|
|
for val in (
|
|
pos.get("contracts"),
|
|
info.get("positionAmt"),
|
|
info.get("size"),
|
|
info.get("pos"),
|
|
info.get("availPos"),
|
|
):
|
|
if val is None or val == "":
|
|
continue
|
|
try:
|
|
x = abs(float(val))
|
|
if x > 0:
|
|
return x
|
|
except (TypeError, ValueError):
|
|
continue
|
|
return 0.0
|
|
|
|
|
|
def sum_unrealized_pnl_from_positions(positions: list[dict[str, Any]] | None) -> float | None:
|
|
total = 0.0
|
|
found = False
|
|
for p in positions or []:
|
|
if not isinstance(p, dict):
|
|
continue
|
|
if position_row_contracts(p) <= 1e-12:
|
|
continue
|
|
upnl = parse_position_unrealized_pnl(p)
|
|
if upnl is None:
|
|
continue
|
|
found = True
|
|
total += float(upnl)
|
|
return round(total, 2) if found else None
|
|
|
|
|
|
def _row_field(row: Any, key: str, default: str = "") -> str:
|
|
if row is None:
|
|
return default
|
|
try:
|
|
if hasattr(row, "keys") and key in row.keys():
|
|
val = row[key]
|
|
elif isinstance(row, dict):
|
|
val = row.get(key)
|
|
else:
|
|
val = None
|
|
except Exception:
|
|
val = None
|
|
return str(val or default).strip()
|
|
|
|
|
|
def sum_unrealized_pnl_from_metrics(
|
|
rows: list[dict[str, Any]] | list[Any],
|
|
get_metrics_fn: Callable[[str, str], dict[str, Any] | None],
|
|
) -> float | None:
|
|
"""按活跃监控单逐笔拉交易所 metrics 汇总(与持仓卡浮盈亏一致)."""
|
|
total = 0.0
|
|
found = False
|
|
for row in rows or []:
|
|
ex_sym = _row_field(row, "exchange_symbol")
|
|
sym = _row_field(row, "symbol")
|
|
direction = _row_field(row, "direction", "long").lower() or "long"
|
|
target = ex_sym or sym
|
|
if not target:
|
|
continue
|
|
metrics = get_metrics_fn(target, direction)
|
|
if not isinstance(metrics, dict):
|
|
continue
|
|
upnl = metrics.get("unrealized_pnl")
|
|
if upnl is None:
|
|
continue
|
|
try:
|
|
total += float(upnl)
|
|
found = True
|
|
except (TypeError, ValueError):
|
|
continue
|
|
return round(total, 2) if found else None
|
|
|
|
|
|
def fetch_unrealized_pnl(fetch_positions_fn: Callable[[], list[dict[str, Any]] | None]) -> float | None:
|
|
try:
|
|
return sum_unrealized_pnl_from_positions(fetch_positions_fn() or [])
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def resolve_instance_unrealized_pnl(
|
|
fetch_positions_fn: Callable[[], list[dict[str, Any]] | None],
|
|
active_rows: list[Any] | None,
|
|
get_metrics_fn: Callable[[str, str], dict[str, Any] | None] | None,
|
|
) -> float | None:
|
|
"""先全量持仓汇总,失败或无数据时回退到活跃监控单 metrics."""
|
|
total = fetch_unrealized_pnl(fetch_positions_fn)
|
|
if total is not None:
|
|
return total
|
|
if active_rows and get_metrics_fn:
|
|
return sum_unrealized_pnl_from_metrics(active_rows, get_metrics_fn)
|
|
return None
|
|
|
|
|
|
def merge_unrealized_pnl_components(*parts: float | None) -> float | None:
|
|
"""合并永续与期权等多路未实现盈亏(任一路有值即参与合计)."""
|
|
total = 0.0
|
|
found = False
|
|
for part in parts:
|
|
if part is None:
|
|
continue
|
|
try:
|
|
total += float(part)
|
|
found = True
|
|
except (TypeError, ValueError):
|
|
continue
|
|
return round(total, 2) if found else None
|