파이썬으로 업비트·빗썸 상장 봇 만들기

pip install "websockets>=14" ccxt

# websockets>=14
import asyncio, json, websockets

API_KEY = "YOUR_KEY"
FEED_URL = "wss://kr.cryptolisting.ws?cex=upbit,bithumb"

async def listen():
    attempt = 0
    while True:
        try:
            async with websockets.connect(
                FEED_URL, additional_headers={"X-API-Key": API_KEY}
            ) as ws:
                attempt = 0
                while True:
                    # heartbeat가 30초마다 오므로 60초 동안 아무것도 없으면 끊긴 연결
                    raw = await asyncio.wait_for(ws.recv(), timeout=60)
                    await handle(json.loads(raw))
        except websockets.ConnectionClosed as e:
            reason = e.rcvd.reason if e.rcvd else ""
            if reason in ("key_expired", "key_invalidated"):
                print(f"키 거부: {reason}")
                return
        except websockets.InvalidStatus as e:
            status = e.response.status_code
            if status in (401, 403):
                print(f"키 거부: HTTP {status}")
                return
            print(f"핸드셰이크 거부: HTTP {status}")   # 429: 간격을 두고 재시도
        except (asyncio.TimeoutError, OSError) as e:
            print(f"연결 끊김: {e!r}")
        await asyncio.sleep(min(2 ** attempt, 300))
        attempt += 1

{
  "type": "announcement",
  "title": "플루언트(BLEND) 신규 거래지원 안내 (KRW, BTC, USDT 마켓)",
  "ticker": "BLEND",
  "publisher": "upbit",
  "listingType": "spot_listing",
  "detectedTimestampUs": 1710345000005000,
  "dispatchTimestampUs": 1710345000006000,
  "abnormalDetectionLatency": false,
  "markets": "KRW,BTC,USDT"
}

BASE_SIZE_USDT = 50     # 토큰당 주문 금액
NON_KRW_FACTOR = 0.5    # BTC/USDT 마켓만 열리는 상장에 대한 본인의 정책

def order_size(msg):
    markets = msg.get("markets")   # 업비트 현물 상장에만 존재, 없으면 알 수 없음
    if markets is None or "KRW" in markets.split(","):
        return BASE_SIZE_USDT
    return BASE_SIZE_USDT * NON_KRW_FACTOR

async def route(msg):
    lt = msg["listingType"]
    tickers = [t for t in msg["ticker"].split(",") if t]
    if not tickers:
        return                                   # SpeedTrial: 티커 가림
    if lt == "spot_listing":
        size = order_size(msg)
        await asyncio.gather(*(buy(t, size, msg) for t in tickers[:MAX_TOKENS_PER_EVENT]))
    elif lt == "caution_released":
        print(f"관심 목록: {msg['publisher']} 유의 종목 해제 {tickers}")
    elif lt == "spot_delisting":
        await asyncio.gather(*(reduce_exposure(t) for t in tickers))

import ccxt.async_support as ccxt

venue = ccxt.binance({"apiKey": "VENUE_KEY", "secret": "VENUE_SECRET"})

def resolve(ticker):
    # 현물 페어 먼저, 그다음 USDT 마진 무기한 선물
    for symbol in (f"{ticker}/USDT", f"{ticker}/USDT:USDT"):
        if symbol in venue.markets:
            return symbol
    return None

async def refresh_markets():
    while True:
        await asyncio.sleep(300)
        await venue.load_markets(reload=True)

# 업비트·빗썸 상장 봇 — pip install "websockets>=14" ccxt
import asyncio, json, websockets
import ccxt.async_support as ccxt

API_KEY = "YOUR_KEY"
FEED_URL = "wss://kr.cryptolisting.ws?cex=upbit,bithumb"
DRY_RUN = True
BASE_SIZE_USDT = 50
NON_KRW_FACTOR = 0.5
MAX_TOKENS_PER_EVENT = 3

venue = ccxt.binance({"apiKey": "VENUE_KEY", "secret": "VENUE_SECRET"})
seen = set()


def order_size(msg):
    markets = msg.get("markets")   # 업비트 현물 상장에만 존재, 없으면 알 수 없음
    if markets is None or "KRW" in markets.split(","):
        return BASE_SIZE_USDT
    return BASE_SIZE_USDT * NON_KRW_FACTOR


def resolve(ticker):
    for symbol in (f"{ticker}/USDT", f"{ticker}/USDT:USDT"):
        if symbol in venue.markets:
            return symbol
    return None


async def buy(ticker, size_usdt, msg):
    symbol = resolve(ticker)
    if symbol is None:
        print(f"건너뜀 {ticker}: {venue.id}에 마켓 없음")
        return
    try:
        price = (await venue.fetch_ticker(symbol)).get("last")
        if not price:
            print(f"건너뜀 {symbol}: 아직 체결가 없음")
            return
        amount = float(venue.amount_to_precision(symbol, size_usdt / price))
        if DRY_RUN:
            print(f"[드라이런] 매수 {amount} {symbol} (~{size_usdt} USDT), {msg['publisher']} 상장")
            return
        order = await venue.create_market_buy_order(symbol, amount)
        print(f"매수 완료 {symbol}: 주문 {order['id']}")
    except ccxt.BaseError as e:
        print(f"주문 실패 {symbol}: {e}")


async def reduce_exposure(ticker):
    symbol = f"{ticker}/USDT"
    if symbol not in venue.markets:
        return
    try:
        held = (await venue.fetch_balance()).get("free", {}).get(ticker) or 0
        if held and DRY_RUN:
            print(f"[드라이런] 매도 {held} {symbol}, 거래지원 종료 공지")
        elif held:
            await venue.create_market_sell_order(symbol, float(venue.amount_to_precision(symbol, held)))
    except ccxt.BaseError as e:
        print(f"축소 실패 {symbol}: {e}")


async def route(msg):
    lt = msg["listingType"]
    tickers = [t for t in msg["ticker"].split(",") if t]
    tickers = [t for t in tickers if (msg["publisher"], lt, t) not in seen]
    seen.update((msg["publisher"], lt, t) for t in tickers)
    if not tickers:
        return                                   # 중복이거나 SpeedTrial 가림
    if lt == "spot_listing":
        size = order_size(msg)
        await asyncio.gather(*(buy(t, size, msg) for t in tickers[:MAX_TOKENS_PER_EVENT]))
    elif lt == "caution_released":
        print(f"관심 목록: {msg['publisher']} 유의 종목 해제 {tickers}")
    elif lt == "spot_delisting":
        await asyncio.gather(*(reduce_exposure(t) for t in tickers))


async def handle(msg):
    kind = msg.get("type")
    if kind == "welcome":
        print(f"연결됨: tier={msg.get('tier')} cex={msg.get('allowedCex')}")
    elif kind == "renewal_notice":
        print(msg.get("title"))
    elif kind == "announcement":              # test_announcement는 절대 매매하지 않음
        if msg.get("abnormalDetectionLatency"):
            print(f"표시된 이벤트, 매매 안 함: {msg.get('title')}")
            return
        await route(msg)


async def listen():
    attempt = 0
    while True:
        try:
            async with websockets.connect(
                FEED_URL, additional_headers={"X-API-Key": API_KEY}
            ) as ws:
                attempt = 0
                while True:
                    raw = await asyncio.wait_for(ws.recv(), timeout=60)
                    await handle(json.loads(raw))
        except websockets.ConnectionClosed as e:
            reason = e.rcvd.reason if e.rcvd else ""
            if reason in ("key_expired", "key_invalidated"):
                print(f"키 거부: {reason}")
                return
        except websockets.InvalidStatus as e:
            status = e.response.status_code
            if status in (401, 403):
                print(f"키 거부: HTTP {status}")
                return
            print(f"핸드셰이크 거부: HTTP {status}")   # 429: 간격을 두고 재시도
        except (asyncio.TimeoutError, OSError) as e:
            print(f"연결 끊김: {e!r}")
        await asyncio.sleep(min(2 ** attempt, 300))
        attempt += 1


async def refresh_markets():
    while True:
        await asyncio.sleep(300)
        try:
            await venue.load_markets(reload=True)
        except ccxt.BaseError as e:
            print(f"마켓 갱신 실패: {e}")


async def main():
    await venue.load_markets()
    refresher = asyncio.create_task(refresh_markets())
    try:
        await listen()
    finally:
        refresher.cancel()
        await venue.close()


asyncio.run(main())