diff --git a/examples/ni.py b/examples/ni.py new file mode 100644 index 0000000..41aedc0 --- /dev/null +++ b/examples/ni.py @@ -0,0 +1,79 @@ +# -*- coding: utf-8 -*- +from pandas_ta.overlap import sma +from pandas_ta.utils import get_offset, verify_series + +# - Standard definition of your custom indicator function (including docs)- + +def ni(close, length=None, centered=False, offset=None, **kwargs): + """ + Example indicator ni + """ + # Validate Arguments + length = int(length) if length and length > 0 else 20 + close = verify_series(close, length) + offset = get_offset(offset) + + if close is None: return + + # Calculate Result + t = int(0.5 * length) + 1 + ma = sma(close, length) + + ni = close - ma.shift(t) + if centered: + ni = (close.shift(t) - ma).shift(-t) + + # Offset + if offset != 0: + ni = ni.shift(offset) + + # Handle fills + if "fillna" in kwargs: + ni.fillna(kwargs["fillna"], inplace=True) + if "fill_method" in kwargs: + ni.fillna(method=kwargs["fill_method"], inplace=True) + + # Name and Categorize it + ni.name = f"ni_{length}" + ni.category = "trend" + + return ni + +ni.__doc__ = \ +"""Example indicator (NI) + +Is an indicator provided solely as an example + +Sources: + https://github.com/twopirllc/pandas-ta/issues/264 + +Calculation: + Default Inputs: + length=20, centered=False + SMA = Simple Moving Average + t = int(0.5 * length) + 1 + + ni = close.shift(t) - SMA(close, length) + if centered: + ni = ni.shift(-t) + +Args: + close (pd.Series): Series of 'close's + length (int): It's period. Default: 20 + centered (bool): Shift the ni back by int(0.5 * length) + 1. Default: False + offset (int): How many periods to offset the result. Default: 0 + +Kwargs: + fillna (value, optional): pd.DataFrame.fillna(value) + fill_method (value, optional): Type of fill method + +Returns: + pd.Series: New feature generated. +""" + +# - Define a matching class method -------------------------------------------- + +def ni_method(self, length=None, offset=None, **kwargs): + close = self._get_column(kwargs.pop("close", "close")) + result = ni(close=close, length=length, offset=offset, **kwargs) + return self._post_process(result, **kwargs) \ No newline at end of file diff --git a/pandas_ta/custom.py b/pandas_ta/custom.py new file mode 100644 index 0000000..1854249 --- /dev/null +++ b/pandas_ta/custom.py @@ -0,0 +1,225 @@ +# -*- coding: utf-8 -*- +import importlib +import os +import sys +from os.path import abspath, join, exists, basename, splitext +from glob import glob +import types +import pandas_ta +import pandas as pd +from pandas_ta import AnalysisIndicators + + +def bind(function_name, function, method): + """ + Helper function to bind the function and class method defined in a custom + indicator module to the active pandas_ta instance. + + Args: + function_name (str): The name of the indicator within pandas_ta + function (fcn): The indicator function + method (fcn): The class method corresponding to the passed function + """ + setattr(pandas_ta, function_name, function) + setattr(AnalysisIndicators, function_name, method) + + +def create_dir(path, create_categories=True, verbose=True): + """ + Helper function to setup a suitable folder structure for working with + custom indicators. You only need to call this once whenever you want to + setup a new custom indicators folder. + + Args: + path (str): Full path to where you want your indicator tree + create_categories (bool): If True create category sub-folders + verbose (bool): If True print verbose output of results + """ + + # ensure that the passed directory exists / is readable + if not exists(path): + os.makedirs(path) + if verbose: + print(f"[i] Created main directory '{path}'.") + + # list the contents of the directory + # dirs = glob(abspath(join(path, '*'))) + + # optionally add any missing category subdirectories + if create_categories: + for sd in [*pandas_ta.Category]: + d = abspath(join(path, sd)) + if not exists(d): + os.makedirs(d) + if verbose: + dirname = basename(d) + print(f"[i] Created an empty sub-directory '{dirname}'.") + + +def get_module_functions(module): + """ + Helper function to get the functions of an imported module as a dictionary. + + Args: + module: python module + + Returns: + dict: module functions mapping + { + "func1_name": func1, + "func2_name": func2,... + } + """ + module_functions = {} + + for name, item in vars(module).items(): + if isinstance(item, types.FunctionType): + module_functions[name] = item + + return module_functions + + +def import_dir(path, verbose=True): + # ensure that the passed directory exists / is readable + if not exists(path): + print(f"[X] Unable to read the directory '{path}'.") + return + + # list the contents of the directory + dirs = glob(abspath(join(path, '*'))) + + # traverse full directory, importing all modules found there + for d in dirs: + dirname = basename(d) + + # only look in directories which are valid pandas_ta categories + if dirname not in [*pandas_ta.Category]: + if verbose: + print(f"[i] Skipping the sub-directory '{dirname}' since it's not a valid pandas_ta category.") + continue + + # for each module found in that category (directory)... + for module in glob(abspath(join(path, dirname, '*.py'))): + module_name = splitext(basename(module))[0] + + # ensure that the supplied path is included in our python path + if d not in sys.path: + sys.path.append(d) + + # (re)load the indicator module + module_functions = load_indicator_module(module_name) + + # figure out which of the modules functions to bind to pandas_ta + fcn_callable = module_functions.get(module_name, None) + fcn_method_callable = module_functions.get(module_name + "_method", None) + + if fcn_callable == None: + print(f"[X] Unable to find a function named '{module_name}' in the module '{module_name}.py'.") + continue + if fcn_method_callable == None: + missing_method = module_name + "_method" + print(f"[X] Unable to find a method function named '{missing_method}' in the module '{module_name}.py'.") + continue + + # add it to the correct category if it's not there yet + if module_name not in pandas_ta.Category[dirname]: + pandas_ta.Category[dirname].append(module_name) + + bind(module_name, fcn_callable, fcn_method_callable) + if verbose: + print(f"[i] Successfully imported the custom indicator '{module}' into category '{dirname}'.") + + +import_dir.__doc__ = \ +""" +Import a directory of custom indicators into pandas_ta + +Args: + path (str): Full path to your indicator tree + verbose (bool): If True verbose output of results + +This method allows you to experiment and develop your own technical analysis +indicators in a separate local directory of your choice but use them seamlessly +together with the existing pandas_ta functions just like if they were part of +pandas_ta. + +If you at some late point would like to push them into the pandas_ta library +you can do so very easily by following the step by step instruction here +https://github.com/twopirllc/pandas-ta/issues/264. + +A brief example of usage: + +1. Loading the 'ta' module: +>>> import pandas as pd +>>> import pandas_ta as ta + +2. Create an empty directory on your machine where you want to work with your +indicators. Invoke pandas_ta.custom.import_dir once to pre-populate it with +sub-folders for all available indicator categories, e.g.: + +>>> import os +>>> from os.path import abspath, join, expanduser +>>> from pandas_ta.custom import create_dir, import_dir +>>> my_dir = abspath(join(expanduser("~"), "my_indicators")) +>>> create_dir(my_dir) + +3. You can now create your own custom indicator e.g. by copying existing +ones from pandas_ta core module and modifying them. + +IMPORTANT: Each custom indicator should have a unique name and have both +a) a function named exactly as the module, e.g. 'ni' if the module is ni.py +b) a matching method used by AnalysisIndicators named as the module but + ending with '_method'. E.g. 'ni_method' + +In essence these modules should look exactly like the standard indicators +available in categories under the pandas_ta-folder. The only difference will +be an addition of a matching class method. + +For an example of the correct structure, look at the example ni.py in the +examples folder. + +The ni.py indicator is a trend indicator so therefore we drop it into the +sub-folder named trend. Thus we have a folder structure like this: + +~/my_indicators/ +│ +├── candles/ +. +. +└── trend/ +. └── ni.py +. +└── volume/ + +4. We can now dynamically load all our custom indicators located in our +designated indicators directory like this: + +>>> import_dir(my_dir) + +If your custom indicator(s) loaded succesfully then it should behave exactly +like all other native indicators in pandas_ta, including help functions. +""" + + +def load_indicator_module(module_name): + """ + Helper function to (re)load an indicator module. + + Returns: + dict: module functions mapping + { + "func1_name": func1, + "func2_name": func2,... + } + + """ + # load module + try: + module = importlib.import_module(module_name) + except Exception as ex: + print(f"[X] An error occurred when attempting to load module {module_name}: {ex}") + sys.exit(1) + + # reload to refresh previously loaded module + module = importlib.reload(module) + return get_module_functions(module) \ No newline at end of file