Import File

Hello I want a day-trading algorithm that I can trade live. I'm going to run this algorithm live by connecting my QuantConnect account to my Schwab trading account. Is it possible for you to code an algorithm like this for me?

I want the algorithm to put together a watchlist one second after 9:30 ET, right after the market opens. To make the watchlist, I want the algorithm to look at three websites. Any stock listed on benzinga.com/premarket in the “Premarket Gainers” section with a “+/-” that is 10% or higher, I want that stock added to my watchlist. Any stock listed on marketchameleon.com/Reports/PremarketTrading in the “Most Active” section or the “Gainers” section with a “% Chg” that is +10% or higher, I want that stock added to my watchlist. Any stock listed on https://www.investing.com/equities/pre-market after you switch over to the “United States Stocks” tab that is +10% or higher, I want that stock added to my watchlist.

I only want the watchlist to include stocks that had $10,000,000 dollar volume during the premarket hours this morning. And I only want the watchlist to include stocks that opened at a share price of $0.75 or higher at the start of regular trading hours this morning.

Once that watchlist is compiled, I want the algorithm to set a buy order for every single stock on the watchlist. The buy order needs to be exactly 10% above the opening share price of the stock. I need the buy order to be for $10 worth of the stock based on the buy price.

Then I want the algorithm to monitor every 10 seconds for when any of the buy orders gets executed. If a buy order is executed, I want the algorithm to immediately set an order-cancels-order exit order. The two prices for this order-cancels-order setup should be exactly 40% above the stock's opening price and exactly 5% above the stock's opening price.

20 seconds before market close, I need the algorithm to liquidate all positions that are currently open as long as they are not the symbol TQQQ and then cancel any orders that are remaining.

Can you code this strategy for me please? Thanks!

The material on this website is provided for informational purposes only and does not constitute an offer to sell, a solicitation to buy, or a recommendation or endorsement for any security or strategy, nor does it constitute an offer to provide investment advisory services by QuantConnect. In addition, the material offers no opinion with respect to the suitability of any security or specific investment. QuantConnect makes no guarantees as to the accuracy or completeness of the views expressed in the website. The views are subject to change, and may have become unreliable for various reasons, including changes in market conditions or economic circumstances. All investments involve risk, including loss of principal. You should consult with an investment professional before making any investment decisions.

def initialize(self) -> None: """ Production-ready, defensive implementation of a premarket breakout OCO strategy targeting Schwab brokerage. Notes: - Scraping third-party pages can break if HTML changes or providers block automated access. We therefore fetch once per session, throttle, and expose a ManualTickers parameter override. In backtests, we skip web fetches entirely. - We proxy the “official RTH open” using the first trade at/after 09:30:00 ET from tick data; minute-bar opens are only known at 09:31. - OCO is emulated at the algorithm level: after entry fill we submit a limit take-profit and a sell stop; when one fills, we cancel the sibling. """ # Environment configuration self.set_start_date(2021, 1, 1) self.set_end_date(2022, 1, 1) self.set_cash(100000) self.set_time_zone("America/New_York") # Security initializer with price seeding and leverage=1 (conservative) base_initializer = BrokerageModelSecurityInitializer(self.brokerage_model, FuncSecuritySeeder(self.get_last_known_prices)) def custom_initializer(security): base_initializer.initialize(security) try: security.set_leverage(1) except Exception as e: self.debug(f"[INIT] Leverage set failed for {security.symbol.value}: {e}") self.set_security_initializer(custom_initializer) # Universe settings: minute resolution, extended hours enabled self.universe_settings.resolution = Resolution.MINUTE self.universe_settings.extended_market_hours = True # Parameters self._allow_fractional = self._get_bool_param("allowFractional", False) # Auto-detect Schwab; keep fractionals disabled by default (flip via QC parameter if brokerage later supports it) model_name = type(self.brokerage_model).__name__ if ("Schwab" in model_name) or ("Charles" in model_name): self._allow_fractional = False self._max_concurrent_symbols = self._get_int_param("MaxConcurrentSymbols", 40) self._entry_limit_offset = self._get_float_param("entryLimitOffset", 0.002) manual_csv = self.get_parameter("ManualTickers") self._manual_tickers = self._parse_manual_tickers(manual_csv) # Sizing constants self._dollar_notional_per_symbol = 10.0 self._min_fractional_notional = 2.0 # Session-scoped caches and state self._watchlist_built_date = datetime.min.date() self._web_cache = {"date": datetime.min.date(), "tickers": []} self._candidates = set() # Symbols that passed premarket filters self._state = {} # None/PendingOpen/PendingEntry/Entered/Exited self._rth_open = {} # Per-symbol RTH open price self._tick_subscribed = set() # Symbols we subscribed to ticks for open capture # Order tracking and OCO linkage self._entry_ticket = {} self._tp_ticket = {} self._sl_ticket = {} self._targets = {} # {symbol: {"takeProfitPx":..., "stopOutPx":..., "openPrice":..., "buyTrigger":...}} self._entry_order_type = {} # "StopLimit" or "StopMarket" # End-of-day flush flag self._flushed_for_close = False # Baseline symbol (SPY) for scheduling and market hours utilities spy_sec = self.add_equity("SPY", resolution=Resolution.SECOND, extended_market_hours=True) self._spy = spy_sec.symbol # 09:30:01 ET kickoff: build watchlist and attempt order placement self.schedule.on(self.date_rules.every_day(self._spy), self.time_rules.at(9, 30, 1), self.build_watchlist_and_place_orders) # 10-second heartbeat using SPY seconds consolidated bars self.consolidate(self._spy, timedelta(seconds=10), self._on_heartbeat_consolidated) self.debug(f"[INIT] model={model_name}, allow_fractional={self._allow_fractional}, max_concurrent={self._max_concurrent_symbols}, offset={self._entry_limit_offset}") # region Core scheduled actions def build_watchlist_and_place_orders(self) -> None: """ 09:30:01 ET: fetch premarket candidates, filter by premarket $ volume (04:00–09:30 ET), establish RTH open, and place 10% breakout stop(-limit) entries. If open not yet known, subscribe to ticks and defer placing orders until available. """ today = self.time.date() if self._watchlist_built_date == today: self.debug("[PREMKT] Watchlist already built today; skipping.") return self._watchlist_built_date = today raw_tickers = self.fetch_premarket_candidates() if len(raw_tickers) == 0: self.debug("[WEB] No tickers from web/manual sources; nothing to do.") return self.debug(f"[WEB] Pulled {len(raw_tickers)} tickers (pre-dedupe/normalize)") normalized = self._normalize_tickers(raw_tickers) unique_syms = sorted(set(normalized)) added_syms = [] for tkr in unique_syms: try: sec = self.add_equity(tkr, resolution=Resolution.MINUTE, extended_market_hours=True) sym = sec.symbol added_syms.append(sym) if sym not in self._state: self._state[sym] = "PendingOpen" except Exception as e: self.debug(f"[PREMKT] Skipping {tkr}: AddEquity failed: {e}") passing = [] for sym in added_syms: pm_dv = self.compute_premarket_dollar_volume(sym) if pm_dv >= 10_000_000: passing.append(sym) else: self.debug(f"[PREMKT] {sym.value} rejected: premarket $vol ${pm_dv:,.0f} < $10,000,000") self._state[sym] = "Exited" self._candidates = set(passing) self.debug(f"[PREMKT] {len(self._candidates)} symbols passed premarket filter") for sym in list(self._candidates): open_px = self.ensure_rth_open(sym) if open_px is None: # RTH open not yet known; subscribe to ticks and wait (heartbeat will retry) self._subscribe_ticks(sym) continue if open_px < 0.75: self.debug(f"[PREMKT] {sym.value} rejected: RTH open {open_px:.4f} < 0.75") self._state[sym] = "Exited" continue self.place_entry_and_record(sym, open_px) def _on_heartbeat_consolidated(self, bar) -> None: """Consolidator callback every ~10 seconds for orchestration.""" try: self.on_ten_second_heartbeat() except Exception as e: self.debug(f"[HB] Exception in heartbeat: {e}") # endregion # region Data handlers def on_data(self, data: Slice) -> None: """ Capture the first tick at/after 09:30:00 ET to proxy the RTH open. Once captured, initiate order placement if state is PendingOpen. """ if data is None: return if len(self._candidates) == 0 and len(self._tick_subscribed) == 0: return rth_open_dt = self._today_time(9, 30, 0) for sym in list(self._tick_subscribed): if sym in self._rth_open: continue ticks = data.ticks.get(sym) if ticks is None or len(ticks) == 0: continue for tk in ticks: if tk.time < rth_open_dt: continue if tk.price <= 0: continue self._rth_open[sym] = float(round(tk.price, 4)) self.debug(f"[OPEN] {sym.value} RTH open via tick: {self._rth_open[sym]:.4f} @ {tk.time}") if self._state.get(sym, "") == "PendingOpen": if self._rth_open[sym] >= 0.75: self.place_entry_and_record(sym, self._rth_open[sym]) else: self.debug(f"[PREMKT] {sym.value} rejected: RTH open {self._rth_open[sym]:.4f} < 0.75") self._state[sym] = "Exited" break def on_order_event(self, order_event: OrderEvent) -> None: """ Order lifecycle management: - On entry fill -> create bracket exits (limit TP, stop-market SL). - On one exit fill -> cancel the sibling. - On entry rejection -> fallback from StopLimit to StopMarket once. """ if order_event is None: return symbol = order_event.symbol ticket = self.transactions.get_order_ticket(order_event.order_id) if ticket is None: return # Log rejections/cancellations if order_event.status in [OrderStatus.CANCELED, OrderStatus.INVALID]: self.debug(f"[ORDER] {symbol.value} {order_event.status}: {order_event.message}") # Entry management if symbol in self._entry_ticket and ticket.order_id == self._entry_ticket[symbol].order_id: if order_event.status == OrderStatus.FILLED: self._state[symbol] = "Entered" qty = float(order_event.fill_quantity) targs = self._targets.get(symbol, {}) tp_px = float(targs.get("takeProfitPx", 0.0)) sl_px = float(targs.get("stopOutPx", 0.0)) self.debug(f"[ENTRY] {symbol.value} filled {qty} @ {order_event.fill_price:.4f}; TP={tp_px:.4f} SL={sl_px:.4f}") # Create bracket exits (idempotent guard inside) self._ensure_brackets(symbol, int(abs(qty)), tp_px, sl_px) elif order_event.status in [OrderStatus.INVALID]: if self._entry_order_type.get(symbol, "") == "StopLimit": self.debug(f"[ENTRY] {symbol.value} entry rejected; retrying with StopMarket") self._place_entry_stop_market_retry(symbol) # Exit bracket OCO emulation if symbol in self._tp_ticket and ticket.order_id == self._tp_ticket[symbol].order_id: if order_event.status == OrderStatus.FILLED: self.debug(f"[BRACKET] {symbol.value} Take-Profit filled @ {order_event.fill_price:.4f}; canceling Stop") self._safe_cancel(self._sl_ticket.get(symbol)) self._cleanup_symbol(symbol) if symbol in self._sl_ticket and ticket.order_id == self._sl_ticket[symbol].order_id: if order_event.status == OrderStatus.FILLED: self.debug(f"[BRACKET] {symbol.value} Stop-Loss filled @ {order_event.fill_price:.4f}; canceling TP") self._safe_cancel(self._tp_ticket.get(symbol)) self._cleanup_symbol(symbol) # endregion # region Heartbeat orchestration def on_ten_second_heartbeat(self) -> None: """ Every ~10 seconds: ensure opens, mend missing brackets, enforce concurrency, and perform end-of-day safety flush 20 seconds before market close. """ now = self.time # EOD liquidation ~20 seconds before the next market close try: next_close = self.securities[self._spy].exchange.hours.get_next_market_close(now, False) except Exception: next_close = now if (not self._flushed_for_close) and (now >= next_close - timedelta(seconds=20)): self._flushed_for_close = True self.debug("[EOD] Flushing: canceling open orders and liquidating positions (except TQQQ)") # 1) Cancel all open orders for ticket in list(self.transactions.get_order_tickets()) or []: self._safe_cancel(ticket) # 2) Liquidate all holdings except TQQQ for holding in list(self.portfolio.values()) or []: if holding is None or not holding.invested: continue sym = holding.symbol if sym.value.upper() == "TQQQ": continue try: self.liquidate(sym) except Exception as e: self.debug(f"[EOD] Liquidate {sym.value} failed: {e}") # 3) Clear OCO state to prevent next-day interference self._clear_all_symbol_state() return # Ensure RTH open and place entries as soon as available for sym in list(self._candidates): if self._state.get(sym) != "PendingOpen": continue open_px = self.ensure_rth_open(sym) if open_px is None: continue if open_px < 0.75: self.debug(f"[PREMKT] {sym.value} rejected: RTH open {open_px:.4f} < 0.75") self._state[sym] = "Exited" continue self.place_entry_and_record(sym, open_px) # Mend missing brackets for entered positions (idempotent) for sym, state in list(self._state.items()): if state != "Entered": continue targs = self._targets.get(sym) if not targs: continue qty = int(abs(self.portfolio[sym].quantity)) if qty <= 0: continue tp_px = float(targs.get("takeProfitPx", 0.0)) sl_px = float(targs.get("stopOutPx", 0.0)) self._ensure_brackets(sym, qty, tp_px, sl_px) # endregion # region Strategy helpers def fetch_premarket_candidates(self) -> list: """ Live only: download premarket lists once per session from: - Benzinga Premarket Gainers - MarketChameleon Premarket Trading - Investing.com US pre-market Backtest: skip web fetches; rely on ManualTickers. """ today = self.time.date() # Return cache if already fetched today if self._web_cache["date"] == today: cached = list(self._web_cache.get("tickers", [])) if len(cached) > 0: self.debug(f"[WEB] Using cached tickers: {len(cached)}") return cached + self._manual_tickers tickers = [] if self.live_mode: pages = [ ("https://www.benzinga.com/premarket", "benzinga"), ("https://marketchameleon.com/Reports/PremarketTrading", "marketchameleon"), ("https://www.investing.com/equities/pre-market", "investing") ] for url, tag in pages: html = self._download_with_retries(url) if len(html) == 0: self.debug(f"[WEB] {tag} download failed/empty; continuing") continue found = self._extract_candidates_from_html(html) if len(found) > 0: tickers.extend(found) self.debug(f"[WEB] {tag} extracted {len(found)} candidates") else: self.debug("[WEB] Backtest mode: skipping web fetch; using ManualTickers only") if len(self._manual_tickers) > 0: self.debug(f"[WEB] ManualTickers appended: {len(self._manual_tickers)}") tickers.extend(self._manual_tickers) self._web_cache["date"] = today self._web_cache["tickers"] = list(tickers) return tickers def compute_premarket_dollar_volume(self, symbol) -> float: """ Sum close*volume for minute bars from 04:00 to 09:30 ET today with extended hours enabled. Defensive against missing data and zero/negative values. """ if symbol is None: return 0.0 start = self._today_time(4, 0, 0) end = self._today_time(9, 30, 0) total = 0.0 try: bars = self.history[TradeBar](symbol, start, end, Resolution.MINUTE, extended_market_hours=True) if bars is None: return 0.0 for bar in bars: px = float(bar.close) vol = float(bar.volume) if px <= 0 or vol <= 0: continue total += px * vol except Exception as e: self.debug(f"[PREMKT] History failed for {symbol.value}: {e}") return 0.0 return float(total) def ensure_rth_open(self, symbol) -> float: """ Return today's RTH open if known: - Prefer first tick at/after 09:30:00 ET (requires tick subscription). - Fallback to minute bar open from the 09:30–09:31 bar once available. - If still unavailable by 09:31:30, skip for the day. """ if symbol in self._rth_open: return float(self._rth_open[symbol]) rth_dt = self._today_time(9, 30, 0) now = self.time if now < rth_dt: self._subscribe_ticks(symbol) return None if now <= rth_dt + timedelta(seconds=30): self._subscribe_ticks(symbol) return None # Minute-open fallback once the 09:30 bar materializes try: one = self.history[TradeBar](symbol, rth_dt, rth_dt + timedelta(minutes=1), Resolution.MINUTE, extended_market_hours=True) first = None for b in one: first = b break if first is not None and float(first.open) > 0: self._rth_open[symbol] = float(round(first.open, 4)) self.debug(f"[OPEN] {symbol.value} RTH open via minute bar: {self._rth_open[symbol]:.4f}") return float(self._rth_open[symbol]) except Exception as e: self.debug(f"[OPEN] {symbol.value} minute-open fallback failed: {e}") # Final attempt after 09:31:30 if now >= rth_dt + timedelta(minutes=1, seconds=30): try: two = self.history[TradeBar](symbol, rth_dt, rth_dt + timedelta(minutes=2), Resolution.MINUTE, extended_market_hours=True) first = None for b in two: first = b break if first is not None and float(first.open) > 0: self._rth_open[symbol] = float(round(first.open, 4)) self.debug(f"[OPEN] {symbol.value} RTH open via minute bar (delayed): {self._rth_open[symbol]:.4f}") return float(self._rth_open[symbol]) else: self.debug(f"[OPEN] {symbol.value} open unavailable by 09:31:30; skipping for the day") self._state[symbol] = "Exited" except Exception as e: self.debug(f"[OPEN] {symbol.value} delayed minute-open failed: {e}") return None def place_entry_and_record(self, symbol, open_price: float) -> None: """ Submit buy StopLimit (preferred) 10% above RTH open with 0.2% default limit offset (configurable), falling back to StopMarket if rejected. Quantity sizing respects fractional capability and min notional. """ if symbol is None or open_price <= 0: return if self._active_entry_count() >= self._max_concurrent_symbols: return buy_trigger = float(round(open_price * 1.10, 4)) limit_px = float(round(buy_trigger * (1.0 + max(0.0, self._entry_limit_offset)), 4)) tp_px = float(round(open_price * 1.40, 4)) sl_px = float(round(open_price * 1.05, 4)) # Sizing qty = 0 if self._allow_fractional: if self._dollar_notional_per_symbol < self._min_fractional_notional: qty = 0 else: denom = float(buy_trigger) if denom <= 0: qty = 0 else: qty = float(round(self._dollar_notional_per_symbol / denom, 4)) else: denom = float(buy_trigger) if denom > 0: qty = int(self._dollar_notional_per_symbol / denom) # floor for positive values else: qty = 0 if (isinstance(qty, (int, float)) and qty <= 0): self.debug(f"[ENTRY] {symbol.value} sizing <= 0 at trigger {buy_trigger:.4f}; skipping") self._state[symbol] = "Exited" return # Day TIF props = OrderProperties() props.time_in_force = TimeInForce.DAY # Submit StopLimit entry; fallback handled in OnOrderEvent upon rejection ticket = None try: ticket = self.stop_limit_order(symbol, qty, buy_trigger, limit_px, order_properties=props, tag=f"Entry SLmt @{buy_trigger:.4f}/{limit_px:.4f}") self._entry_order_type[symbol] = "StopLimit" except Exception as e: self.debug(f"[ENTRY] StopLimit failed for {symbol.value}: {e}; falling back to StopMarket") try: ticket = self.stop_market_order(symbol, qty, buy_trigger, order_properties=props, tag=f"Entry SMark @{buy_trigger:.4f}") self._entry_order_type[symbol] = "StopMarket" except Exception as e2: self.debug(f"[ENTRY] StopMarket failed for {symbol.value}: {e2}; skipping") self._state[symbol] = "Exited" return if ticket is None: self._state[symbol] = "Exited" return self._entry_ticket[symbol] = ticket self._targets[symbol] = {"takeProfitPx": tp_px, "stopOutPx": sl_px, "openPrice": float(round(open_price, 4)), "buyTrigger": buy_trigger} self._state[symbol] = "PendingEntry" self.debug(f"[ENTRY] {symbol.value} submitted qty={qty} stop={buy_trigger:.4f} limit={limit_px:.4f} TP={tp_px:.4f} SL={sl_px:.4f}") # endregion # region Internal utilities def _ensure_brackets(self, symbol, qty: int, tp_px: float, sl_px: float) -> None: """ Ensure both bracket legs exist: - Limit sell TP at 1.40x open - StopMarket sell SL at 1.05x open (below entry; must be a stop order to be executable below the entry trigger) """ if symbol is None or qty <= 0: return # TP limit order if symbol not in self._tp_ticket or self._tp_ticket[symbol] is None or self._tp_ticket[symbol].status in [OrderStatus.CANCELED, OrderStatus.INVALID]: if tp_px > 0: try: self._tp_ticket[symbol] = self.limit_order(symbol, -qty, tp_px, tag=f"TP @{tp_px:.4f}") self.debug(f"[BRACKET] {symbol.value} TP submitted @{tp_px:.4f}") except Exception as e: self.debug(f"[BRACKET] {symbol.value} TP submit failed: {e}") # SL stop-market order if symbol not in self._sl_ticket or self._sl_ticket[symbol] is None or self._sl_ticket[symbol].status in [OrderStatus.CANCELED, OrderStatus.INVALID]: if sl_px > 0: try: self._sl_ticket[symbol] = self.stop_market_order(symbol, -qty, sl_px, tag=f"SL @{sl_px:.4f}") self.debug(f"[BRACKET] {symbol.value} SL submitted @{sl_px:.4f}") except Exception as e: self.debug(f"[BRACKET] {symbol.value} SL submit failed: {e}") def _place_entry_stop_market_retry(self, symbol) -> None: """Retry entry as StopMarket once if a StopLimit entry is rejected/invalid.""" ticket = self._entry_ticket.get(symbol) if ticket is None: return self._safe_cancel(ticket) targs = self._targets.get(symbol, {}) stop = float(targs.get("buyTrigger", 0.0)) if stop <= 0: return qty = 0 if self._allow_fractional: denom = float(stop) if denom > 0: qty = float(round(self._dollar_notional_per_symbol / denom, 4)) else: denom = float(stop) if denom > 0: qty = int(self._dollar_notional_per_symbol / denom) if (isinstance(qty, (int, float)) and qty <= 0): return props = OrderProperties() props.time_in_force = TimeInForce.DAY try: newt = self.stop_market_order(symbol, qty, stop, order_properties=props, tag=f"Entry Retry SMark @{stop:.4f}") if newt is not None: self._entry_ticket[symbol] = newt self._entry_order_type[symbol] = "StopMarket" self.debug(f"[ENTRY] {symbol.value} StopMarket retry submitted @{stop:.4f}") except Exception as e: self.debug(f"[ENTRY] {symbol.value} StopMarket retry failed: {e}") def _subscribe_ticks(self, symbol) -> None: """Ensure tick subscription for the symbol to capture the first RTH tick at/after 09:30.""" if symbol in self._tick_subscribed: return try: self.add_equity(symbol.value, resolution=Resolution.TICK, extended_market_hours=True) self._tick_subscribed.add(symbol) self.debug(f"[OPEN] Subscribed to ticks for {symbol.value}") except Exception as e: self.debug(f"[OPEN] Tick subscribe failed for {symbol.value}: {e}") def _download_with_retries(self, url: str, retries: int = 2) -> str: """Live-only web fetch via QC Download with small retry/backoff.""" txt = "" for i in range(max(1, retries + 1)): try: txt = self.download(url) except Exception as e: txt = "" self.debug(f"[WEB] Download error ({i+1}/{retries+1}) {url}: {e}") if isinstance(txt, str) and len(txt) > 0: break return txt if isinstance(txt, str) else "" def _extract_candidates_from_html(self, html: str) -> list: """ Extremely defensive HTML tokenization (no external libs): - Tokenize alphanumerics plus '.' and '-' to extract potential tickers. - If any percentage >= +10% is detected on the page, we return the collected tokens. - Final dedupe/validation happens downstream. """ if not isinstance(html, str) or len(html) == 0: return [] tokens = [] cur = [] for ch in html: if ("A" <= ch <= "Z") or ("0" <= ch <= "9") or ch in [".", "-"]: cur.append(ch) else: if len(cur) > 0: tokens.append("".join(cur)) cur = [] if len(cur) > 0: tokens.append("".join(cur)) # Percent detection (any >= +10% on the page) has_pos_10 = False num = "" sign = 1 for ch in html: if ch == '+': sign = 1 num = "" elif ch == '-': sign = -1 num = "" elif ch.isdigit() or ch == '.': num += ch elif ch == '%': try: val = float(num) * sign if val >= 10.0: has_pos_10 = True break except Exception: pass num = "" sign = 1 else: num = "" sign = 1 # Normalize candidate tokens cands = [] seen = set() for t in tokens: s = t.strip().upper() if len(s) == 0: continue if len(s) > 8: # prune long tokens continue base = s.split('.') base = base[0] if len(base) > 0 else s if len(base) < 1 or len(base) > 5: continue if not all(('A' <= c <= 'Z') or ('0' <= c <= '9') for c in base): continue if s in seen: continue seen.add(s) cands.append(s) # If >= +10% was detected, return the set; else still return (downstream filters will trim) return cands def _normalize_tickers(self, raw: list) -> list: """Upper-case, basic syntax filtering, allow class suffix ('.' or '-') up to 2 chars.""" out = [] seen = set() for t in raw: if not isinstance(t, str): continue s = t.strip().upper() if len(s) == 0: continue base = s.split('.') base = base[0] if len(base) > 0 else s if len(base) < 1 or len(base) > 5: continue if not all(('A' <= c <= 'Z') or ('0' <= c <= '9') for c in base): continue if s in seen: continue seen.add(s) out.append(s) return out def _today_time(self, hh: int, mm: int, ss: int): """Utility to build a timezone-aware datetime for 'today' in the algorithm's timezone.""" return datetime(self.time.year, self.time.month, self.time.day, hh, mm, ss, tzinfo=self.time_zone) def _active_entry_count(self) -> int: """Count currently active (non-filled/non-canceled) entry tickets for concurrency control.""" cnt = 0 for sym, ticket in list(self._entry_ticket.items()): if ticket is None: continue st = ticket.status if st in [OrderStatus.NEW, OrderStatus.SUBMITTED, OrderStatus.PARTIALLY_FILLED]: cnt += 1 return cnt def _safe_cancel(self, ticket) -> None: """Cancel a ticket if still active; swallow and log exceptions.""" try: if ticket is not None and ticket.status not in [OrderStatus.CANCELED, OrderStatus.FILLED, OrderStatus.INVALID]: ticket.cancel() except Exception as e: self.debug(f"[ORDER] Cancel failed: {e}") def _cleanup_symbol(self, symbol) -> None: """Clear per-symbol OCO/entry state after exits.""" self._tp_ticket.pop(symbol, None) self._sl_ticket.pop(symbol, None) self._entry_ticket.pop(symbol, None) self._entry_order_type.pop(symbol, None) self._targets.pop(symbol, None) self._rth_open.pop(symbol, None) if symbol in self._state: self._state[symbol] = "Exited" def _clear_all_symbol_state(self) -> None: """Clear all symbol state at EOD flush.""" self._tp_ticket.clear() self._sl_ticket.clear() self._entry_ticket.clear() self._entry_order_type.clear() self._targets.clear() self._rth_open.clear() self._candidates.clear() for k in list(self._state.keys()): self._state[k] = "Exited" def _parse_manual_tickers(self, csv_line) -> list: """Parse ManualTickers CSV parameter into a list of tickers.""" if csv_line is None: return [] s = str(csv_line).strip() if len(s) == 0: return [] parts = [p.strip().upper() for p in s.split(',') if len(p.strip()) > 0] return parts def _get_bool_param(self, key: str, default: bool) -> bool: raw = self.get_parameter(key) if raw is None: return bool(default) s = str(raw).strip().lower() return s in ["true", "1", "yes", "y", "on"] def _get_int_param(self, key: str, default: int) -> int: raw = self.get_parameter(key) try: return int(raw) if raw is not None and len(str(raw)) > 0 else int(default) except Exception: return int(default) def _get_float_param(self, key: str, default: float) -> float: raw = self.get_parameter(key) try: return float(raw) if raw is not None and len(str(raw)) > 0 else float(default) except Exception: return float(default) # endregion

Mia Alissi

STAFF Pro ,

Here’s a complete algorithm you can paste into QuantConnect. It follows your requirements: builds the watchlist at 9:30:01 ET from the three premarket pages, filters for ≥$10M premarket dollar volume and RTH open ≥ $0.75, places $10 buy stop orders 10% above the open, and on fills sets an emulated OCO with a take-profit at 1.40× open and a stop at 1.05× open, then performs a 20-seconds-before-close cleanup (skipping TQQQ). Notes:

For the “official open” we reliably proxy the first trade at/after 09:30:00 ET via a temporary Tick subscription. Minute bars only reveal their “Open” at 09:31, so we fall back to that if needed.
Schwab fractional shares are set to off by default here. If your brokerage later supports fractionals through the API, set the parameter allowFractional=true in QC.
OCO is emulated (LEAN/Schwab don’t offer native OCO via API). When one exit fills, the sibling is canceled immediately.
Scraping can break if sites change or block automated access. In backtests we skip web requests and you can supply ManualTickers (CSV parameter) as a fallback. In live, we fetch once, parse defensively, and cache.

Copy this code into a new Python project in QuantConnect:

from AlgorithmImports import *

class PremarketBreakoutOCOSchwab(QCAlgorithm):

Would you like me to:

switch entries to pure StopMarket instead of StopLimit,
raise/lower the 10-second cadence,
cap total capital committed per day, or
add email/Slack notifications for entries and exits?

Sources: Empty Equity And Options400 Benchmark, Basic Template Crypto Algorithm, Order Tickets - QuantConnect.com, Interactive Brokers - QuantConnect.com, Stop Limit Order Regression Algorithm

Retrospective Green Ocelot

6.5k Pro ,

Thanks!

Retrospective Green Ocelot INVESTOR

Update Backtest

Notebook

person upvoted this people upvoted this

To unlock posting to the community forums please complete at least 30% of Boot Camp.
You can continue your Boot Camp training progress from the terminal. We hope to see you in the community soon!

Radically Open-Source Algorithmic Trading Engine

Join Our Discord Channel

Draft Discussions

Bookmarked Discussions

SEARCH DISCUSSIONS

TOP 5 Research Publications

444,100 Quants.

VOTE FOR UPCOMING FEATURES

Import File

Allocate to this Strategy

Organization

Team

Clone Strategy

Previous Ranking

IN THIS RESEARCH

PARTICIPANTS

Discussion Awards

Actions

Join QuantConnect for Free

SIGN IN

Radically Open-Source Algorithmic Trading Engine

Join Our Discord Channel

Draft Discussions

Bookmarked Discussions

SEARCH DISCUSSIONS

TOP 5 Research Publications

444,100 Quants.

VOTE FOR UPCOMING FEATURES

Import File

Allocate to this Strategy

Organization

Team

Clone Strategy

Previous Ranking

IN THIS RESEARCH

PARTICIPANTS

Discussion Awards

SHARE RESEARCH

SHARE DISCUSSION

SHARE ARTICLE

SHARE

Actions

Join QuantConnect for Free