{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# Bab 06 — ML Fundamentals Playground\n",
        "\n",
        "Notebook pendamping Bab 6. Jalankan semua sel untuk melihat split data, baseline, metrics, threshold, dan leakage demo.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "#!/usr/bin/env python3\n",
        "\"\"\"Bab 06 — Machine Learning Fundamentals Playground.\n",
        "\n",
        "Standard library only. Demonstrates:\n",
        "- tabular dataset;\n",
        "- train/validation/test split;\n",
        "- majority baseline;\n",
        "- simple rule-based classifier;\n",
        "- confusion matrix and metrics;\n",
        "- threshold tradeoff;\n",
        "- regression mean baseline;\n",
        "- leakage demo.\n",
        "\"\"\"\n",
        "from __future__ import annotations\n",
        "\n",
        "import math\n",
        "import random\n",
        "from statistics import mean\n",
        "\n",
        "SEED = 42\n",
        "random.seed(SEED)\n",
        "\n",
        "Record = dict[str, float | int | str]\n",
        "\n",
        "\n",
        "def make_student_dataset(n: int = 48) -> list[Record]:\n",
        "    \"\"\"Create a small synthetic dataset for learning support prediction.\"\"\"\n",
        "    rows: list[Record] = []\n",
        "    for student_id in range(1, n + 1):\n",
        "        attendance = round(random.uniform(0.55, 1.0), 2)\n",
        "        task_completion = round(random.uniform(0.35, 1.0), 2)\n",
        "        quiz_score = random.randint(35, 98)\n",
        "        consultation_count = random.randint(0, 5)\n",
        "\n",
        "        risk_score = (\n",
        "            (1 - attendance) * 0.35\n",
        "            + (1 - task_completion) * 0.30\n",
        "            + max(0, 70 - quiz_score) / 70 * 0.25\n",
        "            + max(0, 2 - consultation_count) / 2 * 0.10\n",
        "        )\n",
        "        # Hidden context simulates real-world factors not captured by our simple features:\n",
        "        # family support, illness, device access, motivation, teacher notes, etc.\n",
        "        hidden_context = random.uniform(-0.12, 0.12)\n",
        "        needs_support = int(risk_score + hidden_context >= 0.36)\n",
        "        next_quiz_score = max(0, min(100, int(quiz_score + attendance * 8 + task_completion * 7 - needs_support * 8)))\n",
        "\n",
        "        rows.append(\n",
        "            {\n",
        "                \"student_id\": student_id,\n",
        "                \"attendance\": attendance,\n",
        "                \"task_completion\": task_completion,\n",
        "                \"quiz_score\": quiz_score,\n",
        "                \"consultation_count\": consultation_count,\n",
        "                \"needs_support\": needs_support,\n",
        "                \"next_quiz_score\": next_quiz_score,\n",
        "                # This is intentionally leaky: it is only known after intervention/decision.\n",
        "                \"leaky_after_review_flag\": needs_support,\n",
        "            }\n",
        "        )\n",
        "    return rows\n",
        "\n",
        "\n",
        "def split_data(rows: list[Record], train_ratio: float = 0.6, valid_ratio: float = 0.2) -> tuple[list[Record], list[Record], list[Record]]:\n",
        "    shuffled = rows[:]\n",
        "    random.shuffle(shuffled)\n",
        "    n = len(shuffled)\n",
        "    n_train = int(n * train_ratio)\n",
        "    n_valid = int(n * valid_ratio)\n",
        "    return shuffled[:n_train], shuffled[n_train : n_train + n_valid], shuffled[n_train + n_valid :]\n",
        "\n",
        "\n",
        "def majority_label(rows: list[Record], label: str = \"needs_support\") -> int:\n",
        "    counts = {0: 0, 1: 0}\n",
        "    for row in rows:\n",
        "        counts[int(row[label])] += 1\n",
        "    return 1 if counts[1] >= counts[0] else 0\n",
        "\n",
        "\n",
        "def predict_majority(rows: list[Record], majority: int) -> list[int]:\n",
        "    return [majority for _ in rows]\n",
        "\n",
        "\n",
        "def risk_score(row: Record) -> float:\n",
        "    return (\n",
        "        (1 - float(row[\"attendance\"])) * 0.35\n",
        "        + (1 - float(row[\"task_completion\"])) * 0.30\n",
        "        + max(0, 70 - float(row[\"quiz_score\"])) / 70 * 0.25\n",
        "        + max(0, 2 - float(row[\"consultation_count\"])) / 2 * 0.10\n",
        "    )\n",
        "\n",
        "\n",
        "def predict_rule(rows: list[Record], threshold: float = 0.38) -> list[int]:\n",
        "    return [int(risk_score(row) >= threshold) for row in rows]\n",
        "\n",
        "\n",
        "def predict_leaky(rows: list[Record]) -> list[int]:\n",
        "    return [int(row[\"leaky_after_review_flag\"]) for row in rows]\n",
        "\n",
        "\n",
        "def confusion_matrix(y_true: list[int], y_pred: list[int]) -> dict[str, int]:\n",
        "    if len(y_true) != len(y_pred):\n",
        "        raise ValueError(\"y_true dan y_pred harus sama panjang\")\n",
        "    tp = sum(1 for y, p in zip(y_true, y_pred) if y == 1 and p == 1)\n",
        "    fp = sum(1 for y, p in zip(y_true, y_pred) if y == 0 and p == 1)\n",
        "    tn = sum(1 for y, p in zip(y_true, y_pred) if y == 0 and p == 0)\n",
        "    fn = sum(1 for y, p in zip(y_true, y_pred) if y == 1 and p == 0)\n",
        "    return {\"TP\": tp, \"FP\": fp, \"TN\": tn, \"FN\": fn}\n",
        "\n",
        "\n",
        "def safe_div(num: float, den: float) -> float:\n",
        "    return 0.0 if den == 0 else num / den\n",
        "\n",
        "\n",
        "def classification_metrics(y_true: list[int], y_pred: list[int]) -> dict[str, float | int]:\n",
        "    cm = confusion_matrix(y_true, y_pred)\n",
        "    tp, fp, tn, fn = cm[\"TP\"], cm[\"FP\"], cm[\"TN\"], cm[\"FN\"]\n",
        "    accuracy = safe_div(tp + tn, tp + fp + tn + fn)\n",
        "    precision = safe_div(tp, tp + fp)\n",
        "    recall = safe_div(tp, tp + fn)\n",
        "    f1 = safe_div(2 * precision * recall, precision + recall)\n",
        "    return {**cm, \"accuracy\": accuracy, \"precision\": precision, \"recall\": recall, \"f1\": f1}\n",
        "\n",
        "\n",
        "def print_metrics(name: str, rows: list[Record], preds: list[int]) -> None:\n",
        "    y_true = [int(row[\"needs_support\"]) for row in rows]\n",
        "    metrics = classification_metrics(y_true, preds)\n",
        "    print(f\"\\n{name}\")\n",
        "    print(\"-\" * len(name))\n",
        "    for key in [\"TP\", \"FP\", \"TN\", \"FN\", \"accuracy\", \"precision\", \"recall\", \"f1\"]:\n",
        "        value = metrics[key]\n",
        "        if isinstance(value, float):\n",
        "            print(f\"{key:>9}: {value:.3f}\")\n",
        "        else:\n",
        "            print(f\"{key:>9}: {value}\")\n",
        "\n",
        "\n",
        "def mae(predictions: list[float], actuals: list[float]) -> float:\n",
        "    if len(predictions) != len(actuals):\n",
        "        raise ValueError(\"predictions dan actuals harus sama panjang\")\n",
        "    return mean([abs(p - a) for p, a in zip(predictions, actuals)])\n",
        "\n",
        "\n",
        "def regression_mean_baseline(train: list[Record], target: str = \"next_quiz_score\") -> float:\n",
        "    return mean([float(row[target]) for row in train])\n",
        "\n",
        "\n",
        "def show_error_examples(rows: list[Record], preds: list[int], max_items: int = 3) -> None:\n",
        "    print(\"\\nContoh error analysis\")\n",
        "    print(\"--------------------\")\n",
        "    shown = 0\n",
        "    for row, pred in zip(rows, preds):\n",
        "        true = int(row[\"needs_support\"])\n",
        "        if true != pred:\n",
        "            print(\n",
        "                f\"student={row['student_id']} true={true} pred={pred} \"\n",
        "                f\"attendance={row['attendance']} task={row['task_completion']} quiz={row['quiz_score']} risk={risk_score(row):.3f}\"\n",
        "            )\n",
        "            shown += 1\n",
        "            if shown >= max_items:\n",
        "                break\n",
        "    if shown == 0:\n",
        "        print(\"Tidak ada error pada subset ini. Curiga jika ini terjadi karena fitur bocor.\")\n",
        "\n",
        "\n",
        "def main() -> None:\n",
        "    print(\"Bab 06 — Machine Learning Fundamentals Playground\")\n",
        "    print(\"=\" * 72)\n",
        "    print(f\"seed: {SEED}\")\n",
        "\n",
        "    rows = make_student_dataset()\n",
        "    train, valid, test = split_data(rows)\n",
        "    print(f\"jumlah data: total={len(rows)} train={len(train)} valid={len(valid)} test={len(test)}\")\n",
        "\n",
        "    majority = majority_label(train)\n",
        "    print(f\"majority label di train: {majority}\")\n",
        "\n",
        "    print_metrics(\"Validation — majority baseline\", valid, predict_majority(valid, majority))\n",
        "\n",
        "    for threshold in [0.45, 0.38, 0.30]:\n",
        "        print_metrics(f\"Validation — rule model threshold={threshold}\", valid, predict_rule(valid, threshold))\n",
        "\n",
        "    best_threshold = 0.38\n",
        "    print_metrics(\"Test final — rule model\", test, predict_rule(test, best_threshold))\n",
        "\n",
        "    print_metrics(\"Leakage demo — jangan ditiru\", test, predict_leaky(test))\n",
        "\n",
        "    baseline_value = regression_mean_baseline(train)\n",
        "    actual_next = [float(row[\"next_quiz_score\"]) for row in test]\n",
        "    pred_next = [baseline_value for _ in test]\n",
        "    print(\"\\nRegression mean baseline\")\n",
        "    print(\"------------------------\")\n",
        "    print(f\"prediksi konstan: {baseline_value:.2f}\")\n",
        "    print(f\"MAE test: {mae(pred_next, actual_next):.3f}\")\n",
        "\n",
        "    show_error_examples(test, predict_rule(test, best_threshold))\n",
        "\n",
        "    print(\"\\nCatatan: leakage demo terlihat bagus karena memakai fitur yang sebenarnya tidak boleh tersedia saat prediksi.\")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 1. Buat dataset dan split\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "rows = make_student_dataset()\n",
        "train, valid, test = split_data(rows)\n",
        "print(len(rows), len(train), len(valid), len(test))\n",
        "print(rows[0])\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 2. Majority baseline\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "majority = majority_label(train)\n",
        "print(\"majority\", majority)\n",
        "print_metrics(\"Validation — majority baseline\", valid, predict_majority(valid, majority))\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 3. Rule model dan threshold\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "for threshold in [0.45, 0.38, 0.30]:\n",
        "    print_metrics(f\"Validation — threshold={threshold}\", valid, predict_rule(valid, threshold))\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 4. Test final\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "best_threshold = 0.38\n",
        "print_metrics(\"Test final — rule model\", test, predict_rule(test, best_threshold))\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 5. Leakage demo — jangan ditiru\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "print_metrics(\"Leakage demo\", test, predict_leaky(test))\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 6. Regression mean baseline\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "baseline_value = regression_mean_baseline(train)\n",
        "actual_next = [float(row[\"next_quiz_score\"]) for row in test]\n",
        "pred_next = [baseline_value for _ in test]\n",
        "print(\"prediksi konstan\", baseline_value)\n",
        "print(\"MAE\", mae(pred_next, actual_next))\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 7. Error analysis\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {},
      "outputs": [],
      "source": [
        "show_error_examples(test, predict_rule(test, best_threshold))\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## 8. Challenge\n",
        "\n",
        "Ubah threshold, jumlah data, atau definisi risk score. Catat precision, recall, F1, dan contoh error.\n"
      ]
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "name": "python",
      "pygments_lexer": "ipython3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}