test_cache.py

import pandas as pd
import numpy as np
import joblib
from collections import defaultdict
from data_generator import generate_skewed_data


class MLCache:
    def __init__(self, capacity, model):
        self.capacity = capacity
        self.cache = set()
        self.freq = defaultdict(int)
        self.model = model
        self.hits = 0
        self.requests = 0

    def access(self, key):
        self.requests += 1
        self.freq[key] += 1
        features = np.array([[key, self.freq[key]]])

        if key in self.cache:
            self.hits += 1
        else:
            if len(self.cache) >= self.capacity:
                probs = {
                    k: self.model.predict_proba(np.array([[k, self.freq[k]]]))[0][1]
                    for k in self.cache
                }
                evict = min(probs, key=probs.get)
                self.cache.remove(evict)
            self.cache.add(key)

    def get_hit_rate(self):
        return self.hits / self.requests if self.requests > 0 else 0.0


if __name__ == "__main__":
    NUM_REQUESTS = 10000
    UNIQUE_ITEMS = 100
    CACHE_CAPACITY = 10
    SKEW = 1.3

    print("Generating test request stream...")
    test_data = generate_skewed_data(NUM_REQUESTS, UNIQUE_ITEMS, SKEW)

    print("Loading trained best model...")
    model = joblib.load("models/best_model.pkl")

    print("Running smart cache simulation...")
    cache = MLCache(CACHE_CAPACITY, model)
    for item in test_data:
        cache.access(item)

    hit_rate = cache.get_hit_rate()
    print(f"Smart cache hit rate: {hit_rate:.4f}")