"""Does total-return momentum change any admission or ordering decision? Step 2 of docs/data-integrity-plan.md. A screening test, deliberately cheap: if ranking on total return picks the same names in the same order as ranking on price return, nothing downstream can change and the production `tr_factor` work (Step 3) is unnecessary. Why this needs no schema and no ingestion change: the forward-anchored factor design exists because *production* ingest is incremental, so a back-adjusted series drifts between fetches. Here each symbol's whole history is fetched in one request, and within a single response the adjustment is internally consistent. Method ------ * One `Adjustment.SPLIT` and one `Adjustment.ALL` request per symbol plus SPY, cached to sqlite so re-runs are free and reproducible. Aligned by date, never zipped by index; a date present in only one adjustment is dropped and counted. * At every SPY session with enough history, residual 12-1 momentum is computed twice with the *same* production function: the price-return series against nominal SPY, and the total-return series against total-return SPY. Only bars at or before the decision date are used. * The volatility leg is computed once per date from the SPLIT series and held fixed in both arms, so the only thing that varies is momentum. * Both arms are percentiled with the production `_percentiles`, gated at the production `min_momentum_percentile`, then ordered by `blend_strategy_rank` with the symbol as a deterministic tie-break. The full admitted list is compared - never truncated to `max_positions`, because a lower-ranked name can be the first actionable setup when others fail gates or are already held. Decision rule (pre-registered, docs/data-integrity-plan.md) ---------------------------------------------------------- * no ordered-list difference on any evaluated date -> stop, Step 3 unnecessary * any ordered-list difference -> Step 3 investigates trades * incomplete downloads or no evaluable dates -> inconclusive, fix coverage Usage:: python scripts/research_tr_rank_agreement.py --fetch # populate the cache python scripts/research_tr_rank_agreement.py --time-batch # project the runtime python scripts/research_tr_rank_agreement.py # full screen """ from __future__ import annotations import argparse import asyncio import json import sqlite3 import sys import time from datetime import date, datetime, time as dtime, timedelta, timezone from pathlib import Path ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) DEFAULT_SNAPSHOT = ROOT / "backtest_snapshots" / "survivorship-a.sqlite" DEFAULT_CACHE = ROOT / "backtest_snapshots" / "tr-rank-cache.sqlite" BENCHMARK = "SPY" # --------------------------------------------------------------------------- # cache # --------------------------------------------------------------------------- def open_cache(path: Path) -> sqlite3.Connection: con = sqlite3.connect(path) con.execute( "CREATE TABLE IF NOT EXISTS bars (" " symbol TEXT NOT NULL, date TEXT NOT NULL," " split_close REAL, all_close REAL," " PRIMARY KEY (symbol, date))" ) con.execute( "CREATE TABLE IF NOT EXISTS fetched (" " symbol TEXT PRIMARY KEY, fetched_at TEXT NOT NULL," " split_bars INTEGER, all_bars INTEGER, aligned_bars INTEGER," " dropped_unmatched INTEGER)" ) con.commit() return con async def fetch_pair(symbol: str, start: date, end: date) -> tuple[dict, dict]: """(split_closes, all_closes) keyed by date, each from a single request.""" from alpaca.data.enums import Adjustment from alpaca.data.historical import StockHistoricalDataClient from alpaca.data.requests import StockBarsRequest from alpaca.data.timeframe import TimeFrame from app.config import settings if not settings.alpaca_api_key or not settings.alpaca_api_secret: raise SystemExit("Alpaca credentials are not configured.") client = StockHistoricalDataClient(settings.alpaca_api_key, settings.alpaca_api_secret) alpaca_symbol = symbol.replace("-", ".") window = ( datetime.combine(start, dtime.min, tzinfo=timezone.utc), datetime.combine(end + timedelta(days=1), dtime.min, tzinfo=timezone.utc), ) async def one(adjustment) -> dict[date, float]: request = StockBarsRequest( symbol_or_symbols=alpaca_symbol, timeframe=TimeFrame.Day, start=window[0], end=window[1], adjustment=adjustment, ) last: Exception | None = None for attempt in range(4): try: bars = await asyncio.to_thread(client.get_stock_bars, request) return { b.timestamp.date(): float(b.close) for b in bars.data.get(alpaca_symbol, []) } except Exception as exc: # noqa: BLE001 - retry any transport failure last = exc if attempt < 3: await asyncio.sleep(2 * (attempt + 1)) raise SystemExit(f"Alpaca fetch failed for {symbol}: {last}") return await one(Adjustment.SPLIT), await one(Adjustment.ALL) async def populate_cache(symbols: list[str], start: date, end: date, cache: Path) -> dict: con = open_cache(cache) done = {r[0] for r in con.execute("SELECT symbol FROM fetched")} todo = [s for s in symbols if s not in done] print(f"cache {cache.name}: {len(done)} cached, {len(todo)} to fetch") coverage: dict[str, dict] = {} for index, symbol in enumerate(todo, 1): split, adjusted = await fetch_pair(symbol, start, end) # Align by date. A date in only one adjustment cannot be compared and is # dropped rather than paired with its neighbour. common = sorted(set(split) & set(adjusted)) dropped = len(set(split) ^ set(adjusted)) con.executemany( "INSERT OR REPLACE INTO bars (symbol, date, split_close, all_close) VALUES (?,?,?,?)", [(symbol, d.isoformat(), split[d], adjusted[d]) for d in common], ) con.execute( "INSERT OR REPLACE INTO fetched VALUES (?,?,?,?,?,?)", (symbol, datetime.now(timezone.utc).isoformat(), len(split), len(adjusted), len(common), dropped), ) con.commit() if dropped: print(f" {symbol:<6} aligned={len(common)} dropped_unmatched={dropped}") if index % 50 == 0: print(f" … {index}/{len(todo)}") for row in con.execute("SELECT symbol, split_bars, all_bars, aligned_bars, dropped_unmatched FROM fetched"): coverage[row[0]] = { "split_bars": row[1], "all_bars": row[2], "aligned_bars": row[3], "dropped_unmatched": row[4], } con.close() return coverage def load_series(cache: Path, symbols: list[str]) -> dict[str, tuple[list[date], list[float], list[float]]]: con = open_cache(cache) out: dict[str, tuple[list[date], list[float], list[float]]] = {} for symbol in symbols: rows = con.execute( "SELECT date, split_close, all_close FROM bars WHERE symbol = ? ORDER BY date", (symbol,), ).fetchall() if not rows: continue out[symbol] = ( [date.fromisoformat(r[0]) for r in rows], [float(r[1]) for r in rows], [float(r[2]) for r in rows], ) con.close() return out # --------------------------------------------------------------------------- # screen # --------------------------------------------------------------------------- def snapshot_symbols(snapshot: Path) -> tuple[list[str], list[str], date, date]: """(symbols with bars, symbols with none, window start, window end). A ticker row with no bars in the snapshot cannot rank in either arm, so it is inert rather than missing. Keeping the two apart is what stops a failed download from being quietly reported as agreement. """ con = sqlite3.connect(snapshot) try: rows = con.execute( "SELECT t.symbol, count(o.id) FROM tickers t " "LEFT JOIN ohlcv_records o ON o.ticker_id = t.id GROUP BY t.symbol" ).fetchall() lo, hi = con.execute("SELECT min(date), max(date) FROM ohlcv_records").fetchone() finally: con.close() with_bars = sorted(r[0].upper() for r in rows if r[1]) without_bars = sorted(r[0].upper() for r in rows if not r[1]) return with_bars, without_bars, date.fromisoformat(lo), date.fromisoformat(hi) def ordered_list( momentum: dict[str, float], vol_percentiles: dict[str, float], cutoff: float, ) -> list[str]: """Admitted set at the production cutoff, ordered by the production blend. Symbols are fed to ``_percentiles`` in sorted order in both arms so that ties resolve identically and cannot manufacture a difference. """ from app.services.momentum_service import _percentiles, blend_strategy_rank percentiles = _percentiles({s: momentum[s] for s in sorted(momentum)}) admitted = [s for s, p in percentiles.items() if p >= cutoff] scored = { s: blend_strategy_rank(percentiles[s], vol_percentiles.get(s)) for s in admitted } return sorted(admitted, key=lambda s: (-(scored[s] or 0.0), s)) def mechanism_check( series: dict[str, tuple[list[date], list[float], list[float]]], sample_dates: list[date], ) -> dict: """Is the rank shift explained by dividend yield? For each sampled decision date, correlate a name's trailing 12-month implied yield — read straight off the ALL/SPLIT ratio, which is what separates the two series — against how far its momentum percentile moves between the arms. A strong positive rank correlation means the difference *is* the dividend, not numerical noise. """ from app.services.momentum_service import _MOM_LOOKBACK, _percentiles, compute_residual_12_1_momentum bench_dates, bench_split, bench_all = series[BENCHMARK] bench_price = dict(zip(bench_dates, bench_split)) bench_tr = dict(zip(bench_dates, bench_all)) symbols = sorted(s for s in series if s != BENCHMARK) index_of = {s: {d: i for i, d in enumerate(series[s][0])} for s in symbols} correlations: list[float] = [] used_dates: list[str] = [] for as_of in sample_dates: price_mom: dict[str, float] = {} tr_mom: dict[str, float] = {} yields: dict[str, float] = {} for symbol in symbols: idx = index_of[symbol].get(as_of) if idx is None or idx < _MOM_LOOKBACK: continue dates, split_closes, all_closes = series[symbol] wd, ws, wa = dates[: idx + 1], split_closes[: idx + 1], all_closes[: idx + 1] p = compute_residual_12_1_momentum(wd, ws, bench_price) t = compute_residual_12_1_momentum(wd, wa, bench_tr) if p is None or t is None: continue price_mom[symbol] = p tr_mom[symbol] = t # Ratio drift over the formation window is the accumulated dividend. r_now = wa[-1] / ws[-1] if ws[-1] else None r_then = wa[-_MOM_LOOKBACK] / ws[-_MOM_LOOKBACK] if ws[-_MOM_LOOKBACK] else None if r_now and r_then: yields[symbol] = r_now / r_then - 1.0 common = sorted(set(price_mom) & set(yields)) if len(common) < 30: continue pp = _percentiles({s: price_mom[s] for s in common}) tp = _percentiles({s: tr_mom[s] for s in common}) shift = {s: tp[s] - pp[s] for s in common} correlations.append(_spearman([yields[s] for s in common], [shift[s] for s in common])) used_dates.append(as_of.isoformat()) return { "what_it_measures": ( "Spearman correlation, per sampled date, between a name's trailing 12-month " "implied yield (from the ALL/SPLIT ratio over the formation window) and the " "change in its momentum percentile between the two arms." ), "caveat": ( "Both the stock series and the benchmark leg change between arms, so this is " "strong evidence consistent with a dividend effect - it does not isolate the " "stock's own yield from the beta-weighted benchmark yield." ), "formation_window_bars": _MOM_LOOKBACK, "min_symbols_per_date": 30, "sampled_dates": len(correlations), "sample_date_list": used_dates, "mean_spearman_yield_vs_percentile_shift": ( round(sum(correlations) / len(correlations), 4) if correlations else None ), "min": round(min(correlations), 4) if correlations else None, "max": round(max(correlations), 4) if correlations else None, "per_date_spearman": [round(c, 4) for c in correlations], } def _spearman(x: list[float], y: list[float]) -> float: def rank(v: list[float]) -> list[float]: order = sorted(range(len(v)), key=lambda i: v[i]) out = [0.0] * len(v) for position, i in enumerate(order): out[i] = float(position) return out rx, ry = rank(x), rank(y) n = len(rx) mx, my = sum(rx) / n, sum(ry) / n num = sum((rx[i] - mx) * (ry[i] - my) for i in range(n)) dx = sum((rx[i] - mx) ** 2 for i in range(n)) ** 0.5 dy = sum((ry[i] - my) ** 2 for i in range(n)) ** 0.5 return num / (dx * dy) if dx and dy else 0.0 def disagreement(a: list[str], b: list[str]) -> float: sa, sb = set(a), set(b) union = sa | sb if not union: return 0.0 return len(sa ^ sb) / len(union) def run_screen( series: dict[str, tuple[list[date], list[float], list[float]]], cutoff: float, *, limit_dates: int | None = None, verbose: bool = True, ) -> dict: from app.services.momentum_service import ( _MOM_LOOKBACK, compute_realized_vol_6m, compute_residual_12_1_momentum, _percentiles, ) bench_dates, bench_split, bench_all = series[BENCHMARK] bench_price = dict(zip(bench_dates, bench_split)) bench_tr = dict(zip(bench_dates, bench_all)) symbols = sorted(s for s in series if s != BENCHMARK) index_of: dict[str, dict[date, int]] = { s: {d: i for i, d in enumerate(series[s][0])} for s in symbols } evaluated = 0 skipped_no_history = 0 changed_sets = 0 changed_orders = 0 disagreements: list[float] = [] # Counted directly rather than derived from the Jaccard ratio: a swap moves # two names (one out, one in), so the symmetric difference is twice the # replacement count and the two are easy to conflate. replacements: list[int] = [] only_price_counts: list[int] = [] only_tr_counts: list[int] = [] admitted_sizes: list[int] = [] per_year: dict[int, dict[str, int]] = {} examples: list[dict] = [] decision_dates = [d for d in bench_dates] if limit_dates: decision_dates = decision_dates[-limit_dates:] for as_of in decision_dates: price_mom: dict[str, float] = {} tr_mom: dict[str, float] = {} vol_values: dict[str, float] = {} for symbol in symbols: idx = index_of[symbol].get(as_of) if idx is None or idx < _MOM_LOOKBACK: continue dates, split_closes, all_closes = series[symbol] window_dates = dates[: idx + 1] window_split = split_closes[: idx + 1] window_all = all_closes[: idx + 1] price_value = compute_residual_12_1_momentum(window_dates, window_split, bench_price) tr_value = compute_residual_12_1_momentum(window_dates, window_all, bench_tr) # Equal eligibility: a name enters both arms or neither. if price_value is None or tr_value is None: continue price_mom[symbol] = price_value tr_mom[symbol] = tr_value vol = compute_realized_vol_6m(window_split) if vol is not None: vol_values[symbol] = vol if not price_mom: skipped_no_history += 1 continue evaluated += 1 vol_percentiles = _percentiles({s: vol_values[s] for s in sorted(vol_values)}) price_list = ordered_list(price_mom, vol_percentiles, cutoff) tr_list = ordered_list(tr_mom, vol_percentiles, cutoff) year = per_year.setdefault(as_of.year, {"dates": 0, "changed_sets": 0, "changed_orders": 0}) year["dates"] += 1 set_price, set_tr = set(price_list), set(tr_list) only_price = len(set_price - set_tr) only_tr = len(set_tr - set_price) only_price_counts.append(only_price) only_tr_counts.append(only_tr) admitted_sizes.append(len(price_list)) # A replacement is one name leaving paired with one arriving; when the # two arms admit different counts, the surplus is not a replacement. replacements.append(min(only_price, only_tr)) same_set = set_price == set_tr same_order = price_list == tr_list if not same_set: changed_sets += 1 year["changed_sets"] += 1 if not same_order: changed_orders += 1 year["changed_orders"] += 1 if len(examples) < 10: only_price = [s for s in price_list if s not in set(tr_list)] only_tr = [s for s in tr_list if s not in set(price_list)] examples.append({ "date": as_of.isoformat(), "admitted_price_return": len(price_list), "admitted_total_return": len(tr_list), "only_in_price_return": only_price[:10], "only_in_total_return": only_tr[:10], "order_changed_without_membership_change": same_set, "first_10_price_return": price_list[:10], "first_10_total_return": tr_list[:10], }) disagreements.append(disagreement(price_list, tr_list)) if verbose and evaluated % 100 == 0: print(f" … {evaluated} dates evaluated, {changed_orders} with a different ordered list") return { "evaluated_dates": evaluated, "skipped_dates_no_history": skipped_no_history, "dates_with_different_admitted_set": changed_sets, "dates_with_different_ordered_list": changed_orders, "mean_set_disagreement": (sum(disagreements) / len(disagreements)) if disagreements else 0.0, "max_set_disagreement": max(disagreements) if disagreements else 0.0, "mean_admitted_per_arm": round(sum(admitted_sizes) / len(admitted_sizes), 2) if admitted_sizes else 0.0, "mean_names_only_in_price_return": round(sum(only_price_counts) / len(only_price_counts), 2) if only_price_counts else 0.0, "mean_names_only_in_total_return": round(sum(only_tr_counts) / len(only_tr_counts), 2) if only_tr_counts else 0.0, "mean_replacements_per_date": round(sum(replacements) / len(replacements), 2) if replacements else 0.0, "max_replacements_per_date": max(replacements) if replacements else 0, "mean_symmetric_difference": round( (sum(only_price_counts) + sum(only_tr_counts)) / len(only_price_counts), 2 ) if only_price_counts else 0.0, "per_year": { str(y): v for y, v in sorted(per_year.items()) }, "examples": examples, } # --------------------------------------------------------------------------- # driver # --------------------------------------------------------------------------- async def main(args: argparse.Namespace) -> int: snapshot = Path(args.snapshot) cache = Path(args.cache) symbols, inert, window_start, window_end = snapshot_symbols(snapshot) universe = symbols + [BENCHMARK] print(f"snapshot : {snapshot.name} ({len(symbols)} symbols with bars" + (f", {len(inert)} inert: {', '.join(inert)}" if inert else "") + ")") print(f"window : {window_start} .. {window_end} (clipped to the snapshot)") if args.fetch: await populate_cache(universe, window_start, window_end, cache) return 0 series = load_series(cache, universe) missing = [s for s in universe if s not in series] if missing: raise SystemExit( f"{len(missing)} symbols have bars in the snapshot but none in the cache " f"({', '.join(missing[:8])}). Run with --fetch first; an incomplete download " "makes the screen inconclusive and must never be read as agreement." ) from app.services.admin_service import ACTIVATION_DEFAULTS cutoff = float(ACTIVATION_DEFAULTS["min_momentum_percentile"]) print(f"cutoff : {cutoff} (production min_momentum_percentile)") if args.mechanism: bench_dates = load_series(cache, [BENCHMARK])[BENCHMARK][0] sample = bench_dates[::args.mechanism] print(f"mechanism check over {len(sample)} sampled dates ...") payload = { "generated_at": datetime.now(timezone.utc).isoformat(), "snapshot": str(snapshot), "cache": str(cache), "window": {"start": window_start.isoformat(), "end": window_end.isoformat()}, "universe_size": len(symbols), "sample_every_n_sessions": args.mechanism, "result": mechanism_check(series, sample), } out = Path(args.mechanism_out) out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") r = payload["result"] print(f" sampled dates : {r['sampled_dates']}") print(f" mean Spearman : {r['mean_spearman_yield_vs_percentile_shift']} " f"(min {r['min']}, max {r['max']})") print(f" report : {out}") return 0 if args.time_batch: n = args.time_batch started = time.perf_counter() result = run_screen(series, cutoff, limit_dates=n, verbose=False) elapsed = time.perf_counter() - started total_dates = len(series[BENCHMARK][0]) per_date = elapsed / max(result["evaluated_dates"], 1) print(f"\ntimed {result['evaluated_dates']} evaluated dates in {elapsed:.1f}s " f"({per_date:.2f}s/date)") print(f"projected full run over {total_dates} sessions: " f"{per_date * total_dates / 60:.1f} min") return 0 started = time.perf_counter() result = run_screen(series, cutoff) result["elapsed_seconds"] = round(time.perf_counter() - started, 1) con = open_cache(cache) coverage = { r[0]: {"split_bars": r[1], "all_bars": r[2], "aligned_bars": r[3], "dropped_unmatched": r[4]} for r in con.execute("SELECT symbol, split_bars, all_bars, aligned_bars, dropped_unmatched FROM fetched") } con.close() verdict = ( "stop - no admission or ordering difference in this window" if result["dates_with_different_ordered_list"] == 0 else "proceed to Step 3 - ordered lists differ" ) payload = { "generated_at": datetime.now(timezone.utc).isoformat(), "snapshot": str(snapshot), "cache": str(cache), "window": {"start": window_start.isoformat(), "end": window_end.isoformat()}, "universe_size": len(symbols), "inert_snapshot_symbols": inert, "cutoff": cutoff, "verdict": verdict, "result": result, "coverage": { "symbols_cached": len(coverage), "symbols_with_unmatched_dates": sum(1 for c in coverage.values() if c["dropped_unmatched"]), "total_unmatched_dates": sum(c["dropped_unmatched"] for c in coverage.values()), }, } out = Path(args.out) out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") print(f"\nevaluated dates : {result['evaluated_dates']}") print(f"dates with different admitted set: {result['dates_with_different_admitted_set']}") print(f"dates with different ordered list: {result['dates_with_different_ordered_list']}") print(f"mean / max set disagreement : {result['mean_set_disagreement']:.4f} / {result['max_set_disagreement']:.4f}") print(f"verdict : {verdict}") print(f"report : {out}") return 0 def parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) p.add_argument("--snapshot", default=str(DEFAULT_SNAPSHOT), help="Universe and date bounds come from this snapshot.") p.add_argument("--cache", default=str(DEFAULT_CACHE)) p.add_argument("--fetch", action="store_true", help="Populate the cache and exit.") p.add_argument("--time-batch", type=int, default=None, metavar="N", help="Time the screen over the last N sessions and project the full runtime.") p.add_argument("--mechanism", type=int, default=None, metavar="EVERY_N", help="Correlate trailing dividend yield against the percentile shift on every Nth session.") p.add_argument("--out", default="reports/tr-rank-agreement.json") p.add_argument("--mechanism-out", default="reports/tr-rank-agreement-mechanism.json") return p.parse_args() if __name__ == "__main__": raise SystemExit(asyncio.run(main(parse_args())))