Exercises

import numpy as np

def standardize(x: np.ndarray) -> np.ndarray:
    """Standardize an array to zero mean and unit variance.

    Parameters
    ----------
    x : np.ndarray
        Input array.

    Returns
    -------
    np.ndarray
        Standardized array with mean ~0 and std ~1.
    """
    return (x - x.mean()) / x.std()

# Verify purity
data = np.array([1.0, 2.0, 3.0, 4.0, 5.0])
original = data.copy()
result = standardize(data)
assert np.array_equal(data, original), "Input was modified!"
assert np.isclose(result.mean(), 0.0)
assert np.isclose(result.std(), 1.0)

import numpy as np
from functools import reduce

def apply_transforms(data: np.ndarray, *transforms) -> np.ndarray:
    """Apply a sequence of transformations to data.

    Parameters
    ----------
    data : np.ndarray
        Input data.
    *transforms : callable
        Functions to apply in order.

    Returns
    -------
    np.ndarray
        Transformed data.
    """
    return reduce(lambda d, f: f(d), transforms, data)

# Usage
result = apply_transforms(
    np.array([1, -2, 3, -4, 5]),
    np.abs,
    np.sqrt,
    lambda x: x / x.max(),
)

# Bug: shared mutable default
print(log_experiment("run1"))  # ['run1']
print(log_experiment("run2"))  # ['run1', 'run2'] — unexpected!

def log_experiment(result, history=None):
    if history is None:
        history = []
    history.append(result)
    return history

print(log_experiment("run1"))  # ['run1']
print(log_experiment("run2"))  # ['run2'] — correct!

import functools
import time
import numpy as np

def timer(func):
    @functools.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

@timer
def compute_eigenvalues(n: int) -> np.ndarray:
    """Compute eigenvalues of a random n x n matrix."""
    A = np.random.randn(n, n)
    return np.linalg.eigvals(A)

eigs = compute_eigenvalues(500)
print(compute_eigenvalues.__name__)  # 'compute_eigenvalues'

import time
from contextlib import contextmanager
import numpy as np

@contextmanager
def timed_block(label: str = "Block"):
    print(f"Starting {label}...")
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"Done in {elapsed:.4f}s")

with timed_block("Matrix multiply"):
    A = np.random.randn(1000, 1000)
    B = np.random.randn(1000, 1000)
    C = A @ B

import numpy as np

def make_activation(name: str, alpha: float = 0.01):
    """Create an activation function by name.

    Parameters
    ----------
    name : str
        One of 'relu', 'sigmoid', 'tanh', 'leaky_relu'.
    alpha : float
        Slope for negative values in leaky ReLU.
    """
    if name == "relu":
        def activation(x):
            return np.maximum(0, x)
    elif name == "sigmoid":
        def activation(x):
            return 1 / (1 + np.exp(-x))
    elif name == "tanh":
        def activation(x):
            return np.tanh(x)
    elif name == "leaky_relu":
        def activation(x):
            return np.where(x > 0, x, alpha * x)
    else:
        raise ValueError(f"Unknown activation: {name}")

    activation.__name__ = name
    activation.name = name
    return activation

# Test
relu = make_activation("relu")
leaky = make_activation("leaky_relu", alpha=0.1)
x = np.linspace(-3, 3, 100)
print(relu(x).min())   # 0.0
print(leaky(-1.0))     # -0.1

import functools

def count_calls(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        wrapper.call_count += 1
        return func(*args, **kwargs)
    wrapper.call_count = 0
    wrapper.reset_count = lambda: setattr(wrapper, 'call_count', 0)
    return wrapper

@count_calls
def simulate(snr_db):
    return np.random.rand() < 10**(-snr_db/10)

for _ in range(1000):
    simulate(10)
print(simulate.call_count)  # 1000
simulate.reset_count()
print(simulate.call_count)  # 0

import functools
import time

def retry(max_attempts: int = 3, delay: float = 1.0):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    wait = delay * (2 ** attempt)
                    print(f"{func.__name__} failed (attempt {attempt+1}/"
                          f"{max_attempts}), retrying in {wait:.1f}s: {e}")
                    time.sleep(wait)
            raise last_exception
        return wrapper
    return decorator

@retry(max_attempts=3, delay=0.5)
def download_data(url: str) -> bytes:
    import urllib.request
    return urllib.request.urlopen(url).read()

import os

class WorkingDirectory:
    def __init__(self, path: str, create: bool = True):
        self.path = path
        self.create = create
        self._old_cwd = None

    def __enter__(self):
        self._old_cwd = os.getcwd()
        if self.create:
            os.makedirs(self.path, exist_ok=True)
        os.chdir(self.path)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        os.chdir(self._old_cwd)
        return False

from contextlib import contextmanager

@contextmanager
def working_directory(path: str, create: bool = True):
    old_cwd = os.getcwd()
    if create:
        os.makedirs(path, exist_ok=True)
    os.chdir(path)
    try:
        yield path
    finally:
        os.chdir(old_cwd)

import numpy as np
from functools import partial

# Solution 1: Default argument
callbacks_1 = []
for freq in [100, 200, 400, 800]:
    callbacks_1.append(lambda t, f=freq: np.sin(2 * np.pi * f * t))

# Solution 2: functools.partial
def sine_wave(t, freq):
    return np.sin(2 * np.pi * freq * t)

callbacks_2 = [partial(sine_wave, freq=f) for f in [100, 200, 400, 800]]

# Solution 3: Factory function
def make_sine(freq):
    def sine(t):
        return np.sin(2 * np.pi * freq * t)
    return sine

callbacks_3 = [make_sine(f) for f in [100, 200, 400, 800]]

import functools
import inspect
import numpy as np

def validate_shapes(shape_specs: dict):
    def decorator(func):
        sig = inspect.signature(func)
        param_names = list(sig.parameters.keys())

        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            bound = sig.bind(*args, **kwargs)
            bound.apply_defaults()

            for name, expected in shape_specs.items():
                arr = bound.arguments.get(name)
                if arr is None or not isinstance(arr, np.ndarray):
                    continue
                if len(arr.shape) != len(expected):
                    raise ValueError(
                        f"{func.__name__}: {name} expected "
                        f"{len(expected)}D, got {len(arr.shape)}D"
                    )
                for i, (actual, spec) in enumerate(zip(arr.shape, expected)):
                    if spec is not None and actual != spec:
                        raise ValueError(
                            f"{func.__name__}: {name} dim {i} "
                            f"expected {spec}, got {actual}"
                        )
            return func(*args, **kwargs)
        return wrapper
    return decorator

@validate_shapes({"X": (None, 10), "y": (None,)})
def fit(X, y):
    return X.shape, y.shape

fit(np.zeros((100, 10)), np.zeros(100))   # OK
# fit(np.zeros((100, 5)), np.zeros(100))  # ValueError

import functools
import hashlib
import os
import pickle
import numpy as np

def memoize_disk(cache_dir: str = ".cache"):
    def decorator(func):
        os.makedirs(cache_dir, exist_ok=True)

        def _hash_arg(arg):
            if isinstance(arg, np.ndarray):
                h = hashlib.sha256(arg.tobytes())
                h.update(str(arg.dtype).encode())
                h.update(str(arg.shape).encode())
                return h.hexdigest()
            return hashlib.sha256(repr(arg).encode()).hexdigest()

        def _cache_key(args, kwargs):
            parts = [func.__name__]
            parts.extend(_hash_arg(a) for a in args)
            for k, v in sorted(kwargs.items()):
                parts.append(f"{k}={_hash_arg(v)}")
            combined = "_".join(parts)
            return hashlib.sha256(combined.encode()).hexdigest()

        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            key = _cache_key(args, kwargs)
            path = os.path.join(cache_dir, f"{key}.pkl")
            if os.path.exists(path):
                with open(path, "rb") as f:
                    return pickle.load(f)
            result = func(*args, **kwargs)
            with open(path, "wb") as f:
                pickle.dump(result, f)
            return result

        def clear_cache():
            import shutil
            if os.path.exists(cache_dir):
                shutil.rmtree(cache_dir)
                os.makedirs(cache_dir)

        wrapper.clear_cache = clear_cache
        return wrapper
    return decorator

class Pipeline:
    def __init__(self, *funcs):
        self._funcs = list(funcs)

    def __rshift__(self, other):
        if callable(other):
            return Pipeline(*self._funcs, other)
        raise TypeError(f"Cannot compose with {type(other)}")

    def __call__(self, data):
        result = data
        for func in self._funcs:
            result = func(result)
        return result

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

    def __iter__(self):
        return iter(self._funcs)

    def __repr__(self):
        names = [f.__name__ for f in self._funcs]
        return f"Pipeline({' >> '.join(names)})"

    def describe(self):
        for i, func in enumerate(self._funcs):
            doc = (func.__doc__ or "No description").strip().split('\n')[0]
            print(f"  Step {i+1}: {func.__name__} — {doc}")

# Usage
pipe = Pipeline(np.abs) >> np.sqrt >> np.log1p
print(pipe)       # Pipeline(absolute >> sqrt >> log1p)
print(len(pipe))  # 3
result = pipe(np.array([-4, -1, 0, 1, 4]))

import functools
import warnings

def deprecated(message: str, version: str = ""):
    def decorator(func):
        warned = False

        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            nonlocal warned
            if not warned:
                ver_info = f" (since v{version})" if version else ""
                warnings.warn(
                    f"{func.__name__} is deprecated{ver_info}. {message}",
                    category=DeprecationWarning,
                    stacklevel=2,
                )
                warned = True
            return func(*args, **kwargs)
        return wrapper
    return decorator

@deprecated("Use compute_ber_v2 instead", version="3.0")
def compute_ber(tx, rx):
    """Old BER computation."""
    return np.mean(tx != rx)

import os
import tempfile
from contextlib import ExitStack

def process_multiple_files(file_paths: list[str]) -> dict:
    """Process multiple files, ensuring all are properly closed."""
    results = {}

    with ExitStack() as stack:
        # Open all files — ExitStack closes them all on exit
        handles = []
        for path in file_paths:
            fh = stack.enter_context(open(path, 'r'))
            handles.append((path, fh))

        # Process each file
        for path, fh in handles:
            content = fh.read()
            results[path] = len(content)

    # All files are guaranteed closed here
    return results

# Demo with temporary files
with tempfile.TemporaryDirectory() as tmpdir:
    paths = []
    for i in range(5):
        path = os.path.join(tmpdir, f"data_{i}.txt")
        with open(path, 'w') as f:
            f.write(f"Data for experiment {i}\n" * 100)
        paths.append(path)

    results = process_multiple_files(paths)
    for path, size in results.items():
        print(f"{os.path.basename(path)}: {size} chars")

import functools

_OPTIMIZER_REGISTRY = {}

def register(name: str):
    def decorator(func):
        _OPTIMIZER_REGISTRY[name] = func
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)
        return wrapper
    return decorator

def get_optimizer(name: str):
    if name not in _OPTIMIZER_REGISTRY:
        available = ", ".join(_OPTIMIZER_REGISTRY.keys())
        raise ValueError(f"Unknown optimizer: {name}. Available: {available}")
    return _OPTIMIZER_REGISTRY[name]

def list_optimizers():
    return list(_OPTIMIZER_REGISTRY.keys())

@register("sgd")
def sgd_optimizer(params, lr=0.01):
    """Stochastic Gradient Descent."""
    return {p: p - lr * g for p, g in params.items()}

@register("adam")
def adam_optimizer(params, lr=0.001, beta1=0.9, beta2=0.999):
    """Adam optimizer."""
    return {p: p - lr * g for p, g in params.items()}

# Dispatch
opt = get_optimizer("adam")
print(opt.__name__)         # 'adam_optimizer'
print(list_optimizers())    # ['sgd', 'adam']

import functools
import json
import time
import uuid
import threading
from datetime import datetime

class ExperimentTracker:
    _runs = []
    _call_stack = threading.local()

    @classmethod
    def track(cls, func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            run_id = str(uuid.uuid4())[:8]
            if not hasattr(cls._call_stack, 'stack'):
                cls._call_stack.stack = []
            parent = cls._call_stack.stack[-1] if cls._call_stack.stack else None

            run = {
                "run_id": run_id,
                "parent_id": parent,
                "function": func.__name__,
                "start_time": datetime.now().isoformat(),
                "args": [repr(a)[:100] for a in args],
                "kwargs": {k: repr(v)[:100] for k, v in kwargs.items()},
            }

            cls._call_stack.stack.append(run_id)
            try:
                start = time.perf_counter()
                result = func(*args, **kwargs)
                run["duration"] = time.perf_counter() - start
                run["status"] = "success"
                run["result"] = repr(result)[:200]
                return result
            except Exception as e:
                run["duration"] = time.perf_counter() - start
                run["status"] = "error"
                run["error"] = str(e)
                raise
            finally:
                cls._call_stack.stack.pop()
                cls._runs.append(run)

        return wrapper

    @classmethod
    def save(cls, path: str = "experiment_log.json"):
        with open(path, "w") as f:
            json.dump(cls._runs, f, indent=2)

    @classmethod
    def report(cls):
        print(f"Total runs: {len(cls._runs)}")
        for run in cls._runs:
            status = run["status"]
            print(f"  [{run['run_id']}] {run['function']} "
                  f"({run['duration']:.4f}s) [{status}]")

experiment_tracker = ExperimentTracker.track

import functools
import time
import warnings

def jit_fallback(func=None, *, nopython=True):
    if func is None:
        return lambda f: jit_fallback(f, nopython=nopython)

    original = func

    try:
        import numba
        try:
            jitted = numba.jit(nopython=nopython)(func)
            # Trigger compilation with a test call if possible
            is_jitted = True
        except Exception as e:
            warnings.warn(f"Numba JIT failed for {func.__name__}: {e}")
            jitted = func
            is_jitted = False
    except ImportError:
        jitted = func
        is_jitted = False

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return jitted(*args, **kwargs)

    wrapper.is_jitted = is_jitted
    wrapper.original = original
    wrapper.jitted = jitted

    def benchmark(*args, n=100, **kwargs):
        # Time original
        start = time.perf_counter()
        for _ in range(n):
            original(*args, **kwargs)
        t_orig = (time.perf_counter() - start) / n

        # Time JIT version
        start = time.perf_counter()
        for _ in range(n):
            jitted(*args, **kwargs)
        t_jit = (time.perf_counter() - start) / n

        speedup = t_orig / t_jit if t_jit > 0 else float('inf')
        print(f"{func.__name__}: original={t_orig:.6f}s, "
              f"jit={t_jit:.6f}s, speedup={speedup:.1f}x")
        return {"original": t_orig, "jit": t_jit, "speedup": speedup}

    wrapper.benchmark = benchmark
    return wrapper

@jit_fallback
def monte_carlo_pi(n_samples: int) -> float:
    count = 0
    for i in range(n_samples):
        x = np.random.random()
        y = np.random.random()
        if x*x + y*y <= 1.0:
            count += 1
    return 4.0 * count / n_samples