mhallsmoore / qstrader

QuantStart.com - QSTrader backtesting simulation engine.
https://www.quantstart.com/qstrader/
MIT License
2.93k stars 855 forks source link

ATRIndicator #222

Closed JamesKBowler closed 8 months ago

JamesKBowler commented 7 years ago

Hi all,

Made this today, thought I would share. Anyone has a faster alternative please let me know.

from collections import deque
import numpy as np

from qstrader.indicators.base import AbstractIndicator
from qstrader.event import EventType

class ATRIndicator(AbstractIndicator):
    """
    Requires:
    ticker - The ticker symbol being used for moving averages
    window - Lookback period for exponential moving average
    true range window - Period for calculating ATR

    # Some information about TR
    http://www.investopedia.com/terms/a/atr.asp
    ch_cl = current high - current low
    ch_pc = current high - previous close (absolute value)
    cl_pc = current low - previous close (absolute value)
    tr = max(ch_cl, ch_pc, cl_pc)
    """
    def __init__(self, ticker, window=90):
        self.ticker = ticker
        self.w_bars = deque(maxlen=self.window)
        self.tr_bars = deque(maxlen=self.window)
        self.atr_values = deque(maxlen=self.window)
        self.bars = 0
        self.atr = 0

    def calculate(self, event):
        if (
            event.type == EventType.BAR and event.ticker == self.ticker
        ):
            self.bars += 1
            # Add latest adjusted closing price to the window of bars
            self.w_bars.append(event.adj_close_price)
            # Calculate True Range
            ch_cl = event.high_price - event.low_price
            ch_pc = 0.0
            cl_pc = 0.0
            if self.bars != 1:
                ch_pc = abs(event.high_price - self.w_bars[-2])
                cl_pc = abs(event.low_price - self.w_bars[-2])
            tr = max(ch_cl, ch_pc, cl_pc)
            self.tr_bars.append(tr)
            # Calulate first average point
            if self.bars == self.window:
                avg = np.mean(self.tr_bars)
                self.atr_values.append(avg)
            # Calulate exponential moving average
            elif self.bars > self.window:
                self.atr = int(
                    (self.atr_values[-1] * (self.window-1) + self.tr_bars[-1]
                        ) // self.window
                )
                self.atr_values.append(self.atr)

Access the latest ATR value by calling:

Indicators.dict_ind['ATRIndicator']['APPL'].atr

JamesKBowler commented 7 years ago

Here is the abstract class.

from abc import ABCMeta, abstractmethod
from collections import defaultdict

class AbstractIndicator(object):
    """
    AbstractIndicator is an abstract base class providing an
    interface for all subsequent (inherited) indicator handling
    objects.

    The goal of a (derived) Indicator object is to generate indicator
    values for particular symbols based on the inputs of bars
    generated from a PriceHandler (derived) object.

    This is designed to work both with historic and live data as
    the Indicator object is agnostic to data location.
    """

    __metaclass__ = ABCMeta

    @abstractmethod
    def calculate(self, event):
        """
        Provides the mechanisms to calculate the list of indicators.
        """
        raise NotImplementedError("Should implement calculate()")

class Indicators(AbstractIndicator):
    """
    Indicators is a collection of Indicator
    """
    def __init__(self, *indicators):
        self._dict_ind = defaultdict(dict)
        for i in indicators:
            self._dict_ind[i.__class__.__name__][i.ticker] = i

    def calculate(self, event):
        for indicator, obj in self._dict_ind.items():
            try:
                obj[event.ticker].calculate(event)
            except KeyError:
                pass
enriqueromualdez commented 7 years ago

Hi @JamesKBowler,

Went through the code and it looks perfectly fine to me.

I haven't made an separate ATR indicator, but I've coded up an ATR Stop Loss and Keltner Channels which use the ATR. We have very similar code minus the use of the AbstractIndicator class.

I've noticed that the accuracy of the ATR is off for the first year's worth of bar data (252 bars). This was when compared to TradingView's ATR.

Here's my version of the ATR, but as a TrailingStopLoss. I've redacted the code that's specific to the Stop Loss and kept the ATR calculation:

class ATRStopLoss(AbstractRiskManager):
        self.y_close = [0]
        self.bars = 0
        self.ATR = []
        self.window = window
        self.ma_bars = deque(maxlen=self.window) # For SMA

    def average_true_range(self, event):
        """
        The average_true_range function calculates the ATR
        in two steps: First, it calculates the True Range
        based on the High, Low, and Previous Close price of
        a Bar Event. Next, it calculates the ATR based on the
        specified window using either an Simple Moving
        Average or an Exponential Moving Average.

        The output can then be used to calculate ATR based
        Trailing Stops.
        """
        self.bars += 1
        if self.bars > 1:
            self.y_close.append(event.close_price)

            y_close = self.y_close[-2]
            high = event.high_price
            low = event.low_price

            x = high - low
            if y_close == 0:
                y = 0
                z = 0
            else:
                y = abs(high - y_close)
                z = abs(low - y_close)

            true_range = max(x, y, z)
            self.ma_bars.append(PriceParser.display(true_range))

            if self.bars > self.window:
                if len(self.ATR) == 0:
                    # First ATR (SMA)
                    ATR = np.mean(self.ma_bars)
                    self.ATR.append(ATR)

            if self.bars > self.window + 1:
                if len(self.ATR) > 0:
                    # Subsequent ATR (EMA)
                    previous = self.ATR[-1]
                    multiplier = 2 / (self.window + 1)
                    price = PriceParser.display(true_range)
                    ATR = multiplier * (price - previous) + previous
                    self.ATR.append(ATR)

Like I mentioned earlier, when compared to TradingView's ATR indicator, the values of my ATR decreases to 1-2 pips on the SPY after approximately 252 trading days. I've added a lookback in calculate_signals() in order for the indicator to be as accurate as possible before any backtesting commences.