Learning Analytics in Python — Deep Dive

Build learning analytics pipelines in Python with xAPI event processing, at-risk prediction models, engagement scoring, and real-time dashboards.

Learning analytics systems ingest event streams from learning platforms, compute engagement and performance features, run predictive models, and surface results through dashboards. This guide builds each component.

xAPI Event Processing

The xAPI standard formats learning events as JSON statements:

from dataclasses import dataclass
from datetime import datetime
from typing import Optional
import json

@dataclass
class LearningEvent:
    actor_id: str
    verb: str
    object_id: str
    object_type: str
    timestamp: datetime
    duration_seconds: Optional[float] = None
    score: Optional[float] = None
    max_score: Optional[float] = None
    context: Optional[dict] = None

def parse_xapi_statement(raw: dict) -> LearningEvent:
    """Parse an xAPI statement into a structured event."""
    return LearningEvent(
        actor_id=raw["actor"]["account"]["name"],
        verb=raw["verb"]["id"].split("/")[-1],  # e.g., "completed"
        object_id=raw["object"]["id"],
        object_type=raw["object"].get("definition", {}).get("type", "unknown").split("/")[-1],
        timestamp=datetime.fromisoformat(raw["timestamp"].replace("Z", "+00:00")),
        duration_seconds=_parse_duration(raw.get("result", {}).get("duration")),
        score=raw.get("result", {}).get("score", {}).get("raw"),
        max_score=raw.get("result", {}).get("score", {}).get("max"),
        context=raw.get("context", {}),
    )

def _parse_duration(iso_duration: str | None) -> float | None:
    """Parse ISO 8601 duration to seconds."""
    if not iso_duration:
        return None
    import re
    match = re.match(r'PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+(?:\.\d+)?)S)?', iso_duration)
    if not match:
        return None
    hours = float(match.group(1) or 0)
    minutes = float(match.group(2) or 0)
    seconds = float(match.group(3) or 0)
    return hours * 3600 + minutes * 60 + seconds

Feature Engineering Pipeline

Transform raw events into student-level features for modeling:

import pandas as pd
import numpy as np
from collections import defaultdict

class FeatureEngineer:
    def __init__(self, course_id: str):
        self.course_id = course_id

    def compute_features(self, events: list[LearningEvent],
                          as_of_date: datetime = None) -> pd.DataFrame:
        """Compute per-student feature vectors from event logs."""
        if as_of_date:
            events = [e for e in events if e.timestamp <= as_of_date]

        student_events = defaultdict(list)
        for e in events:
            student_events[e.actor_id].append(e)

        rows = []
        for student_id, evts in student_events.items():
            evts.sort(key=lambda e: e.timestamp)
            rows.append(self._student_features(student_id, evts))

        return pd.DataFrame(rows)

    def _student_features(self, student_id: str,
                           events: list[LearningEvent]) -> dict:
        """Extract features for a single student."""
        video_events = [e for e in events if e.object_type == "video"]
        quiz_events = [e for e in events if e.verb == "completed" and e.score is not None]
        assignment_events = [e for e in events if e.object_type == "assignment"]
        forum_events = [e for e in events if e.verb in ("posted", "replied")]

        # Time-based features
        session_gaps = []
        for i in range(1, len(events)):
            gap = (events[i].timestamp - events[i-1].timestamp).total_seconds()
            if gap < 3600:  # Same session if gap < 1 hour
                session_gaps.append(gap)

        total_time = sum(e.duration_seconds or 0 for e in events)
        unique_days = len({e.timestamp.date() for e in events})

        # Video engagement
        video_time = sum(e.duration_seconds or 0 for e in video_events)
        videos_completed = sum(1 for e in video_events if e.verb == "completed")
        videos_started = sum(1 for e in video_events if e.verb in ("played", "started"))

        # Assessment performance
        quiz_scores = [e.score / e.max_score for e in quiz_events
                       if e.max_score and e.max_score > 0]
        score_trend = self._linear_trend(quiz_scores) if len(quiz_scores) >= 3 else 0.0

        # Submission timing
        late_submissions = sum(1 for e in assignment_events
                               if e.context and e.context.get("late", False))

        return {
            "student_id": student_id,
            "total_events": len(events),
            "unique_active_days": unique_days,
            "total_time_minutes": round(total_time / 60, 1),
            "avg_session_minutes": round(np.mean(session_gaps) / 60, 1) if session_gaps else 0,
            "video_completion_rate": videos_completed / max(videos_started, 1),
            "video_time_minutes": round(video_time / 60, 1),
            "quiz_count": len(quiz_scores),
            "quiz_avg_score": round(np.mean(quiz_scores), 3) if quiz_scores else 0,
            "quiz_score_trend": round(score_trend, 4),
            "forum_posts": len(forum_events),
            "late_submissions": late_submissions,
            "days_since_last_activity": self._days_since_last(events),
            "regularity_score": self._regularity(events),
        }

    def _linear_trend(self, values: list[float]) -> float:
        """Compute linear trend (slope) of a series."""
        if len(values) < 2:
            return 0.0
        x = np.arange(len(values))
        coeffs = np.polyfit(x, values, 1)
        return coeffs[0]

    def _days_since_last(self, events: list[LearningEvent]) -> int:
        if not events:
            return 999
        last = max(e.timestamp for e in events)
        return (datetime.now(last.tzinfo) - last).days

    def _regularity(self, events: list[LearningEvent]) -> float:
        """Measure how regularly the student engages (0-1)."""
        if len(events) < 2:
            return 0.0
        dates = sorted({e.timestamp.date() for e in events})
        if len(dates) < 2:
            return 0.0
        gaps = [(dates[i+1] - dates[i]).days for i in range(len(dates)-1)]
        # Lower std deviation of gaps = more regular
        if np.mean(gaps) == 0:
            return 1.0
        cv = np.std(gaps) / np.mean(gaps)  # Coefficient of variation
        return max(0, 1 - cv)  # Higher = more regular

At-Risk Prediction Model

from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import cross_val_score
from sklearn.calibration import CalibratedClassifierCV
import joblib

class AtRiskPredictor:
    def __init__(self):
        base_model = GradientBoostingClassifier(
            n_estimators=200, max_depth=4,
            learning_rate=0.1, subsample=0.8,
            min_samples_leaf=20,
        )
        # Calibrate probabilities for meaningful risk scores
        self.model = CalibratedClassifierCV(base_model, cv=5)
        self.feature_columns = None

    def train(self, features_df: pd.DataFrame, labels: pd.Series):
        """Train on historical data. Labels: 1=dropped/failed, 0=succeeded."""
        self.feature_columns = [c for c in features_df.columns if c != "student_id"]
        X = features_df[self.feature_columns]
        self.model.fit(X, labels)

        # Cross-validation score
        scores = cross_val_score(self.model, X, labels, cv=5, scoring="roc_auc")
        print(f"Cross-validated AUC: {scores.mean():.3f} (+/- {scores.std():.3f})")

    def predict_risk(self, features_df: pd.DataFrame) -> pd.DataFrame:
        """Predict risk scores for current students."""
        X = features_df[self.feature_columns]
        probs = self.model.predict_proba(X)[:, 1]

        result = features_df[["student_id"]].copy()
        result["risk_score"] = np.round(probs, 3)
        result["risk_level"] = pd.cut(
            probs, bins=[0, 0.3, 0.6, 1.0],
            labels=["low", "moderate", "high"]
        )
        return result.sort_values("risk_score", ascending=False)

    def feature_importance(self) -> pd.DataFrame:
        """Return feature importance rankings."""
        # Access the base estimator from calibrated model
        importances = np.mean([
            est.estimators_[0].feature_importances_
            for est in self.model.calibrated_classifiers_
        ], axis=0)
        return pd.DataFrame({
            "feature": self.feature_columns,
            "importance": importances,
        }).sort_values("importance", ascending=False)

    def save(self, path: str):
        joblib.dump({"model": self.model, "features": self.feature_columns}, path)

    @classmethod
    def load(cls, path: str):
        data = joblib.load(path)
        predictor = cls()
        predictor.model = data["model"]
        predictor.feature_columns = data["features"]
        return predictor

Engagement Scoring

Combine multiple engagement signals into a single composite score:

class EngagementScorer:
    """Compute composite engagement scores with configurable weights."""

    DEFAULT_WEIGHTS = {
        "regularity_score": 0.20,
        "video_completion_rate": 0.15,
        "quiz_avg_score": 0.20,
        "forum_posts_normalized": 0.10,
        "total_time_normalized": 0.15,
        "days_since_last_normalized": 0.20,
    }

    def score(self, features_df: pd.DataFrame) -> pd.DataFrame:
        """Add engagement score column to features DataFrame."""
        df = features_df.copy()

        # Normalize features to 0-1 range
        df["forum_posts_normalized"] = df["forum_posts"].clip(upper=20) / 20
        max_time = df["total_time_minutes"].quantile(0.95)
        df["total_time_normalized"] = (df["total_time_minutes"] / max(max_time, 1)).clip(upper=1)
        df["days_since_last_normalized"] = 1 - (df["days_since_last_activity"].clip(upper=14) / 14)

        # Weighted composite
        df["engagement_score"] = sum(
            df[feature] * weight
            for feature, weight in self.DEFAULT_WEIGHTS.items()
            if feature in df.columns
        )
        df["engagement_score"] = df["engagement_score"].round(3)

        return df

Intervention Automation

from datetime import datetime

class InterventionEngine:
    def __init__(self, risk_threshold: float = 0.6):
        self.risk_threshold = risk_threshold
        self.intervention_log = []

    def generate_interventions(self, risk_df: pd.DataFrame,
                                features_df: pd.DataFrame) -> list[dict]:
        """Generate targeted interventions for at-risk students."""
        high_risk = risk_df[risk_df["risk_score"] >= self.risk_threshold]
        interventions = []

        for _, row in high_risk.iterrows():
            student_id = row["student_id"]
            student_features = features_df[features_df["student_id"] == student_id].iloc[0]

            intervention = {
                "student_id": student_id,
                "risk_score": row["risk_score"],
                "timestamp": datetime.now().isoformat(),
                "actions": [],
            }

            # Determine specific interventions based on feature patterns
            if student_features["days_since_last_activity"] > 7:
                intervention["actions"].append({
                    "type": "email_nudge",
                    "template": "re_engagement",
                    "message": "We noticed you have not logged in recently. "
                              "Your next assignment is due soon.",
                })

            if student_features["video_completion_rate"] < 0.3:
                intervention["actions"].append({
                    "type": "content_recommendation",
                    "template": "video_reminder",
                    "message": "Students who watch the lecture videos score "
                              "25% higher on assessments.",
                })

            if student_features["quiz_score_trend"] < -0.1:
                intervention["actions"].append({
                    "type": "instructor_alert",
                    "template": "declining_performance",
                    "message": f"Student {student_id} shows declining quiz scores. "
                              "Consider reaching out.",
                })

            if intervention["actions"]:
                interventions.append(intervention)
                self.intervention_log.append(intervention)

        return interventions

Dashboard Data API

Serve analytics to frontend dashboards:

from fastapi import FastAPI, Query
from typing import Optional

app = FastAPI(title="Learning Analytics API")

@app.get("/api/course/{course_id}/overview")
def course_overview(course_id: str):
    """Course-level engagement summary."""
    # In production, query from database
    return {
        "total_students": 245,
        "active_this_week": 198,
        "at_risk_count": 23,
        "avg_engagement": 0.72,
        "content_completion": {
            "videos": 0.68,
            "readings": 0.45,
            "quizzes": 0.82,
        },
    }

@app.get("/api/course/{course_id}/at-risk")
def at_risk_students(course_id: str,
                     min_risk: float = Query(0.6, ge=0, le=1),
                     limit: int = Query(20, ge=1, le=100)):
    """List at-risk students with risk scores and recommended actions."""
    # Query prediction results from database
    return {"students": [], "total": 0}

@app.get("/api/student/{student_id}/timeline")
def student_timeline(student_id: str,
                     course_id: Optional[str] = None):
    """Individual student engagement timeline."""
    return {
        "student_id": student_id,
        "weekly_engagement": [],
        "assessment_scores": [],
        "risk_history": [],
    }

Measuring Intervention Effectiveness

Track whether interventions actually improve outcomes using causal inference:

from scipy import stats

def measure_intervention_impact(intervention_group: pd.DataFrame,
                                 control_group: pd.DataFrame,
                                 outcome_column: str = "final_grade") -> dict:
    """Compare outcomes between intervention and control groups."""
    treated = intervention_group[outcome_column].dropna()
    control = control_group[outcome_column].dropna()

    t_stat, p_value = stats.ttest_ind(treated, control, equal_var=False)
    effect_size = (treated.mean() - control.mean()) / control.std()  # Cohen's d

    return {
        "treated_mean": round(treated.mean(), 2),
        "control_mean": round(control.mean(), 2),
        "difference": round(treated.mean() - control.mean(), 2),
        "cohens_d": round(effect_size, 3),
        "p_value": round(p_value, 4),
        "significant": p_value < 0.05,
        "treated_n": len(treated),
        "control_n": len(control),
    }

The one thing to remember: Learning analytics in Python is a pipeline — ingest events, engineer features, predict risk, trigger interventions, measure impact — and the entire system’s value depends on closing that loop so that predictions actually lead to actions that measurably improve student outcomes.

pythonlearning-analyticseducation-technologydata-analysis