"""从成交明细汇总腿盈亏与净盈亏。""" from __future__ import annotations from typing import Any def _as_map(x: Any) -> dict[str, Any]: if isinstance(x, dict): return x try: return dict(x) except Exception: return {} def _leg_option_pnl(rows: list[dict[str, Any]], leg: str) -> float | None: opt_open = next( (f for f in rows if f.get("leg") == leg and f.get("action") == "open"), None, ) opt_close = next( (f for f in rows if f.get("leg") == leg and f.get("action") == "close"), None, ) if not opt_open or not opt_close: return None qty = float(opt_open.get("qty_eth") or opt_close.get("qty_eth") or 0) return (float(opt_close["fill_px"]) - float(opt_open["fill_px"])) * qty def summarize_fills_pnl(fills: list[Any]) -> dict[str, float | None]: """ 价差盈亏按 fill_px;手续费另扣。 净盈亏 = 各腿盈亏之和 − 全部手续费(开+平)。 支持永期(option+perp)与期期(option+option2)。 手续费拆:fees_perp / fees_option(含 option2)/ fees_option2; 滑点合计 slip_total(SIM 记账;LIVE 应为 0)。 """ rows = [_as_map(x) for x in fills] option_pnl = _leg_option_pnl(rows, "option") option2_pnl = _leg_option_pnl(rows, "option2") perp_open = next( (f for f in rows if f.get("leg") == "perp" and f.get("action") == "open"), None, ) perp_close = next( (f for f in rows if f.get("leg") == "perp" and f.get("action") == "close"), None, ) perp_pnl: float | None = None if perp_open and perp_close: qty = float(perp_open.get("qty_eth") or perp_close.get("qty_eth") or 0) side = str(perp_open.get("side") or "") o = float(perp_open["fill_px"]) c = float(perp_close["fill_px"]) if side == "long": perp_pnl = (c - o) * qty else: perp_pnl = (o - c) * qty fees_perp = sum( float(f.get("fee") or 0) for f in rows if str(f.get("leg") or "") == "perp" ) fees_option = sum( float(f.get("fee") or 0) for f in rows if str(f.get("leg") or "") in ("option", "option2") ) fees_option2 = sum( float(f.get("fee") or 0) for f in rows if str(f.get("leg") or "") == "option2" ) fees_total = fees_perp + fees_option slip_total = sum(float(f.get("slip") or 0) for f in rows) parts: list[float] = [] if option_pnl is not None: parts.append(option_pnl) if option2_pnl is not None: parts.append(option2_pnl) if perp_pnl is not None: parts.append(perp_pnl) gross = sum(parts) if parts else None net = (gross - fees_total) if gross is not None else None return { "option_pnl": option_pnl, "option2_pnl": option2_pnl, "perp_pnl": perp_pnl, "fees_perp": fees_perp, "fees_option": fees_option, "fees_option2": fees_option2, "fees_total": fees_total, "slip_total": slip_total, "gross_pnl": gross, "net_pnl": net, }