"""Exogenous fundamental processes and the present-value benchmarks. Per asset k, the dividend paid every `div_interval` ticks is D_t = mu_{s_t} + x_t + J_t, x_t = rho x_{t-1} + sigma eps_t, with a Markov regime s_t in {0,1} shifting the mean dividend mu, and a jump J_t (compound Poisson, only at dividend dates). Innovations eps are correlated across assets through a one-factor structure eps_k = b_k f + sqrt(1 - b_k^2) u_k. Benchmarks (cents per lot, tick-time discount rate r applied per dividend period): - `full_value` : E[PV | true state (mu_{s_t}, x_t)] -> the "latent fundamental value" - `public_value`: E[PV | public information only] -> Kalman/HMM posterior on (s_t, x_t) Both use the same closed form V = mu/r_d + x * rho_d/(1 + r_d - rho_d) where r_d, rho_d are the per-dividend-period rate and persistence; agents never see `full_value`. """ from __future__ import annotations from dataclasses import dataclass, field import numpy as np @dataclass(slots=True) class FundamentalConfig: n_assets: int = 1 div_interval: int = 20 # ticks between dividends r_period: float = 0.0025 # discount rate per dividend period (0.25%) -> P/D ~ 400 mu: tuple[float, ...] = (50.0,) # mean dividend per period in cents, regime 0 (per asset) mu_shift: float = 0.0 # regime 1 mean = mu * (1 + mu_shift); 0 => no regimes p_switch: float = 0.0 # per-period probability of a regime switch rho: float = 0.98 # AR(1) persistence of x per period sigma: float = 2.0 # innovation std of x (cents) jump_prob: float = 0.0 # per-period jump probability jump_sigma: float = 0.0 # jump std (cents of dividend) factor_loading: tuple[float, ...] = (0.0,) # b_k per asset (common factor in innovations) x0: float = 0.0 @dataclass class FundamentalState: regime: np.ndarray # (K,) int x: np.ndarray # (K,) float, latent deviation period: int = 0 # public-information posterior (per asset): mean and variance of x, regime probability x_hat: np.ndarray = field(default_factory=lambda: np.zeros(1)) x_var: np.ndarray = field(default_factory=lambda: np.zeros(1)) p_regime1: np.ndarray = field(default_factory=lambda: np.zeros(1)) x_hat_revealed: np.ndarray = field(default_factory=lambda: np.zeros(1)) x_var_revealed: np.ndarray = field(default_factory=lambda: np.zeros(1)) class Fundamentals: """Draws the dividend path with a dedicated RNG stream so shocks are common random numbers.""" def __init__(self, cfg: FundamentalConfig, rng: np.random.Generator) -> None: self.cfg = cfg self.rng = rng K = cfg.n_assets mu = np.asarray(cfg.mu, dtype=float) self.mu = np.resize(mu, K) b = np.asarray(cfg.factor_loading, dtype=float) self.b = np.resize(b, K) self.state = FundamentalState(regime=np.zeros(K, dtype=int), x=np.full(K, cfg.x0, dtype=float), x_hat=np.full(K, cfg.x0, dtype=float), x_var=np.zeros(K), p_regime1=np.zeros(K), x_hat_revealed=np.full(K, cfg.x0, dtype=float), x_var_revealed=np.zeros(K)) self.history: list[dict] = [] self.x_revealed = np.full(K, cfg.x0, dtype=float) # last announced deviation self.last_dividend = self.regime_mean(self.state.regime) + self.x_revealed self._draw_next() # x_1 is latent from t = 0 def _draw_next(self) -> None: """Draw the upcoming (latent) deviation from the last revealed one and reset the public prior.""" cfg, st, K = self.cfg, self.state, self.cfg.n_assets f = self.rng.standard_normal() u = self.rng.standard_normal(K) eps = self.b * f + np.sqrt(np.clip(1.0 - self.b ** 2, 0.0, 1.0)) * u st.x = cfg.rho * self.x_revealed + cfg.sigma * eps st.x_hat = cfg.rho * st.x_hat_revealed if hasattr(st, "x_hat_revealed") else cfg.rho * self.x_revealed st.x_var = cfg.rho ** 2 * getattr(st, "x_var_revealed", np.zeros(K)) + cfg.sigma ** 2 # ---- closed-form PV ------------------------------------------------------------------ def _pv(self, mu: np.ndarray, x: np.ndarray) -> np.ndarray: r, rho = self.cfg.r_period, self.cfg.rho return mu / r + x / (1.0 + r - rho) def regime_mean(self, regime: np.ndarray) -> np.ndarray: return self.mu * (1.0 + self.cfg.mu_shift * regime) def full_value(self) -> np.ndarray: """Latent fundamental value in cents per lot (never shown to agents).""" return self._pv(self.regime_mean(self.state.regime), self.state.x) def public_value(self) -> np.ndarray: """Expected PV under the public information set (filtered state).""" p1 = self.state.p_regime1 mu = self.mu * (1.0 + self.cfg.mu_shift * p1) return self._pv(mu, self.state.x_hat) def dividend_now(self) -> np.ndarray: return self.regime_mean(self.state.regime) + self.state.x # ---- dynamics -------------------------------------------------------------------------- def step(self) -> np.ndarray: """Pay the dividend of the current period (revealing x_n), then draw x_{n+1}. Returns D_n (cents).""" cfg, st, K = self.cfg, self.state, self.cfg.n_assets if cfg.p_switch > 0: flips = self.rng.random(K) < cfg.p_switch st.regime = np.where(flips, 1 - st.regime, st.regime) jump = np.zeros(K) if cfg.jump_prob > 0: hit = self.rng.random(K) < cfg.jump_prob jump = np.where(hit, self.rng.standard_normal(K) * cfg.jump_sigma, 0.0) st.period += 1 d = self.regime_mean(st.regime) + st.x + jump self.last_dividend = d self.x_revealed = st.x.copy() self._update_public(d) # sets x_hat_revealed / x_var_revealed (exact when jumps and regimes are off) self._draw_next() self.history.append({"period": st.period, "dividend": d.copy(), "x": st.x.copy(), "x_revealed": self.x_revealed.copy(), "regime": st.regime.copy(), "full_value": self.full_value(), "public_value": self.public_value()}) return d def _update_public(self, d: np.ndarray) -> None: """Public inference from the announced dividend: with one regime and no jumps the deviation is read exactly (x_revealed = d - mu); with regimes it is a two-state mixture filter; with jumps a Kalman step.""" cfg, st = self.cfg, self.state obs_var = cfg.jump_prob * cfg.jump_sigma ** 2 + 1e-9 x_pred, v_pred = st.x_hat, st.x_var # prior on x_n carried through the period (news-updated) if cfg.mu_shift == 0.0 or cfg.p_switch == 0.0: K_gain = v_pred / (v_pred + obs_var) st.x_hat_revealed = x_pred + K_gain * (d - self.mu - x_pred) st.x_var_revealed = (1 - K_gain) * v_pred st.p_regime1 = np.zeros_like(st.p_regime1) return p1 = st.p_regime1 * (1 - cfg.p_switch) + (1 - st.p_regime1) * cfg.p_switch mu0, mu1 = self.mu, self.mu * (1 + cfg.mu_shift) s2 = v_pred + obs_var l0 = np.exp(-0.5 * (d - mu0 - x_pred) ** 2 / s2) l1 = np.exp(-0.5 * (d - mu1 - x_pred) ** 2 / s2) post1 = p1 * l1 / (p1 * l1 + (1 - p1) * l0 + 1e-300) K_gain = v_pred / s2 innov = d - (post1 * mu1 + (1 - post1) * mu0) - x_pred st.x_hat_revealed = x_pred + K_gain * innov st.x_var_revealed = (1 - K_gain) * v_pred st.p_regime1 = post1 def public_observe(f: "Fundamentals", asset: int, value: float, noise_sd: float) -> None: """Kalman update of the public posterior on x_t from a public news item reporting x_t with stated noise. Misleading items carry an undisclosed bias, so the public filter is (deliberately) fooled by them.""" st = f.state v = st.x_var[asset] R = max(noise_sd, 0.1) ** 2 K = v / (v + R) st.x_hat[asset] = st.x_hat[asset] + K * (value - st.x_hat[asset]) st.x_var[asset] = (1.0 - K) * v def private_signal(f: Fundamentals, rng: np.random.Generator, precision: float, asset: int) -> float: """A noisy private signal about the latent deviation of the *upcoming* dividend: s = x_n + N(0, sigma^2/precision). precision -> inf : perfect knowledge of x_n ; precision -> 0 : pure noise. x_n is already drawn (it is the state of the economy), so this is information about the latent state, not about future random draws.""" noise_sd = f.cfg.sigma / np.sqrt(max(precision, 1e-9)) return float(f.state.x[asset] + rng.standard_normal() * noise_sd)