"""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_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 = (
"Acme Global Macro Absolute Return Fund"
"1000000.00"
""
"Foo Corp 0%, Due 12/01/2029"
"Foo Corp 0%500000.00"
"50.0DBT"
"CORP"
"Bar ETFBar ETF"
"300000.0030.0"
"ECRF"
"UST 2yrUST"
"200000.0020.0"
"DBTUST"
""
)
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("b") 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 test_taxplan() -> None:
print("taxplan", flush=True)
import fundlab.taxplan as tp
# equity book -> taxable
r = tp.classify("Some US Equity Fund",
buckets=[{"name": "Equity (common)", "pct": 80},
{"name": "Cash/MMF (short-term)", "pct": 20}])
check("equity fund -> TAXABLE", r["location"] == "TAXABLE",
f"{r['location']} {r['score']}")
# bond book -> IRA
r = tp.classify("Some Bond Fund",
buckets=[{"name": "Corporate bond", "pct": 70},
{"name": "US Treasury", "pct": 25},
{"name": "Cash/MMF (short-term)", "pct": 5}])
check("bond fund -> IRA", r["location"] == "IRA",
f"{r['location']} {r['score']}")
# muni name override wins regardless of sleeves
r = tp.classify("X Municipal Bond Fund", betas={"shv": 5.0})
check("muni name -> TAXABLE (munis)",
r["location"] == "TAXABLE (munis)" and r["score"] == 1.0,
f"{r['location']}")
# merger arb: deal gains can be STCG but NII/ROC are mixed -> capped
# to the MIXED band (the 1099 decides), not a clean IRA
r = tp.classify("The Merger Fund",
buckets=[{"name": "Equity (common)", "pct": 90},
{"name": "Cash/MMF (short-term)", "pct": 10}])
check("merger arb capped (STCG) -> MIXED",
r["location"] == "MIXED (check 1099)" and 0.35 < r["score"] <= 0.50,
f"{r['location']} {r['score']}")
# ROC: >= half of the 5y return defers to the investor (LTCG on a
# >1y sale) -> upgraded out of IRA; smaller ROC -> note only
splits = {"X": {"appr_share": 0.05, "payout_12m": 0.03, "tot": 0.40}}
r1 = tp.classify("ROC Fund",
buckets=[{"name": "Bond (corporate)", "pct": 80},
{"name": "Cash/MMF (short-term)", "pct": 20}])
r1["location"] = "IRA" # what the score alone gave
_real_rocs = tp._rocs
# per-share tables record distributions as NEGATIVES (declared amounts)
tp._rocs = lambda: {"X": {"nav": [100.0, 100.0], "tot": [-5.0, -5.0],
"roc": [-4.5, -4.5], "nii": [-0.5, -0.5],
"gains": []}}
tp.finalize("X", r1, splits)
check("ROC-heavy fund upgraded (defers to LTCG)",
r1["location"] == "TAXABLE (defers to LTCG)", r1["location"])
r2 = tp.classify("ROC Fund",
buckets=[{"name": "Corporate bond", "pct": 80},
{"name": "Cash/MMF (short-term)", "pct": 20}])
tp._rocs = lambda: {"X": {"nav": [100.0, 100.0], "tot": [-5.0, -5.0],
"roc": [-1.5, -1.5], "nii": [-3.5, -3.5],
"gains": []}}
tp.finalize("X", r2, splits)
check("small ROC -> note, not a location change",
r2["location"] == "IRA" and "return of capital" in r2["notes"],
r2["location"] + " " + r2["notes"][:60])
tp._rocs = _real_rocs
# money market -> flexible
r = tp.classify("Plain Money Market Account",
buckets=[{"name": "Cash/MMF (short-term)", "pct": 100}])
check("money market -> FLEXIBLE", r["location"] == "FLEXIBLE (cash)",
r["location"])
# mostly pass-through, no sleeves -> MIXED
r = tp.classify("Wrapper Fund",
buckets=[{"name": "Fund/ETF holdings", "pct": 100}])
check("100% FOF no sleeves -> MIXED",
r["location"] == "MIXED (check 1099)", r["location"])
# sleeve proxy: pure rates -> IRA; pure equity -> TAXABLE
check("sleeve proxy rates -> IRA",
tp.classify("F", betas={"tlt": 0.9, "shv": 0.2})["location"]
== "IRA", "")
check("sleeve proxy equity -> TAXABLE",
tp.classify("F", betas={"ivv": 0.8, "qqq": 0.2})["location"]
== "TAXABLE", "")
# manual override for the Leuthold wrappers
r = tp.finalize("LCORX", tp.classify("Leuthold Core Investment",
buckets=[
{"name": "Fund holdings",
"pct": 100}]))
check("LCORX manual -> TAXABLE", r["location"] == "TAXABLE",
r["location"])
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 test_cef() -> None:
import numpy as np
import pandas as pd
from fundlab import cef_screen, cef_character
print("\ncef")
# _perf: flat market price, total-return line +30% over 3y -> the
# whole return is payout, maxDD ~ 0 (monotone up)
n = 3 * 252
idx = pd.bdate_range("2023-01-02", periods=n)
adj = 100.0 * (1 + np.arange(n) / 252 * 0.10)
p = cef_screen._perf(pd.DataFrame({"Close": 100.0,
"Adj Close": adj}, index=idx))
check("_perf t5 approx +30% (3y window)",
p["t5"] is not None and 0.29 < p["t5"] < 0.31, f"{p['t5']}")
check("_perf payout_12m == t12 - p12",
abs(p["payout_12m"] - (p["t12"] - p["p12"])) < 1e-12,
f"{p['payout_12m']}")
check("_perf vol5 positive", p["vol5"] is not None and p["vol5"] > 0)
check("_perf maxdd5 <= 0 (monotone up -> ~0)",
p["maxdd5"] is not None and p["maxdd5"] <= 1e-9,
f"{p['maxdd5']}")
# select(): top return + resilient + payout; NON_FUNDS excluded
scr = {
"AAA": {"sym": "aaa", "t5": 0.5, "n_pos_scen": 0, "payout_12m": 0.0},
"BBB": {"sym": "bbb", "t5": 0.4, "n_pos_scen": 4, "payout_12m": 0.0},
"CCC": {"sym": "ccc", "t5": -0.5, "n_pos_scen": 0, "payout_12m": 0.2},
"CFND": {"sym": "cfnd", "t5": 0.9, "n_pos_scen": 0, "payout_12m": 0.0},
}
sel = cef_character.select(scr)
check("select excludes NON_FUNDS", "cfnd" not in sel, str(sel))
check("select includes top return + resilient + payout",
{"aaa", "bbb", "ccc"} <= set(sel), str(sel))
# CEF annual-report per-share highlights parser
# ----------------------------------------------------------------------
_DOC_FRANKLIN = """\
Financial highlights For a common share of capital stock outstanding
throughout each year ended November 30: 2025 1 2024 1 2023 1 2022 1 2021 1
Net
asset value, beginning of year
$55.82
$38.75
$35.97
$26.53
$17.13
Income
(loss) from operations:
Net
investment loss
(0.13
)
(0.51
)
(0.69
)
(0.43
)
(0.36
)
Net
realized and unrealized gain (loss)
(2.12
)
20.58
5.80
11.65
11.01
Total
income (loss) from operations
(2.25)
20.07
5.11
11.22
10.65
Less
distributions to common
shareholders
from:
Dividends
(0.81
)
(3.00
)
(2.37
)
(1.92
)
(0.54
)
Return of capital
(3.42
)
(0.93
)
Total distributions to common shareholders
4.23
3.00
2.37
1.92
1.47
Anti-dilutive impact of repurchase plan 0.04 0.14 0.22
Dilutive impact of rights offering (0.46) (0.27) (0.13) (0.09) (0.13)
Anti-dilutive impact of tender offer plan 0.00 0.00 0.00 0.00 0.00
Net asset value, end of year $48.88 $55.82 $38.75 $35.97 $26.53
Market price, end of year $45.55 $50.49 $34.50 $30.43 $21.65
"""
_DOC_CLASSIC = """\
Selected Per Share Data Net asset value, beginning of year $ 53.78 $ 39.73
$ 29.96 $ 42.83 $ 28.79 Income from Investment Operations: Net investment
income (a) 0.30 0.43 0.41 (b) 0.33 0.04 Net realized and unrealized gain
(loss) on investments and foreign currency transactions 11.62 13.13 9.32 (c)
(10.28) 17.31 Total from investment operations 11.92 13.56 9.73 (9.95) 17.35
Less Distributions to Stockholders from: Net investment income (0.48) (0.38)
Net realized gains (7.40) (0.44) (2.44) (2.93) Total distributions to
stockholders (7.40) (0.44) (2.92) (3.31) Capital Share Transactions:
Accretion (dilution) to net asset value, resulting from share repurchase
program, tender offer or issuance of shares for the reinvestment of
distributions from net investment income and net realized gains 0.83 0.93
0.04 0.00 (d) 0.00 (d) Net asset value, end of year $ 59.13 $ 53.78 $ 39.73
$ 29.96 $ 42.83 Market value, end of year $ 50.25 $ 44.73 $ 31.10 $ 25.20
$ 35.83
"""
_DOC_ABRDN = """\
PER SHARE OPERATING PERFORMANCE: Net asset value per common share, beginning
of year $16.38 $15.00 $15.49 $21.22 $20.25 Net investment loss (c) (0.12)
(0.07) (0.08) (0.12) (0.17) Net realized and unrealized gains/(losses) on
investments, written options and foreign currency transactions 2.91 3.11 0.87
(4.14) 2.83 Total from investment operations applicable to common
shareholders 2.79 3.04 0.79 (4.26) 2.66 Distributions to common shareholders
from: Net investment income (1.72) (0.69) (0.02) (0.03) (0.42) Net realized
gains (0.10) (0.76) (1.26) (1.44) (1.27) Return of capital (0.21) Total
distributions (1.82) (1.66) (1.28) (1.47) (1.69) Net asset value per common
share, end of year $17.35 $16.38 $15.00 $15.49 $21.22 Market price, end of
year $15.51 $15.08 $12.47 $13.66 $20.80
"""
_DOC_KOREA = """\
The Korea Fund, Inc. Financial Highlights For a share of stock outstanding
throughout each year: Year ended June 30, 2026 2025 2024 2023 2022 Net
asset value, beginning of year $30.87 $28.78 $26.52 $28.54 $54.37 Investment
Operations: Net investment income (1) (0.05) 0.02 0.09 0.19 0.32 Net realized
and change in unrealized gain (loss) 58.44 2.39 2.08 1.06 (17.05) Total from
investment operations 58.39 2.41 2.17 1.25 (16.73) Dividends and
Distributions to Stockholders from: Net investment income (1.38) (0.45)
(0.03) (2.05) Net realized gains (3.27) (7.06) Return of capital (0.02)
Total dividends and distributions to stockholders (1.38) (0.45) (3.32) (9.11)
Common Stock Transactions: Accretion to net asset value resulting from share
repurchases and tender offer 0.13 0.09 0.05 0.01 Net asset value, end of
year $87.88 $30.87 $28.78 $26.52 $28.54 Market price, end of year $75.21
$26.93 $24.13 $23.14 $24.35
"""
def test_cef_annual() -> None:
from fundlab import cef_annual as cn
# Franklin style (EMO): td-split labels, ROC in cols 0 and 4,
# positive-printed total row
h = cn.parse_highlights(_DOC_FRANKLIN)
check("franklin parses 5 cols", h is not None and h["ncols"] == 5, str(h))
check("franklin div share ~66.5%",
abs(h["share_div"] - 8.64 / 12.99) < 0.005, str(h["share_div"]))
check("franklin gains zero", h["share_gains"] == 0.0)
check("franklin roc share ~33.5%",
abs(h["share_roc"] - 4.35 / 12.99) < 0.005, str(h["share_roc"]))
check("franklin FY-end discount ~-6.8%",
abs(h["mkt_end"][0] / h["nav_end"][0] - 1 - (-0.0681)) < 0.001)
# Classic CEF (TWN): printed total row is SHORT (a zero year dropped),
# so totals derive from the NAV identity; gains sit in 4 of 5 cols
# and NII-divs in the OTHER two
h = cn.parse_highlights(_DOC_CLASSIC)
check("classic parses 5 cols", h is not None and h["ncols"] == 5, str(h))
check("classic derived total col0 = 7.40",
abs(h["dist_tot"][0] - 7.40) < 0.01, str(h["dist_tot"]))
check("classic zero-year col2", abs(h["dist_tot"][2]) < 0.01,
str(h["dist_tot"]))
check("classic gains ~93.9%",
abs(h["share_gains"] - 13.21 / 14.07) < 0.005, str(h["share_gains"]))
check("classic NII ~6.1%",
abs(h["share_div"] - 0.86 / 14.07) < 0.005, str(h["share_div"]))
# abrdn style (HQL): "per common share" labels, single-value ROC row
h = cn.parse_highlights(_DOC_ABRDN)
check("abrdn parses 5 cols", h is not None and h["ncols"] == 5, str(h))
check("abrdn div ~36.4%",
abs(h["share_div"] - 2.88 / 7.92) < 0.005, str(h["share_div"]))
check("abrdn gains ~61.0%",
abs(h["share_gains"] - 4.83 / 7.92) < 0.005, str(h["share_gains"]))
check("abrdn roc ~2.7%",
abs(h["share_roc"] - 0.21 / 7.92) < 0.005, str(h["share_roc"]))
# Korea style (KF): "Total dividends and distributions" label (the
# divs regex must not grab the total row), accretion row in cols 1-4
h = cn.parse_highlights(_DOC_KOREA)
check("korea parses 5 cols", h is not None and h["ncols"] == 5, str(h))
check("korea div ~27.4%",
abs(h["share_div"] - 3.91 / 14.26) < 0.01, str(h["share_div"]))
check("korea gains ~72.4%",
abs(h["share_gains"] - 10.33 / 14.26) < 0.01,
str(h["share_gains"]))
# unreconcilable table must be rejected, not guessed
bad = _DOC_CLASSIC.replace("(7.40) (0.44) (2.44) (2.93)",
"(7.99) (0.44) (2.44) (2.93)")
check("unreconcilable table rejected",
cn.parse_highlights(bad) is None)
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_taxplan()
test_drawdown()
test_cef()
test_cef_annual()
test_edgar_live()
print(f"\n{PASS} passed, {FAIL} failed")
return 1 if FAIL else 0
if __name__ == "__main__":
sys.exit(main())
# ----------------------------------------------------------------------