Spaces:
Running on Zero
Running on Zero
File size: 4,915 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 126 127 128 129 130 131 132 133 134 135 136 137 138 | """
scenario_features.py
--------------------
Leakage-safe temporal and scenario feature engineering for Dataset B.
Features verified/generated:
- rolling_txn_15m
- rolling_fraud_rate_15m
- baseline_txn_15m
- baseline_fraud_rate
- velocity_ratio
- fraud_rate_deviation
- amount_deviation
Guarantees:
- Rolling features are computed strictly on past 15-minute rolling windows.
- Baselines are calculated exclusively from early non-spike baseline windows.
- Zero future scenario temporal 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("scenario-features")
def generate_dataset_b_features(df: pd.DataFrame) -> pd.DataFrame:
"""
Ensures all merchant temporal & scenario features are present and cleanly formatted for Dataset B.
"""
LOGGER.info("Generating Dataset B scenario features for %s rows ...", len(df))
df_out = df.copy()
required_cols = [
"rolling_txn_15m",
"rolling_fraud_rate_15m",
"baseline_txn_15m",
"baseline_fraud_rate",
"velocity_ratio",
"fraud_rate_deviation",
"amount_deviation",
]
# Check if features exist, else compute them per scenario
missing = [c for c in required_cols if c not in df_out.columns]
if missing:
LOGGER.info("Computing missing Dataset B features: %s", missing)
frames = []
for scenario_id, group in df_out.groupby("scenario_id"):
grp = group.sort_values("event_time").reset_index(drop=True).copy()
grp["minute_bucket"] = grp["event_time"].dt.floor("min")
# Per-minute aggregations
per_min = (
grp.groupby("minute_bucket", as_index=False)
.agg(
minute_txn_count=("amount", "count"),
minute_fraud_count=("is_fraud", "sum"),
minute_amount_sum=("amount", "sum"),
)
)
per_min["rolling_txn_15m"] = per_min["minute_txn_count"].rolling(15, min_periods=1).sum()
per_min["rolling_fraud_15m"] = per_min["minute_fraud_count"].rolling(15, min_periods=1).sum()
per_min["rolling_fraud_rate_15m"] = per_min["rolling_fraud_15m"] / per_min["rolling_txn_15m"].clip(lower=1)
# Baseline from first 30 minutes
base_window = per_min.iloc[: min(30, len(per_min))]
b_txn_15m = float(base_window["minute_txn_count"].mean() * 15)
b_fraud_rate = float(base_window["minute_fraud_count"].sum() / max(1, base_window["minute_txn_count"].sum()))
b_amt = float(base_window["minute_amount_sum"].mean() / max(1.0, base_window["minute_txn_count"].mean()))
per_min["baseline_txn_15m"] = max(1.0, b_txn_15m)
per_min["baseline_fraud_rate"] = b_fraud_rate
per_min["velocity_ratio"] = per_min["rolling_txn_15m"] / per_min["baseline_txn_15m"]
per_min["fraud_rate_deviation"] = per_min["rolling_fraud_rate_15m"] - per_min["baseline_fraud_rate"]
grp = grp.merge(
per_min[
[
"minute_bucket",
"rolling_txn_15m",
"rolling_fraud_rate_15m",
"baseline_txn_15m",
"baseline_fraud_rate",
"velocity_ratio",
"fraud_rate_deviation",
]
],
on="minute_bucket",
how="left",
)
grp["baseline_amount"] = max(1.0, b_amt)
grp["amount_deviation"] = grp["amount"] / grp["baseline_amount"].clip(lower=1)
grp = grp.drop(columns=["minute_bucket"])
frames.append(grp)
df_out = pd.concat(frames, ignore_index=True)
# Cast feature dtypes cleanly
for c in required_cols:
df_out[c] = df_out[c].astype("float32")
LOGGER.info("Dataset B feature engineering complete. Total columns: %s", len(df_out.columns))
return df_out
def build_and_save_dataset_b_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_b_scenarios.parquet"
if output_path is None:
output_path = root / "data" / "processed" / "dataset_b_features.parquet"
df = pd.read_parquet(input_path)
df_feats = generate_dataset_b_features(df)
df_feats.to_parquet(output_path, index=False)
LOGGER.info("Dataset B features saved to %s", output_path)
return output_path
if __name__ == "__main__":
build_and_save_dataset_b_features()
|