개발/시리즈
나만의 자동매매 프로그램 구축하기 시리즈 - PyQt5로 거래소 API 저장하기 (1)
TF포지
2025. 4. 20. 18:10
CCXT 라이브러리부터 PyQt5 GUI 연동까지
이번 장은 CCXT 라이브러리를 기반으로 GUI 기반 자동매매 시스템 구축을 위해 PyQt5와 연동하시려는분들을 위해 구성되었습니다.
PyQt5 설치 -> CCXT 라이브러리 설치 -> API 연동까지 기본 개발 환경 설치 방법과 API 연동 방법을 순서대로 알려드립니다.
프로젝트 구조
📁 my_pyqt_app/
┣ api_key_diaglog.py # API 연동 메인 코드
┗ register_auth.ui # API 연동 UI전체 코드 확인 가능한 깃 주소 : https://github.com/bksj-ds/tfTutorial/tree/master/QtTutorial/QLabel
git clone https://github.com/bksj-ds/tfTutorial1단계: PyQt5 설치
아직 PyQt 설치를 못하신 분들은 아래 글을 참조해주세요.
2025.04.19 - [개발/PyQt5] - PyQt5 - PyQt5 개발환경 구성과 Designer 설치 (1)
[PyQt5 - PyQt5 개발환경 구성과 Designer 설치 (1)
설치부터 GUI 위젯까지이 가이드는 자동매매 구현을 위해 PyQt5를 처음 시작하는 분들을 위해 구성되었습니다.Anaconda 설치 → 가상환경 생성 → Qt Designer 실행까지 기본 개발 환경 설치 방법을 순
tradeforge.tistory.com](https://tradeforge.tistory.com/2)
2단계: CCXT 라이브러리 설치
# 가상 환경으로 진입
(base) C:\Users\User> activate pyqt5_env
# ccxt 라이브러리 설치
(pyqt5_env) C:\Users\User> pip install ccxt위에 설명된 명령어를 입력하시면 가상 환경(필자는 pyqt5_env)에 ccxt 라이브러리가 설치되고 사용 가능합니다.
CCXT 라이브러리를 왜 사용하는건가요?
자동매매 시스템을 구축하려면 거래소 API와 통신이 가능해야 합니다. 하지만 거래소마다 API 구조, 인증 방식, 응답 포맷이 모두 다르기 때문에, 여러 거래소를 동시에 다루거나 변경하려면 상당한 시간과 비용이 소모됩니다.
CCXT는 이러한 문제를 해결해주는 Python 기반 통합 API 라이브러리입니다.
라이브러리의 주요 특징
| 특징 | 설명 |
| 🧩 표준화된 인터페이스 | Binance, Bybit, Bitget, Kucoin 등 100개 이상 거래소의 API를 하나의 방식으로 통일 |
| 🔐 인증 처리 자동화 | API 키와 시크릿 설정만으로 인증을 간편하게 수행 |
| 📈 다양한 기능 지원 | fetchTicker(), createOrder(), fetchBalance() 등 주요 거래 기능 내장 |
| 🔄 실시간 시장 데이터 조회 | OHLCV, Ticker, Order Book, Positions 등 다양한 마켓 정보 실시간 조회 |
| 🧵 비동기 처리 가능 | asyncio 기반 자동매매 로직에 유연하게 연동 가능 |
💡 예시 코드: 현재 시세 조회
import ccxt
exchange = ccxt.bybit({
'apiKey': 'YOUR_API_KEY',
'secret': 'YOUR_SECRET',
'enableRateLimit': True,
'options': {
'adjustForTimeDifference': True,
'verbose': True,
'defaultType': 'future'
}
})
ticker = exchange.fetch_ticker('BTC/USDT:USDT')
print(f"현재 가격: {ticker['last']}")apiKey와 secret 키를 코드에 입력한 이후 코드를 실행하면 바이비트 선물 거래소의 코인 정보를 확인할 수 있습니다.
코드 실행 방법
코드를 실행하는 방법을 아래에서 확인할 수 있습니다.
# 가상 환경에 설치된 spyder IDE 실행
(pyqt5_env) C:\Users\User> spyder필자는 IDE로써 스파이더를 사용하므로, 해당 시리즈에서도 계속 사용할 예정입니다.
AI의 도움을 받고 싶은 분들은 Cursor를 사용하셔도 좋습니다.
Cursor AI 다운로드 링크 첨부드립니다.
커서 AI (https://www.cursor.com/downloads)
바이비트 API 발급 방법
(https://www.bybit.com/invite?ref=05GZV1Z)
아직 아이디가 없으신 경우 해당 링크로 접속하면 바이비트 가입이 가능합니다.
(https://www.bybit.com/app/user/api-management)
해당 링크로 접속하면 바이비트 API 발급 페이지로 이동합니다.
비트겟 API 발급 방법
아직 아이디가 없으신 경우 해당 링크로 접속하면 비트겟 가입이 가능합니다.
(https://www.bitget.com/asia/account/newapi)
해당 링크로 접속하면 비트겟 API 발급 페이지로 이동합니다.
3단계: PyQt5 GUI 개발
GUI 화면 구성하기
먼저 Qt Deisgner를 활용해서 API키를 입력받는 UI를 구성해보도록 하겠습니다.
파일 -> 새폼 -> Widget 을 클릭하여 새로운 위젯을 생성합니다.
칸이 너무 작아보여 너비를 480으로 수정해주겠습니다.
API키를 입력받기 위한 QLabel을 추가해주도록 하겠습니다.
특정 거래소(비트겟)의 경우 패스워드 입력이 필요하니 패스워드도 함께 추가해주었습니다.
좌측 패널에서 Text Edit도 끌어서 추가해줍니다.
입력된 텍스트를 저장하기 위해 좌측 패널의 Push Button을 추가해줍니다.
이렇게 대충 틀만 잡아주고 폼의 배경을 우클릭 -> 배치 -> 격자형으로 배치를 눌러주면 아주 편하게 틀을 잡아줄 수 있습니다.
자동으로 잡힌 디자인이 마음에 안드니 객체들의 높이를 조정해주고 폼 크기를 조정한 이후 마무리 하겠습니다.
모든 객체들을 클릭한 이후 maximumSize의 높이를 40으로 고정해주었습니다.
객체의 이름은 아래와 같이 설정되었습니다.
심화 단계 : PyQt5 GUI로 거래소 API 연동하기
⚠ 주의) 아래 내용은 개발 기초지식이 없는 경우 복잡하여 프로젝트를 중단시킬 가능성이 있으니, 그냥 코드를 갖다 쓰시는것이 정신건강에 이로울 수 있습니다.
필수 라이브러리 설치
# 가상 환경에 라이브러리 설치
(pyqt5_env) C:\Users\User> pip install ccxt
(pyqt5_env) C:\Users\User> pip install cryptographyGUI에 입력된 정보를 저장하기
저장된 register_auth.ui를 통해 사용자가 입력한 API 정보를 입력 받아보도록 하겠습니다.
API 관련 정보를 저장하는 과정은 아래와 같이 진행됩니다.
[PyQt5 GUI]
│ (API Key, Secret Key, Password)
▼
[ApiKeyDialog]
│
▼ saveButton.click()
[APIManager]
├── encrypt_api_keys() API 키를 암호화
├── decrypt_api_keys() API 키를 복호화
└── initialize_exchange() → CCXT 객체 생성GUI와 파이썬 코드를 연동하는 ApiKeyDialog 클래스 만들기
class ApiKeyDialog(QDialog):
def __init__(self, api_manager: APIManager, ui_path: str):
super().__init__()
self.api_manager = api_manager
# Load the UI
uic.loadUi(ui_path, self)
# Connect the save button to logic
self.saveButton.clicked.connect(self.save_api_keys)
def save_api_keys(self):
api_key = self.apiKeyEdit.toPlainText().strip()
secret_key = self.secretKeyEdit.toPlainText().strip()
password = self.passwordEdit.toPlainText().strip()
if not api_key or not secret_key:
QMessageBox.warning(self, "입력 오류", "API Key와 Secret Key를 모두 입력해주세요.")
return
try:
result = self.api_manager.encrypt_api_keys(api_key, secret_key, password)
if result:
QMessageBox.information(self, "성공", "API 키 저장 완료")
self.close()
else:
QMessageBox.warning(self, "실패", "API 키 저장에 실패했습니다.")
except Exception as e:
QMessageBox.critical(self, "오류", f"API 키 저장 중 오류 발생: {e}")라인별 설명
클래스 정의
class ApiKeyDialog(QDialog):- QDialog를 상속받는 API 키 입력용 다이얼로그 클래스를 정의합니다.
def __init__(self, api_manager: APIManager, ui_path: str):- api_manager: API 키 저장 및 암호화를 담당하는 APIManager 인스턴스
- ui_path: Qt Designer로 제작한 .ui 파일 경로
super().__init__()- 부모 클래스인 QDialog의 생성자 호출
self.api_manager = api_manager- 받은 APIManager를 인스턴스 변수로 저장합니다.
uic.loadUi(ui_path, self)- .ui 파일을 현재 다이얼로그에 로드하여 UI 요소들을 불러옵니다.
self.saveButton.clicked.connect(self.save_api_keys)- UI에 있는 저장 버튼(saveButton)을 save_api_keys 메서드와 연결합니다.
save_api_keys 메서드
def save_api_keys(self):- 저장 버튼 클릭 시 실행될 함수입니다.
api_key = self.apiKeyEdit.toPlainText().strip()
secret_key = self.secretKeyEdit.toPlainText().strip()
password = self.passwordEdit.toPlainText().strip()- UI에서 입력한 API Key, Secret Key, Password 값을 가져와 공백 제거 후 저장합니다.
if not api_key or not secret_key:
QMessageBox.warning(self, "입력 오류", "API Key와 Secret Key를 모두 입력해주세요.")
return- 필수 값인 API Key와 Secret Key가 비어 있다면 경고 메시지를 띄우고 중단합니다.
try:
result = self.api_manager.encrypt_api_keys(api_key, secret_key, password)- APIManager의 encrypt_api_keys 메서드를 호출하여 API 정보를 암호화합니다.
if result:
QMessageBox.information(self, "성공", "API 키 저장 완료")
self.close()- 성공적으로 암호화되면 메시지 박스로 성공을 알리고 다이얼로그를 닫습니다.
else:
QMessageBox.warning(self, "실패", "API 키 저장에 실패했습니다."- 암호화 또는 저장에 실패한 경우 사용자에게 실패 메시지를 보여줍니다.
except Exception as e:
QMessageBox.critical(self, "오류", f"API 키 저장 중 오류 발생: {e}")- 예외가 발생하면 에러 메시지를 출력합니다.
ApiKeyDialog에서 API 키를 암호화, 복호화 및 저장하는 ApiManager 클래스 만들기
API 매니저의 기능
- 아래에 구현된 API 매니저는 API 키의 저장, 호출, 암호화, 복호화 및 자동매매를 위한 ccxt 객체 생성을 담당하는 클래스입니다.
#-------------------------------
#
# API 매니저 통합 클래스
#
#-------------------------------
class APIManager:
def __init__(self, parent, exchange_name="bybit", testnet=False, file_path = "api_keys.txt"):
self.parent = parent # Reference to the main UI window
self.exchange_name = exchange_name
self.testnet = testnet
self.exchange = None
self.api_key = None
self.secret_key = None
self.api_password = None # Needed for Bitget
self.file_path = file_path or f"{exchange_name}_api_keys.enc"
self._initialize_encryption()
def _initialize_encryption(self):
"""Initialize encryption key and Fernet instance"""
# Create a directory for storing keys if it doesn't exist
os.makedirs('.keys', exist_ok=True)
# Path for the encryption key
key_path = os.path.join('.keys', '.master.key')
# Generate or load master key
if os.path.exists(key_path):
with open(key_path, 'rb') as key_file:
self.master_key = key_file.read()
else:
# Generate a random salt
self.salt = os.urandom(16)
# Use PBKDF2 to derive a key
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=self.salt,
iterations=100000,
)
# Generate a random master key
self.master_key = base64.urlsafe_b64encode(os.urandom(32))
# Save the master key
with open(key_path, 'wb') as key_file:
key_file.write(self.master_key)
# Save the salt
with open(os.path.join('.keys', '.salt'), 'wb') as salt_file:
salt_file.write(self.salt)
# Initialize Fernet for encryption/decryption
self.fernet = Fernet(self.master_key)
#########################
#
# API 연동 화면 대응
#
#########################
def load_api_key_if_exists(self, parent):
"""Load encrypted keys from the file if it exists and is readable."""
try:
api_key, secret_key, password = self.decrypt_api_keys()
if api_key and secret_key:
parent.apiKeyEdit.setPlainText(api_key)
parent.secretKeyEdit.setPlainText(secret_key)
if hasattr(parent, 'pwEdit') and password:
parent.pwEdit.setPlainText(password)
except Exception as e:
QMessageBox.warning(None, "오류", f"API 키 파일을 읽는 중 오류 발생: {e}")
def save_keys(self, parent):
"""Save API keys with encryption."""
# Retrieve input values
api_key = parent.apiKeyEdit.toPlainText().strip()
secret_key = parent.secretKeyEdit.toPlainText().strip()
password = parent.pwEdit.toPlainText().strip() if hasattr(parent, 'pwEdit') else None
if api_key and secret_key:
try:
# Save API keys with encryption (including password if provided)
success = self.encrypt_api_keys(api_key, secret_key, password)
if success:
QMessageBox.information(self.parent, "성공", "API 키 등록 성공")
parent.close()
else:
QMessageBox.warning(self.parent, "오류", "API 키 암호화 실패")
except Exception as e:
QMessageBox.warning(self.parent, "오류", f"API 키 저장 중 오류 발생: {e}")
else:
QMessageBox.warning(self.parent, "오류", "API 키와 Secret 키 모두 등록 필요")
#########################
#
# API 키 불러오기
#
#########################
def load_api_key(self):
"""Load encrypted API keys."""
try:
# Try to load encrypted keys
self.api_key, self.secret_key, self.api_password = self.decrypt_api_keys()
# If encryption load fails, try legacy format
if not self.api_key or not self.secret_key:
if os.path.exists(self.file_path.replace('.enc', '.txt')) and os.access(self.file_path.replace('.enc', '.txt'), os.R_OK):
with open(self.file_path.replace('.enc', '.txt'), "r") as file:
lines = file.readlines()
keys = {line.split("=")[0]: line.split("=")[1].strip() for line in lines if "=" in line}
self.api_key = keys.get("API_KEY", "")
self.secret_key = keys.get("SECRET_KEY", "")
self.api_password = keys.get("API_PASSWORD", "")
return bool(self.api_key and self.secret_key)
except Exception as e:
if hasattr(self.parent, 'log_message'):
self.parent.log_message(f"API 키 로드 실패: {e}")
return False
def get_api_keys(self):
"""Retrieve API keys securely."""
if not self.load_api_key():
QMessageBox.warning(self.parent, "오류", "API 키 파일을 찾을 수 없거나 읽을 수 없습니다.")
return False
if not self.api_key or not self.secret_key:
QMessageBox.warning(self.parent, "오류", "API 키와 Secret 키를 등록하세요.")
return False
if (self.exchange_name == "bitget" or self.exchange_name == "okx") and not self.api_password:
QMessageBox.warning(self.parent, "오류", "API 비밀번호가 필요합니다.")
return False
return True
def initialize_exchange(self):
"""Initialize API connection with the selected exchange."""
if not self.get_api_keys():
return
try:
exchange_class = getattr(ccxt, self.exchange_name)
exchange_params = {
'apiKey': self.api_key,
'secret': self.secret_key,
'enableRateLimit': True,
'options': {
'adjustForTimeDifference': True,
'verbose': True,
'defaultType': 'future'
},
}
if self.exchange_name in ['gateio']:
exchange_params.get('options').update({'defaultType': 'swap'}) # Bitget requires password
if self.exchange_name in ['bitget', 'okx']:
exchange_params['password'] = self.api_password # Bitget requires password
self.exchange = exchange_class(exchange_params)
if self.testnet:
self.exchange.set_sandbox_mode(True)
return self.exchange
except Exception as e:
return print(f'{e}')
def encrypt_api_keys(self, api_key, secret_key, password=None):
"""Encrypt API keys using Fernet symmetric encryption"""
try:
# Convert keys to JSON string with password for Bitget/OKX
keys_dict = {
"api_key": api_key,
"secret_key": secret_key,
"exchange": self.exchange_name,
"testnet": self.testnet
}
# Add password for Bitget/OKX
if password and self.exchange_name in ['bitget', 'okx']:
keys_dict["api_password"] = password
keys_json = json.dumps(keys_dict)
# Encrypt the JSON string
encrypted_data = self.fernet.encrypt(keys_json.encode())
# Save encrypted data to file
with open(self.file_path, 'wb') as f:
f.write(encrypted_data)
return True
except Exception as e:
if hasattr(self.parent, 'log_message'):
self.parent.log_message(f"❌ API 키 암호화 실패: {str(e)}")
return False
def decrypt_api_keys(self):
"""Decrypt API keys from encrypted file"""
try:
if not os.path.exists(self.file_path):
return None, None, None
# Read encrypted data
with open(self.file_path, 'rb') as f:
encrypted_data = f.read()
# Decrypt the data
decrypted_json = self.fernet.decrypt(encrypted_data)
keys_dict = json.loads(decrypted_json.decode())
# Return API key, secret key, and password (if exists)
return (
keys_dict.get("api_key"),
keys_dict.get("secret_key"),
keys_dict.get("api_password") # Will be None if not present
)
except Exception as e:
if hasattr(self.parent, 'log_message'):
self.parent.log_message(f"❌ API 키 복호화 실패: {str(e)}")
return None, None, None
def save_api_keys(self, api_key, secret_key):
"""Save API keys securely with encryption"""
return self.encrypt_api_keys(api_key, secret_key)
def load_api_keys(self):
"""Load and decrypt saved API keys"""
return self.decrypt_api_keys()APIManager 클래스 구조 및 동작 원리
1. 클래스 초기화
class APIManager:
def __init__(self, parent, exchange_name="bybit", testnet=False, file_path = "api_keys.txt"):| 항목 | 설명 |
parent |
메인 UI 참조 (PyQt5 기준) |
exchange_name |
거래소 이름 (예: bybit, binance) |
testnet |
테스트넷 여부 |
file_path |
API 키 저장 파일 경로 |
.fernet |
대칭키 암호화를 위한 Fernet 인스턴스 초기화 |
2. 암호화 키 초기화
def _initialize_encryption(self):.keys디렉토리를 생성하여master.key와salt를 저장합니다.- 암호화 키는
PBKDF2HMAC를 사용해 생성됩니다. - 암호화는
Fernet알고리즘을 기반으로 합니다.
3. 저장된 키 로딩 및 UI 연결
def load_api_key_if_exists(self, parent):- 암호화된 파일을 복호화하여 API Key, Secret Key, Password를 UI 필드에 채웁니다.
- UI의
apiKeyEdit,secretKeyEdit,pwEdit와 연결됩니다.
4. 사용자 입력 키 저장
def save_keys(self, parent):- 사용자가 입력한 API Key, Secret Key, Password를 가져와 저장합니다.
- 내부적으로
encrypt_api_keys()를 호출하여 파일로 안전하게 저장합니다.
5. API 키 불러오기
def load_api_key(self):- 암호화된 API 키 파일을 복호화해서
self.api_key,self.secret_key에 저장합니다. - 실패 시, 텍스트 백업 파일 (
.txt)을 대체로 읽습니다.
6. API 키 존재 여부 확인
def get_api_keys(self):- API 키 파일을 불러오고 모든 필수 항목이 존재하는지 검증합니다.
- 거래소에 따라
api_password가 필수인지도 체크합니다.
7. CCXT 연동
def initialize_exchange(self):ccxt라이브러리를 이용하여 거래소 API 객체를 초기화합니다.- 필요 시
sandbox mode(테스트 모드)를 활성화합니다. - 거래소별 옵션도 설정합니다 (예: Bitget은
password필요).
8. API 키 암호화 저장
def encrypt_api_keys(self, api_key, secret_key, password=None):- API Key, Secret Key, Password를 JSON으로 변환 후 암호화합니다.
- 암호화된 데이터를 파일(
.enc)로 저장합니다.
9. API 키 복호화
def decrypt_api_keys(self):- 저장된 암호화 파일을 읽어 복호화합니다.
- JSON 파싱을 통해 키 값을 추출하여 반환합니다.
10. 헬퍼 함수
def save_api_keys(self, api_key, secret_key):- 암호화 저장만 수행하는 래퍼 함수입니다.
def load_api_keys(self):- 암호화된 키를 읽어오는 래퍼 함수입니다.
이렇게 코드를 실행하면
그리고 저장버튼을 눌러보겠습니다.
그리고, txt 파일을 열어보면 암호화되어 잘 저장되어있는 모습을 확인할 수 있습니다.
이렇게 PyQt5를 기반으로 API키를 입력하고 자동매매를 위한 CCXT 객체 생성 과정을 알아보았습니다.
다음 포스팅에선 자산 정보를 GUI에 표시하는 방법을 알아보도록 하겠습니다.
이 글이 도움이 되셨다면 공감 ❤️ 과 궁금하신점들을 댓글로 ✍️ 부탁드립니다.