diff --git a/macrobase_py/__init__.py b/macrobase_py/__init__.py index 56412e6b0..fa619c46b 100644 --- a/macrobase_py/__init__.py +++ b/macrobase_py/__init__.py @@ -1,3 +1,17 @@ +"""Python tooling for MacroBase summarization algorithms.""" + +from .explainers import ( + BatchSummarizer, + AprioriSummarizer, + HeuristicSummarizer, + Explanation, +) + +__all__ = [ + "BatchSummarizer", + "AprioriSummarizer", + "HeuristicSummarizer", + "Explanation", """Python utilities for the MacroBase analytics stack.""" from .pipeline import ( # noqa: F401 diff --git a/macrobase_py/explainers/__init__.py b/macrobase_py/explainers/__init__.py index 3e651bccf..6daf4df80 100644 --- a/macrobase_py/explainers/__init__.py +++ b/macrobase_py/explainers/__init__.py @@ -1,3 +1,16 @@ +"""Explainability primitives for MacroBase-inspired workflows.""" + +from .base import BatchSummarizer +from .apriori import AprioriSummarizer +from .heuristic import HeuristicSummarizer +from .explanations import Explanation + +__all__ = [ + "BatchSummarizer", + "AprioriSummarizer", + "HeuristicSummarizer", + "Explanation", +] """Module placeholder for macrobase_py.explainers.""" __all__: list[str] = [] diff --git a/macrobase_py/explainers/apriori.py b/macrobase_py/explainers/apriori.py new file mode 100644 index 000000000..cfec13b38 --- /dev/null +++ b/macrobase_py/explainers/apriori.py @@ -0,0 +1,74 @@ +"""Apriori-style summarizer implementation.""" + +from __future__ import annotations + +from itertools import combinations +from typing import Dict, List, Sequence, Set, Tuple + +import pandas as pd + +from .base import BatchSummarizer, Combination +from .explanations import Explanation + + +class AprioriSummarizer(BatchSummarizer): + """Enumerate attribute combinations with classic Apriori pruning.""" + + def _run(self, df: pd.DataFrame) -> List[Explanation]: + explanations: List[Explanation] = [] + + # Initialize with frequent singletons. + level_itemsets: List[Tuple[Combination, Dict[str, float]]] = [] + for attribute in self.attributes or []: + for value in self._attribute_values[attribute]: + combination = self._normalize_combination(((attribute, value),)) + if combination is None or not self._is_valid_combination(combination): + continue + metrics = self._compute_metrics(combination) + if not self._has_min_support(metrics): + continue + if self._satisfies_ratio(metrics): + explanation = self._build_explanation(combination, metrics) + explanations.append(explanation) + level_itemsets.append((combination, metrics)) + + k = 2 + while level_itemsets and k <= (self.max_order or len(self.attributes or [])): + candidates = self._generate_candidates([combo for combo, _ in level_itemsets], k) + next_level: List[Tuple[Combination, Dict[str, float]]] = [] + for combination in candidates: + metrics = self._compute_metrics(combination) + if not self._has_min_support(metrics): + continue + if self._satisfies_ratio(metrics): + explanation = self._build_explanation(combination, metrics) + explanations.append(explanation) + next_level.append((combination, metrics)) + level_itemsets = next_level + k += 1 + + return explanations + + def _generate_candidates( + self, previous_level: Sequence[Combination], k: int + ) -> List[Combination]: + candidates: Set[Combination] = set() + previous_set = set(previous_level) + for i in range(len(previous_level)): + for j in range(i + 1, len(previous_level)): + left = previous_level[i] + right = previous_level[j] + if left[:-1] != right[:-1]: + continue + merged = self._normalize_combination(left + right[-1:]) + if merged is None: + continue + if not self._is_valid_combination(merged): + continue + if not self._subsets_are_frequent(merged, previous_set): + continue + candidates.add(merged) + return sorted(candidates) + + +__all__ = ["AprioriSummarizer"] diff --git a/macrobase_py/explainers/base.py b/macrobase_py/explainers/base.py new file mode 100644 index 000000000..422199918 --- /dev/null +++ b/macrobase_py/explainers/base.py @@ -0,0 +1,313 @@ +"""Base classes shared by summarizers.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from collections import OrderedDict +from itertools import combinations +from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple + +import numpy as np +import pandas as pd + +from .explanations import Explanation + +Combination = Tuple[Tuple[str, Any], ...] + + +class BatchSummarizer(ABC): + """Base class for MacroBase-style summarizers operating on pandas data. + + Subclasses implement :meth:`_run` to generate candidate explanations. The + base class handles data validation, ratio metric computation, filtering, and + conversion into :class:`Explanation` objects ready for reporting. + """ + + def __init__( + self, + *, + min_support: float = 0.1, + min_ratio_metric: float = 1.0, + ratio_metric: str = "risk_ratio", + attributes: Optional[Sequence[str]] = None, + outlier_column: str = "is_outlier", + max_order: Optional[int] = None, + top_k: Optional[int] = 20, + functional_dependencies: Optional[Dict[str, Iterable[str]]] = None, + use_fd_hints: bool = False, + ) -> None: + self.min_support = min_support + self.min_ratio_metric = min_ratio_metric + self.ratio_metric = ratio_metric + self.attributes = list(attributes) if attributes is not None else None + self.outlier_column = outlier_column + self.max_order = max_order + self.top_k = top_k + self.functional_dependencies = ( + { + determinant: frozenset(dependents) + for determinant, dependents in (functional_dependencies or {}).items() + } + ) + self.use_fd_hints = use_fd_hints + + self._df: Optional[pd.DataFrame] = None + self._outlier_mask: Optional[pd.Series] = None + self._n_outliers: int = 0 + self._n_inliers: int = 0 + self._total_rows: int = 0 + self._attribute_values: Dict[str, List[Any]] = {} + + # ------------------------------------------------------------------ + # Fluent configuration helpers + # ------------------------------------------------------------------ + def set_min_support(self, value: float) -> "BatchSummarizer": + self.min_support = value + return self + + def set_min_ratio_metric(self, value: float) -> "BatchSummarizer": + self.min_ratio_metric = value + return self + + def set_ratio_metric(self, metric: str) -> "BatchSummarizer": + self.ratio_metric = metric + return self + + def set_attributes(self, attributes: Sequence[str]) -> "BatchSummarizer": + self.attributes = list(attributes) + return self + + def set_outlier_column(self, column: str) -> "BatchSummarizer": + self.outlier_column = column + return self + + def set_max_order(self, max_order: int) -> "BatchSummarizer": + self.max_order = max_order + return self + + def set_top_k(self, top_k: Optional[int]) -> "BatchSummarizer": + self.top_k = top_k + return self + + def set_functional_dependencies( + self, functional_dependencies: Dict[str, Iterable[str]] + ) -> "BatchSummarizer": + self.functional_dependencies = { + determinant: frozenset(dependents) + for determinant, dependents in functional_dependencies.items() + } + return self + + def enable_fd_hints(self, enabled: bool = True) -> "BatchSummarizer": + self.use_fd_hints = enabled + return self + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + def explain(self, df: pd.DataFrame) -> List[Explanation]: + """Generate explanations for the provided data frame.""" + + working_df = df.copy() + self._prepare(working_df) + explanations = self._run(working_df) + return self._finalize(explanations) + + # ------------------------------------------------------------------ + # Internal helpers shared by subclasses + # ------------------------------------------------------------------ + def _prepare(self, df: pd.DataFrame) -> None: + if self.outlier_column not in df.columns: + raise KeyError(f"Outlier column '{self.outlier_column}' not found in frame") + + if self.attributes is None: + self.attributes = [ + col + for col in df.columns + if col != self.outlier_column + ] + else: + missing = [col for col in self.attributes if col not in df.columns] + if missing: + raise KeyError(f"Attributes missing from frame: {missing}") + + for attribute in self.attributes: + if df[attribute].isnull().any(): + raise ValueError( + f"Attribute '{attribute}' contains null values which are not supported" + ) + + self._df = df + self._outlier_mask = df[self.outlier_column].astype(bool) + self._total_rows = len(df) + self._n_outliers = int(self._outlier_mask.sum()) + self._n_inliers = self._total_rows - self._n_outliers + if self._n_outliers == 0: + raise ValueError("At least one outlier is required for summarization") + if self._n_inliers == 0: + raise ValueError("At least one inlier is required for summarization") + + self._attribute_values = { + attr: sorted(df[attr].unique().tolist()) for attr in self.attributes + } + + if self.max_order is None: + self.max_order = len(self.attributes) + + def _finalize(self, explanations: List[Explanation]) -> List[Explanation]: + metric_name = self.ratio_metric + explanations = [ + exp for exp in explanations if self._metric_from_explanation(exp, metric_name) is not None + ] + + explanations.sort( + key=lambda exp: ( + self._metric_from_explanation(exp, metric_name) or -np.inf, + exp.support_outliers, + exp.risk_difference, + ), + reverse=True, + ) + if self.top_k is not None: + explanations = explanations[: self.top_k] + return explanations + + def _metric_from_explanation(self, explanation: Explanation, metric_name: str) -> Optional[float]: + if metric_name == "risk_ratio": + return explanation.risk_ratio + if metric_name == "risk_difference": + return explanation.risk_difference + if metric_name == "support_outliers": + return explanation.support_outliers + if metric_name == "support_total": + return explanation.support_total + raise ValueError(f"Unsupported ratio metric: {metric_name}") + + def _passes_thresholds(self, metrics: Dict[str, float]) -> bool: + return self._has_min_support(metrics) and self._satisfies_ratio(metrics) + + def _metric_value(self, metrics: Dict[str, float]) -> Optional[float]: + metric_name = self.ratio_metric + if metric_name == "risk_ratio": + return metrics["risk_ratio"] + if metric_name == "risk_difference": + return metrics["risk_difference"] + if metric_name == "support_outliers": + return metrics["support_outliers"] + if metric_name == "support_total": + return metrics["support_total"] + raise ValueError(f"Unsupported ratio metric: {metric_name}") + + def _has_min_support(self, metrics: Dict[str, float]) -> bool: + return metrics["support_outliers"] >= self.min_support + + def _satisfies_ratio(self, metrics: Dict[str, float]) -> bool: + metric_value = self._metric_value(metrics) + if metric_value is None: + return False + return metric_value >= self.min_ratio_metric + + def _compute_metrics(self, combination: Combination) -> Dict[str, float]: + assert self._df is not None and self._outlier_mask is not None + + mask = pd.Series(True, index=self._df.index) + for attribute, value in combination: + mask &= self._df[attribute] == value + + outlier_mask = mask & self._outlier_mask + inlier_mask = mask & (~self._outlier_mask) + + count_outliers = int(outlier_mask.sum()) + count_inliers = int(inlier_mask.sum()) + + support_outliers = ( + count_outliers / self._n_outliers if self._n_outliers else 0.0 + ) + support_inliers = ( + count_inliers / self._n_inliers if self._n_inliers else 0.0 + ) + support_total = (count_outliers + count_inliers) / self._total_rows + + if support_inliers == 0.0: + risk_ratio = float("inf") if support_outliers > 0 else 0.0 + else: + risk_ratio = support_outliers / support_inliers + risk_difference = support_outliers - support_inliers + + # Simple difference-of-proportions z-score as significance proxy. + pooled = (count_outliers + count_inliers) / self._total_rows + variance = pooled * (1 - pooled) + denom = 0.0 + if self._n_outliers: + denom += 1 / self._n_outliers + if self._n_inliers: + denom += 1 / self._n_inliers + if variance == 0.0 or denom == 0.0: + significance = 0.0 + else: + significance = (risk_difference) / np.sqrt(variance * denom) + + return { + "support_outliers": support_outliers, + "support_inliers": support_inliers, + "support_total": support_total, + "risk_ratio": risk_ratio, + "risk_difference": risk_difference, + "significance": significance, + } + + def _build_explanation( + self, combination: Combination, metrics: Dict[str, float] + ) -> Explanation: + attribute_map = OrderedDict(combination) + return Explanation( + attribute_map, + metrics["support_outliers"], + metrics["support_inliers"], + metrics["support_total"], + metrics["risk_ratio"], + metrics["risk_difference"], + metrics["significance"], + ) + + def _is_valid_combination(self, combination: Combination) -> bool: + if len(combination) > (self.max_order or len(self.attributes or [])): + return False + if self.use_fd_hints and self.functional_dependencies: + attrs = {attr for attr, _ in combination} + for determinant, dependents in self.functional_dependencies.items(): + if determinant in attrs and dependents & attrs: + return False + return True + + @staticmethod + def _normalize_combination( + combination: Sequence[Tuple[str, Any]] + ) -> Optional[Combination]: + seen: Dict[str, Any] = {} + for attribute, value in combination: + if attribute in seen and seen[attribute] != value: + return None + seen[attribute] = value + ordered = tuple(sorted(seen.items(), key=lambda item: item[0])) + return ordered + + def _subsets_are_frequent( + self, + combination: Combination, + frequent_subsets: Iterable[Combination], + ) -> bool: + frequent_set = set(frequent_subsets) + k = len(combination) + for subset in combinations(combination, k - 1): + normalized = self._normalize_combination(subset) + if normalized is None or normalized not in frequent_set: + return False + return True + + @abstractmethod + def _run(self, df: pd.DataFrame) -> List[Explanation]: + """Return all candidate explanations for ``df``.""" + + +__all__ = ["BatchSummarizer", "Combination"] diff --git a/macrobase_py/explainers/explanations.py b/macrobase_py/explainers/explanations.py new file mode 100644 index 000000000..01af239b4 --- /dev/null +++ b/macrobase_py/explainers/explanations.py @@ -0,0 +1,59 @@ +"""Explanation objects returned by summarizers.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Dict, Any, OrderedDict as OrderedDictType + + +@dataclass(order=True) +class Explanation: + """Container for summarizer output prepared for reporting. + + The ``sort_index`` attribute allows explanations to be sorted by their + utility metric while retaining all rich metadata for downstream + consumption. + """ + + sort_index: float = field(init=False, repr=False, compare=True) + attributes: "OrderedDictType[str, Any]" + support_outliers: float + support_inliers: float + support_total: float + risk_ratio: float + risk_difference: float + significance: float + + def __post_init__(self) -> None: + # Higher is better, so we store the negative value because ``dataclass`` + # ordering compares ``sort_index`` directly. + primary_metric = self.risk_ratio if self.risk_ratio != float("inf") else 1e12 + self.sort_index = -primary_metric + + def to_report_dict(self) -> Dict[str, Any]: + """Return a human-readable mapping ready for business reports.""" + + def _fmt_percent(value: float) -> str: + return f"{value * 100:.1f}%" + + rr = "∞" if self.risk_ratio == float("inf") else f"{self.risk_ratio:.2f}" + return { + "attributes": dict(self.attributes), + "outlier_support": _fmt_percent(self.support_outliers), + "inlier_support": _fmt_percent(self.support_inliers), + "overall_support": _fmt_percent(self.support_total), + "risk_ratio": rr, + "risk_difference": _fmt_percent(self.risk_difference), + "significance_z": f"{self.significance:.2f}", + } + + def __str__(self) -> str: + parts = [f"{k}={v}" for k, v in self.attributes.items()] + return ( + f"Explanation({', '.join(parts)}; " + f"outlier_support={self.support_outliers:.3f}, " + f"inlier_support={self.support_inliers:.3f}, " + f"risk_ratio={'∞' if self.risk_ratio == float('inf') else f'{self.risk_ratio:.2f}'}, " + f"risk_difference={self.risk_difference:.3f}, " + f"significance_z={self.significance:.2f})" + ) diff --git a/macrobase_py/explainers/heuristic.py b/macrobase_py/explainers/heuristic.py new file mode 100644 index 000000000..baa334b75 --- /dev/null +++ b/macrobase_py/explainers/heuristic.py @@ -0,0 +1,88 @@ +"""Heuristic search summarizer for high-dimensional spaces.""" + +from __future__ import annotations + +import heapq +from typing import Dict, List, Sequence, Set, Tuple + +import pandas as pd + +from .base import BatchSummarizer, Combination +from .explanations import Explanation + + +class HeuristicSummarizer(BatchSummarizer): + """Beam-search summarizer that favors combinations with high ratios.""" + + def __init__(self, *args, beam_width: int = 10, **kwargs) -> None: + super().__init__(*args, **kwargs) + self.beam_width = beam_width + + def set_beam_width(self, width: int) -> "HeuristicSummarizer": + self.beam_width = width + return self + + def _run(self, df: pd.DataFrame) -> List[Explanation]: + explanations: List[Explanation] = [] + candidate_heap: List[Tuple[float, Combination, Dict[str, float]]] = [] + + # Seed with singletons. + for attribute in self.attributes or []: + for value in self._attribute_values[attribute]: + combination = self._normalize_combination(((attribute, value),)) + if combination is None or not self._is_valid_combination(combination): + continue + metrics = self._compute_metrics(combination) + if not self._has_min_support(metrics): + continue + if self._satisfies_ratio(metrics): + explanation = self._build_explanation(combination, metrics) + explanations.append(explanation) + score = self._metric_value(metrics) + heapq.heappush(candidate_heap, (-(score or 0.0), combination, metrics)) + + visited: Set[Combination] = {item[1] for item in candidate_heap} + + current_depth = 1 + while candidate_heap and current_depth < (self.max_order or len(self.attributes or [])): + next_candidates: List[Tuple[float, Combination, Dict[str, float]]] = [] + for _ in range(min(self.beam_width, len(candidate_heap))): + score, base_combination, base_metrics = heapq.heappop(candidate_heap) + next_candidates.append((score, base_combination, base_metrics)) + + candidate_heap.clear() + expansions: List[Tuple[float, Combination, Dict[str, float]]] = [] + for _, combination, _ in next_candidates: + present_attrs = {attr for attr, _ in combination} + for attribute in self.attributes or []: + if attribute in present_attrs: + continue + for value in self._attribute_values[attribute]: + new_comb = self._normalize_combination(combination + ((attribute, value),)) + if new_comb is None: + continue + if new_comb in visited: + continue + if not self._is_valid_combination(new_comb): + continue + metrics = self._compute_metrics(new_comb) + if not self._has_min_support(metrics): + continue + if self._satisfies_ratio(metrics): + explanation = self._build_explanation(new_comb, metrics) + explanations.append(explanation) + score_value = self._metric_value(metrics) or 0.0 + heapq.heappush(expansions, (-(score_value), new_comb, metrics)) + visited.add(new_comb) + + # Prepare next level beam. + for _ in range(min(self.beam_width, len(expansions))): + heap_item = heapq.heappop(expansions) + heapq.heappush(candidate_heap, heap_item) + + current_depth += 1 + + return explanations + + +__all__ = ["HeuristicSummarizer"] diff --git a/tests/test_summarizers.py b/tests/test_summarizers.py new file mode 100644 index 000000000..74a81fbd5 --- /dev/null +++ b/tests/test_summarizers.py @@ -0,0 +1,110 @@ +import sys +from pathlib import Path + +import pandas as pd +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from macrobase_py.explainers import AprioriSummarizer, HeuristicSummarizer + + +def build_sample_frame() -> pd.DataFrame: + return pd.DataFrame( + [ + {"region": "US", "browser": "Chrome", "segment": "A", "zip": "94107", "is_outlier": 1}, + {"region": "US", "browser": "Chrome", "segment": "A", "zip": "94107", "is_outlier": 1}, + {"region": "US", "browser": "Chrome", "segment": "B", "zip": "94107", "is_outlier": 1}, + {"region": "US", "browser": "Firefox", "segment": "B", "zip": "94107", "is_outlier": 1}, + {"region": "US", "browser": "Chrome", "segment": "A", "zip": "94107", "is_outlier": 0}, + {"region": "US", "browser": "Chrome", "segment": "B", "zip": "94107", "is_outlier": 0}, + {"region": "US", "browser": "Safari", "segment": "A", "zip": "94107", "is_outlier": 0}, + {"region": "US", "browser": "Firefox", "segment": "B", "zip": "94107", "is_outlier": 0}, + {"region": "EU", "browser": "Chrome", "segment": "A", "zip": "75001", "is_outlier": 0}, + {"region": "EU", "browser": "Chrome", "segment": "B", "zip": "75001", "is_outlier": 0}, + {"region": "EU", "browser": "Firefox", "segment": "B", "zip": "75001", "is_outlier": 0}, + {"region": "ASIA", "browser": "Chrome", "segment": "A", "zip": "100-0001", "is_outlier": 0}, + {"region": "ASIA", "browser": "Chrome", "segment": "B", "zip": "100-0001", "is_outlier": 0}, + {"region": "ASIA", "browser": "Safari", "segment": "B", "zip": "100-0001", "is_outlier": 0}, + ] + ) + + +def test_apriori_summarizer_support_and_ratio(): + df = build_sample_frame() + summarizer = ( + AprioriSummarizer(min_support=0.2, min_ratio_metric=2.0, max_order=3) + .set_ratio_metric("risk_ratio") + .enable_fd_hints(True) + .set_functional_dependencies({"region": ["zip"]}) + ) + + explanations = summarizer.explain(df) + assert explanations, "Expected at least one explanation" + + top = explanations[0] + assert dict(top.attributes) == { + "browser": "Chrome", + "region": "US", + "segment": "A", + } + assert top.support_outliers == pytest.approx(0.5) + assert top.support_inliers == pytest.approx(0.1) + assert top.risk_ratio == pytest.approx(5.0) + assert top.risk_difference == pytest.approx(0.4) + assert top.significance > 0 + + report = top.to_report_dict() + assert report["risk_ratio"] == "5.00" + assert report["outlier_support"].endswith("%") + # Functional dependency hint should prevent redundant attribute usage. + for explanation in explanations: + assert "region" not in explanation.attributes or "zip" not in explanation.attributes + + assert any( + dict(exp.attributes) == {"browser": "Chrome", "region": "US"} + for exp in explanations + ) + + +def test_heuristic_matches_top_combo(): + df = build_sample_frame() + summarizer = HeuristicSummarizer( + min_support=0.2, + min_ratio_metric=2.0, + max_order=3, + beam_width=5, + ).set_ratio_metric("risk_ratio") + + explanations = summarizer.explain(df) + assert explanations, "Expected at least one explanation" + + top = explanations[0] + assert dict(top.attributes) == { + "browser": "Chrome", + "region": "US", + "segment": "A", + } + assert top.risk_ratio == pytest.approx(5.0) + assert top.support_outliers == pytest.approx(0.5) + + # Ensure report formatting is accessible for business users. + report = top.to_report_dict() + assert report["outlier_support"].startswith("50.0") + assert report["risk_difference"].startswith("40.0") + + +def test_invalid_inputs_raise(): + df = build_sample_frame() + summarizer = AprioriSummarizer() + + with pytest.raises(KeyError): + summarizer.set_attributes(["does_not_exist"]).explain(df) + + all_outlier = df.assign(is_outlier=1) + with pytest.raises(ValueError): + AprioriSummarizer().explain(all_outlier) + + all_inlier = df.assign(is_outlier=0) + with pytest.raises(ValueError): + AprioriSummarizer().explain(all_inlier)