Greg Pomerantz
54d26939dc
- fundlab/nport.py: parse the fund's own category/percentage lines, as-of date, net assets and dollar-valued positions from N-PORT schedules of investments (handles per-fund and combined multi-fund family filings; conservative keyword bucketing of positions) - fundlab/nport_cache/<sym>.json: parsed snapshots for 15 of 16 funds (raw SOI HTML kept locally, gitignored; source URLs + filing dates in nport_manifest.json, md5-verified against EDGAR) - atesx: no current SOI found (Anchor's recent filings cover the Income fund) - listed with an honest note - app.py: new 'Fund Lab' tab - pick a fund, see objective, reported composition (bar + table), rough keyword buckets, top positions, and the prospectus strategy excerpt - tests: parser unit tests (section finding, category regex, buckets)
342 lines
13 KiB
Python
342 lines
13 KiB
Python
"""N-PORT schedule-of-investments parsing for the Fund Lab page.
|
|
|
|
The raw SOI HTML files live in ``nport_cache/<sym>.html`` (one per fund,
|
|
fetched from EDGAR; see ``nport_manifest.json`` for the source URL and
|
|
filing date of each). ``build()`` parses them into a compact JSON snapshot
|
|
(``nport_cache/<sym>.json``) that the Streamlit page renders.
|
|
|
|
Parsing is deliberately conservative: each fund family formats its SOI
|
|
differently (tables, div layouts, combined multi-fund files), so we work
|
|
on a whitespace-normalized text *stream* and extract what is reliably
|
|
present:
|
|
|
|
* the fund's own category/percentage lines (the fund's reported
|
|
composition — the star of the report),
|
|
* the as-of date and net assets,
|
|
* position rows carrying a dollar value (where the layout allows),
|
|
* a coarse keyword bucketing of those positions.
|
|
|
|
Anything we cannot parse is simply absent from the snapshot — the page
|
|
shows what we have and flags the rest.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import bisect
|
|
import html as _html
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
CACHE = Path(__file__).parent / "nport_cache"
|
|
MANIFEST = Path(__file__).parent / "nport_manifest.json"
|
|
|
|
_MONTHS = ("January|February|March|April|May|June|July|August|September|"
|
|
"October|November|December")
|
|
|
|
# "<Fund name> ... Schedule of Investments" (variants)
|
|
_SECTION = re.compile(
|
|
r"(?:CONSOLIDATED\s+)?(?:SCHEDULE\s+OF\s+(?:PORTFOLIO\s+)?INVESTMENTS|"
|
|
r"PORTFOLIO\s+OF\s+INVESTMENTS)",
|
|
re.I)
|
|
|
|
# category: "NAME — 12.3%" / "NAME (12.3%)" / "NAME: 12.3%" / "NAME 12.3%"
|
|
# name: starts uppercase, letters and a few punctuation marks, 3..45 chars
|
|
# (the cap keeps merged table rows from spanning a whole line).
|
|
_CAT = re.compile(
|
|
r"(?<![A-Za-z0-9$])([A-Z][A-Za-z][A-Za-z0-9 ,/&'().\-]{1,43}?)"
|
|
r"\s*(?:[-–—:]|\(| )\s*(\d{1,3}\s?\.\d{1,2})\s*%")
|
|
|
|
|
|
def _trim_name(name: str) -> str:
|
|
"""Strip a mixed-case prefix (table header run-on) when the tail is an
|
|
all-caps category name: 'Maturity Fair Value CORPORATE BONDS' ->
|
|
'CORPORATE BONDS'. Title-case names are kept whole."""
|
|
parts = re.split(r"(?<=[a-z0-9\)-])\s+(?=[A-Z])", name)
|
|
if len(parts) > 1:
|
|
tail = parts[-1]
|
|
if tail.replace(" ", "").isupper() and len(tail) >= 4:
|
|
return tail
|
|
return name
|
|
|
|
_DOLLAR = re.compile(r"\$\s?(\d{1,3}(?:,\d{3})*(?:\.\d+)?|\d{2,}(?:\.\d+)?)")
|
|
# a line ending in a (comma-grouped) number: position value without "$"
|
|
_TRAILNUM = re.compile(r"([\d,]{4,}(?:\.\d+)?)\s*$")
|
|
|
|
_ASOF_LONG = re.compile(rf"\b((?:{_MONTHS}) \d{{1,2}}, \d{{4}})\b")
|
|
_ASOF_SHORT = re.compile(r"\b(\d{1,2}\.\d{1,2}\.\d{2,4})\b")
|
|
|
|
_NET = re.compile(r"net assets", re.I)
|
|
|
|
|
|
def rows(html: str) -> list[str]:
|
|
"""Row-aware HTML -> list of non-empty text lines (one per table row
|
|
/ block). Div-wrapped words in some filings stay on separate rows;
|
|
that is why most parsing is done on the joined *stream* instead."""
|
|
x = re.sub(r"(?i)</(tr|p|div|li|h[1-6])>|<br\s*/?>", "\n", html)
|
|
x = re.sub(r"<[^>]+>", " ", x)
|
|
x = _html.unescape(x).replace("\xa0", " ").replace("\r", " ")
|
|
out = []
|
|
for ln in x.split("\n"):
|
|
ln = re.sub(r"[ \t]+", " ", ln).strip()
|
|
if ln:
|
|
out.append(ln)
|
|
return out
|
|
|
|
|
|
def find_section(text: str, tokens: list[str]) -> tuple[int, int] | None:
|
|
"""Locate the fund's own SOI section (possibly in a multi-fund file).
|
|
|
|
``tokens`` are case-insensitive words that must appear in the section
|
|
header (a window around the "Schedule of Investments" phrase, which
|
|
contains or follows the fund name). The section extends over all
|
|
subsequent headers of the SAME fund (page-continuation footers) and
|
|
ends at the first header that no longer names all of the fund's
|
|
tokens.
|
|
"""
|
|
hits = list(_SECTION.finditer(text))
|
|
if not hits:
|
|
return None
|
|
|
|
def score(m: re.Match) -> int:
|
|
near = text[max(0, m.start() - 60): m.start()].lower()
|
|
before = text[max(0, m.start() - 150): m.start()].lower()
|
|
after = text[m.end(): m.end() + 80].lower()
|
|
s = sum(3 for t in tokens if t in near)
|
|
s += sum(1 for t in tokens
|
|
if t not in near and (t in before or t in after))
|
|
return s
|
|
|
|
def names_fund(m: re.Match) -> bool:
|
|
window = (text[max(0, m.start() - 150): m.start()]
|
|
+ text[m.end(): m.end() + 80]).lower()
|
|
return all(t in window for t in tokens)
|
|
|
|
best = None
|
|
for i, m in enumerate(hits):
|
|
s = score(m)
|
|
if s and (best is None or s > best[0]):
|
|
best = (s, i)
|
|
if best is None:
|
|
return None
|
|
start = max(0, hits[best[1]].start() - 200)
|
|
end = min(len(text), start + 400_000)
|
|
for j in range(best[1] + 1, len(hits)):
|
|
if names_fund(hits[j]):
|
|
continue
|
|
# cross-references never terminate, even mid-section
|
|
before = text[max(0, hits[j].start() - 30): hits[j].start()]
|
|
if before.strip().lower().endswith("see"):
|
|
continue
|
|
end = max(start, hits[j].start() - 150)
|
|
break
|
|
return start, end
|
|
|
|
|
|
def _to_float(s: str) -> float:
|
|
try:
|
|
return float(s.replace(",", ""))
|
|
except ValueError:
|
|
return 0.0
|
|
|
|
|
|
def parse_section(stream: str, start: int, end: int, rs: list[str],
|
|
row_starts: list[int]) -> dict:
|
|
seg = stream[start:end]
|
|
# the rows that fall inside the section (plus the partial first row,
|
|
# which may carry the section header)
|
|
i1 = bisect.bisect_right(row_starts, end)
|
|
i0 = bisect.bisect_left(row_starts, start)
|
|
if i0 and row_starts[i0] > start:
|
|
i0 -= 1
|
|
lines = [ln for ln in rs[i0:i1] if ln.strip()]
|
|
|
|
as_of = None
|
|
head = seg[:1500]
|
|
m = _ASOF_LONG.search(head) or _ASOF_SHORT.search(head)
|
|
if m:
|
|
as_of = m.group(1)
|
|
|
|
# -- categories (global, in order of appearance, deduplicated)
|
|
categories: list[dict] = []
|
|
seen: set[str] = set()
|
|
pos = 0
|
|
while len(categories) < 60:
|
|
cm = _CAT.search(seg, pos)
|
|
if cm is None:
|
|
break
|
|
name = _trim_name(cm.group(1).strip(" -–—:("))
|
|
# reject table-run-on junk: digits inside a category name (bond
|
|
# series, dates, page refs), stray cell markers ("a a a"), or
|
|
# column headers absorbed into the name. On a reject, resume from
|
|
# just after the name start so a clean match inside the same
|
|
# run-on is still found.
|
|
if (re.search(r"\d", name)
|
|
or re.search(r"\ba\s+a\b", name, re.I)
|
|
or re.search(r"\b(shares value|value shares|security shares|"
|
|
r"principal amount|fair value)\b", name, re.I)):
|
|
pos = cm.start(1) + 1
|
|
continue
|
|
key = name.lower()
|
|
if key not in seen:
|
|
seen.add(key)
|
|
categories.append({"name": name,
|
|
"pct": float(cm.group(2).replace(" ", ""))})
|
|
pos = cm.end()
|
|
# drop total/net-asset summary lines from the list
|
|
categories = [c for c in categories
|
|
if not re.match(r"^(total\s+)?(net\s+assets|investments)$",
|
|
c["name"], re.I)]
|
|
|
|
# -- positions: rows with a dollar value (last $ on the row) or,
|
|
# failing that, a trailing grouped number
|
|
positions: list[dict] = []
|
|
current_cat = ""
|
|
for ln in lines:
|
|
cm = _CAT.search(ln)
|
|
if cm and not _DOLLAR.search(ln):
|
|
current_cat = cm.group(1).strip()
|
|
dollars = _DOLLAR.findall(ln)
|
|
value = None
|
|
if dollars:
|
|
value = _to_float(dollars[-1])
|
|
elif _TRAILNUM.search(ln):
|
|
value = _to_float(_TRAILNUM.search(ln).group(1))
|
|
if value is None or value <= 0:
|
|
continue
|
|
if _NET.search(ln) or len(ln) < 8:
|
|
continue
|
|
if re.fullmatch(r"[\d,$.()\s]+", ln): # a bare number row
|
|
continue
|
|
positions.append({"text": re.sub(r"\s+", " ", ln)[:160],
|
|
"value": value, "cat": current_cat})
|
|
|
|
# -- net assets: last "net assets" mention with a nearby number
|
|
net_assets = None
|
|
for nm in list(_NET.finditer(seg))[-3:]:
|
|
window = seg[nm.end(): nm.end() + 120]
|
|
dm = _DOLLAR.search(window) or re.search(r"([\d,]{8,}(?:\.\d+)?)", window)
|
|
if dm:
|
|
net_assets = _to_float(dm.group(1))
|
|
break
|
|
return {"as_of": as_of, "categories": categories,
|
|
"positions": positions, "net_assets": net_assets}
|
|
|
|
|
|
# ------------------------------------------------------------------ buckets
|
|
_B_RE = [
|
|
("Futures", r"\b(futur|forward|swap)\b|\bOPEN (CURRENCY|FUTURES)\b|\bCURRENCY (EXCHANGE|FORWARD|OPTION)\b"),
|
|
("Commodities", r"\b(gold|silver|copper|crude|natural gas|oil|energy|agricultur|commodit|palladium|platinum|wheat|corn)\b"),
|
|
("Fund holdings", r"\b(fund|etf)\b, ?class |class [a-z0-9]{1,2}\b|\bexchange[- ]trad(?:ed)? (fund|funds|etf)|\b(open[- ]end|closed[- ]end) fund|\bmoney market fund\b|\bETF\b|\bFund\b(?!s of Investments)"),
|
|
("Agency MBS", r"\bmortgage\b|\bgnma\b|\bfhlmc\b|\bfnma\b|\bagency\b|\bmbs\b"),
|
|
("CMBS / ABS", r"\bcmo\b|commercial mortgage|collateralized mortgage|abs (?:trust|securitization)|asset[- ]?backed"),
|
|
("High yield", r"high[- ]?yield"),
|
|
("US govt", r"treasur|us govt|u\.s\. government|t-bill"),
|
|
("IG credit / munis", r"corporate|sovereign|senior note|notes,|bonds,|munici|nonconvertible"),
|
|
("Cash & T-bills", r"^\s*cash\b|cash item|bank deposit|money market"),
|
|
("Equity (intl)", r"\b(foreign|non-?us|international|emerging|europe|japan|canada|\buk\b|german|french|austral|brazil|india|china|korea|switz|netherland|spain|italy)\b"),
|
|
("Equity (US)", r"\b(inc|corp|corporation|ltd|plc|group|llc|lp)\.?|common stock|equity|stock"),
|
|
]
|
|
_BUCKETS = [(name, re.compile(rx, re.I)) for name, rx in _B_RE]
|
|
|
|
|
|
def classify(text: str, category: str) -> str:
|
|
"""Keyword-bucket one position. Order matters: first match wins."""
|
|
ctx = f"{category} | {text}"
|
|
for name, rx in _BUCKETS:
|
|
if rx.search(ctx):
|
|
return name
|
|
return "Other"
|
|
|
|
|
|
def build(sym: str, fund_tokens: list[str], force: bool = False) -> dict | None:
|
|
"""Parse the cached SOI for ``sym``; returns the snapshot (or None).
|
|
|
|
``fund_tokens`` identifies this fund's section inside multi-fund
|
|
family files (e.g. ["mortgage", "opportunities"]).
|
|
"""
|
|
html_path = CACHE / f"{sym}.html"
|
|
json_path = CACHE / f"{sym}.json"
|
|
if not force and json_path.exists():
|
|
try:
|
|
return json.loads(json_path.read_text())
|
|
except Exception:
|
|
pass
|
|
if not html_path.exists():
|
|
return None
|
|
rs = rows(html_path.read_text())
|
|
stream = " ".join(rs)
|
|
row_starts: list[int] = []
|
|
pos = 0
|
|
for r in rs:
|
|
row_starts.append(pos)
|
|
pos += len(r) + 1
|
|
sec = find_section(stream, fund_tokens)
|
|
if sec is None:
|
|
return None
|
|
seg = parse_section(stream, *sec, rs, row_starts)
|
|
buckets: dict[str, float] = {}
|
|
for p in seg["positions"]:
|
|
b = classify(p["text"], p["cat"])
|
|
buckets[b] = buckets.get(b, 0.0) + p["value"]
|
|
total = sum(buckets.values()) or 1.0
|
|
snap = {
|
|
"sym": sym,
|
|
"as_of": seg["as_of"],
|
|
"net_assets": seg["net_assets"],
|
|
"n_positions": len(seg["positions"]),
|
|
"categories": seg["categories"],
|
|
"buckets": [{"name": k, "value": v, "pct": 100.0 * v / total}
|
|
for k, v in sorted(buckets.items(), key=lambda kv: -kv[1])],
|
|
"top": sorted(seg["positions"], key=lambda p: -p["value"])[:40],
|
|
}
|
|
json_path.write_text(json.dumps(snap, indent=1))
|
|
return snap
|
|
|
|
|
|
# ------------------------------------------------------------------ CLI
|
|
# fund-identifying tokens per ticker (used to find the fund's section in
|
|
# multi-fund family filings)
|
|
FUND_TOKENS = {
|
|
# atesx: no cached SOI (its family's current filings don't cover it);
|
|
# it still shows in the list with the manifest's note
|
|
"atesx": ["anchor", "risk", "equity"],
|
|
"atrfx": ["catalyst", "systematic"],
|
|
"cvsix": ["calamos", "market"],
|
|
"jlpsx": ["large cap core plus"],
|
|
"pmaix": ["victory", "pioneer"],
|
|
"pmfkx": ["victory", "pioneer"],
|
|
"pmorx": ["mortgage", "opportunities"],
|
|
"qspnx": ["style", "premia"],
|
|
"svarx": ["spectrum low volatility"],
|
|
"cosix": ["columbia strategic income"],
|
|
"mbxix": ["millburn", "hedge"],
|
|
"eagmx": ["global macro absolute return"],
|
|
"egrsx": ["global macro absolute return"],
|
|
"lcorx": ["leuthold core"],
|
|
"lcrix": ["leuthold core"],
|
|
"lamhx": ["dividend growth"],
|
|
}
|
|
|
|
|
|
def build_all(force: bool = False) -> dict:
|
|
"""Build/return snapshots for every fund with a cached SOI file."""
|
|
out = {}
|
|
for sym, tk in FUND_TOKENS.items():
|
|
s = build(sym, tk, force=force)
|
|
if s:
|
|
out[sym] = s
|
|
return out
|
|
|
|
|
|
def manifest() -> dict:
|
|
return json.loads(MANIFEST.read_text())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
syms = sys.argv[1:] or list(FUND_TOKENS)
|
|
for s in syms:
|
|
r = build(s, FUND_TOKENS[s], force=True)
|
|
print(s, "->", "ok" if r else "FAILED")
|