daily_aggregator.py
#!/usr/bin/env python3
"""
Daily Aggregator Job - Computes causal ROI metrics
Runs daily at 2am via Cloud Scheduler
"""
import logging
import os
from datetime import datetime, timedelta
from typing import Optional
from google.cloud import bigquery
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Config
PROJECT_ID = os.getenv("GCP_PROJECT_ID")
DATASET_ID = os.getenv("BQ_DATASET", "causal_roi")
DECAY_ALPHA = float(os.getenv("DECAY_ALPHA", "0.85"))
MIN_SAMPLE_SIZE = int(os.getenv("MIN_SAMPLE_SIZE", "30"))
client = bigquery.Client(project=PROJECT_ID)
def compute_edge_metrics(run_date: str):
"""
Compute edge-level ΔROI metrics for a given date.
Joins treatment and control outcomes.
"""
query = f"""
CREATE OR REPLACE TABLE `{PROJECT_ID}.{DATASET_ID}.edge_metrics`
PARTITION BY date
CLUSTER BY path_id, edge_id
AS
WITH action_outcomes AS (
SELECT
a.path_id,
a.action_id,
a.channel,
a.spend,
a.t1,
-- Treatment outcomes
AVG(CASE WHEN o.cohort_id = 'treatment' THEN o.value ELSE NULL END) AS rev_treatment,
-- Control outcomes (avg across control cohorts)
AVG(CASE WHEN o.cohort_id LIKE 'control_%' THEN o.value ELSE NULL END) AS rev_control,
COUNT(DISTINCT o.cohort_id) AS n_cohorts,
COUNT(*) AS n_obs,
-- Latency: action -> outcome
AVG(TIMESTAMP_DIFF(o.t2, a.t1, MILLISECOND)) AS latency_ms
FROM `{PROJECT_ID}.{DATASET_ID}.events_action` a
LEFT JOIN `{PROJECT_ID}.{DATASET_ID}.events_outcome` o
ON a.path_id = o.path_id
AND o.kpi = 'revenue'
AND o.t2 BETWEEN a.t1 AND TIMESTAMP_ADD(a.t1, INTERVAL 7 DAY)
WHERE DATE(a.t1) = '{run_date}'
GROUP BY a.path_id, a.action_id, a.channel, a.spend, a.t1
),
edge_joins AS (
SELECT
ao.*,
e.edge_id,
e.edge_type
FROM action_outcomes ao
JOIN `{PROJECT_ID}.{DATASET_ID}.edges` e
ON e.dst_node_id LIKE CONCAT('%', ao.channel, '%')
AND e.edge_type = 'action->outcome'
)
SELECT
DATE('{run_date}') AS date,
ej.edge_id,
ej.path_id,
ej.channel,
ej.spend,
ej.rev_treatment,
ej.rev_control,
-- ΔROI = (revenue_treat - revenue_ctrl) / spend
SAFE_DIVIDE(
ej.rev_treatment - COALESCE(ej.rev_control, 0),
NULLIF(ej.spend, 0)
) AS delta_roi,
ej.n_obs,
CAST(ej.latency_ms AS INT64) AS latency_ms,
-- Standard error (simplified)
SAFE_DIVIDE(
STDDEV(ej.rev_treatment - COALESCE(ej.rev_control, 0)),
SQRT(ej.n_obs)
) AS std_error,
CURRENT_TIMESTAMP() AS created_at
FROM edge_joins ej
WHERE ej.n_obs >= {MIN_SAMPLE_SIZE}
"""
job = client.query(query)
job.result()
logger.info(f"✓ Computed edge_metrics for {run_date}: {job.total_rows} rows")
return job.total_rows
def compute_path_metrics(run_date: str):
"""
Aggregate edge metrics to path-level metrics.
Computes velocity, PCR, and path-level ROI.
"""
query = f"""
CREATE OR REPLACE TABLE `{PROJECT_ID}.{DATASET_ID}.path_metrics`
PARTITION BY date
CLUSTER BY path_id
AS
WITH path_velocities AS (
SELECT
a.path_id,
-- Decision velocity: median(t_action - t_forecast)
APPROX_QUANTILES(
TIMESTAMP_DIFF(a.t1, f.t0, MILLISECOND),
100
)[SAFE_OFFSET(50)] AS velocity_ms
FROM `{PROJECT_ID}.{DATASET_ID}.events_action` a
JOIN `{PROJECT_ID}.{DATASET_ID}.events_forecast` f
ON a.path_id = f.path_id
WHERE DATE(a.t1) = '{run_date}'
GROUP BY a.path_id
),
path_attribution AS (
SELECT
path_id,
AVG(weight) AS p_causal
FROM `{PROJECT_ID}.{DATASET_ID}.attribution_links`
GROUP BY path_id
),
path_edges AS (
SELECT
em.path_id,
AVG(em.delta_roi) AS roi_delta,
COUNT(DISTINCT em.edge_id) AS n_edges,
SUM(em.n_obs) AS total_obs
FROM `{PROJECT_ID}.{DATASET_ID}.edge_metrics` em
WHERE em.date = '{run_date}'
GROUP BY em.path_id
)
SELECT
DATE('{run_date}') AS date,
pv.path_id,
pv.velocity_ms,
pa.p_causal,
-- PCR computed in post-processing (needs global top-N)
0.0 AS pcr,
NULL AS accuracy,
pe.roi_delta,
pe.n_edges AS n_actions,
pe.total_obs AS n_outcomes,
CURRENT_TIMESTAMP() AS created_at
FROM path_velocities pv
LEFT JOIN path_attribution pa ON pv.path_id = pa.path_id
LEFT JOIN path_edges pe ON pv.path_id = pe.path_id
"""
job = client.query(query)
job.result()
logger.info(f"✓ Computed path_metrics for {run_date}: {job.total_rows} rows")
# Compute PCR for top paths
compute_pcr(run_date)
return job.total_rows
def compute_pcr(run_date: str):
"""
Path Confidence Ratio: probability-weighted precision of top paths.
PCR = Σ(P(causal) × I[top_path]) / Σ(P(causal)_all_top)
"""
query = f"""
MERGE `{PROJECT_ID}.{DATASET_ID}.path_metrics` target
USING (
WITH top_paths AS (
SELECT
path_id,
p_causal,
roi_delta,
ROW_NUMBER() OVER (ORDER BY roi_delta * p_causal DESC) AS rank
FROM `{PROJECT_ID}.{DATASET_ID}.path_metrics`
WHERE date = '{run_date}'
ORDER BY rank
LIMIT 20
),
total_p_causal AS (
SELECT SUM(p_causal) AS total FROM top_paths
)
SELECT
tp.path_id,
tp.p_causal / tpc.total AS pcr
FROM top_paths tp
CROSS JOIN total_p_causal tpc
) source
ON target.path_id = source.path_id
AND target.date = '{run_date}'
WHEN MATCHED THEN
UPDATE SET pcr = source.pcr
"""
job = client.query(query)
job.result()
logger.info(f"✓ Updated PCR for top paths on {run_date}")
def update_edge_weights(run_date: str):
"""
Self-learning: update edge weights with exponential decay.
new_weight = α * old_weight + (1-α) * clipped(ΔROI)
"""
query = f"""
MERGE `{PROJECT_ID}.{DATASET_ID}.edges` target
USING (
SELECT
em.edge_id,
em.delta_roi,
em.latency_ms,
em.n_obs
FROM `{PROJECT_ID}.{DATASET_ID}.edge_metrics` em
WHERE em.date = '{run_date}'
) source
ON target.edge_id = source.edge_id
WHEN MATCHED THEN
UPDATE SET
latest_delta_roi = source.delta_roi,
latest_latency_ms = source.latency_ms,
-- Exponential decay: 85% old + 15% new (clipped to [-5, 5])
decay_weight = {DECAY_ALPHA} * decay_weight +
{1 - DECAY_ALPHA} * GREATEST(-5, LEAST(5, source.delta_roi)),
n_observations = n_observations + source.n_obs,
last_updated = CURRENT_TIMESTAMP()
"""
job = client.query(query)
job.result()
logger.info(
f"✓ Updated edge weights with decay for {run_date}: {job.total_rows} edges"
)
return job.total_rows
def downgrade_stale_edges():
"""
Auto-downgrade edges with insufficient data or stale updates.
"""
query = f"""
UPDATE `{PROJECT_ID}.{DATASET_ID}.edges`
SET
decay_weight = decay_weight * 0.5,
last_updated = CURRENT_TIMESTAMP()
WHERE
(n_observations < {MIN_SAMPLE_SIZE} OR
last_updated < TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 14 DAY))
AND decay_weight > 0.1
"""
job = client.query(query)
job.result()
logger.info(f"✓ Downgraded stale edges: {job.total_rows} edges")
return job.total_rows
def run_daily_aggregation(run_date: Optional[str] = None):
"""
Main orchestrator: runs all aggregation steps.
"""
if run_date is None:
run_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
logger.info(f"🚀 Starting daily aggregation for {run_date}")
try:
# Step 1: Compute edge metrics (treatment vs control)
edge_rows = compute_edge_metrics(run_date)
# Step 2: Aggregate to path metrics
path_rows = compute_path_metrics(run_date)
# Step 3: Update self-learning weights
updated_edges = update_edge_weights(run_date)
# Step 4: Downgrade stale edges
downgraded_edges = downgrade_stale_edges()
logger.info(f"✅ Aggregation complete for {run_date}")
logger.info(f" Edges: {edge_rows}, Paths: {path_rows}")
logger.info(f" Updated: {updated_edges}, Downgraded: {downgraded_edges}")
return {
"status": "success",
"date": run_date,
"edge_metrics": edge_rows,
"path_metrics": path_rows,
"updated_edges": updated_edges,
"downgraded_edges": downgraded_edges,
}
except Exception as e:
logger.error(f"❌ Aggregation failed for {run_date}: {e}")
raise
if __name__ == "__main__":
import sys
run_date = sys.argv[1] if len(sys.argv) > 1 else None
result = run_daily_aggregation(run_date)
print(result)