Back to Community
Batch Average True Range

Recently I've been playing around with Quantopian/Zipline building a library of indicators I might find useful in the future. I came across the previous implementations of ATR shared, and thought I'd share my own stab at it, both iteratively in a custom event window and using a batch transform.

Happy Hacking!

7-11

Previous Posts:
https://www.quantopian.com/posts/average-true-range-basic-implementation
https://www.quantopian.com/posts/python-classes-implementing-true-range-and-average-true-range-indicators

Clone Algorithm
19
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 pandas as pd

from zipline.transforms import utils as tnfs_utils


class ATREventWindow(tnfs_utils.EventWindow):
    def __init__(self, window_length):
        tnfs_utils.EventWindow.__init__(self, window_length=window_length)
        self.atr = 0
        self.alpha = 2.0 / (self.window_length + 1)
        
    def handle_add(self, event):
        if len(self.ticks) < 2:
            return
        
        if self.atr is 0:
            self.atr = self.true_range()
        else:
            self.atr = self.atr + self.alpha * (self.true_range() - self.atr)
    
    def true_range(self):
        data = list(self.ticks)
        high = data[-1]['high']
        low = data[-1]['low']
        prev_close = data[-2]['close_price']
        return max(high - low, abs(high - prev_close), abs(low - prev_close))
        
        
    def handle_remove(self, event):
        pass
    
    def __call__(self):
        return self.atr


def _true_range(d, last_close=None):
    if last_close is None:
        last_close = d['close_price'].shift(1)
    return pd.Panel({'range': d['high'] - d['low'],
                     'range_high': (d['high'] - last_close).abs(),
                     'range_low': (d['low'] - last_close).abs()}).max(axis=0)


def _atr(d, span=14, true_range=None):
    if true_range is None:
        true_range = _true_range(d)
    return pd.ewma(true_range, span=span)


@batch_transform(window_length=14)
def atr(d):
    return _atr(d)


def initialize(context):
    context.sid = sid(26578)
    context.atr_window = ATREventWindow(window_length=14)

       
def handle_data(context, data):
    data[context.sid].dt = data[context.sid].datetime
    context.atr_window.update(data[context.sid])
    
    record(window=context.atr_window())
    
    atrs = atr(data)
    if atrs is None:
        return
    
    record(batch=atrs.ix[-1][context.sid])
This backtest was created using an older version of the backtester. Please re-run this backtest to see results using the latest backtester. Learn more about the recent changes.
There was a runtime error.
8 responses

self.alpha = 2.0 / (self.window_length + 1)

atr = (window_length*atr + tr)/(window_length + 1)
= ((window_length+1) atr + (tr - atr)) / (window_length + 1)
= atr + (1./(window_length + 1)) * (tr - atr)

why the 2.0?

Its arbitrary, but usually I like to give more weight to recent events (and is customary technical analysis) so I always bump up the alpha that way.

Hello Jason,

That looks like a lot of effort went in to it. With TA-Lib we can just use:

import talib

def initialize(context):  
    context.sid = sid(26578)  


def handle_data(context, data):  
    high        = history(15, '1d', 'high')  
    low         = history(15, '1d', 'low')  
    close       = history(15, '1d', 'close_price')  
    ATR         = talib.ATR(high[context.sid], low[context.sid], close[context.sid])  
    record(ATR=float(ATR[-1:]))  

P.

Yes now that talib is available for import on the quantopian platform its much easier, the above code predates this feature.

For a very simple ATR implementation, check out the code for this example.

Dyno, if you want more weight to recent events, just shorten your time window. Your alpha calc is wrong (or at least it doesn't provide the correct ATR). ATR is an exponentially weighted moving average of the True Range, where the weight on the most recent observation (alpha) is equal to 1 / time_period and the weight on previous ATR value is 1 - alpha.

Here is the TaLib implementation of ATR in C.

Clone Algorithm
25
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 math

# Set up the ATR indicator.
atr_data = ta.ATR(timeperiod=15)

def initialize(context):  
    context.stock = sid(26578) # GOOG 

def handle_data(context, data):  
    atr = atr_data(data)[context.stock]
    record(ATR=atr)
    
This backtest was created using an older version of the backtester. Please re-run this backtest to see results using the latest backtester. Learn more about the recent changes.
There was a runtime error.

Check out the code for this ATR example which makes it clear how the calculation is actually done. It manually calculates the ATR values and then compares them to the Talib ATR value. They are an exact match.

Clone Algorithm
25
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
# Set up the ATR indicator.
atr_data = ta.ATR(timeperiod=20)

def initialize(context):  
    context.stock = sid(26578) # GOOG 
    context.bar_counter = 0
    context.atr = 0
    context.prior_atr = 0
    context.prior_close = 0

def handle_data(context, data):  
    context.bar_counter += 1
    time_period = 20  # for ATR calc
    alpha = 1 / float(time_period)  # for ATR calc
    
    high = data[context.stock].high
    low = data[context.stock].low
    close = data[context.stock].close_price
    true_range = max(high, context.prior_close) - min(low, context.prior_close)
    
    if context.bar_counter > time_period + 1:
        # ATR is an exponentially weighted moving average calc
        ATR = alpha * true_range + (1-alpha) * context.prior_atr
    elif context.bar_counter == time_period + 1:
        # the first ATR calc is just a simple average of the TR values over the initial time_period window
        ATR = (context.prior_atr + true_range) / time_period
    elif context.bar_counter > 1:
        # at least two bars needed to calculate true range, so ignore first bar
        # this temporarily stores the TR aggregate for use in the averaging above
        ATR = context.prior_atr + true_range
    else:
        # the first bar can't be used because there is no true range
        ATR = 0
        
    # update values for use in next handle_data call
    context.prior_close = close
    context.prior_atr = ATR
    
    # if not enough data for time_period, set return value to NaN
    if context.bar_counter <= time_period:
        ATR = float('NaN')
    
    # Ta-lib ATR value
    atr = atr_data(data)[context.stock]
    
    # Graph and print results comparing Ta-lib ATR value with calculated ATR
    record(ATR=ATR, Talib_ATR=atr)
    log.info('ATR: {0:.5f}  Talib-Atr: {1:.5f}'.format(ATR, atr))
    
This backtest was created using an older version of the backtester. Please re-run this backtest to see results using the latest backtester. Learn more about the recent changes.
There was a runtime error.

Hey Colin,

It looks like cloning the following code, results in the message: Undefined name 'ta'

Any ideas?

Thanks,

Christos


import math

Set up the ATR indicator.

atr_data = ta.ATR(timeperiod=15)

def initialize(context):
context.stock = sid(26578) # GOOG

def handle_data(context, data):
atr = atr_data(data)[context.stock]
record(ATR=atr)

I believe "ta" is short for "talib" so

import talib as ta

may do the trick.