From 4265a39076c7ad09b1db230498af77ce950c2935 Mon Sep 17 00:00:00 2001 From: Merlin Roe Date: Thu, 11 Jan 2024 00:00:17 +0000 Subject: [PATCH] Renamed Pattern to Amoprh --- CHANGELOG.md | 3 +- hexital/core/hexital.py | 46 ++++----- hexital/exceptions.py | 2 +- hexital/indicators/__init__.py | 2 +- hexital/indicators/amorph.py | 82 ++++++++++++++++ hexital/indicators/pattern.py | 81 ---------------- tests/core/test_hexital.py | 16 ++-- tests/core/test_indicator.py | 26 ++--- ...est_pattern.py => test_amoprh_patterns.py} | 4 +- tests/indicators/test_amorph.py | 91 ++++++++++++++++++ tests/indicators/test_pattern_indicator.py | 94 ------------------- 11 files changed, 215 insertions(+), 232 deletions(-) create mode 100644 hexital/indicators/amorph.py delete mode 100644 hexital/indicators/pattern.py rename tests/indicators/{test_pattern.py => test_amoprh_patterns.py} (96%) create mode 100644 tests/indicators/test_amorph.py delete mode 100644 tests/indicators/test_pattern_indicator.py diff --git a/CHANGELOG.md b/CHANGELOG.md index df61f20..8a03e76 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,7 +11,8 @@ The project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html - Updated collapse candle 'fill' to show essentially doji candle rather than copy prev - Added Patterns: - Hammer Candle -- Updated Hexital/Pattern to accept patterns, movements and custom methods +- Renamed Pattern to Amorph and updated to only require either 'indicator' or 'analysis' +- Updated Hexital/Amorph to accept patterns, movements and custom methods - Major Fix: Re-wrote collapse_candles_timeframe to correctly handle candles,gaps and appending - Fixed Doji pattern - Fixed Supertrend Indicator diff --git a/hexital/core/hexital.py b/hexital/core/hexital.py index 7d83481..efeca95 100644 --- a/hexital/core/hexital.py +++ b/hexital/core/hexital.py @@ -1,11 +1,11 @@ import importlib from copy import deepcopy from datetime import timedelta -from typing import Dict, List, Optional +from typing import Callable, Dict, List, Optional from hexital.core.candle import Candle from hexital.core.indicator import Indicator -from hexital.exceptions import InvalidIndicator, InvalidPattern +from hexital.exceptions import InvalidIndicator, InvalidAnalysis from hexital.lib.candle_extension import collapse_candles_timeframe, reading_by_index DEFAULT = "default" @@ -57,7 +57,7 @@ def _validate_indicators( f"Indicator type invalid 'indicator' must be a dict or Indicator type: {indicator}" ) - pattern_class = getattr(indicator_module, "Pattern") + amorph_class = getattr(indicator_module, "Amorph") if indicator.get("indicator"): indicator_name = indicator.pop("indicator") @@ -70,36 +70,28 @@ def _validate_indicators( new_indicator = indicator_class(**indicator) valid_indicators[new_indicator.name] = new_indicator - elif indicator.get("pattern") and isinstance(indicator.get("pattern"), str): - pattern_name = indicator.pop("pattern") + elif indicator.get("analysis") and isinstance(indicator.get("analysis"), str): + name = indicator.pop("analysis") pattern_module = importlib.import_module("hexital.analysis.patterns") - - try: - pattern_func = getattr(pattern_module, pattern_name) - except AttributeError: - raise InvalidIndicator(f"pattern {pattern_name} does not exist") - - new_indicator = pattern_class(pattern=pattern_func, **indicator) - valid_indicators[new_indicator.name] = new_indicator - - elif indicator.get("movement") and isinstance(indicator.get("movement"), str): - movement_name = indicator.pop("movement") movement_module = importlib.import_module("hexital.analysis.movement") - try: - movement_func = getattr(movement_module, movement_name) - except AttributeError: - raise InvalidIndicator(f"movement {movement_name} does not exist") - new_indicator = pattern_class(pattern=movement_func, **indicator) + pattern_func = getattr(pattern_module, name, None) + analysis_func = getattr(movement_module, name, pattern_func) + if not analysis_func: + raise InvalidAnalysis( + f"analysis {name} does not exist in patterns or movements" + ) + + new_indicator = amorph_class(analysis=analysis_func, **indicator) valid_indicators[new_indicator.name] = new_indicator - elif indicator.get("method") and callable(indicator.get("method")): - method_name = indicator.pop("method") - new_indicator = pattern_class(pattern=method_name, **indicator) + elif indicator.get("analysis") and callable(indicator.get("analysis")): + method_name = indicator.pop("analysis") + new_indicator = amorph_class(analysis=method_name, **indicator) valid_indicators[new_indicator.name] = new_indicator else: - raise InvalidIndicator( - f"Dict Indicator missing 'indicator', 'pattern', 'movement' or 'pattern' name, not: {indicator}" + raise InvalidAnalysis( + f"Dict Indicator missing 'indicator' or 'analysis' name, not: {indicator}" ) for indicator in valid_indicators.values(): @@ -185,7 +177,7 @@ def reading_as_list(self, name: Optional[str] = None) -> List[float | dict | Non return self._indicators[name].as_list return [] - def add_indicator(self, indicator: Indicator | List[Indicator]): + def add_indicator(self, indicator: Indicator | List[Indicator] | Dict[str, str | Callable]): """Add's a new indicator to `Hexital` strategy. This accept either `Indicator` datatypes or dict string versions to be packed. `add_indicator(SMA(period=10))` or `add_indicator({"indicator": "SMA", "period": 10})` diff --git a/hexital/exceptions.py b/hexital/exceptions.py index dbe13c0..8a293c6 100644 --- a/hexital/exceptions.py +++ b/hexital/exceptions.py @@ -3,7 +3,7 @@ def __init__(self, message): super().__init__(message) -class InvalidPattern(Exception): +class InvalidAnalysis(Exception): def __init__(self, message): super().__init__(message) diff --git a/hexital/indicators/__init__.py b/hexital/indicators/__init__.py index 19bd8ce..4b6af59 100644 --- a/hexital/indicators/__init__.py +++ b/hexital/indicators/__init__.py @@ -1,3 +1,4 @@ +from .amorph import Amorph from .adx import ADX from .atr import ATR from .ema import EMA @@ -6,7 +7,6 @@ from .macd import MACD from .managed import Managed from .obv import OBV -from .pattern import Pattern from .rma import RMA from .roc import ROC from .rsi import RSI diff --git a/hexital/indicators/amorph.py b/hexital/indicators/amorph.py new file mode 100644 index 0000000..6b1181e --- /dev/null +++ b/hexital/indicators/amorph.py @@ -0,0 +1,82 @@ +import importlib +import inspect +from copy import deepcopy +from typing import Callable, Optional + +from hexital.core import Indicator +from hexital.exceptions import InvalidAnalysis + + +class Amorph(Indicator): + """Amorph + + Flexible Skeleton Indicator that will use a method + to generate readings on every Candle like indicators. + The given Method is expected to have 'candles' and 'index' as named arguments + + Arguments: + All of Indicator Arguments and All of the given Amorph Arguments as keyword arguments + or use args as a dict of keyword arguments for called analysis + + """ + + _analysis_method: Callable + _analysis_kwargs: dict + + def __init__(self, analysis: str | Callable, args: Optional[dict] = None, **kwargs): + if isinstance(analysis, str): + pattern_module = importlib.import_module("hexital.analysis.patterns") + movement_module = importlib.import_module("hexital.analysis.movement") + pattern_method = getattr(pattern_module, analysis, None) + analysis_func = getattr(movement_module, analysis, pattern_method) + + if analysis_func is not None: + self._analysis_method = analysis_func + + elif callable(analysis): + self._analysis_method = analysis + + if not hasattr(self, "_analysis_method"): + raise InvalidAnalysis(f"The given analysis [{analysis}] is invalid") + + self._analysis_kwargs, kwargs = self._seperate_indicator_attributes(kwargs) + + if isinstance(args, dict): + self._analysis_kwargs.update(args) + + super().__init__(**kwargs) + + @property + def settings(self) -> dict: + """Returns a dict format of how this indicator can be generated""" + settings = self.__dict__ + output = {"analysis": self._analysis_method.__name__} + + for name, value in settings.items(): + if name == "candles": + continue + if name == "timeframe_fill" and self.timeframe is None: + continue + if not name.startswith("_") and value: + output[name] = deepcopy(value) + + return output + + @staticmethod + def _seperate_indicator_attributes(kwargs: dict) -> tuple[dict, dict]: + indicator_attr = inspect.getmembers(Indicator)[1][1].keys() + analysis_args = {} + for argum in list(kwargs.keys()): + if argum not in indicator_attr: + analysis_args[argum] = kwargs.pop(argum) + + return analysis_args, kwargs + + def _generate_name(self) -> str: + name = self._analysis_method.__name__ + if self._analysis_kwargs.get("length"): + name += f"_{self._analysis_kwargs['length']}" + return name + + def _calculate_reading(self, index: int) -> float | dict | None: + return self._analysis_method(candles=self.candles, index=index, **self._analysis_kwargs) diff --git a/hexital/indicators/pattern.py b/hexital/indicators/pattern.py deleted file mode 100644 index a00dd95..0000000 --- a/hexital/indicators/pattern.py +++ /dev/null @@ -1,81 +0,0 @@ -import importlib -import inspect -from copy import deepcopy -from typing import Callable, Optional - -from hexital.core import Indicator -from hexital.exceptions import InvalidPattern - - -class Pattern(Indicator): - """Pattern - - Flexible Skeleton Indicator that will use candle Analysis patterns - to generate pattern results on every Candle like indicators. - - Arguments: - All of Indicator Arguments and All of the given Pattern Arguments as keyword arguments - or use args as a dict of keyword arguments for called pattern - - """ - - _pattern_method: Callable - _pattern_kwargs: dict - - def __init__(self, pattern: str | Callable, args: Optional[dict] = None, **kwargs): - if isinstance(pattern, str): - pattern_module = importlib.import_module("hexital.analysis.patterns") - movement_module = importlib.import_module("hexital.analysis.movement") - pattern_method = getattr(pattern_module, pattern, None) - movement_module = getattr(movement_module, pattern, None) - if pattern_method is not None: - self._pattern_method = pattern_method - elif movement_module is not None: - self._pattern_method = movement_module - elif callable(pattern): - self._pattern_method = pattern - - if not hasattr(self, "_pattern_method"): - raise InvalidPattern(f"The given pattern [{pattern}] is invalid") - - self._pattern_kwargs, kwargs = self._seperate_indicator_attributes(kwargs) - - if isinstance(args, dict): - self._pattern_kwargs.update(args) - - super().__init__(**kwargs) - - @property - def settings(self) -> dict: - """Returns a dict format of how this indicator can be generated""" - settings = self.__dict__ - output = {"pattern": self._pattern_method.__name__} - - for name, value in settings.items(): - if name == "candles": - continue - if name == "timeframe_fill" and self.timeframe is None: - continue - if not name.startswith("_") and value: - output[name] = deepcopy(value) - - return output - - @staticmethod - def _seperate_indicator_attributes(kwargs: dict) -> tuple[dict, dict]: - indicator_attr = inspect.getmembers(Indicator)[1][1].keys() - pattern_args = {} - for argum in list(kwargs.keys()): - if argum not in indicator_attr: - pattern_args[argum] = kwargs.pop(argum) - - return pattern_args, kwargs - - def _generate_name(self) -> str: - name = self._pattern_method.__name__ - if self._pattern_kwargs.get("length"): - name += f"_{self._pattern_kwargs['length']}" - return name - - def _calculate_reading(self, index: int) -> float | dict | None: - return self._pattern_method(candles=self.candles, index=index, **self._pattern_kwargs) diff --git a/tests/core/test_hexital.py b/tests/core/test_hexital.py index 327915b..48e66bd 100644 --- a/tests/core/test_hexital.py +++ b/tests/core/test_hexital.py @@ -3,8 +3,8 @@ import pytest from hexital.core import Candle, Hexital, Indicator -from hexital.exceptions import InvalidIndicator -from hexital.indicators import EMA, OBV, SMA +from hexital.exceptions import InvalidAnalysis, InvalidIndicator +from hexital.indicators import EMA, SMA def fake_pattern(candles: List[Candle], index=-1): @@ -54,7 +54,7 @@ def test_hextial_dict_invalid(self, candles): @pytest.mark.usefixtures("candles") def test_hextial_dict_invalid_missing(self, candles): - with pytest.raises(InvalidIndicator): + with pytest.raises(InvalidAnalysis): Hexital("Test Stratergy", candles, [{"period": 10}]) @pytest.mark.usefixtures("candles", "expected_ema", "expected_sma") @@ -68,20 +68,20 @@ def test_hextial_dict_append(self, candles, expected_ema, expected_sma): ) @pytest.mark.usefixtures("candles") - def test_hextial_dict_pattern(self, candles): - strat = Hexital("Test Stratergy", candles, [{"pattern": "doji"}]) + def test_hextial_dict_analysis_pattern(self, candles): + strat = Hexital("Test Stratergy", candles, [{"analysis": "doji"}]) strat.calculate() assert strat.reading("doji") is not None @pytest.mark.usefixtures("candles") def test_hextial_dict_movement(self, candles): - strat = Hexital("Test Stratergy", candles, [{"movement": "positive"}]) + strat = Hexital("Test Stratergy", candles, [{"analysis": "positive"}]) strat.calculate() assert strat.reading("positive") is not None @pytest.mark.usefixtures("candles") - def test_hextial_dict_pattern_custom(self, candles): - strat = Hexital("Test Stratergy", candles, [{"method": fake_pattern}]) + def test_hextial_dict_analysis_custom(self, candles): + strat = Hexital("Test Stratergy", candles, [{"analysis": fake_pattern}]) strat.calculate() assert strat.reading("fake_pattern") is not None diff --git a/tests/core/test_indicator.py b/tests/core/test_indicator.py index 0597aa4..01edddc 100644 --- a/tests/core/test_indicator.py +++ b/tests/core/test_indicator.py @@ -4,7 +4,7 @@ import pytest from hexital.core import Candle, Indicator -from hexital.indicators.pattern import Pattern +from hexital.indicators.amorph import Amorph @dataclass(kw_only=True) @@ -14,7 +14,7 @@ class FakeIndicator(Indicator): fullname_override: Optional[str] = None name_suffix: Optional[str] = None round_value: int = 4 - + period: int = 10 input_value: str = "close" @@ -49,9 +49,7 @@ def test_name_fulloverride(minimal_candles: List[Candle]): @pytest.mark.usefixtures("minimal_candles") def test_name_fulloverride_and_suffix(minimal_candles: List[Candle]): - test = FakeIndicator( - candles=minimal_candles, fullname_override="FUCK", name_suffix="YOU" - ) + test = FakeIndicator(candles=minimal_candles, fullname_override="FUCK", name_suffix="YOU") test.calculate() assert test.name == "FUCK_YOU" @@ -72,9 +70,7 @@ def test_name_timeframe(minimal_candles: List[Candle]): @pytest.mark.usefixtures("minimal_candles") def test_name_timeframe_override(minimal_candles: List[Candle]): - test = FakeIndicator( - candles=minimal_candles, fullname_override="FUCK", timeframe="t5" - ) + test = FakeIndicator(candles=minimal_candles, fullname_override="FUCK", timeframe="t5") test.calculate() assert test.name == "FUCK" @@ -140,10 +136,10 @@ def test_settings(minimal_candles: List[Candle]): } -def test_settings_pattern(): - test = Pattern(pattern="doji") +def test_settings_analysis(): + test = Amorph(analysis="doji") assert test.settings == { - "pattern": "doji", + "analysis": "doji", "round_value": 4, } @@ -183,9 +179,7 @@ def test_append_dict(): "volume": 19661, } ) - assert test.candles == [ - Candle(17213, 2395, 7813, 3615, 19661, indicators={"Fake_10": 100.0}) - ] + assert test.candles == [Candle(17213, 2395, 7813, 3615, 19661, indicators={"Fake_10": 100.0})] def test_append_dict_list(): @@ -209,9 +203,7 @@ def test_append_list(): test.append([17213, 2395, 7813, 3615, 19661]) - assert test.candles == [ - Candle(17213, 2395, 7813, 3615, 19661, indicators={"Fake_10": 100.0}) - ] + assert test.candles == [Candle(17213, 2395, 7813, 3615, 19661, indicators={"Fake_10": 100.0})] def test_append_list_list(): diff --git a/tests/indicators/test_pattern.py b/tests/indicators/test_amoprh_patterns.py similarity index 96% rename from tests/indicators/test_pattern.py rename to tests/indicators/test_amoprh_patterns.py index e82fb3f..29b25f3 100644 --- a/tests/indicators/test_pattern.py +++ b/tests/indicators/test_amoprh_patterns.py @@ -110,12 +110,12 @@ def show_results(self, result: list, expected: list, verbose: bool): @pytest.mark.usefixtures("candles", "expected_doji") def test_doji(self, candles, expected_doji): - test = indicators.Pattern(pattern=patterns.doji, candles=candles) + test = indicators.Amorph(analysis=patterns.doji, candles=candles) test.calculate() assert self.verify(test.as_list, expected_doji) @pytest.mark.usefixtures("candles", "expected_hammer") def test_hammer(self, candles, expected_hammer): - test = indicators.Pattern(pattern=patterns.hammer, candles=candles) + test = indicators.Amorph(analysis=patterns.hammer, candles=candles) test.calculate() assert self.verify(test.as_list, expected_hammer) diff --git a/tests/indicators/test_amorph.py b/tests/indicators/test_amorph.py new file mode 100644 index 0000000..fd8dee0 --- /dev/null +++ b/tests/indicators/test_amorph.py @@ -0,0 +1,91 @@ +import pytest +from hexital import patterns, movement +from hexital.indicators import Amorph +from hexital.core import Candle +from hexital.exceptions import InvalidAnalysis +from typing import List + + +def fake_pattern(candles: List[Candle], index=-1): + return 1 + + +@pytest.mark.usefixtures("candles") +def test_invalid_amorph(candles): + with pytest.raises(InvalidAnalysis): + Amorph(analysis="FUCK", candles=candles) + + +@pytest.mark.usefixtures("candles") +def test_string_amorph(candles): + test = Amorph(analysis="doji", candles=candles) + test.calculate() + assert test.reading() is not None + + +@pytest.mark.usefixtures("candles") +def test_method_amorph(candles): + test = Amorph(analysis=patterns.doji, candles=candles) + test.calculate() + assert test.reading() is not None + + +@pytest.mark.usefixtures("candles") +def test_amorph_multi_arguments(candles): + test = Amorph(analysis=patterns.doji, candles=candles, length=20) + test.calculate() + assert test.name == "doji_20" + + +@pytest.mark.usefixtures("candles") +def test_amorph_dict_arguments(candles): + test = Amorph(analysis=patterns.doji, candles=candles, args={"length": 20}) + test.calculate() + assert test.name == "doji_20" + + +@pytest.mark.usefixtures("candles") +def test_amorph_merged_aguments(candles): + test = Amorph( + analysis=patterns.doji, + candles=candles, + length=20, + fullname_override="MERGED_ARGS", + ) + test.calculate() + assert test.name == "MERGED_ARGS" + + +@pytest.mark.usefixtures("candles") +def test_string_movement(candles): + test = Amorph(analysis="positive", candles=candles) + test.calculate() + assert test.reading("positive") is not None + + +@pytest.mark.usefixtures("candles") +def test_movement_amorph(candles): + test = Amorph(analysis=movement.positive, candles=candles) + test.calculate() + assert test.reading("positive") is not None + + +@pytest.mark.usefixtures("candles") +def test_movement_amorph_args(candles): + test = Amorph(analysis=movement.positive, candles=candles, fullname_override="boobies") + test.calculate() + assert test.reading("boobies") is not None + + +@pytest.mark.usefixtures("candles") +def test_movement_amorph_kawgs(candles): + test = Amorph(analysis=movement.above, candles=candles, indicator="open", indicator_two="low") + test.calculate() + assert test.reading("above") is not None + + +@pytest.mark.usefixtures("candles") +def test_amorph_custom(candles): + test = Amorph(analysis=fake_pattern, candles=candles) + test.calculate() + assert test.reading("fake_pattern") is not None diff --git a/tests/indicators/test_pattern_indicator.py b/tests/indicators/test_pattern_indicator.py deleted file mode 100644 index 0e9a137..0000000 --- a/tests/indicators/test_pattern_indicator.py +++ /dev/null @@ -1,94 +0,0 @@ -import pytest -from hexital import indicators, patterns, movement -from hexital.core import Candle -from hexital.exceptions import InvalidPattern -from typing import List - - -def fake_pattern(candles: List[Candle], index=-1): - return 1 - - -@pytest.mark.usefixtures("candles") -def test_invalid_pattern(candles): - with pytest.raises(InvalidPattern): - indicators.Pattern(pattern="FUCK", candles=candles) - - -@pytest.mark.usefixtures("candles") -def test_string_pattern(candles): - test = indicators.Pattern(pattern="doji", candles=candles) - test.calculate() - assert test.reading() is not None - - -@pytest.mark.usefixtures("candles") -def test_method_pattern(candles): - test = indicators.Pattern(pattern=patterns.doji, candles=candles) - test.calculate() - assert test.reading() is not None - - -@pytest.mark.usefixtures("candles") -def test_pattern_multi_arguments(candles): - test = indicators.Pattern(pattern=patterns.doji, candles=candles, length=20) - test.calculate() - assert test.name == "doji_20" - - -@pytest.mark.usefixtures("candles") -def test_pattern_dict_arguments(candles): - test = indicators.Pattern(pattern=patterns.doji, candles=candles, args={"length": 20}) - test.calculate() - assert test.name == "doji_20" - - -@pytest.mark.usefixtures("candles") -def test_pattern_merged_aguments(candles): - test = indicators.Pattern( - pattern=patterns.doji, - candles=candles, - length=20, - fullname_override="MERGED_ARGS", - ) - test.calculate() - assert test.name == "MERGED_ARGS" - - -@pytest.mark.usefixtures("candles") -def test_string_movement(candles): - test = indicators.Pattern(pattern="positive", candles=candles) - test.calculate() - assert test.reading("positive") is not None - - -@pytest.mark.usefixtures("candles") -def test_movement_pattern(candles): - test = indicators.Pattern(pattern=movement.positive, candles=candles) - test.calculate() - assert test.reading("positive") is not None - - -@pytest.mark.usefixtures("candles") -def test_movement_pattern_args(candles): - test = indicators.Pattern( - pattern=movement.positive, candles=candles, fullname_override="boobies" - ) - test.calculate() - assert test.reading("boobies") is not None - - -@pytest.mark.usefixtures("candles") -def test_movement_pattern_kawgs(candles): - test = indicators.Pattern( - pattern=movement.above, candles=candles, indicator="open", indicator_two="low" - ) - test.calculate() - assert test.reading("above") is not None - - -@pytest.mark.usefixtures("candles") -def test_pattern_custom(candles): - test = indicators.Pattern(pattern=fake_pattern, candles=candles) - test.calculate() - assert test.reading("fake_pattern") is not None