Fraud Detection & Anomaly Detection Questions

from math import pow
from datetime import datetime

def time_decay_features(transactions, decay):
    """
    transactions: iterable of dicts sorted by 'ts' ascending,
      each {'id':..., 'ts': datetime, 'amount': float}
    decay: float in (0,1), per-hour decay factor
    Returns list of (id, decayed_sum_before_this_tx)
    """
    results = []
    acc = 0.0                # accumulator stores sum(amount * decay^age_hours) relative to last_ts
    last_ts = None

    for tx in transactions:
        ts = tx['ts']
        amt = tx['amount']

        if last_ts is None:
            delta_h = 0.0
        else:
            delta_h = (ts - last_ts).total_seconds() / 3600.0

        # decay the accumulator to current timestamp
        if delta_h != 0:
            acc *= pow(decay, delta_h)

        # acc now equals decayed sum of all prior transactions at this tx time
        results.append((tx['id'], acc))

        # include this transaction into accumulator for future records
        acc += amt  # stored as amount * decay^0

        last_ts = ts

    return results

def precision_at_k(y_true, y_score, k):
    """
    Returns precision among the top-k scored instances.
    y_true: list or iterable of 0/1 labels
    y_score: list or iterable of floats (same length as y_true)
    k: integer, 1 <= k <= len(y_true)
    """
    if k <= 0:
        raise ValueError("k must be >= 1")
    n = len(y_true)
    if n != len(y_score):
        raise ValueError("y_true and y_score must have the same length")
    if k > n:
        raise ValueError("k must be <= number of instances")

    # Pair scores with labels and sort by score descending
    paired = list(zip(y_score, y_true))
    paired.sort(key=lambda x: x[0], reverse=True)

    # Count positives in top-k
    top_k = paired[:k]
    positives = sum(label for (_, label) in top_k)

    return positives / k