Pricing Analytics
Architecture Guide

-- Staging model: enforces baseline/promotional separation
-- All volume classified before entering the elasticity layer
WITH raw_pos AS (
  SELECT
    sku_id, retailer_id, channel, week_start_date,
    units_sold, net_revenue, avg_shelf_price,
    COALESCE(tc.is_promoted, FALSE) AS is_promotional_period,
    tc.promo_depth_pct
  FROM raw.pos_weekly_sales pos
  LEFT JOIN raw.trade_calendar tc
    ON  pos.sku_id = tc.sku_id
    AND pos.week_start_date BETWEEN tc.promo_start AND tc.promo_end
),
classified_volume AS (
  SELECT *,
    CASE WHEN is_promotional_period THEN 0 ELSE units_sold END AS baseline_units,
    CASE WHEN is_promotional_period THEN units_sold ELSE 0 END AS promo_units
  FROM raw_pos
)
SELECT * FROM classified_volume

class CrossSKUElasticityModel:
    """Governed model: separate everyday vs. promotional coefficients,
    cross-SKU substitution, market signal modulation, full lineage."""

    def fit(self, df: pd.DataFrame):
        # CRITICAL: fit separately on baseline vs. promotional periods
        baseline_df = df[df['promotional_units'] == 0]
        promo_df    = df[df['promotional_units'] > 0]

        for sku_id, group in baseline_df.groupby('sku_id'):
            X, y = self._build_features(group, mode='everyday')
            self.everyday_coef[sku_id] = self.model.fit(X, y).coef_[0]

    def scenario(self, sku_id, price_delta_pct, market_signals):
        """Returns predicted volume impact with full parameter lineage."""
        e = self.everyday_coef.get(sku_id, -1.5)
        return {
            "own_elasticity":     round(e, 3),
            "volume_impact_pct":  round(e * (price_delta_pct / 100) * 100, 2),
            "market_signal_inputs": market_signals,  # lineage
            "model_version":      "governed_v1"
        }