diff --git a/.github/workflows/sync-openapi.yml b/.github/workflows/sync-openapi.yml
index d649c77..24a922e 100644
--- a/.github/workflows/sync-openapi.yml
+++ b/.github/workflows/sync-openapi.yml
@@ -2,7 +2,7 @@ name: Sync OpenAPI spec to psxdata-docs
 
 on:
   push:
-    branches: [main]
+    tags: ["v*"]
 
 jobs:
   sync:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9933b85..b30545a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,26 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ---
 
+## [0.3.0] — 2026-09-30
+
+### Added
+
+- `GET /stocks/{symbol}/historical` is now cached. Each symbol's full history is fetched from PSX once and reused for every `start`/`end` range. It stays fresh for 30 minutes during PSX trading hours (Mon–Fri 09:00–17:00 PKT) and until the next trading-day open otherwise. Responses carry `X-Cache: HIT | MISS | STALE`, an `Age` header on cached responses, and `meta.cached: true` when served from cache.
+- When PSX rate-limits the API or is unreachable, `/historical` serves the last cached copy (`X-Cache: STALE`) instead of failing.
+- Optional `REDIS_URL` environment variable: any Redis-compatible server (e.g. Aiven for Valkey) that keeps the cache across restarts and instances. Without it the cache is in-memory only. `HISTORICAL_CACHE_MARKET_TTL` (seconds, default `1800`) tunes trading-hours freshness.
+
+### Fixed
+
+- PSX rate-limiting (HTTP 429 from PSX) surfaced as `500 internal_error` on every endpoint (324 such responses on `/historical` on 2026-09-29). It now returns `503 psx_unavailable` with `Retry-After: 60`, and `/historical` stops calling PSX for 60 seconds after a 429.
+- Invalid `start`/`end` dates, or `start` after `end`, on `/historical` returned `500`; they now return `422 bad_request`.
+
+### Changed
+
+- `start`/`end` on `/historical` must be ISO dates (`YYYY-MM-DD`).
+- The OpenAPI spec is pushed to the docs site on release tags (`v*`) instead of every push to `main`, so public docs change only when a release is made.
+
+---
+
 ## [0.2.1] — 2026-09-26
 
 ### Fixed
diff --git a/README.md b/README.md
index e6c96b4..10eff01 100644
--- a/README.md
+++ b/README.md
@@ -98,6 +98,8 @@ Every response wraps its payload in a consistent envelope.
 }
 ```
 
+`meta.cached` is `true` when the response was served from the API's cache (currently `/stocks/{symbol}/historical`; see [Caching](#caching)).
+
 **Error response**
 ```json
 {
@@ -113,7 +115,7 @@ Every response wraps its payload in a consistent envelope.
 | 404 | `not_found` | Symbol or index does not exist |
 | 429 | `rate_limited` | Exceeded 60 requests/minute per IP |
 | 502 | `upstream_data_error` | Upstream PSX data failed validation |
-| 503 | `psx_unavailable` | PSX website unreachable |
+| 503 | `psx_unavailable` | PSX website unreachable, or PSX is rate-limiting the API (sent with a `Retry-After` header) |
 | 500 | `internal_error` | Unexpected server error |
 
 ---
@@ -124,6 +126,43 @@ Every response wraps its payload in a consistent envelope.
 
 ---
 
+## Caching
+
+`GET /stocks/{symbol}/historical` is served from a cache. PSX always returns a symbol's full history, so the API fetches it once and slices it to your `start`/`end` range.
+
+| When the data was fetched | Stays fresh until |
+| ------------------------- | ----------------- |
+| Mon–Fri 09:00–17:00 PKT (trading hours) | 30 minutes later, but no later than 17:00 |
+| Any other time | The next weekday 09:00 PKT |
+
+Every `/historical` response says where it came from:
+
+| Header | Values |
+| ------ | ------ |
+| `X-Cache` | `HIT` — fresh cached copy · `MISS` — fetched from PSX just now · `STALE` — PSX refused or was unreachable, so the last cached copy was served |
+| `Age` | Seconds since the data was fetched from PSX (on `HIT` and `STALE`) |
+
+`meta.cached` is `true` for `HIT` and `STALE`. If PSX is rate-limiting and no cached copy exists, the API returns `503 psx_unavailable` with `Retry-After: 60`.
+
+---
+
+## Configuration
+
+All settings are optional environment variables.
+
+| Variable | Default | Description |
+| -------- | ------- | ----------- |
+| `REDIS_URL` | unset | Redis-compatible server for the `/historical` cache, e.g. an [Aiven for Valkey](https://aiven.io/valkey) service URI (`rediss://default:<password>@<host>:<port>`). Keeps the cache across restarts and instances. When unset, or when the server is unreachable, the API falls back to an in-memory cache and keeps working. |
+| `HISTORICAL_CACHE_MARKET_TTL` | `1800` | How long, in seconds, `/historical` data stays fresh during PSX trading hours. |
+
+```bash
+docker run -p 8000:8000 -e REDIS_URL="rediss://default:<password>@<host>:<port>" mtauha/psxdata-api
+```
+
+On FastAPI Cloud, set it as an app environment variable (`fastapi cloud env`) rather than committing it.
+
+---
+
 ## Docker
 
 ```bash
diff --git a/api/cache/__init__.py b/api/cache/__init__.py
new file mode 100644
index 0000000..e1574ae
--- /dev/null
+++ b/api/cache/__init__.py
@@ -0,0 +1 @@
+"""Response cache for GET /stocks/{symbol}/historical."""
diff --git a/api/cache/codec.py b/api/cache/codec.py
new file mode 100644
index 0000000..0bc0f3f
--- /dev/null
+++ b/api/cache/codec.py
@@ -0,0 +1,84 @@
+"""Serialisation of cached /historical entries: a versioned, gzip-compressed JSON envelope.
+
+Rows are stored as arrays under a single ``columns`` list, which is roughly 20% smaller than
+repeating the keys on every row.
+"""
+from __future__ import annotations
+
+import gzip
+import json
+import logging
+import zlib
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+FORMAT_VERSION = 1
+COLUMNS: tuple[str, ...] = ("date", "open", "high", "low", "close", "volume", "is_anomaly")
+
+
+@dataclass(frozen=True)
+class CacheEntry:
+    symbol: str
+    fetched_at: datetime
+    fresh_until: datetime
+    rows: list[dict[str, Any]]
+
+
+def _json_default(value: Any) -> Any:
+    # numpy / pandas scalars expose .item() returning the native Python value
+    if hasattr(value, "item"):
+        return value.item()
+    raise TypeError(f"Object of type {type(value).__name__} is not JSON serialisable")
+
+
+def encode(entry: CacheEntry) -> bytes:
+    payload = {
+        "v": FORMAT_VERSION,
+        "symbol": entry.symbol,
+        "fetched_at": entry.fetched_at.isoformat(),
+        "fresh_until": entry.fresh_until.isoformat(),
+        "columns": list(COLUMNS),
+        "rows": [[row.get(col) for col in COLUMNS] for row in entry.rows],
+    }
+    raw = json.dumps(payload, default=_json_default, separators=(",", ":")).encode()
+    return gzip.compress(raw, compresslevel=6)
+
+
+def _aware(value: Any) -> datetime:
+    parsed = datetime.fromisoformat(value)
+    if parsed.tzinfo is None:
+        raise ValueError("cache timestamp is not timezone-aware")
+    return parsed
+
+
+def _row(values: Any) -> dict[str, Any]:
+    if not isinstance(values, list) or len(values) != len(COLUMNS):
+        raise ValueError("cache row has the wrong shape")
+    if not isinstance(values[0], str):
+        raise TypeError("cache row date is not a string")
+    return dict(zip(COLUMNS, values))
+
+
+def decode(blob: bytes) -> CacheEntry | None:
+    """Return the entry, or None for anything that is not a complete v1 envelope."""
+    try:
+        payload = json.loads(gzip.decompress(blob))
+        if payload["v"] != FORMAT_VERSION:
+            logger.warning("Ignoring cache entry with unknown format version %r", payload["v"])
+            return None
+        if payload["columns"] != list(COLUMNS):
+            raise ValueError("cache entry has unexpected columns")
+        if not isinstance(payload["symbol"], str) or not isinstance(payload["rows"], list):
+            raise TypeError("cache entry has malformed fields")
+        return CacheEntry(
+            symbol=payload["symbol"],
+            fetched_at=_aware(payload["fetched_at"]),
+            fresh_until=_aware(payload["fresh_until"]),
+            rows=[_row(values) for values in payload["rows"]],
+        )
+    except (OSError, EOFError, zlib.error, ValueError, KeyError, TypeError) as exc:
+        logger.warning("Ignoring unreadable cache entry (%s)", type(exc).__name__)
+        return None
diff --git a/api/cache/factory.py b/api/cache/factory.py
new file mode 100644
index 0000000..91d9fb7
--- /dev/null
+++ b/api/cache/factory.py
@@ -0,0 +1,76 @@
+"""Build the /historical cache from environment configuration.
+
+``REDIS_URL`` (optional) points at any Redis-compatible server, e.g. Aiven for Valkey. It is
+never logged, only its hostname.
+"""
+from __future__ import annotations
+
+import logging
+from collections.abc import Mapping
+from datetime import timedelta
+from urllib.parse import urlparse
+
+import redis
+from redis.backoff import NoBackoff
+from redis.retry import Retry
+
+from api.cache.freshness import DEFAULT_MARKET_TTL
+from api.cache.historical import HistoricalService
+from api.cache.store import MemoryLRU, RedisStore, TieredStore
+
+logger = logging.getLogger(__name__)
+
+REDIS_TIMEOUT_SECONDS = 0.5
+
+
+def market_ttl_from_env(env: Mapping[str, str]) -> timedelta:
+    raw = env.get("HISTORICAL_CACHE_MARKET_TTL")
+    if raw is None:
+        return DEFAULT_MARKET_TTL
+    try:
+        seconds = int(raw)
+    except ValueError:
+        seconds = 0
+    if seconds <= 0:
+        logger.warning(
+            "Ignoring invalid HISTORICAL_CACHE_MARKET_TTL=%r; using %ds",
+            raw, int(DEFAULT_MARKET_TTL.total_seconds()),
+        )
+        return DEFAULT_MARKET_TTL
+    return timedelta(seconds=seconds)
+
+
+def build_redis_client(url: str | None) -> redis.Redis | None:
+    if not url:
+        return None
+    try:
+        return redis.Redis.from_url(
+            url,
+            socket_connect_timeout=REDIS_TIMEOUT_SECONDS,
+            socket_timeout=REDIS_TIMEOUT_SECONDS,
+            health_check_interval=30,
+            retry=Retry(NoBackoff(), 0),
+        )
+    except ValueError as exc:
+        logger.warning(
+            "REDIS_URL could not be parsed (%s); using in-memory cache only", type(exc).__name__
+        )
+        return None
+
+
+def build_historical_service(env: Mapping[str, str]) -> HistoricalService:
+    url = env.get("REDIS_URL")
+    client = build_redis_client(url)
+    if client is None:
+        logger.info("No cache backend configured; /historical uses in-memory cache only")
+    else:
+        host = urlparse(url or "").hostname or "unknown-host"
+        try:
+            client.ping()
+            logger.info("Cache backend connected: %s", host)
+        except (redis.RedisError, OSError) as exc:
+            logger.warning(
+                "Cache backend unreachable at startup: %s (%s)", host, type(exc).__name__
+            )
+    store = TieredStore(MemoryLRU(), RedisStore(client))
+    return HistoricalService(store, market_ttl=market_ttl_from_env(env))
diff --git a/api/cache/freshness.py b/api/cache/freshness.py
new file mode 100644
index 0000000..5e768f4
--- /dev/null
+++ b/api/cache/freshness.py
@@ -0,0 +1,37 @@
+"""Freshness policy for cached /historical entries, tied to PSX trading hours.
+
+Past rows never change; only today's row moves while the market is open. Inside a padded
+trading window an entry stays fresh for ``market_ttl`` (capped at the window close); outside
+it, an entry stays fresh until the next window opens.
+"""
+from __future__ import annotations
+
+from datetime import date, datetime, time, timedelta, timezone
+from zoneinfo import ZoneInfo
+
+PKT = ZoneInfo("Asia/Karachi")
+WINDOW_OPEN = time(9, 0)
+WINDOW_CLOSE = time(17, 0)
+DEFAULT_MARKET_TTL = timedelta(minutes=30)
+
+
+def _is_weekday(day: date) -> bool:
+    return day.weekday() < 5
+
+
+def fresh_until(fetched_at: datetime, market_ttl: timedelta = DEFAULT_MARKET_TTL) -> datetime:
+    """Return the UTC instant after which data fetched at ``fetched_at`` is stale."""
+    if fetched_at.tzinfo is None:
+        raise ValueError("fetched_at must be timezone-aware")
+    local = fetched_at.astimezone(PKT)
+    today = local.date()
+    window_open = datetime.combine(today, WINDOW_OPEN, tzinfo=PKT)
+    window_close = datetime.combine(today, WINDOW_CLOSE, tzinfo=PKT)
+
+    if _is_weekday(today) and window_open <= local < window_close:
+        return min(local + market_ttl, window_close).astimezone(timezone.utc)
+
+    day = today if _is_weekday(today) and local < window_open else today + timedelta(days=1)
+    while not _is_weekday(day):
+        day += timedelta(days=1)
+    return datetime.combine(day, WINDOW_OPEN, tzinfo=PKT).astimezone(timezone.utc)
diff --git a/api/cache/historical.py b/api/cache/historical.py
new file mode 100644
index 0000000..055cc8e
--- /dev/null
+++ b/api/cache/historical.py
@@ -0,0 +1,125 @@
+"""Cache-first access to full symbol histories.
+
+PSX always returns a symbol's whole history, so one entry per symbol serves every date range.
+Concurrent misses for one symbol share a single PSX fetch via a fixed pool of striped locks.
+When PSX refuses (429) or is unavailable, the last cached copy is served as STALE. After a 429,
+PSX is not called again for ``cooldown_seconds``.
+"""
+from __future__ import annotations
+
+import threading
+import time
+import zlib
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from enum import Enum
+from typing import Any
+
+from psxdata.exceptions import PSXRateLimitError, PSXUnavailableError
+
+from api.cache.codec import CacheEntry, decode, encode
+from api.cache.freshness import DEFAULT_MARKET_TTL, fresh_until
+from api.cache.store import TieredStore
+
+LOCK_STRIPES = 256
+PSX_COOLDOWN_SECONDS = 60
+COOLDOWN_MESSAGE = "PSX is rate-limiting requests; retry later"
+
+Row = dict[str, Any]
+Fetcher = Callable[[str], list[Row]]
+
+
+class CacheStatus(str, Enum):
+    HIT = "HIT"
+    MISS = "MISS"
+    STALE = "STALE"
+
+
+@dataclass(frozen=True)
+class CacheResult:
+    rows: list[Row]
+    status: CacheStatus
+    fetched_at: datetime
+
+
+def cache_key(symbol: str) -> str:
+    return f"psxdata-api:v1:historical:{symbol.upper()}"
+
+
+def _utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class HistoricalService:
+    def __init__(
+        self,
+        store: TieredStore,
+        *,
+        market_ttl: timedelta = DEFAULT_MARKET_TTL,
+        cooldown_seconds: float = PSX_COOLDOWN_SECONDS,
+        now: Callable[[], datetime] = _utcnow,
+        monotonic: Callable[[], float] = time.monotonic,
+    ) -> None:
+        self.store = store
+        self._market_ttl = market_ttl
+        self._cooldown = cooldown_seconds
+        self._now = now
+        self._monotonic = monotonic
+        self._blocked_until = 0.0
+        self._state_lock = threading.Lock()
+        self._locks = [threading.Lock() for _ in range(LOCK_STRIPES)]
+
+    def get(self, symbol: str, fetch: Fetcher) -> CacheResult:
+        symbol = symbol.upper()
+        key = cache_key(symbol)
+        entry = self._load(key)
+        if entry is not None and self._is_fresh(entry):
+            return CacheResult(entry.rows, CacheStatus.HIT, entry.fetched_at)
+
+        with self._lock_for(key):
+            entry = self._load(key)
+            if entry is not None and self._is_fresh(entry):
+                return CacheResult(entry.rows, CacheStatus.HIT, entry.fetched_at)
+            if self._in_cooldown():
+                return self._fallback(entry, PSXRateLimitError(COOLDOWN_MESSAGE))
+            try:
+                rows = fetch(symbol)
+            except PSXRateLimitError as exc:
+                self._start_cooldown()
+                return self._fallback(entry, exc)
+            except PSXUnavailableError as exc:
+                return self._fallback(entry, exc)
+
+            rows = sorted(rows, key=lambda row: row.get("date") or "", reverse=True)
+            fetched_at = self._now()
+            fresh = CacheEntry(symbol, fetched_at, fresh_until(fetched_at, self._market_ttl), rows)
+            self.store.put(key, encode(fresh))
+            return CacheResult(rows, CacheStatus.MISS, fetched_at)
+
+    def close(self) -> None:
+        self.store.close()
+
+    def _load(self, key: str) -> CacheEntry | None:
+        blob = self.store.get(key)
+        return decode(blob) if blob is not None else None
+
+    def _is_fresh(self, entry: CacheEntry) -> bool:
+        return self._now() < entry.fresh_until
+
+    def _lock_for(self, key: str) -> threading.Lock:
+        return self._locks[zlib.crc32(key.encode()) % LOCK_STRIPES]
+
+    def _in_cooldown(self) -> bool:
+        with self._state_lock:
+            return self._monotonic() < self._blocked_until
+
+    def _start_cooldown(self) -> None:
+        with self._state_lock:
+            self._blocked_until = self._monotonic() + self._cooldown
+
+    @staticmethod
+    def _fallback(entry: CacheEntry | None, exc: Exception) -> CacheResult:
+        if entry is None:
+            raise exc
+        return CacheResult(entry.rows, CacheStatus.STALE, entry.fetched_at)
diff --git a/api/cache/store.py b/api/cache/store.py
new file mode 100644
index 0000000..b0c3dc4
--- /dev/null
+++ b/api/cache/store.py
@@ -0,0 +1,145 @@
+"""Two-tier byte store: an in-process LRU (L1) in front of an optional Redis-compatible
+server (L2).
+
+L1 serves almost every hit without a network round-trip. L2 keeps entries across restarts
+and instances. L2 fails open: an outage costs one timeout per backoff window, never an error.
+"""
+from __future__ import annotations
+
+import logging
+import threading
+import time
+from collections import OrderedDict
+from collections.abc import Callable
+
+import redis
+
+logger = logging.getLogger(__name__)
+
+L1_MAX_BYTES = 64 * 1024 * 1024
+L2_TTL_SECONDS = 7 * 24 * 60 * 60
+L2_BACKOFF_SECONDS = 30.0
+
+
+class MemoryLRU:
+    """Thread-safe LRU of bytes values, bounded by their total size."""
+
+    def __init__(self, max_bytes: int = L1_MAX_BYTES) -> None:
+        self._max_bytes = max_bytes
+        self._data: OrderedDict[str, bytes] = OrderedDict()
+        self._size = 0
+        self._lock = threading.Lock()
+
+    @property
+    def size_bytes(self) -> int:
+        return self._size
+
+    def get(self, key: str) -> bytes | None:
+        with self._lock:
+            value = self._data.get(key)
+            if value is not None:
+                self._data.move_to_end(key)
+            return value
+
+    def set(self, key: str, value: bytes) -> None:
+        if len(value) > self._max_bytes:
+            return
+        with self._lock:
+            old = self._data.pop(key, None)
+            if old is not None:
+                self._size -= len(old)
+            self._data[key] = value
+            self._size += len(value)
+            while self._size > self._max_bytes:
+                _, evicted = self._data.popitem(last=False)
+                self._size -= len(evicted)
+
+
+class RedisStore:
+    """Fail-open wrapper around a Redis-compatible client (``None`` means not configured)."""
+
+    def __init__(
+        self,
+        client: redis.Redis | None,
+        *,
+        ttl_seconds: int = L2_TTL_SECONDS,
+        backoff_seconds: float = L2_BACKOFF_SECONDS,
+        monotonic: Callable[[], float] = time.monotonic,
+    ) -> None:
+        self._client = client
+        self._ttl = ttl_seconds
+        self._backoff = backoff_seconds
+        self._monotonic = monotonic
+        self._disabled_until = 0.0
+        self._lock = threading.Lock()
+
+    @property
+    def enabled(self) -> bool:
+        return self._client is not None
+
+    def get(self, key: str) -> bytes | None:
+        client = self._usable_client()
+        if client is None:
+            return None
+        try:
+            value = client.get(key)
+        except (redis.RedisError, OSError) as exc:
+            self._fail("read", exc)
+            return None
+        return value if isinstance(value, bytes) else None
+
+    def set(self, key: str, value: bytes) -> None:
+        client = self._usable_client()
+        if client is None:
+            return
+        try:
+            client.set(key, value, ex=self._ttl)
+        except (redis.RedisError, OSError) as exc:
+            self._fail("write", exc)
+
+    def close(self) -> None:
+        if self._client is None:
+            return
+        try:
+            self._client.close()
+        except (redis.RedisError, OSError):
+            pass
+
+    def _usable_client(self) -> redis.Redis | None:
+        if self._client is None or self._monotonic() < self._disabled_until:
+            return None
+        return self._client
+
+    def _fail(self, operation: str, exc: Exception) -> None:
+        with self._lock:
+            now = self._monotonic()
+            if now >= self._disabled_until:
+                logger.warning(
+                    "Cache backend %s failed (%s); bypassing it for %.0fs",
+                    operation, type(exc).__name__, self._backoff,
+                )
+            self._disabled_until = now + self._backoff
+
+
+class TieredStore:
+    """L1 first; on an L1 miss, L2 (filling L1). Writes go to both."""
+
+    def __init__(self, l1: MemoryLRU, l2: RedisStore) -> None:
+        self.l1 = l1
+        self.l2 = l2
+
+    def get(self, key: str) -> bytes | None:
+        value = self.l1.get(key)
+        if value is not None:
+            return value
+        value = self.l2.get(key)
+        if value is not None:
+            self.l1.set(key, value)
+        return value
+
+    def put(self, key: str, value: bytes) -> None:
+        self.l1.set(key, value)
+        self.l2.set(key, value)
+
+    def close(self) -> None:
+        self.l2.close()
diff --git a/api/dependencies.py b/api/dependencies.py
index 1116fb3..c2ca174 100644
--- a/api/dependencies.py
+++ b/api/dependencies.py
@@ -1,7 +1,11 @@
 """Shared FastAPI dependencies for the API layer."""
+from fastapi import Request
 from slowapi import Limiter
 from slowapi.util import get_remote_address
 
+from api.cache.factory import build_historical_service
+from api.cache.historical import HistoricalService
+
 limiter = Limiter(key_func=get_remote_address, default_limits=["60/minute"])
 
 
@@ -9,6 +13,10 @@ def get_rate_limiter() -> Limiter:
     return limiter
 
 
-def get_cache() -> None:
-    # TODO: return Redis client when Redis layer is added
-    return None
+def get_cache(request: Request) -> HistoricalService:
+    """Return the app's /historical cache; build a memory-only one if lifespan did not run."""
+    service: HistoricalService | None = getattr(request.app.state, "historical_service", None)
+    if service is None:
+        service = build_historical_service({})
+        request.app.state.historical_service = service
+    return service
diff --git a/api/main.py b/api/main.py
index 8b586de..3b4e6b0 100644
--- a/api/main.py
+++ b/api/main.py
@@ -1,16 +1,24 @@
 import logging
+import os
 from contextlib import asynccontextmanager
 
 from fastapi import FastAPI, Request
 from fastapi.exceptions import RequestValidationError
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
-from psxdata.exceptions import InvalidSymbolError, PSXParseError, PSXUnavailableError
+from psxdata.exceptions import (
+    InvalidSymbolError,
+    PSXParseError,
+    PSXRateLimitError,
+    PSXUnavailableError,
+)
 from pydantic import ValidationError
 from slowapi.errors import RateLimitExceeded
 from slowapi.middleware import SlowAPIMiddleware
 from starlette.exceptions import HTTPException as StarletteHTTPException
 
+from api.cache.factory import build_historical_service
+from api.cache.historical import COOLDOWN_MESSAGE, PSX_COOLDOWN_SECONDS
 from api.dependencies import limiter
 from api.routers import router_registry
 
@@ -19,9 +27,13 @@
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    """Manage application lifespan — startup and shutdown events."""
-    # TODO: initialise cache / Redis on startup, close on shutdown.
-    yield
+    """Build the /historical cache on startup and close its backend on shutdown."""
+    service = build_historical_service(os.environ)
+    app.state.historical_service = service
+    try:
+        yield
+    finally:
+        service.close()
 
 
 app = FastAPI(
@@ -71,6 +83,15 @@ async def psx_unavailable_handler(request: Request, exc: PSXUnavailableError) ->
     )
 
 
+@app.exception_handler(PSXRateLimitError)
+async def psx_rate_limit_handler(request: Request, exc: PSXRateLimitError) -> JSONResponse:
+    return JSONResponse(
+        status_code=503,
+        headers={"Retry-After": str(PSX_COOLDOWN_SECONDS)},
+        content={"error": {"status": 503, "code": _ERROR_CODES[503], "message": COOLDOWN_MESSAGE}},
+    )
+
+
 @app.exception_handler(InvalidSymbolError)
 async def invalid_symbol_handler(request: Request, exc: InvalidSymbolError) -> JSONResponse:
     return JSONResponse(
diff --git a/api/requirements.txt b/api/requirements.txt
index 86900f7..1586c4c 100644
--- a/api/requirements.txt
+++ b/api/requirements.txt
@@ -4,4 +4,5 @@ fastapi~=0.100
 uvicorn[standard]~=0.23
 slowapi~=0.1
 redis>=4.5
+tzdata>=2024.1
 httpx~=0.27
diff --git a/api/routers/stocks.py b/api/routers/stocks.py
index 283431b..b277a42 100644
--- a/api/routers/stocks.py
+++ b/api/routers/stocks.py
@@ -1,14 +1,18 @@
 """Stocks router — /stocks and /stocks/{symbol}/* endpoints."""
 from __future__ import annotations
 
-from datetime import datetime, timezone
+import re
+from datetime import date, datetime, timezone
+from typing import Any
 
 import pandas as pd
 import psxdata
-from fastapi import APIRouter, HTTPException, Request
+from fastapi import APIRouter, HTTPException, Request, Response
 
-from api.dependencies import limiter
+from api.cache.historical import CacheStatus
+from api.dependencies import get_cache, limiter
 from api.schemas import (
+    ErrorEnvelope,
     FundamentalsResponse,
     FundamentalsRow,
     HistoricalResponse,
@@ -37,6 +41,52 @@ def _df_to_records(df: pd.DataFrame) -> list[dict]:
     return df.to_dict("records")
 
 
+_HISTORICAL_RESPONSES: dict[int | str, dict[str, Any]] = {
+    200: {
+        "headers": {
+            "X-Cache": {
+                "description": "HIT (fresh cache), MISS (fetched from PSX), or STALE "
+                "(PSX refused or was unreachable; last cached copy served)",
+                "schema": {"type": "string", "enum": ["HIT", "MISS", "STALE"]},
+            },
+            "Age": {
+                "description": "Seconds since the data was fetched from PSX (HIT and STALE only)",
+                "schema": {"type": "integer"},
+            },
+        },
+    },
+    503: {
+        "model": ErrorEnvelope,
+        "description": "PSX is unavailable or rate-limiting, and no cached copy exists",
+        "headers": {
+            "Retry-After": {
+                "description": "Seconds to wait before retrying (sent when PSX is rate-limiting)",
+                "schema": {"type": "integer"},
+            },
+        },
+    },
+}
+
+
+_ISO_DATE = re.compile(r"\d{4}-\d{2}-\d{2}")
+
+
+def _parse_date(name: str, value: str | None) -> date | None:
+    if value is None:
+        return None
+    # fromisoformat also accepts 20240102 and 2024-W01-2; the documented contract is YYYY-MM-DD
+    if _ISO_DATE.fullmatch(value):
+        try:
+            return date.fromisoformat(value)
+        except ValueError:
+            pass
+    raise HTTPException(status_code=422, detail=f"{name} must be an ISO date (YYYY-MM-DD)")
+
+
+def _fetch_full_history(symbol: str) -> list[dict]:
+    return _df_to_records(psxdata.stocks(symbol, cache=False))
+
+
 @router.get("/stocks", response_model=StringListResponse)
 @limiter.limit("60/minute")
 def list_stocks(request: Request, index: str | None = None) -> StringListResponse:
@@ -47,21 +97,42 @@ def list_stocks(request: Request, index: str | None = None) -> StringListRespons
     )
 
 
-@router.get("/stocks/{symbol}/historical", response_model=HistoricalResponse)
+@router.get(
+    "/stocks/{symbol}/historical",
+    response_model=HistoricalResponse,
+    responses=_HISTORICAL_RESPONSES,
+)
 @limiter.limit("60/minute")
 def get_historical(
     request: Request,
+    response: Response,
     symbol: str,
     start: str | None = None,
     end: str | None = None,
 ) -> HistoricalResponse:
-    df = psxdata.stocks(symbol.upper(), start=start, end=end)
-    rows: list[OHLCVRow] = []
-    if not df.empty:
-        rows = [OHLCVRow.model_validate(r) for r in _df_to_records(df)]
+    start_date = _parse_date("start", start)
+    end_date = _parse_date("end", end)
+    if start_date and end_date and start_date > end_date:
+        raise HTTPException(status_code=422, detail="start must not be after end")
+
+    result = get_cache(request).get(symbol.upper(), _fetch_full_history)
+
+    lo = start_date.isoformat() if start_date else None
+    hi = end_date.isoformat() if end_date else None
+    rows = [
+        OHLCVRow.model_validate(r)
+        for r in result.rows
+        if (lo is None or r["date"] >= lo) and (hi is None or r["date"] <= hi)
+    ]
+
+    cached = result.status is not CacheStatus.MISS
+    response.headers["X-Cache"] = result.status.value
+    if cached:
+        age = (datetime.now(timezone.utc) - result.fetched_at).total_seconds()
+        response.headers["Age"] = str(max(0, int(age)))
     return HistoricalResponse(
         data=rows,
-        meta=MetaList(timestamp=_now_iso(), cached=False, count=len(rows)),
+        meta=MetaList(timestamp=_now_iso(), cached=cached, count=len(rows)),
     )
 
 
diff --git a/pyproject.toml b/pyproject.toml
index e1a2c09..88b4c37 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "psxdata-api"
-version = "0.2.1"
+version = "0.3.0"
 description = "FastAPI REST service for Pakistan Stock Exchange (PSX) data"
 readme = "README.md"
 requires-python = ">=3.11"
@@ -16,6 +16,7 @@ dependencies = [
     "slowapi>=0.1",
     "redis>=4.5",
     "httpx>=0.27",
+    "tzdata>=2024.1",
 ]
 
 [project.optional-dependencies]
@@ -23,6 +24,7 @@ dev = [
     "pytest>=7",
     "pytest-cov",
     "pytest-asyncio",
+    "fakeredis>=2.20",
     "httpx>=0.27",
     "ruff>=0.4",
     "mypy>=1.0",
diff --git a/tests/api/conftest.py b/tests/api/conftest.py
new file mode 100644
index 0000000..57a0c68
--- /dev/null
+++ b/tests/api/conftest.py
@@ -0,0 +1,14 @@
+"""Shared fixtures for API tests."""
+import pytest
+
+from api.cache.historical import HistoricalService
+from api.cache.store import MemoryLRU, RedisStore, TieredStore
+from api.main import app
+
+
+@pytest.fixture(autouse=True)
+def fresh_historical_service() -> HistoricalService:
+    """Give every test an empty, Redis-less /historical cache so cached data never leaks."""
+    service = HistoricalService(TieredStore(MemoryLRU(), RedisStore(None)))
+    app.state.historical_service = service
+    return service
diff --git a/tests/api/test_cache_codec.py b/tests/api/test_cache_codec.py
new file mode 100644
index 0000000..a9d5810
--- /dev/null
+++ b/tests/api/test_cache_codec.py
@@ -0,0 +1,90 @@
+"""Tests for the gzip JSON cache envelope."""
+import gzip
+import json
+import logging
+import math
+from datetime import datetime, timezone
+
+import numpy as np
+
+from api.cache.codec import COLUMNS, CacheEntry, decode, encode
+
+T0 = datetime(2026, 9, 29, 5, 0, tzinfo=timezone.utc)
+T1 = datetime(2026, 9, 29, 5, 30, tzinfo=timezone.utc)
+ROWS = [
+    {"date": "2026-09-29", "open": 10.0, "high": 11.0, "low": 9.5, "close": 10.5,
+     "volume": 1000, "is_anomaly": False},
+    {"date": "2026-09-28", "open": 9.0, "high": 10.0, "low": 8.5, "close": 9.5,
+     "volume": 900, "is_anomaly": True},
+]
+
+
+def test_round_trip_preserves_entry_and_row_order() -> None:
+    entry = CacheEntry("SYS", T0, T1, ROWS)
+    assert decode(encode(entry)) == entry
+
+
+def test_round_trip_preserves_nulls_nan_and_numpy_scalars() -> None:
+    rows = [{"date": "2026-09-29", "open": np.float64(1.5), "high": np.float64(2.0),
+             "low": np.float64(1.0), "close": float("nan"), "volume": None,
+             "is_anomaly": np.bool_(True)}]
+    decoded = decode(encode(CacheEntry("SYS", T0, T1, rows)))
+    assert decoded is not None
+    row = decoded.rows[0]
+    assert math.isnan(row.pop("close"))      # NaN stays NaN, exactly as uncached
+    assert row == {"date": "2026-09-29", "open": 1.5, "high": 2.0, "low": 1.0,
+                   "volume": None, "is_anomaly": True}
+
+
+def test_empty_history_round_trips() -> None:
+    entry = CacheEntry("NOPE", T0, T1, [])
+    assert decode(encode(entry)) == entry
+
+
+def test_only_known_columns_are_kept() -> None:
+    rows = [dict(ROWS[0], symbol="SYS", extra=1)]
+    decoded = decode(encode(CacheEntry("SYS", T0, T1, rows)))
+    assert decoded is not None
+    assert set(decoded.rows[0]) == {"date", "open", "high", "low", "close", "volume",
+                                    "is_anomaly"}
+
+
+def test_payload_is_versioned_gzip_json() -> None:
+    payload = json.loads(gzip.decompress(encode(CacheEntry("SYS", T0, T1, ROWS))))
+    assert payload["v"] == 1
+    assert payload["columns"][0] == "date"
+
+
+def test_unreadable_input_returns_none(caplog) -> None:
+    caplog.set_level(logging.WARNING)
+    good = encode(CacheEntry("SYS", T0, T1, ROWS))
+    assert decode(b"not gzip at all") is None
+    assert decode(good[:20]) is None
+    assert decode(gzip.compress(json.dumps({"v": 99}).encode())) is None
+    assert decode(gzip.compress(json.dumps({"v": 1}).encode())) is None
+    assert decode(gzip.compress(b"[1, 2]")) is None
+    assert "cache entry" in caplog.text
+
+
+def _envelope(**overrides) -> bytes:
+    payload = {
+        "v": 1, "symbol": "SYS",
+        "fetched_at": T0.isoformat(), "fresh_until": T1.isoformat(),
+        "columns": list(COLUMNS),
+        "rows": [["2026-09-29", 1.0, 1.0, 1.0, 1.0, 1, False]],
+    }
+    payload.update(overrides)
+    return gzip.compress(json.dumps(payload).encode())
+
+
+def test_structurally_wrong_envelopes_return_none() -> None:
+    assert decode(_envelope()) is not None                                  # control
+    assert decode(_envelope(columns=["other"], rows=[[1]])) is None         # wrong columns
+    assert decode(_envelope(columns=list(reversed(COLUMNS)))) is None       # reordered
+    assert decode(_envelope(rows=[["2026-09-29", 1.0]])) is None            # short row
+    assert decode(_envelope(rows=[[20260929, 1.0, 1.0, 1.0, 1.0, 1, False]])) is None  # date type
+    assert decode(_envelope(rows=["not a list"])) is None
+    assert decode(_envelope(rows={"a": 1})) is None
+    assert decode(_envelope(symbol=7)) is None
+    assert decode(_envelope(fetched_at="2026-09-29T05:00:00")) is None      # naive timestamp
+    assert decode(_envelope(fresh_until=12345)) is None
diff --git a/tests/api/test_cache_factory.py b/tests/api/test_cache_factory.py
new file mode 100644
index 0000000..96cee5b
--- /dev/null
+++ b/tests/api/test_cache_factory.py
@@ -0,0 +1,65 @@
+"""Tests for building the /historical cache from environment variables."""
+import logging
+from datetime import timedelta
+
+import redis
+from fastapi.testclient import TestClient
+
+from api.cache.factory import build_historical_service, build_redis_client, market_ttl_from_env
+from api.main import app
+
+
+def test_market_ttl_defaults_to_30_minutes() -> None:
+    assert market_ttl_from_env({}) == timedelta(minutes=30)
+
+
+def test_market_ttl_from_env_seconds() -> None:
+    assert market_ttl_from_env({"HISTORICAL_CACHE_MARKET_TTL": "900"}) == timedelta(minutes=15)
+
+
+def test_invalid_market_ttl_falls_back_with_warning(caplog) -> None:
+    caplog.set_level(logging.WARNING)
+    assert market_ttl_from_env({"HISTORICAL_CACHE_MARKET_TTL": "abc"}) == timedelta(minutes=30)
+    assert market_ttl_from_env({"HISTORICAL_CACHE_MARKET_TTL": "0"}) == timedelta(minutes=30)
+    assert "HISTORICAL_CACHE_MARKET_TTL" in caplog.text
+
+
+def test_no_redis_url_means_memory_only() -> None:
+    assert build_redis_client(None) is None
+    assert build_redis_client("") is None
+    service = build_historical_service({})
+    assert service.store.l2.enabled is False
+
+
+def test_redis_client_uses_short_timeouts_and_does_not_connect_eagerly() -> None:
+    client = build_redis_client("rediss://default:pw@cache.example.invalid:12345")
+    assert isinstance(client, redis.Redis)
+    kwargs = client.connection_pool.connection_kwargs
+    assert kwargs["socket_timeout"] == 0.5
+    assert kwargs["socket_connect_timeout"] == 0.5
+
+
+def test_invalid_redis_url_falls_back_without_leaking_secret(caplog) -> None:
+    caplog.set_level(logging.DEBUG)
+    service = build_historical_service({"REDIS_URL": "http://default:s3cr3t@cache.example:1"})
+    assert service.store.l2.enabled is False
+    assert "REDIS_URL" in caplog.text
+    assert "s3cr3t" not in caplog.text
+
+
+def test_unreachable_redis_at_startup_logs_host_only(caplog) -> None:
+    caplog.set_level(logging.DEBUG)
+    service = build_historical_service(
+        {"REDIS_URL": "redis://default:s3cr3t@127.0.0.1:1"}  # nothing listens on port 1
+    )
+    assert service.store.l2.enabled is True                   # stays configured; fails open
+    assert "127.0.0.1" in caplog.text
+    assert "s3cr3t" not in caplog.text
+    service.close()
+
+
+def test_lifespan_installs_service_without_redis(monkeypatch) -> None:
+    monkeypatch.delenv("REDIS_URL", raising=False)
+    with TestClient(app) as client:
+        assert client.get("/health").status_code == 200
+        assert app.state.historical_service.store.l2.enabled is False
diff --git a/tests/api/test_cache_freshness.py b/tests/api/test_cache_freshness.py
new file mode 100644
index 0000000..4ceed38
--- /dev/null
+++ b/tests/api/test_cache_freshness.py
@@ -0,0 +1,57 @@
+"""Tests for the trading-hours freshness policy."""
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from api.cache.freshness import DEFAULT_MARKET_TTL, PKT, fresh_until
+
+
+def pkt(y: int, m: int, d: int, hh: int, mm: int = 0, ss: int = 0) -> datetime:
+    return datetime(y, m, d, hh, mm, ss, tzinfo=PKT)
+
+
+# 2026-09-29 is a Tuesday; 2026-10-02 Friday, 10-03 Saturday, 10-04 Sunday, 10-05 Monday.
+CASES = [
+    ("pre-open", pkt(2026, 9, 29, 8, 50), pkt(2026, 9, 29, 9, 0)),
+    ("open edge", pkt(2026, 9, 29, 9, 0), pkt(2026, 9, 29, 9, 30)),
+    ("in window", pkt(2026, 9, 29, 10, 0), pkt(2026, 9, 29, 10, 30)),
+    ("capped at close", pkt(2026, 9, 29, 16, 45), pkt(2026, 9, 29, 17, 0)),
+    ("last second", pkt(2026, 9, 29, 16, 59, 59), pkt(2026, 9, 29, 17, 0)),
+    ("close edge", pkt(2026, 9, 29, 17, 0), pkt(2026, 9, 30, 9, 0)),
+    ("evening", pkt(2026, 9, 29, 17, 1), pkt(2026, 9, 30, 9, 0)),
+    ("before midnight", pkt(2026, 9, 29, 23, 59), pkt(2026, 9, 30, 9, 0)),
+    ("friday evening", pkt(2026, 10, 2, 17, 1), pkt(2026, 10, 5, 9, 0)),
+    ("saturday midday", pkt(2026, 10, 3, 12, 0), pkt(2026, 10, 5, 9, 0)),
+    ("saturday early", pkt(2026, 10, 3, 8, 0), pkt(2026, 10, 5, 9, 0)),
+    ("sunday night", pkt(2026, 10, 4, 20, 0), pkt(2026, 10, 5, 9, 0)),
+    ("monday pre-open", pkt(2026, 10, 5, 3, 0), pkt(2026, 10, 5, 9, 0)),
+]
+
+
+@pytest.mark.parametrize(("label", "fetched", "expected"), CASES, ids=[c[0] for c in CASES])
+def test_fresh_until(label: str, fetched: datetime, expected: datetime) -> None:
+    assert fresh_until(fetched) == expected
+
+
+def test_result_is_utc() -> None:
+    result = fresh_until(pkt(2026, 9, 29, 10, 0))
+    assert result.utcoffset() == timedelta(0)
+
+
+def test_accepts_utc_input() -> None:
+    # 05:00 UTC == 10:00 PKT (UTC+5, no DST)
+    fetched = datetime(2026, 9, 29, 5, 0, tzinfo=timezone.utc)
+    assert fresh_until(fetched) == datetime(2026, 9, 29, 5, 30, tzinfo=timezone.utc)
+
+
+def test_custom_market_ttl() -> None:
+    assert fresh_until(pkt(2026, 9, 29, 10, 0), timedelta(minutes=15)) == pkt(2026, 9, 29, 10, 15)
+
+
+def test_default_market_ttl_is_30_minutes() -> None:
+    assert DEFAULT_MARKET_TTL == timedelta(minutes=30)
+
+
+def test_naive_datetime_rejected() -> None:
+    with pytest.raises(ValueError):
+        fresh_until(datetime(2026, 9, 29, 10, 0))
diff --git a/tests/api/test_cache_historical.py b/tests/api/test_cache_historical.py
new file mode 100644
index 0000000..ed09f02
--- /dev/null
+++ b/tests/api/test_cache_historical.py
@@ -0,0 +1,244 @@
+"""Tests for HistoricalService: hit/miss, expiry, stale fallback, cooldown, single-flight."""
+import gzip
+import json
+import threading
+import zlib
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timedelta, timezone
+
+import fakeredis
+import pytest
+from psxdata.exceptions import (
+    InvalidSymbolError,
+    PSXRateLimitError,
+    PSXServerError,
+    PSXUnavailableError,
+)
+
+from api.cache.historical import LOCK_STRIPES, CacheStatus, HistoricalService, cache_key
+from api.cache.store import MemoryLRU, RedisStore, TieredStore
+
+ASCENDING_ROWS = [
+    {"date": "2026-09-28", "open": 9.0, "high": 10.0, "low": 8.5, "close": 9.5,
+     "volume": 900, "is_anomaly": False},
+    {"date": "2026-09-29", "open": 10.0, "high": 11.0, "low": 9.5, "close": 10.5,
+     "volume": 1000, "is_anomaly": False},
+]
+
+
+class Clock:
+    """Tuesday 2026-09-29 10:00 PKT (inside the trading window, 30 min TTL)."""
+
+    def __init__(self) -> None:
+        self.t = datetime(2026, 9, 29, 5, 0, tzinfo=timezone.utc)
+        self.mono = 1000.0
+
+    def now(self) -> datetime:
+        return self.t
+
+    def monotonic(self) -> float:
+        return self.mono
+
+    def advance(self, seconds: float) -> None:
+        self.t += timedelta(seconds=seconds)
+        self.mono += seconds
+
+
+class FakeFetch:
+    def __init__(self, rows=None) -> None:
+        self.rows = ASCENDING_ROWS if rows is None else rows
+        self.error: Exception | None = None
+        self.calls: list[str] = []
+
+    def __call__(self, symbol: str):
+        self.calls.append(symbol)
+        if self.error is not None:
+            raise self.error
+        return [dict(r) for r in self.rows]
+
+
+def make_service(clock: Clock, l2_client=None) -> HistoricalService:
+    store = TieredStore(MemoryLRU(), RedisStore(l2_client, monotonic=clock.monotonic))
+    return HistoricalService(store, now=clock.now, monotonic=clock.monotonic)
+
+
+@pytest.fixture
+def clock() -> Clock:
+    return Clock()
+
+
+def test_miss_then_hit(clock: Clock) -> None:
+    service, fetch = make_service(clock), FakeFetch()
+    first = service.get("SYS", fetch)
+    second = service.get("SYS", fetch)
+    assert first.status is CacheStatus.MISS
+    assert second.status is CacheStatus.HIT
+    assert fetch.calls == ["SYS"]
+    assert [r["date"] for r in second.rows] == [r["date"] for r in first.rows]
+    assert second.fetched_at == clock.t
+
+
+def test_rows_are_newest_first(clock: Clock) -> None:
+    service = make_service(clock)
+    miss = service.get("SYS", FakeFetch())
+    hit = service.get("SYS", FakeFetch())
+    assert [r["date"] for r in miss.rows] == ["2026-09-29", "2026-09-28"]
+    assert [r["date"] for r in hit.rows] == ["2026-09-29", "2026-09-28"]
+
+
+def test_symbol_is_uppercased(clock: Clock) -> None:
+    service, fetch = make_service(clock), FakeFetch()
+    service.get("sys", fetch)
+    assert service.get("SYS", fetch).status is CacheStatus.HIT
+    assert fetch.calls == ["SYS"]
+    assert cache_key("sys") == "psxdata-api:v1:historical:SYS"
+
+
+def test_expired_entry_is_refetched(clock: Clock) -> None:
+    service, fetch = make_service(clock), FakeFetch()
+    service.get("SYS", fetch)
+    clock.advance(31 * 60)
+    assert service.get("SYS", fetch).status is CacheStatus.MISS
+    assert len(fetch.calls) == 2
+
+
+def test_rate_limit_serves_stale_and_starts_cooldown(clock: Clock) -> None:
+    service, fetch = make_service(clock), FakeFetch()
+    primed = service.get("SYS", fetch)
+    clock.advance(31 * 60)
+
+    fetch.error = PSXRateLimitError("PSX rate limit exceeded (429)")
+    stale = service.get("SYS", fetch)
+    assert stale.status is CacheStatus.STALE
+    assert stale.fetched_at == primed.fetched_at
+    assert len(fetch.calls) == 2
+
+    fetch.error = None                       # PSX would answer now, but we are cooling down
+    clock.advance(30)
+    assert service.get("SYS", fetch).status is CacheStatus.STALE
+    assert len(fetch.calls) == 2
+
+    clock.advance(31)                        # cooldown (60 s) over
+    assert service.get("SYS", fetch).status is CacheStatus.MISS
+    assert len(fetch.calls) == 3
+
+
+def test_rate_limit_without_copy_raises_and_cooldown_blocks_other_symbols(clock: Clock) -> None:
+    service, fetch = make_service(clock), FakeFetch()
+    fetch.error = PSXRateLimitError("PSX rate limit exceeded (429)")
+    with pytest.raises(PSXRateLimitError):
+        service.get("SYS", fetch)
+    fetch.error = None
+    with pytest.raises(PSXRateLimitError, match="rate-limiting"):
+        service.get("HUBC", fetch)           # blocked by cooldown, PSX not called
+    assert fetch.calls == ["SYS"]
+
+
+def test_unavailable_serves_stale_without_cooldown(clock: Clock) -> None:
+    service, fetch = make_service(clock), FakeFetch()
+    service.get("SYS", fetch)
+    clock.advance(31 * 60)
+    fetch.error = PSXServerError("PSX 502")
+    assert service.get("SYS", fetch).status is CacheStatus.STALE
+    fetch.error = None
+    assert service.get("SYS", fetch).status is CacheStatus.MISS
+    assert len(fetch.calls) == 3
+
+
+def test_unavailable_without_copy_raises(clock: Clock) -> None:
+    service, fetch = make_service(clock), FakeFetch()
+    fetch.error = PSXUnavailableError("PSX down")
+    with pytest.raises(PSXUnavailableError):
+        service.get("SYS", fetch)
+
+
+def test_other_errors_propagate_and_are_not_cached(clock: Clock) -> None:
+    service, fetch = make_service(clock), FakeFetch()
+    fetch.error = InvalidSymbolError("nope")
+    with pytest.raises(InvalidSymbolError):
+        service.get("NOPE", fetch)
+    assert service.store.get(cache_key("NOPE")) is None
+
+
+def test_empty_history_is_cached(clock: Clock) -> None:
+    service, fetch = make_service(clock), FakeFetch(rows=[])
+    assert service.get("NOPE", fetch).rows == []
+    hit = service.get("NOPE", fetch)
+    assert hit.status is CacheStatus.HIT
+    assert hit.rows == []
+    assert len(fetch.calls) == 1
+
+
+def test_corrupt_cached_bytes_are_a_miss_and_get_overwritten(clock: Clock) -> None:
+    redis_client = fakeredis.FakeRedis()
+    redis_client.set(cache_key("SYS"), b"written by something else")
+    service, fetch = make_service(clock, redis_client), FakeFetch()
+    assert service.get("SYS", fetch).status is CacheStatus.MISS
+    restarted = make_service(clock, redis_client)            # fresh L1, same L2
+    assert restarted.get("SYS", fetch).status is CacheStatus.HIT
+    assert len(fetch.calls) == 1
+
+
+def test_wrong_schema_envelope_is_refetched(clock: Clock) -> None:
+    far_future = (clock.t + timedelta(days=365)).isoformat()
+    bogus = gzip.compress(json.dumps({
+        "v": 1, "symbol": "SYS", "fetched_at": clock.t.isoformat(), "fresh_until": far_future,
+        "columns": ["other"], "rows": [[1]],
+    }).encode())
+    redis_client = fakeredis.FakeRedis()
+    redis_client.set(cache_key("SYS"), bogus)
+    service, fetch = make_service(clock, redis_client), FakeFetch()
+    result = service.get("SYS", fetch)
+    assert result.status is CacheStatus.MISS
+    assert result.rows[0]["date"] == "2026-09-29"
+    assert len(fetch.calls) == 1
+
+
+def test_entry_survives_restart_via_l2(clock: Clock) -> None:
+    redis_client = fakeredis.FakeRedis()
+    fetch = FakeFetch()
+    make_service(clock, redis_client).get("SYS", fetch)
+    assert make_service(clock, redis_client).get("SYS", fetch).status is CacheStatus.HIT
+    assert len(fetch.calls) == 1
+
+
+def test_concurrent_requests_for_one_symbol_fetch_once(clock: Clock) -> None:
+    service = make_service(clock)
+    started, release = threading.Event(), threading.Event()
+    calls: list[str] = []
+
+    def slow_fetch(symbol: str):
+        calls.append(symbol)
+        started.set()
+        release.wait(5)
+        return [dict(r) for r in ASCENDING_ROWS]
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        first = pool.submit(service.get, "SYS", slow_fetch)
+        assert started.wait(5)
+        others = [pool.submit(service.get, "SYS", slow_fetch) for _ in range(7)]
+        release.set()
+        results = [first.result(5)] + [f.result(5) for f in others]
+
+    assert calls == ["SYS"]
+    assert [r.status for r in results].count(CacheStatus.MISS) == 1
+
+
+def test_different_symbols_do_not_block_each_other(clock: Clock) -> None:
+    stripe = lambda s: zlib.crc32(cache_key(s).encode()) % LOCK_STRIPES  # noqa: E731
+    assert stripe("AAA") != stripe("BBB")
+    service = make_service(clock)
+    started, release = threading.Event(), threading.Event()
+
+    def blocking_fetch(symbol: str):
+        started.set()
+        release.wait(5)
+        return []
+
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        blocked = pool.submit(service.get, "AAA", blocking_fetch)
+        assert started.wait(5)
+        assert service.get("BBB", FakeFetch()).status is CacheStatus.MISS
+        assert not blocked.done()
+        release.set()
+        blocked.result(5)
diff --git a/tests/api/test_cache_store.py b/tests/api/test_cache_store.py
new file mode 100644
index 0000000..9d65b03
--- /dev/null
+++ b/tests/api/test_cache_store.py
@@ -0,0 +1,144 @@
+"""Tests for the L1 memory LRU, the fail-open Redis store, and the tiered store."""
+import logging
+
+import fakeredis
+import redis
+
+from api.cache.store import L2_TTL_SECONDS, MemoryLRU, RedisStore, TieredStore
+
+
+class FakeMonotonic:
+    def __init__(self) -> None:
+        self.t = 1000.0
+
+    def __call__(self) -> float:
+        return self.t
+
+
+class DownRedis:
+    """Stands in for an unreachable server: every call raises."""
+
+    def __init__(self) -> None:
+        self.calls = 0
+
+    def get(self, key):
+        self.calls += 1
+        raise redis.ConnectionError("connection refused")
+
+    def set(self, key, value, ex=None):
+        self.calls += 1
+        raise redis.ConnectionError("connection refused")
+
+    def close(self) -> None:
+        pass
+
+
+class CountingRedis(fakeredis.FakeRedis):
+    gets = 0
+
+    def get(self, name):
+        type(self).gets += 1
+        return super().get(name)
+
+
+# ---- MemoryLRU ----------------------------------------------------------------------------
+
+def test_lru_get_set_and_miss() -> None:
+    lru = MemoryLRU(max_bytes=100)
+    lru.set("a", b"1234")
+    assert lru.get("a") == b"1234"
+    assert lru.get("missing") is None
+
+
+def test_lru_evicts_least_recently_used_by_bytes() -> None:
+    lru = MemoryLRU(max_bytes=10)
+    lru.set("a", b"aaaa")
+    lru.set("b", b"bbbb")
+    lru.get("a")                 # a is now most recently used
+    lru.set("c", b"cccc")        # 12 bytes > 10: evict b
+    assert lru.get("b") is None
+    assert lru.get("a") == b"aaaa"
+    assert lru.get("c") == b"cccc"
+    assert lru.size_bytes == 8
+
+
+def test_lru_replacing_a_key_updates_size() -> None:
+    lru = MemoryLRU(max_bytes=100)
+    lru.set("a", b"1234")
+    lru.set("a", b"12345678")
+    assert lru.size_bytes == 8
+
+
+def test_lru_skips_values_larger_than_capacity() -> None:
+    lru = MemoryLRU(max_bytes=4)
+    lru.set("big", b"12345")
+    assert lru.get("big") is None
+    assert lru.size_bytes == 0
+
+
+# ---- RedisStore ---------------------------------------------------------------------------
+
+def test_redis_store_without_client_is_a_noop() -> None:
+    store = RedisStore(None)
+    store.set("k", b"v")
+    assert store.get("k") is None
+    assert store.enabled is False
+    store.close()
+
+
+def test_redis_store_round_trip_with_expiry() -> None:
+    client = fakeredis.FakeRedis()
+    store = RedisStore(client)
+    store.set("k", b"v")
+    assert store.get("k") == b"v"
+    assert 0 < client.ttl("k") <= L2_TTL_SECONDS
+    assert store.enabled is True
+
+
+def test_redis_store_failure_is_swallowed_logged_once_and_backs_off(caplog) -> None:
+    caplog.set_level(logging.WARNING)
+    clock = FakeMonotonic()
+    client = DownRedis()
+    store = RedisStore(client, monotonic=clock)
+
+    assert store.get("k") is None           # fails, starts 30 s backoff
+    store.set("k", b"v")                    # within backoff: client not touched
+    assert store.get("k") is None
+    assert client.calls == 1
+    assert caplog.text.count("Cache backend") == 1
+
+    clock.t += 31                           # backoff over: tries again
+    assert store.get("k") is None
+    assert client.calls == 2
+
+
+# ---- TieredStore --------------------------------------------------------------------------
+
+def test_tiered_l2_hit_fills_l1() -> None:
+    client = fakeredis.FakeRedis()
+    client.set("k", b"v")
+    tiered = TieredStore(MemoryLRU(), RedisStore(client))
+    assert tiered.get("k") == b"v"
+    assert tiered.l1.get("k") == b"v"
+
+
+def test_tiered_l1_hit_skips_l2() -> None:
+    CountingRedis.gets = 0
+    tiered = TieredStore(MemoryLRU(), RedisStore(CountingRedis()))
+    tiered.put("k", b"v")
+    assert tiered.get("k") == b"v"
+    assert CountingRedis.gets == 0
+
+
+def test_tiered_put_writes_both_tiers() -> None:
+    client = fakeredis.FakeRedis()
+    tiered = TieredStore(MemoryLRU(), RedisStore(client))
+    tiered.put("k", b"v")
+    assert tiered.l1.get("k") == b"v"
+    assert client.get("k") == b"v"
+
+
+def test_tiered_works_when_l2_is_down() -> None:
+    tiered = TieredStore(MemoryLRU(), RedisStore(DownRedis()))
+    tiered.put("k", b"v")
+    assert tiered.get("k") == b"v"
diff --git a/tests/api/test_error_envelope.py b/tests/api/test_error_envelope.py
index e7f3924..0fddc16 100644
--- a/tests/api/test_error_envelope.py
+++ b/tests/api/test_error_envelope.py
@@ -2,11 +2,12 @@
 import pytest
 from fastapi import Request
 from fastapi.testclient import TestClient
-from psxdata.exceptions import InvalidSymbolError, PSXUnavailableError
+from psxdata.exceptions import InvalidSymbolError, PSXRateLimitError, PSXUnavailableError
 
 from api.dependencies import limiter
 from api.main import app
 
+
 # Register sentinel routes at module level to avoid mutating `app` inside test bodies.
 # These routes exist only to trigger specific exception types for testing.
 @app.get("/__test_500")
@@ -19,6 +20,11 @@ def _psx_down():
     raise PSXUnavailableError("PSX server unreachable")
 
 
+@app.get("/__test_503_rate_limited")
+def _psx_rate_limited():
+    raise PSXRateLimitError("PSX rate limit exceeded (429) on https://dps.psx.com.pk/historical")
+
+
 @app.get("/__test_404_symbol")
 def _bad_symbol():
     raise InvalidSymbolError("FAKESYM not found on PSX")
@@ -68,6 +74,16 @@ def test_psx_unavailable_maps_to_503(client: TestClient) -> None:
     assert "PSX server unreachable" in body["error"]["message"]
 
 
+def test_psx_rate_limit_maps_to_503_with_retry_after(client: TestClient) -> None:
+    resp = client.get("/__test_503_rate_limited")
+    assert resp.status_code == 503
+    assert resp.headers["Retry-After"] == "60"
+    body = resp.json()
+    assert body["error"]["status"] == 503
+    assert body["error"]["code"] == "psx_unavailable"
+    assert body["error"]["message"] == "PSX is rate-limiting requests; retry later"
+
+
 def test_invalid_symbol_maps_to_404(client: TestClient) -> None:
     resp = client.get("/__test_404_symbol")
     assert resp.status_code == 404
diff --git a/tests/api/test_historical_endpoint.py b/tests/api/test_historical_endpoint.py
new file mode 100644
index 0000000..4c47672
--- /dev/null
+++ b/tests/api/test_historical_endpoint.py
@@ -0,0 +1,193 @@
+"""Tests for GET /stocks/{symbol}/historical with the cache in front of PSX."""
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+import pandas as pd
+import pytest
+import redis
+from fastapi.testclient import TestClient
+from psxdata.exceptions import PSXRateLimitError
+
+from api.cache.historical import HistoricalService
+from api.cache.store import MemoryLRU, RedisStore, TieredStore
+from api.main import app
+
+
+def _history() -> pd.DataFrame:
+    # PSX order: newest first
+    return pd.DataFrame({
+        "date": pd.to_datetime(["2024-01-03", "2024-01-02", "2024-01-01"]),
+        "open": [3.0, 2.0, 1.0], "high": [3.5, 2.5, 1.5], "low": [2.5, 1.5, 0.5],
+        "close": [3.2, 2.2, 1.2], "volume": [300, 200, 100],
+        "is_anomaly": [False, False, False],
+    })
+
+
+class Clock:
+    """Tuesday 2024-01-02 10:00 PKT — in the past, so real-clock Age is positive."""
+
+    def __init__(self) -> None:
+        self.t = datetime(2024, 1, 2, 5, 0, tzinfo=timezone.utc)
+        self.mono = 1000.0
+
+    def now(self) -> datetime:
+        return self.t
+
+    def monotonic(self) -> float:
+        return self.mono
+
+    def advance(self, seconds: float) -> None:
+        self.t += timedelta(seconds=seconds)
+        self.mono += seconds
+
+
+class DownRedis:
+    def get(self, key):
+        raise redis.ConnectionError("down")
+
+    def set(self, key, value, ex=None):
+        raise redis.ConnectionError("down")
+
+    def close(self) -> None:
+        pass
+
+
+@pytest.fixture
+def client() -> TestClient:
+    return TestClient(app, raise_server_exceptions=False)
+
+
+def _install(service_clock: Clock | None = None, l2_client=None) -> None:
+    kwargs = {}
+    if service_clock is not None:
+        kwargs = {"now": service_clock.now, "monotonic": service_clock.monotonic}
+    store = TieredStore(MemoryLRU(), RedisStore(l2_client))
+    app.state.historical_service = HistoricalService(store, **kwargs)
+
+
+def test_first_request_is_miss_then_hit(client: TestClient) -> None:
+    with patch("psxdata.stocks", return_value=_history()) as mock_stocks:
+        first = client.get("/stocks/SYS/historical")
+        second = client.get("/stocks/SYS/historical")
+    assert first.status_code == 200 and second.status_code == 200
+    assert first.headers["X-Cache"] == "MISS"
+    assert second.headers["X-Cache"] == "HIT"
+    assert "Age" not in first.headers
+    assert int(second.headers["Age"]) >= 0
+    assert first.json()["meta"]["cached"] is False
+    assert second.json()["meta"]["cached"] is True
+    assert second.json()["data"] == first.json()["data"]
+    assert [r["date"] for r in first.json()["data"]] == ["2024-01-03", "2024-01-02", "2024-01-01"]
+    mock_stocks.assert_called_once_with("SYS", cache=False)
+
+
+def test_symbol_case_shares_cache_entry(client: TestClient) -> None:
+    with patch("psxdata.stocks", return_value=_history()) as mock_stocks:
+        client.get("/stocks/sys/historical")
+        resp = client.get("/stocks/SyS/historical")
+    assert resp.headers["X-Cache"] == "HIT"
+    mock_stocks.assert_called_once_with("SYS", cache=False)
+
+
+def test_unknown_symbol_empty_history_is_cached(client: TestClient) -> None:
+    with patch("psxdata.stocks", return_value=pd.DataFrame()) as mock_stocks:
+        first = client.get("/stocks/NOPE/historical")
+        second = client.get("/stocks/NOPE/historical")
+    assert first.status_code == 200 and first.json()["data"] == []
+    assert second.headers["X-Cache"] == "HIT" and second.json()["data"] == []
+    assert mock_stocks.call_count == 1
+
+
+@pytest.mark.parametrize(("query", "expected"), [
+    ("", ["2024-01-03", "2024-01-02", "2024-01-01"]),
+    ("?start=2024-01-02", ["2024-01-03", "2024-01-02"]),
+    ("?end=2024-01-02", ["2024-01-02", "2024-01-01"]),
+    ("?start=2024-01-02&end=2024-01-02", ["2024-01-02"]),
+    ("?start=2025-01-01", []),
+])
+def test_date_range_is_sliced_from_cached_history(
+    client: TestClient, query: str, expected: list[str]
+) -> None:
+    with patch("psxdata.stocks", return_value=_history()):
+        resp = client.get(f"/stocks/SYS/historical{query}")
+    assert resp.status_code == 200
+    assert [r["date"] for r in resp.json()["data"]] == expected
+    assert resp.json()["meta"]["count"] == len(expected)
+
+
+@pytest.mark.parametrize("query", [
+    "?start=2024-13-01",
+    "?end=yesterday",
+    "?start=20240102",
+    "?end=2024-W01-2",
+    "?start=2024-01-03&end=2024-01-01",
+])
+def test_bad_dates_return_422_without_calling_psx(client: TestClient, query: str) -> None:
+    with patch("psxdata.stocks", return_value=_history()) as mock_stocks:
+        resp = client.get(f"/stocks/SYS/historical{query}")
+    assert resp.status_code == 422
+    assert resp.json()["error"]["code"] == "bad_request"
+    mock_stocks.assert_not_called()
+
+
+def test_missing_values_behave_like_uncached_on_miss_and_hit(client: TestClient) -> None:
+    # NaN close: 0.2.x returned 200 with close=null (pydantic serialises NaN as null).
+    nan_close = _history()
+    nan_close.loc[1, "close"] = float("nan")
+    with patch("psxdata.stocks", return_value=nan_close):
+        miss = client.get("/stocks/NANC/historical")
+        hit = client.get("/stocks/NANC/historical")
+    assert miss.status_code == 200 and hit.status_code == 200
+    assert hit.headers["X-Cache"] == "HIT"
+    assert miss.json()["data"][1]["close"] is None
+    assert hit.json()["data"] == miss.json()["data"]
+
+    # Missing Int64 volume: 0.2.x returned 502 (OHLCVRow.volume is required). Unchanged.
+    na_volume = _history()
+    na_volume["volume"] = pd.array([300, pd.NA, 100], dtype="Int64")
+    with patch("psxdata.stocks", return_value=na_volume) as mock_stocks:
+        miss = client.get("/stocks/NAVOL/historical")
+        hit = client.get("/stocks/NAVOL/historical")
+    assert miss.status_code == 502 and hit.status_code == 502
+    assert hit.json()["error"]["code"] == "upstream_data_error"
+    assert mock_stocks.call_count == 1
+
+
+def test_psx_rate_limit_serves_stale_copy(client: TestClient) -> None:
+    clock = Clock()
+    _install(clock)
+    with patch("psxdata.stocks", return_value=_history()):
+        client.get("/stocks/SYS/historical")
+    clock.advance(31 * 60)
+    with patch("psxdata.stocks", side_effect=PSXRateLimitError("429")):
+        resp = client.get("/stocks/SYS/historical")
+    assert resp.status_code == 200
+    assert resp.headers["X-Cache"] == "STALE"
+    assert int(resp.headers["Age"]) > 0
+    assert resp.json()["meta"]["cached"] is True
+    assert len(resp.json()["data"]) == 3
+
+
+def test_psx_rate_limit_without_copy_returns_503(client: TestClient) -> None:
+    with patch("psxdata.stocks", side_effect=PSXRateLimitError("429")):
+        resp = client.get("/stocks/SYS/historical")
+    assert resp.status_code == 503
+    assert resp.headers["Retry-After"] == "60"
+    assert resp.json()["error"]["code"] == "psx_unavailable"
+
+
+def test_redis_down_still_serves(client: TestClient) -> None:
+    _install(l2_client=DownRedis())
+    with patch("psxdata.stocks", return_value=_history()) as mock_stocks:
+        first = client.get("/stocks/SYS/historical")
+        second = client.get("/stocks/SYS/historical")
+    assert first.status_code == 200 and first.headers["X-Cache"] == "MISS"
+    assert second.status_code == 200 and second.headers["X-Cache"] == "HIT"
+    assert mock_stocks.call_count == 1
+
+
+def test_openapi_documents_cache_headers_and_503() -> None:
+    op = app.openapi()["paths"]["/stocks/{symbol}/historical"]["get"]
+    assert "X-Cache" in op["responses"]["200"]["headers"]
+    assert "Age" in op["responses"]["200"]["headers"]
+    assert "Retry-After" in op["responses"]["503"]["headers"]
diff --git a/tests/api/test_stocks.py b/tests/api/test_stocks.py
index cd68cdf..6420bfe 100644
--- a/tests/api/test_stocks.py
+++ b/tests/api/test_stocks.py
@@ -4,7 +4,7 @@
 import pandas as pd
 import pytest
 from fastapi.testclient import TestClient
-from psxdata.exceptions import PSXParseError, PSXUnavailableError
+from psxdata.exceptions import PSXParseError, PSXRateLimitError, PSXUnavailableError
 
 from api.main import app
 
@@ -63,10 +63,10 @@ def test_historical_empty_returns_200_not_404(client: TestClient) -> None:
     assert resp.json()["data"] == []
 
 
-def test_historical_passes_date_params(client: TestClient) -> None:
+def test_historical_fetches_full_history_without_sdk_cache(client: TestClient) -> None:
     with patch("psxdata.stocks", return_value=pd.DataFrame()) as mock_stocks:
         client.get("/stocks/ENGRO/historical?start=2024-01-01&end=2024-12-31")
-    mock_stocks.assert_called_once_with("ENGRO", start="2024-01-01", end="2024-12-31")
+    mock_stocks.assert_called_once_with("ENGRO", cache=False)
 
 
 def test_quote_returns_200(client: TestClient) -> None:
@@ -139,3 +139,11 @@ def test_stocks_cors_header_present(client: TestClient) -> None:
     with patch("psxdata.tickers", return_value=[]):
         resp = client.get("/stocks", headers={"Origin": "https://example.com"})
     assert "access-control-allow-origin" in resp.headers
+
+
+def test_quote_psx_rate_limited_returns_503(client: TestClient) -> None:
+    with patch("psxdata.quote", side_effect=PSXRateLimitError("429")):
+        resp = client.get("/stocks/ENGRO/quote")
+    assert resp.status_code == 503
+    assert resp.headers["Retry-After"] == "60"
+    assert resp.json()["error"]["code"] == "psx_unavailable"