Back to Community
Behavioral Arbitrage Webinar Notebook + Backtest

Finally, I managed to get around to cleaning up this notebook to readable standards. I have also taken the liberty to add the backtest results to combine both the pre and post earnings strategies together.

Clone Algorithm
380
Loading...
Backtest from to with initial capital
Total Returns
--
Alpha
--
Beta
--
Sharpe
--
Sortino
--
Max Drawdown
--
Benchmark Returns
--
Volatility
--
Returns 1 Month 3 Month 6 Month 12 Month
Alpha 1 Month 3 Month 6 Month 12 Month
Beta 1 Month 3 Month 6 Month 12 Month
Sharpe 1 Month 3 Month 6 Month 12 Month
Sortino 1 Month 3 Month 6 Month 12 Month
Volatility 1 Month 3 Month 6 Month 12 Month
Max Drawdown 1 Month 3 Month 6 Month 12 Month
import numpy as np
import pandas as pd
from quantopian.algorithm import attach_pipeline, pipeline_output
from quantopian.pipeline import Pipeline
from quantopian.pipeline.data import Fundamentals
from quantopian.pipeline.data.builtin import USEquityPricing
from quantopian.pipeline.factors import Returns, AverageDollarVolume
from quantopian.pipeline.filters import QTradableStocksUS
from quantopian.pipeline.factors.eventvestor import BusinessDaysUntilNextEarnings, BusinessDaysSincePreviousEarnings
import quantopian.optimize as opt
from quantopian.pipeline.data.zacks import EarningsSurprises

    
def initialize(context):
    context.MAX_LEV = 0.5
    context.MAX_IN_ONE = .1
    
    context.longs = [[]] * 5
    context.shorts = [[]] * 5
    
    attach_pipeline(pre_earnings_pipeline(context), 'pre_pipeline')
    attach_pipeline(post_earnings_pipeline(context), 'post_pipeline')
    
    # < No Slippage Assumed >
    set_slippage(slippage.FixedBasisPointsSlippage(0))
    schedule_function(rebalance, date_rules.every_day(), time_rules.market_open())
    schedule_function(record_vars, date_rules.every_day(), time_rules.market_close())
    
    
def pre_earnings_pipeline(context):
    universe = QTradableStocksUS()
    point_in_time = BusinessDaysUntilNextEarnings().eq(1)
    
    factor = (EarningsSurprises.eps_pct_diff_surp.latest.rank() + Returns(window_length=5).rank())
    filter_factor = EarningsSurprises.eps_std_dev_est.latest
    
    universe = universe & factor.notnan() & filter_factor.notnan()
    universe = filter_factor.rank(mask=universe).percentile_between(70, 100)
    
    universe = universe & point_in_time
    
    longs = factor.zscore(mask=universe).percentile_between(0, 25)
    shorts = factor.zscore(mask=universe).percentile_between(75, 100)

    return Pipeline(
        columns={
            'longs': longs,
            'shorts': shorts,
        },

        screen=universe
    )


def post_earnings_pipeline(context):
    universe = QTradableStocksUS()
    point_in_time = BusinessDaysSincePreviousEarnings().eq(1)
        
    factor = Fundamentals.pe_ratio.latest
    filter_factor = (EarningsSurprises.eps_amt_diff_surp.latest**2)**0.5
    
    universe = universe & factor.notnan() & filter_factor.notnan()
    universe = filter_factor.rank(mask=universe).percentile_between(0, 30)
    
    universe = universe & point_in_time
    
    longs = factor.zscore(mask=universe).percentile_between(0, 25)
    shorts = factor.zscore(mask=universe).percentile_between(75, 100)
    
    return Pipeline(
        columns={
            'longs': longs,
            'shorts': shorts,
        },
        screen=universe
)


def before_trading_start(context, data):
    context.pre_pipeline = pipeline_output('pre_pipeline')
    context.post_pipeline = pipeline_output('post_pipeline')

    def update_record(record, new_item, days_to_hold):
        record.insert(0, new_item)
        while len(record) > days_to_hold and len(record[-1]) == 0:
            del(record[-1])
        if sum(map(lambda l: 0 if len(l) == 0 else 1, record)) > days_to_hold:
            del(record[-1])
    
    update_record(context.longs, context.pre_pipeline.index[
        (context.pre_pipeline['longs'] == True)
    ], 2)
    update_record(context.shorts, context.pre_pipeline.index[
        (context.pre_pipeline['shorts'] == True)
    ], 2)
    update_record(context.longs, context.post_pipeline.index[
        (context.post_pipeline['longs'] == True)
    ], 5)
    update_record(context.shorts, context.post_pipeline.index[
        (context.post_pipeline['shorts'] == True)
    ], 5)
        
        
def rebalance(context, data):
    long_list = [equity for sublist in context.longs for equity in sublist]
    short_list = [equity for sublist in context.shorts for equity in sublist]
    
    for equity in long_list:
        if data.can_trade(equity):
            order_target_percent(equity, min(context.MAX_LEV / len(long_list), context.MAX_IN_ONE))
    for equity in short_list:
        if data.can_trade(equity):
            order_target_percent(equity, -min(context.MAX_LEV / len(short_list), context.MAX_IN_ONE))
    
    for position in context.portfolio.positions:
        if position not in long_list and position not in short_list:
            order_target_percent(position, 0)
            
            
def record_vars(context, data):
    record(leverage=context.account.leverage, pos=len(context.portfolio.positions))
There was a runtime error.
8 responses

And the notebook.

Loading notebook preview...
Notebook previews are currently unavailable.

Wow!! So awesome! Grateful for your generosity! Thank you!

I can't get this to work - Do I need a subscription to run a backtest?

Only if you want to test recent time periods (last 2 years or so). I think you can still get it to work by changing the end date to before Aug 2016

For context, Cheng presented a webinar with us a few weeks back. The recording can be found on YouTube on the Quantopian channel along with lots of other new video content we've been producing this year.

Disclaimer

The material on this website is provided for informational purposes only and does not constitute an offer to sell, a solicitation to buy, or a recommendation or endorsement for any security or strategy, nor does it constitute an offer to provide investment advisory services by Quantopian. In addition, the material offers no opinion with respect to the suitability of any security or specific investment. No information contained herein should be regarded as a suggestion to engage in or refrain from any investment-related course of action as none of Quantopian nor any of its affiliates is undertaking to provide investment advice, act as an adviser to any plan or entity subject to the Employee Retirement Income Security Act of 1974, as amended, individual retirement account or individual retirement annuity, or give advice in a fiduciary capacity with respect to the materials presented herein. If you are an individual retirement or other investor, contact your financial advisor or other fiduciary unrelated to Quantopian about whether any given investment idea, strategy, product or service described herein may be appropriate for your circumstances. All investments involve risk, including loss of principal. Quantopian makes no guarantees as to the accuracy or completeness of the views expressed in the website. The views are subject to change, and may have become unreliable for various reasons, including changes in market conditions or economic circumstances.

Hi Cheng -

Just curious - why did you decide to pay for the data?

@Grant
I wanted to have the OOS test since I initially developed the strategy on free data portion. My meager contest earnings are more or less paying for it as well.

Hi Cheng Peng,

I'm a huge fan of Behavioral Finance and have implemented off Q platform the Herding Effect / Follow the Leader logic in an agent based multi-strategy, swarm optimization, winner takes all and/or ensemble framework. Your Behaviorial Arbitrage logic based on pre and post earnings is spot on.

Could a variant of this logic under Quantopian L/S market neutral framework using Optimize API with constraints be the implementation of your winning contest entry? If so, kudos to you! Then also there is something to be said about using premium alternative data and Behaviorial Finance.