dennisthiessenandClaude Fable 5
c7c60a64f2
Earnings backfill sourced from the public DoltHub earnings repo at a pinned commit rather than the FMP API: reproducible for anyone re-running the study, and it burns no request quota. 12,414 events, 98.6% of symbols with >=8 announcements, 99.2% paired actual/estimate, no keyed duplicates. 2a earnings-gap diagnostic: INFORMATIONAL, no filter shipped. The pre-earnings cohort's right tail was better, so the registered avoid-earnings condition failed. Note the raw 23/266 vs 115/574 incidence gap is largely a duration confound -- severe losses stop out fast and have less time to span an announcement -- so it is not evidence that holding through earnings is safe. 2b SUE: FAIL against the pre-registered +0.03 bar (unconditional IC +0.0151 over 56 reliable windows, momentum-conditional +0.0213). Signs stable across eras, so this is a clean null rather than an ambiguous one, consistent with post-earnings drift having decayed in large caps. Closes the Tier-1 arc: Task 1 dead on deep evidence, Task 2 dead here, Task 3 complete as diagnostic. No in-sample research thread remains open. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1550 lines
58 KiB
Python
1550 lines
58 KiB
Python
"""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())
|