B-delay writes when the script runsDictionary of Applied Machine Learning · reward
Numerical companion to the entry reward: it recomputes what the entry states and prints one line per check
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.
Run it without installing anything:uv run https://dictionaryofml.org/terms/reward.py
uv downloads this script and the pinned NumPy and Matplotlib it needs, then runs it; the script fetches any input file it uses. To keep the output files, download reward.py into a folder and run uv run reward.py there. With NumPy and Matplotlib already installed, python3 reward.py, from any directory — it writes its output files into the current directory. Fixed seeds, so the printed numbers reproduce exactly. Download reward.py · Notebook · Open in Colab
One cell per block of the script: the code, and what that code printed when it last ran here
"""
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).
"""
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
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.
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)
[ok] [B-partial] one episode: 30 directions scored, 11 of them the best one (37%)
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.
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)
[ok] [B-learn] return over the last 20 episodes: learned rule 17.6 vs random -1.6
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.
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)
[ok] [B-delay] reward credited two steps late: return 15.2 (vs 17.6 when credited to the direction that caused it) 3/3 checks pass wrote /Users/junga1/dictionaryappliedml/pythondemos/reward_return.csv, /Users/junga1/dictionaryappliedml/pythondemos/reward.png
B-delay writes when the script runs