| Overall Statistics |
|
Total Trades 157 Average Win 0.22% Average Loss -0.58% Compounding Annual Return -2.651% Drawdown 6.800% Expectancy -0.078 Net Profit -3.498% Sharpe Ratio -0.436 Probabilistic Sharpe Ratio 1.861% Loss Rate 33% Win Rate 67% Profit-Loss Ratio 0.38 Alpha 0 Beta 0 Annual Standard Deviation 0.04 Annual Variance 0.002 Information Ratio -0.436 Tracking Error 0.04 Treynor Ratio 0 Total Fees $157.00 Estimated Strategy Capacity $5000.00 Lowest Capacity Asset TSLA XY1PTCHZ72TI|TSLA UNU3P8Y3WFAD |
from System.Drawing import Color
from AlgorithmImports import *
class Benchmark:
def __init__(self, algo, underlying, shares = 100, indicators = {}):
self.algo = algo
self.underlying = underlying
# Variable to hold the last calculated benchmark value
self.benchmarkCash = None
self.benchmarkShares = shares
self.tradingChart = Chart('Trade Plot')
# On the Trade Plotter Chart we want 3 series: trades and price:
for s in ["Call", "Put", "Straddle"]:
self.tradingChart.AddSeries(Series("Sell {}".format(s), SeriesType.Scatter, '$', Color.Green, ScatterMarkerSymbol.Circle))
self.tradingChart.AddSeries(Series("Buy {}".format(s), SeriesType.Scatter, '$', Color.Red, ScatterMarkerSymbol.Triangle))
self.tradingChart.AddSeries(Series('Price', SeriesType.Line, '$', Color.White))
self.algo.AddChart(self.tradingChart)
self.AddIndicators(indicators)
self.resample = datetime.min
self.resamplePeriod = (self.algo.EndDate - self.algo.StartDate) / 2000
def AddIndicators(self, indicators):
self.indicators = indicators
for name, indicator in indicators.items():
self.algo.AddChart(Chart(name))
def PrintBenchmark(self):
if self.algo.Time <= self.resample: return
self.resample = self.algo.Time + self.resamplePeriod
# self.__PrintBuyHold()
self.__PrintTrades()
self.__PrintCash()
self.__PrintIndicators()
def PrintTrade(self, order, trade, quantity):
''' Prints the price of the option on our trade chart. '''
plotTradeName = ''
tradeStrike = trade.Strike()
if isinstance(tradeStrike, list):
if order.Symbol.ID.OptionRight == OptionRight.Put:
tradeName = "{0} Put".format(trade.name)
elif order.Symbol.ID.OptionRight == OptionRight.Call:
tradeName = "{0} Call".format(trade.name)
tradeStrike = order.Symbol.ID.StrikePrice
else:
tradeName = trade.name
if quantity < 0:
plotTradeName = 'Sell {}'.format(tradeName)
else:
plotTradeName = 'Buy {}'.format(tradeName)
self.algo.Plot('Trade Plot', plotTradeName, tradeStrike)
def __PrintIndicators(self):
''' Prints the indicators array values to the Trade Plot chart. '''
for name, indicator in self.indicators.items():
if name == 'BB':
self.__PlotBB(indicator)
elif name == 'AROON':
self.__PlotAROON(indicator)
elif name == 'MACD':
self.__PlotMACD(indicator)
else:
self.algo.PlotIndicator(name, indicator)
def __PlotBB(self, indicator):
self.algo.Plot('BB', 'Price', self.__UnderlyingPrice())
self.algo.Plot('BB', 'BollingerUpperBand', indicator.UpperBand.Current.Value)
self.algo.Plot('BB', 'BollingerMiddleBand', indicator.MiddleBand.Current.Value)
self.algo.Plot('BB', 'BollingerLowerBand', indicator.LowerBand.Current.Value)
def __PlotMACD(self, indicator):
# self.algo.Plot('MACD', 'MACD', indicator.Current.Value)
# self.algo.Plot('MACD', 'MACDSignal', indicator.Signal.Current.Value)
# self.algo.Plot("MACD", "Price", self.__UnderlyingPrice())
self.algo.Plot("MACD", "Zero", 0)
self.algo.Plot('MACD', 'MACDSignal', indicator.Signal.Current.Value)
def __PlotAROON(self, indicator):
self.algo.Plot('AROON', 'Aroon', indicator.Current.Value)
self.algo.Plot('AROON', 'AroonUp', indicator.AroonUp.Current.Value)
self.algo.Plot('AROON', 'AroonDown', indicator.AroonDown.Current.Value)
def __PrintCash(self):
''' Prints the cash in the portfolio in a separate chart. '''
self.algo.Plot('Cash', 'Options gain', self.algo.Portfolio.Cash)
def __PrintTrades(self):
''' Prints the underlying price on the trades chart. '''
self.algo.Plot('Trade Plot', 'Price', self.__UnderlyingPrice())
def __PrintBuyHold(self):
''' Simulate buy and hold the shares. We use the same number of shares as the backtest.
In this situation is 100 shares + the cash of the portfolio.'''
if not self.benchmarkCash:
self.benchmarkCash = self.algo.Portfolio.TotalPortfolioValue - self.benchmarkShares * self.__UnderlyingPrice()
self.algo.Plot("Strategy Equity", "Buy & Hold", self.benchmarkCash + self.benchmarkShares * self.__UnderlyingPrice())
def __UnderlyingPrice(self):
return self.algo.Securities[self.underlying].Close#region imports from AlgorithmImports import * #endregion #region imports from .Benchmark import Benchmark #endregion # Your New Python File
#region imports
from AlgorithmImports import *
#endregion
class MarketHours:
def __init__(self, algorithm, symbol):
self.algorithm = algorithm
self.hours = algorithm.Securities[symbol].Exchange.Hours
def get_CurrentOpen(self):
return self.hours.GetNextMarketOpen(self.algorithm.Time, False)
def get_CurrentClose(self):
return self.hours.GetNextMarketClose(self.get_CurrentOpen(), False)#region imports
from AlgorithmImports import *
from itertools import groupby
from QuantConnect.Logging import *
from MarketHours import MarketHours
from PortfolioHandler import Handler
from PortfolioHandler.OStrategies import Straddle
#endregion
# TODO:
# - force it to work once an hour
# - force it to work every 5 minutes
# - force it to work once a day
# - check if LEAP is properly rolled so it does not pick a too close one. Ideal is 365 days DTE!
# - try to roll for a credit including the LEAP value
# - when monitoring not just close the position also roll and leave the scanner just for opening positions. (not expected to cause any results if we include credit checks)
# -
# TODO:
# Replicate the result from this spreadsheet: https://docs.google.com/spreadsheets/d/1-Pr-_arX_mdM2O_oDMHnhFZNRBti4N07/edit#gid=2038006255
# Details of best result
# Run Seq. Run Date Code Version Portfolio Return CAGR Percent Change Max Drawdown (%) Max Drawdown ($) Opening Trades Winners Losers Avg Gain (Winners) Avg Loss (Losers) Geometric Sharpe Ratio Baseline
# v2.0038 8/25/2020 2.1 962.83% 32.81% 91.54% -36.35% -$282,701.00 322 211 111 $13,975.38 -$17,891.65 0.955 Loss
# Environmental Backtest Settings Sequence Parameters
# Start Date End Date Initial Cash Fill Orders Close Orders Allocation Straddle Quantity
# 4/2/2012 7/31/2020 $100,000 Mid Price Mid Price 0.50 0
# DTE Threshold Price Threshold Open DTE Min Open DTE Max Price Var Min Price Var Max
# 35 0.065 335 425 -$0.50 $0.50
# DTE Threshold Loss Limit Threshold Price Call Threshold Price Put Threshold Open DTE Min Open DTE Max DTE Max Expansion Open DTE Multiplier Price Var Min Price Var Max Price Markup
# 1 -3.25 0.050 0.050 5 15 30 1.05 -$0.50 $0.50 $0.10
# FINAL FORM: seems like short rolling works great it's the LEAP that gives too little profit (is that the way it needs to be??!)
# THIS DOES NOT SEEM TO WORK!!!
class CustomBuyingPowerModel(BuyingPowerModel):
def GetMaximumOrderQuantityForTargetBuyingPower(self, parameters):
quantity = super().GetMaximumOrderQuantityForTargetBuyingPower(parameters).Quantity
quantity = np.floor(quantity / 100) * 100
return GetMaximumOrderQuantityResult(quantity)
def HasSufficientBuyingPowerForOrder(self, parameters):
return HasSufficientBuyingPowerForOrderResult(True)
class MilkTheCowAlphaModel(AlphaModel):
algorithm = None
def __init__(self, algorithm, ticker, option):
self.ticker = ticker
self.option = option
self.algorithm = algorithm
self.symbol = algorithm.AddEquity(self.ticker)
self.marketHours = MarketHours(algorithm, self.ticker)
self.portfolio = Handler(self.algorithm)
# self.symbol.SetBuyingPowerModel(CustomBuyingPowerModel())
# TEST default expirations:
# self.LeapExpiration = ExpirationRange(335, 425)
# self.ShortExpiration = ExpirationRange(5, 15)
# TEST rolling expirations:
# self.LeapExpiration = ExpirationRange(335, 425)
# self.ShortExpiration = ExpirationRange(5, 30)
self.LeapExpiration = ExpirationRange(335, 425)
self.ShortExpiration = ExpirationRange(5, 30)
self.Log = Logger(self.algorithm)
self.Credit = TradeCredit()
def Update(self, algorithm, data):
insights = []
if algorithm.IsWarmingUp: return insights
if self.ticker not in data.Keys: return insights
self.algorithm.benchmark.PrintBenchmark()
if self.marketHours.get_CurrentClose().hour - 2 > self.algorithm.Time.hour > self.marketHours.get_CurrentOpen().hour + 2:
if self.algorithm.Time.minute == 30:
insights.extend(self.Monitor(data))
if self.algorithm.Time.minute == 5:
insights.extend(self.Scanner(data))
# if self.algorithm.Time.hour > self.marketHours.get_CurrentOpen().hour + 4 and self.algorithm.Time.minute == 15:
# insights.extend(self.Monitor(data))
# if 11 > self.algorithm.Time.hour >= 10 and (self.algorithm.Time.minute == 15 or self.algorithm.Time.minute == 30 or self.algorithm.Time.minute == 45):
# insights.extend(self.Scanner(data))
return Insight.Group(insights)
def Monitor(self, data):
insights = []
stockPrice = self.algorithm.Securities[self.ticker].Price
longStraddles = self.portfolio.Straddles(self.ticker, expiration = self.LeapExpiration.ToArr(), short = False)
shortStraddles = self.portfolio.Straddles(self.ticker, expiration = self.ShortExpiration.ToArr(), short = True)
#### EXIT RULES
# Exiting the strategy is defined as closing both straddles and is triggered by one of two conditions.
# - Long LEAP Straddle DTE < 335
# - Short Straddle indicates over a 325% loss; anything less should just be adjusted, see below.
if len(longStraddles) > 0 and len(shortStraddles) > 0:
longStraddle = longStraddles[0]
shortStraddle = shortStraddles[0]
close = False
# if longStraddle.ExpiresIn(self.algorithm) < 335:
# close = True
# self.Log.Add("EXIT - longStraddle {}: expires in {} < 335".format(longStraddle.ToString(self.algorithm), longStraddle.ExpiresIn(self.algorithm) ))
if shortStraddle.UnrealizedProfit() <= -325:
close = True
self.Log.Add("EXIT - shortStraddle {}: unrealized profit {} <= -335".format(shortStraddle.ToString(self.algorithm), shortStraddle.UnrealizedProfit() ))
if close:
# I'm selling the leap
self.Credit.Sell(leap = longStraddle.AskPrice(self.algorithm))
# And buying the short
self.Credit.Buy(short = shortStraddle.AskPrice(self.algorithm))
self.Log.Add("EXIT - with credit {} / leap(${}) short(${})".format(self.Credit.ToString(), longStraddle.AskPrice(self.algorithm), shortStraddle.AskPrice(self.algorithm)))
return [
Insight.Price(longStraddle.Call.Symbol, Resolution.Minute, 15, InsightDirection.Flat),
Insight.Price(longStraddle.Put.Symbol, Resolution.Minute, 15, InsightDirection.Flat),
Insight.Price(shortStraddle.Call.Symbol, Resolution.Minute, 15, InsightDirection.Flat),
Insight.Price(shortStraddle.Put.Symbol, Resolution.Minute, 15, InsightDirection.Flat)
]
self.Log.Print()
#### LONG LEAP STRADDLE
# ##### Standard Roll Triggers:
# - Less than 335 days expiration (DTE).
# - When underlying price moves more than 6.5% from strike price.
# - When profit on LEAP straddle is > 0.5%
# __Note: IV Opportunity Rolling no longer recommended.__
# __When rolling, select strikes ATM and DTE = 365 +/- 30 (335 – 395).__
if len(longStraddles) > 0:
longStraddle = longStraddles[0]
close = False
if longStraddle.ExpiresIn(self.algorithm) < 335:
close = True
self.Log.Add("ROLL LEAP - longStraddle {}: expires in {} < 335".format(longStraddle.ToString(self.algorithm), longStraddle.ExpiresIn(self.algorithm) ))
if (longStraddle.Strike() / 1.065) > stockPrice or stockPrice > (longStraddle.Strike() * 1.065):
close = True
self.Log.Add("ROLL LEAP - longStraddle {0}: strike / 1.065({1}) > {2} or {2} > strike * 1.065({3})".format(longStraddle.ToString(self.algorithm), (longStraddle.Strike() / 1.065), stockPrice, (longStraddle.Strike() * 1.065)))
# if longStraddle.UnrealizedProfit() > 0.5:
# close = True
# self.Log.Add("ROLL LEAP - longStraddle {}: unrealized profit {} > 0.5%".format(longStraddle.ToString(self.algorithm), longStraddle.UnrealizedProfit() ))
if close:
# I'm selling the leap
self.Credit.Sell(leap = longStraddle.AskPrice(self.algorithm))
self.Log.Add("ROLL LEAP - sell trigger with credit {} / leap(${})".format(self.Credit.ToString(), longStraddle.AskPrice(self.algorithm)))
insights.extend(
[
Insight.Price(longStraddle.Call.Symbol, Resolution.Minute, 15, InsightDirection.Flat),
Insight.Price(longStraddle.Put.Symbol, Resolution.Minute, 15, InsightDirection.Flat)
]
)
self.Log.Print()
# #### SHORT STRADDLE
# ##### Standard Roll Triggers:
# - 1 day to expiration (DTE).
# - When underlying price moves more than 5% from strike price.
# - 2-5 days to expiration (DTE) AND spot price is $1.00 or less away from the strike price.
# - When profit on short straddle is > 20%
# __When rolling, select strikes ATM and select the minimum DTE to yield a net credit, ideally 7 DTE, but with a maximum DTE of 30.__
# > - LEAP: Roll when the underlying price moves more than 6.5% from strike price.
# > - When rolling, select strikes ATM and DTE 335-425 days out.
# > - Short: Roll 1 day prior to expiration or when underlying price moves more than 5% from the strike price.
# > - When rolling select strikes ATM and select the minimum DTE to yield a net credit (max of 30 DTE).
if len(shortStraddles) > 0:
shortStraddle = shortStraddles[0]
close = False
if shortStraddle.ExpiresIn(self.algorithm) <= 1:
close = True
self.Log.Add("ROLL SHORT - shortStraddle {}: expires in {} <= 1".format(shortStraddle.ToString(self.algorithm), shortStraddle.ExpiresIn(self.algorithm) ))
if (shortStraddle.Strike() / 1.05) > stockPrice or stockPrice > (shortStraddle.Strike() * 1.05):
close = True
self.Log.Add("ROLL SHORT - shortStraddle {0}: strike / 1.05({1}) > {2} or {2} > strike * 1.05({3})".format(shortStraddle.ToString(self.algorithm), (shortStraddle.Strike() / 1.05), stockPrice, (shortStraddle.Strike() * 1.05)))
if 2 <= shortStraddle.ExpiresIn(self.algorithm) <= 5 and ((shortStraddle.Strike() - 1) > stockPrice or stockPrice > (shortStraddle.Strike() + 1)):
close = True
self.Log.Add("ROLL SHORT - shortStraddle {0}: strike - 1 ({1}) > {2} or {2} > strike + 1({3})".format(shortStraddle.ToString(self.algorithm), (shortStraddle.Strike() - 1), stockPrice, (shortStraddle.Strike() + 1)))
if shortStraddle.UnrealizedProfit() > 20:
close = True
self.Log.Add("ROLL SHORT - shortStraddle {}: unrealized profit {} > 20%".format(shortStraddle.ToString(self.algorithm), shortStraddle.UnrealizedProfit() ))
if close:
# And buying the short
self.Credit.Buy(short = shortStraddle.AskPrice(self.algorithm))
self.Log.Add("ROLL SHORT - buy trigger with credit {} / short(${})".format(self.Credit.ToString(), shortStraddle.AskPrice(self.algorithm)))
insights.extend(
[
Insight.Price(shortStraddle.Call.Symbol, Resolution.Minute, 15, InsightDirection.Flat),
Insight.Price(shortStraddle.Put.Symbol, Resolution.Minute, 15, InsightDirection.Flat)
]
)
self.Log.Print()
return insights
def Scanner(self, data):
insights = []
lenShortStraddle = len(self.portfolio.Straddles(self.ticker, expiration = [0, self.ShortExpiration.Stop], short = True))
lenLeapStraddle = len(self.portfolio.Straddles(self.ticker, expiration = self.LeapExpiration.ToArr(), short = False))
# SCAN LEAP
# - if no LEAP
if lenLeapStraddle == 0:
longStraddle = self.__FindStraddle(data, expiration = self.LeapExpiration, idealDTE=365)
self.Log.Add("SCANNER LEAP - Trying to find a LEAP")
if longStraddle:
self.Credit.Buy(leap = longStraddle.AskPrice(self.algorithm))
self.Log.Add("SCANNER LEAP - buying LEAP {} with credit {} / ${}".format(longStraddle.ToString(self.algorithm), self.Credit.ToString(), longStraddle.AskPrice(self.algorithm)) )
insights.extend([
Insight.Price(longStraddle.Call, Resolution.Minute, 15, InsightDirection.Up),
Insight.Price(longStraddle.Put, Resolution.Minute, 15, InsightDirection.Up),
])
self.Log.Print()
# SCAN SHORT
# - if LEAP and no SHORT
# First open the leapStraddle and after that the shortStraddle!
if lenShortStraddle == 0 and lenLeapStraddle > 0:
shortStraddle = self.__FindStraddle(data, expiration = self.ShortExpiration, minCredit = self.Credit.LastShort * 1.3)
self.Log.Add("SCANNER SHORT - Trying to find a SHORT")
if shortStraddle:
self.Credit.Sell(short = shortStraddle.AskPrice(self.algorithm))
self.Log.Add("SCANNER SHORT - selling SHORT {} with credit {} / ${}".format(
shortStraddle.ToString(self.algorithm), self.Credit.ToString(), shortStraddle.AskPrice(self.algorithm) ) )
insights.extend([
Insight.Price(shortStraddle.Call, Resolution.Minute, 15, InsightDirection.Down),
Insight.Price(shortStraddle.Put, Resolution.Minute, 15, InsightDirection.Down),
])
self.Log.Print()
return insights
def __FindStraddle(self, data, expiration, minCredit = 0.0, idealDTE = None):
put = None
call = None
stockPrice = self.algorithm.Securities[self.ticker].Price
contracts = self.algorithm.OptionChainProvider.GetOptionContractList(self.ticker, self.algorithm.Time.date())
if len(contracts) == 0 : return None
# # 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.algorithm.Securities[x.ID.Symbol].IsTradable]
contracts = [i for i in contracts if expiration.Start <= (i.ID.Date.date() - self.algorithm.Time.date()).days <= expiration.Stop]
if not contracts: return None
contracts = sorted(contracts, key = lambda x: x.ID.Date, reverse = False)
# Pick all the ATM contracts per day
ATMcontracts = []
for expiry, g in groupby(contracts, lambda x: x.ID.Date):
# Sort by strike as grouping without sorting does not work.
dateGroup = sorted(list(g), key = lambda x: x.ID.StrikePrice, reverse = False)
strikeGroup = []
for strike, sg in groupby(dateGroup, lambda x: x.ID.StrikePrice):
sg = list(sg)
# only select groups that have 2 contracts
if len(sg) == 2:
# assign if (doing basically a min/sorting function):
# - no group is added in the strike group
# - previous strike price - stock price difference is bigger than current strike price - stock price difference
if len(strikeGroup) == 0 or (abs(stockPrice - strikeGroup[0].ID.StrikePrice) > abs(stockPrice - strike)):
strikeGroup = sg
GroupPut = next(filter(lambda option: option.ID.OptionRight == OptionRight.Put, strikeGroup), None)
GroupCall = next(filter(lambda option: option.ID.OptionRight == OptionRight.Call, strikeGroup), None)
ATMcontracts.extend([GroupCall, GroupPut])
contracts = ATMcontracts
# add all the option contracts so we can access all the data.
for c in contracts:
self.algorithm.AddOptionContract(c, Resolution.Minute)
# WARNING!! we have to select AskPrice like this in a separate list because otherwise it seems like the filtering might be too fast or the pointers are messed
# up in python that by adding the option contract right above this could cause issues where AskPrice might be 0!!
puts = [[self.algorithm.Securities[c.Value].AskPrice, c] for c in contracts if c.ID.OptionRight == OptionRight.Put]
put = min(puts, key=lambda x: abs(x[0] - minCredit / 2))[1]
# only select calls that have the same date as our put above for extra insurance that it's a straddle.
calls = [[self.algorithm.Securities[c.Value].AskPrice, c] for c in contracts if c.ID.OptionRight == OptionRight.Call and c.ID.Date == put.ID.Date]
call = min(calls, key=lambda x: abs(x[0] - minCredit / 2))[1]
# if we have an ideal DTE defined then try and compare that with the one selected
if idealDTE != None and self.__ExpiresIn(put) < idealDTE :
idealPut = min(puts, key=lambda x: abs((x[1].ID.Date.date() - self.algorithm.Time.date()).days - idealDTE))[1]
idealCall = [c for c in contracts if c.ID.Date == idealPut.ID.Date and c.ID.OptionRight == OptionRight.Call][0]
# minPremium = self.algorithm.Securities[call.Value].AskPrice + self.algorithm.Securities[put.Value].AskPrice
# idealPremium = self.algorithm.Securities[idealCall.Value].AskPrice + self.algorithm.Securities[idealPut.Value].AskPrice
call = idealCall
put = idealPut
if not put or not call: return None
if put.ID.StrikePrice != call.ID.StrikePrice: return None
return Straddle(put, call)
# 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.ID.Date.date() - self.algorithm.Time.date()).days
def __SignalDeltaPercent(self):
return (self.indicators['MACD'].Current.Value - self.indicators['MACD'].Signal.Current.Value) / self.indicators['MACD'].Fast.Current.Value
class ExpirationRange:
def __init__(self, start, stop):
self.Start = start
self.Stop = stop
def ToArr(self):
return [self.Start, self.Stop]
# Class to hold the credit of our trade so we can separate LEAP profit/credit from short profit but also combine them when needed.
class TradeCredit:
Value = 0
Leap = 0
Short = 0
LastShort = 0
LastLeap = 0
# When we buy something we remove from credit as we are paying from balance to get the options.
def Buy(self, short = 0, leap = 0):
self.Short -= short
self.Leap -= leap
if short != 0: self.LastShort = short
self.Value -= short + leap
return self.Value
# When we are selling we add to credit as we are receiving premium for selling them.
def Sell(self, short = 0, leap = 0):
self.Short += short
self.Leap += leap
if leap != 0: self.LastLeap = leap
self.Value += short + leap
return self.Value
def ToString(self):
return "Total: ${} / Short: ${} (last ${}) / Leap: ${}".format(round(self.Value, 2) * 100, round(self.Short, 2) * 100, round(self.LastShort, 2), round(self.Leap, 2) * 100)
# Make this logger class so we can print logging messages in section for easier read.
class Logger:
Messages = []
Algorithm = None
def __init__(self, algorithm):
self.Algorithm = algorithm
def Add(self, message):
self.Messages.append(message)
def Print(self):
if len(self.Messages) > 0:
self.Algorithm.Log("------***------")
for m in self.Messages:
self.Algorithm.Log(m)
self.Algorithm.Log("------|||------")
self.Messages = []# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
# Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from AlgorithmImports import *
from Selection.OptionUniverseSelectionModel import OptionUniverseSelectionModel
class MilkTheCowOptionSelectionModel(OptionUniverseSelectionModel):
'''Creates option chain universes that select only the two week earliest expiry call contract
and runs a user defined optionChainSymbolSelector every day to enable choosing different option chains'''
def __init__(self, select_option_chain_symbols, expirationRange = [7, 30]):
super().__init__(timedelta(1), select_option_chain_symbols)
self.expirationRange = expirationRange
def Filter(self, filter):
'''Defines the option chain universe filter'''
return (filter.Strikes(-2, +2)
.Expiration(self.expirationRange[0], self.expirationRange[1])
.IncludeWeeklys()
.OnlyApplyFilterAtMarketOpen())from AlgorithmImports import *
class OptionsSpreadExecution(ExecutionModel):
'''Execution model that submits orders while the current spread is tight.
Note this execution model will not work using Resolution.Daily since Exchange.ExchangeOpen will be false, suggested resolution is Minute
'''
def __init__(self, acceptingSpreadPercent=0.005):
'''Initializes a new instance of the SpreadExecutionModel class'''
self.targetsCollection = PortfolioTargetCollection()
# Gets or sets the maximum spread compare to current price in percentage.
self.acceptingSpreadPercent = Math.Abs(acceptingSpreadPercent)
self.executionTimeThreshold = timedelta(minutes = 10)
self.openExecutedOrders = {}
def Execute(self, algorithm, targets):
'''Executes market orders if the spread percentage to price is in desirable range.
Args:
algorithm: The algorithm instance
targets: The portfolio targets'''
# update the complete set of portfolio targets with the new targets
self.UniqueTargetsByStrategy(algorithm, targets)
self.ResetExecutedOrders(algorithm)
# for performance we check count value, OrderByMarginImpact and ClearFulfilled are expensive to call
if self.targetsCollection.Count > 0:
for target in self.targetsCollection.OrderByMarginImpact(algorithm):
symbol = target.Symbol
if not self.TimeToProcess(algorithm, symbol): continue
# calculate remaining quantity to be ordered
unorderedQuantity = OrderSizing.GetUnorderedQuantity(algorithm, target)
# check order entry conditions
if unorderedQuantity != 0:
# get security information
security = algorithm.Securities[symbol]
# TODO: check this against spreads.!!
# if we are selling or buying an option then pick a favorable price
# if we are trying to get out of the trade then execute at market price
# if (target.Quantity != 0 and self.SpreadIsFavorable(security)) or target.Quantity == 0:
stockPrice = security.Underlying.Price
algorithm.MarketOrder(symbol, unorderedQuantity, tag = "Current stock price {0}".format(stockPrice))
self.openExecutedOrders[symbol.Value] = algorithm.Time
self.targetsCollection.ClearFulfilled(algorithm)
def TimeToProcess(self, algorithm, symbol):
# if we executed the market order less than the executionTimeThreshold then skip
key = symbol.Value
openOrders = algorithm.Transactions.GetOpenOrders(symbol)
if key in self.openExecutedOrders.keys():
if (self.openExecutedOrders[key] + self.executionTimeThreshold) > algorithm.Time:
return False
else:
# cancel existing open orders for symbol and try again
algorithm.Transactions.CancelOpenOrders(key, "The order did not fill in the expected threshold.")
return True
else:
# Order was never processed for this Symbol.
return True
def ResetExecutedOrders(self, algorithm):
# attempt to clear targets that have been filled later
self.targetsCollection.ClearFulfilled(algorithm)
# reset openExecutedOrders if no targets present
if self.targetsCollection.Count == 0:
self.openExecutedOrders = {}
# TODO: improve this for insight Groups so we can do spreads
def UniqueTargetsByStrategy(self, algorithm, targets):
# check newly added targets for similar option strategies that have not been filled
for target in targets:
symbol = target.Symbol
targetSecurity = algorithm.Securities[symbol]
# TODO this does not work when we are trying to roll leap straddles and at the same time sell short straddles as they have the same quantity.
if symbol.SecurityType == SecurityType.Option:
# if an old strategy has not been filled then we are going to remove it and allow the new similar one to be tried.
for t in self.targetsCollection:
tSecurity = algorithm.Securities[t.Symbol]
if (t.Symbol.SecurityType == symbol.SecurityType and \
tSecurity.Right == targetSecurity.Right and \
t.Quantity == target.Quantity and \
tSecurity.Expiry == targetSecurity.Expiry):
self.targetsCollection.Remove(t.Symbol)
self.targetsCollection.Add(target)
def SpreadIsFavorable(self, security):
'''Determines if the spread is in desirable range.'''
# Price has to be larger than zero to avoid zero division error, or negative price causing the spread percentage < 0 by error
# Has to be in opening hours of exchange to avoid extreme spread in OTC period
return security.Exchange.ExchangeOpen \
and security.Price > 0 and security.AskPrice > 0 and security.BidPrice > 0 \
and (security.AskPrice - security.BidPrice) / security.Price <= self.acceptingSpreadPercent
from AlgorithmImports import *
from OStrategies import *
import pickle
from itertools import chain
# Class that handles portfolio data. We have here any method that would search the portfolio for any of the contracts we need.
class Handler:
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/calls 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 Straddles(self, underlying, ignoreStored = False, expiration = [3, 750], short = None):
allContracts = self.OptionStrategies(underlying)
contracts = []
for t, options in allContracts.items():
# if we have 2 contracts and they are short
if len(options) != 2:
continue
# - short = None: add straddle
# - short = True and Quantity > 0: continue
# - short = True and Quantity < 0: add short straddle
# - short = False and Quantity > 0: add long straddle
# - short = False and Quantity < 0: continue
if short != None:
if short == True and sum(o.Quantity for o in options) > 0:
continue
if short == False and sum(o.Quantity for o in options) < 0:
continue
# WARNING! THIS ASSUMES WE HAVE OPTIONS ON A CERTAIN DAY THAT ARE EITHER SOLD OR BOUGHT AND JUST ONCE SET PER
# UNDERLYING. WE MIGHT HAVE TO UPDATE THIS
# pick the put and the call
Put = next(filter(lambda option: option.Security.Right == OptionRight.Put, options), None)
Call = next(filter(lambda option: option.Security.Right == OptionRight.Call, options), None)
if not Put or not Call:
continue
# check expiration if it's in range.
if expiration[1] < (Put.Security.Expiry.date() - self.algo.Time.date()).days < expiration[0]:
continue
# if it's a straddle (both strikes are the same)
if Put.Security.StrikePrice == Call.Security.StrikePrice:
contract = Straddle(Put, Call)
if not ignoreStored and contract.StrategyKey() not in self.ReadTrades("Straddles"): continue
contracts.append(contract)
return contracts
def SoldPuts(self, underlying, ignoreStored = False, expiration = [0, 30]):
# select all the calls in our portfolio
allContracts = self.OptionStrategies(underlying, [OptionRight.Put])
contracts = []
for t, puts in allContracts.items():
# if it's more or less than 1 call in the group (per date) skip
if len(puts) != 1:
continue
Put = puts[0]
# if the quantity of the contracts is not sold skip
if Put.Quantity >= 0:
continue
# check expiration if it's in range.
if expiration[1] < (Put.Security.Expiry.date() - self.algo.Time.date()).days < expiration[0]:
continue
contract = SoldPut(Put)
if not ignoreStored and contract.StrategyKey() not in self.ReadTrades("SoldPuts"): continue
contracts.append(contract)
return contracts
def SoldCalls(self, underlying, ignoreStored = False, expiration = [0, 30]):
# select all the calls in our portfolio
allContracts = self.OptionStrategies(underlying, [OptionRight.Call])
contracts = []
for t, calls in allContracts.items():
# if it's more or less than 1 call in the group (per date) skip
if len(calls) != 1:
continue
Call = calls[0]
# if the quantity of the contracts is not sold skip
if Call.Quantity >= 0:
continue
# check expiration if it's in range.
if expiration[1] < (Call.Security.Expiry.date() - self.algo.Time.date()).days < expiration[0]:
continue
contract = SoldCall(Call)
if not ignoreStored and contract.StrategyKey() not in self.ReadTrades("SoldCalls"): continue
contracts.append(contract)
return contracts
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:
# TODO replace the OptionStrategies with the existing Lean code classes.
# OptionStrategies.BullPutSpread(canonicalOption, shortPut.Security.StrikePrice, longPut.Security.StrikePrice, shortPut.Security.Expiry)
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
def FindStrategy(self, key, symbol, strategies = ["IronCondors", "BearCallSpreads", "BullPutSpreads"]):
for strategy in strategies:
contract = next(filter(lambda contract: contract.StrategyKey() == key, getattr(self, strategy)(symbol, True)))
if contract: return contract
return None
def FixAssignment(self, order, strategies = ["Straddles"]):
security = order.Symbol
underlying = security.Underlying.Value
trade = self.GetStoredTrade(security, strategies = strategies)
if not trade: return
# Only sold options can be assigned the bought one expire worthless.
# Sell or buy the shares for the current order
if security.ID.OptionRight == OptionRight.Put:
self.algo.MarketOrder(underlying, - order.AbsoluteQuantity * 100, False)
elif security.ID.OptionRight == OptionRight.Call:
self.algo.MarketOrder(underlying, order.AbsoluteQuantity * 100, False)
portfolioOptions = chain.from_iterable(self.OptionStrategies(underlying).values())
portfolioOptions = [c.Symbol.Value for c in portfolioOptions]
# sell/buy the other options in the strategy/trade
for contract in trade.optionLegs:
contractKey = contract.Value
if contractKey in portfolioOptions:
self.algo.Liquidate(contractKey, tag = "Liquidating {0} from {1}".format(contractKey, trade.StrategyKey()))
return trade
def GetStoredTrade(self, security, strategies = ["Straddles"]):
orderKey = "".join(security.Value.split())
for strategy in strategies:
trades = self.ReadTrades(strategy)
for t in trades:
tradeElements = t.split("_")
strategyName = tradeElements[0]
# select all keys except first that would be the StrategyName
strategyKeys = tradeElements[1:]
if orderKey in strategyKeys:
optionValues = [c.replace(security.Underlying.Value, security.Underlying.Value + " ") for c in strategyKeys]
return eval(strategyName)(*[self.algo.Securities[o].Symbol for o in optionValues])
return None
#
# @param order [OrderEvent]
# @param strategies [Array] // Eg: ["IronCondors", "BearCallSpreads"]
def GetCurrentTrade(self, order, strategies = ["Straddles"]):
symbol = order.Symbol
security = symbol.ID
# get all trades of all strategies
for strategy in strategies:
contracts = getattr(self, strategy)(security.Symbol, True)
for t in contracts:
if symbol.Value in t.StrategyKeys():
return t
return None
# 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, ignoreStored = 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
# TODO update this to use the LEAN versions of the strategies and expand on them. Maybe we don't even have to do that. OPEN and CLOSE are not needed!
class BaseOptionStrategy(QCAlgorithm):
name = ""
optionLegs = []
securityOptionLegs = []
expiryList = []
def __init__(self, name, optionLegs):
self.name = name
self.optionLegs = optionLegs
def ToString(self, algo):
strikeDiffStr = ""
if not isinstance(self.Strike(), list):
difference = round( ( ( self.Strike() - self.UnderlyingPrice(algo) ) / self.UnderlyingPrice(algo) ) * 100, 2)
strikeDiffStr = " - (${underlyingSymbol}) ${underlying} = {difference}%".format(underlyingSymbol = self.Underlying(), underlying = self.UnderlyingPrice(algo), difference = difference)
return "{name}(${strike}{strikeDiffStr}; {expiration}; Exp. {expiresIn} days)".format(
name = self.name, strike = self.Strike(), strikeDiffStr = strikeDiffStr, expiration = self.Expiration(), expiresIn = self.ExpiresIn(algo)
)
# 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):
keysStr = "_".join(["".join(o.split()) for o in self.StrategyKeys()])
return "{}_{}".format(self.NameKey(), keysStr)
def StrategyKeys(self):
if self.IsContract():
ids = [o.Value for o in self.optionLegs]
else:
ids = [o.Symbol.Value for o in self.SecurityOptionLegs()]
return ids
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):
algo.portfolio.AddTrade("{}s".format(self.NameKey()), self.StrategyKeys())
def Close(self, algo):
algo.portfolio.RemoveTrade("{}s".format(self.NameKey()), self.StrategyKeys())
def ExpiryList(self):
if self.IsContract():
exList = [x.Date.date() for x in self.SecurityOptionLegs()]
else:
exList = [x.Expiry.date() for x in self.SecurityOptionLegs()]
self.expiryList = self.expiryList or exList
return self.expiryList
def Expiration(self):
return self.ExpiryList()[0]
def AskPrice(self, algo):
if self.IsContract():
prices = [algo.Securities[o.Value].AskPrice for o in self.optionLegs]
else:
prices = [o.AskPrice for o in self.SecurityOptionLegs()]
return round(sum(prices), 2)
def UnderlyingPrice(self, algo):
return algo.Securities[self.Underlying()].Price
def Strike(self):
if self.IsHolding() or self.IsContract():
strikes = [c.StrikePrice for c in self.SecurityOptionLegs()]
else:
strikes = [c.Strike for c in self.SecurityOptionLegs()]
strikes = list(set(strikes))
if len(strikes) > 1:
return strikes
else:
return strikes[0]
def SecurityOptionLegs(self):
if self.IsHolding():
self.securityOptionLegs = self.securityOptionLegs or [x.Security for x in self.optionLegs]
# is this a contract Symbol?
elif self.IsContract():
self.securityOptionLegs = self.securityOptionLegs or [x.ID for x in self.optionLegs]
else:
self.securityOptionLegs = self.securityOptionLegs or self.optionLegs
return self.securityOptionLegs
def IsContract(self):
return hasattr(self.optionLegs[0], 'ID')
def IsHolding(self):
return isinstance(self.optionLegs[0], OptionHolding)
def IsOption(self):
return isinstance(self.optionLegs[0], Option)
def Underlying(self):
if self.IsHolding() or self.IsContract():
# This is a str. (it might not be in the case of the Holding though?!?)
return self.SecurityOptionLegs()[0].Underlying.Symbol
else:
return self.SecurityOptionLegs()[0].UnderlyingSymbol
def NameKey(self):
return "".join(self.name.split())
class Straddle(BaseOptionStrategy):
Put = None
Call = None
def __init__(self, Put, Call):
BaseOptionStrategy.__init__(self, "Straddle", [Put, Call])
self.Call = Call
self.Put = Put
self.__StrikeCheck()
if self.SameExpiration() == False:
raise Exception("The expiration should be the same for all options.")
def __StrikeCheck(self):
# is this an option holding?
if self.IsHolding():
callStrike = self.Call.Security.StrikePrice
putStrike = self.Put.Security.StrikePrice
# is this a contract Symbol?
elif self.IsContract():
callStrike = self.Call.ID.StrikePrice
putStrike = self.Put.ID.StrikePrice
# is this a OptionChain Symbol?
else:
callStrike = self.Call.Strike
putStrike = self.Put.Strike
if callStrike != putStrike:
raise Exception("The Call strike has to be equal to the Put strike.")
# TODO fix this CoveredCall (SoldCall) strategy here so it works. Change the name also as it's not a CoveredCall that implies the buying of the stock.
class SoldPut(BaseOptionStrategy):
Put = None
def __init__(self, put):
BaseOptionStrategy.__init__(self, "Sold Put", [put])
self.Put = put
# TODO open does not make sense?!!! maybe consider with insights to add??
class SoldCall(BaseOptionStrategy):
Call = None
def __init__(self, call):
BaseOptionStrategy.__init__(self, "Sold Call", [call])
self.Call = call
# TODO open does not make sense?!!! maybe consider with insights to add??
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 __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 __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)
#region imports from AlgorithmImports import * from .Handler import Handler from .OStrategies import * #endregion # Your New Python File
# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
# Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from AlgorithmImports import *
from Selection.OptionUniverseSelectionModel import OptionUniverseSelectionModel
class SafeCallOptionSelectionModel(OptionUniverseSelectionModel):
'''Creates option chain universes that select only the two week earliest expiry call contract
and runs a user defined optionChainSymbolSelector every day to enable choosing different option chains'''
def __init__(self, select_option_chain_symbols, targetExpiration = 14):
super().__init__(timedelta(1), select_option_chain_symbols)
self.targetExpiration = targetExpiration
def Filter(self, filter):
'''Defines the option chain universe filter'''
return (filter.Strikes(+3, +5)
.Expiration(self.targetExpiration, self.targetExpiration * 2)
.IncludeWeeklys()
.OnlyApplyFilterAtMarketOpen())#region imports
from AlgorithmImports import *
from QuantConnect.Logging import *
from MarketHours import MarketHours
from PortfolioHandler import Handler
from PortfolioHandler.OStrategies import SoldCall
#endregion
class SafeSoldCallAlphaModel(AlphaModel):
options = {}
algorithm = None
sliceData = None
def __init__(self, algorithm, ticker, option):
self.ticker = ticker
self.option = option
self.tolerance = 0.1
self.algorithm = algorithm
self.symbol = algorithm.AddEquity(self.ticker)
self.marketHours = MarketHours(algorithm, self.ticker)
self.portfolio = Handler(self.algorithm)
self.targetExpiration = ExpirationRange(7, 30)
self.Log = Logger(self.algorithm)
self.Credit = TradeCredit()
# Set up default Indicators, these indicators are defined on the Value property of incoming data (except ATR and AROON which use the full TradeBar object)
# self.indicators = {
# # 'BB' : algorithm.BB(self.ticker, 20, 1, MovingAverageType.Simple, Resolution.Hour),
# 'RSI' : algorithm.RSI(self.ticker, 14, MovingAverageType.Simple, Resolution.Hour),
# # 'EMA' : algorithm.EMA(self.ticker, 14, Resolution.Hour),
# # 'SMA' : algorithm.SMA(self.ticker, 14, Resolution.Hour),
# 'MACD' : algorithm.MACD(self.ticker, 12, 26, 9, MovingAverageType.Exponential, Resolution.Hour),
# 'MOM' : algorithm.MOM(self.ticker, 20, Resolution.Hour),
# # 'MOMP' : algorithm.MOMP(self.ticker, 20, Resolution.Hour),
# # 'STD' : algorithm.STD(self.ticker, 20, Resolution.Hour),
# # by default if the symbol is a tradebar type then it will be the min of the low property
# # 'MIN' : algorithm.MIN(self.ticker, 14, Resolution.Hour),
# # by default if the symbol is a tradebar type then it will be the max of the high property
# # 'MAX' : algorithm.MAX(self.ticker, 14, Resolution.Hour),
# 'ATR' : algorithm.ATR(self.ticker, 14, MovingAverageType.Simple, Resolution.Hour),
# # 'AROON' : algorithm.AROON(self.ticker, 20, Resolution.Hour)
# }
# self.algorithm.benchmark.AddIndicators(self.indicators)
# TODO: - draw a line on the trade plot that shows the constant strike price and try and make it not touch the stock price.
# - use the ATR ()
def Update(self, algorithm, data):
if algorithm.IsWarmingUp: return []
if self.ticker not in data.Keys: return []
self.algorithm.benchmark.PrintBenchmark()
weekday = self.algorithm.Time.isoweekday()
insights = []
# TODO print the RollPrice to see how the algo determines the proper strike to pick and make sure we are looking further based on that for safety
# IMPORTANT!!: In order to fix the cancelled closing order instance only check to close an order after 12. The reason is the market orders that happen
# before market open are converted to MarketOnOpen orders and it seems this does not get resolved.
# THIS DOES A NEGATIVE RESULT
# if self.algorithm.Time.hour > self.marketHours.get_CurrentOpen().hour + 2:
# # Monitor every minute
# if self.marketHours.get_CurrentClose().hour - 1 > self.algorithm.Time.hour:
# insights.extend(self.MonitorCoveredCall(data))
# # SCAN just once an hour
# if self.algorithm.Time.minute == 5 and self.marketHours.get_CurrentClose().hour - 2 > self.algorithm.Time.hour:
# insights.extend(self.Scanner(data))
# if self.algorithm.Time.hour > self.marketHours.get_CurrentOpen().hour + 4:
# insights.extend(self.MonitorHedgePut(data))
# if weekday == DayOfWeek.Thursday or weekday == DayOfWeek.Tuesday:
# if 11 > self.algorithm.Time.hour >= 10 and (self.algorithm.Time.minute == 15 or self.algorithm.Time.minute == 45):
# insights.extend(self.Scanner(data))
# POSITIVE RESULTS ($2158 / 0.87%)
if self.marketHours.get_CurrentClose().hour - 2 > self.algorithm.Time.hour > self.marketHours.get_CurrentOpen().hour + 2:
if self.algorithm.Time.minute == 30 or self.algorithm.Time.minute == 15 or self.algorithm.Time.minute == 45:
insights.extend(self.MonitorCoveredCall(data))
if self.algorithm.Time.minute == 5:
insights.extend(self.Scanner(data))
return insights
def MonitorHedgePut(self, data):
insights = []
stockPrice = self.algorithm.Securities[self.ticker].Price
for eput in self.portfolio.UnderlyingSoldOptions(self.ticker, OptionRight.Put):
close = False
expiresIn = self.__ExpiresIn(eput.Security)
if eput.UnrealizedProfitPercent * 100 > 90:
close = True
# if already invested in this position then check if it expires today
elif expiresIn == 0 and self.algorithm.Time.hour == self.marketHours.get_CurrentClose().hour - 4:
close = True
# elif stockPrice < eput.Security.StrikePrice:
# close = True
if close:
insights.append(
Insight.Price(eput.Symbol, Resolution.Minute, 1, InsightDirection.Flat)
)
return insights
def MonitorCoveredCall(self, data):
insights = []
# v1
# - profit <= -100
# - profit > 95
# - ExpiresIn == 0 and Time.hour > 15
# v2
# - ExpiresIn <= 2
# - profit > 80
# - profit <= -100
# hedged = self.__IsHedged()
stockPrice = self.algorithm.Securities[self.ticker].Price
for call in self.portfolio.SoldCalls(self.ticker, expiration = [0, self.targetExpiration.Stop]):
close = False
hedge = False
# if ((stockPrice * 1.1 > call.Strike()) and call.UnrealizedProfit() > -50):
# close = True
# self.Log.Add("ROLL - call {call}: stockPrice({stockPrice}) * 1.1 > strike{strike} and profit{profit} > -50".format(
# call = call.ToString(self.algorithm), profit = call.UnrealizedProfit(), stockPrice = stockPrice, strike = call.Strike() )
# )
# if ((stockPrice * 1.2 > call.Strike()) and call.UnrealizedProfit() > -25):
# close = True
# self.Log.Add("ROLL - call {call}: stockPrice({stockPrice}) * 1.2 > strike{strike} and profit{profit} > -25".format(
# call = call.ToString(self.algorithm), profit = call.UnrealizedProfit(), stockPrice = stockPrice, strike = call.Strike() )
# )
# v2 ?
# if call.ExpiresIn(self.algorithm) <= 2:
# close = True
# self.Log.Add("ROLL - call {}: expiresIn{} <= 2".format(call.ToString(self.algorithm), call.ExpiresIn(self.algorithm) ))
# v2 ?
if call.UnrealizedProfit() > 80:
close = True
self.Log.Add("ROLL - call {}: unrealized profit {} > 80".format(call.ToString(self.algorithm), call.UnrealizedProfit() ))
# v1 GOOD RESULT!
# if call.UnrealizedProfit() > 95:
# close = True
# self.Log.Add("ROLL - call {}: unrealized profit {} > 95".format(call.ToString(self.algorithm), call.UnrealizedProfit() ))
# v1 GOOD RESULT!
if call.UnrealizedProfit() <= -100:
close = True
self.Log.Add("ROLL - call {}: unrealized profit {} <= -100".format(call.ToString(self.algorithm), call.UnrealizedProfit() ))
# if stockPrice >= call.Security.StrikePrice / 1.1:
# close = True
# hedge = True
# if stockPrice > call.Strike() and self.algorithm.Time.hour >= 12 and call.ExpiresIn(self.algorithm) < round(self.targetExpiration * 0.8):
# close = True
# hedge = True
# if already invested in this position then check if it expires today
# v1 GOOD RESULT!
if call.ExpiresIn(self.algorithm) == 0 and self.algorithm.Time.hour > 15:
close = True
self.Log.Add("ROLL - call {}: expiresIn == 0 and hour {} > 15".format(call.ToString(self.algorithm), self.algorithm.Time.hour ))
# if self.indicators['MACD'].Signal.Current.Value >= 10 and self.indicators['RSI'].Current.Value >= 70 and stockPrice >= call.Security.StrikePrice / 1.5:
# close = True
# elif (self.__SignalDeltaPercent() < -self.tolerance and self.indicators['RSI'].Current.Value > 70):
# close = True
# hedge = True
# if hedge and not hedged:
# hedgeContract = self.__FindPut(data)
# insights.append(
# Insight.Price(hedgeContract.Symbol, Resolution.Minute, 1, InsightDirection.Down)
# )
if close:
self.Credit.Buy(call.AskPrice(self.algorithm))
self.Log.Add("ROLL - buying with credit {} / ${}".format(self.Credit.ToString(), call.AskPrice(self.algorithm)))
insights.append(
Insight.Price(call.Call.Symbol, Resolution.Minute, 15, InsightDirection.Flat)
)
self.Log.Print()
return insights
def Scanner(self, data):
insights = []
## Buying conditions
# if self.indicators['RSI'].Current.Value < 40: return insights
# IF RSI is > 40 -- NO
# | YES
# if self.indicators['MACD'].Signal.Current.Value >= 10 and self.indicators['RSI'].Current.Value >= 70:
# return insights
# 0 positions covered calls or hedge -- NO
# | YES
# lenSoldCalls = len(self.portfolio.UnderlyingSoldOptions(self.ticker, OptionRight.Call))
lenSoldCalls = len(self.portfolio.SoldCalls(self.ticker, expiration = [0, self.targetExpiration.Stop]))
# if lenSoldCalls > 0 or self.__IsHedged():
# return insights
if lenSoldCalls == 0:
call = self.__FindCall(data)
self.Log.Add("SCANNER - Trying to find a call")
if call:
self.Credit.Sell(call.AskPrice(self.algorithm))
self.Log.Add("SCANNER - selling {} with credit {} / ${}".format(call.ToString(self.algorithm), self.Credit.ToString(), call.AskPrice(self.algorithm)) )
insights.append(
Insight.Price(call.Call, Resolution.Minute, 15, InsightDirection.Down)
)
self.Log.Print()
return insights
def __FindPut(self, data, delta = -0.6):
chain = data.OptionChains.GetValue(self.option)
if not chain: 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) >= self.targetExpiration and x.Right == OptionRight.Put]
# # 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.algorithm.Securities[x.Symbol].IsTradable]
if not contracts: return None
return min(contracts, key=lambda x: abs(x.Greeks.Delta - delta))
def __FindCall(self, data):
call = None
stockPrice = self.algorithm.Securities[self.ticker].Price
minCredit = 1 # default min credit of 100$
# in case we are rolling and we had a loss then roll for a bigger premium
if self.Credit.LastValue > 1:
minCredit = self.Credit.LastValue * 1.1
contracts = self.algorithm.OptionChainProvider.GetOptionContractList(self.ticker, self.algorithm.Time.date())
if len(contracts) == 0 : return None
# # 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.algorithm.Securities[x.ID.Symbol].IsTradable]
# pick withing expiration range
contracts = [i for i in contracts if self.targetExpiration.Start <= (i.ID.Date.date() - self.algorithm.Time.date()).days <= self.targetExpiration.Stop]
if not contracts: return None
# select only calls
contracts = [x for x in contracts if x.ID.OptionRight == OptionRight.Call]
# sort by date
contracts = sorted(contracts, key = lambda x: x.ID.Date, reverse = False)
# TODO make this work based on EMA and some formula below self.__RollPrice
# pick contracts with strikes between 15% and 20% of spot price
contracts = [x for x in contracts if stockPrice * 1.1 < x.ID.StrikePrice < stockPrice * 1.20]
# contracts = ATMcontracts
# add all the option contracts so we can access all the data.
for c in contracts:
self.algorithm.AddOptionContract(c, Resolution.Minute)
calls = [[self.algorithm.Securities[c.Value].AskPrice, c] for c in contracts]
call = min(calls, key=lambda x: abs(x[0] - minCredit))[1]
if not call: return None
return SoldCall(call)
def __ChainFindCall(self, data, delta = 0.01):
chain = data.OptionChains.GetValue(self.option)
if not chain: 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) >= self.targetExpiration and x.Right == OptionRight.Call]
# # 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.algorithm.Securities[x.Symbol].IsTradable]
if not contracts: return None
return min(contracts, key=lambda x: abs(x.Greeks.Delta - delta))
def __IsHedged(self):
return len(self.portfolio.SoldPuts(self.ticker, expiration = [0, self.targetExpiration.Stop])) > 0
# 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.algorithm.Time.date()).days
def __SignalDeltaPercent(self):
return (self.indicators['MACD'].Current.Value - self.indicators['MACD'].Signal.Current.Value) / self.indicators['MACD'].Fast.Current.Value
# Roll to a strike a certain percentage in price up
def __RollPrice(self, option):
atr_slow = self.atr_slow.Current.Value
atr_fast = self.atr_fast.Current.Value
ema_slow = self.ema_slow.Current.Value
ema_fast = self.ema_fast.Current.Value
security = option.Security
strike = security.Symbol.ID.StrikePrice
stock_price = self.Securities[self.underlying].Price
ema_ratio = ema_fast / ema_slow
if not self.underlying in self.last_price:
self.last_price[self.underlying] = list()
history = self.last_price[self.underlying]
self.last_price[self.underlying].append(stock_price)
move = history[0] - history[-1]
#new_strike = ((
# max(atr_slow, atr_fast) / max(ema_fast, stock_price) + 1
#) ** 3) * max(ema_fast, ema_slow, stock_price) #* ema_ratio
# TODO replace the 1.4 by EMA ratio
#new_strike = (max(ema_fast, ema_slow, stock_price) + max(atr_slow, atr_fast)) * ema_ratio
#new_strike = (max(ema_fast, stock_price) + max(atr_slow, atr_fast)) * (ema_ratio ** 1.5)
new_strike = max(ema_slow, ema_fast, stock_price)*1.25 * (ema_ratio ** 2) + max(atr_slow, atr_fast) * 1
#max(ema_fast, stock_price)*1.05 * (ema_ratio ** 2) + move * 0.8,
new_strike = max(
new_strike,
max(ema_fast, stock_price) * 1.25
#stock_price * 1.03
#ema_fast
)
# TODO instead of ATR lets just check how big the move was in the past 5 days and project it into the future
new_strike = min(
new_strike,
max(ema_fast, stock_price)*1.3
)
#new_strike = max(ema_fast, stock_price) * 1.02 + move * 0.8
if len(self.last_price[self.underlying]) > 5:
self.last_price[self.underlying] = self.last_price[self.underlying][-5:]
# dampen falloff
if self.last_roll > new_strike:
new_strike = (self.last_roll + new_strike*2) / 3
self.last_roll = new_strike
# in case of rolling we try to make up for the loss by not moving the strike too much
#if stock_price > strike:
# new_strike = strike * 1.05
return new_strike
class ExpirationRange:
def __init__(self, start, stop):
self.Start = start
self.Stop = stop
def ToArr(self):
return [self.Start, self.Stop]
# Class to hold the credit of our trade so we can separate LEAP profit/credit from short profit but also combine them when needed.
class TradeCredit:
Value = 0
LastValue = 0
# When we buy something we remove from credit as we are paying from balance to get the options.
def Buy(self, value = 0):
self.LastValue = value
self.Value -= value
return self.Value
# When we are selling we add to credit as we are receiving premium for selling them.
def Sell(self, value = 0):
self.Value += value
return self.Value
def ToString(self):
return "Total: ${} (last ${})".format(round(self.Value, 2) * 100, round(self.LastValue, 2))
# Make this logger class so we can print logging messages in section for easier read.
class Logger:
Messages = []
Algorithm = None
def __init__(self, algorithm):
self.Algorithm = algorithm
def Add(self, message):
self.Messages.append(message)
def Print(self):
if len(self.Messages) > 0:
self.Algorithm.Log("------***------")
for m in self.Messages:
self.Algorithm.Log(m)
self.Algorithm.Log("------|||------")
self.Messages = []#region imports
from AlgorithmImports import *
from Risk.MaximumDrawdownPercentPerSecurity import MaximumDrawdownPercentPerSecurity
import numpy as np
from PortfolioHandler import Handler
#endregion
class TrailingStopRisk(RiskManagementModel):
'''Provides an implementation of IRiskManagementModel that limits the drawdown per holding to the specified percentage'''
def __init__(self, maximumDrawdownPercent = 5, profitTarget = None, ticker = None, strategies = [], algo = None):
'''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.algo = algo
self.symbol = Symbol.Create(ticker, SecurityType.Equity, Market.USA)
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 = self.algo.portfolio
portfolio = Handler(algorithm)
# TODO: this works but it's clear it does not get the ObjectStore from our main algo?!! WHY?
# TODO: fix this code to work with any strategy including SoldCalls. It might actually work!
portfolio.SyncStored(self.symbol, self.strategies)
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, InsightDirection.Flat))
else:
if pnl < self.maximumDrawdownPercent:
for c in contract.optionLegs:
targets.append(PortfolioTarget(c.Symbol, InsightDirection.Flat))
return targets
# region imports
from AlgorithmImports import *
# from Alphas.ConstantAlphaModel import ConstantAlphaModel
# from Selection.OptionUniverseSelectionModel import OptionUniverseSelectionModel
# from Execution.ImmediateExecutionModel import ImmediateExecutionModel
from TrailingStopRisk import TrailingStopRisk
from Risk.NullRiskManagementModel import NullRiskManagementModel
from Benchmark import Benchmark
from PortfolioHandler import Handler
# from UniverseSelection import OptionUniverseSelectionModel2
from OptionsSpreadExecution import OptionsSpreadExecution
from SafeSoldCallAlphaModel import SafeSoldCallAlphaModel
from SafeCallOptionSelectionModel import SafeCallOptionSelectionModel
from MilkTheCowAlphaModel import MilkTheCowAlphaModel
from MilkTheCowOptionSelectionModel import MilkTheCowOptionSelectionModel
# endregion
class AddAlphaModelAlgorithm(QCAlgorithm):
def Initialize(self):
''' Initialise the data and resolution required, as well as the cash and start-end dates for your algorithm. All algorithms must initialized.'''
# self.SetStartDate(2012, 2, 4) # Set Start Date
# self.SetEndDate(2020, 31, 7) # Set End Date
self.SetStartDate(2021, 1, 1) # Set Start Date
self.SetEndDate(2022, 5, 1) # Set End Date
self.SetCash(250000) # Set Strategy Cash
# Set settings and account setup
self.UniverseSettings.Resolution = Resolution.Minute
self.UniverseSettings.FillForward = False
# THIS ALSO DOES NOT SEEM TO FIX THE MARGIN ISSUE
# self.Portfolio.MarginCallModel = MarginCallModel.Null
self.SetSecurityInitializer(self.CustomSecurityInitializer)
self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage, AccountType.Margin) # Set InteractiveBrokers Brokerage model
# Main variables
self.ticker = self.GetParameter("ticker")
self.benchmark = Benchmark(self, self.ticker)
self.portfolio = Handler(self)
self.option = Symbol.Create(self.ticker, SecurityType.Option, Market.USA, f"?{self.ticker}")
equity = Symbol.Create(self.ticker, SecurityType.Equity, Market.USA)
# Milk the cow models
# self.SetUniverseSelection(MilkTheCowOptionSelectionModel(self.SelectOptionChainSymbols, expirationRange=[7, 30])) # Options for short Straddle
# self.SetUniverseSelection(MilkTheCowOptionSelectionModel(self.SelectOptionChainSymbols, expirationRange=[365, 395])) # Options for long Straddle
# self.SetAlpha(MilkTheCowAlphaModel(self, self.ticker, self.option))
# self.SetExecution(OptionsSpreadExecution(acceptingSpreadPercent=0.050))
# self.SetRiskManagement(TrailingStopRisk(algo = self, maximumDrawdownPercent = 20, profitTarget = 90, ticker = self.ticker, strategies = ["SoldCalls", "SoldPuts"]))
self.SetPortfolioConstruction(SingleSharePortfolioConstructionModel())
# self.SetExecution(ImmediateExecutionModel())
self.SetRiskManagement(NullRiskManagementModel())
# Safe call models
# self.SetUniverseSelection(SafeCallOptionSelectionModel(self.SelectOptionChainSymbols, targetExpiration=14))
self.SetAlpha(SafeSoldCallAlphaModel(self, self.ticker, self.option))
self.SetExecution(OptionsSpreadExecution(acceptingSpreadPercent=0.050))
# self.SetRiskManagement(TrailingStopRisk(algo = self, maximumDrawdownPercent = 20, profitTarget = 90, ticker = self.ticker, strategies = ["SoldCalls", "SoldPuts"]))
# OTHER models from learning
# self.SetUniverseSelection(OptionUniverseSelectionModel2(timedelta(1), self.SelectOptionChainSymbols))
# self.SetAlpha(ConstantOptionContractAlphaModel(InsightType.Price, InsightDirection.Up, timedelta(hours = 0.5)))
# self.SetExecution(SpreadExecutionModel())
# self.SetExecution(MarketOrderExecutionModel())
# set the buying power model
# security = self.AddEquity(self.ticker)
# security.SetBuyingPowerModel(CustomBuyingPowerModel())
self.SetWarmUp(TimeSpan.FromDays(30))
def SelectOptionChainSymbols(self, utcTime):
return [ self.option ]
def OnOrderEvent(self, orderEvent):
# MILK THE COW ALGO! FROM: https://optionstradingiq.com/calendar-straddle/
# As long as we’re on the topic of rolling issues; what if you get assigned?
# First, keep calm since it’s no big deal.
# Simply close out the shares assigned, close out the other option in the short straddle and just reopen your short straddle using the guidelines.
order = self.Transactions.GetOrderById(orderEvent.OrderId)
if orderEvent.Status == OrderStatus.Filled:
if orderEvent.IsAssignment and order.Type == OrderType.OptionExercise:
# - after each event that assigns an option find out if the option was in a strategy and close it all (sell stock sell the other option/s)
trade = self.portfolio.FixAssignment(order, strategies = ["Straddles", "SoldCalls"])
else:
# - sync with ObjectStore the portfolio strategies i give
# - don't store in the Store the single options just strategies
# - sync after each Event that buys/sells a new option that is not an assignment
self.portfolio.SyncStored(self.ticker, strategies = ['Straddles', 'SoldCalls'])
# this method is expected to ignore the ObjectStore strategies and just filter what is on the portfolio.
trade = self.portfolio.GetCurrentTrade(order, strategies = ["Straddles", "SoldCalls"])
if trade: self.benchmark.PrintTrade(order = order, trade = trade, quantity = order.Quantity)
# https://www.quantconnect.com/forum/discussion/13199/greeks-with-optionchainprovider/p1/comment-38906
# def OptionContractSecurityInitializer(self, security):
# if security.Type == SecurityType.Equity:
# symbol = security.Symbol
# security.VolatilityModel = StandardDeviationOfReturnsVolatilityModel(30, Resolution.Daily)
# for index, row in self.History(symbol, 30, Resolution.Daily).iterrows():
# security.SetMarketPrice(IndicatorDataPoint(index[1], row.close))
# if security.Type == SecurityType.Option:
# security.PriceModel = OptionPriceModels.CrankNicolsonFD()
# https://www.quantconnect.com/forum/discussion/10236/options-delta-always-zero/p1/comment-29181
def CustomSecurityInitializer(self, security):
'''Initialize the security with raw prices'''
security.SetDataNormalizationMode(DataNormalizationMode.Raw)
security.SetMarketPrice(self.GetLastKnownPrice(security))
if security.Type == SecurityType.Equity:
security.VolatilityModel = StandardDeviationOfReturnsVolatilityModel(30)
history = self.History(security.Symbol, 31, Resolution.Daily)
if history.empty or 'close' not in history.columns:
return
for time, row in history.loc[security.Symbol].iterrows():
trade_bar = TradeBar(time, security.Symbol, row.open, row.high, row.low, row.close, row.volume)
security.VolatilityModel.Update(security, trade_bar)
elif security.Type == SecurityType.Option:
security.PriceModel = OptionPriceModels.CrankNicolsonFD() # BlackScholes()
class SingleSharePortfolioConstructionModel(PortfolioConstructionModel):
'''Portfolio construction model that sets target quantities to 1 for up insights and -1 for down insights'''
def CreateTargets(self, algorithm, insights):
targets = []
for insight in insights:
targets.append(PortfolioTarget(insight.Symbol, insight.Direction * self.TargetQuantity(algorithm, insight.Symbol)))
return targets
# Method that defines how many option contracts to sell or buy by symbol.
# Here we can expand this to be variable by Symbol or defined by a parameter or by portfolio alocation based on margin available.
def TargetQuantity(self, algo, symbol):
# algo.CalculateOrderQuantity(symbol, 1)
return 1
class CustomBuyingPowerModel(BuyingPowerModel):
def GetMaximumOrderQuantityForTargetBuyingPower(self, parameters):
quantity = super().GetMaximumOrderQuantityForTargetBuyingPower(parameters).Quantity
quantity = np.floor(quantity / 100) * 100
return GetMaximumOrderQuantityResult(quantity)
def HasSufficientBuyingPowerForOrder(self, parameters):
return HasSufficientBuyingPowerForOrderResult(True)