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

MAINT BBANDS RMA tests ENH Strategy + Watchlist Class + New Notebook

Browse Source
This commit is contained in:
Kevin Johnson
2020-07-21 10:49:33 -07:00
parent 11730dad3f
commit b356fca04d
26 changed files with 3827 additions and 434 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3
View File
@@ -119,10 +119,13 @@ pandas_ta/_wrapper.py
data/datas.csv
data/SPY_5min.csv
data/SPY_1min.csv
data/similang-ch.csv
data/tulip.csv
examples/taplot.py
examples/charting.ipynb
examples/ib_trader.ipynb
examples/example2.ipynb
examples/*.csv
setup.cfg
note.md
driver.py
README.md
+87 -46
View File
@@ -16,7 +16,7 @@ All the indicators return a named Series or a DataFrame in uppercase underscore
* Has 100+ indicators and utility functions.
* Option to use __multiprocessing__ when using df.ta.strategy(). See below.
* Example Jupyter Notebook under the [examples](https://github.com/twopirllc/pandas-ta/tree/master/examples) directory.
* 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.
* __Extended Pandas DataFrame__ as 'ta'.
@@ -25,50 +25,12 @@ All the indicators return a named Series or a DataFrame in uppercase underscore
## __Recent Changes__
* A __Strategy__ Class to help name and group your favorite indicators.
* An experimental and independent __Watchlist__ Class located in the [Examples](https://github.com/twopirllc/pandas-ta/tree/master/examples/watchlist.py) Directory that can be used in conjunction with the new __Strategy__ Class.
* 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 DataFrame Method:__
strategy (strategy)
### __Added indicators:__
Bias (bias)
Choppiness Index (chop)
Chande Kroll Stop (cksp)
Doji (cdl_doji)
Entropy (entropy)
Heikin-Ashi Candles (ha)
Inertia (inertia)
KDJ (kdj)
Parabolic Stop and Reverse (psar)
Price Distance (pdist)
Psycholigical Line (psl)
Percentage Volume Oscillator (pvo)
Relative Volatility Index (rvi)
Supertrend (supertrend)
Weighted Closing Price (wcp)
### __Added utilities:__
Above (above)
Above Value (above_value)
Below (below)
Below Value (below_value)
Cross Value (cross_value)
### __User Added Indicators:__
Aberration (aberration)
BRAR (brar)
### __Corrected Indicators:__
Absolute Price Oscillator (apo)
Aroon & Aroon Oscillator (aroon)
* Fixed indicator and included oscillator in returned dataframe
Bollinger Bands (bbands)
Commodity Channel Index (cci)
Chande Momentum Oscillator (cmo)
Exponential Moving Average (ema)
Moving Average Convergence Divergence (macd)
Relative Vigor Index (rvgi)
Symmetric Weighted Moving Average (swma)
Weighted Moving Average (wma)
## What is a Pandas DataFrame Extension?
@@ -127,9 +89,63 @@ pd.DataFrame().ta.indicators()
help(ta.log_return)
```
## __New DataFrame Method__: _strategy_ with Multiprocessing
## New Class: __Strategy__
### What is a Pandas TA Strategy?
A _Strategy_ is a simple way to name and group your favorite TA indicators. Technically, a _Strategy_ is a simple Data Class to contain list of indicators and their parameters. __Note__: _Strategy_ is experimental and subject to change. Pandas TA comes with two basic Strategies: __AllStrategy__ and __CommonStrategy__.
Strategy is a new __Pandas (TA)__ method to facilitate bulk indicator processing. By default, running ```df.ta.strategy()``` will append __all
* See the [Pandas TA Strategy Examples](https://github.com/twopirllc/pandas-ta/tree/master/examples/PandasTA_Strategy_Examples.ipynb) Notebook for more Examples including _Indicator Composition/Chaining_.
### Strategy Requirements:
- _name_: Some short memorable string. _Note_: Case-insensitive "All" is reserved.
- _ta_: A list of dicts containing keyword arguments to identify the indicator and the indicator's arguments
### Optional Requirements:
- _description_: A more detailed description of what the Strategy tries to capture. Default: None
- _created_: At datetime string of when it was created. Default: Automatically generated.
#### Things to note:
- A Strategy will __fail__ when consumed by Pandas TA if there is no {"kind": "indicator name"} attribute. __Remember__ to check your spelling.
#### Brief Examples
```python
# Builtin All Default Strategy
AllStrategy = Strategy(
name="All",
description="All the indicators with their default settings. Pandas TA default.",
ta=None
)
# Builtin Default (Example) Strategy.
CommonStrategy = Strategy(
name="Common Price and Volume SMAs",
description="Common Price SMAs: 10, 20, 50, 200 and Volume SMA: 20.",
ta=[
{"kind": "sma", "length": 10},
{"kind": "sma", "length": 20},
{"kind": "sma", "length": 50},
{"kind": "sma", "length": 200},
{"kind": "sma", "close": "volume", "length": 20, "prefix": "VOL"}
]
)
# Your Custom Strategy or whatever your TA composition
CustomStrategy = Strategy(
name="Momo and Volatility",
description="SMA 50,200, BBANDS, RSI, MACD and Volume SMA 20",
ta=[
{"kind": "sma", "length": 50},
{"kind": "sma", "length": 200},
{"kind": "bbands", "length": 20},
{"kind": "rsi"},
{"kind": "macd", "fast": 8, "slow": 21},
{"kind": "sma", "close": "volume", "length": 20, "prefix": "VOLUME"},
]
)
```
## __DataFrame Method__: _strategy_ with Multiprocessing
The new __Pandas (TA)__ method __strategy__ is used to facilitate bulk indicator processing. By default, running ```df.ta.strategy()``` will append __all
applicable__ indicators to DataFrame ```df```. Utility methods like ```above```, ```below``` et al are not included.
* The ```ta.strategy()``` method is still __under development__. Future iterations will allow you to load a ```ta.json``` config file with your specific strategy name and parameters to automatically run you bulk indicators.
@@ -171,7 +187,32 @@ df.ta.strategy(fast=10, slow=50, verbose=True)
df.columns
```
## __New DataFrame kwargs__: _prefix_ and _suffix_
### Running a Custom Strategy
While the _Strategy_ Class it has not been fully integrated with the __strategy__ method yet. For now, the following can be done to implement your Custom Strategy.
```python
# Create a Strategy
CustomStrategy = Strategy(
name="Momo and Volatility",
description="SMA 50,200, BBANDS, RSI, MACD and Volume SMA 20",
ta=[
{"kind": "sma", "length": 50},
{"kind": "sma", "length": 200},
{"kind": "bbands", "length": 20},
{"kind": "rsi"},
{"kind": "macd", "fast": 8, "slow": 21},
{"kind": "sma", "close": "volume", "length": 20, "prefix": "VOLUME"},
]
)
#Running it requires the name and ta properties
df.ta.strategy(name=CustomStrategy.name, ta=CustomStrategy.ta)
# Sanity check. Make sure all the columns are there
df.columns
```
## __DataFrame kwargs__: _prefix_ and _suffix_
```python
prehl2 = df.ta.hl2(prefix="pre")
@@ -184,7 +225,7 @@ bothhl2 = df.ta.hl2(prefix="pre", suffix="post")
print(bothhl2.name) # "pre_HL2_post"
```
## __New DataFrame Properties__: _reverse_ & _datetime_ordered_
## __DataFrame Properties__: _reverse_ & _datetime_ordered_
```python
# The 'reverse' is a helper property that returns the DataFrame
@@ -193,7 +234,7 @@ df = df.ta.reverse
# The 'datetime_ordered' property returns True if the DataFrame
# index is of Pandas datetime64 and df.index[0] < df.index[-1]
# Otherwise it return False
# Otherwise it returns False
time_series_in_order = df.ta.datetime_ordered
```
examples/PandasTA_Strategy_Examples.ipynb
+3040
View File
File diff suppressed because it is too large Load Diff
examples/watchlist.py
+202
View File
@@ -0,0 +1,202 @@
# -*- coding: utf-8 -*-
from functools import lru_cache
from pathlib import Path
from random import random
import pandas as pd
from alphaVantageAPI.alphavantage import AlphaVantage # pip install alphaVantage-api
import pandas_ta as ta
class Watchlist(object):
"""Watchlist Class (** This is subject to change! **)
============================================================================
A simple Class to load/download financial market data and automatically
apply Technical Analysis indicators with a Pandas TA Strategy. Default
Strategy: pandas_ta.AllStrategy.
Requirements:
- Pandas TA (pip install pandas_ta)
- AlphaVantage (pip install alphaVantage-api) for the Default Data Source.
To use another Data Source, update the load() method after AV.
Required Arguments:
- tickers: A list of strings containing tickers. Example: ['SPY', 'AAPL']
============================================================================
"""
def __init__(
self,
tickers: list,
tf: str = None,
name: str = None,
strategy: ta.Strategy = None,
ds: object = None,
**kwargs
):
self.tickers = tickers
self.tf = tf
self.verbose = kwargs.pop("verbose", False)
self.name = name
self.data = None
self.kwargs = kwargs
self.ds = ds if ds is not None else None
self.strategy = strategy
def _drop_columns(self, df: pd.DataFrame, cols: list = ['Unnamed: 0', 'date', 'split_coefficient', 'dividend']):
"""Helper methods to drop columns silently."""
df_columns = list(df.columns)
if any(_ in df_columns for _ in cols):
if self.verbose:
print(f"[i] Possible columns dropped: {', '.join(cols)}")
df = df.drop(cols, axis=1, errors='ignore')
return df
def _load_all(self, **kwargs) -> dict:
"""Updates the Watchlist's data property with a dictionary of DataFrames
keyed by ticker."""
if self.tickers is not None and isinstance(self.tickers, list) and len(self.tickers):
self.data = {ticker: self.load(ticker, **kwargs) for ticker in self.tickers}
return self.data
def load(
self,
ticker: str = None,
tf: str = None,
index: str = 'date',
drop: list = ['dividend', 'split_coefficient'],
file_path: str = ".",
**kwargs
) -> pd.DataFrame:
"""Loads or Downloads (if a local csv does not exist) the data from the
Data Source. When successful, it returns a Data Frame for the requested
ticker. If no tickers are given, it loads all the tickers."""
tf = self.tf if tf is None else tf.upper()
if ticker is not None and isinstance(ticker, str):
ticker = str(ticker).upper()
else:
print(f"[!] Loading All: {', '.join(self.tickers)}")
self._load_all(**kwargs)
return
filename_ = f"{ticker}_{tf}.csv"
current_file = Path(file_path) / filename_
# Load local or from Data Source
if current_file.exists():
df = pd.read_csv(filename_, index_col=index)
if not df.ta.datetime_ordered:
df = df.set_index(pd.DatetimeIndex(df.index))
print(f"\n[i] Loaded['{tf}']: {filename_}")
else:
if self.ds is not None and isinstance(self.ds, AlphaVantage):
df = self.ds.data(tf, ticker)
if not df.ta.datetime_ordered:
df = df.set_index(pd.DatetimeIndex(df[index]))
print(f"\n[+] Downloading['{tf}']: {ticker}")
df = self._drop_columns(df) # Remove select columns
if kwargs.pop("analyze", True):
df.ta.strategy(name=self.strategy.name, ta=self.strategy.ta, **kwargs)
df.ticker = ticker # Attach ticker to the DataFrame
return df
@property
def data(self) -> dict:
"""When not None, it contains a dictionary of DataFrames keyed by ticker."""
return self._data
@data.setter
def data(self, value: dict) -> None:
# Later check dict has string keys and DataFrame values
if value is not None and isinstance(value, dict):
if self.verbose: print(f"[+] New data")
self._data = value
else:
self._data = None
@property
def name(self) -> str:
"""The name of the Watchlist. Default: "Watchlist: {Watchlist.tickers}"."""
return self._name
@name.setter
def name(self, value: str) -> None:
if isinstance(value, str):
self._name = str(value)
else:
self._name = f"Watchlist: {', '.join(self.tickers)}"
@property
def strategy(self) -> ta.Strategy:
"""Pandas TA Strategy Class. Default: pandas_ta.AllStrategy"""
return self._strategy
@strategy.setter
def strategy(self, value: ta.Strategy) -> None:
if value is not None and isinstance(value, ta.Strategy):
self._strategy = value
else:
self._strategy = ta.AllStrategy
@property
def tf(self) -> str:
"""Alias for timeframe. Default: 'D'"""
return self._tf
@tf.setter
def tf(self, value: str) -> None:
if isinstance(value, str):
value = str(value)
self._tf = value
else:
self._tf = "D"
@property
def tickers(self) -> list:
"""tickers
If a string, it it converted to a list. Example: 'AAPL' -> ['AAPL']
* Does not accept, comma seperated strings.
If a list, checks if it is a list of strings.
"""
return self._tickers
@tickers.setter
def tickers(self, value: (list, str)) -> None:
if value is None:
print(f"[X] {value} is not a valie Watchlist ticker.")
return
elif isinstance(value, list) and [isinstance(_, str) for _ in value]:
self._tickers = list(map(str.upper, value))
elif isinstance(value, str):
self._tickers = [value.upper()]
self.name = self._tickers
@property
def verbose(self) -> bool:
"""Toggle the verbose property. Default: False"""
return self._verbose
@verbose.setter
def verbose(self, value: bool) -> None:
if isinstance(value, bool):
self._verbose = bool(value)
else:
self._verbose = False
def indicators(self, *args, **kwargs) -> any:
"""Returns the list of indicators that are available with Pandas Ta."""
pd.DataFrame().ta.indicators(*args, **kwargs)
def __repr__(self) -> str:
s = f"Watch(name='{self.name}', tickers[{len(self.tickers)}]='{', '.join(self.tickers)}', tf='{self.tf}', strategy[{self.strategy.total_ta()}]='{self.strategy.name}'"
if self.data is not None:
s += f", data[{len(self.data.keys())}])"
return s
return s + ")"
pandas_ta/core.py
+188 -81
View File
@@ -1,8 +1,11 @@
# -*- coding: utf-8 -*-
from dataclasses import dataclass, field
from datetime import datetime
from functools import wraps
from multiprocessing import cpu_count, Pool
from random import random
from time import perf_counter
from typing import List
import pandas as pd
from pandas.core.base import PandasObject
@@ -17,7 +20,7 @@ from pandas_ta.volatility import *
from pandas_ta.volume import *
from pandas_ta.utils import *
version = ".".join(("0", "1", "75b"))
version = ".".join(("0", "1", "76b"))
def mp_worker(args):
df, method, kwargs = args
@@ -40,11 +43,91 @@ def finalize(method):
return _wrapper
@dataclass
class Strategy:
"""Strategy (Data)Class
A way to name and group your favorite indicators
Args:
name (str): Some short memorable string. Note: Case-insensitive "All" is reserved.
ta (list of dicts): A list of dicts containing keyword arguments where "kind" is the indicator.
description (str): A more detailed description of what the Strategy tries to capture. Default: None
created (str): At datetime string of when it was created. Default: Automatically generated. *Subject to change*
Example TA:
ta = [
{"kind": "sma", "length": 200},
{"kind": "sma", "close": "volume", "length": 50},
{"kind": "bbands", "length": 20},
{"kind": "rsi"},
{"kind": "macd", "fast": 8, "slow": 21},
{"kind": "sma", "close": "volume", "length": 20, "prefix": "VOLUME"},
]
"""
name: str# = None # Required.
ta: List = field(default_factory=list) # Required.
description: str = None # Helpful. More descriptive version or notes or w/e.
created: str = datetime.now().strftime("%m/%d/%Y, %H:%M:%S") # Optional. May change type later to datetime
last_run: str = None # Auto filled
run_time: str = None # Auto filled
def __post_init__(self):
has_name = True
is_ta = False
required_args = ["[X] Strategy requires the following argument(s):"]
name_is_str = isinstance(self.name, str)
ta_is_list = isinstance(self.ta, list)
if self.name is None or not name_is_str:
required_args.append(" - name. Must be a string. Example: \"My TA\". Note: \"all\" is reserved.")
has_name != has_name
if self.ta is None:
self.ta = None
elif self.ta is not None and ta_is_list and self.total_ta() > 0:
# Check that all elements of the list are dicts.
# Does not check if the dicts values are valid indicator kwargs
# User must check indicator documentation for all indicators args.
is_ta = all([isinstance(_, dict) and len(_.keys()) > 0 for _ in self.ta])
else:
s = " - ta. Format is a list of dicts. Example: [{'kind': 'sma', 'length': 10}]"
s += "\n Check the indicator for the correct arguments if you receive this error."
required_args.append(s)
if len(required_args) > 1:
[print(_) for _ in required_args]
return None
def total_ta(self):
return len(self.ta) if self.ta is not None else 0
# All Default Strategy
AllStrategy = Strategy(
name="All",
description="All the indicators with their default settings. Pandas TA default.",
ta=None
)
# Default (Example) Strategy.
CommonStrategy = Strategy(
name="Common Price and Volume SMAs",
description="Common Price SMAs: 10, 20, 50, 200 and Volume SMA: 20.",
ta=[
{"kind": "sma", "length": 10},
{"kind": "sma", "length": 20},
{"kind": "sma", "length": 50},
{"kind": "sma", "length": 200},
{"kind": "sma", "close": "volume", "length": 20, "prefix": "VOL"}
]
)
class BasePandasObject(PandasObject):
"""Simple PandasObject Extension
Ensures the DataFrame is not empty and has columns. It would be a
sad Panda otherwise.
Ensures the DataFrame is not empty and has columns.
It would be a sad Panda otherwise.
Args:
df (pd.DataFrame): Extends Pandas DataFrame
@@ -140,30 +223,37 @@ class AnalysisIndicators(BasePandasObject):
_adjusted = None
_mp = False
def __call__(self, kind=None, alias=None, timed=False, verbose=False, **kwargs):
try:
if isinstance(kind, str):
kind = kind.lower()
fn = getattr(self, kind)
def __call__(
self,
kind: str= None,
alias: str = None,
timed = False,
verbose = False,
**kwargs
):
try:
if isinstance(kind, str):
kind = kind.lower()
fn = getattr(self, kind)
if timed: stime = perf_counter()
if timed: stime = perf_counter()
# Run the indicator
result = fn(**kwargs) # = getattr(self, kind)(**kwargs)
# Run the indicator
result = fn(**kwargs) # = getattr(self, kind)(**kwargs)
# Add an alias if passed
if alias: result.alias = f"{alias}"
# Add an alias if passed
if alias: result.alias = f"{alias}"
if timed:
result.timed = final_time(stime)
print(f"[+] {kind}:{alias + ':' if alias is not None else ''} {result.timed}")
if timed:
result.timed = final_time(stime)
alias_str = alias + ':' if alias is not None else ''
print(f"[+] {kind}:{alias_str} {result.timed}")
return result
else:
self.help()
except: pass
return result
else:
self.help()
except: pass
@property
def adjusted(self) -> str:
@@ -257,11 +347,11 @@ class AnalysisIndicators(BasePandasObject):
match = [i for i, x in enumerate(matches) if x]
# If found, awesome. Return it or return the 'series'.
cols = ', '.join(list(df.columns))
NOT_FOUND = f" [X] Ooops!!!: It's {series not in df.columns}, the series '{series}' not in {cols}"
NOT_FOUND = f"[X] Ooops!!!: It's {series not in df.columns}, the series '{series}' was not found in {cols}"
return df.iloc[:,match[0]] if len(match) else print(NOT_FOUND)
def constants(self, append, lower_bound=-100, upper_bound=100, every=1):
def constants(self, append, lower_bound=-100, upper_bound=100, every=10):
"""Constants
Useful for creating indicator levels or if you need some constant value
@@ -332,58 +422,7 @@ class AnalysisIndicators(BasePandasObject):
s = f"{header}\nTotal Indicators: {total_indicators}\n"
print(f"{s}Abbreviations:\n {', '.join(ta_indicators)}") if total_indicators > 0 else print(s)
# ALL Features
def _all(self, **kwargs):
"""Appends by default all non-excluded indicators to the DataFrame. Used by ta.strategy(**kwargs)"""
cpus = cpu_count()
cores = int(kwargs.pop("cores", cpus))
timed = kwargs.pop("timed", False)
verbose = kwargs.pop("verbose", False)
user_excluded = kwargs.pop("exclude", [])
append = kwargs.setdefault("append", True)
excluded = ["above", "above_value", "below", "below_value",
"cross", "cross_value", "long_run", "short_run", "trend_return", "vp"]
excluded += user_excluded
current_columns = len(self._df.columns)
indicators = self.indicators(as_list=True, exclude=excluded)
print('[+] Strategy "All"')
if verbose:
print(f'[i] Indicators with the following arguments: {kwargs}')
print(f"[i] excluded[{len(excluded)}]: {', '.join(excluded)}")
if timed: stime = perf_counter()
if not self.mp:
# Display multiprocessing tip 10% of the time.
if random() < 0.1:
print(f"[i] Set 'df.ta.mp = True' to enable multiprocessing. This computer has {cpus} cores. Default: False")
methods = [getattr(self, kind) for kind in indicators]
[f(**kwargs) for f in methods]
else:
print(f"[i] multiprocessing: {cores} of {cpu_count()} cores")
pool = Pool(cores)
result = pool.imap_unordered(
mp_worker, ((self._df, ind, kwargs) for ind in indicators), cores
)
pool.close()
pool.join()
# Apply prefixes/suffixes and append to the DataFrame
for r in result:
self._add_prefix_suffix(r, **kwargs)
self._append(r, **kwargs)
print(f"[i] total indicators: {len(indicators)}, columns added: {len(self._df.columns) - current_columns}")
print(f"[i] runtime: {final_time(stime)}\n") if timed else None
def strategy(self, **kwargs):
def strategy(self, *args, **kwargs):
"""Strategy Method
An experimental method that by default runs all applicable indicators.
@@ -393,17 +432,85 @@ class AnalysisIndicators(BasePandasObject):
Args:
name (str, optional): Default: 'all'
exclude (list, optional): Default: []. List of indicator names to exclude.
verbose (bool): Default: False
kwargs:
(optional) Default: {}. Any indicator argument you want to modify.
For example, length=20 or offset=-1 or high=df['High'] ...
"""
name = kwargs.pop("name", "all")
if name is None or name == "" or not isinstance(name, str): # Extra check
name = "all"
self._all(**kwargs) if name == "all" else None
cpus = cpu_count()
name = kwargs.pop("name", None)
if name is None or name.lower() == "all":
name = "All"
print(f"strat.kwargs: {kwargs}")
# removing before sending the rest of kwargs to the indicators
ta = kwargs.pop("ta", None)
mp = kwargs.pop("mp", False)
cores = int(kwargs.pop("cores", cpus))
timed = kwargs.pop("timed", False)
verbose = kwargs.pop("verbose", False)
user_excluded = kwargs.pop("exclude", [])
kwargs["append"] = True
is_all = True if name is None or name.lower() == "all" else False
has_ta = True if ta is not None else False
initial_column_count = len(self._df.columns)
excluded = []
excluded += user_excluded # Exclude user excluded ta if listed
print(f'[+] Strategy "{name}"') if verbose else None
if is_all:
# Exclude utilities special functions
excluded += ["above", "above_value", "below", "below_value", "cross", "cross_value", "long_run", "short_run", "trend_return", "vp"]
ta = self.indicators(as_list=True, exclude=excluded)
else:
for kwds in ta:
kwds["append"] = True
if verbose:
print(f'[i] Indicators with the following arguments: {kwargs}')
if len(excluded) > 0:
print(f"[i] Excluded[{len(excluded)}]: {', '.join(excluded)}")
# Enable multiprocessing if user sets: mp=True
if mp: self.mp = not self.mp
if self.mp:
# TODO: Fix for Custom Strategies
print(f"[i] Multiprocessing: {cores} of {cpu_count()} cores")
pool = Pool(cores)
if timed: stime = perf_counter()
result = pool.imap_unordered(
mp_worker, ((self._df, ind, kwargs) for ind in ta), cores
)
pool.close()
pool.join()
# Apply prefixes/suffixes and appends indicator result to the DataFrame
for r in result:
self._add_prefix_suffix(r, **kwargs)
self._append(r, **kwargs)
if timed: ftime = final_time(stime)
else:
# Display multiprocessing tip 10% of the time.
if random() < 0.1:
print(f"[i] Set 'df.ta.mp = True' to enable multiprocessing. This computer has {cpus} cores. Default: False")
if timed: stime = perf_counter()
if is_all:
indicators = [getattr(self, kind) for kind in ta]
[f(**kwargs) for f in indicators]
else:
[getattr(self, kwds["kind"])(**kwds) for kwds in ta]
if timed: ftime = final_time(stime)
if verbose:
print(f"[i] Total indicators: {len(ta)}")
print(f"[i] Columns added: {len(self._df.columns) - initial_column_count}")
print(f"[i] Runtime: {ftime}") if timed else None
# Candles
pandas_ta/overlap/rma.py
+1 -2
View File
@@ -6,12 +6,11 @@ def rma(close, length=None, offset=None, **kwargs):
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 10
min_periods = int(kwargs['min_periods']) if 'min_periods' in kwargs and kwargs['min_periods'] is not None else length
offset = get_offset(offset)
alpha = (1.0 / length) if length > 0 else 0.5
# Calculate Result
rma = close.ewm(alpha=alpha, min_periods=min_periods).mean()
rma = close.ewm(alpha=alpha).mean()
# Offset
if offset != 0:
pandas_ta/utils.py
+2 -1
View File
@@ -1,5 +1,6 @@
# -*- coding: utf-8 -*-
import math
from pathlib import Path
from time import perf_counter
import numpy as np
@@ -9,7 +10,7 @@ from functools import reduce
from operator import mul
from sys import float_info as sflt
TRADING_DAYS_PER_YEAR = 250
TRADING_DAYS_PER_YEAR = 251
TRADING_HOURS_PER_DAY = 6.5
MINUTES_PER_HOUR = 60
pandas_ta/volatility/bbands.py
+4 -4
View File
@@ -44,15 +44,15 @@ def bbands(close, length=None, std=None, mamode=None, offset=None, **kwargs):
upper.fillna(method=kwargs['fill_method'], inplace=True)
# Name and Categorize it
lower.name = f"BBL_{length}"
mid.name = f"BBM_{length}"
upper.name = f"BBU_{length}"
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'
# Prepare DataFrame to return
data = {lower.name: lower, mid.name: mid, upper.name: upper}
bbandsdf = DataFrame(data)
bbandsdf.name = f"BBANDS_{length}"
bbandsdf.name = f"BBANDS_{length}_{std}"
bbandsdf.category = 'volatility'
return bbandsdf
tests/config.py
+1 -1
View File
@@ -6,7 +6,7 @@ VERBOSE = True
ALERT = f"[!]"
INFO = f"[i]"
CORRELATION = 'corr' #'sem'
CORRELATION = "corr" #"sem"
CORRELATION_THRESHOLD = 0.99 # Less than 0.99 is undesirable
sample_data = read_csv(
tests/context.py
+1 -1
View File
@@ -1,5 +1,5 @@
import os
import sys
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
import pandas_ta
tests/test_indicator_candle.py
+6 -6
View File
@@ -14,11 +14,11 @@ class TestCandle(TestCase):
def setUpClass(cls):
cls.data = sample_data
cls.data.columns = cls.data.columns.str.lower()
cls.open = cls.data['open']
cls.high = cls.data['high']
cls.low = cls.data['low']
cls.close = cls.data['close']
if 'volume' in cls.data.columns: cls.volume = cls.data['volume']
cls.open = cls.data["open"]
cls.high = cls.data["high"]
cls.low = cls.data["low"]
cls.close = cls.data["close"]
if "volume" in cls.data.columns: cls.volume = cls.data["volume"]
@classmethod
def tearDownClass(cls):
@@ -26,7 +26,7 @@ class TestCandle(TestCase):
del cls.high
del cls.low
del cls.close
if hasattr(cls, 'volume'): del cls.volume
if hasattr(cls, "volume"): del cls.volume
del cls.data
tests/test_indicator_candle_ext.py
+2 -2
View File
@@ -26,9 +26,9 @@ class TestCandleExtension(TestCase):
def test_ha_ext(self):
self.data.ta.ha(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-4:]), ['HA_open', 'HA_high', 'HA_low', 'HA_close'])
self.assertEqual(list(self.data.columns[-4:]), ["HA_open", "HA_high", "HA_low", "HA_close"])
def test_cdl_doji_ext(self):
self.data.ta.cdl_doji(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'CDL_DOJI_10_0.1')
self.assertEqual(self.data.columns[-1], "CDL_DOJI_10_0.1")
tests/test_indicator_momentum.py
+43 -43
View File
@@ -14,11 +14,11 @@ class TestMomentum(TestCase):
def setUpClass(cls):
cls.data = sample_data
cls.data.columns = cls.data.columns.str.lower()
cls.open = cls.data['open']
cls.high = cls.data['high']
cls.low = cls.data['low']
cls.close = cls.data['close']
if 'volume' in cls.data.columns: cls.volume = cls.data['volume']
cls.open = cls.data["open"]
cls.high = cls.data["high"]
cls.low = cls.data["low"]
cls.close = cls.data["close"]
if "volume" in cls.data.columns: cls.volume = cls.data["volume"]
@classmethod
def tearDownClass(cls):
@@ -26,7 +26,7 @@ class TestMomentum(TestCase):
del cls.high
del cls.low
del cls.close
if hasattr(cls, 'volume'): del cls.volume
if hasattr(cls, "volume"): del cls.volume
del cls.data
@@ -62,12 +62,12 @@ class TestMomentum(TestCase):
def test_ao(self):
result = pandas_ta.ao(self.high, self.low)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'AO_5_34')
self.assertEqual(result.name, "AO_5_34")
def test_apo(self):
result = pandas_ta.apo(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'APO_12_26')
self.assertEqual(result.name, "APO_12_26")
try:
expected = tal.APO(self.close)
@@ -82,12 +82,12 @@ class TestMomentum(TestCase):
def test_bias(self):
result = pandas_ta.bias(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'BIAS_SMA_26')
self.assertEqual(result.name, "BIAS_SMA_26")
def test_bop(self):
result = pandas_ta.bop(self.open, self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'BOP')
self.assertEqual(result.name, "BOP")
try:
expected = tal.BOP(self.open, self.high, self.low, self.close)
@@ -102,12 +102,12 @@ class TestMomentum(TestCase):
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')
self.assertEqual(result.name, "BRAR_26")
def test_cci(self):
result = pandas_ta.cci(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'CCI_14_0.015')
self.assertEqual(result.name, "CCI_14_0.015")
try:
expected = tal.CCI(self.high, self.low, self.close)
@@ -122,12 +122,12 @@ class TestMomentum(TestCase):
def test_cg(self):
result = pandas_ta.cg(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'CG_10')
self.assertEqual(result.name, "CG_10")
def test_cmo(self):
result = pandas_ta.cmo(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'CMO_14')
self.assertEqual(result.name, "CMO_14")
try:
expected = tal.CMO(self.close)
@@ -142,45 +142,45 @@ class TestMomentum(TestCase):
def test_coppock(self):
result = pandas_ta.coppock(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'COPC_11_14_10')
self.assertEqual(result.name, "COPC_11_14_10")
def test_fisher(self):
result = pandas_ta.fisher(self.high, self.low)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'FISHERT_5')
self.assertEqual(result.name, "FISHERT_5")
def test_inertia(self):
result = pandas_ta.inertia(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'INERTIA_20_14')
self.assertEqual(result.name, "INERTIA_20_14")
result = pandas_ta.inertia(self.close, self.high, self.low, refined=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'INERTIAr_20_14')
self.assertEqual(result.name, "INERTIAr_20_14")
result = pandas_ta.inertia(self.close, self.high, self.low, thirds=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'INERTIAt_20_14')
self.assertEqual(result.name, "INERTIAt_20_14")
def test_kdj(self):
result = pandas_ta.kdj(self.high, self.low, self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'KDJ_9_3')
self.assertEqual(result.name, "KDJ_9_3")
def test_kst(self):
result = pandas_ta.kst(self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'KST_10_15_20_30_10_10_10_15_9')
self.assertEqual(result.name, "KST_10_15_20_30_10_10_10_15_9")
def test_macd(self):
result = pandas_ta.macd(self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'MACD_12_26_9')
self.assertEqual(result.name, "MACD_12_26_9")
try:
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]})
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:
try:
@@ -204,7 +204,7 @@ class TestMomentum(TestCase):
def test_mom(self):
result = pandas_ta.mom(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'MOM_10')
self.assertEqual(result.name, "MOM_10")
try:
expected = tal.MOM(self.close)
@@ -219,32 +219,32 @@ class TestMomentum(TestCase):
def test_ppo(self):
result = pandas_ta.ppo(self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'PPO_12_26_9')
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)
pdt.assert_series_equal(result["PPO_12_26_9"], expected, check_names=False)
except AssertionError as ae:
try:
corr = pandas_ta.utils.df_error_analysis(result['PPO_12_26_9'], expected, col=CORRELATION)
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)
error_analysis(result["PPO_12_26_9"], CORRELATION, ex)
def test_psl(self):
result = pandas_ta.psl(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'PSL_12')
self.assertEqual(result.name, "PSL_12")
def test_pvo(self):
result = pandas_ta.pvo(self.volume)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'PVO_12_26_9')
self.assertEqual(result.name, "PVO_12_26_9")
def test_roc(self):
result = pandas_ta.roc(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'ROC_10')
self.assertEqual(result.name, "ROC_10")
try:
expected = tal.ROC(self.close)
@@ -259,7 +259,7 @@ class TestMomentum(TestCase):
def test_rsi(self):
result = pandas_ta.rsi(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'RSI_14')
self.assertEqual(result.name, "RSI_14")
try:
expected = tal.RSI(self.close)
@@ -274,37 +274,37 @@ class TestMomentum(TestCase):
def test_rvgi(self):
result = pandas_ta.rvgi(self.open, self.high, self.low, self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'RVGI_14_4')
self.assertEqual(result.name, "RVGI_14_4")
def test_slope(self):
result = pandas_ta.slope(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'SLOPE_1')
self.assertEqual(result.name, "SLOPE_1")
def test_slope_as_angle(self):
result = pandas_ta.slope(self.close, as_angle=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'ANGLEr_1')
self.assertEqual(result.name, "ANGLEr_1")
def test_slope_as_angle_to_degrees(self):
result = pandas_ta.slope(self.close, as_angle=True, to_degrees=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'ANGLEd_1')
self.assertEqual(result.name, "ANGLEd_1")
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)
self.assertEqual(result.name, 'STOCH_14_14_14')
self.assertEqual(result.name, "STOCH_14_14_14")
self.assertEqual(len(result.columns), 4)
result = pandas_ta.stoch(self.high, self.low, self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'STOCH_14_5_3')
self.assertEqual(result.name, "STOCH_14_5_3")
try:
tal_stochf = tal.STOCHF(self.high, self.low, self.close)
tal_stoch = tal.STOCH(self.high, self.low, self.close)
tal_stochdf = DataFrame({'STOCHF_14': tal_stochf[0], 'STOCHF_3': tal_stochf[1], 'STOCH_5': tal_stoch[0], 'STOCH_3': tal_stoch[1]})
tal_stochdf = DataFrame({"STOCHF_14": tal_stochf[0], "STOCHF_3": tal_stochf[1], "STOCH_5": tal_stoch[0], "STOCH_3": tal_stoch[1]})
pdt.assert_frame_equal(result, tal_stochdf)
except AssertionError as ae:
try:
@@ -334,17 +334,17 @@ class TestMomentum(TestCase):
def test_trix(self):
result = pandas_ta.trix(self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'TRIX_30_9')
self.assertEqual(result.name, "TRIX_30_9")
def test_tsi(self):
result = pandas_ta.tsi(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'TSI_13_25')
self.assertEqual(result.name, "TSI_13_25")
def test_uo(self):
result = pandas_ta.uo(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'UO_7_14_28')
self.assertEqual(result.name, "UO_7_14_28")
try:
expected = tal.ULTOSC(self.high, self.low, self.close)
@@ -359,7 +359,7 @@ class TestMomentum(TestCase):
def test_willr(self):
result = pandas_ta.willr(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'WILLR_14')
self.assertEqual(result.name, "WILLR_14")
try:
expected = tal.WILLR(self.high, self.low, self.close)
tests/test_indicator_momentum_ext.py
+31 -31
View File
@@ -26,152 +26,152 @@ class TestMomentumExtension(TestCase):
def test_ao_ext(self):
self.data.ta.ao(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'AO_5_34')
self.assertEqual(self.data.columns[-1], "AO_5_34")
def test_apo_ext(self):
self.data.ta.apo(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'APO_12_26')
self.assertEqual(self.data.columns[-1], "APO_12_26")
def test_bias_ext(self):
self.data.ta.bias(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'BIAS_SMA_26')
self.assertEqual(self.data.columns[-1], "BIAS_SMA_26")
def test_bop_ext(self):
self.data.ta.bop(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'BOP')
self.assertEqual(self.data.columns[-1], "BOP")
def test_brar_ext(self):
self.data.ta.brar(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-2:]), ['AR_26', 'BR_26'])
self.assertEqual(list(self.data.columns[-2:]), ["AR_26", "BR_26"])
def test_cci_ext(self):
self.data.ta.cci(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'CCI_14_0.015')
self.assertEqual(self.data.columns[-1], "CCI_14_0.015")
def test_cg_ext(self):
self.data.ta.cg(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'CG_10')
self.assertEqual(self.data.columns[-1], "CG_10")
def test_cmo_ext(self):
self.data.ta.cmo(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'CMO_14')
self.assertEqual(self.data.columns[-1], "CMO_14")
def test_coppock_ext(self):
self.data.ta.coppock(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'COPC_11_14_10')
self.assertEqual(self.data.columns[-1], "COPC_11_14_10")
def test_fisher_ext(self):
self.data.ta.fisher(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'FISHERT_5')
self.assertEqual(self.data.columns[-1], "FISHERT_5")
def test_inertia_ext(self):
self.data.ta.inertia(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'INERTIA_20_14')
self.assertEqual(self.data.columns[-1], "INERTIA_20_14")
def test_inertia_refined_ext(self):
self.data.ta.inertia(refined=True, append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'INERTIAr_20_14')
self.assertEqual(self.data.columns[-1], "INERTIAr_20_14")
def test_inertia_thirds_ext(self):
self.data.ta.inertia(thirds=True, append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'INERTIAt_20_14')
self.assertEqual(self.data.columns[-1], "INERTIAt_20_14")
def test_kdj_ext(self):
self.data.ta.kdj(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-3:]), ['K_9_3', 'D_9_3', 'J_9_3'])
self.assertEqual(list(self.data.columns[-3:]), ["K_9_3", "D_9_3", "J_9_3"])
def test_kst_ext(self):
self.data.ta.kst(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-2:]), ['KST_10_15_20_30_10_10_10_15', 'KSTs_9'])
self.assertEqual(list(self.data.columns[-2:]), ["KST_10_15_20_30_10_10_10_15", "KSTs_9"])
def test_macd_ext(self):
self.data.ta.macd(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-3:]), ['MACD_12_26_9', 'MACDh_12_26_9', 'MACDs_12_26_9'])
self.assertEqual(list(self.data.columns[-3:]), ["MACD_12_26_9", "MACDh_12_26_9", "MACDs_12_26_9"])
def test_mom_ext(self):
self.data.ta.mom(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'MOM_10')
self.assertEqual(self.data.columns[-1], "MOM_10")
def test_ppo_ext(self):
self.data.ta.ppo(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-3:]), ['PPO_12_26_9', 'PPOh_12_26_9', 'PPOs_12_26_9'])
self.assertEqual(list(self.data.columns[-3:]), ["PPO_12_26_9", "PPOh_12_26_9", "PPOs_12_26_9"])
def test_psl_ext(self):
self.data.ta.psl(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'PSL_12')
self.assertEqual(self.data.columns[-1], "PSL_12")
def test_pvo_ext(self):
self.data.ta.pvo(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-3:]), ['PVO_12_26_9', 'PVOh_12_26_9', 'PVOs_12_26_9'])
self.assertEqual(list(self.data.columns[-3:]), ["PVO_12_26_9", "PVOh_12_26_9", "PVOs_12_26_9"])
def test_roc_ext(self):
self.data.ta.roc(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'ROC_10')
self.assertEqual(self.data.columns[-1], "ROC_10")
def test_rsi_ext(self):
self.data.ta.rsi(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'RSI_14')
self.assertEqual(self.data.columns[-1], "RSI_14")
def test_rvgi_ext(self):
self.data.ta.rvgi(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-2:]), ['RVGI_14_4', 'RVGIs_14_4'])
self.assertEqual(list(self.data.columns[-2:]), ["RVGI_14_4", "RVGIs_14_4"])
def test_slope_ext(self):
self.data.ta.slope(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'SLOPE_1')
self.assertEqual(self.data.columns[-1], "SLOPE_1")
self.data.ta.slope(append=True, as_angle=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'ANGLEr_1')
self.assertEqual(self.data.columns[-1], "ANGLEr_1")
self.data.ta.slope(append=True, as_angle=True, to_degrees=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'ANGLEd_1')
self.assertEqual(self.data.columns[-1], "ANGLEd_1")
def test_stoch_ext(self):
self.data.ta.stoch(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-4:]), ['STOCHFk_14', 'STOCHFd_3', 'STOCHk_5', 'STOCHd_3'])
self.assertEqual(list(self.data.columns[-4:]), ["STOCHFk_14", "STOCHFd_3", "STOCHk_5", "STOCHd_3"])
def test_trix_ext(self):
self.data.ta.trix(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-2:]), ['TRIX_30_9', 'TRIXs_30_9'])
self.assertEqual(list(self.data.columns[-2:]), ["TRIX_30_9", "TRIXs_30_9"])
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(self.data.columns[-1], "TSI_13_25")
def test_uo_ext(self):
self.data.ta.uo(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'UO_7_14_28')
self.assertEqual(self.data.columns[-1], "UO_7_14_28")
def test_willr_ext(self):
self.data.ta.willr(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'WILLR_14')
self.assertEqual(self.data.columns[-1], "WILLR_14")
tests/test_indicator_overlap.py
+37 -37
View File
@@ -15,11 +15,11 @@ class TestOverlap(TestCase):
def setUpClass(cls):
cls.data = sample_data
cls.data.columns = cls.data.columns.str.lower()
cls.open = cls.data['open']
cls.high = cls.data['high']
cls.low = cls.data['low']
cls.close = cls.data['close']
if 'volume' in cls.data.columns: cls.volume = cls.data['volume']
cls.open = cls.data["open"]
cls.high = cls.data["high"]
cls.low = cls.data["low"]
cls.close = cls.data["close"]
if "volume" in cls.data.columns: cls.volume = cls.data["volume"]
@classmethod
def tearDownClass(cls):
@@ -27,7 +27,7 @@ class TestOverlap(TestCase):
del cls.high
del cls.low
del cls.close
if hasattr(cls, 'volume'): del cls.volume
if hasattr(cls, "volume"): del cls.volume
del cls.data
@@ -39,7 +39,7 @@ class TestOverlap(TestCase):
def test_dema(self):
result = pandas_ta.dema(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'DEMA_10')
self.assertEqual(result.name, "DEMA_10")
try:
expected = tal.DEMA(self.close, 10)
@@ -54,7 +54,7 @@ class TestOverlap(TestCase):
def test_ema(self):
result = pandas_ta.ema(self.close, presma=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'EMA_10')
self.assertEqual(result.name, "EMA_10")
try:
expected = tal.EMA(self.close, 10)
@@ -69,17 +69,17 @@ class TestOverlap(TestCase):
def test_fwma(self):
result = pandas_ta.fwma(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'FWMA_10')
self.assertEqual(result.name, "FWMA_10")
def test_hl2(self):
result = pandas_ta.hl2(self.high, self.low)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'HL2')
self.assertEqual(result.name, "HL2")
def test_hlc3(self):
result = pandas_ta.hlc3(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'HLC3')
self.assertEqual(result.name, "HLC3")
try:
expected = tal.TYPPRICE(self.high, self.low, self.close)
@@ -94,24 +94,24 @@ class TestOverlap(TestCase):
def test_hma(self):
result = pandas_ta.hma(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'HMA_10')
self.assertEqual(result.name, "HMA_10")
def test_kama(self):
result = pandas_ta.kama(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'KAMA_10_2_30')
self.assertEqual(result.name, "KAMA_10_2_30")
def test_ichimoku(self):
ichimoku, span = pandas_ta.ichimoku(self.high, self.low, self.close)
self.assertIsInstance(ichimoku, DataFrame)
self.assertIsInstance(span, DataFrame)
self.assertEqual(ichimoku.name, 'ICHIMOKU_9_26_52')
self.assertEqual(span.name, 'ICHISPAN_9_26')
self.assertEqual(ichimoku.name, "ICHIMOKU_9_26_52")
self.assertEqual(span.name, "ICHISPAN_9_26")
def test_linreg(self):
result = pandas_ta.linreg(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'LR_14')
self.assertEqual(result.name, "LR_14")
try:
expected = tal.LINEARREG(self.close)
@@ -126,7 +126,7 @@ class TestOverlap(TestCase):
def test_linreg_angle(self):
result = pandas_ta.linreg(self.close, angle=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'LRa_14')
self.assertEqual(result.name, "LRa_14")
try:
expected = tal.LINEARREG_ANGLE(self.close)
@@ -141,7 +141,7 @@ class TestOverlap(TestCase):
def test_linreg_intercept(self):
result = pandas_ta.linreg(self.close, intercept=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'LRb_14')
self.assertEqual(result.name, "LRb_14")
try:
expected = tal.LINEARREG_INTERCEPT(self.close)
@@ -156,12 +156,12 @@ class TestOverlap(TestCase):
def test_linreg_r(self):
result = pandas_ta.linreg(self.close, r=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'LRr_14')
self.assertEqual(result.name, "LRr_14")
def test_linreg_slope(self):
result = pandas_ta.linreg(self.close, slope=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'LRm_14')
self.assertEqual(result.name, "LRm_14")
try:
expected = tal.LINEARREG_SLOPE(self.close)
@@ -176,7 +176,7 @@ class TestOverlap(TestCase):
def test_midpoint(self):
result = pandas_ta.midpoint(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'MIDPOINT_2')
self.assertEqual(result.name, "MIDPOINT_2")
try:
expected = tal.MIDPOINT(self.close, 2)
@@ -191,7 +191,7 @@ class TestOverlap(TestCase):
def test_midprice(self):
result = pandas_ta.midprice(self.high, self.low)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'MIDPRICE_2')
self.assertEqual(result.name, "MIDPRICE_2")
try:
expected = tal.MIDPRICE(self.high, self.low, 2)
@@ -206,27 +206,27 @@ class TestOverlap(TestCase):
def test_ohlc4(self):
result = pandas_ta.ohlc4(self.open, self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'OHLC4')
self.assertEqual(result.name, "OHLC4")
def test_pwma(self):
result = pandas_ta.pwma(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'PWMA_10')
self.assertEqual(result.name, "PWMA_10")
def test_rma(self):
result = pandas_ta.rma(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'RMA_10')
self.assertEqual(result.name, "RMA_10")
def test_sinwma(self):
result = pandas_ta.sinwma(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'SINWMA_14')
self.assertEqual(result.name, "SINWMA_14")
def test_sma(self):
result = pandas_ta.sma(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'SMA_10')
self.assertEqual(result.name, "SMA_10")
try:
expected = tal.SMA(self.close, 10)
@@ -241,17 +241,17 @@ class TestOverlap(TestCase):
def test_swma(self):
result = pandas_ta.swma(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'SWMA_10')
self.assertEqual(result.name, "SWMA_10")
def test_supertrend(self):
result = pandas_ta.supertrend(self.high, self.low, self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'SUPERT_7_3.0')
self.assertEqual(result.name, "SUPERT_7_3.0")
def test_t3(self):
result = pandas_ta.t3(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'T3_10_0.7')
self.assertEqual(result.name, "T3_10_0.7")
try:
expected = tal.T3(self.close, 10)
@@ -266,7 +266,7 @@ class TestOverlap(TestCase):
def test_tema(self):
result = pandas_ta.tema(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'TEMA_10')
self.assertEqual(result.name, "TEMA_10")
try:
expected = tal.TEMA(self.close, 10)
@@ -281,7 +281,7 @@ class TestOverlap(TestCase):
def test_trima(self):
result = pandas_ta.trima(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'TRIMA_10')
self.assertEqual(result.name, "TRIMA_10")
try:
expected = tal.TRIMA(self.close, 10)
@@ -296,17 +296,17 @@ class TestOverlap(TestCase):
def test_vwap(self):
result = pandas_ta.vwap(self.high, self.low, self.close, self.volume)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'VWAP')
self.assertEqual(result.name, "VWAP")
def test_vwma(self):
result = pandas_ta.vwma(self.close, self.volume)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'VWMA_10')
self.assertEqual(result.name, "VWMA_10")
def test_wcp(self):
result = pandas_ta.wcp(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'WCP')
self.assertEqual(result.name, "WCP")
try:
expected = tal.WCLPRICE(self.high, self.low, self.close)
@@ -321,7 +321,7 @@ class TestOverlap(TestCase):
def test_wma(self):
result = pandas_ta.wma(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'WMA_10')
self.assertEqual(result.name, "WMA_10")
try:
expected = tal.WMA(self.close, 10)
@@ -336,4 +336,4 @@ class TestOverlap(TestCase):
def test_zlma(self):
result = pandas_ta.zlma(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'ZL_EMA_10')
self.assertEqual(result.name, "ZL_EMA_10")
tests/test_indicator_overlap_ext.py
+25 -25
View File
@@ -26,87 +26,87 @@ class TestOverlapExtension(TestCase):
def test_dema_ext(self):
self.data.ta.dema(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'DEMA_10')
self.assertEqual(self.data.columns[-1], "DEMA_10")
def test_ema_ext(self):
self.data.ta.ema(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'EMA_10')
self.assertEqual(self.data.columns[-1], "EMA_10")
def test_fwma_ext(self):
self.data.ta.fwma(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'FWMA_10')
self.assertEqual(self.data.columns[-1], "FWMA_10")
def test_hl2_ext(self):
self.data.ta.hl2(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'HL2')
self.assertEqual(self.data.columns[-1], "HL2")
def test_hlc3_ext(self):
self.data.ta.hlc3(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'HLC3')
self.assertEqual(self.data.columns[-1], "HLC3")
def test_hma_ext(self):
self.data.ta.hma(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'HMA_10')
self.assertEqual(self.data.columns[-1], "HMA_10")
def test_kama_ext(self):
self.data.ta.kama(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'KAMA_10_2_30')
self.assertEqual(self.data.columns[-1], "KAMA_10_2_30")
def test_ichimoku_ext(self):
self.data.ta.ichimoku(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-5:]), ['ISA_9', 'ISB_26', 'ITS_9', 'IKS_26', 'ICS_26'])
self.assertEqual(list(self.data.columns[-5:]), ["ISA_9", "ISB_26", "ITS_9", "IKS_26", "ICS_26"])
def test_linreg_ext(self):
self.data.ta.linreg(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'LR_14')
self.assertEqual(self.data.columns[-1], "LR_14")
def test_midpoint_ext(self):
self.data.ta.midpoint(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'MIDPOINT_2')
self.assertEqual(self.data.columns[-1], "MIDPOINT_2")
def test_midprice_ext(self):
self.data.ta.midprice(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'MIDPRICE_2')
self.assertEqual(self.data.columns[-1], "MIDPRICE_2")
def test_ohlc4_ext(self):
self.data.ta.ohlc4(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'OHLC4')
self.assertEqual(self.data.columns[-1], "OHLC4")
def test_pwma_ext(self):
self.data.ta.pwma(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'PWMA_10')
self.assertEqual(self.data.columns[-1], "PWMA_10")
def test_rma_ext(self):
self.data.ta.rma(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'RMA_10')
self.assertEqual(self.data.columns[-1], "RMA_10")
def test_sinwma_ext(self):
self.data.ta.sinwma(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'SINWMA_14')
self.assertEqual(self.data.columns[-1], "SINWMA_14")
def test_sma_ext(self):
self.data.ta.sma(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'SMA_10')
self.assertEqual(self.data.columns[-1], "SMA_10")
def test_swma_ext(self):
self.data.ta.swma(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'SWMA_10')
self.assertEqual(self.data.columns[-1], "SWMA_10")
def test_supertrend_ext(self):
self.data.ta.supertrend(append=True)
@@ -116,39 +116,39 @@ class TestOverlapExtension(TestCase):
def test_t3_ext(self):
self.data.ta.t3(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'T3_10_0.7')
self.assertEqual(self.data.columns[-1], "T3_10_0.7")
def test_tema_ext(self):
self.data.ta.tema(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'TEMA_10')
self.assertEqual(self.data.columns[-1], "TEMA_10")
def test_trima_ext(self):
self.data.ta.trima(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'TRIMA_10')
self.assertEqual(self.data.columns[-1], "TRIMA_10")
def test_vwap_ext(self):
self.data.ta.vwap(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'VWAP')
self.assertEqual(self.data.columns[-1], "VWAP")
def test_vwma_ext(self):
self.data.ta.vwma(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'VWMA_10')
self.assertEqual(self.data.columns[-1], "VWMA_10")
def test_wcp_ext(self):
self.data.ta.wcp(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'WCP')
self.assertEqual(self.data.columns[-1], "WCP")
def test_wma_ext(self):
self.data.ta.wma(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'WMA_10')
self.assertEqual(self.data.columns[-1], "WMA_10")
def test_zlma_ext(self):
self.data.ta.zlma(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'ZL_EMA_10')
self.assertEqual(self.data.columns[-1], "ZL_EMA_10")
tests/test_indicator_performance.py
+11 -11
View File
@@ -10,7 +10,7 @@ class TestPerformace(TestCase):
@classmethod
def setUpClass(cls):
cls.data = sample_data
cls.close = cls.data['close']
cls.close = cls.data["close"]
cls.islong = cls.close > pandas_ta.sma(cls.close, length=50)
@classmethod
@@ -27,42 +27,42 @@ class TestPerformace(TestCase):
def test_log_return(self):
result = pandas_ta.log_return(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'LOGRET_1')
self.assertEqual(result.name, "LOGRET_1")
def test_cum_log_return(self):
result = pandas_ta.log_return(self.close, cumulative=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'CUMLOGRET_1')
self.assertEqual(result.name, "CUMLOGRET_1")
def test_percent_return(self):
result = pandas_ta.percent_return(self.close, cumulative=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'PCTRET_1')
self.assertEqual(result.name, "PCTRET_1")
def test_cum_percent_return(self):
result = pandas_ta.percent_return(self.close, cumulative=True)
self.assertEqual(result.name, 'CUMPCTRET_1')
self.assertEqual(result.name, "CUMPCTRET_1")
def test_log_trend_return(self):
result = pandas_ta.trend_return(self.close, self.islong, log=True, cumulative=False)
self.assertEqual(result.name, 'LTR')
self.assertEqual(result.name, "LTR")
def test_cum_log_trend_return(self):
result = pandas_ta.trend_return(self.close, self.islong, log=True, cumulative=True)
self.assertEqual(result.name, 'CLTR')
self.assertEqual(result.name, "CLTR")
def test_variable_cum_log_trend_return(self):
result = pandas_ta.trend_return(self.close, self.islong, log=True, cumulative=True, variable=True)
self.assertEqual(result.name, 'CLTR')
self.assertEqual(result.name, "CLTR")
def test_pct_trend_return(self):
result = pandas_ta.trend_return(self.close, self.islong, log=False, cumulative=False)
self.assertEqual(result.name, 'PTR')
self.assertEqual(result.name, "PTR")
def test_cum_pct_trend_return(self):
result = pandas_ta.trend_return(self.close, self.islong, log=False, cumulative=True)
self.assertEqual(result.name, 'CPTR')
self.assertEqual(result.name, "CPTR")
def test_variable_pct_log_trend_return(self):
result = pandas_ta.trend_return(self.close, self.islong, log=False, cumulative=True, variable=True)
self.assertEqual(result.name, 'CPTR')
self.assertEqual(result.name, "CPTR")
tests/test_indicator_performance_ext.py
+9 -9
View File
@@ -10,7 +10,7 @@ class TestPerformaceExtension(TestCase):
@classmethod
def setUpClass(cls):
cls.data = sample_data
cls.islong = cls.data['close'] > pandas_ta.sma(cls.data['close'], length=50)
cls.islong = cls.data["close"] > pandas_ta.sma(cls.data["close"], length=50)
@classmethod
def tearDownClass(cls):
@@ -25,39 +25,39 @@ class TestPerformaceExtension(TestCase):
def test_log_return_ext(self):
self.data.ta.log_return(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'LOGRET_1')
self.assertEqual(self.data.columns[-1], "LOGRET_1")
def test_cum_log_return_ext(self):
self.data.ta.log_return(append=True, cumulative=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'CUMLOGRET_1')
self.assertEqual(self.data.columns[-1], "CUMLOGRET_1")
def test_percent_return_ext(self):
self.data.ta.percent_return(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'PCTRET_1')
self.assertEqual(self.data.columns[-1], "PCTRET_1")
def test_cum_percent_return_ext(self):
self.data.ta.percent_return(append=True, cumulative=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'CUMPCTRET_1')
self.assertEqual(self.data.columns[-1], "CUMPCTRET_1")
def test_log_trend_return_ext(self):
self.data.ta.trend_return(trend=self.islong, log=True, cumulative=False, append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'LTR')
self.assertEqual(self.data.columns[-1], "LTR")
def test_cum_log_trend_return_ext(self):
self.data.ta.trend_return(trend=self.islong, log=True, cumulative=True, append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'CLTR')
self.assertEqual(self.data.columns[-1], "CLTR")
def test_pct_trend_return_ext(self):
self.data.ta.trend_return(trend=self.islong, log=False, cumulative=False, append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'PTR')
self.assertEqual(self.data.columns[-1], "PTR")
def test_cum_pct_trend_return_ext(self):
self.data.ta.trend_return(trend=self.islong, log=False, cumulative=True, append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'CPTR')
self.assertEqual(self.data.columns[-1], "CPTR")
tests/test_indicator_statistics.py
+15 -15
View File
@@ -14,11 +14,11 @@ class TestStatistics(TestCase):
def setUpClass(cls):
cls.data = sample_data
cls.data.columns = cls.data.columns.str.lower()
cls.open = cls.data['open']
cls.high = cls.data['high']
cls.low = cls.data['low']
cls.close = cls.data['close']
if 'volume' in cls.data.columns: cls.volume = cls.data['volume']
cls.open = cls.data["open"]
cls.high = cls.data["high"]
cls.low = cls.data["low"]
cls.close = cls.data["close"]
if "volume" in cls.data.columns: cls.volume = cls.data["volume"]
@classmethod
def tearDownClass(cls):
@@ -26,7 +26,7 @@ class TestStatistics(TestCase):
del cls.high
del cls.low
del cls.close
if hasattr(cls, 'volume'): del cls.volume
if hasattr(cls, "volume"): del cls.volume
del cls.data
@@ -36,37 +36,37 @@ class TestStatistics(TestCase):
def test_entropy(self):
result = pandas_ta.entropy(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'ENTP_10')
self.assertEqual(result.name, "ENTP_10")
def test_kurtosis(self):
result = pandas_ta.kurtosis(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'KURT_30')
self.assertEqual(result.name, "KURT_30")
def test_mad(self):
result = pandas_ta.mad(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'MAD_30')
self.assertEqual(result.name, "MAD_30")
def test_median(self):
result = pandas_ta.median(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'MEDIAN_30')
self.assertEqual(result.name, "MEDIAN_30")
def test_quantile(self):
result = pandas_ta.quantile(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'QTL_30_0.5')
self.assertEqual(result.name, "QTL_30_0.5")
def test_skew(self):
result = pandas_ta.skew(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'SKEW_30')
self.assertEqual(result.name, "SKEW_30")
def test_stdev(self):
result = pandas_ta.stdev(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'STDEV_30')
self.assertEqual(result.name, "STDEV_30")
try:
expected = tal.STDDEV(self.close, 30)
@@ -81,7 +81,7 @@ class TestStatistics(TestCase):
def test_variance(self):
result = pandas_ta.variance(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'VAR_30')
self.assertEqual(result.name, "VAR_30")
try:
expected = tal.VAR(self.close, 30)
@@ -96,4 +96,4 @@ class TestStatistics(TestCase):
def test_zscore(self):
result = pandas_ta.zscore(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'Z_30')
self.assertEqual(result.name, "Z_30")
tests/test_indicator_statistics_ext.py
+8 -8
View File
@@ -26,39 +26,39 @@ class TestStatisticsExtension(TestCase):
def test_entropy_ext(self):
self.data.ta.entropy(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'ENTP_10')
self.assertEqual(self.data.columns[-1], "ENTP_10")
def test_kurtosis_ext(self):
self.data.ta.kurtosis(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'KURT_30')
self.assertEqual(self.data.columns[-1], "KURT_30")
def test_mad_ext(self):
self.data.ta.mad(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'MAD_30')
self.assertEqual(self.data.columns[-1], "MAD_30")
def test_median_ext(self):
self.data.ta.median(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'MEDIAN_30')
self.assertEqual(self.data.columns[-1], "MEDIAN_30")
def test_quantile_ext(self):
self.data.ta.quantile(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'QTL_30_0.5')
self.assertEqual(self.data.columns[-1], "QTL_30_0.5")
def test_skew_ext(self):
self.data.ta.skew(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'SKEW_30')
self.assertEqual(self.data.columns[-1], "SKEW_30")
def test_stdev_ext(self):
self.data.ta.stdev(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'STDEV_30')
self.assertEqual(self.data.columns[-1], "STDEV_30")
def test_variance_ext(self):
self.data.ta.variance(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'VAR_30')
self.assertEqual(self.data.columns[-1], "VAR_30")
tests/test_indicator_trend.py
+22 -22
View File
@@ -14,11 +14,11 @@ class TestTrend(TestCase):
def setUpClass(cls):
cls.data = sample_data
cls.data.columns = cls.data.columns.str.lower()
cls.open = cls.data['open']
cls.high = cls.data['high']
cls.low = cls.data['low']
cls.close = cls.data['close']
if 'volume' in cls.data.columns: cls.volume = cls.data['volume']
cls.open = cls.data["open"]
cls.high = cls.data["high"]
cls.low = cls.data["low"]
cls.close = cls.data["close"]
if "volume" in cls.data.columns: cls.volume = cls.data["volume"]
@classmethod
def tearDownClass(cls):
@@ -26,7 +26,7 @@ class TestTrend(TestCase):
del cls.high
del cls.low
del cls.close
if hasattr(cls, 'volume'): del cls.volume
if hasattr(cls, "volume"): del cls.volume
del cls.data
@@ -37,7 +37,7 @@ class TestTrend(TestCase):
def test_adx(self):
result = pandas_ta.adx(self.high, self.low, self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'ADX_14')
self.assertEqual(result.name, "ADX_14")
try:
expected = tal.ADX(self.high, self.low, self.close)
@@ -52,16 +52,16 @@ class TestTrend(TestCase):
def test_amat(self):
result = pandas_ta.amat(self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'AMAT_EMA_8_21_2')
self.assertEqual(result.name, "AMAT_EMA_8_21_2")
def test_aroon(self):
result = pandas_ta.aroon(self.high, self.low)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'AROON_14')
self.assertEqual(result.name, "AROON_14")
try:
expected = tal.AROON(self.high, self.low)
expecteddf = DataFrame({'AROOND_14': expected[0], 'AROONU_14': expected[1]})
expecteddf = DataFrame({"AROOND_14": expected[0], "AROONU_14": expected[1]})
pdt.assert_frame_equal(result, expecteddf)
except AssertionError as ae:
try:
@@ -92,44 +92,44 @@ class TestTrend(TestCase):
def test_chop(self):
result = pandas_ta.chop(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'CHOP_14_1_100')
self.assertEqual(result.name, "CHOP_14_1_100")
def test_cksp(self):
result = pandas_ta.cksp(self.high, self.low, self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'CKSP_10_1_9')
self.assertEqual(result.name, "CKSP_10_1_9")
def test_decreasing(self):
result = pandas_ta.decreasing(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'DEC_1')
self.assertEqual(result.name, "DEC_1")
def test_dpo(self):
result = pandas_ta.dpo(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'DPO_20')
self.assertEqual(result.name, "DPO_20")
def test_increasing(self):
result = pandas_ta.increasing(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'INC_1')
self.assertEqual(result.name, "INC_1")
def test_linear_decay(self):
result = pandas_ta.linear_decay(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'LDECAY_5')
self.assertEqual(result.name, "LDECAY_5")
def test_long_run(self):
result = pandas_ta.long_run(self.close, self.open)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'LR_2')
self.assertEqual(result.name, "LR_2")
def test_psar(self):
result = pandas_ta.psar(self.high, self.low)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'PSAR_0.02_0.2')
self.assertEqual(result.name, "PSAR_0.02_0.2")
# Combine Long and Short SAR's into one SAR value
# Combine Long and Short SAR"s into one SAR value
psar = result[result.columns[:2]].fillna(0)
psar = psar[psar.columns[0]] + psar[psar.columns[1]]
psar.name = result.name
@@ -147,14 +147,14 @@ class TestTrend(TestCase):
def test_qstick(self):
result = pandas_ta.qstick(self.open, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'QS_10')
self.assertEqual(result.name, "QS_10")
def test_short_run(self):
result = pandas_ta.short_run(self.close, self.open)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'SR_2')
self.assertEqual(result.name, "SR_2")
def test_vortex(self):
result = pandas_ta.vortex(self.high, self.low, self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'VTX_14')
self.assertEqual(result.name, "VTX_14")
tests/test_indicator_trend_ext.py
+18 -18
View File
@@ -26,79 +26,79 @@ class TestTrendExtension(TestCase):
def test_adx_ext(self):
self.data.ta.adx(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-3:]), ['ADX_14', 'DMP_14', 'DMN_14'])
self.assertEqual(list(self.data.columns[-3:]), ["ADX_14", "DMP_14", "DMN_14"])
def test_amat_ext(self):
self.data.ta.amat(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-2:]), ['AMAT_LR_2', 'AMAT_SR_2'])
self.assertEqual(list(self.data.columns[-2:]), ["AMAT_LR_2", "AMAT_SR_2"])
def test_aroon_ext(self):
self.data.ta.aroon(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-3:]), ['AROOND_14', 'AROONU_14', 'AROONOSC_14'])
self.assertEqual(list(self.data.columns[-3:]), ["AROOND_14", "AROONU_14", "AROONOSC_14"])
def test_chop_ext(self):
self.data.ta.chop(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'CHOP_14_1_100')
self.assertEqual(self.data.columns[-1], "CHOP_14_1_100")
def test_cksp_ext(self):
self.data.ta.cksp(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'CKSPs_10_1_9')
self.assertEqual(self.data.columns[-1], "CKSPs_10_1_9")
def test_decreasing_ext(self):
self.data.ta.decreasing(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'DEC_1')
self.assertEqual(self.data.columns[-1], "DEC_1")
def test_dpo_ext(self):
self.data.ta.dpo(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'DPO_20')
self.assertEqual(self.data.columns[-1], "DPO_20")
def test_increasing_ext(self):
self.data.ta.increasing(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'INC_1')
self.assertEqual(self.data.columns[-1], "INC_1")
def test_linear_decay_ext(self):
self.data.ta.linear_decay(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'LDECAY_5')
self.assertEqual(self.data.columns[-1], "LDECAY_5")
def test_long_run_ext(self):
# Nothing passed, return self
self.assertEqual(self.data.ta.long_run(append=True).shape, self.data.shape)
fast = self.data.ta.ema('close', 8)
slow = self.data.ta.ema('close', 21)
fast = self.data.ta.ema("close", 8)
slow = self.data.ta.ema("close", 21)
self.data.ta.long_run(fast, slow, append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'LR_2')
self.assertEqual(self.data.columns[-1], "LR_2")
def test_psar_ext(self):
self.data.ta.psar(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-4:]), ['PSARl_0.02_0.2', 'PSARs_0.02_0.2', 'PSARaf_0.02_0.2', 'PSARr_0.02_0.2'])
self.assertEqual(list(self.data.columns[-4:]), ["PSARl_0.02_0.2", "PSARs_0.02_0.2", "PSARaf_0.02_0.2", "PSARr_0.02_0.2"])
def test_qstick_ext(self):
self.data.ta.qstick(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'QS_10')
self.assertEqual(self.data.columns[-1], "QS_10")
def test_short_run_ext(self):
# Nothing passed, return self
self.assertEqual(self.data.ta.short_run(append=True).shape, self.data.shape)
fast = self.data.ta.ema('close', 8)
slow = self.data.ta.ema('close', 21)
fast = self.data.ta.ema("close", 8)
slow = self.data.ta.ema("close", 21)
self.data.ta.short_run(fast, slow, append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'SR_2')
self.assertEqual(self.data.columns[-1], "SR_2")
def test_vortext_ext(self):
self.data.ta.vortex(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-2:]), ['VTXP_14', 'VTXM_14'])
self.assertEqual(list(self.data.columns[-2:]), ["VTXP_14", "VTXM_14"])
tests/test_indicator_volatility.py
+21 -21
View File
@@ -14,11 +14,11 @@ class TestVolatility(TestCase):
def setUpClass(cls):
cls.data = sample_data
cls.data.columns = cls.data.columns.str.lower()
cls.open = cls.data['open']
cls.high = cls.data['high']
cls.low = cls.data['low']
cls.close = cls.data['close']
if 'volume' in cls.data.columns: cls.volume = cls.data['volume']
cls.open = cls.data["open"]
cls.high = cls.data["high"]
cls.low = cls.data["low"]
cls.close = cls.data["close"]
if "volume" in cls.data.columns: cls.volume = cls.data["volume"]
@classmethod
def tearDownClass(cls):
@@ -26,7 +26,7 @@ class TestVolatility(TestCase):
del cls.high
del cls.low
del cls.close
if hasattr(cls, 'volume'): del cls.volume
if hasattr(cls, "volume"): del cls.volume
del cls.data
@@ -37,17 +37,17 @@ class TestVolatility(TestCase):
def test_aberration(self):
result = pandas_ta.aberration(self.high, self.low, self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'ABER_5_15')
self.assertEqual(result.name, "ABER_5_15")
def test_accbands(self):
result = pandas_ta.accbands(self.high, self.low, self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'ACCBANDS_20')
self.assertEqual(result.name, "ACCBANDS_20")
def test_atr(self):
result = pandas_ta.atr(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'ATR_14')
self.assertEqual(result.name, "ATR_14")
try:
expected = tal.ATR(self.high, self.low, self.close)
@@ -62,11 +62,11 @@ class TestVolatility(TestCase):
def test_bbands(self):
result = pandas_ta.bbands(self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'BBANDS_5')
self.assertEqual(result.name, "BBANDS_5_2.0")
try:
expected = tal.BBANDS(self.close)
expecteddf = DataFrame({'BBL_5': expected[0], 'BBM_5': expected[1], 'BBU_5': expected[2]})
expecteddf = DataFrame({"BBL_5_2.0": expected[0], "BBM_5_2.0": expected[1], "BBU_5_2.0": expected[2]})
pdt.assert_frame_equal(result, expecteddf)
except AssertionError as ae:
try:
@@ -90,27 +90,27 @@ class TestVolatility(TestCase):
def test_donchian(self):
result = pandas_ta.donchian(self.close)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'DC_10_20')
self.assertEqual(result.name, "DC_10_20")
result = pandas_ta.donchian(self.close, lower_length=20, upper_length=5)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'DC_20_5')
self.assertEqual(result.name, "DC_20_5")
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")
def test_massi(self):
result = pandas_ta.massi(self.high, self.low)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'MASSI_9_25')
self.assertEqual(result.name, "MASSI_9_25")
def test_natr(self):
result = pandas_ta.natr(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'NATR_14')
self.assertEqual(result.name, "NATR_14")
try:
expected = tal.NATR(self.high, self.low, self.close)
@@ -125,25 +125,25 @@ class TestVolatility(TestCase):
def test_pdist(self):
result = pandas_ta.pdist(self.open, self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'PDIST')
self.assertEqual(result.name, "PDIST")
def test_rvi(self):
result = pandas_ta.rvi(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'RVI_14')
self.assertEqual(result.name, "RVI_14")
result = pandas_ta.rvi(self.close, self.high, self.low, refined=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'RVIr_14')
self.assertEqual(result.name, "RVIr_14")
result = pandas_ta.rvi(self.close, self.high, self.low, thirds=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'RVIt_14')
self.assertEqual(result.name, "RVIt_14")
def test_true_range(self):
result = pandas_ta.true_range(self.high, self.low, self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'TRUERANGE_1')
self.assertEqual(result.name, "TRUERANGE_1")
try:
expected = tal.TRANGE(self.high, self.low, self.close)
tests/test_indicator_volatility_ext.py
+13 -13
View File
@@ -23,64 +23,64 @@ class TestVolatilityExtension(TestCase):
def test_aberration_ext(self):
self.data.ta.aberration(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-4:]), ['ABER_ZG_5_15', 'ABER_SG_5_15', 'ABER_XG_5_15', 'ABER_ATR_5_15'])
self.assertEqual(list(self.data.columns[-4:]), ["ABER_ZG_5_15", "ABER_SG_5_15", "ABER_XG_5_15", "ABER_ATR_5_15"])
def test_accbands_ext(self):
self.data.ta.accbands(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-3:]), ['ACCBL_20', 'ACCBM_20', 'ACCBU_20'])
self.assertEqual(list(self.data.columns[-3:]), ["ACCBL_20", "ACCBM_20", "ACCBU_20"])
def test_atr_ext(self):
self.data.ta.atr(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'ATR_14')
self.assertEqual(self.data.columns[-1], "ATR_14")
def test_bbands_ext(self):
self.data.ta.bbands(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-3:]), ['BBL_5', 'BBM_5', 'BBU_5'])
self.assertEqual(list(self.data.columns[-3:]), ["BBL_5_2.0", "BBM_5_2.0", "BBU_5_2.0"])
def test_donchian_ext(self):
self.data.ta.donchian(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-3:]), ['DCL_10_20', 'DCM_10_20', 'DCU_10_20'])
self.assertEqual(list(self.data.columns[-3:]), ["DCL_10_20", "DCM_10_20", "DCU_10_20"])
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", "KCB_20", "KCU_20"])
def test_massi_ext(self):
self.data.ta.massi(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'MASSI_9_25')
self.assertEqual(self.data.columns[-1], "MASSI_9_25")
def test_natr_ext(self):
self.data.ta.natr(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'NATR_14')
self.assertEqual(self.data.columns[-1], "NATR_14")
def test_pdist_ext(self):
self.data.ta.pdist(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'PDIST')
self.assertEqual(self.data.columns[-1], "PDIST")
def test_rvi_ext(self):
self.data.ta.rvi(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'RVI_14')
self.assertEqual(self.data.columns[-1], "RVI_14")
def test_rvi_refined_ext(self):
self.data.ta.rvi(refined=True, append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'RVIr_14')
self.assertEqual(self.data.columns[-1], "RVIr_14")
def test_rvi_thirds_ext(self):
self.data.ta.rvi(thirds=True, append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'RVIt_14')
self.assertEqual(self.data.columns[-1], "RVIt_14")
def test_true_range_ext(self):
self.data.ta.true_range(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'TRUERANGE_1')
self.assertEqual(self.data.columns[-1], "TRUERANGE_1")
tests/test_indicator_volume.py
+20 -20
View File
@@ -14,11 +14,11 @@ class TestVolume(TestCase):
def setUpClass(cls):
cls.data = sample_data
cls.data.columns = cls.data.columns.str.lower()
cls.open = cls.data['open']
cls.high = cls.data['high']
cls.low = cls.data['low']
cls.close = cls.data['close']
if 'volume' in cls.data.columns: cls.volume_ = cls.data['volume']
cls.open = cls.data["open"]
cls.high = cls.data["high"]
cls.low = cls.data["low"]
cls.close = cls.data["close"]
if "volume" in cls.data.columns: cls.volume_ = cls.data["volume"]
@classmethod
def tearDownClass(cls):
@@ -26,7 +26,7 @@ class TestVolume(TestCase):
del cls.high
del cls.low
del cls.close
if hasattr(cls, 'volume'): del cls.volume_
if hasattr(cls, "volume"): del cls.volume_
del cls.data
@@ -37,7 +37,7 @@ class TestVolume(TestCase):
def test_ad(self):
result = pandas_ta.ad(self.high, self.low, self.close, self.volume_)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'AD')
self.assertEqual(result.name, "AD")
try:
expected = tal.AD(self.high, self.low, self.close, self.volume_)
@@ -52,12 +52,12 @@ class TestVolume(TestCase):
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')
self.assertEqual(result.name, "ADo")
def test_adosc(self):
result = pandas_ta.adosc(self.high, self.low, self.close, self.volume_)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'ADOSC_3_10')
self.assertEqual(result.name, "ADOSC_3_10")
try:
expected = tal.ADOSC(self.high, self.low, self.close, self.volume_)
@@ -72,27 +72,27 @@ 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_2_4_2_2_2")
def test_cmf(self):
result = pandas_ta.cmf(self.high, self.low, self.close, self.volume_)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'CMF_20')
self.assertEqual(result.name, "CMF_20")
def test_efi(self):
result = pandas_ta.efi(self.close, self.volume_)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'EFI_13')
self.assertEqual(result.name, "EFI_13")
def test_eom(self):
result = pandas_ta.eom(self.high, self.low, self.close, self.volume_)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'EOM_14_100000000')
self.assertEqual(result.name, "EOM_14_100000000")
def test_mfi(self):
result = pandas_ta.mfi(self.high, self.low, self.close, self.volume_)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'MFI_14')
self.assertEqual(result.name, "MFI_14")
try:
expected = tal.MFI(self.high, self.low, self.close, self.volume_)
@@ -107,12 +107,12 @@ class TestVolume(TestCase):
def test_nvi(self):
result = pandas_ta.nvi(self.close, self.volume_)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'NVI_1')
self.assertEqual(result.name, "NVI_1")
def test_obv(self):
result = pandas_ta.obv(self.close, self.volume_)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'OBV')
self.assertEqual(result.name, "OBV")
try:
expected = tal.OBV(self.close, self.volume_)
@@ -127,19 +127,19 @@ class TestVolume(TestCase):
def test_pvi(self):
result = pandas_ta.pvi(self.close, self.volume_)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'PVI_1')
self.assertEqual(result.name, "PVI_1")
def test_pvol(self):
result = pandas_ta.pvol(self.close, self.volume_)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'PVOL')
self.assertEqual(result.name, "PVOL")
def test_pvt(self):
result = pandas_ta.pvt(self.close, self.volume_)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'PVT')
self.assertEqual(result.name, "PVT")
def test_vp(self):
result = pandas_ta.vp(self.close, self.volume_)
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'VP_10')
self.assertEqual(result.name, "VP_10")
tests/test_indicator_volume_ext.py
+17 -17
View File
@@ -10,7 +10,7 @@ class TestVolumeExtension(TestCase):
@classmethod
def setUpClass(cls):
cls.data = sample_data
cls.open = cls.data['open']
cls.open = cls.data["open"]
@classmethod
def tearDownClass(cls):
@@ -25,49 +25,49 @@ class TestVolumeExtension(TestCase):
def test_ad_ext(self):
self.data.ta.ad(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'AD')
self.assertEqual(self.data.columns[-1], "AD")
def test_ad_open_ext(self):
self.data.ta.ad(open_=self.open, append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'ADo')
self.assertEqual(self.data.columns[-1], "ADo")
def test_adosc_ext(self):
self.data.ta.adosc(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'ADOSC_3_10')
self.assertEqual(self.data.columns[-1], "ADOSC_3_10")
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'])
# Remove 'OBV' so it does not interfere with test_obv_ext()
self.data.drop('OBV', axis=1, inplace=True)
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"])
# Remove "OBV" so it does not interfere with test_obv_ext()
self.data.drop("OBV", axis=1, inplace=True)
def test_cmf_ext(self):
self.data.ta.cmf(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'CMF_20')
self.assertEqual(self.data.columns[-1], "CMF_20")
def test_efi_ext(self):
self.data.ta.efi(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'EFI_13')
self.assertEqual(self.data.columns[-1], "EFI_13")
def test_eom_ext(self):
self.data.ta.eom(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'EOM_14_100000000')
self.assertEqual(self.data.columns[-1], "EOM_14_100000000")
def test_mfi_ext(self):
self.data.ta.mfi(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'MFI_14')
self.assertEqual(self.data.columns[-1], "MFI_14")
def test_nvi_ext(self):
self.data.ta.nvi(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'NVI_1')
self.assertEqual(self.data.columns[-1], "NVI_1")
# print(f"\nNVI: {self.data.columns[-1]}")
# print(f"NVI: {self.data.columns}")
@@ -76,24 +76,24 @@ class TestVolumeExtension(TestCase):
self.assertIsInstance(self.data, DataFrame)
# print(f"\nOBV: {self.data.columns[-1]}")
# print(f"OBV: {self.data.columns}")
self.assertEqual(self.data.columns[-1], 'OBV')
self.assertEqual(self.data.columns[-1], "OBV")
def test_pvi_ext(self):
self.data.ta.pvi(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'PVI_1')
self.assertEqual(self.data.columns[-1], "PVI_1")
def test_pvol_ext(self):
self.data.ta.pvol(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'PVOL')
self.assertEqual(self.data.columns[-1], "PVOL")
def test_pvt_ext(self):
self.data.ta.pvt(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], 'PVT')
self.assertEqual(self.data.columns[-1], "PVT")
def test_vp_ext(self):
result = self.data.ta.vp()
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, 'VP_10')
self.assertEqual(result.name, "VP_10")