stock_chart_site/backend/app/api/markets.py

"""시장 overview API.

GET /api/markets/movers
  ?market=KOSPI|KOSDAQ|ALL  (기본 ALL)
  ?limit=10                  (각 카테고리 별 상위 N)

응답 (3 카테고리):
  - gainers: 등락률 상위 (전일종가 대비 상승%)
  - losers : 등락률 하위 (하락%)
  - volume : 거래량 상위 (당일 volume)

ohlcv_daily 에서 가장 최근 2 거래일을 잡아 비교한다. 데이터가 한 종목당
2 일치 이상 없을 때는 그 종목은 자연 누락된다.

GET /api/markets/related/{code}
  ?limit=8
  같은 market 에서 등락률이 비슷한 (또는 가장 활발한) 종목 N 개.
  sector 컬럼은 비어있는 경우가 많아, market 동질성만으로 묶는다.
"""
from __future__ import annotations

from fastapi import APIRouter, HTTPException, Query
from sqlalchemy import text

from app.db.connection import get_engine

router = APIRouter(prefix="/api/markets", tags=["markets"])


def _movers_sql(where_market: str) -> str:
    """ohlcv_daily 에서 종목별 최신 2일 비교 + 카테고리 분류.

    market filter 는 symbols 테이블에서 적용. WITH … AS 안에 :market 바인딩이
    있는 단일 쿼리.
    """
    return f"""
        WITH last_dates AS (
          SELECT DISTINCT date
            FROM ohlcv_daily
           ORDER BY date DESC
           LIMIT 2
        ),
        ordered AS (
          SELECT MAX(date) AS d0, MIN(date) AS d1 FROM last_dates
        ),
        symb AS (
          SELECT code, name, market FROM symbols
           WHERE active = TRUE {where_market}
        ),
        latest AS (
          SELECT o.code, o.close AS close0, o.volume AS volume0
            FROM ohlcv_daily o, ordered
           WHERE o.date = ordered.d0
        ),
        prev AS (
          SELECT o.code, o.close AS close1
            FROM ohlcv_daily o, ordered
           WHERE o.date = ordered.d1
        )
        SELECT s.code, s.name, s.market,
               l.close0   AS close,
               p.close1   AS prev_close,
               l.volume0  AS volume,
               CASE WHEN p.close1 > 0
                    THEN (l.close0 - p.close1) / p.close1 * 100
                    ELSE NULL
               END AS pct_change
          FROM symb s
          JOIN latest l ON l.code = s.code
          JOIN prev   p ON p.code = s.code
         WHERE l.close0 IS NOT NULL AND p.close1 IS NOT NULL
    """


@router.get("/movers")
def movers(
    market: str = Query(default="ALL", pattern="^(ALL|KOSPI|KOSDAQ)$"),
    limit: int = Query(default=10, ge=1, le=50),
) -> dict:
    where_market = "" if market == "ALL" else "AND market = :market"
    params: dict[str, object] = {"lim": limit}
    if market != "ALL":
        params["market"] = market

    eng = get_engine()
    base_sql = _movers_sql(where_market)
    with eng.connect() as conn:
        gainers = conn.execute(
            text(base_sql + " ORDER BY pct_change DESC NULLS LAST LIMIT :lim"),
            params,
        ).all()
        losers = conn.execute(
            text(base_sql + " ORDER BY pct_change ASC NULLS LAST LIMIT :lim"),
            params,
        ).all()
        by_volume = conn.execute(
            text(base_sql + " ORDER BY volume DESC NULLS LAST LIMIT :lim"),
            params,
        ).all()

    def _row(r) -> dict:
        return {
            "code": r[0],
            "name": r[1],
            "market": r[2],
            "close": float(r[3]) if r[3] is not None else None,
            "prev_close": float(r[4]) if r[4] is not None else None,
            "volume": int(r[5]) if r[5] is not None else None,
            "pct_change": float(r[6]) if r[6] is not None else None,
        }

    return {
        "market": market,
        "limit": limit,
        "gainers": [_row(r) for r in gainers],
        "losers": [_row(r) for r in losers],
        "by_volume": [_row(r) for r in by_volume],
    }


@router.get("/indices")
def indices(days: int = Query(default=60, ge=2, le=400)) -> dict:
    """KOSPI / KOSDAQ 지수 N일 종가 시계열.

    macro_daily 에 yfinance 가 채우는 'kospi','kosdaq' key 를 그대로 읽음.
    채워져 있지 않으면 series 가 빈 배열로 반환되어 프런트가 '데이터 준비 중' 안내.
    """
    sql = text(
        """
        SELECT date, key, value
          FROM macro_daily
         WHERE key IN ('kospi','kosdaq')
           AND date >= (CURRENT_DATE - (:d || ' day')::interval)
         ORDER BY date ASC
        """
    )
    eng = get_engine()
    with eng.connect() as conn:
        rows = conn.execute(sql, {"d": days}).all()

    series: dict[str, list[dict]] = {"kospi": [], "kosdaq": []}
    for r in rows:
        d, k, v = r[0], r[1], r[2]
        if v is None:
            continue
        series.setdefault(k, []).append({"date": str(d), "value": float(v)})

    def _summary(pts: list[dict]) -> dict:
        if not pts:
            return {"latest": None, "prev": None, "pct_change": None}
        latest = pts[-1]["value"]
        prev = pts[-2]["value"] if len(pts) >= 2 else None
        pct = ((latest - prev) / prev * 100) if (prev and prev > 0) else None
        return {"latest": latest, "prev": prev, "pct_change": pct}

    return {
        "days": days,
        "indices": [
            {
                "key": "kospi",
                "name": "KOSPI",
                "points": series["kospi"],
                **_summary(series["kospi"]),
            },
            {
                "key": "kosdaq",
                "name": "KOSDAQ",
                "points": series["kosdaq"],
                **_summary(series["kosdaq"]),
            },
        ],
    }


@router.get("/related/{code}")
def related(code: str, limit: int = Query(default=8, ge=1, le=30)) -> dict:
    """같은 market 에서 등락률 절대값 기준 가까운 종목 N 개.

    sector 컬럼은 시드 단계에서 NULL 인 경우가 많아 market 만으로 동질성을 잡는다.
    백엔드가 sector 분류를 채우게 되면 같은 sector 우선 정렬로 강화할 수 있음.
    """
    eng = get_engine()
    with eng.connect() as conn:
        target = conn.execute(
            text("SELECT market FROM symbols WHERE code = :c"),
            {"c": code},
        ).first()
        if not target:
            raise HTTPException(status_code=404, detail=f"unknown code: {code}")
        market = target[0]

        rows = conn.execute(
            text(
                _movers_sql("AND market = :market AND code <> :c")
                + " ORDER BY volume DESC NULLS LAST LIMIT :lim"
            ),
            {"market": market, "c": code, "lim": limit},
        ).all()

    return {
        "code": code,
        "market": market,
        "items": [
            {
                "code": r[0],
                "name": r[1],
                "market": r[2],
                "close": float(r[3]) if r[3] is not None else None,
                "prev_close": float(r[4]) if r[4] is not None else None,
                "volume": int(r[5]) if r[5] is not None else None,
                "pct_change": float(r[6]) if r[6] is not None else None,
            }
            for r in rows
        ],
    }