[자동 매매] 보조지표 그리기

지난 글에서는 실시간으로 받아오는 데이터를 TradingView에서 지원하는 라이브러리를 통해 클라이언트 화면에 보여주는 작업을 진행했습니다. 이번 글에서는 이전 데이터들을 받아오고, 가장 기본적인 지표인 거래량과 이동평균선을 추가해 차트에 같이 보여주는 작업을 진행해보도록 하겠습니다.

 

데이터 불러오기

데이터를 불러오기 위해서는 업비트에서 이전 데이터들을 불러오면 됩니다. 캔들(OHLCV)를 받아오는 API는 다음과 같습니다.

import requests

url = f"https://api.upbit.com/v1/candles/{type}"

headers = {"accept": "application/json"}

response = requests.get(url, headers=headers)

print(response.text)

 

초(Second), 분(Minute), 일(Day), 주(Week), 월(Month), 연(Year) 캔들을 제공하며, 해당 시간대에 체결이 발생한 경우에만 생성된다고 나와있습니다. 초기에 데이터를 한 번 불러오고, 그 이후에는 연결된 웹소켓을 통해서 실시간으로 데이터를 받아오면 되기 때문에 주의 사항은 크게 의미가 없을 것 같습니다. 

import { PrevCandleProps } from "@/types/types";
import { useGet } from "./useAPI";

const usePreviousData = (type: string, code: string) => {
  const { data } = useGet<PrevCandleProps[]>(
    `api/candle/${type}/${code}`,
    [type, code],
    true,
    Infinity
  );
  return data;
};
export default usePreviousData;

// 필요한 변수 설정
const code = "KRW-BTC";
const [type, setType] = useState("minutes1");
const prevData: PrevCandleProps[] = usePreviousData(type, code);

 

분봉의 경우에는 /minute/{unit}으로 몇분봉인지 설정을 해줘야하기 때문에, type이 minutes인 경우에는 minutes/로 변환한 후 데이터 요청을 보냅니다. 이 과정은 백엔드에서 진행합니다. 기존에 trading_routers.py와 trading_services.py는 웹소켓용으로 websocket_routers/services.py로 변경하고, api만 요청하는 라우터를 하나 추가로 만들었습니다.

# api_routers.py
from fastapi import APIRouter, WebSocket
from app.services import api_services

router = APIRouter()

@router.get("/candle/{type}/{code}")
async def get_candle_data(type: str, code: str, count: int = 200):
    # type 예: "minutes1", "minutes5", "days"
    if type.startswith("minutes"):
        type = type.replace("minutes", "minutes/")
    data = await api_services.get_candle_data(type, code, count)
    return data

# api_services.py
import json
import requests
import pandas as pd
from app.services import indicators

# 업비트 캔들 데이터 조회
async def get_candle_data(type: str, code: str, count: int = 200):

    url = f"https://api.upbit.com/v1/candles/{type}"
    params = {"market": code, "count": count}
    headers = {"Accept": "application/json"}

    response = requests.get(url, headers=headers, params=params)
    response.raise_for_status()
    data = response.json()

    return data

 

보조지표 만들기

이제 받아온 데이터들로 보조지표를 만들어서 같이 프론트에 넘겨보도록 합시다. 보조지표 생성을 위해서는 pandas와 ta 라이브러리를 사용했습니다. 각각의 보조지표가 뭐를 의미하는지는 알지 못해서 코드만 첨부하겠습니다. 

# indicators.py

import pandas as pd
import ta

def add_indicators(df: pd.DataFrame):
    df["sma5"] = df["close"].rolling(window=5).mean()
    df["sma20"] = df["close"].rolling(window=20).mean()
    df["ema12"] = df["close"].ewm(span=12, adjust=False).mean()
    df["ema26"] = df["close"].ewm(span=26, adjust=False).mean()
    df["macd"] = df["ema12"] - df["ema26"]
    df["signal"] = df["macd"].ewm(span=9, adjust=False).mean()
    df["rsi14"] = ta.momentum.RSIIndicator(df["close"], window=14).rsi()
    df["rsi7"] = ta.momentum.RSIIndicator(df["close"], window=7).rsi()

    bb = ta.volatility.BollingerBands(close=df["close"], window=20, window_dev=2)
    df["bb_middle"] = bb.bollinger_mavg()
    df["bb_upper"] = bb.bollinger_hband()
    df["bb_lower"] = bb.bollinger_lband()

    return df

 

그리고 이 값들을 data를 넘겨줄 때 포함해서 넘겨줍시다.

async def get_candle_data(type: str, code: str, count: int = 200):

    url = f"https://api.upbit.com/v1/candles/{type}"
    params = {"market": code, "count": count}
    headers = {"Accept": "application/json"}

    response = requests.get(url, headers=headers, params=params)
    response.raise_for_status()
    data = response.json()

    df = pd.DataFrame([{
        "market": c["market"],
        "candle_date_time_utc": c["candle_date_time_utc"],
        "candle_date_time_kst": c["candle_date_time_kst"],
        "opening_price": c["opening_price"],
        "high_price": c["high_price"],
        "low_price": c["low_price"],
        "trade_price": c["trade_price"],
        "timestamp": c["timestamp"],
        "candle_acc_trade_price": c["candle_acc_trade_price"],
        "candle_acc_trade_volume": c["candle_acc_trade_volume"],
        "unit": c.get("unit", 1),
        "time": int(pd.to_datetime(c["candle_date_time_utc"]).timestamp()),

        # indicators 계산용 컬럼
        "open": c["opening_price"],
        "high": c["high_price"],
        "low": c["low_price"],
        "close": c["trade_price"],
        "volume": c["candle_acc_trade_volume"],
    } for c in data])

    # 시간순 정렬
    df = df.sort_values("timestamp")

    # 보조지표 추가
    df = indicators.add_indicators(df)

    # JSON 직렬화 안전 처리 (NaN → None)
    result = []
    for row in df.to_dict(orient="records"):
        clean_row = {k: (None if pd.isna(v) else v) for k, v in row.items()}
        result.append(clean_row)

    return result

 

다 포함해서 타입 정의를 해줍시다. 생각보다 받아오는 데이터 종류가 다양하고, 로직이 복잡해서 정적 검사를 하면서 진행하는게 중요해보입니다. 

// 과거 데이터 타입
export interface PrevCandleProps {
  market: string;
  candle_date_time_utc: string;
  candle_date_time_kst: string;
  opening_price: number;
  high_price: number;
  low_price: number;
  trade_price: number;
  timestamp: number;
  candle_acc_trade_price: number;
  candle_acc_trade_volume: number;
  time: number;
  unit: number;
  sma5: number | null;
  sma20: number | null;
  ema12: number | null;
  ema26: number | null;
  macd: number | null;
  signal: number | null;
  rsi14: number | null;
  rsi7: number | null;
  bb_lower: number | null;
  bb_middle: number | null;
  bb_upper: number | null;
}

 

추가로 차트에 들어가는 보조지표들도 만들어줍시다. lightweight 라이브러리에서 addLineSeries와 addHistogramSeries로 쉽게 생성해줄 수 있었습니다. 

import { ChartSeriesProps } from "@/types/types";
import { IChartApi } from "lightweight-charts";

const createSeries = (chart: IChartApi): ChartSeriesProps => {
  const candleSeries = chart.addCandlestickSeries({
    upColor: "#f04452",
    borderUpColor: "#f04452",
    wickUpColor: "#f04452",
    downColor: "#3182f6",
    borderDownColor: "#3182f6",
    wickDownColor: "#3182f6",
  });

  const sma5Series = chart.addLineSeries({ color: "orange", lineWidth: 3 });
  const sma20Series = chart.addLineSeries({ color: "blue", lineWidth: 3 });

  const bbUpperSeries = chart.addLineSeries({ color: "green", lineWidth: 1 });
  const bbMiddleSeries = chart.addLineSeries({ color: "green", lineWidth: 1 });
  const bbLowerSeries = chart.addLineSeries({ color: "green", lineWidth: 1 });

  const macdSeries = chart.addLineSeries({
    color: "red",
    lineWidth: 2,
    priceScaleId: "macd",
  });
  const signalSeries = chart.addLineSeries({
    color: "blue",
    lineWidth: 2,
    priceScaleId: "macd",
  });
  const macdHistSeries = chart.addHistogramSeries({
    priceScaleId: "macd",
    priceFormat: { type: "volume" },
  });

  const rsiSeries = chart.addLineSeries({
    color: "purple",
    lineWidth: 2,
    priceScaleId: "rsi",
  });
  const rsi7Series = chart.addLineSeries({
    color: "pink",
    lineWidth: 2,
    priceScaleId: "rsi",
  });

  // 차트에 추가
  chart.priceScale("macd").applyOptions({
    scaleMargins: { top: 0.75, bottom: 0.15 },
  });
  chart.priceScale("rsi").applyOptions({
    scaleMargins: { top: 0.9, bottom: 0 },
  });
  chart.priceScale("right").applyOptions({
    scaleMargins: { top: 0.1, bottom: 0.3 },
  });

  return {
    candleSeries,
    sma5Series,
    sma20Series,
    bbUpperSeries,
    bbMiddleSeries,
    bbLowerSeries,
    macdSeries,
    signalSeries,
    macdHistSeries,
    rsiSeries,
    rsi7Series,
  };
};

export default createSeries;

 

실시간 데이터 누적하기

데이터들을 받아왔으니, 이제 실시간 데이터들을 보조지표들을 계산한 후에 계속 넘겨줘서 실시간 동향을 파악할 수 있도록 해봅시다. 마찬가지로 분봉의 경우 minutes를 minutes/로 변환한 후 API 요청을 보냅니다. 

def round_time(trade_ts: int, type: str) -> int:
    dt = pd.to_datetime(trade_ts, unit="s")  # UTC 기준 datetime

    if type.startswith("minutes/"):
        unit = int(type.split("/")[1])
        minute = (dt.minute // unit) * unit
        dt = dt.replace(minute=minute, second=0, microsecond=0)
        return int(dt.timestamp())

    elif type == "days":
        dt_kst = dt + pd.Timedelta(hours=9)
        dt_kst = dt_kst.normalize()  # 한국시간 자정
        return int((dt_kst - pd.Timedelta(hours=9)).timestamp())

    elif type == "weeks":
        dt_kst = dt + pd.Timedelta(hours=9)
        monday_kst = dt_kst - pd.Timedelta(days=dt_kst.weekday())
        monday_kst = monday_kst.normalize()
        return int((monday_kst - pd.Timedelta(hours=9)).timestamp())

    elif type == "months":
        dt_kst = dt + pd.Timedelta(hours=9)
        first_day_kst = dt_kst.replace(day=1).normalize()
        return int((first_day_kst - pd.Timedelta(hours=9)).timestamp())

    elif type == "years":
        dt_kst = dt + pd.Timedelta(hours=9)
        first_day_kst = dt_kst.replace(month=1, day=1).normalize()
        return int((first_day_kst - pd.Timedelta(hours=9)).timestamp())

    # fallback → 1분
    return (trade_ts // 60) * 60

async def backend_websocket(websocket: WebSocket, code: str, type: str):
    # Upbit REST API 용 포맷 맞추기
    if type.startswith("minutes"):
        type = type.replace("minutes", "minutes/")

    await websocket.accept()
    try:
        # 초기 과거 데이터
        raw_candles = await api_services.get_candle_data(type, code, 200)
        df = pd.DataFrame([{
            "time": c["timestamp"] // 1000,   # 업비트 REST timestamp = UTC 기준
            "open": c["opening_price"],
            "high": c["high_price"],
            "low": c["low_price"],
            "close": c["trade_price"],
            "volume": c["candle_acc_trade_volume"],
        } for c in raw_candles])
        df = df.sort_values("time")
        df = indicators.add_indicators(df)
        candles[code] = df

        # await websocket.send_json([clean_dict(row) for row in df.to_dict(orient="records")])

        # 실시간 데이터
        while True:
            await asyncio.sleep(0.5)

            if code in last_data:
                raw = last_data[code]
                trade_ts = raw["trade_timestamp"] // 1000  # 초 단위 (UTC)
                price = raw["trade_price"]
                trade_volume = raw["trade_volume"]

                # 같은 체결이면 무시
                if code in last_trade_ts and last_trade_ts[code] == raw["trade_timestamp"]:
                    continue
                last_trade_ts[code] = raw["trade_timestamp"]

                ts = round_time(trade_ts, type)

                df = candles[code]

                if not df.empty and ts < df.iloc[-1]["time"]:
                    continue

                if not df.empty and df.iloc[-1]["time"] == ts:
                    df.at[df.index[-1], "high"] = max(df.iloc[-1]["high"], price)
                    df.at[df.index[-1], "low"] = min(df.iloc[-1]["low"], price)
                    df.at[df.index[-1], "close"] = price
                    df.at[df.index[-1], "volume"] += trade_volume
                else:
                    df.loc[len(df)] = {
                        "time": ts,
                        "open": price,
                        "high": price,
                        "low": price,
                        "close": price,
                        "volume": trade_volume,
                    }

                df = indicators.add_indicators(df)
                candles[code] = df

                last_row = clean_dict(df.iloc[-1].to_dict())
                payload = {**raw, **last_row}
                await websocket.send_json(payload)

    except WebSocketDisconnect:
        print("disconnected")
    except Exception as e:
        print("error in websocket:", e)
    finally:
        await websocket.close()

 

프론트에서 받아온 데이터를 다시 차트에 반영해주도록 합시다.

// hooks/useRealtimeCandle.ts
import { useEffect, useRef } from "react";
import { ChartSeriesProps, CurrentCandleProps } from "@/types/types";
import { CandlestickData, UTCTimestamp } from "lightweight-charts";

export const useRealtimeCandle = (
  code: string,
  type: string,
  series: ChartSeriesProps
) => {
  const currentCandleRef = useRef<CandlestickData | null>(null);

  useEffect(() => {
    if (!series) return;

    const ws = new WebSocket(`ws://localhost:8000/ws/${code}/${type}`);

    ws.onmessage = (event) => {
      const data: CurrentCandleProps = JSON.parse(event.data);
      const candleTime = Number(data.time) as UTCTimestamp;
      const price = data.trade_price;
      let currentCandle = currentCandleRef.current;
      // 이전과 다른 타임스탬프가 들어왔을 때 (새로운 캔들)
      if (!currentCandle || Number(currentCandle.time) !== candleTime) {
        currentCandle = {
          time: candleTime,
          open: price,
          high: price,
          low: price,
          close: price,
        };
      } else {
        // 현재 캔들 이어서 업데이트
        currentCandle.high = Math.max(currentCandle.high, price);
        currentCandle.low = Math.min(currentCandle.low, price);
        currentCandle.close = price;
        series.candleSeries.update(currentCandle);
      }

      // 볼린저 밴드 업데이트
      if (data.bb_upper != null)
        series.bbUpperSeries.update({ time: candleTime, value: data.bb_upper });
      if (data.bb_middle != null)
        series.bbMiddleSeries.update({
          time: candleTime,
          value: data.bb_middle,
        });
      if (data.bb_lower != null)
        series.bbLowerSeries.update({ time: candleTime, value: data.bb_lower });

      if (data.sma5 != null)
        series.sma5Series.update({ time: candleTime, value: data.sma5 });
      if (data.sma20 != null)
        series.sma20Series.update({ time: candleTime, value: data.sma20 });

      // MACD 업데이트
      if (data.macd != null && data.signal != null) {
        series.macdSeries.update({ time: candleTime, value: data.macd });
        series.signalSeries.update({ time: candleTime, value: data.signal });
        const hist = data.macd - data.signal;
        series.macdHistSeries.update({
          time: candleTime,
          value: hist,
          color: hist >= 0 ? "green" : "red",
        });
      }

      // RSI 업데이트
      if (data.rsi14 != null)
        series.rsiSeries.update({ time: candleTime, value: data.rsi14 });
      if (data.rsi7 != null)
        series.rsi7Series.update({ time: candleTime, value: data.rsi7 });

      currentCandleRef.current = currentCandle;
    };

    return () => ws.close();
  }, [code, type, series]);
};

 

결과입니다. 업비트에서 내려주는 캔들 데이터가 최대 200개까지만 가능해서 생각보다 그래프 추이를 보는데에는 한계가 있을 수도 있겠네요. 추가로 라이브러리에서는 UTC 시간대를 받고, 업비트에서 받을 수 있는 데이터는 KST, UTC가 있는데 이를 처리하는 부분에서 애를 먹어서 분봉만 실시간으로 보조지표가 변하도록 설정해뒀습니다. 첫 로딩 시 이전 데이터들을 다 받아오는데 일봉 이상의 경우에는 실시간으로 보조지표를 확인할 이유도 적을 것 같구요. 아마 맨 마지막에 수정될 것 같습니다. 

이번 글에서는 이렇게 이전 데이터와 보조지표를 받아오고, 실시간으로 현재가를 받아오고, 분봉에서는 보조지표도 실시간으로 그리는 걸 구현해봤습니다. 다음 글에서는 보조지표들을 활용해서 차트에 매매/매수 신호를 보내고, 그 기점으로 거래를 진행해보도록 하겠습니다. 

 

언제나처럼 ㅡ 시작은 삽질이지만, 끝은 지식입니다.