feat: Add per-column metrics to summary

diffly/cli.py

@@ -11,6 +11,7 @@
 
 from ._compat import typer
 from ._utils import ABS_TOL_DEFAULT, ABS_TOL_TEMPORAL_DEFAULT, REL_TOL_DEFAULT
+from .metrics import _PRESETS
 
 app = typer.Typer()
 
@@ -129,9 +130,25 @@ def main(
             )
         ),
     ] = [],
+    metric: Annotated[
+        list[str],
+        typer.Option(
+            help=(
+                "Metric presets to display per numerical column. Repeatable. "
+                f"Available: {', '.join(_PRESETS)}."
+            )
+        ),
+    ] = [],
 ) -> None:
     """Compare two `parquet` files and print the comparison result."""
 
+    for name in metric:
+        if name not in _PRESETS:
+            raise typer.BadParameter(
+                f"Unknown metric: {name!r}. Available: {', '.join(_PRESETS)}."
+            )
+    metrics = {name: _PRESETS[name] for name in metric}
+
     comparison = compare_frames(
         pl.scan_parquet(left),
         pl.scan_parquet(right),
@@ -149,6 +166,7 @@ def main(
         right_name=right_name,
         slim=slim,
         hidden_columns=hidden_columns,
+        metrics=metrics,
     )
     if output_json:
         typer.echo(summary.to_json())

diffly/comparison.py

@@ -25,6 +25,7 @@
     lazy_len,
     make_and_validate_mapping,
 )
+from .metrics import Metric
 
 if TYPE_CHECKING:  # pragma: no cover
     # NOTE: We cannot import at runtime as we're otherwise running into circular
@@ -919,6 +920,7 @@ def summary(
         right_name: str = Side.RIGHT,
         slim: bool = False,
         hidden_columns: list[str] | None = None,
+        metrics: Mapping[str, Metric] | None = None,
     ) -> Summary:
         """Generate a summary of all aspects of the comparison.
 
@@ -948,6 +950,14 @@ def summary(
                 advanced users who are familiar with the summary format.
             hidden_columns: Columns for which no values are printed, e.g. because they
                 contain sensitive information.
+            metrics: Optional mapping from display label to a metric callable
+                ``(left_expr, right_expr) -> pl.Expr``. Each callable receives two
+                :class:`polars.Expr` referring to the left and right values of a single
+                numerical column across all joined rows, and must return a scalar
+                aggregation expression. See :mod:`diffly.metrics` for presets
+                (``mean``, ``median``, ``mean_absolute_deviation`` etc.). Metrics are
+                only computed for numerical columns. Prefer short labels — the
+                summary has a fixed width and many or long labels degrade rendering.
 
         Returns:
             A summary which can be printed or written to a file.
@@ -973,6 +983,7 @@ def summary(
             right_name=right_name,
             slim=slim,
             hidden_columns=hidden_columns,
+            metrics=metrics,
         )
 
     # ----------------------------------- UTILITIES ----------------------------------- #

diffly/metrics.py

@@ -0,0 +1,90 @@
+# Copyright (c) QuantCo 2025-2026
+# SPDX-License-Identifier: BSD-3-Clause
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+
+import polars as pl
+import polars.selectors as cs
+
+Metric = Callable[[pl.Expr, pl.Expr], pl.Expr]
+"""A metric is a callable mapping ``(left_expr, right_expr)`` to a scalar aggregation
+expression.
+
+The expressions refer to the left-side and right-side values of a single column across
+all joined rows.
+"""
+
+
+@dataclass(frozen=True)
+class _Metric:
+    """A metric paired with a column-applicability selector.
+
+    Internal only.
+    """
+
+    fn: Metric
+    selector: pl.Expr
+
+
+def _make_numeric_metric(metric: Metric) -> _Metric:
+    return _Metric(fn=metric, selector=cs.numeric())
+
+
+def mean(left: pl.Expr, right: pl.Expr) -> pl.Expr:
+    """Mean of ``right - left``."""
+    return (right - left).mean()
+
+
+def median(left: pl.Expr, right: pl.Expr) -> pl.Expr:
+    """Median of ``right - left``."""
+    return (right - left).median()
+
+
+def min(left: pl.Expr, right: pl.Expr) -> pl.Expr:
+    """Minimum of ``right - left``."""
+    return (right - left).min()
+
+
+def max(left: pl.Expr, right: pl.Expr) -> pl.Expr:
+    """Maximum of ``right - left``."""
+    return (right - left).max()
+
+
+def std(left: pl.Expr, right: pl.Expr) -> pl.Expr:
+    """Standard deviation of ``right - left``."""
+    return (right - left).std()
+
+
+def mean_absolute_deviation(left: pl.Expr, right: pl.Expr) -> pl.Expr:
+    """Mean of ``|right - left|``."""
+    return (right - left).abs().mean()
+
+
+def mean_relative_deviation(left: pl.Expr, right: pl.Expr) -> pl.Expr:
+    """Mean of ``|(right - left) / left|``. Yields ``inf`` or ``null`` where
+    ``left`` is zero."""
+    return ((right - left) / left).abs().mean()
+
+
+def quantile(q: float) -> Metric:
+    """Factory returning a metric that computes the ``q``-quantile of
+    ``right - left``."""
+
+    def _quantile(left: pl.Expr, right: pl.Expr) -> pl.Expr:
+        return (right - left).quantile(q)
+
+    return _quantile
+
+
+_PRESETS: dict[str, Metric] = {
+    "mean": mean,
+    "median": median,
+    "min": min,
+    "max": max,
+    "std": std,
+    "mean_absolute_deviation": mean_absolute_deviation,
+    "mean_relative_deviation": mean_relative_deviation,
+}

diffly/summary.py

@@ -6,11 +6,14 @@
 import dataclasses
 import io
 import json
+from collections.abc import Mapping
 from dataclasses import dataclass
 from datetime import date, datetime, timedelta
 from decimal import Decimal
 from typing import TYPE_CHECKING, Any, Literal, cast
 
+import polars as pl
+import polars.selectors as cs
 from rich import box
 from rich.columns import Columns as RichColumns
 from rich.console import Console, Group, RenderableType
@@ -20,6 +23,7 @@
 from rich.text import Text
 
 from ._utils import Side, capitalize_first
+from .metrics import Metric, _make_numeric_metric
 
 if TYPE_CHECKING:  # pragma: no cover
     from .comparison import DataFrameComparison
@@ -57,6 +61,7 @@ def __init__(
         right_name: str,
         slim: bool,
         hidden_columns: list[str] | None,
+        metrics: Mapping[str, Metric] | None,
     ):
         self.slim = slim
         self._data = _compute_summary_data(
@@ -69,6 +74,7 @@ def __init__(
             right_name=right_name,
             slim=slim,
             hidden_columns=hidden_columns,
+            metrics=metrics,
         )
 
     def format(self, pretty: bool | None = None) -> str:
@@ -565,13 +571,16 @@ def _section_columns(self) -> RenderableType:
         elif not columns:
             display_items.append(Text("All columns match perfectly.", style="italic"))
         else:
-            matches = Table(show_header=False)
+            metric_labels = self._data._metric_labels
+            matches = Table(show_header=bool(metric_labels))
             matches.add_column(
                 "Column",
                 max_width=COLUMN_SECTION_COLUMN_WIDTH,
                 overflow=OVERFLOW,
             )
             matches.add_column("Match Rate", justify="right")
+            for label in metric_labels:
+                matches.add_column(label, justify="right")
             has_top_changes_column = any(
                 c.changes is not None for c in columns if c.match_rate < 1
             )
@@ -583,6 +592,9 @@ def _section_columns(self) -> RenderableType:
                     Text(col.name, style="cyan"),
                     f"{_format_fraction_as_percentage(col.match_rate)}",
                 ]
+                for label in metric_labels:
+                    value = col.metrics.get(label) if col.metrics else None
+                    row_items.append(_format_metric_value(value))
                 if col.changes is not None:
                     change_lines = []
                     for change in col.changes:
@@ -703,6 +715,7 @@ class SummaryDataColumn:
     match_rate: float
     n_total_changes: int
     changes: list[SummaryDataColumnChange] | None
+    metrics: dict[str, Any] | None = None
 
 
 @dataclass
@@ -720,6 +733,7 @@ class SummaryData:
     _other_common_columns: list[str]
     _truncated_left_name: str
     _truncated_right_name: str
+    _metric_labels: list[str] = dataclasses.field(default_factory=list)
 
     def to_dict(self) -> dict[str, Any]:
         def _convert(obj: Any) -> Any:
@@ -758,6 +772,7 @@ def _compute_summary_data(
     right_name: str,
     slim: bool,
     hidden_columns: list[str] | None,
+    metrics: Mapping[str, Metric] | None,
 ) -> SummaryData:
     from .comparison import DataFrameComparison
 
@@ -823,6 +838,10 @@ def _validate_primary_key_hidden_columns() -> None:
             _truncated_right_name=truncated_right,
         )
 
+    metrics_resolved: dict[str, Metric] = dict(metrics or {})
+    metrics_by_column = _compute_column_metrics(comp, metrics_resolved)
+    metric_labels = list(metrics_resolved.keys())
+
     schemas = _compute_schemas(comp, slim)
     rows = _compute_rows(comp, slim)
     columns = _compute_columns(
@@ -831,6 +850,7 @@ def _validate_primary_key_hidden_columns() -> None:
         show_perfect_column_matches,
         top_k_changes_by_column,
         show_sample_primary_key_per_change,
+        metrics_by_column,
     )
     sample_rows_left_only, sample_rows_right_only = _compute_sample_rows(
         comp, sample_k_rows_only
@@ -850,6 +870,7 @@ def _validate_primary_key_hidden_columns() -> None:
         _other_common_columns=comp._other_common_columns,
         _truncated_left_name=truncated_left,
         _truncated_right_name=truncated_right,
+        _metric_labels=metric_labels,
     )
 
 
@@ -911,12 +932,54 @@ def _compute_rows(comp: DataFrameComparison, slim: bool) -> SummaryDataRows | No
     )
 
 
+def _compute_column_metrics(
+    comp: DataFrameComparison,
+    metrics: Mapping[str, Metric],
+) -> dict[str, dict[str, Any]]:
+    if not metrics:
+        return {}
+    if comp.primary_key is None or comp.num_rows_joined() == 0:
+        return {}
+
+    _metrics = {label: _make_numeric_metric(m) for label, m in metrics.items()}
+
+    def select_columns(selector: pl.Expr) -> set[str]:
+        left = set(cs.expand_selector(comp.left_schema, selector))
+        right = set(cs.expand_selector(comp.right_schema, selector))
+        return (left & right) & set(comp._other_common_columns)
+
+    metric_to_columns = {
+        label: select_columns(m.selector) for label, m in _metrics.items()
+    }
+
+    all_columns = sorted(set().union(*metric_to_columns.values()))
+    out: dict[str, dict[str, Any]] = {c: {} for c in all_columns}
+    if not all_columns:
+        return out
+
+    joined = comp.joined(lazy=True)
+    agg_exprs = [
+        metric.fn(
+            pl.col(f"{column}_{Side.LEFT}"),
+            pl.col(f"{column}_{Side.RIGHT}"),
+        ).alias(f"{label}__{column}")
+        for label, metric in _metrics.items()
+        for column in sorted(metric_to_columns[label])
+    ]
+    row = joined.select(agg_exprs).collect().row(0, named=True)
+    for label, columns in metric_to_columns.items():
+        for column in columns:
+            out[column][label] = row[f"{label}__{column}"]
+    return out
+
+
 def _compute_columns(
     comp: DataFrameComparison,
     slim: bool,
     show_perfect_column_matches: bool,
     top_k_changes_by_column: dict[str, int],
     show_sample_primary_key_per_change: bool,
+    metrics_by_column: dict[str, dict[str, Any]],
 ) -> list[SummaryDataColumn] | None:
     # NOTE: We can only compute column matches if there are primary key columns and at
     # least one joined row.
@@ -963,6 +1026,7 @@ def _compute_columns(
                 match_rate=rate,
                 n_total_changes=n_total_changes,
                 changes=changes,
+                metrics=metrics_by_column.get(col_name),
             )
         )
     return columns
@@ -1038,6 +1102,10 @@ def _format_fraction_as_percentage(fraction: float) -> str:
     return f"{percentage:.2f}%"
 
 
+def _yellow(raw: Any) -> str:
+    return f"[yellow]{raw}[/yellow]"
+
+
 def _format_value(value: Any) -> str:
     if isinstance(value, list):
         formatted = [_format_value(x) for x in value]
@@ -1054,7 +1122,15 @@ def _format_value(value: Any) -> str:
         raw = str(value)
     else:
         raw = value
-    return f"[yellow]{raw}[/yellow]"
+    return _yellow(raw)
+
+
+def _format_metric_value(value: Any) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, float):
+        return _yellow(f"{value:.4g}")
+    return _format_value(value)
 
 
 def _trim_whitespaces(s: str) -> str: