Code is refusing to load options chains

Hello Everyone,

I am absolutely struggling to get the options to work on this platform, I seem to find absolutely no tutorials on options and its really annoying to work through.

I have my code below:

from AlgorithmImports import *
from collections import deque
import numpy as np
from datetime import timedelta
class SPYOptionsBot(QCAlgorithm):
    def Initialize(self):
        # Algorithm Settings
        self.SetStartDate(2023, 1, 1)
        self.SetEndDate(2023, 12, 31)
        self.SetCash(100000)
        
        # Add SPY Equity
        self.spy = self.AddEquity("SPY", Resolution.Minute).Symbol
        self.SetBenchmark(self.spy)
        # Add SPY Options
        self.option = self.AddOption("SPY", Resolution.Minute)
        
        # ---- Expiration changed to 1 to 2 weeks ----
        self.min_days_to_expiry = 7   # 1 week out
        self.max_days_to_expiry = 14  # 2 weeks out
        
        self.option.SetFilter(self.FilterOptions)
        # Parameters
        self.lookback_period       = 252   # daily bars for volatility calc
        self.required_bullish_days = 3     # consecutive bullish candles
        self.std_dev_multiplier    = 1.0
        self.max_positions         = 3     # max concurrent positions
        self.position_size         = 0.1   # 10% of portfolio value per trade
        # Track closing prices for bullish pattern detection
        self.previous_closes = deque(maxlen=self.required_bullish_days)
        # Keep track of open positions
        self.open_positions = {}
        # Schedule position management daily at 15:30
        self.Schedule.On(self.DateRules.EveryDay(), 
                         self.TimeRules.At(15, 30), 
                         self.ManagePositions)
    def FilterOptions(self, universe):
        """
        Filter for strikes near the current price, with expiration
        between 1-2 weeks (7 to 14 days).
        """
        return (universe
            .Strikes(-3, 3)
            .Expiration(self.min_days_to_expiry, self.max_days_to_expiry))
    def OnData(self, data):
        # We need SPY bar data to detect bullish pattern
        if self.spy not in data.Bars:
            return
        # Track consecutive bullish closes
        bar = data.Bars[self.spy]
        self.previous_closes.append(bar.Close)
        # Check for bullish pattern and capacity for new trades
        if (len(self.previous_closes) == self.required_bullish_days 
            and all(self.previous_closes[i] > self.previous_closes[i - 1]
                    for i in range(1, self.required_bullish_days))
            and len(self.open_positions) < self.max_positions):
            
            self.Log(f"Detected {self.required_bullish_days} bullish candles. Looking for put options to sell.")
            # Check if we have any option chain data in this slice
            if self.spy in data.OptionChains and len(data.OptionChains[self.spy]) > 0:
                option_chain = data.OptionChains[self.spy]
                self.SellPutOption(option_chain)
            else:
                self.Log("No option chain data is available for SPY at this moment.")
    def SellPutOption(self, option_chain):
        # Current price of the underlying equity
        underlying_price = self.Securities[self.spy].Price
        # Calculate historical volatility (std dev) over specified lookback
        history = self.History(self.spy, self.lookback_period, Resolution.Daily)
        if history.empty:
            self.Log("No historical data available to compute volatility.")
            return
        close_col = 'close' if 'close' in history.columns else 'Close'
        stddev = np.std(history[close_col])
        strike_price = underlying_price - (stddev * self.std_dev_multiplier)
        # Filter for put options that are out-of-the-money (<= strike_price)
        puts = [
            contract for contract in option_chain
            if contract.Right == OptionRight.Put
            and contract.Strike <= strike_price
            and (self.Time + timedelta(days=self.min_days_to_expiry)
                 <= contract.Expiry
                 <= self.Time + timedelta(days=self.max_days_to_expiry))
        ]
        if not puts:
            self.Log(f"No suitable put options found. Target strike: {strike_price}")
            return
        # Sort by "closeness" to strike, then volume (descending), then bid price
        puts.sort(key=lambda x: (
            abs(x.Strike - strike_price),
            -(x.Volume or 0),
            x.BidPrice
        ))
        selected_put = puts[0]
        max_position_value = self.Portfolio.TotalPortfolioValue * self.position_size
        
        # Each contract represents 100 shares
        if selected_put.BidPrice > 0:
            quantity = int(max_position_value // (selected_put.BidPrice * 100))
        else:
            quantity = 0
        if quantity > 0:
            self.Sell(selected_put.Symbol, quantity)
            self.open_positions[selected_put.Symbol] = {
                'entry_price': selected_put.BidPrice,
                'quantity': quantity,
                'entry_time': self.Time,
                'strike': selected_put.Strike
            }
            self.Log(f"Sold {quantity} contracts of {selected_put.Symbol} at {selected_put.BidPrice}")
        else:
            self.Log("Calculated position size is zero; not placing any order.")
    def ManagePositions(self):
        """
        Manage open positions daily at 15:30. Closes positions if:
        1) 50% profit is reached (option price <= 50% of entry)
        2) 100% loss is reached (option price >= 2x entry)
        3) Position is held longer than 15 days
        """
        positions_to_close = []
        for symbol, position in list(self.open_positions.items()):
            if symbol not in self.Securities:
                self.Log(f"Symbol {symbol} not found in Securities.")
                continue
            current_price = self.Securities[symbol].Price
            entry_price   = position['entry_price']
            days_held     = (self.Time - position['entry_time']).days
            # Closing conditions
            if (current_price <= entry_price * 0.5    # 50% profit
                or current_price >= entry_price * 2   # 100% loss
                or days_held >= 15):                  # Time-based exit
                self.Buy(symbol, position['quantity'])
                positions_to_close.append(symbol)
                self.Log(f"Closing position {symbol} after {days_held} days. "
                         f"Entry: {entry_price}, Exit: {current_price}")
        # Remove closed positions from tracking
        for symbol in positions_to_close:
            del self.open_positions[symbol]
    def OnOrderEvent(self, orderEvent):
        if orderEvent.Status == OrderStatus.Filled:
            self.Log(f"Order {orderEvent.OrderId} filled: "
                     f"{orderEvent.Symbol} at {orderEvent.FillPrice}")
    def OnEndOfAlgorithm(self):
        self.Log(f"Algorithm ended. Final portfolio value: "
                 f"${self.Portfolio.TotalPortfolioValue:,.2f}")
+ Expand
- Collapse

Any help would be so appreciated!

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.

from AlgorithmImports import *
from collections import deque
import numpy as np
from datetime import timedelta
class SPYOptionsBot(QCAlgorithm):
def initialize(self) -> None:
# Algorithm Settings
self.set_start_date(2023, 1, 1)
self.set_end_date(2023, 12, 31)
self.set_cash(100000)
# Add SPY Equity
self.spy = self.add_equity("SPY", Resolution.MINUTE).Symbol
self.set_benchmark(self.spy)
# Add SPY Options
self.option = self.add_option("SPY", Resolution.MINUTE)
# Expiration changed to 1 to 2 weeks
self.min_days_to_expiry = 7 # 1 week out
self.max_days_to_expiry = 14 # 2 weeks out
self.option.set_filter(self.filter_options)
# Parameters
self.lookback_period = 252 # daily bars for volatility calc
self.required_bullish_days = 3 # consecutive bullish candles
self.std_dev_multiplier = 1.0
self.max_positions = 3 # max concurrent positions
self.position_size = 0.1 # 10% of portfolio value per trade
# Track closing prices for bullish pattern detection
self.previous_closes: deque = deque(maxlen=self.required_bullish_days)
# Keep track of open positions
self.open_positions: dict = {}
# Schedule position management daily at 15:30
self.schedule.on(self.date_rules.every_day(),
self.time_rules.at(15, 30),
self.manage_positions)
def filter_options(self, universe: OptionFilterUniverse) -> OptionFilterUniverse:
"""
Filter for strikes near the current price, with expiration
between 1-2 weeks (7 to 14 days).
"""
return (universe
.strikes(-3, 3)
.expiration(self.min_days_to_expiry, self.max_days_to_expiry))
def on_data(self, data: Slice) -> None:
# We need SPY bar data to detect bullish pattern
if self.spy not in data.bars:
return
# Track consecutive bullish closes
bar = data.bars[self.spy]
self.previous_closes.append(bar.close)
# Check for bullish pattern and capacity for new trades
if (len(self.previous_closes) == self.required_bullish_days
and all(self.previous_closes[i] > self.previous_closes[i - 1]
for i in range(1, self.required_bullish_days))
and len(self.open_positions) < self.max_positions):
self.log(f"Detected {self.required_bullish_days} bullish candles. Looking for put options to sell.")
# Check if we have any option chain data in this slice
if self.spy in data.option_chains and data.option_chains[self.spy]:
option_chain = data.option_chains[self.spy]
self.sell_put_option(option_chain)
else:
self.log("No option chain data is available for SPY at this moment.")
def sell_put_option(self, option_chain: OptionChain) -> None:
# Current price of the underlying equity
underlying_price = self.securities[self.spy].price
# Calculate historical volatility (std dev) over specified lookback
history = self.history(self.spy, self.lookback_period, Resolution.DAILY)
if history.empty:
self.log("No historical data available to compute volatility.")
return
close_col = 'close' if 'close' in history.columns else 'Close'
stddev = np.std(history[close_col])
strike_price = underlying_price - (stddev * self.std_dev_multiplier)
# Filter for put options that are out-of-the-money (<= strike_price)
puts = [
contract for contract in option_chain
if contract.right == OptionRight.PUT
and contract.strike <= strike_price
and (self.time + timedelta(days=self.min_days_to_expiry)
<= contract.expiry
<= self.time + timedelta(days=self.max_days_to_expiry))
]
if not puts:
self.log(f"No suitable put options found. Target strike: {strike_price}")
return
# Sort by "closeness" to strike, then volume (descending), then bid price
puts.sort(key=lambda x: (
abs(x.strike - strike_price),
-(x.volume or 0),
x.bid_price
))
selected_put = puts[0]
max_position_value = self.portfolio.total_portfolio_value * self.position_size
# Each contract represents 100 shares
if selected_put.bid_price > 0:
quantity = int(max_position_value // (selected_put.bid_price * 100))
else:
quantity = 0
if quantity > 0:
self.sell(selected_put.symbol, quantity)
self.open_positions[selected_put.symbol] = {
'entry_price': selected_put.bid_price,
'quantity': quantity,
'entry_time': self.time,
'strike': selected_put.strike
}
self.log(f"Sold {quantity} contracts of {selected_put.symbol} at {selected_put.bid_price}")
else:
self.log("Calculated position size is zero; not placing any order.")
def manage_positions(self) -> None:
"""
Manage open positions daily at 15:30. Closes positions if:
1) 50% profit is reached (option price <= 50% of entry)
2) 100% loss is reached (option price >= 2x entry)
3) Position is held longer than 15 days
"""
positions_to_close = []
for symbol, position in list(self.open_positions.items()):
if symbol not in self.securities:
self.log(f"Symbol {symbol} not found in Securities.")
continue
current_price = self.securities[symbol].price
entry_price = position['entry_price']
days_held = (self.time - position['entry_time']).days
# Closing conditions
if (current_price <= entry_price * 0.5 # 50% profit
or current_price >= entry_price * 2 # 100% loss
or days_held >= 15): # Time-based exit
self.buy(symbol, position['quantity'])
positions_to_close.append(symbol)
self.log(f"Closing position {symbol} after {days_held} days. "
f"Entry: {entry_price}, Exit: {current_price}")
# Remove closed positions from tracking
for symbol in positions_to_close:
del self.open_positions[symbol]
def on_order_event(self, order_event: OrderEvent) -> None:
if order_event.status == OrderStatus.FILLED:
self.log(f"Order {order_event.order_id} filled: "
f"{order_event.symbol} at {order_event.fill_price}")
def on_end_of_algorithm(self) -> None:
self.log(f"Algorithm ended. Final portfolio value: "
f"${self.portfolio.total_portfolio_value:,.2f}")

+ Expand

Domenico Bezuidenhout INVESTOR

December 2024

I keep getting chain wont load errors

Upvote

Mia Alissi INVESTOR

from AlgorithmImports import *
from collections import deque
import numpy as np
from datetime import timedelta
class SPYOptionsBot(QCAlgorithm):
    def initialize(self) -> None:
        # Algorithm Settings
        self.set_start_date(2023, 1, 1)
        self.set_end_date(2023, 12, 31)
        self.set_cash(100000)
        # Add SPY Equity
        self.spy = self.add_equity("SPY", Resolution.MINUTE).Symbol
        self.set_benchmark(self.spy)
        # Add SPY Options
        self.option = self.add_option("SPY", Resolution.MINUTE)
        # Expiration changed to 1 to 2 weeks
        self.min_days_to_expiry = 7   # 1 week out
        self.max_days_to_expiry = 14  # 2 weeks out
        self.option.set_filter(self.filter_options)
        # Parameters
        self.lookback_period = 252   # daily bars for volatility calc
        self.required_bullish_days = 3     # consecutive bullish candles
        self.std_dev_multiplier = 1.0
        self.max_positions = 3     # max concurrent positions
        self.position_size = 0.1   # 10% of portfolio value per trade
        # Track closing prices for bullish pattern detection
        self.previous_closes: deque = deque(maxlen=self.required_bullish_days)
        # Keep track of open positions
        self.open_positions: dict = {}
        # Schedule position management daily at 15:30
        self.schedule.on(self.date_rules.every_day(), 
                         self.time_rules.at(15, 30), 
                         self.manage_positions)
    def filter_options(self, universe: OptionFilterUniverse) -> OptionFilterUniverse:
        """
        Filter for strikes near the current price, with expiration
        between 1-2 weeks (7 to 14 days).
        """
        return (universe
            .strikes(-3, 3)
            .expiration(self.min_days_to_expiry, self.max_days_to_expiry))
    def on_data(self, data: Slice) -> None:
        # We need SPY bar data to detect bullish pattern
        if self.spy not in data.bars:
            return
        # Track consecutive bullish closes
        bar = data.bars[self.spy]
        self.previous_closes.append(bar.close)
        # Check for bullish pattern and capacity for new trades
        if (len(self.previous_closes) == self.required_bullish_days 
            and all(self.previous_closes[i] > self.previous_closes[i - 1]
                    for i in range(1, self.required_bullish_days))
            and len(self.open_positions) < self.max_positions):
            self.log(f"Detected {self.required_bullish_days} bullish candles. Looking for put options to sell.")
            # Check if we have any option chain data in this slice
            if self.spy in data.option_chains and data.option_chains[self.spy]:
                option_chain = data.option_chains[self.spy]
                self.sell_put_option(option_chain)
            else:
                self.log("No option chain data is available for SPY at this moment.")
    def sell_put_option(self, option_chain: OptionChain) -> None:
        # Current price of the underlying equity
        underlying_price = self.securities[self.spy].price
        # Calculate historical volatility (std dev) over specified lookback
        history = self.history(self.spy, self.lookback_period, Resolution.DAILY)
        if history.empty:
            self.log("No historical data available to compute volatility.")
            return
        close_col = 'close' if 'close' in history.columns else 'Close'
        stddev = np.std(history[close_col])
        strike_price = underlying_price - (stddev * self.std_dev_multiplier)
        # Filter for put options that are out-of-the-money (<= strike_price)
        puts = [
            contract for contract in option_chain
            if contract.right == OptionRight.PUT
            and contract.strike <= strike_price
            and (self.time + timedelta(days=self.min_days_to_expiry)
                 <= contract.expiry
                 <= self.time + timedelta(days=self.max_days_to_expiry))
        ]
        if not puts:
            self.log(f"No suitable put options found. Target strike: {strike_price}")
            return
        # Sort by "closeness" to strike, then volume (descending), then bid price
        puts.sort(key=lambda x: (
            abs(x.strike - strike_price),
            -(x.volume or 0),
            x.bid_price
        ))
        selected_put = puts[0]
        max_position_value = self.portfolio.total_portfolio_value * self.position_size
        # Each contract represents 100 shares
        if selected_put.bid_price > 0:
            quantity = int(max_position_value // (selected_put.bid_price * 100))
        else:
            quantity = 0
        if quantity > 0:
            self.sell(selected_put.symbol, quantity)
            self.open_positions[selected_put.symbol] = {
                'entry_price': selected_put.bid_price,
                'quantity': quantity,
                'entry_time': self.time,
                'strike': selected_put.strike
            }
            self.log(f"Sold {quantity} contracts of {selected_put.symbol} at {selected_put.bid_price}")
        else:
            self.log("Calculated position size is zero; not placing any order.")
    def manage_positions(self) -> None:
        """
        Manage open positions daily at 15:30. Closes positions if:
        1) 50% profit is reached (option price <= 50% of entry)
        2) 100% loss is reached (option price >= 2x entry)
        3) Position is held longer than 15 days
        """
        positions_to_close = []
        for symbol, position in list(self.open_positions.items()):
            if symbol not in self.securities:
                self.log(f"Symbol {symbol} not found in Securities.")
                continue
            current_price = self.securities[symbol].price
            entry_price = position['entry_price']
            days_held = (self.time - position['entry_time']).days
            # Closing conditions
            if (current_price <= entry_price * 0.5    # 50% profit
                or current_price >= entry_price * 2   # 100% loss
                or days_held >= 15):                  # Time-based exit
                self.buy(symbol, position['quantity'])
                positions_to_close.append(symbol)
                self.log(f"Closing position {symbol} after {days_held} days. "
                         f"Entry: {entry_price}, Exit: {current_price}")
        # Remove closed positions from tracking
        for symbol in positions_to_close:
            del self.open_positions[symbol]
    def on_order_event(self, order_event: OrderEvent) -> None:
        if order_event.status == OrderStatus.FILLED:
            self.log(f"Order {order_event.order_id} filled: "
                     f"{order_event.symbol} at {order_event.fill_price}")
    def on_end_of_algorithm(self) -> None:
        self.log(f"Algorithm ended. Final portfolio value: "
                 f"${self.portfolio.total_portfolio_value:,.2f}")
+ Expand
- Collapse

Sources: Upcoming Holiday Momentum for Amazon, Index Option Bull Put Spread Algorithm, Automating the Wheel Strategy, Index Option Bear Put Spread Algorithm, Head & Shoulders TA Pattern Detection

Platform

Radically Open-Source Algorithmic Trading Engine

Join Our Discord Channel

Quarterly Open-Source Trading Competition

Draft Discussions

Bookmarked Discussions

SEARCH DISCUSSIONS

359,700 Quants.

VOTE FOR UPCOMING FEATURES

JOIN OUR Community MAILING LIST

IN THIS RESEARCH

PARTICIPANTS

Actions

Join QuantConnect for Free