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

"""StockAnalysis DB 서비스 — 단일 진실 공급원 (Single Source of Truth)

StockAnalysis ORM 저장 로직을 중앙화하여 3곳의 중복 제거:
  - batch_daily_analysis.py    (bulk)
  - batch_intraday_analysis.py (per-stock, rescore)

PostgreSQL UPSERT(ON CONFLICT DO UPDATE)를 사용하여
DELETE → INSERT 사이의 race condition 을 원자적으로 해결.

워커 패턴 (CLAUDE.md 3.2):
  - session 을 직접 받는 `upsert_analysis` / `bulk_upsert_analysis` 는
    scheduler / batch 에서 기존처럼 사용 가능 (시그니처 불변).
  - 실시간 콜백·인트라데이 분석에서는 `queue_upsert_analysis` /
    `queue_bulk_upsert_analysis` 를 사용하여 DB 락 경합을 방지.
"""
from __future__ import annotations

import logging
from datetime import date, datetime, timezone
from typing import TYPE_CHECKING

from sqlalchemy import select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import AsyncSessionLocal
from app.models.models import StockAnalysis

logger = logging.getLogger(__name__)

if TYPE_CHECKING:
    from app.api.endpoints.recommendations import StockRecommendation


# ─────────────────────────────────────────────────────────────────────────────
# 헬퍼
# ─────────────────────────────────────────────────────────────────────────────
def passes_price_filter(result: "StockRecommendation") -> bool:
    """5,000원 이하 저가주 필터 (CLAUDE.md 5.4).

    True → 통과, False → 필터됨.
    """
    return not (result.current_price and result.current_price < 5000)


# UPSERT 시 갱신할 컬럼 집합 (PK·unique key·최초등록 시각 제외)
_UPSERT_UPDATE_COLS = frozenset({
    "stock_name", "current_price", "change_rate", "volume",
    "rsi", "ma5", "ma20", "ma60",
    "stochastic_k", "stochastic_d",
    "bollinger_upper", "bollinger_lower", "bollinger_middle",
    "signals", "score",
    "ml_score", "ml_confidence", "ml_signal",
    "atr_14", "adx_14",
    "updated_at",
})


def to_analysis_dict(
    stock_code: str,
    stock_name: str,
    result: "StockRecommendation",
    analysis_date: date,
) -> dict:
    """StockRecommendation → StockAnalysis INSERT dict 변환.

    ORM 객체 대신 dict 를 반환하여 단건 UPSERT 와 bulk INSERT 에서 모두 재사용.
    """
    return {
        "stock_code": stock_code,
        "stock_name": stock_name,
        "analysis_date": analysis_date,
        "current_price": result.current_price,
        "change_rate": result.change_rate,
        "volume": result.volume,
        "rsi": result.rsi,
        "ma5": result.ma5,
        "ma20": result.ma20,
        "ma60": result.ma60,
        "stochastic_k": result.stochastic_k,
        "stochastic_d": result.stochastic_d,
        "bollinger_upper": result.bollinger_upper,
        "bollinger_lower": result.bollinger_lower,
        "bollinger_middle": result.bollinger_middle,
        "signals": result.signals,
        "score": result.score,
        "ml_score": result.ml_score,
        "ml_confidence": result.ml_confidence,
        "ml_signal": result.ml_signal,
        "atr_14": result.atr_14,
        "adx_14": result.adx_14,
        "updated_at": datetime.now(timezone.utc).replace(tzinfo=None),
    }


# ─────────────────────────────────────────────────────────────────────────────
# 내부 실행 함수 (db_worker.py 에서도 호출)
# ─────────────────────────────────────────────────────────────────────────────
async def _execute_upsert(session: AsyncSession, rows: list[dict]) -> None:
    """실제 PostgreSQL UPSERT 실행 — session 은 호출 측이 관리."""
    if not rows:
        return
    stmt = pg_insert(StockAnalysis).values(rows)
    stmt = stmt.on_conflict_do_update(
        constraint="uq_stock_analysis_code_date",
        set_={col: stmt.excluded[col] for col in _UPSERT_UPDATE_COLS},
    )
    await session.execute(stmt)


# ─────────────────────────────────────────────────────────────────────────────
# 기존 session 기반 함수 — 시그니처 불변 (CLAUDE.md 4.1)
# ─────────────────────────────────────────────────────────────────────────────
async def upsert_analysis(
    session: AsyncSession,
    stock_code: str,
    stock_name: str,
    result: "StockRecommendation",
    analysis_date: date,
) -> None:
    """단건 UPSERT — session 직접 사용. 배치 스케줄러에서 그대로 호출 가능."""
    row = to_analysis_dict(stock_code, stock_name, result, analysis_date)
    await _execute_upsert(session, [row])


async def bulk_upsert_analysis(
    session: AsyncSession,
    items: list[tuple[str, str, "StockRecommendation"]],
    analysis_date: date,
) -> int:
    """다건 UPSERT — session 직접 사용. 일봉 배치에서 그대로 호출 가능."""
    if not items:
        return 0
    rows = [to_analysis_dict(code, name, r, analysis_date) for code, name, r in items]
    await _execute_upsert(session, rows)
    return len(rows)


# ─────────────────────────────────────────────────────────────────────────────
# NEW: 워커 큐 기반 함수 — 실시간 콜백·인트라데이에서 사용 (CLAUDE.md 3.2)
# ─────────────────────────────────────────────────────────────────────────────
async def queue_upsert_analysis(
    stock_code: str,
    stock_name: str,
    result: "StockRecommendation",
    analysis_date: date,
) -> None:
    """단건 UPSERT 를 db_worker 큐에 위임.

    실시간 스레드 / 인트라데이 분석에서 사용.
    DB 세션을 직접 열지 않으므로 락 경합 없음.
    """
    from app.services.db_worker import enqueue_upsert_analysis
    row = to_analysis_dict(stock_code, stock_name, result, analysis_date)
    await enqueue_upsert_analysis(row)


async def queue_bulk_upsert_analysis(
    items: list[tuple[str, str, "StockRecommendation"]],
    analysis_date: date,
) -> int:
    """다건 UPSERT 를 db_worker 큐에 위임.

    배치 분석 결과를 비동기로 저장할 때 사용.
    API 응답 지연 없이 큐에 넣고 즉시 반환.
    Returns: 큐에 등록된 row 수
    """
    if not items:
        return 0
    from app.services.db_worker import enqueue_bulk_upsert_analysis
    rows = [to_analysis_dict(code, name, r, analysis_date) for code, name, r in items]
    await enqueue_bulk_upsert_analysis(rows)
    return len(rows)


# ─────────────────────────────────────────────────────────────────────────────
# DB 스캐너 — 저장된 분석 결과로 신호 종목 필터링 (API 호출 없음)
# ─────────────────────────────────────────────────────────────────────────────
async def run_db_scanner(analysis_date: date) -> set[str]:
    """배치 DB 기반 코스닥·코스피 스캐너 (추가 API 호출 없음).

    조건:
      - RSI(14) ≤ 30  (과매도)
      - 종가 ≤ BB 하단 × 1.015
      - 시세 ≥ 5,000원
      - ADX ≤ 30  (강한 추세 = 평균회귀 부적합)
      - MA5 ≥ MA20 × 0.95  (강한 하락 추세 제외)

    Returns: 신호 종목 코드 set
    """
    try:
        async with AsyncSessionLocal() as session:
            result = await session.execute(
                select(
                    StockAnalysis.stock_code,
                    StockAnalysis.current_price,
                    StockAnalysis.rsi,
                    StockAnalysis.bollinger_lower,
                    StockAnalysis.adx_14,
                    StockAnalysis.ma5,
                    StockAnalysis.ma20,
                ).where(
                    StockAnalysis.analysis_date == analysis_date,
                    StockAnalysis.current_price >= 5000,
                    StockAnalysis.rsi.isnot(None),
                    StockAnalysis.rsi <= 30,
                    StockAnalysis.bollinger_lower.isnot(None),
                )
            )
            rows = result.all()

        signal_tickers: set[str] = set()
        for row in rows:
            price, bb_lower = row.current_price, row.bollinger_lower
            if not bb_lower or bb_lower == 0:
                continue

            pct = (price - bb_lower) / bb_lower * 100
            if pct > 1.5:
                continue

            adx = row.adx_14
            if adx is not None and adx > 30:
                continue

            ma5, ma20 = row.ma5, row.ma20
            if ma5 is not None and ma20 is not None and ma20 > 0:
                if float(ma5) < float(ma20) * 0.95:
                    continue

            signal_tickers.add(row.stock_code)

        logger.info(
            "🔍 DB 스캐너: %d개 신호 종목 (RSI≤30 + BB하단 + ADX≤30 + MA방향)",
            len(signal_tickers),
        )
        return signal_tickers

    except Exception as exc:
        logger.error("❌ DB 스캐너 실패: %s", exc, exc_info=True)
        return set()