integration parameters are now part of the constructor

Author: Amellgo

src/algorithms/semiparam_algorithms/nvm_semi_param_algorithms/g_estimation_given_mu_rqmc_based.py

@@ -31,7 +31,6 @@ def v_sequence_default_value(n: float) -> float:
 GRID_SIZE_DEFAULT_VALUE: int = 200
 INTEGRATION_TOLERANCE_DEFAULT_VALUE: float = 1e-2
 INTEGRATION_LIMIT_DEFAULT_VALUE: int = 50
-INTEGRATOR_DEFAULT = RQMCIntegrator()
 
 
 class SemiParametricGEstimationGivenMuRQMCBased:
@@ -70,8 +69,7 @@ def __init__(self, sample: Optional[_typing.NDArray[np.float64]] = None, **kwarg
             self.x_data,
             self.grid_size,
             self.integration_tolerance,
-            self.integration_limit,
-            self.integrator
+            self.integration_limit
         ) = self._validate_kwargs(self.n, **kwargs)
         self.denominator: float = 2 * math.pi * self.n
         self.precompute_gamma_grid()
@@ -81,7 +79,7 @@ def __init__(self, sample: Optional[_typing.NDArray[np.float64]] = None, **kwarg
     @staticmethod
     def _validate_kwargs(
         n: int, **kwargs: Unpack[ParamsAnnotation]
-    ) -> tuple[float, float, float, float, List[float], int, float, int, Integrator]:
+    ) -> tuple[float, float, float, float, List[float], int, float, int]:
         mu: float = kwargs.get("mu", MU_DEFAULT_VALUE)
         gmm: float = kwargs.get("gmm", GAMMA_DEFAULT_VALUE)
         u_value: float = kwargs.get("u_value", U_SEQUENCE_DEFAULT_VALUE(n))
@@ -90,8 +88,7 @@ def _validate_kwargs(
         grid_size: int = kwargs.get("grid_size", GRID_SIZE_DEFAULT_VALUE)
         integration_tolerance: float = kwargs.get("integration_tolerance", INTEGRATION_TOLERANCE_DEFAULT_VALUE)
         integration_limit: int = kwargs.get("integration_limit", INTEGRATION_LIMIT_DEFAULT_VALUE)
-        integrator: Integrator = kwargs.get("integrator_type", INTEGRATOR_DEFAULT)
-        return mu, gmm, u_value, v_value, x_data, grid_size, integration_tolerance, integration_limit, integrator
+        return mu, gmm, u_value, v_value, x_data, grid_size, integration_tolerance, integration_limit
 
     def conjugate_psi(self, u: float) -> complex:
         return complex((u**2) / 2, self.mu * u)
@@ -166,16 +163,15 @@ def second_v_integrand(self, v: float, x: float) -> np.ndarray:
         x_power = self.x_powers[x][idx]
         return (self.second_u_integrals[idx] * x_power) / gamma_val
 
-    def compute_integrals_for_x(self, x: float, integrator: Integrator = None) -> float:
+    def compute_integrals_for_x(self, x: float, integrator: Integrator = RQMCIntegrator()) -> float:
         """Compute integrals using RQMC for v-integration."""
-        integrator = integrator or self.integrator
-        first_integral = integrator.compute_integral(func=lambda t: np.sum(self.first_v_integrand(t * self.v_value, x)) * self.v_value).value
+        first_integral = integrator.compute(func=lambda t: np.sum(self.first_v_integrand(t * self.v_value, x)) * self.v_value).value
 
-        second_integral = integrator.compute_integral(func=lambda t: np.sum(self.second_v_integrand(-t * self.v_value, x)) * self.v_value).value
+        second_integral = integrator.compute(func=lambda t: np.sum(self.second_v_integrand(-t * self.v_value, x)) * self.v_value).value
 
         total = (first_integral + second_integral) / self.denominator
         return max(0.0, total.real)
 
     def algorithm(self, sample: np._typing.NDArray) -> EstimateResult:
-        y_data = [self.compute_integrals_for_x(x, self.integrator) for x in self.x_data]
+        y_data = [self.compute_integrals_for_x(x, RQMCIntegrator()) for x in self.x_data]
         return EstimateResult(list_value=y_data, success=True)

src/algorithms/support_algorithms/integrator.py

@@ -13,5 +13,8 @@ class Integrator(Protocol):
 
     """Base class for integral calculation"""
 
-    def compute_integral(self, func: Callable, params: dict) -> IntegrationResult:
+    def __init__(self):
+        ...
+
+    def compute(self, func: Callable) -> IntegrationResult:
         ...

src/algorithms/support_algorithms/quad_integrator.py

@@ -1,11 +1,45 @@
-from typing import Callable
+from typing import Callable, Optional
 
 from scipy.integrate import quad
 from src.algorithms.support_algorithms.integrator import IntegrationResult
 
 class QuadIntegrator:
 
-    def compute_integral(self, func: Callable, params: dict) -> IntegrationResult:
+    def __init__(
+            self,
+            a: float = 0,
+            b: float = 1,
+            args: tuple = (),
+            full_output: int = 0,
+            epsabs: float | int = 1.49e-08,
+            epsrel: float | int = 1.49e-08,
+            limit: float | int = 50,
+            points=None,
+            weight: float | int = None,
+            wvar=None,
+            wopts=None,
+            maxp1: float | int = 50,
+            limlst: int = 50,
+            complex_func: bool = False,
+    ):
+        self.params = {
+            'a': a,
+            'b': b,
+            'args': args,
+            'full_output': full_output,
+            'epsabs': epsabs,
+            'epsrel': epsrel,
+            'limit': limit,
+            'points': points,
+            'weight': weight,
+            'wvar': wvar,
+            'wopts': wopts,
+            'maxp1': maxp1,
+            'limlst': limlst,
+            'complex_func': complex_func
+        }
+
+    def compute(self, func: Callable, params: dict) -> IntegrationResult:
 
         """
         Compute integral via quad integrator
@@ -17,11 +51,11 @@ def compute_integral(self, func: Callable, params: dict) -> IntegrationResult:
         Returns: moment approximation and error tolerance
         """
 
-        full_output_requested = params.pop('full_output', False)
-        quad_res = quad(func, **params)
-        if full_output_requested:
-            value, error, message = quad_res
+        verbose = params.pop('full_output', False)
+        result = quad(func, **params)
+        if verbose:
+            value, error, message = result
         else:
-            value, error = quad_res
+            value, error = result
             message = None
         return IntegrationResult(value, error, message)

src/algorithms/support_algorithms/rqmc.py

@@ -229,17 +229,44 @@ def __call__(self) -> tuple[float, float]:
 
 
 class RQMCIntegrator:
+    """
+    Randomize Quasi Monte Carlo Method
 
-    def compute_integral(self, func: Callable, params: dict) -> IntegrationResult:
+    Args:
+        error_tolerance: pre-specified error tolerance
+        count: number of rows of random values matrix
+        base_n: number of columns of random values matrix
+        i_max: allowed number of cycles
+        a: parameter for quantile of normal distribution
+
+    """
+
+    def __init__(
+            self,
+            error_tolerance: float = 1e-6,
+            count: int = 25,
+            base_n: int = 2 ** 6,
+            i_max: int = 100,
+            a: float = 0.00047,
+    ):
+        self.params = {
+            'error_tolerance': error_tolerance,
+            "count": count,
+            "base_n": base_n,
+            "i_max": i_max,
+            "a": a
+        }
+
+    def compute(self, func: Callable) -> IntegrationResult:
         """
         Compute integral via RQMC integrator
 
         Args:
             func: integrated function
-            params: Parameters of integration algorithm
 
         Returns: moment approximation and error tolerance
         """
 
-        rqmc = RQMC(func, **params)()
-        return IntegrationResult(rqmc[0], rqmc[1])
+        result = RQMC(func, **self.params)()
+        return IntegrationResult(result[0], result[1])
+

src/mixtures/nm_mixture.py

@@ -60,175 +60,155 @@ def _params_validation(self, data_collector: Any, params: dict[str, float | rv_c
             raise ValueError("Gamma cant be zero")
         return data_class
 
-    def _classical_moment(self, n: int, params: dict, integrator: Integrator = None) -> tuple[float, float]:
+    def _classical_moment(self, n: int, integrator: Integrator = QuadIntegrator()) -> tuple[float, float]:
         """
         Compute n-th moment of classical NMM
 
         Args:
             n (): Moment ordinal
-            params (): Parameters of integration algorithm
-            integrator (): type of integrator to computing
+            integrator (): class of integrator with params to computing
 
         Returns: moment approximation and error tolerance
 
         """
         mixture_moment = 0
         error_tolerance = 0
-        integrator = integrator or QuadIntegrator()
         for k in range(0, n + 1):
             for l in range(0, k + 1):
                 coefficient = binom(n, n - k) * binom(k, k - l) * (self.params.beta ** (k - l)) * (
                             self.params.gamma ** l)
-                mixing_moment = integrator.compute_integral(lambda u: self.params.distribution.ppf(u) ** (k - l),
-                                                            {"a": 0, "b": 1, **params})
+                mixing_moment = integrator.compute(lambda u: self.params.distribution.ppf(u) ** (k - l))
                 error_tolerance += (self.params.beta ** (k - l)) * mixing_moment.error
                 mixture_moment += coefficient * (self.params.alpha ** (n - k)) * mixing_moment.value * norm.moment(l)
             return mixture_moment, error_tolerance
 
 
-    def _canonical_moment(self, n: int, params: dict, integrator: Integrator = None) -> tuple[float, float]:
+    def _canonical_moment(self, n: int, integrator: Integrator = QuadIntegrator()) -> tuple[float, float]:
         """
         Compute n-th moment of canonical NMM
 
         Args:
             n (): Moment ordinal
-            params (): Parameters of integration algorithm
-            integrator (): type of integrator to computing
+            integrator (): class of integrator with params to computing
 
         Returns: moment approximation and error tolerance
 
         """
         mixture_moment = 0
         error_tolerance = 0
-        integrator = integrator or QuadIntegrator()
         for k in range(0, n + 1):
             coefficient = binom(n, n - k) * (self.params.sigma ** k)
-            mixing_moment = integrator.compute_integral(lambda u: self.params.distribution.ppf(u) ** (n - k),
-                                                        {"a": 0, "b": 1, **params})
+            mixing_moment = integrator.compute(lambda u: self.params.distribution.ppf(u) ** (n - k))
             error_tolerance += mixing_moment.error
             mixture_moment += coefficient * mixing_moment.value * norm.moment(k)
         return mixture_moment, error_tolerance
 
-    def compute_moment(self, n: int, params: dict, integrator: Integrator = None) -> tuple[float, float]:
+    def compute_moment(self, n: int, integrator: Integrator = QuadIntegrator) -> tuple[float, float]:
         """
         Compute n-th moment of  NMM
 
         Args:
             n (): Moment ordinal
-            params (): Parameters of integration algorithm
-            integrator (): type of integrator to computing
+            integrator (): class of integrator with params to computing
 
         Returns: moment approximation and error tolerance
 
         """
         if isinstance(self.params, _NMMClassicDataCollector):
-            return self._classical_moment(n, params, integrator)
-        return self._canonical_moment(n, params, integrator)
+            return self._classical_moment(n, integrator)
+        return self._canonical_moment(n, integrator)
 
-    def _canonical_compute_cdf(self, x: float, params: dict, integrator: Integrator = None) -> tuple[float, float]:
+    def _canonical_compute_cdf(self, x: float, integrator: Integrator = RQMCIntegrator) -> tuple[float, float]:
         """
         Equation for canonical cdf
         Args:
             x (): point
-            params (): parameters of RQMC algorithm
-            integrator (): type of integrator to computing
+            integrator (): class of integrator with params to computing
 
         Returns: computed cdf and error tolerance
 
         """
-        integrator = integrator or RQMCIntegrator()
-        rqmc = integrator.compute_integral(func=lambda u: norm.cdf((x - self.params.distribution.ppf(u)) / np.abs(self.params.sigma)), **params)
-        return rqmc.value, rqmc.error
+        result = integrator.compute(func=lambda u: norm.cdf((x - self.params.distribution.ppf(u)) / np.abs(self.params.sigma)))
+        return result.value, result.error
 
-    def _classical_compute_cdf(self, x: float, params: dict, integrator: Integrator = None) -> tuple[float, float]:
+    def _classical_compute_cdf(self, x: float, integrator: Integrator = RQMCIntegrator) -> tuple[float, float]:
         """
         Equation for classic cdf
         Args:
             x (): point
-            params (): parameters of RQMC algorithm
-            integrator (): type of integrator to computing
+            integrator (): class of integrator with params to computing
 
         Returns: computed cdf and error tolerance
 
         """
-        integrator = integrator or RQMCIntegrator()
-        rqmc = integrator.compute_integral(func=
+        result = integrator.compute(func=
             lambda u: norm.cdf(
                 (x - self.params.alpha - self.params.beta * self.params.distribution.ppf(u)) / np.abs(self.params.gamma)
-            ),
-            **params
+            )
         )
-        return rqmc.value, rqmc.error
+        return result.value, result.error
 
-    def compute_cdf(self, x: float, params: dict, integrator: Integrator = None) -> tuple[float, float]:
+    def compute_cdf(self, x: float, integrator: Integrator = RQMCIntegrator) -> tuple[float, float]:
         """
         Choose equation for cdf estimation depends on Mixture form
         Args:
             x (): point
-            params (): parameters of RQMC algorithm
-            integrator (): type of integrator to computing
+            integrator (): class of integrator with params to computing
 
         Returns: Computed pdf and error tolerance
 
         """
         if isinstance(self.params, _NMMCanonicalDataCollector):
-            return self._canonical_compute_cdf(x, params, integrator)
-        return self._classical_compute_cdf(x, params, integrator)
+            return self._canonical_compute_cdf(x, integrator)
+        return self._classical_compute_cdf(x, integrator)
 
-    def _canonical_compute_pdf(self, x: float, params: dict, integrator: Integrator = None) -> tuple[float, float]:
+    def _canonical_compute_pdf(self, x: float, integrator: Integrator = RQMCIntegrator) -> tuple[float, float]:
         """
         Equation for canonical pdf
         Args:
             x (): point
-            params (): parameters of RQMC algorithm
-            integrator (): type of integrator to computing
+            integrator (): class of integrator with params to computing
 
         Returns: computed pdf and error tolerance
 
         """
-        integrator = integrator or RQMCIntegrator()
-        rqmc = integrator.compute_integral(func=
+        result = integrator.compute(func=
             lambda u: (1 / np.abs(self.params.sigma))
             * norm.pdf((x - self.params.distribution.ppf(u)) / np.abs(self.params.sigma)),
-            **params
         )
-        return rqmc.value, rqmc.error
+        return result.value, result.error
 
-    def _classical_compute_pdf(self, x: float, params: dict, integrator: Integrator = None) -> tuple[float, float]:
+    def _classical_compute_pdf(self, x: float, integrator: Integrator = RQMCIntegrator()) -> tuple[float, float]:
         """
         Equation for classic pdf
         Args:
             x (): point
-            params (): parameters of RQMC algorithm
-            integrator (): type of integrator to computing
+            integrator (): class of integrator with params to computing
 
         Returns: computed pdf and error tolerance
 
         """
-        integrator = integrator or RQMCIntegrator()
-        rqmc = integrator.compute_integral(func=
+        result = integrator.compute(func=
             lambda u: (1 / np.abs(self.params.gamma))
             * norm.pdf(
                 (x - self.params.alpha - self.params.beta * self.params.distribution.ppf(u)) / np.abs(self.params.gamma)
             ),
-            **params
         )
-        return rqmc.value, rqmc.error
+        return result.value, result.error
 
-    def compute_pdf(self, x: float, params: dict, integrator: Integrator = None) -> tuple[float, float]:
+    def compute_pdf(self, x: float, integrator: Integrator = RQMCIntegrator()) -> tuple[float, float]:
         """
         Choose equation for pdf estimation depends on Mixture form
         Args:
             x (): point
-            params (): parameters of RQMC algorithm
-            integrator (): type of integrator to computing
+            integrator (): class of integrator with params to computing
 
         Returns: Computed pdf and error tolerance
 
         """
         if isinstance(self.params, _NMMCanonicalDataCollector):
-            return self._canonical_compute_pdf(x, params, integrator)
-        return self._classical_compute_pdf(x, params, integrator)
+            return self._canonical_compute_pdf(x, integrator)
+        return self._classical_compute_pdf(x, integrator)
 
     def _classical_compute_log_pdf(self, x: float, params: dict) -> tuple[float, float]:
         """