"""Partitioned WRDS equity inputs for daily, security-level research.
The standard WRDS zone intentionally stays small and monthly. A CRSP daily stock-file pull is a
different workload: the 1926--2012 U.S. common-stock sample has tens of millions of rows and must
never be materialized as one pandas frame. This module therefore treats each stable ``permno``
bucket as an ordinary immutable :func:`numeraire_dataset.zones.wrds.pull_raw` cache and binds those
independent caches, a same-vintage CRSP release marker, the CRSP value-weighted market, and the
daily risk-free rate into one canonical collection manifest.
The collection is an iterator, not a frame. Consumers process one asset bucket at a time and may
load the two small calendar series separately. Licensed observations remain under the configured
data home and are never embedded in the manifest; the manifest contains only non-secret identities,
row counts, bounds, and content/query digests.
``convention="siz"`` is the paper-era default: legacy ``crspm.dsf`` share codes 10/11 and the
legacy ``crspm.dsi`` value-weighted market. ``"ciz"`` is an explicit current-format sensitivity
using CRSP's official share-code mapping. The convention is part of every source and collection
identity, so the two return/delisting regimes cannot be silently mixed.
WRDS ``ff.factors_daily.rf`` is stored as a decimal simple return (for example, 6 bp is ``0.0006``),
not percentage points. Its current history begins in July 1926, later than the January 1926 CRSP
stock and market histories. The clean calendar therefore fails closed for a requested range whose
market sessions are not all covered by RF; it never fills the early gap or invents zero rates.
For the BAB paper horizon, use 1926-07-01 as the effective daily start; this still supplies more
than the required 750 warm-up sessions before the Original factor series starts in April 1929.
"""
from __future__ import annotations
import datetime as dt
import hashlib
import json
import os
import stat
import tempfile
from collections.abc import Iterator, Mapping
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Literal, cast
import pandas as pd
from numeraire_dataset.paths import data_home
from numeraire_dataset.zones import wrds
from numeraire_dataset.zones.raw import RawMeta, read_meta
from numeraire_dataset.zones.steps import get_step
if TYPE_CHECKING: # pragma: no cover - typing only
import wrds as _wrds # pyright: ignore[reportMissingImports]
DailyEquityConvention = Literal["siz", "ciz"]
DEFAULT_DAILY_PARTITIONS = 64
MAX_DAILY_PARTITIONS = 256
CRSP_RELEASE_MARKER_SQL = "select max(mthcaldt) as date from crspm.msf_v2"
_MANIFEST_VERSION = 2
_DATA_VINTAGE_HASH_LEN = 12
_COLLECTION_KIND = "crsp_daily_equity"
_RECIPE = "release-marker-permno-bucket-market-rf-v2"
_RELEASE_MARKER_SOURCE = "crsp_release_marker_msf_v2"
_UNIT_CONTRACTS = {
"release_marker": "crsp_msf_v2_max_mthcaldt_date",
"stock": "crsp_daily_decimal_simple_return",
"market": "crsp_dsi_vwretd_decimal_simple_return",
"risk_free": "ff_factors_daily_rf_decimal_simple_return",
}
_MONTHLY_COLLECTION_KIND = "crsp_monthly_equity_targets_siz"
_MONTHLY_RECIPE = "release-marker-permno-bucket-msf-names-delist-rf-v2"
_MONTHLY_UNIT_CONTRACTS = {
"release_marker": "crsp_msf_v2_max_mthcaldt_date",
"msf": "crsp_monthly_decimal_simple_return",
"msedelist": "crsp_monthly_decimal_delisting_return",
"risk_free": "ff_factors_monthly_rf_decimal_simple_return_source_month_label",
}
def _date(value: str | dt.date | pd.Timestamp, *, label: str) -> dt.date:
try:
stamp = pd.Timestamp(value)
except (TypeError, ValueError) as exc:
raise ValueError(f"{label} must be a valid date") from exc
if pd.isna(stamp):
raise ValueError(f"{label} must be a valid date")
if stamp.tz is not None:
raise ValueError(f"{label} must be timezone-naive")
if stamp != stamp.normalize():
raise ValueError(f"{label} must not contain a time of day")
return stamp.date()
def _date_range(
start_date: str | dt.date | pd.Timestamp,
end_date: str | dt.date | pd.Timestamp,
) -> tuple[dt.date, dt.date]:
start = _date(start_date, label="start_date")
end = _date(end_date, label="end_date")
if start > end:
raise ValueError("start_date must not be after end_date")
return start, end
def _month_range(
start_date: str | dt.date | pd.Timestamp,
end_date: str | dt.date | pd.Timestamp,
) -> tuple[dt.date, dt.date]:
start, end = _date_range(start_date, end_date)
start_month = pd.Timestamp(start).to_period("M")
end_month = pd.Timestamp(end).to_period("M")
return start_month.start_time.date(), end_month.end_time.normalize().date()
def _partition(bucket: int, partitions: int) -> tuple[int, int]:
if isinstance(partitions, bool) or not isinstance(partitions, int):
raise TypeError("partitions must be an integer")
if not 1 <= partitions <= MAX_DAILY_PARTITIONS:
raise ValueError(f"partitions must be in 1..{MAX_DAILY_PARTITIONS}")
if isinstance(bucket, bool) or not isinstance(bucket, int):
raise TypeError("bucket must be an integer")
if not 0 <= bucket < partitions:
raise ValueError("bucket must be in 0..partitions-1")
return bucket, partitions
def _convention(value: str) -> DailyEquityConvention:
if value not in {"siz", "ciz"}:
raise ValueError("convention must be 'siz' or 'ciz'")
return cast("DailyEquityConvention", value)
def _component(value: str, *, label: str) -> str:
if (
not value
or value in {".", ".."}
or any(char in value for char in ("/", "\\", "\0", "@", "#"))
):
raise ValueError(f"{label} must be a safe, non-empty cache component")
return value
def _query_hash(sql: str) -> str:
return "sha256:" + hashlib.sha256(sql.encode("utf-8")).hexdigest()
def _release_marker_date(frame: pd.DataFrame) -> dt.date:
"""Validate the one-cell CRSP release marker and return its normalized date."""
if tuple(frame.columns) != ("date",) or len(frame) != 1:
raise RuntimeError("CRSP release marker must contain exactly one non-null date cell")
try:
return _date(
cast("str | dt.date | pd.Timestamp", frame.iloc[0]["date"]),
label="CRSP release marker",
)
except ValueError as exc:
raise RuntimeError("CRSP release marker must contain exactly one valid date") from exc
def _live_release_marker(conn: _wrds.Connection) -> dt.date:
return _release_marker_date(conn.raw_sql(CRSP_RELEASE_MARKER_SQL))
def _require_unchanged_release_marker(
cached: dt.date,
observed: dt.date,
*,
vintage: str,
stage: str,
) -> None:
if observed != cached:
raise RuntimeError(
f"CRSP release marker changed {stage}; raw inputs under vintage {vintage!r} "
"cannot be mixed or extended. Choose a new vintage and rerun the collection."
)
def _cached_release_marker(
conn: _wrds.Connection,
*,
vintage: str,
home: str | Path | None,
) -> tuple[pd.DataFrame, RawMeta, dt.date]:
"""Create or revalidate the immutable marker raw cache for this vintage."""
frame = wrds.pull_raw(
conn,
_RELEASE_MARKER_SOURCE,
CRSP_RELEASE_MARKER_SQL,
vintage,
home=home,
pit_status="snapshot",
# A release marker is the fixed consistency anchor for a vintage. Even if callers request
# a refresh for a brand-new collection, the existing same-vintage anchor is never replaced.
refresh=False,
)
marker = _release_marker_date(frame)
meta = read_meta(_RELEASE_MARKER_SOURCE, vintage, home=home)
return frame, meta, marker
def _clean_step_identity(
name: str, *, params: Mapping[str, object] | None = None
) -> dict[str, object]:
registered = get_step(name)
return {
"name": registered.name,
"version": registered.version,
"params": {} if params is None else dict(params),
}
def _daily_clean_pipeline(convention: DailyEquityConvention) -> list[dict[str, object]]:
return [
_clean_step_identity("daily_market_rf_clean"),
_clean_step_identity(
"crsp_daily_equity_clean",
params={"convention": convention},
),
_clean_step_identity("daily_stock_excess_clean"),
]
def _monthly_clean_pipeline() -> list[dict[str, object]]:
return [
_clean_step_identity("monthly_risk_free_clean"),
_clean_step_identity(
"crsp_monthly_clean",
params={"exchanges": None, "performance_delist_fill": None},
),
_clean_step_identity("monthly_stock_excess_clean"),
]
[docs]
def crsp_daily_stock_sql(
*,
start_date: str | dt.date | pd.Timestamp,
end_date: str | dt.date | pd.Timestamp,
bucket: int,
partitions: int = DEFAULT_DAILY_PARTITIONS,
convention: DailyEquityConvention = "siz",
) -> str:
"""Return deterministic SQL for one complete-history ``permno`` bucket.
Values are validated and rendered as typed date/integer literals rather than passed through
from arbitrary SQL text. Exact SQL bytes are subsequently pinned by the raw-zone query hash.
The query deliberately orders by security and date so connector/backend row ordering cannot
move the content digest.
"""
start, end = _date_range(start_date, end_date)
bucket, partitions = _partition(bucket, partitions)
convention = _convention(convention)
if convention == "siz":
return (
"select d.permno, d.date, d.ret, n.shrcd, n.exchcd "
"from crspm.dsf as d "
"join crspm.msenames as n "
"on d.permno = n.permno and d.date between n.namedt and n.nameendt "
f"where d.date between date '{start.isoformat()}' and date '{end.isoformat()}' "
"and n.shrcd in (10, 11) "
f"and mod(d.permno, {partitions}) = {bucket} "
"order by d.permno, d.date"
)
return (
"select permno, dlycaldt as date, dlyret as ret, dlydelflg, "
"dlyretmissflg, dlyretdurflg, primaryexch, conditionaltype, tradingstatusflg "
"from crspm.dsf_v2 "
f"where dlycaldt between date '{start.isoformat()}' and date '{end.isoformat()}' "
"and sharetype = 'NS' and securitytype = 'EQTY' and securitysubtype = 'COM' "
"and usincflg = 'Y' and issuertype in ('ACOR', 'CORP') "
f"and mod(permno, {partitions}) = {bucket} "
"order by permno, dlycaldt"
)
[docs]
def crsp_daily_market_sql(
*, start_date: str | dt.date | pd.Timestamp, end_date: str | dt.date | pd.Timestamp
) -> str:
"""CRSP value-weighted U.S. market used by the paper (not an S&P 500 series)."""
start, end = _date_range(start_date, end_date)
return (
"select date, vwretd as market_return from crspm.dsi "
f"where date between date '{start.isoformat()}' and date '{end.isoformat()}' "
"order by date"
)
[docs]
def ff_daily_risk_free_sql(
*, start_date: str | dt.date | pd.Timestamp, end_date: str | dt.date | pd.Timestamp
) -> str:
"""Daily U.S. RF, supplied by WRDS in decimal units and currently available from 1926-07."""
start, end = _date_range(start_date, end_date)
return (
"select date, rf as risk_free from ff.factors_daily "
f"where date between date '{start.isoformat()}' and date '{end.isoformat()}' "
"order by date"
)
[docs]
def crsp_monthly_stock_sql(
*,
start_date: str | dt.date | pd.Timestamp,
end_date: str | dt.date | pd.Timestamp,
bucket: int,
partitions: int = DEFAULT_DAILY_PARTITIONS,
) -> str:
"""One SIZ monthly-stock bucket, independent of names and delisting events."""
start, end = _month_range(start_date, end_date)
bucket, partitions = _partition(bucket, partitions)
return (
"select permno, permco, date, ret, prc, shrout from crspm.msf "
f"where date between date '{start.isoformat()}' and date '{end.isoformat()}' "
f"and mod(permno, {partitions}) = {bucket} "
"order by permno, date"
)
[docs]
def crsp_monthly_names_sql(
*,
start_date: str | dt.date | pd.Timestamp,
end_date: str | dt.date | pd.Timestamp,
bucket: int,
partitions: int = DEFAULT_DAILY_PARTITIONS,
) -> str:
"""Name histories overlapping the target range for one SIZ ``permno`` bucket."""
start, end = _month_range(start_date, end_date)
bucket, partitions = _partition(bucket, partitions)
return (
"select permno, permco, namedt, nameendt, shrcd, exchcd, siccd, ticker "
"from crspm.msenames "
f"where namedt <= date '{end.isoformat()}' "
f"and coalesce(nameendt, date '9999-12-31') >= date '{start.isoformat()}' "
f"and mod(permno, {partitions}) = {bucket} "
"order by permno, namedt"
)
[docs]
def crsp_monthly_delist_sql(
*,
start_date: str | dt.date | pd.Timestamp,
end_date: str | dt.date | pd.Timestamp,
bucket: int,
partitions: int = DEFAULT_DAILY_PARTITIONS,
) -> str:
"""Monthly delisting events pulled independently so terminal-only rows cannot disappear."""
start, end = _month_range(start_date, end_date)
bucket, partitions = _partition(bucket, partitions)
return (
"select permno, dlstdt as date, dlret, dlstcd from crspm.msedelist "
f"where dlstdt between date '{start.isoformat()}' and date '{end.isoformat()}' "
f"and mod(permno, {partitions}) = {bucket} "
"order by permno, dlstdt"
)
[docs]
def ff_monthly_risk_free_sql(
*, start_date: str | dt.date | pd.Timestamp, end_date: str | dt.date | pd.Timestamp
) -> str:
"""Monthly FF RF in decimal units; WRDS source dates label months at their first day."""
start, end = _month_range(start_date, end_date)
return (
"select date, rf as risk_free from ff.factors_monthly "
f"where date between date '{start.isoformat()}' and date '{end.isoformat()}' "
"order by date"
)
[docs]
def siz_monthly_target_queries(
*, start_date: str | dt.date | pd.Timestamp, end_date: str | dt.date | pd.Timestamp
) -> dict[str, str]:
"""Legacy monthly inputs for :func:`zones.clean.crsp_monthly_clean`.
This is separate from the standard 1955+ query family so a long BAB reproduction neither
changes nor triggers the Compustat-oriented standard pull. Pass the returned frames to the
existing cleaner, using ``exchanges=None, performance_delist_fill=None`` for the BAB-paper
universe and non-imputation convention. For scalable/restartable work, prefer the partitioned
:func:`prepare_crsp_monthly_equity_targets` collection.
"""
start, end = _month_range(start_date, end_date)
bounds = f"between date '{start.isoformat()}' and date '{end.isoformat()}'"
return {
"msf": (
"select permno, permco, date, ret, prc, shrout from crspm.msf "
f"where date {bounds} order by permno, date"
),
"msenames": (
"select permno, permco, namedt, nameendt, shrcd, exchcd, siccd, ticker "
"from crspm.msenames "
f"where namedt <= date '{end.isoformat()}' "
f"and coalesce(nameendt, date '9999-12-31') >= date '{start.isoformat()}' "
"order by permno, namedt"
),
"msedelist": (
"select permno, dlstdt as date, dlret, dlstcd from crspm.msedelist "
f"where dlstdt {bounds} order by permno, dlstdt"
),
}
def _source_names(
*, convention: DailyEquityConvention, start: dt.date, end: dt.date, partitions: int
) -> tuple[tuple[str, ...], str, str]:
date_token = f"{start:%Y%m%d}_{end:%Y%m%d}"
width = max(3, len(str(partitions - 1)))
stocks = tuple(
f"crsp_daily_{convention}_{date_token}_p{partitions:03d}_b{bucket:0{width}d}"
for bucket in range(partitions)
)
return stocks, f"crsp_dsi_{date_token}", f"ff_daily_rf_{date_token}"
def _monthly_source_names(
*, start: dt.date, end: dt.date, partitions: int
) -> tuple[tuple[tuple[str, str, str], ...], str]:
date_token = f"{start:%Y%m}_{end:%Y%m}"
width = max(3, len(str(partitions - 1)))
buckets = tuple(
tuple(
f"crsp_monthly_siz_{date_token}_p{partitions:03d}_b{bucket:0{width}d}_{role}"
for role in ("msf", "msenames", "msedelist")
)
for bucket in range(partitions)
)
return cast("tuple[tuple[str, str, str], ...]", buckets), f"ff_monthly_rf_{date_token}"
[docs]
@dataclass(frozen=True)
class DailyEquityManifestEntry:
"""Non-secret identity of one independently cached raw input."""
role: str
source: str
query_hash: str
content_digest: str
content_digest_version: int
row_count: int
pit_status: str
min_date: str | None
max_date: str | None
bucket: int | None = None
[docs]
@dataclass(frozen=True)
class MonthlyEquityManifestEntry:
"""Path-free identity of one monthly SIZ target input."""
role: str
source: str
query_hash: str
content_digest: str
content_digest_version: int
row_count: int
pit_status: str
min_date: str | None
max_date: str | None
bucket: int | None = None
[docs]
@dataclass(frozen=True)
class MonthlyEquityRawPartition:
"""Three independently pulled raw frames for one monthly ``permno`` bucket."""
bucket: int
msf: pd.DataFrame
msenames: pd.DataFrame
msedelist: pd.DataFrame
def _frame_bounds(
frame: pd.DataFrame, *, date_column: str = "date"
) -> tuple[str | None, str | None]:
if date_column not in frame:
raise ValueError(f"raw WRDS frame has no {date_column!r} column")
dates = pd.to_datetime(frame[date_column], errors="coerce")
if dates.isna().any():
raise ValueError("raw WRDS frame contains an invalid date")
if getattr(dates.dt, "tz", None) is not None:
raise ValueError("raw WRDS frame dates must be timezone-naive")
if frame.empty:
return None, None
return dates.min().date().isoformat(), dates.max().date().isoformat()
def _entry(
*, role: str, source: str, meta: RawMeta, frame: pd.DataFrame, bucket: int | None = None
) -> DailyEquityManifestEntry:
minimum, maximum = _frame_bounds(frame)
if meta.source != source:
raise RuntimeError("raw metadata source does not match its requested collection source")
if meta.row_count != len(frame):
raise RuntimeError("raw metadata row count does not match its returned frame")
return DailyEquityManifestEntry(
role=role,
source=source,
bucket=bucket,
query_hash=meta.query_hash,
content_digest=meta.content_digest,
content_digest_version=meta.content_digest_version,
row_count=meta.row_count,
pit_status=meta.pit_status,
min_date=minimum,
max_date=maximum,
)
def _monthly_entry(
*, role: str, source: str, meta: RawMeta, frame: pd.DataFrame, bucket: int | None = None
) -> MonthlyEquityManifestEntry:
minimum, maximum = _frame_bounds(
frame,
date_column="namedt" if role == "msenames" else "date",
)
if meta.source != source:
raise RuntimeError("raw metadata source does not match its requested collection source")
if meta.row_count != len(frame):
raise RuntimeError("raw metadata row count does not match its returned frame")
return MonthlyEquityManifestEntry(
role=role,
source=source,
bucket=bucket,
query_hash=meta.query_hash,
content_digest=meta.content_digest,
content_digest_version=meta.content_digest_version,
row_count=meta.row_count,
pit_status=meta.pit_status,
min_date=minimum,
max_date=maximum,
)
def _manifest_payload(
*,
convention: DailyEquityConvention,
start: dt.date,
end: dt.date,
partitions: int,
vintage: str,
entries: tuple[DailyEquityManifestEntry, ...],
) -> dict[str, object]:
return {
"version": _MANIFEST_VERSION,
"kind": _COLLECTION_KIND,
"recipe": _RECIPE,
"convention": convention,
"start_date": start.isoformat(),
"end_date": end.isoformat(),
"partitions": partitions,
"vintage": vintage,
"pit_status": "snapshot",
"redistributable": False,
"unit_contracts": _UNIT_CONTRACTS,
"clean_pipeline": _daily_clean_pipeline(convention),
"entries": [asdict(entry) for entry in entries],
}
def _monthly_manifest_payload(
*,
start: dt.date,
end: dt.date,
partitions: int,
vintage: str,
entries: tuple[MonthlyEquityManifestEntry, ...],
) -> dict[str, object]:
return {
"version": _MANIFEST_VERSION,
"kind": _MONTHLY_COLLECTION_KIND,
"recipe": _MONTHLY_RECIPE,
"convention": "siz",
"start_date": start.isoformat(),
"end_date": end.isoformat(),
"partitions": partitions,
"vintage": vintage,
"pit_status": "snapshot",
"redistributable": False,
"clean_pipeline": _monthly_clean_pipeline(),
"unit_contracts": _MONTHLY_UNIT_CONTRACTS,
"entries": [asdict(entry) for entry in entries],
}
def _canonical_json(payload: Mapping[str, object]) -> str:
return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
def _manifest_hash(payload: Mapping[str, object]) -> str:
return "sha256:" + hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest()
def _manifest_location(
*,
home: str | Path | None,
convention: DailyEquityConvention,
start: dt.date,
end: dt.date,
partitions: int,
vintage: str,
) -> Path:
vintage = _component(vintage, label="vintage")
root = data_home(home).resolve()
spec = {
"convention": convention,
"start_date": start.isoformat(),
"end_date": end.isoformat(),
"partitions": partitions,
"recipe": _RECIPE,
"clean_pipeline": _daily_clean_pipeline(convention),
}
token = hashlib.sha256(_canonical_json(spec).encode("utf-8")).hexdigest()[:20]
target = root / "collections" / _COLLECTION_KIND / token / vintage / "manifest.json"
current = root
for component in target.relative_to(root).parts:
current /= component
if current.is_symlink():
raise RuntimeError(f"collection manifest path contains a symlink at {current}")
if not target.resolve(strict=False).is_relative_to(root):
raise RuntimeError("collection manifest path resolves outside the configured data home")
return target
def _monthly_manifest_location(
*,
home: str | Path | None,
start: dt.date,
end: dt.date,
partitions: int,
vintage: str,
) -> Path:
vintage = _component(vintage, label="vintage")
root = data_home(home).resolve()
spec = {
"convention": "siz",
"start_date": start.isoformat(),
"end_date": end.isoformat(),
"partitions": partitions,
"recipe": _MONTHLY_RECIPE,
"clean_pipeline": _monthly_clean_pipeline(),
}
token = hashlib.sha256(_canonical_json(spec).encode("utf-8")).hexdigest()[:20]
target = root / "collections" / _MONTHLY_COLLECTION_KIND / token / vintage / "manifest.json"
current = root
for component in target.relative_to(root).parts:
current /= component
if current.is_symlink():
raise RuntimeError(f"collection manifest path contains a symlink at {current}")
if not target.resolve(strict=False).is_relative_to(root):
raise RuntimeError("collection manifest path resolves outside the configured data home")
return target
def _publish_manifest(path: Path, payload: Mapping[str, object]) -> None:
body = (_canonical_json(payload) + "\n").encode("utf-8")
path.parent.mkdir(parents=True, exist_ok=True)
if path.is_symlink() or any(parent.is_symlink() for parent in path.parents[:4]):
raise RuntimeError("collection manifest path contains a symlink")
if path.exists():
if not path.is_file() or stat.S_ISLNK(path.lstat().st_mode):
raise RuntimeError("collection manifest is not a regular file")
if path.read_bytes() != body:
raise RuntimeError(
"immutable collection manifest differs from cached raw identities; "
"choose a new vintage"
)
return
descriptor, temporary_name = tempfile.mkstemp(prefix=".manifest-", dir=path.parent)
temporary = Path(temporary_name)
try:
with os.fdopen(descriptor, "wb") as stream:
stream.write(body)
stream.flush()
os.fsync(stream.fileno())
try:
os.link(temporary, path, follow_symlinks=False)
except FileExistsError:
if path.is_symlink() or path.read_bytes() != body:
raise RuntimeError("concurrent collection manifest publication disagreed") from None
finally:
temporary.unlink(missing_ok=True)
def _parse_manifest(path: Path) -> dict[str, object]:
try:
mode = path.lstat().st_mode
except FileNotFoundError:
raise FileNotFoundError("collection manifest is missing") from None
except OSError as exc:
raise RuntimeError("collection manifest cannot be inspected safely") from exc
if stat.S_ISLNK(mode) or not stat.S_ISREG(mode):
raise RuntimeError("collection manifest is not a regular file")
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError:
raise FileNotFoundError("collection manifest is missing") from None
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
raise RuntimeError("collection manifest is unreadable") from exc
if not isinstance(payload, dict):
raise RuntimeError("collection manifest root must be an object")
return payload
[docs]
@dataclass(frozen=True)
class DailyEquityCollection:
"""Handle for a complete partitioned daily-equity input collection.
``iter_stock_partitions`` validates and yields exactly one cached asset bucket at a time. It
intentionally exposes no method that concatenates the buckets.
"""
convention: DailyEquityConvention
start_date: dt.date
end_date: dt.date
partitions: int
vintage: str
entries: tuple[DailyEquityManifestEntry, ...]
manifest_hash: str
data_vintage: str
manifest_path: Path
_home: Path
_manifest_json: str
[docs]
def manifest(self) -> dict[str, object]:
"""Return a defensive copy of the canonical, path-free manifest."""
payload = json.loads(self._manifest_json)
if not isinstance(payload, dict): # construction invariant; fail closed if memory corrupts
raise RuntimeError("in-memory collection manifest is invalid")
return payload
def _sql_for_entry(self, entry: DailyEquityManifestEntry) -> str:
if entry.role == "release_marker":
if entry.bucket is not None:
raise RuntimeError("release-marker manifest entry cannot have a bucket")
return CRSP_RELEASE_MARKER_SQL
if entry.role == "stock":
if entry.bucket is None:
raise RuntimeError("stock manifest entry has no bucket")
return crsp_daily_stock_sql(
start_date=self.start_date,
end_date=self.end_date,
bucket=entry.bucket,
partitions=self.partitions,
convention=self.convention,
)
if entry.role == "market":
return crsp_daily_market_sql(start_date=self.start_date, end_date=self.end_date)
if entry.role == "risk_free":
return ff_daily_risk_free_sql(start_date=self.start_date, end_date=self.end_date)
raise RuntimeError(f"unknown collection role {entry.role!r}")
def _read(self, entry: DailyEquityManifestEntry) -> pd.DataFrame:
sql = self._sql_for_entry(entry)
frame, meta, _ = wrds._validated_cached_frame(
entry.source,
self.vintage,
home=self._home,
sql=sql,
)
if entry.role == "release_marker":
_release_marker_date(frame)
observed = _entry(
role=entry.role,
source=entry.source,
meta=meta,
frame=frame,
bucket=entry.bucket,
)
if observed != entry:
raise RuntimeError("cached raw input no longer matches the collection manifest")
return frame
def _validate_release_marker_cache(self) -> None:
if not self.entries or self.entries[0].role != "release_marker":
raise RuntimeError("collection has no leading CRSP release marker")
self._read(self.entries[0])
[docs]
def iter_stock_partitions(self) -> Iterator[pd.DataFrame]:
"""Yield validated raw stock buckets in numeric bucket order, never concatenated."""
self._validate_release_marker_cache()
stock_entries = tuple(entry for entry in self.entries if entry.role == "stock")
if tuple(entry.bucket for entry in stock_entries) != tuple(range(self.partitions)):
raise RuntimeError("collection manifest has incomplete or misordered stock buckets")
for entry in stock_entries:
yield self._read(entry)
[docs]
def raw_market(self) -> pd.DataFrame:
"""Read the small CRSP value-weighted market input."""
self._validate_release_marker_cache()
entries = [entry for entry in self.entries if entry.role == "market"]
if len(entries) != 1:
raise RuntimeError("collection must contain exactly one market input")
return self._read(entries[0])
[docs]
def raw_risk_free(self) -> pd.DataFrame:
"""Read the small daily risk-free input."""
self._validate_release_marker_cache()
entries = [entry for entry in self.entries if entry.role == "risk_free"]
if len(entries) != 1:
raise RuntimeError("collection must contain exactly one risk-free input")
return self._read(entries[0])
[docs]
def market_rf(self) -> pd.DataFrame:
"""Return the strict CRSP-market/RF frame; raise on any uncovered market session."""
from numeraire_dataset.zones.clean import daily_market_rf_clean
return daily_market_rf_clean(self.raw_market(), self.raw_risk_free())
[docs]
def iter_clean_stock_partitions(self) -> Iterator[pd.DataFrame]:
"""Yield cleaned stock-excess-return buckets against this collection's daily RF."""
from numeraire_dataset.zones.clean import (
crsp_daily_equity_clean,
daily_stock_excess_clean,
)
market_rf = self.market_rf()
for raw in self.iter_stock_partitions():
clean = crsp_daily_equity_clean(raw, convention=self.convention)
yield daily_stock_excess_clean(clean, market_rf)
[docs]
@dataclass(frozen=True)
class MonthlyEquityTargetCollection:
"""SIZ monthly target returns, streamed as independent ``permno`` buckets."""
start_date: dt.date
end_date: dt.date
partitions: int
vintage: str
entries: tuple[MonthlyEquityManifestEntry, ...]
manifest_hash: str
data_vintage: str
manifest_path: Path
_home: Path
_manifest_json: str
[docs]
def manifest(self) -> dict[str, object]:
"""Return a defensive copy of the frozen, path-free canonical manifest."""
payload = json.loads(self._manifest_json)
if not isinstance(payload, dict):
raise RuntimeError("in-memory collection manifest is invalid")
return payload
def _sql_for_entry(self, entry: MonthlyEquityManifestEntry) -> str:
if entry.role == "release_marker":
if entry.bucket is not None:
raise RuntimeError("monthly release-marker entry cannot have a bucket")
return CRSP_RELEASE_MARKER_SQL
if entry.role == "risk_free":
if entry.bucket is not None:
raise RuntimeError("monthly risk-free manifest entry cannot have a bucket")
return ff_monthly_risk_free_sql(
start_date=self.start_date,
end_date=self.end_date,
)
if entry.bucket is None:
raise RuntimeError("monthly stock-input manifest entry has no bucket")
builders = {
"msf": crsp_monthly_stock_sql,
"msenames": crsp_monthly_names_sql,
"msedelist": crsp_monthly_delist_sql,
}
try:
builder = builders[entry.role]
except KeyError as exc:
raise RuntimeError(f"unknown monthly collection role {entry.role!r}") from exc
return builder(
start_date=self.start_date,
end_date=self.end_date,
bucket=entry.bucket,
partitions=self.partitions,
)
def _read(self, entry: MonthlyEquityManifestEntry) -> pd.DataFrame:
frame, meta, _ = wrds._validated_cached_frame(
entry.source,
self.vintage,
home=self._home,
sql=self._sql_for_entry(entry),
)
if entry.role == "release_marker":
_release_marker_date(frame)
observed = _monthly_entry(
role=entry.role,
source=entry.source,
meta=meta,
frame=frame,
bucket=entry.bucket,
)
if observed != entry:
raise RuntimeError("cached raw input no longer matches the monthly collection manifest")
return frame
def _validate_release_marker_cache(self) -> None:
if not self.entries or self.entries[0].role != "release_marker":
raise RuntimeError("monthly collection has no leading CRSP release marker")
self._read(self.entries[0])
[docs]
def iter_raw_partitions(self) -> Iterator[MonthlyEquityRawPartition]:
"""Yield MSF, overlapping names, and independent delists for one bucket at a time."""
self._validate_release_marker_cache()
for bucket in range(self.partitions):
offset = 1 + bucket * 3
bucket_entries = self.entries[offset : offset + 3]
if tuple(entry.role for entry in bucket_entries) != (
"msf",
"msenames",
"msedelist",
) or tuple(entry.bucket for entry in bucket_entries) != (bucket, bucket, bucket):
raise RuntimeError("monthly collection entries are incomplete or misordered")
yield MonthlyEquityRawPartition(
bucket=bucket,
msf=self._read(bucket_entries[0]),
msenames=self._read(bucket_entries[1]),
msedelist=self._read(bucket_entries[2]),
)
[docs]
def raw_risk_free(self) -> pd.DataFrame:
"""Read the independently cached monthly Fama--French RF input."""
self._validate_release_marker_cache()
if len(self.entries) != self.partitions * 3 + 2:
raise RuntimeError("monthly collection has the wrong number of inputs")
entry = self.entries[-1]
if entry.role != "risk_free" or entry.bucket is not None:
raise RuntimeError("monthly collection has no final risk-free input")
return self._read(entry)
[docs]
def risk_free(self) -> pd.DataFrame:
"""Return decimal RF with its WRDS source-month label normalized to month-end."""
from numeraire_dataset.zones.clean import monthly_risk_free_clean
return monthly_risk_free_clean(self.raw_risk_free())
[docs]
def iter_clean_partitions(self) -> Iterator[pd.DataFrame]:
"""Yield no-exchange-screen SIZ returns with same-month RF, one bucket at a time.
The BAB-paper default keeps common shares but applies no exchange screen and does not invent
an unreported performance-delisting return. ``source_ret``, ``dlret``, adjusted ``ret``,
``risk_free``, and ``excess_ret`` remain side-by-side for audit and downstream targets.
"""
from numeraire_dataset.zones.clean import (
crsp_monthly_clean,
monthly_stock_excess_clean,
)
risk_free = self.risk_free()
for raw in self.iter_raw_partitions():
stock = crsp_monthly_clean(
raw.msf,
raw.msenames,
raw.msedelist,
exchanges=None,
performance_delist_fill=None,
)
yield monthly_stock_excess_clean(stock, risk_free)
def _build_collection(
*,
convention: DailyEquityConvention,
start: dt.date,
end: dt.date,
partitions: int,
vintage: str,
home: str | Path | None,
entries: tuple[DailyEquityManifestEntry, ...],
publish: bool,
) -> DailyEquityCollection:
payload = _manifest_payload(
convention=convention,
start=start,
end=end,
partitions=partitions,
vintage=vintage,
entries=entries,
)
digest = _manifest_hash(payload)
payload["manifest_hash"] = digest
payload["data_vintage"] = (
f"{_COLLECTION_KIND}_{convention}@{digest.removeprefix('sha256:')[:_DATA_VINTAGE_HASH_LEN]}"
)
path = _manifest_location(
home=home,
convention=convention,
start=start,
end=end,
partitions=partitions,
vintage=vintage,
)
if publish:
_publish_manifest(path, payload)
return DailyEquityCollection(
convention=convention,
start_date=start,
end_date=end,
partitions=partitions,
vintage=vintage,
entries=entries,
manifest_hash=digest,
data_vintage=str(payload["data_vintage"]),
manifest_path=path,
_home=data_home(home).resolve(),
_manifest_json=_canonical_json(payload),
)
def _build_monthly_collection(
*,
start: dt.date,
end: dt.date,
partitions: int,
vintage: str,
home: str | Path | None,
entries: tuple[MonthlyEquityManifestEntry, ...],
publish: bool,
) -> MonthlyEquityTargetCollection:
payload = _monthly_manifest_payload(
start=start,
end=end,
partitions=partitions,
vintage=vintage,
entries=entries,
)
digest = _manifest_hash(payload)
payload["manifest_hash"] = digest
payload["data_vintage"] = (
f"{_MONTHLY_COLLECTION_KIND}@{digest.removeprefix('sha256:')[:_DATA_VINTAGE_HASH_LEN]}"
)
path = _monthly_manifest_location(
home=home,
start=start,
end=end,
partitions=partitions,
vintage=vintage,
)
if publish:
_publish_manifest(path, payload)
return MonthlyEquityTargetCollection(
start_date=start,
end_date=end,
partitions=partitions,
vintage=vintage,
entries=entries,
manifest_hash=digest,
data_vintage=str(payload["data_vintage"]),
manifest_path=path,
_home=data_home(home).resolve(),
_manifest_json=_canonical_json(payload),
)
def _validate_entry_layout(
entries: tuple[DailyEquityManifestEntry, ...],
*,
convention: DailyEquityConvention,
start: dt.date,
end: dt.date,
partitions: int,
) -> None:
stock_sources, market_source, rf_source = _source_names(
convention=convention,
start=start,
end=end,
partitions=partitions,
)
if len(entries) != partitions + 3:
raise RuntimeError("collection manifest has the wrong number of inputs")
marker = entries[0]
if (
marker.role != "release_marker"
or marker.source != _RELEASE_MARKER_SOURCE
or marker.bucket is not None
or marker.row_count != 1
or marker.min_date is None
or marker.min_date != marker.max_date
or marker.query_hash != _query_hash(CRSP_RELEASE_MARKER_SQL)
):
raise RuntimeError("collection manifest release marker does not match its requested spec")
stock = entries[1 : partitions + 1]
if tuple(entry.role for entry in stock) != ("stock",) * partitions:
raise RuntimeError("collection manifest stock entries are missing or misordered")
if tuple(entry.bucket for entry in stock) != tuple(range(partitions)):
raise RuntimeError("collection manifest stock buckets are missing or misordered")
if tuple(entry.source for entry in stock) != stock_sources:
raise RuntimeError("collection manifest stock sources do not match its requested spec")
if entries[partitions + 1].role != "market" or entries[partitions + 1].source != market_source:
raise RuntimeError("collection manifest market source does not match its requested spec")
if entries[partitions + 2].role != "risk_free" or entries[partitions + 2].source != rf_source:
raise RuntimeError("collection manifest risk-free source does not match its requested spec")
for entry in entries:
if entry.row_count < 0:
raise RuntimeError("collection manifest row counts cannot be negative")
if entry.pit_status != "snapshot":
raise RuntimeError("daily WRDS collection entries must be revised snapshots")
if entry.role == "release_marker":
sql = CRSP_RELEASE_MARKER_SQL
elif entry.role == "stock":
sql = crsp_daily_stock_sql(
start_date=start,
end_date=end,
bucket=cast(int, entry.bucket),
partitions=partitions,
convention=convention,
)
elif entry.role == "market":
sql = crsp_daily_market_sql(start_date=start, end_date=end)
elif entry.role == "risk_free":
sql = ff_daily_risk_free_sql(start_date=start, end_date=end)
else:
raise RuntimeError(f"unknown collection manifest role {entry.role!r}")
if entry.query_hash != _query_hash(sql):
raise RuntimeError("collection manifest query hash does not match its requested spec")
def _validate_monthly_entry_layout(
entries: tuple[MonthlyEquityManifestEntry, ...],
*,
start: dt.date,
end: dt.date,
partitions: int,
) -> None:
bucket_sources, rf_source = _monthly_source_names(
start=start,
end=end,
partitions=partitions,
)
if len(entries) != partitions * 3 + 2:
raise RuntimeError("monthly collection manifest has the wrong number of inputs")
marker = entries[0]
if (
marker.role != "release_marker"
or marker.source != _RELEASE_MARKER_SOURCE
or marker.bucket is not None
or marker.row_count != 1
or marker.min_date is None
or marker.min_date != marker.max_date
or marker.query_hash != _query_hash(CRSP_RELEASE_MARKER_SQL)
):
raise RuntimeError(
"monthly collection manifest release marker does not match its requested spec"
)
expected_roles = ("msf", "msenames", "msedelist")
builders = {
"msf": crsp_monthly_stock_sql,
"msenames": crsp_monthly_names_sql,
"msedelist": crsp_monthly_delist_sql,
}
for bucket in range(partitions):
offset = 1 + bucket * 3
actual = entries[offset : offset + 3]
if tuple(entry.role for entry in actual) != expected_roles:
raise RuntimeError("monthly collection bucket roles are missing or misordered")
if tuple(entry.bucket for entry in actual) != (bucket, bucket, bucket):
raise RuntimeError("monthly collection bucket numbers are missing or misordered")
if tuple(entry.source for entry in actual) != bucket_sources[bucket]:
raise RuntimeError("monthly collection sources do not match its requested spec")
for entry in actual:
sql = builders[entry.role](
start_date=start,
end_date=end,
bucket=bucket,
partitions=partitions,
)
if entry.query_hash != _query_hash(sql):
raise RuntimeError(
"monthly collection query hash does not match its requested spec"
)
rf_entry = entries[-1]
if rf_entry.role != "risk_free" or rf_entry.bucket is not None or rf_entry.source != rf_source:
raise RuntimeError("monthly collection final risk-free input does not match its spec")
rf_sql = ff_monthly_risk_free_sql(start_date=start, end_date=end)
if rf_entry.query_hash != _query_hash(rf_sql):
raise RuntimeError("monthly collection RF query hash does not match its requested spec")
for entry in entries:
if entry.row_count < 0:
raise RuntimeError("monthly collection row counts cannot be negative")
if entry.pit_status != "snapshot":
raise RuntimeError("monthly WRDS collection entries must be revised snapshots")
[docs]
def prepare_crsp_daily_equity(
conn: _wrds.Connection,
*,
start_date: str | dt.date | pd.Timestamp,
end_date: str | dt.date | pd.Timestamp,
vintage: str | None = None,
convention: DailyEquityConvention = "siz",
partitions: int = DEFAULT_DAILY_PARTITIONS,
home: str | Path | None = None,
refresh: bool = False,
) -> DailyEquityCollection:
"""Pull/cache one bucket at a time and publish its complete collection manifest.
A failed run leaves already-published immutable buckets available for a retry but publishes no
collection manifest until every stock bucket and both calendar inputs are complete. Before any
bucket query and again after all pulls, an immutable same-vintage CRSP release marker must agree
with an immediate live marker; drift requires a new vintage. Frames are released before the
next bucket, bounding client memory by one raw partition.
"""
start, end = _date_range(start_date, end_date)
_, partitions = _partition(0, partitions)
convention = _convention(convention)
vintage = _component(vintage or wrds.default_vintage(), label="vintage")
# Reject an unsafe collection namespace before issuing any licensed query.
_manifest_location(
home=home,
convention=convention,
start=start,
end=end,
partitions=partitions,
vintage=vintage,
)
stock_sources, market_source, rf_source = _source_names(
convention=convention,
start=start,
end=end,
partitions=partitions,
)
marker_frame, marker_meta, cached_marker = _cached_release_marker(
conn,
vintage=vintage,
home=home,
)
entries: list[DailyEquityManifestEntry] = [
_entry(
role="release_marker",
source=_RELEASE_MARKER_SOURCE,
meta=marker_meta,
frame=marker_frame,
)
]
del marker_frame
_require_unchanged_release_marker(
cached_marker,
_live_release_marker(conn),
vintage=vintage,
stage="before the partitioned pull",
)
for bucket, source in enumerate(stock_sources):
sql = crsp_daily_stock_sql(
start_date=start,
end_date=end,
bucket=bucket,
partitions=partitions,
convention=convention,
)
frame = wrds.pull_raw(
conn,
source,
sql,
vintage,
home=home,
pit_status="snapshot",
refresh=refresh,
)
meta = read_meta(source, vintage, home=home)
entries.append(_entry(role="stock", source=source, meta=meta, frame=frame, bucket=bucket))
del frame
for role, source, sql in (
(
"market",
market_source,
crsp_daily_market_sql(start_date=start, end_date=end),
),
("risk_free", rf_source, ff_daily_risk_free_sql(start_date=start, end_date=end)),
):
frame = wrds.pull_raw(
conn,
source,
sql,
vintage,
home=home,
pit_status="snapshot",
refresh=refresh,
)
meta = read_meta(source, vintage, home=home)
entries.append(_entry(role=role, source=source, meta=meta, frame=frame))
del frame
_require_unchanged_release_marker(
cached_marker,
_live_release_marker(conn),
vintage=vintage,
stage="during the partitioned pull",
)
materialized_entries = tuple(entries)
_validate_entry_layout(
materialized_entries,
convention=convention,
start=start,
end=end,
partitions=partitions,
)
return _build_collection(
convention=convention,
start=start,
end=end,
partitions=partitions,
vintage=vintage,
home=home,
entries=materialized_entries,
publish=True,
)
[docs]
def load_crsp_daily_equity(
*,
start_date: str | dt.date | pd.Timestamp,
end_date: str | dt.date | pd.Timestamp,
vintage: str,
convention: DailyEquityConvention = "siz",
partitions: int = DEFAULT_DAILY_PARTITIONS,
home: str | Path | None = None,
) -> DailyEquityCollection:
"""Load a completed collection manifest without opening a WRDS connection.
The small release-marker cache is revalidated immediately. Other sidecars and parquet contents
are revalidated lazily, one input at a time, when the collection iterator/read methods are used.
"""
start, end = _date_range(start_date, end_date)
_, partitions = _partition(0, partitions)
convention = _convention(convention)
vintage = _component(vintage, label="vintage")
path = _manifest_location(
home=home,
convention=convention,
start=start,
end=end,
partitions=partitions,
vintage=vintage,
)
payload = _parse_manifest(path)
raw_entries = payload.get("entries")
if not isinstance(raw_entries, list):
raise RuntimeError("collection manifest entries must be a list")
try:
entries = tuple(DailyEquityManifestEntry(**entry) for entry in raw_entries)
except (TypeError, ValueError) as exc:
raise RuntimeError("collection manifest contains invalid entries") from exc
_validate_entry_layout(
entries,
convention=convention,
start=start,
end=end,
partitions=partitions,
)
expected = _build_collection(
convention=convention,
start=start,
end=end,
partitions=partitions,
vintage=vintage,
home=home,
entries=entries,
publish=False,
)
expected_payload = _manifest_payload(
convention=convention,
start=start,
end=end,
partitions=partitions,
vintage=vintage,
entries=entries,
)
digest = _manifest_hash(expected_payload)
expected_payload["manifest_hash"] = digest
expected_payload["data_vintage"] = (
f"{_COLLECTION_KIND}_{convention}@{digest.removeprefix('sha256:')[:_DATA_VINTAGE_HASH_LEN]}"
)
if payload != expected_payload:
raise RuntimeError("collection manifest identity/hash does not match the requested spec")
expected._validate_release_marker_cache()
return expected
[docs]
def prepare_crsp_monthly_equity_targets(
conn: _wrds.Connection,
*,
start_date: str | dt.date | pd.Timestamp,
end_date: str | dt.date | pd.Timestamp,
vintage: str | None = None,
partitions: int = DEFAULT_DAILY_PARTITIONS,
home: str | Path | None = None,
refresh: bool = False,
) -> MonthlyEquityTargetCollection:
"""Pull a complete SIZ monthly target collection, bounded by one bucket in memory.
Each bucket has three independent immutable raw inputs—MSF, overlapping name histories, and
delisting events—so a terminal event with no MSF row is still available to the outer-join
cleaner. A single small monthly FF RF input follows all bucket roles. A failed run leaves only
reusable raw caches; the canonical collection manifest appears after all ``3*N+2`` inputs,
including the leading CRSP release marker, pass. Marker drift before or during a pull requires
a new vintage.
"""
start, end = _month_range(start_date, end_date)
_, partitions = _partition(0, partitions)
vintage = _component(vintage or wrds.default_vintage(), label="vintage")
# Fail before any licensed query if the collection namespace itself is unsafe.
_monthly_manifest_location(
home=home,
start=start,
end=end,
partitions=partitions,
vintage=vintage,
)
bucket_sources, rf_source = _monthly_source_names(
start=start,
end=end,
partitions=partitions,
)
builders = {
"msf": crsp_monthly_stock_sql,
"msenames": crsp_monthly_names_sql,
"msedelist": crsp_monthly_delist_sql,
}
marker_frame, marker_meta, cached_marker = _cached_release_marker(
conn,
vintage=vintage,
home=home,
)
entries: list[MonthlyEquityManifestEntry] = [
_monthly_entry(
role="release_marker",
source=_RELEASE_MARKER_SOURCE,
meta=marker_meta,
frame=marker_frame,
)
]
del marker_frame
_require_unchanged_release_marker(
cached_marker,
_live_release_marker(conn),
vintage=vintage,
stage="before the partitioned pull",
)
for bucket, sources in enumerate(bucket_sources):
for role, source in zip(("msf", "msenames", "msedelist"), sources, strict=True):
sql = builders[role](
start_date=start,
end_date=end,
bucket=bucket,
partitions=partitions,
)
frame = wrds.pull_raw(
conn,
source,
sql,
vintage,
home=home,
pit_status="snapshot",
refresh=refresh,
)
meta = read_meta(source, vintage, home=home)
entries.append(
_monthly_entry(
role=role,
source=source,
meta=meta,
frame=frame,
bucket=bucket,
)
)
del frame
rf_sql = ff_monthly_risk_free_sql(start_date=start, end_date=end)
frame = wrds.pull_raw(
conn,
rf_source,
rf_sql,
vintage,
home=home,
pit_status="snapshot",
refresh=refresh,
)
meta = read_meta(rf_source, vintage, home=home)
entries.append(
_monthly_entry(
role="risk_free",
source=rf_source,
meta=meta,
frame=frame,
)
)
del frame
_require_unchanged_release_marker(
cached_marker,
_live_release_marker(conn),
vintage=vintage,
stage="during the partitioned pull",
)
materialized_entries = tuple(entries)
_validate_monthly_entry_layout(
materialized_entries,
start=start,
end=end,
partitions=partitions,
)
return _build_monthly_collection(
start=start,
end=end,
partitions=partitions,
vintage=vintage,
home=home,
entries=materialized_entries,
publish=True,
)
[docs]
def load_crsp_monthly_equity_targets(
*,
start_date: str | dt.date | pd.Timestamp,
end_date: str | dt.date | pd.Timestamp,
vintage: str,
partitions: int = DEFAULT_DAILY_PARTITIONS,
home: str | Path | None = None,
) -> MonthlyEquityTargetCollection:
"""Load a complete monthly SIZ target manifest without opening a WRDS connection."""
start, end = _month_range(start_date, end_date)
_, partitions = _partition(0, partitions)
vintage = _component(vintage, label="vintage")
path = _monthly_manifest_location(
home=home,
start=start,
end=end,
partitions=partitions,
vintage=vintage,
)
payload = _parse_manifest(path)
raw_entries = payload.get("entries")
if not isinstance(raw_entries, list):
raise RuntimeError("monthly collection manifest entries must be a list")
try:
entries = tuple(MonthlyEquityManifestEntry(**entry) for entry in raw_entries)
except (TypeError, ValueError) as exc:
raise RuntimeError("monthly collection manifest contains invalid entries") from exc
_validate_monthly_entry_layout(
entries,
start=start,
end=end,
partitions=partitions,
)
expected = _build_monthly_collection(
start=start,
end=end,
partitions=partitions,
vintage=vintage,
home=home,
entries=entries,
publish=False,
)
expected_payload = _monthly_manifest_payload(
start=start,
end=end,
partitions=partitions,
vintage=vintage,
entries=entries,
)
digest = _manifest_hash(expected_payload)
expected_payload["manifest_hash"] = digest
expected_payload["data_vintage"] = (
f"{_MONTHLY_COLLECTION_KIND}@{digest.removeprefix('sha256:')[:_DATA_VINTAGE_HASH_LEN]}"
)
if payload != expected_payload:
raise RuntimeError("monthly collection manifest identity/hash does not match its spec")
expected._validate_release_marker_cache()
return expected