Make the course yours.

import io
import pandas as pd
import numpy as np

# ---- OPTION A: follow along with the example (just run this cell) ----------
# One row = one student. We will predict whether they passed.
example_url = (
    "https://raw.githubusercontent.com/Sreenivas-Sadhu-Prabhakara/"
    "learn-ml-models/main/data/study_habits.csv"
)
df = pd.read_csv(example_url)

# ---- OPTION B: use YOUR OWN file instead -----------------------------------
# 1) Put a "#" in front of the "df = pd.read_csv(example_url)" line above.
# 2) Delete the "#" in front of the four lines below.
# 3) Run the cell and pick your .csv or .xlsx file when Colab asks.
# from google.colab import files
# uploaded = files.upload()
# name = next(iter(uploaded))
# df = pd.read_csv(io.BytesIO(uploaded[name])) if name.lower().endswith(".csv") else pd.read_excel(io.BytesIO(uploaded[name]))

print(f"Rows: {len(df):,}")
print(f"Columns: {list(df.columns)}")
df.head()

# The values below already work for the EXAMPLE dataset, so you can run this
# cell as-is. When you switch to your own file, change the lines marked  <--

# In plain words, describe your project (these are notes for you):
ROW_MEANING = "one student and how they studied"              # <-- what is one row?
PREDICTION_GOAL = "whether the student passed"                # <-- what do you want to predict?
COSTLY_MISTAKE = "telling an at-risk student they will pass"  # <-- which mistake hurts most?

# Is your answer a category or a number?
TASK = "classification"       # <-- "classification" for a category, "regression" for a number

# Which column is the answer, and which columns are the clues?
TARGET = "passed"             # <-- the column you want to predict
FEATURES = [                  # <-- the columns used as clues (never include TARGET)
    "hours_studied",
    "attendance_percent",
    "previous_score",
    "sleep_hours",
    "participated",
]

# How should new data arrive? Keep "random" unless you have a reason to change it.
SPLIT_MODE = "random"         # "random", "time", or "group"
TIME_COLUMN = None             # only for a time split, e.g. "event_date"
GROUP_COLUMN = None            # only for a group split, e.g. "customer_id"
TEST_SIZE = 0.25              # keep 25% of rows for the final test
RANDOM_STATE = 42            # any fixed number, so results repeat

if TASK not in {"classification", "regression"}:
    raise ValueError("TASK must be 'classification' or 'regression'.")

if not 0 < TEST_SIZE < 1:
    raise ValueError("TEST_SIZE must be a number between 0 and 1.")

if TARGET not in df.columns:
    raise ValueError(f"TARGET '{TARGET}' is not a column in your file.")

if not FEATURES:
    raise ValueError("Choose at least one feature.")

if len(FEATURES) != len(set(FEATURES)):
    raise ValueError("Each feature should appear only once in FEATURES.")

if TARGET in FEATURES:
    raise ValueError("Remove TARGET from FEATURES. The answer cannot be a clue.")

extra_columns = [
    column for column in (TIME_COLUMN, GROUP_COLUMN)
    if column is not None
]
required_columns = list(dict.fromkeys(FEATURES + [TARGET] + extra_columns))
missing_columns = [column for column in required_columns if column not in df.columns]

if missing_columns:
    raise ValueError(f"These configured columns do not exist: {missing_columns}")

model_data = df[required_columns].dropna(subset=[TARGET]).copy()

if TASK == "regression":
    model_data[TARGET] = pd.to_numeric(model_data[TARGET], errors="raise")
    if not np.isfinite(model_data[TARGET]).all():
        raise ValueError("A regression target cannot contain infinity.")
elif model_data[TARGET].nunique() < 2:
    raise ValueError("Classification needs at least two target classes.")

numeric_input_features = model_data[FEATURES].select_dtypes(
    include="number"
).columns.tolist()
infinite_counts = np.isinf(model_data[numeric_input_features]).sum()
infinite_counts = infinite_counts[infinite_counts > 0]
if not infinite_counts.empty:
    print("Treating these infinite feature values as missing:")
    display(infinite_counts.to_frame("infinite_values"))
    model_data[numeric_input_features] = model_data[
        numeric_input_features
    ].replace([np.inf, -np.inf], np.nan)

print("One row:", ROW_MEANING)
print("Goal:", PREDICTION_GOAL)
print("Costly mistake:", COSTLY_MISTAKE)
print("Usable rows:", len(model_data))

profile = pd.DataFrame({
    "dtype": model_data[FEATURES].dtypes.astype(str),
    "missing": model_data[FEATURES].isna().sum(),
    "unique_values": model_data[FEATURES].nunique(dropna=True),
})
display(profile)

if TASK == "classification":
    target_profile = pd.DataFrame({
        "rows": model_data[TARGET].value_counts(dropna=False),
        "share": model_data[TARGET].value_counts(normalize=True, dropna=False),
    })
    display(target_profile)
else:
    display(model_data[TARGET].describe().to_frame("target"))

possible_identifiers = []
for column in FEATURES:
    name_looks_like_id = "id" in column.lower().replace("-", "_").split("_")
    mostly_unique_text = (
        not pd.api.types.is_numeric_dtype(model_data[column])
        and model_data[column].nunique(dropna=True) > 0.5 * len(model_data)
    )
    if name_looks_like_id or mostly_unique_text:
        possible_identifiers.append(column)

if possible_identifiers:
    print("Review these possible identifier/high-cardinality columns:", possible_identifiers)

from sklearn.model_selection import train_test_split, GroupShuffleSplit

if SPLIT_MODE == "random":
    stratify = None
    if TASK == "classification":
        class_counts = model_data[TARGET].value_counts()
        if class_counts.min() >= 2:
            stratify = model_data[TARGET]

    train_rows, test_rows = train_test_split(
        model_data,
        test_size=TEST_SIZE,
        random_state=RANDOM_STATE,
        stratify=stratify,
    )

elif SPLIT_MODE == "time":
    if not TIME_COLUMN:
        raise ValueError("Set TIME_COLUMN before using a time split.")

    if model_data[TIME_COLUMN].isna().any():
        raise ValueError(
            f"TIME_COLUMN '{TIME_COLUMN}' contains missing values. "
            "Fill or remove them before splitting."
        )

    model_data[TIME_COLUMN] = pd.to_datetime(
        model_data[TIME_COLUMN], errors="raise"
    )
    ordered = model_data.sort_values(TIME_COLUMN)
    split_at = int(len(ordered) * (1 - TEST_SIZE))

    if split_at < 1 or split_at >= len(ordered):
        raise ValueError("TEST_SIZE leaves an empty train or test set.")

    # Keep identical timestamps on the same side of the boundary.
    test_start_time = ordered.iloc[split_at][TIME_COLUMN]
    train_rows = ordered[ordered[TIME_COLUMN] < test_start_time]
    test_rows = ordered[ordered[TIME_COLUMN] >= test_start_time]

    if train_rows.empty or test_rows.empty:
        raise ValueError(
            "The time boundary leaves an empty train or test set. "
            "Choose a different TEST_SIZE or review the timestamp values."
        )

elif SPLIT_MODE == "group":
    if not GROUP_COLUMN:
        raise ValueError("Set GROUP_COLUMN before using a group split.")

    if model_data[GROUP_COLUMN].isna().any():
        raise ValueError(
            f"GROUP_COLUMN '{GROUP_COLUMN}' contains missing values. "
            "Fill or remove them before splitting."
        )

    if model_data[GROUP_COLUMN].nunique() < 2:
        raise ValueError("A group split needs at least two distinct groups.")

    splitter = GroupShuffleSplit(
        n_splits=1,
        test_size=TEST_SIZE,
        random_state=RANDOM_STATE,
    )
    train_index, test_index = next(
        splitter.split(model_data, groups=model_data[GROUP_COLUMN])
    )
    train_rows = model_data.iloc[train_index]
    test_rows = model_data.iloc[test_index]

else:
    raise ValueError("SPLIT_MODE must be 'random', 'time', or 'group'.")

X_train = train_rows[FEATURES].copy()
X_test = test_rows[FEATURES].copy()
y_train = train_rows[TARGET].copy()
y_test = test_rows[TARGET].copy()

if TASK == "classification" and y_train.nunique() < 2:
    raise ValueError(
        "The training split contains fewer than two classes. "
        "Use more data or choose a different valid split."
    )

print("Training rows:", len(X_train))
print("Testing rows: ", len(X_test))
print("Split mode:   ", SPLIT_MODE)

if SPLIT_MODE == "time":
    print("Train ends:", train_rows[TIME_COLUMN].max())
    print("Test starts:", test_rows[TIME_COLUMN].min())
elif SPLIT_MODE == "group":
    overlap = set(train_rows[GROUP_COLUMN]) & set(test_rows[GROUP_COLUMN])
    print("Groups shared across train and test:", len(overlap))

from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import OneHotEncoder, StandardScaler

numeric_features = X_train.select_dtypes(include="number").columns.tolist()
categorical_features = [
    column for column in FEATURES
    if column not in numeric_features
]

print("Numeric features:", numeric_features)
print("Categorical features:", categorical_features)

high_cardinality = [
    column for column in categorical_features
    if X_train[column].nunique(dropna=True) > 100
]
if high_cardinality:
    print("Review high-cardinality features before continuing:", high_cardinality)

def make_encoder():
    try:
        return OneHotEncoder(handle_unknown="ignore", sparse_output=False)
    except TypeError:  # Compatibility with older scikit-learn versions
        return OneHotEncoder(handle_unknown="ignore", sparse=False)

def make_preprocessor(scale_numeric=False):
    transformers = []

    if numeric_features:
        numeric_steps = [
            ("fill_missing", SimpleImputer(strategy="median")),
        ]
        if scale_numeric:
            numeric_steps.append(("scale", StandardScaler()))

        transformers.append((
            "numeric",
            Pipeline(numeric_steps),
            numeric_features,
        ))

    if categorical_features:
        categorical_steps = Pipeline([
            ("fill_missing", SimpleImputer(strategy="most_frequent")),
            ("encode", make_encoder()),
        ])
        transformers.append((
            "categorical",
            categorical_steps,
            categorical_features,
        ))

    return ColumnTransformer(
        transformers,
        remainder="drop",
        verbose_feature_names_out=False,
    )

import matplotlib.pyplot as plt
from sklearn.dummy import DummyClassifier, DummyRegressor
from sklearn.metrics import (
    accuracy_score,
    balanced_accuracy_score,
    classification_report,
    ConfusionMatrixDisplay,
    f1_score,
    mean_absolute_error,
    mean_squared_error,
    r2_score,
)
from sklearn.model_selection import (
    GroupKFold,
    KFold,
    StratifiedKFold,
    TimeSeriesSplit,
)

results_log = []

# Pass this dictionary into every cross-validation search's fit() call.
# Group labels are needed while folds are constructed, but are never features.
CV_FIT_PARAMS = (
    {"groups": train_rows[GROUP_COLUMN].to_numpy()}
    if SPLIT_MODE == "group"
    else {}
)

def make_cv():
    if SPLIT_MODE == "time":
        folds = min(5, len(y_train) - 1)
        if folds < 2:
            raise ValueError(
                "Time-based CV needs at least three training rows."
            )
        return TimeSeriesSplit(n_splits=folds)

    if SPLIT_MODE == "group":
        groups = train_rows[GROUP_COLUMN]
        folds = min(5, groups.nunique())
        if folds < 2:
            raise ValueError(
                "Group-based CV needs at least two training groups."
            )
        return GroupKFold(n_splits=folds)

    if TASK == "classification":
        folds = min(5, int(y_train.value_counts().min()))
        if folds < 2:
            raise ValueError("Every class needs at least two training rows for CV.")
        return StratifiedKFold(
            n_splits=folds,
            shuffle=True,
            random_state=RANDOM_STATE,
        )

    folds = min(5, len(y_train))
    if folds < 2:
        raise ValueError("Regression needs at least two training rows for CV.")
    return KFold(
        n_splits=folds,
        shuffle=True,
        random_state=RANDOM_STATE,
    )

PRIMARY_SCORING = (
    "balanced_accuracy" if TASK == "classification"
    else "neg_mean_absolute_error"
)

def evaluate_predictions(name, y_true, y_pred, show_details=True):
    if TASK == "classification":
        row = {
            "model": name,
            "accuracy": accuracy_score(y_true, y_pred),
            "balanced_accuracy": balanced_accuracy_score(y_true, y_pred),
            "macro_f1": f1_score(
                y_true, y_pred, average="macro", zero_division=0
            ),
        }
    else:
        row = {
            "model": name,
            "MAE": mean_absolute_error(y_true, y_pred),
            "RMSE": np.sqrt(mean_squared_error(y_true, y_pred)),
            "R2": r2_score(y_true, y_pred),
        }

    results_log[:] = [item for item in results_log if item["model"] != name]
    results_log.append(row)
    display(pd.DataFrame(results_log))

    if show_details and TASK == "classification":
        print(classification_report(y_true, y_pred, zero_division=0))
        ConfusionMatrixDisplay.from_predictions(
            y_true, y_pred, xticks_rotation=45
        )
        plt.show()

    return row

dummy_train = np.zeros((len(y_train), 1))
dummy_test = np.zeros((len(y_test), 1))

if TASK == "classification":
    baseline_model = DummyClassifier(strategy="most_frequent")
else:
    baseline_model = DummyRegressor(strategy="mean")

baseline_model.fit(dummy_train, y_train)
baseline_predictions = baseline_model.predict(dummy_test)
evaluate_predictions("Baseline", y_test, baseline_predictions, show_details=False)