feat(features): implement promotion compute method (#109) (#112)

Author: w7-mgfcode

app/features/featuresets/service.py

@@ -138,6 +138,26 @@ def compute_features(
             result, cols = self._compute_lifecycle_features(result)
             feature_columns.extend(cols)
 
+        # 7. Promotion features (PRP-3.1D — Phase 2)
+        if self.config.promotion_config:
+            promotion_rows_df = getattr(self, "_promotion_rows_df", None)
+            if promotion_rows_df is None:
+                # PRP-3.1E wires the DB JOIN that sets this attribute.
+                # In unit tests, the test sets it directly on the service.
+                # An empty DataFrame is the safe no-op fallback.
+                promotion_rows_df = pd.DataFrame(
+                    columns=[
+                        "product_id",
+                        "store_id",
+                        "kind",
+                        "discount_pct",
+                        "start_date",
+                        "end_date",
+                    ]
+                )
+            result, cols = self._compute_promotion_features(result, promotion_rows_df)
+            feature_columns.extend(cols)
+
         # Compute stats
         null_counts: dict[str, int] = {}
         if feature_columns:
@@ -491,6 +511,116 @@ def _compute_lifecycle_features(self, df: pd.DataFrame) -> tuple[pd.DataFrame, l
 
         return result, columns
 
+    def _compute_promotion_features(
+        self,
+        df: pd.DataFrame,
+        promotion_rows_df: pd.DataFrame,
+    ) -> tuple[pd.DataFrame, list[str]]:
+        """Compute promotion-family features (active + intensity per kind).
+
+        CRITICAL: Time-safe via ``groupby(entity_cols).shift(lag_days)`` on
+        a daily-grain indicator. Per the time-safety contract, the active
+        indicator at row D reads activity from D - lag_days. A promotion
+        active on D itself must NOT appear in active_lag{N} at D.
+
+        Date-range semantics: ``start_date <= D <= end_date`` (both inclusive).
+
+        Chain-wide promotions: rows with ``store_id`` NaN/None apply to
+        EVERY store of that product. Handled via a two-pass match (store-
+        specific OR chain-wide), never via a NaN-key merge.
+
+        Overlapping promotions on the same kind/day reduce via ``max`` over
+        ``discount_pct`` for intensity (Decision §15-C); active stays 0/1.
+
+        Args:
+            df: Sales DataFrame, pre-sorted and cutoff-filtered (per
+                compute_features pipeline).
+            promotion_rows_df: Promotion rows. Columns:
+                ``[product_id, store_id, kind, discount_pct, start_date, end_date]``.
+                ``store_id`` may be NaN (chain-wide). ``discount_pct`` may
+                be NaN (bogo / bundle kinds).
+
+        Returns:
+            Tuple of (DataFrame with new columns, list of new column names).
+        """
+        config = self.config.promotion_config
+        if config is None:
+            raise RuntimeError("_compute_promotion_features called without promotion_config")
+
+        result = df.copy()
+        columns: list[str] = []
+        lag = config.lag_days
+
+        # Defensive re-sort to match the caller invariant.
+        result = result.sort_values([*self.entity_cols, self.date_col])
+        dates = pd.to_datetime(result[self.date_col]).dt.date
+
+        # Deterministic column ordering: sorted kinds, active before intensity.
+        sorted_kinds: tuple[str, ...] = tuple(sorted(config.kinds_to_track))
+
+        for kind in sorted_kinds:
+            kind_rows = promotion_rows_df[promotion_rows_df["kind"] == kind]
+
+            # Per-row daily indicators (D-day truth, BEFORE lag shift).
+            active_today: pd.Series[Any] = pd.Series(0, index=result.index, dtype="int64")
+            intensity_today: pd.Series[Any] = pd.Series(np.nan, index=result.index, dtype="float64")
+
+            # Two-pass match: store-specific then chain-wide. Never merge on NaN keys.
+            store_specific = kind_rows[kind_rows["store_id"].notna()]
+            chain_wide = kind_rows[kind_rows["store_id"].isna()]
+
+            for _, promo in store_specific.iterrows():
+                mask = (
+                    (result["store_id"] == promo["store_id"])
+                    & (result["product_id"] == promo["product_id"])
+                    & (dates >= promo["start_date"])
+                    & (dates <= promo["end_date"])
+                )
+                active_today = active_today.where(~mask, 1)
+                disc = promo["discount_pct"]
+                if pd.notna(disc):
+                    # Overlapping-on-same-kind reduction = max (Decision §15-C).
+                    masked_disc = intensity_today.where(~mask, float(disc))
+                    intensity_today = pd.concat([intensity_today, masked_disc], axis=1).max(axis=1)
+
+            for _, promo in chain_wide.iterrows():
+                mask = (
+                    (result["product_id"] == promo["product_id"])
+                    & (dates >= promo["start_date"])
+                    & (dates <= promo["end_date"])
+                )
+                active_today = active_today.where(~mask, 1)
+                disc = promo["discount_pct"]
+                if pd.notna(disc):
+                    masked_disc = intensity_today.where(~mask, float(disc))
+                    intensity_today = pd.concat([intensity_today, masked_disc], axis=1).max(axis=1)
+
+            # CRITICAL: groupby(entity_cols).shift(lag) — the leakage gate.
+            # Feature at row D reads daily indicator at D - lag.
+            if config.include_active:
+                col = f"promo_{kind}_active_lag{lag}"
+                shifted_active = (
+                    result.assign(_a=active_today)
+                    .groupby(self.entity_cols, observed=True)["_a"]
+                    .shift(lag)
+                )
+                # Nullable Int64 preserves NaN at the start of each series
+                # (mirrors the lag-feature idiom — Decision §15-D).
+                result[col] = shifted_active.astype("Int64")
+                columns.append(col)
+
+            if config.include_intensity:
+                col = f"promo_{kind}_intensity_lag{lag}"
+                shifted_intensity = (
+                    result.assign(_i=intensity_today)
+                    .groupby(self.entity_cols, observed=True)["_i"]
+                    .shift(lag)
+                )
+                result[col] = shifted_intensity.astype("float64")
+                columns.append(col)
+
+        return result, columns
+
 
 class FeatureDataLoader:
     """Async data loader for feature computation.

app/features/featuresets/tests/test_leakage.py

@@ -13,6 +13,7 @@
     FeatureSetConfig,
     LagConfig,
     LifecycleConfig,
+    PromotionConfig,
     RollingConfig,
 )
 from app.features.featuresets.service import FeatureEngineeringService
@@ -418,3 +419,155 @@ def test_lifecycle_group_isolation_no_cross_product_leakage(
                     f"days_since_launch_lag1={actual}, expected={base_lag}. "
                     "Lifecycle lag is mixing across products."
                 )
+
+
+class TestPromotionLeakage:
+    """Tests verifying promotion features never use future data.
+
+    PRP-3.1D — these leakage cases are LOAD-BEARING. They assert that a
+    promotion active on day D MUST NOT appear in day D's
+    ``promo_<kind>_active_lag1`` column; it appears at day D+1 only. The
+    date-range semantics (start_date <= D <= end_date, both inclusive)
+    plus ``groupby(...).shift(lag_days)`` are the mathematical leakage gate.
+    """
+
+    def test_promotion_active_no_leakage_at_same_day(
+        self,
+        sample_time_series: pd.DataFrame,
+        phase2_promotion_rows_df: pd.DataFrame,
+    ) -> None:
+        """CRITICAL: A promotion active on day D MUST NOT appear in lag1 at D."""
+        config = FeatureSetConfig(
+            name="test",
+            entity_columns=("store_id", "product_id"),
+            promotion_config=PromotionConfig(
+                kinds_to_track=("markdown",),
+                include_active=True,
+                include_intensity=False,
+                lag_days=1,
+            ),
+        )
+        service = FeatureEngineeringService(config)
+        service._promotion_rows_df = phase2_promotion_rows_df  # type: ignore[attr-defined]
+        result = service.compute_features(sample_time_series)
+
+        # The fixture's markdown is active 2024-01-07 .. 2024-01-14 (8 days).
+        # promo_markdown_active_lag1 should be 1 on 2024-01-08 .. 2024-01-15.
+        df = result.df.reset_index(drop=True)
+        dates = pd.to_datetime(df["date"]).dt.date
+
+        # Day BEFORE start (D=Jan 6): lag1 reads Jan 5 — inactive. EXPECT 0.
+        assert df.loc[dates == date(2024, 1, 6), "promo_markdown_active_lag1"].iloc[0] == 0
+
+        # Day OF start (D=Jan 7): lag1 reads Jan 6 — inactive. EXPECT 0.
+        #   This is the load-bearing leakage check: same-day MUST NOT leak.
+        assert df.loc[dates == date(2024, 1, 7), "promo_markdown_active_lag1"].iloc[0] == 0, (
+            "LEAKAGE DETECTED: promo active on day D appeared in active_lag1 at day D"
+        )
+
+        # Day AFTER start (D=Jan 8): lag1 reads Jan 7 — active. EXPECT 1.
+        assert df.loc[dates == date(2024, 1, 8), "promo_markdown_active_lag1"].iloc[0] == 1
+
+        # Day AFTER end (D=Jan 15): lag1 reads Jan 14 — last active day. EXPECT 1.
+        assert df.loc[dates == date(2024, 1, 15), "promo_markdown_active_lag1"].iloc[0] == 1
+
+        # Two days AFTER end (D=Jan 16): lag1 reads Jan 15 — inactive. EXPECT 0.
+        assert df.loc[dates == date(2024, 1, 16), "promo_markdown_active_lag1"].iloc[0] == 0
+
+    def test_promotion_boundary_end_date_at_cutoff(
+        self,
+        sample_time_series: pd.DataFrame,
+    ) -> None:
+        """A promo ending exactly on cutoff_date - 1 yields active_lag1=1 at cutoff."""
+        cutoff = date(2024, 1, 15)
+        promo_rows = pd.DataFrame(
+            {
+                "product_id": [1],
+                "store_id": [1],
+                "kind": ["markdown"],
+                "discount_pct": [0.20],
+                "start_date": [date(2024, 1, 10)],
+                "end_date": [date(2024, 1, 14)],  # cutoff - 1
+            }
+        )
+        config = FeatureSetConfig(
+            name="test",
+            entity_columns=("store_id", "product_id"),
+            promotion_config=PromotionConfig(kinds_to_track=("markdown",), lag_days=1),
+        )
+        service = FeatureEngineeringService(config)
+        service._promotion_rows_df = promo_rows  # type: ignore[attr-defined]
+        result = service.compute_features(sample_time_series, cutoff_date=cutoff)
+
+        df = result.df.reset_index(drop=True)
+        dates = pd.to_datetime(df["date"]).dt.date
+        # At cutoff (Jan 15), lag1 reads Jan 14 — end_date, INCLUSIVE → active.
+        last = df.loc[dates == cutoff].iloc[0]
+        assert last["promo_markdown_active_lag1"] == 1, (
+            "Boundary leakage: end_date INCLUSIVE on the previous day failed"
+        )
+
+    def test_promotion_starts_on_cutoff_not_in_lag1(
+        self,
+        sample_time_series: pd.DataFrame,
+    ) -> None:
+        """A promo starting exactly on cutoff is NOT in active_lag1 at cutoff."""
+        cutoff = date(2024, 1, 15)
+        promo_rows = pd.DataFrame(
+            {
+                "product_id": [1],
+                "store_id": [1],
+                "kind": ["markdown"],
+                "discount_pct": [0.20],
+                "start_date": [cutoff],  # starts today
+                "end_date": [date(2024, 1, 25)],
+            }
+        )
+        config = FeatureSetConfig(
+            name="test",
+            entity_columns=("store_id", "product_id"),
+            promotion_config=PromotionConfig(kinds_to_track=("markdown",), lag_days=1),
+        )
+        service = FeatureEngineeringService(config)
+        service._promotion_rows_df = promo_rows  # type: ignore[attr-defined]
+        result = service.compute_features(sample_time_series, cutoff_date=cutoff)
+
+        df = result.df.reset_index(drop=True)
+        dates = pd.to_datetime(df["date"]).dt.date
+        last = df.loc[dates == cutoff].iloc[0]
+        # lag1 reads cutoff - 1 = Jan 14, BEFORE start_date.
+        assert last["promo_markdown_active_lag1"] == 0, (
+            "Same-day leakage: promo starting on D appeared in active_lag1 at D"
+        )
+
+    def test_chain_wide_promo_does_not_bleed_across_products(
+        self,
+        multi_series_time_series: pd.DataFrame,
+    ) -> None:
+        """A chain-wide promo on product=1 must NOT activate features for product=2."""
+        promo_rows = pd.DataFrame(
+            {
+                "product_id": [1],
+                "store_id": [None],  # chain-wide
+                "kind": ["markdown"],
+                "discount_pct": [0.30],
+                "start_date": [date(2024, 1, 3)],
+                "end_date": [date(2024, 1, 7)],
+            }
+        )
+        config = FeatureSetConfig(
+            name="test",
+            entity_columns=("store_id", "product_id"),
+            promotion_config=PromotionConfig(kinds_to_track=("markdown",), lag_days=1),
+        )
+        service = FeatureEngineeringService(config)
+        service._promotion_rows_df = promo_rows  # type: ignore[attr-defined]
+        result = service.compute_features(multi_series_time_series)
+
+        df = result.df
+        # Product 1 should see activity 2024-01-04 .. 2024-01-08 (lag1) -- 5 days x 2 stores.
+        prod1 = df[df["product_id"] == 1]
+        assert int(prod1["promo_markdown_active_lag1"].sum()) == 5 * 2
+        # Product 2 should see ZERO activity (chain-wide is product-scoped).
+        prod2 = df[df["product_id"] == 2]
+        assert int(prod2["promo_markdown_active_lag1"].sum()) == 0