Fail closed on incomplete HK price sources

src/hk_equity_snapshot_pipelines/snapshot_proxy_backtest.py

@@ -35,6 +35,12 @@
 
 PROXY_BACKTEST_VERSION = "hk_snapshot_proxy_cycle_backtest.v1"
 PROXY_RESEARCH_STATUS = "research_proxy_not_live_enablement_evidence"
+PARKED_RUN_STATUS = "PARKED"
+COMPLETED_RUN_STATUS = "COMPLETED"
+SOURCE_DOWNLOAD_FAILED_REASON = "SOURCE_DOWNLOAD_FAILED"
+SYNTHETIC_REQUIRES_RESEARCH_ONLY_REASON = "SYNTHETIC_REQUIRES_RESEARCH_ONLY"
+SYNTHETIC_RESEARCH_ONLY_REASON = "SYNTHETIC_RESEARCH_ONLY"
+PROXY_RESEARCH_ONLY_REASON = "PROXY_RESEARCH_ONLY"
 DEFAULT_BENCHMARK_SYMBOL = "2800.HK"
 DEFAULT_COST_BPS = 20.0
 DEFAULT_TOP_N = 5
@@ -67,6 +73,10 @@
 )
 
 
+class IncompletePriceUniverseError(ValueError):
+    pass
+
+
 @dataclass(frozen=True)
 class ProxyProfile:
     profile: str
@@ -254,11 +264,18 @@ def download_yahoo_price_history(
             frames.append(frame)
         except Exception as exc:  # pragma: no cover - live network errors vary
             failures[symbol] = str(exc)
+    if failures:
+        raise IncompletePriceUniverseError("Yahoo price history request is incomplete")
     if not frames:
         raise ValueError("No Yahoo price history could be downloaded or loaded from cache")
     prices = pd.concat(frames, ignore_index=True)
     meta = {
         "price_source": "yahoo_chart_public_api",
+        "source_kind": "real",
+        "fallback_used": False,
+        "research_only": False,
+        "promotion_eligible": False,
+        "reason_code": PROXY_RESEARCH_ONLY_REASON,
         "start": start,
         "end": end,
         "symbols_requested": list(selected_symbols),
@@ -299,6 +316,11 @@ def generate_synthetic_price_history(
             )
     return pd.DataFrame(rows), {
         "price_source": "deterministic_synthetic_price_history",
+        "source_kind": "synthetic",
+        "fallback_used": False,
+        "research_only": True,
+        "promotion_eligible": False,
+        "reason_code": SYNTHETIC_RESEARCH_ONLY_REASON,
         "start": start,
         "end": end,
         "symbols_requested": list(all_symbols),
@@ -581,6 +603,7 @@ def build_proxy_cycle_backtest(
             as_of = pd.Timestamp(close.index[position])
             feature_cache[as_of] = _feature_frame(close, as_of, benchmark_symbol)
     profile_rows: list[dict[str, Any]] = []
+    source_kind = str(price_meta.get("source_kind", "real"))
     for proxy_profile in profiles:
         backtest = run_profile_proxy_backtest(
             close,
@@ -620,15 +643,23 @@ def build_proxy_cycle_backtest(
     for index, row in enumerate(ranking, start=1):
         row["proxy_rank"] = index
         row["research_recommendation"] = (
-            "keep_first_wave_candidate"
+            "synthetic_research_only_not_promotion_eligible"
+            if source_kind == "synthetic"
+            else "keep_first_wave_candidate"
             if row["promotion_scope"] == FIRST_SNAPSHOT_PROMOTION_SCOPE and row["passes_all_cycle_drawdown_gate"]
             else "research_only_or_reject_pending_real_factor_history"
         )
     return {
         "backtest_version": PROXY_BACKTEST_VERSION,
+        "run_status": COMPLETED_RUN_STATUS,
+        "source_kind": source_kind,
+        "fallback_used": bool(price_meta.get("fallback_used", False)),
+        "research_only": True,
+        "promotion_eligible": False,
+        "reason_code": str(price_meta.get("reason_code", PROXY_RESEARCH_ONLY_REASON)),
         "research_status": PROXY_RESEARCH_STATUS,
         "data_boundary": (
-            "Price history can be real Yahoo chart data or deterministic synthetic fallback. Fundamental, buyback, "
+            "Price history can be real Yahoo chart data or deterministic synthetic research data. Fundamental, buyback, "
             "FCF, Southbound-flow, policy, valuation, and event fields are deterministic simulations where real "
             "point-in-time histories are unavailable. Results are for research triage only and are not live-enable evidence."
         ),
@@ -655,6 +686,18 @@ def build_proxy_cycle_backtest(
     }
 
 
+def _parked_payload(*, source_kind: str, reason_code: str) -> dict[str, Any]:
+    return {
+        "backtest_version": PROXY_BACKTEST_VERSION,
+        "run_status": PARKED_RUN_STATUS,
+        "source_kind": source_kind,
+        "fallback_used": False,
+        "research_only": True,
+        "promotion_eligible": False,
+        "reason_code": reason_code,
+    }
+
+
 def run_proxy_cycle_backtest(
     *,
     start: str = DEFAULT_START_DATE,
@@ -663,7 +706,8 @@ def run_proxy_cycle_backtest(
     benchmark_symbol: str = DEFAULT_BENCHMARK_SYMBOL,
     price_source: str = "yahoo",
     cache_dir: Path = DEFAULT_CACHE_DIR,
-    allow_synthetic_fallback: bool = True,
+    allow_synthetic_fallback: bool = False,
+    research_only: bool = False,
     refresh: bool = False,
     rebalance_frequency: str = "monthly",
     top_n: int = DEFAULT_TOP_N,
@@ -673,6 +717,11 @@ def run_proxy_cycle_backtest(
         raise ValueError("price_source must be yahoo or synthetic")
     price_meta: dict[str, Any]
     if price_source == "synthetic":
+        if not research_only:
+            return _parked_payload(
+                source_kind="synthetic",
+                reason_code=SYNTHETIC_REQUIRES_RESEARCH_ONLY_REASON,
+            )
         prices, price_meta = generate_synthetic_price_history(symbols=symbols, benchmark_symbol=benchmark_symbol, start=start, end=end)
     else:
         try:
@@ -684,16 +733,16 @@ def run_proxy_cycle_backtest(
                 cache_dir=cache_dir,
                 refresh=refresh,
             )
-        except Exception as exc:
-            if not allow_synthetic_fallback:
-                raise
-            prices, price_meta = generate_synthetic_price_history(
-                symbols=symbols,
-                benchmark_symbol=benchmark_symbol,
-                start=start,
-                end=end,
+        except IncompletePriceUniverseError:
+            return _parked_payload(
+                source_kind="unavailable",
+                reason_code=SOURCE_DOWNLOAD_FAILED_REASON,
+            )
+        except Exception:
+            return _parked_payload(
+                source_kind="unavailable",
+                reason_code=SOURCE_DOWNLOAD_FAILED_REASON,
             )
-            price_meta["fallback_reason"] = str(exc)
     return build_proxy_cycle_backtest(
         prices=prices,
         price_meta=price_meta,
@@ -754,7 +803,9 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--cache-dir", type=Path, default=DEFAULT_CACHE_DIR)
     parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR)
     parser.add_argument("--refresh", action="store_true")
-    parser.add_argument("--no-synthetic-fallback", action="store_true")
+    parser.add_argument("--research-only", action="store_true", help="Required to use deterministic synthetic prices.")
+    parser.add_argument("--allow-synthetic-fallback", action="store_true", help=argparse.SUPPRESS)
+    parser.add_argument("--no-synthetic-fallback", action="store_true", help=argparse.SUPPRESS)
     parser.add_argument("--rebalance-frequency", choices=("monthly", "weekly"), default="monthly")
     parser.add_argument("--top-n", type=int, default=DEFAULT_TOP_N)
     parser.add_argument("--cost-bps", type=float, default=DEFAULT_COST_BPS)
@@ -767,16 +818,17 @@ def main(argv: list[str] | None = None) -> int:
         benchmark_symbol=args.benchmark_symbol,
         price_source=args.price_source,
         cache_dir=args.cache_dir,
-        allow_synthetic_fallback=not args.no_synthetic_fallback,
+        allow_synthetic_fallback=args.allow_synthetic_fallback and not args.no_synthetic_fallback,
+        research_only=args.research_only,
         refresh=args.refresh,
         rebalance_frequency=args.rebalance_frequency,
         top_n=args.top_n,
         cost_bps=args.cost_bps,
     )
-    if not args.json:
+    if not args.json and payload["run_status"] != PARKED_RUN_STATUS:
         payload = {**payload, "output_paths": _write_outputs(args.output_dir, payload)}
     print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True))
-    return 0
+    return 0 if payload["run_status"] != PARKED_RUN_STATUS else 2
 
 
 __all__ = [

tests/test_snapshot_proxy_backtest.py

@@ -5,6 +5,10 @@
 import sys
 from pathlib import Path
 
+import pandas as pd
+import pytest
+
+import hk_equity_snapshot_pipelines.snapshot_proxy_backtest as proxy_backtest
 from hk_equity_snapshot_pipelines.snapshot_proxy_backtest import (
     DEFAULT_SYMBOLS,
     PROXY_BACKTEST_VERSION,
@@ -24,10 +28,22 @@ def test_proxy_cycle_backtest_summarizes_long_medium_short_periods():
         start="2020-01-01",
         end="2026-06-03",
     )
+    meta.update(
+        {
+            "price_source": "fake_real_price_history",
+            "source_kind": "real",
+            "research_only": False,
+            "reason_code": "PROXY_RESEARCH_ONLY",
+        }
+    )
 
     payload = build_proxy_cycle_backtest(prices=prices, price_meta=meta, top_n=3, cost_bps=20.0)
 
     assert payload["backtest_version"] == PROXY_BACKTEST_VERSION
+    assert payload["source_kind"] == "real"
+    assert payload["research_only"] is True
+    assert payload["promotion_eligible"] is False
+    assert payload["reason_code"] == "PROXY_RESEARCH_ONLY"
     assert payload["research_status"] == PROXY_RESEARCH_STATUS
     assert payload["config"]["max_drawdown_gate"] == 0.30
     assert set(payload["periods"]) == {"long", "medium", "short"}
@@ -43,13 +59,20 @@ def test_proxy_cycle_backtest_summarizes_long_medium_short_periods():
 def test_run_proxy_cycle_backtest_supports_synthetic_source():
     payload = run_proxy_cycle_backtest(
         price_source="synthetic",
+        research_only=True,
         symbols=DEFAULT_SYMBOLS[:4],
         start="2024-01-01",
         end="2026-06-03",
         top_n=2,
     )
 
     assert payload["price_meta"]["price_source"] == "deterministic_synthetic_price_history"
+    assert payload["source_kind"] == "synthetic"
+    assert payload["fallback_used"] is False
+    assert payload["research_only"] is True
+    assert payload["promotion_eligible"] is False
+    assert payload["reason_code"] == "SYNTHETIC_RESEARCH_ONLY"
+    assert all(row["research_recommendation"] == "synthetic_research_only_not_promotion_eligible" for row in payload["profiles"])
     assert payload["data"]["trading_days"] > 500
     assert payload["ranking"][0]["proxy_rank"] == 1
     assert any(row["profile"] == "hk_low_vol_dividend_quality_snapshot" for row in payload["profiles"])
@@ -62,6 +85,7 @@ def test_proxy_cycle_backtest_script_json_synthetic():
             str(SCRIPT),
             "--price-source",
             "synthetic",
+            "--research-only",
             "--start",
             "2024-01-01",
             "--end",
@@ -88,3 +112,123 @@ def test_proxy_cycle_backtest_script_json_synthetic():
     assert payload["research_status"] == PROXY_RESEARCH_STATUS
     assert payload["periods"]["short"]["end"] == "2026-06-03"
     assert payload["price_meta"]["price_source"] == "deterministic_synthetic_price_history"
+    assert payload["source_kind"] == "synthetic"
+    assert payload["fallback_used"] is False
+    assert payload["research_only"] is True
+    assert payload["promotion_eligible"] is False
+
+
+def test_run_proxy_cycle_backtest_parks_when_yahoo_download_fails(monkeypatch: pytest.MonkeyPatch, tmp_path: Path):
+    def fail_downloader(**_: object):
+        raise RuntimeError("fake downloader failure")
+
+    monkeypatch.setattr(proxy_backtest, "download_yahoo_price_history", fail_downloader)
+
+    payload = run_proxy_cycle_backtest(
+        price_source="yahoo",
+        symbols=DEFAULT_SYMBOLS[:4],
+        start="2024-01-01",
+        end="2026-06-03",
+        cache_dir=tmp_path,
+    )
+
+    assert payload == {
+        "backtest_version": PROXY_BACKTEST_VERSION,
+        "run_status": "PARKED",
+        "source_kind": "unavailable",
+        "fallback_used": False,
+        "research_only": True,
+        "promotion_eligible": False,
+        "reason_code": "SOURCE_DOWNLOAD_FAILED",
+    }
+
+
+def test_run_proxy_cycle_backtest_never_falls_back_after_provider_failure(monkeypatch: pytest.MonkeyPatch, tmp_path: Path):
+    def fail_downloader(**_: object):
+        raise RuntimeError("fake provider failure")
+
+    monkeypatch.setattr(proxy_backtest, "download_yahoo_price_history", fail_downloader)
+
+    payload = run_proxy_cycle_backtest(
+        price_source="yahoo",
+        symbols=DEFAULT_SYMBOLS[:4],
+        start="2024-01-01",
+        end="2026-06-03",
+        cache_dir=tmp_path,
+        allow_synthetic_fallback=True,
+        research_only=True,
+    )
+
+    assert payload["run_status"] == "PARKED"
+    assert payload["source_kind"] == "unavailable"
+    assert payload["reason_code"] == "SOURCE_DOWNLOAD_FAILED"
+
+
+def test_run_proxy_cycle_backtest_requires_research_only_for_synthetic():
+    payload = run_proxy_cycle_backtest(
+        price_source="synthetic",
+        symbols=DEFAULT_SYMBOLS[:4],
+        start="2024-01-01",
+        end="2026-06-03",
+        top_n=2,
+    )
+
+    assert payload["run_status"] == "PARKED"
+    assert payload["source_kind"] == "synthetic"
+    assert payload["fallback_used"] is False
+    assert payload["research_only"] is True
+    assert payload["promotion_eligible"] is False
+    assert payload["reason_code"] == "SYNTHETIC_REQUIRES_RESEARCH_ONLY"
+
+
+@pytest.mark.parametrize("missing_symbol", [DEFAULT_SYMBOLS[1], "2800.HK"])
+def test_proxy_cycle_backtest_cli_parks_without_evidence_when_any_requested_symbol_fails(
+    missing_symbol: str,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+):
+    def fake_download(symbol: str, **_: object) -> pd.DataFrame:
+        if symbol == missing_symbol:
+            raise RuntimeError("fake symbol download failure")
+        return pd.DataFrame(
+            {
+                "date": [pd.Timestamp("2026-06-03")],
+                "symbol": [symbol],
+                "close": [100.0],
+                "volume": [1_000_000],
+            }
+        )
+
+    def reject_evidence_build(**_: object):
+        pytest.fail("partial source data must not reach the evidence builder")
+
+    monkeypatch.setattr(proxy_backtest, "_download_yahoo_symbol", fake_download)
+    monkeypatch.setattr(proxy_backtest, "build_proxy_cycle_backtest", reject_evidence_build)
+    output_dir = tmp_path / "evidence"
+
+    exit_code = proxy_backtest.main(
+        [
+            "--start",
+            "2026-06-01",
+            "--end",
+            "2026-06-03",
+            "--symbol",
+            DEFAULT_SYMBOLS[0],
+            "--symbol",
+            DEFAULT_SYMBOLS[1],
+            "--cache-dir",
+            str(tmp_path / "cache"),
+            "--output-dir",
+            str(output_dir),
+            "--research-only",
+        ]
+    )
+
+    payload = json.loads(capsys.readouterr().out)
+    assert exit_code == 2
+    assert payload["run_status"] == "PARKED"
+    assert payload["reason_code"] == "SOURCE_DOWNLOAD_FAILED"
+    assert payload["research_only"] is True
+    assert payload["promotion_eligible"] is False
+    assert not output_dir.exists()