Merge pull request #8 from twopirllc/trend-indicators

Trend indicators
This commit is contained in:
Kevin Johnson
2019-03-01 14:02:16 -08:00
committed by GitHub
8 changed files with 714 additions and 14 deletions
+19
View File
@@ -22,6 +22,12 @@ A [Pandas DataFrame Extension](https://pandas.pydata.org/pandas-docs/stable/exte
# Getting Started and Examples
## Installation (python 3)
```sh
$ pip install pandas_ta
```
## **Quick Start** using the DataFrame Extension
```python
@@ -146,6 +152,19 @@ Use parameter: cumulative=**True** for cumulative results.
|:--------:|
| ![Example Z Score](/images/SPY_ZScore.png) |
## _Trend_ (6)
* _Average Directional Movement Index_: **adx**
* _Aroon Oscillator_: **aroon**
* _Decreasing_: **decreasing**
* _Detrended Price Oscillator_: **dpo**
* _Increasing_: **increasing**
* _Vortex Indicator_: **vortex**
| _Average Directional Movement Index_ (ADX) |
|:--------:|
| ![Example ADX](/images/SPY_ADX.png) |
## _Volatility_ (8)
* _Acceleration Bands_: **accbands**
+45 -3
View File
@@ -7,6 +7,7 @@ from .momentum import *
from .overlap import *
from .performance import *
from .statistics import *
from .trend import *
from .utils import *
from .volatility import *
from .volume import *
@@ -551,6 +552,49 @@ class AnalysisIndicators(BasePandasObject):
# Trend Indicators
def adx(self, high=None, low=None, close=None, drift=None, offset=None, **kwargs):
high = self._get_column(high, 'high')
low = self._get_column(low, 'low')
close = self._get_column(close, 'close')
result = adx(high=high, low=low, close=close, drift=drift, offset=offset, **kwargs)
self._append(result, **kwargs)
return result
def aroon(self, close=None, length=None, offset=None, **kwargs):
close = self._get_column(close, 'close')
result = aroon(close=close, length=length, offset=offset, **kwargs)
self._append(result, **kwargs)
return result
def decreasing(self, close=None, length=None, asint=True, offset=None, **kwargs):
close = self._get_column(close, 'close')
result = decreasing(close=close, length=length, asint=asint, offset=offset, **kwargs)
self._append(result, **kwargs)
return result
def dpo(self, close=None, length=None, centered=True, offset=None, **kwargs):
close = self._get_column(close, 'close')
result = dpo(close=close, length=length, centered=centered, offset=offset, **kwargs)
self._append(result, **kwargs)
return result
def increasing(self, close=None, length=None, asint=True, offset=None, **kwargs):
close = self._get_column(close, 'close')
result = increasing(close=close, length=length, asint=asint, offset=offset, **kwargs)
self._append(result, **kwargs)
return result
def vortex(self, high=None, low=None, close=None, drift=None, offset=None, **kwargs):
high = self._get_column(high, 'high')
low = self._get_column(low, 'low')
close = self._get_column(close, 'close')
result = vortex(high=high, low=low, close=close, drift=drift, offset=offset, **kwargs)
self._append(result, **kwargs)
return result
# Volatility Indicators
def accbands(self, high=None, low=None, close=None, length=None, c=None, mamode=None, offset=None, **kwargs):
high = self._get_column(high, 'high')
@@ -703,6 +747,4 @@ class AnalysisIndicators(BasePandasObject):
def vp(self, close=None, volume=None, width=None, percent=None, **kwargs):
close = self._get_column(close, 'close')
volume = self._get_column(volume, 'volume')
result = vp(close=close, volume=volume, width=width, percent=percent, **kwargs)
self._append(result, **kwargs)
return result
return vp(close=close, volume=volume, width=width, percent=percent, **kwargs)
+497
View File
@@ -0,0 +1,497 @@
# -*- coding: utf-8 -*-
import numpy as np
import pandas as pd
from .momentum import roc
from .overlap import ema, midprice, rma
from .utils import get_drift, get_offset, verify_series, zero
from .volatility import atr, true_range
def adx(high, low, close, length=None, drift=None, offset=None, **kwargs):
"""Indicator: ADX"""
# Validate Arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
length = length if length and length > 0 else 14
drift = get_drift(drift)
offset = get_offset(offset)
# Calculate Result
_atr = atr(high=high, low=low, close=close, length=length)
up = high - high.shift(drift)
dn = low.shift(drift) - low
pos = ((up > dn) & (up > 0)) * up
neg = ((dn > up) & (dn > 0)) * dn
pos = pos.apply(zero)
neg = neg.apply(zero)
dmp = (100 / _atr) * rma(close=pos, length=length)
dmn = (100 / _atr) * rma(close=neg, length=length)
dx = 100 * (dmp - dmn).abs() / (dmp + dmn)
adx = rma(close=dx, length=length)
# Offset
if offset != 0:
dmp = dmp.shift(offset)
dmn = dmn.shift(offset)
adx = adx.shift(offset)
# Handle fills
if 'fillna' in kwargs:
adx.fillna(kwargs['fillna'], inplace=True)
dmp.fillna(kwargs['fillna'], inplace=True)
dmn.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
adx.fillna(method=kwargs['fill_method'], inplace=True)
dmp.fillna(method=kwargs['fill_method'], inplace=True)
dmn.fillna(method=kwargs['fill_method'], inplace=True)
# Name and Categorize it
adx.name = f"ADX_{length}"
dmp.name = f"DMP_{length}"
dmn.name = f"DMN_{length}"
adx.category = dmp.category = dmn.category = 'trend'
# Prepare DataFrame to return
data = {adx.name: adx, dmp.name: dmp, dmn.name: dmn}
adxdf = pd.DataFrame(data)
adxdf.name = f"ADX_{length}"
adxdf.category = 'trend'
return adxdf
def aroon(close, length=None, offset=None, **kwargs):
"""Indicator: Aroon Oscillator"""
# Validate Arguments
close = verify_series(close)
length = length if length and length > 0 else 14
min_periods = int(kwargs['min_periods']) if 'min_periods' in kwargs and kwargs['min_periods'] is not None else length
offset = get_offset(offset)
# Calculate Result
def maxidx(x):
return 100 * (int(np.argmax(x)) + 1) / length
def minidx(x):
return 100 * (int(np.argmin(x)) + 1) / length
_close = close.rolling(length, min_periods=min_periods)
aroon_up = _close.apply(maxidx, raw=True)
aroon_down = _close.apply(minidx, raw=True)
# Handle fills
if 'fillna' in kwargs:
aroon_up.fillna(kwargs['fillna'], inplace=True)
aroon_down.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
aroon_up.fillna(method=kwargs['fill_method'], inplace=True)
aroon_down.fillna(method=kwargs['fill_method'], inplace=True)
# Offset
if offset != 0:
aroon_up = aroon_up.shift(offset)
aroon_down = aroon_down.shift(offset)
# Name and Categorize it
aroon_up.name = f"AROONU_{length}"
aroon_down.name = f"AROOND_{length}"
aroon_down.category = aroon_up.category = 'trend'
# Prepare DataFrame to return
data = {aroon_down.name: aroon_down, aroon_up.name: aroon_up}
aroondf = pd.DataFrame(data)
aroondf.name = f"AROON_{length}"
aroondf.category = 'trend'
return aroondf
def decreasing(close, length=None, asint=True, offset=None, **kwargs):
"""Indicator: Decreasing"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 1
offset = get_offset(offset)
# Calculate Result
decreasing = close.diff(length) < 0
if asint:
decreasing = decreasing.astype(int)
# Offset
if offset != 0:
decreasing = decreasing.shift(offset)
# Handle fills
if 'fillna' in kwargs:
decreasing.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
decreasing.fillna(method=kwargs['fill_method'], inplace=True)
# Name and Categorize it
decreasing.name = f"DEC_{length}"
decreasing.category = 'trend'
return decreasing
def dpo(close, length=None, centered=True, offset=None, **kwargs):
"""Indicator: Detrend Price Oscillator (DPO)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 1
min_periods = int(kwargs['min_periods']) if 'min_periods' in kwargs and kwargs['min_periods'] is not None else length
offset = get_offset(offset)
# Calculate Result
drift = int(0.5 * length) + 1 # int((0.5 * length) + 1)
dpo = close.shift(drift) - close.rolling(length, min_periods=min_periods).mean()
# dpo = close.shift(drift) - close.rolling(length).mean()
if centered:
dpo = dpo.shift(-drift)
# Offset
if offset != 0:
dpo = dpo.shift(offset)
# Handle fills
if 'fillna' in kwargs:
dpo.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
dpo.fillna(method=kwargs['fill_method'], inplace=True)
# Name and Categorize it
dpo.name = f"DPO_{length}"
dpo.category = 'trend'
return dpo
def increasing(close, length=None, asint=True, offset=None, **kwargs):
"""Indicator: Increasing"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 1
offset = get_offset(offset)
# Calculate Result
increasing = close.diff(length) > 0
if asint:
increasing = increasing.astype(int)
# Offset
if offset != 0:
increasing = increasing.shift(offset)
# Handle fills
if 'fillna' in kwargs:
increasing.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
increasing.fillna(method=kwargs['fill_method'], inplace=True)
# Name and Categorize it
increasing.name = f"INC_{length}"
increasing.category = 'trend'
return increasing
def vortex(high, low, close, length=None, drift=None, offset=None, **kwargs):
"""Indicator: Vortex"""
# Validate arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
length = length if length and length > 0 else 14
min_periods = int(kwargs['min_periods']) if 'min_periods' in kwargs and kwargs['min_periods'] is not None else length
drift = get_drift(drift)
offset = get_offset(offset)
# Calculate Result
tr = true_range(high=high, low=low, close=close)
tr_sum = tr.rolling(length, min_periods=min_periods).sum()
vmp = (high - low.shift(drift)).abs()
vmm = (low - high.shift(drift)).abs()
vip = vmp.rolling(length, min_periods=min_periods).sum() / tr_sum
vim = vmm.rolling(length, min_periods=min_periods).sum() / tr_sum
# Offset
if offset != 0:
vip = vip.shift(offset)
vim = vim.shift(offset)
# Handle fills
if 'fillna' in kwargs:
vip.fillna(kwargs['fillna'], inplace=True)
vim.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
vip.fillna(method=kwargs['fill_method'], inplace=True)
vim.fillna(method=kwargs['fill_method'], inplace=True)
# Name and Categorize it
vip.name = f"VTXP_{length}"
vim.name = f"VTXM_{length}"
vip.category = vim.category = 'trend'
# Prepare DataFrame to return
data = {vip.name: vip, vim.name: vim}
vtxdf = pd.DataFrame(data)
vtxdf.name = f"VTX_{length}"
vtxdf.category = 'trend'
return vtxdf
# Trend Documentation
adx.__doc__ = \
"""Average Directional Movement (ADX)
Average Directional Movement is meant to quantify trend strength by measuring
the amount of movement in a single direction.
Sources:
https://www.tradingtechnologies.com/help/x-study/technical-indicator-definitions/average-directional-movement-adx/
Calculation:
DMI ADX TREND 2.0 by @TraderR0BERT, NETWORTHIE.COM
//Created by @TraderR0BERT, NETWORTHIE.COM, last updated 01/26/2016
//DMI Indicator
//Resolution input option for higher/lower time frames
study(title="DMI ADX TREND 2.0", shorttitle="ADX TREND 2.0")
adxlen = input(14, title="ADX Smoothing")
dilen = input(14, title="DI Length")
thold = input(20, title="Threshold")
threshold = thold
//Script for Indicator
dirmov(len) =>
up = change(high)
down = -change(low)
truerange = rma(tr, len)
plus = fixnan(100 * rma(up > down and up > 0 ? up : 0, len) / truerange)
minus = fixnan(100 * rma(down > up and down > 0 ? down : 0, len) / truerange)
[plus, minus]
adx(dilen, adxlen) =>
[plus, minus] = dirmov(dilen)
sum = plus + minus
adx = 100 * rma(abs(plus - minus) / (sum == 0 ? 1 : sum), adxlen)
[adx, plus, minus]
[sig, up, down] = adx(dilen, adxlen)
osob=input(40,title="Exhaustion Level for ADX, default = 40")
col = sig >= sig[1] ? green : sig <= sig[1] ? red : gray
//Plot Definitions Current Timeframe
p1 = plot(sig, color=col, linewidth = 3, title="ADX")
p2 = plot(sig, color=col, style=circles, linewidth=3, title="ADX")
p3 = plot(up, color=blue, linewidth = 3, title="+DI")
p4 = plot(up, color=blue, style=circles, linewidth=3, title="+DI")
p5 = plot(down, color=fuchsia, linewidth = 3, title="-DI")
p6 = plot(down, color=fuchsia, style=circles, linewidth=3, title="-DI")
h1 = plot(threshold, color=black, linewidth =3, title="Threshold")
trender = (sig >= up or sig >= down) ? 1 : 0
bgcolor(trender>0?black:gray, transp=85)
//Alert Function for ADX crossing Threshold
Up_Cross = crossover(up, threshold)
alertcondition(Up_Cross, title="DMI+ cross", message="DMI+ Crossing Threshold")
Down_Cross = crossover(down, threshold)
alertcondition(Down_Cross, title="DMI- cross", message="DMI- Crossing Threshold")
Args:
high (pd.Series): Series of 'high's
low (pd.Series): Series of 'low's
close (pd.Series): Series of 'close's
length (int): It's period. Default: 14
drift (int): The difference period. Default: 1
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.DataFrame: adx, dmp, dmn columns.
"""
aroon.__doc__ = \
"""Aroon (AROON)
Aroon attempts to identify if a security is trending and how strong.
Sources:
https://www.tradingview.com/wiki/Aroon
https://www.tradingtechnologies.com/help/x-study/technical-indicator-definitions/aroon-ar/
Calculation:
Default Inputs:
length=1
def maxidx(x):
return 100 * (int(np.argmax(x)) + 1) / length
def minidx(x):
return 100 * (int(np.argmin(x)) + 1) / length
_close = close.rolling(length, min_periods=min_periods)
aroon_up = _close.apply(maxidx, raw=True)
aroon_down = _close.apply(minidx, raw=True)
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 1
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.DataFrame: aroon_up, aroon_down columns.
"""
decreasing.__doc__ = \
"""Decreasing
Returns True or False if the series is decreasing over a periods. By default,
it returns True and False as 1 and 0 respectively with kwarg 'asint'.
Sources:
Calculation:
decreasing = close.diff(length) < 0
if asint:
decreasing = decreasing.astype(int)
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 1
asint (bool): Returns as binary. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.Series: New feature generated.
"""
dpo.__doc__ = \
"""Detrend Price Oscillator (DPO)
Is an indicator designed to remove trend from price and make it easier to
identify cycles.
Sources:
http://stockcharts.com/school/doku.php?id=chart_school:technical_indicators:detrended_price_osci
Calculation:
Default Inputs:
length=1, centered=True
SMA = Simple Moving Average
drift = int(0.5 * length) + 1
DPO = close.shift(drift) - SMA(close, length)
if centered:
DPO = DPO.shift(-drift)
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 1
centered (bool): Shift the dpo back by int(0.5 * length) + 1. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.Series: New feature generated.
"""
increasing.__doc__ = \
"""Increasing
Returns True or False if the series is increasing over a periods. By default,
it returns True and False as 1 and 0 respectively with kwarg 'asint'.
Sources:
Calculation:
increasing = close.diff(length) > 0
if asint:
increasing = increasing.astype(int)
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 1
asint (bool): Returns as binary. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.Series: New feature generated.
"""
vortex.__doc__ = \
"""Vortex
Two oscillators that capture positive and negative trend movement.
Sources:
https://stockcharts.com/school/doku.php?id=chart_school:technical_indicators:vortex_indicator
Calculation:
Default Inputs:
length=14, drift=1
TR = True Range
SMA = Simple Moving Average
tr = TR(high, low, close)
tr_sum = tr.rolling(length).sum()
vmp = (high - low.shift(drift)).abs()
vmn = (low - high.shift(drift)).abs()
VIP = vmp.rolling(length).sum() / tr_sum
VIM = vmn.rolling(length).sum() / tr_sum
Args:
high (pd.Series): Series of 'high's
low (pd.Series): Series of 'low's
close (pd.Series): Series of 'close's
length (int): ROC 1 period. Default: 14
drift (int): The difference period. Default: 1
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.DataFrame: vip and vim columns
"""
-7
View File
@@ -1,11 +1,4 @@
# -*- coding: utf-8 -*-
"""
.. module:: volume
:synopsis: Volume Indicators.
.. moduleauthor:: Dario Lopez Padial (Bukosabino)
"""
import numpy as np
import pandas as pd
+1 -1
View File
@@ -6,7 +6,7 @@ long_description = "An easy to use Python 3 Pandas Extension of Technical Analys
setup(
name = "pandas_ta",
packages = ["pandas_ta"],
version = "0.0.9a",
version = "0.1.0a",
description=long_description,
long_description=long_description,
author = "Kevin Johnson",
+94
View File
@@ -0,0 +1,94 @@
from .config import error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD, VERBOSE
from .context import pandas_ta
from unittest import TestCase, skip
import pandas.util.testing as pdt
from pandas import DataFrame, Series
import talib as tal
class TestTrend(TestCase):
@classmethod
def setUpClass(cls):
cls.data = sample_data
cls.open = cls.data['open']
cls.high = cls.data['high']
cls.low = cls.data['low']
cls.close = cls.data['close']
cls.volume = cls.data['volume']
@classmethod
def tearDownClass(cls):
del cls.data
del cls.open
del cls.high
del cls.low
del cls.close
del cls.volume
def setUp(self):
self.trend = pandas_ta.trend
def tearDown(self):
del self.trend
def test_adx(self):
result = self.trend.adx(self.high, self.low, self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'ADX_14')
try:
expected = tal.ADX(self.high, self.low, self.close)
pdt.assert_series_equal(result.iloc[:,0], expected)
except AssertionError as ae:
try:
corr = pandas_ta.utils.df_error_analysis(result.iloc[:,0], expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
def test_aroon(self):
result = self.trend.aroon(self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'AROON_14')
try:
expected = tal.AROON(self.high, self.low)
expecteddf = DataFrame({'AROOND_14': expected[0], 'AROONU_14': expected[1]})
pdt.assert_frame_equal(result, expecteddf)
except AssertionError as ae:
try:
aroond_corr = pandas_ta.utils.df_error_analysis(result.iloc[:,0], expecteddf.iloc[:,0], col=CORRELATION)
self.assertGreater(aroond_corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result.iloc[:,0], CORRELATION, ex)
try:
aroonu_corr = pandas_ta.utils.df_error_analysis(result.iloc[:,1], expecteddf.iloc[:,1], col=CORRELATION)
self.assertGreater(aroonu_corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result.iloc[:,1], CORRELATION, ex, newline=False)
def test_decreasing(self):
result = self.trend.decreasing(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'DEC_1')
def test_dpo(self):
result = self.trend.dpo(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'DPO_1')
def test_increasing(self):
result = self.trend.increasing(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'INC_1')
def test_vortex(self):
result = self.trend.vortex(self.high, self.low, self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'VTX_14')
+54
View File
@@ -0,0 +1,54 @@
from .config import sample_data
from .context import pandas_ta
from unittest import skip, TestCase
from pandas import DataFrame
class TestTrendExtension(TestCase):
@classmethod
def setUpClass(cls):
cls.data = sample_data
@classmethod
def tearDownClass(cls):
del cls.data
def setUp(self):
pass
def tearDown(self):
pass
def test_adx_ext(self):
self.data.ta.adx(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-3:]), ['ADX_14', 'DMP_14', 'DMN_14'])
def test_aroon_ext(self):
self.data.ta.aroon(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-2:]), ['AROOND_14', 'AROONU_14'])
def test_decreasing_ext(self):
self.data.ta.decreasing(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'DEC_1')
def test_dpo_ext(self):
self.data.ta.dpo(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'DPO_1')
def test_increasing_ext(self):
self.data.ta.increasing(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'INC_1')
def test_vortext_ext(self):
self.data.ta.vortex(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-2:]), ['VTXP_14', 'VTXM_14'])
+4 -3
View File
@@ -1,7 +1,7 @@
from .config import sample_data
from .context import pandas_ta
from unittest import skip, TestCase
from unittest import TestCase
from pandas import DataFrame
@@ -73,6 +73,7 @@ class TestVolumeExtension(TestCase):
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'PVT')
@skip('Standalone and does not need to be added to the DataFrame')
def test_vp_ext(self):
pass
result = self.data.ta.vp()
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'VP_10')