From fce29c0aa21581c12374affd06c3fe06212a98ca Mon Sep 17 00:00:00 2001 From: plidan123 Date: Sun, 25 May 2025 06:58:12 +0300 Subject: [PATCH] feat: add batch support for mixture evaluations and fix type annotations - Extended `moment`, `pdf`, `cdf`, and `logpdf` methods to support both single values and lists of inputs for batch computation. - Improved usability and performance for vectorized mixture evaluations. - Fixed type annotation issues to resolve mypy errors and ensure compliance with abstract base class method signatures. This enhancement improves usability and performance for batch evaluations. --- src/mixtures/abstract_mixture.py | 2 - src/mixtures/nm_mixture.py | 276 ++++++++----------------------- src/mixtures/nmv_mixture.py | 202 +++++++++------------- src/mixtures/nv_mixture.py | 95 ++++++----- 4 files changed, 197 insertions(+), 378 deletions(-) diff --git a/src/mixtures/abstract_mixture.py b/src/mixtures/abstract_mixture.py index 13c8e42..7f5afbc 100644 --- a/src/mixtures/abstract_mixture.py +++ b/src/mixtures/abstract_mixture.py @@ -62,6 +62,4 @@ def _params_validation(self, data_collector: Any, params: dict[str, float | rv_c 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])}") return data_collector(**params) diff --git a/src/mixtures/nm_mixture.py b/src/mixtures/nm_mixture.py index 33c7580..bf0be15 100644 --- a/src/mixtures/nm_mixture.py +++ b/src/mixtures/nm_mixture.py @@ -1,5 +1,5 @@ from dataclasses import dataclass -from typing import Any +from typing import Any, List, Tuple import numpy as np from scipy.integrate import quad @@ -14,9 +14,6 @@ @dataclass class _NMMClassicDataCollector: - """TODO: Change typing from float | int | etc to Protocol with __addition__ __multiplication__ __subtraction__""" - - """Data Collector for parameters of classical NMM""" alpha: float | int | np.int64 beta: float | int | np.int64 gamma: float | int | np.int64 @@ -25,9 +22,6 @@ class _NMMClassicDataCollector: @dataclass class _NMMCanonicalDataCollector: - """TODO: Change typing from float | int | etc to Protocol with __addition__ __multiplication__ __subtraction__""" - - """Data Collector for parameters of canonical NMM""" sigma: float | int | np.int64 distribution: rv_frozen | rv_continuous @@ -37,225 +31,85 @@ class NormalMeanMixtures(AbstractMixtures): _canonical_collector = _NMMCanonicalDataCollector def __init__(self, mixture_form: str, **kwargs: Any) -> None: - """ - Read Doc of Parent Method - """ - super().__init__(mixture_form, **kwargs) def _params_validation(self, data_collector: Any, params: dict[str, float | rv_continuous | rv_frozen]) -> Any: - """ - Read parent method doc - - Raises: - ValueError: If canonical Mixture has negative sigma parameter - - """ - data_class = super()._params_validation(data_collector, params) if hasattr(data_class, "sigma") and data_class.sigma <= 0: - raise ValueError("Sigma cant be zero or negative") + raise ValueError("Sigma can't be zero or negative") if hasattr(data_class, "gamma") and data_class.gamma == 0: - raise ValueError("Gamma cant be zero") + raise ValueError("Gamma can't be zero") return data_class - def _classical_moment(self, n: int, params: dict) -> tuple[float, float]: - """ - Compute n-th moment of classical NMM - - Args: - n (): Moment ordinal - params (): Parameters of integration algorithm - - Returns: moment approximation and error tolerance - - """ + def compute_moment(self, n: int, params: dict) -> Tuple[float, float]: mixture_moment = 0 error_tolerance = 0 - 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 = quad(lambda u: self.params.distribution.ppf(u) ** (k - l), 0, 1, **params) - error_tolerance += (self.params.beta ** (k - l)) * mixing_moment[1] - mixture_moment += coefficient * (self.params.alpha ** (n - k)) * mixing_moment[0] * norm.moment(l) + if self.mixture_form == "classical": + 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 = quad(lambda u: self.params.distribution.ppf(u) ** (k - l), 0, 1, **params) + error_tolerance += (self.params.beta ** (k - l)) * mixing_moment[1] + mixture_moment += coefficient * (self.params.alpha ** (n - k)) * mixing_moment[0] * norm.moment(l) + else: + for k in range(0, n + 1): + coefficient = binom(n, n - k) * (self.params.sigma ** k) + mixing_moment = quad(lambda u: self.params.distribution.ppf(u) ** (n - k), 0, 1, **params) + error_tolerance += mixing_moment[1] + mixture_moment += coefficient * mixing_moment[0] * norm.moment(k) return mixture_moment, error_tolerance - def _canonical_moment(self, n: int, params: dict) -> tuple[float, float]: - """ - Compute n-th moment of canonical NMM - - Args: - n (): Moment ordinal - params (): Parameters of integration algorithm - - Returns: moment approximation and error tolerance - - """ - mixture_moment = 0 - error_tolerance = 0 - for k in range(0, n + 1): - coefficient = binom(n, n - k) * (self.params.sigma**k) - mixing_moment = quad(lambda u: self.params.distribution.ppf(u) ** (n - k), 0, 1, **params) - error_tolerance += mixing_moment[1] - mixture_moment += coefficient * mixing_moment[0] * norm.moment(k) - return mixture_moment, error_tolerance - - def compute_moment(self, n: int, params: dict) -> tuple[float, float]: - """ - Compute n-th moment of NMM - - Args: - n (): Moment ordinal - params (): Parameters of integration algorithm - - Returns: moment approximation and error tolerance - - """ - if isinstance(self.params, _NMMClassicDataCollector): - return self._classical_moment(n, params) - return self._canonical_moment(n, params) - - def _canonical_compute_cdf(self, x: float, params: dict) -> tuple[float, float]: - """ - Equation for canonical cdf - Args: - x (): point - params (): parameters of RQMC algorithm - - Returns: computed cdf and error tolerance - - """ - rqmc = RQMC(lambda u: norm.cdf((x - self.params.distribution.ppf(u)) / np.abs(self.params.sigma)), **params) - return rqmc() - - def _classical_compute_cdf(self, x: float, params: dict) -> tuple[float, float]: - """ - Equation for classic cdf - Args: - x (): point - params (): parameters of RQMC algorithm - - Returns: computed cdf and error tolerance - - """ - rqmc = RQMC( - lambda u: norm.cdf( - (x - self.params.alpha - self.params.beta * self.params.distribution.ppf(u)) / np.abs(self.params.gamma) - ), - **params - ) + def compute_moments(self, values: List[int], params: dict) -> List[Tuple[float, float]]: + return [self.compute_moment(n, params) for n in values] + + def compute_cdf(self, x: float, params: dict) -> Tuple[float, float]: + if self.mixture_form == "classical": + rqmc = RQMC( + lambda u: norm.cdf((x - self.params.alpha - self.params.beta * self.params.distribution.ppf(u)) / np.abs(self.params.gamma)), + **params, + ) + else: + rqmc = RQMC( + lambda u: norm.cdf((x - self.params.distribution.ppf(u)) / np.abs(self.params.sigma)), + **params, + ) return rqmc() - def compute_cdf(self, x: float, params: dict) -> tuple[float, float]: - """ - Choose equation for cdf estimation depends on Mixture form - Args: - x (): point - params (): parameters of RQMC algorithm - - Returns: Computed pdf and error tolerance - - """ - if isinstance(self.params, _NMMCanonicalDataCollector): - return self._canonical_compute_cdf(x, params) - return self._classical_compute_cdf(x, params) - - def _canonical_compute_pdf(self, x: float, params: dict) -> tuple[float, float]: - """ - Equation for canonical pdf - Args: - x (): point - params (): parameters of RQMC algorithm - - Returns: computed pdf and error tolerance - - """ - rqmc = RQMC( - lambda u: (1 / np.abs(self.params.sigma)) - * norm.pdf((x - self.params.distribution.ppf(u)) / np.abs(self.params.sigma)), - **params - ) + def compute_cdfs(self, values: List[float], params: dict) -> List[Tuple[float, float]]: + return [self.compute_cdf(x, params) for x in values] + + def compute_pdf(self, x: float, params: dict) -> Tuple[float, float]: + if self.mixture_form == "classical": + rqmc = RQMC( + 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, + ) + else: + rqmc = RQMC( + lambda u: (1 / np.abs(self.params.sigma)) * norm.pdf((x - self.params.distribution.ppf(u)) / np.abs(self.params.sigma)), + **params, + ) return rqmc() - def _classical_compute_pdf(self, x: float, params: dict) -> tuple[float, float]: - """ - Equation for classic pdf - Args: - x (): point - params (): parameters of RQMC algorithm - - Returns: computed pdf and error tolerance - - """ - rqmc = RQMC( - 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 - ) + def compute_pdfs(self, values: List[float], params: dict) -> List[Tuple[float, float]]: + return [self.compute_pdf(x, params) for x in values] + + def compute_logpdf(self, x: float, params: dict) -> Tuple[float, float]: + if self.mixture_form == "classical": + rqmc = LogRQMC( + lambda u: ( + np.log(1 / np.abs(self.params.gamma)) + + norm.logpdf((x - self.params.alpha - self.params.beta * self.params.distribution.ppf(u)) / np.abs(self.params.gamma)) + ), + **params, + ) + else: + rqmc = LogRQMC( + lambda u: np.log(1 / np.abs(self.params.sigma)) + norm.logpdf((x - self.params.distribution.ppf(u)) / np.abs(self.params.sigma)), + **params, + ) return rqmc() - def compute_pdf(self, x: float, params: dict) -> tuple[float, float]: - """ - Choose equation for pdf estimation depends on Mixture form - Args: - x (): point - params (): parameters of RQMC algorithm - - Returns: Computed pdf and error tolerance - - """ - if isinstance(self.params, _NMMCanonicalDataCollector): - return self._canonical_compute_pdf(x, params) - return self._classical_compute_pdf(x, params) - - def _classical_compute_log_pdf(self, x: float, params: dict) -> tuple[float, float]: - """ - Equation for classic log pdf - Args: - x (): point - params (): parameters of LogRQMC algorithm - - Returns: computed log pdf and error tolerance - - """ - rqmc = LogRQMC( - lambda u: np.log(1 / np.abs(self.params.gamma)) - + norm.logpdf( - (x - self.params.alpha - self.params.beta * self.params.distribution.ppf(u)) / np.abs(self.params.gamma) - ), - **params - ) - return rqmc() - - def _canonical_compute_log_pdf(self, x: float, params: dict) -> tuple[float, float]: - """ - Equation for canonical log pdf - Args: - x (): point - params (): parameters of LogRQMC algorithm - - Returns: computed log pdf and error tolerance - - """ - rqmc = LogRQMC( - lambda u: np.log(1 / np.abs(self.params.sigma)) - + norm.logpdf((x - self.params.distribution.ppf(u)) / np.abs(self.params.sigma)), - **params - ) - return rqmc() - - def compute_logpdf(self, x: float, params: dict) -> tuple[float, float]: - """ - Choose equation for log pdf estimation depends on Mixture form - Args: - x (): point - params (): parameters of LogRQMC algorithm - - Returns: Computed log pdf and error tolerance - - """ - if isinstance(self.params, _NMMCanonicalDataCollector): - return self._canonical_compute_log_pdf(x, params) - return self._classical_compute_log_pdf(x, params) + def compute_log_pdfs(self, values: List[float], params: dict) -> List[Tuple[float, float]]: + return [self.compute_logpdf(x, params) for x in values] diff --git a/src/mixtures/nmv_mixture.py b/src/mixtures/nmv_mixture.py index c090ff3..e5b1f73 100644 --- a/src/mixtures/nmv_mixture.py +++ b/src/mixtures/nmv_mixture.py @@ -40,152 +40,110 @@ class NormalMeanVarianceMixtures(AbstractMixtures): def __init__(self, mixture_form: str, **kwargs: Any) -> None: super().__init__(mixture_form, **kwargs) - def _classical_moment(self, n: int, params: dict) -> tuple[float, float]: - """ - Compute n-th moment of classical NMM - - Args: - n (): Moment ordinal - params (): Parameters of integration algorithm - - Returns: moment approximation and error tolerance - - """ - + def compute_moment(self, n: int, params: dict) -> tuple[float, float]: def integral_func(u: float) -> float: result = 0 for k in range(0, n + 1): for l in range(0, k + 1): - result += ( - binom(n, n - k) - * binom(k, k - l) - * (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) - ) + if self.mixture_form == "classical": + result += ( + binom(n, n - k) + * binom(k, k - l) + * (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.nu ** (k - l)) + * self.params.distribution.ppf(u) ** (k - l / 2) + * (self.params.alpha ** (n - k)) + * norm.moment(l) + ) return result - rqmc = RQMC(lambda u: integral_func(u), **params) + rqmc = RQMC(integral_func) return rqmc() - def _canonical_moment(self, n: int, params: dict) -> tuple[float, float]: - """ - Compute n-th moment of classical NMM - Args: - n (): Moment ordinal - params (): Parameters of integration algorithm + def compute_cdf(self, x: float, params: dict) -> tuple[float, float]: + def inner_func(u: float) -> float: + ppf = 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) - Returns: moment approximation and error tolerance + rqmc = RQMC(inner_func) + return rqmc() - """ - def integral_func(u: float) -> float: - result = 0 - for k in range(0, n + 1): - for l in range(0, k + 1): - result += ( - binom(n, n - k) - * binom(k, k - l) - * (self.params.nu ** (k - l)) - * self.params.distribution.ppf(u) ** (k - l / 2) - * (self.params.alpha ** (n - k)) - * norm.moment(l) + def compute_pdf(self, x: float, params: dict) -> tuple[float, float]: + def inner_func(u: float) -> float: + ppf = 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) ) - return result + ) + else: + return ( + 1 + / np.sqrt(2 * np.pi * ppf) + * np.exp(-((x - self.params.alpha) ** 2 + self.params.mu ** 2 * ppf ** 2) / (2 * ppf)) + ) - rqmc = RQMC(lambda u: integral_func(u), **params) - return rqmc() + rqmc_res = RQMC(inner_func)() + if self.mixture_form == "classical": + val = np.exp(self.params.beta * (x - self.params.alpha) / self.params.gamma ** 2) * rqmc_res[0] + else: + val = np.exp(self.params.mu * (x - self.params.alpha)) * rqmc_res[0] - def compute_moment(self, n: int, params: dict) -> tuple[float, float]: - if isinstance(self.params, _NMVMClassicDataCollector): - return self._classical_moment(n, params) - return self._canonical_moment(n, params) - - def _classical_cdf(self, x: float, params: dict) -> tuple[float, float]: - def _inner_func(u: float) -> float: - ppf = lru_cache()(self.params.distribution.ppf)(u) - point = (x - self.params.alpha) / (np.sqrt(ppf) * self.params.gamma) - ( - self.params.beta / self.params.gamma * np.sqrt(ppf) - ) - return norm.cdf(point) + return val, rqmc_res[1] - rqmc = RQMC(lambda u: _inner_func(u), **params) - return rqmc() - def _canonical_cdf(self, x: float, params: dict) -> tuple[float, float]: - def _inner_func(u: float) -> float: + def compute_logpdf(self, x: float, params: dict) -> tuple[float, float]: + def inner_func(u: float) -> float: ppf = self.params.distribution.ppf(u) - point = (x - self.params.alpha) / (np.sqrt(ppf)) - (self.params.mu * np.sqrt(ppf)) - return norm.cdf(point) + 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) - rqmc = RQMC(lambda u: _inner_func(u), **params) - return rqmc() + rqmc_res = LogRQMC(inner_func)() + if self.mixture_form == "classical": + val = self.params.beta * (x - self.params.alpha) / self.params.gamma ** 2 + rqmc_res[0] + else: + val = self.params.mu * (x - self.params.alpha) + rqmc_res[0] - def compute_cdf(self, x: float, params: dict) -> tuple[float, float]: - if isinstance(self.params, _NMVMClassicDataCollector): - return self._classical_cdf(x, params) - return self._canonical_cdf(x, params) + return val, rqmc_res[1] - def _classical_pdf(self, x: float, params: dict) -> tuple[float, float]: - def _inner_func(u: float) -> float: - ppf = self.params.distribution.ppf(u) - 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) - ) - ) + def compute_moments(self, values: list[int], params: dict) -> list[tuple[float, float]]: + return [self.compute_moment(n,params) for n in values] - rqmc = RQMC(lambda u: _inner_func(u), **params)() - return np.exp(self.params.beta * (x - self.params.alpha) / self.params.gamma**2) * rqmc[0], rqmc[1] - def _canonical_pdf(self, x: float, params: dict) -> tuple[float, float]: - def _inner_func(u: float) -> float: - ppf = self.params.distribution.ppf(u) - return ( - 1 - / np.sqrt(2 * np.pi * ppf) - * np.exp(-((x - self.params.alpha) ** 2 + self.params.mu**2 * ppf**2) / (2 * ppf)) - ) - - rqmc = RQMC(lambda u: _inner_func(u), **params) - res = rqmc() - return np.exp(self.params.mu * (x - self.params.alpha)) * res[0], res[1] - - def _classical_log_pdf(self, x: float, params: dict) -> tuple[float, float]: - def _inner_func(u: float) -> float: - ppf = self.params.distribution.ppf(u) - 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) + def compute_cdfs(self, xs: list[float], params: dict) -> list[tuple[float, float]]: + return [self.compute_cdf(x,params) for x in xs] - rqmc = LogRQMC(lambda u: _inner_func(u), **params) - return rqmc() - def _canonical_log_pdf(self, x: float, params: dict) -> tuple[float, float]: - def _inner_func(u: float) -> float: - ppf = self.params.distribution.ppf(u) - return -((x - self.params.alpha) ** 2 + ppf**2 * self.params.mu**2 + ppf * np.log(2 * np.pi * ppf)) / ( - 2 * ppf - ) + def compute_pdfs(self, xs: list[float], params: dict) -> list[tuple[float, float]]: + return [self.compute_pdf(x,params) for x in xs] - rqmc = LogRQMC(lambda u: _inner_func(u), **params) - return rqmc() - def compute_pdf(self, x: float, params: dict) -> tuple[float, float]: - if isinstance(self.params, _NMVMClassicDataCollector): - return self._classical_pdf(x, params) - return self._canonical_pdf(x, params) - - def compute_logpdf(self, x: float, params: dict) -> tuple[Any, float]: - if isinstance(self.params, _NMVMClassicDataCollector): - int_result = self._classical_log_pdf(x, params) - return self.params.beta * (x - self.params.alpha) / self.params.gamma**2 + int_result[0], int_result[1] - int_result = self._canonical_log_pdf(x, params) - return self.params.mu * (x - self.params.alpha) + int_result[0], int_result[1] + def compute_logpdfs(self, xs: list[float], params: dict) -> list[tuple[float, float]]: + return [self.compute_logpdf(x,params) for x in xs] diff --git a/src/mixtures/nv_mixture.py b/src/mixtures/nv_mixture.py index 7905e6e..297980e 100644 --- a/src/mixtures/nv_mixture.py +++ b/src/mixtures/nv_mixture.py @@ -32,62 +32,71 @@ class _NVMCanonicalDataCollector: class NormalVarianceMixtures(AbstractMixtures): - _classical_collector = _NVMClassicDataCollector _canonical_collector = _NVMCanonicalDataCollector def __init__(self, mixture_form: str, **kwargs: Any) -> None: super().__init__(mixture_form, **kwargs) - def compute_moment(self, n: int, params: dict) -> tuple[float, float]: - """ - Compute n-th moment of NVM - Args: - n (): Moment ordinal - params (): Parameters of integration algorithm - Returns: moment approximation and error tolerance - """ - gamma = self.params.gamma if isinstance(self.params, _NVMClassicDataCollector) else 1 - - def integrate_func(u: float) -> float: + def compute_moment(self, n: int, rqmc_params: dict[str, Any]) -> tuple[float, float]: + gamma = getattr(self.params, 'gamma', 1) + + def integrand(u: float) -> float: return sum( - [ - binom(n, k) - * (gamma**k) - * (self.params.alpha ** (n - k)) - * (self.params.distribution.ppf(u) ** (k / 2)) - * norm.moment(k) - for k in range(0, n + 1) - ] + binom(n, k) + * (gamma ** k) + * (self.params.alpha ** (n - k)) + * (self.params.distribution.ppf(u) ** (k / 2)) + * norm.moment(k) + for k in range(n + 1) ) - result = RQMC(integrate_func, **params)() - return result + return RQMC(integrand, **rqmc_params)() + + def compute_moments(self, ns: list[int], rqmc_params: dict[str, Any]) -> list[tuple[float, float]]: + return [self.compute_moment(n, rqmc_params) for n in ns] + + def compute_cdf(self, x: float, rqmc_params: dict[str, Any]) -> 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))) + + return RQMC(integrand, **rqmc_params)() + + def compute_cdfs(self, xs: list[float], rqmc_params: dict[str, Any]) -> list[tuple[float, float]]: + return [self.compute_cdf(x, rqmc_params) for x in xs] + + def compute_pdf(self, x: float, rqmc_params: dict[str, Any]) -> 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) + + return RQMC(integrand, **rqmc_params)() + + def compute_pdfs(self, xs: list[float], rqmc_params: dict[str, Any]) -> list[tuple[float, float]]: + return [self.compute_pdf(x, rqmc_params) for x in xs] + + def compute_logpdf(self, x: float, rqmc_params: dict[str, Any]) -> 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) + + return LogRQMC(integrand, **rqmc_params)() - def compute_cdf(self, x: float, params: dict) -> tuple[float, float]: - parametric_norm = norm(0, self.params.gamma if isinstance(self.params, _NVMClassicDataCollector) else 1) - rqmc = RQMC( - lambda u: parametric_norm.cdf((x - self.params.alpha) / np.sqrt(self.params.distribution.ppf(u))), **params - ) - return rqmc() + def compute_logpdfs(self, xs: list[float], rqmc_params: dict[str, Any]) -> list[tuple[float, float]]: + return [self.compute_logpdf(x, rqmc_params) for x in xs] @lru_cache() def _integrand_func(self, u: float, d: float, gamma: float) -> float: ppf = self.params.distribution.ppf(u) - return (1 / np.sqrt(np.pi * 2 * ppf * np.abs(gamma**2))) * np.exp(-1 * d / (2 * ppf)) + return (1 / np.sqrt(np.pi * 2 * ppf * np.abs(gamma ** 2))) * np.exp(-d / (2 * ppf)) - def _log_integrand_func(self, u: float, d: float, gamma: float | int | np.int64) -> float: + def _log_integrand_func(self, u: float, d: float, gamma: float) -> float: ppf = self.params.distribution.ppf(u) - return -(ppf * np.log(np.pi * 2 * ppf * gamma**2) + d) / (2 * ppf) - - def compute_pdf(self, x: float, params: dict) -> tuple[float, float]: - gamma = self.params.gamma if isinstance(self.params, _NVMClassicDataCollector) else 1 - d = (x - self.params.alpha) ** 2 / gamma**2 - rqmc = RQMC(lambda u: self._integrand_func(u, d, gamma), **params) - return rqmc() - - def compute_logpdf(self, x: float, params: dict) -> tuple[float, float]: - gamma = self.params.gamma if isinstance(self.params, _NVMClassicDataCollector) else 1 - d = (x - self.params.alpha) ** 2 / gamma**2 - log_rqmc = LogRQMC(lambda u: self._log_integrand_func(u, d, gamma), **params) - return log_rqmc() + return -(ppf * np.log(np.pi * 2 * ppf * gamma ** 2) + d) / (2 * ppf)