wassname/pandas-ta
1
0
Fork 0
mirror of https://github.com/wassname/pandas-ta.git synced 2026-08-11 11:22:48 +08:00
Code Issues Packages Projects Releases Wiki Activity

ENH #499 numpy improv ENH #501 smooth_k MAINT sma, supertrend, true_range, atr

Browse Source
This commit is contained in:
Kevin Johnson
2022-03-20 08:23:00 -07:00
parent c9352260ed
commit 01e8a3547c
15 changed files with 676 additions and 656 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+5 -4
View File
File diff suppressed because one or more lines are too long
examples/Speed_Test.ipynb
+541 -524
View File
File diff suppressed because it is too large Load Diff
pandas_ta/core.py
+2 -2
View File
@@ -1583,11 +1583,11 @@ class AnalysisIndicators(object):
result = atr(high=high, low=low, close=close, length=length, mamode=mamode, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def atrts(self, length=None, factor=None, mamode=None, offset: Int = None, **kwargs: DictLike):
def atrts(self, length=None, ma_length=None, multiplier=None, mamode=None, talib=None, drift=None, offset: Int = None, **kwargs: DictLike):
high = self._get_column(kwargs.pop("high", "high"))
low = self._get_column(kwargs.pop("low", "low"))
close = self._get_column(kwargs.pop("close", "close"))
result = atrts(high=high, low=low, close=close, length=length, factor=factor, mamode=mamode, offset=offset, **kwargs)
result = atrts(high=high, low=low, close=close, length=length, ma_length=ma_length, multiplier=multiplier, mamode=mamode, talib=talib, drift=drift, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def bbands(self, length=None, std=None, mamode=None, offset: Int = None, **kwargs: DictLike):
pandas_ta/momentum/stoch.py
+1 -1
View File
@@ -65,7 +65,7 @@ def stoch(
offset = v_offset(offset)
# Calculate
if Imports["talib"] and mode_tal:
if Imports["talib"] and mode_tal and smooth_k > 2:
from talib import STOCH
stoch_ = STOCH(
high, low, close, k, d, tal_ma(mamode), d, tal_ma(mamode)
pandas_ta/overlap/sma.py
+1 -1
View File
@@ -76,7 +76,7 @@ def sma(
offset = v_offset(offset)
# Calculate
if Imports["talib"] and mode_tal:
if Imports["talib"] and mode_tal and length > 1:
from talib import SMA
sma = SMA(close, length)
else:
pandas_ta/overlap/supertrend.py
+4 -3
View File
@@ -44,7 +44,6 @@ def supertrend(
low = v_series(low, length)
close = v_series(close, length)
if high is None or low is None or close is None:
return
@@ -58,9 +57,8 @@ def supertrend(
hl2_ = hl2(high, low)
matr = multiplier * atr(high, low, close, length)
ub = hl2_ + matr # Upperband
lb = hl2_ - matr # Lowerband
ub = hl2_ + matr # Upperband
for i in range(1, m):
if close.iloc[i] > ub.iloc[i - 1]:
dir_[i] = 1
@@ -78,6 +76,9 @@ def supertrend(
else:
trend[i] = short[i] = ub.iloc[i]
trend[0] = nan
dir_[:length] = [nan] * length
_props = f"_{length}_{multiplier}"
df = DataFrame({
f"SUPERT{_props}": trend,
pandas_ta/statistics/tos_stdevall.py
+7 -7
View File
@@ -38,13 +38,6 @@ def tos_stdevall(
mulitples of the standard deviation. Default: returns 7 columns.
"""
# Validate
stds = v_list(stds, [1, 2, 3])
if min(stds) <= 0:
return
if not all(i < j for i, j in zip(stds, stds[1:])):
stds = stds[::-1]
_props = f"TOS_STDEVALL"
if length is None:
length = close.size
@@ -58,6 +51,13 @@ def tos_stdevall(
if close is None:
return
stds = v_list(stds, [1, 2, 3])
if min(stds) <= 0:
return
if not all(i < j for i, j in zip(stds, stds[1:])):
stds = stds[::-1]
ddof = int(ddof) if isinstance(ddof, int) and 0 <= ddof < length else 1
offset = v_offset(offset)
pandas_ta/utils/_core.py
+1 -1
View File
@@ -144,7 +144,7 @@ def _speed_group(
for i in group:
r = df.ta(i, talib=talib, timed=True)
ms = float(r.timed.split(" ")[0].split(" ")[0])
times.append({index_name: i, "secs": ms2secs(ms, p), "ms": ms})
times.append({index_name: i, "ms": ms, "secs": ms2secs(ms, p)})
return times
pandas_ta/volatility/__init__.py
+1 -1
View File
@@ -2,6 +2,7 @@
from .aberration import aberration
from .accbands import accbands
from .atr import atr
from .atrts import atrts
from .bbands import bbands
from .donchian import donchian
from .hwc import hwc
@@ -13,4 +14,3 @@ from .rvi import rvi
from .thermo import thermo
from .true_range import true_range
from .ui import ui
from .atrts import atrts
pandas_ta/volatility/atr.py
+3 -3
View File
@@ -64,8 +64,8 @@ def atr(
)
atr = ma(mamode, tr, length=length, talib=mode_tal)
percentage = kwargs.pop("percent", False)
if percentage:
percent = kwargs.pop("percent", False)
if percent:
atr *= 100 / close
# Offset
@@ -79,7 +79,7 @@ def atr(
atr.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Category
atr.name = f"ATR{mamode[0]}_{length}{'p' if percentage else ''}"
atr.name = f"ATR{mamode[0]}{'p' if percent else ''}_{length}"
atr.category = "volatility"
return atr
pandas_ta/volatility/atrts.py
+95 -105
View File
@@ -1,101 +1,85 @@
# -*- coding: utf-8 -*-
from .true_range import true_range
from pandas_ta import Imports
from pandas_ta.overlap import ma
from numpy import nan, uintc, zeros_like
from pandas import Series
from pandas_ta._typing import Array, DictLike, Int, IntFloat
from pandas_ta.ma import ma
from pandas_ta.maps import Imports
from pandas_ta.utils import v_drift, v_mamode, v_offset
from pandas_ta.utils import v_pos_default, v_series, v_talib
from pandas_ta.volatility import atr
from pandas_ta.utils import get_drift, get_offset, verify_series
from pandas import DataFrame, Series
from functools import partial
try:
from numba import njit
except ImportError:
def njit(_): return _
@njit
def calculateFunc(upTrend, dnTrend, prevP, atr, factor):
if upTrend:
return prevP - atr * factor
elif dnTrend:
return prevP + atr * factor
@njit
def tailingStopFunc(upTrend, dnTrend, prevA, atrts):
if upTrend:
if atrts < prevA: return prevA
elif dnTrend:
if atrts > prevA: return prevA
def np_atrts(x: Array, ma_: Array, atr_: Array, length: Int, ma_length: Int):
m = x.size
k = max(length, ma_length)
def atrts(high, low, close, length=None, factor=None, mamode=None, talib=None, drift=None, offset=None, **kwargs):
"""ATR Trailing Stops (ATRTS)
identifies exit points for long and short positions.
First, an exponential moving average (EMA) of the input is taken to determine the current trend.
Then, the Average True Range (ATR) is calculated and multiplied by a user defined factor.
If the EMA is increasing (uptrend), the ATR product is subtracted from the price or,
if the EMA is decreasing (down trend), it is added to the price, and along with a few details the ATRTS is formed.
The user may change the position (long), input (close), method (EMA), period lengths,
percent factor and show entry option(see trading signals below).
This indicator’s definition is further expressed in the condensed code given in the calculation below.
result = x.copy()
up = zeros_like(x, dtype=uintc)
dn = zeros_like(x, dtype=uintc)
expn = x > ma_
up[expn], dn[~expn] = 1, 1
up[:k], dn[:k] = 0, 0
result[:k] = nan
for i in range(k, m):
pr = result[i - 1]
if up[i]:
result[i] = x[i] - atr_[i]
if result[i] < pr:
result[i] = pr
if dn[i]:
result[i] = x[i] + atr_[i]
if result[i] > pr:
result[i] = pr
long, short = result * up, result * dn
long[long == 0], short[short == 0] = nan, nan
return result, long, short
def atrts(
high: Series, low: Series, close: Series, length: Int = None,
ma_length: Int = None, multiplier: IntFloat = None,
mamode: str = None, talib: bool = None, drift: Int = None,
offset: Int = None, **kwargs: DictLike
) -> Series:
"""ATR Trailing Stop (ATRTS)
Attempts to identify exits for long and short positions using both ATR
and a moving average (MA) to determine the trend.
The Average True Range (ATR) is multiplied by a user defined factor.
If the MA is increasing (uptrend), the ATR product is subtracted from
the price or, if the MA is decreasing (down trend), it is added to the
price, and along with a few details the ATRTS is formed. The user may
change the position (long), input (close), method (EMA), period lengths,
percent factor and show entry option(see trading signals below).
Sources:
https://www.motivewave.com/studies/atr_trailing_stops.htm
Calculation:
//position = pos, user defined, default is long
//input = price, user defined, default is close
//method = moving average (ma), user defined, default is EMA
//period1 = maP, user defined, default is 63
//period2 = artP, user defined, default is 21
//factor = fac, user defined, default is 3
//show entrys = showE, user defined, default is false
//index = current bar number, prev = previous
//LOE = less or equal, MOE = more or equal
//shortP = short position, longP = long position
//index = current bar number
longP = pos == "Long";
shortP = pos == "Short";
atrts = 0, atr = 0;
ma = ma(method, maP, input);
prevP = price[index-1];
prevA = ifNull(price, atrts[index]); //current atrts is plotted at index+1
upTrend = price moreThan ma;
dnTrend = price LOE ma;
atr = atr(index, atrP);
if (upTrend)
atrts = price - fac * atr;
if (atrts lessThan prevA) atrts = prevA;
endIf
if (dnTrend)
atrts = price + fac * atr;
if (atrts moreThan prevA) atrts = prevA;
endIf
Plot: atrts[index+1];
//Signals
sell = false, buy = false;
if (atrts != 0)
if (longP AND upTrend)
sell = price lessThan atrts; //sell to exit
buy = prevP lessThan atrts AND price moreThan atrts AND showE; //buy (enter)
endIf
if (shortP AND dnTrend)
sell = prevP moreThan atrts AND price lessThan atrts AND showE; //sell short (enter)
buy = price moreThan atrts; //buy to cover
endIf
endIf
Args:
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
factor (int): the multiplyer. Default: 3
mamode (str): See ```help(ta.ma)```. Default: 'rma'
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
length (int): ATR length. Default: 14
ma_length (int): MA Length. Default: 20
multiplier (int): ATR multiplier. Default: 3
mamode (str): See ``help(ta.ma)``. Default: 'ema'
talib (bool): If TA Lib is installed and talib is True, Returns the
TA Lib version. Default: True
drift (int): The difference period. Default: 1
offset (int): How many periods to offset the result. Default: 0
Kwargs:
Kwargs:
percent (bool, optional): Return as percentage. Default: False
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
@@ -104,39 +88,44 @@ def atrts(high, low, close, length=None, factor=None, mamode=None, talib=None, d
pd.Series: New feature generated.
"""
# Validate
length = int(length) if length and length > 0 else 21
factor = int(factor) if factor and factor > 0 else 3
mamode = mamode.lower() if mamode and isinstance(mamode, str) else "rma"
high = verify_series(high, length)
low = verify_series(low, length)
close = verify_series(close, length)
drift = get_drift(drift)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
length = v_pos_default(length, 14)
ma_length = v_pos_default(ma_length, 20)
_length = max(length, ma_length)
high = v_series(high, _length)
low = v_series(low, _length)
close = v_series(close, _length)
if high is None or low is None or close is None: return
if high is None or low is None or close is None:
return
# Calculate - start
atr_ = atr(high=high, low=low, close=close, length=length)
ma_ = ma(mamode, close, length=length*3)
upTrend = close > ma_
dnTrend = close <= ma_
prevP = close.shift(1)
multiplier = v_pos_default(multiplier, 3.0)
mamode = v_mamode(mamode, "ema")
mode_tal = v_talib(talib)
drift = v_drift(drift)
offset = v_offset(offset)
func_p = partial(calculateFunc, factor=factor)
atrts_ = [func_p(a,b,c,d) for a,b,c,d in zip(upTrend, dnTrend, prevP, atr_)]
atrts_ = Series(atrts_, index=close.index)
# Calculate
if Imports["talib"] and mode_tal:
from talib import ATR
atr_ = ATR(high, low, close, length)
else:
atr_ = atr(
high=high, low=low, close=close, length=length,
mamode=mamode, drift=drift, talib=mode_tal,
offset=offset, **kwargs
)
#prevA = atrts_.shift(1)
#atrts = [tailingStopFunc(a,b,c,d) for a,b,c,d in zip(upTrend, dnTrend, prevA, atrts_)]
atr_ *= multiplier
ma_ = ma(mamode, close, length=ma_length, talib=mode_tal)
#atrts = Series(atrts, index=close.index)
atrts = atrts_.shift(-1)
# Calculate - end
np_close, np_ma, np_atr = close.values, ma_.values, atr_.values
np_atrts_, _, _ = np_atrts(np_close, np_ma, np_atr, length, ma_length)
percentage = kwargs.pop("percent", False)
if percentage:
atrts *= 100 / close
percent = kwargs.pop("percent", False)
if percent:
np_atrts_ *= 100 / np_close
atrts = Series(np_atrts_, index=close.index)
# Offset
if offset != 0:
@@ -149,7 +138,8 @@ def atrts(high, low, close, length=None, factor=None, mamode=None, talib=None, d
atrts.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
atrts.name = f"ATRTS{mamode[0]}_{length}{'p' if percentage else ''}"
_props = f"ATRTS{mamode[0]}{'p' if percent else ''}"
atrts.name = f"{_props}_{length}_{ma_length}_{multiplier}"
atrts.category = "volatility"
return atrts
pandas_ta/volatility/true_range.py
+3 -3
View File
@@ -49,9 +49,9 @@ def true_range(
from talib import TRANGE
true_range = TRANGE(high, low, close)
else:
high_low_range = non_zero_range(high, low)
prev_close = close.shift(drift)
ranges = [high_low_range, high - prev_close, prev_close - low]
hl_range = non_zero_range(high, low)
pc = close.shift(drift)
ranges = [hl_range, high - pc, pc - low]
true_range = concat(ranges, axis=1)
true_range = true_range.abs().max(axis=1)
true_range.iloc[:drift] = nan
setup.py
+1 -1
View File
@@ -20,7 +20,7 @@ setup(
"pandas_ta.volatility",
"pandas_ta.volume"
],
version=".".join(("0", "3", "55b")),
version=".".join(("0", "3", "56b")),
description=long_description,
long_description=long_description,
author="Kevin Johnson",
tests/test_ext_indicator_volatility.py
+5
View File
@@ -33,6 +33,11 @@ class TestVolatilityExtension(TestCase):
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "ATRr_14")
def test_atrts_ext(self):
self.data.ta.atrts(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "ATRTSe_14_20_3.0")
def test_bbands_ext(self):
self.data.ta.bbands(append=True)
self.assertIsInstance(self.data, DataFrame)
tests/test_indicator_volatility.py
+6
View File
@@ -67,6 +67,12 @@ class TestVolatility(TestCase):
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "ATRr_14")
def test_atrts(self):
"""Volatility: ATRTS"""
result = pandas_ta.atrts(self.high, self.low, self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "ATRTSe_14_20_3.0")
def test_bbands(self):
"""Volatility: BBANDS"""
result = pandas_ta.bbands(self.close, talib=False)