Building an At-Risk Student Flag in Python

This guide builds a working at-risk student flag from three already-normalized inputs: a per-learner engagement frame, a gradebook frame, and an attendance frame. The task is deliberately narrow — join the three into a learner-term feature vector, score it with a weighted logistic model, apply a threshold derived from an alert budget, and emit flags that carry only tokenized identifiers. The modelling rationale behind each step (why logistic, why calibrate, why an alert budget) is covered in the parent Early-Warning At-Risk Scoring reference; here we focus on producing a correct, FERPA-safe flag frame you can hand to an advising team.

The whole procedure fits in one script. What makes it production-worthy rather than a toy is the join discipline (start from the roster so silent-dropout learners survive), the structural-null handling (a missing engagement row means zero, not average), and the tokenization at the boundary (no raw id ever reaches the output).

Prerequisites

Confirm each of these before running — the script assumes all of them are in place.

Step-by-step implementation

1. Start the join from the roster, not from engagement. Left-joining engagement, grades, and attendance onto the roster guarantees that a learner with no activity — the highest-risk case — still appears in the scored set. An inner join would silently drop exactly the students you most need to flag.

2. Convert structural nulls to meaningful values, not medians. A missing engagement row means the learner never engaged, so set engagement_index = 0.0 and login_recency_days to the full term length. Median-imputing here would disguise a silent dropout as an average student.

3. Standardize continuous features within the cohort. Subtract the cohort mean and divide by the standard deviation so a weight learned on one term is meaningful on another, and clip extreme values first so one scripted-download outlier does not dominate the scale.

4. Compute the linear score, then the logistic probability. Multiply each standardized feature by its reviewed weight, sum with the bias, and pass the result through the sigmoid to get a risk probability in [0, 1].

5. Apply calibration before thresholding. A Platt-scaling slope and intercept (fit on a held-out term) map the raw probability to a true frequency, so a 0.30 really means “about 30% of such learners historically struggled.”

6. Derive the threshold from the alert budget. Sort by probability descending and take the threshold at the budget-th learner, so the number of flags matches what the team can act on — this is what keeps the system out of alert-fatigue territory.

7. Emit flags with tokenized ids only. Replace student_id with a salted HMAC-SHA-256 token so the output frame carries a stable join key but no direct identifier, per the FERPA-compliant PII handling rules.

Complete runnable code block

python
import hashlib
import hmac
import logging

import numpy as np
import pandas as pd

logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
logger = logging.getLogger("at_risk_flag")

_SALT = b"rotate-me-from-secrets-manager"

# Protective features (engagement, grade) get negative weights; risk-raising
# features (absences, missing work) get positive weights. Reviewed by faculty.
WEIGHTS: dict[str, float] = {
    "engagement_index": -2.1,
    "login_recency_days": 0.8,
    "running_grade": -3.4,
    "grade_slope": -1.2,
    "missing_submissions": 0.9,
    "absence_rate": 1.6,
    "consecutive_absences": 0.7,
}
BIAS = -0.5
PLATT_A, PLATT_B = 1.0, 0.0        # calibration fit on a held-out term
TERM_LENGTH_DAYS = 112             # used to fill "never logged in"


def tokenize(raw_id: str) -> str:
    """Deterministic, non-reversible token — no raw id reaches the output."""
    digest = hmac.new(_SALT, str(raw_id).encode("utf-8"), hashlib.sha256).hexdigest()
    return f"stu_{digest[:16]}"


def assemble_features(roster: pd.DataFrame, engagement: pd.DataFrame,
                      grades: pd.DataFrame, attendance: pd.DataFrame) -> pd.DataFrame:
    """Left-join every source onto the roster spine so no learner is dropped."""
    df = (roster
          .merge(engagement, on="student_id", how="left")
          .merge(grades, on="student_id", how="left")
          .merge(attendance, on="student_id", how="left"))
    # Structural nulls carry meaning: no activity == zero, not average.
    fills = {
        "engagement_index": 0.0, "login_recency_days": TERM_LENGTH_DAYS,
        "running_grade": 0.0, "grade_slope": 0.0, "missing_submissions": 0.0,
        "absence_rate": 0.0, "consecutive_absences": 0.0,
    }
    return df.fillna(fills)


def _standardize(df: pd.DataFrame, cols: list[str]) -> pd.DataFrame:
    x = df[cols].astype("float64")
    x = x.clip(lower=x.quantile(0.01), upper=x.quantile(0.99), axis=1)
    return (x - x.mean()) / x.std(ddof=0).replace(0.0, 1.0)


def score_and_flag(df: pd.DataFrame, alert_budget: int) -> pd.DataFrame:
    cols = list(WEIGHTS)
    x = _standardize(df, cols)
    z = x.mul(pd.Series(WEIGHTS)).sum(axis=1) + BIAS
    z_cal = PLATT_A * z + PLATT_B
    p = 1.0 / (1.0 + np.exp(-z_cal))                 # logistic probability
    # Threshold = the probability of the budget-th highest-risk learner.
    k = min(alert_budget, len(p))
    tau = p.sort_values(ascending=False).iloc[k - 1] if k else 1.0
    out = pd.DataFrame({
        "learner_token": df["student_id"].map(tokenize),
        "risk_probability": p.round(4),
        "at_risk": p >= tau,
    })
    logger.info("Scored %d learners; flagged %d at tau=%.3f",
                len(out), int(out["at_risk"].sum()), tau)
    return out.sort_values("risk_probability", ascending=False).reset_index(drop=True)


if __name__ == "__main__":
    # Placeholder frames — model ids only, never real student numbers.
    roster = pd.DataFrame({"student_id": ["p1", "p2", "p3", "p4"], "term_id": "2026SP"})
    engagement = pd.DataFrame({"student_id": ["p1", "p2", "p3"],
                               "engagement_index": [0.9, 0.4, 0.1],
                               "login_recency_days": [1, 9, 30]})
    grades = pd.DataFrame({"student_id": ["p1", "p2", "p3", "p4"],
                           "running_grade": [0.88, 0.61, 0.42, 0.0],
                           "grade_slope": [0.02, -0.05, -0.11, 0.0],
                           "missing_submissions": [0, 2, 5, 0]})
    attendance = pd.DataFrame({"student_id": ["p1", "p2", "p3"],
                               "absence_rate": [0.02, 0.15, 0.40],
                               "consecutive_absences": [0, 2, 6]})

    features = assemble_features(roster, engagement, grades, attendance)
    flags = score_and_flag(features, alert_budget=2)
    print(flags.to_string(index=False))

Verification and output validation

Confirm the flag frame is correct and FERPA-safe before handing it to an advising team:

  • No learner dropped. assert len(flags) == len(roster) — the roster-spine join must preserve every enrollment, including the zero-activity learner p4.
  • No raw identifiers. assert "student_id" not in flags.columns and assert flags["learner_token"].str.startswith("stu_").all().
  • Flag count matches the budget. assert flags["at_risk"].sum() <= alert_budget — thresholding on the budget-th learner never exceeds capacity.
  • Silent dropout ranks high. The learner with zero engagement and a zero running grade (p4 here, or p3) should appear near the top of the sorted frame; if it does not, a sign is flipped in WEIGHTS.
  • Probabilities are bounded. assert flags["risk_probability"].between(0, 1).all() — a value outside [0, 1] means the sigmoid was bypassed.

Troubleshooting

  • KeyError on a weight column during scoring. A feature named in WEIGHTS is missing from the assembled frame — usually a source frame was joined on a mismatched key. Confirm every source uses the same student_id column name and that assemble_features ran before score_and_flag.
  • Everyone gets flagged, or no one does. The alert_budget is larger than the cohort or set to zero. It should be a small fraction of enrollment; re-derive it from the support team’s real weekly capacity as described in Early-Warning At-Risk Scoring.
  • The strongest students are flagged as at-risk. A protective feature (running_grade, engagement_index) carries a positive weight, or standardization was skipped so raw scales dominate. Protective features must be negative and _standardize must run first.
  • nan risk probabilities. A structural null slipped past fillna, so a standardized value is nan. Confirm every column in WEIGHTS has an entry in the fills dict inside assemble_features.
  • Flags shift wildly between weekly runs. A fixed threshold plus a drifting cohort, or an un-calibrated model. Keep the budget-derived threshold and re-check the Platt parameters against a held-out term each run.
  • Two rows per learner after the join. A source frame has duplicate student_id rows (more than one row per learner-term). De-duplicate each source to the learner-term grain before assembly.

Part of: Early-Warning At-Risk Scoring