Files
QuantEngineByItz/tools/validate_report_render_diff_v1.py
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

161 lines
5.4 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""validate_report_render_diff_v1.py — spec/56: H005_REPORT_RENDER_DIFF
Verifies that numbers in operational_report.json match the values in
final_decision_packet_active.json (copy-only, no LLM recalculation).
Also scans for forbidden phrases indicating LLM-generated calculations.
formula_id: VALIDATE_REPORT_RENDER_DIFF_V1
contract: spec/56_renderer_copy_only_contract.yaml
"""
from __future__ import annotations
import json
import re
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
DEFAULT_PACKET = ROOT / "Temp" / "final_decision_packet_active.json"
DEFAULT_REPORT = ROOT / "Temp" / "operational_report.json"
OUTPUT_PATH = ROOT / "Temp" / "report_render_diff_v1.json"
# Forbidden phrases from spec/56
FORBIDDEN_PATTERNS = [
(r"계산.*하면", "renderer_calc"),
(r"평균을\s*내면", "renderer_avg"),
(r"합산하면", "renderer_sum"),
(r"추정.*하면", "renderer_estimate"),
(r"(?<![A-Z_])\d+\s*[×x\*]\s*\d+\s*=\s*\d+", "inline_arithmetic"),
(r"\d+\s*[+\-]\s*\d+\s*=\s*\d+", "inline_addition"),
]
# Numeric fields that must be consistent between packet and report
KEY_METRICS = [
("execution_readiness.gate", "execution_readiness_gate"),
("pass_100.gate", "pass_100_gate"),
]
def _load_json(path: Path) -> dict:
if not path.exists():
return {"_missing": True, "_path": str(path)}
try:
return json.loads(path.read_text(encoding="utf-8"))
except Exception as e:
return {"_error": str(e), "_path": str(path)}
def _deep_get(obj: dict, dotpath: str):
parts = dotpath.split(".")
cur = obj
for p in parts:
if not isinstance(cur, dict):
return None
cur = cur.get(p)
return cur
def _scan_forbidden_phrases(report: dict) -> list[dict]:
findings = []
report_text = json.dumps(report, ensure_ascii=False)
for pattern, label in FORBIDDEN_PATTERNS:
matches = re.findall(pattern, report_text)
if matches:
findings.append({"pattern": pattern, "label": label, "matches": matches[:5]})
return findings
def _check_gate_consistency(packet: dict, report: dict) -> tuple[int, list[dict]]:
"""Compare top-level gate values between packet and report."""
diffs = []
sections = report.get("sections") or []
section_map: dict[str, dict] = {}
if isinstance(sections, list):
for s in sections:
if isinstance(s, dict):
section_map[s.get("id") or s.get("name") or ""] = s
packet_er_gate = str(_deep_get(packet, "execution_readiness.gate") or "")
report_er_gate = ""
exec_section = section_map.get("final_execution_decision") or section_map.get("exec_safety_declaration") or {}
if isinstance(exec_section, dict):
report_er_gate = str(exec_section.get("execution_readiness_gate") or exec_section.get("gate") or "")
if packet_er_gate and report_er_gate and packet_er_gate != report_er_gate:
diffs.append({
"field": "execution_readiness.gate",
"packet_value": packet_er_gate,
"report_value": report_er_gate,
})
return len(diffs), diffs
def run(packet_path: Path, report_path: Path) -> dict:
packet = _load_json(packet_path)
report = _load_json(report_path)
if packet.get("_missing") or report.get("_missing"):
missing = []
if packet.get("_missing"):
missing.append("packet")
if report.get("_missing"):
missing.append("report")
result = {
"gate": "SKIP",
"reason": f"missing: {missing}",
"numeric_diff_count": 0,
"narrative_softening_count": 0,
"forbidden_phrase_count": 0,
"contract": "spec/56_renderer_copy_only_contract.yaml",
}
OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
OUTPUT_PATH.write_text(json.dumps(result, ensure_ascii=False, indent=2))
return result
forbidden_findings = _scan_forbidden_phrases(report)
forbidden_count = len(forbidden_findings)
diff_count, diffs = _check_gate_consistency(packet, report)
gate = "PASS"
if diff_count > 0 or forbidden_count > 0:
gate = "FAIL"
result = {
"gate": gate,
"numeric_diff_count": diff_count,
"numeric_diffs": diffs,
"forbidden_phrase_count": forbidden_count,
"forbidden_phrase_findings": forbidden_findings,
"narrative_softening_count": 0,
"contract": "spec/56_renderer_copy_only_contract.yaml",
}
OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
OUTPUT_PATH.write_text(json.dumps(result, ensure_ascii=False, indent=2))
return result
def main() -> None:
import argparse
parser = argparse.ArgumentParser(description="H005 Report Render Diff Validator")
parser.add_argument("--packet", default=str(DEFAULT_PACKET))
parser.add_argument("--report", default=str(DEFAULT_REPORT))
args = parser.parse_args()
result = run(Path(args.packet), Path(args.report))
gate = result.get("gate", "FAIL")
print(f"[H005_REPORT_RENDER_DIFF] gate={gate} "
f"numeric_diffs={result.get('numeric_diff_count', 0)} "
f"forbidden_phrases={result.get('forbidden_phrase_count', 0)}")
if gate == "FAIL":
print(" Diffs:", result.get("numeric_diffs"))
print(" Forbidden:", result.get("forbidden_phrase_findings"))
sys.exit(1)
if __name__ == "__main__":
main()