{
 "nbformat": 4,
 "nbformat_minor": 5,
 "metadata": {
  "kernelspec": {
   "name": "python3",
   "display_name": "Python 3",
   "language": "python"
  },
  "language_info": {
   "name": "python"
  },
  "colab": {
   "name": "reward.ipynb"
  }
 },
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": "# reward \u2014 Python demo\n\nNumerical companion to the entry [reward](https://dictionaryofml.org/terms/reward.html) of the [Dictionary of Applied Machine Learning](https://dictionaryofml.org/): it recomputes what the entry states and prints one line per check.\n\nShows 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.\n\nRequires NumPy and Matplotlib only, and uses fixed seeds, so the printed numbers reproduce exactly. Generated from [`pythondemos/reward.py`](https://dictionaryofml.org/terms/reward.py); CC BY 4.0."
  },
  {
   "cell_type": "code",
   "metadata": {},
   "execution_count": null,
   "outputs": [],
   "source": "# Notebook shim: the script resolves output paths relative to __file__,\n# which a notebook kernel does not define; everything lands in the\n# working directory instead.\nimport os\n__file__ = os.path.join(os.getcwd(), \"reward.py\")\nos.makedirs(\"pythondemos\", exist_ok=True)"
  },
  {
   "cell_type": "code",
   "metadata": {},
   "execution_count": null,
   "outputs": [],
   "source": "\"\"\"\nreward.py \u2014 numerical companion to the glossary entry 'reward'.\n\nPurpose\n-------\nShows what separates a reward from a label: the reward scores only the\naction that was taken, it may arrive delayed and corrupted by noise, and\na reinforcement learning method still learns from it by treating the\nnegative reward as the loss of that action and choosing actions that\nmaximize the cumulative reward.  Self-contained (numpy and matplotlib\nonly), fixed seed.\n\nSetup\n-----\nA vehicle drives along a lane.  Its state is the lateral offset from the\nlane center, on a grid of 7 positions; the lane edges are obstacles at\nthe two ends.  At each time step the vehicle chooses a steering\ndirection (left, straight, right) and a collision sensor returns the\nreward -(distance moved toward the nearest obstacle), i.e. a low reward\nfor a direction that moves the vehicle toward an obstacle, plus noise\nwith noise level 0.2.  An episode has 30 time steps and starts\nat a random offset.  A method learns a steering rule by keeping, for\nevery state and steering direction, the average of the rewards observed\nfor it, and steering in the direction with the largest average.\n\nBlocks\n------\n[B-partial] The reward reveals only how good the chosen direction was:\n            at each time step the sensor scores 1 of the 3 directions,\n            never the best direction itself (a label would), so over an\n            episode fewer than half of the best directions are ever\n            scored.\n[B-learn]   Treating the negative reward as the loss of the direction\n            taken and averaging over 60 episodes, the learned steering\n            rule reaches a larger return (cumulative reward per episode)\n            than steering at random.\n[B-delay]   If the reward arrives two time steps late and is credited to\n            the direction chosen at the time of arrival, the learned\n            rule is worse than when it is credited to the direction that\n            caused it.\n\nOutputs\n-------\nreward_return.csv : per episode, the return of the random rule, of the\n                    learned rule, and of the rule learned from the\n                    late-credited rewards.\nreward.png        : matplotlib preview of that figure (checking only).\n\"\"\"\n\nimport numpy as np\nimport matplotlib\n\nmatplotlib.use(\"Agg\")\nimport matplotlib.pyplot as plt\n\nfrom pathlib import Path\n\nOUT_DIR = Path(__file__).parent\n\nreport = []\n\n\ndef check(name, ok):\n    report.append((name, bool(ok)))\n    print(f\"  [{'ok' if ok else 'FAIL'}] {name}\")\n\n\nrng = np.random.default_rng(0)\n\nN_STATES, CENTER = 7, 3                    # offsets 0..6, lane center 3\nMOVES = np.array([-1, 0, 1])               # left, straight, right\nSTEPS, EPISODES, NOISE = 30, 60, 0.2\n\n\ndef sensor_reward(state, move):\n    \"\"\"-(distance moved toward the nearest obstacle), noisy.\"\"\"\n    before = min(state, N_STATES - 1 - state)\n    after = min(state + move, N_STATES - 1 - (state + move))\n    return float(before - after) - NOISE * abs(move) + NOISE * rng.standard_normal()\n\n\ndef step(state, move):\n    return int(np.clip(state + move, 0, N_STATES - 1))\n\n\ndef best_move(state):\n    return int(np.sign(CENTER - state))\n\n\ndef run_episode(rule, delay=0):\n    \"\"\"One episode; returns the return and the (state, move, reward)\n    triples as credited by the method, with the reward credited `delay`\n    time steps late.\"\"\"\n    s = int(rng.integers(N_STATES)); ret = 0.0; log = []; pending = []\n    for t in range(STEPS):\n        a = rule(s)\n        r = sensor_reward(s, MOVES[a])\n        ret += r\n        pending.append((t + delay, r))\n        arrived = [p for p in pending if p[0] <= t]\n        pending = [p for p in pending if p[0] > t]\n        for _, r_arr in arrived:\n            log.append((s, a, r_arr))      # credited to the current choice\n        s = step(s, MOVES[a])\n    return ret, log"
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": "**[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."
  },
  {
   "cell_type": "code",
   "metadata": {},
   "execution_count": null,
   "outputs": [],
   "source": "ret, log = run_episode(lambda s: int(rng.integers(3)))\nscored_best = sum(1 for s, a, _ in log if MOVES[a] == best_move(s))\ncheck(f\"[B-partial] one episode: {len(log)} directions scored, \"\n      f\"{scored_best} of them the best one ({scored_best / len(log):.0%})\",\n      len(log) == STEPS and scored_best < len(log) / 2)"
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": "**[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."
  },
  {
   "cell_type": "code",
   "metadata": {},
   "execution_count": null,
   "outputs": [],
   "source": "def learn(delay):\n    sums = np.zeros((N_STATES, 3)); counts = np.zeros((N_STATES, 3))\n    returns = []\n    for ep in range(EPISODES):\n        avg = np.where(counts > 0, sums / np.maximum(counts, 1), 0.0)\n\n        def rule(s):\n            if rng.random() < 0.2 or counts[s].sum() == 0:\n                return int(rng.integers(3))        # try a direction\n            return int(np.argmax(avg[s]))          # largest average reward\n        ret, log = run_episode(rule, delay)\n        for s, a, r in log:\n            sums[s, a] += r; counts[s, a] += 1     # -r is the loss of a\n        returns.append(ret)\n    return np.array(returns)\n\n\nreturns_random = np.array([run_episode(lambda s: int(rng.integers(3)))[0]\n                           for _ in range(EPISODES)])\nreturns_learned = learn(delay=0)\nlate = returns_learned[-20:].mean(); rnd = returns_random[-20:].mean()\ncheck(f\"[B-learn]   return over the last 20 episodes: learned rule \"\n      f\"{late:.1f} vs random {rnd:.1f}\", late > rnd + 1.0)"
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": "**[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."
  },
  {
   "cell_type": "code",
   "metadata": {},
   "execution_count": null,
   "outputs": [],
   "source": "returns_delayed = learn(delay=2)\ndl = returns_delayed[-20:].mean()\ncheck(f\"[B-delay]   reward credited two steps late: return {dl:.1f} \"\n      f\"(vs {late:.1f} when credited to the direction that caused it)\",\n      dl < late - 1.0)\n\n# ---------------------------------------------------------------- CSV\nwith open(OUT_DIR / \"reward_return.csv\", \"w\") as fh:\n    fh.write(\"episode,ret_random,ret_learned,ret_delayed\\n\")\n    for i in range(EPISODES):\n        fh.write(f\"{i + 1},{returns_random[i]:.3f},{returns_learned[i]:.3f},\"\n                 f\"{returns_delayed[i]:.3f}\\n\")\n\n# -------------------------------------------------------------- preview\nfig, ax = plt.subplots(figsize=(5.4, 3.8))\nep = np.arange(1, EPISODES + 1)\nax.plot(ep, returns_random, \"k:\", lw=1.2, label=\"steering at random\")\nax.plot(ep, returns_learned, \"k-\", lw=1.4, label=\"rule learned from the rewards\")\nax.plot(ep, returns_delayed, \"k--\", lw=1.2,\n        label=\"rule learned from rewards credited two steps late\")\nax.set_xlabel(\"episode\")\nax.set_ylabel(\"return (cumulative reward)\")\nax.set_title(\"learning a steering rule from the collision-sensor reward\")\nax.legend(frameon=False, fontsize=8)\nfig.tight_layout()\nfig.savefig(OUT_DIR / \"reward.png\", dpi=110)\n\nn_ok = sum(ok for _, ok in report)\nprint(f\"\\n{n_ok}/{len(report)} checks pass\")\nprint(f\"wrote {OUT_DIR / 'reward_return.csv'}, {OUT_DIR / 'reward.png'}\")\nif n_ok != len(report):\n    raise SystemExit(1)"
  }
 ]
}