키움증권 REST API - 실시간시세

국내주식

• WEBSOCKET

실시간시세

기본 정보

※ 00 (주문체결)

실시간 항목 00(주문체결)은 종목코드(item) 등록과 상관 없이 ACCESS TOKEN을 발급한 계좌에 주문 접수, 체결, 정정, 취소 등 매매가 발생할 경우 데이터가 수신됩니다.

Method운영 도메인모의투자 도메인URLFormatContent-Type

POST

wss://api.kiwoom.com:10000

wss://mockapi.kiwoom.com:10000(KRX만 지원가능)

/api/dostk/websocket

JSON

application/json;charset=UTF-8

다른 TR 확인하기

TR명

주문체결

00

• 주문체결00
• 잔고04
• 주식기세0A
• 주식체결0B
• 주식우선호가0C
• 주식호가잔량0D
• 주식시간외호가0E
• 주식당일거래원0F
• ETF NAV0G
• 주식예상체결0H
• 업종지수0J
• 업종등락0U
• 주식종목정보0g
• ELW 이론가0m
• 장시작시간0s
• ELW 지표0u
• 종목프로그램매매0w
• VI발동/해제1h

요청

Body

 

Element	한글명	type	Required	Length	Description
trnm	서비스명	String	Y	10	REG : 등록 , REMOVE : 해지
grp_no	그룹번호	String	Y	4
refresh	기존등록유지여부	String	Y	1	등록(REG)시 0:기존유지안함 1:기존유지(Default) 0일경우 기존등록한 item/type은 해지, 1일경우 기존등록한 item/type 유지 해지(REMOVE)시 값 불필요
data	실시간 등록 리스트	LIST
- item	실시간 등록 요소	String[ ]	N	100
- type	실시간 항목	String[ ]	Y	2	TR 명(0A,0B....)

응답

Body

 

Element	한글명	type	Required	Length	Description
return_code	결과코드	int	N		통신결과에대한 코드 (등록,해지요청시에만 값 전송 0:정상,1:오류 , 데이터 실시간 수신시 미전송)
return_msg	결과메시지	String	N		통신결과에대한메시지
trnm	서비스명	String	N		등록,해지요청시 요청값 반환 , 실시간수신시 REAL 반환
data	실시간 등록리스트	LIST	N
- type	실시간항목	String	N		TR 명(0A,0B....)
- name	실시간 항목명	String	N
- item	실시간 등록 요소	String	N		종목코드
- values	실시간 값 리스트	LIST	N
- - 9201	계좌번호	String	N
- - 9203	주문번호	String	N
- - 9205	관리자사번	String	N
- - 9001	종목코드,업종코드	String	N
- - 912	주문업무분류	String	N
- - 913	주문상태	String	N
- - 302	종목명	String	N
- - 900	주문수량	String	N
- - 901	주문가격	String	N
- - 902	미체결수량	String	N
- - 903	체결누계금액	String	N
- - 904	원주문번호	String	N
- - 905	주문구분	String	N
- - 906	매매구분	String	N
- - 907	매도수구분	String	N
- - 908	주문/체결시간	String	N
- - 909	체결번호	String	N
- - 910	체결가	String	N
- - 911	체결량	String	N
- - 10	현재가	String	N
- - 27	(최우선)매도호가	String	N
- - 28	(최우선)매수호가	String	N
- - 914	단위체결가	String	N
- - 915	단위체결량	String	N
- - 938	당일매매수수료	String	N
- - 939	당일매매세금	String	N
- - 919	거부사유	String	N
- - 920	화면번호	String	N
- - 921	터미널번호	String	N
- - 922	신용구분	String	N		실시간 체결용
- - 923	대출일	String	N		실시간 체결용
- - 10010	시간외단일가_현재가	String	N
- - 2134	거래소구분	String	N		0:통합,1:KRX,2:NXT
- - 2135	거래소구분명	String	N		통합,KRX,NXT
- - 2136	SOR여부	String	N		Y,N

PYTHON 예제. 

import asyncio 
import websockets
import json

# socket 정보
# SOCKET_URL = 'wss://mockapi.kiwoom.com:10000/api/dostk/websocket'  # 모의투자 접속 URL
SOCKET_URL = 'wss://api.kiwoom.com:10000/api/dostk/websocket'  # 접속 URL
ACCESS_TOKEN = '사용자 AccessToken'  # 고객 Access Token

class WebSocketClient:
def __init__(self, uri):
self.uri = uri
self.websocket = None
self.connected = False
self.keep_running = True

# WebSocket 서버에 연결합니다.
async def connect(self):
try:
self.websocket = await websockets.connect(self.uri)
self.connected = True
print("서버와 연결을 시도 중입니다.")

# 로그인 패킷
param = {
'trnm': 'LOGIN',
'token': ACCESS_TOKEN
}

print('실시간 시세 서버로 로그인 패킷을 전송합니다.')
# 웹소켓 연결 시 로그인 정보 전달
await self.send_message(message=param)

except Exception as e:
print(f'Connection error: {e}')
self.connected = False

# 서버에 메시지를 보냅니다. 연결이 없다면 자동으로 연결합니다.
async def send_message(self, message):
if not self.connected:
await self.connect()  # 연결이 끊어졌다면 재연결
if self.connected:
# message가 문자열이 아니면 JSON으로 직렬화
if not isinstance(message, str):
message = json.dumps(message)

await self.websocket.send(message)
print(f'Message sent: {message}')

# 서버에서 오는 메시지를 수신하여 출력합니다.
async def receive_messages(self):
while self.keep_running:
try:
# 서버로부터 수신한 메시지를 JSON 형식으로 파싱
response = json.loads(await self.websocket.recv())

# 메시지 유형이 LOGIN일 경우 로그인 시도 결과 체크
if response.get('trnm') == 'LOGIN':
if response.get('return_code') != 0:
print('로그인 실패하였습니다. : ', response.get('return_msg'))
await self.disconnect()
else:
print('로그인 성공하였습니다.')

# 메시지 유형이 PING일 경우 수신값 그대로 송신
elif response.get('trnm') == 'PING':
await self.send_message(response)

if response.get('trnm') != 'PING':
print(f'실시간 시세 서버 응답 수신: {response}')

except websockets.ConnectionClosed:
print('Connection closed by the server')
self.connected = False
await self.websocket.close()

# WebSocket 실행
async def run(self):
await self.connect()
await self.receive_messages()

# WebSocket 연결 종료
async def disconnect(self):
self.keep_running = False
if self.connected and self.websocket:
await self.websocket.close()
self.connected = False
print('Disconnected from WebSocket server')

async def main():
# WebSocketClient 전역 변수 선언
websocket_client = WebSocketClient(SOCKET_URL)

# WebSocket 클라이언트를 백그라운드에서 실행합니다.
receive_task = asyncio.create_task(websocket_client.run())

# 실시간 항목 등록
await asyncio.sleep(1)
await websocket_client.send_message({ 
'trnm': 'REG', # 서비스명
'grp_no': '1', # 그룹번호
'refresh': '1', # 기존등록유지여부
'data': [{ # 실시간 등록 리스트
'item': [''], # 실시간 등록 요소
'type': ['00'], # 실시간 항목
}]
})

# 수신 작업이 종료될 때까지 대기
await receive_task

# asyncio로 프로그램을 실행합니다.
if __name__ == '__main__':
asyncio.run(main())

---

요청 잔고조회 tr 04

Body

Element	한글명	type	Required	Length	Description
trnm	서비스명	String	Y	10	REG : 등록 , REMOVE : 해지
grp_no	그룹번호	String	Y	4
refresh	기존등록유지여부	String	Y	1	등록(REG)시 0:기존유지안함 1:기존유지(Default) 0일경우 기존등록한 item/type은 해지, 1일경우 기존등록한 item/type 유지 해지(REMOVE)시 값 불필요
data	실시간 등록 리스트	LIST
- item	실시간 등록 요소	String[ ]	N	104
- type	실시간 항목	String[ ]	Y	2	TR 명(0A,0B....)

응답

Body

Element	한글명	type	Required	Length	Description
return_code	결과코드	int	N		통신결과에대한 코드 (등록,해지요청시에만 값 전송 0:정상,1:오류 , 데이터 실시간 수신시 미전송)
return_msg	결과메시지	String	N		통신결과에대한메시지
trnm	서비스명	String	N		등록,해지요청시 요청값 반환 , 실시간수신시 REAL 반환
data	실시간 등록리스트	LIST	N
- type	실시간항목	String	N		TR 명(0A,0B....)
- name	실시간 항목명	String	N
- item	실시간 등록 요소	String	N		종목코드
- values	실시간 값 리스트	LIST	N
- - 9201	계좌번호	String	N
- - 9001	종목코드,업종코드	String	N
- - 917	신용구분	String	N
- - 916	대출일	String	N
- - 302	종목명	String	N
- - 10	현재가	String	N
- - 930	보유수량	String	N
- - 931	매입단가	String	N
- - 932	총매입가(당일누적)	String	N
- - 933	주문가능수량	String	N
- - 945	당일순매수량	String	N
- - 946	매도/매수구분	String	N		계약,주
- - 950	당일총매도손익	String	N
- - 951	Extra Item	String	N
- - 27	(최우선)매도호가	String	N
- - 28	(최우선)매수호가	String	N
- - 307	기준가	String	N
- - 8019	손익률(실현손익)	String	N
- - 957	신용금액	String	N
- - 958	신용이자	String	N
- - 918	만기일	String	N
- - 990	당일실현손익(유가)	String	N
- - 991	당일실현손익율(유가)	String	N
- - 992	당일실현손익(신용)	String	N
- - 993	당일실현손익율(신용)	String	N
- - 959	담보대출수량	String	N
- - 924	Extra Item	String	N

잔고조회 TR 04

import asyncio 
import websockets
import json

# socket 정보
# SOCKET_URL = 'wss://mockapi.kiwoom.com:10000/api/dostk/websocket'  # 모의투자 접속 URL
SOCKET_URL = 'wss://api.kiwoom.com:10000/api/dostk/websocket'  # 접속 URL
ACCESS_TOKEN = '사용자 AccessToken'  # 고객 Access Token

class WebSocketClient:
def __init__(self, uri):
self.uri = uri
self.websocket = None
self.connected = False
self.keep_running = True

# WebSocket 서버에 연결합니다.
async def connect(self):
try:
self.websocket = await websockets.connect(self.uri)
self.connected = True
print("서버와 연결을 시도 중입니다.")

# 로그인 패킷
param = {
'trnm': 'LOGIN',
'token': ACCESS_TOKEN
}

print('실시간 시세 서버로 로그인 패킷을 전송합니다.')
# 웹소켓 연결 시 로그인 정보 전달
await self.send_message(message=param)

except Exception as e:
print(f'Connection error: {e}')
self.connected = False

# 서버에 메시지를 보냅니다. 연결이 없다면 자동으로 연결합니다.
async def send_message(self, message):
if not self.connected:
await self.connect()  # 연결이 끊어졌다면 재연결
if self.connected:
# message가 문자열이 아니면 JSON으로 직렬화
if not isinstance(message, str):
message = json.dumps(message)

await self.websocket.send(message)
print(f'Message sent: {message}')

# 서버에서 오는 메시지를 수신하여 출력합니다.
async def receive_messages(self):
while self.keep_running:
try:
# 서버로부터 수신한 메시지를 JSON 형식으로 파싱
response = json.loads(await self.websocket.recv())

# 메시지 유형이 LOGIN일 경우 로그인 시도 결과 체크
if response.get('trnm') == 'LOGIN':
if response.get('return_code') != 0:
print('로그인 실패하였습니다. : ', response.get('return_msg'))
await self.disconnect()
else:
print('로그인 성공하였습니다.')

# 메시지 유형이 PING일 경우 수신값 그대로 송신
elif response.get('trnm') == 'PING':
await self.send_message(response)

if response.get('trnm') != 'PING':
print(f'실시간 시세 서버 응답 수신: {response}')

except websockets.ConnectionClosed:
print('Connection closed by the server')
self.connected = False
await self.websocket.close()

# WebSocket 실행
async def run(self):
await self.connect()
await self.receive_messages()

# WebSocket 연결 종료
async def disconnect(self):
self.keep_running = False
if self.connected and self.websocket:
await self.websocket.close()
self.connected = False
print('Disconnected from WebSocket server')

async def main():
# WebSocketClient 전역 변수 선언
websocket_client = WebSocketClient(SOCKET_URL)

# WebSocket 클라이언트를 백그라운드에서 실행합니다.
receive_task = asyncio.create_task(websocket_client.run())

# 실시간 항목 등록
await asyncio.sleep(1)
await websocket_client.send_message({ 
'trnm': 'REG', # 서비스명
'grp_no': '1', # 그룹번호
'refresh': '1', # 기존등록유지여부
'data': [{ # 실시간 등록 리스트
'item': [''], # 실시간 등록 요소
'type': ['04'], # 실시간 항목
}]
})

# 수신 작업이 종료될 때까지 대기
await receive_task

# asyncio로 프로그램을 실행합니다.
if __name__ == '__main__':
asyncio.run(main())

---

요청 주식시세 TR 0A

Body

Element	한글명	type	Required	Length	Description
trnm	서비스명	String	Y	10	REG : 등록 , REMOVE : 해지
grp_no	그룹번호	String	Y	4
refresh	기존등록유지여부	String	Y	1	등록(REG)시 0:기존유지안함 1:기존유지(Default) 0일경우 기존등록한 item/type은 해지, 1일경우 기존등록한 item/type 유지 해지(REMOVE)시 값 불필요
data	실시간 등록 리스트	LIST
- item	실시간 등록 요소	String[ ]	N	100	거래소별 종목코드, 업종코드 (KRX:039490,NXT:039490_NX,SOR:039490_AL)
- type	실시간 항목	String[ ]	Y	2	TR 명(0A,0B....)

응답

Body

Element	한글명	type	Required	Length	Description
return_code	결과코드	int	N		통신결과에대한 코드 (등록,해지요청시에만 값 전송 0:정상,1:오류 , 데이터 실시간 수신시 미전송)
return_msg	결과메시지	String	N		통신결과에대한메시지(등록,해지시에만 값 전송,데이터 실시간 수신시 미전송)
trnm	서비스명	String	N		등록,해지요청시 요청값 반환 , 실시간수신시 REAL 반환
data	실시간 등록리스트	LIST	N
- type	실시간항목	String	N		TR 명(0A,0B....)
- name	실시간 항목명	String	N
- item	실시간 등록 요소	String	N		종목코드
- values	실시간 값 리스트	LIST	N
- - 10	현재가	String	N
- - 11	전일대비	String	N
- - 12	등락율	String	N
- - 27	(최우선)매도호가	String	N
- - 28	(최우선)매수호가	String	N
- - 13	누적거래량	String	N
- - 14	누적거래대금	String	N
- - 16	시가	String	N
- - 17	고가	String	N
- - 18	저가	String	N
- - 25	전일대비기호	String	N
- - 26	전일거래량대비(계약,주)	String	N
- - 29	거래대금증감	String	N
- - 30	전일거래량대비(비율)	String	N
- - 31	거래회전율	String	N
- - 32	거래비용	String	N
- - 311	시가총액(억)	String	N
- - 567	상한가발생시간	String	N
- - 568	하한가발생시간	String	N

 

주식시세 TR 0A

import asyncio 
import websockets
import json

# socket 정보
# SOCKET_URL = 'wss://mockapi.kiwoom.com:10000/api/dostk/websocket'  # 모의투자 접속 URL
SOCKET_URL = 'wss://api.kiwoom.com:10000/api/dostk/websocket'  # 접속 URL
ACCESS_TOKEN = '사용자 AccessToken'  # 고객 Access Token

class WebSocketClient:
def __init__(self, uri):
self.uri = uri
self.websocket = None
self.connected = False
self.keep_running = True

# WebSocket 서버에 연결합니다.
async def connect(self):
try:
self.websocket = await websockets.connect(self.uri)
self.connected = True
print("서버와 연결을 시도 중입니다.")

# 로그인 패킷
param = {
'trnm': 'LOGIN',
'token': ACCESS_TOKEN
}

print('실시간 시세 서버로 로그인 패킷을 전송합니다.')
# 웹소켓 연결 시 로그인 정보 전달
await self.send_message(message=param)

except Exception as e:
print(f'Connection error: {e}')
self.connected = False

# 서버에 메시지를 보냅니다. 연결이 없다면 자동으로 연결합니다.
async def send_message(self, message):
if not self.connected:
await self.connect()  # 연결이 끊어졌다면 재연결
if self.connected:
# message가 문자열이 아니면 JSON으로 직렬화
if not isinstance(message, str):
message = json.dumps(message)

await self.websocket.send(message)
print(f'Message sent: {message}')

# 서버에서 오는 메시지를 수신하여 출력합니다.
async def receive_messages(self):
while self.keep_running:
try:
# 서버로부터 수신한 메시지를 JSON 형식으로 파싱
response = json.loads(await self.websocket.recv())

# 메시지 유형이 LOGIN일 경우 로그인 시도 결과 체크
if response.get('trnm') == 'LOGIN':
if response.get('return_code') != 0:
print('로그인 실패하였습니다. : ', response.get('return_msg'))
await self.disconnect()
else:
print('로그인 성공하였습니다.')

# 메시지 유형이 PING일 경우 수신값 그대로 송신
elif response.get('trnm') == 'PING':
await self.send_message(response)

if response.get('trnm') != 'PING':
print(f'실시간 시세 서버 응답 수신: {response}')

except websockets.ConnectionClosed:
print('Connection closed by the server')
self.connected = False
await self.websocket.close()

# WebSocket 실행
async def run(self):
await self.connect()
await self.receive_messages()

# WebSocket 연결 종료
async def disconnect(self):
self.keep_running = False
if self.connected and self.websocket:
await self.websocket.close()
self.connected = False
print('Disconnected from WebSocket server')

async def main():
# WebSocketClient 전역 변수 선언
websocket_client = WebSocketClient(SOCKET_URL)

# WebSocket 클라이언트를 백그라운드에서 실행합니다.
receive_task = asyncio.create_task(websocket_client.run())

# 실시간 항목 등록
await asyncio.sleep(1)
await websocket_client.send_message({ 
'trnm': 'REG', # 서비스명
'grp_no': '1', # 그룹번호
'refresh': '1', # 기존등록유지여부
'data': [{ # 실시간 등록 리스트
'item': ['005930'], # 실시간 등록 요소
'type': ['0A'], # 실시간 항목
}]
})

# 수신 작업이 종료될 때까지 대기
await receive_task

# asyncio로 프로그램을 실행합니다.
if __name__ == '__main__':
asyncio.run(main())

---

요청 주식체결 TR 0B

Body

 

Element	한글명	type	Required	length	Description
trnm	서비스명	String	Y	10	REG : 등록 , REMOVE : 해지
grp_no	그룹번호	String	Y	4
refresh	기존등록유지여부	String	Y	1	등록(REG)시 0:기존유지안함 1:기존유지(Default) 0일경우 기존등록한 item/type은 해지, 1일경우 기존등록한 item/type 유지 해지(REMOVE)시 값 불필요
data	실시간 등록 리스트	LIST
- item	실시간 등록 요소	String[ ]	N	100	거래소별 종목코드, 업종코드 (KRX:039490,NXT:039490_NX,SOR:039490_AL)
- type	실시간 항목	String[ ]	Y	2	TR 명(0A,0B....)

응답

Body

Element	한글명	type	Required	Length	Description
return_code	결과코드	int	N		통신결과에대한 코드 (등록,해지요청시에만 값 전송 0:정상,1:오류 , 데이터 실시간 수신시 미전송)
return_msg	결과메시지	String	N		통신결과에대한메시지
trnm	서비스명	String	N		등록,해지요청시 요청값 반환 , 실시간수신시 REAL 반환
data	실시간 등록리스트	LIST	N
- type	실시간항목	String	N		TR 명(0B,0B....)
- name	실시간 항목명	String	N
- item	실시간 등록 요소	String	N		종목코드
- values	실시간 값 리스트	LIST	N
- - 20	체결시간	String	N
- - 10	현재가	String	N
- - 11	전일대비	String	N
- - 12	등락율	String	N
- - 27	(최우선)매도호가	String	N
- - 28	(최우선)매수호가	String	N
- - 15	거래량	String	N		+는 매수체결,-는 매도체결
- - 13	누적거래량	String	N
- - 14	누적거래대금	String	N
- - 16	시가	String	N
- - 17	고가	String	N
- - 18	저가	String	N
- - 25	전일대비기호	String	N
- - 26	전일거래량대비(계약,주)	String	N
- - 29	거래대금증감	String	N
- - 30	전일거래량대비(비율)	String	N
- - 31	거래회전율	String	N
- - 32	거래비용	String	N
- - 228	체결강도	String	N
- - 311	시가총액(억)	String	N
- - 290	장구분	String	N
- - 691	K.O 접근도	String	N
- - 567	상한가발생시간	String	N
- - 568	하한가발생시간	String	N
- - 851	전일 동시간 거래량 비율	String	N
- - 1890	시가시간	String	N
- - 1891	고가시간	String	N
- - 1892	저가시간	String	N
- - 1030	매도체결량	String	N
- - 1031	매수체결량	String	N
- - 1032	매수비율	String	N
- - 1071	매도체결건수	String	N
- - 1072	매수체결건수	String	N
- - 1313	순간거래대금	String	N
- - 1315	매도체결량_단건	String	N
- - 1316	매수체결량_단건	String	N
- - 1314	순매수체결량	String	N
- - 1497	CFD증거금	String	N
- - 1498	유지증거금	String	N
- - 620	당일거래평균가	String	N
- - 732	CFD거래비용	String	N
- - 852	대주거래비용	String	N
- - 9081	거래소구분	String	N

import asyncio 
import websockets
import json

# socket 정보
# SOCKET_URL = 'wss://mockapi.kiwoom.com:10000/api/dostk/websocket'  # 모의투자 접속 URL
SOCKET_URL = 'wss://api.kiwoom.com:10000/api/dostk/websocket'  # 접속 URL
ACCESS_TOKEN = '사용자 AccessToken'  # 고객 Access Token

class WebSocketClient:
def __init__(self, uri):
self.uri = uri
self.websocket = None
self.connected = False
self.keep_running = True

# WebSocket 서버에 연결합니다.
async def connect(self):
try:
self.websocket = await websockets.connect(self.uri)
self.connected = True
print("서버와 연결을 시도 중입니다.")

# 로그인 패킷
param = {
'trnm': 'LOGIN',
'token': ACCESS_TOKEN
}

print('실시간 시세 서버로 로그인 패킷을 전송합니다.')
# 웹소켓 연결 시 로그인 정보 전달
await self.send_message(message=param)

except Exception as e:
print(f'Connection error: {e}')
self.connected = False

# 서버에 메시지를 보냅니다. 연결이 없다면 자동으로 연결합니다.
async def send_message(self, message):
if not self.connected:
await self.connect()  # 연결이 끊어졌다면 재연결
if self.connected:
# message가 문자열이 아니면 JSON으로 직렬화
if not isinstance(message, str):
message = json.dumps(message)

await self.websocket.send(message)
print(f'Message sent: {message}')

# 서버에서 오는 메시지를 수신하여 출력합니다.
async def receive_messages(self):
while self.keep_running:
try:
# 서버로부터 수신한 메시지를 JSON 형식으로 파싱
response = json.loads(await self.websocket.recv())

# 메시지 유형이 LOGIN일 경우 로그인 시도 결과 체크
if response.get('trnm') == 'LOGIN':
if response.get('return_code') != 0:
print('로그인 실패하였습니다. : ', response.get('return_msg'))
await self.disconnect()
else:
print('로그인 성공하였습니다.')

# 메시지 유형이 PING일 경우 수신값 그대로 송신
elif response.get('trnm') == 'PING':
await self.send_message(response)

if response.get('trnm') != 'PING':
print(f'실시간 시세 서버 응답 수신: {response}')

except websockets.ConnectionClosed:
print('Connection closed by the server')
self.connected = False
await self.websocket.close()

# WebSocket 실행
async def run(self):
await self.connect()
await self.receive_messages()

# WebSocket 연결 종료
async def disconnect(self):
self.keep_running = False
if self.connected and self.websocket:
await self.websocket.close()
self.connected = False
print('Disconnected from WebSocket server')

async def main():
# WebSocketClient 전역 변수 선언
websocket_client = WebSocketClient(SOCKET_URL)

# WebSocket 클라이언트를 백그라운드에서 실행합니다.
receive_task = asyncio.create_task(websocket_client.run())

# 실시간 항목 등록
await asyncio.sleep(1)
await websocket_client.send_message({ 
'trnm': 'REG', # 서비스명
'grp_no': '1', # 그룹번호
'refresh': '1', # 기존등록유지여부
'data': [{ # 실시간 등록 리스트
'item': ['005930'], # 실시간 등록 요소
'type': ['0B'], # 실시간 항목
}]
})

# 수신 작업이 종료될 때까지 대기
await receive_task

# asyncio로 프로그램을 실행합니다.
if __name__ == '__main__':
asyncio.run(main())

---

요청 우선호가 TR 0C

Body

Element한글명typeRequiredLengthDescription

Element

한글명

type

Required

Length

Description

trnm	서비스명	String	Y	10	REG : 등록 , REMOVE : 해지
grp_no	그룹번호	String	Y	4
refresh	기존등록유지여부	String	Y	1	등록(REG)시 0:기존유지안함 1:기존유지(Default) 0일경우 기존등록한 item/type은 해지, 1일경우 기존등록한 item/type 유지 해지(REMOVE)시 값 불필요
data	실시간 등록 리스트	LIST
- item	실시간 등록 요소	String[ ]	N	100	거래소별 종목코드, 업종코드 (KRX:039490,NXT:039490_NX,SOR:039490_AL)
- type	실시간 항목	String[ ]	Y	2	TR 명(0A,0B....)

응답

Body

Element한글명typeRequiredLengthDescription

Element

한글명

type

Required

Length

Description

return_code	결과코드	int	N		통신결과에대한 코드 (등록,해지요청시에만 값 전송 0:정상,1:오류 , 데이터 실시간 수신시 미전송)
return_msg	결과메시지	String	N		통신결과에대한메시지
trnm	서비스명	String	N		등록,해지요청시 요청값 반환 , 실시간수신시 REAL 반환
data	실시간 등록리스트	LIST	N
- type	실시간항목	String	N		TR 명(0A,0B....)
- name	실시간 항목명	String	N
- item	실시간 등록 요소	String	N		종목코드
- values	실시간 값 리스트	LIST	N
- - 27	(최우선)매도호가	String	N
- - 28	(최우선)매수호가	String	N

 

주식 우선호가 0C 예제

import asyncio 
import websockets
import json

# socket 정보
# SOCKET_URL = 'wss://mockapi.kiwoom.com:10000/api/dostk/websocket'  # 모의투자 접속 URL
SOCKET_URL = 'wss://api.kiwoom.com:10000/api/dostk/websocket'  # 접속 URL
ACCESS_TOKEN = '사용자 AccessToken'  # 고객 Access Token

class WebSocketClient:
def __init__(self, uri):
self.uri = uri
self.websocket = None
self.connected = False
self.keep_running = True

# WebSocket 서버에 연결합니다.
async def connect(self):
try:
self.websocket = await websockets.connect(self.uri)
self.connected = True
print("서버와 연결을 시도 중입니다.")

# 로그인 패킷
param = {
'trnm': 'LOGIN',
'token': ACCESS_TOKEN
}

print('실시간 시세 서버로 로그인 패킷을 전송합니다.')
# 웹소켓 연결 시 로그인 정보 전달
await self.send_message(message=param)

except Exception as e:
print(f'Connection error: {e}')
self.connected = False

# 서버에 메시지를 보냅니다. 연결이 없다면 자동으로 연결합니다.
async def send_message(self, message):
if not self.connected:
await self.connect()  # 연결이 끊어졌다면 재연결
if self.connected:
# message가 문자열이 아니면 JSON으로 직렬화
if not isinstance(message, str):
message = json.dumps(message)

await self.websocket.send(message)
print(f'Message sent: {message}')

# 서버에서 오는 메시지를 수신하여 출력합니다.
async def receive_messages(self):
while self.keep_running:
try:
# 서버로부터 수신한 메시지를 JSON 형식으로 파싱
response = json.loads(await self.websocket.recv())

# 메시지 유형이 LOGIN일 경우 로그인 시도 결과 체크
if response.get('trnm') == 'LOGIN':
if response.get('return_code') != 0:
print('로그인 실패하였습니다. : ', response.get('return_msg'))
await self.disconnect()
else:
print('로그인 성공하였습니다.')

# 메시지 유형이 PING일 경우 수신값 그대로 송신
elif response.get('trnm') == 'PING':
await self.send_message(response)

if response.get('trnm') != 'PING':
print(f'실시간 시세 서버 응답 수신: {response}')

except websockets.ConnectionClosed:
print('Connection closed by the server')
self.connected = False
await self.websocket.close()

# WebSocket 실행
async def run(self):
await self.connect()
await self.receive_messages()

# WebSocket 연결 종료
async def disconnect(self):
self.keep_running = False
if self.connected and self.websocket:
await self.websocket.close()
self.connected = False
print('Disconnected from WebSocket server')

async def main():
# WebSocketClient 전역 변수 선언
websocket_client = WebSocketClient(SOCKET_URL)

# WebSocket 클라이언트를 백그라운드에서 실행합니다.
receive_task = asyncio.create_task(websocket_client.run())

# 실시간 항목 등록
await asyncio.sleep(1)
await websocket_client.send_message({ 
'trnm': 'REG', # 서비스명
'grp_no': '1', # 그룹번호
'refresh': '1', # 기존등록유지여부
'data': [{ # 실시간 등록 리스트
'item': ['005930'], # 실시간 등록 요소
'type': ['0C'], # 실시간 항목
}]
})

# 수신 작업이 종료될 때까지 대기
await receive_task

# asyncio로 프로그램을 실행합니다.
if __name__ == '__main__':
asyncio.run(main())

---

 

요청

Body

Element한글명typeRequiredLengthDescription

Element

한글명

type

Required

Length

Description

trnm	서비스명	String	Y	10	REG : 등록 , REMOVE : 해지
grp_no	그룹번호	String	Y	4
refresh	기존등록유지여부	String	Y	1	등록(REG)시 0:기존유지안함 1:기존유지(Default) 0일경우 기존등록한 item/type은 해지, 1일경우 기존등록한 item/type 유지 해지(REMOVE)시 값 불필요
data	실시간 등록 리스트	LIST
- item	실시간 등록 요소	String[ ]	N	100	거래소별 종목코드, 업종코드 (KRX:039490,NXT:039490_NX,SOR:039490_AL)
- type	실시간 항목	String[ ]	Y	2	TR 명(0A,0B....)

응답

Body

Element한글명typeRequiredLengthDescription

Element

한글명

type

Required

Length

Description

return_code	결과코드	int	N		통신결과에대한 코드 (등록,해지요청시에만 값 전송 0:정상,1:오류 , 데이터 실시간 수신시 미전송)
return_msg	결과메시지	String	N		통신결과에대한메시지
trnm	서비스명	String	N		등록,해지요청시 요청값 반환 , 실시간수신시 REAL 반환
data	실시간 등록리스트	LIST	N
- type	실시간항목	String	N		TR 명(0A,0B....)
- name	실시간 항목명	String	N
- item	실시간 등록 요소	String	N		종목코드
- values	실시간 값 리스트	LIST	N
- - 21	호가시간	String	N
- - 41	매도호가1	String	N
- - 61	매도호가수량1	String	N
- - 81	매도호가직전대비1	String	N
- - 51	매수호가1	String	N
- - 71	매수호가수량1	String	N
- - 91	매수호가직전대비1	String	N
- - 42	매도호가2	String	N
- - 62	매도호가수량2	String	N
- - 82	매도호가직전대비2	String	N
- - 52	매수호가2	String	N
- - 72	매수호가수량2	String	N
- - 92	매수호가직전대비2	String	N
- - 43	매도호가3	String	N
- - 63	매도호가수량3	String	N
- - 83	매도호가직전대비3	String	N
- - 53	매수호가3	String	N
- - 73	매수호가수량3	String	N
- - 93	매수호가직전대비3	String	N
- - 44	매도호가4	String	N
- - 64	매도호가수량4	String	N
- - 84	매도호가직전대비4	String	N
- - 54	매수호가4	String	N
- - 74	매수호가수량4	String	N
- - 94	매수호가직전대비4	String	N
- - 45	매도호가5	String	N
- - 65	매도호가수량5	String	N
- - 85	매도호가직전대비5	String	N
- - 55	매수호가5	String	N
- - 75	매수호가수량5	String	N
- - 95	매수호가직전대비5	String	N
- - 46	매도호가6	String	N
- - 66	매도호가수량6	String	N
- - 86	매도호가직전대비6	String	N
- - 56	매수호가6	String	N
- - 76	매수호가수량6	String	N
- - 96	매수호가직전대비6	String	N
- - 47	매도호가7	String	N
- - 67	매도호가수량7	String	N
- - 87	매도호가직전대비7	String	N
- - 57	매수호가7	String	N
- - 77	매수호가수량7	String	N
- - 97	매수호가직전대비7	String	N
- - 48	매도호가8	String	N
- - 68	매도호가수량8	String	N
- - 88	매도호가직전대비8	String	N
- - 58	매수호가8	String	N
- - 78	매수호가수량8	String	N
- - 98	매수호가직전대비8	String	N
- - 49	매도호가9	String	N
- - 69	매도호가수량9	String	N
- - 89	매도호가직전대비9	String	N
- - 59	매수호가9	String	N
- - 79	매수호가수량9	String	N
- - 99	매수호가직전대비9	String	N
- - 50	매도호가10	String	N
- - 70	매도호가수량10	String	N
- - 60	매수호가10	String	N
- - 90	매도호가직전대비10	String	N
- - 80	매수호가수량10	String	N
- - 100	매수호가직전대비10	String	N
- - 121	매도호가총잔량	String	N
- - 122	매도호가총잔량직전대비	String	N
- - 125	매수호가총잔량	String	N
- - 126	매수호가총잔량직전대비	String	N
- - 23	예상체결가	String	N
- - 24	예상체결수량	String	N
- - 128	순매수잔량	String	N
- - 129	매수비율	String	N
- - 138	순매도잔량	String	N
- - 139	매도비율	String	N
- - 200	예상체결가전일종가대비	String	N
- - 201	예상체결가전일종가대비등락율	String	N
- - 238	예상체결가전일종가대비기호	String	N
- - 291	예상체결가	String	N		예상체결 시간동안에만 유효한 값
- - 292	예상체결량	String	N
- - 293	예상체결가전일대비기호	String	N
- - 294	예상체결가전일대비	String	N
- - 295	예상체결가전일대비등락율	String	N
- - 621	LP매도호가수량1	String	N
- - 631	LP매수호가수량1	String	N
- - 622	LP매도호가수량2	String	N
- - 632	LP매수호가수량2	String	N
- - 623	LP매도호가수량3	String	N
- - 633	LP매수호가수량3	String	N
- - 624	LP매도호가수량4	String	N
- - 634	LP매수호가수량4	String	N
- - 625	LP매도호가수량5	String	N
- - 635	LP매수호가수량5	String	N
- - 626	LP매도호가수량6	String	N
- - 636	LP매수호가수량6	String	N
- - 627	LP매도호가수량7	String	N
- - 637	LP매수호가수량7	String	N
- - 628	LP매도호가수량8	String	N
- - 638	LP매수호가수량8	String	N
- - 629	LP매도호가수량9	String	N
- - 639	LP매수호가수량9	String	N
- - 630	LP매도호가수량10	String	N
- - 640	LP매수호가수량10	String	N
- - 13	누적거래량	String	N
- - 299	전일거래량대비예상체결율	String	N
- - 215	장운영구분	String	N
- - 216	투자자별ticker	String	N
- - 6044	KRX 매도호가잔량1	String	N
- - 6045	KRX 매도호가잔량2	String	N
- - 6046	KRX 매도호가잔량3	String	N
- - 6047	KRX 매도호가잔량4	String	N
- - 6048	KRX 매도호가잔량5	String	N
- - 6049	KRX 매도호가잔량6	String	N
- - 6050	KRX 매도호가잔량7	String	N
- - 6051	KRX 매도호가잔량8	String	N
- - 6052	KRX 매도호가잔량9	String	N
- - 6053	KRX 매도호가잔량10	String	N
- - 6054	KRX 매수호가잔량1	String	N
- - 6055	KRX 매수호가잔량2	String	N
- - 6056	KRX 매수호가잔량3	String	N
- - 6057	KRX 매수호가잔량4	String	N
- - 6058	KRX 매수호가잔량5	String	N
- - 6059	KRX 매수호가잔량6	String	N
- - 6060	KRX 매수호가잔량7	String	N
- - 6061	KRX 매수호가잔량8	String	N
- - 6062	KRX 매수호가잔량9	String	N
- - 6063	KRX 매수호가잔량10	String	N
- - 6064	KRX 매도호가총잔량	String	N
- - 6065	KRX 매수호가총잔량	String	N
- - 6066	NXT 매도호가잔량1	String	N
- - 6067	NXT 매도호가잔량2	String	N
- - 6068	NXT 매도호가잔량3	String	N
- - 6069	NXT 매도호가잔량4	String	N
- - 6070	NXT 매도호가잔량5	String	N
- - 6071	NXT 매도호가잔량6	String	N
- - 6072	NXT 매도호가잔량7	String	N
- - 6073	NXT 매도호가잔량8	String	N
- - 6074	NXT 매도호가잔량9	String	N
- - 6075	NXT 매도호가잔량10	String	N
- - 6076	NXT 매수호가잔량1	String	N
- - 6077	NXT 매수호가잔량2	String	N
- - 6078	NXT 매수호가잔량3	String	N
- - 6079	NXT 매수호가잔량4	String	N
- - 6080	NXT 매수호가잔량5	String	N
- - 6081	NXT 매수호가잔량6	String	N
- - 6082	NXT 매수호가잔량7	String	N
- - 6083	NXT 매수호가잔량8	String	N
- - 6084	NXT 매수호가잔량9	String	N
- - 6085	NXT 매수호가잔량10	String	N
- - 6086	NXT 매도호가총잔량	String	N
- - 6087	NXT 매수호가총잔량	String	N
- - 6100	KRX 중간가 매도 총잔량 증감	String	N
- - 6101	KRX 중간가 매도 총잔량	String	N
- - 6102	KRX 중간가	String	N
- - 6103	KRX 중간가 매수 총잔량	String	N
- - 6104	KRX 중간가 매수 총잔량 증감	String	N
- - 6105	NXT중간가 매도 총잔량 증감	String	N
- - 6106	NXT중간가 매도 총잔량	String	N
- - 6107	NXT중간가	String	N
- - 6108	NXT중간가 매수 총잔량	String	N
- - 6109	NXT중간가 매수 총잔량 증감	String	N
- - 6110	KRX중간가대비	String	N		기준가대비
- - 6111	KRX중간가대비 기호	String	N		기준가대비
- - 6112	KRX중간가대비등락율	String	N		기준가대비
- - 6113	NXT중간가대비	String	N		기준가대비
- - 6114	NXT중간가대비 기호	String	N		기준가대비
- - 6115	NXT중간가대비등락율	String	N		기준가대비

주식호가잔량 TR 0D

import asyncio 
import websockets
import json

# socket 정보
# SOCKET_URL = 'wss://mockapi.kiwoom.com:10000/api/dostk/websocket'  # 모의투자 접속 URL
SOCKET_URL = 'wss://api.kiwoom.com:10000/api/dostk/websocket'  # 접속 URL
ACCESS_TOKEN = '사용자 AccessToken'  # 고객 Access Token

class WebSocketClient:
def __init__(self, uri):
self.uri = uri
self.websocket = None
self.connected = False
self.keep_running = True

# WebSocket 서버에 연결합니다.
async def connect(self):
try:
self.websocket = await websockets.connect(self.uri)
self.connected = True
print("서버와 연결을 시도 중입니다.")

# 로그인 패킷
param = {
'trnm': 'LOGIN',
'token': ACCESS_TOKEN
}

print('실시간 시세 서버로 로그인 패킷을 전송합니다.')
# 웹소켓 연결 시 로그인 정보 전달
await self.send_message(message=param)

except Exception as e:
print(f'Connection error: {e}')
self.connected = False

# 서버에 메시지를 보냅니다. 연결이 없다면 자동으로 연결합니다.
async def send_message(self, message):
if not self.connected:
await self.connect()  # 연결이 끊어졌다면 재연결
if self.connected:
# message가 문자열이 아니면 JSON으로 직렬화
if not isinstance(message, str):
message = json.dumps(message)

await self.websocket.send(message)
print(f'Message sent: {message}')

# 서버에서 오는 메시지를 수신하여 출력합니다.
async def receive_messages(self):
while self.keep_running:
try:
# 서버로부터 수신한 메시지를 JSON 형식으로 파싱
response = json.loads(await self.websocket.recv())

# 메시지 유형이 LOGIN일 경우 로그인 시도 결과 체크
if response.get('trnm') == 'LOGIN':
if response.get('return_code') != 0:
print('로그인 실패하였습니다. : ', response.get('return_msg'))
await self.disconnect()
else:
print('로그인 성공하였습니다.')

# 메시지 유형이 PING일 경우 수신값 그대로 송신
elif response.get('trnm') == 'PING':
await self.send_message(response)

if response.get('trnm') != 'PING':
print(f'실시간 시세 서버 응답 수신: {response}')

except websockets.ConnectionClosed:
print('Connection closed by the server')
self.connected = False
await self.websocket.close()

# WebSocket 실행
async def run(self):
await self.connect()
await self.receive_messages()

# WebSocket 연결 종료
async def disconnect(self):
self.keep_running = False
if self.connected and self.websocket:
await self.websocket.close()
self.connected = False
print('Disconnected from WebSocket server')

async def main():
# WebSocketClient 전역 변수 선언
websocket_client = WebSocketClient(SOCKET_URL)

# WebSocket 클라이언트를 백그라운드에서 실행합니다.
receive_task = asyncio.create_task(websocket_client.run())

# 실시간 항목 등록
await asyncio.sleep(1)
await websocket_client.send_message({ 
'trnm': 'REG', # 서비스명
'grp_no': '1', # 그룹번호
'refresh': '1', # 기존등록유지여부
'data': [{ # 실시간 등록 리스트
'item': ['005930'], # 실시간 등록 요소
'type': ['0D'], # 실시간 항목
}]
})

# 수신 작업이 종료될 때까지 대기
await receive_task

# asyncio로 프로그램을 실행합니다.
if __name__ == '__main__':
asyncio.run(main())

---

요청 주식 당일 거래원 TR 0F

Body

 

Element	한글명	type	Required	Length	Description
trnm	서비스명	String	Y	10	REG : 등록 , REMOVE : 해지
grp_no	그룹번호	String	Y	4
refresh	기존등록유지여부	String	Y	1	등록(REG)시 0:기존유지안함 1:기존유지(Default) 0일경우 기존등록한 item/type은 해지, 1일경우 기존등록한 item/type 유지 해지(REMOVE)시 값 불필요
data	실시간 등록 리스트	LIST
- item	실시간 등록 요소	String[ ]	N	100	거래소별 종목코드, 업종코드 (KRX:039490,NXT:039490_NX,SOR:039490_AL)
- type	실시간 항목	String[ ]	Y	2	TR 명(0A,0B....)

응답

Body

Element	한글명	type	Required	Length	Description
return_code	결과코드	int	N		통신결과에대한 코드 (등록,해지요청시에만 값 전송 0:정상,1:오류 , 데이터 실시간 수신시 미전송)
return_msg	결과메시지	String	N		통신결과에대한메시지
trnm	서비스명	String	N		등록,해지요청시 요청값 반환 , 실시간수신시 REAL 반환
data	실시간 등록리스트	LIST	N
- type	실시간항목	String	N		TR 명(0A,0B....)
- name	실시간 항목명	String	N
- item	실시간 등록 요소	String	N		종목코드
- values	실시간 값 리스트	LIST	N
- - 141	매도거래원1	String	N
- - 161	매도거래원수량1	String	N
- - 166	매도거래원별증감1	String	N
- - 146	매도거래원코드1	String	N
- - 271	매도거래원색깔1	String	N
- - 151	매수거래원1	String	N
- - 171	매수거래원수량1	String	N
- - 176	매수거래원별증감1	String	N
- - 156	매수거래원코드1	String	N
- - 281	매수거래원색깔1	String	N
- - 142	매도거래원2	String	N
- - 162	매도거래원수량2	String	N
- - 167	매도거래원별증감2	String	N
- - 147	매도거래원코드2	String	N
- - 272	매도거래원색깔2	String	N
- - 152	매수거래원2	String	N
- - 172	매수거래원수량2	String	N
- - 177	매수거래원별증감2	String	N
- - 157	매수거래원코드2	String	N
- - 282	매수거래원색깔2	String	N
- - 143	매도거래원3	String	N
- - 163	매도거래원수량3	String	N
- - 168	매도거래원별증감3	String	N
- - 148	매도거래원코드3	String	N
- - 273	매도거래원색깔3	String	N
- - 153	매수거래원3	String	N
- - 173	매수거래원수량3	String	N
- - 178	매수거래원별증감3	String	N
- - 158	매수거래원코드3	String	N
- - 283	매수거래원색깔3	String	N
- - 144	매도거래원4	String	N
- - 164	매도거래원수량4	String	N
- - 169	매도거래원별증감4	String	N
- - 149	매도거래원코드4	String	N
- - 274	매도거래원색깔4	String	N
- - 154	매수거래원4	String	N
- - 174	매수거래원수량4	String	N
- - 179	매수거래원별증감4	String	N
- - 159	매수거래원코드4	String	N
- - 284	매수거래원색깔4	String	N
- - 145	매도거래원5	String	N
- - 165	매도거래원수량5	String	N
- - 170	매도거래원별증감5	String	N
- - 150	매도거래원코드5	String	N
- - 275	매도거래원색깔5	String	N
- - 155	매수거래원5	String	N
- - 175	매수거래원수량5	String	N
- - 180	매수거래원별증감5	String	N
- - 160	매수거래원코드5	String	N
- - 285	매수거래원색깔5	String	N
- - 261	외국계매도추정합	String	N
- - 262	외국계매도추정합변동	String	N
- - 263	외국계매수추정합	String	N
- - 264	외국계매수추정합변동	String	N
- - 267	외국계순매수추정합	String	N
- - 268	외국계순매수변동	String	N
- - 337	거래소구분	String	N

주식강일거래원 TR 0F

import asyncio 
import websockets
import json

# socket 정보
# SOCKET_URL = 'wss://mockapi.kiwoom.com:10000/api/dostk/websocket'  # 모의투자 접속 URL
SOCKET_URL = 'wss://api.kiwoom.com:10000/api/dostk/websocket'  # 접속 URL
ACCESS_TOKEN = '사용자 AccessToken'  # 고객 Access Token

class WebSocketClient:
def __init__(self, uri):
self.uri = uri
self.websocket = None
self.connected = False
self.keep_running = True

# WebSocket 서버에 연결합니다.
async def connect(self):
try:
self.websocket = await websockets.connect(self.uri)
self.connected = True
print("서버와 연결을 시도 중입니다.")

# 로그인 패킷
param = {
'trnm': 'LOGIN',
'token': ACCESS_TOKEN
}

print('실시간 시세 서버로 로그인 패킷을 전송합니다.')
# 웹소켓 연결 시 로그인 정보 전달
await self.send_message(message=param)

except Exception as e:
print(f'Connection error: {e}')
self.connected = False

# 서버에 메시지를 보냅니다. 연결이 없다면 자동으로 연결합니다.
async def send_message(self, message):
if not self.connected:
await self.connect()  # 연결이 끊어졌다면 재연결
if self.connected:
# message가 문자열이 아니면 JSON으로 직렬화
if not isinstance(message, str):
message = json.dumps(message)

await self.websocket.send(message)
print(f'Message sent: {message}')

# 서버에서 오는 메시지를 수신하여 출력합니다.
async def receive_messages(self):
while self.keep_running:
try:
# 서버로부터 수신한 메시지를 JSON 형식으로 파싱
response = json.loads(await self.websocket.recv())

# 메시지 유형이 LOGIN일 경우 로그인 시도 결과 체크
if response.get('trnm') == 'LOGIN':
if response.get('return_code') != 0:
print('로그인 실패하였습니다. : ', response.get('return_msg'))
await self.disconnect()
else:
print('로그인 성공하였습니다.')

# 메시지 유형이 PING일 경우 수신값 그대로 송신
elif response.get('trnm') == 'PING':
await self.send_message(response)

if response.get('trnm') != 'PING':
print(f'실시간 시세 서버 응답 수신: {response}')

except websockets.ConnectionClosed:
print('Connection closed by the server')
self.connected = False
await self.websocket.close()

# WebSocket 실행
async def run(self):
await self.connect()
await self.receive_messages()

# WebSocket 연결 종료
async def disconnect(self):
self.keep_running = False
if self.connected and self.websocket:
await self.websocket.close()
self.connected = False
print('Disconnected from WebSocket server')

async def main():
# WebSocketClient 전역 변수 선언
websocket_client = WebSocketClient(SOCKET_URL)

# WebSocket 클라이언트를 백그라운드에서 실행합니다.
receive_task = asyncio.create_task(websocket_client.run())

# 실시간 항목 등록
await asyncio.sleep(1)
await websocket_client.send_message({ 
'trnm': 'REG', # 서비스명
'grp_no': '1', # 그룹번호
'refresh': '1', # 기존등록유지여부
'data': [{ # 실시간 등록 리스트
'item': ['005930'], # 실시간 등록 요소
'type': ['0F'], # 실시간 항목
}]
})

# 수신 작업이 종료될 때까지 대기
await receive_task

# asyncio로 프로그램을 실행합니다.
if __name__ == '__main__':
asyncio.run(main())

---

Element

한글명

type

Required

Length

Description

요청

Body

 

Element	한글명	type	Required	Length	Description
trnm	서비스명	String	Y	10	REG : 등록 , REMOVE : 해지
grp_no	그룹번호	String	Y	4
refresh	기존등록유지여부	String	Y	1	등록(REG)시 0:기존유지안함 1:기존유지(Default) 0일경우 기존등록한 item/type은 해지, 1일경우 기존등록한 item/type 유지 해지(REMOVE)시 값 불필요
data	실시간 등록 리스트	LIST
- item	실시간 등록 요소	String[ ]	N	100	거래소별 종목코드, 업종코드 (KRX:039490,NXT:039490_NX,SOR:039490_AL)
- type	실시간 항목	String[ ]	Y	2	TR 명(0A,0B....)

응답

Body

Element	한글명	type	Required	Length	Description
return_code	결과코드	int	N		통신결과에대한 코드 (등록,해지요청시에만 값 전송 0:정상,1:오류 , 데이터 실시간 수신시 미전송)
return_msg	결과메시지	String	N		통신결과에대한메시지
trnm	서비스명	String	N		등록,해지요청시 요청값 반환 , 실시간수신시 REAL 반환
data	실시간 등록리스트	LIST	N
- type	실시간항목	String	N		TR 명(0A,0B....)
- name	실시간 항목명	String	N
- item	실시간 등록 요소	String	N		종목코드
- values	실시간 값 리스트	LIST	N
- - 36	NAV	String	N
- - 37	NAV전일대비	String	N
- - 38	NAV등락율	String	N
- - 39	추적오차율	String	N
- - 20	체결시간	String	N
- - 10	현재가	String	N
- - 11	전일대비	String	N
- - 12	등락율	String	N
- - 13	누적거래량	String	N
- - 25	전일대비기호	String	N
- - 667	ELW기어링비율	String	N
- - 668	ELW손익분기율	String	N
- - 669	ELW자본지지점	String	N
- - 265	NAV/지수괴리율	String	N
- - 266	NAV/ETF괴리율	String	N

ETF NAV TR 0G

import asyncio 
import websockets
import json

# socket 정보
# SOCKET_URL = 'wss://mockapi.kiwoom.com:10000/api/dostk/websocket'  # 모의투자 접속 URL
SOCKET_URL = 'wss://api.kiwoom.com:10000/api/dostk/websocket'  # 접속 URL
ACCESS_TOKEN = '사용자 AccessToken'  # 고객 Access Token

class WebSocketClient:
def __init__(self, uri):
self.uri = uri
self.websocket = None
self.connected = False
self.keep_running = True

# WebSocket 서버에 연결합니다.
async def connect(self):
try:
self.websocket = await websockets.connect(self.uri)
self.connected = True
print("서버와 연결을 시도 중입니다.")

# 로그인 패킷
param = {
'trnm': 'LOGIN',
'token': ACCESS_TOKEN
}

print('실시간 시세 서버로 로그인 패킷을 전송합니다.')
# 웹소켓 연결 시 로그인 정보 전달
await self.send_message(message=param)

except Exception as e:
print(f'Connection error: {e}')
self.connected = False

# 서버에 메시지를 보냅니다. 연결이 없다면 자동으로 연결합니다.
async def send_message(self, message):
if not self.connected:
await self.connect()  # 연결이 끊어졌다면 재연결
if self.connected:
# message가 문자열이 아니면 JSON으로 직렬화
if not isinstance(message, str):
message = json.dumps(message)

await self.websocket.send(message)
print(f'Message sent: {message}')

# 서버에서 오는 메시지를 수신하여 출력합니다.
async def receive_messages(self):
while self.keep_running:
try:
# 서버로부터 수신한 메시지를 JSON 형식으로 파싱
response = json.loads(await self.websocket.recv())

# 메시지 유형이 LOGIN일 경우 로그인 시도 결과 체크
if response.get('trnm') == 'LOGIN':
if response.get('return_code') != 0:
print('로그인 실패하였습니다. : ', response.get('return_msg'))
await self.disconnect()
else:
print('로그인 성공하였습니다.')

# 메시지 유형이 PING일 경우 수신값 그대로 송신
elif response.get('trnm') == 'PING':
await self.send_message(response)

if response.get('trnm') != 'PING':
print(f'실시간 시세 서버 응답 수신: {response}')

except websockets.ConnectionClosed:
print('Connection closed by the server')
self.connected = False
await self.websocket.close()

# WebSocket 실행
async def run(self):
await self.connect()
await self.receive_messages()

# WebSocket 연결 종료
async def disconnect(self):
self.keep_running = False
if self.connected and self.websocket:
await self.websocket.close()
self.connected = False
print('Disconnected from WebSocket server')

async def main():
# WebSocketClient 전역 변수 선언
websocket_client = WebSocketClient(SOCKET_URL)

# WebSocket 클라이언트를 백그라운드에서 실행합니다.
receive_task = asyncio.create_task(websocket_client.run())

# 실시간 항목 등록
await asyncio.sleep(1)
await websocket_client.send_message({ 
'trnm': 'REG', # 서비스명
'grp_no': '1', # 그룹번호
'refresh': '1', # 기존등록유지여부
'data': [{ # 실시간 등록 리스트
'item': ['069500'], # 실시간 등록 요소
'type': ['0G'], # 실시간 항목
}]
})

# 수신 작업이 종료될 때까지 대기
await receive_task

# asyncio로 프로그램을 실행합니다.
if __name__ == '__main__':
asyncio.run(main())

---

요청 주식예상체결 0H

Body

 

Element	한글명	type	Required	Length	Description
trnm	서비스명	String	Y	10	REG : 등록 , REMOVE : 해지
grp_no	그룹번호	String	Y	4
refresh	기존등록유지여부	String	Y	1	등록(REG)시 0:기존유지안함 1:기존유지(Default) 0일경우 기존등록한 item/type은 해지, 1일경우 기존등록한 item/type 유지 해지(REMOVE)시 값 불필요
data	실시간 등록 리스트	LIST
- item	실시간 등록 요소	String[ ]	N	100	거래소별 종목코드, 업종코드 (KRX:039490,NXT:039490_NX,SOR:039490_AL)
- type	실시간 항목	String[ ]	Y	2	TR 명(0A,0B....)

응답

Body 

Element	한글명	type	Required	Length	Description
return_code	결과코드	int	N		통신결과에대한 코드 (등록,해지요청시에만 값 전송 0:정상,1:오류 , 데이터 실시간 수신시 미전송)
return_msg	결과메시지	String	N		통신결과에대한메시지
trnm	서비스명	String	N		등록,해지요청시 요청값 반환 , 실시간수신시 REAL 반환
data	실시간 등록리스트	LIST	N
- type	실시간항목	String	N		TR 명(0A,0B....)
- name	실시간 항목명	String	N
- item	실시간 등록 요소	String	N		종목코드
- values	실시간 값 리스트	LIST	N
- - 20	체결시간	String	N
- - 10	현재가	String	N
- - 11	전일대비	String	N
- - 12	등락율	String	N
- - 15	거래량	String	N		+는 매수체결, -는 매도체결
- - 13	누적거래량	String	N
- - 25	전일대비기호	String	N

---

요청 업종지수 0J

Body

Element	한글명	type	Required	Length	Description
trnm	서비스명	String	Y	10	REG : 등록 , REMOVE : 해지
grp_no	그룹번호	String	Y	4
refresh	기존등록유지여부	String	Y	1	등록(REG)시 0:기존유지안함 1:기존유지(Default) 0일경우 기존등록한 item/type은 해지, 1일경우 기존등록한 item/type 유지 해지(REMOVE)시 값 불필요
data	실시간 등록 리스트	LIST
- item	실시간 등록 요소	String[ ]	N	100	거래소별 종목코드, 업종코드 (KRX:039490,NXT:039490_NX,SOR:039490_AL)
- type	실시간 항목	String[ ]	Y	2	TR 명(0A,0B....)

응답

Body

 

Element	한글명	type	Required	Length	Description
return_code	결과코드	int	N		통신결과에대한 코드 (등록,해지요청시에만 값 전송 0:정상,1:오류 , 데이터 실시간 수신시 미전송)
return_msg	결과메시지	String	N		통신결과에대한메시지
trnm	서비스명	String	N		등록,해지요청시 요청값 반환 , 실시간수신시 REAL 반환
data	실시간 등록리스트	LIST	N
- type	실시간항목	String	N		TR 명(0A,0B....)
- name	실시간 항목명	String	N
- item	실시간 등록 요소	String	N		종목코드
- values	실시간 값 리스트	LIST	N
- - 20	체결시간	String	N
- - 10	현재가	String	N
- - 11	전일대비	String	N
- - 12	등락율	String	N
- - 15	거래량	String	N		+는 매수체결,-는 매도체결
- - 13	누적거래량	String	N
- - 14	누적거래대금	String	N
- - 16	시가	String	N
- - 17	고가	String	N
- - 18	저가	String	N
- - 25	전일대비기호	String	N
- - 26	전일거래량대비	String	N		계약,주

업종지수 0J 예제

import asyncio 
import websockets
import json

# socket 정보
# SOCKET_URL = 'wss://mockapi.kiwoom.com:10000/api/dostk/websocket'  # 모의투자 접속 URL
SOCKET_URL = 'wss://api.kiwoom.com:10000/api/dostk/websocket'  # 접속 URL
ACCESS_TOKEN = '사용자 AccessToken'  # 고객 Access Token

class WebSocketClient:
def __init__(self, uri):
self.uri = uri
self.websocket = None
self.connected = False
self.keep_running = True

# WebSocket 서버에 연결합니다.
async def connect(self):
try:
self.websocket = await websockets.connect(self.uri)
self.connected = True
print("서버와 연결을 시도 중입니다.")

# 로그인 패킷
param = {
'trnm': 'LOGIN',
'token': ACCESS_TOKEN
}

print('실시간 시세 서버로 로그인 패킷을 전송합니다.')
# 웹소켓 연결 시 로그인 정보 전달
await self.send_message(message=param)

except Exception as e:
print(f'Connection error: {e}')
self.connected = False

# 서버에 메시지를 보냅니다. 연결이 없다면 자동으로 연결합니다.
async def send_message(self, message):
if not self.connected:
await self.connect()  # 연결이 끊어졌다면 재연결
if self.connected:
# message가 문자열이 아니면 JSON으로 직렬화
if not isinstance(message, str):
message = json.dumps(message)

await self.websocket.send(message)
print(f'Message sent: {message}')

# 서버에서 오는 메시지를 수신하여 출력합니다.
async def receive_messages(self):
while self.keep_running:
try:
# 서버로부터 수신한 메시지를 JSON 형식으로 파싱
response = json.loads(await self.websocket.recv())

# 메시지 유형이 LOGIN일 경우 로그인 시도 결과 체크
if response.get('trnm') == 'LOGIN':
if response.get('return_code') != 0:
print('로그인 실패하였습니다. : ', response.get('return_msg'))
await self.disconnect()
else:
print('로그인 성공하였습니다.')

# 메시지 유형이 PING일 경우 수신값 그대로 송신
elif response.get('trnm') == 'PING':
await self.send_message(response)

if response.get('trnm') != 'PING':
print(f'실시간 시세 서버 응답 수신: {response}')

except websockets.ConnectionClosed:
print('Connection closed by the server')
self.connected = False
await self.websocket.close()

# WebSocket 실행
async def run(self):
await self.connect()
await self.receive_messages()

# WebSocket 연결 종료
async def disconnect(self):
self.keep_running = False
if self.connected and self.websocket:
await self.websocket.close()
self.connected = False
print('Disconnected from WebSocket server')

async def main():
# WebSocketClient 전역 변수 선언
websocket_client = WebSocketClient(SOCKET_URL)

# WebSocket 클라이언트를 백그라운드에서 실행합니다.
receive_task = asyncio.create_task(websocket_client.run())

# 실시간 항목 등록
await asyncio.sleep(1)
await websocket_client.send_message({ 
'trnm': 'REG', # 서비스명
'grp_no': '1', # 그룹번호
'refresh': '1', # 기존등록유지여부
'data': [{ # 실시간 등록 리스트
'item': ['001'], # 실시간 등록 요소
'type': ['0J'], # 실시간 항목
}]
})

# 수신 작업이 종료될 때까지 대기
await receive_task

# asyncio로 프로그램을 실행합니다.
if __name__ == '__main__':
asyncio.run(main())

---

요청 업종등락 0U

Body 

Element	한글명	type	Required	Length	Description
trnm	서비스명	String	Y	10	REG : 등록 , REMOVE : 해지
grp_no	그룹번호	String	Y	4
refresh	기존등록유지여부	String	Y	1	등록(REG)시 0:기존유지안함 1:기존유지(Default) 0일경우 기존등록한 item/type은 해지, 1일경우 기존등록한 item/type 유지 해지(REMOVE)시 값 불필요
data	실시간 등록 리스트	LIST
- item	실시간 등록 요소	String[ ]	N	100	거래소별 종목코드, 업종코드 (KRX:039490,NXT:039490_NX,SOR:039490_AL)
- type	실시간 항목	String[ ]	Y	2	TR 명(0A,0B....)

응답

Body

Element	한글명	type	Required	Length	Description
return_code	결과코드	int	N		통신결과에대한 코드 (등록,해지요청시에만 값 전송 0:정상,1:오류 , 데이터 실시간 수신시 미전송)
return_msg	결과메시지	String	N		통신결과에대한메시지
trnm	서비스명	String	N		등록,해지요청시 요청값 반환 , 실시간수신시 REAL 반환
data	실시간 등록리스트	LIST	N
- type	실시간항목	String	N		TR 명(0A,0B....)
- name	실시간 항목명	String	N
- item	실시간 등록 요소	String	N		종목코드
- values	실시간 값 리스트	LIST	N
- - 20	체결시간	String	N
- - 252	상승종목수	String	N
- - 251	상한종목수	String	N
- - 253	보합종목수	String	N
- - 255	하락종목수	String	N
- - 254	하한종목수	String	N
- - 13	누적거래량	String	N
- - 14	누적거래대금	String	N
- - 10	현재가	String	N
- - 11	전일대비	String	N
- - 12	등락율	String	N
- - 256	거래형성종목수	String	N		계약,주
- - 257	거래형성비율	String	N
- - 25	전일대비기호	String	N

업종등락 0U 예제

import asyncio 
import websockets
import json

# socket 정보
# SOCKET_URL = 'wss://mockapi.kiwoom.com:10000/api/dostk/websocket'  # 모의투자 접속 URL
SOCKET_URL = 'wss://api.kiwoom.com:10000/api/dostk/websocket'  # 접속 URL
ACCESS_TOKEN = '사용자 AccessToken'  # 고객 Access Token

class WebSocketClient:
def __init__(self, uri):
self.uri = uri
self.websocket = None
self.connected = False
self.keep_running = True

# WebSocket 서버에 연결합니다.
async def connect(self):
try:
self.websocket = await websockets.connect(self.uri)
self.connected = True
print("서버와 연결을 시도 중입니다.")

# 로그인 패킷
param = {
'trnm': 'LOGIN',
'token': ACCESS_TOKEN
}

print('실시간 시세 서버로 로그인 패킷을 전송합니다.')
# 웹소켓 연결 시 로그인 정보 전달
await self.send_message(message=param)

except Exception as e:
print(f'Connection error: {e}')
self.connected = False

# 서버에 메시지를 보냅니다. 연결이 없다면 자동으로 연결합니다.
async def send_message(self, message):
if not self.connected:
await self.connect()  # 연결이 끊어졌다면 재연결
if self.connected:
# message가 문자열이 아니면 JSON으로 직렬화
if not isinstance(message, str):
message = json.dumps(message)

await self.websocket.send(message)
print(f'Message sent: {message}')

# 서버에서 오는 메시지를 수신하여 출력합니다.
async def receive_messages(self):
while self.keep_running:
try:
# 서버로부터 수신한 메시지를 JSON 형식으로 파싱
response = json.loads(await self.websocket.recv())

# 메시지 유형이 LOGIN일 경우 로그인 시도 결과 체크
if response.get('trnm') == 'LOGIN':
if response.get('return_code') != 0:
print('로그인 실패하였습니다. : ', response.get('return_msg'))
await self.disconnect()
else:
print('로그인 성공하였습니다.')

# 메시지 유형이 PING일 경우 수신값 그대로 송신
elif response.get('trnm') == 'PING':
await self.send_message(response)

if response.get('trnm') != 'PING':
print(f'실시간 시세 서버 응답 수신: {response}')

except websockets.ConnectionClosed:
print('Connection closed by the server')
self.connected = False
await self.websocket.close()

# WebSocket 실행
async def run(self):
await self.connect()
await self.receive_messages()

# WebSocket 연결 종료
async def disconnect(self):
self.keep_running = False
if self.connected and self.websocket:
await self.websocket.close()
self.connected = False
print('Disconnected from WebSocket server')

async def main():
# WebSocketClient 전역 변수 선언
websocket_client = WebSocketClient(SOCKET_URL)

# WebSocket 클라이언트를 백그라운드에서 실행합니다.
receive_task = asyncio.create_task(websocket_client.run())

# 실시간 항목 등록
await asyncio.sleep(1)
await websocket_client.send_message({ 
'trnm': 'REG', # 서비스명
'grp_no': '1', # 그룹번호
'refresh': '1', # 기존등록유지여부
'data': [{ # 실시간 등록 리스트
'item': ['001'], # 실시간 등록 요소
'type': ['0U'], # 실시간 항목
}]
})

# 수신 작업이 종료될 때까지 대기
await receive_task

# asyncio로 프로그램을 실행합니다.
if __name__ == '__main__':
asyncio.run(main())

---

요청 주식종목정보 TR 0g 

Body

 

Element	한글명	type	Required	Length	Description
trnm	서비스명	String	Y	10	REG : 등록 , REMOVE : 해지
grp_no	그룹번호	String	Y	4
refresh	기존등록유지여부	String	Y	1	등록(REG)시 0:기존유지안함 1:기존유지(Default) 0일경우 기존등록한 item/type은 해지, 1일경우 기존등록한 item/type 유지 해지(REMOVE)시 값 불필요
data	실시간 등록 리스트	LIST
- item	실시간 등록 요소	String[ ]	N	100	거래소별 종목코드, 업종코드 (KRX:039490,NXT:039490_NX,SOR:039490_AL)
- type	실시간 항목	String[ ]	Y	2	TR 명(0A,0B....)

응답

Body

Element	한글명	type	Required	Length	Description
return_code	결과코드	int	N		통신결과에대한 코드 (등록,해지요청시에만 값 전송 0:정상,1:오류 , 데이터 실시간 수신시 미전송)
return_msg	결과메시지	String	N		통신결과에대한메시지
trnm	서비스명	String	N		등록,해지요청시 요청값 반환 , 실시간수신시 REAL 반환
data	실시간 등록리스트	LIST	N
- type	실시간항목	String	N		TR 명(0A,0B....)
- name	실시간 항목명	String	N
- item	실시간 등록 요소	String	N		종목코드
- values	실시간 값 리스트	LIST	N
- - 297	임의연장	String	N
- - 592	장전임의연장	String	N
- - 593	장후임의연장	String	N
- - 305	싱힌기	String	N
- - 306	하한가	String	N
- - 307	기준가	String	N
- - 689	조기종료ELW발생	String	N
- - 594	통화단위	String	N
- - 382	증거금율표시	String	N
- - 370	종목정보	String	N

주식종목정보 0g 

import asyncio 
import websockets
import json

# socket 정보
# SOCKET_URL = 'wss://mockapi.kiwoom.com:10000/api/dostk/websocket'  # 모의투자 접속 URL
SOCKET_URL = 'wss://api.kiwoom.com:10000/api/dostk/websocket'  # 접속 URL
ACCESS_TOKEN = '사용자 AccessToken'  # 고객 Access Token

class WebSocketClient:
def __init__(self, uri):
self.uri = uri
self.websocket = None
self.connected = False
self.keep_running = True

# WebSocket 서버에 연결합니다.
async def connect(self):
try:
self.websocket = await websockets.connect(self.uri)
self.connected = True
print("서버와 연결을 시도 중입니다.")

# 로그인 패킷
param = {
'trnm': 'LOGIN',
'token': ACCESS_TOKEN
}

print('실시간 시세 서버로 로그인 패킷을 전송합니다.')
# 웹소켓 연결 시 로그인 정보 전달
await self.send_message(message=param)

except Exception as e:
print(f'Connection error: {e}')
self.connected = False

# 서버에 메시지를 보냅니다. 연결이 없다면 자동으로 연결합니다.
async def send_message(self, message):
if not self.connected:
await self.connect()  # 연결이 끊어졌다면 재연결
if self.connected:
# message가 문자열이 아니면 JSON으로 직렬화
if not isinstance(message, str):
message = json.dumps(message)

await self.websocket.send(message)
print(f'Message sent: {message}')

# 서버에서 오는 메시지를 수신하여 출력합니다.
async def receive_messages(self):
while self.keep_running:
try:
# 서버로부터 수신한 메시지를 JSON 형식으로 파싱
response = json.loads(await self.websocket.recv())

# 메시지 유형이 LOGIN일 경우 로그인 시도 결과 체크
if response.get('trnm') == 'LOGIN':
if response.get('return_code') != 0:
print('로그인 실패하였습니다. : ', response.get('return_msg'))
await self.disconnect()
else:
print('로그인 성공하였습니다.')

# 메시지 유형이 PING일 경우 수신값 그대로 송신
elif response.get('trnm') == 'PING':
await self.send_message(response)

if response.get('trnm') != 'PING':
print(f'실시간 시세 서버 응답 수신: {response}')

except websockets.ConnectionClosed:
print('Connection closed by the server')
self.connected = False
await self.websocket.close()

# WebSocket 실행
async def run(self):
await self.connect()
await self.receive_messages()

# WebSocket 연결 종료
async def disconnect(self):
self.keep_running = False
if self.connected and self.websocket:
await self.websocket.close()
self.connected = False
print('Disconnected from WebSocket server')

async def main():
# WebSocketClient 전역 변수 선언
websocket_client = WebSocketClient(SOCKET_URL)

# WebSocket 클라이언트를 백그라운드에서 실행합니다.
receive_task = asyncio.create_task(websocket_client.run())

# 실시간 항목 등록
await asyncio.sleep(1)
await websocket_client.send_message({ 
'trnm': 'REG', # 서비스명
'grp_no': '1', # 그룹번호
'refresh': '1', # 기존등록유지여부
'data': [{ # 실시간 등록 리스트
'item': ['005930'], # 실시간 등록 요소
'type': ['0g'], # 실시간 항목
}]
})

# 수신 작업이 종료될 때까지 대기
await receive_task

# asyncio로 프로그램을 실행합니다.
if __name__ == '__main__':
asyncio.run(main())

---

요청 장시작시간 0s

Body

 

Element	한글명	type	Required	Length	Description
trnm	서비스명	String	Y	10	REG : 등록 , REMOVE : 해지
grp_no	그룹번호	String	Y	4
refresh	기존등록유지여부	String	Y	1	등록(REG)시 0:기존유지안함 1:기존유지(Default) 0일경우 기존등록한 item/type은 해지, 1일경우 기존등록한 item/type 유지 해지(REMOVE)시 값 불필요
data	실시간 등록 리스트	LIST
- item	실시간 등록 요소	String[ ]	N	100	거래소별 종목코드, 업종코드 (KRX:039490,NXT:039490_NX,SOR:039490_AL)
- type	실시간 항목	String[ ]	Y	2	TR 명(0A,0B....)

응답

Body

Element	한글명	type	Required	Length	Description
return_code	결과코드	int	N		통신결과에대한 코드 (등록,해지요청시에만 값 전송 0:정상,1:오류 , 데이터 실시간 수신시 미전송)
return_msg	결과메시지	String	N		통신결과에대한메시지
trnm	서비스명	String	N		등록,해지요청시 요청값 반환 , 실시간수신시 REAL 반환
data	실시간 등록리스트	LIST	N
- type	실시간항목	String	N		TR 명(0A,0B....)
- name	실시간 항목명	String	N
- item	실시간 등록 요소	String	N		종목코드
- values	실시간 값 리스트	LIST	N
- - 215	장운영구분	String	N		0:KRX장전,3:KRX장시작,P:NXT프리마켓개시,Q:NXT프리마켓종료,R:NXT메인마켓개시,S:NXT메인마켓 종료,T:NXT애프터마켓단일가개시,U:NXT애프터마켓개시,V:NXT종가매매종료,W:NXT애프터마켓종료
- - 20	체결시간	String	N
- - 214	장시작예상잔여시간	String	N

장시작시간 0s

import asyncio 
import websockets
import json

# socket 정보
# SOCKET_URL = 'wss://mockapi.kiwoom.com:10000/api/dostk/websocket'  # 모의투자 접속 URL
SOCKET_URL = 'wss://api.kiwoom.com:10000/api/dostk/websocket'  # 접속 URL
ACCESS_TOKEN = '사용자 AccessToken'  # 고객 Access Token

class WebSocketClient:
def __init__(self, uri):
self.uri = uri
self.websocket = None
self.connected = False
self.keep_running = True

# WebSocket 서버에 연결합니다.
async def connect(self):
try:
self.websocket = await websockets.connect(self.uri)
self.connected = True
print("서버와 연결을 시도 중입니다.")

# 로그인 패킷
param = {
'trnm': 'LOGIN',
'token': ACCESS_TOKEN
}

print('실시간 시세 서버로 로그인 패킷을 전송합니다.')
# 웹소켓 연결 시 로그인 정보 전달
await self.send_message(message=param)

except Exception as e:
print(f'Connection error: {e}')
self.connected = False

# 서버에 메시지를 보냅니다. 연결이 없다면 자동으로 연결합니다.
async def send_message(self, message):
if not self.connected:
await self.connect()  # 연결이 끊어졌다면 재연결
if self.connected:
# message가 문자열이 아니면 JSON으로 직렬화
if not isinstance(message, str):
message = json.dumps(message)

await self.websocket.send(message)
print(f'Message sent: {message}')

# 서버에서 오는 메시지를 수신하여 출력합니다.
async def receive_messages(self):
while self.keep_running:
try:
# 서버로부터 수신한 메시지를 JSON 형식으로 파싱
response = json.loads(await self.websocket.recv())

# 메시지 유형이 LOGIN일 경우 로그인 시도 결과 체크
if response.get('trnm') == 'LOGIN':
if response.get('return_code') != 0:
print('로그인 실패하였습니다. : ', response.get('return_msg'))
await self.disconnect()
else:
print('로그인 성공하였습니다.')

# 메시지 유형이 PING일 경우 수신값 그대로 송신
elif response.get('trnm') == 'PING':
await self.send_message(response)

if response.get('trnm') != 'PING':
print(f'실시간 시세 서버 응답 수신: {response}')

except websockets.ConnectionClosed:
print('Connection closed by the server')
self.connected = False
await self.websocket.close()

# WebSocket 실행
async def run(self):
await self.connect()
await self.receive_messages()

# WebSocket 연결 종료
async def disconnect(self):
self.keep_running = False
if self.connected and self.websocket:
await self.websocket.close()
self.connected = False
print('Disconnected from WebSocket server')

async def main():
# WebSocketClient 전역 변수 선언
websocket_client = WebSocketClient(SOCKET_URL)

# WebSocket 클라이언트를 백그라운드에서 실행합니다.
receive_task = asyncio.create_task(websocket_client.run())

# 실시간 항목 등록
await asyncio.sleep(1)
await websocket_client.send_message({ 
'trnm': 'REG', # 서비스명
'grp_no': '1', # 그룹번호
'refresh': '1', # 기존등록유지여부
'data': [{ # 실시간 등록 리스트
'item': [''], # 실시간 등록 요소
'type': ['0s'], # 실시간 항목
}]
})

# 수신 작업이 종료될 때까지 대기
await receive_task

# asyncio로 프로그램을 실행합니다.
if __name__ == '__main__':
asyncio.run(main())

---

요청 종목프로그램매매 0w

Body

Element	한글명	type	Required	Length	Description
trnm	서비스명	String	Y	10	REG : 등록 , REMOVE : 해지
grp_no	그룹번호	String	Y	4
refresh	기존등록유지여부	String	Y	1	등록(REG)시 0:기존유지안함 1:기존유지(Default) 0일경우 기존등록한 item/type은 해지, 1일경우 기존등록한 item/type 유지 해지(REMOVE)시 값 불필요
data	실시간 등록 리스트	LIST
- item	실시간 등록 요소	String[ ]	N	100	거래소별 종목코드, 업종코드 (KRX:039490,NXT:039490_NX,SOR:039490_AL)
- type	실시간 항목	String[ ]	Y	2	TR 명(0A,0B....)

응답

Body

Element	한글명	type	Required	Length	Description
return_code	결과코드	int	N		통신결과에대한 코드 (등록,해지요청시에만 값 전송 0:정상,1:오류 , 데이터 실시간 수신시 미전송)
return_msg	결과메시지	String	N		통신결과에대한메시지
trnm	서비스명	String	N		등록,해지요청시 요청값 반환 , 실시간수신시 REAL 반환
data	실시간 등록리스트	LIST	N
- type	실시간항목	String	N		TR 명(0A,0B....)
- name	실시간 항목명	String	N
- item	실시간 등록 요소	String	N		종목코드
- values	실시간 값 리스트	LIST	N
- - 20	체결시간	String	N
- - 10	현재가	String	N
- - 25	전일대비기호	String	N
- - 11	전일대비	String	N
- - 12	등락율	String	N
- - 13	누적거래량	String	N
- - 202	매도수량	String	N
- - 204	매도금액	String	N
- - 206	매수수량	String	N
- - 208	매수금액	String	N
- - 210	순매수수량	String	N
- - 211	순매수수량증감	String	N		계약,주
- - 212	순매수금액	String	N
- - 213	순매수금액증감	String	N
- - 214	장시작예상잔여시간	String	N
- - 215	장운영구분	String	N
- - 216	투자자별ticker	String	N

종목 프로그램 매매 0w

import asyncio 
import websockets
import json

# socket 정보
# SOCKET_URL = 'wss://mockapi.kiwoom.com:10000/api/dostk/websocket'  # 모의투자 접속 URL
SOCKET_URL = 'wss://api.kiwoom.com:10000/api/dostk/websocket'  # 접속 URL
ACCESS_TOKEN = '사용자 AccessToken'  # 고객 Access Token

class WebSocketClient:
def __init__(self, uri):
self.uri = uri
self.websocket = None
self.connected = False
self.keep_running = True

# WebSocket 서버에 연결합니다.
async def connect(self):
try:
self.websocket = await websockets.connect(self.uri)
self.connected = True
print("서버와 연결을 시도 중입니다.")

# 로그인 패킷
param = {
'trnm': 'LOGIN',
'token': ACCESS_TOKEN
}

print('실시간 시세 서버로 로그인 패킷을 전송합니다.')
# 웹소켓 연결 시 로그인 정보 전달
await self.send_message(message=param)

except Exception as e:
print(f'Connection error: {e}')
self.connected = False

# 서버에 메시지를 보냅니다. 연결이 없다면 자동으로 연결합니다.
async def send_message(self, message):
if not self.connected:
await self.connect()  # 연결이 끊어졌다면 재연결
if self.connected:
# message가 문자열이 아니면 JSON으로 직렬화
if not isinstance(message, str):
message = json.dumps(message)

await self.websocket.send(message)
print(f'Message sent: {message}')

# 서버에서 오는 메시지를 수신하여 출력합니다.
async def receive_messages(self):
while self.keep_running:
try:
# 서버로부터 수신한 메시지를 JSON 형식으로 파싱
response = json.loads(await self.websocket.recv())

# 메시지 유형이 LOGIN일 경우 로그인 시도 결과 체크
if response.get('trnm') == 'LOGIN':
if response.get('return_code') != 0:
print('로그인 실패하였습니다. : ', response.get('return_msg'))
await self.disconnect()
else:
print('로그인 성공하였습니다.')

# 메시지 유형이 PING일 경우 수신값 그대로 송신
elif response.get('trnm') == 'PING':
await self.send_message(response)

if response.get('trnm') != 'PING':
print(f'실시간 시세 서버 응답 수신: {response}')

except websockets.ConnectionClosed:
print('Connection closed by the server')
self.connected = False
await self.websocket.close()

# WebSocket 실행
async def run(self):
await self.connect()
await self.receive_messages()

# WebSocket 연결 종료
async def disconnect(self):
self.keep_running = False
if self.connected and self.websocket:
await self.websocket.close()
self.connected = False
print('Disconnected from WebSocket server')

async def main():
# WebSocketClient 전역 변수 선언
websocket_client = WebSocketClient(SOCKET_URL)

# WebSocket 클라이언트를 백그라운드에서 실행합니다.
receive_task = asyncio.create_task(websocket_client.run())

# 실시간 항목 등록
await asyncio.sleep(1)
await websocket_client.send_message({ 
'trnm': 'REG', # 서비스명
'grp_no': '1', # 그룹번호
'refresh': '1', # 기존등록유지여부
'data': [{ # 실시간 등록 리스트
'item': ['005930'], # 실시간 등록 요소
'type': ['0w'], # 실시간 항목
}]
})

# 수신 작업이 종료될 때까지 대기
await receive_task

# asyncio로 프로그램을 실행합니다.
if __name__ == '__main__':
asyncio.run(main())

---

요청 vi 발동/해제 1h

Body 

Element	한글명	type	Required	Length	Description
trnm	서비스명	String	Y	10	REG : 등록 , REMOVE : 해지
grp_no	그룹번호	String	Y	4
refresh	기존등록유지여부	String	Y	1	등록(REG)시 0:기존유지안함 1:기존유지(Default) 0일경우 기존등록한 item/type은 해지, 1일경우 기존등록한 item/type 유지 해지(REMOVE)시 값 불필요
data	실시간 등록 리스트	LIST
- item	실시간 등록 요소	String[ ]	N	100	거래소별 종목코드, 업종코드 (KRX:039490,NXT:039490_NX,SOR:039490_AL)
- type	실시간 항목	String[ ]	Y	2	TR 명(0A,0B....)

응답

Body

Element	한글명	type	Required	Length	Description
return_code	결과코드	int	N		통신결과에대한 코드 (등록,해지요청시에만 값 전송 0:정상,1:오류 , 데이터 실시간 수신시 미전송)
return_msg	결과메시지	String	N		통신결과에대한메시지
trnm	서비스명	String	N		등록,해지요청시 요청값 반환 , 실시간수신시 REAL 반환
data	실시간 등록리스트	LIST	N
- type	실시간항목	String	N		TR 명(0A,0B....)
- name	실시간 항목명	String	N
- item	실시간 등록 요소	String	N		종목코드
- values	실시간 값 리스트	LIST	N
- - 9001	종목코드	String	N
- - 302	종목명	String	N
- - 13	누적거래량	String	N
- - 14	누적거래대금	String	N
- - 9068	VI발동구분	String	N
- - 9008	KOSPI,KOSDAQ,전체구분	String	N
- - 9075	장전구분	String	N
- - 1221	VI발동가격	String	N
- - 1223	매매체결처리시각	String	N
- - 1224	VI해제시각	String	N
- - 1225	VI적용구분	String	N		정적/동적/동적+정적
- - 1236	기준가격 정적	String	N		계약,주
- - 1237	기준가격 동적	String	N
- - 1238	괴리율 정적	String	N
- - 1239	괴리율 동적	String	N
- - 1489	VI발동가 등락율	String	N
- - 1490	VI발동횟수	String	N
- - 9069	발동방향구분	String	N
- - 1279	Extra Item	String	N

vi 발동/해제 예제

import asyncio 
import websockets
import json

# socket 정보
# SOCKET_URL = 'wss://mockapi.kiwoom.com:10000/api/dostk/websocket'  # 모의투자 접속 URL
SOCKET_URL = 'wss://api.kiwoom.com:10000/api/dostk/websocket'  # 접속 URL
ACCESS_TOKEN = '사용자 AccessToken'  # 고객 Access Token

class WebSocketClient:
def __init__(self, uri):
self.uri = uri
self.websocket = None
self.connected = False
self.keep_running = True

# WebSocket 서버에 연결합니다.
async def connect(self):
try:
self.websocket = await websockets.connect(self.uri)
self.connected = True
print("서버와 연결을 시도 중입니다.")

# 로그인 패킷
param = {
'trnm': 'LOGIN',
'token': ACCESS_TOKEN
}

print('실시간 시세 서버로 로그인 패킷을 전송합니다.')
# 웹소켓 연결 시 로그인 정보 전달
await self.send_message(message=param)

except Exception as e:
print(f'Connection error: {e}')
self.connected = False

# 서버에 메시지를 보냅니다. 연결이 없다면 자동으로 연결합니다.
async def send_message(self, message):
if not self.connected:
await self.connect()  # 연결이 끊어졌다면 재연결
if self.connected:
# message가 문자열이 아니면 JSON으로 직렬화
if not isinstance(message, str):
message = json.dumps(message)

await self.websocket.send(message)
print(f'Message sent: {message}')

# 서버에서 오는 메시지를 수신하여 출력합니다.
async def receive_messages(self):
while self.keep_running:
try:
# 서버로부터 수신한 메시지를 JSON 형식으로 파싱
response = json.loads(await self.websocket.recv())

# 메시지 유형이 LOGIN일 경우 로그인 시도 결과 체크
if response.get('trnm') == 'LOGIN':
if response.get('return_code') != 0:
print('로그인 실패하였습니다. : ', response.get('return_msg'))
await self.disconnect()
else:
print('로그인 성공하였습니다.')

# 메시지 유형이 PING일 경우 수신값 그대로 송신
elif response.get('trnm') == 'PING':
await self.send_message(response)

if response.get('trnm') != 'PING':
print(f'실시간 시세 서버 응답 수신: {response}')

except websockets.ConnectionClosed:
print('Connection closed by the server')
self.connected = False
await self.websocket.close()

# WebSocket 실행
async def run(self):
await self.connect()
await self.receive_messages()

# WebSocket 연결 종료
async def disconnect(self):
self.keep_running = False
if self.connected and self.websocket:
await self.websocket.close()
self.connected = False
print('Disconnected from WebSocket server')

async def main():
# WebSocketClient 전역 변수 선언
websocket_client = WebSocketClient(SOCKET_URL)

# WebSocket 클라이언트를 백그라운드에서 실행합니다.
receive_task = asyncio.create_task(websocket_client.run())

# 실시간 항목 등록
await asyncio.sleep(1)
await websocket_client.send_message({ 
'trnm': 'REG', # 서비스명
'grp_no': '1', # 그룹번호
'refresh': '1', # 기존등록유지여부
'data': [{ # 실시간 등록 리스트
'item': [''], # 실시간 등록 요소
'type': ['1h'], # 실시간 항목
}]
})

# 수신 작업이 종료될 때까지 대기
await receive_task

# asyncio로 프로그램을 실행합니다.
if __name__ == '__main__':
asyncio.run(main())