45억 795만원을 만들고도 탈락한 전략,
장치 하나로 통과시켰습니다

# -*- coding: utf-8 -*-
"""
AlgoLab 공개 재현 코드 — FVG 되돌림 전략 + 변동성 타겟 사이징
=============================================================
영상: 「45억 만든 자동매매 전략, 탈락시킨 뒤 살렸습니다」

이 파일 하나만 있으면 영상의 모든 숫자를 처음부터 다시 만들 수 있습니다.
필요한 것: python 3.9+, numpy. 데이터는 실행할 때 거래소 공개 API 로 직접 받습니다.

    pip install numpy
    python fvg_voltarget.py

▣ 기준일 고정 (ASOF)
  영상은 2026-09-05 까지의 데이터로 만들었습니다. 오늘 그냥 돌리면 그 뒤 캔들이
  더 붙어서 숫자가 미세하게 달라집니다. 그래서 기준일을 아래에 못 박았습니다.
  최신 데이터로 보고 싶으면 ASOF 를 None 으로 바꾸세요.

▣ 전략 규칙 (영상에서 말한 그대로, 전부 여기 있습니다)
  1. 상승 갭(FVG)이 생기고 20봉 안에 가격이 그 갭 자리로 되돌아오면 산다
  2. 종가가 갭 하단 아래로 내려가거나 20봉 지수이동평균 아래로 내려가면 나온다
  3. 일봉 50일 이동평균선 위에 있을 때만 산다
  4. 최근 42봉 실현변동성(연율)이 목표 50%를 넘으면 그 비율만큼 비중을 줄인다  ← 이번 편의 장치

▣ 잣대 (결과를 보고 바꾸지 않습니다)
  업비트·바이낸스 두 시장 모두 연복리 30% 이상 · 최대낙폭 -35% 이내 · 편도 거래비용 0.1%
  두 시장 중 나쁜 쪽으로 채점합니다.

▣ 한계
  과거 데이터 백테스트입니다. 슬리피지·부분체결·세금은 넣지 않았습니다.
  승률은 30% 수준입니다(열 번 중 일곱 번은 집니다). 미래 수익을 보장하지 않습니다.
"""
import datetime
import json
import os
import sys
import time
import urllib.request

import numpy as np

sys.stdout.reconfigure(encoding="utf-8", errors="replace")

ASOF = datetime.date(2026, 9, 5)   # 영상을 만든 날. None 이면 오늘까지
CACHE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "_data")
FEE = 0.001            # 편도 0.1%
LOOK, EXIT_N = 20, 20  # 갭 되돌림 유효 봉수 / 청산 EMA
FILT_DAYS = 50         # 일봉 이동평균 필터
VT_WIN, VT_TARGET = 42, 0.5   # 변동성 창(봉) / 목표 연율 변동성
WARM = 400             # 지표 워밍업 — 성적은 이 뒤부터만 잰다
VAL_START = "2024-01-01"       # 검증 구간 시작
BARS_PER_DAY = 6       # 4시간봉


# ══════════════════════════════════════════════════════════════════════
# 1) 데이터 — 업비트(원화) · 바이낸스(달러) 4시간봉
# ══════════════════════════════════════════════════════════════════════
def _get(url):
    req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
    return json.loads(urllib.request.urlopen(req, timeout=30).read().decode())


def fetch_upbit(market):
    """업비트 240분봉을 최신부터 역방향으로 긁는다."""
    rows, to = [], ""
    while True:
        url = f"https://api.upbit.com/v1/candles/minutes/240?market={market}&count=200"
        if to:
            url += "&to=" + to
        batch = _get(url)
        if not batch:
            break
        for b in batch:
            rows.append([b["candle_date_time_kst"], b["opening_price"],
                         b["high_price"], b["low_price"], b["trade_price"]])
        to = batch[-1]["candle_date_time_utc"]
        if len(batch) < 200 or to < "2017-01-01":
            break
        time.sleep(0.12)
    rows.sort(key=lambda r: r[0])
    return rows


def fetch_binance(symbol):
    """바이낸스 4h klines 를 오래된 쪽부터 긁는다."""
    rows, start = [], int(datetime.datetime(2017, 8, 1).timestamp() * 1000)
    while True:
        batch = _get(f"https://api.binance.com/api/v3/klines?symbol={symbol}"
                     f"&interval=4h&limit=1000&startTime={start}")
        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, float(b[1]), float(b[2]), float(b[3]), float(b[4])])
        if len(batch) < 1000:
            break
        start = batch[-1][0] + 1
    return rows


def load(tag, getter):
    os.makedirs(CACHE, exist_ok=True)
    path = os.path.join(CACHE, tag + ".json")
    if os.path.exists(path):
        rows = json.load(open(path, encoding="utf-8"))
    else:
        print(f"  {tag} 내려받는 중…")
        rows = getter()
        json.dump(rows, open(path, "w", encoding="utf-8"))
    # ⚠ 자르기는 캐시를 쓰든 새로 받든 똑같이 적용한다(한쪽에만 걸면 두 번째 실행부터 달라진다)
    if ASOF:
        cut = ASOF.isoformat() + "T23:59:59"
        rows = [r for r in rows if r[0] <= cut]
    return rows


def series(rows):
    d = [r[0] for r in rows]
    o, h, l, c = (np.array([r[i] for r in rows], float) for i in (1, 2, 3, 4))
    n = len(c)
    ret = np.zeros(n)
    ret[1:] = c[1:] / c[:-1] - 1
    yrs = (datetime.date.fromisoformat(d[-1][:10])
           - datetime.date.fromisoformat(d[0][:10])).days / 365.25
    return dict(d=d, o=o, h=h, l=l, c=c, n=n, ret=ret, ppy=n / yrs,
                val_i=int(np.searchsorted(np.asarray(d), VAL_START)))


# ══════════════════════════════════════════════════════════════════════
# 2) 전략 — 규칙 1·2·3
# ══════════════════════════════════════════════════════════════════════
def ema(x, n):
    out = np.empty(len(x), float)
    k = 2.0 / (n + 1)
    out[0] = x[0]
    for i in range(1, len(x)):
        out[i] = x[i] * k + out[i - 1] * (1 - k)
    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


def fvg_pullback(o, h, l, c, look=LOOK, exit_n=EXIT_N):
    """상승 갭(3봉: l[t] > h[t-2]) 뒤 look봉 안에 갭 구간으로 되돌아오면 매수.
    청산: 종가가 갭 하단 아래 또는 종가 < EMA(exit_n)."""
    n = len(c)
    e = ema(c, exit_n)
    sig = np.zeros(n)
    pos = 0.0
    z_top = z_bot = np.nan
    z_age = 10 ** 9
    for t in range(2, n):
        if pos == 0.0:
            if l[t] > h[t - 2]:
                z_top, z_bot, z_age = l[t], h[t - 2], 0
            else:
                z_age += 1
            if z_age <= look and not np.isnan(z_top):
                if l[t] <= z_top and c[t] > z_bot:
                    pos = 1.0
                    z_age = 10 ** 9
        else:
            if (not np.isnan(z_bot) and c[t] < z_bot) or c[t] < e[t]:
                pos = 0.0
                z_top = z_bot = np.nan
        sig[t] = pos
    return sig


def with_filter(S):
    """규칙 3 — 일봉 50일선 위에서만 산다(4시간봉이므로 50×6봉)."""
    base = fvg_pullback(S["o"], S["h"], S["l"], S["c"])
    m50 = sma(S["c"], FILT_DAYS * BARS_PER_DAY)
    f = np.zeros(S["n"])
    ok = ~np.isnan(m50)
    f[ok] = (S["c"][ok] > m50[ok]).astype(float)
    return base * f


# ══════════════════════════════════════════════════════════════════════
# 3) 이번 편의 장치 — 변동성 타겟 사이징 (규칙 4)
# ══════════════════════════════════════════════════════════════════════
def vol_target(S, sig, target=VT_TARGET, win=VT_WIN):
    """최근 win봉 실현변동성(연율)이 target 을 넘으면 그 비율만큼 비중을 줄인다.
    변동성 100%면 비중 0.5, 200%면 0.25. 조용하면 그대로 1.0."""
    ppy = int(round(S["ppy"]))
    r, n = S["ret"], S["n"]
    w = np.ones(n)
    for t in range(win, n):
        v = r[t - win + 1:t + 1].std() * np.sqrt(ppy)
        w[t] = min(1.0, target / v) if v > 1e-9 else 1.0
    return sig * w


# ══════════════════════════════════════════════════════════════════════
# 4) 채점
# ══════════════════════════════════════════════════════════════════════
def growth(S, sig, cost=FEE):
    """신호는 **다음 봉**에 반영한다(같은 봉 종가로 사는 미래참조 방지)."""
    pos = np.r_[0.0, sig[:-1]]
    return pos, (1 + pos * S["ret"]) * (1 - cost * np.abs(np.diff(np.r_[0.0, pos])))


def stats(g, ppy):
    eq = np.cumprod(g)
    yrs = len(g) / ppy
    return dict(cagr=eq[-1] ** (1 / yrs) - 1 if eq[-1] > 0 else -1.0,
                mdd=float((eq / np.maximum.accumulate(eq) - 1).min()),
                mult=float(eq[-1]))


def trades(pos, g):
    inpos = pos > 0
    edges = np.diff(inpos.astype(int))
    starts = list(np.where(edges == 1)[0] + 1)
    ends = list(np.where(edges == -1)[0] + 1)
    if inpos[0]:
        starts = [0] + starts
    if len(ends) < len(starts):
        ends.append(len(g) - 1)
    rets = np.array([np.prod(g[s:e + 1]) - 1 for s, e in zip(starts, ends)])
    if not len(rets):
        return 0, float("nan")
    return len(rets), float((rets > 0).mean())


def measure(S, sig):
    pos, g = growth(S, sig)
    st = stats(g[WARM:], S["ppy"])
    stv = stats(g[S["val_i"]:], S["ppy"])
    n, win = trades(pos[WARM:], g[WARM:])
    st.update(val_cagr=stv["cagr"], val_mdd=stv["mdd"], 거래=n, 승률=win,
              비중변경=int(np.sum(np.abs(np.diff(sig[WARM:])) > 1e-9)),
              구간=f'{S["d"][WARM][:10]}~{S["d"][-1][:10]}')
    return st


# ══════════════════════════════════════════════════════════════════════
def main():
    print(f"기준일 ASOF = {ASOF or '오늘'}\n")
    market = {
        "eth": [("업비트", lambda: fetch_upbit("KRW-ETH"), "up_eth"),
                ("바이낸스", lambda: fetch_binance("ETHUSDT"), "bi_eth")],
        "btc": [("업비트", lambda: fetch_upbit("KRW-BTC"), "up_btc"),
                ("바이낸스", lambda: fetch_binance("BTCUSDT"), "bi_btc")],
    }
    for asset in ("eth", "btc"):
        S, BASE = {}, {}
        for tag, getter, key in market[asset]:
            S[tag] = series(load(key, getter))
            BASE[tag] = with_filter(S[tag])

        print(f"\n{'═' * 78}\n  {asset.upper()}  ({S['업비트']['d'][WARM][:10]} ~ "
              f"{S['업비트']['d'][-1][:10]}, 4시간봉, 편도 {FEE * 100:.1f}%)\n{'═' * 78}")
        print(f"  {'':22s}{'연복리':>8}{'최대낙폭':>10}{'배수':>11}"
              f"{'검증 연복리':>12}{'비중변경':>9}")
        for label, make in (("장치 없음 (원래 전략)", lambda t: BASE[t]),
                            ("+ 변동성 타겟 0.5", lambda t: vol_target(S[t], BASE[t]))):
            got = {}
            for tag, _, _ in market[asset]:
                m = measure(S[tag], make(tag))
                got[tag] = m
                print(f"  {label if tag == '업비트' else '':22s}"
                      f"{tag:6s}{m['cagr'] * 100:7.1f}%{m['mdd'] * 100:9.1f}%"
                      f"{m['mult']:10.1f}배{m['val_cagr'] * 100:11.1f}%{m['비중변경']:9d}")
            mn = min(got[t]["cagr"] for t in got)
            wo = min(got[t]["mdd"] for t in got)
            ok = "통과" if (mn >= 0.30 and wo >= -0.35) else "탈락"
            print(f"  {'':22s}→ 나쁜 쪽 연복리 {mn * 100:.1f}% · "
                  f"나쁜 쪽 최대낙폭 {wo * 100:.1f}%  ⇒ 잣대 {ok}")
            print(f"  {'':22s}   거래 {got['업비트']['거래']}건 · "
                  f"승률 {got['업비트']['승률'] * 100:.1f}% (업비트 기준)\n")

    print("\n※ 영상에서 말한 '장치 22개'의 나머지(트레일링·샹들리에·분할청산·분할진입·서킷브레이커)는")
    print("   설명란의 표에 결과를 전부 실었습니다. 낙폭을 기준선 안으로 넣은 것은 변동성 타겟 하나입니다.")


if __name__ == "__main__":
    main()