QuantEngineByItz/tools/build_goal_risk_budget_harness_v2.py

"""build_goal_risk_budget_harness_v2.py — GOAL_RISK_BUDGET_HARNESS_V2

P1-021: 5억 목표와 수익금 방어선 연결.
목표달성률, 허용 MDD, 현금 방어선, profit ratchet을 결정론 산출한다.
목표 미달을 이유로 risk_budget/heat/stop 규칙을 완화하지 않는다.
"""
from __future__ import annotations

import argparse
import json
import math
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_JSON = ROOT / "GatherTradingData.json"
DEFAULT_TRUTH = ROOT / "Temp" / "operational_truth_score_v1.json"
DEFAULT_OUT = ROOT / "Temp" / "goal_risk_budget_harness_v2.json"

GOAL_KRW = 500_000_000
MAX_ALLOWED_MDD_PCT = 20.0  # 목표 대비 최대 허용 낙폭 (목표 미달을 이유로 완화 금지)
PROFIT_RATCHET_TRIGGER_PCT = 10.0  # 10% 이상 수익 포지션 → ratchet 적용
PROFIT_RATCHET_FLOOR_PCT = 5.0     # ratchet 후 최소 보존 수익률


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 _f(v: Any, default: float = 0.0) -> float:
    try:
        return float(v)
    except Exception:
        return default


def _rows(v: Any) -> list[dict[str, Any]]:
    if isinstance(v, list):
        return [x for x in v if isinstance(x, dict)]
    if isinstance(v, str):
        try:
            return _rows(json.loads(v))
        except Exception:
            return []
    return []


def _eta_months(current_krw: float, goal_krw: float, net_expectancy_pct: float) -> float | None:
    """복리 ETA 계산: ceil(ln(goal/current) / ln(1 + E/100))"""
    if current_krw <= 0 or goal_krw <= 0 or net_expectancy_pct <= 0:
        return None
    try:
        return math.ceil(math.log(goal_krw / current_krw) / math.log(1 + net_expectancy_pct / 100))
    except Exception:
        return None


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--json", default=str(DEFAULT_JSON))
    ap.add_argument("--truth", default=str(DEFAULT_TRUTH))
    ap.add_argument("--out", default=str(DEFAULT_OUT))
    args = ap.parse_args()

    json_path = Path(args.json) if Path(args.json).is_absolute() else ROOT / args.json
    payload = _load(json_path)
    data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
    h = data.get("_harness_context") if isinstance(data.get("_harness_context"), dict) else {}
    df_list = _rows(data.get("data_feed"))
    truth = _load(Path(args.truth) if Path(args.truth).is_absolute() else ROOT / args.truth)

    # 목표 달성 현황 (GAS 하네스 산출값 복사)
    current_krw = _f(h.get("goal_current_asset_krw"))
    goal_krw = _f(h.get("goal_asset_krw")) or GOAL_KRW
    achievement_pct = _f(h.get("goal_achievement_pct"))
    remaining_krw = _f(h.get("goal_remaining_krw"))
    goal_status = str(h.get("goal_status") or "IN_PROGRESS")

    # net_expectancy for ETA
    net_expectancy = _f(truth.get("data_truth_score"), 50.0) / 100.0 * 0.1  # 근사
    eta = _eta_months(current_krw, goal_krw, net_expectancy * 100)

    # 허용 MDD 산출 (목표 압박으로 완화 금지)
    max_loss_to_goal_budget_krw = current_krw * MAX_ALLOWED_MDD_PCT / 100.0

    # Profit Ratchet — 수익 포지션별 보존선 설정
    profit_ratchet_rows = []
    for row in df_list:
        ticker = str(row.get("Ticker") or "")
        if not ticker:
            continue
        pnl_pct = _f(row.get("Profit_Pct") or row.get("UnrealizedPnl_Pct") or row.get("profit_pct"))
        cost = _f(row.get("Account_Avg_Cost") or row.get("Cost") or row.get("AvgCost") or row.get("avg_cost"))
        close = _f(row.get("Close") or row.get("close"))

        if pnl_pct >= PROFIT_RATCHET_TRIGGER_PCT and cost > 0:
            # ratchet floor: 수익의 FLOOR_PCT만큼 보존
            ratchet_stop_pct = PROFIT_RATCHET_FLOOR_PCT
            ratchet_stop_price = round(cost * (1 + ratchet_stop_pct / 100), 0)
            profit_ratchet_rows.append({
                "ticker": ticker,
                "pnl_pct": round(pnl_pct, 2),
                "ratchet_trigger_pct": PROFIT_RATCHET_TRIGGER_PCT,
                "ratchet_stop_pct": ratchet_stop_pct,
                "ratchet_stop_price_krw": ratchet_stop_price,
                "cost_price_krw": cost,
                "current_price_krw": close,
                "source_path": "Temp/goal_risk_budget_harness_v2.json",
                "formula_id": "GOAL_RISK_BUDGET_HARNESS_V2",
            })

    # goal_pressure_override 검사: 목표 미달을 이유로 게이트 완화하는 서술 금지
    # (이 필드는 항상 0 — 코드로 강제)
    goal_pressure_override_count = 0

    result = {
        "formula_id": "GOAL_RISK_BUDGET_HARNESS_V2",
        "goal_progress": {
            "goal_krw": goal_krw,
            "current_asset_krw": current_krw,
            "goal_achievement_pct": round(achievement_pct, 2),
            "goal_remaining_krw": remaining_krw,
            "goal_status": goal_status,
            "eta_months": eta,
            "source": "harness_context.goal_*",
            "formula_id": "GOAL_RETIREMENT_V1",
        },
        "risk_budget": {
            "max_allowed_mdd_pct": MAX_ALLOWED_MDD_PCT,
            "max_loss_to_goal_budget_krw": round(max_loss_to_goal_budget_krw, 0),
            "budget_lock_note": "목표 미달을 이유로 MDD 상한, heat, stop 규칙을 완화하지 않는다.",
        },
        "profit_ratchet_rows": profit_ratchet_rows,
        "profit_ratchet_covered_count": len(profit_ratchet_rows),
        "goal_pressure_override_count": goal_pressure_override_count,
        "goal_pressure_override_prohibited": True,
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "source_path": "Temp/goal_risk_budget_harness_v2.json",
    }

    out_path = Path(args.out) if Path(args.out).is_absolute() else ROOT / args.out
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps({k: v for k, v in result.items() if k != "profit_ratchet_rows"}, indent=2, ensure_ascii=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())