Back to Community
Elder Impulse Algo (sort of)

The algo is based on the Elder Impulse system from one of Elder’s classic tradings books. I added touches of my own here and there.

The main idea I used from the system is that bull signals occur when

13-period EMA > previous 13-period EMA and
MACD-Histogram > previous period's MACD-Histogram

... and vise-versa for bear signals.

The algo definitely doesn’t work the way I want it to, especially in the area of def before_trading_start(context, data):
(which returns different results in stock selection compared to when that code is run in Pipeline)

Any recommendations or pointers would be nice as this is one my first algos.

Clone Algorithm
12
Loading...
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
# library imports
import quantopian.algorithm as algo
from quantopian.pipeline import Pipeline
from quantopian.pipeline.data.builtin import USEquityPricing
from quantopian.pipeline.filters import QTradableStocksUS
from quantopian.pipeline.factors import Latest, CustomFactor, AverageDollarVolume, AnnualizedVolatility
import quantopian.optimize as opt
from scipy import stats
import numpy as np
import pandas as pd
import talib

#-------------------------------------------------------------------

def initialize(context):
    """
    Called once at the start of the algorithm.
    """
    
    # Rebalance every day, 10 minutes after market open
    algo.schedule_function(rebalance,
                           algo.date_rules.every_day(),
                           algo.time_rules.market_open(minutes=10)
                          )

    # Record tracking variables at the end of each day
    algo.schedule_function(recordMetrics,
                           algo.date_rules.every_day(),
                           algo.time_rules.market_close()
                           )

    # Create the dynamic stock selector.
    algo.attach_pipeline(make_pipeline(), 'myPipe')
    
    # Set Standard slippage and commission rules
    set_slippage(slippage.FixedBasisPointsSlippage(basis_points=5, volume_limit=0.1))  
    set_commission(commission.PerShare(cost=0.001, min_trade_cost=1)) 
 
    # Read leverage on every minute processed and chart the maximum every day
    context.maxLeverage = 0  
    for i in range(1, 391):  
        schedule_function(maxLeverage, date_rules.every_day(), time_rules.market_open(minutes=i))
#-------------------------------------------------------------------

def make_pipeline():
    """
    A function to create our dynamic stock selector (pipeline) 
    """

    # Base universe set to the QTradableStocksUS, filtererd by latest close price
    latestClose = USEquityPricing.close.latest
    priceRANGE1 = latestClose >= 5.25  
    priceRANGE2 = latestClose <= 40.00
    priceFilter = priceRANGE1 & priceRANGE2
    filterUniverse = QTradableStocksUS() & priceFilter
    
    # Factor for minimum shares traded and minimum average dollar volume   
    volume = (USEquityPricing.volume.latest / 1000000) 
    volumeFilter = volume >= 1.05
    
    dollarVolume = AverageDollarVolume(window_length=30, mask=filterUniverse)
    high_dollarVolume = dollarVolume.top(600) 
    liquidityFilter = high_dollarVolume & volumeFilter
    
    # Factor for average volatilty over a 1-yr trading period
    volatility = AnnualizedVolatility(window_length = 252, mask=filterUniverse)
    volatiltyFilter = volatility.percentile_between(50,100)
    
    # Combined filter screen
    combinedFilter = liquidityFilter & priceFilter & volatiltyFilter
    
    pipe = Pipeline(columns={'volatility': volatility},
                            screen = combinedFilter
                    )
   
    return pipe 

#-------------------------------------------------------------------
            
def before_trading_start(context, data):
    """
    Called every day before market open.
    """
    # These are the securities that we are interested in trading each day
    context.result = algo.pipeline_output('myPipe')
    
    context.stockList = context.result.sort_values(by='volatility',  ascending=False).dropna() 
    
    # Retrieves historical window of close price data (adjusted for splits, mergers,           and dividends)
    priceHistory = data.history(context.stockList.index.unique(), fields="close",       bar_count=250, frequency="1d")
    
    """
    Iterate over the list of equities in our filtered stock list:
     1. calculate EMA and MACD-Histogram values for both 1 & 2 days ago
     2. find the difference between values for both EMA & MACD-Histogram
     3. z-score the magnitude of each difference for EMA + MACD-Histogram
     4. value longs as greatest positive z-score values and shorts as greatest negative  z-score values
    """
    
    ratioHist = []
    ratioEMA = []
    
    for stocks in context.stockList.index.unique().tolist():
        EMA1 = talib.EMA(priceHistory[stocks], timeperiod=13)[-1]
        EMA2 = talib.EMA(priceHistory[stocks], timeperiod=13)[-2]   
        ema_diff = EMA1 - (EMA2)           
        ratioEMA.append(ema_diff)

        macd, signal, hist = talib.MACD(priceHistory[stocks],
                                        fastperiod = 12,
                                        slowperiod = 26,
                                        signalperiod = 9)
        hist_diff = hist[-1] - (hist[-2])    
        ratioHist.append(hist_diff)
    
    # NaNs to 0s
    ema_toSeries  = (pd.Series(ratioEMA).fillna(0))
    hist_toSeries = (pd.Series(ratioHist).fillna(0))

    # Calculate z-scores
    zscoreEMA =  stats.zscore(ema_toSeries)
    zscoreHist = stats.zscore(hist_toSeries)
    zscoreCombined = (zscoreEMA + zscoreHist)

    # Attach z-score list to output dataframe
    zscoreDataFrame = context.result.assign(ratio_ema = ratioEMA, 
                                            ratio_hist = ratioHist, zscore = zscoreCombined).dropna()
    
    # Longs & shorts sorted by the 30 most positive and 30 most negative z-scores
    longs = zscoreDataFrame.loc[(zscoreDataFrame['ratio_hist'] > 0) &
                                (zscoreDataFrame['ratio_ema'] > 0)].nlargest(30, 'zscore')
    shorts = zscoreDataFrame.loc[(zscoreDataFrame['ratio_hist'] < 0) &
                                 (zscoreDataFrame['ratio_ema'] < 0)].nsmallest(30,       'zscore')
    
    context.Longs_Shorts = longs.append(shorts).sort_values('zscore', ascending=False)   
            
    print context.Longs_Shorts
#-------------------------------------------------------------------

def rebalance(context, data):
    """
    Execute orders according to our schedule_function() timing.
    """ 
    # Convert stocks to trade into a series 
    Longs_Shorts_toSeries = pd.Series(context.Longs_Shorts.zscore)
    alphaObjective = opt.MaximizeAlpha(Longs_Shorts_toSeries)    
                                         
    # Contraints set for max leverage, position sizes, and long/short bias
    max_exposure = opt.MaxGrossExposure(0.90)
    max_position_size = opt.PositionConcentration.with_equal_bounds(-.035, .035)
    dollar_neutral = opt.DollarNeutral()

    portfolioConstraints =  [max_exposure,
                             max_position_size,
                             dollar_neutral
                             ]
    
    order_optimal_portfolio(objective = alphaObjective, 
                            constraints = portfolioConstraints)

#-------------------------------------------------------------------

def recordMetrics(context, data):
    """
    Plot variables at the end of each day.
    """
    # Long, short position counter    
    longs = shorts = 0  
    for position in context.portfolio.positions.itervalues():  
        if position.amount > 0:  
            longs += 1  
        if position.amount < 0:  
            shorts += 1  
    
    # Variables to plot at the end of each trading day
    record(Excess_Funds = context.account.available_funds,
           Position_Count = len(context.portfolio.positions),
           Long_Count = longs, Short_Count = shorts) 

# Additional function to record leverage on a intraday basis
def maxLeverage(context, data):  
    if context.account.leverage > context.maxLeverage:  
        context.maxLeverage = context.account.leverage  
        record(Max_Leverage = context.maxLeverage)  
        
#///////////////////////////////////////////////////////////////////
There was a runtime error.
2 responses

Thank you for the indicator code. But just want to point out the Elder Impulse System (EIS) does not quite suit the Zscore sorting strategy. As you know the EIS combines the EMA (trend) with the MACD (momentum). It is best used for an entry/exit identification strategy. Because EMA is a time-weighted (recent weights more) indicator and MACD is also generated through EMA calculation, A high Zscore of EMA and MACD in general means a huge jump bar in very recent days.

Stocks in your filter > $5.25 close price can have very high volatility and have frequent jumps (up/down) causing high Zscore of EMA and MACD. These jumps may move those stocks into your trade list, but may not generate profit for you.

Hey, thanks for the input.
I remember when I was working on this, the rationale behind the zscore code was indeed to search for increased volatility or significant green/red days on stocks. I know that ideally the Elder Impulse strategy also involves how a longer timeframe is doing, such as weekly price action.
...But nevertheless, a problem occurs when there is whipsaw action, like EMA & MACD might be positive one day and negative the next. This is not what I wanted the algo to trade as it creates high turnover... so to avoid this problem I predicted that it would be better to only enter where there is a significant change is those two factors (maybe a big reversal of trend or breakout from lows or a gap up, etc).