""" Horizontal Resistance Breakout Strategy Implementation """ import backtrader as bt from indicators import ( ResistanceLevel, HigherLowsPattern, VolumeMultiplier, CandleBodyRatio, RetracementZone ) from utils import ( is_breakout_candle, is_valid_retracement, calculate_position_size, is_trend_favorable, TradeLogger ) import config class BreakoutStrategy(bt.Strategy): """ Horizontal Resistance Breakout Strategy Entry Conditions: 1. Price breaks above horizontal resistance on 2x volume 2. Price was making higher lows before breakout 3. Price retraces to resistance level on low volume 4. Small body candles during retracement 5. Price above 200 SMA """ params = ( ('resistance_lookback', config.RESISTANCE_LOOKBACK), ('resistance_tolerance', config.RESISTANCE_TOLERANCE), ('higher_lows_lookback', config.HIGHER_LOWS_LOOKBACK), ('min_higher_lows', config.MIN_HIGHER_LOWS), ('volume_lookback', config.VOLUME_LOOKBACK), ('volume_multiplier', config.VOLUME_MULTIPLIER), ('entry_tolerance', config.ENTRY_TOLERANCE), ('sma_period', config.SMA_PERIOD), ('small_body_ratio', config.SMALL_BODY_RATIO), ('low_volume_ratio', config.LOW_VOLUME_RATIO), ('risk_reward_ratio', config.RISK_REWARD_RATIO), ('position_risk', config.POSITION_SIZE), ('max_retracement_bars', config.MAX_RETRACEMENT_BARS), ('printlog', True), ) def __init__(self): # Initialize indicators self.resistance = ResistanceLevel( self.data, lookback=self.p.resistance_lookback, tolerance=self.p.resistance_tolerance ) self.higher_lows = HigherLowsPattern( self.data, lookback=self.p.higher_lows_lookback, min_lows=self.p.min_higher_lows ) self.volume_mult = VolumeMultiplier( self.data, period=self.p.volume_lookback ) self.body_ratio = CandleBodyRatio(self.data) self.retracement = RetracementZone( self.data, entry_tolerance=self.p.entry_tolerance ) self.sma = bt.indicators.SimpleMovingAverage( self.data.close, period=self.p.sma_period ) # Trade tracking self.trade_logger = TradeLogger() self.breakout_detected = False self.breakout_bar = None self.breakout_volume = None self.breakout_resistance = None self.retracement_low = None self.current_trade_id = None def log(self, txt, dt=None): """Logging function""" if self.p.printlog: dt = dt or self.datas[0].datetime.date(0) print(f'{dt.isoformat()} {txt}') def next(self): # Skip if we don't have enough data if len(self) < self.p.sma_period: return # Check if we're already in a position if self.position: self.manage_position() return # Look for breakout setup self.check_for_breakout() # Look for entry after breakout if self.breakout_detected: self.check_for_entry() def check_for_breakout(self): """Check if a valid breakout has occurred""" # Reset if breakout is too old if self.breakout_bar and len(self) - self.breakout_bar > self.p.max_retracement_bars: self.log(f'BREAKOUT RESET: Too old (>{self.p.max_retracement_bars} bars)') self.reset_breakout() return # Check all conditions for breakout if not self.breakout_detected: # Debug: Track each condition debug_info = [] # 1. Price must be above SMA trend_ok = is_trend_favorable(self.sma[0], self.data.close[0]) if not trend_ok: debug_info.append(f"Trend NOT favorable: Close={self.data.close[0]:.2f} vs SMA200={self.sma[0]:.2f}") if len(self) % 50 == 0: # Log every 50 bars to avoid spam self.log('; '.join(debug_info)) return else: debug_info.append(f"✓ Trend OK: Close={self.data.close[0]:.2f} > SMA200={self.sma[0]:.2f}") # 2. Must have resistance level resistance_ok = self.resistance.is_resistance[0] if not resistance_ok: debug_info.append(f"No resistance level detected") if len(self) % 50 == 0: self.log('; '.join(debug_info)) return else: debug_info.append(f"✓ Resistance found: {self.resistance.resistance[0]:.2f}") # 3. Must have higher lows pattern higher_lows_ok = self.higher_lows.pattern_active[0] if not higher_lows_ok: debug_info.append(f"Higher lows pattern not active (count: {self.higher_lows.higher_lows_count[0]})") if len(self) % 50 == 0: self.log('; '.join(debug_info)) return else: debug_info.append(f"✓ Higher lows active (count: {self.higher_lows.higher_lows_count[0]})") # 4. Check for breakout candle breakout_candle = is_breakout_candle(self.data, self.resistance.resistance[0]) if not breakout_candle: debug_info.append(f"No breakout candle: High={self.data.high[0]:.2f} vs Resistance={self.resistance.resistance[0]:.2f}") if len(self) % 50 == 0: self.log('; '.join(debug_info)) return else: debug_info.append(f"✓ Breakout candle detected") # 5. Volume must be 2x average volume_ok = self.volume_mult.volume_mult[0] >= self.p.volume_multiplier if not volume_ok: debug_info.append(f"Volume insufficient: {self.volume_mult.volume_mult[0]:.2f}x vs required {self.p.volume_multiplier}x") if len(self) % 25 == 0: # More frequent for volume issues self.log('; '.join(debug_info)) return else: debug_info.append(f"✓ Volume sufficient: {self.volume_mult.volume_mult[0]:.2f}x") # If we get here, all conditions are met! self.breakout_detected = True self.breakout_bar = len(self) self.breakout_volume = self.data.volume[0] self.breakout_resistance = self.resistance.resistance[0] self.log(f'🚀 BREAKOUT DETECTED! {"; ".join(debug_info)}') def check_for_entry(self): """Check if conditions are met for entry after breakout""" # Check if price has retraced to entry zone upper_bound = self.breakout_resistance * (1 + self.p.entry_tolerance) lower_bound = self.breakout_resistance * (1 - self.p.entry_tolerance) if lower_bound <= self.data.low[0] <= upper_bound: # Update retracement low if self.retracement_low is None: self.retracement_low = self.data.low[0] else: self.retracement_low = min(self.retracement_low, self.data.low[0]) # Check for valid retracement characteristics if is_valid_retracement( self.data, self.body_ratio[0], self.breakout_volume, self.p.small_body_ratio, self.p.low_volume_ratio ): # Calculate position size and enter trade self.enter_trade() def enter_trade(self): """Enter a new position""" # Calculate stop loss (below retracement low) stop_loss = self.retracement_low * 0.995 # 0.5% below retracement low # Calculate position size position_size = calculate_position_size( self.broker.getvalue(), self.p.position_risk, self.data.close[0], stop_loss ) if position_size > 0: # Calculate target risk = self.data.close[0] - stop_loss target = self.data.close[0] + (risk * self.p.risk_reward_ratio) # Place order self.buy(size=position_size) # Log trade self.current_trade_id = self.trade_logger.log_entry( self.datas[0].datetime.date(0), self.data.close[0], position_size, self.breakout_resistance, stop_loss, target ) self.log(f'BUY CREATED, Price: {self.data.close[0]:.2f}, ' f'Size: {position_size}, Stop: {stop_loss:.2f}, ' f'Target: {target:.2f}') # Reset breakout tracking self.reset_breakout() def manage_position(self): """Manage existing position""" if self.current_trade_id is None: return trade = self.trade_logger.trades[self.current_trade_id] # Check stop loss if self.data.low[0] <= trade['stop_loss']: self.close() self.trade_logger.log_exit( self.current_trade_id, self.datas[0].datetime.date(0), trade['stop_loss'], 'Stop Loss' ) self.log(f'STOP LOSS hit at {trade["stop_loss"]:.2f}') self.current_trade_id = None # Check target elif self.data.high[0] >= trade['target']: self.close() self.trade_logger.log_exit( self.current_trade_id, self.datas[0].datetime.date(0), trade['target'], 'Target' ) self.log(f'TARGET hit at {trade["target"]:.2f}') self.current_trade_id = None def reset_breakout(self): """Reset breakout tracking variables""" self.breakout_detected = False self.breakout_bar = None self.breakout_volume = None self.breakout_resistance = None self.retracement_low = None def notify_order(self, order): """Handle order notifications""" if order.status in [order.Submitted, order.Accepted]: return if order.status in [order.Completed]: if order.isbuy(): self.log(f'BUY EXECUTED, Price: {order.executed.price:.2f}, ' f'Size: {order.executed.size}') elif order.issell(): self.log(f'SELL EXECUTED, Price: {order.executed.price:.2f}, ' f'Size: {order.executed.size}') elif order.status in [order.Canceled, order.Margin, order.Rejected]: self.log('Order Canceled/Margin/Rejected') def notify_trade(self, trade): """Handle trade notifications""" if not trade.isclosed: return self.log(f'TRADE PROFIT, Gross: {trade.pnl:.2f}, Net: {trade.pnlcomm:.2f}') def stop(self): """Called when strategy stops""" if self.p.printlog: print('\n' + '='*60) print('STRATEGY PERFORMANCE SUMMARY') print('='*60) summary = self.trade_logger.get_summary() for key, value in summary.items(): print(f'{key}: {value}') print('='*60)