Merge pull request #32 from PySATL/feat/p-value-calc

feat: add P-value calculator

Author: alex98247

pysatl_criterion/cv_calculator/cv_calculator/cv_calculator.py

@@ -5,6 +5,7 @@
     CriticalValueQuery,
     ILimitDistributionStorage,
 )
+from pysatl_criterion.statistics.models import HypothesisType
 
 
 class CVCalculator:
@@ -17,13 +18,20 @@ class CVCalculator:
     def __init__(self, limit_distribution_storage: ILimitDistributionStorage):
         self.limit_distribution_storage = limit_distribution_storage
 
-    def calculate_critical_value(self, criterion_code: str, sample_size: int, sl: float) -> float:
+    def calculate_critical_value(
+        self,
+        criterion_code: str,
+        sample_size: int,
+        sl: float,
+        alternative: HypothesisType = HypothesisType.RIGHT,
+    ) -> float | tuple[float, float]:
         """
         Calculate critical value for given criterion.
 
         :param criterion_code: criterion code.
         :param sample_size: sample size.
         :param sl: significance level.
+        :param alternative: test alternative
 
         :return: critical value.
         """
@@ -40,36 +48,13 @@ def calculate_critical_value(self, criterion_code: str, sample_size: int, sl: fl
 
         ecdf = scipy_stats.ecdf(statistics_values)
 
-        critical_value = float(np.quantile(ecdf.cdf.quantiles, q=1 - sl))
-
-        return critical_value
-
-    def calculate_two_tailed_critical_values(
-        self, criterion_code: str, sample_size: int, sl: float
-    ) -> tuple[float, float]:
-        """
-        Calculate critical values for two-tailed criterion.
-
-        :param criterion_code: criterion code.
-        :param sample_size: sample size.
-        :param sl: significance level.
-
-        :return: critical values.
-        """
-
-        query = CriticalValueQuery(criterion_code=criterion_code, sample_size=sample_size)
-
-        limit_distribution_from_db = self.limit_distribution_storage.get_data_for_cv(query)
-        if limit_distribution_from_db is None:
-            raise ValueError(
-                "Limit distribution for given criterion and sample size does not exist."
-            )
-
-        statistics_values = limit_distribution_from_db.results_statistics
-
-        ecdf = scipy_stats.ecdf(statistics_values)
-
-        critical_value_left = float(np.quantile(ecdf.cdf.quantiles, q=sl / 2))
-        critical_value_right = float(np.quantile(ecdf.cdf.quantiles, q=1 - sl / 2))
-
-        return critical_value_left, critical_value_right
+        if alternative == HypothesisType.RIGHT:
+            return float(np.quantile(ecdf.cdf.quantiles, q=1 - sl))
+        elif alternative == HypothesisType.LEFT:
+            return float(np.quantile(ecdf.cdf.quantiles, q=sl))
+        elif alternative == HypothesisType.TWO_TAILED:
+            left = float(np.quantile(ecdf.cdf.quantiles, q=sl / 2))
+            right = float(np.quantile(ecdf.cdf.quantiles, q=1 - sl / 2))
+            return left, right
+        else:
+            raise ValueError("Unknown alternative.")

pysatl_criterion/p_value_calculator/__init__.py

@@ -0,0 +1,61 @@
+import scipy.stats as scipy_stats
+
+from pysatl_criterion.persistence.limit_distribution.sqlite.sqlite import (
+    SQLiteLimitDistributionStorage,
+)
+from pysatl_criterion.persistence.model.limit_distribution.limit_distribution import (
+    CriticalValueQuery,
+)
+from pysatl_criterion.statistics.models import HypothesisType
+
+
+class PValueCalculator:
+    """
+    P-value calculator.
+
+    :param limit_distribution_storage: limit distribution storage
+    """
+
+    def __init__(self, limit_distribution_storage: SQLiteLimitDistributionStorage):
+        self.limit_distribution_storage = limit_distribution_storage
+
+    def calculate_p_value(
+        self,
+        criterion_code: str,
+        sample_size: int,
+        statistics_value: float,
+        alternative: HypothesisType = HypothesisType.RIGHT,
+    ) -> float:
+        """
+        Calculate p-value.
+
+        :param criterion_code: criterion code
+        :param sample_size: sample size
+        :param statistics_value: statistics value
+        :param alternative: test alternative
+
+        :return: p-value
+        """
+
+        query = CriticalValueQuery(criterion_code=criterion_code, sample_size=sample_size)
+        limit_distribution_from_db = self.limit_distribution_storage.get_data_for_cv(query)
+
+        if limit_distribution_from_db is None:
+            raise ValueError(
+                "Limit distribution for given criterion and sample size does not exist."
+            )
+
+        simulation_results = limit_distribution_from_db.results_statistics
+
+        ecdf = scipy_stats.ecdf(simulation_results)
+
+        cdf_value = ecdf.cdf.evaluate(statistics_value)
+
+        if alternative == HypothesisType.RIGHT:
+            return 1.0 - cdf_value
+        elif alternative == HypothesisType.TWO_TAILED:
+            return 2.0 * min(cdf_value, 1.0 - cdf_value)
+        elif alternative == HypothesisType.LEFT:
+            return cdf_value
+        else:
+            raise ValueError("Unknown alternative")

pysatl_criterion/p_value_calculator/p_value_calculator/p_value_calculator.py

@@ -1,11 +1,43 @@
 from abc import ABC, abstractmethod
+from enum import Enum, auto
 
 from numpy import float64
 
 
-class AbstractStatistic(ABC):
-    two_tailed: bool = False
+class HypothesisType(Enum):
+    """
+    Alternatives for hypotheses.
+    """
+
+    RIGHT = auto()
+    LEFT = auto()
+    TWO_TAILED = auto()
+
+    def check_hypothesis(
+        self,
+        statistic_value,
+        cv: float | tuple[float, float],
+    ) -> bool:
+        """
+        Compares the value of a statistic to a critical value
+        param statistic_value: statistic value
+        param cv: critical value
+
+        return: True if hypothesis is valid, False otherwise
+
+        """
+        if self == HypothesisType.RIGHT:
+            return statistic_value <= cv
+        if self == HypothesisType.LEFT:
+            return statistic_value >= cv
+        if self == HypothesisType.TWO_TAILED:
+            if not isinstance(cv, tuple):
+                raise TypeError("For a TWO_SIDED hypothesis, 'cv' must be a tuple of two floats.")
+            left_cv, right_cv = cv
+            return left_cv <= statistic_value <= right_cv
+
 
+class AbstractStatistic(ABC):
     @staticmethod
     @abstractmethod
     def code() -> str:
@@ -22,19 +54,3 @@ def execute_statistic(self, rvs, **kwargs) -> float | float64:
         :param kwargs: arguments for statistic calculation
         """
         raise NotImplementedError("Method is not implemented")
-
-    def calculate_critical_value(self, rvs_size, sl) -> float | float64 | None:
-        """
-        Calculate critical value for test statistics
-        :param rvs_size: rvs size
-        :param sl: significance level
-        """
-        return None
-
-    def calculate_two_tailed_critical_values(self, rvs_size: int, sl) -> tuple[float, float] | None:
-        """
-        Calculate two-tailed critical values for test statistics
-        :param rvs_size: rvs size
-        :param sl: significance level
-        """
-        return None

pysatl_criterion/statistics/models.py

@@ -1,8 +1,12 @@
 from pysatl_criterion.cv_calculator.cv_calculator.cv_calculator import CVCalculator
+from pysatl_criterion.p_value_calculator.p_value_calculator.p_value_calculator import (
+    PValueCalculator,
+)
 from pysatl_criterion.persistence.limit_distribution.sqlite.sqlite import (
     SQLiteLimitDistributionStorage,
 )
 from pysatl_criterion.statistics.goodness_of_fit import AbstractGoodnessOfFitStatistic
+from pysatl_criterion.statistics.models import HypothesisType
 
 
 class GoodnessOfFitTest:
@@ -11,17 +15,24 @@ class GoodnessOfFitTest:
 
     :param statistics: statistics.
     :param significance_level: significance level.
+    :param test_method: test method either 'critical_value' or 'p_value'.
+    :param alternative: test alternative.
+
     """
 
     def __init__(
         self,
         statistics: AbstractGoodnessOfFitStatistic,
         significance_level: float,
         db_connection_string: str = "sqlite:///limit_distributions.sqlite",
+        test_method: str = "critical_value",
+        alternative: HypothesisType = HypothesisType.RIGHT,
     ):
         self.statistics = statistics
         self.significance_level = significance_level
         self.db_connection_string = db_connection_string
+        self.test_method = test_method
+        self.alternative = alternative
 
     def test(self, data: list[float]) -> bool:
         """
@@ -35,26 +46,29 @@ def test(self, data: list[float]) -> bool:
         limit_distribution_storage = SQLiteLimitDistributionStorage(self.db_connection_string)
         limit_distribution_storage.init()
 
-        cv_calculator = CVCalculator(limit_distribution_storage)
-
         data_size = len(data)
         criterion_code = self.statistics.code()
         statistics_value = self.statistics.execute_statistic(data)
-        if self.statistics.two_tailed:
-            critical_value_left, critical_value_right = (
-                cv_calculator.calculate_two_tailed_critical_values(
-                    criterion_code, data_size, self.significance_level
-                )
+
+        if self.test_method == "critical_value":
+            cv_calculator = CVCalculator(limit_distribution_storage)
+
+            critical_values = cv_calculator.calculate_critical_value(
+                criterion_code,
+                data_size,
+                self.significance_level,
+                self.alternative,
             )
-            if critical_value_left <= statistics_value <= critical_value_right:
-                return True
-            else:
-                return False
-        else:
-            critical_value = cv_calculator.calculate_critical_value(
-                criterion_code, data_size, self.significance_level
+            return self.alternative.check_hypothesis(statistics_value, critical_values)
+
+        elif self.test_method == "p_value":
+            p_value_calculator = PValueCalculator(limit_distribution_storage)
+            p_value = p_value_calculator.calculate_p_value(
+                criterion_code,
+                data_size,
+                statistics_value,
+                self.alternative,
             )
-            if statistics_value <= critical_value:
-                return True
-            else:
-                return False
+            return p_value >= self.significance_level
+        else:
+            raise ValueError("Invalid test method.")

pysatl_criterion/test/goodness_of_fit_test/goodness_of_fit_test.py

@@ -0,0 +1,79 @@
+from unittest.mock import MagicMock
+
+import numpy as np
+import pytest
+
+from pysatl_criterion.cv_calculator.cv_calculator.cv_calculator import CVCalculator
+from pysatl_criterion.statistics.models import HypothesisType
+
+
+def test_calc_critical_value_right_tailed():
+    mock_storage = MagicMock()
+    mock_distribution = MagicMock()
+
+    mock_distribution.results_statistics = np.array(range(100))
+
+    mock_storage.get_data_for_cv.return_value = mock_distribution
+    calculator = CVCalculator(limit_distribution_storage=mock_storage)
+
+    critical_value = calculator.calculate_critical_value(
+        criterion_code="any_code", sample_size=100, sl=0.05, alternative=HypothesisType.RIGHT
+    )
+    assert critical_value == pytest.approx(94.05)
+
+
+def test_calc_critical_value_left_tailed():
+    mock_storage = MagicMock()
+    mock_distribution = MagicMock()
+
+    mock_distribution.results_statistics = np.array(range(100))
+
+    mock_storage.get_data_for_cv.return_value = mock_distribution
+    calculator = CVCalculator(limit_distribution_storage=mock_storage)
+
+    critical_value = calculator.calculate_critical_value(
+        criterion_code="any_code", sample_size=100, sl=0.05, alternative=HypothesisType.LEFT
+    )
+    assert critical_value == pytest.approx(4.95)
+
+
+def test_calc_critical_value_two_tailed():
+    mock_storage = MagicMock()
+    mock_distribution = MagicMock()
+
+    mock_distribution.results_statistics = np.array(range(100))
+
+    mock_storage.get_data_for_cv.return_value = mock_distribution
+    calculator = CVCalculator(limit_distribution_storage=mock_storage)
+    critical_value = calculator.calculate_critical_value(
+        criterion_code="any_code", sample_size=100, sl=0.05, alternative=HypothesisType.TWO_TAILED
+    )
+    values = (2.475, 96.525)
+    assert critical_value == pytest.approx(values)
+
+
+def test_calc_critical_value_raises_limit_distribution_error():
+    mock_storage = MagicMock()
+    mock_storage.get_data_for_cv.return_value = None
+
+    calculator = CVCalculator(limit_distribution_storage=mock_storage)
+
+    with pytest.raises(
+        ValueError, match="Limit distribution for given criterion and sample size does not exist."
+    ):
+        calculator.calculate_critical_value(criterion_code="any_code", sample_size=100, sl=0.05)
+    mock_storage.get_data_for_cv.assert_called_once()
+
+
+def test_calc_critical_value_raises_unknown_alternative():
+    mock_storage = MagicMock()
+    mock_distribution = MagicMock()
+
+    mock_distribution.results_statistics = np.array(range(100))
+    mock_storage.get_data_for_cv.return_value = mock_distribution
+    calculator = CVCalculator(limit_distribution_storage=mock_storage)
+    with pytest.raises(ValueError, match="Unknown alternative"):
+        calculator.calculate_critical_value(
+            criterion_code="any_code", sample_size=100, sl=0.05, alternative="not_valid"
+        )
+    mock_storage.get_data_for_cv.assert_called_once()