Indicators
Indicator Universes
Introduction
An indicator universe uses technical indicators to determine the constituents of the universe. Imagine a universe that only contains assets above their 10-day simple moving average. You can incorporate indicators into any of the types of universes in the Universes chapter. To create an indicator universe, define a helper class that contains the indicators and then define a universe that updates the indicators and selects assets.
Define SymbolData Objects
To make it easy to create and update indicators for each security in the universe, move the indicator logic into a class. In the universe definition, you can create an instance of this class for each security in the universe. The indicators you create in this class should be manual indicators so you can ensure they only update during universe selection.
class SymbolData(object):
def __init__(self, symbol):
self._symbol = symbol
self.tolerance = 1.01
self.fast = ExponentialMovingAverage(100)
self.slow = ExponentialMovingAverage(300)
self.is_uptrend = False
self.scale = 0
def update(self, time, value):
if self.fast.update(time, value) and self.slow.update(time, value):
fast = self.fast.current.value
slow = self.slow.current.value
self.is_uptrend = fast > slow * self.tolerance
if self.is_uptrend:
self.scale = (fast - slow) / ((fast + slow) / 2.0) private class SelectionData
{
public readonly ExponentialMovingAverage Fast;
public readonly ExponentialMovingAverage Slow;
public SelectionData()
{
Fast = new ExponentialMovingAverage(100);
Slow = new ExponentialMovingAverage(300);
}
public decimal ScaledDelta
{
get { return (Fast - Slow)/((Fast + Slow)/2m); }
}
public bool Update(DateTime time, decimal value)
{
return Fast.Update(time, value) && Slow.Update(time, value);
}
}
You need to use a SymbolData class instead of assigning the indicators to the Fundamental object because you can't create custom propertiesattributes on Fundamental objects like you can with Security objects.
Define the Universe
You need to define SymbolData objects before you define the universe that selects securities.
When your universe function receives an object that contains all the possible securities, create a SymbolData object for each new security and update the remaining SymbolData objects with their daily price or some other data point. For example, the following universe definition selects US Equities that have the greatest difference between two moving averages.
class EmaCrossUniverseSelectionAlgorithm(QCAlgorithm):
def initialize(self) -> None:
'''Initialise the data and resolution required, as well as the cash and start-end dates for your algorithm. All algorithms must initialized.'''
self.set_start_date(2010,1,1) #Set Start Date
self.set_end_date(2015,1,1) #Set End Date
self.set_cash(100000) #Set Strategy Cash
self.universe_settings.asynchronous = True
self.universe_settings.resolution = Resolution.DAILY
self.universe_settings.leverage = 2
self.count = 10
self.averages = { }
# this add universe method accepts two parameters:
# - fundamental selection function: accepts an IEnumerable<Fundamental> and returns an IEnumerable<Symbol>
self.add_universe(self.fundamental_selection_function)
# sort the data by daily dollar volume and take the top 'NumberOfSymbols'
def fundamental_selection_function(self, fundamental: List[Fundamental]) -> List[Symbol]:
# We are going to use a dictionary to refer the object that will keep the moving averages
for f in fundamental:
if f.symbol not in self.averages:
self.averages[f.symbol] = SymbolData(f.symbol)
# Updates the SymbolData object with current EOD price
avg = self.averages[f.symbol]
avg.update(f.end_time, f.adjusted_price)
# Filter the values of the dict: we only want up-trending securities
values = list(filter(lambda x: x.is_uptrend, self.averages.values()))
# Sorts the values of the dict: we want those with greater difference between the moving averages
values.sort(key=lambda x: x.scale, reverse=True)
for x in values[:self.count]:
self.log('symbol: ' + str(x.symbol.value) + ' scale: ' + str(x.scale))
# we need to return only the symbol objects
return [ x.symbol for x in values[:self.count] ] namespace QuantConnect.Algorithm.CSharp
{
public class EmaCrossUniverseSelectionAlgorithm : QCAlgorithm
{
// tolerance to prevent bouncing
const decimal Tolerance = 0.01m;
private const int Count = 10;
// use Buffer+Count to leave a little in cash
private const decimal TargetPercent = 0.1m;
private SecurityChanges _changes = SecurityChanges.None;
// holds our fundamental indicators by symbol
private readonly ConcurrentDictionary<Symbol, SelectionData> _averages = new ConcurrentDictionary<Symbol, SelectionData>();
public override void Initialize()
{
UniverseSettings.Asynchronous = true;
UniverseSettings.Leverage = 2.0m;
UniverseSettings.Resolution = Resolution.Daily;
SetStartDate(2010, 01, 01);
SetEndDate(2015, 01, 01);
SetCash(100*1000);
AddUniverse(fundamental =>
{
return (from f in fundamental
// grab th SelectionData instance for this symbol
let avg = _averages.GetOrAdd(f.Symbol, sym => new SelectionData())
// Update returns true when the indicators are ready, so don't accept until they are
where avg.Update(f.EndTime, f.AdjustedPrice)
// only pick symbols who have their 50 day ema over their 100 day ema
where avg.Fast > avg.Slow*(1 + Tolerance)
// prefer symbols with a larger delta by percentage between the two averages
orderby avg.ScaledDelta descending
// we only need to return the symbol and return 'Count' symbols
select f.Symbol).Take(Count);
});
}
}
}
Examples
The following examples demonstrate some common practices for using indicator universes.
Example 1: Universe Selection
The following algorithm selects the stocks within 5% of the 1-year maximum price among the top 100 liquid stocks. To do so, we make use of Maximum indicator to do so. Then, we hold the stocks with price > EMA > SMA, which indicates an upward accelerating trend.
public class ManualIndicatorAlgorithm : QCAlgorithm
{
private Dictionary<Symbol, Maximum> _maximumBySymbol = new();
private Dictionary<Symbol, SymbolData> _symbolData = new();
public override void Initialize()
{
SetStartDate(2021, 1, 1);
SetEndDate(2021, 2, 1);
// Select a popularity-based universe with indicators in a Selection function.
AddUniverse(Selection);
// Set a scheduled event to rebalance daily on the daily indicator signals.
Schedule.On(
DateRules.EveryDay(),
TimeRules.At(9, 31),
Rebalance
);
}
private IEnumerable<Symbol> Selection(IEnumerable<Fundamental> fundamentals)
{
var selected = new List<Symbol>();
// Initially filtered for the top 100 liquid stocks first.
var filtered = fundamentals.OrderByDescending(f => f.DollarVolume)
.Take(100)
.ToList();
foreach (var f in filtered)
{
if (!_maximumBySymbol.TryGetValue(f.Symbol, out var maximum))
{
maximum = new(252);
// Warm up the Maximum indicator for its readiness to use immediately.
var history = History<TradeBar>(f.Symbol, 252, Resolution.Daily);
foreach (var bar in history)
{
maximum.Update(bar.EndTime, bar.Close);
}
_maximumBySymbol[f.Symbol] = maximum;
}
else
{
// Update the indicator with the last known adjusted price daily.
maximum.Update(f.EndTime, f.AdjustedPrice);
}
// Select to trade if the current price is within 5% of the maximum price of the last year.
// Close to the maximum price provides evidence of high popularity for the fund to support the trend.
if (f.AdjustedPrice >= maximum * 0.95m)
{
selected.Add(f.Symbol);
}
}
return selected;
}
private void Rebalance()
{
var symbolsToBuy = new List<Symbol>();
foreach (var (symbol, symbolData) in _symbolData)
{
// Buy the stocks whose prices are above the EMA and the EMA is above the SMA, meaning their trend is upward accelerating.
if (symbolData.IsReady && Securities[symbol].Price > symbolData.Ema && symbolData.Ema > symbolData.Sma)
{
symbolsToBuy.Add(symbol);
}
}
// Equally invest in selected stocks to dissipate capital risk evenly.
var count = symbolsToBuy.Count;
if (count > 0)
{
var targets = symbolsToBuy.Select(symbol => new PortfolioTarget(symbol, 1m / count)).ToList();
// Liquidate the positions that are not on an upward trend or are not popular anymore.
SetHoldings(targets, liquidateExistingHoldings: true);
}
}
public override void OnSecuritiesChanged(SecurityChanges changes)
{
foreach (var added in changes.AddedSecurities)
{
// Instantiate a new instance of SymbolData object to hold indicators to use.
_symbolData[added.Symbol] = new SymbolData(this, added.Symbol);
}
foreach (var removed in changes.RemovedSecurities)
{
// Remove the data subscription of indicators to release computation resources when leaving the universe.
if (_symbolData.Remove(removed.Symbol, out var symbolData))
{
symbolData.Dispose();
}
}
}
private class SymbolData
{
private QCAlgorithm _algorithm;
public Symbol Symbol { get; set; }
public ExponentialMovingAverage Ema { get; set; }
public SimpleMovingAverage Sma { get; set; }
public bool IsReady => Ema.IsReady && Sma.IsReady;
public SymbolData(QCAlgorithm algorithm, Symbol symbol)
{
_algorithm = algorithm;
Symbol = symbol;
// Create an EMA and an SMA manual indicator for trade signal generation.
Ema = new ExponentialMovingAverage(20);
Sma = new SimpleMovingAverage(20);
// Warm up the indicators to ensure their readiness to use them immediately.
algorithm.WarmUpIndicator(symbol, Ema, Resolution.Daily);
algorithm.WarmUpIndicator(symbol, Sma, Resolution.Daily);
// Subscribe to the indicators to update daily price data for updated trade signal generation.
algorithm.RegisterIndicator(symbol, Ema, Resolution.Daily);
algorithm.RegisterIndicator(symbol, Sma, Resolution.Daily);
}
public void Dispose()
{
Ema.Reset();
Sma.Reset();
// Cancel data subscription to release computation resources.
_algorithm.DeregisterIndicator(Ema);
_algorithm.DeregisterIndicator(Sma);
}
}
} class ManualIndicatorAlgorithm(QCAlgorithm):
maximum_by_symbol = {}
def initialize(self) -> None:
self.set_start_date(2021, 1, 1)
self.set_end_date(2021, 2, 1)
# Select a popularity-based universe with indicators in a Selection function.
self._universe = self.add_universe(self.selection)
# Set a scheduled event to rebalance daily on the daily indicator signals.
self.schedule.on(
self.date_rules.every_day(),
self.time_rules.at(9, 31),
self.rebalance
)
def selection(self, fundamentals: List[Fundamental]) -> List[Symbol]:
selected = []
# Initially filtered for the top 100 liquid stocks first.
filtered = sorted(fundamentals, key=lambda f: f.dollar_volume, reverse=True)[:100]
for f in filtered:
if f.symbol not in self.maximum_by_symbol:
self.maximum_by_symbol[f.symbol] = Maximum(252)
# Warm up the Maximum indicator to its readiness to use immediately.
history = self.history[TradeBar](f.symbol, 252, Resolution.DAILY)
for bar in history:
self.maximum_by_symbol[f.symbol].update(bar.end_time, bar.close)
else:
# Update the indicator with the last known adjusted price daily.
self.maximum_by_symbol[f.symbol].update(f.end_time, f.adjusted_price)
# Select to trade if the current price is within 5% of the maximum price of the last year.
# Close to the maximum price provides evidence of high popularity for the fund to support the trend.
if f.adjusted_price >= self.maximum_by_symbol[f.symbol].current.value * 0.95:
selected.append(f.symbol)
return selected
def rebalance(self) -> None:
def to_buy(symbol):
security = self.securities[symbol]
# Buy the stocks whose prices are above the EMA and the EMA is above the SMA, meaning their trend is upward accelerating.
return security.price > security.ema.current.value > security.sma.current.value
symbols_to_buy = [symbol for symbol in self._universe.selected if to_buy(symbol)]
# Equally invest in the selected stocks to dissipate the capital risk evenly.
count = len(symbols_to_buy)
if count > 0:
targets = [PortfolioTarget(symbol, 1 / count) for symbol in symbols_to_buy]
# Liquidate the positions that are not on an upward trend or are not popular anymore.
self.set_holdings(targets, liquidate_existing_holdings=True)
def on_securities_changed(self, changes: SecurityChanges) -> None:
for added in changes.added_securities:
symbol = added.symbol
# Create an EMA and an SMA manual indicator for trade signal generation.
added.ema = ExponentialMovingAverage(20)
added.sma = SimpleMovingAverage(20)
# Warm up the indicators to ensure their readiness to use them immediately.
self.warm_up_indicator(symbol, added.ema, Resolution.DAILY)
self.warm_up_indicator(symbol, added.sma, Resolution.DAILY)
# Subscribe to the indicators to update daily price data for updated trade signal generation.
self.register_indicator(symbol, added.ema, Resolution.DAILY)
self.register_indicator(symbol, added.sma, Resolution.DAILY)
for removed in changes.removed_securities:
# Cancel data subscription to release computation resources when leaving the universe.
self.deregister_indicator(removed.ema)
self.deregister_indicator(removed.sma)
Other Examples
For more examples, see the following algorithms:
Demonstration Algorithms
EmaCrossUniverseSelectionAlgorithm.py Python
EmaCrossUniverseSelectionAlgorithm.cs C#
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:
