Jurisdiction-Specific Fringe Multipliers in Python

A production that photographs its first week in Georgia, moves the unit to New Mexico for two days, and finishes on a Los Angeles stage owes fringe contributions at a different multiplier for the same performer, in the same classification, on consecutive shooting days — and the only variable that changed is where the work was physically performed. The moment a fringe rate is written as a single hardcoded percentage, that cross-state shoot is already miscalculating, because one number cannot be simultaneously correct in three jurisdictions on three dates. The narrow problem this page solves is deterministic multiplier resolution: given a jurisdiction, a job classification, and the calendar day the work was performed, return exactly one fringe multiplier from a versioned rate table, apply it in Decimal, cap the running contribution against a per-employee period ceiling, and emit an audit record a trust-fund examiner or completion guarantor can recompute months later. It is the specific resolution step the parent Pension & Health Fund Calculations engine assumes exists before it applies any rate.

Prerequisites and Context

This page extends the Pension & Health Fund Calculations subsystem within the broader Guild Compliance & Rule Validation Automation reference architecture. It assumes the inbound lines are already clean: heterogeneous timecards and payroll exports are normalized upstream by the Cost Ingestion & Data Parsing Workflows subsystem, and each hour is keyed to exactly one fund code, department, and budget line through Cost Code Standardization. What this page adds on top of that contract is the resolver itself — the function that turns the triple (jurisdiction, classification, work date) into a single defensible multiplier.

The implementation targets Python 3.11+ for standard-library zoneinfo, and uses the same deliberately small stack as the rest of the section: Pydantic v2 for boundary validation (model_validate, field_validator, ConfigDict(frozen=True)); the standard-library decimal, hashlib, and zoneinfo modules for currency-safe arithmetic, deterministic hashing, and timezone-aware work dates. Never build a monetary value or a rate from float — as the official Python decimal module documentation makes explicit, only fixed-point arithmetic under an explicit rounding context reproduces the figure a trust fund’s own remittance math produces. The classifications in play span the guilds a scripted shoot touches at once — the Screen Actors Guild – American Federation of Television and Radio Artists (SAG-AFTRA) for performers, the Directors Guild of America (DGA) for directors and unit production managers, the Writers Guild of America (WGA) for writing staff, and the International Alliance of Theatrical Stage Employees (IATSE) for crew — but the resolver treats a classification as an opaque key, so it never needs to know which union a row belongs to in order to price it correctly.

Step-by-Step Implementation

The resolver has three responsibilities and keeps them separate. First, model each rate as an effective-dated row so the same jurisdiction and classification can carry different multipliers over time. Second, resolve exactly one row for a given work day — zero matches is a missing rate and more than one is an ambiguous overlap, and both are failures rather than a guess. Third, apply the multiplier in Decimal and accrue the result against a per-employee period cap held as a stateful accumulator, so a high-earning department head stops contributing at the guaranteed ceiling instead of over-remitting.

from __future__ import annotations

import hashlib
import json
from datetime import date, datetime
from decimal import Decimal, ROUND_HALF_UP
from zoneinfo import ZoneInfo

from pydantic import BaseModel, ConfigDict, field_validator

CENTS = Decimal("0.01")


class FringeRate(BaseModel):
    """One effective-dated fringe multiplier for a jurisdiction + classification."""

    model_config = ConfigDict(frozen=True)

    jurisdiction: str          # work-state code, e.g. "GA", "NM", "CA"
    classification: str        # opaque guild/work-classification key
    multiplier: Decimal        # fringe as a fraction of the contributory base
    period_cap: Decimal        # per-employee contribution cap for the period
    effective_from: date
    effective_to: date | None = None   # open-ended when None
    agreement_version: str

    @field_validator("multiplier", "period_cap", mode="before")
    @classmethod
    def _no_float(cls, v):
        if isinstance(v, float):
            raise ValueError("rate/money fields must be str or Decimal, never float")
        return Decimal(str(v))

    def covers(self, work_day: date) -> bool:
        if work_day < self.effective_from:
            return False
        return self.effective_to is None or work_day <= self.effective_to


class FringeRequest(BaseModel):
    model_config = ConfigDict(frozen=True)

    employee_id: str
    jurisdiction: str
    classification: str
    contributory_base: Decimal
    worked_at: datetime         # timezone-aware instant of the shoot day

    @field_validator("worked_at")
    @classmethod
    def _aware(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("worked_at must be timezone-aware (use an IANA zone)")
        return v

    @field_validator("contributory_base", mode="before")
    @classmethod
    def _base_decimal(cls, v):
        if isinstance(v, float):
            raise ValueError("contributory_base must be str/Decimal, never float")
        return Decimal(str(v))


class FringeResolver:
    """Resolves a fringe multiplier as a function of (jurisdiction, class, date)."""

    def __init__(self, table: list[FringeRate], work_zone: str = "America/Los_Angeles"):
        # Index by the composite key; each key owns a list of effective-dated rows.
        self._table: dict[tuple[str, str], list[FringeRate]] = {}
        for row in table:
            self._table.setdefault((row.jurisdiction, row.classification), []).append(row)
        self._zone = ZoneInfo(work_zone)
        # Accrued contribution per (employee, period) drives cap enforcement.
        self._accrued: dict[tuple[str, str], Decimal] = {}
        self.quarantine: list[dict] = []

    def _resolve_rate(self, req: FringeRequest, work_day: date) -> FringeRate | None:
        candidates = self._table.get((req.jurisdiction, req.classification), [])
        matches = [r for r in candidates if r.covers(work_day)]
        # Exactly one row must apply: zero is a missing rate, two is an overlap.
        return matches[0] if len(matches) == 1 else None

    def compute(self, req: FringeRequest, period_key: str) -> dict:
        # The jurisdiction is fixed per shoot day; derive the calendar day in the
        # work-state zone so a rate that turns over at midnight is applied right.
        work_day = req.worked_at.astimezone(self._zone).date()
        rate = self._resolve_rate(req, work_day)

        payload = {
            "employee_id": req.employee_id,
            "jurisdiction": req.jurisdiction,
            "classification": req.classification,
            "work_day": work_day.isoformat(),
            "contributory_base": str(req.contributory_base),
        }
        if rate is None:
            return self._quarantine(payload, "no_unique_effective_rate")

        uncapped = (req.contributory_base * rate.multiplier).quantize(
            CENTS, rounding=ROUND_HALF_UP
        )
        # Cap enforcement is stateful: only the headroom under the ceiling accrues.
        acc_key = (req.employee_id, period_key)
        already = self._accrued.get(acc_key, Decimal("0.00"))
        headroom = max(rate.period_cap - already, Decimal("0.00"))
        contribution = min(uncapped, headroom)
        self._accrued[acc_key] = already + contribution

        payload.update({
            "agreement_version": rate.agreement_version,
            "multiplier": str(rate.multiplier),
            "uncapped_contribution": str(uncapped),
            "period_cap": str(rate.period_cap),
            "capped_contribution": str(contribution),
            "period_accrued_after": str(self._accrued[acc_key]),
            "status": "calculated",
        })
        return self._stamp(payload)

    def _quarantine(self, payload: dict, reason: str) -> dict:
        record = {**payload, "status": "quarantined", "reason_code": reason,
                  "requires_manual_override": True}
        record = self._stamp(record)
        self.quarantine.append(record)
        return record

    @staticmethod
    def _stamp(payload: dict) -> dict:
        canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
        payload["payload_sha256"] = hashlib.sha256(canonical).hexdigest()
        return payload

Two design choices carry the audit weight. The composite key is (jurisdiction, classification) and never classification alone, so the New Mexico day and the California day for the same performer resolve independently rather than colliding on a shared rate. And the cap is a running accumulator keyed by employee and period, not a per-line check, because a ceiling only means anything when it remembers what already accrued — a per-transaction test would let every individual line pass while the sum sailed past the guaranteed maximum. Because the resolver reads its rates from an injected table and never touches the network, the same request and the same table always produce the same multiplier, the same capped contribution, and the same hash.

The resolution path filters the versioned rate table down to a single effective row before any arithmetic runs; a work day that matches zero rows or more than one is diverted rather than priced.

Jurisdiction-aware fringe multiplier resolution with a quarantine branch A fringe request carrying a jurisdiction, classification, and timezone-aware work date enters the resolver. The versioned rate table is filtered by the jurisdiction and classification key, and a decision node asks whether exactly one effective-dated row covers the work day. On the resolved path the date-effective multiplier is selected, applied to the contributory base in Decimal, and accrued against the per-employee period cap before an immutable audit entry is written with the multiplier, agreement version, and a SHA-256 hash. On the unresolved path — zero matching rows or an ambiguous overlap — the record is quarantined with a reason code and a SHA-256 payload hash, routed to the fallback chain rather than zero-accrued, and surfaced for manual override, with the exception rejoining the same audit log. Yes — one row none / overlap exception rejoins the ledger Fringe request jurisdiction · class · work date Filter versioned rate table by jurisdiction + classification Effective row for work day? Select date-effective multiplier effective_from ≤ day ≤ effective_to Apply base · accrue toward cap Decimal · per-employee accumulator Immutable audit entry multiplier · version · SHA-256 Quarantine record reason code · SHA-256 payload Route to fallback chain conservative · never zero-accrued Manual override exception report · triage resolved contribution quarantined exception

Audit Trail Requirements

Every evaluation — the capped calculated line and the quarantined exception alike — must serialize to a write-once store so an examiner reads an unbroken sequence rather than only the lines that resolved. At minimum the persisted payload carries: the employee_id, jurisdiction, classification, and the resolved work_day in the work-state zone; the contributory_base and the agreement_version of the row that priced it; the multiplier, the uncapped_contribution, the governing period_cap, the capped_contribution, and the period_accrued_after value that proves where the running total stood after this line; the status; and the payload_sha256 fingerprint of the canonical payload. Because the multiplier and the agreement version travel with the number, a trust fund questioning a single Georgia day a year later can be answered by the archived record alone — the rate that produced the figure is recoverable, not reconstructed from memory.

Persist each record to append-only, write-once storage — object storage under an immutability lock, or an event-sourced table with no in-place update — before any downstream remittance transaction commits, so a crash mid-close leaves a replayable record of intent rather than a silent gap. A later correction is a new compensating entry, never an edit of the posted one, so the accrued figure a remittance was built on stays recoverable. The period_accrued_after field is what makes the cap itself auditable: an examiner can walk the sequence of lines for one employee and confirm the accumulator only ever climbed toward the ceiling and stopped there.

Gotchas and Production Edge Cases

Per-day jurisdiction, not per-shoot. The jurisdiction that governs a fringe multiplier is the state where the day was physically worked, resolved one shoot day at a time — never a single production-level default. A unit that hard-codes the home-office state silently prices every travelling day at the wrong multiplier. Store the jurisdiction on each request, as the model above does, and let a multi-state week produce a different multiplier on each day. The same per-unit discipline applied to cross-border cost data is worked through in Async Batch Processing for Multi-Currency Shoots.

Date-effective rates, resolved by work date. A fringe schedule is not one number but a succession of effective-dated rows, because a bargaining cycle or a state-mandated change moves the multiplier on a specific date. Resolve the row by the day the work was performed, not the day the close happens to run, so a rate that rose on July 1 never retroactively reprices a June timecard. Modeling effective_from and effective_to on every row — and requiring exactly one to cover the work day — is what turns a mid-shoot rate change from a silent repricing into a resolved, logged fact.

The cap is a stateful accumulator. A per-employee period ceiling only holds if it remembers prior accrual. Checking each line against the raw cap in isolation lets every individual line pass while the running total overshoots; the accumulator keyed by (employee_id, period_key) is what actually stops contributions at the guaranteed maximum. Note too that the cap belongs to the employee and the period, not the jurisdiction — a performer whose contributory earnings straddle two states still shares one ceiling, so the accumulator must span jurisdictions even as the multiplier changes beneath it.

Ambiguity is a failure, not a default. Zero matching rows means the rate is missing; two or more means the effective-date ranges overlap and the table is wrong. Neither is a candidate for a guessed multiplier. The resolver quarantines both, and when the cause is a genuinely absent schedule at close time, that record follows the tiered, over-accruing path defined in Compliance Fallback Chains rather than defaulting to a number no auditor can trace.

Float contamination. Never construct a Decimal from a float literal — Decimal(0.075) imports binary rounding error into the multiplier. Read every rate and base through its string form, as Decimal(str(...)) above, so a single fractional cent compounded across a multi-state week never becomes the variance a guarantor asks you to explain. Idempotent replay depends on the same discipline: because the payload hash is deterministic and the store is append-only, re-running a close after a partial failure is safe only if the consumer deduplicates on (employee_id, work_day, payload_sha256).

Up one level: Pension & Health Fund Calculations, part of Guild Compliance & Rule Validation Automation.