File size: 5,840 Bytes
3336428
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
"""
data_preprocessing.py
----------------------
Handles data ingestion, cleaning, type casting, and feature engineering
for the Student Performance Prediction System.

Demonstrates: variables & data types, casting, control structures,
loops, collections (lists/dicts/sets), and functions -- all wired
into a real Pandas/NumPy preprocessing pipeline.
"""

from __future__ import annotations
import numpy as np
import pandas as pd


# --------------------------------------------------------------------
# Column groupings (collections: lists & dicts used throughout)
# --------------------------------------------------------------------
NUMERIC_FEATURES: list[str] = [
    "study_hours_per_day",
    "attendance_percentage",
    "previous_exam_score",
    "sleep_hours",
    "screen_time_hours",
    "age",
]

CATEGORICAL_FEATURES: list[str] = [
    "gender",
    "extracurricular_activities",
    "parental_education",
    "internet_access",
    "study_group",
    "tutoring",
]

TARGET_COLUMN: str = "final_exam_score"

# Ordinal mapping for parental education (dictionary collection)
EDUCATION_ORDER: dict[str, int] = {
    "High School": 0,
    "Bachelors": 1,
    "Masters": 2,
    "PhD": 3,
}

# Binary yes/no columns get mapped with this dict
YES_NO_MAP: dict[str, int] = {"Yes": 1, "No": 0}


def load_data(path: str) -> pd.DataFrame:
    """Load the raw CSV dataset into a DataFrame."""
    df = pd.read_csv(path)
    return df


def cast_types(df: pd.DataFrame) -> pd.DataFrame:
    """Explicit type casting for numeric columns (float/int)."""
    df = df.copy()
    for col in NUMERIC_FEATURES:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors="coerce").astype(float)
    return df


def handle_missing_values(df: pd.DataFrame) -> pd.DataFrame:
    """
    Clean missing data using simple, explainable rules:
    - Numeric columns -> fill with column median (NumPy)
    - Categorical columns -> fill with the mode (most frequent value)
    Loops over each column list rather than blindly calling fillna on
    everything, to keep the logic auditable.
    """
    df = df.copy()

    for col in NUMERIC_FEATURES:
        if col in df.columns and df[col].isna().any():
            median_val = np.nanmedian(df[col].values)
            df[col] = df[col].fillna(median_val)

    for col in CATEGORICAL_FEATURES:
        if col in df.columns and df[col].isna().any():
            mode_val = df[col].mode(dropna=True)
            fill_value = mode_val.iloc[0] if not mode_val.empty else "Unknown"
            df[col] = df[col].fillna(fill_value)

    return df


def engineer_features(df: pd.DataFrame) -> pd.DataFrame:
    """
    Feature engineering using control structures & custom functions.
    Adds:
      - parental_education_score (ordinal encoding via dict + match-case)
      - binary-encoded yes/no columns
      - a composite 'engagement_index' feature
      - a categorical 'performance_band' label (for EDA, not the model)
    """
    df = df.copy()

    # match-case example: encode parental education explicitly
    def encode_education(level: str) -> int:
        match level:
            case "High School":
                return 0
            case "Bachelors":
                return 1
            case "Masters":
                return 2
            case "PhD":
                return 3
            case _:
                return EDUCATION_ORDER.get(level, 0)

    if "parental_education" in df.columns:
        df["parental_education_score"] = df["parental_education"].apply(encode_education)

    # Binary yes/no encodings
    binary_cols = ["extracurricular_activities", "internet_access", "study_group", "tutoring"]
    for col in binary_cols:
        if col in df.columns:
            df[f"{col}_encoded"] = df[col].map(YES_NO_MAP).fillna(0).astype(int)

    # Composite engineered feature (vectorised NumPy arithmetic)
    if {"study_hours_per_day", "attendance_percentage", "screen_time_hours"}.issubset(df.columns):
        df["engagement_index"] = (
            df["study_hours_per_day"] * 10
            + df["attendance_percentage"] * 0.5
            - df["screen_time_hours"] * 3
        )

    # Performance band using a simple loop/if-elif control structure
    def band_for_score(score: float) -> str:
        if score >= 85:
            return "Excellent"
        elif score >= 70:
            return "Good"
        elif score >= 50:
            return "Average"
        else:
            return "Needs Improvement"

    if TARGET_COLUMN in df.columns:
        df["performance_band"] = [band_for_score(s) for s in df[TARGET_COLUMN]]

    return df


def get_feature_matrix(df: pd.DataFrame) -> tuple[pd.DataFrame, pd.Series]:
    """
    Build the final X (features) and y (target) used by the model.
    Returns a tuple (collections concept in action).
    """
    feature_columns = [
        "study_hours_per_day",
        "attendance_percentage",
        "previous_exam_score",
        "sleep_hours",
        "screen_time_hours",
        "parental_education_score",
        "extracurricular_activities_encoded",
        "internet_access_encoded",
        "study_group_encoded",
        "tutoring_encoded",
        "engagement_index",
    ]
    available = [c for c in feature_columns if c in df.columns]
    X = df[available]
    y = df[TARGET_COLUMN]
    return X, y


def run_pipeline(path: str) -> pd.DataFrame:
    """Convenience function chaining the full preprocessing pipeline."""
    df = load_data(path)
    df = cast_types(df)
    df = handle_missing_values(df)
    df = engineer_features(df)
    return df


if __name__ == "__main__":
    processed = run_pipeline("data/student_data.csv")
    print(f"Processed dataset shape: {processed.shape}")
    print(f"Missing values remaining: {int(processed.isna().sum().sum())}")
    print(processed.head())