Revenue Management Pricing & forecasting pipelines
4 min read Pricing

Serving real-time rate recommendations via FastAPI

The read path for pricing has a brutal contract: answer in a few milliseconds, never publish a stale rate, and let the caller reconstruct exactly which decision produced the number. This guide builds that endpoint with FastAPI, implementing the serving layer described in Price Recommendation Serving within the Dynamic Pricing Rule Engines & Optimization pillar. The result is a cache-backed API that fails loud on staleness and stamps every response with the engine version.

Prerequisites

  • Python 3.11+
  • fastapi and an ASGI server such as uvicorn
  • A cache populated by the pricing engine’s write path (in-process dict here; Redis in production)
  • Rates in integer minor units and a canonical currency, per Rate Plan Structuring & Mapping
  • A freshness SLA (TTL) agreed with the consuming Channel Manager Integration Patterns team

Step 1 — Define the recommendation record and cache

The served object is deliberately small and fully self-describing: rate, currency, when it was computed, and which engine version produced it. The computed-at timestamp is what lets the read path enforce freshness.

python 
from __future__ import annotations

import logging
import time
from dataclasses import dataclass

logger = logging.getLogger("pricing.serving")


@dataclass(frozen=True)
class Recommendation:
    rate_cents: int
    currency: str
    computed_at: float       # time.monotonic() at write
    engine_version: str


_CACHE: dict[tuple[str, str, str], Recommendation] = {}
_TTL_SECONDS = 900


def put(property_id: str, room_type: str, stay_date: str,
        rec: Recommendation) -> None:
    _CACHE[(property_id, room_type, stay_date)] = rec

Step 2 — Serve reads, failing loud on staleness

The endpoint returns the rate only when it is within the freshness SLA. A cache miss is a 404; a stale entry is a 503, signaling the caller to fall back or retry rather than publish an old price.

python 
from fastapi import FastAPI, HTTPException

app = FastAPI()


@app.get("/rate/{property_id}/{room_type}/{stay_date}")
def get_rate(property_id: str, room_type: str, stay_date: str) -> dict:
    rec = _CACHE.get((property_id, room_type, stay_date))
    if rec is None:
        logger.warning("miss %s/%s/%s", property_id, room_type, stay_date)
        raise HTTPException(status_code=404, detail="no recommendation")
    age = time.monotonic() - rec.computed_at
    if age > _TTL_SECONDS:
        logger.error("stale %s/%s/%s age=%.0fs", property_id, room_type, stay_date, age)
        raise HTTPException(status_code=503, detail="recommendation stale")
    return {
        "rate_cents": rec.rate_cents,
        "currency": rec.currency,
        "engine_version": rec.engine_version,
        "age_seconds": round(age),
    }

Step 3 — Add a health and freshness probe

Serving infrastructure needs a way for orchestration to know it is healthy and fresh. A liveness check that returns 200 while the cache is full of stale rates is worse than useless; expose freshness explicitly.

python 
@app.get("/healthz")
def healthz() -> dict:
    now = time.monotonic()
    total = len(_CACHE)
    stale = sum(1 for r in _CACHE.values() if now - r.computed_at > _TTL_SECONDS)
    status = "degraded" if total and stale / total > 0.1 else "ok"
    return {"status": status, "cached": total, "stale": stale}

Reporting degraded when more than ten percent of entries are stale turns a silent recompute backlog into an actionable signal your dashboards and alerts can key on.

Verification and testing

FastAPI’s TestClient drives the endpoint end to end, including the staleness path, by writing an entry with a backdated timestamp.

python 
from fastapi.testclient import TestClient


def test_stale_entry_returns_503() -> None:
    client = TestClient(app)
    put("h1", "suite", "2026-08-01",
        Recommendation(rate_cents=22000, currency="USD",
                       computed_at=time.monotonic() - 10_000,   # older than TTL
                       engine_version="v9"))
    resp = client.get("/rate/h1/suite/2026-08-01")
    assert resp.status_code == 503

    put("h1", "suite", "2026-08-02",
        Recommendation(rate_cents=22000, currency="USD",
                       computed_at=time.monotonic(), engine_version="v9"))
    fresh = client.get("/rate/h1/suite/2026-08-02")
    assert fresh.status_code == 200
    assert fresh.json()["engine_version"] == "v9"

Common pitfalls and edge cases

  • Serving stale on miss. Never fall back to an expired entry to avoid a 404; a wrong rate is costlier than a retry.
  • Wall-clock for age. Use time.monotonic() for age math so an NTP adjustment during a long uptime cannot make a rate look fresh.
  • Unversioned responses. Without engine_version, a bad rollout is unattributable; always stamp it.
  • No freshness in health checks. A liveness probe that ignores staleness lets a degraded node keep taking traffic.
  • Blocking recompute in the request. Keep the read path pure-read; enqueue recomputation, don’t run it inline.