Back to Community
Fundamental history based algo

Hello,
I would like to run an algorithm based on the change in the earning per share ratio according to the last quarter.
For example : if actual earning per share > last earning per share , we buy that stock.
Is it possible to get the history of the earning per share ?

Thanks

11 responses

You can do that in Research, and I heard they were planning on adding historical fundamentals to the backtester/live trader, but I don't think it's happened yet.

Hello,

Here is an example algorithm that compares current earnings per share to last earnings per share. With the current implementation, the solution is a bit of a hack, but as we improve pipeline, it will become easier to do this. For now, this implementation should get you what you want -- especially the EPSChange custom factor.

Please let me know if you have any questions.

Clone Algorithm
98
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
from quantopian.algorithm import attach_pipeline, pipeline_output
from quantopian.pipeline import Pipeline
from quantopian.pipeline import CustomFactor
from quantopian.pipeline.data.builtin import USEquityPricing
from quantopian.pipeline.data import morningstar
from quantopian.pipeline.factors import SimpleMovingAverage
import numpy as np

# Create custom factor subclass to calculate change in EPS
class EPSChange(CustomFactor):

    # Use basic_eps from the morningstar fundamentals data
    inputs = [morningstar.earnings_report.basic_eps]
    # Set the window length to be big enough to capture the end of the last quarter
    window_length = 150

    # Compute EPS change since end of last quarter
    def compute(self, today, assets, out, eps):
        out_list = []
        
        # For each security in the universe
        for i in range(eps.shape[1]):
            
            # Store the column of each day's eps for the security in eps_col
            eps_col = eps[:,i]
            
            # Since the basic_eps only changes at the end of each quarter, np.unique
            # will get us the unique EPS values in the 150-day window. return_index
            # is set to True so that the most recent unique values can be referenced later.
            _, idx = np.unique(eps_col, return_index=True)
            
            # Get the 2 most recent unique values in the 150-day value.
            prev_and_curr_eps = eps_col[np.sort(idx)[-2:]]
            
            # If there was only 1 distinct EPS value in the last 150-days,
            # duplicate it so that numpy can create a 2*num_stocks to output
            # in out[:]
            if len(prev_and_curr_eps) == 1:
                prev_and_curr_eps = np.append(prev_and_curr_eps, prev_and_curr_eps[0])
            
            # Store the 2 most recent unique EPS values for the stock in out_list
            out_list.append(prev_and_curr_eps)
            
        # Convert out_list to a 2*num_stocks numpy array
        all_prev_and_curr_eps = np.transpose(np.array(out_list))
        
        # Return the most recent EPS minus the previous EPS
        out[:] = all_prev_and_curr_eps[-1] - all_prev_and_curr_eps[-2]


def initialize(context):

    # Create the pipe
    pipe = Pipeline()
    attach_pipeline(pipe, 'eps-change-example')

    # Construct the custom factor
    eps_change = EPSChange()

    # Add the EPS change factor to the pipe
    pipe.add(eps_change, 'eps_change')
    
    # Create and apply a filter representing all stocks where the current EPS is
    # greater than the previous one
    eps_better_top_100 = eps_change.top(100)
    pipe.set_screen(eps_better_top_100)

def before_trading_start(context, data):
    context.output = pipeline_output('eps-change-example').sort(['eps_change'], ascending=False)

    update_universe(context.output.index)

def handle_data(context, data):

    print "SECURITY LIST"
    log.info("\n" + str(context.output))
There was a runtime error.
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.

Oh duhh, of course, could use pipeline for this :)

Hello,
Thanks for your reply.
@ Jamie I tried using pipeline as you suggested but the backtest doesn't work, maybe because of long computation time in factor EPSchange.
Some one has an idea to deal with this issue ?

Thanks
idriss

Clone Algorithm
4
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
from quantopian.algorithm import attach_pipeline, pipeline_output
from quantopian.pipeline import Pipeline
from quantopian.pipeline import CustomFactor
from quantopian.pipeline.data.builtin import USEquityPricing
from quantopian.pipeline.data import morningstar
from quantopian.pipeline.factors import SimpleMovingAverage
import numpy as np

from pandas import Series, DataFrame
import pandas as pd
import statsmodels
import statsmodels.api 
import datetime as dt
import datetime as datetime

# Create custom factor subclass to calculate change in EPS
class EPSChange(CustomFactor):

    # Use basic_eps from the morningstar fundamentals data
    inputs = [morningstar.earnings_report.basic_eps]
    # Set the window length to be big enough to capture the end of the last quarter
    window_length = 150

    # Compute EPS change since end of last quarter
    def compute(self, today, assets, out, eps):
        out_list = []
        
        # For each security in the universe
        for i in range(eps.shape[1]):
            
            # Store the column of each day's eps for the security in eps_col
            eps_col = eps[:,i]
            
            # Since the basic_eps only changes at the end of each quarter, np.unique
            # will get us the unique EPS values in the 150-day window. return_index
            # is set to True so that the most recent unique values can be referenced later.
            _, idx = np.unique(eps_col, return_index=True)
            
            # Get the 2 most recent unique values in the 150-day value.
            prev_and_curr_eps = eps_col[np.sort(idx)[-2:]]
            
            # If there was only 1 distinct EPS value in the last 150-days,
            # duplicate it so that numpy can create a 2*num_stocks to output
            # in out[:]
            if len(prev_and_curr_eps) == 1:
                prev_and_curr_eps = np.append(prev_and_curr_eps, prev_and_curr_eps[0])
            
            # Store the 2 most recent unique EPS values for the stock in out_list
            out_list.append(prev_and_curr_eps)
            
        # Convert out_list to a 2*num_stocks numpy array
        all_prev_and_curr_eps = np.transpose(np.array(out_list))
        
        # Return the most recent EPS minus the previous EPS
        out[:] = (all_prev_and_curr_eps[-1] - all_prev_and_curr_eps[-2])/all_prev_and_curr_eps[-2]


def initialize(context):
    context.cached_universe = None
    schedule_function(func=regular_allocation,
                      date_rule=date_rules.week_start(),
                      time_rule=time_rules.market_open(),
                      half_days=True
                      )
    # Create the pipe
    pipe = Pipeline()
    attach_pipeline(pipe, 'eps-change-example')

    # Construct the custom factor
    eps_change = EPSChange()

    # Add the EPS change factor to the pipe
    pipe.add(eps_change, 'eps_change')
    
    # Create and apply a filter representing all stocks where the current EPS is
    # greater than the previous one
    eps_better_top_100 = eps_change.top(100)
    pipe.set_screen(eps_better_top_100)

def before_trading_start(context, data):
    context.days += 1
    
    if (context.days % 90 != 0 and context.cached_universe is not None):
        update_universe(context.cached_universe)
        return
    
    else:
        context.days = 0
    
    context.output = pipeline_output('eps-change-example').sort(['eps_change'], ascending=False)
    context.cached_universe=context.output.index
    context.universe=context.output.index
    update_universe(context.universe)
    
    
def handle_data(context, data):

    print "SECURITY LIST"
    log.info("\n" + str(context.output))

def regular_allocation(context, data):
    holdingSids = Series(context.portfolio.positions.keys())
    
    # prevent position creep. Removing any dangling positions we dont know about anymore
    for sid in holdingSids:
        if sid not in context.universe:
            order_target_percent(sid, 0.0)
    
    for sid in context.output.index :
        order_target_percent(sid,1/len(context.output.index))

There was a runtime error.

It looks like you were trying to increment context.days before having initialized it. As well, in regular_allocation, you were looping through sids in context.output.index instead of data. I've attached a working version of your code with the fix. I hope this helps!

Clone Algorithm
9
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
from quantopian.algorithm import attach_pipeline, pipeline_output
from quantopian.pipeline import Pipeline
from quantopian.pipeline import CustomFactor
from quantopian.pipeline.data.builtin import USEquityPricing
from quantopian.pipeline.data import morningstar
from quantopian.pipeline.factors import SimpleMovingAverage
import numpy as np

from pandas import Series, DataFrame
import pandas as pd
import statsmodels
import statsmodels.api 
import datetime as dt
import datetime as datetime

# Create custom factor subclass to calculate change in EPS
class EPSChange(CustomFactor):

    # Use basic_eps from the morningstar fundamentals data
    inputs = [morningstar.earnings_report.basic_eps]
    # Set the window length to be big enough to capture the end of the last quarter
    window_length = 150

    # Compute EPS change since end of last quarter
    def compute(self, today, assets, out, eps):
        out_list = []
        
        # For each security in the universe
        for i in range(eps.shape[1]):
            
            # Store the column of each day's eps for the security in eps_col
            eps_col = eps[:,i]
            
            # Since the basic_eps only changes at the end of each quarter, np.unique
            # will get us the unique EPS values in the 150-day window. return_index
            # is set to True so that the most recent unique values can be referenced later.
            _, idx = np.unique(eps_col, return_index=True)
            
            # Get the 2 most recent unique values in the 150-day value.
            prev_and_curr_eps = eps_col[np.sort(idx)[-2:]]
            
            # If there was only 1 distinct EPS value in the last 150-days,
            # duplicate it so that numpy can create a 2*num_stocks to output
            # in out[:]
            if len(prev_and_curr_eps) == 1:
                prev_and_curr_eps = np.append(prev_and_curr_eps, prev_and_curr_eps[0])
            
            # Store the 2 most recent unique EPS values for the stock in out_list
            out_list.append(prev_and_curr_eps)
            
        # Convert out_list to a 2*num_stocks numpy array
        all_prev_and_curr_eps = np.transpose(np.array(out_list))
        
        # Return the most recent EPS minus the previous EPS
        out[:] = (all_prev_and_curr_eps[-1] - all_prev_and_curr_eps[-2])/all_prev_and_curr_eps[-2]


def initialize(context):
    context.cached_universe = None
    schedule_function(func=regular_allocation,
                      date_rule=date_rules.week_start(),
                      time_rule=time_rules.market_open(),
                      half_days=True
                      )
    # Create the pipe
    pipe = Pipeline()
    attach_pipeline(pipe, 'eps-change-example')

    # Construct the custom factor
    eps_change = EPSChange()

    # Add the EPS change factor to the pipe
    pipe.add(eps_change, 'eps_change')
    
    # Create and apply a filter representing all stocks where the current EPS is
    # greater than the previous one
    eps_better_top_100 = eps_change.top(100)
    pipe.set_screen(eps_better_top_100)
    
    context.days = 0

def before_trading_start(context, data):
    context.days += 1
    
    if (context.days % 90 != 0 and context.cached_universe is not None):
        update_universe(context.cached_universe)
        return
    
    else:
        context.days = 0
    
    context.output = pipeline_output('eps-change-example').sort(['eps_change'], ascending=False)
    context.cached_universe=context.output.index
    context.universe=context.output.index
    update_universe(context.universe)
    
    
def handle_data(context, data):

    print "SECURITY LIST"
    log.info("\n" + str(context.output))

def regular_allocation(context, data):
    holdingSids = Series(context.portfolio.positions.keys())
    
    # prevent position creep. Removing any dangling positions we dont know about anymore
    for sid in holdingSids:
        if sid not in data:
            order_target_percent(sid, 0.0)
    
    for sid in data:
        order_target_percent(sid,1/len(context.output.index))

There was a runtime error.

Hi Jamie,

unfortunately it's not possible to use np.unique(.) to retrieve the last 4 quarters, because it could be that different quarters have the same values.

I've implemented the following logic. It's computationally correct if tested standalone, but within the backtester a TimeoutException always occurs... may the loop is too slow:

days_in_quarter = 70  
NA = object()

def retrieve_ttm(column):  
    ttm = []  
    i = 0  
    j = 0  
    ttm.append(column[i])  
    for previous, current in zip(column, column[1:]):  
        i += 1  
        if np.isnan(previous):  
            previous = NA  
        if np.isnan(current):  
            current = NA  
        if previous != current:  
            if i - j > days_in_quarter:  
                ttm.append(column[j])  
            ttm.append(column[i])  
            j = i

    if len(ttm) < 4:  
        ttm.append(column[i])

    return ttm[-4:]


class RevenueTTM(CustomFactor):  
    inputs = [morningstar.income_statement.total_revenue]  
    window_length = 252

    def compute(self, today, assets, out, revenue):  
        out_list = []  
        for i in range(revenue.shape[1]):  
            column = revenue[:,i]  
            ttm = retrieve_ttm(column)  
            out_list.append(ttm)  
        all_revenues = np.transpose(np.array(out_list))  
        revenue_ttm = np.sum(all_revenues, axis=0)  
        out[:] = revenue_ttm[0]  

The full algorithm is attached.
How may I implement this feature in a more efficient was? Thanks!

Clone Algorithm
46
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
from quantopian.algorithm import attach_pipeline, pipeline_output
from quantopian.pipeline import Pipeline
from quantopian.pipeline import CustomFactor
from quantopian.pipeline.data import morningstar
import numpy as np

days_in_quarter = 70
NA = object()

def retrieve_ttm(column):
    ttm = []
    i = 0
    j = 0
    ttm.append(column[i])
    for previous, current in zip(column, column[1:]):
        i += 1
        if np.isnan(previous):
            previous = NA
        if np.isnan(current):
            current = NA
        if previous != current:
            if i - j > days_in_quarter:
                ttm.append(column[j])
            ttm.append(column[i])
            j = i

    if len(ttm) < 4:
        ttm.append(column[i])

    return ttm[-4:]


class RevenueTTM(CustomFactor):
    inputs = [morningstar.income_statement.total_revenue]
    window_length = 252

    def compute(self, today, assets, out, revenue):
        out_list = []
        
        for i in range(revenue.shape[1]):
            column = revenue[:,i]
            ttm = retrieve_ttm(column)
            
            out_list.append(ttm)
                
        all_revenues = np.transpose(np.array(out_list))
        revenue_ttm = np.sum(all_revenues, axis=0)       
        out[:] = revenue_ttm[0]


def initialize(context):

    # Create the pipe
    pipe = Pipeline()
    attach_pipeline(pipe, 'Revenue-TTM-Example')

    revenue_ttm = RevenueTTM()

    pipe.add(revenue_ttm, 'revenue_ttm')
    
    # Create and apply a filter representing 
    revenue_ttm_top_100 = revenue_ttm.top(100)
    pipe.set_screen(revenue_ttm_top_100)

def before_trading_start(context, data):
    context.output = pipeline_output('Revenue-TTM-Example').sort(['revenue_ttm'], ascending=False)

    update_universe(context.output.index)

def handle_data(context, data):

    print "SECURITY LIST"
    log.info("\n" + str(context.output))
There was a runtime error.

I wonder if one could see the history of xxx_as_of and pick out the values when that changes...

Well, I'm trying to do that... the code above iterate over the last 252 day (a trading year) and store the value on each change or when there isn't any change for more than 70 days (meaning the value in the current quarter is the same as in the previous one).
The problem is than the loop takes too long for Quantopian and a timeout occurs.

I cannot yet figure how to overcome this issue and hope in the help of the community.

I attach now the full algo again because I fear the source code wasn't available in my previous post:

from quantopian.algorithm import attach_pipeline, pipeline_output  
from quantopian.pipeline import Pipeline  
from quantopian.pipeline import CustomFactor  
from quantopian.pipeline.data import morningstar  
import numpy as np

days_in_quarter = 70  
NA = object()

def retrieve_ttm(column):  
    ttm = []  
    i = 0  
    j = 0  
    ttm.append(column[i])  
    for previous, current in zip(column, column[1:]):  
        i += 1  
        if np.isnan(previous):  
            previous = NA  
        if np.isnan(current):  
            current = NA  
        if previous != current:  
            if i - j > days_in_quarter:  
                ttm.append(column[j])  
            ttm.append(column[i])  
            j = i

    if len(ttm) < 4:  
        ttm.append(column[i])

    return ttm[-4:]


class RevenueTTM(CustomFactor):  
    inputs = [morningstar.income_statement.total_revenue]  
    window_length = 252

    def compute(self, today, assets, out, revenue):  
        out_list = []  
        for i in range(revenue.shape[1]):  
            column = revenue[:,i]  
            ttm = retrieve_ttm(column)  
            out_list.append(ttm)  
        all_revenues = np.transpose(np.array(out_list))  
        revenue_ttm = np.sum(all_revenues, axis=0)  
        out[:] = revenue_ttm[0]


def initialize(context):

    # Create the pipe  
    pipe = Pipeline()  
    attach_pipeline(pipe, 'Revenue-TTM-Example')

    revenue_ttm = RevenueTTM()

    pipe.add(revenue_ttm, 'revenue_ttm')  
    # Create and apply a filter representing  
    revenue_ttm_top_100 = revenue_ttm.top(100)  
    pipe.set_screen(revenue_ttm_top_100)

def before_trading_start(context, data):  
    context.output = pipeline_output('Revenue-TTM-Example').sort(['revenue_ttm'], ascending=False)

    update_universe(context.output.index)

def handle_data(context, data):

    print "SECURITY LIST"  
    log.info("\n" + str(context.output))  
Clone Algorithm
46
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
from quantopian.algorithm import attach_pipeline, pipeline_output
from quantopian.pipeline import Pipeline
from quantopian.pipeline import CustomFactor
from quantopian.pipeline.data import morningstar
import numpy as np

days_in_quarter = 70
NA = object()

def retrieve_ttm(column):
    ttm = []
    i = 0
    j = 0
    ttm.append(column[i])
    for previous, current in zip(column, column[1:]):
        i += 1
        if np.isnan(previous):
            previous = NA
        if np.isnan(current):
            current = NA
        if previous != current:
            if i - j > days_in_quarter:
                ttm.append(column[j])
            ttm.append(column[i])
            j = i

    if len(ttm) < 4:
        ttm.append(column[i])

    return ttm[-4:]


class RevenueTTM(CustomFactor):
    inputs = [morningstar.income_statement.total_revenue]
    window_length = 252

    def compute(self, today, assets, out, revenue):
        out_list = []
        
        for i in range(revenue.shape[1]):
            column = revenue[:,i]
            ttm = retrieve_ttm(column)
            
            out_list.append(ttm)
                
        all_revenues = np.transpose(np.array(out_list))
        revenue_ttm = np.sum(all_revenues, axis=0)       
        out[:] = revenue_ttm[0]


def initialize(context):

    # Create the pipe
    pipe = Pipeline()
    attach_pipeline(pipe, 'Revenue-TTM-Example')

    revenue_ttm = RevenueTTM()

    pipe.add(revenue_ttm, 'revenue_ttm')
    
    # Create and apply a filter representing 
    revenue_ttm_top_100 = revenue_ttm.top(100)
    pipe.set_screen(revenue_ttm_top_100)

def before_trading_start(context, data):
    context.output = pipeline_output('Revenue-TTM-Example').sort(['revenue_ttm'], ascending=False)

    update_universe(context.output.index)

def handle_data(context, data):

    print "SECURITY LIST"
    log.info("\n" + str(context.output))
There was a runtime error.

Yeah you definitely shouldn't do any looping in a pipeline factor. You must do it vectorized.

I re-implemented without loops but the Timeout still occurs :-(

Here the new implementation:

days_in_quarter = 70  
num_of_quarters = 4

def retrieve_ttm(array):  
    # Differences between the next and the current value, after havind convertet NaN and Inf to numbers  
    diff = np.diff(np.nan_to_num(array))

    # Track the indexes where the difference isn't zero: a change has occurred. Shift the indexes by one.  
    idx = np.where(diff != 0)[0] + 1

    # Add zero as the first index (by definition)  
    idx = np.insert(idx, 0, 0)

    # Look for periods without changes longer than 'days_in_quarter'  
    dup = np.diff(idx) > days_in_quarter

    # Add an index in order to duplicate the previous value  
    idx = np.append(idx, idx[dup] + days_in_quarter)

    # Sort the index for consistence  
    idx = np.sort(idx)  
    return array[idx[-num_of_quarters:]]  

Hmm that's too bad :(