构建 13F 持仓面板与共同持仓特征
代码 《交易机器学习》
总结
该脚本通过两种方式收集机构持仓数据：可检索精选管理人的 SEC 13F 文件，也可下载 SEC 的季度批量文件。两种路径都会将持仓标准化为统一结构，其中批量路径旨在实现更广泛的覆盖。生成的数据可用于构建机构与股票之间的关联、股票层面特征以及股票共同持仓比较。
该脚本采用多项保护措施处理常见数据问题。如果需要完整的管理人集合，图构建会避开文件不完整的季度，防止将延迟申报误判为退出。构建多头股票特征时会排除期权，并拒绝其报告价值单位与美元计价特征约定冲突的较早文件。脚本还遵守 SEC 请求速率限制。其输出是持仓数据基础，而非交易策略；13F 披露具有周期性，文段并未证明其预测表现，也无法捕捉报告间隔期间的持仓变化。
核心观点
- 脚本既支持下载精选管理人的数据，也支持批量获取季度 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())
```在遵守原作品许可的前提下，附作者信息全文展示。 许可协议: MIT
此摘要由 Stratmill 研究智能体根据原文撰写，并非原文副本。