New Gaps

from AlgorithmImports import *
from datetime import timedelta, time

class TopGainersCustomData(PythonData):
    def get_source(self, config: SubscriptionDataConfig, date: datetime, isLiveMode: bool) -> SubscriptionDataSource:
        # Use a csv file or fixed URI with gainers for your backtest window
        url = "https://.../sim-gainers.csv?dl=1"
        return SubscriptionDataSource(url, SubscriptionTransportMedium.REMOTE_FILE)

    def reader(self, config: SubscriptionDataConfig, line: str, date: datetime, isLiveMode: bool) -> BaseData:
        # Parse line as symbol/date
        if not line.strip() or line.startswith("#"):
            return None
        parts = line.split(',')
        if len(parts) < 1:
            return None
        symbol = parts[0].strip()
        ticker = Symbol.create(symbol, SecurityType.EQUITY, Market.USA)
        data = TopGainersCustomData()
        data.symbol = ticker
        data.time = date
        data.end_time = date
        return data

class CustomDataUniverseAlgorithm(QCAlgorithm):
    def initialize(self) -> None:
        self.set_start_date(2021, 5, 25)
        self.set_end_date(2021, 6, 10)
        self.set_cash(100000)
        self.add_equity("SPY", Resolution.DAILY)
        self.set_benchmark("SPY")

        # Custom universe
        self.universe_settings.resolution = Resolution.DAILY
        self.add_universe(TopGainersCustomData, "sim-top-gainers", self.select_top_gainers)

        self._watchlist: Set[Symbol] = set()
        self._daily_open: Dict[Symbol, float] = {}
        self._open_orders: Dict[Symbol, Tuple[OrderTicket, float, float]] = {}
        self._trades_today: Set[Symbol] = set()

        self._market_open_time = time(9, 30, 0)
        self._market_close_time = time(16, 0, 0)
        self._open_delay = timedelta(seconds=5)
        self._close_delay = timedelta(seconds=10)

        # Schedules
        # (5.0/60.0 minutes after market open => 5 seconds after market open)
        # (10.0/60.0 minutes before market close => 10 seconds before market close)
        self.schedule.on(
            self.date_rules.every_day("SPY"),
            self.time_rules.after_market_open("SPY", 5.0 / 60.0),
            self.create_watchlist
        )
        self.schedule.on(
            self.date_rules.every_day("SPY"),
            self.time_rules.every(timedelta(seconds=1)),
            self.scan_watchlist
        )
        self.schedule.on(
            self.date_rules.every_day("SPY"),
            self.time_rules.before_market_close("SPY", 10.0 / 60.0),
            self.liquidate_and_clear
        )

    def select_top_gainers(self, data: List[BaseData]) -> List[Symbol]:
        # Select all symbols from the custom TopGainers data
        return [x.symbol for x in data]

    def create_watchlist(self) -> None:
        active = [x.symbol for x in self.active_securities.values if x.symbol.value != "SPY"]
        self._watchlist = set(active)
        self._trades_today.clear()
        self._open_orders.clear()
        self._daily_open.clear()

        for symbol in self._watchlist:
            bars = self.history(symbol, 1, Resolution.DAILY)
            if not bars.empty:
                self._daily_open[symbol] = float(bars["open"][0])

        self.log(f"Watchlist for {self.time.strftime('%Y-%m-%d')}: {sorted([s.value for s in self._watchlist])}")

    def scan_watchlist(self) -> None:
        now_time = self.time.time()
        if not self._watchlist or now_time >= self._market_close_time:
            return

        for symbol in list(self._watchlist):
            if symbol in self._trades_today:
                continue
            price = self.securities[symbol].price
            open_price = self._daily_open.get(symbol, 0)
            if open_price > 0 and price >= open_price * 1.3:
                qty = self.calculate_order_quantity(symbol, 0.1)
                ticket = self.market_order(symbol, qty)

                tp = price * 1.35
                sl = price * 0.65
                self._open_orders[symbol] = (ticket, tp, sl)
                self._trades_today.add(symbol)
                self._watchlist.remove(symbol)

    def on_order_event(self, orderEvent: OrderEvent) -> None:
        if orderEvent.status != OrderStatus.FILLED:
            return
        symbol = orderEvent.symbol
        if symbol in self._open_orders:
            entry_ticket, tp, sl = self._open_orders[symbol]
            price = self.securities[symbol].price
            if price >= tp or price <= sl:
                self.liquidate(symbol)
            if not self.portfolio[symbol].invested:
                del self._open_orders[symbol]

    def liquidate_and_clear(self) -> None:
        for symbol in list(self._open_orders.keys()):
            self.liquidate(symbol)
        self.log(f"End of day liquidate/clear at {self.time}")
        self._watchlist.clear()
        self._trades_today.clear()
        self._open_orders.clear()

How it works:

• At 5 seconds after NYSE open, the algorithm builds a watchlist from your daily top gainers custom data universe and logs the symbols for every day.
• Each second, it checks the current price of each watchlist symbol, and buys if the price hits +30% of the day's open.
• After entry, it will liquidate for +35% profit or -35% loss (intraday), and the symbol is not traded again that day.
• All open positions are liquidated ~10 seconds before market close, and the daily state is reset.
• This setup will log all watchlists for each backtest day, so you can verify that SBET or KNW submit trades on relevant dates, if those tickers are included in your custom data.

NOTE:

• Place your simulated top gainers list (including SBET, KNW, etc.) as a CSV or in a fixed URL and update the get_source function as needed.
• If you want a ready format for the CSV, just ask!

Sources: Custom Data Universe Algorithm, Dropbox Base Data Universe Selection Algorithm, Custom Data - QuantConnect.com, Empty Equity And Options400 Benchmark, Universe Data - QuantConnect.com

Thanks!