Anomaly Detection in Payment Streams | Dataford Interview Questions

How would you implement a function to detect anomalies in a stream of incoming payment data?

Implement detect_anomalies(payments, window_size, z_threshold) for an ordered stream. Each payment is a dictionary with user_id, timestamp, and numeric amount; compare each payment only with that user's previous payments in the rolling window. Return zero-based indices where the amount differs from the previous mean by more than z_threshold * (maximum_amount - minimum_amount); with a zero range, any different amount is anomalous. Do not classify a payment until at least two prior payments for that user are available.

Constraints

0 <= len(payments) <= 5000
1 <= window_size
0 < z_threshold
Each payment contains string user_id, numeric timestamp, and numeric amount
Payments are processed in the input order
A payment is evaluated only after at least two prior payments for its user

Function Signature

def detect_anomalies(payments, window_size, z_threshold):

Your question is Anomaly Detection in Payment Streams. Start with the requirements in the Question tab.

Run and submit as often as you like. When you're ready, talk me through your approach or go straight to the code.

Problem

How would you implement a function to detect anomalies in a stream of incoming payment data?

Constraints

0 <= len(payments) <= 5000
1 <= window_size
0 < z_threshold
Each payment contains string user_id, numeric timestamp, and numeric amount
Payments are processed in the input order
A payment is evaluated only after at least two prior payments for its user

Function Signature

def detect_anomalies(payments, window_size, z_threshold):

Your question is Anomaly Detection in Payment Streams. Start with the requirements in the Question tab.

Run and submit as often as you like. When you're ready, talk me through your approach or go straight to the code.