Spaces:
Running on Zero
Running on Zero
File size: 4,786 Bytes
10ec54c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | """
transaction_features.py
------------------------
Leakage-safe transaction-level feature engineering for Dataset A.
Features generated:
- amount_log1p
- hour, day_of_week, is_weekend
- customer_txn_count_past (historical customer transaction count)
- customer_amount_mean_past (historical customer average amount)
- customer_amount_std_past (historical customer amount standard deviation)
- device_txn_count_past (historical device transaction count)
- customer_amount_dev (amount ratio relative to customer's historical average)
- missingness indicators (identity_available, missing_p_email, missing_r_email, missing_addr1, missing_device_info)
Guarantees:
- All rolling/expanding features use strictly observations prior to current transaction index.
- Zero future data leakage.
"""
from __future__ import annotations
import logging
from pathlib import Path
import numpy as np
import pandas as pd
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(message)s",
)
LOGGER = logging.getLogger("transaction-features")
def generate_dataset_a_features(df: pd.DataFrame) -> pd.DataFrame:
"""
Computes leakage-free transaction features for Dataset A.
Assumes df is sorted or will sort df chronologically by event_time.
"""
LOGGER.info("Generating Dataset A features for %s rows ...", len(df))
df_out = df.sort_values("event_time").reset_index(drop=True)
# 1. Basic time features
df_out["hour"] = df_out["event_time"].dt.hour.astype("int8")
df_out["day_of_week"] = df_out["event_time"].dt.dayofweek.astype("int8")
df_out["is_weekend"] = (df_out["day_of_week"] >= 5).astype("int8")
# 2. Amount log1p
df_out["amount_log1p"] = np.log1p(np.clip(df_out["amount"], 0, None)).astype("float32")
# 3. Leakage-safe customer historical features
cust_group = df_out.groupby("customer_proxy_id")
# Expanding count of prior transactions
past_cust_count = cust_group.cumcount().astype("int32")
df_out["customer_txn_count_past"] = past_cust_count
# Past sum of amount (cumsum minus current row)
amt = df_out["amount"].astype("float64")
amt_cumsum = cust_group["amount"].cumsum()
past_amt_sum = (amt_cumsum - amt).values
past_count_arr = past_cust_count.values
valid_mask = past_count_arr > 0
past_amt_mean = np.zeros(len(df_out), dtype="float32")
past_amt_mean[valid_mask] = (past_amt_sum[valid_mask] / past_count_arr[valid_mask]).astype("float32")
df_out["customer_amount_mean_past"] = past_amt_mean
# Past variance of amount
amt_sq = amt ** 2
amt_sq_cumsum = df_out.groupby("customer_proxy_id")["amount"].transform(lambda s: (s.astype("float64")**2).cumsum())
past_amt_sq_sum = (amt_sq_cumsum - amt_sq).values
past_amt_var = np.zeros(len(df_out), dtype="float32")
past_amt_var[valid_mask] = (
(past_amt_sq_sum[valid_mask] / past_count_arr[valid_mask]) - (past_amt_mean[valid_mask] ** 2)
)
df_out["customer_amount_std_past"] = np.sqrt(np.maximum(0.0, past_amt_var)).astype("float32")
# Amount deviation from customer past mean
amt_dev = np.ones(len(df_out), dtype="float32")
amt_dev[valid_mask] = (
df_out["amount"].values[valid_mask] / (past_amt_mean[valid_mask] + 1e-5)
).astype("float32")
df_out["customer_amount_dev"] = amt_dev
# 4. Leakage-safe device historical count
dev_group = df_out.groupby("device_proxy_id")
df_out["device_txn_count_past"] = dev_group.cumcount().astype("int32")
# 5. Missingness indicators
df_out["identity_available"] = (
df_out["DeviceInfo"].notna() | df_out["DeviceType"].notna()
).astype("int8")
df_out["missing_p_email"] = df_out["P_emaildomain"].isna().astype("int8")
df_out["missing_r_email"] = df_out["R_emaildomain"].isna().astype("int8")
df_out["missing_addr1"] = df_out["addr1"].isna().astype("int8")
df_out["missing_device_info"] = df_out["DeviceInfo"].isna().astype("int8")
LOGGER.info("Dataset A feature engineering complete. Total columns: %s", len(df_out.columns))
return df_out
def build_and_save_dataset_a_features(
input_path: Path | None = None,
output_path: Path | None = None,
) -> Path:
root = Path(__file__).resolve().parents[2]
if input_path is None:
input_path = root / "data" / "processed" / "dataset_a_model.parquet"
if output_path is None:
output_path = root / "data" / "processed" / "dataset_a_features.parquet"
df = pd.read_parquet(input_path)
df_feats = generate_dataset_a_features(df)
df_feats.to_parquet(output_path, index=False)
LOGGER.info("Dataset A features saved to %s", output_path)
return output_path
if __name__ == "__main__":
build_and_save_dataset_a_features()
|