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