diff --git a/macrobase_py/__init__.py b/macrobase_py/__init__.py
index fa619c46b..86aa09f4e 100644
--- a/macrobase_py/__init__.py
+++ b/macrobase_py/__init__.py
@@ -1,3 +1,9 @@
+"""Utilities and classifiers for MacroBase Python components."""
+
+from . import classifiers
+
+__all__ = ["classifiers"]
+
 """Python tooling for MacroBase summarization algorithms."""
 
 from .explainers import (
diff --git a/macrobase_py/classifiers/__init__.py b/macrobase_py/classifiers/__init__.py
index 8fdcb478e..deb97e5dd 100644
--- a/macrobase_py/classifiers/__init__.py
+++ b/macrobase_py/classifiers/__init__.py
@@ -1,3 +1,21 @@
+"""Classifier implementations for MacroBase's Python tooling."""
+
+from .base import Classifier
+from .percentile import PercentileOutlierClassifier
+from .predicate import PredicateClassifier
+from .mean_shift import MeanShiftClassifier, CountMeanShiftClassifier
+from .aggregated import AggregatedQuantileClassifier, AggregatedArithmeticClassifier
+
+__all__ = [
+    "Classifier",
+    "PercentileOutlierClassifier",
+    "PredicateClassifier",
+    "MeanShiftClassifier",
+    "CountMeanShiftClassifier",
+    "AggregatedQuantileClassifier",
+    "AggregatedArithmeticClassifier",
+]
+
 """Module placeholder for macrobase_py.classifiers."""
 
 __all__: list[str] = []
diff --git a/macrobase_py/classifiers/aggregated.py b/macrobase_py/classifiers/aggregated.py
new file mode 100644
index 000000000..4a03418fe
--- /dev/null
+++ b/macrobase_py/classifiers/aggregated.py
@@ -0,0 +1,158 @@
+"""Classifiers designed for aggregated (cube) datasets."""
+
+from __future__ import annotations
+
+from typing import Sequence
+
+import numpy as np
+import pandas as pd
+
+from .base import Classifier, ensure_columns
+
+
+def _weighted_quantile(values: np.ndarray, weights: np.ndarray, quantile: float) -> float:
+    total = weights.sum()
+    if total <= 0:
+        raise ValueError("Weights must sum to a positive value.")
+    order = np.argsort(values)
+    values = values[order]
+    weights = weights[order]
+    cumulative = np.cumsum(weights)
+    cutoff = quantile * total
+    index = np.searchsorted(cumulative, cutoff, side="right")
+    index = np.clip(index, 0, len(values) - 1)
+    return float(values[index])
+
+
+class AggregatedQuantileClassifier(Classifier):
+    """Quantile based detector tailored to aggregated data."""
+
+    def __init__(
+        self,
+        metrics: Sequence[str],
+        *,
+        quantile: float = 0.95,
+        higher_is_outlier: bool = True,
+        count_column: str | None = "count",
+    ) -> None:
+        if not metrics:
+            raise ValueError("At least one metric column must be provided.")
+        if not (0.0 < quantile < 1.0):
+            raise ValueError("Quantile must be between 0 and 1 (exclusive).")
+        self.metrics = list(metrics)
+        self.quantile = float(quantile)
+        self.higher_is_outlier = higher_is_outlier
+        self.count_column = count_column
+        self.thresholds: dict[str, float] = {}
+
+    def fit(self, data: pd.DataFrame) -> "AggregatedQuantileClassifier":
+        ensure_columns(data, self.metrics)
+        weights = None
+        if self.count_column is not None and self.count_column in data.columns:
+            counts = data[self.count_column].to_numpy(dtype=float)
+            if np.any(counts < 0):
+                raise ValueError("Counts must be non-negative.")
+            weights = counts
+
+        for metric in self.metrics:
+            series = data[metric].to_numpy(dtype=float)
+            if weights is not None:
+                threshold = _weighted_quantile(series, weights, self.quantile)
+            else:
+                threshold = float(np.quantile(series, self.quantile))
+            if not self.higher_is_outlier:
+                threshold = float(np.quantile(series, 1.0 - self.quantile)) if weights is None else _weighted_quantile(series, weights, 1.0 - self.quantile)
+            self.thresholds[metric] = threshold
+
+        self._mark_fitted()
+        return self
+
+    def score(self, data: pd.DataFrame) -> pd.Series:
+        self._ensure_fitted()
+        ensure_columns(data, self.metrics)
+
+        def per_row(row: pd.Series) -> float:
+            deltas: list[float] = []
+            for metric in self.metrics:
+                value = float(row[metric])
+                threshold = self.thresholds[metric]
+                if self.higher_is_outlier:
+                    deltas.append(value - threshold)
+                else:
+                    deltas.append(threshold - value)
+            return float(np.max(deltas))
+
+        scores = data.apply(per_row, axis=1)
+        return scores
+
+    def predict(self, data: pd.DataFrame) -> pd.Series:
+        scores = self.score(data)
+        return scores > 0.0
+
+
+class AggregatedArithmeticClassifier(Classifier):
+    """Detect groups deviating from the global per-unit average."""
+
+    def __init__(
+        self,
+        metrics: Sequence[str],
+        *,
+        count_column: str = "count",
+        zscore_threshold: float = 2.0,
+    ) -> None:
+        if not metrics:
+            raise ValueError("At least one metric column must be provided.")
+        if zscore_threshold <= 0:
+            raise ValueError("z-score threshold must be positive.")
+        self.metrics = list(metrics)
+        self.count_column = count_column
+        self.zscore_threshold = float(zscore_threshold)
+        self._means: dict[str, float] = {}
+        self._stds: dict[str, float] = {}
+
+    def _per_unit(self, data: pd.DataFrame) -> pd.DataFrame:
+        ensure_columns(data, self.metrics)
+        ensure_columns(data, [self.count_column])
+        counts = data[self.count_column].to_numpy(dtype=float)
+        if np.any(counts <= 0):
+            raise ValueError("Counts must be positive for arithmetic classifier.")
+        values = data[self.metrics].to_numpy(dtype=float)
+        per_unit = values / counts[:, None]
+        return pd.DataFrame(per_unit, columns=self.metrics, index=data.index)
+
+    def fit(self, data: pd.DataFrame) -> "AggregatedArithmeticClassifier":
+        per_unit = self._per_unit(data)
+        counts = data[self.count_column].to_numpy(dtype=float)
+        weights = counts / counts.sum()
+
+        for metric in self.metrics:
+            values = per_unit[metric].to_numpy(dtype=float)
+            mean = float(np.dot(values, weights))
+            variance = float(np.dot((values - mean) ** 2, weights))
+            std = float(np.sqrt(variance)) if variance > 0 else 1e-9
+            self._means[metric] = mean
+            self._stds[metric] = std
+
+        self._mark_fitted()
+        return self
+
+    def score(self, data: pd.DataFrame) -> pd.Series:
+        self._ensure_fitted()
+        per_unit = self._per_unit(data)
+
+        def per_row(row: pd.Series) -> float:
+            zscores: list[float] = []
+            for metric in self.metrics:
+                std = self._stds[metric]
+                mean = self._means[metric]
+                zscore = abs((float(row[metric]) - mean) / std)
+                zscores.append(zscore - self.zscore_threshold)
+            return float(np.max(zscores))
+
+        scores = per_unit.apply(per_row, axis=1)
+        return scores
+
+    def predict(self, data: pd.DataFrame) -> pd.Series:
+        scores = self.score(data)
+        return scores > 0.0
+
diff --git a/macrobase_py/classifiers/base.py b/macrobase_py/classifiers/base.py
new file mode 100644
index 000000000..44a2d1c30
--- /dev/null
+++ b/macrobase_py/classifiers/base.py
@@ -0,0 +1,55 @@
+"""Base classes and helpers for MacroBase classifiers."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Iterable, Sequence
+
+import pandas as pd
+
+
+class Classifier(ABC):
+    """Abstract base classifier with fit/score/predict lifecycle."""
+
+    def __init__(self) -> None:
+        self._fitted: bool = False
+
+    @property
+    def fitted(self) -> bool:
+        """Whether the classifier has been trained."""
+
+        return self._fitted
+
+    def _mark_fitted(self) -> None:
+        self._fitted = True
+
+    def _ensure_fitted(self) -> None:
+        if not self._fitted:
+            raise RuntimeError("Classifier must be fitted before use.")
+
+    @abstractmethod
+    def fit(self, data: pd.DataFrame) -> "Classifier":
+        """Fit the classifier using the provided data."""
+
+    @abstractmethod
+    def score(self, data: pd.DataFrame) -> pd.Series:
+        """Compute outlier scores for the provided data."""
+
+    @abstractmethod
+    def predict(self, data: pd.DataFrame) -> pd.Series:
+        """Return boolean predictions indicating outliers."""
+
+
+def ensure_columns(data: pd.DataFrame, columns: Sequence[str]) -> None:
+    """Validate that the requested columns are present in *data*."""
+
+    missing = [column for column in columns if column not in data.columns]
+    if missing:
+        raise KeyError(f"Missing required columns: {', '.join(missing)}")
+
+
+def to_series(values: Iterable[float], index: pd.Index) -> pd.Series:
+    """Utility to wrap iterables into a :class:`pandas.Series`."""
+
+    return pd.Series(list(values), index=index)
+
diff --git a/macrobase_py/classifiers/mean_shift.py b/macrobase_py/classifiers/mean_shift.py
new file mode 100644
index 000000000..75ca2fcbb
--- /dev/null
+++ b/macrobase_py/classifiers/mean_shift.py
@@ -0,0 +1,160 @@
+"""Mean shift inspired density classifiers."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Iterable, Sequence
+
+import numpy as np
+import pandas as pd
+
+from .base import Classifier, ensure_columns, to_series
+
+
+@dataclass
+class _GaussianBandwidth:
+    value: np.ndarray
+
+
+class _GaussianKDE:
+    """Lightweight Gaussian KDE supporting optional sample weights."""
+
+    def __init__(self, bandwidth: float | Sequence[float] | None = None) -> None:
+        self.bandwidth = bandwidth
+        self.samples: np.ndarray | None = None
+        self.weights: np.ndarray | None = None
+        self._bw: _GaussianBandwidth | None = None
+
+    def fit(self, samples: np.ndarray, weights: np.ndarray | None = None) -> "_GaussianKDE":
+        if samples.ndim != 2:
+            raise ValueError("Samples must be a 2D array.")
+        self.samples = np.asarray(samples, dtype=float)
+        self.weights = None if weights is None else np.asarray(weights, dtype=float)
+        self._bw = _GaussianBandwidth(self._compute_bandwidth())
+        return self
+
+    def score_samples(self, samples: np.ndarray) -> np.ndarray:
+        if self.samples is None or self._bw is None:
+            raise RuntimeError("KDE must be fitted before scoring samples.")
+
+        query = np.asarray(samples, dtype=float)
+        diffs = (query[:, None, :] - self.samples[None, :, :]) / self._bw.value
+        exponent = -0.5 * np.sum(diffs * diffs, axis=2)
+        kernel_vals = np.exp(exponent)
+
+        if self.weights is not None:
+            weights = self.weights / self.weights.sum()
+            density = np.dot(kernel_vals, weights)
+        else:
+            density = kernel_vals.mean(axis=1)
+
+        norm = np.prod(self._bw.value) * np.power(2.0 * np.pi, self.samples.shape[1] / 2.0)
+        density = density / norm
+        return np.log(density + 1e-12)
+
+    def _compute_bandwidth(self) -> np.ndarray:
+        assert self.samples is not None
+        n, d = self.samples.shape
+        if isinstance(self.bandwidth, Sequence):
+            bw = np.asarray(list(self.bandwidth), dtype=float)
+            if bw.shape != (d,):
+                raise ValueError("Bandwidth sequence must have one value per feature.")
+            return np.where(bw > 0, bw, 1.0)
+        if isinstance(self.bandwidth, (int, float)) and self.bandwidth is not None:
+            bw_value = float(self.bandwidth)
+            if bw_value <= 0:
+                raise ValueError("Bandwidth must be positive.")
+            return np.full(d, bw_value)
+
+        std = self.samples.std(axis=0, ddof=1)
+        std = np.where(std > 0, std, 1.0)
+        factor = np.power(n, -1.0 / (d + 4))
+        return std * factor
+
+
+class MeanShiftClassifier(Classifier):
+    """Density based outlier detector using a Gaussian KDE approximation."""
+
+    def __init__(
+        self,
+        metrics: Sequence[str],
+        *,
+        bandwidth: float | Sequence[float] | None = None,
+        contamination: float = 0.05,
+    ) -> None:
+        if not metrics:
+            raise ValueError("At least one metric column must be provided.")
+        if not (0.0 < contamination < 1.0):
+            raise ValueError("Contamination must be within (0, 1).")
+
+        self.metrics = list(metrics)
+        self.bandwidth = bandwidth
+        self.contamination = contamination
+        self._kde = _GaussianKDE(bandwidth)
+        self.threshold: float | None = None
+
+    def _values(self, data: pd.DataFrame) -> np.ndarray:
+        ensure_columns(data, self.metrics)
+        return data[self.metrics].to_numpy(dtype=float)
+
+    def fit(self, data: pd.DataFrame) -> "MeanShiftClassifier":
+        samples = self._values(data)
+        self._kde.fit(samples)
+        log_density = self._kde.score_samples(samples)
+        self.threshold = float(np.quantile(log_density, self.contamination))
+        self._mark_fitted()
+        return self
+
+    def score(self, data: pd.DataFrame) -> pd.Series:
+        self._ensure_fitted()
+        assert self.threshold is not None
+        samples = self._values(data)
+        log_density = self._kde.score_samples(samples)
+        scores = self.threshold - log_density
+        return to_series(scores, data.index)
+
+    def predict(self, data: pd.DataFrame) -> pd.Series:
+        self._ensure_fitted()
+        assert self.threshold is not None
+        samples = self._values(data)
+        log_density = self._kde.score_samples(samples)
+        return to_series(log_density < self.threshold, data.index)
+
+
+class CountMeanShiftClassifier(MeanShiftClassifier):
+    """Mean shift variant that accounts for aggregated counts."""
+
+    def __init__(
+        self,
+        metrics: Sequence[str],
+        *,
+        count_column: str = "count",
+        bandwidth: float | Sequence[float] | None = None,
+        contamination: float = 0.05,
+    ) -> None:
+        super().__init__(metrics, bandwidth=bandwidth, contamination=contamination)
+        self.count_column = count_column
+
+    def _expanded_values(self, data: pd.DataFrame) -> np.ndarray:
+        ensure_columns(data, [self.count_column])
+        values = self._values(data)
+        counts = data[self.count_column].to_numpy(dtype=int)
+        if np.any(counts < 0):
+            raise ValueError("Counts must be non-negative.")
+        expanded: list[np.ndarray] = []
+        for value, count in zip(values, counts, strict=False):
+            if count == 0:
+                continue
+            expanded.extend([value] * int(count))
+        if not expanded:
+            raise ValueError("Aggregated data does not contain any observations.")
+        return np.vstack(expanded)
+
+    def fit(self, data: pd.DataFrame) -> "CountMeanShiftClassifier":
+        expanded = self._expanded_values(data)
+        self._kde.fit(expanded)
+        log_density = self._kde.score_samples(expanded)
+        self.threshold = float(np.quantile(log_density, self.contamination))
+        self._mark_fitted()
+        return self
+
diff --git a/macrobase_py/classifiers/percentile.py b/macrobase_py/classifiers/percentile.py
new file mode 100644
index 000000000..9d751f9e4
--- /dev/null
+++ b/macrobase_py/classifiers/percentile.py
@@ -0,0 +1,68 @@
+"""Percentile-based outlier detection."""
+
+from __future__ import annotations
+
+from typing import Iterable, Sequence
+
+import numpy as np
+import pandas as pd
+
+from .base import Classifier, ensure_columns, to_series
+
+
+class PercentileOutlierClassifier(Classifier):
+    """Flag rows whose metrics exceed a configurable percentile."""
+
+    def __init__(
+        self,
+        metrics: Sequence[str],
+        percentile: float = 95.0,
+        *,
+        higher_is_outlier: bool = True,
+    ) -> None:
+        if not metrics:
+            raise ValueError("At least one metric column must be provided.")
+        if not (0.0 < percentile < 100.0):
+            raise ValueError("Percentile must be between 0 and 100 (exclusive).")
+
+        self.metrics = list(metrics)
+        self.percentile = percentile / 100.0
+        self.higher_is_outlier = higher_is_outlier
+        self.thresholds: dict[str, float] = {}
+
+    def fit(self, data: pd.DataFrame) -> "PercentileOutlierClassifier":
+        ensure_columns(data, self.metrics)
+
+        for metric in self.metrics:
+            series = data[metric].dropna()
+            if series.empty:
+                raise ValueError(f"Metric '{metric}' does not contain any data.")
+
+            quantile = self.percentile if self.higher_is_outlier else 1.0 - self.percentile
+            self.thresholds[metric] = float(series.quantile(quantile))
+
+        self._mark_fitted()
+        return self
+
+    def _metric_scores(self, data: pd.DataFrame) -> Iterable[float]:
+        for _, row in data[self.metrics].iterrows():
+            deltas = []
+            for metric in self.metrics:
+                value = row[metric]
+                threshold = self.thresholds[metric]
+                if self.higher_is_outlier:
+                    deltas.append(value - threshold)
+                else:
+                    deltas.append(threshold - value)
+            yield float(np.max(deltas))
+
+    def score(self, data: pd.DataFrame) -> pd.Series:
+        self._ensure_fitted()
+        ensure_columns(data, self.metrics)
+
+        return to_series(self._metric_scores(data), data.index)
+
+    def predict(self, data: pd.DataFrame) -> pd.Series:
+        scores = self.score(data)
+        return scores > 0.0
+
diff --git a/macrobase_py/classifiers/predicate.py b/macrobase_py/classifiers/predicate.py
new file mode 100644
index 000000000..646ce7ad0
--- /dev/null
+++ b/macrobase_py/classifiers/predicate.py
@@ -0,0 +1,34 @@
+"""Classifier evaluating boolean predicates over dataframe columns."""
+
+from __future__ import annotations
+
+import pandas as pd
+
+from .base import Classifier
+
+
+class PredicateClassifier(Classifier):
+    """Evaluate a boolean predicate to classify rows."""
+
+    def __init__(self, predicate: str) -> None:
+        if not predicate:
+            raise ValueError("A predicate expression must be provided.")
+        self.predicate = predicate
+
+    def fit(self, data: pd.DataFrame) -> "PredicateClassifier":  # noqa: D401
+        """Predicate classifiers do not require fitting."""
+
+        self._mark_fitted()
+        return self
+
+    def score(self, data: pd.DataFrame) -> pd.Series:
+        self._ensure_fitted()
+        mask = data.eval(self.predicate)
+        if mask.dtype != bool:
+            mask = mask.astype(bool)
+        return mask.astype(float)
+
+    def predict(self, data: pd.DataFrame) -> pd.Series:
+        scores = self.score(data)
+        return scores.astype(bool)
+
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 000000000..00db3d1b4
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,8 @@
+import pathlib
+import sys
+
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
diff --git a/tests/test_classifiers.py b/tests/test_classifiers.py
new file mode 100644
index 000000000..62e0ef8c1
--- /dev/null
+++ b/tests/test_classifiers.py
@@ -0,0 +1,108 @@
+import numpy as np
+import pandas as pd
+
+from macrobase_py.classifiers import (
+    AggregatedArithmeticClassifier,
+    AggregatedQuantileClassifier,
+    CountMeanShiftClassifier,
+    MeanShiftClassifier,
+    PercentileOutlierClassifier,
+    PredicateClassifier,
+)
+
+
+def test_percentile_classifier_flags_high_values():
+    data = pd.DataFrame({"metric": [1, 2, 3, 4, 100]})
+    clf = PercentileOutlierClassifier(metrics=["metric"], percentile=80.0)
+    clf.fit(data)
+    scores = clf.score(data)
+    preds = clf.predict(data)
+    assert preds.iloc[-1]
+    assert scores.iloc[-1] > 0
+    assert not preds.iloc[0]
+    assert scores.iloc[0] <= 0
+
+
+def test_predicate_classifier_evaluates_expression():
+    data = pd.DataFrame({"value": [1, 6, 10], "flag": [0, 1, 1]})
+    clf = PredicateClassifier("value > 5 and flag == 1")
+    clf.fit(data)
+    preds = clf.predict(data)
+    scores = clf.score(data)
+    assert preds.tolist() == [False, True, True]
+    assert scores.tolist() == [0.0, 1.0, 1.0]
+
+
+def test_mean_shift_classifier_detects_outliers():
+    rng = np.random.default_rng(0)
+    normal = rng.normal(0, 1, size=(200, 2))
+    outliers = np.array([[5.0, 5.0], [6.0, 6.0]])
+    data = pd.DataFrame(np.vstack([normal, outliers]), columns=["x", "y"])
+
+    clf = MeanShiftClassifier(metrics=["x", "y"], contamination=0.02, bandwidth=1.0)
+    clf.fit(data)
+    preds = clf.predict(data)
+    scores = clf.score(data)
+
+    assert preds.iloc[-1]
+    assert preds.iloc[-2]
+    assert scores.iloc[-1] > 0
+    assert preds.sum() >= 2
+
+
+def test_count_mean_shift_classifier_uses_counts():
+    data = pd.DataFrame(
+        {
+            "x": [0.0, 0.1, 5.0],
+            "y": [0.0, -0.1, 5.0],
+            "count": [200, 180, 2],
+        }
+    )
+
+    clf = CountMeanShiftClassifier(metrics=["x", "y"], count_column="count", bandwidth=0.5, contamination=0.05)
+    clf.fit(data)
+    preds = clf.predict(data)
+    scores = clf.score(data)
+
+    assert preds.iloc[-1]
+    assert scores.iloc[-1] > 0
+    assert not preds.iloc[0]
+
+
+def test_aggregated_quantile_classifier_flags_high_group():
+    data = pd.DataFrame(
+        {
+            "group": list("ABCD"),
+            "count": [20, 25, 22, 5],
+            "metric": [1.0, 1.2, 0.9, 5.0],
+        }
+    )
+
+    clf = AggregatedQuantileClassifier(metrics=["metric"], quantile=0.9, higher_is_outlier=True, count_column="count")
+    clf.fit(data)
+    preds = clf.predict(data)
+    scores = clf.score(data)
+
+    assert preds.iloc[-1]
+    assert scores.iloc[-1] > 0
+    assert not preds.iloc[0]
+
+
+def test_aggregated_arithmetic_classifier_detects_ratio_shift():
+    data = pd.DataFrame(
+        {
+            "group": list("ABC"),
+            "count": [50, 60, 5],
+            "metric": [100.0, 120.0, 70.0],
+        }
+    )
+
+    clf = AggregatedArithmeticClassifier(metrics=["metric"], count_column="count", zscore_threshold=2.0)
+    clf.fit(data)
+    preds = clf.predict(data)
+    scores = clf.score(data)
+
+    assert preds.iloc[-1]
+    assert scores.iloc[-1] > 0
+    assert not preds.iloc[0]
+