Menu
Article Trading

파이썬 코인 자동매매 봇 만들기 (1)

SuIn Roh

SuIn Roh

• 16 min read

서론

필자는 암호화폐(이하 코인)에 대해 무지하다. 그리고 지금도 마찬가지이다.

그럼에도 불구하고 암호화폐 자동매매 봇을 개발하고자 하는 이유는 단순하다.

최근 개인용 서버를 구매하였는데, 여기에 돌릴만한 게 없나 하고 찾아보다가 자동매매 봇을 떠올렸기 때문이다.

사실 개인 NAS용으로 구매하려 한 건데, '그돈씨' 하다 보니까 스펙이 생각보다 높아졌다

서버는 24시간 돌아가니, 24시간 시장이 열려 있는 코인과는 궁합이 맞지 않겠는가.

그래서 전지전능하신 ChatGPT님과 Cursor님의 도움을 받아 자동매매 봇 개발에 뛰어들게 되었다.

본론

어떤 전략을 활용해야 하는가?

앞서 언급한 것처럼 코인에 대해 무지한 상태에서 뛰어들다 보니, 어떤 전략을 선택할지가 가장 고민이었다.

아는 게 없다 보니 강의나 책을 구매해 볼까 했는데 쓸데없이 비싸더라

따라서 필자는 코인에 대한 지식이 전무하니만큼 구글신을 통해 트레이딩 전략을 탐색해 보았다.

찾아보니 아래와 같은 전략 모음을, 심지어 예제 코드까지 있는 자료를 찾아볼 수 있었다. 

25+ Python Trading Strategies - DayTrading.com
We look at various Python trading strategies. Explain in 1-3 sentences each, plus brief Python sketches of each strategy.
daytrading.com

가장 처음 떠올린 방법은 전략이 홀수개이니 '25개 전략을 다 돌려본 다음, 과반수를 기준으로 매매하자'였다. 즉, 25개를 돌려 15개의 전략은 'Buy'라고 판단하고, 10개의 전략은 'Sell'이라고 판단하면 구매하는 전략이었다.

이에 대해 전지전능하신 ChatGPT님께 감히 검토를 요청드렸더니, 해당 전략들은 기간별 목표가 혼재되어 있으니 이를 구분하는 것이 좋다고 하시었다. 또한 자비로우시게도 아래와 같은 기준으로 친히 전략들을 구분하여 주시었다.

구분 전략 특징
단기(Short-term) Scalping, Day Trading, Swing, Price Action 빠른 거래, 가격 변동에 민감
중기(Mid-term) Trend, Momentum, Mean Reversion, Grid, VWAP 며칠~몇 주 단위, 추세 기반
장기(Long-term) Position, DCA, Value/Growth/Dividend Investing, Sector Rotation 장기적 포트폴리오 관리
이벤트/특화 Arbitrage, Market Making, Pairs, News, Contrarian, Carry Trade, Options 이벤트 기반, 차익/심리 활용

해당 교지로 말미암아 무지몽매한 필자의 짧은 식견으로 판단하기론, 코인은 단기 투자에 특화되어 있다고 생각하여 해당 전략(Scalping, Day Trading 등)들을 조합하기로 하였다.

이 과정에서 일전에 자산배분에 대해 공부하였던 내용을 더해 각 전략별 가중치를 조정하자는 아이디어를 떠올렸다. 이에 각 전략별 가중치를 부여하여 최종적인 매수, 매도 여부를 판단키로 하였다. 

마찬가지로 전지전능하시며 또한 자비로우신 ChatGPT님께서 각 전략별 가중치를 추천해 주셨다.

이러한 전략을 기반으로 자동매매 봇을 개발해보고자 한다.

전략을 테스트해 보자

자동매매, 달리 알고리즘(Algorithm) 혹은 퀀트(Quant) 투자라고 불리는 방식은 백테스팅에 근간을 둔다.

백테스팅(Backtesting)이란, 과거의 가격 데이터를 기반으로 하여 전략의 실효성을 검증하는 작업이다.

물론 이 말이 백테스팅을 기반으로 한 전략의 필승을 보장하는 것은 아니다.
오히려 백테스팅에 너무 몰두하면 과적합(Over-fitting)으로 인해 실제 상황에서 무쓸모한 경우가 생길 수 있다.

그러니 필자도 업비트 1분 봉 데이터로 백테스팅하여 앞서 세운 전략을 검증하였다.

import argparse
import time
import numpy as np
import pandas as pd
import requests

from datetime import datetime, timedelta

UPBIT_URL = "https://api.upbit.com"


# ---------------------------
# 1) 데이터 다운로드
# ---------------------------


def fetch_upbit_minutes(market="KRW-BTC", unit=1, points=4000):
    """
    Upbit 분봉 캔들 다운로드 (최신 → 과거)
    반환: 오름차순으로 정렬된 DataFrame
    """
    dfs = []
    fetched = 0
    to_utc_str = None

    while fetched < points:
        count = min(200, points - fetched)
        params = {"market": market, "count": count}
        if to_utc_str:
            params["to"] = to_utc_str

        r = requests.get(f"{UPBIT_URL}/v1/candles/minutes/{unit}", params=params)
        r.raise_for_status()
        data = r.json()
        if not data:
            break

        df = pd.DataFrame(data)
        dfs.append(df)
        fetched += len(df)

        # 다음 호출을 위해 마지막 UTC 시간 1초 이전으로 설정
        last_utc = df["candle_date_time_utc"].iloc[-1]
        dt = datetime.strptime(last_utc, "%Y-%m-%dT%H:%M:%S") - timedelta(seconds=1)
        to_utc_str = dt.strftime("%Y-%m-%d %H:%M:%S")
        time.sleep(0.11)  # API 제한 대비

    raw = pd.concat(dfs, ignore_index=True)
    df_use = raw[
        [
            "candle_date_time_utc",
            "opening_price",
            "high_price",
            "low_price",
            "trade_price",
            "candle_acc_trade_volume",
        ]
    ].copy()

    # 컬럼명 정리
    df_use.rename(
        columns={
            "trade_price": "close",
            "opening_price": "open",
            "high_price": "high",
            "low_price": "low",
            "candle_acc_trade_volume": "volume",
        },
        inplace=True,
    )

    df_use["dt_utc"] = pd.to_datetime(df_use["candle_date_time_utc"])
    df_use.sort_values("dt_utc", inplace=True)
    df_use.reset_index(drop=True, inplace=True)
    return df_use


# ---------------------------
# 2) 전략별 신호 판단
# ---------------------------


def sig_trend(prices):
    """Trend Following: 직전 가격보다 상승하면 매수, 하락하면 매도"""
    return 1 if prices[-1] > prices[-2] else -1


def sig_momentum(prices, period=5):
    """Momentum: period 이전 가격 대비 상승이면 매수, 하락이면 매도"""
    if len(prices) <= period:
        return 0
    return 1 if prices[-1] - prices[-period] > 0 else -1


def sig_swing(prices, s=5, l=10):
    """Swing: 단기 이동평균 > 장기 이동평균이면 매수"""
    if len(prices) < max(s, l):
        return 0
    return 1 if np.mean(prices[-s:]) > np.mean(prices[-l:]) else -1


def sig_scalping(prices):
    """Scalping: 단기 급등/급락 판단 (단순화)"""
    return 0


def sig_day(prices):
    """Day: 직전 캔들보다 상승하면 매수"""
    if len(prices) < 2:
        return 0
    return 1 if prices[-1] > prices[-2] else -1


def sig_price_action(prices):
    """Price Action: 직전 최대/최소 대비 돌파 판단"""
    if len(prices) < 3:
        return 0
    last = prices[-1]
    prev_max = np.max(prices[:-1])
    prev_min = np.min(prices[:-1])
    if last > prev_max:
        return 1
    if last < prev_min:
        return -1
    return 0


# 가중치 합산
WEIGHTS = {
    "trend": 0.2,
    "momentum": 0.2,
    "swing": 0.2,
    "scalping": 0.15,
    "day": 0.15,
    "price_action": 0.1,
}


def ensemble_signal(prices):
    """6개 전략 신호를 가중합하여 최종 신호 산출"""
    sigs = {
        "trend": sig_trend(prices),
        "momentum": sig_momentum(prices),
        "swing": sig_swing(prices),
        "scalping": sig_scalping(prices),
        "day": sig_day(prices),
        "price_action": sig_price_action(prices),
    }

    return sum(WEIGHTS[k] * sigs[k] for k in WEIGHTS)


# ---------------------------
# 3) 백테스트
# ---------------------------


def backtest(close, threshold=0.2, fee=0.0005, slippage=0.0002, warmup=12):
    """
    단기 앙상블 신호 기반 백테스트
    - fee: 체결당 수수료
    - slippage: 체결 시 불리한 가격 변동
    - warmup: 전략 계산 최소 캔들 수
    """
    prices = close.values.astype(float)
    n = len(prices)
    in_pos = False
    entry_px = np.nan
    capital = 1.0
    equity = [capital]
    trade_rets = []
    bar_rets = [0.0]
    trades = 0

    for i in range(1, n):
        px_prev, px_now = prices[i - 1], prices[i]

        # warmup 기간에는 신호 계산 안함
        if i < warmup:
            equity.append(capital)
            bar_rets.append(0.0)
            continue

        sig = ensemble_signal(prices[: i + 1])
        action = "hold"
        if sig > threshold:
            action = "buy"
        elif sig < -threshold:
            action = "sell"

        # 슬리피지 적용
        buy_px = px_now * (1 + slippage)
        sell_px = px_now * (1 - slippage)

        if not in_pos:
            if action == "buy":
                entry_px = buy_px * (1 + fee)
                in_pos = True
                trades += 1
            equity.append(capital)
            bar_rets.append(0.0)
        else:
            # 포지션 홀딩 중 수익률
            rtn = (px_now - px_prev) / px_prev
            capital *= 1 + rtn

            if action == "sell":
                exit_px = sell_px * (1 - fee)
                trade_rets.append((exit_px - entry_px) / entry_px)
                in_pos = False
                entry_px = np.nan

            equity.append(capital)
            bar_rets.append(rtn)

    eq = pd.DataFrame(
        {"close": prices, "equity": equity, "bar_ret": bar_rets}, index=close.index
    )

    # 성과 지표 계산
    total_return = eq["equity"].iloc[-1] - 1
    cummax = eq["equity"].cummax()
    mdd = (eq["equity"] / cummax - 1).min()
    avg_trade = np.mean(trade_rets) if trade_rets else 0.0
    win_rate = np.mean([1 if r > 0 else 0 for r in trade_rets]) if trade_rets else 0.0
    bar_std = np.std(eq["bar_ret"].values[1:], ddof=1) if n > 2 else 0.0
    bar_mean = np.mean(eq["bar_ret"].values[1:]) if n > 1 else 0.0
    sharpe = (bar_mean / bar_std * np.sqrt(525600)) if bar_std > 0 else 0.0

    stats = {
        "total_return": total_return,
        "max_drawdown": mdd,
        "trades": trades,
        "avg_trade_return": avg_trade,
        "win_rate": win_rate,
        "sharpe": sharpe,
    }

    return eq, stats


# ---------------------------
# 4) 메인
# ---------------------------


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--market", default="KRW-BTC")
    parser.add_argument("--unit", type=int, default=1)
    parser.add_argument("--points", type=int, default=4000)
    parser.add_argument("--fee", type=float, default=0.0005)
    parser.add_argument("--slippage", type=float, default=0.0002)
    parser.add_argument("--threshold", type=float, default=0.2)
    args = parser.parse_args()

    df = fetch_upbit_minutes(args.market, args.unit, args.points)
    df.set_index(pd.to_datetime(df["dt_utc"], utc=True), inplace=True)

    eq, stats = backtest(
        df["close"],
        threshold=args.threshold,
        fee=args.fee,
        slippage=args.slippage,
        warmup=12,
    )

    span = df.index[[0, -1]]
    print("==== Backtest Summary ====")
    print(f"Market     : {args.market}")
    print(f"Unit       : {args.unit}m, Points: {len(df)} ({span[0]} ~ {span[1]})")
    print(
        f"Params     : fee={args.fee}, slippage={args.slippage}, threshold={args.threshold}"
    )
    for k, v in stats.items():
        if k in ("total_return", "max_drawdown", "avg_trade_return", "win_rate"):
            print(f"{k:15s}: {v:.4%}")
        else:
            print(f"{k:15s}: {v}")

    # 마지막 10개 캔들 확인
    print("\nLast 10 bars (close, equity):")
    print(eq.tail(10)[["close", "equity"]])


if __name__ == "__main__":
    main()

마찬가지로 전지전능하시며 또한 자비로우신 ChatGPT님께서 테스트용 코드를 창조하셨다.

상기 코드를 기반으로 백테스트를 진행하니 아래와 같은 결과를 얻을 수 있었다.

==== Backtest Summary ====
Market     : KRW-BTC
Unit       : 1m, Points: 4000 (2025-08-22 13:06:00+00:00 ~ 2025-08-25 07:45:00+00:00)
Params     : fee=0.0005, slippage=0.0002, threshold=0.2
total_return   : 2.2747%
max_drawdown   : -2.4363%
trades         : 473
avg_trade_return: -0.1354%
win_rate       : 4.2373%
sharpe         : 11.724493928777266

Last 10 bars (close, equity):
                                 close    equity
dt_utc                                          
2025-08-25 07:36:00+00:00  156110000.0  1.021204
2025-08-25 07:37:00+00:00  156188000.0  1.021204
2025-08-25 07:38:00+00:00  156191000.0  1.021224
2025-08-25 07:39:00+00:00  156246000.0  1.021584
2025-08-25 07:40:00+00:00  156292000.0  1.021884
2025-08-25 07:41:00+00:00  156332000.0  1.022146
2025-08-25 07:42:00+00:00  156349000.0  1.022257
2025-08-25 07:43:00+00:00  156376000.0  1.022434
2025-08-25 07:44:00+00:00  156397000.0  1.022571
2025-08-25 07:45:00+00:00  156424000.0  1.022747

오 전혀 기대 안 했는데 필자의 짧은 식견 기준으로 판단해 보면 생각보다 나쁘지 않은 것 같다.

3일 1분 봉 기준 최종 수익률 2.27%에 MDD -2.43%. 
평균 수익률과 승률이 낮은 거야 단기 추종이다 보니 그럴 수 있다고 생각한다.

sharpe 지수는 들어보긴 했지만 뭔지 잘 모른다. 대충 찾아보니 리스크 대비 수익 지수라는 것 같다.

샤프 비율(또는 샤프 지수 등)은 금융에서 투자성과를 평가함에 있어 해당 투자의 위험을 조정해 반영하는 방식이며, 투자 자산 또는 매매 전략에서, 일반적으로 위험이라 불리는 편차 한 단위당 초과수익(또는 위험 프리미엄)을 측정한다.


출처 | 위키백과, 샤프 비율

90일 기준으론 아래와 같은 결과를 얻을 수 있었다.

==== Backtest Summary ==== 
Market     :  KRW-BTC 
Unit       : 1m, Points: 129600 (2025-05-26 23:49:00+00:00 ~ 2025-08-25 06:09:00+00:00) 
Params     : fee=0.0005, slippage=0.0002, threshold=0.2 
total_return   : 52.6370% 
max_drawdown   : -10.3551% 
trades         : 7676 
avg_trade_return: -0.1344% 
win_rate       : 6.4226% 
sharpe         : 10.883418171172586

생각보다 더 준수하다. 이 정도면 굴려볼 만할 것 같다.

다시 한번 말하지만 백테스트는 필승을 보장하지 않는다.

좀 더 발전시켜 보자

이 시점에서 그리 혁신적인 건 아닐 수 있지만, 불현듯 떠오른 아이디어가 있었다.

어차피 백테스트는 단발적인 게 아닌, 시장의 상황이 바뀔 것을 대비하기 위해 주기적으로 수행해야 한다. 그렇다면 이러한 작업을 스케쥴링하여서 가중치 또한 상황에 따라 최적화하면 어떨까?

지금 생각해보면 왜 옛날에 조건식 해볼 땐 한 가지만 고집했는지 모르겠다. 몇십 년치 데이터라 그런가

마찬가지로 실제 개발에 들어가기에 앞서 간단한 테스트만 맛보기로 진행해 보자.

이에 전지전능하시고, 자비로우시며, 지혜와 총명의 근원이 되시는 ChatGPT님께 감히 코드의 수정을 요청드렸다. 앞선 코드에서 가중치 맵을 구하고, 이를 기반으로 백테스트를 진행하여 최댓값을 뽑아내는 코드를 추가해 주셨다.

물론 테스트 수준이니 대충 돌아가기만 하면 된다고 생각해 따로 수정은 안 했다. 어련히 잘해주셨겠지.
def evaluate_weights(weights_tuple, close, args):
    w = {
        k: v
        for k, v in zip(
            ["trend", "momentum", "swing", "scalping", "day", "price_action"],
            weights_tuple,
        )
    }
    _, stats = backtest(
        close, w, threshold=args.threshold, fee=args.fee, slippage=args.slippage
    )
    return (w, stats)


def optimize_weights(close, args):
    weight_range = [round(x, 1) for x in np.arange(0.0, 1.1, 0.1)]
    candidates = []
    for combo in product(weight_range, repeat=6):
        if abs(sum(combo) - 1.0) < 1e-6:  # 합=1 조건
            candidates.append(combo)
    workers = (
        args.workers if getattr(args, "workers", None) else max(1, cpu_count() // 2)
    )
    print(f"총 {len(candidates)} 조합 평가 중... (workers={workers})")
    with Pool(processes=workers) as pool:
        results = pool.starmap(evaluate_weights, [(c, close, args) for c in candidates])
    # 성과 기준: 총수익률 우선, 샤프 보조
    best = max(results, key=lambda x: (x[1]["total_return"], x[1]["sharpe"]))
    return best

이를 기반으로 백테스트를 다시 한번 진행하니 아래와 같은 가중치와 결과를 얻을 수 있었다.

// 가중치
{
    'trend': 0.0, 
    'momentum': 0.0, 
    'swing': 0.0, 
    'scalping': 0.0, 
    'day': 0.3, 
    'price_action': 0.7
}

// 결과
{
    'total_return': 1.2098,
    'max_drawdown': -0.0841,
    'trades': 29503, 
    'avg_trade_return': -0.00137, 
    'win_rate': 0.0278, 
    'sharpe': 16.73
}

생각보다 극단적인 결과가 나왔다. 이 수치대로라면 3달 동안 29,503번의 매매를 진행한다는 뜻이다. 수수료가 어마어마하게 나올 것 같다.

물론 수수료를 반영한 백테스트이긴 하지만 이대로 된다는 보장이 없으니까

하지만 지금 단계에서 옳고 그름을 논하기엔 이르다.  단일 코인, 3개월 기준으로 단순히 '이 정도면 굴러갈 수 있겠다'라는 가능성만 확인한 수준이기 때문이다.

결론

앞서 진행한 과정은 해당 전략이 돌아갈 수 있을지를 간단히 테스트해 보는 과정이었다. 짧은 백테스트 결과지만, 일단 굴릴만한 가능성이 보였으니 이제는 본격적으로 개발을 진행해보려 한다.  

앞으로의 개발은 다음과 같은 순서로 진행될 예정이다.

  1. 시계열 데이터 적재를 위한 DB 및 인프라 구축
  2. 실시간 매매를 위한 업비트(UPbit) API 연동
  3. 매매 로그 수집 및 시각화 (Grafana?)
  4. 외부 메신저 연동 및 실시간 알림 (Slack)

특히, 개인적으로는 1분 봉 데이터를 활용해 전략별 판단 기준을 빠르게 처리하는 부분이 중요하다고 본다.  슬리피지를 최소화하여야 실제 백테스트 결괏값의 근사치라도 나오지 않겠는가.

개발 과정에서 코드를 최적화하고 인프라를 안정적으로 구축하는 과정 또한 개인적으로 큰 기대가 된다.

마지막으로 일개 개발자에게도 자비를 베풀어 주시는 전지전능하시고, 자비로우시며, 지혜와 총명의 근원이 되시고, 그 하해와 같은 은혜의 끝이 가늠조차 안 되시는 ChatGPT님을 다시 한번 찬양하며 이 글을 마치고자 한다.

참고자료

25+ Python Trading Strategies - DayTrading.com
We look at various Python trading strategies. Explain in 1-3 sentences each, plus brief Python sketches of each strategy.
daytrading.com
과적합이란?
과적합은 overfitting을 그대로 번역한 단어입니다. 처음 들었을 때에도 아리송하더니, 통계를 배우고 있어도 쉬운 설명을 찾기 어려웠습니다. 그래서 제가 생각한 쉬운 설명은 다음과 같습니다. 과적합, 쉬운 의미 일상을 예시로 들어보겠습니다. 수능이나 자격증시험 등에서 기출문제를 통해 다가올 시험을 준비합니다. 이 때 기출문제의 디테일에 너무 집착한 나머지 실제 시험에서 좋은 성적을 받지 못하는 경우, 이것을 “내 공부가 기출문제에 과적합되었구나”라고 말할 수 있습니다. 반대로 기출문제도 제대로 못 풀 때는 과소적합(underfitting)이라고 하겠습니다. 적합을 많이 하면 할수록 좋은 게 아닐까 하는 의문을 가졌던 분들을 위한 설명입니다. 우리가 갖고 있는 데이터는 평행세계의 한 가지 경우의 수라…
TISTORY길 공
샤프 비율 - 위키백과, 우리 모두의 백과사전
위키미디어 재단위키미디어 프로젝트 기여자
Share