Merge pull request #512 from QuantStrategyLab/fix/managed-symbol-extractor-compat

fix: adapt managed-symbol extractor safe-haven keywords

Author: Pigbibi

src/quant_platform_kit/common/feature_snapshot_runtime.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import inspect
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from typing import Any, Callable, Mapping
@@ -266,15 +267,57 @@ def extract_feature_snapshot_managed_symbols(
         return tuple(
             extractor(
                 feature_snapshot,
-                benchmark_symbol=benchmark_symbol,
-                safe_haven=safe_haven_symbol,
+                **_managed_symbols_extractor_kwargs(
+                    extractor,
+                    benchmark_symbol=benchmark_symbol,
+                    safe_haven_symbol=safe_haven_symbol,
+                ),
             )
         )
     if safe_haven_symbol:
         return (safe_haven_symbol,)
     return fallback_symbols
 
 
+def _managed_symbols_extractor_kwargs(
+    extractor: Callable[..., Any],
+    *,
+    benchmark_symbol: str,
+    safe_haven_symbol: str | None,
+) -> dict[str, Any]:
+    """Adapt the two reviewed safe-haven keyword spellings at the plug-in edge.
+
+    Strategy plug-ins are independently versioned. The historical contract
+    accepted ``safe_haven`` while the global-ETF extractor uses the more
+    explicit ``safe_haven_symbol``. Inspecting the callable before calling it
+    preserves both contracts without catching a TypeError raised *inside* the
+    plug-in itself.
+    """
+
+    try:
+        parameters = inspect.signature(extractor).parameters.values()
+    except (TypeError, ValueError):
+        # Opaque callables retain the original, documented keyword spelling.
+        return {
+            "benchmark_symbol": benchmark_symbol,
+            "safe_haven": safe_haven_symbol,
+        }
+
+    accepted = {parameter.name for parameter in parameters}
+    accepts_kwargs = any(
+        parameter.kind is inspect.Parameter.VAR_KEYWORD
+        for parameter in parameters
+    )
+    kwargs: dict[str, Any] = {}
+    if "benchmark_symbol" in accepted or accepts_kwargs:
+        kwargs["benchmark_symbol"] = benchmark_symbol
+    if "safe_haven" in accepted or accepts_kwargs:
+        kwargs["safe_haven"] = safe_haven_symbol
+    elif "safe_haven_symbol" in accepted:
+        kwargs["safe_haven_symbol"] = safe_haven_symbol
+    return kwargs
+
+
 def _apply_runtime_policy(
     runtime_config: dict[str, Any],
     runtime_adapter: StrategyRuntimeAdapter,

tests/test_feature_snapshot_runtime.py

@@ -8,6 +8,7 @@
     FeatureSnapshotContextRequest,
     FeatureSnapshotRuntimeSettings,
     evaluate_feature_snapshot_strategy,
+    extract_feature_snapshot_managed_symbols,
 )
 from quant_platform_kit.strategy_contracts import (
     CallableStrategyEntrypoint,
@@ -42,6 +43,62 @@ def _entrypoint() -> CallableStrategyEntrypoint:
 
 
 class FeatureSnapshotRuntimeTests(unittest.TestCase):
+    def test_managed_symbol_extractor_accepts_safe_haven_symbol_contract(self) -> None:
+        observed: dict[str, object] = {}
+
+        def extractor(
+            _snapshot: object,
+            *,
+            benchmark_symbol: str | None = None,
+            safe_haven_symbol: str | None = None,
+        ) -> tuple[str, ...]:
+            observed["benchmark_symbol"] = benchmark_symbol
+            observed["safe_haven_symbol"] = safe_haven_symbol
+            return ("VT", str(safe_haven_symbol))
+
+        symbols = extract_feature_snapshot_managed_symbols(
+            runtime_adapter=StrategyRuntimeAdapter(
+                managed_symbols_extractor=extractor,
+            ),
+            feature_snapshot=(),
+            benchmark_symbol="VOO",
+            safe_haven_symbol="BIL",
+        )
+
+        self.assertEqual(symbols, ("VT", "BIL"))
+        self.assertEqual(observed, {
+            "benchmark_symbol": "VOO",
+            "safe_haven_symbol": "BIL",
+        })
+
+    def test_managed_symbol_extractor_retains_legacy_safe_haven_contract(self) -> None:
+        observed: dict[str, object] = {}
+
+        def extractor(
+            _snapshot: object,
+            *,
+            benchmark_symbol: str | None = None,
+            safe_haven: str | None = None,
+        ) -> tuple[str, ...]:
+            observed["benchmark_symbol"] = benchmark_symbol
+            observed["safe_haven"] = safe_haven
+            return ("QQQ", str(safe_haven))
+
+        symbols = extract_feature_snapshot_managed_symbols(
+            runtime_adapter=StrategyRuntimeAdapter(
+                managed_symbols_extractor=extractor,
+            ),
+            feature_snapshot=(),
+            benchmark_symbol="QQQ",
+            safe_haven_symbol="BOXX",
+        )
+
+        self.assertEqual(symbols, ("QQQ", "BOXX"))
+        self.assertEqual(observed, {
+            "benchmark_symbol": "QQQ",
+            "safe_haven": "BOXX",
+        })
+
     def test_fail_closes_when_path_missing(self) -> None:
         result = evaluate_feature_snapshot_strategy(
             entrypoint=_entrypoint(),