같은 전략을 자산 29개에 붙였습니다

# -*- coding: utf-8 -*-
"""알고랩 — "전략은 자산을 고른다" 재현 코드 (2026-09-07 공개).

영상에서 말한 숫자 가운데 **신규 14칸 사전등록 외삽**과 **1000만원 경로 4종**을
이 파일 하나로 처음부터 다시 만듭니다. 데이터는 실행할 때 야후·업비트·바이낸스
공개 API 로 직접 받습니다(같은 폴더에 data/ 캐시가 생깁니다).

  · 코인 챔피언 레시피 4종을 손대지 않고 신규 자산 10종(14칸)에 붙입니다.
  · 성적을 보기 전에 추세효율(ER60)만 재서 승/패를 먼저 봉인하고, 그 다음 채점합니다.
  · 1000만원을 넣었을 때의 경로(도지코인·에이다·애플·금)를 뽑습니다.

전체 29칸 표와 주식 96칸 표는 페이지에서 CSV 로 내려받을 수 있습니다
(구 15칸은 지난 편들에서 쓴 시장 캐시에 묶여 있어 이 파일 범위 밖입니다).

기준일은 아래 ASOF 로 못 박혀 있습니다 — 영상과 같은 숫자가 나오게 하려는 것입니다.
최신 데이터로 보려면 ASOF = None 으로 바꾸십시오(숫자는 영상과 달라집니다).

실행:  python run_29.py        (numpy 만 있으면 됩니다)
"""
import json
import os
import sys

import numpy as np

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import datalib as D
import strat_pool as P
from strat_new import squeeze_break, fvg_pullback
from runlib import series_data, make_sig, growth, BARS_PER_DAY
from sweep import span_stats
import overlays as OV

# ── 고정값 ────────────────────────────────────────────────────────────────
ASOF = "2026-09-06"        # 영상을 만든 날. None 이면 오늘까지
FEE = 0.001                # 편도 0.1% — 코인·주식에 동일 적용
SEED = 10_000_000          # 1000만원
ER_THRESH = 0.16           # 강화A 15종에서 관찰된 선. 이 파일이 그 선을 시험한다
BASE_BPD = BARS_PER_DAY["4h"]

# 압축 돌파 파라미터 — 지난 편(신상 스크립트)에서 기준을 통과한 값 그대로
SQ_PARAM = {"q": 0.3, "brk_n": 40, "exit_n": 20}
SQ_FILT = 0


def sc(p, bpd):
    """'일' 단위로 정의한 파라미터를 그 시간봉의 봉수로 환산."""
    return max(2, int(round(p * bpd / BASE_BPD)))


def ones_sig(o, h, l, c):
    return np.ones(len(c))


SPECS = {
    "켈트너BTC형": lambda b: ("keltner", {"n": sc(30, b), "k": 1.5}, 100, None),
    "켈트너ETH형": lambda b: ("keltner", {"n": sc(15, b), "k": 1.25}, 50, None),
    "압축돌파": lambda b: ("squeeze", {k: (sc(v, b) if k in ("brk_n", "exit_n") else v)
                                   for k, v in SQ_PARAM.items()}, SQ_FILT, None),
    "FVG+변타0.5": lambda b: ("fvg", {"look": sc(20, b), "exit_n": sc(20, b)}, 50,
                            {"target": 0.5, "win": sc(42, b)}),
    "필터+변타0.2": lambda b: ("ones", {}, 200, {"target": 0.20, "win": 60 * b}),
}
FN = {"keltner": P.keltner_break, "squeeze": squeeze_break, "fvg": fvg_pullback,
      "ones": ones_sig}
COIN4 = ["켈트너BTC형", "켈트너ETH형", "압축돌파", "FVG+변타0.5"]

NEW_STOCK = {"금(GLD)": "GLD", "은(SLV)": "SLV", "원유(USO)": "USO",
             "달러(UUP)": "UUP", "엔(FXY)": "FXY", "장기채(TLT)": "TLT"}
NEW_CRYPTO = {"SOL": ("KRW-SOL", "SOLUSDT"), "XRP": ("KRW-XRP", "XRPUSDT"),
              "ADA": ("KRW-ADA", "ADAUSDT"), "DOGE": ("KRW-DOGE", "DOGEUSDT")}


# ── 데이터 조달 + ASOF 자르기 ─────────────────────────────────────────────
HERE = os.path.dirname(os.path.abspath(__file__))
FROZEN = os.path.join(HERE, "data_asof")   # 영상에 쓴 그 시점의 캔들을 그대로 동봉


def get(source, symbol, interval):
    """CSV 경로를 돌려준다. 동봉한 data_asof/ 가 있으면 그것을 쓴다.

    ⚠ 왜 얼려서 동봉하는가 — 4시간봉의 **마지막 캔들은 아직 닫히지 않았다.**
       나중에 다시 받으면 그 한 개가 달라져 5년 경로 금액이 0.5~1% 움직인다.
       영상과 같은 숫자를 보시려면 동봉본을 그대로 쓰십시오.
       data_asof/ 를 지우면 공개 API 로 새로 받습니다(그때는 마지막 캔들만큼 달라집니다).
    ⚠ 자르기는 캐시를 쓰든 새로 받든 똑같이 적용한다. 한쪽에만 걸면 두 번째 실행부터 달라진다.
    """
    name = "%s_%s_%s.csv" % (source, symbol.replace("/", "_"), interval)
    frozen = os.path.join(FROZEN, name)
    if os.path.exists(frozen):
        return frozen
    path = D.ensure(source, symbol, interval)
    if not ASOF:
        return path
    os.makedirs(FROZEN, exist_ok=True)
    with open(path, encoding="utf-8") as f, open(frozen, "w", encoding="utf-8") as g:
        g.write(next(f))
        for line in f:
            if line[:10] <= ASOF:
                g.write(line)
    print("  · %s — 새로 받았습니다(마지막 캔들은 영상 시점과 다를 수 있습니다)" % name)
    return frozen


def pair_start(a, b):
    """두 시장 CSV 의 공통 시작일 — 창구성 혼동 방지."""
    def first(p):
        with open(p, encoding="utf-8") as f:
            next(f)
            return next(f).split(",")[0][:10]
    return max(first(a), first(b))


# ── 성적 계산 ─────────────────────────────────────────────────────────────
def cal(c, m):
    return c / abs(m) if m < -1e-9 else float("nan")


def eff_ratio(c, n):
    """추세효율 = |끝-시작| / 이동거리 합. 1에 가까울수록 곧게 간다."""
    c = np.asarray(c, float)
    d = np.abs(np.diff(c))
    num = np.abs(c[n:] - c[:-n])
    den = np.convolve(d, np.ones(n), "valid")
    m = den > 1e-12
    return float(np.mean(num[m] / den[m])) if m.any() else float("nan")


def load(path, tf, filt_days, start=None):
    return series_data(path, tf, max(400, filt_days * BARS_PER_DAY[tf] + 250), start=start)


def run_recipe(S, spec):
    kind, param, filt, vt = spec
    base = make_sig(S, FN[kind], param, filt)
    if vt:
        base = OV.vol_target(S["o"], S["h"], S["l"], S["c"], base,
                             target=vt["target"], win=vt["win"], ppy=S["ppy"])
    pos, g = growth(S, base, FEE)
    a = S["warm"]
    st = span_stats(g[a:], S["ppy"])
    return dict(cagr=float(st["cagr"]), mdd=float(st["mdd"]),
                calmar=cal(st["cagr"], st["mdd"])), pos, g, a


def bh_of(S):
    a = S["warm"]
    st = span_stats(1 + S["ret"][a:], S["ppy"])
    return dict(cagr=float(st["cagr"]), mdd=float(st["mdd"]),
                calmar=cal(st["cagr"], st["mdd"]))


def er60_of(path, tf, start=None):
    bpd = BARS_PER_DAY[tf]
    S = series_data(path, tf, max(400, 100 * bpd + 250), start=start)
    a = S["warm"]
    cd = S["c"][a:][::max(1, bpd)]
    return eff_ratio(cd, 60), float(S["ret"][a:].std() * np.sqrt(S["ppy"]))


def eval_asset(path, tf, start=None):
    """코인 챔피언 4종의 평균 격차(전략 칼마 − 보유 칼마)."""
    gaps = []
    for rname in COIN4:
        spec = SPECS[rname](BARS_PER_DAY[tf])
        S = load(path, tf, spec[2], start=start)
        b = bh_of(S)
        r, _, _, _ = run_recipe(S, spec)
        gaps.append((r["calmar"] - b["calmar"]) if r["calmar"] == r["calmar"] else np.nan)
    return float(np.nanmean(gaps))


# ══════ 1단계 · 성적을 보기 전에 ER60 만 재고 예측을 봉인 ═══════════════════
print("=" * 96)
print("1) 사전 등록 — 신규 10종의 추세효율만 먼저 잰다 (성적 미확인). 임계선 %.2f" % ER_THRESH)
print("=" * 96)
print("%-16s %8s %9s  %s" % ("자산", "ER60", "연변동", "봉인한 예측"))
print("-" * 96)
PRED, PATH, START = {}, {}, {}
for name, sym in NEW_STOCK.items():
    p = get("yahoo", sym, "1d")
    PATH[name], START[name] = (p, "1d"), None
    er, vol = er60_of(p, "1d")
    PRED[name] = "승" if er > ER_THRESH else "패"
    print("%-16s %8.3f %8.1f%%  %s" % (name, er, vol * 100, PRED[name]))
for name, (u, b) in NEW_CRYPTO.items():
    pu, pb = get("upbit", u, "4h"), get("binance", b, "4h")
    st = pair_start(pu, pb)
    for mkt, p in (("업비트", pu), ("바이낸스", pb)):
        key = "%s(%s)4h" % (name, mkt)
        PATH[key], START[key] = (p, "4h"), st
        er, vol = er60_of(p, "4h", start=st)
        PRED[key] = "승" if er > ER_THRESH else "패"
        print("%-16s %8.3f %8.1f%%  %s" % (key, er, vol * 100, PRED[key]))

# ══════ 2단계 · 실측하고 채점 ══════════════════════════════════════════════
print("")
print("=" * 96)
print("2) 실측 — 코인 챔피언 4종을 그대로 붙이고, 봉인한 예측을 채점한다")
print("=" * 96)
print("%-16s %10s %-6s %-6s  %s" % ("자산", "평균격차", "예측", "실제", "채점"))
print("-" * 96)
hit, rows = 0, []
for key, (p, tf) in PATH.items():
    gap = eval_asset(p, tf, start=START[key])
    actual = "승" if gap > 0 else "패"
    ok = actual == PRED[key]
    hit += ok
    rows.append((key, gap, PRED[key], actual, ok))
    print("%-16s %+10.3f %-6s %-6s  %s" % (key, gap, PRED[key], actual, "적중" if ok else "빗나감"))
alt = [r for r in rows if r[0].endswith("4h")]
print("")
print("적중 %d / %d = %d퍼센트  (동전던지기 50퍼센트)"
      % (hit, len(rows), round(100 * hit / len(rows))))
print("알트코인 %d칸 중 실제로 이긴 칸 %d개 · 그중 임계선 아래였던 칸 %d개"
      % (len(alt), sum(1 for r in alt if r[1] > 0),
         sum(1 for r in alt if r[1] > 0 and r[2] == "패")))

# ══════ 3단계 · 1000만원 경로 ══════════════════════════════════════════════
print("")
print("=" * 96)
print("3) 1000만원을 넣었다면 — 크립토는 업비트·바이낸스 중 나쁜 쪽 시장")
print("=" * 96)


def money(path, tf, rname, start=None):
    spec = SPECS[rname](BARS_PER_DAY[tf])
    S = load(path, tf, spec[2], start=start)
    a = S["warm"]
    _, _, g, _ = run_recipe(S, spec)
    eq = SEED * np.cumprod(g[a:])
    bh = SEED * np.cumprod(1 + S["ret"][a:])

    def mdd(x):
        return float((x / np.maximum.accumulate(x) - 1).min())
    return (S["d"][a][:10], S["d"][-1][:10], float(eq[-1]), mdd(eq),
            float(bh[-1]), mdd(bh))


CASES = [
    ("도지코인", get("binance", "DOGEUSDT", "4h"), "4h", "FVG+변타0.5",
     pair_start(get("upbit", "KRW-DOGE", "4h"), get("binance", "DOGEUSDT", "4h"))),
    ("에이다", get("binance", "ADAUSDT", "4h"), "4h", "압축돌파",
     pair_start(get("upbit", "KRW-ADA", "4h"), get("binance", "ADAUSDT", "4h"))),
    ("애플", get("yahoo", "AAPL", "1d"), "1d", "필터+변타0.2", None),
    ("금(GLD)", get("yahoo", "GLD", "1d"), "1d", "켈트너BTC형", None),
]
print("%-9s %-13s %-24s %14s %14s" % ("자산", "레시피", "구간", "그냥 보유", "자동매매"))
print("-" * 96)
for label, p, tf, rname, st in CASES:
    d0, d1, s_end, s_mdd, b_end, b_mdd = money(p, tf, rname, start=st)
    print("%-9s %-13s %s~%s %11s만원 %11s만원"
          % (label, rname, d0, d1,
             "{:,.0f}".format(b_end / 1e4), "{:,.0f}".format(s_end / 1e4)))
    print("%-9s %-13s %-24s 최대낙폭 %6.1f%%   최대낙폭 %6.1f%%"
          % ("", "", "", b_mdd * 100, s_mdd * 100))

print("")
print("한계 — 전 칸 매수만(롱온리) · 종가 체결 · 호가 미끄러짐 미반영 · 편도 0.1%.")
print("알트코인 4종은 현재 거래 중인 종목만 봤습니다(사라진 코인은 표에 없습니다).")
print("그냥 보유 성적은 -90% 구간을 끝까지 버틴다는 전제 위에 있습니다.")
print("과거 데이터 백테스트이며 미래 수익을 보장하지 않습니다.")