feat: delayed ground truth tutorial (POC)

README.md

@@ -72,6 +72,11 @@ python example/scripts/07_setup_grafana.py         # provision monitoring dashbo
 
 See [example/README.md](example/README.md) for details.
 
+**Delayed ground truth tutorial (POC):** a variant that trains on only a
+cutoff of the data, then streams the rest in as batches with ground truth
+arriving after a delay, so you can watch drift/MAE metrics evolve instead of
+landing all at once. See [example/tutorial_streaming/README.md](example/tutorial_streaming/README.md).
+
 ## Teardown
 
 ```bash

example/tutorial_streaming/01_load_and_cutoff.py

@@ -0,0 +1,39 @@
+"""
+Step 1: Generate the synthetic housing dataset and load only the first
+TRAIN_CUTOFF_FRACTION of it into BigQuery offline_features, holding the rest
+back to be streamed in later by 04_simulate_stream.py.
+
+Assumes a fresh mlops dataset (see README.md).
+"""
+
+import os
+from pathlib import Path
+from dotenv import load_dotenv
+from google.cloud import bigquery
+
+from _dataset import generate_rows, N_ROWS
+
+load_dotenv(Path.cwd() / ".env")
+
+PROJECT = os.environ.get("BIGQUERY_PROJECT")
+if not PROJECT:
+    raise SystemExit(
+        "BIGQUERY_PROJECT is not set. Run `deployml get-urls` after `deployml deploy` to write a .env "
+        "with BIGQUERY_PROJECT, MLFLOW_URL, and others. Then re-run this script from the same directory."
+    )
+DATASET = os.getenv("BIGQUERY_DATASET", "mlops")
+TABLE = f"{PROJECT}.{DATASET}.offline_features"
+TRAIN_CUTOFF_FRACTION = 0.4
+
+rows = generate_rows()
+cutoff = int(N_ROWS * TRAIN_CUTOFF_FRACTION)
+train_rows = rows[:cutoff]
+
+client = bigquery.Client(project=PROJECT)
+errors = client.insert_rows_json(TABLE, train_rows)
+
+if errors:
+    print(f"Errors inserting rows: {errors}")
+else:
+    print(f"✓ Loaded {len(train_rows)}/{N_ROWS} rows into {TABLE} (initial training cutoff)")
+    print(f"  {N_ROWS - cutoff} rows held back for 04_simulate_stream.py")

example/tutorial_streaming/04_simulate_stream.py

@@ -0,0 +1,103 @@
+"""
+Step 4: Simulate new incoming data arriving in batches. Each batch is scored
+via FastAPI /predict immediately, then its ground truth is written after a
+short delay, so the predictions and ground_truth tables fill in at different
+times like a real stream.
+
+ponytail: BATCH_DELAY_SECONDS is a real wall-clock sleep, not a simulated
+one, so total runtime is tied to demo time. Upgrade path: write ground truth
+with backdated event_timestamps instead of sleeping, so the full time series
+is visible immediately regardless of how long the script takes to run.
+"""
+
+import os
+import time
+from pathlib import Path
+from datetime import datetime, timezone
+import numpy as np
+import requests
+from dotenv import load_dotenv
+from google.cloud import bigquery
+
+from _dataset import generate_rows, split_into_batches, N_ROWS
+
+load_dotenv(Path.cwd() / ".env")
+
+FASTAPI_URL = os.environ.get("FASTAPI_URL")
+PROJECT = os.environ.get("BIGQUERY_PROJECT")
+DATASET = os.getenv("BIGQUERY_DATASET", "mlops")
+TRAIN_CUTOFF_FRACTION = 0.4
+N_BATCHES = 5
+BATCH_DELAY_SECONDS = 30
+GROUND_TRUTH_NOISE = 15000  # std dev of fake noise around predicted value
+
+
+def score_batch(batch: list[dict]) -> int:
+    success = 0
+    for row in batch:
+        payload = {
+            "entity_id": row["entity_id"],
+            "features": {
+                "bedrooms": row["bedrooms"],
+                "bathrooms": row["bathrooms"],
+                "area_sqft": row["area_sqft"],
+                "lot_size": row["lot_size"],
+                "year_built": row["year_built"],
+                "city": row["city"],
+                "state": row["state"],
+            },
+        }
+        resp = requests.post(f"{FASTAPI_URL}/predict", json=payload, timeout=10)
+        if resp.status_code == 200 and resp.json().get("prediction", -1) != -1:
+            success += 1
+    return success
+
+
+def write_ground_truth(client: bigquery.Client, batch: list[dict]) -> None:
+    entity_ids = [row["entity_id"] for row in batch]
+    query = f"""
+        SELECT entity_id, predicted_value
+        FROM `{PROJECT}.{DATASET}.predictions`
+        WHERE entity_id IN UNNEST(@entity_ids)
+    """
+    job_config = bigquery.QueryJobConfig(
+        query_parameters=[bigquery.ArrayQueryParameter("entity_ids", "STRING", entity_ids)]
+    )
+    predictions = list(client.query(query, job_config=job_config).result())
+
+    now = datetime.now(timezone.utc)
+    rows = [
+        {
+            "entity_id": row["entity_id"],
+            "event_timestamp": now.isoformat(),
+            "actual_value": float(row["predicted_value"]) + np.random.normal(0, GROUND_TRUTH_NOISE),
+        }
+        for row in predictions
+    ]
+    errors = client.insert_rows_json(f"{PROJECT}.{DATASET}.ground_truth", rows)
+    if errors:
+        print(f"  Errors writing ground truth: {errors}")
+    else:
+        print(f"  ✓ Wrote ground truth for {len(rows)} rows")
+
+
+if __name__ == "__main__":
+    if not FASTAPI_URL or not PROJECT:
+        raise SystemExit(
+            "FASTAPI_URL or BIGQUERY_PROJECT missing. Run `deployml get-urls` to write .env."
+        )
+
+    all_rows = generate_rows()
+    cutoff = int(N_ROWS * TRAIN_CUTOFF_FRACTION)
+    stream_rows = all_rows[cutoff:]
+    batches = split_into_batches(stream_rows, N_BATCHES)
+
+    client = bigquery.Client(project=PROJECT)
+    for i, batch in enumerate(batches, start=1):
+        print(f"Batch {i}/{len(batches)}: scoring {len(batch)} rows")
+        success = score_batch(batch)
+        print(f"  ✓ {success}/{len(batch)} predictions successful")
+
+        print(f"  Waiting {BATCH_DELAY_SECONDS}s before ground truth arrives...")
+        time.sleep(BATCH_DELAY_SECONDS)
+        write_ground_truth(client, batch)

example/tutorial_streaming/README.md

@@ -0,0 +1,86 @@
+# Delayed Ground Truth Tutorial (POC)
+
+A small variant of the [end-to-end example](../README.md) that simulates new
+incoming data arriving over time, with ground truth showing up after a delay
+instead of all at once. Built for issue #65.
+
+**Assumption:** this expects a fresh `mlops` BigQuery dataset (no rows yet in
+`offline_features`, `predictions`, `ground_truth`). If you already ran the
+main `example/scripts/` walkthrough against the same deployment, either
+`deployml destroy` + `deployml deploy` again, or point this at a separate
+deployment.
+
+## Prerequisites
+
+Same as the [main example](../README.md#prerequisites): a deployed stack,
+`.env` from `deployml get-urls`, and the same Python dependencies.
+
+## Steps
+
+Run from the project root (where `.env` lives).
+
+**1. Load training data with a cutoff**
+
+```bash
+python example/tutorial_streaming/01_load_and_cutoff.py
+```
+
+Generates the same 500-row synthetic housing dataset as the main example, but
+only loads the first 40% (200 rows) into `offline_features`. The remaining
+300 rows are held back to be streamed in later.
+
+**2. Train a model**
+
+```bash
+python example/scripts/02_train_model.py
+```
+
+Unchanged from the main example. Trains on whatever is in `offline_features`
+— now just the 200-row cutoff instead of the full 500.
+
+**3. Register the model**
+
+```bash
+python example/scripts/03_register_model.py
+```
+
+Unchanged from the main example.
+
+**4. Simulate the incoming data stream**
+
+```bash
+python example/tutorial_streaming/04_simulate_stream.py
+```
+
+Regenerates the same 500 rows (same seed as step 1) and takes the 300 held
+back. Splits them into 5 batches of ~60 rows. For each batch: scores it via
+FastAPI `/predict` (logged to `predictions` automatically), waits 30 seconds,
+then writes ground truth for that batch to `ground_truth`. Takes about 2.5
+minutes total.
+
+**5. Compute drift metrics**
+
+```bash
+python example/scripts/06_compute_drift_metrics.py
+```
+
+Unchanged from the main example.
+
+**6. Set up the Grafana dashboard**
+
+```bash
+python example/scripts/07_setup_grafana.py
+```
+
+Unchanged from the main example. Open `GRAFANA_URL` to watch predictions and
+MAE update batch by batch if you run step 4 again, or re-run step 5 between
+batches.
+
+## Known limitations / future work
+
+- The 30-second delay is a real `time.sleep`, not a simulated one, so it's
+  tied to wall-clock demo time. A follow-up will replace it with backdated
+  `event_timestamp`s so a full time series is visible immediately.
+- This is a POC run manually as scripts. A follow-up will promote it into a
+  `deployml tutorial` CLI command, including support for uploading your own
+  dataset instead of the built-in housing one.

example/tutorial_streaming/_dataset.py

@@ -0,0 +1,50 @@
+"""Shared synthetic housing data generator for the streaming tutorial.
+
+Same feature distributions and formula as example/scripts/01_load_training_data.py,
+factored out so 01_load_and_cutoff.py and 04_simulate_stream.py can both
+regenerate the identical 500 rows (same seed) without persisting anything
+to disk between script runs.
+"""
+
+import uuid
+from datetime import datetime, timezone
+import numpy as np
+
+N_ROWS = 500
+RANDOM_SEED = 42
+
+
+def generate_rows(n_rows: int = N_ROWS, seed: int = RANDOM_SEED) -> list[dict]:
+    rng = np.random.default_rng(seed)
+
+    cities = list(range(5))  # 0-4 representing 5 cities
+    states = list(range(3))  # 0-2 representing 3 states
+
+    bedrooms = rng.integers(1, 6, n_rows).astype(float)
+    bathrooms = rng.integers(1, 4, n_rows).astype(float)
+    area_sqft = rng.integers(800, 4000, n_rows).astype(float)
+    lot_size = rng.integers(2000, 10000, n_rows).astype(float)
+    year_built = rng.integers(1960, 2023, n_rows).astype(float)
+    city = rng.choice(cities, n_rows).astype(float)
+    state = rng.choice(states, n_rows).astype(float)
+
+    now = datetime.now(timezone.utc)
+    return [
+        {
+            "entity_id": str(uuid.uuid4()),
+            "event_timestamp": now.isoformat(),
+            "bedrooms": bedrooms[i],
+            "bathrooms": bathrooms[i],
+            "area_sqft": area_sqft[i],
+            "lot_size": lot_size[i],
+            "year_built": year_built[i],
+            "city": city[i],
+            "state": state[i],
+        }
+        for i in range(n_rows)
+    ]
+
+
+def split_into_batches(rows: list, n_batches: int) -> list[list]:
+    """Split rows into n_batches near-equal chunks, covering every row exactly once."""
+    return [list(batch) for batch in np.array_split(rows, n_batches)]

tests/test_tutorial_streaming.py

@@ -0,0 +1,41 @@
+"""Unit test for the batch-splitting logic in the streaming tutorial. No GCP/FastAPI calls."""
+import importlib.util
+from pathlib import Path
+
+DATASET_MODULE_PATH = Path(__file__).parent.parent / "example" / "tutorial_streaming" / "_dataset.py"
+
+
+def _load_dataset_module():
+    spec = importlib.util.spec_from_file_location("tutorial_streaming_dataset", DATASET_MODULE_PATH)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+split_into_batches = _load_dataset_module().split_into_batches
+
+
+def test_split_into_batches_covers_every_row_once():
+    rows = [{"entity_id": i} for i in range(300)]
+    batches = split_into_batches(rows, 5)
+
+    assert len(batches) == 5
+    flattened = [row for batch in batches for row in batch]
+    assert sorted(row["entity_id"] for row in flattened) == list(range(300))
+
+
+def test_split_into_batches_near_equal_sizes():
+    rows = [{"entity_id": i} for i in range(7)]
+    batches = split_into_batches(rows, 3)
+
+    sizes = sorted(len(b) for b in batches)
+    assert sizes == [2, 2, 3]
+
+
+def test_split_into_batches_more_batches_than_rows():
+    rows = [{"entity_id": i} for i in range(2)]
+    batches = split_into_batches(rows, 5)
+
+    assert len(batches) == 5
+    flattened = [row for batch in batches for row in batch]
+    assert sorted(row["entity_id"] for row in flattened) == [0, 1]