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")

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()