book
Checkout our new book! Hands on AI Trading with Python, QuantConnect, and AWS Learn More arrow

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.

Select Language:
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:

ArgumentData TypeDescriptionDefault Value
date_ruleIDateRuleDate rule that defines what days the universe selection function runs
time_ruleITimeRuleTime rule that defines what times on each day selected by date rule the universe selection function runs
Callable[[datetime], List[Symbol]]selectorSelector function that accepts the current date time and returns the universe's Symbol objects
settingsUniverseSettingsThe 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:

MemberDescription
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.todayTrigger an event once today.
self.date_rules.tomorrowTrigger 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:

Select Language:
# 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:

MemberDescription
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.nowTrigger an event at the current time of day.
self.time_rules.midnightTrigger an event at midnight.
self.time_rules.noonTrigger 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.

Select Language:
time_rule = FuncTimeRule(
    name="CustomTimeRule", 
    create_utc_event_times_function=lambda dates: [d + timedelta(hours=10) for d in dates]
)

Selection Frequency

Dataless scheduled universes run at whatever selection schedule you define by the date_rule and time_rule arguments.

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.

Select Language:
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.

Select Language:
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.

Select Language:
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:

You can also see our Videos. You can also get in touch with us via Discord.

Did you find this page helpful?

Contribute to the documentation: