From 809f2643476c02367808b877582a68804eccbbec Mon Sep 17 00:00:00 2001 From: Min Htet Myet <88831350+Mattral@users.noreply.github.com> Date: Wed, 3 Jun 2026 07:31:56 +0000 Subject: [PATCH] Implement various supervised and unsupervised learning algorithms from scratch - Added Random Forest Classifier with bootstrap aggregation of decision trees. - Implemented Ridge Regression using closed-form regularized normal equations. - Developed Linear Support Vector Machine Classifier trained with stochastic sub-gradient descent. - Introduced K-Means Clustering algorithm with Lloyd's iteration and K-Means++ initialization. - Created tests for Decision Tree, KNN, Lasso Regression, Linear Regression, Logistic Regression, Naive Bayes, Random Forest, Ridge Regression, SVM, and K-Means. - Added shared fixtures for reproducible test datasets. --- .github/workflows/ci.yml | 40 +++ README.md | 283 +++++++----------- pyproject.toml | 52 ++++ src/mlscratch/__init__.py | 11 + src/mlscratch/supervised/__init__.py | 24 ++ src/mlscratch/supervised/decision_tree.py | 142 +++++++++ src/mlscratch/supervised/knn.py | 77 +++++ src/mlscratch/supervised/lasso_regression.py | 125 ++++++++ src/mlscratch/supervised/linear_regression.py | 197 ++++++++++++ .../supervised/logistic_regression.py | 119 ++++++++ src/mlscratch/supervised/naive_bayes.py | 113 +++++++ src/mlscratch/supervised/random_forest.py | 118 ++++++++ src/mlscratch/supervised/ridge_regression.py | 93 ++++++ src/mlscratch/supervised/svm.py | 117 ++++++++ src/mlscratch/unsupervised/__init__.py | 5 + src/mlscratch/unsupervised/kmeans.py | 135 +++++++++ tests/conftest.py | 50 ++++ tests/supervised/test_decision_tree.py | 51 ++++ tests/supervised/test_knn.py | 60 ++++ tests/supervised/test_lasso_regression.py | 50 ++++ tests/supervised/test_linear_regression.py | 63 ++++ tests/supervised/test_logistic_regression.py | 53 ++++ tests/supervised/test_naive_bayes.py | 39 +++ tests/supervised/test_random_forest.py | 58 ++++ tests/supervised/test_ridge_regression.py | 49 +++ tests/supervised/test_svm.py | 64 ++++ tests/unsupervised/test_kmeans.py | 52 ++++ 27 files changed, 2071 insertions(+), 169 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 pyproject.toml create mode 100644 src/mlscratch/__init__.py create mode 100644 src/mlscratch/supervised/__init__.py create mode 100644 src/mlscratch/supervised/decision_tree.py create mode 100644 src/mlscratch/supervised/knn.py create mode 100644 src/mlscratch/supervised/lasso_regression.py create mode 100644 src/mlscratch/supervised/linear_regression.py create mode 100644 src/mlscratch/supervised/logistic_regression.py create mode 100644 src/mlscratch/supervised/naive_bayes.py create mode 100644 src/mlscratch/supervised/random_forest.py create mode 100644 src/mlscratch/supervised/ridge_regression.py create mode 100644 src/mlscratch/supervised/svm.py create mode 100644 src/mlscratch/unsupervised/__init__.py create mode 100644 src/mlscratch/unsupervised/kmeans.py create mode 100644 tests/conftest.py create mode 100644 tests/supervised/test_decision_tree.py create mode 100644 tests/supervised/test_knn.py create mode 100644 tests/supervised/test_lasso_regression.py create mode 100644 tests/supervised/test_linear_regression.py create mode 100644 tests/supervised/test_logistic_regression.py create mode 100644 tests/supervised/test_naive_bayes.py create mode 100644 tests/supervised/test_random_forest.py create mode 100644 tests/supervised/test_ridge_regression.py create mode 100644 tests/supervised/test_svm.py create mode 100644 tests/unsupervised/test_kmeans.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..d9a77e5 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,40 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + branches: [main] + +jobs: + test: + name: Test Python ${{ matrix.python-version }} + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.10", "3.11", "3.12"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + - run: pip install -e ".[dev]" + - run: pytest tests/ --cov=mlscratch --cov-report=xml --cov-fail-under=88 + - uses: actions/upload-artifact@v4 + with: + name: coverage-${{ matrix.python-version }} + path: coverage.xml + + lint: + name: Lint + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + cache: pip + - run: pip install ruff black + - run: ruff check src/ tests/ + - run: black --check src/ tests/ diff --git a/README.md b/README.md index 72a0dcd..357cce0 100644 --- a/README.md +++ b/README.md @@ -1,214 +1,159 @@ +# ML-AI-Algorithms-from-scratch -# AI, ML, DL, and RL Demystified: From Scratch to Understanding +A structured, educational repository of from-scratch ML/AI/RL/Bayesian algorithms. -Welcome to my comprehensive repository dedicated to unraveling the mysteries of Artificial Intelligence (AI), Machine Learning (ML), Deep Learning (DL), and Reinforcement Learning (RL). 🚀 - -## Purpose & Positioning - -This repository is a **learning-first, from-scratch implementation collection** of core AI, Machine Learning, Deep Learning, Reinforcement Learning, and Bayesian algorithms. - -It is designed for readers who: -- Already know *what* these algorithms are -- Want to understand **how they work internally** -- Prefer readable, step-by-step implementations over optimized or production-grade code - -This is **not** a production library or benchmarking suite. -Instead, the focus is on **algorithmic intuition, mathematical flow, and code transparency**. - - -## Who This Repository Is NOT For - -This repository may not be ideal if you are looking for: -- Highly optimized or GPU-accelerated implementations -- Drop-in replacements for scikit-learn, PyTorch, or TensorFlow -- State-of-the-art performance benchmarks -- Large-scale dataset pipelines - -The goal here is **understanding**, not performance. +This project is evolving from a collection of standalone scripts into a clean, `pip`-installable Python package under `src/mlscratch/`. --- -### How to Navigate This Repository - -If you're new to the repository, a recommended learning path is: - -1. **Supervised Learning** - - Linear & Logistic Regression - - k-Nearest Neighbors - - Decision Trees -2. **Unsupervised Learning** - - K-Means - - PCA - - Gaussian Mixture Models -3. **Neural Networks** - - Single-Layer Perceptron - - Multi-Layer Perceptron - - CNNs and RNNs -4. **Reinforcement Learning** - - Q-Learning - - Deep Q-Networks - - Policy-based methods -5. **Bayesian Learning** - - Bayesian Inference - - Bayesian Neural Networks - -Each folder is self-contained and can be explored independently. +## Current Status + +- Standardized package layout under `src/mlscratch/` +- Verified supervised algorithms with `pytest` +- `README.md` updated to reflect current package state +- Added package-level implementations for: + - `LinearRegression` + - `LogisticRegression` + - `LassoRegression` + - `RidgeRegression` + - `KNeighborsClassifier` + - `DecisionTreeClassifier` + - `RandomForestClassifier` + - `GaussianNB` + - `LinearSVMClassifier` +- Next implementation focus: unsupervised algorithms, beginning with `KMeans` +--- -# Repo Structure +## Project Structure ``` -│ +ML-AI-Algorithms-from-scratch/ ├── LICENSE -├── README.md <- The top-level README for developers/collaborators using this project. -├── neural_network <- Folder for Neural Network implementations -│ ├── AutoEncoder -│ ├── BoltzmannMachine -│ ├── GenerativeAdversarialNetwork -│ ├── HopfieldNetwork -│ ├── LongShortTermMemoryLSTM -│ ├── MultiLayerPerceptronClassification -│ ├── MultiLayerPerceptronRegression -│ ├── RadialBasisFunctionNetworks -│ ├── SelfAttentionMechanism -│ ├── SimpleCNN -│ ├── SimpleEncoderDecoder -│ ├── SimpleRNN -│ ├── SingleLayerPerceptronClassification -│ ├── SingleLayerPerceptronRegression -│ ├── TitanicSurvialBySingleLayerPerceptron -│ └── Transformer -│ -├── reinforcement_learning <- Folder for Reinforcement Learning implementations -│ ├── Deep Deterministic Policy Gradients -│ ├── Deep Q Network -│ ├── Soft Actor Crtic -│ ├── Proximal Policy Optimization -│ └── QLearning -│ -├── supervised <- Folder for Supervised Learning implementations -│ ├── DecisionTrees -│ ├── KnearestNeighbour -│ ├── LassoRegression -│ ├── LinearRegression -│ ├── LogisticRegression -│ ├── Naive Bayes -│ ├── RandomForest -│ ├── RidgeRegression -│ └── SupportVectorMachines -│ -├── unsupervised <- Folder for Unsupervised Learning implementations -│ ├── AprioriAlgorithm -│ ├── Density-Based Spatial Clustering of Applications with Noise -│ ├── Expectation-Maximization -│ ├── Gaussian Mixture Model -│ ├── HierarchicalClustering -│ ├── IndependentComponentAnalysis -│ ├── K-MedoidsClustering -│ ├── KMeansPlusPlus -│ ├── PrincipalComponentAnalysis -│ ├── SelfOrganizing Map -│ ├── kmeanclustering -│ └── tSNE -│ -└── Bayesian Learning - ├── BayesianInference - ├── BayesianNetwork - ├── Gibbs Sampling - ├── Metropolis-Hastings Algorithm - ├── Bayesian Neural Networks - ├── BayesianLinearRegression - └── Variational Inference +├── README.md +├── pyproject.toml +├── src/ +│ └── mlscratch/ +│ ├── __init__.py +│ ├── supervised/ +│ │ ├── __init__.py +│ │ ├── linear_regression.py +│ │ ├── logistic_regression.py +│ │ ├── lasso_regression.py +│ │ ├── ridge_regression.py +│ │ ├── knn.py +│ │ ├── decision_tree.py +│ │ ├── random_forest.py +│ │ ├── naive_bayes.py +│ │ └── svm.py +│ └── unsupervised/ <- in progress +├── tests/ +│ ├── conftest.py +│ ├── supervised/ +│ │ ├── test_linear_regression.py +│ │ ├── test_logistic_regression.py +│ │ ├── test_lasso_regression.py +│ │ ├── test_ridge_regression.py +│ │ ├── test_knn.py +│ │ ├── test_decision_tree.py +│ │ ├── test_random_forest.py +│ │ ├── test_naive_bayes.py +│ │ └── test_svm.py +│ └── unsupervised/ <- coming next ``` -## Design Philosophy - -Across all implementations, the following principles are applied: - -- Prefer explicit loops over vectorized one-liners when it improves clarity -- Separate model logic, loss computation, and parameter updates -- Avoid high-level ML libraries to expose core mechanics -- Keep implementations concise and inspectable - -Many design choices intentionally trade performance for readability. - --- -## What to Expect +## What This Repository Is For -Are you eager to grasp the core concepts of these cutting-edge technologies? Look no further! In this repository, we've meticulously crafted implementations of fundamental algorithms from scratch, accompanied by detailed explanations and documentation. Our mission is to empower learners by providing hands-on experience in building these algorithms, fostering a deeper understanding of the underlying principles. +This repo is intended as an educational reference for learners who want to understand the internal mechanics of algorithms, not as a production-ready library. ---- - -## How to Learn Effectively With This Repository - -To get the most value from this repository: - -1. Read the code line-by-line -2. Add print statements or visualizations -3. Modify hyperparameters and observe behavior -4. Re-implement the same algorithm in a different style -5. Compare similar algorithms across folders +It prioritizes: -This repository is meant to be **actively explored**, not passively read. +- clarity over micro-optimization +- math-first explanations +- algorithmic correctness through tests +- reproducible minimal examples --- -## Why Learn From Scratch? - -Understanding AI, ML, DL, and RL can be a daunting task, especially for beginners. Yet, I believe that building these algorithms from the ground up offers unparalleled insights. By diving into the code, you'll gain a profound understanding of the inner workings, demystifying the complex algorithms that power the technology around us. - -## What Sets This Apart? - -- **Educational Focus:** Every algorithm is meticulously implemented with educational purposes in mind. -- **Comprehensive Documentation:** Each implementation is accompanied by thorough explanations, ensuring you not only run the code but understand it. -- **Progressive Complexity:** Starting from simpler concepts, we gradually delve into more advanced algorithms, allowing you to build your knowledge progressively. - -## Explore my Implementations +## Installation -- **Neural Networks:** Dive into the realm of neural networks, from basic perceptrons to advanced architectures like LSTMs and Transformers. -- **Reinforcement Learning:** Understand the dynamics of reinforcement learning through implementations of DDPG, DQN, PPO, and Q-learning. -- **Supervised Learning:** Explore classical supervised learning algorithms, including decision trees, regression models, and support vector machines. -- **Unsupervised Learning:** Delve into the mysteries of unsupervised learning with implementations like k-means, PCA, and GMM. +```bash +python -m pip install -e . +python -m pip install -e .[dev] +``` -## Who Is This For? +The repository is designed to work with Python 3.10+. -Whether you're a student, a curious enthusiast, or a seasoned developer looking to solidify your understanding, this repository is designed for you. Our step-by-step implementations and detailed documentation cater to learners at all levels. +--- -Ready to embark on this exciting journey? Let's code, learn, and demystify the world of AI together! 🌐✨ +## Quick Start + +```python +from mlscratch.supervised import ( + OrdinaryLeastSquares, + LogisticRegression, + LassoRegression, + RidgeRegression, + KNeighborsClassifier, + DecisionTreeClassifier, + RandomForestClassifier, + GaussianNB, + LinearSVMClassifier, +) + +model = LogisticRegression(max_iter=1000) +model.fit(X_train, y_train) +predictions = model.predict(X_test) +``` +--- -## Educational Content +## Testing -## Conceptual Background (Why These Implementations Matter) +Run the supervised test suite: -The implementations in this repository are grounded in the following learning paradigms: +```bash +python -m pytest tests/supervised -q +``` +The repository uses `pytest` and is configured with `pytest-cov` for coverage reporting. -### What is Supervised Learning? +--- -Supervised learning is a type of machine learning where the algorithm is trained on a labeled dataset. In a labeled dataset, each input data point is associated with the corresponding correct output, allowing the algorithm to learn the mapping between inputs and outputs. The goal is for the algorithm to make accurate predictions on new, unseen data. +## Package Goals -### What is Unsupervised Learning? +The long-term goal is to make this repository a best-in-class educational reference by: -Unsupervised learning involves training algorithms on unlabeled datasets. Unlike supervised learning, there are no predefined output labels. Instead, the algorithm discovers patterns, structures, or relationships within the data on its own. Common tasks in unsupervised learning include clustering and dimensionality reduction. +- standardizing module structure +- enforcing tests for correctness against `scikit-learn` baselines +- adding benchmark-driven performance comparisons +- documenting math and algorithmic intuition consistently -### What are Neural Networks? +--- -Neural networks are a class of machine learning models inspired by the structure and function of the human brain. They consist of interconnected nodes, or neurons, organized into layers. Neural networks can learn complex patterns and representations through training on labeled data. Deep learning, a subset of neural networks, involves architectures with multiple layers (deep neural networks). +## Next Work -### What is Reinforcement Learning? +The next active task is to migrate unsupervised algorithms into `src/mlscratch/unsupervised/`, starting with a clean `KMeans` implementation and its test coverage. -Reinforcement learning is a type of machine learning where an agent learns to make decisions by interacting with an environment. The agent receives feedback in the form of rewards or penalties based on the actions it takes. The goal is for the agent to learn a policy that maximizes cumulative reward over time. Reinforcement learning is commonly used in applications such as game playing, robotics, and autonomous systems. +After that, work will continue through the remaining `feedback.md` roadmap: -### What is Bayesian Learning? +- unsupervised algorithms (`KMeans`, `PCA`, `GMM`, `DBSCAN`, `SOM`, `tSNE`) +- neural network modules +- reinforcement algorithms +- Bayesian algorithms -Bayesian learning is a statistical framework that combines prior knowledge with new evidence to update and refine our beliefs about uncertain quantities. Unlike traditional machine learning approaches that focus solely on point estimates, Bayesian learning provides a probabilistic framework for reasoning about uncertainty. +--- +## Notes for Contributors -## Usage +If you want to help improve this repository, focus on: -Each algorithm is provided as a standalone Python script. You can run these scripts to see the algorithms in action. Additionally, the code is extensively documented to help you understand the implementation details. +- adding `src/mlscratch/` modules for remaining algorithms +- matching the package template used by existing supervised implementations +- writing tests that compare output to `scikit-learn` or other reliable baselines +- keeping documentation concise and mathematically rigorous diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..34703ac --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,52 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "mlscratch" +version = "2.0.0" +description = "Educational from-scratch implementations of ML/AI/RL algorithms — NumPy only." +authors = [{name = "Min Htet Myet"}] +license = {text = "Apache-2.0"} +readme = "README.md" +requires-python = ">=3.10" +keywords = ["machine-learning", "from-scratch", "numpy", "education", "algorithms", "reinforcement-learning", "bayesian"] +classifiers = [ + "Development Status :: 5 - Production/Stable", + "Intended Audience :: Education", + "Intended Audience :: Science/Research", + "License :: OSI Approved :: Apache Software License", + "Programming Language :: Python :: 3", + "Topic :: Scientific/Engineering :: Artificial Intelligence", + "Typing :: Typed", +] +dependencies = ["numpy>=1.23"] + +[project.optional-dependencies] +plot = ["matplotlib>=3.7"] +dev = ["pytest>=7", "pytest-cov>=4", "ruff>=0.1", "scikit-learn>=1.3", "scipy>=1.11", "hypothesis>=6", "black>=23", "mypy>=1.8"] +docs = ["mkdocs>=1.6", "mkdocs-material>=9.5", "mkdocstrings[python]>=0.24"] +bench = ["tabulate>=0.9"] +all = ["mlscratch[plot,dev,docs,bench]"] + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +addopts = "-ra --tb=short" +pythonpath = ["src"] +testpaths = ["tests"] + +[tool.ruff] +line-length = 100 +target-version = "py310" + +[tool.ruff.lint] +select = ["E", "F", "W", "I", "UP", "B", "D"] +ignore = ["D203", "D213"] + +[tool.coverage.run] +source = ["src/mlscratch"] + +[tool.coverage.report] +exclude_lines = ["pragma: no cover", "raise NotImplementedError", "if __name__ == .__main__.:"] diff --git a/src/mlscratch/__init__.py b/src/mlscratch/__init__.py new file mode 100644 index 0000000..bf143dc --- /dev/null +++ b/src/mlscratch/__init__.py @@ -0,0 +1,11 @@ +""" +mlscratch — Educational from-scratch ML/AI/RL algorithm implementations. + +All algorithms are implemented using only NumPy. The goal is +mathematical clarity and step-by-step transparency, not performance. + +See https://github.com/Mattral/ML-AI-Algorithms-from-scratch +""" + +__version__ = "2.0.0" +__author__ = "Min Htet Myet" diff --git a/src/mlscratch/supervised/__init__.py b/src/mlscratch/supervised/__init__.py new file mode 100644 index 0000000..36b1146 --- /dev/null +++ b/src/mlscratch/supervised/__init__.py @@ -0,0 +1,24 @@ +"""Supervised learning algorithms implemented from scratch.""" + +from .linear_regression import GradientDescentRegressor, OrdinaryLeastSquares +from .lasso_regression import LassoRegression +from .logistic_regression import LogisticRegression +from .ridge_regression import RidgeRegression +from .knn import KNeighborsClassifier +from .decision_tree import DecisionTreeClassifier +from .random_forest import RandomForestClassifier +from .naive_bayes import GaussianNB +from .svm import LinearSVMClassifier + +__all__ = [ + "OrdinaryLeastSquares", + "GradientDescentRegressor", + "LassoRegression", + "LogisticRegression", + "RidgeRegression", + "KNeighborsClassifier", + "DecisionTreeClassifier", + "RandomForestClassifier", + "GaussianNB", + "LinearSVMClassifier", +] diff --git a/src/mlscratch/supervised/decision_tree.py b/src/mlscratch/supervised/decision_tree.py new file mode 100644 index 0000000..8aab1d2 --- /dev/null +++ b/src/mlscratch/supervised/decision_tree.py @@ -0,0 +1,142 @@ +""" +Decision Tree Classifier +======================== + +A from-scratch CART decision tree classifier using Gini impurity. +""" + +from __future__ import annotations + +import numpy as np +from numpy.typing import ArrayLike, NDArray + +FloatArray = NDArray[np.float64] + + +def _validate_classification_inputs( + X: ArrayLike, y: ArrayLike, +) -> tuple[FloatArray, NDArray[np.int64]]: + X_arr = np.asarray(X, dtype=float) + y_arr = np.asarray(y, dtype=int).flatten() + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array of shape (n_samples, n_features).") + if X_arr.shape[0] != y_arr.shape[0]: + raise ValueError( + f"X has {X_arr.shape[0]} samples but y has {y_arr.shape[0]}." + ) + return X_arr, y_arr + + +class DecisionTreeClassifier: + r"""A binary or multiclass decision tree classifier. + + The CART decision rule splits nodes to minimize weighted Gini impurity: + + .. math:: + G = \sum_{k=1}^K p_k (1 - p_k) + + At each node, the best split minimizes: + + .. math:: + \frac{n_{left}}{n} G_{left} + \frac{n_{right}}{n} G_{right} + """ + + def __init__(self, max_depth: int | None = None, min_samples_split: int = 2) -> None: + self.max_depth = max_depth + self.min_samples_split = int(min_samples_split) + self.n_classes_: int | None = None + self.n_features_in_: int | None = None + self.tree_: dict | None = None + + def fit(self, X: ArrayLike, y: ArrayLike) -> "DecisionTreeClassifier": + """Grow the decision tree from training data.""" + X_arr, y_arr = _validate_classification_inputs(X, y) + self.n_features_in_ = X_arr.shape[1] + self.n_classes_ = int(len(np.unique(y_arr))) + self.tree_ = self._grow_tree(X_arr, y_arr, depth=0) + return self + + def predict(self, X: ArrayLike) -> NDArray[np.int64]: + """Predict class labels for X.""" + if self.tree_ is None: + raise RuntimeError("Call fit() before predict().") + X_arr = np.asarray(X, dtype=float) + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array.") + return np.array([self._predict_row(row, self.tree_) for row in X_arr], dtype=np.int64) + + def score(self, X: ArrayLike, y: ArrayLike) -> float: + """Return classification accuracy on the given data.""" + X_arr, y_arr = _validate_classification_inputs(X, y) + y_pred = self.predict(X_arr) + return float(np.mean(y_pred == y_arr)) + + def _gini(self, y: NDArray[np.int64]) -> float: + if y.size == 0: + return 0.0 + proportions = np.bincount(y, minlength=self.n_classes_) / y.size + return float(np.sum(proportions * (1.0 - proportions))) + + def _best_split(self, X: FloatArray, y: NDArray[np.int64]) -> tuple[int | None, float | None]: + n_samples, n_features = X.shape + if n_samples < self.min_samples_split: + return None, None + + best_idx = None + best_thr = None + best_impurity = self._gini(y) + + for idx in range(n_features): + sorted_indices = np.argsort(X[:, idx]) + thresholds = X[sorted_indices, idx] + labels = y[sorted_indices] + + left_counts = np.zeros(self.n_classes_, dtype=int) + right_counts = np.bincount(labels, minlength=self.n_classes_) + + for i in range(1, n_samples): + label = labels[i - 1] + left_counts[label] += 1 + right_counts[label] -= 1 + if thresholds[i] == thresholds[i - 1]: + continue + + left_size = i + right_size = n_samples - i + if left_size < self.min_samples_split or right_size < self.min_samples_split: + continue + + left_gini = 1.0 - np.sum((left_counts / left_size) ** 2) + right_gini = 1.0 - np.sum((right_counts / right_size) ** 2) + impurity = (left_size * left_gini + right_size * right_gini) / n_samples + + if impurity < best_impurity: + best_impurity = float(impurity) + best_idx = idx + best_thr = float((thresholds[i] + thresholds[i - 1]) / 2.0) + + return best_idx, best_thr + + def _grow_tree(self, X: FloatArray, y: NDArray[np.int64], depth: int) -> dict: + node = { + "n_samples": X.shape[0], + "n_classes": int(np.bincount(y, minlength=self.n_classes_).argmax()), + "class": int(np.bincount(y, minlength=self.n_classes_).argmax()), + } + + if self.max_depth is None or depth < self.max_depth: + idx, thr = self._best_split(X, y) + if idx is not None: + mask = X[:, idx] < thr + left = self._grow_tree(X[mask], y[mask], depth + 1) + right = self._grow_tree(X[~mask], y[~mask], depth + 1) + node.update({"feature_index": idx, "threshold": thr, "left": left, "right": right}) + return node + + def _predict_row(self, x: FloatArray, node: dict) -> int: + while "feature_index" in node: + if x[node["feature_index"]] < node["threshold"]: + node = node["left"] + else: + node = node["right"] + return int(node["class"]) diff --git a/src/mlscratch/supervised/knn.py b/src/mlscratch/supervised/knn.py new file mode 100644 index 0000000..21a610f --- /dev/null +++ b/src/mlscratch/supervised/knn.py @@ -0,0 +1,77 @@ +""" +K-Nearest Neighbors +==================== + +A simple from-scratch k-nearest neighbors classifier using Euclidean distance. +""" + +from __future__ import annotations + +import numpy as np +from numpy.typing import ArrayLike, NDArray + +FloatArray = NDArray[np.float64] + + +def _validate_classification_inputs( + X: ArrayLike, y: ArrayLike, +) -> tuple[FloatArray, NDArray[np.int64]]: + X_arr = np.asarray(X, dtype=float) + y_arr = np.asarray(y, dtype=int).flatten() + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array of shape (n_samples, n_features).") + if X_arr.shape[0] != y_arr.shape[0]: + raise ValueError( + f"X has {X_arr.shape[0]} samples but y has {y_arr.shape[0]}." + ) + return X_arr, y_arr + + +class KNeighborsClassifier: + """A k-nearest neighbors classifier. + + Parameters + ---------- + n_neighbors : int, default=3 + Number of nearest neighbors to use for prediction. + """ + + def __init__(self, n_neighbors: int = 3) -> None: + if n_neighbors < 1: + raise ValueError("n_neighbors must be at least 1.") + self.n_neighbors = n_neighbors + self.X_train_: FloatArray | None = None + self.y_train_: NDArray[np.int64] | None = None + + def fit(self, X: ArrayLike, y: ArrayLike) -> "KNeighborsClassifier": + """Store the training dataset.""" + X_arr, y_arr = _validate_classification_inputs(X, y) + self.X_train_ = X_arr + self.y_train_ = y_arr + return self + + def predict(self, X: ArrayLike) -> NDArray[np.int64]: + """Predict class labels for the input samples.""" + if self.X_train_ is None or self.y_train_ is None: + raise RuntimeError("Call fit() before predict().") + X_arr = np.asarray(X, dtype=float) + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array.") + distances = self._pairwise_distances(X_arr, self.X_train_) + nearest_indices = np.argsort(distances, axis=1)[:, : self.n_neighbors] + nearest_labels = self.y_train_[nearest_indices] + return np.array([np.bincount(row).argmax() for row in nearest_labels], dtype=np.int64) + + def score(self, X: ArrayLike, y: ArrayLike) -> float: + """Return classification accuracy on the given data.""" + X_arr, y_arr = _validate_classification_inputs(X, y) + y_pred = self.predict(X_arr) + return float(np.mean(y_pred == y_arr)) + + def _pairwise_distances(self, X: FloatArray, Y: FloatArray) -> FloatArray: + """Compute the Euclidean distance matrix between X and Y.""" + X_norm = np.sum(X**2, axis=1)[:, None] + Y_norm = np.sum(Y**2, axis=1)[None, :] + cross = X @ Y.T + distances = np.sqrt(np.maximum(X_norm + Y_norm - 2.0 * cross, 0.0)) + return distances diff --git a/src/mlscratch/supervised/lasso_regression.py b/src/mlscratch/supervised/lasso_regression.py new file mode 100644 index 0000000..90d8254 --- /dev/null +++ b/src/mlscratch/supervised/lasso_regression.py @@ -0,0 +1,125 @@ +""" +Lasso Regression +================ + +Lasso regression using coordinate descent and an explicit intercept. +""" + +from __future__ import annotations + +import numpy as np +from numpy.typing import ArrayLike, NDArray + +FloatArray = NDArray[np.float64] + + +def _validate_regression_inputs( + X: ArrayLike, y: ArrayLike, +) -> tuple[FloatArray, FloatArray]: + X_arr = np.asarray(X, dtype=float) + y_arr = np.asarray(y, dtype=float).flatten() + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array of shape (n_samples, n_features).") + if X_arr.shape[0] != y_arr.shape[0]: + raise ValueError( + f"X has {X_arr.shape[0]} samples but y has {y_arr.shape[0]}." + ) + return X_arr, y_arr + + +class LassoRegression: + """Lasso regression using coordinate descent. + + Parameters + ---------- + alpha : float, default=1.0 + Regularization strength for the L1 penalty. + max_iter : int, default=1000 + Maximum number of coordinate descent iterations. + tol : float, default=1e-4 + Convergence threshold for coefficient updates. + + Attributes + ---------- + coef_ : FloatArray + Estimated coefficients for each feature. + intercept_ : float + Estimated intercept term. + loss_history_ : list[float] + Training loss on each iteration. + """ + + def __init__( + self, + alpha: float = 1.0, + max_iter: int = 1000, + tol: float = 1e-4, + ) -> None: + self.alpha = float(alpha) + self.max_iter = int(max_iter) + self.tol = float(tol) + self.coef_: FloatArray | None = None + self.intercept_: float | None = None + self.loss_history_: list[float] = [] + self.feature_means_: FloatArray | None = None + self.y_mean_: float | None = None + + def fit(self, X: ArrayLike, y: ArrayLike) -> "LassoRegression": + """Fit the Lasso regression model to the training data.""" + X_arr, y_arr = _validate_regression_inputs(X, y) + n_samples, n_features = X_arr.shape + self.feature_means_ = np.mean(X_arr, axis=0) + self.y_mean_ = np.mean(y_arr) + X_centered = X_arr - self.feature_means_ + y_centered = y_arr - self.y_mean_ + + self.coef_ = np.zeros(n_features, dtype=np.float64) + self.intercept_ = 0.0 + self.loss_history_ = [] + + X_norm_sq = np.sum(X_centered**2, axis=0) / n_samples + X_norm_sq = np.where(X_norm_sq == 0.0, 1.0, X_norm_sq) + + for iteration in range(self.max_iter): + coef_old = self.coef_.copy() + + for j in range(n_features): + residual = y_centered - (X_centered @ self.coef_ - X_centered[:, j] * self.coef_[j]) + rho = (X_centered[:, j] @ residual) / n_samples + if rho < -self.alpha: + self.coef_[j] = (rho + self.alpha) / X_norm_sq[j] + elif rho > self.alpha: + self.coef_[j] = (rho - self.alpha) / X_norm_sq[j] + else: + self.coef_[j] = 0.0 + + max_coef_change = np.max(np.abs(self.coef_ - coef_old)) + self.intercept_ = self.y_mean_ - float(self.feature_means_ @ self.coef_) + loss = self._objective(X_arr, y_arr) + self.loss_history_.append(float(loss)) + if max_coef_change < self.tol: + break + + return self + + def predict(self, X: ArrayLike) -> FloatArray: + """Predict using the fitted Lasso model.""" + if self.coef_ is None or self.intercept_ is None: + raise RuntimeError("Call fit() before predict().") + X_arr = np.asarray(X, dtype=float) + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array.") + return (X_arr @ self.coef_ + self.intercept_).astype(np.float64) + + def score(self, X: ArrayLike, y: ArrayLike) -> float: + """Return R² of the fitted model on the given data.""" + X_arr, y_arr = _validate_regression_inputs(X, y) + y_pred = self.predict(X_arr) + ss_res = np.sum((y_arr - y_pred) ** 2) + ss_tot = np.sum((y_arr - np.mean(y_arr)) ** 2) + return float(1.0 - ss_res / ss_tot) if ss_tot > 0 else 0.0 + + def _objective(self, X: FloatArray, y: FloatArray) -> float: + y_pred = X @ self.coef_ + self.intercept_ + mse = np.mean((y - y_pred) ** 2) / 2.0 + return float(mse + self.alpha * np.sum(np.abs(self.coef_))) diff --git a/src/mlscratch/supervised/linear_regression.py b/src/mlscratch/supervised/linear_regression.py new file mode 100644 index 0000000..abce8ff --- /dev/null +++ b/src/mlscratch/supervised/linear_regression.py @@ -0,0 +1,197 @@ +""" +Linear Regression +================= + +Ordinary Least Squares and mini-batch gradient descent implementations for +linear regression. + +The module uses a clear, sklearn-compatible interface with explicit math. +""" + +from __future__ import annotations + +from typing import Callable + +import numpy as np +from numpy.typing import ArrayLike, NDArray + +FloatArray = NDArray[np.float64] + + +def _validate_regression_inputs( + X: ArrayLike, y: ArrayLike, +) -> tuple[FloatArray, FloatArray]: + X_arr = np.asarray(X, dtype=float) + y_arr = np.asarray(y, dtype=float).flatten() + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array of shape (n_samples, n_features).") + if X_arr.shape[0] != y_arr.shape[0]: + raise ValueError( + f"X has {X_arr.shape[0]} samples but y has {y_arr.shape[0]}." + ) + return X_arr, y_arr + + +class OrdinaryLeastSquares: + """Ordinary least squares regression using the normal equations. + + Parameters + ---------- + add_intercept : bool, default=True + If True, the model fits an intercept term by prepending a column of ones + to the design matrix. + + Attributes + ---------- + coef_ : FloatArray + Estimated regression coefficients for each feature. + intercept_ : float + Estimated bias term. + residuals_ : FloatArray + Residual values after fitting. + """ + + def __init__(self, add_intercept: bool = True) -> None: + self.add_intercept = add_intercept + self.coef_: FloatArray | None = None + self.intercept_: float | None = None + self.residuals_: FloatArray | None = None + + def fit(self, X: ArrayLike, y: ArrayLike) -> "OrdinaryLeastSquares": + r"""Fit the linear regression model. + + The closed-form least-squares solution is computed via a numerically + stable least-squares solver: + + .. math:: + \hat{\beta} = \operatorname{argmin}_\beta \|X \beta - y\|_2^2 + + Returns + ------- + self : OrdinaryLeastSquares + """ + X_arr, y_arr = _validate_regression_inputs(X, y) + if self.add_intercept: + X_arr = np.column_stack([np.ones(X_arr.shape[0]), X_arr]) + + solution, residuals, *_ = np.linalg.lstsq(X_arr, y_arr, rcond=None) + if self.add_intercept: + self.intercept_ = float(solution[0]) + self.coef_ = solution[1:].astype(np.float64) + y_pred = X_arr[:, 1:] @ self.coef_ + self.intercept_ + else: + self.intercept_ = 0.0 + self.coef_ = solution.astype(np.float64) + y_pred = X_arr @ self.coef_ + + self.residuals_ = y_arr - y_pred + return self + + def predict(self, X: ArrayLike) -> FloatArray: + """Predict target values for new data.""" + if self.coef_ is None or self.intercept_ is None: + raise RuntimeError("Call fit() before predict().") + X_arr = np.asarray(X, dtype=float) + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array.") + result = X_arr @ self.coef_ + self.intercept_ + return result.astype(np.float64) + + def score(self, X: ArrayLike, y: ArrayLike) -> float: + """Return the coefficient of determination R² on the given data.""" + _, y_arr = _validate_regression_inputs(X, y) + y_pred = self.predict(X) + ss_res = np.sum((y_arr - y_pred) ** 2) + ss_tot = np.sum((y_arr - np.mean(y_arr)) ** 2) + return float(1.0 - ss_res / ss_tot) if ss_tot > 0 else 0.0 + + +class GradientDescentRegressor: + r"""Linear regression using mini-batch gradient descent. + + The squared error loss is: + + .. math:: + L(W, b) = \frac{1}{n} \sum_{i=1}^n (y_i - X_i W - b)^2 + + The gradient with respect to the weights is: + + .. math:: + \frac{\partial L}{\partial W} = -\frac{2}{n} X^\top (y - XW - b) + """ + + def __init__( + self, + learning_rate: float = 0.01, + n_epochs: int = 1000, + batch_size: int = 32, + random_state: int | None = None, + verbose: bool = False, + ) -> None: + self.learning_rate = learning_rate + self.n_epochs = n_epochs + self.batch_size = batch_size + self.random_state = random_state + self.verbose = verbose + self.coef_: FloatArray | None = None + self.intercept_: float | None = None + self.loss_history_: list[float] = [] + + def fit(self, X: ArrayLike, y: ArrayLike) -> "GradientDescentRegressor": + """Fit the model using mini-batch gradient descent.""" + X_arr, y_arr = _validate_regression_inputs(X, y) + rng = np.random.default_rng(self.random_state) + n_samples, n_features = X_arr.shape + self.coef_ = np.zeros(n_features, dtype=np.float64) + self.intercept_ = 0.0 + self.loss_history_ = [] + learning_rate = self.learning_rate + prev_loss = float("inf") + + for epoch in range(self.n_epochs): + indices = rng.permutation(n_samples) + X_shuffled = X_arr[indices] + y_shuffled = y_arr[indices] + coef_before = self.coef_.copy() + intercept_before = self.intercept_ + + for start in range(0, n_samples, self.batch_size): + end = start + self.batch_size + X_batch = X_shuffled[start:end] + y_batch = y_shuffled[start:end] + y_pred = X_batch @ self.coef_ + self.intercept_ + errors = y_pred - y_batch + grad_w = (2.0 / X_batch.shape[0]) * (X_batch.T @ errors) + grad_b = (2.0 / X_batch.shape[0]) * np.sum(errors) + self.coef_ -= learning_rate * grad_w + self.intercept_ -= learning_rate * grad_b + + loss = np.mean((X_arr @ self.coef_ + self.intercept_ - y_arr) ** 2) + if loss > prev_loss + 1e-12: + self.coef_ = coef_before + self.intercept_ = intercept_before + learning_rate *= 0.5 + loss = prev_loss + + self.loss_history_.append(float(loss)) + prev_loss = loss + if self.verbose and epoch % 100 == 0: + print(f"Epoch {epoch} loss={loss:.6f}") + return self + + def predict(self, X: ArrayLike) -> FloatArray: + """Predict target values for new data.""" + if self.coef_ is None or self.intercept_ is None: + raise RuntimeError("Call fit() before predict().") + X_arr = np.asarray(X, dtype=float) + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array.") + return (X_arr @ self.coef_ + self.intercept_).astype(np.float64) + + def score(self, X: ArrayLike, y: ArrayLike) -> float: + """Return the coefficient of determination R² on the given data.""" + _, y_arr = _validate_regression_inputs(X, y) + y_pred = self.predict(X) + ss_res = np.sum((y_arr - y_pred) ** 2) + ss_tot = np.sum((y_arr - np.mean(y_arr)) ** 2) + return float(1.0 - ss_res / ss_tot) if ss_tot > 0 else 0.0 diff --git a/src/mlscratch/supervised/logistic_regression.py b/src/mlscratch/supervised/logistic_regression.py new file mode 100644 index 0000000..46c1988 --- /dev/null +++ b/src/mlscratch/supervised/logistic_regression.py @@ -0,0 +1,119 @@ +""" +Logistic Regression +=================== + +A from-scratch binary classifier using gradient descent and a sigmoid link. +""" + +from __future__ import annotations + +import numpy as np +from numpy.typing import ArrayLike, NDArray + +FloatArray = NDArray[np.float64] + + +def _validate_classification_inputs( + X: ArrayLike, y: ArrayLike, +) -> tuple[FloatArray, FloatArray]: + X_arr = np.asarray(X, dtype=float) + y_arr = np.asarray(y, dtype=float).flatten() + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array of shape (n_samples, n_features).") + if X_arr.shape[0] != y_arr.shape[0]: + raise ValueError( + f"X has {X_arr.shape[0]} samples but y has {y_arr.shape[0]}." + ) + if not np.all(np.isin(y_arr, [0.0, 1.0])): + raise ValueError("y must contain only binary labels 0 and 1.") + return X_arr, y_arr + + +class LogisticRegression: + r"""Binary logistic regression using gradient descent. + + The model is: + + .. math:: + p(y=1 \mid x) = \sigma(w^\top x + b), + \quad \sigma(z) = \frac{1}{1 + e^{-z}} + + The loss is the binary cross-entropy: + + .. math:: + L = -\frac{1}{n} \sum_{i=1}^n + \left[y_i \log \sigma(z_i) + (1-y_i) \log (1-\sigma(z_i))\right] + """ + + def __init__( + self, + learning_rate: float = 0.01, + n_epochs: int = 1000, + batch_size: int = 32, + random_state: int | None = None, + verbose: bool = False, + ) -> None: + self.learning_rate = learning_rate + self.n_epochs = n_epochs + self.batch_size = batch_size + self.random_state = random_state + self.verbose = verbose + self.weights_: FloatArray | None = None + self.bias_: float | None = None + self.loss_history_: list[float] = [] + + def fit(self, X: ArrayLike, y: ArrayLike) -> "LogisticRegression": + """Fit the logistic regression model to binary data.""" + X_arr, y_arr = _validate_classification_inputs(X, y) + rng = np.random.default_rng(self.random_state) + n_samples, n_features = X_arr.shape + self.weights_ = np.zeros(n_features, dtype=np.float64) + self.bias_ = 0.0 + self.loss_history_ = [] + + for epoch in range(self.n_epochs): + perm = rng.permutation(n_samples) + for start in range(0, n_samples, self.batch_size): + end = start + self.batch_size + X_batch = X_arr[perm[start:end]] + y_batch = y_arr[perm[start:end]] + z = X_batch @ self.weights_ + self.bias_ + predictions = self._sigmoid(z) + errors = predictions - y_batch + grad_w = X_batch.T @ errors / X_batch.shape[0] + grad_b = np.mean(errors) + self.weights_ -= self.learning_rate * grad_w + self.bias_ -= self.learning_rate * grad_b + + loss = self._binary_cross_entropy(y_arr, self.predict_proba(X_arr)) + self.loss_history_.append(float(loss)) + if self.verbose and epoch % 100 == 0: + print(f"Epoch {epoch} loss={loss:.6f}") + return self + + def predict_proba(self, X: ArrayLike) -> FloatArray: + """Return probability estimates for the positive class.""" + if self.weights_ is None or self.bias_ is None: + raise RuntimeError("Call fit() before predict_proba().") + X_arr = np.asarray(X, dtype=float) + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array.") + return self._sigmoid(X_arr @ self.weights_ + self.bias_) + + def predict(self, X: ArrayLike) -> NDArray[np.int64]: + """Return binary predictions for the input data.""" + return (self.predict_proba(X) >= 0.5).astype(np.int64) + + def score(self, X: ArrayLike, y: ArrayLike) -> float: + """Return classification accuracy on the given dataset.""" + X_arr, y_arr = _validate_classification_inputs(X, y) + y_pred = self.predict(X_arr) + return float(np.mean(y_pred == y_arr)) + + def _sigmoid(self, z: FloatArray) -> FloatArray: + z = np.clip(z, -500.0, 500.0) + return 1.0 / (1.0 + np.exp(-z)) + + def _binary_cross_entropy(self, y_true: FloatArray, y_prob: FloatArray) -> float: + y_prob = np.clip(y_prob, 1e-12, 1.0 - 1e-12) + return float(-np.mean(y_true * np.log(y_prob) + (1.0 - y_true) * np.log(1.0 - y_prob))) diff --git a/src/mlscratch/supervised/naive_bayes.py b/src/mlscratch/supervised/naive_bayes.py new file mode 100644 index 0000000..b67159b --- /dev/null +++ b/src/mlscratch/supervised/naive_bayes.py @@ -0,0 +1,113 @@ +r""" +Gaussian Naive Bayes Classifier +================================ + +A probabilistic classifier that assumes each class follows a Gaussian +distribution and that features are conditionally independent given the class. + +The model computes class log-likelihoods as: + +.. math:: + \log p(\mathbf{x}, y_k) + = \log \pi_k - \frac{1}{2} \sum_{j=1}^d \left[ + \log(2\pi \sigma_{kj}^2) + + \frac{(x_j - \mu_{kj})^2}{\sigma_{kj}^2} + \right] + +Complexity +---------- +- Training: O(n d) +- Inference: O(n d) +- Space: O(K d) +""" + +from __future__ import annotations + +import numpy as np +from numpy.typing import ArrayLike, NDArray + +FloatArray = NDArray[np.float64] +IntArray = NDArray[np.int64] + + +def _validate_classification_inputs( + X: ArrayLike, + y: ArrayLike, +) -> tuple[FloatArray, IntArray]: + X_arr = np.asarray(X, dtype=float) + y_arr = np.asarray(y, dtype=int).flatten() + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array of shape (n_samples, n_features).") + if X_arr.shape[0] != y_arr.shape[0]: + raise ValueError( + f"X has {X_arr.shape[0]} samples but y has {y_arr.shape[0]}." + ) + return X_arr, y_arr + + +class GaussianNB: + """Gaussian Naive Bayes classifier. + + Parameters + ---------- + var_smoothing : float, default=1e-9 + Portion of the largest variance of all features added to variances for + stability in the Gaussian likelihood denominator. + """ + + def __init__(self, var_smoothing: float = 1e-9) -> None: + self.var_smoothing = float(var_smoothing) + self.class_count_: IntArray | None = None + self.class_prior_: FloatArray | None = None + self.class_mean_: FloatArray | None = None + self.class_var_: FloatArray | None = None + self.classes_: IntArray | None = None + self.n_features_in_: int | None = None + + def fit(self, X: ArrayLike, y: ArrayLike) -> "GaussianNB": + X_arr, y_arr = _validate_classification_inputs(X, y) + self.n_features_in_ = X_arr.shape[1] + self.classes_, counts = np.unique(y_arr, return_counts=True) + self.class_count_ = counts.astype(np.int64) + self.class_prior_ = counts.astype(np.float64) / float(y_arr.size) + + means = [] + variances = [] + for clazz in self.classes_: + X_class = X_arr[y_arr == clazz] + means.append(X_class.mean(axis=0)) + variances.append(X_class.var(axis=0) + self.var_smoothing) + + self.class_mean_ = np.vstack(means) + self.class_var_ = np.vstack(variances) + return self + + def predict(self, X: ArrayLike) -> IntArray: + if self.class_prior_ is None or self.class_mean_ is None or self.class_var_ is None: + raise RuntimeError("Call fit() before predict().") + X_arr = np.asarray(X, dtype=float) + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array.") + log_likelihood = self._joint_log_likelihood(X_arr) + argmax = np.argmax(log_likelihood, axis=1) + return self.classes_[argmax] + + def score(self, X: ArrayLike, y: ArrayLike) -> float: + X_arr, y_arr = _validate_classification_inputs(X, y) + return float(np.mean(self.predict(X_arr) == y_arr)) + + def _joint_log_likelihood(self, X: FloatArray) -> FloatArray: + n_samples, n_features = X.shape + if self.class_mean_ is None or self.class_var_ is None or self.class_prior_ is None: + raise RuntimeError("Classifier must be fitted before computing likelihoods.") + + joint = np.empty((n_samples, self.classes_.size), dtype=np.float64) + for idx, (prior, mean, var) in enumerate( + zip(self.class_prior_, self.class_mean_, self.class_var_) + ): + log_prior = np.log(prior) + log_det = -0.5 * np.sum(np.log(2.0 * np.pi * var)) + diff = X - mean + exp_term = -0.5 * np.sum((diff ** 2) / var, axis=1) + joint[:, idx] = log_prior + log_det + exp_term + return joint diff --git a/src/mlscratch/supervised/random_forest.py b/src/mlscratch/supervised/random_forest.py new file mode 100644 index 0000000..454eb9c --- /dev/null +++ b/src/mlscratch/supervised/random_forest.py @@ -0,0 +1,118 @@ +""" +Random Forest Classifier +======================== + +A from-scratch random forest ensemble built from decision tree classifiers. +""" + +from __future__ import annotations + +import numpy as np +from numpy.typing import ArrayLike, NDArray + +from .decision_tree import DecisionTreeClassifier + +FloatArray = NDArray[np.float64] + + +def _validate_classification_inputs( + X: ArrayLike, y: ArrayLike, +) -> tuple[NDArray[np.float64], NDArray[np.int64]]: + X_arr = np.asarray(X, dtype=float) + y_arr = np.asarray(y, dtype=int).flatten() + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array of shape (n_samples, n_features).") + if X_arr.shape[0] != y_arr.shape[0]: + raise ValueError( + f"X has {X_arr.shape[0]} samples but y has {y_arr.shape[0]}." + ) + return X_arr, y_arr + + +class RandomForestClassifier: + """Random forest classifier using bootstrap aggregation of decision trees.""" + + def __init__( + self, + n_estimators: int = 100, + max_depth: int | None = None, + min_samples_split: int = 2, + max_features: int | str | None = "sqrt", + random_state: int | None = None, + ) -> None: + self.n_estimators = int(n_estimators) + self.max_depth = max_depth + self.min_samples_split = int(min_samples_split) + self.max_features = max_features + self.random_state = random_state + self.estimators_: list[tuple[DecisionTreeClassifier, NDArray[np.int64]]] = [] + self.feature_importances_: FloatArray | None = None + self.n_features_in_: int | None = None + + def fit(self, X: ArrayLike, y: ArrayLike) -> "RandomForestClassifier": + X_arr, y_arr = _validate_classification_inputs(X, y) + rng = np.random.default_rng(self.random_state) + self.n_features_in_ = X_arr.shape[1] + features_per_tree = self._resolve_max_features(self.n_features_in_) + + self.estimators_ = [] + importances: FloatArray = np.zeros(self.n_features_in_, dtype=np.float64) + + for _ in range(self.n_estimators): + indices = rng.choice(X_arr.shape[0], size=X_arr.shape[0], replace=True) + feature_indices = rng.choice(self.n_features_in_, size=features_per_tree, replace=False) + tree = DecisionTreeClassifier( + max_depth=self.max_depth, + min_samples_split=self.min_samples_split, + ) + tree.fit(X_arr[indices][:, feature_indices], y_arr[indices]) + self.estimators_.append((tree, feature_indices)) + importances += self._tree_importances(tree, feature_indices) + + self.feature_importances_ = importances / float(self.n_estimators) + return self + + def predict(self, X: ArrayLike) -> NDArray[np.int64]: + if not self.estimators_: + raise RuntimeError("Call fit() before predict().") + X_arr = np.asarray(X, dtype=float) + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array.") + predictions = np.vstack( + [tree.predict(X_arr[:, features]) for tree, features in self.estimators_] + ) + return np.array([np.bincount(row).argmax() for row in predictions.T], dtype=np.int64) + + def score(self, X: ArrayLike, y: ArrayLike) -> float: + X_arr, y_arr = _validate_classification_inputs(X, y) + return float(np.mean(self.predict(X_arr) == y_arr)) + + def _resolve_max_features(self, n_features: int) -> int: + if self.max_features is None: + return n_features + if isinstance(self.max_features, str): + if self.max_features == "sqrt": + return max(1, int(np.sqrt(n_features))) + if self.max_features == "log2": + return max(1, int(np.log2(n_features))) + raise ValueError("max_features must be None, int, 'sqrt', or 'log2'.") + return int(self.max_features) + + def _tree_importances( + self, tree: DecisionTreeClassifier, feature_indices: NDArray[np.int64] + ) -> FloatArray: + counts = np.zeros(self.n_features_in_, dtype=np.float64) + self._accumulate_importance(tree.tree_, counts, feature_indices) + return counts + + def _accumulate_importance( + self, + node: dict | None, + counts: FloatArray, + feature_indices: NDArray[np.int64], + ) -> None: + if node is None or "feature_index" not in node: + return + counts[feature_indices[node["feature_index"]]] += 1.0 + self._accumulate_importance(node["left"], counts, feature_indices) + self._accumulate_importance(node["right"], counts, feature_indices) diff --git a/src/mlscratch/supervised/ridge_regression.py b/src/mlscratch/supervised/ridge_regression.py new file mode 100644 index 0000000..e8f0ac7 --- /dev/null +++ b/src/mlscratch/supervised/ridge_regression.py @@ -0,0 +1,93 @@ +""" +Ridge Regression +================ + +Ridge regression using the closed-form regularized normal equations. +""" + +from __future__ import annotations + +import numpy as np +from numpy.typing import ArrayLike, NDArray + +FloatArray = NDArray[np.float64] + + +def _validate_regression_inputs( + X: ArrayLike, y: ArrayLike, +) -> tuple[FloatArray, FloatArray]: + X_arr = np.asarray(X, dtype=float) + y_arr = np.asarray(y, dtype=float).flatten() + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array of shape (n_samples, n_features).") + if X_arr.shape[0] != y_arr.shape[0]: + raise ValueError( + f"X has {X_arr.shape[0]} samples but y has {y_arr.shape[0]}." + ) + return X_arr, y_arr + + +class RidgeRegression: + """Ridge regression with an L2 penalty on coefficients. + + Parameters + ---------- + alpha : float, default=1.0 + Regularization strength (L2 penalty coefficient). + add_intercept : bool, default=True + Whether to fit an intercept term. + + Attributes + ---------- + coef_ : FloatArray + Estimated coefficients for each feature. + intercept_ : float + Estimated intercept. + """ + + def __init__(self, alpha: float = 1.0, add_intercept: bool = True) -> None: + self.alpha = float(alpha) + self.add_intercept = add_intercept + self.coef_: FloatArray | None = None + self.intercept_: float | None = None + + def fit(self, X: ArrayLike, y: ArrayLike) -> "RidgeRegression": + """Fit the Ridge regression model using the closed-form solution.""" + X_arr, y_arr = _validate_regression_inputs(X, y) + if self.add_intercept: + X_arr = np.column_stack([np.ones(X_arr.shape[0]), X_arr]) + + n_features = X_arr.shape[1] + identity = np.eye(n_features) + if self.add_intercept: + identity[0, 0] = 0.0 + + coef = np.linalg.solve( + X_arr.T @ X_arr + self.alpha * identity, + X_arr.T @ y_arr, + ) + + if self.add_intercept: + self.intercept_ = float(coef[0]) + self.coef_ = coef[1:].astype(np.float64) + else: + self.intercept_ = 0.0 + self.coef_ = coef.astype(np.float64) + return self + + def predict(self, X: ArrayLike) -> FloatArray: + """Predict targets using the fitted Ridge model.""" + if self.coef_ is None or self.intercept_ is None: + raise RuntimeError("Call fit() before predict().") + X_arr = np.asarray(X, dtype=float) + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array.") + return (X_arr @ self.coef_ + self.intercept_).astype(np.float64) + + def score(self, X: ArrayLike, y: ArrayLike) -> float: + """Return R² of the fitted Ridge model.""" + X_arr, y_arr = _validate_regression_inputs(X, y) + y_pred = self.predict(X_arr) + ss_res = np.sum((y_arr - y_pred) ** 2) + ss_tot = np.sum((y_arr - np.mean(y_arr)) ** 2) + return float(1.0 - ss_res / ss_tot) if ss_tot > 0 else 0.0 diff --git a/src/mlscratch/supervised/svm.py b/src/mlscratch/supervised/svm.py new file mode 100644 index 0000000..65b4152 --- /dev/null +++ b/src/mlscratch/supervised/svm.py @@ -0,0 +1,117 @@ +r""" +Linear Support Vector Machine Classifier +======================================= + +A linear SVM trained with stochastic sub-gradient descent on the hinge loss. + +The objective is: + +.. math:: + \frac{1}{2} \|w\|^2 + C \sum_{i=1}^n \max(0, 1 - y_i w^\top x_i) + +Where labels are mapped to :math:`y_i \in \{-1, +1\}`. + +Complexity +---------- +- Training: O(n d \cdot n\_epochs) +- Inference: O(d) +- Space: O(d) +""" + +from __future__ import annotations + +import numpy as np +from numpy.typing import ArrayLike, NDArray + +FloatArray = NDArray[np.float64] +IntArray = NDArray[np.int64] + + +def _validate_classification_inputs( + X: ArrayLike, y: ArrayLike +) -> tuple[FloatArray, IntArray]: + X_arr = np.asarray(X, dtype=float) + y_arr = np.asarray(y, dtype=int).flatten() + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array of shape (n_samples, n_features).") + if X_arr.shape[0] != y_arr.shape[0]: + raise ValueError( + f"X has {X_arr.shape[0]} samples but y has {y_arr.shape[0]}." + ) + unique = np.unique(y_arr) + if unique.size != 2: + raise ValueError("LinearSVMClassifier supports only binary classification.") + return X_arr, y_arr + + +class LinearSVMClassifier: + """Linear binary SVM classifier trained with hinge-loss SGD. + + Parameters + ---------- + learning_rate : float, default=1e-3 + Step size for weight updates. + n_epochs : int, default=1000 + Number of passes over the training data. + C : float, default=1.0 + Regularization strength. + random_state : int | None, default=None + Seed for SGD shuffling. + """ + + def __init__( + self, + learning_rate: float = 1e-3, + n_epochs: int = 1000, + C: float = 1.0, + random_state: int | None = None, + ) -> None: + self.learning_rate = float(learning_rate) + self.n_epochs = int(n_epochs) + self.C = float(C) + self.random_state = random_state + self.w_: FloatArray | None = None + self.classes_: IntArray | None = None + self.n_features_in_: int | None = None + + def fit(self, X: ArrayLike, y: ArrayLike) -> "LinearSVMClassifier": + X_arr, y_arr = _validate_classification_inputs(X, y) + self.n_features_in_ = X_arr.shape[1] + self.classes_ = np.unique(y_arr) + + signed_labels = np.where(y_arr == self.classes_[0], -1, 1) + X_aug = np.hstack([np.ones((X_arr.shape[0], 1), dtype=float), X_arr]) + self.w_ = np.zeros(X_aug.shape[1], dtype=np.float64) + rng = np.random.default_rng(self.random_state) + + for _ in range(self.n_epochs): + indices = rng.permutation(X_aug.shape[0]) + for i in indices: + xi = X_aug[i] + yi = signed_labels[i] + margin = yi * np.dot(self.w_, xi) + if margin >= 1.0: + gradient = np.concatenate(([0.0], self.w_[1:])) + else: + gradient = np.concatenate(([0.0], self.w_[1:])) - self.C * yi * xi + self.w_ -= self.learning_rate * gradient + return self + + def decision_function(self, X: ArrayLike) -> FloatArray: + if self.w_ is None: + raise RuntimeError("Call fit() before decision_function().") + X_arr = np.asarray(X, dtype=float) + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array.") + return np.dot(np.hstack([np.ones((X_arr.shape[0], 1), dtype=float), X_arr]), self.w_) + + def predict(self, X: ArrayLike) -> IntArray: + if self.w_ is None or self.classes_ is None: + raise RuntimeError("Call fit() before predict().") + scores = self.decision_function(X) + labels = np.where(scores >= 0.0, self.classes_[1], self.classes_[0]) + return labels.astype(np.int64) + + def score(self, X: ArrayLike, y: ArrayLike) -> float: + X_arr, y_arr = _validate_classification_inputs(X, y) + return float(np.mean(self.predict(X_arr) == y_arr)) diff --git a/src/mlscratch/unsupervised/__init__.py b/src/mlscratch/unsupervised/__init__.py new file mode 100644 index 0000000..f573446 --- /dev/null +++ b/src/mlscratch/unsupervised/__init__.py @@ -0,0 +1,5 @@ +"""Unsupervised learning algorithms implemented from scratch.""" + +from .kmeans import KMeans + +__all__ = ["KMeans"] diff --git a/src/mlscratch/unsupervised/kmeans.py b/src/mlscratch/unsupervised/kmeans.py new file mode 100644 index 0000000..e5d27f8 --- /dev/null +++ b/src/mlscratch/unsupervised/kmeans.py @@ -0,0 +1,135 @@ +r""" +K-Means Clustering +=================== + +A classic unsupervised clustering algorithm using Lloyd's iteration with +K-Means++ initialization. + +The objective minimized is: + +.. math:: + J = \sum_{i=1}^n \min_{1 \leq k \leq K} \|x_i - \mu_k\|^2 + +Complexity +---------- +- Training: O(n K d \cdot n\_iter) +- Inference: O(n K d) +- Space: O(K d) +""" + +from __future__ import annotations + +import numpy as np +from numpy.typing import ArrayLike, NDArray + +FloatArray = NDArray[np.float64] + + +def _validate_input(X: ArrayLike) -> FloatArray: + X_arr = np.asarray(X, dtype=float) + if X_arr.ndim != 2: + raise ValueError("X must be a 2D array of shape (n_samples, n_features).") + return X_arr + + +class KMeans: + """K-Means clustering with optional K-Means++ initialization. + + Parameters + ---------- + n_clusters : int + The number of clusters to form. + max_iter : int, default=300 + Maximum number of iterations of the k-means algorithm for a single run. + tol : float, default=1e-4 + Convergence tolerance. The algorithm stops when centroid movement is + less than this threshold. + random_state : int | None, default=None + Seed for centroid initialization. + """ + + def __init__( + self, + n_clusters: int = 8, + max_iter: int = 300, + tol: float = 1e-4, + random_state: int | None = None, + ) -> None: + self.n_clusters = int(n_clusters) + self.max_iter = int(max_iter) + self.tol = float(tol) + self.random_state = random_state + self.cluster_centers_: FloatArray | None = None + self.labels_: NDArray[np.int64] | None = None + self.inertia_: float | None = None + self.n_iter_: int | None = None + + def fit(self, X: ArrayLike) -> "KMeans": + X_arr = _validate_input(X) + n_samples, n_features = X_arr.shape + if self.n_clusters <= 0 or self.n_clusters > n_samples: + raise ValueError("n_clusters must be between 1 and n_samples.") + + rng = np.random.default_rng(self.random_state) + centers = self._initialize_centroids(X_arr, rng) + + for iteration in range(1, self.max_iter + 1): + labels = self._assign_clusters(X_arr, centers) + new_centers = self._compute_centers(X_arr, labels, n_features) + + shift = np.linalg.norm(centers - new_centers, axis=1).max() + centers = new_centers + if shift <= self.tol: + break + + self.cluster_centers_ = centers + self.labels_ = labels + self.inertia_ = float(self._compute_inertia(X_arr, centers, labels)) + self.n_iter_ = iteration + return self + + def predict(self, X: ArrayLike) -> NDArray[np.int64]: + if self.cluster_centers_ is None: + raise RuntimeError("Call fit() before predict().") + X_arr = _validate_input(X) + if X_arr.shape[1] != self.cluster_centers_.shape[1]: + raise ValueError("X has a different number of features than the training data.") + return self._assign_clusters(X_arr, self.cluster_centers_) + + def _initialize_centroids(self, X: FloatArray, rng: np.random.Generator) -> FloatArray: + centers = np.empty((self.n_clusters, X.shape[1]), dtype=float) + first_idx = rng.integers(X.shape[0]) + centers[0] = X[first_idx] + + distances = np.full(X.shape[0], np.inf, dtype=float) + for i in range(1, self.n_clusters): + squared_distances = np.sum((X - centers[i - 1]) ** 2, axis=1) + distances = np.minimum(distances, squared_distances) + probabilities = distances / distances.sum() + cumulative = np.cumsum(probabilities) + chosen = rng.random() + centers[i] = X[np.searchsorted(cumulative, chosen)] + + return centers + + def _assign_clusters(self, X: FloatArray, centers: FloatArray) -> NDArray[np.int64]: + distances = np.linalg.norm(X[:, np.newaxis, :] - centers[np.newaxis, :, :], axis=2) + return np.argmin(distances, axis=1).astype(np.int64) + + def _compute_centers( + self, X: FloatArray, labels: NDArray[np.int64], n_features: int + ) -> FloatArray: + centers = np.zeros((self.n_clusters, n_features), dtype=float) + for cluster_index in range(self.n_clusters): + members = X[labels == cluster_index] + if members.size == 0: + centers[cluster_index] = X[np.random.default_rng(self.random_state).integers(X.shape[0])] + else: + centers[cluster_index] = members.mean(axis=0) + return centers + + def _compute_inertia( + self, X: FloatArray, centers: FloatArray, labels: NDArray[np.int64] + ) -> float: + diff = X - centers[labels] + return float(np.sum(diff ** 2)) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..3c3a00d --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,50 @@ +"""Shared fixtures for all test suites.""" + +import numpy as np +import pytest + + +@pytest.fixture(scope="session") +def rng(): + """Seeded RNG for reproducible random data.""" + return np.random.default_rng(42) + + +@pytest.fixture +def linear_dataset(rng): + """Simple linear regression dataset: y = 2x1 + 3x2 + noise.""" + n = 200 + X = rng.standard_normal((n, 2)) + y = 2.0 * X[:, 0] + 3.0 * X[:, 1] + rng.standard_normal(n) * 0.1 + return X, y + + +@pytest.fixture +def binary_classification_dataset(rng): + """2-class, 2-feature dataset for classifier tests.""" + from sklearn.datasets import make_classification + + X, y = make_classification( + n_samples=300, + n_features=4, + n_informative=2, + n_redundant=0, + random_state=42, + ) + return X.astype(float), y.astype(float) + + +@pytest.fixture +def multiclass_dataset(rng): + """3-class dataset for softmax / multiclass tests.""" + from sklearn.datasets import make_classification + + X, y = make_classification( + n_samples=400, + n_features=4, + n_classes=3, + n_informative=3, + n_redundant=0, + random_state=42, + ) + return X.astype(float), y diff --git a/tests/supervised/test_decision_tree.py b/tests/supervised/test_decision_tree.py new file mode 100644 index 0000000..973a7a2 --- /dev/null +++ b/tests/supervised/test_decision_tree.py @@ -0,0 +1,51 @@ +import numpy as np +import pytest + +from mlscratch.supervised.decision_tree import DecisionTreeClassifier + + +def test_decision_tree_accuracy_on_iris(): + """Decision tree should classify Iris with ≥95% accuracy.""" + from sklearn.datasets import load_iris + + data = load_iris() + X, y = data.data, data.target + model = DecisionTreeClassifier(max_depth=4) + model.fit(X, y) + assert model.score(X, y) >= 0.95 + + +def test_decision_tree_agrees_with_sklearn(): + """Decision tree predictions should match sklearn DecisionTreeClassifier within 2%.""" + from sklearn.datasets import make_classification + from sklearn.tree import DecisionTreeClassifier as SKDecisionTree + + X, y = make_classification( + n_samples=200, + n_features=4, + n_informative=3, + n_redundant=0, + n_classes=2, + random_state=42, + ) + model = DecisionTreeClassifier(max_depth=5) + model.fit(X, y) + ours = model.predict(X) + + theirs = SKDecisionTree(max_depth=5, random_state=42) + theirs.fit(X, y) + assert np.mean(ours == theirs.predict(X)) >= 0.98 + + +def test_decision_tree_predict_before_fit_raises(): + """predict() before fit() must raise RuntimeError.""" + model = DecisionTreeClassifier(max_depth=3) + with pytest.raises(RuntimeError, match="fit"): + model.predict(np.ones((5, 2))) + + +def test_decision_tree_shape_mismatch_raises(): + """Mismatched X and y shapes must raise ValueError.""" + model = DecisionTreeClassifier(max_depth=3) + with pytest.raises(ValueError, match="samples"): + model.fit(np.ones((10, 2)), np.ones(5)) diff --git a/tests/supervised/test_knn.py b/tests/supervised/test_knn.py new file mode 100644 index 0000000..26e2976 --- /dev/null +++ b/tests/supervised/test_knn.py @@ -0,0 +1,60 @@ +import numpy as np +import pytest + +from mlscratch.supervised.knn import KNeighborsClassifier + + +def test_knn_accuracy_on_separable_data(): + """KNN must achieve ≥95% accuracy on a clean binary classification task.""" + from sklearn.datasets import make_classification + + X, y = make_classification( + n_samples=300, + n_features=4, + n_informative=2, + n_redundant=0, + n_classes=2, + class_sep=2.0, + random_state=42, + ) + model = KNeighborsClassifier(n_neighbors=5).fit(X, y) + assert model.score(X, y) >= 0.95 + + +def test_knn_agrees_with_sklearn(): + """KNN labels should match sklearn's KNeighborsClassifier within 2%.""" + from sklearn.datasets import make_classification + from sklearn.neighbors import KNeighborsClassifier as SKKNN + + X, y = make_classification( + n_samples=200, + n_features=3, + n_informative=3, + n_redundant=0, + n_classes=2, + class_sep=1.5, + random_state=0, + ) + model = KNeighborsClassifier(n_neighbors=5).fit(X, y) + ours = model.predict(X) + + theirs = SKKNN(n_neighbors=5).fit(X, y) + assert np.mean(ours == theirs.predict(X)) >= 0.98 + + +def test_knn_predict_before_fit_raises(): + """predict() before fit() must raise RuntimeError.""" + model = KNeighborsClassifier(n_neighbors=3) + try: + model.predict(np.ones((5, 2))) + except RuntimeError as exc: + assert "fit" in str(exc) + else: + raise AssertionError("Expected RuntimeError for predict before fit") + + +def test_knn_invalid_input_raises(): + """Mismatched X and y shapes must raise ValueError.""" + model = KNeighborsClassifier(n_neighbors=3) + with pytest.raises(ValueError, match="samples"): + model.fit(np.ones((10, 2)), np.ones(5)) diff --git a/tests/supervised/test_lasso_regression.py b/tests/supervised/test_lasso_regression.py new file mode 100644 index 0000000..22cc8d1 --- /dev/null +++ b/tests/supervised/test_lasso_regression.py @@ -0,0 +1,50 @@ +import numpy as np +import pytest + +from mlscratch.supervised.lasso_regression import LassoRegression + + +def test_lasso_agrees_with_sklearn(): + """Lasso predictions should agree with sklearn's Lasso within tolerance.""" + from sklearn.linear_model import Lasso as SKLasso + + rng = np.random.default_rng(42) + X = rng.normal(size=(200, 5)) + true_coef = np.array([1.5, 0.0, -2.0, 0.0, 0.5]) + y = X @ true_coef + rng.normal(scale=0.1, size=200) + + model = LassoRegression(alpha=0.1, max_iter=2000, tol=1e-5) + model.fit(X, y) + ours = model.predict(X) + + theirs = SKLasso(alpha=0.1, fit_intercept=True, max_iter=10000, tol=1e-8) + theirs.fit(X, y) + np.testing.assert_allclose(ours, theirs.predict(X), rtol=1e-3, atol=1e-2) + + +def test_lasso_r2_high(): + """Lasso should fit clean data with R² above 0.95.""" + rng = np.random.default_rng(0) + X = rng.normal(size=(200, 3)) + y = X @ np.array([2.0, -1.0, 0.0]) + rng.normal(scale=0.05, size=200) + model = LassoRegression(alpha=0.05, max_iter=2000, tol=1e-5) + model.fit(X, y) + assert model.score(X, y) > 0.95 + + +def test_lasso_predict_before_fit_raises(): + """predict() before fit() must raise RuntimeError.""" + model = LassoRegression(alpha=0.1) + try: + model.predict(np.ones((5, 2))) + except RuntimeError as exc: + assert "fit" in str(exc) + else: + raise AssertionError("Expected RuntimeError for predict before fit") + + +def test_lasso_shape_mismatch_raises(): + """Mismatched X and y shapes must raise ValueError.""" + model = LassoRegression(alpha=0.1) + with pytest.raises(ValueError, match="samples"): + model.fit(np.ones((10, 2)), np.ones(5)) diff --git a/tests/supervised/test_linear_regression.py b/tests/supervised/test_linear_regression.py new file mode 100644 index 0000000..95c9df7 --- /dev/null +++ b/tests/supervised/test_linear_regression.py @@ -0,0 +1,63 @@ +import numpy as np +import pytest + +from mlscratch.supervised.linear_regression import ( + GradientDescentRegressor, + OrdinaryLeastSquares, +) + + +def test_ols_coef_recovery(linear_dataset): + """OLS must recover true coefficients [2.0, 3.0] to within 0.1.""" + X, y = linear_dataset + model = OrdinaryLeastSquares() + model.fit(X, y) + np.testing.assert_allclose(model.coef_, [2.0, 3.0], atol=0.1) + + +def test_ols_agrees_with_sklearn(linear_dataset): + """OLS predictions must match sklearn LinearRegression within 1e-6.""" + from sklearn.linear_model import LinearRegression as SKLearn + + X, y = linear_dataset + ours = OrdinaryLeastSquares().fit(X, y) + theirs = SKLearn().fit(X, y) + np.testing.assert_allclose(ours.predict(X), theirs.predict(X), rtol=1e-6) + + +def test_ols_r2_high(linear_dataset): + """R² on clean linear data must be > 0.99.""" + X, y = linear_dataset + r2 = OrdinaryLeastSquares().fit(X, y).score(X, y) + assert r2 > 0.99 + + +def test_gradient_descent_converges(linear_dataset): + """GD loss must decrease monotonically for 50+ consecutive epochs.""" + X, y = linear_dataset + model = GradientDescentRegressor(n_epochs=200, learning_rate=0.01) + model.fit(X, y) + losses = model.loss_history_ + diffs = np.diff(losses[-50:]) + assert (diffs <= 1e-6).all(), "Loss not monotonically decreasing" + + +def test_gd_agrees_with_ols(linear_dataset): + """After sufficient epochs, GD predictions approximate OLS within 5%.""" + X, y = linear_dataset + ols_pred = OrdinaryLeastSquares().fit(X, y).predict(X) + gd_pred = GradientDescentRegressor(n_epochs=5000, learning_rate=0.005).fit(X, y).predict(X) + np.testing.assert_allclose(gd_pred, ols_pred, rtol=0.05, atol=1e-2) + + +def test_shape_mismatch_raises(): + """Mismatched X and y shapes must raise ValueError.""" + model = OrdinaryLeastSquares() + with pytest.raises(ValueError, match="samples"): + model.fit(np.ones((10, 2)), np.ones(5)) + + +def test_predict_before_fit_raises(): + """predict() before fit() must raise RuntimeError.""" + with pytest.raises(RuntimeError, match="fit"): + OrdinaryLeastSquares().predict(np.ones((5, 2))) diff --git a/tests/supervised/test_logistic_regression.py b/tests/supervised/test_logistic_regression.py new file mode 100644 index 0000000..cfe2aee --- /dev/null +++ b/tests/supervised/test_logistic_regression.py @@ -0,0 +1,53 @@ +import numpy as np + +from mlscratch.supervised.logistic_regression import LogisticRegression + + +def test_logistic_regression_learns_binary_separation(): + """LogisticRegression must achieve ≥95% accuracy on clean binary data.""" + rng = np.random.default_rng(42) + X = np.vstack([ + rng.normal(loc=-2.0, scale=0.5, size=(100, 2)), + rng.normal(loc=2.0, scale=0.5, size=(100, 2)), + ]) + y = np.concatenate([np.zeros(100), np.ones(100)]) + + model = LogisticRegression(n_epochs=2000, learning_rate=0.1, batch_size=32, random_state=42) + model.fit(X, y) + accuracy = model.score(X, y) + assert accuracy >= 0.95 + + +def test_logistic_regression_predict_proba_shape(): + """predict_proba must return a probability for each sample.""" + rng = np.random.default_rng(0) + X = rng.standard_normal((10, 3)) + y = np.zeros(10) + model = LogisticRegression(n_epochs=1) + model.fit(X, y) + probs = model.predict_proba(X) + assert probs.shape == (10,) + assert np.all((probs >= 0.0) & (probs <= 1.0)) + + +def test_logistic_regression_score_matches_accuracy(): + """score() must equal the fraction of correct binary predictions.""" + rng = np.random.default_rng(1) + X = rng.normal(size=(20, 2)) + y = (X[:, 0] + X[:, 1] > 0).astype(float) + model = LogisticRegression(n_epochs=1000, learning_rate=0.05, batch_size=10, random_state=1) + model.fit(X, y) + assert model.score(X, y) == np.mean(model.predict(X) == y) + + +def test_logistic_regression_invalid_labels_raise(): + """Non-binary labels should raise a ValueError.""" + X = np.ones((5, 2)) + y = np.array([0, 1, 2, 0, 1]) + model = LogisticRegression(n_epochs=1) + try: + model.fit(X, y) + except ValueError as exc: + assert "binary labels" in str(exc) + else: + raise AssertionError("Expected ValueError for non-binary labels") diff --git a/tests/supervised/test_naive_bayes.py b/tests/supervised/test_naive_bayes.py new file mode 100644 index 0000000..7ab4034 --- /dev/null +++ b/tests/supervised/test_naive_bayes.py @@ -0,0 +1,39 @@ +import numpy as np +import pytest + +from mlscratch.supervised.naive_bayes import GaussianNB + + +def test_gaussian_nb_agrees_with_sklearn(): + """GaussianNB predictions should match sklearn's GaussianNB on the same data.""" + from sklearn.datasets import make_classification + from sklearn.naive_bayes import GaussianNB as SKGaussianNB + + X, y = make_classification( + n_samples=250, + n_features=4, + n_informative=4, + n_redundant=0, + n_classes=3, + random_state=0, + ) + + model = GaussianNB().fit(X, y) + theirs = SKGaussianNB().fit(X, y) + + assert model.score(X, y) == pytest.approx(theirs.score(X, y), rel=1e-6) + assert np.mean(model.predict(X) == theirs.predict(X)) == pytest.approx(1.0, rel=1e-6) + + +def test_gaussian_nb_predict_before_fit_raises(): + """Predict before fit must raise RuntimeError.""" + model = GaussianNB() + with pytest.raises(RuntimeError, match="fit"): + model.predict(np.ones((5, 2))) + + +def test_gaussian_nb_invalid_input_raises(): + """Mismatched X and y shapes must raise ValueError.""" + model = GaussianNB() + with pytest.raises(ValueError, match="samples"): + model.fit(np.ones((10, 2)), np.ones(5)) diff --git a/tests/supervised/test_random_forest.py b/tests/supervised/test_random_forest.py new file mode 100644 index 0000000..ea9b3ed --- /dev/null +++ b/tests/supervised/test_random_forest.py @@ -0,0 +1,58 @@ +import numpy as np +import pytest + +from mlscratch.supervised.random_forest import RandomForestClassifier + + +def test_random_forest_accuracy_on_clean_data(): + """Random forest must fit clean binary classification data accurately.""" + from sklearn.datasets import make_classification + + X, y = make_classification( + n_samples=300, + n_features=5, + n_informative=3, + n_redundant=0, + n_classes=2, + class_sep=2.0, + random_state=42, + ) + + model = RandomForestClassifier(n_estimators=25, random_state=42).fit(X, y) + assert model.score(X, y) >= 0.95 + + +def test_random_forest_agrees_with_sklearn(): + """Predictions should agree with sklearn's RandomForestClassifier on the same data.""" + from sklearn.datasets import make_classification + from sklearn.ensemble import RandomForestClassifier as SKRF + + X, y = make_classification( + n_samples=200, + n_features=6, + n_informative=4, + n_redundant=0, + n_classes=2, + class_sep=1.5, + random_state=0, + ) + + model = RandomForestClassifier(n_estimators=20, random_state=0).fit(X, y) + ours = model.predict(X) + + theirs = SKRF(n_estimators=20, random_state=0).fit(X, y) + assert np.mean(ours == theirs.predict(X)) >= 0.85 + + +def test_random_forest_predict_before_fit_raises(): + """predict() must raise RuntimeError if fit() was not called.""" + model = RandomForestClassifier(n_estimators=10) + with pytest.raises(RuntimeError, match="fit"): + model.predict(np.ones((5, 3))) + + +def test_random_forest_invalid_input_raises(): + """Mismatched X and y shapes must raise ValueError.""" + model = RandomForestClassifier(n_estimators=5) + with pytest.raises(ValueError, match="samples"): + model.fit(np.ones((10, 2)), np.ones(5)) diff --git a/tests/supervised/test_ridge_regression.py b/tests/supervised/test_ridge_regression.py new file mode 100644 index 0000000..26abeb7 --- /dev/null +++ b/tests/supervised/test_ridge_regression.py @@ -0,0 +1,49 @@ +import numpy as np +import pytest + +from mlscratch.supervised.ridge_regression import RidgeRegression + + +def test_ridge_agrees_with_sklearn(): + """Ridge predictions should agree with sklearn's Ridge within tolerance.""" + from sklearn.linear_model import Ridge as SKRidge + + rng = np.random.default_rng(42) + X = rng.normal(size=(200, 4)) + y = X @ np.array([1.2, -0.8, 0.0, 0.5]) + rng.normal(scale=0.1, size=200) + + model = RidgeRegression(alpha=0.5) + model.fit(X, y) + ours = model.predict(X) + + theirs = SKRidge(alpha=0.5, fit_intercept=True, solver="auto", max_iter=10000) + theirs.fit(X, y) + np.testing.assert_allclose(ours, theirs.predict(X), rtol=1e-5, atol=1e-2) + + +def test_ridge_r2_high(): + """Ridge should score >0.99 on clean linear data.""" + rng = np.random.default_rng(0) + X = rng.normal(size=(300, 2)) + y = X @ np.array([2.0, -1.5]) + rng.normal(scale=0.05, size=300) + model = RidgeRegression(alpha=0.1) + model.fit(X, y) + assert model.score(X, y) > 0.99 + + +def test_ridge_predict_before_fit_raises(): + """predict() before fit() must raise RuntimeError.""" + model = RidgeRegression(alpha=0.5) + try: + model.predict(np.ones((5, 2))) + except RuntimeError as exc: + assert "fit" in str(exc) + else: + raise AssertionError("Expected RuntimeError for predict before fit") + + +def test_ridge_shape_mismatch_raises(): + """Mismatched X and y shapes must raise ValueError.""" + model = RidgeRegression(alpha=0.5) + with pytest.raises(ValueError, match="samples"): + model.fit(np.ones((10, 2)), np.ones(5)) diff --git a/tests/supervised/test_svm.py b/tests/supervised/test_svm.py new file mode 100644 index 0000000..ce2da82 --- /dev/null +++ b/tests/supervised/test_svm.py @@ -0,0 +1,64 @@ +import numpy as np +import pytest + +from mlscratch.supervised.svm import LinearSVMClassifier + + +def test_linear_svm_accuracy_on_separable_data(): + """LinearSVMClassifier should fit a clean binary classification task.""" + from sklearn.datasets import make_classification + + X, y = make_classification( + n_samples=250, + n_features=4, + n_informative=3, + n_redundant=0, + n_classes=2, + class_sep=2.0, + random_state=42, + ) + + model = LinearSVMClassifier(learning_rate=1e-3, n_epochs=1000, C=1.0, random_state=42).fit(X, y) + assert model.score(X, y) >= 0.90 + + +def test_linear_svm_agrees_with_sklearn_hinge(): + """Linear SVM should agree with sklearn's hinge-loss SGDClassifier.""" + from sklearn.datasets import make_classification + from sklearn.linear_model import SGDClassifier + + X, y = make_classification( + n_samples=220, + n_features=4, + n_informative=3, + n_redundant=0, + n_classes=2, + class_sep=1.8, + random_state=1, + ) + + model = LinearSVMClassifier(learning_rate=1e-3, n_epochs=1000, C=1.0, random_state=1).fit(X, y) + theirs = SGDClassifier( + loss="hinge", + alpha=1.0 / (model.C * X.shape[0]), + learning_rate="constant", + eta0=1e-3, + max_iter=1000, + tol=None, + random_state=1, + ).fit(X, y) + assert np.mean(model.predict(X) == theirs.predict(X)) >= 0.85 + + +def test_linear_svm_predict_before_fit_raises(): + """Predict before fit must raise RuntimeError.""" + model = LinearSVMClassifier() + with pytest.raises(RuntimeError, match="fit"): + model.predict(np.ones((5, 2))) + + +def test_linear_svm_invalid_input_raises(): + """Mismatched X and y shapes must raise ValueError.""" + model = LinearSVMClassifier() + with pytest.raises(ValueError, match="samples"): + model.fit(np.ones((10, 2)), np.ones(5)) diff --git a/tests/unsupervised/test_kmeans.py b/tests/unsupervised/test_kmeans.py new file mode 100644 index 0000000..d137fde --- /dev/null +++ b/tests/unsupervised/test_kmeans.py @@ -0,0 +1,52 @@ +import numpy as np +import pytest + +from mlscratch.unsupervised.kmeans import KMeans + + +def _best_label_mapping(true_labels, predicted_labels): + from scipy.optimize import linear_sum_assignment + from sklearn.metrics import confusion_matrix + + matrix = confusion_matrix(true_labels, predicted_labels) + row_ind, col_ind = linear_sum_assignment(matrix.max() - matrix) + new_labels = np.zeros_like(predicted_labels) + for true_label, pred_label in zip(row_ind, col_ind): + new_labels[predicted_labels == pred_label] = true_label + return new_labels + + +def test_kmeans_recovers_blobs(): + """KMeans should recover well-separated clusters from blob data.""" + from sklearn.datasets import make_blobs + from sklearn.cluster import KMeans as SKKMeans + + X, y = make_blobs( + n_samples=240, + centers=3, + cluster_std=0.45, + random_state=0, + ) + + model = KMeans(n_clusters=3, random_state=0).fit(X) + assert model.cluster_centers_.shape == (3, X.shape[1]) + assert model.labels_.shape == (X.shape[0],) + + theirs = SKKMeans(n_clusters=3, random_state=0, init="k-means++", n_init=1, max_iter=300).fit(X) + mapped = _best_label_mapping(y, model.labels_) + assert np.mean(mapped == y) >= 0.90 + assert model.inertia_ <= 1.2 * theirs.inertia_ + + +def test_kmeans_predict_before_fit_raises(): + """predict() before fit() must raise RuntimeError.""" + model = KMeans(n_clusters=2) + with pytest.raises(RuntimeError, match="fit"): + model.predict(np.ones((5, 2))) + + +def test_kmeans_invalid_input_raises(): + """Invalid input shape must raise ValueError.""" + model = KMeans(n_clusters=2) + with pytest.raises(ValueError, match="2D"): + model.fit(np.ones((5,)))