# app/src/trading/strategies/trend.py from __future__ import annotations import time from typing import Any from src.integrations.exchange.service import ExchangeService from src.trading.market_analysis.models import ( EmaDistanceState, EntryTimingState, MarketPhase, MarketState, MomentumState, TrendDirection, TrendQuality, TrendStrength, VolatilityState, ) from src.trading.market_analysis.service import MarketAnalysisService from src.trading.strategies.base import StrategyContext from src.trading.strategies.signals import SignalResult, SignalType class TrendStrategy: name = "TREND" # Live-окно цен хранится в памяти стратегии. # Оно нужно не для общего анализа рынка, а для подтверждения, # что цена прямо сейчас действительно движется в нужную сторону. _price_window: dict[str, list[float]] = {} _price_window_updated_at: dict[str, float] = {} # Через сколько секунд live-окно считается устаревшим. _window_ttl_seconds = 60 # Сколько последних live-точек используем для подтверждения импульса. _window_size = 8 # Минимальное изменение цены внутри live-окна. # 0.05 = 0.05%. _threshold_percent = 0.05 # Какая доля движений внутри окна должна идти в сторону сделки. # 0.6 = минимум 60% шагов должны быть в сторону входа. _min_direction_ratio = 0.6 # Основной таймфрейм анализа стратегии. _market_interval = "5m" # PUBLIC API def reset_runtime(self, symbol: str | None = None) -> None: """ Сбрасывает runtime-память стратегии. Используется при: - смене актива; - смене стратегии; - перезапуске автоторговли; - ручном сбросе состояния. Если symbol=None — очищаем все live-окна. Если symbol указан — очищаем только данные этого актива. """ if symbol is None: self._price_window.clear() self._price_window_updated_at.clear() return normalized_symbol = symbol.upper() keys_to_delete = [ key for key in self._price_window.keys() if key.upper() == normalized_symbol ] for key in keys_to_delete: self._price_window.pop(key, None) self._price_window_updated_at.pop(key, None) def analyze(self, context: StrategyContext) -> SignalResult: """ Главная функция стратегии. Последовательность анализа: 1. Получаем market analysis по свечам 5m: - состояние рынка; - тренд; - волатильность; - momentum; - breakout; - структура рынка; - EMA-дистанция; - HTF-тренд 1h. 2. Получаем live snapshot: - bid; - ask; - last price. 3. Рассчитываем рабочую цену анализа: - если есть bid/ask — берём середину; - иначе используем last price. 4. Обновляем live-окно последних цен. 5. Собираем base_payload: - все market metrics; - все причины блокировок; - данные для UI, журнала и execution confidence. 6. Сначала проверяем breakout. Это важно: после выхода из сжатия рынок ещё может выглядеть как NOISY / COMPRESSED / RANGE, но первая сделка часто появляется именно в этот момент. 7. Если breakout не найден — применяем защитные фильтры рынка. 8. Если рынок разрешён — ждём достаточное live-окно. 9. По live-окну подтверждаем направление: - TREND_UP + live-импульс вверх = BUY; - TREND_DOWN + live-импульс вниз = SELL; - иначе HOLD. Важно: Эта функция не открывает сделку. Она только возвращает BUY / SELL / HOLD. Сделку потом открывает ExecutionEngine, если сигнал подтвердился. """ market = MarketAnalysisService().analyze( context.symbol, interval=self._market_interval, limit=200, ) snapshot_result = self._snapshot_or_hold( context=context, market=market, ) if isinstance(snapshot_result, SignalResult): return snapshot_result snapshot = snapshot_result symbol = str(snapshot.get("symbol") or context.symbol) current_price = self._analysis_price(snapshot) if current_price <= 0: return self._invalid_price_hold( symbol=symbol, snapshot=snapshot, market=market, ) prices = self._update_price_window( symbol=symbol, current_price=current_price, ) base_payload = self._base_payload( market=market, symbol=symbol, snapshot=snapshot, current_price=current_price, prices=prices, ) # 1. Сначала проверяем пробой. # Пробой — это исключение из обычной логики фильтров. # Его нельзя ставить после market_block, иначе ранние входы # из сжатия будут отсеиваться как COMPRESSED / NOISY / RANGE. breakout_signal = self._breakout_signal( market=market, base_payload=base_payload, ) if breakout_signal is not None: return breakout_signal # 2. Если пробоя нет — включаем обычную защиту. # Здесь отсекаются плохие условия: # - рынок не в тренде; # - волатильность плохая; # - HTF против входа; # - momentum не подтверждает; # - структура против входа; # - поздний вход. market_block = self._market_block_signal( market=market, base_payload=base_payload, ) if market_block is not None: return market_block # 3. Если рынок хороший, но live-данных ещё мало, # не открываем сделку вслепую. if len(prices) < self._window_size: return self._hold( reason="Недостаточно live-данных для подтверждения TREND.", block_reason="NOT_ENOUGH_LIVE_DATA", block_message="мало данных", payload={ **base_payload, "window_size": len(prices), "required_window_size": self._window_size, }, ) # 4. Подтверждаем направление по live-движению. return self._live_trend_signal( market=market, base_payload=base_payload, prices=prices, ) # STEP 1. SNAPSHOT / PRICE def _snapshot_or_hold( self, *, context: StrategyContext, market: Any, ) -> dict[str, Any] | SignalResult: """ Получает live snapshot с биржи. Snapshot нужен для: - актуальной цены; - bid/ask; - проверки spread дальше по цепочке; - формирования payload для UI и журнала. Если snapshot получить нельзя — стратегия возвращает HOLD. Это безопаснее, чем строить сигнал на устаревших свечах. """ try: return ExchangeService().get_market_snapshot( context.symbol, runtime_key="auto", ) except Exception as exc: return SignalResult( signal=SignalType.HOLD, reason="Не удалось получить рыночный snapshot. Безопасный HOLD.", confidence=0.0, payload={ **dict(market.payload or {}), "strategy": self.name, "symbol": context.symbol, "error": str(exc), "market_analysis": market.payload, "entry_block_reason": "MARKET_SNAPSHOT_ERROR", "entry_block_message": "нет данных рынка", }, ) def _invalid_price_hold( self, *, symbol: str, snapshot: dict[str, Any], market: Any, ) -> SignalResult: """ Возвращает HOLD, если цена из snapshot некорректная. Без валидной цены нельзя: - обновить live-окно; - рассчитать импульс; - рассчитать цену входа. """ return SignalResult( signal=SignalType.HOLD, reason="Некорректная рыночная цена. Безопасный HOLD.", confidence=0.0, payload={ **dict(market.payload or {}), "strategy": self.name, "symbol": symbol, "snapshot": snapshot, "market_analysis": market.payload, "entry_block_reason": "INVALID_MARKET_PRICE", "entry_block_message": "нет цены", }, ) def _analysis_price( self, snapshot: dict[str, Any], ) -> float: """ Выбирает цену для анализа live-импульса. Приоритет: 1. midpoint между bid и ask; 2. last_price; 3. 0.0, если цены нет. Midpoint лучше last_price, потому что меньше зависит от случайного последнего трейда. """ bid = self._safe_float(snapshot.get("bid_price")) ask = self._safe_float(snapshot.get("ask_price")) if bid is not None and ask is not None and bid > 0 and ask > 0: return (bid + ask) / 2 last = self._safe_float(snapshot.get("last_price")) if last is not None: return last return 0.0 # STEP 2. LIVE PRICE WINDOW def _update_price_window( self, *, symbol: str, current_price: float, ) -> list[float]: """ Обновляет live-окно последних цен. Логика: - если окно устарело по TTL — очищаем его; - добавляем новую цену; - если цен больше лимита — удаляем самую старую. Это окно показывает не общий тренд по свечам, а краткосрочное движение прямо сейчас. """ now = time.monotonic() previous_updated_at = self._price_window_updated_at.get(symbol) if ( previous_updated_at is not None and now - previous_updated_at > self._window_ttl_seconds ): self._price_window.pop(symbol, None) self._price_window_updated_at.pop(symbol, None) prices = self._price_window.setdefault(symbol, []) prices.append(current_price) self._price_window_updated_at[symbol] = now if len(prices) > self._window_size: prices.pop(0) return prices # STEP 3. PAYLOAD def _base_payload( self, *, market: Any, symbol: str, snapshot: dict[str, Any], current_price: float, prices: list[float], ) -> dict[str, Any]: """ Собирает единый payload стратегии. Payload используется дальше в: - AutoTradeState; - Telegram UI; - diagnostics; - journal; - execution confidence; - supervisor block logs. Поэтому сюда кладём не только итоговый сигнал, но и все промежуточные признаки рынка. """ return { # Сначала переносим полный payload MarketAnalysisService. # Это защищает от потери новых market-полей: # market_long_score / market_short_score, # last_closed_candle_change_percent, # current_interval_change_percent и т.д. **dict(market.payload or {}), # Ниже стратегия добавляет/переопределяет runtime-поля. "strategy": self.name, "symbol": symbol, "analysis_price": current_price, "last_price": snapshot.get("last_price"), "bid_price": snapshot.get("bid_price"), "ask_price": snapshot.get("ask_price"), "market_state": market.state.value, "market_trend": market.trend.value, "market_volatility": market.volatility.value, "market_analysis_interval": market.interval, "market_analysis_reason": market.reason, "market_analysis": market.payload, "market_trend_strength": market.trend_strength.value, "market_trend_quality": market.trend_quality.value, "market_phase": self._normalized_market_phase(market), "market_phase_direction": self._normalized_market_phase_direction(market), "market_phase_change_percent": market.phase_change_percent, "market_phase_direction_consistency": market.payload.get( "market_phase_direction_consistency" ), "current_interval_change_percent": market.current_interval_change_percent, "current_interval_direction": ( market.current_interval_direction.value if market.current_interval_direction is not None else "UNKNOWN" ), "current_interval_label": market.current_interval_label or market.interval, # Общая оценка рынка 0..100. # Она рассчитана в MarketAnalysisService из всех рыночных факторов # и дальше используется UI / diagnostics / execution / adaptive size. "market_score": market.market_score, "market_score_label": market.market_score_label, "market_long_score": market.market_long_score, "market_short_score": market.market_short_score, "last_closed_candle_change_percent": market.payload.get( "last_closed_candle_change_percent" ), "last_closed_candle_direction": market.payload.get( "last_closed_candle_direction" ), "market_phase_reason": market.phase_reason, "market_structure": ( market.market_structure.value if market.market_structure is not None else "UNKNOWN" ), "market_structure_reason": market.market_structure_reason, "momentum_state": ( market.momentum_state.value if market.momentum_state is not None else "UNKNOWN" ), "momentum_direction": ( market.momentum_direction.value if market.momentum_direction is not None else "UNKNOWN" ), "momentum_change_percent": market.momentum_change_percent, "momentum_strength": market.momentum_strength, "breakout_level": market.breakout_level, "breakout_distance_percent": market.breakout_distance_percent, "breakout_reason": market.breakout_reason, "market_trend_gap_percent": market.trend_gap_percent, "market_trend_consistency": market.trend_consistency, "market_trend_efficiency": market.trend_efficiency, "trend_quality_score": market.trend_quality_score, "ema_distance_atr_ratio": market.ema_distance_atr_ratio, "ema_distance_state": ( market.ema_distance_state.value if market.ema_distance_state is not None else "UNKNOWN" ), "entry_timing_state": ( market.entry_timing_state.value if market.entry_timing_state is not None else "UNKNOWN" ), "entry_timing_reason": market.entry_timing_reason, "candle_noise_score": market.payload.get("candle_noise_score"), "price_position_score": market.payload.get("price_position_score"), "rsi": market.rsi, "rsi_overbought": market.payload.get("rsi_overbought"), "rsi_oversold": market.payload.get("rsi_oversold"), "htf_market_state": ( market.htf_market_state.value if market.htf_market_state is not None else "UNKNOWN" ), "htf_trend": ( market.htf_trend.value if market.htf_trend is not None else "UNKNOWN" ), "htf_trend_strength": ( market.htf_trend_strength.value if market.htf_trend_strength is not None else "UNKNOWN" ), "htf_trend_quality": ( market.htf_trend_quality.value if market.htf_trend_quality is not None else "UNKNOWN" ), "htf_market_phase": ( market.htf_market_phase.value if market.htf_market_phase is not None else "UNKNOWN" ), "htf_alignment": market.htf_alignment, "htf_confirmation_score": market.htf_confirmation_score, "htf_reason": market.htf_reason, "runtime_window_ttl_seconds": self._window_ttl_seconds, "runtime_window_size": len(prices), } # STEP 4. BREAKOUT FIRST def _breakout_signal( self, *, market: Any, base_payload: dict[str, Any], ) -> SignalResult | None: """ Проверяет ранний вход по пробою. Почему breakout проверяется ДО обычных блокировок: - после флэта EMA часто ещё сжаты; - качество может выглядеть шумным; - state может не успеть стать идеальным TREND_UP/TREND_DOWN; - но momentum уже показывает реальный пробой. Условия BUY: - momentum_state = BREAKOUT_UP; - локальный тренд вверх; - momentum вверх; - HTF не против входа. Условия SELL: - momentum_state = BREAKOUT_DOWN; - локальный тренд вниз; - momentum вниз; - HTF не против входа. """ momentum_state = getattr(market, "momentum_state", MomentumState.UNKNOWN) momentum_strength = float(getattr(market, "momentum_strength", 0.0) or 0.0) if ( momentum_state == MomentumState.BREAKOUT_UP and market.trend == TrendDirection.UP and market.momentum_direction == TrendDirection.UP and market.htf_alignment in {"ALIGNED", "SAME_INTERVAL"} ): return SignalResult( signal=SignalType.BUY, reason="BREAKOUT_UP подтверждён momentum/breakout semantic layer.", confidence=self._calculate_breakout_confidence(momentum_strength), payload={ **base_payload, "entry_block_reason": None, "entry_block_message": None, "breakout_signal": True, "expected_direction": "BUY", }, ) if ( momentum_state == MomentumState.BREAKOUT_DOWN and market.trend == TrendDirection.DOWN and market.momentum_direction == TrendDirection.DOWN and market.htf_alignment in {"ALIGNED", "SAME_INTERVAL"} ): return SignalResult( signal=SignalType.SELL, reason="BREAKOUT_DOWN подтверждён momentum/breakout semantic layer.", confidence=self._calculate_breakout_confidence(momentum_strength), payload={ **base_payload, "entry_block_reason": None, "entry_block_message": None, "breakout_signal": True, "expected_direction": "SELL", }, ) return None # STEP 5. MARKET BLOCKS # определить ранний impulse-вход из RANGE/SQUEEZE def _early_impulse_direction(self, market: Any) -> str | None: if market.state not in {MarketState.RANGE, MarketState.LOW_VOLATILITY}: return None if market.market_phase != MarketPhase.IMPULSE: return None if market.htf_alignment not in {"ALIGNED", "SAME_INTERVAL"}: return None if ( market.trend == TrendDirection.UP and market.momentum_direction == TrendDirection.UP and market.momentum_state in { MomentumState.MOMENTUM_UP, MomentumState.BREAKOUT_UP, } ): return "BUY" if ( market.trend == TrendDirection.DOWN and market.momentum_direction == TrendDirection.DOWN and market.momentum_state in { MomentumState.MOMENTUM_DOWN, MomentumState.BREAKOUT_DOWN, } ): return "SELL" return None def _market_block_signal( self, *, market: Any, base_payload: dict[str, Any], ) -> SignalResult | None: early_impulse_direction = self._early_impulse_direction(market) """ Проверяет защитные фильтры рынка. Эта функция НЕ ищет вход. Она только отвечает на вопрос: "Можно ли вообще рассматривать вход по TREND?" Если найден риск — возвращает HOLD с причиной. Если всё нормально — возвращает None. """ if ( market.state not in {MarketState.TREND_UP, MarketState.TREND_DOWN} and early_impulse_direction is None ): return self._hold( reason=f"Market state не подходит для TREND: {market.state.value}.", block_reason="MARKET_STATE_NOT_TREND", block_message="рынок не в тренде", payload=base_payload, ) if market.volatility in { VolatilityState.LOW, VolatilityState.UNKNOWN, }: return self._hold( reason="Волатильность не подходит для входа.", block_reason="BAD_VOLATILITY", block_message="волатильность не подходит", payload=base_payload, ) if market.htf_alignment == "AGAINST": return self._hold( reason="HTF trend против направления входа.", block_reason="HTF_TREND_AGAINST", block_message="старший таймфрейм против входа", payload=base_payload, ) if market.htf_alignment not in {"ALIGNED", "SAME_INTERVAL"}: return self._hold( reason="HTF trend не подтвердил направление входа.", block_reason="HTF_TREND_NOT_CONFIRMED", block_message="старший таймфрейм не подтвердил вход", payload=base_payload, ) # Слабый тренд больше не блокируем всегда. # Если рынок уже TREND_UP / TREND_DOWN, старший ТФ подтверждает вход, # качество не NOISY, а фаза IMPULSE — даём live-окну проверить движение. # Это снижает число пропущенных ранних трендовых входов. soft_trend_allowed = ( market.trend_strength == TrendStrength.WEAK and market.market_phase == MarketPhase.IMPULSE and market.trend_quality != TrendQuality.NOISY and market.htf_alignment in {"ALIGNED", "SAME_INTERVAL"} ) if ( market.trend_strength == TrendStrength.WEAK and not soft_trend_allowed ): return self._hold( reason="TREND есть, но сила тренда слабая.", block_reason="WEAK_MARKET_TREND", block_message="слабый тренд", payload=base_payload, ) # RANGE / SQUEEZE больше не режем полностью, # если уже появился momentum/breakout по направлению тренда. # Иначе бот слишком поздно входит после выхода из флэта. phase_breakout_context = ( ( market.state == MarketState.TREND_UP and market.momentum_state == MomentumState.BREAKOUT_UP and market.momentum_direction == TrendDirection.UP ) or ( market.state == MarketState.TREND_DOWN and market.momentum_state == MomentumState.BREAKOUT_DOWN and market.momentum_direction == TrendDirection.DOWN ) ) if ( market.market_phase in {MarketPhase.RANGE, MarketPhase.SQUEEZE} and not phase_breakout_context ): return self._hold( reason="Фаза рынка не подходит для входа по TREND.", block_reason=f"MARKET_PHASE_{market.market_phase.value}", block_message="фаза рынка не подходит", payload=base_payload, ) structure_block = self._market_structure_block( market=market, base_payload=base_payload, ) if structure_block is not None: return structure_block momentum_block = self._momentum_block( market=market, base_payload=base_payload, ) if momentum_block is not None: return momentum_block # COMPRESSED больше не блокируем здесь жёстко. # Сжатие EMA может быть не плохим рынком, а ранней стадией выхода из флэта. # OVEREXTENDED оставляем блокировкой, потому что это часто поздний вход. if market.ema_distance_state == EmaDistanceState.OVEREXTENDED: return self._hold( reason="EMA-дистанция не подходит для входа.", block_reason=f"EMA_DISTANCE_{market.ema_distance_state.value}", block_message="EMA-дистанция не подходит", payload=base_payload, ) if market.entry_timing_state in { EntryTimingState.LATE, EntryTimingState.CHASING, }: return self._hold( reason="Тайминг входа запоздалый.", block_reason=f"ENTRY_TIMING_{market.entry_timing_state.value}", block_message="тайминг входа запоздалый", payload=base_payload, ) if not market.is_trade_allowed: return self._hold( reason=f"Market filter: {market.reason}", block_reason="MARKET_FILTER_BLOCKED", block_message="рынок сейчас не подходит для входа", payload={ **base_payload, "market_filter_blocked": True, }, ) return None def _market_structure_block( self, *, market: Any, base_payload: dict[str, Any], ) -> SignalResult | None: """ Проверяет структуру рынка. Для LONG структура LH/LL плохая: - lower high; - lower low; - рынок делает понижающиеся экстремумы. Для SHORT структура HH/HL плохая: - higher high; - higher low; - рынок делает повышающиеся экстремумы. """ market_structure = ( market.market_structure.value if market.market_structure is not None else "UNKNOWN" ) if market.state == MarketState.TREND_UP and market_structure == "LH_LL": return self._hold( reason="Структура рынка против LONG.", block_reason="MARKET_STRUCTURE_CONFLICT", block_message="структура рынка против LONG", payload={ **base_payload, "expected_direction": "BUY", }, ) if market.state == MarketState.TREND_DOWN and market_structure == "HH_HL": return self._hold( reason="Структура рынка против SHORT.", block_reason="MARKET_STRUCTURE_CONFLICT", block_message="структура рынка против SHORT", payload={ **base_payload, "expected_direction": "SELL", }, ) return None def _momentum_block( self, *, market: Any, base_payload: dict[str, Any], ) -> SignalResult | None: # Проверяет, не идёт ли momentum явно против входа. # Важно: # FLAT больше не блокируем сразу. # Если общий TREND-контекст хороший, live-окно ниже само проверит, # есть ли реальное движение прямо сейчас. # Блокируем только явный momentum против направления сделки. if market.state == MarketState.TREND_UP: if market.momentum_direction == TrendDirection.DOWN: return self._hold( reason="Momentum явно против LONG.", block_reason="MOMENTUM_CONFLICT", block_message="momentum против LONG", payload={ **base_payload, "expected_direction": "BUY", }, ) if market.state == MarketState.TREND_DOWN: if market.momentum_direction == TrendDirection.UP: return self._hold( reason="Momentum явно против SHORT.", block_reason="MOMENTUM_CONFLICT", block_message="momentum против SHORT", payload={ **base_payload, "expected_direction": "SELL", }, ) return None # ========================================================= # STEP 6. LIVE TREND CONFIRMATION # ========================================================= def _live_trend_signal( self, *, market: Any, base_payload: dict[str, Any], prices: list[float], ) -> SignalResult: """ Финальное подтверждение обычного TREND-входа. Market analysis говорит: - рынок в тренде; - направление известно; - фильтры разрешили вход. Но перед сделкой нужно проверить live-окно: - цена реально пошла в нужную сторону; - движение не единичный случайный тик; - достаточно шагов подтверждают направление. """ first_price = prices[0] last_price = prices[-1] if first_price <= 0: return self._hold( reason="Некорректная стартовая цена в live-окне.", block_reason="INVALID_WINDOW_PRICE", block_message="ошибка цены", payload={ **base_payload, "prices": prices, }, ) change_percent = ((last_price - first_price) / first_price) * 100 direction_ratio = self._direction_ratio(prices, change_percent) payload = { **base_payload, "first_price": first_price, "current_price": last_price, "change_percent": round(change_percent, 5), "direction_ratio": round(direction_ratio, 3), "window_size": len(prices), "threshold_percent": self._threshold_percent, "min_direction_ratio": self._min_direction_ratio, } early_impulse_direction = self._early_impulse_direction(market) if market.state == MarketState.TREND_UP: return self._trend_up_signal( change_percent=change_percent, direction_ratio=direction_ratio, payload=payload, ) if market.state == MarketState.TREND_DOWN: return self._trend_down_signal( change_percent=change_percent, direction_ratio=direction_ratio, payload=payload, ) if early_impulse_direction == "BUY": return self._trend_up_signal( change_percent=change_percent, direction_ratio=direction_ratio, payload={ **payload, "early_impulse_signal": True, "expected_direction": "BUY", }, ) if early_impulse_direction == "SELL": return self._trend_down_signal( change_percent=change_percent, direction_ratio=direction_ratio, payload={ **payload, "early_impulse_signal": True, "expected_direction": "SELL", }, ) return self._hold( reason=f"Market state не подходит для TREND: {market.state.value}.", block_reason="MARKET_STATE_NOT_TREND", block_message="рынок не в тренде", payload=payload, ) def _trend_up_signal( self, *, change_percent: float, direction_ratio: float, payload: dict[str, Any], ) -> SignalResult: """ Формирует BUY, если live-окно подтвердило рост. """ if ( change_percent >= self._threshold_percent and direction_ratio >= self._min_direction_ratio ): return SignalResult( signal=SignalType.BUY, reason="TREND_UP подтверждён market analysis и live-импульсом.", confidence=self._calculate_confidence( change_percent, direction_ratio, ), payload={ **payload, "entry_block_reason": None, "entry_block_message": None, "expected_direction": "BUY", }, ) return self._hold( reason="TREND_UP есть, но live-импульс вверх недостаточно сильный.", block_reason="WEAK_UP_IMPULSE", block_message="слабый импульс вверх", payload={ **payload, "expected_direction": "BUY", }, ) def _trend_down_signal( self, *, change_percent: float, direction_ratio: float, payload: dict[str, Any], ) -> SignalResult: """ Формирует SELL, если live-окно подтвердило снижение. """ if ( change_percent <= -self._threshold_percent and direction_ratio >= self._min_direction_ratio ): return SignalResult( signal=SignalType.SELL, reason="TREND_DOWN подтверждён market analysis и live-импульсом.", confidence=self._calculate_confidence( change_percent, direction_ratio, ), payload={ **payload, "entry_block_reason": None, "entry_block_message": None, "expected_direction": "SELL", }, ) return self._hold( reason="TREND_DOWN есть, но live-импульс вниз недостаточно сильный.", block_reason="WEAK_DOWN_IMPULSE", block_message="слабый импульс вниз", payload={ **payload, "expected_direction": "SELL", }, ) # ========================================================= # RESULT HELPERS # ========================================================= def _hold( self, *, reason: str, block_reason: str, block_message: str, payload: dict[str, Any], ) -> SignalResult: """ Унифицированный HOLD. Все блокировки входа должны проходить через эту функцию, чтобы UI, журнал и supervisor получали одинаковые поля: - entry_block_reason; - entry_block_message. """ return SignalResult( signal=SignalType.HOLD, reason=reason, confidence=0.0, payload={ **payload, "entry_block_reason": block_reason, "entry_block_message": block_message, }, ) # ========================================================= # CALCULATIONS # ========================================================= def _direction_ratio(self, prices: list[float], change_percent: float) -> float: """ Считает долю движений в сторону общего изменения. Пример: prices = [100, 101, 102, 101.5, 103] Если итоговое движение вверх: считаем долю шагов, где цена росла. Если итоговое движение вниз: считаем долю шагов, где цена снижалась. Это защищает от ситуации: цена вроде изменилась на нужный процент, но внутри окна движение было рваным и случайным. """ if len(prices) < 2: return 0.0 up_moves = 0 down_moves = 0 for previous_price, current_price in zip(prices, prices[1:]): if current_price > previous_price: up_moves += 1 elif current_price < previous_price: down_moves += 1 total_moves = max(1, len(prices) - 1) if change_percent >= 0: return up_moves / total_moves return down_moves / total_moves def _calculate_breakout_confidence(self, momentum_strength: float) -> float: """ Рассчитывает confidence для breakout-сигнала. Breakout начинается с базовой уверенности 0.55. Чем сильнее momentum_strength, тем выше confidence. Максимум ограничен 0.95, чтобы execution confidence всё равно учитывал spread, подтверждение и качество исполнения. """ strength_score = min(1.0, max(0.0, momentum_strength) / 2) confidence = 0.55 + (strength_score * 0.35) return round(min(0.95, confidence), 2) def _calculate_confidence( self, change_percent: float, direction_ratio: float, ) -> float: """ Рассчитывает confidence обычного TREND-сигнала. Учитываются: - сила движения относительно threshold; - направленность движения внутри live-окна. Чем сильнее и чище live-импульс, тем выше confidence. """ strength = abs(change_percent) / self._threshold_percent if strength < 1: return 0.0 strength_score = min(1.0, strength / 3) direction_score = min(1.0, direction_ratio) confidence = 0.3 + (strength_score * 0.4) + (direction_score * 0.3) return round(min(1.0, confidence), 2) # ========================================================= # NORMALIZATION # ========================================================= def _normalized_market_phase(self, market: Any) -> str: """ Безопасно возвращает market_phase строкой. """ if market.market_phase is None: return "UNKNOWN" return market.market_phase.value def _normalized_market_phase_direction(self, market: Any) -> str: """ Безопасно возвращает phase_direction строкой. """ if market.phase_direction is None: return "UNKNOWN" return market.phase_direction.value def _safe_float( self, value: float | int | str | None, ) -> float | None: """ Безопасное приведение к float. Нужно потому, что snapshot может вернуть числа: - как float; - как int; - как строку; - как None. """ if value is None: return None try: return float(value) except (TypeError, ValueError): return None