"""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 http.client
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,
http.client.HTTPException) 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"(\d+)", xml)
name = re.search(r"([^<]+)", 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,
ciks: str | None = None) -> list[dict]:
"""EDGAR full-text search.
Hits are sorted by relevance, then date. `ciks` restricts to one
registrant — combined with the cover gates this is how a fund's own
filing is found inside a fund family's filings.
"""
params = {"q": query, "forms": forms or "", "size": str(size)}
if ciks:
params["ciks"] = ciks
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}\.)")
# alternate phrasing: "The Fund's investment objective is ... ." — the
# subject sits BEFORE the 'investment objective' anchor
_SEEK2 = (r"((?:the|its) (?:fund|trust|etf|portfolio)'?s?\s*"
r"investment objective is [^.]{5,300}\.?)")
# boilerplate to reject: "...investment objective is not fundamental ..."
_SEEK2_BAD = re.compile(
r"(?i)^(?:the|its) (?:fund|trust|etf|portfolio)'?s?\s*"
r"investment objective is not\b")
# yet another phrasing: "... (the Fund) investment objective is ... ."
_SEEK3 = r"((?:\(\s*the fund\s*\)?\s+investment objective is [^.]{5,300}\.))"
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 _objective_at(text: str, p: int) -> str | None:
"""Objective sentence at/below the heading at p (any phrasing)."""
m = re.search(_SEEK, text[p: p + 2500], re.I)
if m:
return _clean(m.group(1))
for pat in (_SEEK2, _SEEK3):
m = re.search(pat, text[max(0, p - 200): p + 2500], re.I)
if m and not _SEEK2_BAD.match(m.group(1)):
return _clean(m.group(1))
return None
def _first_objective(text: str) -> str | None:
"""Objective sentence after an 'Investment Objective' heading (TOC
entries come first and have no sentence — try them all), or the first
'seeks ...' sentence early in the document."""
heads = [m.start() for m in re.finditer(r"investment objective", text, re.I)]
for p in heads[:25]:
obj = _objective_at(text, p)
if obj:
return obj
m = re.search(_SEEK, text[:15000], re.I)
if m:
return _clean(m.group(1))
for pat in (_SEEK2, _SEEK3):
m = re.search(pat, text[:15000], re.I)
if m and not _SEEK2_BAD.match(m.group(1)):
return _clean(m.group(1))
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
return _first_objective(text)
# share-class / distribution words that Yahoo appends to fund names but
# that never appear in the fund's registered name
_STOP_WORDS = {"instl", "institutional", "investor", "retail", "admiral",
"shares", "share", "class", "fund", "r6", "r-6",
"r4", "r-4", "r3", "a", "b", "c", "i", "n", "z", "y"}
# Yahoo abbreviates fund names; map the common abbreviations back to the
# registered spelling so fuzzy matching sees the same words on both sides
_ABBR = {"mgd": "managed", "glbl": "global", "macr": "macro",
"abs": "absolute", "ret": "return", "advtg": "advantaged",
"strtgy": "strategy", "invst": "investment", "portf": "portfolio",
"opps": "opportunities", "mkt": "market", "eqy": "equity",
"incm": "income", "dist": "distribution", "div": "dividend",
"ltd": "limited", "intl": "international"}
FUZZY_GATE = 0.85 # name similarity needed to accept a document
FUZZY_FLOOR = 0.70 # weaker matches are still usable when RANKED against
# sibling funds' documents (CIK-scoped passes)
def _norm_name(name: str) -> str:
"""Lowercase, drop share-class words, expand Yahoo abbreviations — the
form used for fuzzy name matching (applied to names AND to the document
regions they are matched against)."""
out = []
for w in name.split():
w = w.lower().strip(".,'")
if w in _STOP_WORDS:
continue
out.append(_ABBR.get(w, w))
return " ".join(out)
def _sim(a: str, b: str) -> float:
from difflib import SequenceMatcher
return SequenceMatcher(None, a, b).ratio()
def _fuzzy_contains(region: str, target: str, threshold: float) -> float:
"""Best similarity of the target against the region, sliding a
target-length window (names wrap across lines in converted HTML, so
line-based matching is not enough). Both sides are normalized."""
region = _norm_name(re.sub(r"\s+", " ", region))
best = 0.0
for i in range(0, max(1, len(region) - len(target) + 1), 8):
s = _sim(target, region[i:i + len(target) + 6])
if s > best:
best = s
if best >= 1.0:
break
return best
def _fuzzy_cover(text: str, name: str, limit: int = 1500,
threshold: float = FUZZY_GATE) -> float:
"""Fuzzy cover gate: the fund's REGISTERED name (which differs from the
Yahoo name by abbreviations: Mgd/Managed, Glbl/Global, Macr/Macro) in
title position on the cover page. Returns the best similarity."""
target = _norm_name(name)
if len(target) < 8:
return 0.0
return _fuzzy_contains(text[:limit], target, threshold)
def extract_objective_family(text: str, name: str,
threshold: float = FUZZY_GATE
) -> tuple[float, str] | None:
"""Find the fund's section inside a FAMILY filing (one prospectus
covering many funds) and extract its objective.
The fund's own section header (name above the 'Fund Summary' / 'Investment
Objective' box) is fuzzy-matched against the fund name — robust to the
abbreviations Yahoo uses. Used only on documents of the fund's OWN
registrant (ticker -> CIK), where a name match is meaningful.
Returns (similarity, objective) or None.
"""
target = _norm_name(name)
if len(target) < 8:
return None
heads = [m.start() for m in re.finditer(r"investment objective", text, re.I)]
for p in heads:
s = _fuzzy_contains(text[max(0, p - 500):p], target, threshold)
if s >= threshold:
obj = _objective_at(text, p)
if obj:
return s, obj
return None
def _eval_doc(text: str, name: str, ticker: str | None = None) -> tuple[float, str]:
"""Run all document-level strategies in order: exact cover gate, fuzzy
cover gate (registered name differs by abbreviations), family section
(large multi-fund filings). Returns (similarity, objective) or None."""
if _on_cover(text, name, ticker, COVER_GATE):
obj = _first_objective(text)
if obj:
return 1.0, obj
sim = _fuzzy_cover(text, name)
if sim >= FUZZY_GATE:
obj = _first_objective(text)
if obj:
return sim, obj
if len(text) > 100_000:
res = extract_objective_family(text, name)
if res:
return res
return None
# section headings that terminate an investment-strategies excerpt
_STRATEGY_STOP = re.compile(
r"\b(?:principal risks|fees and expenses|fund performance|portfolio "
r"turnover|fund fees|purchasing|how to buy|share class)", re.I)
def extract_strategy(text: str, limit: int = 2600) -> str | None:
"""The fund's investment-strategies section, if present.
Prospectus supplements amend strategies inside amendment clauses, so the
best 'Investment Strategies' occurrence is chosen: one immediately
followed by actual strategy prose ("seeks", "invests", "Under normal
circumstances") wins."""
cands = list(re.finditer(
r"(?:[Pp]rincipal )?Investment [Ss]trateg(?:y|ies)"
r"|main investment strategies\??",
text, re.I))
best, best_score = None, -1
for m in cands:
seg = text[m.end(): m.end() + 1000]
score = 0
if re.search(r"\b(seeks|invests|investing|invest \w|"
r"under normal circumstances)\b", seg, re.I):
score += 3
if "?" in seg[:80]: # a Q&A section heading
score += 2
if re.search(r"\?\s*\d{1,3}\s", seg[:200]):
score -= 5 # table of contents ("... ? 8 Who ...")
if re.match(r"\s+risk\b", seg, re.I):
score -= 5 # a "...Investment Strategy Risk" subheading
if _STRATEGY_STOP.search(seg[:400]):
score -= 2
if score > best_score:
best, best_score = m, score
if best is None or best_score < 0:
# no heading: take the prose following the objective sentence
# (Fund Summary boxes run objective -> strategies in sequence)
m = re.search(_SEEK, text[:15000], re.I)
if not m:
return None
seg = text[m.end(): m.end() + limit + 400]
else:
seg = text[best.end(): best.end() + limit + 400]
stop = _STRATEGY_STOP.search(seg)
if stop:
seg = seg[:stop.start()]
out = re.sub(r"\s+", " ", seg).strip(" .")
return out[:limit] if len(out) > 100 else None
def _name_queries(name: str) -> list[str]:
"""FTS queries for a fund name: all significant words, then dropping
the SHORTEST words first (Yahoo abbreviations like Mgd/Glbl/Macr are
short and break quoted-phrase / word queries)."""
words = [w.strip(".,'") for w in name.split()
if w.lower().strip(".,'") not in _STOP_WORDS]
words = [w for w in words if len(w) > 1]
out = []
while len(words) >= 3:
out.append(" ".join(f'"{w}"' for w in words) + ' "seeks"')
words = sorted(words, key=len, reverse=True)[:-1] # drop shortest
return out
def _clean(s: str) -> str:
# "(the Fund) investment objective is ..." -> "The Fund's ..."
s = re.sub(r"^(?:\(\s*)?((?:the|its) (?:fund|trust|etf|portfolio))"
r"\s*\)?\s+investment objective is",
r"\1's investment objective is", s, flags=re.I)
s = re.sub(r"^\(\s*", "", s)
# 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, seen: set, floor: float = FUZZY_GATE) -> dict | None:
"""Fetch FTS hits (relevance order) and return the BEST document from
which an objective can be extracted (see _eval_doc). `floor` is the
minimum name similarity: CIK-scoped passes can accept weaker matches
(sibling funds' documents rank below the right one), other passes
require the full gate."""
best: tuple[float, dict] | None = None
fetched = 0
for h in hits:
if fetched >= max_docs:
break
if (h["cik"], h["accession"], h["filename"]) in seen:
continue
# 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
seen.add((h["cik"], h["accession"], fn))
fetched += 1
if len(raw) > MAX_DOC_BYTES:
continue
res = _eval_doc(to_text(raw), name, ticker)
if res and res[0] >= floor and (best is None or res[0] > best[0]):
best = (res[0], {
"objective": res[1],
"category": classify_category(res[1], name),
"form": h["form"],
"file_date": h["file_date"],
"url": doc_url(h["cik"], h["accession"], fn),
})
return best
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.
Every candidate document is run through _eval_doc (exact cover gate,
then fuzzy cover gate for registered names that differ from the Yahoo
name by abbreviations, then family-section matching). Passes, most
direct first:
0. ticker -> registrant CIK (the fund's own filer): its recent
prospectus filings, then a CIK-scoped full-text search over the
fund name's words (shortest dropped first, to shed Yahoo
abbreviations); the fuzzy gates are safe because the CIK is the
fund's own registrant;
1. fund-name search over prospectus forms;
2. fund-name search over annual reports;
3. fund-name WORDS over prospectus forms (abbreviation-tolerant);
4. ticker search over prospectus forms.
Returns None rather than a wrong fund's objective.
"""
tk = ticker.upper() if ticker else None
seen: set = set()
fallback: tuple[float, dict] | None = None
def _better(a: tuple[float, dict] | None,
b: tuple[float, dict] | None) -> tuple[float, dict] | None:
"""Best of two (similarity, metadata): higher sim, then newer date."""
if a is None:
return b
if b is None:
return a
ka = (a[0], a[1].get("file_date", ""))
kb = (b[0], b[1].get("file_date", ""))
return a if ka >= kb else b
def _try(query: str, forms: str, ciks: str | None = None,
floor: float = FUZZY_GATE) -> dict | None:
res = _try_docs(fts_search(query, forms, size=100, ciks=ciks),
fund_name, tk, max_docs, seen, floor=floor)
return res[1] if res else None
# 0) the fund's own registrant — documents here may carry WEAKER name
# matches (FUZZY_FLOOR): the right fund's document ranks above its
# siblings', so the best match across the CIK's documents wins
if tk:
comp = ticker_to_company(tk)
if comp:
cik = comp[0]
gate_cik: tuple[float, dict] | None = None
floor_cik: tuple[float, dict] | None = None
def _collect(res: tuple[float, dict] | None) -> None:
nonlocal gate_cik, floor_cik
if not res or res[0] < FUZZY_FLOOR:
return
if res[0] >= FUZZY_GATE:
gate_cik = _better(gate_cik, res)
else:
floor_cik = _better(floor_cik, res)
for fl in cik_recent_filings(cik, PROSPECTUS_FORMS, count=15):
if not fl["doc"] or (cik, fl["accession"], fl["doc"]) in seen:
continue
try:
raw = sec_get(doc_url(cik, fl["accession"], fl["doc"]))
except urllib.error.HTTPError:
continue
seen.add((cik, fl["accession"], fl["doc"]))
if len(raw) > MAX_DOC_BYTES:
continue
res = _eval_doc(to_text(raw), fund_name, tk)
if res:
_collect((res[0], {
"objective": res[1],
"category": classify_category(res[1], fund_name),
"form": fl["form"],
"file_date": fl["filed"],
"url": doc_url(cik, fl["accession"], fl["doc"]),
}))
for q in _name_queries(fund_name):
_collect(_try_docs(fts_search(q, PROSPECTUS_FORMS, size=100,
ciks=cik),
fund_name, tk, max_docs, seen,
floor=FUZZY_FLOOR))
if gate_cik:
return gate_cik[1]
# floor-tier (weaker name match) is a FALLBACK: the ordinary
# passes below may find a stronger document
fallback = _better(fallback, floor_cik)
# 1) exact fund name
for forms in (PROSPECTUS_FORMS, ANNUAL_FORMS):
res = _try(f'"{fund_name}" "seeks"', forms)
if res:
return res
# 3) fund name words (abbreviation-tolerant)
for q in _name_queries(fund_name):
res = _try(q, PROSPECTUS_FORMS)
if res:
return res
# 4) ticker
if tk:
res = _try(f'"{tk}" "seeks"', PROSPECTUS_FORMS)
if res:
return res
if fallback:
return fallback[1]
return None