Add market signal artifact tree materializer (#104)

Author: Pigbibi

src/quant_platform_kit/common/__init__.py

@@ -40,6 +40,13 @@
     build_runtime_target,
     resolve_runtime_target_from_env,
 )
+from .market_signal_artifacts import (
+    MARKET_SIGNAL_ARTIFACT_LINK_FIELDS,
+    cache_root_for_market_signal_artifact_tree,
+    local_path_for_gcs_object,
+    materialize_market_signal_artifact_tree,
+    resolve_gcs_artifact_reference,
+)
 from .strategy_plugins import (
     CRISIS_RESPONSE_SHADOW_SUPPORTED_STRATEGIES,
     DEFAULT_STRATEGY_PLUGIN_DEFINITIONS,
@@ -123,11 +130,15 @@
     "SUPPORTED_STRATEGY_PLUGIN_MODES",
     "MACRO_RISK_GOVERNOR_SUPPORTED_STRATEGIES",
     "MARKET_REGIME_CONTROL_SUPPORTED_STRATEGIES",
+    "MARKET_SIGNAL_ARTIFACT_LINK_FIELDS",
     "TACO_REBOUND_SHADOW_SUPPORTED_STRATEGIES",
     "filter_execution_blocking_skips",
     "is_terminal_funding_block",
     "is_terminal_strategy_run_stage",
     "localize_notification_text",
+    "cache_root_for_market_signal_artifact_tree",
+    "local_path_for_gcs_object",
+    "materialize_market_signal_artifact_tree",
     "merge_strategy_plugin_i18n",
     "normalize_skip_reason",
     "normalize_stage",
@@ -162,6 +173,7 @@
     "parse_strategy_plugin_notification_targets",
     "parse_strategy_plugin_mounts",
     "resolve_runtime_target_from_env",
+    "resolve_gcs_artifact_reference",
     "should_alert_strategy_plugin_signal",
     "translate_strategy_plugin_value",
     "translator_uses_zh",

src/quant_platform_kit/common/market_signal_artifacts.py

@@ -0,0 +1,185 @@
+"""Materialization helpers for external market signal artifact trees."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import posixpath
+from pathlib import Path
+from typing import Any, Iterable, Mapping
+
+from .strategy_plugin_artifacts import download_gcs_object, parse_gcs_uri
+
+
+MARKET_SIGNAL_ARTIFACT_LINK_FIELDS = frozenset(
+    {
+        "bundle_path",
+        "catalog_path",
+        "consumer_contract_registry_manifest_path",
+        "handoff_manifest_path",
+        "quality_report_path",
+        "registry_path",
+        "signal_bundle_manifest_path",
+        "source_family_catalog_manifest_path",
+    }
+)
+
+
+def materialize_market_signal_artifact_tree(
+    reference: str,
+    *,
+    cache_dir: Path,
+    client_factory: Any = None,
+    link_fields: Iterable[str] | None = None,
+) -> tuple[Path, dict[str, Any]]:
+    """Return a local path for a market signal artifact and its linked JSON tree."""
+
+    raw_reference = _required_string(reference, field_name="reference")
+    if not raw_reference.startswith("gs://"):
+        local_path = Path(raw_reference).expanduser()
+        return local_path, {
+            "source_uri": None,
+            "local_path": raw_reference,
+            "cache_dir": None,
+            "materialized_count": 0,
+            "materialized_paths": (),
+        }
+
+    fields = frozenset(link_fields or MARKET_SIGNAL_ARTIFACT_LINK_FIELDS)
+    cache_root = cache_root_for_market_signal_artifact_tree(
+        raw_reference,
+        cache_dir=cache_dir,
+    )
+    visited: dict[str, Path] = {}
+    _materialize_gcs_json_tree(
+        raw_reference,
+        cache_root=cache_root,
+        client_factory=client_factory,
+        link_fields=fields,
+        visited=visited,
+    )
+    _, object_name = parse_gcs_uri(raw_reference)
+    local_path = local_path_for_gcs_object(cache_root, object_name)
+    return local_path, {
+        "source_uri": raw_reference,
+        "local_path": str(local_path),
+        "cache_dir": str(cache_root),
+        "materialized_count": len(visited),
+        "materialized_paths": tuple(str(path) for path in visited.values()),
+    }
+
+
+def cache_root_for_market_signal_artifact_tree(reference: str, *, cache_dir: Path) -> Path:
+    raw_reference = _required_string(reference, field_name="reference")
+    digest = hashlib.sha256(raw_reference.encode("utf-8")).hexdigest()[:16]
+    return cache_dir / digest
+
+
+def local_path_for_gcs_object(cache_root: Path, object_name: str) -> Path:
+    raw_object_name = _required_string(object_name, field_name="object_name")
+    if raw_object_name.startswith("/"):
+        raise ValueError(f"GCS object name must be relative: {raw_object_name}")
+    normalized = posixpath.normpath(raw_object_name)
+    if normalized in {"", ".", ".."} or normalized.startswith("../"):
+        raise ValueError(f"GCS object name escapes the cache root: {raw_object_name}")
+    return cache_root.joinpath(*normalized.split("/"))
+
+
+def resolve_gcs_artifact_reference(base_uri: str, reference: str) -> str:
+    raw_reference = _required_string(reference, field_name="reference")
+    if raw_reference.startswith("gs://"):
+        parse_gcs_uri(raw_reference)
+        return raw_reference
+    if "://" in raw_reference:
+        raise ValueError(f"Unsupported market signal artifact reference: {raw_reference}")
+    if raw_reference.startswith("/"):
+        raise ValueError(
+            "GCS market signal artifacts must use relative linked paths or gs:// URIs: "
+            f"{raw_reference}"
+        )
+
+    bucket_name, object_name = parse_gcs_uri(base_uri)
+    base_dir = posixpath.dirname(object_name)
+    resolved = posixpath.normpath(posixpath.join(base_dir, raw_reference))
+    if resolved in {"", ".", ".."} or resolved.startswith("../"):
+        raise ValueError(
+            "GCS market signal artifact reference escapes the bucket root: "
+            f"{raw_reference}"
+        )
+    return f"gs://{bucket_name}/{resolved}"
+
+
+def _materialize_gcs_json_tree(
+    uri: str,
+    *,
+    cache_root: Path,
+    client_factory: Any,
+    link_fields: frozenset[str],
+    visited: dict[str, Path],
+) -> None:
+    if uri in visited:
+        return
+
+    _, object_name = parse_gcs_uri(uri)
+    local_path = local_path_for_gcs_object(cache_root, object_name)
+    download_gcs_object(uri, local_path, client_factory=client_factory)
+    visited[uri] = local_path
+
+    payload = _read_json_object(local_path)
+    if payload is None:
+        return
+    for linked_uri in _iter_linked_gcs_artifact_uris(
+        payload,
+        base_uri=uri,
+        link_fields=link_fields,
+    ):
+        _materialize_gcs_json_tree(
+            linked_uri,
+            cache_root=cache_root,
+            client_factory=client_factory,
+            link_fields=link_fields,
+            visited=visited,
+        )
+
+
+def _read_json_object(path: Path) -> Mapping[str, Any] | list[Any] | None:
+    if path.suffix.lower() != ".json":
+        return None
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError as exc:
+        raise ValueError(f"Invalid JSON market signal artifact: {path}") from exc
+    if not isinstance(payload, (dict, list)):
+        return None
+    return payload
+
+
+def _iter_linked_gcs_artifact_uris(
+    payload: Any,
+    *,
+    base_uri: str,
+    link_fields: frozenset[str],
+) -> Iterable[str]:
+    if isinstance(payload, Mapping):
+        for key, value in payload.items():
+            if key in link_fields and isinstance(value, str) and value.strip():
+                yield resolve_gcs_artifact_reference(base_uri, value.strip())
+            yield from _iter_linked_gcs_artifact_uris(
+                value,
+                base_uri=base_uri,
+                link_fields=link_fields,
+            )
+    elif isinstance(payload, list):
+        for item in payload:
+            yield from _iter_linked_gcs_artifact_uris(
+                item,
+                base_uri=base_uri,
+                link_fields=link_fields,
+            )
+
+
+def _required_string(value: Any, *, field_name: str) -> str:
+    text = str(value or "").strip()
+    if not text:
+        raise ValueError(f"{field_name} must be a non-empty string")
+    return text

tests/test_market_signal_artifacts.py

@@ -0,0 +1,159 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from quant_platform_kit.common.market_signal_artifacts import (
+    cache_root_for_market_signal_artifact_tree,
+    local_path_for_gcs_object,
+    materialize_market_signal_artifact_tree,
+    resolve_gcs_artifact_reference,
+)
+
+
+class _FakeBlob:
+    def __init__(self, payloads: dict[str, str], key: str) -> None:
+        self._payloads = payloads
+        self._key = key
+
+    def download_to_filename(self, destination: str) -> None:
+        Path(destination).parent.mkdir(parents=True, exist_ok=True)
+        Path(destination).write_text(self._payloads[self._key], encoding="utf-8")
+
+
+class _FakeBucket:
+    def __init__(self, payloads: dict[str, str], bucket_name: str) -> None:
+        self._payloads = payloads
+        self._bucket_name = bucket_name
+
+    def blob(self, object_name: str) -> _FakeBlob:
+        return _FakeBlob(self._payloads, f"gs://{self._bucket_name}/{object_name}")
+
+
+class _FakeClient:
+    def __init__(self, payloads: dict[str, str]) -> None:
+        self._payloads = payloads
+
+    def bucket(self, bucket_name: str) -> _FakeBucket:
+        return _FakeBucket(self._payloads, bucket_name)
+
+
+def _json(payload: object) -> str:
+    return json.dumps(payload)
+
+
+def test_materialize_local_market_signal_artifact_tree_does_not_download():
+    local_path, metadata = materialize_market_signal_artifact_tree(
+        "~/signals/platform_handoff_index.json",
+        cache_dir=Path("/tmp/cache"),
+        client_factory=lambda: (_ for _ in ()).throw(AssertionError("should not download")),
+    )
+
+    assert local_path == Path("~/signals/platform_handoff_index.json").expanduser()
+    assert metadata == {
+        "source_uri": None,
+        "local_path": "~/signals/platform_handoff_index.json",
+        "cache_dir": None,
+        "materialized_count": 0,
+        "materialized_paths": (),
+    }
+
+
+def test_materialize_gcs_market_signal_artifact_tree_downloads_linked_artifacts(tmp_path):
+    payloads = {
+        "gs://signals/live/platform_handoffs/index.json": _json(
+            {
+                "handoffs": [
+                    {
+                        "handoff_manifest_path": "2026-06-19/platform_handoff.json",
+                    }
+                ],
+            }
+        ),
+        "gs://signals/live/platform_handoffs/2026-06-19/platform_handoff.json": _json(
+            {
+                "signal_bundle_manifest_path": "../bundles/manifest.json",
+                "source_family_catalog_manifest_path": "../catalog/source.manifest.json",
+                "consumer_contract_registry_manifest_path": "../contracts/registry.manifest.json",
+            }
+        ),
+        "gs://signals/live/platform_handoffs/bundles/manifest.json": _json(
+            {
+                "bundle_path": "signal_bundle.json",
+                "quality_report_path": "quality_report.json",
+            }
+        ),
+        "gs://signals/live/platform_handoffs/bundles/signal_bundle.json": _json(
+            {"schema_version": "test.signal_bundle.v1"}
+        ),
+        "gs://signals/live/platform_handoffs/bundles/quality_report.json": _json(
+            {"schema_version": "test.quality_report.v1"}
+        ),
+        "gs://signals/live/platform_handoffs/catalog/source.manifest.json": _json(
+            {"catalog_path": "signal_source_families.json"}
+        ),
+        "gs://signals/live/platform_handoffs/catalog/signal_source_families.json": _json(
+            {"schema_version": "test.source_catalog.v1"}
+        ),
+        "gs://signals/live/platform_handoffs/contracts/registry.manifest.json": _json(
+            {"registry_path": "market_signal_consumers.json"}
+        ),
+        "gs://signals/live/platform_handoffs/contracts/market_signal_consumers.json": _json(
+            {"schema_version": "test.consumer_registry.v1"}
+        ),
+    }
+
+    local_path, metadata = materialize_market_signal_artifact_tree(
+        "gs://signals/live/platform_handoffs/index.json",
+        cache_dir=tmp_path,
+        client_factory=lambda: _FakeClient(payloads),
+    )
+
+    cache_root = cache_root_for_market_signal_artifact_tree(
+        "gs://signals/live/platform_handoffs/index.json",
+        cache_dir=tmp_path,
+    )
+    assert local_path == cache_root / "live" / "platform_handoffs" / "index.json"
+    assert metadata["source_uri"] == "gs://signals/live/platform_handoffs/index.json"
+    assert metadata["local_path"] == str(local_path)
+    assert metadata["cache_dir"] == str(cache_root)
+    assert metadata["materialized_count"] == len(payloads)
+    assert set(metadata["materialized_paths"]) == {
+        str(local_path_for_gcs_object(cache_root, uri.removeprefix("gs://signals/")))
+        for uri in payloads
+    }
+
+    for uri in payloads:
+        object_name = uri.removeprefix("gs://signals/")
+        assert local_path_for_gcs_object(cache_root, object_name).exists()
+
+
+def test_resolve_gcs_artifact_reference_rejects_non_portable_paths():
+    assert (
+        resolve_gcs_artifact_reference(
+            "gs://bucket/root/platform_handoffs/index.json",
+            "2026-06-19/platform_handoff.json",
+        )
+        == "gs://bucket/root/platform_handoffs/2026-06-19/platform_handoff.json"
+    )
+    assert (
+        resolve_gcs_artifact_reference(
+            "gs://bucket/root/platform_handoffs/2026-06-19/platform_handoff.json",
+            "../bundles/manifest.json",
+        )
+        == "gs://bucket/root/platform_handoffs/bundles/manifest.json"
+    )
+
+    with pytest.raises(ValueError, match="relative linked paths"):
+        resolve_gcs_artifact_reference(
+            "gs://bucket/root/platform_handoffs/index.json",
+            "/tmp/signal_bundle_manifest.json",
+        )
+
+    with pytest.raises(ValueError, match="escapes the bucket root"):
+        resolve_gcs_artifact_reference(
+            "gs://bucket/root/platform_handoffs/index.json",
+            "../../../signal_bundle_manifest.json",
+        )