import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import requests
pandas : 데이터 조작 및 분석을 위한 라이브러리
numpy : 수치 연산을 위한 라이브러리
matplotlib.pyplot : 데이터 시작화를 위한 라이브러리
requests : http 요청을 보내기 위한 라이브러리
def fetch_crypto_data(coin_id, vs_currency, days):
"""
CoinGecko API를 사용하여 암호화폐의 과거 가격 데이터를 수집합니다.
Parameters:
- coin_id: 암호화폐의 ID (예: 'bitcoin')
- vs_currency: 비교 통화 (예: 'usd')
- days: 과거 데이터 수집 기간 (일 단위)
Returns:
- df: 수집된 가격 데이터가 포함된 DataFrame
"""
url = f"https://api.coingecko.com/api/v3/coins/{coin_id}/market_chart"
params = {'vs_currency': vs_currency, 'days': days}
response = requests.get(url, params=params)
if response.status_code != 200:
print(f"Error: API request failed with status code {response.status_code}")
return None
data = response.json()
if 'prices' not in data:
print("Error: 'prices' key not found in the API response")
print("Response data:", data)
return None
prices = data['prices']
df = pd.DataFrame(prices, columns=['timestamp', 'price'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
return df
coin_id, vs_currency, days를 인자로 받아 해당 암호화폐의 과거 가격 데이터를 수집
api 요청이 성공하면, 반환된 json데이터에서 prices 키의 값을 추출하여 데이터 프레임으로 반환
타임스탬프를 datetime 형식으로 변환하고, 이를 인덱스로 설정
def calculate_indicators(df, sma_window=20, bb_window=20, bb_std=2, rsi_period=14):
"""
이동평균선, 볼린저 밴드, RSI를 계산하여 DataFrame에 추가합니다.
Parameters:
- df: 가격 데이터가 포함된 DataFrame
- sma_window: 이동평균선 기간 (기본값: 20)
- bb_window: 볼린저 밴드 기간 (기본값: 20)
- bb_std: 볼린저 밴드 표준편차 배수 (기본값: 2)
- rsi_period: RSI 기간 (기본값: 14)
Returns:
- df: 지표가 추가된 DataFrame
"""
# 이동평균선 (SMA)
df['SMA'] = df['price'].rolling(window=sma_window).mean()
# 볼린저 밴드
df['BB_Middle'] = df['price'].rolling(window=bb_window).mean()
df['BB_Std'] = df['price'].rolling(window=bb_window).std()
df['BB_Upper'] = df['BB_Middle'] + (df['BB_Std'] * bb_std)
df['BB_Lower'] = df['BB_Middle'] - (df['BB_Std'] * bb_std)
# RSI
delta = df['price'].diff()
gain = delta.where(delta > 0, 0)
loss = -delta.where(delta < 0, 0)
avg_gain = gain.rolling(window=rsi_period).mean()
avg_loss = loss.rolling(window=rsi_period).mean()
rs = avg_gain / avg_loss
df['RSI'] = 100 - (100 / (1 + rs))
return df
지표를 통해서 계산한다.
이동평균선 (SMA) : 지정된 기간 (sma_window) 동안의 가격 평균을 계산한다.
볼린저밴드 : 중간선 (BB_Middle), 상단선 (BB_Upper), 하단선 ( BB_Lower)을 계산한다.
RSI : 가격 변화의 평균 상승폭과 평균 하락폭을 기반으로 RSI 값을 계산한다.
df['SMA'] = df['price'].rolling(window=sma_window).mean()
rolling ( window=sma_window ) : pandas의 rolling 함수는 지정된 윈도우 크기 만큼 이동하면서 계산을 수행한다. 여기서는 sma_window 기간 동안의 이동 윈도우를 생성한다.
mean() : 각 윈도우에 대해서 평균을 계산하여 이동평균선을 구한다.
df['BB_Middle'] = df['price'].rolling(window=bb_window).mean()
중간선 : bb_window 기간 동안의 이동 평균선을 계산한다.
delta = df['price'].diff()
delta는 가격 변화량이다.
diff() 함수는 현재 값과 이전 값의 차이를 계산한다. 이를 통해서 각 시점의 가격 변화량을 구한다.
gain = delta.where(delta > 0, 0)
loss = -delta.where(delta < 0, 0)
상승분 gain 과 하락분 loss
where 함수는 조건을 만족하는 경우 해당 값을 유지하고, 그렇지 않은 경우 지정된 값으로 대체한다.
gain : 가격이 상승한 경우에 해당 상승분을 유지하고, 그렇지 않으면 0으로 대체
loss : 가격이 하락한 경우 해당 하락분의 절대값을 유지하고, 그렇지 않으면 0으로 대
avg_gain = gain.rolling(window=rsi_period).mean()
avg_loss = loss.rolling(window=rsi_period).mean()
평균 상승분과 평균 하락분
rsi_period 기간 동안의 평균 상승분과 평균 하락분을 계산한다.
메서드 정리_
1. rolling(window)
지정된 윈도우 크기만큼 이동하면서 계산을 수행하는 함수, 이동평균, 이동 표준편차 등을 계산할 때 사용한다.
2. mean()
평균을 계산
3. std()
표준편차를 계산
4. diff()
현재 값과 이전 값의 차이를 계산
5. where(조건, other)
조건을 만족하는 경우 해당 값을 유지하고, 그렇지 않은 경우 other로 대체한다.
def generate_signals(df):
"""
매수 및 매도 신호를 생성하여 DataFrame에 추가합니다.
Parameters:
- df: 지표가 포함된 DataFrame
Returns:
- df: 매매 신호가 추가된 DataFrame
"""
df['Signal'] = 0
# 매수 신호: RSI가 40 이상이고 가격이 이동평균선 위로 돌파할 때
df.loc[(df['RSI'] >= 40) & (df['price'] > df['SMA']) & (df['price'].shift(1) <= df['SMA'].shift(1)), 'Signal'] = 1
# 매도 신호: 가격이 볼린저 밴드 상단에 도달할 때
df.loc[df['price'] >= df['BB_Upper'], 'Signal'] = -1
return df
매매 신호를 생성한다.
매수를 1 매도를 -1로 친다.
# 4. 백테스팅
def backtest_strategy(df):
"""
전략의 성과를 백테스팅하여 수익률을 계산합니다.
Parameters:
- df: 매매 신호가 포함된 DataFrame
Returns:
- df: 백테스팅 결과가 추가된 DataFrame
"""
# 1. 매매 포지션 설정
df['Position'] = df['Signal'].replace(to_replace=0, method='ffill').shift()
# 2. 시장 수익률 계산
df['Market Return'] = df['price'].pct_change()
# 3. 전략 수익률 계산
df['Strategy Return'] = df['Market Return'] * df['Position']
# 4. 누적 시장 수익률 계산
df['Cumulative Market Return'] = (1 + df['Market Return']).cumprod()
# 5. 누적 전략 수익률 계산
df['Cumulative Strategy Return'] = (1 + df['Strategy Return']).cumprod()
return df
주어진 backtest_strategy 함수는 주가 데이터와 매매 신호를 기반으로 투자 전략의 성과를 평가한다.
1. 매매 포지션 설정
df['signal'] 열에는 매수 +1 매도 -1, 중립 0 의 신호가 포함되어있다.
replace(to_replace=0, method='ffill')은 중립 신호 0 을 이전의 매수 또는 매도 신호로 채운다. 이는 중립 신호가 나올 때 이전의 포지션을 유지함을 의미한다.
shift()는 포지션을 하루뒤로 이동시켜서, 현재의 신호가 다음날의 포지션에 반영되도록 한다.
2. 시장 수익률 계산
pct_change() 함수는 각 날짜의 가격 변동률을 계산한다.
이것은 시장의 일일 수익률이다.
3. 전략 수익률 계산
시장 수익률에 포지션을 곱하여 일일 수익률을 계산한다.
예를 들어서 매수 포지션 +1 일때는 시장 수익률을 그대로 반여하고, 매도 포지션 -1 일때는 시장 수익률의 반대 방향으로 수익이 발생한다.
4. 누적 시장 수익률 계산
(1 + df['Market Return'])은 각 일일 수익률에 1을 더하여 일일 수익 성장률을 계산한다.
cumprod() 함수는 이러한 일일 성장률의 누적 곱을 계산하여, 시작시점부터 현재까지의 누적 시장 수익률을 나타낸다.
5. 누적 전략 수익률 계산
마찬가지이다.