feat(orderbook): 10단계 호가 패널 (KIS asking-price 스냅샷)

backend:
- fetch/kis.py: fetch_orderbook(code) — KIS FHKST01010200 (inquire-asking-price-exp-ccn)
  호출. output1 의 askp1~10/bidp1~10 + 잔량 + 합계, output2 의 현재가/전일대비 파싱.
- api/orderbook.py: GET /api/orderbook/{code}. symbols 매칭 안 되면 404, KIS 키 없으면
  skipped_missing_key 상태로 빈 호가 반환 (502 와 구분).
- main.py: orderbook_router 등록.

frontend:
- lib/api.ts: OrderbookResponse 타입 + api.orderbook(code).
- components/OrderbookPanel.tsx: 매도 10단계(위→아래 가격 내림차순) / 현재가(굵게, 전일대비)
  / 매수 10단계. 잔량 막대는 사이드별 max 정규화로 셀 안에서 자라남. ask=파랑, bid=빨강
  (한국 거래소 관행). 10초 폴링.
- app/[code]/page.tsx: 사이드바에 SymbolSidebar 다음으로 OrderbookPanel 마운트.

실시간 ws 가 아니라 HTTP 폴링이라 KIS rate-limit 친화적. 표시 전용 — 주문/체결 endpoint 는
일절 import 하지 않음 (kis.py 문서 정책).
This commit is contained in:
claude-owner
2026-05-28 19:13:35 +09:00
parent 1a6d953d17
commit fed8ff287e
6 changed files with 362 additions and 0 deletions

View File

@@ -0,0 +1,60 @@
"""호가(orderbook) API — KIS 10단계 호가 스냅샷.
장중에 호출하면 호가 + 잔량 + 현재가가 한 번의 KIS 호출로 따라온다. 실시간 ws 가
아니라 폴링 친화 (프론트가 5~10s 주기로 재호출) 설계 — 표시 전용.
"""
from __future__ import annotations
from fastapi import APIRouter, HTTPException
from sqlalchemy import text
from app.db.connection import get_engine
from app.fetch.kis import SkippedMissingKey, fetch_orderbook
router = APIRouter(prefix="/api/orderbook", tags=["orderbook"])
@router.get("/{code}")
def get_orderbook(code: str) -> dict:
"""code 의 10단계 호가 스냅샷.
- symbols 에 없는 코드는 404.
- KIS 키 미설정이면 status="skipped_missing_key" + 빈 호가 (프론트에서 안내).
- KIS 호출 실패는 502.
"""
eng = get_engine()
with eng.connect() as conn:
sym = conn.execute(
text("SELECT code, name, market FROM symbols WHERE code = :c"),
{"c": code},
).first()
if not sym:
raise HTTPException(status_code=404, detail=f"unknown code: {code}")
try:
snap = fetch_orderbook(code)
except SkippedMissingKey:
return {
"code": sym[0],
"name": sym[1],
"market": sym[2],
"status": "skipped_missing_key",
"ts": None,
"current": None,
"prev_close_diff": None,
"prev_close_pct": None,
"asks": [],
"bids": [],
"ask_total": 0,
"bid_total": 0,
}
except Exception as exc: # noqa: BLE001
raise HTTPException(status_code=502, detail=f"kis orderbook failed: {exc}") from exc
return {
"code": sym[0],
"name": sym[1],
"market": sym[2],
"status": "ok",
**snap,
}

View File

@@ -357,6 +357,95 @@ def fetch_minute_range(
return sorted(accumulated.values(), key=lambda r: r["ts"])
@retry(
stop=stop_after_attempt(2),
wait=wait_exponential(multiplier=1, min=1, max=4),
retry=retry_if_exception_type((httpx.HTTPError, httpx.TimeoutException)),
reraise=True,
)
def fetch_orderbook(code: str) -> dict[str, Any]:
"""10단계 호가 + 잔량 + 현재가 스냅샷 (read-only).
KIS `inquire-asking-price-exp-ccn` (TR FHKST01010200) 호출. 실시간 ws 가 아니라
HTTP 요청 시점의 호가 스냅샷이므로 프론트에서 5~10초 폴링이 자연스럽다.
Returns:
{
"ts": "HH:MM:SS" (KIS 가 준 aspr_acpt_hour),
"current": float | None, # stck_prpr
"prev_close_diff": float | None,
"prev_close_pct": float | None,
"asks": [{level, price, qty}] (level 1..10, 1=가장 낮은 매도호가),
"bids": [{level, price, qty}] (level 1..10, 1=가장 높은 매수호가),
"ask_total": int,
"bid_total": int,
}
"""
if not _has_keys():
raise SkippedMissingKey("kis app_key/secret missing")
url = f"{KIS_BASE}/uapi/domestic-stock/v1/quotations/inquire-asking-price-exp-ccn"
params = {
"FID_COND_MRKT_DIV_CODE": "J",
"FID_INPUT_ISCD": code,
}
with httpx.Client(timeout=10.0) as cli:
resp = cli.get(url, headers=_headers("FHKST01010200"), params=params)
resp.raise_for_status()
data = resp.json()
if data.get("rt_cd") != "0":
raise RuntimeError(f"kis error: {data.get('msg1')} (rt_cd={data.get('rt_cd')})")
o1 = (data.get("output1") or {}) if isinstance(data.get("output1"), dict) else {}
o2 = (data.get("output2") or {}) if isinstance(data.get("output2"), dict) else {}
def _f(d: dict[str, Any], k: str) -> float | None:
v = d.get(k)
if v in (None, ""):
return None
try:
return float(v)
except (TypeError, ValueError):
return None
def _i(d: dict[str, Any], k: str) -> int:
v = d.get(k)
if v in (None, ""):
return 0
try:
return int(float(v))
except (TypeError, ValueError):
return 0
asks = []
bids = []
for lv in range(1, 11):
ap = _f(o1, f"askp{lv}")
aq = _i(o1, f"askp_rsqn{lv}")
bp = _f(o1, f"bidp{lv}")
bq = _i(o1, f"bidp_rsqn{lv}")
if ap is not None and ap > 0:
asks.append({"level": lv, "price": ap, "qty": aq})
if bp is not None and bp > 0:
bids.append({"level": lv, "price": bp, "qty": bq})
raw_hour = o1.get("aspr_acpt_hour") or ""
if isinstance(raw_hour, str) and len(raw_hour) == 6:
ts = f"{raw_hour[0:2]}:{raw_hour[2:4]}:{raw_hour[4:6]}"
else:
ts = None
return {
"ts": ts,
"current": _f(o2, "stck_prpr"),
"prev_close_diff": _f(o2, "prdy_vrss"),
"prev_close_pct": _f(o2, "prdy_ctrt"),
"asks": asks,
"bids": bids,
"ask_total": _i(o1, "total_askp_rsqn"),
"bid_total": _i(o1, "total_bidp_rsqn"),
}
def ping() -> dict[str, Any]:
"""토큰 발급만 시도해서 키 유효성 확인."""
if not _has_keys():

View File

@@ -12,6 +12,7 @@ from app.api.chart import router as chart_router
from app.api.markets import router as markets_router
from app.api.metrics import router as metrics_router
from app.api.news import router as news_router
from app.api.orderbook import router as orderbook_router
from app.api.predict import router as predict_router
from app.api.refresh import router as refresh_router
from app.api.symbols import router as symbols_router
@@ -106,6 +107,7 @@ app.include_router(metrics_router)
app.include_router(news_router)
app.include_router(markets_router)
app.include_router(themes_router)
app.include_router(orderbook_router)
def _resolved_device() -> str: