85da979fbe
- backend/app/seed/themes.py: themes_for_code(code) 헬퍼 추가 (코드 → 소속 테마들)
- backend/app/api/themes.py: GET /api/themes/peer/{code} 신규
· 종목이 속한 모든 테마의 코드 합집합을 peer 풀로
· 단일 SQL (ROW_NUMBER OVER PARTITION BY code) 로 21영업일 누적수익률 일괄 계산
· 본인 / peer 평균 / 상·하위 5종목 반환
- web/lib/api.ts: PeerCompare 타입 + api.peerCompare()
- web/components/PeerComparePanel.tsx: 테마 태그 + 3-칸 요약 (본인/평균/격차) + 상·하위 5
- [code]/page.tsx 에 PeriodReturns 아래 부착. 테마 무소속 종목은 패널 자동 숨김.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
218 lines
7.0 KiB
Python
218 lines
7.0 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_for_code, themes_index
|
|
|
|
router = APIRouter(prefix="/api/themes", tags=["themes"])
|
|
|
|
|
|
@router.get("")
|
|
def list_themes() -> dict:
|
|
return {"items": themes_index(), "total": len(THEMES)}
|
|
|
|
|
|
@router.get("/peer/{code}")
|
|
def peer_compare(
|
|
code: str,
|
|
window: int = Query(default=21, ge=2, le=252),
|
|
limit: int = Query(default=5, ge=1, le=20),
|
|
) -> dict:
|
|
"""동일 테마(들) 내 동종목 비교.
|
|
|
|
- 대상 종목이 속한 모든 테마의 코드 합집합 (자기 자신 제외) 을 peer 풀로 잡고,
|
|
각 종목의 최근 거래일 종가 / window 영업일 전 종가로 누적수익률을 구한다.
|
|
- peer 평균, 대상 종목 값, 상/하위 N 개 반환.
|
|
- 테마 무소속이면 themes=[] 빈 응답 (404 아님 — UI 가 그냥 패널을 숨기게).
|
|
"""
|
|
themes = themes_for_code(code)
|
|
if not themes:
|
|
return {
|
|
"code": code,
|
|
"window": window,
|
|
"themes": [],
|
|
"self_pct": None,
|
|
"peer_avg_pct": None,
|
|
"top": [],
|
|
"bottom": [],
|
|
}
|
|
|
|
# 테마 합집합 코드 풀 (자기 자신 제외)
|
|
peer_codes: set[str] = set()
|
|
for t in themes:
|
|
peer_codes.update(t.codes)
|
|
peer_codes.discard(code)
|
|
pool = [code, *sorted(peer_codes)]
|
|
|
|
sql = text(
|
|
"""
|
|
WITH ranked AS (
|
|
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
|
|
),
|
|
latest AS (
|
|
SELECT code, close AS close_now, date AS date_now FROM ranked WHERE rn = 1
|
|
),
|
|
base AS (
|
|
SELECT code, close AS close_base, date AS date_base FROM ranked WHERE rn = :w
|
|
)
|
|
SELECT s.code, s.name, s.market,
|
|
l.close_now, b.close_base,
|
|
CASE WHEN b.close_base > 0
|
|
THEN (l.close_now - b.close_base) / b.close_base * 100
|
|
ELSE NULL
|
|
END AS pct
|
|
FROM symbols s
|
|
LEFT JOIN latest l ON l.code = s.code
|
|
LEFT JOIN base b ON b.code = s.code
|
|
WHERE s.code = ANY(:codes)
|
|
"""
|
|
)
|
|
|
|
eng = get_engine()
|
|
with eng.connect() as conn:
|
|
rows = conn.execute(sql, {"codes": pool, "w": window}).all()
|
|
|
|
by_code: dict[str, dict] = {}
|
|
for r in rows:
|
|
by_code[r[0]] = {
|
|
"code": r[0],
|
|
"name": r[1],
|
|
"market": r[2],
|
|
"close": float(r[3]) if r[3] is not None else None,
|
|
"base_close": float(r[4]) if r[4] is not None else None,
|
|
"pct_change": float(r[5]) if r[5] is not None else None,
|
|
}
|
|
|
|
self_row = by_code.get(code)
|
|
self_pct = self_row["pct_change"] if self_row else None
|
|
|
|
peers = [
|
|
v for k, v in by_code.items() if k != code and v["pct_change"] is not None
|
|
]
|
|
if peers:
|
|
peer_avg_pct = sum(p["pct_change"] for p in peers) / len(peers)
|
|
peers_sorted = sorted(peers, key=lambda x: x["pct_change"], reverse=True)
|
|
top = peers_sorted[:limit]
|
|
bottom = peers_sorted[-limit:][::-1]
|
|
else:
|
|
peer_avg_pct = None
|
|
top = []
|
|
bottom = []
|
|
|
|
return {
|
|
"code": code,
|
|
"name": self_row["name"] if self_row else code,
|
|
"window": window,
|
|
"themes": [
|
|
{"slug": t.slug, "name": t.name, "code_count": len(t.codes)}
|
|
for t in themes
|
|
],
|
|
"self_pct": self_pct,
|
|
"peer_avg_pct": peer_avg_pct,
|
|
"peer_count": len(peers),
|
|
"top": top,
|
|
"bottom": bottom,
|
|
}
|
|
|
|
|
|
@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,
|
|
}
|