07.4.4.1.13 — AutoTrade Runtime Journal, Execution Refactor & Trade Analytics
This commit is contained in:
318
app/src/trading/auto/position_health.py
Normal file
318
app/src/trading/auto/position_health.py
Normal file
@@ -0,0 +1,318 @@
|
||||
# app/src/trading/auto/position_health.py
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
from src.core.numbers import safe_float
|
||||
from src.core.types import NumericLike
|
||||
from src.trading.auto.state import AutoTradeState
|
||||
|
||||
|
||||
class AutoPositionHealthMixin:
|
||||
# синхронизировать runtime health/risk состояние открытой позиции
|
||||
def _sync_position_health_state(self, state: AutoTradeState) -> None:
|
||||
if state.position_side == "NONE" or state.entry_price is None:
|
||||
state.position_pnl_percent = None
|
||||
state.position_hold_seconds = None
|
||||
state.position_pressure = None
|
||||
state.position_health_score = None
|
||||
state.position_health_status = None
|
||||
state.position_health_reason = None
|
||||
state.position_risk_level = None
|
||||
state.position_risk_reason = None
|
||||
state.position_trend_alignment = None
|
||||
state.position_adverse_momentum = False
|
||||
state.position_exit_pressure = None
|
||||
return
|
||||
|
||||
pnl_percent = self._position_pnl_percent(state)
|
||||
hold_seconds = self._position_hold_seconds(state)
|
||||
trend_alignment = self._position_trend_alignment(state)
|
||||
adverse_momentum = self._has_adverse_position_momentum(state)
|
||||
|
||||
pressure = self._position_pressure(
|
||||
state=state,
|
||||
pnl_percent=pnl_percent,
|
||||
)
|
||||
|
||||
health_score = self._position_health_score(
|
||||
state=state,
|
||||
pnl_percent=pnl_percent,
|
||||
trend_alignment=trend_alignment,
|
||||
adverse_momentum=adverse_momentum,
|
||||
)
|
||||
|
||||
risk_level, risk_reason = self._position_risk_level(
|
||||
state=state,
|
||||
pnl_percent=pnl_percent,
|
||||
trend_alignment=trend_alignment,
|
||||
adverse_momentum=adverse_momentum,
|
||||
)
|
||||
|
||||
state.position_pnl_percent = pnl_percent
|
||||
state.position_hold_seconds = hold_seconds
|
||||
state.position_pressure = pressure
|
||||
state.position_health_score = health_score
|
||||
state.position_health_status = self._position_health_status(health_score)
|
||||
state.position_health_reason = self._position_health_reason(
|
||||
pressure=pressure,
|
||||
trend_alignment=trend_alignment,
|
||||
adverse_momentum=adverse_momentum,
|
||||
)
|
||||
state.position_risk_level = risk_level
|
||||
state.position_risk_reason = risk_reason
|
||||
state.position_trend_alignment = trend_alignment
|
||||
state.position_adverse_momentum = adverse_momentum
|
||||
state.position_exit_pressure = self._position_exit_pressure(
|
||||
state=state,
|
||||
pnl_percent=pnl_percent,
|
||||
risk_level=risk_level,
|
||||
)
|
||||
|
||||
# рассчитать PnL позиции в процентах от notional
|
||||
def _position_pnl_percent(self, state: AutoTradeState) -> float | None:
|
||||
entry_price = safe_float(state.entry_price)
|
||||
size = safe_float(state.position_size)
|
||||
pnl = safe_float(state.unrealized_pnl_usd)
|
||||
|
||||
if entry_price is None or entry_price <= 0:
|
||||
return None
|
||||
|
||||
if size is None or size <= 0:
|
||||
return None
|
||||
|
||||
if pnl is None:
|
||||
return None
|
||||
|
||||
notional = entry_price * size
|
||||
|
||||
if notional <= 0:
|
||||
return None
|
||||
|
||||
return round((pnl / notional) * 100, 4)
|
||||
|
||||
# рассчитать время удержания открытой позиции
|
||||
def _position_hold_seconds(self, state: AutoTradeState) -> int | None:
|
||||
opened_at = getattr(state, "position_opened_monotonic_at", None)
|
||||
|
||||
if opened_at is None:
|
||||
return None
|
||||
|
||||
opened = safe_float(opened_at)
|
||||
|
||||
if opened is None:
|
||||
return None
|
||||
|
||||
return max(0, int(time.monotonic() - opened))
|
||||
|
||||
# определить давление на позицию по PnL
|
||||
def _position_pressure(
|
||||
self,
|
||||
*,
|
||||
state: AutoTradeState,
|
||||
pnl_percent: NumericLike | None,
|
||||
) -> str:
|
||||
pnl = safe_float(state.unrealized_pnl_usd) or 0.0
|
||||
percent = safe_float(pnl_percent)
|
||||
|
||||
if percent is None:
|
||||
if pnl < 0:
|
||||
return "LOSS"
|
||||
|
||||
if pnl > 0:
|
||||
return "PROFIT"
|
||||
|
||||
return "FLAT"
|
||||
|
||||
if percent <= -0.8:
|
||||
return "HIGH_LOSS"
|
||||
|
||||
if percent <= -0.3:
|
||||
return "LOSS"
|
||||
|
||||
if percent >= 0.8:
|
||||
return "STRONG_PROFIT"
|
||||
|
||||
if percent >= 0.3:
|
||||
return "PROFIT"
|
||||
|
||||
return "FLAT"
|
||||
|
||||
# определить alignment позиции относительно тренда
|
||||
def _position_trend_alignment(self, state: AutoTradeState) -> str:
|
||||
side = str(state.position_side or "NONE").upper()
|
||||
market_state = str(state.market_state or "").upper()
|
||||
trend = str(state.market_trend or "").upper()
|
||||
|
||||
if side == "NONE":
|
||||
return "NONE"
|
||||
|
||||
if side == "LONG":
|
||||
if market_state == "TREND_UP" or trend == "UP":
|
||||
return "ALIGNED"
|
||||
|
||||
if market_state == "TREND_DOWN" or trend == "DOWN":
|
||||
return "AGAINST"
|
||||
|
||||
if side == "SHORT":
|
||||
if market_state == "TREND_DOWN" or trend == "DOWN":
|
||||
return "ALIGNED"
|
||||
|
||||
if market_state == "TREND_UP" or trend == "UP":
|
||||
return "AGAINST"
|
||||
|
||||
return "NEUTRAL"
|
||||
|
||||
# проверить, направлен ли momentum против позиции
|
||||
def _has_adverse_position_momentum(self, state: AutoTradeState) -> bool:
|
||||
side = str(state.position_side or "NONE").upper()
|
||||
momentum_direction = str(state.momentum_direction or "").upper()
|
||||
momentum_state = str(state.momentum_state or "").upper()
|
||||
|
||||
if side == "LONG":
|
||||
return (
|
||||
momentum_direction == "DOWN"
|
||||
or momentum_state in {"MOMENTUM_DOWN", "BREAKOUT_DOWN"}
|
||||
)
|
||||
|
||||
if side == "SHORT":
|
||||
return (
|
||||
momentum_direction == "UP"
|
||||
or momentum_state in {"MOMENTUM_UP", "BREAKOUT_UP"}
|
||||
)
|
||||
|
||||
return False
|
||||
|
||||
# рассчитать health score позиции
|
||||
def _position_health_score(
|
||||
self,
|
||||
*,
|
||||
state: AutoTradeState,
|
||||
pnl_percent: NumericLike | None,
|
||||
trend_alignment: str,
|
||||
adverse_momentum: bool,
|
||||
) -> int:
|
||||
score = 100
|
||||
percent = safe_float(pnl_percent)
|
||||
|
||||
if percent is not None:
|
||||
if percent <= -1.0:
|
||||
score -= 35
|
||||
elif percent <= -0.5:
|
||||
score -= 22
|
||||
elif percent < 0:
|
||||
score -= 10
|
||||
elif percent >= 0.8:
|
||||
score += 5
|
||||
|
||||
if trend_alignment == "AGAINST":
|
||||
score -= 25
|
||||
elif trend_alignment == "NEUTRAL":
|
||||
score -= 8
|
||||
|
||||
if adverse_momentum:
|
||||
score -= 20
|
||||
|
||||
if state.execution_quality == "BLOCKED":
|
||||
score -= 15
|
||||
elif state.execution_quality == "WARNING":
|
||||
score -= 8
|
||||
|
||||
if state.market_runtime_degraded:
|
||||
score -= 10
|
||||
|
||||
return max(0, min(100, score))
|
||||
|
||||
# классифицировать health status по score
|
||||
def _position_health_status(self, score: int | None) -> str:
|
||||
if score is None:
|
||||
return "UNKNOWN"
|
||||
|
||||
if score >= 80:
|
||||
return "HEALTHY"
|
||||
|
||||
if score >= 55:
|
||||
return "WATCH"
|
||||
|
||||
if score >= 35:
|
||||
return "PRESSURE"
|
||||
|
||||
return "DANGER"
|
||||
|
||||
# сформировать человекочитаемую причину health состояния
|
||||
def _position_health_reason(
|
||||
self,
|
||||
*,
|
||||
pressure: str,
|
||||
trend_alignment: str,
|
||||
adverse_momentum: bool,
|
||||
) -> str:
|
||||
if trend_alignment == "AGAINST" and adverse_momentum:
|
||||
return "тренд и momentum против позиции"
|
||||
|
||||
if trend_alignment == "AGAINST":
|
||||
return "тренд против позиции"
|
||||
|
||||
if adverse_momentum:
|
||||
return "momentum против позиции"
|
||||
|
||||
if pressure in {"HIGH_LOSS", "LOSS"}:
|
||||
return "позиция под давлением"
|
||||
|
||||
if pressure in {"PROFIT", "STRONG_PROFIT"}:
|
||||
return "позиция в прибыли"
|
||||
|
||||
return "позиция стабильна"
|
||||
|
||||
# определить runtime risk level позиции
|
||||
def _position_risk_level(
|
||||
self,
|
||||
*,
|
||||
state: AutoTradeState,
|
||||
pnl_percent: NumericLike | None,
|
||||
trend_alignment: str,
|
||||
adverse_momentum: bool,
|
||||
) -> tuple[str, str]:
|
||||
percent = safe_float(pnl_percent)
|
||||
|
||||
if state.execution_quality == "BLOCKED":
|
||||
return "HIGH", "исполнение заблокировано"
|
||||
|
||||
if percent is not None and percent <= -1.0:
|
||||
return "HIGH", "сильная просадка позиции"
|
||||
|
||||
if trend_alignment == "AGAINST" and adverse_momentum:
|
||||
return "HIGH", "рынок движется против позиции"
|
||||
|
||||
if percent is not None and percent < 0:
|
||||
if trend_alignment == "AGAINST" or adverse_momentum:
|
||||
return "ELEVATED", "убыток усиливается рыночным контекстом"
|
||||
|
||||
return "MODERATE", "позиция в минусе"
|
||||
|
||||
if adverse_momentum:
|
||||
return "MODERATE", "momentum против позиции"
|
||||
|
||||
return "LOW", "критичных рисков нет"
|
||||
|
||||
# определить давление на выход из позиции
|
||||
def _position_exit_pressure(
|
||||
self,
|
||||
*,
|
||||
state: AutoTradeState,
|
||||
pnl_percent: NumericLike | None,
|
||||
risk_level: str,
|
||||
) -> str:
|
||||
percent = safe_float(pnl_percent)
|
||||
|
||||
if risk_level == "HIGH":
|
||||
return "HIGH"
|
||||
|
||||
if risk_level == "ELEVATED":
|
||||
return "WATCH"
|
||||
|
||||
if percent is not None and percent <= -0.5:
|
||||
return "WATCH"
|
||||
|
||||
return "LOW"
|
||||
Reference in New Issue
Block a user