Deliver BAT-005: KIS token Redis cache (REQ-016, minor-change)
Port the trex-orbit Redis token-cache pattern so the KIS access token
survives daemon restarts and is shared across processes, avoiding KIS
token-issuance rate limits. KISBroker._ensure_token now resolves L1
in-memory -> L2 cache -> issue, storing issued tokens back to the cache.
- adapters/broker/token_cache.py: TokenCache protocol + RedisTokenCache
(key trex:broker:kis:{account_ref}:token, token+expiry+app_key suffix
JSON, ex=ttl, best-effort with graceful fallback) + InMemoryTokenCache.
- kis.py: optional token_cache injection.
- runtime config/daemon: TREX_KIS_REDIS_URL wires a RedisTokenCache.
- pyproject: redis>=5.
Verification: 96 passed, ruff clean — cache hit avoids issuance, cache is
shared across instances (restart/multi-process survival), backend errors
fall back to issuance (no regression), and app_key-suffix mismatch
invalidates a stale cache. Tested deterministically with an in-memory
double (no Redis server required). Processed via the minor-change fast
path: REQ-016 -> Implemented, BAT-005 (batch-lite) -> release-candidate.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -40,6 +40,7 @@ class KISBroker:
|
||||
base_url: str,
|
||||
timeout: float = 5.0,
|
||||
allow_live: bool = False,
|
||||
token_cache=None,
|
||||
) -> None:
|
||||
self._account_ref = account_ref
|
||||
self._app_key = app_key
|
||||
@@ -48,6 +49,7 @@ class KISBroker:
|
||||
self._timeout = timeout
|
||||
self._token: Optional[str] = None
|
||||
self._token_exp: datetime = datetime.now(tz=timezone.utc)
|
||||
self._token_cache = token_cache # optional TokenCache (L2, REQ-016)
|
||||
is_mock = any(h in self._base_url for h in _MOCK_HINTS)
|
||||
if not is_mock and not allow_live:
|
||||
raise RuntimeError(
|
||||
@@ -61,8 +63,16 @@ class KISBroker:
|
||||
|
||||
# --- auth ---
|
||||
def _ensure_token(self) -> str:
|
||||
# L1: in-memory
|
||||
if self._token and datetime.now(tz=timezone.utc) < self._token_exp:
|
||||
return self._token
|
||||
# L2: shared/persistent cache (Redis) — survives restart / multi-process (REQ-016)
|
||||
if self._token_cache is not None:
|
||||
cached = self._token_cache.load(self._account_ref, self._app_key)
|
||||
if cached is not None:
|
||||
self._token, self._token_exp = cached
|
||||
return self._token
|
||||
# L3: issue from KIS
|
||||
resp = httpx.post(
|
||||
f"{self._base_url}/oauth2/tokenP",
|
||||
json={
|
||||
@@ -77,6 +87,8 @@ class KISBroker:
|
||||
self._token = data["access_token"]
|
||||
ttl = int(data.get("expires_in", 3600))
|
||||
self._token_exp = datetime.now(tz=timezone.utc) + timedelta(seconds=max(60, ttl - 300))
|
||||
if self._token_cache is not None:
|
||||
self._token_cache.store(self._account_ref, self._app_key, self._token, self._token_exp)
|
||||
return self._token
|
||||
|
||||
def _headers(self, tr_id: str) -> dict[str, str]:
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
"""KIS token cache (REQ-016).
|
||||
|
||||
A small cache boundary so the KIS access token survives daemon restarts and
|
||||
is shared across processes (avoiding KIS token-issuance rate limits). Ported
|
||||
from the trex-orbit Redis pattern (key trex:broker:kis:{account_ref}:token,
|
||||
app_key suffix safety). All cache operations are best-effort: any backend
|
||||
error degrades to in-memory issuance (graceful fallback).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Optional, Protocol
|
||||
|
||||
_KEY = "trex:broker:kis:{account_ref}:token"
|
||||
|
||||
|
||||
class TokenCache(Protocol):
|
||||
def load(self, account_ref: str, app_key: str) -> Optional[tuple[str, datetime]]:
|
||||
"""Return (token, expires_at_utc) if a valid, matching token is cached."""
|
||||
...
|
||||
|
||||
def store(self, account_ref: str, app_key: str, token: str, expires_at: datetime) -> None:
|
||||
"""Persist token with TTL until expires_at. Best-effort."""
|
||||
...
|
||||
|
||||
|
||||
def _encode(app_key: str, token: str, expires_at: datetime) -> str:
|
||||
return json.dumps({
|
||||
"token": token,
|
||||
"expires_at": expires_at.astimezone(timezone.utc).isoformat(),
|
||||
"app_key_suffix": app_key[-4:],
|
||||
})
|
||||
|
||||
|
||||
def _decode(app_key: str, raw: str) -> Optional[tuple[str, datetime]]:
|
||||
try:
|
||||
data = json.loads(raw)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
if data.get("app_key_suffix") != app_key[-4:]:
|
||||
return None # credential rotated — ignore stale cache (AC-4)
|
||||
token = data.get("token")
|
||||
try:
|
||||
exp = datetime.fromisoformat(data["expires_at"])
|
||||
except (KeyError, TypeError, ValueError):
|
||||
return None
|
||||
if exp.tzinfo is None:
|
||||
exp = exp.replace(tzinfo=timezone.utc)
|
||||
if not token or datetime.now(tz=timezone.utc) >= exp:
|
||||
return None
|
||||
return token, exp
|
||||
|
||||
|
||||
class RedisTokenCache:
|
||||
"""Redis-backed token cache. Best-effort (errors -> miss/no-op)."""
|
||||
|
||||
def __init__(self, redis_url: str, *, redis_module: Any = None) -> None:
|
||||
if redis_module is None:
|
||||
import redis as redis_module # type: ignore[import-untyped]
|
||||
self._client = redis_module.from_url(redis_url, socket_connect_timeout=2)
|
||||
|
||||
def load(self, account_ref: str, app_key: str) -> Optional[tuple[str, datetime]]:
|
||||
try:
|
||||
raw = self._client.get(_KEY.format(account_ref=account_ref))
|
||||
except Exception: # noqa: BLE001 - cache must never break the order path
|
||||
return None
|
||||
if raw is None:
|
||||
return None
|
||||
if isinstance(raw, bytes):
|
||||
raw = raw.decode()
|
||||
return _decode(app_key, raw)
|
||||
|
||||
def store(self, account_ref: str, app_key: str, token: str, expires_at: datetime) -> None:
|
||||
ttl = int((expires_at - datetime.now(tz=timezone.utc)).total_seconds())
|
||||
if ttl <= 0:
|
||||
return
|
||||
try:
|
||||
self._client.set(_KEY.format(account_ref=account_ref),
|
||||
_encode(app_key, token, expires_at), ex=ttl)
|
||||
except Exception: # noqa: BLE001
|
||||
return
|
||||
|
||||
|
||||
class InMemoryTokenCache:
|
||||
"""Process-local cache double (tests / fallback). Shared if the instance is shared."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._store: dict[str, str] = {}
|
||||
|
||||
def load(self, account_ref: str, app_key: str) -> Optional[tuple[str, datetime]]:
|
||||
raw = self._store.get(_KEY.format(account_ref=account_ref))
|
||||
return _decode(app_key, raw) if raw else None
|
||||
|
||||
def store(self, account_ref: str, app_key: str, token: str, expires_at: datetime) -> None:
|
||||
self._store[_KEY.format(account_ref=account_ref)] = _encode(app_key, token, expires_at)
|
||||
Reference in New Issue
Block a user