fix(chart): 10분봉 페이지네이션, 장외 캐시, 예측 horizon 캡 일치

reviewer 지적사항 반영:

1. KIS 분봉이 한 번에 30개만 와서 10분봉이 최대 3개만 나오던 문제 →
   fetch_minute_range() 추가. FID_INPUT_HOUR_1 을 30분씩 후퇴시키며
   페이지네이션, 중복 ts 자연 dedupe, max_pages=20 으로 무한루프 방지.
   _ensure_intraday_fresh 는 last_ts+1m ~ now 빈 구간만 채우므로 평소엔
   1~2 페이지로 끝남.

2. 장외/주말에 매번 KIS 를 때리던 문제 →
   - 주말: 'weekend' 반환, KIS 안 부름 (분봉 endpoint 는 당일만 지원)
   - 평일 장외 + 오늘 데이터 있음: 'cached_closed' 반환
   - 장중 + 10분 이내: 'fresh' 반환
   토큰/조회 제한 다시 밟지 않음.

3. 프론트 horizon 입력 60 vs 백엔드 30 불일치 →
   PredictionPanel 의 cap 을 30 으로 맞춤. 백엔드 predict.py 의 학습/검증
   범위와 일치. placeholder 도 '1~30' 으로 명시.
This commit is contained in:
claude-owner
2026-05-23 01:41:43 +09:00
parent 0a5c634680
commit e610599879
3 changed files with 101 additions and 18 deletions

View File

@@ -304,6 +304,59 @@ def fetch_minute_price(
return out
def fetch_minute_range(
code: str,
from_ts: datetime,
to_ts: datetime,
*,
max_pages: int = 20,
) -> list[dict[str, Any]]:
"""[from_ts, to_ts] 윈도우의 1분봉 전체. KIS 30-bar 페이지를 역순으로 반복 호출.
KIS `inquire-time-itemchartprice` 는 한 번에 최대 30개 1분봉만 주고,
`FID_INPUT_HOUR_1` 기준 그 시각 포함 이전 30분을 반환한다. 그래서 to_ts 부터
시작해서 가장 이른 응답 시각의 -1분을 다음 cursor 로 잡아 from_ts 까지 후퇴.
중복 키 (ts) 는 dict 로 자연 dedupe. 더 이상 새 행이 안 들어오거나 max_pages 도달
하면 종료 (rate-limit/무한루프 방지).
Note: 이 endpoint 는 "당일" 만 지원. from_ts/to_ts 는 같은 날짜여야 한다.
"""
if not _has_keys():
raise SkippedMissingKey("kis app_key/secret missing")
if from_ts >= to_ts:
return []
from datetime import timedelta as _td
accumulated: dict[datetime, dict[str, Any]] = {}
cursor = to_ts
pages = 0
while cursor > from_ts and pages < max_pages:
pages += 1
rows = fetch_minute_price(code, end_hour=cursor.strftime("%H%M%S"))
if not rows:
break
added = 0
for r in rows:
ts = r["ts"]
if ts < from_ts or ts > to_ts:
continue
if ts not in accumulated:
accumulated[ts] = r
added += 1
if added == 0:
# 같은 30개를 또 받았다 — 더 과거가 없거나 KIS 가 똑같은 페이지를 반환.
break
earliest_ts = min(r["ts"] for r in rows)
next_cursor = earliest_ts - _td(minutes=1)
if next_cursor >= cursor:
break
cursor = next_cursor
return sorted(accumulated.values(), key=lambda r: r["ts"])
def ping() -> dict[str, Any]:
"""토큰 발급만 시도해서 키 유효성 확인."""
if not _has_keys():