Prototype: add knowledge database for families transformations

src/pysatl_core/families/__init__.py

@@ -22,6 +22,7 @@
     parametrization,
 )
 from .registry import ParametricFamilyRegister
+from .registry_graph import BinaryOperationType
 
 __all__ = [
     "ParametricFamilyRegister",
@@ -32,6 +33,7 @@
     "constraint",
     "parametrization",
     "configure_families_register",
+    "BinaryOperationType",
     # builtins
     *_builtins_all,
 ]

src/pysatl_core/families/builtins/continuous/__init__.py

@@ -10,11 +10,13 @@
 
 
 from pysatl_core.families.builtins.continuous.exponential import configure_exponential_family
+from pysatl_core.families.builtins.continuous.lognormal import configure_lognormal_family
 from pysatl_core.families.builtins.continuous.normal import configure_normal_family
 from pysatl_core.families.builtins.continuous.uniform import configure_uniform_family
 
 __all__ = [
     "configure_normal_family",
     "configure_uniform_family",
     "configure_exponential_family",
+    "configure_lognormal_family",
 ]

src/pysatl_core/families/builtins/continuous/lognormal.py

@@ -0,0 +1,149 @@
+"""
+Log-Normal distribution family implementation.
+
+Contains the LogNormal family with multiple parameterizations.
+"""
+
+from __future__ import annotations
+
+__author__ = "Fedor Myznikov"
+__copyright__ = "Copyright (c) 2025"
+__license__ = "SPDX-License-Identifier: MIT"
+
+from typing import cast
+
+import numpy as np
+from scipy.special import erf, erfinv
+
+from pysatl_core.distributions.support import ContinuousSupport
+from pysatl_core.families.parametric_family import ParametricFamily
+from pysatl_core.families.parametrizations import (
+    Parametrization,
+    constraint,
+    parametrization,
+)
+from pysatl_core.families.registry import ParametricFamilyRegister
+from pysatl_core.types import (
+    CharacteristicName,
+    FamilyName,
+    NumericArray,
+    UnivariateContinuous,
+)
+
+
+def configure_lognormal_family() -> None:
+    """Configure and register the LogNormal distribution family."""
+
+    if ParametricFamilyRegister.contains(FamilyName.LOGNORMAL):
+        return
+
+    LOGNORMAL_DOC = """
+    Log-Normal distribution.
+
+    If a random variable Y is normally distributed with mean μ and standard deviation σ,
+    then X = exp(Y) follows a log‑normal distribution. Its probability density function is:
+
+        f(x) = 1/(x σ √(2π)) * exp(-(ln x - μ)²/(2σ²)),   for x > 0.
+
+    The distribution is often used to model quantities that cannot be negative,
+    such as incomes, stock prices, or lifetimes.
+    """
+
+    def pdf(parameters: Parametrization, x: NumericArray) -> NumericArray:
+        """Probability density function of the log‑normal distribution."""
+        params = cast(_MeanStd, parameters)
+        mu = params.mu
+        sigma = params.sigma
+
+        if x <= 0:
+            raise ValueError("X must be in [0, +inf)")
+
+        exponent = np.exp(-((np.log(x) - mu) ** 2) / (2 * sigma**2))
+        coefficient = 1 / (x * sigma * np.sqrt(np.pi * 2))
+
+        return cast(NumericArray, coefficient * exponent)
+
+    def cdf(parameters: Parametrization, x: NumericArray) -> NumericArray:
+        parameters = cast(_MeanStd, parameters)
+        if x <= 0:
+            raise ValueError("X must be in [0, +inf)")
+
+        z = (np.log(x) - parameters.mu) / (parameters.sigma * np.sqrt(2))
+        return cast(NumericArray, 0.5 * (1 + erf(z)))
+
+    def ppf(parameters: Parametrization, p: NumericArray) -> NumericArray:
+        if np.any((p < 0) | (p > 1)):
+            raise ValueError("Probability must be in [0, 1]")
+
+        parameters = cast(_MeanStd, parameters)
+        result = np.exp(parameters.mu + np.sqrt(2 * parameters.sigma**2) * erfinv(2 * p - 1))
+        return cast(NumericArray, result)
+
+    def lpdf(parameters: Parametrization, x: NumericArray) -> NumericArray:
+        return np.log(pdf(parameters, x))
+
+    def mean_func(parameters: Parametrization) -> float:
+        """Mean of normal distribution."""
+        parameters = cast(_MeanStd, parameters)
+        return cast(float, np.exp(parameters.mu + parameters.sigma**2 / 2))
+
+    def var_func(parameters: Parametrization) -> float:
+        """Variance of normal distribution."""
+        parameters = cast(_MeanStd, parameters)
+        return cast(
+            float,
+            (np.exp(parameters.sigma**2) - 1) * np.exp(2 * parameters.mu + parameters.sigma**2),
+        )
+
+    def skew_func(parameters: Parametrization) -> float:
+        """Skewness of normal distribution (always 0)."""
+        parameters = cast(_MeanStd, parameters)
+        return cast(
+            float, (np.exp(parameters.sigma**2) + 2) * np.sqrt(np.exp(parameters.sigma**2) - 1)
+        )
+
+    def kurt_func(parameters: Parametrization, excess: bool = False) -> float:
+        parameters = cast(_MeanStd, parameters)
+        if not excess:
+            return 0.0
+        else:
+            return cast(
+                float,
+                np.exp(4 * parameters.sigma**2)
+                + 2 * np.exp(3 * parameters.sigma**2)
+                + 3 * np.exp(2 * parameters.sigma**2)
+                - 6,
+            )
+
+    def _support(_: Parametrization) -> ContinuousSupport:
+        """Support of the log‑normal distribution (0, ∞)."""
+        return ContinuousSupport(left=0.0, left_closed=False, right=np.inf, right_closed=False)
+
+    LogNormal = ParametricFamily(
+        name=FamilyName.LOGNORMAL,
+        distr_type=UnivariateContinuous,
+        distr_parametrizations=["meanStd"],
+        distr_characteristics={
+            CharacteristicName.PDF: pdf,
+            CharacteristicName.CDF: cdf,
+            CharacteristicName.PPF: ppf,
+            CharacteristicName.LPDF: lpdf,
+            CharacteristicName.MEAN: mean_func,
+            CharacteristicName.VAR: var_func,
+            CharacteristicName.SKEW: skew_func,
+            CharacteristicName.KURT: kurt_func,
+        },
+        support_by_parametrization=_support,
+    )
+    LogNormal.__doc__ = LOGNORMAL_DOC
+
+    @parametrization(family=LogNormal, name="meanStd")  # family will be set after Normal is created
+    class _MeanStd(Parametrization):
+        mu: float
+        sigma: float
+
+        @constraint(description="sigma > 0")
+        def check_sigma_positive(self) -> bool:
+            return self.sigma > 0
+
+    ParametricFamilyRegister.register(LogNormal)

src/pysatl_core/families/configuration.py

@@ -25,6 +25,7 @@
 
 from pysatl_core.families.builtins import (
     configure_exponential_family,
+    configure_lognormal_family,
     configure_normal_family,
     configure_uniform_family,
 )
@@ -48,6 +49,7 @@ def configure_families_register() -> ParametricFamilyRegister:
     configure_exponential_family()
     configure_uniform_family()
     configure_normal_family()
+    configure_lognormal_family()
     return ParametricFamilyRegister()
 
 

src/pysatl_core/families/distribution.py

@@ -7,10 +7,15 @@
 
 from __future__ import annotations
 
+import dataclasses
+
+from pysatl_core.families.registry_graph import BinaryOperationType
+
 __author__ = "Leonid Elkin, Mikhail Mikhailov"
 __copyright__ = "Copyright (c) 2025 PySATL project"
 __license__ = "SPDX-License-Identifier: MIT"
 
+from types import NotImplementedType
 from typing import TYPE_CHECKING
 
 from pysatl_core.distributions.distribution import _KEEP, Distribution
@@ -193,3 +198,112 @@ def sample(self, n: int, **options: Any) -> NumericArray:
             the sampling strategy.
         """
         return self.sampling_strategy.sample(n, distr=self, **options)
+
+    def _try_to_transform_with_optimization(
+        self, other: ParametricFamilyDistribution, kind: BinaryOperationType
+    ) -> None | ParametricFamilyDistribution:
+        registry = ParametricFamilyRegister()
+        transform_result = registry.find_binary_transformation(
+            self.family_name, other.family_name, kind
+        )
+        if transform_result is None:
+            return None
+
+        family, transform_parametrization = transform_result
+        new_parametrization = transform_parametrization(
+            self.parametrization.transform_to_base_parametrization(),
+            other.parametrization.transform_to_base_parametrization(),
+        )
+        return family(new_parametrization.name, **dataclasses.asdict(new_parametrization))  # type:ignore[call-overload]
+
+    def __add__(
+        self, other: object
+    ) -> ParametricFamilyDistribution | Distribution | NotImplementedType:
+        """Return ``self + other`` for scalar or distribution operands."""
+        if isinstance(other, ParametricFamilyDistribution):
+            transformation_result = self._try_to_transform_with_optimization(
+                other, BinaryOperationType.ADD
+            )
+            if transformation_result is not None:
+                return transformation_result
+
+        return TransformationOperatorsMixin.__add__(self, other)
+
+    def __radd__(
+        self, other: object
+    ) -> ParametricFamilyDistribution | Distribution | NotImplementedType:
+        """Return ``other + self`` for scalar or distribution operands."""
+        if isinstance(other, ParametricFamilyDistribution):
+            transformation_result = other._try_to_transform_with_optimization(
+                self, BinaryOperationType.ADD
+            )
+            if transformation_result is not None:
+                return transformation_result
+
+        return TransformationOperatorsMixin.__radd__(self, other)
+
+    def __sub__(self, other: object) -> Distribution | NotImplementedType:
+        """Return ``self - other`` for scalar or distribution operands."""
+        if isinstance(other, ParametricFamilyDistribution):
+            transformation_result = self._try_to_transform_with_optimization(
+                other, BinaryOperationType.SUB
+            )
+            if transformation_result is not None:
+                return transformation_result
+
+        return TransformationOperatorsMixin.__sub__(self, other)
+
+    def __rsub__(self, other: object) -> Distribution | NotImplementedType:
+        """Return ``other - self`` for scalar or distribution operands."""
+        if isinstance(other, ParametricFamilyDistribution):
+            transformation_result = other._try_to_transform_with_optimization(
+                self, BinaryOperationType.SUB
+            )
+            if transformation_result is not None:
+                return transformation_result
+
+        return TransformationOperatorsMixin.__rsub__(self, other)
+
+    def __mul__(self, other: object) -> Distribution | NotImplementedType:
+        """Return ``self * other`` for scalar or distribution operands."""
+        if isinstance(other, ParametricFamilyDistribution):
+            transformation_result = self._try_to_transform_with_optimization(
+                other, BinaryOperationType.MUL
+            )
+            if transformation_result is not None:
+                return transformation_result
+
+        return TransformationOperatorsMixin.__mul__(self, other)
+
+    def __rmul__(self, other: object) -> Distribution | NotImplementedType:
+        """Return ``other * self`` for scalar or distribution operands."""
+        if isinstance(other, ParametricFamilyDistribution):
+            transformation_result = other._try_to_transform_with_optimization(
+                self, BinaryOperationType.MUL
+            )
+            if transformation_result is not None:
+                return transformation_result
+
+        return TransformationOperatorsMixin.__rmul__(self, other)
+
+    def __truediv__(self, other: object) -> Distribution | NotImplementedType:
+        """Return ``self / other`` for scalar or distribution operands."""
+        if isinstance(other, ParametricFamilyDistribution):
+            transformation_result = self._try_to_transform_with_optimization(
+                other, BinaryOperationType.DIV
+            )
+            if transformation_result is not None:
+                return transformation_result
+
+        return TransformationOperatorsMixin.__truediv__(self, other)
+
+    def __rtruediv__(self, other: object) -> Distribution | NotImplementedType:
+        """Return ``other / self`` for distribution operands."""
+        if isinstance(other, ParametricFamilyDistribution):
+            transformation_result = other._try_to_transform_with_optimization(
+                self, BinaryOperationType.DIV
+            )
+            if transformation_result is not None:
+                return transformation_result
+
+        return TransformationOperatorsMixin.__rtruediv__(self, other)

src/pysatl_core/families/parametric_family.py

@@ -23,6 +23,7 @@
 from pysatl_core.distributions.computations.computation import AnalyticalComputation
 from pysatl_core.families.distribution import ParametricFamilyDistribution
 from pysatl_core.families.parametrizations import Parametrization, ParametrizationConstraint
+from pysatl_core.families.registry import ParametricFamilyRegister
 from pysatl_core.types import (
     DEFAULT_ANALYTICAL_COMPUTATION_LABEL,
     ComputationFunc,
@@ -384,6 +385,25 @@ def distribution(
 
         parameters = parametrization_class(**parameters_values)
         parameters.validate()
+        registry = ParametricFamilyRegister()
+
+        optimal_result = registry.get_optimal_family(self.name, parameters)
+        if optimal_result is not None:
+            optimal_family, optimal_parametrization = optimal_result
+            distribution_type = optimal_family._distr_type(optimal_parametrization)
+            analytical_computations = optimal_family._build_analytical_computations(
+                optimal_parametrization
+            )
+            return ParametricFamilyDistribution(
+                family_name=optimal_family.name,
+                distribution_type=distribution_type,
+                analytical_computations=analytical_computations,
+                parametrization=optimal_parametrization,
+                support=optimal_family.support_resolver(optimal_parametrization),
+                sampling_strategy=sampling_strategy,
+                computation_strategy=computation_strategy,
+            )
+
         base_parameters = self.to_base(parameters)
         distribution_type = self._distr_type(base_parameters)
         analytical_computations = self._build_analytical_computations(parameters)