Top Watchlist

from AlgorithmImports import *

class GapUpBreakoutAlgorithm(QCAlgorithm):
    def initialize(self) -> None:
        # 1. Basic settings
        self.set_start_date(2021, 1, 1)
        self.set_end_date(2022, 1, 1)
        self.set_cash(100000)
        self._max_candidates = 500  # We'll first narrow the universe to the top 500 daily volume
        self._top_gap_ups = 100      # Then pick the top 100 gap-ups

        # 2. Universe Settings
        self.universe_settings.resolution = Resolution.MINUTE
        self.add_universe(self.coarse_selection_function, self.fine_selection_function)

        # 3. Data tracking (reset daily)
        self._watchlist = {}        # symbol -> open price
        self._entry_prices = {}      # symbol -> entry price
        self._traded_symbols = set() # symbols that have already been traded/removed for the day
        self._previous_close = {}    # symbol -> previous close (from Coarse)

        # 4. Schedule daily events
        #    5 seconds after market open (9:30:05) to pick gap-ups
        #    5 seconds before market close (15:59:55) to liquidate positions & clear watchlist
        self.schedule.on(self.date_rules.every_day("SPY"),
                         self.time_rules.at(9, 30, 5),
                         self.pick_gap_ups)

        self.schedule.on(self.date_rules.every_day("SPY"),
                         self.time_rules.at(15, 59, 55),
                         self.close_positions_end_of_day)

        # Make sure data is adjusted for splits/dividends
        self.set_security_initializer(lambda s: s.set_data_normalization_mode(DataNormalizationMode.ADJUSTED))

    def coarse_selection_function(self, coarse: List[CoarseFundamental]) -> List[Symbol]:
        # Reset daily tracking lists
        self._watchlist.clear()
        self._entry_prices.clear()
        self._traded_symbols.clear()
        self._previous_close.clear()

        # Pre-filter: fundamental data, price > 1, volume > 500k
        filtered = [c for c in coarse if c.has_fundamental_data and c.price > 1 and c.volume > 500000]
        # Sort descending by dollar volume, take top 500
        filtered.sort(key=lambda x: x.dollar_volume, reverse=True)
        selected = filtered[:self._max_candidates]

        # Store previous close for usage in gap-up calculation later
        for c in selected:
            self._previous_close[c.symbol] = c.price

        return [s.symbol for s in selected]

    def fine_selection_function(self, fine: List[FineFundamental]) -> List[Symbol]:
        # Keep the entire set of coarse-selected symbols
        return [f.symbol for f in fine]

    def pick_gap_ups(self) -> None:
        # Calculate gap-up: (Open - PreviousClose) / PreviousClose
        gap_list = []
        for symbol in self._previous_close:
            if symbol in self.securities:
                security = self.securities[symbol]
                prev_close = self._previous_close[symbol]
                current_open = security.open
                # Validate
                if prev_close > 0 and current_open > 0:
                    gap_pct = (current_open - prev_close) / prev_close
                    if gap_pct > 0:  # only positive gap-ups
                        gap_list.append((symbol, gap_pct))

        # Sort by gap descending, take top 100
        gap_list.sort(key=lambda x: x[1], reverse=True)
        gap_list = gap_list[:self._top_gap_ups]

        # Record the open price in our watchlist
        for symbol, _ in gap_list:
            if symbol in self.securities:
                self._watchlist[symbol] = self.securities[symbol].open

        # Make sure these symbols have second-resolution data
        for symbol, _ in gap_list:
            if not self.securities.contains_key(symbol):
                self.add_equity(symbol.value, Resolution.SECOND)

        # Log watchlist
        watch_symbols = [s.value for s in self._watchlist]
        self.debug(f"{self.time} - Daily Watchlist: {watch_symbols}")

    def on_data(self, data: Slice) -> None:
        for symbol, bar in data.bars.items():
            if bar is None:
                continue

            # Special verification logs
            if self.time.date() == datetime(2021, 5, 27).date() and symbol.value.upper() == "SBET":
                self.debug("Verification on SBET for 5/27.")
            if self.time.date() == datetime(2021, 6, 6).date() and symbol.value.upper() == "KNW":
                self.debug("Verification on KNW for 6/6.")

            # Skip if not in watchlist
            if symbol not in self._watchlist:
                continue

            open_price = self._watchlist[symbol]
            if open_price <= 0:
                continue

            current_price = bar.close
            invested = self.portfolio[symbol].invested

            # Entry logic: price >= 30% above open => buy if not traded
            if not invested and symbol not in self._traded_symbols:
                if current_price >= open_price * 1.3:
                    qty = self.calculate_order_quantity(symbol, 0.01)  # 1% of portfolio
                    if qty != 0:
                        self.market_order(symbol, qty)
                        self._traded_symbols.add(symbol)
                        self._entry_prices[symbol] = current_price
                        self.debug(f"BUY {symbol.value} at {current_price}")

            # If currently holding, check for +/- 35% from entry
            else:
                if symbol in self._entry_prices:
                    entry_price = self._entry_prices[symbol]
                    if entry_price > 0:
                        pnl_pct = (current_price - entry_price) / entry_price
                        # Profit target
                        if pnl_pct >= 0.35:
                            self.liquidate(symbol, "TP hit")
                            self._watchlist.pop(symbol, None)
                            self._entry_prices.pop(symbol, None)
                            self.debug(f"Profit target hit. Liquidating {symbol.value} at {current_price}")
                        # Stop loss
                        elif pnl_pct <= -0.35:
                            self.liquidate(symbol, "SL hit")
                            self._watchlist.pop(symbol, None)
                            self._entry_prices.pop(symbol, None)
                            self.debug(f"Stop loss hit. Liquidating {symbol.value} at {current_price}")

    def close_positions_end_of_day(self) -> None:
        # Liquidate all and clear watchlist
        for symbol in list(self._watchlist.keys()):
            if self.portfolio[symbol].invested:
                self.liquidate(symbol, "End of day")
        self._watchlist.clear()
        self._entry_prices.clear()
        self._traded_symbols.clear()
        self.debug(f"{self.time} - End of day liquidation and watchlist cleared.")

This algorithm creates a new daily gap-up watchlist, scans it every second, enters trades as soon as the 30% threshold is hit, manages exits according to your profit/loss requirements, and resets tracking each day. The symbols for each day are logged, and special debug lines for SBET (5/27) and KNW (6/6) are included.

Sources: Opening Range Breakout for Stocks in Play, Gap Side By Side White - QuantConnect.com, docs writing algorithms datasets quiver quantitative wallstreetbets, Upside Gap Two Crows - QuantConnect.com, docs writing algorithms datasets quantconnect us etf constituents

Thanks!