""" FastAPI 애플리케이션 진입점 - CORS, 미들웨어 설정 - WebSocket 실시간 시세 브로드캐스트 - 라우터 등록 및 startup/shutdown 이벤트 관리 """ from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request from fastapi.middleware.cors import CORSMiddleware from contextlib import asynccontextmanager import contextlib import asyncio import json import logging import time import pytz from app.core.config import settings from app.api import api_router from app.services.websocket_manager import websocket_manager from app.core.json_helpers import _dumps from app.services.db_worker import db_writer_worker from app.core.database import engine, Base, AsyncSessionLocal from app.models import models # 모델 임포트 from app.core.market_schedule import is_market_open from sqlalchemy import text # 로깅 설정 import os import logging.handlers _log_level = getattr(logging, settings.LOG_LEVEL.upper(), logging.INFO) _log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' _log_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "logs") os.makedirs(_log_dir, exist_ok=True) _log_file = os.path.join(_log_dir, "backend.log") logging.basicConfig( level=_log_level, format=_log_format, handlers=[ logging.StreamHandler(), logging.handlers.RotatingFileHandler( _log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8" ), ] ) logger = logging.getLogger(__name__) def _normalize_recommendations(rec: dict) -> dict: """사로 다른 포맷의 추청 결과를 recommendations_by_rule 포맷으로 정규화. - get_rule_based_recommendations_realtime() → {"recommendations": [...], ...} - get_rule_based_recommendations_from_db() → {"recommendations_by_rule": {...}, ...} 둘 다 recommendations_by_rule 키를 보장한 상태로 돌려준다. Redis 캐시 데이터와 DB 데이터 모두 이 함수를 통해 정규화. """ if not rec: return rec # 이미 recommendations_by_rule 키가 있으면 그대로 사용 if rec.get("recommendations_by_rule"): return rec # recommendations 리스트를 rule 기준으로 그룹화 raw_list = rec.get("recommendations", []) if not raw_list: return rec by_rule: dict = {} for item in raw_list: rule = ( item.get("rule_name") or item.get("signal") or item.get("strategy") or "general" ) by_rule.setdefault(rule, []).append(item) return {**rec, "recommendations_by_rule": by_rule} async def batch_analysis_scheduler(): """10분 간격 배치 분석 스케줄러. 장중 10분마다 KIS API로 종목 분석 후 stock_analysis DB에 upsert. 분석 완료 후 Redis에 recommendations:latest 캐시를 저장해 auto_trade_scheduler가 DB 조회 없이 바로 매매 판단에 활용할 수 있도록 한다. """ await asyncio.sleep(15) # 부팅 안정화 (auto_trade_scheduler보다 늦게 시작) from app.api.endpoints.recommendations import get_rule_based_recommendations_realtime from app.core.redis import setex_json, TTL_RECOMMENDATIONS, TTL_ANALYSIS interval = int(getattr(settings, "BATCH_ANALYSIS_INTERVAL", 600)) # 기본 10분 logger.info(f"📊 Batch analysis scheduler started (every {interval}s)") try: while True: if not is_market_open(): await asyncio.sleep(1800) continue try: logger.info("🔄 [Batch] 10-min analysis cycle starting...") result = await get_rule_based_recommendations_realtime() count = len(result.get("recommendations", [])) if result else 0 logger.info(f"✅ [Batch] Done: {count} stocks saved/updated in stock_analysis DB") # ── Redis 캐시 업데이트 ───────────────────────────────────────── if result: # 정규화: recommendations_by_rule 키 보장 (auto_trade_scheduler 호환) normalized = _normalize_recommendations(result) ok = await setex_json( "recommendations:latest", TTL_RECOMMENDATIONS, normalized ) if ok: logger.info( f"💾 [Batch] Redis 캐시 갱신 완료: recommendations:latest " f"(rule 수: {len(normalized.get('recommendations_by_rule', {}))}, " f"TTL: {TTL_RECOMMENDATIONS}s)" ) # ===== 개별 종목별 캐시 (병렬 처리 적용) ===== raw_list = result.get("recommendations", []) cache_tasks = [] for item in raw_list: code = item.get("stock_code") or item.get("code") if code: cache_tasks.append(setex_json(f"analysis:{code}", TTL_ANALYSIS, item)) if cache_tasks: await asyncio.gather(*cache_tasks, return_exceptions=True) logger.info(f"💾 [Batch] 개별 종목 분석 데이터 캐시 갱신 완료: {len(cache_tasks)}건 (TTL: {TTL_ANALYSIS}s)") # =============================================== # ─────────────────────────────────────────────────────── except Exception as e: logger.error(f"❌ Batch analysis error: {e}", exc_info=True) await asyncio.sleep(interval) except asyncio.CancelledError: logger.info("⏹️ Batch analysis scheduler stopped") raise async def auto_trade_scheduler(): """1분 간격 자동매매 스케줄러. Redis 캐시(recommendations:latest)를 먼저 확인 → 캐시 히트 시 DB 조회 없이 매매 실행. 케시 미스(30분 TTL 만료) 시만 get_rule_based_recommendations_from_db() 호출. 주문 직전 MCP RISK_ASSESSMENT 호출 → 승인 시만 주문. """ await asyncio.sleep(2) from app.api.endpoints.recommendations import ( get_rule_based_recommendations_from_db, execute_auto_trade_background, ) from app.api.endpoints.portfolio import check_and_execute_exit_trades from app.api.endpoints.mcp import request_mcp_risk_assessment from app.core.redis import get_json logger.info("⏱️ Auto-trade scheduler started (every 60s, Redis→DB fallback mode)") try: while True: try: if not is_market_open(): logger.info("💤 Market is closed. Next check in 30 minutes.") await asyncio.sleep(1800) continue async with AsyncSessionLocal() as session: # 1. 보유 종목 손절/익절 체크 logger.info("💰 Checking portfolio for stop-loss/take-profit...") try: await check_and_execute_exit_trades(session) except Exception as e: logger.error(f"Exit trade check error: {e}", exc_info=True) # 2. 자동매매: Redis 캐시 우선 → cache miss 시만 DB 조회 if settings.AUTO_TRADE_ENABLED: # ── Redis 캐시 확인 (10분 배치 결과, TTL 30분) ──────────────── rec = await get_json("recommendations:latest") if rec: logger.info( "⚡ Redis cache hit: recommendations:latest " f"(rule 수: {len(rec.get('recommendations_by_rule', {}))}) " "→ DB 조회 생략" ) # 포맷 정규화 (다른 키 조합 대응) rec = _normalize_recommendations(rec) else: logger.info("🔄 Redis cache miss → DB에서 추천 조회") rec = await get_rule_based_recommendations_from_db(session) # ────────────────────────────────────────────────────── if rec and rec.get("recommendations_by_rule"): # ── MCP RISK_ASSESSMENT (공통 함수, 이슈 4) ────────── recommendations = rec.get("recommendations_by_rule", {}) trade_plan = { "action": "BUY", "strategy": "rule_based_batch", "recommendation_count": sum( len(v) for v in recommendations.values() ), "rules": list(recommendations.keys()), "top_stocks": [ item.get("stock_code") or item.get("code", "") for rule_items in recommendations.values() for item in ( rule_items[:3] if isinstance(rule_items, list) else [] ) ][:10], } portfolio: dict = {} signals = { "source": "db_recommendations", "rules_triggered": list(recommendations.keys()), } approve, reason = await request_mcp_risk_assessment( trade_plan, portfolio, signals ) # ───────────────────────────────────────────────────── if not approve: logger.warning( f"🛑 MCP RISK_ASSESSMENT DENY → 주문 스킵. 사유: {reason}" ) try: from app.services.db_worker import enqueue_db_write await enqueue_db_write({ "action": "log_mcp_deny", "reason": reason, }) except Exception: pass else: if reason: logger.info( f"✅ MCP RISK_ASSESSMENT APPROVE. 참고: {reason}" ) logger.info("📊 Found recommendations, executing auto-trade") await execute_auto_trade_background(rec, session) else: logger.info("⏭️ Auto-trade skipped: no recommendations") else: logger.info("⏸️ Auto-trade disabled - only monitoring stop-loss/take-profit") except Exception as e: logger.error(f"❌ Auto-trade scheduler error: {e}", exc_info=True) await asyncio.sleep(60) except asyncio.CancelledError: logger.info("⏹️ Auto-trade scheduler stopped") raise async def day_trade_scheduler(): """30초 간격 데이트레이딩 스케줄러. - Opening Range 수집 (09:00~09:30) - 진입 신호 스캔 + 주문 실행 (09:30~15:00) - EOD 강제 청산 (15:20~15:30) - 장 외 시간: 30분 슬립 """ await asyncio.sleep(5) from app.api.endpoints.day_trading import run_day_trading_cycle logger.info("📈 Day-trade scheduler started (every 30s)") try: while True: if not is_market_open(): await asyncio.sleep(1800) continue if not settings.DAY_TRADE_ENABLED: await asyncio.sleep(60) continue try: result = await run_day_trading_cycle() phase = result.get("phase", "") if result.get("entries"): logger.info(f"📈 데이트레이딩 진입: {result['entries']}") if result.get("exits"): logger.info(f"📉 데이트레이딩 청산: {result['exits']}") if phase not in ("market_closed", "outside_window"): logger.debug(f"Day-trade cycle [{phase}]: {result.get('time')}") except Exception as e: logger.error(f"❌ Day-trade scheduler error: {e}", exc_info=True) await asyncio.sleep(30) except asyncio.CancelledError: logger.info("⏹️ Day-trade scheduler stopped") raise @asynccontextmanager async def lifespan(app: FastAPI): """애플리케이션 시작/종료 시 실행될 코드""" logger.info("Starting Auto Stock Trading System...") logger.info(f"KIS API Mode: {'모의투자 (VIRTUAL)' if settings.is_virtual_trading else '실전투자 (REAL)'}") logger.info(f"KIS Base URL: {settings.KOREA_INVESTMENT_API_BASE_URL}") logger.info(f"KIS Account: {settings.KOREA_INVESTMENT_ACCOUNT_NUMBER}") # DB 테이블 자동 생성 try: logger.info(f"Connecting to database: {settings.DATABASE_URL.split('@')[1] if '@' in settings.DATABASE_URL else 'unknown'}") async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) await conn.execute(text(""" ALTER TABLE trade_history ADD COLUMN IF NOT EXISTS total_amount NUMERIC(15, 2), ADD COLUMN IF NOT EXISTS entry_reason TEXT, ADD COLUMN IF NOT EXISTS exit_reason TEXT, ADD COLUMN IF NOT EXISTS signals JSONB, ADD COLUMN IF NOT EXISTS entry_trade_id INTEGER, ADD COLUMN IF NOT EXISTS slippage NUMERIC(10, 2), ADD COLUMN IF NOT EXISTS commission NUMERIC(10, 2), ADD COLUMN IF NOT EXISTS net_profit_loss NUMERIC(15, 2), ADD COLUMN IF NOT EXISTS profit_loss_rate NUMERIC(10, 4); """)) await conn.execute(text(""" UPDATE trade_history SET total_amount = executed_price * executed_quantity WHERE total_amount IS NULL AND executed_price IS NOT NULL AND executed_quantity IS NOT NULL; """)) await conn.execute(text(""" ALTER TABLE stock_analysis ADD COLUMN IF NOT EXISTS atr_14 NUMERIC(15, 2), ADD COLUMN IF NOT EXISTS adx_14 NUMERIC(10, 2), ADD COLUMN IF NOT EXISTS ml_score INTEGER, ADD COLUMN IF NOT EXISTS ml_confidence NUMERIC(5, 4), ADD COLUMN IF NOT EXISTS ml_signal VARCHAR(10); """)) await conn.execute(text(""" DO $$ BEGIN IF NOT EXISTS ( SELECT 1 FROM pg_constraint WHERE conname = 'uq_stock_analysis_code_date' AND conrelid = 'stock_analysis'::regclass ) THEN ALTER TABLE stock_analysis ADD CONSTRAINT uq_stock_analysis_code_date UNIQUE (stock_code, analysis_date); END IF; END $$; """)) await conn.execute(text(""" ALTER TABLE active_trades ADD COLUMN IF NOT EXISTS remaining_pct NUMERIC(5, 2) DEFAULT 1.0, ADD COLUMN IF NOT EXISTS highest_since_entry NUMERIC(15, 2), ADD COLUMN IF NOT EXISTS strategy_name VARCHAR(100); """)) await conn.execute(text(""" CREATE TABLE IF NOT EXISTS trading_sessions ( id SERIAL PRIMARY KEY, session_date DATE UNIQUE NOT NULL, consecutive_losses INTEGER DEFAULT 0, daily_profit_loss NUMERIC(15, 2) DEFAULT 0, is_paused BOOLEAN DEFAULT FALSE, pause_reason TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """)) await conn.execute(text(""" CREATE TABLE IF NOT EXISTS stock_bans ( stock_code VARCHAR(20) PRIMARY KEY, consecutive_losses INTEGER DEFAULT 0 NOT NULL, ban_until TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """)) logger.info("✅ Database tables created successfully (and columns verified)") try: from app.services.loss_tracker import loss_tracker async with AsyncSessionLocal() as session: await loss_tracker.load_from_db(session) except Exception as e: logger.warning(f"⚠️ loss_tracker 초기화 실패 (non-critical): {e}") try: from app.models.models import StockAnalysis, TradeHistory, Portfolio from sqlalchemy import delete from datetime import datetime, timedelta async with AsyncSessionLocal() as session: _kst = pytz.timezone("Asia/Seoul") _today_kst = datetime.now(_kst).date() cutoff_date = _today_kst - timedelta(days=7) result = await session.execute( delete(StockAnalysis).where(StockAnalysis.analysis_date < cutoff_date) ) if result.rowcount > 0: logger.info(f"🗑️ Deleted {result.rowcount} old stock_analysis records") cutoff_datetime = _today_kst - timedelta(days=1) result = await session.execute( delete(TradeHistory).where( TradeHistory.status.in_(['PENDING', 'FAILED']), TradeHistory.created_at < cutoff_datetime, ) ) if result.rowcount > 0: logger.info(f"🗑️ Deleted {result.rowcount} old PENDING/FAILED trades") result = await session.execute( delete(Portfolio).where(Portfolio.quantity <= 0) ) if result.rowcount > 0: logger.info(f"🗑️ Deleted {result.rowcount} empty portfolio records") await session.commit() logger.info("✅ Test data cleanup completed") except Exception as e: logger.warning(f"⚠️ Test data cleanup failed (non-critical): {e}") except ConnectionRefusedError as e: logger.error(f"❌ Database connection refused: {e}") raise except Exception as e: logger.error(f"❌ Failed to create database tables: {e}") import traceback logger.error(traceback.format_exc()) raise try: from app.api.endpoints.day_trading import restore_positions_from_db await restore_positions_from_db() except Exception as e: logger.warning(f"⚠️ 데이트레이딩 포지션 복구 실패 (non-critical): {e}") # ── httpx 싱글톤 클라이언트 초기화 ────────────────────────────────────── from app.api.endpoints.mcp import init_http_client, close_http_client await init_http_client() # ── Redis 클라이언트 초기화 ─────────────────────────────────────────────── from app.core.redis import init_redis, close_redis await init_redis() # 워커 + 스케줄러 시작 app.state.db_worker_task = asyncio.create_task(db_writer_worker()) app.state.auto_trade_task = asyncio.create_task(auto_trade_scheduler()) app.state.batch_analysis_task = asyncio.create_task(batch_analysis_scheduler()) app.state.day_trade_task = asyncio.create_task(day_trade_scheduler()) # KIS 실시간 스트림 시작 from app.services.kis_realtime_stream import run_kis_realtime_stream app.state.kis_stream_task = asyncio.create_task(run_kis_realtime_stream()) logger.info("📡 KIS 실시간 스트림 태스크 시작") yield # 태스크 정리 for task_attr in ( "auto_trade_task", "batch_analysis_task", "day_trade_task", "db_worker_task", "kis_stream_task", ): try: task = getattr(app.state, task_attr, None) if task: task.cancel() with contextlib.suppress(asyncio.CancelledError): await task except Exception as e: logger.warning(f"Failed to stop {task_attr} gracefully: {e}") # ── httpx 클라이언트 종료 ───────────────────────────────────────────────── try: await close_http_client() except Exception as e: logger.warning(f"Failed to close http_client: {e}") # ── Redis 클라이언트 종료 ───────────────────────────────────────────────── try: await close_redis() except Exception as e: logger.warning(f"Failed to close Redis: {e}") logger.info("Shutting down Auto Stock Trading System...") await engine.dispose() app = FastAPI( title=settings.APP_NAME, description="Claude AI와 MCP를 활용한 자동 주식 매매 시스템", version="1.0.0", lifespan=lifespan, debug=settings.DEBUG, ) # CORS 설정 _cors_origins = getattr(settings, "CORS_ORIGINS", None) if _cors_origins: _allowed_origins = [o.strip() for o in _cors_origins.split(",") if o.strip()] else: _allowed_origins = [ "http://localhost:5173", "http://localhost:3000", "http://127.0.0.1:5173", "http://127.0.0.1:3000", ] app.add_middleware( CORSMiddleware, allow_origins=_allowed_origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], expose_headers=["*"], ) logger.info("CORS middleware configured") @app.middleware("http") async def log_requests(request: Request, call_next): start_time = time.time() logger.info(f"🔵 Request: {request.method} {request.url}") response = await call_next(request) process_time = time.time() - start_time logger.info( f"🟢 Response: {request.method} {request.url} " f"- Status: {response.status_code} - Time: {process_time:.3f}s" ) return response app.include_router(api_router, prefix="/api/v1") @app.get("/") async def root(): return {"message": "Auto Stock Trading System API", "version": "1.0.0", "status": "running"} @app.get("/health") async def health_check(): return {"status": "healthy"} @app.websocket("/ws/{client_id}") async def websocket_endpoint(websocket: WebSocket, client_id: str): \"\"\"WebSocket 연결 엔드포인트. 클라이언트→서버 지원 프로토콜: {"type": "ping"} → {"type": "pong"} {"type": "subscribe", "stock_codes": [...]} → ACK {"type": "unsubscribe", "stock_codes": [...]} → ACK {"type": "set_subscriptions", "stock_codes": [...]} → ACK {"type": "get_chart", "stock_code": "...", "period": "D", "limit": 200} → {"type": "chart_data", ...} 그 외 메시지 → echo \"\"\" await websocket_manager.connect(websocket, client_id) try: while True: data = await websocket.receive_text() try: msg = json.loads(data) except (json.JSONDecodeError, AttributeError): await websocket_manager.send_personal_message(f"Echo: {data}", client_id) continue msg_type = msg.get("type", "") if msg_type == "ping": await websocket_manager.send_personal_message( json.dumps({"type": "pong"}), client_id ) continue if msg_type == "subscribe": codes = [str(c) for c in msg.get("stock_codes", [])] websocket_manager.subscribe(client_id, codes) await websocket_manager.send_personal_message( json.dumps({ "type": "ack", "action": "subscribe", "stock_codes": codes, "subscriptions": websocket_manager.get_subscriptions(client_id), }), client_id, ) continue if msg_type == "unsubscribe": codes = [str(c) for c in msg.get("stock_codes", [])] websocket_manager.unsubscribe(client_id, codes) await websocket_manager.send_personal_message( json.dumps({ "type": "ack", "action": "unsubscribe", "stock_codes": codes, "subscriptions": websocket_manager.get_subscriptions(client_id), }), client_id, ) continue if msg_type == "set_subscriptions": codes = [str(c) for c in msg.get("stock_codes", [])] websocket_manager.set_subscriptions(client_id, codes) await websocket_manager.send_personal_message( json.dumps({ "type": "ack", "action": "set_subscriptions", "stock_codes": codes, "subscriptions": websocket_manager.get_subscriptions(client_id), }), client_id, ) # 구독 종목 차트 캐시 사전 워밍업 (백그라운드, 논블로킹) if codes: async def _warmup_chart_cache(stock_codes: list): from app.api.endpoints.market_data import get_chart_data_cached, chart_cache from datetime import datetime for code in stock_codes: cached = chart_cache.get(f"{code}__D") if cached and (datetime.now() - cached["timestamp"]).total_seconds() < 600: continue try: await get_chart_data_cached(code, "D", 100) logger.debug(f"[warmup] {code} 차트 캐시 완료") except Exception as e: logger.debug(f"[warmup] {code} 캐시 실패: {e}") asyncio.create_task(_warmup_chart_cache(codes)) continue if msg_type == "get_chart": stock_code = str(msg.get("stock_code", "")).strip() period = msg.get("period", "D") limit = int(msg.get("limit", 200)) if not stock_code: continue try: from app.api.endpoints.market_data import get_chart_data_cached chart_data = await get_chart_data_cached(stock_code, period, limit) await websocket_manager.send_personal_message( _dumps({"type": "chart_data", "stock_code": stock_code, "period": period, "data": chart_data}), client_id, ) except Exception as e: logger.error(f"get_chart 오류 ({stock_code}): {e}") await websocket_manager.send_personal_message( _dumps({"type": "chart_data", "stock_code": stock_code, "period": period, "data": [], "error": str(e)}), client_id, ) continue await websocket_manager.send_personal_message(f"Echo: {data}", client_id) except WebSocketDisconnect: websocket_manager.disconnect(client_id) logger.info(f"Client {client_id} disconnected") if __name__ == "__main__": import uvicorn uvicorn.run( "app.main:app", host=settings.BACKEND_HOST, port=settings.BACKEND_PORT, reload=settings.DEBUG, )