| Overall Statistics |
|
Total Orders 1278 Average Win 0.04% Average Loss -0.03% Compounding Annual Return -0.704% Drawdown 8.900% Expectancy -0.476 Start Equity 1000000000000 End Equity 912203995110 Net Profit -8.780% Sharpe Ratio -3.608 Sortino Ratio -5.01 Probabilistic Sharpe Ratio 0.000% Loss Rate 77% Win Rate 23% Profit-Loss Ratio 1.31 Alpha -0.02 Beta -0.033 Annual Standard Deviation 0.007 Annual Variance 0 Information Ratio -0.807 Tracking Error 0.141 Treynor Ratio 0.712 Total Fees $0.00 Estimated Strategy Capacity $0 Lowest Capacity Asset SPX 32NKZDDUVN5NY|SPX 31 Portfolio Turnover 0.01% |
#region imports
from AlgorithmImports import *
from collections import deque, namedtuple
#endregion
Trade = namedtuple('Trade', ['Quantity', 'Start_Date', 'End_Date', 'Start_Price', 'End_Price', 'Underlying_Start_Price', 'Underlying_End_Price'], defaults=(None,) * 2)
class AssetTradeData:
def __init__(self, symbol):
self._symbol = symbol
self._buys = {}
self._sells = {}
def update_buy(self, time, order):
if time in self._buys:
raise ValueError('Given time already exists. Please delete the existing element first')
self._buys[time] = order
def update_sell(self, time, order):
if time in self._sells:
raise ValueError('Given time already exists. Please delete the existing element first')
self._sells[time] = order
@property
def buys(self):
return self._buys
@property
def sells(self):
return self._sells
def pair_all_trades(self):
buys: dict[DateTime, Order] = self.buys
sells: dict[DateTime, Order] = self.sells
all_trade_dates = sorted(list(buys.keys()) + list(sells.keys()))
current_total = 0
open_buys: Deque[Order] = deque()
open_sells: Deque[Order] = deque()
trades: List[Trade] = []
for date in all_trade_dates:
if date in buys:
# ToDo: sanity check that can be removed (may be allowed to buy and sell on same day for certain strategies)
if date in sells:
print(date)
#assert date not in sells
# Need a deep copy as we are going to change the order
order = buys[date].Clone()
# ToDo: sanity check that should be removed in future.
assert order.Quantity > 0
trades += self._pair_trades(open_sells, order, trades)
# ToDo: sanity check that should be removed in future.
assert order.Quantity >= 0
if order.Quantity > 0:
assert len(open_sells) == 0
open_buys.append(order)
else: #date in sales
# Need a deep copy as we are going to change the order
order = sells[date].Clone()
assert order.Quantity < 0
trades += self._pair_trades(open_buys, order, trades)
assert order.Quantity <= 0
if order.Quantity < 0:
assert len(open_buys) == 0
open_sells.append(order)
assert len(open_buys) == 0 or len(open_sells) == 0
return trades
def _pair_trades(self, open_trades: List[Order], order: Order, trades: List[Trade]) -> List[Trade]:
'''
Pairs closed trades (new balance 0) from order to open_trades in LIFO style (buy and sell order pairing).
The trades quantity is negative if it is a short trade and positive otherwise.
'''
# Note: this procedure changes the variables that it receives. Do not call directly from outside the class.
paired_trades = []
while len(open_trades) > 0 and abs(open_trades[-1].Quantity) <= abs(order.Quantity) and order.Quantity != 0:
# ToDo: sanity check that should be removed in future.
assert np.sign(open_trades[-1].Quantity * order.Quantity) < 0
trade = self._create_trade(open_trades[-1].Quantity, open_trades[-1], order)
paired_trades.append(trade)
order.Quantity += open_trades[-1].Quantity
open_trades.pop()
# last trade to match in case current order smaller than last existing order
if abs(order.Quantity) > 0 and len(open_trades) > 0:
assert abs(open_trades[-1].Quantity) > abs(order.Quantity)
assert np.sign(open_trades[-1].Quantity * order.Quantity) < 0
trade = self._create_trade(order.Quantity, open_trades[-1], order)
paired_trades.append(trade)
open_trades[-1].Quantity += order.Quantity
order.Quantity = 0
assert open_trades[-1].Quantity != 0
return paired_trades
def _create_trade(self, quantity, start_order, end_order):
return Trade(
quantity,
start_order.LastFillTime,
end_order.LastFillTime,
start_order.Price,
end_order.Price
)
class Trades:
def __init__(self, orders):
self._order_data_by_symbol: dict[Symbol, Any] = {}
orders = [order.Clone() for order in orders]
self._orders: List[Order] = sorted(orders, key=lambda x: x.LastFillTime)
# if data is not given in Eastern timezone convert and get read of timezone info
self._timezone = pytz.timezone('US/Eastern')
self._convert_order_timezone()
self._add_extra_order_data()
# ToDo: Restructure the design to be part of the Asset Data by Smbol class
self._paired_order_data_dict: dict[Symbol, Any] = {}
self.all_paired_data: List[Trade] = []
self.update_orders(orders)
for symbol in self._order_data_by_symbol:
self._paired_order_data_dict[symbol] = self._order_data_by_symbol[symbol].pair_all_trades()
def _convert_order_timezone(self):
for order in self._orders:
if order.LastFillTime.tzinfo is not None:
order.LastFillTime = order.LastFillTime.astimezone(self._timezone).replace(tzinfo=None)
def _add_extra_order_data(self):
# to be overriden in child classes in case some extra data need to be added to orders
pass
def update_orders(self, orders):
for order in orders:
if order.Symbol not in self._order_data_by_symbol:
self._order_data_by_symbol[order.Symbol] = AssetTradeData(order.Symbol)
order_data = self._order_data_by_symbol[order.Symbol]
is_buy = order.Quantity > 0
if is_buy:
order_data.update_buy(time=order.LastFillTime, order=order)
else:
order_data.update_sell(time=order.LastFillTime, order=order)
def __getitem__(self, symbol):
return self._order_data_by_symbol[symbol]
class OptionTrades(Trades):
def __init__(self, orders: List[Order], algo):
for order in orders:
assert order.SecurityType == SecurityType.Option or order.SecurityType == SecurityType.IndexOption
#self._orders = orders
self._underlying_to_orders_dict: Dict[Symbol, List[Order]]
self._algo = algo
super().__init__(orders)
def _add_extra_order_data(self):
'''
Overrides method called in constructor in base class to add underlying data to order data
'''
self._underlying_to_orders_dict = self._create_underlying_order_correspondence()
self._get_underlying_prices()
@property
def underlyings(self):
return list(self._underlying_to_orders_dict.keys())
def _create_underlying_order_correspondence(self):
underlying = {}
for order in self._orders:
if order.Symbol.Underlying not in underlying:
underlying[order.Symbol.Underlying] = [order]
else:
underlying[order.Symbol.Underlying].append(order)
return underlying
def _get_underlying_prices(self):
'''
Adds underlying price to every order as a field UnderlyingPrice.
ToDo: fix dirty solution of adding an extra field to each instance of orders.
'''
for symbol in self._underlying_to_orders_dict:
orders = sorted(self._underlying_to_orders_dict[symbol], key=lambda x: x.LastFillTime)
min_date = orders[0].LastFillTime
max_date = orders[-1].LastFillTime + timedelta(days=1)
df = self._algo.History(symbol, min_date, max_date, Resolution.Hour)
df.index = df.index.droplevel(0)
for order in orders:
#ToDo: Handle below case properly
if order.Type == OrderType.OptionExercise:
order.UnderlyingPrice = None
continue
order.UnderlyingPrice = df.loc[order.LastFillTime, 'close']
def _create_trade(self, quantity, start_order, end_order):
assert start_order.UnderlyingPrice is not None
return Trade(
quantity,
start_order.LastFillTime,
end_order.LastFillTime,
start_order.Price,
end_order.Price,
start_order.UnderlyingPrice,
end_order.UnderlyingPrice
)
def get_df_order_vs_underlying(self, underlying, order_direction = None):
if underlying not in self._underlying_to_orders_dict:
raise ValueError(f"{Symbol} is not in underlyings")
orders = self._underlying_to_orders_dict[underlying]
columns = ['underlying_price', 'order_price', 'percent', 'quantity', 'strike']
index: List[datetime] = [order.LastFillTime for order in orders if order_direction is None or order_direction == order.Direction]
data = np.empty((len(index),len(columns)), dtype='float')
res = pd.DataFrame(columns=columns, index=index, data=data)
for i, order in enumerate(orders):
if order_direction is None or order_direction == order.Direction:
res.iloc[i]['underlying_price'] = order.UnderlyingPrice
res.iloc[i]['order_price'] = order.Price
res.iloc[i]['percent'] = (order.Price / order.UnderlyingPrice) * 100 if order.UnderlyingPrice is not None else None
res.iloc[i]['quantity'] = order.Quantity
res.iloc[i]['strike'] = order.Symbol.ID.StrikePrice
return res
#region imports
from AlgorithmImports import *
from options import *
#endregion
def default_strategy(algo: QCAlgorithm):
monetize_60_day_before_expiry = MonetizeFixedTimeBeforeExpiry(algo=algo, min_days_to_expiry=60)
monetize_90_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=0.9)
monetize_105_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.05)
monetize_110_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.1)
monetize_100_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.0)
monetize_delta_40 = MonetizeByDelta(algo=algo, delta=0.4)
monetize_delta_45 = MonetizeByDelta(algo=algo, delta=0.45)
monetize_delta_50 = MonetizeByDelta(algo=algo, delta=0.5)
monetize_delta_55 = MonetizeByDelta(algo=algo, delta=0.55)
monetize_delta_60 = MonetizeByDelta(algo=algo, delta=0.6)
monetization_list = [monetize_60_day_before_expiry, monetize_90_percent_in_money]
monetize_list = ChainMonetizations(algo=algo, monetizations_list=monetization_list)
spy = algo.AddEquity('SPY').Symbol
spx = algo.AddIndex('SPX').Symbol
strategy = {
'Type': 'Sell',
'Days to Expiry': 360,
'Strike': 1.05,
#'Amount Calculator': FractionToExpiryAmountCalculator(algo=algo, notional_frac=1.0, multiplier=1.5),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=((5*1.5)/180)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=1.5/30),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=1),
'Right': OptionRight.PUT,
'underlying': spx,
#'underlying': spy,
'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spy, 'SPY', Market.USA, '?SPY')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPXW', Market.USA, '?SPXW'),
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPXW', Market.USA, '?SPXW')],
'Resolution': Resolution.MINUTE,
'Execution Time': 90,
'Options Filter Time': 120,
'Before Target Expiry': False,
#'Monetization': monetize_110_percent_in_money,
#'Monetization': monetize_delta_45,
'Monetization': None,
#'Filter': VixFilter(algo=algo, threshold=1.1, days=90),
#'filter': VixThresholdFilter(algo=algo, threshold=20),
'filter': None,
'date_schedule_rule' : algo.DateRules.EveryDay,
#'date_schedule_rule' : algo.DateRules.WeekStart,
'time_schedule_rule': algo.TimeRules.BeforeMarketClose,
'buy_class': BuyByStrikeAndExpiry
}
return strategy
def default_buy_by_delta_strategy(algo: QCAlgorithm):
spy = algo.AddEquity('SPY').Symbol
strategy = default_strategy(algo)
strategy['buy_class'] = BuyByDeltaAndExpiry
strategy.pop('Strike')
strategy['target_delta'] = 0.05
return strategy
def sell_call_strategy(algo: QCAlgorithm()):
algo.set_name('sell_call_105_30')
strategy = default_buy_by_delta_strategy(algo)
strategy['Right'] = OptionRight.CALL
strategy['amount_calculator'] = FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(1/4), cap=None)
strategy['date_schedule_rule'] = algo.DateRules.WeekStart
strategy['time_schedule_rule'] = algo.TimeRules.BeforeMarketClose
strategy['buy_class'] = BuyByStrikeAndExpiry
strategy['Strike'] = 1.05
return strategy
def sell_call_by_delta_strategy(algo: QCAlgorithm()):
algo.set_name('sell_call_delta_5_360')
strategy = default_buy_by_delta_strategy(algo)
strategy['Right'] = OptionRight.CALL
strategy['amount_calculator'] = FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(1/52), cap=None)
strategy['date_schedule_rule'] = algo.DateRules.WeekStart
strategy['time_schedule_rule'] = algo.TimeRules.BeforeMarketClose
return strategy
def default_daily_dynamic_hedge_strategy3(algo: QCAlgorithm):
monetize_60_day_before_expiry = MonetizeFixedTimeBeforeExpiry(algo=algo, min_days_to_expiry=60)
monetize_90_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=0.9)
monetize_105_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.05)
monetize_110_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.1)
monetize_100_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.0)
monetize_delta_40 = MonetizeByDelta(algo=algo, delta=0.4)
monetize_delta_45 = MonetizeByDelta(algo=algo, delta=0.45)
monetize_delta_50 = MonetizeByDelta(algo=algo, delta=0.5)
monetize_delta_55 = MonetizeByDelta(algo=algo, delta=0.55)
monetize_delta_60 = MonetizeByDelta(algo=algo, delta=0.6)
monetize_delta_65 = MonetizeByDelta(algo=algo, delta=0.65)
monetize_delta_70 = MonetizeByDelta(algo=algo, delta=0.7)
monetization_list = [monetize_60_day_before_expiry, monetize_90_percent_in_money]
monetize_list = ChainMonetizations(algo=algo, monetizations_list=monetization_list)
spy = algo.AddEquity('SPY').Symbol
spx = algo.AddIndex('SPX').Symbol
algo.set_name('hedge_90_90_cap_15_mon_55')
intercept: float = 0.14416307
beta: float = -0.21567544
strategy = {
'Type': 'Buy',
'Days to Expiry': 90,
'Strike': 0.9,
'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(4*5*(1.5)/360), cap=0.015 / 52),
'Right': OptionRight.Put,
'underlying': spx,
'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
'Resolution': Resolution.MINUTE,
'Execution Time': 60,
'Options Filter Time': 120,
'Before Target Expiry': False,
'Monetization': monetize_delta_55,
'filter': None,
'date_schedule_rule' : algo.DateRules.week_start,
'time_schedule_rule': algo.TimeRules.before_market_close,
'buy_class': BuyByVIX,
'moneyness_fn': staticmethod(lambda vix: -2. - (vix / 6.)),
# 'moneyness_fn': staticmethod(lambda vix: intercept + beta * vix),
}
return strategy
def default_daily_dynamic_hedge_strategy4(algo: QCAlgorithm):
monetize_60_day_before_expiry = MonetizeFixedTimeBeforeExpiry(algo=algo, min_days_to_expiry=60)
monetize_90_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=0.9)
monetize_105_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.05)
monetize_110_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.1)
monetize_100_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.0)
monetize_delta_40 = MonetizeByDelta(algo=algo, delta=0.4)
monetize_delta_45 = MonetizeByDelta(algo=algo, delta=0.45)
monetize_delta_50 = MonetizeByDelta(algo=algo, delta=0.5)
monetize_delta_55 = MonetizeByDelta(algo=algo, delta=0.55)
monetize_delta_60 = MonetizeByDelta(algo=algo, delta=0.6)
monetize_delta_65 = MonetizeByDelta(algo=algo, delta=0.65)
monetize_delta_70 = MonetizeByDelta(algo=algo, delta=0.7)
monetization_list = [monetize_60_day_before_expiry, monetize_90_percent_in_money]
monetize_list = ChainMonetizations(algo=algo, monetizations_list=monetization_list)
spy = algo.AddEquity('SPY').Symbol
spx = algo.AddIndex('SPX').Symbol
algo.set_name('hedge_90_90_cap_15_mon_55')
# predetermined coefs
intercept: float = 0.14416307
beta: float = -0.21567544
strategy = {
'Type': 'Buy',
'Days to Expiry': 90,
'Strike': 0.9,
'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(4*5*(1.5)/360), cap=0.015 / 52),
'Right': OptionRight.Put,
'underlying': spx,
'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
'Resolution': Resolution.MINUTE,
'Execution Time': 60,
'Options Filter Time': 120,
'Before Target Expiry': False,
'Monetization': monetize_delta_55,
'filter': None,
'date_schedule_rule' : algo.DateRules.week_start,
'time_schedule_rule': algo.TimeRules.before_market_close,
'buy_class': BuyByVIXRegression,
'default_intercept': intercept,
'default_beta': beta,
'coef_date_schedule_rule' : algo.date_rules.year_start,
'coef_time_schedule_rule': algo.time_rules.before_market_close,
'coef_min_daily_period': 30,
}
return strategy
def default_daily_dynamic_hedge_strategy(algo: QCAlgorithm):
monetize_60_day_before_expiry = MonetizeFixedTimeBeforeExpiry(algo=algo, min_days_to_expiry=60)
monetize_90_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=0.9)
monetize_105_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.05)
monetize_110_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.1)
monetize_100_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.0)
monetize_delta_40 = MonetizeByDelta(algo=algo, delta=0.4)
monetize_delta_45 = MonetizeByDelta(algo=algo, delta=0.45)
monetize_delta_50 = MonetizeByDelta(algo=algo, delta=0.5)
monetize_delta_55 = MonetizeByDelta(algo=algo, delta=0.55)
monetize_delta_60 = MonetizeByDelta(algo=algo, delta=0.6)
monetize_delta_65 = MonetizeByDelta(algo=algo, delta=0.65)
monetize_delta_70 = MonetizeByDelta(algo=algo, delta=0.7)
monetization_list = [monetize_60_day_before_expiry, monetize_90_percent_in_money]
monetize_list = ChainMonetizations(algo=algo, monetizations_list=monetization_list)
spy = algo.AddEquity('SPY').Symbol
spx = algo.AddIndex('SPX').Symbol
algo.set_name('hedge_90_90_cap_15_mon_55')
strategy = {
'Type': 'Buy',
'Days to Expiry': 90,
'Strike': 0.9,
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/180)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=((1.5)/360)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(1/5)),
#'amount_calculator': FractionOfHoldingAmountCalculator(algo=algo, notional_frac=1.0, multiplier= 5 * (1.5/360), holding_symbol=spy, cap=0.01/52),
#'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/180)),
'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(4*5*(1.5)/360), cap=0.015 / 52),
#'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(2*5*(1.5)/180), cap=None),
'Right': OptionRight.Put,
'underlying': spx,
#'underlying': spy,
'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spy, 'SPY', Market.USA, '?SPY')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPXW', Market.USA, '?SPXW'),
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
'Resolution': Resolution.MINUTE,
'Execution Time': 60,
'Options Filter Time': 120,
'Before Target Expiry': False,
#'Monetization': monetize_110_percent_in_money,
#'Monetization': monetize_delta_50,
#'Monetization': monetize_delta_45,
'Monetization': monetize_delta_55,
#'Monetization': monetize_delta_50,
#'Monetization': monetize_delta_40,
#'Monetization': monetize_delta_60,
#'Monetization': monetize_delta_65,
#'Monetization': monetize_delta_70,
#'Monetization': None,
#'Filter': VixFilter(algo=algo, threshold=1.1, days=90),
#'filter': VixThresholdFilter(algo=algo, threshold=20),
'filter': None,
#'date_schedule_rule' : algo.DateRules.EveryDay,
'date_schedule_rule' : algo.DateRules.WeekStart,
'time_schedule_rule': algo.TimeRules.BeforeMarketClose,
'buy_class': BuyByStrikeAndExpiry
}
return strategy
def default_daily_dynamic_hedge_strategy1(algo: QCAlgorithm):
monetize_60_day_before_expiry = MonetizeFixedTimeBeforeExpiry(algo=algo, min_days_to_expiry=60)
monetize_90_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=0.9)
monetize_105_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.05)
monetize_110_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.1)
monetize_100_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.0)
monetize_delta_40 = MonetizeByDelta(algo=algo, delta=0.4)
monetize_delta_45 = MonetizeByDelta(algo=algo, delta=0.45)
monetize_delta_50 = MonetizeByDelta(algo=algo, delta=0.5)
monetize_delta_55 = MonetizeByDelta(algo=algo, delta=0.55)
monetize_delta_60 = MonetizeByDelta(algo=algo, delta=0.6)
monetization_list = [monetize_60_day_before_expiry, monetize_90_percent_in_money]
monetize_list = ChainMonetizations(algo=algo, monetizations_list=monetization_list)
spy = algo.AddEquity('SPY').Symbol
spx = algo.AddIndex('SPX').Symbol
strategy = {
'Type': 'Buy',
'Days to Expiry': 360,
'Strike': 0.8,
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=((1.5)/180)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=((1.5)/360)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(1/5)),
#'amount_calculator': FractionOfHoldingAmountCalculator(algo=algo, notional_frac=1.0, multiplier=0.33 * 1.5/360, holding_symbol=spy),
'amount_calculator': FractionOfHoldingAmountCalculator(algo=algo, notional_frac=1.0, multiplier= 1/52, holding_symbol=spy, cap=0.01 / 52),
'Right': OptionRight.Put,
'underlying': spx,
#'underlying': spy,
'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spy, 'SPY', Market.USA, '?SPY')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPXW', Market.USA, '?SPXW'),
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
'Resolution': Resolution.MINUTE,
'Execution Time': 60,
'Options Filter Time': 120,
'Before Target Expiry': False,
#'Monetization': monetize_110_percent_in_money,
'Monetization': monetize_delta_45,
#'Monetization': None,
#'Filter': VixFilter(algo=algo, threshold=1.1, days=90),
#'filter': VixThresholdFilter(algo=algo, threshold=20),
'filter': None,
'date_schedule_rule' : algo.DateRules.EveryDay,
#'date_schedule_rule' : algo.DateRules.WeekStart,
'time_schedule_rule': algo.TimeRules.BeforeMarketClose,
'buy_class': BuyByStrikeAndExpiry
}
return strategy
def main_leg_spread(algo: QCAlgorithm):
monetize_60_day_before_expiry = MonetizeFixedTimeBeforeExpiry(algo=algo, min_days_to_expiry=60)
monetize_90_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=0.9)
monetize_105_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.05)
monetize_110_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.1)
monetize_100_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.0)
monetize_delta_40 = MonetizeByDelta(algo=algo, delta=0.4)
monetize_delta_45 = MonetizeByDelta(algo=algo, delta=0.45)
monetize_delta_50 = MonetizeByDelta(algo=algo, delta=0.5)
monetize_delta_55 = MonetizeByDelta(algo=algo, delta=0.55)
monetize_delta_60 = MonetizeByDelta(algo=algo, delta=0.6)
monetize_delta_65 = MonetizeByDelta(algo=algo, delta=0.65)
monetize_delta_70 = MonetizeByDelta(algo=algo, delta=0.7)
monetization_list = [monetize_60_day_before_expiry, monetize_90_percent_in_money]
monetize_list = ChainMonetizations(algo=algo, monetizations_list=monetization_list)
spy = algo.AddEquity('SPY').Symbol
spx = algo.AddIndex('SPX').Symbol
strategy = {
'Type': 'Buy',
'Days to Expiry': 360,
'Strike': 0.85,
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/180)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=((1.5)/360)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(1/5)),
#'amount_calculator': FractionOfHoldingAmountCalculator(algo=algo, notional_frac=1.0, multiplier= 5 * (1.5/360), holding_symbol=spy, cap=0.01/52),
#'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/180)),
#'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(2*5*(1.5)/360), cap=0.015 / 52),
'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/360), cap=None),
'Right': OptionRight.Put,
'underlying': spx,
#'underlying': spy,
'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spy, 'SPY', Market.USA, '?SPY')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPXW', Market.USA, '?SPXW'),
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
'Resolution': Resolution.MINUTE,
'Execution Time': 60,
'Options Filter Time': 120,
'Before Target Expiry': False,
#'Monetization': monetize_110_percent_in_money,
#'Monetization': monetize_delta_50,
#'Monetization': monetize_delta_45,
#'Monetization': monetize_delta_55,
#'Monetization': monetize_delta_50,
'Monetization': monetize_delta_40,
#'Monetization': monetize_delta_60,
#'Monetization': monetize_delta_65,
#'Monetization': monetize_delta_70,
#'Monetization': None,
#'Filter': VixFilter(algo=algo, threshold=1.1, days=90),
#'filter': VixThresholdFilter(algo=algo, threshold=20),
'filter': None,
#'date_schedule_rule' : algo.DateRules.EveryDay,
'date_schedule_rule' : algo.DateRules.WeekStart,
'time_schedule_rule': algo.TimeRules.BeforeMarketClose,
'buy_class': BuyByStrikeAndExpiry
}
return strategy
def secondary_leg_spread(algo: QCAlgorithm):
monetize_60_day_before_expiry = MonetizeFixedTimeBeforeExpiry(algo=algo, min_days_to_expiry=60)
monetize_90_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=0.9)
monetize_105_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.05)
monetize_110_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.1)
monetize_100_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.0)
monetize_delta_40 = MonetizeByDelta(algo=algo, delta=0.4)
monetize_delta_45 = MonetizeByDelta(algo=algo, delta=0.45)
monetize_delta_50 = MonetizeByDelta(algo=algo, delta=0.5)
monetize_delta_55 = MonetizeByDelta(algo=algo, delta=0.55)
monetize_delta_60 = MonetizeByDelta(algo=algo, delta=0.6)
monetize_delta_65 = MonetizeByDelta(algo=algo, delta=0.65)
monetize_delta_70 = MonetizeByDelta(algo=algo, delta=0.7)
monetize_delta_75 = MonetizeByDelta(algo=algo, delta=0.7)
monetization_list = [monetize_60_day_before_expiry, monetize_90_percent_in_money]
monetize_list = ChainMonetizations(algo=algo, monetizations_list=monetization_list)
spy = algo.AddEquity('SPY').Symbol
spx = algo.AddIndex('SPX').Symbol
strategy = {
'Type': 'Sell',
'Days to Expiry': 360,
'Strike': 0.8,
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/180)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=((1.5)/360)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(1/5)),
#'amount_calculator': FractionOfHoldingAmountCalculator(algo=algo, notional_frac=1.0, multiplier= 5 * (1.5/360), holding_symbol=spy, cap=0.01/52),
#'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/180)),
#'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(2*5*(1.5)/360), cap=0.015 / 52),
'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/360), cap=None),
'Right': OptionRight.Put,
'underlying': spx,
#'underlying': spy,
'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spy, 'SPY', Market.USA, '?SPY')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPXW', Market.USA, '?SPXW'),
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
'Resolution': Resolution.MINUTE,
'Execution Time': 60,
'Options Filter Time': 120,
'Before Target Expiry': False,
#'Filter': VixFilter(algo=algo, threshold=1.1, days=90),
#'filter': VixThresholdFilter(algo=algo, threshold=20),
'filter': None,
#'date_schedule_rule' : algo.DateRules.EveryDay,
'date_schedule_rule' : algo.DateRules.WeekStart,
'time_schedule_rule': algo.TimeRules.BeforeMarketClose,
'buy_class': BuyByStrikeAndExpiry
}
return strategy
def default_spread_strategy(algo: QCAlgorithm):
monetize_60_day_before_expiry = MonetizeFixedTimeBeforeExpiry(algo=algo, min_days_to_expiry=60)
monetize_90_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=0.9)
monetize_105_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.05)
monetize_110_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.1)
monetize_100_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.0)
monetize_delta_40 = MonetizeByDelta(algo=algo, delta=0.4)
monetize_delta_45 = MonetizeByDelta(algo=algo, delta=0.45)
monetize_delta_50 = MonetizeByDelta(algo=algo, delta=0.5)
monetize_delta_55 = MonetizeByDelta(algo=algo, delta=0.55)
monetize_delta_60 = MonetizeByDelta(algo=algo, delta=0.6)
monetize_delta_65 = MonetizeByDelta(algo=algo, delta=0.65)
monetize_delta_70 = MonetizeByDelta(algo=algo, delta=0.7)
monetization_list = [monetize_60_day_before_expiry, monetize_90_percent_in_money]
monetize_list = ChainMonetizations(algo=algo, monetizations_list=monetization_list)
spy = algo.AddEquity('SPY').Symbol
spx = algo.AddIndex('SPX').Symbol
algo.set_name('spread_360_85_80_mon_1_leg')
strategy = {
'Type': 'Buy',
'Days to Expiry': 180,
'Strike': 0.85,
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/180)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=((1.5)/360)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(1/5)),
#'amount_calculator': FractionOfHoldingAmountCalculator(algo=algo, notional_frac=1.0, multiplier= 5 * (1.5/360), holding_symbol=spy, cap=0.01/52),
#'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/180)),
'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(2*5*(1.5)/360), cap=0.015 / 52),
#'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/360), cap=None),
'main_leg': main_leg_spread(algo),
'secondary_leg': secondary_leg_spread(algo),
'Right': OptionRight.Put,
'underlying': spx,
#'underlying': spy,
'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spy, 'SPY', Market.USA, '?SPY')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPXW', Market.USA, '?SPXW'),
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
'Resolution': Resolution.MINUTE,
'Execution Time': 60,
'Options Filter Time': 120,
'Before Target Expiry': False,
#'Monetization': monetize_110_percent_in_money,
'Monetization': MonetizeSpreadByDelta(algo, delta=0.5, monetize_connections=False),
#'Monetization': monetize_delta_45,
#'Monetization': monetize_delta_55,
#'Monetization': monetize_delta_50,
#'Monetization': monetize_delta_40,
#'Monetization': monetize_delta_60,
#'Monetization': monetize_delta_65,
#'Monetization': monetize_delta_70,
#'Monetization': None,
#'Filter': VixFilter(algo=algo, threshold=1.1, days=90),
#'filter': VixThresholdFilter(algo=algo, threshold=20),
'filter': None,
#'date_schedule_rule' : algo.DateRules.EveryDay,
'date_schedule_rule' : algo.DateRules.WeekStart,
'time_schedule_rule': algo.TimeRules.BeforeMarketClose,
'buy_class': BuySpread
}
return strategy
def default_collar_strategy(algo: QCAlgorithm):
monetize_60_day_before_expiry = MonetizeFixedTimeBeforeExpiry(algo=algo, min_days_to_expiry=60)
monetize_90_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=0.9)
monetize_105_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.05)
monetize_110_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.1)
monetize_100_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.0)
monetize_delta_40 = MonetizeByDelta(algo=algo, delta=0.4)
monetize_delta_45 = MonetizeByDelta(algo=algo, delta=0.45)
monetize_delta_50 = MonetizeByDelta(algo=algo, delta=0.5)
monetize_delta_55 = MonetizeByDelta(algo=algo, delta=0.55)
monetize_delta_60 = MonetizeByDelta(algo=algo, delta=0.6)
monetize_delta_65 = MonetizeByDelta(algo=algo, delta=0.65)
monetize_delta_70 = MonetizeByDelta(algo=algo, delta=0.7)
monetization_list = [monetize_60_day_before_expiry, monetize_90_percent_in_money]
monetize_list = ChainMonetizations(algo=algo, monetizations_list=monetization_list)
spy = algo.AddEquity('SPY').Symbol
spx = algo.AddIndex('SPX').Symbol
algo.set_name('spread_360_85_80_mon_1_leg')
strategy = {
'Type': 'Buy',
'Days to Expiry': 180,
'Strike': 0.85,
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/180)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=((1.5)/360)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(1/5)),
#'amount_calculator': FractionOfHoldingAmountCalculator(algo=algo, notional_frac=1.0, multiplier= 5 * (1.5/360), holding_symbol=spy, cap=0.01/52),
#'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/180)),
'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(2*5*(1.5)/360), cap=0.015 / 52),
#'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/360), cap=None),
'main_leg': main_leg_spread(algo),
'secondary_leg': secondary_leg_spread(algo),
'Right': OptionRight.Put,
'underlying': spx,
#'underlying': spy,
'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spy, 'SPY', Market.USA, '?SPY')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPXW', Market.USA, '?SPXW'),
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
'Resolution': Resolution.MINUTE,
'Execution Time': 60,
'Options Filter Time': 120,
'Before Target Expiry': False,
#'Monetization': monetize_110_percent_in_money,
'Monetization': MonetizeSpreadByDelta(algo, delta=0.5, monetize_connections=False),
#'Monetization': monetize_delta_45,
#'Monetization': monetize_delta_55,
#'Monetization': monetize_delta_50,
#'Monetization': monetize_delta_40,
#'Monetization': monetize_delta_60,
#'Monetization': monetize_delta_65,
#'Monetization': monetize_delta_70,
#'Monetization': None,
#'Filter': VixFilter(algo=algo, threshold=1.1, days=90),
#'filter': VixThresholdFilter(algo=algo, threshold=20),
'filter': None,
#'date_schedule_rule' : algo.DateRules.EveryDay,
'date_schedule_rule' : algo.DateRules.WeekStart,
'time_schedule_rule': algo.TimeRules.BeforeMarketClose,
'buy_class': BuyCollar
}
return strategy
def default_daily_dynamic_hedge_strategy2(algo: QCAlgorithm):
monetize_60_day_before_expiry = MonetizeFixedTimeBeforeExpiry(algo=algo, min_days_to_expiry=60)
monetize_90_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=0.9)
monetize_105_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.05)
monetize_110_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.1)
monetize_100_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.0)
monetize_delta_40 = MonetizeByDelta(algo=algo, delta=0.4)
monetize_delta_45 = MonetizeByDelta(algo=algo, delta=0.45)
monetize_delta_50 = MonetizeByDelta(algo=algo, delta=0.5)
monetize_delta_55 = MonetizeByDelta(algo=algo, delta=0.55)
monetize_delta_60 = MonetizeByDelta(algo=algo, delta=0.6)
monetization_list = [monetize_60_day_before_expiry, monetize_90_percent_in_money]
monetize_list = ChainMonetizations(algo=algo, monetizations_list=monetization_list)
spy = algo.AddEquity('SPY').Symbol
spx = algo.AddIndex('SPX').Symbol
strategy = {
'Type': 'Buy',
'Days to Expiry': 360,
'Strike': 0.9,
'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=((1.5)/180)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=((1.5)/360)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(1/5)),
#'amount_calculator': FractionOfHoldingAmountCalculator(algo=algo, notional_frac=1.0, multiplier= 0.33 * (1.5/180), holding_symbol=spy),
'Right': OptionRight.Put,
'underlying': spx,
#'underlying': spy,
'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spy, 'SPY', Market.USA, '?SPY')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPXW', Market.USA, '?SPXW'),
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
'Resolution': Resolution.MINUTE,
'Execution Time': 60,
'Options Filter Time': 120,
'Before Target Expiry': False,
#'Monetization': monetize_110_percent_in_money,
'Monetization': monetize_delta_55,
#'Monetization': None,
#'Filter': VixFilter(algo=algo, threshold=1.1, days=90),
'filter': VixThresholdFilter(algo=algo, threshold=20),
#'Filter': None,
'date_schedule_rule' : algo.DateRules.EveryDay,
#'date_schedule_rule' : algo.DateRules.WeekStart,
'time_schedule_rule': algo.TimeRules.BeforeMarketClose,
'buy_class': BuyByStrikeAndExpiry
}
return strategy
def default_ETF_Dynamic_Collar(algo: QCAlgorithm):
### Remember to set the offset days to 1 in the options file (Tuesday trading)
monetize_60_day_before_expiry = MonetizeFixedTimeBeforeExpiry(algo=algo, min_days_to_expiry=60)
monetize_90_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=0.9)
monetize_105_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.05)
monetize_110_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.1)
monetize_100_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.0)
monetize_delta_40 = MonetizeByDelta(algo=algo, delta=0.4)
monetize_delta_45 = MonetizeByDelta(algo=algo, delta=0.45)
monetize_delta_50 = MonetizeByDelta(algo=algo, delta=0.5)
monetize_delta_55 = MonetizeByDelta(algo=algo, delta=0.55)
monetize_delta_60 = MonetizeByDelta(algo=algo, delta=0.6)
monetize_delta_65 = MonetizeByDelta(algo=algo, delta=0.65)
monetize_delta_70 = MonetizeByDelta(algo=algo, delta=0.7)
monetization_list = [monetize_60_day_before_expiry, monetize_90_percent_in_money]
monetize_list = ChainMonetizations(algo=algo, monetizations_list=monetization_list)
spy = algo.AddEquity('SPY').Symbol
spx = algo.AddIndex('SPX').Symbol
algo.set_name('ETF_Dynamic_Collar')
strategy = {
'Type': 'Buy',
'Days to Expiry': 180,
'Strike': 0.85,
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/180)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=((1.5)/360)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(1/5)),
#'amount_calculator': FractionOfHoldingAmountCalculator(algo=algo, notional_frac=1.0, multiplier= 5 * (1.5/360), holding_symbol=spy, cap=0.01/52),
#'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/180)),
'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(4*5*(1.5)/360), cap=0.03 / 52),
#'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/360), cap=None),
'secondary_budget_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(2/52), cap=None),
'main_leg': main_leg_collar(algo),
'secondary_leg': secondary_leg_collar(algo),
'Right': OptionRight.Put,
'underlying': spx,
#'underlying': spy,
'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spy, 'SPY', Market.USA, '?SPY')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPXW', Market.USA, '?SPXW'),
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
'Resolution': Resolution.MINUTE,
'Execution Time': 60,
'Options Filter Time': 120,
'Before Target Expiry': False,
#'secondary_notional_frac': (2*5*(1.5)/360),
'equalize_legs': False,
'cap_secondary_leg': False,
#'cap_secondary_leg_frac': 0.83,
'cost_target_pct': 0.005 / 52,
#'Monetization': monetize_110_percent_in_money,
'Monetization': MonetizeSpreadByDelta(algo, delta=0.5, monetize_connections=True),
#'Monetization': monetize_delta_45,
#'Monetization': monetize_delta_55,
#'Monetization': monetize_delta_50,
#'Monetization': monetize_delta_40,
#'Monetization': monetize_delta_60,
#'Monetization': monetize_delta_65,
#'Monetization': monetize_delta_70,
#'Monetization': None,
#'Filter': VixFilter(algo=algo, threshold=1.1, days=90),
'filter': VixThresholdFilter(algo=algo, threshold=30),
#'filter': None,
#'date_schedule_rule' : algo.DateRules.EveryDay,
'date_schedule_rule' : algo.DateRules.WeekStart,
'time_schedule_rule': algo.TimeRules.BeforeMarketClose,
'buy_class': BuyCollarByBudgetAlignment
}
return strategy
def main_leg_collar(algo: QCAlgorithm):
monetize_60_day_before_expiry = MonetizeFixedTimeBeforeExpiry(algo=algo, min_days_to_expiry=60)
monetize_90_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=0.9)
monetize_105_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.05)
monetize_110_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.1)
monetize_100_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.0)
monetize_delta_40 = MonetizeByDelta(algo=algo, delta=0.4)
monetize_delta_45 = MonetizeByDelta(algo=algo, delta=0.45)
monetize_delta_50 = MonetizeByDelta(algo=algo, delta=0.5)
monetize_delta_55 = MonetizeByDelta(algo=algo, delta=0.55)
monetize_delta_60 = MonetizeByDelta(algo=algo, delta=0.6)
monetize_delta_65 = MonetizeByDelta(algo=algo, delta=0.65)
monetize_delta_70 = MonetizeByDelta(algo=algo, delta=0.7)
monetization_list = [monetize_60_day_before_expiry, monetize_90_percent_in_money]
monetize_list = ChainMonetizations(algo=algo, monetizations_list=monetization_list)
spy = algo.AddEquity('SPY').Symbol
spx = algo.AddIndex('SPX').Symbol
strategy = {
'Type': 'Buy',
'Days to Expiry': 180,
'Strike': 0.85,
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/180)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=((1.5)/360)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(1/5)),
#'amount_calculator': FractionOfHoldingAmountCalculator(algo=algo, notional_frac=1.0, multiplier= 5 * (1.5/360), holding_symbol=spy, cap=0.01/52),
#'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/180)),
'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(4/52), cap=0.03 / 52),
#'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(2*5*(1.5)/360), cap=None),
'Right': OptionRight.Put,
'underlying': spx,
#'underlying': spy,
'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spy, 'SPY', Market.USA, '?SPY')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPXW', Market.USA, '?SPXW'),
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
'Resolution': Resolution.MINUTE,
'Execution Time': 60,
'Options Filter Time': 120,
'Before Target Expiry': False,
#'Monetization': monetize_110_percent_in_money,
#'Monetization': monetize_delta_50,
#'Monetization': monetize_delta_45,
#'Monetization': monetize_delta_55,
'Monetization': monetize_delta_50,
#'Monetization': monetize_delta_40,
#'Monetization': monetize_delta_60,
#'Monetization': monetize_delta_65,
#'Monetization': monetize_delta_70,
#'Monetization': None,
#'Filter': VixFilter(algo=algo, threshold=1.1, days=90),
#'filter': VixThresholdFilter(algo=algo, threshold=20),
'filter': None,
#'date_schedule_rule' : algo.DateRules.EveryDay,
'date_schedule_rule' : algo.DateRules.WeekStart,
'time_schedule_rule': algo.TimeRules.BeforeMarketClose,
'buy_class': BuyByStrikeAndExpiry
}
return strategy
def secondary_leg_collar(algo: QCAlgorithm):
monetize_60_day_before_expiry = MonetizeFixedTimeBeforeExpiry(algo=algo, min_days_to_expiry=60)
monetize_90_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=0.9)
monetize_105_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.05)
monetize_110_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.1)
monetize_100_percent_in_money = MonetizeRelativeToUnderlying(algo=algo, percent_of_underlying=1.0)
monetize_delta_40 = MonetizeByDelta(algo=algo, delta=0.4)
monetize_delta_45 = MonetizeByDelta(algo=algo, delta=0.45)
monetize_delta_50 = MonetizeByDelta(algo=algo, delta=0.5)
monetize_delta_55 = MonetizeByDelta(algo=algo, delta=0.55)
monetize_delta_60 = MonetizeByDelta(algo=algo, delta=0.6)
monetize_delta_65 = MonetizeByDelta(algo=algo, delta=0.65)
monetize_delta_70 = MonetizeByDelta(algo=algo, delta=0.7)
monetize_delta_75 = MonetizeByDelta(algo=algo, delta=0.7)
monetization_list = [monetize_60_day_before_expiry, monetize_90_percent_in_money]
monetize_list = ChainMonetizations(algo=algo, monetizations_list=monetization_list)
spy = algo.AddEquity('SPY').Symbol
spx = algo.AddIndex('SPX').Symbol
strategy = {
'Type': 'Sell',
'Days to Expiry': 180,
'Strike': 0.8,
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/180)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=((1.5)/360)),
#'amount_calculator': ConstantFractionAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(1/5)),
#'amount_calculator': FractionOfHoldingAmountCalculator(algo=algo, notional_frac=1.0, multiplier= 5 * (1.5/360), holding_symbol=spy, cap=0.01/52),
#'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(5*(1.5)/180)),
#'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(2*5*(1.5)/360), cap=0.03 / 52),
'amount_calculator': FractionOfCashAmountCalculator(algo=algo, notional_frac=1.0, multiplier=(2/52), cap=None),
'Right': OptionRight.Call,
'underlying': spx,
#'underlying': spy,
'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spy, 'SPY', Market.USA, '?SPY')],
#'canonical_option_symbols': [Symbol.CreateCanonicalOption(spx, 'SPXW', Market.USA, '?SPXW'),
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
# Symbol.CreateCanonicalOption(spx, 'SPX', Market.USA, '?SPX')],
'Resolution': Resolution.MINUTE,
'Execution Time': 60,
'Options Filter Time': 120,
'Before Target Expiry': False,
#'Filter': VixFilter(algo=algo, threshold=1.1, days=90),
#'filter': VixThresholdFilter(algo=algo, threshold=20),
'filter': None,
#'date_schedule_rule' : algo.DateRules.EveryDay,
'date_schedule_rule' : algo.DateRules.WeekStart,
'time_schedule_rule': algo.TimeRules.BeforeMarketClose,
'buy_class': BuyByStrikeAndExpiry
}
return strategy# region imports
from AlgorithmImports import *
import QuantLib as ql
# endregion
# using seconds from datetime difference so we can pass 0 day option also
def get_t_days(expiration_date: datetime, calc_date: datetime) -> float:
dt = (expiration_date - calc_date)
days_dt: float = dt.days
seconds_dt: float = dt.seconds / (24*60*60) # days
t: float = seconds_dt + days_dt
return t
def black_model(
option_price: float,
forward_price: float,
strike_price: float,
option_type: int,
expiration_date: datetime,
calc_date: datetime,
discount_factor: float = 1
) -> Tuple[float]:
implied_vol = ql.blackFormulaImpliedStdDev(option_type, strike_price, forward_price, option_price, discount_factor)
strikepayoff = ql.PlainVanillaPayoff(option_type, strike_price)
black = ql.BlackCalculator(strikepayoff, forward_price, implied_vol, discount_factor)
# t: float = (expiration_date - calc_date).days / 360
t: float = get_t_days(expiration_date, calc_date) / 360
implied_vol = implied_vol / np.sqrt(t)
opt_price: float = black.value()
return implied_vol, black.delta(discount_factor * forward_price), opt_price
def black_model_opt_price(
forward_price: float,
strike_price: float,
option_type: int,
implied_vol: float,
expiration_date: datetime,
calc_date: datetime,
discount_factor: float = 1
) -> float:
# t: float = (expiration_date - calc_date).days / 360
t: float = get_t_days(expiration_date, calc_date) / 360
iv: float = implied_vol * np.sqrt(t)
strikepayoff = ql.PlainVanillaPayoff(option_type, strike_price)
black = ql.BlackCalculator(strikepayoff, forward_price, iv, discount_factor)
opt_price: float = black.value()
return opt_price# region imports
from AlgorithmImports import *
import QuantLib as ql
from option_pricing.option_pricing_model import OptionPricingModel, IndexOptionPricingModel, FallbackIVStrategy
# endregion
class ImpliedVolatilityIndicator(PythonIndicator):
def __init__(
self,
algo: QCAlgorithm,
symbol: Symbol,
name: str,
moneyness: float,
target_expiry_days: int = 25,
) -> None:
super().__init__()
self._algo: QCAlgorithm = algo
self._symbol: Symbol = symbol
self._moneyness: float = moneyness
self._target_expiry_days: int = target_expiry_days
self.name: str = name
self.value: float = 0.
if self._symbol.security_type == SecurityType.INDEX:
self._option_pricing: Optional[OptionPricingModel] = IndexOptionPricingModel(algo)
else:
raise TypeError(f"ImpliedVolatilityIndicator.__init__: underlying security type: {self._underlying.security_type} is not supported")
def update(self, input: TradeBar) -> bool:
if not isinstance(input, TradeBar):
raise TypeError('ImpliedVolatilityIndicator.update: input must be a TradeBar')
self.value = 0.
found: bool = False
symbols_to_cleanup: List[Symbol] = []
sec_type: SecurityType = self._algo.securities[self._symbol].type
if sec_type == SecurityType.INDEX:
before_target_option_contract, after_target_option_contract = tuple(
self._find_atm_options_for_index(self._symbol)
)
before_target_future, after_target_future = (None, None)
elif sec_type == SecurityType.FUTURE:
# find two future contracts; one of them has sooner expiry than target expiry; other one expires later
before_target, after_target = self._find_futures_contracts()
if before_target is None and after_target is None:
self._algo.log(
f'ImpliedVolatilityIndicator.update: No futures contracts with targeted expiry found for: {self._symbol}'
)
before_target_option_contract, after_target_option_contract = (None, None)
else:
# subscribe futures
before_target_future, after_target_future = tuple(
self._subscribe_futures(before_target, after_target, symbols_to_cleanup)
)
# find ATM options for two futures
before_target_option_contract, after_target_option_contract = tuple(
self._find_atm_options_for_futures(
before_target_future,
after_target_future
)
)
else:
raise TypeError(f'ImpliedVolatilityIndicator.update: security type {sec_type} is not supported')
found: bool = before_target_option_contract is not None and after_target_option_contract is not None
if not found:
self._algo.log(
f'ImpliedVolatilityIndicator.update: No option contracts with targeted expiry found for: {self._symbol}'
)
else:
# underlying future has been subscribed
# NOTE DataNormalizationMode.RAW for underlying future is required
if all(
self._algo.subscription_manager.subscription_data_config_service.get_subscription_data_configs(c.underlying_symbol)
for c in [before_target_option_contract, after_target_option_contract]
):
# subscribe options
before_target_option, after_target_option = tuple(
self._subscribe_options(
before_target_option_contract,
after_target_option_contract,
symbols_to_cleanup
)
)
# calculate implied volatility
self.value = self._get_implied_volatility(
before_target_option,
after_target_option,
before_target_future,
after_target_future
)
# remove not needed assets from algorithm
for asset in symbols_to_cleanup:
self._algo.remove_security(asset)
return found
def _find_futures_contracts(self) -> Tuple[Optional[Symbol]]:
future_contracts: List[Symbol] = self._algo.future_chain_provider.get_future_contract_list(
self._symbol, self._algo.time
)
if len(future_contracts) == 0:
self._algo.log(
f'ImpliedVolatilityIndicator._find_futures_contract: No futures found for: {self._symbol}'
)
return None, None
target_expiry: datetime = self._algo.time + timedelta(days=self._target_expiry_days)
# select two futures - one of them has sooner expiry than target expiry; other one expires later
before_target: Symbol = max((f for f in future_contracts if f.ID.date <= target_expiry), key=lambda f: f.ID.date, default=None)
after_target: Symbol = min((f for f in future_contracts if f.ID.date >= target_expiry), key=lambda f: f.ID.date, default=None)
return before_target, after_target
# NOTE this could be the same for equity symbol as well
def _find_atm_options_for_index(
self,
index_symbol: Symbol
) -> Tuple[Optional[OptionContract]]:
option_contract_symbols: List = list(self._algo.option_chain(index_symbol, flatten=True).contracts.values())
if len(option_contract_symbols) == 0:
self._algo.log(
f'ImpliedVolatilityIndicator._find_atm_options_for_index: No options found for: {index_symbol}'
)
for i in range(2):
yield None
else:
underlying_price: float = self._algo.securities[index_symbol].ask_price
expiries: List[datetime] = list(map(lambda x: x.expiry, option_contract_symbols))
target_expiry: datetime = self._algo.time + timedelta(days=self._target_expiry_days)
before_target: Symbol = max((e for e in expiries if e <= target_expiry), key=lambda e: e, default=None)
after_target: Symbol = min((e for e in expiries if e >= target_expiry), key=lambda e: e, default=None)
for e in [before_target, after_target]:
contracts: List[OptionContract] = [
x for x in option_contract_symbols
if x.strike != 0
and x.expiry == e
and (1 - self._moneyness) <= abs(underlying_price / x.strike) <= (1 + self._moneyness)
]
if len(contracts) == 0:
self._algo.log(
f'ImpliedVolatilityIndicator._find_atm_options_for_index: No options filtered for: {index_symbol}'
)
yield None
else:
strike: float = max(map(lambda x: x.strike, contracts))
atm_option: OptionContract = next(
filter(
lambda x: x.right == OptionRight.CALL and x.expiry == e and x.strike == strike, contracts
)
)
yield atm_option
def _find_atm_options_for_futures(
self,
before_target_future: Future,
after_target_future: Future
) -> Tuple[Optional[OptionContract]]:
for target_future in [before_target_future, after_target_future]:
option_contract_symbols: List = list(self._algo.option_chain(target_future.symbol, flatten=True).contracts.values())
if len(option_contract_symbols) == 0:
self._algo.log(
f'ImpliedVolatilityIndicator._find_atm_options_for_futures: No options found for: {target_future.symbol}'
)
yield None
else:
underlying_price: float = target_future.ask_price
expiries: List[datetime] = list(map(lambda x: x.expiry, option_contract_symbols))
expiry: datetime = min(expiries)
contracts: List[OptionContract] = [
x for x in option_contract_symbols
if x.strike != 0
and x.expiry == expiry
and (1 - self._moneyness) <= abs(underlying_price / x.strike) <= (1 + self._moneyness)
]
if len(contracts) == 0:
self._algo.log(
f'ImpliedVolatilityIndicator._find_atm_options_for_futures: No options filtered for: {target_future.symbol}'
)
yield None
else:
strike: float = max(map(lambda x: x.strike, contracts))
atm_option: OptionContract = next(
filter(
lambda x: x.right == OptionRight.CALL and x.expiry == expiry and x.strike == strike, contracts
)
)
yield atm_option
def _get_implied_volatility(
self,
before_target_option: OptionContract,
after_target_option: OptionContract,
before_target_future: Optional[Future] = None,
after_target_future: Optional[Future] = None
) -> float:
# calculate implied volatility for each option using black model
if before_target_option.underlying.type == SecurityType.FUTURE:
implied_vols: List[float] = [
self._option_pricing.bs_iv(
option_symbol=o.symbol,
option_price=o.price,
forward_price=f.price,
evaluation_dt=self._algo.time,
fallback_iv_strategy=FallbackIVStrategy.CALL_PUT_PARITY_IV,
discount_factor=1.
)
for o, f in zip(
[before_target_option, after_target_option],
[before_target_future, after_target_future],
)
]
elif before_target_option.underlying.type == SecurityType.INDEX:
spot_price: float = self._algo.current_slice.bars.get(self._symbol).close
pricing_model: IndexOptionPricingModel = IndexOptionPricingModel(self._algo)
implied_vols: List[float] = []
for o in [before_target_option, after_target_option]:
rfr: float = pricing_model.get_risk_free_rate()
dividends: float = pricing_model.get_dividends(o.symbol)
discount_factor: float = pricing_model.get_discount_factor(rfr, dividends, o.expiry)
forward_price: float = pricing_model.get_forward_price(spot_price, discount_factor)
iv: float = self._option_pricing.bs_iv(
option_symbol=o.symbol,
option_price=o.price,
forward_price=forward_price,
evaluation_dt=self._algo.time,
fallback_iv_strategy=FallbackIVStrategy.CALL_PUT_PARITY_IV,
discount_factor=discount_factor
)
implied_vols.append(iv)
# interpolate variances
interpolated_variance: float = self._interpolate(
(before_target_option.expiry - self._algo.time).days,
(after_target_option.expiry - self._algo.time).days,
implied_vols[0] ** 2, # variance
implied_vols[1] ** 2, # variance
self._target_expiry_days
)
# get volatility
return np.sqrt(interpolated_variance)
def _subscribe_futures(
self,
before_target: Symbol,
after_target: Symbol,
symbols_to_cleanup: List[Symbol]
) -> Tuple[Future]:
# subscribe futures to QC algorithm
for s in [before_target, after_target]:
if not self._algo.securities.contains_key(s):
target_future: Future = self._algo.add_future_contract(s)
self._algo.securities[target_future.symbol].set_data_normalization_mode(DataNormalizationMode.RAW)
# add symbol for a later clean up; only those symbols which were not subscribed already (those might be traded)
symbols_to_cleanup.append(target_future.symbol)
else:
target_future: Future = self._algo.securities.get(s)
yield target_future
def _subscribe_options(
self,
before_target_option_contract: OptionContract,
after_target_option_contract: OptionContract,
symbols_to_cleanup: List[Symbol]
) -> Tuple:
# subscribe futures to QC algorithm
for oc in [before_target_option_contract, after_target_option_contract]:
if not self._algo.securities.contains_key(oc):
target_option = self._algo.add_future_option_contract(oc)
# add symbol for a later clean up; only those symbols which were not subscribed already (those might be traded)
symbols_to_cleanup.append(target_option.symbol)
else:
target_option = self._algo.securities.get(oc)
yield target_option
def _interpolate(self, x1: float, x2: float, y1: float, y2: float, x: float) -> float:
return ((y2 - y1) * x + x2 * y1 - x1 * y2) / (x2 - x1)# region imports
from AlgorithmImports import *
from collections import deque
from typing import Optional
# endregion
class RealizedVolatilityIndicator(PythonIndicator):
def __init__(
self,
algo: QCAlgorithm,
symbol: Symbol,
name: str,
daily_period: int,
auto_updates: bool,
resolution: Resolution = Resolution.DAILY
) -> None:
super().__init__()
self._daily_period: int = daily_period
self._auto_updates: bool = auto_updates
self.name: str = name
self.value: float = 0.
self._first_bar: Optional[TradeBar] = None
self._recent_bar: Optional[TradeBar] = None
self._return_values = deque()
self._rolling_sum: float = 0.
self._rolling_sum_of_squares: float = 0.
self._n: Optional[float] = None
# register indicator for automatic updates
if auto_updates:
algo.register_indicator(symbol, self, resolution)
@property
def is_auto_updated(self) -> bool:
return self._auto_updates
def update(self, input: TradeBar) -> bool:
if not isinstance(input, TradeBar):
raise TypeError('RealizedVolatilityIndicator.update: input must be a TradeBar')
if not self._first_bar:
# store first bar
self._first_bar = input
else:
log_return: float = np.log(input.close / self._recent_bar.close)
self._return_values.append(log_return)
# update rolling sums
self._rolling_sum += log_return
self._rolling_sum_of_squares += log_return ** 2
is_ready: bool = (input.end_time - self._first_bar.time).days >= self._daily_period
if is_ready:
# store number of bars
if not self._n:
self._n = len(self._return_values)
mean_value_1: float = self._rolling_sum / self._n
mean_value_2: float = self._rolling_sum_of_squares / self._n
# adjust rolling sums
removed_return: float = self._return_values.popleft()
self._rolling_sum -= removed_return
self._rolling_sum_of_squares -= removed_return ** 2
self.value = np.sqrt(mean_value_2 - (mean_value_1 ** 2)) * np.sqrt(250) # annualized
self._recent_bar = input
return self._n is not None# region imports
from AlgorithmImports import *
from dataclasses import dataclass
from abc import ABC, abstractmethod
from option_pricing.option_pricing_model import OptionPricingModel, IndexOptionPricingModel, FallbackIVStrategy
# endregion
@dataclass
class StressOptionInfo():
underlying_option_symbol: Symbol
spot_price: float
forward_spot_price: float
discount_factor: float
dividends: float
implied_volatility: float
option_price: float
class StressUnderlying(ABC):
def __init__(
self,
algo: QCAlgorithm,
underlying: Symbol,
underlying_price_change_perc: Optional[float],
underlying_vol_change_perc: Optional[float],
vix: Optional[Symbol],
vix_target: Optional[float]
) -> None:
self._algo: QCAlgorithm = algo
self._underlying: Symbol = underlying
self._underlying_price_change_perc: float = underlying_price_change_perc
self._underlying_vol_change_perc: float = underlying_vol_change_perc
self._vix: Optional[Symbol] = vix
self._vix_target: Optional[float] = vix_target
if self._underlying.security_type == SecurityType.INDEX:
self._option_pricing: Optional[OptionPricingModel] = IndexOptionPricingModel(algo)
else:
raise TypeError(f"StressUnderlying.__init__: underlying security type: {self._underlying.security_type} is not supported")
@property
def option_pricing(self) -> OptionPricingModel:
return self._option_pricing
@property
def original_option_holdings(self) -> float:
original_option_holdings: float = sum([
holding.value.holdings_value for holding in self._algo.portfolio
if self._algo.securities[holding.key].type == SecurityType.INDEX_OPTION and \
holding.value.invested
])
return original_option_holdings
@property
def portfolio_value(self) -> float:
# final portfolio value after stress
portfolio_value: float = self._algo.portfolio.total_portfolio_value - self.original_option_holdings + self.option_holdings
return portfolio_value
@property
def option_holdings(self) -> float:
# calculate holdings value for all the stressed options in portfolio
stressed_opt_holdings: float = 0.
stressed_port_options_info: List[StressOptionInfo] = self._stress()
for opt_info in stressed_port_options_info:
option_symbol: Symbol = opt_info.underlying_option_symbol
c_multiplier: int = self._algo.securities[option_symbol].contract_multiplier
quantity: float = self._algo.portfolio[option_symbol].quantity
option_price: float = opt_info.option_price # stressed option price
stressed_holding_value: float = quantity * option_price * c_multiplier
stressed_opt_holdings += stressed_holding_value
return stressed_opt_holdings
@abstractmethod
def _find_port_options(self) -> List[Symbol]:
...
@abstractmethod
def _stress(self) -> List[StressOptionInfo]:
...
class StressIndexUnderlyingFixedPercentage(StressUnderlying):
def __init__(
self,
algo: QCAlgorithm,
underlying: Symbol,
underlying_price_change_perc: Optional[float],
underlying_vol_change_perc: Optional[float],
) -> None:
super(StressIndexUnderlyingFixedPercentage, self).__init__(
algo,
underlying,
underlying_price_change_perc,
underlying_vol_change_perc,
None,
None
)
def _find_port_options(self) -> List[StressOptionInfo]:
result: List[StressOptionInfo] = []
# spot_price: float = self._algo.securities[self._underlying].price
spot_price: float = self._algo.current_slice.bars.get(self._underlying).close
# go through the options in a portfolio for specified underlying, those are INDEX_OPTIONs in this case
for holding in self._algo.portfolio:
if self._algo.securities[holding.key].type == SecurityType.INDEX_OPTION and \
holding.key.underlying == self._underlying and \
holding.value.invested:
rfr: float = self._option_pricing.get_risk_free_rate()
dividend_yld: float = self.option_pricing.get_dividends(holding.key)
discount_factor: float = self.option_pricing.get_discount_factor(rfr, dividend_yld, holding.key.id.date)
forward_price: float = self.option_pricing.get_forward_price(spot_price, discount_factor)
if holding.key.id.date <= self._algo.time:
# self._algo.log(f'StressIndexUnderlying._find_port_options - option {holding.key.value} has already expired; now: {self._algo.time}; expiry: {holding.key.id.date}; time delta: {holding.key.id.date - self._algo.time} days')
continue
iv: float = self._option_pricing.bs_iv(
option_symbol=holding.key,
option_price=self._algo.securities[holding.key].price,
forward_price=forward_price,
evaluation_dt=self._algo.time,
fallback_iv_strategy=FallbackIVStrategy.CALL_PUT_PARITY_IV,
discount_factor=discount_factor
)
opt_price: float = self._option_pricing.black_price(
option_symbol=holding.key,
forward_price=forward_price,
vol=iv,
discount_factor=discount_factor
)
result.append(
StressOptionInfo(
holding.key,
spot_price,
forward_price,
discount_factor,
dividend_yld,
iv,
opt_price
)
)
return result
def _stress(self) -> List[StressOptionInfo]:
port_options_info: List[StressOptionInfo] = self._find_port_options()
if len(port_options_info) == 0:
self._algo.log(f'StressIndexUnderlying._stress No options found in portfolio for {self._underlying} underlying')
return []
# go through the options in portfolio, change underlying price and recalculate options' values
# for a given percentage change in the volatility and spot price of the underlying
stressed_port_options_info: List[StressOptionInfo] = []
for option_info in port_options_info:
option_symbol: Symbol = option_info.underlying_option_symbol
stressed_spot_price: float = option_info.spot_price * (1 + self._underlying_price_change_perc)
stressed_iv: float = option_info.implied_volatility * (1 + self._underlying_vol_change_perc)
# discount_factor and dividends are not changed in this instance
stressed_forward_price: float = self.option_pricing.get_forward_price(
stressed_spot_price,
option_info.discount_factor
)
stressed_opt_price: float = self._option_pricing.black_price(
option_symbol=option_symbol,
forward_price=stressed_forward_price,
vol=stressed_iv,
discount_factor=option_info.discount_factor
)
stressed_port_options_info.append(
StressOptionInfo(
option_symbol,
stressed_spot_price,
stressed_forward_price,
# discount_factor and dividends are not changed in this instance
option_info.discount_factor,
option_info.dividends,
stressed_iv,
stressed_opt_price
)
)
return stressed_port_options_info
class StressIndexUnderlyingUsingVIX(StressUnderlying):
def __init__(
self,
algo: QCAlgorithm,
underlying: Symbol,
underlying_price_change_perc: Optional[float],
vix: Optional[Symbol],
vix_target: Optional[float]
) -> None:
super(StressIndexUnderlyingUsingVIX, self).__init__(
algo,
underlying,
underlying_price_change_perc,
None,
vix,
vix_target
)
def _find_port_options(self) -> List[StressOptionInfo]:
result: List[StressOptionInfo] = []
# spot_price: float = self._algo.securities[self._underlying].price
spot_price: float = self._algo.current_slice.bars.get(self._underlying).close
# go through the options in a portfolio for specified underlying, those are INDEX_OPTIONs in this case
for holding in self._algo.portfolio:
if self._algo.securities[holding.key].type == SecurityType.INDEX_OPTION and \
holding.key.underlying == self._underlying and \
holding.value.invested:
rfr: float = self._option_pricing.get_risk_free_rate()
dividend_yld: float = self.option_pricing.get_dividends(holding.key)
discount_factor: float = self.option_pricing.get_discount_factor(rfr, dividend_yld, holding.key.id.date)
forward_price: float = self.option_pricing.get_forward_price(spot_price, discount_factor)
if holding.key.id.date <= self._algo.time:
# self._algo.log(f'StressIndexUnderlying._find_port_options - option {holding.key.value} has already expired; now: {self._algo.time}; expiry: {holding.key.id.date}; time delta: {holding.key.id.date - self._algo.time} days')
continue
iv: float = self._option_pricing.bs_iv(
option_symbol=holding.key,
option_price=self._algo.securities[holding.key].price,
forward_price=forward_price,
evaluation_dt=self._algo.time,
fallback_iv_strategy=FallbackIVStrategy.CALL_PUT_PARITY_IV,
discount_factor=discount_factor
)
opt_price: float = self._option_pricing.black_price(
option_symbol=holding.key,
forward_price=forward_price,
vol=iv,
discount_factor=discount_factor
)
result.append(
StressOptionInfo(
holding.key,
spot_price,
forward_price,
discount_factor,
dividend_yld,
iv,
opt_price
)
)
return result
def _stress(self) -> List[StressOptionInfo]:
port_options_info: List[StressOptionInfo] = self._find_port_options()
if len(port_options_info) == 0:
self._algo.log(f'StressIndexUnderlying._stress No options found in portfolio for {self._underlying} underlying')
return []
# go through the options in portfolio, change underlying price and recalculate options' values
# for a targeted VIX value when self._vix is defined
stressed_port_options_info: List[StressOptionInfo] = []
for option_info in port_options_info:
option_symbol: Symbol = option_info.underlying_option_symbol
# target VIX value
vix_diff: float = (self._algo.current_slice.bars.get(self._vix).close - self._vix_target) / 100
stressed_spot_price: float = option_info.spot_price * (1 + self._underlying_price_change_perc)
stressed_iv: float = option_info.implied_volatility * (1 + vix_diff)
# discount_factor and dividends are not changed in this instance
stressed_forward_price: float = self.option_pricing.get_forward_price(
stressed_spot_price,
option_info.discount_factor
)
stressed_opt_price: float = self._option_pricing.black_price(
option_symbol=option_symbol,
forward_price=stressed_forward_price,
vol=stressed_iv,
discount_factor=option_info.discount_factor
)
stressed_port_options_info.append(
StressOptionInfo(
option_symbol,
stressed_spot_price,
stressed_forward_price,
# discount_factor and dividends are not changed in this instance
option_info.discount_factor,
option_info.dividends,
stressed_iv,
stressed_opt_price
)
)
return stressed_port_options_info# region imports
from AlgorithmImports import *
from options import *
from stats import *
import config
from indicators.realized_volatility import RealizedVolatilityIndicator
from indicators.implied_volatility import ImpliedVolatilityIndicator
from indicators.stress_test import StressUnderlying, StressIndexUnderlyingFixedPercentage, StressIndexUnderlyingUsingVIX
from option_pricing.iv_monitor import IVMonitor
# endregion
class SPX_Options(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2012, 1, 1)
self.SetEndDate(2025, 1, 1)
self.SetCash(10e11)
self.SetSecurityInitializer(MySecurityInitializer(self.brokerage_model, FuncSecuritySeeder(self.get_last_known_price)))
self.strategies = []
self.days_since_rebalance = 0
spx = self.AddIndex("SPX").Symbol
self.spy = self.add_equity("SPY").Symbol
spy = self.spy
# self.strategies.append(config.default_daily_dynamic_hedge_strategy(algo=self))
# self.strategies.append(config.default_daily_dynamic_hedge_strategy3(algo=self)) # vix function
self.strategies.append(config.default_daily_dynamic_hedge_strategy4(algo=self)) # vix regression
# self.strategies.append(config.default_collar_strategy(algo=self))
# self.strategies.append(config.sell_call_by_delta_strategy(algo=self))
# self.strategies.append(config.default_ETF_Dynamic_Collar(algo=self))
self.realized_option_profits = 0
self._strategy_parser(self.strategies)
self._vix: Symbol = self.add_index("VIX", Resolution.MINUTE).symbol
# stress indicator initialized with vix and underlying percentage change
self._stress_indicator: StressUnderlying = StressIndexUnderlyingUsingVIX(
algo=self,
underlying=spx,
underlying_price_change_perc=-0.1,
vix=self._vix,
vix_target=30
)
# stress indicator initialized with underlying price and volatility percentage change
# self._stress_indicator: StressUnderlying = StressIndexUnderlyingFixedPercentage(
# algo=self,
# underlying=spx,
# underlying_price_change_perc=-0.1,
# underlying_vol_change_perc=-0.1
# )
self._iv_monitor: IVMonitor = IVMonitor(
algo=self,
underlying=spx,
include_dividends=True,
abs_delta_to_notify=0.1
)
# realized volatility indicator
vol_period: int = 21
self._realized_vol_indicator: RealizedVolatilityIndicator = RealizedVolatilityIndicator(
algo=self,
symbol=spx,
name='RealizedVolatility',
daily_period=vol_period,
auto_updates=True,
resolution=Resolution.DAILY # only relevant with auto_updates = True
)
# implied volatility indicator
self._implied_vol_indicator: ImpliedVolatilityIndicator = ImpliedVolatilityIndicator(
algo=self,
symbol=spx,
name='ImpliedVolatility',
moneyness=0.01,
target_expiry_days=60,
)
self.register_indicator(spx, self._implied_vol_indicator, Resolution.DAILY)
# charting
PortfolioValue(self)
OptionsPortfolioValue(self)
Greeks(algo=self, strategies=self.strategies)
HoldingsValue(self, spy)
StressTestPortfolioValue(self, self._stress_indicator)
StressTestHoldings(self, self._stress_indicator)
ImpliedVsRealized(self, self._realized_vol_indicator, self._implied_vol_indicator)
'''
portfolio_contents_logger = LogPortfolioContents(self)
self.Schedule.On(
#self.DateRules.EveryDay('SPX'),
#self.TimeRules.BeforeMarketClose('SPX', 0),
self.DateRules.WeekStart('SPX'),
self.TimeRules.BeforeMarketClose('SPX', 0),
portfolio_contents_logger.log_portfolio_contents
)
'''
'''
self.Schedule.On(
self.DateRules.EveryDay('SPY'),
self.TimeRules.BeforeMarketClose('SPY', 30),
self._additional_spy
)
'''
'''
self.Schedule.On(
self.DateRules.EveryDay('SPY'),
self.TimeRules.BeforeMarketClose('SPY', 30),
self.adjust_spy
)
self.Schedule.On(
self.DateRules.EveryDay(),
self.TimeRules.Midnight,
self.increment_rebalance_day
)
'''
'''
fwrds = forward_prices_and_discounts_structure(self, spx)
self.Schedule.On(
self.DateRules.EveryDay('SPX'),
self.TimeRules.BeforeMarketClose('SPX', 30),
fwrds.get_options
)
'''
def on_order_event(self, order_event: OrderEvent) -> None:
# add new option symbols to IV monitor structure after order is filled
order = self.transactions.get_order_by_id(order_event.order_id)
if order_event.status == OrderStatus.FILLED:
if order_event.symbol.security_type == SecurityType.INDEX_OPTION:
self._iv_monitor.add(order_event.symbol)
# TODO how do we target certain strategy (if there are going to be multiple of them)?
for strategy in self.strategies:
instance: BuyStrategy = strategy['strategy_instance']
if getattr(instance, "update_vix_and_moneyness", None):
if (strategy['Type'] == 'Sell' and order_event.quantity < 0) or \
strategy['Type'] == 'Buy' and order_event.quantity > 0:
instance.update_vix_and_moneyness(order_event.symbol)
def increment_rebalance_day(self):
# update indicator manually
# if not self._realized_vol_indicator.is_auto_updated:
# self._realized_vol_indicator.update(
# self.current_slice.bars.get(self._future.symbol) # TODO this should be an index, this is copied from another codebase
# )
self.days_since_rebalance += 1
def _additional_spy(self):
if self.realized_option_profits > 0:
self.Log('Option profits ' + str(self.realized_option_profits))
price = self.Securities[self.spy].Price
amount = np.floor(self.realized_option_profits / price)
self.MarketOrder(self.spy, amount)
self.realized_option_profits = 0
def adjust_spy(self):
if self.days_since_rebalance >= 30:
if self.portfolio.cash != 0:
amount = round(self.portfolio.cash / self.securities[self.spy].price)
self.market_order(self.spy, amount)
self.days_since_rebalance = 0
def on_data(self, slice: Slice) -> None:
pass
def _strategy_parser(self, strategies):
for strategy in self.strategies:
self.Securities[strategy['underlying']].SetDataNormalizationMode(DataNormalizationMode.Raw)
option_strategy = strategy['buy_class'](
algo=self, **strategy
)
option_strategy.schedule()
strategy['strategy_instance'] = option_strategy
if strategy['Monetization'] is not None:
strategy['Monetization'].add_strategy(option_strategy)
self.Schedule.On(
self.DateRules.EveryDay(option_strategy.underlying),
self.TimeRules.BeforeMarketClose(option_strategy.underlying,
strategy['Execution Time']), strategy['Monetization'].monetize
)
'''
def OnOrderEvent(self, orderEvent: OrderEvent) -> None:
order = self.Transactions.GetOrderById(orderEvent.OrderId)
ticket = orderEvent.Ticket
if orderEvent.Status == OrderStatus.Filled:
if ticket.OrderType == OrderType.OptionExercise:
self.Log('Option Exercised')
if ticket.OrderType == OrderType.OptionExercise and orderEvent.FillPrice > 0:
self.realized_option_profits += orderEvent.FillPrice * orderEvent.AbsoluteFillQuantity
self.Log('Profit')
elif (ticket.SecurityType == SecurityType.Option or ticket.SecurityType == SecurityType.IndexOption) and orderEvent.Direction == OrderDirection.Sell:
if orderEvent.FillPrice > 0:
self.Log('Profit')
self.realized_option_profits += orderEvent.FillPrice * orderEvent.AbsoluteFillQuantity
'''
def on_securities_changed(self, changes):
pass
# self.Log('changed')
class MySecurityInitializer(BrokerageModelSecurityInitializer):
def __init__(self, brokerage_model: IBrokerageModel, security_seeder: ISecuritySeeder) -> None:
super().__init__(brokerage_model, security_seeder)
def Initialize(self, security: Security) -> None:
# First, call the superclass definition
# This method sets the reality models of each security using the default reality models of the brokerage model
super().Initialize(security)
# Next, overwrite the security buying power
security.set_buying_power_model(BuyingPowerModel.Null)
security.set_fee_model(ConstantFeeModel(0))#region imports
from AlgorithmImports import *
#endregion
import numpy as np
import pandas as pd
class Metric(ABC):
@abstractmethod
def calc(data: Union[pd.Series, pd.DataFrame]):
pass
class MeanReturn(Metric):
def __init__(N: int):
self._N = N
def calc(data: Union[pd.Series, pd.DataFrame]):
return data.mean(axis=0) * self._N
class CAGR(Metric):
pass
class StdDev(Metric):
def __init__(N: int):
self._N = N
def calc(data: Union[pd.Series, pd.DataFrame]):
return data.std(axis=0) * np.sqrt(self._N)
class MaxDrawdowns(Metric):
pass
class SharpeRatio(Metric):
def calc(data: Union[pd.Series, pd.DataFrame]):
sr = (weekly_ret_series.mean(axis=0) / weekly_ret_series.std(axis=0)) * np.sqrt(N)
# region imports
from AlgorithmImports import *
# endregion
class ParityPair:
def __init__(self):
self.put = None
self.call = None
self.strike = None
self.expiry = None
def update_parity_pair(self, contract_symbol):
if self.expiry == contract_symbol.id.date and \
self.strike == contract_symbol.id.strike_price:
if contract_symbol.id.OptionRight == OptionRight.PUT:
assert self.call is not None
self.put = contract_symbol
else:
assert contract_symbol.id.OptionRight == OptionRight.PUT and self.put is not None
self.call = contract_symbol
else:
self.call = None
self.put = None
self.expiry = contract_symbol.id.date
self.strike = contract_symbol.id.strike_price
if contract_symbol.id.OptionRight == OptionRight.PUT:
self.put = contract_symbol
else:
assert contract_symbol.id.OptionRight == OptionRight.CALL
self.call = contract_symbol
class forward_prices_and_discounts_structure:
def __init__(self, algo, underlying_symbol):
self._algo = algo
self._underlying_symbol = underlying_symbol
self._dates_dict = {}
def insert_dates(self, dates: List[date]):
for date in dates:
if date not in self._dates_dict:
self._dates_dict[date] = None
def _build_dates_dict_from_portfolio(self):
d = {}
for x in self._algo.portfolio:
symbol = x.key
if symbol.id.SecurityType == SecurityType.INDEX_OPTION and symbol.id.date not in d:
d[symbol.id.date] = None
return d
def _get_parity_pairs(self, contracts):
first_parity_pair = ParityPair()
second_parity_pair = ParityPair()
underlying_price = self._algo.securities[self._underlying_symbol].price
# if the number is an integer or its fractional part is 0.5, there can be two different strikes equally close to it - from below
# and from above, which might lead to unpredicted behaviour. We break parity by subtracting a small quantity.
if underlying_price.is_integer() or math.modf(underlying_price)[0] == 0.5:
underlying_price -= 0.0001
sorted_contracts = sorted(contracts, key=lambda x: abs(x.id.strike_price - underlying_price))
for i in (0,2):
if sorted_contracts[i].id.OptionRight == sorted_contracts[i+1].id.OptionRight:
self._algo.log('a')
assert (
sorted_contracts[i].id.strike_price == sorted_contracts[i+1].id.strike_price and \
sorted_contracts[i].id.OptionRight != sorted_contracts[i+1].id.OptionRight
)
assert sorted_contracts[0].id.strike_price != sorted_contracts[2].id.strike_price
for parity_pair, i in zip((first_parity_pair, second_parity_pair), (0,2)):
parity_pair.update_parity_pair(sorted_contracts[i])
parity_pair.update_parity_pair(sorted_contracts[i+1])
return first_parity_pair, second_parity_pair
def get_df_and_fp(contracts):
pass
def get_options(self):
self._dates_dict = self._build_dates_dict_from_portfolio()
option_contract_symbols = list(self._algo.option_chain(self._underlying_symbol))
dates_contracts_dict = {}
for symbol in option_contract_symbols:
if symbol.id.date in self._dates_dict:
if symbol.id.date in dates_contracts_dict:
dates_contracts_dict[symbol.id.date].append(symbol)
else:
dates_contracts_dict[symbol.id.date] = [symbol]
parity_pairs_dict = {}
for date in dates_contracts_dict:
parity_pairs_dict[date] = self._get_parity_pairs(dates_contracts_dict[date])# region imports
from AlgorithmImports import *
from .option_pricing_model import OptionPricingModel, IndexOptionPricingModel, FallbackIVStrategy
from decimal import Decimal
from pandas.core.frame import DataFrame
# endregion
class IVMonitor():
def __init__(
self,
algo: QCAlgorithm,
underlying: Symbol,
include_dividends: bool = True,
abs_delta_to_notify: float = 0.
) -> None:
self._algo: QCAlgorithm = algo
self._underlying: Symbol = underlying
self._include_dividends: bool = include_dividends
self._abs_delta_to_notify: float = abs_delta_to_notify
self._scheduler()
if self._underlying.security_type == SecurityType.INDEX:
self._option_pricing: Optional[OptionPricingModel] = IndexOptionPricingModel(algo)
else:
raise TypeError(f"IVMonitor.__init__: underlying security type: {self._underlying.security_type} is not supported")
# QC's implied volatility indicator indexed by option symbol
self._iv_by_option: Dict[Symbol, ImpliedVolatility] = {}
def add(self, option_symbol: Symbol) -> None:
if option_symbol not in self._iv_by_option:
iv_indicator: ImpliedVolatility = ImpliedVolatility(
option=option_symbol,
risk_free_rate_model=self._algo.risk_free_interest_rate_model,
dividend_yield_model=DividendYieldProvider.create_for_option(option_symbol) if self._include_dividends else ConstantDividendYieldModel(0),
mirror_option=None,
option_model=OptionPricingModelType.BLACK_SCHOLES
)
# iv_indicator: ImpliedVolatility = self._algo.iv(
# symbol=option_symbol,
# dividend_yield=None if self._include_dividends else 0.,
# option_model=OptionPricingModelType.BLACK_SCHOLES
# )
self._algo.warm_up_indicator(option_symbol, iv_indicator)
self._iv_by_option[option_symbol] = iv_indicator
def _scheduler(self):
self._algo.schedule.on(
self._algo.date_rules.every_day(self._underlying),
self._algo.time_rules.before_market_close(self._underlying, 0),
self._compare_iv
)
def _compare_iv(self) -> None:
# NOTE this can be deleted later - used only for assertion
def trim_to_a_point(num: float, dec_point: int = 4) -> float:
factor: int = 10**dec_point
num = num * factor
num = int(num)
num = num / factor
return num
# NOTE this can be deleted later - used only for assertion
def get_precision(num: float) -> int:
d_num: Decimal = Decimal(str(num))
sign, digits, exp = d_num.as_tuple()
precision: int = abs(exp)
return precision
option_symbols_to_remove: List[Symbol] = []
current_slice: Slice = self._algo.current_slice
underlying_bar: TradeBar = current_slice.bars.get(self._underlying)
# iterate through the option symbols and measure the difference in our black model IV value and QC's indicator IV value
for option_symbol, iv_indicator in self._iv_by_option.items():
# update indicator
opt_bar: QuoteBar = current_slice.quote_bars.get(option_symbol)
if underlying_bar and opt_bar:
iv_indicator.update(IndicatorDataPoint(self._underlying, underlying_bar.end_time, underlying_bar.close))
iv_indicator.update(IndicatorDataPoint(option_symbol, opt_bar.end_time, opt_bar.close))
dte: int = (option_symbol.id.date - self._algo.time).days
if dte <= 0:
option_symbols_to_remove.append(option_symbol)
# self._algo.log(f'IVMonitor._compare_iv - option {option_symbol.value} has already expired; now: {self._algo.time}; expiry: {option_symbol.id.date}; time delta: {option_symbol.id.date - self._algo.time} days')
continue
if not self._algo.portfolio[option_symbol].invested:
continue
if iv_indicator.is_ready:
spot_price: float = underlying_bar.close
if self._include_dividends:
dividends: float = self._option_pricing.get_dividends(option_symbol)
else:
dividends: float = 0.
rfr: float = self._option_pricing.get_risk_free_rate()
discount_factor: float = self._option_pricing.get_discount_factor(rfr, dividends, option_symbol.id.date)
forward_price: float = self._option_pricing.get_forward_price(spot_price, discount_factor)
iv: float = self._option_pricing.bs_iv(
option_symbol=option_symbol,
option_price=opt_bar.close,
forward_price=forward_price,
evaluation_dt=self._algo.time,
fallback_iv_strategy=FallbackIVStrategy.CALL_PUT_PARITY_IV,
discount_factor=discount_factor
)
opt_price: float = self._option_pricing.black_price(
option_symbol=option_symbol,
forward_price=forward_price,
vol=iv,
discount_factor=discount_factor
)
# lagging 1 day
precalculated_iv: float = self._algo.current_slice.option_chains.get(option_symbol.canonical).contracts.get(option_symbol).implied_volatility
qc_indicator_iv: float = iv_indicator.current.value
black_iv: float = iv
# NOTE these can be removed later - assertion only
qc_opt_price: float = opt_bar.close
indicator_opt_price: float = iv_indicator.price.current.value
black_opt_price: float = round(opt_price, get_precision(qc_opt_price)) # round to get rid of black model floating point precision to a point when price is comparable to QC's price
indicator_spot_price: float = iv_indicator.underlying_price.current.value
assert trim_to_a_point(iv_indicator.dividend_yield.current.value) == trim_to_a_point(dividends), f'indicator dividends: {iv_indicator.dividend_yield.current.value}, black dividends: {dividends}'
assert trim_to_a_point(iv_indicator.risk_free_rate.current.value) == trim_to_a_point(rfr), f'indicator RFR: {iv_indicator.risk_free_rate.current.value}, black RFR: {rfr}'
assert indicator_opt_price == qc_opt_price, f"indicator option price does not equal to last available QC option price; QC option price {qc_opt_price}; indicator option price: {indicator_opt_price}"
# assert indicator_opt_price == black_opt_price, f"indicator option price does not equal to black model option price; black option price {black_opt_price}; indicator option price: {indicator_opt_price}"
assert indicator_spot_price == spot_price, f"indicator spot price does not equal to QC spot price; QC spot price {spot_price}; indicator spot price: {indicator_spot_price}"
diff: float = black_iv / qc_indicator_iv - 1
if abs(diff) >= self._abs_delta_to_notify:
self._algo.log(
f'{option_symbol.value} - Black model IV: ' + '{:.2f}'.format(black_iv) + '; QC IV indicator: ' + '{:.2f}'.format(qc_indicator_iv) + '; diff: ' + '{:.2f}'.format(diff*100) + '%; DTE: ' + f'{dte}'
)
# remove expired options
for symbol in option_symbols_to_remove:
del self._iv_by_option[symbol]# region imports
from AlgorithmImports import *
import QuantLib as ql
from abc import ABC, abstractmethod
from dataclasses import dataclass
from math import e
from enum import Enum
# endregion
class FallbackIVStrategy(Enum):
MIRROR_OPTION_IV = 1
CALL_PUT_PARITY_IV = 2
class OptionPricingModel(ABC):
def __init__(self, algo: QCAlgorithm) -> None:
self._algo: QCAlgorithm = algo
@abstractmethod
def get_forward_price(
self,
spot_price: float,
discount_factor: float,
) -> float:
...
@abstractmethod
def get_risk_free_rate(self, calc_dt: Optional[datetime] = None) -> float:
...
@abstractmethod
def get_discount_factor(
self,
rfr: float,
dividends: float,
expiry: datetime,
calc_dt: Optional[datetime] = None
) -> float:
...
@abstractmethod
def get_dividends(self, option_symbol: Symbol, calc_dt: Optional[datetime] = None) -> float:
...
def _get_history_price(self, option_symbol: Symbol) -> Optional[float]:
option_price: Optional[float] = None
history: DataFrame = self._algo.history(QuoteBar, option_symbol, 10, Resolution.HOUR)
if not history.empty:
last_bar = history.iloc[-1]
option_price: float = (last_bar.bidclose + last_bar.askclose) / 2
if option_price is None:
self._algo.log(f'OptionPricingModel._get_history_price - cannot fetch option {option_symbol} price')
return option_price
@staticmethod
def get_t_days(expiration_date: datetime, calc_dt: datetime) -> float:
dt = (expiration_date - calc_dt)
days_dt: float = dt.days
seconds_dt: float = dt.seconds / (24*60*60) # days
t: float = seconds_dt + days_dt
return t
class IndexOptionPricingModel(OptionPricingModel):
def __init__(self, algo: QCAlgorithm) -> None:
super(IndexOptionPricingModel, self).__init__(algo)
def get_forward_price(
self,
spot_price: float,
discount_factor: float
) -> float:
F: float = spot_price / discount_factor
return F
def get_risk_free_rate(self, calc_dt: Optional[datetime] = None) -> float:
if calc_dt is None: calc_dt = self._algo.time
rfr: float = RiskFreeInterestRateModelExtensions.get_risk_free_rate(
self._algo.risk_free_interest_rate_model,
calc_dt, calc_dt
)
return rfr
def get_discount_factor(
self,
rfr: float,
dividends: float,
expiry: datetime,
calc_dt: Optional[datetime] = None
) -> float:
if calc_dt is None: calc_dt = self._algo.time
discount_factor: float = e**((-rfr+dividends)*((expiry - calc_dt).days / 360))
return discount_factor
def get_dividends(self, option_symbol: Symbol, calc_dt: Optional[datetime] = None) -> float:
if calc_dt is None: calc_dt = self._algo.time
# https://www.quantconnect.com/docs/v2/writing-algorithms/reality-modeling/dividend-yield/key-concepts
# DividendYieldProvider, which provides the continuous yield calculated from all dividend payoffs from the underlying Equity over the previous 350 days.
# SPX => SPY dividends
# NDX => QQQ dividends
# VIX => 0
return DividendYieldProvider.create_for_option(option_symbol).get_dividend_yield(calc_dt, self._algo.securities[option_symbol.underlying].price)
def bs_iv(
self,
option_symbol: Symbol, # TODO make this a class property
option_price: float,
forward_price: float,
evaluation_dt: datetime,
fallback_iv_strategy: FallbackIVStrategy,
discount_factor: float = 1
) -> float:
iv: float = self._black_iv(option_symbol, option_price, forward_price, fallback_iv_strategy, discount_factor)
t: float = OptionPricingModel.get_t_days(option_symbol.id.date, evaluation_dt) / 360
# iv *= np.sqrt(t) # iv in annualized terms
iv /= np.sqrt(t) # to match previous results, IV monitor and indicators
return iv
def _black_iv(
self,
option_symbol: Symbol, # TODO make this a class property
option_price: float,
forward_price: float,
fallback_iv_strategy: FallbackIVStrategy,
discount_factor: float = 1,
) -> float:
strike_price: float = option_symbol.id.strike_price
option_type: int = ql.Option.Call if option_symbol.id.option_right == OptionRight.CALL else ql.Option.Put
try:
implied_vol: float = ql.blackFormulaImpliedStdDev(
option_type, strike_price, forward_price, option_price, discount_factor
)
except RuntimeError:
# create mirror option in case there was an exception caught in the first black model solver
mirror_option_symbol: Symbol = Symbol.create_option(
option_symbol.underlying,
option_symbol.id.market,
self._algo.securities[option_symbol].style,
OptionRight.CALL if option_symbol.id.option_right == OptionRight.PUT else OptionRight.PUT,
strike_price,
option_symbol.id.date
)
# if option_type == ql.Option.Call:
# assert (forward_price - strike_price) * discount_factor - option_price > 0
# implied_vol = 0
# else:
# assert (strike_price - forward_price) * discount_factor - option_price < 0
# implied_vol = 0
mirror_option_price: Optional[float] = self._get_history_price(mirror_option_symbol)
if fallback_iv_strategy == FallbackIVStrategy.MIRROR_OPTION_IV:
implied_vol: float = self._mirror_option_iv(
mirror_option_symbol=mirror_option_symbol,
mirror_option_price=mirror_option_price,
forward_price=forward_price,
discount_factor=discount_factor
)
elif fallback_iv_strategy == FallbackIVStrategy.CALL_PUT_PARITY_IV:
implied_vol: float = self._call_put_parity_iv(
option_symbol=option_symbol,
option_price=option_price,
mirror_option_symbol=mirror_option_symbol,
mirror_option_price=mirror_option_price,
forward_price=forward_price,
discount_factor=discount_factor
)
return implied_vol
def _mirror_option_iv(
self,
mirror_option_symbol: Symbol,
mirror_option_price: float,
forward_price: float,
discount_factor: float = 1
) -> float:
strike_price: float = mirror_option_symbol.id.strike_price
option_type: int = ql.Option.Call if mirror_option_symbol.id.option_right == OptionRight.CALL else ql.Option.Put
implied_vol: float = ql.blackFormulaImpliedStdDev(
option_type, strike_price, forward_price, mirror_option_price, discount_factor
)
return implied_vol
def _call_put_parity_iv(
self,
option_symbol: Symbol,
option_price: float,
mirror_option_symbol: Symbol,
mirror_option_price: float,
forward_price: float,
discount_factor: float = 1
) -> float:
strike_price: float = option_symbol.id.strike_price
spot_price: float = self._algo.securities[option_symbol.underlying].price
call_price: Optional[float] = option_price if option_symbol.id.option_right == OptionRight.CALL else mirror_option_price
put_price: Optional[float] = option_price if option_symbol.id.option_right == OptionRight.PUT else mirror_option_price
if call_price is None or put_price is None:
return None
fixed_discount_factor: float = (put_price + spot_price - call_price) / strike_price
fixed_forward_price: float = spot_price / fixed_discount_factor
option_type: int = ql.Option.Call if option_symbol.id.option_right == OptionRight.CALL else ql.Option.Put
implied_vol: float = ql.blackFormulaImpliedStdDev(
option_type, strike_price, fixed_forward_price, option_price, fixed_discount_factor
)
return implied_vol
def black_price(
self,
option_symbol: Symbol,
forward_price:float,
vol: float,
discount_factor=1
) -> float:
"""Compute Black-Scholes option price for given volatility."""
option_type: int = ql.Option.Call if option_symbol.id.option_right == OptionRight.CALL else ql.Option.Put
price: float = ql.blackFormula(option_type, option_symbol.id.strike_price, forward_price, vol, discount_factor)
return price#region imports
from AlgorithmImports import *
from abc import ABC, abstractmethod
import bisect
from pandas.core.frame import DataFrame
from sklearn.linear_model import LinearRegression
#endregion
class Filter(ABC):
@abstractmethod
def filter(self):
pass
class AmountCalculator(ABC):
def __init__(self, algo):
self._algo = algo
@abstractmethod
def calc_amount(self, option):
pass
def _get_amount_as_fraction_of_portfolio(self, option, fraction_of_portfolio_value):
multiplier = option.ContractMultiplier
target_notional = self._algo.Portfolio.TotalPortfolioValue * fraction_of_portfolio_value
notional_of_contract = multiplier * option.Underlying.Price
amount = target_notional / notional_of_contract
return amount
def _get_amount_as_fraction_of_cash(self, option, fraction):
holding_value = self._algo.portfolio.cash
multiplier = option.ContractMultiplier
target_notional = holding_value * fraction
notional_of_contract = multiplier * option.Underlying.Price
amount = target_notional / notional_of_contract
return amount
class BuyStrategy(ABC):
# ToDo: get rid of general params and make specific parameters/
def __init__(
self,
algo: QCAlgorithm,
underlying: Symbol,
canonical_option_symbols: List[Symbol],
amount_calculator: AmountCalculator,
filter: Optional[Filter],
date_schedule_rule: Callable,
time_schedule_rule: Callable,
**params,
):
'''
date_scheduler_rule: a pointer to a function in QCAlgorithm.DateRules
time_schedlue_rule: a pointer to a function in QCAlgorithm.TimeRules
'''
self._params = params
self._resolution = params['Resolution']
self._algo = algo
self._underlying = underlying
self._filter = filter
self._amount_calculator = amount_calculator
self._active_options: Set[Symbol] = set()
self._canonical_options = canonical_option_symbols
self._current_orders = {}
self._time_rule = time_schedule_rule
self._date_rule = date_schedule_rule
self._garbage_collection_set = set() #Securities that we subscribed to but do not need. To be collected by garbage collection.
self._to_execute_dict = {}
self._options_to_buy = None
def schedule(self):
algo = self._algo
algo.Schedule.On(self._date_rule(self._underlying), self._time_rule(self._underlying, self._params['Options Filter Time']), self._scheduled_get_options)
algo.Schedule.On(self._date_rule(self._underlying), self._time_rule(self._underlying, self._params['Execution Time']), self.buy)
#algo.Schedule.On(algo.DateRules.EveryDay(), algo.TimeRules.Midnight, self._garbage_collection)
# determine regression coefs scheduler
if 'coef_date_schedule_rule' in self._params and \
'coef_time_schedule_rule' in self._params:
algo.schedule.on(
self._params['coef_date_schedule_rule'](self._underlying),
self._params['coef_time_schedule_rule'](
self._underlying, self._params['Execution Time']
),
self._determine_coefs
)
def _determine_coefs(self) -> None:
pass
def execute(self):
pass
@property
def underlying(self):
return self._underlying
@property
def active_options(self):
return self._active_options
def _get_options(self):
pass
def _get_amount(self, option):
amount = self._amount_calculator.calc_amount(option)
if self._params['Type'] == 'Sell':
amount = -amount
elif self._params['Type'] != 'Buy':
raise NotImplementedError()
return amount
def _scheduled_get_options(self):
self._options_to_buy = self._get_options()
symbols = self._options_to_buy
if symbols is None:
return
for symbol in symbols:
option = self._subscribe_to_option(symbol)
if not self._algo.securities[symbol].is_tradable:
self._algo.securities[symbol].is_tradable = True
def buy(self):
if self._filter is not None:
if not self._filter.filter():
return
#symbols = self._get_options()
symbols = self._options_to_buy
if symbols is None:
#self._algo.Log('stop')
return
assert len(symbols) == 1
for symbol in symbols:
option = self._subscribe_to_option(symbol)
amount = self._get_amount(option)
if self._algo.securities[symbol].Price == 0:
self._algo.log(f'No price data {self._algo.Time} :: {symbol.Value}')
#return
ticket = self._algo.market_order(symbol=symbol, quantity=amount)
#self._current_orders[ticket.OrderId] = {
# 'ticket': ticket,
#}
#self._algo.Log(f'Time to expiry {(option.expiry - self._algo.time).days}')
self._active_options.add(symbol)
def _get_options_chains(self):
symbols = []
for canonical_option in self._canonical_options:
#symbols += list(self._algo.OptionChainProvider.GetOptionContractList(canonical_option, self._algo.Time))
symbols += list(option.key for option in self._algo.option_chain(canonical_option).contracts)
return symbols
def get_Greeks(self)->Dict[Symbol, Dict[str, float]]:
'''
Obtains Greeks and Implied vol for options traded under current strategy
and still held in portfolio.
ToDo: identify amounts that were bought only using current strategy.
ToDo: right now if another strategy bought same options, there will be
ToDo: double counting.
'''
slice = self._algo.CurrentSlice
contracts: Dict[Symbol, OptionContract] = {}
for canonical_option in self._canonical_options:
if canonical_option in slice.OptionChains:
contracts.update({contract.Symbol: contract for contract in slice.OptionChains[canonical_option]})
res = {}
for option in self._active_options:
if option in self._algo.Portfolio and self._algo.Portfolio[option].Quantity != 0:
if option not in contracts:
self._algo.Log(f'{self._algo.Time} has problem with Greeks')
return
res.update(
{
option: {
'ImpliedVol': contracts[option].ImpliedVolatility,
'Greeks': contracts[option].Greeks
}
}
)
return res
def _garbage_collection(self):
to_remove = []
for symbol in self._active_options:
if symbol not in self._algo.Portfolio or self._algo.Portfolio[symbol].Quantity == 0:
self._algo.RemoveOptionContract(symbol)
to_remove.append(symbol)
for symbol in to_remove:
self._active_options.remove(symbol)
to_remove = []
for symbol in self._garbage_collection_set:
if symbol not in self._algo.Portfolio or self._algo.Portfolio[symbol].Quantity == 0:
self._algo.RemoveOptionContract(symbol)
for symbol in to_remove:
self._garbage_collection.remove(symbol)
def _filter_puts(self, symbols):
puts = [symbol for symbol in symbols if symbol.ID.OptionRight == OptionRight.Put]
return puts
def _filter_calls(self, symbols):
calls = [symbol for symbol in symbols if symbol.ID.OptionRight == OptionRight.Call]
return calls
def _get_n_day_options(self, symbols, days_to_expiry: int):
expiry_symbol_dict = {}
algo_time = datetime.combine(self._algo.time, time.min)
for symbol in symbols:
expiry_date = symbol.ID.Date
if expiry_date in expiry_symbol_dict:
expiry_symbol_dict[expiry_date].append(symbol)
else:
if (expiry_date - algo_time).days >= days_to_expiry:
expiry_symbol_dict[expiry_date] = [symbol]
dates = sorted(expiry_symbol_dict.keys())
for date in dates:
if len(expiry_symbol_dict[date]) > 1:
return expiry_symbol_dict[date]
return None
def _get_options_by_moneyness(self, symbols, moneyness: float):
expiry_symbol_dict = {}
algo_time = datetime.combine(self._algo.time, time.min)
for symbol in symbols:
expiry_date = symbol.ID.Date
if expiry_date in expiry_symbol_dict:
expiry_symbol_dict[expiry_date].append(symbol)
else:
if (expiry_date - algo_time).days >= days_to_expiry:
expiry_symbol_dict[expiry_date] = [symbol]
dates = sorted(expiry_symbol_dict.keys())
for date in dates:
if len(expiry_symbol_dict[date]) > 1:
return expiry_symbol_dict[date]
return None
def _get_options_closest_to_target_expiry(self, symbols, days_to_expiry: int, before=False, after=False):
expiry_symbol_dict = {}
for symbol in symbols:
expiry_date = symbol.ID.Date
if expiry_date in expiry_symbol_dict:
expiry_symbol_dict[expiry_date].append(symbol)
else:
if (expiry_date - self._algo.Time).days >= 1:
if before:
if (symbol.ID.Date - self._algo.Time).days <= days_to_expiry:
expiry_symbol_dict[expiry_date] = [symbol]
elif after:
if (symbol.ID.Date - self._algo.Time).days >= days_to_expiry:
expiry_symbol_dict[expiry_date] = [symbol]
else:
expiry_symbol_dict[expiry_date] = [symbol]
dates = sorted(expiry_symbol_dict.keys(), key=lambda x: abs((x - self._algo.Time).days - days_to_expiry))
for date in dates:
if len(expiry_symbol_dict[date]) > 1:
return expiry_symbol_dict[date]
return None
def _get_puts(self):
symbols = self._get_options_chains()
return self._filter_puts(symbols)
def _get_calls(self):
symbols = self._get_options_chains()
return self._filter_calls(symbols)
def _get_options_closest_to_target_strike(self, symbols, target_strike: float) -> List[Symbol]:
underlying_price: float = self._algo.securities[symbols[0].id.underlying.symbol].price
target_strike_price: float = target_strike * underlying_price
# get negative and positive differences
diff: List[float] = [x.id.strike_price - target_strike * self._algo.securities[x.id.underlying.symbol].price for x in symbols]
ind_bellow: np.ndarray = [diff.index(d) for i, d in enumerate(diff) if d < 0.]
ind_above: np.ndarray = [diff.index(d) for i, d in enumerate(diff) if d >= 0.]
# pick one closest bellow and above target strike price
closest_symbols: List[Symbol] = [
symbols[max(ind_bellow, key=lambda i: diff[i])],
symbols[min(ind_above, key=lambda i: diff[i])]
]
# assert one is bellow and other one is above the target strike price
assert closest_symbols[0].id.strike_price < target_strike_price and closest_symbols[1].id.strike_price > target_strike_price, \
f'1. strike: {closest_symbols[0].id.strike_price}; 2. strike: {closest_symbols[1].id.strike_price}: target strike: {target_strike_price}'
return closest_symbols
def _get_option_closest_to_target_strike(self, symbols, target_strike: float):
# ToDo: Closest from below or above: return here an array of closest and make the calling function pick.
diff = np.array([abs(x.ID.StrikePrice - target_strike * self._algo.Securities[x.ID.Underlying.Symbol].Price) for x in symbols])
ind = np.argmin(diff)
return symbols[ind]
def _subscribe_to_option(self, symbol):
if self._underlying.SecurityType == SecurityType.Index:
return self._algo.AddIndexOptionContract(symbol=symbol, resolution=self._resolution)
elif self._underlying.SecurityType == SecurityType.Equity:
return self._algo.AddOptionContract(symbol=symbol, resolution=self._resolution)
else:
raise NotImplementedError
class BuyByVIX(BuyStrategy):
def __init__(self,
algo: QCAlgorithm,
underlying: Symbol,
canonical_option_symbols: List[Symbol],
amount_calculator: AmountCalculator,
filter: Optional[Filter],
date_schedule_rule: Callable,
time_schedule_rule: Callable,
**params,
):
'''
date_schedule_rule: a pointer to a function in QCAlgorithm.DateRules
time_schedule_rule: a pointer to a function in QCAlgorithm.TimeRules
'''
super().__init__(
algo,
underlying,
canonical_option_symbols,
amount_calculator,
filter,
date_schedule_rule,
time_schedule_rule,
**params
)
self._vix_symbol: Symbol = algo.add_index("VIX", Resolution.MINUTE).symbol
self._moneyness_fn: Callable = params['moneyness_fn']
def _get_options(self):
if self._params['Right'] == OptionRight.Put:
symbols = self._get_puts()
elif self._params['Right'] == OptionRight.Call:
symbols = self._get_calls()
else:
raise NotImplementedError()
symbols = self._get_n_day_options(symbols, self._params['Days to Expiry'])
if symbols is None or len(symbols) == 0:
self._algo.Log('No options to buy')
return None
moneyness: float = self._moneyness_fn(self._algo.current_slice[self._vix_symbol].price) / 100.
target_strike: float = 1. + moneyness
closest_symbols: Symbol = self._get_options_closest_to_target_strike(symbols, target_strike=target_strike)
fn: Callable = min if target_strike <= 1 else max
symbol: Symbol = fn(closest_symbols, key=lambda x: x.id.strike_price)
return [symbol]
class BuyByVIXRegression(BuyStrategy):
def __init__(self,
algo: QCAlgorithm,
underlying: Symbol,
canonical_option_symbols: List[Symbol],
amount_calculator: AmountCalculator,
filter: Optional[Filter],
date_schedule_rule: Callable,
time_schedule_rule: Callable,
**params,
):
'''
date_schedule_rule: a pointer to a function in QCAlgorithm.DateRules
time_schedule_rule: a pointer to a function in QCAlgorithm.TimeRules
'''
super().__init__(
algo,
underlying,
canonical_option_symbols,
amount_calculator,
filter,
date_schedule_rule,
time_schedule_rule,
**params
)
self._vix_symbol: Symbol = algo.add_index("VIX", Resolution.MINUTE).symbol
# vix and moneyness data
self._vix_and_moneyness_df: DataFrame = pd.DataFrame(columns=['VIX', 'moneyness'])
# calculate values
self._intercept: Optional[float] = None
self._beta: Optional[float] = None
# TODO (1) what minimal size of a historical period do we need?
self._min_daily_period: int = self._params['coef_min_daily_period']
def _determine_coefs(self) -> None:
if self._vix_and_moneyness_df.empty:
self._algo.log(f'Not enought data to determine regression coeficients. DataFrame is empty')
return
elapsed_days: int = (self._algo.time - self._vix_and_moneyness_df.index[0]).days
if elapsed_days <= self._min_daily_period:
self._algo.log(f'Not enought data to determine regression coeficients. Elapsed days: {elapsed_days}. Min elapsed daily period: {self._min_daily_period}')
return
# run regression with X=VIX and Y=moneyness
lin_reg = LinearRegression()
lin_reg.fit(
self._vix_and_moneyness_df.VIX.values.reshape(-1,1),
self._vix_and_moneyness_df.moneyness.values.reshape(-1,1)
)
self._intercept = lin_reg.intercept_
self._beta = lin_reg.coef_
def update_vix_and_moneyness(self, option_symbol: Symbol) -> None:
# determine moneyness and store it with the VIX
strike_price: float = option_symbol.id.strike_price
underlying_price: float = self._algo.securities[option_symbol.underlying].price
vix: float = self._algo.current_slice[self._vix_symbol].price
moneyness: float = -(100 - (strike_price / underlying_price) * 100)
self._vix_and_moneyness_df.loc[self._algo.time] = [vix, moneyness]
def _get_options(self):
vix: float = self._algo.current_slice[self._vix_symbol].price
intercept: float = self._params['default_intercept'] if self._intercept is None else self._intercept
beta: float = self._params['default_beta'] if self._beta is None else self._beta
if self._params['Right'] == OptionRight.Put:
symbols = self._get_puts()
elif self._params['Right'] == OptionRight.Call:
symbols = self._get_calls()
else:
raise NotImplementedError()
symbols = self._get_n_day_options(symbols, self._params['Days to Expiry'])
if symbols is None or len(symbols) == 0:
self._algo.Log('No options to buy')
return None
moneyness: float = (intercept + beta * vix) / 100
target_strike: float = 1. + moneyness
closest_symbols: Symbol = self._get_options_closest_to_target_strike(symbols, target_strike=target_strike)
fn: Callable = min if target_strike <= 1 else max
symbol: Symbol = fn(closest_symbols, key=lambda x: x.id.strike_price)
return [symbol]
class BuyByDeltaAndExpiry(BuyStrategy):
def __init__(self,
algo: QCAlgorithm,
underlying: Symbol,
canonical_option_symbols: Symbol,
target_delta: float,
amount_calculator: AmountCalculator,
threshold = 0,
filter: Optional[Filter]=None,
**params,
):
super().__init__(algo=algo, underlying = underlying, canonical_option_symbols = canonical_option_symbols,
amount_calculator = amount_calculator, filter = filter, **params
)
self._threshold = threshold
self._target_delta = target_delta
def _get_option_by_delta(self, options: Dict[Symbol, Option]):
deltas_dict = {}
count = 0
for symbol, option in options.items():
#contract_data = OptionContract(option, self._underlying)
contract_data = OptionContract(option)
contract_data.Time = self._algo.Time
result = option.EvaluatePriceModel(None, contract_data)
delta = abs(result.Greeks.Delta)
if delta in deltas_dict:
count += 1
deltas_dict[delta].append(symbol)
elif delta >= self._threshold:
count += 1
deltas_dict[delta] = [symbol]
if len(deltas_dict) == 0:
return None
deltas = sorted(deltas_dict.keys(), key=lambda x: abs(x - self._target_delta))
#self._algo.Log(f'The delta difference that we get is {abs(deltas[0]-self._target_delta)}')
return deltas_dict[deltas[0]]
def _get_options(self):
if self._params['Right'] == OptionRight.Put:
symbols = self._get_puts()
elif self._params['Right'] == OptionRight.Call:
symbols = self._get_calls()
else:
raise NotImplementedError()
symbols = self._get_n_day_options(symbols, self._params['Days to Expiry'])
if symbols is None or len(symbols) == 0:
self._algo.Log('No options to buy')
return None
new_subscriptions = set()
options = {}
for symbol in symbols:
options[symbol] = self._subscribe_to_option(symbol)
if symbol not in self._algo.ActiveSecurities:
new_subscriptions.add(symbol)
symbols = self._get_option_by_delta(options)
if symbols is None:
self._algo.Log(f'No deltas')
return None
for symbol in symbols:
if symbol in new_subscriptions:
new_subscriptions.remove(symbol)
# Careful. There is an issue that removeSecurity removes the price.
#for subscription in new_subscriptions:
# self._algo.RemoveOptionContract(subscription)
self._garbage_collection_set.update(new_subscriptions)
if len(symbols) > 1:
symbols = [symbols[0]]
assert len(symbols) == 1
return symbols
class BuyByStrikeAndExpiry(BuyStrategy):
def _get_options(self):
if self._params['Right'] == OptionRight.Put:
symbols = self._get_puts()
elif self._params['Right'] == OptionRight.Call:
symbols = self._get_calls()
else:
raise NotImplementedError()
if self._params['Before Target Expiry']:
symbols = self._get_options_closest_to_target_expiry(symbols, days_to_expiry=self._params['Days to Expiry'], before=True, after=False)
else:
symbols = self._get_options_closest_to_target_expiry(symbols, days_to_expiry=self._params['Days to Expiry'], before=False, after=False)
if symbols is None or len(symbols) == 0:
self._algo.Log('No options to buy')
return
symbol = self._get_option_closest_to_target_strike(symbols, target_strike=self._params['Strike'])
return [symbol]
class BuySpread(BuyStrategy):
def __init__(self,
algo: QCAlgorithm,
underlying: Symbol,
canonical_option_symbols: List[Symbol],
amount_calculator: AmountCalculator,
filter: Optional[Filter],
date_schedule_rule: Callable,
time_schedule_rule: Callable,
**params,
):
'''
date_scheduler_rule: a pointer to a function in QCAlgorithm.DateRules
time_schedlue_rule: a pointer to a function in QCAlgorithm.TimeRules
'''
super().__init__(algo,
underlying,
canonical_option_symbols,
amount_calculator,
filter,
date_schedule_rule,
time_schedule_rule,
**params)
assert 'main_leg' in self._params
self._main_leg = self._params['main_leg']
assert 'secondary_leg' in self._params
self._secondary_leg = self._params['secondary_leg']
self._main_amount_calculator = self._main_leg['amount_calculator']
self._connections = {}
self._main_quantities = {}
@property
def connections(self):
return self._connections
def buy(self):
if self._filter is not None:
if not self._filter.filter():
return
symbols = self._options_to_buy
if symbols is None:
#self._algo.Log('stop')
return
assert len(symbols) == 2
options = [self._subscribe_to_option(symbol) for symbol in symbols]
amounts = self._get_amount(options)
for symbol, amount in zip(symbols, amounts):
if self._algo.securities[symbol].Price == 0:
self._algo.log(f'No price data {self._algo.Time} :: {symbol.Value}')
return
ticket = self._algo.market_order(symbol=symbol, quantity=amount)
self._active_options.add(symbols[0])
if symbols[0] in self._connections:
self._main_quantities[symbols[0]] += amounts[0]
if symbols[1] in self._connections[symbols[0]]:
self._connections[symbols[0]][symbols[1]] += amounts[1]
else:
self._connections[symbols[0]][symbols[1]] = amounts[1]
else:
self._connections[symbols[0]] = {symbols[1]: amounts[1]}
self._main_quantities[symbols[0]] = amounts[0]
def _get_amount(self, options):
amounts = []
for leg in [self._main_leg, self._secondary_leg]:
amount = leg['amount_calculator'].calc_amount(options[0])
if leg['Type'] == 'Sell':
amounts.append(-amount)
elif leg['Type'] == 'Buy':
amounts.append(amount)
else:
raise NotImplementedError()
return amounts
def _get_options(self):
res = []
for leg in [self._main_leg, self._secondary_leg]:
params = leg
if params['Right'] == OptionRight.Put:
symbols = self._get_puts()
elif params['Right'] == OptionRight.Call:
symbols = self._get_calls()
else:
raise NotImplementedError()
if leg['Before Target Expiry']:
symbols = self._get_options_closest_to_target_expiry(symbols, days_to_expiry=params['Days to Expiry'], before=True, after=False)
else:
symbols = self._get_options_closest_to_target_expiry(symbols, days_to_expiry=params['Days to Expiry'], before=False, after=False)
if symbols is None or len(symbols) == 0:
self._algo.Log('No options to buy')
return
symbol = self._get_option_closest_to_target_strike(symbols, target_strike=params['Strike'])
res.append(symbol)
return res
class BuyCollar(BuySpread):
def _search_strike_relative_to_budget(self, target_symbol, search_symbols):
target_option = self._subscribe_to_option(target_symbol)
def _get_options(self):
res = []
params = self._main_leg
if params['Right'] == OptionRight.Put:
symbols = self._get_puts()
secondary_symbols = self._get_calls()
elif params['Right'] == OptionRight.Call:
symbols = self._get_calls()
secondary_symbols = self._get_puts()
else:
raise NotImplementedError()
if params['Before Target Expiry']:
symbols = self._get_options_closest_to_target_expiry(symbols, days_to_expiry=params['Days to Expiry'], before=True, after=False)
secondary_symbols = self._get_options_closest_to_target_expiry(secondary_symbols, days_to_expiry=params['Days to Expiry'], before=True, after=False)
else:
symbols = self._get_options_closest_to_target_expiry(symbols, days_to_expiry=params['Days to Expiry'], before=False, after=False)
secondary_symbols = self._get_options_closest_to_target_expiry(secondary_symbols, days_to_expiry=params['Days to Expiry'], before=False, after=False)
if symbols is None or len(symbols) == 0:
self._algo.Log('No options to buy')
return
assert symbols[0].id.date == secondary_symbols[0].id.date
symbol = self._get_option_closest_to_target_strike(symbols, target_strike=params['Strike'])
self._search_strike_relative_to_budget(secondary_symbols, symbol)
res.append(symbol)
return res
class MonetizeOptions(ABC):
def __init__(self, algo, option_strategy: Optional[BuyStrategy]=None):
self._algo = algo
self._option_strategy: Optional[BuyStrategy] = option_strategy
@abstractmethod
def monetize(self):
pass
def add_strategy(self, option_strategy: BuyStrategy):
if self._option_strategy is not None:
raise(ValueError('connection to buys strategy already exists'))
self._option_strategy = option_strategy
class MonetizeFixedTimeBeforeExpiry(MonetizeOptions):
def __init__(self, algo, min_days_to_expiry):
self._min_days_to_expiry = min_days_to_expiry
super().__init__(algo)
def monetize(self):
for holding in self._algo.Portfolio:
if holding.Value.Type == SecurityType.IndexOption or holding.Value.Type == SecurityType.Option and holding.Value.HoldingsValue != 0:
time_to_expiry = (self._algo.Securities[holding.Key].Expiry - self._algo.Time).days
if time_to_expiry <= self._min_days_to_expiry:
self._algo.RemoveSecurity(holding.Key)
class MonetizeRelativeToUnderlying(MonetizeOptions):
def __init__(self, algo, percent_of_underlying):
self._percent_of_underlying = percent_of_underlying
super().__init__(algo)
def monetize(self):
for holding in self._algo.Portfolio:
if (holding.Value.Type == SecurityType.IndexOption or holding.Value.Type == SecurityType.Option) and holding.Value.HoldingsValue > 0:
underlying_price = self._algo.Securities[holding.Value.Security.Underlying.Symbol].Price
if self._algo.Securities[holding.Key].StrikePrice * self._percent_of_underlying > underlying_price:
self._algo.RemoveSecurity(holding.Key)
class MonetizeByDelta(MonetizeOptions):
def __init__(self, algo, delta):
self._delta = delta
super().__init__(algo)
def monetize(self):
algo = self._algo
slice = algo.CurrentSlice
assert len(slice.OptionChains) == 1 or len(slice.OptionChains) == 0
assert self._option_strategy is not None
for canonical_option in self._option_strategy._canonical_options:
if canonical_option not in slice.OptionChains:
continue
option_contracts = slice.OptionChains[canonical_option]
for contract in option_contracts:
if (
contract.Symbol in self._option_strategy.active_options
and contract.Symbol in algo.Portfolio
and algo.Portfolio[contract.Symbol].HoldingsValue != 0
):
if abs(contract.Greeks.Delta) > self._delta:
algo.RemoveSecurity(contract.Symbol)
class MonetizeSpreadByDelta(MonetizeByDelta):
def __init__(self, algo, delta, monetize_connections=True):
self._monetize_connections = monetize_connections
super().__init__(algo, delta)
def monetize(self):
algo = self._algo
slice = algo.current_slice
assert len(slice.option_chains) == 1 or len(slice.option_chains) == 0
assert self._option_strategy is not None
for canonical_option in self._option_strategy._canonical_options:
if canonical_option not in slice.option_chains:
continue
option_contracts = slice.option_chains[canonical_option]
options_dict = {contract.symbol: contract for contract in option_contracts}
symbols_to_pop = []
for symbol in self._option_strategy._connections:
if symbol in options_dict and abs(options_dict[symbol].Greeks.Delta) > self._delta:
quantity = -self._option_strategy._main_quantities[symbol]
quantity1 = 0
algo.market_order(symbol, quantity)
#assert(len(self._option_strategy.connections[symbol])==1)
if self._monetize_connections:
for connected_symbol, quantity_to_offset in self._option_strategy._connections[symbol].items():
quantity1 += quantity_to_offset
algo.market_order(connected_symbol, -quantity_to_offset)
assert quantity == quantity1
symbols_to_pop.append(symbol)
for symbol in symbols_to_pop:
self._option_strategy._connections.pop(symbol)
self._option_strategy._main_quantities.pop(symbol)
class ChainMonetizations(MonetizeOptions):
def __init__(self, algo, monetizations_list):
self._monetizations_list = monetizations_list
super().__init__(algo)
def monetize(self):
for monetization in self._monetizations_list:
monetization.monetize()
class VixThresholdFilter(Filter):
def __init__(self, algo, threshold):
self._algo = algo
self._threshold = threshold
self.vix_symbol = algo.AddIndex("VIX", Resolution.Hour).Symbol
algo.Securities[self.vix_symbol].SetDataNormalizationMode(DataNormalizationMode.Raw)
def filter(self):
if self._algo.Securities[self.vix_symbol].Price > self._threshold:
return False
else:
return True
class VixFilter(Filter):
def __init__(self, algo, threshold, days):
self._algo = algo
self._threshold = threshold
self.vix_symbol = algo.AddIndex("VIX", Resolution.Daily).Symbol
algo.Securities[self.vix_symbol].SetDataNormalizationMode(DataNormalizationMode.Raw)
self.vix_sma = SimpleMovingAverage(days)
self._algo.RegisterIndicator(self.vix_symbol, self.vix_sma, Resolution.Daily)
algo.WarmUpIndicator(self.vix_symbol, self.vix_sma)
def filter(self):
if not self.vix_sma.IsReady:
return False
if self._algo.Securities[self.vix_symbol].Price >= self.vix_sma.Current.Value * self._threshold:
return False
else:
return True
class FractionOfCashAmountCalculator(AmountCalculator):
def __init__(self, algo, notional_frac, multiplier, cap=None):
super().__init__(algo)
self._multiplier = multiplier
self._notional_frac = notional_frac
self._cap = cap
def calc_amount(self, option):
amount = self._get_amount_as_fraction_of_cash(option, self._notional_frac)
amount = round(amount * self._multiplier)
if self._cap is not None and (option.ask_price != 0 or option.bid_price != 0):
assert option.ask_price != 0
assert option.bid_price != 0
holding_value = self._algo.portfolio.cash
option_cost = ((option.ask_price + option.bid_price)/2) * option.ContractMultiplier
max_amount = round((self._cap * holding_value)/option_cost)
amount = min(amount, max_amount)
return amount
class ConstantFractionAmountCalculator(AmountCalculator):
def __init__(self, algo, notional_frac, multiplier, cap=None):
super().__init__(algo)
self._multiplier = multiplier
self._notional_frac = notional_frac
self._cap = cap
def calc_amount(self, option):
amount = self._get_amount_as_fraction_of_portfolio(option, self._notional_frac)
amount = round(amount * self._multiplier)
return amount
class FractionOfHoldingAmountCalculator(AmountCalculator):
def __init__(self, algo, notional_frac, multiplier, holding_symbol, cap=None):
super().__init__(algo)
self._holding = holding_symbol
self._multiplier = multiplier
self._notional_frac = notional_frac
self._cap = cap
def _get_amount_as_fraction_of_holding(self, option, fraction):
holding_value = self._algo.portfolio[self._holding].holdings_value
multiplier = option.ContractMultiplier
target_notional = holding_value * fraction
notional_of_contract = multiplier * option.Underlying.Price
amount = target_notional / notional_of_contract
return amount
def calc_amount(self, option):
amount = self._get_amount_as_fraction_of_holding(option, self._notional_frac)
amount = round(amount * self._multiplier)
if self._cap is not None and option.price != 0:
holding_value = self._algo.portfolio[self._holding].holdings_value
option_cost = option.price * option.ContractMultiplier
max_amount = round((self._cap * holding_value)/option_cost)
amount = min(amount, max_amount)
return amount
class FractionToExpiryAmountCalculator(AmountCalculator):
def __init__(self, algo, notional_frac, multiplier):
super().__init__(algo)
self._multiplier = multiplier
self._notional_frac = notional_frac
def calc_amount(self, option):
amount = self._get_amount_as_fraction_of_portfolio(option, self._notional_frac)
days_to_expiry = (option.Expiry - self._algo.Time).days
amount = round(amount * self._multiplier / days_to_expiry)
return amount
class DeltaHedge:
def __init__(self, algo: QCAlgorithm, underlying: Symbol, canonical_option: Symbol,
date_schedule_rule, time_schedule_rule, multiplier=1):
self._algo: QCAlgorithm = algo
self._underlying = underlying
self._canonical_option = canonical_option
self._time_rule = time_schedule_rule
self._date_rule = date_schedule_rule
self._algo.schedule.on(self._date_rule, self._time_rule, self.delta_hedge)
self._multiplier = multiplier
def delta_hedge(self):
if self._canonical_option not in self._algo.current_slice.option_chains:
self._algo.log('no chain data')
return
if self._algo.securities[self._underlying].exchange.exchange_open:
delta = 0
contracts = self._algo.current_slice.option_chains[self._canonical_option].contracts
for holding in self._algo.portfolio:
symbol = holding.key
if (symbol.ID.SecurityType == SecurityType.Option or symbol.ID.SecurityType == SecurityType.IndexOption) and \
self._algo.portfolio[symbol].quantity != 0: #and symbol.underlying == self._underlying:
if symbol in contracts:
delta += self._algo.portfolio[symbol].quantity * contracts[symbol].greeks.delta * self._algo.securities[symbol].contract_multiplier
else:
self._algo.Log(f'missing greek')
delta *= self._multiplier
#self._algo.Log(f'got delta to hedge {delta}')
quantity = -(delta)
amount_to_hedge = quantity - self._algo.portfolio[self._underlying].quantity
#self._algo.Log(f'additional hedging {amount_to_hedge}')
self._algo.market_order(symbol=self._underlying, quantity=round(amount_to_hedge))
class DeltaHedgeWithFutures(DeltaHedge):
def __init__(self, algo: QCAlgorithm, canonical_option: Symbol, continuous_future: Future,
date_schedule_rule, time_schedule_rule, multiplier=1, filter: Filter = None):
self._algo: QCAlgorithm = algo
self._canonical_option = canonical_option
self._continuous_future = continuous_future
self._current_fut = None
self._filter = filter
self._time_rule = time_schedule_rule
self._date_rule = date_schedule_rule
self._algo.schedule.on(self._date_rule, self._time_rule, self.delta_hedge)
self._multiplier = multiplier
@property
def _current_future(self):
if self._current_fut is None:
self._current_fut = self._continuous_future.mapped
return self._current_fut
def delta_hedge(self):
if self._current_future is None:
self._algo.log('no mapped future to hedge')
return
if self._canonical_option not in self._algo.current_slice.option_chains:
self._algo.log('no chain data')
return
if self._filter is not None and not self._filter.filter():
return
if self._algo.securities[self._current_future].exchange.exchange_open:
delta = 0
contracts = self._algo.current_slice.option_chains[self._canonical_option].contracts
for holding in self._algo.portfolio:
symbol = holding.key
if (symbol.ID.SecurityType == SecurityType.Option or symbol.ID.SecurityType == SecurityType.IndexOption) and \
self._algo.portfolio[symbol].quantity != 0: #and symbol.underlying == self._underlying:
if symbol in contracts:
delta += self._algo.portfolio[symbol].quantity * contracts[symbol].greeks.delta * self._algo.securities[symbol].contract_multiplier
else:
self._algo.Log(f'missing greek')
#self._algo.Log(f'got delta to hedge {delta}')
quantity_to_hedge = - delta // self._algo.securities[self._current_future].symbol_properties.contract_multiplier
if self._to_rollover():
self._algo.log('liquidating')
self._algo.liquidate(self._current_future)
self._current_fut = self._continuous_future.mapped
else:
current_quantity = self._algo.portfolio[self._current_future].quantity
quantity_to_hedge = quantity_to_hedge - current_quantity
#self._algo.Log(f'additional hedging {quantity_to_hedge}')
self._algo.market_order(symbol=self._current_fut, quantity=quantity_to_hedge)
def _to_rollover(self):
if (self._algo.securities[self._current_future].expiry -self._algo.time).days <= 10:
return True
else:
return False
class BuyCollarByBudgetAlignment(BuyCollar):
def _get_options(self):
res = []
params = self._main_leg
if params['Right'] == OptionRight.Put:
symbols = self._get_puts()
secondary_symbols = self._get_calls()
elif params['Right'] == OptionRight.Call:
symbols = self._get_calls()
secondary_symbols = self._get_puts()
else:
raise NotImplementedError()
secondary_params = self._secondary_leg
if params['Before Target Expiry']:
symbols = self._get_options_closest_to_target_expiry(symbols, days_to_expiry=params['Days to Expiry'], before=True, after=False)
secondary_symbols = self._get_options_closest_to_target_expiry(secondary_symbols, days_to_expiry=secondary_params['Days to Expiry'], before=True, after=False)
else:
symbols = self._get_options_closest_to_target_expiry(symbols, days_to_expiry=params['Days to Expiry'], before=False, after=False)
secondary_symbols = self._get_options_closest_to_target_expiry(secondary_symbols, days_to_expiry=secondary_params['Days to Expiry'], before=False, after=False)
if symbols is None or len(symbols) == 0:
self._algo.Log('No options to buy')
return
symbol = self._get_option_closest_to_target_strike(symbols, target_strike=params['Strike'])
res.append(symbol)
secondary_symbol = self._search_strike_relative_to_budget(symbol, secondary_symbols)
res.append(secondary_symbol)
return res
def _bisect_contract_price_search(self, target_price, search_symbols, start, end, notional_target, quantity_cap):
def key_func(symbol):
data = self._algo.history(symbol, 10, resolution=Resolution.HOUR, fill_forward=True)
if not hasattr(data, 'askclose') or not hasattr(data, 'bidclose'):
self._algo.log('no data')
price = (data.askclose.values[-1] + data.bidclose.values[-1]) / 2
res = (target_price - price * quantity_cap) / self._algo.portfolio.cash
s = symbol.id.strike_price
u = self._algo.securities[self._underlying].price
return res
index = bisect.bisect_left(search_symbols, notional_target, key=key_func)
lower_price = key_func(search_symbols[index - 1])
if index == len(search_symbols):
index = index -1
else:
upper_price = key_func(search_symbols[index])
assert lower_price <= notional_target <= upper_price
return search_symbols[index]
def _search_strike_relative_to_budget(self, target_symbol, search_symbols):
option = self._subscribe_to_option(target_symbol)
price = (option.ask_price + option.bid_price) / 2
main_quantity = self._main_leg['amount_calculator'].calc_amount(option)
#secondary_quantity = self._secondary_leg['amount_calculator'].calc_amount(option)
secondary_quantity = self._params['secondary_budget_calculator'].calc_amount(option)
cost = main_quantity * option.contract_multiplier * price
underlying_price = self._algo.securities[self._underlying].price
#time_factor = np.sqrt(self._secondary_leg['Days to Expiry'] / 360)
time_factor = 1
lb = 1-0.3 * time_factor
ub = 1+0.3 * time_factor
if len(search_symbols) == 0:
self._algo_log ('problem')
search_temp = search_symbols
search_symbols = [s for s in search_symbols if s.id.strike_price > lb * underlying_price and s.id.strike_price < ub * underlying_price]
if len(search_symbols) == 0:
self._algo_log ('problem')
#notional_frac = self._params['secondary_notional_frac']
#amt_calc = FractionOfCashAmountCalculator(self._algo, notional_frac = notional_frac, multiplier=1)
notional_target = self._params['cost_target_pct']
self._temp_amount = secondary_quantity
secondary_quantity *= option.contract_multiplier
if search_symbols[0].id.option_right == OptionRight.CALL:
ordered_symbols = sorted(search_symbols, key=lambda x: x.id.strike_price, reverse=False)
#if dealing with put options
else:
ordered_symbols = sorted(search_symbols, key=lambda x: x.id.strike_price, reverse=True)
strikes = [s.id.strike_price for s in ordered_symbols]
#df = self._algo.history(target_symbol, periods=10, resolution=Resolution.MINUTE, fill_forward=True)
secondary_option = self._bisect_contract_price_search(cost, ordered_symbols, start=0, end=len(search_symbols),
notional_target=notional_target, quantity_cap=secondary_quantity)
return secondary_option
def _get_amount(self, options):
amounts = []
for i, leg in enumerate([self._main_leg, self._secondary_leg]):
if self._params['equalize_legs'] and i >=1:
amount = amounts[0]
else:
amount = leg['amount_calculator'].calc_amount(options[i])
if self._params['cap_secondary_leg'] and i >= 1:
amount = min(self._params['cap_secondary_leg_frac'] * abs(amounts[0]), amount)
if leg['Type'] == 'Sell':
amounts.append(-amount)
elif leg['Type'] == 'Buy':
amounts.append(amount)
else:
raise NotImplementedError()
#assert amounts[1] == self._temp_amount
return amounts#region imports
from AlgorithmImports import *
from abc import ABC, abstractmethod
from indicators.stress_test import StressUnderlying
from indicators.realized_volatility import RealizedVolatilityIndicator
from indicators.implied_volatility import ImpliedVolatilityIndicator
#endregion
class Charts(ABC):
def __init__(self, algo, chart_name, series_list):
self._algo = algo
self._chart_name = chart_name
self._series_list = series_list
self._add_chart()
self._scheduler()
def _add_chart(self):
chart = Chart(self._chart_name)
self._algo.AddChart(chart)
for series in self._series_list:
chart.AddSeries(Series(series['name'], series['type'], series['unit']))
def _scheduler(self):
algo = self._algo
algo.Schedule.On(
algo.DateRules.EveryDay('SPX'),
algo.TimeRules.BeforeMarketClose('SPX', 0),
self._update_chart
)
@abstractmethod
def _update_chart(self):
pass
class ImpliedVsRealized(Charts):
def __init__(
self,
algo,
rv_indicator: RealizedVolatilityIndicator,
iv_indicator: ImpliedVolatilityIndicator
):
series_list = [
{
'name': 'Realized Volatility',
'type': SeriesType.Line,
'unit': '%'
},
{
'name': 'Implied Volatility',
'type': SeriesType.Line,
'unit': '%'
},
]
self._rv_indicator: RealizedVolatilityIndicator = rv_indicator
self._iv_indicator: ImpliedVolatilityIndicator = iv_indicator
super().__init__(algo = algo, chart_name = 'Implied vs Realized Volatility', series_list = series_list)
def _update_chart(self):
self._algo.Plot(self._chart_name, self._series_list[0]['name'], self._rv_indicator.value * 100)
self._algo.Plot(self._chart_name, self._series_list[1]['name'], self._iv_indicator.value * 100)
class StressTestHoldings(Charts):
def __init__(self, algo, stress_indicator: StressUnderlying):
series_list = [
{
'name': 'Stressed Option Holdings',
'type': SeriesType.Line,
'unit': '$'
},
{
'name': 'Original Option Holdings',
'type': SeriesType.Line,
'unit': '$'
},
]
self._stress_indicator: StressUnderlying = stress_indicator
super().__init__(algo = algo, chart_name = 'Stress Test Holdings', series_list = series_list)
def _update_chart(self):
self._algo.Plot(self._chart_name, self._series_list[0]['name'], self._stress_indicator.option_holdings)
self._algo.Plot(self._chart_name, self._series_list[1]['name'], self._stress_indicator.original_option_holdings)
class StressTestPortfolioValue(Charts):
def __init__(self, algo, stress_indicator: StressUnderlying):
series_list = [
{
'name': 'Stressed Portfolio Value',
'type': SeriesType.Line,
'unit': '$'
},
{
'name': 'Original Portfolio Value',
'type': SeriesType.Line,
'unit': '$'
}
]
self._stress_indicator: StressUnderlying = stress_indicator
super().__init__(algo = algo, chart_name = 'Stress Test', series_list = series_list)
def _update_chart(self):
self._algo.Plot(self._chart_name, self._series_list[0]['name'], self._stress_indicator.portfolio_value)
self._algo.Plot(self._chart_name, self._series_list[1]['name'], self._algo.portfolio.total_portfolio_value)
class InvestedOptionCount(Charts):
def __init__(self, algo):
series_list = [
{
'name': 'Invested Option Count',
'type': SeriesType.BAR,
'unit': ''
},
]
super().__init__(algo = algo, chart_name = 'Invested Option Count', series_list = series_list)
def _update_chart(self):
options_invested: int = [
holding.value for holding in self._algo.portfolio
if self._algo.securities[holding.key].type == SecurityType.INDEX_OPTION and \
holding.value.invested
]
self._algo.Plot(self._chart_name, self._series_list[0]['name'], len(options_invested))
class PortfolioValue(Charts):
def __init__(self, algo):
series_list = [
{
'name': 'Portfolio Value',
'type': SeriesType.Line,
'unit': '$'
}
]
super().__init__(algo = algo, chart_name = 'Portfolio', series_list = series_list)
def _update_chart(self):
self._algo.Plot(self._chart_name, self._series_list[0]['name'], self._algo.Portfolio.TotalPortfolioValue)
class HoldingsValue(Charts):
def __init__(self, algo, holdings_symbol: Symbol):
series_list = [
{
'name': holdings_symbol.to_string(),
'type': SeriesType.Line,
'unit': '$'
}
]
self._holdings_symbol = holdings_symbol
super().__init__(algo = algo, chart_name = holdings_symbol.to_string(), series_list = series_list)
def _update_chart(self):
self._algo.Plot(
self._chart_name, self._series_list[0]['name'],
self._algo.Portfolio[self._holdings_symbol].holdings_value
)
class OptionsPortfolioValue(Charts):
def __init__(self, algo):
series_list = [
{
'name': 'Options Value',
'type': SeriesType.Line,
'unit': '$'
}
]
super().__init__(algo = algo, chart_name = 'Options Portfolio Value', series_list = series_list)
def _update_chart(self):
value = 0
for holding in self._algo.Portfolio:
if holding.Value.Type == SecurityType.IndexOption or holding.Value.Type == SecurityType.Option:
value += holding.Value.HoldingsValue
self._algo.Plot(self._chart_name, self._series_list[0]['name'], value)
class Greeks(Charts):
def __init__(self, algo, strategies):
self._strategies = strategies
self._algo = algo
series_list = [
{
'name': 'Delta',
'type': SeriesType.Line,
'unit': ''
},
{
'name': 'Gamma',
'type': SeriesType.Line,
'unit': ''
},
{
'name': 'Implied Volatility',
'type': SeriesType.Line,
'unit': ''
},
{
'name': 'Theta',
'type': SeriesType.Line,
'unit': ''
},
{
'name': 'Vega',
'type': SeriesType.Line,
'unit': ''
},
{
'name': 'Theta_per_day',
'type': SeriesType.Line,
'unit': ''
},
]
super().__init__(algo = algo, chart_name = 'Greeks', series_list = series_list)
def _update_chart(self):
algo = self._algo
delta = 0
gamma = 0
implied_vol = 0
theta = 0
vega = 0
theta_per_day = 0
total_options_holdings = 0
for strategy in self._strategies:
greeks = strategy['strategy_instance'].get_Greeks()
if greeks is None:
algo.Log('No Greeks')
return
strategy['CurrentGreeks'] = greeks
for option in strategy['CurrentGreeks']:
delta += strategy['CurrentGreeks'][option]['Greeks'].Delta * algo.Securities[option].ContractMultiplier * algo.Portfolio[option].Quantity
gamma += strategy['CurrentGreeks'][option]['Greeks'].Gamma * algo.Securities[option].ContractMultiplier * algo.Portfolio[option].Quantity
vega += strategy['CurrentGreeks'][option]['Greeks'].Vega * algo.Securities[option].ContractMultiplier * algo.Portfolio[option].Quantity
theta += strategy['CurrentGreeks'][option]['Greeks'].Theta * algo.Securities[option].ContractMultiplier * algo.Portfolio[option].Quantity
theta_per_day += strategy['CurrentGreeks'][option]['Greeks'].ThetaPerDay * algo.Securities[option].ContractMultiplier * algo.Portfolio[option].Quantity
algo.Plot(self._chart_name, "Delta", delta)
algo.Plot(self._chart_name, "Gamma", gamma)
algo.Plot(self._chart_name, "Vega", vega)
algo.Plot(self._chart_name, "Theta", theta)
algo.Plot(self._chart_name, "Theta_per_day", theta_per_day)
class LogPortfolioContents:
def __init__(self, algo):
self._algo = algo
def log_portfolio_contents(self):
algo = self._algo
algo.log(f'Options Portfolio Contents: \n')
for holding in algo.portfolio:
holding = holding.value
if holding.quantity == 0:
continue
if holding.Type == SecurityType.IndexOption or holding.Type == SecurityType.Option:
dict_to_log = {
'symbol': holding.symbol.value,
'expiry': holding.security.expiry,
'strike': holding.security.strike_price,
'weight' : holding.holdings_value / algo.portfolio.total_portfolio_value,
'notional_pct': abs(holding.security.underlying.price * holding.security.contract_multiplier * holding.quantity / algo.portfolio.total_portfolio_value)
}
algo.log(dict_to_log)
algo.log(f'End Portfolio Contents')