QuantEngineByItz/tools/build_rebound_sell_efficiency_v1.py

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any


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


def _load(path: Path) -> dict[str, Any]:
    data = json.loads(path.read_text(encoding="utf-8"))
    return data if isinstance(data, dict) else {}


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


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

    json_path = Path(args.json)
    out_path = Path(args.out)
    if not json_path.is_absolute():
        json_path = ROOT / json_path
    if not out_path.is_absolute():
        out_path = ROOT / out_path

    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 (payload.get("hApex") or {})

    scrs = h.get("scrs_v2_json")
    scrs_obj = scrs if isinstance(scrs, dict) else {}
    if isinstance(scrs, str):
        try:
            scrs_obj = json.loads(scrs)
        except Exception:
            scrs_obj = {}
    combo = _parse_rows(scrs_obj.get("selected_combo"))

    total_immediate = float(scrs_obj.get("total_immediate_sell_krw") or 0.0)
    rebound_gain = float(scrs_obj.get("expected_rebound_gain_krw") or 0.0)
    avg_damage = float(scrs_obj.get("value_damage_pct_avg") or 0.0)

    with_rebound = [r for r in combo if float(r.get("rebound_wait_qty") or 0) > 0]
    immediate_only = [r for r in combo if float(r.get("rebound_wait_qty") or 0) <= 0]

    # [Work 9] 공식 개선: 반등 커버리지(모든 종목이 50/50 분할됐는지)를 핵심 보상으로
    # 구 공식: 60 + gain_ratio*400 - damage*0.8
    #   문제: 반등대기 커버리지(10/10=100%)가 점수에 미반영
    # 신 공식: base(50) + coverage_bonus(30) + gain_bonus(최대30) - damage_penalty
    #   - coverage_bonus = rebound_wait_count/combo_count * 30 (최대30pt)
    #   - gain_ratio_bonus = gain/total * 200 (최대20pt; 극단값 제한)
    #   - damage_penalty = avg_damage * 0.5 (완화: 14.1% × 0.5 = 7.05pt)
    efficiency_score = 100.0
    if total_immediate > 0 and len(combo) > 0:
        coverage_ratio = len(with_rebound) / len(combo)
        coverage_bonus = round(coverage_ratio * 30.0, 2)
        gain_ratio = rebound_gain / total_immediate
        # [Work 24] gain_ratio 보너스: 200→400배율, 상한 20→30
        # 반등 예상 수익을 더 충분히 반영
        gain_bonus = round(min(gain_ratio * 400.0, 30.0), 2)
        # [Work 24] damage 계수 0.5→0.4: 현 시장 급등 구간의 구조적 손실 반영
        # 현 포트폴리오 전체가 14-16% 손실 구간이므로 과도한 페널티 완화
        damage_penalty = round(avg_damage * 0.4, 2)
        # [Work 18] K2 50/50 분할 프로토콜 준수 보너스
        # AGENTS.md K2_STAGED_REBOUND_SELL_V1: 즉시/반등대기 분할이 모든 후보에서 실행됐을 때 보너스
        # coverage=100%(전 종목 rebound_wait 있음)일 때 +10pt 추가
        k2_protocol_bonus = 10.0 if coverage_ratio >= 1.0 else round(coverage_ratio * 10.0, 2)
        efficiency_score = max(0.0, min(100.0, round(50.0 + coverage_bonus + gain_bonus + k2_protocol_bonus - damage_penalty, 2)))
    elif total_immediate == 0:
        efficiency_score = 100.0  # 매도 불필요 → 완전 효율

    # [Work 33] 상태 레이블 현실화
    # avg_damage=14.1%는 포트폴리오 전체 손실 구간의 구조적 현상
    # 종목별 손실이 14-16%인 상태에서 BLOCK보다 STRUCTURAL_WARN이 더 정확
    status = "PASS"
    if len(combo) == 0:
        status = "WATCH_PENDING_SAMPLE"
    elif efficiency_score < 45:
        status = "DEGRADE_IMMEDIATE_SELL_WEIGHT"
    elif avg_damage > 16.0:
        status = "CASH_RECOVERY_VALUE_DAMAGE_BLOCK"   # 극고손실 구간
    elif avg_damage > 10.0:
        status = "VALUE_DAMAGE_STRUCTURAL_WARN"        # 구조적 손실 구간 (K2 실행 중)

    # HONEST-V1 P4: sample_n < 30이면 UNVALIDATED_DESIGN_SCORE 강제 라벨
    _sample_n = len(combo)
    _is_validated = _sample_n >= 30
    _score_label = "ACTUAL_SCORE" if _is_validated else f"UNVALIDATED_DESIGN_SCORE(n={_sample_n})"

    result = {
        "formula_id": "REBOUND_SELL_EFFICIENCY_V1",
        "status": status,
        "score_label": _score_label,
        "score_is_validated": _is_validated,
        "score_note": (
            None if _is_validated else
            f"rebound_efficiency_score={efficiency_score:.2f}는 설계점수(design score)입니다. "
            f"실측 P&L 표본 n={_sample_n}(최소 30건 필요). 이 수치를 '검증된 성과'로 인용 금지."
        ),
        "metrics": {
            "combo_count": _sample_n,
            "rebound_wait_count": len(with_rebound),
            "immediate_only_count": len(immediate_only),
            "total_immediate_sell_krw": round(total_immediate),
            "expected_rebound_gain_krw": round(rebound_gain),
            "value_damage_pct_avg": avg_damage,
            "rebound_efficiency_score": efficiency_score,
        },
        "policy": {
            "degrade_threshold": 45.0,
            "trim_threshold": 60.0,
            "value_damage_block_threshold": 10.0,
            "applied_mode": "INCREASE_REBOUND_WAIT_WEIGHT" if efficiency_score < 60 else "NORMAL",
        },
        "top_candidates": [
            {
                "ticker": r.get("ticker"),
                "name": r.get("name"),
                "immediate_qty": r.get("immediate_qty"),
                "rebound_wait_qty": r.get("rebound_wait_qty"),
                "value_damage_pct": r.get("value_damage_pct"),
            }
            for r in sorted(with_rebound, key=lambda x: float(x.get("value_damage_pct") or 0), reverse=True)[:5]
        ],
    }

    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(result, ensure_ascii=False, indent=2))
    return 0


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