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>
264 lines
11 KiB
Python
264 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DEFAULT_REPORT = ROOT / "Temp" / "operational_report.json"
|
|
DEFAULT_JSON = ROOT / "GatherTradingData.json"
|
|
DEFAULT_ENGINE_GATE = ROOT / "Temp" / "engine_harness_gate_result.json"
|
|
DEFAULT_OUT = ROOT / "Temp" / "strategy_hardening_harness_v1.json"
|
|
|
|
|
|
def _load(path: Path) -> dict[str, Any]:
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
obj = json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return {}
|
|
return obj if isinstance(obj, dict) else {}
|
|
|
|
|
|
def _sections_map(report: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
|
rows = report.get("sections") if isinstance(report.get("sections"), list) else []
|
|
out: dict[str, dict[str, Any]] = {}
|
|
for row in rows:
|
|
if not isinstance(row, dict):
|
|
continue
|
|
name = str(row.get("name") or "").strip()
|
|
if name:
|
|
out[name] = row
|
|
return out
|
|
|
|
|
|
def _has_section(sec: dict[str, dict[str, Any]], name: str) -> bool:
|
|
return name in sec
|
|
|
|
|
|
def _score_binary(flags: list[bool]) -> float:
|
|
if not flags:
|
|
return 0.0
|
|
return round((sum(1 for x in flags if x) / len(flags)) * 100.0, 2)
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--report", default=str(DEFAULT_REPORT))
|
|
ap.add_argument("--json", default=str(DEFAULT_JSON))
|
|
ap.add_argument("--engine-gate", default=str(DEFAULT_ENGINE_GATE))
|
|
ap.add_argument("--out", default=str(DEFAULT_OUT))
|
|
args = ap.parse_args()
|
|
|
|
rp = Path(args.report)
|
|
jp = Path(args.json)
|
|
ep = Path(args.engine_gate)
|
|
op = Path(args.out)
|
|
if not rp.is_absolute():
|
|
rp = ROOT / rp
|
|
if not jp.is_absolute():
|
|
jp = ROOT / jp
|
|
if not ep.is_absolute():
|
|
ep = ROOT / ep
|
|
if not op.is_absolute():
|
|
op = ROOT / op
|
|
|
|
report = _load(rp)
|
|
data_json = _load(jp)
|
|
engine_gate = _load(ep)
|
|
sec = _sections_map(report)
|
|
data = data_json.get("data") if isinstance(data_json.get("data"), dict) else {}
|
|
hctx = data.get("_harness_context") if isinstance(data.get("_harness_context"), dict) else {}
|
|
|
|
# Domain evidence
|
|
routing_serving_score = _score_binary([
|
|
_has_section(sec, "routing_serving_trace"),
|
|
_has_section(sec, "routing_serving_trace_v2"),
|
|
_has_section(sec, "routing_decision_explain_v1"),
|
|
_has_section(sec, "export_gate_diagnosis"),
|
|
])
|
|
decision_score = _score_binary([
|
|
_has_section(sec, "QEH_AUDIT_BLOCK"),
|
|
_has_section(sec, "today_decision_summary_card"),
|
|
_has_section(sec, "shadow_ledger_table"),
|
|
_has_section(sec, "llm_constraint_audit"),
|
|
])
|
|
fundamental_score = _score_binary([
|
|
_has_section(sec, "fundamental_quality_gate_v1"),
|
|
_has_section(sec, "fundamental_multifactor_v2"),
|
|
_has_section(sec, "earnings_growth_quality_v1"),
|
|
_has_section(sec, "market_share_proxy_v1"),
|
|
_has_section(sec, "cashflow_stability_v1"),
|
|
])
|
|
horizon_score = _score_binary([
|
|
_has_section(sec, "horizon_allocation_lock_v1"),
|
|
_has_section(sec, "t1_evaluation_summary_box"),
|
|
_has_section(sec, "benchmark_relative_harness_table"),
|
|
_has_section(sec, "index_relative_health_table"),
|
|
])
|
|
smart_money_score = _score_binary([
|
|
_has_section(sec, "smart_money_liquidity_gate_v1"),
|
|
_has_section(sec, "alpha_lead_table"),
|
|
_has_section(sec, "entry_freshness_gate_table"),
|
|
_has_section(sec, "anti_distribution_table"),
|
|
])
|
|
profit_preservation_score = _score_binary([
|
|
_has_section(sec, "profit_preservation_table"),
|
|
_has_section(sec, "sell_value_preservation_gate_table"),
|
|
_has_section(sec, "scrs_v2_sell_table"),
|
|
_has_section(sec, "mandatory_reduction_plan"),
|
|
])
|
|
cash_raise_score = _score_binary([
|
|
_has_section(sec, "cash_recovery_plan_crdl"),
|
|
_has_section(sec, "smart_cash_raise_table"),
|
|
_has_section(sec, "portfolio_structure_risks"),
|
|
])
|
|
|
|
outcome = _load(ROOT / "Temp" / "outcome_quality_score_v1.json")
|
|
exec_quality = _load(ROOT / "Temp" / "execution_quality_harness_v1.json")
|
|
eval_cov = _load(ROOT / "Temp" / "evaluation_history_coverage_v1.json")
|
|
data_integrity = _load(ROOT / "Temp" / "data_integrity_score_v1.json")
|
|
decision_evidence = _load(ROOT / "Temp" / "decision_evidence_score_v1.json")
|
|
derivation = _load(ROOT / "Temp" / "derivation_validity_score_v1.json")
|
|
algo_guidance = _load(ROOT / "Temp" / "algorithm_guidance_proof_v1.json")
|
|
|
|
# Performance domains are numeric, not just section existence
|
|
t20_pass_rate = float((outcome.get("metrics") or {}).get("t20_pass_rate") or 0.0)
|
|
outcome_score = float(outcome.get("score") or 0.0)
|
|
eq_oper = (exec_quality.get("metrics") or {}).get("operational_t20") if isinstance((exec_quality.get("metrics") or {}).get("operational_t20"), dict) else {}
|
|
execution_expectancy = float(eq_oper.get("expectancy_pct") or 0.0)
|
|
execution_mdd = float(eq_oper.get("max_drawdown_pct") or 0.0)
|
|
execution_win_rate = float(eq_oper.get("win_rate_pct") or 0.0)
|
|
data_integrity_score = float(data_integrity.get("score") or 0.0)
|
|
decision_evidence_score = float(decision_evidence.get("score") or 0.0)
|
|
derivation_score = float(derivation.get("score") or 0.0)
|
|
algo_guidance_score = float(algo_guidance.get("score") or 0.0)
|
|
|
|
# Engine hardening score focuses on deterministic control + measured outcome
|
|
control_score = round((routing_serving_score + decision_score + data_integrity_score + decision_evidence_score + derivation_score + algo_guidance_score) / 6.0, 2)
|
|
perf_subscores = [outcome_score, t20_pass_rate]
|
|
if exec_quality:
|
|
perf_subscores.append(max(0.0, min(100.0, 50.0 + execution_expectancy * 10.0)))
|
|
perf_subscores.append(max(0.0, min(100.0, 100.0 - execution_mdd * 3.0)))
|
|
perf_subscores.append(execution_win_rate)
|
|
performance_score = round(sum(perf_subscores) / max(1, len(perf_subscores)), 2)
|
|
overall = round(control_score * 0.6 + performance_score * 0.4, 2)
|
|
readiness_gate = "PERFORMANCE_READY"
|
|
if data_integrity_score < 100.0:
|
|
readiness_gate = "BLOCKED_DATA_QUALITY"
|
|
elif outcome_score < 60.0 or t20_pass_rate < 60.0 or str(exec_quality.get("gate") or "") in {"FAIL", "WATCH_PENDING_SAMPLE"}:
|
|
readiness_gate = "NOT_PERFORMANCE_READY"
|
|
|
|
# 100% target gap breakdown
|
|
gaps = {
|
|
"routing_serving_gap": round(100.0 - routing_serving_score, 2),
|
|
"decision_gap": round(100.0 - decision_score, 2),
|
|
"fundamental_gap": round(100.0 - fundamental_score, 2),
|
|
"horizon_gap": round(100.0 - horizon_score, 2),
|
|
"smart_money_gap": round(100.0 - smart_money_score, 2),
|
|
"profit_preservation_gap": round(100.0 - profit_preservation_score, 2),
|
|
"cash_raise_gap": round(100.0 - cash_raise_score, 2),
|
|
"outcome_quality_gap": round(100.0 - outcome_score, 2),
|
|
"t20_pass_rate_gap": round(100.0 - t20_pass_rate, 2),
|
|
"execution_expectancy_gap": round(max(0.0, 0.1 - execution_expectancy), 3),
|
|
"execution_mdd_over_gap": round(max(0.0, execution_mdd - 12.0), 2),
|
|
"execution_win_rate_gap": round(max(0.0, 45.0 - execution_win_rate), 2),
|
|
}
|
|
|
|
actions = []
|
|
if outcome_score < 60 or t20_pass_rate < 60:
|
|
actions.append({
|
|
"priority": "P0",
|
|
"action_id": "PERF_RECOVERY_HARNESS_V1",
|
|
"why": "매수/매도 뒷북·설거지 징후가 T20 패스율에 반영됨",
|
|
"required_metrics": ["t20_pass_rate", "watch_miss_rate", "late_chase_block_precision", "rebound_sell_value_damage"],
|
|
"target": {"t20_pass_rate_min": 60.0, "outcome_quality_min": 60.0},
|
|
})
|
|
if str(exec_quality.get("gate") or "") in {"FAIL", "WATCH_PENDING_SAMPLE"}:
|
|
actions.append({
|
|
"priority": "P0",
|
|
"action_id": "EXECUTION_QUALITY_HARNESS_V1",
|
|
"why": "기대값/낙폭/승률을 운영 표본으로 고정 추적해 뒷북·설거지 구조를 계량 통제",
|
|
"required_metrics": ["expectancy_pct", "max_drawdown_pct", "win_rate_pct", "samples"],
|
|
"target": {"expectancy_pct_min": 0.0, "max_drawdown_pct_max": 12.0, "win_rate_pct_min": 45.0},
|
|
})
|
|
if data_integrity_score < 100:
|
|
actions.append({
|
|
"priority": "P0",
|
|
"action_id": "DATA_INTEGRITY_100_LOCK_V1",
|
|
"why": "데이터 완성도 100 미달은 실전 판단 왜곡 유발",
|
|
"required_metrics": ["required_field_completeness_pct", "placeholder_safety_pct", "capture_age_hours"],
|
|
"target": {"data_integrity_score": 100.0},
|
|
})
|
|
if cash_raise_score < 100:
|
|
actions.append({
|
|
"priority": "P1",
|
|
"action_id": "CASH_RAISE_VALUE_DAMAGE_MIN_V1",
|
|
"why": "현금확보 과정에서 주식가치 훼손 최소화 필요",
|
|
"required_metrics": ["immediate_sell_ratio", "rebound_wait_ratio", "value_damage_pct_avg"],
|
|
"target": {"value_damage_pct_avg_max": 10.0},
|
|
})
|
|
if fundamental_score < 100:
|
|
actions.append({
|
|
"priority": "P1",
|
|
"action_id": "FUNDAMENTAL_FEATURE_COMPLETION_V1",
|
|
"why": "펀더멘털 결손은 중장기 판단 신뢰도 저하",
|
|
"required_metrics": ["roe_coverage_pct", "opm_coverage_pct", "ocf_coverage_pct", "fcf_coverage_pct"],
|
|
"target": {"feature_coverage_min": 95.0},
|
|
})
|
|
|
|
result = {
|
|
"formula_id": "STRATEGY_HARDENING_HARNESS_V1",
|
|
"source": {
|
|
"report_json": str(rp),
|
|
"data_json": str(jp),
|
|
"engine_gate_json": str(ep),
|
|
},
|
|
"engine_gate_status": engine_gate.get("status"),
|
|
"domain_scores": {
|
|
"routing_serving": routing_serving_score,
|
|
"decision_governance": decision_score,
|
|
"fundamental": fundamental_score,
|
|
"horizon_short_mid_long": horizon_score,
|
|
"smart_money_liquidity": smart_money_score,
|
|
"profit_preservation": profit_preservation_score,
|
|
"cash_raise_execution": cash_raise_score,
|
|
"outcome_quality": outcome_score,
|
|
"t20_pass_rate": t20_pass_rate,
|
|
"execution_expectancy_pct": execution_expectancy,
|
|
"execution_max_drawdown_pct": execution_mdd,
|
|
"execution_win_rate_pct": execution_win_rate,
|
|
"execution_quality_gate": exec_quality.get("gate"),
|
|
"data_integrity": data_integrity_score,
|
|
"decision_evidence": decision_evidence_score,
|
|
"derivation_validity": derivation_score,
|
|
"algorithm_guidance_proof": algo_guidance_score,
|
|
},
|
|
"meta_scores": {
|
|
"control_score": control_score,
|
|
"performance_score": performance_score,
|
|
"overall_hardening_score": overall,
|
|
"evaluation_history_gate": eval_cov.get("gate"),
|
|
"outcome_gate": outcome.get("gate"),
|
|
"readiness_gate": readiness_gate,
|
|
},
|
|
"gaps_to_100": gaps,
|
|
"hardening_actions": actions,
|
|
"determinism_lock": {
|
|
"llm_numeric_free_will_allowed": False,
|
|
"harness_context_keys": len(hctx.keys()) if isinstance(hctx, dict) else 0,
|
|
},
|
|
}
|
|
|
|
op.parent.mkdir(parents=True, exist_ok=True)
|
|
op.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|