n8n + Claude Self-Healing Data Pipeline: PostgreSQL Anomaly Detection & Auto-Remediation

Scheduled PostgreSQL fetch → statistical/ML anomaly detection → Claude classification & remediation planning → auto-execution → validation & notification

Who

Data engineers, ops teams, organizations needing automated data quality monitoring

You get

Automatically detect data anomalies every 15 minutes, AI-classify and auto-remediate (fill nulls, cap outliers, reformat schema, retry upstream), with human escalation on failure

You need

n8n · postgresql · slack · pagerduty · datadog

Steps

1. Configure trigger and data connections

   After importing the template into n8n, configure the schedule trigger (default every 15 minutes) and the Set Pipeline Parameters node. Enter PostgreSQL connection details, source table name, Slack channel ID, PagerDuty service ID and API key, DataDog API key. Replace placeholders in nodes: <__PLACEHOLDER_VALUE__source_table_name__>, <__PLACEHOLDER_VALUE__Slack channel ID for notifications__>, <__PLACEHOLDER_VALUE__PagerDuty API key__>, <__PLACEHOLDER_VALUE__DataDog API key__>.

2. Run statistical and ML anomaly detection

   Workflow executes automatically: Fetch Latest Data from PostgreSQL pulls last 15 minutes of data; Fetch Historical Baseline Data loads 7-day baseline; Compare with Historical Baseline calculates deviations; Calculate Trend Analysis Metrics computes moving averages and trends; Run ML-Based Anomaly Scoring combines 5 features (baseline deviation, trend velocity, historical pattern, statistical outliers, null rate) into 0-1 anomaly score. Run Statistical Quality Checks executes null rate, IQR outlier, and schema consistency checks.

3. Claude classifies anomaly and plans remediation

   If anomaly score exceeds threshold (default 0.75) or statistical checks detect anomalies, Check If Anomaly Detected branches to AI Anomaly Classifier Agent. This node calls Claude-3-5-sonnet with a system prompt setting it as data quality expert, user message passing anomaly type, affected columns, and statistics summary, requesting JSON output with classification (schema_drift/missing_data/outliers/upstream_failure), severity (low/medium/high/critical), confidence, and explanation. Then AI Remediation Planner Agent calls Claude again to generate a single remediation action (fill_nulls/cap_outliers/reformat_schema/retry_upstream) with parameters, success probability, and reasoning.

   Execution prompt · Source text。AI Anomaly Classifier Agent node options.system field

   Copy prompt

   You are a data quality expert specializing in anomaly detection and classification. Analyze data quality issues and provide accurate classifications with severity assessments.

   Execution prompt · Source text。AI Anomaly Classifier Agent node messages.values[0].content field

   Copy prompt

   =Analyze this data quality anomaly and classify it:
   
   Anomaly Type: {{ $('Run Statistical Quality Checks').first().json.anomalyType }}
   Affected Columns: {{ $('Run Statistical Quality Checks').first().json.affectedColumns }}
   Statistics: {{ JSON.stringify($('Run Statistical Quality Checks').first().json.statistics) }}
   Total Records: {{ $('Run Statistical Quality Checks').first().json.totalRecords }}
   
   Classify this anomaly into one of these categories:
   - schema_drift: Unexpected changes in data structure or types
   - missing_data: High rates of null or missing values
   - outliers: Statistical anomalies in numeric data
   - upstream_failure: Issues with data source or pipeline
   
   Provide your response as JSON with:
   {
     "classification": "category_name",
     "severity": "low|medium|high|critical",
     "confidence": 0-100,
     "explanation": "detailed reasoning"
   }

   Execution prompt · Source text。AI Remediation Planner Agent node options.system field

   Copy prompt

   You are a data engineering expert specializing in automated remediation strategies. Suggest practical, safe remediation actions for data quality issues based on anomaly classification and historical patterns.

   Execution prompt · Source text。AI Remediation Planner Agent node messages.values[0].content field

   Copy prompt

   =Based on this classified anomaly, suggest a specific remediation action:
   
   Classification: {{ $json.classification }}
   Severity: {{ $json.severity }}
   Anomaly Details: {{ $json.explanation }}
   Affected Columns: {{ $('Run Statistical Quality Checks').first().json.affectedColumns }}
   Statistics: {{ JSON.stringify($('Run Statistical Quality Checks').first().json.statistics) }}
   
   Suggest ONE of these remediation actions:
   - fill_nulls: Fill missing values with appropriate defaults
   - cap_outliers: Cap extreme values to acceptable ranges
   - reformat_schema: Fix schema mismatches or type issues
   - retry_upstream: Retry fetching data from source
   
   Provide:
   1. Action (one of the above)
   2. Parameters needed for the action
   3. Expected success probability (0-100)
   4. Reasoning
   
   Return as JSON: {"action": "...", "parameters": {...}, "successProbability": ..., "reasoning": "..."}

4. Execute remediation action

   Query Remediation History from PostgreSQL queries last 30 days of similar anomaly remediation history; Calculate Remediation Success Rate computes historical success rates by action and attaches to remediation plan. Route by Remediation Type routes based on action field to one of four branches: Remediate Missing Values (median for numeric, mode for categorical), Remediate Outliers (IQR bound capping), Remediate Schema Format (convert to expected schema types), Retry Upstream Data Fetch (placeholder, actual retry logic must be implemented). When severity is critical, Check If Critical Severity in parallel triggers Post Critical Alert to PagerDuty to create PagerDuty incident.

5. Validate remediation and notify

   Merge Remediated Data Streams combines branch results; Re-run Quality Checks on Remediated Data re-runs null, outlier, and schema checks. If validation passes (remediationSuccessful=true), Send Success Notification to Slack sends success notification; otherwise Escalate to Human Review in Slack sends human review request with data samples and statistics, triggering Prepare Data Sample for Review to generate representative samples.

6. Log incident and update monitoring

   Log Successful/Failed Remediation Incident writes results to PostgreSQL incident table; Aggregate Incident Logs aggregates; Generate Remediation Report produces comprehensive report with success rates, breakdowns by type/severity/action, top issues, recommendations; Send Metrics to DataDog pushes anomaly count, remediation success rate, health score; Update Historical Baseline Metrics updates baseline table.

Please set up this n8n template for a self-healing data pipeline. First confirm you have: n8n instance, PostgreSQL (source table + incident table + baseline table), Anthropic Claude API key, Slack channel permissions, PagerDuty service ID and API key, DataDog API key. After importing the template, replace all placeholders step by step, test individual node execution, then enable the schedule trigger.