Education and public services

Data mapping

Entity

student or citizen (pseudonymised identifier)

Group

institution or office

Profile

generic; configure time windows with verticals.register

Reference implementation

from datetime import datetime, timedelta
from vertiq import plugins, rules
from vertiq.mapping import Mapping, apply, repair
from vertiq.tasks import predict_event, predict_value

# rows: the source export as a list of dicts, e.g. list(csv.DictReader(open("export.csv")))
mapping = Mapping.parse({"entity": "Chhatra ID", "time": "Date", "type": "Activity", "value": "Score", "group": "School", "attrs": ["Subject", "Section", "Due", "Submitted"], "date_format": "%d-%m-%Y"})
m, notes = repair(mapping, rows)      # validates column names, value types and date format against the rows
ds, problems = apply(m, rows)         # rows that fail to load are reported, not imputed

@plugins.feature("school")
def absence_streak(entity, before, cut):   # receives only events recorded before the cutoff
    days = sorted({e.at.date() for e in before if e.type == "attendance"})
    gap = (cut.date() - days[-1]).days if days else 999
    return {"days_since_present": gap}

task = predict_event(ds, target="attendance", window_days=7, pack="school")
print(task.status, task.report())     # "trained" only if the model outperformed the baseline on held-out entities
if task.status == "trained":
    print(task.score("S4402"))  # probability with contributing factors

size = predict_value(ds, target="assessment", window_days=60, pack="school")
print(size.status, size.report())

flags = rules.run(ds, [
    rules.overdue("Due", "Submitted", tolerance_days=2, type="assignment"),   # assignments submitted more than two days late
])
print(flags["summary"])

This code is executed by the test suite (tests/test_guides.py) against generated records with these column names.