Unifox 시연회 프로젝트 보고서
제작동기
겨울 방학 프로젝트로 개발했었던 트레이딩 봇 프로젝트가 있습니다. 모든 아이디어를 실현시키고 못하고 제출했던 프로젝트입니다. 그래서 이번 시연회 프로젝트를 통해 기존 트레이딩 봇을 더 강력하게 발전시켜보고 싶다는 생각으로 시작하게 되었습니다.
프로젝트 설명
이 프로젝트는 upbit api를 활용하여 자동으로 거래를 수행하는 암호화폐 트레이딩 봇입니다.
도지코인(DOGE)을 대상으로 하며, 24시간 지속적으로 매매를 수행합니다.
실시간으로 시장 가격을 모니터링하며 설정한 가격 범위 내에서 움직임이 감지될 때 매수 결정을 내립니다.
매수한 상태에서 일정 비율 이상의 손실이 발생하면 즉시 손절해서 추가적인 손실를 방지하고, 목표 수익에 도달하면 자동으로 익절해 수익을 얻습니다.
api와 같은 민감한 정보는 env 파일을 통해 안전하게 관리됩니다.
소스코드
환경설정
import jwt import hashlib import os import dotenv import requests import uuid import random import time from urllib.parse import urlencode, unquote dotenv.load_dotenv() # 코드를 실행하면 정말 실제 돈으로 코인이 사지기 때문에 # access_key, secret_key, server_url에 있는 민감한 정보는 dotenv파일에 있습니다. # dotenv파일은 제출하지 않았습니다. # 실행하면 오류가 나지만 버그때문이 아니라는점을 알아주세요 access_key = os.environ['UPBIT_OPEN_API_ACCESS_KEY'] secret_key = os.environ['UPBIT_OPEN_API_SECRET_KEY'] server_url = os.environ['UPBIT_OPEN_API_SERVER_URL'] # 매수 상태 확인 플래그 is_position_open = False # 매수 가격 저장 변수 entry_price = 0 # 손절 비율 STOP_LOSS_RATIO = 0.03 # 3% 손실 시 손절 # 익절 비율 TAKE_PROFIT_RATIO = 0.05 # 5% 이익 시 익절
헤더 생성하기
def make_headers(data=None): payload = { 'access_key': access_key, 'nonce': str(uuid.uuid4()), } if data != None: query_string = unquote(urlencode(data, doseq=True)).encode("utf-8") m = hashlib.sha512() m.update(query_string) query_hash = m.hexdigest() payload['query_hash'] = query_hash payload['query_hash_alg'] = 'SHA512' jwt_token = jwt.encode(payload, secret_key) authorization = 'Bearer {}'.format(jwt_token) headers = { 'Authorization': authorization, } return headers
시장가 매수
def buy_price_order(price): params = { 'market': 'KRW-DOGE', 'side': 'bid', 'ord_type': 'price', 'price': price, } headers = make_headers(data = params) res = requests.post(server_url + '/v1/orders', json=params, headers=headers) return res
지정가 매수
def buy_limit_order(price, volume): params = { 'market': 'KRW-DOGE', 'side': 'bid', 'ord_type': 'limit', 'price': price, 'volume': volume, } headers = make_headers(data = params) res = requests.post(server_url + '/v1/orders', json=params, headers=headers) return res
시장가 매도
def sell_market_order(volume): params = { 'market':'KRW-DOGE', 'side': 'ask', 'ord_type': 'market', 'volume': volume, } headers = make_headers(data = params) res = requests.post(server_url + '/v1/orders', json=params, headers=headers) return res
지정가 매도
def sell_limit_order(price, volume): params = { 'market':'KRW-DOGE', 'side': 'ask', 'ord_type': 'limit', 'price': price, 'volume': volume, } headers = make_headers(data = params) res = requests.post(server_url + '/v1/orders', json=params, headers=headers) return res
개별 주문건 조회
def get_uuid_order(order_uuid): params = { 'uuid': order_uuid } headers = make_headers(data = params) res = requests.get(server_url + '/v1/order', params=params, headers=headers) return res
계좌 조회
params = {} headers = make_headers() res = requests.get(server_url + '/v1/accounts', params=params, headers=headers) return res
현재가 정보 가져오기
def get_ticker_info(ticker): params = { "markets": ticker } res = requests.get(server_url + "/v1/ticker", params=params) return res
DOGE 보유 여부
def check_position(): balances = get_balance().json() doge_balance = next((float(item['balance']) for item in balances if item['currency'] == 'DOGE'), 0) return doge_balance > 0, doge_balance
start 함수
def start(): global is_position_open, entry_price print("===== 암호화폐 자동 거래 시작 =====") print(f"손절 비율: {STOP_LOSS_RATIO*100}%") print(f"익절 비율: {TAKE_PROFIT_RATIO*100}%") # 시작할 때 기존 포지션 확인 is_position_open, _ = check_position() if is_position_open: print("이미 DOGE 코인을 보유하고 있습니다.") # 평균 매수가 계산 기능 추가 while True: try: # 초기 가격 init_price = get_ticker_info('KRW-DOGE').json()[0]['trade_price'] rand_num = random.randint(2, 5) start_time = time.time() time.sleep(rand_num) current_price = get_ticker_info('KRW-DOGE').json() if type(current_price) is list: current_price = current_price[0] current_price = current_price['trade_price'] print(f"현재 가격: {current_price}") # 포지션 확인 is_position_open, doge_balance = check_position() # 이미 매수 상태라면, 손절/익절 체크 if is_position_open and entry_price > 0: profit_ratio = (current_price - entry_price) / entry_price print(f"현재 수익률: {profit_ratio*100:.2f}%") # 손절 조건 if profit_ratio <= -STOP_LOSS_RATIO: print(f"손절 조건 충족! 손실률: {profit_ratio*100:.2f}%") sell_market_order(doge_balance) print(f"{doge_balance} DOGE를 {current_price}원에 손절 매도했습니다.") is_position_open = False entry_price = 0 continue # 익절 조건 elif profit_ratio >= TAKE_PROFIT_RATIO: print(f"익절 조건 충족! 수익률: {profit_ratio*100:.2f}%") sell_market_order(doge_balance) print(f"{doge_balance} DOGE를 {current_price}원에 익절 매도했습니다.") is_position_open = False entry_price = 0 continue # 매수 신호 확인 if not is_position_open and (init_price - init_price * 0.001) <= current_price and current_price <= (init_price + init_price * 0.001): # 현재 금액 KRW_BALANCE = float(get_balance().json()[0]['balance']) # 전략 1: 시장가 매수 (50%) market_amount = KRW_BALANCE * 0.5 CHARGED_MARKET_AMOUNT = market_amount * (1 - 0.0005)# 수수료 고려 print(f"{current_price}원에 시장가 매수합니다.") res = buy_price_order(CHARGED_MARKET_AMOUNT) if res.status_code == 201: # 주문 성공 order_uuid = res.json()["uuid"] order_info = get_uuid_order(order_uuid).json() # 해당 주문의 uuid 가져오기 # uuid로 주문 정보 가져오기 # 포지션 상태 업데이트 is_position_open = True entry_price = current_price print(f"시장가 매수 주문 완료: {CHARGED_MARKET_AMOUNT}원") # 전략 2: 지정가 매수 (30% - 현재가보다 0.5% 낮게) limit_price_1 = current_price * 0.995 limit_amount_1 = KRW_BALANCE * 0.3 limit_volume_1 = limit_amount_1 / limit_price_1 print(f"{limit_price_1}원에 지정가 매수 주문합니다.") buy_limit_order(limit_price_1, limit_volume_1) # 전략 3: 지정가 매수 (20% - 현재가보다 1% 낮게) limit_price_2 = current_price * 0.99 limit_amount_2 = KRW_BALANCE * 0.2 limit_volume_2 = limit_amount_2 / limit_price_2 print(f"{limit_price_2}원에 지정가 매수 주문합니다.") buy_limit_order(limit_price_2, limit_volume_2) # 매수 후 모니터링 while is_position_open: current_price = get_ticker_info('KRW-DOGE').json() if type(current_price) is list: current_price = current_price[0] current_price = current_price['trade_price'] print(f"현재 가격: {current_price}") # 수익률 계산 profit_ratio = (current_price - entry_price) / entry_price print(f"현재 수익률: {profit_ratio*100:.2f}%") # 손절 조건 if profit_ratio <= -STOP_LOSS_RATIO: _, doge_balance = check_position() print(f"손절 조건 충족! 손실률: {profit_ratio*100:.2f}%") sell_market_order(doge_balance) print(f"{doge_balance} DOGE를 {current_price}원에 손절 매도했습니다.") is_position_open = False entry_price = 0 break # 익절 조건 elif profit_ratio >= TAKE_PROFIT_RATIO: _, doge_balance = check_position() print(f"익절 조건 충족! 수익률: {profit_ratio*100:.2f}%") sell_market_order(doge_balance) print(f"{doge_balance} DOGE를 {current_price}원에 익절 매도했습니다.") is_position_open = False entry_price = 0 break # 원래 로직 (가격 범위 이탈 시 매도) if current_price <= (init_price - init_price * 0.001) or current_price >= (init_price + init_price * 0.001): balances = get_balance().json() coin_volume = next((item['balance'] for item in balances if item['currency'] == 'DOGE'), 0) sell_market_order(coin_volume) print(f"{coin_volume} DOGE를 {current_price}원에 매도했습니다.") is_position_open = False entry_price = 0 break time.sleep(0.5) time.sleep(1) except Exception as e: print(f"오류 발생: {e}") time.sleep(5) if __name__ == "__main__": start()
시연영상
느낀점
이번 프로젝트를 테스트하면서 2500원을 잃었습니다. 완벽하지 않은 알고리즘으로 통해서 나온 손해가 발생했는데요. 뼈아픈 손실이 있었지만, 완벽하지 않은 봇의 현실을 체감할 수 있어 뜻 깊은 경험을 해볼 수 있었습니다.
프로젝트를 하며 upbit api와 더 친해질 수 있었고, 기상청이나 챗gpt 등의 api를 활용하여 또 다른 프로젝트를 해보고 싶다는 생각을 하게 되었습니다. 앞으로 더 많은 도전을 통해 다양한 경험을 하고 싶습니다.