feat: batch_analysis_scheduler → Redis SETEX, auto_trade_scheduler → Redis GET (캐시 히트 시 DB 생략)

- batch_analysis_scheduler: recommendations:latest SETEX (TTL 1800s)
- auto_trade_scheduler: Redis GET → cache miss 시 DB fallback
- 포맷 정규화: recommendations 리스트 → recommendations_by_rule dict 자동 변환

backend/app/main.py

@@ -48,14 +48,46 @@ logging.basicConfig(
 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.
-    auto_trade_scheduler가 DB를 빠르게 읽어 매매 판단에 활용한다.
+    분석 완료 후 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
 
     interval = int(getattr(settings, "BATCH_ANALYSIS_INTERVAL", 600))  # 기본 10분
     logger.info(f"📊 Batch analysis scheduler started (every {interval}s)")
@@ -69,6 +101,22 @@ async def batch_analysis_scheduler():
                 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)"
+                        )
+                # ───────────────────────────────────────────────────────
+
             except Exception as e:
                 logger.error(f"❌ Batch analysis error: {e}", exc_info=True)
             await asyncio.sleep(interval)
@@ -79,9 +127,10 @@ async def batch_analysis_scheduler():
 
 async def auto_trade_scheduler():
     """1분 간격 자동매매 스케줄러.
-    DB 캐시에서 추천 종목을 읽어 손절/익절 및 신규 매수를 실행한다.
-    주문 직전에 MCP RISK_ASSESSMENT를 호출해 승인된 경우에만 주문한다.
-    실제 KIS API 분석은 batch_analysis_scheduler(10분)가 담당한다.
+
+    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 (
@@ -89,10 +138,10 @@ async def auto_trade_scheduler():
         execute_auto_trade_background,
     )
     from app.api.endpoints.portfolio import check_and_execute_exit_trades
-    # 공통 MCP 리스크 평가 함수 (이슈 4)
     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, DB mode)")
+    logger.info("⏱️ Auto-trade scheduler started (every 60s, Redis→DB fallback mode)")
     try:
         while True:
             try:
@@ -109,10 +158,22 @@ async def auto_trade_scheduler():
                     except Exception as e:
                         logger.error(f"Exit trade check error: {e}", exc_info=True)
 
-                    # 2. 자동매매: DB 캐시에서 추천 조회
+                    # 2. 자동매매: Redis 캐시 우선 → cache miss 시만 DB 조회
                     if settings.AUTO_TRADE_ENABLED:
-                        logger.info("🔄 Checking DB recommendations for auto-trade...")
+                        # ── 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) ──────────
@@ -134,7 +195,7 @@ async def auto_trade_scheduler():
                                     )
                                 ][:10],
                             }
-                            portfolio: dict = {}  # 확장 시 실제 포트폴리오 요약 전달
+                            portfolio: dict = {}
                             signals = {
                                 "source": "db_recommendations",
                                 "rules_triggered": list(recommendations.keys()),
@@ -163,7 +224,7 @@ async def auto_trade_scheduler():
                                 logger.info("📊 Found recommendations, executing auto-trade")
                                 await execute_auto_trade_background(rec, session)
                         else:
-                            logger.info("⏭️ Auto-trade skipped: no recommendations in DB")
+                            logger.info("⏭️ Auto-trade skipped: no recommendations")
                     else:
                         logger.info("⏸️ Auto-trade disabled - only monitoring stop-loss/take-profit")
             except Exception as e:
@@ -530,13 +591,11 @@ async def websocket_endpoint(websocket: WebSocket, client_id: str):
                     client_id,
                 )
                 # 구독 종목 차트 캐시 사전 워밍업 (백그라운드, 논블로킹)
-                # 사용자가 종목 클릭 전에 캐시가 완성되도록 미리 KIS API 호출
                 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