Greg Pomerantz
e9f8dfc462
- edgar.py: SEC FTS + submissions API; strict cover-gate extraction (name in title position or (TICKER) on the cover; underlying-reference names like leveraged wrappers rejected); 4-pass fetch (ticker->CIK filings, name search, annual reports, ticker search); keyword category classifier. Returns None rather than a wrong fund's objective. - fundinfo.py CLI: curated -> cached -> EDGAR resolution into funds.json - funds_curated.json: human-verified objectives for 13 benchmark-pool funds (iShares/Vanguard family-trust classes the scraper can't reach) - tests: 26 checks incl. live EDGAR fetch of VTSAX
131 lines
4.4 KiB
Python
131 lines
4.4 KiB
Python
"""Resolve fund investment objectives + categories.
|
|
|
|
Usage:
|
|
python -m fundlab.fundinfo SYMBOL [SYMBOL ...] [--refresh] [--no-curate] [--max-docs N]
|
|
|
|
Objective sources, in priority order:
|
|
1. funds_curated.json (human-verified, in this package) — used unless
|
|
--no-curate. Authoritative for the well-known benchmark funds.
|
|
2. SEC EDGAR — the fund's longName (from the Yahoo chart JSON in the
|
|
data root, default ~/prog/fin/stocks) is searched over prospectus /
|
|
annual-report forms; the objective is extracted from the fund's own
|
|
filing and the fund classified.
|
|
|
|
Everything is merged into funds.json next to the repository root. Cached
|
|
EDGAR entries are reused unless --refresh is given. Be polite to the SEC:
|
|
one fund takes ~5-15 requests; run a handful at a time.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
from .edgar import fetch_fund
|
|
|
|
DATA_ROOT = Path("~/prog/fin/stocks").expanduser()
|
|
FUNDS_FILE = Path(__file__).resolve().parent.parent / "funds.json"
|
|
CURATED_FILE = Path(__file__).resolve().parent / "funds_curated.json"
|
|
|
|
|
|
def load_curated() -> dict:
|
|
if CURATED_FILE.exists():
|
|
try:
|
|
return json.loads(CURATED_FILE.read_text())
|
|
except Exception:
|
|
pass
|
|
return {}
|
|
|
|
|
|
def long_name(sym: str, root: Path) -> str | None:
|
|
f = root / f"{sym}.json"
|
|
if not f.exists():
|
|
return None
|
|
try:
|
|
meta = json.loads(f.read_text())["chart"]["result"][0]["meta"]
|
|
return meta.get("longName") or meta.get("shortName")
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
ap = argparse.ArgumentParser(prog="fundlab.fundinfo")
|
|
ap.add_argument("symbols", nargs="+", help="fund ticker(s), e.g. vtsax vwo")
|
|
ap.add_argument("--refresh", action="store_true",
|
|
help="re-fetch from EDGAR even if present in funds.json")
|
|
ap.add_argument("--no-curate", action="store_true",
|
|
help="ignore funds_curated.json and use EDGAR for everything")
|
|
ap.add_argument("--data-root", default=str(DATA_ROOT))
|
|
ap.add_argument("--max-docs", type=int, default=10,
|
|
help="max EDGAR documents to try per fund")
|
|
args = ap.parse_args(argv)
|
|
|
|
root = Path(args.data_root)
|
|
curated = {} if args.no_curate else load_curated()
|
|
funds = {}
|
|
if FUNDS_FILE.exists():
|
|
try:
|
|
funds = json.loads(FUNDS_FILE.read_text())
|
|
except Exception:
|
|
pass
|
|
|
|
rc = 0
|
|
for sym in args.symbols:
|
|
sym = sym.lower()
|
|
# 1) curated (authoritative) — always wins over cache/EDGAR
|
|
if sym in curated:
|
|
c = curated[sym]
|
|
funds[sym] = {
|
|
"name": c.get("name", ""),
|
|
"objective": c["objective"],
|
|
"category": c["category"],
|
|
"source": "curated",
|
|
}
|
|
print(f"= {sym}: curated ({c['category']})")
|
|
continue
|
|
# 2) cached EDGAR result
|
|
if not args.refresh and sym in funds and funds[sym].get("objective"):
|
|
print(f"= {sym}: cached ({funds[sym].get('category')})")
|
|
continue
|
|
# 3) EDGAR
|
|
name = long_name(sym, root)
|
|
if not name:
|
|
print(f"= {sym}: NO chart JSON in {root} — cannot identify fund",
|
|
file=sys.stderr)
|
|
rc = 1
|
|
continue
|
|
print(f"= {sym}: fetching '{name}' from EDGAR ...", flush=True)
|
|
try:
|
|
res = fetch_fund(name, ticker=sym, max_docs=args.max_docs)
|
|
except Exception as e:
|
|
print(f" ERROR: {type(e).__name__}: {e}", file=sys.stderr)
|
|
rc = 1
|
|
continue
|
|
if not res:
|
|
print(f" no investment objective found in EDGAR", file=sys.stderr)
|
|
rc = 1
|
|
continue
|
|
funds[sym] = {
|
|
"name": name,
|
|
"objective": res["objective"],
|
|
"category": res["category"],
|
|
"form": res["form"],
|
|
"file_date": res["file_date"],
|
|
"url": res["url"],
|
|
"fetched": date.today().isoformat(),
|
|
"source": "edgar",
|
|
}
|
|
print(f" category: {res['category']} [{res['form']} {res['file_date']}]")
|
|
print(f" objective: {res['objective']}")
|
|
|
|
FUNDS_FILE.write_text(json.dumps(funds, indent=2))
|
|
print(f"\n{FUNDS_FILE}")
|
|
return rc
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|