diff options
Diffstat (limited to 'openbb_platform/obbject_extensions/charting/openbb_charting/core/plotly_ta/base.py')
-rw-r--r-- | openbb_platform/obbject_extensions/charting/openbb_charting/core/plotly_ta/base.py | 198 |
1 files changed, 198 insertions, 0 deletions
diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/core/plotly_ta/base.py b/openbb_platform/obbject_extensions/charting/openbb_charting/core/plotly_ta/base.py new file mode 100644 index 00000000000..126f3b195fc --- /dev/null +++ b/openbb_platform/obbject_extensions/charting/openbb_charting/core/plotly_ta/base.py @@ -0,0 +1,198 @@ +from typing import Any, Callable, Dict, Iterator, List, Optional, Type + +import pandas as pd + +from .data_classes import ChartIndicators, TAIndicator + + +def columns_regex(df_ta: pd.DataFrame, name: str) -> List[str]: + """Return columns that match regex name""" + column_name = df_ta.filter(regex=rf"{name}(?=[^\d]|$)").columns.tolist() + + return column_name + + +class Indicator: + """Class for technical indicator.""" + + def __init__( + self, + func: Callable, + name: str = "", + **attrs: Any, + ) -> None: + self.func = func + self.name = name + self.attrs = attrs + + def __call__(self, *args: Any, **kwargs: Any) -> Any: + return self.func(*args, **kwargs) + + +class PluginMeta(type): + """Metaclass for all Plotly plugins.""" + + __indicators__: List[Indicator] = [] + __static_methods__: list = [] + __ma_mode__: List[str] = [] + __inchart__: List[str] = [] + __subplots__: List[str] = [] + + def __new__(mcs: Type["PluginMeta"], *args: Any, **kwargs: Any) -> "PluginMeta": + name, bases, attrs = args + indicators: Dict[str, Indicator] = {} + cls_attrs: Dict[str, list] = { + "__ma_mode__": [], + "__inchart__": [], + "__subplots__": [], + } + new_cls = super().__new__(mcs, name, bases, attrs, **kwargs) + for base in reversed(new_cls.__mro__): + for elem, value in base.__dict__.items(): + if elem in indicators: + del indicators[elem] + + is_static_method = isinstance(value, staticmethod) + if is_static_method: + value = value.__func__ # noqa: PLW2901 + if isinstance(value, Indicator): + if is_static_method: + raise TypeError( + f"Indicator {value.name} can't be a static method" + ) + indicators[value.name] = value + elif is_static_method and elem not in new_cls.__static_methods__: + new_cls.__static_methods__.append(elem) + + if elem in ["__ma_mode__", "__inchart__", "__subplots__"]: + cls_attrs[elem].extend(value) + + new_cls.__indicators__ = list(indicators.values()) + new_cls.__static_methods__ = list(set(new_cls.__static_methods__)) + new_cls.__ma_mode__ = list(set(cls_attrs["__ma_mode__"])) + new_cls.__inchart__ = list(set(cls_attrs["__inchart__"])) + new_cls.__subplots__ = list(set(cls_attrs["__subplots__"])) + + return new_cls + + def __iter__(cls: Type["PluginMeta"]) -> Iterator[Indicator]: # type: ignore + return iter(cls.__indicators__) + + # pylint: disable=unused-argument + def __init__(cls, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + + +class PltTA(metaclass=PluginMeta): + """The base class that all Plotly plugins must inherit from.""" + + indicators: ChartIndicators + intraday: bool = False + df_stock: pd.DataFrame + df_ta: pd.DataFrame + df_fib: pd.DataFrame + close_column: Optional[str] = "close" + params: Dict[str, TAIndicator] = {} + inchart_colors: List[str] = [] + show_volume: bool = True + + __static_methods__: list = [] + __indicators__: List[Indicator] = [] + __ma_mode__: List[str] = [] + __inchart__: List[str] = [] + __subplots__: List[str] = [] + + # pylint: disable=unused-argument + def __new__(cls, *args: Any, **kwargs: Any) -> "PltTA": + if cls is PltTA: + raise TypeError("Can't instantiate abstract class Plugin directly") + self = super().__new__(cls) + + static_methods = cls.__static_methods__ + indicators = cls.__indicators__ + + for item in indicators: + # we make sure that the indicator is bound to the instance + if not hasattr(self, item.name): + setattr(self, item.name, item.func.__get__(self, cls)) + + for static_method in static_methods: + if not hasattr(self, static_method): + setattr(self, static_method, staticmethod(getattr(self, static_method))) + + for attr, value in cls.__dict__.items(): + if attr in [ + "__ma_mode__", + "__inchart__", + "__subplots__", + ] and value not in getattr(self, attr): + getattr(self, attr).extend(value) + + return self + + @property + def ma_mode(self) -> List[str]: + return list(set(self.__ma_mode__)) + + @ma_mode.setter + def ma_mode(self, value: List[str]): + self.__ma_mode__ = value + + def add_plugins(self, plugins: List["PltTA"]) -> None: + """Add plugins to current instance""" + for plugin in plugins: + for item in plugin.__indicators__: + # pylint: disable=unnecessary-dunder-call + if not hasattr(self, item.name): + setattr(self, item.name, item.func.__get__(self, type(self))) + self.__indicators__.append(item) + + for static_method in plugin.__static_methods__: + if not hasattr(self, static_method): + setattr( + self, static_method, staticmethod(getattr(self, static_method)) + ) + for attr, value in plugin.__dict__.items(): + if attr in [ + "__ma_mode__", + "__inchart__", + "__subplots__", + ] and value not in getattr(self, attr): + getattr(self, attr).extend(value) + + def remove_plugins(self, plugins: List["PltTA"]) -> None: + """Remove plugins from current instance""" + for plugin in plugins: + for item in plugin.__indicators__: + delattr(self, item.name) + self.__indicators__.remove(item) + + for static_method in plugin.__static_methods__: + delattr(self, static_method) + + def __iter__(self) -> Iterator[Indicator]: + return iter(self.__indicators__) + + def get_float_precision(self) -> str: + """Returns f-string precision format""" + price = self.df_stock[self.close_column].tail(1).values[0] + float_precision = ( + ",.2f" if price > 1.10 else "" if len(str(price)) < 8 else ".6f" + ) + return float_precision + + +def indicator( + name: str = "", + **attrs: Any, +) -> Callable: + """Decorator for adding indicators to a plugin class.""" + attrs["name"] = name + + def decorator(func: Callable) -> Indicator: + if not attrs.pop("name", ""): + name = func.__name__ + + return Indicator(func, name, **attrs) + + return decorator |