scoring.py

# -*- coding: utf-8 -*-
"""채점 엔진 — 포지션 기록에서 수익률과 순위 점수를 낸다.

규칙 (사전등록)
  포지션은 -1.0 ~ +1.0. 범위를 벗어나면 잘라낸다 (레버리지 1 고정)
  포지션 갱신은 매시간 가능. 미갱신 구간은 직전 포지션이 유지된다
  수익률 = 각 구간 포지션 x 그 구간 수익률의 복리
  거래비용 = 포지션 변경분에 왕복 0.02% (미국주식 무료 + 스프레드 추정)
  순위 점수 = -log10(1 - 백분위)   백분위는 그 종목의 '실력0 분포' 기준

이 파일을 바꾸면 selftest() 를 반드시 다시 돌린다. 채점 코드의 오류는 예외로
드러나지 않고 그럴듯한 숫자로 나오므로, 닫힌 형태로 답을 아는 경우들로만 검증한다.
"""
import os, json, math, bisect
import numpy as np
import pandas as pd

ROOT = os.path.dirname(os.path.abspath(__file__))
# 종목별 실거래 비용. 포지션 변경분(|Δw|)에 곱한다.
# 🔑 이걸 물리지 않으면 포지션을 초당 뒤집는 무차별 제출이 이득을 본다.
#    매시간 ±1을 뒤집으면 121일에 |Δw| 합계가 5,808이 되는데, 코인 0.06%면
#    수수료만 자본의 348%다. 규칙이 아니라 비용이 그런 행동을 막는다.
FEES = {
    # 코인 — 바이낸스 테이커 0.05% + 슬리피지 0.01%
    "BTC": 0.0006, "ETH": 0.0006, "SOL": 0.0008,
    # 미국 주식 — 수수료 무료지만 스프레드·슬리피지가 남는다
    "NVDA": 0.0002, "TSLA": 0.0002, "AAPL": 0.0002,
    # 선물 — 계약당 수수료 + 호가폭
    "GOLD": 0.0002, "OIL": 0.0003,
}
FEE_DEFAULT = 0.0006
FEE = FEE_DEFAULT            # selftest에서 임시로 바꿔 쓰는 전역값


def fee_of(asset=None):
    if asset is None:
        return FEE
    return FEES.get(asset, FEE_DEFAULT)
POS_MIN, POS_MAX = -1.0, 1.0

def clip(p):
    """레버리지 1 상한. 넘겨 제출해도 잘라낸다 — 거부하지 않는 이유는
    에이전트가 반올림 오차로 1.0000001을 보내는 일이 흔하기 때문이다."""
    return max(POS_MIN, min(POS_MAX, float(p)))


def equity(positions, prices, asset=None):
    """포지션 기록 + 가격에서 자산 곡선을 만든다.

    positions: 시각→포지션 (제출된 시점만). 사이 구간은 직전 값이 유지된다
    prices:    시각→가격 (채점 격자. 보통 1시간 간격)
    """
    pos = pd.Series(positions, dtype=float).sort_index().map(clip)
    px = pd.Series(prices, dtype=float).sort_index()
    w = pos.reindex(px.index, method="ffill").fillna(0.0)
    r = px.pct_change().fillna(0.0)
    cost = w.diff().abs().fillna(w.abs()) * fee_of(asset)
    g = w.shift(1).fillna(0.0) * r - cost      # 🔑 포지션은 다음 구간에 적용된다
    return (1 + g).cumprod(), g


def total_return(positions, prices, asset=None):
    eq, _ = equity(positions, prices, asset)
    return float(eq.iloc[-1] - 1.0)


def turnover(positions, prices):
    """포지션 변경 총량. 무차별 제출 탐지와 화면 표시에 쓴다."""
    pos = pd.Series(positions, dtype=float).sort_index().map(clip)
    px = pd.Series(prices, dtype=float).sort_index()
    w = pos.reindex(px.index, method="ffill").fillna(0.0)
    return float(w.diff().abs().fillna(w.abs()).sum())


def fee_paid(positions, prices, asset=None):
    return turnover(positions, prices) * fee_of(asset)


def percentile(ret, refv):
    """그 종목의 '실력0 분포'에서 내 수익률의 백분위.

    refv 는 reference.live(가격, 수수료, 노출) 이 만든 정렬된 배열이다.
    시즌 실제 경로로 매일 다시 만든다 — 과거 부트스트랩으로 고정하면
    시즌 중 그 종목이 오를 때 참가자 전원이 부풀려진다.
    """
    return bisect.bisect_left(refv, ret) / len(refv)


def score(ret, refv):
    """운으로 이 성적이 나올 확률의 역수 지수. 2.0이면 1/100, 3.0이면 1/1,000.

    🔑 순위는 종목 안에서만 매긴다. 수익률 분포의 꼬리 모양이 종목마다 달라
       종목 간 직접 비교는 어느 통계량을 맞추어도 편중이 남는다. 우승자는
       최댓값이고 최댓값은 꼬리가 정하기 때문이다. 종목별 시상이 그 비교를 없앤다.

       종목 안에서는 잘 교정된다 — 실력 0 참가자의 백분위 중앙이 0.47~0.54다.
    """
    p = min(percentile(ret, refv), 1 - 1e-6)
    return -math.log10(max(1.0 - p, 1e-6))


def valid_ratio(positions, prices):
    """포지션이 유효했던 구간 비율. 참여 요건 판정에 쓴다."""
    pos = pd.Series(positions, dtype=float).sort_index()
    if not len(pos):
        return 0.0
    px = pd.Series(prices, dtype=float).sort_index()
    w = pos.reindex(px.index, method="ffill")
    return float(w.notna().mean())


# ─────────────────────── 채점기 자체 검증 ───────────────────────
def selftest(verbose=True):
    """닫힌 형태로 답을 아는 경우들. 하나라도 틀리면 채점을 신뢰할 수 없다."""
    out = []
    idx = pd.date_range("2026-01-01", periods=101, freq="h")

    def chk(name, got, want, tol=1e-9):
        ok = abs(got - want) <= tol
        out.append((name, got, want, ok))
        return ok

    # ① 포지션 0이면 수익도 0 — 가격이 어떻게 움직이든
    px = pd.Series(100 * np.cumprod(1 + np.random.default_rng(0).normal(0, .01, 101)), index=idx)
    chk("포지션 0 → 수익 0", total_return({idx[0]: 0.0}, px), 0.0)

    # ② 포지션 +1 · 무비용이면 단순보유와 정확히 같아야
    global FEE
    keep = FEE
    FEE = 0.0
    chk("포지션 +1 = 단순보유", total_return({idx[0]: 1.0}, px),
        float(px.iloc[-1] / px.iloc[0] - 1), 1e-9)
    # ③ 포지션 -1 은 부호가 뒤집힌 복리 (단순 -x 가 아니다)
    w = -1.0
    want = float(np.prod(1 + w * px.pct_change().fillna(0).values[1:]) - 1)
    chk("포지션 -1 = 역방향 복리", total_return({idx[0]: -1.0}, px), want, 1e-9)
    FEE = keep

    # ④ 레버리지 상한 — 2.0을 내도 1.0으로 잘려야
    FEE = 0.0
    chk("포지션 2.0 → 1.0으로 절단",
        total_return({idx[0]: 2.0}, px), total_return({idx[0]: 1.0}, px), 1e-12)
    FEE = keep

    # ⑤ 미래를 보지 않는가 — 가격이 오르기 직전에 포지션을 넣어도
    #    그 구간 수익은 못 먹는다 (포지션은 다음 구간부터 적용)
    p2 = pd.Series([100.0] * 50 + [200.0] * 51, index=idx)
    FEE = 0.0
    late = total_return({idx[50]: 1.0}, p2)      # 점프가 일어난 봉에 진입
    early = total_return({idx[49]: 1.0}, p2)     # 한 봉 먼저 진입
    ok = abs(late) < 1e-9 and early > 0.9
    out.append(("미래 차단 (늦게 들어가면 못 먹음)", late, 0.0, ok))
    FEE = keep

    # ⑥ 비용이 실제로 빠지는가
    a = total_return({idx[0]: 1.0}, px)
    FEE = 0.01
    b = total_return({idx[0]: 1.0}, px)
    FEE = keep
    out.append(("비용이 수익을 낮춤", b, a, b < a))

    # ⑦ 백분위·점수가 단조인가 — 인공 기준분포로 확인
    fake = np.sort(np.random.default_rng(1).normal(0, .3, 4000))
    out.append(("점수 단조 증가", score(0.6, fake), score(0.1, fake),
                score(0.6, fake) > score(0.1, fake)))
    med = float(np.median(fake))
    out.append(("중앙값 근처 점수 ~0.3", score(med, fake), 0.3,
                abs(score(med, fake) - 0.30) < 0.05))

    if verbose:
        print("=" * 72)
        print("채점기 자체 검증")
        print("=" * 72)
        for n, g, w, ok in out:
            print("  %-32s %14.8f  %s" % (n, g, "✅" if ok else "🔴 실패 (기대 %.8f)" % w))
        bad = [x for x in out if not x[3]]
        print("\n  %d/%d 통과%s" % (len(out) - len(bad), len(out),
                                   "" if not bad else "  🔴 실패 %d건 — 채점 중단" % len(bad)))
    return all(x[3] for x in out)


if __name__ == "__main__":
    ok = selftest()
    raise SystemExit(0 if ok else 1)