01_create_tables.sql

-- Causal ROI Dashboard - BigQuery Schema
-- Run this after creating your dataset: bq mk --dataset ${PROJECT_ID}:causal_roi

-- ============================================================================
-- RAW EVENT TABLES (your logging layer writes here)
-- ============================================================================

CREATE TABLE IF NOT EXISTS `causal_roi.events_forecast` (
  path_id STRING NOT NULL,
  hypothesis_id STRING NOT NULL,
  t0 TIMESTAMP NOT NULL,
  features JSON,  -- flexible: {"channel": "email", "segment": "high_value", ...}
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
)
PARTITION BY DATE(t0)
CLUSTER BY path_id, hypothesis_id
OPTIONS(
  description="Forecasts/predictions issued by the system",
  labels=[("env", "prod"), ("domain", "attribution")]
);

CREATE TABLE IF NOT EXISTS `causal_roi.events_action` (
  path_id STRING NOT NULL,
  action_id STRING NOT NULL,
  channel STRING NOT NULL,
  spend FLOAT64 NOT NULL,
  t1 TIMESTAMP NOT NULL,
  metadata JSON,  -- e.g., {"campaign_id": "Q4_promo", "creative": "v2"}
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
)
PARTITION BY DATE(t1)
CLUSTER BY path_id, channel
OPTIONS(
  description="Actions taken (ad spend, emails sent, etc.)"
);

CREATE TABLE IF NOT EXISTS `causal_roi.events_outcome` (
  path_id STRING NOT NULL,
  kpi STRING NOT NULL,  -- "revenue", "conversions", "ltv"
  value FLOAT64 NOT NULL,
  t2 TIMESTAMP NOT NULL,
  cohort_id STRING,  -- "treatment" or "control_{geo}" or "control_{synthetic}"
  attribution_window_days INT64 DEFAULT 7,
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
)
PARTITION BY DATE(t2)
CLUSTER BY path_id, cohort_id, kpi
OPTIONS(
  description="Observed outcomes with treatment/control labels"
);

CREATE TABLE IF NOT EXISTS `causal_roi.attribution_links` (
  path_id STRING NOT NULL,
  method STRING NOT NULL,  -- "shapley", "markov", "geo_holdout", "uplift_model"
  weight FLOAT64 NOT NULL,  -- normalized [0,1] probability score
  window_days INT64 DEFAULT 7,
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP(),
  updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
)
PARTITION BY DATE(created_at)
CLUSTER BY path_id, method
OPTIONS(
  description="Attribution weights from different methods"
);

-- ============================================================================
-- AGGREGATED METRICS (written by daily job)
-- ============================================================================

CREATE TABLE IF NOT EXISTS `causal_roi.edge_metrics` (
  date DATE NOT NULL,
  edge_id STRING NOT NULL,
  path_id STRING NOT NULL,
  channel STRING,
  spend FLOAT64,
  rev_treatment FLOAT64,
  rev_control FLOAT64,
  delta_roi FLOAT64,  -- (rev_treat - rev_ctrl) / spend
  n_obs INT64,
  latency_ms INT64,  -- action→outcome lag
  std_error FLOAT64,  -- optional: for confidence intervals
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
)
PARTITION BY date
CLUSTER BY path_id, edge_id
OPTIONS(
  description="Daily edge-level causal metrics"
);

CREATE TABLE IF NOT EXISTS `causal_roi.path_metrics` (
  date DATE NOT NULL,
  path_id STRING NOT NULL,
  velocity_ms INT64,  -- median(t_action - t_forecast)
  p_causal FLOAT64,  -- from attribution_links (avg weight)
  pcr FLOAT64,  -- path confidence ratio
  accuracy FLOAT64,  -- if tracking hit/miss
  roi_delta FLOAT64,  -- avg edge delta_roi for this path
  n_actions INT64,
  n_outcomes INT64,
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
)
PARTITION BY date
CLUSTER BY path_id
OPTIONS(
  description="Daily path-level rollup metrics"
);

-- ============================================================================
-- GRAPH STRUCTURE (for traversal/visualization)
-- ============================================================================

CREATE TABLE IF NOT EXISTS `causal_roi.nodes` (
  node_id STRING NOT NULL,
  node_type STRING NOT NULL,  -- "signal", "action", "outcome"
  label STRING,
  props JSON,  -- {"name": "email_open", "tier": "critical", ...}
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
)
CLUSTER BY node_id, node_type
OPTIONS(
  description="Knowledge graph nodes"
);

CREATE TABLE IF NOT EXISTS `causal_roi.edges` (
  edge_id STRING NOT NULL,
  src_node_id STRING NOT NULL,
  dst_node_id STRING NOT NULL,
  edge_type STRING NOT NULL,  -- "signal->action", "action->outcome"
  latest_delta_roi FLOAT64,
  latest_latency_ms INT64,
  decay_weight FLOAT64 DEFAULT 1.0,  -- exponential decay for self-learning
  n_observations INT64 DEFAULT 0,
  last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP(),
  props JSON
)
CLUSTER BY src_node_id, dst_node_id, edge_type
OPTIONS(
  description="Knowledge graph edges with learned weights"
);

-- ============================================================================
-- CONTROL/TREATMENT TRACKING (for experimentation)
-- ============================================================================

CREATE TABLE IF NOT EXISTS `causal_roi.control_cohorts` (
  cohort_id STRING NOT NULL,
  path_id STRING NOT NULL,
  method STRING NOT NULL,  -- "geo_holdout", "synthetic_control", "randomized"
  geo_region STRING,  -- if geo-based
  synthetic_weights JSON,  -- if synthetic control: {"region_A": 0.3, ...}
  quality_score FLOAT64,  -- match quality
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
)
PARTITION BY DATE(created_at)
CLUSTER BY path_id, cohort_id
OPTIONS(
  description="Control group definitions for causal inference"
);

-- ============================================================================
-- INDEXES (for fast queries)
-- ============================================================================

-- Note: BigQuery uses clustering instead of indexes
-- Already applied above via CLUSTER BY clauses