ee3e799de1
주요 변경: - 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>
104 lines
3.9 KiB
Python
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())
|