Merge branch 'development'

This commit is contained in:
Kevin Johnson
2020-09-28 11:06:30 -07:00
55 changed files with 8250 additions and 3668 deletions
+3 -1
View File
@@ -116,7 +116,8 @@ env/**
pandas_ta/_wrapper.py
# twopirllc stuff
AlphaVantageAPI
AlphaVantageAPI/
data/datas.csv
data/GLD_D_tv.csv
data/SPY_5min.csv
@@ -129,6 +130,7 @@ data/tulip.csv
examples/cache.sqlite
examples/taplot.py
examples/alpaca_trader.py
examples/ChartTA.ipynb
examples/charting.ipynb
examples/ib_trader.ipynb
+19 -6
View File
@@ -1,14 +1,27 @@
clean:
find . -name '*.pyc' -exec rm -f {} +
.PHONY: all
all:
make test_utils
make test_ta
make test_ext
make test_strats
caches:
find ./pandas_ta | grep -E "(__pycache__|\.pyc|\.pyo$\)"
clean:
find . -name '*.pyc' -exec rm -f {} +
init:
pip install -r requirements.txt
ti:
python -m unittest -v tests/test_indicator*.py
test_ext:
python -m unittest -v tests/test_ext_indicator_*.py
ts:
python -m unittest -v tests/test_strategy.py
test_strats:
python -m unittest -v tests/test_strategy.py
test_ta:
python -m unittest -v tests/test_indicator_*.py
test_utils:
python -m unittest -v tests/test_utils.py
+319 -205
View File
@@ -1,107 +1,88 @@
Pandas TA - A Technical Analysis Library in Python 3
=================
[![Python Version](https://img.shields.io/pypi/pyversions/pandas_ta.svg)](https://pypi.org/project/pandas_ta/)
[![PyPi Version](https://img.shields.io/pypi/v/pandas_ta.svg)](https://pypi.org/project/pandas_ta/)
[![Package Status](https://img.shields.io/pypi/status/pandas_ta.svg)](https://pypi.org/project/pandas_ta/)
[![Downloads](https://img.shields.io/pypi/dm/pandas_ta.svg?style=flat)](https://pypistats.org/packages/pandas_ta)
# **Pandas TA**
# Pandas Technical Analysis Library in _Python 3_
![Example Chart](/images/TA_Chart.png)
_Pandas Technical Analysis_ (**Pandas TA**) is an easy to use library that is built upon Python's Pandas library with more than 120 Indicators and Utility functions. These indicators are commonly used for financial time series datasets with columns or labels: datetime, _open_, _high_, _low_, _close_, _volume_, et al. Many commonly used indicators are included, such as: _Simple Moving Average_ (**sma**) _Moving Average Convergence Divergence_ (**macd**), _Hull Exponential Moving Average_ (**hma**), _Bollinger Bands_ (**bbands**), _On-Balance Volume_ (**obv**), _Aroon & Aroon Oscillator_ (**aroon**), _Squeeze_ (**squeeze**) and **many more**.
**Pandas TA** has three different ways of processing Technical Indicators as described below. The **primary** requirement to run indicators in [Pandas DataFrame Extension](https://pandas.pydata.org/pandas-docs/stable/extending.html) mode, is that _open, high, low, close, volume_ are **lowercase**. Depending on the indicator, they either return a named Series or a DataFrame in uppercase underscore parameter format. For example, MACD(fast=12, slow=26, signal=9) will return a DataFrame with columns: ['MACD_12_26_9', 'MACDh_12_26_9', 'MACDs_12_26_9'].
_Pandas Technical Analysis_ (**Pandas TA**) is an easy to use library that leverages the Pandas library with more than 120 Indicators and Utility functions. Many commonly used indicators are included, such as: _Simple Moving Average_ (**sma**) _Moving Average Convergence Divergence_ (**macd**), _Hull Exponential Moving Average_ (**hma**), _Bollinger Bands_ (**bbands**), _On-Balance Volume_ (**obv**), _Aroon & Aroon Oscillator_ (**aroon**), _Squeeze_ (**squeeze**) and **_many more_**.
## Pandas TA Issues, Ideas and Contributions
<br/>
#### Thanks for trying **Pandas TA**!
# **Table of contents**
Please take a moment to read **this** and the rest of this **README** before posting any issue.
<!--ts-->
* [Features](#features)
* [Installation](#installation)
* [Stable](#stable)
* [Latest Version](#latest-version)
* [Quick Start](#quick-start)
* [Help](#help)
* [Issues and Contributions](#issues-and-contributions)
* [Programming Conventions](#programming-conventions)
* [Pandas TA Strategies](#pandas-ta-strategies)
* [Types of Strategies](#types-of-strategies)
* [DataFrame Properties](#dataframe-properties)
* [Changes](#changes)
* [Indicators by Category](#indicators-by-category)
* [Candles](#candles-3)
* [Momentum](#momentum-34)
* [Overlap](#overlap-27)
* [Performance](#performance-3)
* [Statistics](#statistics-9)
* [Trend](#trend-15)
* [Utility](#utility-5)
* [Volatility](#volatility-12)
* [Volume](#volume-13)
<!--te-->
* ### [Comments and Feedback](https://github.com/twopirllc/pandas-ta/issues)
* Have you read the rest of **this** document?
* Are you running the latest version?
* ```pip install -U git+https://github.com/twopirllc/pandas-ta```
* Have you tried the [Examples](https://github.com/twopirllc/pandas-ta/tree/master/examples/)?
* Did they help?
* What is missing?
* Could you help improve them?
* Did you know you can easily build _Custom Strategies_ with the **[Strategy](https://github.com/twopirllc/pandas-ta/blob/master/examples/PandasTA_Strategy_Examples.ipynb) Class**?
* Documentation could always use improvement. Can you contribute?
* ### [Indicator or Feature Requests & Contributions](https://github.com/twopirllc/pandas-ta/issues)
* Please be as detailed as possible. Links, screenshots, and sometimes data samples are welcome.
* You want a new indicator not currently listed.
* You want an alternate version of an existing indicator.
* The indicator does not match another website, library, broker platform, language, et al.
* Can you contribute?
<!-- * [Specifying Strategies in **Pandas TA**](#specifying-strategies-in-pandas-ta) -->
<!-- * [Multiprocessing](#multiprocessing) -->
## __Features__
<br/>
# **Features**
* Has 120+ indicators and utility functions.
* Easily add prefixes or suffixes or both to columns names. Useful for building Custom Strategies.
* __Extended Pandas DataFrame__ as 'ta'.
* Indicators are tightly correlated with the de facto [TA Lib](https://mrjbq7.github.io/ta-lib/) if they share common indicators.
* Have the need for speed? By using the _strategy_ method, you get **multiprocessing** for free!
* Easily add _prefixes_ or _suffixes_ or both to columns names. Useful for Custom Chained Strategies.
* Example Jupyter Notebooks under the [examples](https://github.com/twopirllc/pandas-ta/tree/master/examples) directory, including how to create Custom Strategies using the new [__Strategy__ Class](https://github.com/twopirllc/pandas-ta/tree/master/examples/PandaTA_Strategy_Examples.ipynb)
* A new 'ta' method called 'strategy'. By default, it runs __all__ the indicators or equivalent ta.AllStrategy.
<br/>
## __Recent Changes__
* A __Strategy__ Class to help name and group your favorite indicators.
* An experimental and independent __Watchlist__ Class located in the [Examples](https://github.com/twopirllc/pandas-ta/tree/master/examples/watchlist.py) Directory that can be used in conjunction with the new __Strategy__ Class.
and _Weighted Moving Average_.
* __Multiprocessing__ is automatically applied to df.ta.strategy() for __All__ indicators or a chosen __Category__ of indicators.
* Improved the calculation performance of indicators: _Exponential Moving Averagage_
* Updated *trend_return* utility to return a more pertinenet trade info for a _trend_. Example can be found in the [AI Example Notebook](https://github.com/twopirllc/pandas-ta/tree/master/examples/AIExample.ipynb). The notebook is still a work in progress and open to colloboration.
## __Breaking Indicators__
* _Stochastic Oscillator_ (**stoch**): Now in line with Trading View's calculation. See: ```help(ta.stoch)```
## __New Indicators__
* _Squeeze_ (**squeeze**). A Momentum indicator. Both John Carter's TTM **and** Lazybear's TradingView versions are implemented. The default is John Carter's, or ```lazybear=False```. Set ```lazybear=True``` to enable Lazybear's.
* _TTM Trend_ (**ttm_trend**). A trend indicator inspired from John Carter's book "Mastering the Trade".
* _SMI Ergodic_ (**smi**) Developed by William Blau, the SMI Ergodic Indicator is the same as the True Strength Index (TSI) except the SMI includes a signal line and oscillator.
* _Gann High-Low Activator_ (**hilo**) The Gann High Low Activator Indicator was created by Robert Krausz in a 1998
issue of Stocks & Commodities Magazine. It is a moving average based trend
indicator consisting of two different simple moving averages.
* _Stochastic RSI_ (**stochrsi**) "Stochastic RSI and Dynamic Momentum Index" was created by Tushar Chande and Stanley Kroll. In line with Trading View's calculation. See: ```help(ta.stochrsi)```
* _Inside Bar_ (**cdl_inside**) An Inside Bar is a bar contained within it's previous bar's high and low See: ```help(ta.cdl_inside)```
## __Updated Indicators__
* _Fisher Transform_ (**fisher**): Added Fisher's default **ema** signal line. To change the length of the signal line, use the argument: ```signal=5```. Default: 5
* _Fisher Transform_ (**fisher**) and _Kaufman's Adaptive Moving Average_ (**kama**): Fixed a bug where their columns were not added to final DataFrame when using the _strategy_ method.
* _Trend Return_ (**trend_return**): Returns a DataFrame now instead of Series.
* _Average True Range_ (**atr**): Added option to return **atr** as a percentage. See: ```help(ta.atr)```
## What is a Pandas DataFrame Extension?
A [Pandas DataFrame Extension](https://pandas.pydata.org/pandas-docs/stable/extending.html), extends a DataFrame allowing one to add more functionality and features to Pandas to suit your needs. As such, it is now easier to run Technical Analysis on existing Financial Time Series without leaving the current DataFrame. This extension by default returns the Indicator result or it can append the result to the existing DataFrame by including the parameter 'append=True' in the method call. Examples below.
# __Getting Started and Examples__
## __Installation__ (python 3)
**Installation**
===================
Stable
------
The ```pip``` version is the last most stable release.
```sh
$ pip install pandas_ta
```
## __Latest Version__
Latest Version
--------------
Best choice!
```sh
$ pip install -U git+https://github.com/twopirllc/pandas-ta
```
## __Quick Start__ using the DataFrame Extension
<br/>
# **Quick Start**
```python
import pandas as pd
import pandas_ta as ta
# Load data
df = pd.read_csv("path/symbol.csv", sep=",")
df = pd.read_csv("path/to/symbol.csv", sep=",")
# Calculate Returns and append to the df DataFrame
df.ta.log_return(cumulative=True, append=True)
@@ -116,136 +97,172 @@ df.tail()
# vv Continue Post Processing vv
```
## __Module and Indicator Help__
<br/>
# **Help**
```python
import pandas as pd
import pandas_ta as ta
# Create a DataFrame so 'ta' can be used.
df = pd.DataFrame()
# Help about this, 'ta', extension
help(pd.DataFrame().ta)
help(df.ta)
# List of all indicators
pd.DataFrame().ta.indicators()
df.ta.indicators()
# Help about the log_return indicator
help(ta.log_return)
```
<br/>
## New Class: __Strategy__
### What is a Pandas TA Strategy?
A _Strategy_ is a simple way to name and group your favorite TA indicators. Technically, a _Strategy_ is a simple Data Class to contain list of indicators and their parameters. __Note__: _Strategy_ is experimental and subject to change. Pandas TA comes with two basic Strategies: __AllStrategy__ and __CommonStrategy__.
# **Issues and Contributions**
* See the [Pandas TA Strategy Examples](https://github.com/twopirllc/pandas-ta/tree/master/examples/PandasTA_Strategy_Examples.ipynb) Notebook for more Examples including _Indicator Composition/Chaining_.
Thanks for trying **Pandas TA**!
### Strategy Requirements:
* ### [Comments and Feedback](https://github.com/twopirllc/pandas-ta/issues)
* Have you read **_this_** document?
* Are you running the latest version?
* ```$ pip install -U git+https://github.com/twopirllc/pandas-ta```
* Have you tried the [Examples](https://github.com/twopirllc/pandas-ta/tree/master/examples/)?
* Did they help?
* What is missing?
* Could you help improve them?
* Did you know you can easily build _Custom Strategies_ with the **[Strategy](https://github.com/twopirllc/pandas-ta/blob/master/examples/PandasTA_Strategy_Examples.ipynb) Class**?
* Documentation could _always_ be improved. Can you help contribute?
* ### [Indicator or Feature Requests & Contributions](https://github.com/twopirllc/pandas-ta/issues)
* Please be as **detailed** as possible. Links, screenshots, and sometimes data samples are welcome.
* You want a new indicator not currently listed.
* You want an alternate version of an existing indicator.
* The indicator does not match another website, library, broker platform, language, et al.
* Do you have correlation analysis to back your claim?
* Can you contribute?
<br/>
**Contributors**
================
_Thank you for your contributions!_
[alexonab](https://github.com/alexonab) | [allahyarzadeh](https://github.com/allahyarzadeh) | [codesutras](https://github.com/codesutras) | [DrPaprikaa](https://github.com/DrPaprikaa) | [FGU1](https://github.com/FGU1) | [lluissalord](https://github.com/lluissalord) | [maxdignan](https://github.com/maxdignan) | [pbrumblay](https://github.com/pbrumblay) | [SoftDevDanial](https://github.com/SoftDevDanial) | [YuvalWein](https://github.com/YuvalWein)
<br/>
**Programming Conventions**
===========================
**Pandas TA** has three primary "styles" of processing Technical Indicators for your use case and/or requirements. They are: _Conventional_, _DataFrame Extension_, and the _Pandas TA Strategy_. Each with increasing levels of abstraction for ease of use. As you become more familiar with **Pandas TA**, the simplicity and speed of using a _Pandas TA Strategy_ may become more apparent. Furthermore, you can create your own indicators through Chaining or Composition. Lastly, each indicator either returns a _Series_ or a _DataFrame_ in Uppercase Underscore format regardless of style.
_Conventional_
====================
You explicitly define the input columns and take care of the output.
* ```sma10 = ta.sma(df["Close"], length=10)```
* Returns a series with name: ```SMA_10```
* ```donchiandf = ta.donchian(df["HIGH"], df["low"], lower_length=10, upper_length=15)```
* Returns a DataFrame named ```DC_10_15``` and column names: ```DCL_10_15, DCM_10_15, DCU_10_15```
* ```ema10_ohlc4 = ta.ema(ta.ohlc4(df["Open"], df["High"], df["Low"], df["Close"]), length=10)```
* Conventional Chaining is possible but more explicit.
* Since it returns a series named ```EMA_10```. If needed, you may need to uniquely name it.
_Pandas TA DataFrame Extension_
====================
Calling ```df.ta``` will automatically lowercase _OHLCVA_ to _ohlcva_: _open, high, low, close, volume_, _adj_close_. By default, ```df.ta``` will use the _ohlcva_ for the indicator arguments removing the need to specify input columns directly.
* ```sma10 = df.ta.sma(length=10)```
* Returns a series with name: ```SMA_10```
* ```ema10_ohlc4 = df.ta.ema(close=df.ta.ohlc4(), length=10, suffix="OHLC4")```
* Returns a series with name: ```EMA_10_OHLC4```
* Chaining Indicators _require_ specifying the input like: ```close=df.ta.ohlc4()```.
* ```donchiandf = df.ta.donchian(lower_length=10, upper_length=15)```
* Returns a DataFrame named ```DC_10_15``` and column names: ```DCL_10_15, DCM_10_15, DCU_10_15```
Same as the last three examples, but appending the results directly to the DataFrame ```df```.
* ```df.ta.sma(length=10, append=True)```
* Appends to ```df``` column name: ```SMA_10```.
* ```df.ta.ema(close=df.ta.ohlc4(append=True), length=10, suffix="OHLC4", append=True)```
* Chaining Indicators _require_ specifying the input like: ```close=df.ta.ohlc4()```.
* ```df.ta.donchian(lower_length=10, upper_length=15, append=True)```
* Appends to ```df``` with column names: ```DCL_10_15, DCM_10_15, DCU_10_15```.
_Pandas TA Strategy_
====================
A **Pandas TA** Strategy is a named group of indicators to be run by the _strategy_ method. All Strategies use **mulitprocessing** _except_ when using the ```col_names``` parameter (see [below](#multiprocessing)). There are different types of _Strategies_ listed in the following section.
<br/>
### Here are the previous _Styles_ implemented using a Strategy Class:
```python
# (1) Create the Strategy
MyStrategy = ta.Strategy(
name="DCSMA10",
ta=[
{"kind": "ohlc4"},
{"kind": "sma", "length": 10},
{"kind": "donchian", "lower_length": 10, "upper_length": 15},
{"kind": "ema", "close": "OHLC4", "length": 10, "suffix": "OHLC4"},
]
)
# (2) Run the Strategy
df.ta.strategy(MyStrategy, **kwargs)
```
<br/><br/>
# **Pandas TA** _Strategies_
The _Strategy_ Class is a simple way to name and group your favorite TA Indicators by using a _Data Class_. **Pandas TA** comes with two prebuilt basic Strategies to help you get started: __AllStrategy__ and __CommonStrategy__. A _Strategy_ can be as simple as the __CommonStrategy__ or as complex as needed using Composition/Chaining.
* When using the _strategy_ method, **all** indicators will be automatically appended to the DataFrame ```df```.
* You are using a Chained Strategy when you have the output of one indicator as input into one or more indicators in the same _Strategy_.
* **Note:** Use the 'prefix' and/or 'suffix' keywords to distinguish the composed indicator from it's default Series.
See the [Pandas TA Strategy Examples Notebook](https://github.com/twopirllc/pandas-ta/tree/master/examples/PandasTA_Strategy_Examples.ipynb) for examples including _Indicator Composition/Chaining_.
Strategy Requirements
---------------------
- _name_: Some short memorable string. _Note_: Case-insensitive "All" is reserved.
- _ta_: A list of dicts containing keyword arguments to identify the indicator and the indicator's arguments
- **Note:** A Strategy will fail when consumed by Pandas TA if there is no ```{"kind": "indicator name"}``` attribute. _Remember_ to check your spelling.
### Optional Requirements:
Optional Parameters
-------------------
- _description_: A more detailed description of what the Strategy tries to capture. Default: None
- _created_: At datetime string of when it was created. Default: Automatically generated.
#### Things to note:
- A Strategy will __fail__ when consumed by Pandas TA if there is no {"kind": "indicator name"} attribute. __Remember__ to check your spelling.
<br/>
#### Brief Examples
```python
# The Builtin All Default Strategy
ta.AllStrategy = ta.Strategy(
name="All",
description="All the indicators with their default settings. Pandas TA default.",
ta=None
)
Types of Strategies
=======================
# The Builtin Default (Example) Strategy.
ta.CommonStrategy = ta.Strategy(
name="Common Price and Volume SMAs",
description="Common Price SMAs: 10, 20, 50, 200 and Volume SMA: 20.",
ta=[
{"kind": "sma", "length": 10},
{"kind": "sma", "length": 20},
{"kind": "sma", "length": 50},
{"kind": "sma", "length": 200},
{"kind": "sma", "close": "volume", "length": 20, "prefix": "VOL"}
]
)
# Your Custom Strategy or whatever your TA composition
CustomStrategy = ta.Strategy(
name="Momo and Volatility",
description="SMA 50,200, BBANDS, RSI, MACD and Volume SMA 20",
ta=[
{"kind": "sma", "length": 50},
{"kind": "sma", "length": 200},
{"kind": "bbands", "length": 20},
{"kind": "rsi"},
{"kind": "macd", "fast": 8, "slow": 21},
{"kind": "sma", "close": "volume", "length": 20, "prefix": "VOLUME"},
]
)
```
## __DataFrame Method__: _strategy_ with Multiprocessing
The new __Pandas (TA)__ method __strategy__ is used to facilitate bulk indicator processing. By default, running ```df.ta.strategy()``` will append __all
applicable__ indicators to DataFrame ```df```. Utility methods like ```above```, ```below``` et al are not included, however they can be included with Custom Strategies.
* The ```ta.strategy()``` method is still __under development__ and subject to change until stable.
```python
# Runs and appends all indicators to the current DataFrame by default
# The resultant DataFrame will be large.
df.ta.strategy()
# Or the string "all"
df.ta.strategy("all")
# Or the ta.AllStrategy
df.ta.strategy(ta.AllStrategy)
# Use verbose if you want to make sure it is running.
df.ta.strategy(verbose=True)
# Use timed if you want to see how long it takes to run.
df.ta.strategy(timed=True)
# Maybe you do not want certain indicators.
# Just exclude (a list of) them.
df.ta.strategy(exclude=["bop", "mom", "percent_return", "wcp", "pvi"], verbose=True)
# Perhaps you want to use different values for indicators.
# This will run ALL indicators that have fast or slow as parameters.
# Check your results and exclude as necessary.
df.ta.strategy(fast=10, slow=50, verbose=True)
# Sanity check. Make sure all the columns are there
df.columns
```
## Running a Builtin, Categorical or Custom Strategy
### __Builtin__
## _Builtin_
```python
# Running the Builtin CommonStrategy as mentioned above
df.ta.strategy(ta.CommonStrategy)
# The Default Strategy is the ta.AllStrategy. The following are equivalent
# df.ta.strategy(ta.AllStrategy)
# df.ta.strategy("All")
# The Default Strategy is the ta.AllStrategy. The following are equivalent:
df.ta.strategy()
df.ta.strategy("All")
df.ta.strategy(ta.AllStrategy)
```
### __Categorical__
## _Categorical_
```python
# List of indicator categories
df.ta.categories
# Running a Categorical Strategy only requires the Category name
df.ta.strategy("Momentum") # Default values for all Momentum indicators
df.ta.strategy("overlap", length=27) # Override all 'length' attributes
df.ta.strategy("overlap", length=42) # Override all Overlap 'length' attributes
```
### __Custom__
## _Custom_
```python
# Create your own Custom Strategy
CustomStrategy = ta.Strategy(
@@ -264,38 +281,67 @@ CustomStrategy = ta.Strategy(
df.ta.strategy(CustomStrategy)
```
## __DataFrame Property__: _categories_
**Multiprocessing**
=======================
The **Pandas TA** _strategy_ method utilizes **multiprocessing** for bulk indicator processing of all Strategy types with **ONE EXCEPTION!** When using the ```col_names``` parameter to rename resultant column(s), the indicators in ```ta``` array will be ran in order.
```python
# List of Pandas TA categories
df = df.ta.categories
```
# Runs and appends all indicators to the current DataFrame by default
# The resultant DataFrame will be large.
df.ta.strategy()
# Or the string "all"
df.ta.strategy("all")
# Or the ta.AllStrategy
df.ta.strategy(ta.AllStrategy)
## __DataFrame Property__: _cores_
# Use verbose if you want to make sure it is running.
df.ta.strategy(verbose=True)
```python
# Set the number of cores to use for strategy multiprocessing
# Defaults to the number of cpus you have
# Use timed if you want to see how long it takes to run.
df.ta.strategy(timed=True)
# Choose the number of cores to use. Default is all available cores.
df.ta.cores = 4
# Returns the number of cores you set or your default number of cpus.
df.ta.cores
# Maybe you do not want certain indicators.
# Just exclude (a list of) them.
df.ta.strategy(exclude=["bop", "mom", "percent_return", "wcp", "pvi"], verbose=True)
# Perhaps you want to use different values for indicators.
# This will run ALL indicators that have fast or slow as parameters.
# Check your results and exclude as necessary.
df.ta.strategy(fast=10, slow=50, verbose=True)
# Sanity check. Make sure all the columns are there
df.columns
```
## __DataFrame Properties__: _reverse_ & _datetime_ordered_
<br/>
## Custom Strategy without Multiprocessing
**Remember** These will not be utilizing **multiprocessing**
```python
# The 'datetime_ordered' property returns True if the DataFrame
# index is of Pandas datetime64 and df.index[0] < df.index[-1]
# Otherwise it returns False
time_series_in_order = df.ta.datetime_ordered
# The 'reverse' is a helper property that returns the DataFrame
# in reverse order
df = df.ta.reverse
NonMPStrategy = ta.Strategy(
name="EMAs, BBs, and MACD",
description="Non Multiprocessing Strategy by rename Columns",
ta=[
{"kind": "ema", "length": 8},
{"kind": "ema", "length": 21},
{"kind": "bbands", "length": 20, "col_names": ("BBL", "BBM", "BBU")},
{"kind": "macd", "fast": 8, "slow": 21, "col_names": ("MACD", "MACD_H", "MACD_S")}
]
)
# Run it
df.ta.strategy(NonMPStrategy)
```
## __DataFrame Property__: *adjusted*
<br/><br/>
# **DataFrame Properties**
## **adjusted**
```python
# Set ta to default to an adjusted column, 'adj_close', overriding default 'close'
@@ -306,28 +352,72 @@ df.ta.sma(length=10, append=True)
df.ta.adjusted = None
```
## __DataFrame kwargs__: _prefix_ and _suffix_
## **categories**
```python
# List of Pandas TA categories
df.ta.categories
```
## **cores**
```python
# Set the number of cores to use for strategy multiprocessing
# Defaults to the number of cpus you have
df.ta.cores = 4
# Returns the number of cores you set or your default number of cpus.
df.ta.cores
```
## **datetime_ordered**
```python
# The 'datetime_ordered' property returns True if the DataFrame
# index is of Pandas datetime64 and df.index[0] < df.index[-1]
# Otherwise it returns False
df.ta.datetime_ordered
```
## **reverse**
```python
# The 'datetime_ordered' property returns True if the DataFrame
# index is of Pandas datetime64 and df.index[0] < df.index[-1]
# Otherwise it returns False
df.ta.datetime_ordered
# The 'reverse' is a helper property that returns the DataFrame
# in reverse order
df.ta.reverse
```
## **prefix & suffix**
```python
# Applying a prefix to the name of an indicator
prehl2 = df.ta.hl2(prefix="pre")
print(prehl2.name) # "pre_HL2"
# Applying a suffix to the name of an indicator
endhl2 = df.ta.hl2(suffix="post")
print(endhl2.name) # "HL2_post"
# Applying a prefix and suffix to the name of an indicator
bothhl2 = df.ta.hl2(prefix="pre", suffix="post")
print(bothhl2.name) # "pre_HL2_post"
```
# __Technical Analysis Indicators__ (_by Category_)
<br/><br/>
## _Candles_ (3)
# **Indicators** (_by Category_)
### **Candles** (3)
* _Doji_: **cdl_doji**
* _Inside Bar_: **cdl_inside**
* _Heikin-Ashi_: **ha**
## _Momentum_ (33)
### **Momentum** (34)
* _Awesome Oscillator_: **ao**
* _Absolute Price Oscillator_: **apo**
@@ -335,6 +425,7 @@ print(bothhl2.name) # "pre_HL2_post"
* _Balance of Power_: **bop**
* _BRAR_: **brar**
* _Commodity Channel Index_: **cci**
* _Chande Forecast Oscillator_: **cfo**
* _Center of Gravity_: **cg**
* _Chande Momentum Oscillator_: **cmo**
* _Coppock Curve_: **coppock**
@@ -369,7 +460,7 @@ print(bothhl2.name) # "pre_HL2_post"
|:--------:|
| ![Example MACD](/images/SPY_MACD.png) |
## _Overlap_ (27)
### **Overlap** (27)
* _Double Exponential Moving Average_: **dema**
* _Exponential Moving Average_: **ema**
@@ -405,7 +496,8 @@ print(bothhl2.name) # "pre_HL2_post"
|:--------:|
| ![Example Chart](/images/TA_Chart.png) |
## _Performance_ (3)
### **Performance** (3)
Use parameter: cumulative=**True** for cumulative results.
@@ -417,7 +509,8 @@ Use parameter: cumulative=**True** for cumulative results.
|:--------:|
| ![Example Cumulative Percent Return](/images/SPY_CumulativePercentReturn.png) |
## _Statistics_ (9)
### **Statistics** (9)
* _Entropy_: **entropy**
* _Kurtosis_: **kurtosis**
@@ -433,17 +526,18 @@ Use parameter: cumulative=**True** for cumulative results.
|:--------:|
| ![Example Z Score](/images/SPY_ZScore.png) |
## _Trend_ (15)
### **Trend** (15)
* _Average Directional Movement Index_: **adx**
* _Archer Moving Averages Trends_: **amat**
* _Aroon & Aroon Oscillator_: **aroon**
* _Choppiness Index_: **chop**
* _Chande Kroll Stop_: **cksp**
* _Decay_: **decay**
* Formally: **linear_decay**
* _Decreasing_: **decreasing**
* _Detrended Price Oscillator_: **dpo**
* _Increasing_: **increasing**
* _Linear Decay_: **linear_decay**
* _Long Run_: **long_run**
* _Parabolic Stop and Reverse_: **psar**
* _Q Stick_: **qstick**
@@ -455,7 +549,7 @@ Use parameter: cumulative=**True** for cumulative results.
|:--------:|
| ![Example ADX](/images/SPY_ADX.png) |
## _Utility_ (5)
### **Utility** (5)
* _Above_: **above**
* _Above Value_: **above_value**
@@ -463,7 +557,7 @@ Use parameter: cumulative=**True** for cumulative results.
* _Below Value_: **below_value**
* _Cross_: **cross**
## _Volatility_ (12)
### **Volatility** (12)
* _Aberration_: **aberration**
* _Acceleration Bands_: **accbands**
@@ -482,7 +576,7 @@ Use parameter: cumulative=**True** for cumulative results.
|:--------:|
| ![Example ATR](/images/SPY_ATR.png) |
## _Volume_ (13)
### **Volume** (13)
* _Accumulation/Distribution Index_: **ad**
* _Accumulation/Distribution Oscillator_: **adosc**
@@ -502,17 +596,37 @@ Use parameter: cumulative=**True** for cumulative results.
|:--------:|
| ![Example OBV](/images/SPY_OBV.png) |
<br/><br/>
# Contributors
* [alexonab](https://github.com/alexonab)
* [allahyarzadeh](https://github.com/allahyarzadeh)
* [DrPaprikaa](https://github.com/DrPaprikaa)
* [FGU1](https://github.com/FGU1)
* [lluissalord](https://github.com/lluissalord)
* [SoftDevDanial](https://github.com/SoftDevDanial)
* [YuvalWein](https://github.com/YuvalWein)
# **Changes**
## **Recent**
* A __Strategy__ Class to help name and group your favorite indicators.
* An _experimental_ and independent __Watchlist__ Class located in the [Examples](https://github.com/twopirllc/pandas-ta/tree/master/examples/watchlist.py) Directory that can be used in conjunction with the new __Strategy__ Class.
* _Linear Regression_ (**linear_regression**) is a new utility method for Simple Linear Regression using _Numpy_ or _Scikit Learn_'s implementation.
# Inspiration
* Original TA-LIB: http://ta-lib.org/
* TradingView: http://www.tradingview.com
## **Breaking**
* _Stochastic Oscillator_ (**stoch**): Now in line with Trading View's calculation. See: ```help(ta.stoch)```
* _Linear Decay_ (**linear_decay**): Renamed to _Decay_ (**decay**) and with the option for Exponential decay using ```mode="exp"```. See: ```help(ta.decay)```
## **New**
* _Chande Forecast Oscillator_ (**cfo**) It calculates the percentage difference between the actual price and the Time Series Forecast (the endpoint of a linear regression line).
* _Gann High-Low Activator_ (**hilo**) The Gann High Low Activator Indicator was created by Robert Krausz in a 1998.
* _Inside Bar_ (**cdl_inside**) An Inside Bar is a bar contained within it's previous bar's high and low See: ```help(ta.cdl_inside)```
* _SMI Ergodic_ (**smi**) Developed by William Blau, the SMI Ergodic Indicator is the same as the True Strength Index (TSI) except the SMI includes a signal line and oscillator.
* _Squeeze_ (**squeeze**). A Momentum indicator. Both John Carter's TTM **and** Lazybear's TradingView versions are implemented. The default is John Carter's, or ```lazybear=False```. Set ```lazybear=True``` to enable Lazybear's.
* _Stochastic RSI_ (**stochrsi**) "Stochastic RSI and Dynamic Momentum Index" was created by Tushar Chande and Stanley Kroll. In line with Trading View's calculation. See: ```help(ta.stochrsi)```
* _TTM Trend_ (**ttm_trend**). A trend indicator inspired from John Carter's book "Mastering the Trade".
issue of Stocks & Commodities Magazine. It is a moving average based trend
indicator consisting of two different simple moving averages.
## **Updated**
* _Trend Return_ (**trend_return**): Returns a DataFrame now instead of Series with pertinenet trade info for a _trend_. An example can be found in the [AI Example Notebook](https://github.com/twopirllc/pandas-ta/tree/master/examples/AIExample.ipynb). The notebook is still a work in progress and open to colloboration.
# **Sources**
* [Original TA-LIB](http://ta-lib.org/)
* [TradingView](http://www.tradingview.com)
* [Sierra Chart](https://search.sierrachart.com/?Query=indicators&submitted=true)
* [FM Labs](https://www.fmlabs.com/reference/default.htm)
* [User 42](https://user42.tuxfamily.org/chart/manual/index.html)
+5242 -702
View File
File diff suppressed because it is too large Load Diff
+9 -9
View File
@@ -45,7 +45,7 @@
"Numpy v1.18.3\n",
"Pandas v1.1.0\n",
"mplfinance v0.12.6a3\n",
"Pandas TA v0.2.02b\n"
"Pandas TA v0.1.72b0\n"
]
}
],
@@ -91,7 +91,7 @@
" # All Data: 0, Last Four Years: 0.25, Last Two Years: 0.5, This Year: 1, Last Half Year: 2, Last Quarter: 4\n",
" yearly_divisor = {\"all\": 0, \"10y\": 0.1, \"5y\": 0.2, \"4y\": 0.25, \"3y\": 1./3, \"2y\": 0.5, \"1y\": 1, \"6mo\": 2, \"3mo\": 4}\n",
" yd = yearly_divisor[tf] if tf in yearly_divisor.keys() else 0\n",
" return int(ta.TRADING_DAYS_PER_YEAR / yd) if yd > 0 else df.shape[0]"
" return int(ta.RATE[\"TRADING_DAYS_PER_YEAR\"] / yd) if yd > 0 else df.shape[0]"
]
},
{
@@ -112,13 +112,13 @@
"text": [
"[!] Loading All: SPY, QQQ, AAPL, TSLA\n",
"[i] Loaded['D']: SPY_D.csv\n",
"[i] Runtime: 35.2548 ms (0.0353 s)\n",
"[i] Runtime: 829.8970 ms (0.8299 s)\n",
"[i] Loaded['D']: QQQ_D.csv\n",
"[i] Runtime: 85.3897 ms (0.0854 s)\n",
"[i] Runtime: 899.8845 ms (0.8999 s)\n",
"[i] Loaded['D']: AAPL_D.csv\n",
"[i] Runtime: 20.9833 ms (0.0210 s)\n",
"[i] Runtime: 813.8567 ms (0.8139 s)\n",
"[i] Loaded['D']: TSLA_D.csv\n",
"[i] Runtime: 34.4865 ms (0.0345 s)\n"
"[i] Runtime: 832.0968 ms (0.8321 s)\n"
]
}
],
@@ -353,7 +353,7 @@
{
"data": {
"text/plain": [
"<matplotlib.axes._subplots.AxesSubplot at 0x11b509040>"
"<matplotlib.axes._subplots.AxesSubplot at 0x10eeb55b0>"
]
},
"execution_count": 9,
@@ -396,7 +396,7 @@
{
"data": {
"text/plain": [
"<matplotlib.axes._subplots.AxesSubplot at 0x10f4258e0>"
"<matplotlib.axes._subplots.AxesSubplot at 0x10f10cf40>"
]
},
"execution_count": 10,
@@ -441,7 +441,7 @@
{
"data": {
"text/plain": [
"<matplotlib.axes._subplots.AxesSubplot at 0x11b876460>"
"<matplotlib.axes._subplots.AxesSubplot at 0x10f24f100>"
]
},
"execution_count": 11,
File diff suppressed because it is too large Load Diff
+186 -212
View File
File diff suppressed because one or more lines are too long
+30 -23
View File
@@ -8,7 +8,7 @@ import pandas as pd # pip install pandas
import yfinance as yf
# yf.pdr_override() # <== that's all it takes :-)
from alphaVantageAPI.alphavantage import AlphaVantage # pip install alphaVantage-api
import alphaVantageAPI as AV # pip install alphaVantage-api
import pandas_ta as ta # pip install pandas_ta
@@ -50,20 +50,24 @@ def colors(colors: str = None, default: str = "GrRd"):
class Watchlist(object):
"""Watchlist Class (** This is subject to change! **)
============================================================================
"""
# Watchlist Class (** This is subject to change! **)
A simple Class to load/download financial market data and automatically
apply Technical Analysis indicators with a Pandas TA Strategy. Default
Strategy: pandas_ta.AllStrategy.
apply Technical Analysis indicators with a Pandas TA Strategy.
Requirements:
Default Strategy: pandas_ta.CommonStrategy
## Package Support:
### Data Source (Default: AlphaVantage)
- AlphaVantage (pip install alphaVantage-api).
- Python Binance (pip install python-binance). # Future Support
- Yahoo Finance (pip install yfinance). # Almost Supported
# Technical Analysis:
- Pandas TA (pip install pandas_ta)
- AlphaVantage (pip install alphaVantage-api) for the Default Data Source.
To use another Data Source, update the load() method after AV.
Required Arguments:
- tickers: A list of strings containing tickers. Example: ['SPY', 'AAPL']
============================================================================
## Required Arguments:
- tickers: A list of strings containing tickers. Example: ["SPY", "AAPL"]
"""
def __init__(
self,
@@ -74,24 +78,28 @@ class Watchlist(object):
ds: object = None,
**kwargs
):
self.tickers = tickers
self.tf = tf
self.verbose = kwargs.pop("verbose", False)
self.debug = kwargs.pop("debug", False)
self.name = name
self.tickers = tickers
self.tf = tf
self.name = name if isinstance(name, str) else f"Watch: {', '.join(tickers)}"
self.data = None
self.kwargs = kwargs
self.strategy = strategy
self._init_data_source(ds)
def _init_data_source(self, ds: object):
if ds is not None:
self.ds = ds
elif isinstance(ds, str) and ds.lower() == "yahoo":
self.ds = yf
else:
AVkwargs = {"api_key": "YOUR API KEY","clean": True, "export": True, "export_path": ".", "output_size": "full", "premium": False}
av_kwargs = kwargs.pop("av_kwargs", AVkwargs)
self.ds = AlphaVantage(**av_kwargs)
AVkwargs = {"api_key": "YOUR API KEY", "clean": True, "export": True, "export_path": ".", "output_size": "full", "premium": False}
self.av_kwargs = self.kwargs.pop("av_kwargs", AVkwargs)
self.file_path = self.av_kwargs["export_path"]
self.ds = AV.AlphaVantage(**self.av_kwargs)
def _drop_columns(self, df: pd.DataFrame, cols: list = ["Unnamed: 0", "date", "split_coefficient", "dividend"]):
"""Helper methods to drop columns silently."""
@@ -115,7 +123,6 @@ class Watchlist(object):
tf: str = None,
index: str = "date",
drop: list = ["dividend", "split_coefficient"],
file_path: str = ".",
**kwargs
) -> pd.DataFrame:
"""Loads or Downloads (if a local csv does not exist) the data from the
@@ -131,18 +138,18 @@ class Watchlist(object):
return
filename_ = f"{ticker}_{tf}.csv"
current_file = Path(file_path) / filename_
current_file = Path(self.file_path) / filename_
# Load local or from Data Source
if current_file.exists():
df = pd.read_csv(filename_, index_col=index)
df = pd.read_csv(current_file, index_col=index)
if not df.ta.datetime_ordered:
df = df.set_index(pd.DatetimeIndex(df.index))
print(f"[i] Loaded['{tf}']: {filename_}")
else:
print(f"[+] Downloading['{tf}']: {ticker}")
if isinstance(self.ds, AlphaVantage):
df = self.ds.data(tf, ticker)
if isinstance(self.ds, AV.AlphaVantage):
df = self.ds.data(ticker, tf)
if not df.ta.datetime_ordered:
df = df.set_index(pd.DatetimeIndex(df[index]))
elif isinstance(self.ds, yfinance):
+29 -2
View File
@@ -18,6 +18,17 @@ except DistributionNotFound:
else:
__version__ = _dist.version
from importlib.util import find_spec
Imports = {
"scipy": find_spec("scipy") is not None,
"sklearn": find_spec("sklearn") is not None,
"statsmodels": find_spec("statsmodels") is not None,
"mplfinance": find_spec("mplfinance") is not None,
"alphaVantage-api ": find_spec("alphaVantageAPI") is not None,
"yfinance": find_spec("yfinance") is not None,
"talib": find_spec("talib") is not None
}
# Not ideal and not dynamic but it works.
# Will find a dynamic solution later.
Category = {
@@ -25,7 +36,7 @@ Category = {
"candles": ["cdl_doji", "cdl_inside", "ha"],
# Momentum
"momentum": ["ao", "apo", "bias", "bop", "brar", "cci", "cg", "cmo", "coppock", "er", "eri", "fisher", "inertia", "kdj", "kst", "macd", "mom", "pgo", "ppo", "psl", "pvo", "roc", "rsi", "rvgi", "slope", "smi", "squeeze", "stoch", "stochrsi", "trix", "tsi", "uo", "willr"],
"momentum": ["ao", "apo", "bias", "bop", "brar", "cci", "cfo", "cg", "cmo", "coppock", "er", "eri", "fisher", "inertia", "kdj", "kst", "macd", "mom", "pgo", "ppo", "psl", "pvo", "roc", "rsi", "rvgi", "slope", "smi", "squeeze", "stoch", "stochrsi", "trix", "tsi", "uo", "willr"],
# Overlap
"overlap": ["dema", "ema", "fwma", "hilo", "hl2", "hlc3", "hma", "ichimoku", "kama", "linreg", "midpoint", "midprice", "ohlc4", "pwma", "rma", "sinwma", "sma", "supertrend", "swma", "t3", "tema", "trima", "vwap", "vwma", "wcp", "wma", "zlma"],
@@ -37,7 +48,7 @@ Category = {
"statistics": ["entropy", "kurtosis", "mad", "median", "quantile", "skew", "stdev", "variance", "zscore"],
# Trend
"trend": ["adx", "amat", "aroon", "chop", "cksp", "decreasing", "dpo", "increasing", "linear_decay", "long_run", "psar", "qstick", "short_run", "ttm_trend", "vortex"],
"trend": ["adx", "amat", "aroon", "chop", "cksp", "decay", "decreasing", "dpo", "increasing", "long_run", "psar", "qstick", "short_run", "ttm_trend", "vortex"],
# Volatility
"volatility": ["aberration", "accbands", "atr", "bbands", "donchian", "kc", "massi", "natr", "pdist", "rvi", "true_range", "ui"],
@@ -46,4 +57,20 @@ Category = {
"volume": ["ad", "adosc", "aobv", "cmf", "efi", "eom", "mfi", "nvi", "obv", "pvi", "pvol", "pvt"],
}
# https://www.worldtimezone.com/markets24.php
EXCHANGE_TZ = {
"NZSX": 12, "ASX": 11,
"TSE": 9, "HKE": 8, "SSE": 8, "SGX": 8,
"NSE": 5.5, "DIFX": 4, "RTS": 3,
"JSE": 2, "FWB": 1, "LSE": 1,
"BMF": -2, "NYSE": -4, "TSX": -4
}
RATE = {
"TRADING_DAYS_PER_YEAR": 252, # Keep even
"TRADING_HOURS_PER_DAY": 6.5,
"MINUTES_PER_HOUR": 60
}
from pandas_ta.core import *
+690 -827
View File
File diff suppressed because it is too large Load Diff
+2 -1
View File
@@ -7,6 +7,7 @@ from .brar import brar
from .cci import cci
from .cg import cg
from .cmo import cmo
from .cfo import cfo
from .coppock import coppock
from .er import er
from .eri import eri
@@ -31,4 +32,4 @@ from .stochrsi import stochrsi
from .trix import trix
from .tsi import tsi
from .uo import uo
from .willr import willr
from .willr import willr
+63
View File
@@ -0,0 +1,63 @@
# -*- coding: utf-8 -*-
from pandas_ta.overlap import linreg
from pandas_ta.utils import get_drift, get_offset, verify_series
def cfo(close, length=None, scalar=None, drift=None, offset=None, **kwargs):
"""Indicator: Chande Forcast Oscillator (CFO)"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 9
scalar = float(scalar) if scalar else 100
drift = get_drift(drift)
offset = get_offset(offset)
#Finding linear regression of Series
cfo = scalar * (close - linreg(close, length=length, tsf=True))
cfo /= close
# Offset
if offset != 0:
cfo = cfo.shift(offset)
# Handle fills
if "fillna" in kwargs:
cfo.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
cfo.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
cfo.name = f"CFO_{length}"
cfo.category = "momentum"
return cfo
cfo.__doc__ = \
"""Chande Forcast Oscillator (CFO)
The Forecast Oscillator calculates the percentage difference between the actual
price and the Time Series Forecast (the endpoint of a linear regression line).
Sources:
https://www.fmlabs.com/reference/default.htm?url=ForecastOscillator.htm
Calculation:
Default Inputs:
length=9, drift=1, scalar=100
LINREG = Linear Regression
CFO = scalar * (close - LINERREG(length, tdf=True)) / close
Args:
close (pd.Series): Series of 'close's
length (int): The period. Default: 9
scalar (float): How much to magnify. Default: 100
drift (int): The short period. Default: 1
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.
"""
+9 -9
View File
@@ -1,14 +1,14 @@
# -*- coding: utf-8 -*-
from math import atan, pi
from ..utils import get_offset, verify_series
from pandas_ta.utils import get_offset, verify_series
def slope(close, length=None, as_angle=None, to_degrees=None, offset=None, **kwargs):
def slope(close, length=None, as_angle=None, to_degrees=None, vertical=None, offset=None, **kwargs):
"""Indicator: Slope"""
# Validate arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 1
as_angle = True if as_angle and isinstance(as_angle, bool) else False
to_degrees = True if to_degrees and isinstance(to_degrees, bool) else False
as_angle = True if isinstance(as_angle, bool) else False
to_degrees = True if isinstance(to_degrees, bool) else False
offset = get_offset(offset)
# Calculate Result
@@ -23,14 +23,14 @@ def slope(close, length=None, as_angle=None, to_degrees=None, offset=None, **kwa
slope = slope.shift(offset)
# Handle fills
if 'fillna' in kwargs:
slope.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
slope.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
slope.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
slope.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
slope.name = f"SLOPE_{length}" if not as_angle else f"ANGLE{'d' if to_degrees else 'r'}_{length}"
slope.category = 'momentum'
slope.category = "momentum"
return slope
+22 -20
View File
@@ -1,20 +1,20 @@
# -*- coding: utf-8 -*-
import math
from ..utils import get_offset, verify_series
from pandas_ta.utils import get_offset, verify_series
def linreg(close, length=None, offset=None, **kwargs):
"""Indicator: Linear Regression"""
# Validate arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 14
min_periods = int(kwargs['min_periods']) if 'min_periods' in kwargs and kwargs['min_periods'] is not None else length
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
offset = get_offset(offset)
angle = kwargs.pop('angle', False)
intercept = kwargs.pop('intercept', False)
degrees = kwargs.pop('degrees', False)
r = kwargs.pop('r', False)
slope = kwargs.pop('slope', False)
tsf = kwargs.pop('tsf', False)
angle = kwargs.pop("angle", False)
intercept = kwargs.pop("intercept", False)
degrees = kwargs.pop("degrees", False)
r = kwargs.pop("r", False)
slope = kwargs.pop("slope", False)
tsf = kwargs.pop("tsf", False)
# Calculate Result
x = range(1, length + 1) # [1, 2, ..., n] from 1 to n keeps Sum(xy) low
@@ -54,10 +54,10 @@ def linreg(close, length=None, offset=None, **kwargs):
linreg = linreg.shift(offset)
# Handle fills
if 'fillna' in kwargs:
linreg.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
linreg.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
linreg.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
linreg.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
linreg.name = f"LR"
@@ -66,7 +66,7 @@ def linreg(close, length=None, offset=None, **kwargs):
if angle: linreg.name += "a"
if r: linreg.name += "r"
linreg.name += f"_{length}"
linreg.category = 'overlap'
linreg.category = "overlap"
return linreg
@@ -75,7 +75,9 @@ def linreg(close, length=None, offset=None, **kwargs):
linreg.__doc__ = \
"""Linear Regression Moving Average (linreg)
Linear Regression Moving Average
Linear Regression Moving Average (LINREG). This is a simplified version of a
Standard Linear Regression. LINREG is a rolling regression of one variable. A
Standard Linear Regression is between two or more variables.
Source: TA Lib
@@ -104,12 +106,12 @@ Args:
offset (int): How many periods to offset the result. Default: 0
Kwargs:
angle (bool, optional): Default: False. If True, returns the angle of the slope in radians
degrees (bool, optional): Default: False. If True, returns the angle of the slope in degrees
intercept (bool, optional): Default: False. If True, returns the angle of the slope in radians
r (bool, optional): Default: False. If True, returns it's correlation 'r'
slope (bool, optional): Default: False. If True, returns the slope
tsf (bool, optional): Default: False. If True, returns the Time Series Forecast value.
angle (bool, optional): Default: False. If True, returns the angle of the slope in radians
degrees (bool, optional): Default: False. If True, returns the angle of the slope in degrees
intercept (bool, optional): Default: False. If True, returns the angle of the slope in radians
r (bool, optional): Default: False. If True, returns it's correlation 'r'
slope (bool, optional): Default: False. If True, returns the slope
tsf (bool, optional): Default: False. If True, returns the Time Series Forecast value.
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
+1 -1
View File
@@ -24,7 +24,7 @@ def vwap(high, low, close, volume, offset=None, **kwargs):
# Name & Category
vwap.name = "VWAP"
vwap.category = 'overlap'
vwap.category = "overlap"
return vwap
+1 -1
View File
@@ -4,10 +4,10 @@ from .amat import amat
from .aroon import aroon
from .chop import chop
from .cksp import cksp
from .decay import decay
from .decreasing import decreasing
from .dpo import dpo
from .increasing import increasing
from .linear_decay import linear_decay
from .long_run import long_run
from .psar import psar
from .qstick import qstick
+11 -10
View File
@@ -12,28 +12,29 @@ def amat(close=None, fast=None, slow=None, mamode=None, lookback=None, offset=No
fast = int(fast) if fast and fast > 0 else 8
slow = int(slow) if slow and slow > 0 else 21
lookback = int(lookback) if lookback and lookback > 0 else 2
mamode = mamode.upper() if mamode else "EMA"
mamode = mamode.lower() if mamode else "ema"
offset = get_offset(offset)
# Calculate Result
if mamode == "EMA":
fast_ma = ema(close=close, length=fast, **kwargs)
slow_ma = ema(close=close, length=slow, **kwargs)
elif mamode == "HMA":
if mamode == "hma":
fast_ma = hma(close=close, length=fast, **kwargs)
slow_ma = hma(close=close, length=slow, **kwargs)
elif mamode == "LINREG":
elif mamode == "linreg":
fast_ma = linreg(close=close, length=fast, **kwargs)
slow_ma = linreg(close=close, length=slow, **kwargs)
elif mamode == "RMA":
elif mamode == "rma":
fast_ma = rma(close=close, length=fast, **kwargs)
slow_ma = rma(close=close, length=slow, **kwargs)
elif mamode == "SMA":
elif mamode == "sma":
fast_ma = sma(close=close, length=fast, **kwargs)
slow_ma = sma(close=close, length=slow, **kwargs)
elif mamode == "WMA":
elif mamode == "wma":
fast_ma = wma(close=close, length=fast, **kwargs)
slow_ma = wma(close=close, length=slow, **kwargs)
else: # "ema"
fast_ma = ema(close=close, length=fast, **kwargs)
slow_ma = ema(close=close, length=slow, **kwargs)
mas_long = long_run(fast_ma, slow_ma, length=lookback)
mas_short = short_run(fast_ma, slow_ma, length=lookback)
@@ -59,7 +60,7 @@ def amat(close=None, fast=None, slow=None, mamode=None, lookback=None, offset=No
})
# Name and Categorize it
amatdf.name = f"AMAT_{mamode}_{fast}_{slow}_{lookback}"
amatdf.name = f"AMAT_{mamode.upper()}_{fast}_{slow}_{lookback}"
amatdf.category = "trend"
return amatdf
@@ -1,16 +1,23 @@
# -*- coding: utf-8 -*-
from math import exp
from pandas import DataFrame
from pandas_ta.utils import get_offset, verify_series
def linear_decay(close, length=None, offset=None, **kwargs):
"""Indicator: Linear Decay"""
def decay(close, kind=None, length=None, mode=None, offset=None, **kwargs):
"""Indicator: Decay"""
# Validate Arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 5
mode = mode.lower() if isinstance(mode, str) else "linear"
offset = get_offset(offset)
# Calculate Result
diff = close.shift(1) - (1 / length)
_mode = "L"
if mode == "exp" or kind == "exponential":
_mode = "EXP"
diff = close.shift(1) - exp(-length)
else: # "linear"
diff = close.shift(1) - (1 / length)
diff[0] = close[0]
tdf = DataFrame({"close": close, "diff": diff, "0": 0})
ld = tdf.max(axis=1)
@@ -25,31 +32,37 @@ def linear_decay(close, length=None, offset=None, **kwargs):
if "fill_method" in kwargs:
ld.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
ld.name = f"LDECAY_{length}"
# Name and Categorize it
ld.name = f"{_mode}DECAY_{length}"
ld.category = "trend"
return ld
linear_decay.__doc__ = \
"""Linear Decay
decay.__doc__ = \
"""Decay
Adds a linear decay moving forward from prior signals like crosses.
Creates a decay moving forward from prior signals like crosses. The default is
"linear". Exponential is optional as "exponential" or "exp".
Sources:
https://tulipindicators.org/decay
Calculation:
Default Inputs:
length=5
max(close, close[-1] - (1 / length), 0)
length=5, mode=None
if mode == "exponential" or mode == "exp":
max(close, close[-1] - exp(-length), 0)
else:
max(close, close[-1] - (1 / length), 0)
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 1
offset (int): How many periods to offset the result. Default: 0
length (int): It's period. Default: 1
mamode (str): Option "exponential" ("exp"). Default: 'linear' or None
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
-452
View File
@@ -1,452 +0,0 @@
# -*- coding: utf-8 -*-
import math
from pathlib import Path
from time import perf_counter
from numpy import argmax, argmin, dot, ones, triu
from numpy import append as npAppend
from numpy import array as npArray
from numpy import ndarray as npNdArray
from numpy import sum as npSum
from pandas import DataFrame, Series
from pandas.api.types import is_datetime64_any_dtype
from functools import reduce
from operator import mul
from sys import float_info as sflt
TRADING_DAYS_PER_YEAR = 252 # Keep even
TRADING_HOURS_PER_DAY = 6.5
MINUTES_PER_HOUR = 60
def _above_below(
series_a: Series,
series_b: Series,
above: bool = True,
asint: bool = True,
offset: int = None,
**kwargs
):
series_a = verify_series(series_a)
series_b = verify_series(series_b)
offset = get_offset(offset)
series_a.apply(zero)
series_b.apply(zero)
# Calculate Result
if above:
current = series_a >= series_b
else:
current = series_a <= series_b
if asint:
current = current.astype(int)
# Offset
if offset != 0:
current = current.shift(offset)
# Name & Category
current.name = f"{series_a.name}_{'A' if above else 'B'}_{series_b.name}"
current.category = "utility"
return current
def above(
series_a: Series,
series_b: Series,
asint: bool = True,
offset: int = None,
**kwargs
):
return _above_below(series_a, series_b, above=True, asint=asint, offset=offset, **kwargs)
def above_value(
series_a: Series,
value: float,
asint: bool = True,
offset: int = None,
**kwargs
):
if not isinstance(value, (int, float, complex)):
print("[X] value is not a number")
return
series_b = Series(value, index=series_a.index, name=f"{value}".replace(".","_"))
return _above_below(series_a, series_b, above=True, asint=asint, offset=offset, **kwargs)
def below(
series_a: Series,
series_b: Series,
asint: bool =True,
offset: int =None
,**kwargs
):
return _above_below(series_a, series_b, above=False, asint=asint, offset=offset, **kwargs)
def below_value(
series_a: Series,
value: float,
asint: bool = True,
offset: int = None,
**kwargs
):
if not isinstance(value, (int, float, complex)):
print("[X] value is not a number")
return
series_b = Series(value, index=series_a.index, name=f"{value}".replace(".","_"))
return _above_below(series_a, series_b, above=False, asint=asint, offset=offset, **kwargs)
def category_files(category: str) -> list:
"""Helper function to return all filenames in the category directory."""
files = [x.stem for x in list(Path(f"pandas_ta/{category}/").glob("*.py")) if x.stem != "__init__"]
return files
def combination(**kwargs):
"""https://stackoverflow.com/questions/4941753/is-there-a-math-ncr-function-in-python"""
n = int(math.fabs(kwargs.pop("n", 1)))
r = int(math.fabs(kwargs.pop("r", 0)))
if kwargs.pop("repetition", False) or kwargs.pop("multichoose", False):
n = n + r - 1
# if r < 0: return None
r = min(n, n - r)
if r == 0:
return 1
numerator = reduce(mul, range(n, n - r, -1), 1)
denominator = reduce(mul, range(1, r + 1), 1)
return numerator // denominator
def cross_value(
series_a: Series,
value: float,
above: bool = True,
asint: bool = True,
offset: int = None,
**kwargs
):
series_b = Series(value, index=series_a.index, name=f"{value}".replace(".","_"))
return cross(series_a, series_b, above, asint, offset, **kwargs)
def cross(
series_a: Series,
series_b: Series,
above: bool = True,
asint: bool = True,
offset: int = None,
**kwargs
):
series_a = verify_series(series_a)
series_b = verify_series(series_b)
offset = get_offset(offset)
series_a.apply(zero)
series_b.apply(zero)
# Calculate Result
current = series_a > series_b # current is above
previous = series_a.shift(1) < series_b.shift(1) # previous is below
# above if both are true, below if both are false
cross = current & previous if above else ~current & ~previous
if asint:
cross = cross.astype(int)
# Offset
if offset != 0:
cross = cross.shift(offset)
# Name & Category
cross.name = f"{series_a.name}_{'XA' if above else 'XB'}_{series_b.name}"
cross.category = "utility"
return cross
def is_datetime_ordered(df: DataFrame or Series) -> bool:
"""Returns True if the index is a datetime and ordered."""
index_is_datetime = is_datetime64_any_dtype(df.index)
try:
ordered = df.index[0] < df.index[-1]
except RuntimeWarning: pass
finally:
return True if index_is_datetime and ordered else False
def signals(indicator, xa, xb, cross_values, xserie, xserie_a, xserie_b, cross_series, offset) -> DataFrame:
df = DataFrame()
if xa is not None and isinstance(xa, (int, float)):
if cross_values:
crossed_above_start = cross_value(indicator, xa, above=True, offset=offset)
crossed_above_end = cross_value(indicator, xa, above=False, offset=offset)
df[crossed_above_start.name] = crossed_above_start
df[crossed_above_end.name] = crossed_above_end
else:
crossed_above = above_value(indicator, xa, offset=offset)
df[crossed_above.name] = crossed_above
if xb is not None and isinstance(xb, (int, float)):
if cross_values:
crossed_below_start = cross_value(indicator, xb, above=True, offset=offset)
crossed_below_end = cross_value(indicator, xb, above=False, offset=offset)
df[crossed_below_start.name] = crossed_below_start
df[crossed_below_end.name] = crossed_below_end
else:
crossed_below = below_value(indicator, xb, offset=offset)
df[crossed_below.name] = crossed_below
# xseries is the default value for both xserie_a and xserie_b
if xserie_a is None:
xserie_a = xserie
if xserie_b is None:
xserie_b = xserie
if xserie_a is not None and verify_series(xserie_a):
if cross_series:
cross_serie_above = cross(indicator, xserie_a, above=True, offset=offset)
else:
cross_serie_above = above(indicator, xserie_a, offset=offset)
df[cross_serie_above.name] = cross_serie_above
if xserie_b is not None and verify_series(xserie_b):
if cross_series:
cross_serie_below = cross(indicator, xserie_b, above=False, offset=offset)
else:
cross_serie_below = below(indicator, xserie_b, offset=offset)
df[cross_serie_below.name] = cross_serie_below
return df
def df_error_analysis(dfA: DataFrame, dfB: DataFrame, **kwargs) -> DataFrame:
"""DataFrame Correlation Analysis helper"""
corr_method = kwargs.pop("corr_method", "pearson")
# Find their differences and correlation
diff = dfA - dfB
corr = dfA.corr(dfB, method=corr_method)
# For plotting
if kwargs.pop("plot", False):
diff.hist()
if diff[diff > 0].any():
diff.plot(kind="kde")
if kwargs.pop("triangular", False):
return corr.where(triu(ones(corr.shape)).astype(bool))
return corr
def fibonacci(n: int = 2, **kwargs) -> npNdArray:
"""Fibonacci Sequence as a numpy array"""
n = int(math.fabs(n)) if n >= 0 else 2
zero = kwargs.pop("zero", False)
if zero:
a, b = 0, 1
else:
n -= 1
a, b = 1, 1
result = npArray([a])
for i in range(0, n):
a, b = b, a + b
result = npAppend(result, a)
weighted = kwargs.pop("weighted", False)
if weighted:
fib_sum = npSum(result)
if fib_sum > 0:
return result / fib_sum
else:
return result
else:
return result
def final_time(stime):
time_diff = perf_counter() - stime
return f"{time_diff * 1000:2.4f} ms ({time_diff:2.4f} s)"
def get_drift(x: int) -> int:
"""Returns an int if not zero, otherwise defaults to one."""
return int(x) if isinstance(x, int) and x != 0 else 1
def get_offset(x: int) -> int:
"""Returns an int, otherwise defaults to zero."""
return int(x) if isinstance(x, int) else 0
def is_percent(x: int or float) -> bool:
if isinstance(x, (int, float)):
return x is not None and x >= 0 and x <= 100
return False
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.
"""
diff = high - low
if diff.eq(0).any().any():
diff += sflt.epsilon
return diff
def pascals_triangle(n: int = None, **kwargs) -> npNdArray:
"""Pascal's Triangle
Returns a numpy array of the nth row of Pascal's Triangle.
n=4 => triangle: [1, 4, 6, 4, 1]
=> 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(math.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_weights = triangle / triangle_sum
inverse_weights = 1 - triangle_weights
weighted = kwargs.pop("weighted", False)
inverse = kwargs.pop("inverse", False)
if weighted and inverse:
return inverse_weights
if weighted:
return triangle_weights
if inverse:
return None
return triangle
def recent_maximum_index(x):
return int(argmax(x[::-1]))
def recent_minimum_index(x):
return int(argmin(x[::-1]))
def signed_series(series: Series, initial: int = None) -> Series:
"""Returns a Signed Series with or without an initial value
Default Example:
series = Series([3, 2, 2, 1, 1, 5, 6, 6, 7, 5])
and returns:
sign = Series([NaN, -1.0, 0.0, -1.0, 0.0, 1.0, 1.0, 0.0, 1.0, -1.0])
"""
series = verify_series(series)
sign = series.diff(1)
sign[sign > 0] = 1
sign[sign < 0] = -1
sign.iloc[0] = initial
return sign
def symmetric_triangle(n: int = None, **kwargs) -> list:
"""Symmetric Triangle with n >= 2
Returns a numpy array of the nth row of Symmetric Triangle.
n=4 => triangle: [1, 2, 2, 1]
=> weighted: [0.16666667 0.33333333 0.33333333 0.16666667]
"""
n = int(math.fabs(n)) if n is not None else 2
if n == 2:
triangle = [1, 1]
if n > 2:
if n % 2 == 0:
front = [i + 1 for i in range(0, math.floor(n/2))]
triangle = front + front[::-1]
else:
front = [i + 1 for i in range(0, math.floor(0.5 * (n + 1)))]
triangle = front.copy()
front.pop()
triangle += front[::-1]
if kwargs.pop("weighted", False):
triangle_sum = npSum(triangle)
triangle_weights = triangle / triangle_sum
return triangle_weights
return triangle
def unsigned_differences(series: Series, amount: int = None, **kwargs) -> Series:
"""Unsigned Differences
Returns two Series, an unsigned positive and unsigned negative series based
on the differences of the original series. The positive series are only the
increases and the negative series is only the decreases.
Default Example:
series = Series([3, 2, 2, 1, 1, 5, 6, 6, 7, 5, 3]) and returns
postive = Series([0, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0])
negative = Series([0, 1, 0, 1, 0, 0, 0, 0, 0, 1, 1])
"""
amount = int(amount) if amount is not None else 1
negative = series.diff(amount)
negative.fillna(0, inplace=True)
positive = negative.copy()
positive[positive <= 0] = 0
positive[positive > 0] = 1
negative[negative >= 0] = 0
negative[negative < 0] = 1
if kwargs.pop("asint", False):
positive = positive.astype(int)
negative = negative.astype(int)
return positive, negative
def verify_series(series: Series) -> Series:
"""If a Pandas Series return it."""
if series is not None and isinstance(series, Series):
return series
def weights(w):
def _dot(x):
return dot(w, x)
return _dot
def zero(x: [int, float]) -> [int, float]:
"""If the value is close to zero, then return zero.
Otherwise return the value."""
return 0 if abs(x) < sflt.epsilon else x
# Candle Functions
def candle_color(open_, close):
color = close.copy().astype(int)
color[close >= open_] = 1
color[close < open_] = -1
return color
def real_body(close, open_):
return non_zero_range(close, open_)
def high_low_range(high, low):
return non_zero_range(high, low)
+6
View File
@@ -0,0 +1,6 @@
# -*- coding: utf-8 -*-
from ._candles import *
from ._core import *
from ._math import *
from ._signals import *
from ._time import *
+19
View File
@@ -0,0 +1,19 @@
# -*- coding: utf-8 -*-
from pandas import Series
from ._core import non_zero_range
def candle_color(open_: Series, close: Series) -> Series:
color = close.copy().astype(int)
color[close >= open_] = 1
color[close < open_] = -1
return color
def high_low_range(high: Series, low: Series) -> Series:
return non_zero_range(high, low)
def real_body(close: Series, open_: Series) -> Series:
return non_zero_range(close, open_)
+109
View File
@@ -0,0 +1,109 @@
# -*- coding: utf-8 -*-
from pathlib import Path
from sys import float_info as sflt
from numpy import argmax, argmin
from pandas import DataFrame, Series
from pandas.api.types import is_datetime64_any_dtype
def category_files(category: str) -> list:
"""Helper function to return all filenames in the category directory."""
files = [x.stem for x in list(Path(f"pandas_ta/{category}/").glob("*.py")) if x.stem != "__init__"]
return files
def get_drift(x: int) -> int:
"""Returns an int if not zero, otherwise defaults to one."""
return int(x) if isinstance(x, int) and x != 0 else 1
def get_offset(x: int) -> int:
"""Returns an int, otherwise defaults to zero."""
return int(x) if isinstance(x, int) else 0
def is_datetime_ordered(df: DataFrame or Series) -> bool:
"""Returns True if the index is a datetime and ordered."""
index_is_datetime = is_datetime64_any_dtype(df.index)
try:
ordered = df.index[0] < df.index[-1]
except RuntimeWarning: pass
finally:
return True if index_is_datetime and ordered else False
def is_percent(x: int or float) -> bool:
if isinstance(x, (int, float)):
return x is not None and x >= 0 and x <= 100
return False
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'.
"""
diff = high - low
if diff.eq(0).any().any():
diff += sflt.epsilon
return diff
def recent_maximum_index(x):
return int(argmax(x[::-1]))
def recent_minimum_index(x):
return int(argmin(x[::-1]))
def signed_series(series: Series, initial: int = None) -> Series:
"""Returns a Signed Series with or without an initial value
Default Example:
series = Series([3, 2, 2, 1, 1, 5, 6, 6, 7, 5])
and returns:
sign = Series([NaN, -1.0, 0.0, -1.0, 0.0, 1.0, 1.0, 0.0, 1.0, -1.0])
"""
series = verify_series(series)
sign = series.diff(1)
sign[sign > 0] = 1
sign[sign < 0] = -1
sign.iloc[0] = initial
return sign
def unsigned_differences(series: Series, amount: int = None, **kwargs) -> Series:
"""Unsigned Differences
Returns two Series, an unsigned positive and unsigned negative series based
on the differences of the original series. The positive series are only the
increases and the negative series is only the decreases.
Default Example:
series = Series([3, 2, 2, 1, 1, 5, 6, 6, 7, 5, 3]) and returns
postive = Series([0, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0])
negative = Series([0, 1, 0, 1, 0, 0, 0, 0, 0, 1, 1])
"""
amount = int(amount) if amount is not None else 1
negative = series.diff(amount)
negative.fillna(0, inplace=True)
positive = negative.copy()
positive[positive <= 0] = 0
positive[positive > 0] = 1
negative[negative >= 0] = 0
negative[negative < 0] = 1
if kwargs.pop("asint", False):
positive = positive.astype(int)
negative = negative.astype(int)
return positive, negative
def verify_series(series: Series) -> Series:
"""If a Pandas Series return it."""
if series is not None and isinstance(series, Series):
return series
+215
View File
@@ -0,0 +1,215 @@
# -*- coding: utf-8 -*-
from functools import reduce
from math import fabs, floor
from operator import mul
from sys import float_info as sflt
from numpy import dot, ones, triu
from numpy import append as npAppend
from numpy import array as npArray
from numpy import corrcoef as npCorrcoef
from numpy import dot
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 ._core import verify_series
def combination(**kwargs):
"""https://stackoverflow.com/questions/4941753/is-there-a-math-ncr-function-in-python"""
n = int(fabs(kwargs.pop("n", 1)))
r = int(fabs(kwargs.pop("r", 0)))
if kwargs.pop("repetition", False) or kwargs.pop("multichoose", False):
n = n + r - 1
# if r < 0: return None
r = min(n, n - r)
if r == 0:
return 1
numerator = reduce(mul, range(n, n - r, -1), 1)
denominator = reduce(mul, range(1, r + 1), 1)
return numerator // denominator
def fibonacci(n: int = 2, **kwargs) -> npNdArray:
"""Fibonacci Sequence as a numpy array"""
n = int(fabs(n)) if n >= 0 else 2
zero = kwargs.pop("zero", False)
if zero:
a, b = 0, 1
else:
n -= 1
a, b = 1, 1
result = npArray([a])
for i in range(0, n):
a, b = b, a + b
result = npAppend(result, a)
weighted = kwargs.pop("weighted", False)
if weighted:
fib_sum = npSum(result)
if fib_sum > 0:
return result / fib_sum
else:
return result
else:
return result
def linear_regression(x: Series, y: Series) -> dict:
"""Classic Linear Regression in Numpy or Scikit-Learn"""
x = verify_series(x)
y = verify_series(y)
m, n = x.size, y.size
if m != n:
print(f"[X] Linear Regression X and y observations do not match: {m} != {n}")
return
if Imports["sklearn"]:
return _linear_regression_sklearn(x, y)
else:
return _linear_regression_np(x, y)
def pascals_triangle(n: int = None, **kwargs) -> npNdArray:
"""Pascal's Triangle
Returns a numpy array of the nth row of Pascal's Triangle.
n=4 => triangle: [1, 4, 6, 4, 1]
=> weighted: [0.0625, 0.25, 0.375, 0.25, 0.0625]
=> inverse weighted: [0.9375, 0.75, 0.625, 0.75, 0.9375]
"""
n = int(fabs(n)) if n is not None else 0
# Calculation
triangle = npArray([combination(n=n, r=i) for i in range(0, n + 1)])
triangle_sum = npSum(triangle)
triangle_weights = triangle / triangle_sum
inverse_weights = 1 - triangle_weights
weighted = kwargs.pop("weighted", False)
inverse = kwargs.pop("inverse", False)
if weighted and inverse:
return inverse_weights
if weighted:
return triangle_weights
if inverse:
return None
return triangle
def symmetric_triangle(n: int = None, **kwargs) -> list:
"""Symmetric Triangle with n >= 2
Returns a numpy array of the nth row of Symmetric Triangle.
n=4 => triangle: [1, 2, 2, 1]
=> weighted: [0.16666667 0.33333333 0.33333333 0.16666667]
"""
n = int(fabs(n)) if n is not None else 2
if n == 2:
triangle = [1, 1]
if n > 2:
if n % 2 == 0:
front = [i + 1 for i in range(0, floor(n/2))]
triangle = front + front[::-1]
else:
front = [i + 1 for i in range(0, floor(0.5 * (n + 1)))]
triangle = front.copy()
front.pop()
triangle += front[::-1]
if kwargs.pop("weighted", False):
triangle_sum = npSum(triangle)
triangle_weights = triangle / triangle_sum
return triangle_weights
return triangle
def weights(w):
def _dot(x):
return dot(w, x)
return _dot
def zero(x: [int, float]) -> [int, float]:
"""If the value is close to zero, then return zero.
Otherwise return itself."""
return 0 if abs(x) < sflt.epsilon else x
# TESTING
def df_error_analysis(dfA: DataFrame, dfB: DataFrame, **kwargs) -> DataFrame:
"""DataFrame Correlation Analysis helper"""
corr_method = kwargs.pop("corr_method", "pearson")
# Find their differences and correlation
diff = dfA - dfB
corr = dfA.corr(dfB, method=corr_method)
# For plotting
if kwargs.pop("plot", False):
diff.hist()
if diff[diff > 0].any():
diff.plot(kind="kde")
if kwargs.pop("triangular", False):
return corr.where(triu(ones(corr.shape)).astype(bool))
return corr
# 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."""
m = x.size
x_sum = x.sum()
y_sum = y.sum()
# 1st row, 2nd col value corr(x, y)
r = npCorrcoef(x, y)[0,1]
r_mixture = m * (x * y).sum() - x_sum * y_sum
b = r_mixture / (m * (x * x).sum() - x_sum * x_sum)
a = y.mean() - b * x.mean()
line = a + b * x
# seterr(divide="ignore", invalid="ignore")
return {
"a": a, "b": b, "r": r,
"t": r / npSqrt((1 - r * r) / (m - 2)),
"line": line
}
def _linear_regression_sklearn(x, y):
"""Simple Linear Regression in Scikit Learn for two 1d arrays for
environments with the sklearn package."""
from sklearn.linear_model import LinearRegression
regression = LinearRegression().fit(DataFrame(x), y=y)
r = regression.score(DataFrame(x), y=y)
a, b = regression.intercept_, regression.coef_[0]
return {
"a": a, "b": b, "r": r,
"t": r / npSqrt((1 - r * r) / (x.size - 2)),
"line": a + b * x
}
+184
View File
@@ -0,0 +1,184 @@
# -*- coding: utf-8 -*-
from pandas import DataFrame, Series
from ._core import get_offset, verify_series
from ._math import zero
def _above_below(
series_a: Series,
series_b: Series,
above: bool = True,
asint: bool = True,
offset: int = None,
**kwargs
):
series_a = verify_series(series_a)
series_b = verify_series(series_b)
offset = get_offset(offset)
series_a.apply(zero)
series_b.apply(zero)
# Calculate Result
if above:
current = series_a >= series_b
else:
current = series_a <= series_b
if asint:
current = current.astype(int)
# Offset
if offset != 0:
current = current.shift(offset)
# Name & Category
current.name = f"{series_a.name}_{'A' if above else 'B'}_{series_b.name}"
current.category = "utility"
return current
def above(
series_a: Series,
series_b: Series,
asint: bool = True,
offset: int = None,
**kwargs
):
return _above_below(series_a, series_b, above=True, asint=asint, offset=offset, **kwargs)
def above_value(
series_a: Series,
value: float,
asint: bool = True,
offset: int = None,
**kwargs
):
if not isinstance(value, (int, float, complex)):
print("[X] value is not a number")
return
series_b = Series(value, index=series_a.index, name=f"{value}".replace(".","_"))
return _above_below(series_a, series_b, above=True, asint=asint, offset=offset, **kwargs)
def below(
series_a: Series,
series_b: Series,
asint: bool =True,
offset: int =None
,**kwargs
):
return _above_below(series_a, series_b, above=False, asint=asint, offset=offset, **kwargs)
def below_value(
series_a: Series,
value: float,
asint: bool = True,
offset: int = None,
**kwargs
):
if not isinstance(value, (int, float, complex)):
print("[X] value is not a number")
return
series_b = Series(value, index=series_a.index, name=f"{value}".replace(".","_"))
return _above_below(series_a, series_b, above=False, asint=asint, offset=offset, **kwargs)
def cross_value(
series_a: Series,
value: float,
above: bool = True,
asint: bool = True,
offset: int = None,
**kwargs
):
series_b = Series(value, index=series_a.index, name=f"{value}".replace(".","_"))
return cross(series_a, series_b, above, asint, offset, **kwargs)
def cross(
series_a: Series,
series_b: Series,
above: bool = True,
asint: bool = True,
offset: int = None,
**kwargs
):
series_a = verify_series(series_a)
series_b = verify_series(series_b)
offset = get_offset(offset)
series_a.apply(zero)
series_b.apply(zero)
# Calculate Result
current = series_a > series_b # current is above
previous = series_a.shift(1) < series_b.shift(1) # previous is below
# above if both are true, below if both are false
cross = current & previous if above else ~current & ~previous
if asint:
cross = cross.astype(int)
# Offset
if offset != 0:
cross = cross.shift(offset)
# Name & Category
cross.name = f"{series_a.name}_{'XA' if above else 'XB'}_{series_b.name}"
cross.category = "utility"
return cross
def signals(indicator, xa, xb, cross_values, xserie, xserie_a, xserie_b, cross_series, offset) -> DataFrame:
df = DataFrame()
if xa is not None and isinstance(xa, (int, float)):
if cross_values:
crossed_above_start = cross_value(indicator, xa, above=True, offset=offset)
crossed_above_end = cross_value(indicator, xa, above=False, offset=offset)
df[crossed_above_start.name] = crossed_above_start
df[crossed_above_end.name] = crossed_above_end
else:
crossed_above = above_value(indicator, xa, offset=offset)
df[crossed_above.name] = crossed_above
if xb is not None and isinstance(xb, (int, float)):
if cross_values:
crossed_below_start = cross_value(indicator, xb, above=True, offset=offset)
crossed_below_end = cross_value(indicator, xb, above=False, offset=offset)
df[crossed_below_start.name] = crossed_below_start
df[crossed_below_end.name] = crossed_below_end
else:
crossed_below = below_value(indicator, xb, offset=offset)
df[crossed_below.name] = crossed_below
# xseries is the default value for both xserie_a and xserie_b
if xserie_a is None:
xserie_a = xserie
if xserie_b is None:
xserie_b = xserie
if xserie_a is not None and verify_series(xserie_a):
if cross_series:
cross_serie_above = cross(indicator, xserie_a, above=True, offset=offset)
else:
cross_serie_above = above(indicator, xserie_a, offset=offset)
df[cross_serie_above.name] = cross_serie_above
if xserie_b is not None and verify_series(xserie_b):
if cross_series:
cross_serie_below = cross(indicator, xserie_b, above=False, offset=offset)
else:
cross_serie_below = below(indicator, xserie_b, offset=offset)
df[cross_serie_below.name] = cross_serie_below
return df
+25
View File
@@ -0,0 +1,25 @@
# -*- coding: utf-8 -*-
from datetime import datetime
from time import perf_counter
from pandas_ta import EXCHANGE_TZ
def final_time(stime):
time_diff = perf_counter() - stime
return f"{time_diff * 1000:2.4f} ms ({time_diff:2.4f} s)"
def get_time(exchange: str = "NYSE", to_string:bool = False) -> (None, str):
tz = EXCHANGE_TZ["NYSE"] # Default is NYSE (Eastern Time Zone)
if isinstance(exchange, str):
exchange = exchange.upper()
tz = EXCHANGE_TZ[exchange]
day_of_year = datetime.utcnow().timetuple().tm_yday
today = datetime.utcnow()
s = f"Today: {today}, "
s += f"Day {day_of_year}/365 ({100 * round(day_of_year/365, 2)}%), "
s += f"{exchange} Time: {(today.timetuple().tm_hour + tz) % 12}:{today.timetuple().tm_min}:{today.timetuple().tm_sec}"
return s if to_string else print(s)
+1 -1
View File
@@ -12,7 +12,7 @@ def accbands(high, low, close, length=None, c=None, drift=None, mamode=None, off
high_low_range = non_zero_range(high, low)
length = int(length) if length and length > 0 else 20
c = float(c) if c and c > 0 else 4
min_periods = int(kwargs['min_periods']) if 'min_periods' in kwargs and kwargs['min_periods'] is not None else length
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
mamode = mamode.lower() if mamode else "sma"
drift = get_drift(drift)
offset = get_offset(offset)
+1 -2
View File
@@ -9,7 +9,6 @@ def bbands(close, length=None, std=None, mamode=None, offset=None, **kwargs):
# Validate arguments
close = verify_series(close)
length = int(length) if length and length > 0 else 5
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
std = float(std) if std and std > 0 else 2.
mamode = mamode.lower() if mamode else "sma"
offset = get_offset(offset)
@@ -52,7 +51,7 @@ def bbands(close, length=None, std=None, mamode=None, offset=None, **kwargs):
data = {lower.name: lower, mid.name: mid, upper.name: upper}
bbandsdf = DataFrame(data)
bbandsdf.name = f"BBANDS_{length}_{std}"
bbandsdf.category = "volatility"
bbandsdf.category = mid.category
return bbandsdf
+12 -12
View File
@@ -9,8 +9,8 @@ def donchian(high, low, lower_length=None, upper_length=None, offset=None, **kwa
low = verify_series(low)
lower_length = int(lower_length) if lower_length and lower_length > 0 else 20
upper_length = int(upper_length) if upper_length and upper_length > 0 else 20
lower_min_periods = int(kwargs['lower_min_periods']) if 'lower_min_periods' in kwargs and kwargs['lower_min_periods'] is not None else lower_length
upper_min_periods = int(kwargs['upper_min_periods']) if 'upper_min_periods' in kwargs and kwargs['upper_min_periods'] is not None else upper_length
lower_min_periods = int(kwargs["lower_min_periods"]) if "lower_min_periods" in kwargs and kwargs["lower_min_periods"] is not None else lower_length
upper_min_periods = int(kwargs["upper_min_periods"]) if "upper_min_periods" in kwargs and kwargs["upper_min_periods"] is not None else upper_length
offset = get_offset(offset)
# Calculate Result
@@ -19,14 +19,14 @@ def donchian(high, low, lower_length=None, upper_length=None, offset=None, **kwa
mid = 0.5 * (lower + upper)
# Handle fills
if 'fillna' in kwargs:
lower.fillna(kwargs['fillna'], inplace=True)
mid.fillna(kwargs['fillna'], inplace=True)
upper.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
lower.fillna(method=kwargs['fill_method'], inplace=True)
mid.fillna(method=kwargs['fill_method'], inplace=True)
upper.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
lower.fillna(kwargs["fillna"], inplace=True)
mid.fillna(kwargs["fillna"], inplace=True)
upper.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
lower.fillna(method=kwargs["fill_method"], inplace=True)
mid.fillna(method=kwargs["fill_method"], inplace=True)
upper.fillna(method=kwargs["fill_method"], inplace=True)
# Offset
if offset != 0:
@@ -38,13 +38,13 @@ def donchian(high, low, lower_length=None, upper_length=None, offset=None, **kwa
lower.name = f"DCL_{lower_length}_{upper_length}"
mid.name = f"DCM_{lower_length}_{upper_length}"
upper.name = f"DCU_{lower_length}_{upper_length}"
mid.category = upper.category = lower.category = 'volatility'
mid.category = upper.category = lower.category = "volatility"
# Prepare DataFrame to return
data = {lower.name: lower, mid.name: mid, upper.name: upper}
dcdf = DataFrame(data)
dcdf.name = f"DC_{lower_length}_{upper_length}"
dcdf.category = 'volatility'
dcdf.category = mid.category
return dcdf
+6 -6
View File
@@ -12,7 +12,7 @@ def massi(high, low, fast=None, slow=None, offset=None, **kwargs):
slow = int(slow) if slow and slow > 0 else 25
if slow < fast:
fast, slow = slow, fast
min_periods = int(kwargs['min_periods']) if 'min_periods' in kwargs and kwargs['min_periods'] is not None else fast
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else fast
offset = get_offset(offset)
# Calculate Result
@@ -27,14 +27,14 @@ def massi(high, low, fast=None, slow=None, offset=None, **kwargs):
massi = massi.shift(offset)
# Handle fills
if 'fillna' in kwargs:
massi.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
massi.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
massi.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
massi.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
massi.name = f"MASSI_{fast}_{slow}"
massi.category = 'volatility'
massi.category = "volatility"
return massi
+6 -6
View File
@@ -9,7 +9,7 @@ def natr(high, low, close, length=None, mamode=None, scalar=None, drift=None, of
low = verify_series(low)
close = verify_series(close)
length = int(length) if length and length > 0 else 14
mamode = mamode.lower() if mamode else 'ema'
mamode = mamode.lower() if mamode else "ema"
scalar = float(scalar) if scalar else 100
drift = get_drift(drift)
offset = get_offset(offset)
@@ -23,14 +23,14 @@ def natr(high, low, close, length=None, mamode=None, scalar=None, drift=None, of
natr = natr.shift(offset)
# Handle fills
if 'fillna' in kwargs:
natr.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
natr.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
natr.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
natr.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
natr.name = f"NATR_{length}"
natr.category = 'volatility'
natr.category = "volatility"
return natr
+1 -1
View File
@@ -22,7 +22,7 @@ def pdist(open_, high, low, close, drift=None, offset=None, **kwargs):
# Name & Category
pdist.name = "PDIST"
pdist.category = 'volatility'
pdist.category = "volatility"
return pdist
+5 -5
View File
@@ -23,14 +23,14 @@ def true_range(high, low, close, drift=None, offset=None, **kwargs):
true_range = true_range.shift(offset)
# Handle fills
if 'fillna' in kwargs:
true_range.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
true_range.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
true_range.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
true_range.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
true_range.name = f"TRUERANGE_{drift}"
true_range.category = 'volatility'
true_range.category = "volatility"
return true_range
+5 -5
View File
@@ -25,14 +25,14 @@ def adosc(high, low, close, volume, open_=None, fast=None, slow=None, offset=Non
adosc = adosc.shift(offset)
# Handle fills
if 'fillna' in kwargs:
adosc.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
adosc.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
adosc.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
adosc.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
adosc.name = f"ADOSC_{fast}_{slow}"
adosc.category = 'volume'
adosc.category = "volume"
return adosc
+6 -6
View File
@@ -10,7 +10,7 @@ def cmf(high, low, close, volume, open_=None, length=None, offset=None, **kwargs
volume = verify_series(volume)
high_low_range = non_zero_range(high, low)
length = int(length) if length and length > 0 else 20
min_periods = int(kwargs['min_periods']) if 'min_periods' in kwargs and kwargs['min_periods'] is not None else length
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
offset = get_offset(offset)
# Calculate Result
@@ -29,14 +29,14 @@ def cmf(high, low, close, volume, open_=None, length=None, offset=None, **kwargs
cmf = cmf.shift(offset)
# Handle fills
if 'fillna' in kwargs:
cmf.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
cmf.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
cmf.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
cmf.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
cmf.name = f"CMF_{length}"
cmf.category = 'volume'
cmf.category = "volume"
return cmf
+14 -14
View File
@@ -18,33 +18,33 @@ def mfi(high, low, close, volume, length=None, drift=None, offset=None, **kwargs
typical_price = hlc3(high=high, low=low, close=close)
raw_money_flow = typical_price * volume
tdf = DataFrame({'diff': 0, 'rmf': raw_money_flow, '+mf': 0, '-mf': 0})
tdf = DataFrame({"diff": 0, "rmf": raw_money_flow, "+mf": 0, "-mf": 0})
tdf.loc[(typical_price.diff(drift) > 0), 'diff'] = 1
tdf.loc[tdf['diff'] == 1, '+mf'] = raw_money_flow
tdf.loc[(typical_price.diff(drift) > 0), "diff"] = 1
tdf.loc[tdf["diff"] == 1, "+mf"] = raw_money_flow
tdf.loc[(typical_price.diff(drift) < 0), 'diff'] = -1
tdf.loc[tdf['diff'] == -1, '-mf'] = raw_money_flow
tdf.loc[(typical_price.diff(drift) < 0), "diff"] = -1
tdf.loc[tdf["diff"] == -1, "-mf"] = raw_money_flow
psum = tdf['+mf'].rolling(length).sum()
nsum = tdf['-mf'].rolling(length).sum()
tdf['mr'] = psum / nsum
psum = tdf["+mf"].rolling(length).sum()
nsum = tdf["-mf"].rolling(length).sum()
tdf["mr"] = psum / nsum
mfi = 100 * psum / (psum + nsum)
tdf['mfi'] = mfi
tdf["mfi"] = mfi
# Offset
if offset != 0:
mfi = mfi.shift(offset)
# Handle fills
if 'fillna' in kwargs:
mfi.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
mfi.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
mfi.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
mfi.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
mfi.name = f"MFI_{length}"
mfi.category = 'volume'
mfi.category = "volume"
return mfi
+6 -6
View File
@@ -8,7 +8,7 @@ def nvi(close, volume, length=None, initial=None, offset=None, **kwargs):
close = verify_series(close)
volume = verify_series(volume)
length = int(length) if length and length > 0 else 1
min_periods = int(kwargs['min_periods']) if 'min_periods' in kwargs and kwargs['min_periods'] is not None else length
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
initial = int(initial) if initial and initial > 0 else 1000
offset = get_offset(offset)
@@ -25,14 +25,14 @@ def nvi(close, volume, length=None, initial=None, offset=None, **kwargs):
nvi = nvi.shift(offset)
# Handle fills
if 'fillna' in kwargs:
nvi.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
nvi.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
nvi.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
nvi.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
nvi.name = f"NVI_{length}"
nvi.category = 'volume'
nvi.category = "volume"
return nvi
+5 -5
View File
@@ -17,14 +17,14 @@ def obv(close, volume, offset=None, **kwargs):
obv = obv.shift(offset)
# Handle fills
if 'fillna' in kwargs:
obv.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
obv.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
obv.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
obv.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
obv.name = f"OBV"
obv.category = 'volume'
obv.category = "volume"
return obv
+6 -6
View File
@@ -8,7 +8,7 @@ def pvi(close, volume, length=None, initial=None, offset=None, **kwargs):
close = verify_series(close)
volume = verify_series(volume)
length = int(length) if length and length > 0 else 1
min_periods = int(kwargs['min_periods']) if 'min_periods' in kwargs and kwargs['min_periods'] is not None else length
min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
initial = int(initial) if initial and initial > 0 else 1000
offset = get_offset(offset)
@@ -25,14 +25,14 @@ def pvi(close, volume, length=None, initial=None, offset=None, **kwargs):
pvi = pvi.shift(offset)
# Handle fills
if 'fillna' in kwargs:
pvi.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
pvi.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
pvi.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
pvi.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
pvi.name = f"PVI_{length}"
pvi.category = 'volume'
pvi.category = "volume"
return pvi
+6 -6
View File
@@ -7,7 +7,7 @@ def pvol(close, volume, offset=None, **kwargs):
close = verify_series(close)
volume = verify_series(volume)
offset = get_offset(offset)
signed = kwargs.pop('signed', False)
signed = kwargs.pop("signed", False)
# Calculate Result
if signed:
@@ -20,14 +20,14 @@ def pvol(close, volume, offset=None, **kwargs):
pvol = pvol.shift(offset)
# Handle fills
if 'fillna' in kwargs:
pvol.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
pvol.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
pvol.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
pvol.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
pvol.name = f"PVOL"
pvol.category = 'volume'
pvol.category = "volume"
return pvol
+5 -5
View File
@@ -19,14 +19,14 @@ def pvt(close, volume, drift=None, offset=None, **kwargs):
pvt = pvt.shift(offset)
# Handle fills
if 'fillna' in kwargs:
pvt.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
pvt.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
pvt.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
pvt.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
pvt.name = f"PVT"
pvt.category = 'volume'
pvt.category = "volume"
return pvt
+6 -6
View File
@@ -9,7 +9,7 @@ def vp(close, volume, width=None, **kwargs):
close = verify_series(close)
volume = verify_series(volume)
width = int(width) if width and width > 0 else 10
sort_close = kwargs.pop('sort_close', False)
sort_close = kwargs.pop("sort_close", False)
# Setup
signed_volume = signed_series(volume, initial=1)
@@ -47,14 +47,14 @@ def vp(close, volume, width=None, **kwargs):
vpdf[total_volume_col] = vpdf[pos_volume_col] + vpdf[neg_volume_col]
# Handle fills
if 'fillna' in kwargs:
vpdf.fillna(kwargs['fillna'], inplace=True)
if 'fill_method' in kwargs:
vpdf.fillna(method=kwargs['fill_method'], inplace=True)
if "fillna" in kwargs:
vpdf.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
vpdf.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
vpdf.name = f"VP_{width}"
vpdf.category = 'volume'
vpdf.category = "volume"
return vpdf
+22 -23
View File
@@ -1,13 +1,12 @@
# -*- coding: utf-8 -*-
from distutils.core import setup
from pandas_ta.core import version
long_description = "An easy to use Python 3 Pandas Extension with 115+ Technical Analysis Indicators. Can be called from a Pandas DataFrame or standalone like TA-Lib. Correlation tested with TA-Lib."
setup(
name ="pandas_ta",
packages =['pandas_ta', 'pandas_ta.candles', 'pandas_ta.momentum', 'pandas_ta.overlap', 'pandas_ta.performance', 'pandas_ta.statistics', 'pandas_ta.trend', 'pandas_ta.volatility', 'pandas_ta.volume'],
version =version,
packages =["pandas_ta", "pandas_ta.candles", "pandas_ta.momentum", "pandas_ta.overlap", "pandas_ta.performance", "pandas_ta.statistics", "pandas_ta.trend", "pandas_ta.utils", "pandas_ta.volatility", "pandas_ta.volume"],
version =".".join(("0", "2", "15b")),
description =long_description,
long_description =long_description,
author ="Kevin Johnson",
@@ -15,36 +14,36 @@ setup(
url ="https://github.com/twopirllc/pandas-ta",
maintainer ="Kevin Johnson",
maintainer_email ="appliedmathkj@gmail.com",
# install_requires=['pandas'],
# install_requires=["pandas"],
download_url ="https://github.com/twopirllc/pandas-ta.git",
keywords =['technical analysis', 'trading', 'python3', 'pandas'],
keywords =["technical analysis", "trading", "python3", "pandas"],
license ="The MIT License (MIT)",
classifiers =[
'Development Status :: 4 - Beta',
'Programming Language :: Python :: 3.6',
'Programming Language :: Python :: 3.7',
'Programming Language :: Python :: 3.8',
'Operating System :: OS Independent',
'License :: OSI Approved :: MIT License',
'Natural Language :: English',
'Intended Audience :: Developers',
'Intended Audience :: Financial and Insurance Industry',
'Intended Audience :: Science/Research',
'Topic :: Office/Business :: Financial',
'Topic :: Office/Business :: Financial :: Investment',
'Topic :: Scientific/Engineering',
'Topic :: Scientific/Engineering :: Information Analysis',
"Development Status :: 4 - Beta",
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Operating System :: OS Independent",
"License :: OSI Approved :: MIT License",
"Natural Language :: English",
"Intended Audience :: Developers",
"Intended Audience :: Financial and Insurance Industry",
"Intended Audience :: Science/Research",
"Topic :: Office/Business :: Financial",
"Topic :: Office/Business :: Financial :: Investment",
"Topic :: Scientific/Engineering",
"Topic :: Scientific/Engineering :: Information Analysis",
],
package_data={
'data': ['data/*.csv'],
"data": ["data/*.csv"],
},
install_requires =['pandas'],
install_requires =["pandas"],
# List additional groups of dependencies here (e.g. development dependencies).
# You can install these using the following syntax, for example:
# $ pip install -e .[dev,test]
extras_require = {
'dev': ['ta-lib', 'jupyterlab'],
'test': ['ta-lib'],
"dev": ["ta-lib", "jupyterlab", "sklearn", "statsmodels"],
"test": ["ta-lib"],
},
)
+5 -2
View File
@@ -1,5 +1,5 @@
import os
from pandas import read_csv
from pandas import DatetimeIndex, read_csv
VERBOSE = True
@@ -13,9 +13,12 @@ sample_data = read_csv(
f"data/SPY_D.csv",
index_col=0,
parse_dates=True,
infer_datetime_format=False,
infer_datetime_format=True,
keep_date_col=True
)
sample_data.set_index(DatetimeIndex(sample_data["date"]), inplace=True, drop=True)
sample_data.drop("date", axis=1, inplace=True)
def error_analysis(df, kind, msg, icon=INFO, newline=True):
if VERBOSE:
@@ -15,12 +15,8 @@ class TestCandleExtension(TestCase):
def tearDownClass(cls):
del cls.data
def setUp(self):
pass
def tearDown(self):
pass
def setUp(self): pass
def tearDown(self): pass
def test_cdl_doji_ext(self):
@@ -1,7 +1,7 @@
from .config import sample_data
from .context import pandas_ta
from unittest import TestCase
from unittest import skip, TestCase
from pandas import DataFrame
@@ -15,12 +15,8 @@ class TestMomentumExtension(TestCase):
def tearDownClass(cls):
del cls.data
def setUp(self):
pass
def tearDown(self):
pass
def setUp(self): pass
def tearDown(self): pass
def test_ao_ext(self):
@@ -53,6 +49,11 @@ class TestMomentumExtension(TestCase):
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "CCI_14_0.015")
def test_cfo_ext(self):
self.data.ta.cfo(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "CFO_9")
def test_cg_ext(self):
self.data.ta.cg(append=True)
self.assertIsInstance(self.data, DataFrame)
@@ -1,7 +1,7 @@
from .config import sample_data
from .context import pandas_ta
from unittest import TestCase
from unittest import skip, TestCase
from pandas import DataFrame
@@ -15,12 +15,8 @@ class TestOverlapExtension(TestCase):
def tearDownClass(cls):
del cls.data
def setUp(self):
pass
def tearDown(self):
pass
def setUp(self): pass
def tearDown(self): pass
def test_dema_ext(self):
@@ -17,7 +17,6 @@ class TestPerformaceExtension(TestCase):
del cls.data
del cls.islong
def setUp(self): pass
def tearDown(self): pass
@@ -15,12 +15,8 @@ class TestStatisticsExtension(TestCase):
def tearDownClass(cls):
del cls.data
def setUp(self):
pass
def tearDown(self):
pass
def setUp(self): pass
def tearDown(self): pass
def test_entropy_ext(self):
@@ -15,12 +15,8 @@ class TestTrendExtension(TestCase):
def tearDownClass(cls):
del cls.data
def setUp(self):
pass
def tearDown(self):
pass
def setUp(self): pass
def tearDown(self): pass
def test_adx_ext(self):
@@ -48,6 +44,15 @@ class TestTrendExtension(TestCase):
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "CKSPs_10_1_9")
def test_decay_ext(self):
self.data.ta.decay(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "LDECAY_5")
self.data.ta.decay(mode="exp", append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "EXPDECAY_5")
def test_decreasing_ext(self):
self.data.ta.decreasing(append=True)
self.assertIsInstance(self.data, DataFrame)
@@ -63,17 +68,12 @@ class TestTrendExtension(TestCase):
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "INC_1")
def test_linear_decay_ext(self):
self.data.ta.linear_decay(append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "LDECAY_5")
def test_long_run_ext(self):
# Nothing passed, return self
self.assertEqual(self.data.ta.long_run(append=True).shape, self.data.shape)
fast = self.data.ta.ema("close", 8)
slow = self.data.ta.ema("close", 21)
fast = self.data.ta.ema(8)
slow = self.data.ta.ema(21)
self.data.ta.long_run(fast, slow, append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "LR_2")
@@ -92,8 +92,8 @@ class TestTrendExtension(TestCase):
# Nothing passed, return self
self.assertEqual(self.data.ta.short_run(append=True).shape, self.data.shape)
fast = self.data.ta.ema("close", 8)
slow = self.data.ta.ema("close", 21)
fast = self.data.ta.ema(8)
slow = self.data.ta.ema(21)
self.data.ta.short_run(fast, slow, append=True)
self.assertIsInstance(self.data, DataFrame)
self.assertEqual(self.data.columns[-1], "SR_2")
@@ -15,7 +15,6 @@ class TestVolatilityExtension(TestCase):
def tearDownClass(cls):
del cls.data
def setUp(self): pass
def tearDown(self): pass
@@ -17,7 +17,6 @@ class TestVolumeExtension(TestCase):
del cls.data
del cls.open
def setUp(self): pass
def tearDown(self): pass
+5
View File
@@ -119,6 +119,11 @@ class TestMomentum(TestCase):
except Exception as ex:
error_analysis(result, CORRELATION, ex)
def test_cfo(self):
result = pandas_ta.cfo(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "CFO_9")
def test_cg(self):
result = pandas_ta.cg(self.close)
self.assertIsInstance(result, Series)
+9 -5
View File
@@ -99,6 +99,15 @@ class TestTrend(TestCase):
self.assertIsInstance(result, DataFrame)
self.assertEqual(result.name, "CKSP_10_1_9")
def test_decay(self):
result = pandas_ta.decay(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "LDECAY_5")
result = pandas_ta.decay(self.close, mode="exp")
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "EXPDECAY_5")
def test_decreasing(self):
result = pandas_ta.decreasing(self.close)
self.assertIsInstance(result, Series)
@@ -114,11 +123,6 @@ class TestTrend(TestCase):
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "INC_1")
def test_linear_decay(self):
result = pandas_ta.linear_decay(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "LDECAY_5")
def test_long_run(self):
result = pandas_ta.long_run(self.close, self.open)
self.assertIsInstance(result, Series)
+122 -129
View File
@@ -1,190 +1,183 @@
# Must run seperately from the rest of the tests
# in order to successfully run
from multiprocessing import cpu_count
from time import perf_counter
from .config import sample_data
from .context import pandas_ta
from unittest import skip, TestCase
from pandas import DataFrame
# Must run seperately from the rest of the tests
# in order to successfully run
from pandas_ta.utils import final_time
cores = 4
cumulative = False
speed_table = False
strategy_timed = False
timed = True
verbose = False
class TestStrategyMethods(TestCase):
@classmethod
def setUpClass(cls):
cls.data = sample_data
cls.data.ta.cores = cores
cls.speed_test = DataFrame()
@classmethod
def tearDownClass(cls):
cls.speed_test = cls.speed_test.T
cls.speed_test.index.name = "Test"
cls.speed_test.columns = ["Columns", "Seconds"]
if cumulative: cls.speed_test["Cum. Seconds"] = cls.speed_test["Seconds"].cumsum()
if speed_table: cls.speed_test.to_csv("tests/speed_test.csv")
if timed:
print(f"[i] Cores: {cls.data.ta.cores}")
print(f"[i] Total Datapoints: {cls.data.shape[0]}")
print(cls.speed_test)
del cls.data
def setUp(self): pass
def tearDown(self): pass
def setUp(self):
self.added_cols = 0
self.category = ""
self.init_cols = len(self.data.columns)
self.time_diff = 0
self.result = None
if verbose: print()
if timed: self.stime = perf_counter()
def tearDown(self):
if timed: self.time_diff = perf_counter() - self.stime
self.added_cols = len(self.data.columns) - self.init_cols
self.assertGreaterEqual(self.added_cols, 1)
self.result = self.data[self.data.columns[-self.added_cols:]]
self.assertIsInstance(self.result, DataFrame)
self.data.drop(columns=self.result.columns, axis=1, inplace=True)
self.speed_test[self.category] = [self.added_cols, self.time_diff]
# @skip
def test_all(self):
init_cols = len(self.data.columns)
self.data.ta.strategy(verbose=False)
added_cols = len(self.data.columns) - init_cols
self.assertGreaterEqual(added_cols, 1)
result = self.data[self.data.columns[-added_cols:]]
self.assertIsInstance(result, DataFrame)
self.data.drop(columns=result.columns, axis=1, inplace=True)
self.category = "All"
self.data.ta.strategy(verbose=verbose, timed=strategy_timed)
@skip
def test_all_strategy(self):
init_cols = len(self.data.columns)
self.data.ta.strategy(pandas_ta.AllStrategy, verbose=False)
added_cols = len(self.data.columns) - init_cols
self.assertGreaterEqual(added_cols, 1)
result = self.data[self.data.columns[-added_cols:]]
self.assertIsInstance(result, DataFrame)
self.data.drop(columns=result.columns, axis=1, inplace=True)
self.data.ta.strategy(pandas_ta.AllStrategy, verbose=verbose, timed=strategy_timed)
@skip
def test_all_name_strategy(self):
init_cols = len(self.data.columns)
self.data.ta.strategy("All", verbose=False)
added_cols = len(self.data.columns) - init_cols
self.assertGreaterEqual(added_cols, 1)
result = self.data[self.data.columns[-added_cols:]]
self.assertIsInstance(result, DataFrame)
self.data.drop(columns=result.columns, axis=1, inplace=True)
self.category = "All"
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
# @skip
def test_candles_category(self):
init_cols = len(self.data.columns)
self.data.ta.strategy("Candles", verbose=False)
added_cols = len(self.data.columns) - init_cols
self.assertGreaterEqual(added_cols, 1)
result = self.data[self.data.columns[-added_cols:]]
self.assertIsInstance(result, DataFrame)
self.data.drop(columns=result.columns, axis=1, inplace=True)
self.category = "Candles"
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
# @skip
def test_common(self):
init_cols = len(self.data.columns)
self.data.ta.strategy(pandas_ta.CommonStrategy, verbose=False)
added_cols = len(self.data.columns) - init_cols
self.assertGreaterEqual(added_cols, 1)
result = self.data[self.data.columns[-added_cols:]]
self.assertIsInstance(result, DataFrame)
self.data.drop(columns=result.columns, axis=1, inplace=True)
self.category = "Common"
self.data.ta.strategy(pandas_ta.CommonStrategy, verbose=verbose, timed=strategy_timed)
# @skip
def test_custom_a(self):
self.category = "Custom A"
momo_bands_sma_ta = [
{"kind":"sma", "length": 50}, # 1
{"kind":"sma", "length": 200}, # 1
{"kind":"bbands", "length": 20}, # 3
{"kind":"macd"}, # 3
{"kind":"rsi"}, # 1
{"kind":"log_return", "cumulative": True}, # 1
{"kind":"sma", "close": "CUMLOGRET_1", "length": 5, "suffix": "CUMLOGRET"}, # 1
{"kind": "rsi"}, # 1
{"kind": "macd"}, # 3
{"kind": "sma", "length": 50}, # 1
{"kind": "sma", "length": 200}, # 1
{"kind": "bbands", "length": 20}, # 3
{"kind": "log_return", "cumulative": True}, # 1
{"kind": "ema", "close": "CUMLOGRET_1", "length": 5, "suffix": "CLR"}
]
custom = pandas_ta.Strategy(
"Momo, Bands and SMAs and Cumulative Log Returns", # name
"Commons with Cumulative Log Return EMA Chain", # name
momo_bands_sma_ta, # ta
"MACD and RSI Momo with BBANDS and SMAs 50 & 200 and Cumulative Log Returns" # description
"Common indicators with specific lengths and a chained indicator" # description
)
self.data.ta.strategy(custom, verbose=verbose, timed=strategy_timed)
init_cols = len(self.data.columns)
self.data.ta.strategy(custom, verbose=False)
added_cols = len(self.data.columns) - init_cols
self.assertEqual(added_cols, 11)
result = self.data[self.data.columns[-added_cols:]]
self.assertIsInstance(result, DataFrame)
self.data.drop(columns=result.columns, axis=1, inplace=True)
@skip
# @skip
def test_custom_args_tuple(self):
self.category = "Custom B"
custom_args_ta = [
{"kind":"fisher", "params": (13, 7)},
{"kind":"macd", "params": (9, 19, 7)},
{"kind":"ema", "params": (5,)},
{"kind":"linreg", "close": "EMA_5", "length": 8, "prefix": "EMA_5"}
{"kind":"fisher", "params": (13, 7)},
]
custom = pandas_ta.Strategy(
"Custom Args Tuple", custom_args_ta,
"Allow for easy filling in indicator arguments without naming them"
"Allow for easy filling in indicator arguments by argument placement."
)
self.data.ta.strategy(custom, verbose=verbose, timed=strategy_timed)
init_cols = len(self.data.columns)
self.data.ta.strategy(custom, verbose=False)
added_cols = len(self.data.columns) - init_cols
def test_custom_col_names_tuple(self):
self.category = "Custom C"
result = self.data[self.data.columns[-added_cols:]]
self.assertIsInstance(result, DataFrame)
self.data.drop(columns=result.columns, axis=1, inplace=True)
custom_args_ta = [
{"kind":"bbands", "col_names": ("LB", "MB", "UB")}
]
custom = pandas_ta.Strategy(
"Custom Col Numbers Tuple", custom_args_ta,
"Allow for easy renaming of resultant columns"
)
self.data.ta.strategy(custom, verbose=verbose, timed=strategy_timed)
# @skip
def test_custom_col_numbers_tuple(self):
self.category = "Custom D"
custom_args_ta = [
{"kind":"macd", "col_numbers": (1,)}
]
custom = pandas_ta.Strategy(
"Custom Col Numbers Tuple", custom_args_ta,
"Allow for easy selection of resultant columns"
)
self.data.ta.strategy(custom, verbose=verbose, timed=strategy_timed)
# @skip
def test_momentum_category(self):
init_cols = len(self.data.columns)
self.data.ta.strategy("Momentum", verbose=False)
added_cols = len(self.data.columns) - init_cols
self.assertGreaterEqual(added_cols, 1)
result = self.data[self.data.columns[-added_cols:]]
self.assertIsInstance(result, DataFrame)
self.data.drop(columns=result.columns, axis=1, inplace=True)
self.category = "Momentum"
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
# @skip
def test_overlap_category(self):
init_cols = len(self.data.columns)
self.data.ta.strategy("Overlap", verbose=False)
added_cols = len(self.data.columns) - init_cols
self.assertGreaterEqual(added_cols, 1)
result = self.data[self.data.columns[-added_cols:]]
self.assertIsInstance(result, DataFrame)
self.data.drop(columns=result.columns, axis=1, inplace=True)
self.category = "Overlap"
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
# @skip
def test_performance_category(self):
init_cols = len(self.data.columns)
self.data.ta.strategy("Performance", verbose=False)
added_cols = len(self.data.columns) - init_cols
self.assertGreaterEqual(added_cols, 1)
result = self.data[self.data.columns[-added_cols:]]
self.assertIsInstance(result, DataFrame)
self.data.drop(columns=result.columns, axis=1, inplace=True)
self.category = "Performance"
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
# @skip
def test_statistics_category(self):
init_cols = len(self.data.columns)
self.data.ta.strategy("Statistics", verbose=False)
added_cols = len(self.data.columns) - init_cols
self.assertGreaterEqual(added_cols, 1)
result = self.data[self.data.columns[-added_cols:]]
self.assertIsInstance(result, DataFrame)
self.data.drop(columns=result.columns, axis=1, inplace=True)
self.category = "Statistics"
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
# @skip
def test_trend_category(self):
init_cols = len(self.data.columns)
self.data.ta.strategy("Trend", verbose=False)
added_cols = len(self.data.columns) - init_cols
self.assertGreaterEqual(added_cols, 1)
result = self.data[self.data.columns[-added_cols:]]
self.assertIsInstance(result, DataFrame)
self.data.drop(columns=result.columns, axis=1, inplace=True)
self.category = "Trend"
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
# @skip
def test_volatility_category(self):
init_cols = len(self.data.columns)
self.data.ta.strategy("Volatility", verbose=False)
added_cols = len(self.data.columns) - init_cols
self.assertGreaterEqual(added_cols, 1)
result = self.data[self.data.columns[-added_cols:]]
self.assertIsInstance(result, DataFrame)
self.data.drop(columns=result.columns, axis=1, inplace=True)
self.category = "Volatility"
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
# @skip
def test_volume_category(self):
init_cols = len(self.data.columns)
self.data.ta.strategy("Volume", verbose=False)
added_cols = len(self.data.columns) - init_cols
self.assertGreaterEqual(added_cols, 1)
result = self.data[self.data.columns[-added_cols:]]
self.assertIsInstance(result, DataFrame)
self.data.drop(columns=result.columns, axis=1, inplace=True)
self.category = "Volume"
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
+74 -59
View File
@@ -1,7 +1,7 @@
from .config import sample_data
from .context import pandas_ta
from unittest import TestCase
from unittest import skip, TestCase
from unittest.mock import patch
import numpy as np
@@ -9,11 +9,11 @@ import numpy.testing as npt
from pandas import DataFrame, Series
data = {
'zero': [0, 0],
'a': [0, 1],
'b': [1, 0],
'c': [1, 1],
'crossed': [0, 1],
"zero": [0, 0],
"a": [0, 1],
"b": [1, 0],
"c": [1, 1],
"crossed": [0, 1],
}
class TestUtilities(TestCase):
@@ -35,80 +35,81 @@ class TestUtilities(TestCase):
def test__add_prefix_suffix(self):
result = self.data.ta.hl2(append=False, prefix="pre")
self.assertEqual(result.name, 'pre_HL2')
self.assertEqual(result.name, "pre_HL2")
result = self.data.ta.hl2(append=False, suffix="suf")
self.assertEqual(result.name, 'HL2_suf')
self.assertEqual(result.name, "HL2_suf")
result = self.data.ta.hl2(append=False, prefix="pre", suffix="suf")
self.assertEqual(result.name, 'pre_HL2_suf')
self.assertEqual(result.name, "pre_HL2_suf")
result = self.data.ta.hl2(append=False, prefix=1, suffix=2)
self.assertEqual(result.name, '1_HL2_2')
self.assertEqual(result.name, "1_HL2_2")
result = self.data.ta.macd(append=False, prefix="pre", suffix="suf")
for col in result.columns:
self.assertTrue(col.startswith('pre_') and col.endswith('_suf'))
self.assertTrue(col.startswith("pre_") and col.endswith("_suf"))
@skip
def test__above_below(self):
result = self.utils._above_below(self.crosseddf['a'], self.crosseddf['zero'], above=True)
result = self.utils._above_below(self.crosseddf["a"], self.crosseddf["zero"], above=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'a_A_zero')
npt.assert_array_equal(result, self.crosseddf['c'])
self.assertEqual(result.name, "a_A_zero")
npt.assert_array_equal(result, self.crosseddf["c"])
result = self.utils._above_below(self.crosseddf['a'], self.crosseddf['zero'], above=False)
result = self.utils._above_below(self.crosseddf["a"], self.crosseddf["zero"], above=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'a_B_zero')
npt.assert_array_equal(result, self.crosseddf['b'])
self.assertEqual(result.name, "a_B_zero")
npt.assert_array_equal(result, self.crosseddf["b"])
result = self.utils._above_below(self.crosseddf['c'], self.crosseddf['zero'], above=True)
result = self.utils._above_below(self.crosseddf["c"], self.crosseddf["zero"], above=True)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'c_A_zero')
npt.assert_array_equal(result, self.crosseddf['c'])
self.assertEqual(result.name, "c_A_zero")
npt.assert_array_equal(result, self.crosseddf["c"])
result = self.utils._above_below(self.crosseddf['c'], self.crosseddf['zero'], above=False)
result = self.utils._above_below(self.crosseddf["c"], self.crosseddf["zero"], above=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'c_B_zero')
npt.assert_array_equal(result, self.crosseddf['zero'])
self.assertEqual(result.name, "c_B_zero")
npt.assert_array_equal(result, self.crosseddf["zero"])
def test_above(self):
result = self.utils.above(self.crosseddf['a'], self.crosseddf['zero'])
result = self.utils.above(self.crosseddf["a"], self.crosseddf["zero"])
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'a_A_zero')
npt.assert_array_equal(result, self.crosseddf['c'])
self.assertEqual(result.name, "a_A_zero")
npt.assert_array_equal(result, self.crosseddf["c"])
result = self.utils.above(self.crosseddf['zero'], self.crosseddf['a'])
result = self.utils.above(self.crosseddf["zero"], self.crosseddf["a"])
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'zero_A_a')
npt.assert_array_equal(result, self.crosseddf['b'])
self.assertEqual(result.name, "zero_A_a")
npt.assert_array_equal(result, self.crosseddf["b"])
def test_above_value(self):
result = self.utils.above_value(self.crosseddf['a'], 0)
result = self.utils.above_value(self.crosseddf["a"], 0)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'a_A_0')
npt.assert_array_equal(result, self.crosseddf['c'])
self.assertEqual(result.name, "a_A_0")
npt.assert_array_equal(result, self.crosseddf["c"])
result = self.utils.above_value(self.crosseddf['a'], self.crosseddf['zero'])
result = self.utils.above_value(self.crosseddf["a"], self.crosseddf["zero"])
self.assertIsNone(result)
def test_below(self):
result = self.utils.below(self.crosseddf['zero'], self.crosseddf['a'])
result = self.utils.below(self.crosseddf["zero"], self.crosseddf["a"])
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'zero_B_a')
npt.assert_array_equal(result, self.crosseddf['c'])
self.assertEqual(result.name, "zero_B_a")
npt.assert_array_equal(result, self.crosseddf["c"])
result = self.utils.below(self.crosseddf['zero'], self.crosseddf['a'])
result = self.utils.below(self.crosseddf["zero"], self.crosseddf["a"])
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'zero_B_a')
npt.assert_array_equal(result, self.crosseddf['c'])
self.assertEqual(result.name, "zero_B_a")
npt.assert_array_equal(result, self.crosseddf["c"])
def test_below_value(self):
result = self.utils.below_value(self.crosseddf['a'], 0)
result = self.utils.below_value(self.crosseddf["a"], 0)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, 'a_B_0')
npt.assert_array_equal(result, self.crosseddf['b'])
self.assertEqual(result.name, "a_B_0")
npt.assert_array_equal(result, self.crosseddf["b"])
result = self.utils.below_value(self.crosseddf['a'], self.crosseddf['zero'])
result = self.utils.below_value(self.crosseddf["a"], self.crosseddf["zero"])
self.assertIsNone(result)
def test_combination(self):
@@ -121,18 +122,18 @@ class TestUtilities(TestCase):
self.assertEqual(self.utils.combination(n=10, r=4, repetition=True), 715)
def test_cross_above(self):
result = self.utils.cross(self.crosseddf['a'], self.crosseddf['b'])
result = self.utils.cross(self.crosseddf["a"], self.crosseddf["b"])
self.assertIsInstance(result, Series)
npt.assert_array_equal(result, self.crosseddf['crossed'])
npt.assert_array_equal(result, self.crosseddf["crossed"])
result = self.utils.cross(self.crosseddf['a'], self.crosseddf['b'], above=True)
result = self.utils.cross(self.crosseddf["a"], self.crosseddf["b"], above=True)
self.assertIsInstance(result, Series)
npt.assert_array_equal(result, self.crosseddf['crossed'])
npt.assert_array_equal(result, self.crosseddf["crossed"])
def test_cross_below(self):
result = self.utils.cross(self.crosseddf['b'], self.crosseddf['a'], above=False)
result = self.utils.cross(self.crosseddf["b"], self.crosseddf["a"], above=False)
self.assertIsInstance(result, Series)
npt.assert_array_equal(result, self.crosseddf['crossed'])
npt.assert_array_equal(result, self.crosseddf["crossed"])
def test_fibonacci(self):
self.assertIs(type(self.utils.fibonacci(zero=True, weighted=False)), np.ndarray)
@@ -154,6 +155,25 @@ class TestUtilities(TestCase):
npt.assert_allclose(self.utils.fibonacci(n=5, zero=True, weighted=True), np.array([0, 1/12, 1/12, 1/6, 1/4, 5/12]))
npt.assert_allclose(self.utils.fibonacci(n=5, zero=False, weighted=True), np.array([1/12, 1/12, 1/6, 1/4, 5/12]))
def test_get_time(self):
result = self.utils.get_time()
result = self.utils.get_time("NZSX")
result = self.utils.get_time("SSE", to_string=True)
self.assertEqual(self.utils.EXCHANGE_TZ["NYSE"], -4)
self.assertIsInstance(result, str)
def test_linear_regression(self):
x = Series([1, 2, 3, 4, 5])
y = Series([1.8, 2.1, 2.7, 3.2, 4])
result = self.utils.linear_regression(x, y)
self.assertIsInstance(result, dict)
self.assertIsInstance(result["a"], float)
self.assertIsInstance(result["b"], float)
self.assertIsInstance(result["r"], float)
self.assertIsInstance(result["t"], float)
self.assertIsInstance(result["line"], Series)
def test_pascals_triangle(self):
self.assertIsNone(self.utils.pascals_triangle(inverse=True), None)
@@ -198,22 +218,17 @@ class TestUtilities(TestCase):
self.assertNotEqual(self.utils.zero(1), 0)
def test_get_drift(self):
for s in [0, None, '', [], {}]:
for s in [0, None, "", [], {}]:
self.assertIsInstance(self.utils.get_drift(s), int)
self.assertEqual(self.utils.get_drift(0), 1)
self.assertEqual(self.utils.get_drift(1.1), 1)
self.assertEqual(self.utils.get_drift(-1.1), -1)
self.assertEqual(self.utils.get_drift(1.999999999999999), 1)
self.assertEqual(self.utils.get_drift(1.9999999999999999), 2)
self.assertEqual(self.utils.get_drift(-10), -10)
self.assertEqual(self.utils.get_drift(-1.1), 1)
def test_get_offset(self):
for s in [0, None, '', [], {}]:
for s in [0, None, "", [], {}]:
self.assertIsInstance(self.utils.get_offset(s), int)
self.assertEqual(self.utils.get_offset(0), 0)
self.assertEqual(self.utils.get_offset(1.1), 1)
self.assertEqual(self.utils.get_offset(-1.1), -1)
self.assertEqual(self.utils.get_offset(1.999999999999999), 1)
self.assertEqual(self.utils.get_offset(1.9999999999999999), 2)
self.assertEqual(self.utils.get_offset(-1.1), 0)
self.assertEqual(self.utils.get_offset(1), 1)