feat: add new stats for graph stats - clique number and independence number

Author: Ivan-Pokhabov

.github/workflows/ci.yaml

@@ -14,7 +14,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [ "3.10", "3.11", "3.12" ]
+        python-version: [ "3.11", "3.12" ]
 
     steps:
       - name: Checkout repository

pyproject.toml

@@ -7,12 +7,13 @@ authors = [
 ]
 license = {text = "MIT"}
 readme = "README.md"
-requires-python = ">=3.9,<3.13"
+requires-python = ">=3.11,<3.13"
 dependencies = [
     "numpy>=1.25.1",
     "scipy>=1.11.2",
     "pandas>=2.2.1",
-    "typing-extensions>=4.12.2"
+    "typing-extensions>=4.12.2",
+    "networkx (>=3.5,<4.0)"
 ]
 
 [project.urls]

pysatl_criterion/persistence/limit_distribution/sqlite/sqlite.py

@@ -120,7 +120,7 @@ def get_data(self, query: LimitDistributionQuery) -> LimitDistributionModel | No
             return None
 
         columns = [col[0] for col in cursor.description]
-        return self._row_to_model(dict(zip(columns, row)))
+        return self._row_to_model(dict(zip(columns, row, strict=False)))
 
     def delete_data(self, query: LimitDistributionQuery) -> None:
         """Delete specific limit distribution data."""
@@ -164,4 +164,4 @@ def get_data_for_cv(self, query: CriticalValueQuery) -> LimitDistributionModel |
             return None
 
         columns = [col[0] for col in cursor.description]
-        return self._row_to_model(dict(zip(columns, row)))
+        return self._row_to_model(dict(zip(columns, row, strict=False)))

pysatl_criterion/statistics/exponent.py

@@ -9,9 +9,12 @@
 from pysatl_criterion.statistics.common import KSStatistic
 from pysatl_criterion.statistics.goodness_of_fit import AbstractGoodnessOfFitStatistic
 from pysatl_criterion.statistics.graph_goodness_of_fit import (
+    AbstractGraphTestStatistic,
     GraphAverageDegreeTestStatistic,
+    GraphCliqueNumberTestStatistic,
     GraphConnectedComponentsTestStatistic,
     GraphEdgesNumberTestStatistic,
+    GraphIndependenceNumberTestStatistic,
     GraphMaxDegreeTestStatistic,
 )
 
@@ -810,73 +813,78 @@ def execute_statistic(self, rvs, **kwargs):
         return hg
 
 
-class GraphEdgesNumberExponentialityGofStatistic(
-    AbstractExponentialityGofStatistic, GraphEdgesNumberTestStatistic
+class AbstractGraphExponentialityGofStatistic(
+    AbstractExponentialityGofStatistic, AbstractGraphTestStatistic
 ):
     @staticmethod
     @override
     def code():
-        super_class = AbstractExponentialityGofStatistic
-        parent_code = super(super_class, super_class).code()
-        return f"EdgesNumber_{parent_code}"
+        parent_code = AbstractExponentialityGofStatistic.code()
+        return f"GRAPH_{parent_code}"
 
     @staticmethod
     @override
     def _compute_dist(rvs):
-        super_class = GraphEdgesNumberTestStatistic
-        parent_code = super(super_class, super_class)._compute_dist(rvs)
-        return parent_code / np.mean(rvs)
+        base_dist = AbstractGraphTestStatistic._compute_dist(rvs)
+        mean = np.mean(rvs)
+        return base_dist / mean if mean != 0 else base_dist
 
 
-class GraphMaxDegreeExponentialityGofStatistic(
-    AbstractExponentialityGofStatistic, GraphMaxDegreeTestStatistic
+class GraphEdgesNumberExponentialityGofStatistic(
+    AbstractGraphExponentialityGofStatistic, GraphEdgesNumberTestStatistic
 ):
     @staticmethod
     @override
     def code():
-        super_class = AbstractExponentialityGofStatistic
-        parent_code = super(super_class, super_class).code()
-        return f"MaxDegree_{parent_code}"
+        parent_code = AbstractGraphExponentialityGofStatistic.code()
+        return f"{GraphEdgesNumberExponentialityGofStatistic.get_stat_name()}_{parent_code}"
 
+
+class GraphMaxDegreeExponentialityGofStatistic(
+    AbstractGraphExponentialityGofStatistic, GraphMaxDegreeTestStatistic
+):
     @staticmethod
     @override
-    def _compute_dist(rvs):
-        super_class = GraphMaxDegreeTestStatistic
-        parent_code = super(super_class, super_class)._compute_dist(rvs)
-        return parent_code / np.mean(rvs)
+    def code():
+        parent_code = AbstractGraphExponentialityGofStatistic.code()
+        return f"{GraphMaxDegreeExponentialityGofStatistic.get_stat_name()}_{parent_code}"
 
 
 class GraphAverageDegreeExponentialityGofStatistic(
-    AbstractExponentialityGofStatistic, GraphAverageDegreeTestStatistic
+    AbstractGraphExponentialityGofStatistic, GraphAverageDegreeTestStatistic
 ):
     @staticmethod
     @override
     def code():
-        super_class = AbstractExponentialityGofStatistic
-        parent_code = super(super_class, super_class).code()
-        return f"AverageDegree_{parent_code}"
+        parent_code = AbstractGraphExponentialityGofStatistic.code()
+        return f"{GraphAverageDegreeExponentialityGofStatistic.get_stat_name()}_{parent_code}"
+
 
+class GraphConnectedComponentsExponentialityGofStatistic(
+    AbstractGraphExponentialityGofStatistic, GraphConnectedComponentsTestStatistic
+):
     @staticmethod
     @override
-    def _compute_dist(rvs):
-        super_class = GraphAverageDegreeTestStatistic
-        parent_dist = super(super_class, super_class)._compute_dist(rvs)
-        return parent_dist / np.mean(rvs)
+    def code():
+        parent_code = AbstractGraphExponentialityGofStatistic.code()
+        return f"{GraphConnectedComponentsExponentialityGofStatistic.get_stat_name()}_{parent_code}"
 
 
-class GraphConnectedComponentsExponentialityGofStatistic(
-    AbstractExponentialityGofStatistic, GraphConnectedComponentsTestStatistic
+class GraphCliqueNumberExponentialityGofStatistic(
+    AbstractGraphExponentialityGofStatistic, GraphCliqueNumberTestStatistic
 ):
     @staticmethod
     @override
     def code():
-        super_class = AbstractExponentialityGofStatistic
-        parent_code = super(super_class, super_class).code()
-        return f"ConnectedComponents_{parent_code}"
+        parent_code = AbstractGraphExponentialityGofStatistic.code()
+        return f"{GraphCliqueNumberExponentialityGofStatistic.get_stat_name()}_{parent_code}"
 
+
+class GraphIndependenceNumberExponentialityGofStatistic(
+    AbstractGraphExponentialityGofStatistic, GraphIndependenceNumberTestStatistic
+):
     @staticmethod
     @override
-    def _compute_dist(rvs):
-        super_class = GraphConnectedComponentsTestStatistic
-        parent_dist = super(super_class, super_class)._compute_dist(rvs)
-        return parent_dist / np.mean(rvs)
+    def code():
+        parent_code = AbstractGraphExponentialityGofStatistic.code()
+        return f"{GraphIndependenceNumberExponentialityGofStatistic.get_stat_name()}_{parent_code}"

pysatl_criterion/statistics/graph_goodness_of_fit.py

@@ -1,5 +1,4 @@
 from abc import ABC
-from typing import Union
 
 import numpy as np
 from numpy import float64
@@ -10,13 +9,21 @@
 
 class AbstractGraphTestStatistic(AbstractGoodnessOfFitStatistic, ABC):
     @override
-    def execute_statistic(self, rvs, **kwargs) -> Union[float, float64]:
+    def execute_statistic(self, rvs, **kwargs) -> float | float64:
         dist = self._compute_dist(rvs)
 
         adjacency_list = self._make_adjacency_list(rvs, dist)
         statistic = self.get_graph_stat(adjacency_list)
         return statistic
 
+    @staticmethod
+    def get_stat_name() -> str:
+        raise NotImplementedError("Method is not implemented")
+
+    @staticmethod
+    def get_graph_stat(graph: list[list[int]]) -> float:
+        raise NotImplementedError("Method is not implemented")
+
     @staticmethod
     def _make_adjacency_list(rvs, dist: float) -> list[list[int]]:
         adjacency_list: list[list[int]] = []
@@ -31,40 +38,51 @@ def _make_adjacency_list(rvs, dist: float) -> list[list[int]]:
         return adjacency_list
 
     @staticmethod
-    def _compute_dist(rvs):  # TODO (normalize for different distributions)
+    def _compute_dist(rvs: list[float]) -> float:  # TODO (normalize for different distributions)
         return (max(rvs) - min(rvs)) / 10
 
-    @staticmethod
-    def get_graph_stat(graph: list[list[int]]):
-        raise NotImplementedError("Method is not implemented")
 
-
-class GraphEdgesNumberTestStatistic(AbstractGraphTestStatistic, ABC):
+class GraphEdgesNumberTestStatistic(AbstractGraphTestStatistic):
     @staticmethod
     @override
-    def get_graph_stat(graph):
+    def get_graph_stat(graph: list[list[int]]) -> float:
         return sum(map(len, graph)) // 2
 
+    @staticmethod
+    @override
+    def get_stat_name() -> str:
+        return "EDGESNUMBER"
+
 
-class GraphMaxDegreeTestStatistic(AbstractGraphTestStatistic, ABC):
+class GraphMaxDegreeTestStatistic(AbstractGraphTestStatistic):
     @staticmethod
     @override
-    def get_graph_stat(graph):
+    def get_graph_stat(graph: list[list[int]]) -> float:
         return max(map(len, graph))
 
+    @staticmethod
+    @override
+    def get_stat_name() -> str:
+        return "MAXDEGREE"
+
 
-class GraphAverageDegreeTestStatistic(AbstractGraphTestStatistic, ABC):
+class GraphAverageDegreeTestStatistic(AbstractGraphTestStatistic):
     @staticmethod
     @override
-    def get_graph_stat(graph):
+    def get_graph_stat(graph: list[list[int]]) -> float:
         degrees = list(map(len, graph))
-        return np.mean(degrees) if degrees != 0 else 0.0
+        return float(np.mean(degrees)) if degrees != 0 else 0.0
+
+    @staticmethod
+    @override
+    def get_stat_name() -> str:
+        return "AVGDEGREE"
 
 
-class GraphConnectedComponentsTestStatistic(AbstractGraphTestStatistic, ABC):
+class GraphConnectedComponentsTestStatistic(AbstractGraphTestStatistic):
     @staticmethod
     @override
-    def get_graph_stat(graph):
+    def get_graph_stat(graph) -> float:
         visited = set()
         components = 0
 
@@ -81,3 +99,57 @@ def dfs(node):
                 dfs(node)
                 components += 1
         return components
+
+    @staticmethod
+    @override
+    def get_stat_name() -> str:
+        return "CONNECTEDCOMPONENTS"
+
+
+class GraphCliqueNumberTestStatistic(AbstractGraphTestStatistic):
+    @override
+    def execute_statistic(self, rvs, **kwargs) -> float | float64:
+        dist = self._compute_dist(rvs)
+        rvs.sort()
+
+        right_border = 0
+        clique_number = 0
+        for left_border in range(len(rvs)):
+            while right_border < len(rvs) and rvs[left_border] + dist > rvs[right_border]:
+                right_border += 1
+            if right_border == len(rvs):
+                clique_number = max(clique_number, right_border - left_border + 1)
+                break
+            clique_number = max(clique_number, right_border - left_border)
+        return clique_number
+
+    @staticmethod
+    @override
+    def get_stat_name() -> str:
+        return "CLIQUENUMBER"
+
+
+class GraphIndependenceNumberTestStatistic(AbstractGraphTestStatistic):
+    @override
+    def execute_statistic(self, rvs, **kwargs) -> float | float64:
+        if not rvs:
+            return 0
+
+        dist = self._compute_dist(rvs)
+        rvs.sort()
+
+        stat = 1
+        last_chosen_position = rvs[0]
+
+        for i in range(1, len(rvs)):
+            current_point = rvs[i]
+            if current_point >= last_chosen_position + dist:
+                stat += 1
+                last_chosen_position = current_point
+
+        return stat
+
+    @staticmethod
+    @override
+    def get_stat_name() -> str:
+        return "INDEPENDENCENUMBER"