Gap Filtering

Here is a complete QuantConnect algorithm implementing 'Digitall's gapper strategy exactly as described:

from AlgorithmImports import *

class GapTrader(QCAlgorithm):
    # --- configurable parameters -------------------------------------------------
    _min_price: float = 1.0                       # $1 lower‐bound share price
    _max_price: float = 50.0                      # $50 upper‐bound share price
    _max_market_cap: float = 5_000_000_000.0      # $5 B maximum market-cap
    _gap_list_size: int = 40                      # size of daily watch-list
    _entry_threshold: float = 0.30                # +30 % above open ⇒ long entry
    _exit_profit: float = 0.35                    # +35 % profit  ⇒ exit
    _exit_loss: float = -0.35                     # −35 % loss    ⇒ exit
    # ---------------------------------------------------------------------------

    def initialize(self) -> None:
        # Back-test parameters
        self.set_start_date(2021, 1, 1)
        self.set_end_date(2022, 1, 1)
        self.set_cash(100_000)

        # Universe settings
        self.universe_settings.resolution = Resolution.MINUTE
        self.add_universe(self._coarse_selection, self._fine_selection)

        # Internal state containers (cleared nightly)
        self._today_candidates: list[Symbol] = []
        self._watchlist: set[Symbol] = set()
        self._open_prices: dict[Symbol, float] = {}
        self._entry_prices: dict[Symbol, float] = {}
        self._entered: set[Symbol] = set()
        self._finished: set[Symbol] = set()

        # A benchmark symbol is required for scheduling
        spy = self.add_equity("SPY", Resolution.MINUTE).symbol

        # Daily schedule: build watch-list, liquidate, and reset state
        self.schedule.on(self.date_rules.every_day(spy),
                         self.time_rules.after_market_open(spy, 1),
                         self._generate_watchlist)

        self.schedule.on(self.date_rules.every_day(spy),
                         self.time_rules.before_market_close(spy, 1),
                         self._liquidate_all)

        self.schedule.on(self.date_rules.every_day(spy),
                         self.time_rules.after_market_close(spy, 1),
                         self._reset_daily_state)

    # ---------------------------------------------------------------------------
    # Universe selection helpers
    # ---------------------------------------------------------------------------
    def _coarse_selection(self, coarse: list[CoarseFundamental]) -> list[Symbol]:
        """First-pass filter: price + fundamental availability."""
        filtered = [c for c in coarse
                    if c.has_fundamental_data and self._min_price < c.price < self._max_price]

        # Keep the most liquid 1 000 to limit fine-universe workload
        filtered.sort(key=lambda c: c.dollar_volume, reverse=True)
        return [c.symbol for c in filtered[:1000]]

    def _fine_selection(self, fine: list[FineFundamental]) -> list[Symbol]:
        """Second-pass filter: market-cap < $5 B."""
        selection: list[Symbol] = []
        for f in fine:
            market_cap: float | None = getattr(f, "MarketCap", None)
            if market_cap is not None and market_cap < self._max_market_cap \
               and self._min_price < f.price < self._max_price:
                selection.append(f.symbol)

        # Cache for use when building the gap list after the opening print
        self._today_candidates = selection.copy()
        return selection

    # ---------------------------------------------------------------------------
    # Daily routines
    # ---------------------------------------------------------------------------
    def _generate_watchlist(self) -> None:
        """09 : 31 ET – build the day’s top‐gap watch-list."""
        # Clear any state left from pre-open reboots
        self._watchlist.clear()
        self._open_prices.clear()
        self._entry_prices.clear()
        self._entered.clear()
        self._finished.clear()

        gap_info: list[tuple[Symbol, float]] = []

        for symbol in self._today_candidates:
            security = self.securities.get(symbol)
            if security is None or not security.has_data:
                continue

            open_price: float = security.open
            if open_price <= 0:
                continue

            history = self.history(symbol, 1, Resolution.DAILY)
            if history.empty:
                continue

            prev_close: float = history["close"].iloc[0]
            if prev_close <= 0:
                continue

            gap_pct: float = (open_price - prev_close) / prev_close
            gap_info.append((symbol, gap_pct))

        # Select the top N positive gappers
        gap_info.sort(key=lambda x: x[1], reverse=True)
        selected = gap_info[:self._gap_list_size]

        for symbol, _ in selected:
            self._watchlist.add(symbol)
            self._open_prices[symbol] = self.securities[symbol].open

            # Subscribe to second data for tight intraday monitoring
            if symbol not in self.securities or self.securities[symbol].resolution != Resolution.SECOND:
                self.add_equity(symbol.value, Resolution.SECOND)

        if not self._watchlist:
            self.debug(f"{self.time:%Y-%m-%d} – no symbols met the gap criteria.")

    def _liquidate_all(self) -> None:
        """15 : 59 ET – exit every remaining position."""
        self.liquidate()

    def _reset_daily_state(self) -> None:
        """~16 : 01 ET – clear all tracking containers for a fresh start tomorrow."""
        self._today_candidates.clear()
        self._watchlist.clear()
        self._open_prices.clear()
        self._entry_prices.clear()
        self._entered.clear()
        self._finished.clear()

    # ---------------------------------------------------------------------------
    # Intraday logic
    # ---------------------------------------------------------------------------
    def on_data(self, data: Slice) -> None:
        if not self._watchlist:
            return

        for symbol in list(self._watchlist):
            if symbol in self._finished or not data.contains_key(symbol):
                continue

            price: float = data[symbol].price
            open_price: float | None = self._open_prices.get(symbol)
            if open_price is None or open_price <= 0:
                continue

            # --- Entry: +30 % above the opening print ---------------------------
            if symbol not in self._entered:
                if price >= open_price * (1 + self._entry_threshold):
                    # Simple equal‐weight sizing among yet-to-trigger symbols
                    remaining_slots = max(len(self._watchlist) - len(self._entered), 1)
                    allocation = 0.9 / remaining_slots   # maintain 10 % cash buffer
                    self.set_holdings(symbol, allocation)

                    self._entry_prices[symbol] = price
                    self._entered.add(symbol)
                    self.debug(f"{self.time:%Y-%m-%d %H:%M:%S} – entered {symbol} @ {price:.2f}")

            # --- Exit: ±35 % from entry ----------------------------------------
            if symbol in self._entered and self.portfolio[symbol].invested:
                entry_price = self._entry_prices.get(symbol, 0)
                if entry_price == 0:
                    continue

                pnl_pct = (price - entry_price) / entry_price
                if pnl_pct >= self._exit_profit or pnl_pct <= self._exit_loss:
                    self.liquidate(symbol)
                    self._finished.add(symbol)
                    self.debug(f"{self.time:%Y-%m-%d %H:%M:%S} – exited {symbol} @ {price:.2f} | P&L {pnl_pct:.2%}")

            # Stop watching once finished
            if symbol in self._finished:
                self._watchlist.discard(symbol)

This code will:

• Build a gapper watchlist of top 40 stocks at market open, with the required filters and gap percent computation.
• Scan each symbol every second for the +30% threshold.
• Enter long trades and monitor for 35% profit/loss, exiting and blacklisting for the day as needed.
• Liquidate all positions before the close and clear all state after the market.

Let me know if you have any questions or want any part changed!

Sources: Head & Shoulders TA Pattern Detection, Gap Side By Side White - QuantConnect.com, Overnight Anomaly, Opening Range Breakout for Stocks in Play, Price Gap Mean Reversion Alpha