gmp/f
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
a09861f39f
f/tests/test_fundlab.py
Greg Pomerantz a09861f39f N-PORT cross-check: verify top candidates' actual holdings 
fundlab/xcheck.py - for each screen candidate, resolve the fund's OWN
registrant CIK (browse-edgar; the 497-cover CIK is the family/trust),
get the exact series name for the ticker (the only reliable
disambiguator between sibling funds), walk the 4 most recent NPORT-P
filing dates, and parse holdings from the interactive NPORT XML
(primary_doc.xml at the accession root - NOT the XSL-rendered view the
submissions API points at). Exact seriesName match > best htm exhibit
parse. Buckets from the authoritative assetCat+issuerCat codes (ABS-O,
ABS-CBDO, DBT+UST/CORP/MUN/NUSS, LON, STIV, RA, EC+RF=fund, ...), not
position-name keywords. Resumable; raw filings cached under
nport_cache/raw/ (gitignored).

nport.py - _SECTION gains the "INVESTMENT PORTFOLIO (unaudited)"
variant (NPORT-EX Sch-F files); find_section/build gain a frac
token-tolerance param (Yahoo names drift from filing names); CMBS/ABS
bucket gains CLO/CBDO/DAC terms.

app Fund Lab - "N-PORT cross-check" expander: per-candidate table
(as-of, n, t5, top code-bucket, #1 position) + per-fund holdings
detail.

RESEARCH.md - cross-check verdicts. 21/22 resolved (qcmmrx is an MMF,
no holdings). The screen's top names are REAL:
- hmezx/mervx = genuine merger arb (equity in deal targets + escrow)
- egrix = 100% wrapper in one macro managed portfolio (underlying not
  NPORT-disclosed); etsix = fund of EV internal multi-strat accounts
- wmnux = discounted/zero-coupon corporate bonds + equity swaps (the
  "equity names" are bond issuers/swap underlyings)
- scfzx/rctix/aflix = securitized credit/CLO/distressed/levered loans
- hicox/fhmix/usmsx/btmix (munis), aguax/femdx (EM sovereign), anglx
  (agency MBS), lpxax (rotated out of prefs into bank/financial debt)
  = genuine missing-factor exposures the 35-sleeve model lacks
- fhcox/dultx/safex = short-duration carry (a short-duration sleeve
  would explain them)

tests/test_fundlab.py - test_xcheck (14 checks): parse_interactive,
code buckets, name-match normalization, series-name disambiguation.
Also: untrack fundlab/streamlit.log; gitignore raw/ + xcheck_run.log.

84 fundlab / 32 app / 14 data tests pass.
2026-08-27 12:43:17 -04:00

474 lines
22 KiB
Python
Raw Blame History

"""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_search() -> None:
print("search engine", flush=True)
from fundlab import dbmine, search, tickers
# precision gate: a different fund sharing some words must fail
check("name gate rejects wrong fund",
search._name_match("PIMCO Access to Global Markets Fund",
"PIMCO Access Income Fund") < 2 / 3, "")
check("name gate accepts right fund",
search._name_match("Fidelity Multi-Asset Income Fund",
"Fidelity Multi-Asset Income") >= 2 / 3, "")
# query ladder handles hyphens + word-count fallbacks
q = tickers._queries("AQR Diversified Event-Driven Fund")
check("query ladder exact first", q[0] ==
'"AQR Diversified Event-Driven Fund"', str(q))
check("query ladder hyphen-free", '"AQR Diversified Event Driven Fund"'
in q, str(q))
check("query ladder 2-word prefix", '"AQR Diversified"' in q, str(q))
# family dedupe: share classes collapse, distinct funds don't
check("family dedupe same fund",
dbmine.family_key("AQR Style Premia Alternative R6")
== dbmine.family_key("AQR Style Premia Alternative I"), "")
check("family dedupe distinct funds",
dbmine.family_key("AQR Style Premia Alternative R6")
!= dbmine.family_key("AQR Managed Futures Strategy I"), "")
# ticker regex: both cover formats
import re
t1 = re.findall(tickers.TICKER_RX, "Ticker Symbol: ABCDX")
t2 = re.findall(tickers.SLASH_RX, "Fidelity Multi-Asset Income Fund /FMSDX ")
check("ticker regex label format", t1 == ["ABCDX"], str(t1))
check("ticker regex slash format", t2 == ["FMSDX"], str(t2))
def test_universe() -> None:
print("edgar universe cover parser", flush=True)
from fundlab import edgar_universe as eu
sample = (
"<SERIES-AND-CLASSES-CONTRACTS-DATA>\n"
"<SERIES>\n<SERIES-NAME>1290 Multi-Alternative Strategies Fund\n"
"<CLASS-CONTRACT>\n<CLASS-CONTRACT-NAME>Class A\n"
"<CLASS-CONTRACT-TICKER-SYMBOL>TNMAX\n</CLASS-CONTRACT>\n"
"<CLASS-CONTRACT-NAME>Class I\n"
"<CLASS-CONTRACT-TICKER-SYMBOL>TNMIX\n</CLASS-CONTRACT>\n"
"</SERIES>\n"
"<SERIES-NAME>1290 High Yield Bond Fund\n"
"<CLASS-CONTRACT-TICKER-SYMBOL>TNHAX\n</CLASS-CONTRACT>\n"
"</SERIES-AND-CLASSES-CONTRACTS-DATA>\n"
"<OTHER>JUNK-TICKER-LINE\n" # after the series block: ignored
)
s = eu.parse_cover(sample)
check("two series", len(s) == 2, str(s))
check("series 1 name", s[0]["name"] ==
"1290 Multi-Alternative Strategies Fund", s[0]["name"])
check("series 1 tickers", s[0]["tickers"] == ["TNMAX", "TNMIX"],
str(s[0]["tickers"]))
check("series 2 tickers", s[1]["tickers"] == ["TNHAX"],
str(s[1]["tickers"]))
check("ticker before any series ignored",
eu.parse_cover("<CLASS-CONTRACT-TICKER-SYMBOL>NOPE\n") == [],
"")
check("malformed ticker rejected",
eu.parse_cover("<SERIES-NAME>F\n"
"<CLASS-CONTRACT-TICKER-SYMBOL>1BAD\n")[0]
["tickers"] == [], "")
def test_overnight() -> None:
print("overnight select", flush=True)
from fundlab import overnight as ov
tickers = {"AAAAX": "Foo Market Neutral Fund",
"AAAIJ": "Foo Market Neutral Fund", # 2nd class
"BBBAX": "Bar Growth Fund",
"CCCAX": "Baz ETF",
"DDDAX": "Qux Short History"}
meta = {"AAAAX": {"days": 3000, "type": "MUTUALFUND", "name": "x"},
"AAAIJ": {"days": 2500, "type": "MUTUALFUND", "name": "x"},
"BBBAX": {"days": 3000, "type": "MUTUALFUND", "name": "y"},
"CCCAX": {"days": 3000, "type": "ETF", "name": "z"},
"DDDAX": {"days": 500, "type": "MUTUALFUND", "name": "q"},
"EEEAX": {"error": "404"}}
sel = ov.select_rows(tickers, meta, min_days=1250)
check("ETF dropped", [r["sym"] for r in sel].count("cccax") == 0,
str(sel))
check("short history dropped",
[r["sym"] for r in sel].count("dddax") == 0, "")
check("classes collapse to longest",
[r["sym"] for r in sel].count("foo") == 0 and
[r for r in sel if r["name"] == "Foo Market Neutral Fund"]
[0]["sym"] == "aaaax", str(sel))
check("alpha_name tagged", sel[0]["alpha_name"] is True,
str(sel[0]))
check("non-alpha name not tagged (but kept)",
any(r["sym"] == "bbabx" or r["sym"] == "bbbax" and
r["alpha_name"] is False for r in sel), str(sel))
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 test_xcheck() -> None:
print("xcheck", flush=True)
import fundlab.xcheck as xc
xml = (
"<root><seriesName>Acme Global Macro Absolute Return Fund</seriesName>"
"<netAssets>1000000.00</netAssets>"
"<invstOrSecs>"
"<invstOrSec><name>Foo Corp 0%, Due 12/01/2029</name>"
"<title>Foo Corp 0%</title><valUSD>500000.00</valUSD>"
"<pctVal>50.0</pctVal><assetCat>DBT</assetCat>"
"<issuerCat>CORP</issuerCat></invstOrSec>"
"<invstOrSec><name>Bar ETF</name><title>Bar ETF</title>"
"<valUSD>300000.00</valUSD><pctVal>30.0</pctVal>"
"<assetCat>EC</assetCat><issuerCat>RF</issuerCat></invstOrSec>"
"<invstOrSec><name>UST 2yr</name><title>UST</title>"
"<valUSD>200000.00</valUSD><pctVal>20.0</pctVal>"
"<assetCat>DBT</assetCat><issuerCat>UST</issuerCat></invstOrSec>"
"</invstOrSecs></root>"
)
d = xc.parse_interactive(xml)
check("parse_interactive name", d is not None
and d["name"] == "Acme Global Macro Absolute Return Fund",
str(d and d.get("name")))
check("parse_interactive n positions", d is not None
and len(d["positions"]) == 3, str(d and len(d["positions"])))
check("parse_interactive net assets",
d is not None and abs(xc._fnum(d["net_assets"]) - 1e6) < 1e-6,
str(d and d.get("net_assets")))
check("parse_interactive non-NPORT -> None",
xc.parse_interactive("<a>b</a>") is None, "")
# code-based buckets
check("bucket DBT/UST", xc.xml_bucket({"asset_cat": "DBT",
"issuer_cat": "UST"})
== "US Treasury", "")
check("bucket DBT/CORP", xc.xml_bucket({"asset_cat": "DBT",
"issuer_cat": "CORP"})
== "Corporate bond", "")
check("bucket ABS-CBDO", xc.xml_bucket({"asset_cat": "ABS-CBDO",
"issuer_cat": "CORP"})
== "CLO (collateralized debt)", "")
check("bucket EC/RF = fund holding", xc.xml_bucket({"asset_cat": "EC",
"issuer_cat": "RF"})
== "Fund/ETF holdings", "")
check("bucket DE = derivative", xc.xml_bucket({"asset_cat": "DE",
"issuer_cat": "CORP"})
== "Derivative / hedge", "")
check("bucket STIV", xc.xml_bucket({"asset_cat": "STIV",
"issuer_cat": "RF"})
== "Cash/MMF (short-term)", "")
# name-match normalization
check("name match exact", xc._name_match(
"Cohen & Steers Low Duration Preferred and Income Fund, Inc.",
"Cohen & Steers Low Duration Preferred & Income Fund,Inc.") == 2, "")
check("name match containment", xc._name_match(
"Eaton Vance Global Macro Absolute Return Fund",
"Eaton Vance Global Macro Absolute Return Advantage Fund") == 1, "")
check("name match none", xc._name_match(
"NexPoint Merger Arbitrage Fund",
"NexPoint Event Driven Fund") == 0, "")
# series-name lookup from covers (cached)
t2c = xc.ticker_to_cik()
s = xc.series_name_for("EGRIX", "0000745463", t2c.get("egrix"))
check("series_name_for EGRIX disambiguates Advantage", s is not None
and s.endswith("Advantage Fund"), str(s))
def main() -> int:
test_pool()
test_text_and_objective()
test_classify()
test_strategy()
test_nport()
test_decompose()
test_search()
test_universe()
test_overnight()
test_curated()
test_xcheck()
test_edgar_live()
print(f"\n{PASS} passed, {FAIL} failed")
return 1 if FAIL else 0
if __name__ == "__main__":
sys.exit(main())
Reference in New Issue View Git Blame Copy Permalink