How to Create a Custom Symbol Selection Function in Python for Your QuantConnect Strategy

Here is a complete, runnable example demonstrating how to write a custom function that requests historical data, calculates a performance metric (in this case, Momentum), and returns a ranked list of symbols to trade.

Strategy Overview

1. Universe: A manual list of liquid technology stocks.
2. Ranking Function: get_ranked_symbols. This function takes a list of symbols, retrieves 30 days of history, calculates the percentage change, and returns the top 3 performers.
3. Execution: A Scheduled Event runs at the start of every month to call this function and rebalance the portfolio.

Python Code

# region imports
from AlgorithmImports import *
# endregion

class RankingFunctionAlgorithm(QCAlgorithm):

    def initialize(self):
        self.set_start_date(2020, 1, 1)
        self.set_end_date(2023, 1, 1)
        self.set_cash(100000)

        # 1. Define a list of tickers to analyze
        tickers = ["AAPL", "MSFT", "GOOG", "AMZN", "TSLA", "NVDA", "AMD", "INTC"]
        self.symbols = [self.add_equity(ticker, Resolution.DAILY).symbol for ticker in tickers]
        
        # 2. Schedule the rebalancing event (e.g., first trading day of the month)
        self.schedule.on(
            self.date_rules.month_start(self.symbols[0]),
            self.time_rules.after_market_open(self.symbols[0], 30),
            self.rebalance
        )
        
        # Parameter: Number of days to look back for ranking
        self.lookback = 30
        # Parameter: Number of top symbols to trade
        self.target_count = 3

    def rebalance(self):
        """
        Scheduled event to rebalance the portfolio based on ranking.
        """
        # Call our custom ranking function
        top_symbols = self.get_ranked_symbols(self.symbols, self.lookback, self.target_count)

        # If no symbols returned (e.g., not enough data), do nothing
        if not top_symbols:
            return

        self.debug(f"Top {self.target_count} Symbols: {[s.value for s in top_symbols]}")

        # 1. Liquidate symbols that are no longer in the top list
        for holding_symbol in self.portfolio.keys():
            if self.portfolio[holding_symbol].invested and holding_symbol not in top_symbols:
                self.liquidate(holding_symbol)

        # 2. Allocate equal weight to the top symbols
        # 1.0 / 3 = 33% allocation each
        target_weight = 1.0 / len(top_symbols)
        
        for symbol in top_symbols:
            self.set_holdings(symbol, target_weight)

    def get_ranked_symbols(self, symbols_list, lookback_days, top_n):
        """
        Custom function to rank symbols based on momentum.
        
        Args:
            symbols_list (list): List of Symbol objects to rank.
            lookback_days (int): Number of days of history to request.
            top_n (int): Number of symbols to return.
            
        Returns:
            list: A list of the top N Symbol objects sorted by performance.
        """
        # 1. Request historical data
        # We request 'lookback_days' + 5 to ensure we have enough trading days to cover weekends/holidays
        history = self.history(symbols_list, lookback_days + 5, Resolution.DAILY)
        
        if history.empty:
            self.debug("History is empty.")
            return []

        # 2. Prepare the data
        # 'close' column is selected. 
        # .unstack(level=0) pivots the table so columns are Symbols and rows are Time.
        closes = history['close'].unstack(level=0)

        # 3. Calculate Momentum (Percentage Change)
        # Formula: (Last Price - Price N days ago) / Price N days ago
        # We use iloc to get the last row and the row 'lookback_days' ago
        # We check if we have enough rows first
        if len(closes) < lookback_days:
            return []
            
        # Calculate returns from the start of the window to the end
        # iloc[-1] is the most recent price, iloc[0] is the oldest price in the window
        returns = (closes.iloc[-1] / closes.iloc[0]) - 1.0

        # 4. Sort and Filter
        # sort_values sorts ascending by default, so we use ascending=False for highest returns first
        sorted_returns = returns.sort_values(ascending=False)
        
        # Take the top N symbols
        top_performers = sorted_returns.head(top_n)
        
        # The index of this series contains the Symbol objects (mapped from the columns of 'closes')
        # We convert the index to a standard python list
        return top_performers.index.tolist()

Key Implementation Details

1. self.history: This is the core API method. It returns a MultiIndex DataFrame (indexes are Symbol and Time).
2. unstack(level=0): This pandas method is crucial. It transforms the data from a "tall" format (stacked) to a "wide" format where every column represents a specific stock symbol. This makes vectorized calculations (like calculating returns for all stocks at once) very efficient.
3. Data Validation: The code checks if history.empty and ensures len(closes) is sufficient before calculating to prevent index errors during the warm-up period.
4. Return Type: The function returns a list of Symbol objects, which can be passed directly to self.set_holdings or self.liquidate.

Q&A

Q: Can I use this function for Universe Selection instead of a fixed list?
A: Yes. In a Coarse/Fine Universe Selection model, you receive coarse or fine objects. You can extract symbols from those objects, pass them to this function, and return the Symbol objects to the framework. However, requesting history for thousands of stocks in CoarseSelection can be slow; it is better to use the coarse object's native properties (like x.Price or x.DollarVolume) for initial filtering.

Q: How do I rank by volatility instead of momentum?
A: You would modify the calculation step in get_ranked_symbols. Instead of calculating percentage change, you would calculate the standard deviation of the percentage changes:

# Calculate daily percent changes
daily_returns = closes.pct_change().dropna()
# Calculate standard deviation (volatility)
volatility = daily_returns.std()
# Sort by lowest volatility (ascending=True)
sorted_volatility = volatility.sort_values(ascending=True)

Q: Why do you add +5 days to the history request?
A: self.history(days=30) requests data for the last 30 calendar days. If there are weekends or holidays, you might only get ~20 trading bars. Adding a buffer ensures you get enough actual data points to perform the calculation. Alternatively, you can use self.history(symbols, 30, Resolution.DAILY) (without specifying days/timedelta) to get exactly 30 trading bars.