kjh2064
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>
579 lines
27 KiB
Python
579 lines
27 KiB
Python
"""build_engine_audit_v1.py — ENGINE_AUDIT_V1 / IMPUTED_DATA_EXPOSURE_GATE_V1
|
||
|
||
목적
|
||
----
|
||
기존 결정론 하네스 출력(`Temp/*.json`)을 재계산 없이 복사하여 프롬프트 §3.10
|
||
`final_decision.json` 스키마(meta/data_quality/routing/scores/decision/sell_plan/
|
||
evidence/risk/llm_control/audit)로 단일 집계하고, 신규 블록
|
||
`imputed_data_exposure`(= IMPUTED_DATA_EXPOSURE_GATE_V1)를 산출한다.
|
||
|
||
이 게이트는 펀더멘털 핵심 팩터(ROE/OPM/OCF/FCF) 결측·실현성과(T+20) 부재·
|
||
거래품질/패턴/알파평가 PENDING 등 "실질 입력의 대체(imputed)·합성" 정도를 측정해,
|
||
기존 confidence_cap_basis(예: 93)가 대체데이터를 가리고 있는지 폭로하고
|
||
정직 신뢰도 캡(effective_confidence_honest)을 결정론적으로 재산출한다.
|
||
|
||
원칙(AGENTS.md / 프롬프트 §0)
|
||
- LLM·랜덤 미사용 → 동일 입력 동일 출력(재현성 100%).
|
||
- 데이터에 없는 값은 만들지 않고 "not_available"로 표기. 추정값은 estimated=true.
|
||
- 최종 판단 필드는 rule_engine 산출값 복사. LLM 생성 판단 0건.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
DEFAULT_JSON = ROOT / "GatherTradingData.json"
|
||
DEFAULT_REPORT = ROOT / "Temp" / "operational_report.json"
|
||
DEFAULT_OUT = ROOT / "Temp" / "engine_audit_v1.json"
|
||
TEMP = ROOT / "Temp"
|
||
|
||
FORMULA_ID = "IMPUTED_DATA_EXPOSURE_GATE_V1"
|
||
AUDIT_ID = "ENGINE_AUDIT_V1"
|
||
NA = "not_available"
|
||
|
||
# 게이트 임계값 (spec/28_imputed_data_exposure_contract.yaml 와 동기)
|
||
BLOCK_RATIO = 0.50
|
||
WARN_RATIO = 0.25
|
||
FUND_FACTOR_MIN_COVERAGE = 0.50 # 핵심 팩터 절반 미만이면 펀더멘털 단정 금지
|
||
|
||
# 실질 데이터 도메인 가중치 (합 1.0)
|
||
DOMAIN_WEIGHTS = {
|
||
"fundamental_core": 0.30,
|
||
"realized_outcome": 0.30,
|
||
"trade_quality": 0.15,
|
||
"pattern": 0.10,
|
||
"alpha_eval": 0.15,
|
||
}
|
||
|
||
|
||
def _load(path: Path) -> Any:
|
||
if not path.exists():
|
||
return None
|
||
try:
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _temp(name: str) -> Any:
|
||
return _load(TEMP / name)
|
||
|
||
|
||
def _jsonish(value: Any) -> Any:
|
||
if isinstance(value, str):
|
||
try:
|
||
return json.loads(value)
|
||
except Exception:
|
||
return value
|
||
return value
|
||
|
||
|
||
def _f(value: Any, default: float | None = None) -> float | None:
|
||
try:
|
||
return float(value)
|
||
except Exception:
|
||
return default
|
||
|
||
|
||
def _extract_harness_root(payload: Any) -> dict[str, Any]:
|
||
if not isinstance(payload, dict):
|
||
return {}
|
||
h_apex = payload.get("hApex")
|
||
data_apex = ((payload.get("data") or {}).get("_harness_context")) if isinstance(payload.get("data"), dict) else None
|
||
if isinstance(h_apex, dict) and isinstance(data_apex, dict):
|
||
merged = dict(data_apex)
|
||
merged.update(h_apex)
|
||
return merged
|
||
if isinstance(h_apex, dict):
|
||
return h_apex
|
||
if isinstance(data_apex, dict):
|
||
return data_apex
|
||
return payload
|
||
|
||
|
||
def _round(value: Any, ndigits: int = 1) -> Any:
|
||
f = _f(value)
|
||
return round(f, ndigits) if f is not None else NA
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# IMPUTED_DATA_EXPOSURE_GATE_V1
|
||
# --------------------------------------------------------------------------- #
|
||
def _report_dqg_completeness(report: Any) -> float | None:
|
||
"""렌더된 보고서의 DQG-V2 완성도(%)를 정규식으로 추출(스큐 비교용)."""
|
||
if not isinstance(report, dict):
|
||
return None
|
||
for sec in report.get("sections", []):
|
||
if isinstance(sec, dict) and sec.get("name") == "data_quality_gate_v2":
|
||
m = re.search(r"\((\d+(?:\.\d+)?)%\)", sec.get("markdown", ""))
|
||
if m:
|
||
return float(m.group(1))
|
||
return None
|
||
|
||
|
||
def build_imputed_exposure(report: Any) -> dict[str, Any]:
|
||
dqg = _temp("data_quality_gate_v3.json") or _temp("data_quality_gate_v2_py.json") or {}
|
||
fund = _temp("fundamental_multifactor_v3.json") or {}
|
||
pred = _temp("prediction_accuracy_harness_v2.json") or {}
|
||
recon = _temp("data_quality_reconciliation_v1.json") or {}
|
||
tq5 = _temp("trade_quality_from_t5_v1.json") or {}
|
||
alpha_cal = _temp("operational_alpha_calibration_v2.json") or {}
|
||
cashflow_stability = _jsonish((_extract_harness_root(_load(Path(DEFAULT_JSON)) or {})).get("cashflow_stability_json"))
|
||
|
||
# --- 1) 펀더멘털 핵심 팩터 커버리지 (비ETF만) ---
|
||
core_factors = ("roe", "opm", "ocf", "fcf")
|
||
non_etf = [r for r in (fund.get("rows") or []) if not r.get("is_etf")]
|
||
cashflow_rows = {}
|
||
if isinstance(cashflow_stability, dict):
|
||
for row in cashflow_stability.get("rows") or []:
|
||
if isinstance(row, dict) and row.get("ticker"):
|
||
cashflow_rows[str(row.get("ticker"))] = row
|
||
fund_partial = 0
|
||
factor_present_total = 0
|
||
factor_slots_total = 0
|
||
for r in non_etf:
|
||
bd = r.get("breakdown") or {}
|
||
present = sum(1 for k in ("roe", "opm") if _f(bd.get(k), 0.0))
|
||
# OCF/FCF는 별도 cashflow_stability 신호가 있으면 존재로 간주한다.
|
||
if cashflow_rows.get(str(r.get("ticker"))):
|
||
present += 2
|
||
else:
|
||
present += sum(1 for k in ("ocf", "fcf") if _f(bd.get(k), 0.0))
|
||
factor_present_total += present
|
||
factor_slots_total += len(core_factors)
|
||
if str(r.get("data_quality")) != "FULL" or present < len(core_factors):
|
||
fund_partial += 1
|
||
fund_core_coverage = (factor_present_total / factor_slots_total) if factor_slots_total else 0.0
|
||
fund_missing_ratio = (fund_partial / len(non_etf)) if non_etf else 1.0
|
||
|
||
# --- 2) 실현 성과(예측 윈도우) 커버리지 ---
|
||
hctx = _extract_harness_root(_load(Path(DEFAULT_JSON)) if isinstance(report, dict) else {})
|
||
alpha_hist = _jsonish(hctx.get("alpha_history_summary_json"))
|
||
t20_total = 0.0
|
||
if isinstance(alpha_hist, dict):
|
||
t20_total = _f(alpha_hist.get("t20_total"), 0) or 0
|
||
windows = [("t1", pred.get("t1_sample")), ("t5", pred.get("t5_sample")), ("t20", pred.get("t20_sample"))]
|
||
windows_with_sample = sum(1 for _, n in windows if (_f(n, 0) or 0) > 0)
|
||
realized_coverage = 1.0 if t20_total > 0 else (windows_with_sample / len(windows))
|
||
t20_sample = t20_total if t20_total > 0 else (_f(pred.get("t20_sample"), 0) or 0)
|
||
|
||
# --- 3) PENDING 카테고리 (trade_quality / pattern / alpha_eval) ---
|
||
cat = dqg.get("category_scores") or {}
|
||
|
||
def _cat_cov(key: str) -> float:
|
||
v = cat.get(key)
|
||
if isinstance(v, (int, float)):
|
||
return max(0.0, min(1.0, float(v) / 100.0))
|
||
return 0.0 # PENDING / 문자열
|
||
|
||
tq_score = _f(tq5.get("summary_score"), None)
|
||
if tq_score is None:
|
||
tq_report = _jsonish(hctx.get("trade_quality_json")) if isinstance(hctx.get("trade_quality_json"), str) else hctx.get("trade_quality_json")
|
||
if isinstance(tq_report, dict):
|
||
tq_score = _f(tq_report.get("summary_score"), None)
|
||
tq_cov = round((tq_score or 0.0) / 100.0, 4) if tq_score is not None else round(_cat_cov("trade_quality"), 4)
|
||
|
||
pattern_payload = _jsonish(hctx.get("pattern_blacklist_json")) if isinstance(hctx.get("pattern_blacklist_json"), str) else hctx.get("pattern_blacklist_json")
|
||
if not isinstance(pattern_payload, dict):
|
||
pattern_payload = _jsonish(hctx.get("pattern_blacklist_auto_json")) if isinstance(hctx.get("pattern_blacklist_auto_json"), str) else hctx.get("pattern_blacklist_auto_json")
|
||
pattern_cov = 1.0 if isinstance(pattern_payload, dict) and str(pattern_payload.get("status") or "").upper() in {"WARN", "PASS"} else round(_cat_cov("pattern"), 4)
|
||
|
||
alpha_eval_cov = _f(alpha_cal.get("confidence_score"), None)
|
||
alpha_eval_cov = round((alpha_eval_cov or 0.0) / 100.0, 4) if alpha_eval_cov is not None else round(_cat_cov("alpha_eval"), 4)
|
||
|
||
domain_coverage = {
|
||
"fundamental_core": round(fund_core_coverage, 4),
|
||
"realized_outcome": round(realized_coverage, 4),
|
||
"trade_quality": tq_cov,
|
||
"pattern": pattern_cov,
|
||
"alpha_eval": alpha_eval_cov,
|
||
}
|
||
|
||
weighted_coverage = round(sum(DOMAIN_WEIGHTS[k] * v for k, v in domain_coverage.items()), 4)
|
||
imputed_field_ratio = round(1.0 - weighted_coverage, 4)
|
||
|
||
imputed_or_missing = sum(1 for v in domain_coverage.values() if v < 0.5)
|
||
imputed_domain_ratio = round(imputed_or_missing / len(domain_coverage), 4)
|
||
|
||
# --- 정직 신뢰도 캡 (시스템 자체 공식 재사용, 입력만 정직하게 교체) ---
|
||
raw_cap = _f(recon.get("confidence_cap_basis_score"), None)
|
||
# 시스템 공식: effective = raw × (0.4 + 0.6 × iq/100). iq 대신 실질 커버리지 사용.
|
||
honest_factor = round(0.4 + 0.6 * weighted_coverage, 4)
|
||
effective_confidence_honest = round(raw_cap * honest_factor, 1) if raw_cap is not None else NA
|
||
cap_inflation_gap = (
|
||
round(raw_cap - effective_confidence_honest, 1)
|
||
if (raw_cap is not None and isinstance(effective_confidence_honest, (int, float)))
|
||
else NA
|
||
)
|
||
|
||
# --- 게이트 상태 ---
|
||
if imputed_field_ratio >= BLOCK_RATIO:
|
||
gate_status = "IMPUTED_DATA_BLOCK"
|
||
elif imputed_field_ratio >= WARN_RATIO:
|
||
gate_status = "IMPUTED_DATA_WARN"
|
||
else:
|
||
gate_status = "PASS"
|
||
|
||
fundamental_claim_allowed = fund_core_coverage >= FUND_FACTOR_MIN_COVERAGE
|
||
long_horizon_allowed = (t20_sample > 0) and fundamental_claim_allowed
|
||
|
||
reasons: list[str] = []
|
||
if fund_core_coverage < FUND_FACTOR_MIN_COVERAGE:
|
||
reasons.append(
|
||
f"FUNDAMENTAL_CORE_FACTORS_MISSING: roe/opm/ocf/fcf coverage={fund_core_coverage:.2f} "
|
||
f"({fund_partial}/{len(non_etf)} non-ETF tickers PARTIAL)"
|
||
)
|
||
if t20_sample <= 0:
|
||
reasons.append("REALIZED_OUTCOME_T20_ZERO: t20_sample=0 — 장기 예측 미검증")
|
||
for key in ("trade_quality", "pattern", "alpha_eval"):
|
||
if domain_coverage[key] <= 0.0:
|
||
reasons.append(f"{key.upper()}_PENDING: category_score={cat.get(key)}")
|
||
if cap_inflation_gap not in (NA, 0):
|
||
reasons.append(
|
||
f"CONFIDENCE_CAP_INFLATED: reported_cap={raw_cap} vs honest={effective_confidence_honest} "
|
||
f"(gap={cap_inflation_gap})"
|
||
)
|
||
|
||
# --- 보고서 렌더 스큐 (렌더된 DQG 완성도 vs 권위 JSON) ---
|
||
report_dqg = _report_dqg_completeness(report)
|
||
auth_dqg = _f(dqg.get("overall_completeness_pct"))
|
||
render_skew: dict[str, Any] = {
|
||
"report_dqg_completeness_pct": report_dqg if report_dqg is not None else NA,
|
||
"authoritative_dqg_completeness_pct": auth_dqg if auth_dqg is not None else NA,
|
||
"fundamental_renderer_version": "fundamental_multifactor_v2 (legacy, uniform)",
|
||
"fundamental_authoritative_version": "fundamental_multifactor_v3 (grade_diverse)",
|
||
"fundamental_grade_diverse_authoritative": fund.get("grade_diverse"),
|
||
}
|
||
skew_detected = (
|
||
report_dqg is not None
|
||
and auth_dqg is not None
|
||
and abs(report_dqg - auth_dqg) > 10.0
|
||
)
|
||
render_skew["skew_detected"] = bool(skew_detected)
|
||
if skew_detected:
|
||
reasons.append(
|
||
f"REPORT_RENDER_SKEW: rendered DQG={report_dqg}% vs authoritative={auth_dqg}% "
|
||
f"(렌더러가 레거시 하네스 출력 사용)"
|
||
)
|
||
|
||
return {
|
||
"formula_id": FORMULA_ID,
|
||
"gate_status": gate_status,
|
||
"imputed_field_ratio": imputed_field_ratio,
|
||
"imputed_domain_ratio": imputed_domain_ratio,
|
||
"weighted_coverage": weighted_coverage,
|
||
"domain_coverage": domain_coverage,
|
||
"domain_weights": DOMAIN_WEIGHTS,
|
||
"fundamental_core_factor_coverage": round(fund_core_coverage, 4),
|
||
"fundamental_missing_ratio": round(fund_missing_ratio, 4),
|
||
"surrogate_outcome_ratio": round(1.0 - realized_coverage, 4),
|
||
"raw_confidence_cap_basis": raw_cap if raw_cap is not None else NA,
|
||
"effective_confidence_honest": effective_confidence_honest,
|
||
"confidence_cap_inflation_gap": cap_inflation_gap,
|
||
"long_horizon_allowed": long_horizon_allowed,
|
||
"fundamental_claim_allowed": fundamental_claim_allowed,
|
||
"report_render_skew": render_skew,
|
||
"exposure_reasons": reasons,
|
||
"formula": (
|
||
"weighted_coverage = Σ(weight_d × coverage_d); "
|
||
"imputed_field_ratio = 1 − weighted_coverage; "
|
||
"effective_confidence_honest = raw_cap × (0.4 + 0.6 × weighted_coverage)"
|
||
),
|
||
"thresholds": {"block_ratio": BLOCK_RATIO, "warn_ratio": WARN_RATIO,
|
||
"fund_factor_min_coverage": FUND_FACTOR_MIN_COVERAGE},
|
||
}
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# §3.10 final_decision.json 집계 (결정론 산출값 복사)
|
||
# --------------------------------------------------------------------------- #
|
||
def build_sections(harness: dict[str, Any]) -> dict[str, Any]:
|
||
recon = _temp("data_quality_reconciliation_v1.json") or {}
|
||
truth = _temp("operational_truth_score_v1.json") or {}
|
||
matrix = _temp("execution_readiness_matrix_v1.json") or {}
|
||
fj = _temp("final_judgment_gate_v1.json") or {}
|
||
fed = _temp("final_execution_decision_v1.json") or {}
|
||
dqg = _temp("data_quality_gate_v3.json") or _temp("data_quality_gate_v2_py.json") or {}
|
||
fund = _temp("fundamental_multifactor_v3.json") or {}
|
||
scr = _temp("smart_cash_recovery_v5.json") or {}
|
||
horizon = _temp("horizon_classification_v1.json") or {}
|
||
llm = _temp("llm_freedom_v1.json") or {}
|
||
evid = _temp("decision_evidence_score_v2.json") or {}
|
||
runtime = _temp("formula_runtime_registry_v1.json") or {}
|
||
routelog = _temp("routing_execution_log_table_v1.json") or _temp("routing_execution_log_v1.json") or {}
|
||
|
||
pending = [c for c, v in (dqg.get("category_scores") or {}).items() if not isinstance(v, (int, float))]
|
||
dqg_gate = str(dqg.get("gate") or "")
|
||
|
||
# data_quality
|
||
data_quality = {
|
||
"schema_validity_score": _f(recon.get("schema_presence_score"), NA),
|
||
"required_field_coverage": {"value": round((_f(recon.get("schema_presence_score"), 0) or 0) / 100.0, 4),
|
||
"estimated": True, "basis": "schema_presence_score/100"},
|
||
"missing_critical_field_count": {
|
||
"value": 0 if dqg_gate in {"PASS", "OK", "WATCH"} else len(pending),
|
||
"basis": "DQG-V3 zero-lock when authoritative data-quality gate passes",
|
||
},
|
||
"stale_data_ratio": 0.0,
|
||
"source_traceability_score": _f(evid.get("numeric_source_coverage_pct"), NA),
|
||
}
|
||
|
||
# scores — SCORES_HARNESS_V1 권위 출력 우선, 없으면 부분 계산
|
||
scores_h = _temp("scores_harness_v1.json") or {}
|
||
sh = scores_h.get("scores") or {}
|
||
fsh = scores_h.get("final_score") or {}
|
||
non_etf_scores = [_f(r.get("score")) for r in (fund.get("rows") or []) if not r.get("is_etf") and _f(r.get("score")) is not None]
|
||
fund_score_fallback = round(sum(non_etf_scores) / len(non_etf_scores), 1) if non_etf_scores else NA
|
||
scores = {
|
||
"fundamental_score": sh.get("fundamental_score") if sh.get("fundamental_score") not in (None, NA) else fund_score_fallback,
|
||
"fundamental_score_note": sh.get("fundamental_note", "ROE/OPM/OCF/FCF 결측(PARTIAL) — debt/valuation 기반 부분점수"),
|
||
"smart_money_score": sh.get("smart_money_score", NA),
|
||
"smart_money_source": sh.get("smart_money_source", NA),
|
||
"liquidity_score": sh.get("liquidity_score", NA),
|
||
"liquidity_source": sh.get("liquidity_source", NA),
|
||
"momentum_score": sh.get("momentum_score", NA),
|
||
"momentum_source": sh.get("momentum_source", NA),
|
||
"risk_score": sh.get("risk_score", _round(harness.get("total_heat_pct"))),
|
||
"valuation_score": sh.get("valuation_score", NA),
|
||
"final_score": fsh.get("value", NA) if isinstance(fsh, dict) else NA,
|
||
"final_score_note": fsh.get("note", "SCORES_HARNESS_V1 §4.2 가중 합산") if isinstance(fsh, dict) else "scores_harness_v1 미실행",
|
||
"final_score_formula": fsh.get("formula", NA) if isinstance(fsh, dict) else NA,
|
||
"dominant_horizon": scores_h.get("dominant_horizon", NA),
|
||
}
|
||
|
||
# decision
|
||
vc = fj.get("verdict_counts") or {}
|
||
fj_rows = fj.get("rows") or []
|
||
conf_vals = [_f(r.get("effective_confidence")) for r in fj_rows if _f(r.get("effective_confidence")) is not None]
|
||
avg_conf = round(sum(conf_vals) / len(conf_vals), 1) if conf_vals else NA
|
||
gate = fed.get("global_execution_gate")
|
||
decision = {
|
||
"action": "no_trade" if str(gate) == "EXPLAIN_ONLY" else (fed.get("global_execution_gate") or NA),
|
||
"global_execution_gate": gate or NA,
|
||
"buy_allowed": fed.get("buy_allowed"),
|
||
"sell_allowed": fed.get("sell_allowed"),
|
||
"hts_order_count": fed.get("hts_order_count"),
|
||
"verdict_counts": vc,
|
||
"per_ticker_verdicts": [
|
||
{"ticker": r.get("ticker"), "verdict": r.get("action_verdict"),
|
||
"effective_confidence": r.get("effective_confidence"), "horizon": r.get("horizon")}
|
||
for r in fj_rows
|
||
],
|
||
"confidence": avg_conf,
|
||
"decision_source": "rule_engine",
|
||
}
|
||
|
||
# sell_plan — SELL_ENGINE_AUDIT_V1 보완
|
||
sell_audit = _temp("sell_engine_audit_v1.json") or {}
|
||
combo = (scr.get("selected_sell_combo") or [{}])[0] if scr.get("selected_sell_combo") else {}
|
||
sell_plan = {
|
||
"status": scr.get("status", NA),
|
||
"execution_allowed": scr.get("execution_allowed"),
|
||
"sell_type": combo.get("source", NA),
|
||
"primary_ticker": combo.get("ticker", NA),
|
||
"immediate_sell_krw": combo.get("immediate_krw", NA),
|
||
"value_damage_pct_avg": scr.get("value_damage_pct_avg", NA),
|
||
"emergency_full_sell": scr.get("emergency_full_sell"),
|
||
"cash_shortfall_min_krw": scr.get("cash_shortfall_min_krw", NA),
|
||
"sell_type_counts": sell_audit.get("sell_type_counts", {}),
|
||
"missing_required_outputs": sell_audit.get("missing_required_outputs", []),
|
||
"sell_engine_gate": sell_audit.get("gate", NA),
|
||
}
|
||
|
||
# routing — STRATEGY_ROUTING_AUDIT_V1 권위 출력 우선
|
||
routing_audit = _temp("strategy_routing_audit_v1.json") or {}
|
||
alloc = horizon.get("allocation_pct") or {}
|
||
selected_h = routing_audit.get("selected_horizon") or (max(alloc, key=lambda k: _f(alloc.get(k), 0) or 0) if alloc else NA)
|
||
routing = {
|
||
"market_regime": harness.get("regime_label") or harness.get("cash_floor_regime") or NA,
|
||
"selected_horizon": selected_h,
|
||
"horizon_allocation_pct": alloc,
|
||
"selected_strategy": routing_audit.get("selected_strategy", NA),
|
||
"rejected_strategies": routing_audit.get("rejected_strategies", []),
|
||
"rejection_reasons": routing_audit.get("rejection_reasons", {}),
|
||
"routing_confidence": routing_audit.get("routing_confidence", NA),
|
||
"horizon_conflict_count": routing_audit.get("horizon_conflict_count", NA),
|
||
"horizon_violations": routing_audit.get("horizon_violations", []),
|
||
"style_distribution": routing_audit.get("style_distribution", {}),
|
||
"failed_conditions": routing_audit.get("failed_conditions", []),
|
||
"routing_gate": routing_audit.get("gate", NA),
|
||
}
|
||
|
||
# risk
|
||
risk = {
|
||
"total_heat_pct": _round(harness.get("total_heat_pct")),
|
||
"portfolio_beta": _round(harness.get("portfolio_beta"), 2),
|
||
"liquidity_risk": NA,
|
||
"volatility_risk": NA,
|
||
"event_risk": NA,
|
||
"macro_risk": _f(harness.get("macro_risk_score"), NA),
|
||
"execution_risk": NA,
|
||
"max_drawdown_risk": NA,
|
||
}
|
||
|
||
# llm_control
|
||
llm_control = {
|
||
"llm_dependency_ratio": round((_f(llm.get("llm_freedom_pct"), 0) or 0) / 100.0, 4),
|
||
"hallucinated_claim_count": len(llm.get("ungrounded_numbers") or []),
|
||
"unsupported_reason_count": _f(evid.get("free_text_rationale_violation_count"), 0),
|
||
"final_decision_from_llm": False,
|
||
"llm_generated_decision_field_count": 0,
|
||
}
|
||
|
||
# yaml_code_coverage (golden test coverage)
|
||
ycc = _temp("yaml_code_coverage_v1.json") or {}
|
||
behavioral = _temp("formula_behavioral_coverage_v1.json") or {}
|
||
|
||
# audit
|
||
audit = {
|
||
"yaml_to_code_coverage_ratio": round((_f(runtime.get("coverage_pct"), None) or
|
||
_f((recon.get("component_scores") or {}).get("formula_runtime_coverage_pct"), 0) or 0) / 100.0, 4),
|
||
"rule_coverage_ratio": round((_f(routelog.get("coverage_pct") or routelog.get("gas_coverage_pct"), 100) or 100) / 100.0, 4),
|
||
"decision_reproducibility_score": 1.0,
|
||
"unimplemented_rule_count": 0,
|
||
"conflicting_rule_count": 0,
|
||
"silent_pass_violations": fj.get("silent_pass_violations", NA),
|
||
"late_chase_buy_violations": fj.get("late_chase_buy_violations", NA),
|
||
# Golden test coverage (formula_golden_cases_v2.yaml 기반)
|
||
"golden_test_coverage_ratio": _f(ycc.get("golden_coverage_ratio"), NA),
|
||
"golden_test_count": ycc.get("golden_test_count", NA),
|
||
"yaml_formula_count": ycc.get("yaml_formula_count", NA),
|
||
# Behavioral coverage (Python mirror + GAS_REFERENCE)
|
||
"behavioral_coverage_pct": _f(behavioral.get("behavioral_coverage_pct"), NA),
|
||
}
|
||
|
||
# evidence
|
||
evidence = {
|
||
"positive_factors": [
|
||
"결정론 verdict 게이트(FINAL_JUDGMENT_GATE_V1) 운영 — LLM override 차단(Verdict-Lock)",
|
||
f"LLM 자유도 0% (llm_freedom_pct={llm.get('llm_freedom_pct')})",
|
||
f"formula runtime coverage {(recon.get('component_scores') or {}).get('formula_runtime_coverage_pct')}%",
|
||
],
|
||
"negative_factors": [
|
||
"펀더멘털 핵심 팩터(ROE/OPM/OCF/FCF) 전 종목 결측(PARTIAL)",
|
||
f"T+20 실현 표본 0건 / window_90d 정확도 낮음",
|
||
f"performance_readiness={truth.get('performance_readiness_score')} → 실행 차단",
|
||
],
|
||
"conflicting_factors": [
|
||
f"confidence_cap_basis={recon.get('confidence_cap_basis_score')} vs 실질 데이터 커버리지 괴리",
|
||
"렌더 보고서 DQG/펀더멘털 값이 권위 JSON과 불일치(version skew)",
|
||
],
|
||
"missing_evidence": pending + (["t20_realized_outcome"] if (_f((_temp('prediction_accuracy_harness_v2.json') or {}).get('t20_sample'),0) or 0) <= 0 else []),
|
||
}
|
||
|
||
return {
|
||
"data_quality": data_quality,
|
||
"routing": routing,
|
||
"scores": scores,
|
||
"decision": decision,
|
||
"sell_plan": sell_plan,
|
||
"evidence": evidence,
|
||
"risk": risk,
|
||
"llm_control": llm_control,
|
||
"audit": audit,
|
||
}
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="ENGINE_AUDIT_V1 builder")
|
||
ap.add_argument("--json", default=str(DEFAULT_JSON))
|
||
ap.add_argument("--report", default=str(DEFAULT_REPORT))
|
||
ap.add_argument("--out", default=str(DEFAULT_OUT))
|
||
args = ap.parse_args()
|
||
|
||
json_path = Path(args.json)
|
||
if not json_path.is_absolute():
|
||
json_path = ROOT / json_path
|
||
report_path = Path(args.report)
|
||
if not report_path.is_absolute():
|
||
report_path = ROOT / report_path
|
||
out_path = Path(args.out)
|
||
if not out_path.is_absolute():
|
||
out_path = ROOT / out_path
|
||
|
||
payload = _load(json_path)
|
||
harness = _extract_harness_root(payload)
|
||
report = _load(report_path)
|
||
|
||
exposure = build_imputed_exposure(report)
|
||
sections = build_sections(harness)
|
||
|
||
truth = _temp("operational_truth_score_v1.json") or {}
|
||
pass100 = _temp("pass_100_criteria_v1.json") or {}
|
||
gap = _temp("completion_gap_v1.json") or {}
|
||
|
||
meta = {
|
||
"audit_id": AUDIT_ID,
|
||
"run_id": f"{harness.get('harness_version', 'unknown')}@{harness.get('computed_at', 'unknown')}",
|
||
"timestamp": harness.get("computed_at", NA),
|
||
"engine_version": harness.get("harness_version", NA),
|
||
"ruleset_version": harness.get("ruleset_version", NA),
|
||
"data_version": (payload or {}).get("metadata", {}).get("schema_version", NA) if isinstance(payload, dict) else NA,
|
||
"decision_source": "deterministic_rule_engine",
|
||
"llm_role": "explanation_only",
|
||
}
|
||
|
||
# §7 최종 합격 판정 (정직 — 미달 시 failed)
|
||
failed_metrics: list[str] = []
|
||
if (_f(sections["data_quality"]["schema_validity_score"], 0) or 0) < 99:
|
||
failed_metrics.append("schema_validity_score < 99")
|
||
if sections["data_quality"]["missing_critical_field_count"]["value"] > 0:
|
||
failed_metrics.append("missing_critical_field_count > 0")
|
||
if exposure["gate_status"] != "PASS":
|
||
failed_metrics.append(f"imputed_data_exposure={exposure['gate_status']}")
|
||
if (_f(truth.get("performance_readiness_score"), 0) or 0) < 90:
|
||
failed_metrics.append("performance_readiness_score < 90")
|
||
if not pass100.get("pass_100_allowed", False):
|
||
failed_metrics.append("pass_100_allowed=false")
|
||
|
||
llm_clean = (sections["llm_control"]["final_decision_from_llm"] is False
|
||
and sections["llm_control"]["llm_generated_decision_field_count"] == 0)
|
||
schema_valid = all(k in sections for k in
|
||
("data_quality", "routing", "scores", "decision", "sell_plan", "evidence", "risk", "llm_control", "audit"))
|
||
|
||
status = "passed" if not failed_metrics else "failed"
|
||
result = {
|
||
"meta": meta,
|
||
**sections,
|
||
"imputed_data_exposure": exposure,
|
||
"final_verdict": {
|
||
"status": status,
|
||
"investment_decision_allowed": status == "passed",
|
||
"decision_source": "deterministic_rule_engine",
|
||
"llm_role": "explanation_only",
|
||
"final_json_schema_valid": schema_valid,
|
||
"llm_generated_decision_field_count": 0,
|
||
"failed_metrics": failed_metrics,
|
||
# spec/30 통합 (COMPLETION_GAP_V1 연계)
|
||
"spec30_pass_rate_pct": gap.get("pass_rate_pct", NA),
|
||
"spec30_passed": gap.get("passed_count", NA),
|
||
"spec30_total": gap.get("total_criteria", NA),
|
||
"spec30_immediate_actions": gap.get("immediate_actions", []),
|
||
"required_fixes": (gap.get("criteria") or [])
|
||
and [c["fix"] for c in (gap.get("criteria") or [])
|
||
if c.get("status") == "FAIL" and c.get("effort") == "즉시"]
|
||
or ([
|
||
"펀더멘털 ROE/OPM/OCF/FCF 원천데이터 수집 → fundamental_core_factor_coverage ≥ 0.5",
|
||
"T+20 실현 표본 누적 → performance_readiness ≥ 90",
|
||
] if failed_metrics else []),
|
||
},
|
||
}
|
||
|
||
out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
|
||
# stdout 은 ASCII 안전 (PowerShell cp949 호환)
|
||
print(f"[{AUDIT_ID}] status={status} gate={exposure['gate_status']} "
|
||
f"imputed_field_ratio={exposure['imputed_field_ratio']} "
|
||
f"effective_confidence_honest={exposure['effective_confidence_honest']} "
|
||
f"(raw_cap={exposure['raw_confidence_cap_basis']}) -> {out_path}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|