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

src/studypulse/model.py
113 строк3 KB
nevers
upload files
13 янв 2026, 17:14
13 янв 2026, 17:14e86a850
"""Risk prediction model for StudyPulse."""

import pandas as pd
from sklearn.linear_model import LogisticRegression

from studypulse.utils import set_deterministic_seed


def train_risk_model(df: pd.DataFrame) -> LogisticRegression:
    """Train a logistic regression model to predict risk flags.

    Args:
        df: DataFrame with student data and risk_flag column.

    Returns:
        Trained LogisticRegression model.
    """
    set_deterministic_seed(42)

    # Features for prediction
    feature_cols = [
        "assignment_score",
        "assignments_submitted",
        "attendance_rate",
        "days_since_last_submission",
        "days_to_deadline",
    ]

    X = df[feature_cols].values
    y = df["risk_flag"].values

    # Train logistic regression
    model = LogisticRegression(random_state=42, max_iter=1000)
    model.fit(X, y)

    return model


def predict_risk(
    df: pd.DataFrame, model: LogisticRegression | None = None
) -> pd.DataFrame:
    """Predict risk flags for students using trained model or heuristic fallback.

    Args:
        df: DataFrame with student data.
        model: Optional pre-trained model. If None, uses heuristic fallback.

    Returns:
        DataFrame with added predicted_risk_flag and predicted_risk_probability columns.
    """
    df = df.copy()

    feature_cols = [
        "assignment_score",
        "assignments_submitted",
        "attendance_rate",
        "days_since_last_submission",
        "days_to_deadline",
    ]

    # Use model if available and dataset is large enough
    if model is not None and len(df) >= 5:
        X = df[feature_cols].values
        df["predicted_risk_flag"] = model.predict(X)
        df["predicted_risk_probability"] = model.predict_proba(X)[:, 1]
    else:
        # Heuristic fallback: deterministic rules
        df["predicted_risk_flag"] = 0
        df["predicted_risk_probability"] = 0.0

        # Apply same rules as risk_flag computation
        risk_mask = (
            (df["assignment_score"] < 60)
            | (df["attendance_rate"] < 0.7)
            | (df["days_since_last_submission"] > 14)
            | (df["days_to_deadline"] < 0)
        )

        df.loc[risk_mask, "predicted_risk_flag"] = 1
        df.loc[risk_mask, "predicted_risk_probability"] = 0.8

    return df


def train_and_predict(df: pd.DataFrame) -> pd.DataFrame:
    """Train model on dataset and predict risks.

    Uses model if dataset is large enough, otherwise falls back to heuristic.

    Args:
        df: DataFrame with student data and risk_flag column.

    Returns:
        DataFrame with added predicted_risk_flag and predicted_risk_probability columns.
    """
    # Compute risk flags first
    from studypulse.metrics import compute_risk_flags

    df_with_flags = compute_risk_flags(df)

    # Train model if dataset is large enough
    model = None
    if len(df_with_flags) >= 10:
        try:
            model = train_risk_model(df_with_flags)
        except Exception:
            # Fallback to heuristic if training fails
            model = None

    # Predict
    result = predict_risk(df_with_flags, model)

    return result