Universes

public class DatalessScheduledUniverseExampleAlgorithm : QCAlgorithm
{
    // Create a dictionary for the universe data where the key is the date and
    // the value is a comma-separated string of stock tickers.
    private Dictionary<DateTime, string> _tickersByDate = new();
    private Universe _universe;

    public override void Initialize()
    {
        SetStartDate(2024, 9, 1);
        SetEndDate(2024, 12, 31);

        // Download the initial universe file.
        DownloadUniverseFile();
        // Add the custom universe.
        UniverseSettings.ExtendedMarketHours = true;
        var spy = QuantConnect.Symbol.Create("SPY", SecurityType.Equity, Market.USA);
        var dateRule = DateRules.EveryDay(spy);
        _universe = AddUniverse(new ScheduledUniverse(dateRule, TimeRules.At(8, 0), SelectAssets));
        // Schedule rebalances for market open.
        Schedule.On(dateRule, TimeRules.AfterMarketOpen(spy, 0), Rebalance);
    }

    private void DownloadUniverseFile()
    {
        // Download the universe CSV file. Dropbox links require the "dl=1" URL parameter.
        var file = 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.
        foreach (var line in file.Split('\n').Skip(1))
        {
            // Skip empty lines.
            if (line.IsNullOrEmpty())
            {
                continue;
            }
            var items = line.Split("\"");
            var date = Parse.DateTimeExact(items[0].Split(",")[0], "yyyy-MM-dd").Date;
            _tickersByDate[date] = items[1];
        }
    }

    private IEnumerable<Symbol> SelectAssets(DateTime dt)
    {
        // When live trading, re-download the CSV file each day to get the new rows.
        if (LiveMode)
        {
            DownloadUniverseFile();
        }
        // Get the current day's data from the CSV file.
        if (!_tickersByDate.TryGetValue(dt.Date, out var tickers))
        {
            // If there isn't an entry for the current date, return an empty universe.
            return Enumerable.Empty<Symbol>();
        }
        // Convert the stock tickers in the CSV file to Symbol objects.
        return tickers
            .Split(',')
            .Select(x => QuantConnect.Symbol.Create(x, SecurityType.Equity, Market.USA));
    }

    private void Rebalance()
    {
        // Form an equal-weighted portfolio of all the universe constituents.
        var weight = 1m / _universe.Selected.Count;
        SetHoldings(_universe.Selected.Select(symbol => new PortfolioTarget(symbol, weight)).ToList(), liquidateExistingHoldings: true);
    }
}

from io import StringIO


class DatalessScheduledUniverseExampleAlgorithm(QCAlgorithm):
    
    def initialize(self) -> None:
        self.set_start_date(2024, 9, 1)
        self.set_end_date(2024, 12, 31)

        # 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], liquidate_existing_holdings=True)

public class DatalessScheduledUniverseDemoAlgorithm : QCAlgorithm
{
    public override void Initialize()
    {
        SetStartDate(2024, 9, 1);
        SetEndDate(2024, 12, 31);
        AddUniverse(new ScheduledUniverse(
            DateRules.MonthStart(), 
            TimeRules.At(8, 0), 
            (dt) => 
            {
                if (dt.Month % 3 == 0)
                {
                    return new[] { QuantConnect.Symbol.Create("SPY", SecurityType.Equity, Market.USA) };
                }
                return Enumerable.Empty<Symbol>();
            }
        ));
    }
}

class DatalessScheduledUniverseDemoAlgorithm(QCAlgorithm):

    def initialize(self):
        self.set_start_date(2024, 9, 1)
        self.set_end_date(2024, 12, 31)
        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 []

public class DatalessScheduledUniverseDemoAlgorithm : QCAlgorithm
{
    private int _month = -1;
    private int _week = -1;
        
    public override void Initialize()
    {
        SetStartDate(2024, 9, 1);
        SetEndDate(2024, 12, 31);
        AddUniverse(new ScheduledUniverse(
            DateRules.WeekStart(), 
            TimeRules.At(8, 0), 
            (dt) => 
            {
                if (dt.Month == _month)
                {
                    if (++_week == 3)
                    {
                        return new[] { QuantConnect.Symbol.Create("VXZ", SecurityType.Equity, Market.USA) };
                    }
                    return Enumerable.Empty<Symbol>();
                }
                _month = dt.Month;
                _week = 0;
                return Enumerable.Empty<Symbol>();
            }
        ));
    }
}

class DatalessScheduledUniverseDemoAlgorithm(QCAlgorithm):

    def initialize(self):
        self._month = -1
        self._week = -1
        self.set_start_date(2024, 9, 1)
        self.set_end_date(2024, 12, 31)
        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 []