Back to Community
Trading on multiple TA-Lib signals

One thing that many traders do is look at multiple signals in conjunction with one another. If all the signals indicate the same thing, they are often more likely to be correct.

Here's an example using three signals generated from TA-Lib. I want this post to serve a couple purposes — introducing the strategy and showing how to use the talib module.

This strategy takes a chunk of stocks, and for each stock, it looks at three signals. If all the signals for a stock are in agreement about whether it is overbought or underbought, then we go short or long. The three signals we use are:

  • Money flow index
  • Commodity channel index
  • Chande momentum oscillator

You can see the thresholds I use and other variables, like time period and the universe size, in the source code. Since the "ta." notation is being deprecated, using functions from the talib module is better.

Interestingly enough, the strategy does very well from about 2002 - 2009. After that, it mostly trends around the benchmark. I wonder if there is a good reason why it did so well for a while then slowed down (the backtest below is run from 2004 - 2006). Feel free to clone the algorithm and adjust it. Also try using other functions from TA-Lib as signals, specifically have a look at the various oscillators.

Clone Algorithm
351
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
import talib
from collections import deque

# Initialization
def initialize(context):
    # Use a chunk of stocks in the dollar volume universe
    set_universe(universe.DollarVolumeUniverse(96, 99))

    # We can ignore slippage and commision because we are just testing the accuracy of signals
    set_slippage(slippage.FixedSlippage(spread=0))
    set_commission(commission.PerTrade(cost=0))
    
    
    context.timeperiod = 14 # Use a signal generated from the past 14 days of price data
    context.levarage = 1.0
    
    # Store the price and volume data in these dictionaries
    context.highs_frame = {}
    context.lows_frame = {}
    context.closes_frame = {}
    context.volumes_frame = {}
    
# Get our indicators. We pass this function the needed data.
def get_indicator(highs_frame, lows_frame, closes_frame, volumes_frame, ttimeperiod):
    # Get money flow index
    mfi = talib.MFI(numpy.array(highs_frame), numpy.array(lows_frame), numpy.array(closes_frame), numpy.array(volumes_frame), ttimeperiod)[-1]
    
    # Get commodity channel index
    cci = talib.CCI(numpy.array(highs_frame), numpy.array(lows_frame), numpy.array(closes_frame), ttimeperiod)[-1]
    
    # Get Chande momentum oscillator
    cmo = talib.CMO(numpy.array(closes_frame), ttimeperiod)[-1]
    
    return mfi, cci, cmo

# This is run once per bar, once per day in this case
def handle_data(context, data):
    
    # Store which stocks we are going long and short on in this bar here
    buys = []
    sells = []
    buy_symbols = []
    sell_symbols = []
    
    # For each stock in our universe
    for stock in data:
        
        # Update our frames of relevant data
        if stock not in context.highs_frame:
            context.highs_frame[stock] = deque(maxlen=context.timeperiod+1)
            context.lows_frame[stock] = deque(maxlen=context.timeperiod+1)
            context.closes_frame[stock] = deque(maxlen=context.timeperiod+1)
            context.volumes_frame[stock] = deque(maxlen=context.timeperiod+1)
        context.highs_frame[stock].append(data[stock].high)
        context.lows_frame[stock].append(data[stock].low)
        context.closes_frame[stock].append(data[stock].close_price)
        context.volumes_frame[stock].append(float(data[stock].volume))
        
        # Make sure the stock has enough data for computation of the signals
        if len(context.highs_frame[stock]) < context.timeperiod+1:
            continue
        
        # Get the indicators
        mfi, cci, cmo = get_indicator(context.highs_frame[stock], context.lows_frame[stock],  context.closes_frame[stock], context.volumes_frame[stock], context.timeperiod)
        
        # Add a stock to our list of short if the signals all indicate that it's overbought
        if cci > 100 and mfi > 80 and cmo > 50:
            sells.append(stock)
            sell_symbols.append(stock.symbol)
        # Add a stock to our list of longs if the signals all indicate that it's underbought
        elif cci < -100 and mfi < 20 and cmo < -50:
            buys.append(stock)
            buy_symbols.append(stock.symbol)
    if len(buy_symbols) + len(sell_symbols) > 0:
        log.info('Longs today: '+str(', '.join(buy_symbols)))
        log.info('Shorts today: '+str(', '.join(sell_symbols)))
            
    # Go through and long or short each stock, making sure we allocate the leveraged fraction of our portfolio
    for stock in data:
        if stock in buys:
            if len(sells) > 0:
                order_target_percent(stock, 0.5*context.levarage/len(buys))
            else:
                order_target_percent(stock, 1.0*context.levarage/len(buys))
        elif stock in sells:
            if len(buys) > 0:
                order_target_percent(stock, -0.5*context.levarage/len(sells))
            else:
                order_target_percent(stock, -1.0*context.levarage/len(sells))
        else:
            order_target(stock, 0)
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.

3 responses

Here's another example. This one only trades one stock, the SPY, and records the indicators. It is also designed for minute mode and live trading, and because of that uses the history functions instead of deques to retrieve past data.

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
import numpy
import talib
from collections import deque

# Initialization
def initialize(context):
    # Use a chunk of stocks in the dollar volume universe
    #set_universe(universe.DollarVolumeUniverse(80, 85))
    
    context.stock = sid(8554) # Just trade the SPY

    # We can ignore slippage and commision because we are just testing the accuracy of signals
    set_slippage(slippage.FixedSlippage(spread=0))
    set_commission(commission.PerTrade(cost=0))
    
    
    context.timeperiod = 14 # Use a signal generated from the past 14 days of price data
    context.levarage = 1.0
    
    # Store the price and volume data in these dictionaries
    context.highs_frame = {}
    context.lows_frame = {}
    context.closes_frame = {}
    context.volumes_frame = {}
    
# Get our indicators. We pass this function the needed data.
def get_indicators(highs_frame, lows_frame, closes_frame, volumes_frame, ttimeperiod):
    # Get money flow index
    mfi = talib.MFI(numpy.array(highs_frame), numpy.array(lows_frame), numpy.array(closes_frame), numpy.array(volumes_frame), ttimeperiod)[-1]
    
    # Get commodity channel index
    cci = talib.CCI(numpy.array(highs_frame), numpy.array(lows_frame), numpy.array(closes_frame), ttimeperiod)[-1]
    
    # Get Chande momentum oscillator
    cmo = talib.CMO(numpy.array(closes_frame), ttimeperiod)[-1]
    
    record(SPY_CCI=cci, SPY_MFI=mfi, SPY_CMO=cmo)
    
    return mfi, cci, cmo

# This is run once per bar, once per day in this case
def handle_data(context, data):
    
    # Store which stocks we are going long and short on in this bar here
    buys = []
    sells = []
    buy_symbols = []
    sell_symbols = []
    
    # For each stock in our universe
    for stock in data:
        
        highs_frame = history(15, "1d", "high")[stock.sid].values
        lows_frame = history(15, "1d", "low")[stock.sid].values
        closes_frame = history(15, "1d", "close_price")[stock.sid].values
        volumes_frame = history(15, "1d", "volume")[stock.sid].values
        
        # Get the indicators
        mfi, cci, cmo = get_indicators(highs_frame, lows_frame, closes_frame, volumes_frame, context.timeperiod)
        
        # Add a stock to our list of short if the signals all indicate that it's overbought
        if cci > 100 and mfi > 80 and cmo > 50:
            sells.append(stock)
            sell_symbols.append(stock.symbol)
        # Add a stock to our list of longs if the signals all indicate that it's underbought
        elif cci < -100 and mfi < 20 and cmo < -50:
            buys.append(stock)
            buy_symbols.append(stock.symbol)
    if len(buy_symbols) + len(sell_symbols) > 0:
        log.info('Longs today: '+str(', '.join(buy_symbols)))
        log.info('Shorts today: '+str(', '.join(sell_symbols)))
            
    # Go through and long or short each stock, making sure we allocate the leveraged fraction of our portfolio
    for stock in data:
        if stock in buys:
            if len(sells) > 0:
                order_target_percent(stock, 0.5*context.levarage/len(buys))
            else:
                order_target_percent(stock, 1.0*context.levarage/len(buys))
        elif stock in sells:
            if len(buys) > 0:
                order_target_percent(stock, -0.5*context.levarage/len(sells))
            else:
                order_target_percent(stock, -1.0*context.levarage/len(sells))
        else:
            order_target(stock, 0)
There was a runtime error.

And here's one more using more signals and slightly different logic.

Clone Algorithm
71
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
import talib
from collections import deque

# Initialization
def initialize(context):
    # Use a chunk of stocks in the dollar volume universe
    set_universe(universe.DollarVolumeUniverse(95, 100))

    # We can ignore slippage and commision because we are just testing the accuracy of signals
    set_slippage(slippage.FixedSlippage(spread=0))
    set_commission(commission.PerTrade(cost=0))
    
    
    context.timeperiod = 14 # Use a signal generated from the past 14 days of price data
    context.levarage = 1.0
    
    # Store the price and volume data in these dictionaries
    context.highs_frame = {}
    context.lows_frame = {}
    context.closes_frame = {}
    context.volumes_frame = {}
    
# Get our indicators. We pass this function the needed data.
def get_indicator(highs_frame, lows_frame, closes_frame, volumes_frame, ttimeperiod):
    # Get money flow index
    mfi = talib.MFI(highs_frame, lows_frame, closes_frame, volumes_frame, ttimeperiod)[-1]
    
    # Get commodity channel index
    cci = talib.CCI(highs_frame, lows_frame, closes_frame, ttimeperiod)[-1]
    
    # Get Chande momentum oscillator
    cmo = talib.CMO(closes_frame, ttimeperiod)[-1]
    
    # Get Aroon oscillator
    aroonosc = talib.AROONOSC(highs_frame, lows_frame, ttimeperiod)[-1]
    
    # Get average directional movement index
    adx = talib.ADX(highs_frame, lows_frame, closes_frame, ttimeperiod)[-1]
    
    # Get relative strength index
    rsi = talib.RSI(closes_frame, ttimeperiod)[-1]
    
    return mfi, cci, cmo, aroonosc, adx, rsi

# This is run once per bar, once per day in this case
def handle_data(context, data):
    
    # Store which stocks we are going long and short on in this bar here
    buys = []
    sells = []
    buy_symbols = []
    sell_symbols = []
    
    # For each stock in our universe
    for stock in data:
        
        # Update our frames of relevant data
        if stock not in context.highs_frame:
            context.highs_frame[stock] = deque(maxlen=context.timeperiod+1)
            context.lows_frame[stock] = deque(maxlen=context.timeperiod+1)
            context.closes_frame[stock] = deque(maxlen=context.timeperiod+1)
            context.volumes_frame[stock] = deque(maxlen=context.timeperiod+1)
        context.highs_frame[stock].append(data[stock].high)
        context.lows_frame[stock].append(data[stock].low)
        context.closes_frame[stock].append(data[stock].close_price)
        context.volumes_frame[stock].append(float(data[stock].volume))
        
        # Make sure the stock has enough data for computation of the signals
        if len(context.highs_frame[stock]) < context.timeperiod+1:
            continue
        
        # Get the indicators
        mfi, cci, cmo, aroonosc, adx, rsi = get_indicator(numpy.array(context.highs_frame[stock]), numpy.array(context.lows_frame[stock]),  numpy.array(context.closes_frame[stock]), numpy.array(context.volumes_frame[stock]), context.timeperiod)
        
        score = 0
        if cci > 100:
            score -= 1
        elif cci < -100:
            score += 1
        #if mfi > 80:
        #    score -= 1
        #elif mfi < 20:
        #    score += 1
        if cmo > 50:
            score -= 1
        elif cmo < -50:
            score += 1
        if aroonosc < -50:
            score -= 1
        elif aroonosc > 50:
            score += 1
        if rsi > 70:
            score -= 1
        elif rsi < 30:
            score += 1

        # Add a stock to our list of shorts if the signals all indicate that it's overbought
        if score <= -2:
            sells.append(stock)
            sell_symbols.append(stock.symbol)
        # Add a stock to our list of longs if the signals all indicate that it's underbought
        elif score >= 2:
            buys.append(stock)
            buy_symbols.append(stock.symbol)
            
            
    if len(buy_symbols) + len(sell_symbols) > 0:
        log.info('Longs today: '+str(', '.join(buy_symbols)))
        log.info('Shorts today: '+str(', '.join(sell_symbols)))
            
    # Go through and long or short each stock, making sure we allocate the leveraged fraction of our portfolio
    for stock in data:
        if stock in buys:
            if len(sells) > 0:
                order_target_percent(stock, 0.5*context.levarage/len(buys))
            else:
                order_target_percent(stock, 1.0*context.levarage/len(buys))
        elif stock in sells:
            if len(buys) > 0:
                order_target_percent(stock, -0.5*context.levarage/len(sells))
            else:
                order_target_percent(stock, -1.0*context.levarage/len(sells))
        else:
            order_target(stock, 0)
There was a runtime error.

Hi. Will anyone be able to re-code this for the new API?