Merge pull request #42 from SynergyX-AI-Pattern/refactor/#41_predict_batch_avg

[REFACTOR] 예측 배치에 top 20 계산 및 저장 로직 추가 #41

Author: eldeoddt

app/core/logging_config.py

@@ -10,4 +10,5 @@ def setup_logging():
         level=LOG_LEVEL,
         format=LOG_FORMAT,
         stream=sys.stdout,
+        force=True,
     )

app/db/base.py

@@ -1,3 +1,8 @@
-from sqlalchemy.orm import declarative_base
+from app.db.base_class import Base
 
-Base = declarative_base()
+from app.models.stock import Stock
+from app.models.stock_detail import StockDetail
+from app.models.prediction import Prediction
+from app.models.pattern import Pattern
+from app.models.pattern_apply import PatternApply
+from app.models.pattern_detection_log import PatternDetectionLog

app/db/base_class.py

@@ -0,0 +1,3 @@
+from sqlalchemy.orm import declarative_base
+
+Base = declarative_base()

app/db/session.py

@@ -1,16 +1,21 @@
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 from app.core.config import settings
+from app.db.base import Base
 
+
+# DB 엔진 생성
 engine = create_engine(
     settings.DATABASE_URL,
-    pool_pre_ping=True,  # 연결 확인 (죽은 연결 자동 제거)
-    echo=False  # SQL 로그 사용
+    pool_pre_ping=True,
+    echo=False,
 )
 
+# 세션팩토리
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
 
+# 의존성 주입용 (FastAPI DI)
 def get_db():
     db = SessionLocal()
     try:

app/models/base_model.py

@@ -1,6 +1,6 @@
 from sqlalchemy import Column, DateTime
 from sqlalchemy.sql import func
-from app.db.base import Base
+from app.db.base_class import Base
 
 
 class BaseTimeModel(Base):

app/models/stock.py

@@ -14,3 +14,4 @@ class Stock(BaseTimeModel):
     # 관계 설정
     predictions = relationship("Prediction", back_populates="stock")
     pattern_applies = relationship("PatternApply", back_populates="stock")
+    stock_detail = relationship("StockDetail", back_populates="stock", uselist=False)

app/models/stock_detail.py

@@ -0,0 +1,21 @@
+from sqlalchemy import Column, BigInteger, Float, JSON, ForeignKey
+from app.models.base_model import BaseTimeModel
+from sqlalchemy.orm import relationship, Mapped
+
+class StockDetail(BaseTimeModel):
+    __tablename__ = "stock_detail"
+
+    id = Column(BigInteger, primary_key=True, autoincrement=True)
+    stock_id = Column(BigInteger, ForeignKey("stock.id"), nullable=False, unique=True)
+
+    price = Column(Float, nullable=True)
+    change_rate = Column(Float, nullable=True)
+    financial_data = Column(JSON, nullable=True)
+    change_amount = Column(Float, nullable=True)
+    version = Column(BigInteger, nullable=True)
+
+    ai_avg_increase = Column(Float, nullable=True, default=None)
+    ai_rank = Column(BigInteger, nullable=True, default=None)
+
+    # 관계
+    stock = relationship("Stock", back_populates="stock_detail")

app/schedulers/predict_batch_runner.py

@@ -35,6 +35,9 @@ def run_batch():
             cooldown_sec=cooldown_sec
         )
 
+        # 평균 상승률, 랭크 계산
+        top3_info = BatchService.update_ai_avg_increase_and_rank()
+
         duration = time.time() - start_time
         logger.info(
             f"[Scheduler] 배치 예측 완료 - 성공: {success_count}, 실패: {fail_count}, 소요: {duration:.2f}s"
@@ -46,7 +49,7 @@ def run_batch():
             import threading
             def run_notification():
                 asyncio.run(
-                    notify_discord_async(success_count, fail_count, duration, failed_symbols, 1404342496221462539))
+                    notify_discord_async(success_count, fail_count, duration, failed_symbols, 1404342496221462539, top3_info=top3_info,))
 
             notification_thread = threading.Thread(target=run_notification)
             notification_thread.start()

app/services/batch_service.py

@@ -1,8 +1,12 @@
 from concurrent.futures import ThreadPoolExecutor, as_completed
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, aliased
+from sqlalchemy import func
 from app.db.session import SessionLocal
 from app.models.stock import Stock
+from app.models.stock_detail import StockDetail
+from app.models.prediction import Prediction
 import time
+import datetime
 from app.crud.prediction import create_prediction_objects, save_predictions
 from app.services.prediction_service import PredictionService
 import logging
@@ -122,3 +126,126 @@ def batch_predict_and_save(start_id: int, end_id: int, max_workers: int = 2):
                     logger.exception(f"[{stock_id}] 처리 중 예외 발생: {str(e)}")
 
         logger.info("전체 배치 완료")
+
+    @staticmethod
+    def update_ai_avg_increase_and_rank(base_date: datetime.date | None = None):
+        """
+        종목별 예측값 기반 실제 존재하는 예측일 기준 15일 평균 상승률 계산 후 stock_detail 업데이트
+        :param base_date: 기준일 (None이면 오늘 날짜 사용)
+        :return: None
+        """
+        db = SessionLocal()
+        try:
+            today = base_date or datetime.date.today()
+            logger.info(f"[BatchService] 평균 상승률 계산 시작 (기준일: {today})")
+
+            # --- 예측 데이터 존재 여부 확인 ---
+            total_preds = db.query(Prediction).count()
+            future_preds = db.query(Prediction).filter(Prediction.target_date > today).count()
+            logger.info(f"[BatchService] Prediction 전체: {total_preds}건 / 기준일 이후: {future_preds}건")
+
+            if future_preds == 0:
+                logger.warning(f"[BatchService] 기준일({today}) 이후 예측 데이터가 없습니다.")
+                return
+
+            # --- 첫 번째 예측일 구하기 ---
+            first_date_subq = (
+                db.query(
+                    Prediction.stock_id,
+                    func.min(Prediction.target_date).label("first_target_date")
+                )
+                .filter(Prediction.target_date > today)
+                .group_by(Prediction.stock_id)
+                .subquery()
+            )
+            logger.debug(f"[BatchService] first_date_subq 생성 완료")
+
+            # --- 첫 예측일의 예측 종가 구하기 ---
+            subquery_first_price = (
+                db.query(
+                    Prediction.stock_id,
+                    Prediction.predicted_close.label("first_predicted_close")
+                )
+                .join(
+                    first_date_subq,
+                    (Prediction.stock_id == first_date_subq.c.stock_id)
+                    & (Prediction.target_date == first_date_subq.c.first_target_date)
+                )
+                .subquery()
+            )
+            first_price_count = db.query(subquery_first_price).count()
+            logger.info(f"[BatchService] 첫 예측일 종가 매핑 완료 ({first_price_count}건)")
+
+            # --- 평균 상승률 계산 ---
+            results = (
+                db.query(
+                    Prediction.stock_id,
+                    func.avg(
+                        (Prediction.predicted_close - subquery_first_price.c.first_predicted_close)
+                        / func.nullif(subquery_first_price.c.first_predicted_close, 0.0)
+                    ).label("avg_increase")
+                )
+                .join(subquery_first_price, Prediction.stock_id == subquery_first_price.c.stock_id)
+                .filter(Prediction.target_date > today)
+                .group_by(Prediction.stock_id)
+                .all()
+            )
+
+            logger.info(f"[BatchService] 평균 상승률 계산 결과: {len(results)}건")
+
+            if not results:
+                logger.warning("[BatchService] 평균 상승률 계산 결과 없음 (JOIN 또는 데이터 매칭 문제 가능)")
+                return
+
+            # --- 상위 3개 샘플 출력 ---
+            sample_logs = [
+                f"stock_id={r.stock_id}, avg_increase={round((r.avg_increase or 0) * 100, 2)}%"
+                for r in results[:3]
+            ]
+            logger.debug(f"[BatchService] 계산 결과 샘플: {sample_logs}")
+
+            # --- 평균 상승률 내림차순 정렬 및 랭킹 부여 ---
+            sorted_results = sorted(results, key=lambda r: r.avg_increase or 0, reverse=True)
+
+            for rank, row in enumerate(sorted_results, start=1):
+                db.query(StockDetail).filter(StockDetail.stock_id == row.stock_id).update(
+                    {
+                        StockDetail.ai_avg_increase: row.avg_increase,
+                        StockDetail.ai_rank: rank,
+                        StockDetail.updated_at: datetime.datetime.utcnow(),
+                    }
+                )
+                if rank <= 3:  # 상위 3개만 출력
+                    logger.debug(
+                        f"[BatchService] UPDATE → stock_id={row.stock_id}, "
+                        f"avg_increase={row.avg_increase}, rank={rank}"
+                    )
+
+            db.commit()
+            logger.info(f"[BatchService] 평균 상승률 및 랭킹 갱신 완료 ({len(sorted_results)}개 종목)")
+
+            # 상위 3개 종목 조회
+            top3 = (
+                db.query(Stock.name, StockDetail.ai_avg_increase)
+                .join(StockDetail, Stock.id == StockDetail.stock_id)
+                .order_by(StockDetail.ai_rank.asc())
+                .limit(3)
+                .all()
+            )
+
+            # 디스코드 알림용 리스트 반환
+            top3_info = [
+                {"name": name, "increase": round((increase or 0) * 100, 2)}
+                for name, increase in top3
+            ]
+
+            return top3_info
+
+        except Exception as e:
+            db.rollback()
+            logger.exception(f"[BatchService] 평균 상승률 계산 중 오류 발생: {e}")
+            raise
+
+        finally:
+            db.close()
+            logger.debug("[BatchService] 세션 종료 완료")

app/tests/services/test_batch_service.py

@@ -0,0 +1,41 @@
+import os
+import datetime
+from dotenv import load_dotenv
+import logging
+
+# --- FastAPI와 동일한 로깅 설정 불러오기 ---
+from app.core.logging_config import setup_logging
+setup_logging()
+
+# --- 환경 변수 로드 ---
+BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))
+dotenv_path = os.path.join(BASE_DIR, ".env")
+load_dotenv(dotenv_path)
+
+from app.db.session import SessionLocal
+from app.services.batch_service import BatchService
+
+
+def test_update_ai_avg_increase_and_rank_real_db():
+    """
+    [통합테스트] 실제 DB 기준으로 평균 상승률 계산 및 stock_detail 업데이트 테스트
+    """
+    db = SessionLocal()
+
+    # 테스트용 기준일 (직접 지정 가능)
+    base_date = datetime.date(2025, 10, 13)
+
+    print(f"[TEST] 기준일: {base_date}")
+
+    try:
+        BatchService.update_ai_avg_increase_and_rank(base_date)
+        db.commit()
+        print("[SUCCESS] 평균 상승률 갱신 완료")
+
+    except Exception as e:
+        db.rollback()
+        print(f"[ERROR] 테스트 중 예외 발생: {e}")
+        raise
+
+    finally:
+        db.close()