[암호화폐 자동매매] 업비트 OpenAPI를 활용하여 MACD 지표 기반 매매 샘플코드
업비트 OpenAPI를 활용하여 MACD 지표 기반 매매 구현을 위하여 ChatGPT에게 샘플 코드를 요청한 결과 공유.
import ccxt
import pandas as pd
from ta.trend import MACD
import time
# 업비트 API 키
API_KEY = 'your_access_key'
SECRET_KEY = 'your_secret_key'
# 업비트 객체 생성
upbit = ccxt.upbit({
'apiKey': API_KEY,
'secret': SECRET_KEY,
})
# 코인 티커 및 기본 설정
symbol = 'KRW-BTC' # 거래할 코인 (예: 비트코인)
interval = '1m' # 캔들 간격 (1분봉)
amount = 0.001 # 거래 수량
def fetch_ohlcv(symbol, interval):
"""OHLCV 데이터 가져오기"""
ohlcv = upbit.fetch_ohlcv(symbol, timeframe=interval, limit=100)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') # 타임스탬프 변환
return df
def calculate_macd(df):
"""MACD 및 Signal Line 계산"""
macd = MACD(df['close'], window_slow=26, window_fast=12, window_sign=9)
df['macd'] = macd.macd()
df['signal'] = macd.macd_signal()
return df
def trade(symbol, side, amount):
"""매수 또는 매도 주문 실행"""
try:
if side == 'buy':
order = upbit.create_market_buy_order(symbol, amount)
elif side == 'sell':
order = upbit.create_market_sell_order(symbol, amount)
print(f"{side.upper()} 주문 성공: {order}")
except Exception as e:
print(f"{side.upper()} 주문 실패: {e}")
def main():
"""메인 실행 로직"""
last_signal = None # 이전 신호 기록
while True:
try:
# OHLCV 데이터 가져오기
df = fetch_ohlcv(symbol, interval)
# MACD 계산
df = calculate_macd(df)
macd = df.iloc[-1]['macd']
signal = df.iloc[-1]['signal']
# 거래 신호 판단
if macd > signal and last_signal != 'buy':
print("매수 신호 발생!")
trade(symbol, 'buy', amount)
last_signal = 'buy'
elif macd < signal and last_signal != 'sell':
print("매도 신호 발생!")
trade(symbol, 'sell', amount)
last_signal = 'sell'
time.sleep(10) # 10초 대기 (API 호출 제한 방지)
except Exception as e:
print(f"에러 발생: {e}")
time.sleep(10)
if __name__ == "__main__":
main()
[필요절차]
1. 업비트 OpenAPI 등록 및 키 발급
2. 필요한 라이브러리 설치:
- ccxt: 암호화폐 거래소 API를 쉽게 사용할 수 있는 라이브러리.
- pandas: 데이터 분석 및 처리.
- ta: 기술적 분석(TA) 지표 계산.
3. MACD 계산
- MACD는 12일 지수이동평균(EMA)과 26일 EMA의 차이로 계산됩니다.
- 신호선(Signal Line)은 MACD의 9일 EMA입니다.
4. 거래 로직
- MACD가 신호선을 상향 돌파하면 매수.
- MACD가 신호선을 하향 돌파하면 매도.
[주요 사항]
1. API 호출 제한:
- 업비트 OpenAPI는 초당 호출 횟수 제한이 있으니, 적절히 time.sleep을 설정하세요.
2. 위험 관리
- 거래 금액과 수량은 충분히 테스트한 후 적용하세요.
- 실제 거래 전 모의 환경에서 반드시 테스트하세요.
3. 보안
- API 키를 코드에 직접 포함하지 말고, 환경 변수나 안전한 파일에 저장해 사용하세요.
4. 오류 처리
- API 응답 오류나 기타 예외에 대비한 예외 처리 로직을 추가하는 것이 중요합니다.
ChatGPT에게 수익률 데이터 표기 기능 추가 요청한 결과.
import ccxt
import pandas as pd
from ta.trend import MACD
import time
# 업비트 API 키
API_KEY = 'your_access_key'
SECRET_KEY = 'your_secret_key'
# 업비트 객체 생성
upbit = ccxt.upbit({
'apiKey': API_KEY,
'secret': SECRET_KEY,
})
# 코인 티커 및 기본 설정
symbol = 'KRW-BTC' # 거래할 코인 (예: 비트코인)
interval = '1m' # 캔들 간격 (1분봉)
amount = 0.001 # 거래 수량
# 수익률 기록용 변수
trade_history = [] # 거래 내역 기록 (매수/매도 가격, 수익 계산)
def fetch_ohlcv(symbol, interval):
"""OHLCV 데이터 가져오기"""
ohlcv = upbit.fetch_ohlcv(symbol, timeframe=interval, limit=100)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') # 타임스탬프 변환
return df
def calculate_macd(df):
"""MACD 및 Signal Line 계산"""
macd = MACD(df['close'], window_slow=26, window_fast=12, window_sign=9)
df['macd'] = macd.macd()
df['signal'] = macd.macd_signal()
return df
def trade(symbol, side, amount):
"""매수 또는 매도 주문 실행"""
try:
if side == 'buy':
order = upbit.create_market_buy_order(symbol, amount)
elif side == 'sell':
order = upbit.create_market_sell_order(symbol, amount)
print(f"{side.upper()} 주문 성공: {order}")
return order['price'] # 주문 성공 시 체결 가격 반환
except Exception as e:
print(f"{side.upper()} 주문 실패: {e}")
return None
def calculate_profit():
"""총 수익률 계산"""
total_profit = 0
for trade in trade_history:
if 'sell_price' in trade: # 매도가 완료된 거래만 수익 계산
profit = (trade['sell_price'] - trade['buy_price']) * trade['amount']
total_profit += profit
return total_profit
def main():
"""메인 실행 로직"""
last_signal = None # 이전 신호 기록
buy_price = None # 마지막 매수 가격
while True:
try:
# OHLCV 데이터 가져오기
df = fetch_ohlcv(symbol, interval)
# MACD 계산
df = calculate_macd(df)
macd = df.iloc[-1]['macd']
signal = df.iloc[-1]['signal']
# 거래 신호 판단
if macd > signal and last_signal != 'buy':
print("매수 신호 발생!")
buy_price = trade(symbol, 'buy', amount)
if buy_price:
trade_history.append({'buy_price': buy_price, 'amount': amount})
last_signal = 'buy'
elif macd < signal and last_signal != 'sell':
print("매도 신호 발생!")
sell_price = trade(symbol, 'sell', amount)
if sell_price and buy_price:
# 수익 계산 및 기록
trade_history[-1]['sell_price'] = sell_price
last_signal = 'sell'
# 현재 수익률 출력
total_profit = calculate_profit()
print(f"현재 총 수익: {total_profit} KRW")
time.sleep(10) # 10초 대기 (API 호출 제한 방지)
except Exception as e:
print(f"에러 발생: {e}")
time.sleep(10)
if __name__ == "__main__":
main()
[주요 변경점]
1. 거래 기록 관리
- trade_history 리스트에 매수/매도 내역을 저장합니다.
- 매수 시 buy_price와 거래 수량을 기록하고, 매도 시 sell_price를 추가합니다.
2. 수익 계산
- calculate_profit() 함수는 trade_history를 기반으로 매도 완료된 거래의 수익을 계산합니다.
- 현재 총 수익이 매 순환마다 출력됩니다.
3. 체결 가격 추적
- trade 함수에서 매수 또는 매도 후 체결된 가격을 반환해 기록에 사용합니다.
[실행 결과]
매수 신호 발생!
BUY 주문 성공: {'price': 45000000, 'amount': 0.001}
매도 신호 발생!
SELL 주문 성공: {'price': 46000000, 'amount': 0.001}
현재 총 수익: 1000 KRW
MACD 기울기의 평균값을 계산하여 조건을 더 강화하려면, MACD 기울기의 최근 값을 기반으로 이동평균을 계산합니다. 이를 통해 단기적인 노이즈를 제거하고 안정적인 추세를 판단할 수 있습니다.
아래는 수정된 코드입니다.
import ccxt
import pandas as pd
from ta.trend import MACD
import time
# 업비트 API 키
API_KEY = 'your_access_key'
SECRET_KEY = 'your_secret_key'
# 업비트 객체 생성
upbit = ccxt.upbit({
'apiKey': API_KEY,
'secret': SECRET_KEY,
})
# 코인 티커 및 기본 설정
symbol = 'KRW-BTC' # 거래할 코인 (예: 비트코인)
interval = '1m' # 캔들 간격 (1분봉)
amount = 0.001 # 거래 수량
slope_window = 5 # MACD 기울기의 평균을 계산할 기간
# 수익률 기록용 변수
trade_history = [] # 거래 내역 기록 (매수/매도 가격, 수익 계산)
def fetch_ohlcv(symbol, interval):
"""OHLCV 데이터 가져오기"""
ohlcv = upbit.fetch_ohlcv(symbol, timeframe=interval, limit=100)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') # 타임스탬프 변환
return df
def calculate_macd(df):
"""MACD 및 Signal Line 계산"""
macd = MACD(df['close'], window_slow=26, window_fast=12, window_sign=9)
df['macd'] = macd.macd()
df['signal'] = macd.macd_signal()
return df
def calculate_macd_slope(df, window):
"""MACD 기울기 및 평균 기울기 계산"""
df['macd_slope'] = df['macd'].diff() # MACD 기울기 (현재 값 - 이전 값)
df['macd_slope_avg'] = df['macd_slope'].rolling(window=window).mean() # 이동평균 기울기
return df
def trade(symbol, side, amount):
"""매수 또는 매도 주문 실행"""
try:
if side == 'buy':
order = upbit.create_market_buy_order(symbol, amount)
elif side == 'sell':
order = upbit.create_market_sell_order(symbol, amount)
print(f"{side.upper()} 주문 성공: {order}")
return order['price'] # 주문 성공 시 체결 가격 반환
except Exception as e:
print(f"{side.upper()} 주문 실패: {e}")
return None
def calculate_profit():
"""총 수익률 계산"""
total_profit = 0
for trade in trade_history:
if 'sell_price' in trade: # 매도가 완료된 거래만 수익 계산
profit = (trade['sell_price'] - trade['buy_price']) * trade['amount']
total_profit += profit
return total_profit
def main():
"""메인 실행 로직"""
last_signal = None # 이전 신호 기록
buy_price = None # 마지막 매수 가격
while True:
try:
# OHLCV 데이터 가져오기
df = fetch_ohlcv(symbol, interval)
# MACD 및 기울기 계산
df = calculate_macd(df)
df = calculate_macd_slope(df, slope_window)
macd = df['macd'].iloc[-1]
signal = df['signal'].iloc[-1]
macd_slope_avg = df['macd_slope_avg'].iloc[-1] # 평균 기울기
# 거래 신호 판단 (평균 기울기 조건 추가)
if macd > signal and macd_slope_avg > 0 and last_signal != 'buy':
print(f"매수 신호 발생! (MACD 평균 기울기: {macd_slope_avg:.6f})")
buy_price = trade(symbol, 'buy', amount)
if buy_price:
trade_history.append({'buy_price': buy_price, 'amount': amount})
last_signal = 'buy'
elif macd < signal and last_signal != 'sell':
print("매도 신호 발생!")
sell_price = trade(symbol, 'sell', amount)
if sell_price and buy_price:
# 수익 계산 및 기록
trade_history[-1]['sell_price'] = sell_price
last_signal = 'sell'
# 현재 수익률 출력
total_profit = calculate_profit()
print(f"현재 총 수익: {total_profit} KRW")
time.sleep(10) # 10초 대기 (API 호출 제한 방지)
except Exception as e:
print(f"에러 발생: {e}")
time.sleep(10)
if __name__ == "__main__":
main()
[주요 변경점]
1. MACD 기울기의 평균값 계산
- calculate_macd_slope 함수에서 macd.diff()를 이용해 기울기를 계산합니다. 이후 rolling(window=window).mean()을 사용해 지정된 기간(slope_window) 동안의 평균 기울기를 계산합니다.
2. 매수 조건 강화
- MACD 기울기의 **평균값(macd_slope_avg)**이 양수일 때만 매수하도록 조건을 추가했습니다.
3. 조정 가능한 이동평균 창 크기
- slope_window 변수로 이동평균 기울기의 계산 기간을 조정할 수 있습니다. (기본값: 5)
4. 출력 정보 추가
- 평균 기울기(macd_slope_avg)를 매수 신호 발생 시 출력하도록 추가했습니다.
MACD 기울기 대신 SMA 20봉 평균이 0보다 큰 경우로 수정.
import ccxt
import pandas as pd
from ta.trend import SMAIndicator
import time
# 업비트 API 키
API_KEY = 'your_access_key'
SECRET_KEY = 'your_secret_key'
# 업비트 객체 생성
upbit = ccxt.upbit({
'apiKey': API_KEY,
'secret': SECRET_KEY,
})
# 코인 티커 및 기본 설정
symbol = 'KRW-BTC' # 거래할 코인 (예: 비트코인)
interval = '1m' # 캔들 간격 (1분봉)
amount = 0.001 # 거래 수량
sma_window = 20 # SMA 계산에 사용할 봉 수
# 수익률 기록용 변수
trade_history = [] # 거래 내역 기록 (매수/매도 가격, 수익 계산)
def fetch_ohlcv(symbol, interval):
"""OHLCV 데이터 가져오기"""
ohlcv = upbit.fetch_ohlcv(symbol, timeframe=interval, limit=100)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') # 타임스탬프 변환
return df
def calculate_macd(df):
"""MACD 및 Signal Line 계산"""
macd = MACD(df['close'], window_slow=26, window_fast=12, window_sign=9)
df['macd'] = macd.macd()
df['signal'] = macd.macd_signal()
return df
def calculate_sma(df, window):
"""SMA(단순 이동 평균) 계산"""
sma = SMAIndicator(df['close'], window=window)
df['sma'] = sma.sma_indicator() # SMA 값을 데이터프레임에 추가
return df
def trade(symbol, side, amount):
"""매수 또는 매도 주문 실행"""
try:
if side == 'buy':
order = upbit.create_market_buy_order(symbol, amount)
elif side == 'sell':
order = upbit.create_market_sell_order(symbol, amount)
print(f"{side.upper()} 주문 성공: {order}")
return order['price'] # 주문 성공 시 체결 가격 반환
except Exception as e:
print(f"{side.upper()} 주문 실패: {e}")
return None
def calculate_profit():
"""총 수익률 계산"""
total_profit = 0
for trade in trade_history:
if 'sell_price' in trade: # 매도가 완료된 거래만 수익 계산
profit = (trade['sell_price'] - trade['buy_price']) * trade['amount']
total_profit += profit
return total_profit
def main():
"""메인 실행 로직"""
last_signal = None # 이전 신호 기록
buy_price = None # 마지막 매수 가격
while True:
try:
# OHLCV 데이터 가져오기
df = fetch_ohlcv(symbol, interval)
# MACD 및 SMA 계산
df = calculate_macd(df)
df = calculate_sma(df, sma_window)
macd = df['macd'].iloc[-1]
signal = df['signal'].iloc[-1]
sma_value = df['sma'].iloc[-1] # 20봉 SMA 값
sma_prev_value = df['sma'].iloc[-2] # 20봉 SMA 이전 값
# 거래 신호 판단 (SMA가 양수인 경우만 매수)
if macd > signal and sma_value > 0 and last_signal != 'buy':
print(f"매수 신호 발생! (SMA 20봉: {sma_value:.2f})")
buy_price = trade(symbol, 'buy', amount)
if buy_price:
trade_history.append({'buy_price': buy_price, 'amount': amount})
last_signal = 'buy'
elif macd < signal and last_signal != 'sell':
print("매도 신호 발생!")
sell_price = trade(symbol, 'sell', amount)
if sell_price and buy_price:
# 수익 계산 및 기록
trade_history[-1]['sell_price'] = sell_price
last_signal = 'sell'
# 현재 수익률 출력
total_profit = calculate_profit()
print(f"현재 총 수익: {total_profit} KRW")
time.sleep(10) # 10초 대기 (API 호출 제한 방지)
except Exception as e:
print(f"에러 발생: {e}")
time.sleep(10)
if __name__ == "__main__":
main()
[주요 변경점]
1. SMA 계산 함수 추가
- ta.trend.SMAIndicator를 사용해 20봉 단순 이동 평균(SMA)을 계산합니다.
2. 매수 조건 수정
- sma_value > 0 조건을 추가하여 SMA의 평균값이 양수일 때만 매수하도록 변경했습니다.
3. 출력 정보 추가
- 매수 신호 발생 시 20봉 SMA 값을 출력하도록 추가했습니다.
4. 기본 설정
- SMA 계산에 사용할 봉 수(sma_window)는 기본값으로 20을 사용하며, 필요에 따라 조정 가능합니다.
ChatGPT가 제시해준 코드에 대한 검증을 추가적으로 수행하길 권장.
끝.
댓글
댓글 쓰기