feat(rec): reconciliation module with PFS pricer — tracks #340

Build the `rec` module to drive calculated payments toward 1:1 parity
with CMS-published ground-truth files. The notebook at
notebooks/pfs_calcs.py revealed our pfs.calcs.payment output diverges
from pfs.carrier_locality for some year × locality × HCPCS combos;
rec lets us measure the gap, track it year-over-year, and extend to
OPPS / IPPS / ASC / DMEPOS without touching the engine.

Design — open/closed:

- src/rec/base.py, engine.py, report.py, cli.py are closed for
  modification. Adding a new payment system = one new file under
  src/rec/pricers/ implementing the Pricer protocol + one line in
  rec.pricers.__init__.PRICERS.

- Pricer is a @runtime_checkable Protocol with ClassVar
  system/description/join_keys/compare_cols and three methods
  (ground_truth, calculated, years_available). isinstance() is used
  by the engine to validate conformance without subclassing.

- reconcile(pricer, con, year) returns a frozen Reconciliation
  dataclass with scalar counts, a per-row deltas polars DataFrame,
  exact/near match counts, and non-fatal warnings. Matched rows are
  classified as exact (≤ tolerance_cents) or near (≤ 1¢). The delta
  table contains <col>_gt / <col>_calc / <col>_delta + abs_max_delta
  + is_exact / is_near flags, sorted by worst delta first.

- rec.report.as_markdown / as_json format the Reconciliation for
  the CLI. markdown includes counts table and top-N deltas.

- src/cli/rec.py is a Typer sub-app registered under `stack rec`:
    stack rec list                    # registered pricers
    stack rec pfs --year 2025         # single year
    stack rec pfs                     # all available years
    stack rec pfs --format json --output out.json
  src/rec/__main__.py mirrors the same via `python -m rec`.

PfsPricer implementation (src/rec/pricers/pfs.py):

- ground_truth: SELECT from pfs.carrier_locality WHERE year = ?
- calculated:
    1. RULES[year].conversion_factor replaces the stale
       pfs.rvu.conv_factor column (rvu file values are unreliable;
       the notebook already did this override).
    2. Filters pfs.rvu to status_code IN ('A', 'T') so carrier-priced
       codes (status 'C') show up as ground_truth_only instead of
       being flagged as bugs.
    3. Cross-joins rvu × distinct(mac, locality) from gpci, then
       calls pfs.calcs.payment.payment() twice (facility=False /
       True), rounds to cents, and joins the two halves.
- join_keys: [year, mac, locality, hcpcs, mod]
- compare_cols: [non_fac_fee, fac_fee] — limiting-charge columns
  are omitted because pfs/pipe.py synthesises them from
  non_fac_fee × 1.0925 at load time; reconciling against a value
  we computed ourselves is circular. Documented in the module
  docstring as a tracked gap (#340 gap #3).

Wiring:

- pyproject.toml: new rec optional dep group (stack[conf,pfs] +
  duckdb + narwhals + typer), added to stack[all], module-name
  list extended to include "rec".
- stack.toml [storage]: rec = "data/rec" for report artifacts.
- src/cli/__init__.py registers rec_app under `stack rec`.

Tests (37, all green):

- tests/rec/conftest.py seeds an in-memory DuckDB with four rows
  designed to exercise the full match matrix: exact, 1¢ near, GT-only
  (carrier-priced), calc-only. Also defines a FakePricer class for
  engine tests that don't need DuckDB.
- tests/rec/test_base.py: protocol conformance, dataclass shape,
  pct_exact / is_perfect derived properties, frozen immutability.
- tests/rec/test_engine.py: counts, exact vs near classification,
  tolerance promotion, delta table shape and ordering, rejection of
  non-conforming objects.
- tests/rec/test_report.py: markdown section headers, JSON
  round-trip.
- tests/rec/pricers/test_pfs.py: registration, protocol conformance,
  ground_truth row count, status-code filter, cent rounding,
  summary counts against the seeded fixture, tolerance promotion,
  missing-year KeyError.

Notebook:

- notebooks/pfs_reconciliation.py — marimo interactive runner with
  year dropdown, tolerance slider, rendered summary markdown, delta
  table, and warnings panel. Mirrors the structure of pfs_calcs.py.

Gaps surfaced by this work (not fixed in this commit, to be tracked
as follow-up issues):

1. pfs.calcs.payment.payment() ignores RuleYear.budget_neutrality_adjustor.
   Silently correct for 2014–2025 since the value is 1.0, but a
   future non-1 year will drift.
2. _LIMITING_CHARGE_FACTOR = 1.0925 is a magic constant inside
   pfs/pipe.py at load time. Should be extracted to pfs.calcs and
   the synthesized columns should be superseded by an actual CMS
   source if one exists.
3. pfs.rvu.conv_factor is stale / often NULL. Single source of
   truth should be pfs.rules.RULES; the column could be dropped.
4. Carrier file doesn't capture the header's stated CF so we can't
   cross-check RULES[year].conversion_factor matches the file at
   ingest time.

notebooks/pfs_reconciliation.py

@@ -0,0 +1,118 @@
+import marimo
+
+__generated_with = "0.21.1"
+app = marimo.App(width="medium")
+
+
+@app.cell(hide_code=True)
+def _():
+    import marimo as mo
+
+    return (mo,)
+
+
+@app.cell(hide_code=True)
+def _(mo):
+    mo.md(
+        """
+    # PFS Reconciliation
+
+    Runs `rec.pricers.pfs.PfsPricer` against `pfs.carrier_locality`
+    for a chosen year and reports the per-row delta.  Goal is perfect
+    1:1 concordance.  Tracks **homelab/stack#340**.
+    """
+    )
+    return
+
+
+@app.cell(hide_code=True)
+def _():
+    from conf import connect
+    from rec.engine import reconcile
+    from rec.pricers.pfs import PfsPricer
+
+    con = connect.duckdb()
+    pricer = PfsPricer()
+    return con, pricer, reconcile
+
+
+@app.cell(hide_code=True)
+def _(con, mo, pricer):
+    _years = pricer.years_available(con)
+    if not _years:
+        year_picker = mo.ui.dropdown(
+            options={"(no data loaded)": 0}, value="(no data loaded)", label="Year"
+        )
+    else:
+        year_picker = mo.ui.dropdown(
+            options={str(y): y for y in _years},
+            value=str(_years[-1]),
+            label="Year",
+        )
+
+    tolerance = mo.ui.slider(
+        start=0, stop=10, step=1, value=0, label="Tolerance (cents)"
+    )
+
+    mo.hstack([year_picker, tolerance], justify="start", gap=1)
+    return tolerance, year_picker
+
+
+@app.cell
+def _(con, mo, pricer, reconcile, tolerance, year_picker):
+    _year = int(year_picker.value) if year_picker.value else 0
+    if _year == 0:
+        result = None
+        mo.md("*Load `pfs.rvu`, `pfs.gpci`, and `pfs.carrier_locality` first.*")
+    else:
+        result = reconcile(pricer, con, _year, tolerance_cents=int(tolerance.value))
+        mo.md(result.summary_md(top_n=25))
+    return (result,)
+
+
+@app.cell(hide_code=True)
+def _(mo):
+    mo.md(
+        """
+    ## Delta table
+
+    Every row in the outer join, sorted by the largest absolute delta.
+    ``is_exact`` uses the tolerance above; ``is_near`` is always a 1¢
+    window.  Null ``fee_gt`` means the row is calculated-only; null
+    ``fee_calc`` means ground-truth-only.
+    """
+    )
+    return
+
+
+@app.cell
+def _(result):
+    _deltas = result.deltas if result is not None else None
+    _deltas
+    return
+
+
+@app.cell(hide_code=True)
+def _(mo):
+    mo.md(
+        """
+    ## Warnings
+
+    Non-fatal issues surfaced by the engine — duplicate join keys,
+    missing columns, rule lookup failures, etc.
+    """
+    )
+    return
+
+
+@app.cell
+def _(mo, result):
+    if result is None or not result.warnings:
+        mo.md("*(none)*")
+    else:
+        mo.md("\n".join(f"- {w}" for w in result.warnings))
+    return
+
+
+if __name__ == "__main__":
+    app.run()

pyproject.toml

@@ -71,6 +71,13 @@ pfs = [
     "duckdb>=1.0.0",
     "narwhals>=2.17.0",
 ]
+rec = [
+    "stack[conf]",
+    "stack[pfs]",
+    "duckdb>=1.0.0",
+    "narwhals>=2.17.0",
+    "typer>=0.24.1",
+]
 rex = [
     "stack[conf]",
     "narwhals>=2.17.0",
@@ -118,6 +125,7 @@ all = [
     "stack[cms]",
     "stack[perf]",
     "stack[pfs]",
+    "stack[rec]",
     "stack[rex]",
     "stack[lake]",
 ]
@@ -160,6 +168,6 @@ markers = [
 ]
 
 [tool.uv.build-backend]
-module-name = ["aco", "api", "bcda", "bib", "bls", "ccw", "cli", "cms", "conf", "opps", "perf", "pfs", "rex", "sem"]
+module-name = ["aco", "api", "bcda", "bib", "bls", "ccw", "cli", "cms", "conf", "opps", "perf", "pfs", "rec", "rex", "sem"]
 namespace = true
 source-exclude = ["compose.yml","infra/**","data/**","notebooks/**","tuva/**","assets/**","docs/**","dev/**","bundle/**","cloud/**"]

src/cli/__init__.py

@@ -21,6 +21,7 @@ from cli.health import health
 from cli.lake import app as lake_app
 from cli.load import app as load_app
 from cli.perf import app as perf_app
+from cli.rec import app as rec_app
 from cli.run import run
 from cli.validate import validate
 
@@ -41,6 +42,11 @@ app.add_typer(db_app, name="db", help="DuckDB utilities.")
 app.add_typer(docs_app, name="docs", help="Documentation generation.")
 app.add_typer(api_app, name="api", help="API server.")
 app.add_typer(perf_app, name="perf", help="Pipeline telemetry utilities.")
+app.add_typer(
+    rec_app,
+    name="rec",
+    help="Reconcile calculated payments against CMS ground-truth files.",
+)
 
 
 def main() -> None:

src/cli/rec.py

@@ -0,0 +1,143 @@
+"""stack rec — reconcile calculated payments against CMS ground truth.
+
+Every pricer under ``rec.pricers`` is exposed as a subcommand.  Run::
+
+    uv run stack rec list                    # show registered pricers
+    uv run stack rec pfs --year 2025          # PFS reconciliation
+    uv run stack rec pfs                      # all available years
+    uv run stack rec pfs --year 2025 --format json
+
+Tracks: homelab/stack#340
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Annotated
+
+import typer
+
+app = typer.Typer(
+    name="rec",
+    no_args_is_help=True,
+    help="Reconcile calculated payments against CMS ground-truth files.",
+)
+
+
+@app.command("list")
+def list_pricers() -> None:
+    """List registered pricers and their compare columns."""
+    from rec.pricers import PRICERS
+
+    if not PRICERS:
+        typer.echo("No pricers registered.")
+        raise typer.Exit(0)
+
+    typer.echo("Registered pricers:")
+    typer.echo("")
+    for name, cls in sorted(PRICERS.items()):
+        typer.echo(f"  {name}")
+        typer.echo(f"    {cls.description}")
+        typer.echo(f"    join:    {', '.join(cls.join_keys)}")
+        typer.echo(f"    compare: {', '.join(cls.compare_cols)}")
+        typer.echo("")
+
+
+@app.command("pfs")
+def reconcile_pfs(
+    year: Annotated[
+        int | None,
+        typer.Option("--year", "-y", help="Year to reconcile; omit for all."),
+    ] = None,
+    format: Annotated[
+        str,
+        typer.Option("--format", "-f", help="Output format: md | json."),
+    ] = "md",
+    output: Annotated[
+        Path | None,
+        typer.Option("--output", "-o", help="Write to file instead of stdout."),
+    ] = None,
+    tolerance_cents: Annotated[
+        int,
+        typer.Option(
+            "--tolerance",
+            "-t",
+            help="Cents of tolerance for an 'exact' match.",
+        ),
+    ] = 0,
+) -> None:
+    """Reconcile PFS calculated fees against carrier_locality."""
+    _run_pricer(
+        "pfs",
+        year=year,
+        format=format,
+        output=output,
+        tolerance_cents=tolerance_cents,
+    )
+
+
+def _run_pricer(
+    name: str,
+    *,
+    year: int | None,
+    format: str,
+    output: Path | None,
+    tolerance_cents: int,
+) -> None:
+    from conf import connect
+    from rec.engine import reconcile, reconcile_all_years
+    from rec.pricers import PRICERS
+
+    if name not in PRICERS:
+        typer.echo(
+            f"Unknown pricer {name!r}. Known: {', '.join(PRICERS)}",
+            err=True,
+        )
+        raise typer.Exit(2)
+    if format not in ("md", "json"):
+        typer.echo(f"Unknown format {format!r}; use md or json", err=True)
+        raise typer.Exit(2)
+
+    pricer = PRICERS[name]()
+    con = connect.duckdb()
+
+    if year is None:
+        results = reconcile_all_years(pricer, con, tolerance_cents=tolerance_cents)
+        if not results:
+            typer.echo(
+                f"No years available for {name}. "
+                f"Load pfs.rvu / pfs.gpci / pfs.carrier_locality first.",
+                err=True,
+            )
+            raise typer.Exit(1)
+    else:
+        results = {year: reconcile(pricer, con, year, tolerance_cents=tolerance_cents)}
+
+    rendered = _render(results, format)
+    if output is None:
+        sys.stdout.write(rendered)
+        if not rendered.endswith("\n"):
+            sys.stdout.write("\n")
+    else:
+        output.parent.mkdir(parents=True, exist_ok=True)
+        output.write_text(rendered)
+        typer.echo(f"Wrote {output}")
+
+
+def _render(results: dict[int, object], format: str) -> str:
+    from rec.report import as_json, as_markdown
+
+    if format == "md":
+        return "\n\n---\n\n".join(
+            as_markdown(r)
+            for _, r in sorted(results.items())  # type: ignore[arg-type]
+        )
+    # JSON: list of per-year objects
+    import json
+
+    items = [
+        json.loads(as_json(r))  # type: ignore[arg-type]
+        for _, r in sorted(results.items())
+    ]
+    return json.dumps(items, indent=2, default=str)

src/rec/__init__.py

@@ -0,0 +1,81 @@
+"""rec — payment reconciliation against CMS ground-truth files.
+
+Runs arbitrary "pricer" implementations against CMS-published payment
+files and reports the row-level delta, iterating toward exact 1:1
+parity.  Generalises beyond PFS: future pricers will cover OPPS,
+IPPS, ASC, DMEPOS, and ESRD.
+
+Architecture — open/closed by design
+------------------------------------
+
+::
+
+    CMS carrier file         Stack payment calc
+    (ground truth)           (rules + table + calcs)
+            │                        │
+            ▼                        ▼
+        ┌───────────────────────────────┐
+        │  rec.engine.reconcile()       │   ← closed for modification
+        │    join on Pricer.join_keys   │
+        │    diff on Pricer.compare_cols│
+        └───────────────────────────────┘
+                        │
+                        ▼
+               Reconciliation
+               (dataclass)
+                        │
+                        ▼
+                 rec.report.as_markdown / as_json
+
+    ┌─────────────────────────────────────┐
+    │  Pricer protocol — open for         │
+    │  extension via rec.pricers.PRICERS  │
+    │                                     │
+    │    pfs     ← rec.pricers.pfs        │
+    │    opps    ← (future)               │
+    │    ipps    ← (future)               │
+    │    asc     ← (future)               │
+    │    dmepos  ← (future)               │
+    └─────────────────────────────────────┘
+
+Adding a new payment system means writing a single file under
+``src/rec/pricers/<system>.py`` that implements the ``Pricer`` protocol
+and registering it in ``rec.pricers.__init__.PRICERS``.  Nothing in
+``base.py``, ``engine.py``, ``report.py``, or ``cli.py`` changes.
+
+Usage
+-----
+
+Run reconciliation from Python::
+
+    from conf import connect
+    from rec import reconcile
+    from rec.pricers.pfs import PfsPricer
+
+    con = connect.duckdb()
+    result = reconcile(PfsPricer(), con, year=2025)
+    print(result.summary_md())
+
+Or from the CLI::
+
+    uv run stack rec pfs --year 2025
+    uv run stack rec list
+
+Tracks: homelab/stack#340
+"""
+
+from __future__ import annotations
+
+from rec.base import Pricer, Reconciliation
+from rec.engine import reconcile, reconcile_all_years
+from rec.pricers import PRICERS
+from rec.pricers.pfs import PfsPricer
+
+__all__ = [
+    "PRICERS",
+    "PfsPricer",
+    "Pricer",
+    "Reconciliation",
+    "reconcile",
+    "reconcile_all_years",
+]

src/rec/__main__.py

@@ -0,0 +1,8 @@
+"""``python -m rec`` entry point — delegates to ``stack rec`` Typer app."""
+
+from __future__ import annotations
+
+from cli.rec import app
+
+if __name__ == "__main__":
+    app()

src/rec/base.py

@@ -0,0 +1,148 @@
+"""Pricer protocol and Reconciliation result dataclass.
+
+This module is **closed for modification**.  Adding a new payment
+system does not require editing anything here — implement the
+``Pricer`` protocol in ``src/rec/pricers/<system>.py`` and register
+the class in ``rec.pricers.__init__.PRICERS``.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import ClassVar, Protocol, runtime_checkable
+
+import duckdb
+import polars as pl
+
+
+@runtime_checkable
+class Pricer(Protocol):
+    """Protocol for a payment-system pricer.
+
+    A pricer compares one payment system's **ground-truth** rates
+    (published by CMS) against the **calculated** rates produced by
+    applying the annual Final Rule to the underlying inputs (RVUs,
+    weights, wage indices, etc.).
+
+    Implementations must set the three ``ClassVar`` fields and
+    provide three methods.  ``@runtime_checkable`` lets the engine
+    validate a pricer with ``isinstance(pricer, Pricer)`` without
+    forcing subclassing.
+
+    Class attributes
+    ----------------
+    system
+        Short identifier — ``"pfs"``, ``"opps"``, ``"asc"``, etc.
+    description
+        Human-readable one-liner for the CLI registry.
+    join_keys
+        Column names that uniquely identify a row on both sides of
+        the reconciliation.  The engine joins on these.
+    compare_cols
+        Dollar-valued columns whose deltas are computed.  Every
+        column listed here must exist in both ``ground_truth`` and
+        ``calculated`` outputs.
+    """
+
+    system: ClassVar[str]
+    description: ClassVar[str]
+    join_keys: ClassVar[list[str]]
+    compare_cols: ClassVar[list[str]]
+
+    def ground_truth(self, con: duckdb.DuckDBPyConnection, year: int) -> pl.DataFrame:
+        """Return the CMS-published rates for a given year."""
+        ...
+
+    def calculated(self, con: duckdb.DuckDBPyConnection, year: int) -> pl.DataFrame:
+        """Return rates computed from rules + inputs for a given year."""
+        ...
+
+    def years_available(self, con: duckdb.DuckDBPyConnection) -> list[int]:
+        """Return years where BOTH ground-truth and calculated data exist."""
+        ...
+
+
+@dataclass(frozen=True)
+class Reconciliation:
+    """Result of a single year's reconciliation for one pricer.
+
+    Scalar counts are accompanied by a ``deltas`` polars DataFrame
+    containing one row per join-key combination with per-column
+    deltas and an ``is_exact`` / ``abs_max_delta`` pair for sorting.
+
+    The dataclass is frozen so instances are safely shareable and
+    can be hashed / compared in tests.
+    """
+
+    system: str
+    """Pricer system identifier (``"pfs"``, etc.)."""
+
+    year: int
+    """Calendar year reconciled."""
+
+    ground_truth_rows: int
+    """Row count from the pricer's ``ground_truth`` query."""
+
+    calculated_rows: int
+    """Row count from the pricer's ``calculated`` query."""
+
+    matched_rows: int
+    """Rows present on BOTH sides after joining on ``join_keys``."""
+
+    ground_truth_only: int
+    """Rows only on the ground-truth side (missing calculation)."""
+
+    calculated_only: int
+    """Rows only on the calculated side (missing from CMS file)."""
+
+    exact_matches: int
+    """Matched rows where every ``compare_col`` is equal to the cent."""
+
+    near_matches: int
+    """Matched rows where every ``compare_col`` differs by ≤ 1 cent."""
+
+    deltas: pl.DataFrame
+    """Per-row delta table.  Columns: join_keys + for each compare col
+    ``<col>_gt`` / ``<col>_calc`` / ``<col>_delta``, plus ``is_exact``,
+    ``is_near``, and ``abs_max_delta``."""
+
+    tolerance_cents: int = 0
+    """Tolerance (in cents) used when computing ``exact_matches``.
+    ``near_matches`` always uses 1 cent."""
+
+    warnings: tuple[str, ...] = field(default_factory=tuple)
+    """Non-fatal warnings surfaced during reconciliation (duplicate
+    join-key rows, null compare columns, missing ``RULES[year]``,
+    etc.)."""
+
+    # ── Derived views ─────────────────────────────────────────────
+
+    @property
+    def pct_exact(self) -> float:
+        """Percentage of matched rows that are exact to the cent."""
+        if self.matched_rows == 0:
+            return 0.0
+        return 100.0 * self.exact_matches / self.matched_rows
+
+    @property
+    def is_perfect(self) -> bool:
+        """True when every row on both sides is matched and exact."""
+        return (
+            self.matched_rows > 0
+            and self.matched_rows == self.ground_truth_rows == self.calculated_rows
+            and self.exact_matches == self.matched_rows
+        )
+
+    # ── Convenience formatters (delegate to rec.report) ───────────
+
+    def summary_md(self, *, top_n: int = 20) -> str:
+        """Render as markdown.  See ``rec.report.as_markdown``."""
+        from rec.report import as_markdown
+
+        return as_markdown(self, top_n=top_n)
+
+    def to_json(self, *, top_n: int = 20) -> str:
+        """Render as JSON.  See ``rec.report.as_json``."""
+        from rec.report import as_json
+
+        return as_json(self, top_n=top_n)

src/rec/engine.py

@@ -0,0 +1,213 @@
+"""Generic reconciliation engine.
+
+Takes a ``Pricer`` instance, runs its ``ground_truth`` and
+``calculated`` methods for a year, full-outer-joins the results on
+``Pricer.join_keys``, diffs the ``Pricer.compare_cols``, and returns
+a ``Reconciliation``.
+
+This module is **closed for modification** — the logic works for
+every pricer that conforms to the protocol.
+"""
+
+from __future__ import annotations
+
+import duckdb
+import polars as pl
+
+from rec.base import Pricer, Reconciliation
+
+
+def reconcile(
+    pricer: Pricer,
+    con: duckdb.DuckDBPyConnection,
+    year: int,
+    *,
+    tolerance_cents: int = 0,
+) -> Reconciliation:
+    """Reconcile one year of a pricer's ground-truth vs calculated.
+
+    Parameters
+    ----------
+    pricer
+        A ``Pricer``-conforming instance (validated at runtime).
+    con
+        DuckDB connection with the pricer's input tables loaded.
+    year
+        Calendar year to reconcile.
+    tolerance_cents
+        Matched rows whose per-column deltas are all ≤ this many
+        cents count as ``exact_matches``.  Default ``0`` requires
+        cent-perfect parity.  ``near_matches`` is always computed at
+        a 1-cent tolerance.
+
+    Returns
+    -------
+    Reconciliation
+        Frozen dataclass with scalar stats and a per-row ``deltas``
+        DataFrame.
+    """
+    if not isinstance(pricer, Pricer):
+        raise TypeError(f"{type(pricer).__name__} does not conform to rec.base.Pricer")
+
+    warnings: list[str] = []
+
+    gt = pricer.ground_truth(con, year)
+    calc = pricer.calculated(con, year)
+
+    _check_columns(gt, pricer.join_keys + pricer.compare_cols, "ground_truth", warnings)
+    _check_columns(calc, pricer.join_keys + pricer.compare_cols, "calculated", warnings)
+
+    _check_unique_keys(gt, pricer.join_keys, "ground_truth", warnings)
+    _check_unique_keys(calc, pricer.join_keys, "calculated", warnings)
+
+    # Normalise dtypes on join keys to avoid silent join failures
+    # when one side has Int64 and the other Utf8 for the same logical
+    # identifier.  Cast everything to string — join keys are always
+    # categorical for our purposes.
+    gt = _stringify_keys(gt, pricer.join_keys)
+    calc = _stringify_keys(calc, pricer.join_keys)
+
+    # Rename compare columns so the left/right survive the outer join.
+    gt_renamed = gt.rename({c: f"{c}_gt" for c in pricer.compare_cols}).select(
+        pricer.join_keys + [f"{c}_gt" for c in pricer.compare_cols]
+    )
+
+    calc_renamed = calc.rename({c: f"{c}_calc" for c in pricer.compare_cols}).select(
+        pricer.join_keys + [f"{c}_calc" for c in pricer.compare_cols]
+    )
+
+    joined = gt_renamed.join(
+        calc_renamed,
+        on=pricer.join_keys,
+        how="full",
+        coalesce=True,
+    )
+
+    # Side membership flags — True when at least one compare column on
+    # that side is non-null.
+    in_gt = pl.any_horizontal(
+        [pl.col(f"{c}_gt").is_not_null() for c in pricer.compare_cols]
+    )
+    in_calc = pl.any_horizontal(
+        [pl.col(f"{c}_calc").is_not_null() for c in pricer.compare_cols]
+    )
+
+    joined = joined.with_columns(
+        in_gt.alias("_in_gt"),
+        in_calc.alias("_in_calc"),
+    )
+
+    # Per-column deltas (rounded to cents).  Null on either side → null delta.
+    delta_exprs = []
+    for c in pricer.compare_cols:
+        delta_exprs.append(
+            ((pl.col(f"{c}_calc") - pl.col(f"{c}_gt")).round(2)).alias(f"{c}_delta")
+        )
+    joined = joined.with_columns(delta_exprs)
+
+    # abs_max_delta — maximum absolute delta across compare cols (null if
+    # any delta is null, i.e. row not matched on that column).
+    joined = joined.with_columns(
+        pl.max_horizontal(
+            [pl.col(f"{c}_delta").abs() for c in pricer.compare_cols]
+        ).alias("abs_max_delta")
+    )
+
+    tolerance_dollars = tolerance_cents / 100.0
+    joined = joined.with_columns(
+        (
+            pl.col("_in_gt")
+            & pl.col("_in_calc")
+            & (pl.col("abs_max_delta") <= tolerance_dollars)
+        ).alias("is_exact"),
+        (
+            pl.col("_in_gt") & pl.col("_in_calc") & (pl.col("abs_max_delta") <= 0.01)
+        ).alias("is_near"),
+    )
+
+    # Scalar stats
+    ground_truth_rows = int(joined.filter(pl.col("_in_gt")).height)
+    calculated_rows = int(joined.filter(pl.col("_in_calc")).height)
+    matched_rows = int(joined.filter(pl.col("_in_gt") & pl.col("_in_calc")).height)
+    ground_truth_only = int(
+        joined.filter(pl.col("_in_gt") & ~pl.col("_in_calc")).height
+    )
+    calculated_only = int(joined.filter(~pl.col("_in_gt") & pl.col("_in_calc")).height)
+    exact_matches = int(joined.filter(pl.col("is_exact")).height)
+    near_matches = int(joined.filter(pl.col("is_near")).height)
+
+    deltas = joined.drop(["_in_gt", "_in_calc"]).sort(
+        by=["abs_max_delta", *pricer.join_keys],
+        descending=[True, *([False] * len(pricer.join_keys))],
+        nulls_last=True,
+    )
+
+    return Reconciliation(
+        system=pricer.system,
+        year=year,
+        ground_truth_rows=ground_truth_rows,
+        calculated_rows=calculated_rows,
+        matched_rows=matched_rows,
+        ground_truth_only=ground_truth_only,
+        calculated_only=calculated_only,
+        exact_matches=exact_matches,
+        near_matches=near_matches,
+        deltas=deltas,
+        tolerance_cents=tolerance_cents,
+        warnings=tuple(warnings),
+    )
+
+
+def reconcile_all_years(
+    pricer: Pricer,
+    con: duckdb.DuckDBPyConnection,
+    *,
+    tolerance_cents: int = 0,
+) -> dict[int, Reconciliation]:
+    """Run ``reconcile`` for every year the pricer reports available."""
+    years = pricer.years_available(con)
+    return {
+        y: reconcile(pricer, con, y, tolerance_cents=tolerance_cents) for y in years
+    }
+
+
+# ── Internal helpers ──────────────────────────────────────────────
+
+
+def _check_columns(
+    df: pl.DataFrame,
+    required: list[str],
+    side: str,
+    warnings: list[str],
+) -> None:
+    missing = [c for c in required if c not in df.columns]
+    if missing:
+        warnings.append(
+            f"{side}: missing required columns {missing} (pricer contract violated)"
+        )
+
+
+def _check_unique_keys(
+    df: pl.DataFrame,
+    keys: list[str],
+    side: str,
+    warnings: list[str],
+) -> None:
+    if df.height == 0:
+        return
+    key_cols = [k for k in keys if k in df.columns]
+    if not key_cols:
+        return
+    dup_count = df.height - df.unique(subset=key_cols).height
+    if dup_count > 0:
+        warnings.append(
+            f"{side}: {dup_count} duplicate rows on join keys {key_cols} "
+            f"(left-join will silently drop)"
+        )
+
+
+def _stringify_keys(df: pl.DataFrame, keys: list[str]) -> pl.DataFrame:
+    casts = [
+        pl.col(k).cast(pl.Utf8).fill_null("").alias(k) for k in keys if k in df.columns
+    ]
+    return df.with_columns(casts) if casts else df

src/rec/pricers/__init__.py

@@ -0,0 +1,22 @@
+"""Registry of payment-system pricers.
+
+Adding a new payment system:
+
+1. Create ``src/rec/pricers/<system>.py`` implementing the
+   ``rec.base.Pricer`` protocol.
+2. Add a single line to ``PRICERS`` below.
+
+Nothing else changes — the engine, report, and CLI all consume the
+registry without knowing about specific systems.
+"""
+
+from __future__ import annotations
+
+from rec.base import Pricer
+from rec.pricers.pfs import PfsPricer
+
+PRICERS: dict[str, type[Pricer]] = {
+    PfsPricer.system: PfsPricer,
+}
+
+__all__ = ["PRICERS", "PfsPricer"]

src/rec/pricers/pfs.py

@@ -0,0 +1,182 @@
+"""PFS pricer — Physician Fee Schedule reconciliation.
+
+Ground truth: ``pfs.carrier_locality`` rows loaded from CMS carrier
+files (one row per year × locality × HCPCS × modifier).
+
+Calculated: join ``pfs.rvu`` × ``pfs.gpci`` for the year, override
+``conv_factor`` with ``RULES[year].conversion_factor``, and call
+``pfs.calcs.payment.payment()`` once for facility and once for
+non-facility.
+
+Status-code filter:
+    Only status ``A`` (algorithm-priced) and ``T`` (paid under
+    algorithm when more than one service on same day) are expected
+    to reconcile to the cent.  Carrier-priced codes (``C``), bundled
+    (``B``), and other status indicators are excluded from the
+    calculated side so they show up as ``ground_truth_only`` rather
+    than being flagged as computation bugs.
+
+Gaps tracked:
+    - Limiting charge columns are stack-synthesized in pipe.py
+      (non_fac_fee × 1.0925), so they are excluded from compare_cols.
+      See homelab/stack#340 gap #3.
+    - Budget-neutrality adjustor exists on RuleYear but isn't
+      applied inside pfs.calcs.payment; this pricer currently assumes
+      adjustor == 1.0 (true for 2014-2025).  See gap #1.
+"""
+
+from __future__ import annotations
+
+from typing import ClassVar
+
+import duckdb
+import polars as pl
+
+from pfs.calcs.payment import payment
+from pfs.rules import RULES
+
+# Status codes whose payment is computed algorithmically and should
+# reconcile to the cent.  Everything else is ground-truth only.
+_ALGORITHMIC_STATUS: tuple[str, ...] = ("A", "T")
+
+
+class PfsPricer:
+    """PFS reconciliation: carrier_locality vs RVU × GPCI × CF."""
+
+    system: ClassVar[str] = "pfs"
+    description: ClassVar[str] = (
+        "Physician Fee Schedule — RVU × GPCI × CF vs pfs.carrier_locality"
+    )
+    join_keys: ClassVar[list[str]] = [
+        "year",
+        "mac",
+        "locality",
+        "hcpcs",
+        "mod",
+    ]
+    compare_cols: ClassVar[list[str]] = ["non_fac_fee", "fac_fee"]
+
+    # ── Ground truth ──────────────────────────────────────────────
+
+    def ground_truth(self, con: duckdb.DuckDBPyConnection, year: int) -> pl.DataFrame:
+        """CMS-published fees from ``pfs.carrier_locality``."""
+        return con.execute(
+            """
+            SELECT
+                year,
+                mac,
+                locality,
+                hcpcs,
+                COALESCE(mod, '') AS mod,
+                non_fac_fee,
+                fac_fee
+            FROM pfs.carrier_locality
+            WHERE year = ?
+            """,
+            [year],
+        ).pl()
+
+    # ── Calculated ────────────────────────────────────────────────
+
+    def calculated(self, con: duckdb.DuckDBPyConnection, year: int) -> pl.DataFrame:
+        """Calculated fees from RVU × GPCI × CF for every locality."""
+        if year not in RULES:
+            raise KeyError(
+                f"PFS RuleYear missing for {year}; "
+                f"add to pfs.rules.RULES before reconciling"
+            )
+        cf = RULES[year].conversion_factor
+
+        rvu = con.execute(
+            f"""
+            SELECT
+                year,
+                hcpcs,
+                COALESCE(mod, '') AS mod,
+                status_code,
+                work_rvu,
+                non_fac_pe_rvu,
+                fac_pe_rvu,
+                mp_rvu
+            FROM pfs.rvu
+            WHERE year = ?
+              AND status_code IN {_ALGORITHMIC_STATUS}
+            """,
+            [year],
+        ).pl()
+
+        gpci = con.execute(
+            """
+            SELECT
+                year,
+                mac,
+                locality,
+                work_gpci,
+                pe_gpci,
+                mp_gpci
+            FROM pfs.gpci
+            WHERE year = ?
+            """,
+            [year],
+        ).pl()
+
+        if rvu.height == 0 or gpci.height == 0:
+            return pl.DataFrame(
+                schema={
+                    "year": pl.Int64,
+                    "mac": pl.Utf8,
+                    "locality": pl.Utf8,
+                    "hcpcs": pl.Utf8,
+                    "mod": pl.Utf8,
+                    "non_fac_fee": pl.Float64,
+                    "fac_fee": pl.Float64,
+                }
+            )
+
+        # Cross-join RVU (one row per hcpcs+mod) with distinct locality
+        # identifiers from GPCI, then inject CF from RULES.  The cross
+        # join is O(n_hcpcs × n_localities) ≈ 8k × 110 ≈ 900k rows —
+        # fine for polars in memory.
+        localities = gpci.select(["mac", "locality"]).unique()
+        rvu_expanded = (
+            rvu.drop("status_code")
+            .join(localities, how="cross")
+            .with_columns(pl.lit(cf).alias("conv_factor"))
+        )
+
+        # payment() joins on (mac, locality) and returns the input with
+        # an added ``payment_amount`` column.
+        nf = payment(rvu_expanded, gpci, facility=False).select(
+            "year",
+            "mac",
+            "locality",
+            "hcpcs",
+            "mod",
+            pl.col("payment_amount").round(2).alias("non_fac_fee"),
+        )
+        fc = payment(rvu_expanded, gpci, facility=True).select(
+            "year",
+            "mac",
+            "locality",
+            "hcpcs",
+            "mod",
+            pl.col("payment_amount").round(2).alias("fac_fee"),
+        )
+
+        return nf.join(fc, on=["year", "mac", "locality", "hcpcs", "mod"], how="inner")
+
+    # ── Available years ───────────────────────────────────────────
+
+    def years_available(self, con: duckdb.DuckDBPyConnection) -> list[int]:
+        """Years with BOTH a carrier file and RVU+GPCI inputs loaded."""
+        rows = con.execute(
+            """
+            SELECT DISTINCT year FROM pfs.carrier_locality
+            INTERSECT
+            SELECT DISTINCT year FROM pfs.rvu
+            INTERSECT
+            SELECT DISTINCT year FROM pfs.gpci
+            ORDER BY 1
+            """
+        ).fetchall()
+        return [int(r[0]) for r in rows if r[0] is not None]

src/rec/report.py

@@ -0,0 +1,126 @@
+"""Markdown and JSON formatters for a ``Reconciliation``.
+
+Kept separate from ``base.py`` so the dataclass stays lightweight
+and the formatters can be extended without touching the data model.
+"""
+
+from __future__ import annotations
+
+import json
+
+import polars as pl
+
+from rec.base import Reconciliation
+
+
+def as_markdown(r: Reconciliation, *, top_n: int = 20) -> str:
+    """Render a Reconciliation as a markdown report.
+
+    Includes counts, match percentage, warnings, and the top-N rows
+    with the largest absolute delta (sorted descending).
+    """
+    lines: list[str] = []
+    lines.append(f"# Reconciliation — `{r.system}` CY{r.year}")
+    lines.append("")
+
+    status = "✓ perfect 1:1" if r.is_perfect else f"{r.pct_exact:.2f}% exact"
+    lines.append(f"**Status:** {status}")
+    lines.append("")
+
+    lines.append("## Counts")
+    lines.append("")
+    lines.append("| Metric | Value |")
+    lines.append("|---|---:|")
+    lines.append(f"| Ground-truth rows | {r.ground_truth_rows:,} |")
+    lines.append(f"| Calculated rows | {r.calculated_rows:,} |")
+    lines.append(f"| Matched (both sides) | {r.matched_rows:,} |")
+    lines.append(f"| Exact matches (≤ {r.tolerance_cents}¢) | {r.exact_matches:,} |")
+    lines.append(f"| Near matches (≤ 1¢) | {r.near_matches:,} |")
+    lines.append(f"| Ground-truth only | {r.ground_truth_only:,} |")
+    lines.append(f"| Calculated only | {r.calculated_only:,} |")
+    lines.append("")
+
+    if r.warnings:
+        lines.append("## Warnings")
+        lines.append("")
+        for w in r.warnings:
+            lines.append(f"- {w}")
+        lines.append("")
+
+    diffs = _top_diffs(r.deltas, top_n)
+    if diffs.height > 0:
+        lines.append(f"## Top {diffs.height} deltas")
+        lines.append("")
+        lines.append(_polars_to_markdown(diffs))
+        lines.append("")
+    else:
+        lines.append("## Top deltas")
+        lines.append("")
+        lines.append("*No non-zero deltas — every matched row is exact.*")
+        lines.append("")
+
+    return "\n".join(lines)
+
+
+def as_json(r: Reconciliation, *, top_n: int = 20) -> str:
+    """Render a Reconciliation as JSON.
+
+    Scalar stats + warnings go at the top level; the top-N delta
+    rows are embedded under ``deltas`` as a list of dicts.
+    """
+    diffs = _top_diffs(r.deltas, top_n)
+    payload = {
+        "system": r.system,
+        "year": r.year,
+        "ground_truth_rows": r.ground_truth_rows,
+        "calculated_rows": r.calculated_rows,
+        "matched_rows": r.matched_rows,
+        "ground_truth_only": r.ground_truth_only,
+        "calculated_only": r.calculated_only,
+        "exact_matches": r.exact_matches,
+        "near_matches": r.near_matches,
+        "tolerance_cents": r.tolerance_cents,
+        "pct_exact": round(r.pct_exact, 4),
+        "is_perfect": r.is_perfect,
+        "warnings": list(r.warnings),
+        "top_deltas": diffs.to_dicts(),
+    }
+    return json.dumps(payload, indent=2, default=str)
+
+
+# ── Internal helpers ──────────────────────────────────────────────
+
+
+def _top_diffs(deltas: pl.DataFrame, top_n: int) -> pl.DataFrame:
+    """Return the top-N non-exact rows by ``abs_max_delta`` descending."""
+    if deltas.height == 0 or "abs_max_delta" not in deltas.columns:
+        return deltas.head(0)
+    # Drop rows that are matched and exact — they add no signal.
+    if "is_exact" in deltas.columns:
+        non_exact = deltas.filter(~pl.col("is_exact").fill_null(False))
+    else:
+        non_exact = deltas
+    return non_exact.head(top_n)
+
+
+def _polars_to_markdown(df: pl.DataFrame) -> str:
+    """Render a small polars DataFrame as a GitHub-style markdown table."""
+    if df.height == 0:
+        return "*(empty)*"
+
+    cols = df.columns
+    header = "| " + " | ".join(cols) + " |"
+    sep = "|" + "|".join(["---"] * len(cols)) + "|"
+    rows = []
+    for row in df.iter_rows():
+        cells = [_fmt_cell(v) for v in row]
+        rows.append("| " + " | ".join(cells) + " |")
+    return "\n".join([header, sep, *rows])
+
+
+def _fmt_cell(v: object) -> str:
+    if v is None:
+        return ""
+    if isinstance(v, float):
+        return f"{v:.2f}"
+    return str(v).replace("|", "\\|")

stack.toml

@@ -85,6 +85,7 @@ zotero = "data/zotero/data/storage"
 bcda = "data/bcda"
 rex = "data/rex"
 pfs = "data/pfs"
+rec = "data/rec"
 cms_log = "data/cms/log.jsonl"
 bcda_log = "data/bcda/log.jsonl"
 

tests/rec/conftest.py

@@ -0,0 +1,181 @@
+"""Shared fixtures for rec tests.
+
+Provides:
+
+* ``con`` — an in-memory DuckDB with a ``pfs`` schema seeded with
+  tiny rvu/gpci/carrier_locality rows designed to exercise the full
+  match/miss matrix:
+
+    row A — exact match (calc round == carrier)
+    row B — 1¢ near-miss (calc rounds to $X.Y1, carrier $X.Y0)
+    row C — ground-truth only (carrier row with status_code = 'C'
+            so calc side skips it)
+    row D — calculated only (RVU row with no carrier file entry)
+
+* ``fake_pricer`` — a hand-rolled Pricer that takes hard-coded polars
+  frames so engine tests don't need DuckDB.
+"""
+
+from __future__ import annotations
+
+from typing import ClassVar
+
+import duckdb
+import polars as pl
+import pytest
+
+from pfs.rules import RULES
+
+# Pick a year that exists in pfs.rules.RULES
+_TEST_YEAR = 2025
+assert _TEST_YEAR in RULES, f"test seed year {_TEST_YEAR} missing from RULES"
+_CF = RULES[_TEST_YEAR].conversion_factor
+
+
+# ── In-memory DuckDB seeded for PfsPricer tests ──────────────────
+
+
+@pytest.fixture
+def con() -> duckdb.DuckDBPyConnection:
+    """DuckDB in-memory with seeded pfs.* tables."""
+    c = duckdb.connect(":memory:")
+    c.execute("CREATE SCHEMA pfs")
+
+    # RVU values engineered so:
+    #   99213 (A):   work=0.97, nfpe=1.04, mp=0.07
+    #   99214 (A):   work=1.50, nfpe=1.56, mp=0.11
+    #   99999 (C):   work=1.00, nfpe=1.00, mp=0.00  -- carrier-priced
+    #   88888 (A):   work=0.50, nfpe=0.40, mp=0.05  -- no carrier row
+    c.execute(
+        """
+        CREATE TABLE pfs.rvu (
+            year INTEGER,
+            hcpcs VARCHAR,
+            mod VARCHAR,
+            status_code VARCHAR,
+            work_rvu DOUBLE,
+            non_fac_pe_rvu DOUBLE,
+            fac_pe_rvu DOUBLE,
+            mp_rvu DOUBLE,
+            conv_factor DOUBLE
+        )
+        """
+    )
+    c.execute(
+        f"""
+        INSERT INTO pfs.rvu VALUES
+            ({_TEST_YEAR}, '99213', NULL, 'A', 0.97, 1.04, 0.41, 0.07, NULL),
+            ({_TEST_YEAR}, '99214', NULL, 'A', 1.50, 1.56, 0.63, 0.11, NULL),
+            ({_TEST_YEAR}, '99999', NULL, 'C', 1.00, 1.00, 1.00, 0.00, NULL),
+            ({_TEST_YEAR}, '88888', NULL, 'A', 0.50, 0.40, 0.20, 0.05, NULL)
+        """
+    )
+
+    # Single locality seeded so expected payments are deterministic.
+    c.execute(
+        """
+        CREATE TABLE pfs.gpci (
+            year INTEGER,
+            mac VARCHAR,
+            locality VARCHAR,
+            locality_name VARCHAR,
+            work_gpci DOUBLE,
+            pe_gpci DOUBLE,
+            mp_gpci DOUBLE
+        )
+        """
+    )
+    c.execute(
+        f"""
+        INSERT INTO pfs.gpci VALUES
+            ({_TEST_YEAR}, '10212', '01', 'ALABAMA', 1.000, 0.880, 0.550)
+        """
+    )
+
+    # Carrier-locality rows.
+    # Row A (99213): calculate exact expected payment and seed it.
+    #   nf = (0.97*1.0 + 1.04*0.88 + 0.07*0.55) * CF
+    #   fac = (0.97*1.0 + 0.41*0.88 + 0.07*0.55) * CF
+    nf_99213 = round((0.97 * 1.0 + 1.04 * 0.88 + 0.07 * 0.55) * _CF, 2)
+    fac_99213 = round((0.97 * 1.0 + 0.41 * 0.88 + 0.07 * 0.55) * _CF, 2)
+    # Row B (99214): seed a value 1¢ off the calc to exercise near_matches.
+    nf_99214_calc = round((1.50 * 1.0 + 1.56 * 0.88 + 0.11 * 0.55) * _CF, 2)
+    fac_99214_calc = round((1.50 * 1.0 + 0.63 * 0.88 + 0.11 * 0.55) * _CF, 2)
+    nf_99214_seeded = round(nf_99214_calc - 0.01, 2)
+    fac_99214_seeded = fac_99214_calc  # only nf is the near miss
+    # Row C (99999): status C, so calc side skips — only GT.
+    nf_99999 = 55.55
+    fac_99999 = 44.44
+    # Row D (88888): calc side only — no carrier row.
+
+    c.execute(
+        """
+        CREATE TABLE pfs.carrier_locality (
+            year INTEGER,
+            mac VARCHAR,
+            locality VARCHAR,
+            hcpcs VARCHAR,
+            mod VARCHAR,
+            non_fac_fee DOUBLE,
+            fac_fee DOUBLE,
+            non_fac_limiting_charge DOUBLE,
+            fac_limiting_charge DOUBLE
+        )
+        """
+    )
+    c.execute(
+        f"""
+        INSERT INTO pfs.carrier_locality VALUES
+            ({_TEST_YEAR}, '10212', '01', '99213', NULL,
+             {nf_99213}, {fac_99213}, NULL, NULL),
+            ({_TEST_YEAR}, '10212', '01', '99214', NULL,
+             {nf_99214_seeded}, {fac_99214_seeded}, NULL, NULL),
+            ({_TEST_YEAR}, '10212', '01', '99999', NULL,
+             {nf_99999}, {fac_99999}, NULL, NULL)
+        """
+    )
+
+    return c
+
+
+@pytest.fixture
+def test_year() -> int:
+    return _TEST_YEAR
+
+
+# ── Fake pricer for engine tests ─────────────────────────────────
+
+
+class _FakePricer:
+    """Hand-rolled pricer that returns hard-coded polars frames."""
+
+    system: ClassVar[str] = "fake"
+    description: ClassVar[str] = "Test pricer — returns fixture frames"
+    join_keys: ClassVar[list[str]] = ["hcpcs", "mod"]
+    compare_cols: ClassVar[list[str]] = ["fee"]
+
+    def ground_truth(self, con, year: int) -> pl.DataFrame:
+        return pl.DataFrame(
+            {
+                "hcpcs": ["A001", "A002", "A003"],  # exact, 1¢ near, gt-only
+                "mod": ["", "", ""],
+                "fee": [10.00, 20.00, 30.00],
+            }
+        )
+
+    def calculated(self, con, year: int) -> pl.DataFrame:
+        return pl.DataFrame(
+            {
+                "hcpcs": ["A001", "A002", "A004"],  # exact, 1¢ near, calc-only
+                "mod": ["", "", ""],
+                "fee": [10.00, 20.01, 40.00],
+            }
+        )
+
+    def years_available(self, con) -> list[int]:
+        return [2025]
+
+
+@pytest.fixture
+def fake_pricer() -> _FakePricer:
+    return _FakePricer()

tests/rec/pricers/test_pfs.py

@@ -0,0 +1,109 @@
+"""Tests for rec.pricers.pfs.PfsPricer with a seeded in-memory DuckDB."""
+
+from __future__ import annotations
+
+from rec.base import Pricer
+from rec.engine import reconcile
+from rec.pricers import PRICERS
+from rec.pricers.pfs import PfsPricer
+
+
+class TestRegistration:
+    def test_in_PRICERS(self) -> None:
+        assert "pfs" in PRICERS
+        assert PRICERS["pfs"] is PfsPricer
+
+    def test_conforms_to_protocol(self) -> None:
+        assert isinstance(PfsPricer(), Pricer)
+
+    def test_class_attributes(self) -> None:
+        assert PfsPricer.system == "pfs"
+        assert PfsPricer.join_keys == ["year", "mac", "locality", "hcpcs", "mod"]
+        assert PfsPricer.compare_cols == ["non_fac_fee", "fac_fee"]
+
+
+class TestGroundTruth:
+    def test_row_count(self, con, test_year) -> None:
+        p = PfsPricer()
+        gt = p.ground_truth(con, test_year)
+        # Seeded 3 carrier rows (99213, 99214, 99999)
+        assert gt.height == 3
+
+    def test_columns(self, con, test_year) -> None:
+        p = PfsPricer()
+        gt = p.ground_truth(con, test_year)
+        for col in [
+            "year",
+            "mac",
+            "locality",
+            "hcpcs",
+            "mod",
+            "non_fac_fee",
+            "fac_fee",
+        ]:
+            assert col in gt.columns
+
+
+class TestCalculated:
+    def test_excludes_carrier_priced(self, con, test_year) -> None:
+        """99999 has status_code='C' so calc side skips it."""
+        p = PfsPricer()
+        calc = p.calculated(con, test_year)
+        hcpcs = calc["hcpcs"].to_list()
+        assert "99999" not in hcpcs
+
+    def test_includes_algorithmic(self, con, test_year) -> None:
+        """99213, 99214, 88888 all have status 'A'."""
+        p = PfsPricer()
+        calc = p.calculated(con, test_year)
+        hcpcs = set(calc["hcpcs"].to_list())
+        assert {"99213", "99214", "88888"}.issubset(hcpcs)
+
+    def test_returns_cents_rounded(self, con, test_year) -> None:
+        p = PfsPricer()
+        calc = p.calculated(con, test_year)
+        # All values should be round to 2 decimals (cents)
+        for fee in calc["non_fac_fee"].to_list() + calc["fac_fee"].to_list():
+            assert fee == round(fee, 2)
+
+
+class TestReconcilePfs:
+    def test_summary_counts(self, con, test_year) -> None:
+        r = reconcile(PfsPricer(), con, test_year)
+        # Ground-truth rows: 99213, 99214, 99999 = 3
+        assert r.ground_truth_rows == 3
+        # Calculated rows: 99213, 99214, 88888 = 3 (99999 skipped)
+        assert r.calculated_rows == 3
+        # Matched: 99213, 99214 = 2
+        assert r.matched_rows == 2
+        # gt_only: 99999
+        assert r.ground_truth_only == 1
+        # calc_only: 88888
+        assert r.calculated_only == 1
+
+    def test_exact_and_near(self, con, test_year) -> None:
+        r = reconcile(PfsPricer(), con, test_year)
+        # 99213 seeded to match exactly; 99214 is 1¢ off on non_fac_fee.
+        assert r.exact_matches == 1
+        assert r.near_matches == 2
+
+    def test_tolerance_promotes(self, con, test_year) -> None:
+        r = reconcile(PfsPricer(), con, test_year, tolerance_cents=1)
+        assert r.exact_matches == 2
+
+    def test_delta_table_contains_hcpcs(self, con, test_year) -> None:
+        r = reconcile(PfsPricer(), con, test_year)
+        hcpcs = set(r.deltas["hcpcs"].to_list())
+        # All four distinct hcpcs from outer join
+        assert hcpcs == {"99213", "99214", "99999", "88888"}
+
+    def test_years_available(self, con, test_year) -> None:
+        p = PfsPricer()
+        assert p.years_available(con) == [test_year]
+
+    def test_missing_year_raises(self, con) -> None:
+        import pytest
+
+        p = PfsPricer()
+        with pytest.raises(KeyError, match="RuleYear missing"):
+            p.calculated(con, 1999)

tests/rec/test_base.py

@@ -0,0 +1,137 @@
+"""Tests for rec.base — Pricer protocol and Reconciliation dataclass."""
+
+from __future__ import annotations
+
+import polars as pl
+
+from rec.base import Pricer, Reconciliation
+
+
+class TestPricerProtocol:
+    def test_fake_pricer_conforms(self, fake_pricer) -> None:
+        assert isinstance(fake_pricer, Pricer)
+
+    def test_pfs_pricer_conforms(self) -> None:
+        from rec.pricers.pfs import PfsPricer
+
+        assert isinstance(PfsPricer(), Pricer)
+
+    def test_non_conforming_object_rejected(self) -> None:
+        class NotAPricer:
+            pass
+
+        assert not isinstance(NotAPricer(), Pricer)
+
+    def test_missing_method_rejected(self) -> None:
+        class Incomplete:
+            system = "x"
+            description = "x"
+            join_keys: list[str] = []
+            compare_cols: list[str] = []
+
+            def ground_truth(self, con, year):
+                return None
+
+            # missing calculated and years_available
+
+        assert not isinstance(Incomplete(), Pricer)
+
+
+class TestReconciliation:
+    def test_scalar_fields(self) -> None:
+        r = Reconciliation(
+            system="x",
+            year=2025,
+            ground_truth_rows=100,
+            calculated_rows=100,
+            matched_rows=99,
+            ground_truth_only=1,
+            calculated_only=1,
+            exact_matches=95,
+            near_matches=98,
+            deltas=pl.DataFrame(),
+        )
+        assert r.system == "x"
+        assert r.year == 2025
+        assert r.matched_rows == 99
+        assert r.tolerance_cents == 0
+
+    def test_pct_exact(self) -> None:
+        r = Reconciliation(
+            system="x",
+            year=2025,
+            ground_truth_rows=100,
+            calculated_rows=100,
+            matched_rows=100,
+            ground_truth_only=0,
+            calculated_only=0,
+            exact_matches=75,
+            near_matches=100,
+            deltas=pl.DataFrame(),
+        )
+        assert r.pct_exact == 75.0
+
+    def test_pct_exact_empty(self) -> None:
+        r = Reconciliation(
+            system="x",
+            year=2025,
+            ground_truth_rows=0,
+            calculated_rows=0,
+            matched_rows=0,
+            ground_truth_only=0,
+            calculated_only=0,
+            exact_matches=0,
+            near_matches=0,
+            deltas=pl.DataFrame(),
+        )
+        assert r.pct_exact == 0.0
+
+    def test_is_perfect_true(self) -> None:
+        r = Reconciliation(
+            system="x",
+            year=2025,
+            ground_truth_rows=10,
+            calculated_rows=10,
+            matched_rows=10,
+            ground_truth_only=0,
+            calculated_only=0,
+            exact_matches=10,
+            near_matches=10,
+            deltas=pl.DataFrame(),
+        )
+        assert r.is_perfect is True
+
+    def test_is_perfect_false_on_unmatched(self) -> None:
+        r = Reconciliation(
+            system="x",
+            year=2025,
+            ground_truth_rows=10,
+            calculated_rows=10,
+            matched_rows=10,
+            ground_truth_only=0,
+            calculated_only=0,
+            exact_matches=9,  # one row off
+            near_matches=10,
+            deltas=pl.DataFrame(),
+        )
+        assert r.is_perfect is False
+
+    def test_frozen(self) -> None:
+        import dataclasses
+
+        r = Reconciliation(
+            system="x",
+            year=2025,
+            ground_truth_rows=0,
+            calculated_rows=0,
+            matched_rows=0,
+            ground_truth_only=0,
+            calculated_only=0,
+            exact_matches=0,
+            near_matches=0,
+            deltas=pl.DataFrame(),
+        )
+        import pytest
+
+        with pytest.raises(dataclasses.FrozenInstanceError):
+            r.year = 2026  # type: ignore[misc]

tests/rec/test_engine.py

@@ -0,0 +1,76 @@
+"""Tests for rec.engine.reconcile with a hand-rolled fake pricer."""
+
+from __future__ import annotations
+
+import pytest
+
+from rec.engine import reconcile, reconcile_all_years
+
+
+class TestReconcile:
+    def test_counts(self, fake_pricer) -> None:
+        r = reconcile(fake_pricer, con=None, year=2025)
+        # FakePricer:
+        #   ground_truth: A001 (10.00), A002 (20.00), A003 (30.00)
+        #   calculated:   A001 (10.00), A002 (20.01), A004 (40.00)
+        # Matched: A001, A002 → 2
+        # gt_only: A003 → 1
+        # calc_only: A004 → 1
+        assert r.ground_truth_rows == 3
+        assert r.calculated_rows == 3
+        assert r.matched_rows == 2
+        assert r.ground_truth_only == 1
+        assert r.calculated_only == 1
+
+    def test_exact_vs_near(self, fake_pricer) -> None:
+        r = reconcile(fake_pricer, con=None, year=2025)
+        # A001 is exact; A002 is 1¢ off → near but not exact at tolerance=0.
+        assert r.exact_matches == 1
+        assert r.near_matches == 2
+
+    def test_tolerance_promotes_near_to_exact(self, fake_pricer) -> None:
+        r = reconcile(fake_pricer, con=None, year=2025, tolerance_cents=1)
+        assert r.exact_matches == 2
+        assert r.tolerance_cents == 1
+
+    def test_delta_table_shape(self, fake_pricer) -> None:
+        r = reconcile(fake_pricer, con=None, year=2025)
+        # All 4 distinct hcpcs appear in the outer join
+        assert r.deltas.height == 4
+        # Columns: join_keys + per-col suffixes + abs_max_delta + flags
+        cols = set(r.deltas.columns)
+        for k in fake_pricer.join_keys:
+            assert k in cols
+        assert "fee_gt" in cols
+        assert "fee_calc" in cols
+        assert "fee_delta" in cols
+        assert "abs_max_delta" in cols
+        assert "is_exact" in cols
+        assert "is_near" in cols
+
+    def test_delta_sorted_by_abs_max_desc(self, fake_pricer) -> None:
+        r = reconcile(fake_pricer, con=None, year=2025)
+        # The top rows should be the ones with largest abs_max_delta
+        # (A004 calc-only and A003 gt-only both have null deltas — with
+        # nulls_last they come last).  Of the real deltas, A002 has 0.01
+        # and A001 has 0.00.
+        non_null = r.deltas.drop_nulls("abs_max_delta")
+        assert non_null.height == 2
+        # A002 (0.01) should be first
+        first = non_null.row(0, named=True)
+        assert first["hcpcs"] == "A002"
+        assert abs(first["abs_max_delta"] - 0.01) < 1e-9
+
+    def test_rejects_non_pricer(self) -> None:
+        class NotAPricer:
+            pass
+
+        with pytest.raises(TypeError, match="does not conform"):
+            reconcile(NotAPricer(), con=None, year=2025)  # type: ignore[arg-type]
+
+
+class TestReconcileAllYears:
+    def test_dispatch(self, fake_pricer) -> None:
+        results = reconcile_all_years(fake_pricer, con=None)
+        assert set(results.keys()) == {2025}
+        assert results[2025].matched_rows == 2

tests/rec/test_report.py

@@ -0,0 +1,57 @@
+"""Tests for rec.report — markdown and JSON formatters."""
+
+from __future__ import annotations
+
+import json
+
+from rec.engine import reconcile
+from rec.report import as_json, as_markdown
+
+
+class TestAsMarkdown:
+    def test_contains_header(self, fake_pricer) -> None:
+        r = reconcile(fake_pricer, con=None, year=2025)
+        md = as_markdown(r)
+        assert "# Reconciliation — `fake` CY2025" in md
+
+    def test_contains_counts_table(self, fake_pricer) -> None:
+        r = reconcile(fake_pricer, con=None, year=2025)
+        md = as_markdown(r)
+        assert "## Counts" in md
+        assert "Ground-truth rows" in md
+        assert "Calculated rows" in md
+        assert "Matched (both sides)" in md
+
+    def test_contains_top_deltas(self, fake_pricer) -> None:
+        r = reconcile(fake_pricer, con=None, year=2025)
+        md = as_markdown(r)
+        assert "## Top" in md
+        # A002 has the 1¢ delta; should appear in top deltas table
+        assert "A002" in md
+
+    def test_perfect_status_line(self, fake_pricer) -> None:
+        r = reconcile(fake_pricer, con=None, year=2025, tolerance_cents=1)
+        md = as_markdown(r)
+        # With tolerance=1¢, both matched rows are exact — but there's
+        # still a gt_only and a calc_only so is_perfect is False.
+        assert "Status:" in md
+
+
+class TestAsJson:
+    def test_valid_json(self, fake_pricer) -> None:
+        r = reconcile(fake_pricer, con=None, year=2025)
+        payload = json.loads(as_json(r))
+        assert isinstance(payload, dict)
+        assert payload["system"] == "fake"
+        assert payload["year"] == 2025
+        assert payload["matched_rows"] == 2
+        assert payload["exact_matches"] == 1
+        assert payload["near_matches"] == 2
+        assert payload["pct_exact"] == 50.0
+
+    def test_top_deltas_embedded(self, fake_pricer) -> None:
+        r = reconcile(fake_pricer, con=None, year=2025)
+        payload = json.loads(as_json(r))
+        assert "top_deltas" in payload
+        assert isinstance(payload["top_deltas"], list)
+        assert len(payload["top_deltas"]) >= 1

uv.lock

@@ -3211,6 +3211,15 @@ pfs = [
     { name = "pyarrow" },
     { name = "pydantic" },
 ]
+rec = [
+    { name = "duckdb" },
+    { name = "fsspec" },
+    { name = "httpx" },
+    { name = "narwhals" },
+    { name = "pyarrow" },
+    { name = "pydantic" },
+    { name = "typer" },
+]
 rex = [
     { name = "duckdb" },
     { name = "fsspec" },
@@ -3256,6 +3265,7 @@ requires-dist = [
     { name = "duckdb", marker = "extra == 'conf'", specifier = ">=1.0.0" },
     { name = "duckdb", marker = "extra == 'opps'", specifier = ">=1.0.0" },
     { name = "duckdb", marker = "extra == 'pfs'", specifier = ">=1.0.0" },
+    { name = "duckdb", marker = "extra == 'rec'", specifier = ">=1.0.0" },
     { name = "fastapi", marker = "extra == 'api'", specifier = ">=0.135.1" },
     { name = "fsspec", marker = "extra == 'bcda'", specifier = ">=2024.1.0" },
     { name = "fsspec", marker = "extra == 'rex'", specifier = ">=2024.1.0" },
@@ -3269,6 +3279,7 @@ requires-dist = [
     { name = "narwhals", marker = "extra == 'cms'", specifier = ">=2.17.0" },
     { name = "narwhals", marker = "extra == 'opps'", specifier = ">=2.17.0" },
     { name = "narwhals", marker = "extra == 'pfs'", specifier = ">=2.17.0" },
+    { name = "narwhals", marker = "extra == 'rec'", specifier = ">=2.17.0" },
     { name = "narwhals", marker = "extra == 'rex'", specifier = ">=2.17.0" },
     { name = "opentelemetry-api", marker = "extra == 'perf'", specifier = ">=1.25.0" },
     { name = "opentelemetry-exporter-otlp-proto-grpc", marker = "extra == 'perf'", specifier = ">=1.25.0" },
@@ -3307,19 +3318,23 @@ requires-dist = [
     { name = "stack", extras = ["conf"], marker = "extra == 'opps'" },
     { name = "stack", extras = ["conf"], marker = "extra == 'perf'" },
     { name = "stack", extras = ["conf"], marker = "extra == 'pfs'" },
+    { name = "stack", extras = ["conf"], marker = "extra == 'rec'" },
     { name = "stack", extras = ["conf"], marker = "extra == 'rex'" },
     { name = "stack", extras = ["lake"], marker = "extra == 'all'" },
     { name = "stack", extras = ["perf"], marker = "extra == 'all'" },
     { name = "stack", extras = ["pfs"], marker = "extra == 'all'" },
+    { name = "stack", extras = ["pfs"], marker = "extra == 'rec'" },
+    { name = "stack", extras = ["rec"], marker = "extra == 'all'" },
     { name = "stack", extras = ["rex"], marker = "extra == 'all'" },
     { name = "stack", extras = ["rex"], marker = "extra == 'opps'" },
     { name = "stack", extras = ["rex"], marker = "extra == 'pfs'" },
     { name = "typer", marker = "extra == 'cli'", specifier = ">=0.24.1" },
+    { name = "typer", marker = "extra == 'rec'", specifier = ">=0.24.1" },
     { name = "uvicorn", marker = "extra == 'api'", specifier = ">=0.41.0" },
     { name = "uvicorn", marker = "extra == 'cli'", specifier = ">=0.41.0" },
     { name = "xlrd", specifier = ">=2.0.2" },
 ]
-provides-extras = ["conf", "aco", "api", "bcda", "bib", "bls", "ccw", "cli", "cms", "opps", "pfs", "rex", "perf", "sem", "lake", "aws", "gcp", "azure", "all"]
+provides-extras = ["conf", "aco", "api", "bcda", "bib", "bls", "ccw", "cli", "cms", "opps", "pfs", "rec", "rex", "perf", "sem", "lake", "aws", "gcp", "azure", "all"]
 
 [package.metadata.requires-dev]
 dev = [