backtest.py

import ccxt # ccxt(여러 거래소의 API를 받아오는 라이브러리) 호출
import pandas as pd #pandas(데이터 분석 라이브러리) 호출, 앞으로 pandas를 pd로 부를것을 명시(alias,별칭)
import numpy as np #numpy(수치 계산 라이브러리. 행렬 등이 포함되어있음)호출, 앞으로 numpy를 np로 부를 것을 명시함
import matplotlib.pyplot as plt #matplotlib(그래프 시각화 라이브러리) 호출, 앞으로 matplotlib를 plt으로 부를 것을 명시함
from datetime import datetime #datetime(날짜와 시간 객체를 다루는 기본 파이썬 모듈)을 호출
import pytz #pytz(전세계 타임존을 다룰 수 있는 라이브러리) 호출, 용도는 바이낸스에서 받은 UTC time을 한국 시간으로 변환하기 위해 사용함

from sklearn.pipeline import Pipeline #머신러닝 라이브러리인 scikit-learn(sklearn)에서 Pipeline(여러 단계를 한번에 묶어서 실행하기 위함)을 호출함.
from sklearn.preprocessing import StandardScaler #머신러닝 라이브러리인 scikit-learn(sklearn)에서 StandardScaler(값을을 표준화(평균 0,표준편차 1로 작업, 통계학에서 계산을 편리하기 위해 사용하는 스킬))를 호출함.
from sklearn.linear_model import LogisticRegression #머신러닝 라이브러리인 scikit-learn(sklearn)에서 LogisticRegression(로지스틱 회귀를 사용하기 위해 호출)을 호출함.
from sklearn.metrics import classification_report, roc_auc_score, confusion_matrix, accuracy_score, precision_score, recall_score #머신러닝 라이브러리인 scikit-learn(sklearn)의 metrics(모델 성능 평가 도구)의 각종 도구들을 호출함.
# classification_report(Precision(정밀도),Recall(재현율),F1 score(정밀도와 재현율의 조화평균)등에 대한 요약 리포트)
# roc_auc_score(Roc-Auc 값을 나타냄. 이 값들은 이진 분류의 성능을 보여주는 척도임)
# confusion_matrix(혼동행렬(TP,TF,NP,NF))
# accuracy_score(정확도),precision_score(정밀도),recall_score(재현율/민감도)
SYMBOL    = "ETH/USDT"   #어떤 코인 거래쌍을 쓸지 지정하는 변수. 원하는 심볼로 변경 가능 (BTC/USDT 등)
TIMEFRAME = "1h" # 캔들 간격(봉 주기) 지정
SINCE     = int(pd.Timestamp("2023-01-01", tz="UTC").timestamp() * 1000) #데이터 수집 시작 지점, 바이낸스는 밀리초 단위를 요구하기때문에 밀리초로 변환
LIMIT     = 1000 #한번 호출할 때 최대 몇개의 봉을 받을지 지정. 거래소마다 제한이 있는데, 바이낸스는 보통 최대 1000이여서 1000으로 설정

ex = ccxt.binance() #ccxt 라이브러리로 바이낸스 거래소 객체 생성, 이 객체를 통해 시세 데이터를 가져올 수 있음

def fetch_ohlcv_all(symbol=None, timeframe=None, since_ms=None, limit=1000): #원하는 심볼(symbol), 타임프레임(timeframe), 시작시간(since), 제한(limit)을 넣으면 여러번 호출해서 전체 데이터를 이어붙이는 함수
    symbol    = symbol    or SYMBOL #함수 호출시 symbol에 대한 값을 주지 않으면, 위에서 지정한 SYMBOL 전역 변수를 기본 값으로 사용
    timeframe = timeframe or TIMEFRAME #함수 호출시 timeframe에 대한 값을 주지 않으면, 위에서 지정한 TIMEFRAME 전역 변수를 기본 값으로 사용
    since_ms  = since_ms  or SINCE #함수 호출시 since에 대한 값을 주지 않으면, 위에서 지정한 SINCE 전역 변수를 기본 값으로 사용

    all_rows, cursor = [], since_ms #all_rows = 데이터를 계속 쌓아둘 리스트, cursor = 현재 불러올 위치(타임 스탬프), 처음에는 since부터 시작
    while True: #ccxt의 fetch_ohlcv를 반복 호출 -> 1회 최대 1000개까지 불러오므로(LIMIT에 의해서) while 루프를 통해서 전체 데이터를 수집하기 위함
        rows = ex.fetch_ohlcv(symbol, timeframe=timeframe, since=cursor, limit=limit) #cctx의 fetch_ohlcv를 반복 호출 -> 1회 최대 1000개까지 불러오므로(LIMIT에 의해서) while 루프를 통해서 전체 데이터를 수집하기 위함
        if not rows: #불러온 데이터가 없으면
            break #루프 종료
        all_rows.extend(rows) #이번에 불러온 데이터를 all_rows 리스트에 추가
        cursor = rows[-1][0] + 1 #마지막 봉의 timestamp를 기준으로 +1해서 다음 구간부터 다시 요청 ->중복 방지, 데이터가 겹치지 않고 이어지게 하기 위함
        if len(rows) < limit: #거래소에서 받아온 데이터가 limit보다 적으면 = 더이상 받아올 데이터가 없음을 의미
            break #그러므로 루프 종료
    df = pd.DataFrame(all_rows, columns=["timestamp","open","high","low","close","volume"]) #쌓아둔 데이터를 pandas DataFrame으로 변환, 열 이름을 timestamp, open, high, low, close, volume으로 지정
    df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True).dt.tz_convert("Asia/Seoul") #timestamp(밀리초)를 datetime 형식으로 변환, UTC 기준으로 읽은 뒤->한국 시간(Asia/Seoul)로 변환
    return df.sort_values("timestamp").reset_index(drop=True) #시간 순으로 정렬 + 인덱스 재설정 -> 최종 dataFrame 반환

df = fetch_ohlcv_all() #위에서 정의한 fetch_ohlcv_all 함수를 실행
print("SYMBOL =", SYMBOL) #현재 심볼이 무엇인지 출력
print("데이터 개수:", len(df)) #데이터가 몇개인지 출력
print(df.head()) #앞부분 5행만 미리 보기. 전체 데이터가 많기때문에,앞부분 샘플만 빠르게 확인하기 위함

def rsi(series, period=14): #RSI(Relative Strength Index) 계산 함수를 호출, 함수의 괄호 안에 들어가는 것은 매개변수들이고, 데이터(시계열)와 기간을 받는데, series는 종가 시계열을 의미하고, period의 수로는 기본값인 14 사용
    delta = series.diff() #한 칸 차분을 통해 직전 값과의 변화율을 계산
    up = delta.clip(lower=0).rolling(period).mean() #상승한 값만 남긴 뒤에, period 동안의 평균을 구함
    down = (-delta.clip(upper=0)).rolling(period).mean()#하락한 값만 남긴 뒤에, period 동안의 평균을 구함
    rs = up / (down + 1e-12) #상승/하락 비율을 구함. (분모가 0이 되는걸 방지하기 위해 1e-12를 추가함)
    return 100 - (100 / (1 + rs)) #값 반환, 반환하는 공식은 전형적인 RSI 공식 = 100 - (100/(1+rs))

def make_features_and_label(df_raw): #원본 데이터(df_raw)를 가공 시작
    df = df_raw.copy() #원본 데이터(df_raw) 복사
    df["ret1"]   = np.log(df["close"]).diff() #수익률 피처(종가의 로그 차분) (한시간 전 대비 수익률)
    df["sma7"]   = df["close"].rolling(7).mean() #단기 이동 평균선(7봉 평균)
    df["sma21"]  = df["close"].rolling(21).mean() #중기 이동 평균선(21봉 평균)
    df["rsi14"]  = rsi(df["close"], 14)#RSI(14) -> 모멘텀 지표 -> 과매도/과매수 확인
    df["vol14"]  = df["ret1"].rolling(14).std() # 변동성 확인(14봉 기준 로그수익률 표준편차)
    df["body"]   = df["close"] - df["open"] #캔들 바디 길이(시가 대비 종가 상승폭(양봉/음봉크기))
    df["range"]  = df["high"] - df["low"] #캔들 전체 범위(고가-저가) = 변동 폭
    df["label"]  = (df["close"].shift(-1) > df["close"]).astype(int) #라벨 작업, 다음봉 종가가 지금 종가보다 크면 1, 작거나 같으면 0 = 이게 우리가 예측할 목표값(y)
    df = df.dropna().reset_index(drop=True) #이동평균/변동성 계산때문에 생긴 NaN 행 제거, 인덱스 리셋
    return df #최종 DataFrame 반환

df_feat = make_features_and_label(df) #가공된 데이터셋(df_feat) 생성
print(df_feat.tail()) #마지막 5행 확인

from sklearn.metrics import classification_report, confusion_matrix, accuracy_score, precision_score, recall_score #성능 평가 지표 함수들 호출

y_pred = (proba_te > 0.5).astype(int) #테스트셋에서 모델이 예측한 상승 확률 = proba_te, 확률이 0.5보다 클때 상승으로 예측(1 반환), 아니면 하락으로 예측(0 반환) -> 이진 분류 예측 결과(y_pred) 생성

tn, fp, fn, tp = confusion_matrix(y_te, y_pred).ravel() #혼동 행렬 계산

# 지표 계산
정확도   = accuracy_score(y_te, y_pred)
정밀도   = precision_score(y_te, y_pred, zero_division=0)
재현율   = recall_score(y_te, y_pred, zero_division=0)   # = 민감도
민감도   = 재현율
특이도   = tn / (tn + fp) if (tn+fp) > 0 else 0.0

print("=== 분류 성능 지표 ===")
print(f"정확도(Accuracy)   : {정확도:.3f}")
print(f"정밀도(Precision)  : {정밀도:.3f}")
print(f"재현율(Recall)     : {재현율:.3f}")
print(f"민감도(Sensitivity): {민감도:.3f}")
print(f"특이도(Specificity): {특이도:.3f}")

print("\n[상세 분류 리포트]")
print(classification_report(y_te, y_pred, target_names=["하락(0)", "상승(1)"]))

FEE = 0.0005              # 전환 비용 (거래 수수료, 왕복 0.5%로 가정)
FUNDING_PER_HOUR = 0.0    # 펀딩/차입 비용 (선물 가정 시)
MAX_POS = 1.0             # 최대 포지션 크기 (롱+1, 숏 -1, 현금 0)
STEP_PER_BAR = 0.25       # 분할 강도
MIN_TRADE = 0.05          # 최소 체결 크기(이보다 작은 체결은 생략)
BAND = 0.02               # 히스테리시스(목표와 현재 차이가 너무 작으면 무시)

# 확률 → 목표 포지션 매핑
# p>목표 구매 수치 → 롱, p<목표 판매 수치 → 숏, 나머지는 현금
def prob_to_target_position(p, max_pos=MAX_POS):
    if p > 0.75: # 목표 구매 수치보다 높을때 구매하는 로직
        scale = (p - 0.75) / 0.25 #구매 강도를 결정하는 로직, 목표 구매 수치보다 확률이 높은 정도를 통해서 0~1 사이의 값만큼 구매 스케일을 결정함
        return np.clip(scale * max_pos, 0.0, max_pos) #결정된 스케일에 맞게 실제 포지션을 잡게끔 함
    elif p < 0.25: #목표 판매 수치보다 낮을때 숏 포지션을 잡게끔 하는 로직
        scale = (0.25 - p) / 0.25 # 숏 강도를 결정하는 로직, 목표 판매 수치보다 낮은 정도를 0~1 사이의 값으로 설정해서 숏 스케일 결정
        return -np.clip(scale * max_pos, 0.0, max_pos) # 결정된 스케일에 맞게 실제 포지션을 잡게끔 함
    else:
        return 0.0 # 중립 구간이면 현금 상태

test = test.copy() # 원본 test DataFrame 복사 (안전하게 사용하기 위함)
test["proba"] = proba_te # 예측된 상승 확률을 proba 컬럼에 추가
ret = test["ret1"].fillna(0) # ret1(수익률)에서 NaN을 0으로 채움

target_now = test["proba"].apply(prob_to_target_position) # 각 시점의 확률 → 목표 포지션 값으로 변환

# 분할 진입/청산 로직
position = np.zeros(len(test)) # 포지션 기록용 배열 (초기값 0)
for i in range(1, len(test)):
    prev = position[i-1] # 이전 시점의 포지션
    tgt  = target_now.iloc[i-1]  # 목표 포지션 (확률 기반), 체결은 다음 봉
    gap  = tgt - prev # 목표와 현재 포지션의 차이
    if abs(gap) < BAND: # 차이가 너무 작으면 무시
        position[i] = prev
        continue
    step = np.clip(gap, -STEP_PER_BAR*MAX_POS, STEP_PER_BAR*MAX_POS) # 한 번에 이동할 수 있는 최대 폭 제한 (분할매매)
    if abs(step) < MIN_TRADE: # 너무 작은 거래는 생략
        position[i] = prev
    else:
        position[i] = prev + step # 조정된 포지션 반영

test["position"] = position # 최종 포지션 배열을 test DataFrame에 저장

# 거래 비용/펀딩 반영
turnover = np.abs(np.diff(np.r_[0.0, position])) # 체결량(포지션 변동 절댓값)
funding_cost = FUNDING_PER_HOUR * np.abs(position) # 포지션 유지 시 펀딩비

net_ret = (position * ret) - (turnover * FEE) - funding_cost # 실제 전략 수익 = (보유포지션×시장수익률) - 수수료 - 펀딩비
equity = (1 + net_ret).cumprod() # 전략의 누적 자산 곡선
bh_equity = (1 + ret).cumprod() # 단순 매수(Buy&Hold) 누적 자산 곡선

def max_drawdown(curve):
    peak = curve.cummax() # 지금까지의 최고점
    dd = (curve/peak - 1.0) # 낙폭 계산
    return dd.min() # 가장 큰 낙폭(최대 손실률)

mdd_strategy = max_drawdown(equity) # 전략의 최대낙폭
mdd_bh = max_drawdown(bh_equity)# 단순매수의 최대낙폭
ann_factor = np.sqrt(24*252) # Sharpe 연율화 계수 (24시간봉 × 252거래일)
sharpe_strategy = (net_ret.mean() / (net_ret.std() + 1e-12)) * ann_factor # 전략 Sharpe ratio, Sharpe Ratio = (투자 수익률 − 무위험 수익률) ÷ 수익률의 표준편차 = 수익률 대비 위험(변동성)을 얼마나 효율적으로 감수했는가?
sharpe_bh = (ret.mean() / (ret.std() + 1e-12)) * ann_factor # 단순매수 Sharpe ratio

print(f"Final equity (strategy): {equity.iloc[-1]:.3f}x") # 전략 최종 수익 배수
print(f"Final equity (B&H)     : {bh_equity.iloc[-1]:.3f}x") # 단순 매수 최종 수익 배수
print(f"MDD (strategy): {mdd_strategy:.2%} | MDD (B&H): {mdd_bh:.2%}") # 최대낙폭 출력
print(f"Sharpe (strategy): {sharpe_strategy:.2f} | Sharpe (B&H): {sharpe_bh:.2f}") # Sharpe ratio 출력

plt.figure(figsize=(12,6)) # 그림 생성, 크기는 가로 12, 세로 6인치
plt.plot(test["timestamp"], equity, label="Strategy") #x축엔 timestamp(시간), y축엔 equity(누적 자산 곡선), 범례엔 Strategy로 표시
plt.plot(test["timestamp"], bh_equity, label="Buy & Hold", alpha=0.8) # Buy & Hold(단순 매수 후 보유) 곡선을 비교군으로 추가, alpha값을 0.8로 해서(약간 투명하게 하기 위함) 전략 곡선이 더 잘보이게 설정
plt.title(f"Equity Curve — {SYMBOL} {TIMEFRAME}") # 그래프 제목 표시
plt.xlabel("Time (KST)") #X축 라벨 : 시간
plt.ylabel("Equity (x)")#Y축 라벨 : 자산 배수
plt.legend() #위에서 label로 지정한 "Strategy", "Buy & Hold" 표시
plt.grid(True) #격자선 표시 (눈금 보기 편하게)
plt.show() #그래프 최종 출력