Weekly Cohort Retention Matrix and Churn Watch with DuckDB

id: cohort-retention-duckdb
namespace: company.team
description: |
  Build a weekly cohort retention matrix from an events parquet file with embedded DuckDB SQL, then let a small Python analyzer flag any cohort whose retention fell further than your threshold. Flagged cohorts post to Slack with the drops and a link to the full report.

inputs:
  - id: cohort_weeks
    type: INT
    defaults: 8
    description: How many of the newest weekly cohorts stay in the comparison
      window, newest included.

  - id: churn_drop_threshold_pct
    type: FLOAT
    defaults: 15.0
    description: Percentage points of retention loss against the trailing average
      that flags a cohort as churning.

tasks:
  # --- 1. seed a sample events parquet ---
  - id: seed_sample_events
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: Write a deterministic sample events parquet with COPY TO so the
      flow runs before you point it at a warehouse.
    outputFiles:
      - events.parquet
    sql: |
      COPY (
        WITH cohort_offsets AS (
          SELECT
            generate_series AS weeks_ago,
            CAST(date_trunc('week', CURRENT_DATE) AS DATE)
              - CAST(7 * generate_series AS INTEGER) AS cohort_week
          FROM generate_series(1, 8)
        ),
        cohort_users AS (
          SELECT
            c.weeks_ago,
            c.cohort_week,
            100 * c.weeks_ago + u.generate_series AS user_num
          FROM cohort_offsets c, generate_series(1, 20) u
        ),
        user_weeks AS (
          SELECT
            cu.*,
            w.generate_series AS weeks_since
          FROM cohort_users cu, generate_series(0, 8) w
          WHERE w.generate_series <= cu.weeks_ago
        ),
        keep_prob AS (
          SELECT
            *,
            CASE weeks_since
              WHEN 0 THEN 100
              WHEN 1 THEN 85
              WHEN 2 THEN 70
              WHEN 3 THEN 60
              WHEN 4 THEN 50
              WHEN 5 THEN 45
              WHEN 6 THEN 40
              WHEN 7 THEN 35
              ELSE 30
            END - CASE WHEN weeks_ago = 1 AND weeks_since >= 1 THEN 40 ELSE 0 END AS keep_pct
          FROM user_weeks
        )
        SELECT
          'user_' || lpad(user_num::VARCHAR, 4, '0') AS user_id,
          cohort_week + CAST(7 * weeks_since AS INTEGER) AS event_date,
          'app_open' AS event_name
        FROM keep_prob
        WHERE (user_num * 37 + weeks_since * 13) % 100 < keep_pct
      ) TO '{{ outputFiles["events.parquet"] }}' (FORMAT parquet);

  # --- 2. cohort retention matrix ---
  - id: cohort_matrix
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: Mount the events parquet through inputFiles, pivot it into one row
      per cohort and weeks-since pair, and COPY the matrix out as CSV.
    inputFiles:
      events.parquet: "{{ outputs.seed_sample_events.outputFiles['events.parquet'] }}"
    outputFiles:
      - cohort_matrix.csv
    sql: |
      COPY (
        WITH events AS (
          SELECT user_id, CAST(event_date AS TIMESTAMP) AS event_ts
          FROM read_parquet('events.parquet')
        ),
        first_activity AS (
          SELECT user_id, date_trunc('week', MIN(event_ts)) AS cohort_week
          FROM events
          GROUP BY user_id
        ),
        weekly AS (
          SELECT
            f.cohort_week,
            f.user_id,
            date_diff('week', f.cohort_week, date_trunc('week', e.event_ts)) AS weeks_since
          FROM events e
          JOIN first_activity f ON e.user_id = f.user_id
        ),
        retained AS (
          SELECT cohort_week, weeks_since, COUNT(DISTINCT user_id) AS retained_users
          FROM weekly
          WHERE weeks_since >= 0
          GROUP BY cohort_week, weeks_since
        ),
        sizes AS (
          SELECT cohort_week, COUNT(DISTINCT user_id) AS cohort_size
          FROM first_activity
          GROUP BY cohort_week
        )
        SELECT
          s.cohort_week,
          r.weeks_since,
          s.cohort_size,
          r.retained_users,
          ROUND(100.0 * r.retained_users / s.cohort_size, 1) AS retention_pct
        FROM retained r
        JOIN sizes s ON r.cohort_week = s.cohort_week
        ORDER BY s.cohort_week, r.weeks_since
      ) TO '{{ outputFiles["cohort_matrix.csv"] }}' (HEADER, DELIMITER ',');

  # --- 3. compare latest cohort against the trailing average ---
  - id: analyze_cohorts
    type: io.kestra.plugin.scripts.python.Script
    description: Read the matrix CSV, compare the newest cohort cell by cell with
      the trailing average, and emit the verdict through the json output
      protocol.
    containerImage: python:3.11-slim
    inputFiles:
      cohort_matrix.csv: "{{ outputs.cohort_matrix.outputFiles['cohort_matrix.csv'] }}"
      config.json: |
        {"cohort_weeks": {{ inputs.cohort_weeks }}, "churn_drop_threshold_pct": {{ inputs.churn_drop_threshold_pct }}}
    outputFiles:
      - cohort_report.json
    script: |
      import csv
      import json
      import os


      def read_config():
          if not os.path.exists("config.json"):
              return {}
          with open("config.json", encoding="utf-8") as handle:
              try:
                  return json.load(handle)
              except ValueError:
                  return {}


      def read_matrix():
          if not os.path.exists("cohort_matrix.csv"):
              return []
          cells = []
          try:
              with open("cohort_matrix.csv", newline="", encoding="utf-8") as handle:
                  for row in csv.DictReader(handle):
                      cohort = (row.get("cohort_week") or "").strip()
                      if not cohort:
                          continue
                      try:
                          weeks_since = int(float(row.get("weeks_since")))
                          retention_pct = float(row.get("retention_pct"))
                          cohort_size = int(float(row.get("cohort_size")))
                      except (TypeError, ValueError):
                          continue
                      if weeks_since < 0 or retention_pct < 0 or cohort_size <= 0:
                          continue
                      cells.append({
                          "cohort_week": cohort,
                          "weeks_since": weeks_since,
                          "cohort_size": cohort_size,
                          "retention_pct": retention_pct,
                      })
          except (csv.Error, UnicodeDecodeError):
              return []
          return cells


      config = read_config()
      try:
          threshold = float(config.get("churn_drop_threshold_pct", 15.0))
      except (TypeError, ValueError):
          threshold = 15.0
      try:
          cohort_weeks = int(config.get("cohort_weeks", 8))
      except (TypeError, ValueError):
          cohort_weeks = 8

      matrix = {}
      for cell in read_matrix():
          matrix.setdefault(cell["cohort_week"], {})[cell["weeks_since"]] = cell["retention_pct"]

      all_cohorts = sorted(matrix)
      cohorts = all_cohorts[-cohort_weeks:] if cohort_weeks > 0 else all_cohorts

      flagged_cohorts = {}
      worst_drop_pct = 0.0
      worst_weeks_since = None
      deltas = []

      if len(cohorts) >= 2:
          latest = cohorts[-1]
          earlier = cohorts[:-1]
          for weeks_since in sorted(matrix[latest]):
              trailing = [matrix[c][weeks_since] for c in earlier if weeks_since in matrix[c]]
              if not trailing:
                  continue
              trailing_avg = sum(trailing) / len(trailing)
              retention_pct = matrix[latest][weeks_since]
              drop = trailing_avg - retention_pct
              deltas.append(retention_pct - trailing_avg)
              if drop > worst_drop_pct:
                  worst_drop_pct = drop
                  worst_weeks_since = weeks_since
              if drop > threshold:
                  flagged_cohorts.setdefault(latest, []).append({
                      "weeks_since": weeks_since,
                      "retention_pct": round(retention_pct, 1),
                      "trailing_avg_pct": round(trailing_avg, 1),
                      "drop_pct": round(drop, 1),
                  })

      if not matrix or not deltas:
          verdict = "insufficient_data"
      elif flagged_cohorts:
          verdict = "churn_flagged"
      else:
          verdict = "healthy"

      flagged = bool(flagged_cohorts)

      if deltas:
          mean_delta = sum(deltas) / len(deltas)
          if mean_delta <= -1.0:
              overall_trend = "declining"
          elif mean_delta >= 1.0:
              overall_trend = "improving"
          else:
              overall_trend = "stable"
      else:
          overall_trend = "unknown"

      if flagged:
          parts = []
          for cohort in sorted(flagged_cohorts):
              worst_cell = max(flagged_cohorts[cohort], key=lambda item: item["drop_pct"])
              parts.append(
                  "%s down %s points at week %s"
                  % (cohort, worst_cell["drop_pct"], worst_cell["weeks_since"])
              )
          slack_line = "; ".join(parts) + "; trend " + overall_trend
      else:
          slack_line = (
              "no cohort fell more than %s points below the trailing average; trend %s"
              % (threshold, overall_trend)
          )

      report = {
          "verdict": verdict,
          "threshold_pct": threshold,
          "cohort_window": cohort_weeks,
          "cohorts_considered": cohorts,
          "flagged_cohorts": flagged_cohorts,
          "worst_drop_pct": round(worst_drop_pct, 1),
          "worst_weeks_since": worst_weeks_since,
          "overall_trend": overall_trend,
          "cohort_count": len(matrix),
          "matrix": matrix,
      }
      with open("cohort_report.json", "w", encoding="utf-8") as handle:
          json.dump(report, handle, indent=2, sort_keys=True)

      payload = {
          "outputs": {
              "verdict": verdict,
              "flagged": flagged,
              "flagged_cohorts": flagged_cohorts,
              "flagged_cohorts_json": json.dumps(flagged_cohorts, sort_keys=True),
              "worst_drop_pct": round(worst_drop_pct, 1),
              "worst_weeks_since": worst_weeks_since,
              "overall_trend": overall_trend,
              "cohort_count": len(matrix),
              "slack_line": slack_line,
          }
      }
      print("::" + json.dumps(payload) + "::")

  # --- 4. route on the verdict ---
  - id: route_churn
    type: io.kestra.plugin.core.flow.If
    description: Alert the channel only when at least one cohort dropped past the
      threshold, otherwise log a green result.
    condition: "{{ outputs.analyze_cohorts.vars.flagged }}"
    then:
      - id: notify_churn
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post the flagged drops, the trend, and the report file reference to
          the team channel.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "COHORT CHURN {{ flow.id }} ({{ execution.id }}): {{ outputs.analyze_cohorts.vars.slack_line }}. Report {{ outputs.analyze_cohorts.outputFiles['cohort_report.json'] }}"
          }
      - id: log_flagged
        type: io.kestra.plugin.core.log.Log
        description: Record the flagged cohorts in the execution log for the audit trail.
        message: "flagged {{ outputs.analyze_cohorts.vars.cohort_count }} cohorts
          window, verdict {{ outputs.analyze_cohorts.vars.verdict }}, worst drop
          {{ outputs.analyze_cohorts.vars.worst_drop_pct }} points"
    else:
      - id: log_healthy
        type: io.kestra.plugin.core.log.Log
        description: Green path, no cohort fell further than the threshold against its
          trailing average.
        message: "ALL GREEN {{ flow.id }}, verdict {{
          outputs.analyze_cohorts.vars.verdict }}, trend {{
          outputs.analyze_cohorts.vars.overall_trend }}, worst drop {{
          outputs.analyze_cohorts.vars.worst_drop_pct }} points"

triggers:
  - id: weekly_refresh
    type: io.kestra.plugin.core.trigger.Schedule
    description: Refresh the retention matrix every Monday morning, disabled until
      you are ready to schedule it.
    cron: "0 6 * * 1"
    disabled: true

errors:
  - id: slack_alert
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED {{ flow.id }} run {{ execution.id }}"}

outputs:
  - id: verdict
    type: STRING
    value: "{{ outputs.analyze_cohorts.vars.verdict }}"

  - id: flagged_cohorts
    type: JSON
    value: "{{ outputs.analyze_cohorts.vars.flagged_cohorts_json }}"

  - id: worst_drop_pct
    type: STRING
    value: "{{ outputs.analyze_cohorts.vars.worst_drop_pct }}"

  - id: report
    type: STRING
    value: "{{ outputs.analyze_cohorts.outputFiles['cohort_report.json'] }}"