feat(stock): webai_cache module (TTLCache for SP-A2)

3개의 TTLCache (portfolio 120s · news 600s · screener 180s) +
헬퍼 함수. screener key는 mode + top_n + weights canonical hash로
분기. 다음 커밋에서 /api/webai/portfolio·news-sentiment·screener/run
3 endpoint에 적용.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-18 21:43:24 +09:00
parent 8c5bfa453f
commit 030365bed0
2 changed files with 135 additions and 0 deletions

View File

@@ -0,0 +1,67 @@
"""SP-A2 — webai_cache module의 cache hit/miss + key 분기 검증."""
import time
import pytest
from app.webai_cache import (
PORTFOLIO_CACHE, NEWS_CACHE, SCREENER_CACHE,
cache_get_portfolio, cache_set_portfolio,
cache_get_news, cache_set_news,
cache_get_screener, cache_set_screener,
_screener_key,
)
def _clear_all():
PORTFOLIO_CACHE.clear()
NEWS_CACHE.clear()
SCREENER_CACHE.clear()
def test_portfolio_cache_miss_then_hit():
_clear_all()
assert cache_get_portfolio() is None
cache_set_portfolio({"holdings": [], "cash": 0})
assert cache_get_portfolio() == {"holdings": [], "cash": 0}
def test_news_cache_key_by_date():
"""date가 다르면 별도 캐시 슬롯."""
_clear_all()
cache_set_news("2026-05-18", {"count": 5})
cache_set_news("2026-05-17", {"count": 3})
assert cache_get_news("2026-05-18") == {"count": 5}
assert cache_get_news("2026-05-17") == {"count": 3}
assert cache_get_news("2026-05-16") is None # not cached
def test_news_cache_latest_key_normalized():
"""date=None은 'latest' 키로 정규화되어 동일 슬롯."""
_clear_all()
cache_set_news(None, {"count": 9})
assert cache_get_news(None) == {"count": 9}
def test_screener_key_includes_mode_and_top_n():
"""screener key는 mode + top_n + weights hash로 분기."""
k_preview = _screener_key("preview", 20, None)
k_preview_w = _screener_key("preview", 20, {"news": 0.3})
k_auto = _screener_key("auto", 20, None)
assert k_preview != k_preview_w
assert k_preview != k_auto
def test_screener_cache_roundtrip():
_clear_all()
payload = {"asof": "2026-05-18", "survivors_count": 17}
cache_set_screener("preview", 20, None, payload)
assert cache_get_screener("preview", 20, None) == payload
assert cache_get_screener("preview", 20, {"news": 0.3}) is None
def test_ttl_expiry_portfolio():
"""짧은 ttl로 만료 확인 — 직접 시간 조작 대신 TTLCache 내부 동작 신뢰."""
from cachetools import TTLCache
short = TTLCache(maxsize=1, ttl=0.1) # 0.1초
short["result"] = "x"
assert short.get("result") == "x"
time.sleep(0.2)
assert short.get("result") is None

68
stock/app/webai_cache.py Normal file
View File

@@ -0,0 +1,68 @@
"""SP-A2 — NAS stock의 /api/webai/* 엔드포인트 in-memory TTLCache.
web-ai 측 캐시(stock_client._TTL)가 miss됐을 때도 NAS에서 같은 데이터를
KIS·LLM 재호출 없이 즉시 반환하기 위한 2-layer 캐시의 server 측.
V1+V2가 동시 호출해도 NAS는 1회만 계산.
TTL 정책 (spec §10 SP-A2):
- portfolio: 120s (web-ai TTL 180s 보다 짧게 — 변경 감지 가능)
- news: 600s (sentiment는 일 단위)
- screener: 180s
"""
from __future__ import annotations
import hashlib
import json
from typing import Any, Optional
from cachetools import TTLCache
PORTFOLIO_CACHE: TTLCache = TTLCache(maxsize=1, ttl=120.0)
NEWS_CACHE: TTLCache = TTLCache(maxsize=10, ttl=600.0)
SCREENER_CACHE: TTLCache = TTLCache(maxsize=10, ttl=180.0)
# ----- portfolio -----
def cache_get_portfolio() -> Optional[Any]:
return PORTFOLIO_CACHE.get("result")
def cache_set_portfolio(value: Any) -> None:
PORTFOLIO_CACHE["result"] = value
# ----- news-sentiment -----
def _news_key(date: Optional[str]) -> str:
return date if date else "latest"
def cache_get_news(date: Optional[str]) -> Optional[Any]:
return NEWS_CACHE.get(_news_key(date))
def cache_set_news(date: Optional[str], value: Any) -> None:
NEWS_CACHE[_news_key(date)] = value
# ----- screener -----
def _screener_key(mode: str, top_n: int, weights: Optional[dict]) -> str:
"""mode + top_n + weights canonical hash. weights 객체 동등성을 키로."""
if weights is None:
w_repr = "none"
else:
# canonical: sorted keys → md5 hex (긴 weights도 짧은 키로)
canon = json.dumps(weights, sort_keys=True, ensure_ascii=False)
w_repr = hashlib.md5(canon.encode("utf-8")).hexdigest()[:12]
return f"{mode}:{top_n}:{w_repr}"
def cache_get_screener(mode: str, top_n: int, weights: Optional[dict]) -> Optional[Any]:
return SCREENER_CACHE.get(_screener_key(mode, top_n, weights))
def cache_set_screener(mode: str, top_n: int, weights: Optional[dict], value: Any) -> None:
SCREENER_CACHE[_screener_key(mode, top_n, weights)] = value