Support leveraged combo shadow runtime

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

scripts/print_strategy_switch_env_plan.py

@@ -245,10 +245,16 @@ def build_switch_plan(
             [
                 "IBKR_FEATURE_SNAPSHOT_PATH",
                 "IBKR_FEATURE_SNAPSHOT_MANIFEST_PATH",
-                "IBKR_STRATEGY_CONFIG_PATH",
                 "IBKR_RECONCILIATION_OUTPUT_PATH",
             ]
         )
+        if config_source_policy == "env_only":
+            optional_env.append("IBKR_STRATEGY_CONFIG_PATH")
+            notes.append(
+                "IBKR_STRATEGY_CONFIG_PATH is optional for this profile; set it only for an explicit shadow/runtime config."
+            )
+        else:
+            remove_if_present.append("IBKR_STRATEGY_CONFIG_PATH")
 
     hints: dict[str, str] = {}
     if requires_feature_snapshot:
@@ -260,6 +266,11 @@ def build_switch_plan(
         hints["feature_snapshot_manifest_filename"] = manifest_filename
     if artifact_paths.bundled_config_path is not None:
         hints["bundled_strategy_config_path"] = str(artifact_paths.bundled_config_path)
+    if definition.profile == "us_equity_combo_leveraged":
+        hints["shadow_352045_strategy_config_path"] = (
+            "package://us_equity_strategies/configs/"
+            "us_equity_combo_leveraged_shadow_352045.json"
+        )
 
     return {
         "platform": IBKR_PLATFORM,

strategy_runtime.py

@@ -47,6 +47,7 @@
 DCA_PROFILES = frozenset({"nasdaq_sp500_smart_dca", "ibit_smart_dca"})
 IBIT_ZSCORE_EXIT_PROFILE = "ibit_smart_dca"
 _MARKET_HISTORY_INPUT = "market_history"
+_MARKET_DATA_INPUT = "market_data"
 _BENCHMARK_HISTORY_INPUT = "benchmark_history"
 _DERIVED_INDICATORS_INPUT = "derived_indicators"
 _PORTFOLIO_SNAPSHOT_INPUT = "portfolio_snapshot"
@@ -459,6 +460,61 @@ def _build_market_history_inputs(
             }
         }
 
+    def _build_direct_market_data_inputs(
+        self,
+        ib,
+        historical_close_loader: Callable[..., Any],
+    ) -> Mapping[str, Any]:
+        if self.profile != "us_equity_combo_leveraged":
+            raise ValueError(f"Unsupported market_data strategy profile {self.profile!r}")
+
+        trend_symbol = str(self.merged_runtime_config.get("market_data_trend_symbol") or "SPY").strip().upper()
+        ma_window = int(
+            self.merged_runtime_config.get(
+                "market_data_ma_window",
+                self.merged_runtime_config.get("sma_period", 200),
+            )
+        )
+        if ma_window <= 0:
+            raise ValueError("market_data_ma_window must be positive")
+        history = historical_close_loader(
+            ib,
+            trend_symbol,
+            duration=str(self.merged_runtime_config.get("market_data_history_duration") or "2 Y"),
+            bar_size=str(self.merged_runtime_config.get("market_data_history_bar_size") or "1 day"),
+        )
+        if isinstance(history, pd.DataFrame):
+            if "close" in history.columns:
+                close_values = history["close"]
+            else:
+                close_values = history.iloc[:, 0]
+        elif isinstance(history, pd.Series):
+            close_values = history
+        else:
+            close_values = [
+                item.get("close") if isinstance(item, Mapping) else getattr(item, "close", item)
+                for item in (history or ())
+            ]
+        close_series = pd.to_numeric(pd.Series(close_values), errors="coerce").dropna()
+        close_series = close_series[close_series > 0]
+        if len(close_series) < ma_window:
+            raise ValueError(
+                f"{self.profile} requires at least {ma_window} positive {trend_symbol} closes, "
+                f"got {len(close_series)}"
+            )
+        trend_price = float(close_series.iloc[-1])
+        trend_ma = float(close_series.tail(ma_window).mean())
+        return {
+            _MARKET_DATA_INPUT: {
+                "spy_above_ma200": trend_price > trend_ma,
+                "trend_symbol": trend_symbol,
+                "trend_price": trend_price,
+                "trend_ma": trend_ma,
+                "trend_ma_window": ma_window,
+                "history_observation_count": int(len(close_series)),
+            }
+        }
+
     def _build_strategy_context(
         self,
         *,
@@ -517,6 +573,16 @@ def evaluate(
                 pacing_sec=pacing_sec,
                 strategy_plugin_signals=strategy_plugin_signals,
             )
+        if _MARKET_DATA_INPUT in self.required_inputs:
+            return self._evaluate_direct_market_data_strategy(
+                ib=ib,
+                current_holdings=current_holdings,
+                historical_close_loader=historical_close_loader,
+                run_as_of=run_as_of,
+                translator=translator,
+                pacing_sec=pacing_sec,
+                strategy_plugin_signals=strategy_plugin_signals,
+            )
         if _MARKET_HISTORY_INPUT in self.required_inputs:
             return self._evaluate_market_data_strategy(
                 ib=ib,
@@ -547,6 +613,81 @@ def evaluate(
             f"{', '.join(sorted(self.required_inputs)) or '<none>'}"
         )
 
+    def _evaluate_direct_market_data_strategy(
+        self,
+        *,
+        ib,
+        current_holdings,
+        historical_close_loader: Callable[..., Any],
+        run_as_of: pd.Timestamp,
+        translator: Callable[[str], str],
+        pacing_sec: float,
+        strategy_plugin_signals=(),
+    ) -> StrategyEvaluationResult:
+        runtime_config = dict(self.runtime_config)
+        runtime_config.setdefault("translator", translator)
+        runtime_config.setdefault("pacing_sec", float(pacing_sec))
+        apply_runtime_policy_to_runtime_config(runtime_config, self.runtime_adapter)
+        portfolio_snapshot = self._fetch_portfolio_snapshot_for_context(
+            ib,
+            required=False,
+        )
+        portfolio_snapshot = self._project_portfolio_snapshot(
+            portfolio_snapshot,
+            self._configured_strategy_symbols(include_ranking_pool=True),
+        )
+        portfolio_snapshot = self._attach_strategy_plugin_metadata(portfolio_snapshot, strategy_plugin_signals)
+        option_chains = self._fetch_option_chains_for_runtime(ib, runtime_config, portfolio_snapshot)
+        if option_chains:
+            runtime_config["option_chains"] = option_chains
+        ctx = self._build_strategy_context(
+            runtime_adapter=self.runtime_adapter,
+            as_of=run_as_of,
+            market_inputs=self._build_direct_market_data_inputs(ib, historical_close_loader),
+            portfolio_snapshot=portfolio_snapshot,
+            runtime_config=runtime_config,
+            current_holdings=current_holdings,
+            ib=ib,
+        )
+        decision = self.entrypoint.evaluate(ctx)
+        managed_symbols = tuple(
+            dict.fromkeys(
+                str(position.symbol).strip().upper()
+                for position in decision.positions
+                if str(position.symbol or "").strip()
+            )
+        )
+        price_fallbacks = self._build_historical_close_map(
+            ib,
+            historical_close_loader,
+            self._build_price_fallback_symbol_list(
+                decision,
+                managed_symbols=managed_symbols,
+                current_holdings=current_holdings,
+            ),
+        )
+        metadata = {
+            "strategy_profile": self.profile,
+            "managed_symbols": managed_symbols,
+            "status_icon": self.status_icon,
+            "dry_run_only": self.runtime_settings.dry_run_only,
+            **build_execution_timing_metadata(
+                signal_date=run_as_of,
+                signal_effective_after_trading_days=(
+                    self.runtime_adapter.runtime_policy.signal_effective_after_trading_days
+                ),
+            ),
+        }
+        if portfolio_snapshot is not None:
+            metadata = self._enrich_portfolio_metadata(metadata, portfolio_snapshot)
+        if "BOXX" in managed_symbols:
+            metadata["safe_haven_symbol"] = "BOXX"
+        if price_fallbacks:
+            metadata["price_fallbacks"] = price_fallbacks
+            metadata["dry_run_price_fallbacks"] = price_fallbacks
+            metadata["price_fallback_source"] = "historical_close"
+        return StrategyEvaluationResult(decision=decision, metadata=metadata)
+
     def _evaluate_market_data_strategy(
         self,
         *,
@@ -942,7 +1083,10 @@ def load_strategy_runtime(
         logger=logger,
     )
     runtime_config: dict[str, Any] = {}
-    if _FEATURE_SNAPSHOT_INPUT in frozenset(entrypoint.manifest.required_inputs):
+    if (
+        _FEATURE_SNAPSHOT_INPUT in frozenset(entrypoint.manifest.required_inputs)
+        or runtime_settings.strategy_config_path
+    ):
         runtime_config = runtime.load_runtime_parameters()
     runtime_config.update(_build_runtime_overrides(runtime_settings))
 

tests/test_runtime_config_support.py

@@ -834,6 +834,10 @@ def test_print_strategy_profile_status_json_matches_registry():
     assert by_profile["global_etf_rotation"]["requires_strategy_config_path"] is False
     assert "nasdaq_sp500_smart_dca" in by_profile
     assert "ibit_smart_dca" in by_profile
+    assert by_profile["us_equity_combo_leveraged"]["profile_group"] == "direct_runtime_inputs"
+    assert by_profile["us_equity_combo_leveraged"]["input_mode"] == "market_data"
+    assert by_profile["us_equity_combo_leveraged"]["requires_strategy_config_path"] is False
+    assert by_profile["us_equity_combo_leveraged"]["config_source_policy"] == "env_only"
     assert "tech_communication_pullback_enhancement" not in by_profile
     assert by_profile["russell_top50_leader_rotation"]["profile_group"] == "snapshot_backed"
     assert by_profile["russell_top50_leader_rotation"]["display_name_zh"] == "罗素Top50领涨"
@@ -927,6 +931,28 @@ def test_print_strategy_switch_env_plan_for_tqqq_growth_income():
     assert "IBKR_FEATURE_SNAPSHOT_PATH" in plan["remove_if_present"]
 
 
+def test_print_strategy_switch_env_plan_keeps_optional_config_for_leveraged_combo_shadow():
+    result = subprocess.run(
+        [sys.executable, str(SWITCH_PLAN_SCRIPT_PATH), "--profile", "us_equity_combo_leveraged", "--json"],
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+
+    plan = json.loads(result.stdout)
+    assert plan["canonical_profile"] == "us_equity_combo_leveraged"
+    assert plan["profile_group"] == "direct_runtime_inputs"
+    assert plan["input_mode"] == "market_data"
+    assert plan["requires_strategy_config_path"] is False
+    assert plan["config_source_policy"] == "env_only"
+    assert "IBKR_STRATEGY_CONFIG_PATH" in plan["optional_env"]
+    assert "IBKR_STRATEGY_CONFIG_PATH" not in plan["remove_if_present"]
+    assert "IBKR_FEATURE_SNAPSHOT_PATH" in plan["remove_if_present"]
+    assert plan["hints"]["shadow_352045_strategy_config_path"].startswith(
+        "package://us_equity_strategies/"
+    )
+
+
 def test_print_strategy_switch_env_plan_for_hk_global_etf_dry_run():
     result = subprocess.run(
         [

tests/test_strategy_runtime.py

@@ -321,6 +321,117 @@ def evaluate(self, ctx):
     assert runtime.status_icon == "🧲"
 
 
+def test_load_strategy_runtime_loads_explicit_config_for_market_data_profile(monkeypatch):
+    class FakeEntrypoint:
+        manifest = StrategyManifest(
+            profile="us_equity_combo_leveraged",
+            domain="quant_combo",
+            display_name="US Equity Combo Leveraged",
+            description="test",
+            required_inputs=frozenset({"market_data"}),
+            default_config={"dynamic": True},
+        )
+
+        def evaluate(self, ctx):
+            return StrategyDecision()
+
+    monkeypatch.setattr(
+        strategy_runtime_module,
+        "load_strategy_definition",
+        lambda raw_profile: SimpleNamespace(profile="us_equity_combo_leveraged"),
+    )
+    monkeypatch.setattr(
+        strategy_runtime_module,
+        "load_strategy_entrypoint_for_profile",
+        lambda raw_profile: FakeEntrypoint(),
+    )
+    monkeypatch.setattr(
+        strategy_runtime_module,
+        "load_strategy_runtime_adapter_for_profile",
+        lambda raw_profile: StrategyRuntimeAdapter(
+            runtime_parameter_loader=lambda **_kwargs: {
+                "tqqq_weight": 0.35,
+                "soxl_weight": 0.20,
+                "boxx_weight": 0.45,
+                "hard_defense_risk_exposure": 0.0,
+            },
+        ),
+    )
+
+    runtime = strategy_runtime_module.load_strategy_runtime(
+        "us_equity_combo_leveraged",
+        runtime_settings=_build_runtime_settings(profile="us_equity_combo_leveraged"),
+        logger=lambda _message: None,
+    )
+
+    assert runtime.runtime_config["tqqq_weight"] == 0.35
+    assert runtime.runtime_config["hard_defense_risk_exposure"] == 0.0
+    assert runtime.merged_runtime_config["dynamic"] is True
+    assert runtime.merged_runtime_config["boxx_weight"] == 0.45
+
+
+def test_direct_market_data_strategy_builds_spy_ma200_context():
+    captured = {}
+
+    class FakeEntrypoint:
+        manifest = StrategyManifest(
+            profile="us_equity_combo_leveraged",
+            domain="quant_combo",
+            display_name="US Equity Combo Leveraged",
+            description="test",
+            required_inputs=frozenset({"market_data"}),
+            default_config={},
+        )
+
+        def evaluate(self, ctx):
+            captured["market_data"] = dict(ctx.market_data)
+            captured["runtime_config"] = dict(ctx.runtime_config)
+            return StrategyDecision(
+                positions=(
+                    PositionTarget(symbol="TQQQ", target_weight=0.35),
+                    PositionTarget(symbol="SOXL", target_weight=0.20),
+                    PositionTarget(symbol="BOXX", target_weight=0.45),
+                )
+            )
+
+    runtime = strategy_runtime_module.LoadedStrategyRuntime(
+        entrypoint=FakeEntrypoint(),
+        runtime_settings=_build_runtime_settings(profile="us_equity_combo_leveraged"),
+        runtime_adapter=StrategyRuntimeAdapter(
+            status_icon="🇺🇸",
+            available_inputs=frozenset({"market_data"}),
+            runtime_policy=StrategyRuntimePolicy(signal_effective_after_trading_days=0),
+        ),
+        runtime_config={"market_data_ma_window": 200},
+        merged_runtime_config={"market_data_ma_window": 200},
+        status_icon="🇺🇸",
+        logger=lambda _message: None,
+    )
+
+    def fake_close_loader(_ib, symbol, **_kwargs):
+        if symbol == "SPY":
+            return strategy_runtime_module.pd.Series([100.0] * 200 + [120.0])
+        return strategy_runtime_module.pd.Series([10.0] * 201)
+
+    result = runtime.evaluate(
+        ib=None,
+        current_holdings=set(),
+        historical_close_loader=fake_close_loader,
+        historical_candle_loader=None,
+        run_as_of=strategy_runtime_module.pd.Timestamp("2026-07-02"),
+        translator=lambda value, **_kwargs: value,
+        pacing_sec=0.0,
+    )
+
+    market_data = captured["market_data"]["market_data"]
+    assert market_data["spy_above_ma200"] is True
+    assert market_data["trend_symbol"] == "SPY"
+    assert market_data["trend_ma_window"] == 200
+    assert captured["runtime_config"]["market_data_ma_window"] == 200
+    assert result.metadata["managed_symbols"] == ("TQQQ", "SOXL", "BOXX")
+    assert result.metadata["safe_haven_symbol"] == "BOXX"
+
+
 def test_dca_overrides_apply_to_runtime_config():
     settings = _build_runtime_settings(
         profile="nasdaq_sp500_smart_dca",