kjh2064/QuantEngineByItz
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(report): 레포트 프로 수준 개선 — gate_trace 정형화, HTS표 재설계, 중복섹션 제거

Browse Source 
- _fmt_gate_trace(): 게이트 요약 compact 출력 (손절✓ 상대손절✓ 현금바닥⊘)
- _concise_hts_input_sheet: gate_trace 제거, 지정가/매도수량/손절가/TP2가/실행스타일 추가
- _immediate_execution_playbook: 게이트요약 compact, sell_sequence 정형화된 표
- _reference_price_ledger: watch_breakout_gate 중복 fallback 제거, prices_json 기준가 원장
- _sparkline: 데이터 4개 미만 시 데이터부족 표시
- SECTION_TITLES: 내부 formula ID 한국어 명칭으로 통일
- report dict: generated_at/section_errors 추가 (PASS)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
김재현
2026-06-14 23:05:47 +09:00
parent 20f0973f74
commit e2820065d1
5 changed files with 411 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/render_operational_report.py
+252 -26
View File
@@ -21,7 +21,7 @@ from src.quant_engine.sector_trend_analysis import build_sector_trend_analysis
SECTION_ORDER = [
"exec_safety_declaration", "final_judgment_table", "final_execution_decision",
"concise_hts_input_sheet", "watch_breakout_gate",
"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",
@@ -40,7 +40,7 @@ SECTION_ORDER = [
"backdata_feature_bank_table", "alpha_lead_table", "anti_distribution_table",
"profit_preservation_table", "smart_cash_raise_table", "execution_quality_table",
"sell_priority_decision_table", "strategy_performance_scoreboard",
"outcome_eval_window_monitor",
"performance_readiness_summary", "operational_eval_queue_summary", "outcome_eval_window_monitor",
"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",
@@ -53,6 +53,7 @@ SECTION_TITLES = {
"final_execution_decision": "최종 실행 결정",
"concise_hts_input_sheet": "HTS 입력 요약표",
"watch_breakout_gate": "투명한 감시 원장 / 돌파 감시 게이트",
"reference_price_ledger": "투명한 감시 원장",
"single_conclusion": "단일 결론",
"immediate_execution_playbook": "즉시 실행 플레이북",
"market_context_learning_note": "시장 컨텍스트 학습 노트",
@@ -68,21 +69,21 @@ SECTION_TITLES = {
"routing_serving_trace": "라우팅 서빙 추적",
"export_gate_diagnosis": "내보내기 게이트 진단",
"QEH_AUDIT_BLOCK": "QEH 감사 블록",
"fundamental_quality_gate_v1": "FUNDAMENTAL_QUALITY_GATE_V1",
"horizon_allocation_lock_v1": "HORIZON_ALLOCATION_LOCK_V1",
"smart_money_liquidity_gate_v1": "SMART_MONEY_LIQUIDITY_GATE_V1",
"routing_serving_trace_v2": "ROUTING_SERVING_DECISION_TRACE_V2",
"fundamental_multifactor_v2": "FUNDAMENTAL_MULTI_FACTOR_SCORE_V2",
"earnings_growth_quality_v1": "EARNINGS_GROWTH_QUALITY_GATE_V1",
"market_share_proxy_v1": "MARKET_SHARE_MOMENTUM_PROXY_V1",
"cashflow_stability_v1": "CASHFLOW_STABILITY_GATE_V1",
"routing_decision_explain_v1": "ROUTING_DECISION_EXPLAIN_LOCK_V1",
"benchmark_relative_harness_table": "benchmark_relative_harness_table",
"index_relative_health_table": "index_relative_health_table",
"entry_freshness_gate_table": "entry_freshness_gate_table",
"sell_value_preservation_gate_table": "sell_value_preservation_gate_table",
"watch_release_checklist": "watch_release_checklist",
"alpha_feedback_loop_report": "alpha_feedback_loop_report",
"fundamental_quality_gate_v1": "펀더멘털 품질 게이트",
"horizon_allocation_lock_v1": "지평선 배분 잠금",
"smart_money_liquidity_gate_v1": "스마트머니 유동성 게이트",
"routing_serving_trace_v2": "라우팅 서빙 결정 추적",
"fundamental_multifactor_v2": "펀더멘털 다중팩터 점수",
"earnings_growth_quality_v1": "실적 성장 품질 게이트",
"market_share_proxy_v1": "시장점유율 모멘텀 프록시",
"cashflow_stability_v1": "현금흐름 안정성 게이트",
"routing_decision_explain_v1": "라우팅 결정 설명",
"benchmark_relative_harness_table": "벤치마크 상대 하네스",
"index_relative_health_table": "지수 상대 건강 테이블",
"entry_freshness_gate_table": "진입 신선도 게이트",
"sell_value_preservation_gate_table": "매도 가치 보존 게이트",
"watch_release_checklist": "관찰 해제 체크리스트",
"alpha_feedback_loop_report": "알파 피드백 루프",
"backdata_feature_bank_table": "백데이터 특성 원장",
"alpha_lead_table": "알파 선행 테이블",
"anti_distribution_table": "분산 매도 위험 테이블",
@@ -91,6 +92,8 @@ SECTION_TITLES = {
"execution_quality_table": "체결 품질 테이블",
"sell_priority_decision_table": "매도 우선순위 결정 테이블",
"strategy_performance_scoreboard": "전략 성과 스코어보드",
"performance_readiness_summary": "성과 준비도 요약",
"operational_eval_queue_summary": "운영 T+20 대기열 요약",
"outcome_eval_window_monitor": "성과 평가 윈도우 모니터",
"decision_trace_table": "판단 추적 테이블",
"anti_whipsaw_reentry_gate": "반등 재진입 감시 게이트",
@@ -170,6 +173,10 @@ def _sparkline(values: list[Any]) -> str:
continue
if not points:
return "n/a"
if len(points) < 4:
def _fp(p: float) -> str:
return f"{p:,.0f}" if abs(p) >= 1000 or p == int(p) else f"{p:.2f}"
return "데이터부족(" + ", ".join(_fp(p) for p in points) + ")"
lo = min(points)
hi = max(points)
bars = "▁▂▃▄▅▆▇█"
@@ -182,6 +189,50 @@ def _sparkline(values: list[Any]) -> str:
return "".join(out)
def _fmt_gate_trace(trace_raw: Any) -> str:
"""gate_trace list → compact readable 게이트 요약 (예: 손절✓ 상대손절✓ 열게이트✓ 현금바닥⊘)."""
if isinstance(trace_raw, str):
try:
trace_raw = json.loads(trace_raw)
except Exception:
return str(trace_raw)[:60]
if not isinstance(trace_raw, list):
return str(trace_raw)[:60]
_labels = {
"STOP_BREACH": "손절",
"RELATIVE_STOP": "상대손절",
"INTRADAY_LOCK": "장중잠금",
"HEAT_GATE": "열게이트",
"MEAN_REVERSION_GATE": "평균회귀",
"CASH_FLOOR": "현금바닥",
"EXIT_POLICY": "청산정책",
}
_icons = {
"PASS": "✓", "FAIL": "✗", "FORCE_EXIT": "✗",
"HARD_BLOCK": "⊘", "INACTIVE": "–", "SKIP": "?",
}
parts = []
for g in trace_raw:
if not isinstance(g, dict):
continue
gate = g.get("gate", "")
result = g.get("result", "")
label = _labels.get(gate, gate[:4])
icon = _icons.get(result, result[:2])
if result == "INACTIVE":
continue
parts.append(f"{label}{icon}")
return " ".join(parts) if parts else "–"
def _build_ticker_lookup(hctx: dict, key: str) -> dict:
"""hctx에서 key를 JSON 파싱 후 ticker → row dict 매핑 반환."""
raw = _sj(hctx.get(key, []))
if not isinstance(raw, list):
return {}
return {row.get("ticker", ""): row for row in raw if isinstance(row, dict)}
# ── PHASE-0 렌더러 ────────────────────────────────────────────────────────────
def _exec_safety_declaration(hctx: dict, se: list) -> str:
@@ -228,7 +279,30 @@ def _concise_hts_input_sheet(hctx: dict, se: list) -> str:
items = _sj(hctx.get("decisions_json", []))
if not isinstance(items, list) or not items:
return _err(se, "concise_hts_input_sheet", "decisions_json 없음")
return _tbl(items, ["ticker", "name", "final_action", "gate_trace", "rs_verdict"])
lpp_map = _build_ticker_lookup(hctx, "limit_price_policy_json")
sqj_map = _build_ticker_lookup(hctx, "sell_quantities_json")
pj_map = _build_ticker_lookup(hctx, "prices_json")
rows = []
for it in items:
ticker = it.get("ticker", "")
lpp = lpp_map.get(ticker, {})
sqj = sqj_map.get(ticker, {})
pj = pj_map.get(ticker, {})
hts_p = lpp.get("hts_limit_price", "")
stop_p = pj.get("stop_price", "")
tp2_p = pj.get("tp2_price", "")
rows.append({
"ticker": ticker,
"종목명": it.get("name", ""),
"매매구분": it.get("final_action", ""),
"지정가": f"{hts_p:,}" if isinstance(hts_p, (int, float)) and hts_p else hts_p or "DATA_MISSING",
"매도수량": sqj.get("sell_qty", "-"),
"손절가": f"{stop_p:,}" if isinstance(stop_p, (int, float)) and stop_p else stop_p or "DATA_MISSING",
"TP2가": f"{tp2_p:,}" if isinstance(tp2_p, (int, float)) and tp2_p else tp2_p or "-",
"RS판정": it.get("rs_verdict", ""),
"실행스타일": lpp.get("execution_style", "-"),
})
return _tbl(rows, ["ticker", "종목명", "매매구분", "지정가", "매도수량", "손절가", "TP2가", "RS판정", "실행스타일"])
def _watch_breakout_gate(hctx: dict, se: list) -> str:
@@ -247,6 +321,37 @@ def _watch_breakout_gate(hctx: dict, se: list) -> str:
return "".join(parts)
def _reference_price_ledger(hctx: dict, se: list) -> str:
ledger = _sj(hctx.get("reference_price_ledger_json", []))
if isinstance(ledger, list) and ledger:
return _tbl(ledger, _first_keys(ledger))
# fallback: prices_json로 기준가 원장 표시
pj = _sj(hctx.get("prices_json", []))
if not isinstance(pj, list) or not pj:
return "기준가 원장 없음 — 하네스 업데이트 필요"
rows = []
for row in pj:
if not isinstance(row, dict):
continue
avg = row.get("avg_cost", "")
stop = row.get("stop_price", "")
tp1 = row.get("tp1_price", "")
tp2 = row.get("tp2_price", "")
rows.append({
"ticker": row.get("ticker", ""),
"종목명": row.get("name", ""),
"평균단가": f"{avg:,}" if isinstance(avg, (int, float)) and avg else avg or "-",
"손절가": f"{stop:,}" if isinstance(stop, (int, float)) and stop else stop or "-",
"TP1가": f"{tp1:,}" if isinstance(tp1, (int, float)) and tp1 else tp1 or "-",
"TP1상태": row.get("tp1_state", ""),
"TP2가": f"{tp2:,}" if isinstance(tp2, (int, float)) and tp2 else tp2 or "-",
"TP2상태": row.get("tp2_state", ""),
"수익률(%)": row.get("profit_pct", ""),
"이익잠금": row.get("profit_lock_stage", ""),
})
return _tbl(rows, ["ticker", "종목명", "평균단가", "손절가", "TP1가", "TP1상태", "TP2가", "TP2상태", "수익률(%)", "이익잠금"])
# ── PHASE-1 렌더러 ────────────────────────────────────────────────────────────
def _single_conclusion(hctx: dict, se: list) -> str:
@@ -270,17 +375,42 @@ def _immediate_execution_playbook(hctx: dict, se: list) -> str:
plan = _sj(hctx.get("cash_recovery_plan_json", {}))
parts = []
if isinstance(items, list) and items:
parts.append("**실행 결정**\n\n" + _tbl(items, ["ticker", "name", "final_action", "gate_trace"]))
exec_rows = []
for it in items:
exec_rows.append({
"ticker": it.get("ticker", ""),
"종목명": it.get("name", ""),
"매매구분": it.get("final_action", ""),
"게이트요약": _fmt_gate_trace(it.get("gate_trace", [])),
"RS판정": it.get("rs_verdict", ""),
})
parts.append("**실행 결정**\n\n" + _tbl(exec_rows, ["ticker", "종목명", "매매구분", "게이트요약", "RS판정"]))
else:
parts.append(_err(se, "immediate_execution_playbook", "decisions_json 없음"))
if isinstance(plan, dict):
sell_seq = plan.get("sell_sequence", "")
sell_seq = _sj(plan.get("sell_sequence", []))
exp_total = plan.get("expected_total_krw", "")
parts.append("\n\n**현금 회수 계획**\n\n" + _kv([
("매도 시퀀스", str(sell_seq)[:120]),
("예상 즉시 회수 (KRW)", plan.get("expected_total_krw", "")),
("예상 즉시 회수 (KRW)", f"{exp_total:,}" if isinstance(exp_total, (int, float)) else exp_total),
("부족액 충족", plan.get("shortfall_met", "")),
("필요 건수", plan.get("items_needed", "")),
]))
if isinstance(sell_seq, list) and sell_seq:
seq_rows = []
for item in sell_seq:
if not isinstance(item, dict):
continue
lp = item.get("limit_price", "")
ek = item.get("expected_krw", "")
seq_rows.append({
"ticker": item.get("ticker", ""),
"종목명": item.get("name", ""),
"수량": item.get("qty", ""),
"지정가": f"{lp:,}" if isinstance(lp, (int, float)) and lp else lp or "-",
"방식": item.get("preserve_style", ""),
"예상회수(KRW)": f"{ek:,}" if isinstance(ek, (int, float)) and ek else ek or "-",
})
parts.append("\n\n**매도 시퀀스**\n\n" + _tbl(seq_rows, ["ticker", "종목명", "수량", "지정가", "방식", "예상회수(KRW)"]))
return "".join(parts)
@@ -921,6 +1051,90 @@ def _strategy_performance_scoreboard(hctx: dict, se: list) -> str:
return _kv(rows)
def _performance_readiness_summary(hctx: dict, se: list) -> str:
oac_path = ROOT / "Temp" / "operational_alpha_calibration_v2.json"
prb_path = ROOT / "Temp" / "performance_readiness_replay_bridge_v1.json"
prb2_path = ROOT / "Temp" / "performance_readiness_replay_bridge_v2.json"
def _load(path: Path) -> dict:
if not path.exists():
return {}
try:
data = json.loads(path.read_text(encoding="utf-8"))
return data if isinstance(data, dict) else {}
except Exception:
return {}
oac = _load(oac_path)
if not oac:
return _err(se, "performance_readiness_summary", "operational_alpha_calibration_v2.json 없음")
prb = _load(prb_path)
prb2 = _load(prb2_path)
live = prb.get("live") if isinstance(prb.get("live"), dict) else {}
replay = prb.get("replay_informational") if isinstance(prb.get("replay_informational"), dict) else {}
metrics = oac.get("metrics") if isinstance(oac.get("metrics"), dict) else {}
rows = [
("CHECK_83 게이트", oac.get("gate", "")),
("confidence_score", oac.get("confidence_score", "")),
("performance_ready", oac.get("performance_ready", "")),
("readiness_reasons", ", ".join(oac.get("readiness_reasons", [])) if isinstance(oac.get("readiness_reasons"), list) else oac.get("readiness_reasons", "")),
("outcome_quality_score", metrics.get("outcome_quality_score", "")),
("t20_operational_sample", metrics.get("t20_operational_sample", "")),
("t5_operational_pass_rate", metrics.get("t5_operational_pass_rate", "")),
("value_damage_pct_avg", metrics.get("value_damage_pct_avg", "")),
]
if prb:
rows += [
("readiness_bridge_gate", prb.get("gate", "")),
("performance_readiness_score", prb.get("performance_readiness_score", "")),
("live_t20_count", live.get("t20_count", "")),
("live_sample_gate", live.get("sample_gate", "")),
("replay_t20_count", replay.get("t20_count", "")),
]
if prb2:
rows += [
("promotion_rule", prb2.get("promotion_rule", "")),
("promotion_allowed", prb2.get("promotion_allowed", "")),
]
rows += [
("operational_gate_note", "live T+20가 30건 미만이면 PERFORMANCE_READY로 승격하지 않음"),
("current_state", "DATA_GATED" if str(oac.get("gate", "")).upper() != "PERFORMANCE_READY" else "READY"),
]
return _kv(rows)
def _operational_eval_queue_summary(hctx: dict, se: list) -> str:
q_path = ROOT / "Temp" / "operational_eval_queue_v1.json"
if not q_path.exists():
return _err(se, "operational_eval_queue_summary", "operational_eval_queue_v1.json 없음")
try:
q = json.loads(q_path.read_text(encoding="utf-8"))
except Exception:
return _err(se, "operational_eval_queue_summary", "operational_eval_queue_v1.json 파싱 실패")
if not isinstance(q, dict):
return _err(se, "operational_eval_queue_summary", "operational_eval_queue_v1.json 구조 오류")
metrics = q.get("metrics") if isinstance(q.get("metrics"), dict) else {}
queue = q.get("queue") if isinstance(q.get("queue"), list) else []
todo = q.get("todo_protocol") if isinstance(q.get("todo_protocol"), list) else []
rows = [
("formula_id", q.get("formula_id", "")),
("as_of", q.get("as_of", "")),
("t20_days_threshold", q.get("t20_days_threshold", "")),
("records_total", metrics.get("records_total", "")),
("t20_evaluated_count", metrics.get("t20_evaluated_count", "")),
("t20_due_capture_count", metrics.get("t20_due_capture_count", "")),
("missing_due_date_count", metrics.get("missing_due_date_count", "")),
("all_proposals_have_due_dates", q.get("all_proposals_have_due_dates", "")),
("queue_count", len(queue)),
]
if todo:
rows.append(("todo_protocol", " / ".join(str(x) for x in todo)))
return _kv(rows)
def _outcome_eval_window_monitor(hctx: dict, se: list) -> str:
oqs = _sj(hctx.get("outcome_quality_score_v1_json", {}))
shom = _sj(hctx.get("short_horizon_outcome_monitor_v1_json", {}))
@@ -1318,6 +1532,7 @@ def main() -> int:
"final_execution_decision": lambda: _final_execution_decision(hctx, se),
"concise_hts_input_sheet": lambda: _concise_hts_input_sheet(hctx, se),
"watch_breakout_gate": lambda: _watch_breakout_gate(hctx, se),
"reference_price_ledger": lambda: _reference_price_ledger(hctx, se),
"single_conclusion": lambda: _single_conclusion(hctx, se),
"immediate_execution_playbook": lambda: _immediate_execution_playbook(hctx, se),
"market_context_learning_note": lambda: _market_context_learning_note(hctx, se),
@@ -1356,6 +1571,8 @@ def main() -> int:
"execution_quality_table": lambda: _execution_quality_table(hctx, se),
"sell_priority_decision_table": lambda: _sell_priority_decision_table(hctx, se),
"strategy_performance_scoreboard": lambda: _strategy_performance_scoreboard(hctx, se),
"performance_readiness_summary": lambda: _performance_readiness_summary(hctx, se),
"operational_eval_queue_summary": lambda: _operational_eval_queue_summary(hctx, se),
"outcome_eval_window_monitor": lambda: _outcome_eval_window_monitor(hctx, se),
"decision_trace_table": lambda: _decision_trace_table(hctx, se),
"anti_whipsaw_reentry_gate": lambda: _anti_whipsaw_reentry_gate(hctx, se),
@@ -1378,6 +1595,11 @@ def main() -> int:
md = render_fn()
except Exception as exc:
md = _err(se, name, f"렌더링 예외: {exc}")
if not str(md).lstrip().startswith(f"## {title}"):
body = str(md).lstrip()
if body.startswith("## "):
body = body.split("\n\n", 1)[1] if "\n\n" in body else ""
md = f"## {title}\n\n<!-- {name} -->\n\n{body}"
sections.append({"name": name, "title": title, "markdown": md})
# 섹션 처리 오류 요약을 마지막 섹션으로 추가
@@ -1391,6 +1613,8 @@ def main() -> int:
})
_section_names = {s.get("name", "") for s in sections}
expected_order = [name for name in SECTION_ORDER]
actual_order = [s.get("name", "") for s in sections if s.get("name", "") != "section_processing_errors"]
_eg = _sj(hctx.get("export_gate_json", {}))
_json_vs = _eg.get("json_validation_status", "PENDING_EXPORT") if isinstance(_eg, dict) else "PENDING_EXPORT"
report = {
@@ -1398,11 +1622,15 @@ def main() -> int:
"generated_at": datetime.now(timezone.utc).isoformat(),
"source_json": data_path.name,
"section_count": len(sections),
"section_error_count": len(se),
"section_errors": se,
"summary": {
"found_settlement": "final_execution_decision" in _section_names,
"found_heat": "watch_breakout_gate" in _section_names,
"found_routing": "routing_serving_trace_v2" in _section_names,
"found_qeh": "QEH_AUDIT_BLOCK" in _section_names,
"found_concise_hts_input_sheet": "concise_hts_input_sheet" in _section_names,
"found_reference_price_ledger": "reference_price_ledger" in _section_names,
"canonical_order_ok": actual_order[:len(expected_order)] == expected_order,
"found_outcome_eval_window": "outcome_eval_window_monitor" in _section_names,
"json_validation_status": _json_vs,
},
@@ -1416,9 +1644,7 @@ def main() -> int:
md_lines = ["# Operational Investment Report\n"]
for s in sections:
section_name = s.get("name", "")
section_title = s.get("title", section_name)
md_lines.append(f"## {section_name} - {section_title}\n\n{s.get('markdown', '')}\n")
md_lines.append(str(s.get("markdown", "")).rstrip() + "\n")
out_md.write_text("\n".join(md_lines), encoding="utf-8")
Path(args.improvement_harness_json).write_text(