Back to Community
Mean reversion of Quality factor over 12 earnings

Attached is an attempt.

Clone Algorithm
83
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
from quantopian.algorithm import attach_pipeline, pipeline_output
from quantopian.pipeline import Pipeline
from quantopian.pipeline.data import morningstar
from quantopian.pipeline.factors import CustomFactor
from quantopian.pipeline.data.builtin import USEquityPricing
from quantopian.pipeline.factors import AverageDollarVolume
from quantopian.pipeline.filters.morningstar import Q1500US

class Quality(CustomFactor):     
    inputs = [morningstar.income_statement.gross_profit, morningstar.balance_sheet.total_assets]
    window_length = 12
    
    def compute(self, today, assets, out, gross_profit, total_assets):
        norm = gross_profit / total_assets
        out[:] = (norm[-1] - np.mean(norm, axis=0)) / np.std(norm, axis=0)
        
def initialize(context):
    schedule_function(my_rebalance, date_rules.week_start(), time_rules.market_open(hours=1))
    make_pipeline()    
         
def make_pipeline():
    base_universe = Q1500US()
    min_filter = USEquityPricing.close.latest > 15
    pipe = Pipeline(screen = base_universe & min_filter)
    attach_pipeline(pipe, 'my_pipeline')
    pipe.add(Quality(),"Quality")
 
def before_trading_start(context, data):
    context.output = pipeline_output('my_pipeline')
    context.security_list = context.output.index
     
def my_rebalance(context,data):
    ranks = context.output["Quality"]
    ranks = ranks.dropna()
    idx = ranks.values.argsort()
    
    longs = []
    shorts = []
    
    for i, sid in enumerate(ranks.index):
        if i in idx[:75]:
            shorts.append(sid)
        elif i in idx[-75:]:
            longs.append(sid)
    
    for sid in shorts:
        order_target_value(sid, context.portfolio.portfolio_value / 150.)
    
    for sid in longs:
        order_target_value(sid, -context.portfolio.portfolio_value / 150.)
        
    for sid in context.portfolio.positions:
        if sid not in longs or sid not in shorts:
            order_target(sid, 0)

def handle_data(context,data):
    pass
There was a runtime error.
1 response

Improved version.

Clone Algorithm
83
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
from quantopian.algorithm import attach_pipeline, pipeline_output
from quantopian.pipeline import Pipeline
from quantopian.pipeline.data import morningstar
from quantopian.pipeline.factors import CustomFactor
from quantopian.pipeline.data.builtin import USEquityPricing
from quantopian.pipeline.factors import AverageDollarVolume
from quantopian.pipeline.filters.morningstar import Q1500US

class Quality(CustomFactor):     
    inputs = [morningstar.income_statement.gross_profit, morningstar.balance_sheet.total_assets]
    window_length = 12
    
    def compute(self, today, assets, out, gross_profit, total_assets):
        norm = gross_profit / total_assets
        out[:] = (norm[-1] - np.mean(norm, axis=0)) / np.std(norm, axis=0)
        
def initialize(context):
    schedule_function(my_rebalance, date_rules.week_start(), time_rules.market_open(hours=1))
    make_pipeline()    
         
def make_pipeline():
    base_universe = Q1500US()
    min_filter = USEquityPricing.close.latest > 15
    pipe = Pipeline(screen = base_universe & min_filter)
    attach_pipeline(pipe, 'my_pipeline')
    pipe.add(Quality(),"Quality")
 
def before_trading_start(context, data):
    context.output = pipeline_output('my_pipeline')
    context.security_list = context.output.index
     
def my_rebalance(context,data):
    ranks = context.output["Quality"]
    ranks = ranks.dropna()
    idx = ranks.values.argsort()
    
    longs = []
    shorts = []
    
    for i, sid in enumerate(ranks.index):
        if i in idx[:25]:
            shorts.append(sid)
        elif i in idx[-25:]:
            longs.append(sid)
    
    for sid in shorts:
        order_target_value(sid, context.portfolio.portfolio_value / 50.)
    
    for sid in longs:
        order_target_value(sid, -context.portfolio.portfolio_value / 50.)
        
    for sid in context.portfolio.positions:
        if sid not in longs and sid not in shorts:
            order_target(sid, 0)

def handle_data(context,data):
    pass
There was a runtime error.