""" 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.database import engine, Base, AsyncSessionLocal from app.models import models # 모델 임포트 from app.core.market_schedule import is_market_open from sqlalchemy import text # 로깅 설정 logging.basicConfig( level=getattr(logging, settings.LOG_LEVEL.upper(), logging.INFO), format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) async def auto_trade_scheduler(): """1분 간격 자동매매 스케줄러. 실시간 분석을 수행하고, 보유 종목 손절/익절을 체크하며, 자동매매를 실행한다. """ # 지연 시작으로 앱 부팅 안정화 await asyncio.sleep(2) from app.api.endpoints.recommendations import ( get_rule_based_recommendations_realtime, execute_auto_trade_background, ) from app.api.endpoints.portfolio import check_and_execute_exit_trades logger.info("⏱️ Auto-trade scheduler started (every 60s)") try: while True: try: # 장 운영 시간 체크 if not is_market_open(): logger.info("💤 Market is closed. Skipping auto-trade cycle. Next check in 30 minutes.") # 장 종료 시에는 30분(1800초) 대기하여 리소스 절약 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. 자동매매가 활성화된 경우에만 신규 매수 진행 if settings.AUTO_TRADE_ENABLED: logger.info("🔄 Starting real-time analysis cycle...") rec = await get_rule_based_recommendations_realtime() if rec and rec.get("recommendations_by_rule"): logger.info(f"📊 Found recommendations, executing auto-trade") await execute_auto_trade_background(rec, session) else: logger.info("⏭️ Auto-trade skipped: no recommendations matched rules") 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...") kis_mode = settings.KIS_MODE.upper() 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) # Ensure trade_history columns exist (self-heal without external migration) 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; """)) # stock_analysis에 ATR / ADX / ML 컬럼 추가 (CLAUDE.md 섹션 6.2) 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); """)) # active_trades에 분할 청산 컬럼 추가 (CLAUDE.md 섹션 7.3) 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); """)) # Risk Management 테이블 생성 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 ); """)) # 종목별 연속 손실 ban 영속화 테이블 (CLAUDE.md 섹션 7.4) 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)") # 연속 손실 ban 상태 복원 (재시작 후에도 유지 — CLAUDE.md 섹션 7.4) 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 date, timedelta async with AsyncSessionLocal() as session: _kst = pytz.timezone("Asia/Seoul") _today_kst = datetime.now(_kst).date() # 오래된 stock_analysis 삭제 (7일 이전) 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") # PENDING/FAILED 오래된 거래 삭제 (1일 이전) 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") # 수량 0인 포트폴리오 삭제 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}") logger.error("Please check if PostgreSQL server is running and accessible") 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}") # 스케줄러 시작 app.state.auto_trade_task = asyncio.create_task(auto_trade_scheduler()) app.state.day_trade_task = asyncio.create_task(day_trade_scheduler()) yield # 종료 시 정리 작업 # 스케줄러 중단 for task_attr in ("auto_trade_task", "day_trade_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}") 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 설정 (반드시 라우터 등록 전에) # allow_origins=["*"] + allow_credentials=True 조합은 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} - Status: {response.status_code} - Time: {process_time:.3f}s") return response # API 라우터 등록 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 연결 엔드포인트""" await websocket_manager.connect(websocket, client_id) try: while True: data = await websocket.receive_text() # ping/pong 하트비트 처리 try: msg = json.loads(data) if msg.get("type") == "ping": await websocket_manager.send_personal_message( json.dumps({"type": "pong"}), client_id ) continue except (json.JSONDecodeError, AttributeError): pass 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, )