Support GCS feature snapshot artifacts

Author: Pigbibi

application/feature_snapshot_service.py

@@ -4,6 +4,7 @@
 
 import hashlib
 import json
+import tempfile
 from dataclasses import dataclass
 from datetime import timezone
 from pathlib import Path
@@ -15,6 +16,7 @@
 DEFAULT_SNAPSHOT_DATE_COLUMNS = ("as_of", "snapshot_date")
 DEFAULT_MAX_SNAPSHOT_MONTH_LAG = 1
 DEFAULT_SNAPSHOT_MANIFEST_SUFFIX = ".manifest.json"
+DEFAULT_ARTIFACT_CACHE_DIR = Path(tempfile.gettempdir()) / "quant_strategy_artifacts"
 
 
 @dataclass(frozen=True)
@@ -99,6 +101,56 @@ def _resolve_manifest_path(snapshot_path: Path, manifest_path: str | None) -> Pa
     return Path(f"{snapshot_path}{DEFAULT_SNAPSHOT_MANIFEST_SUFFIX}")
 
 
+def _is_gcs_uri(reference: str | None) -> bool:
+    return str(reference or "").strip().startswith("gs://")
+
+
+def _resolve_manifest_reference(snapshot_reference: str, manifest_path: str | None) -> str:
+    raw_manifest = str(manifest_path or "").strip()
+    if raw_manifest:
+        return raw_manifest
+    return f"{str(snapshot_reference).strip()}{DEFAULT_SNAPSHOT_MANIFEST_SUFFIX}"
+
+
+def _parse_gcs_uri(uri: str) -> tuple[str, str]:
+    raw_uri = str(uri or "").strip()
+    if not raw_uri.startswith("gs://"):
+        raise ValueError(f"Unsupported GCS URI: {raw_uri}")
+    bucket_name, _, object_name = raw_uri[5:].partition("/")
+    if not bucket_name or not object_name:
+        raise ValueError(f"Invalid GCS URI: {raw_uri}")
+    return bucket_name, object_name
+
+
+def _download_gcs_object(uri: str, destination: Path) -> None:
+    from google.cloud import storage
+
+    bucket_name, object_name = _parse_gcs_uri(uri)
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    client = storage.Client()
+    client.bucket(bucket_name).blob(object_name).download_to_filename(str(destination))
+
+
+def _cache_path_for_remote_artifact(reference: str) -> Path:
+    raw_reference = str(reference or "").strip()
+    digest = hashlib.sha256(raw_reference.encode("utf-8")).hexdigest()[:16]
+    leaf_name = Path(raw_reference).name or "artifact"
+    return DEFAULT_ARTIFACT_CACHE_DIR / digest / leaf_name
+
+
+def _materialize_artifact_path(reference: str) -> tuple[Path, dict[str, object]]:
+    raw_reference = str(reference or "").strip()
+    if not raw_reference:
+        raise ValueError("artifact reference is required")
+
+    if not _is_gcs_uri(raw_reference):
+        return Path(raw_reference), {"source_uri": None, "local_path": raw_reference}
+
+    local_path = _cache_path_for_remote_artifact(raw_reference)
+    _download_gcs_object(raw_reference, local_path)
+    return local_path, {"source_uri": raw_reference, "local_path": str(local_path)}
+
+
 def _normalize_manifest_payload(payload: dict[str, object]) -> dict[str, object]:
     normalized = dict(payload)
     if "snapshot_as_of" in normalized:
@@ -117,7 +169,12 @@ def load_feature_snapshot(path: str) -> pd.DataFrame:
     raw_path = str(path or "").strip()
     if not raw_path:
         raise EnvironmentError("Feature snapshot path is required")
-    snapshot_path = Path(raw_path)
+    try:
+        snapshot_path, _ = _materialize_artifact_path(raw_path)
+    except Exception as exc:
+        raise FileNotFoundError(
+            f"Feature snapshot unavailable: {raw_path} ({type(exc).__name__}: {exc})"
+        ) from exc
     if not snapshot_path.exists():
         raise FileNotFoundError(f"Feature snapshot not found: {snapshot_path}")
     return _load_snapshot_frame(snapshot_path)
@@ -150,6 +207,77 @@ def load_feature_snapshot_guarded(
             ),
         )
 
+    manifest_reference = _resolve_manifest_reference(raw_path, manifest_path)
+    if _is_gcs_uri(raw_path) or _is_gcs_uri(manifest_reference):
+        try:
+            local_snapshot_path, snapshot_artifact_metadata = _materialize_artifact_path(raw_path)
+        except Exception as exc:
+            return FeatureSnapshotGuardResult(
+                frame=None,
+                metadata=_build_guard_metadata(
+                    snapshot_path=raw_path,
+                    decision="fail_closed",
+                    snapshot_exists=False,
+                    snapshot_source_uri=raw_path if _is_gcs_uri(raw_path) else None,
+                    fail_reason=f"feature_snapshot_download_failed:{type(exc).__name__}:{exc}",
+                ),
+            )
+
+        local_manifest_path = None
+        manifest_artifact_metadata = {
+            "source_uri": manifest_reference if _is_gcs_uri(manifest_reference) else None,
+            "local_path": manifest_reference,
+        }
+        manifest_download_error = None
+        try:
+            local_manifest_path, manifest_artifact_metadata = _materialize_artifact_path(
+                manifest_reference
+            )
+        except Exception as exc:
+            manifest_download_error = f"{type(exc).__name__}:{exc}"
+            if require_manifest:
+                return FeatureSnapshotGuardResult(
+                    frame=None,
+                    metadata=_build_guard_metadata(
+                        snapshot_path=raw_path,
+                        decision="fail_closed",
+                        snapshot_exists=True,
+                        snapshot_source_uri=snapshot_artifact_metadata.get("source_uri"),
+                        snapshot_local_path=snapshot_artifact_metadata.get("local_path"),
+                        snapshot_manifest_path=manifest_reference,
+                        snapshot_manifest_exists=False,
+                        snapshot_manifest_source_uri=manifest_artifact_metadata.get("source_uri"),
+                        snapshot_manifest_download_error=manifest_download_error,
+                        fail_reason=f"feature_snapshot_manifest_download_failed:{manifest_download_error}",
+                    ),
+                )
+
+        result = load_feature_snapshot_guarded(
+            str(local_snapshot_path),
+            run_as_of=run_as_of,
+            required_columns=required_columns,
+            snapshot_date_columns=snapshot_date_columns,
+            max_snapshot_month_lag=max_snapshot_month_lag,
+            manifest_path=str(local_manifest_path) if local_manifest_path is not None else None,
+            require_manifest=require_manifest,
+            expected_strategy_profile=expected_strategy_profile,
+            expected_config_name=expected_config_name,
+            expected_config_path=expected_config_path,
+            expected_contract_version=expected_contract_version,
+        )
+        metadata = dict(result.metadata)
+        metadata["feature_snapshot_path"] = raw_path
+        metadata["snapshot_path"] = raw_path
+        metadata["snapshot_source_uri"] = snapshot_artifact_metadata.get("source_uri")
+        metadata["snapshot_local_path"] = snapshot_artifact_metadata.get("local_path")
+        metadata["snapshot_manifest_path"] = manifest_reference
+        metadata["snapshot_manifest_source_uri"] = manifest_artifact_metadata.get("source_uri")
+        metadata["snapshot_manifest_local_path"] = manifest_artifact_metadata.get("local_path")
+        if manifest_download_error is not None:
+            metadata["snapshot_manifest_download_error"] = manifest_download_error
+            metadata["snapshot_manifest_exists"] = False
+        return FeatureSnapshotGuardResult(frame=result.frame, metadata=metadata)
+
     snapshot_path = Path(raw_path)
     manifest_file = _resolve_manifest_path(snapshot_path, manifest_path)
     file_timestamp = None

tests/test_feature_snapshot_service.py

@@ -5,6 +5,7 @@
 import unittest
 from pathlib import Path
 from tempfile import TemporaryDirectory
+from unittest.mock import patch
 
 
 def _sha256_file(path: Path) -> str:
@@ -145,6 +146,87 @@ def test_load_feature_snapshot_guarded_validates_manifest_checksums(self):
         self.assertEqual(result.metadata["snapshot_manifest_strategy_profile"], "tech_pullback_cash_buffer")
         self.assertEqual(result.metadata["snapshot_manifest_config_name"], "tech_pullback_cash_buffer")
 
+    def test_load_feature_snapshot_downloads_gcs_csv(self):
+        _skip_if_missing_pandas()
+        from application.feature_snapshot_service import load_feature_snapshot
+
+        with TemporaryDirectory() as tmp_dir:
+            source_uri = "gs://unit-test-bucket/snapshots/tech_pullback.csv"
+
+            def fake_download(uri: str, destination: Path) -> None:
+                self.assertEqual(uri, source_uri)
+                destination.parent.mkdir(parents=True, exist_ok=True)
+                destination.write_text("symbol,sector,mom_6_1\nAAA,tech,0.1\n", encoding="utf-8")
+
+            with patch("application.feature_snapshot_service._download_gcs_object", side_effect=fake_download):
+                frame = load_feature_snapshot(source_uri)
+
+        self.assertEqual(frame.to_dict(orient="records"), [{"symbol": "AAA", "sector": "tech", "mom_6_1": 0.1}])
+
+    def test_load_feature_snapshot_guarded_downloads_gcs_snapshot_and_manifest(self):
+        _skip_if_missing_pandas()
+        from application.feature_snapshot_service import load_feature_snapshot_guarded
+
+        with TemporaryDirectory() as tmp_dir:
+            config_path = Path(tmp_dir) / "config.json"
+            config_path.write_text(json.dumps({"name": "tech_pullback_cash_buffer"}), encoding="utf-8")
+
+            snapshot_uri = "gs://unit-test-bucket/snapshots/tech_pullback_cash_buffer_feature_snapshot_latest.csv"
+            manifest_uri = f"{snapshot_uri}.manifest.json"
+            downloaded_snapshot_path: Path | None = None
+
+            def fake_download(uri: str, destination: Path) -> None:
+                nonlocal downloaded_snapshot_path
+                destination.parent.mkdir(parents=True, exist_ok=True)
+                if uri == snapshot_uri:
+                    downloaded_snapshot_path = destination
+                    destination.write_text(
+                        "as_of,symbol,sector,mom_6_1\n2026-03-31,AAA,Information Technology,0.1\n",
+                        encoding="utf-8",
+                    )
+                    return
+                if uri == manifest_uri:
+                    self.assertIsNotNone(downloaded_snapshot_path)
+                    destination.write_text(
+                        json.dumps(
+                            {
+                                "contract_version": "tech_pullback_cash_buffer.feature_snapshot.v1",
+                                "strategy_profile": "tech_pullback_cash_buffer",
+                                "config_name": "tech_pullback_cash_buffer",
+                                "config_path": str(config_path),
+                                "config_sha256": _sha256_file(config_path),
+                                "snapshot_path": snapshot_uri,
+                                "snapshot_sha256": _sha256_file(downloaded_snapshot_path),
+                                "snapshot_as_of": "2026-03-31",
+                            }
+                        ),
+                        encoding="utf-8",
+                    )
+                    return
+                raise AssertionError(f"unexpected uri: {uri}")
+
+            with patch("application.feature_snapshot_service._download_gcs_object", side_effect=fake_download):
+                result = load_feature_snapshot_guarded(
+                    snapshot_uri,
+                    run_as_of="2026-04-01",
+                    required_columns=("as_of", "symbol", "sector", "mom_6_1"),
+                    require_manifest=True,
+                    expected_strategy_profile="tech_pullback_cash_buffer",
+                    expected_config_name="tech_pullback_cash_buffer",
+                    expected_config_path=str(config_path),
+                    expected_contract_version="tech_pullback_cash_buffer.feature_snapshot.v1",
+                )
+
+        self.assertIsNotNone(result.frame)
+        self.assertEqual(result.metadata["snapshot_guard_decision"], "proceed")
+        self.assertEqual(
+            result.metadata["feature_snapshot_path"],
+            snapshot_uri,
+        )
+        self.assertEqual(result.metadata["snapshot_manifest_path"], manifest_uri)
+        self.assertEqual(result.metadata["snapshot_source_uri"], snapshot_uri)
+        self.assertEqual(result.metadata["snapshot_manifest_source_uri"], manifest_uri)
+
 
 if __name__ == "__main__":
     unittest.main()