Retail & E-Commerce

┌─────────────────────────────────────────────────────────────────┐
│                    RETAIL CUSTOMER 360 PLATFORM                  │
└─────────────────────────────────────────────────────────────────┘

Source Systems:
┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐
│  POS    │ │E-Commerce│ │ Mobile  │ │ Loyalty │ │Customer │
│ Systems │ │Platform  │ │   App   │ │ Program │ │ Service │
└────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘
     │           │           │           │           │
     └───────────┴───────────┴───────────┴───────────┘
                             │
              ┌──────────────▼──────────────┐
              │     Discovery Agent         │
              │  • Customer touchpoints     │
              │  • Transaction sources      │
              │  • Behavioral signals       │
              └──────────────┬──────────────┘
                             │
              ┌──────────────▼──────────────┐
              │     Identity Resolution     │
              │  • Email matching           │
              │  • Phone matching           │
              │  • Device fingerprinting    │
              │  • Address standardization  │
              └──────────────┬──────────────┘
                             │
              ┌──────────────▼──────────────┐
              │     Golden Customer Record   │
              │  ┌────────────────────────┐  │
              │  │ customer_id            │  │
              │  │ master_email           │  │
              │  │ lifetime_value         │  │
              │  │ preferred_channel      │  │
              │  │ segment                │  │
              │  │ churn_risk_score       │  │
              │  └────────────────────────┘  │
              └──────────────┬──────────────┘
                             │
         ┌───────────────────┼───────────────────┐
         ▼                   ▼                   ▼
   ┌───────────┐      ┌───────────┐      ┌───────────┐
   │Marketing  │      │ In-Store  │      │ Customer  │
   │Automation │      │ Clienteling│     │ Service   │
   └───────────┘      └───────────┘      └───────────┘

-- Retail Customer 360 with Identity Resolution
-- models/gold/dim_customer_360.sql

{{
  config(
    materialized='incremental',
    unique_key='master_customer_id',
    tags=['customer_360', 'retail']
  )
}}

WITH identity_graph AS (
  -- Build identity clusters from all touchpoints
  SELECT
    COALESCE(
      e.customer_id,
      p.customer_id,
      m.customer_id,
      l.member_id
    ) as source_id,
    LOWER(TRIM(COALESCE(e.email, p.email, m.email, l.email))) as email,
    REGEXP_REPLACE(COALESCE(e.phone, p.phone, m.phone, l.phone), '[^0-9]', '') as phone_clean,
    -- Cluster on email first, then phone
    FIRST_VALUE(source_id) OVER (
      PARTITION BY email
      ORDER BY
        CASE
          WHEN e.customer_id IS NOT NULL THEN 1  -- E-commerce is primary
          WHEN l.member_id IS NOT NULL THEN 2    -- Loyalty second
          ELSE 3
        END
    ) as master_customer_id
  FROM {{ ref('silver_ecommerce_customers') }} e
  FULL OUTER JOIN {{ ref('silver_pos_customers') }} p ON e.email = p.email
  FULL OUTER JOIN {{ ref('silver_mobile_users') }} m ON e.email = m.email
  FULL OUTER JOIN {{ ref('silver_loyalty_members') }} l ON e.email = l.email
),

customer_metrics AS (
  SELECT
    ig.master_customer_id,
    -- Lifetime Value
    SUM(t.transaction_total) as lifetime_value,
    COUNT(DISTINCT t.transaction_id) as total_transactions,
    -- Channel Preferences
    MODE(t.channel) as preferred_channel,
    COUNT(DISTINCT t.channel) as channels_used,
    -- Recency
    MAX(t.transaction_date) as last_purchase_date,
    DATEDIFF(day, MAX(t.transaction_date), CURRENT_DATE()) as days_since_purchase,
    -- Frequency
    COUNT(DISTINCT DATE_TRUNC('month', t.transaction_date)) as active_months,
    -- Category Affinity
    ARRAY_AGG(DISTINCT t.category) as category_affinities
  FROM identity_graph ig
  JOIN {{ ref('silver_transactions') }} t ON ig.source_id = t.customer_id
  GROUP BY ig.master_customer_id
)

SELECT
  cm.master_customer_id,
  ig.email as master_email,
  ig.phone_clean as master_phone,
  cm.lifetime_value,
  cm.total_transactions,
  cm.preferred_channel,
  cm.channels_used,
  cm.last_purchase_date,
  cm.days_since_purchase,
  -- Segmentation
  CASE
    WHEN cm.lifetime_value > 5000 AND cm.days_since_purchase < 30 THEN 'VIP_ACTIVE'
    WHEN cm.lifetime_value > 5000 AND cm.days_since_purchase < 90 THEN 'VIP_AT_RISK'
    WHEN cm.lifetime_value > 1000 AND cm.days_since_purchase < 60 THEN 'LOYAL'
    WHEN cm.days_since_purchase < 30 THEN 'ACTIVE'
    WHEN cm.days_since_purchase < 180 THEN 'LAPSING'
    ELSE 'CHURNED'
  END as customer_segment,
  -- Churn Risk Score (0-100)
  LEAST(100, cm.days_since_purchase * 0.5 + (90 - cm.active_months) * 2) as churn_risk_score,
  cm.category_affinities,
  CURRENT_TIMESTAMP() as _updated_at
FROM customer_metrics cm
JOIN identity_graph ig ON cm.master_customer_id = ig.master_customer_id

┌─────────────────────────────────────────────────────────────────┐
│                  INVENTORY & DEMAND PLATFORM                     │
└─────────────────────────────────────────────────────────────────┘

Data Sources:
┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐
│ Warehouse   │  │   Store     │  │  Supplier   │  │  External   │
│ Management  │  │    POS      │  │   Portal    │  │  Signals    │
│   (WMS)     │  │             │  │             │  │(Weather/Cal)│
└──────┬──────┘  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘
       │                │                │                │
       └────────────────┼────────────────┼────────────────┘
                        │
            ┌───────────▼───────────┐
            │   Bronze Layer        │
            │   • Raw inventory     │
            │   • POS transactions  │
            │   • Supplier data     │
            └───────────┬───────────┘
                        │
            ┌───────────▼───────────┐
            │   DQ Profiler Agent   │
            │   • Stock accuracy    │
            │   • Location validity │
            │   • SKU consistency   │
            └───────────┬───────────┘
                        │
            ┌───────────▼───────────┐
            │   Silver Layer        │
            │   • Unified inventory │
            │   • Sales velocity    │
            │   • Lead times        │
            └───────────┬───────────┘
                        │
            ┌───────────▼───────────┐
            │   Gold Layer          │
            │   ┌─────────────────┐ │
            │   │ fact_inventory  │ │
            │   │ • stock_on_hand │ │
            │   │ • days_of_supply│ │
            │   │ • reorder_point │ │
            │   │ • demand_forecast│ │
            │   └─────────────────┘ │
            └───────────┬───────────┘
                        │
         ┌──────────────┼──────────────┐
         ▼              ▼              ▼
   ┌──────────┐  ┌──────────────┐  ┌──────────┐
   │ Replenish│  │   Demand     │  │  Store   │
   │ Planning │  │  Forecasting │  │ Ordering │
   └──────────┘  └──────────────┘  └──────────┘

-- Inventory Position with Demand Sensing
-- models/gold/fact_inventory_position.sql

{{
  config(
    materialized='incremental',
    unique_key='location_sku_date_key',
    partition_by={'field': 'snapshot_date', 'data_type': 'date'},
    cluster_by=['location_id', 'category']
  )
}}

WITH current_inventory AS (
  SELECT
    location_id,
    sku,
    SUM(quantity_on_hand) as stock_on_hand,
    SUM(quantity_in_transit) as in_transit,
    SUM(quantity_reserved) as reserved
  FROM {{ ref('silver_inventory_positions') }}
  WHERE snapshot_date = CURRENT_DATE()
  GROUP BY location_id, sku
),

sales_velocity AS (
  SELECT
    location_id,
    sku,
    -- 7-day rolling average
    AVG(units_sold) OVER (
      PARTITION BY location_id, sku
      ORDER BY sale_date
      ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
    ) as avg_daily_sales_7d,
    -- 28-day rolling average for trend
    AVG(units_sold) OVER (
      PARTITION BY location_id, sku
      ORDER BY sale_date
      ROWS BETWEEN 27 PRECEDING AND CURRENT ROW
    ) as avg_daily_sales_28d,
    -- Day of week patterns
    AVG(units_sold) OVER (
      PARTITION BY location_id, sku, DAYOFWEEK(sale_date)
      ORDER BY sale_date
      ROWS BETWEEN 3 PRECEDING AND CURRENT ROW
    ) as dow_adjusted_demand
  FROM {{ ref('silver_daily_sales') }}
  QUALIFY sale_date = CURRENT_DATE() - 1
),

external_signals AS (
  SELECT
    location_id,
    signal_date,
    weather_impact_factor,
    event_impact_factor,
    promotion_impact_factor
  FROM {{ ref('silver_demand_signals') }}
  WHERE signal_date BETWEEN CURRENT_DATE() AND CURRENT_DATE() + 14
),

demand_forecast AS (
  SELECT
    sv.location_id,
    sv.sku,
    sv.avg_daily_sales_7d,
    -- Adjust forecast with external signals
    sv.avg_daily_sales_7d
      * COALESCE(es.weather_impact_factor, 1.0)
      * COALESCE(es.event_impact_factor, 1.0)
      * COALESCE(es.promotion_impact_factor, 1.0) as adjusted_daily_forecast,
    -- 14-day forward demand
    SUM(sv.avg_daily_sales_7d
      * COALESCE(es.weather_impact_factor, 1.0)
      * COALESCE(es.event_impact_factor, 1.0)) as forecast_14d
  FROM sales_velocity sv
  LEFT JOIN external_signals es ON sv.location_id = es.location_id
  GROUP BY 1, 2, 3, 4
)

SELECT
  {{ dbt_utils.generate_surrogate_key(['ci.location_id', 'ci.sku', 'CURRENT_DATE()']) }}
    as location_sku_date_key,
  CURRENT_DATE() as snapshot_date,
  ci.location_id,
  ci.sku,
  p.category,
  ci.stock_on_hand,
  ci.in_transit,
  ci.reserved,
  ci.stock_on_hand - ci.reserved as available_to_sell,
  df.avg_daily_sales_7d,
  df.adjusted_daily_forecast,
  -- Days of Supply
  CASE
    WHEN df.adjusted_daily_forecast > 0
    THEN ci.stock_on_hand / df.adjusted_daily_forecast
    ELSE 999
  END as days_of_supply,
  -- Stock Status
  CASE
    WHEN ci.stock_on_hand = 0 THEN 'OUT_OF_STOCK'
    WHEN ci.stock_on_hand / NULLIF(df.adjusted_daily_forecast, 0) < 7 THEN 'LOW_STOCK'
    WHEN ci.stock_on_hand / NULLIF(df.adjusted_daily_forecast, 0) > 60 THEN 'OVERSTOCK'
    ELSE 'HEALTHY'
  END as stock_status,
  -- Reorder Signal
  CASE
    WHEN ci.stock_on_hand + ci.in_transit < (df.adjusted_daily_forecast * p.lead_time_days * 1.5)
    THEN TRUE
    ELSE FALSE
  END as reorder_required,
  df.forecast_14d,
  CURRENT_TIMESTAMP() as _loaded_at
FROM current_inventory ci
JOIN {{ ref('dim_product') }} p ON ci.sku = p.sku
LEFT JOIN demand_forecast df ON ci.location_id = df.location_id AND ci.sku = df.sku

┌─────────────────────────────────────────────────────────────────┐
│                  PERSONALIZATION PLATFORM                        │
└─────────────────────────────────────────────────────────────────┘

Behavioral Signals:
┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│   Browse     │ │   Search     │ │   Purchase   │ │   Email      │
│   Clickstream│ │   Queries    │ │   History    │ │   Engagement │
└───────┬──────┘ └───────┬──────┘ └───────┬──────┘ └───────┬──────┘
        │                │                │                │
        └────────────────┴────────────────┴────────────────┘
                                │
                   ┌────────────▼────────────┐
                   │    Event Streaming      │
                   │    (Kafka/Kinesis)      │
                   └────────────┬────────────┘
                                │
               ┌────────────────┼────────────────┐
               ▼                                 ▼
   ┌───────────────────┐              ┌───────────────────┐
   │  Batch Features   │              │  Real-time        │
   │  (Daily refresh)  │              │  Features         │
   │  • LTV score      │              │  • Session intent │
   │  • Category affinity│            │  • Cart context   │
   │  • Segment        │              │  • Recent views   │
   └─────────┬─────────┘              └─────────┬─────────┘
             │                                  │
             └──────────────┬───────────────────┘
                            ▼
               ┌────────────────────────┐
               │     Feature Store      │
               │  ┌──────────────────┐  │
               │  │ customer_features│  │
               │  │ product_features │  │
               │  │ context_features │  │
               │  └──────────────────┘  │
               └────────────┬───────────┘
                            │
         ┌──────────────────┼──────────────────┐
         ▼                  ▼                  ▼
   ┌───────────┐     ┌───────────┐     ┌───────────┐
   │   Web     │     │   Email   │     │   Store   │
   │   Recs    │     │   Content │     │  Clientel │
   └───────────┘     └───────────┘     └───────────┘

-- Customer Feature Store for Personalization
-- models/features/customer_features.sql

{{
  config(
    materialized='incremental',
    unique_key='customer_id',
    tags=['features', 'personalization']
  )
}}

WITH browse_behavior AS (
  SELECT
    customer_id,
    -- Category browsing patterns (last 30 days)
    OBJECT_AGG(
      category,
      TO_VARIANT(OBJECT_CONSTRUCT(
        'views', view_count,
        'time_spent_seconds', total_time,
        'recency_days', days_since_last_view
      ))
    ) as category_browse_scores,
    -- Search intent signals
    ARRAY_AGG(DISTINCT search_term) FILTER (WHERE search_date > CURRENT_DATE() - 7)
      as recent_searches
  FROM {{ ref('silver_browse_sessions') }}
  WHERE session_date > CURRENT_DATE() - 30
  GROUP BY customer_id
),

purchase_patterns AS (
  SELECT
    customer_id,
    -- Category purchase affinity (all time, weighted by recency)
    OBJECT_AGG(
      category,
      TO_VARIANT(OBJECT_CONSTRUCT(
        'purchase_count', purchase_count,
        'total_spent', total_spent,
        'affinity_score', affinity_score
      ))
    ) as category_purchase_affinity,
    -- Price sensitivity
    AVG(CASE WHEN is_discounted THEN 1 ELSE 0 END) as discount_sensitivity,
    PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY item_price) as median_price_point,
    -- Timing patterns
    MODE(DAYOFWEEK(purchase_date)) as preferred_day,
    MODE(HOUR(purchase_timestamp)) as preferred_hour
  FROM {{ ref('silver_transactions') }}
  GROUP BY customer_id
),

engagement_scores AS (
  SELECT
    customer_id,
    -- Email engagement (for content optimization)
    AVG(CASE WHEN opened THEN 1 ELSE 0 END) as email_open_rate,
    AVG(CASE WHEN clicked THEN 1 ELSE 0 END) as email_click_rate,
    -- Content preferences
    ARRAY_AGG(DISTINCT content_type) FILTER (WHERE clicked) as preferred_content_types
  FROM {{ ref('silver_email_engagement') }}
  WHERE sent_date > CURRENT_DATE() - 90
  GROUP BY customer_id
),

product_recommendations AS (
  SELECT
    customer_id,
    -- Collaborative filtering results (pre-computed)
    cf.recommended_skus,
    cf.recommendation_scores,
    -- Recently viewed (for "continue shopping")
    rv.recently_viewed_skus,
    -- Abandoned cart items
    ac.abandoned_skus
  FROM {{ ref('ml_collaborative_filtering') }} cf
  LEFT JOIN {{ ref('silver_recent_views') }} rv USING (customer_id)
  LEFT JOIN {{ ref('silver_abandoned_carts') }} ac USING (customer_id)
)

SELECT
  c.customer_id,
  -- Identity & Segment
  c.customer_segment,
  c.lifetime_value,
  c.churn_risk_score,
  -- Browse Features
  bb.category_browse_scores,
  bb.recent_searches,
  -- Purchase Features
  pp.category_purchase_affinity,
  pp.discount_sensitivity,
  pp.median_price_point,
  pp.preferred_day,
  pp.preferred_hour,
  -- Engagement Features
  es.email_open_rate,
  es.email_click_rate,
  es.preferred_content_types,
  -- Recommendations
  pr.recommended_skus,
  pr.recommendation_scores,
  pr.recently_viewed_skus,
  pr.abandoned_skus,
  -- Feature freshness
  CURRENT_TIMESTAMP() as features_updated_at
FROM {{ ref('dim_customer_360') }} c
LEFT JOIN browse_behavior bb ON c.customer_id = bb.customer_id
LEFT JOIN purchase_patterns pp ON c.customer_id = pp.customer_id
LEFT JOIN engagement_scores es ON c.customer_id = es.customer_id
LEFT JOIN product_recommendations pr ON c.customer_id = pr.customer_id

┌─────────────────────────────────────────────────────────────────┐
│                  STORE OPERATIONS PLATFORM                       │
└─────────────────────────────────────────────────────────────────┘

Data Sources:
┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│   Traffic    │ │   Labor      │ │    POS       │ │  Compliance  │
│   Counters   │ │   System     │ │   Sales      │ │   Scans      │
└───────┬──────┘ └───────┬──────┘ └───────┬──────┘ └───────┬──────┘
        │                │                │                │
        └────────────────┴────────────────┴────────────────┘
                                │
                   ┌────────────▼────────────┐
                   │    Hourly Aggregation   │
                   │    (Bronze → Silver)    │
                   └────────────┬────────────┘
                                │
                   ┌────────────▼────────────┐
                   │   Store Metrics Layer   │
                   │   ┌──────────────────┐  │
                   │   │ Conversion Rate  │  │
                   │   │ Sales per Hour   │  │
                   │   │ Labor Efficiency │  │
                   │   │ Basket Size      │  │
                   │   └──────────────────┘  │
                   └────────────┬────────────┘
                                │
         ┌──────────────────────┼──────────────────────┐
         ▼                      ▼                      ▼
   ┌───────────┐         ┌───────────┐         ┌───────────┐
   │   Labor   │         │   Store   │         │  District │
   │ Scheduling│         │ Dashboards│         │ Comparison│
   └───────────┘         └───────────┘         └───────────┘

-- Store Operations Metrics
-- models/gold/fact_store_hourly_performance.sql

{{
  config(
    materialized='incremental',
    unique_key='store_hour_key',
    partition_by={'field': 'business_date', 'data_type': 'date'}
  )
}}

WITH traffic AS (
  SELECT
    store_id,
    business_date,
    hour_of_day,
    SUM(entries) as hourly_traffic,
    AVG(dwell_time_seconds) as avg_dwell_time
  FROM {{ ref('silver_traffic_counts') }}
  WHERE business_date >= CURRENT_DATE() - 90
  GROUP BY 1, 2, 3
),

sales AS (
  SELECT
    store_id,
    DATE(transaction_timestamp) as business_date,
    HOUR(transaction_timestamp) as hour_of_day,
    COUNT(DISTINCT transaction_id) as transactions,
    SUM(transaction_total) as hourly_sales,
    SUM(item_count) as units_sold,
    COUNT(DISTINCT customer_id) as unique_customers
  FROM {{ ref('silver_transactions') }}
  WHERE DATE(transaction_timestamp) >= CURRENT_DATE() - 90
  GROUP BY 1, 2, 3
),

labor AS (
  SELECT
    store_id,
    work_date as business_date,
    hour_of_day,
    SUM(scheduled_hours) as scheduled_labor_hours,
    SUM(actual_hours) as actual_labor_hours,
    COUNT(DISTINCT employee_id) as staff_count
  FROM {{ ref('silver_labor_hours') }}
  WHERE work_date >= CURRENT_DATE() - 90
  GROUP BY 1, 2, 3
)

SELECT
  {{ dbt_utils.generate_surrogate_key(['t.store_id', 't.business_date', 't.hour_of_day']) }}
    as store_hour_key,
  t.store_id,
  t.business_date,
  t.hour_of_day,
  s.store_name,
  s.region,
  s.district,
  -- Traffic Metrics
  t.hourly_traffic,
  t.avg_dwell_time,
  -- Sales Metrics
  COALESCE(sl.transactions, 0) as transactions,
  COALESCE(sl.hourly_sales, 0) as hourly_sales,
  COALESCE(sl.units_sold, 0) as units_sold,
  -- Conversion
  CASE
    WHEN t.hourly_traffic > 0
    THEN sl.transactions::FLOAT / t.hourly_traffic
    ELSE 0
  END as conversion_rate,
  -- Basket Metrics
  CASE
    WHEN sl.transactions > 0
    THEN sl.hourly_sales / sl.transactions
    ELSE 0
  END as average_transaction_value,
  CASE
    WHEN sl.transactions > 0
    THEN sl.units_sold::FLOAT / sl.transactions
    ELSE 0
  END as average_basket_size,
  -- Labor Metrics
  COALESCE(l.scheduled_labor_hours, 0) as scheduled_hours,
  COALESCE(l.actual_labor_hours, 0) as actual_hours,
  COALESCE(l.staff_count, 0) as staff_count,
  -- Labor Efficiency
  CASE
    WHEN l.actual_labor_hours > 0
    THEN sl.hourly_sales / l.actual_labor_hours
    ELSE 0
  END as sales_per_labor_hour,
  CASE
    WHEN l.actual_labor_hours > 0
    THEN sl.transactions::FLOAT / l.actual_labor_hours
    ELSE 0
  END as transactions_per_labor_hour,
  -- Optimal Staffing Indicator
  CASE
    WHEN t.hourly_traffic / NULLIF(l.staff_count, 0) > 50 THEN 'UNDERSTAFFED'
    WHEN t.hourly_traffic / NULLIF(l.staff_count, 0) < 15 THEN 'OVERSTAFFED'
    ELSE 'OPTIMAL'
  END as staffing_status,
  CURRENT_TIMESTAMP() as _loaded_at
FROM traffic t
JOIN {{ ref('dim_store') }} s ON t.store_id = s.store_id
LEFT JOIN sales sl ON t.store_id = sl.store_id
  AND t.business_date = sl.business_date
  AND t.hour_of_day = sl.hour_of_day
LEFT JOIN labor l ON t.store_id = l.store_id
  AND t.business_date = l.business_date
  AND t.hour_of_day = l.hour_of_day