[codex] Migrate global ETF rotation to feature snapshot

README.md

@@ -25,7 +25,6 @@ These profiles can run from market history, portfolio snapshots, or other runtim
 
 | Profile | Name | Notes |
 | --- | --- | --- |
-| `global_etf_rotation` | Global ETF Rotation | runtime-enabled ETF rotation using market history. |
 | `tqqq_growth_income` | TQQQ Growth Income | runtime-enabled QQQ/TQQQ dual-drive profile with defensive and income sleeves. |
 | `soxl_soxx_trend_income` | SOXL/SOXX Semiconductor Trend Income | runtime-enabled semiconductor ETF trend profile. |
 | `nasdaq_sp500_smart_dca` | Nasdaq 100 / S&P 500 Smart DCA | runtime-enabled buy-only DCA profile for broad US equity ETFs; defaults to monthly fixed-amount DCA with configurable weekly/monthly/quarterly cadence and optional smart sizing. |
@@ -37,6 +36,7 @@ These profiles depend on artifacts produced by `UsEquitySnapshotPipelines` befor
 
 | Profile | Name | Notes |
 | --- | --- | --- |
+| `global_etf_rotation` | Global ETF Rotation | runtime-enabled feature-snapshot ETF rotation with daily canary defense and BIL safe haven. |
 | `russell_top50_leader_rotation` | Russell Top50 Leader Rotation | runtime-enabled feature-snapshot Russell Top50 leader rotation. |
 
 ### Research-only candidates

README.zh-CN.md

@@ -25,7 +25,6 @@ UsEquityStrategies 是 QuantStrategyLab 的美股策略包。为 QuantStrategyLa
 
 | Profile | 名称 | 说明 |
 | --- | --- | --- |
-| `global_etf_rotation` | Global ETF Rotation | 使用 market history 的 runtime-enabled ETF 轮动。 |
 | `tqqq_growth_income` | TQQQ Growth Income | QQQ/TQQQ dual-drive，带防守和 income sleeve。 |
 | `soxl_soxx_trend_income` | SOXL/SOXX Semiconductor Trend Income | 半导体 ETF 趋势策略。 |
 | `nasdaq_sp500_smart_dca` | Nasdaq 100 / S&P 500 Smart DCA | 面向宽基美股 ETF 的买入型 DCA profile；默认月度定额定投，可配置周/月/季频率和智能倍数。 |
@@ -37,6 +36,7 @@ UsEquityStrategies 是 QuantStrategyLab 的美股策略包。为 QuantStrategyLa
 
 | Profile | 名称 | 说明 |
 | --- | --- | --- |
+| `global_etf_rotation` | Global ETF Rotation | 基于 feature snapshot 的 ETF 轮动，带每日 canary 防守和 BIL safe haven。 |
 | `russell_top50_leader_rotation` | Russell Top50 Leader Rotation | 基于 feature snapshot 的 Russell Top50 leader rotation。 |
 
 ### 研究侧候选

docs/us_equity_contract_gap_matrix.md

@@ -47,7 +47,6 @@ comparison with runtime-enabled peers:
 - `russell_1000_multi_factor_defensive`
 - `mega_cap_leader_rotation_top50_balanced`
 - `mega_cap_leader_rotation_dynamic_top20`
-- `mega_cap_leader_rotation_aggressive`
 - `dynamic_mega_leveraged_pullback`
 - `tech_communication_pullback_enhancement`
 

docs/us_equity_runtime_archive.zh-CN.md

@@ -30,7 +30,6 @@ _更新日期：2026-06-03_
 | 已删除 profile | 状态 | 关键结果 |
 | --- | --- | --- |
 | `mega_cap_leader_rotation_dynamic_top20` | 已移除 | CAGR 21.51%，最大回撤 -23.14%。 |
-| `mega_cap_leader_rotation_aggressive` | 已移除 | CAGR 32.42%，最大回撤 -28.64%。 |
 | `dynamic_mega_leveraged_pullback` | 已移除 | CAGR 30.96%，最大回撤 -34.80%。 |
 
 ## 归档口径

docs/us_equity_strategy_status.zh-CN.md

@@ -30,7 +30,6 @@ _更新日期：2026-06-19_
 | `russell_1000_multi_factor_defensive` | 年化只小幅跑赢大盘，最大回撤与大盘接近，实盘价值弱于定投大盘，移除可运行入口。 |
 | `mega_cap_leader_rotation_top50_balanced` | 名称不再贴切；策略仍保留但改名为 `russell_top50_leader_rotation`，旧 profile 不再兼容。 |
 | `mega_cap_leader_rotation_dynamic_top20` | 同期 CAGR 21.51%、最大回撤 -23.14%；收益明显弱于 `russell_top50_leader_rotation` 的 36.41%。 |
-| `mega_cap_leader_rotation_aggressive` | Top50 top3/cap35 CAGR 32.42%、最大回撤 -28.64%；仍弱于 Russell Top50，且更集中。 |
 | `dynamic_mega_leveraged_pullback` | CAGR 30.96%、最大回撤 -34.80%；2x 产品和事件反弹路线更复杂，未优于当前保留路线。 |
 | `tech_communication_pullback_enhancement` | 行业限制在科技/通信，收益明显低于 `russell_top50_leader_rotation`，最大回撤也没有改善；策略实现和 bundled config 仅作为离线研究归档保留。 |
 

src/us_equity_strategies/catalog.py

@@ -47,7 +47,7 @@
 }
 
 STRATEGY_REQUIRED_INPUTS: dict[str, frozenset[str]] = {
-    GLOBAL_ETF_ROTATION_PROFILE: frozenset({"market_history"}),
+    GLOBAL_ETF_ROTATION_PROFILE: frozenset({"feature_snapshot"}),
     TQQQ_GROWTH_INCOME_PROFILE: frozenset({"benchmark_history", "portfolio_snapshot"}),
     SOXL_SOXX_TREND_INCOME_PROFILE: frozenset({"derived_indicators", "portfolio_snapshot"}),
     RUSSELL_TOP50_LEADER_ROTATION_PROFILE: frozenset({"feature_snapshot"}),
@@ -411,10 +411,10 @@ def _build_strategy_definition(
     GLOBAL_ETF_ROTATION_PROFILE: StrategyMetadata(
         canonical_profile=GLOBAL_ETF_ROTATION_PROFILE,
         display_name="Global ETF Rotation",
-        description="Quarterly top-2 global ETF rotation with daily canary defense, SMA250 confidence gating, and BIL safe haven.",
+        description="Quarterly top-2 global ETF rotation backed by snapshot-side ETF universe evidence, with daily canary defense and BIL safe haven.",
         aliases=("global_macro_etf_rotation", GLOBAL_ETF_CONFIDENCE_VOL_GATE_PROFILE),
         cadence="quarterly + daily canary",
-        asset_scope="global_etf_rotation",
+        asset_scope="global_etf_rotation_snapshot",
         benchmark="VOO",
         role="defensive_rotation",
         status="runtime_enabled",

src/us_equity_strategies/entrypoints/__init__.py

@@ -243,21 +243,15 @@ def _evaluate_global_etf_rotation_with_manifest(ctx: StrategyContext, *, manifes
     config["canary_assets"] = list(config.get("canary_assets", ()))
     config.pop("signal_effective_after_trading_days", None)
     pop_execution_only_config(config)
-    market_history = ctx.market_data.get("market_history")
-    if market_history is None:
-        def _empty_market_history(_client, _symbol):
-            return ()
-
-        market_history = _empty_market_history
+    feature_snapshot = require_market_data(ctx, "feature_snapshot")
     translator = config.pop("translator", default_translator)
     config.pop("signal_text_fn", None)
-    weights, signal_desc, is_emergency, canary_str = legacy_global_etf_rotation.compute_signals(
-        ctx.capabilities.get("broker_client"),
+    config.pop("pacing_sec", None)
+    weights, signal_desc, is_emergency, canary_str = legacy_global_etf_rotation.compute_signals_from_feature_snapshot(
+        feature_snapshot,
         get_current_holdings(ctx),
-        get_historical_close=market_history,
         as_of_date=ctx.as_of,
         translator=translator,
-        pacing_sec=float(config.pop("pacing_sec", 0.0)),
         **config,
     )
     weights, income_layer_diagnostics = apply_income_layer_to_weights(
@@ -278,6 +272,8 @@ def _empty_market_history(_client, _symbol):
         market_regime_control_diagnostics.get("market_regime_control_notification_context"),
     )
     diagnostics = {
+        "signal_source": legacy_global_etf_rotation.SIGNAL_SOURCE,
+        "snapshot_contract_version": legacy_global_etf_rotation.SNAPSHOT_CONTRACT_VERSION,
         "signal_description": signal_desc,
         "canary_status": canary_str,
         "actionable": weights is not None,

src/us_equity_strategies/manifests/__init__.py

@@ -38,7 +38,7 @@ def _manifest(
         "Quarterly top-2 global ETF rotation with daily canary defense, SMA250 confidence gating, and BIL safe haven."
     ),
     aliases=("global_macro_etf_rotation", GLOBAL_ETF_CONFIDENCE_VOL_GATE_PROFILE),
-    required_inputs=frozenset({"market_history"}),
+    required_inputs=frozenset({"feature_snapshot"}),
     default_config={
         "ranking_pool": (
             "EWY",

src/us_equity_strategies/runtime_adapters.py

@@ -19,6 +19,7 @@
     resolve_canonical_profile,
 )
 from us_equity_strategies.strategies import (
+    global_etf_rotation as global_etf_rotation_strategy,
     ibit_smart_dca as ibit_smart_dca_strategy,
     mega_cap_leader_rotation as mega_cap_leader_rotation_strategy,
     nasdaq_sp500_smart_dca as nasdaq_sp500_smart_dca_strategy,
@@ -52,6 +53,17 @@
 BASE_RUNTIME_ADAPTERS: dict[str, StrategyRuntimeAdapter] = {
     "global_etf_rotation": StrategyRuntimeAdapter(
         status_icon="🐤",
+        required_feature_columns=global_etf_rotation_strategy.REQUIRED_FEATURE_COLUMNS,
+        snapshot_date_columns=global_etf_rotation_strategy.SNAPSHOT_DATE_COLUMNS,
+        max_snapshot_month_lag=global_etf_rotation_strategy.MAX_SNAPSHOT_MONTH_LAG,
+        require_snapshot_manifest=global_etf_rotation_strategy.REQUIRE_SNAPSHOT_MANIFEST,
+        snapshot_contract_version=global_etf_rotation_strategy.SNAPSHOT_CONTRACT_VERSION,
+        managed_symbols_extractor=global_etf_rotation_strategy.extract_managed_symbols,
+        artifact_contract=StrategyArtifactContract(
+            requires_snapshot_artifacts=True,
+            requires_snapshot_manifest_path=global_etf_rotation_strategy.REQUIRE_SNAPSHOT_MANIFEST,
+            snapshot_contract_version=global_etf_rotation_strategy.SNAPSHOT_CONTRACT_VERSION,
+        ),
         runtime_policy=StrategyRuntimePolicy(signal_effective_after_trading_days=1),
     ),
     "tqqq_growth_income": StrategyRuntimeAdapter(

src/us_equity_strategies/strategies/global_etf_rotation.py

@@ -3,7 +3,7 @@
 import time
 from datetime import datetime
 from importlib import import_module
-from typing import Callable
+from typing import Any, Callable
 
 import numpy as np
 import pandas as pd
@@ -278,3 +278,217 @@ def compute_signals(
     top_str = ", ".join(f"{ticker}({score:.3f})" for ticker, score in top)
     signal_desc = translator("quarterly", n=top_n) + f"\n  Top: {top_str}{confidence_note}"
     return weights, signal_desc, False, canary_str
+
+SIGNAL_SOURCE = "feature_snapshot"
+SNAPSHOT_CONTRACT_VERSION = "global_etf_rotation.feature_snapshot.v1"
+REQUIRED_FEATURE_COLUMNS = frozenset(
+    {
+        "as_of",
+        "symbol",
+        "role",
+        "close",
+        "momentum_13612w",
+        "sma_pass",
+        "eligible",
+        "vol_126",
+    }
+)
+SNAPSHOT_DATE_COLUMNS = ("as_of", "snapshot_date")
+MAX_SNAPSHOT_MONTH_LAG = 1
+REQUIRE_SNAPSHOT_MANIFEST = True
+
+
+def _coerce_bool(value: Any) -> bool:
+    if pd.isna(value):
+        return False
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, (int, float)):
+        return bool(value)
+    normalized = str(value).strip().lower()
+    if normalized in {"1", "true", "yes", "y"}:
+        return True
+    if normalized in {"0", "false", "no", "n"}:
+        return False
+    return bool(normalized)
+
+
+def _to_feature_frame(feature_snapshot) -> pd.DataFrame:
+    frame = (
+        feature_snapshot.copy()
+        if isinstance(feature_snapshot, pd.DataFrame)
+        else pd.DataFrame(list(feature_snapshot))
+    )
+    if frame.empty:
+        raise ValueError("feature_snapshot must contain at least one row")
+    missing = REQUIRED_FEATURE_COLUMNS - set(frame.columns)
+    if missing:
+        missing_text = ", ".join(sorted(missing))
+        raise ValueError(f"feature_snapshot missing required columns: {missing_text}")
+    frame = frame.copy()
+    frame["symbol"] = frame["symbol"].astype(str).str.strip().str.upper()
+    frame["role"] = frame["role"].astype(str).str.strip().str.lower()
+    frame["momentum_13612w"] = pd.to_numeric(frame["momentum_13612w"], errors="coerce")
+    frame["score"] = pd.to_numeric(frame.get("score", frame["momentum_13612w"]), errors="coerce")
+    frame["vol_126"] = pd.to_numeric(frame["vol_126"], errors="coerce")
+    frame["close"] = pd.to_numeric(frame["close"], errors="coerce")
+    frame["sma_pass"] = frame["sma_pass"].map(_coerce_bool)
+    frame["eligible"] = frame["eligible"].map(_coerce_bool)
+    return frame
+
+
+def extract_managed_symbols(
+    feature_snapshot,
+    *,
+    benchmark_symbol: str | None = None,
+    safe_haven_symbol: str | None = None,
+) -> tuple[str, ...]:
+    frame = _to_feature_frame(feature_snapshot)
+    symbols: list[str] = []
+    for symbol in frame.loc[frame["role"].isin({"ranking_pool_etf", "safe_haven"}), "symbol"]:
+        if symbol and symbol not in symbols:
+            symbols.append(str(symbol))
+    safe = str(safe_haven_symbol or "").strip().upper()
+    if safe and safe not in symbols:
+        symbols.append(safe)
+    return tuple(symbols)
+
+
+def _snapshot_rebalance_day(as_of_date, *, rebalance_months) -> bool:
+    tz_ny = pytz.timezone("America/New_York")
+    timestamp = pd.Timestamp(as_of_date)
+    if timestamp.tzinfo is None:
+        now_ny = tz_ny.localize(timestamp.to_pydatetime())
+    else:
+        now_ny = timestamp.tz_convert(tz_ny).to_pydatetime()
+    return _is_rebalance_day(now_ny, rebalance_months=rebalance_months)
+
+
+def _snapshot_confidence_weighting(
+    rows: pd.DataFrame,
+    top: list[tuple[str, float]],
+    *,
+    top_n: int,
+    confidence_metric: str,
+    confidence_threshold: float,
+    confidence_top1_weight: float,
+    confidence_volatility_gate_enabled: bool,
+    confidence_volatility_max_ratio: float,
+) -> tuple[dict[str, float], str]:
+    per_weight = 1.0 / float(top_n)
+    weights = {ticker: per_weight for ticker, _score in top}
+    if top_n != 2 or len(top) != 2:
+        return weights, ""
+    score_pairs = [(str(row.symbol), float(row.rank_score)) for row in rows.itertuples(index=False)]
+    confidence = _score_confidence(score_pairs, metric=str(confidence_metric))
+    top1, top2 = top[0][0], top[1][0]
+    use_confidence_weight = not np.isnan(confidence) and confidence >= float(confidence_threshold)
+    top1_vol = top2_vol = float("nan")
+    if use_confidence_weight and confidence_volatility_gate_enabled:
+        vol_by_symbol = rows.set_index("symbol")["vol_126"].to_dict()
+        top1_vol = float(vol_by_symbol.get(top1, float("nan")))
+        top2_vol = float(vol_by_symbol.get(top2, float("nan")))
+        use_confidence_weight = not (
+            np.isnan(top1_vol)
+            or np.isnan(top2_vol)
+            or top2_vol <= 0.0
+            or top1_vol > top2_vol * float(confidence_volatility_max_ratio)
+        )
+    if not use_confidence_weight:
+        return weights, ""
+    top1_weight = min(1.0, max(per_weight, float(confidence_top1_weight)))
+    weights = {top1: top1_weight, top2: 1.0 - top1_weight}
+    note = f"confidence={confidence:.2f}"
+    if confidence_volatility_gate_enabled:
+        note += f" vol={top1_vol:.2%}/{top2_vol:.2%}"
+    return weights, note
+
+
+def compute_signals_from_feature_snapshot(
+    feature_snapshot,
+    current_holdings,
+    *,
+    as_of_date=None,
+    ranking_pool=RANKING_POOL,
+    canary_assets=CANARY_ASSETS,
+    safe_haven: str = SAFE_HAVEN,
+    top_n: int = TOP_N,
+    hold_bonus: float = HOLD_BONUS,
+    canary_bad_threshold: int = CANARY_BAD_THRESHOLD,
+    rebalance_months=REBALANCE_MONTHS,
+    translator: Callable,
+    sma_period: int = SMA_PERIOD,
+    confidence_weighting_enabled: bool = False,
+    confidence_metric: str = CONFIDENCE_METRIC_Z_GAP,
+    confidence_threshold: float = 1.0,
+    confidence_top1_weight: float = 0.75,
+    confidence_volatility_gate_enabled: bool = False,
+    confidence_volatility_window: int = 126,
+    confidence_volatility_max_ratio: float = 1.3,
+):
+    del sma_period, confidence_volatility_window
+    frame = _to_feature_frame(feature_snapshot)
+    safe_haven = str(safe_haven or SAFE_HAVEN).strip().upper()
+    snapshot_ranking_symbols = frame.loc[frame["role"].eq("ranking_pool_etf"), "symbol"].dropna().astype(str).tolist()
+    snapshot_canary_symbols = frame.loc[frame["role"].eq("canary_asset"), "symbol"].dropna().astype(str).tolist()
+    ranking_symbols = snapshot_ranking_symbols or [symbol.upper() for symbol in ranking_pool]
+    canary_symbols = snapshot_canary_symbols or [symbol.upper() for symbol in canary_assets]
+    current_holding_symbols = {str(symbol or "").strip().upper() for symbol in current_holdings or ()}
+
+    by_symbol = frame.drop_duplicates(subset=["symbol"], keep="last").set_index("symbol")
+    n_bad = 0
+    canary_details: list[str] = []
+    for ticker in canary_symbols:
+        if ticker not in by_symbol.index:
+            n_bad += 1
+            canary_details.append(f"{ticker}:❌(no data)")
+            continue
+        mom = float(by_symbol.at[ticker, "momentum_13612w"])
+        if np.isnan(mom) or mom < 0:
+            n_bad += 1
+            canary_details.append(f"{ticker}:❌({mom:.3f})" if not np.isnan(mom) else f"{ticker}:❌(nan)")
+        else:
+            canary_details.append(f"{ticker}:✅({mom:.3f})")
+    canary_str = ", ".join(canary_details)
+    if n_bad >= int(canary_bad_threshold):
+        signal_desc = translator("emergency", n_bad=n_bad, safe=safe_haven)
+        return {safe_haven: 1.0}, signal_desc, True, canary_str
+
+    rebalance_as_of = as_of_date if as_of_date is not None else pd.Timestamp.now(tz=pytz.timezone("America/New_York"))
+    if not _snapshot_rebalance_day(rebalance_as_of, rebalance_months=rebalance_months):
+        signal_desc = translator("daily_check")
+        return None, signal_desc, False, canary_str
+
+    rows = frame.loc[frame["symbol"].isin(ranking_symbols)].copy()
+    rows = rows.loc[rows["eligible"] & rows["sma_pass"] & rows["momentum_13612w"].notna()].copy()
+    if rows.empty:
+        signal_desc = translator("emergency", n_bad="SMA", safe=safe_haven)
+        return {safe_haven: 1.0}, signal_desc, False, canary_str
+    rows["rank_score"] = rows["score"].fillna(rows["momentum_13612w"])
+    rows.loc[rows["symbol"].isin(current_holding_symbols), "rank_score"] += float(hold_bonus)
+    ranked = rows.sort_values(["rank_score", "momentum_13612w", "symbol"], ascending=[False, False, True])
+    top = [(str(row.symbol), float(row.rank_score)) for row in ranked.head(int(top_n)).itertuples(index=False)]
+    if not top:
+        signal_desc = translator("emergency", n_bad="SMA", safe=safe_haven)
+        return {safe_haven: 1.0}, signal_desc, False, canary_str
+
+    weights = {ticker: 1.0 / float(top_n) for ticker, _score in top}
+    confidence_note = ""
+    if confidence_weighting_enabled:
+        weights, confidence_note = _snapshot_confidence_weighting(
+            ranked,
+            top,
+            top_n=int(top_n),
+            confidence_metric=str(confidence_metric),
+            confidence_threshold=float(confidence_threshold),
+            confidence_top1_weight=float(confidence_top1_weight),
+            confidence_volatility_gate_enabled=bool(confidence_volatility_gate_enabled),
+            confidence_volatility_max_ratio=float(confidence_volatility_max_ratio),
+        )
+    if len(top) < int(top_n):
+        weights[safe_haven] = weights.get(safe_haven, 0.0) + (1.0 / float(top_n)) * (int(top_n) - len(top))
+    selected = ", ".join(f"{ticker}({score:.3f})" for ticker, score in top)
+    signal_desc = f"Global ETF snapshot rotation selected: {selected}"
+    if confidence_note:
+        signal_desc = f"{signal_desc} | {confidence_note}"
+    return weights, signal_desc, False, canary_str

tests/test_catalog.py

@@ -39,7 +39,7 @@ def test_catalog_contains_supported_profiles(self):
         )
         self.assertEqual(
             catalog[GLOBAL_ETF_ROTATION_PROFILE].required_inputs,
-            frozenset({"market_history"}),
+            frozenset({"feature_snapshot"}),
         )
 
         self.assertNotIn("global_etf_confidence_vol_gate", catalog)