전략 1,887개를 걸렀더니
1000만원이 5억 4천만원이 됐습니다.

# -*- coding: utf-8 -*-
# =============================================================================
# AlgoLab · 전략 1,887개 발굴 백테스트 — 공개용
#   영상 「1,000만원 코인 자동매매, 1,887개 백테스트」의 최종 조합 숫자를 그대로 재현합니다.
#   믿지 말고 직접 돌려보세요. 설정값을 바꿔 여러분의 조건으로 실험해도 됩니다.
#
# 실행:  python discovery_backtest.py
# 필요:  Python 3.8+ / numpy  (pip install numpy)
#        데이터는 업비트·바이낸스 공개 API 에서 자동으로 받습니다(API 키 불필요).
#        첫 실행은 캔들 수집에 3~5분 걸립니다(같은 폴더에 CSV 로 캐시되어 두 번째부터는 몇 초).
#
# ── 최종 조합 (영상 그대로) ──────────────────────────────────────────────────
#   · 4시간봉 켈트너 채널 돌파: 가격이 [EMA n + k x ATR n] 상단을 넘으면 전액 매수,
#     EMA 아래로 내려오면 전량 매도(현금). 숏 없음.
#   · 추세 필터: 가격이 일봉 100일 이동평균선(= 4시간봉 600개 SMA) 위에 있을 때만 진입.
#   · 설정: 비트코인 n=50, k=1.25 / 이더리움 n=60, k=2.0. 자금은 반반, 봉마다 리밸런스 가정.
#   · 비용: 비중이 바뀔 때마다 편도 0.1% (업비트 수수료 0.05% + 슬리피지 가정).
#   · 신호는 봉 종가로 확정하고 **다음 봉부터** 반영합니다(미래참조 방지).
#
# ── 기준일(ASOF) 고정 ────────────────────────────────────────────────────────
#   ASOF 이후 데이터를 자르지 않으면 실행 시점마다 숫자가 달라져 영상과 어긋납니다.
#   최신까지 보고 싶으면 ASOF = None 으로 바꾸세요(영상 숫자와는 달라집니다).
#
# ── 이 계산의 한계 (영상에서도 밝힙니다) ─────────────────────────────────────
#   · 설정값은 과거 데이터 전체(2017-12~2026-08)에서 골랐습니다. 과최적화에서
#     완전히 자유롭지 않고, 미래 수익을 보장하지 않습니다.
#   · 4시간봉 종가 체결을 가정합니다. 실제 체결가는 다를 수 있습니다.
#   · 과거 9년에는 2020~21 암호화폐 초강세장이 포함되어 있습니다.
#   · 거래소가 과거 캔들을 미세 보정하는 경우가 있어, 배수의 마지막 자리에서
#     0.1~0.2 수준의 차이가 날 수 있습니다(예: 54.3배 → 54.4배). 연복리·낙폭은 일치합니다.
# =============================================================================
import csv, datetime, json, os, sys, time, urllib.parse, urllib.request

import numpy as np

try:                                             # 윈도우 기본 콘솔(cp949)에서도 죽지 않게
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
except Exception:
    pass

ASOF = "2026-08-31T12:00:00"        # UTC. 이 봉(포함)까지 사용 — 영상 제작 시점과 동일
COST = 0.001                        # 편도 0.1%
HERE = os.path.dirname(os.path.abspath(__file__))


# ── 데이터 수집 (캐시 → 없으면 API. ASOF 자르기는 양쪽에 똑같이 적용) ────────
def _get(url):
    req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0",
                                               "Accept": "application/json"})
    return json.loads(urllib.request.urlopen(req, timeout=30).read().decode())


def fetch_upbit(market):
    rows, to = [], ""
    while True:
        url = ("https://api.upbit.com/v1/candles/minutes/240?market=" + market +
               "&count=200" + ("&to=" + urllib.parse.quote(to) if to else ""))
        batch = _get(url)
        if not batch:
            break
        for b in batch:
            rows.append((b["candle_date_time_utc"], b["opening_price"],
                         b["high_price"], b["low_price"], b["trade_price"]))
        to = batch[-1]["candle_date_time_utc"]
        time.sleep(0.12)
    rows.reverse()
    return rows


def fetch_binance(symbol):
    rows, start = [], int(datetime.datetime(2017, 8, 1).timestamp() * 1000)
    while True:
        url = ("https://api.binance.com/api/v3/klines?symbol=" + symbol +
               "&interval=4h&limit=1000&startTime=" + str(start))
        batch = _get(url)
        if not batch:
            break
        for b in batch:
            t = datetime.datetime.utcfromtimestamp(b[0] / 1000).strftime("%Y-%m-%dT%H:%M:%S")
            rows.append((t, b[1], b[2], b[3], b[4]))
        start = batch[-1][0] + 1
        if len(batch) < 1000:
            break
        time.sleep(0.15)
    return rows


def load(tag, fetch):
    path = os.path.join(HERE, "cache_" + tag + ".csv")
    if not os.path.exists(path):
        print("  " + tag + " 수집 중…")
        rows = fetch()
        with open(path, "w", newline="", encoding="utf-8") as f:
            w = csv.writer(f)
            w.writerow(["date", "open", "high", "low", "close"])
            w.writerows(rows)
    ts, o, h, l, c = [], [], [], [], []
    for r in csv.DictReader(open(path, encoding="utf-8")):
        if ASOF and r["date"] > ASOF:          # ★ 캐시든 신규든 같은 기준으로 자른다
            continue
        ts.append(r["date"])
        o.append(float(r["open"])); h.append(float(r["high"]))
        l.append(float(r["low"])); c.append(float(r["close"]))
    return ts, np.array(o), np.array(h), np.array(l), np.array(c)


# ── 지표 ─────────────────────────────────────────────────────────────────────
def ema(x, n):
    a = 2.0 / (n + 1)
    out = np.empty(len(x)); out[0] = x[0]
    for i in range(1, len(x)):
        out[i] = a * x[i] + (1 - a) * out[i - 1]
    return out


def atr(h, l, c, n):
    pc = np.r_[c[0], c[:-1]]
    tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc)))
    out = np.full(len(tr), np.nan)
    if len(tr) >= n:
        out[n - 1] = tr[:n].mean()
        for i in range(n, len(tr)):
            out[i] = out[i - 1] + (tr[i] - out[i - 1]) / n
    return out


def sma(x, n):
    out = np.full(len(x), np.nan)
    if len(x) >= n:
        cs = np.cumsum(np.insert(x, 0, 0.0))
        out[n - 1:] = (cs[n:] - cs[:-n]) / n
    return out


# ── 전략: 켈트너 돌파 + 일봉 100일선 필터 ────────────────────────────────────
# 켈트너 '상태'(돌파~이탈)와 필터를 따로 두고 곱한다. 필터 때문에 나갔다가
# 가격이 100일선 위로 돌아오면, 켈트너 상태가 살아 있는 한 다시 들어간다.
def signal(o, h, l, c, n, k):
    e = ema(c, n); a = atr(h, l, c, n)
    f100 = sma(c, 600)                          # 4시간봉 600개 = 100일
    sig = np.zeros(len(c)); base = 0.0
    for i in range(len(c)):
        if np.isnan(a[i]):
            sig[i] = 0.0; continue
        if c[i] > e[i] + k * a[i]:              # 돌파 → 상태 ON (같은 봉에선 매수 우선)
            base = 1.0
        elif c[i] < e[i]:                        # EMA 이탈 → 상태 OFF
            base = 0.0
        ok = (not np.isnan(f100[i])) and c[i] > f100[i]
        sig[i] = base if ok else 0.0
    if len(sig) > n:
        sig[:n + 1] = 0.0                        # 워밍업
    return sig


def growth(c, sig):
    ret = np.zeros(len(c)); ret[1:] = c[1:] / c[:-1] - 1
    pos = np.r_[0.0, sig[:-1]]                  # 다음 봉 반영
    return pos, (1 + pos * ret) * (1 - COST * np.abs(np.diff(np.r_[0.0, pos])))


def stats(g, ts):
    eq = np.cumprod(g)
    yrs = (datetime.date.fromisoformat(ts[-1][:10])
           - datetime.date.fromisoformat(ts[0][:10])).days / 365.25
    mdd = float((eq / np.maximum.accumulate(eq) - 1).min())
    return float(eq[-1] ** (1 / yrs) - 1), mdd, float(eq[-1])


def run(tag, btc, eth):
    tb, ob, hb, lb, cb = btc
    te, oe, he, le, ce = eth
    WARM = 400
    _, gb = growth(cb, signal(ob, hb, lb, cb, 50, 1.25))
    _, ge = growth(ce, signal(oe, he, le, ce, 60, 2.0))
    ib_map = {t: i for i, t in enumerate(tb)}
    pairs = [(ib_map[t], i) for i, t in enumerate(te) if t in ib_map]
    ib = np.array([x[0] for x in pairs]); ie = np.array([x[1] for x in pairs])
    m = (ib >= WARM) & (ie >= WARM)
    gc = 0.5 * gb[ib[m]] + 0.5 * ge[ie[m]]
    dl = [te[i] for i in ie[m]]
    cagr, mdd, mult = stats(gc, dl)
    print("\n[" + tag + "]  " + dl[0][:10] + " ~ " + dl[-1][:10])
    print("  반반 조합: 연복리 {:+.1%} · 최대낙폭 {:+.1%} · {:.1f}배 (1000만원 → {:,.0f}만원)"
          .format(cagr, mdd, mult, mult * 1000))
    yr = {}
    for i, t in enumerate(dl):
        yr[t[:4]] = yr.get(t[:4], 1.0) * gc[i]
    print("  연도별: " + " · ".join(y + " {:+.0f}%".format((v - 1) * 100)
                                  for y, v in sorted(yr.items())))


if __name__ == "__main__":
    print("전략 1,887개 발굴 편 — 최종 조합 재현 (기준일 " + str(ASOF) + " UTC)")
    up_btc = load("upbit_btc", lambda: fetch_upbit("KRW-BTC"))
    up_eth = load("upbit_eth", lambda: fetch_upbit("KRW-ETH"))
    run("업비트 원화", up_btc, up_eth)
    try:
        bn_btc = load("binance_btc", lambda: fetch_binance("BTCUSDT"))
        bn_eth = load("binance_eth", lambda: fetch_binance("ETHUSDT"))
        run("바이낸스 USDT", bn_btc, bn_eth)
    except Exception as e:
        print("\n[바이낸스] 수집 실패(" + str(e) + ") — 지역에 따라 API 가 막힐 수 있습니다. 업비트 결과만 확인하세요.")
    print("\n영상 수치: 업비트 +58.0% / -29.6% / 54.3배 · 바이낸스 +41.7% / -29.4%")