feat: 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) -> None:
+        ...
+
+    def compute(self, func: Callable) -> IntegrationResult:
         ...

src/algorithms/support_algorithms/quad_integrator.py

@@ -1,27 +1,60 @@
-from typing import Callable
+from typing import Callable, Any, Sequence
 
 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[Any, ...] = (),
+            full_output: int = 0,
+            epsabs: float | int = 1.49e-08,
+            epsrel: float | int = 1.49e-08,
+            limit: float | int = 50,
+            points: Sequence[float | int] | None = None,
+            weight: float | int | None = None,
+            wvar: Any = None,
+            wopts: Any = 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) -> IntegrationResult:
 
         """
         Compute integral via quad integrator
 
         Args:
             func: integrated function
-            params: Parameters of integration algorithm
 
         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 = self.params.pop('full_output', False)
+        result = quad(func, **self.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,48 @@ def __call__(self) -> tuple[float, float]:
 
 
 class RQMCIntegrator:
+    """
+    Randomize Quasi Monte Carlo Method
+
+    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.error_tolerance = error_tolerance
+        self.count = count
+        self.base_n = base_n
+        self.i_max = i_max
+        self.a = a
 
-    def compute_integral(self, func: Callable, params: dict) -> IntegrationResult:
+    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
         """
+        result = RQMC(
+            func,
+            error_tolerance=self.error_tolerance,
+            count=self.count,
+            base_n=self.base_n,
+            i_max=self.i_max,
+            a=self.a,
+        )()
+        return IntegrationResult(result[0], result[1])
 
-        rqmc = RQMC(func, **params)()
-        return IntegrationResult(rqmc[0], rqmc[1])

src/mixtures/abstract_mixture.py

@@ -5,6 +5,7 @@
 from scipy.stats import rv_continuous
 from scipy.stats.distributions import rv_frozen
 
+from src.algorithms.support_algorithms.integrator import Integrator
 
 class AbstractMixtures(metaclass=ABCMeta):
     """Base class for Mixtures"""
@@ -28,13 +29,13 @@ 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, integrator: Integrator) -> tuple[float, float]: ...
 
     @abstractmethod
-    def compute_cdf(self, x: float, params: dict) -> tuple[float, float]: ...
+    def compute_cdf(self, x: float, integrator: Integrator) -> tuple[float, float]: ...
 
     @abstractmethod
-    def compute_pdf(self, x: float, params: dict) -> tuple[float, float]: ...
+    def compute_pdf(self, x: float, integrator: Integrator) -> tuple[float, float]: ...
 
     @abstractmethod
     def compute_logpdf(self, x: float, params: dict) -> tuple[float, float]: ...