""" 거래 성과 분석 + 전략 자동 최적화 API GET /api/v1/analysis/performance?days=30 — 성과 통계 GET /api/v1/analysis/insights?days=30 — 전략 인사이트 + 개선 제안 POST /api/v1/analysis/generate-journals — 미생성 일지 일괄 생성 POST /api/v1/analysis/optimize-strategy — 전략 최적화 즉시 실행 GET /api/v1/analysis/optimization-history — 최적화 이력 """ import logging from datetime import datetime, date from typing import Any, Dict, List, Optional from fastapi import APIRouter, BackgroundTasks, Depends, Query from app.core.security import require_api_key from sqlalchemy import select, desc from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_db, AsyncSessionLocal from app.models.models import StrategyOptimizationLog, TradeJournal from app.services.trade_analyzer import ( analyze_performance, build_insights, create_journal_record, generate_journal_for_trade, get_uncovered_trades, ) from app.services.strategy_optimizer import ( get_last_optimization_date, run_optimization, save_optimization_log, ) router = APIRouter() logger = logging.getLogger(__name__) # ───────────────────────────────────────────────────────────────────────────── # GET /performance # ───────────────────────────────────────────────────────────────────────────── @router.get("/performance") async def get_performance( days: int = Query(default=30, ge=1, le=365, description="분석 기간 (일)"), db: AsyncSession = Depends(get_db), ) -> Dict[str, Any]: """최근 N일 완료 거래 성과 통계 반환.""" try: return await analyze_performance(db, days=days) except Exception as e: logger.error(f"performance 분석 실패: {e}", exc_info=True) return {"error": str(e), "summary": {}, "total_trades": 0} # ───────────────────────────────────────────────────────────────────────────── # GET /insights # ───────────────────────────────────────────────────────────────────────────── @router.get("/insights") async def get_insights( days: int = Query(default=30, ge=1, le=365), db: AsyncSession = Depends(get_db), ) -> Dict[str, Any]: """전략 인사이트 및 개선 제안 반환.""" try: perf = await analyze_performance(db, days=days) insights = build_insights(perf) last_opt_date = await get_last_optimization_date(db) return { "insights": insights, "performance_summary": perf.get("summary", {}), "last_optimization_date": last_opt_date.isoformat() if last_opt_date else None, "period_days": days, "generated_at": datetime.utcnow().isoformat(), } except Exception as e: logger.error(f"insights 생성 실패: {e}", exc_info=True) return {"insights": [], "error": str(e)} # ───────────────────────────────────────────────────────────────────────────── # POST /generate-journals # ───────────────────────────────────────────────────────────────────────────── async def _generate_journals_task(days: int = 7) -> Dict[str, Any]: """백그라운드에서 미생성 일지를 일괄 생성.""" async with AsyncSessionLocal() as db: try: trades = await get_uncovered_trades(db, days=days) if not trades: return {"created": 0, "message": "생성할 미처리 거래 없음"} perf = await analyze_performance(db, days=30) created = 0 for trade in trades: try: content = generate_journal_for_trade(trade, perf) await create_journal_record(db, trade, content) created += 1 except Exception as e: logger.warning(f"일지 생성 실패 ({trade.stock_code}): {e}") await db.commit() logger.info(f"✅ 매매일지 {created}건 생성 완료 (대상: {len(trades)}건)") return {"created": created, "total_uncovered": len(trades)} except Exception as e: await db.rollback() logger.error(f"일지 일괄 생성 실패: {e}", exc_info=True) return {"created": 0, "error": str(e)} @router.post("/generate-journals", dependencies=[Depends(require_api_key)]) async def generate_journals( background_tasks: BackgroundTasks, days: int = Query(default=7, ge=1, le=30, description="최근 N일 내 미생성 거래 대상"), db: AsyncSession = Depends(get_db), ) -> Dict[str, Any]: """일지가 없는 완료 거래에 대해 템플릿 기반 매매일지를 일괄 생성.""" try: uncovered = await get_uncovered_trades(db, days=days) count = len(uncovered) if count == 0: return {"status": "skipped", "message": "생성할 미처리 거래 없음", "queued": 0} background_tasks.add_task(_generate_journals_task, days) return { "status": "started", "message": f"{count}건의 거래 일지 생성을 시작했습니다.", "queued": count, } except Exception as e: logger.error(f"generate-journals 실패: {e}", exc_info=True) return {"status": "error", "message": str(e)} # ───────────────────────────────────────────────────────────────────────────── # POST /optimize-strategy # ───────────────────────────────────────────────────────────────────────────── # 최적화 진행 상태 (단순 인메모리 플래그) _optimization_running: bool = False @router.post("/optimize-strategy", dependencies=[Depends(require_api_key)]) async def optimize_strategy( background_tasks: BackgroundTasks, days: int = Query(default=30, ge=7, le=180, description="성과 분석 기간"), period_days: int = Query(default=90, ge=30, le=365, description="백테스트 기간"), ) -> Dict[str, Any]: """ 전략 파라미터 최적화를 즉시 실행 (수동 트리거). - 성과 분석 → 파라미터 그리드 서치 → Sharpe 10% 개선 시 자동 적용 - 백그라운드에서 실행 (수 분 소요 가능) """ global _optimization_running if _optimization_running: return {"status": "already_running", "message": "최적화가 이미 실행 중입니다."} async def _opt_task(): global _optimization_running _optimization_running = True try: async with AsyncSessionLocal() as db: perf = await analyze_performance(db, days=days) result = await run_optimization(db, perf, trigger="manual", period_days=period_days) await save_optimization_log(db, result, trigger="manual") logger.info(f"✅ 수동 최적화 완료: applied={result.applied}, reason={result.apply_reason}") except Exception as e: logger.error(f"최적화 실패: {e}", exc_info=True) finally: _optimization_running = False background_tasks.add_task(_opt_task) return { "status": "started", "message": "전략 최적화를 시작했습니다. /optimization-history 에서 결과를 확인하세요.", "performance_days": days, "backtest_period_days": period_days, } # ───────────────────────────────────────────────────────────────────────────── # GET /optimization-history # ───────────────────────────────────────────────────────────────────────────── @router.get("/optimization-history") async def get_optimization_history( limit: int = Query(default=10, ge=1, le=50), db: AsyncSession = Depends(get_db), ) -> List[Dict[str, Any]]: """전략 최적화 실행 이력 반환.""" try: stmt = ( select(StrategyOptimizationLog) .order_by(desc(StrategyOptimizationLog.created_at)) .limit(limit) ) result = await db.execute(stmt) logs = result.scalars().all() return [ { "id": log.id, "run_date": log.run_date.isoformat() if log.run_date else None, "trigger": log.trigger, "period_days": log.period_days, "best_sharpe": float(log.best_sharpe) if log.best_sharpe else None, "current_sharpe": float(log.current_sharpe) if log.current_sharpe else None, "sharpe_improvement_pct": ( round( (float(log.best_sharpe) - float(log.current_sharpe)) / max(abs(float(log.current_sharpe)), 0.001) * 100, 2, ) if log.best_sharpe and log.current_sharpe else None ), "best_win_rate": float(log.best_win_rate) if log.best_win_rate else None, "best_mdd": float(log.best_mdd) if log.best_mdd else None, "applied": log.applied, "apply_reason": log.apply_reason, "best_params": log.best_params, "previous_params": log.previous_params, "total_combinations": log.total_combinations, "created_at": log.created_at.isoformat() if log.created_at else None, } for log in logs ] except Exception as e: logger.error(f"optimization-history 조회 실패: {e}", exc_info=True) return [] # ───────────────────────────────────────────────────────────────────────────── # GET /optimization-status # ───────────────────────────────────────────────────────────────────────────── @router.get("/optimization-status") async def get_optimization_status() -> Dict[str, Any]: """현재 최적화 실행 상태 조회.""" return {"running": _optimization_running}