58802bdc5f
MTF 필터 백테스트, paper/live 체결, 빗썸 Private API 연동 및 운영 스크립트·설계 문서를 추가해 2단계 전략을 실거래 단계에 연결한다. Co-authored-by: Cursor <cursoragent@cursor.com>
144 lines
4.6 KiB
Python
144 lines
4.6 KiB
Python
"""빗썸 Private REST API — 잔고·주문."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
from deepcoin.api.bithumb_auth import auth_headers, dumps_params
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class BithumbPrivateClient:
|
|
"""빗썸 v2.1 Private API 클라이언트."""
|
|
|
|
def __init__(
|
|
self,
|
|
access_key: str,
|
|
secret_key: str,
|
|
base_url: str = "https://api.bithumb.com",
|
|
sleep_sec: float = 0.35,
|
|
retries: int = 3,
|
|
) -> None:
|
|
"""클라이언트를 초기화한다.
|
|
|
|
Args:
|
|
access_key: API Access Key.
|
|
secret_key: API Secret Key.
|
|
base_url: API 베이스 URL.
|
|
sleep_sec: 연속 요청 간 대기(초).
|
|
retries: 실패 시 재시도 횟수.
|
|
"""
|
|
if not access_key or not secret_key:
|
|
raise ValueError("BITHUMB_ACCESS_KEY / BITHUMB_SECRET_KEY 가 필요합니다.")
|
|
self.access_key = access_key
|
|
self.secret_key = secret_key
|
|
self.base_url = base_url.rstrip("/")
|
|
self.sleep_sec = sleep_sec
|
|
self.retries = retries
|
|
self._session = requests.Session()
|
|
|
|
def _request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
params: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""인증 요청을 수행한다."""
|
|
url = f"{self.base_url}{path}"
|
|
body = dumps_params(params) if params else None
|
|
headers = auth_headers(
|
|
self.access_key,
|
|
self.secret_key,
|
|
params=params,
|
|
)
|
|
last_error: Exception | None = None
|
|
for attempt in range(1, self.retries + 1):
|
|
try:
|
|
response = self._session.request(
|
|
method,
|
|
url,
|
|
data=body,
|
|
headers=headers,
|
|
timeout=30,
|
|
)
|
|
if response.status_code == 429:
|
|
wait = self.sleep_sec * attempt * 3
|
|
logger.warning("Rate limit 429 — %ss 대기", wait)
|
|
time.sleep(wait)
|
|
continue
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
time.sleep(self.sleep_sec)
|
|
if not isinstance(payload, (dict, list)):
|
|
raise ValueError(f"Unexpected response: {type(payload)}")
|
|
return payload if isinstance(payload, dict) else {"data": payload}
|
|
except Exception as exc:
|
|
last_error = exc
|
|
wait = self.sleep_sec * attempt * 2
|
|
logger.warning("Private API failed (%s/%s): %s", attempt, self.retries, exc)
|
|
time.sleep(wait)
|
|
if last_error is not None:
|
|
raise last_error
|
|
return {}
|
|
|
|
def get_accounts(self) -> list[dict[str, Any]]:
|
|
"""전체 계좌(잔고)를 조회한다."""
|
|
payload = self._request("GET", "/v1/accounts")
|
|
if isinstance(payload, list):
|
|
return payload
|
|
return payload.get("data", payload) if isinstance(payload.get("data"), list) else []
|
|
|
|
def get_balance(self, currency: str) -> tuple[float, float]:
|
|
"""통화별 잔고를 반환한다.
|
|
|
|
Returns:
|
|
(available, locked) 수량 또는 원화.
|
|
"""
|
|
currency = currency.upper()
|
|
for row in self.get_accounts():
|
|
if str(row.get("currency", "")).upper() == currency:
|
|
return float(row.get("balance", 0)), float(row.get("locked", 0))
|
|
return 0.0, 0.0
|
|
|
|
def market_buy_krw(self, market: str, krw_amount: float) -> dict[str, Any]:
|
|
"""시장가 매수 (원화 금액).
|
|
|
|
Args:
|
|
market: 예) KRW-BTC.
|
|
krw_amount: 매수 원화 금액.
|
|
|
|
Returns:
|
|
주문 응답 dict.
|
|
"""
|
|
params = {
|
|
"market": market,
|
|
"side": "bid",
|
|
"price": str(int(krw_amount)),
|
|
"ord_type": "price",
|
|
}
|
|
return self._request("POST", "/v1/orders", params=params)
|
|
|
|
def market_sell_volume(self, market: str, volume: float) -> dict[str, Any]:
|
|
"""시장가 매도 (코인 수량).
|
|
|
|
Args:
|
|
market: 예) KRW-BTC.
|
|
volume: 매도 수량.
|
|
|
|
Returns:
|
|
주문 응답 dict.
|
|
"""
|
|
params = {
|
|
"market": market,
|
|
"side": "ask",
|
|
"volume": f"{volume:.8f}".rstrip("0").rstrip("."),
|
|
"ord_type": "market",
|
|
}
|
|
return self._request("POST", "/v1/orders", params=params)
|