quant-trading/research/lab/momentum_fundamentals.py

"""
Does financial-statement analysis improve overnight_recovery_top10?

    PYTHONPATH=. uv run --with numba python -m research.lab.momentum_fundamentals [--sensitivity]

Averaging fundamental ranks into the momentum score was already tried and
hurts (fund_eval.py: Sharpe 0.80 -> 0.63; momentum's alpha lives in its most
extreme names). So every variant here keeps momentum's ranking and uses the
statements only to veto or to confirm, the way an analyst would screen a
momentum list. Fundamentals are point-in-time EDGAR values (first filing,
usable the day after it; research/lab/fundamentals.py); they cover ~90% of
S&P 500 members from 2011, so the test runs 2011-2026 (design 2011-2016,
holdout 2017-2026). A name without data is never vetoed.

Variants, fixed before looking at results (7 trials):
  profitable       veto trailing losses: ROE < 0 or operating cash flow < 0
  quality_veto     veto the bottom 20% of a sector-relative quality score
                   (mean rank of ROE, CFO/assets, accruals, gross profit/assets)
  earnings_confirm keep only a positive latest earnings surprise (SUE > 0):
                   price momentum backed by earnings momentum (Chan, Jegadeesh &
                   Lakonishok 1996; Novy-Marx 2015)
  revenue_confirm  keep only positive year-on-year revenue growth
  fund_momentum    top 30 by momentum, then the 10 with the best fundamental
                   momentum (mean rank of SUE, revenue SUE, gross-margin change,
                   ROA change)
  red_flags        veto the worst decile of accruals (Sloan 1996) or of share
                   issuance (dilution)
  fscore           veto a Piotroski-style F-score below 4
Then a neighbourhood check of the two that looked best (``--sensitivity``):
fund_momentum with the top 20 / 40 instead of 30, quality_veto at 10% / 30%.
Base: overnight_mom + rec_126, top 10, every 10 days, buffer 2 (the live
overnight_recovery_top10). Each variant is compared with the base on the
same days (paired difference, Newey-West t) and against QQQ.
"""

from __future__ import annotations

import dataclasses
import json
import os
import warnings

import numpy as np
import pandas as pd

from research.lab import fundamentals as fu
from research.lab.alpha import _nw_se, alpha_stats, bench_returns
from research.lab.factor_scan import RESULTS, load
from research.lab.report import stats
from research.lab.rotation import factor_ranks
from research.lab.us_sweep import _blend, log_trials, run, sector_rank

warnings.filterwarnings("ignore")
SPAN = ("2011-01-03", "2026-09-24")
DESIGN = ("2011-01-03", "2016-12-31")
HOLDOUT = ("2017-01-01", "2026-09-24")


def variants(p, sensitivity: bool = False) -> dict[str, tuple[pd.DataFrame, pd.DataFrame, float]]:
    """name -> (score, eligible mask, rank buffer)."""
    pr = factor_ranks(p, "us")
    F = fu.load(p)
    mom = _blend(pr["overnight_mom"], pr["rec_126"]).where(p.member)
    m = p.member
    ok = lambda veto: m & ~veto.fillna(False)                              # noqa: E731
    pct = lambda x: x.where(m).rank(axis=1, pct=True)                      # noqa: E731
    quality = _blend(*(sector_rank(F[k], p) for k in ("roe", "cfo_a", "accruals", "gp_a")))
    fmom = _blend(*(pct(F[k]) for k in ("sue", "sue_rev", "gm_change", "roa_change")))
    mom_rank = mom.rank(axis=1, ascending=False)
    top30 = mom_rank <= 30
    if sensitivity:
        return {
            "base": (mom, m, 2.0),
            "fund_momentum top20": (fmom.fillna(0.5).where(mom_rank <= 20), m, 1.0),
            "fund_momentum top30": (fmom.fillna(0.5).where(top30), m, 1.0),
            "fund_momentum top40": (fmom.fillna(0.5).where(mom_rank <= 40), m, 1.0),
            "quality_veto 10%": (mom, ok(quality < 0.1), 2.0),
            "quality_veto 30%": (mom, ok(quality < 0.3), 2.0),
        }
    return {
        "base": (mom, m, 2.0),
        "profitable": (mom, ok((F["roe"] < 0) | (F["cfo_a"] < 0)), 2.0),
        "quality_veto": (mom, ok(quality < 0.2), 2.0),
        "earnings_confirm": (mom, ok(F["sue"] <= 0), 2.0),
        "revenue_confirm": (mom, ok(F["rev_growth"] <= 0), 2.0),
        "fund_momentum": (fmom.fillna(0.5).where(top30), m, 1.0),     # no data: neutral
        "red_flags": (mom, ok((pct(F["accruals"]) < 0.1) | (pct(F["net_issuance"]) < 0.1)), 2.0),
        "fscore": (mom, ok(F["fscore"] < 4), 2.0),
    }


def paired_t(d: pd.Series) -> float:
    d = d.dropna()
    X = np.ones((len(d), 1))
    se = _nw_se(X, (d - d.mean()).to_numpy())[0]
    return float(d.mean() / se)


def main() -> None:
    import sys
    sens = "--sensitivity" in sys.argv
    p, _ = load("us")
    b = bench_returns(p.dates)
    R, rows = {}, []
    for name, (score, elig, buf) in variants(p, sens).items():
        pv = dataclasses.replace(p, member=elig)
        st = run(pv, score.where(elig), [SPAN], top_n=10, buffer=buf, every=10)
        R[name] = st["_rets"]
        rows.append({"score": f"momfund:{name}", **{k: v for k, v in st.items() if not k.startswith("_")}})
    log_trials(rows[1:], "momentum_fundamentals" + ("_sensitivity" if sens else ""))
    R = pd.DataFrame(R)
    out = {}
    print(f"{'':18s}{'2011-2026':>34s} | {'2011-2016 (design)':>34s} | {'2017-2026 (holdout)':>34s} | vs base 17-26")
    for name in R.columns:
        r = R[name]
        cells, rec = [], {}
        for label, (a, z) in (("full", SPAN), ("design", DESIGN), ("holdout", HOLDOUT)):
            x = r.loc[a:z]
            s = stats((1 + x).cumprod())
            al = alpha_stats(x, b)
            rec[label] = {"cagr": s["cagr"], "sharpe": s["sharpe"], "mdd": s["mdd"], "alpha": al["alpha"], "t": al["t"]}
            cells.append(f"{s['cagr']*100:5.1f}% sh {s['sharpe']:.2f} mdd {s['mdd']*100:4.0f}% a {al['alpha']*100:5.1f}%")
        d = (r - R["base"]).loc[HOLDOUT[0]:HOLDOUT[1]]
        dd = (r - R["base"]).loc[DESIGN[0]:DESIGN[1]]
        extra = "" if name == "base" else (f"{d.mean()*252*100:+5.1f}%/yr t {paired_t(d):+.2f} "
                                           f"(design {dd.mean()*252*100:+5.1f}% t {paired_t(dd):+.2f})")
        rec["diff_holdout"] = float(d.mean() * 252)
        out[name] = rec
        print(f"{name:18s}" + " | ".join(cells) + f" | {extra}", flush=True)
    yr = (1 + R).groupby(R.index.year).prod() - 1
    print("\nyearly returns (%):")
    print((yr * 100).round(1).to_string())
    tag = "_sensitivity" if sens else ""
    with open(os.path.join(RESULTS, f"momentum_fundamentals{tag}.json"), "w") as fh:
        json.dump(out, fh, indent=1, default=float)
    R.to_pickle(os.path.join(RESULTS, f"momentum_fundamentals{tag}.pkl"))


if __name__ == "__main__":
    main()