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
372 lines
14 KiB
Python
372 lines
14 KiB
Python
"""EDGAR fetcher: fund investment objective + category from SEC filings.
|
|
|
|
Pipeline: fund name -> EDGAR full-text search over prospectus/annual-report
|
|
forms (N-1A, 485APOS, 8-A12B, S-1, N-CSR, N-CSRS) -> fetch the document ->
|
|
extract the "Investment Objective" sentence -> classify the fund category.
|
|
|
|
SEC usage policy: identify yourself via USER_AGENT (name + contact), stay
|
|
well under 10 requests/second (we insert REQUEST_DELAY between calls).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
USER_AGENT = "fundlab (personal research) gmp@wow.st"
|
|
REQUEST_DELAY = 0.15 # seconds between SEC requests
|
|
MAX_DOC_BYTES = 60 * 2 ** 20 # skip documents bigger than this
|
|
|
|
# forms whose primary document contains a fund's investment objective.
|
|
# Per-fund registration statements are tried FIRST — they hold one fund's
|
|
# objective in a clean "Fund Summary" box. Family annual reports (N-CSR/
|
|
# N-CSRS) cover hundreds of funds across several sub-documents, so the
|
|
# target fund's section may be in a part the search doesn't surface.
|
|
# forms whose documents contain a fund's investment objective. Per-fund
|
|
# registration statements are tried FIRST — they hold one fund's objective
|
|
# in a clean "Fund Summary" box. Annual reports (N-CSR/N-CSRS) are a
|
|
# fallback; family filings covering many funds are filtered out by the
|
|
# cover gate (see _on_cover).
|
|
PROSPECTUS_FORMS = "485APOS,485BPOS,497,497K,497F,N-1A,8-A12B,S-1,FWP,424B2,424B3"
|
|
ANNUAL_FORMS = "N-CSR,N-CSRS"
|
|
|
|
# how far into a document the fund's identifier may sit and still count as
|
|
# being ON THE COVER (the document is about this fund). Per-fund
|
|
# prospectus covers fit in ~1200 chars; title position only (~400) for
|
|
# annual reports and ticker-based lookups.
|
|
COVER_GATE = 1200
|
|
TITLE_GATE = 400
|
|
|
|
_last_request = 0.0
|
|
|
|
|
|
def sec_get(url: str, timeout: int = 120, retries: int = 3) -> bytes:
|
|
"""GET with SEC user agent, politeness delay and retry/backoff."""
|
|
global _last_request
|
|
for attempt in range(retries):
|
|
wait = REQUEST_DELAY - (time.monotonic() - _last_request)
|
|
if wait > 0:
|
|
time.sleep(wait)
|
|
_last_request = time.monotonic()
|
|
req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as r:
|
|
return r.read()
|
|
except urllib.error.HTTPError as e:
|
|
if e.code == 404:
|
|
raise
|
|
if e.code in (429, 500, 502, 503) and attempt + 1 < retries:
|
|
time.sleep(2.0 * (attempt + 1))
|
|
continue
|
|
raise
|
|
except (urllib.error.URLError, TimeoutError) as e:
|
|
if attempt + 1 < retries:
|
|
time.sleep(2.0 * (attempt + 1))
|
|
continue
|
|
raise
|
|
raise RuntimeError("unreachable")
|
|
|
|
|
|
def to_text(html: bytes) -> str:
|
|
"""Crude but adequate HTML -> text: block ends become newlines, tags go."""
|
|
t = re.sub(rb"<\s*(br|/p|/div|/tr|/h[1-6])[^>]*>", b"\n", html, flags=re.I)
|
|
t = re.sub(rb"<[^>]+>", b" ", t)
|
|
t = re.sub(rb"&#?\w+;", b" ", t)
|
|
t = t.decode("latin-1", "ignore")
|
|
t = re.sub(r"[ \t]+", " ", t)
|
|
t = re.sub(r"\n\s*", "\n", t)
|
|
# tag stripping eats apostrophes ("Fund s board") — fix the common ones
|
|
t = re.sub(r"\b(Fund|Trust|ETF) s\b", r"\1's", t)
|
|
# abbreviations whose dots would break sentence-level matching
|
|
for a, b in (("U.S.", "US"), ("U.K.", "UK"), ("E.U.", "EU"),
|
|
("e.g.", "eg"), ("i.e.", "ie"), ("No.", "No")):
|
|
t = t.replace(a, b)
|
|
return t
|
|
|
|
|
|
def ticker_to_company(ticker: str) -> tuple[str, str] | None:
|
|
"""ticker -> (cik, conformed company name) via browse-edgar (Atom output)."""
|
|
url = ("https://www.sec.gov/cgi-bin/browse-edgar?action=getcompany"
|
|
f"&CIK={urllib.parse.quote(ticker.upper())}&type=&dateb="
|
|
"&owner=include&count=5&output=atom")
|
|
xml = sec_get(url, timeout=30).decode("latin-1", "ignore")
|
|
cik = re.search(r"<cik>(\d+)</cik>", xml)
|
|
name = re.search(r"<conformed-name>([^<]+)</conformed-name>", xml)
|
|
if not cik:
|
|
return None
|
|
return cik.group(1), (name.group(1) if name else "")
|
|
|
|
|
|
def cik_recent_filings(cik: str, types: str, count: int = 6) -> list[dict]:
|
|
"""Recent filings of a registrant (newest first) via the submissions API."""
|
|
url = f"https://data.sec.gov/submissions/CIK{int(cik):010d}.json"
|
|
d = json.loads(sec_get(url, timeout=60))
|
|
f = d.get("filings", {}).get("recent", {})
|
|
forms = f.get("form", [])
|
|
out = []
|
|
for i, form in enumerate(forms):
|
|
if form not in types.split(","):
|
|
continue
|
|
out.append({
|
|
"accession": f["accessionNumber"][i],
|
|
"form": form,
|
|
"filed": f.get("filingDate", [""] * len(forms))[i],
|
|
"doc": (f.get("primaryDocument") or [""] * len(forms))[i],
|
|
})
|
|
if len(out) >= count:
|
|
break
|
|
return out
|
|
|
|
|
|
def fts_search(query: str, forms: str | None = None, size: int = 20) -> list[dict]:
|
|
"""EDGAR full-text search.
|
|
|
|
Hits are sorted by relevance, then date: a fund's own prospectus ranks
|
|
above documents that merely mention the ticker (funds investing in it,
|
|
family fund lists). Relevance alone is not enough — section attribution
|
|
in extract_objective is the real guard — but ordering by it means the
|
|
first match is usually the right fund's.
|
|
"""
|
|
params = {"q": query, "forms": forms or "", "size": str(size)}
|
|
url = f"https://efts.sec.gov/LATEST/search-index?{urllib.parse.urlencode(params)}"
|
|
d = json.loads(sec_get(url, timeout=60))
|
|
hits = d.get("hits", {}).get("hits", [])
|
|
out, seen = [], set()
|
|
for h in hits:
|
|
s = h.get("_source", {})
|
|
key = (s.get("adsh", ""), h.get("_id", "").split(":", 1)[-1])
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append({
|
|
"score": h.get("_score", 0.0),
|
|
"accession": s.get("adsh", ""),
|
|
"filename": h.get("_id", "").split(":", 1)[-1],
|
|
"cik": (s.get("ciks") or [""])[0],
|
|
"form": s.get("form", ""),
|
|
"file_date": s.get("file_date", ""),
|
|
})
|
|
out.sort(key=lambda x: (x["score"], x["file_date"]), reverse=True)
|
|
return out
|
|
|
|
|
|
def doc_url(cik: str, accession: str, filename: str) -> str:
|
|
return (f"https://www.sec.gov/Archives/edgar/data/{int(cik)}"
|
|
f"/{accession.replace('-', '')}/{filename}")
|
|
|
|
|
|
_SEEK = (r"((?:the|its) (?:fund|trust|etf|portfolio)\s*\)?[^\.\n]{0,120}?"
|
|
r"seeks (?:to |investment results |investment objective)?"
|
|
r"[^.]{10,400}\.)")
|
|
|
|
|
|
def _name_re(name: str) -> str:
|
|
"""Pattern for a fund name; to_text() turns 'U.S.' into 'US', so the
|
|
dots are optional."""
|
|
return re.escape(name).replace(r"U\.S\.", r"U\.?S\.?")
|
|
|
|
|
|
# a name preceded by one of these is a REFERENCE to another fund (a
|
|
# leveraged ETF's cover says "200% of the performance of the Invesco
|
|
# QQQ Trust"), not the title of the document
|
|
_REF_BEFORE = re.compile(
|
|
r"(?:performance\s+of\s+|underlying\s+|index\s+of\s+|versus\s+|"
|
|
r"than\s+|against\s+|of\s+the\s+|the\s+)$", re.I)
|
|
|
|
|
|
def _name_titlelike(text: str, name: str, limit: int) -> bool:
|
|
"""True when the name appears in title position on the cover — not as
|
|
an underlying reference (see _REF_BEFORE)."""
|
|
for m in re.finditer(_name_re(name), text[:limit], re.I):
|
|
before = text[max(0, m.start() - 40):m.start()]
|
|
if not _REF_BEFORE.search(before):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _on_cover(text: str, name: str | None, ticker: str | None,
|
|
limit: int = COVER_GATE) -> str | None:
|
|
"""Cover gate: is this document ABOUT the fund?
|
|
|
|
A fund's own registration statement puts its name (in title position)
|
|
or its ticker in parentheses on the cover page, within the first
|
|
~1200 characters. A document that merely mentions the fund elsewhere
|
|
(family fund lists, funds investing in it, exhibits) does not. Returns
|
|
'name' / 'ticker' / None.
|
|
"""
|
|
head = text[:limit]
|
|
if name and _name_titlelike(text, name, limit):
|
|
return "name"
|
|
if ticker and re.search(r"\(\s*" + re.escape(ticker.upper()) + r"\s*\)",
|
|
head, re.I):
|
|
return "ticker"
|
|
return None
|
|
|
|
|
|
def extract_objective(text: str, name: str, ticker: str | None = None,
|
|
gate: int = COVER_GATE) -> str | None:
|
|
"""Extract the fund's investment objective sentence from a document
|
|
that is about the fund (see _on_cover).
|
|
|
|
Returns None when the document is not about this fund (cover gate
|
|
fails) or no objective sentence is found — never another fund's.
|
|
"""
|
|
if not _on_cover(text, name, ticker, gate):
|
|
return None
|
|
heads = [m.start() for m in re.finditer(r"investment objective", text, re.I)]
|
|
if heads:
|
|
m = re.search(_SEEK, text[heads[0]: heads[0] + 2500], re.I)
|
|
if m:
|
|
return _clean(m.group(1))
|
|
m = re.search(_SEEK, text[:15000], re.I)
|
|
return _clean(m.group(1)) if m else None
|
|
|
|
|
|
def _clean(s: str) -> str:
|
|
# captures like "(the Fund) seeks to ..." leave a stray ')'
|
|
s = re.sub(r"^\)\s*", "", s)
|
|
s = re.sub(r"^((?:the|its) (?:fund|trust|etf|portfolio)"
|
|
r"(?: or (?:fund|trust|etf|portfolio))?)\s*\)\s*",
|
|
r"\1 ", s, flags=re.I)
|
|
s = re.sub(r"\s+", " ", s).strip()
|
|
if s[:1].islower():
|
|
s = s[0].upper() + s[1:]
|
|
return s
|
|
|
|
|
|
def classify_category(objective: str, name: str) -> str:
|
|
"""equity / fixed_income / mixed / alternatives / money_market / other.
|
|
|
|
Keyword rules over objective + fund name; deliberately transparent and
|
|
conservative (unknown -> 'other')."""
|
|
t = f"{objective} {name}".lower()
|
|
if re.search(r"money market", t):
|
|
return "money_market"
|
|
equity = bool(re.search(
|
|
r"equit|stocks?|common stock|capitalization|nasdaq.?100|s&p ?500 "
|
|
r"index|total stock market|russell ?2000", t))
|
|
bond = bool(re.search(
|
|
r"debt securit|bonds?\b|fixed.income|maturit|treasur|credit|yield|"
|
|
r"mortgage|interest.rate", t))
|
|
alt = bool(re.search(
|
|
r"commodit|derivatives|real estate|currenc|private (?:equity|credit)|"
|
|
r"hedge|bullion|precious metal|\bgold\b|\bsilver\b|\bcopper\b|"
|
|
r"\bnatural gas\b|\bcrude oil\b|\bwheat\b|\bcorn\b", t))
|
|
if equity and bond:
|
|
return "mixed"
|
|
if bond:
|
|
return "fixed_income"
|
|
if equity:
|
|
return "equity"
|
|
if alt:
|
|
return "alternatives"
|
|
return "other"
|
|
|
|
|
|
def accession_docs(cik: str, accession: str) -> list[str]:
|
|
"""All .htm files of a filing (family reports are split into parts)."""
|
|
url = (f"https://www.sec.gov/Archives/edgar/data/{int(cik)}"
|
|
f"/{accession.replace('-', '')}/index.json")
|
|
try:
|
|
d = json.loads(sec_get(url, timeout=60))
|
|
except Exception:
|
|
return []
|
|
return [it["name"] for it in d.get("directory", {}).get("item", [])
|
|
if it["name"].lower().endswith((".htm", ".html"))]
|
|
|
|
|
|
def _try_docs(hits: list[dict], name: str, ticker: str | None,
|
|
max_docs: int, gate: int) -> dict | None:
|
|
"""Fetch FTS hits (relevance order) and return the first document that
|
|
passes the cover gate and yields an objective sentence."""
|
|
fetched = 0
|
|
for h in hits:
|
|
if fetched >= max_docs:
|
|
break
|
|
# annual reports: the fund's part may be a sibling document of the
|
|
# same filing, so enumerate the accession's documents
|
|
names = [h["filename"]]
|
|
if h["form"] in ANNUAL_FORMS.split(","):
|
|
names += [n for n in accession_docs(h["cik"], h["accession"])
|
|
if n != h["filename"]]
|
|
for fn in names:
|
|
if fetched >= max_docs:
|
|
break
|
|
try:
|
|
raw = sec_get(doc_url(h["cik"], h["accession"], fn))
|
|
except urllib.error.HTTPError:
|
|
continue
|
|
fetched += 1
|
|
if len(raw) > MAX_DOC_BYTES:
|
|
continue
|
|
obj = extract_objective(to_text(raw), name, ticker, gate)
|
|
if obj:
|
|
return {
|
|
"objective": obj,
|
|
"category": classify_category(obj, name),
|
|
"form": h["form"],
|
|
"file_date": h["file_date"],
|
|
"url": doc_url(h["cik"], h["accession"], fn),
|
|
}
|
|
return None
|
|
|
|
|
|
def fetch_fund(fund_name: str, ticker: str | None = None,
|
|
max_docs: int = 8) -> dict | None:
|
|
"""Full pipeline for one fund. Returns metadata dict or None.
|
|
|
|
Passes, most direct first:
|
|
0. ticker -> registrant CIK (standalone trusts like Invesco QQQ
|
|
Trust / iShares Silver Trust): their most recent prospectus
|
|
filings, cover-gated;
|
|
1. fund-name search over prospectus forms — name in title position
|
|
or (TICKER) on the cover (COVER_GATE);
|
|
2. fund-name search over annual reports — name in title position
|
|
(TITLE_GATE);
|
|
3. ticker search over prospectus forms — (TICKER) in title position
|
|
(TITLE_GATE); catches funds whose registered name changed over
|
|
time.
|
|
Returns None rather than a wrong fund's objective.
|
|
"""
|
|
tk = ticker.upper() if ticker else None
|
|
if tk:
|
|
comp = ticker_to_company(tk)
|
|
if comp:
|
|
for fl in cik_recent_filings(comp[0], PROSPECTUS_FORMS, count=15):
|
|
if not fl["doc"]:
|
|
continue
|
|
try:
|
|
raw = sec_get(doc_url(comp[0], fl["accession"], fl["doc"]))
|
|
except urllib.error.HTTPError:
|
|
continue
|
|
if len(raw) > MAX_DOC_BYTES:
|
|
continue
|
|
obj = extract_objective(to_text(raw), fund_name, tk, COVER_GATE)
|
|
if obj:
|
|
return {
|
|
"objective": obj,
|
|
"category": classify_category(obj, fund_name),
|
|
"form": fl["form"],
|
|
"file_date": fl["filed"],
|
|
"url": doc_url(comp[0], fl["accession"], fl["doc"]),
|
|
}
|
|
q = f'"{fund_name}" "seeks"'
|
|
res = _try_docs(fts_search(q, PROSPECTUS_FORMS, size=100),
|
|
fund_name, tk, max_docs, COVER_GATE)
|
|
if res:
|
|
return res
|
|
res = _try_docs(fts_search(q, ANNUAL_FORMS, size=100),
|
|
fund_name, tk, max_docs, TITLE_GATE)
|
|
if res:
|
|
return res
|
|
if tk:
|
|
res = _try_docs(fts_search(f'"{tk}" "seeks"', PROSPECTUS_FORMS, size=100),
|
|
fund_name, tk, max_docs, TITLE_GATE)
|
|
if res:
|
|
return res
|
|
return None
|