Hands-on lab — Day 171: Feature Engineering

"""
Feature Engineering reference library implementation.
"""
import numpy as np


def encode_cyclical_time(timestamps: np.ndarray, period: float = 24.0) -> tuple[np.ndarray, np.ndarray]:
    """
    Encode periodic timestamps into continuous 2D coordinates:
    sin_feat = sin(2 * pi * t / period)
    cos_feat = cos(2 * pi * t / period)
    Preserves distance continuity across boundary (e.g. 23:59 and 00:01).
    """
    t = np.asarray(timestamps, dtype=float)
    radians = 2.0 * np.pi * t / period
    sin_feat = np.sin(radians)
    cos_feat = np.cos(radians)
    return sin_feat, cos_feat


def compute_polynomial_interactions(X: np.ndarray) -> np.ndarray:
    """
    Generate original features + all pairwise interaction terms x_i * x_j for i <= j.
    """
    X = np.asarray(X, dtype=float)
    n_samples, n_features = X.shape
    
    terms = [X]
    for i in range(n_features):
        for j in range(i, n_features):
            interaction = (X[:, i] * X[:, j])[:, np.newaxis]
            terms.append(interaction)
            
    return np.hstack(terms)


def compute_group_aggregations(
    groups_train: np.ndarray, values_train: np.ndarray, groups_test: np.ndarray
) -> np.ndarray:
    """
    Compute training group statistics (mean, std) and map to test data.
    Unseen test groups fallback to global training statistics.
    Returns array of shape (N_test, 2) [group_mean, group_std].
    """
    groups_tr = np.asarray(groups_train)
    vals_tr = np.asarray(values_train, dtype=float)
    groups_te = np.asarray(groups_test)
    
    global_mean = float(np.mean(vals_tr))
    global_std = float(np.std(vals_tr))
    if global_std < 1e-9:
        global_std = 1.0
        
    unique_groups = np.unique(groups_tr)
    group_stats = {}
    for g in unique_groups:
        mask = groups_tr == g
        g_vals = vals_tr[mask]
        g_mean = float(np.mean(g_vals))
        g_std = float(np.std(g_vals)) if len(g_vals) > 1 else global_std
        if g_std < 1e-9:
            g_std = 1.0
        group_stats[g] = (g_mean, g_std)
        
    out = np.zeros((len(groups_te), 2), dtype=float)
    for i, g in enumerate(groups_te):
        if g in group_stats:
            out[i, 0], out[i, 1] = group_stats[g]
        else:
            out[i, 0], out[i, 1] = global_mean, global_std
            
    return out

"""
Tests for reference feature engineering implementation.
"""
import pytest
import numpy as np
import engineering_lib as fe


def test_cyclical_time_distance_continuity():
    # Hour 23.0 and Hour 0.0 (1 hour apart across midnight boundary)
    times = np.array([23.0, 0.0, 1.0, 12.0])
    sin_feat, cos_feat = fe.encode_cyclical_time(times, period=24.0)
    
    pt_23 = np.array([sin_feat[0], cos_feat[0]])
    pt_0 = np.array([sin_feat[1], cos_feat[1]])
    pt_1 = np.array([sin_feat[2], cos_feat[2]])
    pt_12 = np.array([sin_feat[3], cos_feat[3]])
    
    # Distance between 23:00 and 00:00 must equal distance between 00:00 and 01:00
    dist_23_0 = np.linalg.norm(pt_23 - pt_0)
    dist_0_1 = np.linalg.norm(pt_0 - pt_1)
    dist_0_12 = np.linalg.norm(pt_0 - pt_12)
    
    assert np.isclose(dist_23_0, dist_0_1, atol=1e-5)
    # Opposite times (00:00 vs 12:00) should be at maximum diameter distance 2.0
    assert np.isclose(dist_0_12, 2.0, atol=1e-5)


def test_polynomial_interactions_shape():
    # 3 features: original (3) + pairwise products 3*(3+1)/2 = 6 interactions -> total 9 columns
    X = np.ones((10, 3))
    X_poly = fe.compute_polynomial_interactions(X)
    assert X_poly.shape == (10, 9)


def test_group_aggregations_leak_free():
    groups_tr = np.array(["A", "A", "B", "B"])
    vals_tr = np.array([10.0, 20.0, 100.0, 200.0]) # Mean A = 15, Mean B = 150
    groups_te = np.array(["A", "B", "C"]) # C is unseen
    
    stats_te = fe.compute_group_aggregations(groups_tr, vals_tr, groups_te)
    
    assert np.isclose(stats_te[0, 0], 15.0)  # A mean
    assert np.isclose(stats_te[1, 0], 150.0) # B mean
    # C should fallback to global training mean: (10+20+100+200)/4 = 82.5
    assert np.isclose(stats_te[2, 0], 82.5)

lesson_id: D171
day: 171
kind: feature-engineering-flywheel
languages:
  - python
setup_commands:
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
run_commands:
  - .venv/bin/python examples/engineering_lib.py
test_commands:
  - ./tests/run_tests.sh
cleanup_commands:
  - rm -rf .venv __pycache__ .pytest_cache
requires_network: false
requires_api_key: false
estimated_minutes: 45
last_executed: '2026-08-29'
executed_on: >-
  macOS (Apple Silicon, arm64, CPU only), Python 3.14.0, numpy 2.5.2,
  scikit-learn 1.9.0, pytest 9.1.1, scipy 1.15.2 -- bash tests/run_tests.sh -> 4 checks,
  0 failure(s), exit 0. pytest examples -v -> 3 passed. pytest starter -v -> 2 passed.
  Verified cyclical time coordinate encoding, pairwise polynomial interactions, and leak-free group aggregations.

numpy==2.5.2
scikit-learn==1.9.0
pytest==9.1.1
scipy==1.15.2

"""
Feature Engineering starter library.
"""
import numpy as np


def encode_cyclical_time(timestamps: np.ndarray, period: float = 24.0) -> tuple[np.ndarray, np.ndarray]:
    """Encode periodic timestamps into continuous (sin, cos) cyclical coordinate pairs."""
    raise NotImplementedError("Implement encode_cyclical_time")


def compute_polynomial_interactions(X: np.ndarray) -> np.ndarray:
    """Generate all pairwise product interactions x_i * x_j for i <= j."""
    raise NotImplementedError("Implement compute_polynomial_interactions")


def compute_group_aggregations(groups_train: np.ndarray, values_train: np.ndarray, groups_test: np.ndarray) -> np.ndarray:
    """Compute leak-free group mean and std on training data, mapping onto test data with global fallback."""
    raise NotImplementedError("Implement compute_group_aggregations")

"""
Tests for starter feature engineering.
"""
import pytest
import numpy as np
import engineering_lib as fe


def test_cyclical_stub():
    with pytest.raises(NotImplementedError):
        fe.encode_cyclical_time(np.array([0.0, 12.0]))


def test_poly_stub():
    with pytest.raises(NotImplementedError):
        fe.compute_polynomial_interactions(np.zeros((5, 2)))

#!/usr/bin/env bash
# Day 171 lab harness: "Feature Engineering"
set -u

LAB_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$LAB_DIR"

PYTHON="${PYTHON:-../../../../.venv-tools/bin/python3}"
PYTEST="${PYTEST:-../../../../.venv-tools/bin/pytest}"

CHECKS=0
FAILURES=0

ok() {
  CHECKS=$((CHECKS + 1))
  echo "  ok: $1"
}

fail() {
  CHECKS=$((CHECKS + 1))
  FAILURES=$((FAILURES + 1))
  echo "  FAIL: $1"
}

echo "=== 1. Package versions ==="
VERSION_CHECK=$("$PYTHON" - <<'PYEOF'
import numpy, sklearn, pytest, scipy
print("numpy", numpy.__version__)
print("scikit-learn", sklearn.__version__)
print("pytest", pytest.__version__)
print("scipy", scipy.__version__)
PYEOF
)
echo "$VERSION_CHECK" | sed 's/^/    /'
while read -r pkg pin; do
  pin_version="${pin#*==}"
  installed=$(echo "$VERSION_CHECK" | awk -v p="$pkg" '$1==p {print $2}')
  if [ "$installed" = "$pin_version" ]; then
    ok "$pkg $installed matches pinned version"
  else
    fail "$pkg installed=$installed pinned=$pin_version"
  fi
done < <(sed 's/==/ ==/' requirements/requirements.txt)

echo ""
echo "=== 2. Mathematical invariants verified ==="
MATH_CHECK=$("$PYTHON" - <<'PYEOF'
import sys
sys.path.insert(0, "examples")
import numpy as np
import engineering_lib as fe

# Trigonometric unit circle identity: sin^2 + cos^2 = 1.0
times = np.linspace(0, 24, 100)
s, c = fe.encode_cyclical_time(times, period=24.0)
assert np.allclose(s**2 + c**2, 1.0), "Trig identity sin^2 + cos^2 = 1 violated"

print("MATH_OK")
PYEOF
)

if [ "$MATH_CHECK" = "MATH_OK" ]; then
  ok "Cyclical time encoding unit-circle invariant verified"
else
  fail "Mathematical verification failed: $MATH_CHECK"
fi

echo ""
echo "=== 3. Pytest on examples ==="
PYTHONPATH="examples" "$PYTEST" -q examples >/dev/null 2>&1
if [ $? -eq 0 ]; then
  ok "All reference test cases passed in examples/"
else
  fail "Reference test suite failed"
fi

echo ""
echo "=== 4. Pytest on starter ==="
PYTHONPATH="starter" "$PYTEST" -q starter >/dev/null 2>&1
if [ $? -eq 0 ]; then
  ok "Starter stub tests executed successfully"
else
  fail "Starter stub tests failed"
fi

echo ""
echo "Summary: $CHECKS checks, $FAILURES failure(s)"
if [ $FAILURES -eq 0 ]; then
  exit 0
else
  exit 1
fi