Real-Time Streaming & Caching

The Architectural Primer: Why Kafka and Redis?

In a live financial environment, you cannot afford to drop transactions during a traffic spike, nor can you afford slow response times while a heavy machine learning model processes data.

To solve this, the architecture separates the ingestion of data from the processing of data:

1. Apache Kafka (The Shock Absorber): Instead of applications sending HTTP requests directly to an API, they publish events to a Kafka topic. Kafka holds these messages in a highly available queue. If the downstream ML pipeline gets overwhelmed, Kafka provides natural backpressure, ensuring zero data loss

2. Redis (The Shield): Fraudsters frequently test the same stolen credit card multiple times in rapid succession. Running a full XGBoost inference and database write for identical, back-to-back requests is a waste of compute. Redis acts as a read-through cache; if a transaction was flagged 50 milliseconds ago, the API serves the cached result instantly, bypassing the ML model entirely

The Streaming Consumer

The pipeline starts with a dedicated Python consumer running in its own Kubernetes pod. It continuously polls the transaction_stream Kafka topic. When it pulls a batch of transactions, it forwards them to the internal FastAPI inference service

# src/data_pipeline/consumer.py (Simplified)
def process_transaction(transaction_data):
    try:
        # Wrap the single transaction in a list for batch API processing
        payload = [transaction_data]
        response = requests.post(API_URL, json=payload, timeout=5)
        
        if response.status_code == 200:
            return response.json()["batch_results"][0]
            
    except Exception as e:
        logger.error(f"Failed to call API: {e}")
        return None

# The infinite polling loop
while True:
    msg = consumer.poll(1.0)
    if msg is None or msg.error():
        continue

    tx_data = json.loads(msg.value().decode('utf-8'))
    prediction = process_transaction(tx_data)
    # ... logging logic based on prediction ...

The Inference API: Cache Lookup & Async Write-Back

When the FastAPI service receives the request, it executes a strict three-phase lifecycle to ensure sub-100ms latency.

Phase 1: Cache Lookup Before touching the ML model, the API queries Redis using the transaction_id

# src/api/app.py - Inside predict_batch()
for i, tx in enumerate(transactions):
    cached_result = None
    if r_client:
        cached_json = r_client.get(f"pred:{tx.transaction_id}")
        if cached_json:
            cached_result = json.loads(cached_json)
            cached_result["source"] = "cache" 
    
    if cached_result:
        results_map[i] = cached_result
    else:
        indices_to_compute.append(i)
        txs_to_compute.append(tx)

Phase 2: ML Inference If the transaction isn’t in the cache, the API formats the raw features into a Pandas DataFrame and passes them to the XGBoost model pipeline (which was loaded into memory during the application’s startup lifespan)

if txs_to_compute:
    start_time = time.time()
    df = pd.DataFrame([t.dict(exclude={'transaction_id'}) for t in txs_to_compute])
    
    # Execute XGBoost pipeline
    preds = pipeline.predict(df)
    probs = pipeline.predict_proba(df)[:, 1]
    inference_time = (time.time() - start_time) * 1000

Phase 3: The Write-Back & Async Offloading This is where the system design shines. Generating LLM vector embeddings for the database and writing to PostgreSQL are high-latency operations (often taking hundreds of milliseconds).

Instead of making the user wait for the database write to finish, the API instantly caches the prediction in Redis (with a 1-hour TTL) and delegates the heavy database operations to FastAPI’s BackgroundTasks. The HTTP response is returned immediately

for j, (pred, prob) in enumerate(zip(preds, probs)):
    tx_data = txs_to_compute[j]
    tx_id = tx_data.transaction_id
    
    result = {
        "transaction_id": tx_id,
        "is_fraud": bool(pred),
        "fraud_probability": float(prob),
        "source": "model"
    }
    
    # 1. Cache immediately to shield against rapid retry attacks
    if r_client:
        r_client.setex(f"pred:{tx_id}", 3600, json.dumps(result))
    
    # 2. Offload heavy lifting (Embeddings & SQL) so the API returns instantly
    background_tasks.add_task(
        process_embedding_and_log,
        tx_id, tx_data, pred, prob, inference_time
    )

By decoupling the ingestion stream with Kafka and leveraging Redis alongside asynchronous background tasks, the core inference loop remains entirely compute-bound, resulting in massive throughput capabilities.