"""交易组持仓周期:目标出场以策略平仓时刻为准;永续先平则以永续平仓为准。""" from __future__ import annotations from typing import Any, Mapping, Sequence # 目标平仓(含 15U / 权利金倍数 / 只平永续) _TARGET_REASONS = frozenset({"fixed_usdt", "premium_multiple", "target_perp_only"}) def _ts(v: Any) -> int | None: if v is None: return None try: n = int(v) except (TypeError, ValueError): return None return n if n > 0 else None def _as_map(row: Mapping[str, Any] | Any) -> Mapping[str, Any]: if isinstance(row, Mapping): return row try: return dict(row) except Exception: return {} def first_perp_close_ts_ms(fills: Sequence[Mapping[str, Any] | Any]) -> int | None: """永续平仓成交时间(目标只平永续时作为持仓结束时刻)。""" best: int | None = None for raw in fills: f = _as_map(raw) if str(f.get("leg") or "") != "perp": continue if str(f.get("action") or "") != "close": continue ts = _ts(f.get("ts_ms")) if ts is None: continue if best is None or ts < best: best = ts return best def hold_timing( group: Mapping[str, Any] | Any, fills: Sequence[Mapping[str, Any] | Any] ) -> dict[str, Any]: """ 返回展示用开仓/平仓/持仓时长。 - 开仓:groups.open_at_ms - 平仓(策略持仓周期): - `target_perp_only` / `option_residual`:永续平仓 fill 时间 - 其它已平:groups.close_at_ms(缺则回退成交) """ g = _as_map(group) open_ms = _ts(g.get("open_at_ms")) status = str(g.get("status") or "") reason = str(g.get("close_reason") or "") group_close = _ts(g.get("close_at_ms")) perp_close = first_perp_close_ts_ms(fills) use_perp = reason == "target_perp_only" or status == "option_residual" if use_perp: close_ms = perp_close or group_close basis = "perp" elif status == "open": close_ms = None basis = "open" else: close_ms = group_close if close_ms is None and reason in _TARGET_REASONS: close_ms = perp_close basis = "group" hold_ms: int | None = None if open_ms is not None and close_ms is not None and close_ms >= open_ms: hold_ms = close_ms - open_ms return { "hold_open_at_ms": open_ms, "hold_close_at_ms": close_ms, "hold_ms": hold_ms, "hold_basis": basis, }