update ci and add claude.md for uplift

.github/issues/001-flagship-pipeline-demo.md

@@ -0,0 +1,81 @@
+# Flagship End-to-End ML Pipeline Demo
+
+## Summary
+
+Transform this project into **the** flagship system-level example for Outerbounds, demonstrating a complete ML pipeline with:
+- Real-time data ingestion (CoinGecko crypto market data)
+- Automated dataset accumulation with temporal splits
+- Model training with quality gates
+- Champion/challenger promotion workflow
+- Production validation with `@trigger_on_finish` chaining
+
+## Current State
+
+The project already has the core flows:
+- `IngestMarketDataFlow` - fetches live crypto prices
+- `BuildDatasetFlow` - accumulates snapshots into train/eval datasets
+- `TrainDetectorFlow` - trains Isolation Forest model
+- `EvaluateDetectorFlow` - runs quality gates
+- `PromoteModelFlow` - promotes via run tags
+
+## Proposed Enhancements
+
+### 1. Automated Pipeline Orchestration
+
+Connect flows using `@trigger_on_finish` for fully automated retraining:
+
+```python
+# flows/build_dataset/flow.py
+@trigger_on_finish(flow="IngestMarketDataFlow")
+class BuildDatasetFlow(FlowSpec):
+    """Auto-triggers when new data is ingested."""
+```
+
+Pipeline chain:
+```
+IngestMarketDataFlow (hourly cron)
+    └─► BuildDatasetFlow (@trigger_on_finish)
+        └─► TrainDetectorFlow (@trigger_on_finish, conditional on data volume)
+            └─► EvaluateDetectorFlow (@trigger_on_finish)
+                └─► PromoteModelFlow (manual or auto-promote if gates pass)
+```
+
+### 2. Enhanced Dashboard
+
+Expand `deployments/dashboard/` to show:
+- Pipeline status (which flows have run, pending triggers)
+- Model lineage visualization (which dataset trained which model)
+- Champion history timeline
+- Live anomaly predictions using champion model
+
+### 3. Multi-Asset Support
+
+Extend beyond single anomaly detector:
+- Support multiple model types (IsolationForest, AutoEncoder, etc.)
+- Track models per asset class (BTC detector, ETH detector, etc.)
+- A/B testing between champion and challenger
+
+### 4. Production Metrics
+
+Add `ValidateOutcomesFlow` with:
+- Prediction accuracy tracking (when ground truth available)
+- Drift detection between training and production distributions
+- Automated alerting when performance degrades
+
+## Technical Notes
+
+- Use `@schedule(cron="0 * * * *")` for hourly ingestion
+- Champion management uses Metaflow run tags (`champion` tag on training runs)
+- Branch isolation: `[dev-assets]` config enables reading prod data in dev
+
+## Acceptance Criteria
+
+- [ ] All flows connected via `@trigger_on_finish`
+- [ ] Dashboard shows pipeline status and model lineage
+- [ ] At least 2 model variants tracked (champion + challenger)
+- [ ] Hourly scheduled runs working on Outerbounds
+- [ ] Documentation with architecture diagram
+
+## Labels
+
+`enhancement`, `flagship`, `demo`

.github/workflows/deploy.yml

@@ -31,31 +31,17 @@ jobs:
         fetch-depth: 0
 
     - name: Set up Python
-      uses: actions/setup-python@v1
+      uses: actions/setup-python@v5
       with:
-        python-version: 3.12
+        python-version: "3.12"
 
     - name: Install dependencies
-      run: |
-        python3 -m pip install -U requests
-        python3 -m pip install outerbounds pyyaml
-        python3 -m pip install -U ob-project-utils
+      run: pip install outerbounds ob-project-utils
 
     - name: Configure Outerbounds
       run: |
-        PROJECT_NAME=$(yq .project obproject.toml)
-        DEFAULT_CICD_USER="${PROJECT_NAME//_/-}-cicd"
-        PLATFORM=$(yq .platform obproject.toml)
-        CICD_USER=$(yq ".cicd_user // \"$DEFAULT_CICD_USER\"" obproject.toml)
-        PERIMETER="default"
-        echo "Deployment target:"
-        echo "  Platform: $PLATFORM"
-        echo "  CI/CD User: $CICD_USER"
-        echo "  Perimeter: $PERIMETER"
         outerbounds service-principal-configure \
-          --name $CICD_USER \
-          --deployment-domain $PLATFORM \
-          --perimeter $PERIMETER \
+          --from-obproject-toml \
           --github-actions
 
     - name: Deploy Project

CLAUDE.md

@@ -0,0 +1,96 @@
+# Crypto Anomaly Detection — Model Registry
+
+Production ML systems need a way to train models, evaluate them against quality gates, promote the best to champion, and validate predictions against real outcomes. This project demonstrates that full lifecycle using live cryptocurrency data from CoinGecko — detecting anomalous price movements that may signal crashes.
+
+The model registry patterns here (versioned assets, champion tracking, quality gates, outcome validation) apply to any domain where models are continuously retrained and promoted.
+
+## Architecture
+
+```
+IngestMarketDataFlow (@schedule hourly)
+  → fetches CoinGecko data, writes Parquet, registers market_snapshot DataAsset
+
+BuildDatasetFlow (@schedule daily)
+  → accumulates snapshots into temporal features, time-splits into train/holdout
+  → registers training_dataset + eval_holdout DataAssets
+  → publishes "dataset_ready" event
+
+TrainDetectorFlow (@project_trigger "dataset_ready")
+  → consumes training_dataset, trains Isolation Forest
+  → registers anomaly_detector ModelAsset with metrics in annotations
+
+EvaluateDetectorFlow (@trigger_on_finish TrainDetectorFlow)
+  → loads candidate model + eval_holdout
+  → applies quality gates (min anomaly rate, max contamination)
+  → registers evaluation record, publishes "evaluation_passed"
+
+PromoteModelFlow (manual)
+  → promotes a model version to "champion" via run tags
+
+ValidateOutcomesFlow (@schedule hourly)
+  → loads predictions from ~24h ago, fetches current prices
+  → checks if flagged anomalies actually crashed (>15% drop)
+  → closes the feedback loop: crash_recall, anomaly_precision, false_alarm_rate
+```
+
+Dashboard in `deployments/dashboard/` visualizes model performance and market data.
+
+## Platform features used
+
+- **Data assets**: market_snapshot, training_dataset, eval_holdout — versioned with annotations (row counts, date ranges)
+- **Model assets**: anomaly_detector — versioned with quality metrics in annotations
+- **Events**: `dataset_ready` triggers training, `evaluation_passed` signals quality gate pass
+- **@trigger_on_finish**: EvaluateFlow chains after TrainFlow automatically
+- **@schedule**: Hourly ingest, daily dataset build, hourly outcome validation
+- **Branch config**: `[branch_to_environment]` maps main→production, *→dev
+- **[dev-assets]**: Feature branches read production data from main
+- **Config**: Flow-level JSON configs for model, training, and evaluation parameters via `Config()` objects
+- **Apps**: FastAPI dashboard in `deployments/dashboard/`
+
+## Flows
+
+| Flow | Trigger | Purpose |
+|------|---------|---------|
+| IngestMarketDataFlow | `@schedule(hourly)` | Fetch CoinGecko data, register market_snapshot |
+| BuildDatasetFlow | `@schedule(daily)` | Accumulate snapshots, temporal features, time-split |
+| TrainDetectorFlow | `@project_trigger("dataset_ready")` | Train Isolation Forest, register model asset |
+| EvaluateDetectorFlow | `@trigger_on_finish(TrainDetectorFlow)` | Quality gates, register evaluation record |
+| PromoteModelFlow | Manual | Tag model version as champion |
+| ValidateOutcomesFlow | `@schedule(hourly)` | Compare predictions to actual outcomes |
+
+## CI strategy
+
+Deploy-only (no teardown or promote CI module). Pushes to main, develop, and feature/** trigger deploy. The project deploys flows with schedules and event sensors — teardown would require removing those on branch cleanup, which isn't configured yet. Consider adding teardown for feature branches.
+
+Uses `--from-obproject-toml` for auth — reads platform, project, and perimeter from `obproject.toml` and branch-to-environment mapping.
+
+## Run locally
+
+```bash
+export PYTHONPATH="$PWD"
+
+# Quick: train on fresh data (no assets needed)
+python flows/train/flow.py run
+
+# Full pipeline: ingest → build → train → evaluate
+python flows/ingest/flow.py run
+python flows/build_dataset/flow.py run
+python flows/train/flow.py run
+python flows/evaluate/flow.py run
+
+# Promote a model
+python flows/promote/flow.py run --version latest
+
+# Override config for experiments
+python flows/train/flow.py --config model_config configs/model.json run
+```
+
+`PYTHONPATH` must include project root so `import src` resolves across all flows.
+
+## Good to know
+
+- **Config vs Parameters**: `Config()` objects (`model_config`, `training_config`, `eval_config`) are set at deployment time from JSON files in `configs/`. Parameters override config values at run time. This separation lets you deploy with environment-specific defaults while still allowing manual overrides.
+- **Champion tracking** uses Metaflow run tags, not Asset API tags. Query with `Flow("TrainDetectorFlow").runs("champion")`.
+- **New branches have no historical data**. `[dev-assets] branch = "main"` lets feature branches read production market snapshots while writing new assets to their own namespace.
+- **Two asset registration patterns**: `register_data()` for Metaflow artifacts (snapshots), `register_external_data()` for Parquet blobs on cloud storage (training datasets). Choose based on whether the data is a Python object or a file.
+- **BuildDatasetFlow** needs `min_snapshots_per_coin + 1` snapshots before it can produce a dataset. On a fresh deployment, you need multiple ingest runs first.

deployments/dashboard/app.py

@@ -596,12 +596,16 @@ async def overview(request: Request):
         except Exception as e:
             print(f"[DEBUG] Failed to load champion run {champion_run_id}: {e}")
 
-    return templates.TemplateResponse("overview.html", get_template_context(
-        request, branch,
-        status=status,
-        evaluation=evaluation,
-        champion=champion_details,
-    ))
+    return templates.TemplateResponse(
+        request=request,
+        name="overview.html",
+        context=get_template_context(
+            request, branch,
+            status=status,
+            evaluation=evaluation,
+            champion=champion_details,
+        ),
+    )
 
 
 @app.get("/data", response_class=HTMLResponse)
@@ -632,10 +636,14 @@ async def data_explorer(request: Request):
                 "message": "No data yet",
             })
 
-    return templates.TemplateResponse("data.html", get_template_context(
-        request, branch,
-        data_assets=data_assets,
-    ))
+    return templates.TemplateResponse(
+        request=request,
+        name="data.html",
+        context=get_template_context(
+            request, branch,
+            data_assets=data_assets,
+        ),
+    )
 
 
 @app.get("/models", response_class=HTMLResponse)
@@ -661,13 +669,17 @@ async def model_registry(request: Request):
     # Count champions to validate uniqueness
     champion_count = sum(1 for v in versions if v.get("alias") == "champion")
 
-    return templates.TemplateResponse("models.html", get_template_context(
-        request, branch,
-        versions=versions,
-        error_message=error_message,
-        champion=champion,
-        champion_count=champion_count,
-    ))
+    return templates.TemplateResponse(
+        request=request,
+        name="models.html",
+        context=get_template_context(
+            request, branch,
+            versions=versions,
+            error_message=error_message,
+            champion=champion,
+            champion_count=champion_count,
+        ),
+    )
 
 
 
@@ -885,11 +897,15 @@ async def cards_page(request: Request):
             print(f"[DEBUG] cards_page error for {flow_name}: {e}")
             pass
 
-    return templates.TemplateResponse("cards.html", get_template_context(
-        request, branch,
-        cards=cards_info,
-        flow_runs=flow_runs,  # For intra-flow comparison dropdowns
-    ))
+    return templates.TemplateResponse(
+        request=request,
+        name="cards.html",
+        context=get_template_context(
+            request, branch,
+            cards=cards_info,
+            flow_runs=flow_runs,  # For intra-flow comparison dropdowns
+        ),
+    )
 
 
 # =============================================================================

flows/build_dataset/flow.py

@@ -131,7 +131,7 @@ def build_and_write(self):
         import pandas as pd
 
         snapshot_store = SnapshotStore(self.prj.project, self.prj.branch)
-        dataset_store = DatasetStore(self.prj.project, self.prj.branch)
+        dataset_store = DatasetStore(self.prj.project, self.prj.write_branch)
 
         root, provider = get_datastore_root()
         print(f"Storage: {provider} ({root})")

flows/evaluate/flow.py

@@ -67,14 +67,14 @@ class EvaluateDetectorFlow(ProjectFlow):
     # Data source override - Parameter overrides config at runtime
     eval_data_source = Parameter(
         "eval_data_source",
-        default=None,
-        help="Override eval data source: 'eval_holdout' (default) or 'fresh'. If not set, uses eval_config."
+        default="eval_holdout",
+        help="Eval data source: 'eval_holdout' or 'fresh'"
     )
 
     eval_data_version = Parameter(
         "eval_data_version",
-        default=None,
-        help="Override eval data version (e.g., 'latest', 'latest-1'). If not set, uses eval_config."
+        default="latest",
+        help="Eval data version (e.g., 'latest', 'latest-1')"
     )
 
     @step
@@ -131,8 +131,16 @@ def load_eval_data(self):
         config_source = self.eval_config.get("eval_data_source") if self.eval_config else None
         config_version = self.eval_config.get("eval_data_version") if self.eval_config else None
 
-        data_source = self.eval_data_source or config_source or "eval_holdout"
-        data_version = self.eval_data_version or config_version or "latest"
+        # Guard against string "None" from Parameter serialization in Argo triggers
+        def _resolve(param, config_val, default):
+            if param and str(param) != "None":
+                return param
+            if config_val and str(config_val) != "None":
+                return config_val
+            return default
+
+        data_source = _resolve(self.eval_data_source, config_source, "eval_holdout")
+        data_version = _resolve(self.eval_data_version, config_version, "latest")
 
         print(f"\nEvaluation data source: {data_source}")
         print(f"Evaluation data version: {data_version}")

flows/ingest/flow.py

@@ -75,7 +75,7 @@ def register(self):
         # Write Parquet snapshot to cloud storage for fast-data accumulation
         print("\nWriting Parquet snapshot to storage...")
         try:
-            store = SnapshotStore(self.prj.project, self.prj.branch)
+            store = SnapshotStore(self.prj.project, self.prj.write_branch)
             snapshot_meta = store.write_snapshot(
                 self.feature_set,
                 flow_name=current.flow_name,