암호화폐 거래소 중 최대 규모인 Binance에서 제공하는 주문 데이터 구조는 선물(Futures), 현물(Spot), 마진(Margin) 각市場마다 상이한 스키마를 가집니다. 저는 최근 트레이딩 봇 개발 시 전체 주문 수명주기(Placement → Acknowledgment →_fill → Partial Fill → Cancellation → Rejection)를 하나의 일관된 데이터 흐름으로 관리해야 하는 프로젝트를 진행하면서, Binance 공식 문서에서 누락된 세부 사항들을 직접 테스트하며 정리했습니다.
본 가이드에서는 Binance API 주문 데이터 구조를 REST 엔드포인트, WebSocket 스트림, Python 라이브러리 세 가지 관점에서 비교하고, AI 기반 주문 분석 파이프라인 구축 시 HolySheep AI를 활용하는 방법을 설명합니다.
HolySheep AI vs Binance 공식 API vs 기타 중계 서비스 비교
| 비교 항목 | Binance 공식 API | HolySheep AI 게이트웨이 | 기타 중계 서비스 |
|---|---|---|---|
| API 엔드포인트 | api.binance.com (단일) | https://api.holysheep.ai/v1 | 도메인별 분리 |
| AI 모델 통합 | 없음 (거래 전용) | GPT-4.1, Claude, Gemini, DeepSeek | 제한적 |
| 주문 데이터 분석 | Raw 데이터만 제공 | AI 기반 패턴 인식·이상 탐지 | 없음 또는 유료 애드온 |
| 결제 수단 | 자체 시스템 | 해외 신용카드 불필요, 로컬 결제 | 다양함 |
| 신규 가입 크레딧 | 없음 | ✅ 무료 크레딧 제공 | 불규칙 |
| Rate Limit 처리 | 자체 구현 필요 | 자동 재시도·폴백 | 서비스별 상이 |
| 주문 지연 시간 | ~50ms | ~80ms (AI 오버헤드 포함) | ~100-300ms |
Binance 주문 데이터 구조 핵심 필드 매핑
Binance 주문 응답은market에 따라 필드명이 상이합니다. 아래 표는 현물·선물·마진 거래소의 공통 필드와 차이점을 정리한 것입니다.
| 필드명 | 현물 (Spot) | 선물 (Futures USDT-M) | 마진 (Margin) | 설명 |
|---|---|---|---|---|
orderId |
✅ | ✅ | ✅ | 交易所分配的32位整数订单ID |
orderListId |
✅ (OCO만) | N/A | ✅ | 오더 리스트 그룹 ID |
symbol |
BTCUSDT | BTCUSDT | BTCUSDT | 거래 쌍 심볼 |
status |
NEW, PARTIALLY_FILLED, FILLED, CANCELED, PENDING_CANCEL, REJECTED, EXPIRED | NEW, PARTIALLY_FILLED, FILLED, CANCELED, EXPIRED, NEW_INSURANCE, NEW_ADL | NEW, PARTIALLY_FILLED, FILLED, CANCELED, EXPIRED | 주문 상태 열거형 |
type |
LIMIT, MARKET, STOP_LOSS, STOP_LOSS_LIMIT, TAKE_PROFIT, TAKE_PROFIT_LIMIT, LIMIT_MAKER | LIMIT, MARKET, STOP, STOP_MARKET, TAKE_PROFIT_MARKET, TRAILING_STOP_MARKET | LIMIT, MARKET | 주문 유형 |
side |
BUY, SELL | BUY, SELL | BUY, SELL | 매수/매도 방향 |
price |
정확한 가격 | 정확한 가격 | 정확한 가격 | 지정가 주문 가격 |
stopPrice |
✅ | ✅ | ❌ | 조건 발동 가격 |
origQty |
✅ | ✅ | ✅ | 원래 주문 수량 |
executedQty |
✅ | ✅ | ✅ | 체결된 수량 |
avgPrice |
현물: 없음*, 선물: ✅ | ✅ | ✅ | 평균 체결 가격 |
commission |
✅ | ✅ | ✅ | 수수료 금액 |
commissionAsset |
BNB, USDT 등 | USDT | BNB, USDT 등 | 수수료 결제 자산 |
clientOrderId |
✅ | ✅ | ✅ | 사용자 지정 주문 ID |
updateTime |
✅ | ✅ | ✅ | 최종 업데이트 타임스탬프 |
isIsolated |
N/A | N/A | ✅ | 격리 마진 여부 |
*현물 MARKET 주문의 avgPrice는 거래소에서 제공하지 않으며, 별도 계산 필요
Python으로 Binance 주문 데이터 파싱하기
저는 Binance订单 데이터를 일관된 형태로 정규화하는 유틸리티 클래스를 직접 작성하여 사용합니다. 아래 코드는 현물·선물 응답을统一的 스키마로 변환하는 예시입니다.
# requirements: requests, python-binance (현물), binance-futures-connector (선물)
import requests
import time
from dataclasses import dataclass, asdict
from typing import Optional, Literal
from enum import Enum
class OrderStatus(Enum):
NEW = "NEW"
PARTIALLY_FILLED = "PARTIALLY_FILLED"
FILLED = "FILLED"
CANCELED = "CANCELED"
REJECTED = "REJECTED"
EXPIRED = "EXPIRED"
class OrderSide(Enum):
BUY = "BUY"
SELL = "SELL"
@dataclass
class NormalizedOrder:
"""跨市場統一的訂單數據結構"""
order_id: int
client_order_id: Optional[str]
symbol: str
status: str
side: str
order_type: str
price: Optional[float]
stop_price: Optional[float]
orig_quantity: float
executed_quantity: float
avg_fill_price: Optional[float]
commission: Optional[float]
commission_asset: Optional[str]
created_at: int
updated_at: int
market: Literal["SPOT", "FUTURES", "MARGIN"]
is_isolated: Optional[bool] = None
leverage: Optional[int] = None # 선물 전용
class BinanceOrderParser:
"""Binance订单数据标准化解析器"""
SPOT_BASE = "https://api.binance.com"
FUTURES_BASE = "https://fapi.binance.com"
MARGIN_BASE = "https://api.binance.com"
def __init__(self, api_key: str, api_secret: str):
self.api_key = api_key
self.api_secret = api_secret
def _parse_spot_order(self, raw: dict) -> NormalizedOrder:
"""현물 주문 응답 정규화"""
return NormalizedOrder(
order_id=int(raw["orderId"]),
client_order_id=raw.get("clientOrderId"),
symbol=raw["symbol"],
status=raw["status"],
side=raw["side"],
order_type=raw["type"],
price=float(raw["price"]) if raw.get("price") else None,
stop_price=float(raw["stopPrice"]) if raw.get("stopPrice") else None,
orig_quantity=float(raw["origQty"]),
executed_quantity=float(raw["executedQty"]),
avg_fill_price=self._calc_avg_price_spot(raw),
commission=float(raw["commission"]) if raw.get("commission") else None,
commission_asset=raw.get("commissionAsset"),
created_at=int(raw["time"]),
updated_at=int(raw["updateTime"]),
market="SPOT",
)
def _calc_avg_price_spot(self, raw: dict) -> Optional[float]:
"""현물 시장 avgPrice 미제공 시 체결 내역에서 계산"""
if raw.get("avgPrice"):
return float(raw["avgPrice"])
# MARKET 주문의 경우 별도 조회 필요
executed = float(raw["executedQty"])
if executed == 0:
return None
# fills 배열에서 계산 (주문 생성 시 포함된 경우)
fills = raw.get("fills", [])
if fills:
total_value = sum(float(f["price"]) * float(f["qty"]) for f in fills)
return total_value / executed
return None
def _parse_futures_order(self, raw: dict) -> NormalizedOrder:
"""선물 주문 응답 정규화"""
return NormalizedOrder(
order_id=int(raw["orderId"]),
client_order_id=raw.get("clientOrderId"),
symbol=raw["symbol"],
status=raw["status"],
side=raw["side"],
order_type=raw["type"],
price=float(raw["price"]) if raw.get("price") else None,
stop_price=float(raw["stopPrice"]) if raw.get("stopPrice") else None,
orig_quantity=float(raw["origQty"]),
executed_quantity=float(raw["executedQty"]),
avg_fill_price=float(raw["avgPrice"]) if raw.get("avgPrice") else None,
commission=float(raw["commission"]) if raw.get("commission") else None,
commission_asset=raw.get("commissionAsset", "USDT"),
created_at=int(raw["time"]),
updated_at=int(raw["updateTime"]),
market="FUTURES",
leverage=int(raw.get("leverage", 1)),
)
def normalize_order(self, raw: dict, market: str) -> NormalizedOrder:
"""市场별 주문 데이터 정규화 dispatcher"""
parsers = {
"SPOT": self._parse_spot_order,
"MARGIN": self._parse_spot_order, # 마진은 현물과 필드 유사
"FUTURES": self._parse_futures_order,
}
parser = parsers.get(market.upper())
if not parser:
raise ValueError(f"Unsupported market: {market}")
return parser(raw)
def get_order_with_retry(self, symbol: str, order_id: int,
market: str = "SPOT", retries: int = 3) -> NormalizedOrder:
"""주문 조회 + 재시도 로직"""
bases = {
"SPOT": self.SPOT_BASE,
"MARGIN": self.MARGIN_BASE,
"FUTURES": self.FUTURES_BASE,
}
base = bases[market.upper()]
endpoints = {
"SPOT": "/api/v3/order",
"MARGIN": "/api/v3/order",
"FUTURES": "/fapi/v2/order",
}
for attempt in range(retries):
try:
params = {"symbol": symbol, "orderId": order_id}
headers = {"X-MBX-APIKEY": self.api_key}
# 서명 생략 (실제 구현 시 HMAC-SHA256 필요)
response = requests.get(
f"{base}{endpoints[market.upper()]}",
params=params,
headers=headers,
timeout=5
)
response.raise_for_status()
raw = response.json()
return self.normalize_order(raw, market)
except requests.exceptions.RequestException as e:
if attempt == retries - 1:
raise
time.sleep(2 ** attempt) # 지수 백오프
raise RuntimeError("Failed after all retries")
사용 예시
if __name__ == "__main__":
parser = BinanceOrderParser(
api_key="YOUR_BINANCE_API_KEY",
api_secret="YOUR_BINANCE_API_SECRET"
)
# 예시: 현물 주문 조회
order = parser.get_order_with_retry(
symbol="BTCUSDT",
order_id=123456789,
market="SPOT"
)
print(f"주문ID: {order.order_id}")
print(f"상태: {order.status}")
print(f"체결평균가: {order.avg_fill_price}")
print(f"시장: {order.market}")
WebSocket 실시간 주문 데이터 스트림
Binance는 주문 업데이트를 실시간으로 수신할 수 있는 WebSocket 스트림을 제공합니다. 단일 주문 추적(@order)과 전체 계정 업데이트(@account) 두 가지 방식이 있습니다.
# pip install websockets
import asyncio
import json
import websockets
from dataclasses import dataclass
from typing import Callable, Optional
@dataclass
class OrderUpdate:
"""WebSocket 주문 업데이트 이벤트"""
event_type: str # e.g., "executionReport"
event_time: int # 이벤트 발생 시간 (밀리초)
symbol: str # 거래 쌍
client_order_id: str # 사용자 지정 주문 ID
side: str # BUY/SELL
order_type: str # 주문 유형
order_status: str # 주문 상태
price: str # 주문 가격
stop_price: str #止损价格
orig_quantity: str # 원래 수량
executed_quantity: str # 체결 수량
avg_price: str # 평균 체결 가격
order_id: int # Binance 주문 ID
is_marker_order: bool # 시장가 주문 여부
is_working: bool # 활성 주문 여부
transaction_time: int # 거래 시간
class BinanceWebSocketClient:
"""Binance WebSocket 주문 스트림 클라이언트"""
SPOT_WS = "wss://stream.binance.com:9443/ws"
FUTURES_WS = "wss://fstream.binance.com/ws"
def __init__(self, api_key: str):
self.api_key = api_key
self.subscriptions = []
async def subscribe_order_updates(self, symbol: str,
callback: Callable[[OrderUpdate], None],
market: str = "SPOT") -> None:
"""개별 주문 스트림 구독 (symbol별 실시간 업데이트)"""
stream_name = f"{symbol.lower()}@order"
if market == "FUTURES":
base_url = self.FUTURES_WS
else:
base_url = self.SPOT_WS
subscribe_msg = {
"method": "SUBSCRIBE",
"params": [stream_name],
"id": int(asyncio.get_event_loop().time() * 1000)
}
async with websockets.connect(base_url) as ws:
await ws.send(json.dumps(subscribe_msg))
# 구독 확인
confirm = await ws.recv()
print(f"구독 확인: {confirm}")
# 실시간 메시지 수신 루프
async for message in ws:
data = json.loads(message)
# 주문 업데이트 이벤트만 처리
if data.get("e") == "executionReport":
update = self._parse_execution_report(data, market)
await callback(update)
elif data.get("result") is None and "id" in data:
# 구독 성공 확인 (무시)
continue
def _parse_execution_report(self, data: dict, market: str) -> OrderUpdate:
"""executionReport 이벤트 파싱"""
# 선물 vs 현물 필드명 차이 처리
if market == "FUTURES":
return OrderUpdate(
event_type=data.get("e"),
event_time=data.get("E"),
symbol=data.get("s"),
client_order_id=data.get("c"),
side=data.get("S"),
order_type=data.get("o"),
order_status=data.get("X"), # Binance 상태 코드
price=data.get("p"),
stop_price=data.get("P"),
orig_quantity=data.get("q"),
executed_quantity=data.get("l"), # 체결 수량 (last trade)
avg_price=data.get("ap"),
order_id=int(data.get("i")),
is_marker_order=data.get("m"),
is_working=data.get("w"),
transaction_time=data.get("T"),
)
else:
# 현물/마진
return OrderUpdate(
event_type=data.get("e"),
event_time=data.get("E"),
symbol=data.get("s"),
client_order_id=data.get("c"),
side=data.get("S"),
order_type=data.get("o"),
order_status=data.get("X"),
price=data.get("p"),
stop_price=data.get("p"), # 단일 필드
orig_quantity=data.get("q"),
executed_quantity=data.get("l", "0"),
avg_price=data.get("L"), # 마지막 체결 가격
order_id=int(data.get("i")),
is_marker_order=data.get("m"),
is_working=data.get("w"),
transaction_time=data.get("T"),
)
async def subscribe_account_updates(self, callback: Callable) -> None:
"""계정 전체 업데이트 스트림 구독 (맞춤 설정 필요)"""
#.listenKey 필요 (별도 获取)
listen_key = await self._get_listen_key()
stream_url = f"{self.SPOT_WS}/{listen_key}"
async with websockets.connect(stream_url) as ws:
async for message in ws:
data = json.loads(message)
await callback(data)
async def _get_listen_key(self) -> str:
"""REST API로 listenKey 발급"""
headers = {"X-MBX-APIKEY": self.api_key}
async with websockets.connect(self.SPOT_WS.replace("wss", "https").replace("/ws", "/api/v3")):
# 실제로는 httpx나 aiohttp 사용
pass
raise NotImplementedError("Use httpx for listenKey request")
async def handle_order_update(update: OrderUpdate):
"""주문 업데이트 처리 콜백 예시"""
print(f"[{update.event_type}] {update.symbol} {update.side} "
f"상태:{update.order_status} 체결:{update.executed_quantity}")
# 주문 완료 시 추가 처리
if update.order_status == "FILLED":
print(f"✅ 주문 {update.order_id} 전체 체결 완료")
사용 예시
async def main():
client = BinanceWebSocketClient(api_key="YOUR_BINANCE_API_KEY")
try:
await client.subscribe_order_updates(
symbol="BTCUSDT",
callback=handle_order_update,
market="SPOT"
)
except websockets.exceptions.ConnectionClosed:
print("연결 종료, 재연결 시도...")
await asyncio.sleep(5)
await main()
if __name__ == "__main__":
asyncio.run(main())
HolySheep AI로 주문 데이터 AI 분석 파이프라인 구축
Binance订单数据的量虽大,但传统的规则-based警报系统难以捕捉复杂的市场模式。我在实际项目中,将订单数据通过 HolySheep AI API 连接到大语言模型,实现了以下自动化分析功能:
- 주문 체결 지연 패턴 자동 탐지
- 빈번한 부분 체결 원인 AI 분석
- 시장 돌파 시 주문 밀도 패턴 예측
- 거래 전략 수익성 자동 리포트 생성
# pip install openai httpx python-dotenv
import os
import json
from datetime import datetime
from typing import List, Dict, Optional
from openai import OpenAI
HolySheep AI 설정
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY", # 👈 HolySheep API 키
base_url="https://api.holysheep.ai/v1" # 👈 HolySheep 엔드포인트
)
class OrderAnalyzer:
"""HolySheep AI 기반 Binance 주문 데이터 분석기"""
def __init__(self):
self.client = client
self.model = "gpt-4.1" # HolySheep에서 사용 가능한 모델
def summarize_orders(self, orders: List[Dict]) -> str:
"""여러 주문 데이터의 요약 분석을 생성합니다"""
prompt = f"""당신은 전문 암호화폐 거래 분석가입니다.
아래 Binance 주문 데이터를 분석하여 핵심 인사이트를 제공해주세요:
주문 데이터
{json.dumps(orders, indent=2, ensure_ascii=False)}
분석 요청 사항
1. 전체 주문 성공률 및 평균 체결 시간
2. 손절/익절 비율 분석
3. 부분 체결 발생 패턴
4. 개선이 필요한 영역 및 구체적 권장사항
5. 다음 거래 세션용 실행 가능한 거래 전략
JSON 형식으로 결과를 반환해주세요:
{{
"summary": "요약 설명",
"success_rate": 0.0,
"avg_fill_time_ms": 0,
"partial_fill_count": 0,
"loss_cut_ratio": 0.0,
"take_profit_ratio": 0.0,
"issues": ["문제1", "문제2"],
"recommendations": ["권장1", "권장2"],
"next_strategy": "다음 전략 설명"
}}"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": "당신은 Binance 주문 데이터 분석 전문가입니다. 한국어로 정확하게 응답해주세요."
},
{"role": "user", "content": prompt}
],
temperature=0.3, # 분석이므로 낮은 온도
max_tokens=2000
)
return response.choices[0].message.content
def detect_anomalies(self, orders: List[Dict]) -> List[Dict]:
"""주문 데이터에서 이상치 탐지"""
prompt = f"""아래 Binance 주문 로그에서 비정상적인 패턴을 탐지해주세요:
{json.dumps(orders, indent=2, ensure_ascii=False)}
탐지해야 할 이상치 유형:
- 비정상적으로 오래 체결되는 주문 (>30초)
- 연속된 부분 체결 (3회 이상)
- 취소 후 재주문 패턴
- 비정상적 수수료 발생
이상치만 필터링하여 배열로 반환:
[
{{
"order_id": 12345,
"anomaly_type": "지연_체결",
"description": "설명",
"severity": "high/medium/low",
"recommended_action": "권장 조치"
}}
]"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "한국어로만 응답. JSON 배열만 반환."},
{"role": "user", "content": prompt}
],
temperature=0.1,
max_tokens=1500,
response_format={"type": "json_object"}
)
try:
result = json.loads(response.choices[0].message.content)
return result.get("anomalies", [])
except json.JSONDecodeError:
return []
def generate_trading_report(self, daily_orders: List[Dict],
market_symbol: str) -> str:
"""일일 거래 리포트 자동 생성"""
system_prompt = """당신은高频交易量化分析师입니다。
提供한국어로 상세하고 실행 가능한 거래 리포트를 작성해주세요.
Markdown 형식으로 반환해주세요."""
prompt = f"""## {market_symbol} 일일 거래 리포트
생성일: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
기본 통계
- 총 주문 수: {len(daily_orders)}
- 체결 성공: {sum(1 for o in daily_orders if o.get('status') == 'FILLED')}
- 부분 체결: {sum(1 for o in daily_orders if o.get('status') == 'PARTIALLY_FILLED')}
- 취소: {sum(1 for o in daily_orders if o.get('status') == 'CANCELED')}
상세 주문 내역
{json.dumps(daily_orders[:20], indent=2, ensure_ascii=False)}
위 데이터를 바탕으로 다음 항목 포함 리포트 작성:
1. 일일 성과 요약 (MQL 코멘트 포함)
2. 체결 품질 분석
3. 시장 상황 평가
4. 다음 거래일 전략 수립
5. 위험 관리 권장사항"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
],
temperature=0.4,
max_tokens=3000
)
return response.choices[0].message.content
HolySheep AI 가격 참고 (USD per Million Tokens)
HOLYSHEEP_PRICING = {
"gpt-4.1": {"input": 8.00, "output": 8.00},
"claude-sonnet-4-5": {"input": 15.00, "output": 15.00},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50},
"deepseek-v3.2": {"input": 0.42, "output": 0.42},
}
def estimate_analysis_cost(num_orders: int, model: str = "gpt-4.1") -> Dict:
"""분석 비용 추정 (HolySheep 기준)"""
# 주문 100건 기준 약 500토큰 입력 소모
input_tokens = (num_orders / 100) * 500
output_tokens = 300 # 고정 출력
pricing = HOLYSHEEP_PRICING.get(model, HOLYSHEEP_PRICING["gpt-4.1"])
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
total_cost = input_cost + output_cost
return {
"input_tokens": int(input_tokens),
"output_tokens": output_tokens,
"input_cost_usd": round(input_cost, 6),
"output_cost_usd": round(output_cost, 6),
"total_cost_usd": round(total_cost, 6),
}
사용 예시
if __name__ == "__main__":
# 샘플 주문 데이터
sample_orders = [
{
"orderId": 111111,
"symbol": "BTCUSDT",
"status": "FILLED",
"side": "BUY",
"type": "LIMIT",
"price": 42000.00,
"origQty": 0.1,
"executedQty": 0.1,
"avgPrice": 42005.50,
"commission": 0.042,
"commissionAsset": "USDT",
"time": 1704067200000,
},
{
"orderId": 222222,
"symbol": "ETHUSDT",
"status": "PARTIALLY_FILLED",
"side": "SELL",
"type": "LIMIT",
"price": 2200.00,
"origQty": 2.0,
"executedQty": 0.8,
"avgPrice": 2200.50,
"commission": 0.00176,
"commissionAsset": "USDT",
"time": 1704067300000,
},
]
analyzer = OrderAnalyzer()
# 주문 요약 분석
summary = analyzer.summarize_orders(sample_orders)
print("=== 주문 분석 결과 ===")
print(summary)
# 비용 추정
cost = estimate_analysis_cost(len(sample_orders))
print(f"\n예상 분석 비용: ${cost['total_cost_usd']}")
이런 팀에 적합 / 비적합
| ✅ HolySheep AI가 적합한 팀 | ❌ HolySheep AI가 부적합한 팀 |
|---|---|
|
암호화폐 거래 봇 개발자 다중 거래소 API를 통합 관리해야 하며, AI 기반 분석 기능이 필요한 경우 |
단순 주문 실행만 필요 Binance 공식 API만으로 충분한 고속 거래(HFT) 전략을 실행하는 경우 |
|
해외 신용카드 없는 개발자 국내 결제 수단으로 AI API 비용을 정산하고 싶은 경우 |
엄청난 트래픽 처리 초당 1000+ API 호출이 필요한 대규모 인프라도매の場合 |
|
AI/人嘲 분석 도입 싶은 팀 주문 패턴 분석, 리스크 리포트 자동화를 검토 중인 경우 |
최저 지연 시간 필수 레이턴시가 10ms 이내인 네이티브 API 연결이 필수적인 경우 |
|
다중 모델 비교 실험 GPT, Claude, Gemini, DeepSeek를 동일한 인터페이스에서 테스트하려는 경우 |
복잡한 선물 거래 USDT-M 선물 외币헤지, 분산 마진 등 특수 기능이 필요한 경우 |
가격과 ROI
| 모델 | 입력 비용 ($/MTok) | 출력 비용 ($/MTok) | 100K 토큰 비용 | <
|---|