""" Point-in-time S&P MidCap 400 / SmallCap 600 research universe: month-end holdings of the iShares Core ETFs that track them (IJH, IJR), and Yahoo daily bars for every ticker that was ever held. research/lab/panel_mid.py builds the Panels. PYTHONPATH=. uv run python -m research.lab.fetch_us_mid holdings # ~45 min; then parses PYTHONPATH=. uv run python -m research.lab.fetch_us_mid prices # ~3 h; resumable PYTHONPATH=. uv run python -m research.lab.fetch_us_mid parse # re-parse cached CSVs PYTHONPATH=. uv run python -m research.lab.fetch_us_mid status # counts, weight sums Holdings. BlackRock's product-page API serves past month-end snapshots (``get-fund-document?...&asOfDate=YYYYMMDD&component=holdings``; only the last trading day of a month works) from 2006-09-29 for both funds, except 2017-01..2017-06, which return nothing. Raw CSVs are cached in data/research/ext/ishares//.csv; months that return nothing are remembered in index.json. Requests go out directly (no proxy), one per ~1.2 s. Old snapshots list each security under its LAST ticker in BlackRock's security master (Westwood One is "DIAL" in 2006), with internal codes (``WM1`` is Williams-Sonoma, ``VRX.1``, ``PSYS.``), no weights (quantity and price only), and a few rows without a ticker. ``_resolve`` maps codes and missing tickers to the clean symbol the company name carries in other snapshots, else strips the suffix; panel_mid's price check catches what is still wrong. Rights, warrants and stub positions (under 2% of an equal weight: shares left over after a deletion) are dropped and weights are recomputed from quantity x price. Outputs: holdings.csv (one row per fund, date, ticker) and snapshots.csv (count, equity value, shares outstanding). Prices. Yahoo daily bars from 2005 (auto_adjust=False so the as-traded close can be checked against the holdings file's price), one ticker per request, cached in data/research/ext/yahoo_mid/.csv; tickers Yahoo does not serve (mostly delisted names) are listed in missing.json. The connection is the shell's default one (Yahoo blocks this host's own IP); the proxy in QUANT_MARKET_PROXY is removed from the environment, because the live monitor fetches through it. Pauses 1.5 s between tickers, 60 s after an error, and 10 minutes after five errors in a row; a run of 25 misses is re-checked against SPY in case Yahoo is throttling. """ from __future__ import annotations import argparse import csv import io import json import os import re import time import urllib.request from datetime import date, datetime, timedelta import numpy as np import pandas as pd ROOT = os.path.join("data", "research", "ext") HOLD_DIR = os.path.join(ROOT, "ishares") PX_DIR = os.path.join(ROOT, "yahoo_mid") TIINGO_DIR = os.path.join(ROOT, "tiingo_mid") HOLDINGS = os.path.join(HOLD_DIR, "holdings.csv") FUNDS = {"IJH": 239763, "IJR": 239774} # S&P MidCap 400, S&P SmallCap 600 ETFS = ["IJH", "IJR", "MDY", "SLY", "IWM", "SPY", "QQQ", "BIL", "SHY"] API = ("https://www.ishares.com/varnish-api/blk-one01-product-data/product-data/api/" "v1/get-fund-document?appType=PRODUCT_PAGE&appSubType=ISHARES&targetSite=us-ishares" "&locale=en_US&portfolioId={pid}&userType=individual&asOfDate={d}&component=holdings") DATES_API = ("https://www.ishares.com/varnish-api/blk-one01-product-data/product-data/api/" "v2/get-product-data?appSubType=ISHARES&appType=PRODUCT_PAGE&component=holdings" "&locale=en_US&portfolioId={pid}&targetSite=us-ishares&userType=individual" "&excludeContent=true&includeConfig=true") UA = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124 Safari/537.36" FIRST_MONTH = "2006-01" NOT_STOCK = r"\bRIGHTS?\b|\bRTS\b|WARRANT|\bWTS?\b|\bCVR\b|ESCROW|CONTRA\b|CASH ACCRUAL|VESTING" STUB_REL = 0.02 # positions below 2% of an equal weight are leftovers CASH_SECTORS = {"S-T Securities", "Cash and/or Derivatives", "Cash", "Money Market"} # --------------------------------------------------------------------------- # holdings # --------------------------------------------------------------------------- _direct = urllib.request.build_opener(urllib.request.ProxyHandler({})) def _get(url: str, headers: dict | None = None) -> bytes: req = urllib.request.Request(url, headers={"User-Agent": UA, **(headers or {})}) for attempt in range(4): try: with _direct.open(req, timeout=60) as fh: return fh.read() except Exception as exc: # noqa: BLE001 — retry transient network errors print(f" {type(exc).__name__}: {exc}; retry {attempt + 1}", flush=True) time.sleep(10 * (attempt + 1)) raise RuntimeError(f"giving up on {url}") def _month_end_candidates(month: pd.Period) -> list[date]: """The last four weekdays of a month, latest first (the last trading day is one of them).""" d = month.end_time.date() out = [] while len(out) < 4: if d.weekday() < 5: out.append(d) d -= timedelta(days=1) return out def _as_of(text: str) -> str | None: m = re.search(r'Fund Holdings as of,"([^"]+)"', text) if not m or m.group(1).strip() == "-": return None return datetime.strptime(m.group(1), "%b %d, %Y").strftime("%Y%m%d") def fetch_holdings() -> None: os.makedirs(HOLD_DIR, exist_ok=True) idx_path = os.path.join(HOLD_DIR, "index.json") index = json.load(open(idx_path)) if os.path.exists(idx_path) else {} today = date.today() for fund, pid in FUNDS.items(): fdir = os.path.join(HOLD_DIR, fund) os.makedirs(fdir, exist_ok=True) seen = index.setdefault(fund, {}) months = pd.period_range(FIRST_MONTH, pd.Period(today, "M") - 1, freq="M") # recent dates the product page lists (the latest may be mid-month) meta = json.loads(_get(DATES_API.format(pid=pid), {"Accept": "application/json", "x-application-id": "pp-ui-csr"})) listed = meta["componentsByNameMap"]["holdings"]["containersByNameMap"]["all"][ "dataPointsByNameMap"]["dateList"]["value"] time.sleep(1.2) todo = [(str(m), _month_end_candidates(m)) for m in months] todo += [(f"listed-{d}", [datetime.strptime(str(d), "%Y%m%d").date()]) for d in listed] for key, cands in todo: if key in seen: continue got = None for d in cands: if d >= today: continue ds = d.strftime("%Y%m%d") path = os.path.join(fdir, f"{ds}.csv") if os.path.exists(path): got = ds break text = _get(API.format(pid=pid, d=ds)).decode("utf-8-sig", "replace") time.sleep(1.2) asof = _as_of(text) if asof is None: continue if asof != ds: print(f" {fund} {ds}: file says {asof}", flush=True) with open(os.path.join(fdir, f"{asof}.csv"), "w") as fh: fh.write(text) got = asof break seen[key] = got print(f"{fund} {key}: {got or 'none'}", flush=True) json.dump(index, open(idx_path, "w"), indent=1, sort_keys=True) json.dump(index, open(idx_path, "w"), indent=1, sort_keys=True) parse_all() def _num(x: str) -> float: x = (x or "").replace(",", "").strip() try: return float(x) except ValueError: return np.nan def parse_snapshot(path: str) -> pd.DataFrame: """All rows of one holdings CSV, numbers parsed.""" rows = list(csv.reader(io.open(path, encoding="utf-8-sig"))) h = next(i for i, r in enumerate(rows) if r and r[0] == "Ticker") shares_out = next((_num(r[1]) for r in rows[:h] if len(r) > 1 and r[0] == "Shares Outstanding"), np.nan) head = rows[h] body = [r for r in rows[h + 1:] if len(r) == len(head)] df = pd.DataFrame(body, columns=head) wcol = next((c for c in ("Weight (%)", "Market Weight") if c in df.columns), None) return pd.DataFrame({ "shares_out": shares_out, "ticker_raw": df["Ticker"].str.strip(), "name": df["Name"].str.strip().str.upper(), "sector": df["Sector"].str.strip(), "asset_class": df["Asset Class"].str.strip(), "type": df["Type"].str.strip() if "Type" in df else "-", "quantity": df["Quantity"].map(_num) if "Quantity" in df else np.nan, "price": df["Price"].map(_num) if "Price" in df else np.nan, "market_value": df["Market Value"].map(_num) if "Market Value" in df else np.nan, "weight_reported": df[wcol].map(_num) if wcol else np.nan, }) def _is_equity(df: pd.DataFrame) -> pd.Series: """Stocks only: drops cash, money-market funds, futures, swaps and collateral. A few old rows have no asset class (and no ticker) but a GICS sector and a share price; they are stocks.""" typ_ok = df["type"].isin(["EQUITY", "-", ""]) listed = df["asset_class"] == "Equity" unclassed = (df["asset_class"].isin(["-", ""]) & ~df["sector"].isin(CASH_SECTORS | {"-", ""}) & (df["price"] > 1.0) & (df["price"] != 100.0)) not_stock = df["name"].str.contains(NOT_STOCK, regex=True, na=False) return typ_ok & (listed | unclassed) & ~not_stock _CLEAN = re.compile(r"[A-Z]{1,5}([ .][A-Z])?") def _yahoo_symbol(raw: str) -> str | None: """BlackRock ticker → Yahoo symbol. Class shares use a dash (BRK.B → BRK-B, "MOG A" → MOG-A); BlackRock's disambiguation suffixes are dropped (VRX.1 → VRX, PSYS. → PSYS, LKQ1 → LKQ, "CC WI" → CC). For suffixed codes this is the last resort: ``_resolve`` first looks the company name up (WM1 is Williams-Sonoma, WSM).""" t = re.sub(r"\s+", " ", raw.strip().upper()) if t in ("", "-") or re.search(r"[ ./](WS|RT|RTS)$", t): # rights and warrants return None t = re.sub(r"[ .](WI|US)$", "", t) # when-issued lines, "EHTH US" m = re.fullmatch(r"([A-Z]+)[ .]([A-Z])", t) if m: return f"{m.group(1)}-{m.group(2)}" t = re.sub(r"\.\d*$", "", t) # VRX.1, PSYS. m = re.fullmatch(r"([A-Z]+)\d+([A-Z]?)", t) # LKQ1, RS6, FR1A if m: t = m.group(1) + (f"-{m.group(2)}" if m.group(2) else "") t = t.replace(".", "-").replace(" ", "-").replace("/", "-") return t or None def _norm_name(n: str) -> str: n = str(n).upper().replace("&", " AND ") n = re.sub(r"\bCL(ASS)?\s+([A-Z])\b", r"CL\2", n) n = re.sub(r"[^A-Z0-9 ]", " ", n) n = re.sub(r"\b(INC|INCORPORATED|CORP|CORPORATION|CO|COMPANY|LTD|LIMITED|PLC|THE|NEW|" r"HOLDINGS?|HLDGS?|GROUP|GRP|INTL|INTERNATIONAL)\b", " ", n) return " ".join(n.split()) def _resolve(rows: pd.DataFrame) -> pd.Series: """Yahoo symbol per row. Rows whose ticker is a clean symbol keep it; suffixed codes and missing tickers take the clean symbol the same company name carries in any snapshot (exact normalized name, else a unique prefix match of at least 10 characters), else the stripped code.""" raw = rows["ticker_raw"].fillna("").str.strip().str.upper() clean = raw.str.fullmatch(_CLEAN) nm = rows["name"].map(_norm_name) sym = raw.map(_yahoo_symbol) by_name = (pd.DataFrame({"nm": nm[clean], "sym": sym[clean]}) .groupby("nm")["sym"].agg(lambda s: s.mode().iloc[0])) names = sorted(by_name.index) out = sym.where(clean) for n in nm[~clean].unique(): hit = by_name.get(n) if hit is None and len(n) >= 10: cand = {by_name[k] for k in names if k.startswith(n) or (len(k) >= 10 and n.startswith(k))} hit = cand.pop() if len(cand) == 1 else None if hit is not None: out[(~clean) & (nm == n)] = hit left = out.isna() out[left] = sym[left] return out def parse_all() -> pd.DataFrame: parts = [] for fund in FUNDS: fdir = os.path.join(HOLD_DIR, fund) for f in sorted(os.listdir(fdir)) if os.path.isdir(fdir) else []: if f.endswith(".csv"): df = parse_snapshot(os.path.join(fdir, f)) df.insert(0, "date", pd.Timestamp(f[:-4])) df.insert(0, "fund", fund) parts.append(df) allrows = pd.concat(parts, ignore_index=True) eq = allrows[_is_equity(allrows)].copy() eq["mv"] = eq["quantity"] * eq["price"] eq["mv"] = eq["market_value"].where(eq["market_value"] > 0, eq["mv"]) eq["weight"] = eq["mv"] / eq.groupby(["fund", "date"])["mv"].transform("sum") eq["ticker"] = _resolve(eq) unresolved = eq[eq["ticker"].isna()] if len(unresolved): print(f"{len(unresolved)} equity rows without a ticker (dropped):", sorted(unresolved["name"].unique())[:20]) eq = eq.dropna(subset=["ticker"]).sort_values("mv", ascending=False) # one row per ticker and snapshot (a security held in two lots is summed; # the price is the largest lot's) eq = (eq.groupby(["fund", "date", "ticker"], as_index=False) .agg(name=("name", "first"), ticker_raw=("ticker_raw", "first"), sector=("sector", "first"), quantity=("quantity", "sum"), price=("price", "first"), mv=("mv", "sum"), weight=("weight", "sum"), weight_reported=("weight_reported", "sum"), shares_out=("shares_out", "first"))) # stub positions: a few hundred shares left after a deletion or a # corporate action, not index members (crashed members stay above 2%) rel = eq["weight"] * eq.groupby(["fund", "date"])["ticker"].transform("size") stub = rel < STUB_REL print(f"dropped {stub.sum()} stub positions (< {STUB_REL:.0%} of an equal weight)") eq = eq[~stub].copy() eq["weight"] = eq["weight"] / eq.groupby(["fund", "date"])["weight"].transform("sum") eq.drop(columns="shares_out").to_csv(HOLDINGS, index=False) summ = eq.groupby(["fund", "date"]).agg(n=("ticker", "size"), equity_mv=("mv", "sum"), weight_reported=("weight_reported", "sum"), shares_out=("shares_out", "first")) summ.to_csv(os.path.join(HOLD_DIR, "snapshots.csv")) print(f"Saved {HOLDINGS}: {len(eq)} rows, {eq['ticker'].nunique()} tickers") return eq def load_holdings() -> pd.DataFrame: return pd.read_csv(HOLDINGS, parse_dates=["date"]) # --------------------------------------------------------------------------- # prices # --------------------------------------------------------------------------- def _px_path(t: str) -> str: return os.path.join(PX_DIR, t.replace("^", "_").replace("=", "_").replace("/", "_") + ".csv") def _no_market_proxy() -> None: """Never fetch Yahoo through QUANT_MARKET_PROXY (the live monitor's IP).""" mp = os.environ.get("QUANT_MARKET_PROXY", "") if not mp and os.path.exists(".env"): for line in open(".env"): if line.startswith("QUANT_MARKET_PROXY="): mp = line.split("=", 1)[1].strip().strip("'\"") for k in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY", "all_proxy", "ALL_PROXY"): if mp and os.environ.get(k, "").rstrip("/") == mp.rstrip("/"): del os.environ[k] def fetch_prices(start: str = "2005-01-01", retry_missing: bool = False) -> None: import yfinance as yf from yfinance.exceptions import YFPricesMissingError, YFTzMissingError yf.config.debug.hide_exceptions = False _no_market_proxy() os.makedirs(PX_DIR, exist_ok=True) miss_path = os.path.join(PX_DIR, "missing.json") missing = json.load(open(miss_path)) if os.path.exists(miss_path) else {} tickers = ETFS + sorted(set(load_holdings()["ticker"]) - set(ETFS)) todo = [t for t in tickers if not os.path.exists(_px_path(t)) and (retry_missing or t not in missing)] print(f"{len(tickers)} tickers, {len(todo)} to fetch", flush=True) errors = 0 miss_streak: list[str] = [] i = 0 while i < len(todo): t = todo[i] try: h = yf.Ticker(t).history(start=start, auto_adjust=False, actions=True) if len(h) == 0: raise YFPricesMissingError(t, "empty") h.index = pd.DatetimeIndex(h.index.tz_localize(None).normalize(), name="Date") h.to_csv(_px_path(t)) missing.pop(t, None) print(f"[{i + 1}/{len(todo)}] {t:8s} ok {len(h)} bars {h.index[0].date()}", flush=True) errors, miss_streak = 0, [] except Exception as exc: # noqa: BLE001 — sorted below msg = f"{type(exc).__name__}: {exc}" gone = (isinstance(exc, (YFTzMissingError, YFPricesMissingError)) or "404" in msg or "Not Found" in msg or "delisted" in msg) if not gone: # rate limit or network error: wait, retry the same ticker errors += 1 print(f"[{i + 1}/{len(todo)}] {t:8s} error {msg[:140]}", flush=True) json.dump(missing, open(miss_path, "w"), indent=0, sort_keys=True) time.sleep(600 if errors >= 5 else 60) if errors >= 5: errors = 0 continue missing[t] = datetime.now().isoformat(timespec="seconds") miss_streak.append(t) print(f"[{i + 1}/{len(todo)}] {t:8s} missing", flush=True) if len(miss_streak) >= 25: # a long run of misses: is Yahoo throttling? try: ok = len(yf.Ticker("SPY").history(period="5d")) > 0 except Exception: # noqa: BLE001 ok = False if not ok: print(" canary SPY failed: throttled; redoing the streak after 10 min", flush=True) for m in miss_streak: missing.pop(m, None) i -= len(miss_streak) - 1 miss_streak = [] time.sleep(600) continue miss_streak = [] errors = 0 if (i + 1) % 25 == 0: json.dump(missing, open(miss_path, "w"), indent=0, sort_keys=True) i += 1 time.sleep(1.5) json.dump(missing, open(miss_path, "w"), indent=0, sort_keys=True) print("done", flush=True) def load_bars(t: str) -> pd.DataFrame | None: """Adjusted OHLCV (as yfinance's auto_adjust=True) plus the unadjusted close. Falls back to a Tiingo-converted CSV (see lab.tiingo_merge).""" path = _px_path(t) if not os.path.exists(path): alt = os.path.join(TIINGO_DIR, os.path.basename(path)) if not os.path.exists(alt): return None path = alt h = pd.read_csv(path, index_col=0, parse_dates=True) h = h[~h.index.duplicated(keep="last")].sort_index() ratio = (h["Adj Close"] / h["Close"]).where(h["Close"] > 0) out = pd.DataFrame({k.lower(): h[k] * ratio for k in ("Open", "High", "Low", "Close")}) out["volume"] = h["Volume"] # Yahoo's Close is split-adjusted: undo later splits for the as-traded price splits = h["Stock Splits"].replace(0, 1.0).fillna(1.0) later = splits[::-1].cumprod()[::-1].shift(-1).fillna(1.0) out["raw_close"] = h["Close"] * later return out def status() -> None: eq = load_holdings() tick = set(eq["ticker"]) have = {t for t in tick if os.path.exists(_px_path(t))} miss_path = os.path.join(PX_DIR, "missing.json") missing = set(json.load(open(miss_path))) if os.path.exists(miss_path) else set() print(f"tickers {len(tick)}: prices {len(have)}, missing {len(missing & tick)}, " f"todo {len(tick - have - missing)}") print(snapshot_summary().groupby("fund") .agg(first=("date", "min"), last=("date", "max"), snaps=("date", "size"), n_min=("n", "min"), n_max=("n", "max"), wsum_min=("weight_sum", "min"), wsum_med=("weight_sum", "median"), wsum_max=("weight_sum", "max"))) def snapshot_summary() -> pd.DataFrame: """Per snapshot: equity holdings count and their weight sum, the equity market value over the fund's assets (the files' own cash rows are unusable: cash is sometimes priced at 100). Assets = shares outstanding x the ETF's close, both on today's split basis (the header's share count is split-adjusted, like Yahoo's unadjusted-for-dividends Close).""" s = pd.read_csv(os.path.join(HOLD_DIR, "snapshots.csv"), parse_dates=["date"]) s["weight_sum"] = np.nan for fund in FUNDS: if not os.path.exists(_px_path(fund)): continue close = pd.read_csv(_px_path(fund), index_col=0, parse_dates=True)["Close"] m = s["fund"] == fund px = close.reindex(close.index.union(s.loc[m, "date"])).ffill().reindex(s.loc[m, "date"]) s.loc[m, "weight_sum"] = s.loc[m, "equity_mv"].values / (s.loc[m, "shares_out"].values * px.values) return s def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("step", choices=["holdings", "parse", "prices", "status"]) ap.add_argument("--retry-missing", action="store_true") args = ap.parse_args() if args.step == "holdings": fetch_holdings() elif args.step == "parse": parse_all() elif args.step == "prices": fetch_prices(retry_missing=args.retry_missing) else: status() if __name__ == "__main__": main()