199 lines
8.1 KiB
Python
199 lines
8.1 KiB
Python
"""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()
|
|
directional_bias = ""
|
|
if isinstance((global_risk or {}).get("directional_market_bias"), dict):
|
|
directional_bias = str((global_risk or {}).get("directional_market_bias", {}).get("market_bias") 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 directional_bias != "favorable" 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"]
|