Add daily IBIT BTC platform handoff publish workflow

.github/workflows/publish-platform-handoffs.yml

@@ -0,0 +1,117 @@
+name: Publish Platform Handoffs
+
+on:
+  schedule:
+    # Daily after the prior UTC date's BTC daily close is available.
+    - cron: '15 1 * * *'
+  workflow_dispatch:
+    inputs:
+      execute_publish:
+        description: Upload platform handoffs to GCS. false only builds a GitHub artifact.
+        required: true
+        default: false
+        type: boolean
+      as_of:
+        description: Optional signal as-of date (YYYY-MM-DD).
+        required: false
+        type: string
+      gcs_prefix:
+        description: Optional GCS prefix override for platform handoffs.
+        required: false
+        type: string
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref_name }}
+  cancel-in-progress: false
+
+jobs:
+  publish-ibit-btc-handoff:
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    permissions:
+      contents: read
+      id-token: write
+    env:
+      GCP_PROJECT_ID: ${{ vars.GCP_PROJECT_ID || 'interactivebrokersquant' }}
+      GCP_WORKLOAD_IDENTITY_PROVIDER: ${{ vars.GCP_WORKLOAD_IDENTITY_PROVIDER }}
+      GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT: ${{ vars.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }}
+      GCS_PREFIX: ${{ inputs.gcs_prefix || vars.MARKET_SIGNAL_GCS_PREFIX || 'gs://qsl-runtime-logs-shared/platform_handoffs' }}
+      EXECUTE_PUBLISH: ${{ github.event_name == 'schedule' && 'true' || inputs.execute_publish }}
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+
+      - name: Install package
+        run: |
+          set -euo pipefail
+          python -m pip install --upgrade pip
+          python -m pip install -e .
+
+      - name: Authenticate to Google Cloud
+        if: env.EXECUTE_PUBLISH == 'true'
+        uses: google-github-actions/auth@v3
+        with:
+          workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }}
+
+      - name: Set up gcloud
+        if: env.EXECUTE_PUBLISH == 'true'
+        uses: google-github-actions/setup-gcloud@v3
+        with:
+          project_id: ${{ env.GCP_PROJECT_ID }}
+          version: '>= 416.0.0'
+
+      - name: Build and publish IBIT BTC platform handoff
+        id: publish
+        run: |
+          set -euo pipefail
+          args=(
+            --work-dir data/output
+            --source-version 0.1.1
+            --code-commit "${GITHUB_SHA}"
+            --gcs-prefix "${GCS_PREFIX}"
+          )
+          if [ -n "${{ inputs.as_of }}" ]; then
+            args+=(--as-of "${{ inputs.as_of }}")
+          fi
+          if [ "${EXECUTE_PUBLISH}" = "true" ]; then
+            args+=(--execute)
+          fi
+          python scripts/publish_ibit_btc_platform_handoff.py "${args[@]}"
+          as_of="$(python - <<'PY'
+          import json
+          from pathlib import Path
+          index = json.loads(Path("data/output/platform_handoffs/index.json").read_text(encoding="utf-8"))
+          print(index["handoffs"][-1]["as_of"])
+          PY
+          )"
+          echo "as_of=${as_of}" >> "$GITHUB_OUTPUT"
+
+      - name: Upload generated artifacts
+        uses: actions/upload-artifact@v4
+        with:
+          name: platform-handoffs-ibit-btc-${{ steps.publish.outputs.as_of }}-${{ github.run_id }}
+          path: data/output/platform_handoffs
+          if-no-files-found: error
+          retention-days: 7
+
+      - name: Append job summary
+        run: |
+          set -euo pipefail
+          {
+            echo "## Platform handoff publish"
+            echo
+            echo "- consumer: \`us_equity:ibit_smart_dca\`"
+            echo "- as_of: \`${{ steps.publish.outputs.as_of }}\`"
+            echo "- gcs_prefix: \`${GCS_PREFIX}\`"
+            echo "- execute_publish: \`${EXECUTE_PUBLISH}\`"
+            echo
+            echo "Generated files:"
+            find data/output/platform_handoffs -type f | sort | sed 's/^/- /'
+          } >> "$GITHUB_STEP_SUMMARY"

scripts/publish_ibit_btc_platform_handoff.py

@@ -0,0 +1,255 @@
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import subprocess
+import sys
+import urllib.error
+import urllib.request
+from datetime import UTC, date, datetime, timedelta
+from pathlib import Path
+
+BINANCE_BTCUSDT_DAILY_URL = (
+    "https://api.binance.com/api/v3/klines?symbol=BTCUSDT&interval=1d&limit=800"
+)
+DEFAULT_CONSUMER = "us_equity:ibit_smart_dca"
+DEFAULT_STRATEGY = "ibit_smart_dca"
+DEFAULT_GCS_PREFIX = "gs://qsl-runtime-logs-shared/platform_handoffs"
+
+
+def default_as_of(*, today: date | None = None) -> str:
+    """Return yesterday UTC as the default signal as-of date."""
+    current = today or datetime.now(UTC).date()
+    return (current - timedelta(days=1)).isoformat()
+
+
+def resolve_as_of(*, csv_path: Path, requested: str | None, today: date | None = None) -> str:
+    """Pick the requested as-of when present in the CSV, otherwise the latest row."""
+    rows = _read_csv_dates(csv_path)
+    if not rows:
+        raise ValueError(f"no dates found in {csv_path}")
+    target = requested or default_as_of(today=today)
+    if target in rows:
+        return target
+    return rows[-1]
+
+
+def fetch_binance_btc_daily_csv(output_path: Path) -> int:
+    """Download BTCUSDT daily OHLCV from Binance public API into a local CSV."""
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    try:
+        with urllib.request.urlopen(BINANCE_BTCUSDT_DAILY_URL, timeout=30) as response:
+            payload = json.loads(response.read().decode())
+    except urllib.error.URLError as exc:
+        raise RuntimeError(f"failed to download Binance BTCUSDT daily klines: {exc}") from exc
+
+    if not payload:
+        raise RuntimeError("Binance BTCUSDT daily klines response was empty")
+
+    rows: list[dict[str, object]] = []
+    for entry in payload:
+        timestamp_ms = int(entry[0])
+        day = datetime.fromtimestamp(timestamp_ms / 1000, tz=UTC).date().isoformat()
+        rows.append(
+            {
+                "date": day,
+                "open": float(entry[1]),
+                "high": float(entry[2]),
+                "low": float(entry[3]),
+                "close": float(entry[4]),
+                "volume": float(entry[5]),
+            }
+        )
+
+    with output_path.open("w", newline="", encoding="utf-8") as handle:
+        writer = csv.DictWriter(
+            handle,
+            fieldnames=["date", "open", "high", "low", "close", "volume"],
+        )
+        writer.writeheader()
+        writer.writerows(rows)
+
+    return len(rows)
+
+
+def build_ibit_btc_platform_handoff(
+    *,
+    work_dir: Path,
+    input_csv: Path,
+    as_of: str,
+    code_commit: str,
+    source_version: str,
+    consumer: str = DEFAULT_CONSUMER,
+    strategy: str = DEFAULT_STRATEGY,
+) -> dict[str, Path]:
+    """Build BTC cycle bundle artifacts and the platform handoff index locally."""
+    from market_signal_sources.cli.build_btc_cycle_bundle import main as build_bundle_main
+    from market_signal_sources.cli.publish_platform_signal_handoff import (
+        main as publish_handoff_main,
+    )
+
+    publication_dir = work_dir / "platform_handoffs" / as_of
+    bundle_dir = publication_dir / "bundle"
+    index_path = work_dir / "platform_handoffs" / "index.json"
+    bundle_dir.mkdir(parents=True, exist_ok=True)
+    generated_at = f"{as_of}T00:15:00Z"
+
+    build_exit = build_bundle_main(
+        [
+            "--input-csv",
+            str(input_csv),
+            "--output-dir",
+            str(bundle_dir),
+            "--as-of",
+            as_of,
+            "--provider",
+            "binance_public",
+            "--provider-dataset",
+            "btcusdt_daily_klines",
+            "--source-version",
+            source_version,
+            "--code-commit",
+            code_commit,
+            "--generated-at",
+            generated_at,
+        ]
+    )
+    if build_exit != 0:
+        raise RuntimeError(f"build-btc-cycle-bundle failed with exit code {build_exit}")
+
+    publish_exit = publish_handoff_main(
+        [
+            "--publication-dir",
+            str(publication_dir),
+            "--signal-bundle-manifest",
+            str(bundle_dir / "manifest.json"),
+            "--consumer",
+            consumer,
+            "--strategy",
+            strategy,
+            "--index-path",
+            str(index_path),
+            "--lookup-as-of",
+            as_of,
+        ]
+    )
+    if publish_exit != 0:
+        raise RuntimeError(
+            f"publish-platform-signal-handoff failed with exit code {publish_exit}"
+        )
+
+    return {
+        "publication_dir": publication_dir,
+        "index_path": index_path,
+        "bundle_manifest": bundle_dir / "manifest.json",
+    }
+
+
+def upload_platform_handoffs(*, local_root: Path, gcs_prefix: str) -> None:
+    """Sync the local platform handoffs tree to GCS."""
+    normalized = gcs_prefix.rstrip("/")
+    command = ["gsutil", "-m", "rsync", "-r", str(local_root), normalized]
+    completed = subprocess.run(command, check=False, capture_output=True, text=True)
+    if completed.returncode != 0:
+        message = completed.stderr.strip() or completed.stdout.strip() or "unknown gsutil error"
+        raise RuntimeError(f"gsutil rsync failed: {message}")
+
+
+def _read_csv_dates(csv_path: Path) -> list[str]:
+    with csv_path.open(newline="", encoding="utf-8") as handle:
+        reader = csv.DictReader(handle)
+        if "date" not in (reader.fieldnames or []):
+            raise ValueError(f"{csv_path} is missing a date column")
+        return [str(row["date"]) for row in reader if row.get("date")]
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Fetch BTC daily OHLCV, build the IBIT smart DCA platform handoff, "
+            "and optionally upload it to GCS."
+        )
+    )
+    parser.add_argument(
+        "--work-dir",
+        type=Path,
+        default=Path("data/output"),
+        help="Local build root for platform handoff artifacts.",
+    )
+    parser.add_argument(
+        "--as-of",
+        help="Signal as-of date (YYYY-MM-DD). Defaults to yesterday UTC when present in the CSV.",
+    )
+    parser.add_argument(
+        "--input-csv",
+        type=Path,
+        help="Optional pre-fetched BTC OHLCV CSV. When omitted, Binance public data is downloaded.",
+    )
+    parser.add_argument(
+        "--gcs-prefix",
+        default=DEFAULT_GCS_PREFIX,
+        help="GCS prefix for platform handoffs.",
+    )
+    parser.add_argument(
+        "--execute",
+        action="store_true",
+        help="Upload the generated platform handoffs directory to GCS.",
+    )
+    parser.add_argument(
+        "--source-version",
+        default="0.1.1",
+        help="MarketSignalSources package version recorded in bundle provenance.",
+    )
+    parser.add_argument(
+        "--code-commit",
+        help="Git commit SHA recorded in bundle provenance. Defaults to GITHUB_SHA when set.",
+    )
+    parser.add_argument(
+        "--consumer",
+        default=DEFAULT_CONSUMER,
+        help="Runtime consumer contract to publish.",
+    )
+    parser.add_argument(
+        "--strategy",
+        default=DEFAULT_STRATEGY,
+        help="Platform strategy profile for runtime adapter config.",
+    )
+    args = parser.parse_args(argv)
+
+    work_dir = args.work_dir.resolve()
+    work_dir.mkdir(parents=True, exist_ok=True)
+    input_csv = args.input_csv or (work_dir / "inputs" / "btc_daily.csv")
+    code_commit = args.code_commit or __import__("os").environ.get("GITHUB_SHA", "0" * 40)
+
+    if args.input_csv is None:
+        row_count = fetch_binance_btc_daily_csv(input_csv)
+        print(f"downloaded {row_count} BTCUSDT daily rows to {input_csv}")
+    elif not input_csv.is_file():
+        print(f"error: input CSV not found: {input_csv}", file=sys.stderr)
+        return 2
+
+    as_of = resolve_as_of(csv_path=input_csv, requested=args.as_of)
+    print(f"using as_of={as_of}")
+
+    artifacts = build_ibit_btc_platform_handoff(
+        work_dir=work_dir,
+        input_csv=input_csv,
+        as_of=as_of,
+        code_commit=code_commit,
+        source_version=args.source_version,
+        consumer=args.consumer,
+        strategy=args.strategy,
+    )
+    print(f"built platform handoff index at {artifacts['index_path']}")
+
+    if args.execute:
+        upload_root = work_dir / "platform_handoffs"
+        upload_platform_handoffs(local_root=upload_root, gcs_prefix=args.gcs_prefix)
+        print(f"uploaded {upload_root} to {args.gcs_prefix.rstrip('/')}")
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())