Module.py

"""
Module.py

Defines a modular, fully connected neural network model for binary classification.

Features:
- Flexible architecture construction via sequential Layer stacking
- ReLU, Sigmoid, and Tanh activation support
- Dropout regularization during training
- L2 weight regularization (Ridge)
- Optimization via Adam or standard SGD
- Full forward and backward propagation
- Configurable training with hyperparameter control

Author: Ulaş Sertan Kemeç
Date: 2025-06-19
"""


import Layer
import numpy as np


def make_mini_batches(X, Y, batch_size, shuffle=True):
    m = X.shape[1]
    # Optionally shuffle in unison:
    if shuffle:
        idx = np.random.permutation(m)
        X, Y = X[:, idx], Y[:, idx]

    mini_batches = []
    num_batches = int(np.ceil(m / batch_size))
    for i in range(num_batches):
        start = i * batch_size
        end = start + batch_size
        x_batch = X[:, start:end]
        y_batch = Y[:, start:end]
        mini_batches.append((x_batch, y_batch))
    return mini_batches

class Module:
    def __init__(self, dimensions, activations, lambd=0.0, keep_prob=1.0, learning_rate=0.001, batch_norm=True):
        """
        dimensions: List[int] - number of units per layer (including input/output)
        activations: List[str] - activation functions for each hidden/output layer
        """
        self.layers = []
        self.learning_rate = learning_rate
        self.batch_norm = batch_norm

        if len(activations) != len(dimensions) - 1:
            raise ValueError("Length of activations must be one less than length of dimensions")

        for i in range(1, len(dimensions)):
            layer = Layer.Layer(
                input_dim=dimensions[i - 1],
                units=dimensions[i],
                activation=activations[i - 1],
                lambd=lambd,
                keep_prob=keep_prob,
                learning_rate=learning_rate
            )
            self.layers.append(layer)


    def forward(self, X, dropout=False, training=True, batch_norm=None):
        """
        Perform forward propagation through all layers.

        Parameters:
        - X (ndarray): Input data of shape (input_dim, number_of_examples)
        - dropout (bool): Whether to apply dropout during forward pass
        - batch_norm : Optional[bool]  # override self.batch_norm if not None

        Returns:
        - X (ndarray): Final output (Ŷ) from the network
        """
        use_bn = self.batch_norm if batch_norm is None else batch_norm
        for i, layer in enumerate(self.layers):
            # Output of one layer becomes input to the next
            bn_this_layer = use_bn and (i < len(self.layers) - 1)
            X = layer.forward_propagation(X, dropout, batch_norm=bn_this_layer, training=training)
        return X

    def backward(self, Y, regularization=None, dropout=False):
        """
        Perform backward propagation through all layers.

        Parameters:
        - Y (ndarray): True labels (same shape as output)
        - regularization (str or None): Regularization method (e.g., "L2")
        - dropout (bool): Whether dropout was used in forward pass
        """
        # Get final prediction from the last layer
        Yhat = np.clip(self.layers[-1].A, 1e-10, 1 - 1e-10)

        # Compute dA from loss derivative (binary cross-entropy)
        dA = np.divide(-Y, Yhat) + np.divide(1 - Y, 1 - Yhat)

        # Propagate gradients backward through each layer
        for layer in reversed(self.layers):
            layer.back_propagation(dA, regularization, dropout)
            dA = layer.dA_previous  # Gradient for next layer

    def update_parameters(self, t, beta_values=(0.9, 0.999, 1e-8), optimization=None):
        """
        Update parameters (weights and biases) for all layers.

        Parameters:
        - t (int): Epoch or iteration counter (used in Adam for bias correction)
        - beta_values (tuple): Beta1, Beta2, and epsilon for Adam optimizer
        - optimization (str or None): Optimization algorithm ("Adam" or None for SGD)
        """
        for layer in self.layers:
            layer.update_parameters(t, beta_values, optimization)

    def compute_cost(self, Y, Yhat, regularization=None):
        """
        Compute binary cross-entropy cost, with optional L2 regularization.

        Parameters:
            Y (ndarray): Ground truth labels, shape (1, m)
            Yhat (ndarray): Predicted probabilities, shape (1, m)
            regularization (str or None): Type of regularization ("L2" or None)

        Returns:
            float: Computed total cost
        """
        m = Y.shape[1]
        Yhat = np.clip(Yhat, 1e-10, 1 - 1e-10)

        # Binary cross-entropy loss
        cost = -np.sum(Y * np.log(Yhat)) - np.sum((1 - Y) * np.log(1 - Yhat))
        cost /= m

        # L2 regularization (if specified)
        if regularization == "L2":
            cost_regularization = 0
            for layer in self.layers:
                cost_regularization += np.sum(np.square(layer.W)) * (layer.lambd / (2 * m))
            cost += cost_regularization

        return cost

    def predict(self, X, dropout=False, training=False):
        """
        Predict binary class labels for given input.

        Parameters:
            X (ndarray): Input data of shape (n_features, m_samples)
            dropout (bool): Whether to apply dropout (should be False during inference)

        Returns:
            ndarray: Binary predictions (0 or 1) of shape (1, m_samples)
        """
        Yhat = self.forward(X, dropout, training=training, batch_norm=self.batch_norm)
        prediction = (Yhat >= 0.5).astype(int)
        return prediction

    def train(self, X, Y, epochs,mini_batch_size=None, regularization=None, dropout=False,
              beta_values=(0.9, 0.999, 1e-8), optimization=None, verbose=False, interval = 10):
        """
        Train the model using forward, backward, and parameter update steps.

        Parameters:
            X (ndarray): Input data
            Y (ndarray): Ground truth labels
            epochs (int): Number of training iterations
            regularization (str or None): Regularization type (e.g., "L2")
            dropout (bool): Whether to apply dropout during training
            beta_values (tuple): Adam optimizer parameters
            optimization (str or None): "Adam" or None for SGD
            verbose (bool): Whether to print cost in every 200 epochs
        """

        m_batch = Y.shape[1]
        step = 0
        for epoch in range(1, 1 + epochs):
            if mini_batch_size is None:
                batches = [(X, Y)]
            else:
                batches = make_mini_batches(X, Y, mini_batch_size, shuffle=True)

            epoch_cost_sum, n_seen = 0.0, 0
            for batch_id, (x, y) in enumerate(batches):
                step += 1
                Yhat = self.forward(x, dropout, training=True, batch_norm=self.batch_norm)
                cost = self.compute_cost(y, Yhat, regularization)
                m_batch = y.shape[0] if y.ndim == 2 and y.shape[0] != 1 else y.shape[-1]
                epoch_cost_sum += cost * m_batch
                n_seen += m_batch
                self.backward(y, regularization, dropout)
                self.update_parameters(step, beta_values, optimization)

            epoch_cost = epoch_cost_sum / n_seen


            # Print cost every 200 epochs
            if (epoch % 200 == 0 or epoch == epochs - 1 ) and verbose:
                print(f"Epoch {epoch}: Cost -> {epoch_cost:.6f}")

            decay_interval = epochs // interval
            if decay_interval > 0 and epoch > 0 and epoch % decay_interval == 0:
                for layer in self.layers:
                    layer.learning_rate *= 0.5