research: timestamped Yahoo analyst snapshots and a forward-only EPS-revision overlay

collect archives Yahoo's earnings/revenue estimates, EPS trend, EPS
revisions and price targets as immutable snapshots; a snapshot is usable
only after its created_at, so the vendor's 30/90-day history never reaches
earlier bars. evaluate replays overnight_recovery_top10 with an EPS-revision
re-rank of the quant top 50 on the same ledger and costs and returns NEEDS
EVIDENCE until 252 active sessions from 20 collection dates exist. The
agent's candidate pack now carries each name's analyst context.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CLAUDE.md

@@ -40,6 +40,7 @@ uv run python trader.py compare --market us --book selected_2026_09      # A nam
 
 # Agent-reviewed strategy overnight_recovery_agent_top10: daily loop (skill .claude/skills/agent-stock-review)
 uv run python agent_review.py candidates       # after the US close: top-50 pack -> data/agent/us/candidates_<as_of>.json
+uv run python agent_review.py candidates --analyst-refresh  # immutable Yahoo EPS-trend/revision snapshots + analyst context in the pack
 uv run python agent_review.py submit data/agent/us/draft_<as_of>.json   # before 16:00 New York of the next session
 uv run python agent_review.py evaluate         # score IC, veto outcomes, book vs overnight_recovery_top10
 
@@ -73,6 +74,7 @@ PYTHONPATH=. uv run python -m research.lab.us_target50                   # the b
 PYTHONPATH=. uv run --with numba python -m research.lab.qqq_scan         # alpha vs QQQ of every sleeve / factor / mega-cap variant
 PYTHONPATH=. uv run --with numba python -m research.lab.beat_qqq         # sleeves picked on 2006-16, judged 2017-26; odds of beating QQQ
 PYTHONPATH=. uv run --with numba python -m research.lab.momentum_fundamentals [--sensitivity]  # financial-statement vetoes/confirmations on overnight_recovery_top10
+uv run python -m research.lab.analyst_estimates evaluate  # EPS-revision overlay vs the same ledger/costs; forward only, no live promotion
 uv run python -m research.lab.fetch_insider && PYTHONPATH=. uv run python -m research.lab.insider        # SEC Form 4 insider factors
 PYTHONPATH=. uv run python -m research.lab.fetch_insider --universe mid small && PYTHONPATH=. uv run python -m research.lab.insider_mid  # S&P 400/600 Form 4 factors (all REJECT 2026-10-09)
 PYTHONPATH=. uv run python -m research.lab.fetch_lazy_prices && PYTHONPATH=. uv run python -m research.lab.lazy_prices  # 10-K/10-Q text similarity, 8-K earnings dates, SIC

agent_review.py

@@ -4,7 +4,7 @@ Daily loop of the agent-reviewed strategy ``overnight_recovery_agent_top10``
 scores with a point-in-time stamp, and measure the scores against what
 happened.
 
-    uv run python agent_review.py candidates [--refresh]   # pack for the next session
+    uv run python agent_review.py candidates [--refresh] [--analyst-refresh]
     uv run python agent_review.py submit data/agent/us/draft_<as_of>.json
     uv run python agent_review.py evaluate                  # IC, vetoes, book vs control
 
@@ -75,6 +75,11 @@ def cmd_candidates(args) -> None:
     bars = trader.strategy_bars("us", NAME, tickers, close, open_)
     strat = trader.STRATEGY_REGISTRY[NAME]()
     cand = strat.candidates(bars)
+    from research.lab.analyst_estimates import collect, candidate_context
+    if args.analyst_refresh:
+        collect(list(cand.index))
+    generated_at = datetime.now(timezone.utc).isoformat()
+    analyst = candidate_context(list(cand.index), generated_at)
 
     c = close[cand.index].ffill()
     r = lambda k: (c.iloc[-1] / c.iloc[-1 - k] - 1).round(4)                     # noqa: E731
@@ -109,6 +114,7 @@ def cmd_candidates(args) -> None:
             "overnight_mom_12m": float(row["overnight_mom_12m"]), "recovery_126d": float(row["recovery_126d"]),
             "held_weight": {k: v.get(t, 0.0) for k, v in held.items()},
             **links.get(t, {}),
+            "analyst_estimates": analyst.get(t),
             "previous": ({"as_of": prev["as_of"], "score": p.get("score"), "veto": p.get("veto", False),
                           "summary": p.get("summary", "")} if p else None),
         })
@@ -120,7 +126,10 @@ def cmd_candidates(args) -> None:
         "deadline_utc": str(ar.session_deadline(nxt)),
         # 0: the next session rebalances on these scores; otherwise only vetoes act until then
         "sessions_until_rebalance": int(until),
-        "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
+        "generated_at": generated_at,
+        "analyst_estimates": {"covered": len(analyst),
+                              "eligible": sum(r["eligible"] for r in analyst.values()),
+                              "note": "Known only after snapshot collection; stale fiscal metadata is excluded from the research overlay."},
         "rule": {"agent_weight": ar.AGENT_WEIGHT, "score_range": list(ar.SCORE_RANGE),
                  "veto_reasons": list(ar.VETO_REASONS), "top_n": strat.top_n, "pool": strat.pool,
                  "rebal_freq": strat.rebal_freq},
@@ -290,6 +299,7 @@ def main() -> None:
     sub = ap.add_subparsers(dest="cmd", required=True)
     p = sub.add_parser("candidates", help="write the candidate pack for the next session")
     p.add_argument("--refresh", action="store_true", help="download prices first (the monitor normally has)")
+    p.add_argument("--analyst-refresh", action="store_true", help="archive current Yahoo analyst estimates before writing the pack")
     p = sub.add_parser("submit", help="validate and store an agent scores file")
     p.add_argument("file")
     sub.add_parser("evaluate", help="score IC, veto outcomes, and the book against its control")

data_manager.py

@@ -195,7 +195,7 @@ def _save_unavailable(market: str, unavailable: dict[str, str]) -> None:
         json.dump(unavailable, f, indent=0, sort_keys=True)
 
 
-def _drop_sparse_rows(df: pd.DataFrame) -> pd.DataFrame:
+def _drop_sparse_rows(df: pd.DataFrame, refreshed_columns=None) -> pd.DataFrame:
     """Drop rows where too few of the tickers alive on that date have a bar."""
     df = df.dropna(how="all")
     if df.empty:
@@ -207,6 +207,9 @@ def _drop_sparse_rows(df: pd.DataFrame) -> pd.DataFrame:
     last = np.where(has_any, n - 1 - notna[::-1].argmax(axis=0), -1)
     pos = np.arange(n)[:, None]
     alive = (pos >= first) & (pos <= last + ALIVE_GRACE_ROWS)
+    if refreshed_columns is not None:
+        unrequested = ~df.columns.isin(refreshed_columns)
+        alive[:, unrequested] = notna[:, unrequested]
     coverage = notna.sum(axis=1) / np.maximum(alive.sum(axis=1), 1)
     keep = coverage >= MIN_ROW_COVERAGE
     if not keep.all():
@@ -234,6 +237,7 @@ def _adjustment_ratio(old: pd.Series, new: pd.Series) -> float | None:
 def _merge_refresh(existing: pd.DataFrame | None, new: pd.DataFrame,
                    window_start: pd.Timestamp | None) -> pd.DataFrame:
     """Overwrite requested tickers from ``window_start`` on with freshly downloaded bars."""
+    refreshed_columns = new.columns
     new = new.dropna(axis=1, how="all")
     if existing is None or existing.empty:
         return _drop_sparse_rows(new.sort_index())
@@ -254,7 +258,10 @@ def _merge_refresh(existing: pd.DataFrame | None, new: pd.DataFrame,
                       columns=list(out.columns) + [c for c in new.columns if c not in out.columns])
     window = out.index >= window_start
     out.loc[window, new.columns] = new.reindex(out.index[window]).to_numpy()
-    return _drop_sparse_rows(out.sort_index())
+    # Unrequested columns count only where an actual bar already exists.
+    # They cannot make a newly fetched date look sparse; older history stays intact.
+    valid = _drop_sparse_rows(out.loc[window], refreshed_columns=refreshed_columns).index
+    return out.loc[~window | out.index.isin(valid)].sort_index()
 
 
 def _fill_open(close: pd.DataFrame, open_raw: pd.DataFrame | None) -> pd.DataFrame:

overnight_paper.py

@@ -62,6 +62,7 @@ BUY_CUTOFF = timedelta(minutes=11)      # Alpaca: MOC orders close at 15:50 ET
 SELL_BEFORE_OPEN = timedelta(minutes=20)
 SELL_CUTOFF = timedelta(minutes=2)      # Alpaca: MOO orders close at 09:28 ET
 REPORT_AFTER_OPEN = timedelta(minutes=30)
+ACCEPTED_ORDER_STATUSES = {"accepted", "pending_new", "new", "accepted_for_bidding", "partially_filled", "filled"}
 
 
 # ---------------------------------------------------------------------------
@@ -209,13 +210,17 @@ def compute_signal(cl: Alpaca, session: date) -> dict:
 
 def _state() -> dict:
     path = os.path.join(OUT, "state.json")
-    return json.load(open(path)) if os.path.exists(path) else {}
+    if not os.path.exists(path):
+        return {}
+    with open(path) as fh:
+        return json.load(fh)
 
 
 def _save_state(st: dict) -> None:
     os.makedirs(OUT, exist_ok=True)
     tmp = os.path.join(OUT, ".state.json.tmp")
-    json.dump(st, open(tmp, "w"), indent=1)
+    with open(tmp, "w") as fh:
+        json.dump(st, fh, indent=1)
     os.replace(tmp, os.path.join(OUT, "state.json"))
 
 
@@ -235,10 +240,18 @@ def _submit(cl: Alpaca, kind: str, payload: dict, dry: bool) -> None:
         return
     resp = cl.order(payload)
     _log_order(kind, payload, resp)
-    ok = resp["http"] in (200, 201)
-    dup = resp["http"] == 422 and "client_order_id" in json.dumps(resp).lower()
-    print(f"  {tag} -> {'ok ' + str(resp.get('status')) if ok else 'already sent' if dup else 'HTTP ' + str(resp['http']) + ' ' + str(resp.get('message', resp.get('text', '')))[:120]}",
-          flush=True)
+    if resp["http"] == 422 and "client_order_id" in str(resp.get("message", "")).lower():
+        existing = cl.get("/orders:by_client_order_id", client_order_id=payload["client_order_id"])
+        matches = all(existing.get(k) == payload[k] for k in ("client_order_id", "symbol", "side", "type", "time_in_force"))
+        matches &= float(existing.get("qty", 0)) == float(payload["qty"])
+        if matches and existing.get("id") and existing.get("status") in ACCEPTED_ORDER_STATUSES:
+            print(f"  {tag} -> already accepted ({existing['status']})", flush=True)
+            return
+        raise RuntimeError(f"{tag}: client_order_id belongs to a mismatched or {existing.get('status')} order")
+    if resp["http"] not in (200, 201) or not resp.get("id") or resp.get("status") not in ACCEPTED_ORDER_STATUSES:
+        reason = str(resp.get("message", resp.get("text", resp.get("status", "missing order acknowledgement"))))[:200]
+        raise RuntimeError(f"{tag}: HTTP {resp['http']} {reason}")
+    print(f"  {tag} -> ok {resp['status']}", flush=True)
 
 
 # ---------------------------------------------------------------------------
@@ -276,7 +289,8 @@ def buy(cl: Alpaca, dry: bool = False, force: bool = False) -> None:
         return
     sig = compute_signal(cl, today)
     os.makedirs(os.path.join(OUT, "signals"), exist_ok=True)
-    json.dump(sig, open(os.path.join(OUT, "signals", f"{today}.json"), "w"), indent=1)
+    with open(os.path.join(OUT, "signals", f"{today}.json"), "w") as fh:
+        json.dump(sig, fh, indent=1)
     acct = cl.get("/account")
     equity = float(acct["equity"])
     px = latest_prices(cl, sig["top"])

research/lab/analyst_estimates.py

@@ -0,0 +1,348 @@
+"""Timestamped Yahoo analyst snapshots and a forward-only momentum overlay.
+
+uv run python -m research.lab.analyst_estimates collect --candidates data/agent/us/candidates_<as_of>.json
+uv run python -m research.lab.analyst_estimates evaluate
+
+The vendor's 30/90-day EPS trend is known only when collected. It never
+creates observations on those earlier dates. Price targets are descriptive;
+the research overlay uses EPS revisions within the quant's top 50 names.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import math
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+
+from strategies.agent_review import session_deadline
+from strategies.robust_momentum import RobustMomentumStrategy
+
+ROOT = Path(__file__).resolve().parents[2]
+SNAPSHOTS = ROOT / "data/research/analyst_estimates"
+PROVIDER = "yahoo_finance"
+MAX_AGE_DAYS = 7
+MIN_ANALYSTS = 5
+EPS_FLOOR = 0.25
+POOL = 50
+MIN_COVERAGE = 0.70
+REVISION_WEIGHT = 0.20
+FEATURES = ("eps_revision_30d", "eps_revision_90d", "revision_balance_30d")
+
+
+def _utc(value) -> pd.Timestamp:
+    ts = pd.Timestamp(value)
+    if pd.isna(ts) or ts.tzinfo is None:
+        raise ValueError("Snapshot timestamps must be timezone-aware")
+    return ts.tz_convert("UTC")
+
+
+def _clean(value):
+    if isinstance(value, dict):
+        return {str(k): _clean(v) for k, v in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_clean(v) for v in value]
+    if value is None or pd.isna(value):
+        return None
+    if isinstance(value, np.generic):
+        return value.item()
+    return value
+
+
+def collect(tickers: list[str], directory: Path = SNAPSHOTS) -> Path:
+    import yfinance as yf
+    from research.lab.fetch_us import _proxy_env
+
+    _proxy_env()
+    rows = {}
+    for symbol in dict.fromkeys(tickers):
+        ticker = yf.Ticker(symbol)
+        row = {"source_url": f"https://finance.yahoo.com/quote/{symbol}/analysis/", "errors": []}
+        for field in ("earnings_estimate", "revenue_estimate", "eps_trend", "eps_revisions",
+                      "analyst_price_targets"):
+            try:
+                value = getattr(ticker, field)
+                row[field] = _clean(value.to_dict(orient="index") if isinstance(value, pd.DataFrame) else value)
+                if not row[field]:
+                    row["errors"].append(f"{field}: unavailable")
+            except Exception as exc:  # Preserve missing endpoints; never turn a failure into a zero signal.
+                row[field] = {}
+                row["errors"].append(f"{field}: {type(exc).__name__}")
+        # yfinance's public estimate tables drop endDate. Preserve the fetched
+        # period identity to make fiscal rollovers and stale metadata visible.
+        trend = ticker._analysis._earnings_trend or []
+        row["period_ends"] = {r["period"]: r.get("endDate") for r in trend}
+        row["observed_at"] = datetime.now(timezone.utc).isoformat()
+        rows[symbol] = row
+        print(f"{symbol}: {'available' if row['eps_trend'] else 'missing EPS trend'}", flush=True)
+        time.sleep(0.2)
+    now = datetime.now(timezone.utc)
+    snapshot = {"schema": 1, "provider": PROVIDER, "created_at": now.isoformat(),
+                "yfinance_version": yf.__version__, "rows": rows}
+    directory = Path(directory)
+    directory.mkdir(parents=True, exist_ok=True)
+    path = directory / f"snapshot_{now:%Y%m%dT%H%M%S%fZ}.json"
+    with path.open("x") as fh:
+        json.dump(snapshot, fh, indent=1, allow_nan=False)
+    print(f"Saved {len(rows)} immutable observations -> {path}", flush=True)
+    return path
+
+
+def load_snapshots(directory: Path = SNAPSHOTS) -> list[dict]:
+    snapshots = []
+    for path in sorted(Path(directory).glob("snapshot_*.json")):
+        d = json.loads(path.read_text())
+        if d.get("schema") != 1 or d.get("provider") != PROVIDER:
+            raise ValueError(f"Unsupported analyst snapshot: {path}")
+        created = _utc(d["created_at"])
+        for row in d["rows"].values():
+            if _utc(row["observed_at"]) > created:
+                raise ValueError(f"Observation newer than its snapshot: {path}")
+        snapshots.append(d)
+    return sorted(snapshots, key=lambda d: _utc(d["created_at"]))
+
+
+def _number(value) -> float | None:
+    if isinstance(value, bool):
+        return None
+    try:
+        v = float(value)
+        return v if math.isfinite(v) else None
+    except (ValueError, TypeError):
+        return None
+
+
+def row_features(row: dict) -> dict:
+    period = "+1y"
+    trend = row.get("eps_trend", {}).get(period, {})
+    estimate = row.get("earnings_estimate", {}).get(period, {})
+    revisions = row.get("eps_revisions", {}).get(period, {})
+    result = {"period": period, "period_end": row.get("period_ends", {}).get(period),
+              "observed_at": row["observed_at"], "source_url": row["source_url"],
+              "eps": _number(trend.get("current")),
+              "analysts": _number(estimate.get("numberOfAnalysts")),
+              **{k: None for k in FEATURES}, "eligible": False, "quality_flags": []}
+    end = pd.to_datetime(result["period_end"], errors="coerce")
+    observed = _utc(row["observed_at"]).tz_localize(None).normalize()
+    if pd.isna(end) or end <= observed:
+        result["quality_flags"].append("missing_or_expired_fiscal_period")
+    current_end = pd.to_datetime(row.get("period_ends", {}).get("0y"), errors="coerce")
+    if pd.isna(current_end) or current_end <= observed:
+        result["quality_flags"].append("stale_current_year_metadata")
+    if not pd.isna(end) and not pd.isna(current_end) and end <= current_end:
+        result["quality_flags"].append("inconsistent_fiscal_period_order")
+    if (result["analysts"] or 0) < MIN_ANALYSTS:
+        result["quality_flags"].append("insufficient_analyst_coverage")
+    current = result["eps"]
+    for days in (30, 90):
+        old = _number(trend.get(f"{days}daysAgo"))
+        if current is not None and current > EPS_FLOOR and old is not None and old > EPS_FLOOR:
+            result[f"eps_revision_{days}d"] = current / old - 1.0
+    up, down = (_number(revisions.get(k)) for k in ("upLast30days", "downLast30days"))
+    if up is not None and down is not None and min(up, down) >= 0:
+        result["revision_balance_30d"] = (up - down) / (up + down) if up + down else 0.0
+    if any(result[k] is None for k in FEATURES):
+        result["quality_flags"].append("missing_revisions_or_small_nonpositive_eps")
+    if current is not None and _number(estimate.get("avg")) is not None:
+        if not np.isclose(current, float(estimate["avg"]), rtol=0.01, atol=0.01):
+            result["quality_flags"].append("inconsistent_consensus_eps")
+    if current is not None and current > EPS_FLOOR:
+        lo, hi = (_number(estimate.get(k)) for k in ("low", "high"))
+        result["eps_range_over_mean"] = (hi - lo) / current if lo is not None and hi is not None else None
+    targets = row.get("analyst_price_targets", {})
+    price, target = _number(targets.get("current")), _number(targets.get("median"))
+    result["target_median_upside"] = target / price - 1 if price and price > 0 and target and target > 0 else None
+    result["price_targets"] = targets
+    # Old 0y metadata is a data-quality gate, even when +1y still lies ahead.
+    result["eligible"] = not result["quality_flags"]
+    return result
+
+
+def candidate_context(tickers: list[str], cutoff, snapshots: list[dict] | None = None) -> dict[str, dict]:
+    cutoff = _utc(cutoff)
+    out = {}
+    ordered = load_snapshots() if snapshots is None else sorted(snapshots, key=lambda d: _utc(d["created_at"]))
+    for d in ordered:
+        created = _utc(d["created_at"])
+        if not cutoff - pd.Timedelta(days=MAX_AGE_DAYS) <= created <= cutoff:
+            continue
+        for ticker in tickers:
+            if ticker in d["rows"]:
+                row = d["rows"][ticker]
+                if cutoff - pd.Timedelta(days=MAX_AGE_DAYS) <= _utc(row["observed_at"]) <= cutoff:
+                    out[ticker] = {"provider": d["provider"], "snapshot_created_at": d["created_at"],
+                                   **row_features(row)}
+    return out
+
+
+def feature_history(dates, columns, snapshots: list[dict]) -> dict[str, pd.DataFrame]:
+    frames = {k: pd.DataFrame(np.nan, index=dates, columns=columns) for k in FEATURES}
+    for day in dates:
+        cutoff = session_deadline(day)
+        context = candidate_context(list(columns), cutoff, snapshots)
+        for ticker, row in context.items():
+            if row["eligible"]:
+                for k in FEATURES:
+                    frames[k].at[day, ticker] = row[k]
+    return frames
+
+
+def blend_revisions(base: pd.DataFrame, features: dict[str, pd.DataFrame]) -> tuple[pd.DataFrame, pd.Series]:
+    output = base.copy()
+    active = pd.Series(False, index=base.index)
+    for day, row in base.iterrows():
+        candidates = row.dropna().sort_values(ascending=False, kind="stable").head(POOL).index
+        f = pd.DataFrame({k: features[k].loc[day].reindex(candidates) for k in FEATURES})
+        valid = f.notna().all(axis=1)
+        if len(f) < 10 or valid.sum() < math.ceil(len(f) * MIN_COVERAGE):
+            continue
+        # A tie carries no directional information; missing names stay neutral.
+        rank = (f.loc[valid].rank(method="average") - 1) / max(int(valid.sum()) - 1, 1)
+        score = 0.50 * rank[FEATURES[0]] + 0.25 * rank[FEATURES[1]] + 0.25 * rank[FEATURES[2]]
+        quant = pd.Series(1 - np.arange(len(f)) / max(len(f) - 1, 1), index=candidates)
+        output.loc[day, row.notna()] = -2.0
+        output.loc[day, candidates] = quant + REVISION_WEIGHT * (2 * score.reindex(candidates).fillna(0.5) - 1)
+        active.at[day] = True
+    return output, active
+
+
+class AnalystRevisionStrategy(RobustMomentumStrategy):
+    """Research only: reuse the base strategy's timing, turnover buffer and sizing."""
+    def __init__(self, snapshots: list[dict], **kwargs):
+        super().__init__("overnight_recovery", **kwargs)
+        self.snapshots = snapshots
+        self._quant_score = self._score
+        self._score = self._revision_score
+        self.active = pd.Series(dtype=bool)
+
+    def _revision_score(self, panel):
+        base = self._quant_score(panel)
+        eligible = panel["close"].notna()
+        if "eligible" in panel:
+            eligible &= panel["eligible"].reindex_like(base).fillna(False).astype(bool)
+        base = base.where(eligible)
+        features = feature_history(base.index, base.columns, self.snapshots)
+        score, self.active = blend_revisions(base, features)
+        return score
+
+
+def evaluate(directory: Path = SNAPSHOTS, destination: Path | None = None) -> dict:
+    from backtest_runner import ENGINE_VERSION, load_stock_panels, prepare_strategy, replay_prepared
+    from research.lab.trials import append_trials
+    import metrics
+
+    snapshots = load_snapshots(directory)
+    stamp = datetime.now(timezone.utc)
+    destination = destination or ROOT / f"data/research/results/analyst_revisions_{stamp:%Y%m%dT%H%M%S%fZ}"
+    destination.mkdir(parents=True, exist_ok=False)
+    protocol = {
+        "status": "approved_local_forward_research", "created_at": stamp.isoformat(),
+        "claim": "EPS revisions add information beyond overnight/recovery momentum",
+        "competing_hypothesis": "Analysts follow the same price trend; the overlay adds turnover without alpha",
+        "prediction": "Net CAGR and Sharpe rise on the same sessions after collection; gains survive higher costs",
+        "engine": ENGINE_VERSION, "baseline": "overnight_recovery_top10",
+        "pool": POOL, "revision_weight": REVISION_WEIGHT, "coverage_gate": MIN_COVERAGE,
+        "features": dict(zip(FEATURES, (0.50, 0.25, 0.25))),
+        "target_prices": "descriptive only; not a return/probability or a selection factor",
+        "controls": "same PIT membership, warmup, close fills, top10, rebalance10, buffer2, whole shares",
+        "costs_per_side": [0.001, 0.0025], "capital": 10000,
+        "commission": "max($2, $1.88 + $0.009/share)",
+        "minimum_evidence": "252 active signal sessions, 20 distinct collection dates, active on >=70% of evaluated sessions",
+        "decision_rule": "Research candidate only: CAGR improves and Sharpe +0.10 at both costs; MaxDD deterioration <=2pp",
+        "promotion_boundary": "No live promotion; historical searches already examined are development, not independent OOS",
+        "data_boundary": "Vendor EPS trend is usable only after created_at; stale fiscal metadata and loss/near-zero EPS excluded",
+        "metrics": "calendar-time CAGR, daily-return Sharpe (rf=0), MaxDD, orders, fees, paired net returns",
+        "ablations": "base vs fixed EPS-only overlay; 10bp vs 25bp; no parameter search",
+        "expected_artifacts": ["protocol.json", "results.json", "returns.csv when enough observations exist"],
+        "snapshots": {str(p): hashlib.sha256(p.read_bytes()).hexdigest()
+                      for p in sorted(Path(directory).glob("snapshot_*.json"))},
+        "code_hashes": {p.name: hashlib.sha256(p.read_bytes()).hexdigest()
+                        for p in (Path(__file__), ROOT / "strategies/robust_momentum.py", ROOT / "execution.py",
+                                  ROOT / "signals.py", ROOT / "backtest_runner.py", ROOT / "metrics.py")},
+        "data_hashes": {str(p): hashlib.sha256(p.read_bytes()).hexdigest()
+                        for p in (ROOT / "data/us.csv", ROOT / "data/us_open.csv", ROOT / "data/us_volume.csv",
+                                  ROOT / "data/sp500_history.json") if p.exists()},
+    }
+    (destination / "protocol.json").write_text(json.dumps(protocol, indent=2))
+    result = {"verdict": "NEEDS EVIDENCE", "snapshot_count": len(snapshots), "comparisons": []}
+    if snapshots:
+        stock = load_stock_panels("us", [protocol["baseline"]], None, refresh=False)
+        prices = stock["panels"]["close"]
+        first = _utc(snapshots[0]["created_at"]).tz_convert("America/New_York").date()
+        days = prices.index[prices.index.date >= first]
+        result["priced_sessions_after_collection"] = len(days)
+        if len(days) >= 2:
+            strategy = AnalystRevisionStrategy(snapshots)
+            args = dict(member=stock["member"], start=str(days[0].date()), end=str(days[-1].date()))
+            plan = prepare_strategy(protocol["baseline"], "us", stock["panels"], stock["stocks"],
+                                    strategy=strategy, **args)
+            active = strategy.active.loc[days[0]:days[-1]]
+            result["active_signal_sessions"] = int(active.sum())
+            result["snapshot_collection_dates"] = len({_utc(d["created_at"]).date() for d in snapshots
+                                                       if _utc(d["created_at"]) <= session_deadline(days[-1])})
+            if active.any():
+                start = active.index[active][0] + pd.offsets.BDay(1)
+                # Shifted signals trade one session after the collection cutoff.
+                live_days = plan["prices"].index[plan["prices"].index >= start]
+                result["evaluated_sessions"] = len(live_days)
+                if len(live_days) >= 2:
+                    args["start"] = str(live_days[0].date())
+                    plan = prepare_strategy(protocol["baseline"], "us", stock["panels"], stock["stocks"],
+                                            strategy=strategy, **args)
+                    base = prepare_strategy(protocol["baseline"], "us", stock["panels"], stock["stocks"], **args)
+                    enough = (int(active.sum()) >= 252 and result["snapshot_collection_dates"] >= 20
+                              and float(active.mean()) >= MIN_COVERAGE)
+                    returns = {}
+                    for cost in protocol["costs_per_side"]:
+                        pair = {}
+                        for name, p in (("base", base), ("eps_revisions", plan)):
+                            run = replay_prepared(p, 10000, cost, 2.0, fee_base=1.88, fee_per_share=0.009)
+                            r = run["equity"].pct_change().dropna()
+                            returns[f"{name}_{cost}"] = r
+                            pair[name] = {"orders": len(run["trade_log"]),
+                                          "fees": sum(t["commission"] for t in run["trade_log"]),
+                                          "total_return": float(run["equity"].iloc[-1] / run["equity"].iloc[0] - 1)}
+                            if enough:
+                                pair[name].update(metrics.raw_summary(run["equity"]))
+                        result["comparisons"].append({"cost": cost, **pair})
+                    pd.DataFrame(returns).to_csv(destination / "returns.csv")
+                    if enough:
+                        passed = all(c["eps_revisions"]["annualizedReturn"] > c["base"]["annualizedReturn"]
+                                     and c["eps_revisions"]["sharpeRatio"] >= c["base"]["sharpeRatio"] + 0.10
+                                     and c["eps_revisions"]["maxDrawdown"] >= c["base"]["maxDrawdown"] - 0.02
+                                     for c in result["comparisons"])
+                        result["verdict"] = "FORWARD CANDIDATE" if passed else "RETAIN BASELINE"
+                    append_trials(destination / "trials.jsonl", result["comparisons"], protocol)
+    result["boundary"] = "Insufficient forward evidence never receives annualized performance or a promotion verdict."
+    (destination / "results.json").write_text(json.dumps(result, indent=2, allow_nan=False))
+    print(json.dumps(result, indent=2), flush=True)
+    print(f"Saved -> {destination}", flush=True)
+    return result
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    sub = ap.add_subparsers(dest="action", required=True)
+    p = sub.add_parser("collect")
+    g = p.add_mutually_exclusive_group(required=True)
+    g.add_argument("--candidates", type=Path)
+    g.add_argument("--tickers", nargs="+")
+    p.add_argument("--directory", type=Path, default=SNAPSHOTS)
+    p = sub.add_parser("evaluate")
+    p.add_argument("--directory", type=Path, default=SNAPSHOTS)
+    p.add_argument("--output", type=Path)
+    args = ap.parse_args()
+    if args.action == "collect":
+        tickers = args.tickers if args.tickers else [r["ticker"] for r in json.loads(args.candidates.read_text())["candidates"]]
+        collect(tickers, args.directory)
+    else:
+        evaluate(args.directory, args.output)
+
+
+if __name__ == "__main__":
+    main()

tests/test_analyst_estimates.py

@@ -0,0 +1,143 @@
+"""Analyst vintages must never change past trades or fabricate missing evidence."""
+import copy
+import contextlib
+import io
+import json
+import tempfile
+import unittest
+from unittest import mock
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+
+from research.lab.analyst_estimates import (AnalystRevisionStrategy, FEATURES, blend_revisions,
+                                           candidate_context, evaluate, feature_history, load_snapshots, row_features)
+from strategies.agent_review import session_deadline
+from strategies.robust_momentum import RobustMomentumStrategy
+from tests.test_robust_strategies import _bars
+
+
+def _row(observed_at, value=12.0):
+    return {"observed_at": pd.Timestamp(observed_at).isoformat(), "source_url": "https://finance.yahoo.com/",
+            "period_ends": {"0y": "2030-12-31", "+1y": "2031-12-31"},
+            "eps_trend": {"+1y": {"current": value, "30daysAgo": 10.0, "90daysAgo": 9.0}},
+            "eps_revisions": {"+1y": {"upLast30days": 4, "downLast30days": 1}},
+            "earnings_estimate": {"+1y": {"avg": value, "numberOfAnalysts": 10, "low": 8, "high": 15}},
+            "analyst_price_targets": {"current": 100, "median": 150}}
+
+
+def _snapshot(observed_at, rows):
+    return {"schema": 1, "provider": "yahoo_finance", "created_at": pd.Timestamp(observed_at).isoformat(),
+            "rows": rows}
+
+
+class AnalystVintageTests(unittest.TestCase):
+    def test_vendor_30day_history_cannot_be_used_before_collection(self):
+        stamp = "2026-10-12T18:00:00Z"
+        snapshot = _snapshot(stamp, {"MU": _row(stamp)})
+        dates = pd.bdate_range("2026-10-01", "2026-10-23")
+        features = feature_history(dates, ["MU"], [snapshot])
+        self.assertTrue(features[FEATURES[0]].loc[:"2026-10-09"].isna().all().all())
+        self.assertAlmostEqual(features[FEATURES[0]].at[pd.Timestamp("2026-10-12"), "MU"], .2)
+        self.assertTrue(features[FEATURES[0]].loc["2026-10-20":].isna().all().all())
+        self.assertEqual(candidate_context(["MU"], "2026-10-12T17:00Z", [snapshot]), {})
+
+    def test_collection_after_cutoff_is_used_on_a_later_bar(self):
+        stamp = session_deadline(pd.Timestamp("2026-10-12")) + pd.Timedelta(seconds=1)
+        snapshot = _snapshot(stamp, {"MU": _row(stamp)})
+        features = feature_history(pd.bdate_range("2026-10-12", periods=2), ["MU"], [snapshot])
+        self.assertTrue(pd.isna(features[FEATURES[0]].iloc[0, 0]))
+        self.assertTrue(np.isfinite(features[FEATURES[0]].iloc[1, 0]))
+
+    def test_fiscal_rollover_losses_low_coverage_and_conflicting_eps_do_not_score(self):
+        row = _row("2026-10-12T18:00Z")
+        self.assertTrue(row_features(row)["eligible"])
+        cases = []
+        stale = copy.deepcopy(row); stale["period_ends"]["0y"] = "2026-08-31"; cases.append(stale)
+        expired = copy.deepcopy(row); expired["period_ends"]["+1y"] = "2026-08-31"; cases.append(expired)
+        losses = copy.deepcopy(row); losses["eps_trend"]["+1y"]["30daysAgo"] = -1; cases.append(losses)
+        near_zero = copy.deepcopy(row); near_zero["eps_trend"]["+1y"]["90daysAgo"] = .01; cases.append(near_zero)
+        missing = copy.deepcopy(row); missing["eps_revisions"] = {}; cases.append(missing)
+        sparse = copy.deepcopy(row); sparse["earnings_estimate"]["+1y"]["numberOfAnalysts"] = 2; cases.append(sparse)
+        conflict = copy.deepcopy(row); conflict["earnings_estimate"]["+1y"]["avg"] = 20; cases.append(conflict)
+        for case in cases:
+            self.assertFalse(row_features(case)["eligible"])
+
+    def test_snapshot_rejects_mixed_providers_and_invalid_observation_time(self):
+        stamp = "2026-10-12T18:00Z"
+        for provider, observed in (("other", stamp), ("yahoo_finance", "2026-10-13T18:00Z")):
+            with tempfile.TemporaryDirectory() as tmp:
+                d = _snapshot(stamp, {"MU": _row(observed)})
+                d["provider"] = provider
+                Path(tmp, "snapshot_test.json").write_text(json.dumps(d))
+                with self.assertRaises(ValueError):
+                    load_snapshots(Path(tmp))
+
+
+class AnalystOverlayTests(unittest.TestCase):
+    def setUp(self):
+        self.bars = _bars(n_days=420, n_assets=40)
+        self.kw = dict(top_n=5, rebal_freq=5, buffer=.5, warmup=260)
+
+    def test_no_snapshots_keeps_the_exact_baseline_targets(self):
+        base = RobustMomentumStrategy("overnight_recovery", **self.kw).generate_signals(self.bars)
+        overlay = AnalystRevisionStrategy([], **self.kw).generate_signals(self.bars)
+        pd.testing.assert_frame_equal(base, overlay)
+
+    def test_new_snapshot_changes_future_targets_but_preserves_the_past(self):
+        dates = self.bars["close"].index
+        k = 260 + 5 * 20
+        stamp = session_deadline(dates[k]) - pd.Timedelta(hours=1)
+        base = RobustMomentumStrategy("overnight_recovery", **self.kw).generate_signals(self.bars)
+        held = set(base.columns[base.iloc[k + 1] > 0])
+        rows = {t: _row(stamp, 1.0 if t in held else 30.0) for t in base.columns}
+        snapshot = _snapshot(stamp, rows)
+        strategy = AnalystRevisionStrategy([snapshot], **self.kw)
+        overlay = strategy.generate_signals(self.bars)
+        pd.testing.assert_frame_equal(base.iloc[:k + 1], overlay.iloc[:k + 1])
+        self.assertFalse(np.allclose(base.iloc[k + 1], overlay.iloc[k + 1]))
+        part = {f: x.iloc[:k + 3] for f, x in self.bars.items()}
+        partial = AnalystRevisionStrategy([snapshot], **self.kw).generate_signals(part)
+        pd.testing.assert_frame_equal(partial, overlay.loc[partial.index])
+        self.assertTrue((overlay.sum(axis=1) <= 1 + 1e-9).all())
+
+    def test_insufficient_feature_coverage_retains_base_and_missing_names_are_neutral(self):
+        day = pd.Timestamp("2026-10-12")
+        base = pd.DataFrame([np.linspace(1, 0, 10)], index=[day], columns=list("ABCDEFGHIJ"))
+        features = {k: base.copy() * np.nan for k in FEATURES}
+        for f in features.values():
+            f.iloc[0, :6] = 1.0
+        output, active = blend_revisions(base, features)
+        pd.testing.assert_frame_equal(output, base)
+        self.assertFalse(active.any())
+        for f in features.values():
+            f.iloc[0, :7] = 1.0
+        output, active = blend_revisions(base, features)
+        self.assertTrue(active.any())
+        np.testing.assert_allclose(output, base)  # Tied data and missing data both carry zero adjustment.
+
+    def test_short_forward_replay_uses_the_ledger_without_claiming_annualized_improvement(self):
+        dates = self.bars["close"].index
+        stamp = session_deadline(dates[400]) - pd.Timedelta(hours=1)
+        member = pd.DataFrame(True, index=dates, columns=self.bars["close"].columns)
+        stock = {"panels": self.bars, "member": member, "stocks": list(member.columns)}
+        rows = {t: _row(stamp, 5.0 + i) for i, t in enumerate(member.columns)}
+        with tempfile.TemporaryDirectory() as tmp:
+            directory = Path(tmp, "snapshots"); directory.mkdir()
+            Path(directory, "snapshot_test.json").write_text(json.dumps(_snapshot(stamp, rows)))
+            with mock.patch("backtest_runner.load_stock_panels", return_value=stock), contextlib.redirect_stdout(io.StringIO()):
+                result = evaluate(directory, Path(tmp, "results"))
+            self.assertEqual(result["verdict"], "NEEDS EVIDENCE")
+            self.assertEqual(len(result["comparisons"]), 2)
+            for comparison in result["comparisons"]:
+                for name in ("base", "eps_revisions"):
+                    self.assertGreater(comparison[name]["orders"], 0)
+                    self.assertGreater(comparison[name]["fees"], 0)
+                    self.assertTrue(np.isfinite(comparison[name]["total_return"]))
+                    self.assertNotIn("annualizedReturn", comparison[name])
+                    self.assertNotIn("sharpeRatio", comparison[name])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_live_pipeline.py

@@ -1,6 +1,7 @@
 """Contracts of the live pipeline: price panel refresh, session gating,
 order sizing, and how results are reported."""
 
+import copy
 import unittest
 from datetime import datetime, timezone
 from types import SimpleNamespace
@@ -62,17 +63,47 @@ class PanelRefreshTests(unittest.TestCase):
         self.assertNotIn(panel.index[2], cleaned.index)
         self.assertEqual(len(cleaned), 4)
 
+    def test_subset_refresh_keeps_trading_days_without_erasing_older_history(self):
+        stored = _panel({"A": [np.nan, 10, 11], "B": [np.nan, 20, 21],
+                         "C": [30, 31, 32], "D": [40, 41, 42], "E": [50, 51, 52]})
+        fresh = pd.DataFrame({"A": [11, 12], "B": [21, 22]},
+                             index=pd.to_datetime(["2026-09-16", "2026-09-17"]))
+        merged = data_manager._merge_refresh(stored, fresh, fresh.index[0])
+        self.assertIn(fresh.index[-1], merged.index)
+        self.assertEqual(merged.loc[fresh.index[-1], "A"], 12)
+        self.assertTrue(pd.isna(merged.loc[fresh.index[-1], "C"]))
+        pd.testing.assert_frame_equal(merged.loc[stored.index[:2]], stored.iloc[:2], check_freq=False)
+
+    def test_subset_refresh_still_drops_partial_rows_in_the_requested_universe(self):
+        stored = _panel({c: [10, 10, 10] for c in "ABCDEFGH"})
+        fresh = pd.DataFrame({"A": [10, np.nan], "B": [10, np.nan],
+                              "C": [10, np.nan], "D": [10, 11]},
+                             index=pd.to_datetime(["2026-09-16", "2026-09-17"]))
+        merged = data_manager._merge_refresh(stored, fresh, fresh.index[0])
+        self.assertNotIn(fresh.index[-1], merged.index)
+        self.assertIn(stored.index[0], merged.index)
+
+    def test_failed_requested_columns_still_count_toward_freshness(self):
+        stored = _panel({c: [10, 10, 10] for c in "ABCDEFGH"})
+        fresh = pd.DataFrame({"A": [10, 11], "B": [np.nan, np.nan],
+                              "C": [np.nan, np.nan], "D": [np.nan, np.nan]},
+                             index=pd.to_datetime(["2026-09-16", "2026-09-17"]))
+        merged = data_manager._merge_refresh(stored, fresh, fresh.index[0])
+        self.assertNotIn(fresh.index[-1], merged.index)
+        self.assertEqual(merged.loc[stored.index[-1], "B"], 10)
+
 
 class LiveBookTests(unittest.TestCase):
     BOOK = {"label": "test", "start": {"us": "2026-09-16"}, "capital": {"us": 5_000.0},
             "fee_per_order": {"us": 1.0}, "strategies": {"us": ["momentum"]}}
 
-    def _run(self, state: dict):
-        close = _panel({"AAA": [10, 11, 12, 13, 14], "BBB": [20, 20, 21, 21, 22],
-                        "SPY": [1, 1, 1, 1, 1]})
+    def _run(self, state: dict, close=None, targets=None):
+        if close is None:
+            close = _panel({"AAA": [10, 11, 12, 13, 14], "BBB": [20, 20, 21, 21, 22],
+                            "SPY": [1, 1, 1, 1, 1]})
         args = SimpleNamespace(market="us", strategy="momentum", book="test", capital=None,
                                tx_cost=0.001, fixed_fee=2.0, integer_shares=True)
-        equal = pd.DataFrame(0.5, index=close.index, columns=["AAA", "BBB"])
+        equal = targets if targets is not None else pd.DataFrame(0.5, index=close.index, columns=["AAA", "BBB"])
         with mock.patch.dict(trader.LIVE_BOOKS, {"test": self.BOOK}), \
                 mock.patch.object(trader, "strategy_universe", lambda m, n: (["AAA", "BBB"], "SPY")), \
                 mock.patch.object(trader, "pit_candidates", lambda m, n, s: []), \
@@ -92,6 +123,46 @@ class LiveBookTests(unittest.TestCase):
         with mock.patch.dict(self.BOOK, {"start": {"us": "2026-10-01"}}):
             self.assertIsNone(self._run({}))
 
+    def test_backfilled_session_replays_later_trades_and_is_idempotent(self):
+        close = _panel({"AAA": [10, 11, 12, 13, 14], "BBB": [20, 20, 21, 21, 22],
+                        "SPY": [1, 1, 1, 1, 1]})
+        targets = pd.DataFrame({"AAA": [1, 1, 1, 0, 1], "BBB": [0, 0, 0, 1, 0]}, index=close.index)
+        expected = self._run({}, close, targets)
+        missing = close.index[3]
+        state = self._run({}, close.drop(missing), targets)
+        self.assertNotEqual(state["cash"], expected["cash"])
+        repaired = self._run(state, close, targets)
+        self.assertEqual(repaired["holdings"], expected["holdings"])
+        self.assertAlmostEqual(repaired["cash"], expected["cash"])
+        self.assertEqual(repaired["daily_equity"], expected["daily_equity"])
+        self.assertEqual(repaired["daily_log"], expected["daily_log"])
+        self.assertEqual(repaired["trade_log"], expected["trade_log"])
+        self.assertEqual(repaired["last_evening"], "2026-09-18")
+        before = copy.deepcopy(repaired)
+        self.assertEqual(self._run(repaired, close, targets), before)
+
+    def test_backfilled_first_book_session_restores_original_capital(self):
+        close = _panel({"AAA": [10, 11, 12, 13, 14], "BBB": [20, 20, 21, 21, 22],
+                        "SPY": [1, 1, 1, 1, 1]})
+        expected = self._run({}, close)
+        state = self._run({}, close.drop(close.index[2]))
+        repaired = self._run(state, close)
+        self.assertEqual(repaired["daily_equity"], expected["daily_equity"])
+        self.assertEqual(repaired["holdings"], expected["holdings"])
+        self.assertAlmostEqual(repaired["cash"], expected["cash"])
+
+    def test_legacy_checkpoint_restores_cash_when_old_trade_values_were_rounded(self):
+        state = trader.init_state("us", "momentum", 50)
+        state.update(cash=9.98, holdings={"AAA": 4}, last_evening="2026-09-18",
+                     daily_equity={"2026-09-16": 50, "2026-09-18": 49.98},
+                     daily_log=[{"date": "2026-09-16", "cash": 50, "holdings": {}}],
+                     trade_log=[{"date": "2026-09-18", "action": "BUY", "ticker": "AAA",
+                                 "shares": 4, "value": 40, "commission": 0}])
+        trader._rewind_state(state, pd.Timestamp("2026-09-17"))
+        self.assertEqual(state["cash"], 50)
+        self.assertEqual(state["holdings"], {})
+        self.assertEqual(state["last_evening"], "2026-09-16")
+
     def test_book_strategies_are_live_strategies_of_their_market(self):
         for book, cfg in trader.LIVE_BOOKS.items():
             for market, names in cfg["strategies"].items():

tests/test_overnight_paper.py

@@ -0,0 +1,109 @@
+import tempfile
+import unittest
+from datetime import datetime
+from unittest import mock
+
+import pandas as pd
+
+import overnight_paper as paper
+
+
+class PaperOrders:
+    def __init__(self, responses, held=None):
+        self.responses = iter(responses)
+        self.held = held or []
+        self.pending = []
+        self.sent = []
+        self.existing = None
+
+    def get(self, path, **params):
+        if path == "/account":
+            return {"equity": "10000"}
+        if path == "/positions":
+            return self.held
+        if path == "/orders":
+            return self.pending
+        if path == "/orders:by_client_order_id" and self.existing:
+            return self.existing
+        raise RuntimeError("order not found")
+
+    def order(self, payload):
+        self.sent.append(payload.copy())
+        response = next(self.responses)
+        if isinstance(response, int):
+            response = {"http": response, "message": "insufficient buying power"}
+        else:
+            response = {"http": 201, "id": str(len(self.sent)), "status": "accepted", **payload, **response}
+        if response["http"] == 201 and response.get("status") == "accepted":
+            self.pending.append(response)
+        return response
+
+
+class OvernightOrderTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        day = datetime(2026, 10, 12, 15, 40, tzinfo=paper.NY)
+        self.day = day
+        calendar = pd.DataFrame([{"date": day.date(), "open": day.replace(hour=9, minute=30),
+                                  "close": day.replace(hour=16, minute=0)}]).set_index("date")
+
+        class Clock(datetime):
+            @classmethod
+            def now(cls, tz=None):
+                return day.astimezone(tz)
+
+        patches = [mock.patch.object(paper, "OUT", self.tmp.name),
+                   mock.patch.object(paper, "datetime", Clock),
+                   mock.patch.object(paper, "now_utc", return_value=day),
+                   mock.patch.object(paper, "sessions", return_value=calendar),
+                   mock.patch.object(paper, "compute_signal", return_value={
+                       "top": ["AAA", "BBB"], "prev_close": {"AAA": 100, "BBB": 100}, "bars_through": "2026-10-09"}),
+                   mock.patch.object(paper, "latest_prices", return_value={"AAA": 100, "BBB": 100})]
+        for patch in patches:
+            patch.start()
+            self.addCleanup(patch.stop)
+
+    def test_partial_buy_failure_retries_only_unaccepted_orders(self):
+        client = PaperOrders([{}, 403, {}])
+        with self.assertRaisesRegex(RuntimeError, "HTTP 403"):
+            paper.buy(client)
+        self.assertNotIn("bought", paper._state())
+        paper.buy(client)
+        self.assertEqual([o["symbol"] for o in client.sent], ["AAA", "BBB", "BBB"])
+        self.assertEqual(paper._state()["bought"], "2026-10-12")
+        paper.buy(client)
+        self.assertEqual(len(client.sent), 3)
+
+    def test_partial_sell_failure_does_not_mark_the_session_complete(self):
+        client = PaperOrders([{}, 403, {}], [{"symbol": "AAA", "qty": "10"}, {"symbol": "BBB", "qty": "20"}])
+        with mock.patch.object(paper, "now_utc", return_value=self.day.replace(hour=9, minute=10)):
+            with self.assertRaisesRegex(RuntimeError, "HTTP 403"):
+                paper.sell(client)
+            self.assertNotIn("sold", paper._state())
+            paper.sell(client)
+        self.assertEqual([o["symbol"] for o in client.sent], ["AAA", "BBB", "BBB"])
+        self.assertEqual(paper._state()["sold"], "2026-10-12")
+
+    def test_duplicate_is_success_only_for_the_same_accepted_order(self):
+        payload = {"symbol": "AAA", "qty": "10", "side": "buy", "type": "market",
+                   "time_in_force": "cls", "client_order_id": "onb-2026-10-12-AAA"}
+        for status, quantity, succeeds in [("accepted", "10", True), ("rejected", "10", False), ("accepted", "11", False)]:
+            with self.subTest(status=status, quantity=quantity):
+                client = PaperOrders([{"http": 422, "message": "client_order_id must be unique"}])
+                client.existing = {**payload, "id": "existing", "status": status, "qty": quantity}
+                if succeeds:
+                    paper._submit(client, "moc", payload, False)
+                else:
+                    with self.assertRaises(RuntimeError):
+                        paper._submit(client, "moc", payload, False)
+
+    def test_http_success_with_rejected_status_is_not_completion(self):
+        client = PaperOrders([{"status": "rejected"}])
+        with self.assertRaisesRegex(RuntimeError, "rejected"):
+            paper.buy(client)
+        self.assertNotIn("bought", paper._state())
+
+
+if __name__ == "__main__":
+    unittest.main()

trader.py

@@ -676,8 +676,10 @@ def load_state(market: str, strategy_name: str, book: str | None = None) -> dict
 def save_state(state: dict, market: str, strategy_name: str, book: str | None = None) -> None:
     path = _state_path(market, strategy_name, book)
     os.makedirs(os.path.dirname(path), exist_ok=True)
-    with open(path, "w") as f:
+    tmp = f"{path}.tmp"
+    with open(tmp, "w") as f:
         json.dump(state, f, indent=2, default=str)
+    os.replace(tmp, path)
 
 
 def init_state(market: str, strategy_name: str, capital: float) -> dict:
@@ -1830,12 +1832,44 @@ def refresh_panels(market: str, strategy_names: list[str], label: str) -> None:
         data_market = strategy_data_market(market, name)
         requests.setdefault(data_market, set()).update(
             price_request(market, list(tickers) + [benchmark]))
-        fields.setdefault(data_market, {"close", "open"}).update(strategy_fields(name))
+        fields.setdefault(data_market, {"close", "open", "volume"}).update(strategy_fields(name))
     for data_market, request in requests.items():
         print(f"[{label}] Refreshing {data_market} panel for {len(request)} tickers")
         data_manager.update(data_market, sorted(request), fields=sorted(fields[data_market]))
 
 
+def _rewind_state(state: dict, before: pd.Timestamp) -> None:
+    cutoff = str(before.date())
+    holdings = dict(state["holdings"])
+    cash = state["cash"]
+    # Trade values and commissions retain full precision; daily snapshots round cash.
+    for trade in reversed(state["trade_log"]):
+        if trade["date"] < cutoff:
+            continue
+        ticker, shares = trade["ticker"], trade["shares"]
+        if trade["action"] == "BUY":
+            holdings[ticker] = holdings.get(ticker, 0) - shares
+            cash += trade["value"] + trade["commission"]
+        else:
+            holdings[ticker] = holdings.get(ticker, 0) + shares
+            cash -= trade["value"] - trade["commission"]
+    state["holdings"] = {t: q for t, q in holdings.items() if abs(q) > 1e-8}
+    state["cash"] = cash
+    state["trade_log"] = [t for t in state["trade_log"] if t["date"] < cutoff]
+    state["daily_log"] = [d for d in state.get("daily_log", []) if d["date"] < cutoff]
+    state["daily_equity"] = {d: v for d, v in state["daily_equity"].items() if d < cutoff}
+    state["engine_changes"] = [e for e in state.get("engine_changes", []) if e["first_session"] < cutoff]
+    last = max(state["daily_equity"], default=None)
+    state["last_morning"] = state["last_evening"] = last
+    state["pending_trades"] = None
+    state["engine_version"] = state["daily_log"][-1].get("engine_version") if state["daily_log"] else None
+    if state["daily_log"] and not state["engine_version"]:
+        # The legacy ledger rounded trades, so its recorded snapshot is authoritative.
+        snapshot = state["daily_log"][-1]
+        state["cash"] = snapshot["cash"]
+        state["holdings"] = {t: h["shares"] for t, h in snapshot["holdings"].items()}
+
+
 def _process_sessions(args, state: dict | None, start: pd.Timestamp | None = None,
                       label: str = "auto", refresh: bool = True) -> dict | None:
     """Run every completed session not yet recorded in ``state`` (catch-up safe).
@@ -1876,6 +1910,16 @@ def _process_sessions(args, state: dict | None, start: pd.Timestamp | None = Non
     else:
         last = state.get("last_evening")
         first = book_start(getattr(args, "book", None), market)
+        if last and state.get("daily_equity"):
+            beginning = first if first is not None else pd.Timestamp(min(state["daily_equity"]))
+            recorded = pd.to_datetime(list(state["daily_equity"]))
+            gaps = sessions[(sessions >= beginning) & (sessions <= pd.Timestamp(last))
+                            & ~sessions.isin(recorded)]
+            if len(gaps):
+                print(f"[{label}] {strategy_name}: backfilled {gaps[0].date()}; replaying from that session")
+                _rewind_state(state, gaps[0])
+                last = state.get("last_evening")
+                first = beginning
         if last:
             todo = sessions[sessions > pd.Timestamp(last)]
         else:  # created by a morning run: start with the book's first session
@@ -1910,7 +1954,8 @@ def cmd_auto(args):
 
     Processes every completed session since the last recorded one, so missed
     days (monitor downtime, late data) are caught up in order. A session is
-    only processed after the exchange has closed.
+    only processed after the exchange has closed. Backfilled historical days
+    rewind the ledger and replay the affected sessions in order.
 
     Usage with cron (US market, run at 5pm ET weekdays):
         0 17 * * 1-5  cd /path/to/quant && uv run python trader.py auto --market us