"""永续估算盈亏:固定 taker 手续费(默认单边 0.05%,开+平双边). 浮盈亏仍读交易所;本模块只服务「盈利金额 / 止盈盈利 / 推送 / 记账 pnl_amount」等估算口径. """ from __future__ import annotations import math import os from typing import Optional def _finite(v) -> Optional[float]: try: f = float(v) return f if math.isfinite(f) else None except (TypeError, ValueError): return None def taker_fee_rate() -> float: """单边 taker 费率,默认 0.0005(=0.05%).""" raw = os.getenv("PERP_TAKER_FEE_RATE", "0.0005") rate = _finite(raw) if rate is None or rate < 0: return 0.0005 return rate def notional_usdt(price, qty, contract_size: float = 1.0) -> Optional[float]: """名义价值 U = 价格 × 张数 × 合约面值.""" p = _finite(price) q = _finite(qty) cs = _finite(contract_size) if p is None or q is None or p <= 0 or q <= 0: return None if cs is None or cs <= 0: cs = 1.0 return abs(q) * p * cs def estimate_roundtrip_fee_usdt( entry_price, exit_price, qty=None, contract_size: float = 1.0, *, open_notional: float | None = None, rate: float | None = None, ) -> float: """开+平双边手续费(各单边 rate). 优先用 价×张×面值;若无张数则用 open_notional 估开仓名义, 平仓名义按 exit/entry 缩放. """ fee_rate = taker_fee_rate() if rate is None else float(rate) if fee_rate <= 0: return 0.0 entry = _finite(entry_price) exit_p = _finite(exit_price) open_n = notional_usdt(entry, qty, contract_size) if qty is not None else None if open_n is None: open_n = _finite(open_notional) if open_n is None or open_n <= 0: return 0.0 if entry is not None and entry > 0 and exit_p is not None and exit_p > 0: close_n = open_n * (exit_p / entry) else: close_n = open_n return round(open_n * fee_rate + close_n * fee_rate, 8) def net_pnl_after_fee( gross_pnl, entry_price, exit_price, qty=None, contract_size: float = 1.0, *, open_notional: float | None = None, rate: float | None = None, ) -> Optional[float]: """毛利扣双边手续费后的净盈亏;gross 无效则返回 None.""" gross = _finite(gross_pnl) if gross is None: return None fee = estimate_roundtrip_fee_usdt( entry_price, exit_price, qty, contract_size, open_notional=open_notional, rate=rate, ) return round(gross - fee, 4)