diff --git a/app/services/fundamentals_derivation.py b/app/services/fundamentals_derivation.py index e5d203c..3fada32 100644 --- a/app/services/fundamentals_derivation.py +++ b/app/services/fundamentals_derivation.py @@ -20,7 +20,7 @@ Rules: from __future__ import annotations from dataclasses import dataclass, field -from datetime import date +from datetime import date, datetime, timezone from typing import Any, Iterable _FP_TO_Q = {"Q1": 1, "Q2": 2, "Q3": 3, "FY": 4} @@ -30,7 +30,12 @@ TAPE_LEN = 4 # quarter-tape length # Duration (flow) fields differenced from YTD into discrete quarters + summed to TTM. _FLOW_FIELDS = ( - "revenue", "net_income", "operating_income", "diluted_eps", "cfo", "capex", + "revenue", + "net_income", + "operating_income", + "diluted_eps", + "cfo", + "capex", "depreciation_amortization", ) @@ -44,7 +49,9 @@ class MetricPoint: @dataclass class MetricSeries: value: float | None = None - history: list[MetricPoint] = field(default_factory=list) # oldest -> newest, <= TAPE_LEN + history: list[MetricPoint] = field( + default_factory=list + ) # oldest -> newest, <= TAPE_LEN period_end: date | None = None filed_date: date | None = None @@ -82,7 +89,9 @@ def derive(snapshots: Iterable[Any]) -> DerivedFundamentals: result.ttm_diluted_eps = _ttm(discrete["diluted_eps"], *latest) ttm_cfo = _ttm(discrete["cfo"], *latest) ttm_capex = _ttm(discrete["capex"], *latest) - result.ttm_fcf = None if ttm_cfo is None or ttm_capex is None else ttm_cfo - ttm_capex + result.ttm_fcf = ( + None if ttm_cfo is None or ttm_capex is None else ttm_cfo - ttm_capex + ) # tape = the CONSECUTIVE run of up to TAPE_LEN quarters ending at the latest, # stopping at a gap — so trend text never compares non-adjacent periods. @@ -90,7 +99,9 @@ def derive(snapshots: Iterable[Any]) -> DerivedFundamentals: result.metrics = { "revenue_growth_yoy": _yoy_growth_series(discrete["revenue"], selected, tape), "eps_growth_yoy": _yoy_growth_series(discrete["diluted_eps"], selected, tape), - "operating_margin": _margin_series(discrete["operating_income"], discrete["revenue"], selected, tape), + "operating_margin": _margin_series( + discrete["operating_income"], discrete["revenue"], selected, tape + ), "fcf_margin": _fcf_margin_series(discrete, selected, tape), "net_debt": _instant_series(selected, tape, _net_debt), "net_debt_to_ebitda": _leverage_series(selected, discrete, tape), @@ -102,8 +113,27 @@ def derive(snapshots: Iterable[Any]) -> DerivedFundamentals: return result +def derive_as_of(snapshots: Iterable[Any], as_of: datetime) -> DerivedFundamentals: + """Derive using only SEC filings accepted by the historical cutoff.""" + cutoff = _utc_datetime(as_of) + visible = ( + row + for row in snapshots + if (accepted := getattr(row, "accepted_at", None)) is not None + and _utc_datetime(accepted) <= cutoff + ) + return derive(visible) + + +def _utc_datetime(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + # -- period selection -------------------------------------------------------- + def _select_latest_per_period(snapshots: Iterable[Any]) -> dict[tuple[int, str], Any]: best: dict[tuple[int, str], Any] = {} for row in snapshots: @@ -126,7 +156,9 @@ def _ordered_quarters(selected: dict[tuple[int, str], Any]) -> list[tuple[int, i return sorted((fy, _FP_TO_Q[fp]) for (fy, fp) in selected) -def _consecutive_suffix(quarters: list[tuple[int, int]], n: int) -> list[tuple[int, int]]: +def _consecutive_suffix( + quarters: list[tuple[int, int]], n: int +) -> list[tuple[int, int]]: """The run of up to n quarters ending at the latest, walking back only through adjacent periods (stop at the first gap). Returned oldest -> newest.""" if not quarters: @@ -146,7 +178,10 @@ def _consecutive_suffix(quarters: list[tuple[int, int]], n: int) -> list[tuple[i # -- discrete + TTM ---------------------------------------------------------- -def _discrete_quarters(selected: dict[tuple[int, str], Any], field_name: str) -> dict[tuple[int, int], float]: + +def _discrete_quarters( + selected: dict[tuple[int, str], Any], field_name: str +) -> dict[tuple[int, int], float]: out: dict[tuple[int, int], float] = {} for (fy, fp), row in selected.items(): val = _discrete_value(selected, fy, fp, field_name) @@ -189,6 +224,7 @@ def _pct_change(cur: float | None, prior: float | None) -> float | None: # -- per-metric series (value at latest + tape history) ---------------------- + def _period_end(selected, fy: int, q: int) -> date | None: row = selected.get((fy, _Q_TO_FP[q])) return row.period_end if row is not None else None @@ -196,7 +232,7 @@ def _period_end(selected, fy: int, q: int) -> date | None: def _yoy_growth_series(dq, selected, tape) -> MetricSeries: pts = [] - for (fy, q) in tape: + for fy, q in tape: cur, prior = _ttm(dq, fy, q), _ttm(dq, fy - 1, q) pts.append(MetricPoint(_period_end(selected, fy, q), _pct_change(cur, prior))) return _series(pts) @@ -204,7 +240,7 @@ def _yoy_growth_series(dq, selected, tape) -> MetricSeries: def _margin_series(num_dq, den_dq, selected, tape) -> MetricSeries: pts = [] - for (fy, q) in tape: + for fy, q in tape: num, den = _ttm(num_dq, fy, q), _ttm(den_dq, fy, q) val = None if num is None or not den else num / den * 100.0 pts.append(MetricPoint(_period_end(selected, fy, q), val)) @@ -213,24 +249,38 @@ def _margin_series(num_dq, den_dq, selected, tape) -> MetricSeries: def _fcf_margin_series(discrete, selected, tape) -> MetricSeries: pts = [] - for (fy, q) in tape: - cfo, capex, rev = _ttm(discrete["cfo"], fy, q), _ttm(discrete["capex"], fy, q), _ttm(discrete["revenue"], fy, q) - val = None if cfo is None or capex is None or not rev else (cfo - capex) / rev * 100.0 + for fy, q in tape: + cfo, capex, rev = ( + _ttm(discrete["cfo"], fy, q), + _ttm(discrete["capex"], fy, q), + _ttm(discrete["revenue"], fy, q), + ) + val = ( + None + if cfo is None or capex is None or not rev + else (cfo - capex) / rev * 100.0 + ) pts.append(MetricPoint(_period_end(selected, fy, q), val)) return _series(pts) def _instant_series(selected, tape, fn) -> MetricSeries: - pts = [MetricPoint(_period_end(selected, fy, q), fn(selected.get((fy, _Q_TO_FP[q])))) for (fy, q) in tape] + pts = [ + MetricPoint(_period_end(selected, fy, q), fn(selected.get((fy, _Q_TO_FP[q])))) + for (fy, q) in tape + ] return _series(pts) def _leverage_series(selected, discrete, tape) -> MetricSeries: pts = [] - for (fy, q) in tape: + for fy, q in tape: row = selected.get((fy, _Q_TO_FP[q])) nd = _net_debt(row) - op, da = _ttm(discrete["operating_income"], fy, q), _ttm(discrete["depreciation_amortization"], fy, q) + op, da = ( + _ttm(discrete["operating_income"], fy, q), + _ttm(discrete["depreciation_amortization"], fy, q), + ) ebitda = None if op is None or da is None else op + da # Null when EBITDA <= 0: a negative denominator would flip polarity and a # "lower is better" read would rank a distressed issuer as favorable. @@ -241,7 +291,7 @@ def _leverage_series(selected, discrete, tape) -> MetricSeries: def _share_change_series(selected, tape) -> MetricSeries: pts = [] - for (fy, q) in tape: + for fy, q in tape: cur = _shares(selected.get((fy, _Q_TO_FP[q]))) prior = _shares(selected.get((fy - 1, _Q_TO_FP[q]))) pts.append(MetricPoint(_period_end(selected, fy, q), _pct_change(cur, prior))) diff --git a/app/services/fundamentals_research.py b/app/services/fundamentals_research.py new file mode 100644 index 0000000..f2a6fde --- /dev/null +++ b/app/services/fundamentals_research.py @@ -0,0 +1,157 @@ +"""Pure scoring helpers for point-in-time fundamentals research. + +The runner converts a CIK-deduplicated cross-section into favorable 0..100 +factor ranks and three deliberately small composites. Historical valuation is +absent: stored bars are split-adjusted, while filing-time EPS and share counts +are not guaranteed to be on today's split basis. +""" + +from __future__ import annotations + +import math +from collections.abc import Mapping +from typing import Any + +MIN_CROSS_SECTION = 5 + +FACTOR_POLARITY: dict[str, bool] = { + "revenue_growth_yoy": True, + "eps_growth_yoy": True, + "operating_margin": True, + "fcf_margin": True, + "net_debt_to_ebitda": False, + "share_count_change_yoy": False, +} + +QUALITY_FACTORS = ( + "operating_margin", + "fcf_margin", + "net_debt_to_ebitda", + "share_count_change_yoy", +) +GROWTH_FACTORS = ("revenue_growth_yoy", "eps_growth_yoy") +COMPOSITE_KEYS = ("quality", "growth", "balanced") + + +def raw_features(derived: Any) -> dict[str, float | None]: + """Extract the six research-eligible values from derived fundamentals.""" + metrics = getattr(derived, "metrics", {}) or {} + return { + key: _finite_or_none(getattr(metrics.get(key), "value", None)) + for key in FACTOR_POLARITY + } + + +def cross_section_scores( + features_by_issuer: Mapping[str, Mapping[str, Any]], + *, + min_cross_section: int = MIN_CROSS_SECTION, +) -> dict[str, dict[str, float | None]]: + """Return favorable factor ranks and composites for every issuer. + + Quality needs two of four inputs; growth needs one of two. Balanced requires + both sub-scores and weights them equally, so quality's four inputs do not + mechanically dominate growth's two inputs. + """ + result = { + str(issuer): { + **{key: None for key in FACTOR_POLARITY}, + **{key: None for key in COMPOSITE_KEYS}, + } + for issuer in features_by_issuer + } + + for factor, higher_is_better in FACTOR_POLARITY.items(): + values = { + str(issuer): _finite_or_none(features.get(factor)) + for issuer, features in features_by_issuer.items() + } + ranks = favorable_percentiles( + values, + higher_is_better=higher_is_better, + min_count=min_cross_section, + ) + for issuer, rank in ranks.items(): + result[issuer][factor] = rank + + for scores in result.values(): + quality_values = _available(scores, QUALITY_FACTORS) + growth_values = _available(scores, GROWTH_FACTORS) + if len(quality_values) >= 2: + scores["quality"] = _mean(quality_values) + if growth_values: + scores["growth"] = _mean(growth_values) + if scores["quality"] is not None and scores["growth"] is not None: + scores["balanced"] = _mean( + [float(scores["quality"]), float(scores["growth"])] + ) + return result + + +def favorable_percentiles( + values_by_issuer: Mapping[str, Any], + *, + higher_is_better: bool, + min_count: int = MIN_CROSS_SECTION, +) -> dict[str, float | None]: + """Tie-aware favorable percentile for a deduplicated cross-section.""" + valid = { + str(issuer): float(value) + for issuer, value in values_by_issuer.items() + if _finite_or_none(value) is not None + } + result: dict[str, float | None] = {str(issuer): None for issuer in values_by_issuer} + if len(valid) < min_count: + return result + + for issuer, subject in valid.items(): + others = [value for key, value in valid.items() if key != issuer] + if higher_is_better: + worse = sum(value < subject for value in others) + else: + worse = sum(value > subject for value in others) + tied = sum(value == subject for value in others) + result[issuer] = round( + (worse + 0.5 * tied) / len(others) * 100.0, + 4, + ) + return result + + +def overlay_rank( + strategy_rank: Any, + fundamental_score: Any, + weight: float, + *, + missing_score: float = 50.0, +) -> float | None: + """Blend production rank with fundamentals without changing the gate.""" + base = _finite_or_none(strategy_rank) + if base is None: + return None + if not 0.0 <= weight <= 1.0: + raise ValueError("weight must be between 0 and 1") + score = _finite_or_none(fundamental_score) + if score is None: + score = float(missing_score) + return round((1.0 - weight) * base + weight * score, 4) + + +def _available(scores: Mapping[str, Any], keys: tuple[str, ...]) -> list[float]: + return [ + value for key in keys if (value := _finite_or_none(scores.get(key))) is not None + ] + + +def _mean(values: list[float]) -> float: + return round(sum(values) / len(values), 4) + + +def _finite_or_none(value: Any) -> float | None: + if ( + isinstance(value, (int, float)) + and not isinstance(value, bool) + and math.isfinite(value) + ): + return float(value) + return None diff --git a/docs/research/fundamentals-weight-backtest.md b/docs/research/fundamentals-weight-backtest.md new file mode 100644 index 0000000..6911b75 --- /dev/null +++ b/docs/research/fundamentals-weight-backtest.md @@ -0,0 +1,153 @@ +# Point-in-time fundamentals weight backtest + +Status: pre-registered local research. Running it does not change production. + +## Question + +Does reordering already-qualified long setups with SEC fundamentals improve the +production book's risk-adjusted return? The qualification gate, execution model, +position sizing, capacity, costs, ATR trail, and post-stop re-entry policy remain +unchanged. This isolates the incremental value of fundamentals as a ranking +overlay. + +## Registered experiment + +The control is the production 80/20 residual-momentum / volatility rank. The +runner tests three fundamental composites at weights 10%, 20%, 30%, and 40%: + +- Quality: operating margin, FCF margin, low net-debt/EBITDA, and low dilution. + At least two inputs must exist. +- Growth: revenue growth and diluted-EPS growth. At least one must exist. +- Balanced: equal weight to the quality and growth sub-scores. Both must exist. + +Each raw metric is ranked favorably from 0 to 100 across the CIK-deduplicated +tracked universe. Missing composite scores are neutral at 50. The formula is: + +`final rank = (1 - weight) * production rank + weight * fundamental rank` + +There are 13 registered portfolio trials including the control. That complete +count is used by the Deflated Sharpe calculation. + +Historical P/E and FCF yield are excluded. Stored Alpaca bars are split-adjusted, +while filing-time EPS and shares are not guaranteed to use today's split basis; +mixing them without point-in-time split factors can manufacture valuation moves. +Earnings surprise is also excluded because the completed Dolt SUE study already +failed its promotion bar for this strategy. + +## Point-in-time rule + +Only SEC rows accepted before midnight America/New_York at the start of a signal +date are visible. This is conservative relative to the daily pre-market SEC +import and prevents same-day filings or later amendments from leaking backward. +The derivation then selects the newest visible accession per fiscal period. + +Default windows are fixed before the first run: + +- Train: entry date before 2024-01-01 +- Validation: 2024-01-01 through 2024-12-31 +- Test: entry date on or after 2025-01-01 + +Do not move these boundaries after seeing results. The test window is used only +to check the single arm chosen from train and validation. Reports expose all +registered rows for auditability and correct multiple-testing accounting. + +## 1. Create the portable snapshot + +Run this wherever the production PostgreSQL connection is already configured. +The exporter copies prices, the safe strategy settings, SEC snapshots, and Dolt +earnings rows. It does not copy credentials or unrelated system settings. + +Windows PowerShell: + +```powershell +.venv\Scripts\python.exe scripts\create_backtest_snapshot.py ` + --output backtest_snapshots\fundamentals-backtest.sqlite ` + --force +``` + +Linux production host: + +```bash +.venv/bin/python scripts/create_backtest_snapshot.py \ + --output backtest_snapshots/fundamentals-backtest.sqlite \ + --force +``` + +Copy only the SQLite file to the MacBook. `scp`, a local network share, or an +encrypted USB drive are all fine. Do not copy `.env`. + +## 2. Prepare the MacBook + +Use the same Git commit as the machine that created the report. From the repo: + +```bash +python3.11 -m venv .venv +source .venv/bin/activate +python -m pip install --upgrade pip +python -m pip install -e '.[dev]' +chmod +x scripts/run_fundamentals_macbook.sh +``` + +Put the snapshot at `backtest_snapshots/fundamentals-backtest.sqlite`, or pass a +different path to the launcher. + +## 3. Run it + +```bash +./scripts/run_fundamentals_macbook.sh \ + backtest_snapshots/fundamentals-backtest.sqlite +``` + +The launcher defaults to logical CPU count minus one. Override it if the laptop +gets too warm or memory pressure rises: + +```bash +WORKERS=8 ./scripts/run_fundamentals_macbook.sh \ + backtest_snapshots/fundamentals-backtest.sqlite +``` + +The first run builds two caches under `reports/.cache`: production candidate +replay and point-in-time fundamental scores. If interrupted, rerun the same +command; valid caches are reused. Cache keys include snapshot size and mtime, so +a new snapshot triggers a rebuild. + +## 4. Bring the result back + +The final line names one ZIP such as: + +`reports/fundamentals-overlay-20260723-180000.zip` + +That ZIP contains: + +- the complete JSON report and reproducibility metadata; +- a readable Markdown summary; +- portfolio-arm CSV; +- factor-IC CSV; +- full-period trade CSV for every arm, allowing winner-concentration checks; +- this registered protocol. + +Copy the ZIP into this workspace or attach it in the conversation. The snapshot +itself is not needed for the first evaluation unless a result looks inconsistent. + +## Evaluation order + +1. Data coverage and the accepted-at range. +2. Individual factor IC: sign, magnitude, consistency, and cross-section size. +3. Composite IC in train, validation, and test. +4. Development selection made without the test window. +5. Test Sharpe, CAGR, drawdown, yearly returns, trial-corrected DSR, and trade + overlap versus control. +6. Sensitivity to a few dominant winners and whether the effect is economically + large enough to justify production complexity. + +The mechanical development bar requires train and validation Sharpe not below +control and validation drawdown no more than two percentage points worse. A test +pass is still research evidence, not automatic deployment. + +## Known limitation + +The snapshot contains today's tracked tickers, not historical constituent +membership or delisted names. This creates survivorship bias. The same biased +universe is used for control and overlays, so the local comparison is useful, +but its absolute Sharpe or CAGR must not be presented as an unbiased live +expectation. Forward paper performance remains the true out-of-sample check. diff --git a/scripts/create_backtest_snapshot.py b/scripts/create_backtest_snapshot.py index e49b7b6..f357ed6 100644 --- a/scripts/create_backtest_snapshot.py +++ b/scripts/create_backtest_snapshot.py @@ -1,9 +1,9 @@ -"""Create a minimal local SQLite snapshot for offline backtest research. +"""Create a portable local SQLite snapshot for offline backtest research. -Copies only the data required by app.services.backtest_service.run_backtest: +Copies the data required by the production backtest and fundamentals research: tickers, OHLCV bars, SPY benchmark closes, and the activation / recommendation / -paper-exit settings the run reads. Other system settings are intentionally -skipped to avoid copying secrets into local snapshot files. +paper-exit settings the run reads, immutable SEC snapshots, and Dolt earnings +events. Other system settings are skipped to avoid copying secrets locally. """ from __future__ import annotations @@ -54,7 +54,9 @@ def _parse_args() -> argparse.Namespace: help="SQLite snapshot path to create.", ) parser.add_argument("--batch-size", type=int, default=5000) - parser.add_argument("--force", action="store_true", help="Overwrite an existing snapshot file.") + parser.add_argument( + "--force", action="store_true", help="Overwrite an existing snapshot file." + ) return parser.parse_args() @@ -65,6 +67,7 @@ async def _copy_table( *, batch_size: int, where=None, + row_transform=None, ) -> int: table = model.__table__ columns = list(table.columns) @@ -87,6 +90,8 @@ async def _copy_table( stream = await source.stream(stmt.execution_options(yield_per=batch_size)) async for partition in stream.partitions(batch_size): rows = [dict(row._mapping) for row in partition] + if row_transform is not None: + rows = [row_transform(row) for row in rows] if not rows: continue await dest.execute(insert(table), rows) @@ -105,6 +110,8 @@ async def _main() -> None: from app.database import Base import app.models # noqa: F401 - registers all metadata tables from app.models.benchmark_price import BenchmarkPrice + from app.models.earnings_event import EarningsEvent + from app.models.fundamental_snapshot import FundamentalSnapshot from app.models.ohlcv import OHLCVRecord from app.models.settings import SystemSetting from app.models.ticker import Ticker @@ -123,8 +130,12 @@ async def _main() -> None: connect_args={"server_settings": {"default_transaction_read_only": "on"}}, ) dest_engine = create_async_engine(_sqlite_url(output)) - SourceSession = async_sessionmaker(source_engine, class_=AsyncSession, expire_on_commit=False) - DestSession = async_sessionmaker(dest_engine, class_=AsyncSession, expire_on_commit=False) + SourceSession = async_sessionmaker( + source_engine, class_=AsyncSession, expire_on_commit=False + ) + DestSession = async_sessionmaker( + dest_engine, class_=AsyncSession, expire_on_commit=False + ) print(f"Source: {_hide_password(source_url)}") print(f"Snapshot: {output}") @@ -135,7 +146,9 @@ async def _main() -> None: async with SourceSession() as source, DestSession() as dest: counts = { - "tickers": await _copy_table(source, dest, Ticker, batch_size=args.batch_size), + "tickers": await _copy_table( + source, dest, Ticker, batch_size=args.batch_size + ), "system_settings": await _copy_table( source, dest, @@ -152,9 +165,30 @@ async def _main() -> None: SystemSetting.key.like("paper_%"), ), ), - "benchmark_prices": await _copy_table(source, dest, BenchmarkPrice, batch_size=args.batch_size), - "ohlcv_records": await _copy_table(source, dest, OHLCVRecord, batch_size=args.batch_size), + "benchmark_prices": await _copy_table( + source, dest, BenchmarkPrice, batch_size=args.batch_size + ), + "ohlcv_records": await _copy_table( + source, dest, OHLCVRecord, batch_size=args.batch_size + ), } + # Import-run provenance is operational metadata, not a research input. + # Null it so the portable snapshot needs no data_import_runs rows. + async with SourceSession() as source, DestSession() as dest: + counts["fundamental_snapshots"] = await _copy_table( + source, + dest, + FundamentalSnapshot, + batch_size=args.batch_size, + row_transform=lambda row: {**row, "import_run_id": None}, + ) + counts["earnings_events"] = await _copy_table( + source, + dest, + EarningsEvent, + batch_size=args.batch_size, + row_transform=lambda row: {**row, "import_run_id": None}, + ) finally: await source_engine.dispose() await dest_engine.dispose() diff --git a/scripts/run_fundamentals_macbook.sh b/scripts/run_fundamentals_macbook.sh new file mode 100644 index 0000000..b8bf5e1 --- /dev/null +++ b/scripts/run_fundamentals_macbook.sh @@ -0,0 +1,38 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="$(git rev-parse --show-toplevel)" +cd "$ROOT" + +SNAPSHOT="${1:-backtest_snapshots/fundamentals-backtest.sqlite}" +WORKERS="${WORKERS:-$(sysctl -n hw.logicalcpu 2>/dev/null || echo 8)}" +if [[ "$WORKERS" -gt 1 ]]; then + WORKERS=$((WORKERS - 1)) +fi + +if [[ -x .venv/bin/python ]]; then + PYTHON=.venv/bin/python +else + PYTHON="${PYTHON:-python3}" +fi + +if [[ ! -f "$SNAPSHOT" ]]; then + echo "Snapshot not found: $SNAPSHOT" >&2 + exit 1 +fi + +STAMP="$(date -u +%Y%m%d-%H%M%S)" +OUT="reports/fundamentals-overlay-${STAMP}.json" + +echo "Snapshot: $SNAPSHOT" +echo "Workers: $WORKERS" +echo "Output: $OUT" + +"$PYTHON" scripts/run_fundamentals_research.py "$SNAPSHOT" \ + --workers "$WORKERS" \ + --candidate-cache reports/.cache/fundamentals-candidates.pkl \ + --fundamentals-cache reports/.cache/fundamentals-scores.pkl \ + --out "$OUT" + +echo +echo "Bring this file back for review: ${OUT%.json}.zip" diff --git a/scripts/run_fundamentals_research.py b/scripts/run_fundamentals_research.py new file mode 100644 index 0000000..fe60425 --- /dev/null +++ b/scripts/run_fundamentals_research.py @@ -0,0 +1,1140 @@ +"""Point-in-time fundamentals rank-overlay research. + +The production qualification gate is unchanged. The runner first measures +30-session factor IC, then reorders already-qualified candidates with quality, +growth, or balanced fundamental ranks at 10, 20, 30, and 40 percent weights. +It writes JSON, Markdown, CSV, and a portable ZIP bundle. +""" + +from __future__ import annotations + +import argparse +import asyncio +import csv +import hashlib +import json +import multiprocessing +import os +import pickle +import subprocess +import sys +import zipfile +from collections import defaultdict +from concurrent.futures import ProcessPoolExecutor, as_completed +from datetime import date, datetime, time, timezone +from pathlib import Path +from types import SimpleNamespace +from typing import Any +from zoneinfo import ZoneInfo + +from sqlalchemy import func, select +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)) + +CACHE_VERSION = "fundamentals-overlay-v1" +NY = ZoneInfo("America/New_York") +COMPOSITES = ("quality", "growth", "balanced") +WEIGHTS = (0.10, 0.20, 0.30, 0.40) +ARMS: tuple[dict[str, Any], ...] = ( + { + "id": "control_w00", + "label": "Production 80/20 momentum-volatility rank", + "composite": None, + "weight": 0.0, + }, + *tuple( + { + "id": f"{composite}_w{round(weight * 100):02d}", + "label": f"{composite.title()} overlay {round(weight * 100)}%", + "composite": composite, + "weight": weight, + } + for composite in COMPOSITES + for weight in WEIGHTS + ), +) +N_TRIALS = len(ARMS) + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("snapshot") + parser.add_argument( + "--workers", type=int, default=max(1, multiprocessing.cpu_count() - 1) + ) + parser.add_argument("--out", default=None) + parser.add_argument( + "--candidate-cache", default="reports/.cache/fundamentals-candidates.pkl" + ) + parser.add_argument( + "--fundamentals-cache", default="reports/.cache/fundamentals-scores.pkl" + ) + parser.add_argument("--train-end", default="2024-01-01") + parser.add_argument("--test-start", default="2025-01-01") + parser.add_argument("--allow-spawn", action="store_true") + parser.add_argument("--quiet", action="store_true") + return parser.parse_args() + + +def _sqlite_url(path: Path) -> str: + return f"sqlite+aiosqlite:///{path.resolve().as_posix()}" + + +def _default_out() -> Path: + stamp = datetime.now().strftime("%Y%m%d-%H%M%S") + return Path("reports") / f"fundamentals-overlay-{stamp}.json" + + +def _snapshot_hash(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _cache_key(snapshot: Path, suffix: dict[str, Any]) -> dict[str, Any]: + stat = snapshot.stat() + return { + "version": CACHE_VERSION, + "snapshot": str(snapshot.resolve()), + "size": stat.st_size, + "mtime_ns": stat.st_mtime_ns, + **suffix, + } + + +def _load_cache(path: Path, key: dict[str, Any]) -> Any | None: + if not path.exists(): + return None + with path.open("rb") as handle: + payload = pickle.load(handle) # noqa: S301 - trusted local cache + return payload.get("value") if payload.get("key") == key else None + + +def _save_cache(path: Path, key: dict[str, Any], value: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("wb") as handle: + pickle.dump( + {"key": key, "value": value}, handle, protocol=pickle.HIGHEST_PROTOCOL + ) + + +def _git_commit() -> str | None: + try: + return subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=ROOT, + capture_output=True, + check=True, + text=True, + ).stdout.strip() + except (OSError, subprocess.CalledProcessError): + return None + + +async def _load_snapshot(snapshot: Path, quiet: bool) -> dict[str, Any]: + from app.models.earnings_event import EarningsEvent + from app.models.fundamental_snapshot import FundamentalSnapshot + from app.models.ohlcv import OHLCVRecord + from app.models.ticker import Ticker + from app.services import backtest_service as bt + from app.services.admin_service import get_activation_config + from app.services.paper_trade_service import get_exit_policy + from app.services.recommendation_service import get_recommendation_config + + engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True) + session_factory = async_sessionmaker( + engine, class_=AsyncSession, expire_on_commit=False + ) + try: + async with session_factory() as db: + config = await get_recommendation_config(db) + activation = await get_activation_config(db) + exit_config = await get_exit_policy(db) + benchmark = await bt._load_benchmark_closes_for_backtest( + db, days=None, refresh=False + ) + tickers = list( + (await db.execute(select(Ticker).order_by(Ticker.symbol))).scalars() + ) + snapshots = list( + ( + await db.execute( + select(FundamentalSnapshot).order_by( + FundamentalSnapshot.cik, + FundamentalSnapshot.accepted_at, + ) + ) + ).scalars() + ) + earnings_count = int( + ( + await db.execute(select(func.count()).select_from(EarningsEvent)) + ).scalar_one() + ) + price_bounds = ( + await db.execute( + select( + func.min(OHLCVRecord.date), + func.max(OHLCVRecord.date), + func.count(), + ) + ) + ).one() + prices: dict[str, tuple] = {} + for index, ticker in enumerate(tickers, 1): + columns = await bt._fetch_columns(db, ticker.symbol) + if columns is not None: + prices[ticker.symbol] = columns + if not quiet and index % 50 == 0: + print(f"loaded prices {index}/{len(tickers)}", flush=True) + finally: + await engine.dispose() + + by_cik: dict[str, list[Any]] = defaultdict(list) + for row in snapshots: + by_cik[str(row.cik)].append(row) + ticker_rows = [ + { + "symbol": ticker.symbol, + "cik": str(ticker.cik) if ticker.cik else None, + "sic": ticker.sic, + } + for ticker in tickers + ] + audit = { + "tickers": len(tickers), + "tickers_with_prices": len(prices), + "tickers_with_cik": sum(row["cik"] is not None for row in ticker_rows), + "unique_ciks": len({row["cik"] for row in ticker_rows if row["cik"]}), + "fundamental_rows": len(snapshots), + "fundamental_ciks": len(by_cik), + "accepted_at_min": min((row.accepted_at for row in snapshots), default=None), + "accepted_at_max": max((row.accepted_at for row in snapshots), default=None), + "earnings_rows": earnings_count, + "price_date_min": price_bounds[0], + "price_date_max": price_bounds[1], + "price_rows": int(price_bounds[2] or 0), + } + return { + "config": config, + "activation": activation, + "exit_config": exit_config, + "benchmark": benchmark, + "ticker_rows": ticker_rows, + "prices": prices, + "snapshots_by_cik": dict(by_cik), + "audit": audit, + } + + +def _representatives( + ticker_rows: list[dict], prices: dict[str, tuple] +) -> dict[str, str]: + reps: dict[str, str] = {} + for row in ticker_rows: + cik = row.get("cik") + symbol = str(row["symbol"]) + if not cik or symbol not in prices: + continue + if cik not in reps or symbol < reps[cik]: + reps[cik] = symbol + return reps + + +def _build_candidates( + snapshot: Path, + data: dict[str, Any], + args: argparse.Namespace, +) -> tuple[list[dict], int]: + from app.services import backtest_service as bt + from scripts import run_research_matrix as shared + + cache_path = Path(args.candidate_cache) + key = _cache_key(snapshot, {"kind": "daily-production-candidates"}) + cached = _load_cache(cache_path, key) + if cached is not None: + if not args.quiet: + print(f"loaded candidate cache {cache_path}", flush=True) + return list(cached["qualified"]), int(cached["entry_candidate_count"]) + + prices = data["prices"] + workers = max(1, min(args.workers, max(1, multiprocessing.cpu_count() - 1))) + replay_rows: list[dict] = [] + replay_start = date(1900, 1, 1) + + def replay_one(symbol: str, columns: tuple) -> list[dict]: + return bt._replay_candidates_for_period( + symbol, + columns, + data["config"], + data["activation"], + data["benchmark"], + replay_start, + "daily", + True, + True, + ) + + if workers == 1: + for index, (symbol, columns) in enumerate(prices.items(), 1): + replay_rows.extend(replay_one(symbol, columns)) + if not args.quiet and index % 25 == 0: + print(f"replay {index}/{len(prices)}", flush=True) + else: + context = bt._mp_context() or multiprocessing.get_context("spawn") + with ProcessPoolExecutor(max_workers=workers, mp_context=context) as pool: + futures = { + pool.submit( + bt._replay_candidates_for_period, + symbol, + columns, + data["config"], + data["activation"], + data["benchmark"], + replay_start, + "daily", + True, + True, + ): symbol + for symbol, columns in prices.items() + } + for index, future in enumerate(as_completed(futures), 1): + replay_rows.extend(future.result()) + if not args.quiet and index % 25 == 0: + print(f"replay {index}/{len(futures)}", flush=True) + + setups = [row for row in replay_rows if not row.get("_rank_only")] + observations = [row for row in replay_rows if row.get("_universe_rank_observation")] + ranks = shared._live_universe_rank_map( + observations, + data["benchmark"], + bt.STRATEGY_RANK_MOMENTUM_WEIGHT, + ) + cutoff = float(data["activation"].get("min_momentum_percentile", 80.0)) + qualified: list[dict] = [] + for setup in setups: + if setup.get("direction") != "long": + continue + identity = (str(setup["symbol"]), str(setup["date"])) + rank = ranks.get(identity) + if rank is None: + continue + candidate = { + key: value + for key, value in setup.items() + if not key.startswith("_universe_") + } + candidate[bt.PRODUCTION_PERCENTILE_KEY] = rank["momentum_percentile"] + candidate[bt.VOL_PERCENTILE_KEY] = rank["volatility_percentile"] + candidate[bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY] = rank["strategy_rank"] + candidate["qualified"] = bt._momentum_qualifies(candidate, cutoff) + if candidate["qualified"]: + qualified.append(candidate) + + if not qualified: + raise RuntimeError("no qualified long candidates after replay") + value = {"qualified": qualified, "entry_candidate_count": len(setups)} + _save_cache(cache_path, key, value) + if not args.quiet: + print(f"wrote candidate cache {cache_path}", flush=True) + return qualified, len(setups) + + +def _weekly_factor_dates( + representatives: dict[str, str], prices: dict[str, tuple] +) -> set[date]: + from app.services import backtest_service as bt + + dates: set[date] = set() + for symbol in representatives.values(): + columns = prices[symbol] + records = [ + SimpleNamespace(date=date.fromordinal(int(value))) for value in columns[0] + ] + for index in bt._weekly_asof_indices(records): + if index >= bt.MIN_LOOKBACK - 1 and index + bt.HORIZON < len(records): + dates.add(records[index].date) + return dates + + +def _utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +def _coverage_summary(rows: list[dict[str, int]]) -> dict[str, Any]: + if not rows: + return {} + keys = sorted(rows[0]) + result: dict[str, Any] = {"dates": len(rows)} + for key in keys: + values = sorted(row[key] for row in rows) + middle = len(values) // 2 + median = ( + values[middle] + if len(values) % 2 + else (values[middle - 1] + values[middle]) / 2 + ) + result[key] = { + "min": values[0], + "median": median, + "max": values[-1], + } + return result + + +def _build_scores( + snapshot: Path, + dates: set[date], + representatives: dict[str, str], + snapshots_by_cik: dict[str, list[Any]], + args: argparse.Namespace, +) -> tuple[dict[str, dict[str, dict[str, float | None]]], dict[str, Any]]: + from app.services import fundamentals_derivation as derivation + from app.services import fundamentals_research as research + + ordered_dates = sorted(dates) + date_fingerprint = hashlib.sha256( + "|".join(value.isoformat() for value in ordered_dates).encode() + ).hexdigest() + cache_path = Path(args.fundamentals_cache) + key = _cache_key( + snapshot, + { + "kind": "point-in-time-scores", + "date_fingerprint": date_fingerprint, + "availability": "accepted before signal-date midnight America/New_York", + }, + ) + cached = _load_cache(cache_path, key) + if cached is not None: + if not args.quiet: + print(f"loaded fundamentals cache {cache_path}", flush=True) + return cached["scores"], cached["coverage"] + + eligible_ciks = sorted(set(representatives) & set(snapshots_by_cik)) + rows_by_cik = { + cik: sorted(snapshots_by_cik[cik], key=lambda row: _utc(row.accepted_at)) + for cik in eligible_ciks + } + positions = {cik: 0 for cik in eligible_ciks} + visible = {cik: [] for cik in eligible_ciks} + current_features: dict[str, dict[str, float | None]] = {} + scores_by_date: dict[str, dict[str, dict[str, float | None]]] = {} + coverage_rows: list[dict[str, int]] = [] + + for date_index, signal_date in enumerate(ordered_dates, 1): + cutoff = datetime.combine(signal_date, time.min, tzinfo=NY).astimezone( + timezone.utc + ) + for cik in eligible_ciks: + rows = rows_by_cik[cik] + position = positions[cik] + changed = False + while position < len(rows) and _utc(rows[position].accepted_at) <= cutoff: + visible[cik].append(rows[position]) + position += 1 + changed = True + positions[cik] = position + if changed: + current_features[cik] = research.raw_features( + derivation.derive(visible[cik]) + ) + scores = research.cross_section_scores(current_features) + scores_by_date[signal_date.isoformat()] = scores + coverage_rows.append( + { + key: sum(row.get(key) is not None for row in scores.values()) + for key in (*research.FACTOR_POLARITY, *research.COMPOSITE_KEYS) + } + ) + if not args.quiet and date_index % 100 == 0: + print(f"fundamentals dates {date_index}/{len(ordered_dates)}", flush=True) + + coverage = _coverage_summary(coverage_rows) + value = {"scores": scores_by_date, "coverage": coverage} + _save_cache(cache_path, key, value) + if not args.quiet: + print(f"wrote fundamentals cache {cache_path}", flush=True) + return scores_by_date, coverage + + +def _factor_diagnostics( + representatives: dict[str, str], + prices: dict[str, tuple], + scores_by_date: dict[str, dict[str, dict[str, float | None]]], + train_end: date, + test_start: date, +) -> dict[str, list[dict]]: + from app.services import backtest_service as bt + from app.services import fundamentals_research as research + + signal_keys = (*research.FACTOR_POLARITY, *research.COMPOSITE_KEYS) + observations: list[dict[str, Any]] = [] + for cik, symbol in representatives.items(): + columns = prices[symbol] + ordinals, _opens, _highs, _lows, closes, _volumes = columns + records = [ + SimpleNamespace(date=date.fromordinal(int(value))) for value in ordinals + ] + for index in bt._weekly_asof_indices(records): + forward_index = index + bt.HORIZON + if index < bt.MIN_LOOKBACK - 1 or forward_index >= len(records): + continue + if closes[index] <= 0: + continue + signal_date = records[index].date + score = scores_by_date.get(signal_date.isoformat(), {}).get(cik, {}) + forward = closes[forward_index] / closes[index] - 1.0 + iso = signal_date.isocalendar() + for key in signal_keys: + value = score.get(key) + if value is not None: + observations.append( + { + "signal": key, + "date": signal_date, + "week": (iso.year, iso.week), + "value": float(value), + "forward": float(forward), + "symbol": symbol, + } + ) + + windows = { + "train": lambda value: value < train_end, + "validation": lambda value: train_end <= value < test_start, + "test": lambda value: value >= test_start, + "full": lambda _value: True, + } + result: dict[str, list[dict]] = {} + for window, predicate in windows.items(): + collected: dict = defaultdict(lambda: defaultdict(list)) + for row in observations: + if predicate(row["date"]): + collected[row["signal"]][row["week"]].append( + { + "val": row["value"], + "fwd": row["forward"], + "symbol": row["symbol"], + } + ) + result[window] = bt._signal_evaluation(collected) + return result + + +def _attach_overlay_ranks( + candidates: list[dict], + ticker_rows: list[dict], + scores_by_date: dict[str, dict[str, dict[str, float | None]]], +) -> dict[str, Any]: + from app.services import backtest_service as bt + from app.services import fundamentals_research as research + + symbol_to_cik = { + str(row["symbol"]): row.get("cik") for row in ticker_rows if row.get("cik") + } + covered = {composite: 0 for composite in COMPOSITES} + for candidate in candidates: + cik = symbol_to_cik.get(str(candidate["symbol"])) + score = scores_by_date.get(str(candidate["date"]), {}).get(cik, {}) + for composite in COMPOSITES: + value = score.get(composite) + candidate[f"fund_{composite}"] = value + if value is not None: + covered[composite] += 1 + base = candidate.get(bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY) + for arm in ARMS: + if arm["composite"] is None: + continue + candidate[_ranking_key(arm)] = research.overlay_rank( + base, + score.get(str(arm["composite"])), + float(arm["weight"]), + ) + total = len(candidates) + return { + composite: { + "candidates": count, + "pct": round(count / total * 100.0, 2) if total else 0.0, + } + for composite, count in covered.items() + } + + +def _ranking_key(arm: dict[str, Any]) -> str: + from app.services import backtest_service as bt + + if arm["composite"] is None: + return bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY + return "fund_overlay_{}_{:02d}".format( + arm["composite"], round(float(arm["weight"]) * 100) + ) + + +def _window(arm: dict[str, Any], name: str) -> dict[str, Any] | None: + return next( + (row for row in arm.get("windows", []) if row.get("window") == name), + None, + ) + + +def _trade_overlap(subject: set[str], control: set[str]) -> dict[str, Any]: + union = subject | control + return { + "overlap_pct": round(len(subject & control) / len(union) * 100.0, 2) + if union + else 100.0, + "added": len(subject - control), + "removed": len(control - subject), + } + + +def _winner_concentration(details: list[dict[str, Any]]) -> dict[str, Any]: + pnls = sorted( + (float(row["pnl"]) for row in details if row.get("pnl") is not None), + reverse=True, + ) + rs = sorted( + (float(row["r"]) for row in details if row.get("r") is not None), + reverse=True, + ) + top_pnl = sum(pnls[:5]) + total_pnl = sum(pnls) + remaining_rs = rs[5:] + return { + "top5_pnl": round(top_pnl, 2) if pnls else None, + "net_pnl_ex_top5": round(total_pnl - top_pnl, 2) if pnls else None, + "top5_share_of_positive_net_pct": ( + round(top_pnl / total_pnl * 100.0, 2) if total_pnl > 0 else None + ), + "avg_r_ex_top5": ( + round(sum(remaining_rs) / len(remaining_rs), 4) if remaining_rs else None + ), + } + + +def _run_arm( + arm: dict[str, Any], + candidates: list[dict], + data: dict[str, Any], + train_end: date, + test_start: date, +) -> tuple[dict[str, Any], dict[str, set[str]]]: + from app.services import backtest_service as bt + from scripts import run_research_matrix as shared + + strategy = next( + row for row in bt.PORTFOLIO_MONITOR_STRATEGIES if row.get("is_production") + ) + entry = bt._entry_variant_config(str(strategy["entry_variant"])) + if entry is None: + raise RuntimeError("production entry configuration missing") + exit_config = data["exit_config"] + exit_policy = bt.LIVE_EXIT_MODE_TO_SIM.get( + str(exit_config.get("mode", "atr_trailing")), "atr_trail3" + ) + hold_days = int(exit_config.get("hold_days", 30)) + trail = float(exit_config.get("atr_multiplier", bt.ATR_TRAIL_MULTIPLIER)) + ranking_key = _ranking_key(arm) + reentry = bt._make_gate_reset_reentry_fn( + candidates, + data["prices"], + cadence="daily", + ranking_key=ranking_key, + ) + windows: list[dict[str, Any]] = [] + trades_by_window: dict[str, set[str]] = {} + full_trade_details: list[dict[str, Any]] = [] + for name, start, end in ( + ("train", None, train_end), + ("validation", train_end, test_start), + ("test", test_start, None), + ("full", None, None), + ): + sim = bt._simulate_portfolio( + candidates, + data["prices"], + data["benchmark"], + exit_policy, + hold_days, + ranking_key=ranking_key, + max_positions=int(entry["max_positions"]), + risk_per_trade=float(entry["risk_per_trade"]), + atr_trail_multiplier=trail, + post_stop_reentry_fn=reentry, + start_date=start, + end_date=end, + fill_mode=bt.FILL_MODE_CLOSE, + include_trades=True, + ) + if sim is None: + windows.append({"window": name, "error": "no trades"}) + trades_by_window[name] = set() + continue + shared._assert_calendar_truncation(sim, hold_days, bt.FILL_MODE_CLOSE) + details = sim.pop("trade_details", []) + sim["winner_concentration"] = _winner_concentration(details) + if name == "full": + full_trade_details = details + trades_by_window[name] = { + "{}:{}".format(row.get("symbol"), row.get("entry_date")) for row in details + } + for heavy in ("equity_curve", "benchmark_curve", "reentry_events"): + sim.pop(heavy, None) + dsr = bt.deflated_sharpe_ratio( + sim.get("sharpe"), + sim.get("sharpe_se"), + N_TRIALS, + n_returns=sim.get("n_returns"), + return_skew=sim.get("return_skew"), + return_kurtosis=sim.get("return_kurtosis"), + ) + windows.append({"window": name, "dsr": dsr, **sim}) + return ( + { + "id": arm["id"], + "label": arm["label"], + "composite": arm["composite"], + "weight": arm["weight"], + "ranking_key": ranking_key, + "windows": windows, + "full_trade_details": full_trade_details, + }, + trades_by_window, + ) + + +def _development_grade(control: dict, arm: dict) -> dict[str, Any]: + control_train = _window(control, "train") or {} + control_validation = _window(control, "validation") or {} + arm_train = _window(arm, "train") or {} + arm_validation = _window(arm, "validation") or {} + required = ( + control_train.get("sharpe"), + control_validation.get("sharpe"), + control_validation.get("max_drawdown_pct"), + arm_train.get("sharpe"), + arm_validation.get("sharpe"), + arm_validation.get("max_drawdown_pct"), + ) + if any(value is None for value in required): + return {"pass": False, "reason": "missing train or validation statistic"} + checks = { + "train_sharpe_not_worse": arm_train["sharpe"] >= control_train["sharpe"], + "validation_sharpe_not_worse": ( + arm_validation["sharpe"] >= control_validation["sharpe"] + ), + "validation_drawdown_within_2pp": ( + arm_validation["max_drawdown_pct"] + <= control_validation["max_drawdown_pct"] + 2.0 + ), + } + return { + "pass": all(checks.values()), + "checks": checks, + "train_sharpe_delta": round(arm_train["sharpe"] - control_train["sharpe"], 4), + "validation_sharpe_delta": round( + arm_validation["sharpe"] - control_validation["sharpe"], 4 + ), + } + + +def _final_check(control: dict, selected: dict | None) -> dict[str, Any] | None: + if selected is None: + return None + control_test = _window(control, "test") or {} + selected_test = _window(selected, "test") or {} + values = ( + control_test.get("sharpe"), + control_test.get("max_drawdown_pct"), + selected_test.get("sharpe"), + selected_test.get("max_drawdown_pct"), + ) + if any(value is None for value in values): + return {"pass": False, "reason": "missing test statistic"} + checks = { + "test_sharpe_not_worse": selected_test["sharpe"] >= control_test["sharpe"], + "test_drawdown_within_2pp": ( + selected_test["max_drawdown_pct"] <= control_test["max_drawdown_pct"] + 2.0 + ), + } + return { + "arm_id": selected["id"], + "pass": all(checks.values()), + "checks": checks, + "test_sharpe_delta": round(selected_test["sharpe"] - control_test["sharpe"], 4), + "note": "Research evidence only; passing does not change production.", + } + + +def _fmt(value: Any) -> str: + if value is None: + return "—" + return f"{value:.4g}" if isinstance(value, float) else str(value) + + +def _markdown(report: dict[str, Any]) -> str: + lines = [ + "# Point-in-time fundamentals overlay research", + "", + "Generated: {}".format(report.get("generated_at")), + "", + "## Protocol", + "", + "- Train ends before **{}**.".format(report["splits"]["train_end"]), + "- Validation runs until **{}**.".format(report["splits"]["test_start"]), + "- Test starts at that date and is not used to select the arm.", + "- Qualification is unchanged; fundamentals only reorder qualified longs.", + "- SEC filings become visible at midnight New York time after acceptance.", + "- Pre-registered portfolio trials for DSR: **{}**.".format(report["n_trials"]), + "", + "## Data warnings", + "", + ] + lines.extend(f"- {warning}" for warning in report.get("warnings", [])) + lines.extend( + [ + "", + "## Factor IC", + "", + "| window | signal | IC | t | positive | quintile spread | weeks | N |", + "|---|---|---:|---:|---:|---:|---:|---:|", + ] + ) + for window, rows in report.get("factor_ic", {}).items(): + for row in rows: + lines.append( + "| {} | {} | {} | {} | {} | {} | {} | {} |".format( + window, + row.get("signal"), + _fmt(row.get("mean_ic")), + _fmt(row.get("ic_t_stat")), + _fmt(row.get("ic_positive_pct")), + _fmt(row.get("mean_quintile_spread")), + row.get("weeks"), + _fmt(row.get("avg_cross_section")), + ) + ) + lines.extend( + [ + "", + "## Portfolio arms", + "", + "| arm | window | Sharpe | SE | DSR | CAGR | MaxDD | Calmar | trades | overlap |", + "|---|---|---:|---:|---:|---:|---:|---:|---:|---:|", + ] + ) + for arm in report.get("arms", []): + for row in arm.get("windows", []): + overlap = row.get("selection_vs_control", {}).get("overlap_pct") + lines.append( + "| {} | {} | {} | {} | {} | {} | {} | {} | {} | {} |".format( + arm["id"], + row.get("window"), + _fmt(row.get("sharpe")), + _fmt(row.get("sharpe_se")), + _fmt(row.get("dsr")), + _fmt(row.get("cagr_pct")), + _fmt(row.get("max_drawdown_pct")), + _fmt(row.get("calmar")), + _fmt(row.get("trades")), + _fmt(overlap), + ) + ) + selection = report.get("development_selection") + lines.extend(["", "## Mechanical selection", ""]) + if selection: + lines.append( + "- Development-selected arm: **{}**. ".format(selection["arm_id"]) + + "The test result is reported only as a final check." + ) + lines.append("- Final check: `{}`".format(report.get("final_check"))) + else: + lines.append("- No overlay passed the train + validation requirements.") + lines.extend(["", "Production remains unchanged pending human review.", ""]) + return "\n".join(lines) + + +def _write_outputs(report: dict[str, Any], out: Path, *, bundle: bool) -> None: + out.parent.mkdir(parents=True, exist_ok=True) + out.write_text(json.dumps(report, indent=2, default=str), encoding="utf-8") + markdown_path = out.with_suffix(".md") + markdown_path.write_text(_markdown(report), encoding="utf-8") + + arms_csv = out.with_name(f"{out.stem}-arms.csv") + with arms_csv.open("w", newline="", encoding="utf-8") as handle: + writer = csv.writer(handle) + writer.writerow( + [ + "arm", + "composite", + "weight", + "window", + "sharpe", + "sharpe_se", + "dsr", + "cagr_pct", + "max_drawdown_pct", + "calmar", + "trades", + "overlap_pct", + "top5_pnl_share_pct", + "avg_r_ex_top5", + ] + ) + for arm in report.get("arms", []): + for row in arm.get("windows", []): + writer.writerow( + [ + arm["id"], + arm["composite"], + arm["weight"], + row.get("window"), + row.get("sharpe"), + row.get("sharpe_se"), + row.get("dsr"), + row.get("cagr_pct"), + row.get("max_drawdown_pct"), + row.get("calmar"), + row.get("trades"), + row.get("selection_vs_control", {}).get("overlap_pct"), + row.get("winner_concentration", {}).get( + "top5_share_of_positive_net_pct" + ), + row.get("winner_concentration", {}).get("avg_r_ex_top5"), + ] + ) + + factor_csv = out.with_name(f"{out.stem}-factor-ic.csv") + with factor_csv.open("w", newline="", encoding="utf-8") as handle: + writer = csv.writer(handle) + writer.writerow( + [ + "window", + "signal", + "mean_ic", + "ic_t_stat", + "ic_positive_pct", + "mean_quintile_spread", + "weeks", + "avg_cross_section", + "reliable", + ] + ) + for window, rows in report.get("factor_ic", {}).items(): + for row in rows: + writer.writerow( + [ + window, + row.get("signal"), + row.get("mean_ic"), + row.get("ic_t_stat"), + row.get("ic_positive_pct"), + row.get("mean_quintile_spread"), + row.get("weeks"), + row.get("avg_cross_section"), + row.get("reliable"), + ] + ) + + trades_csv = out.with_name(f"{out.stem}-trades.csv") + trade_columns = [ + "arm", + "symbol", + "entry_date", + "exit_date", + "r", + "pnl", + "reason", + "hold", + "entry", + "exit", + "risk_dollars", + ] + with trades_csv.open("w", newline="", encoding="utf-8") as handle: + writer = csv.DictWriter(handle, fieldnames=trade_columns, extrasaction="ignore") + writer.writeheader() + for arm in report.get("arms", []): + for trade in arm.get("full_trade_details", []): + writer.writerow({"arm": arm["id"], **trade}) + + if bundle: + bundle_path = out.with_suffix(".zip") + with zipfile.ZipFile(bundle_path, "w", zipfile.ZIP_DEFLATED) as archive: + for path in (out, markdown_path, arms_csv, factor_csv, trades_csv): + archive.write(path, arcname=path.name) + protocol = ROOT / "docs" / "research" / "fundamentals-weight-backtest.md" + if protocol.exists(): + archive.write(protocol, arcname=protocol.name) + + +async def _main() -> None: + args = _parse_args() + snapshot = Path(args.snapshot) + out = Path(args.out) if args.out else _default_out() + if not snapshot.exists(): + raise SystemExit(f"snapshot not found: {snapshot}") + if args.workers < 1: + raise SystemExit("--workers must be positive") + train_end = date.fromisoformat(args.train_end) + test_start = date.fromisoformat(args.test_start) + if train_end >= test_start: + raise SystemExit("--train-end must be earlier than --test-start") + + os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1" + if args.allow_spawn: + os.environ["BACKTEST_ALLOW_SPAWN"] = "1" + + data = await _load_snapshot(snapshot, args.quiet) + audit = data["audit"] + if audit["fundamental_rows"] == 0 or audit["fundamental_ciks"] < 5: + raise SystemExit( + "snapshot lacks usable fundamental_snapshots; create a fresh export " + "with scripts/create_backtest_snapshot.py" + ) + representatives = _representatives(data["ticker_rows"], data["prices"]) + candidates, entry_candidate_count = _build_candidates(snapshot, data, args) + factor_dates = _weekly_factor_dates(representatives, data["prices"]) + all_dates = factor_dates | { + date.fromisoformat(str(candidate["date"])) for candidate in candidates + } + scores_by_date, score_coverage = _build_scores( + snapshot, + all_dates, + representatives, + data["snapshots_by_cik"], + args, + ) + candidate_coverage = _attach_overlay_ranks( + candidates, data["ticker_rows"], scores_by_date + ) + factor_ic = _factor_diagnostics( + representatives, + data["prices"], + scores_by_date, + train_end, + test_start, + ) + + report: dict[str, Any] = { + "generated_at": datetime.now(timezone.utc).isoformat(), + "git_commit": _git_commit(), + "snapshot": str(snapshot.resolve()), + "snapshot_sha256": _snapshot_hash(snapshot), + "splits": { + "train_end": train_end.isoformat(), + "test_start": test_start.isoformat(), + "windows": { + "train": f"entry date < {train_end.isoformat()}", + "validation": ( + f"{train_end.isoformat()} <= entry date < {test_start.isoformat()}" + ), + "test": f"entry date >= {test_start.isoformat()}", + }, + }, + "n_trials": N_TRIALS, + "pre_registered_arms": list(ARMS), + "protocol": { + "qualification": "unchanged production gate; rank overlay only", + "cadence": "daily", + "fill_mode": "close; production near-close proxy", + "horizon_sessions": 30, + "filing_availability": ( + "accepted_at before signal-date midnight America/New_York; " + "conservative match for the daily pre-market SEC import" + ), + "missing_fundamental_score": 50.0, + "selection": ( + "highest validation Sharpe among arms with train and validation " + "Sharpe not below control and validation drawdown within 2pp" + ), + "production_mutation": False, + }, + "warnings": [ + "Current tracked universe only: historical constituent membership and " + "delisted names are unavailable, so absolute results have survivorship bias.", + "Historical valuation is excluded because split-adjusted bars cannot be " + "safely combined with filing-time EPS and shares without split factors.", + "Earnings surprise is excluded because the completed SUE study already " + "failed its promotion bar for this strategy.", + "The test window remains research evidence, not a pristine future sample; " + "live paper performance is still the final out-of-sample check.", + ], + "data_audit": audit, + "strategy_config": { + "recommendation": data["config"], + "activation": data["activation"], + "exit": data["exit_config"], + }, + "score_cross_section_coverage": score_coverage, + "qualified_candidate_coverage": candidate_coverage, + "entry_candidate_count": entry_candidate_count, + "qualified_candidates": len(candidates), + "factor_ic": factor_ic, + "arms": [], + "development_grades": {}, + "development_selection": None, + "final_check": None, + } + _write_outputs(report, out, bundle=False) + + trade_sets: dict[str, dict[str, set[str]]] = {} + control: dict[str, Any] | None = None + for arm in ARMS: + if not args.quiet: + print("running {}".format(arm["id"]), flush=True) + result, arm_trades = _run_arm(arm, candidates, data, train_end, test_start) + trade_sets[str(arm["id"])] = arm_trades + if control is None: + control = result + else: + for row in result["windows"]: + name = str(row["window"]) + row["selection_vs_control"] = _trade_overlap( + arm_trades.get(name, set()), + trade_sets["control_w00"].get(name, set()), + ) + report["development_grades"][str(arm["id"])] = _development_grade( + control, result + ) + report["arms"].append(result) + _write_outputs(report, out, bundle=False) + + if control is None: + raise RuntimeError("control arm did not run") + eligible = [ + arm + for arm in report["arms"][1:] + if report["development_grades"].get(arm["id"], {}).get("pass") + ] + selected = max( + eligible, + key=lambda arm: ( + float((_window(arm, "validation") or {}).get("sharpe") or -999.0), + -float(arm["weight"]), + ), + default=None, + ) + if selected is not None: + report["development_selection"] = { + "arm_id": selected["id"], + "chosen_without_test": True, + "validation_sharpe": (_window(selected, "validation") or {}).get("sharpe"), + } + report["final_check"] = _final_check(control, selected) + report["completed_at"] = datetime.now(timezone.utc).isoformat() + _write_outputs(report, out, bundle=True) + print(f"wrote {out}", flush=True) + bundle_path = out.with_suffix(".zip") + print("wrote {}".format(bundle_path), flush=True) + + +if __name__ == "__main__": + asyncio.run(_main()) diff --git a/tests/unit/test_fundamentals_derivation.py b/tests/unit/test_fundamentals_derivation.py index ca249f4..a6fc804 100644 --- a/tests/unit/test_fundamentals_derivation.py +++ b/tests/unit/test_fundamentals_derivation.py @@ -2,7 +2,7 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import date, datetime, timezone import pytest @@ -38,7 +38,9 @@ _ENDS = { # period_end per (fy, quarter index 0..3) } -def _year(fy, discretes: dict[str, list[float]], instants: dict[str, list] | None = None): +def _year( + fy, discretes: dict[str, list[float]], instants: dict[str, list] | None = None +): """Build 4 snapshot rows (Q1,Q2,Q3,FY) with YTD-cumulative flow fields from the given per-quarter discrete values; instants set as-is per quarter.""" rows = [] @@ -55,22 +57,38 @@ def _year(fy, discretes: dict[str, list[float]], instants: dict[str, list] | Non def _two_years(): rev25 = [100, 110, 120, 130] rev26 = [110, 121, 132, 143] # +10% each quarter YoY - rows = _year(2025, { - "revenue": rev25, - "operating_income": [x * 0.2 for x in rev25], - "diluted_eps": [1.0, 1.1, 1.2, 1.3], - "cfo": [x * 0.25 for x in rev25], - "capex": [x * 0.05 for x in rev25], - "depreciation_amortization": [x * 0.05 for x in rev25], - }, instants={"shares_outstanding": [1000, 1000, 1000, 1000], "cash_and_st_investments": [40] * 4, "total_debt": [140] * 4}) - rows += _year(2026, { - "revenue": rev26, - "operating_income": [x * 0.2 for x in rev26], - "diluted_eps": [1.1, 1.21, 1.32, 1.43], - "cfo": [x * 0.25 for x in rev26], - "capex": [x * 0.05 for x in rev26], - "depreciation_amortization": [x * 0.05 for x in rev26], - }, instants={"shares_outstanding": [900, 900, 900, 900], "cash_and_st_investments": [50] * 4, "total_debt": [150] * 4}) + rows = _year( + 2025, + { + "revenue": rev25, + "operating_income": [x * 0.2 for x in rev25], + "diluted_eps": [1.0, 1.1, 1.2, 1.3], + "cfo": [x * 0.25 for x in rev25], + "capex": [x * 0.05 for x in rev25], + "depreciation_amortization": [x * 0.05 for x in rev25], + }, + instants={ + "shares_outstanding": [1000, 1000, 1000, 1000], + "cash_and_st_investments": [40] * 4, + "total_debt": [140] * 4, + }, + ) + rows += _year( + 2026, + { + "revenue": rev26, + "operating_income": [x * 0.2 for x in rev26], + "diluted_eps": [1.1, 1.21, 1.32, 1.43], + "cfo": [x * 0.25 for x in rev26], + "capex": [x * 0.05 for x in rev26], + "depreciation_amortization": [x * 0.05 for x in rev26], + }, + instants={ + "shares_outstanding": [900, 900, 900, 900], + "cash_and_st_investments": [50] * 4, + "total_debt": [150] * 4, + }, + ) return rows @@ -97,9 +115,11 @@ def test_net_debt_leverage_and_share_dilution(): # net debt = total_debt - cash = 150 - 50 = 100 (latest instant) assert d.metrics["net_debt"].value == pytest.approx(100.0) # EBITDA TTM = TTM operating_income + TTM D&A; net_debt/ebitda - op_ttm = 506 * 0.2 # 101.2 - da_ttm = 506 * 0.05 # 25.3 - assert d.metrics["net_debt_to_ebitda"].value == pytest.approx(100.0 / (op_ttm + da_ttm), rel=1e-6) + op_ttm = 506 * 0.2 # 101.2 + da_ttm = 506 * 0.05 # 25.3 + assert d.metrics["net_debt_to_ebitda"].value == pytest.approx( + 100.0 / (op_ttm + da_ttm), rel=1e-6 + ) # shares 900 vs 1000 a year earlier -> -10% (buyback) assert d.metrics["share_count_change_yoy"].value == pytest.approx(-10.0, abs=1e-6) @@ -130,7 +150,9 @@ def test_net_debt_requires_both_components(): r.total_debt = None d = fd.derive(rows) assert d.metrics["net_debt"].value is None - assert d.metrics["net_debt_to_ebitda"].value is None # net debt null -> leverage null + assert ( + d.metrics["net_debt_to_ebitda"].value is None + ) # net debt null -> leverage null def test_leverage_null_when_ebitda_nonpositive(): @@ -140,15 +162,23 @@ def test_leverage_null_when_ebitda_nonpositive(): r.depreciation_amortization = 1 d = fd.derive(rows) assert d.metrics["net_debt"].value == pytest.approx(100.0) # net debt still valid - assert d.metrics["net_debt_to_ebitda"].value is None # but leverage nulled + assert d.metrics["net_debt_to_ebitda"].value is None # but leverage nulled def test_tape_stops_at_a_gap(): - rows = [r for r in _two_years() if not (r.fiscal_year == 2026 and r.fiscal_period == "Q1")] + rows = [ + r + for r in _two_years() + if not (r.fiscal_year == 2026 and r.fiscal_period == "Q1") + ] d = fd.derive(rows) hist = d.metrics["operating_margin"].history # consecutive suffix ending at FY2026: Q2, Q3, FY (not compressed across the Q1 gap) - assert [p.period_end for p in hist] == [date(2026, 3, 31), date(2026, 6, 30), date(2026, 9, 30)] + assert [p.period_end for p in hist] == [ + date(2026, 3, 31), + date(2026, 6, 30), + date(2026, 9, 30), + ] def test_yoy_growth_null_when_prior_nonpositive(): @@ -160,14 +190,48 @@ def test_yoy_growth_null_when_prior_nonpositive(): assert d.metrics["eps_growth_yoy"].value is None # loss->profit is not a % +def test_derive_as_of_excludes_future_amendment(): + rows = _two_years() + original = next( + row for row in rows if row.fiscal_year == 2026 and row.fiscal_period == "FY" + ) + amendment = replace( + original, + accepted_at=datetime(2027, 1, 1, tzinfo=UTC), + revenue=999999, + ) + before = fd.derive_as_of([*rows, amendment], datetime(2026, 12, 31, tzinfo=UTC)) + after = fd.derive_as_of([*rows, amendment], datetime(2027, 1, 2, tzinfo=UTC)) + assert before.metrics["revenue_growth_yoy"].value == pytest.approx(10.0) + assert after.metrics["revenue_growth_yoy"].value != pytest.approx(10.0) + + +def test_derive_as_of_treats_sqlite_naive_acceptance_as_utc(): + rows = _two_years() + rows[0].accepted_at = datetime(2025, 1, 1) + result = fd.derive_as_of(rows, datetime(2027, 1, 1, tzinfo=UTC)) + assert result.latest_period_end == date(2026, 9, 30) + + def test_amendment_selection_newest_accepted_wins(): rows = _two_years() # an amendment to FY2026 FY restates revenue YTD higher, accepted later - amended = Snap(2026, "FY", date(2026, 9, 30), date(2026, 11, 1), - datetime(2027, 1, 1, tzinfo=UTC), revenue=999999, - operating_income=100, diluted_eps=1.43, cfo=100, capex=10, - depreciation_amortization=25, shares_outstanding=900, - cash_and_st_investments=50, total_debt=150) + amended = Snap( + 2026, + "FY", + date(2026, 9, 30), + date(2026, 11, 1), + datetime(2027, 1, 1, tzinfo=UTC), + revenue=999999, + operating_income=100, + diluted_eps=1.43, + cfo=100, + capex=10, + depreciation_amortization=25, + shares_outstanding=900, + cash_and_st_investments=50, + total_debt=150, + ) d = fd.derive(rows + [amended]) # Q4 revenue discrete now uses the amended YTD(FY)=999999 minus YTD(Q3)=363 # so TTM/growth reflects the amendment, proving newest accepted_at won. diff --git a/tests/unit/test_fundamentals_research.py b/tests/unit/test_fundamentals_research.py new file mode 100644 index 0000000..c13dde2 --- /dev/null +++ b/tests/unit/test_fundamentals_research.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +import math + +import pytest + +from app.services import fundamentals_research as research + + +def test_favorable_percentiles_are_tie_aware(): + ranks = research.favorable_percentiles( + {"a": 3, "b": 3, "c": 3, "d": 3, "e": 3}, + higher_is_better=True, + ) + assert set(ranks.values()) == {50.0} + + +def test_lower_is_better_flips_the_rank(): + ranks = research.favorable_percentiles( + {"a": 1, "b": 2, "c": 3, "d": 4, "e": 5}, + higher_is_better=False, + ) + assert ranks["a"] == 100.0 + assert ranks["e"] == 0.0 + + +def test_invalid_and_thin_cross_sections_stay_null(): + ranks = research.favorable_percentiles( + {"a": 1, "b": 2, "c": math.nan, "d": None, "e": 5}, + higher_is_better=True, + ) + assert all(value is None for value in ranks.values()) + + +def test_composites_use_equal_subgroup_weighting(): + features = { + str(index): { + "operating_margin": index, + "fcf_margin": index, + "net_debt_to_ebitda": 6 - index, + "share_count_change_yoy": 6 - index, + "revenue_growth_yoy": index, + "eps_growth_yoy": index, + } + for index in range(1, 6) + } + scores = research.cross_section_scores(features) + assert scores["5"]["quality"] == 100.0 + assert scores["5"]["growth"] == 100.0 + assert scores["5"]["balanced"] == 100.0 + assert scores["3"]["balanced"] == 50.0 + + +def test_overlay_uses_neutral_missing_score_and_validates_weight(): + assert research.overlay_rank(90, None, 0.2) == 82.0 + assert research.overlay_rank(90, 100, 0.2) == 92.0 + with pytest.raises(ValueError, match="weight"): + research.overlay_rank(90, 50, 1.1) diff --git a/tests/unit/test_fundamentals_research_runner.py b/tests/unit/test_fundamentals_research_runner.py new file mode 100644 index 0000000..6ce7853 --- /dev/null +++ b/tests/unit/test_fundamentals_research_runner.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import zipfile + +from scripts import run_fundamentals_research as runner + + +def _window(name, sharpe, drawdown): + return { + "window": name, + "sharpe": sharpe, + "sharpe_se": 0.2, + "dsr": 0.8, + "cagr_pct": 10.0, + "max_drawdown_pct": drawdown, + "calmar": 1.0, + "trades": 20, + } + + +def test_arm_matrix_is_bounded_and_pre_registered(): + assert runner.N_TRIALS == 13 + assert runner.ARMS[0]["id"] == "control_w00" + assert {arm["weight"] for arm in runner.ARMS[1:]} == {0.1, 0.2, 0.3, 0.4} + assert {arm["composite"] for arm in runner.ARMS[1:]} == { + "quality", + "growth", + "balanced", + } + + +def test_development_grade_does_not_read_test_window(): + control = { + "windows": [ + _window("train", 1.0, 10.0), + _window("validation", 1.0, 10.0), + _window("test", 9.0, 1.0), + ] + } + arm = { + "windows": [ + _window("train", 1.1, 10.0), + _window("validation", 1.2, 11.0), + _window("test", -9.0, 90.0), + ] + } + assert runner._development_grade(control, arm)["pass"] is True + + +def test_output_bundle_is_self_contained(tmp_path): + report = { + "generated_at": "2026-07-23T00:00:00Z", + "splits": {"train_end": "2024-01-01", "test_start": "2025-01-01"}, + "n_trials": 13, + "warnings": ["survivorship bias"], + "factor_ic": {"full": []}, + "arms": [], + "development_selection": None, + "final_check": None, + } + output = tmp_path / "result.json" + runner._write_outputs(report, output, bundle=True) + with zipfile.ZipFile(output.with_suffix(".zip")) as archive: + names = set(archive.namelist()) + assert { + "result.json", + "result.md", + "result-arms.csv", + "result-factor-ic.csv", + "result-trades.csv", + } <= names + + +def test_winner_concentration_exposes_top_five_dependence(): + details = [ + {"pnl": value, "r": value / 10} for value in (100, 90, 80, 70, 60, -10, -20) + ] + result = runner._winner_concentration(details) + assert result["top5_pnl"] == 400 + assert result["net_pnl_ex_top5"] == -30 + assert result["avg_r_ex_top5"] == -1.5