wassname/pandas-ta
1
0
Fork 0
mirror of https://github.com/wassname/pandas-ta.git synced 2026-06-27 16:10:07 +08:00
Code Issues Packages Projects Releases Wiki Activity

ENH #125 @244 returns None instead and runs relevant indicators in strategy MAINT refactor

Browse Source
This commit is contained in:
Kevin Johnson
2021-03-13 10:19:40 -08:00
parent 11978449e5
commit 2773f17d25
122 changed files with 564 additions and 323 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pandas_ta/candles/cdl_doji.py
+8 -6
View File
@@ -1,22 +1,24 @@
# -*- coding: utf-8 -*-
from pandas_ta.overlap import sma
from pandas_ta.utils import get_offset, high_low_range, is_percent
from pandas_ta.utils import non_zero_range, real_body, verify_series
from pandas_ta.utils import real_body, verify_series
def cdl_doji( open_, high, low, close, length=None, factor=None, scalar=None, asint=True, offset=None, **kwargs):
def cdl_doji(open_, high, low, close, length=None, factor=None, scalar=None, asint=True, offset=None, **kwargs):
"""Candle Type: Doji"""
# Validate Arguments
open_ = verify_series(open_)
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
length = int(length) if length and length > 0 else 10
factor = float(factor) if is_percent(factor) else 10
scalar = float(scalar) if scalar else 100
open_ = verify_series(open_, length)
high = verify_series(high, length)
low = verify_series(low, length)
close = verify_series(close, length)
offset = get_offset(offset)
naive = kwargs.pop("naive", False)
if open_ is None or high is None or low is None or close is None: return
# Calculate Result
body = real_body(open_, close).abs()
hl_range = high_low_range(high, low).abs()
pandas_ta/candles/cdl_inside.py
+2 -3
View File
@@ -1,7 +1,6 @@
# -*- coding: utf-8 -*-
from pandas import DataFrame, set_option
from pandas_ta.utils import candle_color, get_drift, get_offset
from pandas_ta.utils import non_zero_range, real_body, verify_series
from pandas_ta.utils import candle_color, get_offset
from pandas_ta.utils import verify_series
def cdl_inside(open_, high, low, close, asbool=False, offset=None, **kwargs):
pandas_ta/candles/ha.py
-1
View File
@@ -1,5 +1,4 @@
# -*- coding: utf-8 -*-
import numpy as np
from pandas import DataFrame
from pandas_ta.utils import get_offset, verify_series
pandas_ta/core.py
+24 -9
View File
@@ -1,13 +1,13 @@
# -*- coding: utf-8 -*-
from dataclasses import dataclass, field
from datetime import datetime
from math import log as mlog
from multiprocessing import cpu_count, Pool
from time import perf_counter
from typing import List, Tuple
import pandas as pd
from numpy import ndarray as npndarray
from numpy import log10 as npLog10
from numpy import ndarray as npNdarray
from pandas.core.base import PandasObject
from pandas_ta import version, Category
@@ -118,9 +118,7 @@ class BasePandasObject(PandasObject):
"""
def __init__(self, df, **kwargs):
if df.empty:
return
if df.empty: return
if len(df.columns) > 0:
common_names = {
"Date": "date",
@@ -237,6 +235,7 @@ class AnalysisIndicators(BasePandasObject):
_adjusted = None
_cores = cpu_count()
_mp = False
_time_range = "years"
# DataFrame Behavioral Methods
def __call__(
@@ -326,6 +325,19 @@ class AnalysisIndicators(BasePandasObject):
"""Reverses the DataFrame. Simply: df.iloc[::-1]"""
return self._df.iloc[::-1]
@property
def time_range(self) -> str:
""""""
return total_time(self._df, self._time_range)
@time_range.setter
def time_range(self, value: str) -> None:
"""property: df.ta.mp = False (Default)"""
if value is not None and isinstance(value, str):
self._time_range = value
else:
self._time_range = "years"
@property
def version(self) -> str:
"""Returns the version."""
@@ -425,8 +437,10 @@ class AnalysisIndicators(BasePandasObject):
* Applies prefixes and/or suffixes
* Appends the result to main DataFrame
"""
verbose = kwargs.pop("verbose", False)
if not isinstance(result, (pd.Series, pd.DataFrame)):
print(f"[X] Oops! The result was not a Series or DataFrame.")
if verbose:
print(f"[X] Oops! The result was not a Series or DataFrame.")
return self._df
else:
# Append only specific columns to the dataframe (via
@@ -495,7 +509,7 @@ class AnalysisIndicators(BasePandasObject):
Returns nothing to the user. Either adds or removes constant ranges
from the working DataFrame.
"""
if isinstance(values, npndarray) or isinstance(values, list):
if isinstance(values, npNdarray) or isinstance(values, list):
if append:
for x in values:
self._df[f"{x}"] = x
@@ -528,6 +542,7 @@ class AnalysisIndicators(BasePandasObject):
"datetime_ordered",
"mp",
"reverse",
"time_range",
"version",
]
@@ -654,9 +669,9 @@ class AnalysisIndicators(BasePandasObject):
_total_ta = len(ta)
pool = Pool(self.cores)
# Some magic to optimize chunksize for speed based on total ta indicators
_chunksize = mp_chunksize - 1 if mp_chunksize > _total_ta else int(mlog(_total_ta)) + 1
_chunksize = mp_chunksize - 1 if mp_chunksize > _total_ta else int(npLog10(_total_ta)) + 1
if verbose:
print(f"[i] Multiprocessing: {self.cores} of {cpu_count()} cores of {_total_ta} indicators.")
print(f"[i] Multiprocessing: {_chunksize} chunks over {cpu_count()} cores for {_total_ta} indicators.")
results = None
if mode["custom"]:
pandas_ta/cycles/ebsw.py
+12 -6
View File
@@ -1,6 +1,10 @@
# -*- coding: utf-8 -*-
import math
from numpy import cos as npCos
from numpy import exp as npExp
from numpy import NaN as npNaN
from numpy import pi as npPi
from numpy import sin as npSin
from numpy import sqrt as npSqrt
from pandas import Series
from pandas_ta.utils import get_offset, verify_series
@@ -8,11 +12,13 @@ from pandas_ta.utils import get_offset, verify_series
def ebsw(close, length=None, bars=None, offset=None, **kwargs):
"""Indicator: Even Better SineWave (EBSW)"""
# Validate arguments
close = verify_series(close)
length = int(length) if length and length > 38 else 40
bars = int(bars) if bars and bars > 0 else 10
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# variables
alpha1 = HP = 0 # alpha and HighPass
a1 = b1 = c1 = c2 = c3 = 0
@@ -26,12 +32,12 @@ def ebsw(close, length=None, bars=None, offset=None, **kwargs):
result = [npNaN for _ in range(0, length - 1)] + [0]
for i in range(length, m):
# HighPass filter cyclic components whose periods are shorter than Duration input
alpha1 = (1 - math.sin(360 / length)) / math.cos(360 / length)
alpha1 = (1 - npSin(360 / length)) / npCos(360 / length)
HP = 0.5 * (1 + alpha1) * (close[i] - lastClose) + alpha1 * lastHP
# Smooth with a Super Smoother Filter from equation 3-3
a1 = math.exp(-math.sqrt(2) * math.pi / bars)
b1 = 2 * a1 * math.cos(math.sqrt(2) * 180 / bars)
a1 = npExp(-npSqrt(2) * npPi / bars)
b1 = 2 * a1 * npCos(npSqrt(2) * 180 / bars)
c2 = b1
c3 = -1 * a1 * a1
c1 = 1 - c2 - c3
@@ -43,7 +49,7 @@ def ebsw(close, length=None, bars=None, offset=None, **kwargs):
Pwr = (Filt * Filt + FilterHist[1] * FilterHist[1] + FilterHist[0] * FilterHist[0]) / 3
# Normalize the Average Wave to Square Root of the Average Power
Wave = Wave / math.sqrt(Pwr)
Wave = Wave / npSqrt(Pwr)
# update storage, result
FilterHist.append(Filt) # append new Filt value
pandas_ta/momentum/ao.py
+5 -2
View File
@@ -6,14 +6,17 @@ from pandas_ta.utils import get_offset, verify_series
def ao(high, low, fast=None, slow=None, offset=None, **kwargs):
"""Indicator: Awesome Oscillator (AO)"""
# Validate Arguments
high = verify_series(high)
low = verify_series(low)
fast = int(fast) if fast and fast > 0 else 5
slow = int(slow) if slow and slow > 0 else 34
if slow < fast:
fast, slow = slow, fast
_length = max(fast, slow)
high = verify_series(high, _length)
low = verify_series(low, _length)
offset = get_offset(offset)
if high is None or low is None: return
# Calculate Result
median_price = 0.5 * (high + low)
fast_sma = sma(median_price, fast)
pandas_ta/momentum/apo.py
+3 -1
View File
@@ -6,13 +6,15 @@ from pandas_ta.utils import get_offset, verify_series
def apo(close, fast=None, slow=None, offset=None, **kwargs):
"""Indicator: Absolute Price Oscillator (APO)"""
# Validate Arguments
close = verify_series(close)
fast = int(fast) if fast and fast > 0 else 12
slow = int(slow) if slow and slow > 0 else 26
if slow < fast:
fast, slow = slow, fast
close = verify_series(close, max(fast, slow))
offset = get_offset(offset)
if close is None: return
# Calculate Result
fastma = sma(close, length=fast)
slowma = sma(close, length=slow)
pandas_ta/momentum/bias.py
+3 -1
View File
@@ -6,11 +6,13 @@ from pandas_ta.utils import get_offset, verify_series
def bias(close, length=None, mamode=None, offset=None, **kwargs):
"""Indicator: Bias (BIAS)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 26
mamode = mamode if isinstance(mamode, str) else "sma"
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
bma = ma(mamode, close, length=length, **kwargs)
bias = (close / bma) - 1
pandas_ta/momentum/brar.py
+6 -4
View File
@@ -6,17 +6,19 @@ from pandas_ta.utils import get_drift, get_offset, non_zero_range, verify_series
def brar(open_, high, low, close, length=None, scalar=None, drift=None, offset=None, **kwargs):
"""Indicator: BRAR (BRAR)"""
# Validate Arguments
open_ = verify_series(open_)
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
length = int(length) if length and length > 0 else 26
scalar = float(scalar) if scalar else 100
high_open_range = non_zero_range(high, open_)
open_low_range = non_zero_range(open_, low)
open_ = verify_series(open_, length)
high = verify_series(high, length)
low = verify_series(low, length)
close = verify_series(close, length)
drift = get_drift(drift)
offset = get_offset(offset)
if open_ is None or high is None or low is None or close is None: return
# Calculate Result
hcy = non_zero_range(high, close.shift(drift))
cyl = non_zero_range(close.shift(drift), low)
pandas_ta/momentum/cci.py
+5 -3
View File
@@ -7,13 +7,15 @@ from pandas_ta.utils import get_offset, verify_series
def cci(high, low, close, length=None, c=None, offset=None, **kwargs):
"""Indicator: Commodity Channel Index (CCI)"""
# Validate Arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
length = int(length) if length and length > 0 else 14
c = float(c) if c and c > 0 else 0.015
high = verify_series(high, length)
low = verify_series(low, length)
close = verify_series(close, length)
offset = get_offset(offset)
if high is None or low is None or close is None: return
# Calculate Result
typical_price = hlc3(high=high, low=low, close=close)
mean_typical_price = sma(typical_price, length=length)
pandas_ta/momentum/cfo.py
+3 -1
View File
@@ -6,12 +6,14 @@ from pandas_ta.utils import get_drift, get_offset, verify_series
def cfo(close, length=None, scalar=None, drift=None, offset=None, **kwargs):
"""Indicator: Chande Forcast Oscillator (CFO)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 9
scalar = float(scalar) if scalar else 100
close = verify_series(close, length)
drift = get_drift(drift)
offset = get_offset(offset)
if close is None: return
# Finding linear regression of Series
cfo = scalar * (close - linreg(close, length=length, tsf=True))
cfo /= close
pandas_ta/momentum/cg.py
+3 -1
View File
@@ -5,10 +5,12 @@ from pandas_ta.utils import get_offset, verify_series, weights
def cg(close, length=None, offset=None, **kwargs):
"""Indicator: Center of Gravity (CG)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
coefficients = [length - i for i in range(0, length)]
numerator = -close.rolling(length).apply(weights(coefficients), raw=True)
pandas_ta/momentum/cmo.py
+4 -2
View File
@@ -6,18 +6,20 @@ from pandas_ta.utils import get_drift, get_offset, verify_series
def cmo(close, length=None, scalar=None, drift=None, offset=None, **kwargs):
"""Indicator: Chande Momentum Oscillator (CMO)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 14
scalar = float(scalar) if scalar else 100
talib = kwargs.pop("talib", True)
close = verify_series(close, length)
drift = get_drift(drift)
offset = get_offset(offset)
if close is None: return
# Calculate Result
mom = close.diff(drift)
positive = mom.copy().clip(lower=0)
negative = mom.copy().clip(upper=0).abs()
talib = kwargs.pop("talib", True)
if talib:
pos_ = rma(positive, length)
neg_ = rma(negative, length)
pandas_ta/momentum/coppock.py
+3 -1
View File
@@ -7,12 +7,14 @@ from pandas_ta.utils import get_offset, verify_series
def coppock(close, length=None, fast=None, slow=None, offset=None, **kwargs):
"""Indicator: Coppock Curve (COPC)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
fast = int(fast) if fast and fast > 0 else 11
slow = int(slow) if slow and slow > 0 else 14
close = verify_series(close, max(length, fast, slow))
offset = get_offset(offset)
if close is None: return
# Calculate Result
total_roc = roc(close, fast) + roc(close, slow)
coppock = wma(total_roc, length)
pandas_ta/momentum/er.py
+3 -1
View File
@@ -6,11 +6,13 @@ from pandas_ta.utils import get_drift, get_offset, verify_series, signals
def er(close, length=None, drift=None, offset=None, **kwargs):
"""Indicator: Efficiency Ratio (ER)"""
# Validate arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
close = verify_series(close, length)
offset = get_offset(offset)
drift = get_drift(drift)
if close is None: return
# Calculate Result
abs_diff = close.diff(length).abs()
abs_volatility = close.diff(drift).abs()
pandas_ta/momentum/eri.py
+5 -3
View File
@@ -7,12 +7,14 @@ from pandas_ta.utils import get_offset, verify_series
def eri(high, low, close, length=None, offset=None, **kwargs):
"""Indicator: Elder Ray Index (ERI)"""
# Validate arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
length = int(length) if length and length > 0 else 13
high = verify_series(high, length)
low = verify_series(low, length)
close = verify_series(close, length)
offset = get_offset(offset)
if high is None or low is None or close is None: return
# Calculate Result
ema_ = ema(close, length)
bull = high - ema_
pandas_ta/momentum/fisher.py
+5 -2
View File
@@ -9,12 +9,15 @@ from pandas_ta.utils import get_offset, high_low_range, verify_series, zero
def fisher(high, low, length=None, signal=None, offset=None, **kwargs):
"""Indicator: Fisher Transform (FISHT)"""
# Validate Arguments
high = verify_series(high)
low = verify_series(low)
length = int(length) if length and length > 0 else 9
signal = int(signal) if signal and signal > 0 else 1
_length = max(length, signal)
high = verify_series(high, _length)
low = verify_series(low, _length)
offset = get_offset(offset)
if high is None or low is None: return
# Calculate Result
hl2_ = hl2(high, low)
highest_hl2 = hl2_.rolling(length).max()
pandas_ta/momentum/inertia.py
+8 -4
View File
@@ -1,25 +1,29 @@
# -*- coding: utf-8 -*-
from pandas_ta.overlap import linreg
from pandas_ta.volatility import rvi
from pandas_ta.utils import get_drift, get_offset, non_zero_range, verify_series
from pandas_ta.utils import get_drift, get_offset, verify_series
def inertia(close=None, high=None, low=None, length=None, rvi_length=None, scalar=None, refined=None, thirds=None, mamode=None, drift=None, offset=None, **kwargs):
"""Indicator: Inertia (INERTIA)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 20
rvi_length = int(rvi_length) if rvi_length and rvi_length > 0 else 14
scalar = float(scalar) if scalar and scalar > 0 else 100
refined = False if refined is None else True
thirds = False if thirds is None else True
mamode = mamode if isinstance(mamode, str) else "ema"
_length = max(length, rvi_length)
close = verify_series(close, _length)
drift = get_drift(drift)
offset = get_offset(offset)
if close is None: return
if refined or thirds:
high = verify_series(high)
low = verify_series(low)
high = verify_series(high, _length)
low = verify_series(low, _length)
if high is None or low is None: return
# Calculate Result
if refined:
pandas_ta/momentum/kdj.py
+6 -3
View File
@@ -7,13 +7,16 @@ from pandas_ta.utils import get_offset, non_zero_range, verify_series
def kdj(high=None, low=None, close=None, length=None, signal=None, offset=None, **kwargs):
"""Indicator: KDJ (KDJ)"""
# Validate Arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
length = int(length) if length and length > 0 else 9
signal = int(signal) if signal and signal > 0 else 3
_length = max(length, signal)
high = verify_series(high, _length)
low = verify_series(low, _length)
close = verify_series(close, _length)
offset = get_offset(offset)
if high is None or low is None or close is None: return
# Calculate Result
highest_high = high.rolling(length).max()
lowest_low = low.rolling(length).min()
pandas_ta/momentum/kst.py
+4 -1
View File
@@ -7,7 +7,6 @@ from pandas_ta.utils import get_drift, get_offset, verify_series
def kst(close, roc1=None, roc2=None, roc3=None, roc4=None, sma1=None, sma2=None, sma3=None, sma4=None, signal=None, drift=None, offset=None, **kwargs):
"""Indicator: 'Know Sure Thing' (KST)"""
# Validate arguments
close = verify_series(close)
roc1 = int(roc1) if roc1 and roc1 > 0 else 10
roc2 = int(roc2) if roc2 and roc2 > 0 else 15
roc3 = int(roc3) if roc3 and roc3 > 0 else 20
@@ -19,9 +18,13 @@ def kst(close, roc1=None, roc2=None, roc3=None, roc4=None, sma1=None, sma2=None,
sma4 = int(sma4) if sma4 and sma4 > 0 else 15
signal = int(signal) if signal and signal > 0 else 9
_length = max(roc1, roc2, roc3, roc4, sma1, sma2, sma3, sma4, signal)
close = verify_series(close, _length)
drift = get_drift(drift)
offset = get_offset(offset)
if close is None: return
# Calculate Result
rocma1 = roc(close, roc1).rolling(sma1).mean()
rocma2 = roc(close, roc2).rolling(sma2).mean()
pandas_ta/momentum/macd.py
+4 -2
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
from pandas import DataFrame, concat
from pandas import concat, DataFrame
from pandas_ta.overlap import ema
from pandas_ta.utils import get_offset, verify_series, signals
@@ -7,14 +7,16 @@ from pandas_ta.utils import get_offset, verify_series, signals
def macd(close, fast=None, slow=None, signal=None, offset=None, **kwargs):
"""Indicator: Moving Average, Convergence/Divergence (MACD)"""
# Validate arguments
close = verify_series(close)
fast = int(fast) if fast and fast > 0 else 12
slow = int(slow) if slow and slow > 0 else 26
signal = int(signal) if signal and signal > 0 else 9
if slow < fast:
fast, slow = slow, fast
close = verify_series(close, max(fast, slow, signal))
offset = get_offset(offset)
if close is None: return
# Calculate Result
fastma = ema(close, length=fast)
slowma = ema(close, length=slow)
pandas_ta/momentum/mom.py
+3 -1
View File
@@ -5,10 +5,12 @@ from pandas_ta.utils import get_offset, verify_series
def mom(close, length=None, offset=None, **kwargs):
"""Indicator: Momentum (MOM)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
mom = close.diff(length)
pandas_ta/momentum/pgo.py
+5 -3
View File
@@ -7,12 +7,14 @@ from pandas_ta.utils import get_offset, verify_series
def pgo(high, low, close, length=None, offset=None, **kwargs):
"""Indicator: Pretty Good Oscillator (PGO)"""
# Validate arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
length = int(length) if length and length > 0 else 14
high = verify_series(high, length)
low = verify_series(low, length)
close = verify_series(close, length)
offset = get_offset(offset)
if high is None or low is None or close is None: return
# Calculate Result
pgo = close - sma(close, length)
pgo /= ema(atr(high, low, close, length), length)
pandas_ta/momentum/ppo.py
+3 -1
View File
@@ -7,15 +7,17 @@ from pandas_ta.utils import get_offset, verify_series
def ppo(close, fast=None, slow=None, signal=None, scalar=None, offset=None, **kwargs):
"""Indicator: Percentage Price Oscillator (PPO)"""
# Validate Arguments
close = verify_series(close)
fast = int(fast) if fast and fast > 0 else 12
slow = int(slow) if slow and slow > 0 else 26
signal = int(signal) if signal and signal > 0 else 9
scalar = float(scalar) if scalar else 100
if slow < fast:
fast, slow = slow, fast
close = verify_series(close, max(fast, slow, signal))
offset = get_offset(offset)
if close is None: return
# Calculate Result
fastma = sma(close, length=fast)
slowma = sma(close, length=slow)
pandas_ta/momentum/psl.py
+3 -1
View File
@@ -6,12 +6,14 @@ from pandas_ta.utils import get_drift, get_offset, verify_series
def psl(close, open_=None, length=None, scalar=None, drift=None, offset=None, **kwargs):
"""Indicator: Psychological Line (PSL)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 12
scalar = float(scalar) if scalar and scalar > 0 else 100
close = verify_series(close, length)
drift = get_drift(drift)
offset = get_offset(offset)
if close is None: return
# Calculate Result
if open_ is not None:
open_ = verify_series(open_)
pandas_ta/momentum/pvo.py
+3 -1
View File
@@ -7,15 +7,17 @@ from pandas_ta.utils import get_offset, verify_series
def pvo(volume, fast=None, slow=None, signal=None, scalar=None, offset=None, **kwargs):
"""Indicator: Percentage Volume Oscillator (PVO)"""
# Validate Arguments
volume = verify_series(volume)
fast = int(fast) if fast and fast > 0 else 12
slow = int(slow) if slow and slow > 0 else 26
signal = int(signal) if signal and signal > 0 else 9
scalar = float(scalar) if scalar else 100
if slow < fast:
fast, slow = slow, fast
volume = verify_series(volume, max(fast, slow, signal))
offset = get_offset(offset)
if volume is None: return
# Calculate Result
fastma = ema(volume, length=fast)
slowma = ema(volume, length=slow)
pandas_ta/momentum/qqe.py
+4 -2
View File
@@ -12,14 +12,17 @@ from pandas_ta.utils import get_drift, get_offset, verify_series
def qqe(close, length=None, smooth=None, factor=None, mamode=None, drift=None, offset=None, **kwargs):
"""Indicator: Quantitative Qualitative Estimation (QQE)"""
# Validate arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 14
smooth = int(smooth) if smooth and smooth > 0 else 5
factor = float(factor) if factor else 4.236
wilders_length = 2 * length - 1
mamode = mamode if isinstance(mamode, str) else "ema"
close = verify_series(close, max(length, smooth, wilders_length))
drift = get_drift(drift)
offset = get_offset(offset)
if close is None: return
# Calculate Result
rsi_ = rsi(close, length)
_mode = mamode.lower()[0] if mamode != "ema" else ""
@@ -30,7 +33,6 @@ def qqe(close, length=None, smooth=None, factor=None, mamode=None, drift=None, o
# Double Smooth the RSI MA True Range using Wilder's Length with a default
# width of 4.236.
wilders_length = 2 * length - 1
smoothed_rsi_tr_ma = ma("ema", rsi_ma_tr, length=wilders_length)
dar = factor * ma("ema", smoothed_rsi_tr_ma, length=wilders_length)
pandas_ta/momentum/roc.py
+3 -1
View File
@@ -6,10 +6,12 @@ from pandas_ta.utils import get_offset, verify_series
def roc(close, length=None, offset=None, **kwargs):
"""Indicator: Rate of Change (ROC)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
roc = 100 * mom(close=close, length=length) / close.shift(length)
pandas_ta/momentum/rsi.py
+3 -1
View File
@@ -7,12 +7,14 @@ from pandas_ta.utils import get_drift, get_offset, verify_series, signals
def rsi(close, length=None, scalar=None, drift=None, offset=None, **kwargs):
"""Indicator: Relative Strength Index (RSI)"""
# Validate arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 14
scalar = float(scalar) if scalar else 100
close = verify_series(close, length)
drift = get_drift(drift)
offset = get_offset(offset)
if close is None: return
# Calculate Result
negative = close.diff(drift)
positive = negative.copy()
pandas_ta/momentum/rsx.py
+5 -2
View File
@@ -1,16 +1,19 @@
# -*- coding: utf-8 -*-
from numpy import NaN as npNaN
from pandas import DataFrame, Series, concat
from pandas import concat, DataFrame, Series
from pandas_ta.utils import get_drift, get_offset, verify_series, signals
def rsx(close, length=None, drift=None, offset=None, **kwargs):
"""Indicator: Relative Strength Xtra (inspired by Jurik RSX)"""
# Validate arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 14
close = verify_series(close, length)
drift = get_drift(drift)
offset = get_offset(offset)
if close is None: return
# variables
vC, v1C = 0, 0
v4, v8, v10, v14, v18, v20 = 0, 0, 0, 0, 0, 0
pandas_ta/momentum/rvgi.py
+7 -4
View File
@@ -7,16 +7,19 @@ from pandas_ta.utils import get_offset, non_zero_range, verify_series
def rvgi(open_, high, low, close, length=None, swma_length=None, offset=None, **kwargs):
"""Indicator: Relative Vigor Index (RVGI)"""
# Validate Arguments
open_ = verify_series(open_)
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
high_low_range = non_zero_range(high, low)
close_open_range = non_zero_range(close, open_)
length = int(length) if length and length > 0 else 14
swma_length = int(swma_length) if swma_length and swma_length > 0 else 4
_length = max(length, swma_length)
open_ = verify_series(open_, _length)
high = verify_series(high, _length)
low = verify_series(low, _length)
close = verify_series(close, _length)
offset = get_offset(offset)
if open_ is None or high is None or low is None or close is None: return
# Calculate Result
numerator = swma(close_open_range, length=swma_length).rolling(length).sum()
denominator = swma(high_low_range, length=swma_length).rolling(length).sum()
pandas_ta/momentum/slope.py
+7 -4
View File
@@ -1,23 +1,26 @@
# -*- coding: utf-8 -*-
from math import atan, pi
from numpy import arctan as npAtan
from numpy import pi as npPi
from pandas_ta.utils import get_offset, verify_series
def slope( close, length=None, as_angle=None, to_degrees=None, vertical=None, offset=None, **kwargs):
"""Indicator: Slope"""
# Validate arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 1
as_angle = True if isinstance(as_angle, bool) else False
to_degrees = True if isinstance(to_degrees, bool) else False
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
slope = close.diff(length) / length
if as_angle:
slope = slope.apply(atan)
slope = slope.apply(npAtan)
if to_degrees:
slope *= 180 / pi
slope *= 180 / npPi
# Offset
if offset != 0:
pandas_ta/momentum/smi.py
+5 -3
View File
@@ -1,22 +1,24 @@
# -*- coding: utf-8 -*-
from pandas import concat, DataFrame
from pandas import DataFrame
from .tsi import tsi
from pandas_ta.overlap import ema
from pandas_ta.utils import get_offset, verify_series, signals
from pandas_ta.utils import get_offset, verify_series
def smi(close, fast=None, slow=None, signal=None, scalar=None, offset=None, **kwargs):
"""Indicator: SMI Ergodic Indicator (SMIIO)"""
# Validate arguments
close = verify_series(close)
fast = int(fast) if fast and fast > 0 else 5
slow = int(slow) if slow and slow > 0 else 20
signal = int(signal) if signal and signal > 0 else 5
if slow < fast:
fast, slow = slow, fast
scalar = float(scalar) if scalar else 1
close = verify_series(close, max(fast, slow, signal))
offset = get_offset(offset)
if close is None: return
# Calculate Result
smi = tsi(close, fast=fast, slow=slow, scalar=scalar)
signalma = ema(smi, signal)
pandas_ta/momentum/squeeze.py
+7 -6
View File
@@ -1,7 +1,6 @@
# -*- coding: utf-8 -*-
from numpy import NaN as npNaN
from pandas import DataFrame
from pandas_ta.momentum import mom
from pandas_ta.overlap import ema, linreg, sma
from pandas_ta.trend import decreasing, increasing
@@ -13,17 +12,19 @@ from pandas_ta.utils import unsigned_differences, verify_series
def squeeze(high, low, close, bb_length=None, bb_std=None, kc_length=None, kc_scalar=None, mom_length=None, mom_smooth=None, use_tr=None, offset=None, **kwargs):
"""Indicator: Squeeze Momentum (SQZ)"""
# Validate arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
offset = get_offset(offset)
bb_length = int(bb_length) if bb_length and bb_length > 0 else 20
bb_std = float(bb_std) if bb_std and bb_std > 0 else 2.0
kc_length = int(kc_length) if kc_length and kc_length > 0 else 20
kc_scalar = float(kc_scalar) if kc_scalar and kc_scalar > 0 else 1.5
mom_length = int(mom_length) if mom_length and mom_length > 0 else 12
mom_smooth = int(mom_smooth) if mom_smooth and mom_smooth > 0 else 6
_length = max(bb_length, kc_length, mom_length, mom_smooth)
high = verify_series(high, _length)
low = verify_series(low, _length)
close = verify_series(close, _length)
offset = get_offset(offset)
if high is None or low is None or close is None: return
use_tr = kwargs.setdefault("tr", True)
asint = kwargs.pop("asint", True)
pandas_ta/momentum/stoch.py
+6 -3
View File
@@ -7,14 +7,17 @@ from pandas_ta.utils import get_offset, non_zero_range, verify_series
def stoch(high, low, close, k=None, d=None, smooth_k=None, offset=None, **kwargs):
"""Indicator: Stochastic Oscillator (STOCH)"""
# Validate arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
k = k if k and k > 0 else 14
d = d if d and d > 0 else 3
smooth_k = smooth_k if smooth_k and smooth_k > 0 else 3
_length = max(k, d, smooth_k)
high = verify_series(high, _length)
low = verify_series(low, _length)
close = verify_series(close, _length)
offset = get_offset(offset)
if high is None or low is None or close is None: return
# Calculate Result
lowest_low = low.rolling(k).min()
highest_high = high.rolling(k).max()
pandas_ta/momentum/stochrsi.py
+3 -1
View File
@@ -8,13 +8,15 @@ from pandas_ta.utils import get_offset, non_zero_range, verify_series
def stochrsi(close, length=None, rsi_length=None, k=None, d=None, offset=None, **kwargs):
"""Indicator: Stochastic RSI Oscillator (STOCHRSI)"""
# Validate arguments
close = verify_series(close)
length = length if length and length > 0 else 14
rsi_length = rsi_length if rsi_length and rsi_length > 0 else 14
k = k if k and k > 0 else 3
d = d if d and d > 0 else 3
close = verify_series(close, max(length, rsi_length, k, d))
offset = get_offset(offset)
if close is None: return
# Calculate Result
rsi_ = rsi(close, length=rsi_length)
lowest_rsi = rsi_.rolling(length).min()
pandas_ta/momentum/trix.py
+3 -1
View File
@@ -7,13 +7,15 @@ from pandas_ta.utils import get_drift, get_offset, verify_series
def trix(close, length=None, signal=None, scalar=None, drift=None, offset=None, **kwargs):
"""Indicator: Trix (TRIX)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 30
signal = int(signal) if signal and signal > 0 else 9
scalar = float(scalar) if scalar else 100
close = verify_series(close, max(length, signal))
drift = get_drift(drift)
offset = get_offset(offset)
if close is None: return
# Calculate Result
ema1 = ema(close=close, length=length, **kwargs)
ema2 = ema(close=ema1, length=length, **kwargs)
pandas_ta/momentum/tsi.py
+3 -1
View File
@@ -6,16 +6,18 @@ from pandas_ta.utils import get_drift, get_offset, verify_series
def tsi(close, fast=None, slow=None, scalar=None, drift=None, offset=None, **kwargs):
"""Indicator: True Strength Index (TSI)"""
# Validate Arguments
close = verify_series(close)
fast = int(fast) if fast and fast > 0 else 13
slow = int(slow) if slow and slow > 0 else 25
# if slow < fast:
# fast, slow = slow, fast
scalar = float(scalar) if scalar else 100
close = verify_series(close, max(fast, slow))
drift = get_drift(drift)
offset = get_offset(offset)
if "length" in kwargs: kwargs.pop("length")
if close is None: return
# Calculate Result
diff = close.diff(drift)
slow_ema = ema(close=diff, length=slow, **kwargs)
pandas_ta/momentum/uo.py
+11 -11
View File
@@ -6,20 +6,20 @@ from pandas_ta.utils import get_drift, get_offset, verify_series
def uo(high, low, close, fast=None, medium=None, slow=None, fast_w=None, medium_w=None, slow_w=None, drift=None, offset=None, **kwargs):
"""Indicator: Ultimate Oscillator (UO)"""
# Validate arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
fast = int(fast) if fast and fast > 0 else 7
fast_w = float(fast_w) if fast_w and fast_w > 0 else 4.0
medium = int(medium) if medium and medium > 0 else 14
medium_w = float(medium_w) if medium_w and medium_w > 0 else 2.0
slow = int(slow) if slow and slow > 0 else 28
slow_w = float(slow_w) if slow_w and slow_w > 0 else 1.0
_length = max(fast, medium, slow)
high = verify_series(high, _length)
low = verify_series(low, _length)
close = verify_series(close, _length)
drift = get_drift(drift)
offset = get_offset(offset)
fast = int(fast) if fast and fast > 0 else 7
fast_w = float(fast_w) if fast_w and fast_w > 0 else 4.0
medium = int(medium) if medium and medium > 0 else 14
medium_w = float(medium_w) if medium_w and medium_w > 0 else 2.0
slow = int(slow) if slow and slow > 0 else 28
slow_w = float(slow_w) if slow_w and slow_w > 0 else 1.0
if high is None or low is None or close is None: return
# Calculate Result
tdf = DataFrame({
pandas_ta/momentum/willr.py
+7 -4
View File
@@ -1,17 +1,20 @@
# -*- coding: utf-8 -*-
from pandas_ta.utils import get_drift, get_offset, verify_series
from pandas_ta.utils import get_offset, verify_series
def willr(high, low, close, length=None, offset=None, **kwargs):
"""Indicator: William's Percent R (WILLR)"""
# Validate arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
length = int(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
_length = max(length, min_periods)
high = verify_series(high, _length)
low = verify_series(low, _length)
close = verify_series(close, _length)
offset = get_offset(offset)
if high is None or low is None or close is None: return
# Calculate Result
lowest_low = low.rolling(length, min_periods=min_periods).min()
highest_high = high.rolling(length, min_periods=min_periods).max()
pandas_ta/overlap/alma.py
+8 -9
View File
@@ -1,25 +1,27 @@
# -*- coding: utf-8 -*-
from numpy import exp as npExp
from numpy import NaN as npNaN
from pandas import Series
from pandas_ta.utils import get_offset, verify_series
import math
def alma(close, length=None, sigma=None, distribution_offset=None, offset=None, **kwargs):
"""Indicator: Arnaud Legoux Moving Average (ALMA)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
sigma = float(sigma) if sigma and sigma > 0 else 6.0
distribution_offset = float(distribution_offset) if distribution_offset and distribution_offset > 0 else 0.85
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Pre-Calculations
m = distribution_offset * (length - 1)
s = length / sigma
wtd = list(range(length))
for i in range(0, length):
wtd[i] = math.exp(-1 * ((i - m) * (i - m)) / (2 * s * s))
wtd[i] = npExp(-1 * ((i - m) * (i - m)) / (2 * s * s))
# Calculate Result
result = [npNaN for _ in range(0, length - 1)] + [0]
@@ -27,15 +29,12 @@ def alma(close, length=None, sigma=None, distribution_offset=None, offset=None,
window_sum = 0
cum_sum = 0
for j in range(0, length):
# wtd = math.exp(-1 * ((j - m) * (j - m)) / (2 * s * s)) # moved to pre-calc for efficiency
# wtd = exp(-1 * ((j - m) * (j - m)) / (2 * s * s)) # moved to pre-calc for efficiency
window_sum = window_sum + wtd[j] * close[i - j]
cum_sum = cum_sum + wtd[j]
almean = window_sum / cum_sum
if i == length:
result.append(npNaN) # additional one bar NaN as pre-roll
else:
result.append(almean)
almean = window_sum / cum_sum
result.append(npNaN) if i == length else result.append(almean)
alma = Series(result, index=close.index)
pandas_ta/overlap/dema.py
+3 -1
View File
@@ -6,10 +6,12 @@ from pandas_ta.utils import get_offset, verify_series
def dema(close, length=None, offset=None, **kwargs):
"""Indicator: Double Exponential Moving Average (DEMA)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
ema1 = ema(close=close, length=length)
ema2 = ema(close=ema1, length=length)
pandas_ta/overlap/ema.py
+3 -1
View File
@@ -6,12 +6,14 @@ from pandas_ta.utils import get_offset, verify_series
def ema(close, length=None, offset=None, **kwargs):
"""Indicator: Exponential Moving Average (EMA)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
adjust = kwargs.pop("adjust", False)
sma = kwargs.pop("sma", True)
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
if sma:
close = close.copy()
pandas_ta/overlap/fwma.py
+3 -1
View File
@@ -5,11 +5,13 @@ from pandas_ta.utils import fibonacci, get_offset, verify_series, weights
def fwma(close, length=None, asc=None, offset=None, **kwargs):
"""Indicator: Fibonacci's Weighted Moving Average (FWMA)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
asc = asc if asc else True
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
fibs = fibonacci(n=length, weighted=True)
fwma = close.rolling(length, min_periods=length).apply(weights(fibs), raw=True)
pandas_ta/overlap/hilo.py
+6 -4
View File
@@ -1,7 +1,6 @@
# -*- coding: utf-8 -*-
from numpy import NaN as npNaN
from pandas import DataFrame, Series
# from pandas_ta.overlap.ma import ma
from .ma import ma
from pandas_ta.utils import get_offset, verify_series
@@ -9,14 +8,17 @@ from pandas_ta.utils import get_offset, verify_series
def hilo(high, low, close, high_length=None, low_length=None, mamode=None, offset=None, **kwargs):
"""Indicator: Gann HiLo (HiLo)"""
# Validate Arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
high_length = int(high_length) if high_length and high_length > 0 else 13
low_length = int(low_length) if low_length and low_length > 0 else 21
mamode = mamode.lower() if isinstance(mamode, str) else "sma"
_length = max(high_length, low_length)
high = verify_series(high, _length)
low = verify_series(low, _length)
close = verify_series(close, _length)
offset = get_offset(offset)
if high is None or low is None or close is None: return
# Calculate Result
m = close.size
hilo = Series(npNaN, index=close.index)
pandas_ta/overlap/hma.py
+6 -4
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
from math import sqrt
from numpy import sqrt as npSqrt
from .wma import wma
from pandas_ta.utils import get_offset, verify_series
@@ -7,13 +7,15 @@ from pandas_ta.utils import get_offset, verify_series
def hma(close, length=None, offset=None, **kwargs):
"""Indicator: Hull Moving Average (HMA)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
half_length = int(length / 2)
sqrt_length = int(sqrt(length))
sqrt_length = int(npSqrt(length))
wmaf = wma(close=close, length=half_length)
wmas = wma(close=close, length=length)
@@ -44,7 +46,7 @@ Calculation:
length=10
WMA = Weighted Moving Average
half_length = int(0.5 * length)
sqrt_length = int(math.sqrt(length))
sqrt_length = int(sqrt(length))
wmaf = WMA(close, half_length)
wmas = WMA(close, length)
pandas_ta/overlap/hwma.py
+2 -4
View File
@@ -6,13 +6,12 @@ from pandas_ta.utils import get_offset, verify_series
def hwma(close, na=None, nb=None, nc=None, offset=None, **kwargs):
"""Indicator: Holt-Winter Moving Average"""
# Validate Arguments
close = verify_series(close)
na = float(na) if na and na > 0 and na < 1 else 0.2
nb = float(nb) if nb and nb > 0 and nb < 1 else 0.1
nc = float(nc) if nc and nc > 0 and nc < 1 else 0.1
close = verify_series(close)
offset = get_offset(offset)
# Calculate Result
last_a = last_v = 0
last_f = close[0]
@@ -24,8 +23,7 @@ def hwma(close, na=None, nb=None, nc=None, offset=None, **kwargs):
V = (1.0 - nb) * (last_v + last_a) + nb * (F - last_f)
A = (1.0 - nc) * last_a + nc * (V - last_v)
result.append((F + V + 0.5 * A))
# update values
last_a, last_f, last_v = A, F, V
last_a, last_f, last_v = A, F, V # update values
hwma = Series(result, index=close.index)
pandas_ta/overlap/ichimoku.py
+6 -3
View File
@@ -6,14 +6,17 @@ from pandas_ta.utils import get_offset, verify_series
def ichimoku(high, low, close, tenkan=None, kijun=None, senkou=None, offset=None, **kwargs):
"""Indicator: Ichimoku Kinkō Hyō (Ichimoku)"""
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
tenkan = int(tenkan) if tenkan and tenkan > 0 else 9
kijun = int(kijun) if kijun and kijun > 0 else 26
senkou = int(senkou) if senkou and senkou > 0 else 52
_length = max(tenkan, kijun, senkou)
high = verify_series(high, _length)
low = verify_series(low, _length)
close = verify_series(close, _length)
offset = get_offset(offset)
if high is None or low is None or close is None: return None, None
# Calculate Result
tenkan_sen = midprice(high=high, low=low, length=tenkan)
kijun_sen = midprice(high=high, low=low, length=kijun)
pandas_ta/overlap/kama.py
+5 -4
View File
@@ -1,22 +1,22 @@
# -*- coding: utf-8 -*-
from numpy import NaN as npNaN
from pandas import Series, DataFrame
from pandas import Series
from pandas_ta.utils import get_drift, get_offset, non_zero_range, verify_series
def kama(close, length=None, fast=None, slow=None, drift=None, offset=None, **kwargs):
"""Indicator: Kaufman's Adaptive Moving Average (KAMA)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
fast = int(fast) if fast and fast > 0 else 2
slow = int(slow) if slow and slow > 0 else 30
close = verify_series(close, max(fast, slow, length))
drift = get_drift(drift)
offset = get_offset(offset)
# Calculate Result
m = close.size
if close is None: return
# Calculate Result
def weight(length: int) -> float:
return 2 / (length + 1)
@@ -30,6 +30,7 @@ def kama(close, length=None, fast=None, slow=None, drift=None, offset=None, **kw
x = er * (fr - sr) + sr
sc = x * x
m = close.size
result = [npNaN for _ in range(0, length - 1)] + [0]
for i in range(length, m):
result.append(sc[i] * close[i] + (1 - sc[i]) * result[i - 1])
pandas_ta/overlap/linreg.py
+9 -5
View File
@@ -1,13 +1,15 @@
# -*- coding: utf-8 -*-
import math
from numpy import arctan as npAtan
from numpy import pi as npPi
from numpy import sqrt as npSqrt
from pandas_ta.utils import get_offset, verify_series
def linreg(close, length=None, offset=None, **kwargs):
"""Indicator: Linear Regression"""
# Validate arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 14
close = verify_series(close, length)
offset = get_offset(offset)
angle = kwargs.pop("angle", False)
intercept = kwargs.pop("intercept", False)
@@ -16,6 +18,8 @@ def linreg(close, length=None, offset=None, **kwargs):
slope = kwargs.pop("slope", False)
tsf = kwargs.pop("tsf", False)
if close is None: return
# Calculate Result
x = range(1, length + 1) # [1, 2, ..., n] from 1 to n keeps Sum(xy) low
x_sum = 0.5 * length * (length + 1)
@@ -34,15 +38,15 @@ def linreg(close, length=None, offset=None, **kwargs):
return b
if angle:
theta = math.atan(m)
theta = npAtan(m)
if degrees:
theta *= 180 / math.pi
theta *= 180 / npPi
return theta
if r:
y2_sum = (series * series).sum()
rn = length * xy_sum - x_sum * y_sum
rd = math.sqrt(divisor * (length * y2_sum - y_sum * y_sum))
rd = npSqrt(divisor * (length * y2_sum - y_sum * y_sum))
return rn / rd
return m * length + b if tsf else m * (length - 1) + b
pandas_ta/overlap/mcgd.py
+3 -1
View File
@@ -5,11 +5,13 @@ from pandas_ta.utils import get_offset, verify_series
def mcgd(close, length=None, offset=None, c=None, **kwargs):
"""Indicator: McGinley Dynamic Indicator"""
# Validate arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
c = float(c) if c and 0 < c <= 1 else 1
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
close = close.copy()
pandas_ta/overlap/midpoint.py
+3 -1
View File
@@ -5,11 +5,13 @@ from pandas_ta.utils import get_offset, verify_series
def midpoint(close, length=None, offset=None, **kwargs):
"""Indicator: Midpoint"""
# Validate arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 2
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
close = verify_series(close, max(length, min_periods))
offset = get_offset(offset)
if close is None: return
# Calculate Result
lowest = close.rolling(length, min_periods=min_periods).min()
highest = close.rolling(length, min_periods=min_periods).max()
pandas_ta/overlap/midprice.py
+5 -2
View File
@@ -5,12 +5,15 @@ from pandas_ta.utils import get_offset, verify_series
def midprice(high, low, length=None, offset=None, **kwargs):
"""Indicator: Midprice"""
# Validate arguments
high = verify_series(high)
low = verify_series(low)
length = int(length) if length and length > 0 else 2
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
_length = max(length, min_periods)
high = verify_series(high, _length)
low = verify_series(low, _length)
offset = get_offset(offset)
if high is None or low is None: return
# Calculate Result
lowest_low = low.rolling(length, min_periods=min_periods).min()
highest_high = high.rolling(length, min_periods=min_periods).max()
pandas_ta/overlap/pwma.py
+3 -2
View File
@@ -5,12 +5,13 @@ from pandas_ta.utils import get_offset, pascals_triangle, verify_series, weights
def pwma(close, length=None, asc=None, offset=None, **kwargs):
"""Indicator: Pascals Weighted Moving Average (PWMA)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
asc = asc if asc else True
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
triangle = pascals_triangle(n=length - 1, weighted=True)
pwma = close.rolling(length, min_periods=length).apply(weights(triangle), raw=True)
pandas_ta/overlap/rma.py
+4 -2
View File
@@ -5,10 +5,12 @@ from pandas_ta.utils import get_offset, verify_series
def rma(close, length=None, offset=None, **kwargs):
"""Indicator: wildeR's Moving Average (RMA)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
offset = get_offset(offset)
alpha = (1.0 / length) if length > 0 else 0.5
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
rma = close.ewm(alpha=alpha, min_periods=length).mean()
pandas_ta/overlap/sinwma.py
+6 -4
View File
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
from math import pi
from math import sin
from numpy import pi as npPi
from numpy import sin as npSin
from pandas import Series
from pandas_ta.utils import get_offset, verify_series, weights
@@ -8,12 +8,14 @@ from pandas_ta.utils import get_offset, verify_series, weights
def sinwma(close, length=None, offset=None, **kwargs):
"""Indicator: Sine Weighted Moving Average (SINWMA) by Everget of TradingView"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 14
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
sines = Series([sin((i + 1) * pi / (length + 1)) for i in range(0, length)])
sines = Series([npSin((i + 1) * npPi / (length + 1)) for i in range(0, length)])
w = sines / sines.sum()
sinwma = close.rolling(length, min_periods=length).apply(weights(w), raw=True)
pandas_ta/overlap/sma.py
+3 -1
View File
@@ -5,11 +5,13 @@ from pandas_ta.utils import get_offset, verify_series
def sma(close, length=None, offset=None, **kwargs):
"""Indicator: Simple Moving Average (SMA)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
close = verify_series(close, max(length, min_periods))
offset = get_offset(offset)
if close is None: return
# Calculate Result
sma = close.rolling(length, min_periods=min_periods).mean()
pandas_ta/overlap/ssf.py
+3 -2
View File
@@ -3,18 +3,19 @@ from numpy import cos as npCos
from numpy import exp as npExp
from numpy import pi as npPi
from numpy import sqrt as npSqrt
from pandas import Series
from pandas_ta.utils import get_offset, verify_series
def ssf(close, length=None, poles=None, offset=None, **kwargs):
"""Indicator: Ehler's Super Smoother Filter (SSF)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
poles = int(poles) if poles in [2, 3] else 2
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
m = close.size
ssf = close.copy()
pandas_ta/overlap/supertrend.py
+5 -3
View File
@@ -9,13 +9,15 @@ from pandas_ta.utils import get_offset, verify_series
def supertrend(high, low, close, length=None, multiplier=None, offset=None, **kwargs):
"""Indicator: Supertrend"""
# Validate Arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
length = int(length) if length and length > 0 else 7
multiplier = float(multiplier) if multiplier and multiplier > 0 else 3.0
high = verify_series(high, length)
low = verify_series(low, length)
close = verify_series(close, length)
offset = get_offset(offset)
if high is None or low is None or close is None: return
# Calculate Results
m = close.size
dir_, trend = [1] * m, [0] * m
pandas_ta/overlap/swma.py
+4 -2
View File
@@ -5,12 +5,14 @@ from pandas_ta.utils import get_offset, symmetric_triangle, verify_series, weigh
def swma(close, length=None, asc=None, offset=None, **kwargs):
"""Indicator: Symmetric Weighted Moving Average (SWMA)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
# min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
asc = asc if asc else True
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
triangle = symmetric_triangle(length, weighted=True)
swma = close.rolling(length, min_periods=length).apply(weights(triangle), raw=True)
pandas_ta/overlap/t3.py
+3 -1
View File
@@ -6,11 +6,13 @@ from pandas_ta.utils import get_offset, verify_series
def t3(close, length=None, a=None, offset=None, **kwargs):
"""Indicator: T3"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
a = float(a) if a and a > 0 and a < 1 else 0.7
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
c1 = -a * a**2
c2 = 3 * a**2 + 3 * a**3
pandas_ta/overlap/tema.py
+3 -1
View File
@@ -6,10 +6,12 @@ from pandas_ta.utils import get_offset, verify_series
def tema(close, length=None, offset=None, **kwargs):
"""Indicator: Triple Exponential Moving Average (TEMA)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
ema1 = ema(close=close, length=length, **kwargs)
ema2 = ema(close=ema1, length=length, **kwargs)
pandas_ta/overlap/trima.py
+4 -2
View File
@@ -6,10 +6,12 @@ from pandas_ta.utils import get_offset, verify_series
def trima(close, length=None, offset=None, **kwargs):
"""Indicator: Triangular Moving Average (TRIMA)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
half_length = round(0.5 * (length + 1))
sma1 = sma(close, length=half_length)
@@ -41,7 +43,7 @@ Calculation:
Default Inputs:
length=10
SMA = Simple Moving Average
half_length = math.round(0.5 * (length + 1))
half_length = round(0.5 * (length + 1))
SMA1 = SMA(close, half_length)
TRIMA = SMA(SMA1, half_length)
pandas_ta/overlap/vidya.py
+3 -1
View File
@@ -7,11 +7,13 @@ from pandas_ta.utils import get_drift, get_offset, verify_series
def vidya(close, length=None, drift=None, offset=None, **kwargs):
"""Indicator: Variable Index Dynamic Average (VIDYA)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 14
close = verify_series(close, length)
drift = get_drift(drift)
offset = get_offset(offset)
if close is None: return
def _cmo(source: Series, n:int , d: int):
"""Chande Momentum Oscillator (CMO) Patch
For some reason: from pandas_ta.momentum import cmo causes
pandas_ta/overlap/vwma.py
+4 -2
View File
@@ -6,11 +6,13 @@ from pandas_ta.utils import get_offset, verify_series
def vwma(close, volume, length=None, offset=None, **kwargs):
"""Indicator: Volume Weighted Moving Average (VWMA)"""
# Validate Arguments
close = verify_series(close)
volume = verify_series(volume)
length = int(length) if length and length > 0 else 10
close = verify_series(close, length)
volume = verify_series(volume, length)
offset = get_offset(offset)
if close is None or volume is None: return
# Calculate Result
pv = close * volume
vwma = sma(close=pv, length=length) / sma(close=volume, length=length)
pandas_ta/overlap/wma.py
+3 -1
View File
@@ -8,11 +8,13 @@ from pandas_ta.utils import get_offset, verify_series
def wma(close, length=None, asc=None, offset=None, **kwargs):
"""Indicator: Weighted Moving Average (WMA)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
asc = asc if asc else True
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
total_weight = 0.5 * length * (length + 1)
weights_ = Series(npArange(1, length + 1))
pandas_ta/overlap/zlma.py
+4 -2
View File
@@ -8,10 +8,12 @@ from pandas_ta.utils import get_offset, verify_series
def zlma(close, length=None, mamode=None, offset=None, **kwargs):
"""Indicator: Zero Lag Moving Average (ZLMA)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
offset = get_offset(offset)
mamode = mamode.lower() if isinstance(mamode, str) else "ema"
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
lag = int(0.5 * (length - 1))
pandas_ta/performance/log_return.py
+3 -1
View File
@@ -6,10 +6,12 @@ from pandas_ta.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
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
log_return = nplog(close).diff(periods=length)
pandas_ta/performance/percent_return.py
+3 -1
View File
@@ -5,10 +5,12 @@ from pandas_ta.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
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
pct_return = close.pct_change(length)
pandas_ta/performance/trend_return.py
+1 -1
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
from pandas import DataFrame, Series
from pandas import DataFrame
from .log_return import log_return
from .percent_return import percent_return
from pandas_ta.utils import get_offset, verify_series, zero
pandas_ta/statistics/entropy.py
+3 -1
View File
@@ -6,11 +6,13 @@ from pandas_ta.utils import get_offset, verify_series
def entropy(close, length=None, base=None, offset=None, **kwargs):
"""Indicator: Entropy (ENTP)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
base = float(base) if base and base > 0 else 2.0
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
p = close / close.rolling(length).sum()
entropy = (-p * npLog(p) / npLog(base)).rolling(length).sum()
pandas_ta/statistics/kurtosis.py
+3 -1
View File
@@ -5,11 +5,13 @@ from pandas_ta.utils import get_offset, verify_series
def kurtosis(close, length=None, offset=None, **kwargs):
"""Indicator: Kurtosis"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 30
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
close = verify_series(close, max(length, min_periods))
offset = get_offset(offset)
if close is None: return
# Calculate Result
kurtosis = close.rolling(length, min_periods=min_periods).kurt()
pandas_ta/statistics/mad.py
+3 -1
View File
@@ -6,11 +6,13 @@ from pandas_ta.utils import get_offset, verify_series
def mad(close, length=None, offset=None, **kwargs):
"""Indicator: Mean Absolute Deviation"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 30
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
close = verify_series(close, max(length, min_periods))
offset = get_offset(offset)
if close is None: return
# Calculate Result
def mad_(series):
"""Mean Absolute Deviation"""
pandas_ta/statistics/median.py
+3 -1
View File
@@ -5,11 +5,13 @@ from pandas_ta.utils import get_offset, verify_series
def median(close, length=None, offset=None, **kwargs):
"""Indicator: Median"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 30
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
close = verify_series(close, max(length, min_periods))
offset = get_offset(offset)
if close is None: return
# Calculate Result
median = close.rolling(length, min_periods=min_periods).median()
pandas_ta/statistics/quantile.py
+3 -1
View File
@@ -5,12 +5,14 @@ from pandas_ta.utils import get_offset, verify_series
def quantile(close, length=None, q=None, offset=None, **kwargs):
"""Indicator: Quantile"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 30
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
q = float(q) if q and q > 0 and q < 1 else 0.5
close = verify_series(close, max(length, min_periods))
offset = get_offset(offset)
if close is None: return
# Calculate Result
quantile = close.rolling(length, min_periods=min_periods).quantile(q)
pandas_ta/statistics/skew.py
+3 -1
View File
@@ -5,11 +5,13 @@ from pandas_ta.utils import get_offset, verify_series
def skew(close, length=None, offset=None, **kwargs):
"""Indicator: Skew"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 30
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
close = verify_series(close, max(length, min_periods))
offset = get_offset(offset)
if close is None: return
# Calculate Result
skew = close.rolling(length, min_periods=min_periods).skew()
pandas_ta/statistics/stdev.py
+3 -1
View File
@@ -7,11 +7,13 @@ from pandas_ta.utils import get_offset, verify_series
def stdev(close, length=None, ddof=1, offset=None, **kwargs):
"""Indicator: Standard Deviation"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 30
ddof = int(ddof) if ddof >= 0 and ddof < length else 1
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
stdev = variance(close=close, length=length, ddof=ddof).apply(npsqrt)
pandas_ta/statistics/variance.py
+3 -2
View File
@@ -5,13 +5,14 @@ from pandas_ta.utils import get_offset, verify_series
def variance(close, length=None, ddof=None, offset=None, **kwargs):
"""Indicator: Variance"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 1 else 30
ddof = int(ddof) if ddof and ddof >= 0 and ddof < length else 0
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
close = verify_series(close, max(length, min_periods))
offset = get_offset(offset)
if close is None: return
# Calculate Result
variance = close.rolling(length, min_periods=min_periods).var(ddof)
pandas_ta/statistics/zscore.py
+3 -1
View File
@@ -7,11 +7,13 @@ from pandas_ta.utils import get_offset, verify_series
def zscore(close, length=None, std=None, offset=None, **kwargs):
"""Indicator: Z Score"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 1 else 30
std = float(std) if std and std > 1 else 1
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
std *= stdev(close=close, length=length, **kwargs)
mean = sma(close=close, length=length, **kwargs)
pandas_ta/trend/adx.py
+5 -3
View File
@@ -8,14 +8,16 @@ from pandas_ta.utils import get_drift, get_offset, verify_series, zero
def adx(high, low, close, length=None, scalar=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
scalar = float(scalar) if scalar else 100
high = verify_series(high, length)
low = verify_series(low, length)
close = verify_series(close, length)
drift = get_drift(drift)
offset = get_offset(offset)
if high is None or low is None or close is None: return
# Calculate Result
atr_ = atr(high=high, low=low, close=close, length=length)
pandas_ta/trend/amat.py
+3 -1
View File
@@ -9,14 +9,16 @@ from pandas_ta.utils import get_offset, verify_series
def amat(close=None, fast=None, slow=None, mamode=None, lookback=None, slope_length=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.lower() if isinstance(mamode, str) else "ema"
close = verify_series(close, max(fast, slow, lookback))
offset = get_offset(offset)
if "length" in kwargs: kwargs.pop("length")
if close is None: return
# # Calculate Result
fast_ma = ma(mamode, close, length=fast, **kwargs)
slow_ma = ma(mamode, close, length=slow, **kwargs)
pandas_ta/trend/aroon.py
+4 -2
View File
@@ -7,12 +7,14 @@ from pandas_ta.utils import recent_maximum_index, recent_minimum_index
def aroon(high, low, length=None, scalar=None, offset=None, **kwargs):
"""Indicator: Aroon & Aroon Oscillator"""
# Validate Arguments
high = verify_series(high)
low = verify_series(low)
length = length if length and length > 0 else 14
scalar = float(scalar) if scalar else 100
high = verify_series(high, length)
low = verify_series(low, length)
offset = get_offset(offset)
if high is None or low is None: return
# Calculate Result
periods_from_hh = high.rolling(length + 1).apply(recent_maximum_index, raw=True)
periods_from_ll = low.rolling(length + 1).apply(recent_minimum_index, raw=True)
pandas_ta/trend/chop.py
+5 -4
View File
@@ -1,6 +1,5 @@
# -*- coding: utf-8 -*-
from numpy import log10 as npLog10
from pandas import DataFrame
from pandas_ta.volatility import atr
from pandas_ta.utils import get_offset, get_drift, verify_series
@@ -8,15 +7,17 @@ from pandas_ta.utils import get_offset, get_drift, verify_series
def chop(high, low, close, length=None, atr_length=None, scalar=None, drift=None, offset=None, **kwargs):
"""Indicator: Choppiness Index (CHOP)"""
# Validate Arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
length = int(length) if length and length > 0 else 14
atr_length = int(atr_length) if atr_length is not None and atr_length > 0 else 1
scalar = float(scalar) if scalar else 100
high = verify_series(high, length)
low = verify_series(low, length)
close = verify_series(close, length)
drift = get_drift(drift)
offset = get_offset(offset)
if high is None or low is None or close is None: return
# Calculate Result
diff = high.rolling(length).max() - low.rolling(length).min()
pandas_ta/trend/cksp.py
+6 -3
View File
@@ -7,14 +7,17 @@ from pandas_ta.utils import get_offset, verify_series
def cksp(high, low, close, p=None, x=None, q=None, offset=None, **kwargs):
"""Indicator: Chande Kroll Stop (CKSP)"""
# Validate Arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
p = int(p) if p and p > 0 else 10
x = float(x) if x and x > 0 else 1
q = int(q) if q and q > 0 else 9
_length = max(p, q, x)
high = verify_series(high, _length)
low = verify_series(low, _length)
close = verify_series(close, _length)
offset = get_offset(offset)
if high is None or low is None or close is None: return
# Calculate Result
atr_ = atr(high=high, low=low, close=close, length=p)
pandas_ta/trend/decay.py
+5 -3
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
from math import exp
from numpy import exp as npExp
from pandas import DataFrame
from pandas_ta.utils import get_offset, verify_series
@@ -7,16 +7,18 @@ from pandas_ta.utils import get_offset, verify_series
def decay(close, kind=None, length=None, mode=None, offset=None, **kwargs):
"""Indicator: Decay"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 5
mode = mode.lower() if isinstance(mode, str) else "linear"
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
_mode = "L"
if mode == "exp" or kind == "exponential":
_mode = "EXP"
diff = close.shift(1) - exp(-length)
diff = close.shift(1) - npExp(-length)
else: # "linear"
diff = close.shift(1) - (1 / length)
diff[0] = close[0]
pandas_ta/trend/decreasing.py
+3 -1
View File
@@ -5,12 +5,14 @@ from pandas_ta.utils import get_offset, verify_series
def decreasing(close, length=None, strict=None, asint=None, offset=None, **kwargs):
"""Indicator: Decreasing"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 1
strict = strict if isinstance(strict, bool) else False
asint = asint if isinstance(asint, bool) else True
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
def stricly_decreasing(series, n):
return all([i > j for i,j in zip(series[-n:], series[1:])])
pandas_ta/trend/dpo.py
+3 -1
View File
@@ -6,10 +6,12 @@ from pandas_ta.utils import get_offset, verify_series
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 20
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
t = int(0.5 * length) + 1
ma = sma(close, length)
pandas_ta/trend/increasing.py
+3 -1
View File
@@ -5,12 +5,14 @@ from pandas_ta.utils import get_offset, verify_series
def increasing(close, length=None, strict=None, asint=None, offset=None, **kwargs):
"""Indicator: Increasing"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 1
strict = strict if isinstance(strict, bool) else False
asint = asint if isinstance(asint, bool) else True
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
def stricly_increasing(series, n):
return all([i < j for i,j in zip(series[-n:], series[1:])])
pandas_ta/trend/long_run.py
+4 -2
View File
@@ -7,11 +7,13 @@ from pandas_ta.utils import get_offset, verify_series
def long_run(fast, slow, length=None, offset=None, **kwargs):
"""Indicator: Long Run"""
# Validate Arguments
fast = verify_series(fast)
slow = verify_series(slow)
length = int(length) if length and length > 0 else 2
fast = verify_series(fast, length)
slow = verify_series(slow, length)
offset = get_offset(offset)
if fast is None or slow is None: return
# Calculate Result
pb = increasing(fast, length) & decreasing(slow, length) # potential bottom or bottom
bi = increasing(fast, length) & increasing(slow, length) # fast and slow are increasing
pandas_ta/trend/qstick.py
+5 -3
View File
@@ -6,11 +6,13 @@ from pandas_ta.utils import get_offset, non_zero_range, verify_series
def qstick(open_, close, length=None, offset=None, **kwargs):
"""Indicator: Q Stick"""
# Validate Arguments
open_ = verify_series(open_)
close = verify_series(close)
length = int(length) if length and length > 0 else 10
offset = get_offset(offset)
ma = kwargs.pop("ma", "sma")
open_ = verify_series(open_, length)
close = verify_series(close, length)
offset = get_offset(offset)
if open_ is None or close is None: return
# Calculate Result
diff = non_zero_range(close, open_)
pandas_ta/trend/short_run.py
+4 -2
View File
@@ -7,11 +7,13 @@ from pandas_ta.utils import get_offset, verify_series
def short_run(fast, slow, length=None, offset=None, **kwargs):
"""Indicator: Short Run"""
# Validate Arguments
fast = verify_series(fast)
slow = verify_series(slow)
length = int(length) if length and length > 0 else 2
fast = verify_series(fast, length)
slow = verify_series(slow, length)
offset = get_offset(offset)
if fast is None or slow is None: return
# Calculate Result
pt = decreasing(fast, length) & increasing(slow, length) # potential top or top
bd = decreasing(fast, length) & decreasing(slow, length) # fast and slow are decreasing
pandas_ta/trend/ttm_trend.py
+5 -3
View File
@@ -7,12 +7,14 @@ from pandas_ta.utils import get_offset, verify_series
def ttm_trend(high, low, close, length=None, offset=None, **kwargs):
"""Indicator: TTM Trend (TTM_TRND)"""
# Validate arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
length = int(length) if length and length > 0 else 6
high = verify_series(high, length)
low = verify_series(low, length)
close = verify_series(close, length)
offset = get_offset(offset)
if high is None or low is None or close is None: return
# Calculate Result
trend_avg = hl2(high, low)
for i in range(1, length):
pandas_ta/trend/vortex.py
+6 -3
View File
@@ -7,14 +7,17 @@ from pandas_ta.utils import get_drift, get_offset, verify_series, zero
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
_length = max(length, min_periods)
high = verify_series(high, _length)
low = verify_series(low, _length)
close = verify_series(close, _length)
drift = get_drift(drift)
offset = get_offset(offset)
if high is None or low is None or close is None: return
# Calculate Result
tr = true_range(high=high, low=low, close=close)
tr_sum = tr.rolling(length, min_periods=min_periods).sum()
pandas_ta/utils/_candles.py
+1 -1
View File
@@ -15,5 +15,5 @@ def high_low_range(high: Series, low: Series) -> Series:
return non_zero_range(high, low)
def real_body(close: Series, open_: Series) -> Series:
def real_body(open_: Series, close: Series) -> Series:
return non_zero_range(close, open_)
pandas_ta/utils/_core.py
+5 -3
View File
@@ -3,6 +3,7 @@ from pathlib import Path
from sys import float_info as sflt
from numpy import argmax, argmin
from numpy import NaN as npNaN
from pandas import DataFrame, Series
from pandas.api.types import is_datetime64_any_dtype
@@ -105,7 +106,8 @@ def unsigned_differences(series: Series, amount: int = None, **kwargs) -> Series
return positive, negative
def verify_series(series: Series) -> Series:
"""If a Pandas Series return it."""
def verify_series(series: Series, min_length: int = None) -> Series:
"""If a Pandas Series and it meets the min_length of the indicator return it."""
has_length = min_length is not None and isinstance(min_length, int)
if series is not None and isinstance(series, Series):
return series
return None if has_length and series.size < min_length else series
pandas_ta/utils/_math.py
+10 -8
View File
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
from functools import reduce
from math import fabs, floor
from math import floor as mfloor
from operator import mul
from sys import float_info as sflt
from typing import List, Optional, Tuple
@@ -11,6 +11,8 @@ from numpy import append as npAppend
from numpy import array as npArray
from numpy import corrcoef as npCorrcoef
from numpy import dot as npDot
from numpy import fabs as npFabs
from numpy import floor as npFloor
from numpy import exp as npExp
from numpy import log as npLog
from numpy import NaN as npNaN
@@ -27,8 +29,8 @@ from ._core import verify_series
def combination(**kwargs: dict) -> int:
"""https://stackoverflow.com/questions/4941753/is-there-a-math-ncr-function-in-python"""
n = int(fabs(kwargs.pop("n", 1)))
r = int(fabs(kwargs.pop("r", 0)))
n = int(npFabs(kwargs.pop("n", 1)))
r = int(npFabs(kwargs.pop("r", 0)))
if kwargs.pop("repetition", False) or kwargs.pop("multichoose", False):
n = n + r - 1
@@ -45,7 +47,7 @@ def combination(**kwargs: dict) -> int:
def fibonacci(n: int = 2, **kwargs: dict) -> npNdArray:
"""Fibonacci Sequence as a numpy array"""
n = int(fabs(n)) if n >= 0 else 2
n = int(npFabs(n)) if n >= 0 else 2
zero = kwargs.pop("zero", False)
if zero:
@@ -119,7 +121,7 @@ def pascals_triangle(n: int = None, **kwargs: dict) -> npNdArray:
=> weighted: [0.0625, 0.25, 0.375, 0.25, 0.0625]
=> inverse weighted: [0.9375, 0.75, 0.625, 0.75, 0.9375]
"""
n = int(fabs(n)) if n is not None else 0
n = int(npFabs(n)) if n is not None else 0
# Calculation
triangle = npArray([combination(n=n, r=i) for i in range(0, n + 1)])
@@ -146,7 +148,7 @@ def symmetric_triangle(n: int = None, **kwargs: dict) -> Optional[List[int]]:
n=4 => triangle: [1, 2, 2, 1]
=> weighted: [0.16666667 0.33333333 0.33333333 0.16666667]
"""
n = int(fabs(n)) if n is not None else 2
n = int(npFabs(n)) if n is not None else 2
triangle = None
if n == 2:
@@ -154,10 +156,10 @@ def symmetric_triangle(n: int = None, **kwargs: dict) -> Optional[List[int]]:
if n > 2:
if n % 2 == 0:
front = [i + 1 for i in range(0, floor(n / 2))]
front = [i + 1 for i in range(0, mfloor(n / 2))]
triangle = front + front[::-1]
else:
front = [i + 1 for i in range(0, floor(0.5 * (n + 1)))]
front = [i + 1 for i in range(0, mfloor(0.5 * (n + 1)))]
triangle = front.copy()
front.pop()
triangle += front[::-1]
pandas_ta/utils/_time.py
+11 -5
View File
@@ -3,9 +3,10 @@ from datetime import datetime
from time import localtime, perf_counter
from typing import Tuple
from pandas import DataFrame, Series, Timestamp
from pandas import DataFrame, DatetimeIndex, Timestamp
from pandas_ta import EXCHANGE_TZ, RATE
from pandas_ta.utils import verify_series
def df_dates(df: DataFrame, dates: Tuple[str, list] = None) -> DataFrame:
@@ -18,6 +19,10 @@ def df_dates(df: DataFrame, dates: Tuple[str, list] = None) -> DataFrame:
def df_month_to_date(df: DataFrame) -> DataFrame:
"""Yields the Month-to-Date (MTD) DataFrame"""
# if df.empty: print("[X] Month-to-Date not in range"); return
# print(df.shape)
# print(type(df))
# print(df)
return df[df.index >= Timestamp.now().strftime("%Y-%m-01")]
@@ -70,11 +75,12 @@ def get_time(exchange: str = "NYSE", full:bool = True, to_string:bool = False) -
return s if to_string else print(s)
def total_time(series: Series, tf: str = "years") -> float:
"""Calculates the total time of a Series. Options: 'months', 'weeks',
'days', 'hours', 'minutes' and 'seconds'. Default: 'years'.
def total_time(df: DataFrame, tf: str = "years") -> float:
"""Calculates the total time of a DataFrame. Difference of the Last and
First index. Options: 'months', 'weeks', 'days', 'hours', 'minutes'
and 'seconds'. Default: 'years'.
Useful for annualization."""
time_diff = series.index[-1] - series.index[0]
time_diff = df.index[-1] - df.index[0]
TimeFrame = {
"years": time_diff.days / RATE["TRADING_DAYS_PER_YEAR"],
"months": time_diff.days / 30.417,
pandas_ta/volatility/aberration.py
+7 -4
View File
@@ -3,19 +3,22 @@
from pandas import DataFrame
from .atr import atr
from pandas_ta.overlap import hlc3, sma
from pandas_ta.utils import get_offset, non_zero_range, verify_series
from pandas_ta.utils import get_offset, verify_series
def aberration(high, low, close, length=None, atr_length=None, offset=None, **kwargs):
"""Indicator: Aberration (ABER)"""
# Validate arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
length = int(length) if length and length > 0 else 5
atr_length = int(atr_length) if atr_length and atr_length > 0 else 15
_length = max(atr_length, length)
high = verify_series(high, _length)
low = verify_series(low, _length)
close = verify_series(close, _length)
offset = get_offset(offset)
if high is None or low is None or close is None: return
# Calculate Result
atr_ = atr(high=high, low=low, close=close, length=atr_length)
jg = hlc3(high=high, low=low, close=close)
pandas_ta/volatility/accbands.py
+6 -4
View File
@@ -7,17 +7,19 @@ from pandas_ta.utils import get_drift, get_offset, non_zero_range, verify_series
def accbands(high, low, close, length=None, c=None, drift=None, mamode=None, offset=None, **kwargs):
"""Indicator: Acceleration Bands (ACCBANDS)"""
# Validate arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
high_low_range = non_zero_range(high, low)
length = int(length) if length and length > 0 else 20
c = float(c) if c and c > 0 else 4
mamode = mamode if isinstance(mamode, str) else "sma"
high = verify_series(high, length)
low = verify_series(low, length)
close = verify_series(close, length)
drift = get_drift(drift)
offset = get_offset(offset)
if high is None or low is None or close is None: return
# Calculate Result
high_low_range = non_zero_range(high, low)
hl_ratio = high_low_range / (high + low)
hl_ratio *= c
_lower = low * (1 - hl_ratio)
pandas_ta/volatility/atr.py
+5 -3
View File
@@ -7,14 +7,16 @@ from pandas_ta.utils import get_drift, get_offset, verify_series
def atr(high, low, close, length=None, mamode=None, drift=None, offset=None, **kwargs):
"""Indicator: Average True Range (ATR)"""
# Validate arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
length = int(length) if length and length > 0 else 14
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)
if high is None or low is None or close is None: return
# Calculate Result
tr = true_range(high=high, low=low, close=close, drift=drift)
atr = ma(mamode, tr, length=length)

Some files were not shown because too many files have changed in this diff Show More