| Overall Statistics |
|
Total Orders 1258 Average Win 0.12% Average Loss -0.03% Compounding Annual Return -1.003% Drawdown 12.400% Expectancy -0.662 Start Equity 1000000000000 End Equity 877312151015 Net Profit -12.269% Sharpe Ratio -2.425 Sortino Ratio -3.15 Probabilistic Sharpe Ratio 0.000% Loss Rate 93% Win Rate 7% Profit-Loss Ratio 3.51 Alpha -0.021 Beta -0.048 Annual Standard Deviation 0.01 Annual Variance 0 Information Ratio -0.814 Tracking Error 0.144 Treynor Ratio 0.525 Total Fees $0.00 Estimated Strategy Capacity $0 Lowest Capacity Asset SPX 32OCJVZOPY5PQ|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_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': 'Sell', '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
# 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 # endregion # TODO use black model from option pricing class class ImpliedVolatilityIndicator(PythonIndicator): def __init__( self, algo: QCAlgorithm, continuous_future: Future, name: str, moneyness: float, target_expiry_days: int = 25, ) -> None: super().__init__() self._algo: QCAlgorithm = algo self._continuous_future: Future = continuous_future self._moneyness: float = moneyness self._target_expiry_days: int = target_expiry_days self.name: str = name self.value: float = 0. 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 # 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() symbols_to_cleanup: List[Symbol] = [] 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._continuous_future.symbol}' ) 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( before_target_future, after_target_future ) ) found = 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._continuous_future.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._continuous_future.symbol, self._algo.time ) if len(future_contracts) == 0: self._algo.log( f'ImpliedVolatilityIndicator._find_futures_contract: No futures found for: {self._continuous_future.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 def _find_atm_options( 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(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: No options found for: {target_future.symbol}' ) yield None 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: 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, after_target_option, before_target_future: Future, after_target_future: Future ) -> float: # calculate implied volatility for each option using black model implied_vols: List[float] = [ self._black_model( o.ask_price, f.ask_price, o.strike_price, ql.Option.Call if o.right == OptionRight.CALL else ql.Option.Put, o.expiry, self._algo.time ) for o, f in zip( [before_target_option, after_target_option], [before_target_future, after_target_future], ) ] # 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 _black_model( self, option_price: float, forward_price: float, strike_price: float, option_type: int, expiration_date: datetime, calc_date: datetime, discount_factor: float = 1. ) -> float: implied_vol: float = ql.blackFormulaImpliedStdDev(option_type, strike_price, forward_price, option_price, discount_factor) # strikepayoff: float = ql.PlainVanillaPayoff(option_type, strike_price) # black = ql.BlackCalculator(strikepayoff, forward_price, implied_vol, discount_factor) t: float = (expiration_date - calc_date).days / 360 implied_vol: float = implied_vol / np.sqrt(t) return implied_vol #, black.delta(discount_factor * forward_price) 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 # 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 @abstractmethod def _find_port_options(self) -> List[Symbol]: ... @abstractmethod def _stress(self) -> List[StressOptionInfo]: ... class StressIndexUnderlying(StressUnderlying): 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: super(StressIndexUnderlying, self).__init__( algo, underlying, underlying_price_change_perc, underlying_vol_change_perc, vix, vix_target ) @classmethod def using_fixed_percentage( cls, algo: QCAlgorithm, underlying: Symbol, underlying_price_change_perc: float, underlying_vol_change_perc: float ) -> 'StressIndexUnderlying': return cls(algo, underlying, underlying_price_change_perc, underlying_vol_change_perc, None, None) @classmethod def using_vix( cls, algo: QCAlgorithm, underlying: Symbol, vix: Symbol, vix_target: float ) -> 'StressIndexUnderlying': return cls(algo, underlying, None, None, vix, vix_target) @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 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() dividends: float = self.option_pricing.get_dividends(holding.key) discount_factor: float = self.option_pricing.get_discount_factor(rfr, dividends, 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 black: BlackModelResult = OptionPricingModel.black_model( self._algo.securities[holding.key].price, forward_price, holding.key.id.strike_price, holding.key.id.option_right, holding.key.id.date, self._algo.time, discount_factor ) result.append( StressOptionInfo( holding.key, spot_price, forward_price, discount_factor, dividends, black.implied_volatility, black.option_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 # OR # 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 if self._vix is not None: vix_diff: float = (self._algo.current_slice.bars.get(self._vix).close - self._vix_target) / 100 # vix_diff: float = (self._algo.securities[self._vix].price - self._vix_target) / 100 stressed_spot_price: float = option_info.spot_price stressed_iv: float = option_info.implied_volatility * (1 + vix_diff) else: 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 = OptionPricingModel.black_model_opt_price( stressed_forward_price, option_symbol.id.strike_price, option_symbol.id.option_right, stressed_iv, option_symbol.id.date, self._algo.time, 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 StressIndexUnderlying 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.BrokerageModel, FuncSecuritySeeder(self.GetLastKnownPrices))) 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.realized_option_profits = 0 self._strategy_parser(self.strategies) self._vix: Symbol = self.add_index("VIX", Resolution.MINUTE).symbol # stress indicator initialized with fixed percentage # self._stress_indicator: StressIndexUnderlying = StressIndexUnderlying.using_fixed_percentage( # self, # underlying = spx, # underlying_price_change_perc=-0.1, # underlying_vol_change_perc=-0.1 # ) # # OR # # stress indicator initialized with vix self._stress_indicator: StressIndexUnderlying = StressIndexUnderlying.using_vix( algo=self, underlying=spx, vix=self._vix, vix_target=18 ) self._iv_monitor: IVMonitor = IVMonitor( algo=self, underlying=spx, include_dividends=True, abs_delta_to_notify=0.05 ) # 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, # continuous_future=self._future, # name='ImpliedVolatility', # moneyness=self._moneyness, # target_expiry_days=60, # ) # self.register_indicator(self._future.symbol, 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, None) ''' 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) 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) # ) 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 from decimal import Decimal # 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) black: BlackModelResult = OptionPricingModel.black_model( opt_bar.close, forward_price, option_symbol.id.strike_price, option_symbol.id.option_right, option_symbol.id.date, self._algo.time, 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 = black.implied_volatility # 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(black.option_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 # endregion @dataclass class BlackModelResult(): implied_volatility: float delta: float option_price: float 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: ... @staticmethod def black_model( option_price: float, forward_price: float, strike_price: float, option_right: OptionRight, expiration_date: datetime, calc_dt: datetime, discount_factor: float = 1 ) -> BlackModelResult: option_type = ql.Option.Call if option_right == OptionRight.CALL else ql.Option.Put 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 = OptionPricingModel.get_t_days(expiration_date, calc_dt) / 360 implied_vol = implied_vol / np.sqrt(t) opt_price: float = black.value() return BlackModelResult( implied_vol, black.delta(discount_factor * forward_price), opt_price ) @staticmethod def black_model_opt_price( forward_price: float, strike_price: float, option_right: OptionRight, implied_vol: float, expiration_date: datetime, calc_dt: datetime, discount_factor: float = 1 ) -> float: option_type = ql.Option.Call if option_right == OptionRight.CALL else ql.Option.Put # t: float = (expiration_date - calc_date).days / 360 t: float = OptionPricingModel.get_t_days(expiration_date, calc_dt) / 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 @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)
#region imports from AlgorithmImports import * from abc import ABC, abstractmethod #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) 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_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_option_closest_to_target_strike(self, symbols, target_strike: int): # 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 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 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 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 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
#region imports from AlgorithmImports import * from abc import ABC, abstractmethod from indicators.stress_test import StressIndexUnderlying 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: StressIndexUnderlying): series_list = [ { 'name': 'Stressed Option Holdings', 'type': SeriesType.Line, 'unit': '$' }, { 'name': 'Original Option Holdings', 'type': SeriesType.Line, 'unit': '$' }, ] self._stress_indicator: StressIndexUnderlying = 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: StressIndexUnderlying): series_list = [ { 'name': 'Stressed Portfolio Value', 'type': SeriesType.Line, 'unit': '$' }, { 'name': 'Original Portfolio Value', 'type': SeriesType.Line, 'unit': '$' } ] self._stress_indicator: StressIndexUnderlying = 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')
