fixed

Author: plidan123

src/mixtures/abstract_mixture.py

@@ -126,4 +126,4 @@ def _params_validation(
                 raise ValueError(
                     f"Type mismatch: {name} should be {names_and_types[name]}, not {type(value)}"
                 )
-        return data_collector(**params)
+        return data_collector(**params)

src/mixtures/nm_mixture.py

@@ -69,7 +69,7 @@ def mix(u: float) -> float:
                 error += coeff * res.error * norm.moment(k)
         return mixture_moment, error
 
-    def _compute_cdf(self, x: float) -> Tuple[float, float]:
+    def _compute_cdf(self, x: float, params: Dict[str, Any]) -> Tuple[float, float]:
         if self.mixture_form == "classical":
             def fn(u: float) -> float:
                 return norm.cdf((x - self.params.alpha - self.params.beta * self.params.distribution.ppf(u)) / abs(self.params.gamma))

src/mixtures/nmv_mixture.py

@@ -1,29 +1,25 @@
 from dataclasses import dataclass
-from functools import lru_cache
-from typing import Any, Type, Dict
+from typing import Any, Type, Dict, Tuple
 
 import numpy as np
 from scipy.special import binom
 from scipy.stats import norm, 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
-from src.algorithms.support_algorithms.log_rqmc import LogRQMC
-from src.algorithms.support_algorithms.quad_integrator import QuadIntegrator
 from src.mixtures.abstract_mixture import AbstractMixtures
 
 @dataclass
 class _NMVMClassicDataCollector:
     alpha: float | int | np.int64
-    beta: float | int | np.int64
+    beta:  float | int | np.int64
     gamma: float | int | np.int64
     distribution: rv_frozen | rv_continuous
 
 @dataclass
 class _NMVMCanonicalDataCollector:
     alpha: float | int | np.int64
-    mu: float | int | np.int64
+    mu:    float | int | np.int64
     distribution: rv_frozen | rv_continuous
 
 class NormalMeanVarianceMixtures(AbstractMixtures):
@@ -33,96 +29,66 @@ class NormalMeanVarianceMixtures(AbstractMixtures):
     def __init__(
         self,
         mixture_form: str,
-        integrator_cls: Type[Integrator] = RQMCIntegrator,
-        integrator_params: Dict[str, Any] = None,
+        integrator_cls: Type[Integrator],
+        integrator_params: Dict[str, Any] | None = None,
         **kwargs: Any
     ) -> None:
         super().__init__(mixture_form, integrator_cls=integrator_cls, integrator_params=integrator_params, **kwargs)
 
-    def _compute_moment(self, n: int) -> tuple[float, float]:
+    def _compute_moment(self, n: int) -> Tuple[float, float]:
         def integrand(u: float) -> float:
-            result = 0.0
+            s = 0.0
             for k in range(n + 1):
                 for l in range(k + 1):
                     if self.mixture_form == "classical":
-                        result += (
-                            binom(n, n - k)
-                            * binom(k, k - l)
-                            * (self.params.beta ** (k - l))
+                        coef = binom(n, n - k) * binom(k, k - l)
+                        term = (
+                            (self.params.beta ** (k - l))
                             * (self.params.gamma ** l)
                             * (self.params.distribution.ppf(u) ** (k - l/2))
+                            * (self.params.alpha ** (n - k))
                             * norm.moment(l)
                         )
                     else:
-                        result += (
-                            binom(n, n - k)
-                            * binom(k, k - l)
-                            * (self.params.mu ** (k - l))
+                        coef = binom(n, n - k) * binom(k, k - l)
+                        term = (
+                            (self.params.nu ** (k - l))
                             * (self.params.distribution.ppf(u) ** (k - l/2))
+                            * (self.params.alpha ** (n - k))
                             * norm.moment(l)
                         )
-            return result
+                    s += coef * term if self.mixture_form == "classical" else term
+            return s
 
-        integrator = self.integrator_cls(**(self.integrator_params or {}))
-        result = integrator.compute(integrand)
-        return result.value, result.error
+        res = self.integrator_cls(**(self.integrator_params or {})).compute(integrand)
+        return res.value, res.error
 
-    def _compute_cdf(self, x: float) -> tuple[float, float]:
+    def _compute_cdf(self, x: float) -> Tuple[float, float]:
         def integrand(u: float) -> float:
-            ppf = self.params.distribution.ppf(u)
+            p = self.params.distribution.ppf(u)
             if self.mixture_form == "classical":
-                point = (x - self.params.alpha) / (np.sqrt(ppf) * self.params.gamma) - (self.params.beta / self.params.gamma * np.sqrt(ppf))
-            else:
-                point = (x - self.params.alpha) / np.sqrt(ppf) - (self.params.mu * np.sqrt(ppf))
-            return norm.cdf(point)
+                return norm.cdf((x - self.params.alpha - self.params.beta * p) / abs(self.params.gamma))
+            return norm.cdf((x - self.params.alpha) / np.sqrt(p) - self.params.mu * np.sqrt(p))
 
-        integrator = self.integrator_cls(**(self.integrator_params or {}))
-        result = integrator.compute(integrand)
-        return result.value, result.error
+        res = self.integrator_cls(**(self.integrator_params or {})).compute(integrand)
+        return res.value, res.error
 
-    def _compute_pdf(self, x: float) -> tuple[float, float]:
+    def _compute_pdf(self, x: float) -> Tuple[float, float]:
         def integrand(u: float) -> float:
-            ppf = self.params.distribution.ppf(u)
+            p = self.params.distribution.ppf(u)
             if self.mixture_form == "classical":
-                return (
-                    1 / np.sqrt(2 * np.pi * ppf * self.params.gamma ** 2)
-                    * np.exp(-((x - self.params.alpha) ** 2 + self.params.beta ** 2 * ppf ** 2) / (2 * ppf * self.params.gamma ** 2))
-                )
-            else:
-                return (
-                    1 / np.sqrt(2 * np.pi * ppf)
-                    * np.exp(-((x - self.params.alpha) ** 2 + self.params.mu ** 2 * ppf ** 2) / (2 * ppf))
-                )
+                return (1/abs(self.params.gamma)) * norm.pdf((x - self.params.alpha - self.params.beta * p)/abs(self.params.gamma))
+            return (1/abs(self.params.mu)) * norm.pdf((x - p)/abs(self.params.mu))
 
-        integrator = self.integrator_cls(**(self.integrator_params or {}))
-        result = integrator.compute(integrand)
-        if self.mixture_form == "classical":
-            val = np.exp(self.params.beta * (x - self.params.alpha) / self.params.gamma ** 2) * result.value
-        else:
-            val = np.exp(self.params.mu * (x - self.params.alpha)) * result.value
-        return val, result.error
+        res = self.integrator_cls(**(self.integrator_params or {})).compute(integrand)
+        return res.value, res.error
 
-    def _compute_logpdf(self, x: float) -> tuple[float, float]:
+    def _compute_logpdf(self, x: float) -> Tuple[float, float]:
         def integrand(u: float) -> float:
-            ppf = self.params.distribution.ppf(u)
+            p = self.params.distribution.ppf(u)
             if self.mixture_form == "classical":
-                return -(
-                    (x - self.params.alpha) ** 2
-                    + ppf ** 2 * self.params.beta ** 2
-                    + ppf * self.params.gamma ** 2 * np.log(2 * np.pi * ppf * self.params.gamma ** 2)
-                ) / (2 * ppf * self.params.gamma ** 2)
-            else:
-                return -((x - self.params.alpha) ** 2 + ppf ** 2 * self.params.mu ** 2 + ppf * np.log(2 * np.pi * ppf)) / (2 * ppf)
+                return np.log(1/abs(self.params.gamma)) + norm.logpdf((x - self.params.alpha - self.params.beta * p)/abs(self.params.gamma))
+            return np.log(1/abs(self.params.mu)) + norm.logpdf((x - p)/abs(self.params.mu))
 
-        integrator = self.integrator_cls(**(self.integrator_params or {}))
-        result = integrator.compute(integrand)
-        if self.mixture_form == "classical":
-            val = self.params.beta * (x - self.params.alpha) / self.params.gamma ** 2 + result.value
-        else:
-            val = self.params.mu * (x - self.params.alpha) + result.value
-        return val, result.error
-
-    @lru_cache()
-    def _integrand_func(self, u: float, d: float, gamma: float) -> float:
-        ppf = self.params.distribution.ppf(u)
-        return (1 / np.sqrt(2 * np.pi * ppf * abs(gamma) ** 2)) * np.exp(-d / (2 * ppf))
+        res = self.integrator_cls(**(self.integrator_params or {})).compute(integrand)
+        return res.value, res.error

src/mixtures/nv_mixture.py

@@ -35,12 +35,13 @@ def __init__(
         integrator_params: Dict[str, Any] = None,
         **kwargs: Any
     ) -> None:
-        super().__init__(mixture_form, **kwargs)
+        super().__init__(mixture_form, integrator_cls=integrator_cls, integrator_params=integrator_params, **kwargs)
         self.integrator_cls = integrator_cls
         self.integrator_params = integrator_params or {}
 
-    def _compute_moment(self, n: int, __: Dict[str, Any]) -> tuple[float, float]:
+    def _compute_moment(self, n: int) -> tuple[float, float]:
         gamma = getattr(self.params, 'gamma', 1)
+
         def integrand(u: float) -> float:
             return sum(
                 binom(n, k)
@@ -50,34 +51,40 @@ def integrand(u: float) -> float:
                 * norm.moment(k)
                 for k in range(n + 1)
             )
+
         integrator = self.integrator_cls(**self.integrator_params)
         result = integrator.compute(integrand)
         return result.value, result.error
 
-    def _compute_cdf(self, x: float, __: Dict[str, Any]) -> tuple[float, float]:
+    def _compute_cdf(self, x: float) -> tuple[float, float]:
         gamma = getattr(self.params, 'gamma', 1)
         param_norm = norm(0, gamma)
+
         def integrand(u: float) -> float:
             return param_norm.cdf((x - self.params.alpha) / np.sqrt(self.params.distribution.ppf(u)))
+
         integrator = self.integrator_cls(**self.integrator_params)
         result = integrator.compute(integrand)
         return result.value, result.error
 
-    def _compute_pdf(self, x: float, __: Dict[str, Any]) -> tuple[float, float]:
+    def _compute_pdf(self, x: float) -> tuple[float, float]:
         gamma = getattr(self.params, 'gamma', 1)
         d = (x - self.params.alpha) ** 2 / gamma ** 2
+
         def integrand(u: float) -> float:
             return self._integrand_func(u, d, gamma)
+
         integrator = self.integrator_cls(**self.integrator_params)
         result = integrator.compute(integrand)
         return result.value, result.error
 
-    def _compute_logpdf(self, x: float, __: Dict[str, Any]) -> tuple[float, float]:
+    def _compute_logpdf(self, x: float) -> tuple[float, float]:
         gamma = getattr(self.params, 'gamma', 1)
         d = (x - self.params.alpha) ** 2 / gamma ** 2
+
         def integrand(u: float) -> float:
             return self._log_integrand_func(u, d, gamma)
-        # For log-pdf you may choose LogRQMC or any integrator that supports it
+
         integrator = self.integrator_cls(**self.integrator_params)
         result = integrator.compute(integrand)
         return result.value, result.error