Workflow Automation Code Samples: A Daily Reconciliation On Polars 2.0 With A Signed-Off Break Report

"""Positions reconciliation on Polars 2.0: portfolio system against the custodian file.

Polars 2.0 (6 October 2026) turns on out-of-core processing by default and, in several places, stops
keeping input order wherever the query does not ask for it. A join's output order was never a promise.
A break report must be reproducible, so this pipeline sorts explicitly, validates the join keys and
refuses a file whose columns have changed.

Requires polars 2.0.0. Tested on Python 3.14.4 and 3.11.14 with mypy --strict. Sample data only.
"""

from __future__ import annotations

import hashlib
from pathlib import Path

import polars as pl

KEYS = ["account", "isin"]
SCHEMA: dict[str, pl.DataType] = {"account": pl.String(), "isin": pl.String(), "quantity": pl.Decimal(18, 4)}


def load(path: Path, side: str) -> pl.LazyFrame:
    with path.open(encoding="utf-8") as fh:
        header = fh.readline().strip().split(",")
    if header != list(SCHEMA):
        raise ValueError(f"{path}: expected columns {list(SCHEMA)}, found {header}")
    return pl.scan_csv(path, schema=SCHEMA).rename({"quantity": f"qty_{side}"})


def breaks(internal: Path, custodian: Path) -> pl.DataFrame:
    joined = load(internal, "internal").join(
        load(custodian, "custodian"), on=KEYS, how="full", coalesce=True, validate="1:1",
    )
    return (
        joined.with_columns(
            pl.when(pl.col("qty_custodian").is_null()).then(pl.lit("missing_at_custodian"))
            .when(pl.col("qty_internal").is_null()).then(pl.lit("missing_internally"))
            .when(pl.col("qty_internal") != pl.col("qty_custodian")).then(pl.lit("quantity_mismatch"))
            .otherwise(pl.lit(None)).alias("break_type"),
            (pl.col("qty_internal").fill_null(0) - pl.col("qty_custodian").fill_null(0)).alias("difference"),
        )
        .filter(pl.col("break_type").is_not_null())
        .sort(KEYS)  # explicit order: never rely on the order a join happens to return
        .collect()
    )


def fingerprint(df: pl.DataFrame) -> str:
    """Same inputs, same report, same hash: what a reviewer signs is what the auditor sees."""
    return hashlib.sha256(df.write_csv().encode()).hexdigest()


if __name__ == "__main__":
    Path("internal.csv").write_text(
        "account,isin,quantity\n"
        "ACC-20417,GB00B03MLX29,1000\nACC-20417,GB0002634946,250\n"
        "ACC-11893,GB0007980591,40\nACC-30001,GB00B03MLX29,75.5\n")
    Path("custodian.csv").write_text(
        "account,isin,quantity\n"
        "ACC-20417,GB00B03MLX29,1500\nACC-20417,GB0002634946,250\n"
        "ACC-30001,GB00B03MLX29,75.5\nACC-44120,GB0002634946,10\n")
    report = breaks(Path("internal.csv"), Path("custodian.csv"))
    print(report)
    print("fingerprint:", fingerprint(report)[:16], "| repeat:", fingerprint(breaks(Path("internal.csv"),
                                                                                    Path("custodian.csv")))[:16])
    Path("custodian_v2.csv").write_text("account,isin,quantity,price\nACC-20417,GB00B03MLX29,1500,3600\n")
    try:
        breaks(Path("internal.csv"), Path("custodian_v2.csv"))
    except ValueError as exc:
        print("refused:", exc)

Requires polars 2.0.0. Tested on Python 3.14.4 and 3.11.14 with mypy 2.4.0 --strict. Sample data.

"""Sign-off for the daily break report: a named reviewer resolves every break with a reason, and the
sign-off is bound to the report's fingerprint so a regenerated or edited report needs a fresh sign-off.

Requires polars 2.0.0 (via recon.py). Tested on Python 3.14.4 and 3.11.14 with mypy --strict.
"""

from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Literal

import polars as pl

from recon import breaks, fingerprint

Resolution = Literal["book_correction", "chase_counterparty", "timing_difference", "escalate"]


@dataclass(frozen=True)
class Signoff:
    account: str
    isin: str
    resolution: Resolution
    reason: str


def sign_off(report: pl.DataFrame, reviewer: str, decisions: list[Signoff], log: Path) -> str:
    open_breaks = {(str(r["account"]), str(r["isin"])) for r in report.iter_rows(named=True)}
    decided = {(d.account, d.isin) for d in decisions}
    missing = sorted(open_breaks - decided)
    if missing:
        raise ValueError(f"no decision for {missing}")
    if any(len(d.reason.strip()) < 10 for d in decisions):
        raise ValueError("every decision needs a reason a colleague could follow")
    lines = log.read_text().splitlines() if log.exists() else []
    prev = json.loads(lines[-1])["hash"] if lines else "0" * 64
    body = {"report_fingerprint": fingerprint(report), "reviewer": reviewer, "prev": prev,
            "at": datetime.now(timezone.utc).isoformat(),
            "decisions": [vars(d) for d in sorted(decisions, key=lambda d: (d.account, d.isin))]}
    digest = hashlib.sha256(json.dumps(body, sort_keys=True).encode()).hexdigest()
    with log.open("a", encoding="utf-8") as fh:
        fh.write(json.dumps({**body, "hash": digest}) + "\n")
    return digest


if __name__ == "__main__":
    report = breaks(Path("internal.csv"), Path("custodian.csv"))  # files written by recon.py
    log = Path("signoffs.jsonl")
    log.unlink(missing_ok=True)
    partial = [Signoff("ACC-20417", "GB00B03MLX29", "book_correction",
                       "Scrip dividend of 500 units credited at custodian on 14 Oct, not yet booked")]
    try:
        sign_off(report, "ops.lead@example.co.uk", partial, log)
    except ValueError as exc:
        print("refused:", exc)
    complete = partial + [
        Signoff("ACC-11893", "GB0007980591", "chase_counterparty", "Custodian file omits the account, chaser drafted"),
        Signoff("ACC-44120", "GB0002634946", "escalate", "Account not in portfolio system, needs onboarding review"),
    ]
    print("signed:", sign_off(report, "ops.lead@example.co.uk", complete, log)[:16])

Requires polars 2.0.0 via recon.py. Tested on Python 3.14.4 and 3.11.14 with mypy 2.4.0 --strict.