1"""Firm-quarter outcomes: reveal dates, seasonal-random-walk expectations and adverse-outcome labels (PIT_PROTOCOL §4).23Nothing here uses prices. The expectation E_c[X_q] uses quarters q-1..q-8 *restricted to those available at the cutoff*;4samples.py passes the cutoff-specific availability mask. The quarterly series X_q is the first (only) revision.5"""6from __future__ import annotations78import datetime as dt910import polars as pl1112TARGET_VARS = ["op_inc", "net_inc", "cash_op", "rev"]131415def match_release_dates(wide: pl.DataFrame, events: pl.DataFrame, max_lag_days: int = 30) -> pl.DataFrame:16 """Attach A_q (first release in [E_q, F_q + 10 d]) and its timing to each firm-quarter; flag exclusions."""17 e = events.select(["entity_id", pl.col("event_date").cast(pl.Date), "release_time", "eps_estimate", "eps_reported"]).sort(18 ["entity_id", "event_date"])19 w = wide.select(["entity_id", "fiscal_period_end", "filing_date"]).with_columns(20 pl.col("fiscal_period_end").alias("_key")).sort(["entity_id", "_key"])21 # forward as-of: first event on/after the fiscal period end22 j = w.join_asof(e, left_on="_key", right_on="event_date", by="entity_id", strategy="forward")23 j = j.with_columns([24 pl.col("event_date").alias("A_q"),25 (pl.col("event_date") <= pl.col("filing_date") + pl.duration(days=10)).alias("_within"),26 ])27 j = j.with_columns([28 pl.when(pl.col("_within")).then(pl.col("A_q")).otherwise(None).alias("A_q"),29 pl.when(pl.col("_within")).then(pl.col("release_time")).otherwise(None).alias("release_time"),30 pl.when(pl.col("_within")).then(pl.col("eps_estimate")).otherwise(None).alias("eps_estimate"),31 pl.when(pl.col("_within")).then(pl.col("eps_reported")).otherwise(None).alias("eps_reported"),32 ])33 j = j.with_columns([34 (pl.col("filing_date") - pl.col("A_q")).dt.total_days().alias("filing_minus_release_days"),35 ])36 j = j.with_columns([37 pl.when(pl.col("A_q").is_null()).then(pl.lit("no_release"))38 .when(pl.col("filing_minus_release_days") < 0).then(pl.lit("filing_before_release"))39 .when(pl.col("filing_minus_release_days") > max_lag_days).then(pl.lit("release_lag_gt_30d"))40 .otherwise(pl.lit("ok")).alias("release_match"),41 ])42 return wide.join(j.select(["entity_id", "fiscal_period_end", "A_q", "release_time", "eps_estimate", "eps_reported",43 "filing_minus_release_days", "release_match"]), on=["entity_id", "fiscal_period_end"], how="left")444546def seasonal_expectation(w: pl.DataFrame, var: str) -> pl.DataFrame:47 """E[X_q] = X_{q-4} + mean_{k=0..3}(X_{q-1-k} - X_{q-5-k}); requires q-1..q-8 (by quarter index, regular calendar)."""48 x = pl.col(var)49 lags = {k: x.shift(k).over("entity_id") for k in range(1, 9)}50 drift = sum((lags[1 + k] - lags[5 + k]) for k in range(4)) / 4.051 return w.with_columns([52 (lags[4] + drift).alias(f"E_{var}"),53 lags[4].alias(f"{var}_lag4"),54 lags[1].alias(f"{var}_lag1"),55 (x - (lags[4] + drift)).alias(f"surprise_{var}"),56 ])575859def build_quarter_table(wide: pl.DataFrame) -> pl.DataFrame:60 """Quarter-level target ingredients (unscaled). Regularity: consecutive quarters 60-120 days apart."""61 w = wide.sort(["entity_id", "fiscal_period_end"])62 w = w.with_columns(pl.col("fiscal_period_end").diff().over("entity_id").dt.total_days().alias("days_since_prev_q"))63 # a quarter is "regular" if the previous 8 gaps are all within [60, 120]64 ok = (pl.col("days_since_prev_q") >= 60) & (pl.col("days_since_prev_q") <= 120)65 w = w.with_columns(ok.cast(pl.Int8).alias("_gap_ok"))66 w = w.with_columns(pl.col("_gap_ok").rolling_sum(window_size=8, min_samples=8).over("entity_id").alias("_gap_ok8"))67 for v in TARGET_VARS:68 w = seasonal_expectation(w, v)69 # secondary ingredients70 w = w.with_columns([71 pl.col("roa").shift(4).over("entity_id").alias("roa_lag4"),72 (pl.col("assets_c") / (pl.col("liab_c").abs() + 1e-9)).alias("current_ratio"),73 ])74 w = w.with_columns([75 pl.col("current_ratio").shift(4).over("entity_id").alias("current_ratio_lag4"),76 pl.col("intr_cover").shift(4).over("entity_id").alias("intr_cover_lag4"),77 pl.col("assets_tot").shift(1).over("entity_id").alias("assets_tot_lag1"),78 ])79 return w808182def finalize_labels(q: pl.DataFrame, threshold: float = -0.005) -> pl.DataFrame:83 """Scale by the latest *available* total assets (lag-1 quarter) and build binary targets."""84 ta = pl.col("assets_tot_lag1").abs() + 1e-985 q = q.with_columns([86 (pl.col("surprise_op_inc") / ta).alias("S_opinc"),87 (pl.col("surprise_net_inc") / ta).alias("S_netinc"),88 (pl.col("surprise_cash_op") / ta).alias("S_ocf"),89 (pl.col("surprise_rev") / ta).alias("S_rev"),90 ((pl.col("cash_op") - pl.col("cash_op_lag4")) / ta).alias("ocf_innov"),91 (pl.col("roa") - pl.col("roa_lag4")).alias("roa_yoy"),92 ])93 q = q.with_columns([94 (pl.col("S_opinc") <= threshold).cast(pl.Int8).alias("y_adverse_opinc"),95 (pl.col("S_netinc") <= threshold).cast(pl.Int8).alias("y_adverse_netinc"),96 (pl.col("roa_yoy") <= -0.01).cast(pl.Int8).alias("y_roa_decline"),97 (pl.col("ocf_innov") <= -0.01).cast(pl.Int8).alias("y_ocf_negative"),98 (((pl.col("current_ratio") / (pl.col("current_ratio_lag4") + 1e-9)) <= 0.9) |99 ((pl.col("intr_cover") < 1.0) & (pl.col("intr_cover_lag4") >= 1.0))).cast(pl.Int8).alias("y_liquidity_deterioration"),100 pl.when(pl.col("eps_estimate").is_not_null() & pl.col("eps_reported").is_not_null())101 .then((pl.col("eps_reported") < pl.col("eps_estimate")).cast(pl.Int8)).otherwise(None).alias("y_eps_miss"),102 (pl.col("S_opinc").abs() >= 0.005).cast(pl.Int8).alias("y_abs_surprise"), # symmetric placebo target (volatility channel)103 ])104 return q105