{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# \ud83e\udde0 ML LeetCode Interview Guide: Data Structures & Algorithms in Pure Python\n", "### Core ML, Probability, and Systems Problems Implemented from Scratch\n", "\n", "This notebook covers **LeetCode-style algorithmic questions** tailored for Machine Learning Engineer (MLE) and Applied Scientist interviews. Unlike classical model implementation rounds, these questions evaluate **algorithmic thinking, data structure selection, numerical edge cases, and time/space complexity** in **pure Python** (using only standard libraries: `math`, `random`, `bisect`, `heapq`, `collections`).\n", "\n", "---\n", "\n", "## \ud83d\udccc Table of Contents\n", "1. [Weighted Random Sampling (Prefix Sums & Binary Search)](#1-weighted-random-sampling)\n", "2. [Reservoir Sampling (Uniform Streaming with $\\mathcal{O}(1)$ Memory)](#2-reservoir-sampling)\n", "3. [Sort Transformed Array (Convex Quadratic & Two Pointers)](#3-sort-transformed-array)\n", "4. [Find Minimum of a Convex Function (Ternary Search)](#4-find-minimum-of-a-convex-function)\n", "5. [Expected Calibration Error - ECE (Bucket Binning)](#5-expected-calibration-error-ece)\n", "6. [Exact ROC-AUC with Tie Handling (Wilcoxon Rank-Sum)](#6-exact-roc-auc-with-tie-handling)\n", "7. [Precision-Recall Curve & Average Precision (Sweep-Line)](#7-precision-recall-curve--average-precision)\n", "8. [Dot Product of Two Sparse Vectors (Two Pointers & Binary Search)](#8-dot-product-of-two-sparse-vectors)\n", "9. [Top-K Cosine Similarity Retrieval (Min-Heap with `heapq`)](#9-top-k-cosine-similarity-retrieval)\n", "10. [TF-IDF & Inverted Index Document Retrieval](#10-tf-idf--inverted-index-document-retrieval)\n", "11. [Streaming Running Median (Two Heaps)](#11-streaming-running-median)\n", "12. [Edit Distance & Word Error Rate - WER (Dynamic Programming)](#12-edit-distance--word-error-rate-wer)\n", "13. [Image Rotation by Arbitrary Angle (Inverse Mapping & Bilinear Interpolation)](#13-image-rotation-by-arbitrary-angle)\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 1. Weighted Random Sampling\n", "**LeetCode Analog**: [LC 528: Random Pick with Weight](https://leetcode.com/problems/random-pick-with-weight/) \n", "**ML Context**: Sampling tokens with temperature in LLMs, categorical distribution sampling, AdaBoost sample reweighting.\n", "\n", "### Algorithmic Principle\n", "Given positive weights $w = [w_0, w_1, \\dots, w_{N-1}]$:\n", "1. Construct cumulative prefix sums: $S_i = \\sum_{j=0}^i w_j$.\n", "2. Draw a uniform random float $u \\in [0, S_{N-1})$.\n", "3. Use binary search (`bisect.bisect_right`) to find the smallest index $i$ where $u < S_i$.\n", "\n", "- **Time Complexity**: $\\mathcal{O}(N)$ initialization, $\\mathcal{O}(\\log N)$ per sample.\n", "- **Space Complexity**: $\\mathcal{O}(N)$ for prefix sums.\n", "- **Interview Trap**: Using linear scan $\\mathcal{O}(N)$ instead of binary search $\\mathcal{O}(\\log N)$; handling floating-point roundoff at boundary $S_{N-1}$.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import random\n", "import bisect\n", "from collections import Counter\n", "from typing import List\n", "\n", "class WeightedRandomSampler:\n", " \"\"\"\n", " Weighted random sampling using Prefix Sums + Binary Search.\n", " Time: O(N) init, O(log N) sample. Space: O(N).\n", " \"\"\"\n", " def __init__(self, weights: List[float]):\n", " if not weights or any(w < 0 for w in weights):\n", " raise ValueError(\"Weights must be non-empty and non-negative.\")\n", " \n", " self.prefix_sums: List[float] = []\n", " running_sum = 0.0\n", " for w in weights:\n", " running_sum += w\n", " self.prefix_sums.append(running_sum)\n", " self.total_weight = running_sum\n", "\n", " def sample(self) -> int:\n", " target = random.random() * self.total_weight\n", " # bisect_right finds the first prefix_sum strictly greater than target\n", " return bisect.bisect_right(self.prefix_sums, target)\n", "\n", "# Verification Test\n", "random.seed(42)\n", "weights = [1.0, 3.0, 6.0] # Expected: ~10%, ~30%, ~60%\n", "sampler = WeightedRandomSampler(weights)\n", "trials = 10000\n", "samples = [sampler.sample() for _ in range(trials)]\n", "counts = Counter(samples)\n", "\n", "print(\"Empirical distribution over 10,000 samples:\")\n", "for idx, weight in enumerate(weights):\n", " prob = counts[idx] / trials\n", " expected = weight / sum(weights)\n", " print(f\" Index {idx}: observed = {prob:.4f}, expected = {expected:.4f}\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 2. Reservoir Sampling\n", "**LeetCode Analog**: [LC 382: Linked List Random Node](https://leetcode.com/problems/linked-list-random-node/), [LC 398: Random Pick Index](https://leetcode.com/problems/random-pick-index/) \n", "**ML Context**: Sampling from infinite/unknown streaming data pipelines, replay buffers in reinforcement learning without loading full stream into memory.\n", "\n", "### Algorithmic Principle\n", "To sample $k$ items uniformly from a stream of unknown length $N$:\n", "1. Store the first $k$ elements into the reservoir.\n", "2. For each incoming $i$-th element ($i \\ge k$, 0-indexed):\n", " - Pick a random integer $j \\in [0, i]$.\n", " - If $j < k$, replace `reservoir[j]` with the new element.\n", "3. Every element seen so far has an exact $\\frac{k}{i+1}$ probability of being in the reservoir.\n", "\n", "- **Time Complexity**: $\\mathcal{O}(N)$ stream scan.\n", "- **Space Complexity**: $\\mathcal{O}(k)$ auxiliary memory.\n", "- **Interview Trap**: Random range off-by-one errors (must pick $j \\in [0, i]$ inclusive).\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from typing import List, Iterable, Any\n", "import random\n", "from collections import Counter\n", "\n", "def reservoir_sample(stream: Iterable[Any], k: int) -> List[Any]:\n", " \"\"\"\n", " Uniformly samples k items from an incoming stream of unknown length.\n", " Time: O(N), Space: O(k).\n", " \"\"\"\n", " if k <= 0:\n", " return []\n", " \n", " reservoir = []\n", " for i, item in enumerate(stream):\n", " if i < k:\n", " reservoir.append(item)\n", " else:\n", " j = random.randint(0, i)\n", " if j < k:\n", " reservoir[j] = item\n", " return reservoir\n", "\n", "# Verification Test: Monte Carlo uniformity check\n", "random.seed(42)\n", "stream_items = list(range(10)) # Stream items: 0..9\n", "k = 2\n", "trials = 10000\n", "freq = Counter()\n", "\n", "for _ in range(trials):\n", " sample = reservoir_sample(stream_items, k)\n", " freq.update(sample)\n", "\n", "print(f\"Sample frequency over {trials} trials (expected ~{trials * (k / 10):.0f} per item):\")\n", "for item in sorted(freq.keys()):\n", " print(f\" Item {item}: count = {freq[item]}\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 3. Sort Transformed Array under Convex Quadratic Function\n", "**LeetCode Analog**: [LC 360: Sort Transformed Array](https://leetcode.com/problems/sort-transformed-array/) \n", "**ML Context**: Evaluating convex/concave quadratic losses ($f(x) = ax^2 + bx + c$) over sorted inputs; exploiting convexity to avoid $\\mathcal{O}(N \\log N)$ sorting.\n", "\n", "### Algorithmic Principle\n", "A quadratic function $f(x) = ax^2 + bx + c$ has a single extremum (vertex at $x = -b / (2a)$):\n", "- **Convex ($a > 0$)**: Parabola opens upward; maximums occur at the outer boundaries. Use two pointers (`left=0`, `right=N-1`) and fill the result array **from back to front**.\n", "- **Concave ($a < 0$)**: Parabola opens downward; minimums occur at the outer boundaries. Fill the result array **from front to back**.\n", "- **Linear ($a = 0$)**: If $b \\ge 0$, array is already sorted; if $b < 0$, reverse array.\n", "\n", "- **Time Complexity**: $\\mathcal{O}(N)$ linear time.\n", "- **Space Complexity**: $\\mathcal{O}(N)$ for output array.\n", "- **Interview Trap**: Forgetting to handle $a < 0$ vs $a > 0$ pointer fill directions.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from typing import List\n", "\n", "def sort_transformed_array(nums: List[int], a: int, b: int, c: int) -> List[int]:\n", " \"\"\"\n", " Transforms sorted nums by f(x) = a*x^2 + b*x + c and returns sorted list in O(N).\n", " Time: O(N), Space: O(N).\n", " \"\"\"\n", " def f(x: int) -> int:\n", " return a * x * x + b * x + c\n", "\n", " n = len(nums)\n", " res = [0] * n\n", " left, right = 0, n - 1\n", "\n", " if a >= 0:\n", " # Convex: fill largest values from back to front\n", " idx = n - 1\n", " while left <= right:\n", " fl, fr = f(nums[left]), f(nums[right])\n", " if fl >= fr:\n", " res[idx] = fl\n", " left += 1\n", " else:\n", " res[idx] = fr\n", " right -= 1\n", " idx -= 1\n", " else:\n", " # Concave: fill smallest values from front to back\n", " idx = 0\n", " while left <= right:\n", " fl, fr = f(nums[left]), f(nums[right])\n", " if fl <= fr:\n", " res[idx] = fl\n", " left += 1\n", " else:\n", " res[idx] = fr\n", " right -= 1\n", " idx += 1\n", "\n", " return res\n", "\n", "# Verification Test\n", "nums = [-4, -2, 2, 4]\n", "# Convex case (a = 1 > 0)\n", "res_convex = sort_transformed_array(nums, a=1, b=3, c=5)\n", "print(\"Convex transformed (a=1, b=3, c=5):\", res_convex)\n", "assert res_convex == sorted([1*x**2 + 3*x + 5 for x in nums])\n", "\n", "# Concave case (a = -1 < 0)\n", "res_concave = sort_transformed_array(nums, a=-1, b=3, c=5)\n", "print(\"Concave transformed (a=-1, b=3, c=5):\", res_concave)\n", "assert res_concave == sorted([-1*x**2 + 3*x + 5 for x in nums])\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 4. Find Minimum of a Convex Function (Ternary Search)\n", "**LeetCode Analog**: [LC 852: Peak Index in a Mountain Array](https://leetcode.com/problems/peak-index-in-a-mountain-array/) (Continuous variant) \n", "**ML Context**: Derivative-free 1D line search, tuning continuous hyperparameters with unimodal validation loss (e.g. learning rate, regularization weight).\n", "\n", "### Algorithmic Principle\n", "For a strictly unimodal function $f(x)$ on $[a, b]$:\n", "1. Divide $[a, b]$ into three equal segments using two midpoints:\n", " $$m_1 = a + \\frac{b - a}{3}, \\quad m_2 = b - \\frac{b - a}{3}$$\n", "2. If $f(m_1) < f(m_2)$, the minimum cannot lie in $(m_2, b]$; set $b = m_2$.\n", "3. Otherwise, the minimum cannot lie in $[a, m_1)$; set $a = m_1$.\n", "4. Repeat until $|b - a| < \\epsilon$.\n", "\n", "- **Time Complexity**: $\\mathcal{O}(\\log((b - a)/\\epsilon))$.\n", "- **Space Complexity**: $\\mathcal{O}(1)$.\n", "- **Interview Trap**: Confusing strict unimodality with local plateau regions where ternary search can get stuck.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from typing import Callable\n", "\n", "def ternary_search_min(f: Callable[[float], float], a: float, b: float, tol: float = 1e-7) -> float:\n", " \"\"\"\n", " Finds x that minimizes unimodal function f on [a, b].\n", " Time: O(log((b - a) / tol)), Space: O(1).\n", " \"\"\"\n", " left, right = a, b\n", " while right - left > tol:\n", " m1 = left + (right - left) / 3.0\n", " m2 = right - (right - left) / 3.0\n", " if f(m1) < f(m2):\n", " right = m2\n", " else:\n", " left = m1\n", " return (left + right) / 2.0\n", "\n", "# Verification Test: Minimize f(x) = (x - 3.5)^2 + 10 (minimum at x = 3.5)\n", "f = lambda x: (x - 3.5)**2 + 10.0\n", "x_min = ternary_search_min(f, a=0.0, b=10.0)\n", "\n", "print(f\"Optimal x found: {x_min:.6f} (True minimum = 3.500000)\")\n", "print(f\"Function value at minimum: {f(x_min):.6f}\")\n", "assert abs(x_min - 3.5) < 1e-5\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 5. Expected Calibration Error (ECE)\n", "**ML Context**: Model calibration. A well-calibrated classifier satisfies $\\mathbb{P}(\\hat{y} = y \\mid \\hat{p} = p) = p$. Required for confidence estimation, Platt scaling, and temperature scaling.\n", "\n", "### Algorithmic Principle\n", "1. Group $N$ predictions into $M$ equal-width bins $B_1, \\dots, B_M$ over $[0, 1]$.\n", "2. For each bin $B_m$ with $|B_m| > 0$:\n", " $$\\text{acc}(B_m) = \\frac{1}{|B_m|} \\sum_{i \\in B_m} \\mathbb{I}(y_i = 1), \\quad \\text{conf}(B_m) = \\frac{1}{|B_m|} \\sum_{i \\in B_m} p_i$$\n", "3. Compute weighted calibration metrics:\n", " $$\\text{ECE} = \\sum_{m=1}^M \\frac{|B_m|}{N} |\\text{acc}(B_m) - \\text{conf}(B_m)|, \\quad \\text{MCE} = \\max_{m} |\\text{acc}(B_m) - \\text{conf}(B_m)|$$\n", "\n", "- **Time Complexity**: $\\mathcal{O}(N + M)$ single pass binning.\n", "- **Space Complexity**: $\\mathcal{O}(M)$ bin accumulators.\n", "- **Interview Trap**: Edge case $p_i = 1.0$ mapping to index $M$ (clamp to $M - 1$).\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from typing import List, Dict\n", "\n", "def compute_ece(probs: List[float], labels: List[int], n_bins: int = 10) -> Dict[str, float]:\n", " \"\"\"\n", " Computes Expected Calibration Error (ECE) and Maximum Calibration Error (MCE).\n", " Time: O(N + M), Space: O(M).\n", " \"\"\"\n", " if len(probs) != len(labels):\n", " raise ValueError(\"probs and labels must have the same length.\")\n", " \n", " bin_counts = [0] * n_bins\n", " bin_correct = [0] * n_bins\n", " bin_conf_sum = [0.0] * n_bins\n", "\n", " for p, y in zip(probs, labels):\n", " # Map probability to bin [0, n_bins-1] with boundary clamping\n", " b_idx = min(int(p * n_bins), n_bins - 1)\n", " bin_counts[b_idx] += 1\n", " bin_correct[b_idx] += (1 if y == 1 else 0)\n", " bin_conf_sum[b_idx] += p\n", "\n", " total_samples = len(probs)\n", " ece = 0.0\n", " mce = 0.0\n", "\n", " print(\"Bin-level statistics:\")\n", " for i in range(n_bins):\n", " if bin_counts[i] > 0:\n", " acc = bin_correct[i] / bin_counts[i]\n", " conf = bin_conf_sum[i] / bin_counts[i]\n", " gap = abs(acc - conf)\n", " ece += (bin_counts[i] / total_samples) * gap\n", " mce = max(mce, gap)\n", " bin_range = f\"[{i/n_bins:.1f}, {(i+1)/n_bins:.1f})\"\n", " print(f\" Bin {bin_range}: Count={bin_counts[i]}, Acc={acc:.3f}, Conf={conf:.3f}, Gap={gap:.3f}\")\n", "\n", " return {\"ece\": ece, \"mce\": mce}\n", "\n", "# Verification Test\n", "probs = [0.1, 0.15, 0.2, 0.75, 0.8, 0.95]\n", "labels = [0, 0, 1, 1, 1, 1]\n", "res = compute_ece(probs, labels, n_bins=5)\n", "print(f\"\\nECE: {res['ece']:.4f}, MCE: {res['mce']:.4f}\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 6. Exact ROC-AUC with Tie Handling\n", "**ML Context**: Ranking and classification evaluation. Measures $P(\\hat{p}_+ > \\hat{p}_-)$ \u2014 probability that a randomly chosen positive scores higher than a randomly chosen negative.\n", "\n", "### Algorithmic Principle\n", "Using the Wilcoxon-Mann-Whitney rank-sum formulation:\n", "1. Sort pairs $(p_i, y_i)$ ascending by score.\n", "2. Group tied prediction scores and assign them their **average rank**.\n", "3. Compute sum of positive ranks $R_+ = \\sum_{i \\in \\text{pos}} \\text{rank}_i$.\n", "4. Calculate AUC directly:\n", " $$\\text{AUC} = \\frac{R_+ - \\frac{n_+ (n_+ + 1)}{2}}{n_+ \\cdot n_-}$$\n", "\n", "- **Time Complexity**: $\\mathcal{O}(N \\log N)$ to sort, $\\mathcal{O}(N)$ rank sweep.\n", "- **Space Complexity**: $\\mathcal{O}(N)$.\n", "- **Interview Trap**: Naive rank assignment inflates/deflates AUC when multiple candidates have identical prediction scores.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from typing import List\n", "\n", "def compute_roc_auc(probs: List[float], labels: List[int]) -> float:\n", " \"\"\"\n", " Computes exact ROC-AUC with Wilcoxon rank-sum and average-rank tie handling.\n", " Time: O(N log N), Space: O(N).\n", " \"\"\"\n", " if len(probs) != len(labels):\n", " raise ValueError(\"Length mismatch between probs and labels.\")\n", " \n", " n_pos = sum(labels)\n", " n_neg = len(labels) - n_pos\n", " if n_pos == 0 or n_neg == 0:\n", " raise ValueError(\"ROC-AUC requires both positive and negative classes.\")\n", "\n", " # Sort ascending by predicted score\n", " indexed = sorted(enumerate(zip(probs, labels)), key=lambda x: x[1][0])\n", " \n", " rank_sum_pos = 0.0\n", " i = 0\n", " n = len(indexed)\n", " \n", " # Sweep through items, handling tied intervals [i, j)\n", " while i < n:\n", " j = i\n", " while j < n and indexed[j][1][0] == indexed[i][1][0]:\n", " j += 1\n", " \n", " # 1-based ranks from (i + 1) to j have average rank:\n", " avg_rank = ((i + 1) + j) / 2.0\n", " for k in range(i, j):\n", " if indexed[k][1][1] == 1:\n", " rank_sum_pos += avg_rank\n", " i = j\n", "\n", " auc = (rank_sum_pos - (n_pos * (n_pos + 1) / 2.0)) / (n_pos * n_neg)\n", " return auc\n", "\n", "# Verification Test\n", "# 1. Distinct scores: positives at 0.35, 0.8; negatives at 0.1, 0.4 -> pairs: 3/4 = 0.75\n", "probs = [0.1, 0.4, 0.35, 0.8]\n", "labels = [0, 0, 1, 1]\n", "print(\"AUC (distinct scores):\", compute_roc_auc(probs, labels))\n", "assert compute_roc_auc(probs, labels) == 0.75\n", "\n", "# 2. Tied scores test: random guessing with identical predictions must yield exactly 0.5\n", "probs_tied = [0.5, 0.5, 0.5, 0.5]\n", "labels_tied = [1, 0, 1, 0]\n", "print(\"AUC (tied scores):\", compute_roc_auc(probs_tied, labels_tied))\n", "assert compute_roc_auc(probs_tied, labels_tied) == 0.5\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 7. Precision-Recall Curve & Average Precision (AP)\n", "**ML Context**: Information retrieval, highly imbalanced classification, object detection benchmark (mAP).\n", "\n", "### Algorithmic Principle\n", "1. Sort pairs $(p_i, y_i)$ descending by predicted probability.\n", "2. Sweep down the list, updating running counts of $TP$ and $FP$.\n", "3. At each step $k$, compute:\n", " $$\\text{Recall}_k = \\frac{TP_k}{\\text{total\\_pos}}, \\quad \\text{Precision}_k = \\frac{TP_k}{TP_k + FP_k}$$\n", "4. Integrate area under the curve via step-wise Riemann sum:\n", " $$\\text{AP} = \\sum_k (\\text{Recall}_k - \\text{Recall}_{k-1}) \\cdot \\text{Precision}_k$$\n", "\n", "- **Time Complexity**: $\\mathcal{O}(N \\log N)$.\n", "- **Space Complexity**: $\\mathcal{O}(N)$.\n", "- **Interview Trap**: Dividing by zero when no positives are predicted at early thresholds ($TP + FP = 0$).\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from typing import List\n", "\n", "def compute_pr_auc(probs: List[float], labels: List[int]) -> float:\n", " \"\"\"\n", " Computes Average Precision (Area under Precision-Recall curve).\n", " Time: O(N log N), Space: O(N).\n", " \"\"\"\n", " total_pos = sum(labels)\n", " if total_pos == 0:\n", " return 0.0\n", " \n", " # Sort descending by predicted probability\n", " pairs = sorted(zip(probs, labels), key=lambda x: x[0], reverse=True)\n", " \n", " tp = 0\n", " fp = 0\n", " prev_recall = 0.0\n", " ap = 0.0\n", "\n", " print(\"PR curve sweep steps:\")\n", " for p, label in pairs:\n", " if label == 1:\n", " tp += 1\n", " else:\n", " fp += 1\n", " \n", " recall = tp / total_pos\n", " precision = tp / (tp + fp)\n", " # Step area contribution: delta_recall * precision\n", " delta_recall = recall - prev_recall\n", " ap += delta_recall * precision\n", " prev_recall = recall\n", " print(f\" Score={p:.2f}, Label={label} -> Precision={precision:.3f}, Recall={recall:.3f}, delta_R={delta_recall:.3f}\")\n", "\n", " return ap\n", "\n", "# Verification Test\n", "probs = [0.95, 0.85, 0.60, 0.20]\n", "labels = [1, 1, 0, 1]\n", "ap = compute_pr_auc(probs, labels)\n", "print(f\"\\nAverage Precision (PR-AUC): {ap:.4f}\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 8. Dot Product of Two Sparse Vectors\n", "**LeetCode Analog**: [LC 1570: Dot Product of Two Sparse Vectors](https://leetcode.com/problems/dot-product-of-two-sparse-vectors/) \n", "**ML Context**: Large vocabulary bag-of-words, sparse feature embeddings, recommendation systems feature cross.\n", "\n", "### Algorithmic Principle\n", "Interviewers test representation choice and complexity trade-offs:\n", "1. **Sorted Index-Value Pairs with Two Pointers**: $\\mathcal{O}(L_1 + L_2)$ time, $\\mathcal{O}(1)$ extra space, optimal cache locality.\n", "2. **Binary Search Dispatch for Length Imbalance**: When $L_1 \\ll L_2$, binary searching each index of the sparser vector into the denser vector achieves $\\mathcal{O}(L_1 \\log L_2)$.\n", "\n", "- **Time Complexity**: $\\min(L_1 + L_2, \\; L_1 \\log L_2)$ where $L = $ number of non-zero elements.\n", "- **Space Complexity**: $\\mathcal{O}(L)$ for pair storage.\n", "- **Interview Trap**: Using Hash Maps, which have higher memory overhead and hash collision variance vs cache-friendly sorted lists.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import bisect\n", "from typing import List, Tuple\n", "\n", "class SparseVector:\n", " \"\"\"\n", " Stores non-zero values as sorted (index, value) tuples.\n", " Supports Two Pointers O(L1 + L2) and Binary Search O(L1 log L2).\n", " \"\"\"\n", " def __init__(self, nums: List[int]):\n", " self.pairs: List[Tuple[int, int]] = [\n", " (i, val) for i, val in enumerate(nums) if val != 0\n", " ]\n", "\n", " def dot_product(self, other: 'SparseVector') -> int:\n", " p1, p2 = self.pairs, other.pairs\n", " l1, l2 = len(p1), len(p2)\n", " if l1 == 0 or l2 == 0:\n", " return 0\n", "\n", " # Dispatch to Binary Search if one vector is much sparser (e.g. >8x difference)\n", " if l1 * 8 < l2:\n", " return self._dot_binary_search(p1, p2)\n", " elif l2 * 8 < l1:\n", " return self._dot_binary_search(p2, p1)\n", "\n", " # Method 1: Two Pointers O(L1 + L2)\n", " res = 0\n", " i = j = 0\n", " while i < l1 and j < l2:\n", " idx1, val1 = p1[i]\n", " idx2, val2 = p2[j]\n", " if idx1 == idx2:\n", " res += val1 * val2\n", " i += 1\n", " j += 1\n", " elif idx1 < idx2:\n", " i += 1\n", " else:\n", " j += 1\n", " return res\n", "\n", " def _dot_binary_search(self, small: List[Tuple[int, int]], large: List[Tuple[int, int]]) -> int:\n", " res = 0\n", " large_indices = [idx for idx, _ in large]\n", " for idx, val in small:\n", " pos = bisect.bisect_left(large_indices, idx)\n", " if pos < len(large) and large[pos][0] == idx:\n", " res += val * large[pos][1]\n", " return res\n", "\n", "# Verification Test\n", "v1 = SparseVector([1, 0, 0, 2, 3])\n", "v2 = SparseVector([0, 3, 0, 4, 0])\n", "res = v1.dot_product(v2)\n", "print(\"Dot product (v1 @ v2):\", res)\n", "assert res == 8 # index 3: 2 * 4 = 8\n", "\n", "# Imbalanced vectors test\n", "v_sparse = SparseVector([0] * 50 + [4] + [0] * 949)\n", "v_dense = SparseVector([1] * 1000)\n", "assert v_sparse.dot_product(v_dense) == 4\n", "print(\"Imbalanced dot product test passed.\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 9. Top-K Cosine Similarity Retrieval\n", "**LeetCode Analog**: [LC 215: Kth Largest Element](https://leetcode.com/problems/kth-largest-element-in-an-array/), [LC 347: Top K Frequent Elements](https://leetcode.com/problems/top-k-frequent-elements/) \n", "**ML Context**: Vector search, candidate generation in recommendation systems, RAG nearest-neighbor retrieval.\n", "\n", "### Algorithmic Principle\n", "1. Compute cosine similarity: $\\text{sim}(u, v) = \\frac{u \\cdot v}{\\|u\\|_2 \\|v\\|_2}$.\n", "2. Maintain a **Min-Heap** (`heapq`) of fixed size $k$ containing `(similarity, doc_id)`.\n", "3. For each candidate vector:\n", " - If heap has $< k$ elements, push candidate.\n", " - If `similarity > heap[0].similarity`, replace top via `heapreplace`.\n", "4. Return candidates sorted descending by score.\n", "\n", "- **Time Complexity**: $\\mathcal{O}(N \\cdot D + N \\log k)$ where $N=$ candidates, $D=$ dimension.\n", "- **Space Complexity**: $\\mathcal{O}(k)$ auxiliary heap memory.\n", "- **Interview Trap**: Full sorting $\\mathcal{O}(N \\log N)$ instead of heap $\\mathcal{O}(N \\log k)$; division by zero on zero-norm vectors.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import math\n", "import heapq\n", "from typing import List, Tuple\n", "\n", "def top_k_cosine_similarity(query: List[float], candidates: List[Tuple[str, List[float]]], k: int) -> List[Tuple[str, float]]:\n", " \"\"\"\n", " Retrieves top-k candidates with highest cosine similarity using a min-heap.\n", " Time: O(N * D + N log k), Space: O(k).\n", " \"\"\"\n", " def norm(vec: List[float]) -> float:\n", " return math.sqrt(sum(x * x for x in vec))\n", "\n", " q_norm = norm(query)\n", " if q_norm == 0.0 or k <= 0:\n", " return []\n", "\n", " min_heap: List[Tuple[float, str]] = []\n", "\n", " for doc_id, vec in candidates:\n", " v_norm = norm(vec)\n", " sim = sum(q * v for q, v in zip(query, vec)) / (q_norm * v_norm) if v_norm > 0 else 0.0\n", "\n", " if len(min_heap) < k:\n", " heapq.heappush(min_heap, (sim, doc_id))\n", " elif sim > min_heap[0][0]:\n", " heapq.heapreplace(min_heap, (sim, doc_id))\n", "\n", " # Sort descending\n", " return sorted([(doc_id, sim) for sim, doc_id in min_heap], key=lambda x: x[1], reverse=True)\n", "\n", "# Verification Test\n", "query = [1.0, 1.0, 0.0]\n", "docs = [\n", " (\"doc_1\", [1.0, 1.0, 0.0]), # Perfect alignment: sim = 1.0\n", " (\"doc_2\", [1.0, 0.0, 0.0]), # Partial overlap: sim = 1 / sqrt(2) ~ 0.707\n", " (\"doc_3\", [0.0, 0.0, 5.0]), # Orthogonal: sim = 0.0\n", " (\"doc_4\", [2.0, 2.0, 0.0]) # Scaled version: sim = 1.0\n", "]\n", "\n", "top2 = top_k_cosine_similarity(query, docs, k=2)\n", "print(\"Top-2 retrieved documents:\")\n", "for doc_id, sim in top2:\n", " print(f\" {doc_id}: cosine similarity = {sim:.4f}\")\n", "assert len(top2) == 2 and abs(top2[0][1] - 1.0) < 1e-5\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 10. TF-IDF & Inverted Index Document Retrieval\n", "**ML Context**: Lexical retrieval baseline, BM25 foundation, sparse keyword search in hybrid search engines.\n", "\n", "### Algorithmic Principle\n", "1. **Inverted Index**: Map each term $t \\to \\{ \\text{doc\\_id}: \\text{count} \\}$.\n", "2. **Term Frequency**: $\\text{TF}(t, d) = \\frac{\\text{count}(t, d)}{\\text{total\\_words}(d)}$.\n", "3. **Inverse Document Frequency**: $\\text{IDF}(t) = \\log\\left(\\frac{N + 1}{\\text{DF}(t) + 1}\\right) + 1.0$ (smoothed).\n", "4. Score document $d$ for query $Q$: $\\text{Score}(d, Q) = \\sum_{t \\in Q} \\text{TF}(t, d) \\cdot \\text{IDF}(t)$.\n", "\n", "- **Time Complexity**: Indexing: $\\mathcal{O}(\\text{tokens})$; Querying: $\\mathcal{O}(\\sum_{t \\in Q} \\text{postings}(t))$.\n", "- **Space Complexity**: $\\mathcal{O}(\\text{vocabulary} + \\text{postings})$.\n", "- **Interview Trap**: Not smoothing IDF (leading to division by zero on unseen query terms).\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import math\n", "from collections import defaultdict, Counter\n", "from typing import List, Tuple, Dict\n", "\n", "class SimpleTFIDFIndex:\n", " \"\"\"\n", " Inverted Index scoring documents with smoothed TF-IDF.\n", " \"\"\"\n", " def __init__(self):\n", " # term -> dict(doc_id -> term_count)\n", " self.inverted_index: Dict[str, Dict[str, int]] = defaultdict(dict)\n", " self.doc_lengths: Dict[str, int] = {}\n", " self.total_docs = 0\n", "\n", " def add_document(self, doc_id: str, text: str):\n", " tokens = text.lower().split()\n", " self.doc_lengths[doc_id] = len(tokens)\n", " counts = Counter(tokens)\n", " for term, cnt in counts.items():\n", " self.inverted_index[term][doc_id] = cnt\n", " self.total_docs += 1\n", "\n", " def query(self, query_text: str, top_k: int = 5) -> List[Tuple[str, float]]:\n", " query_terms = query_text.lower().split()\n", " scores: Dict[str, float] = defaultdict(float)\n", "\n", " for term in query_terms:\n", " if term not in self.inverted_index:\n", " continue\n", " postings = self.inverted_index[term]\n", " df = len(postings)\n", " idf = math.log((self.total_docs + 1) / (df + 1)) + 1.0\n", "\n", " for doc_id, cnt in postings.items():\n", " tf = cnt / self.doc_lengths[doc_id]\n", " scores[doc_id] += tf * idf\n", "\n", " ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)\n", " return ranked[:top_k]\n", "\n", "# Verification Test\n", "corpus = {\n", " \"nlp_doc\": \"transformer neural networks natural language processing attention\",\n", " \"cv_doc\": \"convolutional neural networks computer vision image classification\",\n", " \"food_doc\": \"delicious italian pasta recipe dinner food\"\n", "}\n", "\n", "index = SimpleTFIDFIndex()\n", "for doc_id, text in corpus.items():\n", " index.add_document(doc_id, text)\n", "\n", "query_res = index.query(\"neural networks vision\", top_k=2)\n", "print(\"Query results for 'neural networks vision':\")\n", "for doc_id, score in query_res:\n", " print(f\" {doc_id}: score = {score:.4f}\")\n", "assert query_res[0][0] == \"cv_doc\" # Contains vision + neural + networks\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 11. Streaming Running Median (Two Heaps)\n", "**LeetCode Analog**: [LC 295: Find Median from Data Stream](https://leetcode.com/problems/find-median-from-data-stream/) \n", "**ML Context**: Real-time feature drift detection, tracking online loss/latency distribution percentiles.\n", "\n", "### Algorithmic Principle\n", "Maintain two balanced heaps:\n", "- `small`: **Max-Heap** (stores negated numbers) containing the lower half of data.\n", "- `large`: **Min-Heap** containing the upper half of data.\n", "- **Invariants**:\n", " 1. Every element in `small` $\\le$ every element in `large`.\n", " 2. Size invariant: $\\text{len}(\\text{small}) \\in \\{ \\text{len}(\\text{large}), \\text{len}(\\text{large}) + 1 \\}$.\n", "- **Median**: If odd count, return top of `small`. If even, return average of tops.\n", "\n", "- **Time Complexity**: $\\mathcal{O}(\\log N)$ insertion, $\\mathcal{O}(1)$ median query.\n", "- **Space Complexity**: $\\mathcal{O}(N)$ stream storage.\n", "- **Interview Trap**: Python only has min-heaps (`heapq`), so you must negate numbers (`-x`) to implement a max-heap.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import heapq\n", "from typing import List\n", "\n", "class StreamingMedian:\n", " \"\"\"\n", " Computes median from data stream using dual heaps.\n", " Time: O(log N) add, O(1) median. Space: O(N).\n", " \"\"\"\n", " def __init__(self):\n", " self.small: List[float] = [] # Max-heap (stores -val)\n", " self.large: List[float] = [] # Min-heap (stores val)\n", "\n", " def add_num(self, num: float):\n", " # 1. Push to small (max-heap)\n", " heapq.heappush(self.small, -num)\n", " \n", " # 2. Ensure order invariant: max(small) <= min(large)\n", " if self.small and self.large and (-self.small[0] > self.large[0]):\n", " val = -heapq.heappop(self.small)\n", " heapq.heappush(self.large, val)\n", "\n", " # 3. Balance size invariant: len(small) == len(large) or len(small) == len(large) + 1\n", " if len(self.small) > len(self.large) + 1:\n", " val = -heapq.heappop(self.small)\n", " heapq.heappush(self.large, val)\n", " elif len(self.large) > len(self.small):\n", " val = heapq.heappop(self.large)\n", " heapq.heappush(self.small, -val)\n", "\n", " def find_median(self) -> float:\n", " if len(self.small) > len(self.large):\n", " return float(-self.small[0])\n", " return (-self.small[0] + self.large[0]) / 2.0\n", "\n", "# Verification Test\n", "sm = StreamingMedian()\n", "stream = [5, 2, 8, 1, 9, 3]\n", "medians = []\n", "for x in stream:\n", " sm.add_num(x)\n", " medians.append(sm.find_median())\n", "\n", "print(\"Stream numbers added:\", stream)\n", "print(\"Running medians:\", medians)\n", "assert medians[-1] == 4.0 # sorted: [1, 2, 3, 5, 8, 9] -> (3 + 5)/2 = 4.0\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 12. Edit Distance & Word Error Rate (WER)\n", "**LeetCode Analog**: [LC 72: Edit Distance](https://leetcode.com/problems/edit-distance/) \n", "**ML Context**: Evaluation metric for Automatic Speech Recognition (ASR), OCR, and text generation. $\\text{WER} = \\frac{S + D + I}{N_{\\text{ref}}}$.\n", "\n", "### Algorithmic Principle\n", "Dynamic Programming over token sequences of reference $R$ ($M$ words) and hypothesis $H$ ($N$ words):\n", "$$DP[i][j] = \\begin{cases} DP[i-1][j-1] & \\text{if } R[i-1] = H[j-1] \\\\ 1 + \\min(DP[i-1][j-1], DP[i-1][j], DP[i][j-1]) & \\text{otherwise} \\end{cases}$$\n", "Where transitions correspond to **Substitution**, **Deletion**, and **Insertion**. Optimized to **1D DP row** to save memory.\n", "\n", "- **Time Complexity**: $\\mathcal{O}(M \\cdot N)$.\n", "- **Space Complexity**: $\\mathcal{O}(\\min(M, N))$ with row optimization.\n", "- **Interview Trap**: Operating on characters instead of words/tokens (ASR WER operates on whitespace-delimited word tokens).\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def compute_wer(reference: str, hypothesis: str) -> float:\n", " \"\"\"\n", " Computes Word Error Rate (WER) using Levenshtein distance on words.\n", " Time: O(M * N), Space: O(N).\n", " \"\"\"\n", " ref = reference.strip().split()\n", " hyp = hypothesis.strip().split()\n", " m, n = len(ref), len(hyp)\n", " if m == 0:\n", " return float(n)\n", "\n", " # 1D DP row: dp[j] stores distance between ref[:i] and hyp[:j]\n", " dp = list(range(n + 1))\n", "\n", " for i in range(1, m + 1):\n", " prev = dp[0]\n", " dp[0] = i\n", " for j in range(1, n + 1):\n", " temp = dp[j]\n", " if ref[i - 1] == hyp[j - 1]:\n", " dp[j] = prev\n", " else:\n", " dp[j] = 1 + min(prev, # substitution\n", " dp[j], # deletion\n", " dp[j - 1]) # insertion\n", " prev = temp\n", "\n", " total_edits = dp[n]\n", " return total_edits / m\n", "\n", "# Verification Test\n", "ref_text = \"the quick brown fox jumps over the lazy dog\"\n", "hyp_text = \"the fast brown fox jumps above the dog\"\n", "\n", "wer = compute_wer(ref_text, hyp_text)\n", "print(f\"Reference: '{ref_text}' ({len(ref_text.split())} words)\")\n", "print(f\"Hypothesis: '{hyp_text}' ({len(hyp_text.split())} words)\")\n", "print(f\"WER: {wer:.4f} (3 edits: quick->fast, over->above, lazy deleted)\")\n", "assert abs(wer - (3 / 9)) < 1e-5\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## 13. Image Rotation by Arbitrary Angle\n", "**ML / CV Context**: Data augmentation for computer vision, spatial transformers, affine matrix transformations without OpenCV/PIL.\n", "\n", "### Algorithmic Principle\n", "Why standard forward mapping fails:\n", "- **Forward Mapping ($x, y \\to x', y'$)** leaves blank pixels / holes because discrete integers do not map 1-to-1 onto the output grid.\n", "- **Inverse Mapping ($x', y' \\to x, y$)**: Iterate through every destination pixel $(x', y')$ and sample backward coordinates $(x, y) = R(-\\theta)(x' - c_x, y' - c_y) + (c_x, c_y)$.\n", "- **Bilinear Interpolation**: For non-integer source coordinates $(x, y)$, sample the 4 neighboring integer pixels using fractional distances $dx, dy$:\n", " $$I(x, y) = (1 - dx)(1 - dy) I_{00} + dx(1 - dy) I_{10} + (1 - dx)dy I_{01} + dx \\cdot dy I_{11}$$\n", "\n", "- **Time Complexity**: $\\mathcal{O}(H \\cdot W)$.\n", "- **Space Complexity**: $\\mathcal{O}(H \\cdot W)$ for output image.\n", "- **Interview Trap**: Using forward mapping instead of inverse mapping; out-of-bounds index errors when $(x_1, y_1)$ touches image boundary.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import math\n", "from typing import List\n", "\n", "def rotate_image(image: List[List[float]], angle_degrees: float, fill_value: float = 0.0) -> List[List[float]]:\n", " \"\"\"\n", " Rotates a 2D grayscale image by an arbitrary angle clockwise around its center.\n", " Uses Inverse Mapping + Bilinear Interpolation.\n", " Time: O(H * W), Space: O(H * W).\n", " \"\"\"\n", " H = len(image)\n", " W = len(image[0]) if H > 0 else 0\n", " if H == 0 or W == 0:\n", " return []\n", "\n", " cx = (W - 1) / 2.0\n", " cy = (H - 1) / 2.0\n", "\n", " # Clockwise rotation in image coordinates (y points down)\n", " theta = math.radians(angle_degrees)\n", " cos_t = math.cos(theta)\n", " sin_t = math.sin(theta)\n", "\n", " out = [[fill_value for _ in range(W)] for _ in range(H)]\n", "\n", " for yp in range(H):\n", " for xp in range(W):\n", " # Inverse mapping: rotate backwards by -theta\n", " dx_dest = xp - cx\n", " dy_dest = yp - cy\n", " x_src = dx_dest * cos_t + dy_dest * sin_t + cx\n", " y_src = -dx_dest * sin_t + dy_dest * cos_t + cy\n", "\n", " # Bilinear interpolation bounds\n", " x0 = math.floor(x_src)\n", " x1 = x0 + 1\n", " y0 = math.floor(y_src)\n", " y1 = y0 + 1\n", "\n", " if 0 <= x0 < W and 0 <= x1 < W and 0 <= y0 < H and 0 <= y1 < H:\n", " wx = x_src - x0\n", " wy = y_src - y0\n", "\n", " val = ( (1 - wx) * (1 - wy) * image[y0][x0] +\n", " wx * (1 - wy) * image[y0][x1] +\n", " (1 - wx) * wy * image[y1][x0] +\n", " wx * wy * image[y1][x1] )\n", " out[yp][xp] = round(val, 2)\n", " elif 0 <= round(x_src) < W and 0 <= round(y_src) < H:\n", " out[yp][xp] = image[round(y_src)][round(x_src)]\n", "\n", " return out\n", "\n", "# Verification Test: Rotate 5x5 pattern with cross\n", "img = [\n", " [0.0, 0.0, 1.0, 0.0, 0.0],\n", " [0.0, 0.0, 1.0, 0.0, 0.0],\n", " [1.0, 1.0, 1.0, 1.0, 1.0],\n", " [0.0, 0.0, 1.0, 0.0, 0.0],\n", " [0.0, 0.0, 1.0, 0.0, 0.0]\n", "]\n", "\n", "print(\"Original 5x5 pattern:\")\n", "for row in img:\n", " print(\" \", [int(v) for v in row])\n", "\n", "rotated_45 = rotate_image(img, 45.0)\n", "print(\"\\nRotated by 45 degrees (bilinear interpolation):\")\n", "for row in rotated_45:\n", " print(\" \", [round(v, 2) for v in row])\n", "\n", "assert rotate_image(img, 0.0) == img\n", "assert rotated_45[2][2] == 1.0 # Center stays invariant\n", "print(\"\\nImage rotation test passed!\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "## \ud83c\udfaf MLE Algorithmic Interview Quick Reference\n", "\n", "| Pattern | Problem | Key Library / Technique | Complexity |\n", "|---|---|---|---|\n", "| **Binary Search** | Weighted Sampling | `bisect.bisect_right` on Prefix Sums | $\\mathcal{O}(\\log N)$ query |\n", "| **Streaming** | Reservoir Sampling | Replace element with prob $k/(i+1)$ | $\\mathcal{O}(N)$ time, $\\mathcal{O}(k)$ space |\n", "| **Two Pointers** | Sort Transformed Array | Monotonicity from ends towards vertex | $\\mathcal{O}(N)$ time |\n", "| **Divide & Conquer** | Ternary Search | Discard $1/3$ of interval per iteration | $\\mathcal{O}(\\log(1/\\epsilon))$ time |\n", "| **Binning** | Expected Calibration Error | Equal-width bucket accumulation | $\\mathcal{O}(N + M)$ time |\n", "| **Sorting / Rank Sum**| Exact ROC-AUC | Wilcoxon-Mann-Whitney with tied ranks | $\\mathcal{O}(N \\log N)$ time |\n", "| **Sweep-Line** | Average Precision (PR-AUC) | Descending score threshold integration | $\\mathcal{O}(N \\log N)$ time |\n", "| **Two Pointers / Bisect**| Sparse Vector Dot Product | Sorted `(index, value)` pairs | $\\mathcal{O}(L_1 + L_2)$ or $\\mathcal{O}(L_1 \\log L_2)$ |\n", "| **Min-Heap** | Top-K Cosine Retrieval | `heapq.heapreplace` of size $k$ | $\\mathcal{O}(N \\cdot D + N \\log k)$ |\n", "| **Hash Map / Postings**| TF-IDF Inverted Index | Term frequency $\\times$ smoothed IDF | $\\mathcal{O}(\\text{postings})$ |\n", "| **Two Heaps** | Streaming Median | Max-heap (lower) + Min-heap (upper) | $\\mathcal{O}(\\log N)$ add, $\\mathcal{O}(1)$ median |\n", "| **Dynamic Programming**| Edit Distance / WER | 1D space-optimized DP table | $\\mathcal{O}(M \\cdot N)$ time, $\\mathcal{O}(N)$ space |\n", "| **Coordinate Geometry** | Arbitrary Image Rotation | Inverse mapping + Bilinear interpolation | $\\mathcal{O}(H \\cdot W)$ |\n" ] } ], "metadata": { "language_info": { "name": "python", "version": "3.10.0" }, "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" } }, "nbformat": 4, "nbformat_minor": 5 }