1"""News arrivals and exogenous treatment shocks.23* Public news: at random ticks a public item reveals x_t with noise (credible, low noise) or with a4 deliberate bias (misleading). Every agent can read public news; whether they act on it is policy.5* Private signals: drawn per agent at activation time by `private_signal` (see assets.fundamentals)6 with an agent-specific precision; informed agents receive them at heterogeneous arrival times.7* Shocks: scheduled treatment events — a jump in the fundamental mean (macro news), a change in the8 discount rate (monetary policy), a funding shock (leverage cap cut) or a liquidity shock (market9 makers withdraw). Shocks are logged so every treatment/control pair shares the same timing.10"""11from __future__ import annotations1213from dataclasses import dataclass, field1415import numpy as np161718@dataclass(slots=True)19class NewsItem:20 time: int21 asset: int22 value: float # reported x (cents) possibly biased23 credible: bool24 noise_sd: float25 text: str = ""262728@dataclass(slots=True)29class NewsConfig:30 rate: float = 0.02 # per-tick probability of a public news item (per asset)31 noise_sd: float = 1.0 # noise of credible news (cents)32 misleading_share: float = 0.0 # share of news items that are biased33 misleading_bias: float = 10.0 # bias (cents) applied with random sign343536@dataclass(slots=True)37class Shock:38 time: int39 kind: str # 'fundamental_jump' | 'rate' | 'funding' | 'liquidity' | 'placebo'40 size: float = 0.041 asset: int | None = None42 duration: int = 0434445@dataclass(slots=True)46class ShockSchedule:47 shocks: list[Shock] = field(default_factory=list)4849 def at(self, t: int) -> list[Shock]:50 return [s for s in self.shocks if s.time == t]515253class NewsFeed:54 def __init__(self, cfg: NewsConfig, rng: np.random.Generator, n_assets: int) -> None:55 self.cfg, self.rng, self.K = cfg, rng, n_assets56 self.items: list[NewsItem] = []5758 def step(self, t: int, x_true: np.ndarray) -> list[NewsItem]:59 out = []60 for k in range(self.K):61 if self.rng.random() < self.cfg.rate:62 misleading = self.rng.random() < self.cfg.misleading_share63 noise = self.rng.standard_normal() * self.cfg.noise_sd64 bias = (self.cfg.misleading_bias * (1 if self.rng.random() < 0.5 else -1)) if misleading else 0.065 val = float(x_true[k] + noise + bias)66 direction = "above" if val > 0 else "below"67 item = NewsItem(time=t, asset=k, value=val, credible=not misleading, noise_sd=self.cfg.noise_sd,68 text=f"Analyst note: next dividend expected {abs(val):.1f} cents {direction} "69 f"its long-run mean.")70 out.append(item)71 self.items.extend(out)72 return out7374 def recent(self, t: int, window: int, asset: int | None = None) -> list[NewsItem]:75 return [n for n in self.items if t - window <= n.time <= t and (asset is None or n.asset == asset)]76