target-weight 승격 증거 일관성 검증 강화

target-weight pilot 증거의 날짜와 params hash 일관성을 승격 및 live gate에서 검증합니다.

Author: easygap

README.md

@@ -231,4 +231,4 @@ Target-weight pilot 승인/재시도 보강: `tools/paper_pilot_control.py --ena
 
 Target-weight 실행 차단 기록: `--execute`가 pilot cap validation에서 막히면 주문·체결·증거 수집 없이 session JSON artifact에 차단 사유를 남깁니다. runtime pilot session은 쓰지 않아 cap을 고친 뒤 같은 거래일 계획을 다시 점검할 수 있습니다.
 
-Target-weight 승격 증거 보강: target-weight 계열 전략은 일반 `execution_backed=True` paper record만으로 promotion evidence day를 채우지 않습니다. `pilot_paper`/authorized record가 target-weight plan과 execution proof를 포함하고, liquidity/pre-trade risk/order result/fill/position reconciliation complete 및 plan/execution params hash 일치를 만족한 날만 승격 카운트에 들어갑니다. live gate도 이 proof summary가 없거나 invalid day가 있으면 target-weight live 전환을 차단합니다.
+Target-weight 승격 증거 보강: target-weight 계열 전략은 일반 `execution_backed=True` paper record만으로 promotion evidence day를 채우지 않습니다. `pilot_paper`/authorized record가 target-weight plan과 execution proof를 포함하고, record date와 plan trade day가 일치하며, liquidity/pre-trade risk/order result/fill/position reconciliation complete 및 plan/execution params hash 일치를 만족한 날만 승격 카운트에 들어갑니다. 60영업일 전체 verified pilot evidence는 하나의 params hash로 고정되어야 하며, live gate도 canonical metadata의 params hash와 paper evidence params hash가 다르면 target-weight live 전환을 차단합니다.

core/live_gate.py

@@ -197,9 +197,26 @@ def _is_target_weight_strategy(strategy_name: str) -> bool:
     return strategy_name.startswith("target_weight_")
 
 
+def _canonical_target_weight_params_hash(
+    metadata: dict[str, Any],
+    strategy_name: str,
+) -> str | None:
+    specs = metadata.get("strategy_specs")
+    if not isinstance(specs, list):
+        return None
+    for spec in specs:
+        if not isinstance(spec, dict):
+            continue
+        if spec.get("candidate_id") == strategy_name:
+            params_hash = spec.get("params_hash")
+            return params_hash if isinstance(params_hash, str) and params_hash else None
+    return None
+
+
 def _validate_target_weight_evidence_summary(
     strategy_name: str,
     evidence: dict[str, Any],
+    canonical_params_hash: str | None = None,
 ) -> list[str]:
     if not _is_target_weight_strategy(strategy_name):
         return []
@@ -214,6 +231,18 @@ def _validate_target_weight_evidence_summary(
     invalid_days = _as_int(summary.get("invalid_days")) or 0
     if summary.get("all_promotable_days_verified") is not True:
         issues.append("target-weight promotable evidence가 모두 검증된 pilot_paper 실행 증거가 아님.")
+    if summary.get("params_hash_consistent") is not True:
+        issues.append("target-weight pilot evidence params_hash가 60영업일 전체에서 일관되지 않음.")
+    evidence_params_hash = summary.get("params_hash") or evidence.get("target_weight_params_hash")
+    if not isinstance(evidence_params_hash, str) or not evidence_params_hash:
+        issues.append("target-weight pilot evidence params_hash 누락.")
+    elif not canonical_params_hash:
+        issues.append("target-weight canonical params_hash 누락.")
+    elif evidence_params_hash != canonical_params_hash:
+        issues.append(
+            "target-weight canonical params_hash 불일치: "
+            f"evidence={evidence_params_hash}, canonical={canonical_params_hash}."
+        )
     if valid_days < promotable_days or valid_days < 60:
         issues.append(
             "target-weight verified pilot_paper evidence 60영업일 미달 "
@@ -434,6 +463,12 @@ def validate_live_readiness(
             issues.append("paper evidence win_rate 45% 미달.")
         if frozen_days > 0:
             issues.append("paper evidence에 frozen day가 존재함.")
-        issues.extend(_validate_target_weight_evidence_summary(strategy_name, evidence))
+        issues.extend(
+            _validate_target_weight_evidence_summary(
+                strategy_name,
+                evidence,
+                canonical_params_hash=_canonical_target_weight_params_hash(metadata, strategy_name),
+            )
+        )
 
     return issues

core/paper_evidence.py

@@ -1181,6 +1181,10 @@ def _target_weight_record_proof_status(strategy: str, record: dict) -> tuple[boo
         return False, "missing_target_weight_execution"
     if plan.get("candidate_id") != strategy:
         return False, "target_weight_candidate_mismatch"
+    plan_trade_day = plan.get("trade_day")
+    record_date = record.get("date")
+    if plan_trade_day and record_date and plan_trade_day != record_date:
+        return False, "target_weight_trade_day_mismatch"
 
     plan_hash = plan.get("params_hash")
     execution_hash = execution.get("params_hash")
@@ -1205,6 +1209,17 @@ def _target_weight_record_proof_status(strategy: str, record: dict) -> tuple[boo
     return True, "verified_target_weight_pilot_execution"
 
 
+def _target_weight_record_params_hash(record: dict) -> str | None:
+    caps = record.get("pilot_caps_snapshot") or {}
+    plan = caps.get("target_weight_plan") or {}
+    execution = caps.get("target_weight_execution") or {}
+    plan_hash = plan.get("params_hash")
+    execution_hash = execution.get("params_hash")
+    if isinstance(plan_hash, str) and plan_hash and plan_hash == execution_hash:
+        return plan_hash
+    return None
+
+
 def _split_target_weight_promotion_records(
     strategy: str,
     records: list[dict],
@@ -1252,15 +1267,31 @@ def generate_promotion_package(strategy: str) -> tuple[Path | None, Path | None]
     target_weight_valid_records: list[dict] = []
     target_weight_invalid_records: list[dict] = []
     target_weight_invalid_reasons: dict[str, int] = {}
+    target_weight_params_hashes: list[str] = []
     if target_weight_required:
         (
             target_weight_valid_records,
             target_weight_invalid_records,
             target_weight_invalid_reasons,
         ) = _split_target_weight_promotion_records(strategy, execution_records)
+        target_weight_params_hashes = sorted({
+            params_hash
+            for record in target_weight_valid_records
+            if (params_hash := _target_weight_record_params_hash(record))
+        })
         records = target_weight_valid_records
     else:
         records = execution_records
+    target_weight_params_hash = (
+        target_weight_params_hashes[0] if len(target_weight_params_hashes) == 1 else None
+    )
+    target_weight_params_hash_consistent = (
+        not target_weight_required
+        or (
+            len(target_weight_valid_records) > 0
+            and len(target_weight_params_hashes) == 1
+        )
+    )
     total_days = len(records)
     promotable_days = total_days
     shadow_days = len(shadow_records)
@@ -1331,6 +1362,11 @@ def generate_promotion_package(strategy: str) -> tuple[Path | None, Path | None]
         block_reasons.append(
             "target_weight_invalid_execution_evidence=%d" % len(target_weight_invalid_records)
         )
+    if target_weight_required and target_weight_valid_records and not target_weight_params_hash_consistent:
+        blocked = True
+        block_reasons.append(
+            "target_weight_params_hash_drift=%d" % len(target_weight_params_hashes)
+        )
     if frozen_days > 0:
         blocked = True
         block_reasons.append("frozen_days=%d" % frozen_days)
@@ -1427,15 +1463,20 @@ def generate_promotion_package(strategy: str) -> tuple[Path | None, Path | None]
             "valid_pilot_days": len(target_weight_valid_records),
             "invalid_days": len(target_weight_invalid_records),
             "invalid_reasons": target_weight_invalid_reasons,
+            "params_hash": target_weight_params_hash,
+            "params_hashes": target_weight_params_hashes,
+            "params_hash_consistent": target_weight_params_hash_consistent,
             "all_promotable_days_verified": (
                 not target_weight_required
                 or (
                     len(target_weight_invalid_records) == 0
                     and len(target_weight_valid_records) == promotable_days
                     and promotable_days > 0
+                    and target_weight_params_hash_consistent
                 )
             ),
         },
+        "target_weight_params_hash": target_weight_params_hash,
         "target_weight_verified_pilot_days": len(target_weight_valid_records),
         "target_weight_invalid_days": len(target_weight_invalid_records),
         # backward compat aliases

docs/PROJECT_GUIDE.md

@@ -718,6 +718,7 @@ full paper 신규 BUY는 preflight status artifact와 runtime state가 모두 
 | ✅ **target-weight pilot enable guard 추가** | `tools/paper_pilot_control.py --enable`이 target-weight 후보를 승인할 때 pilot auth 기록 전에 readiness audit을 재실행하고, 운영자가 요청한 cap이 현재 plan/launch readiness/유동성 preflight/비용 반영 pre-trade risk를 만족할 때만 승인한다. stale plan이나 추천 cap 미충족 상태에서 auth만 먼저 열리는 경로를 차단한다 |
 | ✅ **target-weight cap validation artifact 추가** | `--execute`가 pilot cap validation에서 막혀도 예외로 조기 종료하지 않고 session JSON artifact에 차단 사유, skipped orders, evidence block reason을 남긴다. 주문/체결/증거 수집/runtime pilot session 저장은 하지 않아 cap 조정 후 재점검할 수 있다 |
 | ✅ **target-weight promotion proof guard 추가** | promotion package와 live gate가 target-weight 계열 전략의 promotable day를 일반 `execution_backed=True` record로 인정하지 않는다. `pilot_paper`/authorized record가 target-weight plan/execution proof를 포함하고 liquidity/pre-trade/order/fill/position complete와 plan/execution params hash 일치를 만족한 날만 승격 카운트에 포함한다 |
+| ✅ **target-weight proof consistency guard 추가** | target-weight promotion evidence는 record date와 plan trade day가 일치해야 하며, 60영업일 verified pilot evidence 전체가 하나의 params hash로 고정되어야 한다. live gate는 paper evidence의 target-weight params hash와 canonical `strategy_specs.params_hash`가 다르면 실전 전환을 차단한다 |
 | ✅ **target-weight liquidity preflight 추가** | `core.target_weight_rotation.build_target_weight_plan()`이 주문 종목별 최근 20일 평균 거래대금을 plan diagnostics에 기록하고, `tools/target_weight_rotation_pilot.py`가 주문별 notional이 평균 거래대금의 기본 5%(`--max-order-adv-pct`)를 초과하거나 유동성 diagnostics가 누락되면 readiness audit과 `--execute`를 fail-closed 차단한다. 차단 결과는 session artifact, readiness JSON/MD, pilot session의 `target_weight_execution.liquidity_check`에 남기며 기존 pilot evidence 재사용도 liquidity-complete 조건을 요구한다 |
 | ✅ **target-weight pre-trade risk 추가** | `RiskManager.calculate_transaction_costs()`의 수수료/세금/동적 슬리피지 예상 체결가를 재사용해 plan 전체 주문을 제출 전 시뮬레이션한다. 예상 현금 부족, 최소 현금비중, 최대 투자비중, 종목별 최대 비중, 최대 보유 종목 수 위반은 readiness audit과 `--execute`에서 fail-closed 차단하고, session/readiness/pilot evidence snapshot에 `pre_trade_risk_check`와 cost summary를 남긴다. 매도 주문은 plan diagnostics의 기존 평균매입가를 사용해 세금/양도세 옵션과도 연결된다 |
 | ✅ **target-weight completed rerun block 추가** | same-candidate/trade-day pilot session artifact가 이미 `execution_complete=True`이고 실제 주문 실행 수량이 있으면 `--allow-rerun`을 줘도 재실행을 차단한다. `--allow-rerun`은 부분 실행/중단 세션 복구에만 사용해 완료된 실행 증거가 중복 주문으로 오염되지 않게 한다 |

tests/test_live_gate.py

@@ -57,26 +57,29 @@ def _write_bundle(
     evaluation_errors=None,
     walk_forward_errors=None,
     metric_overrides=None,
+    strategy_specs=None,
 ):
     allowed_modes = ["backtest", "paper", "live"] if allowed_modes is None else allowed_modes
     generated_at = generated_at or datetime(2026, 4, 29, 12, 0, 0).isoformat()
     snapshot_manifest = snapshot_manifest or _snapshot_manifest()
 
-    _write_json(
-        promotion_dir / "run_metadata.json",
-        {
-            "schema_version": LIVE_GATE_SCHEMA_VERSION,
-            "artifact_type": LIVE_GATE_ARTIFACT_TYPE,
-            "commit_hash": commit_hash,
-            "config_yaml_hash": yaml_hash,
-            "config_resolved_hash": resolved_hash,
-            "generated_at": generated_at,
-            "data_snapshot_hash": snapshot_manifest["data_snapshot_hash"],
-            "data_snapshot_manifest": snapshot_manifest,
-            "evaluation_errors": evaluation_errors or {},
-            "walk_forward_errors": walk_forward_errors or {},
-        },
-    )
+    metadata = {
+        "schema_version": LIVE_GATE_SCHEMA_VERSION,
+        "artifact_type": LIVE_GATE_ARTIFACT_TYPE,
+        "commit_hash": commit_hash,
+        "config_yaml_hash": yaml_hash,
+        "config_resolved_hash": resolved_hash,
+        "generated_at": generated_at,
+        "data_snapshot_hash": snapshot_manifest["data_snapshot_hash"],
+        "data_snapshot_manifest": snapshot_manifest,
+        "evaluation_errors": evaluation_errors or {},
+        "walk_forward_errors": walk_forward_errors or {},
+    }
+    if strategy_specs is not None:
+        metadata["strategy_specs"] = strategy_specs
+    elif strategy.startswith("target_weight_"):
+        metadata["strategy_specs"] = [{"candidate_id": strategy, "params_hash": "hash"}]
+    _write_json(promotion_dir / "run_metadata.json", metadata)
     metrics = {"total_return": 12.0, "sharpe": 0.6, "profit_factor": 1.3}
     metrics.update(metric_overrides or {})
     _write_json(
@@ -404,10 +407,14 @@ def test_target_weight_live_gate_accepts_verified_pilot_evidence(tmp_path):
             "valid_pilot_days": 60,
             "invalid_days": 0,
             "invalid_reasons": {},
+            "params_hash": "hash",
+            "params_hashes": ["hash"],
+            "params_hash_consistent": True,
             "all_promotable_days_verified": True,
         },
         target_weight_verified_pilot_days=60,
         target_weight_invalid_days=0,
+        target_weight_params_hash="hash",
     )
 
     issues = validate_live_readiness(
@@ -420,3 +427,42 @@ def test_target_weight_live_gate_accepts_verified_pilot_evidence(tmp_path):
     )
 
     assert issues == []
+
+
+def test_target_weight_live_gate_blocks_params_hash_mismatch(tmp_path):
+    strategy = "target_weight_rotation_test"
+    promotion_dir = tmp_path / "reports" / "promotion"
+    evidence_dir = tmp_path / "reports" / "paper_evidence"
+    _write_bundle(
+        promotion_dir,
+        strategy=strategy,
+        strategy_specs=[{"candidate_id": strategy, "params_hash": "canonical-hash"}],
+    )
+    _write_evidence(
+        evidence_dir,
+        strategy=strategy,
+        target_weight_evidence={
+            "required": True,
+            "valid_pilot_days": 60,
+            "invalid_days": 0,
+            "invalid_reasons": {},
+            "params_hash": "evidence-hash",
+            "params_hashes": ["evidence-hash"],
+            "params_hash_consistent": True,
+            "all_promotable_days_verified": True,
+        },
+        target_weight_verified_pilot_days=60,
+        target_weight_invalid_days=0,
+        target_weight_params_hash="evidence-hash",
+    )
+
+    issues = validate_live_readiness(
+        DummyConfig(),
+        strategy,
+        promotion_dir=promotion_dir,
+        evidence_dir=evidence_dir,
+        current_git_hash="abc123",
+        now=datetime(2026, 4, 29, 12, 0, 0),
+    )
+
+    assert any("target-weight canonical params_hash 불일치" in issue for issue in issues)