Programming & Software Engineering

Data Structures & Algorithms

Built-in Data Structures

Python
# Performance comparison — why choosing right structure matters
import time

data_list = list(range(1_000_000))
data_set = set(data_list)

# Searching for an element
target = 999_999

start = time.time()
_ = target in data_list   # O(n) — scans every element
print(f"List: {time.time()-start:.6f}s")

start = time.time()
_ = target in data_set    # O(1) — hash lookup
print(f"Set:  {time.time()-start:.6f}s")
# Set is ~1000x faster for membership testing!

Big-O Notation

Searching & Sorting

Python
# Binary Search — O(log n)
def binary_search(arr, target):
    left, right = 0, len(arr) - 1
    while left <= right:
        mid = (left + right) // 2
        if arr[mid] == target:
            return mid
        elif arr[mid] < target:
            left = mid + 1
        else:
            right = mid - 1
    return -1

# Quick Sort — O(n log n) average
def quicksort(arr):
    if len(arr) <= 1:
        return arr
    pivot = arr[len(arr) // 2]
    left = [x for x in arr if x < pivot]
    middle = [x for x in arr if x == pivot]
    right = [x for x in arr if x > pivot]
    return quicksort(left) + middle + quicksort(right)

print(quicksort([38, 27, 43, 3, 9, 82, 10]))

# Stack implementation
class Stack:
    def __init__(self): self.items = []
    def push(self, item): self.items.append(item)
    def pop(self): return self.items.pop()
    def peek(self): return self.items[-1]
    def is_empty(self): return len(self.items) == 0

Python
import heapq

class TaskScheduler:
    def __init__(self):
        self.heap = []
        self.counter = 0

    def add_task(self, task, priority):
        heapq.heappush(self.heap, (priority, self.counter, task))
        self.counter += 1

    def get_next(self):
        if self.heap:
            priority, _, task = heapq.heappop(self.heap)
            return task
        return None

scheduler = TaskScheduler()
scheduler.add_task("Fix critical bug", 1)
scheduler.add_task("Write docs", 5)
scheduler.add_task("Deploy to prod", 2)
scheduler.add_task("Code review", 3)

while (task := scheduler.get_next()):
    print(f"Executing: {task}")

def merge_sort(arr):
    if len(arr) <= 1: return arr
    mid = len(arr) // 2
    left = merge_sort(arr[:mid])
    right = merge_sort(arr[mid:])
    return merge(left, right)

def merge(left, right):
    result, i, j = [], 0, 0
    while i < len(left) and j < len(right):
        if left[i] <= right[j]:
            result.append(left[i]); i += 1
        else:
            result.append(right[j]); j += 1
    result.extend(left[i:]); result.extend(right[j:])
    return result

class QueueFromStacks:
    def __init__(self):
        self.in_stack = []
        self.out_stack = []
    def enqueue(self, item):
        self.in_stack.append(item)
    def dequeue(self):
        if not self.out_stack:
            while self.in_stack:
                self.out_stack.append(self.in_stack.pop())
        return self.out_stack.pop()

Object-Oriented Programming

Classes & Objects

Python
class NeuralLayer:
    def __init__(self, input_size, output_size, activation='relu'):
        self.weights = [[0.0] * input_size for _ in range(output_size)]
        self.bias = [0.0] * output_size
        self.activation = activation
        self._name = f"Layer({input_size}→{output_size})"  # private

    def __repr__(self):
        return f"NeuralLayer({self._name}, act={self.activation})"

    def __len__(self):
        return len(self.weights)

    def param_count(self):
        return len(self.weights) * len(self.weights[0]) + len(self.bias)

layer = NeuralLayer(128, 64)
print(layer)              # NeuralLayer(Layer(128→64), act=relu)
print(len(layer))         # 64
print(layer.param_count()) # 8256

Inheritance & Polymorphism

Python
from abc import ABC, abstractmethod

class Shape(ABC):
    @abstractmethod
    def area(self): pass

    @abstractmethod
    def perimeter(self): pass

    def describe(self):
        return f"{self.__class__.__name__}: area={self.area():.2f}"

class Circle(Shape):
    def __init__(self, radius):
        self.radius = radius
    def area(self): return 3.14159 * self.radius ** 2
    def perimeter(self): return 2 * 3.14159 * self.radius

class Rectangle(Shape):
    def __init__(self, w, h):
        self.w, self.h = w, h
    def area(self): return self.w * self.h
    def perimeter(self): return 2 * (self.w + self.h)

# Polymorphism — same interface, different behavior
shapes = [Circle(5), Rectangle(4, 6), Circle(3)]
for s in shapes:
    print(s.describe())

Design Patterns

Python
# Singleton — only one instance ever
class DatabaseConnection:
    _instance = None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

# Factory — create objects without specifying exact class
class ModelFactory:
    @staticmethod
    def create(model_type):
        models = {'linear': LinearModel, 'tree': TreeModel}
        return models[model_type]()

Python
from datetime import datetime, timedelta

class Book:
    def __init__(self, title, author, isbn):
        self.title, self.author, self.isbn = title, author, isbn
        self.is_available = True
    def __str__(self): return f"'{self.title}' by {self.author}"

class Member:
    def __init__(self, name, member_id):
        self.name, self.id = name, member_id
        self.borrowed = []

class Library:
    def __init__(self, name):
        self.name = name
        self.books, self.members, self.loans = {}, {}, []

    def add_book(self, book):
        self.books[book.isbn] = book

    def borrow(self, isbn, member_id):
        book = self.books.get(isbn)
        member = self.members.get(member_id)
        if book and member and book.is_available:
            book.is_available = False
            due = datetime.now() + timedelta(days=14)
            self.loans.append({'book': book, 'member': member, 'due': due})
            member.borrowed.append(book)
            print(f"✓ {member.name} borrowed {book}. Due: {due:%Y-%m-%d}")

    def return_book(self, isbn):
        book = self.books.get(isbn)
        if book:
            book.is_available = True
            print(f"✓ {book} returned")

lib = Library("City Library")
lib.add_book(Book("Deep Learning", "Goodfellow", "978-0"))
lib.members["M1"] = Member("Alice", "M1")
lib.borrow("978-0", "M1")

class Vector2D:
    def __init__(self, x, y): self.x, self.y = x, y
    def __add__(self, other): return Vector2D(self.x+other.x, self.y+other.y)
    def __eq__(self, other): return self.x==other.x and self.y==other.y
    def __repr__(self): return f"Vec({self.x},{self.y})"

v = Vector2D(1,2) + Vector2D(3,4)  # Vec(4,6)

Functional Programming Concepts

First-Class Functions & Lambdas

Python
# Functions are objects — can be passed, returned, stored
def apply_twice(func, value):
    return func(func(value))

print(apply_twice(lambda x: x * 2, 3))  # 12
print(apply_twice(lambda x: x + 10, 5)) # 25

# map, filter, reduce
nums = [1, 2, 3, 4, 5, 6, 7, 8]
squares = list(map(lambda x: x**2, nums))         # [1,4,9,16,25,36,49,64]
evens = list(filter(lambda x: x%2==0, nums))     # [2,4,6,8]

from functools import reduce
total = reduce(lambda a, b: a + b, nums)           # 36

# List comprehension alternative (more Pythonic)
squares = [x**2 for x in nums]
evens = [x for x in nums if x % 2 == 0]

Closures & functools

Python
# Closure — function that remembers its enclosing scope
def make_multiplier(factor):
    def multiply(x):
        return x * factor  # 'factor' is captured from outer scope
    return multiply

double = make_multiplier(2)
triple = make_multiplier(3)
print(double(5))  # 10
print(triple(5))  # 15

# Memoization with lru_cache
from functools import lru_cache

@lru_cache(maxsize=None)
def fibonacci(n):
    if n < 2: return n
    return fibonacci(n-1) + fibonacci(n-2)

print(fibonacci(100))  # Instant! Without cache: years

# Partial application
from functools import partial
import json

pretty_json = partial(json.dumps, indent=2, sort_keys=True)
print(pretty_json({"name": "AI", "type": "ML"}))

Python
from functools import reduce

# Pipeline: compose functions left-to-right
def pipeline(*funcs):
    return lambda x: reduce(lambda v, f: f(v), funcs, x)

# Data processing steps
clean = lambda data: [s.strip().lower() for s in data]
remove_empty = lambda data: [s for s in data if s]
remove_dupes = lambda data: list(dict.fromkeys(data))
sort_alpha = lambda data: sorted(data)

process = pipeline(clean, remove_empty, remove_dupes, sort_alpha)

raw = ["  Python ", "  java", "", "PYTHON", "  Go  ", "java"]
result = process(raw)
print(result)  # ['go', 'java', 'python']

result = list(map(lambda x: x**2, filter(lambda x: x%3==0, range(20))))
# [0, 9, 36, 81, 144, 225, 324]
# List comprehension is more Pythonic, but map/filter is more composable

@lru_cache(maxsize=None)
def factorial(n):
    if n <= 1: return 1
    return n * factorial(n - 1)
print(factorial(100))  # Computed instantly with caching

Memory Management & Generators

Generators: Lazy Evaluation

Python
import sys

# List vs Generator — memory comparison
big_list = [x**2 for x in range(1_000_000)]
big_gen = (x**2 for x in range(1_000_000))

print(f"List:      {sys.getsizeof(big_list):>10,} bytes")  # ~8.5 MB
print(f"Generator: {sys.getsizeof(big_gen):>10,} bytes")   # ~200 bytes!

# Custom generator with yield
def read_large_file(filepath):
    with open(filepath) as f:
        for line in f:
            yield line.strip()  # One line at a time, not whole file

# itertools — lazy iteration toolkit
from itertools import chain, islice, count, cycle

# Chain multiple iterables
combined = chain([1,2], [3,4], [5,6])  # 1,2,3,4,5,6

# Take first N from infinite generator
first_10_evens = list(islice((x for x in count() if x%2==0), 10))

# __slots__ — reduce memory per instance
class PointSlots:
    __slots__ = ['x', 'y']
    def __init__(self, x, y): self.x, self.y = x, y
# Uses ~40% less memory than regular class with __dict__

Python
import csv

def read_csv_rows(filepath):
    with open(filepath) as f:
        reader = csv.DictReader(f)
        for row in reader:
            yield row

def filter_active(rows):
    for row in rows:
        if row.get('status') == 'active':
            yield row

def extract_emails(rows):
    for row in rows:
        yield row['email']

# Pipeline — processes 1M rows using constant memory!
# rows = read_csv_rows('users_1M.csv')
# active = filter_active(rows)
# emails = extract_emails(active)
# for email in emails:
#     send_newsletter(email)

def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

from itertools import islice
print(list(islice(fibonacci(), 10)))  # [0,1,1,2,3,5,8,13,21,34]

Decorators & Context Managers

Building Decorators

Python
import time
from functools import wraps

# Timer decorator
def timer(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"⏱ {func.__name__} took {elapsed:.4f}s")
        return result
    return wrapper

# Retry decorator with arguments
def retry(max_attempts=3, delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f"Attempt {attempt+1} failed: {e}")
                    time.sleep(delay)
            raise Exception(f"Failed after {max_attempts} attempts")
        return wrapper
    return decorator

@timer
@retry(max_attempts=3)
def fetch_data(url):
    print(f"Fetching {url}...")
    return {"status": "ok"}

Context Managers

Python
# Class-based context manager
class DatabaseTransaction:
    def __enter__(self):
        print("BEGIN TRANSACTION")
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            print(f"ROLLBACK — Error: {exc_val}")
        else:
            print("COMMIT")
        return False

# contextlib — simpler decorator-based approach
from contextlib import contextmanager

@contextmanager
def temp_directory():
    import tempfile, shutil
    dirpath = tempfile.mkdtemp()
    try:
        yield dirpath
    finally:
        shutil.rmtree(dirpath)

with temp_directory() as tmpdir:
    print(f"Working in {tmpdir}")
# Directory is automatically cleaned up

Python
import time, logging
from functools import wraps

logging.basicConfig(level=logging.INFO)

def log_and_time(logger=None):
    def decorator(func):
        log = logger or logging.getLogger(func.__module__)
        @wraps(func)
        def wrapper(*args, **kwargs):
            log.info(f"▶ {func.__name__} started")
            start = time.perf_counter()
            try:
                result = func(*args, **kwargs)
                elapsed = time.perf_counter() - start
                log.info(f"✓ {func.__name__} completed in {elapsed:.3f}s")
                return result
            except Exception as e:
                elapsed = time.perf_counter() - start
                log.error(f"✗ {func.__name__} failed after {elapsed:.3f}s: {e}")
                raise
        return wrapper
    return decorator

@log_and_time()
def train_model(epochs):
    time.sleep(0.5)
    return {"accuracy": 0.95}

train_model(10)

def memoize(func):
    cache = {}
    @wraps(func)
    def wrapper(*args):
        if args not in cache:
            cache[args] = func(*args)
        return cache[args]
    wrapper.cache = cache
    return wrapper

@contextmanager
def suppress(*exceptions):
    try:
        yield
    except exceptions:
        pass

with suppress(FileNotFoundError):
    open('nonexistent.txt')  # Silently ignored

NumPy — Array Operations

Array Creation & Indexing

Python
import numpy as np

# Creation
a = np.array([1,2,3,4,5])
zeros = np.zeros((3,4))
rand = np.random.randn(3,3)
grid = np.arange(0, 100, 5)
space = np.linspace(0, 1, 50)

# Boolean masking — powerful filtering
data = np.random.randn(1000)
positives = data[data > 0]        # All positive values
outliers = data[np.abs(data) > 2]  # Beyond 2 std devs

# Vectorized operations — 100x faster than loops
import time
size = 1_000_000
a = np.random.randn(size)

start = time.time()
result_loop = [x**2 + 2*x + 1 for x in a]
print(f"Loop:  {time.time()-start:.3f}s")

start = time.time()
result_np = a**2 + 2*a + 1
print(f"NumPy: {time.time()-start:.3f}s")  # ~50-100x faster

Broadcasting & Linear Algebra

Python
# Broadcasting: (3,3) + (3,1) → auto-expands
matrix = np.ones((3,3))
col_vec = np.array([[10],[20],[30]])
result = matrix + col_vec  # Each row gets different offset

# Linear algebra for ML
X = np.random.randn(100, 3)
w = np.random.randn(3)
y = X @ w  # Matrix-vector multiplication (predictions)

# Normal equation: w = (XᵀX)⁻¹Xᵀy
X_b = np.c_[np.ones((100,1)), X]  # Add bias column
w_optimal = np.linalg.inv(X_b.T @ X_b) @ X_b.T @ y

Python
import numpy as np

# Generate data: y = 3x₁ + 5x₂ + 7 + noise
np.random.seed(42)
X = np.random.randn(200, 2)
y = 3*X[:,0] + 5*X[:,1] + 7 + np.random.randn(200)*0.5

# Add bias, split data
X_b = np.c_[np.ones((200,1)), X]
X_train, X_test = X_b[:160], X_b[160:]
y_train, y_test = y[:160], y[160:]

# Gradient descent
w = np.zeros(3)
lr, epochs = 0.01, 1000
for i in range(epochs):
    preds = X_train @ w
    error = preds - y_train
    gradient = (2/len(y_train)) * X_train.T @ error
    w -= lr * gradient

print(f"Learned weights: bias={w[0]:.2f}, w1={w[1]:.2f}, w2={w[2]:.2f}")
# Expected: ~7, ~3, ~5
rmse = np.sqrt(np.mean((X_test @ w - y_test)**2))
print(f"Test RMSE: {rmse:.4f}")

X = np.random.randn(100, 5) * 10 + 50
X_norm = (X - X.mean(axis=0)) / X.std(axis=0)
print(X_norm.mean(axis=0).round(10))  # ~[0, 0, 0, 0, 0]
print(X_norm.std(axis=0).round(10))   # ~[1, 1, 1, 1, 1]

Pandas — Data Wrangling

Python
import pandas as pd
import numpy as np

# Create from dict
df = pd.DataFrame({
    'name': ['Alice','Bob','Charlie','Diana','Eve'],
    'dept': ['ML','Web','ML','Data','Web'],
    'salary': [95000,82000,105000,78000,88000],
    'exp_years': [5, 3, 8, 2, 4]
})

# Selecting & filtering
ml_team = df[df['dept'] == 'ML']
senior = df.query('exp_years >= 5 and salary > 90000')

# GroupBy — split-apply-combine
dept_stats = df.groupby('dept').agg(
    avg_salary=('salary', 'mean'),
    headcount=('name', 'count'),
    max_exp=('exp_years', 'max')
)

# Missing data handling
df.loc[1, 'salary'] = np.nan
df['salary'] = df['salary'].fillna(df['salary'].median())

# Apply custom function
df['tax_bracket'] = df['salary'].apply(
    lambda s: 'High' if s > 90000 else 'Standard'
)

Python
df = pd.read_csv('https://raw.githubusercontent.com/datasciencedojo/datasets/master/titanic.csv')

# Exploration
print(df.shape)                  # (891, 12)
print(df['Survived'].value_counts())
print(df.isnull().sum())          # Age: 177, Cabin: 687 missing

# Survival rate by class and gender
survival = df.groupby(['Pclass', 'Sex'])['Survived'].mean()
print(survival.round(2))
# 1st class females: 97% survived, 3rd class males: 14%

# Feature engineering
df['Age'] = df['Age'].fillna(df['Age'].median())
df['FamilySize'] = df['SibSp'] + df['Parch'] + 1
df['IsAlone'] = (df['FamilySize'] == 1).astype(int)

orders = pd.DataFrame({'id':[1,2,3], 'product':['A','B','A']})
prices = pd.DataFrame({'product':['A','B'], 'price':[100,200]})
merged = orders.merge(prices, on='product')
print(merged)

top3 = df.groupby('dept')['salary'].mean().nlargest(3)

Matplotlib & Seaborn — Visualization

Python
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np

# Multi-panel dashboard
fig, axes = plt.subplots(2, 2, figsize=(12, 10))

# 1. Line plot
x = np.linspace(0, 10, 100)
axes[0,0].plot(x, np.sin(x), '#6366f1', linewidth=2, label='sin')
axes[0,0].plot(x, np.cos(x), '#f59e0b', linewidth=2, label='cos')
axes[0,0].legend(); axes[0,0].set_title('Trigonometric Functions')

# 2. Histogram
data = np.random.normal(0, 1, 1000)
axes[0,1].hist(data, bins=30, color='#10b981', edgecolor='white')
axes[0,1].set_title('Normal Distribution')

# 3. Scatter
x = np.random.randn(100)
y = 2*x + np.random.randn(100)*0.5
axes[1,0].scatter(x, y, alpha=0.7, c='#ec4899')
axes[1,0].set_title('Correlation')

# 4. Bar chart
categories = ['Python', 'JS', 'Java', 'C++']
values = [85, 72, 68, 55]
axes[1,1].barh(categories, values, color='#6366f1')
axes[1,1].set_title('Language Popularity')

plt.tight_layout()
plt.savefig('dashboard.png', dpi=150)

# Seaborn — statistical plots
sns.set_theme(style='whitegrid')
tips = sns.load_dataset('tips')
sns.boxplot(data=tips, x='day', y='total_bill', hue='time')
plt.title('Bill Distribution by Day')

df = sns.load_dataset('iris')
corr = df.select_dtypes(include='number').corr()
sns.heatmap(corr, annot=True, cmap='coolwarm', center=0)

plt.annotate('Peak', xy=(5, 100), xytext=(6, 120),
            arrowprops=dict(arrowstyle='->', color='red'),
            fontsize=12, color='red')

SciPy & Jupyter Notebooks

Python
from scipy import optimize, stats
import numpy as np

# Curve fitting — find best parameters
def model(x, a, b, c):
    return a * np.exp(-b * x) + c

x_data = np.linspace(0, 4, 50)
y_data = model(x_data, 2.5, 1.3, 0.5) + np.random.normal(0, 0.1, 50)

params, cov = optimize.curve_fit(model, x_data, y_data)
print(f"Fitted: a={params[0]:.2f}, b={params[1]:.2f}, c={params[2]:.2f}")

# T-test: are two groups statistically different?
group_a = np.random.normal(100, 10, 50)
group_b = np.random.normal(105, 10, 50)
t_stat, p_value = stats.ttest_ind(group_a, group_b)
print(f"t={t_stat:.3f}, p={p_value:.4f}")
print("Significant!" if p_value < 0.05 else "Not significant")

Python
import numpy as np
from scipy import stats

# Control: old button, Treatment: new button
np.random.seed(42)
control_clicks = np.random.binomial(1, 0.12, 1000)   # 12% CTR
treatment_clicks = np.random.binomial(1, 0.15, 1000) # 15% CTR

print(f"Control CTR:   {control_clicks.mean():.1%}")
print(f"Treatment CTR: {treatment_clicks.mean():.1%}")

# Chi-squared test for proportions
from scipy.stats import chi2_contingency
table = np.array([[control_clicks.sum(), 1000-control_clicks.sum()],
                  [treatment_clicks.sum(), 1000-treatment_clicks.sum()]])
chi2, p, dof, expected = chi2_contingency(table)
print(f"\nChi² = {chi2:.3f}, p-value = {p:.4f}")
print("→ Deploy new button!" if p < 0.05 else "→ Keep old button")

result = optimize.minimize(lambda x: (x-3)**2+2, x0=0)
print(f"Minimum at x={result.x[0]:.4f}, f(x)={result.fun:.4f}")

Version Control with Git

Bash
# Initialize & first commit
git init
git add .
git commit -m "Initial commit: project structure"

# Branching workflow
git checkout -b feature/add-login      # Create & switch
# ... make changes ...
git add -A
git commit -m "feat: add user login endpoint"
git checkout main
git merge feature/add-login            # Merge feature into main
git branch -d feature/add-login        # Clean up branch

# Working with remotes
git remote add origin https://github.com/user/repo.git
git push -u origin main
git pull origin main                   # Fetch + merge

# Undo mistakes
git stash                              # Save uncommitted changes
git reset --soft HEAD~1                # Undo last commit, keep changes
git log --oneline --graph -10          # Visual history

__pycache__/
*.pyc
.env
*.sqlite
node_modules/
dist/
*.egg-info/
.ipynb_checkpoints/
data/*.csv
models/*.pkl
wandb/

Testing & Debugging

Python
# calculator.py
def add(a, b): return a + b
def divide(a, b):
    if b == 0: raise ValueError("Cannot divide by zero")
    return a / b

# test_calculator.py
import pytest
from calculator import add, divide

def test_add():
    assert add(2, 3) == 5
    assert add(-1, 1) == 0
    assert add(0, 0) == 0

def test_divide_by_zero():
    with pytest.raises(ValueError):
        divide(10, 0)

# Parametrize — test multiple inputs at once
@pytest.mark.parametrize("a,b,expected", [
    (10, 2, 5), (9, 3, 3), (7, 2, 3.5)
])
def test_divide(a, b, expected):
    assert divide(a, b) == expected

# Fixtures — shared setup
@pytest.fixture
def sample_data():
    return [1, 2, 3, 4, 5]

def test_sum(sample_data):
    assert sum(sample_data) == 15

Bash
# Run tests with coverage
pytest test_calculator.py -v --cov=calculator --cov-report=term-missing

Debugging & Logging

Python
import logging

logging.basicConfig(level=logging.INFO,
    format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(__name__)

def train_model(data):
    logger.info(f"Training on {len(data)} samples")
    try:
        # Training logic
        logger.info("Training complete")
    except Exception as e:
        logger.error(f"Training failed: {e}")
        raise

# Debug with breakpoint() — drops into pdb
def buggy_function(x):
    result = x * 2
    breakpoint()  # Execution pauses here → inspect variables
    return result + 1

def test_read_file(tmp_path):
    f = tmp_path / "test.txt"
    f.write_text("hello world")
    assert f.read_text() == "hello world"

Code Modularization, REST APIs & Docker

Clean Code & Modules

Python
# Project structure
# ml_project/
# ├── src/
# │   ├── __init__.py
# │   ├── model.py
# │   ├── preprocess.py
# │   └── api.py
# ├── tests/
# ├── Dockerfile
# └── requirements.txt

FastAPI — Modern Python API

Python
from fastapi import FastAPI
from pydantic import BaseModel
import pickle
import numpy as np

app = FastAPI(title="ML Prediction API")

class PredictionInput(BaseModel):
    features: list[float]

class PredictionOutput(BaseModel):
    prediction: str
    confidence: float

# Load model on startup
with open("model.pkl", "rb") as f:
    model = pickle.load(f)

@app.post("/predict", response_model=PredictionOutput)
async def predict(data: PredictionInput):
    X = np.array(data.features).reshape(1, -1)
    pred = model.predict(X)[0]
    proba = model.predict_proba(X).max()
    return PredictionOutput(prediction=str(pred), confidence=float(proba))

@app.get("/health")
async def health():
    return {"status": "healthy"}

Docker — Containerization

Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "src.api:app", "--host", "0.0.0.0", "--port", "8000"]

Bash
# Build and run
docker build -t ml-api .
docker run -p 8000:8000 ml-api

# docker-compose for multi-service
# docker-compose.yml:
# services:
#   api:  build: . ports: ["8000:8000"]
#   db:   image: postgres:15

GPU vs CPU, CUDA & Distributed Computing

CPU vs GPU Architecture

A GPU's thousands of cores can execute the same operation on thousands of data elements simultaneously (SIMD — Single Instruction Multiple Data). Matrix multiplication — the core of neural networks — is perfectly parallel: each output element is an independent dot product.

GPU-Accelerated Python

Python
# CuPy — NumPy on GPU (drop-in replacement)
import cupy as cp
import numpy as np
import time

size = 10000

# CPU (NumPy)
a_cpu = np.random.randn(size, size).astype(np.float32)
b_cpu = np.random.randn(size, size).astype(np.float32)
start = time.time()
c_cpu = a_cpu @ b_cpu
print(f"CPU: {time.time()-start:.3f}s")

# GPU (CuPy)
a_gpu = cp.array(a_cpu)
b_gpu = cp.array(b_cpu)
start = time.time()
c_gpu = a_gpu @ b_gpu
cp.cuda.Stream.null.synchronize()
print(f"GPU: {time.time()-start:.3f}s")
# GPU is typically 10-50x faster for large matrix multiplications

# Numba — JIT compile Python to GPU kernels
from numba import cuda
import math

@cuda.jit
def vector_add_gpu(a, b, result):
    idx = cuda.grid(1)
    if idx < a.size:
        result[idx] = a[idx] + b[idx]

CUDA Concepts

CUDA organizes parallel execution into a hierarchy:

Distributed Computing

Python
# PyTorch Data Parallel — split batches across GPUs
import torch
import torch.nn as nn

model = nn.Linear(1000, 100)

# Data Parallelism: same model, split data across GPUs
if torch.cuda.device_count() > 1:
    model = nn.DataParallel(model)
model = model.cuda()

# Model Parallelism: split model layers across GPUs
# layer1 → GPU0, layer2 → GPU1 (for models too large for one GPU)

Memory Bottlenecks & Mixed Precision

Python
# Mixed Precision Training with PyTorch
from torch.cuda.amp import autocast, GradScaler

scaler = GradScaler()

for batch in dataloader:
    optimizer.zero_grad()
    with autocast():                  # FP16 forward pass
        output = model(batch)
        loss = criterion(output, labels)
    scaler.scale(loss).backward()      # Scaled FP16 backward
    scaler.step(optimizer)
    scaler.update()
# ~2x faster training, ~50% less memory!

Python
import numpy as np
import time

sizes = [100, 500, 1000, 2000, 5000]
results = []

for n in sizes:
    A = np.random.randn(n, n).astype(np.float32)
    B = np.random.randn(n, n).astype(np.float32)

    start = time.time()
    C = A @ B
    cpu_time = time.time() - start

    gflops = (2 * n**3) / cpu_time / 1e9
    results.append((n, cpu_time, gflops))
    print(f"Size {n:>5}×{n:<5} → {cpu_time:.3f}s ({gflops:.1f} GFLOPS)")

# With GPU (CuPy):
# try:
#     import cupy as cp
#     A_gpu = cp.array(A); B_gpu = cp.array(B)
#     start = time.time()
#     C_gpu = A_gpu @ B_gpu
#     cp.cuda.Stream.null.synchronize()
#     gpu_time = time.time() - start
#     print(f"GPU: {gpu_time:.3f}s → {cpu_time/gpu_time:.1f}x speedup")