QuantEngineByItz/tools/validate_factor_lifecycle_completeness_v1.py

#!/usr/bin/env python3
"""validate_factor_lifecycle_completeness_v1.py — FACTOR_LIFECYCLE_COMPLETENESS_V1

실측 기반 팩터 생애주기 레지스트리 정합성을 검증한다.
- spec/factor_lifecycle_registry.yaml과 Temp/factor_shadow_eligibility_v1.json를 대조.
- 실측상 BLOCKED인데 명세상 shadow/active로 지정된 하드 정합성 위반 탐지 (FAIL)
- 실측상 ELIGIBLE인데 명세상 draft로 묶여서 shadow 승격 자격이 충분한 팩터들 탐지 및 요약 보고
- spec/13_formula_registry.yaml와 factor_lifecycle_registry.yaml 간의 팩터 개수 정합성 검증 (누락 시 WARN)

출력: Temp/factor_lifecycle_completeness_v1.json
"""
from __future__ import annotations

import argparse
import json
import yaml
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
TEMP = ROOT / "Temp"
FORMULA_ID = "FACTOR_LIFECYCLE_COMPLETENESS_V1"


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--registry", default="spec/factor_lifecycle_registry.yaml")
    ap.add_argument("--eligibility", default="Temp/factor_shadow_eligibility_v1.json")
    ap.add_argument("--formula-reg", default="spec/13_formula_registry.yaml")
    ap.add_argument("--out", default="Temp/factor_lifecycle_completeness_v1.json")
    args = ap.parse_args()

    reg_path = ROOT / args.registry if not Path(args.registry).is_absolute() else Path(args.registry)
    elig_path = ROOT / args.eligibility if not Path(args.eligibility).is_absolute() else Path(args.eligibility)
    freg_path = ROOT / args.formula_reg if not Path(args.formula_reg).is_absolute() else Path(args.formula_reg)
    out_path = ROOT / args.out if not Path(args.out).is_absolute() else Path(args.out)

    if not reg_path.exists():
        print(f"[ERROR] Registry not found: {reg_path}")
        return 1
    if not elig_path.exists():
        print(f"[ERROR] Eligibility file not found: {elig_path}")
        return 1

    # Load inputs
    registry_data = yaml.safe_load(reg_path.read_text(encoding="utf-8")) or {}
    eligibility_data = json.loads(elig_path.read_text(encoding="utf-8")) or {}
    
    formula_reg_data = {}
    if freg_path.exists():
        formula_reg_data = yaml.safe_load(freg_path.read_text(encoding="utf-8")) or {}

    factors = registry_data.get("factors") or []
    elig_rows = eligibility_data.get("rows") or []
    elig_map = {r["factor_id"]: r for r in elig_rows if "factor_id" in r}

    # 1. 팩터 개수 정합성 검증
    # formula_registry의 formulas 목록
    freg_formulas = formula_reg_data.get("formula_registry", {}).get("formulas", {})
    freg_keys = set(freg_formulas.keys())
    registry_keys = {f.get("factor_id") for f in factors if isinstance(f, dict)}
    
    missing_in_lifecycle = list(freg_keys - registry_keys)
    extra_in_lifecycle = list(registry_keys - freg_keys)

    violations = []
    shadow_ready_candidates = []
    
    # 2. 실측-명세 정합성 검사
    for f in factors:
        if not isinstance(f, dict):
            continue
        fid = f.get("factor_id", "UNKNOWN")
        promotion_gate = f.get("promotion_gate", "draft").lower()
        
        elig_info = elig_map.get(fid)
        if not elig_info:
            violations.append({
                "factor_id": fid,
                "type": "MISSING_ELIGIBILITY_DATA",
                "message": f"실측 섀도우 적격성 데이터에 팩터 {fid}가 누락되었습니다."
            })
            continue
            
        eligibility = elig_info.get("eligibility", "UNKNOWN")
        
        # 하드 위반: 데이터 결측(BLOCKED 또는 NO_REQUIRED_DATA)인데 shadow나 active로 지정된 경우
        if eligibility in ("BLOCKED", "NO_REQUIRED_DATA") and promotion_gate in ("shadow", "active", "candidate"):
            violations.append({
                "factor_id": fid,
                "type": "PROMOTION_VIOLATION",
                "message": f"팩터 {fid}는 실측 데이터 결측 상태({eligibility})이나 명세 상 {promotion_gate}로 승급되어 있습니다."
            })
            
        # shadow 승격 자격이 충분한 draft 팩터 탐지
        if eligibility == "ELIGIBLE" and promotion_gate == "draft":
            shadow_ready_candidates.append({
                "factor_id": fid,
                "coverage_pct": elig_info.get("coverage_pct"),
                "present_count": elig_info.get("present_count"),
                "required_field_count": elig_info.get("required_field_count")
            })

    # 전체 게이트 상태 결정
    # 하드 위반(PROMOTION_VIOLATION 등)이 있으면 FAIL
    gate_status = "FAIL" if violations else "PASS"
    
    # 3. 결과 JSON 조립
    result = {
        "formula_id": FORMULA_ID,
        "gate": gate_status,
        "summary": {
            "total_factors_in_lifecycle": len(factors),
            "total_formulas_in_registry": len(freg_keys),
            "missing_factors_count": len(missing_in_lifecycle),
            "extra_factors_count": len(extra_in_lifecycle),
            "shadow_ready_count": len(shadow_ready_candidates),
            "violation_count": len(violations),
        },
        "shadow_ready_candidates": shadow_ready_candidates,
        "missing_in_lifecycle": missing_in_lifecycle,
        "extra_in_lifecycle": extra_in_lifecycle,
        "violations": violations,
        "note": (
            "FACTOR_LIFECYCLE_COMPLETENESS_V1: 팩터 생애주기 명세와 실측 데이터 일치성 검증. "
            "BLOCKED/NO_REQUIRED_DATA 팩터가 shadow/active로 승급되어 있으면 FAIL 판정. "
            "draft 상태 중 GatherTradingData.json에 모든 입력 필드가 실측된 팩터는 shadow_ready_candidates로 분류."
        )
    }

    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    
    print(f"[{FORMULA_ID}] gate={gate_status} total={len(factors)} shadow_ready={len(shadow_ready_candidates)} violations={len(violations)}")
    if shadow_ready_candidates:
        print(f"  Shadow-ready candidates (first 5): {[c['factor_id'] for c in shadow_ready_candidates[:5]]}")
    if violations:
        print(f"  [VIOLATION] First violation: {violations[0]['message']}")
        
    return 0 if gate_status == "PASS" else 1


if __name__ == "__main__":
    raise SystemExit(main())