diff --git a/tools/build_honest_performance_guard_v2.py b/tools/build_honest_performance_guard_v2.py new file mode 100644 index 0000000..9c359e3 --- /dev/null +++ b/tools/build_honest_performance_guard_v2.py @@ -0,0 +1,298 @@ +#!/usr/bin/env python3 +""" +build_honest_performance_guard_v2.py +──────────────────────────────────────────────────────────────────────── +정직 성과증빙 하네스 V2 (P0_01 단계) + +P0_01: design vs validated 분리를 엄격하게 + +모든 *_score 필드에 score_kind ∈ {DESIGN, VALIDATED} 라벨을 강제하고, +VALIDATED는 live_sample_n >= 30일 때만 허용한다. +보고서에 노출되는 점수는 VALIDATED만 허용. + +출력: + - Temp/honest_performance_guard_v2.json + - Temp/p0_01_strictness_report.json +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path +from datetime import datetime +from typing import Any + +ROOT = Path(__file__).resolve().parent.parent + +# 입력 파일 +OP_REPORT = ROOT / "Temp" / "operational_report.json" +REBOUND_EFF = ROOT / "Temp" / "rebound_sell_efficiency_v1.json" +LATE_CHASE = ROOT / "Temp" / "late_chase_attribution_v1.json" +PREDICTION_ACC = ROOT / "Temp" / "prediction_accuracy_harness_v2.json" + +# 출력 파일 +OUTPUT_V2 = ROOT / "Temp" / "honest_performance_guard_v2.json" +REPORT_P001 = ROOT / "Temp" / "p0_01_strictness_report.json" + +SAMPLE_THRESHOLD = 30 +ACCEPTED_SCORE_KINDS = {"DESIGN", "VALIDATED"} + +if sys.stdout.encoding and sys.stdout.encoding.lower() not in ("utf-8", "utf8"): + sys.stdout = open(sys.stdout.fileno(), mode="w", encoding="utf-8", buffering=1) + + +def load_json(p: Path) -> dict | list: + if not p.exists(): + return {} + try: + return json.loads(p.read_text(encoding="utf-8")) + except Exception as e: + print(f"[WARN] Failed to load {p.name}: {e}") + return {} + + +def check_all_scores_have_kind_and_sample_n(obj: Any, path: str = "") -> list[dict]: + """모든 *_score 필드가 score_kind와 sample_n을 가지는지 검사.""" + violations = [] + + if isinstance(obj, dict): + for key, value in obj.items(): + current_path = f"{path}.{key}" if path else key + + # *_score 필드 검사 + if key.endswith("_score"): + if not isinstance(value, dict): + violations.append({ + "path": current_path, + "issue": "SCORE_NOT_DICT", + "value": value, + "detail": f"점수가 dict가 아님. 값={value}" + }) + else: + # score_kind 검사 + score_kind = value.get("score_kind") + sample_n = value.get("sample_n") + score_value = value.get("value") + + if score_kind is None: + violations.append({ + "path": current_path, + "issue": "MISSING_SCORE_KIND", + "detail": "score_kind 필드 누락" + }) + elif score_kind not in ACCEPTED_SCORE_KINDS: + violations.append({ + "path": current_path, + "issue": "INVALID_SCORE_KIND", + "value": score_kind, + "detail": f"허용되지 않는 값: {score_kind}" + }) + + if sample_n is None: + violations.append({ + "path": current_path, + "issue": "MISSING_SAMPLE_N", + "detail": "sample_n 필드 누락" + }) + + # VALIDATED인데 sample_n < 30 검사 + if score_kind == "VALIDATED" and isinstance(sample_n, int): + if sample_n < SAMPLE_THRESHOLD: + violations.append({ + "path": current_path, + "issue": "INVALID_VALIDATED_LABEL", + "sample_n": sample_n, + "detail": f"VALIDATED 라벨인데 sample_n={sample_n} < {SAMPLE_THRESHOLD}" + }) + + # 재귀 검사 + elif isinstance(value, (dict, list)): + violations.extend(check_all_scores_have_kind_and_sample_n(value, current_path)) + + elif isinstance(obj, list): + for i, item in enumerate(obj): + current_path = f"{path}[{i}]" + violations.extend(check_all_scores_have_kind_and_sample_n(item, current_path)) + + return violations + + +def build_strictness_report(rebound: dict, chase: dict, pred_acc: dict) -> dict: + """P0_01 엄격성 검사 보고서 작성.""" + report = { + "phase": "P0_01_DESIGN_VS_VALIDATED_SEPARATION", + "generated_at": datetime.now().isoformat(), + "threshold_sample_min": SAMPLE_THRESHOLD, + "findings": { + "rebound_efficiency": {}, + "late_chase_attribution": {}, + "prediction_accuracy": {} + }, + "violations": [], + "corrections_required": [] + } + + # 1. rebound_efficiency 검사 + rb_metrics = rebound.get("metrics", {}) + rb_combo = rb_metrics.get("combo_count", 0) + rb_score = rb_metrics.get("rebound_efficiency_score", 0) + + report["findings"]["rebound_efficiency"] = { + "metric_name": "rebound_efficiency_score", + "current_value": rb_score, + "sample_n": rb_combo, + "meets_validated_threshold": rb_combo >= SAMPLE_THRESHOLD, + "required_score_kind": "VALIDATED" if rb_combo >= SAMPLE_THRESHOLD else "DESIGN", + "annotation_suffix": f" [설계점수, n={rb_combo}]" if rb_combo < SAMPLE_THRESHOLD else "" + } + + if rb_combo < SAMPLE_THRESHOLD: + report["corrections_required"].append({ + "metric": "rebound_efficiency_score", + "action": "ANNOTATE_DESIGN", + "new_structure": { + "score_kind": "DESIGN", + "value": rb_score, + "sample_n": rb_combo, + "annotation": f"n={rb_combo} < {SAMPLE_THRESHOLD}. 실측 미검증." + } + }) + + # 2. late_chase_attribution 검사 + chase_metrics = chase.get("metrics", {}) + chase_sample = chase_metrics.get("sample_n", 0) + chase_rate = chase_metrics.get("chase_entry_rate_pct", 0) + + report["findings"]["late_chase_attribution"] = { + "metric_name": "late_chase_attribution", + "current_value": chase_rate, + "sample_n": chase_sample, + "meets_validated_threshold": chase_sample >= SAMPLE_THRESHOLD, + "required_score_kind": "VALIDATED" if chase_sample >= SAMPLE_THRESHOLD else "DESIGN" + } + + if chase_sample < SAMPLE_THRESHOLD: + report["corrections_required"].append({ + "metric": "late_chase_attribution", + "action": "ANNOTATE_DESIGN", + "new_structure": { + "score_kind": "DESIGN", + "value": chase_rate, + "sample_n": chase_sample, + "annotation": f"뒷박 차단 효과 미검증 (n={chase_sample})" + } + }) + + # 3. prediction_accuracy 검사 + t5_sample = pred_acc.get("t5_sample", 0) + t5_rate = pred_acc.get("t5_op_rate", 0) + + report["findings"]["prediction_accuracy"] = { + "metric_name": "t5_match_rate_pct", + "current_value": t5_rate, + "sample_n": t5_sample, + "meets_validated_threshold": t5_sample >= SAMPLE_THRESHOLD, + "required_score_kind": "VALIDATED" if t5_sample >= SAMPLE_THRESHOLD else "DESIGN" + } + + if t5_sample < SAMPLE_THRESHOLD: + report["corrections_required"].append({ + "metric": "t5_match_rate_pct", + "action": "ANNOTATE_DESIGN", + "new_structure": { + "score_kind": "DESIGN", + "value": t5_rate, + "sample_n": t5_sample, + "annotation": f"실측 미검증 (n={t5_sample})" + } + }) + + # 최종 verdict + report["verdict"] = { + "all_scores_properly_labeled": len(report["corrections_required"]) == 0, + "required_corrections_count": len(report["corrections_required"]), + "status": "PASS" if len(report["corrections_required"]) == 0 else "FAIL_CORRECTION_REQUIRED" + } + + return report + + +def main() -> int: + print("=" * 80) + print(" P0_01: Design vs Validated 엄격한 분리") + print("=" * 80) + + # 입력 로드 + rebound = load_json(REBOUND_EFF) + chase = load_json(LATE_CHASE) + pred_acc = load_json(PREDICTION_ACC) + + # P0_01 보고서 생성 + p001_report = build_strictness_report(rebound, chase, pred_acc) + + print(f"\n[1] 재정렬 효율 (rebound_efficiency_score)") + rb_find = p001_report["findings"]["rebound_efficiency"] + print(f" 현재값: {rb_find['current_value']}") + print(f" 표본 수: {rb_find['sample_n']} / {SAMPLE_THRESHOLD}") + print(f" 필수 라벨: {rb_find['required_score_kind']}") + + print(f"\n[2] 뒷박 매수 (late_chase_attribution)") + chase_find = p001_report["findings"]["late_chase_attribution"] + print(f" 현재값: {chase_find['current_value']}") + print(f" 표본 수: {chase_find['sample_n']} / {SAMPLE_THRESHOLD}") + print(f" 필수 라벨: {chase_find['required_score_kind']}") + + print(f"\n[3] 예측 정확도 (T+5 일치율)") + pred_find = p001_report["findings"]["prediction_accuracy"] + print(f" 현재값: {pred_find['current_value']}%") + print(f" 표본 수: {pred_find['sample_n']} / {SAMPLE_THRESHOLD}") + print(f" 필수 라벨: {pred_find['required_score_kind']}") + + print(f"\n[결과]") + print(f" 필요한 수정: {p001_report['verdict']['required_corrections_count']}") + print(f" 상태: {p001_report['verdict']['status']}") + + # 보고서 저장 + REPORT_P001.write_text( + json.dumps(p001_report, ensure_ascii=False, indent=2), + encoding="utf-8" + ) + print(f"\n✓ P0_01 보고서 저장: {REPORT_P001.name}") + + # V2 가드 생성 + guard_v2 = { + "schema_version": "honest_performance_guard_v2", + "generated_at": datetime.now().isoformat(), + "p0_01_strictness": p001_report["verdict"], + "required_corrections": p001_report["corrections_required"], + "action_plan": [ + { + "step": 1, + "title": "모든 *_score 필드를 dict 구조로 변환", + "fields": ["score_kind", "value", "sample_n", "annotation"] + }, + { + "step": 2, + "title": "각 필드에 score_kind ∈ {DESIGN, VALIDATED} 할당", + "rule": "sample_n >= 30 → VALIDATED, else → DESIGN" + }, + { + "step": 3, + "title": "보고서 노출 규칙 적용", + "rule": "DESIGN 점수는 보고서 요약에 단독 노출 금지. (설계, n=N) 접미사 필수" + } + ] + } + + OUTPUT_V2.write_text( + json.dumps(guard_v2, ensure_ascii=False, indent=2), + encoding="utf-8" + ) + print(f"✓ P0_01 가드 저장: {OUTPUT_V2.name}") + + return 0 if p001_report['verdict']['status'] == "PASS" else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/build_p0_02_masking_removal.py b/tools/build_p0_02_masking_removal.py new file mode 100644 index 0000000..88e35a6 --- /dev/null +++ b/tools/build_p0_02_masking_removal.py @@ -0,0 +1,173 @@ +#!/usr/bin/env python3 +""" +build_p0_02_masking_removal.py +──────────────────────────────────────────────────────────────────────── +P0_02: 값 손상 지표에서 adjusted 마스킹 제거 + +핵심 변경: +1. value_damage_raw_pct는 게이트 입력으로 사용 (항상 raw 값) +2. value_damage_adjusted_pct는 annotation only (참고용) +3. cap_pass=false를 summary에 그대로 전파 +4. 같은 지표가 3개의 다른 값을 가지는 문제 제거 + +출력: + - Temp/p0_02_masking_removal_report.json +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path +from datetime import datetime + +ROOT = Path(__file__).resolve().parent.parent + +# 입력 파일 +CASH_RECOVERY = ROOT / "Temp" / "cash_recovery_optimizer_v4.json" +SMART_CASH_V7 = ROOT / "Temp" / "smart_cash_recovery_v7_authoritative.json" + +# 출력 파일 +REPORT_P002 = ROOT / "Temp" / "p0_02_masking_removal_report.json" + +if sys.stdout.encoding and sys.stdout.encoding.lower() not in ("utf-8", "utf8"): + sys.stdout = open(sys.stdout.fileno(), mode="w", encoding="utf-8", buffering=1) + + +def load_json(p: Path) -> dict: + if not p.exists(): + return {} + try: + return json.loads(p.read_text(encoding="utf-8")) + except Exception as e: + print(f"[WARN] Failed to load {p.name}: {e}") + return {} + + +def find_masking_violations(cash_rec: dict, smart_v7: dict) -> list[dict]: + """adjusted가 0.0으로 강제되는 부분 찾기.""" + violations = [] + + # 1. cash_recovery_optimizer_v4에서 raw vs adjusted 충돌 검사 + raw_dmg = cash_rec.get("value_damage_raw_pct", 0) + adj_dmg = cash_rec.get("value_damage_adjusted_pct", 0) + + if raw_dmg > 0 and adj_dmg == 0.0: + violations.append({ + "location": "cash_recovery_optimizer_v4.json", + "issue": "ADJUSTED_MASKED_TO_ZERO", + "raw_value": raw_dmg, + "adjusted_value": adj_dmg, + "detail": f"raw={raw_dmg} > adjusted={adj_dmg}. adjusted가 마스킹되었을 가능성.", + "severity": "CRITICAL" + }) + + # 2. smart_cash_recovery_v7에서 마스킹 검사 + raw_v7 = smart_v7.get("raw_value_damage_pct_avg") + opt_v7 = smart_v7.get("optimized_value_damage_pct_avg") + cap_pass = smart_v7.get("cap_pass") + + if raw_v7 is not None and opt_v7 is not None: + if raw_v7 > opt_v7 and cap_pass == False: + violations.append({ + "location": "smart_cash_recovery_v7_authoritative.json", + "issue": "CAP_FAIL_NOT_PROPAGATED", + "raw_value": raw_v7, + "optimized_value": opt_v7, + "cap_pass": cap_pass, + "detail": f"raw={raw_v7} > optimized={opt_v7} AND cap_pass=false. 하지만 summary에 cap_pass 정보가 전파되지 않을 가능성.", + "severity": "HIGH" + }) + + return violations + + +def build_p002_report(cash_rec: dict, smart_v7: dict) -> dict: + """P0_02 마스킹 제거 보고서.""" + violations = find_masking_violations(cash_rec, smart_v7) + + report = { + "phase": "P0_02_NO_ADJUSTED_MASKING", + "generated_at": datetime.now().isoformat(), + "violations_found": len(violations), + "violations": violations, + "required_actions": [ + { + "action": "USE_RAW_AS_GATE_INPUT", + "description": "value_damage_raw_pct를 게이트 입력으로 항상 사용", + "fields": ["value_damage_raw_pct"], + "rule": "gate_input = raw, adjusted는 annotation only" + }, + { + "action": "REMOVE_MASKING_LOGIC", + "description": "adjusted=0.0 강제 로직 제거", + "files": [ + "tools/build_cash_recovery_optimizer_v4.py (line 109-113)", + "tools/build_value_preservation_scorer_v1.py" + ] + }, + { + "action": "PROPAGATE_CAP_PASS", + "description": "cap_pass=false를 summary에 명시적으로 표시", + "example": "raw=15.7 > cap=10.0 → cap_pass=false (summary에 명시)" + } + ], + "metric_structure": { + "before_p002": { + "value_damage_raw_pct": 15.7, + "value_damage_adjusted_pct": 0.0, + "cap_pass": "MISSING_FROM_SUMMARY" + }, + "after_p002": { + "value_damage_raw_pct": 15.7, + "value_damage_adjusted_pct": {"value": 15.7, "annotation": "raw 값 (cap=10.0)"}, + "cap_pass": True, + "gate_input": "value_damage_raw_pct (항상 raw)" + } + } + } + + return report + + +def main() -> int: + print("=" * 80) + print(" P0_02: Adjusted 마스킹 제거 및 Raw 값 복원") + print("=" * 80) + + # 입력 로드 + cash_rec = load_json(CASH_RECOVERY) + smart_v7 = load_json(SMART_CASH_V7) + + # P0_02 보고서 생성 + p002_report = build_p002_report(cash_rec, smart_v7) + + print(f"\n[검사 결과]") + print(f" 마스킹 위반 발견: {p002_report['violations_found']}") + + for i, v in enumerate(p002_report["violations"], 1): + print(f"\n [{i}] {v['issue']}") + print(f" 위치: {v['location']}") + print(f" 심각도: {v['severity']}") + if "raw_value" in v: + print(f" raw: {v['raw_value']}") + if "adjusted_value" in v: + print(f" adjusted: {v['adjusted_value']}") + + print(f"\n[필수 조치]") + for i, action in enumerate(p002_report["required_actions"], 1): + print(f" {i}. {action['action']}") + print(f" → {action['description']}") + + # 보고서 저장 + REPORT_P002.write_text( + json.dumps(p002_report, ensure_ascii=False, indent=2), + encoding="utf-8" + ) + print(f"\n✓ P0_02 보고서 저장: {REPORT_P002.name}") + + return 0 if p002_report['violations_found'] == 0 else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/build_p0_03_unified_coverage.py b/tools/build_p0_03_unified_coverage.py new file mode 100644 index 0000000..3533811 --- /dev/null +++ b/tools/build_p0_03_unified_coverage.py @@ -0,0 +1,223 @@ +#!/usr/bin/env python3 +""" +build_p0_03_unified_coverage.py +──────────────────────────────────────────────────────────────────────── +P0_03: 커버리지 분모 통일 — 288 vs 204 분모 불일치 해소 + +핵심 변경: +1. spec/13_formula_registry.yaml에서 active=true 공식만 수집 (단일 분모) +2. deprecated/orphan을 active=false로 명시 +3. 골든 커버리지를 단일 분모로만 계산 +4. 장식용 100% 필드 전면 삭제 + +출력: + - Temp/p0_03_unified_coverage.json + - Temp/p0_03_denominator_audit.json +""" + +from __future__ import annotations + +import json +import sys +import re +from pathlib import Path +from datetime import datetime +from typing import Any + +ROOT = Path(__file__).resolve().parent.parent + +# 입력 파일 +FORMULA_REGISTRY = ROOT / "spec" / "13_formula_registry.yaml" +YAML_CODE_COVERAGE = ROOT / "Temp" / "yaml_code_coverage_v1.json" + +# 출력 파일 +OUTPUT_UNIFIED = ROOT / "Temp" / "p0_03_unified_coverage.json" +AUDIT_REPORT = ROOT / "Temp" / "p0_03_denominator_audit.json" + +if sys.stdout.encoding and sys.stdout.encoding.lower() not in ("utf-8", "utf8"): + sys.stdout = open(sys.stdout.fileno(), mode="w", encoding="utf-8", buffering=1) + + +def load_yaml_simple(p: Path) -> dict: + """간단한 YAML 파싱 (설치된 라이브러리 없이).""" + if not p.exists(): + return {} + text = p.read_text(encoding="utf-8") + result = {} + + # execution_order 섹션 찾기 + in_exec_order = False + current_list = [] + + for line in text.split("\n"): + stripped = line.strip() + + if stripped.startswith("execution_order:"): + in_exec_order = True + continue + + if in_exec_order: + if stripped.startswith("- "): + formula_id = stripped[2:].strip() + if formula_id: + current_list.append(formula_id) + elif stripped and not stripped.startswith("-") and not stripped.startswith("#"): + # 다음 섹션 시작 + in_exec_order = False + + result["execution_order"] = current_list + return result + + +def load_json(p: Path) -> dict | list: + if not p.exists(): + return {} if p.suffix == ".json" else {} + try: + return json.loads(p.read_text(encoding="utf-8")) + except Exception as e: + print(f"[WARN] Failed to load {p.name}: {e}") + return {} + + +def build_denominator_audit(formula_registry: dict, yaml_coverage: dict) -> dict: + """분모 감사 보고서.""" + audit = { + "generated_at": datetime.now().isoformat(), + "findings": {} + } + + # 1. execution_order에서 active=true 공식 수집 + exec_order = formula_registry.get("execution_order", []) + active_count = len(exec_order) + + audit["findings"]["execution_order"] = { + "total_in_registry": active_count, + "formula_ids": exec_order[:10] + (["..."] if len(exec_order) > 10 else []) + } + + # 2. yaml_code_coverage와의 비교 + yaml_cov = yaml_coverage.get("coverage_summary", {}) + yaml_formula_count = yaml_cov.get("formula_total", 0) + orphan_count = yaml_cov.get("orphan_code_formula_count", 0) + + audit["findings"]["yaml_code_coverage"] = { + "formula_total": yaml_formula_count, + "orphan_code_formula_count": orphan_count, + "effective_denominator": yaml_formula_count - orphan_count + } + + # 3. 분모 불일치 진단 + expected_denominator = active_count + actual_288 = 288 # 기존 분모 + actual_204 = 204 # 다른 분모 + + audit["findings"]["denominator_collision"] = { + "expected_unified": expected_denominator, + "legacy_288": actual_288, + "legacy_204": actual_204, + "collision_exists": (actual_288 != actual_204), + "recommendation": f"Use execution_order count={expected_denominator} as SINGLE source of truth" + } + + return audit + + +def build_unified_coverage(formula_registry: dict) -> dict: + """통일된 커버리지 계산.""" + exec_order = formula_registry.get("execution_order", []) + active_formula_count = len(exec_order) + + # 현재는 exec_order 개수를 분모로 사용하는 것만 계산 + unified = { + "schema_version": "unified_coverage_v1", + "generated_at": datetime.now().isoformat(), + "denominator_single_source": "spec/13_formula_registry.yaml:execution_order", + "active_formula_count": active_formula_count, + "coverage_calculation_rule": { + "numerator": "GAS implementation + Python harness implementation", + "denominator": "execution_order count (deprecated/orphan excluded)", + "formula": f"coverage_pct = (gs_impl_count + py_impl_count) / {active_formula_count} * 100" + }, + "required_corrections": [ + { + "issue": "DECORATIVE_100_FIELD_REMOVAL", + "description": "'adjusted_coverage_pct (참고용, PASS 미사용)' 같은 필드 제거", + "action": "Delete all '**_adjusted', '**_参考용' fields" + }, + { + "issue": "SINGLE_DENOMINATOR_LOCK", + "description": "모든 커버리지 계산을 execution_order 분모로 통일", + "action": f"Use denominator={active_formula_count} for all coverage metrics" + }, + { + "issue": "GOLDEN_COVERAGE_UNIFICATION", + "description": "골든 커버리지를 64.93 / 90.2 / 67.93 / 100 중 1개로 선택 불가", + "action": "Calculate single golden_coverage_ratio with execution_order denominator" + } + ], + "implementation_checklist": [ + {"step": 1, "task": "measure_yaml_gs_ps_coverage.py 업데이트 (active=true만 수집)"}, + {"step": 2, "task": "deprecated/orphan formula를 spec/13_formula_registry.yaml에서 active=false로 명시"}, + {"step": 3, "task": "모든 *_adjusted, *_참고용 필드 제거"}, + {"step": 4, "task": "validate_golden_coverage_100.py 업데이트 (단일 분모 검증)"} + ] + } + + return unified + + +def main() -> int: + print("=" * 80) + print(" P0_03: 커버리지 분모 통일 (288 vs 204 불일치 해소)") + print("=" * 80) + + # 입력 로드 + formula_reg = load_yaml_simple(FORMULA_REGISTRY) + yaml_cov = load_json(YAML_CODE_COVERAGE) + + # 분모 감사 + denominator_audit = build_denominator_audit(formula_reg, yaml_cov) + + print(f"\n[1] Execution Order 공식 수") + print(f" 총 개수: {denominator_audit['findings']['execution_order']['total_in_registry']}") + + print(f"\n[2] YAML 코드 커버리지") + yaml_find = denominator_audit["findings"]["yaml_code_coverage"] + print(f" 공식 총 수: {yaml_find['formula_total']}") + print(f" 고아 공식: {yaml_find['orphan_code_formula_count']}") + print(f" 유효 분모: {yaml_find['effective_denominator']}") + + print(f"\n[3] 분모 불일치 진단") + denom_find = denominator_audit["findings"]["denominator_collision"] + print(f" 기대값(execution_order): {denom_find['expected_unified']}") + print(f" 기존 분모 1: {denom_find['legacy_288']}") + print(f" 기존 분모 2: {denom_find['legacy_204']}") + print(f" 충돌: {'YES' if denom_find['collision_exists'] else 'NO'}") + + # 통일된 커버리지 계산 + unified_cov = build_unified_coverage(formula_reg) + + print(f"\n[4] 필수 수정사항") + for i, corr in enumerate(unified_cov['required_corrections'], 1): + print(f" {i}. {corr['issue']}") + print(f" → {corr['description']}") + print(f" → {corr['action']}") + + # 보고서 저장 + AUDIT_REPORT.write_text( + json.dumps(denominator_audit, ensure_ascii=False, indent=2), + encoding="utf-8" + ) + print(f"\n✓ P0_03 분모 감사 저장: {AUDIT_REPORT.name}") + + OUTPUT_UNIFIED.write_text( + json.dumps(unified_cov, ensure_ascii=False, indent=2), + encoding="utf-8" + ) + print(f"✓ P0_03 통일 커버리지 저장: {OUTPUT_UNIFIED.name}") + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/build_p1_01_execution_verdict_unify.py b/tools/build_p1_01_execution_verdict_unify.py new file mode 100644 index 0000000..3e6dfaa --- /dev/null +++ b/tools/build_p1_01_execution_verdict_unify.py @@ -0,0 +1,247 @@ +#!/usr/bin/env python3 +""" +build_p1_01_execution_verdict_unify.py +──────────────────────────────────────────────────────────────────────── +P1_01: 실행 권위 단일화 — '발행금지(FAIL_BLOCK_PUBLISH)인데 주문 7건' 충돌 제거 + +문제: + - operational_report.json: published_verdict=FAIL_BLOCK_PUBLISH (발행 금지) + - final_execution_decision_v4.json: hts_order_count=7, sell_allowed=true + - v8 README: hts_order_count=0, sell_allowed=false + +해결: + 1. global_execution_gate를 final_decision_packet_v2 단 한 곳에서만 산출 + 2. FAIL_BLOCK_PUBLISH → hts_order_count=0 강제 + 3. 모든 섹션이 이 단일 게이트를 '복사'만 함 + +출력: + - Temp/p1_01_execution_authority_report.json +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path +from datetime import datetime + +ROOT = Path(__file__).resolve().parent.parent + +# 입력 파일 +OP_REPORT = ROOT / "Temp" / "operational_report.json" +FINAL_EXEC = ROOT / "Temp" / "final_execution_decision_v4.json" + +# 출력 파일 +REPORT_P101 = ROOT / "Temp" / "p1_01_execution_authority_report.json" + +if sys.stdout.encoding and sys.stdout.encoding.lower() not in ("utf-8", "utf8"): + sys.stdout = open(sys.stdout.fileno(), mode="w", encoding="utf-8", buffering=1) + + +def load_json(p: Path) -> dict: + if not p.exists(): + return {} + try: + return json.loads(p.read_text(encoding="utf-8")) + except Exception as e: + print(f"[WARN] Failed to load {p.name}: {e}") + return {} + + +def audit_execution_verdict_collision(op_report: dict, final_exec: dict) -> dict: + """실행 권위 충돌 감사.""" + audit = { + "generated_at": datetime.now().isoformat(), + "authority_sources": {}, + "collisions_found": [] + } + + # 1. operational_report 판정 + op_verdict = op_report.get("sections", []) + op_final_judgment = {} + for section in op_verdict: + if section.get("name") == "final_judgment_table": + op_final_judgment = section + break + + op_verdict_summary = "UNKNOWN" + if op_final_judgment: + # operational_report의 최종 판정 찾기 + markdown = op_final_judgment.get("markdown", "") + if "EXPORT_READY" in markdown or "JSON_VALID" in markdown: + op_verdict_summary = "EXPORT_READY" + elif "FAIL" in markdown or "BLOCK" in markdown: + op_verdict_summary = "FAIL_BLOCK_PUBLISH" + + audit["authority_sources"]["operational_report"] = { + "source_file": "Temp/operational_report.json", + "verdict": op_verdict_summary, + "hts_order_count_implied": 0 if "FAIL" in op_verdict_summary else "CHECK_FINAL_EXEC" + } + + # 2. final_execution_decision_v4 판정 + final_exec_verdict = final_exec.get("global_execution_gate", "UNKNOWN") + hts_order_count = final_exec.get("hts_order_count", 0) + sell_allowed = final_exec.get("sell_allowed", False) + + audit["authority_sources"]["final_execution_decision_v4"] = { + "source_file": "Temp/final_execution_decision_v4.json", + "global_execution_gate": final_exec_verdict, + "hts_order_count": hts_order_count, + "sell_allowed": sell_allowed + } + + # 3. 충돌 검사 + if op_verdict_summary == "FAIL_BLOCK_PUBLISH" and hts_order_count > 0: + audit["collisions_found"].append({ + "collision_type": "VERDICT_ORDER_COUNT_MISMATCH", + "severity": "CRITICAL_FINANCIAL_LOSS_RISK", + "description": f"발행금지 판정인데 {hts_order_count}건 주문 생성", + "source_1": "operational_report.json → FAIL_BLOCK_PUBLISH", + "source_2": f"final_execution_decision_v4.json → hts_order_count={hts_order_count}", + "consequence": "금지된 주문이 실제 HTS에 입력될 수 있음", + "remediation": "hts_order_count=0 강제, 또는 published_verdict=EXPORT_READY 정정" + }) + + if sell_allowed and op_verdict_summary == "FAIL_BLOCK_PUBLISH": + audit["collisions_found"].append({ + "collision_type": "SELL_PERMISSION_MISMATCH", + "severity": "CRITICAL", + "description": "발행금지인데 sell_allowed=true", + "source_1": "operational_report.json → FAIL_BLOCK_PUBLISH", + "source_2": "final_execution_decision_v4.json → sell_allowed=true", + "remediation": "operational_report의 최종 판정 = final_execution_decision의 권위로 통일" + }) + + return audit + + +def build_p101_report(audit: dict) -> dict: + """P1_01 보고서.""" + report = { + "phase": "P1_01_ONE_GATE_TO_RULE_THEM", + "generated_at": datetime.now().isoformat(), + "audit_findings": audit, + "authority_ladder": [ + { + "rank": 1, + "authority": "Temp/final_decision_packet_v2.json", + "field": "global_execution_gate", + "meaning": "실행 여부의 단일 진실 (ONLY SOURCE)", + "current_status": "NOT VERIFIED" + }, + { + "rank": 2, + "authority": "spec/33_execution_precedence_lock.yaml", + "field": "execution_rules", + "meaning": "FAIL_BLOCK_PUBLISH이면 hts_order_count=0 강제" + }, + { + "rank": 3, + "authority": "모든 보고서 섹션", + "field": "global_execution_gate (복사만)", + "meaning": "rank 1의 값을 '복사'만. 재해석/조건부 판정 금지" + } + ], + "required_corrections": [ + { + "correction_id": "P1_01_A", + "title": "final_decision_packet_v2 = single truth", + "steps": [ + "final_decision_packet_v2.json에 global_execution_gate 필드 필수", + "값: EXPORT_READY | FAIL_BLOCK_PUBLISH | ... (명시적 열거)", + "모든 다른 파일은 이 값을 참조만 함" + ] + }, + { + "correction_id": "P1_01_B", + "title": "FAIL_BLOCK_PUBLISH → hts_order_count=0 강제", + "condition": "if global_execution_gate == FAIL_BLOCK_PUBLISH", + "action": "hts_order_count = 0 (override)", + "location": "spec/33_execution_precedence_lock.yaml" + }, + { + "correction_id": "P1_01_C", + "title": "모든 섹션 = 복사 렌더링만", + "files": [ + "operational_report.json", + "final_execution_decision_v4.json", + "prompts/analysis_prompt.md" + ], + "rule": "global_execution_gate = final_decision_packet_v2.global_execution_gate (조건 없음)" + } + ], + "validation_gates": [ + { + "gate": "COLLISION_COUNT", + "expected": 0, + "actual": len(audit.get("collisions_found", [])), + "status": "PASS" if len(audit.get("collisions_found", [])) == 0 else "FAIL" + }, + { + "gate": "AUTHORITY_SOURCES", + "expected": 1, + "actual": "multiple", + "status": "FAIL" if len(audit.get("authority_sources", {})) > 1 else "PASS" + }, + { + "gate": "EXECUTION_VERDICT_SOURCE_COUNT", + "expected": 1, + "rule": "final_decision_packet_v2에서만 산출" + } + ] + } + + return report + + +def main() -> int: + print("=" * 80) + print(" P1_01: 실행 권위 단일화 — '발행금지인데 주문 7건' 충돌 제거") + print("=" * 80) + + # 입력 로드 + op_report = load_json(OP_REPORT) + final_exec = load_json(FINAL_EXEC) + + # 감사 + audit = audit_execution_verdict_collision(op_report, final_exec) + + print(f"\n[1] 권위 출처 분석") + for source_name, details in audit["authority_sources"].items(): + print(f" {source_name}") + for key, val in details.items(): + if key != "source_file": + print(f" {key}: {val}") + + print(f"\n[2] 충돌 발견") + if not audit["collisions_found"]: + print(f" 충돌 0개 — 데이터 미검증") + else: + for i, collision in enumerate(audit["collisions_found"], 1): + print(f" [{i}] {collision['collision_type']} (심각도: {collision['severity']})") + print(f" → {collision['description']}") + + # 보고서 생성 + p101_report = build_p101_report(audit) + + print(f"\n[3] 필수 수정사항") + for corr in p101_report["required_corrections"]: + print(f" {corr['correction_id']}: {corr['title']}") + if "condition" in corr: + print(f" 조건: {corr['condition']}") + if "action" in corr: + print(f" 조치: {corr['action']}") + + # 보고서 저장 + REPORT_P101.write_text( + json.dumps(p101_report, ensure_ascii=False, indent=2), + encoding="utf-8" + ) + print(f"\n✓ P1_01 보고서 저장: {REPORT_P101.name}") + + return 0 if len(audit["collisions_found"]) == 0 else 1 + + +if __name__ == "__main__": + sys.exit(main())