"""Position health guard for paper/live trade lifecycle management. The entry signal answers "can we enter?". This module answers the later question: "is this open position still behaving like the original setup?". Keep the decision logic pure so paper trading, live sync, and future review jobs can reuse the same vocabulary. """ from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime def _safe_float(value, default: float = 0.0) -> float: try: if value is None or value == "": return default return float(value) except Exception: return default def _parse_time(value: str) -> datetime | None: if not value: return None try: return datetime.fromisoformat(str(value).replace("Z", "+00:00")) except Exception: return None def _pnl_pct(side: str, entry_price: float, price: float) -> float: if entry_price <= 0 or price <= 0: return 0.0 if str(side or "long").lower() == "short": return (entry_price / price - 1) * 100 return (price / entry_price - 1) * 100 def _age_hours(opened_at: str, event_time: str) -> float: start = _parse_time(opened_at) end = _parse_time(event_time) or datetime.now() if not start: return 0.0 return max(0.0, (end - start).total_seconds() / 3600) def _cfg_bool(config: dict, key: str, default: bool) -> bool: value = config.get(key, default) if isinstance(value, str): return value.strip().lower() in {"1", "true", "yes", "on"} return bool(value) @dataclass(frozen=True) class PositionHealthDecision: action: str = "hold" reason: str = "" health_score: float = 100.0 guard_stop: float = 0.0 detail: dict = field(default_factory=dict) def as_dict(self) -> dict: return { "action": self.action, "reason": self.reason, "health_score": round(self.health_score, 4), "guard_stop": round(self.guard_stop, 12) if self.guard_stop > 0 else 0, "detail": self.detail, } def position_health_metrics(trade: dict, current_price: float, event_time: str) -> dict: side = str(trade.get("side") or "long").strip().lower() or "long" entry = _safe_float(trade.get("entry_price")) current = _safe_float(current_price) max_price = max(_safe_float(trade.get("max_price")) or entry, current, entry) min_price = min(_safe_float(trade.get("min_price")) or entry, current, entry) current_pnl = _pnl_pct(side, entry, current) if side == "short": max_pnl = _pnl_pct(side, entry, min_price) min_pnl = _pnl_pct(side, entry, max_price) else: max_pnl = _pnl_pct(side, entry, max_price) min_pnl = _pnl_pct(side, entry, min_price) profit_giveback_pct = 0.0 if max_pnl > 0: profit_giveback_pct = max(0.0, (max_pnl - current_pnl) / max_pnl * 100) return { "side": side, "entry_price": entry, "current_price": current, "age_hours": round(_age_hours(str(trade.get("opened_at") or ""), event_time), 6), "pnl_pct": round(current_pnl, 6), "max_pnl_pct": round(max_pnl, 6), "min_pnl_pct": round(min_pnl, 6), "profit_giveback_pct": round(profit_giveback_pct, 6), "trailing_active": _safe_float(trade.get("trailing_stop")) > 0, } def evaluate_position_health( *, trade: dict, current_price: float, event_time: str, config: dict | None = None, global_risk: dict | None = None, ) -> PositionHealthDecision: """Return a deterministic action for an open position. Actions: - hold: keep normal management. - tighten_stop: raise the protective stop near breakeven. - close: exit early because the original setup is no longer healthy. """ cfg = config if isinstance(config, dict) else {} if not _cfg_bool(cfg, "position_guard_enabled", True): return PositionHealthDecision(action="hold", reason="disabled", detail={"enabled": False}) metrics = position_health_metrics(trade, current_price, event_time) detail = {"enabled": True, **metrics} if metrics["entry_price"] <= 0 or metrics["current_price"] <= 0: return PositionHealthDecision(action="hold", reason="missing_price", detail=detail) if metrics["trailing_active"]: return PositionHealthDecision(action="hold", reason="already_protected_by_trailing", health_score=90, detail=detail) age = _safe_float(metrics["age_hours"]) pnl = _safe_float(metrics["pnl_pct"]) max_pnl = _safe_float(metrics["max_pnl_pct"]) giveback = _safe_float(metrics["profit_giveback_pct"]) health_score = 100.0 health_score -= min(35.0, max(0.0, age - 2.0) * 2.0) if max_pnl <= 0: health_score -= 15.0 if giveback > 0: health_score -= min(30.0, giveback * 0.25) if pnl < 0: health_score -= min(20.0, abs(pnl) * 3.0) risk_level = str((global_risk or {}).get("risk_level") or "").strip().lower() critical_exit_enabled = _cfg_bool(cfg, "position_guard_critical_exit_enabled", True) critical_min_age = max(0.0, _safe_float(cfg.get("position_guard_critical_min_age_hours"), 0.5)) critical_max_pnl = _safe_float(cfg.get("position_guard_critical_max_pnl_pct"), 1.0) if critical_exit_enabled and risk_level == "critical" and age >= critical_min_age and pnl <= critical_max_pnl: detail["global_risk"] = global_risk or {} return PositionHealthDecision( action="close", reason="market_risk_unprotected", health_score=max(0.0, health_score - 25.0), detail=detail, ) giveback_enabled = _cfg_bool(cfg, "position_guard_profit_giveback_enabled", True) giveback_start = max(0.0, _safe_float(cfg.get("position_guard_giveback_min_max_pnl_pct"), 2.0)) giveback_exit = max(0.0, _safe_float(cfg.get("position_guard_giveback_exit_pct"), 70.0)) giveback_exit_pnl = _safe_float(cfg.get("position_guard_giveback_exit_current_pnl_pct"), 0.6) if giveback_enabled and max_pnl >= giveback_start and giveback >= giveback_exit and pnl <= giveback_exit_pnl: return PositionHealthDecision( action="close", reason="profit_giveback_before_trailing", health_score=max(0.0, health_score - 20.0), detail=detail, ) hard_hours = max(0.0, _safe_float(cfg.get("position_guard_hard_hours"), 18.0)) hard_min_max_pnl = max(0.0, _safe_float(cfg.get("position_guard_hard_min_max_pnl_pct"), 2.5)) if hard_hours > 0 and age >= hard_hours and max_pnl < hard_min_max_pnl: return PositionHealthDecision( action="close", reason="position_timeout_weak", health_score=max(0.0, health_score - 30.0), detail={**detail, "hard_hours": hard_hours, "hard_min_max_pnl_pct": hard_min_max_pnl}, ) soft_hours = max(0.0, _safe_float(cfg.get("position_guard_soft_hours"), 6.0)) soft_min_max_pnl = max(0.0, _safe_float(cfg.get("position_guard_soft_min_max_pnl_pct"), 1.5)) lock_profit = _safe_float(cfg.get("position_guard_tighten_lock_profit_pct"), 0.15) if soft_hours > 0 and age >= soft_hours and max_pnl < soft_min_max_pnl: guard_stop = metrics["entry_price"] * (1 + lock_profit / 100) if metrics["side"] == "short": guard_stop = metrics["entry_price"] * (1 - lock_profit / 100) return PositionHealthDecision( action="tighten_stop", reason="position_timeout_soft", health_score=max(0.0, health_score - 15.0), guard_stop=guard_stop, detail={**detail, "soft_hours": soft_hours, "soft_min_max_pnl_pct": soft_min_max_pnl}, ) return PositionHealthDecision(action="hold", reason="healthy", health_score=max(0.0, health_score), detail=detail) __all__ = ["PositionHealthDecision", "evaluate_position_health", "position_health_metrics"]