증권 API를 사용하면 Python 코드에서 시장 데이터를 조회하거나 주문을 전송할 수 있다.
하지만 실제 자동매매 시스템을 만들 때는 단순히 requests.get() 한 번 호출하는 것보다 다음 구조를 이해하는 것이 중요하다.
인증
→ REST 조회
→ 연속 조회
→ 실시간 WebSocket
→ 주문
→ 체결 이벤트
→ 상태 동기화증권사마다 API 규격은 다르지만 전체 구조는 상당히 비슷하다.
REST와 WebSocket의 역할
REST API
필요할 때 요청하고 응답을 받는다.
주로 다음 작업에 적합하다.
- 계좌 정보
- 종목 정보
- 과거 시세
- 현재가 조회
- 주문 요청
- 주문 상태 조회
Client
→ HTTP Request
→ Server
→ HTTP ResponseWebSocket
연결을 유지한 상태에서 서버가 새로운 데이터를 밀어준다.
주로 다음에 적합하다.
- 실시간 체결
- 호가
- 주문 접수 이벤트
- 체결 이벤트
Client
↔ Persistent Connection
↔ ServerOAuth와 접근 토큰
현대 Open API에서는 APP KEY, SECRET KEY 등으로 접근 토큰을 발급받는 형태가 흔하다.
개념적인 요청:
import requests
def request_token(
base_url: str,
app_key: str,
app_secret: str,
) -> str:
response = requests.post(
f"{base_url}/oauth/token",
json={
"app_key": app_key,
"app_secret": app_secret,
},
timeout=10,
)
response.raise_for_status()
data = response.json()
return data[
"access_token"
]실제 엔드포인트와 파라미터 이름은 사용하는 증권사의 공식 문서를 확인해야 한다.
Secret은 환경 변수로 관리하기
import os
app_key = os.environ[
"BROKER_APP_KEY"
]
app_secret = os.environ[
"BROKER_APP_SECRET"
]다음 값은 Git 저장소에 넣지 않는다.
APP KEY
SECRET KEY
ACCESS TOKEN
계좌번호
인증서 비밀번호공통 API 클라이언트
from dataclasses import dataclass
import requests
@dataclass
class BrokerConfig:
base_url: str
token: str
class BrokerClient:
def __init__(
self,
config: BrokerConfig,
):
self.config = config
self.session = (
requests.Session()
)
def _headers(self) -> dict:
return {
"Authorization": (
"Bearer "
+ self.config.token
),
"Content-Type": (
"application/json"
),
}
def get(
self,
path: str,
params: dict | None = None,
) -> dict:
response = (
self.session.get(
self.config.base_url
+ path,
headers=self._headers(),
params=params,
timeout=10,
)
)
response.raise_for_status()
return response.json()HTTP 계층을 한 곳으로 모으면 인증 헤더, timeout, 오류 처리를 일관되게 적용할 수 있다.
현재가 조회 예제 구조
def get_quote(
client: BrokerClient,
symbol: str,
) -> dict:
return client.get(
"/quotes",
{
"symbol": symbol,
},
)실제 API 응답이 다음처럼 복잡하더라도:
{
"result": {
"code": "AAA",
"current_price": "52100",
"volume": "152301"
}
}프로그램 내부에서는 정규화된 객체로 바꾸는 것이 좋다.
from dataclasses import dataclass
@dataclass
class Quote:
symbol: str
price: int
volume: intdef parse_quote(
data: dict,
) -> Quote:
result = data["result"]
return Quote(
symbol=result["code"],
price=int(
result[
"current_price"
]
),
volume=int(
result["volume"]
),
)API 응답을 그대로 전략에 넘기지 않는 이유
증권사가 API 필드명을 변경하면 전략 코드 전체가 영향을 받을 수 있다.
외부 API 응답
→ Adapter
→ 내부 모델
→ 전략이 구조를 사용하면 외부 API 변경을 adapter 계층에서 흡수할 수 있다.
연속 조회와 Pagination
과거 일봉을 한 번의 요청으로 전부 받을 수 없는 경우가 많다.
응답 예:
{
"items": [],
"next_key": "ABC123"
}반복 조회 구조:
def fetch_all_pages(
client,
path,
):
rows = []
next_key = None
while True:
params = {}
if next_key:
params[
"next_key"
] = next_key
data = client.get(
path,
params,
)
rows.extend(
data["items"]
)
next_key = data.get(
"next_key"
)
if not next_key:
break
return rows실제 API에서는 continuation header나 별도 필드를 사용할 수 있다.
호출 제한 Rate Limit
증권 API에는 호출 제한이 존재할 수 있다.
잘못된 방식:
for symbol in symbols:
client.get(
"/quote",
{
"symbol": symbol,
},
)수천 종목에 대해 제한 없이 호출하면 차단될 수 있다.
필요한 대응:
- API 문서의 초당/분당 호출량 확인
- batch API가 있다면 활용
- 요청 큐 사용
- 캐싱
- backoff
- 연속조회 규칙 준수
조회 API 재시도
조회 요청은 네트워크 오류에 대해 제한적으로 재시도할 수 있다.
import time
import requests
def fetch_with_retry(
func,
max_attempts=3,
):
for attempt in range(
1,
max_attempts + 1,
):
try:
return func()
except (
requests.Timeout,
requests.ConnectionError,
):
if (
attempt
== max_attempts
):
raise
time.sleep(
attempt
)하지만 주문 요청에는 이 패턴을 그대로 적용하면 위험하다.
주문 요청 재시도가 위험한 이유
주문 요청
→ 서버는 접수 완료
→ 응답 전달 중 네트워크 timeout클라이언트는 실패로 보지만 서버에서는 이미 주문이 들어갔을 수 있다.
이때 같은 주문을 재전송하면:
첫 번째 주문
+
두 번째 주문
=
중복 주문따라서 timeout이 발생하면 먼저 주문 내역 또는 클라이언트 주문 ID를 통해 접수 여부를 확인해야 한다.
Client Order ID
지원되는 API라면 주문마다 고유한 식별자를 만드는 방식이 유용하다.
from uuid import uuid4
client_order_id = (
str(uuid4())
)주문 기록:
order = {
"client_order_id": (
client_order_id
),
"symbol": "AAA",
"side": "BUY",
"quantity": 3,
}중복 주문 방지와 추적에 도움이 된다.
WebSocket 기본 구조
Python에서는 다양한 WebSocket 라이브러리를 사용할 수 있다.
개념적인 비동기 구조:
import asyncio
import json
import websockets
async def listen_quotes(
url: str,
token: str,
):
async with (
websockets.connect(
url
)
as websocket
):
subscribe = {
"type": "subscribe",
"token": token,
"symbols": [
"AAA",
"BBB",
],
}
await websocket.send(
json.dumps(
subscribe
)
)
async for message in (
websocket
):
data = json.loads(
message
)
print(data)실제 메시지 규격과 인증 방식은 증권사별로 다르다.
실시간 데이터는 큐로 분리하기
WebSocket 콜백 안에서 전략 계산과 DB 저장을 모두 처리하면 수신이 밀릴 수 있다.
좋은 구조:
WebSocket Receiver
↓
Queue
↙ ↘
Storage StrategyPython의 asyncio.Queue를 사용할 수 있다.
import asyncio
quote_queue = (
asyncio.Queue()
)수신:
async def receiver():
while True:
quote = await (
receive_quote()
)
await quote_queue.put(
quote
)처리:
async def worker():
while True:
quote = await (
quote_queue.get()
)
try:
await process_quote(
quote
)
finally:
quote_queue.task_done()연결이 끊어졌을 때
실시간 연결은 언제든 끊길 수 있다.
대응해야 하는 상황:
- 인터넷 단절
- 서버 점검
- 토큰 만료
- heartbeat 실패
- 구독 제한
- 프로그램 예외
재접속 시에는 단순히 연결만 다시 하는 것이 아니라 구독 종목과 마지막 처리 상태를 복구해야 한다.
Heartbeat
WebSocket 서비스는 연결 상태 확인을 위해 ping/pong 또는 heartbeat 메시지를 사용할 수 있다.
Client
→ ping
Server
→ pong일정 시간 응답이 없으면 연결을 비정상 상태로 판단할 수 있다.
주문 이벤트 모델
from dataclasses import dataclass
from datetime import datetime
@dataclass
class OrderEvent:
order_id: str
symbol: str
status: str
filled_quantity: int
filled_price: float | None
timestamp: datetime체결 이벤트:
event = OrderEvent(
order_id="O-1001",
symbol="AAA",
status="FILLED",
filled_quantity=3,
filled_price=52100,
timestamp=datetime.now(),
)부분 체결
10주 주문했다고 한 번에 10주가 모두 체결되는 것은 아니다.
주문 10주
→ 3주 체결
→ 4주 체결
→ 3주 체결따라서 포지션은 주문 수량이 아니라 실제 체결 수량을 기준으로 갱신해야 한다.
포지션 상태 동기화
프로그램 내부 값만 믿으면 안 된다.
예:
내 프로그램: 10주 보유
증권사 계좌: 8주 보유이런 차이가 생길 수 있다.
주기적으로 실제 계좌 상태와 내부 상태를 비교하는 reconciliation 과정이 필요하다.
def reconcile(
local_positions,
broker_positions,
):
...Paper Trading과 Production 분리
환경 변수를 이용할 수 있다.
mode = os.getenv(
"TRADING_MODE",
"paper",
)if mode == "live":
service = LiveOrderService()
else:
service = PaperOrderService()기본값을 paper로 두는 것이 안전하다.
주문 전 체크
def validate_order(
quantity: int,
price: float,
) -> None:
if quantity <= 0:
raise ValueError(
"수량은 1 이상이어야 합니다."
)
if price <= 0:
raise ValueError(
"가격은 0보다 커야 합니다."
)추가로 확인할 것:
- 시장 운영 시간
- 매수 가능 금액
- 최대 주문 금액
- 종목별 보유 한도
- 중복 주문
- 매매 정지 여부
- 가격 단위
- 주문 가능 시장
로그를 반드시 남긴다
import logging
logger = logging.getLogger(
__name__
)logger.info(
"order submitted",
extra={
"symbol": "AAA",
"quantity": 3,
},
)최소한 다음 이벤트는 기록하는 것이 좋다.
- 로그인
- API 오류
- 데이터 수신
- 신호 생성
- 주문 요청
- 주문 응답
- 체결
- 취소
- 시스템 종료
핵심 정리
- REST API는 조회·주문처럼 요청과 응답이 명확한 작업에, WebSocket은 실시간 데이터와 이벤트 수신에 적합하다.
- 외부 API 응답은 adapter를 통해 내부 데이터 모델로 변환하면 전략 코드가 증권사 규격에 덜 의존한다.
- 대량 데이터 조회에서는 pagination과 호출 제한을 반드시 고려해야 한다.
- 조회 API의 재시도 패턴을 주문 API에 그대로 적용하면 중복 주문이 발생할 수 있다.
- 자동매매에서는 주문 요청 상태와 실제 체결 상태를 구분해야 한다.
- 실시간 WebSocket 수신과 데이터 처리 로직을 큐로 분리하면 안정성이 높아진다.
- 프로그램 내부 포지션과 실제 계좌 상태를 주기적으로 동기화해야 한다.
- 실전 환경보다 모의투자 모드를 기본값으로 두고 충분히 검증한 뒤 전환하는 것이 안전하다.
