Skip to content

Commit

Permalink
Renamed Pattern to Amoprh
Browse files Browse the repository at this point in the history
  • Loading branch information
MerlinR committed Jan 11, 2024
1 parent 313c503 commit 4265a39
Show file tree
Hide file tree
Showing 11 changed files with 215 additions and 232 deletions.
3 changes: 2 additions & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
46 changes: 19 additions & 27 deletions hexital/core/hexital.py
Original file line number Diff line number Diff line change
@@ -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"
Expand Down Expand Up @@ -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")
Expand All @@ -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():
Expand Down Expand Up @@ -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})`
Expand Down
2 changes: 1 addition & 1 deletion hexital/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ def __init__(self, message):
super().__init__(message)


class InvalidPattern(Exception):
class InvalidAnalysis(Exception):
def __init__(self, message):
super().__init__(message)

Expand Down
2 changes: 1 addition & 1 deletion hexital/indicators/__init__.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
from .amorph import Amorph
from .adx import ADX
from .atr import ATR
from .ema import EMA
Expand All @@ -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
Expand Down
82 changes: 82 additions & 0 deletions hexital/indicators/amorph.py
Original file line number Diff line number Diff line change
@@ -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)
81 changes: 0 additions & 81 deletions hexital/indicators/pattern.py

This file was deleted.

16 changes: 8 additions & 8 deletions tests/core/test_hexital.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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")
Expand All @@ -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

Expand Down
Loading

0 comments on commit 4265a39

Please sign in to comment.