app/main.py

# -*- coding: utf-8 -*-
"""FINCHAL 서버 — REST + MCP를 한 프로세스에서 서빙한다.



REST 는 사람과 웹 화면이 쓰고, MCP 는 에이전트가 쓴다.

둘이 같은 원장·같은 채점기를 본다. 경로가 갈리면 성적이 갈린다.

"""
import os, sys, json, time, base64, hmac, hashlib, secrets, urllib.parse, urllib.request
import numpy as np, pandas as pd
from fastapi import FastAPI, Request, Header, HTTPException
from fastapi.responses import (JSONResponse, PlainTextResponse, FileResponse,
                               RedirectResponse)
from fastapi.staticfiles import StaticFiles

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path[:0] = [HERE, os.path.dirname(HERE)]
import seasons as S
import scoring as SC
import reference as RF
import ledger as L
import mcp as MCP
import fetch as FE
import i18n as I
import baselines as BL

CEIL = json.load(open(os.path.join(os.path.dirname(HERE), "announce_ceiling.json"),
                     encoding="utf-8"))["ceiling"]

# ─────────────────────── Hugging Face 로그인 ───────────────────────
# 신원은 참가자가 위조할 수 없는 곳에서 와야 한다. 자유 입력 ID를 두면
# 남의 이름으로 제출할 수 있고, 그러면 순위표가 의미를 잃는다.
OAUTH_ID = os.environ.get("OAUTH_CLIENT_ID", "")
OAUTH_SECRET = os.environ.get("OAUTH_CLIENT_SECRET", "")
OAUTH_ISS = os.environ.get("OPENID_PROVIDER_URL", "https://huggingface.co")
SPACE_HOST = os.environ.get("SPACE_HOST", "")
COOKIE = "finchal_session"
# 다른 출처의 iframe 안에서 뜨므로 우리 쿠키는 3자 쿠키다. Lax 는 거기서 버려지고
# None 은 Secure 를 요구하며 Secure 는 HTTPS 를 요구한다 — Space 에서는 참, 로컬에서는 거짓.
IN_FRAME = bool(SPACE_HOST)
COOKIE_KW = ({"samesite": "none", "secure": True} if IN_FRAME
             else {"samesite": "lax", "secure": False})
# 🔴 재시작할 때마다 바뀌면 모든 세션이 끊긴다. Space 변수로 고정할 것.
SESSION_KEY = os.environ.get("FINCHAL_SESSION_KEY") or secrets.token_hex(16)

app = FastAPI(title="FINCHAL")
S1 = S.SEASONS[1]
# 개발 중 화면 확인용. 시즌 개막일에는 채점할 구간이 없어 차트와 순위가 비므로,
# 개막 전에 화면을 검증하려면 시작일을 앞당겨 띄운다.
if os.environ.get("FINCHAL_OPENS"):
    S1 = dict(S1, opens=os.environ["FINCHAL_OPENS"])
ASSETS = S1["assets"]

_PX = {"at": 0.0, "d": {}}
TTL = 900


_SRC = {}


def prices(asset, interval="1d", bars=800):
    """가격 캐시. 어느 소스가 살아 있든 이 함수 밖에서는 신경 쓰지 않는다."""
    k = "%s|%s" % (asset, interval)
    if time.time() - _PX["at"] > TTL:
        _PX["d"].clear(); _PX["at"] = time.time()
    if k not in _PX["d"]:
        s, src, trail = FE.daily(asset, 2200)
        _SRC[asset] = {"source": src, "trail": trail,
                       "bars": 0 if s is None else len(s)}
        if s is None:
            # 🔴 빈 시리즈를 캐시에 넣으면 TTL 동안 계속 빈 값을 준다.
            #    소스가 잠깐 막힌 것과 영영 막힌 것을 구분해야 한다.
            return pd.Series(dtype=float)
        _PX["d"][k] = s
    return _PX["d"][k].iloc[-bars:]


_HX = {"at": 0.0, "d": {}}


def hourly_px(asset):
    """시간봉 캐시. 채점 격자가 이걸 쓴다."""
    if time.time() - _HX["at"] > 600:
        _HX["d"].clear(); _HX["at"] = time.time()
    if asset not in _HX["d"]:
        _HX["d"][asset] = FE.hourly(asset, 2000)
    return _HX["d"][asset]


def season_px(asset):
    """시즌 개막 이후 구간. 채점과 기준분포가 이걸 본다.



    🔑 시간봉이 있으면 시간봉으로 잰다. 일봉으로 채점하면 개막일에 하루를

       통째로 기다려야 첫 수익률이 나오는데, 규칙은 '매시간 갱신 가능'이라고

       말하고 있다. 재는 단위와 약속한 단위가 다르면 안 된다.

       시간봉이 아직 없는 종목은 일봉으로 내려간다 — 비어 있는 것보다 낫다.

    """
    op = pd.Timestamp(S1["opens"])
    h = hourly_px(asset)
    if h is not None and len(h):
        sub = h[h.index >= op]
        if len(sub) >= 2:
            return sub
    d = prices(asset, "1d", 2000)
    return d[d.index >= op]


def rules_doc(lang="ko"):
    T = lambda d: I.t(d, lang)
    return {
        "lang": lang,
        "season": S1["name"], "opens": S1["opens"], "closes": S1["closes"],
        "days": (pd.Timestamp(S1["closes"]) - pd.Timestamp(S1["opens"])).days,
        "prize": T(I.RULES["prize"]) % (S1["prize_usd_per_asset"],
                                        S1["prize_total_usd"]),
        "leverage": T(I.RULES["leverage"]),
        "update": T(I.RULES["update"]),
        "fees": {k: SC.fee_of(k) for k in ASSETS},
        "fee_note": T(I.RULES["fee_note"]),
        "scoring": T(I.RULES["scoring"]),
        "participation": T(I.RULES["participation"]),
        "assets": {k: {"label": T(I.ASSET[k]["label"]), "why": T(I.ASSET[k]["why"]),
                       "vol_annual": v["vol"], "hours": v["hours"]}
                   for k, v in ASSETS.items()},
        "types": {k: I.t(v, lang) for k, v in I.TYPE.items()},
        "warning": T(I.RULES["warning"]),
        "manifesto": T(I.MANIFESTO["title"]) + " — " + T(I.MANIFESTO["sub"]),
        # 운의 한계선. 실력 0 참가자를 20,000회 돌려 낸 상위 5% 지점이다.
        # 이 값 아래의 수익은 운으로도 설명된다.
        "luck_ceiling": {k: round(v["p95"], 4) for k, v in CEIL.items()},
        "luck_note": T(I.RULES["luck_note"]),
    }


def _sign(payload):
    raw = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode().rstrip("=")
    sig = hmac.new(SESSION_KEY.encode(), raw.encode(), hashlib.sha256).hexdigest()[:32]
    return raw + "." + sig


def _unsign(token):
    try:
        raw, sig = (token or "").rsplit(".", 1)
    except ValueError:
        return None
    good = hmac.new(SESSION_KEY.encode(), raw.encode(), hashlib.sha256).hexdigest()[:32]
    if not hmac.compare_digest(sig, good):
        return None
    pad = "=" * (-len(raw) % 4)
    try:
        return json.loads(base64.urlsafe_b64decode(raw + pad))
    except Exception:
        return None


def current_user(request):
    return _unsign(request.cookies.get(COOKIE))


def _redirect_uri(request):
    if SPACE_HOST:
        return "https://%s/auth/callback" % SPACE_HOST
    return str(request.base_url).rstrip("/") + "/auth/callback"


def lang_of(request, explicit=None):
    """브라우저 언어를 자동 식별한다. 사용자가 고른 값이 있으면 그게 이긴다."""
    return I.pick(request.headers.get("accept-language"),
                  explicit or request.cookies.get("finchal_lang")
                  or request.query_params.get("lang"))


def me_from(key, lang="ko"):
    u = L.user_by_key(key or "")
    if not u:
        raise PermissionError(I.t(I.HINT["need_key"], lang))
    return u


def score_asset(asset):
    """그 종목 참가자 전원을 채점한다. 기준분포는 시즌 실제 경로로 만든다."""
    px = season_px(asset)
    if len(px) < 2:
        # 🔴 개막일에는 채점할 구간이 없다. 그렇다고 빈 표를 내보내면
        #    이미 접수한 참가자에게 "아무도 없다"고 말하는 셈이고, 방문자에게는
        #    고장으로 보인다. 접수된 사실은 그대로 보여 주고 수치만 비워 둔다.
        users = {u["user"]: u for u in L.all_users()}
        seen, out = set(), []
        for r in L.positions(asset):
            seen.add(r["u"])
        for uid in seen:
            u = users.get(uid, {})
            last = L.latest(asset, uid)
            out.append({"user": uid, "strategy": u.get("strategy", "-"),
                        "type": u.get("type", "agent"),
                        "joined": u.get("joined", ""),
                        "ret": None, "score": None, "luck_p95": None,
                        "turnover": None, "fee_paid": None, "valid": None,
                        "position": last["p"] if last else None,
                        "pending": True,
                        "n_sub": len(L.positions(asset, uid))})
        out.sort(key=lambda x: x["joined"])
        for i, r in enumerate(out, 1):
            r["rank"] = i
        return out
    fee = SC.fee_of(asset)
    # 🔴 표본이 얇으면 실시간 기준분포의 폭이 거의 0이라, 손실을 낸 참가자에게도
    #    '운의 한계선 돌파'가 붙는다. 실제로 금에서 −0.033%에 표시가 붙었다.
    #    잡음을 성적으로 보여주는 것이 이 대회가 없애려는 문제 그 자체이므로,
    #    구간이 충분히 쌓이기 전에는 판정 자체를 내리지 않는다.
    span_days = (px.index[-1] - px.index[0]).days
    luck_ready = span_days >= 5 and len(px) >= 30
    recs = L.positions(asset)
    rows = []
    by_user = {}
    for r in recs:
        by_user.setdefault(r["u"], {})[pd.Timestamp(r["t"]).tz_localize(None)] = r["p"]
    users = {u["user"]: u for u in L.all_users()}
    for uid, pos in by_user.items():
        ret = SC.total_return(pos, px, asset)
        expo = float(np.abs(pd.Series(pos).reindex(
            px.index, method="ffill").fillna(0.0)).mean())
        refv = RF.live(px, fee, expo=max(expo, 0.05), n=3000)
        u = users.get(uid, {})
        rows.append({
            "user": uid, "strategy": u.get("strategy", "-"),
            "type": u.get("type", "agent"),
            "joined": u.get("joined", ""),
            "ret": ret, "score": SC.score(ret, refv),
            "luck_p95": float(np.percentile(refv, 95)),
            "luck_ready": luck_ready, "span_days": span_days,
            "turnover": SC.turnover(pos, px),
            "fee_paid": SC.fee_paid(pos, px, asset),
            "valid": SC.valid_ratio(pos, px),
            "n_sub": len(L.positions(asset, uid)),
        })
    rows.sort(key=lambda x: -x["ret"])
    for i, r in enumerate(rows, 1):
        r["rank"] = i
    return rows


_SC = {}
SC_TTL = 300


def score_cached(asset):
    """채점은 참가자마다 기준분포 3,000경로를 돌린다. 매 요청마다 하면 화면이 죽는다."""
    e = _SC.get(asset)
    if e and time.time() - e[0] < SC_TTL:
        return e[1]
    r = score_asset(asset)
    _SC[asset] = (time.time(), r)
    return r


def curves(asset, top=12):
    """참가자별 누적 수익률 곡선. 화면의 꺾은선 그래프가 이걸 그린다."""
    px = season_px(asset)
    if len(px) < 2:
        return {"t": [], "series": []}
    rows = score_cached(asset)
    users = {u["user"]: u for u in L.all_users()}
    keep = [r["user"] for r in rows[:top]]
    ser = []
    for uid in keep:
        pos = {}
        for r in L.positions(asset, uid):
            pos[pd.Timestamp(r["t"]).tz_localize(None)] = r["p"]
        if not pos:
            continue
        eq, _ = SC.equity(pos, px, asset)
        u = users.get(uid, {})
        ser.append({"user": uid, "strategy": u.get("strategy", "-"),
                    "type": u.get("type", "agent"),
                    "v": [round(float(x - 1) * 100, 4) for x in eq]})
    # 시장 자체도 한 줄로 깔아 준다 — 참가자가 무엇을 이겨야 하는지 보이게
    mk = (px / px.iloc[0] - 1) * 100
    return {"t": [t.isoformat() for t in px.index],
            "market": [round(float(x), 4) for x in mk],
            "series": ser}


# ─────────────────────── REST ───────────────────────
@app.get("/api/rules")
def api_rules(request: Request, lang: str = None):
    return {"ok": True, **rules_doc(lang_of(request, lang))}


@app.get("/api/me")
def api_me(request: Request):
    u = current_user(request)
    out = {"signed_in": bool(u), "user": u,
           "oauth_configured": bool(OAUTH_ID and OAUTH_SECRET),
           "lang": lang_of(request)}
    if u:
        # 로그인한 사람은 등록을 따로 할 필요가 없다. 키를 바로 준다.
        out["key"] = L.make_key(u["name"])
        rec = next((x for x in L.all_users() if x["user"] == u["name"]), None)
        out["strategy"] = (rec or {}).get("strategy", "")
        out["joined"] = (rec or {}).get("joined")
    return out


@app.get("/login")
def login(request: Request):
    if not (OAUTH_ID and OAUTH_SECRET):
        raise HTTPException(503, "OAuth is not configured (needs hf_oauth in README)")
    state = secrets.token_urlsafe(16)
    q = urllib.parse.urlencode({
        "client_id": OAUTH_ID, "redirect_uri": _redirect_uri(request),
        "response_type": "code", "scope": "openid profile", "state": state})
    r = RedirectResponse("%s/oauth/authorize?%s" % (OAUTH_ISS.rstrip("/"), q))
    # 제공자가 되돌려주는 state 를 이 쿠키와 대조하는 것이,
    # 제3자가 남의 로그인을 가로채 완료하는 것을 막는 유일한 수단이다.
    r.set_cookie("finchal_state", state, max_age=600, httponly=True, **COOKIE_KW)
    return r


@app.get("/auth/callback")
def callback(request: Request, code: str = "", state: str = ""):
    lang = lang_of(request)
    saved = request.cookies.get("finchal_state")
    if not code:
        raise HTTPException(400, "No authorization code — please sign in again")
    if not saved:
        # 쿠키가 저장되지 않은 것이다. 위조가 아니라 브라우저 정책 문제다.
        raise HTTPException(400, (
            "브라우저가 쿠키를 저장하지 못했습니다. 이 페이지를 새 탭에서 직접 열고"
            "(.hf.space 주소) 다시 로그인해 주세요. 3자 쿠키 차단이 원인일 수 있습니다."
            if lang == "ko" else
            "Your browser did not store the cookie. Open this page directly in a new tab "
            "(the .hf.space address) and sign in again — third-party cookie blocking is "
            "the usual cause."))
    if state != saved:
        raise HTTPException(400, "Sign-in state mismatch — please try again")
    body = urllib.parse.urlencode({
        "client_id": OAUTH_ID, "client_secret": OAUTH_SECRET,
        "grant_type": "authorization_code", "code": code,
        "redirect_uri": _redirect_uri(request)}).encode()
    tr = urllib.request.Request(OAUTH_ISS.rstrip("/") + "/oauth/token", data=body,
                                headers={"Content-Type": "application/x-www-form-urlencoded"})
    with urllib.request.urlopen(tr, timeout=20) as r:
        tok = json.loads(r.read().decode())
    ur = urllib.request.Request(OAUTH_ISS.rstrip("/") + "/oauth/userinfo",
                                headers={"Authorization": "Bearer " + tok["access_token"]})
    with urllib.request.urlopen(ur, timeout=20) as r:
        ui = json.loads(r.read().decode())
    user = {"name": ui.get("preferred_username") or ui.get("name"),
            "picture": ui.get("picture"), "at": int(time.time())}
    resp = RedirectResponse("/")
    resp.set_cookie(COOKIE, _sign(user), max_age=60 * 60 * 24 * 14,
                    httponly=True, **COOKIE_KW)
    resp.delete_cookie("finchal_state")
    return resp


@app.get("/logout")
def logout():
    r = RedirectResponse("/")
    r.delete_cookie(COOKIE)
    return r


@app.get("/api/lang/{code}")
def set_lang(code: str):
    """언어 선택을 기억한다. 자동 식별보다 사용자의 선택이 우선한다."""
    if code not in I.LANGS:
        return JSONResponse({"ok": False, "error": "unknown language"}, 400)
    r = JSONResponse({"ok": True, "lang": code})
    r.set_cookie("finchal_lang", code, max_age=60 * 60 * 24 * 365, **COOKIE_KW)
    return r


@app.get("/api/data")
def api_data(asset: str, bars: int = 500, interval: str = "1d"):
    if asset not in ASSETS:
        return JSONResponse({"ok": False, "error": "알 수 없는 종목"}, 400)
    if interval == "1h":
        h = hourly_px(asset)
        if h is None or not len(h):
            return JSONResponse({"ok": False,
                                 "error": "시간봉 없음 / no hourly bars yet"}, 400)
        return {"ok": True, "asset": asset, "interval": "1h",
                "bars": [{"t": t.isoformat(), "c": float(v)}
                         for t, v in h.iloc[-min(max(bars, 10), 5000):].items()]}
    if interval != "1d":
        return JSONResponse({"ok": False,
                             "error": "1d 또는 1h / 1d or 1h only"}, 400)
    s = prices(asset, interval, min(max(bars, 10), 5000))
    return {"ok": True, "asset": asset, "interval": interval,
            "bars": [{"t": t.isoformat(), "c": float(v)} for t, v in s.items()]}


# 🔴 자율 에이전트는 브라우저 OAuth 를 통과할 수 없다. 그렇다고 아무나 이름을
#    지어 등록하게 두면 남의 이름으로 제출할 수 있어 순위표가 무의미해진다.
#    그래서 별도 관리 키를 가진 요청만 지정 이름으로 등록할 수 있게 한다.
#    이 키는 Space 시크릿이고, 발급 대상은 운영자가 직접 확인한 계정뿐이다.
ADMIN_KEY = os.environ.get("FINCHAL_ADMIN_KEY", "")


@app.post("/api/agent/enroll")
async def api_agent_enroll(req: Request, x_finchal_admin: str = Header(None)):
    """헤드리스 에이전트 등록. 관리 키가 있어야 한다."""
    if not ADMIN_KEY or (x_finchal_admin or "").strip() != ADMIN_KEY:
        return JSONResponse({"ok": False, "error": "not authorized"}, 403)
    b = await req.json()
    u = L.register(b["user"], b.get("strategy", "-"), b.get("type", "agent"),
                   b.get("assets", list(ASSETS)))
    return {"ok": True, **u}


@app.post("/api/register")
async def api_register(req: Request):
    b = await req.json()
    sess = current_user(req)
    if OAUTH_ID and OAUTH_SECRET:
        if not (sess and sess.get("name")):
            return JSONResponse({"ok": False, "error":
                ("Hugging Face 로그인이 필요합니다" if lang_of(req) == "ko"
                 else "Please sign in with Hugging Face")}, 401)
        who = sess["name"]          # 신원은 제출자가 정하지 않는다
    else:
        who = b["user"]
    u = L.register(who, b.get("strategy", "-"),
                   b.get("type", "human" if sess else "agent"),
                   b.get("assets", list(ASSETS)))
    return {"ok": True, **u}


@app.post("/api/submit")
async def api_submit(req: Request, authorization: str = Header(None),

                     x_finchal_key: str = Header(None)):
    b = await req.json()
    lang = lang_of(req)
    # Authorization 은 앞단 프록시가 자기 인증에 쓰기도 한다. 그때 참가키가
    # 앱까지 오지 않는다. 전용 헤더를 함께 받아 그 충돌을 피한다.
    key = ((x_finchal_key or "").strip()
           or (authorization or "").replace("Bearer ", "").strip()
           or b.get("key", ""))
    # 🔑 화면에서 온 요청은 Hugging Face 로그인 쿠키가 곧 신원이다.
    #    자유 입력 ID를 허용하면 남의 이름으로 제출할 수 있고 순위표가 무의미해진다.
    sess = current_user(req)
    if sess and sess.get("name"):
        u = L.register(sess["name"], b.get("strategy", "-"),
                       b.get("type", "human"), list(ASSETS))
    else:
        if OAUTH_ID and OAUTH_SECRET and not key:
            return JSONResponse({"ok": False, "error":
                ("Hugging Face 로그인이 필요합니다" if lang == "ko"
                 else "Please sign in with Hugging Face")}, 401)
        try:
            u = me_from(key, lang)
        except PermissionError as e:
            return JSONResponse({"ok": False, "error": str(e)}, 401)
    if b["asset"] not in ASSETS:
        return JSONResponse({"ok": False, "error": "알 수 없는 종목"}, 400)
    r = L.submit(u["user"], b["asset"], b["position"], b.get("source", "rest"))
    _SC.pop(b["asset"], None)      # 방금 낸 사람이 표에서 자기를 못 찾으면 안 된다
    return {"ok": True, **r}


@app.get("/api/leaderboard")
def api_leaderboard(asset: str):
    if asset not in ASSETS:
        return JSONResponse({"ok": False, "error": "알 수 없는 종목"}, 400)
    return {"ok": True, "asset": asset, "rows": score_cached(asset)}


@app.get("/api/curves")
def api_curves(asset: str, top: int = 12):
    if asset not in ASSETS:
        return JSONResponse({"ok": False, "error": "알 수 없는 종목"}, 400)
    return {"ok": True, "asset": asset, **curves(asset, top)}


_BLC = {}


@app.get("/api/baselines")
def api_baselines(request: Request, asset: str, lang: str = None):
    """교과서 규칙들을 최근 구간에 적용한 되돌아보기. 참가자 곡선의 척도로 쓴다."""
    if asset not in ASSETS:
        return JSONResponse({"ok": False, "error": "알 수 없는 종목"}, 400)
    g = lang_of(request, lang)
    k = "%s|%s" % (asset, g)
    e = _BLC.get(k)
    if not e or time.time() - e[0] > 1800:
        _BLC[k] = (time.time(), BL.curves(prices(asset, "1d", 400), asset, SC, g))
    return {"ok": True, "asset": asset, **_BLC[k][1]}


@app.get("/api/manifesto")
def api_manifesto(request: Request, lang: str = None):
    g = lang_of(request, lang)
    T = lambda d: I.t(d, g)
    M = I.MANIFESTO
    s2 = S.SEASONS.get(2, {})
    return {"ok": True, "lang": g,
            "title": T(M["title"]), "sub": T(M["sub"]),
            "acts": [[T(a), T(b)] for a, b in M["acts"]],
            "why": [[T(a), T(b)] for a, b in M["why"]],
            "how": [[T(a), T(b)] for a, b in M["how"]],
            "our_stake": T(M["our_stake"]),
            "anchors": {k: T(v) for k, v in I.ANCHOR.items()},
            "season2": {"name": ("시즌 2 · 예고" if g == "ko" else "Season 2 · Coming"),
                        "teaser": ("선물과 원자재, 그리고 외환" if g == "ko"
                                   else "Futures, commodities and currencies"),
                        "note": ("종목은 시즌 1 종료 후 확정합니다." if g == "ko"
                                 else "Assets will be announced after Season 1 closes.")}}


# ─────────────────────── MCP ───────────────────────
def _call(name, args, key=None, lang="ko"):
    T = lambda d: I.t(d, lang)
    if name == "get_rules":
        return rules_doc(lang)
    if name == "get_data":
        a = args["asset"]
        if a not in ASSETS:
            return {"error": T(I.HINT["unknown_asset"]) % list(ASSETS)}
        iv = args.get("interval", "1d")
        n = min(args.get("bars", 500), 5000)
        if iv == "1h":
            h = hourly_px(a)
            if h is None or not len(h):
                return {"error": ("이 종목은 아직 시간봉이 없습니다. '1d' 를 쓰십시오."
                                  if lang == "ko" else
                                  "No hourly bars for this asset yet. Use '1d'.")}
            return {"asset": a, "interval": "1h",
                    "bars": [{"t": t.isoformat(sep=" "), "c": round(float(v), 6)}
                             for t, v in h.iloc[-n:].items()]}
        if iv != "1d":
            # 요청을 조용히 무시하지 않는다. 못 주는 것은 못 준다고 말한다.
            return {"error": ("interval 은 '1d' 또는 '1h' 만 가능합니다."
                              if lang == "ko" else
                              "interval must be '1d' or '1h'.")}
        s = prices(a, iv, n)
        return {"asset": a, "interval": "1d",
                "bars": [{"t": t.date().isoformat(), "c": round(float(v), 6)}
                         for t, v in s.items()]}
    if name == "submit_position":
        u = me_from(key, lang)
        a = args["asset"]
        if a not in ASSETS:
            return {"error": T(I.HINT["unknown_asset"]) % list(ASSETS)}
        r = L.submit(u["user"], a, args["position"], "mcp")
        _SC.pop(a, None)
        return {"accepted": True, "asset": a, "position": r["p"], "at": r["t"],
                "note": T(I.HINT["kept"])}
    if name == "check_score":
        u = me_from(key, lang)
        want = [args["asset"]] if args.get("asset") else u.get("assets", list(ASSETS))
        out = []
        for a in want:
            if a not in ASSETS:
                continue
            rows = score_asset(a)
            mine = next((r for r in rows if r["user"] == u["user"]), None)
            if not mine:
                last = L.latest(a, u["user"])
                out.append({"asset": a, "status": T(I.HINT["pending"]),
                            "position": last["p"] if last else None,
                            "hint": T(I.HINT["have_pos"] if last
                                      else I.HINT["no_pos"])})
            elif mine.get("pending") or mine.get("ret") is None:
                # 🔴 표에는 올라와 있지만 아직 잴 구간이 없다. 여기서 수치를
                #    반올림하려 들면 None 에 걸려 도구 호출 자체가 실패한다.
                out.append({"asset": a, "status": T(I.HINT["pending"]),
                            "position": mine.get("position"),
                            "of": len(rows),
                            "hint": T(I.HINT["have_pos"])})
            else:
                out.append({"asset": a, "rank": mine["rank"], "of": len(rows),
                            "return": round(mine["ret"], 4),
                            "luck_ceiling": round(mine["luck_p95"], 4),
                            "score": round(mine["score"], 2),
                            "fee_paid": round(mine["fee_paid"], 4),
                            "hint": T(I.HINT["over"] if mine["ret"] > mine["luck_p95"]
                                      else I.HINT["under"])})
        return {"user": u["user"], "strategy": u.get("strategy"), "results": out}
    return {"error": "미구현"}


@app.post("/mcp")
async def mcp_endpoint(req: Request, authorization: str = Header(None),

                       x_finchal_key: str = Header(None)):
    body = await req.json()
    lang = lang_of(req)
    key = ((x_finchal_key or "").strip()
           or (authorization or "").replace("Bearer ", "").strip())
    resp = MCP.handle(body, lambda n, a: _call(n, a, key, lang), lang)
    if resp is None:
        return JSONResponse({}, 202)
    return JSONResponse(resp)


@app.get("/llms.txt", response_class=PlainTextResponse)
def llms(request: Request, lang: str = None):
    """에이전트가 먼저 읽는 요약. 도구 설명과 같은 언어여야 한다."""
    g = lang_of(request, lang)
    r = rules_doc(g)
    a = ", ".join("%s (%s)" % (k, v["label"]) for k, v in r["assets"].items())
    N = "\n"
    if g == "en":
        return (("# %s" + N + N) % r["manifesto"] +
                "You take real positions and compete on returns. "
                "This is trading, not calling direction." + N + N +
                ("Assets: %s" + N) % a +
                ("Season: %s to %s" + N) % (r["opens"], r["closes"]) +
                ("Prize: %s" + N) % r["prize"] +
                ("Position: %s" + N) % r["leverage"] +
                ("Updates: %s" + N) % r["update"] +
                "Fees: charged on position changes only. "
                "Flipping often loses money." + N +
                ("Scoring: %s" + N + N) % r["scoring"] +
                "Start: get_rules() -> get_data() -> build your own model "
                "-> submit_position()" + N)
    return (("# %s" + N + N) % r["manifesto"] +
            "당신은 실제 포지션을 잡아 수익률로 겨룹니다. 방향 예측이 아니라 매매입니다."
            + N + N +
            ("종목: %s" + N) % a +
            ("기간: %s ~ %s" + N) % (r["opens"], r["closes"]) +
            ("상금: %s" + N) % r["prize"] +
            ("포지션: %s" + N) % r["leverage"] +
            ("갱신: %s" + N) % r["update"] +
            "수수료: 포지션 변경분에만. 자주 뒤집으면 손해입니다" + N +
            ("채점: %s" + N + N) % r["scoring"] +
            "시작: get_rules() → get_data() → 직접 모델링 → submit_position()" + N)


STATIC = os.path.join(HERE, "static")
if os.path.isdir(STATIC):
    app.mount("/static", StaticFiles(directory=STATIC), name="static")


@app.get("/")
def index():
    f = os.path.join(STATIC, "index.html")
    if not os.path.exists(f):
        return JSONResponse({"ok": True, "note": "화면 준비 중", **rules_doc()})
    return FileResponse(f, headers={"Cache-Control": "no-cache"})


# ─────────────────────── 기준선 자동 제출 ───────────────────────
def _anchor_loop():
    """하루 한 번 기준선 셋의 포지션을 낸다.



    🔑 실패해도 서버를 죽이면 안 된다. 기준선은 척도이지 서비스가 아니다.

       yfinance가 잠깐 막혀도 화면과 참가자 제출은 계속 돌아가야 한다.

    """
    import traceback
    import anchors as A
    while True:
        try:
            A.run(list(ASSETS), lambda a: prices(a, "1d", 1500),
                  opens=S1["opens"] + "T00:00:00+00:00")
            _SC.clear()
        except Exception:
            traceback.print_exc()
        time.sleep(6 * 3600)


# 🔑 원장 복원은 **다른 무엇보다 먼저**다. 복원 전에 부팅 표식을 쓰면
#    그 표식이 미러를 덮어 원장을 지운다.
_RESTORE = L.restore()
_DUR = L.boot_check()


def _flush_loop():
    """원장을 주기적으로 미러에 올린다. 제출마다 커밋하면 저장소가 터진다."""
    import traceback
    while True:
        time.sleep(90)
        try:
            L.flush()
        except Exception:
            traceback.print_exc()


@app.on_event("startup")
def _startup():
    import threading
    print("[finchal] 원장 %s · 미러 %s" % (L.ROOT, _RESTORE))
    if L.MIRROR:
        threading.Thread(target=_flush_loop, daemon=True).start()
    if os.environ.get("FINCHAL_NO_ANCHOR"):
        return
    threading.Thread(target=_anchor_loop, daemon=True).start()


@app.get("/api/diag")
def api_diag():
    """어느 소스에서 시세가 오는지 실측한다.



    🔑 로컬에서 되는지는 아무 의미가 없다. 막히는 곳은 서버다.

    """
    out = {}
    for a in ASSETS:
        s = prices(a, "1d", 5)
        out[a] = {**_SRC.get(a, {}),
                  "last": None if not len(s) else round(float(s.iloc[-1]), 4),
                  "at": None if not len(s) else s.index[-1].date().isoformat()}
    return {"ok": all(v.get("bars") for v in out.values()), "assets": out}


@app.get("/health")
def health():
    return {"ok": True, "season": S1["name"], "assets": list(ASSETS),
            "ledger": L.ROOT, "durability": _DUR[0], "boots": _DUR[1],
            "mirror": L.mirror_status(), "restore": _RESTORE,
            "warn": None if (_DUR[0] == "확인됨" or L.mirror_status()["restored"])
                    else "원장이 재시작을 견딘다는 증거가 없습니다. "
                         "미러(FINCHAL_MIRROR)나 영속 스토리지를 붙이십시오."}