dennisthiessen/signal-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'research/fip-breadth-ic' — park Phase B fip breadth

Browse Source 
Brings env-gated liquid-breadth harness hooks, research tooling, compact
evidence, and the completion-manifest race guard. No production behavior
change when liquid env vars are unset. Nothing to deploy.
This commit is contained in:
dennisthiessen
2026-07-19 00:32:48 +02:00
parent c2c7244d1a 2311999e57
commit 8f285acb00
12 changed files with 3004 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+2 -2
View File
@@ -263,7 +263,7 @@ A systematic single-variable sweep (offline prod snapshot, production gate/rank/
Two findings future sessions must not re-litigate: Two findings future sessions must not re-litigate:
- **The "inverse-vol sizing win" (July 2026) was mis-attributed — do not resurrect.** The diagnostic sized `notional = equity × 1% / vol_6m`, and the 20% notional cap bound on 95% of entries, so it actually measured "~5 positions × 20% notional each" — a concentration/risk-appetite bump economically equivalent to raising risk to 1.5%, not vol-managed sizing. Genuine inverse-vol sizing (risk budget × median-vol/vol) cuts max drawdown to −18.2% but costs ~58pp total return at flat Sharpe: a risk-preference trade, not edge. - **The "inverse-vol sizing win" (July 2026) was mis-attributed — do not resurrect.** The diagnostic sized `notional = equity × 1% / vol_6m`, and the 20% notional cap bound on 95% of entries, so it actually measured "~5 positions × 20% notional each" — a concentration/risk-appetite bump economically equivalent to raising risk to 1.5%, not vol-managed sizing. Genuine inverse-vol sizing (risk budget × median-vol/vol) cuts max drawdown to −18.2% but costs ~58pp total return at flat Sharpe: a risk-preference trade, not edge.
- **`fip_id` — Da/Gurun/Warachka information discreteness over the 12-1 formation window — is the strongest cross-sectional signal measured on this universe: IC −0.045, t = −2.91, correct sign (continuous-information winners outperform).** It clears the iron-rule bar in isolation but does not improve this book (the momentum gate already captures the effect in-sample). It is the prime ranking/gate candidate **if the universe broadens** (e.g. `nasdaq_all`). - **`fip_id` — Da/Gurun/Warachka information discreteness over the 12-1 formation window — is the strongest cross-sectional signal on the *production* universe: IC −0.045, t = −2.91, correct sign (continuous-information winners outperform).** It clears the iron-rule bar in isolation but does not improve this book (the momentum gate already captures the effect in-sample). **Phase B (liquid-1500, research branch only):** unconditional fip fails iron rule (−0.017 / t −1.85); mom-conditional fip (−0.088 / t −4.58) is a *book-tilt candidate only* after a baseline breadth mom book is proven. Do **not** cite the orphaned 21:14 row (+0.0575) — it raced a partial `research.sqlite`. See `docs/research/fip-breadth-ic.md`.
### The iron rule for strategy changes ### The iron rule for strategy changes
@@ -281,7 +281,7 @@ Corollaries: never let an unvalidated score gate setups; the outcome evaluator m
1. **Forward monitor the promoted strategy** — the production UI now behaves like a portfolio monitor for the current strategy, with selectable lookbacks and SPY comparison. Forward paper-trade months are the only evidence the snapshot cannot provide; the July 2026 tuning pass closed every in-sample lead. (Trailing-stop sensitivity and the max-15 capacity check are done — see the tuning table above.) 1. **Forward monitor the promoted strategy** — the production UI now behaves like a portfolio monitor for the current strategy, with selectable lookbacks and SPY comparison. Forward paper-trade months are the only evidence the snapshot cannot provide; the July 2026 tuning pass closed every in-sample lead. (Trailing-stop sensitivity and the max-15 capacity check are done — see the tuning table above.)
2. **Signal context snapshots** — accumulate point-in-time composite/sentiment/fundamental context for every new setup so the discretionary overlay can be tested forward-only. 2. **Signal context snapshots** — accumulate point-in-time composite/sentiment/fundamental context for every new setup so the discretionary overlay can be tested forward-only.
3. **More breadth, not more history** — widening the ranked universe (e.g. `nasdaq_all`) strengthens each week's cross-section and the IC t-stat, even if only the top slice is traded. Now doubly motivated: it is also where the strong `fip_id` signal (see tuning findings) could become tradeable. (Deeper history was considered and declined.) 3. **Breadth is no longer free leverage** — Phase B found residual-mom t-stat *fell* on liquid-1500 vs the 505-name fingerprint (0.055/1.98 → 0.029/1.33). Any breadth book must clear a pre-registered baseline arm before fip tilts mean anything. (Deeper history was considered and declined.)
## Key Use Cases ## Key Use Cases
app/services/backtest_service.py
+337 -74
View File
@@ -30,6 +30,11 @@ Environment variables (see also run_backtest_snapshot.py):
BACKTEST_FALLBACK_CLEAR_AIR_ONLY=1 BACKTEST_FALLBACK_CLEAR_AIR_ONLY=1
BACKTEST_RESEARCH_EXITS=1 BACKTEST_RESEARCH_EXITS=1
BACKTEST_MIN_RR_SWEEP=1 BACKTEST_MIN_RR_SWEEP=1
Broad-universe signal research (local snapshots only; inert when unset):
BACKTEST_LIQUID_BREADTH=1500 # PIT top-N by 63d median $vol, price floor
BACKTEST_LIQUID_MIN_PRICE=5 # USD close floor at as-of (default 5)
BACKTEST_SIGNAL_EVAL_ONLY=1 # skip portfolio_sim / monitor (signal IC only)
""" """
from __future__ import annotations from __future__ import annotations
@@ -876,20 +881,86 @@ def _signal_values(
return out return out
def _liquid_breadth_top_n() -> int:
"""0 = off (production path). N > 0 enables PIT top-N $vol mask for signal IC."""
raw = os.getenv("BACKTEST_LIQUID_BREADTH", "").strip()
if not raw:
return 0
try:
return max(0, int(raw))
except ValueError:
return 0
def _liquid_min_price() -> float:
raw = os.getenv("BACKTEST_LIQUID_MIN_PRICE", "5").strip() or "5"
try:
return max(0.0, float(raw))
except ValueError:
return 5.0
def _signal_eval_only() -> bool:
return os.getenv("BACKTEST_SIGNAL_EVAL_ONLY", "").strip() in ("1", "true", "yes")
async def _load_research_rank_only_symbols(db: AsyncSession) -> set[str]:
"""Symbols that feed signal IC only (no GTL/candidate replay).
Optional side table ``research_rank_only`` on research snapshots. Missing
table → empty set (production path unchanged).
"""
from sqlalchemy import text
try:
result = await db.execute(text("SELECT symbol FROM research_rank_only"))
return {str(row[0]).upper() for row in result.fetchall() if row[0]}
except Exception:
return set()
def _median_dollar_vol_63(
closes: list[float], volumes: list[float], i: int, lookback: int = 63
) -> float | None:
"""Rolling median of close×volume over ``lookback`` bars ending at ``i`` (inclusive)."""
if i + 1 < lookback or lookback < 2:
return None
dvs: list[float] = []
for k in range(i - lookback + 1, i + 1):
if closes[k] > 0 and volumes[k] >= 0:
dvs.append(closes[k] * float(volumes[k]))
if len(dvs) < max(20, lookback // 2):
return None
dvs_sorted = sorted(dvs)
mid = len(dvs_sorted) // 2
if len(dvs_sorted) % 2:
return dvs_sorted[mid]
return 0.5 * (dvs_sorted[mid - 1] + dvs_sorted[mid])
def _accumulate_signal_series( def _accumulate_signal_series(
records: list, records: list,
collected: dict, collected: dict,
benchmark_closes: dict[date, float] | None = None, benchmark_closes: dict[date, float] | None = None,
*,
symbol: str | None = None,
) -> None: ) -> None:
"""For each weekly as-of bar, emit (signal, forward-return) pairs keyed by ISO """For each weekly as-of bar, emit (signal, forward-return) pairs keyed by ISO
week into ``collected[name][week_key]``. Forward return is close-to-close over week into ``collected[name][week_key]``. Forward return is close-to-close over
HORIZON trading days. Mutates ``collected`` (a dict of dict of list).""" HORIZON trading days. Mutates ``collected`` (a dict of dict of list).
When ``BACKTEST_LIQUID_BREADTH`` is set, observations are dicts with PIT
liquidity fields for the mask; otherwise plain ``(val, fwd)`` tuples so the
production signal path stays unchanged.
"""
n = len(records) n = len(records)
if n < HORIZON + 21: if n < HORIZON + 21:
return return
closes = [float(r.close) for r in records] closes = [float(r.close) for r in records]
highs = [float(r.high) for r in records] highs = [float(r.high) for r in records]
volumes = [float(getattr(r, "volume", 0) or 0) for r in records]
dates = [r.date for r in records] dates = [r.date for r in records]
liquid_mode = _liquid_breadth_top_n() > 0
for i in _weekly_asof_indices(records): for i in _weekly_asof_indices(records):
j = i + HORIZON j = i + HORIZON
if j >= n or closes[i] <= 0: if j >= n or closes[i] <= 0:
@@ -897,8 +968,18 @@ def _accumulate_signal_series(
fwd = closes[j] / closes[i] - 1.0 fwd = closes[j] / closes[i] - 1.0
iso = records[i].date.isocalendar() iso = records[i].date.isocalendar()
week_key = (iso[0], iso[1]) week_key = (iso[0], iso[1])
dvol = _median_dollar_vol_63(closes, volumes, i) if liquid_mode else None
for name, val in _signal_values(dates, closes, highs, i, benchmark_closes).items(): for name, val in _signal_values(dates, closes, highs, i, benchmark_closes).items():
collected[name][week_key].append((val, fwd)) if liquid_mode:
collected[name][week_key].append({
"val": val,
"fwd": fwd,
"close": closes[i],
"median_dvol_63": dvol,
"symbol": symbol,
})
else:
collected[name][week_key].append((val, fwd))
def _rank(xs: list[float]) -> list[float]: def _rank(xs: list[float]) -> list[float]:
@@ -937,6 +1018,110 @@ def _spearman(xs: list[float], ys: list[float]) -> float | None:
return _pearson(_rank(xs), _rank(ys)) return _pearson(_rank(xs), _rank(ys))
def _obs_val_fwd(rec: object) -> tuple[float, float] | None:
"""Unpack a signal observation: ``(val, fwd)`` or research dict form."""
if isinstance(rec, dict):
try:
return float(rec["val"]), float(rec["fwd"])
except (KeyError, TypeError, ValueError):
return None
if isinstance(rec, (tuple, list)) and len(rec) >= 2:
try:
return float(rec[0]), float(rec[1])
except (TypeError, ValueError):
return None
return None
def _filter_liquid_breadth_week(
recs: list,
*,
top_n: int,
min_price: float,
) -> list[tuple[float, float]]:
"""Point-in-time top-N by median $vol among names with price ≥ floor.
Ranking is relative (IEX volume undercount is OK for order stats). Membership
is recomputed every week from as-of bars — never frozen from today's liquidity.
"""
kept = _filter_liquid_breadth_week_rich(
recs, top_n=top_n, min_price=min_price
)
return [(float(r["val"]), float(r["fwd"])) for r in kept]
def _filter_liquid_breadth_week_rich(
recs: list,
*,
top_n: int,
min_price: float,
) -> list[dict]:
"""Same mask as ``_filter_liquid_breadth_week``, returning rich rows.
Single source for harness IC and research diagnostics. Eligible pool =
dict observations with close ≥ min_price and median_dvol_63 > 0; then
keep top_n by dollar volume (highest first). Non-dict legacy tuples are
not eligible for the liquid mask (they have no dvol).
"""
eligible: list[tuple[float, dict]] = [] # (-dvol, row)
for rec in recs:
if not isinstance(rec, dict):
continue
close = rec.get("close")
dvol = rec.get("median_dvol_63")
if close is None or float(close) < min_price:
continue
if dvol is None or float(dvol) <= 0:
continue
pair = _obs_val_fwd(rec)
if pair is None:
continue
row = {
"val": pair[0],
"fwd": pair[1],
"close": float(close),
"median_dvol_63": float(dvol),
"symbol": rec.get("symbol"),
}
# Preserve optional research fields for mom-conditional diagnostics.
for key in ("mom_12_1", "mom_12_1_resid", "vol_6m", "fip_id"):
if key in rec and rec[key] is not None:
row[key] = rec[key]
eligible.append((-float(dvol), row))
eligible.sort(key=lambda item: item[0])
return [row for _, row in eligible[:top_n]]
def _liquid_breadth_week_stats(
recs: list,
*,
top_n: int,
min_price: float,
) -> dict[str, int | bool]:
"""Pre/post mask counts for reconciling avg_cross_section semantics."""
raw = len(recs)
eligible = 0
for rec in recs:
if not isinstance(rec, dict):
continue
close = rec.get("close")
dvol = rec.get("median_dvol_63")
if close is None or float(close) < min_price:
continue
if dvol is None or float(dvol) <= 0:
continue
if _obs_val_fwd(rec) is None:
continue
eligible += 1
post = min(eligible, top_n) if top_n > 0 else eligible
return {
"raw_pool": raw,
"eligible_pre_mask": eligible,
"post_mask": post,
"mask_binds": bool(top_n > 0 and eligible > top_n),
}
def _quintile_spread(pairs: list[tuple[float, float]]) -> float | None: def _quintile_spread(pairs: list[tuple[float, float]]) -> float | None:
"""Mean forward return of the top signal-quintile minus the bottom quintile.""" """Mean forward return of the top signal-quintile minus the bottom quintile."""
n = len(pairs) n = len(pairs)
@@ -982,10 +1167,16 @@ def _signal_evaluation(collected: dict) -> list[dict]:
IC is measured on NON-OVERLAPPING forward windows (weeks thinned to ~HORIZON IC is measured on NON-OVERLAPPING forward windows (weeks thinned to ~HORIZON
apart) so the t-stat isn't inflated by autocorrelation. A signal with no edge apart) so the t-stat isn't inflated by autocorrelation. A signal with no edge
lands near IC 0 / spread 0; one with too few independent windows is flagged lands near IC 0 / score 0; one with too few independent windows is flagged
unreliable rather than trusted on a lucky handful. unreliable rather than trusted on a lucky handful.
When ``BACKTEST_LIQUID_BREADTH=N`` is set, each week's cross-section is first
restricted to the top-N names by point-in-time 63d median dollar volume
(price ≥ BACKTEST_LIQUID_MIN_PRICE). Production path (flag unset) is unchanged.
""" """
stride = max(1, round(HORIZON / 5)) # ISO weeks spanned by the forward window stride = max(1, round(HORIZON / 5)) # ISO weeks spanned by the forward window
top_n = _liquid_breadth_top_n()
min_price = _liquid_min_price()
rows: list[dict] = [] rows: list[dict] = []
for name in sorted(collected): for name in sorted(collected):
weeks_map = collected[name] weeks_map = collected[name]
@@ -994,15 +1185,37 @@ def _signal_evaluation(collected: dict) -> list[dict]:
ics: list[float] = [] ics: list[float] = []
spreads: list[float] = [] spreads: list[float] = []
sizes: list[int] = [] sizes: list[int] = []
raw_sizes: list[int] = []
eligible_sizes: list[int] = []
bind_flags: list[bool] = []
for wk in kept: for wk in kept:
recs = weeks_map[wk] recs = weeks_map[wk]
ic = _spearman([r[0] for r in recs], [r[1] for r in recs]) if top_n > 0:
stats = _liquid_breadth_week_stats(
recs, top_n=top_n, min_price=min_price
)
raw_sizes.append(int(stats["raw_pool"]))
eligible_sizes.append(int(stats["eligible_pre_mask"]))
bind_flags.append(bool(stats["mask_binds"]))
pairs = _filter_liquid_breadth_week(
recs, top_n=top_n, min_price=min_price
)
else:
pairs = []
for rec in recs:
pair = _obs_val_fwd(rec)
if pair is not None:
pairs.append(pair)
if len(pairs) < MIN_CROSS_SECTION:
continue
ic = _spearman([p[0] for p in pairs], [p[1] for p in pairs])
if ic is not None: if ic is not None:
ics.append(ic) ics.append(ic)
spread = _quintile_spread(recs) spread = _quintile_spread(pairs)
if spread is not None: if spread is not None:
spreads.append(spread) spreads.append(spread)
sizes.append(len(recs)) # avg_cross_section is ALWAYS post-mask pair count (the IC sample).
sizes.append(len(pairs))
if not ics: if not ics:
continue continue
mean_ic = sum(ics) / len(ics) mean_ic = sum(ics) / len(ics)
@@ -1011,7 +1224,7 @@ def _signal_evaluation(collected: dict) -> list[dict]:
else: else:
std = 0.0 std = 0.0
t_stat = mean_ic / std * math.sqrt(len(ics)) if std > 0 else None t_stat = mean_ic / std * math.sqrt(len(ics)) if std > 0 else None
rows.append({ row = {
"signal": name, "signal": name,
"weeks": len(ics), "weeks": len(ics),
"avg_cross_section": round(sum(sizes) / len(sizes), 1) if sizes else None, "avg_cross_section": round(sum(sizes) / len(sizes), 1) if sizes else None,
@@ -1020,16 +1233,36 @@ def _signal_evaluation(collected: dict) -> list[dict]:
"ic_positive_pct": round(sum(1 for x in ics if x > 0) / len(ics) * 100, 1), "ic_positive_pct": round(sum(1 for x in ics if x > 0) / len(ics) * 100, 1),
"mean_quintile_spread": round(sum(spreads) / len(spreads), 4) if spreads else None, "mean_quintile_spread": round(sum(spreads) / len(spreads), 4) if spreads else None,
"reliable": len(ics) >= MIN_RELIABLE_PERIODS, "reliable": len(ics) >= MIN_RELIABLE_PERIODS,
}) }
if top_n > 0:
row["liquid_breadth_top_n"] = top_n
row["liquid_min_price"] = min_price
# Explicit pre/post mask diagnostics (reconcile "did top-N bind?").
if raw_sizes:
row["avg_raw_pool"] = round(sum(raw_sizes) / len(raw_sizes), 1)
if eligible_sizes:
row["avg_eligible_pre_mask"] = round(
sum(eligible_sizes) / len(eligible_sizes), 1
)
if bind_flags:
row["mask_binds_pct"] = round(
sum(1 for b in bind_flags if b) / len(bind_flags) * 100, 1
)
rows.append(row)
rows.sort(key=lambda r: r["mean_ic"], reverse=True) rows.sort(key=lambda r: r["mean_ic"], reverse=True)
return rows return rows
def _signal_series(records: list, benchmark_closes: dict[date, float] | None = None) -> dict: def _signal_series(
records: list,
benchmark_closes: dict[date, float] | None = None,
*,
symbol: str | None = None,
) -> dict:
"""Per-ticker signal/forward-return series as a PLAIN (picklable) nested dict """Per-ticker signal/forward-return series as a PLAIN (picklable) nested dict
— no defaultdict/lambda — so it can cross a process boundary.""" — no defaultdict/lambda — so it can cross a process boundary."""
tmp: dict = defaultdict(lambda: defaultdict(list)) tmp: dict = defaultdict(lambda: defaultdict(list))
_accumulate_signal_series(records, tmp, benchmark_closes) _accumulate_signal_series(records, tmp, benchmark_closes, symbol=symbol)
return {name: dict(weeks) for name, weeks in tmp.items()} return {name: dict(weeks) for name, weeks in tmp.items()}
@@ -1041,10 +1274,15 @@ def _replay_and_signals(
benchmark_closes: dict[date, float] | None = None, benchmark_closes: dict[date, float] | None = None,
target_model: str = PRODUCTION_GTL_TARGET_MODEL, target_model: str = PRODUCTION_GTL_TARGET_MODEL,
cadence: str = DEFAULT_BACKTEST_CADENCE, cadence: str = DEFAULT_BACKTEST_CADENCE,
signal_only: bool = False,
) -> tuple[list[dict], dict]: ) -> tuple[list[dict], dict]:
"""The CPU-bound per-ticker work, as a top-level (picklable) function so it can """The CPU-bound per-ticker work, as a top-level (picklable) function so it can
run in a worker process. Takes primitive column arrays (cheap to pickle), run in a worker process. Takes primitive column arrays (cheap to pickle),
rebuilds bar objects, and returns (candidates, signal_series).""" rebuilds bar objects, and returns (candidates, signal_series).
``signal_only=True`` (research rank-only names): skip GTL/candidate replay so
the production portfolio book is never polluted by broad-universe tickers.
"""
date_ords, opens, highs, lows, closes, volumes = columns date_ords, opens, highs, lows, closes, volumes = columns
bars = [ bars = [
SimpleNamespace( SimpleNamespace(
@@ -1052,8 +1290,9 @@ def _replay_and_signals(
) )
for o, op, hi, lo, cl, vo in zip(date_ords, opens, highs, lows, closes, volumes) for o, op, hi, lo, cl, vo in zip(date_ords, opens, highs, lows, closes, volumes)
] ]
return ( candidates: list[dict] = []
_replay_ticker( if not signal_only:
candidates = _replay_ticker(
symbol, symbol,
bars, bars,
config, config,
@@ -1061,8 +1300,10 @@ def _replay_and_signals(
benchmark_closes, benchmark_closes,
target_model, target_model,
cadence, cadence,
), )
_signal_series(bars, benchmark_closes), return (
candidates,
_signal_series(bars, benchmark_closes, symbol=symbol),
) )
@@ -3789,6 +4030,12 @@ async def run_backtest(
result = await db.execute(select(Ticker).order_by(Ticker.symbol)) result = await db.execute(select(Ticker).order_by(Ticker.symbol))
tickers = list(result.scalars().all()) tickers = list(result.scalars().all())
total = len(tickers) total = len(tickers)
rank_only_symbols = await _load_research_rank_only_symbols(db)
if rank_only_symbols:
logger.info(json.dumps({
"event": "backtest_rank_only_loaded",
"count": len(rank_only_symbols),
}))
candidates: list[dict] = [] candidates: list[dict] = []
# Signal IC remains a weekly, non-overlapping diagnostic regardless of the # Signal IC remains a weekly, non-overlapping diagnostic regardless of the
@@ -3847,10 +4094,16 @@ async def run_backtest(
continue continue
if columns is not None: if columns is not None:
futures.append(loop.run_in_executor( futures.append(loop.run_in_executor(
pool, _replay_and_signals, ticker.symbol, columns, config, activation, pool,
_replay_and_signals,
ticker.symbol,
columns,
config,
activation,
benchmark_closes, benchmark_closes,
target_model, target_model,
cadence, cadence,
ticker.symbol in rank_only_symbols,
)) ))
for result in await asyncio.gather(*futures, return_exceptions=True): for result in await asyncio.gather(*futures, return_exceptions=True):
if isinstance(result, Exception): if isinstance(result, Exception):
@@ -3870,10 +4123,15 @@ async def run_backtest(
columns = await _fetch_columns(db, ticker.symbol) columns = await _fetch_columns(db, ticker.symbol)
if columns is not None: if columns is not None:
_merge(await asyncio.to_thread( _merge(await asyncio.to_thread(
_replay_and_signals, ticker.symbol, columns, config, activation, _replay_and_signals,
ticker.symbol,
columns,
config,
activation,
benchmark_closes, benchmark_closes,
target_model, target_model,
cadence, cadence,
ticker.symbol in rank_only_symbols,
)) ))
except Exception: except Exception:
logger.exception("Backtest replay failed for %s", ticker.symbol) logger.exception("Backtest replay failed for %s", ticker.symbol)
@@ -3916,73 +4174,75 @@ async def run_backtest(
portfolio_monitor_report: dict | None = None portfolio_monitor_report: dict | None = None
holdout_report: dict | None = None holdout_report: dict | None = None
min_rr_sweep_report: dict | None = None min_rr_sweep_report: dict | None = None
try: if not _signal_eval_only():
qual_symbols = sorted({
c["symbol"]
for c in candidates
if c.get("qualified")
or any(_qualifies_strategy_variant(c, cfg) for cfg in STRATEGY_VARIANTS)
})
price_columns: dict[str, tuple] = {}
for sym in qual_symbols:
cols = await _fetch_columns(db, sym)
if cols is not None:
price_columns[sym] = cols
spy_closes: dict | None = None
try: try:
oldest = min((cols[0][0] for cols in price_columns.values()), default=None) qual_symbols = sorted({
days_needed = None c["symbol"]
if oldest is not None and not _offline_snapshot_mode(): for c in candidates
days_needed = (date.today() - date.fromordinal(oldest)).days + 30 if c.get("qualified")
spy_closes = await _load_benchmark_closes_for_backtest( or any(_qualifies_strategy_variant(c, cfg) for cfg in STRATEGY_VARIANTS)
db, days=days_needed, refresh=oldest is not None })
) price_columns: dict[str, tuple] = {}
except Exception: for sym in qual_symbols:
logger.exception("Benchmark load for the portfolio sim failed") cols = await _fetch_columns(db, sym)
if cols is not None:
price_columns[sym] = cols
for policy in ("target", "hold"): spy_closes: dict | None = None
sim = _simulate_portfolio( try:
candidates, price_columns, spy_closes, policy, hold_horizon oldest = min((cols[0][0] for cols in price_columns.values()), default=None)
) days_needed = None
if sim is not None: if oldest is not None and not _offline_snapshot_mode():
sim_policies.append({"policy": policy, **sim}) days_needed = (date.today() - date.fromordinal(oldest)).days + 30
strategy_variant_rows = _strategy_variant_sims( spy_closes = await _load_benchmark_closes_for_backtest(
candidates, price_columns, spy_closes, hold_horizon db, days=days_needed, refresh=oldest is not None
) )
exit_policy_rows = _exit_policy_sims( except Exception:
candidates, price_columns, spy_closes, hold_horizon logger.exception("Benchmark load for the portfolio sim failed")
)
live_exit_policy: dict | None = None
try:
from app.services.paper_trade_service import get_exit_policy
live_exit_policy = await get_exit_policy(db) for policy in ("target", "hold"):
except Exception: sim = _simulate_portfolio(
logger.exception("Live exit policy load failed; monitor uses defaults") candidates, price_columns, spy_closes, policy, hold_horizon
portfolio_monitor_report = _portfolio_monitor( )
candidates, price_columns, spy_closes, hold_horizon, if sim is not None:
live_exit_policy=live_exit_policy, sim_policies.append({"policy": policy, **sim})
cadence=cadence, strategy_variant_rows = _strategy_variant_sims(
) candidates, price_columns, spy_closes, hold_horizon
split = _holdout_split() )
if split is not None: exit_policy_rows = _exit_policy_sims(
holdout_report = _holdout_evaluation( candidates, price_columns, spy_closes, hold_horizon
candidates, price_columns, spy_closes, hold_horizon, split, )
live_exit_policy: dict | None = None
try:
from app.services.paper_trade_service import get_exit_policy
live_exit_policy = await get_exit_policy(db)
except Exception:
logger.exception("Live exit policy load failed; monitor uses defaults")
portfolio_monitor_report = _portfolio_monitor(
candidates, price_columns, spy_closes, hold_horizon,
live_exit_policy=live_exit_policy, live_exit_policy=live_exit_policy,
cadence=cadence, cadence=cadence,
) )
if _min_rr_sweep_enabled(): split = _holdout_split()
min_rr_sweep_report = _min_rr_sweep( if split is not None:
candidates, price_columns, spy_closes, activation, current_min_pct, holdout_report = _holdout_evaluation(
hold_horizon, live_exit_policy=live_exit_policy, cadence=cadence, candidates, price_columns, spy_closes, hold_horizon, split,
) live_exit_policy=live_exit_policy,
except Exception: cadence=cadence,
logger.exception("Portfolio simulation failed") )
if _min_rr_sweep_enabled():
min_rr_sweep_report = _min_rr_sweep(
candidates, price_columns, spy_closes, activation, current_min_pct,
hold_horizon, live_exit_policy=live_exit_policy, cadence=cadence,
)
except Exception:
logger.exception("Portfolio simulation failed")
report = { report = {
"generated_at": datetime.now(timezone.utc).isoformat(), "generated_at": datetime.now(timezone.utc).isoformat(),
"tickers": total, "tickers": total,
"rank_only_tickers": len(rank_only_symbols),
"candidates": len(candidates), "candidates": len(candidates),
"qualified": len(qualified), "qualified": len(qualified),
"params": { "params": {
@@ -3999,6 +4259,9 @@ async def run_backtest(
"target_model_label": BACKTEST_TARGET_MODELS[target_model], "target_model_label": BACKTEST_TARGET_MODELS[target_model],
"is_production_target_model": target_model == PRODUCTION_GTL_TARGET_MODEL, "is_production_target_model": target_model == PRODUCTION_GTL_TARGET_MODEL,
"production_reentry_policy": PRODUCTION_REENTRY_POLICY, "production_reentry_policy": PRODUCTION_REENTRY_POLICY,
"liquid_breadth_top_n": _liquid_breadth_top_n() or None,
"liquid_min_price": _liquid_min_price() if _liquid_breadth_top_n() else None,
"signal_eval_only": _signal_eval_only(),
}, },
"activation": activation, "activation": activation,
"overall_qualified": _bucket_stats(qualified), "overall_qualified": _bucket_stats(qualified),
docs/research/README.md
+8 -3
View File
@@ -140,9 +140,9 @@ knobs.
| Lead | Why it's interesting | Blocker | | Lead | Why it's interesting | Blocker |
|---|---|---| |---|---|---|
| **Near-close / MOC execution (ops)** | Recovers overnight momentum drift left on the table by a morning EU scan; evidence closed | Implement schedule + partial-bar scan path; one qualifying scan/day only | | **Near-close / MOC execution (ops)** | Recovers overnight momentum drift left on the table by a morning EU scan; evidence closed | Schedule + fill_mode shipped; live paper validation ongoing |
| **`fip_id`** (information discreteness over the 12-1 window) | **Strongest cross-sectional signal measured on this universe** — IC −0.045, t = −2.91, correct sign; re-derived fingerprint matched Phase A | Doesn't improve *this* book. Revisit when the universe broadens — **after** execution path is decided | | **`fip_id` / liquid breadth** | Fingerprint −0.045 / t −2.91; liquid unconditional **−0.017 / t −1.85** (not green); mom-conditional **−0.088 / t −4.58** | **Parked.** Orphan +0.0575 died (snapshot race). Breadth did not strengthen resid-mom t-stat. Optional reopen = pre-registered two-arm liquid-1500 book first. See [fip-breadth-ic.md](fip-breadth-ic.md) |
| **Broader universe** (`nasdaq_all`) | Strengthens every week's cross-section and the IC t-stat | Grade under the fill mode you will trade | | **Broader universe** | Composition changes factor signs (fip tug-of-war); vol-tilt on breadth is only a **directional hypothesis** (auth. −0.048 / t −1.36) | Any prod broaden must re-validate 80/20 tilt; offline research only; research.sqlite requires completion manifest |
| **Forward paper-trade record** | The only true out-of-sample evidence the snapshot cannot give | Time; mark entries at actual near-close fill once ops ships | | **Forward paper-trade record** | The only true out-of-sample evidence the snapshot cannot give | Time; mark entries at actual near-close fill once ops ships |
| **Better target model for clear-air names** | The return is demonstrably there (#2 wins on raw CAGR in *both* train and test); it's the *flat* 3× ATR target that makes it too expensive in risk | Needs a per-name model, not a constant k×ATR | | **Better target model for clear-air names** | The return is demonstrably there (#2 wins on raw CAGR in *both* train and test); it's the *flat* 3× ATR target that makes it too expensive in risk | Needs a per-name model, not a constant k×ATR |
@@ -169,6 +169,11 @@ knobs.
6. **Fill timing is part of the strategy.** Close-fill reports are not deployable 6. **Fill timing is part of the strategy.** Close-fill reports are not deployable
numbers for an overnight scanner. Grade promotion under the fill mode you will numbers for an overnight scanner. Grade promotion under the fill mode you will
actually trade. actually trade.
7. **Incomplete research artifacts are not results.** The Phase B +0.0575 / t +5.12
liquid-fip row was orphaned within hours: it raced a partially built
`research.sqlite`. Extender now writes a completion manifest; breadth mode
refuses without a match. Same class of protection as calendar-truncation
asserts — do not re-mythologize numbers computed on half a universe.
--- ---
docs/research/fip-breadth-ic.md
+216
View File
@@ -0,0 +1,216 @@
# Broad-universe fip_id IC research (Phase B)
**Status:** **Parked / closed for now.** Unconditional fip not green; mom-conditional lead logged; breadth-momentum thesis challenged. No book sim until reopen.
**Production impact:** none. Display card remains context-only. No deploy from this work.
**Artifacts:** research log + compact reports + env-gated harness hooks; tooling stays for a future reopen.
## Scope
- Research only — production universe, gate, scanner, schedule unchanged.
- Snapshot: `research.sqlite` (~4,650 tickers = prod + nasdaq_all extend).
- Liquid mask: top **1,500** by point-in-time 63d median $vol, price ≥ **$5**/week.
- **Completion manifest required:** extender writes `<snapshot>.manifest.json`; breadth runners refuse without a matching complete manifest (see §Race guard).
## Caveats
- Survivorship bias (today’s constituents, history backfilled).
- IEX volume undercount → relative $vol rank only.
- Pool skew: Nasdaq-heavy; missing pure NYSE mid-caps.
- Do not mix multi-signal tables across universe baselines.
- **Do not cite orphaned 21:14 numbers** (see below).
---
## Fingerprint (505-name prod)
| | Expected | Observed |
|---|---:|---:|
| mean IC | −0.045 | **−0.045** |
| t-stat | −2.9 | **−2.91** |
| weeks / N / reliable | ≥12 / ~500 / true | 35 / 497.7 / true |
**Pass.** Formula + pipeline trustworthy.
Residual momentum on the same fingerprint (what the production book ranks on): **IC +0.055 / t +1.98**.
---
## The orphan (21:14) — root cause
| Source | fip IC (liquid ~1500) | t |
|---|---:|---:|
| Orphan run 21:14 (removed from tree; was `fip-breadth-20260718-211440-breadth.json`) | **+0.0575** | **+5.12** |
| Single-sourced recompute on complete snapshot (2026-07-19) | **−0.0168** | **−1.85** |
That is a **sign disagreement** on the same intended quantity. Method rule: the number you cannot reconcile is the number you cannot use.
### Verdict: orphaned — raced the snapshot build
**Not** “orphaned, unexplained.” The mechanism is derivable from the table itself:
1. **Code was not the difference.** Reconcile shows the old harness path and the new shared filter produce **identical** results on current data (−0.0168 / −1.85). The implementation fork is closed.
2. **Data was the difference.** On today’s complete snapshot the liquid mask **binds in 97.1% of weeks** at top-N = 1,500. Dense signals (e.g. `vol_6m`) post-mask at **exactly 1,500**. The orphaned report’s `vol_6m` averaged **~1,475** cross-section — a masked run on complete data cannot do that. At 21:14 the eligible pool was smaller than 1,500 and the mask never bound.
3. **Timeline fits.** Extender fixes landed ~20:32 / 20:34; full fetch takes ~30 minutes; breadth run fired **21:14** against a partially built `research.sqlite`. Every number in that report was computed on an incomplete universe.
**Do not cite +0.0575 / t +5.12.** It survived less than six hours of contact with project discipline — that is the system working, not time wasted. The orphan JSON was **deleted from the tree** (still in Git history) so it cannot be re-imported as evidence.
**Kept artifacts**
| File | Role |
|---|---|
| `reports/fip-reconcile-20260719-000520.json` | Authoritative single-sourced ICs (compact; membership dumps stripped) |
| `reports/fip-breadth-20260718-211440-fingerprint.json` | Prod fingerprint pass |
### Race guard (same class as calendar truncation)
| Piece | Behavior |
|---|---|
| `extend_snapshot_universe.py` | Clears any prior manifest on start; on full completion writes `<output>.manifest.json` with `complete=true`, ticker / OHLCV / rank_only counts, `finished_at`. `--limit` smoke runs write `complete=false`. |
| `run_fip_breadth_research.py` / `run_fip_breadth_diagnostics.py` | **Refuse** breadth mode unless a matching complete manifest exists and live counts equal the recorded totals. |
Helper: `scripts/research_snapshot_manifest.py`.
---
## Authoritative unconditional liquid fip (post-reconciliation)
| metric | value |
|---|---:|
| mean_ic | **−0.0168** |
| ic_t_stat | **−1.85** |
| weeks | 35 |
| avg_cross_section (**post-mask IC sample**) | 1471.2 |
| avg_raw_pool | 3214.4 |
| avg_eligible_pre_mask | **2338.4** |
| mask_binds_pct | **97.1%** |
| reliable | true |
**Mask binds hard** on complete data (eligible ≫ 1500). Post-mask IC N for fip is ~1471 because not every liquid name has a valid 12-1 fip path — that is signal availability, not a non-binding mask. Contrast orphan `vol_6m` avg N ~1475 vs complete-data `vol_6m` avg N **1500**.
Harness `_signal_evaluation` vs manual IC through the same filter: **exact match** (−0.0168 / −1.85).
**Iron rule unconditional:** **not green** (|IC| 0.017 < 0.03), correct mild-negative sign.
---
## Context table (orphaned 21:14 vs authoritative) — kill the myth numbers
The context table died with the orphan. **−0.16 must not survive in the log.**
| signal (liquid ~1500) | orphaned (21:14) | authoritative (shared filter) | consequence |
|---|---:|---:|---|
| **vol_6m** | −0.16 / t **−6.1** | **−0.048 / t −1.36** | “High-vol tilt harmful on breadth” **downgrades from finding to directional hypothesis** — not significant |
| **raw mom** (`mom_12_1`) | +0.10 / t +4.6 | **+0.046 / t +1.91** | Below iron-rule bar on this pool |
| **resid mom** (`mom_12_1_resid`) | +0.04 / t +2.3 | **+0.029 / t +1.33** | Ditto, and weaker than raw |
### Breadth-momentum thesis — challenged
That last pair is the sobering one. Momentum on liquid breadth is **marginal**. The “more breadth strengthens the momentum t-stat” thesis that motivated Phase B is **empirically wrong on this pool**: same 35 weeks, triple the names, residual-mom t-stat **fell** versus the 505-name fingerprint (**0.055 / 1.98** → **0.029 / 1.33**). The clean momentum edge lives in the large-cap universe already traded.
Meanwhile the strongest reliable signal on liquid breadth is now **mom-conditional fip** (−0.088 / −4.58) — but a fip tilt presupposes a breadth momentum book worth tilting, and that is no longer free.
---
## Compositional story (supported)
`fip_id = sign(PRET)×(%neg−%pos)` pools:
- **Continuous winners** → want **negative** IC
- **Continuous bleeders** → want **positive** IC
| check | IC | t | read |
|---|---:|---:|---|
| Prod-universe subset inside liquid | **−0.044** | **−2.88** | Matches fingerprint → compositional, not regime change |
| Tier 1–800 (senior) | **−0.035** | **−2.99** | Winner leg |
| Tier 801–1500 (junior) | **+0.014** | +1.25 | More bleeder / junk weight |
| Lagged membership (prior-week $vol) | −0.010 | −0.93 | Same sign as same-week; not a +5σ leak artifact |
**Do not log “on Nasdaq, jumpy paths outperform.”** That would mythologize an orphaned +0.06.
---
## Platform-relevant test: momentum-conditional fip
Among liquid top-1500, keep **mom_12_1 ≥ P80** (~294 names/week):
| metric | value |
|---|---:|
| mean_ic | **−0.0879** |
| ic_t_stat | **−4.58** |
| ic_positive_pct | 22.9% |
| weeks | 35 |
| reliable | **true** |
Computed on the **same single-sourced path** as the authoritative −0.017. This is the paper’s claim and the only version a gate could consume.
| Decision | |
|---|---|
| Unconditional fip | **Closed** for production |
| Mom-conditional fip | **Alive as book-tilt candidate only** — and only after a baseline breadth book proves itself |
| Display card | Stays |
| Production change | **None** |
---
## Vol-tilt warning (softened)
| signal (liquid, single-sourced) | IC | t |
|---|---:|---:|
| vol_6m | −0.048 | **−1.36** |
| mom_12_1 | +0.046 | +1.91 |
| mom_12_1_resid | +0.029 | +1.33 |
High-vol names **tend** to underperform on this pool relative to a clean S&P-like book — that is a **directional hypothesis**, not a finding. Production **80/20 high-vol tilt** was validated on S&P-like names. If the universe ever broadens in production, re-validate that tilt; do not treat the orphaned −0.16 / t −6.1 as evidence.
---
## What this means for the book experiment
A fip tilt presupposes a breadth momentum book worth tilting — **that is no longer free.**
**Caution against over-reacting the other way:** modest cross-sectional IC does not preclude a good book. The 505-name book turns resid-mom IC ~0.055 into Sharpe ~2 because the gate trades the **extreme tail**, not the linear sort. The breadth book might still work; it just has to **prove it** before the fip arm means anything. If the baseline cannot clearly beat the existing production book’s territory, fip’s future is a footnote regardless of −4.58.
### Parked next step (if reopened): pre-registered two-arm design
Not started — **design only**, pre-register before any sim:
| Arm | Definition |
|---|---|
| **A — baseline** | Top-quintile residual (or raw — pick one and lock) momentum book on liquid-1500; **no fip**; honest costs; next-open or near-close fills; production-like capacity / risk / stops |
| **B — +fip tilt** | Same book + mom-conditional fip tilt (among mom winners, prefer smoother paths / negative fip_id) |
| Grade on | Spec |
|---|---|
| Split | Entry-date train / validation (`BACKTEST_HOLDOUT_SPLIT` naming — not pristine holdout) |
| Metrics | Sharpe + Mertens/Lo SE, PSR, **DSR**; max DD; turnover; cost drag |
| Promote bar | Arm A must be in production-book territory first; Arm B must beat A on validation with DSR-aware multiple-testing honesty |
| Fail-closed | If A fails, fip is a footnote; do not shop tilts on a dead baseline |
---
## How to re-run (research branch only)
```powershell
# 1) Full extend writes completion manifest (required)
.\.venv\Scripts\python.exe scripts\extend_snapshot_universe.py `
--source backtest_snapshots\prod.sqlite `
--output backtest_snapshots\research.sqlite
# 2) Breadth / diagnostics refuse without matching manifest
.\.venv\Scripts\python.exe scripts\run_fip_breadth_diagnostics.py `
--research-snapshot backtest_snapshots\research.sqlite `
--prod-snapshot backtest_snapshots\prod.sqlite `
--workers 6 --allow-spawn
```
---
## Bottom line
1. Formal iron-rule screen: **not green** either before or after reconciliation.
2. **+0.0575 / +5.12 is orphaned: raced the snapshot build** — authoritative unconditional liquid fip is **−0.017 / −1.9**; mask binds (~97%) on complete data.
3. Context-table myths die with the orphan: **vol −0.16 is not real**; authoritative vol is **−0.048 / t −1.36** (directional only).
4. Compositional tug-of-war is the right story; jumpiness premium is not.
5. **Breadth does not strengthen residual-mom t-stat** on this pool (0.055/1.98 → 0.029/1.33).
6. **Mom-conditional −0.088 / −4.6 stands** on the single-sourced path → optional next step is a **pre-registered two-arm breadth book** (baseline first), not a gate wire-in.
7. Manifest guard is in place so the race cannot recur silently.
reports/README.md
+20
View File
@@ -41,3 +41,23 @@ rejected stop-adjustment path, and add no decision evidence beyond the final
daily matrix and narrative. Their matching one-off runners were removed too. daily matrix and narrative. Their matching one-off runners were removed too.
All remain recoverable from Git history. Rebuildable candidate pickle caches All remain recoverable from Git history. Rebuildable candidate pickle caches
are intentionally ignored and must not be committed. are intentionally ignored and must not be committed.
### Phase B fip breadth IC (2026-07-18/19) — compact evidence
Canonical artifacts:
- `fip-reconcile-20260719-000520.json` — single-sourced authoritative ICs
(unconditional liquid fip, tiers, prod-subset, mom-conditional, context
signals). Membership symbol dumps stripped after the decision; narrative in
[`docs/research/fip-breadth-ic.md`](../docs/research/fip-breadth-ic.md).
- `fip-breadth-20260718-211440-fingerprint.json` — prod-snapshot fingerprint
pass (fip IC −0.045 / t −2.91).
Removed as superseded / dangerous intermediate noise (recoverable from Git):
- `fip-breadth-20260718-211440-breadth.json` (+ wrapper) — **orphaned** +0.0575
/ t +5.12 from racing a partial `research.sqlite`. Kept out of the tree so it
cannot be re-mythologized.
- `fip-breadth-20260718-194828*.json` — fingerprint-only partial run.
- `fip-breadth-diagnostics-20260718-213705.json` and `…-213908.json` — dual-path
diagnostics superseded by the single-sourced reconcile.
reports/fip-breadth-20260718-211440-fingerprint.json
+578
View File
@@ -0,0 +1,578 @@
{
"generated_at": "2026-07-18T19:16:37.954856+00:00",
"tickers": 506,
"rank_only_tickers": 0,
"candidates": 202765,
"qualified": 1086,
"params": {
"step_days": 5,
"step_sessions": 5,
"entry_cadence": "weekly",
"signal_eval_cadence": "weekly",
"horizon_days": 30,
"min_lookback": 60,
"cost_per_side_pct": 0.1,
"target_model": "production_gtl",
"target_model_label": "Live GTL (production)",
"is_production_target_model": true,
"production_reentry_policy": "gate_reset",
"liquid_breadth_top_n": null,
"liquid_min_price": null,
"signal_eval_only": true
},
"activation": {
"min_momentum_percentile": 80.0,
"min_rr": 2.0,
"min_confidence": 0.0,
"require_high_conviction": false,
"exclude_conflicts": false,
"exclude_neutral": true
},
"overall_qualified": {
"total": 1086,
"wins": 379,
"losses": 591,
"expired": 116,
"hit_rate": 39.1,
"avg_r": 0.255,
"total_r": 276.76,
"net_avg_r": 0.209,
"net_total_r": 226.56,
"best_r": 8.85,
"worst_r": -3.38,
"avg_hold_days": 12.0,
"net_r_per_day": 0.0174,
"median_net_r": -1.031,
"profit_factor": 1.34,
"net_avg_r_ex_top5": 0.049
},
"overall_all": {
"total": 202765,
"wins": 82220,
"losses": 113809,
"expired": 6736,
"hit_rate": 41.9,
"avg_r": -0.04,
"total_r": -8186.18,
"net_avg_r": -0.095,
"net_total_r": -19184.55,
"best_r": 9.24,
"worst_r": -16.42,
"avg_hold_days": 8.2,
"net_r_per_day": -0.0115,
"median_net_r": -1.035,
"profit_factor": 0.85,
"net_avg_r_ex_top5": -0.22
},
"by_direction": {
"long": {
"total": 1086,
"wins": 379,
"losses": 591,
"expired": 116,
"hit_rate": 39.1,
"avg_r": 0.255,
"total_r": 276.76,
"net_avg_r": 0.209,
"net_total_r": 226.56,
"best_r": 8.85,
"worst_r": -3.38,
"avg_hold_days": 12.0,
"net_r_per_day": 0.0174,
"median_net_r": -1.031,
"profit_factor": 1.34,
"net_avg_r_ex_top5": 0.049
},
"short": {
"total": 0,
"wins": 0,
"losses": 0,
"expired": 0,
"hit_rate": null,
"avg_r": null,
"total_r": null,
"net_avg_r": null,
"net_total_r": null,
"best_r": null,
"worst_r": null,
"avg_hold_days": null,
"net_r_per_day": null,
"median_net_r": null,
"profit_factor": null,
"net_avg_r_ex_top5": null
}
},
"min_momentum_percentile": 80.0,
"sweep": [
{
"min_momentum_percentile": 90.0,
"total": 497,
"wins": 177,
"losses": 269,
"expired": 51,
"hit_rate": 39.7,
"avg_r": 0.276,
"total_r": 137.05,
"net_avg_r": 0.235,
"net_total_r": 116.55,
"best_r": 8.85,
"worst_r": -3.38,
"avg_hold_days": 11.8,
"net_r_per_day": 0.0199,
"median_net_r": -1.026,
"profit_factor": 1.39,
"net_avg_r_ex_top5": 0.071
},
{
"min_momentum_percentile": 80.0,
"total": 1086,
"wins": 379,
"losses": 591,
"expired": 116,
"hit_rate": 39.1,
"avg_r": 0.255,
"total_r": 276.76,
"net_avg_r": 0.209,
"net_total_r": 226.56,
"best_r": 8.85,
"worst_r": -3.38,
"avg_hold_days": 12.0,
"net_r_per_day": 0.0174,
"median_net_r": -1.031,
"profit_factor": 1.34,
"net_avg_r_ex_top5": 0.049
},
{
"min_momentum_percentile": 70.0,
"total": 1841,
"wins": 597,
"losses": 1062,
"expired": 182,
"hit_rate": 36.0,
"avg_r": 0.152,
"total_r": 280.26,
"net_avg_r": 0.104,
"net_total_r": 190.75,
"best_r": 8.85,
"worst_r": -4.21,
"avg_hold_days": 11.8,
"net_r_per_day": 0.0088,
"median_net_r": -1.037,
"profit_factor": 1.16,
"net_avg_r_ex_top5": -0.055
},
{
"min_momentum_percentile": 60.0,
"total": 2772,
"wins": 873,
"losses": 1611,
"expired": 288,
"hit_rate": 35.1,
"avg_r": 0.126,
"total_r": 348.07,
"net_avg_r": 0.075,
"net_total_r": 209.25,
"best_r": 8.85,
"worst_r": -4.21,
"avg_hold_days": 12.0,
"net_r_per_day": 0.0063,
"median_net_r": -1.04,
"profit_factor": 1.12,
"net_avg_r_ex_top5": -0.077
},
{
"min_momentum_percentile": 50.0,
"total": 3901,
"wins": 1182,
"losses": 2295,
"expired": 424,
"hit_rate": 34.0,
"avg_r": 0.089,
"total_r": 345.37,
"net_avg_r": 0.038,
"net_total_r": 146.96,
"best_r": 8.85,
"worst_r": -4.86,
"avg_hold_days": 12.1,
"net_r_per_day": 0.0031,
"median_net_r": -1.042,
"profit_factor": 1.06,
"net_avg_r_ex_top5": -0.114
},
{
"min_momentum_percentile": 0.0,
"total": 14588,
"wins": 3719,
"losses": 9271,
"expired": 1598,
"hit_rate": 28.6,
"avg_r": -0.065,
"total_r": -952.08,
"net_avg_r": -0.115,
"net_total_r": -1676.86,
"best_r": 9.24,
"worst_r": -15.94,
"avg_hold_days": 12.1,
"net_r_per_day": -0.0095,
"median_net_r": -1.043,
"profit_factor": 0.84,
"net_avg_r_ex_top5": -0.275
}
],
"gate_ablation": [
{
"variant": "all_floors",
"total": 1086,
"wins": 379,
"losses": 591,
"expired": 116,
"hit_rate": 39.1,
"avg_r": 0.255,
"total_r": 276.76,
"net_avg_r": 0.209,
"net_total_r": 226.56,
"best_r": 8.85,
"worst_r": -3.38,
"avg_hold_days": 12.0,
"net_r_per_day": 0.0174,
"median_net_r": -1.031,
"profit_factor": 1.34,
"net_avg_r_ex_top5": 0.049,
"hold_days": 30,
"hold_avg_r": 0.631,
"hold_net_avg_r": 0.585,
"hold_total_r": 684.97
},
{
"variant": "no_confidence_floor",
"total": 1093,
"wins": 380,
"losses": 596,
"expired": 117,
"hit_rate": 38.9,
"avg_r": 0.25,
"total_r": 273.55,
"net_avg_r": 0.204,
"net_total_r": 222.99,
"best_r": 8.85,
"worst_r": -3.38,
"avg_hold_days": 11.9,
"net_r_per_day": 0.0171,
"median_net_r": -1.031,
"profit_factor": 1.33,
"net_avg_r_ex_top5": 0.045,
"hold_days": 30,
"hold_avg_r": 0.626,
"hold_net_avg_r": 0.58,
"hold_total_r": 684.08
},
{
"variant": "no_rr_floor",
"total": 6849,
"wins": 3235,
"losses": 3409,
"expired": 205,
"hit_rate": 48.7,
"avg_r": 0.112,
"total_r": 770.17,
"net_avg_r": 0.061,
"net_total_r": 418.96,
"best_r": 8.85,
"worst_r": -5.16,
"avg_hold_days": 7.9,
"net_r_per_day": 0.0078,
"median_net_r": -0.061,
"profit_factor": 1.11,
"net_avg_r_ex_top5": -0.061,
"hold_days": 30,
"hold_avg_r": 0.354,
"hold_net_avg_r": 0.303,
"hold_total_r": 2425.86
},
{
"variant": "no_neutral_exclusion",
"total": 2313,
"wins": 770,
"losses": 1279,
"expired": 264,
"hit_rate": 37.6,
"avg_r": 0.2,
"total_r": 462.35,
"net_avg_r": 0.154,
"net_total_r": 355.89,
"best_r": 8.85,
"worst_r": -3.38,
"avg_hold_days": 12.6,
"net_r_per_day": 0.0122,
"median_net_r": -1.031,
"profit_factor": 1.25,
"net_avg_r_ex_top5": 0.004,
"hold_days": 30,
"hold_avg_r": 0.583,
"hold_net_avg_r": 0.537,
"hold_total_r": 1348.89
},
{
"variant": "momentum_only",
"total": 14696,
"wins": 6827,
"losses": 7359,
"expired": 510,
"hit_rate": 48.1,
"avg_r": 0.114,
"total_r": 1669.64,
"net_avg_r": 0.064,
"net_total_r": 936.93,
"best_r": 8.85,
"worst_r": -5.71,
"avg_hold_days": 8.4,
"net_r_per_day": 0.0076,
"median_net_r": -1.013,
"profit_factor": 1.11,
"net_avg_r_ex_top5": -0.055,
"hold_days": 30,
"hold_avg_r": 0.395,
"hold_net_avg_r": 0.345,
"hold_total_r": 5798.82
}
],
"gate_ablation_note": "Each row re-qualifies the same candidates at the current momentum cutoff (80) with one floor removed (long-only while the momentum gate is active). If dropping a floor doesn't hurt net expectancy, that floor isn't pulling its weight. The Hold columns grade the same variants under the hold-to-horizon time exit instead of the S/R target \u2014 the view that matters if the exit policy moves to a fixed hold.",
"time_exit_sweep": [
{
"hold_days": 5,
"total": 1086,
"wins": 603,
"win_rate": 55.5,
"avg_r": 0.175,
"total_r": 190.16,
"net_avg_r": 0.129,
"net_total_r": 139.97,
"best_r": 5.09,
"worst_r": -2.51,
"avg_hold_days": 4.5,
"net_r_per_day": 0.0285,
"median_net_r": 0.115,
"profit_factor": 1.36,
"net_avg_r_ex_top5": -0.002
},
{
"hold_days": 10,
"total": 1086,
"wins": 559,
"win_rate": 51.5,
"avg_r": 0.357,
"total_r": 387.9,
"net_avg_r": 0.311,
"net_total_r": 337.7,
"best_r": 6.73,
"worst_r": -2.51,
"avg_hold_days": 7.9,
"net_r_per_day": 0.0395,
"median_net_r": 0.031,
"profit_factor": 1.67,
"net_avg_r_ex_top5": 0.112
},
{
"hold_days": 21,
"total": 1086,
"wins": 487,
"win_rate": 44.8,
"avg_r": 0.525,
"total_r": 570.33,
"net_avg_r": 0.479,
"net_total_r": 520.14,
"best_r": 9.86,
"worst_r": -3.38,
"avg_hold_days": 13.7,
"net_r_per_day": 0.0349,
"median_net_r": -1.027,
"profit_factor": 1.81,
"net_avg_r_ex_top5": 0.191
},
{
"hold_days": 30,
"total": 1086,
"wins": 434,
"win_rate": 40.0,
"avg_r": 0.631,
"total_r": 684.97,
"net_avg_r": 0.585,
"net_total_r": 634.78,
"best_r": 12.87,
"worst_r": -3.38,
"avg_hold_days": 17.8,
"net_r_per_day": 0.0329,
"median_net_r": -1.033,
"profit_factor": 1.9,
"net_avg_r_ex_top5": 0.212
}
],
"portfolio_sim": {
"params": {
"starting_capital": 10000.0,
"max_positions": 10,
"risk_per_trade_pct": 1.0,
"notional_cap_pct": 20.0,
"cost_per_side_pct": 0.1,
"hold_days": 30
},
"policies": [],
"note": "One capital-constrained book over the same qualified setups the tables above grade per-setup: at most 10 concurrent positions (one per ticker), best momentum first, fixed-fractional risk sizing with a no-leverage cap, entries at the detection close, stops filled at the worse of stop or open. 'target' races the S/R target against the stop (timeout at the horizon); 'hold' keeps the initial stop and exits at the horizon close. SPY return is price-only over the same window. In-sample; no dividends."
},
"strategy_variants": {
"variants": [],
"note": "Research-only hold-to-horizon portfolio variants. Production now uses residual 12-1 momentum at cutoff 80; the remaining rows compare the legacy raw rank, raw cutoff 90, one max-15 capacity check, and volatility overlays."
},
"exit_policy_variants": {
"variants": [],
"note": "Research-only exit policies over the residual/high-vol 80/20 entry candidate. Every row uses the same entry qualification/ranking and changes only the exit discipline."
},
"portfolio_monitor": null,
"production_cadence_comparison": null,
"holdout": null,
"min_rr_sweep": null,
"target_model_diagnostics": {
"target_model": "production_gtl",
"target_model_label": "Live GTL (production)",
"candidate_count": 202765,
"primary_source_counts": {
"pivot_point": 196290,
"range_grid": 180036
},
"primary_round_only": 0,
"primary_strength_100": 138596,
"avg_primary_strength": 80.109,
"avg_primary_distance_atr": 2.293,
"avg_primary_rejection_count": 41.908,
"avg_raw_level_count": 53.204,
"avg_gate_level_count": 53.204
},
"signal_eval": [
{
"signal": "vol_6m",
"weeks": 39,
"avg_cross_section": 498.2,
"mean_ic": 0.0609,
"ic_t_stat": 1.48,
"ic_positive_pct": 64.1,
"mean_quintile_spread": 0.0337,
"reliable": true
},
{
"signal": "mom_12_1_resid",
"weeks": 35,
"avg_cross_section": 497.7,
"mean_ic": 0.0552,
"ic_t_stat": 1.98,
"ic_positive_pct": 60.0,
"mean_quintile_spread": 0.0207,
"reliable": true
},
{
"signal": "mom_12_1",
"weeks": 35,
"avg_cross_section": 497.7,
"mean_ic": 0.0531,
"ic_t_stat": 1.61,
"ic_positive_pct": 65.7,
"mean_quintile_spread": 0.0206,
"reliable": true
},
{
"signal": "trend_200",
"weeks": 37,
"avg_cross_section": 497.9,
"mean_ic": 0.0161,
"ic_t_stat": 0.44,
"ic_positive_pct": 59.5,
"mean_quintile_spread": 0.006,
"reliable": true
},
{
"signal": "reversal_1m",
"weeks": 43,
"avg_cross_section": 498.7,
"mean_ic": 0.0059,
"ic_t_stat": 0.22,
"ic_positive_pct": 53.5,
"mean_quintile_spread": 0.0053,
"reliable": true
},
{
"signal": "mom_6_1",
"weeks": 39,
"avg_cross_section": 498.2,
"mean_ic": 0.0051,
"ic_t_stat": 0.21,
"ic_positive_pct": 56.4,
"mean_quintile_spread": 0.0087,
"reliable": true
},
{
"signal": "mom_3_1",
"weeks": 42,
"avg_cross_section": 498.5,
"mean_ic": -0.0064,
"ic_t_stat": -0.25,
"ic_positive_pct": 50.0,
"mean_quintile_spread": 0.0046,
"reliable": true
},
{
"signal": "high_52w",
"weeks": 35,
"avg_cross_section": 497.7,
"mean_ic": -0.0086,
"ic_t_stat": -0.26,
"ic_positive_pct": 54.3,
"mean_quintile_spread": -0.0088,
"reliable": true
},
{
"signal": "fip_id",
"weeks": 35,
"avg_cross_section": 497.7,
"mean_ic": -0.045,
"ic_t_stat": -2.91,
"ic_positive_pct": 25.7,
"mean_quintile_spread": -0.0168,
"reliable": true
}
],
"signal_eval_note": "Cross-sectional rank-IC of price-only signals vs the forward 30-day return (min 20 names/window). |IC| \u2273 0.03 with a consistent sign is a real (if small) edge; near 0 means ranking on it sorts nothing. Momentum factors and high_52w are expected positive; reversal_1m and vol_6m expected negative (mean-reversion / low-vol anomaly). IC is measured on non-overlapping windows; signals with fewer than 12 independent windows are flagged unreliable (too few regimes \u2014 deepen history with the Data Backfill job).",
"note": "Sentiment & fundamentals held neutral (no point-in-time history). Stops fill at the worse of the stop or the bar's open (gaps through the stop are modeled, so a loss can exceed \u22121R); targets never fill better than their level. ~6 months \u2248 one market regime \u2014 treat as directional, not gospel.",
"recommendation": {
"headline": "Trade the qualified list long-only; hold 30 trading days with the initial ATR stop.",
"items": [
{
"topic": "exit",
"text": "Legacy exit diagnostic: hold 30 trading days with the initial stop (+0.58R net/trade vs +0.21R for the S/R target exit)."
},
{
"topic": "gate",
"text": "Gate: the confidence floor adds nothing \u2014 dropping it costs +0.01R/trade and adds 7 trades."
},
{
"topic": "gate",
"text": "Gate: keep the R:R floor (worth +0.28R/trade under the hold exit)."
},
{
"topic": "gate",
"text": "Gate: keep the NEUTRAL exclusion (worth +0.05R/trade under the hold exit)."
},
{
"topic": "cutoff",
"text": "Residual-momentum cutoff: 90 has the best per-trade net (+0.23R over 497 setups)."
},
{
"topic": "robustness",
"text": "Robustness: expectancy survives removing the top 5% of winners (+0.21R net/trade under the recommended 30d hold) \u2014 the edge is not a handful of outliers."
}
],
"note": "Derived from this report's numbers on every run \u2014 the advice flips if the data does."
},
"research_recommendation": {
"items": [],
"note": "Strategy variants unavailable; re-run the backtest after benchmark data is present."
}
}
reports/fip-reconcile-20260719-000520.json
+125
View File
@@ -0,0 +1,125 @@
{
"generated_at": "2026-07-19T00:05:20.113638",
"research_snapshot": "C:\\Workspace\\signal-platform\\backtest_snapshots\\research.sqlite",
"top_n": 1500,
"min_price": 5.0,
"prod_subset_n": 506,
"panel_tickers": 4403,
"single_source": "diagnostics uses harness _signal_series + _filter_liquid_breadth_week_rich only (no parallel mask)",
"avg_cross_section_semantics": "avg_cross_section = post-mask IC sample size. avg_raw_pool = pre-filter observations. avg_eligible_pre_mask = pass price+dvol before top-N. mask_binds_pct = weeks where eligible_pre_mask > top_n.",
"harness_self_consistent": true,
"checks": {
"fip_harness_signal_eval": {
"note": "Authoritative harness _signal_evaluation on collected fip_id",
"signal": "fip_id",
"weeks": 35,
"avg_cross_section": 1471.2,
"mean_ic": -0.0168,
"ic_t_stat": -1.85,
"ic_positive_pct": 40.0,
"mean_quintile_spread": -0.0052,
"reliable": true,
"liquid_breadth_top_n": 1500,
"liquid_min_price": 5.0,
"avg_raw_pool": 3214.4,
"avg_eligible_pre_mask": 2338.4,
"mask_binds_pct": 97.1
},
"fip_same_week_via_shared_filter": {
"note": "Same collected data, IC via shared _filter_liquid_breadth_week_rich",
"mean_ic": -0.0168,
"ic_t_stat": -1.85,
"weeks": 35,
"avg_cross_section": 1471.2,
"ic_positive_pct": 40.0,
"reliable": true
},
"fip_lagged_membership_1w": {
"note": "Top-N by prior-week $vol on current fip pool (shared filter)",
"mean_ic": -0.0102,
"ic_t_stat": -0.93,
"weeks": 35,
"avg_cross_section": 1471.2,
"ic_positive_pct": 40.0,
"reliable": true
},
"fip_tier_1_800": {
"note": "Senior liquid ranks 1\u2013800",
"mean_ic": -0.035,
"ic_t_stat": -2.99,
"weeks": 35,
"avg_cross_section": 791.2,
"ic_positive_pct": 25.7,
"reliable": true
},
"fip_tier_801_1500": {
"note": "Junior liquid ranks 801\u2013top_n",
"mean_ic": 0.0141,
"ic_t_stat": 1.25,
"weeks": 35,
"avg_cross_section": 700.0,
"ic_positive_pct": 60.0,
"reliable": true
},
"fip_prod_universe_subset": {
"note": "Prod.sqlite symbols inside liquid fip set",
"mean_ic": -0.0444,
"ic_t_stat": -2.88,
"weeks": 35,
"avg_cross_section": 497.5,
"ic_positive_pct": 25.7,
"reliable": true
},
"fip_momentum_conditional_top20pct": {
"note": "Among liquid fip set, mom_12_1 \u2265 P80 (paper / gate-relevant)",
"mean_ic": -0.0879,
"ic_t_stat": -4.58,
"weeks": 35,
"avg_cross_section": 294.3,
"ic_positive_pct": 22.9,
"reliable": true
},
"vol_6m_liquid": {
"note": "vol_6m through shared filter",
"mean_ic": -0.0478,
"ic_t_stat": -1.36,
"weeks": 35,
"avg_cross_section": 1500.0,
"ic_positive_pct": 37.1,
"reliable": true
},
"mom_12_1_liquid": {
"note": "raw mom through shared filter",
"mean_ic": 0.0462,
"ic_t_stat": 1.91,
"weeks": 35,
"avg_cross_section": 1471.2,
"ic_positive_pct": 65.7,
"reliable": true
},
"mom_12_1_resid_liquid": {
"note": "residual mom through shared filter",
"mean_ic": 0.0289,
"ic_t_stat": 1.33,
"weeks": 35,
"avg_cross_section": 1471.2,
"ic_positive_pct": 60.0,
"reliable": true
}
},
"interpretation": {
"harness_and_shared_filter_agree": true,
"mask_binds_pct": 97.1,
"avg_eligible_pre_mask": 2338.4,
"avg_raw_pool": 3214.4,
"prod_subset_still_negative": true,
"junior_tier_more_positive": true,
"lag_same_sign_as_same_week": true,
"mom_conditional_negative_and_reliable": true,
"orphan_plus_five_sigma": "Prior report fip-breadth-20260718-211440-breadth.json listed fip IC +0.0575 / t +5.12. This single-sourced recompute is the authoritative number; if it disagrees, the +0.0575 row is orphaned.",
"compositional_story": "fip_id pools continuous winners (neg IC) vs continuous bleeders (pos IC). Prod-subset and senior liquid stay negative; junior liquid is less negative / positive \u2014 composition, not jumpiness premium.",
"vol_tilt_warning": "High-vol names underperform on breadth relative to S&P-like books. Re-validate production 80/20 high-vol tilt before any universe broaden."
},
"platform_verdict": "Mom-conditional fip ALIVE as book-tilt candidate (needs book sim) \u2014 not production wire-in. Unconditional fip not green.",
"membership_dumps_note": "Removed 5 week membership symbol lists from the committed artifact (compact decision evidence). Full dumps recoverable from git history of this file pre-cleanup."
}
scripts/extend_snapshot_universe.py
+429
View File
@@ -0,0 +1,429 @@
"""Extend a *copy* of the production backtest snapshot with broad-universe OHLCV.
Research only — never writes to production Postgres.
Pipeline
--------
1. Copy ``--source`` snapshot (default ``backtest_snapshots/prod.sqlite``) to
``--output`` (default ``backtest_snapshots/research.sqlite``).
2. Resolve symbol pool = nasdaq_all ∪ sp500 via ``ticker_universe_service``.
3. Fetch ~5y daily bars from Alpaca for symbols missing (or short) in the copy.
4. Insert new tickers + OHLCV; mark them in side table ``research_rank_only``
so the harness can feed signal IC without GTL/candidate replay.
5. Write a **completion manifest** (``<output>.manifest.json``) with ticker /
OHLCV / rank_only counts and finished-at. Breadth runners refuse to start
without a matching complete manifest — same class of guard as calendar
truncation (see 2026-07-18 21:14 race: orphaned +0.0575 on a partial pool).
Resume-friendly: re-running skips symbols that already have ≥ ``--min-bars``.
A ``--limit`` smoke run writes ``complete: false`` so breadth mode still refuses.
Example
-------
python scripts/extend_snapshot_universe.py \\
--source backtest_snapshots/prod.sqlite \\
--output backtest_snapshots/research.sqlite \\
--force-copy
# smoke: first 50 missing symbols only
python scripts/extend_snapshot_universe.py --limit 50
"""
from __future__ import annotations
import argparse
import asyncio
import shutil
import sys
import time
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from sqlalchemy import create_engine, text
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
def _parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description=__doc__)
p.add_argument(
"--source",
default="backtest_snapshots/prod.sqlite",
help="Existing prod snapshot to copy (read-only after copy).",
)
p.add_argument(
"--output",
default="backtest_snapshots/research.sqlite",
help="Research snapshot path (created/updated).",
)
p.add_argument(
"--force-copy",
action="store_true",
help="Overwrite output by re-copying from source first.",
)
p.add_argument(
"--history-days",
type=int,
default=1825,
help="OHLCV lookback days (~5y). Default 1825.",
)
p.add_argument(
"--min-bars",
type=int,
default=260,
help="Skip re-fetch when a symbol already has this many bars.",
)
p.add_argument(
"--limit",
type=int,
default=None,
help="Max *new* symbols to fetch (smoke tests).",
)
p.add_argument(
"--sleep",
type=float,
default=0.15,
help="Seconds between Alpaca symbol requests (rate-limit cushion).",
)
p.add_argument(
"--max-retries",
type=int,
default=5,
help="Retries per symbol on RateLimitError.",
)
p.add_argument("--quiet", action="store_true")
return p.parse_args()
def _ensure_rank_only_table(engine) -> None:
"""DDL in its own connection/transaction (don't share with ORM Session)."""
with engine.begin() as conn:
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS research_rank_only (
ticker_id INTEGER PRIMARY KEY,
symbol TEXT NOT NULL UNIQUE
)
"""
)
)
async def _resolve_pool() -> tuple[list[str], dict[str, str]]:
"""Return sorted unique symbols and source labels.
Offline-safe: does **not** use production Postgres or SystemSetting cache
(those require a schema). Public sources first, then FMP, then seeds.
"""
from app.services.ticker_universe_service import (
_SEED_UNIVERSES,
_fetch_universe_symbols_from_fmp,
_fetch_universe_symbols_from_public,
_normalise_symbols,
)
sources: dict[str, str] = {}
symbols: set[str] = set()
for universe in ("nasdaq_all", "sp500"):
cleaned: list[str] = []
src = "none"
public_symbols, public_failures, public_source = (
await _fetch_universe_symbols_from_public(universe)
)
cleaned = _normalise_symbols(public_symbols)
if cleaned:
src = public_source or "public"
else:
if public_failures:
print(
f" WARNING: public fetch {universe}: "
f"{'; '.join(public_failures[:3])}"
)
try:
fmp_symbols = await _fetch_universe_symbols_from_fmp(universe)
cleaned = _normalise_symbols(fmp_symbols)
if cleaned:
src = "fmp"
except Exception as exc:
print(f" WARNING: FMP fetch {universe}: {exc}")
if not cleaned:
cleaned = _normalise_symbols(_SEED_UNIVERSES.get(universe, []))
if cleaned:
src = "seed"
print(
f" WARNING: {universe} fell back to seed list "
f"({len(cleaned)} symbols) — not full universe"
)
if not cleaned:
print(f" WARNING: universe {universe} returned no symbols")
continue
sources[universe] = src
symbols.update(cleaned)
print(f" {universe}: {len(cleaned)} symbols (source={src})")
return sorted(symbols), sources
async def _fetch_symbol_bars(
provider,
symbol: str,
start: date,
end: date,
*,
max_retries: int,
sleep_s: float,
) -> list:
from app.exceptions import ProviderError, RateLimitError
for attempt in range(max_retries):
try:
bars = await provider.fetch_ohlcv(symbol, start, end)
if sleep_s > 0:
await asyncio.sleep(sleep_s)
return bars
except RateLimitError:
wait = min(60.0, 2.0 ** attempt)
print(f" rate limited on {symbol}; sleep {wait:.0f}s")
await asyncio.sleep(wait)
except ProviderError as exc:
if attempt + 1 >= max_retries:
raise
await asyncio.sleep(1.0)
_ = exc
return []
async def _main() -> None:
# ROOT is already on sys.path; keep the helper import path-local.
from research_snapshot_manifest import ( # type: ignore[import-not-found]
clear_manifest,
write_completion_manifest,
)
args = _parse_args()
source = Path(args.source)
output = Path(args.output)
if not source.exists():
raise SystemExit(f"Source snapshot not found: {source}")
# Any rebuild/update invalidates prior completion until we finish cleanly.
clear_manifest(output)
if args.force_copy or not output.exists():
output.parent.mkdir(parents=True, exist_ok=True)
if output.exists():
output.unlink()
print(f"Copying {source} → {output}")
shutil.copy2(source, output)
else:
print(f"Updating existing research snapshot: {output}")
from app.config import settings
from app.providers.alpaca import AlpacaOHLCVProvider
if not settings.alpaca_api_key or not settings.alpaca_api_secret:
raise SystemExit("ALPACA_API_KEY / ALPACA_API_SECRET required in .env")
provider = AlpacaOHLCVProvider(settings.alpaca_api_key, settings.alpaca_api_secret)
end = date.today()
start = end - timedelta(days=int(args.history_days))
print("Resolving universe pool (nasdaq_all ∪ sp500)…")
pool, sources = await _resolve_pool()
print(f"Pool size: {len(pool)} (sources={sources})")
# Sync sqlite via raw SQL — one short transaction per symbol so a failed
# write never leaves the session in "transaction is inactive".
engine = create_engine(
f"sqlite:///{output.resolve().as_posix()}",
future=True,
)
_ensure_rank_only_table(engine)
with engine.connect() as conn:
existing_rows = conn.execute(
text("SELECT id, symbol FROM tickers")
).fetchall()
existing_ids = {str(sym): int(tid) for tid, sym in existing_rows}
prod_symbols = set(existing_ids)
bar_counts: dict[str, int] = {}
for sym, tid in existing_ids.items():
n = conn.execute(
text("SELECT COUNT(*) FROM ohlcv_records WHERE ticker_id = :tid"),
{"tid": tid},
).scalar_one()
bar_counts[sym] = int(n)
to_fetch: list[str] = []
for sym in pool:
if sym in existing_ids and bar_counts.get(sym, 0) >= args.min_bars:
continue
to_fetch.append(sym)
if args.limit is not None:
to_fetch = to_fetch[: max(0, int(args.limit))]
print(f"Symbols to fetch/extend: {len(to_fetch)}")
ok = 0
fail = 0
t0 = time.monotonic()
insert_ohlcv = text(
"""
INSERT INTO ohlcv_records
(ticker_id, date, open, high, low, close, volume, created_at)
VALUES
(:ticker_id, :date, :open, :high, :low, :close, :volume, :created_at)
"""
)
for index, sym in enumerate(to_fetch, 1):
try:
bars = await _fetch_symbol_bars(
provider,
sym,
start,
end,
max_retries=args.max_retries,
sleep_s=args.sleep,
)
except Exception as exc:
fail += 1
if not args.quiet:
print(f" [{index}/{len(to_fetch)}] {sym} FAIL {exc}")
continue
if not bars:
fail += 1
if not args.quiet:
print(f" [{index}/{len(to_fetch)}] {sym} empty")
continue
try:
with engine.begin() as write:
ticker_id = existing_ids.get(sym)
is_new = ticker_id is None
if is_new:
write.execute(
text(
"INSERT INTO tickers (symbol, name, created_at) "
"VALUES (:sym, NULL, :created)"
),
{
"sym": sym,
"created": datetime.now(timezone.utc).isoformat(),
},
)
ticker_id = int(
write.execute(
text("SELECT id FROM tickers WHERE symbol = :sym"),
{"sym": sym},
).scalar_one()
)
existing_ids[sym] = ticker_id
write.execute(
text(
"DELETE FROM ohlcv_records WHERE ticker_id = :tid "
"AND date >= :start AND date <= :end"
),
{
"tid": ticker_id,
"start": start.isoformat(),
"end": end.isoformat(),
},
)
now = datetime.now(timezone.utc).replace(tzinfo=None)
write.execute(
insert_ohlcv,
[
{
"ticker_id": ticker_id,
"date": b.date.isoformat(),
"open": float(b.open),
"high": float(b.high),
"low": float(b.low),
"close": float(b.close),
"volume": int(b.volume),
"created_at": now.isoformat(),
}
for b in bars
],
)
if is_new:
write.execute(
text(
"INSERT OR REPLACE INTO research_rank_only "
"(ticker_id, symbol) VALUES (:tid, :sym)"
),
{"tid": ticker_id, "sym": sym},
)
except Exception as exc:
fail += 1
if not args.quiet:
print(f" [{index}/{len(to_fetch)}] {sym} WRITE FAIL {exc}")
continue
ok += 1
if not args.quiet and (index % 25 == 0 or index == len(to_fetch)):
elapsed = time.monotonic() - t0
print(
f" progress {index}/{len(to_fetch)} ok={ok} fail={fail} "
f"elapsed={elapsed/60:.1f}m last={sym} bars={len(bars)}"
)
rank_only_n = conn.execute(
text("SELECT COUNT(*) FROM research_rank_only")
).scalar_one()
ticker_n = conn.execute(text("SELECT COUNT(*) FROM tickers")).scalar_one()
ohlcv_n = conn.execute(
text("SELECT COUNT(*) FROM ohlcv_records")
).scalar_one()
# Full planned work only when --limit is unset. Smoke runs stay incomplete
# so breadth mode cannot mythologize a 50-symbol toy pool.
is_complete = args.limit is None
manifest_path = write_completion_manifest(
output,
complete=is_complete,
sources=sources,
history_days=int(args.history_days),
min_bars=int(args.min_bars),
fetch_ok=ok,
fetch_fail=fail,
limit=args.limit,
extra={
"prod_symbols_at_start": len(prod_symbols),
"pool_size": len(pool),
"to_fetch": len(to_fetch),
},
)
print("Done.")
print(f" output: {output}")
print(f" tickers: {ticker_n}")
print(f" ohlcv rows: {ohlcv_n}")
print(f" research_rank_only: {rank_only_n}")
print(f" fetched ok/fail: {ok}/{fail}")
print(
f" completion manifest: {manifest_path} "
f"(complete={is_complete})"
)
if not is_complete:
print(
" NOTE: --limit set → complete=false; breadth runners will refuse "
"this snapshot until a full extend finishes."
)
if __name__ == "__main__":
asyncio.run(_main())
scripts/research_snapshot_manifest.py
+172
View File
@@ -0,0 +1,172 @@
"""Completion manifest for research.sqlite — cheap race guard.
The 2026-07-18 21:14 breadth run fired while ``extend_snapshot_universe`` was
still (or had just been) building the snapshot. Harness and shared-filter
recomputes agree on *complete* data, so the orphaned +0.0575 was incomplete
universe, not a code path bug.
Same class of protection as calendar-truncation assertions in the research
matrix: refuse to read results from a half-built artifact.
Layout
------
Sidecar path: ``<snapshot>.manifest.json`` next to the sqlite file
(e.g. ``backtest_snapshots/research.sqlite.manifest.json``).
"""
from __future__ import annotations
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from sqlalchemy import create_engine, text
MANIFEST_SCHEMA_VERSION = 1
def manifest_path_for(snapshot: Path) -> Path:
"""Sidecar path for a research snapshot."""
return Path(str(snapshot) + ".manifest.json")
def _count_snapshot(snapshot: Path) -> dict[str, int]:
engine = create_engine(
f"sqlite:///{snapshot.resolve().as_posix()}",
future=True,
)
try:
with engine.connect() as conn:
ticker_n = int(conn.execute(text("SELECT COUNT(*) FROM tickers")).scalar_one())
ohlcv_n = int(
conn.execute(text("SELECT COUNT(*) FROM ohlcv_records")).scalar_one()
)
try:
rank_only_n = int(
conn.execute(text("SELECT COUNT(*) FROM research_rank_only")).scalar_one()
)
except Exception:
rank_only_n = 0
finally:
engine.dispose()
return {
"ticker_count": ticker_n,
"ohlcv_row_count": ohlcv_n,
"rank_only_count": rank_only_n,
}
def write_completion_manifest(
snapshot: Path,
*,
complete: bool,
sources: dict[str, str] | None = None,
history_days: int | None = None,
min_bars: int | None = None,
fetch_ok: int | None = None,
fetch_fail: int | None = None,
limit: int | None = None,
extra: dict[str, Any] | None = None,
) -> Path:
"""Write (or overwrite) the sidecar completion manifest for *snapshot*."""
snapshot = Path(snapshot)
counts = _count_snapshot(snapshot) if snapshot.exists() else {
"ticker_count": 0,
"ohlcv_row_count": 0,
"rank_only_count": 0,
}
payload: dict[str, Any] = {
"schema_version": MANIFEST_SCHEMA_VERSION,
"snapshot": snapshot.name,
"snapshot_resolved": str(snapshot.resolve()) if snapshot.exists() else str(snapshot),
"complete": bool(complete),
"finished_at": datetime.now(timezone.utc).isoformat(),
**counts,
"sources": sources or {},
"history_days": history_days,
"min_bars": min_bars,
"fetch_ok": fetch_ok,
"fetch_fail": fetch_fail,
"limit": limit,
}
if extra:
payload["extra"] = extra
path = manifest_path_for(snapshot)
path.write_text(json.dumps(payload, indent=2, default=str) + "\n", encoding="utf-8")
return path
def clear_manifest(snapshot: Path) -> None:
"""Remove any existing completion manifest (start of a rebuild)."""
path = manifest_path_for(Path(snapshot))
if path.exists():
path.unlink()
def load_manifest(snapshot: Path) -> dict[str, Any] | None:
path = manifest_path_for(Path(snapshot))
if not path.exists():
return None
return json.loads(path.read_text(encoding="utf-8"))
def assert_research_snapshot_complete(snapshot: Path) -> dict[str, Any]:
"""Refuse breadth-mode work unless the extender finished cleanly.
Raises ``SystemExit`` with a clear message on any failure (missing
manifest, incomplete flag, or live counts that no longer match the
recorded totals — e.g. a mid-run overwrite of the sqlite file).
"""
snapshot = Path(snapshot)
if not snapshot.exists():
raise SystemExit(
f"Research snapshot missing: {snapshot}\n"
"Build it with: python scripts/extend_snapshot_universe.py"
)
path = manifest_path_for(snapshot)
if not path.exists():
raise SystemExit(
f"Research snapshot completion manifest missing: {path}\n"
"Refusing breadth run — this is the guard that would have caught "
"the 2026-07-18 21:14 race against a half-built research.sqlite.\n"
"Re-run extend_snapshot_universe.py to completion (no --limit), "
"or for a trusted existing full snapshot:\n"
" python -c \"from pathlib import Path; "
"from scripts.research_snapshot_manifest import write_completion_manifest; "
f"write_completion_manifest(Path(r'{snapshot}'), complete=True)\""
)
try:
manifest = json.loads(path.read_text(encoding="utf-8"))
except json.JSONDecodeError as exc:
raise SystemExit(f"Corrupt research snapshot manifest {path}: {exc}") from exc
if not manifest.get("complete"):
raise SystemExit(
f"Research snapshot marked incomplete in {path}\n"
f"(finished_at={manifest.get('finished_at')}, limit={manifest.get('limit')}).\n"
"Re-run extend_snapshot_universe.py without --limit until Done."
)
live = _count_snapshot(snapshot)
mismatches: list[str] = []
for key in ("ticker_count", "ohlcv_row_count", "rank_only_count"):
recorded = manifest.get(key)
if recorded is None:
mismatches.append(f"{key}: missing in manifest")
continue
if int(recorded) != int(live[key]):
mismatches.append(
f"{key}: manifest={recorded} live={live[key]}"
)
if mismatches:
raise SystemExit(
"Research snapshot does not match its completion manifest "
f"({path}). Likely a partial rewrite or concurrent extend:\n - "
+ "\n - ".join(mismatches)
+ "\nRe-run extend_snapshot_universe.py to completion."
)
return {**manifest, "live_counts": live}
scripts/run_fip_breadth_diagnostics.py
+669
View File
@@ -0,0 +1,669 @@
"""fip_id breadth diagnostics — single-sourced through harness mask helpers.
Uses the same collection + ``_filter_liquid_breadth_week_rich`` as
``run_backtest`` signal_eval. No parallel mask implementation.
Single-sourced liquid-breadth fip diagnostics through harness mask helpers.
Re-runs unconditional / tier / prod-subset / mom-conditional ICs and context
signals. Requires a complete research.sqlite completion manifest.
Research branch only. Example:
.\\.venv\\Scripts\\python.exe scripts\\run_fip_breadth_diagnostics.py ^
--research-snapshot backtest_snapshots\\research.sqlite ^
--prod-snapshot backtest_snapshots\\prod.sqlite ^
--workers 6 --allow-spawn
"""
from __future__ import annotations
import argparse
import json
import math
import multiprocessing as mp
import os
import sys
from collections import defaultdict
from concurrent.futures import ProcessPoolExecutor, as_completed
from datetime import date, datetime
from pathlib import Path
from typing import Any
from sqlalchemy import create_engine, text
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
# Match production signal_eval cadence / reliability bars.
MIN_CROSS = 20
MIN_RELIABLE = 12
MOM_WINNER_PCT = 80.0
def _parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("--research-snapshot", default="backtest_snapshots/research.sqlite")
p.add_argument("--prod-snapshot", default="backtest_snapshots/prod.sqlite")
p.add_argument("--top-n", type=int, default=1500)
p.add_argument("--min-price", type=float, default=5.0)
p.add_argument("--workers", type=int, default=max(1, (mp.cpu_count() or 4) - 1))
p.add_argument("--allow-spawn", action="store_true")
p.add_argument(
"--dump-weeks",
type=int,
default=0,
help="Weeks of liquid membership symbol lists to embed (default 0 — keep reports compact)",
)
p.add_argument("--out", default=None)
p.add_argument("--quiet", action="store_true")
return p.parse_args()
def _week_ord(wk: tuple[int, int]) -> int:
return int(wk[0]) * 53 + int(wk[1])
def _nonoverlap(weeks: list[tuple[int, int]], stride: int) -> list[tuple[int, int]]:
from app.services.backtest_service import _nonoverlapping_weeks
return _nonoverlapping_weeks(weeks, stride)
def _ic_from_weekly(
week_pairs: dict[tuple[int, int], list[tuple[float, float]]],
) -> dict[str, Any]:
from app.services.backtest_service import HORIZON, _spearman
stride = max(1, round(HORIZON / 5))
usable = [wk for wk, ps in week_pairs.items() if len(ps) >= MIN_CROSS]
kept = _nonoverlap(usable, stride)
ics: list[float] = []
sizes: list[int] = []
for wk in kept:
ps = week_pairs[wk]
if len(ps) < MIN_CROSS:
continue
ic = _spearman([p[0] for p in ps], [p[1] for p in ps])
if ic is not None:
ics.append(ic)
sizes.append(len(ps))
if not ics:
return {
"mean_ic": None,
"ic_t_stat": None,
"weeks": 0,
"avg_cross_section": None,
"ic_positive_pct": None,
"reliable": False,
}
mean_ic = sum(ics) / len(ics)
if len(ics) > 1:
var = sum((x - mean_ic) ** 2 for x in ics) / (len(ics) - 1)
std = math.sqrt(var) if var > 0 else 0.0
t_stat = mean_ic / std * math.sqrt(len(ics)) if std > 0 else None
else:
t_stat = None
return {
"mean_ic": round(mean_ic, 4),
"ic_t_stat": round(t_stat, 2) if t_stat is not None else None,
"weeks": len(ics),
"avg_cross_section": round(sum(sizes) / len(sizes), 1),
"ic_positive_pct": round(sum(1 for x in ics if x > 0) / len(ics) * 100, 1),
"reliable": len(ics) >= MIN_RELIABLE,
}
def _worker(payload: tuple) -> dict:
"""Return harness-style signal series for one ticker (liquid-mode dicts)."""
symbol, ords, opens, highs, lows, closes, volumes, spy = payload
from types import SimpleNamespace
from app.services.backtest_service import _signal_series
bars = [
SimpleNamespace(
date=date.fromordinal(int(o)),
open=float(op),
high=float(hi),
low=float(lo),
close=float(cl),
volume=float(vo),
)
for o, op, hi, lo, cl, vo in zip(ords, opens, highs, lows, closes, volumes)
]
return _signal_series(bars, spy, symbol=symbol)
def _load_spy(conn) -> dict[date, float]:
rows = conn.execute(
text("SELECT date, close FROM benchmark_prices WHERE symbol='SPY' ORDER BY date")
).fetchall()
out: dict[date, float] = {}
for d, c in rows:
if isinstance(d, str):
d = date.fromisoformat(d[:10])
out[d] = float(c)
return out
def _load_job(conn, symbol: str, spy: dict) -> tuple | None:
tid = conn.execute(
text("SELECT id FROM tickers WHERE symbol=:s"), {"s": symbol}
).scalar()
if tid is None:
return None
rows = conn.execute(
text(
"SELECT date, open, high, low, close, volume FROM ohlcv_records "
"WHERE ticker_id=:t ORDER BY date"
),
{"t": tid},
).fetchall()
if len(rows) < 90:
return None
ords, opens, highs, lows, closes, vols = [], [], [], [], [], []
for d, o, h, l, c, v in rows:
if isinstance(d, str):
d = date.fromisoformat(d[:10])
ords.append(d.toordinal())
opens.append(float(o))
highs.append(float(h))
lows.append(float(l))
closes.append(float(c))
vols.append(float(v or 0))
return (symbol, ords, opens, highs, lows, closes, vols, spy)
def main() -> None:
args = _parse_args()
research = Path(args.research_snapshot)
prod = Path(args.prod_snapshot)
# Refuse half-built research.sqlite (2026-07-18 21:14 race).
scripts_dir = Path(__file__).resolve().parent
if str(scripts_dir) not in sys.path:
sys.path.insert(0, str(scripts_dir))
from research_snapshot_manifest import ( # type: ignore[import-not-found]
assert_research_snapshot_complete,
)
manifest = assert_research_snapshot_complete(research)
if not args.quiet:
print(
f"Manifest ok: tickers={manifest.get('ticker_count')} "
f"ohlcv={manifest.get('ohlcv_row_count')} "
f"finished_at={manifest.get('finished_at')}",
flush=True,
)
# Force harness liquid-mode collection (same env as breadth run).
os.environ["BACKTEST_LIQUID_BREADTH"] = str(int(args.top_n))
os.environ["BACKTEST_LIQUID_MIN_PRICE"] = str(float(args.min_price))
if args.allow_spawn:
os.environ["BACKTEST_ALLOW_SPAWN"] = "1"
from app.services.backtest_service import (
HORIZON,
_filter_liquid_breadth_week_rich,
_liquid_breadth_week_stats,
_signal_evaluation,
)
eng = create_engine(f"sqlite:///{research.resolve().as_posix()}")
prod_symbols: set[str] = set()
if prod.exists():
peng = create_engine(f"sqlite:///{prod.resolve().as_posix()}")
with peng.connect() as c:
prod_symbols = {
str(r[0]) for r in c.execute(text("SELECT symbol FROM tickers"))
}
peng.dispose()
with eng.connect() as conn:
spy = _load_spy(conn)
symbols = [
str(r[0])
for r in conn.execute(text("SELECT symbol FROM tickers ORDER BY symbol"))
]
jobs = []
for i, sym in enumerate(symbols, 1):
job = _load_job(conn, sym, spy)
if job is not None:
jobs.append(job)
if not args.quiet and i % 500 == 0:
print(f" queued {i}/{len(symbols)}", flush=True)
if not args.quiet:
print(f"Collecting harness signal series for {len(jobs)} tickers…", flush=True)
collected: dict = defaultdict(lambda: defaultdict(list))
workers = max(1, int(args.workers))
def _merge(series: dict) -> None:
for name, weeks in series.items():
for wk, recs in weeks.items():
# week keys may arrive as lists after JSON; normalize to tuple
key = tuple(wk) if not isinstance(wk, tuple) else wk
collected[name][key].extend(recs)
if workers == 1:
for j, job in enumerate(jobs, 1):
_merge(_worker(job))
if not args.quiet and j % 200 == 0:
print(f" series {j}/{len(jobs)}", flush=True)
else:
ctx = mp.get_context("spawn") if args.allow_spawn or sys.platform == "win32" else None
with ProcessPoolExecutor(max_workers=workers, mp_context=ctx) as pool:
futs = [pool.submit(_worker, job) for job in jobs]
for j, fut in enumerate(as_completed(futs), 1):
try:
_merge(fut.result())
except Exception as exc:
if not args.quiet:
print(f" worker error: {exc}", flush=True)
if not args.quiet and j % 200 == 0:
print(f" series {j}/{len(jobs)}", flush=True)
# --- Harness signal_eval (authoritative unconditional ICs) ---
harness_rows = _signal_evaluation(dict(collected))
harness_by_name = {r["signal"]: r for r in harness_rows}
top_n = int(args.top_n)
min_price = float(args.min_price)
fip_weeks = collected.get("fip_id") or {}
mom_weeks = collected.get("mom_12_1") or {}
vol_weeks = collected.get("vol_6m") or {}
momr_weeks = collected.get("mom_12_1_resid") or {}
# Index mom/vol by (week, symbol) for joins
def _index(weeks_map: dict) -> dict[tuple, dict]:
out: dict[tuple, dict] = {}
for wk, recs in weeks_map.items():
key_wk = tuple(wk) if not isinstance(wk, tuple) else wk
for rec in recs:
if not isinstance(rec, dict):
continue
sym = rec.get("symbol")
if not sym:
continue
out[(key_wk, str(sym))] = rec
return out
mom_ix = _index(mom_weeks)
vol_ix = _index(vol_weeks)
momr_ix = _index(momr_weeks)
# Per-week membership + extended checks via shared rich filter
same_week: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
lag_week: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
tier_hi: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
tier_lo: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
prod_sub: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
mom_cond: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
vol_pairs: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
mom_pairs: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
momr_pairs: dict[tuple, list[tuple[float, float]]] = defaultdict(list)
ordered = sorted((tuple(w) for w in fip_weeks.keys()), key=_week_ord)
prev: dict[tuple, tuple] = {}
for i, wk in enumerate(ordered):
if i:
prev[wk] = ordered[i - 1]
# Prior-week dvol for lag: (symbol, week) from fip recs
dvol_sw: dict[tuple[str, tuple], float] = {}
for wk, recs in fip_weeks.items():
key_wk = tuple(wk) if not isinstance(wk, tuple) else wk
for rec in recs:
if isinstance(rec, dict) and rec.get("symbol") and rec.get("median_dvol_63"):
dvol_sw[(str(rec["symbol"]), key_wk)] = float(rec["median_dvol_63"])
membership_dumps: list[dict] = []
dump_count = 0
stride = max(1, round(HORIZON / 5))
dump_weeks = _nonoverlap(ordered, stride)[: max(0, int(args.dump_weeks))]
for wk_raw, recs in fip_weeks.items():
wk = tuple(wk_raw) if not isinstance(wk_raw, tuple) else wk_raw
stats = _liquid_breadth_week_stats(recs, top_n=top_n, min_price=min_price)
rich = _filter_liquid_breadth_week_rich(
recs, top_n=top_n, min_price=min_price
)
for rank, row in enumerate(rich, 1):
same_week[wk].append((float(row["val"]), float(row["fwd"])))
if rank <= 800:
tier_hi[wk].append((float(row["val"]), float(row["fwd"])))
elif rank <= top_n:
tier_lo[wk].append((float(row["val"]), float(row["fwd"])))
sym = row.get("symbol")
if sym and str(sym) in prod_symbols:
prod_sub[wk].append((float(row["val"]), float(row["fwd"])))
# Join mom for conditional
mrec = mom_ix.get((wk, str(sym))) if sym else None
if mrec is not None:
row["mom_12_1"] = mrec.get("val")
# Mom-conditional among liquid fip set
with_mom = [
r for r in rich
if r.get("mom_12_1") is not None or mom_ix.get((wk, str(r.get("symbol"))))
]
# ensure mom filled
for r in with_mom:
if r.get("mom_12_1") is None and r.get("symbol"):
m = mom_ix.get((wk, str(r["symbol"])))
if m is not None:
r["mom_12_1"] = m["val"]
with_mom = [r for r in rich if r.get("mom_12_1") is not None]
if len(with_mom) >= MIN_CROSS:
with_mom.sort(key=lambda r: float(r["mom_12_1"]))
cut = int(math.floor(len(with_mom) * (MOM_WINNER_PCT / 100.0)))
for r in with_mom[cut:]:
mom_cond[wk].append((float(r["val"]), float(r["fwd"])))
# Context signals via same shared filter on their own pools
for r in _filter_liquid_breadth_week_rich(
vol_weeks.get(wk_raw) or vol_weeks.get(wk) or [],
top_n=top_n,
min_price=min_price,
):
vol_pairs[wk].append((float(r["val"]), float(r["fwd"])))
for r in _filter_liquid_breadth_week_rich(
mom_weeks.get(wk_raw) or mom_weeks.get(wk) or [],
top_n=top_n,
min_price=min_price,
):
mom_pairs[wk].append((float(r["val"]), float(r["fwd"])))
for r in _filter_liquid_breadth_week_rich(
momr_weeks.get(wk_raw) or momr_weeks.get(wk) or [],
top_n=top_n,
min_price=min_price,
):
momr_pairs[wk].append((float(r["val"]), float(r["fwd"])))
# Lagged membership using prior week dvol on current fip pool
pw = prev.get(wk)
if pw is not None:
lagged_recs = []
for rec in recs:
if not isinstance(rec, dict) or not rec.get("symbol"):
continue
pdv = dvol_sw.get((str(rec["symbol"]), pw))
if pdv is None or pdv <= 0:
continue
# Clone with lag dvol for ranking
lagged_recs.append({
**rec,
"median_dvol_63": pdv,
})
for r in _filter_liquid_breadth_week_rich(
lagged_recs, top_n=top_n, min_price=min_price
):
lag_week[wk].append((float(r["val"]), float(r["fwd"])))
if wk in dump_weeks and dump_count < args.dump_weeks:
membership_dumps.append({
"week": list(wk),
"stats": stats,
"symbols": sorted(
str(r["symbol"]) for r in rich if r.get("symbol")
),
"n_symbols": len(rich),
})
dump_count += 1
# IC rows
checks = {
"fip_harness_signal_eval": {
"note": "Authoritative harness _signal_evaluation on collected fip_id",
**(harness_by_name.get("fip_id") or {}),
},
"fip_same_week_via_shared_filter": {
"note": "Same collected data, IC via shared _filter_liquid_breadth_week_rich",
**_ic_from_weekly(same_week),
},
"fip_lagged_membership_1w": {
"note": "Top-N by prior-week $vol on current fip pool (shared filter)",
**_ic_from_weekly(lag_week),
},
"fip_tier_1_800": {
"note": "Senior liquid ranks 1–800",
**_ic_from_weekly(tier_hi),
},
"fip_tier_801_1500": {
"note": "Junior liquid ranks 801–top_n",
**_ic_from_weekly(tier_lo),
},
"fip_prod_universe_subset": {
"note": "Prod.sqlite symbols inside liquid fip set",
**_ic_from_weekly(prod_sub),
},
"fip_momentum_conditional_top20pct": {
"note": (
f"Among liquid fip set, mom_12_1 ≥ P{MOM_WINNER_PCT:.0f} "
"(paper / gate-relevant)"
),
**_ic_from_weekly(mom_cond),
},
"vol_6m_liquid": {
"note": "vol_6m through shared filter",
**_ic_from_weekly(vol_pairs),
},
"mom_12_1_liquid": {
"note": "raw mom through shared filter",
**_ic_from_weekly(mom_pairs),
},
"mom_12_1_resid_liquid": {
"note": "residual mom through shared filter",
**_ic_from_weekly(momr_pairs),
},
}
h = checks["fip_harness_signal_eval"]
s = checks["fip_same_week_via_shared_filter"]
cond = checks["fip_momentum_conditional_top20pct"]
prod = checks["fip_prod_universe_subset"]
hi = checks["fip_tier_1_800"]
lo = checks["fip_tier_801_1500"]
lag = checks["fip_lagged_membership_1w"]
# Self-consistency: harness eval vs manual IC on same filter must match
harness_ic = h.get("mean_ic")
shared_ic = s.get("mean_ic")
consistent = (
harness_ic is not None
and shared_ic is not None
and abs(float(harness_ic) - float(shared_ic)) < 0.005
)
mom_alive = (
cond.get("mean_ic") is not None
and float(cond["mean_ic"]) < 0
and abs(float(cond["mean_ic"])) >= 0.03
and bool(cond.get("reliable"))
)
results = {
"generated_at": datetime.now().isoformat(),
"research_snapshot": str(research.resolve()),
"top_n": top_n,
"min_price": min_price,
"prod_subset_n": len(prod_symbols),
"panel_tickers": len(jobs),
"single_source": (
"diagnostics uses harness _signal_series + "
"_filter_liquid_breadth_week_rich only (no parallel mask)"
),
"avg_cross_section_semantics": (
"avg_cross_section = post-mask IC sample size. "
"avg_raw_pool = pre-filter observations. "
"avg_eligible_pre_mask = pass price+dvol before top-N. "
"mask_binds_pct = weeks where eligible_pre_mask > top_n."
),
"harness_self_consistent": consistent,
"checks": checks,
"membership_dumps": membership_dumps,
"interpretation": {
"harness_and_shared_filter_agree": consistent,
"mask_binds_pct": h.get("mask_binds_pct"),
"avg_eligible_pre_mask": h.get("avg_eligible_pre_mask"),
"avg_raw_pool": h.get("avg_raw_pool"),
"prod_subset_still_negative": (
prod.get("mean_ic") is not None and float(prod["mean_ic"]) < 0
),
"junior_tier_more_positive": (
lo.get("mean_ic") is not None
and hi.get("mean_ic") is not None
and float(lo["mean_ic"]) > float(hi["mean_ic"])
),
"lag_same_sign_as_same_week": (
lag.get("mean_ic") is not None
and s.get("mean_ic") is not None
and (float(lag["mean_ic"]) < 0) == (float(s["mean_ic"]) < 0)
),
"mom_conditional_negative_and_reliable": mom_alive,
"orphan_plus_five_sigma": (
"Orphaned 21:14 row (+0.0575 / t +5.12) raced a partial "
"research.sqlite and was removed from reports/ (Git history only). "
"Harness path and shared filter agree on complete data."
),
"compositional_story": (
"fip_id pools continuous winners (neg IC) vs continuous bleeders "
"(pos IC). Prod-subset and senior liquid stay negative; junior "
"liquid is less negative / positive — composition, not jumpiness premium."
),
"vol_tilt_warning": (
"Authoritative liquid vol_6m IC ≈ −0.048 / t ≈ −1.36 — directional "
"hypothesis only, not significant. Do not cite the orphaned −0.16 / "
"t −6.1. Re-validate production 80/20 high-vol tilt before any "
"universe broaden; it is not a settled finding on this pool."
),
"breadth_momentum_thesis": (
"Residual mom on liquid-1500 is +0.029 / t +1.33 vs fingerprint "
"0.055 / t 1.98 on 505 names — more breadth did not strengthen the "
"momentum t-stat on this pool. Clean mom edge lives in the large-cap "
"universe already traded. A fip tilt presupposes a breadth mom book "
"worth tilting; that baseline must be proven first."
),
},
"platform_verdict": (
"Mom-conditional fip ALIVE as book-tilt candidate only — requires a "
"pre-registered two-arm breadth book (baseline liquid-1500 mom vs +fip "
"tilt) before any gate talk. Unconditional fip not green. Production: none."
if mom_alive
else (
"fip CLOSED for production: mom-conditional does not clear iron rule "
"on single-sourced path. Display card is the resting place."
)
),
"research_snapshot_manifest": {
"finished_at": manifest.get("finished_at"),
"ticker_count": manifest.get("ticker_count"),
"ohlcv_row_count": manifest.get("ohlcv_row_count"),
"rank_only_count": manifest.get("rank_only_count"),
"complete": manifest.get("complete"),
},
}
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
out = Path(args.out) if args.out else Path("reports") / f"fip-reconcile-{stamp}.json"
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps(results, indent=2, default=str), encoding="utf-8")
# Append a machine reconciliation stub next to the JSON only — never clobber
# the curated research log at docs/research/fip-breadth-ic.md.
_update_md(out.with_suffix(".md"), results, out)
if not args.quiet:
print("=== Harness fip_id (authoritative) ===")
print(json.dumps(h, indent=2, default=str))
print("=== Shared-filter same-week (must match) ===")
print(json.dumps(s, indent=2, default=str))
print("=== Mom-conditional ===")
print(json.dumps(cond, indent=2, default=str))
print("self_consistent:", consistent)
print("platform_verdict:", results["platform_verdict"])
print(f"Wrote {out}")
def _update_md(path: Path, results: dict, artifact: Path) -> None:
checks = results["checks"]
interp = results["interpretation"]
h = checks.get("fip_harness_signal_eval") or {}
lines = [
"",
"---",
"",
f"## Reconciliation ({results['generated_at'][:10]})",
"",
"### Problem",
"",
"Machine stub only — curated narrative lives in `docs/research/fip-breadth-ic.md`.",
"",
f"- **Single source:** {results.get('single_source')}",
f"- Harness vs shared-filter agree: "
f"**{interp.get('harness_and_shared_filter_agree')}**",
"",
"### Authoritative unconditional fip (liquid top-N, post-mask)",
"",
f"| metric | value |",
f"|---|---|",
f"| mean_ic | {h.get('mean_ic')} |",
f"| ic_t_stat | {h.get('ic_t_stat')} |",
f"| weeks | {h.get('weeks')} |",
f"| avg_cross_section (post-mask) | {h.get('avg_cross_section')} |",
f"| avg_raw_pool | {h.get('avg_raw_pool')} |",
f"| avg_eligible_pre_mask | {h.get('avg_eligible_pre_mask')} |",
f"| mask_binds_pct | {h.get('mask_binds_pct')} |",
f"| reliable | {h.get('reliable')} |",
"",
"### Checks (single-sourced)",
"",
"| check | mean_ic | t | weeks | avg N | reliable |",
"|---|---:|---:|---:|---:|---|",
]
for key in [
"fip_harness_signal_eval",
"fip_same_week_via_shared_filter",
"fip_lagged_membership_1w",
"fip_tier_1_800",
"fip_tier_801_1500",
"fip_prod_universe_subset",
"fip_momentum_conditional_top20pct",
"vol_6m_liquid",
"mom_12_1_liquid",
"mom_12_1_resid_liquid",
]:
row = checks.get(key) or {}
lines.append(
f"| {key} | {row.get('mean_ic')} | {row.get('ic_t_stat')} | "
f"{row.get('weeks')} | {row.get('avg_cross_section')} | {row.get('reliable')} |"
)
lines.extend([
"",
"### Flags",
"",
f"- Prod subset still negative: **{interp.get('prod_subset_still_negative')}**",
f"- Junior tier more positive than senior: **{interp.get('junior_tier_more_positive')}**",
f"- Lag same sign as same-week: **{interp.get('lag_same_sign_as_same_week')}**",
f"- Mom-conditional negative + reliable: **{interp.get('mom_conditional_negative_and_reliable')}**",
"",
"### Platform verdict (post-reconciliation)",
"",
results.get("platform_verdict", ""),
"",
"### Vol-tilt / breadth-momentum notes",
"",
interp.get("vol_tilt_warning", ""),
"",
interp.get("breadth_momentum_thesis", ""),
"",
f"Artifact: `{artifact.as_posix()}`",
"",
])
# Always overwrite the machine stub (never the curated research log).
path.write_text("\n".join(lines).lstrip() + "\n", encoding="utf-8")
if __name__ == "__main__":
main()
scripts/run_fip_breadth_research.py
+315
View File
@@ -0,0 +1,315 @@
"""Phase B: fip_id IC on liquid-breadth cross-section (local research only).
1. Fingerprint check on the unextended prod snapshot (must ≈ IC −0.045 / t −2.9).
2. Assert research.sqlite has a matching **completion manifest** (race guard).
3. Run signal_eval on research.sqlite with BACKTEST_LIQUID_BREADTH=1500 PIT mask.
4. Write a research report under docs/research/ and reports/.
Does not modify production DB, gate, scanner, or schedule.
Example
-------
# After extend_snapshot_universe.py has built research.sqlite:
python scripts/run_fip_breadth_research.py \\
--prod-snapshot backtest_snapshots/prod.sqlite \\
--research-snapshot backtest_snapshots/research.sqlite \\
--workers 6 --allow-spawn
"""
from __future__ import annotations
import argparse
import asyncio
import json
import os
import sys
from datetime import datetime
from pathlib import Path
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))
FINGERPRINT_IC = -0.045
FINGERPRINT_T = -2.9
FINGERPRINT_IC_TOL = 0.015
FINGERPRINT_T_TOL = 0.6
def _sqlite_url(path: Path) -> str:
return f"sqlite+aiosqlite:///{path.resolve().as_posix()}"
def _parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("--prod-snapshot", default="backtest_snapshots/prod.sqlite")
p.add_argument("--research-snapshot", default="backtest_snapshots/research.sqlite")
p.add_argument("--workers", type=int, default=6)
p.add_argument("--allow-spawn", action="store_true")
p.add_argument("--skip-fingerprint", action="store_true")
p.add_argument("--skip-research", action="store_true")
p.add_argument("--liquid-breadth", type=int, default=1500)
p.add_argument("--min-price", type=float, default=5.0)
p.add_argument(
"--out",
default=None,
help="JSON report path (default reports/fip-breadth-YYYYMMDD.json)",
)
p.add_argument("--quiet", action="store_true")
return p.parse_args()
def _find_fip(signal_eval: list[dict]) -> dict | None:
for row in signal_eval or []:
if row.get("signal") == "fip_id":
return row
return None
def _verdict(row: dict | None) -> dict:
if row is None:
return {
"green": False,
"reason": "fip_id missing from signal_eval",
}
mean_ic = row.get("mean_ic")
t_stat = row.get("ic_t_stat")
reliable = bool(row.get("reliable"))
if mean_ic is None or t_stat is None:
return {"green": False, "reason": "missing mean_ic or ic_t_stat", "row": row}
sign_ok = mean_ic < 0
mag_ok = abs(float(mean_ic)) >= 0.03
green = sign_ok and mag_ok and reliable
return {
"green": green,
"reason": (
"iron rule cleared — follow-up proposal only, not production wire-in"
if green
else "iron rule not met on liquid-breadth cross-section"
),
"checks": {
"mean_ic": mean_ic,
"abs_mean_ic_ge_0_03": mag_ok,
"sign_negative": sign_ok,
"ic_t_stat": t_stat,
"reliable": reliable,
"weeks": row.get("weeks"),
"avg_cross_section": row.get("avg_cross_section"),
},
"row": row,
}
async def _run_signal_eval(snapshot: Path, *, workers: int, quiet: bool) -> dict:
from app.config import settings
from app.services.backtest_service import run_backtest
settings.backtest_workers = workers
engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
def progress(done: int, total: int, symbol: str) -> None:
if quiet:
return
print(f" progress {done}/{total} {symbol}", end="\r")
try:
async with Session() as db:
report = await run_backtest(db, progress_cb=progress, cadence="weekly")
finally:
await engine.dispose()
if not quiet:
print()
return report
def _write_md(path: Path, payload: dict) -> None:
fp = payload.get("fingerprint") or {}
br = payload.get("breadth") or {}
v = payload.get("verdict") or {}
lines = [
"# Broad-universe fip_id IC research (Phase B)",
"",
f"Generated: {payload.get('generated_at')}",
"",
"## Scope",
"",
"- **Research only** — production universe, gate, scanner, schedule unchanged.",
"- Price-only signal harness; no sentiment/fundamentals on the broad tier.",
"- Point-in-time liquidity mask: top "
f"**{payload.get('liquid_breadth_top_n')}** by 63d median $vol, "
f"price ≥ **${payload.get('liquid_min_price')}** at as-of.",
"",
"## Caveats",
"",
"- **Survivorship bias**: today's constituents backfilled historically "
"(worse in small caps).",
"- **IEX volume undercount**: relative $vol rank only, not absolute floors.",
"- **Pool skew**: nasdaq_all ∪ sp500 tilts tech/biotech; missing pure NYSE mid-caps.",
"",
"## Fingerprint (505-name prod snapshot)",
"",
f"- Expected: IC ≈ {FINGERPRINT_IC}, t ≈ {FINGERPRINT_T}",
f"- Observed: IC = {fp.get('mean_ic')}, t = {fp.get('ic_t_stat')}, "
f"weeks = {fp.get('weeks')}, reliable = {fp.get('reliable')}",
f"- Pass: **{fp.get('pass')}**",
"",
"## Liquid-breadth signal_eval (fip_id)",
"",
]
row = br.get("row") or br
if row:
lines.extend([
f"| metric | value |",
f"|---|---|",
f"| mean_ic | {row.get('mean_ic')} |",
f"| ic_t_stat | {row.get('ic_t_stat')} |",
f"| ic_positive_pct | {row.get('ic_positive_pct')} |",
f"| weeks | {row.get('weeks')} |",
f"| avg_cross_section | {row.get('avg_cross_section')} |",
f"| reliable | {row.get('reliable')} |",
f"| mean_quintile_spread | {row.get('mean_quintile_spread')} |",
"",
])
else:
lines.append("_No breadth result (run skipped or failed)._")
lines.append("")
lines.extend([
"## Verdict (iron rule)",
"",
f"- **Green: {v.get('green')}**",
f"- {v.get('reason')}",
f"- Checks: `{json.dumps(v.get('checks') or {}, default=str)}`",
"",
"A green verdict authorizes a **follow-up proposal** only "
"(two-tier universe / gate revalidation) — **not** production wire-in.",
"",
"## Artifacts",
"",
f"- Fingerprint report: `{payload.get('fingerprint_report_path')}`",
f"- Breadth report: `{payload.get('breadth_report_path')}`",
"",
])
path.write_text("\n".join(lines), encoding="utf-8")
async def _main() -> None:
args = _parse_args()
prod = Path(args.prod_snapshot)
research = Path(args.research_snapshot)
if not prod.exists():
raise SystemExit(f"Prod snapshot missing: {prod}")
os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1"
if args.allow_spawn:
os.environ["BACKTEST_ALLOW_SPAWN"] = "1"
os.environ["BACKTEST_SIGNAL_EVAL_ONLY"] = "1"
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
out_json = Path(args.out) if args.out else Path("reports") / f"fip-breadth-{stamp}.json"
out_json.parent.mkdir(parents=True, exist_ok=True)
# Never clobber the curated research log (docs/research/fip-breadth-ic.md).
# Machine summary goes next to the JSON report only.
out_md = out_json.with_suffix(".md")
payload: dict = {
"generated_at": datetime.now().isoformat(),
"liquid_breadth_top_n": args.liquid_breadth,
"liquid_min_price": args.min_price,
"fingerprint": None,
"breadth": None,
"verdict": None,
}
# --- 1) Fingerprint ---
if not args.skip_fingerprint:
# Clear liquid breadth for fingerprint
os.environ.pop("BACKTEST_LIQUID_BREADTH", None)
os.environ.pop("BACKTEST_LIQUID_MIN_PRICE", None)
if not args.quiet:
print(f"Fingerprint run on {prod}…")
fp_report = await _run_signal_eval(prod, workers=args.workers, quiet=args.quiet)
fp_path = out_json.with_name(out_json.stem + "-fingerprint.json")
fp_path.write_text(json.dumps(fp_report, indent=2, default=str), encoding="utf-8")
fip = _find_fip(fp_report.get("signal_eval") or [])
if fip is None:
raise SystemExit("ABORT: fip_id missing from fingerprint signal_eval")
ic_ok = abs(float(fip["mean_ic"]) - FINGERPRINT_IC) <= FINGERPRINT_IC_TOL
t_ok = abs(float(fip["ic_t_stat"]) - FINGERPRINT_T) <= FINGERPRINT_T_TOL
passed = ic_ok and t_ok and bool(fip.get("reliable"))
payload["fingerprint"] = {
**fip,
"pass": passed,
"expected_ic": FINGERPRINT_IC,
"expected_t": FINGERPRINT_T,
}
payload["fingerprint_report_path"] = str(fp_path)
if not args.quiet:
print(
f"Fingerprint fip_id IC={fip.get('mean_ic')} t={fip.get('ic_t_stat')} "
f"pass={passed}"
)
if not passed:
out_json.write_text(json.dumps(payload, indent=2, default=str), encoding="utf-8")
raise SystemExit(
"ABORT: fingerprint mismatch — investigate before trusting breadth runs "
f"(got IC={fip.get('mean_ic')} t={fip.get('ic_t_stat')})"
)
# --- 2) Breadth ---
if not args.skip_research:
# Refuse half-built research.sqlite (2026-07-18 21:14 race).
scripts_dir = Path(__file__).resolve().parent
if str(scripts_dir) not in sys.path:
sys.path.insert(0, str(scripts_dir))
from research_snapshot_manifest import ( # type: ignore[import-not-found]
assert_research_snapshot_complete,
)
manifest = assert_research_snapshot_complete(research)
payload["research_snapshot_manifest"] = {
"finished_at": manifest.get("finished_at"),
"ticker_count": manifest.get("ticker_count"),
"ohlcv_row_count": manifest.get("ohlcv_row_count"),
"rank_only_count": manifest.get("rank_only_count"),
"complete": manifest.get("complete"),
}
os.environ["BACKTEST_LIQUID_BREADTH"] = str(int(args.liquid_breadth))
os.environ["BACKTEST_LIQUID_MIN_PRICE"] = str(float(args.min_price))
if not args.quiet:
print(
f"Breadth run on {research} "
f"(top {args.liquid_breadth}, min_price={args.min_price}; "
f"manifest ok tickers={manifest.get('ticker_count')} "
f"finished_at={manifest.get('finished_at')})…"
)
br_report = await _run_signal_eval(
research, workers=args.workers, quiet=args.quiet
)
br_path = out_json.with_name(out_json.stem + "-breadth.json")
br_path.write_text(json.dumps(br_report, indent=2, default=str), encoding="utf-8")
fip_b = _find_fip(br_report.get("signal_eval") or [])
payload["breadth"] = fip_b or {"error": "fip_id missing"}
payload["breadth_report_path"] = str(br_path)
payload["breadth_tickers"] = br_report.get("tickers")
payload["breadth_rank_only_tickers"] = br_report.get("rank_only_tickers")
payload["verdict"] = _verdict(fip_b)
if not args.quiet:
print(
f"Breadth fip_id IC={ (fip_b or {}).get('mean_ic') } "
f"t={ (fip_b or {}).get('ic_t_stat') } "
f"green={payload['verdict'].get('green')}"
)
out_json.write_text(json.dumps(payload, indent=2, default=str), encoding="utf-8")
out_md.parent.mkdir(parents=True, exist_ok=True)
_write_md(out_md, payload)
if not args.quiet:
print(f"Wrote {out_json}")
print(f"Wrote {out_md}")
if __name__ == "__main__":
asyncio.run(_main())
tests/unit/test_research_snapshot_manifest.py
+133
View File
@@ -0,0 +1,133 @@
"""Completion-manifest guard for research.sqlite breadth runs."""
from __future__ import annotations
import json
import sys
from pathlib import Path
import pytest
from sqlalchemy import create_engine, text
ROOT = Path(__file__).resolve().parents[2]
SCRIPTS = ROOT / "scripts"
if str(SCRIPTS) not in sys.path:
sys.path.insert(0, str(SCRIPTS))
from research_snapshot_manifest import ( # noqa: E402
assert_research_snapshot_complete,
clear_manifest,
load_manifest,
manifest_path_for,
write_completion_manifest,
)
def _tiny_research_db(path: Path, *, tickers: int = 3, bars_each: int = 5) -> None:
engine = create_engine(f"sqlite:///{path.resolve().as_posix()}", future=True)
with engine.begin() as conn:
conn.execute(
text(
"CREATE TABLE tickers ("
"id INTEGER PRIMARY KEY, symbol TEXT NOT NULL UNIQUE, "
"name TEXT, created_at TEXT)"
)
)
conn.execute(
text(
"CREATE TABLE ohlcv_records ("
"id INTEGER PRIMARY KEY, ticker_id INTEGER, date TEXT, "
"open REAL, high REAL, low REAL, close REAL, volume INTEGER, "
"created_at TEXT)"
)
)
conn.execute(
text(
"CREATE TABLE research_rank_only ("
"ticker_id INTEGER PRIMARY KEY, symbol TEXT NOT NULL UNIQUE)"
)
)
for i in range(tickers):
sym = f"T{i}"
conn.execute(
text(
"INSERT INTO tickers (id, symbol, name, created_at) "
"VALUES (:id, :sym, NULL, '2026-01-01')"
),
{"id": i + 1, "sym": sym},
)
if i > 0:
conn.execute(
text(
"INSERT INTO research_rank_only (ticker_id, symbol) "
"VALUES (:id, :sym)"
),
{"id": i + 1, "sym": sym},
)
for d in range(bars_each):
conn.execute(
text(
"INSERT INTO ohlcv_records "
"(ticker_id, date, open, high, low, close, volume, created_at) "
"VALUES (:tid, :date, 1,1,1,1,100, '2026-01-01')"
),
{"tid": i + 1, "date": f"2026-01-{d+1:02d}"},
)
engine.dispose()
def test_write_and_assert_complete(tmp_path: Path) -> None:
snap = tmp_path / "research.sqlite"
_tiny_research_db(snap)
path = write_completion_manifest(snap, complete=True, sources={"t": "unit"})
assert path == manifest_path_for(snap)
assert path.exists()
m = assert_research_snapshot_complete(snap)
assert m["complete"] is True
assert m["ticker_count"] == 3
assert m["ohlcv_row_count"] == 15
assert m["rank_only_count"] == 2
assert m["live_counts"]["ticker_count"] == 3
def test_refuse_missing_manifest(tmp_path: Path) -> None:
snap = tmp_path / "research.sqlite"
_tiny_research_db(snap)
with pytest.raises(SystemExit, match="manifest missing"):
assert_research_snapshot_complete(snap)
def test_refuse_incomplete_flag(tmp_path: Path) -> None:
snap = tmp_path / "research.sqlite"
_tiny_research_db(snap)
write_completion_manifest(snap, complete=False, limit=50)
with pytest.raises(SystemExit, match="marked incomplete"):
assert_research_snapshot_complete(snap)
def test_refuse_count_mismatch(tmp_path: Path) -> None:
snap = tmp_path / "research.sqlite"
_tiny_research_db(snap)
write_completion_manifest(snap, complete=True)
# Tamper: change live DB after manifest written
engine = create_engine(f"sqlite:///{snap.resolve().as_posix()}", future=True)
with engine.begin() as conn:
conn.execute(
text(
"INSERT INTO tickers (id, symbol, name, created_at) "
"VALUES (99, 'EXTRA', NULL, '2026-01-01')"
)
)
engine.dispose()
with pytest.raises(SystemExit, match="does not match"):
assert_research_snapshot_complete(snap)
def test_clear_manifest(tmp_path: Path) -> None:
snap = tmp_path / "research.sqlite"
_tiny_research_db(snap)
write_completion_manifest(snap, complete=True)
assert load_manifest(snap) is not None
clear_manifest(snap)
assert load_manifest(snap) is None