src/studypulse/metrics.py - master - nevers/hackINTERPOL

src/studypulse/metrics.py
131 строка4 KB
nevers
upload files
13 янв 2026, 17:06
13 янв 2026, 17:06c306496
"""Analytics and risk flag computation for StudyPulse."""

from typing import Any

import pandas as pd


def compute_descriptive_stats(df: pd.DataFrame) -> dict[str, Any]:
    """Compute descriptive statistics for the dataset.

    Args:
        df: DataFrame with student data.

    Returns:
        Dictionary of descriptive statistics.
    """
    stats: dict[str, Any] = {}

    stats["total_students"] = len(df)
    stats["mean_assignment_score"] = float(df["assignment_score"].mean())
    stats["median_assignment_score"] = float(df["assignment_score"].median())
    stats["std_assignment_score"] = float(df["assignment_score"].std())
    stats["mean_attendance_rate"] = float(df["attendance_rate"].mean())
    stats["mean_assignments_submitted"] = float(df["assignments_submitted"].mean())
    stats["mean_days_since_last_submission"] = float(
        df["days_since_last_submission"].mean()
    )

    # Count students at risk by various criteria
    stats["low_score_count"] = int(len(df[df["assignment_score"] < 60]))
    stats["low_attendance_count"] = int(len(df[df["attendance_rate"] < 0.7]))
    stats["inactive_count"] = int(
        len(df[df["days_since_last_submission"] > 14])
    )
    stats["behind_deadline_count"] = int(len(df[df["days_to_deadline"] < 0]))

    return stats


def compute_risk_flags(df: pd.DataFrame) -> pd.DataFrame:
    """Compute risk flags for each student.

    Args:
        df: DataFrame with student data.

    Returns:
        DataFrame with added risk_flag column (0=low risk, 1=high risk).
    """
    df = df.copy()

    # Initialize risk flag
    df["risk_flag"] = 0

    # Flag students with low scores
    df.loc[df["assignment_score"] < 60, "risk_flag"] = 1

    # Flag students with low attendance
    df.loc[df["attendance_rate"] < 0.7, "risk_flag"] = 1

    # Flag inactive students (no submission in 14+ days)
    df.loc[df["days_since_last_submission"] > 14, "risk_flag"] = 1

    # Flag students behind deadline
    df.loc[df["days_to_deadline"] < 0, "risk_flag"] = 1

    return df


def compute_risk_score(df: pd.DataFrame) -> pd.DataFrame:
    """Compute a composite risk score (0-100) for each student.

    Args:
        df: DataFrame with student data.

    Returns:
        DataFrame with added risk_score column.
    """
    df = df.copy()

    # Normalize components (higher = more risk)
    score_risk = 100 - df["assignment_score"]  # Low score = high risk
    attendance_risk = (1 - df["attendance_rate"]) * 100  # Low attendance = high risk
    inactivity_risk = (
        df["days_since_last_submission"] / 30.0 * 100
    ).clip(0, 100)  # More days = more risk
    deadline_risk = (
        (-df["days_to_deadline"] / 7.0 * 100).clip(0, 100)
        if (df["days_to_deadline"] < 0).any()
        else pd.Series([0] * len(df))
    )

    # Weighted combination
    df["risk_score"] = (
        0.4 * score_risk
        + 0.3 * attendance_risk
        + 0.2 * inactivity_risk
        + 0.1 * deadline_risk
    )

    return df


def get_top_risks(df: pd.DataFrame, top_n: int = 10) -> pd.DataFrame:
    """Get top N students by risk score.

    Args:
        df: DataFrame with risk_score column.
        top_n: Number of top risks to return.

    Returns:
        DataFrame sorted by risk_score (descending), limited to top_n.
    """
    if "risk_score" not in df.columns:
        df = compute_risk_score(df)

    top_risks = (
        df.nlargest(top_n, "risk_score")[
            [
                "student_id",
                "student_name",
                "assignment_score",
                "attendance_rate",
                "days_since_last_submission",
                "risk_score",
            ]
        ]
        .copy()
        .reset_index(drop=True)
    )

    return top_risks