Universes
Dataless Scheduled Universes
Introduction
A dataless scheduled universe let's you select a set of assets on a specific schedule.
You can control which days the other types of universe run by adjusting the Schedule universe setting.
However, the schedule
universe setting doesn't accept a TimeRule
argument, so you can't control the time of the day they run.
In contrast, a dataless scheduled universe accepts a TimeRule
argument, but its selection function only receives the algorithm time.
Create Universes
To add a dataless scheduled universe, in the initialize
method, call the add_universe
method with a ScheduledUniverse
object.
class MyUniverseAlgorithm(QCAlgorithm): def initialize(self) -> None: self.universe_settings.asynchronous = True self._universe = self.add_universe( ScheduledUniverse( self.date_rules.month_start(), self.time_rules.at(8, 0), self._select_symbols ) ) def _select_symbols(self, dt): return [Symbol.create("SPY", SecurityType.EQUITY, Market.USA)]
The following table describes the arguments the model accepts:
Argument | Data Type | Description | Default Value |
---|---|---|---|
date_rule | IDateRule | Date rule that defines what days the universe selection function runs | |
time_rule | ITimeRule | Time rule that defines what times on each day selected by date rule the universe selection function runs | |
Callable[[datetime], List[Symbol]] |
selector | Selector function that accepts the current date time and returns the universe's Symbol objects | |
settings | UniverseSettings | The universe settings. If you don't provide an argument, the model uses the algorithm.universe_settings by default. | None |
Date Rules
The following table describes the supported DateRules
:
Member | Description |
---|---|
self.date_rules.set_default_time_zone(time_zone: DateTimeZone)
| Sets the time zone for the DateRules object used in all methods in this table. The default time zone is the algorithm time zone. |
self.date_rules.on(year: int, month: int, day: int)
| Trigger an event on a specific date. |
self.date_rules.on(dates: List[datetime])
| Trigger an event on specific dates. |
self.date_rules.every_day()
| Trigger an event every day. |
self.date_rules.every_day(symbol: Symbol, extended_market_hours: bool = False)
| Trigger an event every day a specific symbol is trading. |
self.date_rules.every(days: List[DayOfWeek])
| Trigger an event on specific days throughout the week. To view the DayOfWeek enum members, see DayOfWeek Enum in the .NET documentation. |
self.date_rules.month_start(days_offset: int = 0)
| Trigger an event on the first day of each month plus an offset. |
self.date_rules.month_start(symbol: Symbol, daysOffset: int = 0)
| Trigger an event on the first tradable date of each month for a specific symbol plus an offset. |
self.date_rules.month_end(days_offset: int = 0)
| Trigger an event on the last day of each month minus an offset. |
self.date_rules.month_end(symbol: Symbol, daysOffset: int = 0)
| Trigger an event on the last tradable date of each month for a specific symbol minus an offset. |
self.date_rules.week_start(days_offset: int = 0)
| Trigger an event on the first day of each week plus an offset. |
self.date_rules.week_start(symbol: Symbol, days_offset: int = 0)
| Trigger an event on the first tradable date of each week for a specific symbol plus an offset. |
self.date_rules.week_end(days_offset: int = 0)
| Trigger an event on the last day of each week minus an offset. |
self.date_rules.week_end(symbol: Symbol, days_offset: int = 0)
| Trigger an event on the last tradable date of each week for a specific symbol minus an offset. |
self.date_rules.year_start(days_offset: int = 0)
| Trigger an event on the first day of each year plus an offset. |
self.date_rules.year_start(symbol: Symbol, days_offset: int = 0)
| Trigger an event on the first tradable date of each year for a specific symbol plus an offset. |
self.date_rules.year_end(days_offset: int = 0)
| Trigger an event on the last day of each year minus an offset. |
self.date_rules.year_end(symbol: Symbol, days_offset: int = 0)
| Trigger an event on the last tradable date of each year for a specific symbol minus an offset. |
self.date_rules.today | Trigger an event once today. |
self.date_rules.tomorrow | Trigger an event once tomorrow. |
To define custom date rules, create a FuncDateRule
object.
The FuncDateRule
constructor expects a name
argument of type str
and a get_dates_function
argument of type Callable[[datetime, datetime], List[datetime]]
.
The get_dates_function
function receives the start and end dates of the algorithm and returns a list of dates for the date rule.
In live trading, the end date is 12/31/2025.
The following example demonstrates how to define a date rule that represents the 10th day of each month:
# Create a date rule that specifies the 10th day of each month. date_rule = FuncDateRule( name="10th_day_of_the_month", get_dates_function=lambda start, end: [ datetime(year, month, 10) for year in range(start.year, end.year) for month in range(1,12) ] )
Time Rules
The following table describes the supported TimeRules
:
Member | Description |
---|---|
self.time_rules.set_default_time_zone(time_zone: DateTimeZone)
| Sets the time zone for the TimeRules object used in all methods in this table, except when a different time zone is given. The default time zone is the algorithm time zone. |
self.time_rules.before_market_open(symbol: Symbol, minutes_before_open: float = 0, extended_market_open: bool = False)
| Trigger an event a few minutes before market open for a specific symbol (default is 0). This rule doesn't work for Crypto securities or custom data. |
self.time_rules.after_market_open(symbol: Symbol, minutes_after_open: float = 0, extended_market_open: bool = False)
| Trigger an event a few minutes after market open for a specific symbol (default is 0). This rule doesn't work for Crypto securities or custom data. |
self.time_rules.before_market_close(symbol: Symbol, minutes_before_close: float = 0, extended_market_open: bool = False)
| Trigger an event a few minutes before market close for a specific symbol (default is 0). This rule doesn't work for Crypto securities or custom data. |
self.time_rules.after_market_close(symbol: Symbol, minutes_after_close: float = 0, extended_market_open: bool = False)
| Trigger an event a few minutes after market close for a specific symbol (default is 0). This rule doesn't work for Crypto securities or custom data. |
self.time_rules.every(interval: timedelta)
| Trigger an event every period interval starting at midnight. |
self.time_rules.now | Trigger an event at the current time of day. |
self.time_rules.midnight | Trigger an event at midnight. |
self.time_rules.noon | Trigger an event at noon. |
self.time_rules.at(hour: int, minute: int, second: int = 0)
| Trigger an event at a specific time of day (e.g. 13:10). |
self.time_rules.at(hour: int, minute: int, second: int, time_zone: DateTimeZone)
| Trigger an event at a specific time of day in the given time zone (e.g. 13:10 UTC). |
To define custom time rules, create a FuncTimeRule
object.
The FuncTimeRule
constructor expects a name
argument of type str
and a create_utc_event_times_function
argument of type Callable[[List[datetime]], List[datetime]]
.
The function receives the list of dates from the date rule and then returns a list of datetime
that define the time rule.
time_rule = FuncTimeRule( name="CustomTimeRule", create_utc_event_times_function=lambda dates: [d + timedelta(hours=10) for d in dates] )
Examples
The following examples demonstrate some common practices for dataless scheduled universes.
Example 1: Download External Universe Files
The following algorithm downloads a CSV file from Dropbox. The first column in the file contains the universe date. The second column in the file contains a list of Equity tickers that represents the universe for the day. The dataless scheduled universe parses the file contents each trading day and adds the universe constituents.
from io import StringIO class DatalessScheduledUniverseExampleAlgorithm(QCAlgorithm): def initialize(self) -> None: self.set_start_date(2015, 1, 1) self.set_end_date(2016, 1, 1) # Download the initial universe file. self._download_universe_file() # Add the custom universe. self.universe_settings.extended_market_hours = True spy = Symbol.create('SPY', SecurityType.EQUITY, Market.USA) date_rule = self.date_rules.every_day(spy) self._universe = self.add_universe( ScheduledUniverse(date_rule, self.time_rules.at(8, 0), self._select_assets) ) # Schedule rebalances for market open. self.schedule.on(date_rule, self.time_rules.after_market_open(spy, 0), self._rebalance) def _download_universe_file(self): # Download the universe CSV file. Dropbox links require the "dl=1" URL parameter. file = self.download( "https://www.dropbox.com/scl/fi/fbrxitk4ec3w91nse1raa/df.csv?rlkey=7r042rukzkthp7y1srloyhkov&st=5r4sdfwd&dl=1" ) # Convert the CSV file data into a dictionary where the key is the date and # the value is a comma-separated string of stock tickers. df = pd.read_csv(StringIO(file), index_col=0).iloc[:, 0] df.index = pd.to_datetime(df.index).date self._tickers_by_date = df.to_dict() def _select_assets(self, dt: datetime) -> List[Symbol]: # When live trading, re-download the CSV file each day to get the new rows. if self.live_mode: self._download_universe_file() # Get the current day's data from the CSV file. data = self._tickers_by_date.get(dt.date(), '') # If there isn't an entry for the current date, return an empty universe. if not data: return [] # Convert the stock tickers in the CSV file to Symbol objects. return [Symbol.create(x, SecurityType.EQUITY, Market.USA) for x in data.split(",")] def _rebalance(self) -> None: # Form an equal-weighted portfolio of all the universe constituents. symbols = list(self._universe.selected) weight = 1 / len(symbols) self.set_holdings([PortfolioTarget(symbol, weight) for symbol in symbols], True)
Example 2: Quarter End Selection
The following algorithm selects SPY on the last month of each quarter. For the remaining months, it selects no assets.
class DatalessScheduledUniverseDemoAlgorithm(QCAlgorithm): def initialize(self): self.add_universe( ScheduledUniverse( self.date_rules.month_start(), self.time_rules.at(8, 0), self._select_assets ) ) def _select_assets(self, dt: datetime) -> List[Symbol]: if dt.month % 3 == 0: return [Symbol.create("SPY", SecurityType.EQUITY, Market.USA)] return []
Example 3: Third Week VIX
Standard Options expire at end of the third week of each month. The following algorithm selects VIX-related products on the third week to trade the foreseeable increase in volatility.
class DatalessScheduledUniverseDemoAlgorithm(QCAlgorithm): def initialize(self): self._month = -1 self._week = -1 self.add_universe( ScheduledUniverse( self.date_rules.week_start(), self.time_rules.at(8, 0), self._select_assets ) ) def _select_assets(self, dt: datetime) -> List[Symbol]: if dt.month == self._month: self._week += 1 if self._week == 3: return [Symbol.create("VXZ", SecurityType.EQUITY, Market.USA)] return [] self._month = dt.month self._week = 0 return []
Other Examples
For more examples, see the following algorithms: