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

ENH squeeze ind added + notebooks updated MAINT refactoring DOC updates

Browse Source
This commit is contained in:
Kevin Johnson
2020-08-13 15:24:58 -07:00
parent fa24a4d96b
commit a3ef3872cb
25 changed files with 1946 additions and 1418 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3 -2
View File
@@ -119,6 +119,8 @@ pandas_ta/_wrapper.py
data/datas.csv
data/SPY_5min.csv
data/SPY_1min.csv
data/SPY_D_lbsz.csv
data/TV_5min.csv
data/similang-ch.csv
data/tulip.csv
examples/taplot.py
@@ -131,6 +133,7 @@ note.md
driver.py
doc.py
bt.ipynb
scratch.ipynb
ta_extension.ipynb
kerasmodeller.ipynb
Charts.ipynb
@@ -142,5 +145,3 @@ simple.ipynb
ta.json
# Indicator
pandas_ta/overlap/ft.py
pandas_ta/overlap/psar.py
README.md
+41 -19
View File
@@ -6,22 +6,44 @@
# __Technical Analysis Library in Python 3.7__
![Example Chart](/images/TA_Chart.png)
__Pandas Technical Analysis__ (Pandas TA) is an easy to use library that is built upon Python's Pandas library with more than 100 Indicators and Utility functions. These indicators are commonly used for financial time series datasets with columns or labels similar to: datetime, open, high, low, close, volume, et al. Many commonly used indicators are included, such as: _Simple Moving Average_ (*SMA*) _Moving Average Convergence Divergence_ (*MACD*), _Hull Exponential Moving Average_ (*HMA*), _Bollinger Bands_ (*BBANDS*), _On-Balance Volume_ (*OBV*), _Aroon & Aroon Oscillator_ (*AROON*) and more.
_Pandas Technical Analysis_ (**Pandas TA**) is an easy to use library that is built upon Python's Pandas library with more than 115 Indicators and Utility functions. These indicators are commonly used for financial time series datasets with columns or labels: datetime, open, high, low, close, volume, et al. Many commonly used indicators are included, such as: _Simple Moving Average_ (**sma**) _Moving Average Convergence Divergence_ (**macd**), _Hull Exponential Moving Average_ (**hma**), _Bollinger Bands_ (**bbands**), _On-Balance Volume_ (**obv**), _Aroon & Aroon Oscillator_ (**aroon**), _Squeeze_ (**squeeze**) and **many more**.
This version contains both the orignal code branch as well as a newly refactored branch with the option to use [Pandas DataFrame Extension](https://pandas.pydata.org/pandas-docs/stable/extending.html) mode.
All the indicators return a named Series or a DataFrame in uppercase underscore parameter format. For example, MACD(fast=12, slow=26, signal=9) will return a DataFrame with columns: ['MACD_12_26_9', 'MACDh_12_26_9', 'MACDs_12_26_9'].
**Pandas TA** has three different ways of processing Technical Indicators as described below. The **primary** requirement to run indicators in [Pandas DataFrame Extension](https://pandas.pydata.org/pandas-docs/stable/extending.html) mode, is that _open, high, low, close, volume_ are **lowercase**. Depending on the indicator, they either return a named Series or a DataFrame in uppercase underscore parameter format. For example, MACD(fast=12, slow=26, signal=9) will return a DataFrame with columns: ['MACD_12_26_9', 'MACDh_12_26_9', 'MACDs_12_26_9'].
## Pandas TA Issues, Ideas and Contributions
#### Thanks for trying **Pandas TA**!
Please take a moment to read **this** and the rest of this **README** before posting any issue.
* ### [Comments and Feedback](https://github.com/twopirllc/pandas-ta/issues)
* Have you read the rest of **this** document?
* Are you running the latest version?
* Have you tried the [Examples](https://github.com/twopirllc/pandas-ta/tree/master/examples/)?
* Did they help?
* What is missing?
* Could you help improve them?
* Did you know you can easily build _Custom Strategies_ with the **[Strategy](https://github.com/twopirllc/pandas-ta/blob/master/examples/PandasTA_Strategy_Examples.ipynb) Class**?
* Documentation needs improvement. Can you contribute?
* ### [Indicator or Feature Requests & Contributions](https://github.com/twopirllc/pandas-ta/issues)
* Please be as detailed and concise as possible. Links and screenshots and sometimes data samples are welcome.
* You want a new indicator not currently listed.
* You want an alternate version of an existing indicator.
* The indicator does not match another website, library, broker platform, language, et al.
* Can you contribute?
## __Features__
* Has 110+ indicators and utility functions.
* Option to use __multiprocessing__ when using df.ta.strategy(). See below.
* Example Jupyter Notebooks under the [examples](https://github.com/twopirllc/pandas-ta/tree/master/examples) directory, including how to create Custom Strategies using the new [__Strategy__ Class](https://github.com/twopirllc/pandas-ta/tree/master/examples/PandaTA_Strategy_Examples.ipynb)
* A new 'ta' method called 'strategy'. By default, it runs __all__ the indicators.
* Abbreviated Indicator names as listed below.
* Has 115+ indicators and utility functions.
* __Extended Pandas DataFrame__ as 'ta'.
* Indicators are correlation tested against the de facto [TA Lib](https://mrjbq7.github.io/ta-lib/) if they share common indicators.
* Example Jupyter Notebooks under the [examples](https://github.com/twopirllc/pandas-ta/tree/master/examples) directory, including how to create Custom Strategies using the new [__Strategy__ Class](https://github.com/twopirllc/pandas-ta/tree/master/examples/PandaTA_Strategy_Examples.ipynb)
* Option to use __multiprocessing__ when using df.ta.strategy(). See below.
* Easily add prefixes or suffixes or both to columns names.
* Categories similar to [TA-lib](https://github.com/mrjbq7/ta-lib/tree/master/docs/func_groups) and tightly correlated with TA Lib in testing.
* A new 'ta' method called 'strategy'. By default, it runs __all__ the indicators or equivalent ta.AllStrategy.
## __Recent Changes__
@@ -30,7 +52,7 @@ All the indicators return a named Series or a DataFrame in uppercase underscore
* Improved the calculation performance of indicators: _Exponential Moving Averagage_
and _Weighted Moving Average_.
* Removed internal core optimizations when running ```df.ta.strategy('all')``` with multiprocessing. See the ```ta.strategy()``` method for more details.
* __New Indicators:__ Kaufman's _Efficiency Ratio_ **er**, Johnson's _Pretty Good Oscillator_ **pgo**, _Elder Ray Index_ **eri**, Martin's _Ulcer Index_ **ui**
* __New Indicators:__ Kaufman's _Efficiency Ratio_ **er**, Johnson's _Pretty Good Oscillator_ **pgo**, _Elder Ray Index_ **eri**, Martin's _Ulcer Index_ **ui**, _Squeeze_ **squeeze** (John Carter's TTM **and** Lazybear's TradingView versions)
## What is a Pandas DataFrame Extension?
@@ -59,7 +81,7 @@ import pandas as pd
import pandas_ta as ta
# Load data
df = pd.read_csv('symbol.csv', sep=',')
df = pd.read_csv("path/symbol.csv", sep=",")
# Calculate Returns and append to the df DataFrame
df.ta.log_return(cumulative=True, append=True)
@@ -161,8 +183,8 @@ df.ta.mp = True
# Runs and appends all indicators to the current DataFrame by default
# The resultant DataFrame will be large.
df.ta.strategy()
# Or equivalently use name='all'
df.ta.strategy(name='all')
# Or equivalently use name="all"
df.ta.strategy(name="all")
# Use verbose if you want to make sure it is running.
df.ta.strategy(verbose=True)
@@ -177,7 +199,7 @@ df.ta.strategy(cores=2)
# Maybe you do not want certain indicators.
# Just exclude (a list of) them.
df.ta.strategy(exclude=['bop', 'mom', 'percent_return', 'wcp', 'pvi'], verbose=True)
df.ta.strategy(exclude=["bop", "mom", "percent_return", "wcp", "pvi"], verbose=True)
# Perhaps you want to use different values for indicators.
# This will run ALL indicators that have fast or slow as parameters.
@@ -255,7 +277,7 @@ time_series_in_order = df.ta.datetime_ordered
```python
# Set ta to default to an adjusted column, 'adj_close', overriding default 'close'
df.ta.adjusted = 'adj_close'
df.ta.adjusted = "adj_close"
df.ta.sma(length=10, append=True)
# To reset back to 'close', set adjusted back to None
@@ -269,7 +291,7 @@ df.ta.adjusted = None
* _Doji_: **cdl_doji**
* _Heikin-Ashi_: **ha**
## _Momentum_ (30)
## _Momentum_ (31)
* _Awesome Oscillator_: **ao**
* _Absolute Price Oscillator_: **apo**
@@ -295,7 +317,9 @@ df.ta.adjusted = None
* _Rate of Change_: **roc**
* _Relative Strength Index_: **rsi**
* _Relative Vigor Index_: **rvgi**
* _Slope_: **slope*
* _Slope_: **slope**
* _Squeeze_: **squeeze**
* Default is John Carter's. Enable Lazybear's by ```lazybear=True```
* _Stochastic Oscillator_: **stoch**
* _Trix_: **trix**
* _True strength index_: **tsi**
@@ -446,7 +470,5 @@ Use parameter: cumulative=**True** for cumulative results.
# Inspiration
* TradingView: http://www.tradingview.com
* Original TA-LIB: http://ta-lib.org/
Please leave any comments, feedback, suggestions, or indicator requests.
* TradingView: http://www.tradingview.com
examples/PandasTA_Strategy_Examples.ipynb
+505 -509
View File
File diff suppressed because it is too large Load Diff
examples/example.ipynb
+919 -669
View File
File diff suppressed because one or more lines are too long
pandas_ta/core.py
+46 -32
View File
@@ -8,6 +8,7 @@ from time import perf_counter
from typing import List
import pandas as pd
from numpy import ndarray as npndarray
from pandas_ta import categories
from pandas.core.base import PandasObject
@@ -21,21 +22,23 @@ from pandas_ta.volatility import *
from pandas_ta.volume import *
from pandas_ta.utils import *
version = ".".join(("0", "1", "81b"))
version = ".".join(("0", "1", "90b"))
# Dictionary of files for each category, used in df.ta.strategy()
Category = {name: category_files(name) for name in categories}
def mp_worker(args):
"""Multiprocessing Worker to handle different Methods."""
df, method, kwargs = args
if method != 'ichimoku':
if method != "ichimoku":
return df.ta(kind=method, **kwargs)
else:
return df.ta(kind=method, **kwargs)[0]
def finalize(method):
"""Adds Prefixes/Suffixes if given and Appends Results if True"""
@wraps(method)
def _wrapper(*class_methods, **method_kwargs):
cm = class_methods[0]
@@ -142,7 +145,7 @@ class BasePandasObject(PandasObject):
if len(df.columns) > 0:
self._df = df
else:
raise AttributeError(f" [X] No columns!")
raise AttributeError(f"[X] No columns!")
def __call__(self, kind, *args, **kwargs):
raise NotImplementedError()
@@ -355,41 +358,43 @@ class AnalysisIndicators(BasePandasObject):
return df.iloc[:,match[0]] if len(match) else print(NOT_FOUND)
def constants(self, append, lower_bound=-100, upper_bound=100, every=10):
def constants(self, append: bool, values: list):
"""Constants
Useful for creating indicator levels or if you need some constant value
easily added to your DataFrame.
Add or remove constants to the DataFrame easily with Numpy's arrays or
lists. Useful when you need easily accessible horizontal lines for
charting.
Add constant '1' to the DataFrame
>>> df.ta.constants(True, 1, 1, 1)
>>> df.ta.constants(True, [1])
Remove constant '1' to the DataFrame
>>> df.ta.constants(False, 1, 1, 1)
>>> df.ta.constants(False, [1])
Adding constants that range of constants from -4 to 4 inclusive
>>> df.ta.constants(True, -4, 4, 1)
Removing constants that range of constants from -4 to 4 inclusive
>>> df.ta.constants(False, -4, 4, 1)
Adding the constants for the charts
>>> import numpy as np
>>> chart_lines = np.append(np.arange(-4, 5, 1), np.arange(-100, 110, 10))
>>> df.ta.constants(True, chart_lines)
Removing some constants from the DataFrame
>>> df.ta.constants(False, np.array([-60, -40, 40, 60]))
Args:
append (bool): Default: None. If True, appends the range of constants to the
working DataFrame. If False, it removes the constant range from the working
DataFrame.
lower_bound (int): Default: -100. Lowest integer for the constant range.
upper_bound (int): Default: 100. Largest integer for the constant range.
every (int): Default: 10. How often to include a new constant.
append (bool): If True, appends a Numpy range of constants to the
working DataFrame. If False, it removes the constant range from
the working DataFrame. Default: None.
Returns:
Returns nothing to the user. Either adds or removes constant ranges from the
working DataFrame.
Returns the appended constants
Returns nothing to the user. Either adds or removes constant ranges
from the working DataFrame.
"""
levels = [x for x in range(lower_bound, upper_bound + 1) if x % every == 0]
if append:
for x in levels:
self._df[f'{x}'] = x
else:
for x in levels:
del self._df[f'{x}']
if isinstance(values, npndarray) or isinstance(values, list):
if append:
for x in values:
self._df[f"{x}"] = x
return self._df[self._df.columns[-len(values):]]
else:
for x in values:
del self._df[f"{x}"]
def indicators(self, **kwargs):
@@ -397,10 +402,10 @@ class AnalysisIndicators(BasePandasObject):
Args:
kwargs:
as_list (bool, optional): Default: False. When True, it returns a list
of the indicators. Helpful you want to filter out what you want to run.
exclude (list, optional): Default: None. The passed in list will be
excluded from the indicators list.
as_list (bool, optional): Default: False. When True, it
returns a list of the indicators.
exclude (list, optional): Default: None. The passed in list
will be excluded from the indicators list.
Returns:
Prints the list of indicators. If as_list=True, then a list.
@@ -469,7 +474,7 @@ class AnalysisIndicators(BasePandasObject):
excluded_functions = ["above", "above_value", "below", "below_value", "cross", "cross_value", "long_run", "short_run", "trend_return", "vp"]
excluded += user_excluded # Exclude user excluded ta if listed
print(f"\n[+] Strategy: {name}") if verbose else None
print(f"[+] Strategy: {name}") if verbose else None
if name.lower() in categories or is_all:
# Exclude special functions
excluded += excluded_functions
@@ -750,6 +755,15 @@ class AnalysisIndicators(BasePandasObject):
result = slope(close=close, length=length, offset=offset, **kwargs)
return result
@finalize
def squeeze(self, high=None, low=None, close=None, bb_length=None, bb_std=None, kc_length=None, kc_scalar=None, mom_length=None, mom_smooth=None, use_tr=None, offset=None, **kwargs):
high = self._get_column(high, 'high')
low = self._get_column(low, 'low')
close = self._get_column(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)
return result
@finalize
def stoch(self, high=None, low=None, close=None, fast_k=None, slow_k=None, slow_d=None, offset=None, **kwargs):
high = self._get_column(high, 'high')
pandas_ta/momentum/__init__.py
+1
View File
@@ -24,6 +24,7 @@ from .roc import roc
from .rsi import rsi
from .rvgi import rvgi
from .slope import slope
from .squeeze import squeeze
from .stoch import stoch
from .trix import trix
from .tsi import tsi
pandas_ta/momentum/rsi.py
+14 -16
View File
@@ -29,31 +29,29 @@ def rsi(close, length=None, scalar=None, drift=None, offset=None, **kwargs):
rsi = rsi.shift(offset)
# Handle fills
if 'fillna' in kwargs:
rsi.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
rsi.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
rsi.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
rsi.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
rsi.name = f"RSI_{length}"
rsi.category = 'momentum'
rsi.category = "momentum"
signal_indicators = kwargs.pop('signal_indicators', False)
signal_indicators = kwargs.pop("signal_indicators", False)
if signal_indicators:
signalsdf = concat(
[
DataFrame(
{rsi.name: rsi}
),
DataFrame({rsi.name: rsi}),
signals(
indicator=rsi,
xa=kwargs.pop('xa', 80),
xb=kwargs.pop('xb', 20),
xserie=kwargs.pop('xserie', None),
xserie_a=kwargs.pop('xserie_a', None),
xserie_b=kwargs.pop('xserie_b', None),
cross_values=kwargs.pop('cross_values', False),
cross_series=kwargs.pop('cross_series', True),
xa=kwargs.pop("xa", 80),
xb=kwargs.pop("xb", 20),
xserie=kwargs.pop("xserie", None),
xserie_a=kwargs.pop("xserie_a", None),
xserie_b=kwargs.pop("xserie_b", None),
cross_values=kwargs.pop("cross_values", False),
cross_series=kwargs.pop("cross_series", True),
offset=offset,
),
],
pandas_ta/momentum/squeeze.py
+222
View File
@@ -0,0 +1,222 @@
# -*- coding: utf-8 -*-
from numpy import NaN as npNaN
from pandas import concat, DataFrame
from pandas_ta.momentum import mom
from pandas_ta.overlap import ema, linreg, sma
from pandas_ta.statistics import stdev
from pandas_ta.trend import decreasing, increasing
from pandas_ta.volatility import bbands, kc, true_range
from pandas_ta.utils import get_drift, get_offset, high_low_range
from pandas_ta.utils import unsigned_differences, verify_series
def squeeze(high, low, close, bb_length=None, bb_std=None, kc_length=None, kc_scalar=None, mom_length=None, mom_smooth=None, use_tr=None, offset=None, **kwargs):
"""Indicator: Squeeze Momentum (SQZ)"""
# Validate arguments
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
offset = get_offset(offset)
bb_length = int(bb_length) if bb_length and bb_length > 0 else 20
bb_std = float(bb_std) if bb_std and bb_std > 0 else 2.
kc_length = int(kc_length) if kc_length and kc_length > 0 else 20
kc_scalar = float(kc_scalar) if kc_scalar and kc_scalar > 0 else 1.5
mom_length = int(mom_length) if mom_length and mom_length > 0 else 12
mom_smooth = int(mom_smooth) if mom_smooth and mom_smooth > 0 else 6
use_tr = kwargs.setdefault("tr", True)
asint = kwargs.pop("asint", True)
mamode = kwargs.pop("mamode", "sma").lower()
lazybear = kwargs.pop("lazybear", False)
detailed = kwargs.pop("detailed", False)
def simplify_columns(df, n=3):
df.columns = df.columns.str.lower()
return [c.split('_')[0][n-1:n] for c in df.columns]
# Calculate Result
bbd = bbands(close, length=bb_length, std=bb_std, mamode=mamode)
kch = kc(high, low, close, length=kc_length, scalar=kc_scalar, mamode=mamode, tr=use_tr)
# Simplify KC and BBAND column names for dynamic access
bbd.columns = simplify_columns(bbd)
kch.columns = simplify_columns(kch)
if lazybear:
highest_high = high.rolling(kc_length).max()
lowest_low = low.rolling(kc_length).min()
avg_ = 0.25 * (highest_high + lowest_low) + 0.5 * kch.b
squeeze = linreg(close - avg_, length=kc_length)
else:
momo = mom(close, length=mom_length)
if mamode == "ema":
squeeze = ema(momo, length=mom_smooth)
else:
squeeze = sma(momo, length=mom_smooth)
# Classify Squeezes
squeeze_on = (bbd.l > kch.l) & (bbd.u < kch.u)
squeeze_off = (bbd.l < kch.l) & (bbd.u > kch.u)
no_squeeze = ~squeeze_on & ~squeeze_off
# Offset
if offset != 0:
squeeze = squeeze.shift(offset)
squeeze_on = squeeze_on.shift(offset)
squeeze_off = squeeze_off.shift(offset)
no_squeeze = no_squeeze.shift(offset)
# Handle fills
if "fillna" in kwargs:
squeeze.fillna(kwargs["fillna"], inplace=True)
squeeze_on.fillna(kwargs["fillna"], inplace=True)
squeeze_off.fillna(kwargs["fillna"], inplace=True)
no_squeeze.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
squeeze.fillna(method=kwargs["fill_method"], inplace=True)
squeeze_on.fillna(method=kwargs["fill_method"], inplace=True)
squeeze_off.fillna(method=kwargs["fill_method"], inplace=True)
no_squeeze.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
_props = "" if use_tr else "hlr"
_props += f"_{bb_length}_{bb_std}_{kc_length}_{kc_scalar}"
_props += "_LB" if lazybear else ""
squeeze.name = f"SQZ{_props}"
data = {
squeeze.name: squeeze,
f"SQZ_ON": squeeze_on.astype(int) if asint else squeeze_on,
f"SQZ_OFF": squeeze_off.astype(int) if asint else squeeze_off,
f"SQZ_NO": no_squeeze.astype(int) if asint else no_squeeze
}
df = DataFrame(data)
df.name = squeeze.name
df.category = squeeze.category = "momentum"
# Detailed Squeeze Series
if detailed:
pos_squeeze = squeeze[squeeze >= 0]
neg_squeeze = squeeze[squeeze < 0]
pos_inc, pos_dec = unsigned_differences(pos_squeeze, asint=True)
neg_inc, neg_dec = unsigned_differences(neg_squeeze, asint=True)
pos_inc *= squeeze
pos_dec *= squeeze
neg_dec *= squeeze
neg_inc *= squeeze
pos_inc.replace(0, npNaN, inplace=True)
pos_dec.replace(0, npNaN, inplace=True)
neg_dec.replace(0, npNaN, inplace=True)
neg_inc.replace(0, npNaN, inplace=True)
sqz_inc = squeeze * increasing(squeeze)
sqz_dec = squeeze * decreasing(squeeze)
sqz_inc.replace(0, npNaN, inplace=True)
sqz_dec.replace(0, npNaN, inplace=True)
# Handle fills
if "fillna" in kwargs:
sqz_inc.fillna(kwargs["fillna"], inplace=True)
sqz_dec.fillna(kwargs["fillna"], inplace=True)
pos_inc.fillna(kwargs["fillna"], inplace=True)
pos_dec.fillna(kwargs["fillna"], inplace=True)
neg_dec.fillna(kwargs["fillna"], inplace=True)
neg_inc.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
sqz_inc.fillna(method=kwargs["fill_method"], inplace=True)
sqz_dec.fillna(method=kwargs["fill_method"], inplace=True)
pos_inc.fillna(method=kwargs["fill_method"], inplace=True)
pos_dec.fillna(method=kwargs["fill_method"], inplace=True)
neg_dec.fillna(method=kwargs["fill_method"], inplace=True)
neg_inc.fillna(method=kwargs["fill_method"], inplace=True)
df[f"SQZ_INC"] = sqz_inc
df[f"SQZ_DEC"] = sqz_dec
df[f"SQZ_PINC"] = pos_inc
df[f"SQZ_PDEC"] = pos_dec
df[f"SQZ_NDEC"] = neg_dec
df[f"SQZ_NINC"] = neg_inc
return df
squeeze.__doc__ = \
"""Squeeze (SQZ)
The default is based on John Carter's "TTM Squeeze" indicator, as discussed
in his book "Mastering the Trade" (chapter 11). The Squeeze indicator attempts
to capture the relationship between two studies: Bollinger Bands® and Keltner's
Channels. When the volatility increases, so does the distance between the bands,
conversely, when the volatility declines, the distance also decreases. It finds
sections of the Bollinger Bands® study which fall inside the Keltner's Channels.
Sources:
https://tradestation.tradingappstore.com/products/TTMSqueeze
https://www.tradingview.com/scripts/lazybear/
https://tlc.thinkorswim.com/center/reference/Tech-Indicators/studies-library/T-U/TTM-Squeeze
Calculation:
Default Inputs:
bb_length=20, bb_std=2, kc_length=20, kc_scalar=1.5, mom_length=12,
mom_smooth=12, tr=True, lazybear=False,
BB = Bollinger Bands
KC = Keltner Channels
MOM = Momentum
SMA = Simple Moving Average
EMA = Exponential Moving Average
TR = True Range
RANGE = TR(high, low, close) if using_tr else high - low
BB_LOW, BB_MID, BB_HIGH = BB(close, bb_length, std=bb_std)
KC_LOW, KC_MID, KC_HIGH = KC(high, low, close, kc_length, kc_scalar, TR)
if lazybear:
HH = high.rolling(kc_length).max()
LL = low.rolling(kc_length).min()
AVG = 0.25 * (HH + LL) + 0.5 * KC_MID
SQZ = linreg(close - AVG, kc_length)
else:
MOMO = MOM(close, mom_length)
if mamode == "ema":
SQZ = EMA(MOMO, mom_smooth)
else:
SQZ = EMA(momo, mom_smooth)
SQZ_ON = (BB_LOW > KC_LOW) and (BB_HIGH < KC_HIGH)
SQZ_OFF = (BB_LOW < KC_LOW) and (BB_HIGH > KC_HIGH)
NO_SQZ = !SQZ_ON and !SQZ_OFF
Args:
high (pd.Series): Series of 'high's
low (pd.Series): Series of 'low's
close (pd.Series): Series of 'close's
bb_length (int): Bollinger Bands period. Default: 20
bb_std (float): Bollinger Bands Std. Dev. Default: 2
kc_length (int): Keltner Channel period. Default: 20
kc_scalar (float): Keltner Channel scalar. Default: 1.5
mom_length (int): Momentum Period. Default: 12
mom_smooth (int): Smoothing Period of Momentum. Default: 6
mamode (str): Only "ema" or "sma". Default: "sma"
offset (int): How many periods to offset the result. Default: 0
Kwargs:
tr (value, optional): Use True Range for Keltner Channels. Default: True
asint (value, optional): Use integers instead of bool. Default: True
mamode (value, optional): Which MA to use. Default: "sma"
lazybear (value, optional): Use LazyBear's TradingView implementation.
Default: False
detailed (value, optional): Return additional variations of SQZ for
visualization. Default: False
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.DataFrame: SQZ, SQZ_ON, SQZ_OFF, NO_SQZ columns by default. More
detailed columns if 'detailed' kwarg is True.
"""
pandas_ta/momentum/willr.py
+10 -10
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
from ..utils import get_drift, get_offset, verify_series
from pandas_ta.utils import get_drift, get_offset, verify_series
def willr(high, low, close, length=None, offset=None, **kwargs):
"""Indicator: William's Percent R (WILLR)"""
@@ -8,7 +8,7 @@ def willr(high, low, close, length=None, offset=None, **kwargs):
low = verify_series(low)
close = verify_series(close)
length = int(length) if length and length > 0 else 14
min_periods = int(kwargs['min_periods']) if 'min_periods' in kwargs and kwargs['min_periods'] is not None else length
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
offset = get_offset(offset)
# Calculate Result
@@ -22,14 +22,14 @@ def willr(high, low, close, length=None, offset=None, **kwargs):
willr = willr.shift(offset)
# Handle fills
if 'fillna' in kwargs:
willr.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
willr.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
willr.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
willr.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
willr.name = f"WILLR_{length}"
willr.category = 'momentum'
willr.category = "momentum"
return willr
@@ -47,10 +47,10 @@ Sources:
Calculation:
Default Inputs:
length=20
lowest_low = low.rolling(length).min()
highest_high = high.rolling(length).max()
LL = low.rolling(length).min()
HH = high.rolling(length).max()
WILLR = 100 * ((close - lowest_low) / (highest_high - lowest_low) - 1)
WILLR = 100 * ((close - LL) / (HH - LL) - 1)
Args:
high (pd.Series): Series of 'high's
pandas_ta/performance/log_return.py
+6 -6
View File
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
from numpy import log as nplog
from ..utils import get_offset, verify_series
from pandas_ta.utils import get_offset, verify_series
def log_return(close, length=None, cumulative=False, offset=None, **kwargs):
"""Indicator: Log Return"""
@@ -20,14 +20,14 @@ def log_return(close, length=None, cumulative=False, offset=None, **kwargs):
log_return = log_return.shift(offset)
# Handle fills
if 'fillna' in kwargs:
log_return.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
log_return.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
log_return.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
log_return.fillna(method=kwargs["fill_method"], inplace=True)
# Name & Category
log_return.name = f"{'CUM' if cumulative else ''}LOGRET_{length}"
log_return.category = 'performance'
log_return.category = "performance"
return log_return
pandas_ta/performance/percent_return.py
+2 -2
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
from ..utils import get_offset, verify_series
from pandas_ta.utils import get_offset, verify_series
def percent_return(close, length=None, cumulative=False, offset=None, **kwargs):
"""Indicator: Percent Return"""
@@ -20,7 +20,7 @@ def percent_return(close, length=None, cumulative=False, offset=None, **kwargs):
# Name & Category
pct_return.name = f"{'CUM' if cumulative else ''}PCTRET_{length}"
pct_return.category = 'performance'
pct_return.category = "performance"
return pct_return
pandas_ta/performance/trend_return.py
+10 -6
View File
@@ -2,23 +2,27 @@
from pandas import Series
from .log_return import log_return
from .percent_return import percent_return
from ..utils import get_offset, verify_series, zero
from pandas_ta.utils import get_offset, verify_series, zero
def trend_return(close, trend, log=True, cumulative=None, offset=None, trend_reset=0, **kwargs):
def trend_return(close, trend, log=True, cumulative=None, trend_reset=0, offset=None, **kwargs):
"""Indicator: Trend Return"""
# Validate Arguments
close = verify_series(close)
trend = verify_series(trend)
offset = get_offset(offset)
cumulative = cumulative if cumulative is not None and isinstance(cumulative, bool) else False
trend_reset = int(trend_reset) if trend_reset and isinstance(trend_reset, int) else 0
offset = get_offset(offset)
# Calculate Result
returns = log_return(close, cumulative=False) if log else percent_return(close, cumulative=False)
tsum = 0
m = trend.size
if log:
returns = log_return(close, cumulative=False)
else:
returns = percent_return(close, cumulative=False)
trend = trend.astype(int)
returns = (trend * returns).apply(zero)
tsum = 0
m = trend.size
result = []
for i in range(0, m):
if trend[i] == trend_reset:
pandas_ta/trend/increasing.py
+6 -6
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
from ..utils import get_offset, verify_series
from pandas_ta.utils import get_offset, verify_series
def increasing(close, length=None, asint=True, offset=None, **kwargs):
"""Indicator: Increasing"""
@@ -18,14 +18,14 @@ def increasing(close, length=None, asint=True, offset=None, **kwargs):
increasing = increasing.shift(offset)
# Handle fills
if 'fillna' in kwargs:
increasing.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
increasing.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
increasing.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
increasing.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
increasing.name = f"INC_{length}"
increasing.category = 'trend'
increasing.category = "trend"
return increasing
pandas_ta/utils.py
+23 -34
View File
@@ -45,7 +45,7 @@ def _above_below(
# Name & Category
current.name = f"{series_a.name}_{'A' if above else 'B'}_{series_b.name}"
current.category = 'utility'
current.category = "utility"
return current
@@ -70,7 +70,7 @@ def above_value(
if not isinstance(value, (int, float, complex)):
print("[X] value is not a number")
return
series_b = pd.Series(value, index=series_a.index, name=f"{value}".replace('.','_'))
series_b = pd.Series(value, index=series_a.index, name=f"{value}".replace(".","_"))
return _above_below(series_a, series_b, above=True, asint=asint, offset=offset, **kwargs)
@@ -94,7 +94,7 @@ def below_value(
if not isinstance(value, (int, float, complex)):
print("[X] value is not a number")
return
series_b = pd.Series(value, index=series_a.index, name=f"{value}".replace('.','_'))
series_b = pd.Series(value, index=series_a.index, name=f"{value}".replace(".","_"))
return _above_below(series_a, series_b, above=False, asint=asint, offset=offset, **kwargs)
@@ -106,10 +106,10 @@ def category_files(category: str) -> list:
def combination(**kwargs):
"""https://stackoverflow.com/questions/4941753/is-there-a-math-ncr-function-in-python"""
n = int(math.fabs(kwargs.pop('n', 1)))
r = int(math.fabs(kwargs.pop('r', 0)))
n = int(math.fabs(kwargs.pop("n", 1)))
r = int(math.fabs(kwargs.pop("r", 0)))
if kwargs.pop('repetition', False) or kwargs.pop('multichoose', False):
if kwargs.pop("repetition", False) or kwargs.pop("multichoose", False):
n = n + r - 1
# if r < 0: return None
@@ -130,7 +130,7 @@ def cross_value(
offset: int = None,
**kwargs
):
series_b = pd.Series(value, index=series_a.index, name=f"{value}".replace('.','_'))
series_b = pd.Series(value, index=series_a.index, name=f"{value}".replace(".","_"))
return cross(series_a, series_b, above, asint, offset, **kwargs)
@@ -164,7 +164,7 @@ def cross(
# Name & Category
cross.name = f"{series_a.name}_{'XA' if above else 'XB'}_{series_b.name}"
cross.category = 'utility'
cross.category = "utility"
return cross
@@ -228,37 +228,26 @@ def signals(indicator, xa, xb, cross_values, xserie, xserie_a, xserie_b, cross_s
def df_error_analysis(dfA: pd.DataFrame, dfB: pd.DataFrame, **kwargs) -> pd.DataFrame:
""" """
col = kwargs.pop('col', None)
corr_method = kwargs.pop('corr_method', 'pearson')
col = kwargs.pop("col", None)
corr_method = kwargs.pop("corr_method", "pearson")
# Find their differences
# Find their differences and correlation
diff = dfA - dfB
df = pd.DataFrame({'diff': diff.describe()})
extra = pd.DataFrame(
[diff.var(), diff.mad(), diff.sem(), dfA.corr(dfB, method=corr_method)],
index=['var', 'mad', 'sem', 'corr']
)
# Append the differences to the DataFrame
df = df['diff'].append(extra, ignore_index=False)[0]
corr = dfA.corr(dfB, method=corr_method)
# For plotting
if kwargs.pop('plot', False):
if kwargs.pop("plot", False):
diff.hist()
if diff[diff > 0].any():
diff.plot(kind='kde')
if col is not None:
return df[col]
else:
return df
diff.plot(kind="kde")
return corr
def fibonacci(**kwargs) -> np.ndarray:
"""Fibonacci Sequence as a numpy array"""
n = int(math.fabs(kwargs.pop('n', 2)))
zero = kwargs.pop('zero', False)
weighted = kwargs.pop('weighted', False)
n = int(math.fabs(kwargs.pop("n", 2)))
zero = kwargs.pop("zero", False)
weighted = kwargs.pop("weighted", False)
if zero:
a, b = 0, 1
@@ -322,8 +311,8 @@ def pascals_triangle(n: int = None, **kwargs) -> np.ndarray:
=> inverse weighted: [0.9375, 0.75, 0.625, 0.75, 0.9375]
"""
n = int(math.fabs(n)) if n is not None else 0
weighted = kwargs.pop('weighted', False)
inverse = kwargs.pop('inverse', False)
weighted = kwargs.pop("weighted", False)
inverse = kwargs.pop("inverse", False)
# Calculation
triangle = np.array([combination(n=n, r=i) for i in range(0, n + 1)])
@@ -373,7 +362,7 @@ def symmetric_triangle(n: int = None, **kwargs) -> list:
=> weighted: [0.16666667 0.33333333 0.33333333 0.16666667]
"""
n = int(math.fabs(n)) if n is not None else 2
weighted = kwargs.pop('weighted', False)
weighted = kwargs.pop("weighted", False)
if n == 2:
triangle = [1, 1]
@@ -418,7 +407,7 @@ def unsigned_differences(series: pd.Series, amount: int = None, **kwargs) -> pd.
negative[negative >= 0] = 0
negative[negative < 0] = 1
if kwargs.pop('asint', False):
if kwargs.pop("asint", False):
positive = positive.astype(int)
negative = negative.astype(int)
@@ -440,7 +429,7 @@ def weights(w):
def zero(x: [int, float]) -> [int, float]:
"""If the value is close to zero, then return zero.
Otherwise return the value."""
return 0 if -sflt.epsilon < x and x < sflt.epsilon else x
return 0 if abs(x) < sflt.epsilon else x
# Candle Functions
pandas_ta/volatility/bbands.py
+19 -20
View File
@@ -1,27 +1,26 @@
# -*- coding: utf-8 -*-
from pandas import DataFrame
from ..overlap.ema import ema
from ..overlap.sma import sma
from ..statistics.stdev import stdev
from ..utils import get_offset, verify_series
from pandas_ta.overlap import ema, sma
from pandas_ta.statistics import stdev
from pandas_ta.utils import get_offset, verify_series
def bbands(close, length=None, std=None, mamode=None, offset=None, **kwargs):
"""Indicator: Bollinger Bands (BBANDS)"""
# Validate arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 5
min_periods = int(kwargs['min_periods']) if 'min_periods' in kwargs and kwargs['min_periods'] is not None else length
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
std = float(std) if std and std > 0 else 2.
mamode = mamode.lower() if mamode else 'sma'
mamode = mamode.lower() if mamode else "sma"
offset = get_offset(offset)
# Calculate Result
standard_deviation = stdev(close=close, length=length)
deviations = std * standard_deviation
if mamode is None or mamode == 'sma':
if mamode is None or mamode == "sma":
mid = sma(close=close, length=length)
elif mamode == 'ema':
elif mamode == "ema":
mid = ema(close=close, length=length, **kwargs)
lower = mid - deviations
@@ -34,26 +33,26 @@ def bbands(close, length=None, std=None, mamode=None, offset=None, **kwargs):
upper = upper.shift(offset)
# Handle fills
if 'fillna' in kwargs:
lower.fillna(kwargs['fillna'], inplace=True)
mid.fillna(kwargs['fillna'], inplace=True)
upper.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
lower.fillna(method=kwargs['fill_method'], inplace=True)
mid.fillna(method=kwargs['fill_method'], inplace=True)
upper.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
lower.fillna(kwargs["fillna"], inplace=True)
mid.fillna(kwargs["fillna"], inplace=True)
upper.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
lower.fillna(method=kwargs["fill_method"], inplace=True)
mid.fillna(method=kwargs["fill_method"], inplace=True)
upper.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
lower.name = f"BBL_{length}_{std}"
mid.name = f"BBM_{length}_{std}"
upper.name = f"BBU_{length}_{std}"
mid.category = upper.category = lower.category = 'volatility'
mid.category = upper.category = lower.category = "volatility"
# Prepare DataFrame to return
data = {lower.name: lower, mid.name: mid, upper.name: upper}
bbandsdf = DataFrame(data)
bbandsdf.name = f"BBANDS_{length}_{std}"
bbandsdf.category = 'volatility'
bbandsdf.category = "volatility"
return bbandsdf
@@ -74,7 +73,7 @@ Calculation:
SMA = Simple Moving Average
STDEV = Standard Deviation
stdev = STDEV(close, length)
if 'ema':
if "ema":
MID = EMA(close, length)
else:
MID = SMA(close, length)
@@ -86,7 +85,7 @@ Args:
close (pd.Series): Series of 'close's
length (int): The short period. Default: 20
std (int): The long period. Default: 2
mamode (str): Two options: None or 'ema'. Default: 'ema'
mamode (str): Two options: None or "ema". Default: "ema"
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/volatility/kc.py
+40 -32
View File
@@ -2,9 +2,9 @@
from numpy import sqrt as npsqrt
from pandas import DataFrame
from .atr import atr
from ..overlap.hlc3 import hlc3
from ..statistics.variance import variance
from ..utils import get_offset, non_zero_range, verify_series
from .true_range import true_range
from pandas_ta.overlap import ema, hlc3, sma
from pandas_ta.utils import get_offset, high_low_range, non_zero_range, verify_series
def kc(high, low, close, length=None, scalar=None, mamode=None, offset=None, **kwargs):
@@ -14,22 +14,26 @@ def kc(high, low, close, length=None, scalar=None, mamode=None, offset=None, **k
low = verify_series(low)
close = verify_series(close)
length = int(length) if length and length > 0 else 20
min_periods = int(kwargs['min_periods']) if 'min_periods' in kwargs and kwargs['min_periods'] is not None else length
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
scalar = float(scalar) if scalar and scalar > 0 else 2
use_tr = kwargs.pop("tr", True)
mamode = mamode.lower() if mamode else None
offset = get_offset(offset)
# Calculate Result
std = variance(close=close, length=length).apply(npsqrt)
if mamode == 'ema':
basis = close.ewm(span=length, min_periods=min_periods).mean()
band = atr(high=high, low=low, close=close)
if use_tr:
range_ = true_range(high, low, close)
else:
hl_range = non_zero_range(high, low)
typical_price = hlc3(high=high, low=low, close=close)
basis = typical_price.rolling(length, min_periods=min_periods).mean()
band = hl_range.rolling(length, min_periods=min_periods).mean()
range_ = high_low_range(high, low)
_mode = ""
if mamode == "sma":
basis = sma(close, length)
band = sma(range_, length=length)
_mode += "s"
elif mamode is None or mamode == "ema":
basis = ema(close, length=length)
band = ema(range_, length=length)
lower = basis - scalar * band
upper = basis + scalar * band
@@ -41,26 +45,27 @@ def kc(high, low, close, length=None, scalar=None, mamode=None, offset=None, **k
upper = upper.shift(offset)
# Handle fills
if 'fillna' in kwargs:
lower.fillna(kwargs['fillna'], inplace=True)
basis.fillna(kwargs['fillna'], inplace=True)
upper.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
lower.fillna(method=kwargs['fill_method'], inplace=True)
basis.fillna(method=kwargs['fill_method'], inplace=True)
upper.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
lower.fillna(kwargs["fillna"], inplace=True)
basis.fillna(kwargs["fillna"], inplace=True)
upper.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
lower.fillna(method=kwargs["fill_method"], inplace=True)
basis.fillna(method=kwargs["fill_method"], inplace=True)
upper.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
lower.name = f"KCL_{length}"
basis.name = f"KCB_{length}"
upper.name = f"KCU_{length}"
basis.category = upper.category = lower.category = 'volatility'
_props = f"{_mode if len(_mode) else ''}_{length}_{scalar}"
lower.name = f"KCL{_props}"
basis.name = f"KCB{_props}"
upper.name = f"KCU{_props}"
basis.category = upper.category = lower.category = "volatility"
# Prepare DataFrame to return
data = {lower.name: lower, basis.name: basis, upper.name: upper}
kcdf = DataFrame(data)
kcdf.name = f"KC_{length}"
kcdf.category = 'volatility'
kcdf.name = f"KC{_props}"
kcdf.category = basis.category
return kcdf
@@ -77,14 +82,17 @@ Sources:
Calculation:
Default Inputs:
length=20, scalar=2
length=20, scalar=2, mamode=None
ATR = Average True Range
EMA = Exponential Moving Average
SMA = Simple Moving Average
if 'ema':
BAND = ATR(high, low, close)
if mamode == "ema":
BASIS = EMA(close, length)
BAND = ATR(high, low, close)
else:
elif mamode == "sma":
BASIS = SMA(close, length)
else: # Typical Price
hl_range = high - low
tp = typical_price = hlc3(high, low, close)
BASIS = SMA(tp, length)
@@ -99,7 +107,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: None or 'ema'. Default: 'ema'
mamode (str): Two options: None or "ema". Default: "ema"
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/volume/aobv.py
+25 -30
View File
@@ -1,14 +1,9 @@
# -*- coding: utf-8 -*-
from pandas import DataFrame
from .obv import obv
from ..overlap.ema import ema
from ..overlap.hma import hma
from ..overlap.linreg import linreg
from ..overlap.sma import sma
from ..overlap.wma import wma
from ..trend.long_run import long_run
from ..trend.short_run import short_run
from ..utils import get_offset, verify_series
from pandas_ta.overlap import ema, hma, linreg, sma, wma
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):
"""Indicator: Archer On Balance Volume (AOBV)"""
@@ -16,31 +11,31 @@ def aobv(close, volume, fast=None, slow=None, mamode=None, max_lookback=None, mi
close = verify_series(close)
volume = verify_series(volume)
offset = get_offset(offset)
fast = int(fast) if fast and fast > 0 else 2
slow = int(slow) if slow and slow > 0 else 4
fast = int(fast) if fast and fast > 0 else 4
slow = int(slow) if slow and slow > 0 else 12
max_lookback = int(max_lookback) if max_lookback and max_lookback > 0 else 2
min_lookback = int(min_lookback) if min_lookback and min_lookback > 0 else 2
if slow < fast:
fast, slow = slow, fast
mamode = mamode.upper() if mamode else None
run_length = kwargs.pop('run_length', 2)
run_length = kwargs.pop("run_length", 2)
# Calculate Result
obv_ = obv(close=close, volume=volume, **kwargs)
if mamode is None or mamode == 'EMA':
mamode = 'EMA'
if mamode is None or mamode == "EMA":
mamode = "EMA"
maf = ema(close=obv_, length=fast, **kwargs)
mas = ema(close=obv_, length=slow, **kwargs)
elif mamode == 'HMA':
elif mamode == "HMA":
maf = hma(close=obv_, length=fast, **kwargs)
mas = hma(close=obv_, length=slow, **kwargs)
elif mamode == 'LINREG':
elif mamode == "LINREG":
maf = linreg(close=obv_, length=fast, **kwargs)
mas = linreg(close=obv_, length=slow, **kwargs)
elif mamode == 'SMA':
elif mamode == "SMA":
maf = sma(close=obv_, length=fast, **kwargs)
mas = sma(close=obv_, length=slow, **kwargs)
elif mamode == 'WMA':
elif mamode == "WMA":
maf = wma(close=obv_, length=fast, **kwargs)
mas = wma(close=obv_, length=slow, **kwargs)
@@ -57,18 +52,18 @@ def aobv(close, volume, fast=None, slow=None, mamode=None, max_lookback=None, mi
obv_short = obv_short.shift(offset)
# # Handle fills
if 'fillna' in kwargs:
obv_.fillna(kwargs['fillna'], inplace=True)
maf.fillna(kwargs['fillna'], inplace=True)
mas.fillna(kwargs['fillna'], inplace=True)
obv_long.fillna(kwargs['fillna'], inplace=True)
obv_short.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
obv_.fillna(method=kwargs['fill_method'], inplace=True)
maf.fillna(method=kwargs['fill_method'], inplace=True)
mas.fillna(method=kwargs['fill_method'], inplace=True)
obv_long.fillna(method=kwargs['fill_method'], inplace=True)
obv_short.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
obv_.fillna(kwargs["fillna"], inplace=True)
maf.fillna(kwargs["fillna"], inplace=True)
mas.fillna(kwargs["fillna"], inplace=True)
obv_long.fillna(kwargs["fillna"], inplace=True)
obv_short.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
obv_.fillna(method=kwargs["fill_method"], inplace=True)
maf.fillna(method=kwargs["fill_method"], inplace=True)
mas.fillna(method=kwargs["fill_method"], inplace=True)
obv_long.fillna(method=kwargs["fill_method"], inplace=True)
obv_short.fillna(method=kwargs["fill_method"], inplace=True)
# Prepare DataFrame to return
data = {
@@ -84,6 +79,6 @@ def aobv(close, volume, fast=None, slow=None, mamode=None, max_lookback=None, mi
# Name and Categorize it
aobvdf.name = f"AOBV_{mamode}_{fast}_{slow}_{min_lookback}_{max_lookback}_{run_length}"
aobvdf.category = 'volume'
aobvdf.category = "volume"
return aobvdf
pandas_ta/volume/efi.py
+11 -11
View File
@@ -1,5 +1,6 @@
# -*- coding: utf-8 -*-
from ..utils import get_drift, get_offset, verify_series
from pandas_ta.overlap import ema, sma
from pandas_ta.utils import get_drift, get_offset, verify_series
def efi(close, volume, length=None, drift=None, mamode=None, offset=None, **kwargs):
"""Indicator: Elder's Force Index (EFI)"""
@@ -7,7 +8,6 @@ def efi(close, volume, length=None, drift=None, mamode=None, offset=None, **kwar
close = verify_series(close)
volume = verify_series(volume)
length = int(length) if length and length > 0 else 13
min_periods = int(kwargs['min_periods']) if 'min_periods' in kwargs and kwargs['min_periods'] is not None else length
drift = get_drift(drift)
mamode = mamode.lower() if mamode else None
offset = get_offset(offset)
@@ -15,24 +15,24 @@ def efi(close, volume, length=None, drift=None, mamode=None, offset=None, **kwar
# Calculate Result
pv_diff = close.diff(drift) * volume
if mamode == 'sma':
efi = pv_diff.rolling(length, min_periods=min_periods).mean()
if mamode == "sma":
efi = sma(pv_diff, length)
else:
efi = pv_diff.ewm(span=length, min_periods=min_periods).mean()
efi = ema(pv_diff, length)
# Offset
if offset != 0:
efi = efi.shift(offset)
# Handle fills
if 'fillna' in kwargs:
efi.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
efi.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
efi.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
efi.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
efi.name = f"EFI_{length}"
efi.category = 'volume'
efi.category = "volume"
return efi
@@ -65,7 +65,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): Two options: None or "sma". Default: None
offset (int): How many periods to offset the result. Default: 0
Kwargs:
pandas_ta/volume/eom.py
+9 -10
View File
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
from ..overlap.hl2 import hl2
from ..utils import get_drift, get_offset, non_zero_range, verify_series
from pandas_ta.overlap import hl2, sma
from pandas_ta.utils import get_drift, get_offset, non_zero_range, verify_series
def eom(high, low, close, volume, length=None, divisor=None, drift=None, offset=None, **kwargs):
"""Indicator: Ease of Movement (EOM)"""
@@ -9,33 +9,32 @@ def eom(high, low, close, volume, length=None, divisor=None, drift=None, offset=
low = verify_series(low)
close = verify_series(close)
volume = verify_series(volume)
high_low_range = non_zero_range(high, low)
length = int(length) if length and length > 0 else 14
min_periods = int(kwargs['min_periods']) if 'min_periods' in kwargs and kwargs['min_periods'] is not None else length
divisor = divisor if divisor and divisor > 0 else 100000000
drift = get_drift(drift)
offset = get_offset(offset)
# Calculate Result
high_low_range = non_zero_range(high, low)
distance = hl2(high=high, low=low) - hl2(high=high.shift(drift), low=low.shift(drift))
box_ratio = volume / divisor
box_ratio /= high_low_range
eom = distance / box_ratio
eom = eom.rolling(length, min_periods=min_periods).mean()
eom = sma(eom, length=length)
# Offset
if offset != 0:
eom = eom.shift(offset)
# Handle fills
if 'fillna' in kwargs:
eom.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
eom.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
eom.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
eom.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
eom.name = f"EOM_{length}_{divisor}"
eom.category = 'volume'
eom.category = "volume"
return eom
tests/test_indicator_momentum.py
+17
View File
@@ -306,6 +306,23 @@ class TestMomentum(TestCase):
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "ANGLEd_1")
def test_squeeze(self):
result = pandas_ta.squeeze(self.high, self.low, self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "SQZ_20_2.0_20_1.5")
result = pandas_ta.squeeze(self.high, self.low, self.close, tr=False)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "SQZhlr_20_2.0_20_1.5")
result = pandas_ta.squeeze(self.high, self.low, self.close, lazybear=True)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "SQZ_20_2.0_20_1.5_LB")
result = pandas_ta.squeeze(self.high, self.low, self.close, tr=False, lazybear=True)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "SQZhlr_20_2.0_20_1.5_LB")
def test_stoch(self):
result = pandas_ta.stoch(self.high, self.low, self.close, fast_k=14, slow_k=14, slow_d=14)
self.assertIsInstance(result, DataFrame)
tests/test_indicator_momentum_ext.py
+9
View File
@@ -166,6 +166,15 @@ class TestMomentumExtension(TestCase):
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "ANGLEd_1")
def test_squeeze_ext(self):
self.data.ta.squeeze(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-4:]), ["SQZ_20_2.0_20_1.5", "SQZ_ON", "SQZ_OFF", "SQZ_NO"])
self.data.ta.squeeze(tr=False, append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-4:]), ["SQZ_ON", "SQZ_OFF", "SQZ_NO", "SQZhlr_20_2.0_20_1.5"])
def test_stoch_ext(self):
self.data.ta.stoch(append=True)
self.assertIsInstance(self.data, DataFrame)
tests/test_indicator_volatility.py
+5 -1
View File
@@ -100,7 +100,11 @@ class TestVolatility(TestCase):
def test_kc(self):
result = pandas_ta.kc(self.high, self.low, self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "KC_20")
self.assertEqual(result.name, "KC_20_2")
result = pandas_ta.kc(self.high, self.low, self.close, mamode="sma")
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "KCs_20_2")
def test_massi(self):
result = pandas_ta.massi(self.high, self.low)
tests/test_indicator_volatility_ext.py
+1 -1
View File
@@ -48,7 +48,7 @@ class TestVolatilityExtension(TestCase):
def test_kc_ext(self):
self.data.ta.kc(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-3:]), ["KCL_20", "KCB_20", "KCU_20"])
self.assertEqual(list(self.data.columns[-3:]), ["KCL_20_2", "KCB_20_2", "KCU_20_2"])
def test_massi_ext(self):
self.data.ta.massi(append=True)
tests/test_indicator_volume.py
+1 -1
View File
@@ -72,7 +72,7 @@ class TestVolume(TestCase):
def test_aobv(self):
result = pandas_ta.aobv(self.close, self.volume_)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "AOBV_EMA_2_4_2_2_2")
self.assertEqual(result.name, "AOBV_EMA_4_12_2_2_2")
def test_cmf(self):
result = pandas_ta.cmf(self.high, self.low, self.close, self.volume_)
tests/test_indicator_volume_ext.py
+1 -1
View File
@@ -40,7 +40,7 @@ class TestVolumeExtension(TestCase):
def test_aobv_ext(self):
self.data.ta.aobv(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-7:]), ["OBV", "OBV_min_2", "OBV_max_2", "OBV_EMA_2", "OBV_EMA_4", "AOBV_LR_2", "AOBV_SR_2"])
self.assertEqual(list(self.data.columns[-7:]), ["OBV", "OBV_min_2", "OBV_max_2", "OBV_EMA_4", "OBV_EMA_12", "AOBV_LR_2", "AOBV_SR_2"])
# Remove "OBV" so it does not interfere with test_obv_ext()
self.data.drop("OBV", axis=1, inplace=True)