From 7355df6ebf56c0639f52f8ff430de38063093902 Mon Sep 17 00:00:00 2001 From: YoungWooSeo Date: Fri, 6 Mar 2026 10:57:40 +0900 Subject: [PATCH 1/3] refactor(phase3): Create realtime module core structure --- backend/app/realtime/__init__.py | 26 ++++ backend/app/realtime/event_handler.py | 128 ++++++++++++++++++++ backend/app/realtime/market_scheduler.py | 145 +++++++++++++++++++++++ 3 files changed, 299 insertions(+) create mode 100644 backend/app/realtime/__init__.py create mode 100644 backend/app/realtime/event_handler.py create mode 100644 backend/app/realtime/market_scheduler.py diff --git a/backend/app/realtime/__init__.py b/backend/app/realtime/__init__.py new file mode 100644 index 0000000..a530ae7 --- /dev/null +++ b/backend/app/realtime/__init__.py @@ -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", +] diff --git a/backend/app/realtime/event_handler.py b/backend/app/realtime/event_handler.py new file mode 100644 index 0000000..3601429 --- /dev/null +++ b/backend/app/realtime/event_handler.py @@ -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() diff --git a/backend/app/realtime/market_scheduler.py b/backend/app/realtime/market_scheduler.py new file mode 100644 index 0000000..b844f7f --- /dev/null +++ b/backend/app/realtime/market_scheduler.py @@ -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() From b1eab5305a06e33449dda85dd9c84c1bdda1364a Mon Sep 17 00:00:00 2001 From: YoungWooSeo Date: Fri, 6 Mar 2026 10:58:28 +0900 Subject: [PATCH 2/3] refactor(phase3): Add KIS WebSocket client implementation --- backend/app/realtime/websocket_client.py | 244 +++++++++++++++++++++++ 1 file changed, 244 insertions(+) create mode 100644 backend/app/realtime/websocket_client.py diff --git a/backend/app/realtime/websocket_client.py b/backend/app/realtime/websocket_client.py new file mode 100644 index 0000000..e9011eb --- /dev/null +++ b/backend/app/realtime/websocket_client.py @@ -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 + ) From 48aa5440ec653b932eb6991c0ddebe37bdf80d2e Mon Sep 17 00:00:00 2001 From: YoungWooSeo Date: Fri, 6 Mar 2026 11:00:03 +0900 Subject: [PATCH 3/3] refactor(phase3): Add realtime module examples and README --- backend/app/realtime/README.md | 275 +++++++++++++++++++++++++++++++ backend/app/realtime/examples.py | 238 ++++++++++++++++++++++++++ 2 files changed, 513 insertions(+) create mode 100644 backend/app/realtime/README.md create mode 100644 backend/app/realtime/examples.py diff --git a/backend/app/realtime/README.md b/backend/app/realtime/README.md new file mode 100644 index 0000000..b0c55dc --- /dev/null +++ b/backend/app/realtime/README.md @@ -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 +**작성자**: 서영우 diff --git a/backend/app/realtime/examples.py b/backend/app/realtime/examples.py new file mode 100644 index 0000000..0900e73 --- /dev/null +++ b/backend/app/realtime/examples.py @@ -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())