| Overall Statistics |
|
Total Trades 264 Average Win 0.13% Average Loss -0.24% Compounding Annual Return 2.314% Drawdown 4.700% Expectancy -0.032 Net Profit 2.126% Sharpe Ratio 0.418 Probabilistic Sharpe Ratio 25.376% Loss Rate 37% Win Rate 63% Profit-Loss Ratio 0.53 Alpha 0 Beta 0 Annual Standard Deviation 0.04 Annual Variance 0.002 Information Ratio 0.418 Tracking Error 0.04 Treynor Ratio 0 Total Fees $262.00 Estimated Strategy Capacity $7900000.00 Lowest Capacity Asset SPY 31J2MQO5S86W6|SPY R735QTJ8XC9X |
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:
self.tradingChart.AddSeries(Series('Sell Call', SeriesType.Scatter, '$', Color.Green, ScatterMarkerSymbol.Circle))
self.tradingChart.AddSeries(Series('Buy Call', SeriesType.Scatter, '$', Color.Red, ScatterMarkerSymbol.Triangle))
# self.tradingChart.AddSeries(Series('Sell Put', SeriesType.Scatter, '$', Color.Green, ScatterMarkerSymbol.Circle))
# self.tradingChart.AddSeries(Series('Buy Put', 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):
''' Prints the price of the option on our trade chart. '''
plotTradeName = ''
security = order.Symbol.ID
if security.OptionRight == OptionRight.Call:
optionType = 'Call'
elif security.OptionRight == OptionRight.Put:
optionType = 'Put'
if order.Quantity < 0:
plotTradeName = 'Sell {}'.format(optionType)
else:
plotTradeName = 'Buy {}'.format(optionType)
self.algo.Plot('Trade Plot', plotTradeName, security.StrikePrice)
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
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.LeapExpiration = ExpirationRange(365, 395)
self.ShortExpiration = ExpirationRange(7, 14)
self.vixSymbol = algorithm.AddIndex("VIX", Resolution.Minute).Symbol
self.VIX = 0
self.Credit = 0.0
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 data.ContainsKey(self.vixSymbol):
self.VIX = data[self.vixSymbol].Close
if self.marketHours.get_CurrentClose().hour - 2 > self.algorithm.Time.hour > self.marketHours.get_CurrentOpen().hour + 2:
insights.extend(self.Monitor(data))
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.StraddleOptions(self.ticker, expiration = self.LeapExpiration.ToArr(), short = False)
shortStraddles = self.portfolio.StraddleOptions(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
if shortStraddle.UnrealizedProfit() <= -325:
close = True
if close:
self.Credit = 0.0
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)
]
#### 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
if (longStraddle.Strike() / 1.065) > stockPrice or stockPrice > (longStraddle.Strike() * 1.065):
close = True
if longStraddle.UnrealizedProfit() > 0.5:
close = True
if close:
insights.extend(
[
Insight.Price(longStraddle.Call.Symbol, Resolution.Minute, 15, InsightDirection.Flat),
Insight.Price(longStraddle.Put.Symbol, Resolution.Minute, 15, InsightDirection.Flat)
]
)
# #### 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
if (shortStraddle.Strike() / 1.05) > stockPrice or stockPrice > (shortStraddle.Strike() * 1.05):
close = True
# i placed 10 here instead of 1$ it's too little. 10 might be also
# if 2 >= shortStraddle.ExpiresIn(self.algorithm) <= 5 and stockPrice > shortStraddle.Strike() + 1:
# close = True
if shortStraddle.UnrealizedProfit() > 20:
close = True
if close:
self.Credit -= 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)
]
)
return insights
def Scanner(self, data):
insights = []
# VIX is below 35 -- NO
# | YES
if self.VIX > 35: return insights
lenShortStraddle = len(self.portfolio.StraddleOptions(self.ticker, expiration = self.ShortExpiration.ToArr(), short = True))
lenLeapStraddle = len(self.portfolio.StraddleOptions(self.ticker, expiration = self.LeapExpiration.ToArr(), short = False))
# SCAN LEAP
# - if no LEAP
if lenLeapStraddle == 0:
longStraddle = self.__FindStraddle(data, expiration = self.LeapExpiration)
if longStraddle:
insights.extend([
Insight.Price(longStraddle.Call, Resolution.Minute, 15, InsightDirection.Up),
Insight.Price(longStraddle.Put, Resolution.Minute, 15, InsightDirection.Up),
])
# 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)
if shortStraddle:
self.Credit += shortStraddle.AskPrice(self.algorithm)
# reset credit if we did not manage to find a straddle to offset it.
# this ensures we have a reset in terms of expiration.
if self.Credit < 0.0: self.Credit = 0.0
insights.extend([
Insight.Price(shortStraddle.Call, Resolution.Minute, 15, InsightDirection.Down),
Insight.Price(shortStraddle.Put, Resolution.Minute, 15, InsightDirection.Down),
])
return insights
def __FindStraddle(self, data, expiration, minCredit = 0.0):
put = None
call = None
stockPrice = self.algorithm.Securities[self.ticker].Price
# if we have positive credit then we don't care
if minCredit > 0.0: minCredit = 0.0
# if we have negative credit then remove the negative sign.
if minCredit < 0.0: minCredit = abs(minCredit)
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):
group = list(g)
calls = [c for c in group if c.ID.OptionRight == OptionRight.Call]
puts = [c for c in group if c.ID.OptionRight == OptionRight.Put]
if not calls or not puts: continue
ATMcontracts.extend([min(calls, key=lambda x: abs(stockPrice - x.ID.StrikePrice))])
ATMcontracts.extend([min(puts, key=lambda x: abs(stockPrice - x.ID.StrikePrice))])
contracts = ATMcontracts
# add all the option contracts so we can access all the data.
for c in contracts:
self.algorithm.AddOptionContract(c, Resolution.Minute)
# try and get put and call that have a higher credit than specified min.
put = min([c for c in contracts if c.ID.OptionRight == OptionRight.Put], key=lambda x: abs(self.algorithm.Securities[c.Value].AskPrice - minCredit / 2))
call = min([c for c in contracts if c.ID.OptionRight == OptionRight.Call], key=lambda x: abs(self.algorithm.Securities[c.Value].AskPrice - minCredit / 2))
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.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
class ExpirationRange:
def __init__(self, start, stop):
self.Start = start
self.Stop = stop
def ToArr(self):
return [self.Start, self.Stop]# 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
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:
if (t.Symbol.SecurityType == symbol.SecurityType and \
algorithm.Securities[t.Symbol].Right == algorithm.Securities[symbol].Right and \
t.Quantity == target.Quantity):
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
# 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
# TODO: add the method that would return OptionStrategies class instances for sold calls.
# 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 StraddleOptions(self, underlying, expiration = [7, 30], short = True):
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 = 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 == True and sum(o.Quantity for o in options) > 0:
continue
if short == False and sum(o.Quantity for o in options) < 0:
continue
# 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)
# 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)
contracts.append(contract)
return contracts
def SoldPuts(self, underlying, ignoreStored = False):
# select all the calls in our portfolio
allContracts = self.OptionStrategies(underlying, [OptionRight.Put])
contracts = []
for t, puts in allContracts.items():
if len(puts) == 1 and puts[0].Quantity < 0:
put = puts[0]
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):
# select all the calls in our portfolio
allContracts = self.OptionStrategies(underlying, [OptionRight.Call])
contracts = []
for t, calls in allContracts.items():
if len(calls) == 1 and calls[0].Quantity < 0:
call = calls[0]
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
# 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
# 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
# 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):
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 sum(prices)
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)
# 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 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 Strike(self):
if self.IsHolding():
callStrike = self.Call.Security.StrikePrice
else:
callStrike = self.Call.Strike
return callStrike
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):
shortPut = None
def __init__(self, shortPut):
BaseOptionStrategy.__init__(self, "Sold Put", [shortPut])
self.shortPut = shortPut
# TODO open does not make sense?!!! maybe consider with insights to add??
class SoldCall(BaseOptionStrategy):
shortCall = None
def __init__(self, shortCall):
BaseOptionStrategy.__init__(self, "Sold Call", [shortCall])
self.shortCall = shortCall
# 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 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)
#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(+40, +70)
.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
#endregion
class SafeSoldCallAlphaModel(AlphaModel):
options = {}
algorithm = None
sliceData = None
def __init__(self, algorithm, ticker, option, targetExpiration = 14):
self.ticker = ticker
self.option = option
self.tolerance = 0.1
self.algorithm = algorithm
self.targetExpiration = targetExpiration
self.symbol = algorithm.AddEquity(self.ticker)
self.marketHours = MarketHours(algorithm, self.ticker)
self.portfolio = Handler(self.algorithm)
# 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 = []
# 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.
if self.algorithm.Time.hour > self.marketHours.get_CurrentOpen().hour + 4:
insights.extend(self.MonitorCoveredCall(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))
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 - 1:
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 = []
# hedged = self.__IsHedged()
stockPrice = self.algorithm.Securities[self.ticker].Price
for ecall in self.portfolio.UnderlyingSoldOptions(self.ticker, OptionRight.Call):
close = False
hedge = False
expiresIn = self.__ExpiresIn(ecall.Security)
if ecall.UnrealizedProfitPercent * 100 > 95:
close = True
# elif stockPrice >= ecall.Security.StrikePrice / 1.1:
# close = True
# hedge = True
elif stockPrice > ecall.Security.StrikePrice and self.algorithm.Time.hour >= 12 and expiresIn < round(self.targetExpiration * 0.8):
close = True
# hedge = True
# if already invested in this position then check if it expires today
elif expiresIn == 0 and self.algorithm.Time.hour > 15:
close = True
elif self.indicators['MACD'].Signal.Current.Value >= 10 and self.indicators['RSI'].Current.Value >= 70 and stockPrice >= ecall.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:
insights.append(
Insight.Price(ecall.Symbol, Resolution.Minute, 1, InsightDirection.Flat)
)
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
if len(self.portfolio.UnderlyingSoldOptions(self.ticker, OptionRight.Call)) > 0 or \
self.__IsHedged():
return insights
call = self.__FindCall(data)
if call is not None:
insights.append(Insight.Price(
call.Symbol,
Resolution.Minute,
1,
InsightDirection.Down
))
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, 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.UnderlyingSoldOptions(self.ticker, OptionRight.Put)) > 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
#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 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(2020, 1, 1) # Set Start Date
self.SetEndDate(2020, 12, 1) # Set End Date
self.SetCash(200000) # Set Strategy Cash
# Set settings and account setup
self.UniverseSettings.Resolution = Resolution.Minute
self.UniverseSettings.FillForward = False
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.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, targetExpiration=14))
# 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())
self.SetWarmUp(TimeSpan.FromDays(30))
def SelectOptionChainSymbols(self, utcTime):
return [ self.option ]
def OnOrderEvent(self, orderEvent):
if orderEvent.Status == OrderStatus.Filled:
order = self.Transactions.GetOrderById(orderEvent.OrderId)
self.benchmark.PrintTrade(order)
# 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(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, symbol):
return 1