Polygon, Alpaca API 및 Python을 활용한 자동 주식 포트폴리오 백테스팅 파이프라인 구축: 성과 분석 및 위험 관리 최적화
자동화된 주식 포트폴리오 백테스팅 파이프라인은 과거 데이터를 기반으로 투자 전략의 효율성을 평가하고 최적화하는 데 필수적입니다. 이 글에서는 Polygon의 정확한 주식 데이터와 Alpaca API의 편리한 거래 기능을 결합하여 Python으로 백테스팅 파이프라인을 구축하고, 성과 분석 및 위험 관리를 개선하는 방법을 상세히 설명합니다. 더 이상 스프레드시트에 시간을 낭비하지 마세요. 이 파이프라인을 통해 더 빠르고, 정확하며, 데이터 기반의 투자 결정을 내릴 수 있습니다.
1. The Challenge / Context
주식 시장은 끊임없이 변동하며, 성공적인 투자를 위해서는 체계적이고 데이터 기반의 접근 방식이 필요합니다. 과거의 데이터에 근거하여 투자 전략을 테스트하는 백테스팅은 리스크를 줄이고 잠재적인 수익성을 평가하는 데 중요한 역할을 합니다. 하지만 수동으로 데이터를 수집하고 분석하는 과정은 시간이 많이 소요되고 오류가 발생하기 쉽습니다. 특히, 정확하고 포괄적인 주식 시장 데이터를 확보하는 것과, 백테스팅 결과를 효과적으로 분석하고 시각화하는 것이 주요 과제입니다. 현재 많은 개인 투자자 및 소규모 투자 그룹은 백테스팅에 필요한 데이터와 도구에 대한 접근성이 낮아 어려움을 겪고 있습니다. 이는 정보 비대칭을 심화시키고, 투자 결정의 품질 저하로 이어질 수 있습니다.
2. Deep Dive: Alpaca API 및 Polygon.io
Alpaca API는 프로그래밍 방식으로 주식, ETF 및 기타 금융 상품을 거래할 수 있는 플랫폼입니다. 주요 특징은 다음과 같습니다.
- REST API: 간단하고 직관적인 REST API를 제공하여 다양한 프로그래밍 언어에서 쉽게 사용할 수 있습니다.
- 실시간 데이터 스트리밍: 실시간 주식 시세 및 시장 데이터를 제공하여 백테스팅 및 실시간 거래에 필요한 정보를 얻을 수 있습니다.
- 페이퍼 트레이딩 (Paper Trading): 실제 돈을 사용하지 않고 가상으로 거래를 연습할 수 있는 환경을 제공하여 투자 전략을 테스트하고 개선할 수 있습니다.
- 커미션 프리 거래: 커미션 수수료 없이 주식 및 ETF를 거래할 수 있어 비용 효율적인 투자 환경을 제공합니다.
Polygon.io는 주식, 옵션, 외환 및 암호화폐 시장에 대한 실시간 및 과거 데이터를 제공하는 데이터 공급 업체입니다. 주요 특징은 다음과 같습니다.
- 광범위한 데이터 범위: 다양한 금융 시장에 대한 데이터를 제공하여 포괄적인 백테스팅을 수행할 수 있습니다.
- 정확하고 안정적인 데이터: 높은 품질의 데이터를 제공하여 신뢰할 수 있는 백테스팅 결과를 얻을 수 있습니다.
- REST API: Alpaca와 마찬가지로 REST API를 제공하여 데이터를 쉽게 통합할 수 있습니다.
- 다양한 데이터 유형: 주식 시세, 거래량, 재무 제표 등 다양한 데이터 유형을 제공하여 심층적인 분석을 수행할 수 있습니다.
Alpaca API를 거래 플랫폼으로, Polygon.io를 데이터 공급원으로 활용하면 강력하고 유연한 백테스팅 파이프라인을 구축할 수 있습니다.
3. Step-by-Step Guide / Implementation
이 섹션에서는 Polygon 데이터와 Alpaca API를 이용하여 자동화된 백테스팅 파이프라인을 Python으로 구축하는 방법을 단계별로 설명합니다.
Step 1: 환경 설정 및 라이브러리 설치
먼저 필요한 라이브러리를 설치하고 환경을 설정합니다. Python 3.7 이상이 설치되어 있는지 확인하십시오. 다음 명령을 사용하여 필요한 라이브러리를 설치합니다.
pip install alpaca-trade-api pandas python-dotenv polygon-api-client matplotlib
Step 2: API 키 설정
Alpaca API와 Polygon.io API를 사용하려면 API 키가 필요합니다. Alpaca와 Polygon 웹사이트에서 계정을 생성하고 API 키를 발급받으십시오. `.env` 파일을 생성하여 API 키를 안전하게 저장합니다.
# .env 파일 내용
ALPACA_API_KEY="YOUR_ALPACA_API_KEY"
ALPACA_SECRET_KEY="YOUR_ALPACA_SECRET_KEY"
POLYGON_API_KEY="YOUR_POLYGON_API_KEY"
`.env` 파일에서 API 키를 로드하는 코드는 다음과 같습니다.
import os
from dotenv import load_dotenv
load_dotenv()
ALPACA_API_KEY = os.getenv("ALPACA_API_KEY")
ALPACA_SECRET_KEY = os.getenv("ALPACA_SECRET_KEY")
POLYGON_API_KEY = os.getenv("POLYGON_API_KEY")
Step 3: 데이터 수집 함수 구현
Polygon API를 사용하여 과거 주식 데이터를 수집하는 함수를 구현합니다. 다음은 특정 기간 동안 특정 주식의 OHLC (Open, High, Low, Close) 데이터를 가져오는 함수입니다.
from polygon import RESTClient
import pandas as pd
from datetime import datetime
def get_stock_data(symbol, start_date, end_date, polygon_api_key):
"""
Polygon API를 사용하여 과거 주식 데이터를 수집합니다.
Args:
symbol (str): 주식 티커 심볼 (예: "AAPL").
start_date (str): 데이터 시작 날짜 (YYYY-MM-DD 형식).
end_date (str): 데이터 종료 날짜 (YYYY-MM-DD 형식).
polygon_api_key (str): Polygon API 키.
Returns:
pandas.DataFrame: OHLC 데이터를 포함하는 Pandas DataFrame.
데이터를 가져오는 데 실패하면 None을 반환합니다.
"""
try:
client = RESTClient(polygon_api_key)
start_datetime = datetime.strptime(start_date, '%Y-%m-%d')
end_datetime = datetime.strptime(end_date, '%Y-%m-%d')
# API 호출에 시간 제한을 둡니다. (이 예에서는 60초)
iterator = client.get_aggs(
symbol,
1,
"day",
start_datetime,
end_datetime,
raw_response=False,
limit=5000 # API 호출당 최대 5000개의 결과를 가져올 수 있습니다.
)
data = []
for item in iterator:
data.append({
'date': datetime.fromtimestamp(item.timestamp/1000).strftime('%Y-%m-%d'), # 밀리초 단위이므로 1000으로 나눕니다.
'open': item.open,
'high': item.high,
'low': item.low,
'close': item.close,
'volume': item.volume
})
df = pd.DataFrame(data)
df['date'] = pd.to_datetime(df['date'])
df.set_index('date', inplace=True)
return df
except Exception as e:
print(f"Error fetching data for {symbol}: {e}")
return None
Step 4: 백테스팅 로직 구현
수집된 데이터를 기반으로 백테스팅 로직을 구현합니다. 간단한 이동 평균 교차 전략을 예로 들어 보겠습니다. 50일 이동 평균이 200일 이동 평균을 상향 돌파할 때 매수하고, 하향 돌파할 때 매도하는 전략입니다.
def moving_average_crossover(data, short_window, long_window, initial_capital=100000):
"""
이동 평균 교차 전략을 백테스팅합니다.
Args:
data (pandas.DataFrame): OHLC 데이터를 포함하는 Pandas DataFrame.
short_window (int): 단기 이동 평균 기간.
long_window (int): 장기 이동 평균 기간.
initial_capital (float): 초기 자본금.
Returns:
pandas.DataFrame: 거래 내역 및 포트폴리오 가치를 포함하는 Pandas DataFrame.
"""
signals = pd.DataFrame(index=data.index)
signals['signal'] = 0.0
signals['short_mavg'] = data['close'].rolling(window=short_window, min_periods=short_window).mean()
signals['long_mavg'] = data['close'].rolling(window=long_window, min_periods=long_window).mean()
signals['signal'][short_window:] = np.where(signals['short_mavg'][short_window:] > signals['long_mavg'][short_window:], 1.0, 0.0)
signals['positions'] = signals['signal'].diff()
# 백테스팅 로직
initial_capital = float(initial_capital)
positions = pd.DataFrame(index=signals.index)
positions['stock'] = 100 * signals['signal'] # 가정: 주가와 상관없이 항상 100주를 매수/매도
portfolio = initial_capital - (data['close'] * positions['stock']).cumsum() #포트폴리오 가치
positions['portfolio'] = portfolio
return positions
Step 5: 성과 분석
백테스팅 결과를 기반으로 성과를 분석합니다. 총 수익, 샤프 지수, 최대 낙폭 (Maximum Drawdown) 등의 지표를 계산하여 전략의 효율성을 평가합니다.
import numpy as np
def calculate_performance_metrics(returns, risk_free_rate=0.0):
"""
포트폴리오 수익률을 기반으로 성과 지표를 계산합니다.
Args:
returns (pandas.Series): 포트폴리오 수익률 시계열 데이터.
risk_free_rate (float): 무위험 이자율 (연간, 소수점).
Returns:
dict: 성과 지표를 포함하는 딕셔너리.
"""
total_return = (returns + 1).prod() - 1
annualized_return = (1 + total_return)**(252/len(returns)) - 1 # 연간 수익률 (영업일 기준 252일)
# 샤프 지수 계산
excess_returns = returns - risk_free_rate/252 # 일일 초과 수익률
sharpe_ratio = np.sqrt(252) * (excess_returns.mean() / excess_returns.std())
# 최대 낙폭 계산
cumulative_returns = (1 + returns).cumprod()
peak = cumulative_returns.expanding(min_periods=1).max()
drawdown = (cumulative_returns / peak) - 1
max_drawdown = drawdown.min()
return {
'Total Return': total_return,
'Annualized Return': annualized_return,
'Sharpe Ratio': sharpe_ratio,
'Max Drawdown': max_drawdown
}
Step 6: 위험 관리 (Risk Management)
백테스팅 결과에서 최대 낙폭(Maximum Drawdown)을 확인하고, 포지션 사이즈 조정 또는 손절매 (Stop-Loss) 전략을 추가하여 위험을 관리합니다. 예를 들어, 최대 낙폭이 특정 수준을 초과하면 포지션 사이즈를 줄이거나, 손절매 가격을 설정하여 손실을 제한할 수 있습니다.
def apply_stop_loss(positions, data, stop_loss_percentage):
"""손절매 전략을 적용합니다.
Args:
positions (pd.DataFrame): 거래 포지션 데이터프레임 (매수/매도 시그널 포함).
data (pd.DataFrame): 주식 가격 데이터프레임 (종가 기준).
stop_loss_percentage (float): 손절매 비율 (예: 0.05는 5% 손절매).
Returns:
pd.DataFrame: 손절매가 적용된 포지션 데이터프레임.
"""
positions['stop_loss_hit'] = False # 손절매 트리거 여부를 추적하는 열 추가
initial_price = 0
for i in range(1, len(positions)):
if positions['positions'][i-1] != 0 and initial_price == 0:
initial_price = data['close'][i-1]
if positions['positions'][i-1] > 0: # 매수 포지션인 경우 손절매 가격 계산
stop_loss_price = initial_price * (1 - stop_loss_percentage)
if data['close'][i] < stop_loss_price:
positions['positions'][i] = -positions['positions'][i-1] # 매도 시그널로 변경
positions['stop_loss_hit'][i] = True
initial_price = 0
elif positions['positions'][i-1] < 0: # 매도 포지션인 경우 손절매 가격 계산
stop_loss_price = initial_price * (1 + stop_loss_percentage)
if data['close'][i] > stop_loss_price:
positions['positions'][i] = -positions['positions'][i-1] # 매수 시그널로 변경
positions['stop_loss_hit'][i] = True
initial_price = 0
return positions
Step 7: 전체 파이프라인 실행
위에서 구현한 함수들을 사용하여 전체 백테스팅 파이프라인을 실행합니다.
if __name__ == '__main__':
# 설정
symbol = "AAPL"
start_date = "2020-01-01"
end_date = "2023-12-31"
short_window = 50
long_window = 200
initial_capital = 100000
stop_loss_percentage = 0.05 # 손절매 비율 (5%)
# 데이터 수집
data = get_stock_data(symbol, start_date, end_date, POLYGON_API_KEY)
if data is None:
print("데이터를 가져오는 데 실패했습니다. 프로그램을 종료합니다.")
exit()
# 백테스팅
positions = moving_average_crossover(data, short_window, long_window, initial_capital)
# 손절매 적용
positions = apply_stop_loss(positions, data, stop_loss_percentage)
# 수익률 계산
returns = positions['portfolio'].pct_change().dropna()
# 성과 분석
performance_metrics = calculate_performance_metrics(returns)
print(performance_metrics)
# 시각화
import matplotlib.pyplot as plt
plt.figure(figsize=(14, 7))
plt.plot(positions['portfolio'])
plt.title(f'{symbol} 백테스팅 결과')
plt.xlabel('날짜')
plt.ylabel('포트폴리오 가치')
plt.show()
4. Real-world Use Case / Example
개인적으로, 초기에는 다양한 투자 전략을 수동으로 백테스팅하면서 상당한 시간을 소비했습니다. 엑셀 스프레드시트를 사용하여 데이터를 분석하고 시각화하는 과정은 복잡하고 비효율적이었습니다. 이 파이프라인을 구축한 후, 새로운 투자 아이디어를 테스트하고 기존 전략을 최적화하는 데 필요한 시간을 획기적으로 줄일 수 있었습니다. 예를 들어, 특정 경제 지표에 따라 매수/매도 시점을 결정하는 전략을 테스트하는 데 몇 시간이 걸렸지만, 이 파이프라인을 사용하면 몇 분 안에 결과를 얻을 수 있습니다. 또한, 손절매 및 포지션 사이즈 조정과 같은 위험 관리 요소를 통합하여 포트폴리오의 안정성을 높일 수 있었습니다.
5. Pros & Cons / Critical Analysis
- Pros:
- 자동화된 백테스팅: 데이터 수집부터 성과 분석까지 전체 과정을 자동화하여 시간과 노력을 절약합니다.
- 정확하고 신뢰할 수 있는 데이터: Polygon API를 통해 정확하고 안정적인 주식 시장 데이터를 확보할 수 있습니다.
- 유연하고 확장 가능한 아키텍처: Python 코드를 수정하여 다양한 투자 전략 및 위험 관리 기법을 쉽게 통합할 수 있습니다.
- 비용 효율적인 솔루션: Alpaca API의 커미션 프리 거래 기능을 활용하여 거래 비용을 절감할 수 있습니다.
- Cons:
- 데이터 비용: Polygon API의 데이터 사용량에 따라 비용이 발생할 수 있습니다. 무료 플랜은 제한적입니다.
- API 의존성: Alpaca API 및 Polygon API의 안정성에 의존적입니다. API 장애 시 백테스팅 파이프라인이 중단될 수 있습니다.
- 과거 데이터의 한계: 과거의 성과가 미래의 성과를 보장하지 않습니다. 백테스팅 결과는 참고 자료로만 활용해야 합니다.
- 초기 설정 및 유지 관리: API 키 관리, 코드 유지 관리, 데이터 파이프라인 모니터링 등 초기 설정 및 지속적인 관리가 필요합니다.
6. FAQ
- Q: Polygon API의 데이터 사용량 제한은 어떻게 되나요?
A: Polygon API는 다양한 구독 플랜을 제공하며, 플랜에 따라 데이터 사용량 제한이 다릅니다. 자세한 내용은 Polygon 웹사이트를 참조하십시오. - Q: Alpaca API를 사용하여 실제 거래를 할 수 있나요?
A: 네, Alpaca API를 사용하여 실제 주식 거래를 할 수 있습니다. 하지만 먼저 계정을 개설하고 KYC (Know Your Customer) 인증을 완료해야 합니다. - Q: 백테스팅 결과를 어떻게 해석해야 하나요?
A: 백테스팅 결과는 투자 전략의 잠재적인 성과를 예측하는 데 도움이 될 수 있지만, 과거의 성과가 미래의 성과를 보장하지 않습니다. 백테스팅 결과를 다양한 시나리오와 결합하여 분석하고, 위험 관리 요소를 고려하여 투자 결정을 내려야 합니다. - Q: 이 파이프라인을 다른 금융 상품 (예: 암호화폐)에 적용할 수 있나요?
A: 네, Polygon API 및 Alpaca API가 지원하는 금융 상품에 따라 파이프라인을 수정하여 적용할 수 있습니다. 하지만 각 상품의 특성에 맞게 백테스팅 로직 및 위험 관리 기법을 조정해야 합니다.
7. Conclusion
이 글에서는 Polygon API와 Alpaca API를 활용하여 자동화된 주식 포트폴리오 백테스팅 파이프라인을 구축하는 방법을 자세히 설명했습니다. 이 파이프라인을 통해 투자 전략의 효율성을 평가하고, 위험을 관리하며, 데이터 기반의 투자 결정을 내릴 수 있습니다. 지금 바로 이 코드를 사용하여 자신만의 백테스팅 파이프라인을 구축하고, 투자 전략을 최적화하십시오. 추가적인 정보 및 API 사용법은 Alpaca API 공식 문서 및 Polygon API 공식 문서를 참조하십시오.
