#region imports
from AlgorithmImports import *
#endregion
class BaseOptionStrategy(QCAlgorithm):
name = ""
optionLegs = []
securityOptionLegs = []
expiryList = []
def __init__(self, name, optionLegs):
self.name = name
self.optionLegs = optionLegs
# Method that returns the number of days this strategy expires in. If we have multiple explirations we return an array.
def ExpiresIn(self, algo):
expirations = list(set(self.ExpiryList()))
if len(expirations) > 1:
return [(ex - algo.Time.date()).days for ex in expirations]
else:
return (expirations[0] - algo.Time.date()).days
def StrategyKey(self):
strikesStr = "_".join([str(x.StrikePrice if self.IsHolding() else x.Strike) for x in self.SecurityOptionLegs()])
return "{}_{}_{}".format(self.NameKey(), str(self.Expiration()), strikesStr)
def UnrealizedProfit(self):
if not self.IsHolding(): raise Exception("The {} strategy does not hold OptionHolding instances.".format(self.name))
return sum([c.UnrealizedProfitPercent for c in self.optionLegs]) / len(self.optionLegs) * 100
# Checks if the expiration is the same
def SameExpiration(self):
expirations = list(set(self.ExpiryList()))
if len(expirations) > 1:
return False
else:
return True
def Open(self, algo, quantity, log = True):
raise Exception("This method is not implemented")
def Close(self, algo):
algo.portfolio.RemoveTrade("{}s".format(self.NameKey()), self.StrategyKey())
for contract in self.optionLegs:
algo.Log("Liquidating {}".format(contract.Symbol))
algo.Liquidate(contract.Symbol, tag = self.StrategyKey())
def ExpiryList(self):
self.expiryList = self.expiryList or [x.Expiry.date() for x in self.SecurityOptionLegs()]
return self.expiryList
def Expiration(self):
return self.ExpiryList()[0]
def SecurityOptionLegs(self):
if self.IsHolding():
self.securityOptionLegs = self.securityOptionLegs or [x.Security for x in self.optionLegs]
else:
self.securityOptionLegs = self.securityOptionLegs or self.optionLegs
return self.securityOptionLegs
def IsHolding(self):
return isinstance(self.optionLegs[0], OptionHolding)
# TODO simplify this method to use SecurityOptionLegs
def Underlying(self):
if self.IsHolding():
return option.Security.Underlying.Symbol
else:
return option.UnderlyingSymbol
def NameKey(self):
return "".join(self.name.split())
class BullPutSpread(BaseOptionStrategy):
shortPut = None
longPut = None
def __init__(self, shortPut, longPut):
BaseOptionStrategy.__init__(self, "Bull Put Spread", [shortPut, longPut])
self.longPut = longPut
self.shortPut = shortPut
self.__StrikeCheck()
if self.SameExpiration() == False:
raise Exception("The expiration should be the same for all options.")
def Open(self, algo, quantity, log = True):
if log: algo.portfolio.AddTrade("{}s".format(self.NameKey()), self.StrategyKey())
algo.Log("Bull put Long Call {} and Short Call {}".format(self.longPut.Symbol, self.shortPut.Symbol))
algo.Buy(self.longPut.Symbol, quantity)
algo.Sell(self.shortPut.Symbol, quantity)
def __StrikeCheck(self):
if self.IsHolding():
longStrike = self.longPut.Security.StrikePrice
shortStrike = self.shortPut.Security.StrikePrice
else:
longStrike = self.longPut.Strike
shortStrike = self.shortPut.Strike
if longStrike > shortStrike:
raise Exception("The longPut strike has to be lower than the shortPut strike.")
class BearCallSpread(BaseOptionStrategy):
shortCall = None
longCall = None
def __init__(self, shortCall, longCall):
BaseOptionStrategy.__init__(self, "Bear Call Spread", [shortCall, longCall])
self.longCall = longCall
self.shortCall = shortCall
self.__StrikeCheck()
if self.SameExpiration() == False:
raise Exception("The expiration should be the same for all options.")
def Open(self, algo, quantity, log = True):
if log: algo.portfolio.AddTrade("{}s".format(self.NameKey()), self.StrategyKey())
algo.Log("Bear call Long Call {} and Short Call {}".format(self.longCall.Symbol, self.shortCall.Symbol))
algo.Buy(self.longCall.Symbol, quantity)
algo.Sell(self.shortCall.Symbol, quantity)
def __StrikeCheck(self):
if self.IsHolding():
longStrike = self.longCall.Security.StrikePrice
shortStrike = self.shortCall.Security.StrikePrice
else:
longStrike = self.longCall.Strike
shortStrike = self.shortCall.Strike
if longStrike < shortStrike:
raise Exception("The longCall strike has to be higher than the shortCall strike.")
# An iron condor is an options strategy consisting of two puts (one long and one short) and two calls (one long and one short), and four strike prices, all with the same expiration date.
# The iron condor earns the maximum profit when the underlying asset closes between the middle strike prices at expiration.
class IronCondor(BaseOptionStrategy):
bullPutSpread = None
bearCallSpread = None
def __init__(self, longCall, shortCall, longPut, shortPut):
BaseOptionStrategy.__init__(self, "Iron Condor", [longCall, shortCall, longPut, shortPut])
self.bullPutSpread = BullPutSpread(shortPut, longPut)
self.bearCallSpread = BearCallSpread(shortCall, longCall)
def Open(self, algo, quantity, log = True):
if log: algo.portfolio.AddTrade("{}s".format(self.NameKey()), self.StrategyKey())
self.bullPutSpread.Open(algo, quantity, False)
self.bearCallSpread.Open(algo, quantity, False)
from AlgorithmImports import *
from itertools import groupby
from OptionStrategies import BullPutSpread, BearCallSpread, IronCondor
class OptionsFinder:
def __init__(self, algo, underlying, rangeStart, rangeStop, minStrike, maxStrike):
self.algo = algo
self.underlying = underlying
self.rangeStart = rangeStart
self.rangeStop = rangeStop
self.minStrike = minStrike
self.maxStrike = maxStrike
# Right now the below lines could be conditional as we don't need them if we are not using
# the contractBid method where we look for credit and not really the strike.
self.underlyingOption = self.algo.AddOption(self.underlying, Resolution.Hour)
self.underlyingOption.SetFilter(
lambda u: u.IncludeWeeklys()
.Strikes(self.minStrike, self.maxStrike)
.Expiration(timedelta(self.rangeStart), timedelta(self.rangeStop))
)
self.underlyingOption.PriceModel = OptionPriceModels.CrankNicolsonFD()
# Just adds the options data.
def AddContract(self, strike = None, bid = None, optionType = OptionRight.Call):
''' We are adding the option contract data to the algo '''
# Replace this line below with different methods that filter the contracts in a more special way.
if strike is not None:
return self.__GetContractStrike(strike, optionType)
if bid is not None:
return self.__GetContractBid(bid, optionType)
def FindIronCondor(self, expiration, longCall, shortCall, longPut, shortPut):
# Check if there is options data for this symbol.
# Important that when we do GetValue we use the underlyingOption.Symbol because the
# key returned by the equity is `TSLA` and the one from option is `?TSLA`
chain = self.algo.sliceData.OptionChains.GetValue(self.underlyingOption.Symbol)
if chain is None: return None
PutSpread = None
CallSpread = None
# The way we are defining expiration here is by taking an absolute value. So it might just be __ExpiresIn(x) > expiration
contracts = [x for x in chain if self.__ExpiresIn(x) == expiration]
# only tradable contracts
# !!IMPORTANT!!: to escape the error `Backtest Handled Error: The security with symbol 'SPY 220216P00425000' is marked as non-tradable.`
contracts = [x for x in contracts if self.algo.Securities[x.Symbol].IsTradable]
# Make sure we have the contracts sorted by Strike and Expiry
contracts = sorted(contracts, key = lambda x: (x.Expiry, x.Strike))
calls = [x for x in contracts if x.Right == OptionRight.Call]
if len(calls) == 0: return None
soldCall = min(calls, key=lambda x: abs(x.Greeks.Delta - shortCall["delta"]))
# strike is above soldCall by value
boughtCall = min(calls, key=lambda x: abs(x.Strike - (soldCall.Strike + longCall["value"])))
if soldCall is not None and boughtCall is not None:
CallSpread = BearCallSpread(soldCall, boughtCall)
puts = [x for x in contracts if x.Right == OptionRight.Put]
if len(puts) == 0: return None
soldPut = min(puts, key=lambda x: abs(x.Greeks.Delta - shortPut["delta"]))
# strike is below soldPut by value
boughtPut = min(puts, key=lambda x: abs(x.Strike - (soldPut.Strike - longPut["value"])))
if soldPut is not None and boughtPut is not None:
PutSpread = BullPutSpread(soldPut, boughtPut)
if CallSpread is not None and PutSpread is not None:
return IronCondor(CallSpread.longCall, CallSpread.shortCall, PutSpread.longPut, PutSpread.shortPut)
return None
def FindBearCallSpread(self, expiration, shortCall, longCall):
# Check if there is options data for this symbol.
# Important that when we do GetValue we use the underlyingOption.Symbol because the
# key returned by the equity is `TSLA` and the one from option is `?TSLA`
chain = self.algo.sliceData.OptionChains.GetValue(self.underlyingOption.Symbol)
if chain is None: return None
# The way we are defining expiration here is by taking an absolute value. So it might just be __ExpiresIn(x) > expiration
contracts = [x for x in chain if self.__ExpiresIn(x) >= expiration]
# only tradable contracts
# !!IMPORTANT!!: to escape the error `Backtest Handled Error: The security with symbol 'SPY 220216P00425000' is marked as non-tradable.`
contracts = [x for x in contracts if self.algo.Securities[x.Symbol].IsTradable]
# Select only calls
contracts = [x for x in contracts if x.Right == OptionRight.Call]
# Make sure we have the contracts sorted by Strike and Expiry
contracts = sorted(contracts, key = lambda x: (x.Expiry, x.Strike))
# Iterate over the contracts that are grouped per day.
for expiry, group in groupby(contracts, lambda x: x.Expiry):
group = list(group)
# sort the options by credit
options = sorted(group, key = lambda x: x.Greeks.Delta)
soldCall = min(options, key=lambda x: abs(x.Greeks.Delta - shortCall["delta"]))
# strike is above soldCall by value
boughtCall = min(options, key=lambda x: abs(x.Strike - (soldCall.Strike + longCall["value"])))
if soldCall is not None and boughtCall is not None:
return BearCallSpread(soldCall, boughtCall)
return None
def FindBullPutSpread(self, expiration, shortPut, longPut):
# Check if there is options data for this symbol.
# Important that when we do GetValue we use the underlyingOption.Symbol because the
# key returned by the equity is `TSLA` and the one from option is `?TSLA`
chain = self.algo.sliceData.OptionChains.GetValue(self.underlyingOption.Symbol)
if chain is None: return None
# The way we are defining expiration here is by taking an absolute value. So it might just be __ExpiresIn(x) > expiration
contracts = [x for x in chain if self.__ExpiresIn(x) >= expiration]
# only tradable contracts
# !!IMPORTANT!!: to escape the error `Backtest Handled Error: The security with symbol 'SPY 220216P00425000' is marked as non-tradable.`
contracts = [x for x in contracts if self.algo.Securities[x.Symbol].IsTradable]
# Select only puts
contracts = [x for x in contracts if x.Right == OptionRight.Put]
# Make sure we have the contracts sorted by Strike and Expiry
contracts = sorted(contracts, key = lambda x: (x.Expiry, x.Strike))
# Iterate over the contracts that are grouped per day.
for expiry, group in groupby(contracts, lambda x: x.Expiry):
group = list(group)
# sort the options by credit
options = sorted(group, key = lambda x: x.Greeks.Delta)
soldPut = min(options, key=lambda x: abs(x.Greeks.Delta - shortPut["delta"]))
# strike is below soldPut by value
boughtPut = min(options, key=lambda x: abs(x.Strike - (soldPut.Strike - longPut["value"])))
if soldPut is not None and boughtPut is not None:
return BullPutSpread(soldPut, boughtPut)
return None
def ContractsWithCredit(self, bid):
# Check if there is options data for this symbol.
# Important that when we do GetValue we use the underlyingOption.Symbol because the
# key returned by the equity is `TSLA` and the one from option is `?TSLA`
chain = self.algo.sliceData.OptionChains.GetValue(self.underlyingOption.Symbol)
if chain is None: return None
callContract = None
putContract = None
# Don't look at contracts that are closer than the defined rangeStart.
contracts = [x for x in chain if self.__ExpiresIn(x) >= self.rangeStart and self.__ExpiresIn(x) <= self.rangeStop]
# Limit the Strikes to be between -5% and +5% of the UnderlyingPrice.
contracts = [x for x in contracts if x.UnderlyingLastPrice / 1.05 <= x.Strike <= x.UnderlyingLastPrice * 1.05 ]
# Make sure we have the contracts sorted by Strike and Expiry
contracts = sorted(contracts, key = lambda x: (x.Expiry, x.Strike))
# Iterate over the contracts that are grouped per day.
for expiry, group in groupby(contracts, lambda x: x.Expiry):
group = list(group)
# sort group by type
group = sorted(group, key = lambda x: x.Right)
if callContract is not None and putContract is not None:
return [callContract.Symbol, putContract.Symbol]
# Reset the contracts after each expiration day. We want to force the options to Expire on the same date.
callContract = None
putContract = None
for right, rightGroup in groupby(group, lambda x: x.Right):
rightGroup = list(rightGroup)
# sort the options by credit
creditOptions = sorted(rightGroup, key = lambda x: x.BidPrice, reverse = True)
if right == OptionRight.Call:
# find any contract that has a BidPrice > bid
creditCalls = [x for x in creditOptions if x.BidPrice > bid]
creditCalls = sorted(creditCalls, key = lambda x: x.BidPrice)
# if
# we do have a call that can replace the one we are buying for a credit then return that.
# else
# sort the calls by BidPrice and pick the highest bid contract
if len(creditCalls) > 0:
return [creditCalls[0].Symbol]
else:
# TODO: here instead of picking the bigest credit contract (this would be the one with the smallest Strike)
# we should consider picking one that is closer to strike as possible.
callContract = creditOptions[0]
# Only look for PUT contracts if we can't find a call contract with enough credit to replace the bid value.
if right == OptionRight.Put and callContract is not None:
# find any contract that has a BidPrice > bid - callContract.BidPrice
creditPuts = [x for x in creditOptions if x.BidPrice > (bid - callContract.BidPrice)]
creditPuts = sorted(creditPuts, key = lambda x: x.BidPrice)
if len(creditPuts) > 0:
putContract = creditPuts[0]
return []
# Method that returns a boolean if the security expires in the given days
# @param security [Security] the option contract
def __ExpiresIn(self, security):
return (security.Expiry.date() - self.algo.Time.date()).days
def __GetContractBid(self, bid, optionType):
# TODO: this method can be changed to return PUTs and CALLs if the strike selected is not
# sensible enough. Like don't allow selection of contracts furthen than 15% of the stock price.
# This should force the selection of puts to offset the calls or the other way around until we can get into
# a situation where just calls are present.
# Check if there is options data for this symbol.
# Important that when we do GetValue we use the underlyingOption.Symbol because the
# key returned by the equity is `TSLA` and the one from option is `?TSLA`
chain = self.algo.sliceData.OptionChains.GetValue(self.underlyingOption.Symbol)
if chain is None: return None
# Don't look at contracts that are closer than the defined rangeStart.
contracts = [x for x in chain if (x.Expiry.date() - self.algo.Time.date()).days > self.rangeStart]
contracts = sorted(
contracts,
key = lambda x: abs(x.UnderlyingLastPrice - x.Strike)
)
# Get the contracts that have more credit than the one we are buying. The bid here might be too small
# so we should consider sorting by:
# - strike > stock price
# - closest expiration
# - closest strike to stockPrice with credit! SORT THIS BELOW?!?
# TODO: right now we have a problem as it keeps going down in strike when it's loosing. Maybe send in the profit and if it's over -100 -200% switch to PUT?!
# TODO: consider using a trend indicator like EMA to determine a contract type switch CALL -> PUT etc.
contracts = [x for x in contracts if x.BidPrice >= bid * 1.20]
# If possible only trade the same contract type. This should have a limit of how low it should go before it switches to a put.
# Maybe by doing the limit we can remove the trend indicator.
if optionType == OptionRight.Call:
typeContracts = [x for x in contracts if x.Right == optionType and (x.UnderlyingLastPrice / 1.05) < x.Strike]
if len(typeContracts) > 0:
contracts = typeContracts
if optionType == OptionRight.Put:
typeContracts = [x for x in contracts if x.Right == optionType and (x.UnderlyingLastPrice * 1.05) < x.Strike]
if len(typeContracts) > 0:
contracts = typeContracts
# Grab us the contract nearest expiry
contracts = sorted(contracts, key = lambda x: x.Expiry)
if len(contracts) == 0:
return None
return contracts[0].Symbol
def __GetContractStrike(self, strike, optionType):
filtered_contracts = self.__DateOptionFilter()
if len(filtered_contracts) == 0: return str()
else:
contracts = self.__FindOptionsStrike(filtered_contracts, strike, optionType)
if len(contracts):
self.algo.AddOptionContract(contracts[0], Resolution.Daily)
return contracts[0]
else:
return str()
def __FindOptionsBid(self, contracts, bid, optionType = OptionRight.Call):
''' We are filtering based on bid price. '''
# select only the specified options types
contracts = [x for x in contracts if x.ID.OptionRight == optionType]
# TODO!! THis does not work. Most probably we have to do self.algo.AddOptionContract on all contracts above and then
#. use that to filter by BidPrice as the value does not seem to exist. :(
# pick a contract that is above the bid price provided.
contracts = [x for x in contracts if x.ID.BidPrice >= bid]
contracts = sorted(
contracts,
key = lambda x: abs(bid - x.ID.BidPrice)
)
# prefer shorter expirations
contracts = sorted(
contracts,
key = lambda x: x.ID.Date,
reverse=False
)
# sorted the contracts according to their expiration dates and choose the ATM options
return contracts
# Returns options that are ATM.
def __FindOptionsStrike(self, contracts, strike, optionType = OptionRight.Call):
''' We are filtering based on strike rank. '''
# select only the specified options types
contracts = [x for x in contracts if x.ID.OptionRight == optionType]
# never want to go beow strike
contracts = [x for x in contracts if x.ID.StrikePrice >= strike]
contracts = sorted(
contracts,
key = lambda x: abs(strike - x.ID.StrikePrice)
)
# prefer shorter expirations
contracts = sorted(
contracts,
key = lambda x: x.ID.Date,
reverse=False
)
# sorted the contracts according to their expiration dates and choose the ATM options
return contracts
# find the strike price of ATM option
# atm_strike = sorted(contracts,
# key = lambda x: abs(x.ID.StrikePrice - strike))
# atm_strike = atm_strike[0].ID.StrikePrice
# strike_list = sorted(set([i.ID.StrikePrice for i in contracts]))
# # find the index of ATM strike in the sorted strike list
# atm_strike_rank = strike_list.index(atm_strike)
# try:
# strikes = strike_list[(atm_strike_rank + self.minStrike):(atm_strike_rank + self.maxStrike)]
# except:
# strikes = strike_list
# filtered_contracts = [i for i in contracts if i.ID.StrikePrice in strikes]
# # select only call options
# call = [x for x in filtered_contracts if x.ID.OptionRight == optionType]
# # sorted the contracts according to their expiration dates and choose the ATM options
# return sorted(sorted(call, key = lambda x: abs(strike - x.ID.StrikePrice)),
# key = lambda x: x.ID.Date, reverse=True)
def __DateOptionFilter(self):
''' We are filtering the options based on the expiration date. It does return weekly contracts as well. '''
contracts = self.algo.OptionChainProvider.GetOptionContractList(self.underlying, self.algo.Time.date())
if len(contracts) == 0 : return []
# fitler the contracts based on the expiry range
contract_list = [i for i in contracts if self.rangeStart < (i.ID.Date.date() - self.algo.Time.date()).days < self.rangeStop]
return contract_list
# Returns options that can be rolled for a credit and higher strike
def __CreditUpOptions(self, contracts, existing_contract):
# TODO: this!
return []from AlgorithmImports import *
from OptionStrategies import BullPutSpread, BearCallSpread, IronCondor
import pickle
# Class that handles portfolio data. We have here any method that would search the portfolio for any of the contracts we need.
class PortfolioHandler:
def __init__(self, algo):
self.algo = algo
# Create a RollingWindow to store the last of the trades bids for selling options.
self.lastTradeBid = 0
# Returns all the covered calls of the specified underlying
# @param underlying [String]
# @param optionType [OptionRight.Call | OptionRight.Put]
# @param maxDays [Integer] number of days in the future that the contracts are filtered by
def UnderlyingSoldOptions(self, underlying, optionType, maxDays = 60):
contracts = []
for option in self.algo.Portfolio.Values:
security = option.Security
if (option.Type == SecurityType.Option and
str(security.Underlying) == underlying and
security.Right == optionType and
option.Quantity < 0 and
(security.Expiry.date() - self.algo.Time.date()).days < maxDays):
contracts.append(option)
return contracts
def OptionStrategies(self, underlying, types = [OptionRight.Call, OptionRight.Put]):
allContracts = {}
# select all the puts in our portfolio
for option in self.algo.Portfolio.Values:
security = option.Security
if (option.Type == SecurityType.Option and
security.Right in types and
str(security.Underlying) == underlying and
option.Quantity != 0):
allContracts.setdefault(int(security.Expiry.timestamp()), []).append(option)
return allContracts
def BullPutSpreads(self, underlying, ignoreStored = False):
# select all the puts in our portfolio
allContracts = self.OptionStrategies(underlying, [OptionRight.Put])
contracts = []
# if we have 2 contracts per expiration then we have a put spread. Let's filter for bull put spreads now.
# shortPut: higher strike than longPut // sold
# longPut: lower strike than shortPut // bought
for t, puts in allContracts.items():
# if we have 2 puts with equal quantities then we have a put spread
if len(puts) == 2 and sum(put.Quantity for put in puts) == 0:
shortPut = next(filter(lambda put: put.Quantity < 0, puts), None)
longPut = next(filter(lambda put: put.Quantity > 0, puts), None)
if shortPut.Security.StrikePrice > longPut.Security.StrikePrice:
contract = BullPutSpread(shortPut, longPut)
if not ignoreStored and contract.StrategyKey() not in self.ReadTrades("BullPutSpreads"): continue
contracts.append(contract)
return contracts
def BearCallSpreads(self, underlying, ignoreStored = False):
# select all the calls in our portfolio
allContracts = self.OptionStrategies(underlying, [OptionRight.Call])
contracts = []
# if we have 2 contracts per expiration then we have a call spread. Let's filter for bear call spreads now.
# shortCall: lower strike than longCall // sold
# longCall: higher strike than shortCall // bought
for t, calls in allContracts.items():
# if we have 2 calls with equal quantities then we have a call spread
if len(calls) == 2 and sum(call.Quantity for call in calls) == 0:
shortCall = next(filter(lambda call: call.Quantity < 0, calls), None)
longCall = next(filter(lambda call: call.Quantity > 0, calls), None)
if shortCall.Security.StrikePrice < longCall.Security.StrikePrice:
contract = BearCallSpread(shortCall, longCall)
if not ignoreStored and contract.StrategyKey() not in self.ReadTrades("BearCallSpreads"): continue
contracts.append(contract)
return contracts
def IronCondors(self, underlying, ignoreStored = False):
allContracts = self.OptionStrategies(underlying)
contracts = []
# if we have 4 allContracts per expiration then we have an iron condor
for t, c in allContracts.items():
if len(c) == 4:
calls = [call for call in c if call.Security.Right == OptionRight.Call]
puts = [put for put in c if put.Security.Right == OptionRight.Put]
# if we have 2 calls and 2 puts with equal quantities then we have a condor
if (len(calls) == 2 and
sum(call.Quantity for call in calls) == 0 and
len(puts) == 2 and
sum(put.Quantity for put in puts) == 0):
shortCall = next(filter(lambda call: call.Quantity < 0, calls), None)
longCall = next(filter(lambda call: call.Quantity > 0, calls), None)
shortPut = next(filter(lambda put: put.Quantity < 0, puts), None)
longPut = next(filter(lambda put: put.Quantity > 0, puts), None)
contract = IronCondor(longCall, shortCall, longPut, shortPut)
if not ignoreStored and contract.StrategyKey() not in self.ReadTrades("IronCondors"): continue
contracts.append(contract)
return contracts
# Updates the data in the ObjectStore to reflect the trades/strategies in the portfolio.
# @param symbol [Symbol]
# @param strategies [Array] // Eg: ["IronCondors", "BearCallSpreads"]
def SyncStored(self, symbol, strategies):
for strategy in strategies:
strategyKeys = [c.StrategyKey() for c in getattr(self, strategy)(symbol, True)]
self.update_ObjectStoreKey(strategy, strategyKeys)
# Removes all keys from the object store thus clearing all data.
def clear_ObjectStore(self):
keys = [str(j).split(',')[0][1:] for _, j in enumerate(self.algo.ObjectStore.GetEnumerator())]
for key in keys:
self.algo.ObjectStore.Delete(key)
# Updates the object store key with the new value without checking for the existing data.
# @param key [String]
# @param value [Array]
def update_ObjectStoreKey(self, key, value):
self.algo.ObjectStore.SaveBytes(str(key), pickle.dumps(value))
# Add trades to the object store like the following params
# @param key [String] // IronCondors
# @param value [OptionStrategy] // Eg: IronCondor
def AddTrade(self, key, value):
jsonObj = self.ReadTrades(key)
if value not in jsonObj: jsonObj.append(value)
self.algo.ObjectStore.SaveBytes(str(key), pickle.dumps(jsonObj))
# Remove trades from the object store by these params
# @param key [String] // IronCondors
# @param value [OptionStrategy] // Eg: IronCondor
def RemoveTrade(self, key, value):
jsonObj = self.ReadTrades(key)
jsonObj.remove(value)
self.algo.ObjectStore.SaveBytes(str(key), pickle.dumps(jsonObj))
def ReadTrades(self, key):
jsonObj = []
if self.algo.ObjectStore.ContainsKey(key):
deserialized = bytes(self.algo.ObjectStore.ReadBytes(key))
jsonObj = (pickle.loads(deserialized))
if jsonObj is None: jsonObj = []
jsonObj = list(set(jsonObj)) # there should be unique values in our array
return jsonObj
def PrintPortfolio(self):
# self.Debug("Securities:")
# self.Securities
# contains Securities that you subscribe to but it does not mean that you are invested.
# calling self.AddOptionContract will add the option to self.Securities
for kvp in self.Securities:
symbol = kvp.Key # key of the array
security = kvp.Value # value of the array (these are not attributes)
holdings = security.Holdings
self.Debug(str(security.Symbol))
# self.Debug(str(security.Underlying))
# self.Debug(str(security.Holdings))
# self.Debug("Portfolio:")
# self.Portfolio
# contains the Security objects that you are invested in.
for kvp in self.Portfolio:
symbol = kvp.Key
holding = kvp.Value
holdings = holding.Quantity
# self.Debug(str(holding.Holdings))
#region imports
from AlgorithmImports import *
#endregion
from Risk.MaximumDrawdownPercentPerSecurity import MaximumDrawdownPercentPerSecurity
import numpy as np
from PortfolioHandler import PortfolioHandler
class TrailingStop(RiskManagementModel):
'''Provides an implementation of IRiskManagementModel that limits the drawdown per holding to the specified percentage'''
def __init__(self, maximumDrawdownPercent = 0.05, profitTarget = None, symbol = None, strategies = []):
'''Initializes a new instance of the MaximumDrawdownPercentPerSecurity class
Args:
maximumDrawdownPercent: The maximum percentage drawdown allowed for any single security holding'''
self.maximumDrawdownPercent = -abs(maximumDrawdownPercent)
self.profitTarget = profitTarget
self.strategies = strategies
self.symbol = symbol
self.assetBestPnl = {}
def ManageRisk(self, algorithm, targets):
'''Manages the algorithm's risk at each time step
Args:
algorithm: The algorithm instance
targets: The current portfolio targets to be assessed for risk'''
targets = []
portfolio = PortfolioHandler(algorithm)
for strategy in self.strategies:
for contract in getattr(portfolio, strategy)(self.symbol):
key = contract.StrategyKey()
if key not in self.assetBestPnl.keys():
self.assetBestPnl[key] = contract.UnrealizedProfit()
self.assetBestPnl[key] = np.maximum(self.assetBestPnl[key], contract.UnrealizedProfit())
pnl = contract.UnrealizedProfit() - self.assetBestPnl[key]
# To handle profitTarget like 50% from when bought think of checking for
if self.profitTarget is not None:
if self.assetBestPnl[key] >= self.profitTarget and pnl < self.maximumDrawdownPercent:
for c in contract.optionLegs:
targets.append(PortfolioTarget(c.Symbol, 0))
else:
if pnl < self.maximumDrawdownPercent:
for c in contract.optionLegs:
targets.append(PortfolioTarget(c.Symbol, 0))
# for kvp in algorithm.Securities:
# security = kvp.Value
# symbol = security.Symbol.Value
# if symbol not in self.assetBestPnl.keys():
# self.assetBestPnl[symbol] = security.Holdings.UnrealizedProfitPercent
# self.assetBestPnl[symbol] = np.maximum(self.assetBestPnl[symbol], security.Holdings.UnrealizedProfitPercent)
# if not security.Invested:
# continue
# pnl = security.Holdings.UnrealizedProfitPercent - self.assetBestPnl[symbol]
# if pnl < self.maximumDrawdownPercent:
# # liquidate
# targets.append(PortfolioTarget(symbol, 0))
return targets
# region imports
from AlgorithmImports import *
from PortfolioHandler import PortfolioHandler
from OptionsFinder import OptionsFinder
from TrailingStop import TrailingStop
# endregion
# https://app.optionalpha.com/community/templates/post/10-my-ic-10delta-1dte-10wide-stop-hedge
# In 2020, I made just under $40,000 using iron condors on SPY in addition to working a job, with two 1DTE $1-wide trades per week and as much as
# 300 contracts per spread (usually $30-60k per iron condor). The initial capital allocation was less than $1,000, and I gradually deposited approximately
# $20,000 capital over the course of a year. This strategy was accomplished with 5 Delta, 2 standard deviation iron condors (spreads opened at 0.04,
# sold at 0.01 for 3% profit compounded twice a week) with a mean-reverting ATM hedge debit spread implemented in the event of a loss, and a loss was
# realized about every three months. The original IC was stopped out at a price of 0.50, aka when the short leg hit ATM. Yet this strategy experienced
# flaws in the 2021 market, as 2 standard deviation moves became increasingly prominent, with high volatility gaps occurring as often as twice a week in
# Q4 2021 and Q1 2022 regardless of implied volatility. Furthermore, one loss was equivalent to approximately 1 or 2 months of profit if left un-hedged.
# Today, I present to you the evolution of my methodology.
# But first, here is some background information.
# SCANNER:
# VARIABLES:
# symbol: SPY
# amount: up to 50% of bot net liq
# vix_below: 30
# vix_criteria: true|false // Turning VIX Criteria off bypasses the VIX Below restriction.
# expiration: 1 day
# entry_restriction: true|false // Turning this off bypasses the Tuesday or Thursday and timed entry criteria.
# long_call: 10$ above short call
# short_call: .10 delta
# short_put: -.10 delta
# long_put: 10$ below short put
# price: 100% of bid/ask // optionalpha smartpricing (not possible on quant)
# VIX is below 30 -- NO
# | YES
# SPY 14 day RSI intraday is above 30 and below 70 -- NO
# | YES
# (Today is Tuesday || Thursday) && (0 positions Short put spread && Short call spread && Iron condor) -- NO
# | YES
# Market time is 9:45am || 10:00am || 10:15am -- NO
# | YES
# Iron condor available (Symbol, Expiration, Short call, Long call, Short put, Long put / SPY, 1 day, .10 delta, 10$ above short call, -.10 delta, 10$ below short put) -- NO
# | YES
# Bot has available capital to open amount (50% of net liq) of iron condor at Price (100% of bid/ask) -- NO
# | YES
# OPEN IRON CONDOR
# MONITORS:
# -- IC | Closer w/ Hedge Opener
# VARIABLES:
# symbol: SPY
# amount: up to 50% of bot net liq
# hedge_expiration: 2 - 3 days
# long_call: 10$ above short call
# short_call: .30 delta
# short_put: -.30 delta
# long_put: 10$ below short put
# price: 100% of bid/ask // optionalpha smartpricing (not possible on quant)
# For each Iron condor position -- NO
# | YES
# Position premium decreased by 95% (95% profit)
# | YES | NO
# Close position |
# |
# (stock price > short call strike || stock price < short put strike) && current market after 12:00pm && position expires in 0 days
# | YES | NO
# | Current market time is after 03:15pm && position expires in 0 days -- NO
# | | YES
# | Close Position
# |
# -------------------------- stock price > short call strike
# | YES | NO
# ---------------------- ------------------------------------
# (symbol, hedge_expiration, short_call, long_call) avl -- NO stock price < short put strike -- NO
# | YES | YES
# Bot has capital for same quantity as Short call spread (to hedge) -- NO (symbol, hedge_expiration, short_put, long_put) avl -- NO
# | YES | YES
# Close Iron condor above -- NO Bot has capital for same quantity as Short put spread (to hedge) -- NO
# | YES | YES
# OPEN (symbol, hedge_expiration, short_call, long_call) Close Iron condor above -- NO
# | YES
# OPEN (symbol, hedge_expiration, short_put, long_put)
# -- CALL & PUT SPREAD HEDGE ADVANCED CLOSER (2x)
# VARIABLES:
# absolute_profit: 90%
# trailing_profit: True | False // Turn on to allow for Trailing Profit SmartStops that increase the probability of success.
# target: 50% // this is trailing profit and i assume it would mean cumulative profit?!?
# tail: 10%
# pdt: True | False // Turn on to wait 1 Market Day to close. Turn off to allow 0 DTE closing. Default now set to off.
# For each Short call spread position -- NO
# | YES
# Position premium decreased by absolute_profit (90% profit) || (trailing_profit && position trails a trailing % target by trailing_profit [50%])
# | YES | NO
# Close Position |
# ptd toggle is TRUE
# | YES | NO
# ------------------------ ------------------
# stock price > short call strike && position opened > 1 market day -- NO stock price > short call strike -- NO
# | YES | YES
# Close position Close position
# NOTES:
# - implement trailing stops https://medium.datadriveninvestor.com/cut-your-trading-losses-in-5-steps-on-quantconnect-ca7e2995a0ec
class IronPaycheck(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2021, 10, 1) # Set Start Date
self.SetEndDate(2022, 3, 1)
self.SetCash(100000) # Set Strategy Cash
symbol = self.GetParameter("symbol")
self.expiration = 1 # day
self.entryRestriction = True # Turning this off bypasses the Tuesday or Thursday and timed entry criteria.
equity = self.AddEquity(symbol, Resolution.Minute)
self.symbol = equity.Symbol
self.vixSymbol = self.AddIndex("VIX", Resolution.Minute).Symbol
self.VIX = 0
self.RSIValue = self.RSI(self.symbol, 14)
self.SetWarmUp(timedelta(30))
equity.SetDataNormalizationMode(DataNormalizationMode.Raw) # Not sure what this does yet
# To fix the missing price history for option contracts https://www.quantconnect.com/forum/discussion/8779/security-doesn-039-t-have-a-bar-of-data-trade-error-options-trading/p1
self.SetSecurityInitializer(lambda x: x.SetMarketPrice(self.GetLastKnownPrice(x)))
self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage) # not sure what this does
self.sliceData = None
self.portfolio = PortfolioHandler(self)
self.optionHandler = OptionsFinder(self, self.symbol, self.expiration, self.expiration + 1, -20, 20)
self.SetRiskManagement(TrailingStop(maximumDrawdownPercent = 0.1, symbol = self.symbol, strategies = ["BullPutSpreads", "BearCallSpreads"]))
# Sync up the ObjectStore with the portfolio to be sure we have the upmost data.
self.portfolio.SyncStored(self.symbol, ["IronCondors", "BullPutSpreads", "BearCallSpreads"])
# if not self.LiveMode: self.portfolio.clear_ObjectStore()
# (Today is Tuesday || Thursday)
# | YES
# Market time is 10:00am || 10:15am || 10:45am -- NO
# | YES
# !!IMPORTANT!!: to escape the error `Warning: fill at stale price (01/31/2022 16:00:00 America/New_York)` trade after 10:00am
scannerTimes = [self.TimeRules.At(10, 0), self.TimeRules.At(10, 15), self.TimeRules.At(10, 45)]
if self.entryRestriction:
dateRules = self.DateRules.Every(DayOfWeek.Tuesday, DayOfWeek.Thursday)
else:
dateRules = self.DateRules.EveryDay(self.symbol)
for time in scannerTimes:
self.Schedule.On(dateRules, time, self.Scanner)
# -- IC | Closer w/ Hedge Opener
self.Schedule.On(
self.DateRules.EveryDay(self.symbol), \
self.TimeRules.Every(timedelta(minutes=15)), \
self.MonitorIronCondor
)
# -- CALL & PUT SPREAD HEDGE ADVANCED CLOSER (2x)
self.Schedule.On(
self.DateRules.EveryDay(self.symbol), \
self.TimeRules.Every(timedelta(minutes=15)), \
self.MonitorHedges
)
def OnOrderEvent(self, orderEvent):
if orderEvent.IsAssignment:
order = self.Transactions.GetOrderById(orderEvent.OrderId)
if order.Type == OrderType.OptionExercise:
self.Log(str(order))
def OnAssignmentOrderEvent(self, assignmentEvent):
order = self.Transactions.GetOrderById(assignmentEvent.OrderId)
security = order.Symbol
underlyingPrice = self.Securities[self.symbol].Price
option = self.Securities[security.Value]
tag = "/stock price: {}".format(underlyingPrice)
tag += " /option: {}".format(security.Value)
tag += " /option strike: {}".format(option.StrikePrice)
tag += " /ext val: {}".format(option.Price - abs(option.StrikePrice - underlyingPrice))
# TODO: add in the tag all the details of why we had an assignment
# - strike difference (i think this is probably the cause 5%-10% probably max allowed)
# - extinsic value
# - profit value (here i might have to look into the combined profit value)
# - expires in days
if security.ID.OptionRight == OptionRight.Put:
self.MarketOrder(security.Underlying.Value, - order.AbsoluteQuantity * 100, False, tag)
# self.Sell(security.Underlying.Value, order.AbsoluteQuantity * 100)
elif security.ID.OptionRight == OptionRight.Call:
self.MarketOrder(security.Underlying.Value, order.AbsoluteQuantity * 100, False, tag)
# self.Buy(security.Underlying.Value, order.AbsoluteQuantity * 100)
self.Log(str(assignmentEvent))
# TODO see how we can add/remove orders here instead of manually. Right now we are updating the ObjectStore before the trade happens.
# def OnOrderEvent(self, orderEvent):
# order = self.Transactions.GetOrderById(orderEvent.OrderId)
# if orderEvent.Status == OrderStatus.Filled:
# self.Log("{0}: {1}: {2}".format(self.Time, order.Type, orderEvent))
def OnData(self, data: Slice):
if self.IsWarmingUp: return
self.sliceData = data
if data.ContainsKey(self.vixSymbol):
self.VIX = data[self.vixSymbol].Close
# self.Scanner(data)
# -- CALL & PUT SPREAD HEDGE ADVANCED CLOSER (2x)
def MonitorHedges(self):
if self.IsWarmingUp: return
# VARIABLES:
absoluteProfit = 90 # %
trailingProfit = True # | False // Turn on to allow for Trailing Profit SmartStops that increase the probability of success.
target = 50 # % // this is trailing profit and i assume it would mean cumulative profit?!?
# tail = 10 # %
pdt = False # | True // Turn on to wait 1 Market Day to close. Turn off to allow 0 DTE closing. Default now set to off.
stockPrice = self.Securities[self.symbol].Price
# For each Short call/put spread position -- NO
for callSpread in self.portfolio.BearCallSpreads(self.symbol):
# TODO: set a trailing profit algo?! That's a trailing stop loss set only after a certain profit target was reached
if callSpread.UnrealizedProfit() >= absoluteProfit: # or (trailingProfit and ):
callSpread.Close(self)
else:
if pdt:
if stockPrice > callSpread.shortCall.Security.StrikePrice and createdAt.day > 1:
# Close position
callSpread.Close(self)
else:
if stockPrice > callSpread.shortCall.Security.StrikePrice:
# Close position
callSpread.Close(self)
# For each Short call/put spread position -- NO
for putSpread in self.portfolio.BullPutSpreads(self.symbol):
# TODO: set a trailing profit algo?! That's a trailing stop loss set only after a certain profit target was reached
if putSpread.UnrealizedProfit() >= absoluteProfit: # or (trailingProfit and ):
putSpread.Close(self)
else:
if pdt:
# stock price < short put strike && position opened > 1 market day -- NO
if stockPrice < putSpread.shortPut.Security.StrikePrice and createdAt.day > 1:
# Close position
putSpread.Close(self)
else:
# stock price < short put
if stockPrice < putSpread.shortPut.Security.StrikePrice:
# Close position
putSpread.Close(self)
# -- IC | Closer w/ Hedge Opener
def MonitorIronCondor(self):
if self.IsWarmingUp: return
# VARIABLES:
# amount: up to 50% of bot net liq
hedgeExpiration = 2 - 3 # days
hedgeLongCall = {"value": 10} # $ above short call
hedgeShortCall = {"delta": .30} # delta
hedgeShortPut = {"delta": -.30} # delta
hedgeLongPut = {"value": 10} # $ below short put
# price: 100% of bid/ask // optionalpha smartpricing (not possible on quant)
# For each Iron condor position -- NO
# | YES
for ic in self.portfolio.IronCondors(self.symbol):
# Position premium decreased by 95% (95% profit)
# | YES | NO
# Close position |
if ic.UnrealizedProfit() > 95:
ic.Close(self)
continue
shortCall = ic.bearCallSpread.shortCall
shortPut = ic.bullPutSpread.shortPut
stockPrice = self.Securities[self.symbol].Price
expiresIn = ic.ExpiresIn(self)
if expiresIn != 0: return
# (stock price > short call strike || stock price < short put strike) && current market after 12:00pm && position expires in 0 days
if (stockPrice > shortCall.Security.StrikePrice or stockPrice < shortPut.Security.StrikePrice) and self.Time.hour >= 12 and expiresIn == 0:
# stock price > short call strike
if stockPrice > shortCall.Security.StrikePrice:
# (symbol, hedge_expiration, short_call, long_call) avl -- NO
hedgeCallSpread = self.optionHandler.FindBearCallSpread(expiration = hedgeExpiration, shortCall = hedgeShortCall, longCall = hedgeLongCall)
if hedgeCallSpread is not None:
# Bot has capital for same quantity as Short call spread (to hedge) -- NO
# Close Iron condor above -- NO
ic.Close(self)
# OPEN (symbol, hedge_expiration, short_call, long_call)
hedgeCallSpread.Open(self, self.OrderSize())
elif stockPrice < shortPut.Security.StrikePrice:
# (symbol, hedge_expiration, short_put, long_put) avl -- NO
hedgePutSpread = self.optionHandler.FindBullPutSpread(expiration = hedgeExpiration, shortPut = hedgeShortPut, longPut = hedgeLongPut)
if hedgePutSpread is not None:
# Bot has capital for same quantity as Short put spread (to hedge) -- NO
# Close Iron condor above -- NO
ic.Close(self)
# OPEN (symbol, hedge_expiration, short_put, long_put)
hedgePutSpread.Open(self, self.OrderSize())
elif self.Time.hour >= 15 and self.Time.minute >= 15 and expiresIn == 0:
# Current market time is after 15:15 && position expires in 0 days -- NO
ic.Close(self)
def Scanner(self):
if self.IsWarmingUp: return
# amount: up to 50% of bot net liq
vixBelow = 30
vixCriteria = False # Turning VIX Criteria off bypasses the VIX Below restriction.
# self.expiration = 1 # day
longCall = {"value": 10} # $ above short call
shortCall = {"delta": .10} # delta
shortPut = {"delta": -.10} # delta
longPut = {"value": 10} # $ below short put
# price: 100% of bid/ask // optionalpha smartpricing (not possible on quant)
if self.sliceData is None:
return
# VIX is below 30 -- NO
# | YES
if vixCriteria and self.VIX > vixBelow:
return
# SPY 14 day RSI intraday is above 30 and below 70 -- NO
# | YES
if 30 > self.RSIValue.Current.Value or self.RSIValue.Current.Value > 70:
return
# 0 positions Short put spread && Short call spread && Iron condor -- NO
# | YES
if len(self.portfolio.IronCondors(self.symbol)) > 0 or len(self.portfolio.BullPutSpreads(self.symbol)) > 0 or len(self.portfolio.BearCallSpreads(self.symbol)) > 0:
return
# Iron condor available (Symbol, Expiration, Short call, Long call, Short put, Long put / SPY, 1 day, .10 delta, 10$ above short call, -.10 delta, 10$ below short put) -- NO
# | YES
ironCondor = self.optionHandler.FindIronCondor(expiration = self.expiration, longCall = longCall, shortCall = shortCall, longPut = longPut, shortPut = shortPut)
if ironCondor is not None:
# Bot has available capital to open amount (50% of net liq) of iron condor at Price (100% of bid/ask) -- NO
# | YES
# OPEN IRON CONDOR
ironCondor.Open(self, self.OrderSize())
def OrderSize(self):
return 10
# leverage = self.Securities[self.symbol].Leverage
# margin = self.Portfolio.MarginRemaining
# orderSize = margin * leverage / self.Securities[self.symbol].Close
