Build an in-memory live price cache

Most strategy code shouldn't touch the tick queue directly — it should read "current price of X" from a simple, thread-safe cache that the tick consumer keeps updated.

import threading
import time

class PriceCache:
    def __init__(self):
        self._lock = threading.Lock()
        self._prices: dict[int, dict] = {}   # token -> {price, ts, ohlc, ...}

    def update(self, tick: dict):
        with self._lock:
            self._prices[tick["instrument_token"]] = {
                "price": tick["last_price"],
                "ts": time.time(),
                "ohlc": tick.get("ohlc"),
                "depth": tick.get("depth"),
            }

    def get(self, token: int) -> dict | None:
        with self._lock:
            return self._prices.get(token)

    def get_price(self, token: int) -> float | None:
        entry = self.get(token)
        return entry["price"] if entry else None

    def is_stale(self, token: int, max_age: float = 30.0) -> bool:
        entry = self.get(token)
        return entry is None or (time.time() - entry["ts"]) > max_age

cache = PriceCache()

def process_tick(tick):
    cache.update(tick)

# elsewhere, in strategy logic:
price = cache.get_price(infy_token)
if price is not None and not cache.is_stale(infy_token):
    evaluate_signal(price)

Why a cache instead of reading the queue directly in strategy code

1. Decoupling — strategy logic asks "what's the price now?" without knowing anything about queues, threads, or WebSocket internals.
2. Latest-value semantics — a strategy loop usually wants the *most recent* price, not to process every intermediate tick — the cache naturally gives you that by overwriting on each update.
3. Staleness is checkable — is_stale() lets every consumer of price data independently guard against acting on data from a dead feed (chapter 40), instead of duplicating that check everywhere ad hoc.

This cache is the foundation the strategy chapters (79+) build signal logic on top of.

Next: 043 — Anatomy of an order