Automating SAG-AFTRA Pension Contributions Tracking in Python

The specific automation task on this page is narrow and unforgiving: given a stream of validated payroll lines for a performer, compute the employer pension contribution that is actually owed to the Screen Actors Guild-American Federation of Television and Radio Artists (SAG-AFTRA) trust — no more, no less — while a per-plan-year contribution ceiling caps the base and a shifting mix of contributory and non-contributory pay codes tries to distort it. This is the edge case the parent Pension & Health Fund Calculations engine defers to it: before a rate is applied, the pensionable base must be isolated from meal penalties, rest-period premiums, per diem, and reimbursements, and a running fiscal-year total must decide how much of a given line still fits under the ceiling. Get the segregation wrong and you either over-remit — money the production cannot claw back — or under-remit, which surfaces months later as a trust-fund audit finding, a held payment, or a completion guarantor refusing to release the bond until the discrepancy is reconciled. A spreadsheet cannot hold a per-performer running total across a fractured shooting schedule and stay defensible; a deterministic Python ledger can.

Prerequisites and Context

This page extends the Pension & Health Fund Calculations guide within the broader Guild Compliance & Rule Validation Automation architecture. It assumes the incoming lines are already normalized: heterogeneous timecards and payroll exports are cleaned upstream by the Cost Ingestion & Data Parsing Workflows subsystem, where Schema Validation & Error Handling quarantines malformed payloads and Async Batch Processing absorbs vendor API rate limits. What this page adds is the SAG-AFTRA-specific segregation and accrual logic that the generic contribution engine intentionally does not own.

The implementation targets Python 3.11+ for standard-library zoneinfo, and leans on a small stack: Pydantic v2 for boundary validation (model_validate, field_validator); the standard-library decimal, hashlib, and zoneinfo modules for currency-safe arithmetic, deterministic audit hashing, and timezone-aware work dates. Never use float for money — as the official Python decimal module documentation makes explicit, only an explicit precision and rounding context yields the reproducible output a trust fund’s remittance specification demands. On the contract side, the operative clauses are the SAG-AFTRA Pension Plan employer-contribution rate applied to covered (contributory) earnings and the annual contribution ceiling above which no further pension accrues for that performer in the plan year. Which secondary payments carry a contribution at all — reuse and residual income in particular — is resolved separately by the reuse-window rules in SAG-AFTRA Residuals Logic before a line ever reaches this ledger.

Step-by-Step Implementation

The logic proceeds in four steps: classify each pay code as contributory or not, validate the line at the boundary, fix its provenance with a canonical hash, then accrue against a per-performer, per-plan-year ceiling that stops the moment headroom is exhausted.

Step 1 — Classify pay codes. A PayCode enum carries an explicit contributory flag. Only contributory codes feed the pensionable base; meal penalties, rest-period premiums, per diem, and mileage are masked out. This classification is the single most audited decision on the page, so it is data, not an if buried in a loop.

Step 2 — Validate at the boundary. Each PayrollLine is a Pydantic v2 model: amount is a non-negative Decimal, and worked_at must be timezone-aware, because a contribution ceiling that resets on a plan-year boundary cannot be resolved from an ambiguous local timestamp.

Step 3 — Fix provenance before arithmetic. The line is serialized to canonical, key-sorted JSON and hashed with SHA-256, so re-running a disputed week yields an identical fingerprint — the idempotency property that lets an accountant replay the close deterministically.

Step 4 — Accrue against the ceiling. The ledger holds a running contributory total per performer per plan year. The portion of a new contributory line that still fits under the ceiling is min(line_amount, ceiling − accrued_so_far); the pension contribution is that headroom-clamped base times the employer rate, quantized to whole cents.

from __future__ import annotations

import hashlib
import json
from datetime import datetime
from decimal import Decimal, ROUND_HALF_UP, getcontext
from enum import Enum
from zoneinfo import ZoneInfo

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

# Deterministic context for all fringe arithmetic — never float for money.
getcontext().prec = 34
CENTS = Decimal("0.01")
PRODUCTION_TZ = ZoneInfo("America/Los_Angeles")

# Illustrative placeholders — load the live rate and ceiling from a signed,
# version-controlled schedule; they change every bargaining cycle.
EMPLOYER_PENSION_RATE = Decimal("0.06")
ANNUAL_CONTRIBUTION_CEILING = Decimal("250000.00")


class PayCode(Enum):
    """Each code declares whether it feeds the pensionable base."""
    SESSION_FEE = ("SESSION_FEE", True)
    ADJUSTED_REST = ("ADJUSTED_REST", True)      # contributory premium
    OVERTIME = ("OVERTIME", True)
    MEAL_PENALTY = ("MEAL_PENALTY", False)       # punitive, non-contributory
    PER_DIEM = ("PER_DIEM", False)
    MILEAGE = ("MILEAGE", False)

    def __init__(self, code: str, contributory: bool):
        self.code = code
        self.contributory = contributory


class PayrollLine(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True)

    performer_id: str = Field(min_length=1)
    pay_code: PayCode
    amount: Decimal
    worked_at: datetime

    @field_validator("amount")
    @classmethod
    def _non_negative(cls, v: Decimal) -> Decimal:
        if v < 0:
            raise ValueError("pay amount cannot be negative")
        return v

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

    def plan_year(self) -> int:
        # Normalize to the production timezone before deriving the plan year,
        # so a late-night wrap on Dec 31 lands in the correct year.
        return self.worked_at.astimezone(PRODUCTION_TZ).year

    def audit_hash(self) -> str:
        canonical = json.dumps(
            {
                "performer_id": self.performer_id,
                "pay_code": self.pay_code.code,
                "amount": str(self.amount),
                "worked_at": self.worked_at.astimezone(PRODUCTION_TZ).isoformat(),
            },
            sort_keys=True,
            separators=(",", ":"),
        )
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


class ContributionEntry(BaseModel):
    audit_hash: str
    performer_id: str
    plan_year: int
    contributory_base: Decimal   # portion of this line under the ceiling
    pension: Decimal
    ceiling_reached: bool
    status: str
    computed_at: datetime


class PensionAccrualLedger:
    """Tracks running contributory earnings per performer per plan year."""

    def __init__(self, rate: Decimal, ceiling: Decimal):
        self._rate = rate
        self._ceiling = ceiling
        self._accrued: dict[tuple[str, int], Decimal] = {}
        self.quarantine: list[dict] = []

    def apply(self, line: PayrollLine) -> ContributionEntry:
        key = (line.performer_id, line.plan_year())
        accrued = self._accrued.get(key, Decimal("0.00"))

        # Non-contributory codes never touch the pensionable base.
        if not line.pay_code.contributory:
            base = Decimal("0.00")
        else:
            headroom = max(self._ceiling - accrued, Decimal("0.00"))
            base = min(line.amount, headroom)
            self._accrued[key] = accrued + base

        pension = (base * self._rate).quantize(CENTS, ROUND_HALF_UP)
        return ContributionEntry(
            audit_hash=line.audit_hash(),
            performer_id=line.performer_id,
            plan_year=line.plan_year(),
            contributory_base=base,
            pension=pension,
            ceiling_reached=self._accrued.get(key, accrued) >= self._ceiling,
            status="calculated",
            computed_at=datetime.now(PRODUCTION_TZ),
        )


def ingest(raw: dict, ledger: PensionAccrualLedger) -> ContributionEntry | None:
    """Boundary parse: malformed records are logged, never silently coerced."""
    try:
        line = PayrollLine.model_validate(raw)
    except ValidationError as exc:
        ledger.quarantine.append({"reason": "boundary_validation", "errors": exc.errors()})
        return None
    return ledger.apply(line)

Two properties are load-bearing for audit defensibility. First, every monetary value is Decimal, and min/max/subtraction on Decimal preserve that type, so no float ever touches a contribution. Second, worked_at is normalized to a single production timezone via zoneinfo before the plan year and the hash are derived, so the same shoot day always canonicalizes into the same year and the same fingerprint regardless of the offset the source system emitted.

The decision path — mask non-contributory codes, clamp the contributory base to the remaining ceiling headroom, then apply the employer rate and write an immutable entry — is summarized below, including the quarantine branch for records that fail at the boundary.

SAG-AFTRA pension accrual decision path from raw payroll line to immutable ledger entry A raw payroll line is parsed at the boundary with Pydantic model_validate. A first decision asks whether the record is schema-valid; if not it is routed to a quarantine store with its reason and original payload and goes no further. A valid line reaches a second decision: is the pay code contributory? A non-contributory code such as a meal penalty or per diem is masked to a pensionable base of 0.00. A contributory code has its base clamped to the remaining ceiling headroom, base equals the minimum of the line amount and the plan-year ceiling minus the amount already accrued. Both base paths converge on one step that multiplies the base by the employer rate and quantizes to whole cents, then appends an immutable entry stamped with a SHA-256 hash to a write-once WORM log. Raw payroll line Pydantic model_validate schema valid? no Quarantine reason + original payload yes contributory pay code? no base = 0.00 non-contributory masked yes base = min(amount, ceiling − accrued) pension = base × rate quantize to whole cents Append immutable entry + SHA-256 to WORM log

Audit Trail Requirements

A pension accrual is only trustworthy if every cent is reconstructable after the fact. Each calculated entry must be persisted with, at minimum: performer_id, plan_year, pay_code, the contributory_base actually applied (which is not the raw line amount whenever the ceiling clamps it), the resulting pension, the ceiling_reached flag, the SHA-256 audit_hash, and a timezone-aware computed_at. The hash is what makes the record idempotent: recomputing the same line must reproduce the same fingerprint, so a disputed week can be replayed and byte-compared rather than argued.

These entries belong in write-once, append-only storage — an object-lock (WORM) bucket or an insert-only ledger table — never an updatable row that can be silently edited between the remittance and the audit. When a SAG-AFTRA examiner questions a single line, the query is performer_id + plan_year + audit_hash, and the answer must be the exact base, rate, and running total that produced the number. Store the agreement_version of the rate schedule alongside each entry as well, so a rate that changed mid-plan-year is always attributable to the correct bargaining cycle. Records that fail — an unknown pay code, a boundary validation error — go to the quarantine store with their reason and original payload, never into the ledger as a resolved zero, because a silent zero is indistinguishable at audit time from a legitimately zero contribution.

Gotchas and Production Edge Cases

Plan-year boundaries and DST. The contribution ceiling resets on a plan-year boundary, so the timezone in which “which year” is decided is not cosmetic. A wrap shot logged at 2026-12-31T23:30 in a UTC-offset export can fall into 2027 if you compare naive timestamps — silently opening a fresh ceiling and under-accruing the prior year. Normalizing to a single IANA production zone before deriving plan_year() closes this. The same discipline defends against the spring/autumn daylight-saving transitions that make offset-based arithmetic ambiguous.

Multi-location shoots. A performer working across jurisdictions in one plan year still has a single SAG-AFTRA ceiling, not one per location. The ledger key is (performer_id, plan_year) deliberately — never keyed on jurisdiction — so out-of-state days accumulate into the same running total instead of resetting the cap.

Overlapping and out-of-order premiums. Payroll corrections arrive late and out of sequence. Because the base is clamped by live remaining headroom at the moment each line is applied, replaying lines in a different order can shift which line absorbs the last dollars under the ceiling — the per-line split differs even though the plan-year total is identical. If per-line attribution must be stable, sort by worked_at before accrual and treat the accrual pass as idempotent on the SHA-256 hash so a re-sent line is recognized and not double-counted.

Non-contributory leakage. The most common defect is an overtime or rest-period premium that is genuinely contributory being lumped with a meal penalty that is not, or vice versa. Keep the contributory flag on the PayCode itself and add a code only through a reviewed change — never infer contributory status from an amount or a free-text memo. Where a governing agreement other than SAG-AFTRA sets the premium (for example, the turnaround penalties in DGA Overtime & Turnaround Rules), resolve that classification upstream so this ledger only ever sees a settled contributory/non-contributory verdict.

Rate availability at close. If the signed rate schedule is not present when the close runs, do not fall back to a hardcoded constant. Route the gap through the tiered Compliance Fallback Chains so the calculation is flagged with the data source it used rather than quietly applying a stale rate — itself an audit finding.

Up: Pension & Health Fund Calculations