diff --git a/src/pysatl_core/__init__.py b/src/pysatl_core/__init__.py index 751fd03..2ac393b 100644 --- a/src/pysatl_core/__init__.py +++ b/src/pysatl_core/__init__.py @@ -14,6 +14,8 @@ from .distributions import __all__ as _distr_all from .families import * from .families import __all__ as _family_all +from .transformations import * +from .transformations import __all__ as _transformations_all from .types import * from .types import __all__ as _types_all @@ -23,8 +25,10 @@ *_distr_all, *_family_all, *_types_all, + *_transformations_all, ] del _distr_all del _family_all del _types_all +del _transformations_all diff --git a/src/pysatl_core/distributions/distribution.py b/src/pysatl_core/distributions/distribution.py index d9c6bc9..867cd8c 100644 --- a/src/pysatl_core/distributions/distribution.py +++ b/src/pysatl_core/distributions/distribution.py @@ -74,6 +74,24 @@ def computation_strategy(self) -> ComputationStrategy: ... @property def support(self) -> Support | None: ... + def _new_computation_strategy( + self, computation_strategy: ComputationStrategy | None | object = _KEEP + ) -> ComputationStrategy | None: + return ( + self.computation_strategy + if computation_strategy is _KEEP + else cast(ComputationStrategy | None, computation_strategy) + ) + + def _new_sampling_strategy( + self, sampling_strategy: SamplingStrategy | None | object = _KEEP + ) -> SamplingStrategy | None: + return ( + self.sampling_strategy + if sampling_strategy is _KEEP + else cast(SamplingStrategy | None, sampling_strategy) + ) + def _clone_with_strategies( self, *, diff --git a/src/pysatl_core/families/distribution.py b/src/pysatl_core/families/distribution.py index 45c77c4..bf1978c 100644 --- a/src/pysatl_core/families/distribution.py +++ b/src/pysatl_core/families/distribution.py @@ -197,25 +197,13 @@ def _clone_with_strategies( computation_strategy: ComputationStrategy | None | object = _KEEP, ) -> ParametricFamilyDistribution: """Return a copy of this distribution with updated strategies.""" - new_sampling: SamplingStrategy | None = ( - self._sampling_strategy - if sampling_strategy is _KEEP - else cast(SamplingStrategy | None, sampling_strategy) - ) - - new_computation: ComputationStrategy | None = ( - self._computation_strategy - if computation_strategy is _KEEP - else cast(ComputationStrategy | None, computation_strategy) - ) - return ParametricFamilyDistribution( family_name=self._family_name, distribution_type=self._distribution_type, parametrization=self._parametrization, support=self._support, - sampling_strategy=new_sampling, - computation_strategy=new_computation, + sampling_strategy=self._new_sampling_strategy(sampling_strategy), + computation_strategy=self._new_computation_strategy(computation_strategy), ) @property diff --git a/src/pysatl_core/transformations/__init__.py b/src/pysatl_core/transformations/__init__.py new file mode 100644 index 0000000..c1424c2 --- /dev/null +++ b/src/pysatl_core/transformations/__init__.py @@ -0,0 +1,35 @@ +""" +Transformations framework for derived probability distributions. + +This package provides the base primitives for constructing distributions +obtained from other distributions, together with approximation interfaces +and concrete transformation implementations. +""" + +__author__ = "Leonid Elkin, Mikhail Mikhailov, Fedor Myznikov" +__copyright__ = "Copyright (c) 2025 PySATL project" +__license__ = "SPDX-License-Identifier: MIT" + +from .approximations import * +from .approximations import __all__ as _approximations_all +from .distribution import * +from .distribution import __all__ as _distribution_all +from .operations import * +from .operations import __all__ as _operations_all +from .transformation_method import * +from .transformation_method import __all__ as _methods_all + +__all__ = [ + *_approximations_all, + *_distribution_all, + *_operations_all, + *_methods_all, +] + +del _approximations_all + +del _distribution_all + +del _operations_all + +del _methods_all diff --git a/src/pysatl_core/transformations/approximations/__init__.py b/src/pysatl_core/transformations/approximations/__init__.py new file mode 100644 index 0000000..1bcf7b3 --- /dev/null +++ b/src/pysatl_core/transformations/approximations/__init__.py @@ -0,0 +1,25 @@ +""" +Approximation utilities for transformed distributions. + +This subpackage contains approximation interfaces and concrete +approximators that can materialize analytical characteristics for +complex transformation trees. +""" + +__author__ = "Leonid Elkin" +__copyright__ = "Copyright (c) 2025 PySATL project" +__license__ = "SPDX-License-Identifier: MIT" + +from .approximation import * +from .approximation import __all__ as _approximation_all +from .chebyshev import * +from .chebyshev import __all__ as _chebyshev_all + +__all__ = [ + *_approximation_all, + *_chebyshev_all, +] + +del _approximation_all + +del _chebyshev_all diff --git a/src/pysatl_core/transformations/approximations/approximation.py b/src/pysatl_core/transformations/approximations/approximation.py new file mode 100644 index 0000000..9f30ffb --- /dev/null +++ b/src/pysatl_core/transformations/approximations/approximation.py @@ -0,0 +1,59 @@ +""" +Approximation interfaces for derived distributions. + +This module intentionally keeps the approximation layer minimal. +At this stage the public abstraction is a single protocol describing +objects that can materialize an :class:`ApproximatedDistribution` +from a :class:`DerivedDistribution`. +""" + +from __future__ import annotations + +__author__ = "Leonid Elkin" +__copyright__ = "Copyright (c) 2025 PySATL project" +__license__ = "SPDX-License-Identifier: MIT" + +from typing import TYPE_CHECKING, Any, Protocol + +if TYPE_CHECKING: + from pysatl_core.transformations.distribution import ( + ApproximatedDistribution, + DerivedDistribution, + ) + + +class DistributionApproximator(Protocol): + """ + Protocol for objects approximating a derived distribution. + + Implementations may use interpolation, tabulation, polynomial + approximation, or any other technique, as long as they return a new + derived distribution with materialized analytical computations. + """ + + def approximate( + self, + distribution: DerivedDistribution, + **options: Any, + ) -> ApproximatedDistribution: + """ + Build an approximated distribution. + + Parameters + ---------- + distribution : DerivedDistribution + Distribution to approximate. + **options : Any + Extra approximation options. + + Returns + ------- + ApproximatedDistribution + Materialized approximation of the input distribution. + """ + ... + + +__all__ = [ + "DistributionApproximator", +] diff --git a/src/pysatl_core/transformations/approximations/chebyshev.py b/src/pysatl_core/transformations/approximations/chebyshev.py new file mode 100644 index 0000000..e6f92f6 --- /dev/null +++ b/src/pysatl_core/transformations/approximations/chebyshev.py @@ -0,0 +1,226 @@ +""" +Chebyshev-based approximation for transformed distributions. + +This module provides a simple example approximator that materializes chosen +univariate continuous characteristics with Chebyshev polynomials. +The implementation intentionally focuses on the architectural integration +rather than on exhaustive numerical guarantees. +""" + +from __future__ import annotations + +__author__ = "Leonid Elkin" +__copyright__ = "Copyright (c) 2025 PySATL project" +__license__ = "SPDX-License-Identifier: MIT" + +from collections.abc import Mapping, Sequence +from typing import Any, cast + +import numpy as np +from numpy.polynomial import Chebyshev + +from pysatl_core.distributions.computation import AnalyticalComputation +from pysatl_core.distributions.support import ContinuousSupport +from pysatl_core.transformations.approximations.approximation import DistributionApproximator +from pysatl_core.transformations.distribution import ApproximatedDistribution, DerivedDistribution +from pysatl_core.types import ( + ApproximationName, + CharacteristicName, + GenericCharacteristicName, + Kind, + NumericArray, +) + + +class ChebyshevApproximator(DistributionApproximator): + """ + Approximate selected characteristics with Chebyshev polynomials. + + Parameters + ---------- + degree : int, default=32 + Polynomial degree used by the approximation. + sample_size : int, optional + Number of grid points used to fit the polynomial. If omitted, a value + derived from ``degree`` is used. + characteristics : Sequence[GenericCharacteristicName] or None, optional + Characteristics to materialize. If omitted, all direct analytical + computations of the source derived distribution are approximated. + domains : Mapping[GenericCharacteristicName, tuple[float, float]] or None, optional + Explicit approximation domains for selected characteristics. If omitted, + the domain is inferred from the transformed distribution support for + ``pdf`` and ``cdf`` and from ``[0, 1]`` for ``ppf``. + + Notes + ----- + This implementation currently targets univariate continuous + characteristics. It is meant as the first example approximator and may be + extended later with stricter error control and domain management. + """ + + def __init__( + self, + *, + degree: int = 32, + sample_size: int | None = None, + characteristics: Sequence[GenericCharacteristicName] | None = None, + domains: Mapping[GenericCharacteristicName, tuple[float, float]] | None = None, + ) -> None: + if degree < 0: + raise ValueError("degree must be non-negative.") + + self._degree = degree + self._sample_size = sample_size + self._characteristics = tuple(characteristics) if characteristics is not None else None + self._domains = dict(domains) if domains is not None else {} + + def approximate( + self, + distribution: DerivedDistribution, + **options: Any, + ) -> ApproximatedDistribution: + """ + Build an approximated distribution using Chebyshev polynomials. + + Parameters + ---------- + distribution : DerivedDistribution + Distribution to approximate. + **options : Any + Extra options forwarded to queried source characteristics. + + Returns + ------- + ApproximatedDistribution + Distribution whose analytical computations are represented by + Chebyshev approximations. + """ + self._validate_distribution(distribution) + + characteristics = self._select_characteristics(distribution) + analytical_computations: dict[ + GenericCharacteristicName, AnalyticalComputation[Any, Any] + ] = {} + + for characteristic_name in characteristics: + domain = self._resolve_domain(distribution, characteristic_name) + analytical_computations[characteristic_name] = self._build_computation( + distribution=distribution, + characteristic_name=characteristic_name, + domain=domain, + **options, + ) + + return ApproximatedDistribution( + source_distribution=distribution, + approximation_name=ApproximationName.CHEBYSHEV, + distribution_type=distribution.distribution_type, + bases={}, + analytical_computations=analytical_computations, + support=distribution.support, + sampling_strategy=distribution.sampling_strategy, + computation_strategy=distribution.computation_strategy, + ) + + def _validate_distribution(self, distribution: DerivedDistribution) -> None: + """ + Validate that the distribution is supported by the approximator. + """ + distribution_type = distribution.distribution_type + kind = getattr(distribution_type, "kind", None) + dimension = getattr(distribution_type, "dimension", None) + + if kind != Kind.CONTINUOUS or dimension != 1: + raise TypeError( + "ChebyshevApproximator currently supports only univariate continuous distributions." + ) + + def _select_characteristics( + self, + distribution: DerivedDistribution, + ) -> tuple[GenericCharacteristicName, ...]: + """ + Determine which characteristics should be approximated. + """ + if self._characteristics is not None: + selected = self._characteristics + else: + selected = tuple(distribution.analytical_computations) + + if not selected: + raise ValueError( + "No characteristics were selected for approximation. " + "Pass them explicitly or provide direct analytical computations." + ) + + return tuple(selected) + + def _resolve_domain( + self, + distribution: DerivedDistribution, + characteristic_name: GenericCharacteristicName, + ) -> tuple[float, float]: + """ + Resolve the approximation domain for a characteristic. + """ + if characteristic_name in self._domains: + return self._domains[characteristic_name] + + if characteristic_name == CharacteristicName.PPF: + return 0.0, 1.0 + + support = distribution.support + if ( + isinstance(support, ContinuousSupport) + and np.isfinite(support.left) + and np.isfinite(support.right) + ): + return float(support.left), float(support.right) + + raise ValueError( + "Could not infer a finite approximation domain for the requested characteristic." + ) + + def _build_computation( + self, + *, + distribution: DerivedDistribution, + characteristic_name: GenericCharacteristicName, + domain: tuple[float, float], + **options: Any, + ) -> AnalyticalComputation[Any, Any]: + """ + Approximate a single characteristic with a Chebyshev polynomial. + """ + left, right = domain + if left >= right: + raise ValueError("Approximation domain must satisfy left < right.") + + sample_size = self._sample_size or max(2 * self._degree + 1, 33) + method = distribution.query_method(characteristic_name, **options) + + grid = np.linspace(left, right, sample_size, dtype=float) + values = np.asarray(method(grid), dtype=float) + if values.shape != grid.shape: + raise ValueError( + "ChebyshevApproximator expects one-dimensional NumPy semantics " + "for approximated methods." + ) + + polynomial = Chebyshev.fit(grid, values, deg=self._degree, domain=[left, right]) + + def _func(data: Any, /, **_: Any) -> Any: + array = np.asarray(data, dtype=float) + approximated = np.asarray(polynomial(array), dtype=float) + + if np.ndim(array) == 0: + return float(approximated) + + return cast(NumericArray, approximated) + + return AnalyticalComputation(target=characteristic_name, func=_func) + + +__all__ = [ + "ChebyshevApproximator", +] diff --git a/src/pysatl_core/transformations/distribution.py b/src/pysatl_core/transformations/distribution.py new file mode 100644 index 0000000..dc03afb --- /dev/null +++ b/src/pysatl_core/transformations/distribution.py @@ -0,0 +1,325 @@ +""" +Base classes for transformed distributions. + +This module introduces the first architectural layer for derived +probability distributions produced by transformations. The goal is to +keep them fully compatible with the existing :class:`Distribution` +protocol and computation graph while still preserving transformation +metadata. +""" + +from __future__ import annotations + +__author__ = "Leonid Elkin" +__copyright__ = "Copyright (c) 2025 PySATL project" +__license__ = "SPDX-License-Identifier: MIT" + +from collections.abc import Mapping +from numbers import Real +from types import NotImplementedType +from typing import TYPE_CHECKING, Any + +from pysatl_core.distributions.computation import AnalyticalComputation +from pysatl_core.distributions.distribution import _KEEP, Distribution +from pysatl_core.distributions.strategies import ( + ComputationStrategy, + DefaultComputationStrategy, + DefaultSamplingUnivariateStrategy, + SamplingStrategy, +) +from pysatl_core.types import ( + ApproximationName, + DistributionType, + GenericCharacteristicName, + ParentRole, + TransformationName, +) + +if TYPE_CHECKING: + from pysatl_core.distributions.support import Support + from pysatl_core.transformations.approximations.approximation import DistributionApproximator + from pysatl_core.transformations.operations.affine import AffineDistribution + + +class DerivedDistribution(Distribution): + """ + Base class for distributions obtained from one or more parents. + + Parameters + ---------- + distribution_type : DistributionType + Type descriptor of the derived distribution. + bases : Mapping[ParentRole, Distribution] + Parent distributions participating in the transformation. + analytical_computations : Mapping[GenericCharacteristicName, AnalyticalComputation[Any, Any]] + Direct analytical computations attached to the derived distribution. + transformation_name : TransformationName + Logical name of the transformation. + support : Support | None, optional + Support of the transformed distribution. + sampling_strategy : SamplingStrategy | None, optional + Strategy used to generate random samples. If omitted, a default + univariate inverse-transform strategy is installed when the derived + distribution provides an analytical PPF. Otherwise a placeholder + strategy is used. + computation_strategy : ComputationStrategy | None, optional + Strategy used to resolve characteristics. + """ + + def __init__( + self, + *, + distribution_type: DistributionType, + bases: Mapping[ParentRole, Distribution], + analytical_computations: Mapping[ + GenericCharacteristicName, AnalyticalComputation[Any, Any] + ], + transformation_name: TransformationName, + support: Support | None = None, + sampling_strategy: SamplingStrategy | None = None, + computation_strategy: ComputationStrategy | None = None, + ) -> None: + self._distribution_type = distribution_type + self._bases = dict(bases) + self._analytical_computations = dict(analytical_computations) + self._transformation_name = transformation_name + self._support = support + self._sampling_strategy = sampling_strategy or DefaultSamplingUnivariateStrategy() + self._computation_strategy = computation_strategy or DefaultComputationStrategy() + + @property + def distribution_type(self) -> DistributionType: + """Get the type descriptor of the derived distribution.""" + return self._distribution_type + + @property + def bases(self) -> Mapping[ParentRole, Distribution]: + """Get parent distributions grouped by their logical roles.""" + return self._bases + + @property + def transformation_name(self) -> TransformationName: + """Get the logical name of the transformation.""" + return self._transformation_name + + @property + def analytical_computations( + self, + ) -> Mapping[GenericCharacteristicName, AnalyticalComputation[Any, Any]]: + """Get direct analytical computations of the derived distribution.""" + return self._analytical_computations + + @property + def sampling_strategy(self) -> SamplingStrategy: + """Get the sampling strategy.""" + return self._sampling_strategy + + @property + def computation_strategy(self) -> ComputationStrategy: + """Get the characteristic resolution strategy.""" + return self._computation_strategy + + @property + def support(self) -> Support | None: + """Get the support of the derived distribution.""" + return self._support + + def approximate( + self, + approximator: DistributionApproximator, + **options: Any, + ) -> ApproximatedDistribution: + """ + Approximate the current derivation tree. + + Parameters + ---------- + approximator : DistributionApproximator + External approximation object responsible for creating the new + distribution instance. + **options : Any + Extra options forwarded to the approximator. + + Returns + ------- + ApproximatedDistribution + Approximated representation of the current derived distribution. + """ + return approximator.approximate(self, **options) + + def _clone_with_strategies( + self, + *, + sampling_strategy: SamplingStrategy | None | object = _KEEP, + computation_strategy: ComputationStrategy | None | object = _KEEP, + ) -> DerivedDistribution: + """ + Return a copy of the derived distribution with updated strategies. + + Subclasses representing concrete transformations should normally + override this method to preserve their own constructor parameters. + """ + return DerivedDistribution( + distribution_type=self.distribution_type, + bases=self.bases, + analytical_computations=self.analytical_computations, + transformation_name=self.transformation_name, + support=self.support, + sampling_strategy=self._new_sampling_strategy(sampling_strategy), + computation_strategy=self._new_computation_strategy(computation_strategy), + ) + + def _affine(self, scale: float, shift: float) -> AffineDistribution: + """ + Build an affine transformation of the current distribution. + + Parameters + ---------- + scale : float + Multiplicative coefficient. + shift : float + Additive coefficient. + + Returns + ------- + AffineDistribution + Derived distribution representing ``scale * X + shift``. + """ + from pysatl_core.transformations.operations.affine import AffineDistribution + + return AffineDistribution(self, scale=scale, shift=shift) + + def __add__(self, other: object) -> DerivedDistribution | NotImplementedType: + """Return ``self + scalar`` as an affine transformation.""" + if isinstance(other, Real): + return self._affine(scale=1.0, shift=float(other)) + return NotImplemented + + def __radd__(self, other: object) -> DerivedDistribution | NotImplementedType: + """Return ``scalar + self`` as an affine transformation.""" + if isinstance(other, Real): + return self._affine(scale=1.0, shift=float(other)) + return NotImplemented + + def __sub__(self, other: object) -> DerivedDistribution | NotImplementedType: + """Return ``self - scalar`` as an affine transformation.""" + if isinstance(other, Real): + return self._affine(scale=1.0, shift=-float(other)) + return NotImplemented + + def __rsub__(self, other: object) -> DerivedDistribution | NotImplementedType: + """Return ``scalar - self`` as an affine transformation.""" + if isinstance(other, Real): + return self._affine(scale=-1.0, shift=float(other)) + return NotImplemented + + def __mul__(self, other: object) -> DerivedDistribution | NotImplementedType: + """Return ``self * scalar`` as an affine transformation.""" + if isinstance(other, Real): + return self._affine(scale=float(other), shift=0.0) + return NotImplemented + + def __rmul__(self, other: object) -> DerivedDistribution | NotImplementedType: + """Return ``scalar * self`` as an affine transformation.""" + if isinstance(other, Real): + return self._affine(scale=float(other), shift=0.0) + return NotImplemented + + def __truediv__(self, other: object) -> DerivedDistribution | NotImplementedType: + """Return ``self / scalar`` as an affine transformation.""" + if isinstance(other, Real): + divisor = float(other) + if divisor == 0.0: + raise ZeroDivisionError("Cannot divide a distribution by zero.") + return self._affine(scale=1.0 / divisor, shift=0.0) + return NotImplemented + + def __neg__(self) -> DerivedDistribution: + """Return ``-self`` as an affine transformation.""" + return self._affine(scale=-1.0, shift=0.0) + + +class ApproximatedDistribution(DerivedDistribution): + """ + Derived distribution whose analytical computations were materialized by an + external approximator. + + Parameters + ---------- + source_distribution : DerivedDistribution + Original distribution being approximated. + approximation_name : ApproximationName + Name of the approximation procedure. + distribution_type : DistributionType + Type descriptor of the approximated distribution. + bases : Mapping[ParentRole, Distribution] + Parent distributions preserved by the approximated distribution. + analytical_computations : Mapping[GenericCharacteristicName, AnalyticalComputation[Any, Any]] + Materialized analytical computations produced by the approximator. + support : Support | None, optional + Support of the approximated distribution. + sampling_strategy : SamplingStrategy | None, optional + Sampling strategy to expose. + computation_strategy : ComputationStrategy | None, optional + Characteristic resolution strategy. + """ + + def __init__( + self, + *, + source_distribution: DerivedDistribution, + approximation_name: ApproximationName, + distribution_type: DistributionType, + bases: Mapping[ParentRole, Distribution], + analytical_computations: Mapping[ + GenericCharacteristicName, AnalyticalComputation[Any, Any] + ], + support: Support | None = None, + sampling_strategy: SamplingStrategy | None = None, + computation_strategy: ComputationStrategy | None = None, + ) -> None: + super().__init__( + distribution_type=distribution_type, + bases=bases, + analytical_computations=analytical_computations, + transformation_name=TransformationName.APPROXIMATION, + support=support, + sampling_strategy=sampling_strategy, + computation_strategy=computation_strategy, + ) + self._source_distribution = source_distribution + self._approximation_name = approximation_name + + @property + def source_distribution(self) -> DerivedDistribution: + """Get the original non-approximated distribution.""" + return self._source_distribution + + @property + def approximation_name(self) -> ApproximationName: + """Get the name of the approximation procedure.""" + return self._approximation_name + + def _clone_with_strategies( + self, + *, + sampling_strategy: SamplingStrategy | None | object = _KEEP, + computation_strategy: ComputationStrategy | None | object = _KEEP, + ) -> ApproximatedDistribution: + """Return a copy of the approximated distribution with updated strategies.""" + return ApproximatedDistribution( + source_distribution=self._source_distribution, + approximation_name=self.approximation_name, + distribution_type=self.distribution_type, + bases=self.bases, + analytical_computations=self.analytical_computations, + support=self.support, + sampling_strategy=self._new_sampling_strategy(sampling_strategy), + computation_strategy=self._new_computation_strategy(computation_strategy), + ) + + +__all__ = [ + "ApproximatedDistribution", + "DerivedDistribution", +] diff --git a/src/pysatl_core/transformations/operations/__init__.py b/src/pysatl_core/transformations/operations/__init__.py new file mode 100644 index 0000000..d1bca09 --- /dev/null +++ b/src/pysatl_core/transformations/operations/__init__.py @@ -0,0 +1,19 @@ +""" +Concrete distribution transformation operations. + +This subpackage contains concrete transformed-distribution implementations. +At the moment it provides the first affine transformation primitive. +""" + +__author__ = "Leonid Elkin" +__copyright__ = "Copyright (c) 2025 PySATL project" +__license__ = "SPDX-License-Identifier: MIT" + +from .affine import * +from .affine import __all__ as _affine_all + +__all__ = [ + *_affine_all, +] + +del _affine_all diff --git a/src/pysatl_core/transformations/operations/affine.py b/src/pysatl_core/transformations/operations/affine.py new file mode 100644 index 0000000..6389944 --- /dev/null +++ b/src/pysatl_core/transformations/operations/affine.py @@ -0,0 +1,565 @@ +""" +Affine transformation for probability distributions. + +This module implements the transformation ``Y = aX + b``. The +transformation is represented as a derived distribution with analytical +computations built from parent methods resolved through ``query_method()``. +""" + +from __future__ import annotations + +__author__ = "Leonid Elkin" +__copyright__ = "Copyright (c) 2025 PySATL project" +__license__ = "SPDX-License-Identifier: MIT" + +from collections.abc import Mapping +from typing import Any, cast + +import numpy as np + +from pysatl_core.distributions.computation import Method +from pysatl_core.distributions.distribution import Distribution +from pysatl_core.distributions.support import ( + ContinuousSupport, + ExplicitTableDiscreteSupport, + Support, +) +from pysatl_core.transformations.distribution import DerivedDistribution +from pysatl_core.transformations.transformation_method import ( + ResolvedSourceMethods, + SourceRequirements, + TransformationMethod, +) +from pysatl_core.types import ( + CharacteristicName, + ComplexArray, + ComputationFunc, + DistributionType, + GenericCharacteristicName, + Kind, + NumericArray, + ParentRole, + TransformationName, +) + +_BASE_ROLE: ParentRole = "base" + + +class AffineDistribution(DerivedDistribution): + """ + Distribution obtained from the affine transformation ``Y = aX + b``. + + Parameters + ---------- + base_distribution : Distribution + Source distribution being transformed. + scale : float + Multiplicative coefficient ``a``. + shift : float, default=0.0 + Additive coefficient ``b``. + + Notes + ----- + The current implementation focuses on one-dimensional continuous and + discrete distributions. + """ + + def __init__( + self, + base_distribution: Distribution, + *, + scale: float, + shift: float = 0.0, + ) -> None: + if scale == 0.0: + raise ValueError("scale must be non-zero for an affine transformation.") + + self._base_distribution = base_distribution + self._scale = float(scale) + self._shift = float(shift) + distribution_type = self._validate_distribution_type(base_distribution.distribution_type) + bases: dict[ParentRole, Distribution] = {_BASE_ROLE: base_distribution} + + super().__init__( + distribution_type=distribution_type, + bases=bases, + analytical_computations=self._build_analytical_computations( + distribution_type=distribution_type, + bases=bases, + ), + transformation_name=TransformationName.AFFINE, + support=self._transform_support(base_distribution.support), + ) + + @property + def base_distribution(self) -> Distribution: + """Get the source distribution.""" + return self._base_distribution + + @property + def scale(self) -> float: + """Get the multiplicative coefficient ``a``.""" + return self._scale + + @property + def shift(self) -> float: + """Get the additive coefficient ``b``.""" + return self._shift + + def _validate_distribution_type(self, distribution_type: DistributionType) -> DistributionType: + """ + Validate that the affine transformation can be applied. + + Parameters + ---------- + distribution_type : DistributionType + Distribution type descriptor of the parent distribution. + + Returns + ------- + DistributionType + The validated distribution type. + + Raises + ------ + TypeError + If the distribution is not one-dimensional continuous or discrete. + """ + dimension = getattr(distribution_type, "dimension", None) + kind = getattr(distribution_type, "kind", None) + + if dimension != 1: + raise TypeError( + "AffineDistribution currently supports only one-dimensional distributions." + ) + + if kind not in {Kind.CONTINUOUS, Kind.DISCRETE}: + raise TypeError("Unsupported distribution kind for affine transformation.") + + return distribution_type + + def _build_analytical_computations( + self, + *, + distribution_type: DistributionType, + bases: Mapping[ParentRole, Distribution], + ) -> Mapping[GenericCharacteristicName, TransformationMethod[Any, Any]]: + """ + Build analytical computations for the affine transformation. + + Parameters + ---------- + distribution_type : DistributionType + Type descriptor of the transformed distribution. + bases : Mapping[ParentRole, Distribution] + Parent distributions grouped by logical role. + + Returns + ------- + Mapping[GenericCharacteristicName, TransformationMethod[Any, Any]] + Analytical computations exposed by the transformed distribution. + + Raises + ------ + TypeError + If the distribution kind is not supported. + """ + computations: dict[GenericCharacteristicName, TransformationMethod[Any, Any]] = {} + kind = getattr(distribution_type, "kind", None) + + computations[CharacteristicName.CF] = TransformationMethod.from_parents( + target=CharacteristicName.CF, + transformation=TransformationName.AFFINE, + bases=bases, + source_requirements=self._requirements(CharacteristicName.CF), + evaluator=self._make_cf, + ) + computations[CharacteristicName.MEAN] = TransformationMethod.from_parents( + target=CharacteristicName.MEAN, + transformation=TransformationName.AFFINE, + bases=bases, + source_requirements=self._requirements(CharacteristicName.MEAN), + evaluator=self._make_mean, + ) + computations[CharacteristicName.VAR] = TransformationMethod.from_parents( + target=CharacteristicName.VAR, + transformation=TransformationName.AFFINE, + bases=bases, + source_requirements=self._requirements(CharacteristicName.VAR), + evaluator=self._make_var, + ) + computations[CharacteristicName.SKEW] = TransformationMethod.from_parents( + target=CharacteristicName.SKEW, + transformation=TransformationName.AFFINE, + bases=bases, + source_requirements=self._requirements(CharacteristicName.SKEW), + evaluator=self._make_skew, + ) + computations[CharacteristicName.KURT] = TransformationMethod.from_parents( + target=CharacteristicName.KURT, + transformation=TransformationName.AFFINE, + bases=bases, + source_requirements=self._requirements(CharacteristicName.KURT), + evaluator=self._make_kurt, + ) + + if kind == Kind.CONTINUOUS: + computations[CharacteristicName.CDF] = TransformationMethod.from_parents( + target=CharacteristicName.CDF, + transformation=TransformationName.AFFINE, + bases=bases, + source_requirements=self._requirements(CharacteristicName.CDF), + evaluator=self._make_continuous_cdf, + ) + computations[CharacteristicName.PDF] = TransformationMethod.from_parents( + target=CharacteristicName.PDF, + transformation=TransformationName.AFFINE, + bases=bases, + source_requirements=self._requirements(CharacteristicName.PDF), + evaluator=self._make_continuous_pdf, + ) + computations[CharacteristicName.PPF] = TransformationMethod.from_parents( + target=CharacteristicName.PPF, + transformation=TransformationName.AFFINE, + bases=bases, + source_requirements=self._requirements(CharacteristicName.PPF), + evaluator=self._make_continuous_ppf, + ) + return computations + + if kind == Kind.DISCRETE: + computations[CharacteristicName.PMF] = TransformationMethod.from_parents( + target=CharacteristicName.PMF, + transformation=TransformationName.AFFINE, + bases=bases, + source_requirements=self._requirements(CharacteristicName.PMF), + evaluator=self._make_discrete_pmf, + ) + computations[CharacteristicName.CDF] = TransformationMethod.from_parents( + target=CharacteristicName.CDF, + transformation=TransformationName.AFFINE, + bases=bases, + source_requirements=( + self._requirements(CharacteristicName.CDF) + if self.scale > 0.0 + else self._requirements(CharacteristicName.CDF, CharacteristicName.PMF) + ), + evaluator=( + self._make_discrete_cdf + if self.scale > 0.0 + else self._make_discrete_cdf_negative_scale + ), + ) + computations[CharacteristicName.PPF] = TransformationMethod.from_parents( + target=CharacteristicName.PPF, + transformation=TransformationName.AFFINE, + bases=bases, + source_requirements=self._requirements(CharacteristicName.PPF), + evaluator=self._make_discrete_ppf, + ) + return computations + + raise TypeError("Unsupported distribution kind for affine transformation.") + + def _requirements( + self, + *characteristics: GenericCharacteristicName, + ) -> SourceRequirements: + """ + Build source requirements for the base distribution. + + Parameters + ---------- + *characteristics : GenericCharacteristicName + Parent characteristics required to evaluate a transformed one. + + Returns + ------- + SourceRequirements + Requirements grouped by the single base role. + """ + return {_BASE_ROLE: tuple(characteristics)} + + def _make_continuous_cdf( + self, + sources: ResolvedSourceMethods, + ) -> ComputationFunc[NumericArray, NumericArray]: + """ + Build the transformed CDF for a continuous base distribution. + + For ``Y = aX + b`` the formula is + ``F_Y(y) = F_X((y - b) / a)`` for ``a > 0`` and + ``F_Y(y) = 1 - F_X((y - b) / a)`` for ``a < 0``. + """ + base_cdf = cast( + Method[NumericArray, NumericArray], sources[_BASE_ROLE][CharacteristicName.CDF] + ) + + def _cdf(data: NumericArray, **options: Any) -> NumericArray: + values = base_cdf((data - self.shift) / self.scale, **options) + return 1.0 - values if self.scale < 0.0 else values + + return cast(ComputationFunc[NumericArray, NumericArray], _cdf) + + def _make_continuous_pdf( + self, + sources: ResolvedSourceMethods, + ) -> ComputationFunc[NumericArray, NumericArray]: + """ + Build the transformed PDF for a continuous base distribution. + + The affine density transformation is + ``f_Y(y) = f_X((y - b) / a) / |a|``. + """ + base_pdf = cast( + Method[NumericArray, NumericArray], sources[_BASE_ROLE][CharacteristicName.PDF] + ) + + def _pdf(data: NumericArray, **options: Any) -> NumericArray: + return cast( + NumericArray, + base_pdf((data - self.shift) / self.scale, **options) / abs(self.scale), + ) + + return cast(ComputationFunc[NumericArray, NumericArray], _pdf) + + def _make_continuous_ppf( + self, + sources: ResolvedSourceMethods, + ) -> ComputationFunc[NumericArray, NumericArray]: + """ + Build the transformed PPF for a continuous base distribution. + + For positive scale the quantiles are transformed directly. For + negative scale the probabilities are mirrored as ``1 - p``. + """ + base_ppf = cast( + Method[NumericArray, NumericArray], sources[_BASE_ROLE][CharacteristicName.PPF] + ) + + def _ppf(data: NumericArray, **options: Any) -> NumericArray: + probabilities = data if self.scale > 0.0 else 1.0 - data + return self.scale * base_ppf(probabilities, **options) + self.shift + + return cast(ComputationFunc[NumericArray, NumericArray], _ppf) + + def _make_discrete_cdf( + self, + sources: ResolvedSourceMethods, + ) -> ComputationFunc[NumericArray, NumericArray]: + """ + Build the transformed CDF for a discrete base distribution with ``a > 0``. + """ + base_cdf = cast( + Method[NumericArray, NumericArray], sources[_BASE_ROLE][CharacteristicName.CDF] + ) + + def _cdf(data: NumericArray, **options: Any) -> NumericArray: + return base_cdf((data - self.shift) / self.scale, **options) + + return cast(ComputationFunc[NumericArray, NumericArray], _cdf) + + def _make_discrete_cdf_negative_scale( + self, + sources: ResolvedSourceMethods, + ) -> ComputationFunc[NumericArray, NumericArray]: + """ + Build the transformed CDF for a discrete base distribution with ``a < 0``. + + For a decreasing affine map the lower tail of ``Y`` becomes the upper + tail of ``X``. Using both CDF and PMF avoids support-specific logic and + preserves jump values exactly. + """ + base_cdf = cast( + Method[NumericArray, NumericArray], sources[_BASE_ROLE][CharacteristicName.CDF] + ) + base_pmf = cast( + Method[NumericArray, NumericArray], sources[_BASE_ROLE][CharacteristicName.PMF] + ) + + def _cdf(data: NumericArray, **options: Any) -> NumericArray: + x = (data - self.shift) / self.scale + return np.asarray(1.0 - base_cdf(x, **options) + base_pmf(x, **options)) + + return cast(ComputationFunc[NumericArray, NumericArray], _cdf) + + def _make_discrete_pmf( + self, + sources: ResolvedSourceMethods, + ) -> ComputationFunc[NumericArray, NumericArray]: + """ + Build the transformed PMF for a discrete base distribution. + + Since the affine map is bijective for ``a != 0``, the probability mass + is preserved without any Jacobian factor. + """ + base_pmf = cast( + Method[NumericArray, NumericArray], sources[_BASE_ROLE][CharacteristicName.PMF] + ) + + def _pmf(data: NumericArray, **options: Any) -> NumericArray: + return base_pmf((data - self.shift) / self.scale, **options) + + return cast(ComputationFunc[NumericArray, NumericArray], _pmf) + + def _make_discrete_ppf( + self, + sources: ResolvedSourceMethods, + ) -> ComputationFunc[NumericArray, NumericArray]: + """ + Build the transformed PPF for a discrete base distribution. + + For positive scale the quantiles are transformed directly. For + negative scale the lower-tail quantile of ``Y`` corresponds to the + strict upper-tail quantile of ``X``, implemented through + ``nextafter(1 - p, 1)``. + """ + base_ppf = cast( + Method[NumericArray, NumericArray], sources[_BASE_ROLE][CharacteristicName.PPF] + ) + + def _ppf(data: NumericArray, **options: Any) -> NumericArray: + x = data if self.scale > 0.0 else np.nextafter(1.0 - data, 1.0) + return self.scale * base_ppf(x, **options) + self.shift + + return cast(ComputationFunc[NumericArray, NumericArray], _ppf) + + def _make_cf( + self, + sources: ResolvedSourceMethods, + ) -> ComputationFunc[NumericArray, ComplexArray]: + """ + Build the characteristic function of the affine transform. + + The formula is ``phi_Y(t) = exp(i b t) * phi_X(a t)``. + """ + base_cf = cast( + Method[NumericArray, ComplexArray], sources[_BASE_ROLE][CharacteristicName.CF] + ) + + def _cf(data: NumericArray, **options: Any) -> ComplexArray: + return cast( + ComplexArray, np.exp(1j * self.shift * data) * base_cf(self.scale * data, **options) + ) + + return cast(ComputationFunc[NumericArray, ComplexArray], _cf) + + def _make_mean(self, sources: ResolvedSourceMethods) -> ComputationFunc[Any, float]: + """Build the transformed mean.""" + base_mean = cast(Method[Any, float], sources[_BASE_ROLE][CharacteristicName.MEAN]) + + def _mean(**options: Any) -> float: + return self.scale * base_mean(**options) + self.shift + + return _mean + + def _make_var(self, sources: ResolvedSourceMethods) -> ComputationFunc[Any, float]: + """Build the transformed variance.""" + base_var = cast(Method[Any, float], sources[_BASE_ROLE][CharacteristicName.VAR]) + + def _var(**options: Any) -> float: + return self.scale**2 * base_var(**options) + + return _var + + def _make_skew(self, sources: ResolvedSourceMethods) -> ComputationFunc[Any, float]: + """ + Build the transformed skewness. + + Multiplication by a negative constant flips the sign of skewness. + """ + base_skew = cast(Method[Any, float], sources[_BASE_ROLE][CharacteristicName.SKEW]) + + def _skew(**options: Any) -> float: + sign = -1.0 if self.scale < 0.0 else 1.0 + return sign * base_skew(**options) + + return _skew + + def _make_kurt(self, sources: ResolvedSourceMethods) -> ComputationFunc[Any, float]: + """ + Build the transformed kurtosis. + + Kurtosis is invariant under affine transformations with non-zero scale. + """ + base_kurt = cast(Method[Any, float], sources[_BASE_ROLE][CharacteristicName.KURT]) + + def _kurt(**options: Any) -> float: + return base_kurt(**options) + + return _kurt + + def _transform_support(self, support: Support | None) -> Support | None: + """ + Transform the parent support when its structure is known. + + Notes + ----- + Some support types are intentionally left unhandled for now. + In such cases the transformed distribution exposes ``None``. + """ + if support is None: + return None + + if isinstance(support, ContinuousSupport): + return self._transform_continuous_support(support) + + if isinstance(support, ExplicitTableDiscreteSupport): + transformed_points = np.asarray(support.points, dtype=float) * self.scale + self.shift + return ExplicitTableDiscreteSupport(points=transformed_points, assume_sorted=False) + + return None + + def _transform_continuous_support(self, support: ContinuousSupport) -> ContinuousSupport: + """ + Transform a continuous interval support under the affine map. + """ + left = float(self.scale * support.left + self.shift) + right = float(self.scale * support.right + self.shift) + + if self.scale > 0.0: + return ContinuousSupport( + left=left, + right=right, + left_closed=support.left_closed, + right_closed=support.right_closed, + ) + + return ContinuousSupport( + left=right, + right=left, + left_closed=support.right_closed, + right_closed=support.left_closed, + ) + + +def affine( + distribution: Distribution, + *, + scale: float, + shift: float = 0.0, +) -> AffineDistribution: + """ + Apply the affine transformation ``Y = aX + b`` to a distribution. + + Parameters + ---------- + distribution : Distribution + Source distribution. + scale : float + Multiplicative coefficient ``a``. + shift : float, default=0.0 + Additive coefficient ``b``. + + Returns + ------- + AffineDistribution + Derived distribution representing the transformed random variable. + """ + return AffineDistribution(distribution, scale=scale, shift=shift) + + +__all__ = [ + "AffineDistribution", + "affine", +] diff --git a/src/pysatl_core/transformations/transformation_method.py b/src/pysatl_core/transformations/transformation_method.py new file mode 100644 index 0000000..1da956f --- /dev/null +++ b/src/pysatl_core/transformations/transformation_method.py @@ -0,0 +1,122 @@ +""" +Transformation computation primitives. + +This module defines analytical computations produced by distribution +transformations. A transformation method remains compatible with AnalyticalComputation +so that transformed distributions continue to participate in the +existing characteristic graph and computation strategy. +""" + +from __future__ import annotations + +__author__ = "Leonid Elkin" +__copyright__ = "Copyright (c) 2025 PySATL project" +__license__ = "SPDX-License-Identifier: MIT" + +from collections.abc import Callable, Mapping +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any + +from pysatl_core.distributions.computation import AnalyticalComputation, Method +from pysatl_core.types import ( + ComputationFunc, + GenericCharacteristicName, + ParentRole, + TransformationName, +) + +if TYPE_CHECKING: + from pysatl_core.distributions.distribution import Distribution + +type SourceRequirements = dict[ParentRole, tuple[GenericCharacteristicName, ...]] +"""Required parent characteristics grouped by logical parent role.""" + +type ResolvedSourceMethods = dict[ + ParentRole, + dict[GenericCharacteristicName, Method[Any, Any]], +] +"""Resolved parent methods grouped by logical parent role and characteristic.""" + +type TransformationEvaluator[In, Out] = Callable[[ResolvedSourceMethods], ComputationFunc[In, Out]] +"""Factory producing a bound computation function from resolved parent methods.""" + + +@dataclass(frozen=True, slots=True) +class TransformationMethod[In, Out](AnalyticalComputation[In, Out]): + """ + Analytical computation originating from a transformation. + + Parameters + ---------- + target : GenericCharacteristicName + Name of the target characteristic produced by the transformation. + func : ComputationFunc[In, Out] + Bound callable implementing the transformed characteristic. + transformation : TransformationName + Logical name of the transformation that created this computation. + source_requirements : SourceRequirements + Required parent characteristics used to build the computation. + """ + + transformation: TransformationName + source_requirements: SourceRequirements = field(default_factory=dict) + + @classmethod + def from_parents( + cls, + *, + target: GenericCharacteristicName, + transformation: TransformationName, + bases: Mapping[ParentRole, Distribution], + source_requirements: SourceRequirements, + evaluator: TransformationEvaluator[In, Out], + ) -> TransformationMethod[In, Out]: + """ + Build a transformation method from parent distributions. + + Parent requirements are resolved through ``query_method()``. This + allows a transformation to depend on either directly analytical + parent characteristics or characteristics obtained from the parent + computation graph. + + Parameters + ---------- + target : GenericCharacteristicName + Target characteristic produced by the method. + transformation : TransformationName + Logical transformation name. + bases : Mapping[ParentRole, Distribution] + Parent distributions grouped by role. + source_requirements : SourceRequirements + Required parent characteristics. + evaluator : TransformationEvaluator[InT, OutT] + Factory producing the bound transformed computation from the + resolved parent methods. + + Returns + ------- + TransformationMethod[InT, OutT] + Bound transformation method. + """ + resolved: ResolvedSourceMethods = { + role: { + characteristic: bases[role].query_method(characteristic) + for characteristic in characteristics + } + for role, characteristics in source_requirements.items() + } + + return cls( + target=target, + func=evaluator(resolved), + transformation=transformation, + source_requirements=source_requirements, + ) + + +__all__ = [ + "ResolvedSourceMethods", + "SourceRequirements", + "TransformationEvaluator", + "TransformationMethod", +] diff --git a/src/pysatl_core/types.py b/src/pysatl_core/types.py index 15cd90e..f473b89 100644 --- a/src/pysatl_core/types.py +++ b/src/pysatl_core/types.py @@ -169,6 +169,7 @@ def __post_init__(self) -> None: @overload def contains(self, x: Number) -> bool: ... + @overload def contains(self, x: NumericArray) -> BoolArray: ... @@ -252,6 +253,8 @@ def shape(self) -> ContinuousSupportShape1D: implementations may or may not accept them, and wrappers typically forward ``**options`` dynamically. """ +type ParentRole = str +"""Type alias for logical roles of parent distributions in a transformation.""" class CharacteristicName(StrEnum): @@ -283,12 +286,53 @@ class CharacteristicName(StrEnum): STANDARD_MOMENT = "standardized_moment" # unimplemented in graph yet +class TransformationName(StrEnum): + """ + Enumeration of built-in distribution transformations. + + Attributes + ---------- + AFFINE + Affine transformation ``aX + b``. + BINARY + Binary operation on two parent distributions. + FUNCTION + Functional transformation ``f(X)``. + FINITE_MIXTURE + Finite weighted mixture of component distributions. + APPROXIMATION + Materialized approximation of a transformed distribution. + """ + + AFFINE = "affine" + BINARY = "binary" + FUNCTION = "function" + FINITE_MIXTURE = "finite_mixture" + APPROXIMATION = "approximation" + + class FamilyName(StrEnum): NORMAL = "Normal" CONTINUOUS_UNIFORM = "ContinuousUniform" EXPONENTIAL = "Exponential" +class ApproximationName(StrEnum): + """ + Enumeration of built-in approximation procedures. + + Attributes + ---------- + CUSTOM + Generic user-provided approximation procedure. + CHEBYSHEV + Approximation based on Chebyshev polynomials. + """ + + CUSTOM = "custom" + CHEBYSHEV = "chebyshev" + + __all__ = [ "Kind", "EuclideanDistributionType", @@ -297,6 +341,9 @@ class FamilyName(StrEnum): "GenericCharacteristicName", "ParametrizationName", "ComputationFunc", + "TransformationName", + "ParentRole", + "ApproximationName", "DistributionType", "Interval1D", "ContinuousSupportShape1D",