ACA Tracking Logic

ACA Tracking Logic is the stateful compliance engine that turns raw timekeeping into defensible full-time determinations, and it sits inside the Core Architecture & Compliance Mapping for Payroll Systems framework as a downstream consumer of normalized payroll events. The Affordable Care Act Employer Shared Responsibility Provisions under IRC § 4980H require deterministic tracking of Hours of Service (HOS) and full-time status across rolling measurement periods, and a single rounding error, timezone ambiguity, or duplicated punch can flip an employee from part-time to full-time — or push a borderline employer over the 50-FTE Applicable Large Employer (ALE) line — and propagate directly into incorrect Form 1094-C/1095-C filings and § 4980H(a)/(b) penalty exposure. This module ingests heterogeneous time records, normalizes them against statutory HOS definitions, aggregates them over measurement windows, and emits audit-ready status flags with an immutable trail that withstands IRS examination.

Data Normalization & Boundary Enforcement

Raw time and attendance data enters in incompatible shapes: biometric clock-ins, HRIS leave records, payroll batch adjustments, and third-party scheduling exports. Before any determination runs, the ingestion layer must collapse these into a single canonical HoursOfService schema and enforce the Data Boundary Definitions that keep exempt and non-exempt classifications, contractor records, and seasonal worker pools from cross-contaminating each other. Boundary validation happens at the edge — schema checks, timezone alignment to the employer’s primary worksite, and explicit rejection of malformed or duplicate punches — so that the determination engine only ever sees clean, attributable records.

The statutory definition is the field-level constraint that matters most. Under 26 CFR § 54.4980H-1(a)(24), an Hour of Service is each hour for which an employee is paid or entitled to payment for the performance of duties, plus each hour of paid leave for vacation, holiday, illness, incapacity, jury duty, military duty, or leave of absence. Hours for which no compensation is owed — unpaid FMLA leave, unpaid USERRA leave, and unpaid personal leave outside a bona fide plan — are excluded. The normalization schema must therefore carry an explicit HOS category on every record so the engine can include or exclude deterministically rather than inferring intent later.

Quarantine conditions specific to ACA tracking:

• Negative or impossible durations. Any record with negative hours or more than 24 hours in a calendar day is a clock malfunction or a merge artifact; route it to quarantine, never silently clamp it into a determination.
• Missing HOS category. A record without a resolvable category cannot be classified as countable or non-countable and must not default to “counts.” Quarantine it.
• Ambiguous DST intervals. Punches that straddle a daylight-saving transition produce phantom or missing hours when converted naively. Normalize to the worksite zone, log the gap or overlap, and route unresolved intervals to a fallback handler.
• Duplicate punches. Re-uploaded files and retried batches replay records already counted. Deduplicate on a content hash at ingestion so a measurement window is never double-credited.

Hours are stored and summed with Decimal precision rather than binary floating point. While HOS is not currency, the 130-hour monthly boundary is a hard step function: an employee at 129.99 versus 130.00 hours produces a different statutory status, so cumulative IEEE-754 drift across thousands of records is a correctness defect, not a cosmetic one.

from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from decimal import Decimal, ROUND_HALF_UP, InvalidOperation
from enum import Enum
from typing import Optional, Dict, List, Tuple
import hashlib
import logging

logger = logging.getLogger(__name__)

HOS_QUANTUM = Decimal("0.01")          # store hours to 1/100th
MAX_DAILY_HOS = Decimal("24")


class HOSCategory(str, Enum):
    PAID_WORK = "paid_work"
    PAID_LEAVE = "paid_leave"      # vacation/holiday/paid sick — counts under § 54.4980H-1(a)(24)
    UNPAID_LEAVE = "unpaid_leave"  # unpaid FMLA/USERRA/personal — excluded from HOS
    EXCLUDED = "excluded"          # unpaid breaks, bona fide volunteer, contractor hours


class ProcessingStatus(str, Enum):
    ACCEPTED = "accepted"
    NON_HOS = "non_hos"          # valid record that simply does not count toward HOS
    QUARANTINED = "quarantined"  # malformed/ambiguous — needs human review


@dataclass(frozen=True)
class TimeRecord:
    employee_id: str
    record_date: date
    hours: Decimal
    category: HOSCategory
    source_system: str
    worksite_tz: str = "America/New_York"

    @property
    def record_hash(self) -> str:
        payload = f"{self.employee_id}|{self.record_date.isoformat()}|{self.hours}|{self.category.value}"
        return hashlib.sha256(payload.encode()).hexdigest()


@dataclass
class AuditEvent:
    record_id: str
    timestamp: datetime
    status: ProcessingStatus
    reason: Optional[str] = None
    metadata: Dict = field(default_factory=dict)

Jurisdictional Resolution & Effective Dating

The federal § 4980H full-time definition is the floor, not the whole rule. Several jurisdictions impose parallel full-time or coverage definitions that a national employer must track alongside the federal threshold — Massachusetts Fair Share and Health Connector reporting, Hawaii’s Prepaid Health Care Act (which defines full-time at 20 hours per week), California’s SB 1255 wage-statement requirements, and state individual-mandate reporting in California, New Jersey, Rhode Island, and the District of Columbia that drives 1095 distribution deadlines. ACA tracking therefore resolves rules with an explicit override hierarchy — Municipal > State > Federal — selecting the most protective applicable definition for the employee’s primary worksite while still computing the federal determination for ALE counting and IRS filing.

This is the same temporal precision required by FLSA Threshold Mapping, where exempt/non-exempt status turns on an effective-dated salary floor; here the dated artifact is the measurement-period rule set rather than a salary number. Every threshold and every measurement-period definition carries a half-open effective window, and the controlling rule is selected against the period being evaluated — never against today. Inserting datetime.now() into a determination for a prior measurement period is the single most common source of silently wrong ACA history.

@dataclass(frozen=True)
class FullTimeRule:
    jurisdiction: str            # "US", "US-MA", "US-HI", ...
    effective_start: date
    effective_end: Optional[date]  # None == open-ended; window is [start, end)
    monthly_hos_threshold: Decimal
    precedence: int              # 0 federal, 1 state, 2 municipal


def resolve_rule(rules: List[FullTimeRule], worksite: str, as_of: date) -> FullTimeRule:
    """Most-protective applicable rule for a worksite at a measurement date.

    Window test is half-open so adjacent statutes never both match a date.
    """
    candidates = [
        r for r in rules
        if r.jurisdiction in ("US", worksite)
        and r.effective_start <= as_of
        and (r.effective_end is None or as_of < r.effective_end)
    ]
    if not candidates:
        raise LookupError(f"no_rule worksite={worksite} as_of={as_of.isoformat()}")

    # Overlap detection within a single jurisdiction is a config defect, not a tie to break.
    for jur in {c.jurisdiction for c in candidates}:
        same = [c for c in candidates if c.jurisdiction == jur]
        if len(same) > 1:
            raise ValueError(f"overlapping_windows jurisdiction={jur} as_of={as_of.isoformat()}")

    # Highest precedence wins; ties broken toward the lower (more protective) threshold.
    return min(
        candidates,
        key=lambda r: (-r.precedence, r.monthly_hos_threshold),
    )

The federal full-time test compares aggregated monthly hours against a fixed boundary. Expressed as a formula, an employee is full-time for a calendar month when

$\text{HOS}_{\text{month}} \ge 130 \quad\text{or, weekly,}\quad \overline{h}_{\text{week}} \ge 30,$

and ALE status for a year depends on the average monthly full-time-equivalent count, where non-full-time hours roll up as

$\text{FTE}_{\text{month}} = \left\lfloor \frac{\sum \min(\text{HOS}_{\text{emp}},\,120)}{120} \right\rfloor.$

The 130-hour monthly figure — not 4.33 × 30 — is the controlling number under 26 CFR § 54.4980H-3, and non-full-time hours are capped at 120 per employee per month before division.

Production Implementation Pattern

The normalization engine maps each canonical record to a countable or non-countable result, quarantines anomalies, and aggregates HOS deterministically at the employee-month level. Determinism is the contract: identical input payloads must yield identical totals and classifications regardless of execution order or retry count. All arithmetic is Decimal, all logging is structured key=value so the lines are copy-paste safe into production log pipelines, and every record produces exactly one audit event.

class HOSNormalizer:
    """Classify and normalize a single TimeRecord. Pure and idempotent per record."""

    COUNTABLE = {HOSCategory.PAID_WORK, HOSCategory.PAID_LEAVE}

    def __init__(self) -> None:
        self.audit_trail: List[AuditEvent] = []

    def normalize(self, record: TimeRecord) -> Tuple[Optional[Decimal], AuditEvent]:
        now = datetime.now(timezone.utc)
        try:
            hours = record.hours
            if hours.is_nan() or hours < 0 or hours > MAX_DAILY_HOS:
                raise ValueError(f"out_of_range hours={hours}")
        except (InvalidOperation, ValueError) as exc:
            return self._emit(record, now, ProcessingStatus.QUARANTINED, str(exc))

        if record.category not in self.COUNTABLE:
            return self._emit(
                record, now, ProcessingStatus.NON_HOS,
                reason=f"non_hos_category category={record.category.value}",
            )

        normalized = hours.quantize(HOS_QUANTUM, rounding=ROUND_HALF_UP)
        _, event = self._emit(
            record, now, ProcessingStatus.ACCEPTED, normalized=normalized
        )
        return normalized, event

    def _emit(
        self,
        record: TimeRecord,
        ts: datetime,
        status: ProcessingStatus,
        reason: Optional[str] = None,
        normalized: Optional[Decimal] = None,
    ) -> Tuple[Optional[Decimal], AuditEvent]:
        event = AuditEvent(
            record_id=record.record_hash,
            timestamp=ts,
            status=status,
            reason=reason,
            metadata={
                "source": record.source_system,
                "category": record.category.value,
                "normalized": str(normalized) if normalized is not None else None,
            },
        )
        self.audit_trail.append(event)
        if status is ProcessingStatus.QUARANTINED:
            logger.warning("record_quarantined record=%s reason=%s", record.record_hash, reason)
        return (normalized if status is ProcessingStatus.ACCEPTED else None), event


@dataclass(frozen=True)
class MeasurementPeriod:
    employee_id: str
    start_date: date
    end_date: date          # inclusive calendar bound of the lookback window
    is_lookback: bool = True


class FTEAggregator:
    """Aggregate normalized HOS to a monthly full-time determination."""

    def __init__(self, normalizer: HOSNormalizer) -> None:
        self.normalizer = normalizer

    def determine(
        self,
        records: List[TimeRecord],
        period: MeasurementPeriod,
        rules: List[FullTimeRule],
        worksite: str,
    ) -> Dict[str, str]:
        rule = resolve_rule(rules, worksite, period.start_date)
        seen: set[str] = set()
        totals: Dict[str, Decimal] = {}

        for rec in records:
            if not (period.start_date <= rec.record_date <= period.end_date):
                continue
            if rec.record_hash in seen:          # idempotent dedupe
                continue
            seen.add(rec.record_hash)
            normalized, _ = self.normalizer.normalize(rec)
            if normalized is None:               # non-HOS or quarantined: do not count
                continue
            totals[rec.employee_id] = totals.get(rec.employee_id, Decimal("0")) + normalized

        statuses: Dict[str, str] = {}
        near = rule.monthly_hos_threshold * Decimal("0.8")
        for emp_id, total in totals.items():
            if total >= rule.monthly_hos_threshold:
                status = "FULL_TIME"
            elif total >= near:
                status = "NEAR_THRESHOLD"
            else:
                status = "PART_TIME"
            statuses[emp_id] = status
            logger.info(
                "fte_determined emp=%s hos=%s threshold=%s status=%s jurisdiction=%s",
                emp_id, total, rule.monthly_hos_threshold, status, rule.jurisdiction,
            )
        return statuses

Records that fail validation route to a quarantine queue rather than aborting the batch; the decision logic for what is recoverable versus blocking is shared with Fallback Routing Strategies, so ACA tracking reuses the same dead-letter and review-queue plumbing rather than inventing its own.

Compliance Verification & Fallback Routing

Before any determination feeds 1094-C/1095-C generation, the pipeline runs a fixed verification checklist. Each item is a hard gate: a failure freezes downstream filing rather than emitting a defensible-looking but wrong status.

1. HOS reconciliation. The sum of daily normalized hours for an employee-month must equal the aggregator’s monthly total. A delta greater than Decimal("0.01") indicates a dropped or double-counted record and is a hard stop.
2. Decimal precision check. Assert no value in the path is a float. A reconciliation that passes by a hair under float can fail under Decimal; ban floats in the determination modules with a lint rule and assert the type at the boundary.
3. Threshold boundary tests. Run fixtures at 129.99, 130.00, and 130.01 monthly hours against the resolved rule and assert PART_TIME / NEAR_THRESHOLD, FULL_TIME, FULL_TIME. Crossing the boundary must produce a recorded stability-period assignment.
4. Effective-date drift test. Re-resolve the controlling rule for a prior measurement period and assert it returns the rule that was in force then, not the current one. This catches datetime.now() leaks.
5. Fallback activation test. Inject a malformed and a duplicate record and assert both are quarantined/deduplicated, the batch still completes, and the quarantine counter increments.
6. Audit immutability. Serialize every AuditEvent to a write-once ledger or object store with SHA-256 checksums; the manifest hash must be reproducible from the sorted event stream.

class ComplianceVerifier:
    RECON_TOLERANCE = Decimal("0.01")
    QUARANTINE_PAUSE_RATE = Decimal("0.05")  # >5% quarantined -> EMERGENCY_PAUSE

    def verify(
        self,
        statuses: Dict[str, str],
        totals: Dict[str, Decimal],
        rule: FullTimeRule,
    ) -> bool:
        for emp_id, status in statuses.items():
            hos = totals.get(emp_id, Decimal("0"))
            ft = hos >= rule.monthly_hos_threshold
            if (status == "FULL_TIME") != ft and status != "NEAR_THRESHOLD":
                logger.error(
                    "compliance_violation emp=%s hos=%s status=%s threshold=%s",
                    emp_id, hos, status, rule.monthly_hos_threshold,
                )
                return False
        return True

    def audit_manifest(self, events: List[AuditEvent]) -> str:
        lines = [
            f"{e.timestamp.isoformat()}|{e.record_id}|{e.status.value}|{e.reason or 'OK'}"
            for e in sorted(events, key=lambda x: (x.timestamp, x.record_id))
        ]
        return hashlib.sha256("\n".join(lines).encode()).hexdigest()

    def should_pause(self, events: List[AuditEvent]) -> bool:
        if not events:
            return False
        quarantined = sum(1 for e in events if e.status is ProcessingStatus.QUARANTINED)
        rate = Decimal(quarantined) / Decimal(len(events))
        if rate >= self.QUARANTINE_PAUSE_RATE:
            logger.critical("emergency_pause quarantine_rate=%s", rate)
            return True
        return False

If more than five percent of a run’s records quarantine, the orchestrator trips an EMERGENCY_PAUSE and halts until a human clears the backlog — a degraded upstream feed must never silently shrink everyone’s hours. The detailed rolling-window mechanics, administrative-period handling, and stability-period locks live in Automating ACA full-time equivalent tracking.

Failure Modes & Gotchas

• Float-stored hours. Storing HOS as float lets binary rounding accumulate so an employee genuinely at 130.00 hours reconciles to 129.999…, landing on the wrong side of the step function. Fix: Decimal end-to-end, quantized once, with a reconciliation assertion to 0.01.
• Counting unpaid leave as HOS. Treating unpaid FMLA or unpaid personal leave as countable inflates totals and creates false full-time determinations and over-reporting on 1095-C. Fix: require an explicit HOSCategory per record and count only PAID_WORK and PAID_LEAVE under § 54.4980H-1(a)(24); never default a missing category to “counts.”
• datetime.now() in a prior-period determination. Resolving the threshold or measurement-period rule against the current date silently re-classifies closed history when a statute changes. Fix: effective-dated rule resolution keyed on the measurement period, with overlapping windows rejected at load time.
• DST and timezone double/zero counting. Aggregating in UTC without worksite alignment turns a spring-forward shift into 23 logged hours and a fall-back shift into 25, skewing the monthly total. Fix: normalize to the worksite zone, log DST gaps/overlaps, and quarantine unresolved intervals.
• Replayed batches. A re-uploaded timecard file or a retried job re-credits hours already counted, pushing borderline employees over 130. Fix: idempotent dedupe on record_hash inside the aggregation window, exactly as the ingestion layer dedupes on ingest.

Frequently Asked Questions

Related

• Automating ACA full-time equivalent tracking — rolling lookback, administrative, and stability period mechanics with 1094-C/1095-C generation.
• FLSA Threshold Mapping — effective-dated exempt/non-exempt resolution that classifies the workers ACA tracking measures.
• Data Boundary Definitions — the canonical-record contract and Decimal-precision boundaries enforced at ingestion.
• Fallback Routing Strategies — quarantine, dead-letter, and review-queue routing reused by the determination engine.
• Core Architecture & Compliance Mapping for Payroll Systems — the parent architecture this determination engine plugs into.

External Compliance References

• IRS Employer Shared Responsibility Provisions (IRC § 4980H)
• IRS ACA Information Returns (Forms 1094-C/1095-C)
• 26 CFR § 54.4980H-1 — Definitions (Hours of Service)
• 26 CFR § 54.4980H-3 — Determining full-time employees