auto-stock/backend/app/services/strategy_optimizer.py

"""
전략 자동 최적화 서비스

[역할]
- 최근 거래 성과 약점 분석 → 파라미터 범위 결정
- BacktestEngine을 사용한 그리드 서치로 최적 파라미터 탐색
- 평가 지표: Sharpe * 0.4 + win_rate * 0.3 + (1 - |MDD|/100) * 0.3
- Sharpe 10% 이상 개선 시 strategy_config 자동 적용
  (단 `settings.STRATEGY_AUTO_APPLY_ENABLED` 기본 False — 계산·로그만 하고 DB 쓰기는 차단)
- StrategyOptimizationLog 테이블에 이력 저장

[사용법]
    from app.services.strategy_optimizer import run_optimization

    async with AsyncSessionLocal() as db:
        result = await run_optimization(db, perf_data, trigger="manual")
"""
import asyncio
import logging
from datetime import date, datetime, timedelta
from typing import Any, Dict, List, Optional, Tuple
from dataclasses import dataclass

from sqlalchemy import select, desc
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import AsyncSessionLocal
from app.models.models import StrategyOptimizationLog, StockDailyPrice
from app.services.strategy_config_service import strategy_cfg
from app.core.config import settings
from app.services.backtest_engine import BacktestEngine, BacktestConfig
from app.core.timeutil import now_kst

logger = logging.getLogger(__name__)


# ─────────────────────────────────────────────────────────────────────────────
# 결과 데이터클래스
# ─────────────────────────────────────────────────────────────────────────────

@dataclass
class OptimizationResult:
    best_params: Dict[str, Any]
    best_score: float
    best_sharpe: float
    best_win_rate: float
    best_mdd: float
    current_sharpe: float
    previous_params: Dict[str, Any]
    applied: bool
    apply_reason: str
    total_combinations: int
    # IS/OOS 검증 결과 (v2 추가)
    oos_sharpe: float = 0.0
    oos_win_rate: float = 0.0
    oos_mdd: float = 0.0
    is_oos_sharpe_ratio: float = 0.0  # OOS Sharpe / IS Sharpe (1.0에 가까울수록 건전)


# ─────────────────────────────────────────────────────────────────────────────
# 파라미터 범위 결정 (성과 약점 기반)
# ─────────────────────────────────────────────────────────────────────────────

def _build_param_grid(perf_data: Dict[str, Any]) -> Dict[str, List[Any]]:
    """
    성과 약점을 분석하여 그리드 서치할 파라미터 범위를 결정.

    규칙:
    - 승률 < 50%    → RSI_OVERSOLD_LEVEL 강화 (더 낮은 값)
    - avg_loss > 손절 기준 → ATR_STOP_MULTIPLIER, STOP_LOSS_PERCENT 조정
    - MDD 과다(< -15%) → MAX_POSITION_SIZE, MAX_HOLDINGS_COUNT 조정
    - 항상 RSI 기간 변화도 포함 (성과 개선 여지 높음)
    """
    summary = perf_data.get("summary", {})
    win_rate = summary.get("win_rate", 50.0)
    avg_loss = summary.get("avg_loss_pct", -5.0)

    grid: Dict[str, List[Any]] = {}

    # RSI_OVERSOLD_LEVEL: 승률 기반
    if win_rate < 40:
        grid["RSI_OVERSOLD_LEVEL"] = [20.0, 25.0, 28.0]
    elif win_rate < 50:
        grid["RSI_OVERSOLD_LEVEL"] = [25.0, 28.0, 30.0]
    else:
        grid["RSI_OVERSOLD_LEVEL"] = [28.0, 30.0, 32.0]

    # STOP_LOSS_PERCENT: 평균 손실 기반
    if avg_loss < -8.0:
        grid["STOP_LOSS_PERCENT"] = [-5.0, -6.0, -7.0]
    elif avg_loss < -6.0:
        grid["STOP_LOSS_PERCENT"] = [-5.0, -6.0]
    else:
        grid["STOP_LOSS_PERCENT"] = [-6.0, -7.0, -8.0]

    # ATR_STOP_MULTIPLIER: 항상 포함
    grid["ATR_STOP_MULTIPLIER"] = [2.0, 2.5, 3.0]

    # BOLLINGER_STDDEV
    grid["BOLLINGER_STDDEV"] = [1.8, 2.0, 2.2]

    return grid


def _cartesian_product(grid: Dict[str, List[Any]]) -> List[Dict[str, Any]]:
    """딕셔너리 그리드의 카르테시안 곱을 반환."""
    keys = list(grid.keys())
    values = list(grid.values())

    def _product(lists):
        result = [[]]
        for lst in lists:
            result = [x + [y] for x in result for y in lst]
        return result

    combos = _product(values)
    return [dict(zip(keys, combo)) for combo in combos]


# ─────────────────────────────────────────────────────────────────────────────
# 현재 전략 파라미터 수집
# ─────────────────────────────────────────────────────────────────────────────

async def _get_current_params() -> Dict[str, Any]:
    """현재 strategy_cfg 값들을 읽어 반환."""
    await strategy_cfg._refresh()
    return {
        # [2026-09-27] 하드코딩 폴백 → config.py 정합 (우선순위: strategy_cfg(DB) > settings)
        "RSI_OVERSOLD_LEVEL": strategy_cfg.get("RSI_OVERSOLD_LEVEL", settings.RSI_OVERSOLD_LEVEL),  # 30.0 동일
        "STOP_LOSS_PERCENT": strategy_cfg.get("STOP_LOSS_PERCENT", -abs(settings.STOP_LOSS_PERCENT)),  # 옛 폴백 -6.0 → -7.0
        "ATR_STOP_MULTIPLIER": strategy_cfg.get("ATR_STOP_MULTIPLIER", settings.ATR_STOP_MULTIPLIER),  # 2.5 동일
        "BOLLINGER_STDDEV": strategy_cfg.get("BOLLINGER_STDDEV", settings.BOLLINGER_STDDEV),  # 2.0 동일
        "MAX_POSITION_SIZE": strategy_cfg.get("MAX_POSITION_SIZE", settings.MAX_POSITION_SIZE),  # 0.04 동일
        "MAX_HOLDINGS_COUNT": strategy_cfg.get("MAX_HOLDINGS_COUNT", settings.MAX_HOLDINGS_COUNT),  # 20 동일
    }


# ─────────────────────────────────────────────────────────────────────────────
# 백테스트로 파라미터 평가
# ─────────────────────────────────────────────────────────────────────────────

def _compute_score(sharpe: float, win_rate: float, mdd: float) -> float:
    """통합 평가 지표 계산: Sharpe*0.4 + win_rate*0.3 + (1-|mdd|/100)*0.3"""
    mdd_penalty = max(0.0, 1.0 - abs(mdd) / 100.0)
    return sharpe * 0.4 + (win_rate / 100.0) * 0.3 + mdd_penalty * 0.3


async def _evaluate_params(
    params: Dict[str, Any],
    stock_codes: List[str],
    start_date: date,
    end_date: date,
    initial_capital: float,
) -> Tuple[float, float, float, float]:
    """
    단일 파라미터 조합으로 백테스트 실행.
    Returns: (score, sharpe, win_rate, mdd)
    """
    try:
        backtest_params = {
            # [2026-09-27] 폴백 config.py 정합 (옛값: 30.0 / -6.0)
            "MEAN_REVERSION_RSI_THRESHOLD": params.get("RSI_OVERSOLD_LEVEL", settings.RSI_OVERSOLD_LEVEL),
            "MEAN_REVERSION_STOP_LOSS_PCT": params.get("STOP_LOSS_PERCENT", -abs(settings.STOP_LOSS_PERCENT)),
            "MAX_TOTAL_POSITIONS": min(5, len(stock_codes)),
        }
        config = BacktestConfig(
            stock_codes=stock_codes,
            start_date=start_date,
            end_date=end_date,
            initial_capital=initial_capital,
            strategy_params=backtest_params,
            max_positions=min(5, len(stock_codes)),
            position_size=params.get("MAX_POSITION_SIZE", 0.2),
        )
        engine = BacktestEngine()
        result = await engine.run_backtest(config)

        sharpe = float(result.sharpe_ratio or 0.0)
        win_rate = float(result.win_rate or 0.0)
        mdd = float(result.mdd_pct or 0.0)
        score = _compute_score(sharpe, win_rate, mdd)
        return score, sharpe, win_rate, mdd
    except Exception as e:
        logger.debug(f"백테스트 실패 (파라미터: {params}): {e}")
        return -1.0, 0.0, 0.0, 0.0


# ─────────────────────────────────────────────────────────────────────────────
# 대표 종목 선택
# ─────────────────────────────────────────────────────────────────────────────

async def _get_representative_stocks(period_days: int = 90) -> List[str]:
    """
    최근 N일 거래가 있는 대표 종목 10개 반환.
    백테스트 실행 시간을 줄이기 위해 종목 수 제한.
    """
    from app.models.models import TradeHistory
    from sqlalchemy import func, and_
    from datetime import datetime, timedelta

    since = now_kst() - timedelta(days=period_days)

    async with AsyncSessionLocal() as db:
        stmt = (
            select(TradeHistory.stock_code, func.count(TradeHistory.id).label("cnt"))
            .where(TradeHistory.created_at >= since)
            .group_by(TradeHistory.stock_code)
            .order_by(desc("cnt"))
            .limit(10)
        )
        result = await db.execute(stmt)
        codes = [row[0] for row in result.fetchall()]

    # 거래 이력 없으면 stock_daily_prices에서 최근 종목 fallback
    if not codes:
        async with AsyncSessionLocal() as db:
            stmt = (
                select(StockDailyPrice.stock_code)
                .where(StockDailyPrice.trade_date >= (date.today() - timedelta(days=period_days)))
                .group_by(StockDailyPrice.stock_code)
                .limit(10)
            )
            result = await db.execute(stmt)
            codes = [row[0] for row in result.fetchall()]

    return codes or []


# ─────────────────────────────────────────────────────────────────────────────
# 파라미터 적용
# ─────────────────────────────────────────────────────────────────────────────

async def apply_params(params: Dict[str, Any], reason: str = "auto_optimizer") -> bool:
    """strategy_cfg에 파라미터를 일괄 적용.

    [2026-09-29] `settings.STRATEGY_AUTO_APPLY_ENABLED` (기본 **False**) 가 꺼져 있으면
    **아무것도 쓰지 않고 False 를 반환한다.** 이 함수가 DB `strategy_configs` 를 바꾸는
    유일한 지점이고(§0 우선순위 2위), 호출 게이트가 "최근 7일 SELL ≥ 5건"이라 5거래
    표본의 그리드 서치가 RSI 진입·손절·포지션 크기를 덮어쓸 수 있다 — CLAUDE.md
    §8.1-5 / §8.4(파라미터당 100거래) / §8.5-1 과 정면 충돌한다.

    가드는 **`try` 밖**에 둔다(§4). try 안에 두면 예외 시 가드가 조용히 통과한다.
    호출부(`run_optimization`)에도 같은 가드가 있고 상세 로그를 남긴다 — 여기는
    다른 호출부가 생겨도 뚫리지 않게 하는 최종 방어선이다.
    """
    if not settings.STRATEGY_AUTO_APPLY_ENABLED:
        logger.warning(
            "🚫 전략 파라미터 자동 적용 차단 (STRATEGY_AUTO_APPLY_ENABLED=False) — "
            "적용하려던 값: %s (사유: %s). 반영하려면 전략설정 UI 에서 수동으로 하거나 "
            "CLAUDE.md §8.4 백테스트 요건 충족 후 플래그를 켤 것.",
            params, reason,
        )
        return False

    try:
        for key, value in params.items():
            await strategy_cfg.upsert(key, str(value), updated_by=reason)
        logger.info(f"✅ 전략 파라미터 자동 적용 완료: {list(params.keys())} (사유: {reason})")
        return True
    except Exception as e:
        logger.error(f"❌ 파라미터 적용 실패: {e}", exc_info=True)
        return False


# ─────────────────────────────────────────────────────────────────────────────
# 최적화 이력 저장
# ─────────────────────────────────────────────────────────────────────────────

async def save_optimization_log(db: AsyncSession, result: OptimizationResult, trigger: str) -> StrategyOptimizationLog:
    """최적화 결과를 DB에 저장."""
    log = StrategyOptimizationLog(
        run_date=date.today(),
        trigger=trigger,
        period_days=90,
        best_params=result.best_params,
        best_sharpe=result.best_sharpe,
        best_win_rate=result.best_win_rate,
        best_mdd=result.best_mdd,
        current_sharpe=result.current_sharpe,
        previous_params=result.previous_params,
        applied=result.applied,
        apply_reason=result.apply_reason,
        total_combinations=result.total_combinations,
    )
    db.add(log)
    await db.commit()
    await db.refresh(log)
    return log


# ─────────────────────────────────────────────────────────────────────────────
# 메인 최적화 함수
# ─────────────────────────────────────────────────────────────────────────────

async def run_optimization(
    db: AsyncSession,
    perf_data: Dict[str, Any],
    trigger: str = "auto",
    period_days: int = 90,
    initial_capital: float = 10_000_000,
) -> OptimizationResult:
    """
    성과 데이터 기반 전략 파라미터 최적화.

    Args:
        db: AsyncSession
        perf_data: analyze_performance() 반환값
        trigger: "auto" | "manual"
        period_days: 백테스트 기간 (일)
        initial_capital: 백테스트 초기 자본

    Returns:
        OptimizationResult
    """
    logger.info(f"전략 최적화 시작 (trigger={trigger}, period={period_days}일)")

    # 1. 현재 파라미터 저장 (롤백용)
    prev_params = await _get_current_params()

    # 2. 그리드 구성
    param_grid = _build_param_grid(perf_data)
    combinations = _cartesian_product(param_grid)
    total_combos = len(combinations)
    logger.info(f"   그리드 조합 수: {total_combos}")

    # 3. 대표 종목 선택
    stock_codes = await _get_representative_stocks(period_days)
    if not stock_codes:
        logger.warning("백테스트 대상 종목 없음. 최적화 건너뜀.")
        return OptimizationResult(
            best_params=prev_params,
            best_score=0.0,
            best_sharpe=0.0,
            best_win_rate=0.0,
            best_mdd=0.0,
            current_sharpe=0.0,
            previous_params=prev_params,
            applied=False,
            apply_reason="no_stock_data",
            total_combinations=0,
        )

    # ── IS/OOS 기간 분할 (70% IS / 30% OOS, 날짜 기준) ──
    end_date = date.today()
    start_date = end_date - timedelta(days=period_days)
    is_days = int(period_days * 0.7)
    is_end_date = start_date + timedelta(days=is_days)
    oos_start_date = is_end_date + timedelta(days=1)
    oos_end_date = end_date

    logger.info(
        f"   IS 기간: {start_date} ~ {is_end_date} ({is_days}일), "
        f"OOS 기간: {oos_start_date} ~ {oos_end_date} ({period_days - is_days}일)"
    )

    # 4. 현재 파라미터 기준 Sharpe 계산 (IS 기간)
    cur_score, cur_sharpe, cur_wr, cur_mdd = await _evaluate_params(
        prev_params, stock_codes, start_date, is_end_date, initial_capital
    )
    logger.info(f"   현재 IS 성과 -> Sharpe: {cur_sharpe:.3f}, 승률: {cur_wr:.1f}%, MDD: {cur_mdd:.1f}%")

    # 5. 그리드 서치 — IS 기간에서 최적 파라미터 탐색 (배치 5개씩 병렬)
    best_score = cur_score
    best_params = dict(prev_params)
    best_sharpe = cur_sharpe
    best_win_rate = cur_wr
    best_mdd = cur_mdd

    batch_size = 5
    for i in range(0, len(combinations), batch_size):
        batch = combinations[i:i + batch_size]
        tasks = [
            _evaluate_params(p, stock_codes, start_date, is_end_date, initial_capital)
            for p in batch
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for combo, res in zip(batch, results):
            if isinstance(res, Exception):
                continue
            score, sharpe, wr, mdd = res
            if score > best_score:
                best_score = score
                best_sharpe = sharpe
                best_win_rate = wr
                best_mdd = mdd
                best_params = {**prev_params, **combo}

    logger.info(f"   IS 최적 성과 -> Sharpe: {best_sharpe:.3f}, 승률: {best_win_rate:.1f}%, MDD: {best_mdd:.1f}%")

    # 6. OOS 기간에서 최적 파라미터 검증 (과적합 탐지)
    oos_score, oos_sharpe, oos_wr, oos_mdd = await _evaluate_params(
        best_params, stock_codes, oos_start_date, oos_end_date, initial_capital
    )
    is_oos_ratio = (oos_sharpe / best_sharpe) if best_sharpe > 0 else 0.0
    logger.info(
        f"   OOS 검증 -> Sharpe: {oos_sharpe:.3f}, 승률: {oos_wr:.1f}%, MDD: {oos_mdd:.1f}%, "
        f"IS/OOS Sharpe 비율: {is_oos_ratio:.2f}"
    )

    if is_oos_ratio < 0.5 and best_sharpe > 0:
        logger.warning(
            f"   [과적합 경고] OOS Sharpe / IS Sharpe = {is_oos_ratio:.2f} < 0.5 "
            "-- 파라미터가 IS 기간에 과적합되었을 가능성 높음"
        )

    # 7. 자동 적용 여부 결정
    # 조건: IS Sharpe 10% 이상 개선 AND OOS Sharpe > 0 (OOS에서도 양의 성과)
    improvement_threshold = 1.1
    _is_improved = (
        cur_sharpe > 0
        and best_sharpe >= cur_sharpe * improvement_threshold
        and best_params != prev_params
    )
    if cur_sharpe <= 0 and best_sharpe > 0:
        _is_improved = True

    # OOS 검증 게이트: OOS Sharpe > 0 이어야 실전 적용 허가
    _oos_passed = oos_sharpe > 0
    _should_apply = _is_improved and _oos_passed

    applied = False
    apply_reason = ""

    if _should_apply:
        improvement_pct = ((best_sharpe - cur_sharpe) / max(abs(cur_sharpe), 0.001)) * 100
        apply_reason = (
            f"IS Sharpe {improvement_pct:.1f}% 향상 ({cur_sharpe:.3f} -> {best_sharpe:.3f}), "
            f"OOS Sharpe: {oos_sharpe:.3f}, IS/OOS 비율: {is_oos_ratio:.2f}"
        )
        # 최적화된 파라미터만 적용 (prev와 다른 키만)
        changed = {k: v for k, v in best_params.items() if str(v) != str(prev_params.get(k, ""))}
        # 가드는 try 밖 (§4) — 예외로 통과되면 실매매 파라미터가 바뀐다
        if changed and not settings.STRATEGY_AUTO_APPLY_ENABLED:
            _diff = ", ".join(
                f"{k}: {prev_params.get(k, '(없음)')} → {v}" for k, v in sorted(changed.items())
            )
            logger.warning(
                "🚫 [Optimizer] 자동 적용 차단 (STRATEGY_AUTO_APPLY_ENABLED=False). "
                "적용될 뻔한 변경 %d건 — %s | IS Sharpe %.3f → %.3f, OOS Sharpe %.3f, "
                "IS/OOS %.2f, 승률 %.1f%%, MDD %.1f%% | %s",
                len(changed), _diff, cur_sharpe, best_sharpe, oos_sharpe, is_oos_ratio,
                best_win_rate, best_mdd, apply_reason,
            )
            apply_reason = (
                f"자동 적용 차단(STRATEGY_AUTO_APPLY_ENABLED=False) — 제안: {_diff} | 근거: {apply_reason}"
            )
        elif changed:
            applied = await apply_params(changed, reason="auto_optimizer")
            if applied:
                logger.info(f"자동 적용 완료: {apply_reason}")
    else:
        if cur_sharpe <= 0 and best_sharpe <= 0:
            apply_reason = "insufficient_data"
        elif not _oos_passed and _is_improved:
            apply_reason = f"OOS 검증 실패 (OOS Sharpe: {oos_sharpe:.3f} <= 0, 과적합 의심)"
        elif best_sharpe < cur_sharpe * improvement_threshold:
            pct = ((best_sharpe - cur_sharpe) / max(abs(cur_sharpe), 0.001)) * 100
            apply_reason = f"IS 개선 미달 ({pct:+.1f}% < 10%)"
        else:
            apply_reason = "no_improvement"
        logger.info(f"자동 적용 건너뜀: {apply_reason}")

    return OptimizationResult(
        best_params=best_params,
        best_score=best_score,
        best_sharpe=best_sharpe,
        best_win_rate=best_win_rate,
        best_mdd=best_mdd,
        current_sharpe=cur_sharpe,
        previous_params=prev_params,
        applied=applied,
        apply_reason=apply_reason,
        total_combinations=total_combos,
        oos_sharpe=oos_sharpe,
        oos_win_rate=oos_wr,
        oos_mdd=oos_mdd,
        is_oos_sharpe_ratio=round(is_oos_ratio, 4),
    )


# ─────────────────────────────────────────────────────────────────────────────
# 마지막 최적화 이력 조회
# ─────────────────────────────────────────────────────────────────────────────

async def get_last_optimization_date(db: AsyncSession) -> Optional[date]:
    """마지막 최적화 실행일 반환 (없으면 None)."""
    stmt = select(StrategyOptimizationLog.run_date).order_by(
        desc(StrategyOptimizationLog.created_at)
    ).limit(1)
    result = await db.execute(stmt)
    row = result.scalar_one_or_none()
    return row