You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1758 lines
80 KiB
Python
1758 lines
80 KiB
Python
"""
|
|
ETF holdings fetcher: given ETF ticker and optional date, return holdings at that date
|
|
using the closest prior filing. Converts CUSIP -> ticker using local mapping.
|
|
|
|
This fetcher uses SEC submissions endpoint to locate NPORT-P filings for the ETF's CIK,
|
|
and downloads the primary XML to parse holdings (CUSIP, shares, value). It then maps
|
|
CUSIP to ticker by local `cusip_map` table when possible.
|
|
"""
|
|
|
|
from typing import List, Dict, Optional, Tuple
|
|
import time as _time
|
|
from datetime import datetime, timedelta, timezone
|
|
import aiohttp
|
|
import asyncio
|
|
from bs4 import BeautifulSoup
|
|
import re
|
|
import os
|
|
import json
|
|
import hashlib
|
|
import random
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, and_, delete
|
|
|
|
from app.models.etf import ETFCIKMap, CusipMap, ETFSeriesMap, ETFHoldingsSnapshot, ETFHolding
|
|
from app.core.config import settings
|
|
from app.services.sec_http_client import SECHttpClient
|
|
|
|
# Optional external ETF scraper integration (provider websites)
|
|
try:
|
|
from etf_scraper import ETFScraper
|
|
_HAS_ETF_SCRAPER = True
|
|
except Exception:
|
|
_HAS_ETF_SCRAPER = False
|
|
|
|
|
|
class ETFHoldingsFetcher:
|
|
def __init__(self):
|
|
self._http = SECHttpClient("Stock Oracle ETF Fetcher")
|
|
self.sec_base_data = self._http.sec_base_data
|
|
self.sec_base_archives = "https://www.sec.gov/Archives/edgar/data"
|
|
# Keep legacy attributes that point to shared client internals
|
|
self.http_timeout = self._http.http_timeout
|
|
self._req_sem = self._http._req_sem
|
|
self._text_cache = self._http._text_cache
|
|
self._json_cache = self._http._json_cache
|
|
self._cache_dir = self._http._cache_dir
|
|
self._user_agent = self._http._user_agent
|
|
|
|
@property
|
|
def _deadline(self) -> Optional[float]:
|
|
return self._http._deadline
|
|
|
|
@_deadline.setter
|
|
def _deadline(self, value: Optional[float]) -> None:
|
|
self._http._deadline = value
|
|
|
|
def _normalize_snapshot_datetime(self, dt: datetime) -> datetime:
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
|
|
async def _persist_snapshot_and_holdings(
|
|
self,
|
|
db: AsyncSession,
|
|
*,
|
|
ticker: str,
|
|
snapshot_date: datetime,
|
|
source: str,
|
|
cik: Optional[str],
|
|
filing_accession: Optional[str],
|
|
xml_url: Optional[str],
|
|
metadata: Optional[Dict],
|
|
holdings: List[Dict],
|
|
) -> Optional[str]:
|
|
try:
|
|
snap_dt = self._normalize_snapshot_datetime(snapshot_date)
|
|
existing = await db.execute(
|
|
select(ETFHoldingsSnapshot).where(
|
|
and_(
|
|
ETFHoldingsSnapshot.ticker == ticker,
|
|
ETFHoldingsSnapshot.snapshot_date == snap_dt,
|
|
)
|
|
)
|
|
)
|
|
snapshot = existing.scalar_one_or_none()
|
|
if snapshot is None:
|
|
snapshot = ETFHoldingsSnapshot(
|
|
ticker=ticker,
|
|
snapshot_date=snap_dt,
|
|
source=source,
|
|
cik=cik,
|
|
filing_accession=filing_accession,
|
|
xml_url=xml_url,
|
|
metadata_json=metadata or {},
|
|
)
|
|
db.add(snapshot)
|
|
await db.flush()
|
|
else:
|
|
snapshot.source = source
|
|
snapshot.cik = cik
|
|
snapshot.filing_accession = filing_accession
|
|
snapshot.xml_url = xml_url
|
|
snapshot.metadata_json = metadata or {}
|
|
await db.execute(delete(ETFHolding).where(ETFHolding.snapshot_id == snapshot.id))
|
|
|
|
for h in holdings or []:
|
|
try:
|
|
db.add(
|
|
ETFHolding(
|
|
snapshot_id=snapshot.id,
|
|
name=h.get("name"),
|
|
cusip=h.get("cusip"),
|
|
ticker=h.get("ticker"),
|
|
shares=(float(h.get("shares")) if h.get("shares") is not None else None),
|
|
value=(float(h.get("value")) if h.get("value") is not None else None),
|
|
percentage=(float(h.get("percentage")) if h.get("percentage") is not None else None),
|
|
)
|
|
)
|
|
except Exception:
|
|
continue
|
|
await db.commit()
|
|
return str(snapshot.id)
|
|
except Exception:
|
|
try:
|
|
await db.rollback()
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
async def _load_snapshot_holdings(
|
|
self,
|
|
db: AsyncSession,
|
|
*,
|
|
ticker: str,
|
|
as_of_date: Optional[datetime],
|
|
) -> Optional[Dict]:
|
|
tkr = ticker.upper()
|
|
if as_of_date is not None:
|
|
dt_norm = self._normalize_snapshot_datetime(as_of_date)
|
|
q = select(ETFHoldingsSnapshot).where(
|
|
and_(
|
|
ETFHoldingsSnapshot.ticker == tkr,
|
|
ETFHoldingsSnapshot.snapshot_date <= dt_norm,
|
|
)
|
|
).order_by(ETFHoldingsSnapshot.snapshot_date.desc()).limit(1)
|
|
else:
|
|
q = select(ETFHoldingsSnapshot).where(ETFHoldingsSnapshot.ticker == tkr).order_by(ETFHoldingsSnapshot.snapshot_date.desc()).limit(1)
|
|
res = await db.execute(q)
|
|
snap = res.scalar_one_or_none()
|
|
if not snap:
|
|
return None
|
|
hres = await db.execute(select(ETFHolding).where(ETFHolding.snapshot_id == snap.id))
|
|
rows = hres.scalars().all()
|
|
holdings = []
|
|
for r in rows:
|
|
holdings.append({
|
|
"name": r.name,
|
|
"cusip": r.cusip,
|
|
"ticker": r.ticker,
|
|
"shares": r.shares,
|
|
"value": r.value,
|
|
"percentage": r.percentage,
|
|
})
|
|
return {"snapshot": snap, "holdings": holdings}
|
|
|
|
def _quick_count_holdings(self, xml_text: str) -> int:
|
|
"""Fast approximate count of holdings without full XML parsing.
|
|
Counts occurrences of the most common holding element names.
|
|
"""
|
|
if not xml_text:
|
|
return 0
|
|
try:
|
|
# Simple lowercase search with regex allowing namespace prefixes
|
|
lt = xml_text.lower()
|
|
import re as _re
|
|
cnt = 0
|
|
cnt += len(_re.findall(r"<[^>]*fundreportedholding", lt))
|
|
cnt += len(_re.findall(r"<[^>]*invstorsec", lt))
|
|
cnt += len(_re.findall(r"<[^>]*investmentorsec", lt))
|
|
return cnt
|
|
except Exception:
|
|
return 0
|
|
|
|
def _cache_path(self, url: str) -> str:
|
|
return self._http._cache_path(url)
|
|
|
|
async def _get_cik_record(self, db: AsyncSession, ticker: str) -> Optional[Dict[str, Optional[str]]]:
|
|
result = await db.execute(select(ETFCIKMap.cik, ETFCIKMap.name).where(ETFCIKMap.ticker == ticker.upper()))
|
|
row = result.first()
|
|
if not row:
|
|
return None
|
|
# series/class override if available
|
|
srow = await db.execute(select(ETFSeriesMap.series_id, ETFSeriesMap.class_id).where(ETFSeriesMap.ticker == ticker.upper()))
|
|
s = srow.first()
|
|
return {"cik": row[0], "name": row[1], "series_id": (s[0] if s else None), "class_id": (s[1] if s else None)}
|
|
|
|
async def _fetch_json(self, url: str) -> dict:
|
|
return await self._http.fetch_json(url)
|
|
|
|
async def _fetch_text(self, url: str) -> str:
|
|
return await self._http.fetch_text(
|
|
url, accept="application/xml, text/xml;q=0.9, text/html;q=0.8"
|
|
)
|
|
|
|
async def _find_best_filing_and_xml(
|
|
self,
|
|
cik: str,
|
|
target_date: Optional[datetime],
|
|
*,
|
|
ticker: Optional[str] = None,
|
|
fund_name: Optional[str] = None,
|
|
series_id: Optional[str] = None,
|
|
class_id: Optional[str] = None,
|
|
scan_limit: int = 18,
|
|
) -> Optional[Tuple[str, str, str]]:
|
|
"""Return (filing_date, accession_number, xml_url) for closest prior NPORT-P filing.
|
|
|
|
For trust-level CIKs with many series, scan recent accessions and pick the XML whose
|
|
<seriesName> best matches the fund_name tokens or whose index hints include the ticker.
|
|
"""
|
|
cik_digits = "".join(ch for ch in str(cik) if ch.isdigit())
|
|
if not cik_digits:
|
|
return None
|
|
url = f"{self.sec_base_data}/submissions/CIK{int(cik_digits):010d}.json"
|
|
data = await self._fetch_json(url)
|
|
filings: List[Tuple[datetime, str]] = []
|
|
def add_from_block(block: dict):
|
|
forms = block.get("form", [])
|
|
dates = block.get("filingDate", [])
|
|
accessions = block.get("accessionNumber", [])
|
|
for form, dt_str, acc in zip(forms, dates, accessions):
|
|
if form != "NPORT-P":
|
|
continue
|
|
try:
|
|
dt = datetime.strptime(dt_str, "%Y-%m-%d").replace(tzinfo=timezone.utc)
|
|
except Exception:
|
|
continue
|
|
filings.append((dt, acc))
|
|
|
|
recent = data.get("filings", {}).get("recent", {})
|
|
add_from_block(recent)
|
|
|
|
# If target_date is earlier than the earliest in recent, fetch older yearly submission files
|
|
files_meta = data.get("filings", {}).get("files", []) or []
|
|
if target_date and filings:
|
|
earliest_dt = min(dt for dt, _ in filings)
|
|
if earliest_dt > target_date and files_meta:
|
|
# Try up to 6 older files
|
|
for meta in files_meta[:6]:
|
|
name = meta.get("name")
|
|
if not name:
|
|
continue
|
|
# name looks like 'CIK0001100663-2020.json'
|
|
older_url = f"{self.sec_base_data}/submissions/{name}"
|
|
try:
|
|
older = await self._fetch_json(older_url)
|
|
older_recent = older.get("filings", {}).get("recent", {})
|
|
add_from_block(older_recent)
|
|
earliest_dt = min(dt for dt, _ in filings)
|
|
if earliest_dt <= target_date:
|
|
break
|
|
except Exception:
|
|
continue
|
|
if not filings:
|
|
return None
|
|
filings.sort(key=lambda x: x[0])
|
|
|
|
# Candidate list: newest first, filtered by date if provided.
|
|
# Ensure we include multiple prior days (not just the top day) to avoid missing the target series.
|
|
if target_date:
|
|
eligible = [(dt, acc) for dt, acc in filings if dt <= target_date]
|
|
if not eligible:
|
|
early_dt, early_acc = filings[0]
|
|
xml_url = await self._download_primary_xml(
|
|
cik_digits,
|
|
early_acc,
|
|
ticker=ticker,
|
|
fund_name=fund_name,
|
|
series_id=series_id,
|
|
class_id=class_id,
|
|
)
|
|
return (early_dt.strftime("%Y-%m-%d"), early_acc, xml_url)
|
|
else:
|
|
eligible = filings
|
|
eligible_rev = list(reversed(eligible))
|
|
# Group by date (YYYY-MM-DD)
|
|
grouped: Dict[str, List[Tuple[datetime, str]]] = {}
|
|
for dt, acc in eligible_rev:
|
|
key = dt.strftime("%Y-%m-%d")
|
|
grouped.setdefault(key, []).append((dt, acc))
|
|
# Interleave per-day groups: take up to per_day_limit from each day, then next day, etc.
|
|
per_day_limit = 60
|
|
total_limit = max(200, scan_limit * 10)
|
|
cands: List[Tuple[datetime, str]] = []
|
|
for day_key in grouped.keys():
|
|
day_items = grouped[day_key][:per_day_limit]
|
|
cands.extend(day_items)
|
|
if len(cands) >= total_limit:
|
|
break
|
|
|
|
# Establish time budget for the remainder of this selection
|
|
deadline = _time.monotonic() + 22.0
|
|
|
|
# Primary-doc first pass: avoid index.htm to reduce 429s.
|
|
# Try a limited number of accessions, fetch primary XML(s), and score by series match and holdings count.
|
|
sid = (series_id or "").strip()
|
|
cid = (class_id or "").strip() if class_id else None
|
|
primary_limit = 40
|
|
best_primary: Optional[Tuple[int, datetime, str, str]] = None # (score, dt, acc, url)
|
|
for dt, acc in cands[:primary_limit]:
|
|
if _time.monotonic() > deadline:
|
|
break
|
|
acc_clean = acc.replace("-", "")
|
|
for rel in ("primary_doc.xml", "xslFormNPORT-P_X01/primary_doc.xml"):
|
|
if _time.monotonic() > deadline:
|
|
break
|
|
try:
|
|
base_url = f"{self.sec_base_archives}/{int(cik_digits)}/{acc_clean}/"
|
|
url = base_url + rel
|
|
xml_text = await self._fetch_text(url)
|
|
score = 0
|
|
if sid and sid in xml_text:
|
|
score += 120
|
|
if cid and cid in xml_text:
|
|
score += 40
|
|
# Token-based signal (partial match threshold)
|
|
if fund_name:
|
|
fname_l = fund_name.strip().lower()
|
|
toks = [tok for tok in fname_l.replace("(", " ").replace(")", " ").replace(",", " ").split() if len(tok) >= 3]
|
|
if toks:
|
|
low = xml_text.lower()
|
|
matched = sum(1 for tok in toks if tok in low)
|
|
if matched >= max(2, len(toks) // 2):
|
|
score += 30
|
|
# Quick approximate holdings count to avoid heavy parsing in this pass
|
|
try:
|
|
cnt = self._quick_count_holdings(xml_text)
|
|
score += max(0, 40 - abs(cnt - 129))
|
|
if 80 <= cnt <= 200:
|
|
# Return early only if we also have a strong identity signal
|
|
strong_match = False
|
|
if sid and sid in xml_text:
|
|
strong_match = True
|
|
if cid and cid in xml_text:
|
|
strong_match = True
|
|
if not strong_match and fund_name:
|
|
fname_l = fund_name.strip().lower()
|
|
toks = [tok for tok in fname_l.replace("(", " ").replace(")", " ").replace(",", " ").split() if len(tok) >= 3]
|
|
if toks:
|
|
low = xml_text.lower()
|
|
matched = sum(1 for tok in toks if tok in low)
|
|
if matched >= max(2, len(toks) // 2):
|
|
strong_match = True
|
|
if strong_match:
|
|
return (dt.strftime("%Y-%m-%d"), acc, url)
|
|
except Exception:
|
|
pass
|
|
if best_primary is None or score > best_primary[0] or (score == best_primary[0] and dt > best_primary[1]):
|
|
best_primary = (score, dt, acc, url)
|
|
except Exception:
|
|
continue
|
|
|
|
# Prepare tokens
|
|
fname = (fund_name or "").strip().lower()
|
|
tokens = [tok for tok in fname.replace("(", " ").replace(")", " ").replace(",", " ").split() if len(tok) >= 3]
|
|
if ticker:
|
|
tkn = ticker.strip().lower()
|
|
if tkn and tkn not in tokens:
|
|
tokens.append(tkn)
|
|
|
|
# Try to find matching seriesName in XML for each accession; also test alternates when holdings too low
|
|
fallback: Optional[Tuple[str, str, str]] = None
|
|
sid = (series_id or "").strip()
|
|
cid = (class_id or "").strip()
|
|
# Build list of (dt, acc, candidate_urls)
|
|
filing_candidates: List[Tuple[datetime, str, List[str]]] = []
|
|
for dt, acc in cands:
|
|
if _time.monotonic() > deadline:
|
|
break
|
|
try:
|
|
urls = await self._list_candidate_docs(
|
|
cik_digits,
|
|
acc,
|
|
ticker=ticker,
|
|
fund_name=fund_name,
|
|
series_id=series_id,
|
|
class_id=class_id,
|
|
)
|
|
if urls:
|
|
filing_candidates.append((dt, acc, urls[:10]))
|
|
except Exception:
|
|
continue
|
|
|
|
# Flatten URLs with priority (newer filings first)
|
|
flat: List[Tuple[datetime, str, str]] = []
|
|
for dt, acc, urls in filing_candidates:
|
|
for u in urls:
|
|
flat.append((dt, acc, u))
|
|
|
|
# Limit total candidates to avoid long scans
|
|
flat = flat[:60]
|
|
|
|
# Concurrently evaluate candidates with a small pool
|
|
from asyncio import Semaphore, create_task, wait, FIRST_COMPLETED
|
|
sem = Semaphore(6)
|
|
|
|
async def eval_candidate(dt: datetime, acc: str, url: str):
|
|
if _time.monotonic() > deadline:
|
|
return None
|
|
async with sem:
|
|
try:
|
|
xml_text = await self._fetch_text(url)
|
|
# Score
|
|
score = 0
|
|
# Fast-path: exact series/class id substring in XML text
|
|
if sid and sid in xml_text:
|
|
score += 200
|
|
if cid and cid in xml_text:
|
|
score += 50
|
|
# Token hint from fund name in raw text to avoid full parse (partial match)
|
|
if tokens:
|
|
low = xml_text.lower()
|
|
matched = sum(1 for tok in tokens if tok in low)
|
|
if matched >= max(2, len(tokens) // 2):
|
|
score += 30
|
|
# Approximate holdings count quickly to avoid expensive parsing per candidate
|
|
try:
|
|
cnt = self._quick_count_holdings(xml_text)
|
|
# Prefer close to 129
|
|
score += max(0, 40 - abs(cnt - 129))
|
|
# Penalize very large counts
|
|
if cnt > 400:
|
|
score -= 120
|
|
except Exception:
|
|
cnt = 0
|
|
return (score, dt, acc, url)
|
|
except Exception:
|
|
return None
|
|
|
|
tasks = [create_task(eval_candidate(dt, acc, url)) for dt, acc, url in flat]
|
|
best: Optional[Tuple[int, datetime, str, str]] = None
|
|
for t in tasks:
|
|
if _time.monotonic() > deadline:
|
|
break
|
|
done, pending = await wait(tasks, timeout=0.2, return_when=FIRST_COMPLETED)
|
|
for d in done:
|
|
res = d.result()
|
|
if res is None:
|
|
continue
|
|
if best is None or res[0] > best[0] or (res[0] == best[0] and res[1] > best[1]):
|
|
best = res
|
|
if best and best[0] >= 100:
|
|
# Exact series match found
|
|
break
|
|
|
|
# Cancel remaining tasks
|
|
for t in tasks:
|
|
if not t.done():
|
|
t.cancel()
|
|
|
|
if best:
|
|
_, dt, acc, url = best
|
|
return (dt.strftime("%Y-%m-%d"), acc, url)
|
|
|
|
# If primary pass produced a fallback, validate it has holdings; otherwise, try index candidates for that accession
|
|
if best_primary:
|
|
_, dt, acc, url = best_primary
|
|
try:
|
|
xml_text = await self._fetch_text(url)
|
|
cnt = self._quick_count_holdings(xml_text)
|
|
strong_match = False
|
|
if sid and sid in xml_text:
|
|
strong_match = True
|
|
if cid and cid in xml_text:
|
|
strong_match = True
|
|
if not strong_match and fund_name:
|
|
fname_l = fund_name.strip().lower()
|
|
toks = [tok for tok in fname_l.replace("(", " ").replace(")", " ").replace(",", " ").split() if len(tok) >= 3]
|
|
if toks:
|
|
low = xml_text.lower()
|
|
matched = sum(1 for tok in toks if tok in low)
|
|
if matched >= max(2, len(toks) // 2):
|
|
strong_match = True
|
|
except Exception:
|
|
cnt = 0
|
|
strong_match = False
|
|
if cnt >= 50 and strong_match:
|
|
return (dt.strftime("%Y-%m-%d"), acc, url)
|
|
# Try index candidates for this accession with stronger size/score preference and pick first with decent count
|
|
try:
|
|
cand_urls = await self._list_candidate_docs(cik_digits, acc, ticker=ticker, fund_name=fund_name, series_id=series_id, class_id=class_id)
|
|
best_cand: Optional[Tuple[int, str]] = None # (cnt, url)
|
|
for cu in cand_urls[:20]:
|
|
try:
|
|
xt = await self._fetch_text(cu)
|
|
c = self._quick_count_holdings(xt)
|
|
if c >= 80:
|
|
return (dt.strftime("%Y-%m-%d"), acc, cu)
|
|
if best_cand is None or c > best_cand[0]:
|
|
best_cand = (c, cu)
|
|
except Exception:
|
|
continue
|
|
if best_cand is not None and best_cand[0] > 0:
|
|
return (dt.strftime("%Y-%m-%d"), acc, best_cand[1])
|
|
except Exception:
|
|
pass
|
|
|
|
if fallback:
|
|
return fallback
|
|
# Last resort: use best candidate from index (not necessarily primary) of newest prior
|
|
dt, acc = cands[0]
|
|
try:
|
|
cand_urls = await self._list_candidate_docs(cik_digits, acc, ticker=ticker, fund_name=fund_name, series_id=series_id, class_id=class_id)
|
|
best_cand: Optional[Tuple[int, str]] = None
|
|
for cu in cand_urls[:20]:
|
|
try:
|
|
xt = await self._fetch_text(cu)
|
|
c = self._quick_count_holdings(xt)
|
|
if c >= 80:
|
|
return (dt.strftime("%Y-%m-%d"), acc, cu)
|
|
if best_cand is None or c > best_cand[0]:
|
|
best_cand = (c, cu)
|
|
except Exception:
|
|
continue
|
|
# Fallback to primary if nothing else
|
|
xml_url = best_cand[1] if (best_cand and best_cand[0] > 0) else await self._download_primary_xml(cik_digits, acc, ticker=ticker, fund_name=fund_name, series_id=series_id, class_id=class_id)
|
|
return (dt.strftime("%Y-%m-%d"), acc, xml_url)
|
|
except Exception:
|
|
xml_url = await self._download_primary_xml(cik_digits, acc, ticker=ticker, fund_name=fund_name, series_id=series_id, class_id=class_id)
|
|
return (dt.strftime("%Y-%m-%d"), acc, xml_url)
|
|
|
|
async def _download_primary_xml(self, cik: str, accession_number: str, ticker: Optional[str] = None, fund_name: Optional[str] = None, series_id: Optional[str] = None, class_id: Optional[str] = None) -> Optional[str]:
|
|
"""Find a primary XML URL for a filing. Prefer direct primary paths to avoid index 429."""
|
|
# CIK in archives path is typically non-padded digits
|
|
cik_digits = "".join(ch for ch in str(cik) if ch.isdigit())
|
|
acc_clean = accession_number.replace("-", "")
|
|
base_dir = f"{self.sec_base_archives}/{int(cik_digits)}/{acc_clean}/"
|
|
# First try direct primary paths
|
|
for rel in ("primary_doc.xml", "xslFormNPORT-P_X01/primary_doc.xml"):
|
|
try:
|
|
url = base_dir + rel
|
|
_ = await self._fetch_text(url)
|
|
return url
|
|
except Exception:
|
|
continue
|
|
# Fallback to index page if direct paths fail
|
|
index_url = f"{base_dir}{accession_number}-index.htm"
|
|
html = await self._fetch_text(index_url)
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
|
|
def mk_abs(href: str) -> str:
|
|
if href.startswith("/"):
|
|
return f"https://www.sec.gov{href}"
|
|
return f"{index_url.rsplit('/', 1)[0]}/{href}"
|
|
|
|
# Collect candidate XML docs that likely contain full holdings
|
|
candidates = [] # list of (score, size_bytes, url)
|
|
|
|
def parse_size(text: str) -> int:
|
|
t = (text or "").upper().strip()
|
|
num = 0.0
|
|
unit = 1
|
|
parts = t.split()
|
|
if not parts:
|
|
return 0
|
|
try:
|
|
num = float(parts[0])
|
|
except Exception:
|
|
return 0
|
|
if len(parts) > 1:
|
|
u = parts[1]
|
|
if u.startswith("KB"):
|
|
unit = 1024
|
|
elif u.startswith("MB"):
|
|
unit = 1024 * 1024
|
|
elif u.startswith("B"):
|
|
unit = 1
|
|
return int(num * unit)
|
|
|
|
tkr = (ticker or "").strip().lower()
|
|
fname = (fund_name or "").strip().lower()
|
|
# Tokenize fund name to boost relevance
|
|
name_tokens = [tok for tok in fname.replace("(", " ").replace(")", " ").replace(",", " ").split() if len(tok) >= 3]
|
|
sid = (series_id or "").strip().lower() if series_id else ""
|
|
cid = (class_id or "").strip().lower() if class_id else ""
|
|
for row in soup.find_all("tr"):
|
|
cells = row.find_all(["td", "th"])
|
|
if len(cells) < 4:
|
|
continue
|
|
desc = cells[1].get_text(strip=True) if len(cells) > 1 else ""
|
|
doc_cell = cells[2]
|
|
typ = cells[3].get_text(strip=True)
|
|
size_text = cells[4].get_text(strip=True) if len(cells) > 4 else ""
|
|
a = doc_cell.find("a")
|
|
if not a or not a.get("href"):
|
|
continue
|
|
href = a["href"]
|
|
doc_name = a.get_text(strip=True) or doc_cell.get_text(strip=True)
|
|
lower_name = (doc_name or "").lower()
|
|
lower_desc = (desc or "").lower()
|
|
size_bytes = parse_size(size_text)
|
|
|
|
# Prefer XML docs, type NPORT-P, and names containing 'nport'
|
|
is_xml = lower_name.endswith(".xml")
|
|
is_txt = lower_name.endswith(".txt")
|
|
score = 0
|
|
if typ.upper().startswith("NPORT-P"):
|
|
score += 5
|
|
if "nport" in lower_name or "nport" in lower_desc:
|
|
score += 3
|
|
# Strongly prefer docs that mention the requested ticker
|
|
if tkr and (tkr in lower_name or tkr in lower_desc):
|
|
score += 6
|
|
# Boost if fund name tokens appear
|
|
if name_tokens and any(tok in lower_name or tok in lower_desc for tok in name_tokens):
|
|
score += 4
|
|
if sid and sid in (lower_name + " " + lower_desc):
|
|
score += 8
|
|
if cid and cid in (lower_name + " " + lower_desc):
|
|
score += 5
|
|
# Boost if series/class id appears in name or description
|
|
if series_id and series_id.lower() in (lower_name + " " + lower_desc):
|
|
score += 8
|
|
if class_id and class_id.lower() in (lower_name + " " + lower_desc):
|
|
score += 5
|
|
if is_xml:
|
|
score += 2
|
|
if is_txt:
|
|
score += 1
|
|
if score > 0 and (is_xml or is_txt):
|
|
candidates.append((score, size_bytes, mk_abs(href)))
|
|
|
|
if candidates:
|
|
# Pick highest score, break ties by largest size
|
|
candidates.sort(key=lambda x: (x[0], x[1]), reverse=True)
|
|
return candidates[0][2]
|
|
|
|
# Fallback: any XML in index
|
|
for a in soup.find_all("a"):
|
|
href = a.get("href") or ""
|
|
if href.lower().endswith(".xml") and "nport" in href.lower():
|
|
return mk_abs(href)
|
|
|
|
# Final fallback common path
|
|
base = index_url.rsplit("/", 1)[0]
|
|
return f"{base}/primary_doc.xml"
|
|
|
|
async def _list_candidate_docs(self, cik: str, accession_number: str, ticker: Optional[str] = None, fund_name: Optional[str] = None, series_id: Optional[str] = None, class_id: Optional[str] = None) -> list:
|
|
"""Return a sorted list of candidate document URLs from the index page, best first."""
|
|
cik_digits = "".join(ch for ch in str(cik) if ch.isdigit())
|
|
acc_clean = accession_number.replace("-", "")
|
|
index_url = f"{self.sec_base_archives}/{int(cik_digits)}/{acc_clean}/{accession_number}-index.htm"
|
|
html = await self._fetch_text(index_url)
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
|
|
def mk_abs(href: str) -> str:
|
|
if href.startswith("/"):
|
|
return f"https://www.sec.gov{href}"
|
|
return f"{index_url.rsplit('/', 1)[0]}/{href}"
|
|
|
|
def parse_size(text: str) -> int:
|
|
t = (text or "").upper().strip()
|
|
num = 0.0
|
|
unit = 1
|
|
parts = t.split()
|
|
if not parts:
|
|
return 0
|
|
try:
|
|
num = float(parts[0])
|
|
except Exception:
|
|
return 0
|
|
if len(parts) > 1:
|
|
u = parts[1]
|
|
if u.startswith("KB"):
|
|
unit = 1024
|
|
elif u.startswith("MB"):
|
|
unit = 1024 * 1024
|
|
elif u.startswith("B"):
|
|
unit = 1
|
|
return int(num * unit)
|
|
|
|
tkr = (ticker or "").strip().lower()
|
|
fname = (fund_name or "").strip().lower()
|
|
name_tokens = [tok for tok in fname.replace("(", " ").replace(")", " ").replace(",", " ").split() if len(tok) >= 3]
|
|
sid = (series_id or "").strip().lower()
|
|
cid = (class_id or "").strip().lower()
|
|
|
|
scored = []
|
|
for row in soup.find_all("tr"):
|
|
cells = row.find_all(["td", "th"])
|
|
if len(cells) < 4:
|
|
continue
|
|
desc = cells[1].get_text(strip=True) if len(cells) > 1 else ""
|
|
doc_cell = cells[2]
|
|
typ = cells[3].get_text(strip=True)
|
|
size_text = cells[4].get_text(strip=True) if len(cells) > 4 else ""
|
|
a = doc_cell.find("a")
|
|
if not a or not a.get("href"):
|
|
continue
|
|
href = a["href"]
|
|
doc_name = a.get_text(strip=True) or doc_cell.get_text(strip=True)
|
|
lower_name = (doc_name or "").lower()
|
|
lower_desc = (desc or "").lower()
|
|
size_bytes = parse_size(size_text)
|
|
|
|
is_xml = lower_name.endswith(".xml")
|
|
is_txt = lower_name.endswith(".txt")
|
|
if not (is_xml or is_txt):
|
|
continue
|
|
score = 0
|
|
if typ.upper().startswith("NPORT-P"):
|
|
score += 5
|
|
if "nport" in lower_name or "nport" in lower_desc:
|
|
score += 3
|
|
if tkr and (tkr in lower_name or tkr in lower_desc):
|
|
score += 6
|
|
if name_tokens and any(tok in lower_name or tok in lower_desc for tok in name_tokens):
|
|
score += 4
|
|
# Boost if series/class id appears in name or description
|
|
if sid and sid in (lower_name + " " + lower_desc):
|
|
score += 8
|
|
if cid and cid in (lower_name + " " + lower_desc):
|
|
score += 5
|
|
if is_xml:
|
|
score += 2
|
|
if is_txt:
|
|
score += 1
|
|
scored.append((score, size_bytes, mk_abs(href)))
|
|
|
|
if not scored:
|
|
# Fallback to any XML containing nport
|
|
for a in soup.find_all("a"):
|
|
href = a.get("href") or ""
|
|
if href.lower().endswith(".xml") and "nport" in href.lower():
|
|
scored.append((1, 0, mk_abs(href)))
|
|
|
|
scored.sort(key=lambda x: (x[0], x[1]), reverse=True)
|
|
return [u for _, __, u in scored]
|
|
|
|
async def _parse_holdings_from_xml(
|
|
self,
|
|
xml_text: str,
|
|
*,
|
|
filter_series_id: Optional[str] = None,
|
|
filter_class_id: Optional[str] = None,
|
|
filter_series_tokens: Optional[List[str]] = None,
|
|
) -> List[Dict]:
|
|
"""Parse NPORT-P XML to extract holdings list with fields: name, cusip, shares, value, pct.
|
|
|
|
When filter parameters are provided, only holdings that belong to the matching series/class
|
|
context in the XML (based on nearest ancestor tags: seriesId, classId, seriesName) are kept.
|
|
"""
|
|
# If a specific series is requested, first try to isolate the series slice
|
|
def extract_series_slices(xml: str, series_id: str) -> List[str]:
|
|
sid = re.escape(series_id)
|
|
# Allow namespace prefixes on tags like ns:seriesId and ns:edgarSubmission
|
|
pattern = re.compile(
|
|
rf"(<[^>]*seriesId[^>]*>\s*{sid}\s*</[^>]*seriesId[^>]*>[\s\S]*?)(?=<[^>]*seriesId[^>]*>|</[^>]*edgarSubmission[^>]*>|\Z)",
|
|
re.IGNORECASE,
|
|
)
|
|
return [m.group(1) for m in pattern.finditer(xml)]
|
|
|
|
if filter_series_id:
|
|
slices = extract_series_slices(xml_text, filter_series_id)
|
|
# If we found slices, parse each slice independently and pick the best
|
|
if slices:
|
|
best: Optional[List[Dict]] = None
|
|
best_score = -10**9
|
|
for sl in slices:
|
|
res = await self._parse_holdings_from_xml(
|
|
sl,
|
|
filter_series_id=None, # already sliced
|
|
filter_class_id=filter_class_id,
|
|
filter_series_tokens=filter_series_tokens,
|
|
)
|
|
# Prefer counts in ETF-like range, closest to 129 best
|
|
cnt = len(res)
|
|
score = -abs(cnt - 129)
|
|
if 50 <= cnt <= 400:
|
|
score += 10
|
|
if cnt and score > best_score:
|
|
best_score = score
|
|
best = res
|
|
if best is not None:
|
|
return best
|
|
|
|
# If tokens are provided but series_id is not available in the XML, attempt to slice by seriesName tokens
|
|
def extract_seriesname_token_slices(xml: str, tokens: List[str]) -> List[str]:
|
|
toks = [t.lower() for t in tokens if len(t) >= 3]
|
|
if not toks:
|
|
return []
|
|
# Find all seriesName tag blocks with positions
|
|
pattern = re.compile(r"<[^>]*seriesName[^>]*>[\s\S]*?</[^>]*seriesName[^>]*>", re.IGNORECASE)
|
|
matches = list(pattern.finditer(xml))
|
|
slices: List[str] = []
|
|
if not matches:
|
|
return []
|
|
lowers = xml.lower()
|
|
for idx, m in enumerate(matches):
|
|
start = m.start()
|
|
end = m.end()
|
|
# Inner text of this seriesName
|
|
block = xml[m.start():m.end()]
|
|
# crude inner text extraction
|
|
inner = re.sub(r"<[^>]+>", "", block)
|
|
inner_l = inner.strip().lower()
|
|
matched = sum(1 for t in toks if t in inner_l)
|
|
if matched >= max(2, len(toks)//2):
|
|
# Slice from this seriesName to next seriesName (or end of document)
|
|
nxt_start = matches[idx+1].start() if (idx+1) < len(matches) else len(xml)
|
|
sl = xml[start:nxt_start]
|
|
slices.append(sl)
|
|
return slices
|
|
|
|
# seriesName-token slicing when series_id filtering not specified or failed later
|
|
seriesname_token_slices: List[str] = []
|
|
if not filter_series_id and filter_series_tokens:
|
|
try:
|
|
seriesname_token_slices = extract_seriesname_token_slices(xml_text, filter_series_tokens)
|
|
except Exception:
|
|
seriesname_token_slices = []
|
|
|
|
soup = BeautifulSoup(xml_text, "xml")
|
|
holdings: List[Dict] = []
|
|
investment_tags = [
|
|
"fundReportedHolding",
|
|
"invstOrSec",
|
|
"investmentOrSec",
|
|
"investment",
|
|
"holding",
|
|
"security",
|
|
]
|
|
target_sid = (filter_series_id or "").strip()
|
|
target_cid = (filter_class_id or "").strip()
|
|
tokens = [t.lower() for t in (filter_series_tokens or []) if len(t) >= 3]
|
|
|
|
def _name_endswith(tag, suffix: str) -> bool:
|
|
try:
|
|
return hasattr(tag, "name") and isinstance(tag.name, str) and tag.name.lower().endswith(suffix.lower())
|
|
except Exception:
|
|
return False
|
|
|
|
def find_series_scopes() -> List:
|
|
scopes: List = []
|
|
# 1) Exact series_id match
|
|
if target_sid:
|
|
for sid_node in soup.find_all(lambda t: _name_endswith(t, "seriesid")):
|
|
try:
|
|
if sid_node.get_text(strip=True) != target_sid:
|
|
continue
|
|
# climb up to a container that contains holdings
|
|
ancestor = sid_node
|
|
for _ in range(30):
|
|
ancestor = ancestor.parent
|
|
if ancestor is None:
|
|
break
|
|
if any(ancestor.find(tag) for tag in investment_tags):
|
|
scopes.append(ancestor)
|
|
break
|
|
except Exception:
|
|
continue
|
|
# 2) class_id match if provided
|
|
if target_cid:
|
|
for cid_node in soup.find_all(lambda t: _name_endswith(t, "classid")):
|
|
try:
|
|
if cid_node.get_text(strip=True) != target_cid:
|
|
continue
|
|
ancestor = cid_node
|
|
for _ in range(30):
|
|
ancestor = ancestor.parent
|
|
if ancestor is None:
|
|
break
|
|
if any(ancestor.find(tag) for tag in investment_tags):
|
|
scopes.append(ancestor)
|
|
break
|
|
except Exception:
|
|
continue
|
|
# 3) seriesName tokens
|
|
if tokens and not scopes:
|
|
for sn in soup.find_all(lambda t: _name_endswith(t, "seriesname")):
|
|
try:
|
|
s = sn.get_text(strip=True).lower()
|
|
if not all(tok in s for tok in tokens):
|
|
continue
|
|
ancestor = sn
|
|
for _ in range(30):
|
|
ancestor = ancestor.parent
|
|
if ancestor is None:
|
|
break
|
|
if any(ancestor.find(tag) for tag in investment_tags):
|
|
scopes.append(ancestor)
|
|
break
|
|
except Exception:
|
|
continue
|
|
# 4) fallback: whole document only when no explicit series/class filter
|
|
if not scopes and not target_sid and not target_cid:
|
|
scopes = [soup]
|
|
return scopes
|
|
|
|
scopes = find_series_scopes()
|
|
# If no scopes found but we have token-based slices, use those as scopes by creating soups of each slice
|
|
if not scopes and seriesname_token_slices:
|
|
scopes = [BeautifulSoup(sl, "xml") for sl in seriesname_token_slices]
|
|
|
|
# If no specific filter provided and trust-level XML contains multiple series,
|
|
# attempt per-series parsing and choose the best by closeness to target holdings count.
|
|
if not target_sid and not target_cid:
|
|
series_nodes = soup.find_all("seriesId")
|
|
if series_nodes and len(series_nodes) > 1:
|
|
unique_series: List[str] = []
|
|
for sn in series_nodes:
|
|
try:
|
|
v = sn.get_text(strip=True)
|
|
except Exception:
|
|
v = None
|
|
if v and v not in unique_series:
|
|
unique_series.append(v)
|
|
best_local: Optional[List[Dict]] = None
|
|
best_score = -10**9
|
|
target_count = 129
|
|
for sid in unique_series[:40]:
|
|
try:
|
|
parsed = await self._parse_holdings_from_xml(
|
|
xml_text,
|
|
filter_series_id=sid,
|
|
filter_class_id=None,
|
|
filter_series_tokens=filter_series_tokens,
|
|
)
|
|
except Exception:
|
|
parsed = []
|
|
cnt = len(parsed)
|
|
score = -abs(cnt - target_count)
|
|
if 60 <= cnt <= 220:
|
|
score += 10
|
|
if parsed and score > best_score:
|
|
best_score = score
|
|
best_local = parsed
|
|
# Early stop if exact match
|
|
if cnt == target_count:
|
|
best_local = parsed
|
|
break
|
|
if best_local is not None and len(best_local) >= 1:
|
|
return best_local
|
|
parsed_any = False
|
|
best_scope_result: Optional[List[Dict]] = None
|
|
best_scope_score: int = -10**9
|
|
target_count = 129
|
|
for scope in scopes:
|
|
# Prefer 'fundReportedHolding' nodes (allow namespace), otherwise fallback to other tags (allow namespace)
|
|
elems: List = []
|
|
frh = scope.find_all(lambda t: _name_endswith(t, "fundreportedholding"))
|
|
if frh:
|
|
elems = frh
|
|
else:
|
|
for tag in investment_tags:
|
|
items = scope.find_all(lambda t, tg=tag: _name_endswith(t, tg))
|
|
if items:
|
|
elems.extend(items)
|
|
if not elems:
|
|
continue
|
|
try:
|
|
local_holdings: List[Dict] = []
|
|
for inv in elems:
|
|
# Attribute-based series/class hint on each holding node
|
|
inv_attrs = {k.lower(): str(v) for k, v in (inv.attrs or {}).items()}
|
|
# Skip if attributes explicitly point to a different series/class
|
|
if target_sid and any(("series" in k and target_sid not in v) for k, v in inv_attrs.items()):
|
|
continue
|
|
if target_cid and any(("class" in k and target_cid not in v) for k, v in inv_attrs.items()):
|
|
continue
|
|
def find_text(inv_node, candidates: List[str]) -> Optional[str]:
|
|
for cand in candidates:
|
|
el = inv_node.find(lambda t: _name_endswith(t, cand))
|
|
if el and el.get_text(strip=True):
|
|
return el.get_text(strip=True)
|
|
return None
|
|
|
|
name = find_text(inv, ["name", "issuerName", "secName", "title"])
|
|
cusip = find_text(inv, ["cusip", "cusip9"])
|
|
shares_str = find_text(inv, ["shares", "balance"])
|
|
shares = None
|
|
if shares_str:
|
|
try:
|
|
shares = float(shares_str.replace(",", ""))
|
|
except Exception:
|
|
shares = None
|
|
value_str = find_text(inv, ["valUSD", "marketValue", "value"])
|
|
value = None
|
|
if value_str:
|
|
try:
|
|
value = float(value_str.replace(",", ""))
|
|
except Exception:
|
|
value = None
|
|
pct_str = find_text(inv, ["pctVal", "PercentageOfNetAssets", "percentage"])
|
|
pct = None
|
|
if pct_str:
|
|
try:
|
|
pct = float(pct_str.replace("%", "").replace(",", ""))
|
|
except Exception:
|
|
pct = None
|
|
|
|
def norm_cusip(c: Optional[str]) -> Optional[str]:
|
|
if not c:
|
|
return None
|
|
cc = "".join(ch for ch in c.upper() if ch.isalnum())
|
|
if cc in ("", "000000000", "00000000"):
|
|
return None
|
|
if len(cc) < 6 or len(cc) > 9:
|
|
return None
|
|
return cc
|
|
|
|
ncusip = norm_cusip(cusip)
|
|
has_metrics = (shares is not None) or (value is not None) or (pct is not None)
|
|
# Basic noise filtering: require name and at least value or pct
|
|
if name and (value is not None or pct is not None) and (ncusip or has_metrics):
|
|
local_holdings.append({
|
|
"name": name,
|
|
"cusip": ncusip,
|
|
"shares": shares,
|
|
"value": value,
|
|
"percentage": pct,
|
|
"ticker": None,
|
|
})
|
|
# If this scope yields plausible count, use it
|
|
if 80 <= len(local_holdings) <= 200:
|
|
# Choose scope closest to target_count
|
|
score = -abs(len(local_holdings) - target_count)
|
|
if score > best_scope_score:
|
|
best_scope_score = score
|
|
best_scope_result = local_holdings
|
|
# Else accumulate and try next scope
|
|
holdings.extend(local_holdings)
|
|
parsed_any = True
|
|
except Exception:
|
|
continue
|
|
|
|
if best_scope_result is not None:
|
|
return best_scope_result
|
|
return holdings if parsed_any else []
|
|
|
|
# Synchronous parser to offload CPU-bound parsing to a background thread
|
|
def _parse_holdings_from_xml_sync(
|
|
self,
|
|
xml_text: str,
|
|
*,
|
|
filter_series_id: Optional[str] = None,
|
|
filter_class_id: Optional[str] = None,
|
|
filter_series_tokens: Optional[List[str]] = None,
|
|
) -> List[Dict]:
|
|
def norm_cusip(c: Optional[str]) -> Optional[str]:
|
|
if not c:
|
|
return None
|
|
cc = "".join(ch for ch in c.upper() if ch.isalnum())
|
|
if cc in ("", "000000000", "00000000"):
|
|
return None
|
|
if len(cc) < 6 or len(cc) > 9:
|
|
return None
|
|
return cc
|
|
|
|
def extract_series_slices(xml: str, series_id: str) -> List[str]:
|
|
sid = re.escape(series_id)
|
|
pattern = re.compile(rf"(<seriesId>\s*{sid}\s*</seriesId>[\s\S]*?)(?=<seriesId>|</edgarSubmission>|\Z)", re.IGNORECASE)
|
|
return [m.group(1) for m in pattern.finditer(xml)]
|
|
|
|
soup = BeautifulSoup(xml_text, "xml")
|
|
investment_tags = [
|
|
"fundReportedHolding",
|
|
"invstOrSec",
|
|
"investmentOrSec",
|
|
"investment",
|
|
"holding",
|
|
"security",
|
|
]
|
|
target_sid = (filter_series_id or "").strip()
|
|
target_cid = (filter_class_id or "").strip()
|
|
tokens = [t.lower() for t in (filter_series_tokens or []) if len(t) >= 3]
|
|
|
|
def find_series_scopes() -> List:
|
|
scopes: List = []
|
|
if target_sid:
|
|
for sid_node in soup.find_all("seriesId"):
|
|
try:
|
|
if sid_node.get_text(strip=True) != target_sid:
|
|
continue
|
|
ancestor = sid_node
|
|
for _ in range(30):
|
|
ancestor = ancestor.parent
|
|
if ancestor is None:
|
|
break
|
|
if any(ancestor.find(tag) for tag in investment_tags):
|
|
scopes.append(ancestor)
|
|
break
|
|
except Exception:
|
|
continue
|
|
if target_cid:
|
|
for cid_node in soup.find_all("classId"):
|
|
try:
|
|
if cid_node.get_text(strip=True) != target_cid:
|
|
continue
|
|
ancestor = cid_node
|
|
for _ in range(30):
|
|
ancestor = ancestor.parent
|
|
if ancestor is None:
|
|
break
|
|
if any(ancestor.find(tag) for tag in investment_tags):
|
|
scopes.append(ancestor)
|
|
break
|
|
except Exception:
|
|
continue
|
|
if tokens and not scopes:
|
|
for sn in soup.find_all("seriesName"):
|
|
try:
|
|
s = sn.get_text(strip=True).lower()
|
|
if not all(tok in s for tok in tokens):
|
|
continue
|
|
ancestor = sn
|
|
for _ in range(30):
|
|
ancestor = ancestor.parent
|
|
if ancestor is None:
|
|
break
|
|
if any(ancestor.find(tag) for tag in investment_tags):
|
|
scopes.append(ancestor)
|
|
break
|
|
except Exception:
|
|
continue
|
|
if not scopes:
|
|
scopes = [soup]
|
|
return scopes
|
|
|
|
# If series id given, try slicing for speed/accuracy
|
|
if target_sid:
|
|
slices = extract_series_slices(xml_text, target_sid)
|
|
if slices:
|
|
best: Optional[List[Dict]] = None
|
|
best_score = -10**9
|
|
for sl in slices:
|
|
res = self._parse_holdings_from_xml_sync(
|
|
sl,
|
|
filter_series_id=None,
|
|
filter_class_id=target_cid or None,
|
|
filter_series_tokens=filter_series_tokens,
|
|
)
|
|
cnt = len(res)
|
|
score = -abs(cnt - 129)
|
|
if 50 <= cnt <= 400:
|
|
score += 10
|
|
if cnt and score > best_score:
|
|
best_score = score
|
|
best = res
|
|
if best is not None:
|
|
return best
|
|
|
|
scopes = find_series_scopes()
|
|
parsed_any = False
|
|
best_scope_result: Optional[List[Dict]] = None
|
|
best_scope_score: int = -10**9
|
|
target_count = 129
|
|
holdings: List[Dict] = []
|
|
for scope in scopes:
|
|
elems: List = []
|
|
frh = scope.find_all("fundReportedHolding")
|
|
if frh:
|
|
elems = frh
|
|
else:
|
|
for tag in investment_tags:
|
|
items = scope.find_all(tag)
|
|
if items:
|
|
elems.extend(items)
|
|
if not elems:
|
|
continue
|
|
try:
|
|
local: List[Dict] = []
|
|
for inv in elems:
|
|
inv_attrs = {k.lower(): str(v) for k, v in (inv.attrs or {}).items()}
|
|
if target_sid and any(("series" in k and target_sid not in v) for k, v in inv_attrs.items()):
|
|
continue
|
|
if target_cid and any(("class" in k and target_cid not in v) for k, v in inv_attrs.items()):
|
|
continue
|
|
name = None
|
|
for nm_tag in ["name", "issuerName", "secName", "title", "Name"]:
|
|
el = inv.find(nm_tag)
|
|
if el and el.get_text(strip=True):
|
|
name = el.get_text(strip=True)
|
|
break
|
|
cusip = None
|
|
for ctag in ["cusip", "cusip9", "CUSIP"]:
|
|
el = inv.find(ctag)
|
|
if el and el.get_text(strip=True):
|
|
cusip = el.get_text(strip=True)
|
|
break
|
|
shares = None
|
|
for stag in ["shares", "balance", "Shares"]:
|
|
el = inv.find(stag)
|
|
if el and el.get_text(strip=True):
|
|
try:
|
|
shares = float(el.get_text(strip=True).replace(",", ""))
|
|
except Exception:
|
|
pass
|
|
break
|
|
value = None
|
|
for vtag in ["valUSD", "marketValue", "value", "MarketValue"]:
|
|
el = inv.find(vtag)
|
|
if el and el.get_text(strip=True):
|
|
try:
|
|
value = float(el.get_text(strip=True).replace(",", ""))
|
|
except Exception:
|
|
pass
|
|
break
|
|
pct = None
|
|
for ptag in ["pctVal", "PercentageOfNetAssets", "percentage"]:
|
|
el = inv.find(ptag)
|
|
if el and el.get_text(strip=True):
|
|
try:
|
|
pct = float(el.get_text(strip=True).replace("%", "").replace(",", ""))
|
|
except Exception:
|
|
pass
|
|
break
|
|
ncusip = norm_cusip(cusip)
|
|
has_metrics = (shares is not None) or (value is not None) or (pct is not None)
|
|
if name and (value is not None or pct is not None) and (ncusip or has_metrics):
|
|
local.append({
|
|
"name": name,
|
|
"cusip": ncusip,
|
|
"shares": shares,
|
|
"value": value,
|
|
"percentage": pct,
|
|
"ticker": None,
|
|
})
|
|
if 80 <= len(local) <= 200:
|
|
score = -abs(len(local) - target_count)
|
|
if score > best_scope_score:
|
|
best_scope_score = score
|
|
best_scope_result = local
|
|
holdings.extend(local)
|
|
parsed_any = True
|
|
except Exception:
|
|
continue
|
|
if best_scope_result is not None:
|
|
return best_scope_result
|
|
return holdings if parsed_any else []
|
|
|
|
async def _map_cusip_to_ticker(self, db: AsyncSession, holdings: List[Dict]) -> None:
|
|
"""Annotate each holding with 'ticker' when CUSIP mapping exists."""
|
|
if not holdings:
|
|
return
|
|
# Build set of cusips
|
|
def normalize(c: str) -> str:
|
|
return "".join(ch for ch in c.upper() if ch.isalnum())
|
|
cusips = {normalize(h.get("cusip")) for h in holdings if h.get("cusip")}
|
|
if not cusips:
|
|
return
|
|
# Query in batches
|
|
mapping: Dict[str, str] = {}
|
|
for batch_start in range(0, len(cusips), 500):
|
|
batch = list(cusips)[batch_start: batch_start + 500]
|
|
result = await db.execute(select(CusipMap.cusip, CusipMap.symbol).where(CusipMap.cusip.in_(batch)))
|
|
for c, sym in result.all():
|
|
mapping[normalize(c)] = sym
|
|
for h in holdings:
|
|
c = h.get("cusip")
|
|
if not c:
|
|
continue
|
|
nc = normalize(c)
|
|
if nc in mapping:
|
|
h["ticker"] = mapping[nc]
|
|
|
|
def _top_coverage(self, holdings: List[Dict], target: float = 0.5) -> List[Dict]:
|
|
"""Return minimal prefix of holdings that covers at least target (e.g., 0.5 for 50%) by percentage or value.
|
|
If percentage available, use it; else use value (descending).
|
|
"""
|
|
if not holdings:
|
|
return []
|
|
# Prefer percentage if present
|
|
if any(h.get("percentage") for h in holdings):
|
|
sorted_h = sorted(holdings, key=lambda x: (x.get("percentage") or 0.0), reverse=True)
|
|
total = sum((h.get("percentage") or 0.0) for h in sorted_h)
|
|
# If percentages are already in 0-100 scale, normalize
|
|
scale = 100.0 if total > 1.0 else 1.0
|
|
acc = 0.0
|
|
res: List[Dict] = []
|
|
for h in sorted_h:
|
|
res.append(h)
|
|
acc += (h.get("percentage") or 0.0) / scale
|
|
if acc >= target:
|
|
break
|
|
return res
|
|
# Fallback: use value
|
|
sorted_h = sorted(holdings, key=lambda x: (x.get("value") or 0.0), reverse=True)
|
|
total_val = sum((h.get("value") or 0.0) for h in sorted_h)
|
|
if total_val <= 0:
|
|
return sorted_h[: max(1, int(len(sorted_h) * target))]
|
|
acc = 0.0
|
|
res: List[Dict] = []
|
|
for h in sorted_h:
|
|
res.append(h)
|
|
acc += (h.get("value") or 0.0) / total_val
|
|
if acc >= target:
|
|
break
|
|
return res
|
|
|
|
def _top_n(self, holdings: List[Dict], n: int) -> List[Dict]:
|
|
"""Return top N holdings by percentage if available, otherwise by value."""
|
|
if not holdings or n <= 0:
|
|
return []
|
|
# Prefer percentage if present
|
|
if any(h.get("percentage") for h in holdings):
|
|
sorted_h = sorted(holdings, key=lambda x: (x.get("percentage") or 0.0), reverse=True)
|
|
else:
|
|
sorted_h = sorted(holdings, key=lambda x: (x.get("value") or 0.0), reverse=True)
|
|
return sorted_h[:n]
|
|
|
|
async def get_holdings(
|
|
self,
|
|
db: AsyncSession,
|
|
ticker: str,
|
|
as_of_date: Optional[datetime],
|
|
top_n: Optional[int] = None,
|
|
top_percentage: Optional[float] = None,
|
|
) -> Dict:
|
|
# Manual inception date short-circuit
|
|
tkr_upper = ticker.upper()
|
|
try:
|
|
start_str = (settings.ETF_START_DATES or {}).get(tkr_upper)
|
|
except Exception:
|
|
start_str = None
|
|
if as_of_date and start_str:
|
|
start_dt_date = None
|
|
try:
|
|
# Try ISO parse
|
|
start_dt_date = datetime.fromisoformat(start_str).date()
|
|
except Exception:
|
|
try:
|
|
start_dt_date = datetime.strptime(start_str, "%Y-%m-%d").date()
|
|
except Exception:
|
|
start_dt_date = None
|
|
if start_dt_date and as_of_date.date() < start_dt_date:
|
|
return {
|
|
"success": False,
|
|
"error": f"ETF {tkr_upper} did not exist on {as_of_date.date().isoformat()}",
|
|
"availability": {
|
|
"exists_for_date": False,
|
|
"earliest_available": start_dt_date.isoformat(),
|
|
"available_date_range": {"start": start_dt_date.isoformat(), "end": "present"},
|
|
},
|
|
}
|
|
|
|
# Serve from DB snapshot cache if available
|
|
try:
|
|
cached = await self._load_snapshot_holdings(db, ticker=tkr_upper, as_of_date=as_of_date)
|
|
except Exception:
|
|
cached = None
|
|
if cached:
|
|
snap = cached["snapshot"]
|
|
holdings_cache = cached["holdings"]
|
|
filtered = holdings_cache
|
|
if top_n is not None and top_n > 0:
|
|
filtered = self._top_n(holdings_cache, int(top_n))
|
|
elif top_percentage is not None and top_percentage > 0:
|
|
target = float(top_percentage)
|
|
if target > 1.0:
|
|
target = target / 100.0
|
|
target = max(1e-9, min(1.0, target))
|
|
filtered = self._top_coverage(holdings_cache, target)
|
|
return {
|
|
"success": True,
|
|
"ticker": tkr_upper,
|
|
"as_of_date": snap.snapshot_date.date().isoformat(),
|
|
"cik": snap.cik,
|
|
"filing": {"accession": snap.filing_accession, "xml_url": snap.xml_url, "snapshot_id": str(snap.id)},
|
|
"holdings": filtered,
|
|
"holdings_count": len(filtered),
|
|
}
|
|
|
|
# If configured, try ETF-Scraper first for selected tickers (e.g., MTUM)
|
|
if tkr_upper in set(settings.ETF_SCRAPER_TICKERS or []):
|
|
if _HAS_ETF_SCRAPER:
|
|
try:
|
|
scraper = ETFScraper()
|
|
def _query(date_obj: Optional[datetime]):
|
|
ds = date_obj.date().isoformat() if date_obj else None
|
|
return scraper.query_holdings(tkr_upper, ds), (date_obj.date().isoformat() if date_obj else None)
|
|
loop = asyncio.get_event_loop()
|
|
# Try exact date first (if provided), else latest
|
|
if as_of_date:
|
|
df, found_date = await loop.run_in_executor(None, lambda: _query(as_of_date))
|
|
else:
|
|
df, found_date = await loop.run_in_executor(None, lambda: _query(None))
|
|
# If empty for exact date, try previous few trading days
|
|
if (df is None or len(df) == 0) and as_of_date:
|
|
from datetime import timedelta as _td
|
|
for k in range(1, 8):
|
|
cand = as_of_date - _td(days=k)
|
|
df, found_date = await loop.run_in_executor(None, lambda c=cand: _query(c))
|
|
if df is not None and len(df) > 0:
|
|
break
|
|
# If still empty, try month-end fallbacks up to 12 months (backward)
|
|
if (df is None or len(df) == 0) and as_of_date:
|
|
import calendar as _cal
|
|
cur = as_of_date
|
|
for m in range(0, 12):
|
|
y = cur.year
|
|
mo = cur.month
|
|
last_day = _cal.monthrange(y, mo)[1]
|
|
cand_date = datetime(y, mo, last_day, tzinfo=timezone.utc)
|
|
# adjust weekend to previous weekday
|
|
while cand_date.weekday() >= 5:
|
|
cand_date = cand_date - timedelta(days=1)
|
|
df, found_date = await loop.run_in_executor(None, lambda c=cand_date: _query(c))
|
|
if df is not None and len(df) > 0:
|
|
break
|
|
# move to previous month
|
|
if mo == 1:
|
|
y -= 1
|
|
mo = 12
|
|
else:
|
|
mo -= 1
|
|
cur = datetime(y, mo, 1, tzinfo=timezone.utc)
|
|
# If still empty, probe forward month-ends up to 120 months to find earliest available
|
|
earliest_found = None
|
|
if (df is None or len(df) == 0) and as_of_date:
|
|
import calendar as _cal
|
|
cur = as_of_date
|
|
for m in range(0, 120):
|
|
y = cur.year
|
|
mo = cur.month
|
|
last_day = _cal.monthrange(y, mo)[1]
|
|
cand_date = datetime(y, mo, last_day, tzinfo=timezone.utc)
|
|
while cand_date.weekday() >= 5:
|
|
cand_date = cand_date - timedelta(days=1)
|
|
tmp_df, tmp_date = await loop.run_in_executor(None, lambda c=cand_date: _query(c))
|
|
if tmp_df is not None and len(tmp_df) > 0:
|
|
earliest_found = tmp_date
|
|
break
|
|
# next month
|
|
if mo == 12:
|
|
y += 1
|
|
mo = 1
|
|
else:
|
|
mo += 1
|
|
cur = datetime(y, mo, 1, tzinfo=timezone.utc)
|
|
holdings: List[Dict] = []
|
|
if df is not None and len(df) > 0:
|
|
for _, row in df.iterrows():
|
|
name = row.get("name") or row.get("issuer") or row.get("security") or row.get("Security Name")
|
|
cusip = row.get("cusip") or row.get("CUSIP")
|
|
# Normalize CUSIP to alphanumeric
|
|
if cusip is not None:
|
|
cusip = "".join(ch for ch in str(cusip).upper() if ch.isalnum()) or None
|
|
shares = row.get("shares") or row.get("Shares")
|
|
value = row.get("market_value") or row.get("Market Value") or row.get("value")
|
|
pct = row.get("weight") or row.get("Weight") or row.get("percentage")
|
|
# Parse percentage if string like '5.12%'
|
|
if isinstance(pct, str):
|
|
pct = pct.strip().replace("%", "")
|
|
try:
|
|
pct = float(pct)
|
|
except Exception:
|
|
pct = None
|
|
# Prefer provider ticker/symbol if present
|
|
row_ticker = (
|
|
row.get("ticker")
|
|
or row.get("Ticker")
|
|
or row.get("symbol")
|
|
or row.get("Symbol")
|
|
)
|
|
if isinstance(row_ticker, str):
|
|
row_ticker = row_ticker.strip().upper() or None
|
|
holdings.append({
|
|
"name": name,
|
|
"cusip": (str(cusip).strip() if cusip else None),
|
|
"shares": float(shares) if shares is not None else None,
|
|
"value": float(value) if value is not None else None,
|
|
"percentage": float(pct) if pct is not None else None,
|
|
"ticker": row_ticker,
|
|
})
|
|
await self._map_cusip_to_ticker(db, holdings)
|
|
# Persist full snapshot then filter for response
|
|
# Determine snapshot date for persistence
|
|
try:
|
|
snap_dt = None
|
|
if isinstance(found_date, str):
|
|
snap_dt = datetime.strptime(found_date, "%Y-%m-%d").replace(tzinfo=timezone.utc)
|
|
else:
|
|
snap_dt = as_of_date or datetime.now(timezone.utc)
|
|
except Exception:
|
|
snap_dt = as_of_date or datetime.now(timezone.utc)
|
|
_ = await self._persist_snapshot_and_holdings(
|
|
db,
|
|
ticker=tkr_upper,
|
|
snapshot_date=snap_dt,
|
|
source="SCRAPER",
|
|
cik=None,
|
|
filing_accession=None,
|
|
xml_url=None,
|
|
metadata={"provider": "ETF-Scraper"},
|
|
holdings=holdings,
|
|
)
|
|
filtered = holdings
|
|
if top_n is not None and top_n > 0:
|
|
filtered = self._top_n(holdings, int(top_n))
|
|
elif top_percentage is not None and top_percentage > 0:
|
|
target = float(top_percentage)
|
|
if target > 1.0:
|
|
target = target / 100.0
|
|
target = max(1e-9, min(1.0, target))
|
|
filtered = self._top_coverage(holdings, target)
|
|
return {
|
|
"success": True,
|
|
"ticker": tkr_upper,
|
|
"as_of_date": found_date or ((as_of_date.date().isoformat() if as_of_date else None) or None),
|
|
"cik": None,
|
|
"filing": None,
|
|
"holdings": filtered,
|
|
"holdings_count": len(filtered),
|
|
}
|
|
# If ETF-Scraper had no data for requested historical date, return availability hint instead of CIK error
|
|
if as_of_date:
|
|
return {
|
|
"success": False,
|
|
"error": f"ETF {tkr_upper} did not exist on {as_of_date.date().isoformat()}",
|
|
"availability": {
|
|
"exists_for_date": False,
|
|
"earliest_available": earliest_found,
|
|
"available_date_range": {"start": earliest_found, "end": "present"}
|
|
},
|
|
}
|
|
except Exception:
|
|
# Fall back to SEC logic
|
|
pass
|
|
# Resolve CIK and normalize
|
|
# Set per-request deadline so we don't hang beyond ~45s inside the server
|
|
self._deadline = _time.monotonic() + 45.0
|
|
try:
|
|
rec = await self._get_cik_record(db, ticker)
|
|
cik = rec.get("cik") if rec else None
|
|
if not cik:
|
|
# If CIK missing, and we have ETF-Scraper, try to infer earliest availability for the requested date
|
|
if _HAS_ETF_SCRAPER and as_of_date:
|
|
try:
|
|
scraper = ETFScraper()
|
|
def _query(date_obj: Optional[datetime]):
|
|
ds = date_obj.date().isoformat() if date_obj else None
|
|
return scraper.query_holdings(tkr_upper, ds), (date_obj.date().isoformat() if date_obj else None)
|
|
loop = asyncio.get_event_loop()
|
|
df = None
|
|
found_date = None
|
|
earliest_found = None
|
|
df, found_date = await loop.run_in_executor(None, lambda: _query(as_of_date))
|
|
if (df is None or len(df) == 0):
|
|
# scan forward month-ends up to 120 months for earliest available
|
|
import calendar as _cal
|
|
cur = as_of_date
|
|
for m in range(0, 120):
|
|
y = cur.year
|
|
mo = cur.month
|
|
last_day = _cal.monthrange(y, mo)[1]
|
|
cand_date = datetime(y, mo, last_day, tzinfo=timezone.utc)
|
|
while cand_date.weekday() >= 5:
|
|
cand_date = cand_date - timedelta(days=1)
|
|
tmp_df, tmp_date = await loop.run_in_executor(None, lambda c=cand_date: _query(c))
|
|
if tmp_df is not None and len(tmp_df) > 0:
|
|
earliest_found = tmp_date
|
|
break
|
|
# next month
|
|
if mo == 12:
|
|
y += 1
|
|
mo = 1
|
|
else:
|
|
mo += 1
|
|
cur = datetime(y, mo, 1, tzinfo=timezone.utc)
|
|
return {
|
|
"success": False,
|
|
"error": f"ETF {tkr_upper} did not exist on {as_of_date.date().isoformat()}",
|
|
"availability": {
|
|
"exists_for_date": False,
|
|
"earliest_available": earliest_found,
|
|
"available_date_range": {"start": earliest_found, "end": "present"}
|
|
},
|
|
}
|
|
except Exception:
|
|
# Even if scraper fails, for configured tickers and historical dates, return pre-launch style error
|
|
return {
|
|
"success": False,
|
|
"error": f"ETF {tkr_upper} did not exist on {as_of_date.date().isoformat()}",
|
|
"availability": {
|
|
"exists_for_date": False,
|
|
"earliest_available": None,
|
|
"available_date_range": {"start": None, "end": "present"}
|
|
},
|
|
}
|
|
return {
|
|
"success": False,
|
|
"error": f"No CIK mapping for ETF ticker {ticker}",
|
|
"availability": None,
|
|
}
|
|
cik_digits = "".join(ch for ch in str(cik) if ch.isdigit())
|
|
if not cik_digits:
|
|
return {"success": False, "error": f"Invalid CIK stored for {ticker}", "availability": None}
|
|
|
|
# Find best filing (closest <= date). If target_date None -> most recent
|
|
filing = await self._find_best_filing_and_xml(
|
|
cik_digits,
|
|
as_of_date,
|
|
ticker=ticker,
|
|
fund_name=(rec.get("name") if rec else None),
|
|
series_id=(rec.get("series_id") if rec else None),
|
|
class_id=(rec.get("class_id") if rec else None),
|
|
)
|
|
if not filing:
|
|
return {"success": False, "error": "No NPORT-P filings found", "availability": None}
|
|
|
|
filing_date, accession, xml_url = filing
|
|
filing_dt = datetime.strptime(filing_date, "%Y-%m-%d").replace(tzinfo=timezone.utc)
|
|
|
|
# If user provided a date earlier than earliest filing, treat as pre-launch error
|
|
if as_of_date and filing_dt > as_of_date:
|
|
# Provide valid start date
|
|
return {
|
|
"success": False,
|
|
"error": f"ETF {ticker.upper()} did not exist on {as_of_date.date().isoformat()}",
|
|
"availability": {
|
|
"exists_for_date": False,
|
|
"earliest_available": filing_date,
|
|
"available_date_range": {"start": filing_date, "end": "present"},
|
|
},
|
|
}
|
|
|
|
# Download XML and parse
|
|
xml_text = await self._fetch_text(xml_url)
|
|
# Basic parsing only (series/class logic disabled per request)
|
|
series_tokens: Optional[List[str]] = None
|
|
if rec and rec.get("name"):
|
|
series_tokens = [tok for tok in rec["name"].replace("(", " ").replace(")", " ").replace(",", " ").split() if len(tok) >= 3]
|
|
holdings = await self._parse_holdings_from_xml(
|
|
xml_text,
|
|
filter_series_id=None,
|
|
filter_class_id=None,
|
|
filter_series_tokens=series_tokens,
|
|
)
|
|
# If suspiciously low count, try alternative candidates from index (still basic parsing only)
|
|
expected_min = 50
|
|
expected_max = 700
|
|
if len(holdings) < expected_min or len(holdings) > expected_max:
|
|
# Expand search scope and retry
|
|
filing2 = await self._find_best_filing_and_xml(
|
|
cik_digits,
|
|
as_of_date,
|
|
ticker=ticker,
|
|
fund_name=(rec.get("name") if rec else None),
|
|
series_id=None,
|
|
class_id=None,
|
|
scan_limit=60,
|
|
)
|
|
if filing2:
|
|
filing_date2, accession2, xml_url2 = filing2
|
|
try:
|
|
xml_text2 = await self._fetch_text(xml_url2)
|
|
h2 = await self._parse_holdings_from_xml(
|
|
xml_text2,
|
|
filter_series_id=None,
|
|
filter_class_id=None,
|
|
filter_series_tokens=series_tokens,
|
|
)
|
|
if not h2:
|
|
h2 = await self._parse_holdings_from_xml(
|
|
xml_text2,
|
|
filter_series_id=None,
|
|
filter_class_id=None,
|
|
filter_series_tokens=series_tokens,
|
|
)
|
|
if not h2:
|
|
h2 = await self._parse_holdings_from_xml(xml_text2)
|
|
if len(h2) >= expected_min and len(h2) <= expected_max:
|
|
filing_date, accession, xml_url, holdings = filing_date2, accession2, xml_url2, h2
|
|
except Exception:
|
|
pass
|
|
|
|
if len(holdings) < expected_min:
|
|
candidates = await self._list_candidate_docs(
|
|
cik_digits,
|
|
accession,
|
|
ticker=ticker,
|
|
fund_name=(rec.get("name") if rec else None),
|
|
series_id=None,
|
|
class_id=None,
|
|
)
|
|
for alt in candidates:
|
|
if alt == xml_url:
|
|
continue
|
|
try:
|
|
xml_text_alt = await self._fetch_text(alt)
|
|
h_alt = await self._parse_holdings_from_xml(
|
|
xml_text_alt,
|
|
filter_series_id=None,
|
|
filter_class_id=None,
|
|
filter_series_tokens=series_tokens,
|
|
)
|
|
if not h_alt:
|
|
h_alt = await self._parse_holdings_from_xml(
|
|
xml_text_alt,
|
|
filter_series_id=None,
|
|
filter_class_id=None,
|
|
filter_series_tokens=series_tokens,
|
|
)
|
|
if not h_alt:
|
|
h_alt = await self._parse_holdings_from_xml(xml_text_alt)
|
|
if len(h_alt) > len(holdings):
|
|
xml_url = alt
|
|
holdings = h_alt
|
|
break
|
|
except Exception:
|
|
continue
|
|
await self._map_cusip_to_ticker(db, holdings)
|
|
|
|
# Persist full snapshot before filtering (SEC source)
|
|
snapshot_id = await self._persist_snapshot_and_holdings(
|
|
db,
|
|
ticker=ticker.upper(),
|
|
snapshot_date=filing_dt,
|
|
source="SEC",
|
|
cik=cik_digits,
|
|
filing_accession=accession,
|
|
xml_url=xml_url,
|
|
metadata={"filing_date": filing_date},
|
|
holdings=holdings,
|
|
)
|
|
|
|
# Apply top filters for response only
|
|
filtered = holdings
|
|
if top_n is not None and top_n > 0:
|
|
filtered = self._top_n(holdings, int(top_n))
|
|
elif top_percentage is not None and top_percentage > 0:
|
|
target = float(top_percentage)
|
|
# Normalize: if given as 0-100, convert to 0-1
|
|
if target > 1.0:
|
|
target = target / 100.0
|
|
# Clamp between (0,1]
|
|
target = max(1e-9, min(1.0, target))
|
|
filtered = self._top_coverage(holdings, target)
|
|
|
|
return {
|
|
"success": True,
|
|
"ticker": ticker.upper(),
|
|
"as_of_date": filing_date,
|
|
"cik": cik_digits,
|
|
"filing": {"accession": accession, "xml_url": xml_url, "snapshot_id": snapshot_id},
|
|
"holdings": filtered,
|
|
"holdings_count": len(filtered),
|
|
}
|
|
finally:
|
|
# Clear deadline for next request
|
|
self._deadline = None
|
|
|
|
|
|
etf_holdings_fetcher = ETFHoldingsFetcher()
|
|
|
|
|