Top Gaps

Here is a QuantConnect algorithm in Python that captures your described strategy requirements:

from AlgorithmImports import *

class GapUpMomentumAlgorithm(QCAlgorithm):
    """
    Intraday gap‐up momentum strategy

    1. 09:35 ET – build a daily watchlist of the 20 US stocks with the largest
       positive open-gap: (open – prev-close) / prev-close.
    2. From 09:35 ET until the bell, scan the watchlist every second.
       • Buy ONCE per stock the first time its price ≥ 30 % above today’s open.
    3. Manage each position with a 35 % take-profit and 35 % stop-loss.
    4. 15:55 ET – liquidate all holdings and reset state for the next session.
    """

    # ------------------------------------------------------------
    # Initialization
    # ------------------------------------------------------------
    def initialize(self):
        self.set_start_date(2021, 1, 1)
        self.set_end_date(2022, 1, 1)
        self.set_cash(100_000)

        self.universe_settings.resolution = Resolution.MINUTE
        self._benchmark = self.add_equity("SPY", Resolution.MINUTE).symbol

        # Daily-cleared state containers
        self._universe_size = 500           # liquid universe size
        self._watchlist = []                # top-gap symbols
        self._open_prices = {}              # {symbol: session open price}
        self._entry_prices = {}             # {symbol: entry price}
        self._held_today = set()            # symbols already traded
        self._last_scan_time = datetime.min # debounce helper

        # Universe selection & scheduled tasks
        self.add_universe(self._coarse_selection)

        self.schedule.on(
            self.date_rules.every_day(self._benchmark.value),
            self.time_rules.after_market_open(self._benchmark.value, 5),
            self._generate_daily_watchlist
        )

        self.schedule.on(
            self.date_rules.every_day(self._benchmark.value),
            self.time_rules.before_market_close(self._benchmark.value, 5),
            self._eod_cleanup
        )

    # ------------------------------------------------------------
    # Coarse universe – top N by dollar volume
    # ------------------------------------------------------------
    def _coarse_selection(self, coarse):
        liquid = sorted(coarse, key=lambda x: x.dollar_volume, reverse=True)
        selected = [c.symbol for c in liquid[: self._universe_size]]
        self._candidate_symbols = selected          # save for the morning scan
        return selected

    # ------------------------------------------------------------
    # 09:35 ET – build the day’s watchlist
    # ------------------------------------------------------------
    def _generate_daily_watchlist(self):
        # fresh start every session
        self._watchlist.clear()
        self._open_prices.clear()
        self._entry_prices.clear()
        self._held_today.clear()

        if not getattr(self, "_candidate_symbols", None):
            self.debug("[WatchList] No universe candidates.")
            return

        symbols = list(self._candidate_symbols)

        # prev-day close
        daily = self.history(symbols, 2, Resolution.DAILY)
        if daily.empty:
            self.debug("[WatchList] Missing daily history.")
            return
        prev_close = {s: float(daily.loc[s].iloc[0].close)
                      for s in symbols if s in daily.index.levels[0]}

        # today’s 09:30 open
        minute = self.history(symbols, 1, Resolution.MINUTE)
        if minute.empty:
            self.debug("[WatchList] Missing minute history.")
            return
        open_px = {s: float(minute.loc[s].iloc[0].open)
                   for s in symbols if s in minute.index.levels[0]}

        # positive gaps
        gaps = []
        for s in symbols:
            if s in prev_close and s in open_px and prev_close[s] > 0:
                gap = (open_px[s] - prev_close[s]) / prev_close[s] * 100
                if gap > 0:
                    gaps.append((s, gap))

        if not gaps:
            self.debug("[WatchList] No positive gaps today.")
            return

        gaps.sort(key=lambda x: x[1], reverse=True)
        self._watchlist = [s for s, _ in gaps[:20]]
        self._open_prices = {s: open_px[s] for s in self._watchlist}

        # subscribe to second bars for fast intraday scans
        for s in self._watchlist:
            if not self.securities.contains_key(s) or \
               self.securities[s].resolution != Resolution.SECOND:
                self.add_equity(s.value, Resolution.SECOND)

        self.debug(f"[WatchList] {self.time.date()} → {len(self._watchlist)} symbols: "
                   f"{[s.value for s in self._watchlist]}")

    # ------------------------------------------------------------
    # OnData – run once/second for entries & exits
    # ------------------------------------------------------------
    def on_data(self, data):
        if not self._watchlist:
            return

        # debounce to 1 Hz
        if (self.time - self._last_scan_time) < timedelta(seconds=1):
            return
        self._last_scan_time = self.time

        # -------- Entries -------------------------------------------------
        for s in list(self._watchlist):
            if s in self._held_today or not data.contains_key(s):
                continue

            price = float(data[s].close if hasattr(data[s], "Close") else data[s].value)
            if price <= 0 or s not in self._open_prices:
                continue

            if price >= self._open_prices[s] * 1.30:
                qty = self.calculate_order_quantity(s, 0.05)   # 5 % allocation
                if qty:
                    self.market_order(s, qty, tag="30% breakout")
                    self._entry_prices[s] = price
                    self._held_today.add(s)

        # -------- Exits (TP / SL) -----------------------------------------
        for s in list(self._held_today):
            if not self.portfolio[s].invested:
                self._cleanup_symbol(s)
                continue
            if not data.contains_key(s):
                continue

            price = float(data[s].close if hasattr(data[s], "Close") else data[s].value)
            entry = self._entry_prices.get(s, 0)
            if not entry:
                continue

            if price >= entry * 1.35:
                self.liquidate(s, tag="TP 35%")
            elif price <= entry * 0.65:
                self.liquidate(s, tag="SL 35%")

    # ------------------------------------------------------------
    # OrderEvent – housekeeping after fills
    # ------------------------------------------------------------
    def on_order_event(self, order_event):
        if order_event.status == OrderStatus.FILLED and \
           order_event.direction == OrderDirection.SELL:
            self._cleanup_symbol(order_event.symbol)

    # helper to purge symbol from today’s state
    def _cleanup_symbol(self, symbol):
        self._held_today.discard(symbol)
        self._entry_prices.pop(symbol, None)
        if symbol in self._watchlist:
            self._watchlist.remove(symbol)

    # ------------------------------------------------------------
    # 15:55 ET – end-of-day cleanup
    # ------------------------------------------------------------
    def _eod_cleanup(self):
        self.liquidate()
        self._watchlist.clear()
        self._open_prices.clear()
        self._entry_prices.clear()
        self._held_today.clear()
        self.debug(f"[EOD] {self.time.date()} → reset complete.")

• The code creates and scans the top 20 gap-up stocks daily, automatically manages positions, enforces buy/sell rules, and resets cleanly after every trading day.
• Comments clarify the logic so you can customize it further easily.

Sources: Momentum And State Of Market Filters, Head & Shoulders TA Pattern Detection, Overnight Anomaly, Opening Range Breakout for Stocks in Play, Gap Side By Side White - QuantConnect.com