QuantEngineByItz/tools/validate_report_section_completeness_v1.py

#!/usr/bin/env python3
"""
validate_report_section_completeness_v1.py
모든 REPORT_SECTION_ORDER 섹션이 operational_report.json에 존재하고 비어있지 않은지 검증.
section_errors도 검사하여 처리 오류가 있으면 WARN 표기.
누락 섹션이 있으면 gate=FAIL 로 종료(exit 1).
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]

REPORT_SECTION_ORDER = [
    "exec_safety_declaration", "final_judgment_table", "final_execution_decision",
    "concise_hts_input_sheet", "watch_breakout_gate",
    "reference_price_ledger",
    "single_conclusion", "immediate_execution_playbook", "market_context_learning_note",
    "portfolio_performance_summary",
    "portfolio_sector_exposure_summary",
    "sector_universe_refresh_audit_v1",
    "sector_trend_analysis_v1",
    "etf_representative_monitor_v1",
    "performance_readiness_summary",
    "operational_eval_queue_summary",
    "investment_quality_headline", "operational_truth_score",
    "execution_readiness_matrix", "pass_100_criteria",
    "today_decision_summary_card", "routing_serving_trace",
    "export_gate_diagnosis", "QEH_AUDIT_BLOCK",
    "backdata_feature_bank_table", "alpha_lead_table", "anti_distribution_table",
    "profit_preservation_table", "smart_cash_raise_table", "execution_quality_table",
    "decision_trace_table", "anti_whipsaw_reentry_gate", "proposal_reference_sheet",
    "satellite_buy_proposal_sheet", "core_satellite_timing_gate_table",
    "engine_feedback_loop_report", "prediction_evaluation_improvement_report",
    "rule_lifecycle_governance_report",
]

MISSING_DATA_TOKEN = "DATA_MISSING — 하네스 업데이트 필요"


def _missing_category(section_name: str) -> str:
    if section_name in {"fundamental_quality_gate_v1", "horizon_allocation_lock_v1", "smart_money_liquidity_gate_v1"}:
        return "core_signal_gap"
    if section_name in {"benchmark_relative_harness_table", "index_relative_health_table", "entry_freshness_gate_table", "sell_value_preservation_gate_table", "watch_release_checklist"}:
        return "market_gate_gap"
    if section_name in {"engine_feedback_loop_report", "prediction_evaluation_improvement_report", "performance_readiness_summary"}:
        return "performance_gate_gap"
    if section_name in {"alpha_lead_table", "anti_distribution_table", "profit_preservation_table", "smart_cash_raise_table", "execution_quality_table", "sell_priority_decision_table"}:
        return "decision_table_gap"
    return "other_gap"


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--report-json", default=str(ROOT / "Temp" / "operational_report.json"))
    args = ap.parse_args()

    path = Path(args.report_json)
    if not path.exists():
        print(f"[오류] 리포트 JSON 없음: {path}")
        return 1

    report = json.loads(path.read_text(encoding="utf-8"))
    sections_list = report.get("sections", [])
    section_errors = report.get("section_errors", [])

    present = {s["name"] for s in sections_list if s.get("name")}
    non_empty = {
        s["name"] for s in sections_list
        if s.get("name") and s.get("markdown", "").strip()
    }

    missing  = [n for n in REPORT_SECTION_ORDER if n not in present]
    empty    = [n for n in REPORT_SECTION_ORDER if n in present and n not in non_empty]
    err_names = [e["section"] for e in section_errors if isinstance(e, dict)]
    missing_data_rows = []
    for section in sections_list:
        if not isinstance(section, dict):
            continue
        name = str(section.get("name") or "")
        markdown = str(section.get("markdown") or "")
        if not name or name == "section_processing_errors":
            continue
        if MISSING_DATA_TOKEN not in markdown:
            continue
        missing_data_rows.append({
            "section": name,
            "category": _missing_category(name),
            "missing_line_count": sum(1 for line in markdown.splitlines() if MISSING_DATA_TOKEN in line),
        })

    # 결과 출력
    print(f"REPORT_SECTION_ORDER 기준: {len(REPORT_SECTION_ORDER)}개 섹션 검사")
    print(f"  - 존재:         {len([n for n in REPORT_SECTION_ORDER if n in present])}개")
    print(f"  - 누락(missing): {len(missing)}개")
    print(f"  - 빈 섹션(empty): {len(empty)}개")
    print(f"  - 처리 오류 섹션: {len(section_errors)}개")

    if missing:
        print("\n[FAIL] 누락 섹션 목록:")
        for n in missing:
            print(f"  MISSING: {n}")

    if empty:
        print("\n[WARN] 빈 섹션 목록:")
        for n in empty:
            print(f"  EMPTY: {n}")

    if section_errors:
        print(f"\n[WARN] 섹션 처리 오류 ({len(section_errors)}건):")
        for e in section_errors:
            if isinstance(e, dict):
                print(f"  ERROR: {e.get('section', '?')} — {e.get('error', '?')}")

    result = {
        "validator": "validate_report_section_completeness_v1",
        "total_expected": len(REPORT_SECTION_ORDER),
        "present": len([n for n in REPORT_SECTION_ORDER if n in present]),
        "missing": missing,
        "empty": empty,
        "section_error_count": len(section_errors),
        "section_errors": section_errors,
        "gate": "FAIL" if missing else "PASS",
    }

    out = ROOT / "Temp" / "report_section_completeness.json"
    out.write_text(json.dumps(result, indent=2, ensure_ascii=False), encoding="utf-8")
    inventory_out = ROOT / "Temp" / "missing_data_inventory_v1.json"
    inventory_result = {
        "validator": "validate_report_section_completeness_v1",
        "section_count": len(sections_list),
        "missing_section_count": len(missing_data_rows),
        "categories": {},
        "sections": missing_data_rows,
    }
    for row in missing_data_rows:
        cat = row["category"]
        inventory_result["categories"][cat] = inventory_result["categories"].get(cat, 0) + 1
    inventory_out.write_text(json.dumps(inventory_result, indent=2, ensure_ascii=False), encoding="utf-8")
    print(f"\nREPORT_SECTION_COMPLETENESS: gate={result['gate']} missing={len(missing)} empty={len(empty)} section_errors={len(section_errors)}")
    print(f"OUTPUT: {out}")
    print(f"MISSING_DATA_INVENTORY: sections={len(missing_data_rows)} OUTPUT: {inventory_out}")

    if missing:
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())