dekun
103 lines
3.1 KiB
Python
103 lines
3.1 KiB
Python
"""从成交明细汇总腿盈亏与净盈亏。"""
|
||
|
||
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,
|
||
}
|