From 879437bc423cb98525c6d3b82269f0fb50b4922d Mon Sep 17 00:00:00 2001 From: David Brazda Date: Fri, 18 Oct 2024 14:14:49 +0200 Subject: [PATCH] update --- .gitignore | 1 + setup.py | 2 +- tests/cuvwap.ipynb | 76 +++++++++++++++++++++++++ ttools/indicators/CUVWAP.py | 68 +++++++++++++++++++++++ ttools/indicators/__init__.py | 0 ttools/vbtindicators.py | 101 ++++++++-------------------------- 6 files changed, 170 insertions(+), 78 deletions(-) create mode 100644 tests/cuvwap.ipynb create mode 100644 ttools/indicators/CUVWAP.py create mode 100644 ttools/indicators/__init__.py diff --git a/.gitignore b/.gitignore index 82f9275..52f4bde 100644 --- a/.gitignore +++ b/.gitignore @@ -25,6 +25,7 @@ share/python-wheels/ .installed.cfg *.egg MANIFEST +.vscode # PyInstaller # Usually these files are written by a python script from a template diff --git a/setup.py b/setup.py index 7331e43..cf1a940 100644 --- a/setup.py +++ b/setup.py @@ -2,7 +2,7 @@ from setuptools import setup, find_packages setup( name='ttools', - version='0.2.8', + version='0.2.9', packages=find_packages(), install_requires=[ 'vectorbtpro', diff --git a/tests/cuvwap.ipynb b/tests/cuvwap.ipynb new file mode 100644 index 0000000..ebbf018 --- /dev/null +++ b/tests/cuvwap.ipynb @@ -0,0 +1,76 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": 4, + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "The autoreload extension is already loaded. To reload it, use:\n", + " %reload_ext autoreload\n" + ] + }, + { + "ename": "ModuleNotFoundError", + "evalue": "No module named 'indicators'", + "output_type": "error", + "traceback": [ + "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m", + "\u001b[0;31mModuleNotFoundError\u001b[0m Traceback (most recent call last)", + "Cell \u001b[0;32mIn[4], line 5\u001b[0m\n\u001b[1;32m 3\u001b[0m \u001b[38;5;28;01mimport\u001b[39;00m \u001b[38;5;21;01mvectorbtpro\u001b[39;00m \u001b[38;5;28;01mas\u001b[39;00m \u001b[38;5;21;01mvbt\u001b[39;00m\n\u001b[1;32m 4\u001b[0m \u001b[38;5;28;01mfrom\u001b[39;00m \u001b[38;5;21;01mttools\u001b[39;00m\u001b[38;5;21;01m.\u001b[39;00m\u001b[38;5;21;01mvbtindicators\u001b[39;00m \u001b[38;5;28;01mimport\u001b[39;00m register_custom_inds\n\u001b[0;32m----> 5\u001b[0m \u001b[43mregister_custom_inds\u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;28;43;01mNone\u001b[39;49;00m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[38;5;124;43moverride\u001b[39;49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[43m)\u001b[49m\n\u001b[1;32m 6\u001b[0m \u001b[38;5;66;03m#chopiness = vbt.indicator(\"technical:CHOPINESS\").run(s12_data.open, s12_data.high, s12_data.low, s12_data.close, s12_data.volume, window = 100)\u001b[39;00m\n\u001b[1;32m 7\u001b[0m \u001b[38;5;66;03m#vwap_cum_roll = vbt.indicator(\"technical:ROLLING_VWAP\").run(s12_data.open, s12_data.high, s12_data.low, s12_data.close, s12_data.volume, window = 100, min_periods = 5)\u001b[39;00m\n\u001b[1;32m 8\u001b[0m \u001b[38;5;66;03m#vwap_cum_d = vbt.indicator(\"ttools:CUVWAP\").run(s12_data.high, s12_data.low, s12_data.close, s12_data.volume, anchor=\"D\", drag=50)\u001b[39;00m\n\u001b[1;32m 9\u001b[0m \u001b[38;5;66;03m#vwap_lin_angle = vbt.indicator(\"talib:LINEARREG_ANGLE\").run(vwap_cum_d.vwap, timeperiod=2)\u001b[39;00m\n\u001b[1;32m 11\u001b[0m vbt\u001b[38;5;241m.\u001b[39mIF\u001b[38;5;241m.\u001b[39mlist_indicators(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mttools\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n", + "File \u001b[0;32m~/Documents/Development/python/ttools/ttools/vbtindicators.py:28\u001b[0m, in \u001b[0;36mregister_custom_inds\u001b[0;34m(indicator_name, if_exists)\u001b[0m\n\u001b[1;32m 26\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m file_name\u001b[38;5;241m.\u001b[39mendswith(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124m.py\u001b[39m\u001b[38;5;124m\"\u001b[39m) \u001b[38;5;129;01mand\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m file_name\u001b[38;5;241m.\u001b[39mstartswith(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124m_\u001b[39m\u001b[38;5;124m\"\u001b[39m):\n\u001b[1;32m 27\u001b[0m module_name \u001b[38;5;241m=\u001b[39m file_name[:\u001b[38;5;241m-\u001b[39m\u001b[38;5;241m3\u001b[39m]\n\u001b[0;32m---> 28\u001b[0m module \u001b[38;5;241m=\u001b[39m \u001b[43mimportlib\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mimport_module\u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;124;43mf\u001b[39;49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[38;5;124;43mindicators.\u001b[39;49m\u001b[38;5;132;43;01m{\u001b[39;49;00m\u001b[43mfile_name\u001b[49m\u001b[43m[\u001b[49m\u001b[43m:\u001b[49m\u001b[38;5;241;43m-\u001b[39;49m\u001b[38;5;241;43m3\u001b[39;49m\u001b[43m]\u001b[49m\u001b[38;5;132;43;01m}\u001b[39;49;00m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[43m)\u001b[49m\n\u001b[1;32m 29\u001b[0m \u001b[38;5;28;01mfor\u001b[39;00m var_name, var_value \u001b[38;5;129;01min\u001b[39;00m \u001b[38;5;28mvars\u001b[39m(module)\u001b[38;5;241m.\u001b[39mitems():\n\u001b[1;32m 30\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m var_name\u001b[38;5;241m.\u001b[39mstartswith(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mIND_\u001b[39m\u001b[38;5;124m\"\u001b[39m) \u001b[38;5;129;01mand\u001b[39;00m \u001b[38;5;28misinstance\u001b[39m(var_value, vbt\u001b[38;5;241m.\u001b[39mIndicatorFactory):\n", + "File \u001b[0;32m/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/importlib/__init__.py:126\u001b[0m, in \u001b[0;36mimport_module\u001b[0;34m(name, package)\u001b[0m\n\u001b[1;32m 124\u001b[0m \u001b[38;5;28;01mbreak\u001b[39;00m\n\u001b[1;32m 125\u001b[0m level \u001b[38;5;241m+\u001b[39m\u001b[38;5;241m=\u001b[39m \u001b[38;5;241m1\u001b[39m\n\u001b[0;32m--> 126\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[43m_bootstrap\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_gcd_import\u001b[49m\u001b[43m(\u001b[49m\u001b[43mname\u001b[49m\u001b[43m[\u001b[49m\u001b[43mlevel\u001b[49m\u001b[43m:\u001b[49m\u001b[43m]\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mpackage\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mlevel\u001b[49m\u001b[43m)\u001b[49m\n", + "File \u001b[0;32m:1050\u001b[0m, in \u001b[0;36m_gcd_import\u001b[0;34m(name, package, level)\u001b[0m\n", + "File \u001b[0;32m:1027\u001b[0m, in \u001b[0;36m_find_and_load\u001b[0;34m(name, import_)\u001b[0m\n", + "File \u001b[0;32m:992\u001b[0m, in \u001b[0;36m_find_and_load_unlocked\u001b[0;34m(name, import_)\u001b[0m\n", + "File \u001b[0;32m:241\u001b[0m, in \u001b[0;36m_call_with_frames_removed\u001b[0;34m(f, *args, **kwds)\u001b[0m\n", + "File \u001b[0;32m:1050\u001b[0m, in \u001b[0;36m_gcd_import\u001b[0;34m(name, package, level)\u001b[0m\n", + "File \u001b[0;32m:1027\u001b[0m, in \u001b[0;36m_find_and_load\u001b[0;34m(name, import_)\u001b[0m\n", + "File \u001b[0;32m:1004\u001b[0m, in \u001b[0;36m_find_and_load_unlocked\u001b[0;34m(name, import_)\u001b[0m\n", + "\u001b[0;31mModuleNotFoundError\u001b[0m: No module named 'indicators'" + ] + } + ], + "source": [ + "%load_ext autoreload\n", + "%autoreload 2\n", + "import vectorbtpro as vbt\n", + "from ttools.vbtindicators import register_custom_inds\n", + "from ttools.indicators import CUVWAP\n", + "\n", + "\n", + "register_custom_inds(None, \"override\")\n", + "#chopiness = vbt.indicator(\"technical:CHOPINESS\").run(s12_data.open, s12_data.high, s12_data.low, s12_data.close, s12_data.volume, window = 100)\n", + "#vwap_cum_roll = vbt.indicator(\"technical:ROLLING_VWAP\").run(s12_data.open, s12_data.high, s12_data.low, s12_data.close, s12_data.volume, window = 100, min_periods = 5)\n", + "#vwap_cum_d = vbt.indicator(\"ttools:CUVWAP\").run(s12_data.high, s12_data.low, s12_data.close, s12_data.volume, anchor=\"D\", drag=50)\n", + "#vwap_lin_angle = vbt.indicator(\"talib:LINEARREG_ANGLE\").run(vwap_cum_d.vwap, timeperiod=2)\n", + "\n", + "vbt.IF.list_indicators(\"ttools\")\n" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": ".venv", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.10.11" + } + }, + "nbformat": 4, + "nbformat_minor": 2 +} diff --git a/ttools/indicators/CUVWAP.py b/ttools/indicators/CUVWAP.py new file mode 100644 index 0000000..8a9038d --- /dev/null +++ b/ttools/indicators/CUVWAP.py @@ -0,0 +1,68 @@ +import numpy as np +from numba import jit +import vectorbtpro as vbt +from vectorbtpro import _typing as tp +from vectorbtpro.base.wrapping import ArrayWrapper +from vectorbtpro.utils.template import RepFunc + +""" +Contains custom indicators for vectorbtpro. + +import and run register_custom_inds() to register all custom indicators. + +They are available under `vbt.IF.list_indicators("ttols")` +""" + +def substitute_anchor(wrapper: ArrayWrapper, anchor: tp.Optional[tp.FrequencyLike]) -> tp.Array1d: + """Substitute reset frequency by group lens. It is array of number of elements of each group.""" + if anchor is None: + return np.array([wrapper.shape[0]]) + return wrapper.get_index_grouper(anchor).get_group_lens() + +@jit(nopython=True) +def vwap_cum(high, low, close, volume, group_lens, drag): + #anchor based grouping - prepare group indexes + group_end_idxs = np.cumsum(group_lens) + group_start_idxs = group_end_idxs - group_lens + + #prepare output + out = np.full(volume.shape, np.nan, dtype=np.float_) + + hlcc4 = (high + low + close + close) / 4 + + #iterate over groups + for group in range(len(group_lens)): + from_i = group_start_idxs[group] + from_i = max(0, from_i - drag) + to_i = group_end_idxs[group] + nom_cumsum = 0 + denum_cumsum = 0 + #for each group do this (it is just np.cumsum(hlcc4 * volume) / np.sum(volume) iteratively) + for i in range(from_i, to_i): + nom_cumsum += volume[i] * hlcc4[i] + denum_cumsum += volume[i] + if denum_cumsum == 0: + out[i] = np.nan + else: + out[i] = nom_cumsum / denum_cumsum + return out + +""" +cumulative anchored vwap indicator on HLCC4 price, anchor = "D", "h", or "min" ... +drag = 0 - overlap with previous group. takes into account last N elements from previous group +when calculating (simulating v2realbot logic) +""" +IND_CUVWAP = vbt.IF( + class_name='CUVWAP', + module_name='ttools', + input_names=['high', 'low', 'close', 'volume'], + param_names=['anchor', "drag"], + output_names=['vwap'] +).with_apply_func(vwap_cum, + takes_1d=True, + param_settings=dict( + anchor=dict(template=RepFunc(substitute_anchor)), + ), + anchor="D", + drag=0 + ) \ No newline at end of file diff --git a/ttools/indicators/__init__.py b/ttools/indicators/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ttools/vbtindicators.py b/ttools/vbtindicators.py index 650026d..55dd204 100644 --- a/ttools/vbtindicators.py +++ b/ttools/vbtindicators.py @@ -1,91 +1,38 @@ -import numpy as np -from numba import jit + import vectorbtpro as vbt -from vectorbtpro import _typing as tp -from vectorbtpro.base.wrapping import ArrayWrapper -from vectorbtpro.utils.template import RepFunc - -""" -Contains custom indicators for vectorbtpro. - -import and run register_custom_inds() to register all custom indicators. - -They are available under `vbt.IF.list_indicators("ttols")` -""" - -def substitute_anchor(wrapper: ArrayWrapper, anchor: tp.Optional[tp.FrequencyLike]) -> tp.Array1d: - """Substitute reset frequency by group lens. It is array of number of elements of each group.""" - if anchor is None: - return np.array([wrapper.shape[0]]) - return wrapper.get_index_grouper(anchor).get_group_lens() - -@jit(nopython=True) -def vwap_cum(high, low, close, volume, group_lens, drag): - #anchor based grouping - prepare group indexes - group_end_idxs = np.cumsum(group_lens) - group_start_idxs = group_end_idxs - group_lens - - #prepare output - out = np.full(volume.shape, np.nan, dtype=np.float_) - - hlcc4 = (high + low + close + close) / 4 - - #iterate over groups - for group in range(len(group_lens)): - from_i = group_start_idxs[group] - from_i = max(0, from_i - drag) - to_i = group_end_idxs[group] - nom_cumsum = 0 - denum_cumsum = 0 - #for each group do this (it is just np.cumsum(hlcc4 * volume) / np.sum(volume) iteratively) - for i in range(from_i, to_i): - nom_cumsum += volume[i] * hlcc4[i] - denum_cumsum += volume[i] - if denum_cumsum == 0: - out[i] = np.nan - else: - out[i] = nom_cumsum / denum_cumsum - return out - -""" -cumulative anchored vwap indicator on HLCC4 price, anchor = "D", "h", or "min" ... -drag = 0 - overlap with previous group. takes into account last N elements from previous group -when calculating (simulating v2realbot logic) -""" -IND_CUVWAP = vbt.IF( - class_name='CUVWAP', - module_name='ttools', - input_names=['high', 'low', 'close', 'volume'], - param_names=['anchor', "drag"], - output_names=['vwap'] -).with_apply_func(vwap_cum, - takes_1d=True, - param_settings=dict( - anchor=dict(template=RepFunc(substitute_anchor)), - ), - anchor="D", - drag=0 - ) +import importlib +import os def register_custom_inds(indicator_name: str = None, if_exists: str ="skip"): """Register a custom indicator or all custom indicators. + Each indicator is as NAME.py file in the `ttools.indicators` directory. It should + contain variable IND_{NAME} with the indicator factory object. + If `indicator_name` is provided, only the indicator with that name is registered. - Otherwise, all indicators are registered - they are the ones starting with "IND_" . + Otherwise, all indicators found in the directory are registered - in each file + variable with name starting with "IND_" . Argument `if_exists` can be "raise", "skip", or "override". """ + indicators_dir = os.path.join(os.path.dirname(__file__), "indicators") if indicator_name is not None: - var_name = f"IND_{indicator_name}" - var_value = globals().get(var_name) - if var_value is not None and isinstance(var_value, vbt.IndicatorFactory): - vbt.IF.register_custom_indicator(var_value, location="ttools", if_exists=if_exists) - else: - raise ValueError(f"Indicator '{indicator_name}' not found") - else: - for var_name, var_value in globals().items(): - if var_name.startswith("IND_"): + module_name = indicator_name + file_path = os.path.join(indicators_dir, f"{module_name}.py") + if os.path.exists(file_path): + module = importlib.import_module(f"ttools.indicators.{module_name}") + var_name = f"IND_{indicator_name}" + var_value = getattr(module, var_name, None) + if var_value is not None: vbt.IF.register_custom_indicator(var_value, location="ttools", if_exists=if_exists) + else: + for file_name in os.listdir(indicators_dir): + if file_name.endswith(".py") and not file_name.startswith("_"): + module_name = file_name[:-3] + module = importlib.import_module(f"ttools.indicators.{module_name}") + for var_name, var_value in vars(module).items(): + if var_name.startswith("IND_"): + vbt.IF.register_custom_indicator(var_value, location="ttools", if_exists=if_exists) def deregister_custom_inds(indicator_name: str = None): """Deregister a custom indicator or all custom indicators.