{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.12.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceType":"competition","sourceId":3004,"databundleVersionId":861823}],"dockerImageVersionId":31286,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import torch\nimport matplotlib.pyplot as plt\nimport numpy as np\nimport math\nfrom torch.utils.data import DataLoader","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2026-03-14T10:04:16.843336Z","iopub.execute_input":"2026-03-14T10:04:16.843925Z","iopub.status.idle":"2026-03-14T10:04:16.849191Z","shell.execute_reply.started":"2026-03-14T10:04:16.843890Z","shell.execute_reply":"2026-03-14T10:04:16.847996Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def zero_padding(input_img, pad = 0):\n    padded_img = np.pad(input_img, ((0,0), (pad, pad), (pad, pad), (0,0)))\n    tensor = torch.from_numpy(padded_img)\n    return tensor","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-14T10:04:21.500204Z","iopub.execute_input":"2026-03-14T10:04:21.500947Z","iopub.status.idle":"2026-03-14T10:04:21.505892Z","shell.execute_reply.started":"2026-03-14T10:04:21.500912Z","shell.execute_reply":"2026-03-14T10:04:21.504915Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class Conv2D:\n    def __init__(self, n_filters, kernel_size, stride = None, pad = 0, pad_mode = 'pad'):\n        # pad_mode = ['same', 'valid', 'pad']\n        # default 'pad'\n        # same: auto calculate 'pad' parameter to keep the size of img same as input\n        # valid: no padding\n        # pad: use 'pad' hyperparameter\n\n        \n        self.n_filters = n_filters\n        self.kernel_size = kernel_size\n        self.optimizer = None\n        self.init_params = False\n        self.forward_cache = None\n\n        if pad_mode == 'valid':\n            pad = 0\n        elif pad_mode == 'pad':\n            pass\n        elif pad_mode == 'same':\n            assert stride == 1, \"'same' padding requires stride = 1\"\n            pad = (kernel_size - 1) // 2\n        else:\n            raise ValueError(\"pad_mode must be 'same', 'valid', or 'pad'\")\n        stride = stride if stride else kernel_size\n            \n        self.hparams = {'pad' : pad, 'stride' : stride}\n        \n    def relu(self, tensor):\n        cache = tensor\n        a = tensor * (tensor > 0).float()\n        return a, cache\n\n    def relu_grad(self, dA, relu_cache):\n        mask = (relu_cache > 0).float()\n        dZ = dA * mask\n        return dZ\n        \n    def single_conv(self, a_prev, W, b):\n        assert a_prev.size() == W.size(), f'Expected A and W has same size, got A:{a_prev.size()}, W:{W.size()}'\n        z = a_prev * W\n        a = torch.sum(z) + b\n        return a\n\n    def forward(self, A_prev):\n        # A_prev size: (m, input_height, input_width, channel)\n        # height_output = (input_height + 2 * pad - kernel_size) // 2 + 1\n        # width_output = (input_width + 2 * pad - kernel_size) // 2 + 1\n        # A_out: (m, height_output, width_output, n_filters)\n        \n        relu_caches = []\n        (m, n_h, n_w, n_C_prev) = A_prev.size()\n        if not self.init_params:\n            self.W = torch.randn(size = (self.kernel_size, self.kernel_size, n_C_prev, self.n_filters))\n            self.b = torch.randn(size = (1, 1, 1, self.n_filters))\n            self.init_params = True\n\n        pad = self.hparams['pad']\n        stride = self.hparams['stride']\n        f = self.kernel_size\n        new_h = (n_h + 2 * pad - f) // stride + 1\n        new_w = (n_w + 2 * pad - f) // stride + 1\n        new_C = self.n_filters\n        output_size = (m, new_h, new_w, new_C)\n        Z = torch.zeros(size = output_size)\n        A_out = torch.zeros(size = output_size)\n        A_prev_pad = zero_padding(A_prev, pad)\n        \n        for i in range(m):\n            for v in range(new_h):\n                v_start = v * stride\n                v_end = v * stride + f\n                \n                for h in range(new_w):\n                    h_start = h * stride\n                    h_end = h * stride + f\n                    \n                    for c in range(new_C):\n                        a_slice_prev = A_prev_pad[i, v_start : v_end, h_start : h_end, :]   \n                        Z[i, v, h, c] = self.single_conv(a_slice_prev, self.W[:, :, :, c], self.b[0, 0, 0, c])\n            A_out[i], cache = self.relu(Z[i])\n            relu_caches.append(cache)\n\n        self.forward_cache = (A_prev, relu_caches)\n        return A_out\n\n    def backward(self, dA):\n        # dA size: (m, output_height, output_width, n_filters)\n\n        assert self.init_params == True, 'Parameters are not initialized yet.'\n        W, b = self.W, self.b\n        pad = self.hparams['pad']\n        stride = self.hparams['stride']\n        f = self.kernel_size\n        \n        (A_prev, relu_caches) = self.forward_cache\n        (m, n_h_prev, n_w_prev, n_C_prev) = A_prev.size()\n        (m, n_H, n_W, n_C) = dA.size()\n        dZ_size = dA.size()\n        dW_size = W.size()\n        db_size = b.size()\n\n        dZ = torch.zeros(size = dZ_size)\n        dW = torch.zeros(size = dW_size)\n        db = torch.zeros(size = db_size)\n        dA_prev = torch.zeros(size = A_prev.size())\n        dA_prev_pad = zero_padding(dA_prev, pad)\n        A_prev_pad = zero_padding(A_prev, pad)\n\n        for i in range(m):\n            dZ[i] = self.relu_grad(dA[i], relu_caches[i])\n            a_prev_pad = A_prev_pad[i]\n            da_prev_pad = dA_prev_pad[i]\n            # Loop through vertical axis\n            for v in range(n_H):\n                v_start = v * stride\n                v_end = v * stride + f\n                # Loop through horizontal axis\n                for h in range(n_W):\n                    h_start = h * stride\n                    h_end = h * stride + f\n                    # Loop through channels\n                    for c in range(n_C):\n                        a_slice = a_prev_pad[v_start : v_end, h_start : h_end, :]\n                        da_prev_pad[v_start : v_end, h_start : h_end, :] += W[:, :, :, c] * dZ[i, v, h, c]\n                        dW[:, :, :, c] += a_slice * dZ[i, v, h, c]\n                        db[0, 0, 0, c] += dZ[i, v, h, c]\n                # Update dA_prev \n                if pad > 0:\n                    dA_prev[i] = da_prev_pad[pad : -pad, pad : -pad, :]\n                else: \n                    dA_prev[i] = da_prev_pad\n                    \n        self.dW = dW / m\n        self.db = db / m\n        self.update_params()\n        return dA_prev\n\n    def update_params(self):\n        assert self.optimizer is not None, 'Optimizer is None.'\n        assert self.dW.size() == self.W.size(), 'dW and W must have same size.'\n        assert self.db.size() == self.b.size(), 'db and b must have same size.'\n\n        opt = self.optimizer\n        self.W = opt.optimize(self.W, self.dW)\n        self.b = opt.optimize(self.b, self.db)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-14T10:04:22.709940Z","iopub.execute_input":"2026-03-14T10:04:22.710467Z","iopub.status.idle":"2026-03-14T10:04:22.732593Z","shell.execute_reply.started":"2026-03-14T10:04:22.710434Z","shell.execute_reply":"2026-03-14T10:04:22.731514Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class MaxPooling2D:\n    def __init__(self, kernel_size, stride = None):\n        self.kernel_size = kernel_size\n        self.stride = stride if stride else kernel_size\n        self.forward_cache = None\n\n    def forward(self, A_prev):\n        f = self.kernel_size\n        stride = self.stride\n        (m, n_h, n_w, n_C) = A_prev.size()\n        new_h = (n_h - f) // stride + 1\n        new_w = (n_w - f) // stride + 1\n        output_size = (m, new_h, new_w, n_C)\n        A_out = torch.zeros(size = output_size)\n        self.forward_cache = A_prev\n\n        for i in range(m):\n            for v in range(new_h):\n                v_start = v * stride\n                v_end = v * stride + f\n                for h in range(new_w):\n                    h_start = h * stride\n                    h_end = h * stride + f\n                    for c in range(n_C):\n                        a_slice = A_prev[i, v_start : v_end, h_start : h_end, c]\n                        A_out[i, v, h, c] = torch.max(a_slice)\n        return A_out\n\n    def create_mask(self, a_slice):\n        return (a_slice == torch.max(a_slice)).float()\n\n    def backward(self, dA):\n        f = self.kernel_size\n        stride = self.stride\n        (m, new_h, new_w, n_C) = dA.size()\n        A_prev = self.forward_cache\n        dA_prev = torch.zeros(size = A_prev.size())\n        \n        for i in range(m):\n            for v in range(new_h):\n                v_start = v * stride\n                v_end = v * stride + f\n                for h in range(new_w):\n                    h_start = h * stride\n                    h_end = h * stride + f\n                    for c in range(n_C):\n                        A_prev_slice = A_prev[i, v_start : v_end, h_start : h_end, c]\n                        dA_slice = dA[i, v, h, c]\n                        dA_prev[i, v_start : v_end, h_start : h_end, c] += dA_slice * self.create_mask(A_prev_slice)\n        return dA_prev","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-14T10:04:38.916052Z","iopub.execute_input":"2026-03-14T10:04:38.916459Z","iopub.status.idle":"2026-03-14T10:04:38.927651Z","shell.execute_reply.started":"2026-03-14T10:04:38.916429Z","shell.execute_reply":"2026-03-14T10:04:38.926865Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class Flatten:\n    def __init__(self):\n        self.forward_cache = None\n        \n    def forward(self, A_prev):\n        self.forward_cache = A_prev.size()\n        return torch.reshape(A_prev, shape = (A_prev.size()[0], -1))\n\n    def backward(self, dA):\n        A_prev_size = self.forward_cache\n        return torch.reshape(dA, shape = A_prev_size)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-14T10:04:45.573175Z","iopub.execute_input":"2026-03-14T10:04:45.573699Z","iopub.status.idle":"2026-03-14T10:04:45.579385Z","shell.execute_reply.started":"2026-03-14T10:04:45.573668Z","shell.execute_reply":"2026-03-14T10:04:45.578362Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class Adam:\n    def __init__(self, lr=0.001, beta1=0.9, beta2=0.999, epsilon=1e-8):\n        self.lr = lr\n        self.beta1 = beta1\n        self.beta2 = beta2\n        self.epsilon = epsilon\n        \n        # Dictionaries to hold moving averages, keyed by tensor memory ID\n        self.v = {}\n        self.s = {}\n        self.t = {}\n\n    def optimize(self, param, grad):\n        # Unique identifier for the specific W or b tensor\n        pid = id(param)\n        \n        if pid not in self.v:\n            self.v[pid] = torch.zeros_like(param)\n            self.s[pid] = torch.zeros_like(param)\n            self.t[pid] = 0\n            \n        self.t[pid] += 1\n        t = self.t[pid]\n        self.v[pid] = self.beta1 * self.v[pid] + (1 - self.beta1) * grad\n        self.s[pid] = self.beta2 * self.s[pid] + (1 - self.beta2) * (grad ** 2)\n        v_corrected = self.v[pid] / (1 - self.beta1 ** t)\n        s_corrected = self.s[pid] / (1 - self.beta2 ** t)\n        # Using .data ensures the memory ID (pid) does not change!\n        param.data -= self.lr * v_corrected / (torch.sqrt(s_corrected) + self.epsilon)\n        \n        return param","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-14T10:05:14.452193Z","iopub.execute_input":"2026-03-14T10:05:14.452590Z","iopub.status.idle":"2026-03-14T10:05:14.461041Z","shell.execute_reply.started":"2026-03-14T10:05:14.452562Z","shell.execute_reply":"2026-03-14T10:05:14.459778Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class Dense:\n    def __init__(self, units, activation = 'relu'):\n        self.n_neurons = units\n        self.activation = activation\n        self.optimizer = None\n        self.init_params = False\n        self.forward_cache = None\n\n    def relu(self, tensor):\n        cache = tensor\n        a = tensor * (tensor > 0).float()\n        return a, cache\n\n    def relu_grad(self, dA, relu_cache):\n        mask = (relu_cache > 0).float()\n        dZ = dA * mask\n        return dZ\n\n    def softmax(self, tensor):\n        cache = tensor\n        # Shift values down for numerical stability\n        shift_tensor = tensor - torch.max(tensor, dim=1, keepdim=True)[0]\n        exps = torch.exp(shift_tensor)\n        a = exps / torch.sum(exps, dim=1, keepdim=True)\n        return a, cache\n\n    def forward(self, A_prev):\n        # A_prev size: (m, n_in)\n        (m, n_in) = A_prev.size()\n        \n        if not self.init_params:\n            # Multiplied by 0.01 to prevent exploding gradients on initialization\n            self.W = torch.randn(size=(n_in, self.n_neurons)) * 0.01\n            self.b = torch.zeros(size=(1, self.n_neurons))\n            self.init_params = True\n\n        # Z = A_prev * W + b\n        Z = torch.matmul(A_prev, self.W) + self.b\n        \n        # Apply activation\n        if self.activation == 'softmax':\n            A_out, cache = self.softmax(Z)\n        else:\n            A_out, cache = self.relu(Z)\n        self.forward_cache = (A_prev, cache)\n        return A_out\n\n    def backward(self, dA):\n        # dA size: (m, n_neurons)\n        assert self.init_params == True, 'Parameters are not initialized yet.'\n        W, b = self.W, self.b\n        \n        (A_prev, relu_cache) = self.forward_cache\n        m = A_prev.size(0)\n\n        if self.activation == 'softmax':\n            dZ = dA # A - Y\n        else:\n            # Backprop through ReLU\n            dZ = self.relu_grad(dA, relu_cache)\n        \n        # Calculate gradients (averaged over the batch m)\n        self.dW = torch.matmul(A_prev.T, dZ) / m\n        self.db = torch.sum(dZ, dim=0, keepdim=True) / m\n        \n        # Calculate gradient for the previous layer\n        dA_prev = torch.matmul(dZ, self.W.T)\n        \n        self.update_params()\n        return dA_prev\n    \n    def update_params(self):\n        assert self.optimizer is not None, 'Optimizer is None.'\n        assert self.dW.size() == self.W.size(), f'dW and W must have same size, got {self.dW.size()} and {self.W.size()}'\n        assert self.db.size() == self.b.size(), f'db and b must have same size, got {self.db.size()} and {self.b.size()}'\n\n        opt = self.optimizer\n        self.W = opt.optimize(self.W, self.dW)\n        self.b = opt.optimize(self.b, self.db)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-14T10:05:24.279321Z","iopub.execute_input":"2026-03-14T10:05:24.279740Z","iopub.status.idle":"2026-03-14T10:05:24.293018Z","shell.execute_reply.started":"2026-03-14T10:05:24.279711Z","shell.execute_reply":"2026-03-14T10:05:24.291671Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class CNN:\n    def __init__(self, optimizer):\n        self.layers = []\n        self.optimizer = optimizer\n        self.epsilon = 1e-8\n        \n        # History for plotting\n        self.history = {'loss': [], 'accuracy': []}\n\n    def add(self, layer):\n        # Attach the global optimizer to layers that have learnable parameters\n        if hasattr(layer, 'optimizer'):\n            layer.optimizer = self.optimizer\n        self.layers.append(layer)\n\n    def forward(self, X):\n        A = X\n        for layer in self.layers:\n            A = layer.forward(A)\n        return A\n\n    def backward(self, dA):\n        # Traverse layers in reverse order\n        for layer in reversed(self.layers):\n            dA = layer.backward(dA)\n\n    def compute_loss(self, Y_pred, Y_true):\n        # Categorical Cross-Entropy\n        Y_pred_clipped = torch.clamp(Y_pred, self.epsilon, 1.0 - self.epsilon)\n        m = Y_pred.size(0)\n        loss = -torch.sum(Y_true * torch.log(Y_pred_clipped)) / m\n        return loss.item()\n\n    def compute_accuracy(self, Y_pred, Y_true):\n        # Convert one-hot vectors back to class indices\n        predictions = torch.argmax(Y_pred, dim=1)\n        labels = torch.argmax(Y_true, dim=1)\n        correct = torch.sum(predictions == labels).item()\n        return correct / Y_true.size(0)\n\n    def fit(self, X_train, Y_train, epochs, batch_size, verbose=True):\n        m = X_train.size(0)\n        num_batches = m // batch_size\n        \n        for epoch in range(epochs):\n            epoch_loss = 0\n            epoch_acc = 0\n            \n            # Shuffle data for each epoch\n            indices = torch.randperm(m)\n            X_shuffled = X_train[indices]\n            Y_shuffled = Y_train[indices]\n            \n            for b in range(num_batches):\n                start = b * batch_size\n                end = start + batch_size\n                \n                X_batch = X_shuffled[start:end]\n                Y_batch = Y_shuffled[start:end]\n                \n                # 1. Forward Pass\n                Y_pred = self.forward(X_batch)\n                \n                # 2. Compute Metrics\n                batch_loss = self.compute_loss(Y_pred, Y_batch)\n                batch_acc = self.compute_accuracy(Y_pred, Y_batch)\n                \n                epoch_loss += batch_loss\n                epoch_acc += batch_acc\n                \n                # 3. Backward Pass\n                # The shortcut derivative for Softmax + Categorical Cross Entropy\n                dA = Y_pred - Y_batch \n                self.backward(dA)\n                \n            # Average metrics over all batches\n            epoch_loss /= num_batches\n            epoch_acc /= num_batches\n            \n            self.history['loss'].append(epoch_loss)\n            self.history['accuracy'].append(epoch_acc)\n            \n            if verbose:\n                print(f\"Epoch {epoch+1}/{epochs} - Loss: {epoch_loss:.4f} - Accuracy: {epoch_acc:.4f}\")\n\n    def predict(self, X):\n        # Returns raw probabilities\n        return self.forward(X)\n\n    def evaluate(self, X_test, Y_test):\n        Y_pred = self.predict(X_test)\n        loss = self.compute_loss(Y_pred, Y_test)\n        acc = self.compute_accuracy(Y_pred, Y_test)\n        print(f\"Test Loss: {loss:.4f} - Test Accuracy: {acc:.4f}\")\n        return loss, acc\n\n    def plot_history(self):\n        epochs = range(1, len(self.history['loss']) + 1)\n        \n        plt.figure(figsize=(12, 5))\n        \n        # Plot Loss\n        plt.subplot(1, 2, 1)\n        plt.plot(epochs, self.history['loss'], label='Training Loss', color='red')\n        plt.title('Training Loss')\n        plt.xlabel('Epochs')\n        plt.ylabel('Loss')\n        plt.grid(True)\n        \n        # Plot Accuracy\n        plt.subplot(1, 2, 2)\n        plt.plot(epochs, self.history['accuracy'], label='Training Accuracy', color='blue')\n        plt.title('Training Accuracy')\n        plt.xlabel('Epochs')\n        plt.ylabel('Accuracy')\n        plt.grid(True)\n        \n        plt.tight_layout()\n        plt.show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-14T10:06:03.959970Z","iopub.execute_input":"2026-03-14T10:06:03.960457Z","iopub.status.idle":"2026-03-14T10:06:03.976315Z","shell.execute_reply.started":"2026-03-14T10:06:03.960424Z","shell.execute_reply":"2026-03-14T10:06:03.975213Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import pandas as pd\nimport numpy as np\n\ntrain_df = pd.read_csv('/kaggle/input/competitions/digit-recognizer/train.csv')\ntest_df = pd.read_csv('/kaggle/input/competitions/digit-recognizer/test.csv')\n\nY_train_raw = train_df['label'].values\nX_train_raw = train_df.drop('label', axis=1).values\nX_test_raw = test_df.values\n\nX_train = torch.tensor(X_train_raw, dtype=torch.float32).reshape(-1, 28, 28, 1) / 255.0\nX_test = torch.tensor(X_test_raw, dtype=torch.float32).reshape(-1, 28, 28, 1) / 255.0\n\nm = len(Y_train_raw)\nY_train = torch.zeros((m, 10), dtype=torch.float32)\nY_train[torch.arange(m), Y_train_raw] = 1.0\n\nadam_opt = Adam(lr=0.005)\nmodel = CNN(optimizer=adam_opt)\n\nmodel.add(Conv2D(n_filters=8, kernel_size=3, stride=1, pad_mode='same'))\nmodel.add(MaxPooling2D(kernel_size=2, stride=2))\nmodel.add(Flatten())\nmodel.add(Dense(units=64, activation='relu'))\nmodel.add(Dense(units=10, activation='softmax'))\n\nprint(f'Start training...')\nmodel.fit(X_train[ : 3000], Y_train[ : 3000], epochs=5, batch_size=128, verbose=True)\nprint(f'Done.')\n\npredictions = []\nbatch_size = 128\n\nfor i in range(0, len(X_test), batch_size):\n    X_batch = X_test[i : i + batch_size]\n    Y_pred_batch = model.predict(X_batch)\n    batch_preds = torch.argmax(Y_pred_batch, dim=1)\n    predictions.extend(batch_preds.tolist())\n\nsubmission = pd.DataFrame({\n    'ImageId': range(1, len(predictions) + 1),\n    'Label': predictions\n})\n\nsubmission.to_csv('submission.csv', index=False)\nprint('Submit successfully.')","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-14T10:13:06.958384Z","iopub.execute_input":"2026-03-14T10:13:06.959383Z","iopub.status.idle":"2026-03-14T10:13:13.818578Z","shell.execute_reply.started":"2026-03-14T10:13:06.959346Z","shell.execute_reply":"2026-03-14T10:13:13.817402Z"}},"outputs":[],"execution_count":null}]}