diff --git a/stock/app/test_webai_cache.py b/stock/app/test_webai_cache.py new file mode 100644 index 0000000..ccfa244 --- /dev/null +++ b/stock/app/test_webai_cache.py @@ -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 diff --git a/stock/app/webai_cache.py b/stock/app/webai_cache.py new file mode 100644 index 0000000..991c042 --- /dev/null +++ b/stock/app/webai_cache.py @@ -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