Cutoff Volume

from AlgorithmImports import *

class GapMomentumAlgorithm(QCAlgorithm):
    """Gap-momentum intraday strategy"""

    def initialize(self) -> None:
        # Back-test period & cash
        self.set_start_date(2021, 1, 1)
        self.set_end_date(2022, 1, 1)
        self.set_cash(100000)

        # Reference asset – required before use in scheduling rules
        self._schedule_ref_symbol = self.add_equity("SPY", Resolution.MINUTE).symbol

        # Universe configuration
        self.universe_settings.resolution = Resolution.MINUTE
        self.universe_settings.leverage   = 1.0

        # ---------- state containers ----------
        self._coarse_symbols_today = []    # List[Symbol]
        self._open_prices          = {}    # Dict[Symbol, float]
        self._watchlist            = set() # Set[Symbol]
        self._pending_entry        = {}    # Dict[int, Symbol]   (orderId ➜ Symbol)
        self._exit_order_ids       = {}    # Dict[Symbol, List[int]]
        self._open_scan_time       = datetime.min
        self._gap_scan_completed   = False
        # --------------------------------------

        # Universe: all US equities priced ≥ $0.50
        self.add_universe(self._select_coarse)

        # Market-open callback (granularity: minutes).  
        # Gap scan runs 5 s later inside OnData.
        self.schedule.on(
            self.date_rules.every_day(self._schedule_ref_symbol),
            self.time_rules.after_market_open(self._schedule_ref_symbol, 0),
            self._prepare_gap_scan
        )

        # 10 s before close: flatten & reset (minute-level schedule)
        self.schedule.on(
            self.date_rules.every_day(self._schedule_ref_symbol),
            self.time_rules.before_market_close(self._schedule_ref_symbol, 0),
            self._before_market_close
        )

    # ---------- Universe selection ----------
    def _select_coarse(self, coarse) -> List[Symbol]:
        """Mid-night coarse pass – retain symbols trading ≥ $0.50."""
        symbols = [c.symbol for c in coarse if c.price >= 0.5]
        self._coarse_symbols_today = symbols
        return symbols

    # ---------- Schedulers ----------
    def _prepare_gap_scan(self) -> None:
        """Flag that the real gap scan should execute 5 seconds after the bell."""
        self._gap_scan_completed = False
        self._open_scan_time     = self.time + timedelta(seconds=5)

    def _before_market_close(self) -> None:
        """Liquidate, cancel orders, and clear per-day state."""
        self.liquidate()
        for ticket in self.transactions.get_open_order_tickets():
            self.transactions.cancel_order(ticket.order_id)

        self._watchlist.clear()
        self._open_prices.clear()
        self._exit_order_ids.clear()
        self._gap_scan_completed = False

    # ---------- Main data handler ----------
    def on_data(self, data: Slice) -> None:
        # 1) run the gap scan exactly once, 5 s after open
        if not self._gap_scan_completed and self.time >= self._open_scan_time:
            self._run_gap_scan()
            self._gap_scan_completed = True

        # 2) continuously monitor the watchlist for entry/exit logic
        self._monitor_watchlist(data)

    # ---------- Build daily watchlist ----------
    def _run_gap_scan(self) -> None:
        if not self._coarse_symbols_today:
            return

        for symbol in self._coarse_symbols_today:
            security = self.securities[symbol]
            if not security.has_data:
                continue

            open_price = security.price
            if open_price < 0.5:
                included, reason = False, "Open < $0.50"
            else:
                history = self.history(symbol, 1, Resolution.DAILY)
                if history.empty:
                    continue
                prev_close = history.iloc[-1].close
                included   = open_price >= 1.5 * prev_close if prev_close > 0 else False
                reason     = "Gap >= 50%" if included else "Gap < 50%"

            # Sole debug output requirement
            if symbol.value.upper() == "SBET" and self.time.date() == datetime(2025, 5, 27).date():
                self.debug(f"SBET {'included' if included else 'excluded'} – {reason}")

            if not included:
                continue

            # Add qualifying symbol to daily watchlist
            self._watchlist.add(symbol)
            self._open_prices[symbol] = open_price

            # Subscribe to second-resolution data for rapid monitoring
            if security.resolution != Resolution.SECOND:
                self.add_equity(symbol.value, Resolution.SECOND)

    # ---------- Intraday monitoring & trade logic ----------
    def _monitor_watchlist(self, data: Slice) -> None:
        for symbol in list(self._watchlist):
            bar = data.bars.get(symbol) or data.quote_bars.get(symbol) or data.ticks.get(symbol)
            if bar is None:
                continue

            price      = bar.close if hasattr(bar, "Close") else bar.last_price
            open_price = self._open_prices.get(symbol, 0.0)
            if price <= 0 or open_price == 0:
                continue

            # Entry: +30 % relative to open, no current position nor pending order
            if (
                price >= 1.3 * open_price
                and not self.portfolio[symbol].invested
                and symbol not in self._pending_entry.values()
            ):
                qty = int(self.portfolio.total_portfolio_value * 0.10 / price)
                if qty > 0:
                    ticket = self.market_order(symbol, qty)
                    self._pending_entry[ticket.order_id] = symbol
                else:
                    self._watchlist.remove(symbol)  # avoid endless scanning on illiquid stocks

    # ---------- Order-status events ----------
    def on_order_event(self, order_event: OrderEvent) -> None:
        order_id = order_event.order_id
        symbol   = order_event.symbol

        # Entry filled ➜ immediately place TP & SL orders
        if order_id in self._pending_entry and order_event.status == OrderStatus.FILLED:
            self._pending_entry.pop(order_id, None)

            entry_price = order_event.fill_price
            if entry_price <= 0:
                return

            qty = self.portfolio[symbol].quantity
            if qty == 0:
                return

            tp_price = round(entry_price * 1.35, 4)  # +35 %
            sl_price = round(entry_price * 0.65, 4)  # –35 %

            tp = self.limit_order(symbol, -qty, tp_price, "TP")
            sl = self.stop_market_order(symbol, -qty, sl_price, "SL")
            self._exit_order_ids[symbol] = [tp.order_id, sl.order_id]
            return

        # One of the exit orders filled ➜ cancel the sibling & retire symbol
        if symbol in self._exit_order_ids and order_id in self._exit_order_ids[symbol]:
            if order_event.status == OrderStatus.FILLED:
                for sibling_id in self._exit_order_ids[symbol]:
                    if sibling_id != order_id:
                        sibling_ticket = self.transactions.get_order_by_id(sibling_id)
                        if sibling_ticket and sibling_ticket.status not in (
                            OrderStatus.FILLED,
                            OrderStatus.CANCELED,
                        ):
                            self.transactions.cancel_order(sibling_id)

                self._watchlist.discard(symbol)
                self._exit_order_ids.pop(symbol, None)

Key points:

• Daily 5s-after-bell scan for US stocks with gap/filter logic and watchlist reset
• Only logs SBET decision for 2025-05-27; zero other logs
• Scans watchlist every second for entry; +30% triggers buy
• Trades closed for +35%/-35% and post-trade symbol is removed from watchlist
• Liquidates & clears 10s before close

Sources: Head & Shoulders TA Pattern Detection, Gap Side By Side White - QuantConnect.com, Momentum And State Of Market Filters, Intraday ETF Momentum, Momentum Short Term Reversal Strategy

Thanks!