Flagging Consecutive Absences in Attendance Data

This guide covers one precise task: given a normalized attendance fact table, find every learner whose recent sessions contain an unbroken run of N or more absences, and emit a flag for it. Unlike a rolling attendance rate, which averages a sudden collapse against prior good weeks, a consecutive-run detector reads the ordered session sequence directly and fires the moment a student stops attending — days before a rate crosses any cohort floor. It is the run-detection companion to the attendance anomaly detection guide and pairs with the rate-based z-score outlier detector; the two catch different failure shapes.

The technique is run-length encoding (RLE). Sort each student’s sessions by time, collapse the state sequence into (state, run_length) tuples, and test the longest absence run against a threshold. The diagram traces one student’s sequence: an early two-session absence that stays below the threshold, and a later four-session run that crosses it and flags.

Prerequisites

Confirm each of these before running the procedure — the script depends on all of them.

Python 3.10+ (the code uses str | None unions, itertools.groupby, and dataclasses).
pandas==2.2.* installed in the virtualenv (used only for I/O and grouping; the run logic is pure standard library).
A normalized attendance export with canonical states applied — one row per (student_token, course_section_id, session_ts, canonical_state), from the attendance state normalization rules.
Student identifiers already tokenized to salted SHA-256 surrogates (or a salt in ATTENDANCE_HASH_SALT to tokenize at read time). Raw ids must never enter the frame or a flag.
A chosen run threshold N (three to five consecutive sessions is typical) and a decision on how LATE and EXCUSED states break or extend a run — this guide treats EXCUSED as a run-breaker, not an absence.

Step-by-step implementation

1. Load and sort per student within a section. Read the events and sort by (student_token, course_section_id, session_ts); run detection is meaningless unless sessions are in strict chronological order, and an out-of-order frame silently fabricates or hides runs.

2. Classify each session as absence, present, or neutral. Map ABSENT and UNEXCUSED to “absence,” PRESENT and LATE to “present,” and treat EXCUSED and UNRECORDED as neutral run-breakers — a documented excused absence should interrupt a run, not extend it, so a student on medical leave is never flagged as disappearing.

3. Collapse each student’s session states into runs. Use itertools.groupby over the ordered classification to produce (label, run_length) tuples — the run-length encoding. This is the core transform, and it is a pure function of the ordered sequence.

4. Take the longest absence run per student-section. From the encoded tuples, keep the maximum run_length whose label is “absence.” A student with several short runs is a different signal from one long run, and only the longest is tested against the threshold.

5. Flag where the longest run meets the threshold. Emit a flag when max_absence_run >= N, carrying the run length and the window of session timestamps it spans so an advisor can see exactly which sessions triggered it.

6. Record provenance, not justifications. The flag stores the token, section, run length, and the first and last session of the run — never the free-text absence reasons, which may reference protected medical or disciplinary information and belong only in the source system.

Complete runnable code block

python

import os
import hashlib
from dataclasses import dataclass, asdict
from datetime import datetime
from itertools import groupby

import pandas as pd

RUN_N = 3                                     # flag runs of this length or longer
ABSENCE = {"ABSENT", "UNEXCUSED"}             # states that build an absence run
PRESENT = {"PRESENT", "LATE"}                 # states that break a run
# EXCUSED / UNRECORDED are neutral: they break a run without counting as absence.
SALT = os.environ.get("ATTENDANCE_HASH_SALT", "rotate-me-from-secrets").encode()


def tokenize(raw_id: str) -> str:
    """Salted SHA-256 — FERPA-safe; used only if ids arrive un-tokenized."""
    return "stu_" + hashlib.sha256(SALT + raw_id.encode()).hexdigest()[:24]


def _classify(state: str) -> str:
    if state in ABSENCE:
        return "absence"
    if state in PRESENT:
        return "present"
    return "neutral"  # EXCUSED / UNRECORDED break the run without extending it


@dataclass(frozen=True)
class AbsenceRunFlag:
    student_token: str
    course_section_id: int
    max_absence_run: int
    run_start: datetime
    run_end: datetime


def longest_absence_run(sessions: pd.DataFrame) -> AbsenceRunFlag | None:
    """RLE over one student's time-ordered sessions; return the longest run."""
    s = sessions.sort_values("session_ts")
    labels = [_classify(x) for x in s["canonical_state"]]
    times = list(s["session_ts"])
    best_len, best_span = 0, (None, None)
    idx = 0
    for label, group in groupby(labels):
        length = sum(1 for _ in group)
        if label == "absence" and length > best_len:
            best_len = length
            best_span = (times[idx], times[idx + length - 1])
        idx += length
    if best_len < RUN_N:
        return None
    return AbsenceRunFlag(
        student_token=str(s["student_token"].iloc[0]),
        course_section_id=int(s["course_section_id"].iloc[0]),
        max_absence_run=best_len,
        run_start=best_span[0],
        run_end=best_span[1],
    )


def flag_consecutive_absences(events: pd.DataFrame) -> pd.DataFrame:
    flags: list[dict] = []
    grouped = events.groupby(["student_token", "course_section_id"], sort=False)
    for _, section in grouped:
        flag = longest_absence_run(section)
        if flag is not None:
            flags.append(asdict(flag))
    return pd.DataFrame(flags, columns=[
        "student_token", "course_section_id",
        "max_absence_run", "run_start", "run_end",
    ])


if __name__ == "__main__":
    events = pd.read_parquet(os.environ["ATTENDANCE_PARQUET"])
    events["session_ts"] = pd.to_datetime(events["session_ts"], utc=True)
    flagged = flag_consecutive_absences(events)
    print(f"{len(flagged)} student-sections with a run of >= {RUN_N} absences")
    print(flagged.to_string(index=False))

Verification and output validation

Confirm the detector is correct before any flag reaches an advisor:

RLE is order-dependent. Shuffle one student’s rows and assert the flag is unchanged only after the internal sort runs; the function sorts by session_ts, so a scrambled input must still produce the same run.
Threshold boundary. Build a student with exactly N consecutive absences and assert they flag; build one with N-1 and assert they do not. Off-by-one here is the classic RLE bug.
Excused breaks a run. Insert an EXCUSED session in the middle of a long absence stretch and assert the run splits and no longer meets the threshold — an excused day is not a disappearance.
Present breaks a run. Same test with PRESENT in the middle; the two sub-runs are each shorter than the combined stretch.
Tokens only. assert flagged["student_token"].str.startswith("stu_").all() (or matches your surrogate format) so no raw identifier is present.
Span is inside the data. Assert run_start <= run_end and both are drawn from the student’s actual session timestamps.

Troubleshooting

A student flags whose absences were not actually consecutive. The frame was not sorted, so groupby ran over an arbitrary order and merged non-adjacent absences. The function sorts internally; if you replaced that with a pre-sort, confirm it sorts by session_ts within each student-section.
A student on medical leave is flagged. Their leave arrived as ABSENT instead of EXCUSED, so it extended the run. Fix the excused-absence coding upstream in attendance state normalization; EXCUSED is a run-breaker here by design.
A run spans a term break or long gap and looks longer than it is. groupby counts adjacency, not calendar distance, so two absences separated by a semester read as consecutive sessions. Filter to the current term, or split runs on a maximum inter-session gap before encoding.
IndexError when building the run span. The idx + length - 1 span index drifted because the label list and timestamp list fell out of alignment. Keep both derived from the same sorted frame in the same order, as the reference code does.
No flags at all on data you expect to flag. N is set higher than any real run, or absences are coded with a state not in the ABSENCE set (for example a vendor-specific UNEXCUSED_TARDY). Print the distinct canonical_state values and confirm they match the sets, then confirm RUN_N.
Duplicate flags for one student. Duplicate session rows inflate a run. Deduplicate on (student_token, course_section_id, session_ts) before encoding — the same dedup the attendance state normalization rules apply at ingestion.

Attendance Anomaly Detection — the parent layer that combines this run detector with z-score and rate-of-change tests.
Detecting Attendance Outliers with a Z-Score in Python — the cohort-relative rate test that catches outliers a run detector misses.
Attendance State Normalization Rules — the canonical states this detector classifies, and where excused-versus-absent fidelity is enforced.
Early-Warning At-Risk Scoring — the downstream model that consumes an absence-run flag as a high-precision behavioral signal.

Part of: Attendance Anomaly Detection