Gahow Wang
0e1e0a202c
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>
144 lines
7.8 KiB
Python
144 lines
7.8 KiB
Python
"""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()
|