""" reward.py — numerical companion to the glossary entry 'reward'. Purpose ------- Shows what separates a reward from a label: the reward scores only the action that was taken, it may arrive delayed and corrupted by noise, and a reinforcement learning method still learns from it by treating the negative reward as the loss of that action and choosing actions that maximize the cumulative reward. Self-contained (numpy and matplotlib only), fixed seed. Setup ----- A vehicle drives along a lane. Its state is the lateral offset from the lane center, on a grid of 7 positions; the lane edges are obstacles at the two ends. At each time step the vehicle chooses a steering direction (left, straight, right) and a collision sensor returns the reward -(distance moved toward the nearest obstacle), i.e. a low reward for a direction that moves the vehicle toward an obstacle, plus noise with noise level 0.2. An episode has 30 time steps and starts at a random offset. A method learns a steering rule by keeping, for every state and steering direction, the average of the rewards observed for it, and steering in the direction with the largest average. Blocks ------ [B-partial] The reward reveals only how good the chosen direction was: at each time step the sensor scores 1 of the 3 directions, never the best direction itself (a label would), so over an episode fewer than half of the best directions are ever scored. [B-learn] Treating the negative reward as the loss of the direction taken and averaging over 60 episodes, the learned steering rule reaches a larger return (cumulative reward per episode) than steering at random. [B-delay] If the reward arrives two time steps late and is credited to the direction chosen at the time of arrival, the learned rule is worse than when it is credited to the direction that caused it. Outputs ------- reward_return.csv : per episode, the return of the random rule, of the learned rule, and of the rule learned from the late-credited rewards. reward.png : matplotlib preview of that figure (checking only). """ # /// script # requires-python = ">=3.10" # dependencies = ["numpy==2.3.4", "matplotlib==3.10.7"] # /// import numpy as np import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt from pathlib import Path OUT_DIR = Path(__file__).parent report = [] def check(name, ok): report.append((name, bool(ok))) print(f" [{'ok' if ok else 'FAIL'}] {name}") rng = np.random.default_rng(0) N_STATES, CENTER = 7, 3 # offsets 0..6, lane center 3 MOVES = np.array([-1, 0, 1]) # left, straight, right STEPS, EPISODES, NOISE = 30, 60, 0.2 def sensor_reward(state, move): """-(distance moved toward the nearest obstacle), noisy.""" before = min(state, N_STATES - 1 - state) after = min(state + move, N_STATES - 1 - (state + move)) return float(before - after) - NOISE * abs(move) + NOISE * rng.standard_normal() def step(state, move): return int(np.clip(state + move, 0, N_STATES - 1)) def best_move(state): return int(np.sign(CENTER - state)) def run_episode(rule, delay=0): """One episode; returns the return and the (state, move, reward) triples as credited by the method, with the reward credited `delay` time steps late.""" s = int(rng.integers(N_STATES)); ret = 0.0; log = []; pending = [] for t in range(STEPS): a = rule(s) r = sensor_reward(s, MOVES[a]) ret += r pending.append((t + delay, r)) arrived = [p for p in pending if p[0] <= t] pending = [p for p in pending if p[0] > t] for _, r_arr in arrived: log.append((s, a, r_arr)) # credited to the current choice s = step(s, MOVES[a]) return ret, log # -------------------------------------------------------------- [B-partial] ret, log = run_episode(lambda s: int(rng.integers(3))) scored_best = sum(1 for s, a, _ in log if MOVES[a] == best_move(s)) check(f"[B-partial] one episode: {len(log)} directions scored, " f"{scored_best} of them the best one ({scored_best / len(log):.0%})", len(log) == STEPS and scored_best < len(log) / 2) # ---------------------------------------------------------------- [B-learn] def learn(delay): sums = np.zeros((N_STATES, 3)); counts = np.zeros((N_STATES, 3)) returns = [] for ep in range(EPISODES): avg = np.where(counts > 0, sums / np.maximum(counts, 1), 0.0) def rule(s): if rng.random() < 0.2 or counts[s].sum() == 0: return int(rng.integers(3)) # try a direction return int(np.argmax(avg[s])) # largest average reward ret, log = run_episode(rule, delay) for s, a, r in log: sums[s, a] += r; counts[s, a] += 1 # -r is the loss of a returns.append(ret) return np.array(returns) returns_random = np.array([run_episode(lambda s: int(rng.integers(3)))[0] for _ in range(EPISODES)]) returns_learned = learn(delay=0) late = returns_learned[-20:].mean(); rnd = returns_random[-20:].mean() check(f"[B-learn] return over the last 20 episodes: learned rule " f"{late:.1f} vs random {rnd:.1f}", late > rnd + 1.0) # ---------------------------------------------------------------- [B-delay] returns_delayed = learn(delay=2) dl = returns_delayed[-20:].mean() check(f"[B-delay] reward credited two steps late: return {dl:.1f} " f"(vs {late:.1f} when credited to the direction that caused it)", dl < late - 1.0) # ---------------------------------------------------------------- CSV with open(OUT_DIR / "reward_return.csv", "w") as fh: fh.write("episode,ret_random,ret_learned,ret_delayed\n") for i in range(EPISODES): fh.write(f"{i + 1},{returns_random[i]:.3f},{returns_learned[i]:.3f}," f"{returns_delayed[i]:.3f}\n") # -------------------------------------------------------------- preview fig, ax = plt.subplots(figsize=(5.4, 3.8)) ep = np.arange(1, EPISODES + 1) ax.plot(ep, returns_random, "k:", lw=1.2, label="steering at random") ax.plot(ep, returns_learned, "k-", lw=1.4, label="rule learned from the rewards") ax.plot(ep, returns_delayed, "k--", lw=1.2, label="rule learned from rewards credited two steps late") ax.set_xlabel("episode") ax.set_ylabel("return (cumulative reward)") ax.set_title("learning a steering rule from the collision-sensor reward") ax.legend(frameon=False, fontsize=8) fig.tight_layout() fig.savefig(OUT_DIR / "reward.png", dpi=110) n_ok = sum(ok for _, ok in report) print(f"\n{n_ok}/{len(report)} checks pass") print(f"wrote {OUT_DIR / 'reward_return.csv'}, {OUT_DIR / 'reward.png'}") if n_ok != len(report): raise SystemExit(1)