07.4.3.19.1 - Position-aware Flip Protection

app/src/telegram/handlers/auto/ui.py

@@ -208,6 +208,7 @@ def _build_active_position_text(state) -> str:
         f"<b>Доступно</b> · $ {_format_money_compact(available)}",
         f"<b>Зарезервировано</b> · $ {_format_money_compact(reserved)}",
         f"<b>P&L</b> {_format_signed_usd_with_direction(pnl)}",
+        *_execution_block_lines(state),
         "",
         (
             f"{side_icon} <b>{_asset_symbol(state.symbol)}</b> · "

app/src/trading/auto/state.py

@@ -92,4 +92,16 @@ class AutoTradeState:
     allocated_balance_usd: float = 1000.0
 
     # зафиксированный результат закрытых paper-сделок
-    realized_pnl_usd: float = 0.0
+    realized_pnl_usd: float = 0.0
+
+    # последнее execution-действие
+    last_execution_action: str | None = None
+
+    # последняя execution-причина
+    last_execution_reason: str | None = None
+
+    # последняя причина блокировки flip
+    last_flip_block_reason: str | None = None
+
+    # время последнего успешного flip
+    last_flip_at: str | None = None

app/src/trading/execution/engine.py

@@ -26,6 +26,11 @@ class _ExecutionPrice:
 class ExecutionEngine:
     _position = PositionState()
     _size_precision = 5
+    _min_flip_confidence = 0.75
+    _min_flip_repeat_count = 3
+    _min_flip_hold_seconds = 60
+    _loss_flip_confidence = 0.9
+    _last_flip_block_key: str | None = None
 
     def get_position(self) -> PositionState:
         return type(self)._position
@@ -46,6 +51,11 @@ class ExecutionEngine:
             return ExecutionDecision("NONE", False, "Сигнал ещё не готов к execution.")
 
         if self._should_flip_position(state):
+            flip_block_reason = self._flip_block_reason(state)
+
+            if flip_block_reason is not None:
+                return self._block_flip(state, flip_block_reason)
+
             return self._flip_position(state)
 
         if state.last_signal == "BUY":
@@ -112,6 +122,10 @@ class ExecutionEngine:
         )
 
         self._sync_state_from_position(state)
+        state.execution_block_reason = None
+        state.last_flip_block_reason = None
+        state.last_execution_action = action
+        state.last_execution_reason = f"Paper ENTRY {side} открыта."
 
         payload = {
             "execution_type": "ENTRY",
@@ -210,6 +224,12 @@ class ExecutionEngine:
         )
 
         self._sync_state_from_position(state)
+        state.execution_block_reason = None
+        state.last_flip_block_reason = None
+        state.last_execution_action = f"FLIP_{old_side}_TO_{new_side}"
+        state.last_execution_reason = "Paper FLIP выполнен."
+        state.last_flip_at = now
+        type(self)._last_flip_block_key = None
 
         payload = {
             "execution_type": "FLIP",
@@ -333,6 +353,19 @@ class ExecutionEngine:
 
         type(self)._position = PositionState()
         self._sync_state_from_position(state)
+        state.execution_block_reason = None
+        state.last_flip_block_reason = None
+        state.last_execution_action = (
+            f"FORCE_CLOSE_{forced_reason}"
+            if forced_reason is not None
+            else "CLOSE"
+        )
+        state.last_execution_reason = (
+            f"Paper EXIT выполнена по риску: {forced_reason}."
+            if forced_reason is not None
+            else "Paper EXIT выполнена."
+        )
+        type(self)._last_flip_block_key = None
 
         if forced_reason is not None:
             return ExecutionDecision(
@@ -431,6 +464,106 @@ class ExecutionEngine:
 
         return False
 
+    def _flip_block_reason(self, state: AutoTradeState) -> str | None:
+        position = type(self)._position
+
+        confidence = float(state.last_signal_confidence or 0.0)
+        repeat_count = int(state.last_signal_repeat_count or 0)
+        unrealized_pnl = float(state.unrealized_pnl_usd or 0.0)
+        hold_seconds = self._position_hold_seconds(position)
+
+        if confidence < self._min_flip_confidence:
+            return (
+                "Flip blocked: signal confidence "
+                f"{confidence:.2f} < {self._min_flip_confidence:.2f}."
+            )
+
+        if repeat_count < self._min_flip_repeat_count:
+            return (
+                "Flip blocked: repeat count "
+                f"{repeat_count} < {self._min_flip_repeat_count}."
+            )
+
+        if hold_seconds is not None and hold_seconds < self._min_flip_hold_seconds:
+            return (
+                "Flip blocked: position hold time "
+                f"{hold_seconds}s < {self._min_flip_hold_seconds}s."
+            )
+
+        if unrealized_pnl < 0 and confidence < self._loss_flip_confidence:
+            return (
+                "Flip blocked: position is negative and signal is not strong enough "
+                f"({confidence:.2f} < {self._loss_flip_confidence:.2f})."
+            )
+
+        return None
+
+    def _block_flip(
+        self,
+        state: AutoTradeState,
+        reason: str,
+    ) -> ExecutionDecision:
+        position = type(self)._position
+        confidence = float(state.last_signal_confidence or 0.0)
+
+        state.execution_block_reason = reason
+        state.last_flip_block_reason = reason
+        state.last_execution_action = "FLIP_BLOCKED"
+        state.last_execution_reason = reason
+
+        block_key = (
+            f"{position.side}:"
+            f"{state.last_signal}:"
+            f"{state.last_signal_repeat_count}:"
+            f"{confidence:.2f}:"
+            f"{reason}"
+        )
+
+        if block_key != type(self)._last_flip_block_key:
+            type(self)._last_flip_block_key = block_key
+
+            payload = {
+                "execution_type": "FLIP_BLOCKED",
+                "symbol": state.symbol,
+                "position_side": position.side,
+                "signal": state.last_signal,
+                "confidence": confidence,
+                "repeat_count": state.last_signal_repeat_count,
+                "reason": reason,
+                "unrealized_pnl_usd": state.unrealized_pnl_usd,
+                "opened_at": position.opened_at,
+                "updated_at": position.updated_at,
+            }
+
+            JournalService().log_ui_info(
+                event_type="paper_flip_blocked",
+                message=f"Paper FLIP заблокирован: {reason}",
+                screen="auto",
+                action="paper_execution",
+                payload=payload,
+            )
+
+            EventBus.emit("paper_flip_blocked", payload)
+
+        return ExecutionDecision("NONE", False, reason)
+
+    def _position_hold_seconds(self, position: PositionState) -> int | None:
+        if not position.opened_at:
+            return None
+
+        try:
+            opened_at = datetime.strptime(position.opened_at, "%H:%M:%S")
+            now = datetime.strptime(self._now_time(), "%H:%M:%S")
+
+            seconds = int((now - opened_at).total_seconds())
+
+            if seconds < 0:
+                seconds += 24 * 60 * 60
+
+            return seconds
+        except Exception:
+            return None
+
     def _target_side_from_signal(self, signal: str | None) -> str | None:
         if signal == "BUY":
             return "LONG"