Negative Values

Hello is it possible to code the following strategy? I want my algorithm backtest to pull a few columns from this spreadsheet: https://docs.google.com/spreadsheets/d/e/2PACX-1vQuZAKy82QmT1dvy-nA2apz1vabq47Pv09sfNim2eYhbDfmmaBXA2sEH2rRAcnpCKgfMmbxath_V0XF/pub?gid=1367730565&single=true&output=csv

I want my algorithm backtest to set up some buy orders one second after the market opens each day. The spreadsheet will show which stocks are wanted for which day in the Date and Ticker columns. The buy orders want to be set one cent above the number shown in the Premarket High column for each stock on each date. The buy order wants to be for $10,000 worth of each stock. In other words, I want the backtest to set up a buy order right after the market opens to buy each stock on each date as soon as it breaks above its premarket high by one cent.

Each day during the regular market hours, I then want the algorithm backtest to check every ten seconds to see if any of my buy orders have triggered. If one of my buy orders has been triggered, I want an order-cancels-order set up to exit the position. The order-cancels-order should be set up to exit that stock position once it hits a 15% profit or once it hits a 15% loss, whichever comes first.

20 seconds before the market closes, I want the algorithm to liquidate any remaining positions and clear all orders and watchlists to get ready for the next day with its new stocks. I want to be able to control the start and end date of the backtests too.

Could you code this strategy in python for me? 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.

High-level behavior: - Each morning ~30 minutes before the US market open, downloads a Google Sheets CSV containing tickers and their premarket highs for the trading date. It prepares the symbols for the day and sets up a deferred entry time at 09:30:01 America/New_York. - At 09:30:01, submits Buy StopMarketOrders for each symbol at (premarket high + $0.01). - Upon an entry fill, immediately places an OCO-style bracket: * Take-profit LimitOrder at +take_profit_pct * StopMarketOrder at -stop_loss_pct If one exit fills, the sibling is canceled; partial fills dynamically resize the sibling. - Every 10 seconds during RTH, performs a light scan to ensure bracket orders exist/sync. - ~20 seconds before the closing bell, cancels orders, liquidates, and resets state for the next day. User knobs (edit near top of initialize): - Start/End dates: SetStartDate(...) and SetEndDate(...) - dollars_per_trade (default 10000) - take_profit_pct (default 0.15) - stop_loss_pct (default 0.15) - stop entry buffer: +$0.01 above premarket high """ def initialize(self) -> None: # ===================== USER-EDITABLE DATES ===================== # Edit these dates for backtests # Example: self.set_start_date(2021, 1, 1); self.set_end_date(2022, 1, 1) self.set_start_date(2021, 1, 1) self.set_end_date(2022, 1, 1) # ================================================================= # Large default cash to support multiple $10k entries per day self.set_cash(1_000_000) # Use a security seeder so initial prices are available self.set_security_initializer( BrokerageModelSecurityInitializer( self.brokerage_model, FuncSecuritySeeder(self.get_last_known_prices) ) ) # Anchor schedule to SPY sessions. Add once at DAILY resolution. spy = self.add_equity("SPY", Resolution.DAILY) spy.set_data_normalization_mode(DataNormalizationMode.RAW) self._spy = spy.symbol # ----------------------- Tunable parameters ----------------------- self.dollars_per_trade: float = 10000.0 self.take_profit_pct: float = 0.15 self.stop_loss_pct: float = 0.15 self.stop_buffer: float = 0.01 # +$0.01 above premarket high # ------------------------------------------------------------------ # ---------------------- State tracking dicts ---------------------- self.today_targets: Set[Symbol] = set() self.entry_tickets: Dict[Symbol, OrderTicket] = {} self.exit_tickets: Dict[Symbol, Dict[str, OrderTicket]] = {} self.active_symbols: Set[Symbol] = set() self.symbol_by_ticker: Dict[str, Symbol] = {} self.data_by_date: Dict[Any, pd.DataFrame] = {} self._premarket_highs: Dict[Symbol, float] = {} self._entries_submitted_today: Set[Symbol] = set() # Timing helpers self.defer_entries_until: datetime = datetime(2100, 1, 1) self.last_scan_time: datetime = datetime(2000, 1, 1) self._entries_armed: bool = False self._prepare_close: bool = False self._next_close_time: datetime = datetime(2100, 1, 1) # CSV source self._csv_url: str = ( "https://docs.google.com/spreadsheets/d/e/2PACX-1vQuZAKy82QmT1dvy-nA2apz1vabq47Pv09sfNim2eYhbDfmmaBXA2sEH2rRAcnpCKgfMmbxath_V0XF/pub?gid=1367730565&single=true&output=csv" ) self._csv_cache_raw: str = "" # Initial load of the CSV and preprocessing self._refresh_csv() # Schedule a morning refresh + prepare ~30 minutes before market open self.schedule.on( self.date_rules.every_day("SPY"), self.time_rules.before_market_open("SPY", 30), self._morning_prep ) # At the exact market open, arm entries for 09:30:01 self.schedule.on( self.date_rules.every_day("SPY"), self.time_rules.after_market_open("SPY", 0), self._arm_entries_at_open ) # One minute before the close, set a flag and capture the exact close time self.schedule.on( self.date_rules.every_day("SPY"), self.time_rules.before_market_close("SPY", 1), self._flag_prepare_close ) # ============================== SCHEDULERS =============================== def _morning_prep(self) -> None: # Re-download data (best-effort). If it fails, keep prior cache. self._refresh_csv() self._prepare_today() def _arm_entries_at_open(self) -> None: # Set entries to be allowed at 09:30:01 for today next_open, _ = self._get_market_times_for_today() self.defer_entries_until = next_open + timedelta(seconds=1) self._entries_armed = True # Reset submitted set for the day self._entries_submitted_today = set() def _flag_prepare_close(self) -> None: # Capture the exact next regular close time and set a flag to start the EOD sequence _, next_close = self._get_market_times_for_today() self._next_close_time = next_close self._prepare_close = True # ============================= CORE CALLBACKS ============================ def on_data(self, data: Slice) -> None: # Submit entries at or after 09:30:01 when armed if self._entries_armed and len(self.today_targets) > 0 and len(self._entries_submitted_today) == 0: if self.time >= self.defer_entries_until: self._submit_entries_for_today() # Light scan every 10 seconds during regular trading hours if (self.time - self.last_scan_time).total_seconds() >= 10: self.last_scan_time = self.time if self.is_market_open(self._spy): self._scan_and_sync() # End-of-day handling ~20 seconds before the close if self._prepare_close: # Guard for a valid next close timestamp if self._next_close_time > self.time: seconds_to_close = (self._next_close_time - self.time).total_seconds() if seconds_to_close <= 20: self._end_of_day_cleanup() else: # If somehow crossed, cleanup immediately self._end_of_day_cleanup() def on_order_event(self, order_event: OrderEvent) -> None: if order_event is None: return symbol = order_event.symbol order = self.transactions.get_order_by_id(order_event.order_id) if order is None: return # Handle entry fills if symbol in self.entry_tickets and self.entry_tickets[symbol] is not None: entry_ticket = self.entry_tickets[symbol] if order_event.order_id == entry_ticket.order_id: # If entry is filled if order_event.status == OrderStatus.FILLED: fill_qty = int(abs(order_event.fill_quantity)) avg_fill_price = float(order_event.fill_price) if fill_qty > 0 and avg_fill_price > 0: self._place_oco_exits(symbol, fill_qty, avg_fill_price, linked_entry_id=entry_ticket.order_id) elif order_event.status in [OrderStatus.CANCELED, OrderStatus.INVALID]: # Clean up stale entry state self.entry_tickets.pop(symbol, None) # Handle exit fills and partials if symbol in self.exit_tickets: legs = self.exit_tickets.get(symbol, {}) tp_ticket = legs.get("tp", None) sl_ticket = legs.get("sl", None) # If either leg generated the event if (tp_ticket is not None and order_event.order_id == tp_ticket.order_id) or \ (sl_ticket is not None and order_event.order_id == sl_ticket.order_id): # On full fill, cancel sibling if order_event.status == OrderStatus.FILLED: sibling = sl_ticket if order_event.order_id == (tp_ticket.order_id if tp_ticket else -1) else tp_ticket if sibling is not None and sibling.status not in [OrderStatus.CANCELED, OrderStatus.FILLED]: try: sibling.cancel("Sibling filled; OCO cancellation") except Exception as e: self.debug(f"Cancel sibling failed {symbol.value}: {str(e)}") # Clear exit tickets after position flat if self.portfolio[symbol].quantity == 0: self.exit_tickets.pop(symbol, None) else: # Partial fills or updates: sync sibling to remaining quantity self._sync_exit_orders(symbol) # ============================= ORDER HELPERS ============================= def _submit_entries_for_today(self) -> None: placed_any = False for symbol in list(self.today_targets): if symbol in self._entries_submitted_today: continue if symbol not in self._premarket_highs: # Safety: skip if we don't have a premarket high continue premkt_high = float(self._premarket_highs.get(symbol, 0.0)) stop_price = float(max(0.01, premkt_high + self.stop_buffer)) if stop_price <= 0: continue qty_float = self.dollars_per_trade / stop_price if stop_price > 0 else 0.0 qty = int(np.floor(qty_float)) if qty_float > 0 else 0 if qty <= 0: self.debug(f"Qty 0 for {symbol.value}; stop {stop_price:.2f} > dollars_per_trade {self.dollars_per_trade:.2f}") continue if symbol in self.entry_tickets and self.entry_tickets[symbol] is not None: # Avoid duplicate entries for same symbol continue tag = f"ENTRY {symbol.value} stop={stop_price:.2f} premkt={premkt_high:.2f} qty={qty}" try: ticket = self.stop_market_order(symbol, qty, stop_price) if ticket is not None: # Apply tag via update for compatibility fields = UpdateOrderFields() fields.tag = tag ticket.update(fields) self.entry_tickets[symbol] = ticket self._entries_submitted_today.add(symbol) self.active_symbols.add(symbol) placed_any = True except Exception as e: self.debug(f"Failed placing entry for {symbol.value}: {str(e)}") if placed_any: self.debug(f"Submitted {len(self._entries_submitted_today)} entries at {self.time}") def _place_oco_exits(self, symbol: Symbol, qty: int, avg_price: float, linked_entry_id: int) -> None: qty = int(max(0, qty)) if qty <= 0 or avg_price <= 0: return tp_price = float(avg_price * (1.0 + self.take_profit_pct)) sl_price = float(avg_price * (1.0 - self.stop_loss_pct)) tp_tag = f"EXIT-TP {symbol.value} qty={qty} px={tp_price:.2f} linkEntry={linked_entry_id}" sl_tag = f"EXIT-SL {symbol.value} qty={qty} px={sl_price:.2f} linkEntry={linked_entry_id}" try: tp_ticket = self.limit_order(symbol, -qty, tp_price) sl_ticket = self.stop_market_order(symbol, -qty, sl_price) if tp_ticket is not None and sl_ticket is not None: # Tag after creation f_1 = UpdateOrderFields(); f_1.tag = tp_tag; tp_ticket.update(f_1) f_2 = UpdateOrderFields(); f_2.tag = sl_tag; sl_ticket.update(f_2) self.exit_tickets[symbol] = {"tp": tp_ticket, "sl": sl_ticket} except Exception as e: self.debug(f"Failed placing exits for {symbol.value}: {str(e)}") def _sync_exit_orders(self, symbol: Symbol) -> None: if symbol not in self.exit_tickets: return legs = self.exit_tickets.get(symbol, {}) tp_ticket = legs.get("tp", None) sl_ticket = legs.get("sl", None) remaining_qty = int(abs(self.portfolio[symbol].quantity)) if remaining_qty <= 0: # Cancel both if flat for t in [tp_ticket, sl_ticket]: if t is not None and t.status not in [OrderStatus.FILLED, OrderStatus.CANCELED]: try: t.cancel("Position flat; cancel exits") except Exception: pass self.exit_tickets.pop(symbol, None) return # Update sibling quantities to remaining for t in [tp_ticket, sl_ticket]: if t is None: continue if t.status in [OrderStatus.FILLED, OrderStatus.CANCELED]: continue try: fields = UpdateOrderFields() fields.quantity = -remaining_qty # exits are sells t.update(fields) except Exception: # Ignore update failures; can be due to race conditions pass # ============================== DAILY PREP =============================== def _prepare_today(self) -> None: # Cleanup any lingering state/orders from prior day self._cancel_all_open_orders() self._clear_state_but_keep_spy() # Determine today's rows from CSV today_dt = self.time.date() rows = self._get_today_rows(today_dt) # Add securities for today's tickers unique_tickers = sorted(list({row["ticker"] for row in rows})) for ticker in unique_tickers: symbol = self._add_symbol_if_needed(ticker) self.today_targets.add(symbol) self.active_symbols.add(symbol) # Track premarket highs for row in rows: if row["ticker"] == ticker: try: self._premarket_highs[symbol] = float(row["premarket_high"]) if float(row["premarket_high"]) > 0 else 0.0 except Exception: self._premarket_highs[symbol] = 0.0 # Reset arming flags for today self._entries_armed = False self.defer_entries_until = datetime(2100, 1, 1) self._entries_submitted_today = set() self.debug(f"Prepared {len(self.today_targets)} targets for {today_dt}") # =============================== SCANNING ================================ def _scan_and_sync(self) -> None: # Ensure entry fills have exits; and quantities are synced for symbol, entry_ticket in list(self.entry_tickets.items()): if entry_ticket is None: continue status = entry_ticket.status if status == OrderStatus.FILLED: # Place exits if missing if symbol not in self.exit_tickets: avg_fill_price = self._get_ticket_fill_price(entry_ticket) fill_qty = int(abs(self.portfolio[symbol].quantity)) if fill_qty > 0 and avg_fill_price > 0: self._place_oco_exits(symbol, fill_qty, avg_fill_price, linked_entry_id=entry_ticket.order_id) else: self._sync_exit_orders(symbol) elif status in [OrderStatus.CANCELED, OrderStatus.INVALID]: # Clean up stale entries self.entry_tickets.pop(symbol, None) # Also sync exits for any active symbols we might have for symbol in list(self.active_symbols): self._sync_exit_orders(symbol) # ============================== EOD CLEANUP ============================== def _end_of_day_cleanup(self) -> None: # Cancel open orders self._cancel_all_open_orders() # Liquidate positions for active symbols for symbol in list(self.active_symbols): try: if self.portfolio[symbol].quantity != 0: self.liquidate(symbol, tag="EOD cleanup") except Exception: pass # Remove non-SPY securities from universe for ticker, symbol in list(self.symbol_by_ticker.items()): if symbol != self._spy: try: self.remove_security(symbol) except Exception: pass # Reset state self._clear_state_but_keep_spy() self._prepare_close = False self.defer_entries_until = datetime(2100, 1, 1) self._entries_armed = False self._entries_submitted_today = set() self.debug(f"EOD cleanup completed at {self.time}") def _cancel_all_open_orders(self) -> None: # Attempt to cancel all entry and exit tickets for ticket in list(self.entry_tickets.values()): if ticket is not None and ticket.status not in [OrderStatus.FILLED, OrderStatus.CANCELED]: try: ticket.cancel("Daily reset") except Exception: pass for legs in list(self.exit_tickets.values()): for ticket in [legs.get("tp", None), legs.get("sl", None)]: if ticket is not None and ticket.status not in [OrderStatus.FILLED, OrderStatus.CANCELED]: try: ticket.cancel("Daily reset") except Exception: pass def _clear_state_but_keep_spy(self) -> None: # Preserve SPY, clear all other state self.today_targets = set() self.active_symbols = set() # Keep SPY mapping self.symbol_by_ticker = {"SPY": self._spy} self.entry_tickets = {} self.exit_tickets = {} self._premarket_highs = {} # ============================ DATA MANAGEMENT ============================ def _refresh_csv(self) -> None: try: raw = self.download(self._csv_url) if raw is None or len(raw) == 0: if len(self._csv_cache_raw) == 0: self.debug("CSV download failed and no cache available.") else: self.debug("CSV download failed; using cached data.") return self._csv_cache_raw = raw df = pd.read_csv(StringIO(raw)) df = self._normalize_columns(df) # Validate required columns required = {"date", "ticker", "premarkethigh"} if not required.issubset(set([c for c in df.columns])): self.debug(f"CSV missing required columns: {required - set([c for c in df.columns])}") return # Clean types df["ticker"] = df["ticker"].astype(str).str.upper().str.strip() # Coerce date; allow e.g., '2024-01-02' df["date"] = pd.to_datetime(df["date"], errors="coerce").dt.date # Coerce numeric premarket high df["premarkethigh"] = pd.to_numeric(df["premarkethigh"], errors="coerce") # Drop invalid rows df = df.dropna(subset=["date", "ticker", "premarkethigh"]).copy() df = df[df["premarkethigh"] > 0].copy() # Build by-date dictionary of per-day DataFrames (dedupe tickers per day) data_by_date: Dict[Any, pd.DataFrame] = {} for d, grp in df.groupby("date"): # Deduplicate by ticker, keeping the last occurrence grp = grp.sort_index().drop_duplicates(subset=["ticker"], keep="last").copy() data_by_date[d] = grp[["date", "ticker", "premarkethigh"]].copy() self.data_by_date = data_by_date self.debug(f"CSV loaded: {len(df)} rows, {len(self.data_by_date)} dates available") except Exception as e: self.debug(f"CSV processing error: {str(e)}") # Keep prior cache def _normalize_columns(self, df: pd.DataFrame) -> pd.DataFrame: # Normalize incoming column names: trim, lower, remove spaces/dashes def norm(s: str) -> str: s_2 = str(s).strip().lower() s_2 = re.sub(r"[\\s\\-]", "", s_2) return s_2 colmap: Dict[str, str] = {} for c in list(df.columns): k = norm(c) if k in ["date", "tradedate"]: colmap[c] = "date" elif k in ["ticker", "symbol", "tickername"]: colmap[c] = "ticker" elif k in ["premarkethigh", "premarket", "premarketh", "premkt", "premarkethighprice"]: colmap[c] = "premarkethigh" df = df.rename(columns=colmap) return df def _get_today_rows(self, d: date) -> List[Dict[str, Any]]: rows: List[Dict[str, Any]] = [] if d in self.data_by_date: day_df = self.data_by_date.get(d, pd.DataFrame()) if day_df is not None and len(day_df) > 0: for _, r in day_df.iterrows(): try: ticker = str(r["ticker"]).upper().strip() premkt = float(r["premarkethigh"]) if float(r["premarkethigh"]) > 0 else 0.0 if len(ticker) > 0 and premkt > 0: rows.append({"ticker": ticker, "premarket_high": premkt}) except Exception: continue return rows def _get_market_times_for_today(self): # Using SPY to anchor market times exchange = self.securities[self._spy].Exchange next_open = exchange.hours.get_next_market_open(self.time, False) next_close = exchange.hours.get_next_market_close(self.time, False) return next_open, next_close def _add_symbol_if_needed(self, ticker: str) -> Symbol: t = ticker.upper().strip() if t in self.symbol_by_ticker: return self.symbol_by_ticker[t] security = self.add_equity(t, Resolution.SECOND, extended_market_hours=False) security.set_data_normalization_mode(DataNormalizationMode.RAW) symbol = security.symbol self.symbol_by_ticker[t] = symbol return symbol # ================================ UTILITIES ============================== def _get_ticket_fill_price(self, ticket: OrderTicket) -> float: if ticket is None: return 0.0 avg = 0.0 try: fills = [e for e in ticket.order_events if e.fill_price is not None and e.fill_quantity is not None and e.fill_quantity != 0] if len(fills) > 0: # Compute a weighted average fill price total_qty = sum([abs(int(e.fill_quantity)) for e in fills]) if total_qty > 0: wsum = sum([abs(int(e.fill_quantity)) * float(e.fill_price) for e in fills]) avg = float(wsum / total_qty) if total_qty != 0 else 0.0 except Exception: avg = 0.0 return avg

Mia Alissi

STAFF Pro ,

Here’s a complete, ready-to-paste QuantConnect Python algorithm that implements Retrospective’s strategy exactly as requested:

from AlgorithmImports import * import pandas as pd import numpy as np from io import StringIO from datetime import datetime, timedelta, date from typing import Any, Dict, List, Set import re class PremarketBreakoutOCODailyAlgorithm(QCAlgorithm): """ PremarketBreakoutOCODailyAlgorithm

Sources: Order Tickets - QuantConnect.com, Empty Equity And Options400 Benchmark, Data Preparation - QuantConnect.com, docs writing algorithms securities asset classes us equity requesting data, Stop Market Orders - QuantConnect.com

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,200 Quants.

VOTE FOR UPCOMING FEATURES

Negative Values

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,200 Quants.

VOTE FOR UPCOMING FEATURES

Negative Values

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