auto-stock/backend/app/services/swing/scheduler.py

"""스윙 트레이딩 독립 스케줄러

main.py의 lifespan에서 asyncio.create_task()로 시작되는
백그라운드 태스크. 기존 auto_trade_scheduler()를 대체.

60초 주기 실행 흐름:
  1. 장 오픈 체크 (장외 시 30분 대기)
  2. 스윙 포지션 손절/익절 체크 (check_and_execute_exit_v2)
  3. MDD 체크 + 강제 축소 실행 (10% → 30% 축소, 20% → 50% 축소)
  4. AUTO_TRADE_ENABLED 체크
  5. Redis 캐시 또는 DB에서 추천 조회
  6. MCP RISK_ASSESSMENT → 승인 시 스윙 매수 실행

MDD 축소 (CLAUDE.md 5.3):
  - reduction_applied == "10pct": 30% 축소 (REDUCE_AND_PAUSE)
  - reduction_applied == "20pct": 50% 축소 (FORCE_REDUCE, 레버리지 ETF 우선)

추천 데이터 소스 (우선순위):
  1. Redis: "recommendations:latest" 키 (배치 분석 결과 캐시)
  2. DB: get_rule_based_recommendations_from_db() (Redis 미스 시 폴백)
"""
import asyncio
import logging

import pytz
from datetime import timedelta as _td

from app.core.config import settings
from app.core.database import AsyncSessionLocal
from app.core.market_schedule import is_market_open
from app.services.mdd_manager import mdd_manager
from app.services.swing.exit_service import check_and_execute_exit_v2
from app.services.swing.entry_service import execute_swing_entries

logger = logging.getLogger(__name__)
_KST = pytz.timezone("Asia/Seoul")


def _normalize_recommendations(rec: dict) -> dict:
    """recommendations_by_rule 키 정규화.

    Redis 캐시에 "recommendations" 키로 저장된 경우
    "recommendations_by_rule"로 변환하여 execute_swing_entries()와 호환.

    Args:
        rec: 추천 딕셔너리

    Returns:
        dict: 정규화된 추천 딕셔너리
    """
    if not isinstance(rec, dict):
        return rec
    if "recommendations_by_rule" in rec:
        return rec
    if "recommendations" in rec:
        rec["recommendations_by_rule"] = rec["recommendations"]
    return rec


async def swing_scheduler() -> None:
    """60초 주기 스윙 트레이딩 스케줄러.

    무한 루프로 실행되며 asyncio.CancelledError로만 종료.
    각 사이클에서 예외 발생 시 로그 기록 후 다음 사이클 계속.

    DB 세션은 사이클마다 AsyncSessionLocal()로 새로 생성하여
    장시간 세션 유지로 인한 커넥션 풀 고갈 방지.
    """
    await asyncio.sleep(2)
    from app.api.endpoints.recommendations import (
        get_rule_based_recommendations_from_db,
    )
    from app.api.endpoints.mcp import request_mcp_risk_assessment
    from app.core.redis import get_json

    logger.info("📈 [Swing] Scheduler started (every 60s)")
    try:
        while True:
            try:
                logger.info("[Swing] ── 사이클 시작 ──")

                if not is_market_open():
                    logger.info("[Swing] 장 미오픈 — 30분 대기")
                    await asyncio.sleep(1800)
                    continue

                # ── Phase 1: 청산 체크 (별도 세션) ──
                async with AsyncSessionLocal() as exit_session:
                    try:
                        exits = await check_and_execute_exit_v2(exit_session)
                        if exits:
                            logger.info(f"[Swing] 청산 실행: {len(exits)}건")
                    except Exception as e:
                        logger.error(f"[Swing] 청산 체크 오류: {e}", exc_info=True)

                # ── Phase 2: MDD 체크 + 강제 축소 (세션 불필요) ──
                can_open_new = await mdd_manager.can_open_new_position()
                logger.info(f"[Swing] MDD 체크: can_open={can_open_new}")
                if not can_open_new:
                    logger.warning("[Swing] MDD 초과 — 신규 진입 차단")
                    if mdd_manager.reduction_applied:
                        try:
                            reduce_pct = 0.50 if mdd_manager.reduction_applied == "20pct" else 0.30
                            mdd_action = {
                                "action": "FORCE_REDUCE" if mdd_manager.reduction_applied == "20pct" else "REDUCE_AND_PAUSE",
                                "reduce_pct": reduce_pct,
                                "prioritize_leverage": mdd_manager.reduction_applied == "20pct",
                            }
                            results = await mdd_manager.execute_reduction(mdd_action)
                            if results:
                                logger.warning(f"[Swing] MDD 강제 축소 완료: {len(results)}종목")
                        except Exception as e:
                            logger.error(f"[Swing] MDD 강제 축소 실패: {e}", exc_info=True)
                    await asyncio.sleep(60)
                    continue

                # 3. 자동매매 활성화 여부
                if not settings.AUTO_TRADE_ENABLED:
                    logger.info("[Swing] AUTO_TRADE_ENABLED=False — 스킵")
                    await asyncio.sleep(60)
                    continue

                # ── Phase 3: 추천 조회 + 진입 실행 (별도 세션) ──
                # 4. Redis 캐시 확인 → cache miss 시 DB 조회
                rec = await get_json("recommendations:latest")
                if rec:
                    rec = _normalize_recommendations(rec)
                    logger.info(f"[Swing] Redis 캐시 히트 (추천 {len(rec.get('recommendations_by_rule', []))}그룹)")
                else:
                    logger.info("[Swing] Redis 캐시 미스 → DB 조회")
                    async with AsyncSessionLocal() as rec_session:
                        rec = await get_rule_based_recommendations_from_db(rec_session)

                if not rec or not rec.get("recommendations_by_rule"):
                    logger.info("[Swing] 추천 데이터 없음 — 스킵")
                    await asyncio.sleep(60)
                    continue

                # 5. MCP RISK_ASSESSMENT
                recommendations = rec.get("recommendations_by_rule", [])
                if isinstance(recommendations, dict):
                    rec_count = sum(len(v) for v in recommendations.values())
                    top_stocks = [
                        item.get("stock_code", "")
                        for items in recommendations.values()
                        for item in (items[:3] if isinstance(items, list) else [])
                    ][:10]
                else:
                    rec_count = sum(
                        len(g.get("stocks", [])) for g in recommendations
                        if isinstance(g, dict)
                    )
                    top_stocks = [
                        s.get("stock_code", "")
                        for g in recommendations if isinstance(g, dict)
                        for s in g.get("stocks", [])[:3]
                    ][:10]

                trade_plan = {
                    "action": "BUY",
                    "strategy": "swing",
                    "total_recommendations": rec_count,
                    "top_stocks": top_stocks,
                }

                logger.info(f"[Swing] MCP 리스크 평가 요청: {len(top_stocks)}종목")
                mcp_approved, mcp_reason = await request_mcp_risk_assessment(
                    trade_plan, {"strategy": "swing"}, {"top_stocks": top_stocks}
                )
                logger.info(f"[Swing] MCP 결과: approved={mcp_approved}, reason={mcp_reason}")
                if not mcp_approved:
                    logger.warning(f"[Swing] MCP 거부: {mcp_reason}")
                    from app.services.db_worker import enqueue_db_write
                    await enqueue_db_write({
                        "type": "log_mcp_deny",
                        "reason": mcp_reason,
                        "trade_plan": trade_plan,
                    })
                    await asyncio.sleep(60)
                    continue

                # 6. 스윙 진입 실행 (별도 세션)
                async with AsyncSessionLocal() as entry_session:
                    logger.info(f"[Swing] 진입 실행 시작: {top_stocks[:5]}")
                    await execute_swing_entries(rec, entry_session)
                    logger.info("[Swing] ── 사이클 완료 ──")

            except Exception as e:
                logger.error(f"[Swing] 스케줄러 사이클 오류: {e}", exc_info=True)

            await asyncio.sleep(60)

    except asyncio.CancelledError:
        logger.info("[Swing] Scheduler stopped")
        raise