from __future__ import annotations from dataclasses import asdict, dataclass, field from typing import Any from crypto_spot_bot.models import Candle, Ticker @dataclass(slots=True) class PatternResult: label: str score: float description: str tags: list[str] = field(default_factory=list) metrics: dict[str, Any] = field(default_factory=dict) def as_dict(self) -> dict[str, Any]: return asdict(self) class PatternAnalyzer: def analyze(self, candles: list[Candle], ticker: Ticker | None = None) -> PatternResult: if len(candles) < 30: return PatternResult( label="мало данных", score=0.0, description="Недостаточно свечей для анализа шаблонов.", tags=["мало данных"], ) latest = candles[-1] previous = candles[-2] high20 = max(candle.high for candle in candles[-20:]) low20 = min(candle.low for candle in candles[-20:]) close_3 = candles[-4].close if len(candles) >= 4 else candles[0].close close_10 = candles[-11].close if len(candles) >= 11 else candles[0].close close_20 = candles[-21].close if len(candles) >= 21 else candles[0].close ret_3 = _percent_change(latest.close, close_3) ret_10 = _percent_change(latest.close, close_10) ret_20 = _percent_change(latest.close, close_20) body = abs(latest.close - latest.open) lower_wick = max(0.0, min(latest.open, latest.close) - latest.low) upper_wick = max(0.0, latest.high - max(latest.open, latest.close)) atr_percent = (latest.atr_14 / latest.close * 100) if latest.atr_14 and latest.close else 0.0 volume_ratio = ( latest.volume / latest.volume_ma_20 if latest.volume_ma_20 and latest.volume_ma_20 > 0 else 0.0 ) ema_gap_percent = ( (latest.ema_50 - latest.ema_200) / latest.ema_200 * 100 if latest.ema_50 and latest.ema_200 else 0.0 ) spread_percent = ticker.spread_percent if ticker else 0.0 uptrend = bool( latest.ema_20 and latest.ema_50 and latest.ema_200 and latest.ema_20 >= latest.ema_50 >= latest.ema_200 and latest.close >= latest.ema_50 ) downtrend = bool( latest.ema_20 and latest.ema_50 and latest.ema_200 and latest.ema_20 <= latest.ema_50 <= latest.ema_200 and latest.close <= latest.ema_50 ) pullback = bool( latest.ema_20 and uptrend and latest.close <= latest.ema_20 * 1.012 and latest.rsi_14 is not None and 35 <= latest.rsi_14 <= 58 ) oversold_reversal = bool( latest.rsi_14 is not None and latest.rsi_14 <= 35 and latest.close > previous.close and lower_wick >= body * 1.2 ) stabilized_drop = _stabilized_drop( candles=candles, latest=latest, previous=previous, ret_3=ret_3, ret_10=ret_10, ret_20=ret_20, atr_percent=atr_percent, volume_ratio=volume_ratio, lower_wick=lower_wick, body=body, ) breakout = bool( latest.close >= high20 * 0.995 and volume_ratio >= 1.15 and latest.close > latest.open ) breakdown = bool( latest.close <= low20 * 1.005 and volume_ratio >= 1.1 and latest.close < latest.open ) fast_drop = bool(ret_3 <= -max(1.2, atr_percent * 1.8) or (latest.rsi_14 or 100) <= 25) range_market = bool(abs(ret_20) <= max(0.8, atr_percent * 1.2) and abs(ema_gap_percent) <= 0.35) volume_spike = volume_ratio >= 1.6 tags: list[str] = [] if uptrend: tags.append("восходящий тренд") if downtrend: tags.append("нисходящий тренд") if pullback: tags.append("откат к средней") if oversold_reversal: tags.append("перепроданность с разворотом") if stabilized_drop: tags.append("стабилизация после падения") if breakout: tags.append("пробой") if breakdown: tags.append("пробой вниз") if fast_drop: tags.append("ускоренное падение") if range_market: tags.append("боковик") if volume_spike: tags.append("объемный всплеск") label, score, description = _classify( pullback=pullback, oversold_reversal=oversold_reversal, stabilized_drop=stabilized_drop, breakout=breakout, breakdown=breakdown, fast_drop=fast_drop, range_market=range_market, uptrend=uptrend, downtrend=downtrend, ) metrics = { "ret_3_percent": ret_3, "ret_10_percent": ret_10, "ret_20_percent": ret_20, "atr_percent": atr_percent, "volume_ratio": volume_ratio, "ema_gap_percent": ema_gap_percent, "spread_percent": spread_percent, "rsi_14": latest.rsi_14, "high20": high20, "low20": low20, "body": body, "lower_wick": lower_wick, "upper_wick": upper_wick, "stabilized_drop": stabilized_drop, } return PatternResult( label=label, score=round(score, 4), description=description, tags=tags or ["нейтрально"], metrics=metrics, ) def _classify( *, pullback: bool, oversold_reversal: bool, stabilized_drop: bool, breakout: bool, breakdown: bool, fast_drop: bool, range_market: bool, uptrend: bool, downtrend: bool, ) -> tuple[str, float, str]: if fast_drop and breakdown: return "ускоренное падение", 0.18, "Цена быстро падает на повышенном объеме; входы ограничиваются." if breakdown: return "пробой вниз", 0.24, "Цена у нижней границы диапазона с давлением продавцов." if pullback: return "трендовый откат", 0.76, "Восходящий тренд сохраняется, цена откатилась к средней." if oversold_reversal: return "разворот после перепроданности", 0.68, "RSI низкий, но последняя свеча показывает попытку разворота." if stabilized_drop: return "стабилизация после падения", 0.58, "После снижения падение замедлилось; возможен короткий отскок." if breakout: return "пробой вверх", 0.72, "Цена обновляет верхнюю область диапазона с подтверждением объемом." if uptrend: return "восходящее продолжение", 0.64, "EMA и цена подтверждают восходящее продолжение." if range_market: return "боковик", 0.48, "Цена движется в диапазоне без сильного направления." if downtrend: return "нисходящий тренд", 0.28, "EMA и цена показывают нисходящее направление." return "нейтрально", 0.50, "Сильного шаблона входа не обнаружено." def _percent_change(current: float, previous: float) -> float: return ((current - previous) / previous * 100) if previous else 0.0 def _stabilized_drop( *, candles: list[Candle], latest: Candle, previous: Candle, ret_3: float, ret_10: float, ret_20: float, atr_percent: float, volume_ratio: float, lower_wick: float, body: float, ) -> bool: recent_drop = ret_10 <= -max(0.35, atr_percent * 1.1) or ret_20 <= -max(0.6, atr_percent * 1.6) if not recent_drop or latest.rsi_14 is None or latest.rsi_14 > 52: return False recent_lows = [candle.low for candle in candles[-5:-1]] no_new_low = bool(recent_lows) and latest.low >= min(recent_lows) * 0.999 bounce_from_low = ((latest.close - min(candle.low for candle in candles[-6:])) / latest.close * 100) if latest.close else 0.0 body_base = max(body, latest.close * 0.0001) absorption = lower_wick >= body_base * 0.6 or bounce_from_low >= max(0.08, atr_percent * 0.3) momentum_stabilized = latest.close >= previous.close or abs(ret_3) <= max(0.25, atr_percent * 0.8) or no_new_low volume_present = volume_ratio >= 0.55 continuing_drop = latest.close < previous.close and not no_new_low and ret_3 <= -max(0.6, atr_percent * 1.2) return bool(momentum_stabilized and absorption and volume_present and not continuing_drop)