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.

351
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

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
sells = []
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:
if len(buy_symbols) + len(sell_symbols) > 0:
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 len(sells) > 0:
else:
elif stock in sells:
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.

46
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

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
sells = []
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:
if len(buy_symbols) + len(sell_symbols) > 0:
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 len(sells) > 0:
else:
elif stock in sells:
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.

71
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

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

# 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
sells = []
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:

if len(buy_symbols) + len(sell_symbols) > 0:
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 len(sells) > 0:
order_target(stock, 0)