From 7355df6ebf56c0639f52f8ff430de38063093902 Mon Sep 17 00:00:00 2001 From: YoungWooSeo Date: Fri, 6 Mar 2026 10:57:40 +0900 Subject: [PATCH] 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()