데이터 파이프라인 구축

AI 트레이딩의 첫걸음, 데이터 파이프라인

성공적인 AI 트레이딩 시스템은 양질의 데이터에서 시작됩니다. 시장을 분석하고, 전략을 검증하고, 실시간으로 거래를 실행하는 모든 과정에 데이터가 필요하기 때문이죠. 아무리 뛰어난 알고리즘이라도 부정확하거나 오래된 데이터 위에서는 무용지물입니다.

AI 에이전트에게는 두 종류의 데이터가 필요합니다. 첫째는 과거의 시장 움직임을 담은 '과거 데이터(Historical Data)'이고, 둘째는 현재 시장 상황을 실시간으로 알려주는 '스트리밍 데이터(Streaming Data)'입니다. 이 두 가지 데이터를 안정적으로 수집하고 관리하는 체계를 '데이터 파이프라인'이라고 부릅니다. 이제부터 견고한 데이터 파이프라인을 구축하는 방법을 단계별로 알아보겠습니다.

과거 데이터 수집: REST API

과거 데이터는 트레이딩 전략을 만들고 검증(백테스팅)하는 데 필수적입니다. 이 데이터는 보통 거래소에서 제공하는 REST API를 통해 얻을 수 있습니다. REST API는 우리가 특정 기간과 종목의 데이터를 달라고 '요청(Request)'하면, 거래소가 해당 데이터를 '응답(Response)'으로 보내주는 방식입니다. 마치 도서관에서 원하는 책을 검색해서 대출하는 것과 비슷하죠.

여기서는 개발자 친화적인 환경으로 잘 알려진 Alpaca API를 사용하여 과거 OHLCV(시가, 고가, 저가, 종가, 거래량) 데이터를 가져오는 방법을 살펴보겠습니다.

import alpaca_trade_api as tradeapi
import pandas as pd

# API 키 설정 (실제 키로 교체해야 합니다)
API_KEY = 'YOUR_API_KEY'
API_SECRET = 'YOUR_API_SECRET'
BASE_URL = 'https://paper-api.alpaca.markets' # 페이퍼 트레이딩 환경

# Alpaca API 클라이언트 초기화
api = tradeapi.REST(API_KEY, API_SECRET, base_url=BASE_URL)

# 데이터 요청 파라미터 설정
symbol = 'AAPL'
timeframe = '1Day' # 1일봉 데이터
start_date = '2023-01-01'
end_date = '2023-12-31'

# 과거 OHLCV 데이터 가져오기
barset = api.get_bars(
    symbol,
    timeframe,
    start=start_date,
    end=end_date
).df

# 결과 확인
print(barset.head())

위 코드를 실행하면 지정된 기간의 애플(AAPL) 일봉 데이터가 Pandas DataFrame 형태로 반환됩니다. 하지만 수년 치의 분봉 데이터처럼 대용량 데이터를 한 번에 요청하면 문제가 생길 수 있습니다. 대부분의 API는 단시간에 너무 많은 요청을 보내는 것을 막는 '호출 제한(Rate Limit)' 정책이 있기 때문입니다. 따라서 대량의 데이터를 수집할 때는 기간을 잘게 나누어 여러 번에 걸쳐 요청하고, 중간에 잠시 쉬어가는 로직을 추가하는 것이 안정적입니다.

실시간 데이터 스트리밍: WebSocket

과거 데이터로 전략을 세웠다면, 이제 실시간 시장 상황에 맞춰 거래를 실행해야 합니다. 이때 필요한 것이 바로 WebSocket API입니다. REST API가 필요할 때마다 데이터를 '가져오는(pull)' 방식이라면, WebSocket은 한 번 연결을 맺어두면 거래소에서 새로운 데이터가 발생할 때마다 '밀어주는(push)' 방식입니다.

이 덕분에 우리는 가격 변동, 체결 정보 등을 지연 시간 없이 즉시 받아볼 수 있습니다. 실시간 시세에 빠르게 반응해야 하는 트레이딩 에이전트에게는 필수적인 기술입니다.

import alpaca_trade_api as tradeapi
import asyncio

# API 키 설정
API_KEY = 'YOUR_API_KEY'
API_SECRET = 'YOUR_API_SECRET'

# 스트림 클라이언트 초기화
stream = tradeapi.Stream(API_KEY, API_SECRET)

# 데이터 수신 시 실행될 비동기 함수 정의
async def on_trade(t):
    print(f'New trade received: {t}')

# 특정 종목의 거래(trade) 데이터 구독
stream.subscribe_trades(on_trade, 'AAPL')

# 스트림 실행 (이벤트 루프 시작)
stream.run()

이 코드는 'AAPL' 주식에 새로운 거래가 체결될 때마다 해당 정보를 화면에 출력합니다. WebSocket은 지속적으로 데이터를 보내주므로, 이 데이터를 받아서 데이터베이스에 차곡차곡 쌓아두는 프로세스를 만들어야 합니다. 이렇게 수집된 실시간 데이터는 나중에 더 정교한 분석을 위한 소중한 자산이 됩니다.

데이터 정제 및 관리

API로 데이터를 수집하다 보면 여러 문제에 부딪힙니다. 데이터가 중간에 누락되거나(결측치), 거래소마다 시간 기준(타임존)이 달라 데이터가 뒤섞일 수 있습니다. 이런 '데이터 노이즈'는 AI 모델의 성능을 크게 저하시키는 주범입니다.

Pandas는 이런 시계열 데이터를 다루는 데 매우 강력한 도구입니다. 결측치는 앞뒤 데이터로 채우거나(보간), 타임존은 tz_convert 같은 함수로 통일할 수 있습니다. 예를 들어, 뉴욕 증시 데이터는 'America/New_York' 타임존을 기준으로 다루어야 합니다.

# 타임존을 UTC에서 뉴욕 시간으로 변환
barset_localized = barset.tz_localize('UTC').tz_convert('America/New_York')

# 결측치가 있는지 확인
missing_values = barset_localized.isnull().sum()
print(f'\nMissing values:\n{missing_values}')

# 선형 보간법으로 결측치 채우기 (예시)
barset_filled = barset_localized.fillna(method='ffill')

데이터 정제까지 마쳤다면, 이제 이 데이터들을 효율적으로 저장하고 관리할 차례입니다. 매번 파일로 저장하거나 메모리에 전부 올려두는 방식은 데이터가 커질수록 비효율적입니다. 이때 데이터베이스(DB)를 사용하면 대용량 데이터를 안정적으로 관리하고, 필요할 때 원하는 만큼만 빠르게 조회할 수 있습니다.

PostgreSQL과 같은 관계형 데이터베이스나 Supabase 같은 클라우드 기반 DB를 활용하면 데이터 파이프라인을 더욱 견고하게 만들 수 있습니다. 잘 설계된 DB 스키마는 데이터의 정합성을 보장하고, AI 에이전트가 언제든 과거와 현재 데이터를 넘나들며 분석할 수 있는 환경을 제공합니다.

데이터 종류	수집 방법	주요 특징	사용 목적
과거 데이터	REST API	요청 기반(Pull)	백테스팅, 모델 학습
실시간 데이터	WebSocket API	푸시 기반(Push)	실시간 거래, 모니터링

Quiz Questions 1/5

AI 트레이딩 시스템을 구축할 때 필요한 두 가지 주요 데이터 유형은 무엇인가요?

Quiz Questions 2/5

과거 데이터를 수집하여 전략을 검증(백테스팅)하는 데 주로 사용되는 API 방식과, 실시간 시세 변동을 지연 없이 받아보는 데 주로 사용되는 API 방식을 올바르게 짝지은 것은 무엇인가요?

AI 트레이딩 에이전트 입문 및 설계