from __future__ import annotations from crypto_spot_bot.config import Settings from crypto_spot_bot.models import Candle, Position, Signal, Ticker, utc_now NEGATIVE_LONG_PATTERNS = {"нисходящий тренд", "пробой вниз", "ускоренное падение"} class SpotStrategy: def __init__(self, settings: Settings): self.settings = settings def entry_signal( self, symbol: str, candles: list[Candle], ticker: Ticker | None, open_positions_for_symbol: int, pattern: dict | None = None, learning: dict | None = None, llm: dict | None = None, forecast: dict | None = None, account: dict | None = None, trend_candles: list[Candle] | None = None, ) -> Signal: if self.settings.strategy_mode == "torch_forecast": return _torch_forecast_entry_signal( settings=self.settings, symbol=symbol, candles=candles, ticker=ticker, open_positions_for_symbol=open_positions_for_symbol, pattern=pattern or {}, llm=llm or {}, forecast=forecast or {}, account=account, ) if self.settings.strategy_mode == "trend_macd": return _trend_macd_entry_signal( settings=self.settings, symbol=symbol, candles=candles, trend_candles=trend_candles or [], ticker=ticker, open_positions_for_symbol=open_positions_for_symbol, account=account, ) if ticker is None: return Signal(symbol, "HOLD", 0.0, "нет ticker-данных") if len(candles) < 200: return Signal(symbol, "HOLD", 0.0, "недостаточно свечей для EMA200") latest = candles[-1] previous = candles[-2] if len(candles) >= 2 else latest if not _has_entry_indicators(latest): return Signal(symbol, "HOLD", 0.0, "индикаторы еще не готовы") spread_ok = ticker.spread_percent <= self.settings.max_spread_percent liquidity_ok = ticker.turnover_24h >= self.settings.min_24h_turnover_usdt trend_ok = latest.close > latest.ema_200 or latest.ema_20 > latest.ema_50 pullback_ok = 35 <= latest.rsi_14 <= 58 and latest.close <= latest.ema_20 * 1.012 momentum_ok = latest.ema_20 >= latest.ema_50 or latest.close > previous.close volume_ok = latest.volume_ma_20 is not None and latest.volume >= latest.volume_ma_20 * 0.75 atr_percent = (latest.atr_14 / latest.close) * 100 if latest.close else 0.0 volatility_ok = 0.04 <= atr_percent <= 6.0 weights = { "spread": 0.18, "liquidity": 0.14, "trend": 0.16, "pullback": 0.18, "momentum": 0.14, "volume": 0.10, "volatility": 0.10, } score = ( weights["spread"] * float(spread_ok) + weights["liquidity"] * float(liquidity_ok) + weights["trend"] * float(trend_ok) + weights["pullback"] * float(pullback_ok) + weights["momentum"] * float(momentum_ok) + weights["volume"] * float(volume_ok) + weights["volatility"] * float(volatility_ok) ) pattern = pattern or {} learning = learning or {} llm = llm or {} forecast = forecast or {} pattern_label = str(pattern.get("label") or "") pattern_score = float(pattern.get("score", 0.5) or 0.5) pattern_adjustment = ( (pattern_score - 0.5) * self.settings.pattern_score_weight if self.settings.pattern_analysis_enabled else 0.0 ) learning_adjustment = float(learning.get("confidence_adjustment", 0.0) or 0.0) forecast_adjustment = ( float(forecast.get("confidence_adjustment", 0.0) or 0.0) if self.settings.time_series_forecast_enabled else 0.0 ) adaptive = _adaptive_rules(learning) adaptive_entry_adjustment = _adaptive_threshold_adjustment(adaptive) falling_market = _falling_market(latest, previous, pattern_label, llm) llm_adjustment = float(llm.get("confidence_adjustment", 0.0) or 0.0) rebound = _rebound_state( settings=self.settings, candles=candles, latest=latest, previous=previous, pattern=pattern, llm=llm, spread_ok=spread_ok, liquidity_ok=liquidity_ok, volume_ok=volume_ok, volatility_ok=volatility_ok, atr_percent=atr_percent, ) adaptive_blocks_entry = _adaptive_blocks_entry(adaptive, falling_market, rebound["active"]) base_final_score = score + pattern_adjustment + learning_adjustment + llm_adjustment + forecast_adjustment rebound_entry_score = float(rebound.get("entry_score", 0.0) or 0.0) final_score = _clamp(max(base_final_score, rebound_entry_score), 0.0, 1.0) learning_blocks_entry = _learning_blocks_entry( learning=learning, learning_adjustment=learning_adjustment, min_samples=self.settings.learning_min_samples, max_adjustment=self.settings.learning_max_adjustment, enabled=self.settings.learning_enabled, ) llm_blocks_entry = bool(llm.get("block_entry", False)) and self.settings.llm_advisor_enabled forecast_blocks_entry = ( bool(forecast.get("block_entry", False)) and self.settings.time_series_forecast_enabled and bool(forecast.get("usable", False)) ) grid = _grid_state( settings=self.settings, latest=latest, pattern=pattern, llm=llm, atr_percent=atr_percent, spread_ok=spread_ok, liquidity_ok=liquidity_ok, volatility_ok=volatility_ok, ) base_entry_threshold = ( self.settings.grid_entry_confidence if grid["active"] else self.settings.rebound_entry_confidence if rebound["active"] else self.settings.min_signal_confidence ) entry_threshold = _clamp(base_entry_threshold + adaptive_entry_adjustment, 0.45, 0.92) negative_pattern = ( self.settings.pattern_analysis_enabled and pattern_label in NEGATIVE_LONG_PATTERNS and pattern_score <= 0.32 ) pattern_blocks_entry = negative_pattern and not ( rebound["active"] and rebound_entry_score >= entry_threshold ) position_sizing = _position_sizing( settings=self.settings, final_score=final_score, grid_active=grid["active"], rebound_active=rebound["active"], forecast=forecast, adaptive=adaptive, account=account, ) position_notional = float(position_sizing["notional_usdt"]) trade_mode = "GRID" if grid["active"] else "REBOUND" if rebound["active"] else "NORMAL" diagnostics = { "base_score": round(score, 4), "pattern_adjustment": round(pattern_adjustment, 4), "learning_adjustment": round(learning_adjustment, 4), "llm_adjustment": round(llm_adjustment, 4), "forecast_adjustment": round(forecast_adjustment, 4), "rebound_probability": rebound["probability"], "rebound_entry_score": round(rebound_entry_score, 4), "final_score": round(final_score, 4), "entry_blocked_by_pattern": pattern_blocks_entry, "entry_blocked_by_learning": learning_blocks_entry, "entry_blocked_by_adaptive_rules": adaptive_blocks_entry, "adaptive_block_reason": _adaptive_block_reason(adaptive, falling_market, rebound["active"]), "entry_blocked_by_llm": llm_blocks_entry, "entry_blocked_by_forecast": forecast_blocks_entry, "falling_market": falling_market, "open_positions_for_symbol": open_positions_for_symbol, "position_notional_usdt": position_notional, "position_sizing": position_sizing, "trade_mode": trade_mode, "base_entry_threshold": round(base_entry_threshold, 4), "adaptive_entry_threshold_adjustment": round(adaptive_entry_adjustment, 4), "entry_threshold": round(entry_threshold, 4), "adaptive_rules": adaptive, "stop_loss_percent": _adaptive_percent( adaptive, "stop_loss_percent", self.settings.stop_loss_percent, 0.003, 0.08 ), "take_profit_percent": _adaptive_percent( adaptive, "take_profit_percent", self.settings.take_profit_percent, 0.003, 0.20 ), "trailing_stop_percent": _adaptive_percent( adaptive, "trailing_stop_percent", self.settings.trailing_stop_percent, 0.003, 0.08 ), "grid": grid, "rebound": rebound, "pattern": pattern, "learning": learning, "llm": llm, "forecast": forecast, "spread_percent": round(ticker.spread_percent, 5), "turnover_24h": ticker.turnover_24h, "rsi_14": latest.rsi_14, "ema_20": latest.ema_20, "ema_50": latest.ema_50, "ema_200": latest.ema_200, "volume": latest.volume, "volume_ma_20": latest.volume_ma_20, "atr_percent": atr_percent, "checks": { "spread_ok": spread_ok, "liquidity_ok": liquidity_ok, "trend_ok": trend_ok, "pullback_ok": pullback_ok, "momentum_ok": momentum_ok, "volume_ok": volume_ok, "volatility_ok": volatility_ok, "rebound_active": rebound["active"], }, } suffix = _decision_suffix(pattern, learning, llm) if pattern_blocks_entry: return Signal( symbol, "HOLD", round(final_score, 4), f"покупка заблокирована отрицательным LONG-шаблоном: {pattern_label}{suffix}", diagnostics, ) if learning_blocks_entry: return Signal( symbol, "HOLD", round(final_score, 4), f"покупка заблокирована обучением: похожие сделки были убыточными{suffix}", diagnostics, ) if adaptive_blocks_entry: return Signal( symbol, "HOLD", round(final_score, 4), f"покупка заблокирована адаптивными правилами обучения: символ или шаблон в стоп-листе{suffix}", diagnostics, ) if llm_blocks_entry: return Signal( symbol, "HOLD", round(final_score, 4), f"покупка заблокирована LLM Advisor: {llm.get('reason_ru') or 'модель вернула block_entry=true'}{suffix}", diagnostics, ) if forecast_blocks_entry: return Signal( symbol, "HOLD", round(final_score, 4), f"покупка заблокирована прогнозом временного ряда: {forecast.get('reason') or 'ожидаемое движение вниз'}{suffix}", diagnostics, ) if grid["active"] and not grid["buy_zone"] and not rebound["active"]: return Signal( symbol, "HOLD", round(final_score, 4), f"grid-режим активен, но цена не в зоне покупки: {grid['reason']}{suffix}", diagnostics, ) if final_score >= entry_threshold: mode_reason = ( f"grid-режим: покупка в нижней части диапазона, размер {position_notional:.2f} USDT" if grid["active"] else f"rebound-сценарий: падение стабилизировалось, вероятность {rebound['probability']:.2f}, размер {position_notional:.2f} USDT" if rebound["active"] else f"условия покупки набрали достаточную оценку, размер {position_notional:.2f} USDT" ) return Signal( symbol, "BUY", round(final_score, 4), f"{mode_reason}{suffix}", diagnostics, ) return Signal( symbol, "HOLD", round(final_score, 4), f"оценка входа ниже порога{suffix}", diagnostics, ) def _legacy_exit_signal( self, position: Position, candles: list[Candle], ticker: Ticker | None, learning: dict | None = None, ) -> Signal: if ticker is None: return Signal(position.symbol, "HOLD", 0.0, "нет ticker-данных для выхода") if not candles: return Signal(position.symbol, "HOLD", 0.0, "нет свечей для выхода") latest = candles[-1] previous = candles[-2] if len(candles) >= 2 else latest price = ticker.last_price trailing = position.trailing_stop(self.settings.trailing_stop_percent) diagnostics = { "price": price, "entry_price": position.entry_price, "stop_loss": position.stop_loss, "take_profit": position.take_profit, "highest_price": position.highest_price, "trailing_stop": trailing, "rsi_14": latest.rsi_14, "ema_20": latest.ema_20, "ema_50": latest.ema_50, } if price <= position.stop_loss: return Signal(position.symbol, "SELL", 1.0, "сработал стоп-лосс", diagnostics) if price >= position.take_profit: return Signal(position.symbol, "SELL", 0.96, "сработал тейк-профит", diagnostics) if trailing is not None and price <= trailing: return Signal(position.symbol, "SELL", 0.90, "сработал трейлинг-стоп выше цены входа", diagnostics) hold_seconds = (utc_now() - position.opened_at).total_seconds() diagnostics["hold_seconds"] = hold_seconds if hold_seconds < self.settings.min_hold_seconds: return Signal(position.symbol, "HOLD", 0.45, "минимальное время удержания еще не прошло", diagnostics) if adaptive.get("reduce_exposure") and adaptive.get("reduce_now"): return Signal( position.symbol, "SELL", 0.88, "обучение снижает общую экспозицию до целевого уровня", diagnostics, ) if latest.rsi_14 is not None and latest.rsi_14 >= 72 and latest.close < previous.close: return Signal(position.symbol, "SELL", 0.76, "RSI высокий и цена начала снижаться", diagnostics) if ( latest.ema_20 is not None and latest.ema_50 is not None and latest.ema_20 < latest.ema_50 and latest.close < latest.ema_50 ): return Signal(position.symbol, "SELL", 0.70, "краткосрочный тренд ослаб ниже EMA50", diagnostics) return Signal(position.symbol, "HOLD", 0.35, "условия выхода не выполнены", diagnostics) def exit_signal( self, position: Position, candles: list[Candle], ticker: Ticker | None, learning: dict | None = None, forecast: dict | None = None, ) -> Signal: if self.settings.strategy_mode == "torch_forecast": return _torch_forecast_exit_signal(self.settings, position, candles, ticker, forecast or {}) if self.settings.strategy_mode == "trend_macd": return _trend_macd_exit_signal(self.settings, position, candles, ticker) if ticker is None: return Signal(position.symbol, "HOLD", 0.0, "нет ticker-данных для выхода") if not candles: return Signal(position.symbol, "HOLD", 0.0, "нет свечей для выхода") latest = candles[-1] previous = candles[-2] if len(candles) >= 2 else latest price = ticker.last_price adaptive = _adaptive_rules(learning or {}) forecast = forecast or {} stop_loss_percent = _adaptive_percent( adaptive, "stop_loss_percent", self.settings.stop_loss_percent, 0.003, 0.08 ) take_profit_percent = _adaptive_percent( adaptive, "take_profit_percent", self.settings.take_profit_percent, 0.003, 0.20 ) trailing_percent = _adaptive_percent( adaptive, "trailing_stop_percent", self.settings.trailing_stop_percent, 0.003, 0.08 ) effective_stop_loss = max(position.stop_loss, position.entry_price * (1 - stop_loss_percent)) effective_take_profit = position.entry_price * (1 + take_profit_percent) trailing = position.trailing_stop(trailing_percent) estimated_exit_net_percent = _estimated_exit_net_percent(position, price, self.settings) diagnostics = { "price": price, "entry_price": position.entry_price, "stop_loss": effective_stop_loss, "take_profit": effective_take_profit, "highest_price": position.highest_price, "trailing_stop": trailing, "rsi_14": latest.rsi_14, "ema_20": latest.ema_20, "ema_50": latest.ema_50, "adaptive_rules": adaptive, "forecast": forecast, "estimated_exit_net_percent": round(estimated_exit_net_percent, 4), "min_exit_profit_percent": float(adaptive.get("min_exit_profit_percent", 0.0) or 0.0), } if price <= effective_stop_loss: return Signal(position.symbol, "SELL", 1.0, "сработал стоп-лосс", diagnostics) if price >= effective_take_profit: return Signal(position.symbol, "SELL", 0.96, "сработал тейк-профит", diagnostics) if trailing is not None and price <= trailing: return Signal(position.symbol, "SELL", 0.90, "сработал трейлинг-стоп выше цены входа", diagnostics) hold_seconds = (utc_now() - position.opened_at).total_seconds() diagnostics["hold_seconds"] = hold_seconds adaptive_min_hold = int(float(adaptive.get("min_hold_seconds", self.settings.min_hold_seconds) or 0)) min_hold_seconds = max(self.settings.min_hold_seconds, adaptive_min_hold) diagnostics["min_hold_seconds"] = min_hold_seconds if adaptive.get("reduce_exposure") and adaptive.get("reduce_now") and hold_seconds >= min_hold_seconds: return Signal( position.symbol, "SELL", 0.88, "обучение снижает общую экспозицию до целевого уровня", diagnostics, ) if hold_seconds < min_hold_seconds: return Signal(position.symbol, "HOLD", 0.45, "минимальное время удержания еще не прошло", diagnostics) forecast_exit = _forecast_exit_signal( forecast=forecast, position=position, price=price, estimated_exit_net_percent=estimated_exit_net_percent, stop_loss_percent=stop_loss_percent, min_edge_percent=self.settings.time_series_min_edge_percent, ) if forecast_exit is not None: action, confidence, reason = forecast_exit return Signal(position.symbol, action, confidence, reason, diagnostics) if latest.rsi_14 is not None and latest.rsi_14 >= 72 and latest.close < previous.close: if _adaptive_indicator_exit_allowed(adaptive, "rsi_exit_mode", estimated_exit_net_percent): return Signal(position.symbol, "SELL", 0.76, "RSI высокий и цена начала снижаться", diagnostics) return Signal( position.symbol, "HOLD", 0.44, "обучение удерживает позицию: RSI-выход убыточен после издержек", diagnostics, ) if ( latest.ema_20 is not None and latest.ema_50 is not None and latest.ema_20 < latest.ema_50 and latest.close < latest.ema_50 ): if _adaptive_indicator_exit_allowed(adaptive, "ema_exit_mode", estimated_exit_net_percent): return Signal(position.symbol, "SELL", 0.70, "краткосрочный тренд ослаб ниже EMA50", diagnostics) return Signal( position.symbol, "HOLD", 0.44, "обучение удерживает позицию: EMA50-выход убыточен после издержек", diagnostics, ) return Signal(position.symbol, "HOLD", 0.35, "условия выхода не выполнены", diagnostics) def _has_entry_indicators(candle: Candle) -> bool: return all( value is not None for value in ( candle.ema_20, candle.ema_50, candle.ema_200, candle.rsi_14, candle.atr_14, candle.volume_ma_20, ) ) def _trend_macd_entry_signal( *, settings: Settings, symbol: str, candles: list[Candle], trend_candles: list[Candle], ticker: Ticker | None, open_positions_for_symbol: int, account: dict | None, ) -> Signal: if ticker is None: return Signal(symbol, "HOLD", 0.0, "нет ticker-данных") if open_positions_for_symbol > 0: return Signal(symbol, "HOLD", 0.0, "позиция по паре уже открыта") if len(candles) < 60: return Signal(symbol, "HOLD", 0.0, "недостаточно 1h свечей для trend_macd") if len(trend_candles) < 200: return Signal(symbol, "HOLD", 0.0, "недостаточно 1d свечей для EMA200") latest = candles[-1] previous = candles[-2] trend_latest = trend_candles[-1] if not _has_trend_entry_indicators(latest, previous, trend_latest): return Signal(symbol, "HOLD", 0.0, "индикаторы trend_macd еще не готовы") spread_ok = ticker.spread_percent <= settings.max_spread_percent liquidity_ok = ticker.turnover_24h >= settings.min_24h_turnover_usdt daily_trend_ok = bool(trend_latest.close > trend_latest.ema_200 and trend_latest.ema_50 > trend_latest.ema_200) macd_cross_up = _macd_crossed_up(previous, latest) price_above_ema50 = bool(latest.close > latest.ema_50) rsi_min = min(settings.trend_rsi_min, settings.trend_rsi_max) rsi_max = max(settings.trend_rsi_min, settings.trend_rsi_max) rsi_ok = bool(rsi_min <= latest.rsi_14 <= rsi_max) stop_loss_percent = _clamp(settings.stop_loss_percent, 0.003, 0.08) sizing = _trend_position_sizing(settings, account, stop_loss_percent) position_notional = float(sizing["notional_usdt"]) checks = { "spread_ok": spread_ok, "liquidity_ok": liquidity_ok, "daily_trend_ok": daily_trend_ok, "macd_cross_up": macd_cross_up, "price_above_ema50": price_above_ema50, "rsi_ok": rsi_ok, "risk_size_ok": position_notional >= settings.min_position_usdt, } diagnostics = { "strategy_mode": "trend_macd", "trade_mode": "TREND_MACD", "position_notional_usdt": position_notional, "position_sizing": sizing, "stop_loss_percent": stop_loss_percent, "atr_trailing_multiplier": _clamp(settings.atr_trailing_multiplier, 0.5, 10.0), "entry_timeframe": settings.base_interval, "trend_timeframe": settings.trend_interval, "rsi_14": latest.rsi_14, "rsi_min": rsi_min, "rsi_max": rsi_max, "ema_50": latest.ema_50, "macd": latest.macd, "macd_signal": latest.macd_signal, "trend_close": trend_latest.close, "trend_ema_50": trend_latest.ema_50, "trend_ema_200": trend_latest.ema_200, "spread_percent": round(ticker.spread_percent, 5), "turnover_24h": ticker.turnover_24h, "checks": checks, "grid": {"enabled": False, "active": False}, "rebound": {"enabled": False, "active": False}, "forecast": {}, "learning": {}, "llm": {}, } if all(checks.values()): return Signal( symbol, "BUY", 0.86, f"trend_macd: 1d тренд вверх, MACD пересек signal вверх, RSI {latest.rsi_14:.1f}, размер {position_notional:.2f} USDT", diagnostics, ) failed = ", ".join(name for name, ok in checks.items() if not ok) return Signal(symbol, "HOLD", 0.35, f"trend_macd: условия входа не выполнены ({failed})", diagnostics) def _trend_macd_exit_signal( settings: Settings, position: Position, candles: list[Candle], ticker: Ticker | None, ) -> Signal: if ticker is None: return Signal(position.symbol, "HOLD", 0.0, "нет ticker-данных для выхода") if len(candles) < 2: return Signal(position.symbol, "HOLD", 0.0, "недостаточно 1h свечей для выхода") latest = candles[-1] previous = candles[-2] price = ticker.last_price stop_loss_percent = _clamp(settings.stop_loss_percent, 0.003, 0.08) effective_stop_loss = max(position.stop_loss, position.entry_price * (1 - stop_loss_percent)) atr_multiplier = _clamp(settings.atr_trailing_multiplier, 0.5, 10.0) atr_trailing_stop = None if latest.atr_14 is not None and position.highest_price > position.entry_price: atr_trailing_stop = max(effective_stop_loss, position.highest_price - latest.atr_14 * atr_multiplier) macd_cross_down = _macd_crossed_down(previous, latest) close_below_ema50 = latest.ema_50 is not None and latest.close < latest.ema_50 diagnostics = { "strategy_mode": "trend_macd", "price": price, "entry_price": position.entry_price, "stop_loss": effective_stop_loss, "atr_trailing_stop": atr_trailing_stop, "atr_trailing_multiplier": atr_multiplier, "highest_price": position.highest_price, "ema_50": latest.ema_50, "rsi_14": latest.rsi_14, "atr_14": latest.atr_14, "macd": latest.macd, "macd_signal": latest.macd_signal, "macd_cross_down": macd_cross_down, "close_below_ema50": close_below_ema50, } if price <= effective_stop_loss: return Signal(position.symbol, "SELL", 1.0, "trend_macd: сработал стоп-лосс", diagnostics) if atr_trailing_stop is not None and price <= atr_trailing_stop: return Signal(position.symbol, "SELL", 0.94, "trend_macd: сработал ATR trailing stop", diagnostics) if macd_cross_down: return Signal(position.symbol, "SELL", 0.84, "trend_macd: MACD пересек signal вниз", diagnostics) if close_below_ema50: return Signal(position.symbol, "SELL", 0.82, "trend_macd: 1h свеча закрылась ниже EMA50", diagnostics) return Signal(position.symbol, "HOLD", 0.35, "trend_macd: условия выхода не выполнены", diagnostics) def _torch_forecast_entry_signal( *, settings: Settings, symbol: str, candles: list[Candle] | None, ticker: Ticker | None, open_positions_for_symbol: int, pattern: dict, llm: dict, forecast: dict, account: dict | None, ) -> Signal: if ticker is None: return Signal(symbol, "HOLD", 0.0, "torch_forecast: no ticker data") if open_positions_for_symbol >= _dynamic_symbol_position_limit(settings): return Signal(symbol, "HOLD", 0.0, "torch_forecast: symbol position limit reached") stop_loss_percent = _clamp(settings.stop_loss_percent, 0.003, 0.08) sizing = _torch_forecast_position_sizing(settings, account, stop_loss_percent, forecast) position_notional = float(sizing["notional_usdt"]) expected_return = _safe_float(forecast.get("expected_return_percent"), 0.0) probability_up = _safe_float(forecast.get("probability_up"), 0.5) skill = _safe_float(forecast.get("skill"), 0.0) min_edge = max(0.0, settings.time_series_min_edge_percent) min_probability = _torch_min_probability(settings) probe_min_edge = max(0.0, min(settings.time_series_probe_min_edge_percent, min_edge)) probe_min_probability = round( _clamp(settings.time_series_probe_min_probability_up, min_probability, 0.85), 4, ) full_edge_ok = expected_return >= min_edge probe_edge_ok = bool( settings.time_series_probe_enabled and not full_edge_ok and expected_return >= probe_min_edge and probability_up >= probe_min_probability ) edge_mode = "full" if full_edge_ok else ("probe" if probe_edge_ok else "blocked") if probe_edge_ok and position_notional > 0: probe_multiplier = _clamp(settings.time_series_probe_size_multiplier, 0.05, 1.0) position_notional = round( min( settings.max_position_usdt, max(settings.min_position_usdt, position_notional * probe_multiplier), ), 2, ) sizing = { **sizing, "notional_usdt": position_notional, "probe_size_multiplier": round(probe_multiplier, 4), "edge_mode": "probe", } confidence = _torch_forecast_confidence(settings, forecast) spread_ok = ticker.spread_percent <= settings.max_spread_percent liquidity_ok = ticker.turnover_24h >= settings.min_24h_turnover_usdt model_ok = _is_torch_forecast(forecast) quality_gate_ok = forecast.get("quality_gate_passed") is not False rebound = _torch_rebound_overlay( settings=settings, candles=candles or [], ticker=ticker, pattern=pattern, llm=llm, spread_ok=spread_ok, liquidity_ok=liquidity_ok, ) rebound_model_probability_min = round( _clamp(settings.time_series_probe_min_probability_up, 0.50, 0.75), 4, ) missing_torch_model = _missing_torch_model(forecast) model_rebound_entry_ok = bool( rebound.get("active") and model_ok and quality_gate_ok and bool(forecast.get("usable", False)) and not bool(forecast.get("block_entry", False)) and expected_return >= 0.0 and probability_up >= rebound_model_probability_min and skill > 0.0 and confidence >= settings.time_series_min_confidence ) fallback_rebound_entry_ok = bool( settings.time_series_rebound_fallback_enabled and rebound.get("active") and missing_torch_model and quality_gate_ok and not bool(forecast.get("block_entry", False)) and confidence >= settings.time_series_min_confidence ) rebound_entry_ok = model_rebound_entry_ok or fallback_rebound_entry_ok if rebound_entry_ok and position_notional > 0: rebound_cap = max(settings.min_position_usdt, settings.rebound_max_position_usdt) position_notional = round( min(settings.max_position_usdt, rebound_cap, max(settings.min_position_usdt, position_notional)), 2, ) sizing_method = "torch_forecast_rebound_fallback" if fallback_rebound_entry_ok else "torch_forecast_rebound" sizing = { **sizing, "method": sizing_method, "notional_usdt": position_notional, "edge_mode": "rebound_fallback" if fallback_rebound_entry_ok else "rebound", "rebound_probability": rebound.get("probability", 0.0), } edge_mode = "rebound_fallback" if fallback_rebound_entry_ok else "rebound" checks = { "torch_model_ok": model_ok, "quality_gate_ok": quality_gate_ok, "forecast_usable": bool(forecast.get("usable", False)), "forecast_not_blocked": not bool(forecast.get("block_entry", False)), "expected_edge_ok": full_edge_ok or probe_edge_ok, "probability_ok": probability_up >= min_probability, "skill_ok": skill > 0.0, "confidence_ok": confidence >= settings.time_series_min_confidence, "spread_ok": spread_ok, "liquidity_ok": liquidity_ok, "risk_size_ok": position_notional >= settings.min_position_usdt, } diagnostics = { "strategy_mode": "torch_forecast", "trade_mode": "TORCH_FORECAST", "forecast": forecast, "position_notional_usdt": position_notional, "position_sizing": sizing, "stop_loss_percent": stop_loss_percent, "atr_trailing_multiplier": _clamp(settings.atr_trailing_multiplier, 0.5, 10.0), "expected_return_percent": expected_return, "min_edge_percent": min_edge, "probe_enabled": settings.time_series_probe_enabled, "probe_min_edge_percent": probe_min_edge, "probe_min_probability_up": probe_min_probability, "edge_mode": edge_mode, "probability_up": probability_up, "min_probability_up": min_probability, "rebound_model_probability_min": rebound_model_probability_min, "missing_torch_model": missing_torch_model, "time_series_rebound_fallback_enabled": settings.time_series_rebound_fallback_enabled, "model_rebound_entry_ok": model_rebound_entry_ok, "fallback_rebound_entry_ok": fallback_rebound_entry_ok, "rebound_entry_ok": rebound_entry_ok, "min_confidence": settings.time_series_min_confidence, "skill": skill, "quality_gate": forecast.get("quality_gate", {}), "quality_gate_passed": forecast.get("quality_gate_passed"), "spread_percent": round(ticker.spread_percent, 5), "turnover_24h": ticker.turnover_24h, "checks": checks, "grid": {"enabled": False, "active": False}, "rebound": rebound, "learning": {}, "llm": {}, } base_entry_ok = all(checks.values()) if base_entry_ok or rebound_entry_ok: buy_confidence = max(confidence, float(rebound.get("probability", 0.0) or 0.0)) if rebound_entry_ok else confidence entry_path = edge_mode if rebound_entry_ok and not base_entry_ok else edge_mode diagnostics["entry_path"] = entry_path if fallback_rebound_entry_ok and not base_entry_ok: reason = ( "torch_forecast: rebound fallback confirmed without PyTorch model; " f"rebound_probability={float(rebound.get('probability', 0.0) or 0.0):.3f}, " f"size={position_notional:.2f} USDT" ) elif rebound_entry_ok and not base_entry_ok: reason = ( "torch_forecast: rebound overlay confirmed; " f"model={forecast.get('model')}, p_up={probability_up:.3f}, " f"expected={expected_return:.4f}%, rebound_probability={float(rebound.get('probability', 0.0) or 0.0):.3f}, " f"size={position_notional:.2f} USDT" ) else: reason = ( "torch_forecast: PyTorch edge confirmed; " f"model={forecast.get('model')}, p_up={probability_up:.3f}, " f"expected={expected_return:.4f}%, edge_mode={edge_mode}, " f"size={position_notional:.2f} USDT" ) return Signal( symbol, "BUY", round(_clamp(buy_confidence, 0.0, 0.96), 4), reason, diagnostics, ) failed = ", ".join(name for name, ok in checks.items() if not ok) return Signal(symbol, "HOLD", confidence, f"torch_forecast: entry blocked ({failed})", diagnostics) def _torch_rebound_overlay( *, settings: Settings, candles: list[Candle], ticker: Ticker, pattern: dict, llm: dict, spread_ok: bool, liquidity_ok: bool, ) -> dict: if not settings.rebound_trading_enabled: return {"enabled": False, "active": False, "reason": "rebound trading disabled"} if len(candles) < 21: return {"enabled": True, "active": False, "reason": "not enough candles for rebound"} latest = candles[-1] previous = candles[-2] if len(candles) >= 2 else latest if not _has_entry_indicators(latest): return {"enabled": True, "active": False, "reason": "entry indicators are not ready"} volume_ok = latest.volume_ma_20 is not None and latest.volume >= latest.volume_ma_20 * 0.75 atr_percent = (latest.atr_14 / latest.close) * 100 if latest.close and latest.atr_14 is not None else 0.0 volatility_ok = 0.04 <= atr_percent <= 6.0 return _rebound_state( settings=settings, candles=candles, latest=latest, previous=previous, pattern=pattern, llm=llm, spread_ok=spread_ok, liquidity_ok=liquidity_ok, volume_ok=volume_ok, volatility_ok=volatility_ok, atr_percent=atr_percent, ) def _torch_forecast_exit_signal( settings: Settings, position: Position, candles: list[Candle], ticker: Ticker | None, forecast: dict, ) -> Signal: if ticker is None: return Signal(position.symbol, "HOLD", 0.0, "torch_forecast: no ticker data for exit") latest = candles[-1] if candles else None price = ticker.last_price stop_loss_percent = _clamp(settings.stop_loss_percent, 0.003, 0.08) effective_stop_loss = max(position.stop_loss, position.entry_price * (1 - stop_loss_percent)) atr_multiplier = _clamp(settings.atr_trailing_multiplier, 0.5, 10.0) atr_trailing_stop = None if latest and latest.atr_14 is not None and position.highest_price > position.entry_price: atr_trailing_stop = max(effective_stop_loss, position.highest_price - latest.atr_14 * atr_multiplier) expected_return = _safe_float(forecast.get("expected_return_percent"), 0.0) probability_up = _safe_float(forecast.get("probability_up"), 0.5) skill = _safe_float(forecast.get("skill"), 0.0) min_edge = max(0.0, settings.time_series_min_edge_percent) min_probability = _torch_min_probability(settings) estimated_exit_net_percent = _estimated_exit_net_percent(position, price, settings) entry_path = str(position.entry_diagnostics.get("entry_path", "")) entry_edge_mode = str(position.entry_diagnostics.get("edge_mode", "")) rebound_fallback_position = entry_path == "rebound_fallback" or entry_edge_mode == "rebound_fallback" diagnostics = { "strategy_mode": "torch_forecast", "price": price, "entry_price": position.entry_price, "stop_loss": effective_stop_loss, "take_profit": position.take_profit, "atr_trailing_stop": atr_trailing_stop, "atr_trailing_multiplier": atr_multiplier, "highest_price": position.highest_price, "entry_path": entry_path, "entry_edge_mode": entry_edge_mode, "rebound_fallback_position": rebound_fallback_position, "forecast": forecast, "expected_return_percent": expected_return, "min_edge_percent": min_edge, "probability_up": probability_up, "min_probability_up": min_probability, "skill": skill, "estimated_exit_net_percent": round(estimated_exit_net_percent, 4), "atr_14": latest.atr_14 if latest else None, } if price <= effective_stop_loss: return Signal(position.symbol, "SELL", 1.0, "torch_forecast: stop-loss hit", diagnostics) if price >= position.take_profit: return Signal(position.symbol, "SELL", 0.96, "torch_forecast: take-profit hit", diagnostics) if atr_trailing_stop is not None and price <= atr_trailing_stop: return Signal(position.symbol, "SELL", 0.94, "torch_forecast: ATR trailing stop hit", diagnostics) if not _is_torch_forecast(forecast): if rebound_fallback_position: hold_seconds = (utc_now() - position.opened_at).total_seconds() diagnostics["hold_seconds"] = hold_seconds if hold_seconds < settings.min_hold_seconds: return Signal( position.symbol, "HOLD", 0.45, "torch_forecast: rebound fallback minimum hold", diagnostics, ) return Signal( position.symbol, "HOLD", 0.42, "torch_forecast: rebound fallback hold without PyTorch model", diagnostics, ) return Signal(position.symbol, "SELL", 0.78, "torch_forecast: no valid PyTorch forecast to hold", diagnostics) if bool(forecast.get("block_entry", False)) or expected_return <= 0.0 or probability_up <= 0.50: return Signal( position.symbol, "SELL", 0.86, ( "torch_forecast: PyTorch forecast turned negative; " f"p_up={probability_up:.3f}, expected={expected_return:.4f}%" ), diagnostics, ) weak_hold = expected_return < min_edge or probability_up < min_probability or skill <= 0.0 if weak_hold and estimated_exit_net_percent >= 0: return Signal( position.symbol, "SELL", 0.74, ( "torch_forecast: PyTorch no longer confirms enough edge; " f"p_up={probability_up:.3f}, expected={expected_return:.4f}%" ), diagnostics, ) return Signal(position.symbol, "HOLD", 0.35, "torch_forecast: PyTorch hold confirmed", diagnostics) def _is_torch_forecast(forecast: dict) -> bool: model = str(forecast.get("model", "")).strip().lower() return bool(forecast.get("usable", False)) and model in {"torch_lstm", "torch_gru"} def _missing_torch_model(forecast: dict) -> bool: model = str(forecast.get("model", "")).strip().lower() reason = str(forecast.get("reason", "")).lower() return ( not bool(forecast.get("usable", False)) and model in {"", "none"} and "no valid pytorch" in reason ) def _torch_min_probability(settings: Settings) -> float: return round(_clamp(settings.time_series_min_probability_up, 0.45, 0.75), 4) def _dynamic_symbol_position_limit(settings: Settings) -> int: exposure_based_limit = int(settings.max_symbol_exposure_usdt // max(settings.min_position_usdt, 0.01)) return max(1, settings.max_positions_per_symbol, exposure_based_limit) def _torch_forecast_confidence(settings: Settings, forecast: dict) -> float: expected_return = max(0.0, _safe_float(forecast.get("expected_return_percent"), 0.0)) probability_up = _safe_float(forecast.get("probability_up"), 0.5) skill = max(0.0, _safe_float(forecast.get("skill"), 0.0)) min_edge = max(0.01, settings.time_series_min_edge_percent) edge_strength = _clamp(expected_return / max(min_edge * 4.0, 0.01), 0.0, 1.0) probability_strength = _clamp((probability_up - 0.50) / 0.25, 0.0, 1.0) skill_strength = _clamp(skill / 0.35, 0.0, 1.0) confidence = 0.45 + probability_strength * 0.30 + edge_strength * 0.20 + skill_strength * 0.10 return round(_clamp(confidence, 0.0, 0.96), 4) def _torch_forecast_position_sizing( settings: Settings, account: dict | None, stop_loss_percent: float, forecast: dict, ) -> dict[str, float | str]: base = _trend_position_sizing(settings, account, stop_loss_percent) base_notional = float(base["notional_usdt"]) if base_notional <= 0: notional = 0.0 edge_multiplier = probability_multiplier = skill_multiplier = 0.0 else: expected_return = max(0.0, _safe_float(forecast.get("expected_return_percent"), 0.0)) probability_up = _safe_float(forecast.get("probability_up"), 0.5) skill = max(0.0, _safe_float(forecast.get("skill"), 0.0)) min_edge = max(0.01, settings.time_series_min_edge_percent) edge_multiplier = _clamp(expected_return / max(min_edge * 3.0, 0.01), 0.25, 1.15) probability_multiplier = _clamp(0.75 + (probability_up - 0.55) * 3.0, 0.50, 1.20) skill_multiplier = _clamp(0.85 + skill * 0.60, 0.60, 1.15) raw = base_notional * edge_multiplier * probability_multiplier * skill_multiplier notional = 0.0 if raw < settings.min_position_usdt else min(raw, settings.max_position_usdt) return { **base, "method": "torch_forecast_risk", "notional_usdt": round(notional, 2), "base_notional_usdt": base["notional_usdt"], "torch_edge_multiplier": round(edge_multiplier, 4), "torch_probability_multiplier": round(probability_multiplier, 4), "torch_skill_multiplier": round(skill_multiplier, 4), } def _has_trend_entry_indicators(current: Candle, previous: Candle, trend: Candle) -> bool: return all( value is not None for value in ( current.ema_50, current.rsi_14, current.atr_14, current.macd, current.macd_signal, previous.macd, previous.macd_signal, trend.ema_50, trend.ema_200, ) ) def _macd_crossed_up(previous: Candle, current: Candle) -> bool: if None in (previous.macd, previous.macd_signal, current.macd, current.macd_signal): return False return bool(previous.macd <= previous.macd_signal and current.macd > current.macd_signal) def _macd_crossed_down(previous: Candle, current: Candle) -> bool: if None in (previous.macd, previous.macd_signal, current.macd, current.macd_signal): return False return bool(previous.macd >= previous.macd_signal and current.macd < current.macd_signal) def _trend_position_sizing( settings: Settings, account: dict | None, stop_loss_percent: float, ) -> dict[str, float | str]: equity = _safe_float((account or {}).get("equity"), settings.starting_balance_usdt) if equity <= 0: equity = settings.starting_balance_usdt risk_fraction = _clamp(settings.risk_per_trade_percent, 0.0, 0.01) guard_multiplier = _risk_guard_multiplier(account) risk_fraction *= guard_multiplier risk_usdt = equity * risk_fraction raw_notional = risk_usdt / max(stop_loss_percent, 0.0001) high = max(0.0, settings.max_position_usdt) low = max(0.0, settings.min_position_usdt) notional = 0.0 if raw_notional < low else min(raw_notional, high) return { "method": "fixed_fractional_risk", "risk_per_trade_percent": round(risk_fraction * 100, 4), "risk_guard_multiplier": round(guard_multiplier, 4), "risk_usdt": round(risk_usdt, 4), "stop_loss_percent": round(stop_loss_percent * 100, 4), "raw_notional_usdt": round(raw_notional, 4), "notional_usdt": round(notional, 2), "equity_usdt": round(equity, 2), } def _decision_suffix(pattern: dict, learning: dict, llm: dict | None = None) -> str: parts: list[str] = [] label = pattern.get("label") if label: parts.append(f"шаблон: {label}") reason = learning.get("reason") adjustment = float(learning.get("confidence_adjustment", 0.0) or 0.0) if reason and adjustment != 0: parts.append(f"обучение: {reason}") llm = llm or {} llm_reason = llm.get("reason_ru") llm_adjustment = float(llm.get("confidence_adjustment", 0.0) or 0.0) if llm_reason and (llm_adjustment != 0 or llm.get("block_entry")): parts.append(f"LLM: {llm_reason}") return " (" + "; ".join(parts) + ")" if parts else "" def _clamp(value: float, low: float, high: float) -> float: return max(low, min(high, value)) def _position_sizing( *, settings: Settings, final_score: float, grid_active: bool, rebound_active: bool, forecast: dict | None = None, adaptive: dict | None = None, account: dict | None = None, ) -> dict[str, float | bool | str]: low = max(0.0, settings.min_position_usdt) high = max(low, settings.max_position_usdt) if grid_active: high = max(low, min(high, settings.grid_max_position_usdt)) elif rebound_active: high = max(low, min(high, settings.rebound_max_position_usdt)) denominator = max(0.0001, 1.0 - settings.min_signal_confidence) confidence_ratio = _clamp((final_score - settings.min_signal_confidence) / denominator, 0.0, 1.0) confidence_notional = low + (high - low) * confidence_ratio risk_multiplier = _position_risk_multiplier(forecast, adaptive) * _risk_guard_multiplier(account) method = "confidence" raw = confidence_notional kelly = _kelly_position( settings=settings, final_score=final_score, forecast=forecast or {}, adaptive=adaptive or {}, account=account, ) if settings.kelly_sizing_enabled: method = "fractional_kelly" raw = float(kelly["kelly_notional_usdt"]) raw *= risk_multiplier notional = round(_clamp(raw, low, high), 2) return { "method": method, "enabled": bool(settings.kelly_sizing_enabled), "notional_usdt": notional, "confidence_notional_usdt": round(confidence_notional, 2), "risk_multiplier": round(risk_multiplier, 4), "low_cap_usdt": round(low, 2), "high_cap_usdt": round(high, 2), **kelly, } def _position_risk_multiplier(forecast: dict | None, adaptive: dict | None) -> float: multiplier = 1.0 forecast = forecast or {} if forecast.get("usable"): probability_up = _safe_float(forecast.get("probability_up"), 0.5) volatility_percent = _safe_float(forecast.get("volatility_percent"), 0.0) if probability_up < 0.52: multiplier *= 0.75 elif probability_up >= 0.60: multiplier *= 1.08 if volatility_percent >= 0.8: multiplier *= 0.70 learning_multiplier = _safe_float((adaptive or {}).get("effective_position_size_multiplier"), 1.0) multiplier *= _clamp(learning_multiplier, 0.25, 2.0) return multiplier def _risk_guard_multiplier(account: dict | None) -> float: guard = (account or {}).get("risk_guard") if not isinstance(guard, dict): return 1.0 try: value = float(guard.get("position_size_multiplier", 1.0)) except (TypeError, ValueError): value = 1.0 return _clamp(value, 0.0, 1.0) def _kelly_position( *, settings: Settings, final_score: float, forecast: dict, adaptive: dict, account: dict | None, ) -> dict[str, float | bool | str]: confidence_probability = _confidence_probability(final_score, settings.min_signal_confidence) probability_source = "confidence" probability = confidence_probability if forecast.get("usable"): probability = _safe_float(forecast.get("probability_up"), confidence_probability) probability_source = "forecast" probability = _clamp(probability, 0.0, 1.0) stop_loss = _adaptive_percent(adaptive, "stop_loss_percent", settings.stop_loss_percent, 0.003, 0.08) take_profit = _adaptive_percent(adaptive, "take_profit_percent", settings.take_profit_percent, 0.003, 0.20) round_trip_cost = max(0.0, 2.0 * (settings.taker_fee_rate + settings.slippage_rate)) win_return = max(0.0, take_profit - round_trip_cost) loss_return = max(0.0001, stop_loss + round_trip_cost) reward_loss_ratio = win_return / loss_return if loss_return > 0 else 0.0 full_kelly = probability - ((1.0 - probability) / reward_loss_ratio) if reward_loss_ratio > 0 else 0.0 full_kelly = max(0.0, full_kelly) fractional_kelly = full_kelly * _clamp(settings.kelly_fraction, 0.0, 1.0) effective_fraction = _clamp(fractional_kelly, 0.0, _clamp(settings.kelly_max_fraction, 0.0, 1.0)) bankroll = _safe_float((account or {}).get("equity"), settings.starting_balance_usdt) if bankroll <= 0: bankroll = settings.starting_balance_usdt kelly_notional = max(0.0, bankroll * effective_fraction) return { "kelly_probability": round(probability, 4), "kelly_probability_source": probability_source, "kelly_reward_loss_ratio": round(reward_loss_ratio, 4), "kelly_full_fraction": round(full_kelly, 4), "kelly_fractional_fraction": round(fractional_kelly, 4), "kelly_effective_fraction": round(effective_fraction, 4), "kelly_bankroll_usdt": round(bankroll, 2), "kelly_notional_usdt": round(kelly_notional, 2), } def _confidence_probability(final_score: float, min_signal_confidence: float) -> float: denominator = max(0.0001, 1.0 - min_signal_confidence) ratio = _clamp((final_score - min_signal_confidence) / denominator, 0.0, 1.0) return 0.50 + ratio * 0.18 def _grid_state( *, settings: Settings, latest: Candle, pattern: dict, llm: dict, atr_percent: float, spread_ok: bool, liquidity_ok: bool, volatility_ok: bool, ) -> dict: metrics = pattern.get("metrics") or {} high20 = _safe_float(metrics.get("high20"), latest.high) low20 = _safe_float(metrics.get("low20"), latest.low) width = max(0.0, high20 - low20) range_position = _clamp((latest.close - low20) / width, 0.0, 1.0) if width else 0.5 range_width_percent = (width / latest.close * 100) if latest.close else 0.0 label = str(pattern.get("label", "")).lower() tags = {str(tag).lower() for tag in pattern.get("tags", [])} llm_regime = str(llm.get("market_regime", "")).lower() llm_grid = bool(llm.get("grid_suitable", False)) ema_gap = abs(_safe_float(metrics.get("ema_gap_percent"), 999.0)) ret_20 = abs(_safe_float(metrics.get("ret_20_percent"), 999.0)) range_like = ( "боковик" in label or "боковик" in tags or llm_regime == "range" or llm_grid or (ema_gap <= 0.35 and ret_20 <= max(0.8, atr_percent * 1.2)) ) dangerous = ( label in NEGATIVE_LONG_PATTERNS or llm_regime in {"downtrend", "breakdown", "panic"} or bool(llm.get("block_entry", False)) ) active = bool( settings.grid_trading_enabled and range_like and not dangerous and spread_ok and liquidity_ok and volatility_ok and width > 0 ) buy_zone = bool(active and range_position <= _clamp(settings.grid_buy_zone, 0.05, 0.95)) reason = ( f"диапазон {range_position:.2f}, ширина {range_width_percent:.2f}%" if active else "условия grid-режима не подтверждены" ) return { "enabled": settings.grid_trading_enabled, "active": active, "buy_zone": buy_zone, "range_position": round(range_position, 4), "range_width_percent": round(range_width_percent, 4), "buy_zone_limit": round(_clamp(settings.grid_buy_zone, 0.05, 0.95), 4), "llm_grid_suitable": llm_grid, "range_like": range_like, "dangerous": dangerous, "reason": reason, } def _rebound_state( *, settings: Settings, candles: list[Candle], latest: Candle, previous: Candle, pattern: dict, llm: dict, spread_ok: bool, liquidity_ok: bool, volume_ok: bool, volatility_ok: bool, atr_percent: float, ) -> dict: metrics = pattern.get("metrics") or {} ret_3 = _safe_float( metrics.get("ret_3_percent"), _percent_change(latest.close, candles[-4].close) if len(candles) >= 4 else 0.0, ) ret_10 = _safe_float( metrics.get("ret_10_percent"), _percent_change(latest.close, candles[-11].close) if len(candles) >= 11 else 0.0, ) ret_20 = _safe_float( metrics.get("ret_20_percent"), _percent_change(latest.close, candles[-21].close) if len(candles) >= 21 else 0.0, ) label = str(pattern.get("label") or "").lower() tags = {str(tag).lower() for tag in pattern.get("tags", [])} llm_regime = str(llm.get("market_regime", "")).lower() rsi = _safe_float(latest.rsi_14, 50.0) previous_rsi = _safe_float(previous.rsi_14, rsi) volume_ratio = latest.volume / latest.volume_ma_20 if latest.volume_ma_20 and latest.volume_ma_20 > 0 else 0.0 body = abs(latest.close - latest.open) lower_wick = max(0.0, min(latest.open, latest.close) - latest.low) low6 = min(candle.low for candle in candles[-6:]) high6 = max(candle.high for candle in candles[-6:]) recent_lows = [candle.low for candle in candles[-6:-1]] no_new_low = bool(recent_lows) and latest.low >= min(recent_lows) * 0.999 bounce_from_low = ((latest.close - low6) / latest.close * 100) if latest.close else 0.0 range_width = max(high6 - low6, latest.close * 0.0001) range_position = _clamp((latest.close - low6) / range_width, 0.0, 1.0) recent_drop_depth = max(abs(min(ret_10, 0.0)), abs(min(ret_20, 0.0)) * 0.65) pattern_down = ( label in NEGATIVE_LONG_PATTERNS or any(tag in NEGATIVE_LONG_PATTERNS for tag in tags) or any(marker in label for marker in ("нисход", "пад", "пробой вниз", "ускор")) ) price_drop = ret_10 <= -max(0.35, atr_percent * 1.1) or ret_20 <= -max(0.6, atr_percent * 1.6) recent_drop = bool(price_drop and (pattern_down or ret_10 < 0 or ret_20 < 0)) body_base = max(body, latest.close * 0.0001) wick_absorption = lower_wick >= body_base * 0.6 bounced = bounce_from_low >= max(0.08, atr_percent * 0.3) or range_position >= 0.18 momentum_stabilized = latest.close >= previous.close or abs(ret_3) <= max(0.25, atr_percent * 0.8) or no_new_low rsi_zone = 24 <= rsi <= 52 rsi_improving = rsi >= previous_rsi or rsi <= 38 market_ok = spread_ok and liquidity_ok and volatility_ok continuing_collapse = bool( latest.close < previous.close and not no_new_low and ret_3 <= -max(0.6, atr_percent * 1.2) and rsi < 34 ) panic_regime = llm_regime in {"panic", "breakdown"} drop_score = _clamp(recent_drop_depth / max(0.45, atr_percent * 2.0), 0.0, 1.0) stabilization_score = 1.0 if latest.close >= previous.close else 0.75 if no_new_low else 0.55 if momentum_stabilized else 0.0 absorption_score = _clamp(max(lower_wick / (body_base * 1.4), bounce_from_low / max(0.08, atr_percent * 0.7)), 0.0, 1.0) rsi_score = 1.0 if rsi_zone and rsi_improving else 0.65 if rsi_zone else 0.0 volume_score = _clamp(volume_ratio / 1.2, 0.0, 1.0) market_score = 1.0 if market_ok else 0.0 probability = ( drop_score * 0.22 + stabilization_score * 0.24 + absorption_score * 0.20 + rsi_score * 0.18 + volume_score * 0.08 + market_score * 0.08 ) if continuing_collapse or panic_regime: probability = min(probability, 0.45) if not recent_drop: probability = min(probability, 0.50) if not market_ok: probability = min(probability, 0.55) min_probability = _clamp(settings.rebound_min_probability, 0.45, 0.9) active = bool( settings.rebound_trading_enabled and recent_drop and momentum_stabilized and (wick_absorption or bounced) and rsi_zone and market_ok and volume_ok and not continuing_collapse and not panic_regime and probability >= min_probability ) return { "enabled": settings.rebound_trading_enabled, "active": active, "probability": round(_clamp(probability, 0.0, 1.0), 4), "entry_score": round(_clamp(probability, 0.0, 1.0), 4) if active else 0.0, "min_probability": round(min_probability, 4), "recent_drop": recent_drop, "momentum_stabilized": momentum_stabilized, "wick_absorption": wick_absorption, "bounced_from_low": bounced, "rsi_zone": rsi_zone, "rsi_improving": rsi_improving, "market_ok": market_ok, "volume_ratio": round(volume_ratio, 4), "ret_3_percent": round(ret_3, 4), "ret_10_percent": round(ret_10, 4), "ret_20_percent": round(ret_20, 4), "bounce_from_low_percent": round(bounce_from_low, 4), "range_position_6": round(range_position, 4), "continuing_collapse": continuing_collapse, "panic_regime": panic_regime, "reason": ( "падение замедлилось, есть признаки короткого отскока" if active else "rebound-сигнал не подтвержден" ), } def _safe_float(value: object, default: float = 0.0) -> float: try: return float(value) except (TypeError, ValueError): return default def _percent_change(current: float, previous: float) -> float: return ((current - previous) / previous * 100) if previous else 0.0 def _adaptive_rules(learning: dict | None) -> dict: learning = learning or {} rules = learning.get("adaptive_rules", learning) return dict(rules) if isinstance(rules, dict) else {} def _adaptive_threshold_adjustment(adaptive: dict) -> float: raw = adaptive.get("effective_entry_threshold_adjustment", adaptive.get("entry_threshold_adjustment", 0.0)) return _clamp(_safe_float(raw, 0.0), -0.18, 0.18) def _adaptive_blocks_entry(adaptive: dict, falling_market: bool = False, rebound_confirmed: bool = False) -> bool: if adaptive.get("allow_new_entries") is False: return True if adaptive.get("over_target_exposure"): return True if adaptive.get("symbol_blocked") or adaptive.get("pattern_blocked"): return True if adaptive.get("bad_market_entry_block") and falling_market and not rebound_confirmed: return True return False def _adaptive_block_reason(adaptive: dict, falling_market: bool = False, rebound_confirmed: bool = False) -> str: if adaptive.get("allow_new_entries") is False: return "новые входы выключены режимом обучения" if adaptive.get("over_target_exposure"): return "экспозиция выше цели обучения" if adaptive.get("symbol_blocked"): return "символ в стоп-листе обучения" if adaptive.get("pattern_blocked"): return "шаблон в стоп-листе обучения" if adaptive.get("bad_market_entry_block") and falling_market and not rebound_confirmed: return "падающий рынок, добор запрещен" return "адаптивное правило" def _falling_market(latest: Candle, previous: Candle, pattern_label: str, llm: dict) -> bool: label = pattern_label.lower() llm_regime = str(llm.get("market_regime", "")).lower() ema_down = ( latest.ema_20 is not None and latest.ema_50 is not None and latest.close < latest.ema_50 and latest.ema_20 < latest.ema_50 ) momentum_down = latest.close < previous.close and (latest.rsi_14 is None or latest.rsi_14 < 50) pattern_down = any(marker in label for marker in ("нисход", "пад", "пробой вниз", "ускор")) llm_down = llm_regime in {"downtrend", "breakdown", "panic"} return bool(ema_down or (momentum_down and pattern_down) or llm_down) def _adaptive_percent(adaptive: dict, key: str, default: float, low: float, high: float) -> float: return _clamp(_safe_float(adaptive.get(key), default), low, high) def _estimated_exit_net_percent(position: Position, price: float, settings: Settings) -> float: if position.entry_price <= 0: return 0.0 gross_percent = ((price - position.entry_price) / position.entry_price) * 100 round_trip_cost_percent = (settings.taker_fee_rate * 2 + settings.slippage_rate * 2) * 100 return gross_percent - round_trip_cost_percent def _adaptive_indicator_exit_allowed(adaptive: dict, mode_key: str, estimated_exit_net_percent: float) -> bool: mode = str(adaptive.get(mode_key, "normal")).lower() if mode != "profit_only": return True min_exit_profit = _safe_float(adaptive.get("min_exit_profit_percent"), 0.0) return estimated_exit_net_percent >= min_exit_profit def _forecast_exit_signal( *, forecast: dict, position: Position, price: float, estimated_exit_net_percent: float, stop_loss_percent: float, min_edge_percent: float, ) -> tuple[str, float, str] | None: if not forecast.get("usable"): return None skill = _safe_float(forecast.get("skill"), 0.0) expected_return = _safe_float(forecast.get("expected_return_percent"), 0.0) probability_up = _safe_float(forecast.get("probability_up"), 0.5) min_edge = max(0.0, min_edge_percent) strong_negative = skill > 0.02 and expected_return <= -max(min_edge, 0.03) and probability_up <= 0.44 if not strong_negative: return None reason = forecast.get("reason") or "ожидается снижение" if estimated_exit_net_percent >= 0: return "SELL", 0.82, f"прогноз временного ряда ухудшился: {reason}; фиксируем результат" loss_from_entry = ((price - position.entry_price) / position.entry_price) if position.entry_price else 0.0 soft_loss_limit = -max(0.003, stop_loss_percent * 0.35) if loss_from_entry <= soft_loss_limit: return "SELL", 0.84, f"прогноз временного ряда ухудшился: {reason}; ограничиваем убыток до stop-loss" return None def _learning_blocks_entry( *, learning: dict, learning_adjustment: float, min_samples: int, max_adjustment: float, enabled: bool, ) -> bool: if not enabled: return False sample_size = int(learning.get("sample_size", 0) or 0) net_pnl = float(learning.get("net_pnl", 0.0) or 0.0) win_rate = float(learning.get("win_rate", 0.0) or 0.0) strong_negative_adjustment = -max(0.06, max_adjustment * 0.65) return ( sample_size >= min_samples and net_pnl < 0 and win_rate <= 0.25 and learning_adjustment <= strong_negative_adjustment )