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

Applying Research

Mean Reversion

Introduction

This page explains how to you can use the Research Environment to develop and test a Mean Reversion hypothesis, then put the hypothesis in production.

Create Hypothesis

Imagine that we've developed the following hypothesis: stocks that are below 1 standard deviation of their 30-day-mean are due to revert and increase in value, statistically around 85% chance if we assume the return series is stationary and the price series is a Random Process. We've developed the following code in research to pick out such stocks from a preselected basket of stocks.

Import Libraries

We'll need to import libraries to help with data processing. Import numpy and scipy libraries by the following:

Select Language:
import numpy as np
from scipy.stats import norm, zscore

Get Historical Data

To begin, we retrieve historical data for researching.

  1. Instantiate a QuantBook.
  2. Select Language:
    qb = QuantBook()
  3. Select the desired tickers for research.
  4. Select Language:
    assets = ["SHY", "TLT", "SHV", "TLH", "EDV", "BIL",
              "SPTL", "TBT", "TMF", "TMV", "TBF", "VGSH", "VGIT",
              "VGLT", "SCHO", "SCHR", "SPTS", "GOVT"]
  5. Call the add_equity method with the tickers, and their corresponding resolution.
  6. Select Language:
    for i in range(len(assets)):
        qb.add_equity(assets[i],Resolution.MINUTE)

    If you do not pass a resolution argument, Resolution.MINUTE is used by default.

  7. Call the history method with qb.securities.keys for all tickers, time argument(s), and resolution to request historical data for the symbol.
  8. Select Language:
    history = qb.history(qb.securities.keys(), datetime(2021, 1, 1), datetime(2021, 12, 31), Resolution.DAILY)
    Historical data

Prepare Data

We'll have to process our data to get an extent of the signal on how much the stock is deviated from its norm for each ticker.

  1. Select the close column and then call the unstack method.
  2. Select Language:
    df = history['close'].unstack(level=0)
  3. Calculate the truth value of the most recent price being less than 1 standard deviation away from the mean price.
  4. Select Language:
    classifier = df.le(df.rolling(30).mean() - df.rolling(30).std())
  5. Get the z-score for the True values, then compute the expected return and probability (used for Insight magnitude and confidence).
  6. Select Language:
    z_score = df.apply(zscore)[classifier]
    magnitude = -z_score * df.rolling(30).std() / df.shift(1)
    confidence = (-z_score).apply(norm.cdf)
  7. Call fillna to fill NaNs with 0.
  8. Select Language:
    magnitude.fillna(0, inplace=True)
    confidence.fillna(0, inplace=True)
  9. Get our trading weight, we'd take a long only portfolio and normalized to total weight = 1.
  10. Select Language:
    weight = confidence - 1 / (magnitude + 1)
    weight = weight[weight > 0].fillna(0)
    sum_ = np.sum(weight, axis=1)
    for i in range(weight.shape[0]):
        if sum_[i] > 0:
            weight.iloc[i] = weight.iloc[i] / sum_[i]
        else:
            weight.iloc[i] = 0
    weight = weight.iloc[:-1]

Test Hypothesis

We would test the performance of this strategy. To do so, we would make use of the calculated weight for portfolio optimization.

  1. Get the total daily return series.
  2. Select Language:
    ret = pd.Series(index=range(df.shape[0] - 1))
    for i in range(df.shape[0] - 1):
        ret[i] = weight.iloc[i] @ df.pct_change().iloc[i + 1].T
  3. Call cumprod to get the cumulative return.
  4. total_ret = (ret + 1).cumprod()
  5. Set index for visualization.
  6. total_ret.index = weight.index
  7. Display the result.
  8. Select Language:
    total_ret.plot(title='Strategy Equity Curve', figsize=(15, 10))
    plt.show()
    Mean reversion equity line plot

Set Up Algorithm

Once we are confident in our hypothesis, we can export this code into backtesting. One way to accomodate this model into research is to create a scheduled event which uses our model to pick stocks and goes long.

Select Language:
def initialize(self) -> None:

    #1. Required: Five years of backtest history
    self.set_start_date(2014, 1, 1)

    #2. Required: Alpha Streams Models:
    self.set_brokerage_model(BrokerageName.ALPHA_STREAMS)

    #3. Required: Significant AUM Capacity
    self.set_cash(1000000)

    #4. Required: Benchmark to SPY
    self.set_benchmark("SPY")
    
    self.set_portfolio_construction(InsightWeightingPortfolioConstructionModel())
    self.set_execution(ImmediateExecutionModel())

    self.assets = ["SHY", "TLT", "IEI", "SHV", "TLH", "EDV", "BIL",
                    "SPTL", "TBT", "TMF", "TMV", "TBF", "VGSH", "VGIT",
                    "VGLT", "SCHO", "SCHR", "SPTS", "GOVT"]
    
    # Add Equity ------------------------------------------------ 
    for i in range(len(self.assets)):
        self.add_equity(self.assets[i], Resolution.MINUTE)
    
    # Set Scheduled Event Method For Our Model
    self.schedule.on(self.date_rules.every_day(), self.time_rules.before_market_close("SHY", 5), self.every_day_before_market_close)

Now we export our model into the scheduled event method. We will switch qb with self and replace methods with their QCAlgorithm counterparts as needed. In this example, this is not an issue because all the methods we used in research also exist in QCAlgorithm.

Select Language:
def every_day_before_market_close(self) -> None:
    qb = self
    # Fetch history on our universe
    df = qb.history(list(qb.securities.keys()), 30, Resolution.DAILY)
    if df.empty: return

    # Make all of them into a single time index.
    df = df.close.unstack(level=0)

    # Calculate the truth value of the most recent price being less than 1 std away from the mean
    classifier = df.le(df.mean().subtract(df.std())).iloc[-1]
    if not classifier.any(): return

    # Get the z-score for the True values, then compute the expected return and probability
    z_score = df.apply(zscore)[[classifier.index[i] for i in range(classifier.size) if classifier.iloc[i]]]

    magnitude = -z_score * df.std() / df
    confidence = (-z_score).apply(norm.cdf)

    # Get the latest values
    magnitude = magnitude.iloc[-1].fillna(0)
    confidence = confidence.iloc[-1].fillna(0)

    # Get the weights, then zip together to iterate over later
    weight = confidence - 1 / (magnitude + 1)
    weight = weight[weight > 0].fillna(0)
    sum_ = np.sum(weight)
    if sum_ > 0:
        weight = (weight) / sum_
        selected = zip(weight.index, magnitude, confidence, weight)
    else:
        return

    # ==============================
    
    insights = []
    
    for symbol, magnitude, confidence, weight in selected:
        insights.append( Insight.price(symbol, timedelta(days=1), InsightDirection.UP, magnitude, confidence, None, weight) )

    self.emit_insights(insights)

Examples

The below code snippets concludes the above jupyter research notebook content.

Select Language:
from scipy.stats import norm, zscore

# Instantiate a QuantBook.
qb = QuantBook()

# Select the desired tickers for research.
symbols = {}
assets = ["SHY", "TLT", "SHV", "TLH", "EDV", "BIL",
          "SPTL", "TBT", "TMF", "TMV", "TBF", "VGSH", "VGIT",
          "VGLT", "SCHO", "SCHR", "SPTS", "GOVT"]

# Call the AddEquity method with the tickers, and its corresponding resolution. Then store their Symbols. Resolution.Minute is used by default. 
for i in range(len(assets)):
    symbols[assets[i]] = qb.add_equity(assets[i], Resolution.MINUTE).symbol

# Call the History method with qb.Securities.Keys for all tickers, time argument(s), and resolution to request historical data for the symbol.
history = qb.history(qb.securities.Keys, datetime(2021, 1, 1), datetime(2021, 12, 31), Resolution.DAILY)

# Select the close column and then call the unstack method.
df = history['close'].unstack(level=0)

# Calculate the truth value of the most recent price being less than 1 standard deviation away from the mean price.
classifier = df.le(df.rolling(30).mean() - df.rolling(30).std())

# Get the z-score for the True values, then compute the expected return and probability (used for Insight magnitude and confidence).
z_score = df.apply(zscore)[classifier]
magnitude = -z_score * df.rolling(30).std() / df
confidence = (-z_score).apply(norm.cdf)

# Call fillna to fill NaNs with 0
magnitude.fillna(0, inplace=True)
confidence.fillna(0, inplace=True)

# Get our trading weight, we'd take a long only portfolio and normalized to total weight = 1
weight = confidence - 1 / (magnitude + 1)
weight = weight[weight > 0].fillna(0)
sum_ = np.sum(weight, axis=1)
for i in range(weight.shape[0]):
    if sum_[i] > 0:
        weight.iloc[i] = weight.iloc[i] / sum_[i]
    else:
        weight.iloc[i] = 0
weight = weight.iloc[:-1]

# Get the total daily return series
ret = pd.Series(index=range(df.shape[0] - 1))
for i in range(df.shape[0] - 1):
    ret[i] = weight.iloc[i] @ df.pct_change().iloc[i + 1].T
    
# Call cumprod to get the cumulative return
total_ret = (ret + 1).cumprod()

# Set index for visualization
total_ret.index = weight.index

# Plot the result
total_ret.plot(title='Strategy Equity Curve', figsize=(15, 10))
plt.show()

The below code snippets concludes the algorithm set up.

Select Language:
from scipy.stats import norm, zscore

class MeanReversionDemo(QCAlgorithm):

    def initialize(self) -> None:
        #1. Required: Five years of backtest history
        self.set_start_date(2014, 1, 1)
    
        #2. Required: Alpha Streams Models:
        self.set_brokerage_model(BrokerageName.ALPHA_STREAMS)
    
        #3. Required: Significant AUM Capacity
        self.set_cash(1000000)
    
        #4. Required: Benchmark to SPY
        self.set_benchmark("SPY")
        
        self.set_portfolio_construction(InsightWeightingPortfolioConstructionModel())
        self.set_execution(ImmediateExecutionModel())
    
        self.assets = ["SHY", "TLT", "IEI", "SHV", "TLH", "EDV", "BIL",
                        "SPTL", "TBT", "TMF", "TMV", "TBF", "VGSH", "VGIT",
                        "VGLT", "SCHO", "SCHR", "SPTS", "GOVT"]
        
        # Add Equity ------------------------------------------------ 
        for i in range(len(self.assets)):
            self.add_equity(self.assets[i], Resolution.MINUTE).symbol
        
        # Set Scheduled Event Method For Our Model
        self.schedule.on(self.date_rules.every_day(), self.time_rules.before_market_close("SHY", 5), self.every_day_before_market_close)
        
        
    def every_day_before_market_close(self) -> None:
        qb = self
        # Fetch history on our universe
        df = qb.history(list(qb.securities.keys()), 30, Resolution.DAILY)
        if df.empty: return
    
        # Make all of them into a single time index.
        df = df.close.unstack(level=0)
    
        # Calculate the truth value of the most recent price being less than 1 std away from the mean
        classifier = df.le(df.mean().subtract(df.std())).iloc[-1]
        if not classifier.any(): return
    
        # Get the z-score for the True values, then compute the expected return and probability
        z_score = df.apply(zscore)[[classifier.index[i] for i in range(classifier.size) if classifier.iloc[i]]]
    
        magnitude = -z_score * df.std() / df
        confidence = (-z_score).apply(norm.cdf)
    
        # Get the latest values
        magnitude = magnitude.iloc[-1].fillna(0)
        confidence = confidence.iloc[-1].fillna(0)
    
        # Get the weights, then zip together to iterate over later
        weight = confidence - 1 / (magnitude + 1)
        weight = weight[weight > 0].fillna(0)
        sum_ = np.sum(weight)
        if sum_ > 0:
            weight = (weight) / sum_
            selected = zip(weight.index, magnitude, confidence, weight)
        else:
            return
    
        # ==============================
        
        insights = []
        
        for symbol, magnitude, confidence, weight in selected:
            insights.append( Insight.price(symbol, timedelta(days=1), InsightDirection.UP, magnitude, confidence, None, weight) )
    
        self.emit_insights(insights)

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: