Universes
Equity CFD
Introduction
An Equity CFD universe lets you select a set of Equity CFD assets. To select the underlying Equities, you can use a fundamental universe to select assets based on corporate fundamental data, an ETF constituents universe to select the CFD versions of stocks in an exchange-traded fund, or an alternative data universe to select assets based on alternative data. This feature is powerful for European clients seeking to trade US ETF products.
Only the Interactive Brokers CFD integration supports trading Stock-CFD products.
QuantConnect doesn't have historical data for Interactive Brokers CFD products;
however, you can use the live_mode
flag to swap to the CFD equivalents for live trading.
In live trading, you must include Interactive Brokers as a data provider to trade Equity-CFD products.
Create Universes Using Fundamentals
To add a fundamental universe, in the initialize
method, pass a filter function to the add_universe
method. The filter function receives a list of Fundamental
objects and must return a list of Symbol
objects. The Symbol
objects you return from the function are the constituents of the fundamental universe and LEAN automatically creates subscriptions for them. In live mode, call the Symbol.create
method to swap for a CFD version of the same Symbol
. Don't call add_cfd
in the filter function.
# Create a universe for CFD securities based on fundamental data. class MyUniverseAlgorithm(QCAlgorithm): def initialize(self) -> None: self.universe_settings.asynchronous = True # Pass the filter function as an argument to the AddUniverse method. self._universe = self.add_universe(self._fundamental_function) def _fundamental_function(self, fundamental: List[Fundamental]) -> List[Symbol]: symbols = [c.symbol for c in fundamental if c.has_fundamental_data] if self.live_mode: return [Symbol.create(x.value, SecurityType.CFD, Market.INTERACTIVE_BROKERS) for x in symbols] return symbols
Example
The simplest example of accessing the fundamental object would be harnessing the iconic PE ratio for a stock. This is a ratio of the price it commands to the earnings of a stock. The lower the PE ratio for a stock, the more affordable it appears.
# Select the top 50 by dollar volume and then select the top 10 by their PE ratio. self.universe_settings.asynchronous = True self._universe = self.add_universe(self._fundamental_selection_function) def _fundamental_selection_function(self, fundamental: List[Fundamental]) -> List[Symbol]: filtered = [f for f in fundamental if f.price > 10 and f.has_fundamental_data and not np.isnan(f.valuation_ratios.pe_ratio)] sorted_by_dollar_volume = sorted(filtered, key=lambda f: f.dollar_volume, reverse=True)[:100] sorted_by_pe_ratio = sorted(sorted_by_dollar_volume, key=lambda f: f.valuation_ratios.pe_ratio, reverse=False)[:10] symbols = [f.symbol for f in sorted_by_pe_ratio] if self.live_mode: return [Symbol.create(x.value, SecurityType.CFD, Market.INTERACTIVE_BROKERS) for x in symbols] return symbols
Practical Limitations
Fundamental universes allow you to select an unlimited universe of assets to analyze. Each asset in the universe consumes approximately 5MB of RAM, so you may quickly run out of memory if your universe filter selects many assets. If you backtest your algorithms in the Algorithm Lab, familiarize yourself with the RAM capacity of your backtesting and live trading nodes. To keep your algorithm fast and efficient, only subscribe to the assets you need.
Create Universes using ETF Constituents
To add an ETF Constituents universe, in the initialize
method, call the universe.etf
method with a filter function, and pass it to the add_universe
method. The filter function receives a list of ETFConstituentUniverse
objects and must return a list of Symbol
objects. The Symbol
objects you return from the function are the constituents of the fundamental universe and LEAN automatically creates subscriptions for them. In live mode, call the Symbol.create
method to swap for a CFD version of the same Symbol
. Don't call add_cfd
in the filter function.
class MyUniverseAlgorithm(QCAlgorithm): def initialize(self) -> None: self.universe_settings.asynchronous = True self._universe = self.universe.etf("SPY", self.universe_settings, self._etf_constituents_filter) self.add_universe(self._universe) def _etf_constituents_filter(self, constituents: List[ETFConstituentUniverse]) -> List[Symbol]: # Get the 10 securities with the largest weight in the index selected = sorted([c for c in constituents if c.weight], key=lambda c: c.weight, reverse=True)[:10] symbols = [c.symbol for c in selected] if self.live_mode: return [Symbol.create(x.value, SecurityType.CFD, Market.INTERACTIVE_BROKERS) for x in symbols] return symbols
Selection Frequency
Equity universes run on a daily basis by default. To adjust the selection schedule, see Schedule.
Live Trading Considerations
The live data for fundamental universe selection arrives at 6/7 AM Eastern Time (ET), so fundamental universe selection runs for live algorithms between 7 and 8 AM ET. This timing allows you to place trades before the market opens. Don't schedule anything for midnight because the universe selection data isn't ready yet.
Examples
The following examples demonstrate common practices for implementing Equity CFD's universe.
Example 1: Short Bottom-Weighted SPY Constituents
A subset of the SPY constituents outperforms the SPY, while many underperform the overall index. To exclude the ETF constituents that underperform the index, the following algorithm buys the SPY CFD and shorts the CFD of the 100 constituents in the index with the smallest weight in the ETF.
class EquityCfdUniverseAlgorithm(QCAlgorithm): def initialize(self) -> None: self.set_start_date(2023, 6, 1) self.set_cash(10_000_000) self.settings.minimum_order_margin_portfolio_percentage = 0 # Request SPY to trade. In live mode, request the IB CFD self._spy = self.add_cfd("SPY", market= Market.InteractiveBrokers).symbol if self.live_mode else self.add_equity("SPY").symbol # The SPY ETF provides market hours for scheduled events. spy = Symbol.create("SPY", SecurityType.EQUITY, Market.USA) # Refilter and rebalance weekly to avoid over-trading and high transaction costs. self.universe_settings.schedule.on(self.date_rules.week_end(spy)) # Add a universe to select the CFD contracts of the SPY constituents. self._universe = self.add_universe(self.universe.etf(spy, universe_filter_func=self._select_assets)) # Create a scheduled event to rebalance the portfolio. self.schedule.on( self.date_rules.week_start(spy), self.time_rules.after_market_open(spy, 1), self._rebalance ) def _select_assets(self, constituents: List[ETFConstituentUniverse]) -> List[Symbol]: # Cache the constituent weights in a dictionary for filtering and position sizing. self._etf_weight_by_symbol = {c.symbol: c.weight for c in constituents if c.weight} # Select the CFD contracts of the bottom-weighted 100 constituents. # They should have negative excess returns. def get_symbol(symbol): if self.live_mode: return Symbol.create(symbol.value, SecurityType.CFD, Market.INTERACTIVE_BROKERS) return symbol return [ get_symbol(symbol) for symbol, _ in sorted( self._etf_weight_by_symbol.items(), key=lambda x: x[1] )[:100] ] def _rebalance(self) -> None: # Get the ETF weight of all the assets currently in the universe. weight_by_symbol = { symbol: self._etf_weight_by_symbol[symbol] for symbol in self._universe.selected } # Sell the CFD of the 100 ETF constituents with the lowest weight due to expected negative excess return. targets = [PortfolioTarget(symbol, -weight) for symbol, weight in weight_by_symbol.items()] # Buy the SPY CFD to eliminate systematic risk. targets.append(PortfolioTarget(self._spy, 1 - sum(weight_by_symbol.values()))) # Place orders to rebalance the portfolio. self.set_holdings(targets, True)