Merge pull request #20 from SynergyX-AI-Pattern/bug/#18_prediction_batch_upsert

[FIX] 예측 배치, 배포 오류 해결 #18

Author: eldeoddt

.github/workflows/fastapi-dev-ci-cd.yml

@@ -43,10 +43,10 @@ jobs:
           rsync -avz -e "ssh -i private_key.pem -o StrictHostKeyChecking=no" \
             --exclude venv --exclude .git ./ $EC2_USERNAME@$EC2_HOST:/home/$EC2_USERNAME/fastapi-app/
 
-          # .env 파일 전송
+          # .env 전송
           scp -i private_key.pem -o StrictHostKeyChecking=no .env $EC2_USERNAME@$EC2_HOST:/home/$EC2_USERNAME/fastapi-app/.env
 
-          # 서버 실행 스크립트
+          # 서버 설정 및 재시작
           ssh -i private_key.pem -o StrictHostKeyChecking=no $EC2_USERNAME@$EC2_HOST "
             set -e
             cd /home/$EC2_USERNAME/fastapi-app
@@ -62,14 +62,11 @@ jobs:
             pip install --upgrade pip
             pip install -r requirements.txt
 
-            # fast api 서버 재시작
-            pkill -f 'uvicorn' || true
-            export PYTHONPATH=$(pwd)
-          
-            echo '[INFO] FastAPI 실행 중...'
-            nohup ./venv/bin/uvicorn app.main:app --host 0.0.0.0 --port 8000 > fastapi.log 2>&1 &
-            sleep 8
-            tail -n 50 fastapi.log
+            # systemd 서비스 재시작
+            echo '[INFO] FastAPI 서비스 재시작...'
+            sudo systemctl daemon-reload
+            sudo systemctl restart fastapi
+            sudo systemctl status fastapi --no-pager -l
           "
 
           rm -f private_key.pem

app/main.py

@@ -3,7 +3,6 @@
 from fastapi import FastAPI
 import logging
 from dotenv import load_dotenv
-from app.core.config import settings
 from app.core.logging_config import setup_logging
 from app.exceptions.base import APIException
 from app.exceptions.exception_handlers import api_exception_handler
@@ -16,6 +15,7 @@
 from starlette.exceptions import HTTPException as StarletteHTTPException
 from app.schedulers.predict_batch_runner import start_batch_scheduler
 from app.schedulers.pattern_detection_runner import start_pattern_detection_scheduler
+from app.core.config import settings
 
 load_dotenv()
 setup_logging()
@@ -27,17 +27,23 @@
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     if settings.ENV == "prod":
-        try:
-            logger.info("prod - 배치 스케줄러 실행 시작")
-            start_batch_scheduler()
-        except Exception as e:
-            logger.exception(f"배치 스케줄러 시작 실패: {e}")
-
-        try:
-            logger.info("prod - 패턴 감지 스케줄러 실행 시작")
-            start_pattern_detection_scheduler()
-        except Exception as e:
-            logger.exception(f"패턴 감지 스케줄러 시작 실패: {e}")
+        # 중복 실행 방지
+        if not getattr(app.state, "schedulers_started", False):
+            try:
+                logger.info("prod - 배치 스케줄러 실행 시작")
+                start_batch_scheduler()
+            except Exception as e:
+                logger.exception(f"배치 스케줄러 시작 실패: {e}")
+
+            try:
+                logger.info("prod - 패턴 감지 스케줄러 실행 시작")
+                start_pattern_detection_scheduler()
+            except Exception as e:
+                logger.exception(f"패턴 감지 스케줄러 시작 실패: {e}")
+
+            app.state.schedulers_started = True
+        else:
+            logger.info("스케줄러가 이미 시작되어 있어 생략")
     yield
 
 

app/schedulers/predict_batch_runner.py

@@ -2,7 +2,6 @@
 import logging
 import time
 import asyncio
-from concurrent.futures import ThreadPoolExecutor, as_completed
 from app.services.batch_service import BatchService
 from app.utils.discord_notifier import notify_discord_async
 
@@ -12,48 +11,42 @@
 def start_batch_scheduler():
     scheduler = BackgroundScheduler(timezone="Asia/Seoul")
 
-    @scheduler.scheduled_job('cron', hour=19, minute=15)
+    @scheduler.scheduled_job(
+        'cron',
+        hour=16, minute=5, day_of_week='mon-fri',
+        coalesce=False,  # 밀린 작업 버림
+        max_instances=1,  # 중복 실행 방지
+        misfire_grace_time=600  # 지연 허용 10분
+    )
     def run_batch():
-        logger.info("[Scheduler] 18:05 배치 예측 시작")
-        start = time.time()
-
-        success_count = 0
-        fail_count = 0
-        failed_symbols = [] # 실패한 종목
-
-        start_id = 1
-        end_id = 100
-        max_workers = 3
-
-        with ThreadPoolExecutor(max_workers=max_workers) as executor:
-            futures = {
-                executor.submit(BatchService.process_single_stock, stock_id): stock_id
-                for stock_id in range(start_id, end_id + 1)
-            }
-
-            for future in as_completed(futures):
-                stock_id = futures[future]
-                try:
-                    stock_id, success, symbol_or_msg = future.result()
-                    if success:
-                        success_count += 1
-                    else:
-                        fail_count += 1
-                        failed_symbols.append(symbol_or_msg)  # symbol 저장
-                except Exception as e:
-                    logger.exception(f"[Batch] [{stock_id}] 처리 중 예외 발생: {str(e)}")
-                    fail_count += 1
+        logger.info("[Scheduler] 16:05 배치 예측 시작")
+        start_time = time.time()
+
+        start_id, end_id = 1, 100
+        chunk_size = 20
+        max_workers = 6
+        cooldown_sec = 2
+
+        success_count, fail_count, failed_symbols = BatchService.run_batch_in_chunks(
+            start_id=start_id,
+            end_id=end_id,
+            chunk_size=chunk_size,
+            max_workers=max_workers,
+            cooldown_sec=cooldown_sec
+        )
 
-        duration = time.time() - start
+        duration = time.time() - start_time
         logger.info(
-            f"[Scheduler] 배치 예측 완료 - 성공: {success_count}, 실패: {fail_count}, 소요 시간: {duration:.2f}s"
+            f"[Scheduler] 배치 예측 완료 - 성공: {success_count}, 실패: {fail_count}, 소요: {duration:.2f}s"
         )
 
+        # Discord 알림
         # 스레드에서 비동기 함수 실행
         try:
             import threading
             def run_notification():
-                asyncio.run(notify_discord_async(success_count, fail_count, duration, failed_symbols))
+                asyncio.run(
+                    notify_discord_async(success_count, fail_count, duration, failed_symbols, 1404342496221462539))
 
             notification_thread = threading.Thread(target=run_notification)
             notification_thread.start()

app/services/batch_service.py

@@ -29,16 +29,37 @@ def run_batch_in_chunks(start_id=1, end_id=100, chunk_size=20, max_workers=6, co
         """
         logger.info(f"분할 배치 시작: {start_id} ~ {end_id} (chunk: {chunk_size})")
 
+        success_count = 0
+        fail_count = 0
+        failed_symbols = []
+
         for start in range(start_id, end_id + 1, chunk_size):
             end = min(start + chunk_size - 1, end_id)
 
             logger.info(f"예측 구간: {start} ~ {end}")
-            BatchService.batch_predict_and_save(start_id=start, end_id=end, max_workers=max_workers)
+            with ThreadPoolExecutor(max_workers=max_workers) as executor:
+                futures = {
+                    executor.submit(BatchService.process_single_stock, stock_id): stock_id
+                    for stock_id in range(start, end + 1)
+                }
+                for future in as_completed(futures):
+                    stock_id = futures[future]
+                    try:
+                        stock_id, success, symbol_or_msg = future.result()
+                        if success:
+                            success_count += 1
+                        else:
+                            fail_count += 1
+                            failed_symbols.append(symbol_or_msg)
+                    except Exception as e:
+                        logger.exception(f"[{stock_id}] 처리 중 예외 발생: {str(e)}")
+                        fail_count += 1
 
             if cooldown_sec > 0:
                 time.sleep(cooldown_sec)
 
         logger.info("전체 분할 배치 완료")
+        return success_count, fail_count, failed_symbols
 
     @staticmethod
     def process_single_stock(stock_id: int) -> tuple[int, bool, str]:

app/utils/discord_notifier.py

@@ -5,8 +5,6 @@
 
 logger = logging.getLogger(__name__)
 
-DISCORD_WEBHOOK_URL = os.getenv("DISCORD_WEBHOOK_URL")
-
 
 def format_duration(seconds: float) -> str:
     seconds = int(seconds)
@@ -29,11 +27,15 @@ async def notify_discord_async(
         success_count: int,
         fail_count: int,
         duration: float,
-        failed_symbols: list[str] = None
+        failed_symbols: list[str] = None,
+        mention_role_id: str = None
 ) -> None:
     """
     배치 예측 결과를 디스코드로 전송하는 비동기 함수
     """
+    import os
+    DISCORD_WEBHOOK_URL = os.getenv("DISCORD_WEBHOOK_URL")
+
     if not DISCORD_WEBHOOK_URL:
         logger.warning("DISCORD_WEBHOOK_URL이 설정되지 않았습니다.")
         return
@@ -42,8 +44,11 @@ async def notify_discord_async(
     formatted_duration = format_duration(duration)
     today_str = datetime.now().strftime('%Y-%m-%d')
 
+    # 역할 멘션 문자열 생성
+    mention_str = f"<@&{mention_role_id}> " if mention_role_id else ""
+
     content = f"""
-📊 배치 예측 완료 ({today_str})
+{mention_str}📊 배치 예측 완료 ({today_str})
 - 종목 수: {total}
 - 성공: {success_count}
 - 실패: {fail_count}