66a28cc7ca
- seed/themes.py: 반도체/2차전지/바이오/게임/엔터/자동차/금융/조선·방산/플랫폼·AI/유통·소비재/에너지 정적 매핑
- /api/themes (인덱스) + /api/themes/{slug} (종목 카드 + 최신 종가/등락)
- 홈에 ThemeTiles 그리드, /themes/{slug} 상세 페이지
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
112 lines
3.6 KiB
Python
112 lines
3.6 KiB
Python
"""테마/카테고리 API.
|
|
|
|
토스 '주식 골라보기' 의 테마 그리드와 비슷한 결.
|
|
정적 매핑(seed/themes.py)을 베이스로, DB 에서 최신 종가/등락만 채워 응답.
|
|
|
|
GET /api/themes
|
|
- 모든 테마 인덱스 (slug/name/description/code_count)
|
|
|
|
GET /api/themes/{slug}
|
|
?limit=20
|
|
- 해당 테마의 종목 카드 (종목명/시장/종가/등락률). 거래량 기준 정렬.
|
|
- ohlcv_daily 가 아직 안 채워져 있으면 close/pct_change 가 NULL.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
from sqlalchemy import text
|
|
|
|
from app.db.connection import get_engine
|
|
from app.seed.themes import THEMES, theme_by_slug, themes_index
|
|
|
|
router = APIRouter(prefix="/api/themes", tags=["themes"])
|
|
|
|
|
|
@router.get("")
|
|
def list_themes() -> dict:
|
|
return {"items": themes_index(), "total": len(THEMES)}
|
|
|
|
|
|
@router.get("/{slug}")
|
|
def get_theme(slug: str, limit: int = Query(default=30, ge=1, le=100)) -> dict:
|
|
th = theme_by_slug(slug)
|
|
if th is None:
|
|
raise HTTPException(status_code=404, detail=f"unknown theme: {slug}")
|
|
|
|
codes = list(th.codes)[:limit]
|
|
if not codes:
|
|
return {
|
|
"slug": th.slug,
|
|
"name": th.name,
|
|
"description": th.description,
|
|
"items": [],
|
|
}
|
|
|
|
sql = text(
|
|
"""
|
|
WITH last2 AS (
|
|
SELECT DISTINCT date FROM ohlcv_daily ORDER BY date DESC LIMIT 2
|
|
),
|
|
bounds AS (SELECT MAX(date) AS d0, MIN(date) AS d1 FROM last2),
|
|
latest AS (
|
|
SELECT o.code, o.close AS close0, o.volume AS vol0
|
|
FROM ohlcv_daily o, bounds WHERE o.date = bounds.d0
|
|
),
|
|
prev AS (
|
|
SELECT o.code, o.close AS close1
|
|
FROM ohlcv_daily o, bounds WHERE o.date = bounds.d1
|
|
)
|
|
SELECT s.code, s.name, s.market,
|
|
l.close0 AS close,
|
|
p.close1 AS prev_close,
|
|
l.vol0 AS volume,
|
|
CASE WHEN p.close1 > 0 THEN (l.close0 - p.close1) / p.close1 * 100 ELSE NULL END AS pct_change
|
|
FROM symbols s
|
|
LEFT JOIN latest l ON l.code = s.code
|
|
LEFT JOIN prev p ON p.code = s.code
|
|
WHERE s.code = ANY(:codes)
|
|
ORDER BY vol0 DESC NULLS LAST, s.name ASC
|
|
"""
|
|
)
|
|
|
|
eng = get_engine()
|
|
with eng.connect() as conn:
|
|
rows = conn.execute(sql, {"codes": codes}).all()
|
|
|
|
# symbols 테이블에 시드만 있고 codes 일부가 누락된 경우, 그래도 코드/이름 자리는 비워서 카드 표시.
|
|
by_code = {r[0]: r for r in rows}
|
|
items: list[dict] = []
|
|
for c in codes:
|
|
r = by_code.get(c)
|
|
if r is None:
|
|
items.append(
|
|
{
|
|
"code": c,
|
|
"name": c, # symbols 미등재 — 코드로 노출
|
|
"market": None,
|
|
"close": None,
|
|
"prev_close": None,
|
|
"volume": None,
|
|
"pct_change": None,
|
|
}
|
|
)
|
|
else:
|
|
items.append(
|
|
{
|
|
"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 {
|
|
"slug": th.slug,
|
|
"name": th.name,
|
|
"description": th.description,
|
|
"items": items,
|
|
}
|