app/anchors.py

# -*- coding: utf-8 -*-
"""기준선 봇 — 참가자와 같은 표에 놓일 척도.

ODC가 승인약과 불활성 화합물을 같은 표에 넣어 척도를 보여주듯,
리더보드에 기준 전략을 함께 올린다. 척도가 없는 순위표는 읽을 수 없다.

🔴 주최측 모델은 넣지 않는다.
   이 챌린지의 명분은 "누가 더 나은 예측 모델을 만드는가"이지
   "주최자가 제일 잘한다"가 아니다. 주최자가 자기 모델을 참가자 표에 올리면
   그 명분이 흐려진다. 기준선은 척도이지 경쟁자가 아니다.

기준선 셋 — 전부 공개된 교과서 기법이다
  BUYHOLD    항상 +1. 이걸 못 이기면 아무 의미 없다
  VOLTARGET  지수가중 변동성으로 크기만 조절. 방향은 롱 고정
  RANDOM     매일 -1/0/+1 무작위. 바닥

🔑 VOLTARGET 은 방향을 예측하지 않는다.
   방향 예측은 60개 변수로 다섯 단계에 걸쳐 시험했고 전부 예측력이 없었다.
   기준선이 방향을 맞히는 척하면, 기각한 것을 공개면에 올리는 셈이 된다.
   그래서 이 기준선은 '얼마나 크게 들 것인가'만 정한다.
"""
import os, sys
import numpy as np, pandas as pd

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path[:0] = [HERE, os.path.dirname(HERE)]
import ledger as L

ANCHORS = {
    "BUYHOLD": "단순 보유",
    "VOLTARGET": "변동성 타겟팅",
    "RANDOM": "무작위",
}
SPAN = 60


def position(kind, px, seed=0):
    """오늘 잡을 포지션. -1 ~ +1."""
    if kind == "BUYHOLD":
        return 1.0

    if kind == "RANDOM":
        # 날짜를 시드에 넣어 같은 날 같은 값 — 채점이 재현 가능해야 한다
        g = np.random.default_rng(
            int(pd.Timestamp.utcnow().strftime("%Y%m%d")) * 97 + seed)
        return float(g.choice([-1.0, 0.0, 1.0]))

    s = np.log(px).diff().ewm(span=SPAN).std().dropna()
    if len(s) < SPAN * 2:
        return 0.0
    # 목표 변동성 = 그 종목 과거 중앙값. 변동성이 낮으면 크게, 높으면 작게 든다.
    target = float(s.iloc[:len(s) // 2].median())
    return float(max(0.0, min(1.0, target / float(s.iloc[-1]))))


def run(assets, prices_fn, dry=False, opens=None):
    """기준선 셋의 오늘 포지션을 계산해 원장에 넣는다.

    🔴 기준선은 **시즌 개막 시점부터** 존재해야 한다. 서버가 처음 뜬 순간에
       넣으면 그 전 구간이 비어, 단순 보유가 시장을 그대로 따라가지 않는다.
       실제로 개막일에 시장 −0.82% 인데 단순 보유가 +0.12% 로 나왔다 —
       기준선이 하락 구간을 통째로 비켜간 것이라 척도 구실을 못 한다.

    🔑 값이 그대로면 다시 쓰지 않는다. 서버가 재시작할 때마다 같은 포지션을
       쌓으면 원장이 부풀고, 무엇이 실제 판단 변경인지 읽을 수 없게 된다.
    """
    if not dry:
        for name in ANCHORS:
            L.register(name, ANCHORS[name], "anchor", list(assets))
    out = []
    for i, a in enumerate(assets):
        px = prices_fn(a)
        if px is None or len(px) < SPAN * 2:
            continue
        for name in ANCHORS:
            p = round(position(name, px, seed=i), 4)
            note = ""
            if not dry:
                if opens and not L.positions(a, name):
                    # 개막 시점에 먼저 심는다. 그래야 척도가 전 구간을 덮는다.
                    L.seed(name, a, p, opens, "anchor-open")
                    note = "개막 시점 기록"
                last = L.latest(a, name)
                if last is None or abs(last["p"] - p) > 1e-9:
                    L.submit(name, a, p, "anchor")
                    note = note or "갱신"
                else:
                    note = note or "변동 없음"
            out.append({"anchor": name, "asset": a, "position": p, "note": note})
    return out


if __name__ == "__main__":
    import main as M
    dry = "--dry" in sys.argv
    rows = run(list(M.ASSETS), lambda a: M.prices(a, "1d", 1500), dry=dry)
    print("기준선 포지션 %s\n" % ("(모의)" if dry else "제출 완료"))
    by = {}
    for r in rows:
        by.setdefault(r["asset"], {})[r["anchor"]] = r["position"]
    print("  %-6s" % "종목" + "".join("%12s" % k for k in ANCHORS))
    for a, v in by.items():
        print("  %-6s" % a + "".join("%12.3f" % v.get(k, 0) for k in ANCHORS))