Overall Statistics
Total Orders
447
Average Win
0.04%
Average Loss
-0.03%
Compounding Annual Return
-0.738%
Drawdown
1.200%
Expectancy
-0.123
Start Equity
100000
End Equity
99190.23
Net Profit
-0.810%
Sharpe Ratio
-6.84
Sortino Ratio
-4.598
Probabilistic Sharpe Ratio
1.710%
Loss Rate
62%
Win Rate
38%
Profit-Loss Ratio
1.33
Alpha
-0.06
Beta
0.003
Annual Standard Deviation
0.009
Annual Variance
0
Information Ratio
-1.728
Tracking Error
0.105
Treynor Ratio
-20.351
Total Fees
$0.00
Estimated Strategy Capacity
$13000000.00
Lowest Capacity Asset
QQQ RIWIV7K5Z9LX
Portfolio Turnover
52.01%
from AlgorithmImports import *

class OpeningRangeBreakoutAlgorithm(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2024, 1, 1)
        self.SetEndDate(2025, 2, 4)
        self.SetCash(100000)
        
        # Define symbol and resolution
        self.symbol = self.AddEquity("QQQ", Resolution.Minute).Symbol
        self.set_brokerage_model(BrokerageName.ALPACA, AccountType.MARGIN)
        
        self.trailing_stop_percent = 0.0003  # 0.02% trailing stop
        self.retest_threshold = 0.05 # threshold for retest of breakout range
        self.range = 5 # opening range to use (in minutes)
        self.tail_size = 1.2 # size of the reject tail compared to candle body required (1 = same size, 2 = twice the size)

        # Signal tracking
        self.opening_high = 0
        self.opening_low = 0
        self.breakout_up = False
        self.breakout_down = False
        self.retest_up = False
        self.retest_down = False
        
        # Trade management
        self.entry_price = 0
        self.stop_loss = 0
        self.trailing_stop_price = 0
        self.open_position = False
        self.current_option = None
        self.rejects_up = 0
        self.rejects_down = 0
        self.retests_up = 0
        self.retests_down = 0
        self.breakouts_down = 0
        self.breakouts_up = 0
    
        self.schedule.on(self.date_rules.every_day(), 
                        self.time_rules.before_market_close(self.symbol),
                        self.ResetDailyValues)

    def ResetDailyValues(self):
        """Reset daily tracking variables"""
        self.opening_high = 0
        self.opening_low = float('inf')  # Use infinity so first price sets the low
        self.breakout_up = False
        self.breakout_down = False
        self.retest_up = False
        self.retest_down = False
        self.liquidate()
        self.open_position = False
        
        
            
    def HasLongTail(self, bar, breakout_type='up'):
        """Check if candle has a long rejection tail"""
        body = abs(bar.Close - bar.Open)
        if breakout_type == 'up':
            tail = bar.Low - min(bar.Open, bar.Close)
        else:
            tail = max(bar.Open, bar.Close) - bar.High
        return abs(tail) > (body * self.tail_size)
    
    def CheckRetest(self, current_bar, lows, highs, breakout_level, breakout_type='up'):
        """
        Check if retest conditions are met across 3 bars
        
        Parameters:
        current_bar: Current price bar
        lows: List of previous bars' lows [bar_minus_2_low, bar_minus_1_low]
        highs: List of previous bars' highs [bar_minus_2_high, bar_minus_1_high]
        breakout_level: Price level to test for breakout
        breakout_type: Direction of breakout ('up' or 'down')
        """
        if breakout_type == 'up':
            # Check if either of the previous bars stayed above breakout level
            prev_bars_condition = any(low > breakout_level for low in lows)
            
            # Current bar must come down to test the breakout level and close above it
            current_bar_condition = (
                current_bar.Low <= breakout_level + self.retest_threshold and 
                current_bar.Close > breakout_level
            )
            
            return prev_bars_condition and current_bar_condition
        
        elif breakout_type == 'down':
            
            prev_bars_condition = any(high < breakout_level for high in highs)

            # Current bar must move above breakout level and close below it (confirm rejection)
            current_bar_condition = (
                current_bar.High >= breakout_level - self.retest_threshold and  # Retest the level
                current_bar.Close < breakout_level  # Closes below
            )

            return prev_bars_condition and current_bar_condition

    
    def OnData(self, data):
        if not data.ContainsKey(self.symbol):
            return
        
        symbol_quantity = self.Portfolio[self.symbol].Quantity
        #self.Debug(f"Rejects up: {self.rejects_up}, Rejects down: {self.rejects_down}, Retests up: {self.retests_up}, Retests down: {self.retests_down}, Breaks up: {self.breakouts_up} Breaks down: {self.breakouts_down}")

        current_bar = data[self.symbol]
        if current_bar is None:
            self.Debug(f"No data for {self.symbol} at {self.Time}")
            return
            
        current_price = current_bar.Close

        # Update trailing stop if we have a position
        if symbol_quantity > 0:
            # If this is a new position, set initial trailing stop
            if self.trailing_stop_price == 0:
                self.trailing_stop_price = current_price * (1 - self.trailing_stop_percent)
            
            # Update trailing stop if price moves higher
            if current_price > self.trailing_stop_price / (1 - self.trailing_stop_percent):
                self.trailing_stop_price = current_price * (1 - self.trailing_stop_percent)
                self.Debug(f"Updated trailing stop to: {self.trailing_stop_price}")

            # Check if trailing stop is hit
            if current_price <= self.trailing_stop_price:
                self.Liquidate(self.symbol)
                self.open_position = False
                self.Debug(f"Trailing stop triggered at: {current_price}")
                self.Log(f"Trailing stop exit: {str(self.Time)}")
                
                self.trailing_stop_price = 0
                
        
        if symbol_quantity < 0:
            # Initialize trailing stop for new short position
            if self.trailing_stop_price == 0:
                self.trailing_stop_price = current_price * (1 + self.trailing_stop_percent)
            
            # Update trailing stop if price moves lower
            if current_price < self.trailing_stop_price / (1 + self.trailing_stop_percent):
                self.trailing_stop_price = current_price * (1 + self.trailing_stop_percent)
                self.Debug(f"Updated short trailing stop to: {self.trailing_stop_price}")

            # Check if trailing stop is hit
            if current_price >= self.trailing_stop_price:
                self.Liquidate(self.symbol)
                self.open_position = False
                self.Debug(f"Short trailing stop triggered at: {current_price}")
                self.Log(f"Short trailing stop exit: {str(self.Time)}")
                
                if self.current_option is not None:
                    self.Liquidate(self.current_option)
                    self.open_position = False
                    self.current_option = None
                
                self.trailing_stop_price = 0
                
        if self.open_position == True:
            #self.Debug("open position")
            return

        
        # Assuming self.range is X minutes for the opening range
        if self.time.hour == 14 and 30 <= self.time.minute < 30 + self.range:
            self.opening_high = max(self.opening_high, current_bar.High)
            self.opening_low = self.opening_low if self.opening_low > 0 else current_bar.Low

        
        # Signal generation after opening range
        if self.Time.hour > 14 or (self.Time.hour == 14 and self.Time.minute >= 30 + self.range):
            
            # Upside Breakout and Retest Detection
            if current_bar.Close > self.opening_high:
                self.breakout_up = True
                self.breakouts_up += 1
            
            if self.breakout_up and len(self.history(self.symbol, 3, Resolution.Minute)) >= 3:
                history_bars = self.History(self.symbol, 3, Resolution.Minute)
                bar_minus_2 = history_bars.iloc[0]  # Oldest bar
                bar_minus_1 = history_bars.iloc[1]  # Middle bar
                
                self.retest_up = self.CheckRetest(
                    current_bar,
                    [bar_minus_2['low'], bar_minus_1['low']],  # List of previous lows
                    [bar_minus_2['high'], bar_minus_1['high']],  # List of previous highs
                    self.opening_high,
                    'up'
                )
                if self.retest_up:
                    self.retests_up += 1
                
                # Signal conditions with rejection
                if self.retest_up and self.HasLongTail(current_bar, 'up'):
                    self.rejects_up += 1
                    self.market_order(self.symbol, 100)
                    self.open_position = True
                    # Place a trailing stop order for the long position
                    self.Debug(f"Long Entry with Retest Rejection: {current_bar.Close}")

            
            # Downside Breakout and Retest Detection
            if current_bar.Close < self.opening_low:
                self.breakout_down = True
                self.breakouts_down += 1
            
            if self.breakout_down and len(self.History(self.symbol, 3, Resolution.Minute)) >= 3:
                history_bars = self.History(self.symbol, 3, Resolution.Minute)
                bar_minus_2 = history_bars.iloc[0]  # Oldest bar
                bar_minus_1 = history_bars.iloc[1]  # Middle bar
                
                self.retest_down = self.CheckRetest(
                    current_bar,
                    [bar_minus_2['low'], bar_minus_1['low']],  
                    [bar_minus_2['high'], bar_minus_1['high']],  
                    self.opening_low,  
                    'down'
                )

                if self.retest_down:
                    self.retests_down += 1
                # Signal conditions with rejection
                if self.retest_down and self.HasLongTail(current_bar, 'down'):
                    self.rejects_down += 1
                    self.market_order(self.symbol, -100)
                    self.open_position = True
                    #self.BuyPutOption(data)
                    self.Debug(f"Short Entry with Retest Rejection: {current_bar.Close}")
                    
    
    
    def OnOrderEvent(self, orderEvent):
        if orderEvent.Status == OrderStatus.Filled:
            self.Debug(f"Order Filled: {orderEvent}")