Escalation Formula Mapping

Escalation formula mapping is the layer that converts a lease’s rent-adjustment language — a fixed annual bump, a CPI-indexed reset, a negotiated step schedule, or a capped/floored hybrid — into a deterministic function that billing and forecasting systems can call without ever re-reading the contract. It sits inside Core Architecture & Lease Taxonomy, and it begins exactly where labeling ends: once a clause classification system has tagged a span as rent_escalation, this stage parses the actual rate, index, cadence, and constraint boundaries, then exposes a callable that returns the rent owed in any month of the term. Get this layer wrong and a single mis-mapped cap silently overstates net operating income across an entire portfolio.

The specific challenge this page solves: escalation prose is wildly inconsistent across regions, asset classes, and drafting attorneys, yet the output has to be a single, precise, reproducible number that an auditor can reconstruct years later. The pattern used in production lease pipelines is to decouple the contractual trigger from the calculation engine — extraction emits a strict data contract, and a side-effect-free engine consumes it. Everything below is the engineering of that contract and that engine.

Where this fits in the pipeline

Mapping is one stage in a longer chain, and keeping its responsibilities narrow is what makes it auditable:

• Upstream of it: document ingestion, clause classification, and the metadata normalization standards that guarantee dates arrive as ISO-8601 and amounts arrive as typed decimals rather than free text.
• This stage: parse the escalation parameters into a validated EscalationFormula, and expose a pure function rent_for(month) that never mutates state and never hits the network mid-calculation.
• Downstream of it: the validated formula feeds lease data models for storage, and ambiguous or low-confidence parses divert to fallback routing logic for manual review rather than emitting a guessed number.

A mapper that also tries to classify the clause, or resolve amendment precedence, becomes impossible to reason about. It answers one question — “given this term and this month, what is the rent?” — and hands everything else off.

Prerequisites and environment setup

The reference implementation targets Python 3.11+ and a deliberately small dependency set so the engine stays fast, vendorable into a worker image, and free of floating-point surprises.

Install with pip install "pydantic>=2.6" python-dateutil. Two environment assumptions matter. First, every monetary value enters as Decimal, never float — floating-point arithmetic introduces rounding drift that compounds across a multi-year, multi-tenant portfolio, so the abstraction layer must coerce amounts to Decimal at the ingestion boundary. Second, index values (CPI, RPI) are passed in by the caller from an internal index store; the engine never fetches them inline, which keeps rent_for() pure, idempotent, and unit-testable without network mocks.

Escalation decision logic

Routing is a layered decision keyed on escalation_type, not a single branch. The table below is the operational contract the implementation encodes — read it as the spec, the code as the proof.

The math each branch encodes is small and explicit. A fixed-percentage term compounds the rate once per completed cadence:

$$ R_n = R_0 ,(1 + r)^{\lfloor n / m \rfloor} $$

A CPI-indexed term applies the year-over-year index change to the base, clamped between the negotiated floor and cap:

$$ R_n = R_0 \left( 1 + \min!\Big( \max\big( \tfrac{I_n - I_0}{I_0},; f \big),; c \Big) \right) $$

where ( R_0 ) is the base rent, ( r ) the fixed rate, ( m ) the cadence in months, ( n ) the months elapsed, ( I_n ) and ( I_0 ) the current and base index values, and ( f ) and ( c ) the floor and cap as fractions. The single most common production bug is omitting the floor: when CPI prints negative, an unclamped formula reduces rent, which almost no commercial lease actually permits.

Primary implementation

The contract below is engineered for the lease domain: a frozen pydantic v2 model that rejects inconsistent terms at construction time, and a pure calculation method that applies ROUND_HALF_UP only at the final step so intermediate truncation never accumulates. The frozen=True config makes each instance an immutable source of truth that billing, forecasting, and audit tools can all consume identically.

from __future__ import annotations

from datetime import date
from decimal import Decimal, ROUND_HALF_UP
from enum import Enum
from typing import Optional

from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

CENTS = Decimal("0.01")


class EscalationType(str, Enum):
    FIXED_PCT = "fixed_pct"
    CPI_INDEXED = "cpi_indexed"
    STEP_RENT = "step_rent"


class EscalationFormula(BaseModel):
    """Immutable, auditable contract for one lease escalation term."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    lease_id: str = Field(min_length=1)
    effective_start: date
    effective_end: Optional[date] = None
    base_amount: Decimal = Field(gt=0, decimal_places=2)
    escalation_type: EscalationType
    # Fixed: percent per period. CPI: the base index value the lease was struck at.
    rate_or_index: Optional[Decimal] = None
    frequency_months: int = Field(gt=0, le=120)
    cap_pct: Optional[Decimal] = Field(default=None, ge=0)
    floor_pct: Optional[Decimal] = Field(default=None, ge=0)
    # Step rent: {month_offset_from_start: absolute_rent_amount}
    step_schedule: Optional[dict[int, Decimal]] = None

    @field_validator("step_schedule")
    @classmethod
    def _steps_non_negative(cls, v: Optional[dict[int, Decimal]]):
        if v and any(k < 0 or amt < 0 for k, amt in v.items()):
            raise ValueError("step_schedule offsets and amounts must be non-negative")
        return v

    @model_validator(mode="after")
    def _check_consistency(self) -> "EscalationFormula":
        # Temporal sanity — an end before a start corrupts every downstream rollup.
        if self.effective_end and self.effective_end <= self.effective_start:
            raise ValueError("effective_end must be after effective_start")
        # A cap below a floor is an unsatisfiable constraint, almost always an
        # extraction error where the two values were transposed.
        if (
            self.cap_pct is not None
            and self.floor_pct is not None
            and self.cap_pct < self.floor_pct
        ):
            raise ValueError("cap_pct cannot be lower than floor_pct")
        # Each escalation type carries its own non-negotiable parameter.
        if self.escalation_type is EscalationType.FIXED_PCT and self.rate_or_index is None:
            raise ValueError("fixed_pct escalation requires rate_or_index")
        if self.escalation_type is EscalationType.STEP_RENT and not self.step_schedule:
            raise ValueError("step_rent escalation requires a step_schedule")
        if self.escalation_type is EscalationType.CPI_INDEXED and self.rate_or_index is None:
            raise ValueError("cpi_indexed escalation requires a base index (rate_or_index)")
        return self

    def rent_for(
        self,
        months_elapsed: int,
        current_index: Optional[Decimal] = None,
        prior_index: Optional[Decimal] = None,
    ) -> Decimal:
        """Return the rent owed `months_elapsed` after effective_start.

        Pure and idempotent: identical inputs always yield identical Decimals,
        rounded HALF_UP only once, at the very end.
        """
        if months_elapsed < 0:
            raise ValueError("months_elapsed must be non-negative")

        if self.escalation_type is EscalationType.STEP_RENT:
            return self._step_rent(months_elapsed)
        if self.escalation_type is EscalationType.FIXED_PCT:
            return self._fixed_pct(months_elapsed)
        return self._cpi_indexed(current_index, prior_index)

    def _step_rent(self, months_elapsed: int) -> Decimal:
        # The authoritative amount is the latest step at or before this month;
        # before the first step, the base amount governs.
        applicable = [k for k in self.step_schedule if k <= months_elapsed]
        amount = self.step_schedule[max(applicable)] if applicable else self.base_amount
        return amount.quantize(CENTS, rounding=ROUND_HALF_UP)

    def _fixed_pct(self, months_elapsed: int) -> Decimal:
        rate = self.rate_or_index / Decimal("100")
        periods = months_elapsed // self.frequency_months
        return (self.base_amount * (Decimal("1") + rate) ** periods).quantize(
            CENTS, rounding=ROUND_HALF_UP
        )

    def _cpi_indexed(
        self, current_index: Optional[Decimal], prior_index: Optional[Decimal]
    ) -> Decimal:
        if current_index is None or prior_index is None:
            raise ValueError("cpi_indexed calculation requires current and prior index")
        if prior_index == 0:
            raise ValueError("prior index cannot be zero")

        change = (current_index - prior_index) / prior_index
        # Clamp BEFORE applying — the floor protects against deflationary CPI
        # cutting rent, which commercial leases almost never permit.
        if self.floor_pct is not None:
            change = max(change, self.floor_pct / Decimal("100"))
        if self.cap_pct is not None:
            change = min(change, self.cap_pct / Decimal("100"))
        return (self.base_amount * (Decimal("1") + change)).quantize(
            CENTS, rounding=ROUND_HALF_UP
        )

Validation and quality gates

The constructor already rejects structurally impossible terms, but a production mapper needs gates that catch semantically wrong parses before they reach billing. Three checks earn their keep.

First, treat the parse as untrusted and attach a confidence score; anything below threshold is diverted rather than emitted. The mapper should never invent a number it is unsure of.

from pydantic import BaseModel
from typing import Optional


class MappedEscalation(BaseModel):
    formula: EscalationFormula
    confidence: float
    needs_review: bool = False
    is_provisional: bool = False
    review_reason: Optional[str] = None


CONFIDENCE_THRESHOLD = 0.80


def gate(mapped: MappedEscalation) -> MappedEscalation:
    """Divert low-confidence or self-contradictory parses to manual review."""
    reasons = []
    if mapped.confidence < CONFIDENCE_THRESHOLD:
        reasons.append(f"confidence {mapped.confidence:.2f} below {CONFIDENCE_THRESHOLD}")
    f = mapped.formula
    # A CPI lease with no cap is unusual enough in some portfolios to flag.
    if f.escalation_type is EscalationType.CPI_INDEXED and f.cap_pct is None:
        reasons.append("cpi_indexed term has no cap — verify against the executed lease")
    if reasons:
        return mapped.model_copy(
            update={"needs_review": True, "review_reason": "; ".join(reasons)}
        )
    return mapped

Second, enforce idempotency as a test invariant: calling rent_for(n) twice, or recomputing the whole schedule, must produce byte-identical Decimal values. Property-based tests over random months and index series catch any accidental statefulness.

Third, route schema failures, not just low confidence, to a dead-letter path. A ValidationError from the pydantic constructor carries the original payload; preserve it intact so a fixed pattern can replay the record later instead of losing the lease entirely.

Troubleshooting

The delayed-index case is the most common operational failure in legacy portfolios. When the official CPI release lags the billing date, queue the calculation, substitute a trailing average, flag the record, and reconcile on publication rather than blocking the run:

def provisional_index(history: list[Decimal], window: int = 12) -> Decimal:
    """Trailing-average stand-in when the official index print is late."""
    if not history:
        raise ValueError("no index history available for provisional rate")
    recent = history[-window:]
    return sum(recent) / Decimal(len(recent))

The reconciliation pass later recomputes with the official value and posts the delta, so the books self-correct without a human chasing every late release.

Performance and scale notes

For portfolio-wide rent roll generation the engine is CPU-bound on Decimal math, not I/O, so the wins come from doing less redundant work. Construct each EscalationFormula once and cache it keyed by lease_id plus an amendment version — re-validating an immutable, frozen model on every billing cycle is pure waste. Batch index lookups: fetch the full month’s CPI/RPI table once and pass slices into rent_for() rather than querying per lease. Because the method is pure, schedule generation parallelizes cleanly across processes; partition by property so each worker owns a disjoint lease set and no shared state needs locking. For multi-year forecasts, memoize the compounded factor (1 + r) ** periods per (rate, periods) pair, since thousands of leases on the same fixed step share identical factors.

Amendments deserve their own note: never mutate a formula in place. Maintain a timeline of versioned EscalationFormula instances, each bounded by effective_start and effective_end, so historical rent rolls remain reproducible and an audit can replay any prior period exactly as it was billed.

Frequently asked questions

Related

• Handling CAM Charge Variations in Lease Taxonomy Design — why operating-expense recoveries route through a separate taxonomy from base-rent escalations.
• Clause Classification Systems — the upstream stage that tags a span rent_escalation before this engine parses it.
• Lease Data Models — where versioned formulas are stored and amendment precedence is resolved.
• Metadata Normalization Standards — the ISO-8601 and typed-decimal contract this engine depends on at its input boundary.
• Fallback Routing Logic — where low-confidence and self-contradictory parses divert for manual review.

← Back to Core Architecture & Lease Taxonomy

← Back to Core Architecture & Lease Taxonomy