reference.py

# -*- coding: utf-8 -*-
"""기준분포 — '실력 0이었다면 어디까지 갔을까'를 계산한다.

두 가지를 구분한다.

  개막 전 추정치 (announce)
    참가자에게 "이 종목은 운으로 이만큼까지 나온다"고 미리 알려주는 값.
    과거 5년을 블록 부트스트랩해서 낸다. 화면의 '운의 한계선'에 쓴다.

  채점용 기준 (live)
    실제 순위를 정하는 값. 시즌이 진행된 만큼의 **실제 경로** 위에서
    실력 0 참가자를 돌려 만든다. 매일 다시 계산한다.

왜 나누는가
  기준을 과거 부트스트랩으로 고정하면, 시즌 중 그 종목이 크게 오를 때 참가자
  전원의 성적이 함께 부풀려져 순위가 시장 방향을 따라간다. 올바른 반사실은
  '시장이 실제로 그렇게 움직였을 때 운으로 낼 수 있었던 성적'이다.

🔑 개막 전에 고정·공개하는 것은 숫자가 아니라 **방법**이다.
   숫자를 고정하면 시장이 예상과 다르게 움직였을 때 채점이 통째로 어긋난다.

🔑 기준 참가자도 참가자와 똑같이 수수료를 낸다. 안 물리면 참가자만 비용을
   지고 시작한다. 수수료가 높은 종목일수록 그 차이가 커진다.
"""
import os, json, warnings
import numpy as np, pandas as pd
warnings.filterwarnings("ignore")

ROOT = os.path.dirname(os.path.abspath(__file__))
BLOCK = 5
N_LIVE = 4000
N_ANNOUNCE = 20000


def _sim(returns, n, expo, fee, seed, block=None):
    """실력 0 참가자 n명의 최종 수익률. expo 만큼의 시간을 포지션에 노출한다."""
    g = np.random.default_rng(seed)
    r = np.asarray(returns, dtype=float)
    L = len(r)
    out = np.empty(n)
    for i in range(n):
        if block:                                   # 부트스트랩 (개막 전 추정)
            nb = int(np.ceil(L / block))
            st = g.integers(0, max(len(r) - block, 1), size=nb)
            path = np.concatenate([r[j:j+block] for j in st])[:L]
        else:                                       # 실제 경로 (채점)
            path = r
        act = g.random(L) < expo
        pos = np.where(act, g.choice([-1.0, 1.0], size=L), 0.0)
        cost = np.abs(np.diff(np.r_[0.0, pos])) * fee
        out[i] = np.prod(1 + np.r_[0.0, pos[:-1]] * path - cost) - 1
    return np.sort(out)


def live(prices, fee, expo=2/3, n=N_LIVE, seed=None):
    """채점용 — 시즌 경과분의 실제 경로 위에서 만든다. 매일 다시 부른다.

    seed 를 날짜로 주면 같은 날 같은 값이 나와 재현이 된다.
    """
    r = pd.Series(prices, dtype=float).sort_index().pct_change().fillna(0.0).values
    s = seed if seed is not None else int(pd.Timestamp.utcnow().strftime("%Y%m%d"))
    return _sim(r, n, expo, fee, s)


def announce(returns, fee, days, expo=2/3, n=N_ANNOUNCE, seed=20260824):
    """개막 전 추정 — 과거를 블록 부트스트랩해서 '이만큼까지 나온다'를 낸다."""
    r = np.asarray(returns, dtype=float)
    # 시즌 길이만큼만 쓰도록 잘라 붙인다
    reps = int(np.ceil(days / len(r))) + 1
    pool = np.tile(r, reps)[:max(len(r), days * 3)]
    return _sim(pool[:days] if len(pool) >= days else pool, n, expo, fee, seed, block=BLOCK)


def pct_of(ret, refv):
    import bisect
    return bisect.bisect_left(refv, ret) / len(refv)


def score_of(ret, refv):
    import math
    p = min(pct_of(ret, refv), 1 - 1e-6)
    return -math.log10(max(1.0 - p, 1e-6))


# ─────────────────────── 개막 공지값 생성 ───────────────────────
if __name__ == "__main__":
    import seasons as S
    import scoring as SC
    import yfinance as yf

    s1 = S.SEASONS[1]
    days = (pd.Timestamp(s1["closes"]) - pd.Timestamp(s1["opens"])).days
    print("시즌 1 개막 공지값 — %d일 · 수수료 반영\n" % days)
    print("  %-6s %-8s %8s %11s %11s %11s"
          % ("종목", "데이터", "수수료", "운 중앙", "운 상위5%", "운 상위1%"))
    out = {}
    for code, a in s1["assets"].items():
        src = a["src"]
        if code == "BTC":
            p = os.path.join(os.path.dirname(ROOT), "btc-oracle", "data", "hourly.parquet")
            H = pd.read_parquet(p).sort_index()
            if "close" not in H.columns:
                H = H.rename(columns={"c": "close"})
            px = H.close.groupby(H.index.normalize()).last()
        else:
            d = yf.download(src, period="5y", interval="1d",
                            progress=False, auto_adjust=True)["Close"]
            px = (d.iloc[:, 0] if isinstance(d, pd.DataFrame) else d).dropna()
        r = px.pct_change().dropna().values
        fee = SC.fee_of(code)
        c = announce(r, fee, days)
        out[code] = {"p50": float(np.median(c)), "p95": float(np.percentile(c, 95)),
                     "p99": float(np.percentile(c, 99)), "fee": fee, "days": days}
        print("  %-6s %-8s %7.2f%% %10.1f%% %10.1f%% %10.1f%%"
              % (code, src, fee * 100, np.median(c) * 100,
                 np.percentile(c, 95) * 100, np.percentile(c, 99) * 100))
    json.dump({"season": 1, "opens": s1["opens"], "closes": s1["closes"], "days": days,
               "method": "블록부트스트랩 5일 · 무작위 포지션 · 레버리지1 · 수수료 반영",
               "note": "개막 공지용 추정치. 실제 채점은 시즌 실제 경로로 매일 재계산한다.",
               "ceiling": out},
              open(os.path.join(ROOT, "announce_ceiling.json"), "w"),
              ensure_ascii=False, indent=1)
    print("\n  화면의 '운의 한계선'에는 상위5%를 쓴다.")
    print("  저장: finchal/announce_ceiling.json")