main.py

"""
Purchase propensity → GA4 Measurement Protocol sync
===================================================
Runs daily at 06:00 via Cloud Scheduler. For each user active in the last 24h:
  1. Build behavioral features from the last 30 days of GA4 export data
  2. Score them with the BQML model trained weekly on Sundays
  3. Push the propensity bucket as a GA4 user property via Measurement Protocol

The user property feeds GA4 audiences, which sync to Google Ads for remarketing.
Only consented users are scored. Recent buyers (last 30 days) are excluded.

Pipeline:
  Sunday 00:01 → training.sql rebuilds the model in BigQuery
  Daily 06:00  → this function scores active users and pushes to GA4

Required environment variables:
  GCP_PROJECT          GCP project ID
  GA4_EXPORT_DATASET   e.g. analytics_123456789
  ML_DATASET           e.g. ga4_ml
  GA4_MEASUREMENT_ID   e.g. G-XXXXXXXXXX
  GA4_API_SECRET       Measurement Protocol API secret

Optional environment variables:
  GA4_TIMEZONE         IANA timezone matching GA4 property (default: UTC)
                       e.g. "Europe/Stockholm", "America/New_York"
  MAX_BYTES_BILLED     BQ query cost cap in bytes (default: 10 GB)
  MODEL_STALE_DAYS     Warn if model is older than N days (default: 10)
  DEBUG_VALIDATE_RATE  Fraction of payloads validated against GA4 debug endpoint (default: 0.01)
"""

import logging
import os
import random
import statistics
import time
from concurrent.futures import ThreadPoolExecutor
import requests
from google.cloud import bigquery
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# --- Configuration -----------------------------------------------------------
# All config loaded from environment variables. Required vars use os.environ[]
# which fails fast at cold-start if missing — better than discovering it later.

PROJECT        = os.environ["GCP_PROJECT"]
GA4_DATASET    = os.environ["GA4_EXPORT_DATASET"]
ML_DATASET     = os.environ["ML_DATASET"]
MEASUREMENT_ID = os.environ["GA4_MEASUREMENT_ID"]
API_SECRET     = os.environ["GA4_API_SECRET"]

# Timezone must match GA4 property's reporting timezone. The events_intraday_*
# table suffix follows the GA4 property timezone, not UTC, so without this
# we'd query the wrong table near midnight.
GA4_TIMEZONE        = os.environ.get("GA4_TIMEZONE", "UTC")
MAX_BYTES_BILLED    = int(os.environ.get("MAX_BYTES_BILLED", 10 * 1024**3))  # 10 GB safety cap
MODEL_STALE_DAYS    = int(os.environ.get("MODEL_STALE_DAYS", 10))
DEBUG_VALIDATE_RATE = float(os.environ.get("DEBUG_VALIDATE_RATE", 0.01))

# Production endpoint silently accepts malformed payloads (always returns 204).
# Debug endpoint returns actual validation errors as JSON — we sample 1% of
# payloads against it to catch silent failures.
MP_URL       = f"https://www.google-analytics.com/mp/collect?measurement_id={MEASUREMENT_ID}&api_secret={API_SECRET}"
MP_DEBUG_URL = f"https://www.google-analytics.com/debug/mp/collect?measurement_id={MEASUREMENT_ID}&api_secret={API_SECRET}"

# Cloud Logging picks up severity levels automatically from the logging module.
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
log = logging.getLogger("propensity-sync")

# Module-level client is reused across invocations (warm starts) and across
# the 25 worker threads — connection pooling is automatic.
bq = bigquery.Client()

# HTTP session with automatic retry on transient failures.
# - 3 retries on 429 (rate limit) and 5xx (server errors)
# - Backoff factor 1 means delays of 1s, 2s, 4s between attempts
# - allowed_methods=["POST"] because MP is POST-only
_session = requests.Session()
_session.mount("https://", HTTPAdapter(max_retries=Retry(
    total=3,
    backoff_factor=1,
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["POST"],
)))

# Three-bucket scheme. Thresholds chosen to match model calibration —
# revisit these after first ML.EVALUATE run if distribution is skewed.
# Order matters: thresholds checked top-down, first match wins.
BUCKETS = [(0.70, "high"), (0.40, "mid")]
def bucket(p: float) -> str:
    """Map a probability score (0-1) to a propensity bucket label."""
    for threshold, label in BUCKETS:
        if p >= threshold:
            return label
    return "low"

# --- SQL ---------------------------------------------------------------------
# Feature definitions here MUST match training.sql exactly. If you add or
# change a feature, update both files in lockstep — ML.PREDICT will fail on
# schema mismatch.
#
# Differences from training SQL (intentional):
#   - Scoring uses last 30 days of features (fresh data for prediction)
#   - Training uses 180 days of features + 30 days of labels (more positives)
#   - Scoring filters to "active in last 24 hours" (cost optimization)
#   - Scoring filters to "consented users" (legal requirement for ad push)
#
# Within-feature-window recency slices (engagement_last_7d, engagement_last_30d):
#   - In training: last 7d/30d of the 180-day feature window
#   - In scoring:  last 7d/30d of the 30-day feature window (so 30d == window)
#   - This means at scoring time engagement_last_30d_seconds equals
#     total_engagement_seconds. That's expected — the model uses them as
#     separate signals during training; keep both for schema parity.

SQL = f"""
WITH
  -- Users who already bought in the last 30 days: exclude entirely.
  -- No point spending compute or ad budget retargeting recent customers.
  recent_buyers AS (
    SELECT DISTINCT user_pseudo_id
    FROM `{PROJECT}.{GA4_DATASET}.events_*`
    WHERE _TABLE_SUFFIX BETWEEN FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE('{GA4_TIMEZONE}'), INTERVAL 30 DAY))
                            AND FORMAT_DATE('%Y%m%d', CURRENT_DATE('{GA4_TIMEZONE}'))
      AND event_name = 'purchase'
  ),

  -- Users active in the last 24 hours who have granted ads personalization consent.
  -- This is the cost-control hook: scanning the relevant date suffixes is cheap;
  -- joining to 30-day features only for these users keeps total scan small.
  --
  -- We UNION both events_intraday_* and events_* because at 06:00 yesterday's data
  -- may live in either table depending on when GA4 finalized the daily export.
  -- Wildcards safely skip missing table suffixes, so this works in all states.
  active AS (
    SELECT DISTINCT user_pseudo_id FROM (
      SELECT user_pseudo_id, event_timestamp, event_params
      FROM `{PROJECT}.{GA4_DATASET}.events_*`
      WHERE _TABLE_SUFFIX BETWEEN FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE('{GA4_TIMEZONE}'), INTERVAL 1 DAY))
                              AND FORMAT_DATE('%Y%m%d', CURRENT_DATE('{GA4_TIMEZONE}'))
      UNION ALL
      SELECT user_pseudo_id, event_timestamp, event_params
      FROM `{PROJECT}.{GA4_DATASET}.events_intraday_*`
      WHERE _TABLE_SUFFIX BETWEEN FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE('{GA4_TIMEZONE}'), INTERVAL 1 DAY))
                              AND FORMAT_DATE('%Y%m%d', CURRENT_DATE('{GA4_TIMEZONE}'))
    )
    WHERE TIMESTAMP_MICROS(event_timestamp) >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 24 HOUR)
      AND (SELECT value.string_value FROM UNNEST(event_params)
           WHERE key = 'ads_personalization_consent') = 'GRANTED'
      AND user_pseudo_id NOT IN (SELECT user_pseudo_id FROM recent_buyers)
  ),

  -- Per-session aggregates enable avg/max session features.
  -- (user_pseudo_id, ga_session_id) is the unique session key; ga_session_id
  -- alone is a session-start unix timestamp and can collide across users.
  sessions AS (
    SELECT
      user_pseudo_id,
      (SELECT value.int_value FROM UNNEST(event_params) WHERE key='ga_session_id') AS session_id,
      MAX(IFNULL((SELECT value.int_value FROM UNNEST(event_params) WHERE key='session_engaged'),0)) AS engaged,
      SUM(IFNULL((SELECT value.int_value FROM UNNEST(event_params) WHERE key='engagement_time_msec'),0))/1000 AS session_engagement_seconds,
      COUNT(*) AS events_in_session
    FROM `{PROJECT}.{GA4_DATASET}.events_*`
    WHERE _TABLE_SUFFIX BETWEEN FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE('{GA4_TIMEZONE}'), INTERVAL 30 DAY))
                            AND FORMAT_DATE('%Y%m%d', CURRENT_DATE('{GA4_TIMEZONE}'))
      AND user_pseudo_id IN (SELECT user_pseudo_id FROM active)
      AND (SELECT value.int_value FROM UNNEST(event_params) WHERE key='ga_session_id') IS NOT NULL
    GROUP BY user_pseudo_id, session_id
  ),

  user_sessions AS (
    SELECT
      user_pseudo_id,
      COUNT(*) AS sessions_total,
      COUNTIF(engaged = 1) AS engaged_sessions,
      SAFE_DIVIDE(COUNTIF(engaged = 1), COUNT(*)) AS engaged_session_rate,
      AVG(events_in_session) AS avg_session_depth,
      AVG(session_engagement_seconds) AS avg_session_engagement_seconds,
      MAX(session_engagement_seconds) AS max_session_engagement_seconds
    FROM sessions
    GROUP BY user_pseudo_id
  ),

  -- Per-user event aggregates including recency-windowed engagement.
  -- 7d and 30d windows are within the 30-day scoring window — at scoring
  -- time engagement_last_30d_seconds == total_engagement_seconds. The model
  -- still uses both because training learned weights on a 180-day window
  -- where they differ.
  user_events AS (
    SELECT
      user_pseudo_id,

      -- Engagement: total, last 30d, last 7d
      SUM(IFNULL((SELECT value.int_value FROM UNNEST(event_params) WHERE key='engagement_time_msec'),0))/1000 AS total_engagement_seconds,
      SUM(IF(DATE(TIMESTAMP_MICROS(event_timestamp), '{GA4_TIMEZONE}') >= DATE_SUB(CURRENT_DATE('{GA4_TIMEZONE}'), INTERVAL 7 DAY),
             IFNULL((SELECT value.int_value FROM UNNEST(event_params) WHERE key='engagement_time_msec'),0),
             0))/1000 AS engagement_last_7d_seconds,
      SUM(IF(DATE(TIMESTAMP_MICROS(event_timestamp), '{GA4_TIMEZONE}') >= DATE_SUB(CURRENT_DATE('{GA4_TIMEZONE}'), INTERVAL 30 DAY),
             IFNULL((SELECT value.int_value FROM UNNEST(event_params) WHERE key='engagement_time_msec'),0),
             0))/1000 AS engagement_last_30d_seconds,

      -- Categorical features with NULL sentinel for stable encoding
      COALESCE(ANY_VALUE(device.category), '(none)')         AS device_category,
      COALESCE(ANY_VALUE(device.operating_system), '(none)') AS os,
      COALESCE(ANY_VALUE(geo.country), '(none)')             AS country,
      COALESCE(ANY_VALUE(traffic_source.medium), '(none)')   AS traffic_medium,
      COALESCE(ANY_VALUE(traffic_source.source), '(none)')   AS traffic_source,

      -- Full standard GA4 ecommerce funnel
      COUNTIF(event_name='view_item_list')    AS view_item_list_count,
      COUNTIF(event_name='select_item')       AS select_item_count,
      COUNTIF(event_name='view_item')         AS view_item_count,
      COUNTIF(event_name='add_to_wishlist')   AS add_to_wishlist_count,
      COUNTIF(event_name='add_to_cart')       AS add_to_cart_count,
      COUNTIF(event_name='view_cart')         AS view_cart_count,
      COUNTIF(event_name='remove_from_cart')  AS remove_from_cart_count,
      COUNTIF(event_name='begin_checkout')    AS begin_checkout_count,
      COUNTIF(event_name='add_shipping_info') AS add_shipping_info_count,
      COUNTIF(event_name='add_payment_info')  AS add_payment_info_count,

      -- Page-level activity
      COUNTIF(event_name='page_view') AS page_view_count,
      COUNT(*) AS total_events,

      -- Recency
      DATE_DIFF(CURRENT_DATE('{GA4_TIMEZONE}'),
                MAX(DATE(TIMESTAMP_MICROS(event_timestamp), '{GA4_TIMEZONE}')),
                DAY) AS days_since_last_visit

    FROM `{PROJECT}.{GA4_DATASET}.events_*`
    WHERE _TABLE_SUFFIX BETWEEN FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE('{GA4_TIMEZONE}'), INTERVAL 30 DAY))
                            AND FORMAT_DATE('%Y%m%d', CURRENT_DATE('{GA4_TIMEZONE}'))
      AND user_pseudo_id IN (SELECT user_pseudo_id FROM active)
    GROUP BY user_pseudo_id
  ),

  -- Funnel ratios — usually the strongest single predictors because they
  -- normalize for activity volume and capture pure intent.
  feats AS (
    SELECT
      ue.* EXCEPT(user_pseudo_id),
      us.sessions_total,
      us.engaged_sessions,
      us.engaged_session_rate,
      us.avg_session_depth,
      us.avg_session_engagement_seconds,
      us.max_session_engagement_seconds,
      SAFE_DIVIDE(ue.view_item_count,         NULLIF(ue.view_item_list_count, 0))  AS list_to_view_ratio,
      SAFE_DIVIDE(ue.add_to_cart_count,       NULLIF(ue.view_item_count, 0))       AS view_to_cart_ratio,
      SAFE_DIVIDE(ue.begin_checkout_count,    NULLIF(ue.add_to_cart_count, 0))     AS cart_to_checkout_ratio,
      SAFE_DIVIDE(ue.add_payment_info_count,  NULLIF(ue.begin_checkout_count, 0))  AS checkout_to_payment_ratio,
      SAFE_DIVIDE(ue.remove_from_cart_count,  NULLIF(ue.add_to_cart_count, 0))     AS cart_abandon_ratio,
      ue.user_pseudo_id
    FROM user_events ue
    LEFT JOIN user_sessions us USING(user_pseudo_id)
  ),

  -- Run BQML inference. predicted_will_convert_probs is an array of
  -- {{label, prob}} pairs — we extract the probability of class 1 (will convert).
  scored AS (
    SELECT
      user_pseudo_id,
      (SELECT prob FROM UNNEST(predicted_will_convert_probs) WHERE label = 1) AS p
    FROM ML.PREDICT(MODEL `{PROJECT}.{ML_DATASET}.purchase_propensity`, TABLE feats)
  )
-- Filter NULLs defensively: a corrupted model could return NULL probabilities,
-- which would crash bucket() with TypeError. Better to drop them silently.
SELECT user_pseudo_id, p FROM scored WHERE p IS NOT NULL
"""

# Separate query to check when the model was last trained. If the weekly
# Sunday job fails silently, we'd keep scoring with an increasingly stale
# model. This catches that within MODEL_STALE_DAYS.
MODEL_FRESHNESS_SQL = f"""
SELECT TIMESTAMP_DIFF(CURRENT_TIMESTAMP(), last_modified_time, DAY) AS age_days
FROM `{PROJECT}.{ML_DATASET}.INFORMATION_SCHEMA.MODELS`
WHERE model_name = 'purchase_propensity'
"""

# --- Helpers -----------------------------------------------------------------

def check_model_freshness() -> None:
    """
    Log model age. Warns if older than MODEL_STALE_DAYS, errors if missing.
    Doesn't block the run — a stale model is still better than no scoring.
    The warning gets picked up by Cloud Monitoring for alerting.
    """
    try:
        row = next(iter(bq.query(MODEL_FRESHNESS_SQL).result()), None)
        if row is None:
            log.error("Model not found in INFORMATION_SCHEMA — has training ever run?")
            return
        if row.age_days > MODEL_STALE_DAYS:
            log.warning("Model is %d days old (threshold: %d) — training job may be failing",
                        row.age_days, MODEL_STALE_DAYS)
        else:
            log.info("Model age: %d days", row.age_days)
    except Exception as e:
        log.warning("Model freshness check failed: %s", e)

def build_payload(user_pseudo_id: str, p: float) -> dict:
    """
    Construct a Measurement Protocol payload for one user.

    The payload has two semantically distinct parts:
      - user_properties: the persistent propensity bucket. This is what
        GA4 audiences and Google Ads remarketing use.
      - events: required by the MP endpoint (it rejects payloads without
        at least one event). We piggyback the score and bucket as event
        params so they're queryable as custom dimensions/metrics in GA4
        for model performance analysis.

    non_personalized_ads=False asserts that this user has consented to
    ad personalization — which is true here because the SQL already filtered
    to ads_personalization_consent = 'GRANTED'.
    """
    b = bucket(p)
    return {
        "client_id": user_pseudo_id,
        "non_personalized_ads": False,
        "user_properties": {
            "purchase_propensity": {"value": b}
        },
        "events": [{
            "name": "propensity_sync",
            "params": {
                "propensity_score": round(p, 3),
                "propensity_bucket": b,
            },
        }],
    }

def validate_payload(payload: dict) -> None:
    """
    Send a payload to GA4's debug endpoint to surface validation errors.
    The production endpoint returns 204 even for malformed payloads, so
    without this sampling we'd never know if we were pushing garbage.
    Called on DEBUG_VALIDATE_RATE fraction of payloads (default 1%).
    """
    try:
        r = _session.post(MP_DEBUG_URL, json=payload, timeout=5)
        msgs = r.json().get("validationMessages", [])
        if msgs:
            log.error("MP validation errors for client_id=%s: %s", payload["client_id"], msgs)
    except Exception as e:
        log.warning("Debug validation failed: %s", e)

def send(row) -> tuple[bool, float]:
    """
    POST one user's payload to GA4 Measurement Protocol.
    Returns (success_bool, latency_seconds) for the parent's aggregation.
    Retries on transient failures are handled by the session adapter.
    """
    payload = build_payload(row.user_pseudo_id, row.p)
    if random.random() < DEBUG_VALIDATE_RATE:
        validate_payload(payload)
    start = time.monotonic()
    try:
        r = _session.post(MP_URL, json=payload, timeout=5)
        # MP returns 204 No Content on success.
        return (r.status_code == 204, time.monotonic() - start)
    except requests.RequestException as e:
        log.warning("MP send failed for client_id=%s: %s", row.user_pseudo_id, e)
        return (False, time.monotonic() - start)

def p95(latencies: list[float]) -> float:
    """
    Return p95 latency. statistics.quantiles needs at least 2 values;
    smaller lists return their max (or 0). Avoids the off-by-one bugs
    you get with manual indexing on percentiles.
    """
    if len(latencies) < 2:
        return latencies[0] if latencies else 0.0
    # n=20 splits into 20 equal buckets; index 18 is the boundary at 95%.
    return statistics.quantiles(latencies, n=20)[18]

# --- Entry point -------------------------------------------------------------

def run(request):
    """
    Cloud Function HTTP entry point.

    Flow:
      1. Check model freshness (warns if stale, doesn't block)
      2. Query BigQuery for scored active users
      3. POST each user's propensity bucket to GA4 in parallel
      4. Log aggregate metrics for monitoring and alerting

    Returns plain text — the actual data product is the side effect of
    writing user properties to GA4. The string return value just gives
    Cloud Scheduler a 200 OK.
    """
    t0 = time.monotonic()
    check_model_freshness()

    # maximum_bytes_billed caps cost: if the query would scan more than
    # MAX_BYTES_BILLED, BigQuery aborts before charging. Fails the function
    # loud, which is what we want — better than a surprise bill.
    job = bq.query(SQL, job_config=bigquery.QueryJobConfig(
        maximum_bytes_billed=MAX_BYTES_BILLED,
    ))
    rows = list(job.result())
    bq_gb = (job.total_bytes_billed or 0) / 1024**3
    log.info("BQ query: %d rows, %.3f GB billed", len(rows), bq_gb)

    if not rows:
        # Empty result on a daily run is unusual — possible causes: GA4 export delay,
        # consent rejection across all traffic, or all users excluded as recent buyers.
        # Worth a WARNING (not ERROR) so Cloud Monitoring can pick it up if it persists.
        log.warning("No active consented users to sync")
        return ("No active consented users to sync", 200)

    # 25 worker threads is a balance between throughput and not getting
    # rate-limited by Google's Measurement Protocol endpoint. Tune if
    # you see 429s in the retry logs.
    with ThreadPoolExecutor(max_workers=25) as ex:
        results = list(ex.map(send, rows))

    successes = [latency for ok, latency in results if ok]
    failures  = len(results) - len(successes)
    duration  = time.monotonic() - t0

    log.info("Sync complete: %d ok, %d failed, p95 latency %.2fs, total %.1fs",
             len(successes), failures, p95(successes), duration)

    # 10% failure rate threshold — above this, something systemic is wrong
    # (MP outage, malformed payloads, network issues). Logged at ERROR
    # severity so Cloud Monitoring alerts can fire on it.
    if failures > len(rows) * 0.1:
        log.error("High failure rate: %d/%d", failures, len(rows))

    return (f"Synced {len(successes)}/{len(rows)} users", 200)