Greg Pomerantz
7b27920a77
- extract_strategy(): finds the 'Principal Investment Strategies' / 'main investment strategies' section, scores candidates (strategy prose +3, Q&A heading +2, TOC -5, risk subheading -5, stop-heading -2), truncates at the next section heading; falls back to the prose after the objective sentence when no heading exists - fundinfo --strategy [--refresh]: populates the strategy field of funds.json from each fund's EDGAR document - funds.json now carries objective + strategy for 20 funds (the 9 curated index funds have no strategy: their objective is the strategy)
213 lines
9.5 KiB
Python
213 lines
9.5 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_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_curated()
|
|
test_edgar_live()
|
|
print(f"\n{PASS} passed, {FAIL} failed")
|
|
return 1 if FAIL else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|