shuim/auto-stock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge: phase3 realtime module refactor into main

Browse Source 
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
shuim
2026-03-06 11:05:39 +09:00
parent eef9620aa6 48aa5440ec
commit 6d39069d3f
6 changed files with 1056 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

275
backend/app/realtime/README.md Normal file
View File

@@ -0,0 +1,275 @@
# Realtime Module
WebSocket 기반 실시간 시세 수신 및 이벤트 처리 모듈
## 모듈 구조
```
realtime/
├── __init__.py # 모듈 내보내기
├── websocket_client.py # KIS WebSocket 클라이언트
├── event_handler.py # 이벤트 핸들러 관리자
├── market_scheduler.py # 장 스케줄 관리
├── examples.py # 사용 예시
└── README.md
```
## 사용 방법
### 1. 기본 WebSocket 연결
```python
from app.realtime import KISWebSocketClient, event_handler, EventType
from app.realtime.websocket_client import create_websocket_client
# 핸들러 등록
async def on_price_update(event_data):
print(f"{event_data['stock_code']}: {event_data['current_price']}원")
event_handler.register(EventType.PRICE_UPDATE, on_price_update)
# WebSocket 연결
ws_client = create_websocket_client()
await ws_client.connect()
# 종목 구독
await ws_client.subscribe_price("005930") # 삼성전자
# 메시지 수신 시작
await ws_client.listen()
```
### 2. 이벤트 핸들러
```python
from app.realtime import event_handler, EventType
# 특정 이벤트 핸들러
async def handle_price_update(event_data):
stock_code = event_data['stock_code']
price = event_data['current_price']
print(f"{stock_code}: {price}원")
event_handler.register(EventType.PRICE_UPDATE, handle_price_update)
# 글로벌 핸들러 (모든 이벤트 수신)
async def global_handler(event_type, event_data):
print(f"[이벤트] {event_type.value}: {event_data}")
event_handler.register_global(global_handler)
# 핸들러 해제
event_handler.unregister(EventType.PRICE_UPDATE, handle_price_update)
```
### 3. 장 스케줄러
```python
from app.realtime import market_scheduler, MarketStatus
# 현재 장 상태 확인
status = market_scheduler.get_current_status()
print(f"현재 장 상태: {status}")
# 장 개장 여부
if market_scheduler.is_market_open():
print("장 개장 중")
# 다음 장 시작까지 남은 시간
seconds = market_scheduler.get_seconds_until_market_open()
print(f"장 시작까지 {seconds}초 남음")
# 콜백 등록
async def on_market_open():
print("🔔 장 시작!")
# WebSocket 연결, 구독 등
async def on_market_close():
print("🔕 장 종료!")
# WebSocket 연결 해제
market_scheduler.register_callbacks(
on_open=on_market_open,
on_close=on_market_close
)
# 스케줄러 실행
await market_scheduler.run()
```
### 4. 실시간 전략 실행
```python
from app.realtime import event_handler, EventType
from app.strategy import strategy_manager
from app.trading import order_manager, market_data_provider
# 핸들러: 실시간 시세 → 전략 실행 → 주문
async def on_price_update(event_data):
stock_code = event_data['stock_code']
current_price = event_data['current_price']
# 시장 데이터 구성
market_data = {
"current_price": current_price,
"minute_chart": await market_data_provider.get_minute_chart(stock_code),
}
# 전략 실행
signal = await strategy_manager.run_strategy(
"DayTrading",
stock_code=stock_code,
market_data=market_data
)
# 신호 발생 시 주문
if signal:
await order_manager.place_order(
stock_code=signal.stock_code,
order_type=signal.signal_type.value,
quantity=10,
price=signal.price
)
event_handler.register(EventType.PRICE_UPDATE, on_price_update)
```
### 5. 전체 시스템 통합
```python
import asyncio
from app.realtime import event_handler, EventType, market_scheduler
from app.realtime.websocket_client import create_websocket_client
from app.strategy import strategy_manager
from app.strategy.strategies import DayTradingStrategy
from app.trading import order_manager, market_data_provider
# 1. 전략 등록
strategy = DayTradingStrategy()
strategy_manager.register_strategy(strategy)
# 2. 이벤트 핸들러
async def on_price_update(event_data):
# ... 전략 실행 로직
pass
event_handler.register(EventType.PRICE_UPDATE, on_price_update)
# 3. 장 스케줄러
ws_client = None
async def on_market_open():
global ws_client
ws_client = create_websocket_client()
await ws_client.connect()
# 관심 종목 구독
for code in ["005930", "000660"]:
await ws_client.subscribe_price(code)
asyncio.create_task(ws_client.listen())
async def on_market_close():
if ws_client:
await ws_client.disconnect()
market_scheduler.register_callbacks(
on_open=on_market_open,
on_close=on_market_close
)
# 4. 실행
await market_scheduler.run()
```
## 주요 기능
### KISWebSocketClient
- ✅ KIS API WebSocket 연결
- ✅ 실시간 현재가/호가 수신
- ✅ 체결 데이터 수신
- ✅ 종목 구독/구독해제
- ✅ 자동 재연결
- ✅ 이벤트 기반 데이터 전달
### EventHandler
- ✅ 7가지 이벤트 타입
- `PRICE_UPDATE`: 현재가 업데이트
- `ORDERBOOK_UPDATE`: 호가 업데이트
- `EXECUTION`: 체결 발생
- `ORDER_CONFIRM`: 주문 확인
- `ERROR`: 에러
- `CONNECTED`: 연결
- `DISCONNECTED`: 연결 해제
- ✅ 타입별 핸들러 등록
- ✅ 글로벌 핸들러 (모든 이벤트)
- ✅ 비동기 핸들러 실행
- ✅ 에러 핸들링
### MarketScheduler
- ✅ 4가지 장 상태
- `PRE_MARKET`: 장 전 (08:00~09:00)
- `MARKET_OPEN`: 정규장 (09:00~15:30)
- `AFTER_MARKET`: 장 후 (15:30~18:00)
- `MARKET_CLOSED`: 휴장
- ✅ 장 시작/종료 콜백
- ✅ 남은 시간 계산
- ✅ 자동 상태 감지
## 이벤트 데이터 구조
### PRICE_UPDATE
```python
{
"stock_code": "005930",
"stock_name": "삼성전자",
"current_price": 75000,
"change_rate": 1.5,
"volume": 1000000,
"timestamp": datetime(...)
}
```
### EXECUTION
```python
{
"order_id": "0000123456",
"stock_code": "005930",
"executed_price": 75000,
"executed_quantity": 10,
"timestamp": datetime(...)
}
```
## 설치 요구사항
```bash
pip install websockets
```
## 사용 예시 파일
상세한 예시는 `examples.py` 참고:
- 기본 실시간 시세 수신
- 실시간 전략 실행
- 장 스케줄러 활용
- 전체 시스템 통합
## Phase 3 완료 체크리스트
- [x] `realtime/websocket_client.py` 생성
- [x] `realtime/event_handler.py` 생성
- [x] `realtime/market_scheduler.py` 생성
- [x] `realtime/examples.py` 생성
- [x] README 문서 작성
## 다음 단계
- [ ] 기존 WebSocket 코드 마이그레이션
- [ ] API 엔드포인트 통합
- [ ] 프론트엔드 WebSocket 연결
---
**작성일**: 2026-03-06
**작성자**: 서영우

26
backend/app/realtime/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
"""
Realtime Module
WebSocket 기반 실시간 시세 수신 및 이벤트 처리
- KIS WebSocket 연결 관리
- 실시간 호가/체결 수신
- 이벤트 핸들러 등록/실행
- 장 스케줄 관리
"""
from app.realtime.websocket_client import KISWebSocketClient
from app.realtime.event_handler import EventHandler, EventType
from app.realtime.market_scheduler import MarketScheduler, MarketStatus
__all__ = [
# WebSocket
"KISWebSocketClient",
# Event
"EventHandler",
"EventType",
# Scheduler
"MarketScheduler",
"MarketStatus",
]

128
backend/app/realtime/event_handler.py Normal file
View File

@@ -0,0 +1,128 @@
"""
Event Handler
실시간 시세 이벤트 핸들러 관리
"""
from enum import Enum
from typing import Dict, List, Callable, Any, Awaitable
import logging
import asyncio
logger = logging.getLogger(__name__)
class EventType(Enum):
"""이벤트 타입"""
PRICE_UPDATE = "PRICE_UPDATE" # 현재가 업데이트
ORDERBOOK_UPDATE = "ORDERBOOK_UPDATE" # 호가 업데이트
EXECUTION = "EXECUTION" # 체결 발생
ORDER_CONFIRM = "ORDER_CONFIRM" # 주문 체결 확인
ERROR = "ERROR" # 에러 발생
CONNECTED = "CONNECTED" # 연결 성공
DISCONNECTED = "DISCONNECTED" # 연결 해제
class EventHandler:
"""
실시간 이벤트 핸들러 관리자
이벤트 타입별로 핸들러 함수를 등록하고 실행
"""
def __init__(self):
self.handlers: Dict[EventType, List[Callable]] = {
event_type: [] for event_type in EventType
}
self.global_handlers: List[Callable] = [] # 모든 이벤트 수신
logger.info("📡 EventHandler 초기화")
def register(
self,
event_type: EventType,
handler: Callable[[Dict[str, Any]], Awaitable[None]]
):
"""
이벤트 핸들러 등록
Args:
event_type: 이벤트 타입
handler: 비동기 핸들러 함수 (event_data를 받아 처리)
"""
if handler not in self.handlers[event_type]:
self.handlers[event_type].append(handler)
logger.info(f"✅ 핸들러 등록: {event_type.value} -> {handler.__name__}")
def unregister(self, event_type: EventType, handler: Callable):
"""핸들러 등록 해제"""
if handler in self.handlers[event_type]:
self.handlers[event_type].remove(handler)
logger.info(f"❌ 핸들러 해제: {event_type.value} -> {handler.__name__}")
def register_global(self, handler: Callable[[EventType, Dict[str, Any]], Awaitable[None]]):
"""모든 이벤트를 수신하는 글로벌 핸들러 등록"""
if handler not in self.global_handlers:
self.global_handlers.append(handler)
logger.info(f"🌍 글로벌 핸들러 등록: {handler.__name__}")
async def emit(
self,
event_type: EventType,
event_data: Dict[str, Any]
):
"""
이벤트 발생 및 핸들러 실행
Args:
event_type: 이벤트 타입
event_data: 이벤트 데이터
"""
logger.debug(f"📤 이벤트 발생: {event_type.value}")
# 타입별 핸들러 실행
handlers = self.handlers.get(event_type, [])
tasks = []
for handler in handlers:
try:
task = asyncio.create_task(handler(event_data))
tasks.append(task)
except Exception as e:
logger.error(f"❌ 핸들러 실행 오류 [{handler.__name__}]: {e}", exc_info=True)
# 글로벌 핸들러 실행
for global_handler in self.global_handlers:
try:
task = asyncio.create_task(global_handler(event_type, event_data))
tasks.append(task)
except Exception as e:
logger.error(f"❌ 글로벌 핸들러 오류: {e}", exc_info=True)
# 모든 핸들러 완료 대기 (non-blocking)
if tasks:
await asyncio.gather(*tasks, return_exceptions=True)
def clear(self, event_type: Optional[EventType] = None):
"""핸들러 초기화"""
if event_type:
self.handlers[event_type].clear()
logger.info(f"🧹 핸들러 초기화: {event_type.value}")
else:
for handlers in self.handlers.values():
handlers.clear()
self.global_handlers.clear()
logger.info("🧹 전체 핸들러 초기화")
def get_stats(self) -> Dict[str, Any]:
"""핸들러 통계"""
return {
"handlers": {
event_type.value: len(handlers)
for event_type, handlers in self.handlers.items()
},
"global_handlers": len(self.global_handlers)
}
# 싱글톤
event_handler = EventHandler()

238
backend/app/realtime/examples.py Normal file
View File

@@ -0,0 +1,238 @@
"""
Realtime Module Usage Examples
Realtime 모듈 사용 예시
"""
import asyncio
import logging
from typing import Dict, Any
from app.realtime import KISWebSocketClient, event_handler, EventType, market_scheduler
from app.realtime.websocket_client import create_websocket_client
from app.strategy import strategy_manager
from app.trading import order_manager, market_data_provider
logger = logging.getLogger(__name__)
# ── 예시 1: 기본 실시간 시세 수신 ────────────────────────
async def example_basic_price_stream():
"""기본 실시간 시세 수신 예시"""
# 핸들러 등록
async def on_price_update(event_data: Dict[str, Any]):
stock_code = event_data['stock_code']
price = event_data['current_price']
print(f"💰 {stock_code}: {price}원")
event_handler.register(EventType.PRICE_UPDATE, on_price_update)
# WebSocket 연결
ws_client = create_websocket_client()
await ws_client.connect()
# 종목 구독
await ws_client.subscribe_price("005930") # 삼성전자
await ws_client.subscribe_price("000660") # SK하이닉스
# 메시지 수신 (무한 루프)
await ws_client.listen()
# ── 예시 2: 실시간 전략 실행 ───────────────────────────
async def example_realtime_strategy():
"""실시간 시세로 전략 실행"""
from app.strategy.strategies import DayTradingStrategy
# 전략 등록
strategy = DayTradingStrategy()
strategy_manager.register_strategy(strategy)
# 실시간 시세 수신 시 전략 실행
async def on_price_update(event_data: Dict[str, Any]):
stock_code = event_data['stock_code']
current_price = event_data['current_price']
# 시장 데이터 구성
market_data = {
"current_price": current_price,
"minute_chart": await market_data_provider.get_minute_chart(stock_code),
}
# 전략 실행
signal = await strategy_manager.run_strategy(
"DayTrading",
stock_code=stock_code,
market_data=market_data
)
# 신호 발생 시 주문
if signal:
logger.info(f"📡 신호 발생: {signal}")
result = await order_manager.place_order(
stock_code=signal.stock_code,
order_type=signal.signal_type.value,
quantity=10,
price=signal.price
)
logger.info(f"📤 주문 결과: {result}")
event_handler.register(EventType.PRICE_UPDATE, on_price_update)
# WebSocket 실행
ws_client = create_websocket_client()
await ws_client.run() # 자동 재연결 포함
# ── 예시 3: 장 스케줄러 활용 ───────────────────────────
async def example_market_scheduler():
"""장 스케줄러 활용 예시"""
ws_client = None
# 장 시작 콜백
async def on_market_open():
nonlocal ws_client
logger.info("🔔 장 시작 - WebSocket 연결 및 구독")
ws_client = create_websocket_client()
await ws_client.connect()
# 관심 종목 구독
watch_list = ["005930", "000660", "035420"]
for stock_code in watch_list:
await ws_client.subscribe_price(stock_code)
# 수신 시작
asyncio.create_task(ws_client.listen())
# 장 종료 콜백
async def on_market_close():
logger.info("🔕 장 종료 - WebSocket 연결 해제")
if ws_client:
await ws_client.disconnect()
# 콜백 등록
market_scheduler.register_callbacks(
on_open=on_market_open,
on_close=on_market_close
)
# 스케줄러 실행
await market_scheduler.run()
# ── 예시 4: 복합 예시 (전체 시스템) ──────────────────────
async def example_full_system():
"""전체 자동매매 시스템 예시"""
from app.strategy.strategies import DayTradingStrategy
# 1. 전략 등록
strategy = DayTradingStrategy(
config={
"max_positions": 3,
"stop_loss_pct": 0.025
}
)
strategy_manager.register_strategy(strategy, priority=10)
# 2. 이벤트 핸들러 등록
async def on_price_update(event_data: Dict[str, Any]):
"""실시간 시세 → 전략 실행 → 주문"""
stock_code = event_data['stock_code']
current_price = event_data['current_price']
# 시장 데이터 수집
market_data = {
"current_price": current_price,
"minute_chart": await market_data_provider.get_minute_chart(stock_code),
}
# 포트폴리오 데이터
balance = await order_manager.get_balance()
# ... 보유 종목 파싱
# 전략 실행
signal = await strategy_manager.run_strategy(
"DayTrading",
stock_code=stock_code,
market_data=market_data
)
if signal:
# 주문 실행
from app.strategy.signal import SignalType
if signal.signal_type == SignalType.BUY:
await order_manager.place_order(
stock_code=signal.stock_code,
order_type="BUY",
quantity=10,
price=signal.price
)
elif signal.signal_type in [SignalType.SELL, SignalType.STOP_LOSS, SignalType.TAKE_PROFIT]:
await order_manager.place_order(
stock_code=signal.stock_code,
order_type="SELL",
quantity=signal.quantity,
price=signal.price
)
event_handler.register(EventType.PRICE_UPDATE, on_price_update)
# 3. 장 스케줄러 설정
ws_client = None
async def on_market_open():
nonlocal ws_client
logger.info("🔔 장 시작")
# WebSocket 연결
ws_client = create_websocket_client()
await ws_client.connect()
# 관심 종목 구독
watch_list = ["005930", "000660", "035420", "005380"]
for code in watch_list:
await ws_client.subscribe_price(code)
# 수신 시작
asyncio.create_task(ws_client.listen())
async def on_market_close():
logger.info("🔕 장 종료")
# WebSocket 연결 해제
if ws_client:
await ws_client.disconnect()
market_scheduler.register_callbacks(
on_open=on_market_open,
on_close=on_market_close
)
# 4. 시스템 실행
logger.info("🚀 자동매매 시스템 시작")
await market_scheduler.run()
# ── 메인 함수 ─────────────────────────────────────────
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
# 예시 선택 실행
# asyncio.run(example_basic_price_stream())
# asyncio.run(example_realtime_strategy())
# asyncio.run(example_market_scheduler())
asyncio.run(example_full_system())

145
backend/app/realtime/market_scheduler.py Normal file
View File

@@ -0,0 +1,145 @@
"""
Market Scheduler
장 시간 관리 및 스케줄링
"""
from enum import Enum
from datetime import datetime, time
from typing import Optional, Callable, Awaitable
import logging
import asyncio
import pytz
logger = logging.getLogger(__name__)
class MarketStatus(Enum):
"""장 상태"""
PRE_MARKET = "PRE_MARKET" # 장 전 (08:00~09:00)
MARKET_OPEN = "MARKET_OPEN" # 정규장 (09:00~15:30)
AFTER_MARKET = "AFTER_MARKET" # 장 후 (15:30~18:00)
MARKET_CLOSED = "MARKET_CLOSED" # 휴장 (18:00~08:00)
class MarketScheduler:
"""
장 시간 관리 및 스케줄 실행
- 현재 장 상태 파악
- 장 시작/종료 콜백 실행
- 다음 장 시작까지 대기
"""
# 한국 시간 기준
PRE_MARKET_START = time(8, 0)
MARKET_OPEN = time(9, 0)
MARKET_CLOSE = time(15, 30)
AFTER_MARKET_END = time(18, 0)
def __init__(self):
self.kst = pytz.timezone("Asia/Seoul")
self.current_status: Optional[MarketStatus] = None
self.on_market_open: Optional[Callable[[], Awaitable[None]]] = None
self.on_market_close: Optional[Callable[[], Awaitable[None]]] = None
logger.info("📅 MarketScheduler 초기화")
def get_current_status(self) -> MarketStatus:
"""현재 장 상태 조회"""
now = datetime.now(self.kst).time()
if self.PRE_MARKET_START <= now < self.MARKET_OPEN:
return MarketStatus.PRE_MARKET
elif self.MARKET_OPEN <= now < self.MARKET_CLOSE:
return MarketStatus.MARKET_OPEN
elif self.MARKET_CLOSE <= now < self.AFTER_MARKET_END:
return MarketStatus.AFTER_MARKET
else:
return MarketStatus.MARKET_CLOSED
def is_market_open(self) -> bool:
"""현재 정규장 개장 여부"""
return self.get_current_status() == MarketStatus.MARKET_OPEN
def register_callbacks(
self,
on_open: Optional[Callable[[], Awaitable[None]]] = None,
on_close: Optional[Callable[[], Awaitable[None]]] = None
):
"""장 시작/종료 콜백 등록"""
if on_open:
self.on_market_open = on_open
logger.info("✅ 장 시작 콜백 등록")
if on_close:
self.on_market_close = on_close
logger.info("✅ 장 종료 콜백 등록")
async def run(self):
"""
스케줄러 실행 (무한 루프)
장 상태 변화 감지 및 콜백 실행
"""
logger.info("🚀 MarketScheduler 시작")
while True:
try:
status = self.get_current_status()
# 상태 변화 감지
if status != self.current_status:
logger.info(f"📊 장 상태 변경: {self.current_status} → {status}")
# 장 시작
if status == MarketStatus.MARKET_OPEN and self.on_market_open:
logger.info("🔔 장 시작 콜백 실행")
await self.on_market_open()
# 장 종료
elif self.current_status == MarketStatus.MARKET_OPEN and status == MarketStatus.AFTER_MARKET:
if self.on_market_close:
logger.info("🔕 장 종료 콜백 실행")
await self.on_market_close()
self.current_status = status
# 1분마다 상태 체크
await asyncio.sleep(60)
except Exception as e:
logger.error(f"❌ 스케줄러 오류: {e}", exc_info=True)
await asyncio.sleep(60)
def get_seconds_until_market_open(self) -> Optional[int]:
"""다음 장 시작까지 남은 시간 (초)"""
now = datetime.now(self.kst)
today = now.date()
# 오늘 장 시작 시간
market_open_today = datetime.combine(today, self.MARKET_OPEN, tzinfo=self.kst)
if now.time() < self.MARKET_OPEN:
# 오늘 장 시작 전
return int((market_open_today - now).total_seconds())
else:
# 내일 장 시작 대기
from datetime import timedelta
market_open_tomorrow = market_open_today + timedelta(days=1)
return int((market_open_tomorrow - now).total_seconds())
def get_seconds_until_market_close(self) -> Optional[int]:
"""장 종료까지 남은 시간 (초)"""
now = datetime.now(self.kst)
today = now.date()
market_close_today = datetime.combine(today, self.MARKET_CLOSE, tzinfo=self.kst)
if now.time() < self.MARKET_CLOSE:
return int((market_close_today - now).total_seconds())
else:
return 0
# 싱글톤
market_scheduler = MarketScheduler()

244
backend/app/realtime/websocket_client.py Normal file
View File

@@ -0,0 +1,244 @@
"""
KIS WebSocket Client
KIS API WebSocket 연결 및 실시간 데이터 수신
"""
import json
import logging
import asyncio
from typing import Dict, Any, Optional, Set, Callable
from datetime import datetime
import pytz
try:
import websockets
from websockets.client import WebSocketClientProtocol
except ImportError:
websockets = None
WebSocketClientProtocol = None
from app.core.config import settings
from app.realtime.event_handler import event_handler, EventType
logger = logging.getLogger(__name__)
class KISWebSocketClient:
"""
KIS API WebSocket 클라이언트
실시간 호가/체결 데이터 수신
"""
def __init__(self, app_key: str, app_secret: str):
if websockets is None:
raise ImportError("websockets 패키지가 설치되지 않았습니다: pip install websockets")
self.app_key = app_key
self.app_secret = app_secret
self.ws: Optional[WebSocketClientProtocol] = None
self.subscriptions: Set[str] = set() # 구독 중인 종목코드
self.is_connected = False
self.kst = pytz.timezone("Asia/Seoul")
# WebSocket URL (모의투자 vs 실계좌)
if "vts" in settings.KOREA_INVESTMENT_API_BASE_URL.lower():
self.ws_url = "ws://ops.koreainvestment.com:31000" # 모의투자
else:
self.ws_url = "ws://ops.koreainvestment.com:21000" # 실계좌
logger.info(f"🔌 WebSocket 클라이언트 초기화: {self.ws_url}")
async def connect(self):
"""
WebSocket 연결"""
try:
logger.info(f"🔌 WebSocket 연결 시도: {self.ws_url}")
self.ws = await websockets.connect(self.ws_url)
self.is_connected = True
logger.info("✅ WebSocket 연결 성공")
await event_handler.emit(EventType.CONNECTED, {"timestamp": datetime.now(self.kst)})
except Exception as e:
logger.error(f"❌ WebSocket 연결 실패: {e}", exc_info=True)
self.is_connected = False
raise
async def disconnect(self):
"""
WebSocket 연결 해제"""
if self.ws:
await self.ws.close()
self.is_connected = False
logger.info("🔌 WebSocket 연결 해제")
await event_handler.emit(EventType.DISCONNECTED, {"timestamp": datetime.now(self.kst)})
async def subscribe_price(self, stock_code: str):
"""
주식 현재가 구독
Args:
stock_code: 종목코드 (6자리)
"""
if not self.is_connected:
raise ConnectionError("WebSocket이 연결되지 않았습니다")
# KIS WebSocket 프로토콜
subscribe_msg = {
"header": {
"approval_key": self.app_key,
"custtype": "P",
"tr_type": "1", # 등록
"content-type": "utf-8"
},
"body": {
"input": {
"tr_id": "H0STCNT0", # 현재가 호가
"tr_key": stock_code
}
}
}
await self.ws.send(json.dumps(subscribe_msg))
self.subscriptions.add(stock_code)
logger.info(f"📡 구독 등록: {stock_code}")
async def unsubscribe_price(self, stock_code: str):
"""주식 현재가 구독 해제"""
if not self.is_connected:
return
unsubscribe_msg = {
"header": {
"approval_key": self.app_key,
"custtype": "P",
"tr_type": "2", # 해제
"content-type": "utf-8"
},
"body": {
"input": {
"tr_id": "H0STCNT0",
"tr_key": stock_code
}
}
}
await self.ws.send(json.dumps(unsubscribe_msg))
self.subscriptions.discard(stock_code)
logger.info(f"❌ 구독 해제: {stock_code}")
async def listen(self):
"""
메시지 수신 루프 (무한 대기)
수신한 데이터를 파싱하여 이벤트 발생
"""
if not self.is_connected:
raise ConnectionError("WebSocket이 연결되지 않았습니다")
logger.info("👂 WebSocket 메시지 수신 시작")
try:
async for message in self.ws:
try:
await self._handle_message(message)
except Exception as e:
logger.error(f"❌ 메시지 처리 오류: {e}", exc_info=True)
await event_handler.emit(EventType.ERROR, {"error": str(e)})
except websockets.exceptions.ConnectionClosed:
logger.warning("⚠️ WebSocket 연결 끔김")
self.is_connected = False
await event_handler.emit(EventType.DISCONNECTED, {"reason": "connection_closed"})
async def _handle_message(self, message: str):
"""
수신한 메시지 파싱 및 이벤트 발생
Args:
message: WebSocket 메시지 (JSON 문자열)
"""
try:
data = json.loads(message)
except json.JSONDecodeError:
# 텍스트 형식일 수 있음 (KIS API 특성)
logger.debug(f"텍스트 메시지 수신: {message[:100]}...")
return
# TR_ID 확인
tr_id = data.get("header", {}).get("tr_id")
if tr_id == "H0STCNT0": # 현재가 호가
await self._handle_price_update(data)
elif tr_id == "H0STCNI0": # 체결
await self._handle_execution(data)
else:
logger.debug(f"알 수 없는 TR_ID: {tr_id}")
async def _handle_price_update(self, data: Dict[str, Any]):
"""현재가 업데이트 처리"""
body = data.get("body", {})
output = body.get("output", {})
event_data = {
"stock_code": output.get("MKSC_SHRN_ISCD", ""),
"stock_name": output.get("HTS_KOR_ISNM", ""),
"current_price": int(output.get("STCK_PRPR", 0)),
"change_rate": float(output.get("PRDY_CTRT", 0)),
"volume": int(output.get("ACML_VOL", 0)),
"timestamp": datetime.now(self.kst),
}
await event_handler.emit(EventType.PRICE_UPDATE, event_data)
async def _handle_execution(self, data: Dict[str, Any]):
"""체결 데이터 처리"""
body = data.get("body", {})
output = body.get("output", {})
event_data = {
"order_id": output.get("ODNO", ""),
"stock_code": output.get("PDNO", ""),
"executed_price": int(output.get("CNTR_UNPR", 0)),
"executed_quantity": int(output.get("CNTR_QTY", 0)),
"timestamp": datetime.now(self.kst),
}
await event_handler.emit(EventType.EXECUTION, event_data)
def get_subscriptions(self) -> Set[str]:
"""현재 구독 중인 종목 목록"""
return self.subscriptions.copy()
async def run(self):
"""
WebSocket 클라이언트 실행 (연결 + 수신)
자동 재연결 지원
"""
while True:
try:
await self.connect()
await self.listen()
except Exception as e:
logger.error(f"❌ WebSocket 오류: {e}")
await event_handler.emit(EventType.ERROR, {"error": str(e)})
# 5초 후 재연결 시도
logger.info("🔄 5초 후 재연결 시도...")
await asyncio.sleep(5)
# 클라이언트 인스턴스 생성 헬퍼
def create_websocket_client() -> KISWebSocketClient:
"""
KIS WebSocket 클라이언트 인스턴스 생성"""
return KISWebSocketClient(
app_key=settings.KOREA_INVESTMENT_APP_KEY,
app_secret=settings.KOREA_INVESTMENT_APP_SECRET
)