from __future__ import annotations from dataclasses import dataclass from statistics import mean from typing import Iterable, Mapping from config import BotConfig from near_miss import effective_required_drop_pct, leverage_for_coin @dataclass(frozen=True) class EntryDecision: enter: bool reason: str drop_pct: float = 0.0 leverage: float = 0.0 roe_drop_pct: float = 0.0 required_drop_pct: float = 0.0 @dataclass class PositionState: entry_ts: float high_px: float atr_sl_px: float trailing_active: bool = False be_active: bool = False @dataclass(frozen=True) class ExitDecision: close: bool reason: str = "" profit_pct: float = 0.0 roe_pct: float = 0.0 def _std(values: list[float]) -> float: if len(values) < 2: return 0.0 avg = mean(values) return (sum((value - avg) ** 2 for value in values) / len(values)) ** 0.5 def _prices_without_current(history: Iterable[tuple[float, float]], current_price: float) -> list[float]: prices = [price for _, price in history] if prices and prices[-1] == current_price: prices = prices[:-1] return prices def _sma(values: list[float], n: int) -> float: if not values: return 0.0 return mean(values[-min(n, len(values)):]) def _rsi(values: list[float], n: int = 14) -> float: if len(values) < 2: return 50.0 deltas = [values[i] - values[i - 1] for i in range(1, len(values))] window = deltas[-min(n, len(deltas)):] gains = [max(delta, 0.0) for delta in window] losses = [abs(min(delta, 0.0)) for delta in window] avg_gain = mean(gains) if gains else 0.0 avg_loss = mean(losses) if losses else 0.0 if avg_loss == 0: return 100.0 if avg_gain > 0 else 50.0 rs = avg_gain / avg_loss return 100.0 - (100.0 / (1.0 + rs)) def _bollinger(values: list[float], n: int = 20, width: float = 2.0) -> tuple[float, float, float]: window = values[-min(n, len(values)):] or [0.0] mid = mean(window) std = _std(window) return mid - width * std, mid, mid + width * std def _basic_entry_guards( *, coin: str, volume_24h: float, open_positions: set[str], cooldowns: Mapping[str, float], now_ts: float, cfg: BotConfig, ) -> EntryDecision | None: coin = coin.upper() if volume_24h < cfg.min_volume_24h: return EntryDecision(False, "low_volume") if coin in open_positions: return EntryDecision(False, "already_open") if coin in cfg.blacklist: return EntryDecision(False, "blacklist") if cfg.allowed_coins and coin not in cfg.allowed_coins: return EntryDecision(False, "not_allowed_coin") if coin in cooldowns and (now_ts - float(cooldowns[coin])) < (cfg.cooldown_minutes * 60): return EntryDecision(False, "cooldown") return None def should_enter_flash_crash( *, coin: str, history: Iterable[tuple[float, float]], current_price: float, volume_24h: float, baseline_price: float, open_positions: set[str], cooldowns: Mapping[str, float], now_ts: float, cfg: BotConfig, ) -> EntryDecision: coin = coin.upper() guard = _basic_entry_guards( coin=coin, volume_24h=volume_24h, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg, ) if guard is not None: return guard prices = [price for _, price in history] if not prices: return EntryDecision(False, "insufficient_history") high_price = max(prices) if high_price <= 0 or current_price <= 0: return EntryDecision(False, "bad_price") drop_pct = ((current_price - high_price) / high_price) * 100 leverage = leverage_for_coin(cfg, coin) roe_drop_pct = drop_pct * leverage required_drop_pct = effective_required_drop_pct(cfg, coin) if drop_pct > -required_drop_pct: return EntryDecision(False, "no_flash_crash", drop_pct, leverage, roe_drop_pct, required_drop_pct) if baseline_price > 0 and current_price > baseline_price: return EntryDecision(False, "above_baseline", drop_pct, leverage, roe_drop_pct, required_drop_pct) return EntryDecision(True, "flash_crash", drop_pct, leverage, roe_drop_pct, required_drop_pct) def should_enter_volatility_squeeze_breakout( *, coin: str, history: Iterable[tuple[float, float]], current_price: float, volume_24h: float, open_positions: set[str], cooldowns: Mapping[str, float], now_ts: float, cfg: BotConfig, ) -> EntryDecision: coin = coin.upper() guard = _basic_entry_guards( coin=coin, volume_24h=volume_24h, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg, ) if guard is not None: return guard prices = [price for _, price in history] if prices and prices[-1] == current_price: prices = prices[:-1] lookback = max(2, int(cfg.squeeze_lookback_ticks)) if len(prices) < lookback or current_price <= 0: return EntryDecision(False, "insufficient_history") squeeze_window = prices[-lookback:] basis = mean(squeeze_window) if basis <= 0: return EntryDecision(False, "bad_price") band_width_pct = (_std(squeeze_window) * 4.0 / basis) * 100.0 if band_width_pct > cfg.squeeze_max_band_width_pct: return EntryDecision(False, "not_compressed", band_width_pct, leverage_for_coin(cfg, coin), 0.0, cfg.squeeze_max_band_width_pct) recent_high = max(squeeze_window) breakout_pct = ((current_price - recent_high) / recent_high) * 100.0 if recent_high > 0 else 0.0 leverage = leverage_for_coin(cfg, coin) roe_breakout_pct = breakout_pct * leverage if breakout_pct < cfg.squeeze_breakout_pct: return EntryDecision(False, "no_breakout", breakout_pct, leverage, roe_breakout_pct, cfg.squeeze_breakout_pct) return EntryDecision(True, "squeeze_breakout", breakout_pct, leverage, roe_breakout_pct, cfg.squeeze_breakout_pct) def should_enter_trend_pullback_sma_vwap( *, coin: str, history: Iterable[tuple[float, float]], current_price: float, volume_24h: float, open_positions: set[str], cooldowns: Mapping[str, float], now_ts: float, cfg: BotConfig, ) -> EntryDecision: guard = _basic_entry_guards(coin=coin, volume_24h=volume_24h, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg) if guard is not None: return guard prices = _prices_without_current(history, current_price) + [current_price] if len(prices) < 8 or current_price <= 0: return EntryDecision(False, "insufficient_history") sma_fast = _sma(prices, 5) sma_slow = _sma(prices, min(12, len(prices))) vwap_proxy = _sma(prices, min(8, len(prices))) rsi = _rsi(prices, min(14, max(2, len(prices) - 1))) if sma_fast <= sma_slow: return EntryDecision(False, "trend_not_up") if not (cfg.trend_pullback_min_rsi <= rsi <= cfg.trend_pullback_max_rsi): return EntryDecision(False, "rsi_out_of_pullback_range", rsi, leverage_for_coin(cfg, coin), 0.0, cfg.trend_pullback_max_rsi) distance_pct = abs((current_price - vwap_proxy) / vwap_proxy) * 100.0 if vwap_proxy > 0 else 0.0 if distance_pct > cfg.trend_pullback_max_distance_pct: return EntryDecision(False, "too_far_from_vwap", distance_pct, leverage_for_coin(cfg, coin), 0.0, cfg.trend_pullback_max_distance_pct) return EntryDecision(True, "trend_pullback_sma_vwap", distance_pct, leverage_for_coin(cfg, coin), rsi, cfg.trend_pullback_max_distance_pct) def should_enter_bollinger_rsi_mean_reversion( *, coin: str, history: Iterable[tuple[float, float]], current_price: float, volume_24h: float, open_positions: set[str], cooldowns: Mapping[str, float], now_ts: float, cfg: BotConfig, ) -> EntryDecision: guard = _basic_entry_guards(coin=coin, volume_24h=volume_24h, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg) if guard is not None: return guard prior = _prices_without_current(history, current_price) prices = prior + [current_price] if len(prior) < 6 or current_price <= 0: return EntryDecision(False, "insufficient_history") lower, mid, _upper = _bollinger(prior, min(20, len(prior))) rsi = _rsi(prices, min(14, max(2, len(prices) - 1))) tolerance = lower * (cfg.mean_reversion_band_tolerance_pct / 100.0) if current_price > lower + tolerance: return EntryDecision(False, "not_near_lower_band", ((current_price - lower) / lower) * 100.0 if lower > 0 else 0.0, leverage_for_coin(cfg, coin), rsi, cfg.mean_reversion_band_tolerance_pct) if rsi > cfg.mean_reversion_max_rsi: return EntryDecision(False, "rsi_not_oversold", rsi, leverage_for_coin(cfg, coin), 0.0, cfg.mean_reversion_max_rsi) if mid > 0 and current_price < mid * 0.90: return EntryDecision(False, "falling_knife") return EntryDecision(True, "bollinger_rsi_mean_reversion", ((current_price - lower) / lower) * 100.0 if lower > 0 else 0.0, leverage_for_coin(cfg, coin), rsi, cfg.mean_reversion_max_rsi) def should_enter_confirmed_squeeze_breakout( *, coin: str, history: Iterable[tuple[float, float]], current_price: float, volume_24h: float, open_positions: set[str], cooldowns: Mapping[str, float], now_ts: float, cfg: BotConfig, ) -> EntryDecision: base = should_enter_volatility_squeeze_breakout(coin=coin, history=history, current_price=current_price, volume_24h=volume_24h, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg) if not base.enter: return base prices = _prices_without_current(history, current_price) + [current_price] sma_fast = _sma(prices, min(5, len(prices))) sma_slow = _sma(prices, min(12, len(prices))) if sma_fast < sma_slow: return EntryDecision(False, "squeeze_not_trend_confirmed", base.drop_pct, base.leverage, base.roe_drop_pct, base.required_drop_pct) return EntryDecision(True, "confirmed_squeeze_breakout", base.drop_pct, base.leverage, base.roe_drop_pct, base.required_drop_pct) def should_enter_hybrid_survival_squeeze( *, coin: str, history: Iterable[tuple[float, float]], current_price: float, volume_24h: float, baseline_price: float, open_positions: set[str], cooldowns: Mapping[str, float], now_ts: float, cfg: BotConfig, ) -> EntryDecision: """Hybrid paper entry: v60/v59 survival base plus confirmed squeeze alpha. The squeeze module is preferred when present. Flash-crash/survival entries remain allowed only after a short moving-average reclaim, which filters the paper-trade pattern where hard stops dominated falling-knife entries. """ squeeze = should_enter_confirmed_squeeze_breakout( coin=coin, history=history, current_price=current_price, volume_24h=volume_24h, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg, ) if squeeze.enter: return EntryDecision(True, "hybrid_squeeze_breakout", squeeze.drop_pct, squeeze.leverage, squeeze.roe_drop_pct, squeeze.required_drop_pct) if squeeze.reason in {"low_volume", "already_open", "blacklist", "not_allowed_coin", "cooldown", "bad_price"}: return squeeze flash = should_enter_flash_crash( coin=coin, history=history, current_price=current_price, volume_24h=volume_24h, baseline_price=baseline_price, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg, ) if not flash.enter: return flash prices = _prices_without_current(history, current_price) + [current_price] sma_fast = _sma(prices, min(5, len(prices))) if current_price < sma_fast: return EntryDecision(False, "hybrid_flash_not_reclaiming_sma", flash.drop_pct, flash.leverage, flash.roe_drop_pct, flash.required_drop_pct) return EntryDecision(True, "hybrid_survival_rebound", flash.drop_pct, flash.leverage, flash.roe_drop_pct, flash.required_drop_pct) def should_enter_multi_day_trend_investment( *, coin: str, history: Iterable[tuple[float, float]], current_price: float, volume_24h: float, open_positions: set[str], cooldowns: Mapping[str, float], now_ts: float, cfg: BotConfig, ) -> EntryDecision: guard = _basic_entry_guards(coin=coin, volume_24h=volume_24h, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg) if guard is not None: return guard prices = _prices_without_current(history, current_price) + [current_price] if len(prices) < 10 or current_price <= 0: return EntryDecision(False, "insufficient_history") sma_fast = _sma(prices, min(5, len(prices))) sma_slow = _sma(prices, min(10, len(prices))) if sma_slow <= 0: return EntryDecision(False, "bad_price") trend_pct = ((sma_fast - sma_slow) / sma_slow) * 100.0 if trend_pct < cfg.multi_day_min_trend_pct: return EntryDecision(False, "multi_day_trend_too_weak", trend_pct, leverage_for_coin(cfg, coin), 0.0, cfg.multi_day_min_trend_pct) distance_pct = abs((current_price - sma_fast) / sma_fast) * 100.0 if sma_fast > 0 else 0.0 if distance_pct > cfg.multi_day_max_pullback_distance_pct: return EntryDecision(False, "multi_day_entry_too_extended", distance_pct, leverage_for_coin(cfg, coin), 0.0, cfg.multi_day_max_pullback_distance_pct) return EntryDecision(True, "multi_day_trend_investment", trend_pct, leverage_for_coin(cfg, coin), distance_pct, cfg.multi_day_min_trend_pct) def _momentum_pct(values: list[float], lookback: int) -> float | None: if len(values) < lookback + 1: return None reference = values[-lookback - 1] current = values[-1] if reference <= 0: return None return ((current - reference) / reference) * 100.0 def should_enter_relative_strength_rotation( *, coin: str, history: Iterable[tuple[float, float]], current_price: float, volume_24h: float, open_positions: set[str], cooldowns: Mapping[str, float], now_ts: float, cfg: BotConfig, market_histories: Mapping[str, Iterable[tuple[float, float]]] | None = None, ) -> EntryDecision: guard = _basic_entry_guards(coin=coin, volume_24h=volume_24h, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg) if guard is not None: return guard lookback = max(1, int(cfg.relative_strength_lookback_ticks)) histories = market_histories or {coin.upper(): history} scores: list[tuple[float, str]] = [] for candidate, candidate_history in histories.items(): values = [price for _ts, price in candidate_history] if candidate.upper() == coin.upper() and (not values or values[-1] != current_price): values = values + [current_price] score = _momentum_pct(values, lookback) if score is not None: scores.append((score, candidate.upper())) min_momentum = float(cfg.relative_strength_min_momentum_pct) eligible = [(score, candidate) for score, candidate in scores if score >= min_momentum] own_score = next((score for score, candidate in scores if candidate == coin.upper()), 0.0) if len(eligible) < int(cfg.relative_strength_min_positive_candidates): return EntryDecision(False, "market_breadth_too_thin", own_score, leverage_for_coin(cfg, coin), 0.0, min_momentum) eligible.sort(key=lambda item: (item[0], item[1]), reverse=True) top = {candidate for _score, candidate in eligible[: max(1, int(cfg.relative_strength_top_n))]} if coin.upper() not in top: return EntryDecision(False, "not_top_relative_strength", own_score, leverage_for_coin(cfg, coin), 0.0, min_momentum) return EntryDecision(True, "relative_strength_rotation", own_score, leverage_for_coin(cfg, coin), 0.0, min_momentum) def should_enter_by_strategy_family( *, coin: str, history: Iterable[tuple[float, float]], current_price: float, volume_24h: float, baseline_price: float = 0.0, open_positions: set[str], cooldowns: Mapping[str, float], now_ts: float, cfg: BotConfig, market_histories: Mapping[str, Iterable[tuple[float, float]]] | None = None, ) -> EntryDecision: if cfg.strategy_family == "hybrid_survival_squeeze": return should_enter_hybrid_survival_squeeze(coin=coin, history=history, current_price=current_price, volume_24h=volume_24h, baseline_price=baseline_price, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg) if cfg.strategy_family == "volatility_squeeze_breakout": return should_enter_volatility_squeeze_breakout(coin=coin, history=history, current_price=current_price, volume_24h=volume_24h, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg) if cfg.strategy_family == "confirmed_squeeze_breakout": return should_enter_confirmed_squeeze_breakout(coin=coin, history=history, current_price=current_price, volume_24h=volume_24h, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg) if cfg.strategy_family == "trend_pullback_sma_vwap": return should_enter_trend_pullback_sma_vwap(coin=coin, history=history, current_price=current_price, volume_24h=volume_24h, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg) if cfg.strategy_family == "bollinger_rsi_mean_reversion": return should_enter_bollinger_rsi_mean_reversion(coin=coin, history=history, current_price=current_price, volume_24h=volume_24h, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg) if cfg.strategy_family == "multi_day_trend_investment": return should_enter_multi_day_trend_investment(coin=coin, history=history, current_price=current_price, volume_24h=volume_24h, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg) if cfg.strategy_family == "relative_strength_rotation": return should_enter_relative_strength_rotation(coin=coin, history=history, current_price=current_price, volume_24h=volume_24h, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg, market_histories=market_histories) return should_enter_flash_crash(coin=coin, history=history, current_price=current_price, volume_24h=volume_24h, baseline_price=baseline_price, open_positions=open_positions, cooldowns=cooldowns, now_ts=now_ts, cfg=cfg) def exit_long_position(*, entry_px: float, current_px: float, leverage: float, state: PositionState, now_ts: float, cfg: BotConfig) -> ExitDecision: if entry_px <= 0 or current_px <= 0: return ExitDecision(False, "bad_price") state.high_px = max(state.high_px, current_px) profit_pct = ((current_px - entry_px) / entry_px) * 100 max_profit_pct = ((state.high_px - entry_px) / entry_px) * 100 roe_pct = profit_pct * leverage time_in_trade_mins = (now_ts - state.entry_ts) / 60 if not state.be_active and not state.trailing_active and max_profit_pct >= cfg.break_even_activation_pct: state.atr_sl_px = entry_px state.be_active = True if not state.trailing_active and max_profit_pct >= cfg.v_shape_activation_pct: state.trailing_active = True if state.trailing_active and profit_pct <= (max_profit_pct - cfg.v_shape_trail_dist_pct): return ExitDecision(True, f"V-Shape Trail Exit ({roe_pct:+.2f}% ROE)", profit_pct, roe_pct) if not state.trailing_active and time_in_trade_mins > cfg.dead_fish_time_limit_mins: return ExitDecision(True, f"Dead Fish Time-Stop (Nach {int(time_in_trade_mins)} Min ohne Rebound)", profit_pct, roe_pct) if not state.trailing_active and (current_px <= state.atr_sl_px or profit_pct <= -cfg.max_hard_stop_pct): return ExitDecision(True, f"SL Hit ({roe_pct:.2f}% ROE)", profit_pct, roe_pct) return ExitDecision(False, "hold", profit_pct, roe_pct)