performance refactor and amat indicator added

This commit is contained in:
Kevin Johnson
2019-05-18 08:40:06 -07:00
parent a069323953
commit 8a54aed07c
17 changed files with 620 additions and 409 deletions
+4 -1
View File
@@ -132,4 +132,7 @@ ta_extension.ipynb
Charts.ipynb
pandas_pips
reqs.txt
requirements.txt
requirements.txt
qd.py
_performance.py
simple.ipynb
+2 -1
View File
@@ -163,9 +163,10 @@ Use parameter: cumulative=**True** for cumulative results.
|:--------:|
| ![Example Z Score](/images/SPY_ZScore.png) |
## _Trend_ (9)
## _Trend_ (10)
* _Average Directional Movement Index_: **adx**
* _Archer Moving Averages Trends_: **amat**
* _Aroon Oscillator_: **aroon**
* _Decreasing_: **decreasing**
* _Detrended Price Oscillator_: **dpo**
+281 -174
View File
File diff suppressed because one or more lines are too long
+22
View File
@@ -2,4 +2,26 @@ name = "pandas_ta"
"""
.. moduleauthor:: Kevin Johnson
"""
from pkg_resources import get_distribution, DistributionNotFound
import os.path
try:
_dist = get_distribution('pandas_ta')
# Normalize case for Windows systems
dist_loc = os.path.normcase(_dist.location)
here = os.path.normcase(__file__)
if not here.startswith(os.path.join(dist_loc, 'pandas_ta')):
# not installed, but there is another version that *is*
raise DistributionNotFound
except DistributionNotFound:
__version__ = 'Please install this project with setup.py'
else:
__version__ = _dist.version
# Performance
from .performance.log_return import log_return
from .performance.percent_return import percent_return
from .performance.trend_return import trend_return
# DataFrame Extension
from .core import *
+11 -6
View File
@@ -5,15 +5,12 @@ from pandas.core.base import PandasObject
from .momentum import *
from .overlap import *
from .performance import *
from .statistics import *
from .trend import *
from .utils import *
from .volatility import *
from .volume import *
class BasePandasObject(PandasObject):
"""Simple PandasObject Extension
@@ -505,21 +502,23 @@ class AnalysisIndicators(BasePandasObject):
# Performance Indicators
def log_return(self, close=None, length=None, cumulative=False, percent=False, offset=None, **kwargs):
close = self._get_column(close, 'close')
from pandas_ta.performance.log_return import log_return
result = log_return(close=close, length=length, cumulative=cumulative, percent=percent, offset=offset, **kwargs)
self._append(result, **kwargs)
# print(f"result:\n{result}")
return result
def percent_return(self, close=None, length=None, cumulative=False, percent=False, offset=None, **kwargs):
close = self._get_column(close, 'close')
from pandas_ta.performance.percent_return import percent_return
result = percent_return(close=close, length=length, cumulative=cumulative, percent=percent, offset=offset, **kwargs)
self._append(result, **kwargs)
return result
def trend_return(self, close=None, trend=None, log=True, cumulative=True, offset=None, **kwargs):
def trend_return(self, close=None, trend=None, log=None, cumulative=None, offset=None, trend_reset=None, **kwargs):
close = self._get_column(close, 'close')
trend = self._get_column(trend, f"{trend}")
result = trend_return(close=close, trend=trend, log=log, cumulative=cumulative, offset=offset, **kwargs)
from pandas_ta.performance.trend_return import trend_return
result = trend_return(close=close, trend=trend, log=log, cumulative=cumulative, offset=offset, trend_reset=trend_reset, **kwargs)
self._append(result, **kwargs)
return result
@@ -584,6 +583,12 @@ class AnalysisIndicators(BasePandasObject):
self._append(result, **kwargs)
return result
def amat(self, close=None, fast=None, slow=None, mamode=None, lookback=None, offset=None, **kwargs):
close = self._get_column(close, 'close')
result = amat(close=close, fast=fast, slow=slow, mamode=mamode, lookback=lookback, 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)
-205
View File
@@ -1,205 +0,0 @@
# -*- coding: utf-8 -*-
import numpy as np
import pandas as pd
from .utils import get_offset, verify_series, zero
def log_return(close, length=None, cumulative=False, offset=None, **kwargs):
"""Indicator: Log Return"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 1
offset = get_offset(offset)
# Calculate Result
log_return = np.log(close).diff(periods=length)
if cumulative:
log_return = log_return.cumsum()
# Offset
if offset != 0:
log_return = log_return.shift(offset)
# Name & Category
log_return.name = f"{'CUM' if cumulative else ''}LOGRET_{length}"
log_return.category = 'performance'
return log_return
def percent_return(close, length=None, cumulative=False, offset=None, **kwargs):
"""Indicator: Percent Return"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 1
offset = get_offset(offset)
# Calculate Result
pct_return = close.pct_change(length)
if cumulative:
pct_return = pct_return.cumsum()
# Offset
if offset != 0:
pct_return = pct_return.shift(offset)
# Name & Category
pct_return.name = f"{'CUM' if cumulative else ''}PCTRET_{length}"
pct_return.category = 'performance'
return pct_return
def trend_return(close, trend, trend_reset=0, log=True, cumulative=True, offset=None, **kwargs):
"""Indicator: Trend Return"""
# Validate Arguments
close = verify_series(close)
trend = verify_series(trend)
offset = get_offset(offset)
variable = kwargs.pop('variable', True)
# Calculate Result
returns = log_return(close, cumulative=False) if log else percent_return(close, cumulative=False)
m = trend.size
tsum = 0
trend = trend.astype(int)
returns = (trend * returns).apply(zero)
result = []
for i in range(0, m):
if trend[i] == trend_reset:
tsum = 0
else:
return_ = returns[i]
if cumulative:
tsum += return_
else:
tsum = return_
result.append(tsum)
trend_return = pd.Series(result)
if variable:
trend_return += returns
# Offset
if offset != 0:
trend_return = trend_return.shift(offset)
# Name & Category
trend_return.name = f"{'C' if cumulative else ''}{'L' if log else 'P'}TR"
trend_return.category = 'performance'
return trend_return
log_return.__doc__ = \
"""Log Return
Calculates the logarithmic return of a Series.
See also: help(df.ta.log_return) for additional **kwargs a valid 'df'.
Sources:
https://stackoverflow.com/questions/31287552/logarithmic-returns-in-pandas-dataframe
Calculation:
Default Inputs:
length=1, cumulative=False
LOGRET = log( close.diff(periods=length) )
CUMLOGRET = LOGRET.cumsum() if cumulative
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 20
cumulative (bool): If True, returns the cumulative returns. Default: False
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.
"""
percent_return.__doc__ = \
"""Percent Return
Calculates the percent return of a Series.
See also: help(df.ta.percent_return) for additional **kwargs a valid 'df'.
Sources:
https://stackoverflow.com/questions/31287552/logarithmic-returns-in-pandas-dataframe
Calculation:
Default Inputs:
length=1, cumulative=False
PCTRET = close.pct_change(length)
CUMPCTRET = PCTRET.cumsum() if cumulative
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 20
cumulative (bool): If True, returns the cumulative returns. Default: False
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.
"""
trend_return.__doc__ = \
"""Trend Return
Calculates the (Cumulative) Returns of a Trend as defined by some conditional.
By default it calculates log returns but can also use percent change.
Sources: Kevin Johnson
Calculation:
Default Inputs:
trend_reset=0, log=True, cumulative=False
sum = 0
returns = log_return if log else percent_return # These are not cumulative
returns = (trend * returns).apply(zero)
for i, in range(0, trend.size):
if item == trend_reset:
sum = 0
else:
return_ = returns.iloc[i]
if cumulative:
sum += return_
else:
sum = return_
trend_return.append(sum)
if cumulative and variable:
trend_return += returns
Args:
close (pd.Series): Series of 'close's
trend (pd.Series): Series of 'trend's. Preferably 0's and 1's.
trend_reset (value): Value used to identify if a trend has ended. Default: 0
log (bool): Calculate logarithmic returns. Default: True
cumulative (bool): If True, returns the cumulative returns. Default: False
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
variable (bool, optional): Whether to include if return fluxuations in the cumulative returns.
Returns:
pd.Series: New feature generated.
"""
+5
View File
@@ -0,0 +1,5 @@
# -*- coding: utf-8 -*-
from os.path import dirname, basename, isfile, join
import glob
modules = glob.glob(join(dirname(__file__), "*.py"))
__all__ = [basename(f)[:-3] for f in modules if isfile(f) and not f.endswith('__init__.py')]
+57
View File
@@ -0,0 +1,57 @@
# -*- coding: utf-8 -*-
from numpy import log as nplog
from ..utils import get_offset, verify_series
def log_return(close, length=None, cumulative=False, offset=None, **kwargs):
"""Indicator: Log Return"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 1
offset = get_offset(offset)
# Calculate Result
log_return = nplog(close).diff(periods=length)
if cumulative:
log_return = log_return.cumsum()
# Offset
if offset != 0:
log_return = log_return.shift(offset)
# Name & Category
log_return.name = f"{'CUM' if cumulative else ''}LOGRET_{length}"
log_return.category = 'performance'
return log_return
log_return.__doc__ = \
"""Log Return
Calculates the logarithmic return of a Series.
See also: help(df.ta.log_return) for additional **kwargs a valid 'df'.
Sources:
https://stackoverflow.com/questions/31287552/logarithmic-returns-in-pandas-dataframe
Calculation:
Default Inputs:
length=1, cumulative=False
LOGRET = log( close.diff(periods=length) )
CUMLOGRET = LOGRET.cumsum() if cumulative
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 20
cumulative (bool): If True, returns the cumulative returns. Default: False
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.
"""
+56
View File
@@ -0,0 +1,56 @@
# -*- coding: utf-8 -*-
from ..utils import get_offset, verify_series
def percent_return(close, length=None, cumulative=False, offset=None, **kwargs):
"""Indicator: Percent Return"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 1
offset = get_offset(offset)
# Calculate Result
pct_return = close.pct_change(length)
if cumulative:
pct_return = pct_return.cumsum()
# Offset
if offset != 0:
pct_return = pct_return.shift(offset)
# Name & Category
pct_return.name = f"{'CUM' if cumulative else ''}PCTRET_{length}"
pct_return.category = 'performance'
return pct_return
percent_return.__doc__ = \
"""Percent Return
Calculates the percent return of a Series.
See also: help(df.ta.percent_return) for additional **kwargs a valid 'df'.
Sources:
https://stackoverflow.com/questions/31287552/logarithmic-returns-in-pandas-dataframe
Calculation:
Default Inputs:
length=1, cumulative=False
PCTRET = close.pct_change(length)
CUMPCTRET = PCTRET.cumsum() if cumulative
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 20
cumulative (bool): If True, returns the cumulative returns. Default: False
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.
"""
+92
View File
@@ -0,0 +1,92 @@
# -*- coding: utf-8 -*-
from pandas import Series
from .log_return import log_return
from .percent_return import percent_return
from ..utils import get_offset, verify_series, zero
def trend_return(close, trend, log=None, cumulative=None, offset=None, trend_reset=0, **kwargs):
"""Indicator: Trend Return"""
# Validate Arguments
close = verify_series(close)
trend = verify_series(trend)
offset = get_offset(offset)
trend_reset = int(trend_reset) if trend_reset and isinstance(trend_reset, int) else 0
# Calculate Result
returns = log_return(close, cumulative=False) if log else percent_return(close, cumulative=False)
m = trend.size
tsum = 0
trend = trend.astype(int)
returns = (trend * returns).apply(zero)
result = []
for i in range(0, m):
if trend[i] == trend_reset:
tsum = 0
else:
return_ = returns[i]
if cumulative:
tsum += return_
else:
tsum = return_
result.append(tsum)
trend_return = Series(result)
# Offset
if offset != 0:
trend_return = trend_return.shift(offset)
# Name & Category
trend_return.name = f"{'C' if cumulative else ''}{'L' if log else 'P'}TR"
trend_return.category = 'performance'
return trend_return
trend_return.__doc__ = \
"""Trend Return
Calculates the (Cumulative) Returns of a Trend as defined by some conditional.
By default it calculates log returns but can also use percent change.
Sources: Kevin Johnson
Calculation:
Default Inputs:
trend_reset=0, log=True, cumulative=False
sum = 0
returns = log_return if log else percent_return # These are not cumulative
returns = (trend * returns).apply(zero)
for i, in range(0, trend.size):
if item == trend_reset:
sum = 0
else:
return_ = returns.iloc[i]
if cumulative:
sum += return_
else:
sum = return_
trend_return.append(sum)
if cumulative and variable:
trend_return += returns
Args:
close (pd.Series): Series of 'close's
trend (pd.Series): Series of 'trend's. Preferably 0's and 1's.
trend_reset (value): Value used to identify if a trend has ended. Default: 0
log (bool): Calculate logarithmic returns. Default: True
cumulative (bool): If True, returns the cumulative returns. Default: False
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
variable (bool, optional): Whether to include if return fluxuations in the cumulative returns.
Returns:
pd.Series: New feature generated.
"""
+60
View File
@@ -69,6 +69,66 @@ def adx(high, low, close, length=None, drift=None, offset=None, **kwargs):
return adxdf
def amat(close=None, fast=None, slow=None, mamode=None, lookback=None, offset=None, **kwargs):
"""Indicator: Archer Moving Averages Trends (AMAT)"""
# Validate Arguments
close = verify_series(close)
fast = int(fast) if fast and fast > 0 else 8
slow = int(slow) if slow and slow > 0 else 21
lookback = int(lookback) if lookback and lookback > 0 else 2
mamode = mamode.upper() if mamode else 'EMA'
offset = get_offset(offset)
# Calculate Result
if mamode == 'EMA':
fast_ma = ema(close=close, length=fast, **kwargs)
slow_ma = ema(close=close, length=slow, **kwargs)
elif mamode == 'HMA':
fast_ma = hma(close=close, length=fast, **kwargs)
slow_ma = hma(close=close, length=slow, **kwargs)
elif mamode == 'LINREG':
fast_ma = linreg(close=close, length=fast, **kwargs)
slow_ma = linreg(close=close, length=slow, **kwargs)
elif mamode == 'RMA':
fast_ma = rma(close=close, length=fast, **kwargs)
slow_ma = rma(close=close, length=slow, **kwargs)
elif mamode == 'SMA':
fast_ma = sma(close=close, length=fast, **kwargs)
slow_ma = sma(close=close, length=slow, **kwargs)
elif mamode == 'WMA':
fast_ma = wma(close=close, length=fast, **kwargs)
slow_ma = wma(close=close, length=slow, **kwargs)
mas_long = long_run(fast_ma, slow_ma, length=lookback)
mas_short = short_run(fast_ma, slow_ma, length=lookback)
# Offset
if offset != 0:
mas_long = mas_long.shift(offset)
mas_short = mas_short.shift(offset)
# # Handle fills
if 'fillna' in kwargs:
mas_long.fillna(kwargs['fillna'], inplace=True)
mas_short.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
mas_long.fillna(method=kwargs['fill_method'], inplace=True)
mas_short.fillna(method=kwargs['fill_method'], inplace=True)
# Prepare DataFrame to return
amatdf = pd.DataFrame({
f"AMAT_{mas_long.name}": mas_long,
f"AMAT_{mas_short.name}": mas_short
})
# Name and Categorize it
amatdf.name = f"AMAT_{mamode}_{fast}_{slow}_{lookback}"
amatdf.category = 'trend'
return amatdf
def aroon(close, length=None, offset=None, **kwargs):
"""Indicator: Aroon Oscillator"""
# Validate Arguments
+5 -5
View File
@@ -27,7 +27,7 @@ def combination(**kwargs):
return numerator // denominator
def cross(series_a, series_b, above=True, asint=True, offset=None, **kwargs):
def cross(series_a:pd.Series, series_b:pd.Series, above:bool =True, asint:bool =True, offset:int =None, **kwargs):
series_a = verify_series(series_a)
series_b = verify_series(series_b)
offset = get_offset(offset)
@@ -55,7 +55,7 @@ def cross(series_a, series_b, above=True, asint=True, offset=None, **kwargs):
return cross
def df_error_analysis(dfA, dfB, **kwargs):
def df_error_analysis(dfA:pd.DataFrame, dfB:pd.DataFrame, **kwargs):
""" """
col = kwargs.pop('col', None)
corr_method = kwargs.pop('corr_method', 'pearson')
@@ -115,7 +115,7 @@ def get_offset(x:int):
return int(x) if x else 0
def pascals_triangle(n=None, **kwargs):
def pascals_triangle(n:int =None, **kwargs):
"""Pascal's Triangle
Returns a numpy array of the nth row of Pascal's Triangle.
@@ -143,7 +143,7 @@ def pascals_triangle(n=None, **kwargs):
return triangle
def signed_series(series:pd.Series, initial:int = None):
def signed_series(series:pd.Series, initial:int =None):
"""Returns a Signed Series with or without an initial value"""
series = verify_series(series)
sign = series.diff(1)
@@ -153,7 +153,7 @@ def signed_series(series:pd.Series, initial:int = None):
return sign
def symmetric_triangle(n=None, **kwargs):
def symmetric_triangle(n:int =None, **kwargs):
n = int(math.fabs(n)) if n is not None else 2
weighted = kwargs.pop('weighted', False)
+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.1.17b",
version = "0.1.20b",
description=long_description,
long_description=long_description,
author = "Kevin Johnson",
+1 -1
View File
@@ -15,5 +15,5 @@ sample_data = read_csv(f"data/SPY_D.csv", index_col=0, parse_dates=True, infer_d
def error_analysis(df, kind, msg, icon=INFO, newline=True):
if VERBOSE:
s = f" {icon} {df.name}['{kind}']: {msg}"
if newline: s = '\n' + s
if newline: s = f"\n{s}"
print(s)
+13 -15
View File
@@ -20,51 +20,49 @@ class TestPerformace(TestCase):
del cls.islong
def setUp(self):
self.performance = pandas_ta.performance
def tearDown(self):
del self.performance
def setUp(self): pass
def tearDown(self): pass
def test_log_return(self):
result = self.performance.log_return(self.close)
result = pandas_ta.log_return(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'LOGRET_1')
def test_cum_log_return(self):
result = self.performance.log_return(self.close, cumulative=True)
result = pandas_ta.log_return(self.close, cumulative=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'CUMLOGRET_1')
def test_percent_return(self):
result = self.performance.percent_return(self.close)
result = pandas_ta.percent_return(self.close, cumulative=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'PCTRET_1')
def test_cum_percent_return(self):
result = self.performance.percent_return(self.close, cumulative=True)
result = pandas_ta.percent_return(self.close, cumulative=True)
self.assertEqual(result.name, 'CUMPCTRET_1')
def test_log_trend_return(self):
result = self.performance.trend_return(self.close, self.islong, log=True, cumulative=False)
result = pandas_ta.trend_return(self.close, self.islong, log=True, cumulative=False)
self.assertEqual(result.name, 'LTR')
def test_cum_log_trend_return(self):
result = self.performance.trend_return(self.close, self.islong, log=True, cumulative=True)
result = pandas_ta.trend_return(self.close, self.islong, log=True, cumulative=True)
self.assertEqual(result.name, 'CLTR')
def test_variable_cum_log_trend_return(self):
result = self.performance.trend_return(self.close, self.islong, log=True, cumulative=True, variable=True)
result = pandas_ta.trend_return(self.close, self.islong, log=True, cumulative=True, variable=True)
self.assertEqual(result.name, 'CLTR')
def test_pct_trend_return(self):
result = self.performance.trend_return(self.close, self.islong, log=False, cumulative=False)
result = pandas_ta.trend_return(self.close, self.islong, log=False, cumulative=False)
self.assertEqual(result.name, 'PTR')
def test_cum_pct_trend_return(self):
result = self.performance.trend_return(self.close, self.islong, log=False, cumulative=True)
result = pandas_ta.trend_return(self.close, self.islong, log=False, cumulative=True)
self.assertEqual(result.name, 'CPTR')
def test_variable_pct_log_trend_return(self):
result = self.performance.trend_return(self.close, self.islong, log=False, cumulative=True, variable=True)
result = pandas_ta.trend_return(self.close, self.islong, log=False, cumulative=True, variable=True)
self.assertEqual(result.name, 'CPTR')
+5
View File
@@ -52,6 +52,11 @@ class TestTrend(TestCase):
except Exception as ex:
error_analysis(result, CORRELATION, ex)
def test_amat(self):
result = self.trend.amat(self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'AMAT_EMA_8_21_2')
def test_aroon(self):
result = self.trend.aroon(self.close)
self.assertIsInstance(result, DataFrame)
+5
View File
@@ -28,6 +28,11 @@ class TestTrendExtension(TestCase):
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-3:]), ['ADX_14', 'DMP_14', 'DMN_14'])
def test_amat_ext(self):
self.data.ta.amat(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-2:]), ['AMAT_LR_2', 'AMAT_SR_2'])
def test_aroon_ext(self):
self.data.ta.aroon(append=True)
self.assertIsInstance(self.data, DataFrame)