Add Chapter 11 (Policy Gradient REINFORCE) with Python 3.9-compatible typing and CI

Author: srikanthbaride

.github/workflows/ch11.yml

@@ -1,4 +1,4 @@
-name: ch11
+name: ch11 — Policy Gradient (REINFORCE)
 on:
   push:
     paths: ['ch11_policy_gradient/**', '.github/workflows/ch11.yml']
@@ -8,7 +8,8 @@ jobs:
   test:
     runs-on: ubuntu-latest
     strategy:
-      matrix: { python-version: ['3.9','3.10','3.11'] }
+      matrix:
+        python-version: ['3.8','3.9','3.10','3.11']
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5
@@ -18,5 +19,6 @@ jobs:
       - run: |
           python -m pip install -U pip
           pip install -r ch11_policy_gradient/requirements.txt
-      - env: { PYTHONPATH: . }
+      - env:
+          PYTHONPATH: .
         run: pytest -q ch11_policy_gradient/tests

ch11_policy_gradient/README_ch11.md

@@ -1,4 +1,5 @@
 # Chapter 11 — Policy Gradient Fundamentals (REINFORCE)
+
 Quickstart:
 ```bash
 pip install -r ch11_policy_gradient/requirements.txt

ch11_policy_gradient/agents/reinforce.py

@@ -15,30 +15,39 @@ class Reinforce:
     alpha: float = 0.05
     normalize_adv: bool = True
     baseline_fn: Optional[Callable[[object], float]] = None
-    seed: int | None = None
+    seed: Optional[int] = None
+
     def __post_init__(self):
         self.rng = np.random.default_rng(self.seed)
+
     def run_episode_discrete(self, env, policy, feature_fn: Callable[[object], np.ndarray]):
-        s = env.reset(); S,A,R,L = [],[],[],[]; done=False
+        s = env.reset()
+        S, A, R, L = [], [], [], []
+        done = False
         while not done:
-            x = feature_fn(s); a = policy.sample(x)
-            logp,_ = policy.logprob_and_grad(x,a)
+            x = feature_fn(s)
+            a = policy.sample(x)
+            logp, _ = policy.logprob_and_grad(x, a)
             ns, r, done, _ = env.step(a)
-            S.append(s); A.append(a); R.append(r); L.append(logp); s = ns
-        return Trajectory(S,A,R,L)
+            S.append(s); A.append(a); R.append(r); L.append(logp)
+            s = ns
+        return Trajectory(S, A, R, L)
+
     def update_discrete(self, traj: Trajectory, policy, feature_fn: Callable[[object], np.ndarray]):
         G = returns_to_go(traj.rewards, self.gamma)
         if self.baseline_fn is not None:
-            b = np.array([self.baseline_fn(s) for s in traj.states], dtype=float); adv = G - b
+            b = np.array([self.baseline_fn(s) for s in traj.states], dtype=float)
+            adv = G - b
         else:
             adv = G.copy()
         if self.normalize_adv:
-            # Only standardize when there is variability; for 1-step episodes std==0 leads to zero updates.
             if len(adv) >= 2 and np.std(adv) > 1e-8:
                 adv = standardize(adv)
+
         total_grad = np.zeros_like(policy.theta)
-        for s,a,adv_t in zip(traj.states, traj.actions, adv):
-            x = feature_fn(s); _, grad = policy.logprob_and_grad(x,a)
+        for s, a, adv_t in zip(traj.states, traj.actions, adv):
+            x = feature_fn(s)
+            _, grad = policy.logprob_and_grad(x, a)
             total_grad += adv_t * grad
         policy.theta += self.alpha * total_grad
         return {"G": G, "adv": adv}

ch11_policy_gradient/envs/bandit.py

@@ -1,16 +1,22 @@
 import numpy as np
 from dataclasses import dataclass
+from typing import Optional, Tuple
 
 @dataclass
 class TwoArmedBandit:
-    q_star: tuple[float, float] = (1.0, 1.5)
-    seed: int | None = None
+    q_star: Tuple[float, float] = (1.0, 1.5)
+    seed: Optional[int] = None
+
     def __post_init__(self):
         self.rng = np.random.default_rng(self.seed)
+
     @property
-    def nA(self): return 2
+    def nA(self):
+        return 2
+
     def reset(self):
-        return np.array([1.0], dtype=float)  # x(s)=1
+        return np.array([1.0], dtype=float)
+
     def step(self, a: int):
         assert a in (0,1)
         r = float(self.rng.normal(self.q_star[a], 1.0))

ch11_policy_gradient/examples/bandit_softmax.py

@@ -8,12 +8,15 @@ def run(episodes=200, seed=0):
     x = np.array([1.0], dtype=float)
     policy = SoftmaxPolicy(nA=2, d=1, seed=seed)
     algo = Reinforce(gamma=1.0, alpha=0.05, normalize_adv=True, baseline_fn=None, seed=seed)
+
     probs_hist = []
+
     class EPEnv:
         def reset(self): return x
         def step(self, a):
             _, r, done, _ = env.step(a)
             return None, r, True, {}
+
     for _ in range(episodes):
         traj = algo.run_episode_discrete(EPEnv(), policy, lambda s: s)
         algo.update_discrete(traj, policy, lambda s: s)

ch11_policy_gradient/policies/gaussian.py

@@ -1,18 +1,23 @@
 import numpy as np
 from dataclasses import dataclass
+from typing import Optional
 
 @dataclass
 class GaussianPolicy1D:
     mu: float = 0.0
     log_sigma: float = 0.0
-    seed: int | None = None
+    seed: Optional[int] = None
+
     def __post_init__(self):
         self.rng = np.random.default_rng(self.seed)
+
     @property
     def sigma(self) -> float:
         return float(np.exp(self.log_sigma))
+
     def sample(self, _x=None) -> float:
         return float(self.rng.normal(self.mu, self.sigma))
+
     def logprob_and_grad(self, a: float, _x=None):
         sigma2 = self.sigma ** 2
         logp = -0.5 * ((a - self.mu) ** 2 / sigma2 + np.log(2*np.pi*sigma2))

ch11_policy_gradient/policies/softmax.py

@@ -1,24 +1,35 @@
 import numpy as np
 from dataclasses import dataclass
+from typing import Optional
 
 @dataclass
 class SoftmaxPolicy:
     nA: int
     d: int
-    theta: np.ndarray | None = None
-    seed: int | None = None
+    theta: Optional[np.ndarray] = None
+    seed: Optional[int] = None
+
     def __post_init__(self):
         if self.theta is None:
             self.theta = np.zeros((self.nA, self.d), dtype=float)
         self.rng = np.random.default_rng(self.seed)
+
     def prefs(self, x: np.ndarray) -> np.ndarray:
         return self.theta @ x
+
     def probs(self, x: np.ndarray) -> np.ndarray:
-        h = self.prefs(x); h -= np.max(h)
-        e = np.exp(h); return e / e.sum()
+        h = self.prefs(x)
+        h -= np.max(h)
+        e = np.exp(h)
+        return e / e.sum()
+
     def sample(self, x: np.ndarray) -> int:
-        p = self.probs(x); return int(self.rng.choice(self.nA, p=p))
+        p = self.probs(x)
+        return int(self.rng.choice(self.nA, p=p))
+
     def logprob_and_grad(self, x: np.ndarray, a: int):
-        p = self.probs(x); logp = float(np.log(p[a] + 1e-12))
-        grad = -np.outer(p, x); grad[a, :] += x
+        p = self.probs(x)
+        logp = float(np.log(p[a] + 1e-12))
+        grad = -np.outer(p, x)
+        grad[a, :] += x
         return logp, grad

ch11_policy_gradient/tests/conftest.py

@@ -1,4 +1,4 @@
-# Ensure repo root on sys.path so `import ch11_policy_gradient` works from any cwd
+# Ensure repo root is on sys.path so `import ch11_policy_gradient` works
 import os, sys
 ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))
 if ROOT not in sys.path:

ch11_policy_gradient/utils/returns.py

@@ -1,9 +1,14 @@
 import numpy as np
+
 def returns_to_go(rewards, gamma: float) -> np.ndarray:
-    G = np.zeros(len(rewards), dtype=float); g = 0.0
+    G = np.zeros(len(rewards), dtype=float)
+    g = 0.0
     for t in reversed(range(len(rewards))):
-        g = rewards[t] + gamma * g; G[t] = g
+        g = rewards[t] + gamma * g
+        G[t] = g
     return G
+
 def standardize(x: np.ndarray, eps: float = 1e-8) -> np.ndarray:
-    mu, std = x.mean(), x.std()
+    mu = x.mean()
+    std = x.std()
     return (x - mu) / (std + eps)