fixed

Author: plidan123

jupiter_examples/nm_sigma_estimation_comparison.ipynb

@@ -250,7 +250,7 @@
     "    \"\"\"\n",
     "    generator = NMGenerator()\n",
     "    mixture = NormalMeanMixtures(\"canonical\", sigma=real_sigma, distribution=distribution)\n",
-    "    return generator.canonical_generate(mixture, sample_len)\n",
+    "    return generator.generate(mixture, sample_len)\n",
     "\n",
     "def estimate_sigma_eigenvalue_based(sample, real_sigma, search_area, a, b):\n",
     "    sample_len = len(sample)\n",

src/generators/nm_generator.py

@@ -9,7 +9,7 @@
 class NMGenerator(AbstractGenerator):
 
     @staticmethod
-    def classical_generate(mixture: AbstractMixtures, size: int) -> tpg.NDArray:
+    def generate(mixture: AbstractMixtures, size: int) -> tpg.NDArray:
         """Generate a sample of given size. Classical form of NMM
 
         Args:
@@ -27,25 +27,6 @@ def classical_generate(mixture: AbstractMixtures, size: int) -> tpg.NDArray:
             raise ValueError("Mixture must be NormalMeanMixtures")
         mixing_values = mixture.params.distribution.rvs(size=size)
         normal_values = scipy.stats.norm.rvs(size=size)
+        if mixture.mixture_form == "canonical":
+            return mixing_values + mixture.params.sigma * normal_values
         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

@@ -9,7 +9,7 @@
 class NMVGenerator(AbstractGenerator):
 
     @staticmethod
-    def classical_generate(mixture: AbstractMixtures, size: int) -> tpg.NDArray:
+    def generate(mixture: AbstractMixtures, size: int) -> tpg.NDArray:
         """Generate a sample of given size. Classical form of NMVM
 
         Args:
@@ -27,29 +27,10 @@ def classical_generate(mixture: AbstractMixtures, size: int) -> tpg.NDArray:
             raise ValueError("Mixture must be NormalMeanMixtures")
         mixing_values = mixture.params.distribution.rvs(size=size)
         normal_values = scipy.stats.norm.rvs(size=size)
+        if mixture.mixture_form == "canonical":
+            return mixture.params.alpha + mixture.params.mu * mixing_values + (mixing_values ** 0.5) * normal_values
         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
+        )

src/generators/nv_generator.py

@@ -9,7 +9,7 @@
 class NVGenerator(AbstractGenerator):
 
     @staticmethod
-    def classical_generate(mixture: AbstractMixtures, size: int) -> tpg.NDArray:
+    def generate(mixture: AbstractMixtures, size: int) -> tpg.NDArray:
         """Generate a sample of given size. Classical form of NVM
 
         Args:
@@ -27,25 +27,6 @@ def classical_generate(mixture: AbstractMixtures, size: int) -> tpg.NDArray:
             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.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 NVM
-
-        Args:
-            mixture: Normal Variance Mixtures
-            size: length of sample
-
-        Returns: sample of given size
-
-        Raises:
-            ValueError: If mixture type is not Normal Variance Mixtures
-
-        """
-
-        if not isinstance(mixture, NormalVarianceMixtures):
-            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 + (mixing_values**0.5) * normal_values
+        if mixture.mixture_form == "canonical":
+            return mixture.params.alpha + (mixing_values ** 0.5) * normal_values
+        return mixture.params.alpha + mixture.params.gamma * (mixing_values**0.5) * normal_values

src/mixtures/abstract_mixture.py

@@ -1,25 +1,39 @@
 from abc import ABCMeta, abstractmethod
 from dataclasses import fields
-from typing import Any
+from typing import Any, List, Tuple, Union, Dict, Type
+import numpy as np
+from numpy.typing import NDArray
 
 from scipy.stats import rv_continuous
 from scipy.stats.distributions import rv_frozen
 
+from src.algorithms.support_algorithms.integrator import Integrator
+from src.algorithms.support_algorithms.rqmc import RQMCIntegrator  # default integrator
 
 class AbstractMixtures(metaclass=ABCMeta):
     """Base class for Mixtures"""
 
     _classical_collector: Any
     _canonical_collector: Any
 
-    @abstractmethod
-    def __init__(self, mixture_form: str, **kwargs: Any) -> None:
+    def __init__(
+        self,
+        mixture_form: str,
+        integrator_cls: Type[Integrator] = RQMCIntegrator,
+        integrator_params: Dict[str, Any] = None,
+        **kwargs: Any
+    ) -> None:
         """
-
         Args:
-            mixture_form: Form of Mixture classical or Canonical
-            **kwargs: Parameters of Mixture
+            mixture_form: Form of Mixture classical or canonical
+            integrator_cls: Class implementing Integrator protocol (default: RQMCIntegrator)
+            integrator_params: Parameters for integrator constructor (default: {{}})
+            **kwargs: Parameters of Mixture (alpha, gamma, etc.)
         """
+        self.mixture_form = mixture_form
+        self.integrator_cls = integrator_cls
+        self.integrator_params = integrator_params or {}
+
         if mixture_form == "classical":
             self.params = self._params_validation(self._classical_collector, kwargs)
         elif mixture_form == "canonical":
@@ -28,40 +42,88 @@ def __init__(self, mixture_form: str, **kwargs: Any) -> None:
             raise AssertionError(f"Unknown mixture form: {mixture_form}")
 
     @abstractmethod
-    def compute_moment(self, n: int, params: dict) -> tuple[float, float]: ...
+    def _compute_moment(self, n: int) -> Tuple[float, float]:
+        ...
+
+    def compute_moment(
+        self,
+        x: Union[List[int], int, NDArray[np.float64]]
+    ) -> Union[List[Tuple[float, float]], Tuple[float, float], NDArray[Any]]:
+        if isinstance(x, np.ndarray):
+            return np.array([self._compute_moment(xp) for xp in x], dtype=object)
+        elif isinstance(x, list):
+            return [self._compute_moment(xp) for xp in x]
+        elif isinstance(x, int):
+            return self._compute_moment(x)
+        else:
+            raise TypeError(f"Unsupported type for x: {type(x)}")
 
     @abstractmethod
-    def compute_cdf(self, x: float, params: dict) -> tuple[float, float]: ...
+    def _compute_pdf(self, x: float) -> Tuple[float, float]:
+        ...
+
+    def compute_pdf(
+        self,
+        x: Union[List[float], float, NDArray[np.float64]]
+    ) -> Union[List[Tuple[float, float]], Tuple[float, float], NDArray[Any]]:
+        if isinstance(x, np.ndarray):
+            return np.array([self._compute_pdf(xp) for xp in x], dtype=object)
+        elif isinstance(x, list):
+            return [self._compute_pdf(xp) for xp in x]
+        elif isinstance(x, float):
+            return self._compute_pdf(x)
+        else:
+            raise TypeError(f"Unsupported type for x: {type(x)}")
 
     @abstractmethod
-    def compute_pdf(self, x: float, params: dict) -> tuple[float, float]: ...
+    def _compute_logpdf(self, x: float) -> Tuple[float, float]:
+        ...
+
+    def compute_logpdf(
+        self,
+        x: Union[List[float], float, NDArray[np.float64]]
+    ) -> Union[List[Tuple[float, float]], Tuple[float, float], NDArray[Any]]:
+        if isinstance(x, np.ndarray):
+            return np.array([self._compute_logpdf(xp) for xp in x], dtype=object)
+        elif isinstance(x, list):
+            return [self._compute_logpdf(xp) for xp in x]
+        elif isinstance(x, float):
+            return self._compute_logpdf(x)
+        else:
+            raise TypeError(f"Unsupported type for x: {type(x)}")
 
     @abstractmethod
-    def compute_logpdf(self, x: float, params: dict) -> tuple[float, float]: ...
-
-    def _params_validation(self, data_collector: Any, params: dict[str, float | rv_continuous | rv_frozen]) -> Any:
-        """Mixture Parameters Validation
-
-        Args:
-            data_collector: Dataclass that collect parameters of Mixture
-            params: Input parameters
-
-        Returns: Instance of dataclass
-
-        Raises:
-            ValueError: If given parameters is unexpected
-            ValueError: If parameter type is invalid
-            ValueError: If parameters age not given
-
-        """
-
+    def _compute_cdf(self, x: float) -> Tuple[float, float]:
+        ...
+
+    def compute_cdf(
+        self,
+        x: Union[List[float], float, NDArray[np.float64]]
+    ) -> Union[List[Tuple[float, float]], Tuple[float, float], NDArray[Any]]:
+        if isinstance(x, np.ndarray):
+            return np.array([self._compute_cdf(xp) for xp in x], dtype=object)
+        elif isinstance(x, list):
+            return [self._compute_cdf(xp) for xp in x]
+        elif isinstance(x, float):
+            return self._compute_cdf(x)
+        else:
+            raise TypeError(f"Unsupported type for x: {type(x)}")
+
+    def _params_validation(
+        self,
+        data_collector: Any,
+        params: dict[str, float | rv_continuous | rv_frozen]
+    ) -> Any:
+        """Mixture Parameters Validation"""
         dataclass_fields = fields(data_collector)
         if len(params) != len(dataclass_fields):
             raise ValueError(f"Expected {len(dataclass_fields)} arguments, got {len(params)}")
-        names_and_types = dict((field.name, field.type) for field in dataclass_fields)
-        for pair in params.items():
-            if pair[0] not in names_and_types:
-                raise ValueError(f"Unexpected parameter {pair[0]}")
-            if not isinstance(pair[1], names_and_types[pair[0]]):
-                raise ValueError(f"Type missmatch: {pair[0]} should be {names_and_types[pair[0]]}, not {type(pair[1])}")
+        names_and_types = {field.name: field.type for field in dataclass_fields}
+        for name, value in params.items():
+            if name not in names_and_types:
+                raise ValueError(f"Unexpected parameter {name}")
+            if not isinstance(value, names_and_types[name]):
+                raise ValueError(
+                    f"Type mismatch: {name} should be {names_and_types[name]}, not {type(value)}"
+                )
         return data_collector(**params)

src/procedures/semiparametric/nvm_semiparametric/g_estimation_given_mu_rqmc_based.py

@@ -32,7 +32,7 @@ def v_sequence_default_value(n: float) -> float:
 INTEGRATION_LIMIT_DEFAULT_VALUE: int = 50
 
 
-class NMVEstimationDensityInvMTquadRQMCBased:
+class SemiParametricGEstimationGivenMuRQMCBased:
     """Estimation of mixing density function g (xi density function) of NVM mixture represented in canonical form Y =
     alpha + mu*xi + sqrt(xi)*N, where alpha = 0 and mu is given.
 
@@ -52,13 +52,13 @@ class ParamsAnnotation(TypedDict, total=False):
         integration_tolerance: float
         integration_limit: int
 
-    def __init__(self, sample: Optional[np.ndarray] = None, **kwargs: Unpack[ParamsAnnotation]):
+    def __init__(self, sample: Optional[_typing.NDArray[np.float64]] = None, **kwargs: Unpack[ParamsAnnotation]):
         self.x_powers: Dict[float, np.ndarray] = {}
         self.second_u_integrals: np.ndarray
         self.first_u_integrals: np.ndarray
         self.gamma_grid: np.ndarray
         self.v_grid: np.ndarray
-        self.sample: np.ndarray = np.array([]) if sample is None else sample
+        self.sample: _typing.NDArray[np.float64] = np.array([]) if sample is None else sample
         self.n: int = len(self.sample)
         (
             self.mu,
@@ -171,6 +171,6 @@ def compute_integrals_for_x(self, x: float) -> float:
         total = (first_integral + second_integral) / self.denominator
         return max(0.0, total.real)
 
-    def compute(self, sample: np.ndarray) -> EstimateResult:
+    def algorithm(self, sample: np._typing.NDArray) -> EstimateResult:
         y_data = [self.compute_integrals_for_x(x) for x in self.x_data]
         return EstimateResult(list_value=y_data, success=True)

src/procedures/support/integrator.py

@@ -0,0 +1,20 @@
+from dataclasses import dataclass
+from typing import Any, Protocol, Callable, Optional
+
+
+@dataclass
+class IntegrationResult:
+    value: float
+    error: float
+    message: Optional[dict[str, Any]] | None = None
+
+
+class Integrator(Protocol):
+
+    """Base class for integral calculation"""
+
+    def __init__(self) -> None:
+        ...
+
+    def compute(self, func: Callable) -> IntegrationResult:
+        ...