"""Run Task 2 earnings-gap (2a) and SUE/PEAD (2b) research. Both experiments use the manifest-guarded research snapshot restricted to the production symbol set. Earnings data remain in the real earnings_events table on the production snapshot. This runner is research-only and never changes production configuration or integrates a signal/filter. """ from __future__ import annotations import argparse import asyncio import bisect import json import math import os import sqlite3 import sys from collections import defaultdict from concurrent.futures import ProcessPoolExecutor from datetime import date, datetime, timezone from pathlib import Path from types import SimpleNamespace from typing import Any from sqlalchemy import create_engine, select, text from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from app.ssl_bootstrap import bootstrap_ssl # noqa: E402 bootstrap_ssl() IRON_IC_BAR = 0.03 SUE_CARRY_DAYS = 63 SUE_TRAIL = 8 SUE_MIN_TRAIL = 4 COST_PER_SIDE = 0.001 MIN_PRE2021_DEPTH_PCT = 80.0 def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--snapshot", default="backtest_snapshots/research.sqlite") parser.add_argument( "--universe-snapshot", default="backtest_snapshots/prod.sqlite" ) parser.add_argument( "--earnings-snapshot", default="backtest_snapshots/prod.sqlite" ) parser.add_argument("--workers", type=int, default=6) parser.add_argument("--allow-spawn", action="store_true") parser.add_argument("--quiet", action="store_true") parser.add_argument("--stamp", default=None) return parser.parse_args() def _sqlite_url(path: Path) -> str: return f"sqlite+aiosqlite:///{path.resolve().as_posix()}" def _read_symbols(snapshot: Path) -> list[str]: engine = create_engine(f"sqlite:///{snapshot.resolve().as_posix()}", future=True) try: with engine.connect() as conn: return [ str(row[0]).upper().replace(".", "-") for row in conn.execute(text("SELECT symbol FROM tickers ORDER BY symbol")) ] finally: engine.dispose() def _snapshot_depth(snapshot: Path, symbols: set[str]) -> dict[str, Any]: from scripts.research_snapshot_manifest import assert_research_snapshot_complete manifest = assert_research_snapshot_complete(snapshot) connection = sqlite3.connect(snapshot) try: rows = connection.execute( """ SELECT t.symbol, MIN(o.date), MAX(o.date), COUNT(o.id) FROM tickers t LEFT JOIN ohlcv_records o ON o.ticker_id=t.id GROUP BY t.id, t.symbol """ ).fetchall() benchmark = connection.execute( "SELECT COUNT(*), MIN(date), MAX(date) FROM benchmark_prices " "WHERE symbol='SPY'" ).fetchone() finally: connection.close() by_symbol = {str(row[0]).upper(): row for row in rows} selected = [by_symbol[symbol] for symbol in sorted(symbols) if symbol in by_symbol] missing = sorted(symbols - set(by_symbol)) usable = [row for row in selected if int(row[3] or 0) > 0] zero_bar = sorted(str(row[0]) for row in selected if int(row[3] or 0) == 0) counts = sorted(int(row[3]) for row in usable) pre2021 = [row for row in usable if row[1] and str(row[1]) < "2021-01-01"] pre_pct = round(len(pre2021) / max(1, len(usable)) * 100.0, 1) shallow = [ { "symbol": str(row[0]), "first_bar": row[1], "last_bar": row[2], "bars": int(row[3] or 0), } for row in sorted(usable, key=lambda item: int(item[3] or 0)) if int(row[3] or 0) < 1000 ] gate_pass = ( len(usable) >= 505 and len(missing) + len(zero_bar) <= 1 and pre_pct >= MIN_PRE2021_DEPTH_PCT and int(benchmark[0] or 0) >= 1000 and benchmark[1] is not None and str(benchmark[1]) < "2021-01-01" ) return { "manifest": manifest, "requested_symbols": len(symbols), "tradable_symbols": len(usable), "missing_symbols": missing, "zero_bar_symbols": zero_bar, "bar_count": { "min": min(counts) if counts else 0, "median": counts[len(counts) // 2] if counts else 0, "max": max(counts) if counts else 0, }, "symbols_with_pre2021_bars": len(pre2021), "symbols_with_pre2021_bars_pct": pre_pct, "shallow_symbols_lt_1000_bars": shallow, "price_window": { "min": min(str(row[1]) for row in usable if row[1]), "max": max(str(row[2]) for row in usable if row[2]), }, "benchmark_spy": { "rows": int(benchmark[0] or 0), "min": benchmark[1], "max": benchmark[2], }, "gate_threshold": { "min_tradable_symbols": 505, "max_missing_or_zero_bar": 1, "min_symbols_with_pre2021_bars_pct": MIN_PRE2021_DEPTH_PCT, "benchmark_min_rows": 1000, "benchmark_must_begin_pre2021": True, }, "gate_pass": gate_pass, } def _load_backfill_status() -> dict[str, Any]: path = Path("reports/earnings-backfill-status.json") if not path.exists(): raise SystemExit(f"Missing backfill status: {path}") return json.loads(path.read_text(encoding="utf-8")) def _load_earnings(snapshot: Path, symbols: set[str]) -> list[dict[str, Any]]: engine = create_engine(f"sqlite:///{snapshot.resolve().as_posix()}", future=True) try: with engine.connect() as conn: tables = { str(row[0]) for row in conn.execute( text("SELECT name FROM sqlite_master WHERE type='table'") ) } if "earnings_events" not in tables: raise SystemExit("earnings_events table missing") event_columns = { str(row[1]) for row in conn.execute(text("PRAGMA table_info(earnings_events)")) } period_expression = ( "period_end_date" if "period_end_date" in event_columns else "NULL" ) rows = conn.execute( text( f""" SELECT symbol, announce_date, announce_time, eps_estimate, eps_actual, revenue_estimate, revenue_actual, source, {period_expression} AS period_end_date FROM earnings_events ORDER BY symbol, announce_date """ ) ).fetchall() finally: engine.dispose() result = [] for row in rows: symbol = str(row[0]).upper().replace(".", "-") if symbol not in symbols: continue result.append( { "symbol": symbol, "announce_date": date.fromisoformat(str(row[1])[:10]), "announce_time": str(row[2]).lower() if row[2] else None, "eps_estimate": row[3], "eps_actual": row[4], "revenue_estimate": row[5], "revenue_actual": row[6], "source": row[7], "period_end_date": ( date.fromisoformat(str(row[8])[:10]) if row[8] else None ), } ) return result def _load_surprise_history( snapshot: Path, symbols: set[str] ) -> dict[str, list[dict[str, Any]]]: engine = create_engine(f"sqlite:///{snapshot.resolve().as_posix()}", future=True) try: with engine.connect() as conn: tables = { str(row[0]) for row in conn.execute( text("SELECT name FROM sqlite_master WHERE type='table'") ) } if "earnings_surprise_history" not in tables: return {} rows = conn.execute( text( """ SELECT symbol, period_end_date, eps_estimate, eps_actual FROM earnings_surprise_history ORDER BY symbol, period_end_date """ ) ).fetchall() finally: engine.dispose() result: dict[str, list[dict[str, Any]]] = defaultdict(list) for row in rows: symbol = str(row[0]).upper().replace(".", "-") if symbol not in symbols: continue result[symbol].append( { "period_end_date": date.fromisoformat(str(row[1])[:10]), "eps_estimate": row[2], "eps_actual": row[3], } ) return dict(result) def _pct(numerator: int, denominator: int) -> float: return round(numerator / max(1, denominator) * 100.0, 1) def _data_quality( events: list[dict[str, Any]], symbols: set[str], *, window_start: date, window_end: date, backfill_status: dict[str, Any], ) -> dict[str, Any]: in_window = [ event for event in events if window_start <= event["announce_date"] <= window_end ] by_symbol: dict[str, list[dict[str, Any]]] = defaultdict(list) for event in in_window: by_symbol[event["symbol"]].append(event) paired = [ event for event in in_window if event.get("eps_estimate") is not None and event.get("eps_actual") is not None ] paired_by_symbol: dict[str, int] = defaultdict(int) for event in paired: paired_by_symbol[event["symbol"]] += 1 ge8 = sum(len(by_symbol.get(symbol, [])) >= 8 for symbol in symbols) ge8_paired = sum(paired_by_symbol.get(symbol, 0) >= 8 for symbol in symbols) recognised_sessions = {"bmo", "amc", "during"} session_known = sum( str(event.get("announce_time") or "").lower() in recognised_sessions for event in in_window ) session_pct = _pct(session_known, len(in_window)) session_reliable = session_pct >= 80.0 far_off = [] annual_rates: list[float] = [] for symbol in sorted(symbols): symbol_events = sorted( by_symbol.get(symbol, []), key=lambda event: event["announce_date"] ) if not symbol_events: far_off.append( {"symbol": symbol, "events": 0, "events_per_year": 0.0} ) continue first = symbol_events[0]["announce_date"] last = symbol_events[-1]["announce_date"] active_years = max(1.0, (last - first).days / 365.25) rate = len(symbol_events) / active_years annual_rates.append(rate) if rate < 2.0 or rate > 6.0: far_off.append( { "symbol": symbol, "events": len(symbol_events), "first": first.isoformat(), "last": last.isoformat(), "events_per_year": round(rate, 2), } ) keys = [(event["symbol"], event["announce_date"]) for event in in_window] duplicate_rows_in_table = len(keys) - len(set(keys)) paired_pct = _pct(len(paired), len(in_window)) ge8_paired_pct = _pct(ge8_paired, len(symbols)) fallback_needed = paired_pct < 50.0 or ge8_paired_pct < 50.0 return { "window": {"from": window_start.isoformat(), "to": window_end.isoformat()}, "prod_symbols": len(symbols), "events": len(in_window), "symbols_with_any_event": len(by_symbol), "symbols_with_ge8_announcements": ge8, "symbols_with_ge8_announcements_pct": _pct(ge8, len(symbols)), "symbols_with_ge8_paired_announcements": ge8_paired, "symbols_with_ge8_paired_announcements_pct": ge8_paired_pct, "events_with_actual_and_estimate": len(paired), "events_with_actual_and_estimate_pct": paired_pct, "duplicate_rows_in_table": duplicate_rows_in_table, "duplicate_rows_fetched": backfill_status.get( "duplicate_rows_logged_total", 0 ), "restated_rows_fetched": backfill_status.get( "restated_rows_logged_total", 0 ), "dedupe_policy": backfill_status.get("dedupe_policy"), "events_per_symbol_year": { "mean_active_span_rate": ( round(sum(annual_rates) / len(annual_rates), 2) if annual_rates else None ), "expected": "approximately 4", "far_off_rule": "active-span rate <2 or >6, plus zero-event symbols", "far_off_count": len(far_off), "far_off_symbols": far_off, }, "announcement_session": { "recognised_bmo_amc_or_during": session_known, "recognised_pct": session_pct, "reliable": session_reliable, "assessment": ( "usable" if session_reliable else "missing/unreliable; do not use same-day availability" ), }, "point_in_time_policy": "announce_date_plus_1_trading_day_for_all_events", "sue_scaling": { "primary": "eps_surprise_over_stdev_of_prior_8_surprises_min_4", "fallback_trigger": ( "paired event coverage <50% or symbols with >=8 paired events <50%" ), "fallback_needed": fallback_needed, "fallback_name": ( "eps_surprise_over_price" if fallback_needed else "not_used" ), }, "backfill": backfill_status, } def _percentile(values: list[float], quantile: float) -> float | None: if not values: return None ordered = sorted(values) if len(ordered) == 1: return ordered[0] location = quantile * (len(ordered) - 1) lower = int(math.floor(location)) upper = int(math.ceil(location)) if lower == upper: return ordered[lower] weight = location - lower return ordered[lower] * (1.0 - weight) + ordered[upper] * weight def _r_dist(values: list[float]) -> dict[str, Any]: if not values: return {"count": 0} return { "count": len(values), "mean_r": round(sum(values) / len(values), 4), "median_r": round(float(_percentile(values, 0.50)), 4), "win_rate": round(sum(value > 0 for value in values) / len(values), 4), "p05_r": round(float(_percentile(values, 0.05)), 4), "p95_r": round(float(_percentile(values, 0.95)), 4), "min_r": round(min(values), 4), "max_r": round(max(values), 4), } def _first_session_after(announcement: date, calendar: list[date]) -> date | None: index = bisect.bisect_right(calendar, announcement) return calendar[index] if index < len(calendar) else None def _entry_in_last_three_sessions( entry: date, announcement: date, calendar: list[date] ) -> bool: index = bisect.bisect_left(calendar, announcement) return entry in calendar[max(0, index - 3) : index] def _analyse_2a_trades( details: list[dict[str, Any]], events: list[dict[str, Any]], calendar: list[date], *, cost_per_side: float, ) -> dict[str, Any]: by_symbol: dict[str, list[date]] = defaultdict(list) for event in events: by_symbol[event["symbol"]].append(event["announce_date"]) for dates in by_symbol.values(): dates.sort() parsed = [] for trade in details: symbol = str(trade.get("symbol") or "").upper() entry_raw = trade.get("entry_date") exit_raw = trade.get("exit_date") raw_r = trade.get("r") if entry_raw is None or exit_raw is None or raw_r is None: continue entry_date = date.fromisoformat(str(entry_raw)[:10]) exit_date = date.fromisoformat(str(exit_raw)[:10]) entry_price = float(trade.get("entry") or 0.0) initial_stop = float(trade.get("initial_stop") or 0.0) exit_fill = float(trade.get("fill") or 0.0) risk = entry_price - initial_stop if risk <= 0: continue net_r = float(raw_r) - cost_per_side * (entry_price + exit_fill) / risk announcements = by_symbol.get(symbol, []) strict_hold = [ event_date for event_date in announcements if entry_date < event_date < exit_date ] pre_entry = any( _entry_in_last_three_sessions(entry_date, event_date, calendar) for event_date in announcements ) is_stop = str(trade.get("reason") or "") in {"stop", "trailing_stop"} stop_after = is_stop and any( _first_session_after(event_date, calendar) == exit_date for event_date in announcements ) parsed.append( { "symbol": symbol, "entry": entry_date, "exit": exit_date, "net_r": net_r, "earnings_strictly_in_hold": bool(strict_hold), "pre_earnings_entry": pre_entry, "is_stop": is_stop, "stop_within_1d_after_earnings": stop_after, } ) losses = [trade for trade in parsed if trade["net_r"] <= -1.0] losses_with = [trade for trade in losses if trade["earnings_strictly_in_hold"]] all_with = [trade for trade in parsed if trade["earnings_strictly_in_hold"]] pre = [trade["net_r"] for trade in parsed if trade["pre_earnings_entry"]] other = [trade["net_r"] for trade in parsed if not trade["pre_earnings_entry"]] stops = [trade for trade in parsed if trade["is_stop"]] stops_after = [ trade["net_r"] for trade in stops if trade["stop_within_1d_after_earnings"] ] other_stops = [ trade["net_r"] for trade in stops if not trade["stop_within_1d_after_earnings"] ] all_other_exits = [ trade["net_r"] for trade in parsed if not trade["stop_within_1d_after_earnings"] ] pre_dist = _r_dist(pre) other_dist = _r_dist(other) left_delta = None right_delta = None if pre and other: left_delta = round(pre_dist["p05_r"] - other_dist["p05_r"], 4) right_delta = round(pre_dist["p95_r"] - other_dist["p95_r"], 4) tail_condition = ( left_delta is not None and left_delta < 0 and right_delta is not None and right_delta <= 0 ) return { "verdict": "INFORMATIONAL", "costs": {"per_side": cost_per_side, "r_is_net_of_round_trip_costs": True}, "closed_trades": len(parsed), "q1_loss_concentration": { "loss_definition": "realized_net_R <= -1.0", "holding_period_definition": "announcement strictly after entry and before exit", "losses_count": len(losses), "losses_with_announcement_count": len(losses_with), "losses_with_announcement_fraction": ( round(len(losses_with) / len(losses), 4) if losses else None ), "all_trades_with_announcement_count": len(all_with), "all_trades_with_announcement_fraction": ( round(len(all_with) / len(parsed), 4) if parsed else None ), }, "q2_entries_within_3_trading_days_before_announcement": { "pre_earnings": pre_dist, "all_other_entries": other_dist, "tail_deltas_pre_minus_other": { "p05_r": left_delta, "p95_r": right_delta, }, "directional_tail_condition_present": tail_condition, "tail_read": ( "Directional left-worse/right-not-better condition is present; " "materiality and any filter design require separate human approval." if tail_condition else "Registered directional tail condition is not present." ), }, "q3_stop_exits_within_1_trading_day_after_announcement": { "stops_after_earnings": _r_dist(stops_after), "all_other_stops": _r_dist(other_stops), "all_other_exits": _r_dist(all_other_exits), }, "implementation": "REPORT_ONLY_NO_FILTER_ARM_NO_FILTER_CHANGE", } async def _run_2a( snapshot: Path, events: list[dict[str, Any]], symbols: set[str], *, analysis_start: date, analysis_end: date, workers: int, quiet: bool, ) -> dict[str, Any]: from app.config import settings from app.models.ticker import Ticker from app.services import backtest_service as bt from app.services.admin_service import get_activation_config from app.services.benchmark_service import load_benchmark_closes from app.services.paper_trade_service import get_exit_policy from app.services.recommendation_service import get_recommendation_config os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1" settings.backtest_workers = workers engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True) session_factory = async_sessionmaker( engine, class_=AsyncSession, expire_on_commit=False ) try: async with session_factory() as db: config = await get_recommendation_config(db) activation = await get_activation_config(db) exit_config = await get_exit_policy(db) tickers = list( ( await db.execute( select(Ticker) .where(Ticker.symbol.in_(sorted(symbols))) .order_by(Ticker.symbol) ) ).scalars() ) spy = await load_benchmark_closes(db, "SPY") prices: dict[str, tuple] = {} replay_inputs: list[tuple[str, tuple]] = [] for index, ticker in enumerate(tickers): if not quiet and index % 50 == 0: print(f" 2a load {index}/{len(tickers)}", flush=True) columns = await bt._fetch_columns(db, ticker.symbol) if columns is None: continue prices[ticker.symbol] = columns replay_inputs.append((ticker.symbol, columns)) finally: await engine.dispose() candidates: list[dict] = [] process_count = bt._backtest_worker_count() context = bt._mp_context() if process_count > 1 else None if context is not None: loop = asyncio.get_running_loop() chunk_size = process_count * 2 with ProcessPoolExecutor( max_workers=process_count, mp_context=context ) as pool: for start in range(0, len(replay_inputs), chunk_size): batch = replay_inputs[start : start + chunk_size] futures = [ loop.run_in_executor( pool, bt._replay_candidates_for_period, symbol, columns, config, activation, spy, analysis_start, "weekly", True, False, ) for symbol, columns in batch ] for ticker_candidates in await asyncio.gather(*futures): candidates.extend(ticker_candidates) if not quiet: print( f" 2a replay {min(start + len(batch), len(replay_inputs))}/" f"{len(replay_inputs)} workers={process_count}", flush=True, ) else: for index, (symbol, columns) in enumerate(replay_inputs): if not quiet and index % 25 == 0: print(f" 2a replay {index}/{len(replay_inputs)}", flush=True) ticker_candidates = bt._replay_candidates_for_period( symbol, columns, config, activation, spy, analysis_start, "weekly", True, False, ) candidates.extend(ticker_candidates) bt._assign_momentum_percentiles(candidates) bt._assign_residual_momentum_percentiles(candidates) bt._assign_low_volatility_percentiles(candidates) bt._assign_activation_momentum_percentiles(candidates) bt._assign_residual_high_vol_blend(candidates) cutoff = float(activation.get("min_momentum_percentile", 80.0)) for candidate in candidates: candidate["qualified"] = bt._momentum_qualifies(candidate, cutoff) longs = [ candidate for candidate in candidates if candidate.get("qualified") and candidate.get("direction") == "long" ] strategy = next( row for row in bt.PORTFOLIO_MONITOR_STRATEGIES if row.get("is_production") ) entry_config = bt._entry_variant_config(str(strategy["entry_variant"])) if entry_config is None: raise RuntimeError("Production entry configuration missing") ranking_key = str( entry_config.get("ranking_key") or entry_config["percentile_key"] ) exit_policy = bt.LIVE_EXIT_MODE_TO_SIM.get( str(exit_config.get("mode", "atr_trailing")), "atr_trail3" ) hold_days = int(exit_config.get("hold_days", 30)) trail = float(exit_config.get("atr_multiplier", bt.ATR_TRAIL_MULTIPLIER)) reentry = bt._make_gate_reset_reentry_fn( longs, prices, cadence="weekly", ranking_key=ranking_key ) simulation = bt._simulate_portfolio( longs, prices, spy, exit_policy, hold_days, ranking_key=ranking_key, max_positions=int(entry_config["max_positions"]), risk_per_trade=float(entry_config["risk_per_trade"]), atr_trail_multiplier=trail, post_stop_reentry_fn=reentry, fill_mode=bt.FILL_MODE_CLOSE, cost_per_side=COST_PER_SIDE, include_trades=True, ) if simulation is None: raise RuntimeError("Production-config simulation returned no result") calendar = sorted( { date.fromordinal(int(ordinal)) for columns in prices.values() for ordinal in columns[0] } ) all_trade_details = simulation.get("trade_details") or [] eligible_trade_details = [] for trade in all_trade_details: entry_raw = trade.get("entry_date") exit_raw = trade.get("exit_date") if entry_raw is None or exit_raw is None: continue entry_date = date.fromisoformat(str(entry_raw)[:10]) exit_date = date.fromisoformat(str(exit_raw)[:10]) if analysis_start <= entry_date and exit_date <= analysis_end: eligible_trade_details.append(trade) analysis = _analyse_2a_trades( eligible_trade_details, events, calendar, cost_per_side=COST_PER_SIDE, ) analysis["analysis_window"] = { "from": analysis_start.isoformat(), "to": analysis_end.isoformat(), "rule": "entry_on_or_after_start_and_exit_on_or_before_end", "simulation_trades_total": len(all_trade_details), "trades_excluded_outside_earnings_coverage": ( len(all_trade_details) - len(eligible_trade_details) ), } analysis["run_config"] = { "universe_symbols": len(tickers), "fill_mode": "close", "cost_per_side": COST_PER_SIDE, "momentum_cutoff": cutoff, "exit_policy": exit_policy, "hold_days": hold_days, "max_positions": int(entry_config["max_positions"]), "risk_per_trade": float(entry_config["risk_per_trade"]), } analysis["sim_summary"] = { key: simulation.get(key) for key in ( "start_date", "end_date", "trades", "sharpe", "cagr_pct", "max_drawdown_pct", "total_return_pct", ) } return analysis def _build_sue_series( events_by_symbol: dict[str, list[dict[str, Any]]], prices: dict[str, tuple], *, use_price_fallback: bool, surprise_history_by_symbol: dict[str, list[dict[str, Any]]] | None = None, ) -> tuple[dict[str, dict[date, float]], dict[str, int]]: result: dict[str, dict[date, float]] = {} standard_values = 0 fallback_values = 0 history_scaled_values = 0 dropped_insufficient_history = 0 dropped_missing_period_alignment = 0 dropped_zero_stdev = 0 for symbol, columns in prices.items(): dates = [date.fromordinal(int(value)) for value in columns[0]] closes = [float(value) for value in columns[4]] if not dates: continue surprises = [] for event in events_by_symbol.get(symbol.upper(), []): actual = event.get("eps_actual") estimate = event.get("eps_estimate") if actual is None or estimate is None: continue surprises.append( { "announce_date": event["announce_date"], "period_end_date": event.get("period_end_date"), "surprise": float(actual) - float(estimate), } ) surprises.sort(key=lambda item: item["announce_date"]) history = [] if surprise_history_by_symbol is not None: for row in surprise_history_by_symbol.get(symbol.upper(), []): actual = row.get("eps_actual") estimate = row.get("eps_estimate") if actual is None or estimate is None: continue history.append( ( row["period_end_date"], float(actual) - float(estimate), ) ) history.sort(key=lambda item: item[0]) live: dict[date, float] = {} for index, event in enumerate(surprises): announcement = event["announce_date"] surprise = float(event["surprise"]) period_end = event.get("period_end_date") if surprise_history_by_symbol is not None: if period_end is None: dropped_missing_period_alignment += 1 continue trailing = [ value for history_period, value in history if history_period < period_end ][-SUE_TRAIL:] else: trailing = [ float(prior["surprise"]) for prior in surprises[max(0, index - SUE_TRAIL) : index] ] sue = None if len(trailing) >= SUE_MIN_TRAIL: mean = sum(trailing) / len(trailing) variance = sum((value - mean) ** 2 for value in trailing) / ( len(trailing) - 1 ) stdev = math.sqrt(variance) if variance > 0 else 0.0 if stdev > 1e-12: sue = surprise / stdev standard_values += 1 if surprise_history_by_symbol is not None: history_scaled_values += 1 else: dropped_zero_stdev += 1 else: dropped_insufficient_history += 1 if sue is None and use_price_fallback: price_index = bisect.bisect_right(dates, announcement) - 1 if price_index >= 0 and closes[price_index] > 0: sue = surprise / closes[price_index] fallback_values += 1 if sue is None or not math.isfinite(sue): continue start_index = bisect.bisect_right(dates, announcement) if start_index >= len(dates): continue end_index = min(len(dates), start_index + SUE_CARRY_DAYS) for trading_index in range(start_index, end_index): live[dates[trading_index]] = float(sue) if live: result[symbol.upper()] = live return result, { "standard_scaled_events": standard_values, "events_scaled_from_period_history": history_scaled_values, "price_fallback_events": fallback_values, "dropped_insufficient_trailing_history": dropped_insufficient_history, "dropped_missing_period_alignment": dropped_missing_period_alignment, "dropped_zero_stdev": dropped_zero_stdev, } def _find_signal(rows: list[dict[str, Any]], signal: str) -> dict[str, Any] | None: return next((row for row in rows if row.get("signal") == signal), None) def _mechanical_sue_grade( sue_row: dict[str, Any] | None, pre_row: dict[str, Any] | None, post_row: dict[str, Any] | None, ) -> tuple[bool, bool]: sign_stable = bool( pre_row and post_row and float(pre_row.get("mean_ic", 0.0)) > 0 and float(post_row.get("mean_ic", 0.0)) > 0 ) passed = bool( sue_row and float(sue_row.get("mean_ic", -999.0)) >= IRON_IC_BAR and bool(sue_row.get("reliable")) and sign_stable ) return passed, sign_stable def _conditional_ic(bt, sue_weeks: dict) -> dict[str, Any]: usable = [ week for week, records in sue_weeks.items() if len(records) >= bt.MIN_CROSS_SECTION ] kept = bt._nonoverlapping_weeks( usable, max(1, round(bt.HORIZON / 5)) ) values = [] sizes = [] for week in kept: records = [ record for record in sue_weeks[week] if record.get("mom_12_1") is not None ] if len(records) < bt.MIN_CROSS_SECTION: continue ordered = sorted(records, key=lambda record: float(record["mom_12_1"])) top = ordered[-max(1, len(ordered) // 5) :] if len(top) < 5: continue ic = bt._spearman( [float(record["val"]) for record in top], [float(record["fwd"]) for record in top], ) if ic is not None: values.append(float(ic)) sizes.append(len(top)) if not values: return {"mean_ic": None, "weeks": 0, "avg_cross_section": None} mean = sum(values) / len(values) if len(values) > 1: stdev = math.sqrt( sum((value - mean) ** 2 for value in values) / (len(values) - 1) ) t_stat = mean / stdev * math.sqrt(len(values)) if stdev > 0 else None else: t_stat = None return { "mean_ic": round(mean, 4), "ic_t_stat": round(t_stat, 2) if t_stat is not None else None, "weeks": len(values), "avg_cross_section": round(sum(sizes) / len(sizes), 1), "population": "top_mom_12_1_quintile_only", } async def _run_2b( snapshot: Path, events: list[dict[str, Any]], surprise_history: dict[str, list[dict[str, Any]]], symbols: set[str], *, quality: dict[str, Any], workers: int, quiet: bool, ) -> dict[str, Any]: from app.config import settings from app.models.ticker import Ticker from app.services import backtest_service as bt from app.services.benchmark_service import load_benchmark_closes os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1" os.environ["BACKTEST_SIGNAL_EVAL_ONLY"] = "1" settings.backtest_workers = workers engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True) session_factory = async_sessionmaker( engine, class_=AsyncSession, expire_on_commit=False ) prices: dict[str, tuple] = {} try: async with session_factory() as db: tickers = list( ( await db.execute( select(Ticker) .where(Ticker.symbol.in_(sorted(symbols))) .order_by(Ticker.symbol) ) ).scalars() ) spy = await load_benchmark_closes(db, "SPY") for index, ticker in enumerate(tickers): if not quiet and index % 25 == 0: print(f" 2b signals {index}/{len(tickers)}", flush=True) columns = await bt._fetch_columns(db, ticker.symbol) if columns is not None: prices[ticker.symbol] = columns finally: await engine.dispose() events_by_symbol: dict[str, list[dict[str, Any]]] = defaultdict(list) for event in events: events_by_symbol[event["symbol"]].append(event) fallback = bool(quality["sue_scaling"]["fallback_needed"]) sue_map, scaling_counts = _build_sue_series( events_by_symbol, prices, use_price_fallback=fallback, surprise_history_by_symbol=surprise_history, ) sue_all: dict = defaultdict(list) identical: dict = defaultdict(lambda: defaultdict(list)) for symbol, columns in prices.items(): dates = [date.fromordinal(int(value)) for value in columns[0]] highs = [float(value) for value in columns[2]] closes = [float(value) for value in columns[4]] volumes = [float(value) for value in columns[5]] records = [ SimpleNamespace( date=dates[index], close=closes[index], high=highs[index], volume=volumes[index], ) for index in range(len(dates)) ] live_sue = sue_map.get(symbol.upper(), {}) if not live_sue: continue for index in bt._weekly_asof_indices(records): forward_index = index + bt.HORIZON if forward_index >= len(records) or closes[index] <= 0: continue as_of = dates[index] sue_value = live_sue.get(as_of) if sue_value is None: continue forward = closes[forward_index] / closes[index] - 1.0 signal_values = bt._signal_values( dates, closes, highs, index, spy ) momentum = signal_values.get("mom_12_1") residual = signal_values.get("mom_12_1_resid") iso = as_of.isocalendar() week = (iso.year, iso.week) sue_record = { "val": float(sue_value), "fwd": float(forward), "symbol": symbol, "mom_12_1": momentum, } sue_all[week].append(sue_record) if momentum is not None and residual is not None: identical["sue_latest"][week].append(sue_record) identical["mom_12_1"][week].append( {"val": float(momentum), "fwd": forward, "symbol": symbol} ) identical["mom_12_1_resid"][week].append( {"val": float(residual), "fwd": forward, "symbol": symbol} ) full_eval = bt._signal_evaluation({"sue_latest": sue_all}) identical_eval = bt._signal_evaluation(identical) pre_eval = bt._signal_evaluation( {"sue_latest": {week: rows for week, rows in sue_all.items() if week[0] < 2021}} ) post_eval = bt._signal_evaluation( {"sue_latest": {week: rows for week, rows in sue_all.items() if week[0] >= 2021}} ) sue_row = _find_signal(full_eval, "sue_latest") pre_row = _find_signal(pre_eval, "sue_latest") post_row = _find_signal(post_eval, "sue_latest") pass_grade, sign_stable = _mechanical_sue_grade(sue_row, pre_row, post_row) verdict = "PASS" if pass_grade else "FAIL" verdict_detail = ( "SUE candidate confirmed - book-integration design (tilt vs second gate) " "is PENDING_HUMAN. Do not integrate anything yourself." if pass_grade else "SUE DEAD for this stack" ) weekly_sizes = [len(rows) for rows in sue_all.values()] average_weekly_n = ( round(sum(weekly_sizes) / len(weekly_sizes), 1) if weekly_sizes else 0.0 ) average_scored_n = sue_row.get("avg_cross_section") if sue_row else None thin = average_scored_n is None or float(average_scored_n) < 100.0 side_by_side = { signal: _find_signal(identical_eval, signal) for signal in ("sue_latest", "mom_12_1", "mom_12_1_resid") } return { "verdict": verdict, "verdict_detail": verdict_detail, "grade_rule": { "mean_ic_ge_0_03_positive": ( bool(sue_row and float(sue_row.get("mean_ic", -999.0)) >= IRON_IC_BAR) ), "reliable_ge_12_windows": bool(sue_row and sue_row.get("reliable")), "positive_sign_pre_and_post_2021": sign_stable, "pass": pass_grade, }, "sue_unconditional": sue_row, "era_split": {"pre_2021": pre_row, "post_2021": post_row}, "signal_eval_identical_cross_sections": side_by_side, "identical_cross_section_definition": ( "same week-symbol-forward-return cells where sue_latest, mom_12_1, " "and mom_12_1_resid are all non-null" ), "momentum_conditional_top_quintile": _conditional_ic(bt, sue_all), "coverage": { "symbols_with_live_sue": len(sue_map), "avg_weekly_live_n_all_weeks": average_weekly_n, "avg_cross_section_n_scored_nonoverlap": average_scored_n, "thin_cross_section_lt_100": thin, "warning": ( "THIN CROSS-SECTION: fewer than 100 live SUE names per scored week." if thin else None ), }, "scaling": { "method": quality["sue_scaling"]["primary"], "fallback": quality["sue_scaling"]["fallback_name"], "counts": scaling_counts, "pre_coverage_history_policy": ( "period-end EPS surprises may scale later events but are never " "treated as live signals without an announcement date" ), "availability": "announce_date_plus_1_trading_day", "carry_trading_days": SUE_CARRY_DAYS, }, "universe_symbols": len(prices), } def _format(value: Any) -> str: if value is None: return "-" if isinstance(value, bool): return "true" if value else "false" return str(value) def _quality_markdown(quality: dict[str, Any], depth: dict[str, Any]) -> str: annual = quality["events_per_symbol_year"] session = quality["announcement_session"] backfill = quality["backfill"] source = backfill.get("source") or {} lines = [ "### Data quality gate", "", ( f"Approved earnings window: {quality['window']['from']} to " f"{quality['window']['to']}. Source mode: {backfill.get('mode')}." ), "", "| check | result |", "|---|---:|", f"| Prod symbols requested / tradable | {depth['requested_symbols']} / {depth['tradable_symbols']} |", f"| Manifest complete + live counts match | {depth['manifest'].get('complete')} |", f"| Prod symbols with pre-2021 bars | {depth['symbols_with_pre2021_bars']} ({depth['symbols_with_pre2021_bars_pct']}%) |", f"| SPY benchmark depth | {depth['benchmark_spy']['rows']} rows, {depth['benchmark_spy']['min']} to {depth['benchmark_spy']['max']} |", f"| Snapshot depth gate | {depth['gate_pass']} |", f"| Bulk source windows / requests logged | {backfill.get('bulk_windows_done')}/{backfill.get('bulk_windows_total')} / {backfill.get('bulk_requests_logged_total')} |", f"| Source repository / pinned commit | {source.get('repository')} @ {source.get('commit')} |", f"| Source license / upstream provider documented | {source.get('license')} / {source.get('upstream_provider_documented')} |", f"| Existing-source conflicts preserved | {backfill.get('conflicting_existing_rows')} rows / {backfill.get('conflicting_existing_fields')} fields |", f"| Symbols with >=8 announcements | {quality['symbols_with_ge8_announcements']} ({quality['symbols_with_ge8_announcements_pct']}%) |", f"| Symbols with >=8 paired announcements | {quality['symbols_with_ge8_paired_announcements']} ({quality['symbols_with_ge8_paired_announcements_pct']}%) |", f"| Events with estimate + actual | {quality['events_with_actual_and_estimate']}/{quality['events']} ({quality['events_with_actual_and_estimate_pct']}%) |", f"| Duplicate rows in keyed table | {quality['duplicate_rows_in_table']} |", f"| Duplicate / restated payload rows fetched | {quality['duplicate_rows_fetched']} / {quality['restated_rows_fetched']} |", f"| Mean announcements per active symbol-year | {annual['mean_active_span_rate']} (expected about 4) |", f"| Symbols far off (<2 or >6/year, incl. zero) | {annual['far_off_count']} |", f"| Recognised BMO/AMC/during | {session['recognised_pct']}% (reliable={session['reliable']}) |", f"| Point-in-time policy | {quality['point_in_time_policy']} |", f"| SUE price fallback | {quality['sue_scaling']['fallback_name']} |", "", f"Deduplication: {quality['dedupe_policy']}", "", "Far-off announcement-rate symbols: " + (", ".join(row["symbol"] for row in annual["far_off_symbols"]) or "none"), ] return "\n".join(lines) def _dist_markdown(label: str, row: dict[str, Any]) -> str: return ( f"| {label} | {_format(row.get('count'))} | {_format(row.get('mean_r'))} | " f"{_format(row.get('median_r'))} | {_format(row.get('win_rate'))} | " f"{_format(row.get('p05_r'))} | {_format(row.get('p95_r'))} |" ) def _two_a_markdown(result: dict[str, Any]) -> str: q1 = result["q1_loss_concentration"] q2 = result["q2_entries_within_3_trading_days_before_announcement"] q3 = result["q3_stop_exits_within_1_trading_day_after_announcement"] window = result["analysis_window"] lines = [ "### Experiment 2a - earnings-gap risk diagnostic", "", "Verdict: **INFORMATIONAL**. Report-only; no filter arm or implementation.", "", ( f"Trade cohort is restricted to the approved earnings-coverage window " f"{window['from']} to {window['to']}; " f"{window['trades_excluded_outside_earnings_coverage']} simulated trades " "outside that window were excluded." ), "", "| cohort | count | fraction |", "|---|---:|---:|", f"| Realized net R <= -1.0 | {q1['losses_count']} | - |", f"| Losses with announcement strictly inside hold | {q1['losses_with_announcement_count']} | {q1['losses_with_announcement_fraction']} |", f"| All trades with announcement strictly inside hold | {q1['all_trades_with_announcement_count']} | {q1['all_trades_with_announcement_fraction']} |", "", "| Entry cohort | count | mean R | median R | win rate | p05 R | p95 R |", "|---|---:|---:|---:|---:|---:|---:|", _dist_markdown("Within 3 sessions before earnings", q2["pre_earnings"]), _dist_markdown("All other entries", q2["all_other_entries"]), "", f"Tail deltas (pre minus other): p05={q2['tail_deltas_pre_minus_other']['p05_r']}, p95={q2['tail_deltas_pre_minus_other']['p95_r']}.", "", q2["tail_read"], "", "| Exit cohort | count | mean R | median R | win rate | p05 R | p95 R |", "|---|---:|---:|---:|---:|---:|---:|", _dist_markdown("Stops within 1 session after earnings", q3["stops_after_earnings"]), _dist_markdown("All other stops", q3["all_other_stops"]), _dist_markdown("All other exits", q3["all_other_exits"]), ] return "\n".join(lines) def _ic_row(label: str, row: dict[str, Any] | None) -> str: row = row or {} return ( f"| {label} | {_format(row.get('mean_ic'))} | {_format(row.get('ic_t_stat'))} | " f"{_format(row.get('weeks'))} | {_format(row.get('avg_cross_section'))} | " f"{_format(row.get('ic_positive_pct'))} | {_format(row.get('reliable'))} |" ) def _two_b_markdown(result: dict[str, Any]) -> str: side = result["signal_eval_identical_cross_sections"] era = result["era_split"] coverage = result["coverage"] conditional = result["momentum_conditional_top_quintile"] lines = [ "### Experiment 2b - SUE / post-earnings drift", "", f"Mechanical verdict: **{result['verdict']}** - {result['verdict_detail']}", "", "Identical cross-sections:", "", "| signal | mean IC | t | windows | avg N | IC positive % | reliable |", "|---|---:|---:|---:|---:|---:|---|", _ic_row("sue_latest", side.get("sue_latest")), _ic_row("mom_12_1", side.get("mom_12_1")), _ic_row("mom_12_1_resid", side.get("mom_12_1_resid")), "", "Unconditional SUE grade row:", "", "| signal | mean IC | t | windows | avg N | IC positive % | reliable |", "|---|---:|---:|---:|---:|---:|---|", _ic_row("sue_latest", result.get("sue_unconditional")), "", "Era stability:", "", "| era | mean IC | t | windows | avg N | IC positive % | reliable |", "|---|---:|---:|---:|---:|---:|---|", _ic_row("pre-2021", era.get("pre_2021")), _ic_row("post-2021", era.get("post_2021")), "", f"Coverage: {coverage['symbols_with_live_sue']} symbols with live SUE; avg weekly N={coverage['avg_weekly_live_n_all_weeks']}; scored non-overlap avg N={coverage['avg_cross_section_n_scored_nonoverlap']}.", "", coverage.get("warning") or "Cross-section is not flagged thin at the registered <100-name read.", "", f"Momentum-conditional top-quintile SUE: mean IC={conditional.get('mean_ic')}, t={conditional.get('ic_t_stat')}, windows={conditional.get('weeks')}, avg N={conditional.get('avg_cross_section')}.", ] return "\n".join(lines) def _write_reports( *, stamp: str, generated_at: str, snapshot: Path, depth: dict[str, Any], quality: dict[str, Any], result_2a: dict[str, Any], result_2b: dict[str, Any], ) -> tuple[Path, Path]: reports = Path("reports") reports.mkdir(parents=True, exist_ok=True) path_2a = reports / f"earnings-2a-gap-{stamp}.json" path_2b = reports / f"earnings-2b-sue-{stamp}.json" common = { "generated_at": generated_at, "snapshot": str(snapshot.resolve()), "snapshot_depth": depth, "data_quality": quality, "production_impact": "none", } payload_2a = {**common, "experiment": "2a", "result": result_2a} payload_2b = {**common, "experiment": "2b", "result": result_2b} path_2a.write_text( json.dumps(payload_2a, indent=2, default=str) + "\n", encoding="utf-8" ) path_2b.write_text( json.dumps(payload_2b, indent=2, default=str) + "\n", encoding="utf-8" ) path_2a.with_suffix(".md").write_text( "# Earnings Task 2a - gap diagnostic\n\n" + _quality_markdown(quality, depth) + "\n\n" + _two_a_markdown(result_2a) + "\n", encoding="utf-8", ) path_2b.with_suffix(".md").write_text( "# Earnings Task 2b - SUE / PEAD\n\n" + _quality_markdown(quality, depth) + "\n\n" + _two_b_markdown(result_2b) + "\n", encoding="utf-8", ) return path_2a, path_2b def _update_research_doc( *, depth: dict[str, Any], quality: dict[str, Any], result_2a: dict[str, Any], result_2b: dict[str, Any], path_2a: Path, path_2b: Path, ) -> None: path = Path("docs/research/earnings-gap-and-sue.md") existing = path.read_text(encoding="utf-8") if path.exists() else "# Earnings gap and SUE" marker = "## Results" index = existing.find(marker) preregistration = existing[:index].rstrip() if index >= 0 else existing.rstrip() final_status = ( "Task 2 CLOSED (SUE PASS→PENDING_HUMAN)" if result_2b["verdict"] == "PASS" else "Task 2 CLOSED (SUE DEAD)" ) body = [ preregistration, "", "## Results", "", _quality_markdown(quality, depth), "", _two_a_markdown(result_2a), "", _two_b_markdown(result_2b), "", "## Artifacts", "", f"- `{path_2a.as_posix()}` and companion Markdown", f"- `{path_2b.as_posix()}` and companion Markdown", "- `reports/earnings-backfill-status.json`", "", "Production changes: **none**. No earnings filter or SUE integration was implemented.", "", f"## Final status: **{final_status}**", "", ] path.write_text("\n".join(body), encoding="utf-8") async def _main() -> None: args = _parse_args() snapshot = Path(args.snapshot) universe_snapshot = Path(args.universe_snapshot) earnings_snapshot = Path(args.earnings_snapshot) for path in (snapshot, universe_snapshot, earnings_snapshot): if not path.exists(): raise SystemExit(f"Missing snapshot: {path}") if args.allow_spawn: os.environ["BACKTEST_ALLOW_SPAWN"] = "1" requested_symbols = set(_read_symbols(universe_snapshot)) depth = _snapshot_depth(snapshot, requested_symbols) print( "Snapshot guard:", json.dumps( { "manifest_complete": depth["manifest"].get("complete"), "tradable": depth["tradable_symbols"], "pre2021_pct": depth["symbols_with_pre2021_bars_pct"], "benchmark": depth["benchmark_spy"], "gate_pass": depth["gate_pass"], }, default=str, ), ) if not depth["gate_pass"]: raise SystemExit( "Snapshot depth gate failed. Repair the production-universe depth and " "SPY benchmark before running either experiment." ) tradable_symbols = requested_symbols - set(depth["missing_symbols"]) - set( depth["zero_bar_symbols"] ) backfill_status = _load_backfill_status() price_start = date.fromisoformat(depth["price_window"]["min"]) price_end = date.fromisoformat(depth["price_window"]["max"]) backfill_window = backfill_status.get("window") or {} coverage_start = ( date.fromisoformat(backfill_window["from"]) if backfill_window.get("from") else None ) coverage_end = ( date.fromisoformat(backfill_window["to"]) if backfill_window.get("to") else None ) approved_shorter_window = bool( backfill_status.get("mode") == "dolthub_public_bulk_clone" and (backfill_status.get("coverage_amendment") or {}).get( "approved_by_user" ) ) backfill_covers_approved_window = bool( coverage_start and coverage_end and coverage_end >= price_end and (coverage_start <= price_start or approved_shorter_window) ) allowed_modes = {"fmp_bulk_date_range_only", "dolthub_public_bulk_clone"} if ( backfill_status.get("mode") not in allowed_modes or not backfill_status.get("complete") or not backfill_covers_approved_window ): raise SystemExit( "Bulk earnings backfill is incomplete or does not cover its approved " "research window through the price snapshot end." ) if coverage_start is None or coverage_end is None: raise SystemExit("Backfill status is missing its approved coverage window.") analysis_start = max(price_start, coverage_start) analysis_end = min(price_end, coverage_end) events = [ event for event in _load_earnings(earnings_snapshot, tradable_symbols) if analysis_start <= event["announce_date"] <= analysis_end ] surprise_history = _load_surprise_history( earnings_snapshot, tradable_symbols ) quality = _data_quality( events, tradable_symbols, window_start=analysis_start, window_end=analysis_end, backfill_status=backfill_status, ) print( "Earnings quality:", json.dumps( { "events": quality["events"], "symbols_ge8_pct": quality["symbols_with_ge8_announcements_pct"], "paired_pct": quality["events_with_actual_and_estimate_pct"], "session": quality["announcement_session"], "fallback": quality["sue_scaling"]["fallback_name"], } ), ) stamp = args.stamp or datetime.now().strftime("%Y%m%d-%H%M%S") generated_at = datetime.now(timezone.utc).isoformat() reports = Path("reports") reports.mkdir(parents=True, exist_ok=True) print("Running Experiment 2a...") result_2a = await _run_2a( snapshot, events, tradable_symbols, analysis_start=analysis_start, analysis_end=analysis_end, workers=args.workers, quiet=args.quiet, ) checkpoint_2a = reports / f"earnings-2a-gap-{stamp}.checkpoint.json" checkpoint_2a.write_text( json.dumps( { "generated_at": generated_at, "snapshot_depth": depth, "data_quality": quality, "result": result_2a, }, indent=2, default=str, ) + "\n", encoding="utf-8", ) print(f"Wrote 2a checkpoint: {checkpoint_2a}", flush=True) print("Running Experiment 2b...") result_2b = await _run_2b( snapshot, events, surprise_history, tradable_symbols, quality=quality, workers=args.workers, quiet=args.quiet, ) checkpoint_2b = reports / f"earnings-2b-sue-{stamp}.checkpoint.json" checkpoint_2b.write_text( json.dumps( { "generated_at": generated_at, "snapshot_depth": depth, "data_quality": quality, "result": result_2b, }, indent=2, default=str, ) + "\n", encoding="utf-8", ) print(f"Wrote 2b checkpoint: {checkpoint_2b}", flush=True) path_2a, path_2b = _write_reports( stamp=stamp, generated_at=generated_at, snapshot=snapshot, depth=depth, quality=quality, result_2a=result_2a, result_2b=result_2b, ) _update_research_doc( depth=depth, quality=quality, result_2a=result_2a, result_2b=result_2b, path_2a=path_2a, path_2b=path_2b, ) print(f"2a verdict: {result_2a['verdict']}") print(f"2b verdict: {result_2b['verdict']} - {result_2b['verdict_detail']}") print(f"Wrote {path_2a} and {path_2b} (+ Markdown companions)") if __name__ == "__main__": asyncio.run(_main())