stock_chart_site/backend/app/api/symbols.py

"""종목 검색 / 메타 API."""
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/symbols", tags=["symbols"])


@router.get("/search")
def search_symbols(
    q: str = Query(..., min_length=1, max_length=40, description="종목명 또는 코드 prefix/부분 일치"),
    limit: int = Query(default=20, ge=1, le=100),
    seed_only: bool = Query(default=False, description="true 면 학습/배치 대상 10종목만"),
    with_sparkline: bool = Query(
        default=False,
        description="true 면 각 결과에 최근 30 종가 + 전일대비 동봉 (검색 결과 미니차트용)",
    ),
) -> dict:
    """이름은 trigram + ILIKE, 코드는 prefix 매치.

    우선순위:
      1) 코드가 정확히 같으면 가장 위
      2) 이름 prefix 매치
      3) 이름 부분 매치 (trigram similarity)

    with_sparkline=true 시:
      - 매치된 코드들에 대해 ohlcv_daily 최근 30 거래일 종가 한 번에 조회
      - items[*].sparkline (list[float]) + items[*].close + items[*].pct_change 채움
      - 데이터 없는 종목은 sparkline=[], close=None
    """
    q_norm = q.strip()
    if not q_norm:
        raise HTTPException(status_code=400, detail="empty query")

    eng = get_engine()
    where_seed = "AND is_seed = TRUE" if seed_only else ""
    sql = text(
        f"""
        WITH ranked AS (
          SELECT code, name, market, sector, is_seed,
            CASE
              WHEN code = :q                          THEN 0
              WHEN code LIKE :prefix                  THEN 1
              WHEN name LIKE :prefix                  THEN 2
              WHEN name ILIKE :contains               THEN 3
              ELSE 4
            END AS rank,
            similarity(name, :q) AS sim
            FROM symbols
           WHERE active = TRUE
             {where_seed}
             AND (code LIKE :prefix OR name ILIKE :contains OR similarity(name, :q) > 0.2)
        )
        SELECT code, name, market, sector, is_seed
          FROM ranked
         ORDER BY rank ASC, sim DESC, name ASC
         LIMIT :lim
        """
    )
    with eng.connect() as conn:
        rows = conn.execute(
            sql,
            {
                "q": q_norm,
                "prefix": f"{q_norm}%",
                "contains": f"%{q_norm}%",
                "lim": limit,
            },
        ).all()

        codes = [r[0] for r in rows]
        spark_by_code: dict[str, list[float]] = {c: [] for c in codes}
        if with_sparkline and codes:
            # 최근 30 거래일 종가 — 종목별로 ORDER BY date ASC.
            spark_rows = conn.execute(
                text(
                    """
                    SELECT code, date, close FROM (
                      SELECT code, date, close,
                             ROW_NUMBER() OVER (PARTITION BY code ORDER BY date DESC) AS rn
                        FROM ohlcv_daily
                       WHERE code = ANY(:codes) AND close IS NOT NULL
                    ) t
                    WHERE rn <= 30
                    ORDER BY code, date ASC
                    """
                ),
                {"codes": codes},
            ).all()
            for code, _d, close in spark_rows:
                spark_by_code.setdefault(code, []).append(float(close))

    def _item(r) -> dict:
        code = r[0]
        pts = spark_by_code.get(code, []) if with_sparkline else None
        out: dict[str, object] = {
            "code": code,
            "name": r[1],
            "market": r[2],
            "sector": r[3],
            "is_seed": bool(r[4]),
        }
        if with_sparkline:
            out["sparkline"] = pts or []
            out["close"] = pts[-1] if pts else None
            out["pct_change"] = (
                (pts[-1] - pts[-2]) / pts[-2] * 100 if (pts and len(pts) >= 2 and pts[-2]) else None
            )
        return out

    return {
        "q": q_norm,
        "count": len(rows),
        "items": [_item(r) for r in rows],
    }


@router.get("/{code}")
def get_symbol(code: str) -> dict:
    eng = get_engine()
    with eng.connect() as conn:
        row = conn.execute(
            text(
                "SELECT code, name, market, sector, is_seed, active, created_at "
                "FROM symbols WHERE code = :c"
            ),
            {"c": code},
        ).first()
    if not row:
        raise HTTPException(status_code=404, detail=f"unknown code: {code}")
    return {
        "code": row[0],
        "name": row[1],
        "market": row[2],
        "sector": row[3],
        "is_seed": bool(row[4]),
        "active": bool(row[5]),
        "created_at": str(row[6]) if row[6] else None,
    }