diff --git a/scripts/print_strategy_switch_env_plan.py b/scripts/print_strategy_switch_env_plan.py index 10e82f6..6ea9d31 100644 --- a/scripts/print_strategy_switch_env_plan.py +++ b/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, diff --git a/strategy_runtime.py b/strategy_runtime.py index d3ab5d3..8e41407 100644 --- a/strategy_runtime.py +++ b/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 ''}" ) + 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)) diff --git a/tests/test_runtime_config_support.py b/tests/test_runtime_config_support.py index 4de1209..15312d8 100644 --- a/tests/test_runtime_config_support.py +++ b/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( [ diff --git a/tests/test_strategy_runtime.py b/tests/test_strategy_runtime.py index 734be31..ee99013 100644 --- a/tests/test_strategy_runtime.py +++ b/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",