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)
649 lines
26 KiB
Python
649 lines
26 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 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"<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,
|
|
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
|