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())