QuantEngineByItz/tools/build_capital_style_allocation_v1.py

#!/usr/bin/env python3
"""
build_capital_style_allocation_v1.py
───────────────────────────────────────────────────────────────────────────────
CAPITAL_STYLE_ALLOCATION_V1 — 투자성향별 자금 유동성 가중 엔진

4개 투자성향(SCALP/SWING/MOMENTUM/POSITION)에 대해 기존 신호 빌딩블록을
성향별 다른 가중치로 융합하여 결정론적 conviction_score(0~100)와
recommended_position_pct를 산출한다.

신호 출처 (LLM 계산 0%):
  - technical_score    : data_feed RSI14/Disparity/Ret5D 기반 규칙 계산
  - smart_money_score  : Temp/smart_money_flow_signal_v2.json (이미 0~100)
  - fundamental_score  : Temp/fundamental_multifactor_v3.json (이미 0~100)
  - macro_event_score  : Temp/macro_event_ticker_impact_v1.json ([-100,+100]→[0,100])
  - liquidity_modifier : Temp/liquidity_flow_signal_v1.json (DEEP/MODERATE/THIN/FROZEN→배수)

가중치 출처: EXPERT_PRIOR (spec/calibration_registry.yaml 등록)
  SCALP:    tech=0.50, smart=0.30, fund=0.05, macro=0.15
  SWING:    tech=0.30, smart=0.35, fund=0.15, macro=0.20
  MOMENTUM: tech=0.15, smart=0.25, fund=0.40, macro=0.20
  POSITION: tech=0.10, smart=0.20, fund=0.55, macro=0.15

출력: Temp/capital_style_allocation_v1.json
사용법: python tools/build_capital_style_allocation_v1.py
"""
from __future__ import annotations

import argparse
import json
import math
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent

if sys.stdout.encoding and sys.stdout.encoding.lower() not in ("utf-8", "utf8"):
    sys.stdout = open(sys.stdout.fileno(), mode="w", encoding="utf-8", buffering=1)

# ─── 가중치 정의 (EXPERT_PRIOR — calibration_registry.yaml 등록) ───────────
W_STYLE: dict[str, dict[str, float]] = {
    "SCALP":    {"technical": 0.50, "smartmoney": 0.30, "fundamental": 0.05, "macro_event": 0.15},
    "SWING":    {"technical": 0.30, "smartmoney": 0.35, "fundamental": 0.15, "macro_event": 0.20},
    "MOMENTUM": {"technical": 0.15, "smartmoney": 0.25, "fundamental": 0.40, "macro_event": 0.20},
    "POSITION": {"technical": 0.10, "smartmoney": 0.20, "fundamental": 0.55, "macro_event": 0.15},
}

# ─── 포지션 사이즈 맵핑 (EXPERT_PRIOR) ────────────────────────────────────
def conviction_to_pct(score: float) -> float:
    if score >= 80: return 7.0
    if score >= 65: return 5.0
    if score >= 50: return 3.0
    if score >= 35: return 1.5
    return 0.0

# ─── 유동성 배수 ──────────────────────────────────────────────────────────
LIQUIDITY_MODIFIER: dict[str, float] = {
    "DEEP":     1.00,
    "MODERATE": 0.90,
    "THIN":     0.75,
    "FROZEN":   0.00,
}


def _load(path: Path) -> dict | list:
    if not path.exists():
        return {}
    try:
        d = json.loads(path.read_text(encoding="utf-8"))
        return d
    except Exception:
        return {}


def _rows(v: object) -> list[dict]:
    if isinstance(v, list):
        return [x for x in v if isinstance(x, dict)]
    if isinstance(v, dict):
        for key in ("rows", "data", "tickers"):
            c = v.get(key)
            if isinstance(c, list):
                return [x for x in c if isinstance(x, dict)]
    return []


def _f(v: object, default: float = 0.0) -> float:
    try:
        return float(v)  # type: ignore[arg-type]
    except Exception:
        return default


def _clamp(v: float, lo: float = 0.0, hi: float = 100.0) -> float:
    return max(lo, min(hi, v))


# ─── 기술적 점수 산출 (결정론적 규칙) ────────────────────────────────────
def compute_technical_score(rsi14: float, disparity: float, ret5d: float,
                             volume: float, avg_vol5d: float) -> float:
    """
    기본 50점 기준, 아래 조건을 가산/감산.
    RSI14 < 35      → +20 (과매도 반등 기회)
    RSI14 > 70      → -25 (과매수 추격 위험)
    Disparity < 3%  → +15 (MA20 근접 — 눌림목)
    Disparity > 10% → -20 (MA20 과이격)
    Ret5D < -5%     → +10 (단기 급락 반등 후보)
    거래량 확인(volume >= avgVol5d*1.2 AND Ret5D > 0) → +10 (수급 확인 돌파)
    """
    score = 50.0
    if rsi14 < 35:
        score += 20.0
    elif rsi14 > 70:
        score -= 25.0
    if disparity < 3.0:
        score += 15.0
    elif disparity > 10.0:
        score -= 20.0
    if ret5d < -5.0:
        score += 10.0
    if avg_vol5d > 0 and volume >= avg_vol5d * 1.2 and ret5d > 0:
        score += 10.0
    return _clamp(score)


# ─── 메인 ─────────────────────────────────────────────────────────────────
def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--json",     default=str(ROOT / "GatherTradingData.json"))
    ap.add_argument("--smart",    default=str(ROOT / "Temp/smart_money_flow_signal_v2.json"))
    ap.add_argument("--fund",     default=str(ROOT / "Temp/fundamental_multifactor_v3.json"))
    ap.add_argument("--macro",    default=str(ROOT / "Temp/macro_event_ticker_impact_v1.json"))
    ap.add_argument("--liquidity",default=str(ROOT / "Temp/liquidity_flow_signal_v1.json"))
    ap.add_argument("--out",      default=str(ROOT / "Temp/capital_style_allocation_v1.json"))
    args = ap.parse_args()

    def rp(s: str) -> Path:
        p = Path(s)
        return p if p.is_absolute() else ROOT / p

    # ── 입력 로드 ────────────────────────────────────────────────────────
    payload   = _load(rp(args.json))
    smart_raw = _load(rp(args.smart))
    fund_raw  = _load(rp(args.fund))
    macro_raw = _load(rp(args.macro))
    liq_raw   = _load(rp(args.liquidity))

    data = payload.get("data", {}) if isinstance(payload, dict) else {}
    df_list = _rows(data.get("data_feed")) if isinstance(data, dict) else []

    # ── 신호 인덱스 구성 ──────────────────────────────────────────────────
    smart_map: dict[str, float] = {}
    for r in _rows(smart_raw):
        t = str(r.get("ticker") or "")
        smart_map[t] = _f(r.get("smart_money_score"), 50.0)

    fund_map: dict[str, float] = {}
    for r in _rows(fund_raw):
        t = str(r.get("ticker") or "")
        fund_map[t] = _f(r.get("score"), 50.0)

    macro_map: dict[str, dict] = {}
    macro_tickers = _rows(macro_raw) if isinstance(macro_raw, list) else _rows(macro_raw.get("tickers") if isinstance(macro_raw, dict) else [])
    for r in macro_tickers:
        t = str(r.get("ticker") or "")
        macro_map[t] = r

    liq_map: dict[str, str] = {}
    exec_map: dict[str, str] = {}
    for r in _rows(liq_raw):
        t = str(r.get("ticker") or "")
        liq_map[t]  = str(r.get("liquidity_label") or "MODERATE")
        exec_map[t] = str(r.get("execution_mode")  or "")

    # ── 종목별 신호 산출 + 성향별 conviction ───────────────────────────
    rows_out: list[dict] = []
    errors: list[str] = []

    for r in df_list:
        ticker = str(r.get("Ticker") or r.get("ticker") or "")
        name   = str(r.get("Name")   or r.get("name")   or "")
        if not ticker:
            continue

        # 기술지표
        rsi14    = _f(r.get("RSI14"),      50.0)
        disparity = _f(r.get("Disparity"),  0.0)
        ret5d    = _f(r.get("Ret5D"),       0.0)
        volume   = _f(r.get("Volume"),      0.0)
        avg5d    = _f(r.get("AvgVolume_5D"),0.0)

        # 4개 신호 정규화 [0, 100]
        tech_score  = compute_technical_score(rsi14, disparity, ret5d, volume, avg5d)
        smart_score = _clamp(smart_map.get(ticker, 50.0))
        fund_score  = _clamp(fund_map.get(ticker, 50.0))

        macro_info = macro_map.get(ticker, {})
        macro_impact = _f(macro_info.get("primary_impact_score"), 0.0)
        macro_gate   = str(macro_info.get("primary_gate") or "NEUTRAL")
        if macro_gate == "AVOID_NEW_BUY":
            macro_score = 0.0
        else:
            macro_score = _clamp((macro_impact + 100.0) / 2.0)  # [-100,+100]→[0,100]

        # 유동성 배수
        liq_label  = liq_map.get(ticker, "MODERATE")
        exec_mode  = exec_map.get(ticker, "")
        if exec_mode == "FROZEN" or liq_label == "FROZEN":
            liq_modifier = 0.0
        else:
            liq_modifier = LIQUIDITY_MODIFIER.get(liq_label, 0.90)

        signal_breakdown = {
            "technical_score":  round(tech_score, 2),
            "smart_money_score": round(smart_score, 2),
            "fundamental_score": round(fund_score, 2),
            "macro_event_score": round(macro_score, 2),
            "macro_gate":        macro_gate,
            "liquidity_label":   liq_label,
            "liquidity_modifier": liq_modifier,
        }

        # 4개 성향별 conviction 산출
        style_rows: list[dict] = []
        for style, weights in W_STYLE.items():
            raw = (weights["technical"]    * tech_score
                 + weights["smartmoney"]   * smart_score
                 + weights["fundamental"]  * fund_score
                 + weights["macro_event"]  * macro_score)
            conviction = _clamp(round(raw * liq_modifier, 2))
            rec_pct    = conviction_to_pct(conviction)

            # 범위 검증
            if not (0.0 <= conviction <= 100.0):
                errors.append(f"{ticker}.{style}: conviction={conviction} out of [0,100]")
            if not (0.0 <= rec_pct <= 7.0):
                errors.append(f"{ticker}.{style}: recommended_pct={rec_pct} out of [0,7]")

            style_rows.append({
                "style":              style,
                "conviction_score":   conviction,
                "recommended_pct":    rec_pct,
                "raw_weighted_score": round(raw, 2),
            })

        rows_out.append({
            "ticker":           ticker,
            "name":             name,
            "signal_breakdown": signal_breakdown,
            "styles":           style_rows,
            "formula_id":       "CAPITAL_STYLE_ALLOCATION_V1",
        })

    gate = "PASS" if not errors and rows_out else ("FAIL" if errors else "NO_DATA")

    best_summary: dict[str, object] = {}
    if rows_out:
        ranked: list[tuple[float, float, dict[str, object], dict[str, object]]] = []
        for row in rows_out:
            styles = [s for s in (row.get("styles") or []) if isinstance(s, dict)]
            if not styles:
                continue
            best_style = max(styles, key=lambda s: _f(s.get("conviction_score"), 0.0))
            best_conviction = _f(best_style.get("conviction_score"), 0.0)
            best_pct = _f(best_style.get("recommended_pct"), 0.0)
            ranked.append((best_conviction, best_pct, row, best_style))
        if ranked:
            ranked.sort(key=lambda item: (item[0], item[1], str(item[2].get("ticker") or "")), reverse=True)
            best_conviction, best_pct, best_row, best_style = ranked[0]
            best_summary = {
                "capital_style_conviction": round(best_conviction, 2),
                "capital_style_label": best_style.get("style") or "UNKNOWN",
                "capital_style_ticker": best_row.get("ticker") or "",
                "capital_style_name": best_row.get("name") or "",
                "capital_style_recommended_pct": round(best_pct, 2),
            }

    result = {
        "formula_id":     "CAPITAL_STYLE_ALLOCATION_V1",
        "gate":           gate,
        "ticker_count":   len(rows_out),
        "style_list":     list(W_STYLE.keys()),
        "weights":        W_STYLE,
        "rows":           rows_out,
        "errors":         errors,
        **best_summary,
        "meta": {
            "weight_source":    "EXPERT_PRIOR",
            "sample_n":         0,
            "llm_computed":     False,
            "deterministic":    True,
            "unvalidated_weight_label": "UNVALIDATED_WEIGHT",
            "calibration_note": "spec/calibration_registry.yaml 등록. "
                                "실측 30건 누적 후 PROVISIONAL→CALIBRATED 승격 필요.",
        },
    }

    out_path = rp(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")

    sep = "=" * 70
    print(sep)
    print("  CAPITAL_STYLE_ALLOCATION_V1")
    print(sep)
    print(f"  gate={gate}  tickers={len(rows_out)}  errors={len(errors)}")
    for r in rows_out[:3]:
        best = max(r["styles"], key=lambda s: s["conviction_score"])
        print(f"  {r['ticker']:<10} {r['name'][:12]:<12} "
              f"best_style={best['style']:<10} conviction={best['conviction_score']:.1f} "
              f"rec_pct={best['recommended_pct']:.1f}%")
    if len(rows_out) > 3:
        print(f"  ... 외 {len(rows_out)-3}개")
    if errors:
        print(f"\n  [!] 오류 {len(errors)}건:")
        for e in errors[:5]:
            print(f"    {e}")
    print(f"\n  → 저장: {out_path}")
    print(f"  {'CAPITAL_ALLOC_BUILD_OK' if gate=='PASS' else 'CAPITAL_ALLOC_BUILD_FAIL'}\n")

    return 0 if gate == "PASS" else 1


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