| Overall Statistics |
|
Total Trades 0 Average Win 0% Average Loss 0% Compounding Annual Return 0% Drawdown 0% Expectancy 0 Net Profit 0% Sharpe Ratio 0 Probabilistic Sharpe Ratio 0% Loss Rate 0% Win Rate 0% Profit-Loss Ratio 0 Alpha 0 Beta 0 Annual Standard Deviation 0 Annual Variance 0 Information Ratio 2.898 Tracking Error 0.12 Treynor Ratio 0 Total Fees $0.00 Estimated Strategy Capacity $0 Lowest Capacity Asset |
#region imports
from AlgorithmImports import *
import datetime as dt
#endregion
CASH = 100000
START_DATE = '01-01-2022' #'DD-MM-YYYY'
END_DATE = '05-01-2022' # 'DD-MM-YYYY', Can be set as None
FREE_PORTFOLIO_VALUE_PCT = 0.025
BULL_NUMBER = 4
BEAR_NUMBER = 1
FIXED_SL = 1.5
FIXED_TARGET = 0.15
DAYS_TO_EXPIRY = 45
STRIKE_DISTANCE = 0.05
BUY_LOTS = 1
SELL_LOTS = 1
HV_LOOKBACK = 600
# Universe filters
PRICE_FLOOR = 20
PRICE_CEIL = 40
VOLUME_FLOOR = 500000
NUMBER_FROM_UNIVERSE = 100
IV_FLOOR = 0.4
IV_CEIL = 0.6
############### Parsing Logic #############
######## Do not edit anything below ######
START_DATE = dt.datetime.strptime(START_DATE, '%d-%m-%Y')
if END_DATE:
END_DATE = dt.datetime.strptime(END_DATE, '%d-%m-%Y')
'''
Portfolio construction
----------------------
First, we define a start day when we want to perform analysis (for example 19 March 2022) and then run it continuously
until a certain day (1 September 2022). The broker fee structure is set to Interactive Brokers. We set the resolution to hourly.
Next, we allocate some capital (for example, we set the portfolio to 30 000$ ) and define how many assets
that to be invested: 6 bull and 4 bear options spreads.
A bull spread is when we buy a call (ITM) and sell the call (OTM) with a higher strike.
Our bull spread will be Long leg= market price -0.5%; Short leg = market price + 0.5%
A bear spread is when we buy a call and sell the call with a lower strike. Our bear spread will be:
Long leg= market price +0.5%; Short leg = market price - 0.5%
Universe Selection
------------------
We define the universe of equities that we would like to scan using the following criteria :
totalVolume price more than 500,000 (average daily trading volume)
stock price more than 10 and less than 90;
IV more than 40
after that, we take 6 top performers for bull and 4 worst performers for bear using MACD (confirm current trend)
sorted by 10 days exponential Moving Average (EMA) removing outliers (35%+)
once we know the equities we will be investing in, we buy appropriate call spreads using the definition above.
when we purchase our options, we capture (i.e. log file) IV, greeks (Delta, Gamma, Vega, Vomma,T heta, Rho)
as well as a spread (Bid/Ask) as well as the price we got our order filled at.
On Liquidation of any asset, get another equity from current selection and enter that one.
Do not reenter an equity on same day.
Portfolio monitoring
--------------------
we set up a "stop loss" rule: we sell our spread if it loses 25% of its initial value
we set up a "sell-off" rule: we sell if our spread increases in price by 15%
when one option spread is liquidated (be it due to "stop loss" or "sell-off"), we replace it with another
contract (for example: if we sell a bull spread, then we have 5 bull spreads left, so we add one more bull spread).
the portfolio should be fully invested at all times.
spread call strikes can be rounded, but have to be equally spaced. for example stock at $51.89, then we choose a
medium point to be $52 then our differential is $52*0.005=0.26; if there are no options with strikes 51.74 and 52.26,
then we take the nearest pair (51.75 and 52.25) or (51.50 and 52.50) or (51 and 53)
'''
# region imports
from AlgorithmImports import *
from risk_management import FixedStopRMModel
import pandas as pd
from constants import *
import pickle
from numpy import sqrt,mean,log,diff
from scipy import stats
# endregion
class VerticalSpread(QCAlgorithm):
def Initialize(self):
self.SetBacktestDetails()
# setup state storage in initialize method
self.stateData = {}
self.option_symbols = {}
self.new_day = True
self.AddUniverse(self.CoarseFilterFunction)
self.equity_store = pd.DataFrame(columns=['equities'])
#-------------------------------------------------------------------------------
def SetBacktestDetails(self):
"""Set the backtest details."""
self.SetStartDate(START_DATE)
if END_DATE:
self.SetEndDate(END_DATE)
self.SetCash(CASH)
self.SetWarmup(30, Resolution.Daily)
#---------------------------------------------------------------------------------
def OnData(self, slice):
if not self.IsWarmingUp and self.Time.hour==10 and self.new_day:
# Filters coarse selected equities based on Expiry, Right and IV
# Sets self.bulls as top equities on this criteria
self.bulls = pd.DataFrame(columns=['IV','HV','HV_Percentile'])
for symbol in self.selected:
slice = self.CurrentSlice
chain = slice.OptionChains.get(self.option_symbols[symbol])
if not chain: continue
# sorted the optionchain by expiration date and choose the furthest date for the given dates to expiry
expiry = sorted(chain, key=lambda x: abs((x.Expiry - self.Time).days - DAYS_TO_EXPIRY))[0].Expiry
# filter the call options from the contracts expires on that date
target_calls = [i for i in chain if i.Expiry == expiry and i.Right == OptionRight.Call]
calls = [i for i in target_calls if IV_CEIL > i.ImpliedVolatility > IV_FLOOR]
calls = sorted(calls,key=lambda x: x.ImpliedVolatility,reverse=True)
if not calls: continue
self.bulls.loc[symbol,'IV'] = float(calls[0].ImpliedVolatility)
self.bulls.loc[symbol,'HV'],self.bulls.loc[symbol,'HV_Percentile'] = self.stateData[symbol].calcHV(self,HV_LOOKBACK)
self.bulls = self.bulls.astype(float)
self.bulls = self.bulls.loc[self.bulls['HV'] > (self.bulls['IV']*1.2)]
self.bulls = self.bulls.loc[self.bulls['HV_Percentile'].lt(35)]
self.bulls = self.bulls.nlargest(BULL_NUMBER,'IV')
if not self.bulls.empty:
self.Log(f'Top {BULL_NUMBER} bulls for {self.Time.date()} are {self.bulls.index.to_list()}')
self.equity_store.loc[self.Time.date()] = [[x.Value for x in self.bulls.index.to_list()]]
self.new_day = False
else:
self.Log(f'No bulls passed the criteria for {self.Time.date()}')
def OnEndOfAlgorithm(self):
# serializedString = pickle.dumps(
self.ObjectStore.Save("DKAVS", self.equity_store.reset_index().to_json(date_unit='ns'))
# self.ObjectStore.Save("DKAVS", serializedString)
#---------------------------------------------------------------------------------
def OnSecuritiesChanged (self, changes):
for x in changes.AddedSecurities:
if (x.Symbol.SecurityType != SecurityType.Equity) or (x.Symbol.Value=='SPY'): continue
option = self.AddOption(x.Symbol.Value, Resolution.Minute)
option.SetFilter(-1, +1, timedelta(DAYS_TO_EXPIRY-15), timedelta(DAYS_TO_EXPIRY+15))
option.PriceModel = OptionPriceModels.CrankNicolsonFD()
self.option_symbols[x.Symbol] = option.Symbol
for x in changes.RemovedSecurities:
if (x.Symbol.SecurityType != SecurityType.Equity) or (x.Symbol.Value=='SPY'): continue
self.RemoveOptionContract(self.option_symbols[x.Symbol])
#---------------------------------------------------------------------------------
def CoarseFilterFunction(self, coarse: List[CoarseFundamental]) -> List[Symbol]:
# We are going to use a dictionary to refer the object that will keep the moving averages
onlyEquities = [x for x in coarse if x.HasFundamentalData]
for c in onlyEquities:
if c.Symbol not in self.stateData:
self.stateData[c.Symbol] = SelectionData(c.Symbol, 10)
# Updates the SymbolData object with current EOD price
avg = self.stateData[c.Symbol]
avg.update(c.EndTime, c.AdjustedPrice, c.DollarVolume)
if self.IsWarmingUp:
return []
# Filter the values of the dict to those above EMA and more than $500000 vol.
values = [x for x in self.stateData.values() if (PRICE_FLOOR < x.price < PRICE_CEIL) and x.volume > VOLUME_FLOOR]
values = [x for x in values if (x.macd.Current.Value > x.macd.Signal.Current.Value) and (x.macd.Current.Value > 0)]
# sort by the largest in ema.
values.sort(key=lambda x: x.ema, reverse=True)
self.selected = [x.symbol for x in values]
invested = [x.Key for x in self.Portfolio if x.Value.Invested]
# we need to return only the symbol objects
return self.selected + invested
#-----------------------------------------------------------------------------------------
# def OnOrderEvent(self, orderEvent: OrderEvent) -> None:
# order = self.Transactions.GetOrderById(orderEvent.OrderId)
# if orderEvent.Status == OrderStatus.Filled:
# self.Debug(f"{self.Time}: {order.Type}: {orderEvent}: {orderEvent.Symbol} Filled at {orderEvent.FillPrice}")
#----------------------------------------------------------------------------------------
def OnEndOfDay(self):
self.new_day = True
self.Log(f'Total Securities in Universe: {len(self.ActiveSecurities)}')
#--------------------------------------------------------------------------------------
class SelectionData(object):
def __init__(self, symbol, period):
self.symbol = symbol
self.ema = ExponentialMovingAverage(period)
self.macd = MovingAverageConvergenceDivergence(12, 26, 9)
self.is_above_ema = False
self.volume = 0
def update(self, time, price, volume):
self.price = price
self.volume = volume
self.ema.Update(time, price)
self.macd.Update(time, price)
def calcHV(self, algo, HVLookback):
closes = algo.History(self.symbol, HVLookback, Resolution.Daily).close
diffs = diff(log(closes))
diffs_mean = mean(diffs)
diff_square = [(diffs[i]-diffs_mean)**2 for i in range(0,len(diffs))]
sigma = sqrt(sum(diff_square)*(1.0/(len(diffs)-1)))
self.HV = round(sigma*sqrt(252),2)
self.HVP = round(stats.percentileofscore(closes.iloc[-252:], closes.iloc[-1]),2)
return self.HV,self.HVP
'''
Portfolio construction
----------------------
First, we define a start day when we want to perform analysis (for example 19 March 2022) and then run it continuously
until a certain day (1 September 2022). The broker fee structure is set to Interactive Brokers. We set the resolution to hourly.
Next, we allocate some capital (for example, we set the portfolio to 30 000$ ) and define how many assets
that to be invested: 6 bull and 4 bear options spreads.
A bull spread is when we buy a call (ITM) and sell the call (OTM) with a higher strike.
Our bull spread will be Long leg= market price -0.5%; Short leg = market price + 0.5%
A bear spread is when we buy a call and sell the call with a lower strike. Our bear spread will be:
Long leg= market price +0.5%; Short leg = market price - 0.5%
Universe Selection
------------------
We define the universe of equities that we would like to scan using the following criteria :
totalVolume price more than 500,000 (average daily trading volume)
stock price more than 10 and less than 90;
IV more than 40
after that, we take 6 top performers for bull and 4 worst performers for bear using MACD (confirm current trend)
sorted by 10 days exponential Moving Average (EMA) removing outliers (35%+)
once we know the equities we will be investing in, we buy appropriate call spreads using the definition above.
when we purchase our options, we capture (i.e. log file) IV, greeks (Delta, Gamma, Vega, Vomma,T heta, Rho)
as well as a spread (Bid/Ask) as well as the price we got our order filled at.
On Liquidation of any asset, get another equity from current selection and enter that one.
Do not reenter an equity on same day.
Portfolio monitoring
--------------------
we set up a "stop loss" rule: we sell our spread if it loses 25% of its initial value
we set up a "sell-off" rule: we sell if our spread increases in price by 15%
when one option spread is liquidated (be it due to "stop loss" or "sell-off"), we replace it with another
contract (for example: if we sell a bull spread, then we have 5 bull spreads left, so we add one more bull spread).
the portfolio should be fully invested at all times.
spread call strikes can be rounded, but have to be equally spaced. for example stock at $51.89, then we choose a
medium point to be $52 then our differential is $52*0.005=0.26; if there are no options with strikes 51.74 and 52.26,
then we take the nearest pair (51.75 and 52.25) or (51.50 and 52.50) or (51 and 53)
'''
# region imports
from AlgorithmImports import *
from risk_management import FixedStopRMModel
import pandas as pd
from constants import *
import pickle
# endregion
class VerticalSpread(QCAlgorithm):
def Initialize(self):
self.SetBacktestDetails()
# setup state storage in initialize method
self.stateData = {}
self.option_symbols = {}
self.SetWarmup(30,Resolution.Daily)
self.LoadEquityStore()
self.AddUniverse(self.CoarseFilterFunction)
self.AddRiskManagement(FixedStopRMModel(self))
self.Schedule.On(self.DateRules.EveryDay(),self.TimeRules.At(15,0),self.ExitBeforeExpiry)
self.open_bulls = {}
self.open_bears = {}
self.closed_today = []
def LoadEquityStore(self):
self.equity_store = pd.read_json(self.ObjectStore.Read("DKAVS"))
self.equity_store['index'] = pd.to_datetime(self.equity_store['index']).dt.date
self.equity_store.set_index('index',inplace=True)
#-------------------------------------------------------------------------------
def SetBacktestDetails(self):
"""Set the backtest details."""
self.SetStartDate(START_DATE)
if END_DATE:
self.SetEndDate(END_DATE)
self.SetCash(CASH)
self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage,
AccountType.Margin)
self.SetWarmup(30, Resolution.Daily)
# Adjust the cash buffer from the default 2.5% to custom setting
self.Settings.FreePortfolioValuePercentage = FREE_PORTFOLIO_VALUE_PCT
# self.Settings.DataSubscriptionLimit = 500
#---------------------------------------------------------------------------------
def OnData(self, slice):
if not self.IsWarmingUp:
if len(self.open_bulls) < BULL_NUMBER:
# We'll filter equities based on their options IV in the first minute
self.bulls = pd.DataFrame(columns=['IV','Filtered_Calls'])
self.GetBulls()
if self.bulls.empty:
return
# for bull in self.bulls.index:
# sd = self.stateData[bull]
# self.Log(f'Bull: {bull.Value}; with ema {sd.ema.Current.Value}')
# self.Log(f'Bull: {bull.Value}; with macd signal {sd.macd.Signal.Current.Value}')
self.OpenBullCallSpreads(self.bulls)
#---------------------------------------------------------------------------------
def GetBulls(self):
# Filters coarse selected equities based on Expiry, Right and IV
# Sets self.bulls as top equities on this criteria
for symbol in self.selected:
slice = self.CurrentSlice
chain = slice.OptionChains.get(self.option_symbols[symbol])
if not chain: continue
# sorted the optionchain by expiration date and choose the furthest date for the given dates to expiry
expiry = sorted(chain, key=lambda x: abs((x.Expiry - self.Time).days - DAYS_TO_EXPIRY))[0].Expiry
# filter the call options from the contracts expires on that date
target_calls = [i for i in chain if i.Expiry == expiry and i.Right == OptionRight.Call]
calls = [i for i in target_calls if IV_CEIL > i.ImpliedVolatility > IV_FLOOR]
calls = sorted(calls,key=lambda x: x.ImpliedVolatility,reverse=True)
if not calls: continue
self.bulls.loc[symbol,'IV'] = float(calls[0].ImpliedVolatility)
self.bulls.loc[symbol,'Filtered_Calls'] = target_calls
exclusions = self.GetExclusions() # Exclude already invested and same day liquidated equities
self.bulls = self.bulls[~self.bulls.index.isin(exclusions)]
self.bulls['IV'] = self.bulls['IV'].astype(float)
self.bulls = self.bulls.nlargest(BULL_NUMBER - len(self.open_bulls),'IV')
#------------------------------------------------------------------------------------------
def GetExclusions(self):
portfolio_syms = [x[0].Underlying for x in self.open_bulls.keys()]
today_liquidated_syms = [x[0].Underlying for x in self.closed_today]
return portfolio_syms + today_liquidated_syms
#------------------------------------------------------------------------------------------
def GetOptionsForSymbol(self,symbol,calls):
slice = self.CurrentSlice
underlying_price = slice[symbol].Price
itm_calls = [x for x in calls if x.Strike < underlying_price]
itm_calls = sorted(itm_calls, key=lambda x: abs(x.Strike - (underlying_price*(1-STRIKE_DISTANCE))))
otm_calls = [x for x in calls if x.Strike > underlying_price]
otm_calls = sorted(otm_calls, key=lambda x: abs(x.Strike - (underlying_price*(1+STRIKE_DISTANCE))))
return itm_calls, otm_calls, underlying_price
#---------------------------------------------------------------------------------
def OpenBullCallSpreads(self, bulls):
for symbol,row in bulls.iterrows():
itm_calls,otm_calls,uprice = self.GetOptionsForSymbol(symbol,row['Filtered_Calls'])
if (not itm_calls) or (not otm_calls): return
self.Log(f'Underlying Price: {uprice}')
# Buy call option contract with lower strike
self.Buy(itm_calls[0].Symbol, BUY_LOTS)
self.Log(f'Bought {itm_calls[0].Symbol} with attributes-> Expiry: {itm_calls[0].Expiry}, \
Strike: {itm_calls[0].Strike}, Ask: {itm_calls[0].AskPrice}, \
Delta: {itm_calls[0].Greeks.Delta}, Gamma: {itm_calls[0].Greeks.Gamma}, \
Vega: {itm_calls[0].Greeks.Vega}, Theta: {itm_calls[0].Greeks.Theta}, Rho: {itm_calls[0].Greeks.Rho}')
# Sell call option contract with higher strike
self.Sell(otm_calls[0].Symbol, SELL_LOTS)
self.Log(f'Sold {otm_calls[0].Symbol} with attributes-> Expiry: {otm_calls[0].Expiry}, \
Strike: {otm_calls[0].Strike}, Bid: {otm_calls[0].BidPrice}, \
Delta: {otm_calls[0].Greeks.Delta}, Gamma: {otm_calls[0].Greeks.Gamma}, \
Vega: {otm_calls[0].Greeks.Vega}, Theta: {otm_calls[0].Greeks.Theta}, Rho: {otm_calls[0].Greeks.Rho}')
self.open_bulls[(itm_calls[0].Symbol,otm_calls[0].Symbol)] = self.Time
#---------------------------------------------------------------------------------
def OnSecuritiesChanged (self, changes):
for x in changes.AddedSecurities:
x.SetFeeModel(CustomFeeModel())
if (x.Symbol.SecurityType != SecurityType.Equity) or (x.Symbol.Value=='SPY'): continue
option = self.AddOption(x.Symbol.Value, Resolution.Minute)
option.SetFilter(-20, +20, timedelta(DAYS_TO_EXPIRY-15), timedelta(DAYS_TO_EXPIRY+15))
option.PriceModel = OptionPriceModels.CrankNicolsonFD()
self.option_symbols[x.Symbol] = option.Symbol
for x in changes.RemovedSecurities:
if (x.Symbol.SecurityType != SecurityType.Equity) or (x.Symbol.Value=='SPY'): continue
self.RemoveOptionContract(self.option_symbols[x.Symbol])
#---------------------------------------------------------------------------------
def CoarseFilterFunction(self, coarse: List[CoarseFundamental]) -> List[Symbol]:
# We are going to use a dictionary to refer the object that will keep the moving averages
onlyEquities = [x for x in coarse if x.HasFundamentalData]
if self.IsWarmingUp:
return []
if (self.Time.date() not in self.equity_store.index):
return []
today_equities = self.equity_store.loc[self.Time.date()].iloc[0]
self.selected = [x.Symbol for x in onlyEquities if x.Symbol.Value in today_equities]
invested = [x.Key for x in self.Portfolio if x.Value.Invested]
# we need to return only the symbol objects
return self.selected + invested
#-----------------------------------------------------------------------------------------
# def OnOrderEvent(self, orderEvent: OrderEvent) -> None:
# order = self.Transactions.GetOrderById(orderEvent.OrderId)
# if orderEvent.Status == OrderStatus.Filled:
# self.Debug(f"{self.Time}: {order.Type}: {orderEvent}: {orderEvent.Symbol} Filled at {orderEvent.FillPrice}")
#----------------------------------------------------------------------------------------
def OnEndOfDay(self):
self.closed_today = []
self.Log(f'Total Securities in Universe: {len(self.ActiveSecurities)}')
#--------------------- Exit before Expiry and Dividend --------------------------------------
def ExitBeforeExpiry(self):
to_be_removed = []
for (contract1,contract2) in self.open_bulls.keys():
slice = self.CurrentSlice
security1 = self.Securities[contract1]
security2 = self.Securities[contract2]
if (security1.Expiry.date()==self.Time.date()) or (slice.Dividends.get(contract1.Underlying)):
self.ExitOption(contract1.Underlying,contract1,'ExitBeforeExpiry')
self.ExitOption(contract2.Underlying,contract2,'ExitBeforeExpiry')
to_be_removed.append((contract1,contract2))
self.Log(f'{self.Time}--> Exited before Expiry {security1.Expiry} of {contract1}')
self.Log(f'{self.Time}--> Exited before Expiry {security2.Expiry} of {contract2}')
for contracts in to_be_removed:
del self.open_bulls[contracts]
#------------------------------------------------------------------------------------------
def ExitOption(self,symbol,contract,msg):
self.Liquidate(contract, msg)
#--------------------------------------------------------------------------------------
class CustomFeeModel:
def GetOrderFee(self, parameters):
fee = 0
return OrderFee(CashAmount(fee, 'USD'))
#region imports
from AlgorithmImports import *
from constants import *
#endregion
# Your New Python File
class FixedStopRMModel(RiskManagementModel):
'''Provides an implementation of IRiskManagementModel that limits the maximum possible loss
measured from the highest unrealized profit
- Uses Fixed Stop
'''
def __init__(self, algo):
'''Initializes a new instance of the FixedStopRMModel class
Args:
maximumDrawdownPercent: The maximum percentage drawdown allowed for algorithm portfolio
compared with the highest unrealized profit, defaults to 5% drawdown'''
self.algo = algo
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
'''
riskAdjustedTargets = list()
bulls = [x for x in self.algo.open_bulls.keys()]
for (self.sym1,self.sym2) in bulls:
if algorithm.Time == self.algo.open_bulls[(self.sym1,self.sym2)]:
continue
self.security1 = algorithm.Securities[self.sym1]
self.security2 = algorithm.Securities[self.sym2]
# Remove if not invested
if not self.security1.Invested or not self.security2.Invested:
continue
profit1 = self.security1.Holdings.UnrealizedProfit
value1 = self.security1.Holdings.HoldingsCost
profit2 = self.security2.Holdings.UnrealizedProfit
value2 = self.security2.Holdings.HoldingsCost
self.profitPercent = (profit1 + profit2)/(value1+value2)
stop = FIXED_SL*-1
target = FIXED_TARGET
if (self.profitPercent <= stop):
riskAdjustedTargets = self.ExitOptions(algorithm, riskAdjustedTargets,criteria='stop')
elif (self.profitPercent >= target):
riskAdjustedTargets = self.ExitOptions(algorithm, riskAdjustedTargets,criteria='target')
return riskAdjustedTargets
def ExitOptions(self, algorithm, riskAdjustedTargets,criteria):
algorithm.Log(f'''Liquidated as spread for {(self.sym1.Value,self.sym2.Value)} reached
{criteria} at PnL%: {self.profitPercent} at Underlying: {self.security1.Underlying.Price}''')
riskAdjustedTargets.append(PortfolioTarget(self.sym1, 0))
riskAdjustedTargets.append(PortfolioTarget(self.sym2, 0))
del self.algo.open_bulls[(self.sym1,self.sym2)]
self.algo.closed_today.append((self.sym1,self.sym2))
return riskAdjustedTargets