Init: Refactor NMM mixture, separate generators and mixtures, tests r…

src/algorithms/__init__.py

@@ -0,0 +1,5 @@
+from src.algorithms.semiparam_algorithms.nvm_semi_param_algorithms.mu_estimation import SemiParametricMuEstimation
+from src.register.register import Registry
+
+ALGORITHM_REGISTRY: Registry = Registry()
+ALGORITHM_REGISTRY.register("mu_estimation", "NMVSemiparametric")(SemiParametricMuEstimation)

src/algorithms/nvm_semi_param_algorithms/mu_estimation.py → src/algorithms/semiparam_algorithms/nvm_semi_param_algorithms/mu_estimation.py

@@ -1,10 +1,12 @@
 import math
-from typing import Callable, TypedDict, Unpack
+from typing import Callable, Optional, TypedDict, Unpack
 
 import mpmath
 import numpy as np
 from numpy import _typing
 
+from src.estimators.estimate_result import EstimateResult
+
 M_DEFAULT_VALUE = 1000
 TOLERANCE_DEFAULT_VALUE = 10**-5
 OMEGA_DEFAULT_VALUE = lambda x: -1 * math.sin(x) if abs(x) <= math.pi else 0
@@ -33,7 +35,7 @@ class ParamsAnnotation(TypedDict, total=False):
         omega: Callable[[float], float]
         max_iterations: float
 
-    def __init__(self, sample: _typing.ArrayLike = None, **kwargs: Unpack[ParamsAnnotation]):
+    def __init__(self, sample: Optional[_typing.ArrayLike] = None, **kwargs: Unpack[ParamsAnnotation]):
         self.sample = np.array([]) if sample is None else sample
         self.m, self.tolerance, self.omega, self.max_iterations = self._validate_kwargs(**kwargs)
 
@@ -82,7 +84,7 @@ def __w(self, p: float, sample: np._typing.NDArray) -> float:
             y += e * self.omega(x)
         return y
 
-    def algorithm(self, sample: np._typing.NDArray) -> float:
+    def algorithm(self, sample: np._typing.NDArray) -> EstimateResult:
         """Root of this function is an estimation of mu
 
         Args:
@@ -91,25 +93,25 @@ def algorithm(self, sample: np._typing.NDArray) -> float:
         Returns: estimated mu value
 
         """
-
         if self.__w(0, sample) == 0:
-            return 0
+            return EstimateResult(value=0, success=True)
         if self.__w(0, sample) > 0:
-            return -1 * self.algorithm(-1 * sample)
+            second_result = self.algorithm(-1 * sample)
+            return EstimateResult(-1 * second_result.value, second_result.success)
         if self.__w(self.m, sample) < 0:
-            return self.m
+            return EstimateResult(value=self.m, success=False)
 
         left, right = 0.0, self.m
         iteration = 0
         while left <= right:
             mid = (right + left) / 2
             if iteration > self.max_iterations:
-                return mid
+                return EstimateResult(value=mid, success=False)
             iteration += 1
             if abs(self.__w(mid, sample)) < self.tolerance:
-                return mid
+                return EstimateResult(value=mid, success=True)
             elif self.__w(mid, sample) < 0:
                 left = mid
             else:
                 right = mid
-        return -1
+        return EstimateResult(value=-1, success=False)

src/estimators/abstract_estimator.py

@@ -0,0 +1,42 @@
+from abc import abstractmethod
+from typing import Self
+
+from numpy import _typing
+
+from src.algorithms import ALGORITHM_REGISTRY
+from src.estimators.estimate_result import EstimateResult
+
+
+class AbstractEstimator:
+    def __init__(self, algorithm_name: str, params: dict | None = None) -> None:
+        self.algorithm_name = algorithm_name
+        if params is None:
+            self.params = dict()
+        else:
+            self.params = params
+        self.estimate_result = EstimateResult()
+        self._registry = ALGORITHM_REGISTRY
+        self._purpose = ""
+
+    def get_params(self) -> dict:
+        return {"algorithm_name": self.algorithm_name, "params": self.params, "estimated_result": self.estimate_result}
+
+    def set_params(self, algorithm_name: str, params: dict | None = None) -> Self:
+        self.algorithm_name = algorithm_name
+        if params is None:
+            self.params = dict()
+        else:
+            self.params = params
+        return self
+
+    def get_available_algorithms(self) -> list[tuple[str, str]]:
+        return [key for key in self._registry.register_of_names.keys() if key[1] == self._purpose]
+
+    def estimate(self, sample: _typing.ArrayLike) -> EstimateResult:
+        cls = None
+        if (self.algorithm_name, self._purpose) in self._registry.register_of_names:
+            cls = self._registry.dispatch(self.algorithm_name, self._purpose)(sample, **self.params)
+        if cls is None:
+            raise ValueError("This algorithm does not exist")
+        self.estimate_result = cls.algorithm(sample)
+        return self.estimate_result

src/estimators/estimate_result.py

@@ -0,0 +1,8 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class EstimateResult:
+    value: float = -1
+    success: bool = False
+    message: str = "No message"

src/estimators/parametric/abstract_parametric_estimator.py

@@ -0,0 +1,12 @@
+from numpy import _typing
+
+from src.estimators.abstract_estimator import AbstractEstimator
+from src.estimators.estimate_result import EstimateResult
+
+
+class AbstractParametricEstimator(AbstractEstimator):
+    def __init__(self, algorithm_name: str, params: dict | None = None) -> None:
+        super().__init__(algorithm_name, params)
+
+    def estimate(self, sample: _typing.ArrayLike) -> EstimateResult:
+        return super().estimate(sample)

src/estimators/parametric/nm_parametric_estimator.py

@@ -0,0 +1,13 @@
+from numpy import _typing
+
+from src.estimators.estimate_result import EstimateResult
+from src.estimators.parametric.abstract_parametric_estimator import AbstractParametricEstimator
+
+
+class NMParametricEstimator(AbstractParametricEstimator):
+    def __init__(self, algorithm_name: str, params: dict | None = None) -> None:
+        super().__init__(algorithm_name, params)
+        self._purpose = "NMParametric"
+
+    def estimate(self, sample: _typing.ArrayLike) -> EstimateResult:
+        return super().estimate(sample)

src/estimators/parametric/nmv_parametric_estimator.py

@@ -0,0 +1,13 @@
+from numpy import _typing
+
+from src.estimators.estimate_result import EstimateResult
+from src.estimators.parametric.abstract_parametric_estimator import AbstractParametricEstimator
+
+
+class NMVParametricEstimator(AbstractParametricEstimator):
+    def __init__(self, algorithm_name: str, params: dict | None = None) -> None:
+        super().__init__(algorithm_name, params)
+        self._purpose = "NMVParametric"
+
+    def estimate(self, sample: _typing.ArrayLike) -> EstimateResult:
+        return super().estimate(sample)

src/estimators/parametric/nv_parametric_estimator.py

@@ -0,0 +1,13 @@
+from numpy import _typing
+
+from src.estimators.estimate_result import EstimateResult
+from src.estimators.parametric.abstract_parametric_estimator import AbstractParametricEstimator
+
+
+class NVParametricEstimator(AbstractParametricEstimator):
+    def __init__(self, algorithm_name: str, params: dict | None = None) -> None:
+        super().__init__(algorithm_name, params)
+        self._purpose = "NVParametric"
+
+    def estimate(self, sample: _typing.ArrayLike) -> EstimateResult:
+        return super().estimate(sample)

src/estimators/semiparametric/abstract_semiparametric_estimator.py

@@ -0,0 +1,14 @@
+from abc import abstractmethod
+
+from numpy import _typing
+
+from src.estimators.abstract_estimator import AbstractEstimator
+from src.estimators.estimate_result import EstimateResult
+
+
+class AbstractSemiParametricEstimator(AbstractEstimator):
+    def __init__(self, algorithm_name: str, params: dict | None = None) -> None:
+        super().__init__(algorithm_name, params)
+
+    def estimate(self, sample: _typing.ArrayLike) -> EstimateResult:
+        return super().estimate(sample)

src/estimators/semiparametric/nm_semiparametric_estimator.py

@@ -0,0 +1,13 @@
+from numpy import _typing
+
+from src.estimators.estimate_result import EstimateResult
+from src.estimators.semiparametric.abstract_semiparametric_estimator import AbstractSemiParametricEstimator
+
+
+class NMSemiParametricEstimator(AbstractSemiParametricEstimator):
+    def __init__(self, algorithm_name: str, params: dict | None = None) -> None:
+        super().__init__(algorithm_name, params)
+        self._purpose = "NMSemiparametric"
+
+    def estimate(self, sample: _typing.ArrayLike) -> EstimateResult:
+        return super().estimate(sample)

src/estimators/semiparametric/nmv_semiparametric_estimator.py

@@ -0,0 +1,13 @@
+from numpy import _typing
+
+from src.estimators.estimate_result import EstimateResult
+from src.estimators.semiparametric.abstract_semiparametric_estimator import AbstractSemiParametricEstimator
+
+
+class NMVSemiParametricEstimator(AbstractSemiParametricEstimator):
+    def __init__(self, algorithm_name: str, params: dict | None = None) -> None:
+        super().__init__(algorithm_name, params)
+        self._purpose = "NMVSemiparametric"
+
+    def estimate(self, sample: _typing.ArrayLike) -> EstimateResult:
+        return super().estimate(sample)

src/estimators/semiparametric/nv_semiparametric_estimator.py

@@ -0,0 +1,13 @@
+from numpy import _typing
+
+from src.estimators.estimate_result import EstimateResult
+from src.estimators.semiparametric.abstract_semiparametric_estimator import AbstractSemiParametricEstimator
+
+
+class NVSemiParametricEstimator(AbstractSemiParametricEstimator):
+    def __init__(self, algorithm_name: str, params: dict | None = None) -> None:
+        super().__init__(algorithm_name, params)
+        self._purpose = "NVSemiparametric"
+
+    def estimate(self, sample: _typing.ArrayLike) -> EstimateResult:
+        return super().estimate(sample)

src/generators/abstract_generator.py

@@ -0,0 +1,35 @@
+from abc import abstractmethod
+
+import numpy._typing as tpg
+
+from src.mixtures.abstract_mixture import AbstractMixtures
+
+
+class AbstractGenerator:
+    @staticmethod
+    @abstractmethod
+    def canonical_generate(mixture: AbstractMixtures, size: int) -> tpg.NDArray: ...
+
+    """Generate a sample of given size. Classical form of Mixture
+
+    Args:
+        mixture: NMM | NVM | NMVM
+        size: length of sample
+
+    Returns: sample of given size
+
+    """
+
+    @staticmethod
+    @abstractmethod
+    def classical_generate(mixture: AbstractMixtures, size: int) -> tpg.NDArray: ...
+
+    """Generate a sample of given size. Canonical form of Mixture
+
+    Args:
+        mixture: NMM | NVM | NMVM
+        size: length of sample
+
+    Returns: sample of given size
+
+    """

src/generators/nm_generator.py

@@ -0,0 +1,51 @@
+import numpy._typing as tpg
+import scipy
+
+from src.generators.abstract_generator import AbstractGenerator
+from src.mixtures.abstract_mixture import AbstractMixtures
+from src.mixtures.nm_mixture import NormalMeanMixtures
+
+
+class NMGenerator(AbstractGenerator):
+
+    @staticmethod
+    def classical_generate(mixture: AbstractMixtures, size: int) -> tpg.NDArray:
+        """Generate a sample of given size. Classical form of NMM
+
+        Args:
+            mixture: Normal Mean Mixture
+            size: length of sample
+
+        Returns: sample of given size
+
+        Raises:
+            ValueError: If mixture is not a Normal Mean Mixture
+
+        """
+
+        if not isinstance(mixture, NormalMeanMixtures):
+            raise ValueError("Mixture must be NormalMeanMixtures")
+        mixing_values = mixture.params.distribution.rvs(size=size)
+        normal_values = scipy.stats.norm.rvs(size=size)
+        return mixture.params.alpha + mixture.params.beta * mixing_values + mixture.params.gamma * normal_values
+
+    @staticmethod
+    def canonical_generate(mixture: AbstractMixtures, size: int) -> tpg.NDArray:
+        """Generate a sample of given size. Canonical form of NMM
+
+        Args:
+            mixture: Normal Mean Mixture
+            size: length of sample
+
+        Returns: sample of given size
+
+        Raises:
+            ValueError: If mixture is not a Normal Mean Mixture
+
+        """
+
+        if not isinstance(mixture, NormalMeanMixtures):
+            raise ValueError("Mixture must be NormalMeanMixtures")
+        mixing_values = mixture.params.distribution.rvs(size=size)
+        normal_values = scipy.stats.norm.rvs(size=size)
+        return mixing_values + mixture.params.sigma * normal_values

src/generators/nmv_generator.py

@@ -0,0 +1,55 @@
+import numpy._typing as tpg
+import scipy
+
+from src.generators.abstract_generator import AbstractGenerator
+from src.mixtures.abstract_mixture import AbstractMixtures
+from src.mixtures.nmv_mixture import NormalMeanVarianceMixtures
+
+
+class NMVGenerator(AbstractGenerator):
+
+    @staticmethod
+    def classical_generate(mixture: AbstractMixtures, size: int) -> tpg.NDArray:
+        """Generate a sample of given size. Classical form of NMVM
+
+        Args:
+            mixture: Normal Mean Variance Mixtures
+            size: length of sample
+
+        Returns: sample of given size
+
+        Raises:
+            ValueError: If mixture type is not Normal Mean Variance Mixtures
+
+        """
+
+        if not isinstance(mixture, NormalMeanVarianceMixtures):
+            raise ValueError("Mixture must be NormalMeanMixtures")
+        mixing_values = mixture.params.distribution.rvs(size=size)
+        normal_values = scipy.stats.norm.rvs(size=size)
+        return (
+            mixture.params.alpha
+            + mixture.params.beta * mixing_values
+            + mixture.params.gamma * (mixing_values**0.5) * normal_values
+        )
+
+    @staticmethod
+    def canonical_generate(mixture: AbstractMixtures, size: int) -> tpg.NDArray:
+        """Generate a sample of given size. Canonical form of NMVM
+
+        Args:
+            mixture: Normal Mean Variance Mixtures
+            size: length of sample
+
+        Returns: sample of given size
+
+        Raises:
+            ValueError: If mixture type is not Normal Mean Variance Mixtures
+
+        """
+
+        if not isinstance(mixture, NormalMeanVarianceMixtures):
+            raise ValueError("Mixture must be NormalMeanMixtures")
+        mixing_values = mixture.params.distribution.rvs(size=size)
+        normal_values = scipy.stats.norm.rvs(size=size)
+        return mixture.params.alpha + mixture.params.mu * mixing_values + (mixing_values**0.5) * normal_values