QuantEngineByItz/tools/build_market_share_signal_v2.py

"""MARKET_SHARE_SIGNAL_V2 — 시장점유율 프록시 시그널 산출기.

실제 매출 기반 점유율 데이터가 없는 환경에서 3중 프록시를 사용한다:
  1. AvgTradeValue_20D_M — 20일 평균 거래대금(억 기준)  : 유동성/시장 영향력
  2. Frg_20D + Inst_20D  — 외인/기관 20일 누적 순매수   : 수급 강도
  3. Ret20D              — 20일 수익률                  : 상대 모멘텀

비-ETF 유니버스 내 백분위를 산출하여 GAINING/STABLE/LOSING을 결정한다.

백분위 계산:
  상위 33% → GAINING
  중간 34% → STABLE
  하위 33% → LOSING

ETF, 데이터 미수집 → NO_PEER_DATA

confidence: 항상 LOW (proxy 기반)
proxy_basis: "trade_volume_20d+flow+momentum"

참고: Revenue/시장점유율 실데이터 수집 후 HIGH confidence로 업그레이드 예정.
"""
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" / "market_share_signal_v2.json"

_PERCENTILE_GAINING = 67.0   # 상위 33% (≥ 67번째 백분위)
_PERCENTILE_LOSING = 33.0    # 하위 33% (< 33번째 백분위)


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


def _rows(v: Any) -> list[dict[str, Any]]:
    if isinstance(v, list):
        return [x for x in v if isinstance(x, dict)]
    return []


def _f(v: Any, default: float | None = None) -> float | None:
    if v is None or v == "" or v == "N/A":
        return default
    try:
        return float(v)
    except (TypeError, ValueError):
        return default


import re as _re

# ETF 브랜드명 패턴 (이름 기반 1차 판별)
_ETF_NAME_RE = _re.compile(
    r"\b(KODEX|TIGER|KINDEX|ARIRANG|HANARO|KOSEF|TREX|SOL|FOCUS|ACE|TIMEFOLIO|PLUS)\b",
    _re.IGNORECASE,
)
# 6자리 중 5번째 자리 이후에 알파벳이 있는 ETF 티커 패턴 (예: 0117V0)
_ETF_TICKER_RE = _re.compile(r"^\d{4}[A-Z]\d")


def _is_etf(r: dict[str, Any]) -> bool:
    """ETF 여부 판별.

    판별 순서:
    1. Name에 ETF 브랜드명 포함 → ETF
    2. Ticker가 ETF 형식(영문자 포함 6자리) → ETF
    3. EPS / Forward_PE / PBR 모두 없을 때 → 재무 데이터 부재(ETF)
       단, 위 1·2가 모두 false면 비-ETF로 처리 (재무 데이터 미수집 주식 보호)
    """
    name = str(r.get("Name") or r.get("name") or "")
    ticker = str(r.get("Ticker") or r.get("ticker") or "")
    if _ETF_NAME_RE.search(name):
        return True
    if _ETF_TICKER_RE.match(ticker):
        return True
    # 재무 데이터가 있으면 비-ETF로 간주; 없어도 이름/티커 기반 판별이 false면 비-ETF 취급
    return False


def _composite_score(r: dict[str, Any]) -> float | None:
    """거래대금 + 수급 + 모멘텀 합산 점수 (정규화된 상대값)."""
    tv = _f(r.get("AvgTradeValue_20D_M"))
    frg = _f(r.get("Frg_20D"))
    inst = _f(r.get("Inst_20D"))
    ret20 = _f(r.get("Ret20D"))

    if tv is None:
        return None  # 점수 산출 불가

    # 각 요소 점수 (가중치)
    # 거래대금: 50% — 유동성/영향력의 핵심 지표
    tv_score = tv  # 억 단위, 이후 백분위 계산에서 정규화

    # 수급: 30% — 외인+기관 합산 순매수
    flow_score = 0.0
    if frg is not None:
        flow_score += frg * 0.5
    if inst is not None:
        flow_score += inst * 0.5

    # 모멘텀: 20% — 20일 수익률
    mom_score = ret20 if ret20 is not None else 0.0

    # 가중 합산을 위해 각 요소를 거래대금 단위로 스케일링
    # 거래대금이 가장 큰 절대값이므로 기준으로 사용
    composite = tv_score + (flow_score / 1e6 if flow_score != 0 else 0.0) + (mom_score * tv_score * 0.01)
    return composite


def _percentile_rank(value: float, sorted_values: list[float]) -> float:
    """sorted_values 내에서 value의 백분위 계산 (0~100)."""
    if not sorted_values:
        return 50.0
    n = len(sorted_values)
    rank = sum(1 for v in sorted_values if v <= value)
    return rank / n * 100.0


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) if Path(args.json).is_absolute() else ROOT / args.json
    out_path = Path(args.out) if Path(args.out).is_absolute() else ROOT / args.out

    payload = _load(json_path)
    data = payload.get("data") if isinstance(payload.get("data"), dict) else {}

    # data_feed 우선, universe 폴백
    df_list = _rows(data.get("data_feed")) or _rows(data.get("universe"))

    tickers_seen: set[str] = set()
    all_rows: list[dict[str, Any]] = []

    for r in df_list:
        ticker = str(r.get("Ticker") or r.get("ticker") or "")
        if not ticker or ticker in tickers_seen:
            continue
        tickers_seen.add(ticker)
        all_rows.append(r)

    # 비-ETF 유니버스만 백분위 산출
    non_etf_rows = [r for r in all_rows if not _is_etf(r)]
    non_etf_scores: dict[str, float] = {}

    for r in non_etf_rows:
        ticker = str(r.get("Ticker") or r.get("ticker") or "")
        score = _composite_score(r)
        if score is not None:
            non_etf_scores[ticker] = score

    sorted_scores = sorted(non_etf_scores.values())

    rows: list[dict[str, Any]] = []
    label_counts: dict[str, int] = {}

    for r in all_rows:
        ticker = str(r.get("Ticker") or r.get("ticker") or "")
        name = str(r.get("Name") or r.get("name") or "")

        if _is_etf(r):
            state = "NO_PEER_DATA"
            conf = "N/A"
            proxy_basis = "etf_excluded"
            pct_rank: float | None = None
            composite: float | None = None
        elif ticker not in non_etf_scores:
            state = "NO_PEER_DATA"
            conf = "N/A"
            proxy_basis = "trade_value_missing"
            pct_rank = None
            composite = None
        else:
            composite = non_etf_scores[ticker]
            pct_rank = _percentile_rank(composite, sorted_scores)
            if pct_rank >= _PERCENTILE_GAINING:
                state = "GAINING"
            elif pct_rank < _PERCENTILE_LOSING:
                state = "LOSING"
            else:
                state = "STABLE"
            conf = "LOW"
            proxy_basis = "trade_volume_20d+flow+momentum"

        entry = {
            "ticker": ticker,
            "name": name,
            "market_share_state": state,
            "confidence": conf,
            "proxy_basis": proxy_basis,
            "percentile_rank": round(pct_rank, 1) if pct_rank is not None else None,
            "composite_score": round(composite, 2) if composite is not None else None,
            "is_etf": _is_etf(r),
            "formula_id": "MARKET_SHARE_SIGNAL_V2",
        }
        rows.append(entry)
        label_counts[state] = label_counts.get(state, 0) + 1

    # 게이트: 비-ETF 중 GAINING/STABLE/LOSING이 고루 분포해야 신뢰성 있음
    non_etf_labeled = [r for r in rows if not r["is_etf"] and r["market_share_state"] != "NO_PEER_DATA"]
    unique_states = {r["market_share_state"] for r in non_etf_labeled}
    gate = "PASS" if len(unique_states) >= 2 else ("CAUTION" if non_etf_labeled else "FAIL")

    out = {
        "formula_id": "MARKET_SHARE_SIGNAL_V2",
        "gate": gate,
        "proxy_method": "trade_volume_20d+frg_inst_flow+ret20d_momentum",
        "confidence": "LOW",
        "non_etf_scored_count": len(non_etf_scores),
        "unique_states": sorted(unique_states),
        "label_counts": label_counts,
        "row_count": len(rows),
        "rows": rows,
    }
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(out, ensure_ascii=False, indent=2), encoding="utf-8")
    status = "MARKET_SHARE_SIGNAL_V2_OK" if gate != "FAIL" else "MARKET_SHARE_SIGNAL_V2_FAIL"
    print(
        f"MARKET_SHARE_SIGNAL_V2 gate={gate} rows={len(rows)} "
        f"non_etf_scored={len(non_etf_scores)} unique_states={sorted(unique_states)} "
        f"labels={label_counts}"
    )
    print(status)
    return 0


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