feat(chart): 10m 실시간 / 일·주·월 토글 / 오늘 마커 / 예측 거래일 선택

- backend/app/api/chart.py: interval=10m|1d|1w|1mo. 10m 은 ohlcv_1m 을
  time_bucket(10min) 으로 집계, stale(>10분) 이면 KIS 분봉 fetch 후 재조회.
  1w/1mo 는 ohlcv_daily 를 date_trunc 로 집계. today 필드 추가.
- backend/app/fetch/kis.py: fetch_minute_price() 추가 (tr_id FHKST03010200).
  KIS 응답 KST 시각을 tz-aware datetime 으로 변환, 오름차순 정렬.
- web/lib/api.ts: ChartInterval 타입, getChart(interval), predict(horizons[]).
- web/components/StockChart.tsx: 10m 이면 timeVisible. 일·주·월에서 오늘
  화살표 마커 표시. ISO datetime 도 파싱.
- web/components/PredictionPanel.tsx: 단기/중기/장기 프리셋 + 사용자 직접
  지정 (예: 1,2,3,7). API 에 horizons 배열 전달.
- web/app/[code]/page.tsx: interval 칩 (10분/일/주/월). 10m 일 때 60초마다
  폴링. interval 별 기본 lookback (10m=1, 1d=180, 1w=730, 1mo=1825).
This commit is contained in:
claude-owner
2026-05-23 01:34:29 +09:00
parent 928c2160f9
commit 0a5c634680
6 changed files with 503 additions and 81 deletions

View File

@@ -233,6 +233,77 @@ def fetch_daily_price(
return out
@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_minute_price(
code: str,
*,
end_hour: str | None = None,
) -> list[dict[str, Any]]:
"""당일 1분봉 시세 조회 (read-only). 최신 30개 캔들을 반환.
KIS 분봉 endpoint (`inquire-time-itemchartprice`) 는 base 시각 (FID_INPUT_HOUR_1) 부터
역순으로 최대 30개의 1분봉을 돌려준다. base 를 비우면 KIS 가 가장 최근 시각으로 해석.
즉 장중 호출 → 직전 30분 / 장 종료 후 호출 → 15:00~15:30 의 30분.
Returns: [{ts: datetime(KST aware), open, high, low, close, volume}, ...]
ts 오름차순 정렬.
Note: 이 endpoint 는 "당일" 분봉만 지원. 어제 이전 분봉은 별도 endpoint 가 필요한데,
이 사이트의 사용 패턴 (장중 라이브 차트) 에는 당일 데이터로 충분하다.
"""
if not _has_keys():
raise SkippedMissingKey("kis app_key/secret missing")
url = f"{KIS_BASE}/uapi/domestic-stock/v1/quotations/inquire-time-itemchartprice"
params = {
"FID_ETC_CLS_CODE": "",
"FID_COND_MRKT_DIV_CODE": "J",
"FID_INPUT_ISCD": code,
# 비우면 KIS 가 "지금" 으로 해석. 장 마감 후엔 15:30:00 부근 데이터.
"FID_INPUT_HOUR_1": end_hour or "",
"FID_PW_DATA_INCU_YN": "Y", # 과거 데이터 포함 (장 시작 직후 빈 데이터 방지)
}
with httpx.Client(timeout=15.0) as cli:
resp = cli.get(url, headers=_headers("FHKST03010200"), 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')})")
# KIS 응답은 KST. tz-aware 로 변환해서 DB (TIMESTAMPTZ) 에 안전 적재.
from datetime import timedelta, timezone as _tz
KST = _tz(timedelta(hours=9))
out: list[dict[str, Any]] = []
for row in data.get("output2", []) or []:
raw_date = row.get("stck_bsop_date")
raw_hour = row.get("stck_cntg_hour")
if not raw_date or not raw_hour:
continue
try:
ts = datetime.strptime(raw_date + raw_hour.zfill(6), "%Y%m%d%H%M%S").replace(tzinfo=KST)
except ValueError:
continue
out.append(
{
"ts": ts,
"open": float(row.get("stck_oprc") or 0),
"high": float(row.get("stck_hgpr") or 0),
"low": float(row.get("stck_lwpr") or 0),
# 분봉에서는 종가가 stck_prpr (현재가) 로 옴
"close": float(row.get("stck_prpr") or row.get("stck_clpr") or 0),
"volume": int(row.get("cntg_vol") or 0),
}
)
# KIS 응답은 보통 최신→과거 역순. UI/DB 적재 편의를 위해 오름차순으로 뒤집는다.
out.sort(key=lambda r: r["ts"])
return out
def ping() -> dict[str, Any]:
"""토큰 발급만 시도해서 키 유효성 확인."""
if not _has_keys():