From 0e1e0a202cafb57684ac3fcaabca2148acd4f9e1 Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Mon, 12 Oct 2026 05:16:19 -0700 Subject: [PATCH] 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 --- CLAUDE.md | 2 + agent_review.py | 14 +- research/lab/analyst_estimates.py | 348 ++++++++++++++++++++++++++++++ tests/test_analyst_estimates.py | 143 ++++++++++++ 4 files changed, 505 insertions(+), 2 deletions(-) create mode 100644 research/lab/analyst_estimates.py create mode 100644 tests/test_analyst_estimates.py diff --git a/CLAUDE.md b/CLAUDE.md index bbd8934..193f1d5 100644 --- a/CLAUDE.md +++ b/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_.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_.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 diff --git a/agent_review.py b/agent_review.py index d51bf14..69c49ea 100644 --- a/agent_review.py +++ b/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_.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") diff --git a/research/lab/analyst_estimates.py b/research/lab/analyst_estimates.py new file mode 100644 index 0000000..b89d9bf --- /dev/null +++ b/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_.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() diff --git a/tests/test_analyst_estimates.py b/tests/test_analyst_estimates.py new file mode 100644 index 0000000..94b0992 --- /dev/null +++ b/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()