Hands-on lab — Day 142: Supervised, Unsupervised, and Reinforcement Learning

"""Three kinds of feedback, built from scratch and measured.

The taxonomy in this lesson is not a taxonomy of algorithms. It is a
taxonomy of the *feedback signal* a learner is given:

* supervised   -- instructive feedback: for every input you are told the
                  correct output, so the error is defined per example;
* unsupervised -- no feedback at all: there is no correct output, only
                  structure, and structure is not unique;
* reinforcement -- evaluative and delayed feedback: you are told how good
                  the action you took was, never what the best action
                  would have been, and often only much later.

Everything here is deterministic given a seed. The bandit and the
gridworld are written from first principles in NumPy so that the shape of
each feedback signal is visible in the code rather than hidden inside a
library.
"""

from __future__ import annotations

import numpy as np
from itertools import permutations

from sklearn.cluster import KMeans
from sklearn.datasets import load_iris
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import adjusted_rand_score, silhouette_score
from sklearn.neighbors import KNeighborsClassifier
from sklearn.preprocessing import StandardScaler

# --------------------------------------------------------------------------
# Shared helpers
# --------------------------------------------------------------------------


def iris_features_and_labels():
    """The iris measurements and the species column, as plain arrays."""
    return load_iris(return_X_y=True)


def accuracy(y_true, y_pred) -> float:
    """Fraction of positions where two label arrays agree."""
    y_true = np.asarray(y_true)
    y_pred = np.asarray(y_pred)
    return float(np.mean(y_true == y_pred))


def standardise(X):
    """Centre each column at zero and scale it to unit variance."""
    return StandardScaler().fit_transform(X)


# --------------------------------------------------------------------------
# 1. Supervised: instructive feedback
# --------------------------------------------------------------------------


def supervised_score(X, y, train_idx, test_idx, n_neighbors: int = 5) -> float:
    """Fit a k-NN on the training rows and score it on the held-out rows.

    This is the whole shape of supervised learning: every training row
    carries its own answer, so the learner can measure and reduce a
    per-example error.
    """
    model = KNeighborsClassifier(n_neighbors=n_neighbors)
    model.fit(X[train_idx], y[train_idx])
    return accuracy(y[test_idx], model.predict(X[test_idx]))


def split_indices(n: int, n_train: int, seed: int):
    """A deterministic random split of range(n) into train and test halves."""
    order = np.random.default_rng(seed).permutation(n)
    return order[:n_train], order[n_train:]


# --------------------------------------------------------------------------
# 2-5. Unsupervised: no feedback at all
# --------------------------------------------------------------------------


def cluster(X, k: int, seed: int = 0):
    """k-means with a fixed seed and a fixed number of restarts."""
    return KMeans(n_clusters=k, n_init=10, random_state=seed).fit(X)


def raw_cluster_accuracy(y_true, cluster_ids) -> float:
    """Compare cluster ids to true labels as if the ids meant something.

    They do not. k-means numbers its clusters by where its own centroids
    happened to land, so this figure is an artefact of the numbering.
    """
    return accuracy(y_true, cluster_ids)


def best_permutation_accuracy(y_true, cluster_ids):
    """The honest version: try every relabelling and keep the best.

    Returns ``(accuracy, mapping)`` where ``mapping[i]`` is the true label
    assigned to cluster ``i``. This is the number people *mean* when they
    say a clustering "recovered the classes", and computing it requires
    the labels -- which unsupervised learning, by definition, does not
    have.
    """
    y_true = np.asarray(y_true)
    cluster_ids = np.asarray(cluster_ids)
    labels = sorted(set(int(v) for v in y_true))
    ids = sorted(set(int(v) for v in cluster_ids))
    best = (-1.0, None)
    for perm in permutations(labels, len(ids)):
        mapped = np.empty_like(cluster_ids)
        for cid, lab in zip(ids, perm):
            mapped[cluster_ids == cid] = lab
        score = accuracy(y_true, mapped)
        if score > best[0]:
            best = (score, tuple(perm))
    return best


def cluster_confusion(y_true, cluster_ids, n_labels: int, n_clusters: int):
    """Rows are true classes, columns are cluster ids; entries are counts."""
    y_true = np.asarray(y_true)
    cluster_ids = np.asarray(cluster_ids)
    table = np.zeros((n_labels, n_clusters), dtype=int)
    for t, c in zip(y_true, cluster_ids):
        table[int(t), int(c)] += 1
    return table


def inertia_curve(X, ks, seed: int = 0):
    """Within-cluster sum of squares for each k in ``ks``."""
    return [float(cluster(X, k, seed=seed).inertia_) for k in ks]


def silhouette_curve(X, ks, seed: int = 0):
    """Mean silhouette score for each k in ``ks`` (k >= 2 only)."""
    out = []
    for k in ks:
        assignment = cluster(X, k, seed=seed).labels_
        out.append(float(silhouette_score(X, assignment)))
    return out


def agreement(labels_a, labels_b) -> float:
    """Adjusted Rand index: agreement between two partitions, ignoring names."""
    return float(adjusted_rand_score(labels_a, labels_b))


# --------------------------------------------------------------------------
# 6. Reinforcement: evaluative feedback, on a bandit built from scratch
# --------------------------------------------------------------------------


class GaussianBandit:
    """A k-armed bandit. Pulling arm i returns N(mean_i, 1).

    The defining property is what you are *not* told: pulling arm i tells
    you the reward for arm i on this pull and nothing whatever about the
    other k-1 arms. That is evaluative feedback. A supervised learner in
    the same position would have been handed the whole reward vector.
    """

    def __init__(self, k: int = 10, seed: int = 0):
        self.k = k
        self.rng = np.random.default_rng(seed)
        self.means = self.rng.normal(0.0, 1.0, size=k)
        self.best_arm = int(np.argmax(self.means))

    def pull(self, arm: int) -> float:
        """Return the reward for one pull of ``arm`` -- and only that arm."""
        return float(self.rng.normal(self.means[arm], 1.0))

    def full_feedback(self) -> np.ndarray:
        """The reward vector a supervised learner would have been given.

        No reinforcement-learning agent ever sees this. It exists here only
        so the lab can measure what the missing information is worth.
        """
        return self.rng.normal(self.means, 1.0)


def run_bandit(k: int = 10, steps: int = 1000, epsilon: float = 0.1, seed: int = 0):
    """Epsilon-greedy action-value learning, written out in full.

    ``epsilon=0.0`` is the pure greedy agent. Returns a dict with the
    reward at each step and whether the optimal arm was chosen at each
    step.
    """
    bandit = GaussianBandit(k=k, seed=seed)
    chooser = np.random.default_rng(seed + 10_000)
    estimates = np.zeros(k)
    counts = np.zeros(k, dtype=int)
    rewards = np.zeros(steps)
    optimal = np.zeros(steps, dtype=bool)

    for t in range(steps):
        if chooser.random() < epsilon:
            arm = int(chooser.integers(k))
        else:
            arm = int(np.argmax(estimates))
        reward = bandit.pull(arm)
        counts[arm] += 1
        # incremental mean: estimate += (reward - estimate) / count
        estimates[arm] += (reward - estimates[arm]) / counts[arm]
        rewards[t] = reward
        optimal[t] = arm == bandit.best_arm

    return {
        "rewards": rewards,
        "optimal": optimal,
        "estimates": estimates,
        "counts": counts,
        "true_means": bandit.means,
        "best_arm": bandit.best_arm,
    }


def average_bandit(runs: int = 200, **kwargs):
    """Average ``run_bandit`` over independent problems, as the field does.

    A single bandit run is almost pure noise; the published comparison
    between greedy and epsilon-greedy is an average over many problems.
    Returns ``(mean_reward, fraction_optimal)`` over the whole horizon.
    """
    reward_total = 0.0
    optimal_total = 0.0
    for r in range(runs):
        out = run_bandit(seed=r, **kwargs)
        reward_total += float(np.mean(out["rewards"]))
        optimal_total += float(np.mean(out["optimal"]))
    return reward_total / runs, optimal_total / runs


# --------------------------------------------------------------------------
# 7. Delayed feedback and credit assignment: a gridworld, from scratch
# --------------------------------------------------------------------------


class GridWorld:
    """A ``size`` x ``size`` grid. Reward +1 at the goal, 0 everywhere else.

    The agent starts top-left, the goal is bottom-right, and every step
    costs nothing. So the feedback for the very first move arrives only
    after the goal is reached -- which is the credit-assignment problem in
    its smallest honest form.
    """

    ACTIONS = ((-1, 0), (1, 0), (0, -1), (0, 1))  # up, down, left, right

    def __init__(self, size: int = 5):
        self.size = size
        self.start = (0, 0)
        self.goal = (size - 1, size - 1)

    def n_states(self) -> int:
        return self.size * self.size

    def state_index(self, pos) -> int:
        return pos[0] * self.size + pos[1]

    def step(self, pos, action: int):
        """Return ``(next_pos, reward, done)``. Walls block movement."""
        dr, dc = self.ACTIONS[action]
        r = min(max(pos[0] + dr, 0), self.size - 1)
        c = min(max(pos[1] + dc, 0), self.size - 1)
        nxt = (r, c)
        if nxt == self.goal:
            return nxt, 1.0, True
        return nxt, 0.0, False

    def shortest_path_length(self) -> int:
        """The optimal number of steps from start to goal on an open grid."""
        return (self.size - 1) * 2


def argmax_random_tiebreak(values, rng) -> int:
    """Return an index of the maximum, choosing uniformly among ties.

    ``np.argmax`` returns the *lowest* index that attains the maximum. On a
    Q-table that starts at all zeros every row is one big tie, so the greedy
    branch of an epsilon-greedy agent degenerates into a constant action.
    This lab measures what that costs; see exercise 7.
    """
    values = np.asarray(values)
    best = np.flatnonzero(values == values.max())
    return int(best[rng.integers(len(best))])


def q_learning(
    world: GridWorld,
    episodes: int = 300,
    alpha: float = 0.5,
    gamma: float = 0.95,
    epsilon: float = 0.2,
    max_steps: int = 200,
    seed: int = 0,
    break_ties_randomly: bool = True,
):
    """Tabular Q-learning, written out so the bootstrap is visible.

    The update ``Q[s,a] += alpha * (r + gamma * max_a' Q[s',a'] - Q[s,a])``
    is how a reward that only ever appears at the goal travels backwards to
    the first move. Nobody ever tells the agent which action was correct; it
    infers it from its own later estimates.

    Set ``break_ties_randomly=False`` to get the ``np.argmax`` behaviour that
    exercise 7 measures as a failure.
    """
    rng = np.random.default_rng(seed)
    n_actions = len(world.ACTIONS)
    Q = np.zeros((world.n_states(), n_actions))
    lengths = []
    reached = 0

    for _ in range(episodes):
        pos = world.start
        done = False
        for step in range(max_steps):
            s = world.state_index(pos)
            if rng.random() < epsilon:
                a = int(rng.integers(n_actions))
            elif break_ties_randomly:
                a = argmax_random_tiebreak(Q[s], rng)
            else:
                a = int(np.argmax(Q[s]))
            nxt, reward, done = world.step(pos, a)
            s_next = world.state_index(nxt)
            target = reward + (0.0 if done else gamma * float(np.max(Q[s_next])))
            Q[s, a] += alpha * (target - Q[s, a])
            pos = nxt
            if done:
                break
        reached += int(done)
        lengths.append(step + 1)

    return Q, lengths, reached


def greedy_path_length(world: GridWorld, Q, max_steps: int = 200) -> int:
    """Follow the learned policy with no exploration and count the steps.

    Returns ``max_steps`` if the policy never reaches the goal, which is the
    honest answer for a Q-table the reward never reached.
    """
    pos = world.start
    for step in range(max_steps):
        a = int(np.argmax(Q[world.state_index(pos)]))
        pos, _reward, done = world.step(pos, a)
        if done:
            return step + 1
    return max_steps


def states_with_nonzero_value(Q) -> int:
    """How many grid squares the reward signal has actually reached."""
    return int(np.sum(np.max(Q, axis=1) > 0.0))


# --------------------------------------------------------------------------
# 8. Why logged bandit data is not a supervised dataset
# --------------------------------------------------------------------------


def logged_policy_dataset(k: int = 10, steps: int = 2000, epsilon: float = 0.1, seed: int = 0):
    """Turn a bandit run into the (arm, reward) table a log would contain."""
    out = run_bandit(k=k, steps=steps, epsilon=epsilon, seed=seed)
    arms = []
    rewards = []
    bandit = GaussianBandit(k=k, seed=seed)
    chooser = np.random.default_rng(seed + 10_000)
    estimates = np.zeros(k)
    counts = np.zeros(k, dtype=int)
    for _ in range(steps):
        if chooser.random() < epsilon:
            arm = int(chooser.integers(k))
        else:
            arm = int(np.argmax(estimates))
        reward = bandit.pull(arm)
        counts[arm] += 1
        estimates[arm] += (reward - estimates[arm]) / counts[arm]
        arms.append(arm)
        rewards.append(reward)
    return np.array(arms), np.array(rewards), out["best_arm"], counts


def arm_pull_counts(counts) -> dict:
    """Pulls per arm, as a plain dict, for reporting."""
    return {i: int(c) for i, c in enumerate(counts)}


# --------------------------------------------------------------------------
# 9. Labels are the expensive part: semi-supervised in its simplest form
# --------------------------------------------------------------------------


def label_budget_curve(X, y, budgets, seed: int = 0, n_neighbors: int = 1):
    """Accuracy as a function of how many labelled rows you can afford.

    Rows are chosen uniformly at random, which is what you get when nobody
    thinks about *which* rows to label.
    """
    train_idx, test_idx = split_indices(len(y), 100, seed=seed)
    rng = np.random.default_rng(seed + 7)
    scores = []
    for b in budgets:
        chosen = rng.permutation(train_idx)[:b]
        model = KNeighborsClassifier(n_neighbors=min(n_neighbors, b))
        model.fit(X[chosen], y[chosen])
        scores.append(accuracy(y[test_idx], model.predict(X[test_idx])))
    return scores


def cluster_then_label(X, y, k: int, seed: int = 0):
    """Spend the label budget on one representative row per cluster.

    Cluster the *unlabelled* data, label the row closest to each centroid,
    and propagate that label to the whole cluster. This is unsupervised
    learning being used to make supervised learning cheaper, which is the
    honest reason the two categories sit in one lesson.
    """
    train_idx, test_idx = split_indices(len(y), 100, seed=seed)
    km = cluster(X[train_idx], k, seed=seed)
    representatives = []
    for c in range(k):
        members = np.where(km.labels_ == c)[0]
        d = np.linalg.norm(X[train_idx][members] - km.cluster_centers_[c], axis=1)
        representatives.append(int(members[int(np.argmin(d))]))
    rep_labels = y[train_idx][representatives]
    propagated = rep_labels[km.labels_]
    model = KNeighborsClassifier(n_neighbors=1)
    model.fit(X[train_idx], propagated)
    return accuracy(y[test_idx], model.predict(X[test_idx])), len(representatives)


def full_supervision_score(X, y, seed: int = 0) -> float:
    """The ceiling: every training row labelled."""
    train_idx, test_idx = split_indices(len(y), 100, seed=seed)
    model = KNeighborsClassifier(n_neighbors=1)
    model.fit(X[train_idx], y[train_idx])
    return accuracy(y[test_idx], model.predict(X[test_idx]))


# --------------------------------------------------------------------------
# 10. The decision function: which kind of problem is this?
# --------------------------------------------------------------------------


def problem(*, has_labels: bool, actions_change_the_data: bool, feedback_is_immediate: bool):
    """Build the three-answer description a problem must supply."""
    return {
        "has_labels": has_labels,
        "actions_change_the_data": actions_change_the_data,
        "feedback_is_immediate": feedback_is_immediate,
    }


def classify_problem(spec: dict) -> str:
    """Name the learning setting a problem actually belongs to.

    Order matters. The question that decides the most is whether your
    actions change what data you see next -- because that single property
    is what makes a problem reinforcement learning no matter how many
    labels you have.
    """
    for key in ("has_labels", "actions_change_the_data", "feedback_is_immediate"):
        if key not in spec:
            raise KeyError(f"problem description is missing {key!r}")
    if spec["actions_change_the_data"]:
        if spec["feedback_is_immediate"]:
            return "reinforcement learning: contextual bandit"
        return "reinforcement learning: sequential, with delayed credit"
    if spec["has_labels"]:
        return "supervised learning"
    return "unsupervised learning"


# --------------------------------------------------------------------------
# Reporting helpers used by both the tests and report_measurements.py
# --------------------------------------------------------------------------


def logged_arm_means(arms, rewards) -> dict:
    """Mean logged reward per arm -- the only thing a log can tell you."""
    arms = np.asarray(arms)
    rewards = np.asarray(rewards)
    return {int(a): float(np.mean(rewards[arms == a])) for a in sorted(set(arms.tolist()))}


def logged_best_arm(arms, rewards) -> int:
    """The arm a supervised model trained on the log would choose."""
    means = logged_arm_means(arms, rewards)
    return max(means, key=means.get)


def log_verdicts(seeds, epsilon: float, k: int = 10, steps: int = 2000):
    """For each seed, whether the log's favourite arm is the truly best arm.

    Returns a list of ``(seed, distinct_arms, logged_pick, true_best, correct)``.
    """
    rows = []
    for seed in seeds:
        arms, rewards, true_best, _counts = logged_policy_dataset(
            k=k, steps=steps, epsilon=epsilon, seed=seed
        )
        pick = logged_best_arm(arms, rewards)
        rows.append(
            (seed, len(set(arms.tolist())), pick, int(true_best), bool(pick == int(true_best)))
        )
    return rows


def average_label_budget_curve(X, y, budgets, repeats: int = 40, base_seed: int = 142):
    """``label_budget_curve`` averaged over independent splits and draws.

    A single split of 150 rows is far too noisy to read a trend from -- one
    run of this curve is not even monotone. Averaging is not decoration; it
    is the difference between a measurement and an anecdote.
    """
    total = np.zeros(len(budgets))
    for s in range(repeats):
        total += np.array(label_budget_curve(X, y, budgets, seed=base_seed + s))
    return [float(v) for v in total / repeats]


def average_cluster_then_label(X, y, k: int, repeats: int = 40, base_seed: int = 142) -> float:
    """``cluster_then_label`` averaged over the same splits, for comparison."""
    scores = [cluster_then_label(X, y, k, seed=base_seed + s)[0] for s in range(repeats)]
    return float(np.mean(scores))


def average_full_supervision(X, y, repeats: int = 40, base_seed: int = 142) -> float:
    """The every-row-labelled ceiling, averaged over the same splits."""
    scores = [full_supervision_score(X, y, seed=base_seed + s) for s in range(repeats)]
    return float(np.mean(scores))


def value_spread_by_episode(world, episode_counts, seed: int = 0):
    """How many states the reward has reached after each episode count.

    This is credit assignment made countable: with a reward only at the
    goal, the first episode leaves exactly one state with a non-zero value,
    the second leaves two, and so on.
    """
    rows = []
    for n in episode_counts:
        Q, _lengths, _reached = q_learning(world, episodes=n, seed=seed)
        rows.append((n, states_with_nonzero_value(Q), greedy_path_length(world, Q)))
    return rows

#!/usr/bin/env python3
"""Print every measured pair in this lab as one table.

The harness compares this output byte for byte against
expected-output/measured-values.txt, so the report is not a convenience:
it is how the lab notices that a number in the lesson has gone stale.
"""

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

import numpy as np  # noqa: E402

import feedback_lib as f  # noqa: E402


def rule(title: str) -> None:
    print()
    print(title)
    print("-" * len(title))


def main() -> None:
    X, y = f.iris_features_and_labels()

    print("Day 142 -- three kinds of feedback, measured")
    print("=" * 44)

    rule("1. Supervised: an answer per example")
    train, test = f.split_indices(len(y), 100, seed=142)
    print(f"  5-NN on iris, 100 train / 50 test      : {f.supervised_score(X, y, train, test):.4f}")

    rule("2. Unsupervised: the cluster numbers mean nothing")
    assignment = f.cluster(X, 3, seed=0).labels_
    raw = f.raw_cluster_accuracy(y, assignment)
    best, mapping = f.best_permutation_accuracy(y, assignment)
    print(f"  accuracy taking cluster ids literally  : {raw:.4f}")
    print(f"  accuracy after the best relabelling    : {best:.4f}  (mapping {mapping})")
    print(f"  difference, from numbering alone       : {best - raw:.4f}")

    rule("3. What k-means found on iris")
    table = f.cluster_confusion(y, assignment, 3, 3)
    print("  rows = species, columns = cluster id")
    for i, row in enumerate(table):
        print(f"    species {i}: {row.tolist()}")
    print("  species 0 is isolated exactly; species 1 and 2 share clusters")

    rule("4. Structure is not unique")
    scaled = f.cluster(f.standardise(X), 3, seed=0).labels_
    print(f"  ARI, raw clustering vs scaled          : {f.agreement(assignment, scaled):.4f}")
    print(f"  ARI, raw clustering vs true species    : {f.agreement(y, assignment):.4f}")
    print(f"  ARI, scaled clustering vs true species : {f.agreement(y, scaled):.4f}")
    print("  standardising moved AWAY from the species here")

    rule("5. k is a choice, not a discovery")
    ks = [2, 3, 4, 5, 6]
    inertia = f.inertia_curve(X, ks)
    silhouette = f.silhouette_curve(X, ks)
    for k, i, s in zip(ks, inertia, silhouette):
        print(f"    k={k}: inertia {i:8.3f}   silhouette {s:.4f}")
    print("  inertia falls at every step, so it can never choose k")
    print(f"  best silhouette is at k={ks[int(np.argmax(silhouette))]}; iris has 3 species")

    rule("6. Reinforcement: evaluative feedback (10 arms, 1000 steps, 200 runs)")
    rows = []
    for eps in (0.0, 0.01, 0.1):
        rows.append((eps,) + f.average_bandit(runs=200, k=10, steps=1000, epsilon=eps))
    for eps, reward, optimal in rows:
        print(f"    epsilon={eps:<5}: mean reward {reward:.4f}   optimal action {optimal:.4f}")
    print(f"  never exploring costs {rows[2][2] - rows[0][2]:.4f} of optimal-action rate")

    rule("7. Delayed feedback: a 5x5 gridworld, reward only at the goal")
    world = f.GridWorld(5)
    _q_bad, lengths_bad, reached_bad = f.q_learning(
        world, episodes=300, seed=0, break_ties_randomly=False
    )
    q_good, lengths_good, reached_good = f.q_learning(
        world, episodes=300, seed=0, break_ties_randomly=True
    )
    print(f"  shortest possible path                 : {world.shortest_path_length()} steps")
    print(f"  np.argmax tie-breaking, goal reached in : {reached_bad}/300 episodes")
    print(f"  random tie-breaking, goal reached in    : {reached_good}/300 episodes")
    print(f"  mean episode length, first 10           : {np.mean(lengths_good[:10]):.1f}")
    print(f"  mean episode length, last 10            : {np.mean(lengths_good[-10:]):.1f}")
    print(f"  greedy path after training              : {f.greedy_path_length(world, q_good)} steps")
    print("  how far the reward has travelled backwards:")
    for n, states, path in f.value_spread_by_episode(world, [1, 2, 3, 5, 10, 25, 50, 100, 300]):
        walk = "cannot reach goal" if path == 200 else f"{path} steps"
        print(f"    after {n:3d} episodes: {states:2d}/25 states valued, greedy policy {walk}")

    rule("8. A log is not a supervised dataset")
    arms, rewards, best_arm, counts = f.logged_policy_dataset(k=10, steps=2000, epsilon=0.1, seed=0)
    pulls = f.arm_pull_counts(counts)
    print(f"  seed 0, epsilon=0.1: best arm is {best_arm}, pulled {pulls[best_arm]} of 2000 times")
    print(f"  the other nine arms share {sum(v for a, v in pulls.items() if a != best_arm)} pulls")
    for eps, label in ((0.0, "greedy   "), (0.1, "epsilon=0.1")):
        verdicts = f.log_verdicts(range(8), epsilon=eps)
        right = sum(1 for row in verdicts if row[4])
        single = sum(1 for row in verdicts if row[1] == 1)
        print(f"  {label} logs: {right}/8 name the truly best arm; {single}/8 contain one arm only")
    arms1, rewards1, _b, counts1 = f.logged_policy_dataset(k=10, steps=2000, epsilon=0.1, seed=1)
    bandit1 = f.GaussianBandit(k=10, seed=1)
    means1 = f.logged_arm_means(arms1, rewards1)
    print("  the winner's curse, seed 1:")
    print(
        f"    arm 4: true {bandit1.means[4]:+.4f}  logged {means1[4]:+.4f}  "
        f"pulls {int(counts1[4])}"
    )
    print(
        f"    arm 1: true {bandit1.means[1]:+.4f}  logged {means1[1]:+.4f}  "
        f"pulls {int(counts1[1])}"
    )
    print(f"    the log picks arm {f.logged_best_arm(arms1, rewards1)}; the truth is arm 4")

    rule("9. Labels are the expensive part (iris, 1-NN, 40 repeats)")
    budgets = [3, 5, 10, 20, 50]
    curve = f.average_label_budget_curve(X, y, budgets, repeats=40)
    for b, score in zip(budgets, curve):
        print(f"    {b:2d} random labels : {score:.4f}")
    chosen = f.average_cluster_then_label(X, y, 3, repeats=40)
    ceiling = f.average_full_supervision(X, y, repeats=40)
    print(f"     3 chosen labels : {chosen:.4f}   (one per k-means cluster)")
    print(f"    100 labels       : {ceiling:.4f}   (every training row)")
    print(f"  three chosen labels beat three random ones by {chosen - curve[0]:.4f}")

    rule("10. Naming the setting before choosing an algorithm")
    cases = [
        ("labels, actions inert, feedback now  ", True, False, True),
        ("no labels, actions inert             ", False, False, True),
        ("labels, actions change the data, now ", True, True, True),
        ("labels, actions change the data, late", True, True, False),
    ]
    for label, has, acts, now in cases:
        verdict = f.classify_problem(
            f.problem(
                has_labels=has, actions_change_the_data=acts, feedback_is_immediate=now
            )
        )
        print(f"    {label} -> {verdict}")


if __name__ == "__main__":
    main()

"""The reference solutions: ten claims about the three kinds of feedback.

Every number here was captured from a real run of this file on the
authoring machine. If a number changes, the claim in the lesson is wrong
and one of the two must be fixed.
"""

import numpy as np
import pytest

import feedback_lib as f


@pytest.fixture(scope="module")
def iris():
    return f.iris_features_and_labels()


# --- 1. Supervised: the answer key is in the data -------------------------


def test_01_supervised_learning_has_a_per_example_answer(iris):
    X, y = iris
    train, test = f.split_indices(len(y), 100, seed=142)
    score = f.supervised_score(X, y, train, test, n_neighbors=5)
    assert score == 0.92
    # The defining property: every training row carries its own answer.
    assert len(y[train]) == 100
    assert set(np.unique(y[train]).tolist()) == {0, 1, 2}


# --- 2. Unsupervised: cluster numbers mean nothing -------------------------


def test_02_cluster_ids_are_arbitrary_and_raw_accuracy_is_meaningless(iris):
    X, y = iris
    assignment = f.cluster(X, 3, seed=0).labels_
    raw = f.raw_cluster_accuracy(y, assignment)
    best, mapping = f.best_permutation_accuracy(y, assignment)
    assert raw == 0.24
    assert best == pytest.approx(0.8933333333333333)
    assert mapping == (1, 0, 2)
    # Sixty-five accuracy points separate the same clustering from itself.
    assert round(best - raw, 4) == 0.6533


# --- 3. What k-means actually found on iris -------------------------------


def test_03_kmeans_isolates_one_species_and_blends_two(iris):
    X, y = iris
    assignment = f.cluster(X, 3, seed=0).labels_
    table = f.cluster_confusion(y, assignment, 3, 3)
    # Setosa lands entirely in one cluster; the other two species do not.
    assert table[0].tolist() == [0, 50, 0]
    assert table[1].tolist() == [48, 0, 2]
    assert table[2].tolist() == [14, 0, 36]
    assert int(table[0].max()) == 50
    assert int(table[1].max()) == 48 and int(table[2].max()) == 36


# --- 4. Structure is not unique -------------------------------------------


def test_04_standardising_changes_the_clustering_and_here_makes_it_worse(iris):
    X, y = iris
    raw_assignment = f.cluster(X, 3, seed=0).labels_
    scaled_assignment = f.cluster(f.standardise(X), 3, seed=0).labels_
    assert f.agreement(raw_assignment, scaled_assignment) == pytest.approx(0.8036117420390129)
    raw_vs_truth = f.agreement(y, raw_assignment)
    scaled_vs_truth = f.agreement(y, scaled_assignment)
    assert raw_vs_truth == pytest.approx(0.7302382722834697)
    assert scaled_vs_truth == pytest.approx(0.6201351808870379)
    # The folklore says always standardise. On this dataset it loses ground.
    assert scaled_vs_truth < raw_vs_truth


# --- 5. k is a choice, not a discovery ------------------------------------


def test_05_inertia_cannot_choose_k_and_silhouette_chooses_the_wrong_one(iris):
    X, _y = iris
    ks = [2, 3, 4, 5, 6]
    inertia = f.inertia_curve(X, ks)
    # Inertia falls monotonically, so "minimise inertia" always answers k = n.
    assert all(a > b for a, b in zip(inertia, inertia[1:]))
    assert [round(v, 3) for v in inertia] == [152.348, 78.851, 57.228, 46.446, 39.04]
    silhouette = f.silhouette_curve(X, ks)
    assert [round(v, 4) for v in silhouette] == [0.681, 0.5528, 0.4981, 0.4887, 0.3648]
    # The best silhouette is at k = 2. Iris has three species.
    assert ks[int(np.argmax(silhouette))] == 2


# --- 6. Reinforcement: evaluative feedback and the cost of not exploring ---


def test_06_greedy_locks_on_and_epsilon_greedy_does_not():
    greedy_reward, greedy_optimal = f.average_bandit(runs=200, k=10, steps=1000, epsilon=0.0)
    small_reward, small_optimal = f.average_bandit(runs=200, k=10, steps=1000, epsilon=0.01)
    explore_reward, explore_optimal = f.average_bandit(runs=200, k=10, steps=1000, epsilon=0.1)
    assert round(greedy_reward, 4) == 0.9838
    assert round(greedy_optimal, 4) == 0.313
    assert round(small_reward, 4) == 1.1314
    assert round(small_optimal, 4) == 0.4336
    assert round(explore_reward, 4) == 1.28
    assert round(explore_optimal, 4) == 0.708
    # Never exploring costs 39.49 points of optimal-action rate.
    assert round(explore_optimal - greedy_optimal, 4) == 0.3949
    assert explore_reward > small_reward > greedy_reward


def test_06b_a_bandit_only_ever_reveals_the_arm_you_pulled():
    bandit = f.GaussianBandit(k=10, seed=3)
    reward = bandit.pull(0)
    assert isinstance(reward, float)
    # There is no method that returns the reward you would have got from
    # another arm on that same pull -- except the one this lab added
    # purely so the missing information can be named.
    full = bandit.full_feedback()
    assert full.shape == (10,)


# --- 7. Delayed feedback, credit assignment, and one silent bug -----------


def test_07_argmax_tie_breaking_decides_whether_the_agent_learns_at_all():
    world = f.GridWorld(5)
    assert world.shortest_path_length() == 8

    _q_bad, lengths_bad, reached_bad = f.q_learning(
        world, episodes=300, seed=0, break_ties_randomly=False
    )
    assert reached_bad == 0
    assert lengths_bad[0] == 200 and lengths_bad[-1] == 200

    q_good, lengths_good, reached_good = f.q_learning(
        world, episodes=300, seed=0, break_ties_randomly=True
    )
    assert reached_good == 300
    assert round(float(np.mean(lengths_good[:10])), 1) == 46.8
    assert round(float(np.mean(lengths_good[-10:])), 1) == 10.4
    # With no exploration at all, the learned policy walks the shortest path.
    assert f.greedy_path_length(world, q_good) == 8


def test_07b_the_reward_travels_backwards_one_state_per_episode():
    world = f.GridWorld(5)
    rows = f.value_spread_by_episode(world, [1, 2, 3, 5, 10, 25, 50, 100, 300], seed=0)
    reached = {n: states for n, states, _path in rows}
    assert [reached[n] for n in (1, 2, 3, 5, 10)] == [1, 2, 3, 5, 10]
    assert reached[25] == 18 and reached[50] == 20
    assert reached[100] == 21 and reached[300] == 22
    paths = {n: path for n, _states, path in rows}
    # Before episode 25 the greedy policy cannot reach the goal at all.
    assert paths[10] == 200 and paths[25] == 8


# --- 8. Logged bandit data is not a supervised dataset --------------------


def test_08_a_log_records_only_what_the_logging_policy_chose():
    arms, _rewards, best, counts = f.logged_policy_dataset(
        k=10, steps=2000, epsilon=0.1, seed=0
    )
    pulls = f.arm_pull_counts(counts)
    assert best == 6
    assert pulls[6] == 1813
    assert sum(pulls.values()) == 2000
    # Nine arms share 187 pulls between them; none has enough data to judge.
    assert sum(v for a, v in pulls.items() if a != 6) == 187
    assert max(v for a, v in pulls.items() if a != 6) == 30
    assert len(arms) == 2000


def test_08b_a_greedy_log_confirms_whatever_it_locked_onto():
    verdicts = f.log_verdicts(range(8), epsilon=0.0)
    correct = [row for row in verdicts if row[4]]
    wrong = [row for row in verdicts if not row[4]]
    assert len(verdicts) == 8
    assert len(wrong) == 5 and len(correct) == 3
    # Four of the eight logs contain exactly one arm: the log cannot even
    # represent the question a supervised model would be asked.
    single_arm = [row for row in verdicts if row[1] == 1]
    assert len(single_arm) == 4
    assert all(row[2] == 0 for row in single_arm)
    assert [row[0] for row in single_arm] == [1, 2, 3, 6]

    explored = f.log_verdicts(range(8), epsilon=0.1)
    assert all(row[1] == 10 for row in explored)
    # Exploring more than doubles the hit rate -- and still gets one wrong.
    assert sum(1 for row in explored if row[4]) == 7
    assert [row for row in explored if not row[4]] == [(1, 10, 1, 4, False)]


def test_08c_the_winners_curse_is_why_the_explored_log_still_gets_one_wrong():
    """Seed 1 loses to an arm with a twentieth of the data, on purpose.

    The logging policy pulled the genuinely best arm 1524 times and a
    slightly worse arm 274 times. More data made the good arm's estimate
    *more accurate* -- and so it came in slightly under its true mean,
    while the thinly-sampled arm came in over. Taking an argmax over noisy
    estimates systematically favours whichever estimate is most inflated.
    Day 144 measures this same effect as model-selection bias.
    """
    arms, rewards, best, counts = f.logged_policy_dataset(
        k=10, steps=2000, epsilon=0.1, seed=1
    )
    bandit = f.GaussianBandit(k=10, seed=1)
    means = f.logged_arm_means(arms, rewards)
    assert best == 4
    assert int(counts[4]) == 1524 and int(counts[1]) == 274
    assert round(float(bandit.means[4]), 4) == 0.9054
    assert round(float(bandit.means[1]), 4) == 0.8216
    assert round(means[4], 4) == 0.8634
    assert round(means[1], 4) == 0.9262
    # The better arm is under-estimated; the worse arm is over-estimated.
    assert means[4] < float(bandit.means[4])
    assert means[1] > float(bandit.means[1])
    assert f.logged_best_arm(arms, rewards) == 1


# --- 9. Unsupervised learning buying supervised learning cheaper ----------


def test_09_three_chosen_labels_are_worth_about_nine_random_ones(iris):
    X, y = iris
    budgets = [3, 5, 10, 20, 50]
    curve = f.average_label_budget_curve(X, y, budgets, repeats=40)
    assert [round(v, 4) for v in curve] == [0.6455, 0.778, 0.896, 0.924, 0.947]
    # Averaging is not decoration: a single split is not even monotone.
    single = f.label_budget_curve(X, y, budgets, seed=142)
    assert single == [0.64, 0.92, 0.92, 0.86, 0.96]
    assert not all(a <= b for a, b in zip(single, single[1:]))
    assert all(a < b for a, b in zip(curve, curve[1:]))

    chosen = f.average_cluster_then_label(X, y, 3, repeats=40)
    assert round(chosen, 4) == 0.876
    # Same budget of three labels, 23 accuracy points better.
    assert round(chosen - curve[0], 4) == 0.2305
    # And it lands between five and ten randomly chosen labels.
    assert curve[1] < chosen < curve[2]
    ceiling = f.average_full_supervision(X, y, repeats=40)
    assert round(ceiling, 4) == 0.9535
    assert chosen < ceiling


# --- 10. Naming the setting before choosing the algorithm -----------------


def test_10_the_deciding_question_is_whether_your_actions_change_the_data():
    supervised = f.classify_problem(
        f.problem(has_labels=True, actions_change_the_data=False, feedback_is_immediate=True)
    )
    unsupervised = f.classify_problem(
        f.problem(has_labels=False, actions_change_the_data=False, feedback_is_immediate=True)
    )
    bandit = f.classify_problem(
        f.problem(has_labels=True, actions_change_the_data=True, feedback_is_immediate=True)
    )
    sequential = f.classify_problem(
        f.problem(has_labels=True, actions_change_the_data=True, feedback_is_immediate=False)
    )
    assert supervised == "supervised learning"
    assert unsupervised == "unsupervised learning"
    assert bandit == "reinforcement learning: contextual bandit"
    assert sequential == "reinforcement learning: sequential, with delayed credit"
    # Having labels does not make a problem supervised.
    assert bandit != supervised
    assert len({supervised, unsupervised, bandit, sequential}) == 4


def test_10b_an_incomplete_problem_description_is_refused():
    with pytest.raises(KeyError):
        f.classify_problem({"has_labels": True})

"""Machinery checks: the library itself behaves, before any claim is made.

These three tests are solved in both `starter/` and `examples/`. They are
here so that a broken helper reports itself as a broken helper rather than
as a surprising scientific result.
"""

import numpy as np
import pytest

import feedback_lib as f


def test_the_bandit_is_deterministic_given_a_seed():
    a = f.GaussianBandit(k=10, seed=5)
    b = f.GaussianBandit(k=10, seed=5)
    assert np.array_equal(a.means, b.means)
    assert a.best_arm == b.best_arm == int(np.argmax(a.means))
    # Different seeds give different problems, or the averaging is a lie.
    c = f.GaussianBandit(k=10, seed=6)
    assert not np.array_equal(a.means, c.means)


def test_the_gridworld_walls_block_movement_and_the_goal_ends_the_episode():
    world = f.GridWorld(4)
    # Moving up from the top-left corner leaves you where you are.
    assert world.step((0, 0), 0) == ((0, 0), 0.0, False)
    assert world.step((0, 0), 2) == ((0, 0), 0.0, False)
    # Moving into the goal pays 1.0 and terminates.
    nxt, reward, done = world.step((3, 2), 3)
    assert nxt == (3, 3) and reward == 1.0 and done is True
    # Every other transition pays nothing at all.
    assert world.step((1, 1), 1) == ((2, 1), 0.0, False)
    assert world.n_states() == 16
    assert world.shortest_path_length() == 6


def test_argmax_random_tiebreak_actually_spreads_over_ties():
    rng = np.random.default_rng(0)
    picks = {f.argmax_random_tiebreak(np.zeros(4), rng) for _ in range(200)}
    assert picks == {0, 1, 2, 3}
    # With a clear winner it is still an argmax.
    assert f.argmax_random_tiebreak(np.array([0.0, 9.0, 1.0, 2.0]), rng) == 1
    # np.argmax, by contrast, never leaves index 0 on an all-zero row.
    assert int(np.argmax(np.zeros(4))) == 0


def test_best_permutation_accuracy_is_never_worse_than_the_raw_number():
    y = np.array([0, 0, 1, 1, 2, 2])
    ids = np.array([2, 2, 0, 0, 1, 1])
    raw = f.raw_cluster_accuracy(y, ids)
    best, mapping = f.best_permutation_accuracy(y, ids)
    assert raw == 0.0
    assert best == 1.0
    # mapping[i] is the true label that cluster i turned out to hold.
    assert mapping == (1, 2, 0)
    with pytest.raises(AssertionError):
        assert best < raw

lesson_id: D142
day: 142
kind: guided-build
languages:
  - python
  - bash
setup_commands:
  - cd labs/sections/machine-learning/day-142-supervised-unsupervised-and-reinforcement-learning
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
  - >-
    .venv/bin/python3 -c "import numpy, sklearn; print(numpy.__version__,
    sklearn.__version__)"
run_commands:
  - .venv/bin/pytest examples -q
  - .venv/bin/pytest starter -q
  - .venv/bin/python3 examples/report_measurements.py
test_commands:
  - bash tests/run_tests.sh
cleanup_commands:
  - >-
    find . -path ./.venv -prune -o -type d -name '__pycache__' -print -exec rm -rf -- {}
    +
  - rm -rf .pytest_cache
  - 'rm -rf .venv  # optional: removes the lab virtual environment'
  - 'git checkout -- starter/  # optional: reset your work'
requires_network: true
requires_api_key: false
estimated_minutes: 60
last_executed: '2026-08-27'
executed_on: >-
  macOS 26.5.2 (Apple Silicon, arm64), Python 3.14.0, numpy 2.5.2, scikit-learn 1.9.0,
  pytest 9.1.1, bash 3.2.57 -- bash tests/run_tests.sh -> 14 checks, 0 failure(s), exit
  0. pytest examples -q -> 19 passed. pytest starter -q -> 4 passed, 15 skipped (the
  four machinery checks in test_feedback_lib.py are solved in both directories; the
  fifteen exercise stubs in starter/test_feedback_claims.py are untouched). Everything
  ran through a real lab-local .venv created by the documented setup commands;
  scikit-learn pulled in scipy 1.18.1, joblib 1.5.3 and threadpoolctl 3.6.0 as its own
  dependencies, none of which this lab imports directly. After the pip install the lab
  is fully offline -- the iris measurements come from a copy bundled inside the
  installed scikit-learn package, every other dataset is generated on the spot from a
  seeded numpy.random.default_rng, and harness check 9 confirms no URL appears anywhere
  in starter/ or examples/ source. Section 7 of the harness copies examples/ into a
  mktemp-d scratch directory, confirms 19 passed, rewrites `assert reached_good == 300`
  to `assert reached_good == 299`, confirms a non-zero exit naming the failing test, and
  removes the scratch directory. Separately, by hand, `assert raw == 0.24` was changed
  to 0.99 in examples/test_feedback_claims.py and the whole harness re-run: it reported
  14 checks, 2 failure(s) and exited 1 (both the pytest run and the pytest-free direct
  reproduction in section 2 caught it); the file was restored and the harness returned
  to 14 checks, 0 failure(s), exit 0. MEASURED PAIRS, all captured verbatim in
  expected-output/measured-values.txt. (1) Supervised learning on iris: a 5-NN on a
  100/50 split scores 0.92, and the defining property is that all 100 training rows
  carry an answer. (2) The same k-means partition of iris scores 0.24 read literally and
  0.8933333333333333 after the best relabelling (1, 0, 2) -- a 0.6533 difference
  produced by cluster numbering alone, with no change to the partition. (3) k-means at
  k=3 isolates species 0 exactly (row [0, 50, 0]) and blends species 1 and 2 (rows [48,
  0, 2] and [14, 0, 36]); finding that out required the labels the method does not have.
  (4) Standardising changes the clustering (adjusted Rand index 0.8036117420390129
  between raw and scaled) and here moves AWAY from the species -- 0.6201351808870379
  scaled against 0.7302382722834697 raw -- so the folklore advice loses on this dataset.
  (5) Inertia at k = 2..6 is [152.348, 78.851, 57.228, 46.446, 39.04], falling at every
  step, so minimising it always answers k = n; the best mean silhouette in [0.681,
  0.5528, 0.4981, 0.4887, 0.3648] is at k=2, while iris has three species. (6) Ten-armed
  bandit averaged over 200 problems and 1000 steps: greedy scores mean reward 0.9838 and
  takes the best arm 0.3130 of the time, epsilon=0.01 scores 1.1314 and 0.4336, and
  epsilon=0.1 scores 1.2800 and 0.7080 -- never exploring costs 0.3949 of optimal-action
  rate. (7) On a 5x5 gridworld whose shortest path is 8 steps and whose only reward is
  at the goal, Q-learning with np.argmax tie-breaking reaches the goal in 0 of 300
  episodes while random tie-breaking reaches it in 300 of 300; episode length falls from
  a 46.8 mean over the first ten to 10.4 over the last ten, and the trained greedy
  policy walks exactly 8 steps. The reward travels backwards exactly one state per
  episode at first -- 1, 2, 3, 5 and 10 states valued after 1, 2, 3, 5 and 10 episodes,
  then 18, 20, 21 and 22 after 25, 50, 100 and 300. (8) A log is not a supervised
  dataset: an epsilon=0.1 agent pulled the best arm 1813 of 2000 times, leaving 187
  pulls for the other nine arms and at most 30 for any one of them. Across eight seeds,
  3 of 8 greedy logs name the truly best arm and 4 of 8 contain exactly one arm; with
  epsilon=0.1 all eight logs contain all ten arms and 7 of 8 are right. (9) The one
  explored log that is still wrong is the winner's curse measured directly: at seed 1
  the truly best arm 4 (true mean 0.9054) was pulled 1524 times and reads 0.8634, while
  arm 1 (true mean 0.8216) was pulled 274 times and reads 0.9262, so the argmax picks
  arm 1 -- more data made the better arm's estimate more accurate and therefore lower.
  (10) Label budgets on iris averaged over 40 splits: 3, 5, 10, 20 and 50 random labels
  score 0.6455, 0.778, 0.896, 0.924 and 0.947 against a 0.9535 ceiling with all 100
  labelled, while three labels chosen one per k-means cluster score 0.876 -- 0.2305
  better than three random ones, and worth roughly nine of them. FOUR HONESTY CALLS.
  FIRST: standardising before clustering is near-universal advice and it makes iris
  worse, measured; the lab reports the measurement and explains the cause rather than
  quietly using whichever preprocessing looked better. SECOND: the tie-breaking failure
  in exercise 7 was found by writing the agent the obvious way and watching it never
  learn -- it is kept in the library behind a flag, and measured, rather than silently
  fixed. THIRD: the epsilon=0.1 logs get 7 of 8 right, not 8 of 8; the single failure
  was investigated rather than reseeded away, and turned out to be the winner's curse,
  which is a better lesson than a clean sweep would have been. FOURTH: the single-split
  label-budget curve is not monotone (five labels beat twenty), and the lab asserts that
  it is not, keeping it beside the averaged curve so the reader can see what one split
  is worth.

# Requirements

`requirements.txt` pins the three packages this lab imports directly, at
the exact versions the captured output in `expected-output/` was produced
with:

```
numpy==2.5.2
scikit-learn==1.9.0
pytest==9.1.1
```

Installing scikit-learn also pulls in scipy, joblib and threadpoolctl as
its own dependencies. This lab imports none of them directly and does not
pin them; the versions present during capture are recorded in
`../expected-output/FIELDS.md`.

## Why the versions are pinned exactly

Most of the numbers in this lab come from a seeded
`numpy.random.default_rng`. NumPy's documentation is explicit that
`Generator` makes no promise of stream compatibility between versions, so
a different NumPy can legitimately produce a different stream from the
same seed and every sampled figure would move. Pinning is what makes
"assert this equals 0.708" an honest assertion rather than a trap.

The structural results do not depend on the pins at all —
`expected-output/FIELDS.md` separates the two categories explicitly, and
it is worth reading before you conclude that a mismatch is a bug.

## Installing

From the lab directory:

```bash
python3 -m venv .venv
.venv/bin/pip install -r requirements/requirements.txt
```

The install step needs the network. Everything after it is offline: iris
ships inside the installed scikit-learn package, and every other dataset
in this lab is generated on the spot from a seeded generator.

## Free and open-source status

All three packages are free and open source — NumPy and scikit-learn
under the BSD 3-Clause licence, pytest under the MIT licence. There is no
paid tier, no account and no API key anywhere in this lab.

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

# Day 142 lab brief — Three Kinds of Feedback

Yesterday you learned what a model score does not mean. Today you learn
what kind of question you are even asking, and the answer turns out not
to depend on the algorithm at all. It depends on the **feedback signal**
your learner is given.

There are exactly three shapes of feedback, and every named branch of the
field falls out of them:

| Setting | What you are told | What you are never told |
| --- | --- | --- |
| Supervised | the correct output, for every input | nothing — you have the answer key |
| Unsupervised | nothing | whether any answer you found is right |
| Reinforcement | how good the action you took was | what the best action would have been |

That third row is the one people underestimate. **Evaluative feedback is
not weak supervision; it is a different kind of information.** A
supervised learner is told the answer. A reinforcement learner is told a
score for the one thing it tried, and must work out the counterfactual on
its own — by trying something else, which costs it reward.

## The three claims you are here to measure

1. **Unsupervised learning has no answer key, and pretending otherwise
   produces nonsense.** The same k-means partition of iris scores `0.24`
   or `0.8933` depending only on how you number the clusters. Sixty-five
   accuracy points, from arithmetic that has nothing to do with the data.
2. **A reinforcement learner that never explores never finds out.** A
   purely greedy agent on a ten-armed bandit takes the best arm 31% of
   the time. Spending 10% of its pulls on random exploration raises that
   to 71%. It is the same algorithm, the same problem, one parameter.
3. **A log of what a policy did is not a supervised dataset.** Eight
   greedy agents each produce two thousand rows of clean, honest, correct
   data — and five of the eight logs name the wrong arm as best. Four of
   them contain exactly one arm. There is no model that can fix that,
   because the missing rows were never collected.

## Two things you will find that the textbook does not say

Both were measured while building this lab, and both are in the exercises
because they are more instructive than the tidy version.

- **Standardising the features makes k-means *worse* on iris** (adjusted
  Rand index 0.620 scaled against 0.730 raw). "Always scale before
  clustering" is good default advice and it loses here, because iris's
  four features are already in the same unit and standardising promotes
  the noisiest of them. Exercise 4 measures it.
- **`np.argmax` decides whether your agent learns at all.** It returns
  the *lowest* index attaining the maximum, so on a Q-table that starts
  at all zeros the greedy branch of an ε-greedy agent is a constant
  action. The agent in exercise 7 reaches the goal in **0 of 300**
  episodes with `np.argmax` and **300 of 300** with random tie-breaking.
  Nothing throws. Nothing warns. The agent simply never learns, and the
  only symptom is a flat curve.

## How to work

1. Build the environment (see the lab `README.md`).
2. Run `.venv/bin/pytest starter -q`. You will see four passes (the
   machinery checks in `test_feedback_lib.py`) and fifteen skips.
3. Replace one `pytest.skip(...)` at a time with real code. The skip text
   names the exact helpers and the exact values to assert. None of it is
   guesswork.
4. Print the measured pair in every exercise. A number you did not print
   is a number you did not look at.
5. When you want the whole measured table at once, run
   `.venv/bin/python3 examples/report_measurements.py`.

Do not run `pytest starter examples` in one invocation. Both directories
define `feedback_lib.py`, `test_feedback_lib.py` and
`test_feedback_claims.py`; pytest aborts on the module-name collision.
Run them separately, always.

## The exercises

| # | What it establishes |
| --- | --- |
| 1 | Supervised learning means an answer exists for every training row |
| 2 | Cluster ids are arbitrary; raw accuracy against them is meaningless |
| 3 | What k-means actually found on iris — one species clean, two blended |
| 4 | Structure is not unique: standardising changes it, and here for the worse |
| 5 | Inertia cannot choose k, and silhouette chooses the wrong one |
| 6 | Evaluative feedback: the measured price of never exploring |
| 6b | A bandit reveals only the arm you pulled |
| 7 | Delayed credit — and the tie-breaking bug that silently prevents learning |
| 7b | The reward travels backwards exactly one state per episode |
| 8 | A log records only what the logging policy chose |
| 8b | A greedy log confirms whatever it locked onto |
| 8c | The winner's curse: why even a well-explored log gets one wrong |
| 9 | Labels are the expensive part, and clustering makes them go further |
| 10 | Naming the setting before choosing an algorithm |
| 10b | A function that refuses an incomplete description |

Exercise 8c is worth lingering on. The log that gets it wrong pulled the
genuinely best arm 1524 times and a slightly worse arm 274 times. More
data made the good arm's estimate *more accurate*, so it landed slightly
under its true mean, while the thinly-sampled arm landed over. Taking an
argmax over noisy estimates systematically favours whichever estimate is
most inflated. You will meet that exact effect again on Day 144, wearing
different clothes and called model-selection bias.

"""Three kinds of feedback, built from scratch and measured.

The taxonomy in this lesson is not a taxonomy of algorithms. It is a
taxonomy of the *feedback signal* a learner is given:

* supervised   -- instructive feedback: for every input you are told the
                  correct output, so the error is defined per example;
* unsupervised -- no feedback at all: there is no correct output, only
                  structure, and structure is not unique;
* reinforcement -- evaluative and delayed feedback: you are told how good
                  the action you took was, never what the best action
                  would have been, and often only much later.

Everything here is deterministic given a seed. The bandit and the
gridworld are written from first principles in NumPy so that the shape of
each feedback signal is visible in the code rather than hidden inside a
library.
"""

from __future__ import annotations

import numpy as np
from itertools import permutations

from sklearn.cluster import KMeans
from sklearn.datasets import load_iris
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import adjusted_rand_score, silhouette_score
from sklearn.neighbors import KNeighborsClassifier
from sklearn.preprocessing import StandardScaler

# --------------------------------------------------------------------------
# Shared helpers
# --------------------------------------------------------------------------


def iris_features_and_labels():
    """The iris measurements and the species column, as plain arrays."""
    return load_iris(return_X_y=True)


def accuracy(y_true, y_pred) -> float:
    """Fraction of positions where two label arrays agree."""
    y_true = np.asarray(y_true)
    y_pred = np.asarray(y_pred)
    return float(np.mean(y_true == y_pred))


def standardise(X):
    """Centre each column at zero and scale it to unit variance."""
    return StandardScaler().fit_transform(X)


# --------------------------------------------------------------------------
# 1. Supervised: instructive feedback
# --------------------------------------------------------------------------


def supervised_score(X, y, train_idx, test_idx, n_neighbors: int = 5) -> float:
    """Fit a k-NN on the training rows and score it on the held-out rows.

    This is the whole shape of supervised learning: every training row
    carries its own answer, so the learner can measure and reduce a
    per-example error.
    """
    model = KNeighborsClassifier(n_neighbors=n_neighbors)
    model.fit(X[train_idx], y[train_idx])
    return accuracy(y[test_idx], model.predict(X[test_idx]))


def split_indices(n: int, n_train: int, seed: int):
    """A deterministic random split of range(n) into train and test halves."""
    order = np.random.default_rng(seed).permutation(n)
    return order[:n_train], order[n_train:]


# --------------------------------------------------------------------------
# 2-5. Unsupervised: no feedback at all
# --------------------------------------------------------------------------


def cluster(X, k: int, seed: int = 0):
    """k-means with a fixed seed and a fixed number of restarts."""
    return KMeans(n_clusters=k, n_init=10, random_state=seed).fit(X)


def raw_cluster_accuracy(y_true, cluster_ids) -> float:
    """Compare cluster ids to true labels as if the ids meant something.

    They do not. k-means numbers its clusters by where its own centroids
    happened to land, so this figure is an artefact of the numbering.
    """
    return accuracy(y_true, cluster_ids)


def best_permutation_accuracy(y_true, cluster_ids):
    """The honest version: try every relabelling and keep the best.

    Returns ``(accuracy, mapping)`` where ``mapping[i]`` is the true label
    assigned to cluster ``i``. This is the number people *mean* when they
    say a clustering "recovered the classes", and computing it requires
    the labels -- which unsupervised learning, by definition, does not
    have.
    """
    y_true = np.asarray(y_true)
    cluster_ids = np.asarray(cluster_ids)
    labels = sorted(set(int(v) for v in y_true))
    ids = sorted(set(int(v) for v in cluster_ids))
    best = (-1.0, None)
    for perm in permutations(labels, len(ids)):
        mapped = np.empty_like(cluster_ids)
        for cid, lab in zip(ids, perm):
            mapped[cluster_ids == cid] = lab
        score = accuracy(y_true, mapped)
        if score > best[0]:
            best = (score, tuple(perm))
    return best


def cluster_confusion(y_true, cluster_ids, n_labels: int, n_clusters: int):
    """Rows are true classes, columns are cluster ids; entries are counts."""
    y_true = np.asarray(y_true)
    cluster_ids = np.asarray(cluster_ids)
    table = np.zeros((n_labels, n_clusters), dtype=int)
    for t, c in zip(y_true, cluster_ids):
        table[int(t), int(c)] += 1
    return table


def inertia_curve(X, ks, seed: int = 0):
    """Within-cluster sum of squares for each k in ``ks``."""
    return [float(cluster(X, k, seed=seed).inertia_) for k in ks]


def silhouette_curve(X, ks, seed: int = 0):
    """Mean silhouette score for each k in ``ks`` (k >= 2 only)."""
    out = []
    for k in ks:
        assignment = cluster(X, k, seed=seed).labels_
        out.append(float(silhouette_score(X, assignment)))
    return out


def agreement(labels_a, labels_b) -> float:
    """Adjusted Rand index: agreement between two partitions, ignoring names."""
    return float(adjusted_rand_score(labels_a, labels_b))


# --------------------------------------------------------------------------
# 6. Reinforcement: evaluative feedback, on a bandit built from scratch
# --------------------------------------------------------------------------


class GaussianBandit:
    """A k-armed bandit. Pulling arm i returns N(mean_i, 1).

    The defining property is what you are *not* told: pulling arm i tells
    you the reward for arm i on this pull and nothing whatever about the
    other k-1 arms. That is evaluative feedback. A supervised learner in
    the same position would have been handed the whole reward vector.
    """

    def __init__(self, k: int = 10, seed: int = 0):
        self.k = k
        self.rng = np.random.default_rng(seed)
        self.means = self.rng.normal(0.0, 1.0, size=k)
        self.best_arm = int(np.argmax(self.means))

    def pull(self, arm: int) -> float:
        """Return the reward for one pull of ``arm`` -- and only that arm."""
        return float(self.rng.normal(self.means[arm], 1.0))

    def full_feedback(self) -> np.ndarray:
        """The reward vector a supervised learner would have been given.

        No reinforcement-learning agent ever sees this. It exists here only
        so the lab can measure what the missing information is worth.
        """
        return self.rng.normal(self.means, 1.0)


def run_bandit(k: int = 10, steps: int = 1000, epsilon: float = 0.1, seed: int = 0):
    """Epsilon-greedy action-value learning, written out in full.

    ``epsilon=0.0`` is the pure greedy agent. Returns a dict with the
    reward at each step and whether the optimal arm was chosen at each
    step.
    """
    bandit = GaussianBandit(k=k, seed=seed)
    chooser = np.random.default_rng(seed + 10_000)
    estimates = np.zeros(k)
    counts = np.zeros(k, dtype=int)
    rewards = np.zeros(steps)
    optimal = np.zeros(steps, dtype=bool)

    for t in range(steps):
        if chooser.random() < epsilon:
            arm = int(chooser.integers(k))
        else:
            arm = int(np.argmax(estimates))
        reward = bandit.pull(arm)
        counts[arm] += 1
        # incremental mean: estimate += (reward - estimate) / count
        estimates[arm] += (reward - estimates[arm]) / counts[arm]
        rewards[t] = reward
        optimal[t] = arm == bandit.best_arm

    return {
        "rewards": rewards,
        "optimal": optimal,
        "estimates": estimates,
        "counts": counts,
        "true_means": bandit.means,
        "best_arm": bandit.best_arm,
    }


def average_bandit(runs: int = 200, **kwargs):
    """Average ``run_bandit`` over independent problems, as the field does.

    A single bandit run is almost pure noise; the published comparison
    between greedy and epsilon-greedy is an average over many problems.
    Returns ``(mean_reward, fraction_optimal)`` over the whole horizon.
    """
    reward_total = 0.0
    optimal_total = 0.0
    for r in range(runs):
        out = run_bandit(seed=r, **kwargs)
        reward_total += float(np.mean(out["rewards"]))
        optimal_total += float(np.mean(out["optimal"]))
    return reward_total / runs, optimal_total / runs


# --------------------------------------------------------------------------
# 7. Delayed feedback and credit assignment: a gridworld, from scratch
# --------------------------------------------------------------------------


class GridWorld:
    """A ``size`` x ``size`` grid. Reward +1 at the goal, 0 everywhere else.

    The agent starts top-left, the goal is bottom-right, and every step
    costs nothing. So the feedback for the very first move arrives only
    after the goal is reached -- which is the credit-assignment problem in
    its smallest honest form.
    """

    ACTIONS = ((-1, 0), (1, 0), (0, -1), (0, 1))  # up, down, left, right

    def __init__(self, size: int = 5):
        self.size = size
        self.start = (0, 0)
        self.goal = (size - 1, size - 1)

    def n_states(self) -> int:
        return self.size * self.size

    def state_index(self, pos) -> int:
        return pos[0] * self.size + pos[1]

    def step(self, pos, action: int):
        """Return ``(next_pos, reward, done)``. Walls block movement."""
        dr, dc = self.ACTIONS[action]
        r = min(max(pos[0] + dr, 0), self.size - 1)
        c = min(max(pos[1] + dc, 0), self.size - 1)
        nxt = (r, c)
        if nxt == self.goal:
            return nxt, 1.0, True
        return nxt, 0.0, False

    def shortest_path_length(self) -> int:
        """The optimal number of steps from start to goal on an open grid."""
        return (self.size - 1) * 2


def argmax_random_tiebreak(values, rng) -> int:
    """Return an index of the maximum, choosing uniformly among ties.

    ``np.argmax`` returns the *lowest* index that attains the maximum. On a
    Q-table that starts at all zeros every row is one big tie, so the greedy
    branch of an epsilon-greedy agent degenerates into a constant action.
    This lab measures what that costs; see exercise 7.
    """
    values = np.asarray(values)
    best = np.flatnonzero(values == values.max())
    return int(best[rng.integers(len(best))])


def q_learning(
    world: GridWorld,
    episodes: int = 300,
    alpha: float = 0.5,
    gamma: float = 0.95,
    epsilon: float = 0.2,
    max_steps: int = 200,
    seed: int = 0,
    break_ties_randomly: bool = True,
):
    """Tabular Q-learning, written out so the bootstrap is visible.

    The update ``Q[s,a] += alpha * (r + gamma * max_a' Q[s',a'] - Q[s,a])``
    is how a reward that only ever appears at the goal travels backwards to
    the first move. Nobody ever tells the agent which action was correct; it
    infers it from its own later estimates.

    Set ``break_ties_randomly=False`` to get the ``np.argmax`` behaviour that
    exercise 7 measures as a failure.
    """
    rng = np.random.default_rng(seed)
    n_actions = len(world.ACTIONS)
    Q = np.zeros((world.n_states(), n_actions))
    lengths = []
    reached = 0

    for _ in range(episodes):
        pos = world.start
        done = False
        for step in range(max_steps):
            s = world.state_index(pos)
            if rng.random() < epsilon:
                a = int(rng.integers(n_actions))
            elif break_ties_randomly:
                a = argmax_random_tiebreak(Q[s], rng)
            else:
                a = int(np.argmax(Q[s]))
            nxt, reward, done = world.step(pos, a)
            s_next = world.state_index(nxt)
            target = reward + (0.0 if done else gamma * float(np.max(Q[s_next])))
            Q[s, a] += alpha * (target - Q[s, a])
            pos = nxt
            if done:
                break
        reached += int(done)
        lengths.append(step + 1)

    return Q, lengths, reached


def greedy_path_length(world: GridWorld, Q, max_steps: int = 200) -> int:
    """Follow the learned policy with no exploration and count the steps.

    Returns ``max_steps`` if the policy never reaches the goal, which is the
    honest answer for a Q-table the reward never reached.
    """
    pos = world.start
    for step in range(max_steps):
        a = int(np.argmax(Q[world.state_index(pos)]))
        pos, _reward, done = world.step(pos, a)
        if done:
            return step + 1
    return max_steps


def states_with_nonzero_value(Q) -> int:
    """How many grid squares the reward signal has actually reached."""
    return int(np.sum(np.max(Q, axis=1) > 0.0))


# --------------------------------------------------------------------------
# 8. Why logged bandit data is not a supervised dataset
# --------------------------------------------------------------------------


def logged_policy_dataset(k: int = 10, steps: int = 2000, epsilon: float = 0.1, seed: int = 0):
    """Turn a bandit run into the (arm, reward) table a log would contain."""
    out = run_bandit(k=k, steps=steps, epsilon=epsilon, seed=seed)
    arms = []
    rewards = []
    bandit = GaussianBandit(k=k, seed=seed)
    chooser = np.random.default_rng(seed + 10_000)
    estimates = np.zeros(k)
    counts = np.zeros(k, dtype=int)
    for _ in range(steps):
        if chooser.random() < epsilon:
            arm = int(chooser.integers(k))
        else:
            arm = int(np.argmax(estimates))
        reward = bandit.pull(arm)
        counts[arm] += 1
        estimates[arm] += (reward - estimates[arm]) / counts[arm]
        arms.append(arm)
        rewards.append(reward)
    return np.array(arms), np.array(rewards), out["best_arm"], counts


def arm_pull_counts(counts) -> dict:
    """Pulls per arm, as a plain dict, for reporting."""
    return {i: int(c) for i, c in enumerate(counts)}


# --------------------------------------------------------------------------
# 9. Labels are the expensive part: semi-supervised in its simplest form
# --------------------------------------------------------------------------


def label_budget_curve(X, y, budgets, seed: int = 0, n_neighbors: int = 1):
    """Accuracy as a function of how many labelled rows you can afford.

    Rows are chosen uniformly at random, which is what you get when nobody
    thinks about *which* rows to label.
    """
    train_idx, test_idx = split_indices(len(y), 100, seed=seed)
    rng = np.random.default_rng(seed + 7)
    scores = []
    for b in budgets:
        chosen = rng.permutation(train_idx)[:b]
        model = KNeighborsClassifier(n_neighbors=min(n_neighbors, b))
        model.fit(X[chosen], y[chosen])
        scores.append(accuracy(y[test_idx], model.predict(X[test_idx])))
    return scores


def cluster_then_label(X, y, k: int, seed: int = 0):
    """Spend the label budget on one representative row per cluster.

    Cluster the *unlabelled* data, label the row closest to each centroid,
    and propagate that label to the whole cluster. This is unsupervised
    learning being used to make supervised learning cheaper, which is the
    honest reason the two categories sit in one lesson.
    """
    train_idx, test_idx = split_indices(len(y), 100, seed=seed)
    km = cluster(X[train_idx], k, seed=seed)
    representatives = []
    for c in range(k):
        members = np.where(km.labels_ == c)[0]
        d = np.linalg.norm(X[train_idx][members] - km.cluster_centers_[c], axis=1)
        representatives.append(int(members[int(np.argmin(d))]))
    rep_labels = y[train_idx][representatives]
    propagated = rep_labels[km.labels_]
    model = KNeighborsClassifier(n_neighbors=1)
    model.fit(X[train_idx], propagated)
    return accuracy(y[test_idx], model.predict(X[test_idx])), len(representatives)


def full_supervision_score(X, y, seed: int = 0) -> float:
    """The ceiling: every training row labelled."""
    train_idx, test_idx = split_indices(len(y), 100, seed=seed)
    model = KNeighborsClassifier(n_neighbors=1)
    model.fit(X[train_idx], y[train_idx])
    return accuracy(y[test_idx], model.predict(X[test_idx]))


# --------------------------------------------------------------------------
# 10. The decision function: which kind of problem is this?
# --------------------------------------------------------------------------


def problem(*, has_labels: bool, actions_change_the_data: bool, feedback_is_immediate: bool):
    """Build the three-answer description a problem must supply."""
    return {
        "has_labels": has_labels,
        "actions_change_the_data": actions_change_the_data,
        "feedback_is_immediate": feedback_is_immediate,
    }


def classify_problem(spec: dict) -> str:
    """Name the learning setting a problem actually belongs to.

    Order matters. The question that decides the most is whether your
    actions change what data you see next -- because that single property
    is what makes a problem reinforcement learning no matter how many
    labels you have.
    """
    for key in ("has_labels", "actions_change_the_data", "feedback_is_immediate"):
        if key not in spec:
            raise KeyError(f"problem description is missing {key!r}")
    if spec["actions_change_the_data"]:
        if spec["feedback_is_immediate"]:
            return "reinforcement learning: contextual bandit"
        return "reinforcement learning: sequential, with delayed credit"
    if spec["has_labels"]:
        return "supervised learning"
    return "unsupervised learning"


# --------------------------------------------------------------------------
# Reporting helpers used by both the tests and report_measurements.py
# --------------------------------------------------------------------------


def logged_arm_means(arms, rewards) -> dict:
    """Mean logged reward per arm -- the only thing a log can tell you."""
    arms = np.asarray(arms)
    rewards = np.asarray(rewards)
    return {int(a): float(np.mean(rewards[arms == a])) for a in sorted(set(arms.tolist()))}


def logged_best_arm(arms, rewards) -> int:
    """The arm a supervised model trained on the log would choose."""
    means = logged_arm_means(arms, rewards)
    return max(means, key=means.get)


def log_verdicts(seeds, epsilon: float, k: int = 10, steps: int = 2000):
    """For each seed, whether the log's favourite arm is the truly best arm.

    Returns a list of ``(seed, distinct_arms, logged_pick, true_best, correct)``.
    """
    rows = []
    for seed in seeds:
        arms, rewards, true_best, _counts = logged_policy_dataset(
            k=k, steps=steps, epsilon=epsilon, seed=seed
        )
        pick = logged_best_arm(arms, rewards)
        rows.append(
            (seed, len(set(arms.tolist())), pick, int(true_best), bool(pick == int(true_best)))
        )
    return rows


def average_label_budget_curve(X, y, budgets, repeats: int = 40, base_seed: int = 142):
    """``label_budget_curve`` averaged over independent splits and draws.

    A single split of 150 rows is far too noisy to read a trend from -- one
    run of this curve is not even monotone. Averaging is not decoration; it
    is the difference between a measurement and an anecdote.
    """
    total = np.zeros(len(budgets))
    for s in range(repeats):
        total += np.array(label_budget_curve(X, y, budgets, seed=base_seed + s))
    return [float(v) for v in total / repeats]


def average_cluster_then_label(X, y, k: int, repeats: int = 40, base_seed: int = 142) -> float:
    """``cluster_then_label`` averaged over the same splits, for comparison."""
    scores = [cluster_then_label(X, y, k, seed=base_seed + s)[0] for s in range(repeats)]
    return float(np.mean(scores))


def average_full_supervision(X, y, repeats: int = 40, base_seed: int = 142) -> float:
    """The every-row-labelled ceiling, averaged over the same splits."""
    scores = [full_supervision_score(X, y, seed=base_seed + s) for s in range(repeats)]
    return float(np.mean(scores))


def value_spread_by_episode(world, episode_counts, seed: int = 0):
    """How many states the reward has reached after each episode count.

    This is credit assignment made countable: with a reward only at the
    goal, the first episode leaves exactly one state with a non-zero value,
    the second leaves two, and so on.
    """
    rows = []
    for n in episode_counts:
        Q, _lengths, _reached = q_learning(world, episodes=n, seed=seed)
        rows.append((n, states_with_nonzero_value(Q), greedy_path_length(world, Q)))
    return rows

"""Twelve exercises in the three kinds of feedback a learner can be given.

Read `00_brief.md` first. Each function below is a `pytest.skip` naming
exactly what to build and what to assert; replace the skip with real code.
`feedback_lib.py` is complete -- it is the machinery, not the exercise.

Run this suite on its own:

    .venv/bin/pytest starter -q

Never run `pytest starter examples` in one invocation: both directories
define modules with the same names and pytest aborts on the collision.
"""

import numpy as np  # noqa: F401  (you will need it)
import pytest

import feedback_lib as f  # noqa: F401  (you will need it)


@pytest.fixture(scope="module")
def iris():
    return f.iris_features_and_labels()


def test_01_supervised_learning_has_a_per_example_answer(iris):
    pytest.skip(
        "Split iris with f.split_indices(150, 100, seed=142). Score a "
        "5-NN with f.supervised_score and assert it is exactly 0.92. Then "
        "assert the training half really carries 100 answers and that all "
        "three species appear in it. The point is not the score: it is "
        "that an answer exists for every single training row."
    )


def test_02_cluster_ids_are_arbitrary_and_raw_accuracy_is_meaningless(iris):
    pytest.skip(
        "Cluster iris with f.cluster(X, 3, seed=0). Compare the cluster "
        "ids to the species with f.raw_cluster_accuracy -- assert 0.24. "
        "Then use f.best_permutation_accuracy and assert 0.8933333333333333 "
        "with mapping (1, 0, 2). Assert the gap is 0.6533. Both numbers "
        "describe the identical partition; only the numbering differs."
    )


def test_03_kmeans_isolates_one_species_and_blends_two(iris):
    pytest.skip(
        "Build f.cluster_confusion(y, assignment, 3, 3) for the same "
        "clustering. Assert the three rows are [0, 50, 0], [48, 0, 2] and "
        "[14, 0, 36]. Say in a comment which species k-means separated "
        "perfectly and which two it could not tell apart -- and note that "
        "you needed the labels to find that out."
    )


def test_04_standardising_changes_the_clustering_and_here_makes_it_worse(iris):
    pytest.skip(
        "Cluster raw iris and f.standardise(X) at k=3, seed=0. Assert "
        "f.agreement between the two clusterings is 0.8036117420390129. "
        "Assert agreement with the true species is 0.7302382722834697 raw "
        "and 0.6201351808870379 scaled, then assert scaled < raw. The "
        "textbook advice loses on this dataset; report what you measured."
    )


def test_05_inertia_cannot_choose_k_and_silhouette_chooses_the_wrong_one(iris):
    pytest.skip(
        "For ks = [2, 3, 4, 5, 6] assert f.inertia_curve rounds to "
        "[152.348, 78.851, 57.228, 46.446, 39.04] and is strictly "
        "decreasing -- so minimising it always answers k = n. Assert "
        "f.silhouette_curve rounds to [0.681, 0.5528, 0.4981, 0.4887, "
        "0.3648] and that its argmax is k=2, not the 3 species iris has."
    )


def test_06_greedy_locks_on_and_epsilon_greedy_does_not():
    pytest.skip(
        "Run f.average_bandit(runs=200, k=10, steps=1000, epsilon=e) for "
        "e in (0.0, 0.01, 0.1). Assert mean rewards 0.9838, 1.1314, 1.28 "
        "and optimal-action rates 0.313, 0.4336, 0.708. Assert the gap "
        "between epsilon=0.1 and greedy is 0.3949, and that reward rises "
        "with epsilon across all three."
    )


def test_06b_a_bandit_only_ever_reveals_the_arm_you_pulled():
    pytest.skip(
        "Create f.GaussianBandit(k=10, seed=3). Call pull(0) and assert "
        "you got back one float. Then call full_feedback() and assert it "
        "has shape (10,). Write a comment naming what full_feedback() "
        "represents and why no real agent is ever handed it."
    )


def test_07_argmax_tie_breaking_decides_whether_the_agent_learns_at_all():
    pytest.skip(
        "On f.GridWorld(5) (shortest path 8), run f.q_learning for 300 "
        "episodes at seed 0 twice: break_ties_randomly=False and True. "
        "Assert the first reaches the goal in 0 of 300 episodes and the "
        "second in 300 of 300. For the second assert mean episode length "
        "46.8 over the first ten and 10.4 over the last ten, and that "
        "f.greedy_path_length is exactly 8."
    )


def test_07b_the_reward_travels_backwards_one_state_per_episode():
    pytest.skip(
        "Call f.value_spread_by_episode(world, [1, 2, 3, 5, 10, 25, 50, "
        "100, 300]). Assert the valued-state counts for the first five "
        "entries are [1, 2, 3, 5, 10] -- exactly one new state per "
        "episode. Assert 18, 20, 21 and 22 for 25, 50, 100 and 300. "
        "Assert the greedy path is still unreachable at 10 episodes and "
        "is 8 steps at 25."
    )


def test_08_a_log_records_only_what_the_logging_policy_chose():
    pytest.skip(
        "Call f.logged_policy_dataset(k=10, steps=2000, epsilon=0.1, "
        "seed=0). Assert the best arm is 6 and that it was pulled 1813 of "
        "2000 times, that the other nine arms share 187 pulls, and that "
        "the busiest of those nine has only 30. That imbalance is not a "
        "flaw in the log; it is what a good policy produces."
    )


def test_08b_a_greedy_log_confirms_whatever_it_locked_onto():
    pytest.skip(
        "Call f.log_verdicts(range(8), epsilon=0.0). Assert 5 of the 8 "
        "logs name the wrong arm and 4 contain exactly one arm (seeds 1, "
        "2, 3 and 6, all arm 0). Then call it with epsilon=0.1: assert "
        "all 8 logs contain all 10 arms, that 7 of 8 are now correct, and "
        "that the one failure is (1, 10, 1, 4, False)."
    )


def test_08c_the_winners_curse_is_why_the_explored_log_still_gets_one_wrong():
    pytest.skip(
        "For seed 1 at epsilon=0.1: assert the true best arm is 4 with "
        "true mean 0.9054 and 1524 pulls, and that arm 1 has true mean "
        "0.8216 and 274 pulls. Assert the logged means are 0.8634 and "
        "0.9262. Assert arm 4 is under-estimated and arm 1 over-estimated, "
        "and that f.logged_best_arm therefore returns 1. An argmax over "
        "noisy estimates favours the most inflated one. Remember this."
    )


def test_09_three_chosen_labels_are_worth_about_nine_random_ones(iris):
    pytest.skip(
        "Assert f.average_label_budget_curve(X, y, [3, 5, 10, 20, 50], "
        "repeats=40) rounds to [0.6455, 0.778, 0.896, 0.924, 0.947] and "
        "is strictly increasing, while the single-seed 142 curve is [0.64, "
        "0.92, 0.92, 0.86, 0.96] and is NOT. Assert "
        "f.average_cluster_then_label(X, y, 3, repeats=40) is 0.876, that "
        "it beats three random labels by 0.2305, that it sits between the "
        "5- and 10-label figures, and that it stays under the 0.9535 "
        "ceiling from f.average_full_supervision."
    )


def test_10_the_deciding_question_is_whether_your_actions_change_the_data():
    pytest.skip(
        "Run f.classify_problem on all four combinations built by "
        "f.problem: (labels, inert, immediate), (no labels, inert, "
        "immediate), (labels, actions change data, immediate) and "
        "(labels, actions change data, delayed). Assert the four verdicts "
        "are 'supervised learning', 'unsupervised learning', "
        "'reinforcement learning: contextual bandit' and 'reinforcement "
        "learning: sequential, with delayed credit', and that all four "
        "are distinct. Having labels did not make the third one supervised."
    )


def test_10b_an_incomplete_problem_description_is_refused():
    pytest.skip(
        "Assert f.classify_problem({'has_labels': True}) raises KeyError. "
        "A function that guesses at a missing field is worse than one "
        "that refuses, because the guess is invisible in the output."
    )

"""Machinery checks: the library itself behaves, before any claim is made.

These three tests are solved in both `starter/` and `examples/`. They are
here so that a broken helper reports itself as a broken helper rather than
as a surprising scientific result.
"""

import numpy as np
import pytest

import feedback_lib as f


def test_the_bandit_is_deterministic_given_a_seed():
    a = f.GaussianBandit(k=10, seed=5)
    b = f.GaussianBandit(k=10, seed=5)
    assert np.array_equal(a.means, b.means)
    assert a.best_arm == b.best_arm == int(np.argmax(a.means))
    # Different seeds give different problems, or the averaging is a lie.
    c = f.GaussianBandit(k=10, seed=6)
    assert not np.array_equal(a.means, c.means)


def test_the_gridworld_walls_block_movement_and_the_goal_ends_the_episode():
    world = f.GridWorld(4)
    # Moving up from the top-left corner leaves you where you are.
    assert world.step((0, 0), 0) == ((0, 0), 0.0, False)
    assert world.step((0, 0), 2) == ((0, 0), 0.0, False)
    # Moving into the goal pays 1.0 and terminates.
    nxt, reward, done = world.step((3, 2), 3)
    assert nxt == (3, 3) and reward == 1.0 and done is True
    # Every other transition pays nothing at all.
    assert world.step((1, 1), 1) == ((2, 1), 0.0, False)
    assert world.n_states() == 16
    assert world.shortest_path_length() == 6


def test_argmax_random_tiebreak_actually_spreads_over_ties():
    rng = np.random.default_rng(0)
    picks = {f.argmax_random_tiebreak(np.zeros(4), rng) for _ in range(200)}
    assert picks == {0, 1, 2, 3}
    # With a clear winner it is still an argmax.
    assert f.argmax_random_tiebreak(np.array([0.0, 9.0, 1.0, 2.0]), rng) == 1
    # np.argmax, by contrast, never leaves index 0 on an all-zero row.
    assert int(np.argmax(np.zeros(4))) == 0


def test_best_permutation_accuracy_is_never_worse_than_the_raw_number():
    y = np.array([0, 0, 1, 1, 2, 2])
    ids = np.array([2, 2, 0, 0, 1, 1])
    raw = f.raw_cluster_accuracy(y, ids)
    best, mapping = f.best_permutation_accuracy(y, ids)
    assert raw == 0.0
    assert best == 1.0
    # mapping[i] is the true label that cluster i turned out to hold.
    assert mapping == (1, 2, 0)
    with pytest.raises(AssertionError):
        assert best < raw

#!/usr/bin/env bash
# Day 142 lab harness: "Three Kinds of Feedback"
#
# Prints "N checks, M failure(s)" and exits 0 only when M is zero.
set -u

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

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

# Clear caches at the START so the final cleanliness check measures what
# THIS run left behind, not what a previous `pytest starter -q` left.
find . -path ./.venv -prune -o -type d -name '__pycache__' -exec rm -rf -- {} + 2>/dev/null
rm -rf .pytest_cache

CHECKS=0
FAILURES=0

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

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

if [ ! -x "$PYTHON" ]; then
  echo "No lab .venv found at $PYTHON."
  echo "Run: python3 -m venv .venv && .venv/bin/pip install -r requirements/requirements.txt"
  exit 2
fi

echo "1. Installed versions match requirements/requirements.txt"
VERSION_CHECK=$("$PYTHON" - <<'PYEOF'
import numpy, sklearn, pytest
print("numpy", numpy.__version__)
print("scikit-learn", sklearn.__version__)
print("pytest", pytest.__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 the pin"
  else
    fail "$pkg installed=$installed pinned=$pin_version"
  fi
done < <(sed 's/==/ ==/' requirements/requirements.txt)

echo ""
echo "2. Every published claim, reproduced directly (no pytest involved)"
DIRECT_CHECK=$("$PYTHON" - <<'PYEOF'
import sys
sys.path.insert(0, "examples")

import numpy as np

import feedback_lib as f

errors = []


def expect(label, got, want):
    if got != want:
        errors.append(f"{label}: expected {want}, got {got}")


X, y = f.iris_features_and_labels()

# 1. Supervised: an answer per example
train, test = f.split_indices(len(y), 100, seed=142)
expect("5-NN on iris", f.supervised_score(X, y, train, test, n_neighbors=5), 0.92)
expect("training rows", len(train), 100)

# 2. Cluster ids are arbitrary
assignment = f.cluster(X, 3, seed=0).labels_
raw = f.raw_cluster_accuracy(y, assignment)
best, mapping = f.best_permutation_accuracy(y, assignment)
expect("raw cluster accuracy", raw, 0.24)
expect("best-permutation accuracy", round(best, 10), 0.8933333333)
expect("relabelling", mapping, (1, 0, 2))
expect("gap from numbering alone", round(best - raw, 4), 0.6533)

# 3. What k-means found
table = f.cluster_confusion(y, assignment, 3, 3)
expect("species 0 row", table[0].tolist(), [0, 50, 0])
expect("species 1 row", table[1].tolist(), [48, 0, 2])
expect("species 2 row", table[2].tolist(), [14, 0, 36])

# 4. Structure is not unique
scaled = f.cluster(f.standardise(X), 3, seed=0).labels_
expect("ARI raw vs scaled", round(f.agreement(assignment, scaled), 10), 0.803611742)
raw_truth = f.agreement(y, assignment)
scaled_truth = f.agreement(y, scaled)
expect("ARI raw vs species", round(raw_truth, 10), 0.7302382723)
expect("ARI scaled vs species", round(scaled_truth, 10), 0.6201351809)
if scaled_truth >= raw_truth:
    errors.append("standardising did not move away from the species, contradicting exercise 4")

# 5. k is a choice
ks = [2, 3, 4, 5, 6]
inertia = f.inertia_curve(X, ks)
silhouette = f.silhouette_curve(X, ks)
expect("inertia curve", [round(v, 3) for v in inertia], [152.348, 78.851, 57.228, 46.446, 39.04])
expect(
    "silhouette curve",
    [round(v, 4) for v in silhouette],
    [0.681, 0.5528, 0.4981, 0.4887, 0.3648],
)
if not all(a > b for a, b in zip(inertia, inertia[1:])):
    errors.append("inertia was not monotonically decreasing")
expect("silhouette's chosen k", ks[int(np.argmax(silhouette))], 2)

# 6. Evaluative feedback and the cost of not exploring
greedy = f.average_bandit(runs=200, k=10, steps=1000, epsilon=0.0)
small = f.average_bandit(runs=200, k=10, steps=1000, epsilon=0.01)
explore = f.average_bandit(runs=200, k=10, steps=1000, epsilon=0.1)
expect("greedy mean reward", round(greedy[0], 4), 0.9838)
expect("greedy optimal rate", round(greedy[1], 4), 0.313)
expect("epsilon=0.01 mean reward", round(small[0], 4), 1.1314)
expect("epsilon=0.01 optimal rate", round(small[1], 4), 0.4336)
expect("epsilon=0.1 mean reward", round(explore[0], 4), 1.28)
expect("epsilon=0.1 optimal rate", round(explore[1], 4), 0.708)
expect("cost of never exploring", round(explore[1] - greedy[1], 4), 0.3949)

# 7. Delayed feedback and the tie-breaking bug
world = f.GridWorld(5)
expect("shortest path", world.shortest_path_length(), 8)
_qb, lengths_bad, reached_bad = f.q_learning(world, episodes=300, seed=0, break_ties_randomly=False)
q_good, lengths_good, reached_good = f.q_learning(
    world, episodes=300, seed=0, break_ties_randomly=True
)
expect("np.argmax tie-breaking, goals reached", reached_bad, 0)
expect("random tie-breaking, goals reached", reached_good, 300)
expect("first ten episodes", round(float(np.mean(lengths_good[:10])), 1), 46.8)
expect("last ten episodes", round(float(np.mean(lengths_good[-10:])), 1), 10.4)
expect("greedy path after training", f.greedy_path_length(world, q_good), 8)
spread = f.value_spread_by_episode(world, [1, 2, 3, 5, 10, 25, 50, 100, 300])
expect("states valued per episode", [s for _n, s, _p in spread], [1, 2, 3, 5, 10, 18, 20, 21, 22])
expect("greedy path at 10 episodes", spread[4][2], 200)
expect("greedy path at 25 episodes", spread[5][2], 8)

# 8. A log is not a supervised dataset
arms, _rewards, best_arm, counts = f.logged_policy_dataset(k=10, steps=2000, epsilon=0.1, seed=0)
pulls = f.arm_pull_counts(counts)
expect("logged best arm", best_arm, 6)
expect("pulls of the best arm", pulls[6], 1813)
expect("pulls shared by the other nine", sum(v for a, v in pulls.items() if a != 6), 187)
greedy_logs = f.log_verdicts(range(8), epsilon=0.0)
explored_logs = f.log_verdicts(range(8), epsilon=0.1)
expect("greedy logs that are right", sum(1 for r in greedy_logs if r[4]), 3)
expect("greedy logs holding one arm", sum(1 for r in greedy_logs if r[1] == 1), 4)
expect("explored logs that are right", sum(1 for r in explored_logs if r[4]), 7)
expect("explored logs holding all ten arms", sum(1 for r in explored_logs if r[1] == 10), 8)
arms1, rewards1, _b1, counts1 = f.logged_policy_dataset(k=10, steps=2000, epsilon=0.1, seed=1)
bandit1 = f.GaussianBandit(k=10, seed=1)
means1 = f.logged_arm_means(arms1, rewards1)
expect("winner's curse: pulls of arm 4", int(counts1[4]), 1524)
expect("winner's curse: pulls of arm 1", int(counts1[1]), 274)
expect("winner's curse: true mean of arm 4", round(float(bandit1.means[4]), 4), 0.9054)
expect("winner's curse: true mean of arm 1", round(float(bandit1.means[1]), 4), 0.8216)
expect("winner's curse: logged mean of arm 4", round(means1[4], 4), 0.8634)
expect("winner's curse: logged mean of arm 1", round(means1[1], 4), 0.9262)
expect("winner's curse: the log's pick", f.logged_best_arm(arms1, rewards1), 1)

# 9. Labels are the expensive part
budgets = [3, 5, 10, 20, 50]
curve = f.average_label_budget_curve(X, y, budgets, repeats=40)
expect("averaged budget curve", [round(v, 4) for v in curve], [0.6455, 0.778, 0.896, 0.924, 0.947])
single = f.label_budget_curve(X, y, budgets, seed=142)
expect("single-split budget curve", single, [0.64, 0.92, 0.92, 0.86, 0.96])
if all(a <= b for a, b in zip(single, single[1:])):
    errors.append("the single-split curve was monotone, so averaging is not motivated")
if not all(a < b for a, b in zip(curve, curve[1:])):
    errors.append("the averaged curve was not strictly increasing")
chosen = f.average_cluster_then_label(X, y, 3, repeats=40)
ceiling = f.average_full_supervision(X, y, repeats=40)
expect("three chosen labels", round(chosen, 4), 0.876)
expect("every row labelled", round(ceiling, 4), 0.9535)
expect("what choosing is worth", round(chosen - curve[0], 4), 0.2305)
if not (curve[1] < chosen < curve[2]):
    errors.append("three chosen labels did not land between five and ten random ones")

# 10. Naming the setting
verdicts = [
    f.classify_problem(f.problem(has_labels=True, actions_change_the_data=False, feedback_is_immediate=True)),
    f.classify_problem(f.problem(has_labels=False, actions_change_the_data=False, feedback_is_immediate=True)),
    f.classify_problem(f.problem(has_labels=True, actions_change_the_data=True, feedback_is_immediate=True)),
    f.classify_problem(f.problem(has_labels=True, actions_change_the_data=True, feedback_is_immediate=False)),
]
expect(
    "the four verdicts",
    verdicts,
    [
        "supervised learning",
        "unsupervised learning",
        "reinforcement learning: contextual bandit",
        "reinforcement learning: sequential, with delayed credit",
    ],
)
try:
    f.classify_problem({"has_labels": True})
except KeyError:
    pass
else:
    errors.append("classify_problem accepted an incomplete problem description")

if errors:
    for e in errors:
        print("ERROR:", e)
    sys.exit(1)
print("all direct checks passed")
PYEOF
)
if echo "$DIRECT_CHECK" | grep -q "all direct checks passed"; then
  ok "exercises 1-10 reproduced directly against feedback_lib, no pytest involved"
else
  fail "direct library checks failed"
  echo "$DIRECT_CHECK" | sed 's/^/    /'
fi

echo ""
echo "3. examples/ passes in full"
EXAMPLES_OUT=$("$PYTEST" examples -q 2>&1)
if echo "$EXAMPLES_OUT" | tail -1 | grep -qE "^19 passed"; then
  ok "pytest examples -q -> 19 passed"
else
  fail "pytest examples -q did not report 19 passed"
  echo "$EXAMPLES_OUT" | tail -20 | sed 's/^/    /'
fi

echo ""
echo "4. starter/ is an untouched skeleton"
STARTER_OUT=$("$PYTEST" starter -q 2>&1)
if echo "$STARTER_OUT" | tail -1 | grep -qE "4 passed, 15 skipped"; then
  ok "pytest starter -q -> 4 passed, 15 skipped (the machinery checks pass; the fifteen exercises are stubs)"
else
  fail "pytest starter -q did not report 4 passed, 15 skipped"
  echo "$STARTER_OUT" | tail -20 | sed 's/^/    /'
fi

echo ""
echo "5. pytest examples starter (one invocation) aborts on the module-name collision"
COMBINED_OUT=$("$PYTEST" examples starter 2>&1)
if echo "$COMBINED_OUT" | grep -q "import file mismatch"; then
  ok "combined invocation reports import file mismatch, as documented -- never run starter and examples together"
else
  fail "combined invocation did not fail with import file mismatch as expected"
fi

echo ""
echo "6. The report reproduces the captured table exactly"
REPORT_OUT=$("$PYTHON" examples/report_measurements.py 2>&1)
if [ "$REPORT_OUT" = "$(cat expected-output/measured-values.txt)" ]; then
  ok "report_measurements.py output is byte-identical to expected-output/measured-values.txt"
else
  fail "report_measurements.py drifted from expected-output/measured-values.txt"
  echo "$REPORT_OUT" | diff - expected-output/measured-values.txt | head -20 | sed 's/^/    /'
fi

echo ""
echo "7. Proof the harness can fail"
SCRATCH=$(mktemp -d "${TMPDIR:-/tmp}/d142-scratch.XXXXXX")
cp examples/*.py "$SCRATCH"/
SCRATCH_OUT=$("$PYTEST" "$SCRATCH" -q 2>&1)
if echo "$SCRATCH_OUT" | tail -1 | grep -qE "^19 passed"; then
  ok "scratch copy of examples/ passes before it is broken"
else
  fail "scratch copy did not pass before being broken: $(echo "$SCRATCH_OUT" | tail -3)"
fi
"$PYTHON" - "$SCRATCH/test_feedback_claims.py" <<'PYEOF'
import sys
path = sys.argv[1]
text = open(path).read()
needle = "assert reached_good == 300"
replacement = "assert reached_good == 299"
assert needle in text, "could not find the assertion to break"
open(path, "w").write(text.replace(needle, replacement, 1))
PYEOF
BROKEN_OUT=$("$PYTEST" "$SCRATCH" -q 2>&1)
BROKEN_STATUS=$?
if [ "$BROKEN_STATUS" -ne 0 ] && echo "$BROKEN_OUT" | grep -q "test_07_argmax_tie_breaking_decides_whether_the_agent_learns_at_all"; then
  ok "breaking exercise 7's assertion produces a non-zero exit and names the failing test"
else
  fail "broken copy did not fail as expected (exit=$BROKEN_STATUS)"
fi
rm -rf "$SCRATCH"

echo ""
echo "8. The gridworld and the bandit are genuinely deterministic"
DETERMINISM=$("$PYTHON" - <<'PYEOF'
import sys
sys.path.insert(0, "examples")
import numpy as np
import feedback_lib as f

world = f.GridWorld(5)
a = f.q_learning(world, episodes=40, seed=7)[1]
b = f.q_learning(world, episodes=40, seed=7)[1]
c = f.q_learning(world, episodes=40, seed=8)[1]
bandit_a = f.run_bandit(seed=7)["rewards"]
bandit_b = f.run_bandit(seed=7)["rewards"]
if a != b:
    print("ERROR: two runs at the same seed disagreed")
elif a == c:
    print("ERROR: two runs at different seeds were identical")
elif not np.array_equal(bandit_a, bandit_b):
    print("ERROR: the bandit was not reproducible at a fixed seed")
else:
    print("deterministic")
PYEOF
)
if [ "$DETERMINISM" = "deterministic" ]; then
  ok "same seed reproduces exactly; different seeds do not"
else
  fail "determinism check failed: $DETERMINISM"
fi

echo ""
echo "9. Offline, and nothing left behind"
if ! grep -rInE "https?://" examples/*.py starter/*.py > /dev/null 2>&1; then
  ok "no URLs inside examples/ or starter/ source -- this lab reaches no network"
else
  fail "found a URL inside examples/ or starter/"
fi
LEFTOVER_PYC=$(find . -path ./.venv -prune -o -type d -name '__pycache__' -print 2>/dev/null)
if [ -z "$LEFTOVER_PYC" ]; then
  ok "no __pycache__ left behind"
else
  find . -path ./.venv -prune -o -type d -name '__pycache__' -exec rm -rf -- {} + 2>/dev/null
  ok "no __pycache__ left behind (cleaned during this run)"
fi
if [ ! -d .pytest_cache ]; then
  ok "no .pytest_cache left behind"
else
  rm -rf .pytest_cache
  ok "no .pytest_cache left behind (cleaned during this run)"
fi

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