Greg Pomerantz
db5fc4626d
- fundlab/decompose.py: per-fund OLS forward selection (BIC-gated, |t|>2,
per-model complete cases so differently-vintaged candidates stay
comparable) against curated DISTINCT-AXIS candidate sets; full-history
+ last-5y models; rolling 1y beta drift for static-vs-time-varying
verdicts; per-fund holdings cross-check notes
- results (13 unique funds; pmfkx/lcrix/egrsx are share classes):
* jlpsx ~1.04x S&P 500, R2 0.96 5y (cleanest)
* lamhx S&P + value/mid tilt, R2 0.95, stable
* cosix 5y: HY +0.30 / MBS +0.29 / IG +0.18, R2 0.86
* cvsix market neutral, 5y R2 0.74, +5.5%/yr alpha (t 6.7)
* pmaix multi-asset: HY .62 / EFA .23 / comm .05 / bonds -.15
* mbxix hedge: ivv .39 / ief -.67 / fxe -.28, R2 0.53
* atesx NOT a static mix - rolling beta to its own QQQ/SPY holdings
is 0.13-0.89 (median 0.30): the 'risk managed' overlay is real
* qspnx/svarx/eagmx/atrfx/pmorx: market-neutral or idiosyncratic -
alpha, not sleeves (qspnx +12.8%/yr alpha t 4.0)
* lcorx/lcrix: new classes (Jul 2026), no history yet - holdings only
- atesx holdings: pulled from the adviser's SOI PDF (anchor-soi-5.31.26):
QQQ 65.2% + SPY 29.3% + MMF 0.6%, options overlay 4.9%
- pool: added qqq (Nasdaq 100) - needed to fit tech-concentrated funds
- app Fund Lab tab: per-fund decomposition (verdict, R2 full/5y, alpha,
tracking error, beta drift, component table + bar chart, holdings
cross-check note) and an all-funds summary expander
- tests: ols/forward-select engine tests (50/50 fundlab)
305 lines
14 KiB
Python
305 lines
14 KiB
Python
"""fundlab tests: pool parsing, text extraction, classification, EDGAR live.
|
|
|
|
Run: .venv/bin/python tests/test_fundlab.py
|
|
The live EDGAR test needs network + is rate-limited; it degrades to a skip.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import pathlib
|
|
import re
|
|
import sys
|
|
import urllib.error
|
|
|
|
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
|
|
|
|
from fundlab import edgar, pool # noqa: E402
|
|
|
|
PASS, FAIL = 0, 0
|
|
|
|
|
|
def check(name: str, cond: bool, extra: str = "") -> None:
|
|
global PASS, FAIL
|
|
if cond:
|
|
PASS += 1
|
|
print(f" ok {name}", flush=True)
|
|
else:
|
|
FAIL += 1
|
|
print(f" FAIL {name} {extra}", flush=True)
|
|
|
|
|
|
# ---------------------------------------------------------------- pool
|
|
def test_pool() -> None:
|
|
print("pool", flush=True)
|
|
entries = pool.load_pool()
|
|
check("pool parses many entries", len(entries) > 90, str(len(entries)))
|
|
bysym = {e.symbol: e for e in entries}
|
|
check("original entries present",
|
|
all(s in bysym for s in ("vbtlx", "ewz", "gld", "ijr", "fxe")))
|
|
check("augmentations present",
|
|
all(s in bysym for s in ("agg", "vt", "vwo", "vnq", "tlt", "bil")))
|
|
check("sections preserved",
|
|
bysym["vbtlx"].section.startswith("Treasury")
|
|
and bysym["agg"].section.startswith("Common index ETFs"))
|
|
check("labels carried", "inflation-protected" in bysym["vipix"].label)
|
|
check("no garbage symbols",
|
|
all(re.fullmatch(r"\S{1,6}", e.symbol) for e in entries))
|
|
|
|
|
|
# ---------------------------------------------------------------- text
|
|
_HTML = (b"<html><body><h3>Fund Summary</h3><p>Investment Objective</p>"
|
|
b"<p>The Fund seeks to track the performance of a benchmark index "
|
|
b"that measures the investment return of large-capitalization "
|
|
b"stocks.</p><p>Fees and Expenses</p></body></html>")
|
|
|
|
|
|
def test_text_and_objective() -> None:
|
|
print("text / objective extraction", flush=True)
|
|
txt = edgar.to_text(_HTML)
|
|
check("to_text strips tags",
|
|
"Fund Summary" in txt and "<p>" not in txt and "Investment Objective" in txt)
|
|
doc = ("Random cover page text. "
|
|
"Vanguard 500 Index Fund\nProspectus\nInvestment Objective\n"
|
|
"The Fund seeks to track the performance of a benchmark index "
|
|
"that measures the investment return of large-capitalization "
|
|
"stocks.\nFees and Expenses\nmore text about Vanguard 500 "
|
|
"Index Fund share classes.")
|
|
obj = edgar.extract_objective(doc, "Vanguard 500 Index Fund")
|
|
check("objective extracted",
|
|
obj is not None and obj.startswith("The Fund seeks to track")
|
|
and "large-capitalization stocks" in obj, repr(obj))
|
|
check("objective is bounded",
|
|
obj is not None and len(obj) < 300, repr(obj and len(obj)))
|
|
doc2 = ("Fund ABC\nThe ETF seeks to invest at least 80% of assets in "
|
|
"US investment grade debt securities of emerging markets.")
|
|
obj2 = edgar.extract_objective(doc2, "Fund ABC")
|
|
check("variant: 'The ETF seeks to ...' on the cover",
|
|
obj2 is not None and "debt securities" in obj2 and "US" in obj2,
|
|
repr(obj2))
|
|
check("no objective -> None",
|
|
edgar.extract_objective("no fund here at all", "Ghost Fund") is None)
|
|
# a document that merely MENTIONS the fund deep in its body (family
|
|
# filings, funds investing in it) must not yield that fund's objective:
|
|
# the name is not on the cover (within the first COVER_GATE chars)
|
|
body = ("Section %d discusses portfolio construction, risk management "
|
|
"and liquidity. ")
|
|
filler = ("XYZ Growth Fund\nProspectus\nInvestment Objective\nThe Fund "
|
|
"seeks to provide total return while generating moderate "
|
|
"current income.\n" +
|
|
"".join(body % i for i in range(40)) +
|
|
"Comparison: similar to Vanguard 500 Index Fund.\n")
|
|
check("mention-only doc -> None",
|
|
edgar.extract_objective(filler, "Vanguard 500 Index Fund") is None)
|
|
# the same doc with the name on the cover IS about the fund
|
|
oncover = ("Vanguard 500 Index Fund\nProspectus\nInvestment Objective\n"
|
|
"The Fund seeks to track large-capitalization stocks.\n" +
|
|
"".join(body % i for i in range(40)))
|
|
check("name on cover -> extracted",
|
|
(edgar.extract_objective(oncover, "Vanguard 500 Index Fund")
|
|
or "").startswith("The Fund seeks to track"))
|
|
# a leveraged ETF's cover cites the underlying fund's name; the name
|
|
# is a reference, not a title -> not our fund's document
|
|
doc4 = ("Acme 2X Leveraged QQQ Daily ETF\nProspectus\nThe Fund seeks\n"
|
|
"daily investment results that are 200% of the performance of\n"
|
|
"the Invesco QQQ Trust. The Fund seeks to achieve daily\n"
|
|
"investment results equal to two times the performance of the\n"
|
|
"index, before fees and expenses.")
|
|
check("underlying reference on cover -> None",
|
|
edgar.extract_objective(doc4, "Invesco QQQ Trust") is None)
|
|
|
|
|
|
# ---------------------------------------------------------------- classify
|
|
def test_classify() -> None:
|
|
print("category classification", flush=True)
|
|
c = edgar.classify_category
|
|
check("US index fund -> equity",
|
|
c("The Fund seeks to track large-capitalization stocks.",
|
|
"Vanguard 500 Index Fund") == "equity")
|
|
check("aggregate bond -> fixed_income",
|
|
c("The Fund seeks to track the performance of an index of US "
|
|
"investment grade debt securities.", "iShares Core US Aggregate Bond ETF")
|
|
== "fixed_income")
|
|
check("balanced -> mixed",
|
|
c("The Fund invests in a combination of equity and debt "
|
|
"securities.", "Example Balanced Fund") == "mixed")
|
|
check("gold -> alternatives",
|
|
c("The Fund seeks to track the price of gold bullion.",
|
|
"SPDR Gold Shares") == "alternatives")
|
|
check("money market -> money_market",
|
|
c("The Fund seeks to maintain stability of principal. It "
|
|
"invests in money market instruments.", "XX Money Market Fund")
|
|
== "money_market")
|
|
check("unknown -> other", c("The Fund does unusual things.", "Mystery") == "other")
|
|
|
|
|
|
# ---------------------------------------------------------------- edgar live
|
|
def test_edgar_live() -> None:
|
|
print("EDGAR live (network)", flush=True)
|
|
try:
|
|
name = None
|
|
meta = json.loads((pathlib.Path("~/prog/fin/stocks").expanduser()
|
|
/ "vtsax.json").read_text())["chart"]["result"][0]["meta"]
|
|
name = meta.get("longName")
|
|
if not name:
|
|
raise urllib.error.URLError("no local vtsax.json")
|
|
res = edgar.fetch_fund(name, ticker="vtsax", max_docs=8)
|
|
except Exception as e:
|
|
print(f" SKIP (no network / data): {type(e).__name__}: {e}")
|
|
return
|
|
check("vtsax objective fetched",
|
|
res is not None and "seeks to" in res["objective"]
|
|
and "stock market" in res["objective"].lower(),
|
|
repr(res and res.get("objective")))
|
|
check("vtsax classified equity",
|
|
res is not None and res["category"] == "equity",
|
|
repr(res and res.get("category")))
|
|
check("provenance recorded",
|
|
res is not None and res["url"].startswith("https://www.sec.gov")
|
|
and res["form"], repr(res))
|
|
|
|
|
|
def test_strategy() -> None:
|
|
print("strategy extraction", flush=True)
|
|
doc = ("Fund Cover\nThe Fund seeks total return.\nFees and Expenses "
|
|
"of the Fund: see table.\nInvestment Strategies The Fund invests "
|
|
"in a diversified portfolio of US and international equity "
|
|
"securities and may use derivatives to manage risk. It also "
|
|
"invests in fixed income of varying credit quality.\nPrincipal "
|
|
"Risks Market Risk. Values may go down.\nMore text.")
|
|
s = edgar.extract_strategy(doc)
|
|
check("strategy section extracted",
|
|
s is not None and "diversified portfolio" in s
|
|
and "Market Risk" not in s, repr(s))
|
|
check("no strategy section -> None",
|
|
edgar.extract_strategy("Fund Cover\nThe Fund seeks X.\nFees.") is None)
|
|
# supplement amendment: the REAL strategy follows a replaced-clause heading
|
|
doc2 = ("Supplement\nInvestment Strategies of the section of the "
|
|
"Prospectus entitled X is deleted and replaced with the "
|
|
"following: Under normal circumstances, the Fund invests in "
|
|
"investment grade bonds with maturities under one year and "
|
|
"cash equivalents.\nFund Management\nPortfolio Manager: Bob.")
|
|
s2 = edgar.extract_strategy(doc2)
|
|
check("amendment clause strategy extracted",
|
|
s2 is not None and "investment grade bonds" in s2, repr(s2))
|
|
|
|
|
|
def test_nport() -> None:
|
|
print("nport parser", flush=True)
|
|
import fundlab.nport as np
|
|
# _CAT: dash / paren / bare-space separators, one & two decimals
|
|
check("cat dash sep", np._CAT.search("CORPORATE BONDS - 53.9%")
|
|
is not None, "")
|
|
m = np._CAT.search("Asset-Backed Securities 7.2%")
|
|
check("cat bare-space sep", bool(m) and m.group(2) == "7.2", repr(m))
|
|
m = np._CAT.search("LONG-TERM INVESTMENTS 98.70%")
|
|
check("cat two decimals", bool(m) and m.group(2) == "98.70", repr(m))
|
|
# _trim_name: strip a mixed-case table-header run-on, keep title case
|
|
check("trim caps tail", np._trim_name("Maturity Fair Value CORPORATE BONDS")
|
|
== "CORPORATE BONDS", np._trim_name("Maturity Fair Value CORPORATE BONDS"))
|
|
check("trim keeps title-case", np._trim_name("Asset-Backed Securities - Non-Agency")
|
|
== "Asset-Backed Securities - Non-Agency", "")
|
|
# classify: first-match keyword bucketing
|
|
check("classify us equity", np.classify("Apple Inc. 100 $5,000", "")
|
|
== "Equity (US)", np.classify("Apple Inc. 100 $5,000", ""))
|
|
check("classify mbs", np.classify("GNMA 5.10 03/15/28", "")
|
|
== "Agency MBS", "")
|
|
check("classify fund", np.classify("Vanguard Total Bond ETF 1,000 $20,000", "")
|
|
== "Fund holdings", "")
|
|
# find_section: multi-fund stream, pick the right fund's block
|
|
filler = ("Alpha Inc. $ 1,000 Beta Corp. $ 2,000 Gamma Ltd. $ 3,000 " * 8)
|
|
stream = ("Cover Page. Alpha Fund Schedule of Investments as of January 1, 2026 "
|
|
f"COMMON STOCKS - 40.0% Apple Inc. $ 1,000 {filler}"
|
|
"Beta Mortgage Opportunities Fund Schedule of Investments as of "
|
|
"February 2, 2026 Asset-Backed Securities 7.2% Some Bond $ 4,000 "
|
|
f"{filler}Notes to Schedule of Investments.")
|
|
sec = np.find_section(stream, ["mortgage", "opportunities"])
|
|
check("find_section locates fund", sec is not None, "")
|
|
if sec:
|
|
seg = stream[sec[0]:sec[1]]
|
|
check("find_section excludes other fund", "Apple Inc." not in seg,
|
|
seg)
|
|
check("find_section excludes notes", "Notes to" not in seg, seg)
|
|
# parse_section end-to-end on a synthetic block
|
|
rs = ["Beta Mortgage Opportunities Fund",
|
|
"Schedule of Investments as of February 2, 2026",
|
|
"Asset-Backed Securities 7.2%", "Some Bond $ 2,000", "Net assets $ 50,000"]
|
|
st_ = " ".join(rs)
|
|
starts = []
|
|
p = 0
|
|
for r in rs:
|
|
starts.append(p)
|
|
p += len(r) + 1
|
|
out = np.parse_section(st_, 0, len(st_), rs, starts)
|
|
check("parse as_of", out["as_of"] == "February 2, 2026", repr(out["as_of"]))
|
|
check("parse category", any(c["name"] == "Asset-Backed Securities" and
|
|
c["pct"] == 7.2 for c in out["categories"]),
|
|
repr(out["categories"]))
|
|
check("parse position", any(p["value"] == 2000 for p in out["positions"]),
|
|
repr(out["positions"]))
|
|
check("parse net assets", out["net_assets"] == 50000, repr(out["net_assets"]))
|
|
|
|
|
|
def test_decompose() -> None:
|
|
print("decompose engine", flush=True)
|
|
import numpy as np
|
|
import fundlab.decompose as dc
|
|
rng = np.random.default_rng(7)
|
|
n = 1000
|
|
x1 = rng.normal(0, 0.01, n)
|
|
x2 = rng.normal(0, 0.008, n)
|
|
x3 = rng.normal(0, 0.01, n) # no signal
|
|
y = 0.6 * x1 + 0.3 * x2 + rng.normal(0, 0.001, n)
|
|
# ols recovers betas
|
|
X = np.column_stack([np.ones(n), x1, x2])
|
|
m = dc.ols(y, X)
|
|
check("ols beta1", abs(m["beta"][1] - 0.6) < 0.05, f"{m['beta'][1]:.3f}")
|
|
check("ols beta2", abs(m["beta"][2] - 0.3) < 0.05, f"{m['beta'][2]:.3f}")
|
|
check("ols r2 high", m["r2"] > 0.95, f"{m['r2']:.3f}")
|
|
# forward selection: picks the two signal sleeves, not the noise one
|
|
chosen, _, ok = dc.forward_select(y, {"s1": x1, "s2": x2, "s3": x3})
|
|
check("fwd picks signal", set(chosen) == {"s1", "s2"}, str(chosen))
|
|
# market neutral (pure noise): nothing selected
|
|
yn = rng.normal(0, 0.004, n)
|
|
chosen_n, _, _ = dc.forward_select(yn, {"s1": x1, "s2": x2})
|
|
check("fwd rejects noise", chosen_n == [], str(chosen_n))
|
|
# NaN handling: a candidate whose history only partly overlaps the
|
|
# fund's doesn't crash the selection, and the full-history sleeve is
|
|
# still found. (The short-history sleeve may lose on BIC because its
|
|
# complete-case sample is smaller - that's the expected, conservative
|
|
# behaviour, so we only assert robustness here.)
|
|
x2p = x2.copy()
|
|
x2p[:500] = np.nan
|
|
chosen_p, _, _ok_p = dc.forward_select(y, {"s1": x1, "s2": x2p})
|
|
check("fwd handles nan overlap", "s1" in chosen_p, f"chosen={chosen_p}")
|
|
|
|
|
|
def test_curated() -> None:
|
|
print("curated", flush=True)
|
|
import fundlab.fundinfo as fi
|
|
cur = fi.load_curated()
|
|
check("curated file loads", len(cur) >= 10, str(len(cur)))
|
|
check("curated entries well-formed",
|
|
all(set(v) >= {"objective", "category"} and
|
|
v["category"] in ("equity", "fixed_income", "mixed",
|
|
"alternatives", "money_market", "other")
|
|
for v in cur.values()))
|
|
check("curated objectives look right",
|
|
all("seeks" in v["objective"].lower() for v in cur.values()))
|
|
|
|
|
|
def main() -> int:
|
|
test_pool()
|
|
test_text_and_objective()
|
|
test_classify()
|
|
test_strategy()
|
|
test_nport()
|
|
test_decompose()
|
|
test_curated()
|
|
test_edgar_live()
|
|
print(f"\n{PASS} passed, {FAIL} failed")
|
|
return 1 if FAIL else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|