Orchestrating Batch Severity Scoring with Celery

This guide sits under Automated Severity Scoring Models: it shows how to score millions of open claims overnight — or backfill a whole book after a model change — with Celery, using chunked tasks, a fan-out/reduce chord, idempotent retries, and poison-pill isolation so one malformed claim never sinks the run.

Real-time severity scoring answers “what is this one claim worth?” as the FNOL lands. Batch scoring answers a different question entirely: re-score the entire open inventory. You reach for it when a recalibrated model ships, when a new feature column backfills, when a nightly reserve-adequacy job needs a fresh severity tier on every unclosed claim, or when a state adopts a rule that changes the tier boundaries. That inventory is not thousands of claims — for a mid-size carrier it is millions, and the run has to finish inside an operational window before the day shift opens their queues.

A single-process loop that pulls claims and calls model.predict() in sequence fails this on three counts. It is too slow: even at one millisecond per claim, four million claims is over an hour of pure inference, before any I/O. It is not resumable: kill the process at claim two million and you either re-score everything or hand-roll a checkpoint. And it is fragile in the worst way — one claim with a corrupt feature vector raises inside the loop, and every claim after it goes unscored. Batch severity scoring needs horizontal fan-out across workers, exactly-once accounting so a retried chunk does not double-write, isolation so a poison-pill row is quarantined rather than fatal, and a final reduce step that reconciles what actually got scored against what was dispatched. This page builds that with Celery primitives.

This targets Python 3.10+ with Celery 5, a Redis or RabbitMQ broker, and a result backend that supports chords (Redis or a database backend — the RPC backend does not). Structured audit lines go through structlog. Pin the majors so serialization and chord semantics do not drift under you:

Copy
python -m venv .venv && source .venv/bin/activate
pip install "celery[redis]==5.4.*" "structlog==24.*" "redis==5.*"

Before this stage runs, the scoring function itself must already exist as a pure, importable callable — score_claim(features) -> SeverityResult — from the real-time path described in the parent Automated Severity Scoring Models guide. Batch orchestration reuses that exact function; it must never fork a second copy of the scoring logic, or the overnight run and the live path will disagree. The tier boundaries the scorer applies should come from the versioned thresholds produced by Calibrating Severity Thresholds with Historical Loss Data, captured by version into each run so the run is replayable.

The unit of work is a chunk of claim ids, not a single claim: chunking amortizes broker round-trips and keeps the result set small. The task must be idempotent because acks_late means a worker that dies mid-chunk will have the whole chunk redelivered. Key every write on (claim_id, run_id) so a redelivered chunk upserts rather than double-writes, and wrap each row so one bad claim is quarantined, not fatal.

Copy
from __future__ import annotations

from dataclasses import dataclass, asdict
from decimal import Decimal

import structlog
from celery import Celery

app = Celery("severity_batch")
app.conf.update(
    task_acks_late=True,               # redeliver on worker crash
    task_reject_on_worker_lost=True,
    task_serializer="json",
    result_serializer="json",
    worker_prefetch_multiplier=1,      # fair dispatch for long tasks
)
log = structlog.get_logger("severity.batch")


@dataclass(frozen=True, slots=True)
class ChunkResult:
    run_id: str
    scored: int
    quarantined: int
    skipped: int


@app.task(
    bind=True,
    autoretry_for=(TimeoutError, ConnectionError),
    retry_backoff=True,          # exponential: 1s, 2s, 4s, ...
    retry_backoff_max=120,
    retry_jitter=True,
    max_retries=5,
)
def score_chunk(self, claim_ids: list[str], run_id: str) -> dict:
    scored = quarantined = skipped = 0
    for claim_id in claim_ids:
        try:
            features = load_features(claim_id)          # your data access
            if features is None:
                skipped += 1
                continue
            result = score_claim(features)               # reused pure scorer
            upsert_severity(claim_id, run_id, result)    # keyed (claim, run)
            scored += 1
        except PoisonClaim as exc:
            dead_letter(claim_id, run_id, reason=str(exc))
            quarantined += 1
            log.warning("severity.batch.poison", claim_id=claim_id,
                        run_id=run_id, reason=str(exc))
    return asdict(ChunkResult(run_id, scored, quarantined, skipped))

PoisonClaim is raised deliberately by load_features/score_claim for a row whose features cannot be validated — a null coverage limit, a non-Decimal loss amount, an unmappable peril code. It is caught inside the loop so the chunk continues; the transient exceptions in autoretry_for are not caught here, so they bubble up and trigger Celery’s backoff retry of the whole chunk.

A chord is a group of tasks plus a callback that runs once, after every task in the group succeeds, receiving the list of their return values. That is precisely fan-out plus reduce: dispatch one score_chunk per chunk, and let the chord invoke reconcile with all the ChunkResults.

Copy
from celery import chord, group

CHUNK_SIZE = 500


def _chunk(seq: list[str], size: int) -> list[list[str]]:
    return [seq[i:i + size] for i in range(0, len(seq), size)]


def dispatch_batch(claim_ids: list[str], run_id: str, model_version: str) -> str:
    chunks = _chunk(claim_ids, CHUNK_SIZE)
    log.info("severity.batch.dispatch", run_id=run_id, total=len(claim_ids),
             chunks=len(chunks), model_version=model_version)
    header = group(score_chunk.s(c, run_id) for c in chunks)
    callback = reconcile.s(run_id=run_id, dispatched=len(claim_ids),
                           model_version=model_version)
    async_result = chord(header)(callback)
    return async_result.id          # track the chord for run accounting

The callback proves the run is complete and accounted for. Every dispatched claim must land in exactly one bucket — scored, quarantined, or skipped. If the sum does not equal the dispatched total, a chunk was lost and the run manifest must not be signed as clean.

Copy
from celery import shared_task


class BatchImbalance(Exception):
    """Accounted rows do not equal dispatched rows — run is not trustworthy."""


@shared_task
def reconcile(results: list[dict], *, run_id: str, dispatched: int,
              model_version: str) -> dict:
    scored = sum(r["scored"] for r in results)
    quarantined = sum(r["quarantined"] for r in results)
    skipped = sum(r["skipped"] for r in results)
    accounted = scored + quarantined + skipped
    manifest = {
        "run_id": run_id, "model_version": model_version,
        "dispatched": dispatched, "scored": scored,
        "quarantined": quarantined, "skipped": skipped,
        "chunks": len(results),
    }
    if accounted != dispatched:
        log.error("severity.batch.imbalance", **manifest, accounted=accounted)
        raise BatchImbalance(f"{accounted} accounted != {dispatched} dispatched")
    write_run_manifest(manifest)      # append-only, signed
    log.info("severity.batch.complete", **manifest)
    return manifest

You do not need a live broker to test orchestration logic. Set task_always_eager so tasks run synchronously in-process and exceptions propagate to the caller, then assert on the reconciliation math and on poison-pill isolation. Keep a separate integration test that runs against a real broker in CI, but the fast unit suite below is what proves correctness.

Copy
import pytest
from severity_batch import (app, score_chunk, reconcile, BatchImbalance,
                            ChunkResult)


@pytest.fixture(autouse=True)
def eager():
    app.conf.task_always_eager = True
    app.conf.task_eager_propagates = True
    yield
    app.conf.task_always_eager = False


def test_poison_row_isolated_not_fatal(monkeypatch):
    # one of three claims is a poison pill; the other two still score
    result = score_chunk.apply(args=(["ok-1", "bad-2", "ok-3"], "run-A")).get()
    assert result["scored"] == 2
    assert result["quarantined"] == 1
    assert result["skipped"] == 0


def test_reconcile_balances():
    parts = [asdict_result(2, 1, 0), asdict_result(3, 0, 1)]
    manifest = reconcile.apply(
        args=(parts,),
        kwargs=dict(run_id="run-A", dispatched=7, model_version="v3"),
    ).get()
    assert manifest["scored"] == 5 and manifest["quarantined"] == 1


def test_reconcile_fails_closed_on_lost_chunk():
    parts = [asdict_result(2, 0, 0)]         # only 2 accounted
    with pytest.raises(BatchImbalance):
        reconcile.apply(
            args=(parts,),
            kwargs=dict(run_id="run-A", dispatched=7, model_version="v3"),
        ).get()


def asdict_result(scored, quarantined, skipped):
    return {"run_id": "run-A", "scored": scored,
            "quarantined": quarantined, "skipped": skipped}

Run with python -m pytest -q. The load-bearing assertions are that a poison row raises the quarantine count without lowering the scored count, and that reconcile refuses to sign a manifest whose buckets do not sum to the dispatched total.

A batch re-score changes the severity tier — and therefore potentially the routing and reserve — of claims that were already scored under a prior model. That makes the run manifest a regulated artifact: it must record the run_id, the exact model_version and threshold version applied, the as-of snapshot the inventory was drawn from, and the four counts. Write it to the same append-only ledger the real-time path uses, so an examiner can answer “why did this claim’s severity change on this date?” by replaying the signed manifest rather than trusting a mutable database column. Because score_chunk reuses the identical scoring function as the live path and keys every write on (claim_id, run_id), the batch verdict is reproducible and cannot silently diverge from the online decision — the NAIC market-conduct expectation for consistent, explainable claims handling.

• Chord callback never fires. Symptom: the group finishes but reconcile never runs. Cause: a result backend that does not support chord synchronization (the RPC backend). Fix: use the Redis or a database result backend, and confirm result_backend is set before the chord is built.
• Retried chunk double-writes severity rows. Symptom: scored count exceeds dispatched after a worker restart. Cause: writes keyed on claim_id alone, so a redelivered acks_late chunk inserts twice. Fix: make upsert_severity idempotent on (claim_id, run_id) and use an upsert, never a plain insert.
• One bad claim kills the whole chunk. Symptom: hundreds of claims in a chunk go unscored because of a single row. Cause: the row raised an exception that escaped the per-row try. Fix: validate features into an explicit PoisonClaim and catch it inside the loop; let only transient errors in autoretry_for bubble to the retry.
• Retry storm hammers the feature store. Symptom: a transient outage triggers synchronized retries that overload recovery. Cause: fixed-interval retries with no jitter. Fix: keep retry_backoff=True with retry_jitter=True and a sane retry_backoff_max, and cap max_retries so a genuinely dead chunk dead-letters instead of retrying forever.
• Run never finishes inside the window. Symptom: overnight batch spills into business hours. Cause: worker_prefetch_multiplier too high, so a few workers hoard chunks while others idle. Fix: set worker_prefetch_multiplier=1 for long tasks and scale worker concurrency to match the inference cost per chunk.

• Automated Severity Scoring Models — the parent guide whose scoring function this batch run reuses
• Calibrating Severity Thresholds with Historical Loss Data — where the versioned tier boundaries applied in each run come from
• Dynamic Threshold Tuning — why a threshold change is the event that most often triggers a full backfill
• Parallelizing Adjuster Assignment with Asyncio Task Graphs — the complementary concurrency model for the assignment stage
• Claims Triage & Routing Engines — the control plane this batch job feeds