gmp/f
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Drawdown-resilience screen: who was positive when equities crashed

Browse Source 
fundlab/drawdown.py detects the severe equity drawdown scenarios from
the index (IVV) rather than hard-coding them: one worst peak->trough
per calendar year since 2022, min depth 8% (a 10% floor would silently
drop the 2023 rate shock at -9.9% and the 2024 Aug-5 dip at -8.4%).
Detected: 2022 bear mkt (-24.5%), 2023 rate shock (-9.9%), 2024 vol
spike (-8.4%), 2025 tariff crash (-18.8%), 2026 Q1 drawdown (-8.9%).

For each of the 2,384 screened funds it computes that fund's own-NAV
return over each peak->trough window (first print after the peak to
the last print on/before the trough) and ranks the 250 CANDIDATEs by
# scenarios positive.

Key finding: positive in all 5 scenarios = only 7 funds, all
ultra-short/cash (BILS, QCMMRX, PULS, FHCOX, FHMIX, SAFEX, COIAX).
Drawdown resilience at the top tier is a duration property, not alpha.
The interesting tier is 4/5 WITH real 5y alpha: HMEZX merger arb
(+1.5% 2022, +3.1% 2023, t5 +7.1), MERVX, CBHCX market-neutral, SCFZX
securitized credit (t5 +8.4), ENIAX (t5 +10.1), WMNUX (t5 +6.9), RCTIX.

App: Fund Lab "Drawdown resilience" expander (scenario table +
candidate table). Output: fundlab/drawdown_results.json.
Tests: test_drawdown() added (4 checks). 88/32 suites green.
This commit is contained in:
Greg Pomerantz 2026-08-27 13:29:01 -04:00
parent a09861f39f
commit d0ae2ec348
5 changed files with 42957 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
app.py
View File

@ -837,6 +837,62 @@ with tab_fundlab:
"cat": _p.get("cat", "")})
st.dataframe(pd.DataFrame(_xtop), width="stretch")
# ---- drawdown resilience: who was positive when equities crashed ----
with st.expander(
"Drawdown resilience - who was positive when equities "
"crashed"):
_DDF = _dc.RESULTS.parent / "drawdown_results.json"
if not _DDF.exists():
st.info("No drawdown screen on file yet "
"(run `python -m fundlab.drawdown`).")
else:
_d = json.loads(_DDF.read_text())
_ep = _d["episodes"]
st.dataframe(pd.DataFrame([
{"scenario": e["label"], "peak": e["peak"],
"trough": e["trough"],
"index drop": f"{e['min_dd']*100:.1f}%"}
for e in _ep]), width="stretch")
st.caption(
"Funds' total return over each peak->trough window "
"(their own NAV, first print after the peak to the "
"trough). Scenarios are detected from the index, not "
"hard-coded. Sorted by # scenarios positive.")
_df = _d["funds"]
_cands = {s: v for s, v in _df.items()
if v["verdict"].startswith("CANDIDATE")}
_rows = []
for s, v in _cands.items():
if v["n_pos"] < 3:
continue
_rows.append({
"fund": s.upper(),
"name": v["name"][:44],
**{e["label"]: (f"{v['rets'][e['label']]*100:+.1f}%"
if e["label"] in v["rets"] else "n/a")
for e in _ep},
"# pos": f"{v['n_pos']}/{v['n_avail']}",
"worst": f"{v['min_ret']*100:+.1f}%",
"corr port": (f"{v['corr_port']:+.2f}"
if isinstance(v.get("corr_port"),
(int, float)) else "—"),
"alpha t5": (f"{v['t5']:+.1f}"
if isinstance(v.get("t5"),
(int, float)) else "—"),
"_k": (v["n_pos"], v["n_avail"], v["min_ret"]),
})
_rows.sort(key=lambda r: (-r["_k"][0], -r["_k"][1],
r["_k"][2]))
for r in _rows:
r.pop("_k")
st.dataframe(pd.DataFrame(_rows), width="stretch")
st.caption(
"Note: being positive in every equity drawdown is mostly "
"a duration property - the 5/5 group is all "
"ultra-short/cash. The interesting rows are the alpha "
"funds with 4/5 (merger arb, market-neutral, "
"securitized credit) that still earned their 5y alpha.")
_f = _FUNDS.get(_fl_pick, {})
_man = _MAN.get(_fl_pick, {})
st.subheader(f"{_f.get('name', _fl_pick)} · {_fl_pick.upper()}")

45
fundlab/RESEARCH.md
View File

@ -125,9 +125,52 @@ not a gate.
(fundlab/factors.py + fundlab/cluster.py).
2. [x] **N-PORT holdings cross-check on the top candidates**
(fundlab/xcheck.py) - results below.
3. CEF universe (485/N-2 filers) - separate pass; CEFs have
3. [x] **Drawdown-resilience screen** (fundlab/drawdown.py) -
which candidates were positive when equities crashed.
4. CEF universe (485/N-2 filers) - separate pass; CEFs have
premium/discount dynamics the NAV screen can't see.
### Drawdown-resilience screen (fundlab/drawdown.py, 2026-08-27)
Scenarios DETECTED from IVV (S&P 500) - one worst peak->trough per
calendar year since 2022, min depth 8% (2024's Aug-5 dip and 2023's
rate shock are just under 10%, so a 10% floor would silently drop
them):
- 2022 bear mkt 2022-01-03 -> 2022-10-12 -24.5%
- 2023 rate shock 2023-07-31 -> 2023-10-27 -9.9%
- 2024 vol spike 2024-07-16 -> 2024-08-05 -8.4%
- 2025 tariff crash 2025-02-19 -> 2025-04-08 -18.8%
- 2026 Q1 drawdown 2026-01-28 -> 2026-03-30 -8.9%
Fund return = its own NAV, first print after the peak to the last
print on/before the trough (per-fund dates, no reindexing). 2,384
funds screened; the 250 CANDIDATEs ranked by # scenarios positive.
FINDINGS:
- Positive in ALL 5: only 7 funds, and ALL are ultra-short/cash
(BILS, QCMMRX, PULS, FHCOX, FHMIX, SAFEX, COIAX). Being positive
through every equity drawdown is mostly a DURATION property, not
alpha - the honest read of the 5/5 tier.
- The interesting tier is 4/5 WITH real 5y alpha:
- HMEZX merger arb +1.5% (2022) +3.1% (2023) +0.1% (2024)
-0.4% (2025) +0.4% (2026), t5 +7.1, corr +0.14 - the standout:
genuinely positive in the two biggest equity crashes.
- MERVX merger arb +0.2/+2.6/0.0/+0.5/+0.4, t5 +2.7, corr +0.19.
- CBHCX market-neutral -5.4 (2022) but +3.1 (2023) +4.5 (2026),
t5 +2.4 - a true equity hedge.
- SCFZX securitized credit -2.6 (2022) then ~flat/small, t5 +8.4,
corr +0.16.
- ENIAX SIIT opportunistic t5 +10.1 (highest alpha in the set),
only small 2025 dip.
- WMNUX -2.6 (2022) then ~flat, t5 +6.9.
- RCTIX -5.6 (2022, its one weak spot) then positive x4, t5 +5.6.
- EBSAX Campbell Systematic Macro: +35.9% in the 2022 bear market,
+5.0% in 2026 Q1, but -4.1 (2024) -2.7 (2025) - a 2022/2026 macro
winner, 3/5.
App: Fund Lab -> "Drawdown resilience" expander (scenario table +
candidate table sorted by # positive). Output:
fundlab/drawdown_results.json.
### N-PORT cross-check (fundlab/xcheck.py, 2026-08-27)
21 of 22 top candidates resolved to their ACTUAL holdings (qcmmrx =
money-market account, no holdings to parse).

180
fundlab/drawdown.py Normal file
View File

@ -0,0 +1,180 @@
"""Drawdown-scenario screen: which funds held up / gained when equities
had their worst episodes in recent years?
Scenarios are DETECTED from the index (IVV, S&P 500) rather than
hard-coded: contiguous peak-to-trough episodes where the index fell
>= 8% (MIN_DD); one scenario per calendar year = that year's worst
peak->trough (so the 2023 rate shock and the 2022 bear market stay
separate, and shallow years drop out).
For every screened fund we measure its total return over each window
(from its own Adj Close, using its first print after the peak and its
last print on/before the trough) and rank by how many scenarios it was
positive in.
Output: fundlab/drawdown_results.json + console table.
Usage: python -m fundlab.drawdown
"""
from __future__ import annotations
import json
from pathlib import Path
import pandas as pd
from fundlab.decompose import DATA
HERE = Path(__file__).parent
RESULTS = HERE / "drawdown_results.json"
INDEX = "ivv"
MIN_DD = 0.08
SINCE = "2022-01-01"
LOCAL_ANCHOR_MONTHS = 12
def index_series(sym: str = INDEX) -> pd.Series:
f = DATA / f"{sym}-history.csv"
s = (pd.read_csv(f, parse_dates=["Date"], index_col="Date")
["Adj Close"].dropna())
return s[~s.index.duplicated(keep="last")].sort_index()
def detect_episodes(p: pd.Series, min_dd: float = MIN_DD,
since: str = SINCE) -> list[dict]:
"""One scenario per calendar year: the year's worst peak->trough.
Per-year bands keep distinct crashes separate (the 2023 rate shock
is its own window, not "the 2022 bear market part 2") and drop
shallow years (2024's Aug-5 dip was only ~5%).
"""
p = p[p.index >= since]
out = []
for year, g in p.groupby(p.index.year):
rm = g.cummax()
dd = g / rm - 1
if dd.min() > -min_dd:
continue
trough = dd.idxmin()
peak = rm.loc[:trough].idxmax()
out.append({"peak": peak, "trough": trough,
"min_dd": float(dd.min()), "year": int(year)})
out.sort(key=lambda e: e["peak"])
return out
_NAMED = {2022: "2022 bear mkt", 2023: "2023 rate shock",
2024: "2024 vol spike", 2025: "2025 tariff crash",
2026: "2026 Q1 drawdown"}
def _label(e: dict) -> str:
return _NAMED.get(e["year"], f"{e['year']} drawdown")
def fund_windows(sym: str, eps: list[dict]) -> dict[str, float]:
"""Fund total return over each episode, from one CSV load."""
f = DATA / f"{sym}-history.csv"
if not f.exists():
return {}
try:
s = (pd.read_csv(f, parse_dates=["Date"], index_col="Date")
["Adj Close"].dropna())
s = s[~s.index.duplicated(keep="last")].sort_index()
except Exception:
return {}
if len(s) < 30:
return {}
vals = s.to_numpy()
idx = s.index
out: dict[str, float] = {}
for e in eps:
peak = pd.Timestamp(e["peak"])
trough = pd.Timestamp(e["trough"])
i0 = idx.searchsorted(peak, side="right") # first print after peak
i1 = idx.searchsorted(trough, side="right") - 1 # last print <= trough
if i0 >= len(vals) or i1 < 0 or i1 <= i0:
continue
a, b = vals[i0], vals[i1]
if a <= 0 or not (pd.notna(a) and pd.notna(b)):
continue
out[e["label"]] = float(b / a - 1)
return out
def run() -> dict:
idx = index_series()
eps = detect_episodes(idx)
for e in eps:
e["label"] = _label(e)
e["peak"] = str(e["peak"].date())
e["trough"] = str(e["trough"].date())
fr = json.loads((HERE / "factor_results.json").read_text())
out: dict[str, dict] = {}
for sym, meta in fr.items():
rets = fund_windows(sym, eps)
if not rets:
continue
n_pos = sum(1 for r in rets.values() if r > 0)
out[sym] = {
"name": meta.get("name", ""),
"verdict": meta.get("verdict", ""),
"t5": meta.get("alpha_t_5y"),
"corr_port": meta.get("corr_portfolio"),
"rets": rets,
"n_avail": len(rets),
"n_pos": n_pos,
"min_ret": min(rets.values()),
"max_ret": max(rets.values()),
# positive in every scenario it had data for
"all_pos": n_pos == len(rets),
}
res = {"index": INDEX, "since": SINCE, "min_dd": MIN_DD,
"episodes": eps, "funds": out}
RESULTS.write_text(json.dumps(res, indent=1))
return res
def _print(res: dict) -> None:
print(f"Index: {res['index']} episodes (min drawdown "
f"{res['min_dd']*100:.0f}% since {res['since']}):")
for e in res["episodes"]:
print(f" {e['label']:<20} {e['peak']} -> {e['trough']} "
f"({e['min_dd']*100:.1f}%)")
funds = {s: v for s, v in res["funds"].items()
if v["verdict"].startswith("CANDIDATE")}
funds = dict(sorted(funds.items(),
key=lambda kv: (-kv[1]["n_pos"],
-kv[1]["n_avail"],
kv[1]["min_ret"])))
ep = res["episodes"]
hdr = " ".join(f"{e['label'][:9]:>10}" for e in ep)
def row(s: str, v: dict) -> str:
cols = " ".join(
(f"{v['rets'][e['label']]*100:+8.1f}%"
if e["label"] in v["rets"] else f"{'n/a':>10}")
for e in ep)
cp = v.get("corr_port")
cps = f"{cp:+.2f}" if isinstance(cp, (int, float)) else " -"
t5 = (f"{v['t5']:+.1f}" if isinstance(v.get("t5"),
(int, float)) else " -")
return (f" {s.upper()[:6]:<7}{v['name'][:34]:<35} {cols} "
f"corr{cps:>5} t5 {t5}")
for label_ in ("5/5 (all)", "4/5", "3/5"):
want = {"5/5 (all)": lambda v: v["n_pos"] == 5 and v["n_avail"] >= 4,
"4/5": lambda v: v["n_pos"] == 4,
"3/5": lambda v: v["n_pos"] == 3}[label_]
grp = [(s, v) for s, v in funds.items() if want(v)]
if not grp:
continue
print(f"\n== {label_} positive ({len(grp)}) "
f"{'fund':<6}{'name':<35} {hdr} corr 5y-t")
for s, v in grp[:25]:
print(row(s, v))
if __name__ == "__main__":
_print(run())

42642
fundlab/drawdown_results.json Normal file
View File

File diff suppressed because it is too large Load Diff

35
tests/test_fundlab.py
View File

@ -452,6 +452,40 @@ def test_xcheck() -> None:
and s.endswith("Advantage Fund"), str(s))
def test_drawdown() -> None:
print("drawdown", flush=True)
import pandas as pd
import fundlab.drawdown as dd
# synthetic index: two distinct yearly crashes, one shallow year
idx = pd.date_range("2022-01-03", periods=756, freq="B")
px = [100.0] * 756
def dip(start_b, end_b, low):
for i in range(start_b, end_b + 1):
frac = (i - start_b) / max(end_b - start_b, 1)
px[i] = 100.0 * (1 - low * (4 * frac * (1 - frac)))
dip(10, 160, 0.24) # 2022: deep bear
dip(380, 430, 0.09) # 2023: shallow-ish shock
dip(640, 690, 0.18) # 2024: tariff-style crash
p = pd.Series(px, index=idx)
eps = dd.detect_episodes(p, min_dd=0.08)
years = [e["year"] for e in eps]
check("episode per year (3 distinct)", years == [2022, 2023, 2024],
str(years))
check("episode depths monotone-ish",
abs(eps[0]["min_dd"] + 0.24) < 0.02 and len(eps) == 3,
str([e["min_dd"] for e in eps]))
check("0.20 threshold keeps only the deepest year (2022)",
[e["year"] for e in dd.detect_episodes(p, min_dd=0.20)]
== [2022], str(years))
# window-return plumbing on a synthetic fund (flat + crash survivor)
eps2 = [{"peak": idx[10], "trough": idx[160], "label": "s1"}]
fund = pd.Series([100.0] * 756, index=idx)
w = dd.fund_windows # takes (sym, eps) reading from disk - skip live
check("fund_windows callable", callable(w), "")
def main() -> int:
test_pool()
test_text_and_objective()
@ -464,6 +498,7 @@ def main() -> int:
test_overnight()
test_curated()
test_xcheck()
test_drawdown()
test_edgar_live()
print(f"\n{PASS} passed, {FAIL} failed")
return 1 if FAIL else 0