"""Clean-zone builders: CRSP monthly, Compustat annual/formation, and CCM links.
Each builder is a pure ``frame(s) -> frame`` step following the tidy-finance / CIZ (CRSP-Compustat
merged, "Center for Research in Security Prices" conventions) recipe, with every filter/threshold an
**explicit parameter** so it participates in the recipe hash (preprocessing is pinned like the
method). These transform *already-pulled* raw frames, so they are testable on synthetic frames with
the WRDS schema — no live credentials needed. Live validation uses credentialed reference checks.
"""
from __future__ import annotations
from typing import Literal
import numpy as np
import pandas as pd
from numeraire_dataset.zones.steps import step
# CRSP delisting codes with a known performance-related delisting return convention (CIZ / Shumway):
# a missing delisting return for these codes is replaced by -30% (partial recovery assumption).
_PERF_DELIST = (500, *range(520, 585))
_PERF_DELIST_FILL = -0.30
def _daily_dates(frame: pd.DataFrame, *, column: str, context: str) -> pd.Series:
if column not in frame:
raise ValueError(f"{context} has no {column!r} column")
try:
dates = pd.to_datetime(frame[column], errors="raise")
except (TypeError, ValueError) as exc:
raise ValueError(f"{context} contains an invalid date") from exc
if dates.isna().any():
raise ValueError(f"{context} contains NaT")
if getattr(dates.dt, "tz", None) is not None:
raise ValueError(f"{context} dates must be timezone-naive")
if not dates.eq(dates.dt.normalize()).all():
raise ValueError(f"{context} dates must not contain a time of day")
return dates
def _numeric_daily_return(
frame: pd.DataFrame,
*,
column: str,
context: str,
allow_missing: bool,
) -> pd.Series:
if column not in frame:
raise ValueError(f"{context} has no {column!r} column")
try:
values = pd.to_numeric(frame[column], errors="raise").astype("float64")
except (TypeError, ValueError) as exc:
raise TypeError(f"{context} {column!r} values must be numeric") from exc
array = values.to_numpy(dtype=np.float64)
if np.isinf(array).any():
raise ValueError(f"{context} {column!r} contains an infinite value")
if not allow_missing and np.isnan(array).any():
raise ValueError(f"{context} {column!r} contains a missing value")
return values
def _safe_log1p(values: pd.Series) -> pd.Series:
"""Log simple returns without warning or erasing a legitimate source-space ``-1``."""
eligible = values.where(values > -1.0)
out = pd.Series(np.nan, index=values.index, dtype="float64")
valid = eligible.notna()
out.loc[valid] = np.log1p(eligible.loc[valid].to_numpy(dtype=np.float64))
return out
[docs]
@step(
name="crsp_daily_equity_clean",
version=1,
inputs=["daily"],
output="crsp_daily_equity_clean",
)
def crsp_daily_equity_clean(
daily: pd.DataFrame,
*,
convention: Literal["siz", "ciz"] = "siz",
) -> pd.DataFrame:
"""Validate one CRSP daily-equity bucket without imputing an observation.
Legacy/SIZ inputs must already be filtered to share codes 10/11; CIZ inputs carry the official
return/delisting diagnostic flags. ``ret == -1`` is a valid total return and is retained in
``ret``. Its source-space logarithm is undefined and therefore remains missing in
``source_log_return``. The paper's beta uses *excess* log returns; obtain those through
:func:`daily_stock_excess_clean` after the daily risk-free calendar has been validated.
"""
if convention not in {"siz", "ciz"}:
raise ValueError("convention must be 'siz' or 'ciz'")
required = {"permno", "date", "ret"}
required.update(
{"shrcd", "exchcd"}
if convention == "siz"
else {
"dlydelflg",
"dlyretmissflg",
"dlyretdurflg",
"primaryexch",
"conditionaltype",
"tradingstatusflg",
}
)
missing = sorted(required.difference(daily.columns))
if missing:
raise ValueError(f"daily CRSP {convention.upper()} input lacks columns: {missing}")
if not daily.columns.is_unique:
raise ValueError("daily CRSP input has duplicate columns")
out = daily.copy()
out["date"] = _daily_dates(out, column="date", context="daily CRSP input")
permno = pd.to_numeric(out["permno"], errors="raise")
if permno.isna().any() or not np.equal(permno, np.floor(permno)).all():
raise ValueError("daily CRSP permno values must be finite integers")
out["permno"] = permno.astype("int64")
if bool(out.duplicated(["permno", "date"]).any()):
raise ValueError("daily CRSP input has duplicate (permno, date) rows")
out["ret"] = _numeric_daily_return(
out,
column="ret",
context="daily CRSP input",
allow_missing=True,
)
finite = out["ret"].notna()
if bool((out.loc[finite, "ret"] < -1.0).any()):
raise ValueError("daily CRSP input contains a simple return below -1")
if convention == "siz":
share_codes = pd.to_numeric(out["shrcd"], errors="raise")
if share_codes.isna().any() or not share_codes.isin([10, 11]).all():
raise ValueError("legacy daily CRSP input must contain only share codes 10/11")
out["shrcd"] = share_codes.astype("int64")
exchanges = pd.to_numeric(out["exchcd"], errors="raise")
if exchanges.isna().any():
raise ValueError("legacy daily CRSP exchange codes cannot be missing")
out["exchcd"] = exchanges.astype("int64")
out["source_log_return"] = _safe_log1p(out["ret"])
leading = ["permno", "date", "ret", "source_log_return"]
trailing = [column for column in out.columns if column not in leading]
return (
out.loc[:, [*leading, *trailing]]
.sort_values(["permno", "date"], kind="stable")
.reset_index(drop=True)
)
[docs]
@step(
name="daily_market_rf_clean",
version=1,
inputs=["market", "risk_free"],
output="daily_market_rf_clean",
)
def daily_market_rf_clean(market: pd.DataFrame, risk_free: pd.DataFrame) -> pd.DataFrame:
"""Strictly align the CRSP value-weighted market and daily Fama--French risk-free rate.
Both calendars must match exactly. No date is forward-filled or assigned a zero rate. Inputs
and outputs are decimal simple returns. WRDS ``ff.factors_daily.rf`` has been verified in this
scale (historical values through 2012 peak at ``0.0006``); conservative magnitude guards fail
on obvious percent inputs while allowing historical daily market moves.
"""
for frame, value, context in (
(market, "market_return", "daily market input"),
(risk_free, "risk_free", "daily risk-free input"),
):
if not frame.columns.is_unique:
raise ValueError(f"{context} has duplicate columns")
if value not in frame:
raise ValueError(f"{context} has no {value!r} column")
dates = _daily_dates(frame, column="date", context=context)
if bool(dates.duplicated().any()):
raise ValueError(f"{context} has duplicate dates")
if frame.empty:
raise ValueError(f"{context} is empty")
left = market.loc[:, ["date", "market_return"]].copy()
right = risk_free.loc[:, ["date", "risk_free"]].copy()
left["date"] = _daily_dates(left, column="date", context="daily market input")
right["date"] = _daily_dates(right, column="date", context="daily risk-free input")
left["market_return"] = _numeric_daily_return(
left,
column="market_return",
context="daily market input",
allow_missing=False,
)
right["risk_free"] = _numeric_daily_return(
right,
column="risk_free",
context="daily risk-free input",
allow_missing=False,
)
if bool((left["market_return"].abs() > 1.0).any()):
raise ValueError("daily market returns exceed the decimal-unit safety bound")
if bool((right["risk_free"].abs() > 0.01).any()):
raise ValueError("daily risk-free returns exceed the decimal-unit safety bound")
merged = left.merge(right, on="date", how="outer", validate="1:1", indicator=True)
if not merged["_merge"].eq("both").all():
missing_market = int(merged["_merge"].eq("right_only").sum())
missing_rf = int(merged["_merge"].eq("left_only").sum())
raise ValueError(
"daily market and risk-free calendars differ "
f"(market missing {missing_market}, risk-free missing {missing_rf})"
)
merged = merged.drop(columns="_merge")
merged["market_excess"] = merged["market_return"] - merged["risk_free"]
if bool((merged["market_excess"] <= -1.0).any()):
raise ValueError("daily market excess return is not log-safe")
merged["market_log_excess"] = _safe_log1p(merged["market_excess"])
if merged["market_log_excess"].isna().any():
raise ValueError("daily market excess log return is missing")
return merged.sort_values("date", kind="stable").reset_index(drop=True)
[docs]
@step(
name="daily_stock_excess_clean",
version=1,
inputs=["stock", "market_rf"],
output="daily_stock_excess_clean",
)
def daily_stock_excess_clean(stock: pd.DataFrame, market_rf: pd.DataFrame) -> pd.DataFrame:
"""Subtract the same-date daily RF from one clean stock bucket.
Absent stock observations remain absent and later become explicit missing values when a caller
reindexes to the market-session calendar. Every present stock date must have an RF row.
``beta_log_return`` is missing for source or excess returns at/below -100%, while the original
total ``ret`` remains available for return accounting.
"""
required_stock = {"permno", "date", "ret"}
missing_stock = sorted(required_stock.difference(stock.columns))
if missing_stock:
raise ValueError(f"clean daily stock input lacks columns: {missing_stock}")
required_calendar = {"date", "risk_free"}
missing_calendar = sorted(required_calendar.difference(market_rf.columns))
if missing_calendar:
raise ValueError(f"daily market/RF input lacks columns: {missing_calendar}")
if not stock.columns.is_unique or not market_rf.columns.is_unique:
raise ValueError("daily stock and market/RF inputs must have unique columns")
if bool(stock.duplicated(["permno", "date"]).any()):
raise ValueError("clean daily stock input has duplicate (permno, date) rows")
if bool(market_rf.duplicated(["date"]).any()):
raise ValueError("daily market/RF input has duplicate dates")
out = stock.copy()
out["date"] = _daily_dates(out, column="date", context="clean daily stock input")
out["ret"] = _numeric_daily_return(
out,
column="ret",
context="clean daily stock input",
allow_missing=True,
)
calendar = market_rf.loc[:, ["date", "risk_free"]].copy()
calendar["date"] = _daily_dates(calendar, column="date", context="daily market/RF input")
calendar["risk_free"] = _numeric_daily_return(
calendar,
column="risk_free",
context="daily market/RF input",
allow_missing=False,
)
out = out.merge(calendar, on="date", how="left", validate="m:1", indicator=True)
if not out["_merge"].eq("both").all():
raise ValueError("at least one stock date is absent from the daily market/RF calendar")
out = out.drop(columns="_merge")
out["excess_ret"] = out["ret"] - out["risk_free"]
out["beta_log_return"] = _safe_log1p(out["excess_ret"])
leading = ["permno", "date", "ret", "risk_free", "excess_ret", "beta_log_return"]
trailing = [column for column in out.columns if column not in leading]
return (
out.loc[:, [*leading, *trailing]]
.sort_values(["permno", "date"], kind="stable")
.reset_index(drop=True)
)
[docs]
@step(
name="crsp_monthly_clean",
version=3,
inputs=["msf", "msenames", "msedelist"],
output="crsp_monthly_clean",
)
def crsp_monthly_clean(
msf: pd.DataFrame,
msenames: pd.DataFrame,
msedelist: pd.DataFrame,
*,
share_codes: tuple[int, ...] = (10, 11),
exchanges: tuple[int, ...] | None = (1, 2, 3),
performance_delist_fill: float | None = _PERF_DELIST_FILL,
) -> pd.DataFrame:
"""Tidy monthly CRSP: common-stock filter, delisting-adjusted returns, month-end mktcap.
``msf`` = monthly stock file (``permno, date, ret, prc, shrout``); ``msenames`` = the name
history (``permno, namedt, nameendt, shrcd, exchcd``); ``msedelist`` = delisting returns
(``permno, date, dlret, dlstcd``). Keeps ``shrcd in share_codes`` and, unless
``exchanges=None``, ``exchcd in exchanges`` (default NYSE/AMEX/NASDAQ), adjusts returns for
delisting (missing perf-delist returns filled at ``performance_delist_fill``, -30% by default),
and computes
``mktcap = |prc| * shrout`` (thousands -> $millions). Pass
``performance_delist_fill=None`` to preserve an unreported performance-delisting return as
missing; the choice is explicit recipe input for reproductions whose paper does not state an
imputation convention.
Merge cardinality is validated so a duplicated name/delist record raises instead of silently
duplicating a firm-month's return: the name-window join must resolve to at most one name record
per firm-month (overlapping ``msenames`` windows raise), and the stock and delisting inputs must
each contain at most one row per normalized ``(permno, month)``. Those two files are
**outer-joined** so a terminal month present only in ``msedelist`` is retained. Such a row has
``is_terminal=True``; its market capitalization remains missing because no price or shares are
fabricated.
A **genuinely missing** monthly ``ret`` (no usable delisting return that month) is kept as
``NaN`` rather than fabricated as a flat ``0.0``. Only a terminal month with an observed (or
performance-code-imputed) ``dlret`` treats a missing price return as zero so that the delisting
return still enters the combined return. ``source_ret``, ``dlret``, and ``dlstcd`` are retained
alongside adjusted ``ret`` so the combination identity and any -30% convention remain
inspectable. When present in the raw frames, ``permco`` and ``siccd`` are retained as
method-neutral firm and industry identifiers.
"""
df = msf.copy()
if performance_delist_fill is not None and (
not np.isfinite(performance_delist_fill) or performance_delist_fill < -1.0
):
raise ValueError("performance_delist_fill must be None or a finite simple return >= -1")
df["_msf_asof_date"] = pd.to_datetime(df["date"])
df["date"] = df["_msf_asof_date"] + pd.offsets.MonthEnd(0)
de = msedelist.copy()
de["_delist_asof_date"] = pd.to_datetime(de["date"])
de["date"] = de["_delist_asof_date"] + pd.offsets.MonthEnd(0)
# ``validate='1:1'`` bites after normalization: two daily delisting records in one calendar
# month cannot silently duplicate the corresponding monthly stock observation.
df = df.merge(
de[["permno", "date", "_delist_asof_date", "dlret", "dlstcd"]],
on=["permno", "date"],
how="outer",
validate="1:1",
indicator="_delist_origin",
)
df["is_terminal"] = df["_delist_origin"] != "left_only"
# Name histories end on the actual delisting day, which may precede month-end. Use that event
# date for terminal rows; comparing the normalized month-end to ``nameendt`` would wrongly
# discard exactly the delist-only row that the outer join is meant to preserve.
df["_name_asof_date"] = df["_delist_asof_date"].fillna(df["_msf_asof_date"])
names = msenames.copy()
# join the name record in force at each date (namedt <= date <= nameendt)
names["namedt"] = pd.to_datetime(names["namedt"])
names["nameendt"] = pd.to_datetime(names["nameendt"]).fillna(pd.Timestamp("2100-12-31"))
name_cols = ["permno", "namedt", "nameendt", "shrcd", "exchcd"]
if "siccd" in names:
names = names.rename(columns={"siccd": "_name_siccd"})
name_cols.append("_name_siccd")
if "permco" in names:
# An msf-supplied permco (when a caller has one) wins; the name-history value fills the
# delist-only rows that have no stock-file observation.
names = names.rename(columns={"permco": "_name_permco"})
name_cols.append("_name_permco")
# a permno legitimately has multiple name records (successive windows), so the join is m:m by
# construction; the window filter must then leave at most one row per firm-month. Overlapping
# windows would duplicate a return, so we guard the post-filter (permno, date) uniqueness.
df = df.merge(names[name_cols], on="permno", how="left")
df = df[(df["_name_asof_date"] >= df["namedt"]) & (df["_name_asof_date"] <= df["nameendt"])]
universe = df["shrcd"].isin(share_codes)
if exchanges is not None:
universe &= df["exchcd"].isin(exchanges)
df = df[universe]
dup = df.duplicated(subset=["permno", "date"])
if dup.any():
n = int(dup.sum())
raise ValueError(
f"overlapping msenames windows produced {n} duplicate (permno, date) row(s); "
"a firm-month resolved to more than one name record"
)
if "_name_permco" in df:
name_permco = pd.to_numeric(df["_name_permco"], errors="raise")
if "permco" in df:
df["permco"] = pd.to_numeric(df["permco"], errors="raise").fillna(name_permco)
else:
df["permco"] = name_permco
if "_name_siccd" in df:
name_siccd = pd.to_numeric(df["_name_siccd"], errors="raise")
if "siccd" in df:
df["siccd"] = pd.to_numeric(df["siccd"], errors="raise").fillna(name_siccd)
else:
df["siccd"] = name_siccd
# Empty WRDS results commonly arrive with object-typed numeric columns. Normalize explicitly
# before fill/arithmetic so pandas cannot silently downcast objects today or change behavior in
# a future release. ``errors='raise'`` keeps malformed non-numeric observations fail-closed.
df["ret"] = _numeric_daily_return(
df,
column="ret",
context="monthly CRSP stock input",
allow_missing=True,
)
if bool((df.loc[df["ret"].notna(), "ret"] < -1.0).any()):
raise ValueError("monthly CRSP stock input contains a simple return below -1")
df["source_ret"] = df["ret"]
df["dlret"] = _numeric_daily_return(
df,
column="dlret",
context="monthly CRSP delisting input",
allow_missing=True,
)
if bool((df.loc[df["dlret"].notna(), "dlret"] < -1.0).any()):
raise ValueError("monthly CRSP delisting input contains a simple return below -1")
perf = df["is_terminal"] & df["dlstcd"].isin(_PERF_DELIST) & df["dlret"].isna()
if performance_delist_fill is not None:
df.loc[perf, "dlret"] = performance_delist_fill
has_delist = df["dlret"].notna() # an observed or imputed delisting return applies this month
dlret = df["dlret"].fillna(0.0)
# Keep a genuinely missing return NaN; only zero-fill when there is a usable delisting return.
ret = df["ret"].where(~(df["ret"].isna() & has_delist), 0.0)
df["ret"] = (1.0 + ret) * (1.0 + dlret) - 1.0
df["mktcap"] = (df["prc"].abs() * df["shrout"]) / 1000.0 # shrout in thousands -> $millions
out_cols = ["permno"]
if "permco" in df:
out_cols.append("permco")
out_cols.extend(["date", "source_ret", "ret", "mktcap", "exchcd", "shrcd"])
if "siccd" in df:
out_cols.append("siccd")
out_cols.extend(["dlret", "dlstcd", "is_terminal"])
out = df[out_cols].copy()
out = out.sort_values(["permno", "date"], kind="stable").reset_index(drop=True)
return out
[docs]
@step(
name="monthly_risk_free_clean",
version=1,
inputs=["risk_free"],
output="monthly_risk_free_clean",
)
def monthly_risk_free_clean(risk_free: pd.DataFrame) -> pd.DataFrame:
"""Normalize WRDS Fama--French monthly RF from source-month labels to month-end.
``ff.factors_monthly`` uses first-of-month date labels and decimal simple returns. Through the
1926-07--2012-03 paper horizon, its observed maximum is 0.0135. This cleaner preserves the
decimal values, converts labels to realized-return month-end, and rejects duplicate months,
missing/non-finite values, and obvious percentage-point inputs. It never fills a missing month.
"""
if not risk_free.columns.is_unique:
raise ValueError("monthly risk-free input has duplicate columns")
if risk_free.empty:
raise ValueError("monthly risk-free input is empty")
out = risk_free.loc[:, ["date", "risk_free"]].copy()
source_dates = _daily_dates(out, column="date", context="monthly risk-free input")
if not source_dates.dt.is_month_start.all():
raise ValueError("monthly risk-free source dates must label months at their first day")
out["date"] = source_dates + pd.offsets.MonthEnd(0)
if bool(out["date"].duplicated().any()):
raise ValueError("monthly risk-free input has duplicate months")
out["risk_free"] = _numeric_daily_return(
out,
column="risk_free",
context="monthly risk-free input",
allow_missing=False,
)
if bool((out["risk_free"].abs() > 0.05).any()):
raise ValueError("monthly risk-free returns exceed the decimal-unit safety bound")
return out.sort_values("date", kind="stable").reset_index(drop=True)
[docs]
@step(
name="monthly_stock_excess_clean",
version=1,
inputs=["stock", "risk_free"],
output="monthly_stock_excess_clean",
)
def monthly_stock_excess_clean(stock: pd.DataFrame, risk_free: pd.DataFrame) -> pd.DataFrame:
"""Attach same-month RF and an excess return to one clean monthly stock bucket."""
required = {"permno", "date", "source_ret", "ret"}
missing = sorted(required.difference(stock.columns))
if missing:
raise ValueError(f"clean monthly stock input lacks columns: {missing}")
if not stock.columns.is_unique or not risk_free.columns.is_unique:
raise ValueError("monthly stock and risk-free inputs must have unique columns")
if bool(stock.duplicated(["permno", "date"]).any()):
raise ValueError("clean monthly stock input has duplicate (permno, date) rows")
if bool(risk_free.duplicated(["date"]).any()):
raise ValueError("monthly risk-free input has duplicate dates")
out = stock.copy()
out["date"] = _daily_dates(out, column="date", context="clean monthly stock input")
if not out["date"].eq(out["date"] + pd.offsets.MonthEnd(0)).all():
raise ValueError("clean monthly stock dates must be month-end")
out["source_ret"] = _numeric_daily_return(
out,
column="source_ret",
context="clean monthly stock input",
allow_missing=True,
)
out["ret"] = _numeric_daily_return(
out,
column="ret",
context="clean monthly stock input",
allow_missing=True,
)
calendar = risk_free.loc[:, ["date", "risk_free"]].copy()
calendar["date"] = _daily_dates(
calendar,
column="date",
context="monthly risk-free input",
)
if not calendar["date"].eq(calendar["date"] + pd.offsets.MonthEnd(0)).all():
raise ValueError("monthly risk-free dates must be month-end")
calendar["risk_free"] = _numeric_daily_return(
calendar,
column="risk_free",
context="monthly risk-free input",
allow_missing=False,
)
out = out.merge(calendar, on="date", how="left", validate="m:1", indicator=True)
if not out["_merge"].eq("both").all():
raise ValueError("at least one stock month is absent from the monthly risk-free calendar")
out = out.drop(columns="_merge")
out["excess_ret"] = out["ret"] - out["risk_free"]
leading = ["permno", "date", "source_ret", "ret", "risk_free", "excess_ret"]
trailing = [column for column in out.columns if column not in leading]
return (
out.loc[:, [*leading, *trailing]]
.sort_values(["permno", "date"], kind="stable")
.reset_index(drop=True)
)
[docs]
@step(name="compustat_annual_clean", version=5, inputs=["funda"], output="compustat_annual_clean")
def compustat_annual_clean(
funda: pd.DataFrame,
*,
min_book_equity: float = 0.0,
report_lag_months: int = 6,
) -> pd.DataFrame:
"""Tidy annual Compustat: book equity, operating profitability, and asset-growth investment.
``be`` = stockholders' equity (``seq``, else ``ceq+pstk``, else ``at-lt``) + deferred taxes
(``txditc``) - preferred stock (``pstkrv``, else ``pstkl``, else ``pstk``); rows with
``be <= min_book_equity`` are dropped. ``op`` = (``revt`` - ``cogs`` - ``xsga`` - ``xint``)/be.
``inv`` = year-over-year growth in total assets (``at``), by ``gvkey``. Growth is measured only
between an observation and the *immediately preceding fiscal year* — a ``datadate`` exactly
twelve months earlier. It is missing for a firm's first observation, across a skipped year or a
mid-history fiscal-year-end change (where the prior row is not twelve months back), and when
either the current or the immediately preceding row contains a missing asset value. The builder
never carries an older asset value across such a gap or a missing-value row; row-adjacency alone
(which pandas' ``pct_change`` would use) is not treated as year-adjacency.
**Heuristic point-in-time availability.** Annual fundamentals are *not* public at the
fiscal-period end (``datadate``); the 10-K is filed months later. Joining a return panel at
``datadate`` would look ahead. So this emits an **``avail_date`` = ``datadate`` +
``report_lag_months``** column (month-end stamped) and downstream view builders join
point-in-time on ``avail_date``, never on ``datadate``. The default ``report_lag_months=6`` is
a **per-record fixed-lag heuristic**, not the Fama--French calendar-year portfolio-formation
rule. In particular, adding six months to a
January or March fiscal-year end would make it eligible during that same calendar year, whereas
the Fama--French rule waits until the following June. The lag remains an explicit parameter for
backward compatibility and exploratory robustness; changing it changes the recipe hash. Use
:func:`compustat_annual_formation_clean` when the study requires the calendar-year rule. (The
sibling ``compustat_quarterly_clean`` dates availability off the actual ``rdq`` announcement,
which is finer-grained; the annual file carries no announcement date, so a fixed lag is used.)
Output columns: ``gvkey, datadate, avail_date, be, op, inv, at``.
"""
df = funda.copy()
df["datadate"] = pd.to_datetime(df["datadate"]) + pd.offsets.MonthEnd(0)
if df.duplicated(["gvkey", "datadate"]).any():
raise ValueError("annual Compustat input has duplicate gvkey/datadate rows")
df = df.sort_values(["gvkey", "datadate"], kind="stable")
# Calculate on the complete firm history before any book-equity filter. A row with missing
# equity may disappear from the final table, but its missing asset observation must still break
# both adjacent growth rates rather than connecting the next year to an older value.
#
# ``pct_change`` measures growth between *row-adjacent* observations, which is a genuine
# year-over-year change only when the prior row is the immediately preceding fiscal year. A
# history that skips a year (e.g. a 2000 record followed directly by 2002) would otherwise
# report the two-year jump as a single year of growth. Require the prior observation to sit
# exactly twelve months earlier; growth across any other gap (a skipped year or a mid-history
# fiscal-year-end change) is left missing rather than mislabeled.
raw_growth = df.groupby("gvkey")["at"].pct_change(fill_method=None)
month_index = df["datadate"].dt.year * 12 + df["datadate"].dt.month
prior_gap = month_index - month_index.groupby(df["gvkey"]).shift(1)
df["inv"] = raw_growth.where(prior_gap.eq(12))
def _c(col: str) -> pd.Series:
return df[col] if col in df else pd.Series(np.nan, index=df.index)
equity = _c("seq").fillna(_c("ceq") + _c("pstk")).fillna(_c("at") - _c("lt"))
pref = _c("pstkrv").fillna(_c("pstkl")).fillna(_c("pstk")).fillna(0.0)
df["be"] = equity + _c("txditc").fillna(0.0) - pref
df = df[df["be"] > min_book_equity]
df["op"] = (
_c("revt") - _c("cogs").fillna(0.0) - _c("xsga").fillna(0.0) - _c("xint").fillna(0.0)
) / df["be"]
# public-availability date: fiscal-period end plus a conservative reporting lag (month-end).
df["avail_date"] = df["datadate"] + pd.offsets.MonthEnd(report_lag_months)
out = df[["gvkey", "datadate", "avail_date", "be", "op", "inv", "at"]].reset_index(drop=True)
return out
_ANNUAL_FORMATION_FIELDS = ("revt", "cogs", "at", "gp", "fyear", "fyr", "sic")
[docs]
@step(
name="compustat_quarterly_clean",
version=1,
inputs=["fundq"],
output="compustat_quarterly_clean",
)
def compustat_quarterly_clean(
fundq: pd.DataFrame,
*,
min_book_equity: float = 0.0,
rdq_fallback_months: int = 4,
) -> pd.DataFrame:
"""Tidy quarterly Compustat: fresh-earnings ROE with an announcement-dated availability month.
Reproduces the Hou-Xue-Zhang ROE construction from the quarterly file: ``roe_q`` = income
before extraordinary items (``ibq``) over **one-quarter-lagged** book equity, where quarterly
book equity = shareholders' equity (``seqq``, else ``ceqq+pstkq``, else ``atq-ltq``) + deferred
taxes (``txditcq``) - preferred stock (``pstkrq``, else ``pstkq``). Rows whose lagged book
equity is ``<= min_book_equity`` yield a missing ROE and are dropped.
The point-in-time part is the **availability month** ``avail_date``: HXZ time ROE off the most
recent *public* quarterly earnings announcement, Compustat item ``rdq``. We use the announcement
date where present; when ``rdq`` is missing we fall back to ``datadate + rdq_fallback_months``
(default 4 months — conservative relative to the ~2-month typical filing gap, so a missing
announcement date never manufactures look-ahead). ``avail_date`` is floored at ``datadate`` so
a stray early ``rdq`` cannot predate the fiscal quarter it reports. Both are month-end stamped;
the downstream monthly panel maps each firm-month to its most recently *available* ROE.
Output columns: ``gvkey, datadate, rdq, roe_q, avail_date`` (one row per fiscal quarter).
"""
df = fundq.copy()
df["datadate"] = pd.to_datetime(df["datadate"]) + pd.offsets.MonthEnd(0)
df = df.sort_values(["gvkey", "datadate"], kind="stable").reset_index(drop=True)
def _c(col: str) -> pd.Series:
return df[col] if col in df else pd.Series(np.nan, index=df.index)
equity = _c("seqq").fillna(_c("ceqq") + _c("pstkq")).fillna(_c("atq") - _c("ltq"))
pref = _c("pstkrq").fillna(_c("pstkq")).fillna(0.0)
df["be_q"] = equity + _c("txditcq").fillna(0.0) - pref
df["be_q_lag"] = df.groupby("gvkey")["be_q"].shift(1)
df["roe_q"] = _c("ibq") / df["be_q_lag"].where(df["be_q_lag"] > min_book_equity)
# availability = announcement month; fallback to a conservative reporting lag; never < datadate
rdq = pd.to_datetime(_c("rdq"), errors="coerce") + pd.offsets.MonthEnd(0)
fallback = df["datadate"] + pd.offsets.MonthEnd(rdq_fallback_months)
df["avail_date"] = rdq.fillna(fallback).clip(lower=df["datadate"])
out = df[["gvkey", "datadate", "rdq", "roe_q", "avail_date"]].dropna(subset=["roe_q"])
return out.reset_index(drop=True)
# NOTE (CCM link table): we read the WRDS link-history view `crsp.ccmxpf_lnkhist`. tidyfinance's
# `list_supported_datasets()` advertises the CCM source as `crsp.ccmxpf_linktable`, but its code
# actually queries `crsp.ccmxpf_lnkhist` too, so the download is correct — the label is only a
# cosmetic metadata discrepancy (verified live: both views exist and resolve; `linktable` merely
# adds a `usedflag` column). NOT a functional bug; no upstream fix warranted.
# NOTE (WRDS schema choice): WRDS ships the same CRSP tables under three update cadences —
# `crsp` (annual), `crspq` (quarterly), `crspm` (monthly). Prefer `crspm.*` when pulling: fresher
# vintages, identical table layout. The schema belongs in the raw pull's `(source, vintage)`
# identity / query_hash, not in this step (which only sees the frame), so switching cadence never
# silently changes a recipe.
[docs]
@step(name="ccm_links_clean", version=1, inputs=["ccmxpf_linkhist"], output="ccm_links_clean")
def ccm_links_clean(
ccmxpf_linkhist: pd.DataFrame,
*,
link_types: tuple[str, ...] = ("LU", "LC"),
link_prims: tuple[str, ...] = ("P", "C"),
) -> pd.DataFrame:
"""Tidy CRSP-Compustat links: primary, reliable gvkey<->permno links with valid date windows.
Keeps ``linktype in link_types`` (LU/LC) and ``linkprim in link_prims`` (P/C), maps to
``gvkey, permno, linkdt, linkenddt`` with an open link end filled to a sentinel far-future date.
"""
df = ccmxpf_linkhist.copy()
df = df[df["linktype"].isin(link_types) & df["linkprim"].isin(link_prims)]
df["linkdt"] = pd.to_datetime(df["linkdt"])
df["linkenddt"] = pd.to_datetime(df["linkenddt"]).fillna(pd.Timestamp("2100-12-31"))
out = df.rename(columns={"lpermno": "permno"})[["gvkey", "permno", "linkdt", "linkenddt"]]
out = out.dropna(subset=["permno"]).astype({"permno": "int64"})
return out.sort_values(["gvkey", "linkdt"], kind="stable").reset_index(drop=True)