| Overall Statistics |
|
Total Orders 489 Average Win 0.12% Average Loss -0.03% Compounding Annual Return -0.682% Drawdown 5.500% Expectancy -0.444 Start Equity 1000000000000 End Equity 966722949995 Net Profit -3.328% Sharpe Ratio -3.623 Sortino Ratio -4.999 Probabilistic Sharpe Ratio 0.008% Loss Rate 89% Win Rate 11% Profit-Loss Ratio 4.16 Alpha -0.028 Beta -0.031 Annual Standard Deviation 0.009 Annual Variance 0 Information Ratio -0.69 Tracking Error 0.18 Treynor Ratio 1.021 Total Fees $0.00 Estimated Strategy Capacity $13000.00 Lowest Capacity Asset SPX 32MMIPHPVXK3Y|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 * from options import * from option_port_tools import * from stats import * import config # endregion class SPX_Options(QCAlgorithm): def Initialize(self): self.SetStartDate(2020, 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) # stress indicator self._stress_indicator: StressIndexUnderlying = StressIndexUnderlying( self, underlying = spx, underlying_price_change_perc=0.1, underlying_vol_change_perc=0.1 ) PortfolioValue(self) OptionsPortfolioValue(self) Greeks(algo=self, strategies=self.strategies) HoldingsValue(self, spy) StressTest(self, self._stress_indicator) ''' portfolio_contents_logger = LogPortfolioContents(self) self.Schedule.On( #self.DateRules.EveryDay('SPX'), #self.TimeRules.BeforeMarketClose('SPX', 0), self.DateRules.WeekStart('SPX'), self.TimeRules.BeforeMarketClose('SPX', 0), portfolio_contents_logger.log_portfolio_contents ) ''' ''' self.Schedule.On( self.DateRules.EveryDay('SPY'), self.TimeRules.BeforeMarketClose('SPY', 30), self._additional_spy ) ''' ''' self.Schedule.On( self.DateRules.EveryDay('SPY'), self.TimeRules.BeforeMarketClose('SPY', 30), self.adjust_spy ) self.Schedule.On( self.DateRules.EveryDay(), self.TimeRules.Midnight, self.increment_rebalance_day ) ''' ''' fwrds = forward_prices_and_discounts_structure(self, spx) self.Schedule.On( self.DateRules.EveryDay('SPX'), self.TimeRules.BeforeMarketClose('SPX', 30), fwrds.get_options ) ''' def _stress(self): self._stress_indicator._stress() def increment_rebalance_day(self): 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 OnData(self, slice): pass # if not self.portfolio[self.spy].invested: # self.market_order(self.spy, 1) 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): 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 * from math import e from abc import ABC, abstractmethod from greeks import * from dataclasses import dataclass # 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: float = 0.1, underlying_vol_change_perc: float = 0.1 ) -> 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 @abstractmethod def _get_forward_price( self, spot_price: float, expiry: datetime, discount_factor: float, dividends: float ) -> float: ... # NOTE _get_discount_factor should probably take some parameters for a different underlyings in the future @abstractmethod def _get_discount_factor(self, expiry: datetime) -> float: ... @abstractmethod def _get_dividends(self, underlying: Symbol) -> float: ... @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: float = 0.1, underlying_vol_change_perc: float = 0.1 ) -> None: super(StressIndexUnderlying, self).__init__( algo, underlying, underlying_price_change_perc, underlying_vol_change_perc ) @property def portfolio_value(self) -> float: # index options holdings in portfolio 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 ]) # for buy options strategy with nothing else in a portfolio # assert self._algo.portfolio.total_holdings_value == original_option_holdings # final portfolio value after stress portfolio_value: float = self._algo.portfolio.total_portfolio_value - 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 _get_forward_price( self, spot_price: float, expiry: datetime, discount_factor: float, dividends: float ) -> float: # TODO inline # F-future price, S-spot price, r-discount factor, q-dividends spot_price: float = spot_price r: float = discount_factor q: float = dividends T: float = expiry t: float = self._algo.time F: float = spot_price*(e**((r-q)*((T-t).days / 360))) return F def _get_discount_factor(self, expiry: datetime) -> float: rfr: float = RiskFreeInterestRateModelExtensions.get_risk_free_rate( self._algo.risk_free_interest_rate_model, self._algo.time, self._algo.time ) df: float = e**(rfr*((expiry - self._algo.time).days / 360)) return df def _get_dividends(self, underlying: Symbol) -> float: return 0. def _find_port_options(self) -> List[StressOptionInfo]: spot_price: float = self._algo.securities[self._underlying].price dividends: float = self._get_dividends(self._underlying) result: List[StressOptionInfo] = [] # 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: discount_factor: float = self._get_discount_factor(holding.key.ID.date) forward_price: float = self._get_forward_price(spot_price, holding.key.ID.date, discount_factor, dividends) # TODO (1) should I find option in portfolio holdings that is already expired according to QC id.date value? 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: Tuple[float] = black_model( self._algo.securities[holding.key].price, forward_price, holding.key.ID.strike_price, ql.Option.Call if holding.key.ID.option_right == OptionRight.CALL else ql.Option.Put, holding.key.ID.date, self._algo.time, discount_factor ) result.append( StressOptionInfo( holding.key, spot_price, forward_price, discount_factor, dividends, black[0], # IV black[2] # 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 stressed_port_options_info: List[StressOptionInfo] = [] for option_info in port_options_info: option_symbol: Symbol = option_info.underlying_option_symbol stressed_spot_price: float = option_info.spot_price * (1 + self._underlying_price_change_perc) stressed_iv: float = option_info.implied_volatility * (1 + self._underlying_vol_change_perc) # discount_factor and dividends are not changed in this instance stressed_forward_price: float = self._get_forward_price( stressed_spot_price, option_symbol.ID.date, option_info.discount_factor, option_info.dividends ) stressed_opt_price: float = black_model_opt_price( stressed_forward_price, option_symbol.ID.strike_price, ql.Option.Call if option_symbol.ID.option_right == OptionRight.CALL else ql.Option.Put, 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 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 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 option_port_tools import StressOptionInfo, StressIndexUnderlying #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'])) 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 StressTest(Charts): def __init__(self, algo, stress_indicator: StressIndexUnderlying): series_list = [ { 'name': 'Stressed Portfolio Value', 'type': SeriesType.Line }, { 'name': 'Original Portfolio Value', 'type': SeriesType.Line } ] 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 PortfolioValue(Charts): def __init__(self, algo): series_list = [ { 'name': 'Portfolio Value', 'type': SeriesType.Line } ] 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 } ] 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 } ] 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 }, { 'name': 'Gamma', 'type': SeriesType.Line }, { 'name': 'Implied Volatility', 'type': SeriesType.Line }, { 'name': 'Theta', 'type': SeriesType.Line }, { 'name': 'Vega', 'type': SeriesType.Line }, { 'name': 'Theta_per_day', 'type': SeriesType.Line }, ] 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')
