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()