Katkat

Implement a CSAT Sliding Window Tracker

DecagonSoftware Engineer · Technical Screen

Last updated

This question evaluates the ability to design and implement efficient data structures and algorithms for a sliding time-window tracker, covering time-based expiration, stateful updates, rolling averages with rounding, and nearest-rank percentile computation.

Design and implement a class that tracks customer satisfaction scores over a sliding time window.

A customer support conversation has:

  • conversation_id: str: a unique identifier for the conversation.
  • score: int: an integer score from 1 to 5, inclusive.
  • timestamp: int: an integer timestamp. Calls to record are made with strictly increasing timestamps.

The class should support the following API:

class CSATTracker:
 def __init__(self, window_size: int):
 pass

 def record(self, conversation_id: str, score: int, timestamp: int) -> None:
 pass

 def get_average(self, current_time: int) -> float:
 pass

 def update(self, conversation_id: str, score: int) -> None:
 pass

 def get_percentile(self, current_time: int, p: int) -> int | None:
 pass

Requirements:

  1. __init__(window_size) initializes the tracker. A conversation is considered active at time current_time if current_time - timestamp <= window_size.
  2. record(conversation_id, score, timestamp) records a new conversation. You may assume each conversation_id is recorded at most once.
  3. get_average(current_time) returns the average score among all active conversations at current_time, rounded to two decimal places using round(x, 2). If there are no active conversations, return 0.0.
  4. update(conversation_id, score) updates the score of an existing conversation only if that conversation is still active according to the latest time observed by the tracker. Updating a score must not change the original timestamp or ordering. If the conversation does not exist or has expired, do nothing.
  5. get_percentile(current_time, p) returns the score at percentile p among active conversations at current_time, where 1 <= p <= 100. Use the nearest-rank definition: sort active scores in ascending order and return the element at index ceil(p / 100 * n) - 1. If there are no active conversations, return None.

Optimize for efficient expiration of old conversations and efficient queries.