"""Live WRDS pulls into the raw zone, then the clean builders and provenance lock.
This is the seam used by credentialed reproductions. It:
1. connects to WRDS with credentials taken from the **environment by default**, or from explicit
in-process arguments (never written to disk or committed) — the connector's own ``.pgpass`` is
neither required nor created here;
2. pulls each raw table with a ``SELECT`` and caches it as parquet under
``raw/<source>/<vintage>/`` with a ``_meta.json`` sidecar, so the *licensed bytes live only in
the outside-repo data home* (``paths.data_home``), never in any git history; parquet and sidecar
are staged and atomically published as one complete immutable directory;
3. runs the deterministic clean-zone builders (:mod:`numeraire_dataset.zones.clean`) over the raw
frames and records every artifact — raw vintage and built clean table — in a
:class:`~numeraire_dataset.zones.lock.DataLock`, from which a numeraire result's ``data_vintage``
stamp is derived.
The WRDS schema (``crspm`` monthly-cadence tables, preferred over ``crsp``) is part of the raw
pull's identity: it appears in the ``SELECT`` and therefore in the ``query_hash`` recorded in each
raw ``_meta.json``. Heavy imports (``wrds``) are lazy so the module is importable without the
``[wrds]`` extra; only :func:`connect` imports it. Pull functions accept an already-open connection.
"""
from __future__ import annotations
import datetime as dt
import hashlib
import json
import os
import stat
import tempfile
import time
from collections.abc import Iterator
from contextlib import contextmanager, suppress
from pathlib import Path
from typing import TYPE_CHECKING, Any
import pandas as pd
from numeraire_dataset.zones.lock import DataLock
from numeraire_dataset.zones.raw import (
RawMeta,
_raw_path,
_validate_cache_file,
_write_meta_at,
read_meta,
)
from numeraire_dataset.zones.steps import (
CONTENT_DIGEST_VERSION,
content_digest,
run_step,
)
if TYPE_CHECKING: # pragma: no cover - typing only
# Optional [wrds] extra — not installed in the default/dev env. The lazy runtime import in
# connect() carries the same guard; this one only feeds the return/param annotations below.
import wrds as _wrds # pyright: ignore[reportMissingImports]
_PARQUET = "data.parquet"
_META = "_meta.json"
_RAW_INPUT_IDENTITY_VERSION = 1
def _raw_input_identity(meta: RawMeta) -> str:
"""Canonical raw identity chained into a clean recipe hash.
Raw bytes alone are not sufficient provenance: whether those bytes are an as-published vintage
or a latest revised snapshot changes their methodological admissibility. The digest contract
version is equally part of the interpretation of ``content_digest``. A versioned canonical
JSON payload keeps all fields role-bound and unambiguous without exposing SQL or credentials.
"""
payload = {
"content_digest": meta.content_digest,
"content_digest_version": meta.content_digest_version,
"pit_status": meta.pit_status,
"query_hash": meta.query_hash,
"source": meta.source,
"vintage": meta.vintage,
}
canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
return f"raw-input-v{_RAW_INPUT_IDENTITY_VERSION}:{canonical}"
[docs]
def default_vintage() -> str:
"""The default raw-vintage stamp for a WRDS pull, resolved **at call time**.
Uses the explicit ``WRDS_VINTAGE`` environment variable when set; otherwise the current
year-month (UTC), e.g. ``"2026-07"``. Resolving live avoids a hard-coded, silently-stale label
that would cache/lock a fresh pull under a wrong vintage. Pass an explicit ``vintage=`` to pin a
pull to a specific release date (a dated point-in-time vintage).
"""
return os.environ.get("WRDS_VINTAGE") or dt.datetime.now(dt.UTC).strftime("%Y-%m")
# Standard raw pulls, keyed by raw ``source`` name. crspm (monthly cadence) is preferred over crsp.
# The schema is baked into the SQL, so it participates in the pull's query_hash identity.
STANDARD_QUERIES: dict[str, str] = {
"crspm_msf": (
"select permno, permco, date, ret, prc, shrout from crspm.msf where date >= '1955-01-01'"
),
"crspm_msenames": (
"select permno, permco, namedt, nameendt, shrcd, exchcd, siccd, ticker from crspm.msenames"
),
"crspm_msedelist": (
"select permno, dlstdt as date, dlret, dlstcd from crspm.msedelist "
"where dlstdt >= '1955-01-01'"
),
"comp_funda": (
"select gvkey, datadate, seq, ceq, pstk, at, lt, txditc, pstkrv, pstkl, "
"revt, cogs, xsga, xint, ib, fyear, fyr, sich as sic, gp from comp.funda "
"where indfmt='INDL' and datafmt='STD' and popsrc='D' and consol='C' "
"and datadate >= '1954-01-01'"
),
"comp_fundq": (
"select gvkey, datadate, rdq, ibq, seqq, ceqq, pstkq, atq, ltq, txditcq, pstkrq "
"from comp.fundq "
"where indfmt='INDL' and datafmt='STD' and popsrc='D' and consol='C' "
"and datadate >= '1954-01-01'"
),
"crsp_ccmlink": (
"select gvkey, lpermno, linktype, linkprim, linkdt, linkenddt from crsp.ccmxpf_lnkhist"
),
"ff_factors_monthly": ("select date, mktrf, smb, hml, rf, umd from ff.factors_monthly"),
}
# Which raw sources feed each clean step (input-name -> raw source), so the lock chains correctly.
_CLEAN_INPUTS: dict[str, dict[str, str]] = {
"crsp_monthly_clean": {
"msf": "crspm_msf",
"msenames": "crspm_msenames",
"msedelist": "crspm_msedelist",
},
"compustat_annual_clean": {"funda": "comp_funda"},
"compustat_annual_formation_clean": {"funda": "comp_funda"},
"compustat_quarterly_clean": {"fundq": "comp_fundq"},
"ccm_links_clean": {"ccmxpf_linkhist": "crsp_ccmlink"},
}
[docs]
def wrds_available() -> bool:
"""Whether WRDS credentials are reachable in the environment (no connection is attempted)."""
return bool(os.environ.get("WRDS_USERNAME") or os.environ.get("WRDS_USER")) and bool(
os.environ.get("WRDS_PASSWORD")
)
[docs]
def connect(
*, username: str | None = None, password: str | None = None, **kwargs: Any
) -> _wrds.Connection:
"""Open a WRDS connection using credentials from the environment (or explicit arguments).
Reads ``WRDS_USERNAME`` (or ``WRDS_USER``) and ``WRDS_PASSWORD``. Credentials are passed
straight to the connector for this process only — nothing is written to disk and no ``.pgpass``
is created. Extra ``kwargs`` pass through to :class:`wrds.Connection`.
"""
import wrds # pyright: ignore[reportMissingImports] # lazy: only the [wrds] extra provides it
user = username or os.environ.get("WRDS_USERNAME") or os.environ.get("WRDS_USER")
if not user:
raise RuntimeError("set WRDS_USERNAME (or WRDS_USER) in the environment for a WRDS pull")
pw = password or os.environ.get("WRDS_PASSWORD")
conn_kwargs: dict[str, Any] = {"wrds_username": user}
if pw:
conn_kwargs["wrds_password"] = pw
conn_kwargs.update(kwargs)
return wrds.Connection(**conn_kwargs)
def _query_hash(sql: str) -> str:
"""Hash the exact SQL text sent to WRDS as the raw pull's transform identity.
SQL-aware canonicalization would require a real parser: generic whitespace folding corrupts
string literals (``'A B'`` and ``'A B'`` can select different rows). Exact bytes fail safely
on cosmetically reformatted queries instead of ever colliding semantically distinct queries.
"""
return "sha256:" + hashlib.sha256(sql.encode("utf-8")).hexdigest()
def _is_sha256_digest(value: object) -> bool:
if not isinstance(value, str) or not value.startswith("sha256:"):
return False
digest = value.removeprefix("sha256:")
return len(digest) == 64 and all(char in "0123456789abcdef" for char in digest)
def _validate_cached_query(
source: str,
vintage: str,
sql: str,
*,
home: str | Path | None = None,
) -> RawMeta:
"""Fail closed when a same-label raw cache was produced by different SQL.
Cache paths are keyed by ``(source, vintage)`` for human readability, while the sidecar pins
the exact query. A standard query can gain fields without changing either label, so accepting a
stale parquet solely because the path exists would bypass the new schema and corrupt the clean
recipe's claimed inputs. The caller must choose a new vintage instead.
"""
expected = _query_hash(sql)
try:
meta = read_meta(source, vintage, home=home)
except FileNotFoundError as exc:
raise RuntimeError(
f"cached raw {source}@{vintage} query hash mismatch "
f"(expected {expected}, observed <missing>); "
"choose a new vintage rather than overwriting an immutable raw cache"
) from exc
except (OSError, TypeError, ValueError) as exc:
raise RuntimeError(
f"cached raw {source}@{vintage} metadata is invalid; "
"choose a new vintage rather than overwriting an immutable raw cache"
) from exc
if meta.source != source or meta.vintage != vintage:
raise RuntimeError(
f"cached raw {source}@{vintage} metadata identity mismatch; "
"choose a new vintage rather than overwriting an immutable raw cache"
)
actual: object = meta.query_hash
if actual != expected:
observed = actual if _is_sha256_digest(actual) else "<invalid>"
raise RuntimeError(
f"cached raw {source}@{vintage} query hash mismatch "
f"(expected {expected}, observed {observed}); "
"choose a new vintage rather than overwriting an immutable raw cache"
)
return meta
def _validated_cached_frame(
source: str,
vintage: str,
*,
home: str | Path | None = None,
sql: str | None = None,
) -> tuple[pd.DataFrame, RawMeta, str]:
"""Read one complete cache and verify identity, query, rows, schema, and values once."""
directory = _raw_path(source, vintage, home=home)
parquet = directory / _PARQUET
meta_path = directory / _META
try:
_validate_cache_file(parquet, label="raw parquet")
_validate_cache_file(meta_path, label="raw metadata sidecar")
except FileNotFoundError as exc:
raise FileNotFoundError(
f"no complete cached raw pull at {directory} — run pull_raw first"
) from exc
if not parquet.exists() or not meta_path.exists(): # race-safe recheck before readers open them
raise FileNotFoundError(f"no complete cached raw pull at {directory} — run pull_raw first")
if sql is None:
try:
meta = read_meta(source, vintage, home=home)
except (OSError, TypeError, ValueError) as exc:
raise RuntimeError(f"cached raw {source}@{vintage} metadata is invalid") from exc
if meta.source != source or meta.vintage != vintage:
raise RuntimeError(f"cached raw {source}@{vintage} metadata identity mismatch")
if not _is_sha256_digest(meta.query_hash):
raise RuntimeError(f"cached raw {source}@{vintage} metadata query hash is invalid")
else:
meta = _validate_cached_query(source, vintage, sql, home=home)
if meta.content_digest_version != CONTENT_DIGEST_VERSION:
raise RuntimeError(
f"cached raw {source}@{vintage} uses unsupported content-digest contract "
f"v{meta.content_digest_version}; choose a new vintage"
)
try:
frame = pd.read_parquet(parquet)
except (OSError, TypeError, ValueError) as exc:
raise RuntimeError(
f"cached raw {source}@{vintage} parquet is unreadable; "
"choose a new vintage rather than trusting an incomplete cache"
) from exc
digest = content_digest(frame)
if meta.content_digest != digest or meta.row_count != len(frame):
raise RuntimeError(
f"cached raw {source}@{vintage} content metadata does not match its parquet; "
"choose a new vintage rather than trusting a mutated or incomplete cache"
)
return frame, meta, digest
def _discard_incomplete_cache(directory: Path) -> None:
"""Remove only the two known files from an unpublished legacy/incomplete cache directory."""
if directory.is_symlink():
raise RuntimeError(f"incomplete raw cache at {directory} is a symlink; inspect it manually")
if not directory.exists():
return
children = tuple(directory.iterdir())
if any(child.is_symlink() for child in children):
raise RuntimeError(
f"incomplete raw cache at {directory} contains a symlink; inspect it manually"
)
unexpected = [child.name for child in children if child.name not in {_PARQUET, _META}]
if unexpected or any(child.is_dir() for child in children):
raise RuntimeError(
f"incomplete raw cache at {directory} contains unexpected files; inspect it manually"
)
for child in children:
child.unlink()
directory.rmdir()
def _cleanup_staging(directory: Path, *, root: Path | None = None) -> None:
"""Best-effort cleanup for this function's exact, unpublished staging directory."""
root = directory.parents[2] if root is None else root
_validate_internal_path(directory, root, label="raw staging directory")
if not directory.exists():
return
for name in (_PARQUET, _META):
(directory / name).unlink(missing_ok=True)
# An unexpected writer-created file is safer left for inspection than recursively removed.
with suppress(OSError):
directory.rmdir()
def _vintage_token(vintage: str) -> str:
"""Return a fixed-width token so staging/lock names are prefix-unambiguous."""
return hashlib.sha256(vintage.encode("utf-8")).hexdigest()
def _staging_prefix(vintage: str) -> str:
return f"{_vintage_token(vintage)}-"
def _source_token(source: str) -> str:
return hashlib.sha256(source.encode("utf-8")).hexdigest()
def _cache_internal_root(directory: Path) -> Path:
"""Return data_home for ``data_home/raw/source/vintage`` without following user labels."""
return directory.parents[2]
def _validate_internal_path(path: Path, root: Path, *, label: str) -> Path:
"""Reject symlinks and containment escapes in a cache-internal namespace path."""
root = root.resolve()
try:
relative = path.relative_to(root)
except ValueError as exc:
raise RuntimeError(f"{label} is outside the configured data home") from exc
current = root
for part in relative.parts:
current = current / part
if current.is_symlink():
raise RuntimeError(f"{label} contains a symlink at {current}; inspect it manually")
try:
resolved = path.resolve(strict=False)
except OSError as exc:
raise RuntimeError(f"{label} cannot be resolved safely") from exc
if not resolved.is_relative_to(root):
raise RuntimeError(f"{label} resolves outside the configured data home")
return path
def _ensure_internal_directory(path: Path, root: Path, *, label: str) -> Path:
"""Create an internal directory only after and before containment/symlink checks."""
_validate_internal_path(path, root, label=label)
path.mkdir(parents=True, exist_ok=True)
_validate_internal_path(path, root, label=label)
if not path.is_dir():
raise RuntimeError(f"{label} is not a directory")
return path
def _staging_parent(directory: Path, source: str) -> Path:
root = _cache_internal_root(directory)
path = root / ".raw-staging" / _source_token(source)
return _validate_internal_path(path, root, label="raw staging namespace")
def _writer_lock_path(directory: Path, source: str, vintage: str) -> Path:
root = _cache_internal_root(directory)
path = root / ".raw-locks" / _source_token(source) / f"{_vintage_token(vintage)}.lock"
return _validate_internal_path(path, root, label="raw writer-lock namespace")
def _cleanup_stale_staging(parent: Path, vintage: str) -> None:
"""Remove abandoned staging data while holding the vintage's exclusive writer lock."""
root = parent.parents[1]
_validate_internal_path(parent, root, label="raw staging namespace")
prefix = _staging_prefix(vintage)
for directory in (child for child in parent.iterdir() if child.name.startswith(prefix)):
if not directory.is_dir() or directory.is_symlink():
raise RuntimeError(f"unexpected raw staging artifact at {directory}; inspect manually")
_cleanup_staging(directory)
if directory.exists():
raise RuntimeError(f"raw staging directory at {directory} could not be safely cleaned")
@contextmanager
def _cache_writer_lock(path: Path) -> Iterator[None]:
"""Serialize first publication of one vintage; OS locks are released after crashes."""
root = path.parents[2]
_ensure_internal_directory(path.parent, root, label="raw writer-lock namespace")
_validate_internal_path(path, root, label="raw writer-lock file")
flags = os.O_RDWR | os.O_CREAT | os.O_APPEND | getattr(os, "O_NOFOLLOW", 0)
if os.name == "nt": # pragma: no cover - platform flag
flags |= getattr(os, "O_BINARY", 0)
descriptor = os.open(path, flags, 0o600)
with os.fdopen(descriptor, "a+b") as stream:
if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode):
raise RuntimeError(f"raw writer-lock file at {path} is not a regular file")
_validate_internal_path(path, root, label="raw writer-lock file")
if os.name == "nt": # pragma: no cover - exercised on Windows CI/users
import msvcrt
stream.seek(0, os.SEEK_END)
if stream.tell() == 0:
stream.write(b"\0")
stream.flush()
stream.seek(0)
while True:
try:
msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1)
break
except OSError:
time.sleep(0.05)
try:
yield
finally:
stream.seek(0)
msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1)
else:
import fcntl
fcntl.flock(stream.fileno(), fcntl.LOCK_EX)
try:
yield
finally:
fcntl.flock(stream.fileno(), fcntl.LOCK_UN)
[docs]
def raw_cached(source: str, vintage: str, *, home: str | Path | None = None) -> bool:
"""Whether a complete parquet + metadata cache exists for ``(source, vintage)``."""
directory = _raw_path(source, vintage, home=home)
try:
_validate_cache_file(directory / _PARQUET, label="raw parquet")
_validate_cache_file(directory / _META, label="raw metadata sidecar")
except (FileNotFoundError, RuntimeError):
return False
return True
[docs]
def read_raw(source: str, vintage: str, *, home: str | Path | None = None) -> pd.DataFrame:
"""Read a cached raw frame after validating its metadata, schema, and values."""
frame, _, _ = _validated_cached_frame(source, vintage, home=home)
return frame
def _pull_raw_record(
conn: _wrds.Connection,
source: str,
sql: str,
vintage: str,
*,
home: str | Path | None,
pit_status: str,
refresh: bool,
) -> tuple[pd.DataFrame, RawMeta, str]:
"""Implementation returning the validated frame and provenance without hashing twice."""
directory = _raw_path(source, vintage, home=home)
internal_root = _cache_internal_root(directory)
parquet = directory / _PARQUET
meta_path = directory / _META
if parquet.exists() and meta_path.exists():
if refresh:
raise RuntimeError(
f"raw cache {source}@{vintage} is immutable and cannot be refreshed in place; "
"choose a new vintage"
)
return _validated_cached_frame(source, vintage, home=home, sql=sql)
staging_parent = _staging_parent(directory, source)
lock_path = _writer_lock_path(directory, source, vintage)
with _cache_writer_lock(lock_path):
directory.parent.mkdir(parents=True, exist_ok=True)
# Re-run containment after creating the parent to catch a concurrent symlink substitution.
_raw_path(source, vintage, home=home)
_ensure_internal_directory(
staging_parent,
internal_root,
label="raw staging namespace",
)
if directory.parent.stat().st_dev != staging_parent.stat().st_dev:
raise RuntimeError(
"raw cache and staging namespaces must share a filesystem for atomic publication"
)
# Recheck after waiting: exactly one process performs the licensed query and publication;
# all followers validate and adopt its result.
if parquet.exists() and meta_path.exists():
if refresh:
raise RuntimeError(
f"raw cache {source}@{vintage} is immutable and cannot be refreshed in place; "
"choose a new vintage"
)
_cleanup_stale_staging(staging_parent, vintage)
return _validated_cached_frame(source, vintage, home=home, sql=sql)
if directory.exists():
# A directory missing either committed file was never a complete immutable cache. It
# is safe to discard only the two known artifacts and retry the exact same vintage.
_discard_incomplete_cache(directory)
_cleanup_stale_staging(staging_parent, vintage)
staging = Path(tempfile.mkdtemp(prefix=_staging_prefix(vintage), dir=staging_parent))
_validate_internal_path(staging, internal_root, label="raw staging directory")
try:
frame = conn.raw_sql(sql)
staged_parquet = staging / _PARQUET
frame.to_parquet(staged_parquet)
# WRDS-scale frames can exceed local memory. Release the connector representation
# before reading back the persisted representation used for digest/return.
del frame
persisted = pd.read_parquet(staged_parquet)
digest = content_digest(persisted)
meta = RawMeta(
source=source,
vintage=vintage,
pulled_at=dt.datetime.now(dt.UTC).isoformat(timespec="seconds"),
content_digest=digest,
row_count=len(persisted),
query_hash=_query_hash(sql),
pit_status=pit_status,
content_digest_version=CONTENT_DIGEST_VERSION,
)
_write_meta_at(meta, staging)
if directory.exists():
# A non-cooperating older writer may still win. Release the staging frame before
# reading its winner so this fallback cannot double peak memory.
del persisted
return _validated_cached_frame(source, vintage, home=home, sql=sql)
try:
_validate_internal_path(staging, internal_root, label="raw staging directory")
_raw_path(source, vintage, home=home)
staging.rename(directory)
except OSError:
if directory.exists():
del persisted
return _validated_cached_frame(source, vintage, home=home, sql=sql)
raise
return persisted, meta, digest
finally:
_cleanup_staging(staging, root=internal_root)
[docs]
def pull_raw(
conn: _wrds.Connection,
source: str,
sql: str,
vintage: str | None = None,
*,
home: str | Path | None = None,
pit_status: str = "snapshot",
refresh: bool = False,
) -> pd.DataFrame:
"""Pull ``sql`` into a frame and cache it as parquet + ``_meta.json`` under the raw zone.
Returns the cached frame without hitting WRDS when it already exists. Raw vintages are
immutable: ``refresh=True`` may populate a new vintage, but it never overwrites an existing
``(source, vintage)`` cache. The licensed bytes stay in the outside-repo data home.
``pit_status`` defaults to ``"snapshot"`` — a latest, revised WRDS pull is not itself
point-in-time. ``vintage`` defaults to :func:`default_vintage` (``WRDS_VINTAGE`` or the current
year-month).
"""
vintage = vintage or default_vintage()
frame, _, _ = _pull_raw_record(
conn,
source,
sql,
vintage,
home=home,
pit_status=pit_status,
refresh=refresh,
)
return frame
[docs]
def pull_standard(
conn: _wrds.Connection,
*,
vintage: str | None = None,
home: str | Path | None = None,
sources: tuple[str, ...] | None = None,
refresh: bool = False,
) -> dict[str, pd.DataFrame]:
"""Pull (and cache) the standard raw tables; returns ``{source: frame}``."""
vintage = vintage or default_vintage()
names = sources if sources is not None else tuple(STANDARD_QUERIES)
out: dict[str, pd.DataFrame] = {}
for source in names:
out[source] = pull_raw(
conn, source, STANDARD_QUERIES[source], vintage, home=home, refresh=refresh
)
return out
[docs]
def load_clean(
conn: _wrds.Connection | None = None,
*,
vintage: str | None = None,
home: str | Path | None = None,
lock: DataLock | None = None,
write_lock: bool = True,
refresh: bool = False,
) -> tuple[dict[str, pd.DataFrame], DataLock]:
"""Pull the standard raw tables, run the clean builders, and record the provenance lock.
Returns ``(frames, lock)`` where ``frames`` maps a friendly name to a frame:
``crsp`` / ``compustat`` / ``compustat_formation`` / ``compustat_q`` / ``ccm`` (built clean
tables — ``compustat_formation`` carries the following-June annual formation convention and
``compustat_q`` is the quarterly fresh-earnings ROE table), ``ff`` (raw Fama-French factors),
plus the raw
``funda`` and ``msenames`` (needed for characteristics the clean tables drop, e.g. Compustat
``ib`` or the CRSP ``ticker`` map). The returned :class:`DataLock` carries a
``data_vintage`` stamp for each clean table; it is written to ``data.lock.json`` in the data
home when ``write_lock`` (the numeraire-facing provenance record).
A live ``conn`` is required unless every raw table is already cached (then pass ``conn=None``).
"""
vintage = vintage or default_vintage()
lock = lock if lock is not None else DataLock()
raw: dict[str, pd.DataFrame] = {}
raw_meta: dict[str, RawMeta] = {}
for source, sql in STANDARD_QUERIES.items():
if conn is None:
if not raw_cached(source, vintage, home=home):
raise RuntimeError(f"raw {source}@{vintage} not cached and no live conn given")
frame, meta, digest = _validated_cached_frame(source, vintage, home=home, sql=sql)
else:
frame, meta, digest = _pull_raw_record(
conn,
source,
sql,
vintage,
home=home,
pit_status="snapshot",
refresh=refresh,
)
raw[source] = frame
raw_meta[source] = meta
lock.add_raw(
source,
vintage,
digest,
query_hash=meta.query_hash,
pit_status=meta.pit_status,
content_digest_version=meta.content_digest_version,
)
frames: dict[str, pd.DataFrame] = {}
friendly = {
"crsp_monthly_clean": "crsp",
"compustat_annual_clean": "compustat",
"compustat_annual_formation_clean": "compustat_formation",
"compustat_quarterly_clean": "compustat_q",
"ccm_links_clean": "ccm",
}
for step_name, inputs in _CLEAN_INPUTS.items():
step_frames = {input_name: raw[src] for input_name, src in inputs.items()}
# Chain the complete methodological identity into the clean recipe hash: source/vintage,
# semantic bytes + digest contract, exact-query hash, and snapshot/vintage PIT status. The
# human-readable source@vintage keys are still recorded by role in the clean lock entry.
input_hashes = {
input_name: _raw_input_identity(raw_meta[src]) for input_name, src in inputs.items()
}
input_labels = {input_name: f"{src}@{vintage}" for input_name, src in inputs.items()}
built = run_step(step_name, step_frames, input_hashes=input_hashes)
lock.add_clean(built, inputs=input_labels)
frames[friendly[step_name]] = built.frame
frames["ff"] = raw["ff_factors_monthly"]
frames["funda"] = raw["comp_funda"]
frames["msenames"] = raw["crspm_msenames"]
if write_lock:
from numeraire_dataset.paths import data_home
lock.write(data_home(home))
return frames, lock