"""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"

Fund Summary

Investment Objective

" b"

The Fund seeks to track the performance of a benchmark index " b"that measures the investment return of large-capitalization " b"stocks.

Fees and Expenses

") 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 "

" 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 = ( "\n" "\n1290 Multi-Alternative Strategies Fund\n" "\nClass A\n" "TNMAX\n\n" "Class I\n" "TNMIX\n\n" "\n" "1290 High Yield Bond Fund\n" "TNHAX\n\n" "\n" "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("NOPE\n") == [], "") check("malformed ticker rejected", eu.parse_cover("F\n" "1BAD\n")[0] ["tickers"] == [], "") 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_search() test_universe() test_curated() test_edgar_live() print(f"\n{PASS} passed, {FAIL} failed") return 1 if FAIL else 0 if __name__ == "__main__": sys.exit(main())