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

Merge branch 'development' into main

Browse Source
This commit is contained in:
Kevin Johnson
2021-07-28 11:50:31 -07:00
parent e44ff1b30e 0d9c5b9668
commit bc3b292bf1
90 changed files with 1182 additions and 350 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3 -2
View File
@@ -131,7 +131,6 @@ AlphaVantageAPI/
# Data & NB Exclusions
*.csv
jnb/*.ipynb
data/datas.csv
data/f500.csv
data/GLD_D_tv.csv
@@ -142,4 +141,6 @@ data/SPY_D_TV2.csv
data/SPY_D_TV3.csv
data/TV_5min.csv
data/tulip.csv
examples/*.csv
examples/*.csv
jnb/*.ipynb
jnb/*.txt
README.md
+40 -16
View File
File diff suppressed because one or more lines are too long
examples/ni.py
+79
View File
@@ -0,0 +1,79 @@
# -*- coding: utf-8 -*-
from pandas_ta.overlap import sma
from pandas_ta.utils import get_offset, verify_series
# - Standard definition of your custom indicator function (including docs)-
def ni(close, length=None, centered=False, offset=None, **kwargs):
"""
Example indicator ni
"""
# Validate Arguments
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)
ni = close - ma.shift(t)
if centered:
ni = (close.shift(t) - ma).shift(-t)
# Offset
if offset != 0:
ni = ni.shift(offset)
# Handle fills
if "fillna" in kwargs:
ni.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
ni.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
ni.name = f"ni_{length}"
ni.category = "trend"
return ni
ni.__doc__ = \
"""Example indicator (NI)
Is an indicator provided solely as an example
Sources:
https://github.com/twopirllc/pandas-ta/issues/264
Calculation:
Default Inputs:
length=20, centered=False
SMA = Simple Moving Average
t = int(0.5 * length) + 1
ni = close.shift(t) - SMA(close, length)
if centered:
ni = ni.shift(-t)
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 20
centered (bool): Shift the ni back by int(0.5 * length) + 1. Default: False
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.Series: New feature generated.
"""
# - Define a matching class method --------------------------------------------
def ni_method(self, length=None, offset=None, **kwargs):
close = self._get_column(kwargs.pop("close", "close"))
result = ni(close=close, length=length, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
pandas_ta/__init__.py
+3 -3
View File
@@ -55,9 +55,9 @@ Category = {
# Overlap
"overlap": [
"alma", "dema", "ema", "fwma", "hilo", "hl2", "hlc3", "hma", "ichimoku",
"kama", "linreg", "mcgd", "midpoint", "midprice", "ohlc4", "pwma", "rma",
"sinwma", "sma", "ssf", "supertrend", "swma", "t3", "tema", "trima",
"vidya", "vwap", "vwma", "wcp", "wma", "zlma"
"jma", "kama", "linreg", "mcgd", "midpoint", "midprice", "ohlc4",
"pwma", "rma", "sinwma", "sma", "ssf", "supertrend", "swma", "t3",
"tema", "trima", "vidya", "vwap", "vwma", "wcp", "wma", "zlma"
],
# Performance
"performance": ["log_return", "percent_return"],
pandas_ta/core.py
+38 -30
View File
@@ -4,6 +4,7 @@ from multiprocessing import cpu_count, Pool
from pathlib import Path
from time import perf_counter
from typing import List, Tuple
from warnings import simplefilter
import pandas as pd
from numpy import log10 as npLog10
@@ -250,15 +251,15 @@ class AnalysisIndicators(BasePandasObject):
_time_range = "years"
_last_run = get_time(_exchange, to_string=True)
# def __init__(self, pandas_obj):
# # self._validate(pandas_obj)
# self._df = pandas_obj
# self._last_run = get_time(self._exchange, to_string=True)
def __init__(self, pandas_obj):
self._validate(pandas_obj)
self._df = pandas_obj
self._last_run = get_time(self._exchange, to_string=True)
# @staticmethod
# def _validate(df: Tuple[pd.DataFrame, pd.Series]):
# if isinstance(df, pd.Series) or isinstance(df, pd.DataFrame):
# raise AttributeError("[X] Must be either a Pandas Series or DataFrame.")
@staticmethod
def _validate(obj: Tuple[pd.DataFrame, pd.Series]):
if not isinstance(obj, pd.DataFrame) and not isinstance(obj, pd.Series):
raise AttributeError("[X] Must be either a Pandas Series or DataFrame.")
# DataFrame Behavioral Methods
def __call__(
@@ -400,8 +401,9 @@ class AnalysisIndicators(BasePandasObject):
df = self._df
if df is None or result is None: return
else:
simplefilter(action="ignore", category=pd.errors.PerformanceWarning)
if "col_names" in kwargs and not isinstance(kwargs["col_names"], tuple):
kwargs["col_names"] = (kwargs["col_names"],)
kwargs["col_names"] = (kwargs["col_names"],) # Note: tuple(kwargs["col_names"]) doesn't work
if isinstance(result, pd.DataFrame):
# If specified in kwargs, rename the columns.
@@ -761,10 +763,10 @@ class AnalysisIndicators(BasePandasObject):
else:
# Without multiprocessing:
if verbose:
_col_msg = f"[i] No mulitproccessing (cores = 0)."
if has_col_names:
print(f"[i] No mulitproccessing support for 'col_names' option.")
else:
print(f"[i] No mulitproccessing (cores = 0).")
_col_msg = f"[i] No mulitproccessing support for 'col_names' option."
print(_col_msg)
if mode["custom"]:
if Imports["tqdm"] and verbose:
@@ -784,6 +786,7 @@ class AnalysisIndicators(BasePandasObject):
else:
for ind in ta:
getattr(self, ind)(*tuple(), **kwargs)
self._last_run = get_time(self.exchange, to_string=True)
# Apply prefixes/suffixes and appends indicator results to the DataFrame
[self._post_process(r, **kwargs) for r in results]
@@ -900,9 +903,9 @@ class AnalysisIndicators(BasePandasObject):
result = ao(high=high, low=low, fast=fast, slow=slow, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def apo(self, fast=None, slow=None, offset=None, **kwargs):
def apo(self, fast=None, slow=None, mamode=None, offset=None, **kwargs):
close = self._get_column(kwargs.pop("close", "close"))
result = apo(close=close, fast=fast, slow=slow, offset=offset, **kwargs)
result = apo(close=close, fast=fast, slow=slow, mamode=mamode, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def bias(self, length=None, mamode=None, offset=None, **kwargs):
@@ -958,10 +961,10 @@ class AnalysisIndicators(BasePandasObject):
result = cti(close=close, length=length, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def dm(self, drift=None, offset=None, **kwargs):
def dm(self, drift=None, offset=None, mamode=None, **kwargs):
high = self._get_column(kwargs.pop("high", "high"))
low = self._get_column(kwargs.pop("low", "low"))
result = dm(high=high, low=low, drift=drift, offset=offset, **kwargs)
result = dm(high=high, low=low, drift=drift, mamode=mamode, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def er(self, length=None, drift=None, offset=None, **kwargs):
@@ -1078,18 +1081,18 @@ class AnalysisIndicators(BasePandasObject):
result = smi(close=close, fast=fast, slow=slow, signal=signal, scalar=scalar, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def squeeze(self, bb_length=None, bb_std=None, kc_length=None, kc_scalar=None, mom_length=None, mom_smooth=None, use_tr=None, offset=None, **kwargs):
def squeeze(self, bb_length=None, bb_std=None, kc_length=None, kc_scalar=None, mom_length=None, mom_smooth=None, use_tr=None, mamode=None, offset=None, **kwargs):
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 = squeeze(high=high, low=low, close=close, bb_length=bb_length, bb_std=bb_std, kc_length=kc_length, kc_scalar=kc_scalar, mom_length=mom_length, mom_smooth=mom_smooth, use_tr=use_tr, offset=offset, **kwargs)
result = squeeze(high=high, low=low, close=close, bb_length=bb_length, bb_std=bb_std, kc_length=kc_length, kc_scalar=kc_scalar, mom_length=mom_length, mom_smooth=mom_smooth, use_tr=use_tr, mamode=mamode, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def squeeze_pro(self, bb_length=None, bb_std=None, kc_length=None, kc_scalar_wide=None, kc_scalar_normal=None, kc_scalar_narrow=None, mom_length=None, mom_smooth=None, use_tr=None, offset=None, **kwargs):
def squeeze_pro(self, bb_length=None, bb_std=None, kc_length=None, kc_scalar_wide=None, kc_scalar_normal=None, kc_scalar_narrow=None, mom_length=None, mom_smooth=None, use_tr=None, mamode=None, offset=None, **kwargs):
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 = squeeze_pro(high=high, low=low, close=close, bb_length=bb_length, bb_std=bb_std, kc_length=kc_length, kc_scalar_wide=kc_scalar_wide, kc_scalar_normal=kc_scalar_normal, kc_scalar_narrow=kc_scalar_narrow, mom_length=mom_length, mom_smooth=mom_smooth, use_tr=use_tr, offset=offset, **kwargs)
result = squeeze_pro(high=high, low=low, close=close, bb_length=bb_length, bb_std=bb_std, kc_length=kc_length, kc_scalar_wide=kc_scalar_wide, kc_scalar_normal=kc_scalar_normal, kc_scalar_narrow=kc_scalar_narrow, mom_length=mom_length, mom_smooth=mom_smooth, use_tr=use_tr, mamode=mamode, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def stc(self, ma1=None, ma2=None, osc=None, tclength=None, fast=None, slow=None, factor=None, offset=None, **kwargs):
@@ -1097,18 +1100,18 @@ class AnalysisIndicators(BasePandasObject):
result = stc(close=close, ma1=ma1, ma2=ma2, osc=osc, tclength=tclength, fast=fast, slow=slow, factor=factor, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def stoch(self, fast_k=None, slow_k=None, slow_d=None, offset=None, **kwargs):
def stoch(self, fast_k=None, slow_k=None, slow_d=None, mamode=None, offset=None, **kwargs):
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 = stoch(high=high, low=low, close=close, fast_k=fast_k, slow_k=slow_k, slow_d=slow_d, offset=offset, **kwargs)
result = stoch(high=high, low=low, close=close, fast_k=fast_k, slow_k=slow_k, slow_d=slow_d, mamode=mamode, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def stochrsi(self, length=None, rsi_length=None, k=None, d=None, offset=None, **kwargs):
def stochrsi(self, length=None, rsi_length=None, k=None, d=None, mamode=None, offset=None, **kwargs):
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 = stochrsi(high=high, low=low, close=close, length=length, rsi_length=rsi_length, k=k, d=d, offset=offset, **kwargs)
result = stochrsi(high=high, low=low, close=close, length=length, rsi_length=rsi_length, k=k, d=d, mamode=mamode, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def td_seq(self, asint=None, offset=None, show_all=None, **kwargs):
@@ -1121,9 +1124,9 @@ class AnalysisIndicators(BasePandasObject):
result = trix(close=close, length=length, signal=signal, scalar=scalar, drift=drift, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def tsi(self, fast=None, slow=None, drift=None, offset=None, **kwargs):
def tsi(self, fast=None, slow=None, drift=None, mamode=None, offset=None, **kwargs):
close = self._get_column(kwargs.pop("close", "close"))
result = tsi(close=close, fast=fast, slow=slow, drift=drift, offset=offset, **kwargs)
result = tsi(close=close, fast=fast, slow=slow, drift=drift, mamode=mamode, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def uo(self, fast=None, medium=None, slow=None, fast_w=None, medium_w=None, slow_w=None, drift=None, offset=None, **kwargs):
@@ -1191,16 +1194,21 @@ class AnalysisIndicators(BasePandasObject):
result = hwma(close=close, na=na, nb=nb, nc=nc, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def jma(self, length=None, phase=None, offset=None, **kwargs):
close = self._get_column(kwargs.pop("close", "close"))
result = jma(close=close, length=length, phase=phase, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def kama(self, length=None, fast=None, slow=None, offset=None, **kwargs):
close = self._get_column(kwargs.pop("close", "close"))
result = kama(close=close, length=length, fast=fast, slow=slow, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def ichimoku(self, tenkan=None, kijun=None, senkou=None, offset=None, **kwargs):
def ichimoku(self, tenkan=None, kijun=None, senkou=None, include_chikou=True, offset=None, **kwargs):
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, span = ichimoku(high=high, low=low, close=close, tenkan=tenkan, kijun=kijun, senkou=senkou, offset=offset, **kwargs)
result, span = ichimoku(high=high, low=low, close=close, tenkan=tenkan, kijun=kijun, senkou=senkou, include_chikou=include_chikou, offset=offset, **kwargs)
self._add_prefix_suffix(result, **kwargs)
self._add_prefix_suffix(span, **kwargs)
self._append(result, **kwargs)
@@ -1416,11 +1424,11 @@ class AnalysisIndicators(BasePandasObject):
result = chop(high=high, low=low, close=close, length=length, atr_length=atr_length, scalar=scalar, drift=drift, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def cksp(self, p=None, x=None, q=None, offset=None, **kwargs):
def cksp(self, p=None, x=None, q=None, mamode=None, offset=None, **kwargs):
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 = cksp(high=high, low=low, close=close, p=p, x=x, q=q, offset=offset, **kwargs)
result = cksp(high=high, low=low, close=close, p=p, x=x, q=q, mamode=mamode, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def decay(self, length=None, mode=None, offset=None, **kwargs):
pandas_ta/custom.py
+226
View File
@@ -0,0 +1,226 @@
# -*- coding: utf-8 -*-
import importlib
import os
import sys
import types
from os.path import abspath, join, exists, basename, splitext
from glob import glob
import pandas_ta
from pandas_ta import AnalysisIndicators
def bind(function_name, function, method):
"""
Helper function to bind the function and class method defined in a custom
indicator module to the active pandas_ta instance.
Args:
function_name (str): The name of the indicator within pandas_ta
function (fcn): The indicator function
method (fcn): The class method corresponding to the passed function
"""
setattr(pandas_ta, function_name, function)
setattr(AnalysisIndicators, function_name, method)
def create_dir(path, create_categories=True, verbose=True):
"""
Helper function to setup a suitable folder structure for working with
custom indicators. You only need to call this once whenever you want to
setup a new custom indicators folder.
Args:
path (str): Full path to where you want your indicator tree
create_categories (bool): If True create category sub-folders
verbose (bool): If True print verbose output of results
"""
# ensure that the passed directory exists / is readable
if not exists(path):
os.makedirs(path)
if verbose:
print(f"[i] Created main directory '{path}'.")
# list the contents of the directory
# dirs = glob(abspath(join(path, '*')))
# optionally add any missing category subdirectories
if create_categories:
for sd in [*pandas_ta.Category]:
d = abspath(join(path, sd))
if not exists(d):
os.makedirs(d)
if verbose:
dirname = basename(d)
print(f"[i] Created an empty sub-directory '{dirname}'.")
def get_module_functions(module):
"""
Helper function to get the functions of an imported module as a dictionary.
Args:
module: python module
Returns:
dict: module functions mapping
{
"func1_name": func1,
"func2_name": func2,...
}
"""
module_functions = {}
for name, item in vars(module).items():
if isinstance(item, types.FunctionType):
module_functions[name] = item
return module_functions
def import_dir(path, verbose=True):
# ensure that the passed directory exists / is readable
if not exists(path):
print(f"[X] Unable to read the directory '{path}'.")
return
# list the contents of the directory
dirs = glob(abspath(join(path, "*")))
# traverse full directory, importing all modules found there
for d in dirs:
dirname = basename(d)
# only look in directories which are valid pandas_ta categories
if dirname not in [*pandas_ta.Category]:
if verbose:
print(f"[i] Skipping the sub-directory '{dirname}' since it's not a valid pandas_ta category.")
continue
# for each module found in that category (directory)...
for module in glob(abspath(join(path, dirname, "*.py"))):
module_name = splitext(basename(module))[0]
# ensure that the supplied path is included in our python path
if d not in sys.path:
sys.path.append(d)
# (re)load the indicator module
module_functions = load_indicator_module(module_name)
# figure out which of the modules functions to bind to pandas_ta
fcn_callable = module_functions.get(module_name, None)
fcn_method_callable = module_functions.get(f"{module_name}_method", None)
if fcn_callable == None:
print(f"[X] Unable to find a function named '{module_name}' in the module '{module_name}.py'.")
continue
if fcn_method_callable == None:
missing_method = f"{module_name}_method"
print(f"[X] Unable to find a method function named '{missing_method}' in the module '{module_name}.py'.")
continue
# add it to the correct category if it's not there yet
if module_name not in pandas_ta.Category[dirname]:
pandas_ta.Category[dirname].append(module_name)
bind(module_name, fcn_callable, fcn_method_callable)
if verbose:
print(f"[i] Successfully imported the custom indicator '{module}' into category '{dirname}'.")
import_dir.__doc__ = \
"""
Import a directory of custom indicators into pandas_ta
Args:
path (str): Full path to your indicator tree
verbose (bool): If True verbose output of results
This method allows you to experiment and develop your own technical analysis
indicators in a separate local directory of your choice but use them seamlessly
together with the existing pandas_ta functions just like if they were part of
pandas_ta.
If you at some late point would like to push them into the pandas_ta library
you can do so very easily by following the step by step instruction here
https://github.com/twopirllc/pandas-ta/issues/355.
A brief example of usage:
1. Loading the 'ta' module:
>>> import pandas as pd
>>> import pandas_ta as ta
2. Create an empty directory on your machine where you want to work with your
indicators. Invoke pandas_ta.custom.import_dir once to pre-populate it with
sub-folders for all available indicator categories, e.g.:
>>> import os
>>> from os.path import abspath, join, expanduser
>>> from pandas_ta.custom import create_dir, import_dir
>>> ta_dir = abspath(join(expanduser("~"), "my_indicators"))
>>> create_dir(ta_dir)
3. You can now create your own custom indicator e.g. by copying existing
ones from pandas_ta core module and modifying them.
IMPORTANT: Each custom indicator should have a unique name and have both
a) a function named exactly as the module, e.g. 'ni' if the module is ni.py
b) a matching method used by AnalysisIndicators named as the module but
ending with '_method'. E.g. 'ni_method'
In essence these modules should look exactly like the standard indicators
available in categories under the pandas_ta-folder. The only difference will
be an addition of a matching class method.
For an example of the correct structure, look at the example ni.py in the
examples folder.
The ni.py indicator is a trend indicator so therefore we drop it into the
sub-folder named trend. Thus we have a folder structure like this:
~/my_indicators/
│
├── candles/
.
.
└── trend/
. └── ni.py
.
└── volume/
4. We can now dynamically load all our custom indicators located in our
designated indicators directory like this:
>>> import_dir(ta_dir)
If your custom indicator(s) loaded succesfully then it should behave exactly
like all other native indicators in pandas_ta, including help functions.
"""
def load_indicator_module(name):
"""
Helper function to (re)load an indicator module.
Returns:
dict: module functions mapping
{
"func1_name": func1,
"func2_name": func2,...
}
"""
# load module
try:
module = importlib.import_module(name)
except Exception as ex:
print(f"[X] An error occurred when attempting to load module {name}: {ex}")
sys.exit(1)
# reload to refresh previously loaded module
module = importlib.reload(module)
return get_module_functions(module)
pandas_ta/momentum/apo.py
+12 -7
View File
@@ -1,10 +1,10 @@
# -*- coding: utf-8 -*-
from pandas_ta import Imports
from pandas_ta.overlap import sma
from pandas_ta.utils import get_offset, verify_series
from pandas_ta.overlap import ma
from pandas_ta.utils import get_offset, tal_ma, verify_series
def apo(close, fast=None, slow=None, offset=None, **kwargs):
def apo(close, fast=None, slow=None, mamode=None, talib=None, offset=None, **kwargs):
"""Indicator: Absolute Price Oscillator (APO)"""
# Validate Arguments
fast = int(fast) if fast and fast > 0 else 12
@@ -12,17 +12,19 @@ def apo(close, fast=None, slow=None, offset=None, **kwargs):
if slow < fast:
fast, slow = slow, fast
close = verify_series(close, max(fast, slow))
mamode = mamode if isinstance(mamode, str) else "sma"
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import APO
apo = APO(close, fast, slow)
apo = APO(close, fast, slow, tal_ma(mamode))
else:
fastma = sma(close, length=fast)
slowma = sma(close, length=slow)
fastma = ma(mamode, close, length=fast)
slowma = ma(mamode, close, length=slow)
apo = fastma - slowma
# Offset
@@ -62,6 +64,9 @@ Args:
close (pd.Series): Series of 'close's
fast (int): The short period. Default: 12
slow (int): The long period. Default: 26
mamode (str): See ```help(ta.ma)```. Default: 'sma'
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/momentum/bias.py
+1 -1
View File
@@ -53,7 +53,7 @@ Calculation:
Args:
close (pd.Series): Series of 'close's
length (int): The period. Default: 26
mamode (str): Options: 'ema', 'hma', 'rma', 'sma', 'wma'. Default: 'sma'
mamode (str): See ```help(ta.ma)```. Default: 'sma'
drift (int): The short period. Default: 1
offset (int): How many periods to offset the result. Default: 0
pandas_ta/momentum/bop.py
+5 -2
View File
@@ -3,7 +3,7 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, non_zero_range, verify_series
def bop(open_, high, low, close, scalar=None, offset=None, **kwargs):
def bop(open_, high, low, close, scalar=None, talib=None, offset=None, **kwargs):
"""Indicator: Balance of Power (BOP)"""
# Validate Arguments
open_ = verify_series(open_)
@@ -12,9 +12,10 @@ def bop(open_, high, low, close, scalar=None, offset=None, **kwargs):
close = verify_series(close)
scalar = float(scalar) if scalar else 1
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import BOP
bop = BOP(open_, high, low, close)
else:
@@ -56,6 +57,8 @@ Args:
low (pd.Series): Series of 'low's
close (pd.Series): Series of 'close's
scalar (float): How much to magnify. Default: 1
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/momentum/cci.py
+5 -2
View File
@@ -5,7 +5,7 @@ from pandas_ta.statistics.mad import mad
from pandas_ta.utils import get_offset, verify_series
def cci(high, low, close, length=None, c=None, offset=None, **kwargs):
def cci(high, low, close, length=None, c=None, talib=None, offset=None, **kwargs):
"""Indicator: Commodity Channel Index (CCI)"""
# Validate Arguments
length = int(length) if length and length > 0 else 14
@@ -14,11 +14,12 @@ def cci(high, low, close, length=None, c=None, offset=None, **kwargs):
low = verify_series(low, length)
close = verify_series(close, length)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if high is None or low is None or close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import CCI
cci = CCI(high, low, close, length)
else:
@@ -71,6 +72,8 @@ Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 14
c (float): Scaling Constant. Default: 0.015
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/momentum/cmo.py
+7 -4
View File
@@ -4,7 +4,7 @@ from pandas_ta.overlap import rma
from pandas_ta.utils import get_drift, get_offset, verify_series
def cmo(close, length=None, scalar=None, drift=None, offset=None, **kwargs):
def cmo(close, length=None, scalar=None, talib=None, drift=None, offset=None, **kwargs):
"""Indicator: Chande Momentum Oscillator (CMO)"""
# Validate Arguments
length = int(length) if length and length > 0 else 14
@@ -12,11 +12,12 @@ def cmo(close, length=None, scalar=None, drift=None, offset=None, **kwargs):
close = verify_series(close, length)
drift = get_drift(drift)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import CMO
cmo = CMO(close, length)
else:
@@ -24,8 +25,7 @@ def cmo(close, length=None, scalar=None, drift=None, offset=None, **kwargs):
positive = mom.copy().clip(lower=0)
negative = mom.copy().clip(upper=0).abs()
talib = kwargs.pop("talib", True)
if talib:
if mode_tal:
pos_ = rma(positive, length)
neg_ = rma(negative, length)
else:
@@ -71,6 +71,9 @@ Calculation:
Args:
close (pd.Series): Series of 'close's
scalar (float): How much to magnify. Default: 100
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. If TA Lib is not installed but talib is True, it runs the Python
version TA Lib. Default: True
drift (int): The short period. Default: 1
offset (int): How many periods to offset the result. Default: 0
pandas_ta/momentum/dm.py
+8 -3
View File
@@ -5,7 +5,7 @@ from pandas_ta.overlap import ma
from pandas_ta.utils import get_offset, verify_series, get_drift, zero
def dm(high, low, length=None, mamode=None, drift=None, offset=None, **kwargs):
def dm(high, low, length=None, mamode=None, talib=None, drift=None, offset=None, **kwargs):
"""Indicator: DM"""
# Validate Arguments
length = int(length) if length and length > 0 else 14
@@ -14,13 +14,15 @@ def dm(high, low, length=None, mamode=None, drift=None, offset=None, **kwargs):
low = verify_series(low)
drift = get_drift(drift)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if high is None or low is None:
return
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import MINUS_DM, PLUS_DM
pos, neg = PLUS_DM(high, low), MINUS_DM(high, low)
pos = PLUS_DM(high, low, length)
neg = MINUS_DM(high, low, length)
else:
up = high - high.shift(drift)
dn = low.shift(drift) - low
@@ -85,6 +87,9 @@ Calculation:
Args:
high (pd.Series): Series of 'high's
low (pd.Series): Series of 'low's
mamode (str): See ```help(ta.ma)```. Default: 'rma'
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
pandas_ta/momentum/fisher.py
+1 -1
View File
@@ -3,7 +3,7 @@ from numpy import log as nplog
from numpy import nan as npNaN
from pandas import DataFrame, Series
from pandas_ta.overlap import hl2
from pandas_ta.utils import get_offset, high_low_range, verify_series, zero
from pandas_ta.utils import get_offset, high_low_range, verify_series
def fisher(high, low, length=None, signal=None, offset=None, **kwargs):
pandas_ta/momentum/inertia.py
+3
View File
@@ -78,6 +78,9 @@ Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 20
rvi_length (int): RVI period. Default: 14
refined (bool): Use 'refined' calculation. Default: False
thirds (bool): Use 'thirds' calculation. Default: False
mamode (str): See ```help(ta.ma)```. Default: 'ema'
drift (int): The difference period. Default: 1
offset (int): How many periods to offset the result. Default: 0
pandas_ta/momentum/macd.py
+26 -7
View File
@@ -5,7 +5,7 @@ from pandas_ta.overlap import ema
from pandas_ta.utils import get_offset, verify_series, signals
def macd(close, fast=None, slow=None, signal=None, offset=None, **kwargs):
def macd(close, fast=None, slow=None, signal=None, talib=None, offset=None, **kwargs):
"""Indicator: Moving Average, Convergence/Divergence (MACD)"""
# Validate arguments
fast = int(fast) if fast and fast > 0 else 12
@@ -15,13 +15,16 @@ def macd(close, fast=None, slow=None, signal=None, offset=None, **kwargs):
fast, slow = slow, fast
close = verify_series(close, max(fast, slow, signal))
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
as_mode = kwargs.setdefault("asmode", False)
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import MACD
macd, signalma, histogram = MACD(close, fast, slow)
macd, signalma, histogram = MACD(close, fast, slow, signal)
else:
fastma = ema(close, length=fast)
slowma = ema(close, length=slow)
@@ -30,6 +33,11 @@ def macd(close, fast=None, slow=None, signal=None, offset=None, **kwargs):
signalma = ema(close=macd.loc[macd.first_valid_index():,], length=signal)
histogram = macd - signalma
if as_mode:
macd = macd - signalma
signalma = ema(close=macd.loc[macd.first_valid_index():,], length=signal)
histogram = macd - signalma
# Offset
if offset != 0:
macd = macd.shift(offset)
@@ -47,16 +55,17 @@ def macd(close, fast=None, slow=None, signal=None, offset=None, **kwargs):
signalma.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
_asmode = "AS" if as_mode else ""
_props = f"_{fast}_{slow}_{signal}"
macd.name = f"MACD{_props}"
histogram.name = f"MACDh{_props}"
signalma.name = f"MACDs{_props}"
macd.name = f"MACD{_asmode}{_props}"
histogram.name = f"MACD{_asmode}h{_props}"
signalma.name = f"MACD{_asmode}s{_props}"
macd.category = histogram.category = signalma.category = "momentum"
# Prepare DataFrame to return
data = {macd.name: macd, histogram.name: histogram, signalma.name: signalma}
df = DataFrame(data)
df.name = f"MACD{_props}"
df.name = f"MACD{_asmode}{_props}"
df.category = macd.category
signal_indicators = kwargs.pop("signal_indicators", False)
@@ -105,6 +114,7 @@ the difference of MACD and Signal.
Sources:
https://www.tradingview.com/wiki/MACD_(Moving_Average_Convergence/Divergence)
AS Mode: https://tr.tradingview.com/script/YFlKXHnP/
Calculation:
Default Inputs:
@@ -114,14 +124,23 @@ Calculation:
Signal = EMA(MACD, signal)
Histogram = MACD - Signal
if asmode:
MACD = MACD - Signal
Signal = EMA(MACD, signal)
Histogram = MACD - Signal
Args:
close (pd.Series): Series of 'close's
fast (int): The short period. Default: 12
slow (int): The long period. Default: 26
signal (int): The signal period. Default: 9
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
asmode (value, optional): When True, enables AS version of MACD.
Default: False
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
pandas_ta/momentum/mom.py
+5 -2
View File
@@ -3,17 +3,18 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def mom(close, length=None, offset=None, **kwargs):
def mom(close, length=None, talib=None, offset=None, **kwargs):
"""Indicator: Momentum (MOM)"""
# Validate Arguments
length = int(length) if length and length > 0 else 10
close = verify_series(close, length)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import MOM
mom = MOM(close, length)
else:
@@ -53,6 +54,8 @@ Calculation:
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 1
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/momentum/ppo.py
+8 -5
View File
@@ -2,10 +2,10 @@
from pandas import DataFrame
from pandas_ta import Imports
from pandas_ta.overlap import ma
from pandas_ta.utils import get_offset, verify_series
from pandas_ta.utils import get_offset, tal_ma, verify_series
def ppo(close, fast=None, slow=None, signal=None, scalar=None, mamode=None, offset=None, **kwargs):
def ppo(close, fast=None, slow=None, signal=None, scalar=None, mamode=None, talib=None, offset=None, **kwargs):
"""Indicator: Percentage Price Oscillator (PPO)"""
# Validate Arguments
fast = int(fast) if fast and fast > 0 else 12
@@ -17,13 +17,14 @@ def ppo(close, fast=None, slow=None, signal=None, scalar=None, mamode=None, offs
fast, slow = slow, fast
close = verify_series(close, max(fast, slow, signal))
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import PPO
ppo = PPO(close, fast, slow)
ppo = PPO(close, fast, slow, tal_ma(mamode))
else:
fastma = ma(mamode, close, length=fast)
slowma = ma(mamode, close, length=slow)
@@ -90,7 +91,9 @@ Args:
slow(int): The long period. Default: 26
signal(int): The signal period. Default: 9
scalar (float): How much to magnify. Default: 100
mamode (str): Options: 'ema', 'hma', 'rma', 'sma', 'wma'. Default: 'sma'
mamode (str): See ```help(ta.ma)```. Default: 'sma'
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset(int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/momentum/qqe.py
+1 -2
View File
@@ -143,8 +143,7 @@ Args:
length (int): RSI period. Default: 14
smooth (int): RSI smoothing period. Default: 5
factor (float): QQE Factor. Default: 4.236
mamode (str): Smoothing MA type: "ema", "hma", "rma", "sma" or "wma".
Default: "ema"
mamode (str): See ```help(ta.ma)```. Default: 'sma'
drift (int): The difference period. Default: 1
offset (int): How many periods to offset the result. Default: 0
pandas_ta/momentum/roc.py
+6 -2
View File
@@ -4,18 +4,19 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def roc(close, length=None, scalar=None, offset=None, **kwargs):
def roc(close, length=None, scalar=None, talib=None, offset=None, **kwargs):
"""Indicator: Rate of Change (ROC)"""
# Validate Arguments
length = int(length) if length and length > 0 else 10
scalar = float(scalar) if scalar and scalar > 0 else 100
close = verify_series(close, length)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import ROC
roc = ROC(close, length)
else:
@@ -57,6 +58,9 @@ Calculation:
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 1
scalar (float): How much to magnify. Default: 100
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/momentum/rsi.py
+5 -2
View File
@@ -5,7 +5,7 @@ from pandas_ta.overlap import rma
from pandas_ta.utils import get_drift, get_offset, verify_series, signals
def rsi(close, length=None, scalar=None, drift=None, offset=None, **kwargs):
def rsi(close, length=None, scalar=None, talib=None, drift=None, offset=None, **kwargs):
"""Indicator: Relative Strength Index (RSI)"""
# Validate arguments
length = int(length) if length and length > 0 else 14
@@ -13,11 +13,12 @@ def rsi(close, length=None, scalar=None, drift=None, offset=None, **kwargs):
close = verify_series(close, length)
drift = get_drift(drift)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import RSI
rsi = RSI(close, length)
else:
@@ -99,6 +100,8 @@ Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 14
scalar (float): How much to magnify. Default: 100
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
pandas_ta/momentum/smi.py
+3 -2
View File
@@ -20,8 +20,9 @@ def smi(close, fast=None, slow=None, signal=None, scalar=None, offset=None, **kw
if close is None: return
# Calculate Result
smi = tsi(close, fast=fast, slow=slow, scalar=scalar)
signalma = ema(smi, signal)
tsi_df = tsi(close, fast=fast, slow=slow, signal=signal, scalar=scalar)
smi = tsi_df.iloc[:, 0]
signalma = tsi_df.iloc[:, 1]
osc = smi - signalma
# Offset
pandas_ta/momentum/squeeze.py
+2 -2
View File
@@ -9,7 +9,7 @@ from pandas_ta.utils import get_offset
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):
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, mamode=None, offset=None, **kwargs):
"""Indicator: Squeeze Momentum (SQZ)"""
# Validate arguments
bb_length = int(bb_length) if bb_length and bb_length > 0 else 20
@@ -30,7 +30,7 @@ def squeeze(high, low, close, bb_length=None, bb_std=None, kc_length=None, kc_sc
asint = kwargs.pop("asint", True)
detailed = kwargs.pop("detailed", False)
lazybear = kwargs.pop("lazybear", False)
mamode = kwargs.pop("mamode", "sma").lower()
mamode = mamode if isinstance(mamode, str) else "sma"
def simplify_columns(df, n=3):
df.columns = df.columns.str.lower()
pandas_ta/momentum/squeeze_pro.py
+2 -2
View File
@@ -9,7 +9,7 @@ from pandas_ta.utils import get_offset
from pandas_ta.utils import unsigned_differences, verify_series
def squeeze_pro(high, low, close, bb_length=None, bb_std=None, kc_length=None, kc_scalar_wide=None, kc_scalar_normal=None, kc_scalar_narrow=None, mom_length=None, mom_smooth=None, use_tr=None, offset=None, **kwargs):
def squeeze_pro(high, low, close, bb_length=None, bb_std=None, kc_length=None, kc_scalar_wide=None, kc_scalar_normal=None, kc_scalar_narrow=None, mom_length=None, mom_smooth=None, use_tr=None, mamode=None, offset=None, **kwargs):
"""Indicator: Squeeze Momentum (SQZ) PRO"""
# Validate arguments
bb_length = int(bb_length) if bb_length and bb_length > 0 else 20
@@ -35,7 +35,7 @@ def squeeze_pro(high, low, close, bb_length=None, bb_std=None, kc_length=None, k
use_tr = kwargs.setdefault("tr", True)
asint = kwargs.pop("asint", True)
detailed = kwargs.pop("detailed", False)
mamode = kwargs.pop("mamode", "sma").lower()
mamode = mamode if isinstance(mamode, str) else "sma"
def simplify_columns(df, n=3):
df.columns = df.columns.str.lower()
pandas_ta/momentum/stoch.py
+6 -5
View File
@@ -1,10 +1,10 @@
# -*- coding: utf-8 -*-
from pandas import DataFrame
from pandas_ta.overlap import sma
from pandas_ta.overlap import ma
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):
def stoch(high, low, close, k=None, d=None, smooth_k=None, mamode=None, offset=None, **kwargs):
"""Indicator: Stochastic Oscillator (STOCH)"""
# Validate arguments
k = k if k and k > 0 else 14
@@ -15,6 +15,7 @@ def stoch(high, low, close, k=None, d=None, smooth_k=None, offset=None, **kwargs
low = verify_series(low, _length)
close = verify_series(close, _length)
offset = get_offset(offset)
mamode = mamode if isinstance(mamode, str) else "sma"
if high is None or low is None or close is None: return
@@ -25,8 +26,8 @@ def stoch(high, low, close, k=None, d=None, smooth_k=None, offset=None, **kwargs
stoch = 100 * (close - lowest_low)
stoch /= non_zero_range(highest_high, lowest_low)
stoch_k = sma(stoch, length=smooth_k)
stoch_d = sma(stoch_k, length=d)
stoch_k = ma(mamode, stoch.loc[stoch.first_valid_index():,], length=smooth_k)
stoch_d = ma(mamode, stoch_k.loc[stoch_k.first_valid_index():,], length=d)
# Offset
if offset != 0:
@@ -53,7 +54,6 @@ def stoch(high, low, close, k=None, d=None, smooth_k=None, offset=None, **kwargs
df = DataFrame(data)
df.name = f"{_name}{_props}"
df.category = stoch_k.category
return df
@@ -91,6 +91,7 @@ Args:
k (int): The Fast %K period. Default: 14
d (int): The Slow %K period. Default: 3
smooth_k (int): The Slow %D period. Default: 3
mamode (str): See ```help(ta.ma)```. Default: 'sma'
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/momentum/stochrsi.py
+6 -4
View File
@@ -1,11 +1,11 @@
# -*- coding: utf-8 -*-
from pandas import DataFrame
from .rsi import rsi
from pandas_ta.overlap import sma
from pandas_ta.overlap import ma
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):
def stochrsi(close, length=None, rsi_length=None, k=None, d=None, mamode=None, offset=None, **kwargs):
"""Indicator: Stochastic RSI Oscillator (STOCHRSI)"""
# Validate arguments
length = length if length and length > 0 else 14
@@ -14,6 +14,7 @@ def stochrsi(close, length=None, rsi_length=None, k=None, d=None, offset=None, *
d = d if d and d > 0 else 3
close = verify_series(close, max(length, rsi_length, k, d))
offset = get_offset(offset)
mamode = mamode if isinstance(mamode, str) else "sma"
if close is None: return
@@ -25,8 +26,8 @@ def stochrsi(close, length=None, rsi_length=None, k=None, d=None, offset=None, *
stoch = 100 * (rsi_ - lowest_rsi)
stoch /= non_zero_range(highest_rsi, lowest_rsi)
stochrsi_k = sma(stoch, length=k)
stochrsi_d = sma(stochrsi_k, length=d)
stochrsi_k = ma(mamode, stoch, length=k)
stochrsi_d = ma(mamode, stochrsi_k, length=d)
# Offset
if offset != 0:
@@ -92,6 +93,7 @@ Args:
rsi_length (int): RSI period. Default: 14
k (int): The Fast %K period. Default: 3
d (int): The Slow %K period. Default: 3
mamode (str): See ```help(ta.ma)```. Default: 'sma'
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/momentum/tsi.py
+24 -7
View File
@@ -1,19 +1,22 @@
# -*- coding: utf-8 -*-
from pandas_ta.overlap import ema
from pandas import DataFrame
from pandas_ta.overlap import ema, ma
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):
def tsi(close, fast=None, slow=None, signal=None, scalar=None, mamode=None, drift=None, offset=None, **kwargs):
"""Indicator: True Strength Index (TSI)"""
# Validate Arguments
fast = int(fast) if fast and fast > 0 else 13
slow = int(slow) if slow and slow > 0 else 25
signal = int(signal) if signal and signal > 0 else 13
# 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)
mamode = mamode if isinstance(mamode, str) else "ema"
if "length" in kwargs: kwargs.pop("length")
if close is None: return
@@ -28,22 +31,32 @@ def tsi(close, fast=None, slow=None, scalar=None, drift=None, offset=None, **kwa
abs_fast_slow_ema = ema(close=abs_slow_ema, length=fast, **kwargs)
tsi = scalar * fast_slow_ema / abs_fast_slow_ema
tsi_signal = ma(mamode, tsi, length=signal)
# Offset
if offset != 0:
tsi = tsi.shift(offset)
tsi_signal = tsi_signal.shift(offset)
# Handle fills
if "fillna" in kwargs:
tsi.fillna(kwargs["fillna"], inplace=True)
tsi_signal.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
tsi.fillna(method=kwargs["fill_method"], inplace=True)
tsi_signal.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
tsi.name = f"TSI_{fast}_{slow}"
tsi.category = "momentum"
tsi.name = f"TSI_{fast}_{slow}_{signal}"
tsi_signal.name = f"TSIs_{fast}_{slow}_{signal}"
tsi.category = tsi_signal.category = "momentum"
return tsi
# Prepare DataFrame to return
df = DataFrame({tsi.name: tsi, tsi_signal.name: tsi_signal})
df.name = f"TSI_{fast}_{slow}_{signal}"
df.category = "momentum"
return df
tsi.__doc__ = \
@@ -58,7 +71,7 @@ Sources:
Calculation:
Default Inputs:
fast=13, slow=25, scalar=100, drift=1
fast=13, slow=25, signal=13, scalar=100, drift=1
EMA = Exponential Moving Average
diff = close.diff(drift)
@@ -69,12 +82,16 @@ Calculation:
abema = abs_diff_fast_slow_ema = EMA(abs_diff_slow_ema, fast)
TSI = scalar * fast_slow_ema / abema
Signal = EMA(TSI, signal)
Args:
close (pd.Series): Series of 'close's
fast (int): The short period. Default: 13
slow (int): The long period. Default: 25
signal (int): The signal period. Default: 13
scalar (float): How much to magnify. Default: 100
mamode (str): Moving Average of TSI Signal Line.
See ```help(ta.ma)```. Default: 'ema'
drift (int): The difference period. Default: 1
offset (int): How many periods to offset the result. Default: 0
@@ -83,5 +100,5 @@ Kwargs:
fill_method (value, optional): Type of fill method
Returns:
pd.Series: New feature generated.
pd.DataFrame: tsi, signal.
"""
pandas_ta/momentum/uo.py
+6 -3
View File
@@ -4,7 +4,7 @@ from pandas_ta import Imports
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):
def uo(high, low, close, fast=None, medium=None, slow=None, fast_w=None, medium_w=None, slow_w=None, talib=None, drift=None, offset=None, **kwargs):
"""Indicator: Ultimate Oscillator (UO)"""
# Validate arguments
fast = int(fast) if fast and fast > 0 else 7
@@ -19,13 +19,14 @@ def uo(high, low, close, fast=None, medium=None, slow=None, fast_w=None, medium_
close = verify_series(close, _length)
drift = get_drift(drift)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if high is None or low is None or close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import ULTOSC
uo = ULTOSC(high, low, close)
uo = ULTOSC(high, low, close, fast, medium, slow)
else:
tdf = DataFrame({
"high": high,
@@ -101,6 +102,8 @@ Args:
fast_w (float): The Fast %K period. Default: 4.0
medium_w (float): The Slow %K period. Default: 2.0
slow_w (float): The Slow %D period. Default: 1.0
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
pandas_ta/momentum/willr.py
+5 -2
View File
@@ -3,7 +3,7 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def willr(high, low, close, length=None, offset=None, **kwargs):
def willr(high, low, close, length=None, talib=None, offset=None, **kwargs):
"""Indicator: William's Percent R (WILLR)"""
# Validate arguments
length = int(length) if length and length > 0 else 14
@@ -13,11 +13,12 @@ def willr(high, low, close, length=None, offset=None, **kwargs):
low = verify_series(low, _length)
close = verify_series(close, _length)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if high is None or low is None or close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import WILLR
willr = WILLR(high, low, close, length)
else:
@@ -65,6 +66,8 @@ Args:
low (pd.Series): Series of 'low's
close (pd.Series): Series of 'close's
length (int): It's period. Default: 14
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/overlap/__init__.py
+1
View File
@@ -9,6 +9,7 @@ from .hlc3 import hlc3
from .hma import hma
from .hwma import hwma
from .ichimoku import ichimoku
from .jma import jma
from .kama import kama
from .linreg import linreg
from .ma import ma
pandas_ta/overlap/dema.py
+5 -2
View File
@@ -4,17 +4,18 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def dema(close, length=None, offset=None, **kwargs):
def dema(close, length=None, talib=None, offset=None, **kwargs):
"""Indicator: Double Exponential Moving Average (DEMA)"""
# Validate Arguments
length = int(length) if length and length > 0 else 10
close = verify_series(close, length)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import DEMA
dema = DEMA(close, length)
else:
@@ -60,6 +61,8 @@ Calculation:
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 10
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/overlap/ema.py
+5 -2
View File
@@ -4,7 +4,7 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def ema(close, length=None, offset=None, **kwargs):
def ema(close, length=None, talib=None, offset=None, **kwargs):
"""Indicator: Exponential Moving Average (EMA)"""
# Validate Arguments
length = int(length) if length and length > 0 else 10
@@ -12,11 +12,12 @@ def ema(close, length=None, offset=None, **kwargs):
sma = kwargs.pop("sma", True)
close = verify_series(close, length)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import EMA
ema = EMA(close, length)
else:
@@ -69,6 +70,8 @@ Calculation:
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 10
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/overlap/hilo.py
+1 -1
View File
@@ -115,7 +115,7 @@ Args:
close (pd.Series): Series of 'close's
high_length (int): It's period. Default: 13
low_length (int): It's period. Default: 21
mamode (str): Options: 'sma' or 'ema'. Default: 'sma'
mamode (str): See ```help(ta.ma)```. Default: 'sma'
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/overlap/hlc3.py
+3 -2
View File
@@ -3,16 +3,17 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def hlc3(high, low, close, offset=None, **kwargs):
def hlc3(high, low, close, talib=None, offset=None, **kwargs):
"""Indicator: HLC3"""
# Validate Arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import TYPPRICE
hlc3 = TYPPRICE(high, low, close)
else:
pandas_ta/overlap/ichimoku.py
+7 -2
View File
@@ -4,7 +4,7 @@ from .midprice import midprice
from pandas_ta.utils import get_offset, verify_series
def ichimoku(high, low, close, tenkan=None, kijun=None, senkou=None, offset=None, **kwargs):
def ichimoku(high, low, close, tenkan=None, kijun=None, senkou=None, include_chikou=True, offset=None, **kwargs):
"""Indicator: Ichimoku Kinkō Hyō (Ichimoku)"""
tenkan = int(tenkan) if tenkan and tenkan > 0 else 9
kijun = int(kijun) if kijun and kijun > 0 else 26
@@ -14,6 +14,8 @@ def ichimoku(high, low, close, tenkan=None, kijun=None, senkou=None, offset=None
low = verify_series(low, _length)
close = verify_series(close, _length)
offset = get_offset(offset)
if not kwargs.get("lookahead", True):
include_chikou = False
if high is None or low is None or close is None: return None, None
@@ -65,8 +67,10 @@ def ichimoku(high, low, close, tenkan=None, kijun=None, senkou=None, offset=None
span_b.name: span_b,
tenkan_sen.name: tenkan_sen,
kijun_sen.name: kijun_sen,
chikou_span.name: chikou_span,
}
if include_chikou:
data[chikou_span.name] = chikou_span
ichimokudf = DataFrame(data)
ichimokudf.name = f"ICHIMOKU_{tenkan}_{kijun}_{senkou}"
ichimokudf.category = "overlap"
@@ -121,6 +125,7 @@ Args:
tenkan (int): Tenkan period. Default: 9
kijun (int): Kijun period. Default: 26
senkou (int): Senkou period. Default: 52
include_chikou (bool): Whether to include chikou component. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/overlap/jma.py
+123
View File
@@ -0,0 +1,123 @@
# -*- coding: utf-8 -*-
from numpy import average as npAverage
from numpy import nan as npNaN
from numpy import log as npLog
from numpy import power as npPower
from numpy import sqrt as npSqrt
from numpy import zeros_like as npZeroslike
from pandas import Series
from pandas_ta.utils import get_offset, verify_series
def jma(close, length=None, phase=None, offset=None, **kwargs):
"""Indicator: Jurik Moving Average (JMA)"""
# Validate Arguments
_length = int(length) if length and length > 0 else 7
phase = float(phase) if phase and phase != 0 else 0
close = verify_series(close, _length)
offset = get_offset(offset)
if close is None: return
# Define base variables
jma = npZeroslike(close)
volty = npZeroslike(close)
v_sum = npZeroslike(close)
kv = det0 = det1 = ma2 = 0.0
jma[0] = ma1 = uBand = lBand = close[0]
# Static variables
sum_length = 10
length = 0.5 * (_length - 1)
pr = 0.5 if phase < -100 else 2.5 if phase > 100 else 1.5 + phase * 0.01
length1 = max((npLog(npSqrt(length)) / npLog(2.0)) + 2.0, 0)
pow1 = max(length1 - 2.0, 0.5)
length2 = length1 * npSqrt(length)
bet = length2 / (length2 + 1)
beta = 0.45 * (_length - 1) / (0.45 * (_length - 1) + 2.0)
m = close.shape[0]
for i in range(1, m):
price = close[i]
# Price volatility
del1 = price - uBand
del2 = price - lBand
volty[i] = max(abs(del1),abs(del2)) if abs(del1)!=abs(del2) else 0
# Relative price volatility factor
v_sum[i] = v_sum[i - 1] + (volty[i] - volty[max(i - sum_length, 0)]) / sum_length
avg_volty = npAverage(v_sum[max(i - 65, 0):i + 1])
d_volty = 0 if avg_volty ==0 else volty[i] / avg_volty
r_volty = max(1.0, min(npPower(length1, 1 / pow1), d_volty))
# Jurik volatility bands
pow2 = npPower(r_volty, pow1)
kv = npPower(bet, npSqrt(pow2))
uBand = price if (del1 > 0) else price - (kv * del1)
lBand = price if (del2 < 0) else price - (kv * del2)
# Jurik Dynamic Factor
power = npPower(r_volty, pow1)
alpha = npPower(beta, power)
# 1st stage - prelimimary smoothing by adaptive EMA
ma1 = ((1 - alpha) * price) + (alpha * ma1)
# 2nd stage - one more prelimimary smoothing by Kalman filter
det0 = ((price - ma1) * (1 - beta)) + (beta * det0)
ma2 = ma1 + pr * det0
# 3rd stage - final smoothing by unique Jurik adaptive filter
det1 = ((ma2 - jma[i - 1]) * (1 - alpha) * (1 - alpha)) + (alpha * alpha * det1)
jma[i] = jma[i-1] + det1
# Remove initial lookback data and convert to pandas frame
jma[0:_length - 1] = npNaN
jma = Series(jma, index=close.index)
# Offset
if offset != 0:
jma = jma.shift(offset)
# Handle fills
if "fillna" in kwargs:
jma.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
jma.fillna(method=kwargs["fill_method"], inplace=True)
# Name & Category
jma.name = f"JMA_{_length}_{phase}"
jma.category = "overlap"
return jma
jma.__doc__ = \
"""Jurik Moving Average Average (JMA)
Mark Jurik's Moving Average (JMA) attempts to eliminate noise to see the "true"
underlying activity. It has extremely low lag, is very smooth and is responsive
to market gaps.
Sources:
https://c.mql5.com/forextsd/forum/164/jurik_1.pdf
https://www.prorealcode.com/prorealtime-indicators/jurik-volatility-bands/
Calculation:
Default Inputs:
length=7, phase=0
Args:
close (pd.Series): Series of 'close's
length (int): Period of calculation. Default: 7
phase (float): How heavy/light the average is [-100, 100]. Default: 0
offset (int): How many lengths to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.Series: New feature generated.
"""
pandas_ta/overlap/linreg.py
+14 -2
View File
@@ -3,7 +3,7 @@ from numpy import array as npArray
from numpy import arctan as npAtan
from numpy import nan as npNaN
from numpy import pi as npPi
from numpy.lib.stride_tricks import sliding_window_view
from numpy.version import version as npVersion
from pandas import Series
from pandas_ta.utils import get_offset, verify_series
@@ -54,7 +54,19 @@ def linreg(close, length=None, offset=None, **kwargs):
return m * length + b if tsf else m * (length - 1) + b
linreg_ = [linear_regression(_) for _ in sliding_window_view(npArray(close), length)]
def rolling_window(array, length):
"""https://github.com/twopirllc/pandas-ta/issues/285"""
strides = array.strides + (array.strides[-1],)
shape = array.shape[:-1] + (array.shape[-1] - length + 1, length)
return as_strided(array, shape=shape, strides=strides)
if npVersion >= "1.20.0":
from numpy.lib.stride_tricks import sliding_window_view
linreg_ = [linear_regression(_) for _ in sliding_window_view(npArray(close), length)]
else:
from numpy.lib.stride_tricks import as_strided
linreg_ = [linear_regression(_) for _ in rolling_window(npArray(close), length)]
linreg = Series([npNaN] * (length - 1) + linreg_, index=close.index)
# Offset
pandas_ta/overlap/midpoint.py
+3 -2
View File
@@ -3,18 +3,19 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def midpoint(close, length=None, offset=None, **kwargs):
def midpoint(close, length=None, talib=None, offset=None, **kwargs):
"""Indicator: Midpoint"""
# Validate arguments
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)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import MIDPOINT
midpoint = MIDPOINT(close, length)
else:
pandas_ta/overlap/midprice.py
+3 -2
View File
@@ -3,7 +3,7 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def midprice(high, low, length=None, offset=None, **kwargs):
def midprice(high, low, length=None, talib=None, offset=None, **kwargs):
"""Indicator: Midprice"""
# Validate arguments
length = int(length) if length and length > 0 else 2
@@ -12,11 +12,12 @@ def midprice(high, low, length=None, offset=None, **kwargs):
high = verify_series(high, _length)
low = verify_series(low, _length)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if high is None or low is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import MIDPRICE
midprice = MIDPRICE(high, low, length)
else:
pandas_ta/overlap/sma.py
+5 -2
View File
@@ -3,18 +3,19 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def sma(close, length=None, offset=None, **kwargs):
def sma(close, length=None, talib=None, offset=None, **kwargs):
"""Indicator: Simple Moving Average (SMA)"""
# Validate Arguments
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)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import SMA
sma = SMA(close, length)
else:
@@ -54,6 +55,8 @@ Calculation:
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 10
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/overlap/t3.py
+6 -3
View File
@@ -4,20 +4,21 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def t3(close, length=None, a=None, offset=None, **kwargs):
def t3(close, length=None, a=None, talib=None, offset=None, **kwargs):
"""Indicator: T3"""
# Validate Arguments
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)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import T3
t3 = T3(close, length)
t3 = T3(close, length, a)
else:
c1 = -a * a**2
c2 = 3 * a**2 + 3 * a**3
@@ -77,6 +78,8 @@ Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 10
a (float): 0 < a < 1. Default: 0.7
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/overlap/tema.py
+5 -2
View File
@@ -4,17 +4,18 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def tema(close, length=None, offset=None, **kwargs):
def tema(close, length=None, talib=None, offset=None, **kwargs):
"""Indicator: Triple Exponential Moving Average (TEMA)"""
# Validate Arguments
length = int(length) if length and length > 0 else 10
close = verify_series(close, length)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import TEMA
tema = TEMA(close, length)
else:
@@ -60,6 +61,8 @@ Calculation:
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 10
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/overlap/trima.py
+5 -2
View File
@@ -4,17 +4,18 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def trima(close, length=None, offset=None, **kwargs):
def trima(close, length=None, talib=None, offset=None, **kwargs):
"""Indicator: Triangular Moving Average (TRIMA)"""
# Validate Arguments
length = int(length) if length and length > 0 else 10
close = verify_series(close, length)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import TRIMA
trima = TRIMA(close, length)
else:
@@ -61,6 +62,8 @@ Calculation:
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 10
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/overlap/wcp.py
+5 -2
View File
@@ -3,16 +3,17 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def wcp(high, low, close, offset=None, **kwargs):
def wcp(high, low, close, talib=None, offset=None, **kwargs):
"""Indicator: Weighted Closing Price (WCP)"""
# Validate Arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import WCLPRICE
wcp = WCLPRICE(high, low, close)
else:
@@ -51,6 +52,8 @@ Args:
high (pd.Series): Series of 'high's
low (pd.Series): Series of 'low's
close (pd.Series): Series of 'close's
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/overlap/wma.py
+5 -2
View File
@@ -4,18 +4,19 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def wma(close, length=None, asc=None, offset=None, **kwargs):
def wma(close, length=None, asc=None, talib=None, offset=None, **kwargs):
"""Indicator: Weighted Moving Average (WMA)"""
# Validate Arguments
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)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import WMA
wma = WMA(close, length)
else:
@@ -78,6 +79,8 @@ Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 10
asc (bool): Recent values weigh more. Default: True
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/statistics/stdev.py
+6 -3
View File
@@ -5,18 +5,19 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def stdev(close, length=None, ddof=None, offset=None, **kwargs):
def stdev(close, length=None, ddof=None, talib=None, offset=None, **kwargs):
"""Indicator: Standard Deviation"""
# Validate Arguments
length = int(length) if length and length > 0 else 30
ddof = int(ddof) if ddof and ddof >= 0 and ddof < length else 1
ddof = int(ddof) if isinstance(ddof, int) and ddof >= 0 and ddof < length else 1
close = verify_series(close, length)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import STDDEV
stdev = STDDEV(close, length)
else:
@@ -56,6 +57,8 @@ Args:
ddof (int): Delta Degrees of Freedom.
The divisor used in calculations is N - ddof,
where N represents the number of elements. Default: 1
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/statistics/variance.py
+6 -3
View File
@@ -3,19 +3,20 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def variance(close, length=None, ddof=None, offset=None, **kwargs):
def variance(close, length=None, ddof=None, talib=None, offset=None, **kwargs):
"""Indicator: Variance"""
# Validate Arguments
length = int(length) if length and length > 1 else 30
ddof = int(ddof) if ddof and ddof >= 0 and ddof < length else 0
ddof = int(ddof) if isinstance(ddof, int) and ddof >= 0 and ddof < length else 1
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)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import VAR
variance = VAR(close, length)
else:
@@ -54,6 +55,8 @@ Args:
ddof (int): Delta Degrees of Freedom.
The divisor used in calculations is N - ddof,
where N represents the number of elements. Default: 0
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/trend/adx.py
+2 -1
View File
@@ -5,7 +5,7 @@ from pandas_ta.volatility import atr
from pandas_ta.utils import get_drift, get_offset, verify_series, zero
def adx(high, low, close, length=None, lensig=None, mamode=None, scalar=None, drift=None, offset=None, **kwargs):
def adx(high, low, close, length=None, lensig=None, scalar=None, mamode=None, drift=None, offset=None, **kwargs):
"""Indicator: ADX"""
# Validate Arguments
length = length if length and length > 0 else 14
@@ -138,6 +138,7 @@ Args:
length (int): It's period. Default: 14
lensig (int): Signal Length. Like TradingView's default ADX. Default: length
scalar (float): How much to magnify. Default: 100
mamode (str): See ```help(ta.ma)```. Default: 'rma'
drift (int): The difference period. Default: 1
offset (int): How many periods to offset the result. Default: 0
pandas_ta/trend/amat.py
+1 -1
View File
@@ -6,7 +6,7 @@ from pandas_ta.overlap import ma
from pandas_ta.utils import get_offset, verify_series
def amat(close=None, fast=None, slow=None, mamode=None, lookback=None, offset=None, **kwargs):
def amat(close=None, fast=None, slow=None, lookback=None, mamode=None, offset=None, **kwargs):
"""Indicator: Archer Moving Averages Trends (AMAT)"""
# Validate Arguments
fast = int(fast) if fast and fast > 0 else 8
pandas_ta/trend/aroon.py
+5 -2
View File
@@ -5,7 +5,7 @@ from pandas_ta.utils import get_offset, verify_series
from pandas_ta.utils import recent_maximum_index, recent_minimum_index
def aroon(high, low, length=None, scalar=None, offset=None, **kwargs):
def aroon(high, low, length=None, scalar=None, talib=None, offset=None, **kwargs):
"""Indicator: Aroon & Aroon Oscillator"""
# Validate Arguments
length = length if length and length > 0 else 14
@@ -13,11 +13,12 @@ def aroon(high, low, length=None, scalar=None, offset=None, **kwargs):
high = verify_series(high, length)
low = verify_series(low, length)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if high is None or low is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import AROON, AROONOSC
aroon_down, aroon_up = AROON(high, low, length)
aroon_osc = AROONOSC(high, low, length)
@@ -94,6 +95,8 @@ Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 14
scalar (float): How much to magnify. Default: 100
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/trend/cksp.py
+3 -3
View File
@@ -4,7 +4,7 @@ from pandas_ta.volatility import atr
from pandas_ta.utils import get_offset, verify_series
def cksp(high, low, close, p=None, x=None, q=None, offset=None, tvmode=None, **kwargs):
def cksp(high, low, close, p=None, x=None, q=None, tvmode=None, offset=None, **kwargs):
"""Indicator: Chande Kroll Stop (CKSP)"""
# Validate Arguments
# TV defaults=(10,1,9), book defaults = (10,3,20)
@@ -23,7 +23,7 @@ def cksp(high, low, close, p=None, x=None, q=None, offset=None, tvmode=None, **k
mamode = "rma" if tvmode is True else "sma"
# Calculate Result
atr_ = atr(high=high, low=low, close=close, length=p, mamode = mamode)
atr_ = atr(high=high, low=low, close=close, length=p, mamode=mamode)
long_stop_ = high.rolling(p).max() - x * atr_
long_stop = long_stop_.rolling(q).max()
@@ -90,8 +90,8 @@ Args:
p (int): ATR and first stop period. Default: 10 in both modes
x (float): ATR scalar. Default: 1 in TV mode, 3 otherwise
q (int): Second stop period. Default: 9 in TV mode, 20 otherwise
offset (int): How many periods to offset the result. Default: 0
tvmode (bool): Trading View or book implementation mode. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
pandas_ta/trend/decay.py
+1 -1
View File
@@ -63,7 +63,7 @@ Calculation:
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 1
mamode (str): Option "exponential" ("exp"). Default: 'linear' or None
mode (str): If 'exp' then "exponential" decay. Default: 'linear'
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/trend/dpo.py
+2
View File
@@ -9,6 +9,8 @@ def dpo(close, length=None, centered=True, offset=None, **kwargs):
length = int(length) if length and length > 0 else 20
close = verify_series(close, length)
offset = get_offset(offset)
if not kwargs.get("lookahead", True):
centered = False
if close is None: return
pandas_ta/trend/psar.py
+2 -2
View File
@@ -37,7 +37,7 @@ def psar(high, low, close=None, af0=None, af=None, max_af=None, offset=None, **k
long = Series(npNaN, index=high.index)
short = long.copy()
reversal = Series(False, index=high.index)
reversal = Series(0, index=high.index)
_af = long.copy()
_af.iloc[0:2] = af0
@@ -81,7 +81,7 @@ def psar(high, low, close=None, af0=None, af=None, max_af=None, offset=None, **k
long.iloc[row] = sar
_af.iloc[row] = af
reversal.iloc[row] = reverse
reversal.iloc[row] = int(reverse)
# Offset
if offset != 0:
pandas_ta/utils/_core.py
+18
View File
@@ -6,6 +6,7 @@ from sys import float_info as sflt
from numpy import argmax, argmin
from pandas import DataFrame, Series
from pandas.api.types import is_datetime64_any_dtype
from pandas_ta import Imports
def _camelCase2Title(x: str):
@@ -82,6 +83,23 @@ def signed_series(series: Series, initial: int = None) -> Series:
return sign
def tal_ma(name: str) -> int:
"""Helper Function that returns the Enum value for TA Lib's MA Type"""
if Imports["talib"] and isinstance(name, str) and len(name) > 1:
from talib import MA_Type
name = name.lower()
if name == "sma": return MA_Type.SMA # 0
elif name == "ema": return MA_Type.EMA # 1
elif name == "wma": return MA_Type.WMA # 2
elif name == "dema": return MA_Type.DEMA # 3
elif name == "tema": return MA_Type.TEMA # 4
elif name == "trima": return MA_Type.TRIMA # 5
elif name == "kama": return MA_Type.KAMA # 6
elif name == "mama": return MA_Type.MAMA # 7
elif name == "t3": return MA_Type.T3 # 8
return 0 # Default: SMA -> 0
def unsigned_differences(series: Series, amount: int = None, **kwargs) -> Series:
"""Unsigned Differences
Returns two Series, an unsigned positive and unsigned negative series based
pandas_ta/utils/data/yahoofinance.py
+8 -1
View File
@@ -87,7 +87,14 @@ def yf(ticker: str, **kwargs):
# Ticker Info & Chart History
yfd = yfra.Ticker(ticker)
df = yfd.history(period=period, interval=interval, proxy=proxy, **kwargs)
try:
df = yfd.history(period=period, interval=interval, proxy=proxy, **kwargs)
except:
if yfra.__version__ == "0.1.60":
print(f"[!] If history is not downloading, see yfinance Issue #760 by user djl0.")
print(f"[!] https://github.com/ranaroussi/yfinance/issues/760#issuecomment-877355832")
return
if df.empty: return
df.name = ticker
pandas_ta/volatility/accbands.py
+1 -1
View File
@@ -93,7 +93,7 @@ Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 10
c (int): Multiplier. Default: 4
mamode (str): Two options: None or 'ema'. Default: 'ema'
mamode (str): See ```help(ta.ma)```. Default: 'sma'
drift (int): The difference period. Default: 1
offset (int): How many periods to offset the result. Default: 0
pandas_ta/volatility/atr.py
+7 -4
View File
@@ -5,7 +5,7 @@ from pandas_ta.overlap import ma
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):
def atr(high, low, close, length=None, mamode=None, talib=None, drift=None, offset=None, **kwargs):
"""Indicator: Average True Range (ATR)"""
# Validate arguments
length = int(length) if length and length > 0 else 14
@@ -15,13 +15,14 @@ def atr(high, low, close, length=None, mamode=None, drift=None, offset=None, **k
close = verify_series(close, length)
drift = get_drift(drift)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if high is None or low is None or close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import ATR
atr = ATR(high, low, close)
atr = ATR(high, low, close, length)
else:
tr = true_range(high=high, low=low, close=close, drift=drift)
atr = ma(mamode, tr, length=length)
@@ -83,7 +84,9 @@ Args:
low (pd.Series): Series of 'low's
close (pd.Series): Series of 'close's
length (int): It's period. Default: 14
mamode (str): "sma", "ema", "wma" or "rma". Default: "rma"
mamode (str): See ```help(ta.ma)```. Default: 'rma'
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
pandas_ta/volatility/bbands.py
+8 -5
View File
@@ -3,10 +3,10 @@ from pandas import DataFrame
from pandas_ta import Imports
from pandas_ta.overlap import ma
from pandas_ta.statistics import stdev
from pandas_ta.utils import get_offset, non_zero_range, verify_series
from pandas_ta.utils import get_offset, non_zero_range, tal_ma, verify_series
def bbands(close, length=None, std=None, mamode=None, ddof=0, offset=None, **kwargs):
def bbands(close, length=None, std=None, ddof=0, mamode=None, talib=None, offset=None, **kwargs):
"""Indicator: Bollinger Bands (BBANDS)"""
# Validate arguments
length = int(length) if length and length > 0 else 5
@@ -15,13 +15,14 @@ def bbands(close, length=None, std=None, mamode=None, ddof=0, offset=None, **kwa
ddof = int(ddof) if ddof >= 0 and ddof < length else 1
close = verify_series(close, length)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import BBANDS
upper, mid, lower = BBANDS(close, length)
upper, mid, lower = BBANDS(close, length, std, std, tal_ma(mamode))
else:
standard_deviation = stdev(close=close, length=length, ddof=ddof)
deviations = std * standard_deviation
@@ -108,8 +109,10 @@ Args:
close (pd.Series): Series of 'close's
length (int): The short period. Default: 5
std (int): The long period. Default: 2
mamode (str): Two options: "sma" or "ema". Default: "sma"
ddof (int): Degrees of Freedom to use. Default: 0
mamode (str): See ```help(ta.ma)```. Default: 'sma'
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/volatility/hwc.py
+8 -8
View File
@@ -85,25 +85,25 @@ def hwc(close, na=None, nb=None, nc=None, nd=None, scalar=None, channel_eval=Non
# Name and Categorize it
# suffix = f'{str(na).replace(".", "")}-{str(nb).replace(".", "")}-{str(nc).replace(".", "")}'
hwc.name = 'HW-MID'
hwc_upper.name = "HW-UPPER"
hwc_lower.name = "HW-LOWER"
hwc.name = "HWM"
hwc_upper.name = "HWU"
hwc_lower.name = "HWL"
hwc.category = hwc_upper.category = hwc_lower.category = "volatility"
if channel_eval:
hwc_width.name = 'HW-WIDTH'
hwc_pctwidth.name = 'HW-PCTW'
hwc_width.name = "HWW"
hwc_pctwidth.name = "HWPCT"
# Prepare DataFrame to return
if channel_eval:
data = {hwc.name: hwc, hwc_upper.name: hwc_upper, hwc_lower.name: hwc_lower,
hwc_width.name: hwc_width, hwc_pctwidth.name: hwc_pctwidth}
df = DataFrame(data)
df.name = "hwc"
df.name = "HWC"
df.category = hwc.category
else:
data = {hwc.name: hwc, hwc_upper.name: hwc_upper, hwc_lower.name: hwc_lower}
df = DataFrame(data)
df.name = "hwc"
df.name = "HWC"
df.category = hwc.category
return df
@@ -149,5 +149,5 @@ Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.DataFrame: HW-MID, HW-UPPER, HW-LOWER columns.
pd.DataFrame: HWM (Mid), HWU (Upper), HWL (Lower) columns.
"""
pandas_ta/volatility/kc.py
+1 -1
View File
@@ -100,7 +100,7 @@ Args:
close (pd.Series): Series of 'close's
length (int): The short period. Default: 20
scalar (float): A positive float to scale the bands. Default: 2
mamode (str): Two options: "sma" or "ema". Default: "ema"
mamode (str): See ```help(ta.ma)```. Default: 'ema'
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/volatility/natr.py
+7 -3
View File
@@ -4,7 +4,7 @@ from pandas_ta import Imports
from pandas_ta.utils import get_drift, get_offset, verify_series
def natr(high, low, close, length=None, mamode=None, scalar=None, drift=None, offset=None, **kwargs):
def natr(high, low, close, length=None, scalar=None, mamode=None, talib=None, drift=None, offset=None, **kwargs):
"""Indicator: Normalized Average True Range (NATR)"""
# Validate arguments
length = int(length) if length and length > 0 else 14
@@ -15,13 +15,14 @@ def natr(high, low, close, length=None, mamode=None, scalar=None, drift=None, of
close = verify_series(close, length)
drift = get_drift(drift)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if high is None or low is None or close is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import NATR
natr = NATR(high, low, close)
natr = NATR(high, low, close, length)
else:
natr = scalar / close
natr *= atr(high=high, low=low, close=close, length=length, mamode=mamode, drift=drift, offset=offset, **kwargs)
@@ -63,6 +64,9 @@ Args:
close (pd.Series): Series of 'close's
length (int): The short period. Default: 20
scalar (float): How much to magnify. Default: 100
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
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/volatility/rvi.py
+1 -1
View File
@@ -101,10 +101,10 @@ Args:
close (pd.Series): Series of 'close's
length (int): The short period. Default: 14
scalar (float): A positive float to scale the bands. Default: 100
mamode (str): Options: 'sma' or 'ema'. Default: 'sma'
refined (bool): Use 'refined' calculation which is the average of
RVI(high) and RVI(low) instead of RVI(close). Default: False
thirds (bool): Average of high, low and close. Default: False
mamode (str): See ```help(ta.ma)```. Default: 'ema'
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/volatility/thermo.py
+1 -1
View File
@@ -112,8 +112,8 @@ Args:
long(int): The buy factor
short(float): The sell factor
length (int): The period. Default: 20
mamode (str): See ```help(ta.ma)```. Default: 'ema'
drift (int): The diff period. Default: 1
mamode (str): Three options: "ema", "sma", or "hma". Default: "ema"
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/volatility/true_range.py
+5 -2
View File
@@ -5,7 +5,7 @@ from pandas_ta import Imports
from pandas_ta.utils import get_drift, get_offset, non_zero_range, verify_series
def true_range(high, low, close, drift=None, offset=None, **kwargs):
def true_range(high, low, close, talib=None, drift=None, offset=None, **kwargs):
"""Indicator: True Range"""
# Validate arguments
high = verify_series(high)
@@ -13,9 +13,10 @@ def true_range(high, low, close, drift=None, offset=None, **kwargs):
close = verify_series(close)
drift = get_drift(drift)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import TRANGE
true_range = TRANGE(high, low, close)
else:
@@ -63,6 +64,8 @@ Args:
high (pd.Series): Series of 'high's
low (pd.Series): Series of 'low's
close (pd.Series): Series of 'close's
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
drift (int): The shift period. Default: 1
offset (int): How many periods to offset the result. Default: 0
pandas_ta/volume/ad.py
+5 -2
View File
@@ -3,7 +3,7 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, non_zero_range, verify_series
def ad(high, low, close, volume, open_=None, offset=None, **kwargs):
def ad(high, low, close, volume, open_=None, talib=None, offset=None, **kwargs):
"""Indicator: Accumulation/Distribution (AD)"""
# Validate Arguments
high = verify_series(high)
@@ -11,9 +11,10 @@ def ad(high, low, close, volume, open_=None, offset=None, **kwargs):
close = verify_series(close)
volume = verify_series(volume)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import AD
ad = AD(high, low, close, volume)
else:
@@ -70,6 +71,8 @@ Args:
close (pd.Series): Series of 'close's
volume (pd.Series): Series of 'volume's
open (pd.Series): Series of 'open's
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/volume/adosc.py
+6 -3
View File
@@ -5,7 +5,7 @@ from pandas_ta.overlap import ema
from pandas_ta.utils import get_offset, verify_series
def adosc(high, low, close, volume, open_=None, fast=None, slow=None, offset=None, **kwargs):
def adosc(high, low, close, volume, open_=None, fast=None, slow=None, talib=None, offset=None, **kwargs):
"""Indicator: Accumulation/Distribution Oscillator"""
# Validate Arguments
fast = int(fast) if fast and fast > 0 else 3
@@ -17,13 +17,14 @@ def adosc(high, low, close, volume, open_=None, fast=None, slow=None, offset=Non
volume = verify_series(volume, _length)
offset = get_offset(offset)
if "length" in kwargs: kwargs.pop("length")
mode_tal = bool(talib) if isinstance(talib, bool) else True
if high is None or low is None or close is None or volume is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import ADOSC
adosc = ADOSC(high, low, close, volume)
adosc = ADOSC(high, low, close, volume, fast, slow)
else:
ad_ = ad(high=high, low=low, close=close, volume=volume, open_=open_)
fast_ad = ema(close=ad_, length=fast, **kwargs)
@@ -74,6 +75,8 @@ Args:
volume (pd.Series): Series of 'volume's
fast (int): The short period. Default: 12
slow (int): The long period. Default: 26
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/volume/aobv.py
+1 -1
View File
@@ -6,7 +6,7 @@ from pandas_ta.trend import long_run, short_run
from pandas_ta.utils import get_offset, verify_series
def aobv(close, volume, fast=None, slow=None, mamode=None, max_lookback=None, min_lookback=None, offset=None, **kwargs):
def aobv(close, volume, fast=None, slow=None, max_lookback=None, min_lookback=None, mamode=None, offset=None, **kwargs):
"""Indicator: Archer On Balance Volume (AOBV)"""
# Validate arguments
fast = int(fast) if fast and fast > 0 else 4
pandas_ta/volume/efi.py
+2 -2
View File
@@ -3,7 +3,7 @@ from pandas_ta.overlap import ma
from pandas_ta.utils import get_drift, get_offset, verify_series
def efi(close, volume, length=None, drift=None, mamode=None, offset=None, **kwargs):
def efi(close, volume, length=None, mamode=None, drift=None, offset=None, **kwargs):
"""Indicator: Elder's Force Index (EFI)"""
# Validate arguments
length = int(length) if length and length > 0 else 13
@@ -63,7 +63,7 @@ Args:
volume (pd.Series): Series of 'volume's
length (int): The short period. Default: 13
drift (int): The diff period. Default: 1
mamode (str): Two options: None or "sma". Default: None
mamode (str): See ```help(ta.ma)```. Default: 'ema'
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/volume/kvo.py
+23 -38
View File
@@ -1,18 +1,17 @@
# -*- coding: utf-8 -*-
from numpy import where as npWhere
from pandas import DataFrame
from pandas_ta.overlap import hlc3, ma
from pandas_ta.utils import get_drift, get_offset, non_zero_range, verify_series
from pandas_ta.utils import get_drift, get_offset, signed_series, verify_series
def kvo(high, low, close, volume, fast=None, slow=None, length_sig=None, mamode=None, drift=None, offset=None, **kwargs):
def kvo(high, low, close, volume, fast=None, slow=None, signal=None, mamode=None, drift=None, offset=None, **kwargs):
"""Indicator: Klinger Volume Oscillator (KVO)"""
# Validate arguments
fast = int(fast) if fast and fast > 0 else 34
slow = int(slow) if slow and slow > 0 else 55
length_sig = int(length_sig) if length_sig and length_sig > 0 else 13
signal = int(signal) if signal and signal > 0 else 13
mamode = mamode.lower() if mamode and isinstance(mamode, str) else "ema"
_length = max(fast, slow, length_sig)
_length = max(fast, slow, signal)
high = verify_series(high, _length)
low = verify_series(low, _length)
close = verify_series(close, _length)
@@ -23,19 +22,10 @@ def kvo(high, low, close, volume, fast=None, slow=None, length_sig=None, mamode=
if high is None or low is None or close is None or volume is None: return
# Calculate Result
mom = hlc3(high, low, close).diff(drift)
trend = npWhere(mom > 0, 1, 0) + npWhere(mom < 0, -1, 0)
dm = non_zero_range(high, low)
m = high.size
cm = [0] * m
for i in range(1, m):
cm[i] = (cm[i - 1] + dm[i]) if trend[i] == trend[i - 1] else (dm[i - 1] + dm[i])
vf = 100 * volume * trend * abs(2 * dm / cm - 1)
kvo = ma(mamode, vf, length=fast) - ma(mamode, vf, length=slow)
kvo_signal = ma(mamode, kvo, length=length_sig)
signed_volume = volume * signed_series(hlc3(high, low, close), 1)
sv = signed_volume.loc[signed_volume.first_valid_index():,]
kvo = ma(mamode, sv, length=fast) - ma(mamode, sv, length=slow)
kvo_signal = ma(mamode, kvo.loc[kvo.first_valid_index():,], length=signal)
# Offset
if offset != 0:
@@ -51,17 +41,18 @@ def kvo(high, low, close, volume, fast=None, slow=None, length_sig=None, mamode=
kvo_signal.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
kvo.name = f"KVO_{fast}_{slow}"
kvo_signal.name = f"KVOSig_{length_sig}"
_props = f"_{fast}_{slow}_{signal}"
kvo.name = f"KVO{_props}"
kvo_signal.name = f"KVOs{_props}"
kvo.category = kvo_signal.category = "volume"
# Prepare DataFrame to return
data = {kvo.name: kvo, kvo_signal.name: kvo_signal}
kvoandsig = DataFrame(data)
kvoandsig.name = f"KVO_{fast}_{slow}_{length_sig}"
kvoandsig.category = kvo.category
df = DataFrame(data)
df.name = f"KVO{_props}"
df.category = kvo.category
return kvoandsig
return df
kvo.__doc__ = \
@@ -71,23 +62,17 @@ This indicator was developed by Stephen J. Klinger. It is designed to predict
price reversals in a market by comparing volume to price.
Sources:
https://www.tradingview.com/script/Qnn7ymRK-Klinger-Volume-Oscillator/
https://www.investopedia.com/terms/k/klingeroscillator.asp
https://www.daytrading.com/klinger-volume-oscillator
Calculation:
Default Inputs:
fast=34, slow=55, length_sig=13, drift=1
MOM = HLC3.diff(drift)
NEG_TREND = -1 if MOM < 0 else 0
POS_TREND = 1 if MOM > 0 else 0
TREND = POS_TREND + NEG_TREND
DM = high - low
CM = [CMt-1 + DMt if TRENDt == TRENDt-1 else DMt-1 + DMt]
vf = 100 * volume * TREND * abs(2 * dm / cm - 1)
kvo = ema(vf, fast) - ema(vf, slow)
kvo_signal = ema(kvo, length_sig)
fast=34, slow=55, signal=13, drift=1
EMA = Exponential Moving Average
SV = volume * signed_series(HLC3, 1)
KVO = EMA(SV, fast) - EMA(SV, slow)
Signal = EMA(KVO, signal)
Args:
high (pd.Series): Series of 'high's
@@ -97,7 +82,7 @@ Args:
fast (int): The fast period. Default: 34
long (int): The long period. Default: 55
length_sig (int): The signal period. Default: 13
mamode (str): "sma", "ema", "wma" or "rma". Default: "ema"
mamode (str): See ```help(ta.ma)```. Default: 'ema'
offset (int): How many periods to offset the result. Default: 0
Kwargs:
@@ -105,5 +90,5 @@ Kwargs:
fill_method (value, optional): Type of fill method
Returns:
pd.DataFrame: kvo and kvo_signal columns.
pd.DataFrame: KVO and Signal columns.
"""
pandas_ta/volume/mfi.py
+6 -3
View File
@@ -5,7 +5,7 @@ from pandas_ta.overlap import hlc3
from pandas_ta.utils import get_drift, get_offset, verify_series
def mfi(high, low, close, volume, length=None, drift=None, offset=None, **kwargs):
def mfi(high, low, close, volume, length=None, talib=None, drift=None, offset=None, **kwargs):
"""Indicator: Money Flow Index (MFI)"""
# Validate arguments
length = int(length) if length and length > 0 else 14
@@ -15,13 +15,14 @@ def mfi(high, low, close, volume, length=None, drift=None, offset=None, **kwargs
volume = verify_series(volume, length)
drift = get_drift(drift)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if high is None or low is None or close is None or volume is None: return
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import MFI
mfi = MFI(high, low, close, volume)
mfi = MFI(high, low, close, volume, length)
else:
typical_price = hlc3(high=high, low=low, close=close)
raw_money_flow = typical_price * volume
@@ -84,6 +85,8 @@ Args:
close (pd.Series): Series of 'close's
volume (pd.Series): Series of 'volume's
length (int): The sum period. Default: 14
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
pandas_ta/volume/nvi.py
+1 -1
View File
@@ -17,7 +17,7 @@ def nvi(close, volume, length=None, initial=None, offset=None, **kwargs):
# Calculate Result
roc_ = roc(close=close, length=length)
signed_volume = signed_series(volume, initial=1)
signed_volume = signed_series(volume, 1)
nvi = signed_volume[signed_volume < 0].abs() * roc_
nvi.fillna(0, inplace=True)
nvi.iloc[0] = initial
pandas_ta/volume/obv.py
+5 -2
View File
@@ -3,15 +3,16 @@ from pandas_ta import Imports
from pandas_ta.utils import get_offset, signed_series, verify_series
def obv(close, volume, offset=None, **kwargs):
def obv(close, volume, talib=None, offset=None, **kwargs):
"""Indicator: On Balance Volume (OBV)"""
# Validate arguments
close = verify_series(close)
volume = verify_series(volume)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
# Calculate Result
if Imports["talib"]:
if Imports["talib"] and mode_tal:
from talib import OBV
obv = OBV(close, volume)
else:
@@ -53,6 +54,8 @@ Calculation:
Args:
close (pd.Series): Series of 'close's
volume (pd.Series): Series of 'volume's
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/volume/pvi.py
+2 -3
View File
@@ -16,9 +16,8 @@ def pvi(close, volume, length=None, initial=None, offset=None, **kwargs):
if close is None or volume is None: return
# Calculate Result
roc_ = roc(close=close, length=length)
signed_volume = signed_series(volume, initial=1)
pvi = signed_volume[signed_volume > 0].abs() * roc_
signed_volume = signed_series(volume, 1)
pvi = roc(close=close, length=length) * signed_volume[signed_volume > 0].abs()
pvi.fillna(0, inplace=True)
pvi.iloc[0] = initial
pvi = pvi.cumsum()
pandas_ta/volume/pvol.py
+2 -3
View File
@@ -11,10 +11,9 @@ def pvol(close, volume, offset=None, **kwargs):
signed = kwargs.pop("signed", False)
# Calculate Result
pvol = close * volume
if signed:
pvol = signed_series(close, 1) * close * volume
else:
pvol = close * volume
pvol *= signed_series(close, 1)
# Offset
if offset != 0:
pandas_ta/volume/vp.py
+3 -3
View File
@@ -16,10 +16,10 @@ def vp(close, volume, width=None, **kwargs):
if close is None or volume is None: return
# Setup
signed_price = signed_series(close, initial=1)
pos_volume = signed_price[signed_price > 0] * volume
signed_price = signed_series(close, 1)
pos_volume = volume * signed_price[signed_price > 0]
pos_volume.name = volume.name
neg_volume = signed_price[signed_price < 0] * -volume
neg_volume = -volume * signed_price[signed_price < 0]
neg_volume.name = volume.name
vp = concat([close, pos_volume, neg_volume], axis=1)
requirements.txt
+5 -2
View File
@@ -1,2 +1,5 @@
numpy>=1.20.2
pandas>=1.2.4
numpy==1.19.5
pandas==1.2.0
python-dateutil==2.8.1
pytz==2021.1
six==1.16.0
setup.py
+1 -1
View File
@@ -19,7 +19,7 @@ setup(
"pandas_ta.volatility",
"pandas_ta.volume"
],
version=".".join(("0", "3", "02b")),
version=".".join(("0", "3", "14b")),
description=long_description,
long_description=long_description,
author="Kevin Johnson",
tests/test_ext_indicator_candle.py
+1 -2
View File
@@ -1,7 +1,7 @@
from .config import sample_data
from .context import pandas_ta
from unittest import TestCase
from unittest import TestCase, skip
from pandas import DataFrame
@@ -17,7 +17,6 @@ class TestCandleExtension(TestCase):
def setUp(self): pass
def tearDown(self): pass
def test_cdl_doji_ext(self):
self.data.ta.cdl_pattern("doji", append=True)
self.assertIsInstance(self.data, DataFrame)
tests/test_ext_indicator_momentum.py
+1 -1
View File
@@ -247,7 +247,7 @@ class TestMomentumExtension(TestCase):
def test_tsi_ext(self):
self.data.ta.tsi(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "TSI_13_25")
self.assertEqual(list(self.data.columns[-2:]), ["TSI_13_25_13", "TSIs_13_25_13"])
def test_uo_ext(self):
self.data.ta.uo(append=True)
tests/test_ext_indicator_overlap_ext.py
+5
View File
@@ -63,6 +63,11 @@ class TestOverlapExtension(TestCase):
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "HWMA_0.2_0.1_0.1")
def test_jma_ext(self):
self.data.ta.jma(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "JMA_7_0")
def test_kama_ext(self):
self.data.ta.kama(append=True)
self.assertIsInstance(self.data, DataFrame)
tests/test_ext_indicator_volume.py
+1 -1
View File
@@ -63,7 +63,7 @@ class TestVolumeExtension(TestCase):
def test_kvo_ext(self):
self.data.ta.kvo(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "KVOSig_13")
self.assertEqual(list(self.data.columns[-2:]), ["KVO_34_55_13", "KVOs_34_55_13"])
def test_mfi_ext(self):
self.data.ta.mfi(append=True)
tests/test_indicator_candle.py
+1 -1
View File
@@ -58,7 +58,7 @@ class TestCandle(TestCase):
try:
expected = tal.CDLDOJI(self.open, self.high, self.low, self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
tests/test_indicator_momentum.py
+82 -29
View File
@@ -65,60 +65,72 @@ class TestMomentum(TestCase):
self.assertEqual(result.name, "AO_5_34")
def test_apo(self):
result = pandas_ta.apo(self.close)
result = pandas_ta.apo(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "APO_12_26")
try:
expected = tal.APO(self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.apo(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "APO_12_26")
def test_bias(self):
result = pandas_ta.bias(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "BIAS_SMA_26")
def test_bop(self):
result = pandas_ta.bop(self.open, self.high, self.low, self.close)
result = pandas_ta.bop(self.open, self.high, self.low, self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "BOP")
try:
expected = tal.BOP(self.open, self.high, self.low, self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.bop(self.open, self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "BOP")
def test_brar(self):
result = pandas_ta.brar(self.open, self.high, self.low, self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "BRAR_26")
def test_cci(self):
result = pandas_ta.cci(self.high, self.low, self.close)
result = pandas_ta.cci(self.high, self.low, self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "CCI_14_0.015")
try:
expected = tal.CCI(self.high, self.low, self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.cci(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "CCI_14_0.015")
def test_cfo(self):
result = pandas_ta.cfo(self.close)
self.assertIsInstance(result, Series)
@@ -137,13 +149,17 @@ class TestMomentum(TestCase):
try:
expected = tal.CMO(self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.cmo(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "CMO_14")
def test_coppock(self):
result = pandas_ta.coppock(self.close)
self.assertIsInstance(result, Series)
@@ -160,7 +176,7 @@ class TestMomentum(TestCase):
self.assertEqual(result.name, "ER_10")
def test_dm(self):
result = pandas_ta.dm(self.high, self.low)
result = pandas_ta.dm(self.high, self.low, talib=False)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "DM_14")
@@ -169,7 +185,7 @@ class TestMomentum(TestCase):
expected_neg = tal.MINUS_DM(self.high, self.low)
expecteddf = DataFrame({"DMP_14": expected_pos, "DMN_14": expected_neg})
pdt.assert_frame_equal(result, expecteddf)
except AssertionError as ae:
except AssertionError:
try:
dmp = pandas_ta.utils.df_error_analysis(result.iloc[:,0], expecteddf.iloc[:,0], col=CORRELATION)
self.assertGreater(dmp, CORRELATION_THRESHOLD)
@@ -182,6 +198,10 @@ class TestMomentum(TestCase):
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.dm(self.high, self.low)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "DM_14")
def test_eri(self):
result = pandas_ta.eri(self.high, self.low, self.close)
self.assertIsInstance(result, DataFrame)
@@ -216,7 +236,7 @@ class TestMomentum(TestCase):
self.assertEqual(result.name, "KST_10_15_20_30_10_10_10_15_9")
def test_macd(self):
result = pandas_ta.macd(self.close)
result = pandas_ta.macd(self.close, talib=False)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "MACD_12_26_9")
@@ -224,7 +244,7 @@ class TestMomentum(TestCase):
expected = tal.MACD(self.close)
expecteddf = DataFrame({"MACD_12_26_9": expected[0], "MACDh_12_26_9": expected[2], "MACDs_12_26_9": expected[1]})
pdt.assert_frame_equal(result, expecteddf)
except AssertionError as ae:
except AssertionError:
try:
macd_corr = pandas_ta.utils.df_error_analysis(result.iloc[:, 0], expecteddf.iloc[:, 0], col=CORRELATION)
self.assertGreater(macd_corr, CORRELATION_THRESHOLD)
@@ -243,41 +263,58 @@ class TestMomentum(TestCase):
except Exception as ex:
error_analysis(result.iloc[:, 2], CORRELATION, ex, newline=False)
result = pandas_ta.macd(self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "MACD_12_26_9")
def test_macdas(self):
result = pandas_ta.macd(self.close, asmode=True)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "MACDAS_12_26_9")
def test_mom(self):
result = pandas_ta.mom(self.close)
result = pandas_ta.mom(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "MOM_10")
try:
expected = tal.MOM(self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.mom(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "MOM_10")
def test_pgo(self):
result = pandas_ta.pgo(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "PGO_14")
def test_ppo(self):
result = pandas_ta.ppo(self.close)
result = pandas_ta.ppo(self.close, talib=False)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "PPO_12_26_9")
try:
expected = tal.PPO(self.close)
pdt.assert_series_equal(result["PPO_12_26_9"], expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result["PPO_12_26_9"], expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result["PPO_12_26_9"], CORRELATION, ex)
result = pandas_ta.ppo(self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "PPO_12_26_9")
def test_psl(self):
result = pandas_ta.psl(self.close)
self.assertIsInstance(result, Series)
@@ -294,35 +331,43 @@ class TestMomentum(TestCase):
self.assertEqual(result.name, "QQE_14_5_4.236")
def test_roc(self):
result = pandas_ta.roc(self.close)
result = pandas_ta.roc(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "ROC_10")
try:
expected = tal.ROC(self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.roc(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "ROC_10")
def test_rsi(self):
result = pandas_ta.rsi(self.close)
result = pandas_ta.rsi(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "RSI_14")
try:
expected = tal.RSI(self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.rsi(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "RSI_14")
def test_rsx(self):
result = pandas_ta.rsx(self.close)
self.assertIsInstance(result, Series)
@@ -406,10 +451,10 @@ class TestMomentum(TestCase):
self.assertEqual(result.name, "STOCH_14_3_3")
try:
expected = tal.STOCH(self.high, self.low, self.close, 14, 3, 0, 3)
expecteddf = DataFrame({"STOCHk_14_3_0_3": expected[0], "STOCHd_14_3_0_3": expected[1]})
expected = tal.STOCH(self.high, self.low, self.close, 14, 3, 0, 3, 0)
expecteddf = DataFrame({"STOCHk_14_3_0_3_0": expected[0], "STOCHd_14_3_0_3": expected[1]})
pdt.assert_frame_equal(result, expecteddf)
except AssertionError as ae:
except AssertionError:
try:
stochk_corr = pandas_ta.utils.df_error_analysis(result.iloc[:, 0], expecteddf.iloc[:, 0], col=CORRELATION)
self.assertGreater(stochk_corr, CORRELATION_THRESHOLD)
@@ -432,7 +477,7 @@ class TestMomentum(TestCase):
expected = tal.STOCHRSI(self.close, 14, 14, 3, 0)
expecteddf = DataFrame({"STOCHRSIk_14_14_0_3": expected[0], "STOCHRSId_14_14_3_0": expected[1]})
pdt.assert_frame_equal(result, expecteddf)
except AssertionError as ae:
except AssertionError:
try:
stochrsid_corr = pandas_ta.utils.df_error_analysis(result.iloc[:, 0], expecteddf.iloc[:, 1], col=CORRELATION)
self.assertGreater(stochrsid_corr, CORRELATION_THRESHOLD)
@@ -453,35 +498,43 @@ class TestMomentum(TestCase):
def test_tsi(self):
result = pandas_ta.tsi(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "TSI_13_25")
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "TSI_13_25_13")
def test_uo(self):
result = pandas_ta.uo(self.high, self.low, self.close)
result = pandas_ta.uo(self.high, self.low, self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "UO_7_14_28")
try:
expected = tal.ULTOSC(self.high, self.low, self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.uo(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "UO_7_14_28")
def test_willr(self):
result = pandas_ta.willr(self.high, self.low, self.close)
result = pandas_ta.willr(self.high, self.low, self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "WILLR_14")
try:
expected = tal.WILLR(self.high, self.low, self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.willr(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "WILLR_14")
tests/test_indicator_overlap.py
+107 -29
View File
@@ -40,20 +40,24 @@ class TestOverlap(TestCase):
self.assertEqual(result.name, "ALMA_10_6.0_0.85")
def test_dema(self):
result = pandas_ta.dema(self.close)
result = pandas_ta.dema(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "DEMA_10")
try:
expected = tal.DEMA(self.close, 10)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.dema(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "DEMA_10")
def test_ema(self):
result = pandas_ta.ema(self.close, presma=False)
self.assertIsInstance(result, Series)
@@ -62,13 +66,30 @@ class TestOverlap(TestCase):
try:
expected = tal.EMA(self.close, 10)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.ema(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "EMA_10")
try:
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.ema(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "EMA_10")
def test_fwma(self):
result = pandas_ta.fwma(self.close)
self.assertIsInstance(result, Series)
@@ -85,20 +106,24 @@ class TestOverlap(TestCase):
self.assertEqual(result.name, "HL2")
def test_hlc3(self):
result = pandas_ta.hlc3(self.high, self.low, self.close)
result = pandas_ta.hlc3(self.high, self.low, self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "HLC3")
try:
expected = tal.TYPPRICE(self.high, self.low, self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.hlc3(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "HLC3")
def test_hma(self):
result = pandas_ta.hma(self.close)
self.assertIsInstance(result, Series)
@@ -114,6 +139,11 @@ class TestOverlap(TestCase):
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "KAMA_10_2_30")
def test_jma(self):
result = pandas_ta.jma(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "JMA_7_0")
def test_ichimoku(self):
ichimoku, span = pandas_ta.ichimoku(self.high, self.low, self.close)
self.assertIsInstance(ichimoku, DataFrame)
@@ -122,70 +152,86 @@ class TestOverlap(TestCase):
self.assertEqual(span.name, "ICHISPAN_9_26")
def test_linreg(self):
result = pandas_ta.linreg(self.close)
result = pandas_ta.linreg(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "LR_14")
try:
expected = tal.LINEARREG(self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.linreg(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "LR_14")
def test_linreg_angle(self):
result = pandas_ta.linreg(self.close, angle=True)
result = pandas_ta.linreg(self.close, angle=True, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "LRa_14")
try:
expected = tal.LINEARREG_ANGLE(self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.linreg(self.close, angle=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "LRa_14")
def test_linreg_intercept(self):
result = pandas_ta.linreg(self.close, intercept=True)
result = pandas_ta.linreg(self.close, intercept=True, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "LRb_14")
try:
expected = tal.LINEARREG_INTERCEPT(self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.linreg(self.close, intercept=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "LRb_14")
def test_linreg_r(self):
result = pandas_ta.linreg(self.close, r=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "LRr_14")
def test_linreg_slope(self):
result = pandas_ta.linreg(self.close, slope=True)
result = pandas_ta.linreg(self.close, slope=True, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "LRm_14")
try:
expected = tal.LINEARREG_SLOPE(self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.linreg(self.close, slope=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "LRm_14")
def test_ma(self):
result = pandas_ta.ma()
self.assertIsInstance(result, list)
@@ -205,35 +251,43 @@ class TestOverlap(TestCase):
self.assertEqual(result.name, "MCGD_10")
def test_midpoint(self):
result = pandas_ta.midpoint(self.close)
result = pandas_ta.midpoint(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "MIDPOINT_2")
try:
expected = tal.MIDPOINT(self.close, 2)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.midpoint(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "MIDPOINT_2")
def test_midprice(self):
result = pandas_ta.midprice(self.high, self.low)
result = pandas_ta.midprice(self.high, self.low, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "MIDPRICE_2")
try:
expected = tal.MIDPRICE(self.high, self.low, 2)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.midprice(self.high, self.low)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "MIDPRICE_2")
def test_ohlc4(self):
result = pandas_ta.ohlc4(self.open, self.high, self.low, self.close)
self.assertIsInstance(result, Series)
@@ -255,20 +309,24 @@ class TestOverlap(TestCase):
self.assertEqual(result.name, "SINWMA_14")
def test_sma(self):
result = pandas_ta.sma(self.close)
result = pandas_ta.sma(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "SMA_10")
try:
expected = tal.SMA(self.close, 10)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.sma(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "SMA_10")
def test_ssf(self):
result = pandas_ta.ssf(self.close, poles=2)
self.assertIsInstance(result, Series)
@@ -289,50 +347,62 @@ class TestOverlap(TestCase):
self.assertEqual(result.name, "SUPERT_7_3.0")
def test_t3(self):
result = pandas_ta.t3(self.close)
result = pandas_ta.t3(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "T3_10_0.7")
try:
expected = tal.T3(self.close, 10)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.t3(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "T3_10_0.7")
def test_tema(self):
result = pandas_ta.tema(self.close)
result = pandas_ta.tema(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "TEMA_10")
try:
expected = tal.TEMA(self.close, 10)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.tema(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "TEMA_10")
def test_trima(self):
result = pandas_ta.trima(self.close)
result = pandas_ta.trima(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "TRIMA_10")
try:
expected = tal.TRIMA(self.close, 10)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.trima(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "TRIMA_10")
def test_vidya(self):
result = pandas_ta.vidya(self.close)
self.assertIsInstance(result, Series)
@@ -349,35 +419,43 @@ class TestOverlap(TestCase):
self.assertEqual(result.name, "VWMA_10")
def test_wcp(self):
result = pandas_ta.wcp(self.high, self.low, self.close)
result = pandas_ta.wcp(self.high, self.low, self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "WCP")
try:
expected = tal.WCLPRICE(self.high, self.low, self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.wcp(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "WCP")
def test_wma(self):
result = pandas_ta.wma(self.close)
result = pandas_ta.wma(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "WMA_10")
try:
expected = tal.WMA(self.close, 10)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.wma(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "WMA_10")
def test_zlma(self):
result = pandas_ta.zlma(self.close)
self.assertIsInstance(result, Series)
tests/test_indicator_statistics.py
+12 -4
View File
@@ -65,20 +65,24 @@ class TestStatistics(TestCase):
self.assertEqual(result.name, "SKEW_30")
def test_stdev(self):
result = pandas_ta.stdev(self.close)
result = pandas_ta.stdev(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "STDEV_30")
try:
expected = tal.STDDEV(self.close, 30)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.stdev(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "STDEV_30")
def test_tos_sdtevall(self):
result = pandas_ta.tos_stdevall(self.close)
self.assertIsInstance(result, DataFrame)
@@ -96,20 +100,24 @@ class TestStatistics(TestCase):
self.assertEqual(len(result.columns), 5)
def test_variance(self):
result = pandas_ta.variance(self.close)
result = pandas_ta.variance(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "VAR_30")
try:
expected = tal.VAR(self.close, 30)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.variance(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "VAR_30")
def test_zscore(self):
result = pandas_ta.zscore(self.close)
self.assertIsInstance(result, Series)
tests/test_indicator_trend.py
+16 -6
View File
@@ -35,27 +35,31 @@ class TestTrend(TestCase):
def test_adx(self):
result = pandas_ta.adx(self.high, self.low, self.close)
result = pandas_ta.adx(self.high, self.low, self.close, talib=False)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "ADX_14")
try:
expected = tal.ADX(self.high, self.low, self.close)
pdt.assert_series_equal(result.iloc[:, 0], expected)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result.iloc[:, 0], expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.adx(self.high, self.low, self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "ADX_14")
def test_amat(self):
result = pandas_ta.amat(self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "AMATe_8_21_2")
def test_aroon(self):
result = pandas_ta.aroon(self.high, self.low)
result = pandas_ta.aroon(self.high, self.low, talib=False)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "AROON_14")
@@ -63,7 +67,7 @@ class TestTrend(TestCase):
expected = tal.AROON(self.high, self.low)
expecteddf = DataFrame({"AROOND_14": expected[0], "AROONU_14": expected[1]})
pdt.assert_frame_equal(result, expecteddf)
except AssertionError as ae:
except AssertionError:
try:
aroond_corr = pandas_ta.utils.df_error_analysis(result.iloc[:, 0], expecteddf.iloc[:, 0], col=CORRELATION)
self.assertGreater(aroond_corr, CORRELATION_THRESHOLD)
@@ -76,13 +80,19 @@ class TestTrend(TestCase):
except Exception as ex:
error_analysis(result.iloc[:, 1], CORRELATION, ex, newline=False)
result = pandas_ta.aroon(self.high, self.low)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "AROON_14")
def test_aroon_osc(self):
result = pandas_ta.aroon(self.high, self.low)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "AROON_14")
try:
expected = tal.AROONOSC(self.high, self.low)
pdt.assert_series_equal(result.iloc[:, 2], expected)
except AssertionError as ae:
except AssertionError:
try:
aroond_corr = pandas_ta.utils.df_error_analysis(result.iloc[:,2], expected,col=CORRELATION)
self.assertGreater(aroond_corr, CORRELATION_THRESHOLD)
@@ -158,7 +168,7 @@ class TestTrend(TestCase):
try:
expected = tal.SAR(self.high, self.low)
pdt.assert_series_equal(psar, expected)
except AssertionError as ae:
except AssertionError:
try:
psar_corr = pandas_ta.utils.df_error_analysis(psar, expected, col=CORRELATION)
self.assertGreater(psar_corr, CORRELATION_THRESHOLD)
tests/test_indicator_volatility.py
+28 -12
View File
@@ -45,26 +45,26 @@ class TestVolatility(TestCase):
self.assertEqual(result.name, "ACCBANDS_20")
def test_atr(self):
result = pandas_ta.atr(self.high, self.low, self.close)
result = pandas_ta.atr(self.high, self.low, self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "ATRr_14")
try:
expected = tal.ATR(self.high, self.low, self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
def test_bbands(self):
result = pandas_ta.bbands(self.close, ddof=0)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "BBANDS_5_2.0")
result = pandas_ta.atr(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "ATRr_14")
result = pandas_ta.bbands(self.close, ddof=1)
def test_bbands(self):
result = pandas_ta.bbands(self.close, talib=False)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "BBANDS_5_2.0")
@@ -72,7 +72,7 @@ class TestVolatility(TestCase):
expected = tal.BBANDS(self.close)
expecteddf = DataFrame({"BBU_5_2.0": expected[0], "BBM_5_2.0": expected[1], "BBL_5_2.0": expected[2]})
pdt.assert_frame_equal(result, expecteddf)
except AssertionError as ae:
except AssertionError:
try:
bbl_corr = pandas_ta.utils.df_error_analysis(result.iloc[:, 0], expecteddf.iloc[:,0], col=CORRELATION)
self.assertGreater(bbl_corr, CORRELATION_THRESHOLD)
@@ -91,6 +91,14 @@ class TestVolatility(TestCase):
except Exception as ex:
error_analysis(result.iloc[:, 2], CORRELATION, ex, newline=False)
result = pandas_ta.bbands(self.close, ddof=0)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "BBANDS_5_2.0")
result = pandas_ta.bbands(self.close, ddof=1)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "BBANDS_5_2.0")
def test_donchian(self):
result = pandas_ta.donchian(self.high, self.low)
self.assertIsInstance(result, DataFrame)
@@ -115,20 +123,24 @@ class TestVolatility(TestCase):
self.assertEqual(result.name, "MASSI_9_25")
def test_natr(self):
result = pandas_ta.natr(self.high, self.low, self.close)
result = pandas_ta.natr(self.high, self.low, self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "NATR_14")
try:
expected = tal.NATR(self.high, self.low, self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.natr(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "NATR_14")
def test_pdist(self):
result = pandas_ta.pdist(self.open, self.high, self.low, self.close)
self.assertIsInstance(result, Series)
@@ -153,20 +165,24 @@ class TestVolatility(TestCase):
self.assertEqual(result.name, "THERMO_20_2_0.5")
def test_true_range(self):
result = pandas_ta.true_range(self.high, self.low, self.close)
result = pandas_ta.true_range(self.high, self.low, self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "TRUERANGE_1")
try:
expected = tal.TRANGE(self.high, self.low, self.close)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.true_range(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "TRUERANGE_1")
def test_ui(self):
result = pandas_ta.ui(self.close)
self.assertIsInstance(result, Series)
tests/test_indicator_volume.py
+24 -8
View File
@@ -35,40 +35,48 @@ class TestVolume(TestCase):
def test_ad(self):
result = pandas_ta.ad(self.high, self.low, self.close, self.volume_)
result = pandas_ta.ad(self.high, self.low, self.close, self.volume_, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "AD")
try:
expected = tal.AD(self.high, self.low, self.close, self.volume_)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.ad(self.high, self.low, self.close, self.volume_)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "AD")
def test_ad_open(self):
result = pandas_ta.ad(self.high, self.low, self.close, self.volume_, self.open)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "ADo")
def test_adosc(self):
result = pandas_ta.adosc(self.high, self.low, self.close, self.volume_)
result = pandas_ta.adosc(self.high, self.low, self.close, self.volume_, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "ADOSC_3_10")
try:
expected = tal.ADOSC(self.high, self.low, self.close, self.volume_)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.adosc(self.high, self.low, self.close, self.volume_)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "ADOSC_3_10")
def test_aobv(self):
result = pandas_ta.aobv(self.close, self.volume_)
self.assertIsInstance(result, DataFrame)
@@ -95,40 +103,48 @@ class TestVolume(TestCase):
self.assertEqual(result.name, "KVO_34_55_13")
def test_mfi(self):
result = pandas_ta.mfi(self.high, self.low, self.close, self.volume_)
result = pandas_ta.mfi(self.high, self.low, self.close, self.volume_, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "MFI_14")
try:
expected = tal.MFI(self.high, self.low, self.close, self.volume_)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.mfi(self.high, self.low, self.close, self.volume_)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "MFI_14")
def test_nvi(self):
result = pandas_ta.nvi(self.close, self.volume_)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "NVI_1")
def test_obv(self):
result = pandas_ta.obv(self.close, self.volume_)
result = pandas_ta.obv(self.close, self.volume_, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "OBV")
try:
expected = tal.OBV(self.close, self.volume_)
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError as ae:
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.obv(self.close, self.volume_)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "OBV")
def test_pvi(self):
result = pandas_ta.pvi(self.close, self.volume_)
self.assertIsInstance(result, Series)
tests/test_utils.py
+15
View File
@@ -149,12 +149,15 @@ class TestUtilities(TestCase):
result = self.utils.df_dates(self.data, ["1999-11-01", "2020-08-15", "2020-08-24", "2020-08-25", "2020-08-26", "2020-08-27"])
self.assertEqual(5, result.shape[0])
@skip
def test_df_month_to_date(self):
result = self.utils.df_month_to_date(self.data)
@skip
def test_df_quarter_to_date(self):
result = self.utils.df_quarter_to_date(self.data)
@skip
def test_df_year_to_date(self):
result = self.utils.df_year_to_date(self.data)
@@ -274,6 +277,18 @@ class TestUtilities(TestCase):
npt.assert_array_equal(self.utils.symmetric_triangle(n=5), array_5)
npt.assert_array_equal(self.utils.symmetric_triangle(n=5, weighted=True), array_5w)
def test_tal_ma(self):
self.assertEqual(self.utils.tal_ma("sma"), 0)
self.assertEqual(self.utils.tal_ma("Sma"), 0)
self.assertEqual(self.utils.tal_ma("ema"), 1)
self.assertEqual(self.utils.tal_ma("wma"), 2)
self.assertEqual(self.utils.tal_ma("dema"), 3)
self.assertEqual(self.utils.tal_ma("tema"), 4)
self.assertEqual(self.utils.tal_ma("trima"), 5)
self.assertEqual(self.utils.tal_ma("kama"), 6)
self.assertEqual(self.utils.tal_ma("mama"), 7)
self.assertEqual(self.utils.tal_ma("t3"), 8)
def test_zero(self):
self.assertEqual(self.utils.zero(-0.0000000000000001), 0)
self.assertEqual(self.utils.zero(0), 0)
tests/test_utils_metrics.py
+2 -4
View File
@@ -1,10 +1,9 @@
from .config import sample_data
from .context import pandas_ta
from unittest import skip, TestCase
from pandas import DataFrame
from .config import sample_data
from .context import pandas_ta
class TestUtilityMetrics(TestCase):
@@ -97,7 +96,6 @@ class TestUtilityMetrics(TestCase):
def test_pure_profit_score(self):
result = pandas_ta.pure_profit_score(self.close)
self.assertIsInstance(result, float)
self.assertGreaterEqual(result, 0)
def test_sharpe_ratio(self):