breakout-strategy-engine

# Breakout Strategy Engine ## Pre-Built Breakout Strategies with Confirmation Filters ```python import pandas as pd import numpy as np from dataclasses import dataclass from typing import Optional @dataclass class BreakoutSignal: symbol: str direction: str # "long" or "short" entry: float stop_loss: float target: float strategy: str confirmation: list[str] strength: float # 0-1 class BreakoutEngine: # ═══════════════════════════════════════ # 1. BOLLINGER SQUEEZE BREAKOUT # ═══════════════════════════════════════ @staticmethod def bollinger_squeeze(df: pd.DataFrame, bb_period: int = 20, kc_period: int = 20, kc_mult: float = 1.5) -> dict: """Bollinger inside Keltner Channel = squeeze. Breakout when squeeze releases.""" close = df["close"] bb_mid = close.rolling(bb_period).mean() bb_std = close.rolling(bb_period).std() bb_upper = bb_mid + 2 * bb_std bb_lower = bb_mid - 2 * bb_std atr = ((df["high"] - df["low"]).rolling(kc_period).mean()) kc_upper = bb_mid + kc_mult * atr kc_lower = bb_mid - kc_mult * atr squeeze_on = (bb_lower > kc_lower) & (bb_upper < kc_upper) squeeze_off = ~squeeze_on # Squeeze just released squeeze_fire = squeeze_off & squeeze_on.shift(1) # Direction from momentum momentum = close - close.rolling(bb_period).mean() direction = np.where(momentum > 0, "long", "short") df_out = df.copy() df_out["squeeze_on"] = squeeze_on df_out["squeeze_fire"] = squeeze_fire df_out["direction"] = direction df_out["bb_width"] = (bb_upper - bb_lower) / bb_mid * 100 current = df_out.iloc[-1] return { "strategy": "bollinger_squeeze", "squeeze_active": bool(current["squeeze_on"]), "squeeze_firing": bool(current["squeeze_fire"]), "direction": current["direction"], "bb_width": round(current["bb_width"], 3), "bars_in_squeeze": int(squeeze_on.iloc[-20:].sum()), "signal": "BREAKOUT FIRING" if current["squeeze_fire"] else "SQUEEZE BUILDING" if current["squeeze_on"] else "NO SQUEEZE", } # ═══════════════════════════════════════ # 2. RANGE BREAKOUT (Donchian) # ═══════════════════════════════════════ @staticmethod def donchian_breakout(df: pd.DataFrame, period: int = 20, atr_mult: float = 1.5) -> dict: """Break above/below N-period high/low with ATR confirmation.""" high_n = df["high"].rolling(period).max().shift(1) low_n = df["low"].rolling(period).min().shift(1) atr_val = ((df["high"] - df["low"]).rolling(14).mean()) close = df["close"] long_break = close > high_n short_break = close < low_n # Volume confirmation vol_confirm = df["volume"] > df["volume"].rolling(20).mean() * 1.5 current = df.iloc[-1] return { "strategy": "donchian_breakout", "upper_channel": round(high_n.iloc[-1], 5), "lower_channel": round(low_n.iloc[-1], 5), "current_price": round(current["close"], 5), "long_breakout": bool(long_break.iloc[-1]), "short_breakout": bool(short_break.iloc[-1]), "volume_confirmed": bool(vol_confirm.iloc[-1]), "atr": round(atr_val.iloc[-1], 5), "stop_long": round(high_n.iloc[-1] - atr_mult * atr_val.iloc[-1], 5), "stop_short": round(low_n.iloc[-1] + atr_mult * atr_val.iloc[-1], 5), } # ═══════════════════════════════════════ # 3. MOMENTUM BREAKOUT # ═══════════════════════════════════════ @staticmethod def momentum_breakout(df: pd.DataFrame) -> dict: """Multi-filter momentum breakout: ADX + volume + close above/below structure.""" close = df["close"] atr = (df["high"] - df["low"]).rolling(14).mean() # ADX proxy plus_dm = df["high"].diff().clip(lower=0).rolling(14).mean() minus_dm = (-df["low"].diff()).clip(lower=0).rolling(14).mean() dx = abs(plus_dm - minus_dm) / (plus_dm + minus_dm + 1e-10) * 100 adx = dx.rolling(14).mean() # Momentum mom_10 = close.pct_change(10) vol_ratio = df["volume"] / df["volume"].rolling(20).mean() # Structure break high_20 = df["high"].rolling(20).max() low_20 = df["low"].rolling(20).min() current = df.iloc[-1] filters = [] if adx.iloc[-1] > 25: filters.append("ADX>25 (trending)") if vol_ratio.iloc[-1] > 1.5: filters.append("Volume 1.5x avg") if current["close"] > high_20.iloc[-2]: filters.append("New 20-bar high") if current["close"] < low_20.iloc[-2]: filters.append("New 20-bar low") if abs(mom_10.iloc[-1]) > 0.01: filters.append("Strong 10-bar momentum") direction = "long" if mom_10.iloc[-1] > 0 else "short" return { "strategy": "momentum_breakout", "direction": direction, "adx": round(adx.iloc[-1], 1), "momentum_10": round(mom_10.iloc[-1] * 100, 2), "volume_ratio": round(vol_ratio.iloc[-1], 2), "confirmations": filters, "n_confirmations": len(filters), "signal_quality": "A+" if len(filters) >= 4 else "A" if len(filters) >= 3 else "B" if len(filters) >= 2 else "C", "atr_stop": round(atr.iloc[-1] * 2, 5), } # ═══════════════════════════════════════ # FALSE BREAKOUT FILTER # ═══════════════════════════════════════ @staticmethod def false_breakout_probability(df: pd.DataFrame, lookback: int = 100) -> dict: """Historical false breakout rate for current pair to calibrate expectations.""" high_n = df["high"].rolling(20).max().shift(1) low_n = df["low"].rolling(20).min().shift(1) breakouts = (df["close"] > high_n) | (df["close"] < low_n) # A breakout is false if price returns inside range within 5 bars false_count = 0 total = 0 for i in range(20, len(df) - 5): if breakouts.iloc[i]: total += 1 future = df.iloc[i+1:i+6] mid = (high_n.iloc[i] + low_n.iloc[i]) / 2 if (future["close"] < high_n.iloc[i]).any() and (future["close"] > low_n.iloc[i]).any(): false_count += 1 rate = false_count / max(total, 1) return { "false_breakout_rate": round(rate * 100, 1), "total_breakouts": total, "recommendation": "Wait for retest" if rate > 0.5 else "Trade breakout with confirmation", } @staticmethod def scan_all(df: pd.DataFrame, symbol: str = "") -> dict: return { "symbol": symbol, "squeeze": BreakoutEngine.bollinger_squeeze(df), "donchian": BreakoutEngine.donchian_breakout(df), "momentum": BreakoutEngine.momentum_breakout(df), "false_breakout_rate": BreakoutEngine.false_breakout_probability(df), } ``` --- ## Automated Breakout Detection with Liquidity Sweep (CodeTrading Python) > Source: "Automated Break Out Detection in Python" by CodeTrading (Nov 2025) ### Why Simple Breakouts Fail A candle closing above a recent high is NOT enough — price often reverts back (fake breakout). A stronger breakout pattern is preceded by a **liquidity sweep**. ### The Liquidity-Confirmed Breakout Pattern **Bullish breakout with sweep:** 1. Identify pivot highs and pivot lows within a lookback window (e.g., 40 candles) 2. Find the most recent pivot high (resistance level) 3. Check if a pivot low formed AFTER that high AND is **lower than all previous pivot lows** in the window = liquidity sweep 4. Current candle closes above the pivot high AND previous candle was below it = **breakout confirmed** 5. Filter: only take bullish breakouts when last 15 candles are ALL above EMA (trend confirmation) **Bearish breakout with sweep:** Mirror logic — pivot low → higher high sweep → breakdown below pivot low ### Python Implementation Details **Pivot Detection (no look-ahead bias):** ```python def mark_pivots(df, window=7, high_col='high', low_col='low'): # Compare candle high to W neighbors left AND right # Pivot high: higher than all W neighbors # Pivot low: lower than all W neighbors # CRITICAL: stop at current_candle - window + 1 to avoid future data ``` **Breakout Detection Function:** ```python def detect_breakout(candle_idx, back_candles=40, window=5): # 1. Find pivots in [candle_idx - back_candles : candle_idx - window + 1] # 2. Find last pivot high value # 3. Check if close[candle_idx] > pivot_high AND close[candle_idx-1] < pivot_high # 4. Find pivot low between last_pivot_high and breakout candle # 5. Confirm pivot_low < min(all previous pivot lows) = SWEEP # Return: 2 = bullish breakout, 1 = bearish breakout, 0 = no signal ``` **Trend Filter (EMA-based):** - Check if last N candles (e.g., 15) are ALL above EMA → uptrend → only bullish breakouts - ALL below EMA → downtrend → only bearish breakouts - Adjustable selectivity: 10, 15, 20, 25 candles ### Backtest Results (EURUSD, 20 years hourly data, 2003-2023) **Optimization surface (TP/SL ratio vs ATR multiplier):** - Best zone: TP/SL ratio 4-5x, ATR multiplier 4+ - Returns: 5-52% depending on parameters (no leverage, no commission) - Drawdown minimized at ATR mult ~1, TP/SL ~1 (but returns limited to 5-8%) ### Key Parameters ```python back_candles = 40 # Lookback window for pattern detection pivot_window = 5-7 # Neighbors for pivot detection ema_candles = 15 # Candles for trend confirmation atr_mult = 1-5 # Stop loss = ATR * multiplier tp_sl_ratio = 1-5 # Take profit / stop loss ratio ``` ### Usage Recommendation - Best as an **alert indicator** — bot detects pattern, human manages trade - Run simultaneously on multiple assets to multiply signal count without losing precision - Trade management (TP/SL optimization) is asset-specific — needs tuning per instrument See also: `smart-money-trap-detector` (fake breakout detection), `liquidity-order-flow-mapper`, `chart-pattern-scanner`