13F ہولڈنگز پینل اور مشترک ملکیت کے فیچرز بنانا

# 13f_download.py


```py
#!/usr/bin/env python3
"""Download 13F institutional holdings from SEC EDGAR.

Two modes share one CLI and produce the same canonical schema:

  --mode per-cik (default)
      Walk the SEC JSON submissions API for a curated list of large
      institutional investors (Berkshire, Bridgewater, Renaissance, Two
      Sigma, DE Shaw, AQR, Citadel, Millennium, Point72, Tiger Global),
      fetch each 13F-HR filing's XML information table, and assemble a
      multi-quarter holdings panel. Used by Ch22 NB 07 and Ch23 graph /
      RAG notebooks.

  --mode bulk
      Download the SEC's pre-assembled quarterly bulk data set (one 80 MB
      zip with all 13F filings in a 3-month window), parse INFOTABLE +
      COVERPAGE + SUBMISSION, and normalize to the same column schema.
      Used by Ch4 NB 05 to demonstrate the bulk data source at full
      universe scale (~5K filers, ~3M holdings per quarter).

Output layout under `$ML4T_DATA_PATH/equities/positioning/13f/`:

    (per-cik)
      institutional_holdings.parquet    raw holdings: cik, accession_no,
                                        issuer, cusip, value_thousands,
                                        shares, put_call, report_date,
                                        filing_date, company_name
      institution_stock_edges.parquet   institution → stock edge list
      stock_features.parquet            stock-level features
      coownership_matrix.npy            stock × stock similarity
      coownership_stocks.txt            row/col CUSIPs

    (bulk, per quarter)
      bulk/<YYYYQN>/institutional_holdings.parquet   canonical schema,
                                                     ~3M rows
      bulk/<YYYYQN>/bulk_13f.zip                     cached raw zip

Usage:
    # per-cik — default
    python data/equities/positioning/13f_download.py
    python data/equities/positioning/13f_download.py --num-filings 8
    python data/equities/positioning/13f_download.py --max-institutions 3

    # bulk — one or more quarters (filing windows, SEC's own labels)
    python data/equities/positioning/13f_download.py --mode bulk --quarters 2024Q3
    python data/equities/positioning/13f_download.py --mode bulk --quarters 2024Q2,2024Q3

Rate-limited to respect SEC's 10 requests/sec policy.
"""

from __future__ import annotations

import argparse
import calendar
import io
import re
import time
import xml.etree.ElementTree as ET
import zipfile
from collections.abc import Iterable
from datetime import date
from pathlib import Path

import numpy as np
import polars as pl
import requests

from utils.downloading import resolve_data_dir

SEC_HEADERS = {"User-Agent": "ML4T Book stefan@ml4t.io"}
RATE_LIMIT_SECONDS = 0.1
QUARTER_RE = re.compile(r"^(\d{4})Q([1-4])$")

INSTITUTIONS: list[tuple[str, str]] = [
    ("Berkshire Hathaway", "0001067983"),
    ("Bridgewater Associates", "0001350694"),
    ("Renaissance Technologies", "0001037389"),
    ("Two Sigma Investments", "0001450144"),
    ("DE Shaw", "0001009207"),
    ("AQR Capital", "0001167557"),
    ("Citadel Advisors", "0001423053"),
    ("Millennium Management", "0001273087"),
    ("Point72 Asset Management", "0001603466"),
    ("Tiger Global", "0001167483"),
]


def get_recent_13f_filings(cik: str, num_filings: int) -> list[dict]:
    """Fetch recent 13F-HR filing metadata from SEC EDGAR."""
    url = f"https://data.sec.gov/submissions/CIK{cik}.json"
    try:
        resp = requests.get(url, headers=SEC_HEADERS, timeout=10)
        resp.raise_for_status()
        data = resp.json()
    except Exception as e:
        print(f"  Error fetching {cik}: {e}")
        return []

    filings = []
    recent = data.get("filings", {}).get("recent", {})
    for i, form in enumerate(recent.get("form", [])):
        if form == "13F-HR" and len(filings) < num_filings:
            filings.append(
                {
                    "cik": cik,
                    "company_name": data.get("name", "Unknown"),
                    "accession_number": recent["accessionNumber"][i],
                    "report_date": recent["reportDate"][i],
                    "filing_date": recent["filingDate"][i],
                }
            )
    return filings


def fetch_13f_xml_root(cik: str, accession: str) -> ET.Element | None:
    """Fetch and parse the XML information table for one 13F filing."""
    acc_clean = accession.replace("-", "")
    base_url = f"https://www.sec.gov/Archives/edgar/data/{int(cik)}/{acc_clean}/"
    try:
        idx_resp = requests.get(base_url + "index.json", headers=SEC_HEADERS, timeout=10)
        idx_resp.raise_for_status()
        idx_data = idx_resp.json()
    except Exception:
        return None

    xml_files = [
        item["name"]
        for item in idx_data.get("directory", {}).get("item", [])
        if item["name"].endswith(".xml") and item["name"] != "primary_doc.xml"
    ]
    if not xml_files:
        return None

    try:
        resp = requests.get(base_url + xml_files[0], headers=SEC_HEADERS, timeout=30)
        resp.raise_for_status()
        return ET.fromstring(resp.content)
    except Exception:
        return None


def parse_13f_holdings(cik: str, accession: str) -> list[dict]:
    """Parse holdings rows from a 13F information table XML."""
    root = fetch_13f_xml_root(cik, accession)
    if root is None:
        return []

    ns_map = {"ns": "http://www.sec.gov/edgar/document/thirteenf/informationtable"}
    holdings = []
    for info_table in root.findall(".//ns:infoTable", ns_map):
        name = info_table.findtext("ns:nameOfIssuer", "", ns_map)
        cusip = info_table.findtext("ns:cusip", "", ns_map)
        value_str = info_table.findtext("ns:value", "0", ns_map)
        shares_elem = info_table.find("ns:shrsOrPrnAmt/ns:sshPrnamt", ns_map)
        shares_str = shares_elem.text if shares_elem is not None else "0"
        put_call_text = info_table.findtext("ns:putCall", "", ns_map).strip().upper()
        holdings.append(
            {
                "cik": cik,
                "accession_no": accession,
                "issuer": name.strip(),
                "cusip": cusip.strip(),
                "value_thousands": int(value_str) if value_str.isdigit() else 0,
                "shares": int(shares_str) if shares_str.isdigit() else 0,
                "put_call": put_call_text or None,
            }
        )
    return holdings


def _complete_report_dates(holdings_df: pl.DataFrame, expected_ciks: set[str]) -> list[date]:
    """Return report dates every expected institution filed for, newest first.

    13F filings are due 45 days after quarter end, so a download run inside that
    window sees the newest quarter from the early filers only. Building the graph
    from it would drop the late filers and read their absence as a mass exit in
    the quarter-over-quarter ownership change. Both the graph quarter and the
    quarter it is compared against are therefore drawn from this list, so a
    partially filed quarter can never enter either side of the comparison.

    Coverage is measured over every disclosed row, not just long equity: a
    manager who filed but disclosed only options has still filed, and counting
    that as a missing filing would step back a quarter for no reason.
    """
    covered = (
        holdings_df.filter(pl.col("cik").is_in(list(expected_ciks)))
        .group_by("report_date")
        .agg(pl.col("cik").n_unique().alias("n_ciks"))
    )
    complete = covered.filter(pl.col("n_ciks") == len(expected_ciks))
    if complete.is_empty():
        raise ValueError(
            f"No 13F report date is covered by all {len(expected_ciks)} requested "
            "institutions, so no quarter can be built without treating missing "
            "filers as exits. Request more filings per institution with "
            "--num-filings, or narrow the institution list."
        )
    dates = complete["report_date"].sort(descending=True).to_list()
    newest = holdings_df["report_date"].max()
    if dates[0] != newest:
        filed = set(holdings_df.filter(pl.col("report_date") == newest)["cik"].unique().to_list())
        print(
            f"Report date {newest} has only {len(filed)} of {len(expected_ciks)} "
            f"institutions filed; building the graph from {dates[0]} instead. "
            f"Awaiting: {', '.join(sorted(expected_ciks - filed))}"
        )
    return dates


def _reject_pre_2023_reporting_units(equity: pl.DataFrame) -> None:
    """Refuse filings that report value in thousands rather than dollars.

    The SEC switched 13F market value from thousands to whole dollars on
    2023-01-03. The producer keeps the legacy `value_thousands` column name but
    publishes the derived features as `*_usd`, which only holds for filings
    after the switch. Mixing the two would be wrong by a factor of 1,000, and
    silently so.
    """
    earliest = equity["filing_date"].min()
    if earliest is not None and earliest.isoformat() < "2023-01-03":
        raise ValueError(
            f"13F holdings reach back to {earliest}, before the SEC switched "
            "market value from thousands to dollars on 2023-01-03. The derived "
            "artifacts label value in USD and cannot mix the two conventions. "
            "Reduce --num-filings so the window starts after that date."
        )


def build_features_and_matrix(
    holdings_df: pl.DataFrame,
    expected_ciks: Iterable[str] | None = None,
) -> tuple[pl.DataFrame, pl.DataFrame, np.ndarray, list[str]]:
    """Build the latest positive-equity graph and its point-in-time features.

    Args:
        holdings_df: Canonical 13F holdings, including `put_call` and `report_date`.
        expected_ciks: The institutions the caller requested, used to step back
            from a quarter they have not all filed for yet. Without it there is
            no way to tell a manager who has not filed from one who left the
            universe, so the newest quarter present is used as-is.
    """
    if "put_call" not in holdings_df.columns:
        raise ValueError("13F holdings must preserve the SEC putCall field.")
    equity = holdings_df.filter(
        pl.col("put_call").fill_null("").cast(pl.Utf8).str.strip_chars() == ""
    )
    if equity.is_empty():
        raise ValueError("The 13F holdings contain no long-equity positions.")
    _reject_pre_2023_reporting_units(equity)
    complete_dates = (
        None if expected_ciks is None else _complete_report_dates(holdings_df, set(expected_ciks))
    )
    latest_report_date = (
        equity["report_date"].max() if complete_dates is None else complete_dates[0]
    )
    latest_rows = equity.filter(pl.col("report_date") == latest_report_date)
    if latest_rows.is_empty():
        raise ValueError("The latest 13F report date has no positive long-equity positions.")
    # Availability is taken over every disclosure for the selected quarter, not only the
    # long-equity ones the graph is built from. A manager that files later with options
    # only makes the quarter complete - `_complete_report_dates` counts any disclosed row
    # as evidence it filed - without advancing a timestamp read off `latest_rows`, so the
    # graph would claim to have been available before that filing was public. That is
    # lookahead: the quarter was not usable until the last of its filings landed.
    latest_timestamp = holdings_df.filter(pl.col("report_date") == latest_report_date)[
        "filing_date"
    ].max()
    issuer_names = (
        latest_rows.group_by(["cusip", "issuer"])
        .agg(pl.col("value_thousands").sum().alias("issuer_value"))
        .sort(
            ["cusip", "issuer_value", "issuer"],
            descending=[False, True, False],
        )
        .unique(subset="cusip", keep="first", maintain_order=True)
        .select("cusip", pl.col("issuer").alias("stock_name"))
    )
    latest = (
        latest_rows.group_by(["cik", "cusip"])
        .agg(
            pl.col("company_name").sort().first().alias("institution_name"),
            pl.col("value_thousands").cast(pl.Float64).sum().alias("reported_value_usd"),
            pl.col("shares").sum().alias("shares"),
        )
        .join(issuer_names, on="cusip", how="left")
        .filter(pl.col("reported_value_usd") > 0)
        .with_columns(
            pl.lit(latest_report_date).alias("report_date"),
            pl.lit(latest_timestamp).alias("timestamp"),
        )
        .sort(["cik", "cusip"])
    )

    edge_list = latest.select(
        pl.col("cik").alias("institution_id"),
        pl.col("cusip").alias("stock_id"),
        "institution_name",
        "stock_name",
        pl.col("reported_value_usd").alias("weight_value"),
        pl.col("shares").alias("weight_shares"),
        "report_date",
        "timestamp",
    ).sort(["institution_id", "stock_id"])

    institution_count = latest["cik"].n_unique()
    stock_features = (
        latest.group_by("cusip")
        .agg(
            pl.col("stock_name").first().alias("issuer_name"),
            pl.col("cik").n_unique().alias("n_inst_holders"),
            pl.col("reported_value_usd").sum().alias("total_inst_value_usd"),
            pl.col("reported_value_usd").mean().alias("avg_position_size_usd"),
            pl.col("reported_value_usd").std().fill_null(0).alias("position_size_std_usd"),
            pl.col("timestamp").max().alias("timestamp"),
            (pl.col("reported_value_usd") / pl.col("reported_value_usd").sum())
            .pow(2)
            .sum()
            .alias("ownership_hhi"),
        )
        .with_columns(
            (pl.col("n_inst_holders") / institution_count).alias("inst_coverage_pct"),
            (
                pl.col("position_size_std_usd")
                / pl.col("avg_position_size_usd").clip(lower_bound=1)
            ).alias("position_cv"),
        )
        .sort("cusip")
    )

    position_panel = (
        equity.group_by(["cik", "cusip", "report_date"])
        .agg(pl.col("value_thousands").cast(pl.Float64).sum().alias("reported_value_usd"))
        .filter(pl.col("reported_value_usd") > 0)
    )
    # Compare the graph's quarter against another quarter every institution filed
    # for. A partially filed quarter on either side would read as mass entries or
    # exits rather than as real ownership change.
    eligible = (
        position_panel["report_date"].unique().to_list()
        if complete_dates is None
        else complete_dates
    )
    prior_periods = sorted((d for d in eligible if d < latest_report_date), reverse=True)
    if prior_periods:
        current_period, prior_period = latest_report_date, prior_periods[0]
        stock_quarter = position_panel.group_by(["cusip", "report_date"]).agg(
            pl.col("reported_value_usd").sum().alias("quarter_value_usd")
        )
        prior = stock_quarter.filter(pl.col("report_date") == prior_period).select(
            "cusip", pl.col("quarter_value_usd").alias("prior_value_usd")
        )
        current = stock_quarter.filter(pl.col("report_date") == current_period).select(
            "cusip", pl.col("quarter_value_usd").alias("current_value_usd")
        )
        changes = (
            prior.join(current, on="cusip", how="full", coalesce=True)
            .with_columns(
                pl.col("prior_value_usd").fill_null(0),
                pl.col("current_value_usd").fill_null(0),
            )
            .with_columns(
                (pl.col("current_value_usd") - pl.col("prior_value_usd")).alias(
                    "inst_value_change_usd"
                ),
                pl.when(pl.col("prior_value_usd") > 0)
                .then(
                    (pl.col("current_value_usd") - pl.col("prior_value_usd"))
                    / pl.col("prior_value_usd")
                )
                .otherwise(None)
                .alias("inst_pct_change"),
            )
            .select("cusip", "inst_value_change_usd", "inst_pct_change")
        )
        stock_features = stock_features.join(changes, on="cusip", how="left").with_columns(
            pl.col("inst_value_change_usd").fill_null(0)
        )
    else:
        # Keep the artifact schema fixed. With a single quarter there is nothing
        # to compare against, so the change is zero dollars and an undefined rate
        # rather than a missing column that consumers would have to test for.
        stock_features = stock_features.with_columns(
            pl.lit(0.0, dtype=pl.Float64).alias("inst_value_change_usd"),
            pl.lit(None, dtype=pl.Float64).alias("inst_pct_change"),
        )
    stock_features = stock_features.sort("cusip")

    # Co-ownership similarity matrix
    stocks = sorted(latest["cusip"].unique().to_list())
    institutions = sorted(latest["cik"].unique().to_list())
    stock_idx = {s: i for i, s in enumerate(stocks)}
    inst_idx = {c: i for i, c in enumerate(institutions)}
    ownership = np.zeros((len(institutions), len(stocks)), dtype=np.float32)
    for row in latest.iter_rows(named=True):
        ownership[inst_idx[row["cik"]], stock_idx[row["cusip"]]] = row["reported_value_usd"]
    row_sums = ownership.sum(axis=1, keepdims=True)
    row_sums[row_sums == 0] = 1
    ownership_norm = ownership / row_sums
    coown = ownership_norm.T @ ownership_norm
    diag = np.sqrt(np.diag(coown))
    diag[diag == 0] = 1
    similarity = coown / np.outer(diag, diag)

    return stock_features, edge_list, similarity, stocks


# --- Bulk mode (SEC quarterly data sets) ---


def _bulk_zip_url(quarter: str) -> str:
    """Map a filing-window label like '2024Q3' to the SEC bulk zip URL.

    SEC labels 13F data sets by filing-date window, not report quarter:
      Q1 = Mar–May, Q2 = Jun–Aug, Q3 = Sep–Nov, Q4 = Dec (year) – Feb (year+1).
    """
    m = QUARTER_RE.match(quarter)
    if not m:
        raise ValueError(f"Invalid quarter label {quarter!r}; expected format YYYYQN (e.g. 2024Q3)")
    year, q = int(m.group(1)), int(m.group(2))
    if q == 1:
        window = f"01mar{year}-31may{year}"
    elif q == 2:
        window = f"01jun{year}-31aug{year}"
    elif q == 3:
        window = f"01sep{year}-30nov{year}"
    else:  # Q4 straddles the year boundary
        feb_last = 29 if calendar.isleap(year + 1) else 28
        window = f"01dec{year}-{feb_last:02d}feb{year + 1}"
    return f"https://www.sec.gov/files/structureddata/data/form-13f-data-sets/{window}_form13f.zip"


def _download_bulk_zip(quarter: str, target: Path) -> Path:
    """Download the bulk 13F zip for one quarter, skipping if cached."""
    if target.exists():
        print(f"  Using cached zip: {target} ({target.stat().st_size / 1e6:.1f} MB)")
        return target
    url = _bulk_zip_url(quarter)
    print(f"  Fetching {url}")
    resp = requests.get(url, headers=SEC_HEADERS, timeout=600)
    resp.raise_for_status()
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_bytes(resp.content)
    print(f"  Downloaded {len(resp.content) / 1e6:.1f} MB → {target}")
    return target


def _read_bulk_tsv(
    archive: zipfile.ZipFile, name: str, overrides: dict | None = None
) -> pl.DataFrame:
    """Read one TSV inside the bulk zip as a Polars DataFrame."""
    with archive.open(name) as f:
        buf = io.BytesIO(f.read())
    return pl.read_csv(
        buf,
        separator="\t",
        infer_schema_length=10_000,
        schema_overrides=overrides or {},
    )


def _normalize_bulk_to_canonical(
    infotable: pl.DataFrame,
    coverpage: pl.DataFrame,
    submission: pl.DataFrame,
) -> pl.DataFrame:
    """Join the three bulk tables into the canonical per-cik schema.

    Output columns: cik, accession_no, issuer, cusip, value_thousands,
    shares, put_call, report_date, filing_date, company_name.
    """
    # SEC SUBMISSION.FILING_DATE is "DD-MON-YYYY" uppercase (e.g. "31-OCT-2024").
    # Parse to a Date so downstream filter by start_date/end_date works.
    submission = submission.filter(pl.col("SUBMISSIONTYPE") == "13F-HR").select(
        [
            pl.col("ACCESSION_NUMBER"),
            pl.col("CIK").cast(pl.Utf8).str.zfill(10).alias("cik"),
            pl.col("PERIODOFREPORT")
            .str.to_date(format="%d-%b-%Y", strict=False)
            .alias("report_date"),
            pl.col("FILING_DATE").str.to_date(format="%d-%b-%Y", strict=False).alias("filing_date"),
        ]
    )
    coverpage = coverpage.select(
        [
            pl.col("ACCESSION_NUMBER"),
            pl.col("FILINGMANAGER_NAME").alias("company_name"),
        ]
    )
    # INFOTABLE is the big table — keep only what the canonical schema needs.
    put_call = (
        pl.col("PUTCALL")
        .cast(pl.Utf8)
        .str.strip_chars()
        .replace("", None)
        .str.to_uppercase()
        .alias("put_call")
        if "PUTCALL" in infotable.columns
        else pl.lit(None, dtype=pl.Utf8).alias("put_call")
    )
    holdings = infotable.select(
        [
            pl.col("ACCESSION_NUMBER").alias("accession_no"),
            pl.col("NAMEOFISSUER").alias("issuer"),
            pl.col("CUSIP").alias("cusip"),
            pl.col("VALUE").cast(pl.Int64).alias("value_thousands"),
            pl.col("SSHPRNAMT").cast(pl.Int64).alias("shares"),
            put_call,
            pl.col("ACCESSION_NUMBER"),
        ]
    )

    # Inner-joins drop any holdings whose submission type isn't 13F-HR.
    return (
        holdings.join(submission, on="ACCESSION_NUMBER", how="inner")
        .join(coverpage, on="ACCESSION_NUMBER", how="inner")
        .select(
            [
                "cik",
                "accession_no",
                "issuer",
                "cusip",
                "value_thousands",
                "shares",
                "put_call",
                "report_date",
                "filing_date",
                "company_name",
            ]
        )
    )


def _run_bulk(quarters: list[str], bulk_root: Path) -> int:
    """Download + normalize one or more quarterly bulk sets."""
    for quarter in quarters:
        q_dir = bulk_root / quarter
        zip_path = q_dir / "bulk_13f.zip"
        out_path = q_dir / "institutional_holdings.parquet"

        print(f"\n{quarter}:")
        _download_bulk_zip(quarter, zip_path)

        with zipfile.ZipFile(zip_path) as archive:
            members = set(archive.namelist())
            required = {"INFOTABLE.tsv", "COVERPAGE.tsv", "SUBMISSION.tsv"}
            missing = required - members
            if missing:
                print(f"  ERROR: zip missing required tables: {sorted(missing)}")
                return 1

            infotable = _read_bulk_tsv(
                archive,
                "INFOTABLE.tsv",
                overrides={"OTHERMANAGER": pl.Utf8, "FIGI": pl.Utf8},
            )
            coverpage = _read_bulk_tsv(archive, "COVERPAGE.tsv")
            submission = _read_bulk_tsv(archive, "SUBMISSION.tsv")

        print(
            f"  Parsed {len(infotable):,} holdings  "
            f"{len(coverpage):,} coverpages  "
            f"{len(submission):,} submissions"
        )

        canonical = _normalize_bulk_to_canonical(infotable, coverpage, submission)
        canonical.write_parquet(out_path)

        n_cik = canonical["cik"].n_unique()
        n_issuers = canonical["issuer"].n_unique()
        total_value = canonical["value_thousands"].sum() / 1e12
        print(
            f"  Wrote {out_path.name}  "
            f"({len(canonical):,} rows, {n_cik:,} managers, "
            f"{n_issuers:,} unique issuers, ${total_value:.1f}T total)"
        )
    return 0


# --- Per-CIK mode (curated institutions via JSON submissions API) ---


def _run_per_cik(
    output_dir: Path,
    num_filings: int,
    max_institutions: int,
) -> int:
    institutions = INSTITUTIONS[:max_institutions] if max_institutions else INSTITUTIONS

    print(f"Downloading 13F data to: {output_dir}")
    print(f"Institutions: {len(institutions)}  Filings each: {num_filings}")

    all_filings: list[dict] = []
    for name, cik in institutions:
        filings = get_recent_13f_filings(cik, num_filings)
        all_filings.extend(filings)
        print(f"  {name}: {len(filings)} filings")
        time.sleep(RATE_LIMIT_SECONDS)

    if not all_filings:
        print("No filings retrieved.")
        return 1
    filings_df = pl.DataFrame(all_filings)

    all_holdings: list[dict] = []
    for row in filings_df.iter_rows(named=True):
        holdings = parse_13f_holdings(row["cik"], row["accession_number"])
        for h in holdings:
            h["report_date"] = row["report_date"]
            h["filing_date"] = row["filing_date"]
            h["company_name"] = row["company_name"]
        all_holdings.extend(holdings)
        print(
            f"  {row['company_name'][:32]:<32} {row['filing_date']}  {len(holdings):>5} positions"
        )
        time.sleep(RATE_LIMIT_SECONDS)

    if not all_holdings:
        print("No holdings parsed.")
        return 1

    holdings_df = pl.from_dicts(all_holdings, infer_schema_length=None).with_columns(
        pl.col("report_date").str.to_date(),
        pl.col("filing_date").str.to_date(),
    )
    stock_features, edge_list, coown_matrix, stocks = build_features_and_matrix(
        holdings_df, expected_ciks=[cik for _, cik in institutions]
    )

    holdings_path = output_dir / "institutional_holdings.parquet"
    edges_path = output_dir / "institution_stock_edges.parquet"
    features_path = output_dir / "stock_features.parquet"
    matrix_path = output_dir / "coownership_matrix.npy"
    stocks_path = output_dir / "coownership_stocks.txt"

    holdings_df.write_parquet(holdings_path)
    edge_list.write_parquet(edges_path)
    stock_features.write_parquet(features_path)
    np.save(matrix_path, coown_matrix)
    stocks_path.write_text("\n".join(stocks))

    print("")
    print(f"Wrote {holdings_path.name}   ({len(holdings_df):,} rows)")
    print(f"Wrote {edges_path.name}      ({len(edge_list):,} rows)")
    print(f"Wrote {features_path.name}   ({len(stock_features):,} stocks)")
    print(f"Wrote {matrix_path.name}     ({coown_matrix.shape})")
    print(f"Wrote {stocks_path.name}")
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description="Download SEC 13F institutional holdings")
    parser.add_argument(
        "--mode",
        choices=["per-cik", "bulk"],
        default="per-cik",
        help="per-cik: curated institutions via SEC JSON API (default). "
        "bulk: SEC quarterly bulk data sets (~80 MB zip per quarter).",
    )
    parser.add_argument(
        "--data-path",
        type=Path,
        default=None,
        help="Override output root (default: $ML4T_DATA_PATH)",
    )
    # Per-CIK args
    parser.add_argument(
        "--num-filings",
        type=int,
        default=4,
        help="[per-cik] Number of recent 13F-HR filings per institution (default 4)",
    )
    parser.add_argument(
        "--max-institutions",
        type=int,
        default=0,
        help="[per-cik] Limit to first N institutions (0 = all)",
    )
    # Bulk args
    parser.add_argument(
        "--quarters",
        type=str,
        default="",
        help="[bulk] Comma-separated filing-window labels (e.g. '2024Q2,2024Q3'). "
        "SEC labels by filing date: Q1=Mar-May, Q2=Jun-Aug, Q3=Sep-Nov, Q4=Dec-Feb.",
    )
    args = parser.parse_args()

    data_path = resolve_data_dir(args.data_path)
    root = data_path / "equities" / "positioning" / "13f"
    root.mkdir(parents=True, exist_ok=True)

    if args.mode == "bulk":
        if not args.quarters:
            parser.error("--mode bulk requires --quarters (e.g. --quarters 2024Q3)")
        quarters = [q.strip() for q in args.quarters.split(",") if q.strip()]
        # Validate early so a bad label doesn't surface only after a long download.
        for q in quarters:
            _bulk_zip_url(q)
        return _run_bulk(quarters, root / "bulk")

    return _run_per_cik(root, args.num_filings, args.max_institutions)


if __name__ == "__main__":
    raise SystemExit(main())

```