Initializing a Pipeline

The first step of any algorithm using the Pipeline API is to create and register an empty Pipeline object.

from quantopian.algorithm import attach_pipeline, pipeline_output
from quantopian.pipeline import Pipeline
...
def initialize(context):
    pipe = Pipeline()
    attach_pipeline(pipe, name='my_pipeline')

A Pipeline is an object that represents computation we would like to perform every day. A freshly-constructed pipeline is empty, which means it doesn’t yet know how to compute anything, and it won’t produce any values if we ask for its outputs. We’ll see below how to provide our Pipeline with expressions to compute.

Just constructing a new pipeline doesn’t do anything by itself: we have to tell our algorithm to use the new pipeline, and we have to provide a name for the pipeline so that we can identify it later when we ask for results. The attach_pipeline() function from quantopian.algorithm accomplishes both of these tasks.

Importing Datasets

Before we can build computations for our pipeline to execute, we need a way to identify the inputs to those computations. In our example, we’ll use data from the USEquityPricing dataset, which we import from quantopian.pipeline.data.builtin.

from quantopian.pipeline.data.builtin import USEquityPricing

USEquityPricing is an example of a DataSet. The most important thing to understand about DataSets is that they do not hold actual data. DataSets are simply collections of objects that tell the Pipeline API where and how to find the inputs to computations. Since these objects often correspond to database columns, we refer to the attributes of DataSets as columns.

USEquityPricing provides five columns:

• USEquityPricing.open
• USEquityPricing.high
• USEquityPricing.low
• USEquityPricing.close
• USEquityPricing.volume

Many datasets besides USEquityPricing are available on Quantopian. These include corporate fundamental data, news sentiment, macroeconomic indicators, and more. Morningstar fundamental data is available in the quantopian.pipeline.data.Fundamentals dataset. All other datasets are namespaced by provider under quantopian.pipeline.data.

Pipeline-compatible datasets are currently importable from the following modules:

• quantopian.pipeline.data.estimize (Estimize)
• quantopian.pipeline.data.eventvestor (EventVestor)
• quantopian.pipeline.data.psychsignal (PsychSignal)
• quantopian.pipeline.data.quandl (Quandl)
• quantopian.pipeline.data.sentdex (Sentdex)

Example algorithms and notebooks for working with each of these datasets can be found on the Quantopian Data page.

Building Computations

Once we’ve imported the datasets we intend to use in our algorithm, our next step is to build the transformations we want our Pipeline to compute each day.

sma_10 = SimpleMovingAverage(inputs=[USEquityPricing.close], window_length=10)
sma_30 = SimpleMovingAverage(inputs=[USEquityPricing.close], window_length=30)

The SimpleMovingAverage class used here is an example of a Factor. Factors, in the Pipeline API, are objects that represent reductions on trailing windows of data. Every Factor stores four pieces of state:

1. inputs: A list of BoundColumn objects describing the inputs to the Factor.
2. window_length : An integer describing how many rows of historical data the Factor needs to be provided each day.
3. dtype: A numpy dtype object representing the type of values computed by the Factor. Most factors are of dtype float64, indicating that they produce numerical values represented as 64-bit floats. Factors can also be of dtype datetime64[ns].
4. A compute function that operates on the data described by inputs and window_length.

Example Factor: Volume-Weighted Average Price

When we compute a Factor for a day on which we have N assets in our database, the underlying Pipeline API engine provides that Factor’s compute function a two-dimensional array of shape (window_length x N) for each input in inputs. The job of the compute function is to produce a one-dimensional array of length N as an output.

The figure to the right shows the computation performed on a single day by another built-in Factor, VWAP.

prices_under_5 = (sma_10 < 5)

The next line of our example algorithm constructs a Filter. Like Factors, Filters are reductions over input data defined by datasets. The difference between Filters and Factors is that Filters produce boolean-valued outputs, whereas Factors produce numerical- or datetime-valued outputs. The expression (sma_10 < 5) uses Operator Overloading to construct a Filter instance whose compute is equivalent to “compute 10-day moving average price for each asset, then return an array containing True for all assets whose computed value was less than 5, otherwise containing False.

Quantopian provides a library of factors (SimpleMovingAverage, RSI, VWAP and MaxDrawdown) which will continue to grow. You can also create custom factors. See the examples, and API Documentation for more information.

Adding Computations to a Pipeline

Pipelines support two broad classes of operations: adding new columns and screening out unwanted rows.

To add a new column, we call the add() method of our pipeline with a Filter or Factor and a name for the column to be created.

pipe.add(sma_10, name='sma_10')
pipe.add(sma_30, name='sma_30')

These calls to add tell our pipeline to compute a 10-day SMA with the name "sma_10" and our 30-day SMA with the name "sma_30".

The other important operation supported by pipelines is screening out unwanted rows our final results.

pipe.set_screen(prices_under_5)

The call to set_screen informs our pipeline that each day we want to throw away any rows for whose assets our prices_under_5 Filter produced a value of False.

Using Results

By the end of our initialize function, we’ve already defined the core logic of our algorithm. All that remains to be done is to ask for the results of the pipeline attached to our algorithm.

def before_trading_start(context, data):
    results = pipeline_output('my_pipeline')
    print results.head(5)

The return value of pipeline_output will be a pandas.DataFrame containing a column for each call we made to Pipeline.add() in our initialize method and containing a row for each asset that our pipeline’s screen. The printed values should look like this:

Most algorithms save their pipeline Pipeline outputs on context for use in functions other than before_trading_start. For example, an algorithm might compute a set of target portfolio weights as a Factor, store the target weights on context, and then use the stored weights in a rebalance function scheduled to run once per day.

context.pipeline_results = results

Custom Factors

Quantopian provides a large library of built-in factors in the quantopian.pipeline.factors module.

One of the most powerful features of the Pipeline API is that it allows users to define their own custom factors. The easiest way to do this is to subclass quantopian.pipeline.CustomFactor and implement a compute method whose signature is:

def compute(self, today, assets, out, *inputs):
    ...

An instance of CustomFactor that’s been added to a pipeline will have its compute method called every day with data defined by the values supplied to its constructor as inputs and window_length.

For example, if we define and add a CustomFactor like this:

class MyFactor(CustomFactor):
    def compute(self, today, asset_ids, out, values):
       out[:] = do_something_with_values(values)

def initialize(context):
    p = attach_pipeline(Pipeline(), 'my_pipeline')
    my_factor = MyFactor(inputs=[USEquityPricing.close], window_length=5)
    p.add(my_factor, 'my_factor')

then every simulation day my_factor.compute() will be called with data as follows:

• values will all be 5 x N numpy array, where N is roughly number of assets in our database on the day in question (generally between 8000-10000). Each column of values will contain the last 5 close prices of one asset. There will be 5 rows in the array because we passed window_length=5 to the MyFactor constructor.
• out will be an empty array of length N. The job of compute is to write output values into out. We write values directly into an output array rather than returning them to avoid making unnecessary copies. This is an uncommon idiom in Python, but very common in languages like C and Fortran that care about high performance numerical code. Many numpy functions take an out parameter for similar reasons.
• asset_ids will be an integer array of length N containing security ids corresponding to each column of values.
• today will be a datetime64[ns] representing the day for which compute is being called.

Default Inputs

Some factors are naturally parametrizable on both inputs and window_length. It makes reasonable sense, for example, to take a SimpleMovingAverage of more than one length or over more than one input. For this reason, factors like SimpleMovingAverage don’t provide defaults for inputs or window_length. Every time you construct an instance of SimpleMovingAverage, you have to pass a window_length and an inputs list.

Many factors, however, are naturally defined as operating on a specific dataset or on a specific window size. VWAP should always be called with inputs=[USEquityPricing.close, USEquityPricing.volume], and RSI is canonically calculated over a 14-day window. For factors, it’s convenient to be able to provide a default. If values are not passed for window_length or inputs, the CustomFactor constructor will try to fall back to class-level attributes with the same names. This means that we can implement a VWAP-like CustomFactor by defining an inputs list as a class level attribute.

import numpy as np

class PriceRange(CustomFactor):
    """
    Computes the difference between the highest high and the lowest
    low of each over an arbitrary input range.
    """
    inputs = [USEquityPricing.high, USEquityPricing.low]

    def compute(self, today, assets, out, highs, lows):
        out[:] = np.nanmax(highs, axis=0) - np.nanmin(lows, axis=0)

...
# We'll automatically use high and low as inputs because we declared them as
# defaults for the class.
price_range_10 = PriceRange(window_length=10)

Fundamental Data

Quantopian provides a number of Pipeline-compatible fundamental data fields sourced from Morningstar. Each field exists as a BoundColumn under the quantopian.pipeline.data.Fundamentals dataset. These fields can be passed as inputs to Pipeline computations.

For example, Fundamentals.market_cap is a column representing the most recently reported market cap for each asset on each date. There are over 900 total columns available in the Fundamentals dataset. See the Quantopian Fundamentals Reference for a full description of all such attributes.

Masking Factors

When computing a CustomFactor we are sometimes only interested in computing over a certain set of stocks, especially if our CustomFactor’s compute method is computationally expensive. We can restrict the set of stocks over which we would like to compute by passing a Filter to our CustomFactor upon instantiation via the mask parameter. When passed a mask, a CustomFactor will only compute values over stocks for which the Filter returns True. All other stocks for which the Filter returned False will be filled with missing values.

Example:

Suppose we want to compute a factor over only the top 500 stocks by dollar volume. We can define a CustomFactor and Filter as follows:

from quantopian.algorithm import attach_pipeline
from quantopian.pipeline import CustomFactor, Pipeline
from quantopian.pipeline.data.builtin import USEquityPricing
from quantopian.pipeline.factors import AverageDollarVolume

def do_something_expensive(some_input):
    # Not actually expensive. This is just an example.
    return 42

class MyFactor(CustomFactor):
    inputs = [USEquityPricing.close]
    window_length = 252

    def compute(self, today, assets, out, close):
        out[:] = do_something_expensive(close)

def initialize(context):
    pipeline = attach_pipeline(Pipeline(), 'my_pipeline')

    dollar_volume = AverageDollarVolume(window_length=30)
    high_dollar_volume = dollar_volume.top(500)

    # Pass our filter to our custom factor via the mask parameter.
    my_factor = MyFactor(mask=high_dollar_volume)
    pipeline.add(my_factor, 'my_factor')

Slicing Factors

Using a technique we call slicing, it is possible to extract the values of a Factor for a single asset. Slicing can be thought of as extracting a single column of a Factor, where each column corresponds to an asset. Slices are created by indexing into an ordinary factor, keyed by asset. These Slice objects can then be used as an input to a CustomFactor.

When a Slice object is used as an input to a custom factor, it always returns an N x 1 column vector of values, where N is the window length. For example, a slice of a Returns factor would output a column vector of the N previous returns values for a given security.

Example:

Create a Returns factor and extract the column for AAPL.

from quantopian.pipeline import CustomFactor
from quantopian.pipeline.factors import Returns

returns = Returns(window_length=30)
returns_aapl = returns[sid(24)]

class MyFactor(CustomFactor):
    inputs = [returns_aapl]
    window_length = 5

    def compute(self, today, assets, out, returns_aapl):
        # `returns_aapl` is a 5 x 1 numpy array of the last 5 days of
        # returns values for AAPL. For example, it might look like:
        # [[ .01],
        #  [ .02],
        #  [-.01],
        #  [ .03],
        #  [ .01]]
        pass

Note: Only slices of certain factors can be used as inputs. These factors include Returns and any factors created from rank or zscore. The reason for this is that these factors produce normalized values, so they are safe for use as inputs to other factors.

Note: Slice objects cannot be added as a column to a pipeline. Each day, a slice only computes a value for the single asset with which it is associated, whereas ordinary factors compute a value for every asset. In the current pipeline framework there would be no reasonable way to represent the single-valued output of a Slice term.

Normalizing Results

It is often desirable to normalize the output of a factor before using it in an algorithm. In this context, normalizing means re-scaling a set of values in a way that makes comparisons between values more meaningful.

Such re-scaling can be useful for comparing the results of different factors. A technical indicator like RSI might produce an output bounded between 1 and 100, whereas a fundamental ratio might produce a values of any real number. Normalizing two incomensurable factors via an operation like Z-Score makes it easier to incorporate both factors into a larger model, because doing so reduces both factors to dimensionless quantities.

The base Factor class provides several methods for normalizing factor outputs. The demean() method transforms the output of a factor by computing the mean of each row and subtracting it from every entry in the row. Similarly, the zscore() method subtracts the mean from each row and then divides by the standard deviation of the row.

Example:

Suppose that f is a Factor which would produce the following output:

             AAPL   MSFT    MCD     BK
2017-03-13    1.0    2.0    3.0    4.0
2017-03-14    1.5    2.5    3.5    1.0
2017-03-15    2.0    3.0    4.0    1.5
2017-03-16    2.5    3.5    1.0    2.0

f.demean() constructs a new Factor that will produce its output by first computing f and then subtracting the mean from each row:

             AAPL   MSFT    MCD     BK
2017-03-13 -1.500 -0.500  0.500  1.500
2017-03-14 -0.625  0.375  1.375 -1.125
2017-03-15 -0.625  0.375  1.375 -1.125
2017-03-16  0.250  1.250 -1.250 -0.250

f.zscore() works similarly, except that it also divides each row by the standard deviation of the row.

             AAPL   MSFT    MCD     BK
2017-03-13   99.0    2.0    3.0    4.0
2017-03-14    1.5   99.0    3.5    1.0
2017-03-15    2.0    3.0   99.0    1.5
2017-03-16    2.5    3.5    1.0   99.0

>>> f = MyFactor(...)
>>> non_outliers = f.percentile_between(0, 75)

             AAPL   MSFT    MCD     BK
2017-03-13  False   True   True   True
2017-03-14   True  False   True   True
2017-03-15   True   True  False   True
2017-03-16   True   True   True  False

>>> normalized = f.demean(mask=non_outliers)

             AAPL   MSFT    MCD     BK
2017-03-13    NaN -1.000  0.000  1.000
2017-03-14 -0.500    NaN  1.500 -1.000
2017-03-15 -0.166  0.833    NaN -0.666
2017-03-16  0.166  1.166 -1.333    NaN

             AAPL   MSFT    MCD     BK
2017-03-13    1.0    2.0    3.0    4.0
2017-03-14    1.5    2.5    3.5    1.0
2017-03-15    2.0    3.0    4.0    1.5
2017-03-16    2.5    3.5    1.0    2.0

             AAPL   MSFT    MCD     BK
2017-03-13      1      1      2      2
2017-03-14      1      1      2      2
2017-03-15      1      1      2      2
2017-03-16      1      1      2      2

>>> f.demean(groupby=c)

             AAPL   MSFT    MCD     BK
2017-03-13 -0.500  0.500 -0.500  0.500
2017-03-14 -0.500  0.500  1.250 -1.250
2017-03-15 -0.500  0.500  1.250 -1.250
2017-03-16 -0.500  0.500 -0.500  0.500

Working with Dates

Most of the data available in the Pipeline API is numerically-valued. There are, however, some pipeline datasets that contain non-numeric data.

The most common class of non-numeric data available in the Pipeline API is data representing dates and times of events. An example of such a dataset is the EventVestor EarningsCalendar, which provides two columns:

• EarningsCalendar.next_announcement
• EarningsCalendar.previous_announcement

When supplied as inputs to a CustomFactor, these columns provide numpy arrays of dtype numpy.datetime64 representing, for each pair of asset and simulation date, the dates of the next and previous earnings announcements for the asset as of the simulation date.

For example, Microsoft’s earnings announcement calendar for 2014 was as follows:

If we specify EarningsCalendar.next_announcement as an input to a CustomFactor with a window_length of 1, we’ll be passed the following data for MSFT in our compute function:

• From 2014-01-24 to 2014-04-24, the next announcement for MSFT will be 2014-04-24.
• From 2014-04-25 to 2014-07-22, the next announcement will be 2014-07-22.
• From 2014-07-23 to 2014-10-23, the next announcement will be 2014-10-23.

If a company had not yet announced its next earnings on a given simulation date, next_announcement will contain a value of numpy.NaT which stands for “Not a Time”. Note that unlike the special floating-point value NaN, (“Not a Number”), NaT compares equal to itself.

The previous_announcement column behaves similarly, except that it fills backwards rather than forwards, producing the date of the most-recent historical announcement as of any given date.

Working with Strings

The Pipeline API also supports string-based expressions. There are two major uses for string data in pipeline:

1. String-typed Classifiers can be used to perform grouped normalizations via the same methods (demean(), zscore(), etc.) available on integer-typed classifiers.

   For example, we can build a Factor that Z-Scores equity returns by country of business by doing:

   from quantopian.pipeline.data import morningstar as mstar
   from quantopian.pipeline.factors import Returns
   
   country_id = mstar.company_reference.country_id.latest
   daily_returns = Returns(window_length=2)
   zscored = daily_returns.zscore(groupby=country_id)
   

2. String expressions can be converted into Filters via string matching methods like startswith(), endswith(), has_substring(), and matches().

   For example, we can build a Filter that accepts only stocks whose exchange name starts with 'OTC' by doing:

   from quantopian.pipeline.data import morningstar as mstar
   
   exchange_id = mstar.share_class_reference.exchange_id.latest
   is_over_the_counter = exchange_id.startswith('OTC')