Files
QuantEngineByItz/tools/validate_report_section_completeness_v1.py
T
kjh2064 ee3e799de1 feat: 리밸런싱 엔진 V1 + GAS 버그 수정 (2026-06-13)
주요 변경:
- tools/build_rebalance_engine_v1.py: REBALANCE_ENGINE_V1 신규
  * account_snapshot 직접 합산(_build_snap_position_map) → 소수주 분리 행 병합
  * 레짐 소스 macro.REGIME_PRELIM 최우선 (GAS 와 동일)
- src/gas_adapter_parts/gdf_06_rebalance.gs: runRebalanceSheet_() 신규
  * Logger.log / getSpreadsheet_() 로 run_all 연동 수정
- src/gas_adapter_parts/gdc_01_fetch_fundamentals.gs
  * _mergePositionRecord_(): 소수주 중복 행 합산 신규
  * parseInt → parseFloat (qty, availQty)
- src/gas_adapter_parts/gdf_01_price_metrics.gs
  * 미보유 종목 SELL_READY → WATCH_EXIT_SIGNAL
- spec/41_release_dag.yaml: build_rebalance_sheet 노드 추가 (step_count 63)
- spec/51_formula_lifecycle_registry.yaml: REBALANCE_ENGINE_V1 등록

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-13 13:20:14 +09:00

104 lines
3.9 KiB
Python

#!/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",
"single_conclusion", "immediate_execution_playbook", "market_context_learning_note",
"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",
]
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)]
# 결과 출력
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")
print(f"\nREPORT_SECTION_COMPLETENESS: gate={result['gate']} missing={len(missing)} empty={len(empty)} section_errors={len(section_errors)}")
print(f"OUTPUT: {out}")
if missing:
return 1
return 0
if __name__ == "__main__":
sys.exit(main())