Classifying Controllable Expenses with Python

Classifying controllable CAM expenses in Python turns a table of posted general-ledger codes into a per-line controllability tag a lease cap can trust, resolving each account through a two-tier rule engine that a reviewer can audit back to the clause or portfolio rule that decided it. This is the focused build of the classifier that the controllable vs non-controllable expense segregation architecture specifies: where the parent topic lays out the data model and the reasons the boundary matters, this page walks the actual code that reads a recoverable pool, applies range-based portfolio defaults, layers lease-level overrides on top, resolves the two tiers by strict precedence, and emits a tagged ledger the cap stage can consume. Every account it tags arrives already posted by GL code mapping for CAM expenses, so the classifier never guesses what a line is — it only decides whether the landlord can influence it. The last section adds a period-over-period drift check, the safeguard that catches a tax range silently flipping into the controllable pool before a tenant’s consultant does.

Context & When to Use This Approach #

Reach for this classifier the moment a reconciliation carries a controllable expense cap, because a cap is only honest if it clamps the controllable slice and nothing else. If a lease has no cap and recovers the entire operating pool as a straight pass-through, the controllability tag is decorative and you can skip the machinery. But most modern office and retail leases cap the growth of controllable operating expenses at a negotiated percentage, and the instant that clause exists, every recoverable dollar has to be sorted into a pool the cap may touch and a pool it may not — property taxes, insurance, and utility tariffs on one side; landscaping, management fees, and non-union labor on the other. Getting a single tax line onto the wrong side overbills the tenant in a way that clears a desk review and fails an audit.

The approach here is deliberately rule-driven rather than heuristic. A classifier that reads keywords out of an invoice memo — flagging anything containing “tax” as non-controllable — is the wrong instinct: memos are inconsistent, adversarial vendors mislabel line items, and a fuzzy match has no clause to point to when a tenant challenges the split. Instead the classifier keys off the resolved GL account, which is a fact established upstream and reconciled, and it expresses controllability as two tiers of explicit rules: portfolio-wide defaults that cover the common case across the whole book, and per-lease overrides that encode the handful of boundaries a specific tenant negotiated. That separation is what lets one property manager onboard a newly acquired asset by inheriting the entire default set and adding only its leases’ exceptions, rather than authoring a fresh controllability map per building.

Two design commitments run through the whole implementation. First, money is decimal.Decimal end to end; a classifier that leaked a fraction of a cent while summing a bucket would break the conservation check that proves no dollar was lost in the split, and binary float cannot hold a cent exactly. Second, an account no rule resolves is never guessed — it is routed to a review bucket that blocks the reconciliation until a human maps it. Guessing an unknown as controllable pulls an unclassified cost under a cap; guessing it as non-controllable strips a genuine controllable of the ceiling the tenant bargained for. Both are silent errors, and silent errors in a recovery calculation are the expensive kind.

Step-by-Step Implementation #

The build proceeds in five steps: assemble the portfolio default rules as account ranges, register the lease’s overrides, apply the classifier under strict precedence, emit the tagged ledger, and drift-check the result against the prior period.

Step 1 — Build the GL→controllability rule map #

The portfolio defaults are expressed as GL account ranges, not individual codes. A rule that says the entire 7100–7199 real-estate-tax range is non-controllable survives a chart expansion that a rule enumerating 7110, 7115, 7120 one by one does not — the day a new tax sub-account opens, the range rule already covers it while the enumerated rule silently drops it into review. The data model captures a posted line, a range rule, and a lease override as three frozen dataclasses, with a RuleSource enum so every resolved tag records which tier decided it.

from __future__ import annotations

from dataclasses import dataclass, field
from decimal import Decimal, ROUND_HALF_UP
from enum import Enum

CENTS = Decimal("0.01")


class ControlClass(str, Enum):
    """The controllability label a cap engine reads before it clamps a pool."""
    CONTROLLABLE = "controllable"
    NON_CONTROLLABLE = "non_controllable"
    REVIEW = "review"                # unresolved: routed to a human, never inferred


class RuleSource(str, Enum):
    """Which tier decided a line's label—kept for an auditable trail."""
    LEASE_OVERRIDE = "lease_override"
    PORTFOLIO_DEFAULT = "portfolio_default"
    UNRESOLVED = "unresolved"


@dataclass(frozen=True)
class PostedLine:
    """A recoverable general-ledger line awaiting a controllability tag."""
    line_id: str
    gl_account: str                  # numeric GL code, e.g. "6420" grounds, "7110" taxes
    memo: str
    amount: Decimal


@dataclass(frozen=True)
class GLRule:
    """A portfolio default covering a GL account RANGE, not a single code."""
    rule_id: str
    low: str                         # inclusive lower GL code
    high: str                        # inclusive upper GL code
    control_class: ControlClass


@dataclass(frozen=True)
class LeaseOverride:
    """One lease redefining a single account's controllability by clause."""
    gl_account: str
    control_class: ControlClass
    lease_version: str
    clause_ref: str


def build_rule_map() -> list[GLRule]:
    """Assemble book-wide defaults as account RANGES, ordered narrowest first
    so the most specific range wins if two were ever to overlap."""
    rules = [
        GLRule("R-grounds", "6400", "6499", ControlClass.CONTROLLABLE),        # landscaping, sweeping
        GLRule("R-mgmt", "6500", "6599", ControlClass.CONTROLLABLE),           # management fees
        GLRule("R-supplies", "6600", "6699", ControlClass.CONTROLLABLE),       # common-area supplies
        GLRule("R-tax", "7100", "7199", ControlClass.NON_CONTROLLABLE),        # real-estate taxes
        GLRule("R-insurance", "7200", "7299", ControlClass.NON_CONTROLLABLE),  # premiums
        GLRule("R-utility", "7300", "7399", ControlClass.NON_CONTROLLABLE),    # utilities, snow/ice
    ]
    return sorted(rules, key=lambda r: int(r.high) - int(r.low))


def match_portfolio(account: str, rules: list[GLRule]) -> GLRule | None:
    """Return the first portfolio range that contains the account, or None."""
    code = int(account)
    for rule in rules:
        if int(rule.low) <= code <= int(rule.high):
            return rule
    return None

Step 2 — Register lease-level overrides #

An override is a lease redefining what “controllable” means for its own cap — one anchor tenant folds common-area utilities into the controllable pool it caps; another carves porter labor out of it. Overrides are keyed to a lease_version, and only the version being reconciled is indexed, so reopening a prior year replays that year’s boundary rather than today’s. Indexing by account also makes the precedence lookup in the next step an O(1) dictionary hit instead of a scan.

@dataclass(frozen=True)
class Classification:
    """A resolved controllability tag plus the provenance that justifies it."""
    line_id: str
    gl_account: str
    control_class: ControlClass
    source: RuleSource
    rule_ref: str                    # rule_id or clause_ref; "" when unresolved
    amount: Decimal


def index_overrides(
    overrides: list[LeaseOverride],
    lease_version: str,
) -> dict[str, LeaseOverride]:
    """Index a lease's overrides by account for the version being reconciled."""
    return {
        override.gl_account: override
        for override in overrides
        if override.lease_version == lease_version
    }

Step 3 — Apply the classifier under strict precedence #

The resolver encodes the precedence directly: a lease override for the account wins; absent one, the portfolio range default applies; absent both, the line is tagged REVIEW. Formally the resolved class is a pure function of the account and the two rule tiers,

where $o$ is the override map and $d$ the portfolio default. Because resolution depends only on the account — never on line order, amount, or memo — rerunning the classifier over the same ledger with the same rule versions yields a bit-for-bit identical partition, which is exactly the determinism an audit relies on when a statement is reopened years later. Each result carries its source and rule_ref, so a reviewer can trace any tag to the clause or portfolio rule that produced it.

def classify_line(
    line: PostedLine,
    rules: list[GLRule],
    override_index: dict[str, LeaseOverride],
) -> Classification:
    """Resolve one line by strict precedence: a lease override outranks the
    portfolio default, and an account no tier resolves is sent to REVIEW."""
    override = override_index.get(line.gl_account)
    if override is not None:                       # precedence 1: the lease redefines it
        return Classification(
            line.line_id, line.gl_account, override.control_class,
            RuleSource.LEASE_OVERRIDE, override.clause_ref, line.amount,
        )

    default = match_portfolio(line.gl_account, rules)
    if default is not None:                        # precedence 2: the book-wide default
        return Classification(
            line.line_id, line.gl_account, default.control_class,
            RuleSource.PORTFOLIO_DEFAULT, default.rule_id, line.amount,
        )

    return Classification(                         # precedence 3: never guess an unknown
        line.line_id, line.gl_account, ControlClass.REVIEW,
        RuleSource.UNRESOLVED, "", line.amount,
    )

Step 4 — Emit the tagged ledger #

The classifier’s output is a TaggedLedger: the full list of tagged lines plus three subtotals derived from them. Deriving the buckets from the tagged rows, rather than accumulating them as the loop runs, keeps a single source of truth and lets the conservation invariant act as a proof — the controllable, non-controllable, and review totals must reconstruct the pool the classifier received, to the cent. Only the controllable_total is eligible to reach a cap; the non-controllable total is recovered uncapped, and any dollar in review blocks the reconciliation until it is cleared.

@dataclass
class TaggedLedger:
    """The classifier's output: per-line tags plus conserved bucket totals."""
    rows: list[Classification] = field(default_factory=list)

    def _total(self, cls: ControlClass) -> Decimal:
        subtotal = sum(
            (row.amount for row in self.rows if row.control_class is cls),
            Decimal("0.00"),
        )
        return subtotal.quantize(CENTS, rounding=ROUND_HALF_UP)

    @property
    def controllable_total(self) -> Decimal:
        return self._total(ControlClass.CONTROLLABLE)

    @property
    def non_controllable_total(self) -> Decimal:
        return self._total(ControlClass.NON_CONTROLLABLE)

    @property
    def review_total(self) -> Decimal:
        return self._total(ControlClass.REVIEW)

    @property
    def pool_total(self) -> Decimal:
        """The conservation base: every tagged dollar, whatever its bucket."""
        return (
            self.controllable_total
            + self.non_controllable_total
            + self.review_total
        )


def classify_ledger(
    lines: list[PostedLine],
    rules: list[GLRule],
    overrides: list[LeaseOverride],
    lease_version: str,
) -> TaggedLedger:
    """Tag an entire recoverable pool for one lease version."""
    index = index_overrides(overrides, lease_version)
    return TaggedLedger([classify_line(line, rules, index) for line in lines])

Step 5 — Drift-check the classification against the prior period #

The failure a static classifier cannot catch on its own is a boundary that moved between periods — a new override that flips a tax account into controllable, or a chart migration that reassigns an account into a different default range. detect_drift compares this period’s tag for each account against the prior period’s and returns only the accounts that changed. The drift set is simply

Most drift is legitimate — a lease renewal renegotiated the cap definition — but every entry deserves a human glance before the statement goes out, because the one illegitimate flip in the set is the one that overbills a tenant.

def detect_drift(
    prior: TaggedLedger,
    current: TaggedLedger,
) -> dict[str, tuple[ControlClass, ControlClass]]:
    """Flag accounts whose label changed between two periods. A silent flip—
    a tax range that became controllable—is the failure this catches."""
    prior_by_account = {row.gl_account: row.control_class for row in prior.rows}
    drift: dict[str, tuple[ControlClass, ControlClass]] = {}
    for row in current.rows:
        was = prior_by_account.get(row.gl_account)
        if was is not None and was is not row.control_class:
            drift[row.gl_account] = (was, row.control_class)
    return drift

Gotchas & Known Limitations #

• Non-numeric GL codes break the range match. match_portfolio calls int() on the account, so an alphanumeric code such as CAM-TAX raises a ValueError rather than falling through to review. A production book with mixed code formats should normalize codes upstream in GL code mapping for CAM expenses, or the matcher must be widened to a prefix or lookup strategy before it meets a real chart of accounts.
• A single line cannot straddle the boundary. The classic case is a capital improvement whose gross cost is non-controllable but whose lease-permitted amortized annual slice re-enters the controllable pool. A one-tag-per-line classifier cannot represent that split. The correct fix is upstream: the amortized slice is posted as its own GL line with its own account, so the classifier sees two clean lines instead of one it must divide.
• Overrides must never widen their blast radius. An override targets one named account, by design — it is a dictionary keyed on a single code, not a range. Letting an override span a range would quietly recapture accounts the clause never mentioned. Keep overrides account-specific and let only the portfolio defaults use ranges.
• A category that must never be capped needs a hard floor. Real-estate taxes are the archetypal line no legitimate lease caps as controllable. A defensive layer can pin such account ranges as non-overridable so that even a malformed override cannot pull a tax code under a cap; the drift check is the detective control, but a hard floor is the preventive one.
• A non-empty review bucket blocks the run. Any dollar in review is an account the rules could not resolve — a new vendor code, a post-acquisition migration, a clause the abstraction has not captured yet. The reconciliation must halt until review clears to zero, because a cap applied to a pool that still hides unclassified dollars produces a confidently wrong statement.

Verification #

Verification rests on three properties: an override outranks the default and stays traceable to its clause, the buckets conserve the pool exactly, and the drift check catches a flipped tax range. Because every amount is Decimal, the conservation assertion uses exact equality rather than a floating-point tolerance, so a one-cent leak fails the test instead of hiding under an approximate comparison.

from decimal import Decimal


def _lines() -> list[PostedLine]:
    # Harbor Point Plaza, tenant suite 210, 2026 recoverable pool.
    return [
        PostedLine("H1", "6410", "Grounds & landscaping", Decimal("12750.00")),
        PostedLine("H2", "6520", "Property management fee", Decimal("8600.00")),
        PostedLine("H3", "7110", "Real-estate taxes", Decimal("54300.00")),
        PostedLine("H4", "7320", "Common-area electricity", Decimal("16240.00")),
    ]


def test_override_beats_default_and_stays_traceable() -> None:
    rules = build_rule_map()
    # Suite 210's lease folds utilities into its controllable cap (clause 7.2).
    overrides = [LeaseOverride("7320", ControlClass.CONTROLLABLE, "2026.1", "§7.2(a)")]
    ledger = classify_ledger(_lines(), rules, overrides, "2026.1")

    assert ledger.controllable_total == Decimal("37590.00")   # 12750 + 8600 + 16240
    assert ledger.non_controllable_total == Decimal("54300.00")
    assert ledger.review_total == Decimal("0.00")

    utility = next(row for row in ledger.rows if row.line_id == "H4")
    assert utility.control_class is ControlClass.CONTROLLABLE
    assert utility.source is RuleSource.LEASE_OVERRIDE
    assert utility.rule_ref == "§7.2(a)"


def test_conservation_and_unmapped_is_quarantined() -> None:
    rules = build_rule_map()
    lines = _lines() + [PostedLine("H5", "9905", "New vendor, unmapped", Decimal("1980.00"))]
    ledger = classify_ledger(lines, rules, overrides=[], lease_version="2026.1")

    assert ledger.review_total == Decimal("1980.00")
    gross = sum((line.amount for line in lines), Decimal("0.00"))
    assert ledger.pool_total == gross     # exact, because every amount is Decimal


def test_drift_catches_a_flipped_tax_range() -> None:
    rules = build_rule_map()
    prior = classify_ledger(_lines(), rules, overrides=[], lease_version="2026.1")
    # A bad override this year flips a tax account into controllable.
    bad = [LeaseOverride("7110", ControlClass.CONTROLLABLE, "2026.1", "§?")]
    current = classify_ledger(_lines(), rules, bad, "2026.1")

    drift = detect_drift(prior, current)
    assert drift["7110"] == (ControlClass.NON_CONTROLLABLE, ControlClass.CONTROLLABLE)

Run the tests with pytest, or exercise the classifier interactively against a fixture whose correct partition you already know from the lease clause. The override test guards the case a portfolio-wide default cannot express — a single tenant that moved the boundary — and proves the redefinition applies to that ledger without disturbing the book-wide labeling every other property depends on. Once these pass, the controllable_total this classifier emits is ready to hand to the cap stage in applying cumulative caps to controllable CAM, where the segregated bucket built here is finally clamped by the lease ceiling — the calculation this whole classification effort exists to feed, and the reason the controllable vs non-controllable expense segregation layer insists the boundary be right before any dollar is billed.

Related #

• Controllable vs non-controllable expense segregation — the segregation architecture and data contracts this classifier implements.
• Applying cumulative caps to controllable CAM — how the controllable bucket produced here is clamped by a compounding, base-year ceiling.
• GL code mapping for CAM expenses — the upstream stage that assigns the resolved accounts this classifier keys its rules to.