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

MAINT refactoring BUG #436 #463 polygon import ENH numba DOC updates

Browse Source
This commit is contained in:
Kevin Johnson
2022-01-14 15:53:19 -08:00
parent 3aac80c2c1
commit 8e02623866
53 changed files with 1640 additions and 1316 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+7 -4
View File
@@ -56,7 +56,7 @@ _Pandas Technical Analysis_ (**Pandas TA**) is an easy to use library that lever
* [Candles](#candles-64)
* [Cycles](#cycles-2)
* [Momentum](#momentum-42)
* [Overlap](#overlap-35)
* [Overlap](#overlap-36)
* [Performance](#performance-3)
* [Statistics](#statistics-11)
* [Trend](#trend-19)
@@ -118,7 +118,7 @@ $ pip install pandas_ta
Latest Version
--------------
Best choice! Version: *0.3.41b*
Best choice! Version: *0.3.42b*
* Includes all fixes and updates between **pypi** and what is covered in this README.
```sh
$ pip install -U git+https://github.com/twopirllc/pandas-ta
@@ -752,7 +752,7 @@ df = df.ta.cdl_pattern(name=["doji", "inside"])
<br/>
### **Overlap** (35)
### **Overlap** (36)
* _Bill Williams Alligator_: **alligator**
* _Arnaud Legoux Moving Average_: **alma**
@@ -781,6 +781,7 @@ df = df.ta.cdl_pattern(name=["doji", "inside"])
* _Simple Moving Average_: **sma**
* _Smoothed Moving Average_: **smma**
* _Ehler's Super Smoother Filter_: **ssf**
* _Ehler's Super Smoother Filter (3 Poles)_: **ssf3**
* _Supertrend_: **supertrend**
* _Symmetric Weighted Moving Average_: **swma**
* _T3 Moving Average_: **t3**
@@ -1048,9 +1049,11 @@ help(ta.sample)
## **Updated Indicators**
* _Average True Range_ (**atr**): The default ```mamode``` is now "**RMA**" and with the same ```mamode``` options as TradingView. See ```help(ta.atr)```.
* _Exponential Moving Average_ (**ema**): The argument ```sma``` has been renamed ```presma`` to avoid potential name collision. When ```presma=True```, then the Pandas TA version will bootstrap **ema** like TA Lib. See ```help(ta.ema)```.
* _Kaufman Adaptive Moving Average_ (**kama**): An ```mamode``` as been added with default "**SMA**" to properly boostrap **kama**. _Note_: Not all MAs are usable. See ```help(ta.kama)```.
* _Linear Regression_ (**linreg**): Checks **numpy**'s version to determine whether to utilize the ```as_strided``` method or the newer ```sliding_window_view``` method. This should resolve Issues with Google Colab and it's delayed dependency updates as well as TensorFlow's dependencies as discussed in Issues [#285](https://github.com/twopirllc/pandas-ta/issues/285) and [#329](https://github.com/twopirllc/pandas-ta/issues/329).
* _Moving Average Convergence Divergence_ (**macd**): New argument ```asmode``` enables AS version of MACD. Default is False. See ```help(ta.macd)```.
* _Ehler's Super Smoother Filter_ (**ssf**): Some new arguments (```pi```, ```sqrt2```) were added to control the precision of the calculation since it varies by author and user. Additionally, the ```poles``` argument has been removed. For 3 Poles, see ```help(ta.ssf3)```. See ```help(ta.ssf)```.
* _Ehler's Super Smoother Filter (3 Poles)_ (**ssf3**): Was split from ```ta.ssf``` and also has addtional arguments arguments (```pi```, ```sqrt3```). See ```help(ta.ssf3)```.
* _Standard Deviation_ (**stdev**): To use ```ddof``` argument, also set ```talib=False```. The ```ddof``` argument is not available if you have TA Lib installed in your environment. Same goes for **variance**. See ```help(ta.stdev)```.
* _Variance_ (**variance**): To use ```ddof``` argument, also set ```talib=False```. The ```ddof``` argument is not available if you have TA Lib installed in your environment. Same goes for **stdev**. See ```help(ta.variance)```.
* _Volume Profile_ (**vp**): Calculation improvements. See [Pull Request #320](https://github.com/twopirllc/pandas-ta/pull/320) See ```help(ta.vp)```.
examples/AIExample.ipynb
+319 -307
View File
File diff suppressed because one or more lines are too long
examples/PandasTA_Strategy_Examples.ipynb
+67 -21
View File
File diff suppressed because one or more lines are too long
examples/VectorBT_Backtest_with_Pandas_TA.ipynb
+297 -297
View File
File diff suppressed because one or more lines are too long
examples/example.ipynb
+105 -100
View File
File diff suppressed because one or more lines are too long
examples/watchlist.py
+4 -2
View File
@@ -2,7 +2,6 @@
import datetime as dt
from pathlib import Path
from random import random
from typing import Tuple
import pandas as pd # pip install pandas
@@ -19,7 +18,10 @@ import alphaVantageAPI as AV # pip install alphaVantage-api
import pandas_ta as ta # pip install pandas_ta
def colors(colors: str = None, default: str = "GrRd"):
def colors(colors: str = None, default: str = "GrRd") -> dict:
"""A Helper Function to that returns a dict of 'common' color groups.
- Modify per use case or preferred theme
"""
aliases = {
# Pairs
"BkGy": ["black", "gray"],
pandas_ta/__init__.py
+5 -3
View File
@@ -33,7 +33,7 @@ Imports = {
"tqdm": find_spec("tqdm") is not None,
"vectorbt": find_spec("vectorbt") is not None,
"yfinance": find_spec("yfinance") is not None,
'polygon': find_spec('polygon') is not None,
"polygon": find_spec("polygon") is not None,
}
# Not ideal and not dynamic but it works.
@@ -58,8 +58,8 @@ Category = {
"alligator", "alma", "dema", "ema", "fwma", "hilo", "hl2", "hlc3",
"hma", "hwma", "ichimoku", "jma", "kama", "linreg", "mcgd", "midpoint",
"midprice", "ohlc4", "pwma", "rma", "sinwma", "sma", "smma", "ssf",
"supertrend", "swma", "t3", "tema", "trima", "vidya", "vwap", "vwma",
"wcp", "wma", "zlma"
"ssf3", "supertrend", "swma", "t3", "tema", "trima", "vidya", "vwap",
"vwma", "wcp", "wma", "zlma"
],
# Performance
"performance": ["log_return", "percent_return"],
@@ -117,4 +117,6 @@ RATE = {
"YEARLY": 1,
}
import numpy as np
import pandas as pd
from pandas_ta.core import *
pandas_ta/core.py
+23 -18
View File
@@ -1,17 +1,13 @@
# -*- coding: utf-8 -*-
from dataclasses import dataclass, field
from multiprocessing import cpu_count, Pool
from pathlib import Path
from time import perf_counter
from typing import List, Tuple
from warnings import simplefilter
import pandas as pd
from numpy import log10 as npLog10
from numpy import ndarray as npNdarray
from pandas.core.base import PandasObject
from pandas_ta import Category, Imports, version
from pandas_ta import Category, Imports, np, pd, version
from pandas_ta.candles.cdl_pattern import ALL_PATTERNS
from pandas_ta.candles import *
from pandas_ta.cycles import *
@@ -452,7 +448,7 @@ 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}' was not found in {cols}"
NOT_FOUND = f"[X] Ooops!!! It's {series not in df.columns}, the column named '{series}' was not found in {cols}"
return df.iloc[:, match[0]] if len(match) else print(NOT_FOUND)
def _indicators_by_category(self, name: str) -> list:
@@ -545,7 +541,7 @@ class AnalysisIndicators(BasePandasObject):
Returns nothing to the user. Either adds or removes constant ranges
from the working DataFrame.
"""
if isinstance(values, npNdarray) or isinstance(values, list):
if isinstance(values, np.ndarray) or isinstance(values, list):
if append:
for x in values:
self._df[f"{x}"] = x
@@ -736,7 +732,7 @@ class AnalysisIndicators(BasePandasObject):
_total_ta = len(ta)
with Pool(self.cores) as pool:
# Some magic to optimize chunksize for speed based on total ta indicators
_chunksize = mp_chunksize - 1 if mp_chunksize > _total_ta else int(npLog10(_total_ta)) + 1
_chunksize = mp_chunksize - 1 if mp_chunksize > _total_ta else int(np.log10(_total_ta)) + 1
if verbose:
print(f"[i] Multiprocessing {_total_ta} indicators with {_chunksize} chunks and {self.cores}/{cpu_count()} cpus.")
@@ -861,9 +857,13 @@ class AnalysisIndicators(BasePandasObject):
Exits if the DataFrame is empty or None
Otherwise it returns a DataFrame
"""
# ds = kwargs.pop("ds", "yahoo")
ds = f"{ds.lower()}" if ds is not None and isinstance(ds, str) else "yahoo"
# _frequencies = ["1s", "5s", "15s", "30s", "1m", "5m", "15m", "30m", "45m", "1h", "2h", "4h", "D", "W", "M"]
_ds = "yahoo"
ds = f"{ds.lower()}" if ds is not None and isinstance(ds, str) else _ds
strategy = kwargs.pop("strategy", None)
if isinstance(ticker, str):
tickers = [ticker]
# Fetch the Data
if ds == "polygon":
@@ -882,7 +882,8 @@ class AnalysisIndicators(BasePandasObject):
df.columns = df.columns.str.lower()
self._df = df
if strategy is not None: self.strategy(strategy, **kwargs)
# if strategy is not None: self.strategy(strategy, **kwargs)
if strategy is not None: return self.strategy(strategy, returns=True, **kwargs)
return df
@@ -918,9 +919,9 @@ class AnalysisIndicators(BasePandasObject):
result = ebsw(close=close, length=length, bars=bars, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def reflex(self, close=None, length=None, smooth=None, offset=None, **kwargs):
def reflex(self, close=None, length=None, smooth=None, alpha=None, pi=None, sqrt2=None, offset=None, **kwargs):
close = self._get_column(kwargs.pop("close", "close"))
result = reflex(close=close, length=length, smooth=smooth, offset=offset, **kwargs)
result = reflex(close=close, length=length, smooth=smooth, alpha=alpha, pi=pi, sqrt2=sqrt2, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
# Momentum
@@ -1315,9 +1316,14 @@ class AnalysisIndicators(BasePandasObject):
result = smma(close=close, length=length, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def ssf(self, length=None, poles=None, offset=None, **kwargs):
def ssf(self, length=None, everget=None, pi=None, sqrt2=None, offset=None, **kwargs):
close = self._get_column(kwargs.pop("close", "close"))
result = ssf(close=close, length=length, poles=poles, offset=offset, **kwargs)
result = ssf(close=close, length=length, everget=everget, pi=pi, sqrt2=sqrt2, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def ssf3(self, length=None, pi=None, sqrt3=None, offset=None, **kwargs):
close = self._get_column(kwargs.pop("close", "close"))
result = ssf3(close=close, length=length, pi=pi, sqrt3=sqrt3, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def supertrend(self, length=None, multiplier=None, offset=None, **kwargs):
@@ -1536,9 +1542,9 @@ class AnalysisIndicators(BasePandasObject):
result = supertrend(high=high, low=low, close=close, period=period, multiplier=multiplier, mamode=mamode, drift=drift, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def trendflex(self, close=None, length=None, smooth=None, offset=None, **kwargs):
def trendflex(self, close=None, length=None, smooth=None, alpha=None, pi=None, sqrt2=None, offset=None, **kwargs):
close = self._get_column(kwargs.pop("close", "close"))
result = trendflex(close=close, length=length, smooth=smooth, offset=offset, **kwargs)
result = trendflex(close=close, length=length, smooth=smooth, alpha=alpha, pi=pi, sqrt2=sqrt2, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
def tsignals(self, trend=None, asbool=None, trend_reset=None, trend_offset=None, offset=None, **kwargs):
@@ -1605,7 +1611,6 @@ class AnalysisIndicators(BasePandasObject):
def cross_value(self, value=None, above=True, asint=True, offset=None, **kwargs):
a = self._get_column(kwargs.pop("close", "a"))
# a = self._get_column(a, f"{a}")
result = cross_value(series_a=a, value=value, above=above, asint=asint, offset=offset, **kwargs)
return self._post_process(result, **kwargs)
pandas_ta/cycles/ebsw.py
+17 -26
View File
@@ -1,14 +1,5 @@
# -*- coding: utf-8 -*-
from numpy import cos as npCos
from numpy import exp as npExp
from numpy import nan as npNaN
from numpy import pi as npPi
from numpy import sin as npSin
from numpy import sqrt as npSqrt
from numpy import zeros as npZeros
from numpy import roll as npRoll
from numpy import mean as npMean
from pandas import Series
from pandas_ta import np, pd
from pandas_ta.utils import get_offset, verify_series
@@ -72,15 +63,15 @@ def ebsw(close, length=None, bars=None, offset=None, initial_version=False, **kw
# Calculate Result
m = close.size
result = [npNaN for _ in range(0, length - 1)] + [0]
result = [np.nan for _ in range(0, length - 1)] + [0]
for i in range(length, m):
# HighPass filter cyclic components whose periods are shorter than Duration input
alpha1 = (1 - npSin(360 / length)) / npCos(360 / length)
alpha1 = (1 - np.sin(360 / length)) / np.cos(360 / length)
hp = 0.5 * (1 + alpha1) * (close[i] - lastClose) + alpha1 * lastHP
# Smooth with a Super Smoother Filter from equation 3-3
a1 = npExp(-npSqrt(2) * npPi / bars)
b1 = 2 * a1 * npCos(npSqrt(2) * 180 / bars)
a1 = np.exp(-np.sqrt(2) * np.pi / bars)
b1 = 2 * a1 * np.cos(np.sqrt(2) * 180 / bars)
c2 = b1
c3 = -1 * a1 * a1
c1 = 1 - c2 - c3
@@ -92,7 +83,7 @@ def ebsw(close, length=None, bars=None, offset=None, initial_version=False, **kw
power_ = (filter_ * filter_ + filtHist[1] * filtHist[1] + filtHist[0] * filtHist[0]) / 3
# Normalize the Average Wave to Square Root of the Average Power
wave = wave / npSqrt(power_)
wave = wave / np.sqrt(power_)
# update storage, result
filtHist.append(filter_) # append new filter_ value
@@ -104,15 +95,15 @@ def ebsw(close, length=None, bars=None, offset=None, initial_version=False, **kw
else: # this version is the default version
# Instance Variables
lastHP = lastClose = 0
filtHist = npZeros(3)
result = [npNaN] * (length - 1) + [0]
filtHist = np.zeros(3)
result = [np.nan] * (length - 1) + [0]
# Calculate constants
angle = 2 * npPi / length
alpha1 = (1 - npSin(angle)) / npCos(angle)
ang = 2 ** .5 * npPi / bars
a1 = npExp(-ang)
c2 = 2 * a1 * npCos(ang)
angle = 2 * np.pi / length
alpha1 = (1 - np.sin(angle)) / np.cos(angle)
ang = 2 ** .5 * np.pi / bars
a1 = np.exp(-ang)
c2 = 2 * a1 * np.cos(ang)
c3 = -a1 ** 2
c1 = 1 - c2 - c3
@@ -120,12 +111,12 @@ def ebsw(close, length=None, bars=None, offset=None, initial_version=False, **kw
hp = 0.5 * (1 + alpha1) * (close[i] - lastClose) + alpha1 * lastHP
# Rotate filters to overwrite oldest value
filtHist = npRoll(filtHist, -1)
filtHist = np.roll(filtHist, -1)
filtHist[-1] = 0.5 * c1 * (hp + lastHP) + c2 * filtHist[1] + c3 * filtHist[0]
# Wave calculation
wave = npMean(filtHist)
rms = npSqrt(npMean(filtHist ** 2))
wave = np.mean(filtHist)
rms = np.sqrt(np.mean(filtHist ** 2))
wave = wave / rms
# Update past values
@@ -133,7 +124,7 @@ def ebsw(close, length=None, bars=None, offset=None, initial_version=False, **kw
lastClose = close[i]
result.append(wave)
ebsw = Series(result, index=close.index)
ebsw = pd.Series(result, index=close.index)
# Offset
if offset != 0:
pandas_ta/cycles/reflex.py
+52 -47
View File
@@ -1,15 +1,43 @@
# -*- coding: utf-8 -*-
from numpy import nan as npNaN
from numpy import cos as npCos
from numpy import exp as npExp
from numpy import full as npFull
from numpy import pi as npPI
from numpy import sqrt as npSqrt
from pandas import Series
from pandas_ta import np, pd
from pandas_ta.utils import get_offset, verify_series
try:
from numba import njit
except ImportError:
njit = lambda _: _
def reflex(close, length=None, smooth=None, alpha=None, offset=None, **kwargs):
@njit
def np_reflex(x: np.ndarray, n: int, k: int, alpha: float, pi: float, sqrt2: float):
m, ratio = x.size, 2 * sqrt2 / k
a = np.exp(-pi * ratio)
b = 2 * a * np.cos(180 * ratio)
c = a * a - b + 1
_f = np.zeros_like(x)
_ms = np.zeros_like(x)
result = np.zeros_like(x)
for i in range(2, m):
_f[i] = 0.5 * c * (x[i] + x[i - 1]) + b * _f[i - 1] - a * a * _f[i - 2]
for i in range(n, m):
slope = (_f[i - n] - _f[i]) / n
_sum = 0
for j in range(1, n):
_sum += _f[i] - _f[i - j] + j * slope
_sum /= n
_ms[i] = alpha * _sum * _sum + (1 - alpha) * _ms[i - 1]
if _ms[i] != 0.0:
result[i] = _sum / np.sqrt(_ms[i])
return result
def reflex(close, length=None, smooth=None, alpha=None, pi=None, sqrt2=None, offset=None, **kwargs):
"""Reflex (reflex)
John F. Ehlers introduced two indicators within the article
@@ -22,13 +50,20 @@ def reflex(close, length=None, smooth=None, alpha=None, offset=None, **kwargs):
a separate control parameter for the internal applied SuperSmoother.
Sources:
http://traders.com/Documentation/FEEDbk_docs/2020/02/TradersTips.html
https://www.prorealcode.com/prorealtime-indicators/reflex-and-trendflex-indicators-john-f-ehlers/
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 20
smooth (int): Period of internal SuperSmoother. Default: 20
alpha (float: Alpha weight of Difference Sums. Default: 0.04
alpha (float): Alpha weight of Difference Sums. Default: 0.04
pi (float): The value of PI to use. The default is Ehler's
truncated value 3.14159. Adjust the value for more precision.
Default: 3.14159
sqrt2 (float): The value of sqrt(2) to use. The default is Ehler's
truncated value 1.414. Adjust the value for more precision.
Default: 1.414
offset (int): How many periods to offset the result. Default: 0
Kwargs:
@@ -39,49 +74,19 @@ def reflex(close, length=None, smooth=None, alpha=None, offset=None, **kwargs):
pd.Series: New feature generated.
"""
# Validate arguments
close = verify_series(close, length)
length = int(length) if isinstance(length, int) and length > 0 else 20
smooth = int(smooth) if isinstance(smooth, int) and smooth > 0 else 20
alpha = float(alpha) if isinstance(alpha, float) and alpha > 0 else 0.04
pi = float(pi) if isinstance(pi, float) and pi > 0 else 3.14159
sqrt2 = float(sqrt2) if isinstance(sqrt2, float) and sqrt2 > 0 else 1.414
close = verify_series(close, max(length, smooth))
offset = get_offset(offset)
# Precalculations
sqrt2 = npSqrt(2)
m = close.size
a1 = npExp(-sqrt2 * npPI / smooth)
b1 = 2 * a1 * npCos(sqrt2 * 180 / smooth)
c2 = b1
c3 = -a1 * a1
c1 = 1 - c2 - c3
filter_ = npFull(m, 0)
ms = npFull(m, 0)
reflex = npFull(m, npNaN)
# Calculation
for i in range(2, m):
# Gently smooth the data in a SuperSmoother
filter_[i] = 0.5 * c1 * (close[i] + close[i - 1]) + c2 * filter_[i - 1] + c3 * filter_[i - 2]
# Length is assumed cycle period
slope = (filter_[i - length] - filter_[i]) / length
# Sum the differences
sum_ = 0
for count in range(1, length):
sum_ = sum_ + (filter_[i] + count * slope) - filter_[i - count]
sum_ = sum_ / length
# Normalize in terms of Standard Deviations
ms[i] = alpha * sum_ * sum_ + (1 - alpha) * ms[i - 1]
if ms[i] != 0:
reflex[i] = sum_ / npSqrt(ms[i])
else:
reflex[i] = sum_ / 0.00001
result = Series(reflex, index=close.index)
# Neutralize pre-roll phase
result.iloc[0:length] = npNaN
# Calculate Result
np_close = close.values
result = np_reflex(np_close, length, smooth, alpha, pi, sqrt2)
result[:length] = np.nan
result = pd.Series(result, index=close.index)
# Offset
if offset != 0:
pandas_ta/momentum/squeeze_pro.py
+8 -9
View File
@@ -1,6 +1,5 @@
# -*- coding: utf-8 -*-
from numpy import NaN as npNaN
from pandas import DataFrame
from pandas_ta import np, pd
from pandas_ta.momentum import mom
from pandas_ta.overlap import ema, sma
from pandas_ta.trend import decreasing, increasing
@@ -145,7 +144,7 @@ def squeeze_pro(high, low, close, bb_length=None, bb_std=None, kc_length=None, k
f"SQZPRO_OFF": squeeze_off_wide.astype(int) if asint else squeeze_off_wide,
f"SQZPRO_NO": no_squeeze.astype(int) if asint else no_squeeze,
}
df = DataFrame(data)
df = pd.DataFrame(data)
df.name = squeeze.name
df.category = squeeze.category = "momentum"
@@ -162,15 +161,15 @@ def squeeze_pro(high, low, close, bb_length=None, bb_std=None, kc_length=None, k
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)
pos_inc.replace(0, np.nan, inplace=True)
pos_dec.replace(0, np.nan, inplace=True)
neg_dec.replace(0, np.nan, inplace=True)
neg_inc.replace(0, np.nan, 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)
sqz_inc.replace(0, np.nan, inplace=True)
sqz_dec.replace(0, np.nan, inplace=True)
# Handle fills
if "fillna" in kwargs:
pandas_ta/overlap/__init__.py
+1
View File
@@ -24,6 +24,7 @@ from .sinwma import sinwma
from .sma import sma
from .smma import smma
from .ssf import ssf
from .ssf3 import ssf3
from .supertrend import supertrend
from .swma import swma
from .t3 import t3
pandas_ta/overlap/ema.py
+29 -11
View File
@@ -1,10 +1,27 @@
# -*- coding: utf-8 -*-
from numpy import nan as npNaN
from pandas_ta import Imports
from pandas_ta import Imports, np
from pandas_ta.utils import get_offset, verify_series
try:
from numba import njit
except ImportError:
njit = lambda _: _
def ema(close, length=None, talib=None, offset=None, **kwargs):
# Almost there
# @njit
# def np_ema(x: np.ndarray, n: int):
# m = x.size
# result = np.zeros(m)
# a = 1 / (n + 1)
# for i in range(1, m):
# result[i] = a * x[i - 1] + (1 - a) * x[i]
# result[0] = np.nan
# return result
# # return np_prepend(result, n - 1)
def ema(close, length=None, talib=None, presma=None, offset=None, **kwargs):
"""Exponential Moving Average (EMA)
The Exponential Moving Average is more responsive moving average compared to the
@@ -20,13 +37,14 @@ def ema(close, length=None, talib=None, offset=None, **kwargs):
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 10
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
talib (bool): If TA Lib is installed and talib=True, it returns the
TA Lib values. Default: True
presma (bool, optional): If True, uses SMA for initial value like TA Lib.
Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
adjust (bool, optional): Default: False
sma (bool, optional): If True, uses SMA for initial value. Default: True
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
@@ -35,11 +53,11 @@ def ema(close, length=None, talib=None, offset=None, **kwargs):
"""
# Validate Arguments
length = int(length) if length and length > 0 else 10
adjust = kwargs.pop("adjust", False)
sma = kwargs.pop("sma", True)
presma = bool(presma) if isinstance(presma, bool) else True
mode_tal = bool(talib) if isinstance(talib, bool) else True
close = verify_series(close, length)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
adjust = kwargs.pop("adjust", False)
if close is None: return
@@ -48,10 +66,10 @@ def ema(close, length=None, talib=None, offset=None, **kwargs):
from talib import EMA
ema = EMA(close, length)
else:
if sma:
if presma: # TA Lib implementation
close = close.copy()
sma_nth = close[0:length].mean()
close[:length - 1] = npNaN
close[:length - 1] = np.nan
close.iloc[length - 1] = sma_nth
ema = close.ewm(span=length, adjust=adjust).mean()
pandas_ta/overlap/sma.py
+31 -3
View File
@@ -1,6 +1,32 @@
# -*- coding: utf-8 -*-
from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
from pandas_ta import Imports, np, pd
from pandas_ta.utils import get_offset, np_prepend, verify_series
try:
from numba import njit
except ImportError:
njit = lambda _: _
@njit
def np_sma(x: np.ndarray, n: int):
"""https://github.com/numba/numba/issues/4119"""
result = np.convolve(np.ones(n) / n, x)[n - 1:1 - n]
return np_prepend(result, n - 1)
## SMA: Alternative Implementations
# @njit
# def np_sma(x: np.ndarray, n: int):
# result = np.convolve(x, np.ones(n), mode="valid") / n
# return np_prepend(result, n - 1)
# @njit
# def np_sma(x: np.ndarray, n: int):
# csum = np.cumsum(x, dtype=float)
# csum[n:] = csum[n:] - csum[:-n]
# result = csum[n - 1:] / n
# return np_prepend(result, n - 1)
def sma(close, length=None, talib=None, offset=None, **kwargs):
@@ -42,7 +68,9 @@ def sma(close, length=None, talib=None, offset=None, **kwargs):
from talib import SMA
sma = SMA(close, length)
else:
sma = close.rolling(length, min_periods=min_periods).mean()
np_close = close.values
sma = np_sma(np_close, length)
sma = pd.Series(sma, index=close.index)
# Offset
if offset != 0:
pandas_ta/overlap/ssf.py
+67 -47
View File
@@ -1,33 +1,73 @@
# -*- coding: utf-8 -*-
from numpy import cos as npCos
from numpy import exp as npExp
from numpy import nan as npNaN
from numpy import pi as npPi
from numpy import sqrt as npSqrt
from pandas_ta import np, pd
from pandas_ta.utils import get_offset, verify_series
try:
from numba import njit
except ImportError:
njit = lambda _: _
def ssf(close, length=None, poles=None, offset=None, **kwargs):
@njit
def np_ssf(x: np.ndarray, n: int, pi: float, sqrt2: float):
"""Ehler's Super Smoother Filter
http://traders.com/documentation/feedbk_docs/2014/01/traderstips.html
"""
m, ratio, result = x.size, sqrt2 / n, np.copy(x)
a = np.exp(-pi * ratio)
b = 2 * a * np.cos(180 * ratio)
c = a * a - b + 1
for i in range(2, m):
result[i] = 0.5 * c * (x[i] + x[i - 1]) + b * result[i - 1] \
- a * a * result[i - 2]
return result
@njit
def np_ssf_everget(x: np.ndarray, n: int, pi: float, sqrt2: float):
"""John F. Ehler's Super Smoother Filter by Everget (2 poles), Tradingview
https://www.tradingview.com/script/VdJy0yBJ-Ehlers-Super-Smoother-Filter/
"""
m, arg, result = x.size, pi * sqrt2 / n, np.copy(x)
a = np.exp(-arg)
b = 2 * a * np.cos(arg)
for i in range(2, m):
result[i] = 0.5 * (a * a - b + 1) * (x[i] + x[i - 1]) \
+ b * result[i - 1] - a * a * result[i - 2]
return result
def ssf(close, length=None, everget=None, pi=None, sqrt2=None, offset=None, **kwargs):
"""Ehler's Super Smoother Filter (SSF) © 2013
John F. Ehlers's solution to reduce lag and remove aliasing noise with his
research in aerospace analog filter design. This indicator comes with two
versions determined by the keyword poles. By default, it uses two poles but
there is an option for three poles. Since SSF is a (Resursive) Digital Filter,
the number of poles determine how many prior recursive SSF bars to include in
the design of the filter. So two poles uses two prior SSF bars and three poles
uses three prior SSF bars for their filter calculations.
research in aerospace analog filter design. This implementation had two
poles. Since SSF is a (Resursive) Digital Filter, the number of poles
determine how many prior recursive SSF bars to include in the filter design.
For Everget's calculation on TradingView, set arguments:
pi = np.pi, sqrt2 = np.sqrt(2)
Sources:
http://www.stockspotter.com/files/PredictiveIndicators.pdf
http://traders.com/documentation/feedbk_docs/2014/01/traderstips.html
https://www.tradingview.com/script/VdJy0yBJ-Ehlers-Super-Smoother-Filter/
https://www.mql5.com/en/code/588
https://www.mql5.com/en/code/589
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 10
poles (int): The number of poles to use, either 2 or 3. Default: 2
length (int): It's period. Default: 20
everget (bool): Everget's implementation of ssf that uses pi instead of
180 for the b factor of ssf. Default: False
pi (float): The value of PI to use. The default is Ehler's
truncated value 3.14159. Adjust the value for more precision.
Default: 3.14159
sqrt2 (float): The value of sqrt(2) to use. The default is Ehler's
truncated value 1.414. Adjust the value for more precision.
Default: 1.414
offset (int): How many periods to offset the result. Default: 0
Kwargs:
@@ -38,42 +78,22 @@ def ssf(close, length=None, poles=None, offset=None, **kwargs):
pd.Series: New feature generated.
"""
# Validate Arguments
length = int(length) if isinstance(length, int) and length > 0 else 10
poles = int(poles) if isinstance(poles, int) and poles in [2, 3] else 2
length = int(length) if isinstance(length, int) and length > 0 else 20
everget = bool(everget) if isinstance(everget, bool) else False
pi = float(pi) if isinstance(pi, float) and pi > 0 else 3.14159
sqrt2 = float(sqrt2) if isinstance(sqrt2, float) and sqrt2 > 0 else 1.414
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
m = close.size
ssf = close.copy()
if poles == 3:
x = npPi / length # x = PI / n
a0 = npExp(-x) # e^(-x)
b0 = 2 * a0 * npCos(npSqrt(3) * x) # 2e^(-x)*cos(3^(.5) * x)
c0 = a0 * a0 # e^(-2x)
c4 = c0 * c0 # e^(-4x)
c3 = -c0 * (1 + b0) # -e^(-2x) * (1 + 2e^(-x)*cos(3^(.5) * x))
c2 = c0 + b0 # e^(-2x) + 2e^(-x)*cos(3^(.5) * x)
c1 = 1 - c2 - c3 - c4
for i in range(poles, m):
ssf.iloc[i] = c1 * close.iloc[i] + c2 * ssf.iloc[i - 1] + c3 * ssf.iloc[i - 2] + c4 * ssf.iloc[i - 3]
else: # poles == 2
x = npPi * npSqrt(2) / length # x = PI * 2^(.5) / n
a0 = npExp(-x) # e^(-x)
a1 = -a0 * a0 # -e^(-2x)
b1 = 2 * a0 * npCos(x) # 2e^(-x)*cos(x)
c1 = 1 - a1 - b1 # e^(-2x) - 2e^(-x)*cos(x) + 1
for i in range(poles, m):
ssf.iloc[i] = c1 * close.iloc[i] + b1 * ssf.iloc[i - 1] + a1 * ssf.iloc[i - 2]
ssf.iloc[:length] = npNaN
np_close = close.values
if everget:
ssf = np_ssf_everget(np_close, length, pi, sqrt2)
else:
ssf = np_ssf(np_close, length, pi, sqrt2)
ssf = pd.Series(ssf, index=close.index)
# Offset
if offset != 0:
@@ -86,7 +106,7 @@ def ssf(close, length=None, poles=None, offset=None, **kwargs):
ssf.fillna(method=kwargs["fill_method"], inplace=True)
# Name & Category
ssf.name = f"SSF_{length}_{poles}"
ssf.name = f"SSF{'e' if everget else ''}_{length}"
ssf.category = "overlap"
return ssf
pandas_ta/overlap/ssf3.py
+93
View File
@@ -0,0 +1,93 @@
# -*- coding: utf-8 -*-
from pandas_ta import np, pd
from pandas_ta.utils import get_offset, verify_series
try:
from numba import njit
except ImportError:
njit = lambda _: _
@njit
def np_ssf3(x: np.ndarray, n: int, pi: float, sqrt3: float):
"""John F. Ehler's Super Smoother Filter by Everget (3 poles), Tradingview
https://www.tradingview.com/script/VdJy0yBJ-Ehlers-Super-Smoother-Filter/"""
m, result = x.size, np.copy(x)
a = np.exp(-pi / n)
b = 2 * a * np.cos(-pi * sqrt3 / n)
c = a * a
d4 = c * c
d3 = -c * (1 + b)
d2 = b + c
d1 = 1 - d2 - d3 - d4
for i in range(3, m):
result[i] = d1 * x[i] + d2 * result[i - 1] \
+ d3 * result[i - 2] + d4 * result[i - 3]
return result
def ssf3(close, length=None, pi=None, sqrt3=None, offset=None, **kwargs):
"""Ehler's 3 Pole Super Smoother Filter (SSF) © 2013
John F. Ehlers's solution to reduce lag and remove aliasing noise with his
research in aerospace analog filter design. This is implementation has three
poles. Since SSF is a (Resursive) Digital Filter, the number of poles
determine how many prior recursive SSF bars to include in the filter design.
For Everget's calculation on TradingView, set arguments:
pi = np.pi, sqrt3 = 1.738
Sources:
https://www.tradingview.com/script/VdJy0yBJ-Ehlers-Super-Smoother-Filter/
https://www.mql5.com/en/code/589
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 20
pi (float): The value of PI to use. The default is Ehler's
truncated value 3.14159. Adjust the value for more precision.
Default: 3.14159
sqrt3 (float): The value of sqrt(3) to use. The default is Ehler's
truncated value 1.732. Adjust the value for more precision.
Default: 1.732
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.Series: New feature generated.
"""
# Validate Arguments
length = int(length) if isinstance(length, int) and length > 0 else 20
pi = float(pi) if isinstance(pi, float) and pi > 0 else 3.14159
sqrt3 = float(sqrt3) if isinstance(sqrt3, float) and sqrt3 > 0 else 1.732
close = verify_series(close, length)
offset = get_offset(offset)
if close is None: return
# Calculate Result
np_close = close.values
ssf = np_ssf3(np_close, length, pi, sqrt3)
ssf = pd.Series(ssf, index=close.index)
# Offset
if offset != 0:
ssf = ssf.shift(offset)
# Handle fills
if "fillna" in kwargs:
ssf.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
ssf.fillna(method=kwargs["fill_method"], inplace=True)
# Name & Category
ssf.name = f"SSF3_{length}"
ssf.category = "overlap"
return ssf
pandas_ta/trend/cksp.py
+1 -1
View File
@@ -40,6 +40,7 @@ def cksp(high, low, close, p=None, x=None, q=None, tvmode=None, offset=None, **k
pd.DataFrame: long and short columns.
"""
# Validate Arguments
tvmode = tvmode if isinstance(tvmode, bool) else True
p = int(p) if p and p > 0 else 10
x = float(x) if x and x > 0 else 1 if tvmode is True else 3
q = int(q) if q and q > 0 else 9 if tvmode is True else 20
@@ -51,7 +52,6 @@ def cksp(high, low, close, p=None, x=None, q=None, tvmode=None, offset=None, **k
if high is None or low is None or close is None: return
offset = get_offset(offset)
tvmode = tvmode if isinstance(tvmode, bool) else True
mamode = "rma" if tvmode is True else "sma"
# Calculate Result
pandas_ta/trend/trendflex.py
+59 -48
View File
@@ -1,17 +1,44 @@
# -*- coding: utf-8 -*-
from numpy import nan as npNaN
from numpy import cos as npCos
from numpy import exp as npExp
from numpy import full as npFull
from numpy import pi as npPI
from numpy import sqrt as npSqrt
from numpy import sqrt as npSqrt
from pandas import Series
from pandas_ta import np, pd
from pandas_ta.utils import get_offset, verify_series
try:
from numba import njit
except ImportError:
njit = lambda _: _
def trendflex(close, length=None, smooth=None, alpha=None, offset=None, **kwargs):
"""Trendflex (trendflex)
@njit
def np_trendflex(x: np.ndarray, n: int, k: int, alpha: float, pi: float, sqrt2: float):
"""Ehler's Trendflex
http://traders.com/Documentation/FEEDbk_docs/2020/02/TradersTips.html"""
m, ratio = x.size, 2 * sqrt2 / k
a = np.exp(-pi * ratio)
b = 2 * a * np.cos(180 * ratio)
c = a * a - b + 1
_f = np.zeros_like(x)
_ms = np.zeros_like(x)
result = np.zeros_like(x)
for i in range(2, m):
_f[i] = 0.5 * c * (x[i] + x[i - 1]) + b * _f[i - 1] - a * a * _f[i - 2]
for i in range(n, m):
_sum = 0
for j in range(1, n):
_sum += _f[i] - _f[i - j]
_sum /= n
_ms[i] = alpha * _sum * _sum + (1 - alpha) * _ms[i - 1]
if _ms[i] != 0.0:
result[i] = _sum / np.sqrt(_ms[i])
return result
def trendflex(close, length=None, smooth=None, alpha=None, pi=None, sqrt2=None, offset=None, **kwargs):
"""Trendflex (TRENDFLEX)
John F. Ehlers introduced two indicators within the article "Reflex: A New
Zero-Lag Indicator” in February 2020, TASC magazine. One of which is the
@@ -23,14 +50,21 @@ def trendflex(close, length=None, smooth=None, alpha=None, offset=None, **kwargs
a separate control parameter for the internal applied SuperSmoother.
Sources:
http://traders.com/Documentation/FEEDbk_docs/2020/02/TradersTips.html
https://www.prorealcode.com/prorealtime-indicators/reflex-and-trendflex-indicators-john-f-ehlers/
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 20
length (int): It's period. Default: 20
smooth (int): Period of internal SuperSmoother Default: 20
alpha (float: Alpha weight of Difference Sums. Default: 0.04
offset (int): How many periods to offset the result. Default: 0
alpha (float): Alpha weight of Difference Sums. Default: 0.04
pi (float): The value of PI to use. The default is Ehler's
truncated value 3.14159. Adjust the value for more precision.
Default: 3.14159
sqrt2 (float): The value of sqrt(2) to use. The default is Ehler's
truncated value 1.414. Adjust the value for more precision.
Default: 1.414
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
@@ -43,43 +77,20 @@ def trendflex(close, length=None, smooth=None, alpha=None, offset=None, **kwargs
length = int(length) if isinstance(length, int) and length > 0 else 20
smooth = int(smooth) if isinstance(smooth, int) and smooth > 0 else 20
alpha = float(alpha) if isinstance(alpha, float) and alpha > 0 else 0.04
close = verify_series(close, length)
pi = float(pi) if isinstance(pi, float) and pi > 0 else 3.14159
sqrt2 = float(sqrt2) if isinstance(sqrt2, float) and sqrt2 > 0 else 1.414
close = verify_series(close, max(length, smooth))
offset = get_offset(offset)
# Precalculations
sqrt2 = npSqrt(2)
m = close.size
a1 = npExp(-sqrt2 * npPI / smooth)
b1 = 2 * a1 * npCos(sqrt2 * 180 / smooth)
c2 = b1
c3 = -a1 * a1
c1 = 1 - c2 - c3
filter_ = npFull(m, 0)
ms = npFull(m, 0)
trendflex = list(filter_)
if close is None: return
# Calculation
for i in range(2, m):
# Gently smooth the data in a SuperSmoother
filter_[i] = 0.5 * c1 * (close[i] + 0.5 * close[i - 1]) + c2 * filter_[i - 1] + c3 * filter_[i - 2]
# Sum the differences
sum_ = 0
for count in range(1, length):
sum_ = sum_ + filter_[i] - filter_[i - count]
sum_ = sum_ / length
# Normalize in terms of Standard Deviations
ms[i] = alpha * sum_ * sum_ + (1 - alpha) * ms[i - 1]
if ms[i] != 0:
trendflex[i] = sum_ / npSqrt(ms[i])
else:
trendflex[i] = sum_ / 0.00001
result = Series(trendflex, index=close.index)
# Neutralize pre-roll phase
result.iloc[0:length] = npNaN
# Calculate Result
np_close = close.values
result = np_trendflex(np_close, length, smooth, alpha, pi, sqrt2)
# print(f"\nresult:\n{result}\n")
result[:length] = np.nan
# print(f"result:\n{result}")
result = pd.Series(result, index=close.index)
# Offset
if offset != 0:
@@ -92,7 +103,7 @@ def trendflex(close, length=None, smooth=None, alpha=None, offset=None, **kwargs
result.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
result.name = f"TRENDFLEX_{length}_{smooth}"
result.name = f"TRENDFLEX_{length}_{smooth}_{alpha}"
result.category = "trend"
return result
pandas_ta/utils/__init__.py
+2 -1
View File
@@ -3,7 +3,8 @@ from ._candles import *
from ._core import *
from ._math import *
from ._metrics import *
from ._numba import *
from ._signals import *
from ._stats import *
from ._time import *
from .data import *
from .data import *
pandas_ta/utils/_core.py
+52 -2
View File
@@ -7,6 +7,7 @@ from numpy import argmax, argmin
from pandas import DataFrame, Series
from pandas.api.types import is_datetime64_any_dtype
from pandas_ta import Imports
from pandas_ta import pd
def _camelCase2Title(x: str):
@@ -52,7 +53,8 @@ def is_percent(x: int or float) -> bool:
def non_zero_range(high: Series, low: Series) -> Series:
"""Returns the difference of two series and adds epsilon to any zero values. This occurs commonly in crypto data when 'high' = 'low'."""
"""Returns the difference of two series and adds epsilon to any zero values.
This occurs commonly in crypto data when 'high' = 'low'."""
diff = high - low
if diff.eq(0).any().any():
diff += sflt.epsilon
@@ -142,4 +144,52 @@ def verify_series(series: Series, min_length: int = None) -> Series:
"""If a Pandas Series and it meets the min_length of the indicator return it."""
has_length = min_length is not None and isinstance(min_length, int)
if series is not None and isinstance(series, Series):
return None if has_length and series.size < min_length else series
return None if has_length and series.size < min_length else series
def performance(df:DataFrame,
excluded:list = None, other:list = None, top:int = None,
sortby:str = "secs", ascending:bool = False,
gradient:int = False, places:int = 5
) -> DataFrame:
if df.empty: return
top = int(top) if isinstance(top, int) and top > 0 else None
data = []
df = df.copy()
if isinstance(excluded, list) and len(excluded) > 0:
indicators = df.ta.indicators(as_list=True, exclude=excluded)
else:
indicators = df.ta.indicators(as_list=True)
_index_name = "Indicator"
if len(indicators):
for indicator in indicators:
result = df.ta(indicator, timed=True)
ms = float(result.timed.split(" ")[0].split(" ")[0])
data.append({_index_name: indicator, "secs": round(0.001 * ms, places), "ms": ms})
if isinstance(other, list) and len(other) > 0:
for indicator in other:
result = df.ta(indicator, timed=True)
ms = float(result.timed.split(" ")[0].split(" ")[0])
data.append({_index_name: indicator, "secs": round(0.001 * ms, places), "ms": ms})
tdf = DataFrame.from_dict(data)
tdf.set_index(_index_name, inplace=True)
tdf.sort_values(by=sortby, ascending=ascending, inplace=True)
total_timedf = pd.DataFrame(tdf.describe().loc[['min', '50%', 'mean', 'max']]).T
total_timedf["total"] = tdf.sum(axis=0).T
_div = "=" * 60
_observations = f" Observations: {df.shape[0]}"
_quick_slow = "Quickest" if ascending else "Slowest"
_title = f" {_quick_slow} Indicators"
_perfstats = f"Time Stats:\n{total_timedf.T}"
if top:
_title = f" {_quick_slow} {top} Indicators [{tdf.shape[0]}]"
tdf = tdf.head(top)
print(f"\n{_div}\n{_title}\n{_observations}\n{_div}\n{tdf}\n\n{_div}\n{_perfstats}\n\n{_div}\n")
if isinstance(gradient, bool) and gradient: return tdf.style.background_gradient("autumn_r")
return tdf
pandas_ta/utils/_math.py
+30 -45
View File
@@ -5,31 +5,16 @@ from operator import mul
from sys import float_info as sflt
from typing import List, Optional, Tuple
from numpy import ones, triu
from numpy import all as npAll
from numpy import append as npAppend
from numpy import array as npArray
from numpy import corrcoef as npCorrcoef
from numpy import dot as npDot
from numpy import fabs as npFabs
from numpy import exp as npExp
from numpy import log as npLog
from numpy import nan as npNaN
from numpy import ndarray as npNdArray
from numpy import seterr
from numpy import sqrt as npSqrt
from numpy import sum as npSum
from pandas import DataFrame, Series
from pandas_ta import Imports
from pandas_ta import Imports, np
from ._core import verify_series
def combination(**kwargs: dict) -> int:
"""https://stackoverflow.com/questions/4941753/is-there-a-math-ncr-function-in-python"""
n = int(npFabs(kwargs.pop("n", 1)))
r = int(npFabs(kwargs.pop("r", 0)))
n = int(np.fabs(kwargs.pop("n", 1)))
r = int(np.fabs(kwargs.pop("r", 0)))
if kwargs.pop("repetition", False) or kwargs.pop("multichoose", False):
n = n + r - 1
@@ -67,9 +52,9 @@ def erf(x: Tuple[int, float]):
return sign * y # erf(-x) = -erf(x)
def fibonacci(n: int = 2, **kwargs: dict) -> npNdArray:
def fibonacci(n: int = 2, **kwargs: dict) -> np.ndarray:
"""Fibonacci Sequence as a numpy array"""
n = int(npFabs(n)) if n >= 0 else 2
n = int(np.fabs(n)) if n >= 0 else 2
zero = kwargs.pop("zero", False)
if zero:
@@ -78,14 +63,14 @@ def fibonacci(n: int = 2, **kwargs: dict) -> npNdArray:
n -= 1
a, b = 1, 1
result = npArray([a])
result = np.array([a])
for _ in range(0, n):
a, b = b, a + b
result = npAppend(result, a)
result = np.append(result, a)
weighted = kwargs.pop("weighted", False)
if weighted:
fib_sum = npSum(result)
fib_sum = np.sum(result)
if fib_sum > 0:
return result / fib_sum
else:
@@ -103,13 +88,13 @@ def geometric_mean(series: Series) -> float:
has_zeros = 0 in series.values
if has_zeros:
series = series.fillna(0) + 1
if npAll(series > 0):
if np.all(series > 0):
mean = series.prod() ** (1 / n)
return mean if not has_zeros else mean - 1
return 0
def hpoly(array: npArray, x: Tuple[int, float]) -> float:
def hpoly(array: np.array, x: Tuple[int, float]) -> float:
"""Horner Calculation for Polynomial Evaluation (hpoly)
array: np.array of polynomial coefficients
@@ -126,8 +111,8 @@ def hpoly(array: npArray, x: Tuple[int, float]) -> float:
hpoly(coeffs_0, x) => -1224.25
hpoly(coeffs_1, x) or hpoly(coeffs_2, x) => -1224.25 # Faster
"""
if not isinstance(array, npNdArray):
array = npArray(array)
if not isinstance(array, np.ndarray):
array = np.array(array)
m, y = array.size, array[0]
@@ -157,12 +142,12 @@ def log_geometric_mean(series: Series) -> float:
if n < 2: return 0
else:
series = series.fillna(0) + 1
if npAll(series > 0):
return npExp(npLog(series).sum() / n) - 1
if np.all(series > 0):
return np.exp(np.log(series).sum() / n) - 1
return 0
def pascals_triangle(n: int = None, **kwargs: dict) -> npNdArray:
def pascals_triangle(n: int = None, **kwargs: dict) -> np.ndarray:
"""Pascal's Triangle
Returns a numpy array of the nth row of Pascal's Triangle.
@@ -170,11 +155,11 @@ def pascals_triangle(n: int = None, **kwargs: dict) -> npNdArray:
=> weighted: [0.0625, 0.25, 0.375, 0.25, 0.0625]
=> inverse weighted: [0.9375, 0.75, 0.625, 0.75, 0.9375]
"""
n = int(npFabs(n)) if n is not None else 0
n = int(np.fabs(n)) if n is not None else 0
# Calculation
triangle = npArray([combination(n=n, r=i) for i in range(0, n + 1)])
triangle_sum = npSum(triangle)
triangle = np.array([combination(n=n, r=i) for i in range(0, n + 1)])
triangle_sum = np.sum(triangle)
triangle_weights = triangle / triangle_sum
inverse_weights = 1 - triangle_weights
@@ -211,7 +196,7 @@ def symmetric_triangle(n: int = None, **kwargs: dict) -> Optional[List[int]]:
n=4 => triangle: [1, 2, 2, 1]
=> weighted: [0.16666667 0.33333333 0.33333333 0.16666667]
"""
n = int(npFabs(n)) if n is not None else 2
n = int(np.fabs(n)) if n is not None else 2
triangle = None
if n == 2:
@@ -228,17 +213,17 @@ def symmetric_triangle(n: int = None, **kwargs: dict) -> Optional[List[int]]:
triangle += front[::-1]
if kwargs.pop("weighted", False) and isinstance(triangle, list):
triangle_sum = npSum(triangle)
triangle_sum = np.sum(triangle)
triangle_weights = triangle / triangle_sum
return triangle_weights
return triangle
def weights(w: npNdArray):
def weights(w: np.ndarray):
"""Calculates the dot product of weights with values x"""
def _dot(x):
return npDot(w, x)
return np.dot(w, x)
return _dot
@@ -265,7 +250,7 @@ def df_error_analysis(dfA: DataFrame, dfB: DataFrame, **kwargs: dict) -> DataFra
diff.plot(kind="kde")
if kwargs.pop("triangular", False):
return corr.where(triu(ones(corr.shape)).astype(bool))
return corr.where(np.triu(np.ones(corr.shape)).astype(bool))
return corr
@@ -273,13 +258,13 @@ def df_error_analysis(dfA: DataFrame, dfB: DataFrame, **kwargs: dict) -> DataFra
# PRIVATE
def _linear_regression_np(x: Series, y: Series) -> dict:
"""Simple Linear Regression in Numpy for two 1d arrays for environments without the sklearn package."""
result = {"a": npNaN, "b": npNaN, "r": npNaN, "t": npNaN, "line": npNaN}
result = {"a": np.nan, "b": np.nan, "r": np.nan, "t": np.nan, "line": np.nan}
x_sum = x.sum()
y_sum = y.sum()
if int(x_sum) != 0:
# 1st row, 2nd col value corr(x, y)
r = npCorrcoef(x, y)[0, 1]
r = np.corrcoef(x, y)[0, 1]
m = x.size
r_mix = m * (x * y).sum() - x_sum * y_sum
@@ -287,14 +272,14 @@ def _linear_regression_np(x: Series, y: Series) -> dict:
a = y.mean() - b * x.mean()
line = a + b * x
_np_err = seterr()
seterr(divide="ignore", invalid="ignore")
_np_err = np.seterr()
np.seterr(divide="ignore", invalid="ignore")
result = {
"a": a, "b": b, "r": r,
"t": r / npSqrt((1 - r * r) / (m - 2)),
"t": r / np.sqrt((1 - r * r) / (m - 2)),
"line": line,
}
seterr(divide=_np_err["divide"], invalid=_np_err["invalid"])
np.seterr(divide=_np_err["divide"], invalid=_np_err["invalid"])
return result
@@ -310,7 +295,7 @@ def _linear_regression_sklearn(x: Series, y: Series) -> dict:
result = {
"a": a, "b": b, "r": r,
"t": r / npSqrt((1 - r * r) / (x.size - 2)),
"t": r / np.sqrt((1 - r * r) / (x.size - 2)),
"line": a + b * x
}
return result
pandas_ta/utils/_metrics.py
+9 -12
View File
@@ -1,15 +1,12 @@
# -*- coding: utf-8 -*-
from typing import Tuple
from numpy import log as npLog
from numpy import nan as npNaN
from numpy import sqrt as npSqrt
from pandas import Series, Timedelta
from ._core import verify_series
from ._time import total_time
from ._math import linear_regression, log_geometric_mean
from pandas_ta import RATE
from ._time import total_time
from pandas_ta import RATE, np
from pandas_ta.performance import drawdown, log_return, percent_return
@@ -70,8 +67,8 @@ def downside_deviation(returns: Series, benchmark_rate: float = 0.0, tf: str = "
downside = adjusted_benchmark_rate - returns
downside_sum_of_squares = (downside[downside > 0] ** 2).sum()
downside_deviation = npSqrt(downside_sum_of_squares / (returns.shape[0] - 1))
return downside_deviation * npSqrt(days_per_year)
downside_deviation = np.sqrt(downside_sum_of_squares / (returns.shape[0] - 1))
return downside_deviation * np.sqrt(days_per_year)
def jensens_alpha(returns: Series, benchmark_returns: Series) -> float:
@@ -99,7 +96,7 @@ def log_max_drawdown(close: Series) -> float:
>>> result = ta.log_max_drawdown(close)
"""
close = verify_series(close)
log_return = npLog(close.iloc[-1]) - npLog(close.iloc[0])
log_return = np.log(close.iloc[-1]) - np.log(close.iloc[0])
return log_return - max_drawdown(close, method="log")
@@ -155,7 +152,7 @@ def optimal_leverage(
# sharpe = sharpe_ratio(close, benchmark_rate=benchmark_rate, log=log, use_cagr=use_cagr, period=period)
period_mu = period * returns.mean()
period_std = npSqrt(period) * returns.std()
period_std = np.sqrt(period) * returns.std()
mean_excess_return = period_mu - benchmark_rate
# sharpe = mean_excess_return / period_std
@@ -177,7 +174,7 @@ def pure_profit_score(close: Series) -> Tuple[float, int]:
close_index = Series(0, index=close.reset_index().index)
r = linear_regression(close_index, close)["r"]
if r is not npNaN:
if r is not np.nan:
return r * cagr(close)
return 0
@@ -204,7 +201,7 @@ def sharpe_ratio(close: Series, benchmark_rate: float = 0.0, log: bool = False,
return cagr(close) / volatility(close, returns, log=log)
else:
period_mu = period * returns.mean()
period_std = npSqrt(period) * returns.std()
period_std = np.sqrt(period) * returns.std()
return (period_mu - benchmark_rate) / period_std
@@ -253,5 +250,5 @@ def volatility(close: Series, tf: str = "years", returns: bool = False, log: boo
# factor = returns.shape[0] / total_time(returns, tf)
# if kwargs.pop("nearest_day", False) and tf.lower() == "years":
# factor = int(factor + 1)
# return npSqrt(factor) * returns.std()
# return np.sqrt(factor) * returns.std()
return returns
pandas_ta/utils/_numba.py
+57
View File
@@ -0,0 +1,57 @@
# -*- coding: utf-8 -*-
from pandas_ta import np
# try:
# from numba import jit, njit
# except ImportError as e:
# from pandas_ta.utils._shim import jit, njit
try:
from numba import njit
except ImportError:
njit = lambda x: x
# Utilities
@njit
def np_prepend(x: np.ndarray, n: int, value=np.nan):
"""Append array x to an array of values, typically nan."""
return np.append(np.array([value] * n), x)
@njit
def np_shift(x: np.ndarray, n: int, value=np.nan):
"""np shift
shift5 - preallocate empty array and assign slice by chrisaycock
https://stackoverflow.com/questions/30399534/shift-elements-in-a-numpy-array
"""
result = np.empty_like(x)
if n > 0:
result[:n] = value
result[n:] = x[:-n]
elif n < 0:
result[n:] = value
result[:n] = x[-n:]
else:
result[:] = x
return result
# Uncategorized
# @njit
# def np_roofing_filter(x: np.ndarray, n: int, k: int, pi: float, sqrt2: float):
# """Ehler's Roofing Filter (INCOMPLETE)
# http://traders.com/documentation/feedbk_docs/2014/01/traderstips.html"""
# m, hp = x.size, np.copy(x)
# # a = exp(-pi * sqrt(2) / n)
# # b = 2 * a * cos(180 * sqrt(2) / n)
# rsqrt2 = 1 / np.sqrt2
# a = (np.cos(rsqrt2 * 360 / n) + np.sin(rsqrt2 * 360 / n) - 1)
# a /= np.cos(rsqrt2 * 360 / n)
# b, c = 1 - a, (1 - a / 2)
# for i in range(2, m):
# hp = c * c * (x[i] - 2 * x[i - 1] + x[i - 2]) \
# + 2 * b * hp[i - 1] - b * b * hp[i - 2]
# result = np_ssf(hp, k, pi, rsqrt2)
# return result
pandas_ta/utils/_stats.py
+14 -20
View File
@@ -1,14 +1,7 @@
# -*- coding: utf-8 -*-
from typing import Tuple
from numpy import array as npArray
from numpy import infty as npInfty
from numpy import log as npLog
from numpy import nan as npNaN
from numpy import pi as npPi
from numpy import sqrt as npSqrt
from pandas_ta import Imports
from pandas_ta import Imports, np
from ._math import hpoly
@@ -16,12 +9,12 @@ def _gaussian_poly_coefficients():
"""Three pairs of Polynomial Approximation Coefficients
for the Gaussian Normal CDF"""
p0 = npArray([
p0 = np.array([
-5.99633501014107895267E1, 9.80010754185999661536E1,
-5.66762857469070293439E1, 1.39312609387279679503E1,
-1.23916583867381258016E0
])
q0 = npArray([
q0 = np.array([
1.00000000000000000000E0, 1.95448858338141759834E0,
4.67627912898881538453E0, 8.63602421390890590575E1,
-2.25462687854119370527E2, 2.00260212380060660359E2,
@@ -29,14 +22,14 @@ def _gaussian_poly_coefficients():
-1.18331621121330003142E0
])
p1 = npArray([
p1 = np.array([
4.05544892305962419923E0, 3.15251094599893866154E1,
5.71628192246421288162E1, 4.40805073893200834700E1,
1.46849561928858024014E1, 2.18663306850790267539E0,
-1.40256079171354495875E-1, -3.50424626827848203418E-2,
-8.57456785154685413611E-4
])
q1 = npArray([
q1 = np.array([
1.00000000000000000000E0, 1.57799883256466749731E1,
4.53907635128879210584E1, 4.13172038254672030440E1,
1.50425385692907503408E1, 2.50464946208309415979E0,
@@ -44,14 +37,14 @@ def _gaussian_poly_coefficients():
-9.33259480895457427372E-4
])
p2 = npArray([
p2 = np.array([
3.23774891776946035970E0, 6.91522889068984211695E0,
3.93881025292474443415E0, 1.33303460815807542389E0,
2.01485389549179081538E-1, 1.23716634817820021358E-2,
3.01581553508235416007E-4, 2.65806974686737550832E-6,
6.23974539184983293730E-9
])
q2 = npArray([
q2 = np.array([
1.00000000000000000000E0, 6.02427039364742014255E0,
3.67983563856160859403E0, 1.37702099489081330271E0,
2.16236993594496635890E-1, 1.34204006088543189037E-2,
@@ -80,13 +73,14 @@ def inv_norm(value: Tuple[float, int]) -> Tuple[float, None]:
negate = True
v = value
if v == 0.0: return -npInfty
if v == 1.0: return npInfty
if v < 0.0 or value > 1.0: return npNaN
# if v == 0.0: return -npInfty
if v == 0.0: return -np.infty
if v == 1.0: return np.infty
if v < 0.0 or value > 1.0: return np.nan
p0, q0, p1, q1, p2, q2 = _gaussian_poly_coefficients()
sqrt2pi = npSqrt(2 * npPi)
sqrt2pi = np.sqrt(2 * np.pi)
threshold = 0.13533528323661269189
if v > 1.0 - threshold:
v, negate = 1.0 - v, False
@@ -99,8 +93,8 @@ def inv_norm(value: Tuple[float, int]) -> Tuple[float, None]:
y *= sqrt2pi
return y
y = npSqrt(-2.0 * npLog(v))
y0 = y - npLog(y) / y
y = np.sqrt(-2.0 * np.log(v))
y0 = y - np.log(y) / y
z = 1.0 / y
if y < 8.0:
pandas_ta/utils/_time.py
+19 -10
View File
@@ -1,14 +1,13 @@
# -*- coding: utf-8 -*-
from datetime import datetime
from time import localtime, perf_counter
from typing import Tuple
from typing import Tuple, Union
from pandas import DataFrame, Timestamp
from pandas_ta import EXCHANGE_TZ, RATE
from pandas import Timestamp
from pandas_ta import EXCHANGE_TZ, pd, RATE
def df_dates(df: DataFrame, dates: Tuple[str, list] = None) -> DataFrame:
def df_dates(df: pd.DataFrame, dates: Tuple[str, list] = None) -> pd.DataFrame:
"""Yields the DataFrame with the given dates"""
if dates is None: return None
if not isinstance(dates, list):
@@ -16,14 +15,14 @@ def df_dates(df: DataFrame, dates: Tuple[str, list] = None) -> DataFrame:
return df[df.index.isin(dates)]
def df_month_to_date(df: DataFrame) -> DataFrame:
def df_month_to_date(df: pd.DataFrame) -> pd.DataFrame:
"""Yields the Month-to-Date (MTD) DataFrame"""
in_mtd = df.index >= Timestamp.now().strftime("%Y-%m-01")
if any(in_mtd): return df[in_mtd]
return df
def df_quarter_to_date(df: DataFrame) -> DataFrame:
def df_quarter_to_date(df: pd.DataFrame) -> pd.DataFrame:
"""Yields the Quarter-to-Date (QTD) DataFrame"""
now = Timestamp.now()
for m in [1, 4, 7, 10]:
@@ -33,7 +32,7 @@ def df_quarter_to_date(df: DataFrame) -> DataFrame:
return df[df.index >= now.strftime("%Y-%m-01")]
def df_year_to_date(df: DataFrame) -> DataFrame:
def df_year_to_date(df: pd.DataFrame) -> pd.DataFrame:
"""Yields the Year-to-Date (YTD) DataFrame"""
in_ytd = df.index >= Timestamp.now().strftime("%Y-01-01")
if any(in_ytd): return df[in_ytd]
@@ -75,7 +74,7 @@ def get_time(exchange: str = "NYSE", full:bool = True, to_string:bool = False) -
return s if to_string else print(s)
def total_time(df: DataFrame, tf: str = "years") -> float:
def total_time(df: pd.DataFrame, tf: str = "years") -> float:
"""Calculates the total time of a DataFrame. Difference of the Last and
First index. Options: 'months', 'weeks', 'days', 'hours', 'minutes'
and 'seconds'. Default: 'years'.
@@ -96,7 +95,7 @@ def total_time(df: DataFrame, tf: str = "years") -> float:
return TimeFrame["years"]
def to_utc(df: DataFrame) -> DataFrame:
def to_utc(df: pd.DataFrame) -> pd.DataFrame:
"""Either localizes the DataFrame Index to UTC or it applies tz_convert to
set the Index to UTC.
"""
@@ -108,6 +107,16 @@ def to_utc(df: DataFrame) -> DataFrame:
return df
def unix_convert(ts: Union[int, pd.Series]) -> Union[datetime, str]:
"""
Converts timestamps from polygon to readable datetime strings.
:param ts: The timestamp(s). An integer posix timestamp or a pd.Series of timestamps
:return: The converted datetime string
"""
return pd.to_datetime(ts, unit="ms")
# Aliases
mtd = df_month_to_date
qtd = df_quarter_to_date
pandas_ta/utils/data/polygon_api.py
+162 -172
View File
@@ -1,12 +1,8 @@
from pandas import DataFrame
from pandas_ta import Imports, RATE, version
import datetime
import polygon
import pandas as pd
from typing import Union
import logging
LOGGER = logging.getLogger(__name__)
from pandas import DataFrame
from pandas_ta import Imports, pd, RATE, version
from pandas_ta.utils import unix_convert
def polygon_api(ticker: str, **kwargs):
@@ -53,7 +49,6 @@ def polygon_api(ticker: str, **kwargs):
* ``verbose`` - Prints Company Information "info" and a Chart History
header to the screen. Default: False
"""
LOGGER.info(f"[!] kwargs: {kwargs}")
verbose = kwargs.pop("verbose", False)
kind = kwargs.pop("kind", "nothing").lower()
api_key = kwargs.pop("api_key", None)
@@ -68,7 +63,7 @@ def polygon_api(ticker: str, **kwargs):
if ticker is not None and isinstance(ticker, str):
ticker = ticker.upper()
else:
raise ValueError("Ticker symbol name must be a valid name string. Eg: \'AMD\'")
raise ValueError(f"Ticker symbol name must be a valid name string. Eg: \'AMD\'")
start_date = kwargs.pop("start_date", (datetime.date.today() - datetime.timedelta(days=525)))
end_date = kwargs.pop("end_date", datetime.date.today())
@@ -76,198 +71,193 @@ def polygon_api(ticker: str, **kwargs):
multiplier = kwargs.pop("multiplier", 1)
timespan = kwargs.pop("timespan", "day")
LOGGER.info(f"start date: {start_date} || end date: {end_date} || limit: {limit} || "
f"multiplier: {multiplier} || timespan: {timespan}")
with polygon.StocksClient(api_key) as polygon_client:
resp = polygon_client.get_aggregate_bars(ticker, start_date, end_date, limit=limit,
multiplier=multiplier, timespan=timespan)
df = DataFrame()
if "results" in resp.keys():
df = pd.DataFrame.from_dict(resp["results"])
df = df.set_index(pd.DatetimeIndex(unix_convert(df["t"])))
df.index.name = "DateTime"
# reorder then rename
df = df[["o", "h", "l", "c", "v", "vw", "n"]]
_columns = {
"o": "Open", "h": "High", "l": "Low", "c": "Close",
"v": "Volume", "vw": "VWAP", "n": "Trades"
}
df.rename(columns=_columns, errors="ignore", inplace=True)
if df.empty:
print(f"[X] Could not find: {ticker} with 'get_aggregate_bars()'.")
if not Imports["polygon"]:
print(f"[X] Please install yfinance to use this method. (pip install yfinance)")
return
df.name = ticker
if Imports["polygon"] and ticker is not None:
import polygon as polyapi
# ADDITIONAL DATA FLOW
ref_client, stock_client = polygon.ReferenceClient(api_key), polygon.StocksClient(api_key)
with polyapi.StocksClient(api_key) as polygon_client:
resp = polygon_client.get_aggregate_bars(ticker, start_date, end_date, limit=limit,
multiplier=multiplier, timespan=timespan)
div = "=" * 53 # Max div width is 80
# ALL THE INFORMATION
if kind in ["all", "info"] or verbose:
print("\n==== Company Information " + div)
details = ref_client.get_ticker_details(ticker)
details_vx = ref_client.get_ticker_details_vx(ticker)["results"]
df = DataFrame()
if "results" in resp.keys():
df = pd.DataFrame.from_dict(resp["results"])
df = df.set_index(pd.DatetimeIndex(unix_convert(df["t"])))
df.index.name = "DateTime"
# reorder then rename
df = df[["o", "h", "l", "c", "v", "vw", "n"]]
_columns = {
"o": "Open", "h": "High", "l": "Low", "c": "Close",
"v": "Volume", "vw": "VWAP", "n": "Trades"
}
df.rename(columns=_columns, errors="ignore", inplace=True)
has_name = "name" in details_vx and len(details_vx['name'])
has_ticker = "ticker" in details_vx and len(details_vx['ticker'])
if not has_ticker: details_vx['ticker'] = ticker
if has_name and has_ticker:
print(f"{details_vx['name']} [{details_vx['ticker']}]")
else:
print(f"{details_vx['ticker']}")
if df.empty:
print(f"[X] Could not find: {ticker} with 'get_aggregate_bars()'.")
return
df.name = ticker
# TODO: polygon returns hell lotta data for market info across a few endpoints. I don't know which ones to
# include here lol. I wrote the ones i felt were important. Feel free to suggest more.
# Yeah. It needs some additional modifications since details and details_vx are
# not equal and sparse depending on asset of ticker
# ADDITIONAL DATA FLOW
ref_client, stock_client = polyapi.ReferenceClient(api_key), polyapi.StocksClient(api_key)
# Common Information
# print(f"{details['hq_address']}. {details['hq_country']}\nPhone: {details_vx['phone_number']}\n"
# f"Website: {details['url']} || Employees: {details['employees']}\nSector: {details['sector']} || "
# f"Industry: {details['industry']}\n\n==== Market Information {div}\n"
# f"Market: {details_vx['market'].upper()} || locale: {details_vx['locale'].upper()} || "
# f"Exchange: {details['exchange']} || Symbol: {details['symbol']}\nMarket Shares: "
# f"{details_vx['market_cap']} || Outstanding Shares: {details_vx['outstanding_shares']}\n")
has_hq_address = "hq_address" in details and len(details['hq_address'])
has_vx_address = "address" in details_vx and len(details_vx["address"])
if has_hq_address:
print(f"{details['hq_address']}\n{details['hq_state']}, {details['hq_country']}")
elif has_vx_address:
has_vx_address1 = "address1" in details_vx['address'] and len(details_vx['address']['address1'])
has_vx_address2 = "address2" in details_vx['address'] and len(details_vx['address']['address2'])
if has_vx_address1 and has_vx_address2:
print(f"{details_vx['address']['address1']}\n{details_vx['address']['address2']}\n{details_vx['address']['city']}, {details_vx['address']['state']} {details_vx['address']['postal_code']}")
elif has_vx_address1:
print(f"{details_vx['address']['address1']}\n{details_vx['address']['city']}, {details_vx['address']['state']} {details_vx['address']['postal_code']}")
div = "=" * 53 # Max div width is 80
# ALL THE INFORMATION
if kind in ["all", "info"] or verbose:
print("\n==== Company Information " + div)
details = ref_client.get_ticker_details(ticker)
details_vx = ref_client.get_ticker_details_vx(ticker)["results"]
has_phone = "phone" in details and len(details['phone'])
has_vx_phone = "phone_number" in details_vx and len(details_vx['phone_number'])
if has_phone or has_vx_phone:
_phone = details_vx['phone_number'] or details['phone']
if len(_phone): print(f"Phone: {_phone}")
has_name = "name" in details_vx and len(details_vx['name'])
has_ticker = "ticker" in details_vx and len(details_vx['ticker'])
if not has_ticker: details_vx['ticker'] = ticker
if has_name and has_ticker:
print(f"{details_vx['name']} [{details_vx['ticker']}]")
else:
print(f"{details_vx['ticker']}")
# Market Information
has_market = "locale" in details_vx and len(details_vx["locale"])
has_exchange = "primary_exchange" in details_vx and len(details_vx["primary_exchange"])
print("\n==== Market Information " + div)
if has_market and has_exchange and has_ticker:
print(f"Market | Exchange | Symbol".ljust(39), f"{details_vx['locale'].upper()} | {details_vx['primary_exchange']} | {details_vx['ticker']}".rjust(40))
# TODO: polygon returns hell lotta data for market info across a few endpoints. I don't know which ones to
# include here lol. I wrote the ones i felt were important. Feel free to suggest more.
# Yeah. It needs some additional modifications since details and details_vx are
# not equal and sparse depending on asset of ticker
print()
if "market_cap" in details_vx:
print(f"Market Cap.".ljust(39), f"{details_vx['market_cap']:,} ({details_vx['market_cap']/1000000:,.2f} MM)".rjust(40))
if "outstanding_shares" in details_vx:
print(f"Shares Outstanding".ljust(39), f"{details_vx['outstanding_shares']:,}".rjust(40))
# Common Information
# print(f"{details['hq_address']}. {details['hq_country']}\nPhone: {details_vx['phone_number']}\n"
# f"Website: {details['url']} || Employees: {details['employees']}\nSector: {details['sector']} || "
# f"Industry: {details['industry']}\n\n==== Market Information {div}\n"
# f"Market: {details_vx['market'].upper()} || locale: {details_vx['locale'].upper()} || "
# f"Exchange: {details['exchange']} || Symbol: {details['symbol']}\nMarket Shares: "
# f"{details_vx['market_cap']} || Outstanding Shares: {details_vx['outstanding_shares']}\n")
has_hq_address = "hq_address" in details and len(details['hq_address'])
has_vx_address = "address" in details_vx and len(details_vx["address"])
if has_hq_address:
print(f"{details['hq_address']}\n{details['hq_state']}, {details['hq_country']}")
elif has_vx_address:
has_vx_address1 = "address1" in details_vx['address'] and len(details_vx['address']['address1'])
has_vx_address2 = "address2" in details_vx['address'] and len(details_vx['address']['address2'])
if has_vx_address1 and has_vx_address2:
print(f"{details_vx['address']['address1']}\n{details_vx['address']['address2']}\n{details_vx['address']['city']}, {details_vx['address']['state']} {details_vx['address']['postal_code']}")
elif has_vx_address1:
print(f"{details_vx['address']['address1']}\n{details_vx['address']['city']}, {details_vx['address']['state']} {details_vx['address']['postal_code']}")
# Price Info
snap_res = stock_client.get_snapshot(ticker)
print(f"\n==== Price Information ==={div}")
try:
snap = snap_res["ticker"]
has_phone = "phone" in details and len(details['phone'])
has_vx_phone = "phone_number" in details_vx and len(details_vx['phone_number'])
if has_phone or has_vx_phone:
_phone = details_vx['phone_number'] or details['phone']
if len(_phone): print(f"Phone: {_phone}")
# TODO: Convert to DF and print similar to YF
print(f"\nCurrent Price: {snap['lastTrade']['p']} || Today\'s Change: ${snap_res['todaysChange']} - "
f"{snap_res['todaysChangePerc']}%\nBid: {snap['lastQuote']['p']} x {snap['lastQuote']['s']} || Ask: "
f"{snap['lastQuote']['P']} x {snap['lastQuote']['S']} || Spread: "
f"{round(snap['lastQuote']['P'] - snap['lastQuote']['p'], 4)}\nOpen: {snap['day']['o']} || High: "
f"{snap['day']['h']} || Low: {snap['day']['l']} || Close: {snap['day']['c']} || Volume: "
f"{snap['day']['v']} || VWA: {snap['day']['vw']}")
except KeyError:
print(f"* Snapshot not found for {ticker}. Can not print price information.\n"
f"* Note: Snapshot data is cleared at 12am EST and gets populated as data is\n"
f" received from the exchanges. This can happen as early as 4am EST.\n"
f'* Requires a "Stocks Starter" subscription')
# Market Information
has_market = "locale" in details_vx and len(details_vx["locale"])
has_exchange = "primary_exchange" in details_vx and len(details_vx["primary_exchange"])
print("\n==== Market Information " + div)
if has_market and has_exchange and has_ticker:
print(f"Market | Exchange | Symbol".ljust(39), f"{details_vx['locale'].upper()} | {details_vx['primary_exchange']} | {details_vx['ticker']}".rjust(40))
# Splits and Dividends
# divs, splits = ref_client.get_stock_dividends(ticker), ref_client.get_stock_splits(ticker)
# # TODO: spits and dividends endpoints from polygon return a huge list. not sure if that entire list is useful
# print(f"\nNumber of dividends: {divs['count']} || Number of splits: {splits['count']}\n")
print()
if "market_cap" in details_vx:
print(f"Market Cap.".ljust(39), f"{details_vx['market_cap']:,} ({details_vx['market_cap']/1000000:,.2f} MM)".rjust(40))
if "outstanding_shares" in details_vx:
print(f"Shares Outstanding".ljust(39), f"{details_vx['outstanding_shares']:,}".rjust(40))
# TODO: financials endpoint on polygon returns a huge response. I doubt if that's useful to be displayed.
# Price Info
snap_res = stock_client.get_snapshot(ticker)
print(f"\n==== Price Information ==={div}")
try:
snap = snap_res["ticker"]
# Option Chains
if kind in ["option_chains", "oc"]:
_contract_type = kwargs.pop("contract_type", "all").lower()
contract_type = None if _contract_type == "all" else _contract_type
contract_limit = kwargs.pop("contract_limit", 10)
# TODO: Convert to DF and print similar to YF
print(f"\nCurrent Price: {snap['lastTrade']['p']} || Today\'s Change: ${snap_res['todaysChange']} - "
f"{snap_res['todaysChangePerc']}%\nBid: {snap['lastQuote']['p']} x {snap['lastQuote']['s']} || Ask: "
f"{snap['lastQuote']['P']} x {snap['lastQuote']['S']} || Spread: "
f"{round(snap['lastQuote']['P'] - snap['lastQuote']['p'], 4)}\nOpen: {snap['day']['o']} || High: "
f"{snap['day']['h']} || Low: {snap['day']['l']} || Close: {snap['day']['c']} || Volume: "
f"{snap['day']['v']} || VWA: {snap['day']['vw']}")
except KeyError:
print(f"* Snapshot not found for {ticker}. Can not print price information.\n"
f"* Note: Snapshot data is cleared at 12am EST and gets populated as data is\n"
f" received from the exchanges. This can happen as early as 4am EST.\n"
f'* Requires a "Stocks Starter" subscription')
call_chain = put_chain = None
if contract_type is None:
call_chain = ref_client.get_option_contracts(
ticker, limit=contract_limit,
contract_type="call"
)
# Splits and Dividends
# divs, splits = ref_client.get_stock_dividends(ticker), ref_client.get_stock_splits(ticker)
# # TODO: spits and dividends endpoints from polygon return a huge list. not sure if that entire list is useful
# print(f"\nNumber of dividends: {divs['count']} || Number of splits: {splits['count']}\n")
put_chain = ref_client.get_option_contracts(
ticker, limit=contract_limit,
contract_type="put"
)
else:
if contract_type == "call":
# TODO: financials endpoint on polygon returns a huge response. I doubt if that's useful to be displayed.
# Option Chains
if kind in ["option_chains", "oc"]:
_contract_type = kwargs.pop("contract_type", "all").lower()
contract_type = None if _contract_type == "all" else _contract_type
contract_limit = kwargs.pop("contract_limit", 10)
call_chain = put_chain = None
if contract_type is None:
call_chain = ref_client.get_option_contracts(
ticker, limit=contract_limit,
contract_type="call"
)
if contract_type == "put":
put_chain = ref_client.get_option_contracts(
ticker, limit=contract_limit,
contract_type="put"
)
else:
if contract_type == "call":
call_chain = ref_client.get_option_contracts(
ticker, limit=contract_limit,
contract_type="call"
)
if contract_type == "put":
put_chain = ref_client.get_option_contracts(
ticker, limit=contract_limit,
contract_type="put"
)
if call_chain is not None or put_chain is not None:
print(f"\n==== Option Chains {div}")
if call_chain is not None or put_chain is not None:
print(f"\n==== Option Chains {div}")
def _cleandf(chain: dict):
exp_dates = [x["expiration_date"] for x in chain]
df = DataFrame().from_records(chain)
df = df[["ticker", "strike_price", "expiration_date", "exercise_style"]]
df.columns = ["Contract", "Strike", "Exp. Date", "Style"]
df.set_index("Exp. Date", inplace=True)
return exp_dates, df
def _cleandf(chain: dict):
exp_dates = [x["expiration_date"] for x in chain]
df = DataFrame().from_records(chain)
df = df[["ticker", "strike_price", "expiration_date", "exercise_style"]]
df.columns = ["Contract", "Strike", "Exp. Date", "Style"]
df.set_index("Exp. Date", inplace=True)
return exp_dates, df
if call_chain is not None and len(call_chain["results"]):
exp_dates, calldf = _cleandf(call_chain["results"])
if contract_type == "call":
print(f"\n{ticker} Calls for {exp_dates[0]}\n{calldf}")
if call_chain is not None and len(call_chain["results"]):
exp_dates, calldf = _cleandf(call_chain["results"])
if contract_type == "call":
print(f"\n{ticker} Calls for {exp_dates[0]}\n{calldf}")
if put_chain is not None and len(put_chain["results"]):
exp_dates, putdf = _cleandf(put_chain["results"])
if contract_type == "put":
print(f"\n{ticker} Puts for {exp_dates[0]}\n{putdf}")
if put_chain is not None and len(put_chain["results"]):
exp_dates, putdf = _cleandf(put_chain["results"])
if contract_type == "put":
print(f"\n{ticker} Puts for {exp_dates[0]}\n{putdf}")
if contract_type is None:
alldf = pd.merge(calldf.reset_index(), putdf.reset_index(), on="Strike")
alldf.rename(
columns={"Contract_x": "Calls", "Contract_y": "Puts", "Exp. Date_x": "Exp. Date"},
inplace=True
)
alldf.set_index("Exp. Date", inplace=True)
alldf = alldf[["Calls", "Strike", "Puts"]]
print(f"\n{ticker} Calls & Puts for {exp_dates[0]}\n{alldf}")
else:
print(f"\nNo option chains data found for {ticker}.")
if contract_type is None:
alldf = pd.merge(calldf.reset_index(), putdf.reset_index(), on="Strike")
alldf.rename(
columns={"Contract_x": "Calls", "Contract_y": "Puts", "Exp. Date_x": "Exp. Date"},
inplace=True
)
alldf.set_index("Exp. Date", inplace=True)
alldf = alldf[["Calls", "Strike", "Puts"]]
print(f"\n{ticker} Calls & Puts for {exp_dates[0]}\n{alldf}")
else:
print(f"\nNo option chains data found for {ticker}.")
if verbose:
_chart_history = \
f"\n==== Chart History " + div + \
f"\n[*] Pandas TA v{version} & polygon API" + \
f"\n[+] Downloading {ticker} [{start_date} : {end_date}] from Polygon (www.polygon.io/)\n{'='*80}\n"
print(_chart_history)
if verbose:
_chart_history = \
f"\n==== Chart History " + div + \
f"\n[*] Pandas TA v{version} & polygon API" + \
f"\n[+] Downloading {ticker} [{start_date} : {end_date}] from Polygon (www.polygon.io/)\n{'='*80}\n"
print(_chart_history)
if show is not None and isinstance(show, int) and show > 0:
print(f"\n{df.name}\n{df.tail(show)}\n")
if show is not None and isinstance(show, int) and show > 0:
print(f"\n{df.name}\n{df.tail(show)}\n")
return df
def unix_convert(ts: Union[int, pd.Series]) -> Union[datetime.datetime, str]:
"""
Converts timestamps from polygon to readable datetime strings.
:param ts: The timestamp(s). An integer posix timestamp or a pd.Series of timestamps
:return: The converted datetime string
"""
return pd.to_datetime(ts, unit="ms")
return df
else:
return DataFrame()
pandas_ta/utils/data/processes.py
+30 -52
View File
@@ -2,30 +2,8 @@
# -*- coding: utf-8 -*-
import datetime as dt
from random import choice as rChoice
from numpy import absolute as npAbsolute
from numpy import any as npAny
from numpy import array as npArray
from numpy import concatenate as npConcat
from numpy import cumsum as npCumsum
from numpy import flip as npFlip
from numpy import mean as npMean
from numpy import max as npMax
from numpy import min as npMin
from numpy import ndarray as npNdArray
from numpy import std as npStd
from numpy import where as npWhere
from numpy import zeros as npZeros
from numpy.random import normal as npNormal
from numpy.random import randint as npRandInt
from numpy.random.mtrand import randint as npRandInt
from numpy.random import choice as npChoice
from pandas import DataFrame, date_range
from pandas_ta import Imports, RATE
from pandas_ta import Imports, RATE, np
class sample(object):
@@ -154,7 +132,7 @@ class sample(object):
_generate() method to build a sample realization with the given
arguments.
"""
_random_symbol = ''.join([rChoice("ABCDEFGHIJKLMNOPQRSTUVWXYZ") for _ in range(npRandInt(3, 6))])
_random_symbol = ''.join([rChoice("ABCDEFGHIJKLMNOPQRSTUVWXYZ") for _ in range(np.random.randint(3, 6))])
self._name = str(name) if name is not None and isinstance(name, str) else _random_symbol
self._process = str(process).lower() if process is not None and isinstance(process, str) and process in self._processes else None
self._noise = str(noise).lower() if noise is not None and isinstance(noise, str) and noise in self._noises else None
@@ -185,15 +163,15 @@ class sample(object):
self._verbose = verbose if verbose is not None and isinstance(verbose, bool) else False
if self._process == "rand":
self._process = npChoice(self._processes[:-2])
self._process = np.random.choice(self._processes[:-2])
if self._noise == "rand":
self._noise = npChoice(self._noises[:-1])
self._noise = np.random.choice(self._noises[:-1])
self._generate() # Run it
def _bernoulli_mask(self, array, percent:float = None, p:float = None):
def _bernoulli_mask(self, array: np.ndarray, percent:float = None, p:float = None):
"""Bernoulli Mask - Positive or Negative"""
if array.size > 0:
percent = float(percent) if percent is not None and isinstance(percent, float) else self.noise_percent
@@ -204,7 +182,7 @@ class sample(object):
def _bernoulli_process(self):
"""Bernoulli Process"""
return npRandInt(2, size=self.length)
return np.random.randint(2, size=self.length)
def _generate(self):
@@ -241,20 +219,20 @@ class sample(object):
_npns = f"{self.name} | {self.process} {self.noise+' ' if self.noise is not None else ''}{self.np.size}"
_s0n = f"s0: {round(self.np[0], self._precision)}, sN: {round(self.np[-1], self._precision)}"
_msmm = f"mu: {round(npMean(self.np), self._precision)}, sigma: {round(npStd(self.np), self._precision)}"
_msmm = f"mu: {round(np.mean(self.np), self._precision)}, sigma: {round(np.std(self.np), self._precision)}"
self._dfname = f"{_npns} | {_s0n} | {_msmm}"
if self._verbose: print(self._dfname)
def nonnegative(self, array: npNdArray = None):
def nonnegative(self, array: np.ndarray = None):
"""Vertical Translation the 'array' where the resultant 'array' has
non-negative values."""
if isinstance(array, npNdArray):
if isinstance(array, np.ndarray):
return self._nonnegative(array)
return array
def _nonnegative(self, array):
def _nonnegative(self, array: np.ndarray):
"""Translates the array up by the minimum of the 'array' if any values
are negative."""
if array.size > 0 and any(array < 0):
@@ -263,25 +241,25 @@ class sample(object):
return array
def _normal_mask(self, array):
def _normal_mask(self, array: np.ndarray):
"""A method to add some additional randomness to the realized
process. Applies a mask based on the Normal Distribution and the 'array's
mean and standard deviation."""
if array.size > 0:
norm = npNormal(npMean(array), npStd(array), size=self.length)
norm = np.random.normal(np.mean(array), np.std(array), size=self.length)
return array * self.noise_percent * norm
return array
def orientation(self, array, mode:str = None):
def orientation(self, array: np.ndarray, mode: str = None):
"""Orients the 'array' either by Inversion, Reversal, or an
Inverted Reversal."""
if isinstance(array, npNdArray):
if isinstance(array, np.ndarray):
return self._orientation(array, mode=mode)
return array
def _orientation(self, array, mode:str = None):
def _orientation(self, array: np.ndarray, mode: str = None):
"""Orients the 'array' either by Inversion, Reversal, or an
Inverted Reversal."""
_modes = ["i", "r", "ir", "ri", None, "rand"]
@@ -289,16 +267,16 @@ class sample(object):
result = array
if mode is None: return result
if mode == "rand": mode = npChoice(_modes[3:])
if mode == "rand": mode = np.random.choice(_modes[3:])
if mode == "i":
mid = 0.5 * (npMin(array) + npMax(array))
mid = 0.5 * (np.min(array) + np.max(array))
inv = mid - array
diff = inv - inv[0]
result = array[0] + diff if array[0] > 0 else diff - array[0]
if mode == "r":
result = npFlip(array) - (array[-1] - array[0])
result = np.flip(array) - (array[-1] - array[0])
if mode in ["ir", "ri"]:
result = self._orientation(self._orientation(array, "i"), "r")
@@ -306,22 +284,22 @@ class sample(object):
return result
def scale(self, array, mode:str):
def scale(self, array: np.ndarray, mode: str):
"""Mean, Normal or Standard scaling of the 'array'."""
if isinstance(array, npNdArray):
if isinstance(array, np.ndarray):
return self._scaler(array, mode=mode)
return array
def _scaler(self, array, mode:str):
def _scaler(self, array: np.ndarray, mode: str):
"""Scaling: mean, normal, standard"""
result = array
if mode is None: return result
if mode == "rand": mode = npChoice(self._scales[3:])
if mode == "rand": mode = np.random.choice(self._scales[3:])
min_, max_ = npMin(array), npMax(array)
range_ = npAbsolute(max_ - min_)
mu_, std_ = npMean(array), npStd(array)
min_, max_ = np.min(array), np.max(array)
range_ = np.absolute(max_ - min_)
mu_, std_ = np.mean(array), np.std(array)
if mode == "m" and range_ > 0: # "mean"
result = ((array - mu_) / range_)
@@ -335,7 +313,7 @@ class sample(object):
return result
def _simple_random_walk(self, up:float = None, down:float = None) -> npArray:
def _simple_random_walk(self, up:float = None, down:float = None) -> np.array:
"""Simple Random Walk
Sources:
@@ -345,8 +323,8 @@ class sample(object):
down = float(down) if down is not None and isinstance(down, (int, float)) else -1.0
if up < down: down, up = up, down
x = npConcat(([0.0], npWhere(npRandInt(0, 2, size=self.length - 1) == 0, down, up)))
return npCumsum(x).astype(float)
x = np.concatenate(([0.0], np.where(np.random.randint(0, 2, size=self.length - 1) == 0, down, up)))
return np.cumsum(x).astype(float)
def _stoch_noise(self):
@@ -354,7 +332,7 @@ class sample(object):
Otherwise, it returns 0 noise.
"""
_desc = f"[+] "
result = npZeros(self.length, dtype=float)
result = np.zeros(self.length, dtype=float)
if self._noise is not None and Imports["stochastic"]:
from stochastic import random as st_random
@@ -403,7 +381,7 @@ class sample(object):
# Initial Value (s0) adjustment
result = result + result[0] if result[0] > self.s0 else result - result[0]
if result is not None and npAny(result) and self._verbose: print(_desc)
if result is not None and np.any(result) and self._verbose: print(_desc)
return result
pandas_ta/utils/data/yahoofinance.py
+2 -2
View File
@@ -180,11 +180,11 @@ def yf(ticker: str, **kwargs):
print(f"Insiders % | Institution %".ljust(39), f"{100 * ticker_info['heldPercentInsiders']:.4f}% | {100 * ticker_info['heldPercentInstitutions']:.4f}%".rjust(40))
print()
if "bookValue" in ticker_info and ticker_info['bookValue'] is not None or "priceToBook" in ticker_info and ticker_info['priceToBook'] is not None or "pegRatio" in ticker_info and ticker_info['pegRatio'] is not None:
if "bookValue" in ticker_info and ticker_info['bookValue'] is not None and "priceToBook" in ticker_info and ticker_info['priceToBook'] is not None and "pegRatio" in ticker_info and ticker_info['pegRatio'] is not None:
print(f"Book Value | Price to Book | Peg Ratio".ljust(39), f"{ticker_info['priceToBook']} | {ticker_info['priceToBook']} | {ticker_info['pegRatio']}".rjust(40))
if "forwardPE" in ticker_info and ticker_info['forwardPE'] is not None:
print(f"Forward PE".ljust(39), f"{ticker_info['forwardPE']}".rjust(40))
if "forwardEps" in ticker_info and ticker_info['forwardEps'] is not None or "trailingEps" in ticker_info and ticker_info['trailingEps'] is not None:
if "forwardEps" in ticker_info and ticker_info['forwardEps'] is not None and "trailingEps" in ticker_info and ticker_info['trailingEps'] is not None:
print(f"Forward EPS | Trailing EPS".ljust(39), f"{ticker_info['forwardEps']} | {ticker_info['trailingEps']}".rjust(40))
if "enterpriseValue" in ticker_info and ticker_info['enterpriseValue'] is not None:
print(f"Enterprise Value".ljust(39), f"{ticker_info['enterpriseValue']:,}".rjust(40))
pandas_ta/volume/kvo.py
+2 -2
View File
@@ -20,8 +20,8 @@ def kvo(high, low, close, volume, fast=None, slow=None, signal=None, mamode=None
close (pd.Series): Series of 'close's
volume (pd.Series): Series of 'volume's
fast (int): The fast period. Default: 34
long (int): The long period. Default: 55
length_sig (int): The signal period. Default: 13
slow (int): The slow period. Default: 55
signal (int): The signal period. Default: 13
mamode (str): See ```help(ta.ma)```. Default: 'ema'
offset (int): How many periods to offset the result. Default: 0
setup.py
+1 -1
View File
@@ -19,7 +19,7 @@ setup(
"pandas_ta.volatility",
"pandas_ta.volume"
],
version=".".join(("0", "3", "41b")),
version=".".join(("0", "3", "42b")),
description=long_description,
long_description=long_description,
author="Kevin Johnson",
tests/config.py
+4 -3
View File
@@ -1,5 +1,5 @@
import os
from pandas import DatetimeIndex, read_csv
from pandas import DataFrame, DatetimeIndex, read_csv
VERBOSE = True
@@ -18,9 +18,10 @@ sample_data = read_csv(
)
sample_data.set_index(DatetimeIndex(sample_data["date"]), inplace=True, drop=True)
sample_data.drop("date", axis=1, inplace=True)
sample_data = sample_data[:200] # First 200
# sample_data = sample_data[:200] # First 200
# sample_data = sample_data[100:300] # Decreasing Segment
# sample_data = sample_data[-200:] # Last 200
# sample_data = sample_data[:80]
def error_analysis(df, kind, msg, icon=INFO, newline=True):
if VERBOSE:
tests/context.py
+1
View File
@@ -1,3 +1,4 @@
# -*- coding: utf-8 -*-
import os
import sys
tests/test_ext_indicator_candle.py
+1 -1
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
from .config import sample_data
from .context import pandas_ta
from unittest import TestCase, skip
from pandas import DataFrame
tests/test_ext_indicator_cycles.py
+1 -2
View File
@@ -1,6 +1,5 @@
from pandas.core.series import Series
# -*- coding: utf-8 -*-
from .config import sample_data
from .context import pandas_ta
from unittest import TestCase
from pandas import DataFrame
tests/test_ext_indicator_momentum.py
+1 -1
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
from .config import sample_data
from .context import pandas_ta
from unittest import skip, TestCase
from pandas import DataFrame
tests/test_ext_indicator_overlap_ext.py
+6 -5
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
from .config import sample_data
from .context import pandas_ta
from unittest import skip, TestCase
from pandas import DataFrame
@@ -134,13 +134,14 @@ class TestOverlapExtension(TestCase):
self.assertEqual(self.data.columns[-1], "SMMA_7")
def test_ssf_ext(self):
self.data.ta.ssf(append=True, poles=2)
self.data.ta.ssf(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "SSF_10_2")
self.assertEqual(self.data.columns[-1], "SSF_20")
self.data.ta.ssf(append=True, poles=3)
def test_ssf3_ext(self):
self.data.ta.ssf3(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "SSF_10_3")
self.assertEqual(self.data.columns[-1], "SSF3_20")
def test_swma_ext(self):
self.data.ta.swma(append=True)
tests/test_ext_indicator_performance.py
+1
View File
@@ -1,3 +1,4 @@
# -*- coding: utf-8 -*-
from .config import sample_data
from .context import pandas_ta
tests/test_ext_indicator_statistics.py
+1 -1
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
from .config import sample_data
from .context import pandas_ta
from unittest import skip, TestCase
from pandas import DataFrame
tests/test_ext_indicator_trend.py
+2 -2
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
from .config import sample_data
from .context import pandas_ta
from unittest import skip, TestCase
from pandas import DataFrame
@@ -119,7 +119,7 @@ class TestTrendExtension(TestCase):
def test_trendflex_ext(self):
self.data.ta.trendflex(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(list(self.data.columns[-1:]), ["TRENDFLEX_20_20"])
self.assertEqual(list(self.data.columns[-1:]), ["TRENDFLEX_20_20_0.04"])
def test_ttm_trend_ext(self):
self.data.ta.ttm_trend(append=True)
tests/test_ext_indicator_volatility.py
+1 -1
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
from .config import sample_data
from .context import pandas_ta
from unittest import TestCase
from pandas import DataFrame
tests/test_ext_indicator_volume.py
+1 -1
View File
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
from unittest.case import skip
from .config import sample_data
from .context import pandas_ta
from unittest import TestCase
from pandas import DataFrame
tests/test_indicator_candle.py
+1
View File
@@ -1,3 +1,4 @@
# -*- coding: utf-8 -*-
from .config import error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD, VERBOSE
from .context import pandas_ta
tests/test_indicator_cycles.py
+5 -6
View File
@@ -1,11 +1,10 @@
from .config import error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD, VERBOSE
# -*- coding: utf-8 -*-
from .config import sample_data, CORRELATION, CORRELATION_THRESHOLD, VERBOSE
from .context import pandas_ta
from unittest import TestCase, skip
import pandas.testing as pdt
from pandas import DataFrame, Series
import talib as tal
from pandas import Series
class TestCycles(TestCase):
@@ -40,7 +39,7 @@ class TestCycles(TestCase):
self.assertEqual(result.name, "EBSW_40_10")
def test_reflext(self):
def test_reflex(self):
result = pandas_ta.reflex(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "REFLEX_20_20_0.04")
self.assertEqual(result.name, "REFLEX_20_20_0.04")
tests/test_indicator_momentum.py
+2 -1
View File
@@ -1,4 +1,5 @@
from .config import error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD, VERBOSE
# -*- coding: utf-8 -*-
from .config import error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD
from .context import pandas_ta
from unittest import TestCase, skip
tests/test_indicator_overlap.py
+22 -17
View File
@@ -1,4 +1,4 @@
from .config import CORRELATION, CORRELATION_THRESHOLD, error_analysis, sample_data, VERBOSE
from .config import CORRELATION, CORRELATION_THRESHOLD, error_analysis, sample_data
from .context import pandas_ta
from unittest import TestCase, skip
@@ -65,7 +65,8 @@ class TestOverlap(TestCase):
self.assertEqual(result.name, "DEMA_10")
def test_ema(self):
result = pandas_ta.ema(self.close, presma=False)
# For TA Lib comparison
result = pandas_ta.ema(self.close, talib=False, presma=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "EMA_10")
@@ -79,20 +80,19 @@ class TestOverlap(TestCase):
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.ema(self.close, talib=False)
result = pandas_ta.ema(self.close, talib=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "EMA_10")
try:
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError:
try:
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.ema(self.close, talib=False, presma=False, adjust=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "EMA_10")
result = pandas_ta.ema(self.close)
result = pandas_ta.ema(self.close, talib=False, presma=False, adjust=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "EMA_10")
result = pandas_ta.ema(self.close, talib=False, presma=True, adjust=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "EMA_10")
@@ -329,7 +329,7 @@ class TestOverlap(TestCase):
except Exception as ex:
error_analysis(result, CORRELATION, ex)
result = pandas_ta.sma(self.close)
result = pandas_ta.sma(self.close, talib=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "SMA_10")
@@ -339,13 +339,18 @@ class TestOverlap(TestCase):
self.assertEqual(result.name, "SMMA_7")
def test_ssf(self):
result = pandas_ta.ssf(self.close, poles=2)
result = pandas_ta.ssf(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "SSF_10_2")
self.assertEqual(result.name, "SSF_20")
result = pandas_ta.ssf(self.close, poles=3)
result = pandas_ta.ssf(self.close, pi=pandas_ta.np.pi, sqrt2=pandas_ta.np.sqrt(2), everget=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "SSF_10_3")
self.assertEqual(result.name, "SSFe_20")
def test_ssf3(self):
result = pandas_ta.ssf3(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "SSF3_20")
def test_swma(self):
result = pandas_ta.swma(self.close)
tests/test_indicator_performance.py
+1
View File
@@ -1,3 +1,4 @@
# -*- coding: utf-8 -*-
from .config import sample_data
from .context import pandas_ta
tests/test_indicator_statistics.py
+2 -1
View File
@@ -1,4 +1,5 @@
from .config import error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD, VERBOSE
# -*- coding: utf-8 -*-
from .config import error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD
from .context import pandas_ta
from unittest import skip, TestCase
tests/test_indicator_trend.py
+4 -4
View File
@@ -1,9 +1,9 @@
from .config import error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD, VERBOSE
from .config import error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD
from .context import pandas_ta
from unittest import TestCase, skip
from numpy import NaN as npNaN
import numpy as np
import pandas.testing as pdt
from pandas import DataFrame, Series
@@ -165,7 +165,7 @@ class TestTrend(TestCase):
# 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.iloc[0] = npNaN
psar.iloc[0] = np.nan
psar.name = result.name
try:
@@ -192,7 +192,7 @@ class TestTrend(TestCase):
def test_trendflex(self):
result = pandas_ta.trendflex(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "TRENDFLEX_20_20")
self.assertEqual(result.name, "TRENDFLEX_20_20_0.04")
def test_ttm_trend(self):
result = pandas_ta.ttm_trend(self.high, self.low, self.close)
tests/test_indicator_volatility.py
+2 -1
View File
@@ -1,4 +1,5 @@
from .config import error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD, VERBOSE
# -*- coding: utf-8 -*-
from .config import error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD
from .context import pandas_ta
from unittest import TestCase, skip
tests/test_indicator_volume.py
+2
View File
@@ -1,3 +1,4 @@
# -*- coding: utf-8 -*-
from .config import error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD, VERBOSE
from .context import pandas_ta
@@ -97,6 +98,7 @@ class TestVolume(TestCase):
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "EOM_14_100000000")
# @skip
def test_kvo(self):
result = pandas_ta.kvo(self.high, self.low, self.close, self.volume_)
self.assertIsInstance(result, DataFrame)
tests/test_strategy.py
+2 -1
View File
@@ -1,3 +1,4 @@
# -*- coding: utf-8 -*-
# Must run seperately from the rest of the tests
# in order to successfully run
from multiprocessing import cpu_count
@@ -11,7 +12,7 @@ from pandas import DataFrame
# Strategy Testing Parameters
cores = cpu_count()
cores = cpu_count() - 1
cumulative = False
speed_table = False
strategy_timed = False
tests/test_utils.py
+8 -2
View File
@@ -1,3 +1,4 @@
# -*- coding: utf-8 -*-
from .config import sample_data
from .context import pandas_ta
@@ -151,8 +152,8 @@ class TestUtilities(TestCase):
# result = self.utils.df_dates(self.data, ["1999-11-01", "2020-08-15", "2020-08-24", "2020-08-25", "2020-08-26", "2020-08-27"])
# self.assertEqual(5, result.shape[0])
result = self.utils.df_dates(self.data, ["1999-11-01", "2000-03-15"])
self.assertEqual(2, result.shape[0])
# result = self.utils.df_dates(self.data, ["1999-11-01", "2000-03-15"])
# self.assertEqual(2, result.shape[0])
@skip
def test_df_month_to_date(self):
@@ -286,6 +287,11 @@ class TestUtilities(TestCase):
npt.assert_array_equal(self.utils.pascals_triangle(n=5, weighted=True), array_5w)
npt.assert_array_equal(self.utils.pascals_triangle(n=5, weighted=True, inverse=True), array_5iw)
def test__performance(self):
_excluded = ["above", "above_value", "below", "below_value", "cross", "cross_value", "ichimoku"]
result = self.utils.performance(self.data, _excluded, top=10, ascending=False, places=4)
self.assertIsInstance(result, DataFrame)
def test_symmetric_triangle(self):
npt.assert_array_equal(self.utils.symmetric_triangle(), np.array([1,1]))
npt.assert_array_equal(self.utils.symmetric_triangle(weighted=True), np.array([0.5, 0.5]))
tests/test_utils_metrics.py
+4 -3
View File
@@ -1,10 +1,11 @@
# -*- coding: utf-8 -*-
from .config import sample_data
from .context import pandas_ta
from unittest import skip, TestCase
from pandas import DataFrame
from .config import sample_data
from .context import pandas_ta
class TestUtilityMetrics(TestCase):
@classmethod