from __future__ import annotations
import argparse
import hashlib
import json
import webbrowser
from dataclasses import dataclass
from pathlib import Path
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
FACULTY_PROGRAMMES = {
"Medicine": ["Medicine", "Nursing", "Public Health"],
"Engineering": ["Computer Science", "Mechanical Engineering", "Cybernetics"],
"Humanities": ["History", "Linguistics", "Philosophy"],
"Social Sciences": ["Economics", "Political Science", "Psychology"],
"Business": ["Accounting", "Finance", "Management"],
"Education": ["Primary Education", "Special Needs Education", "Pedagogy"],
"Law": ["Law", "International Law", "Public Administration"],
"Natural Sciences": ["Biology", "Chemistry", "Mathematics"],
}
TERMS = {
"2025A": {
"start": pd.Timestamp("2025-08-18"),
"end": pd.Timestamp("2025-12-07"),
"label_observed_after": pd.Timestamp("2026-01-15"),
},
"2026S": {
"start": pd.Timestamp("2026-01-12"),
"end": pd.Timestamp("2026-05-03"),
"label_observed_after": pd.Timestamp("2026-06-15"),
},
}
MODEL_VERSION = "student-risk-demo-0.1"
@dataclass
class SourceData:
sis_students: pd.DataFrame
identity_map: pd.DataFrame
status_history: pd.DataFrame
enrollment_terms: pd.DataFrame
lms_activity_daily: pd.DataFrame
erp_finance_snapshots: pd.DataFrame
campus_activity_daily: pd.DataFrame
def sigmoid(values: np.ndarray) -> np.ndarray:
return 1.0 / (1.0 + np.exp(-np.clip(values, -50, 50)))
def term_dates(term_id: str) -> pd.DatetimeIndex:
term = TERMS[term_id]
return pd.date_range(term["start"], term["end"], freq="D")
def choose_programmes(faculties: np.ndarray, rng: np.random.Generator) -> list[str]:
programmes = []
for faculty in faculties:
programmes.append(rng.choice(FACULTY_PROGRAMMES[faculty]))
return programmes
def generate_sources(n_students: int, seed: int) -> SourceData:
rng = np.random.default_rng(seed)
faculties = np.array(list(FACULTY_PROGRAMMES.keys()))
canonical_ids = np.array([f"STU-{100000 + i}" for i in range(n_students)])
faculty = rng.choice(faculties, size=n_students, p=[0.11, 0.15, 0.1, 0.15, 0.13, 0.12, 0.09, 0.15])
programme = np.array(choose_programmes(faculty, rng))
advisor_id = np.array([f"ADV-{rng.integers(100, 175)}" for _ in range(n_students)])
study_mode = rng.choice(["campus", "hybrid", "online-heavy"], size=n_students, p=[0.5, 0.35, 0.15])
gender = rng.choice(["female", "male", "non-binary_or_not_shared"], size=n_students, p=[0.53, 0.45, 0.02])
age_band = rng.choice(["under_23", "23_29", "30_plus"], size=n_students, p=[0.62, 0.27, 0.11])
international = rng.random(n_students) < 0.17
first_generation = rng.random(n_students) < 0.24
commute_minutes = np.clip(rng.normal(28, 18, n_students), 2, 120).round().astype(int)
ability = rng.normal(0, 1, n_students)
financial_stress = rng.random(n_students) < (0.18 + 0.08 * first_generation + 0.04 * international)
base_gpa = np.clip(3.05 + 0.38 * ability - 0.16 * financial_stress + rng.normal(0, 0.28, n_students), 1.0, 4.0)
sis_students = pd.DataFrame(
{
"canonical_student_id": canonical_ids,
"sis_student_id": [f"FS-{200000 + i}" for i in range(n_students)],
"birth_year": 2026 - rng.choice([20, 21, 22, 23, 24, 27, 32], size=n_students, p=[0.18, 0.2, 0.19, 0.15, 0.1, 0.11, 0.07]),
"gender": gender,
"age_band": age_band,
"international_student": international,
"first_generation_student": first_generation,
"home_commute_minutes": commute_minutes,
"preferred_study_mode": study_mode,
"faculty": faculty,
"programme": programme,
"advisor_id": advisor_id,
"privacy_notice_version": "2026-01",
}
)
lms_ids = [f"CANVAS-{300000 + i}" for i in range(n_students)]
erp_ids = [f"UNIT4-{400000 + i}" for i in range(n_students)]
badge_ids = [f"BADGE-{500000 + i}" for i in range(n_students)]
source_rows = []
for _, row in sis_students.iterrows():
source_rows.extend(
[
(row.canonical_student_id, "SIS", row.sis_student_id, "2024-08-01", None, 1.0, "system_of_record"),
(row.canonical_student_id, "LMS", lms_ids[row.name], "2024-08-01", None, 0.995, "verified_sso_match"),
(row.canonical_student_id, "ERP", erp_ids[row.name], "2024-08-01", None, 0.99, "national_id_hash_match"),
(row.canonical_student_id, "CAMPUS", badge_ids[row.name], "2024-08-01", None, 0.985, "card_registry_match"),
]
)
identity_map = pd.DataFrame(
source_rows,
columns=[
"canonical_student_id",
"source_system",
"source_person_id",
"valid_from",
"valid_to",
"match_confidence",
"match_method",
],
)
identity_map["valid_from"] = pd.to_datetime(identity_map["valid_from"])
identity_map["valid_to"] = pd.to_datetime(identity_map["valid_to"])
enrollment_rows = []
status_rows = []
lms_rows = []
erp_rows = []
campus_rows = []
for i, student in sis_students.iterrows():
current_programme = student.programme
switched_programme = rng.random() < 0.045
programme_2026 = current_programme
if switched_programme:
choices = [p for p in FACULTY_PROGRAMMES[student.faculty] if p != current_programme]
programme_2026 = rng.choice(choices) if choices else current_programme
status_rows.append(
{
"sis_student_id": student.sis_student_id,
"status": "active",
"faculty": student.faculty,
"programme": current_programme,
"valid_from": pd.Timestamp("2024-08-01"),
"valid_to": pd.Timestamp("2026-01-12") if switched_programme else pd.NaT,
"change_recorded_at": pd.Timestamp("2024-08-01"),
"source_sequence": 1,
}
)
if switched_programme:
status_rows.append(
{
"sis_student_id": student.sis_student_id,
"status": "active",
"faculty": student.faculty,
"programme": programme_2026,
"valid_from": pd.Timestamp("2026-01-12"),
"valid_to": pd.NaT,
"change_recorded_at": pd.Timestamp("2025-12-18"),
"source_sequence": 2,
}
)
continues_to_next_term = True
for term_id in TERMS:
if term_id == "2026S" and not continues_to_next_term:
continue
term = TERMS[term_id]
term_index = 0 if term_id == "2025A" else 1
prior_gpa = np.clip(base_gpa[i] + rng.normal(0, 0.12) - 0.07 * term_index, 1.0, 4.0)
credits_registered = int(rng.choice([20, 25, 30, 35], p=[0.06, 0.12, 0.76, 0.06]))
academic_standing = "normal"
if prior_gpa < 2.15:
academic_standing = "probation"
elif prior_gpa < 2.45:
academic_standing = "watch"
latent_disengagement = rng.normal(0, 1) + 0.55 * (prior_gpa < 2.3) + 0.45 * financial_stress[i]
dropout_probability = sigmoid(
np.array(
[
-3.25
+ 0.9 * max(0, 2.6 - prior_gpa)
+ 0.75 * financial_stress[i]
+ 0.85 * latent_disengagement
+ 0.35 * (academic_standing == "probation")
]
)
)[0]
# Approved leave is separate from dropout and should not be treated as the same target.
approved_leave_probability = 0.025 + 0.015 * (student.age_band == "30_plus")
approved_leave_next_term = rng.random() < approved_leave_probability
dropout_next_term = (rng.random() < dropout_probability) and not approved_leave_next_term
enrollment_rows.append(
{
"sis_student_id": student.sis_student_id,
"canonical_student_id": student.canonical_student_id,
"term_id": term_id,
"term_start_date": term["start"],
"faculty": student.faculty,
"programme": current_programme if term_id == "2025A" else programme_2026,
"credits_registered": credits_registered,
"credits_completed_previous_term": int(np.clip(rng.normal(23 + 5 * (prior_gpa - 2.5), 6), 0, 35)),
"prior_gpa": round(float(prior_gpa), 2),
"academic_standing": academic_standing,
"approved_leave_next_term": bool(approved_leave_next_term),
"dropout_label_next_term": bool(dropout_next_term),
"label_observed_after": term["label_observed_after"],
"available_at": term["start"] - pd.Timedelta(days=14),
"financial_stress_seed": bool(financial_stress[i]),
"latent_disengagement_seed": float(latent_disengagement),
}
)
if dropout_next_term:
status_rows.append(
{
"sis_student_id": student.sis_student_id,
"status": "withdrawn",
"faculty": student.faculty,
"programme": current_programme if term_id == "2025A" else programme_2026,
"valid_from": term["end"] + pd.Timedelta(days=18),
"valid_to": pd.NaT,
"change_recorded_at": term["label_observed_after"],
"source_sequence": 99,
}
)
if term_id == "2025A":
continues_to_next_term = False
generate_term_events(
rng=rng,
student=student,
term_id=term_id,
prior_gpa=prior_gpa,
financial_stress=bool(financial_stress[i]),
latent_disengagement=float(latent_disengagement),
dropout_next_term=bool(dropout_next_term),
lms_user_id=lms_ids[i],
erp_account_id=erp_ids[i],
badge_id=badge_ids[i],
lms_rows=lms_rows,
erp_rows=erp_rows,
campus_rows=campus_rows,
)
return SourceData(
sis_students=sis_students,
identity_map=identity_map,
status_history=pd.DataFrame(status_rows),
enrollment_terms=pd.DataFrame(enrollment_rows),
lms_activity_daily=pd.DataFrame(lms_rows),
erp_finance_snapshots=pd.DataFrame(erp_rows),
campus_activity_daily=pd.DataFrame(campus_rows),
)
def generate_term_events(
rng: np.random.Generator,
student: pd.Series,
term_id: str,
prior_gpa: float,
financial_stress: bool,
latent_disengagement: float,
dropout_next_term: bool,
lms_user_id: str,
erp_account_id: str,
badge_id: str,
lms_rows: list[dict],
erp_rows: list[dict],
campus_rows: list[dict],
) -> None:
term = TERMS[term_id]
dates = term_dates(term_id)
mode = student.preferred_study_mode
lms_base = {"campus": 1.2, "hybrid": 1.55, "online-heavy": 2.05}[mode]
campus_base = {"campus": 3.1, "hybrid": 2.1, "online-heavy": 0.65}[mode]
if student.home_commute_minutes > 55:
campus_base *= 0.76
if bool(student.international_student):
lms_base *= 1.05
ability_boost = np.clip((prior_gpa - 2.3) / 1.7, 0, 0.7)
for day in dates:
week = int((day - term["start"]).days // 7) + 1
is_weekend = day.weekday() >= 5
decline = 1.0
if dropout_next_term and week >= 4:
decline = max(0.16, 1.0 - 0.105 * (week - 3) - 0.05 * max(0, latent_disengagement))
noise = rng.lognormal(mean=0, sigma=0.18)
weekday_factor = 0.42 if is_weekend else 1.0
login_lambda = max(0.03, lms_base * decline * weekday_factor * noise + 0.1 * ability_boost)
login_count = int(rng.poisson(login_lambda))
course_views = int(rng.poisson(max(0.1, login_count * rng.uniform(2.0, 5.5))))
video_minutes = int(rng.gamma(shape=1.5, scale=18) * min(1.8, login_count / 2.2)) if login_count else 0
forum_posts = int(rng.poisson(0.05 + 0.12 * login_count + (0.04 if mode == "online-heavy" else 0)))
quiz_attempts = int(rng.poisson(0.1 + 0.18 * login_count)) if week <= 12 else 0
assignment_due = int(day.weekday() == 2 and 2 <= week <= 12)
submission_probability = 0.88 + 0.06 * ability_boost - 0.22 * dropout_next_term * (week >= 4) - 0.08 * financial_stress
assignment_submitted = int(assignment_due and rng.random() < np.clip(submission_probability, 0.1, 0.98))
assignment_late = int(assignment_submitted and rng.random() < (0.09 + 0.18 * dropout_next_term * (week >= 4)))
available_delay_hours = int(rng.integers(2, 38))
lms_rows.append(
{
"lms_user_id": lms_user_id,
"term_id": term_id,
"activity_date": day,
"login_count": login_count,
"course_views": course_views,
"assignment_due_count": assignment_due,
"assignment_submitted_count": assignment_submitted,
"late_submission_count": assignment_late,
"forum_post_count": forum_posts,
"video_minutes": video_minutes,
"quiz_attempt_count": quiz_attempts,
"available_at": day + pd.Timedelta(hours=available_delay_hours),
}
)
campus_decline = decline if dropout_next_term and week >= 4 else 1.0
campus_lambda = max(0.01, campus_base * campus_decline * weekday_factor * rng.lognormal(0, 0.2))
building_entries = int(rng.poisson(campus_lambda))
wifi_minutes = int(rng.gamma(2.0, 60) * min(1.0, building_entries / 3.0)) if building_entries else 0
library_entries = int(rng.poisson(0.14 * building_entries + 0.03 * (week <= 12)))
campus_rows.append(
{
"badge_id": badge_id,
"term_id": term_id,
"event_date": day,
"building_entry_count": building_entries,
"library_entry_count": library_entries,
"wifi_minutes": wifi_minutes,
"campus_zone_count": int(min(building_entries, rng.integers(1, 6))) if building_entries else 0,
"available_at": day + pd.Timedelta(hours=int(rng.integers(4, 48))),
}
)
for snap_date in pd.date_range(term["start"] - pd.Timedelta(days=7), term["end"], freq="7D"):
week = max(0, int((snap_date - term["start"]).days // 7) + 1)
overdue_days = 0
balance = 0.0
tuition_paid = True
if financial_stress:
tuition_paid = week >= 9 and rng.random() < 0.35
if not tuition_paid and week >= 3:
overdue_days = int((week - 2) * 7 + rng.integers(0, 5))
balance = float(np.clip(rng.normal(8500, 2500), 1600, 18000))
scholarship_amount = float(rng.choice([0, 5000, 10000, 15000], p=[0.72, 0.12, 0.1, 0.06]))
erp_rows.append(
{
"erp_account_id": erp_account_id,
"term_id": term_id,
"status_date": snap_date,
"tuition_paid": bool(tuition_paid),
"outstanding_balance_nok": round(balance, 2),
"payment_overdue_days": overdue_days,
"scholarship_amount_nok": scholarship_amount,
"payment_plan": bool(financial_stress and rng.random() < 0.45),
"aid_status": rng.choice(["none", "applied", "approved", "delayed"], p=[0.5, 0.16, 0.27, 0.07]),
"available_at": snap_date + pd.Timedelta(hours=int(rng.integers(8, 72))),
}
)
def map_identity(df: pd.DataFrame, identity_map: pd.DataFrame, source_system: str, id_col: str) -> pd.DataFrame:
mapping = identity_map.loc[identity_map["source_system"].eq(source_system), ["canonical_student_id", "source_person_id"]]
return df.merge(mapping, left_on=id_col, right_on="source_person_id", how="left").drop(columns=["source_person_id"])
def build_feature_snapshot(sources: SourceData, term_id: str, as_of_date: pd.Timestamp) -> pd.DataFrame:
term_start = TERMS[term_id]["start"]
students = sources.sis_students.copy()
enroll = sources.enrollment_terms.loc[sources.enrollment_terms["term_id"].eq(term_id)].copy()
enroll = enroll.loc[enroll["available_at"].le(as_of_date)]
status = sources.status_history.copy()
status["valid_to_filled"] = status["valid_to"].fillna(pd.Timestamp("2100-01-01"))
status_at = status.loc[
status["change_recorded_at"].le(as_of_date)
& status["valid_from"].le(as_of_date)
& status["valid_to_filled"].gt(as_of_date)
].sort_values(["sis_student_id", "source_sequence"])
status_at = status_at.drop_duplicates("sis_student_id", keep="last")
status_at = status_at[["sis_student_id", "status", "faculty", "programme"]].rename(
columns={"status": "student_status_asof", "faculty": "faculty_asof", "programme": "programme_asof"}
)
lms = map_identity(sources.lms_activity_daily, sources.identity_map, "LMS", "lms_user_id")
lms = lms.loc[lms["term_id"].eq(term_id) & lms["activity_date"].le(as_of_date) & lms["available_at"].le(as_of_date)]
lms_14 = lms.loc[lms["activity_date"].gt(as_of_date - pd.Timedelta(days=14))]
lms_to_date = lms.copy()
last_login = (
lms.loc[lms["login_count"].gt(0)]
.groupby("canonical_student_id")["activity_date"]
.max()
.rename("last_lms_login_date")
.reset_index()
)
lms_features = lms_14.groupby("canonical_student_id").agg(
lms_logins_14d=("login_count", "sum"),
lms_course_views_14d=("course_views", "sum"),
video_minutes_14d=("video_minutes", "sum"),
forum_posts_14d=("forum_post_count", "sum"),
quiz_attempts_14d=("quiz_attempt_count", "sum"),
)
assessment_features = lms_to_date.groupby("canonical_student_id").agg(
assignments_due_to_date=("assignment_due_count", "sum"),
assignments_submitted_to_date=("assignment_submitted_count", "sum"),
late_submissions_to_date=("late_submission_count", "sum"),
)
lms_features = lms_features.join(assessment_features, how="outer").reset_index()
lms_features = lms_features.merge(last_login, on="canonical_student_id", how="left")
campus = map_identity(sources.campus_activity_daily, sources.identity_map, "CAMPUS", "badge_id")
campus = campus.loc[campus["term_id"].eq(term_id) & campus["event_date"].le(as_of_date) & campus["available_at"].le(as_of_date)]
campus["active_campus_day"] = (campus["building_entry_count"].gt(0) | campus["wifi_minutes"].gt(0)).astype(int)
campus_14 = campus.loc[campus["event_date"].gt(as_of_date - pd.Timedelta(days=14))]
campus_30 = campus.loc[campus["event_date"].gt(as_of_date - pd.Timedelta(days=30))]
campus_start = campus.loc[campus["event_date"].between(term_start, term_start + pd.Timedelta(days=13))]
campus_features = campus_14.groupby("canonical_student_id").agg(
campus_days_14d=("active_campus_day", "sum"),
building_entries_14d=("building_entry_count", "sum"),
campus_wifi_minutes_14d=("wifi_minutes", "sum"),
)
campus_features = campus_features.join(
campus_30.groupby("canonical_student_id").agg(library_entries_30d=("library_entry_count", "sum")),
how="outer",
)
campus_features = campus_features.join(
campus_start.groupby("canonical_student_id").agg(campus_days_first_14d=("active_campus_day", "sum")),
how="outer",
).reset_index()
erp = map_identity(sources.erp_finance_snapshots, sources.identity_map, "ERP", "erp_account_id")
erp = erp.loc[erp["term_id"].eq(term_id) & erp["status_date"].le(as_of_date) & erp["available_at"].le(as_of_date)]
erp_latest = erp.sort_values(["canonical_student_id", "status_date"]).drop_duplicates("canonical_student_id", keep="last")
erp_features = erp_latest[
[
"canonical_student_id",
"tuition_paid",
"outstanding_balance_nok",
"payment_overdue_days",
"scholarship_amount_nok",
"payment_plan",
"aid_status",
]
]
features = (
enroll.merge(students, on="canonical_student_id", how="left", suffixes=("", "_student"))
.merge(status_at, on="sis_student_id", how="left")
.merge(lms_features, on="canonical_student_id", how="left")
.merge(campus_features, on="canonical_student_id", how="left")
.merge(erp_features, on="canonical_student_id", how="left")
)
fill_zero = [
"lms_logins_14d",
"lms_course_views_14d",
"video_minutes_14d",
"forum_posts_14d",
"quiz_attempts_14d",
"assignments_due_to_date",
"assignments_submitted_to_date",
"late_submissions_to_date",
"campus_days_14d",
"building_entries_14d",
"campus_wifi_minutes_14d",
"library_entries_30d",
"campus_days_first_14d",
"outstanding_balance_nok",
"payment_overdue_days",
"scholarship_amount_nok",
]
for col in fill_zero:
if col in features:
features[col] = features[col].fillna(0)
features["as_of_date"] = as_of_date
features["term_week"] = int((as_of_date - term_start).days // 7 + 1)
features["student_status_asof"] = features["student_status_asof"].fillna("unknown")
features["faculty_asof"] = features["faculty_asof"].fillna(features["faculty"])
features["programme_asof"] = features["programme_asof"].fillna(features["programme"])
features["tuition_paid"] = features["tuition_paid"].fillna(False).astype(bool)
features["payment_plan"] = features["payment_plan"].fillna(False).astype(bool)
features["aid_status"] = features["aid_status"].fillna("unknown")
features["last_lms_login_date"] = pd.to_datetime(features["last_lms_login_date"])
features["days_since_last_lms_login"] = (as_of_date - features["last_lms_login_date"]).dt.days
features["days_since_last_lms_login"] = features["days_since_last_lms_login"].fillna(999)
features["missing_assignments_to_date"] = (
features["assignments_due_to_date"] - features["assignments_submitted_to_date"]
).clip(lower=0)
features["submission_completion_rate_to_date"] = np.where(
features["assignments_due_to_date"].gt(0),
features["assignments_submitted_to_date"] / features["assignments_due_to_date"],
1.0,
)
features["campus_drop_14d_vs_start"] = features["campus_days_first_14d"] - features["campus_days_14d"]
features["outstanding_balance_log1p"] = np.log1p(features["outstanding_balance_nok"])
features["financial_overdue_flag"] = features["payment_overdue_days"].gt(0).astype(int)
return features
class NumpyLogisticRegression:
def __init__(self, learning_rate: float = 0.16, epochs: int = 2200, l2: float = 0.02):
self.learning_rate = learning_rate
self.epochs = epochs
self.l2 = l2
self.weights: np.ndarray | None = None
def fit(self, x: np.ndarray, y: np.ndarray) -> "NumpyLogisticRegression":
x_bias = np.c_[np.ones(x.shape[0]), x]
weights = np.zeros(x_bias.shape[1])
for _ in range(self.epochs):
preds = sigmoid(x_bias @ weights)
gradient = (x_bias.T @ (preds - y)) / y.size
gradient[1:] += self.l2 * weights[1:] / y.size
weights -= self.learning_rate * gradient
self.weights = weights
return self
def predict_proba(self, x: np.ndarray) -> np.ndarray:
if self.weights is None:
raise RuntimeError("Model has not been fit.")
x_bias = np.c_[np.ones(x.shape[0]), x]
return sigmoid(x_bias @ self.weights)
def prepare_design_matrix(
train: pd.DataFrame,
score: pd.DataFrame,
numeric_cols: list[str],
categorical_cols: list[str],
) -> tuple[np.ndarray, np.ndarray, pd.DataFrame, pd.Series, pd.Series]:
train_design = pd.get_dummies(train[numeric_cols + categorical_cols], columns=categorical_cols, dummy_na=True)
score_design = pd.get_dummies(score[numeric_cols + categorical_cols], columns=categorical_cols, dummy_na=True)
train_design, score_design = train_design.align(score_design, join="outer", axis=1, fill_value=0)
train_design = train_design.astype(float)
score_design = score_design.astype(float)
medians = train_design.median()
train_design = train_design.fillna(medians)
score_design = score_design.fillna(medians)
means = train_design.mean()
stds = train_design.std().replace(0, 1.0)
return (
((train_design - means) / stds).to_numpy(),
((score_design - means) / stds).to_numpy(),
train_design,
means,
stds,
)
def transform_with_design(
df: pd.DataFrame,
columns: list[str],
means: pd.Series,
stds: pd.Series,
numeric_cols: list[str],
categorical_cols: list[str],
) -> np.ndarray:
design = pd.get_dummies(df[numeric_cols + categorical_cols], columns=categorical_cols, dummy_na=True)
design = design.reindex(columns=columns, fill_value=0).astype(float)
design = design.fillna(means)
return ((design - means) / stds).to_numpy()
def roc_auc_score(y_true: np.ndarray, y_score: np.ndarray) -> float:
y_true = y_true.astype(int)
n_pos = int(y_true.sum())
n_neg = int(y_true.size - n_pos)
if n_pos == 0 or n_neg == 0:
return float("nan")
order = np.argsort(y_score)
ranks = np.empty_like(order, dtype=float)
ranks[order] = np.arange(1, y_score.size + 1)
pos_rank_sum = ranks[y_true == 1].sum()
return float((pos_rank_sum - n_pos * (n_pos + 1) / 2) / (n_pos * n_neg))
def classification_metrics(y_true: np.ndarray, y_score: np.ndarray, top_fraction: float) -> dict:
threshold = float(np.quantile(y_score, 1 - top_fraction))
flagged = y_score >= threshold
positives = y_true.astype(bool)
tp = int((flagged & positives).sum())
fp = int((flagged & ~positives).sum())
fn = int((~flagged & positives).sum())
tn = int((~flagged & ~positives).sum())
precision = tp / max(1, tp + fp)
recall = tp / max(1, tp + fn)
fpr = fp / max(1, fp + tn)
return {
"roc_auc": roc_auc_score(y_true, y_score),
"threshold_for_top_fraction": threshold,
"flagged_fraction": float(flagged.mean()),
"precision_at_capacity": precision,
"recall_at_capacity": recall,
"false_positive_rate_at_capacity": fpr,
}
def assign_bands(scores: pd.Series) -> pd.Series:
red_cut = scores.quantile(0.94)
amber_cut = scores.quantile(0.80)
return pd.Series(np.where(scores >= red_cut, "red", np.where(scores >= amber_cut, "amber", "green")), index=scores.index)
def reason_codes(row: pd.Series) -> list[str]:
reasons = []
if row["missing_assignments_to_date"] >= 1 or row["submission_completion_rate_to_date"] < 0.85:
reasons.append(
f"assessment: {int(row['missing_assignments_to_date'])} missed assignments, "
f"{row['submission_completion_rate_to_date']:.0%} submitted"
)
if row["days_since_last_lms_login"] >= 4 or row["lms_logins_14d"] <= 12:
reasons.append(f"digital engagement: {int(row['days_since_last_lms_login'])} days since last LMS login")
if row["payment_overdue_days"] > 0 or row["outstanding_balance_nok"] > 5000:
reasons.append(
f"finance: NOK {row['outstanding_balance_nok']:.0f} balance, "
f"{int(row['payment_overdue_days'])} overdue days"
)
if row["preferred_study_mode"] != "online-heavy" and row["campus_drop_14d_vs_start"] >= 3:
reasons.append(f"campus: {int(row['campus_drop_14d_vs_start'])} fewer active campus days vs term start")
if row["prior_gpa"] < 2.5:
reasons.append(f"academic history: prior GPA {row['prior_gpa']:.2f}")
if not reasons:
reasons.append("advisor review: multiple moderate deviations, no single dominant driver")
return reasons[:3]
def feature_hash(row: pd.Series, columns: list[str]) -> str:
payload = {}
for col in columns:
value = row[col]
if isinstance(value, (pd.Timestamp,)):
value = value.isoformat()
elif isinstance(value, (np.integer, np.floating)):
value = value.item()
payload[col] = value
return hashlib.sha256(json.dumps(payload, sort_keys=True, default=str).encode("utf-8")).hexdigest()[:18]
def fairness_report(scored: pd.DataFrame, threshold: float) -> pd.DataFrame:
rows = []
flagged = scored["risk_score"].ge(threshold)
label = scored["dropout_label_next_term"].astype(bool)
for group_col in ["gender", "international_student", "first_generation_student", "faculty_asof", "preferred_study_mode"]:
for value, frame in scored.groupby(group_col, dropna=False):
idx = frame.index
group_flagged = flagged.loc[idx]
group_label = label.loc[idx]
rows.append(
{
"group": group_col,
"value": str(value),
"students": len(frame),
"label_rate_demo_only": round(float(group_label.mean()), 4),
"avg_risk_score": round(float(frame["risk_score"].mean()), 4),
"flag_rate": round(float(group_flagged.mean()), 4),
"recall_if_labels_later_confirm": round(float((group_flagged & group_label).sum() / max(1, group_label.sum())), 4),
"false_positive_rate_if_labels_later_confirm": round(
float((group_flagged & ~group_label).sum() / max(1, (~group_label).sum())), 4
),
}
)
return pd.DataFrame(rows)
def data_dictionary() -> pd.DataFrame:
rows = [
("sis_students", "canonical_student_id", "string", "Pseudonymous university-wide student key."),
("sis_students", "faculty/programme", "string", "Point-in-time academic ownership from SIS."),
("sis_students", "gender, age_band, international_student", "string/bool", "Sensitive or demographic fields used for fairness audit, not default model input."),
("identity_map", "source_system/source_person_id", "string", "Maps SIS/LMS/ERP/CAMPUS identifiers to canonical student keys."),
("status_history", "valid_from/valid_to", "timestamp", "Type-2 history for programme/status changes."),
("enrollment_terms", "prior_gpa", "float", "Academic context available before the term starts."),
("enrollment_terms", "dropout_label_next_term", "bool", "Training label observed only after the term; never available at scoring time."),
("lms_activity_daily", "login_count/course_views/video_minutes", "integer", "Daily digital engagement events."),
("lms_activity_daily", "assignment_due/submitted/late", "integer", "Assessment progress signals."),
("erp_finance_snapshots", "outstanding_balance_nok/payment_overdue_days", "numeric", "Financial friction snapshot from ERP."),
("campus_activity_daily", "building_entry_count/wifi_minutes/library_entry_count", "integer", "Physical campus engagement aggregated by day."),
("feature_student_week", "as_of_date", "date", "Point-in-time feature timestamp used for leakage control."),
("risk_predictions", "risk_score/risk_band/top_reasons", "numeric/string", "Advisor-facing prediction output with explanation."),
("access_audit", "viewer/action/purpose/timestamp", "string/timestamp", "Evidence for who accessed predictions and why."),
]
return pd.DataFrame(rows, columns=["table", "column", "type", "description"])
def write_outputs(sources: SourceData, scored: pd.DataFrame, validation: pd.DataFrame, fairness: pd.DataFrame, out_dir: Path) -> Path:
out_dir.mkdir(parents=True, exist_ok=True)
source_dir = out_dir / "source_samples"
source_dir.mkdir(exist_ok=True)
sources.sis_students.to_csv(source_dir / "sis_students.csv", index=False)
sources.identity_map.to_csv(source_dir / "identity_map.csv", index=False)
sources.status_history.to_csv(source_dir / "status_history.csv", index=False)
sources.enrollment_terms.to_csv(source_dir / "enrollment_terms.csv", index=False)
sources.lms_activity_daily.to_csv(source_dir / "lms_activity_daily.csv", index=False)
sources.erp_finance_snapshots.to_csv(source_dir / "erp_finance_snapshots.csv", index=False)
sources.campus_activity_daily.to_csv(source_dir / "campus_activity_daily.csv", index=False)
scored.to_csv(out_dir / "feature_student_week_and_predictions.csv", index=False)
advisor_cols = [
"prediction_id",
"canonical_student_id",
"term_id",
"as_of_date",
"advisor_id",
"faculty_asof",
"programme_asof",
"risk_score",
"risk_band",
"top_reasons",
"model_version",
"feature_snapshot_hash",
]
advisor_view = scored[advisor_cols].copy()
advisor_view["risk_band_order"] = advisor_view["risk_band"].map({"red": 0, "amber": 1, "green": 2})
advisor_view = advisor_view.sort_values(["risk_band_order", "risk_score"], ascending=[True, False]).drop(
columns=["risk_band_order"]
)
advisor_view.to_csv(out_dir / "advisor_risk_list.csv", index=False)
validation.to_csv(out_dir / "model_validation_metrics.csv", index=False)
fairness.to_csv(out_dir / "fairness_report.csv", index=False)
data_dictionary().to_csv(out_dir / "data_dictionary.csv", index=False)
access_audit(scored).to_csv(out_dir / "access_audit_sample.csv", index=False)
plot_outputs(scored, fairness, out_dir)
dashboard_path = write_dashboard(scored, validation, fairness, out_dir)
return dashboard_path
def access_audit(scored: pd.DataFrame) -> pd.DataFrame:
viewed = scored.sort_values("risk_score", ascending=False).head(80).copy()
rows = []
for n, (_, row) in enumerate(viewed.iterrows(), start=1):
rows.append(
{
"audit_event_id": f"AUD-{n:05d}",
"event_timestamp": row["as_of_date"] + pd.Timedelta(hours=9, minutes=n % 50),
"viewer_user_id": row["advisor_id"],
"viewer_role": "academic_advisor",
"action": "view_student_risk_prediction",
"prediction_id": row["prediction_id"],
"canonical_student_id": row["canonical_student_id"],
"purpose": "student_support_intervention",
"legal_basis": "public_task_with_documented_DPIA",
"fields_returned": "risk_band, risk_score, top_reasons, programme, advisor_case_context",
"model_version": row["model_version"],
}
)
return pd.DataFrame(rows)
def plot_outputs(scored: pd.DataFrame, fairness: pd.DataFrame, out_dir: Path) -> None:
fig, ax = plt.subplots(figsize=(9, 4.8))
for band, color in [("green", "#3a7d44"), ("amber", "#b98517"), ("red", "#b3261e")]:
values = scored.loc[scored["risk_band"].eq(band), "risk_score"]
ax.hist(values, bins=24, alpha=0.72, label=band, color=color)
ax.set_title("Risk score distribution by advisor band")
ax.set_xlabel("Predicted dropout risk")
ax.set_ylabel("Students")
ax.legend()
fig.tight_layout()
fig.savefig(out_dir / "risk_distribution.png", dpi=160)
plt.close(fig)
selected = fairness.loc[fairness["group"].isin(["preferred_study_mode", "international_student", "first_generation_student"])]
selected = selected.sort_values(["group", "value"])
fig, ax = plt.subplots(figsize=(9, 4.8))
labels = selected["group"].str.replace("_", " ") + ": " + selected["value"]
ax.barh(labels, selected["flag_rate"], color="#386fa4")
ax.set_title("Flag rate checks across selected groups")
ax.set_xlabel("Flag rate at advisor capacity")
ax.set_xlim(0, max(0.25, float(selected["flag_rate"].max()) + 0.03))
fig.tight_layout()
fig.savefig(out_dir / "fairness_flag_rates.png", dpi=160)
plt.close(fig)
def dashboard_payload(scored: pd.DataFrame, validation: pd.DataFrame, fairness: pd.DataFrame) -> dict:
advisor_cols = [
"prediction_id",
"canonical_student_id",
"advisor_id",
"faculty_asof",
"programme_asof",
"preferred_study_mode",
"academic_standing",
"student_status_asof",
"aid_status",
"risk_score",
"risk_band",
"top_reasons",
"prior_gpa",
"credits_registered",
"credits_completed_previous_term",
"lms_logins_14d",
"lms_course_views_14d",
"video_minutes_14d",
"forum_posts_14d",
"quiz_attempts_14d",
"days_since_last_lms_login",
"missing_assignments_to_date",
"late_submissions_to_date",
"submission_completion_rate_to_date",
"payment_overdue_days",
"outstanding_balance_nok",
"campus_days_14d",
"building_entries_14d",
"library_entries_30d",
"campus_drop_14d_vs_start",
"feature_snapshot_hash",
]
advisor_rows = scored[advisor_cols].copy()
advisor_rows["risk_score"] = advisor_rows["risk_score"].round(4)
advisor_rows["prior_gpa"] = advisor_rows["prior_gpa"].round(2)
advisor_rows["submission_completion_rate_to_date"] = advisor_rows["submission_completion_rate_to_date"].round(4)
advisor_rows["outstanding_balance_nok"] = advisor_rows["outstanding_balance_nok"].round(0).astype(int)
for column in [
"credits_registered",
"credits_completed_previous_term",
"lms_logins_14d",
"lms_course_views_14d",
"video_minutes_14d",
"forum_posts_14d",
"quiz_attempts_14d",
"days_since_last_lms_login",
"missing_assignments_to_date",
"late_submissions_to_date",
"payment_overdue_days",
"campus_days_14d",
"building_entries_14d",
"library_entries_30d",
"campus_drop_14d_vs_start",
]:
advisor_rows[column] = advisor_rows[column].round(0).astype(int)
advisor_rows["risk_band_order"] = advisor_rows["risk_band"].map({"red": 0, "amber": 1, "green": 2})
advisor_rows = advisor_rows.sort_values(["risk_band_order", "risk_score"], ascending=[True, False]).drop(
columns=["risk_band_order"]
)
faculty = (
scored.groupby("faculty_asof")
.agg(
students=("canonical_student_id", "count"),
avg_risk=("risk_score", "mean"),
red=("risk_band", lambda values: int((values == "red").sum())),
amber=("risk_band", lambda values: int((values == "amber").sum())),
)
.reset_index()
.sort_values("avg_risk", ascending=False)
)
faculty["avg_risk"] = faculty["avg_risk"].round(4)
band_counts = scored["risk_band"].value_counts().reindex(["red", "amber", "green"], fill_value=0)
validation_rows = validation.round(4).to_dict(orient="records")
fairness_rows = fairness.to_dict(orient="records")
return {
"generated_at": pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S"),
"as_of_date": str(scored["as_of_date"].iloc[0].date()),
"model_version": str(scored["model_version"].iloc[0]),
"students_scored": int(len(scored)),
"band_counts": {band: int(count) for band, count in band_counts.items()},
"average_risk": round(float(scored["risk_score"].mean()), 4),
"validation": validation_rows,
"faculty": faculty.to_dict(orient="records"),
"fairness": fairness_rows,
"students": advisor_rows.to_dict(orient="records"),
}
def write_dashboard(scored: pd.DataFrame, validation: pd.DataFrame, fairness: pd.DataFrame, out_dir: Path) -> Path:
payload = dashboard_payload(scored, validation, fairness)
data_json = json.dumps(payload, default=str)
html = DASHBOARD_TEMPLATE.replace("__DASHBOARD_DATA__", data_json)
dashboard_path = out_dir / "demo_dashboard.html"
dashboard_path.write_text(html, encoding="utf-8")
return dashboard_path
DASHBOARD_TEMPLATE = r"""
Student Outcome Intelligence Demo
Student Outcome Intelligence Demo
Interactive front end for the generated dropout-risk prototype: advisor risk list, model validation, faculty patterns, fairness checks, and audit-ready explanation fields.
As of
Model
Generated
Students scored
0
Active students in the Spring 2026 dummy scoring cohort.
Red cases
0
Highest-priority advisor queue.
Amber cases
0
Students worth monitoring or light-touch outreach.
Holdout ROC-AUC
0.00
Demo validation using point-in-time Autumn 2025 snapshots.
Band v Student Faculty / Programme Score Signals Reasons
Advisor
No students match the current filters.
Group
Value
Students
Label rate
Avg score
Flag rate
Recall
False positive rate
"""
def run_demo(out_dir: Path, n_students: int, seed: int, open_dashboard: bool) -> None:
sources = generate_sources(n_students=n_students, seed=seed)
train_snapshots = [
build_feature_snapshot(sources, "2025A", pd.Timestamp("2025-09-15")),
build_feature_snapshot(sources, "2025A", pd.Timestamp("2025-10-06")),
build_feature_snapshot(sources, "2025A", pd.Timestamp("2025-10-27")),
]
train_all = pd.concat(train_snapshots, ignore_index=True)
score_features = build_feature_snapshot(sources, "2026S", pd.Timestamp("2026-02-23"))
numeric_cols = [
"prior_gpa",
"credits_registered",
"credits_completed_previous_term",
"term_week",
"lms_logins_14d",
"lms_course_views_14d",
"video_minutes_14d",
"forum_posts_14d",
"quiz_attempts_14d",
"days_since_last_lms_login",
"missing_assignments_to_date",
"late_submissions_to_date",
"submission_completion_rate_to_date",
"outstanding_balance_log1p",
"payment_overdue_days",
"financial_overdue_flag",
"campus_days_14d",
"building_entries_14d",
"library_entries_30d",
"campus_drop_14d_vs_start",
]
categorical_cols = ["faculty_asof", "academic_standing", "preferred_study_mode", "aid_status", "student_status_asof"]
# Protected and sensitive demographic fields are retained for audit reporting, not default model input.
x_train, x_score, train_design, means, stds = prepare_design_matrix(train_all, score_features, numeric_cols, categorical_cols)
y_train = train_all["dropout_label_next_term"].astype(int).to_numpy()
model = NumpyLogisticRegression().fit(x_train, y_train)
score_features = score_features.copy()
score_features["risk_score"] = model.predict_proba(x_score)
score_features["risk_band"] = assign_bands(score_features["risk_score"])
score_features["model_version"] = MODEL_VERSION
score_features["prediction_id"] = [f"PRED-2026S-{i:06d}" for i in range(1, len(score_features) + 1)]
score_features["top_reasons"] = score_features.apply(lambda row: " | ".join(reason_codes(row)), axis=1)
hash_cols = numeric_cols + categorical_cols + ["as_of_date", "term_id"]
score_features["feature_snapshot_hash"] = score_features.apply(lambda row: feature_hash(row, hash_cols), axis=1)
# Demo-only validation: in a real deployment, the 2026S labels are unknown at scoring time.
rng = np.random.default_rng(seed + 101)
mask = rng.random(len(train_all)) < 0.25
holdout = train_all.loc[mask].copy()
train_part = train_all.loc[~mask].copy()
x_train_part, x_holdout, train_design_part, means_part, stds_part = prepare_design_matrix(
train_part, holdout, numeric_cols, categorical_cols
)
holdout_model = NumpyLogisticRegression().fit(x_train_part, train_part["dropout_label_next_term"].astype(int).to_numpy())
holdout_scores = holdout_model.predict_proba(x_holdout)
holdout_metrics = classification_metrics(holdout["dropout_label_next_term"].astype(int).to_numpy(), holdout_scores, 0.12)
demo_metrics = classification_metrics(
score_features["dropout_label_next_term"].astype(int).to_numpy(), score_features["risk_score"].to_numpy(), 0.12
)
validation = pd.DataFrame(
[
{"dataset": "2025A_holdout_point_in_time", **{k: round(v, 4) for k, v in holdout_metrics.items()}},
{"dataset": "2026S_labels_for_demo_only", **{k: round(v, 4) for k, v in demo_metrics.items()}},
]
)
threshold = float(np.quantile(score_features["risk_score"], 0.88))
fairness = fairness_report(score_features, threshold)
dashboard_path = write_outputs(sources, score_features, validation, fairness, out_dir)
red = int(score_features["risk_band"].eq("red").sum())
amber = int(score_features["risk_band"].eq("amber").sum())
print(f"Wrote outputs to: {out_dir.resolve()}")
print(f"Students scored: {len(score_features):,}")
print(f"Advisor bands: red={red}, amber={amber}, green={len(score_features) - red - amber}")
print(f"Validation ROC-AUC (holdout): {validation.loc[0, 'roc_auc']}")
print(f"Advisor list: {(out_dir / 'advisor_risk_list.csv').resolve()}")
print(f"Dashboard: {dashboard_path.resolve()}")
if open_dashboard:
opened = webbrowser.open(dashboard_path.resolve().as_uri())
if opened:
print("Opened dashboard in your default browser.")
else:
print("Could not open the browser automatically; open the dashboard path above.")
def main() -> None:
parser = argparse.ArgumentParser(description="Generate a dummy Student Outcome Intelligence Platform demo.")
parser.add_argument("--out", default="outputs", help="Output folder.")
parser.add_argument("--students", type=int, default=1200, help="Number of dummy students to generate.")
parser.add_argument("--seed", type=int, default=42, help="Random seed.")
parser.add_argument(
"--no-open-dashboard",
action="store_true",
help="Generate the HTML dashboard but do not open it in a browser.",
)
args = parser.parse_args()
run_demo(Path(args.out), n_students=args.students, seed=args.seed, open_dashboard=not args.no_open_dashboard)
if __name__ == "__main__":
main()