Overall Statistics |
Total Orders 3162 Average Win 0.07% Average Loss -0.06% Compounding Annual Return -1.911% Drawdown 16.900% Expectancy -0.100 Start Equity 50000 End Equity 45394.40 Net Profit -9.211% Sharpe Ratio -0.713 Sortino Ratio -0.201 Probabilistic Sharpe Ratio 0.199% Loss Rate 61% Win Rate 39% Profit-Loss Ratio 1.28 Alpha -0.03 Beta -0.001 Annual Standard Deviation 0.042 Annual Variance 0.002 Information Ratio -0.656 Tracking Error 0.164 Treynor Ratio 42.116 Total Fees $3193.95 Estimated Strategy Capacity $3300000.00 Lowest Capacity Asset IVV RUTTRZ1RC7L1 Portfolio Turnover 91.50% |
#region imports from AlgorithmImports import * #endregion class ArbitrageAlphaModel(AlphaModel): """ This class monitors the intraday bid and ask prices of two correlated ETFs. When the bid price of ETF A (B) diverts high enough away from the ask price of ETF B (A) such that the profit_pct_threshold is reached, we start a timer. If the arbitrage opportunity is still present after the specified timesteps, we enter the arbitrage trade by going long ETF B (A) and short ETF A (B). When the spread reverts back to where the bid of ETF B (A) >= the ask of ETF A (B) for the same number of timesteps, we exit the trade. To address a trending historical spread between the two ETFs, we adjust the spread by removing the mean spread over a lookback window. """ _symbols = [] # IVV, SPY _entry_timer = [0, 0] _exit_timer = [0, 0] _spread_adjusters = [0, 0] _long_side = -1 _consolidators = {} _history = {} def __init__(self, order_delay = 3, profit_pct_threshold = 0.02, window_size = 400): """ Input: - order_delay The number of timesteps to wait while an arbitrage opportunity is present before emitting insights (>= 0) - profit_pct_threshold The amount of adjusted profit there must be in an arbitrage opportunity to signal a potential entry (> 0) - window_size The length of the lookback window used to calculate the spread adjusters (to address the trending spread b/w ETFs) (> 0) """ self._order_delay = order_delay self._pct_threshold = profit_pct_threshold / 100 self._window_size = window_size self._consolidated_update = 0 def update(self, algorithm, data): """ Called each time our alpha model receives a new data slice. Input: - algorithm Algorithm instance running the backtest - data Data for the current time step in the backtest Returns a list of Insights to the portfolio construction model. """ if algorithm.is_warming_up: return [] quotebars = self._get_quotebars(data) if not quotebars: return [] # Ensure we are not within 5 minutes of either the open or close exchange = algorithm.securities['SPY'].exchange if (not exchange.date_time_is_open(algorithm.time - timedelta(minutes=5)) and exchange.date_time_is_open(algorithm.time + timedelta(minutes=5))): return [] # Search for entries for i in range(2): if quotebars[abs(i-1)].bid.close / quotebars[i].ask.close - self._spread_adjusters[abs(i-1)] >= self._pct_threshold: self._entry_timer[i] += 1 if self._entry_timer[i] == self._order_delay: self._exit_timer = [0, 0] if self._long_side == i: return [] self._long_side = i return [Insight.price(self._symbols[i], timedelta(days=9999), InsightDirection.UP), Insight.price(self._symbols[abs(i-1)], timedelta(days=9999), InsightDirection.DOWN)] else: return [] self._entry_timer[i] = 0 # Search for an exit if self._long_side >= 0: # In a position if quotebars[self._long_side].bid.close / quotebars[abs(self._long_side-1)].ask.close - self._spread_adjusters[self._long_side] >= 0: # Exit signal self._exit_timer[self._long_side] += 1 if self._exit_timer[self._long_side] == self._order_delay: # Exit signal lasted long enough self._exit_timer[self._long_side] = 0 i = self._long_side self._long_side = -1 return [Insight.price(self._symbols[i], timedelta(days=9999), InsightDirection.FLAT), Insight.price(self._symbols[abs(i-1)], timedelta(days=9999), InsightDirection.FLAT)] else: return [] return [] def on_securities_changed(self, algorithm, changes): """ Called each time our universe has changed. Inputs: - algorithm Algorithm instance running the backtest - changes The additions and subtractions to the algorithm's security subscriptions """ added_symbols = [security.symbol for security in changes.added_securities] if len(added_symbols) > 0: self._symbols.extend(added_symbols) if len(self._symbols) != 2: algorithm.error(f"ArbitrageAlphaModel must have 2 symbols to trade") algorithm.quit() return history = algorithm.history(self._symbols, self._window_size, Resolution.SECOND)[['bidclose', 'askclose']] starting_row_count = min([history.loc[symbol].shape[0] for symbol in self._symbols]) for symbol in self._symbols: self._history[symbol] = {'bids': history.loc[symbol].bidclose.to_numpy()[-starting_row_count:], 'asks': history.loc[symbol].askclose.to_numpy()[-starting_row_count:]} self._update_spread_adjusters() # Setup daily consolidators to update spread_adjusters for symbol in self._symbols: self._consolidators[symbol] = QuoteBarConsolidator(1) self._consolidators[symbol].data_consolidated += self._custom_daily_handler algorithm.subscription_manager.add_consolidator(symbol, self._consolidators[symbol]) for removed in changes.removed_securities: algorithm.subscription_manager.remove_consolidator(removed.symbol, self._consolidators[removed.symbol]) def _custom_daily_handler(self, sender, consolidated): """ Updates the rolling lookback window with the latest data. Inputs - sender Function calling the consolidator - consolidated Tradebar representing the latest completed trading day """ # Add new data point to history while removing expired history self._history[consolidated.symbol]['bids'] = np.append(self._history[consolidated.symbol]['bids'][-self._window_size:], consolidated.bid.close) self._history[consolidated.symbol]['asks'] = np.append(self._history[consolidated.symbol]['asks'][-self._window_size:], consolidated.ask.close) # After updating the history of both symbols, update the spread adjusters self._consolidated_update += 1 if self._consolidated_update == 2: self._consolidated_update = 0 self._update_spread_adjusters() def _get_quotebars(self, data): """ Extracts the QuoteBars from the given slice. Inputs - data Latest slice object the algorithm has received Returns the QuoteBars for the symbols we are trading. """ if not all([data.quote_bars.contains_key(symbol) for symbol in self._symbols]): return [] quotebars = [data.quote_bars[self._symbols[i]] for i in range(2)] if not all([q is not None for q in quotebars]): return [] # Ensure ask > bid for each ETF if not all([q.ask.close > q.bid.close for q in quotebars]): return [] return quotebars def _update_spread_adjusters(self): """ Updates the spread adjuster by finding the mean of the trailing spread ratio b/w ETFs. """ for i in range(2): numerator_history = self._history[self._symbols[i]]['bids'] denominator_history = self._history[self._symbols[abs(i-1)]]['asks'] self._spread_adjusters[i] = (numerator_history / denominator_history).mean()
#region imports from AlgorithmImports import * from alpha import ArbitrageAlphaModel #endregion class IntradayIndexETFArbitrageAlgorithm(QCAlgorithm): def initialize(self): self.set_start_date(2015, 8, 11) self.set_end_date(2020, 8, 11) self.set_cash(50000) tickers = ['IVV', 'SPY'] symbols = [ Symbol.create(t, SecurityType.EQUITY, Market.USA) for t in tickers ] self.set_universe_selection(ManualUniverseSelectionModel(symbols)) self.universe_settings.resolution = Resolution.SECOND self.add_alpha(ArbitrageAlphaModel()) self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel(lambda _: None)) self.settings.rebalance_portfolio_on_security_changes = False self.set_execution(ImmediateExecutionModel())