SLIs, SLOs, SLAs Definition and Implementation Questions

WITH latest AS (
  SELECT DISTINCT ON (txn_id)
    txn_id,
    event_type,
    event_time,
    (event_time AT TIME ZONE 'UTC')::date AS event_date
  FROM events
  ORDER BY txn_id, event_time DESC -- latest event per txn
),
per_day AS (
  SELECT
    event_date,
    COUNT(*) FILTER (WHERE event_type = 'initiated') AS initiated,
    COUNT(*) FILTER (WHERE event_type = 'confirmed') AS confirmed
  FROM latest
  GROUP BY event_date
)
SELECT event_date, confirmed, initiated,
       CASE WHEN initiated=0 THEN NULL ELSE confirmed::decimal/initiated END AS success_rate
FROM per_day
ORDER BY event_date;

from collections import deque
from datetime import datetime, timedelta
from typing import List, Tuple, Optional

def compute_burn_rate(events: List[Tuple[str, bool]],
                      slo_error_rate: float = 0.001,
                      sample_interval_min: int = 5,
                      window_min: int = 60) -> List[Tuple[str, Optional[float]]]:
    """
    events: list of (ISOtimestamp, success_bool). success_bool=True => success.
    Returns list of (ISO sample time, burn_rate or None if no data).
    """
    if not events:
        return []
    # parse and sort
    parsed = [(datetime.fromisoformat(t), not ok) for t, ok in events]  # store is_error
    parsed.sort(key=lambda x: x[0])
    start = parsed[0][0]
    end = parsed[-1][0]
    sample_interval = timedelta(minutes=sample_interval_min)
    window = timedelta(minutes=window_min)

    q = deque()  # (ts, is_error)
    total = 0
    errors = 0
    res = []
    i = 0  # index into parsed events

    sample_time = start.replace(second=0, microsecond=0)
    # ensure first sample covers first event
    while sample_time + window < start:
        sample_time += sample_interval

    while sample_time <= end:
        window_start = sample_time
        window_end = sample_time + window

        # add events up to window_end
        while i < len(parsed) and parsed[i][0] < window_end:
            ts, is_err = parsed[i]
            q.append((ts, is_err))
            total += 1
            errors += 1 if is_err else 0
            i += 1

        # remove events before window_start
        while q and q[0][0] < window_start:
            _, is_err = q.popleft()
            total -= 1
            errors -= 1 if is_err else 0

        if total == 0:
            burn = None
        else:
            err_rate = errors / total
            burn = err_rate / slo_error_rate

        res.append((window_start.isoformat(), burn))
        sample_time += sample_interval

    return res