452 lines
17 KiB
Python
452 lines
17 KiB
Python
# app/src/trading/auto/position_health.py
|
|
|
|
from __future__ import annotations
|
|
|
|
from src.core.numbers import safe_float
|
|
from src.core.types import NumericLike
|
|
from src.trading.auto.state import AutoTradeState
|
|
from src.trading.auto.state_reset import reset_position_health_state
|
|
from src.trading.execution.constants import (
|
|
EXECUTION_QUALITY_BLOCKED,
|
|
EXECUTION_QUALITY_WARNING,
|
|
MARKET_VOLATILITY_HIGH_STATES,
|
|
POSITION_CURRENT_INTERVAL_ADVERSE_MOVE_PERCENT,
|
|
POSITION_CURRENT_INTERVAL_RISK_MOVE_PERCENT,
|
|
POSITION_EXIT_PRESSURE_LOSS_PERCENT,
|
|
POSITION_HEALTH_DANGER,
|
|
POSITION_HEALTH_HEALTHY,
|
|
POSITION_HEALTH_PNL_GOOD_PROFIT_PERCENT,
|
|
POSITION_HEALTH_PNL_HARD_LOSS_PERCENT,
|
|
POSITION_HEALTH_PNL_HIGH_PRESSURE_PERCENT,
|
|
POSITION_HEALTH_PNL_PRESSURE_PERCENT,
|
|
POSITION_HEALTH_PRESSURE,
|
|
POSITION_HEALTH_UNKNOWN,
|
|
POSITION_HEALTH_WATCH,
|
|
POSITION_MOMENTUM_STRONG,
|
|
POSITION_RISK_ELEVATED,
|
|
POSITION_RISK_HIGH,
|
|
POSITION_RISK_LOW,
|
|
POSITION_RISK_MODERATE,
|
|
POSITION_STOP_LOSS_RATIO_CRITICAL,
|
|
POSITION_STOP_LOSS_RATIO_WARNING,
|
|
POSITION_STOP_LOSS_RATIO_WATCH,
|
|
get_position_health_thresholds,
|
|
)
|
|
|
|
|
|
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:
|
|
reset_position_health_state(state)
|
|
return
|
|
|
|
# PnL % и время удержания больше не считаем здесь.
|
|
# Эти значения должны приходить из единого расчёта position_metrics.py
|
|
# через execution/position_runtime.py.
|
|
pnl_percent = safe_float(state.position_pnl_percent)
|
|
hold_seconds = state.position_hold_seconds
|
|
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,
|
|
)
|
|
|
|
def _health_thresholds(self, state: AutoTradeState) -> dict[str, float]:
|
|
return get_position_health_thresholds(state.symbol)
|
|
|
|
# определить давление на позицию по 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"
|
|
|
|
thresholds = self._health_thresholds(state)
|
|
|
|
if percent <= thresholds["high_loss"]:
|
|
return "HIGH_LOSS"
|
|
|
|
if percent <= thresholds["loss"]:
|
|
return "LOSS"
|
|
|
|
if percent >= thresholds["strong_profit"]:
|
|
return "STRONG_PROFIT"
|
|
|
|
if percent >= thresholds["profit"]:
|
|
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()
|
|
htf_trend = str(getattr(state, "htf_trend", "") or "").upper()
|
|
htf_alignment = str(getattr(state, "htf_alignment", "") or "").upper()
|
|
|
|
if side == "NONE":
|
|
return "NONE"
|
|
|
|
# HTF AGAINST важнее локального тренда:
|
|
# если старший таймфрейм против позиции, позиция считается рискованной.
|
|
if htf_alignment == "AGAINST":
|
|
return "AGAINST"
|
|
|
|
if side == "LONG":
|
|
if market_state == "TREND_UP" or trend == "UP":
|
|
if htf_trend in {"DOWN"}:
|
|
return "NEUTRAL"
|
|
return "ALIGNED"
|
|
|
|
if market_state == "TREND_DOWN" or trend == "DOWN":
|
|
return "AGAINST"
|
|
|
|
if side == "SHORT":
|
|
if market_state == "TREND_DOWN" or trend == "DOWN":
|
|
if htf_trend in {"UP"}:
|
|
return "NEUTRAL"
|
|
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()
|
|
momentum_strength = safe_float(getattr(state, "momentum_strength", None)) or 0.0
|
|
|
|
current_interval_direction = str(
|
|
getattr(state, "current_interval_direction", "") or ""
|
|
).upper()
|
|
current_interval_change_percent = safe_float(
|
|
getattr(state, "current_interval_change_percent", None)
|
|
)
|
|
|
|
current_interval_move_abs = abs(current_interval_change_percent or 0.0)
|
|
|
|
current_interval_against_long = (
|
|
current_interval_direction == "DOWN"
|
|
and current_interval_move_abs >= POSITION_CURRENT_INTERVAL_ADVERSE_MOVE_PERCENT
|
|
)
|
|
|
|
current_interval_against_short = (
|
|
current_interval_direction == "UP"
|
|
and current_interval_move_abs >= POSITION_CURRENT_INTERVAL_ADVERSE_MOVE_PERCENT
|
|
)
|
|
|
|
if side == "LONG":
|
|
return (
|
|
momentum_state in {"MOMENTUM_DOWN", "BREAKOUT_DOWN"}
|
|
or current_interval_against_long
|
|
or (
|
|
momentum_direction == "DOWN"
|
|
and momentum_strength >= POSITION_MOMENTUM_STRONG
|
|
)
|
|
)
|
|
|
|
if side == "SHORT":
|
|
return (
|
|
momentum_state in {"MOMENTUM_UP", "BREAKOUT_UP"}
|
|
or current_interval_against_short
|
|
or (
|
|
momentum_direction == "UP"
|
|
and momentum_strength >= POSITION_MOMENTUM_STRONG
|
|
)
|
|
)
|
|
|
|
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)
|
|
stop_loss_percent = safe_float(getattr(state, "stop_loss_percent", None))
|
|
htf_alignment = str(getattr(state, "htf_alignment", "") or "").upper()
|
|
market_structure = str(getattr(state, "market_structure", "") or "").upper()
|
|
market_phase = str(getattr(state, "market_phase", "") or "").upper()
|
|
trend_quality = str(getattr(state, "market_trend_quality", "") or "").upper()
|
|
volatility = str(getattr(state, "market_volatility", "") or "").upper()
|
|
|
|
if percent is not None:
|
|
if percent <= POSITION_HEALTH_PNL_HARD_LOSS_PERCENT:
|
|
score -= 40
|
|
elif percent <= POSITION_HEALTH_PNL_HIGH_PRESSURE_PERCENT:
|
|
score -= 30
|
|
elif percent <= POSITION_HEALTH_PNL_PRESSURE_PERCENT:
|
|
score -= 18
|
|
elif percent < 0:
|
|
score -= 8
|
|
elif percent >= POSITION_HEALTH_PNL_GOOD_PROFIT_PERCENT:
|
|
score += 5
|
|
|
|
# Если позиция прошла большую часть stop loss — ухудшаем score заранее.
|
|
if stop_loss_percent is not None and stop_loss_percent > 0:
|
|
loss_ratio = abs(percent) / stop_loss_percent if percent < 0 else 0.0
|
|
|
|
if loss_ratio >= POSITION_STOP_LOSS_RATIO_CRITICAL:
|
|
score -= 25
|
|
elif loss_ratio >= POSITION_STOP_LOSS_RATIO_WARNING:
|
|
score -= 15
|
|
|
|
if trend_alignment == "AGAINST":
|
|
score -= 30
|
|
elif trend_alignment == "NEUTRAL":
|
|
score -= 10
|
|
|
|
if adverse_momentum:
|
|
score -= 25
|
|
|
|
if htf_alignment == "AGAINST":
|
|
score -= 20
|
|
elif htf_alignment == "NEUTRAL":
|
|
score -= 8
|
|
|
|
if market_structure == "MIXED":
|
|
score -= 12
|
|
elif market_structure == "LH_LL" and state.position_side == "LONG":
|
|
score -= 18
|
|
elif market_structure == "HH_HL" and state.position_side == "SHORT":
|
|
score -= 18
|
|
|
|
if market_phase in {"RANGE", "SQUEEZE"}:
|
|
score -= 10
|
|
elif market_phase == "PULLBACK" and trend_alignment != "ALIGNED":
|
|
score -= 12
|
|
|
|
if trend_quality == "NOISY":
|
|
score -= 12
|
|
|
|
if volatility in MARKET_VOLATILITY_HIGH_STATES:
|
|
score -= 12
|
|
|
|
if state.execution_quality == EXECUTION_QUALITY_BLOCKED:
|
|
score -= 15
|
|
elif state.execution_quality == 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 POSITION_HEALTH_UNKNOWN
|
|
|
|
if score >= 80:
|
|
return POSITION_HEALTH_HEALTHY
|
|
|
|
if score >= 60:
|
|
return POSITION_HEALTH_WATCH
|
|
|
|
if score >= 40:
|
|
return POSITION_HEALTH_PRESSURE
|
|
|
|
return POSITION_HEALTH_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)
|
|
stop_loss_percent = safe_float(getattr(state, "stop_loss_percent", None))
|
|
htf_alignment = str(getattr(state, "htf_alignment", "") or "").upper()
|
|
market_structure = str(getattr(state, "market_structure", "") or "").upper()
|
|
volatility = str(getattr(state, "market_volatility", "") or "").upper()
|
|
|
|
current_interval_direction = str(
|
|
getattr(state, "current_interval_direction", "") or ""
|
|
).upper()
|
|
current_interval_change_percent = safe_float(
|
|
getattr(state, "current_interval_change_percent", None)
|
|
)
|
|
current_interval_move_abs = abs(current_interval_change_percent or 0.0)
|
|
|
|
current_interval_against_position = (
|
|
(
|
|
state.position_side == "LONG"
|
|
and current_interval_direction == "DOWN"
|
|
)
|
|
or (
|
|
state.position_side == "SHORT"
|
|
and current_interval_direction == "UP"
|
|
)
|
|
)
|
|
|
|
if state.execution_quality == EXECUTION_QUALITY_BLOCKED:
|
|
return POSITION_RISK_HIGH, "исполнение заблокировано"
|
|
|
|
if percent is not None:
|
|
if percent <= POSITION_HEALTH_PNL_HARD_LOSS_PERCENT:
|
|
return POSITION_RISK_HIGH, "сильная просадка позиции"
|
|
|
|
if stop_loss_percent is not None and stop_loss_percent > 0 and percent < 0:
|
|
loss_ratio = abs(percent) / stop_loss_percent
|
|
|
|
if loss_ratio >= POSITION_STOP_LOSS_RATIO_CRITICAL:
|
|
return POSITION_RISK_HIGH, "позиция близко к stop loss"
|
|
|
|
if loss_ratio >= POSITION_STOP_LOSS_RATIO_WARNING:
|
|
return POSITION_RISK_ELEVATED, "позиция прошла больше половины stop loss"
|
|
|
|
if trend_alignment == "AGAINST" and adverse_momentum:
|
|
return POSITION_RISK_HIGH, "рынок движется против позиции"
|
|
|
|
if htf_alignment == "AGAINST" and adverse_momentum:
|
|
return POSITION_RISK_HIGH, "старший таймфрейм и momentum против позиции"
|
|
|
|
if (
|
|
state.position_side == "LONG"
|
|
and market_structure == "LH_LL"
|
|
):
|
|
return POSITION_RISK_ELEVATED, "структура рынка против LONG"
|
|
|
|
if (
|
|
state.position_side == "SHORT"
|
|
and market_structure == "HH_HL"
|
|
):
|
|
return POSITION_RISK_ELEVATED, "структура рынка против SHORT"
|
|
|
|
if volatility in MARKET_VOLATILITY_HIGH_STATES and percent is not None and percent < 0:
|
|
return POSITION_RISK_ELEVATED, "убыток в высокой волатильности"
|
|
|
|
if percent is not None and percent < 0:
|
|
if trend_alignment == "AGAINST" or adverse_momentum:
|
|
return POSITION_RISK_ELEVATED, "убыток усиливается рыночным контекстом"
|
|
|
|
return POSITION_RISK_MODERATE, "позиция в минусе"
|
|
|
|
if current_interval_against_position and current_interval_move_abs >= POSITION_CURRENT_INTERVAL_RISK_MOVE_PERCENT:
|
|
return POSITION_RISK_ELEVATED, "текущая 5м свеча против позиции"
|
|
|
|
if current_interval_against_position and percent is not None and percent < 0:
|
|
return POSITION_RISK_ELEVATED, "убыток усиливается текущей 5м свечой"
|
|
|
|
if adverse_momentum:
|
|
return POSITION_RISK_MODERATE, "momentum против позиции"
|
|
|
|
if htf_alignment == "AGAINST":
|
|
return POSITION_RISK_MODERATE, "старший таймфрейм против позиции"
|
|
|
|
return POSITION_RISK_LOW, "критичных рисков нет"
|
|
|
|
# определить давление на выход из позиции
|
|
def _position_exit_pressure(
|
|
self,
|
|
*,
|
|
state: AutoTradeState,
|
|
pnl_percent: NumericLike | None,
|
|
risk_level: str,
|
|
) -> str:
|
|
percent = safe_float(pnl_percent)
|
|
stop_loss_percent = safe_float(getattr(state, "stop_loss_percent", None))
|
|
|
|
if risk_level == POSITION_RISK_HIGH:
|
|
return "HIGH"
|
|
|
|
if risk_level in {POSITION_RISK_ELEVATED, POSITION_RISK_MODERATE}:
|
|
return "WATCH"
|
|
|
|
if percent is not None:
|
|
if percent <= POSITION_EXIT_PRESSURE_LOSS_PERCENT:
|
|
return "WATCH"
|
|
|
|
if stop_loss_percent is not None and stop_loss_percent > 0 and percent < 0:
|
|
loss_ratio = abs(percent) / stop_loss_percent
|
|
|
|
if loss_ratio >= POSITION_STOP_LOSS_RATIO_WATCH:
|
|
return "WATCH"
|
|
|
|
return "LOW" |