data-pipeline

What this skill does

# data-pipeline

Orchestrate marketing data collection, transformation, and reporting workflows.

## Triggers


Alternate expressions and non-obvious activations (primary phrases are matched automatically from the skill description):

- "ETL [source] to [dest]" → data pipeline creation shorthand
- "ELT" → extract-load-transform pipeline
- "dbt" / "Airflow" / "Spark" → tool-specific pipeline requests

## Purpose

This skill manages marketing data workflows by:
- Collecting data from multiple marketing platforms
- Transforming raw data into actionable metrics
- Aggregating cross-channel performance
- Generating automated reports
- Maintaining data quality and consistency

## Behavior

When triggered, this skill:

1. **Identifies data sources**:
   - List connected platforms
   - Check API credentials/access
   - Determine data freshness requirements

2. **Collects raw data**:
   - Pull metrics from each platform
   - Handle pagination and rate limits
   - Store raw data snapshots

3. **Transforms data**:
   - Normalize naming conventions
   - Calculate derived metrics
   - Apply attribution models
   - Aggregate across channels

4. **Validates data**:
   - Check for anomalies
   - Validate against thresholds
   - Flag data quality issues

5. **Stores and reports**:
   - Update data warehouse/storage
   - Generate summary reports
   - Trigger alerts if needed

## Supported Platforms

### Advertising Platforms

```yaml
advertising:
  google_ads:
    metrics:
      - impressions
      - clicks
      - cost
      - conversions
      - conversion_value
    dimensions:
      - campaign
      - ad_group
      - keyword
      - device
    refresh_frequency: 4h

  meta_ads:
    metrics:
      - impressions
      - reach
      - clicks
      - spend
      - conversions
    dimensions:
      - campaign
      - ad_set
      - ad
      - placement
    refresh_frequency: 4h

  linkedin_ads:
    metrics:
      - impressions
      - clicks
      - cost
      - leads
      - conversions
    dimensions:
      - campaign
      - creative
      - audience
    refresh_frequency: daily
```

### Analytics Platforms

```yaml
analytics:
  google_analytics:
    metrics:
      - sessions
      - users
      - pageviews
      - bounce_rate
      - conversions
      - revenue
    dimensions:
      - source_medium
      - campaign
      - landing_page
      - device
    refresh_frequency: 4h

  mixpanel:
    metrics:
      - events
      - unique_users
      - retention
      - funnel_conversion
    dimensions:
      - event_name
      - user_properties
    refresh_frequency: real-time

  amplitude:
    metrics:
      - events
      - users
      - retention
      - conversion
    dimensions:
      - event_type
      - user_segment
    refresh_frequency: real-time
```

### Email Platforms

```yaml
email:
  mailchimp:
    metrics:
      - sends
      - opens
      - clicks
      - bounces
      - unsubscribes
    dimensions:
      - campaign
      - list
      - segment
    refresh_frequency: 1h

  hubspot:
    metrics:
      - sends
      - opens
      - clicks
      - contacts_created
      - deals_influenced
    dimensions:
      - campaign
      - email_type
      - lifecycle_stage
    refresh_frequency: 1h

  sendgrid:
    metrics:
      - delivered
      - opens
      - clicks
      - bounces
      - spam_reports
    refresh_frequency: real-time
```

### Social Platforms

```yaml
social:
  instagram:
    metrics:
      - reach
      - impressions
      - engagement
      - followers
      - saves
      - shares
    dimensions:
      - post_type
      - content_category
    refresh_frequency: daily

  linkedin:
    metrics:
      - impressions
      - engagement
      - followers
      - clicks
    dimensions:
      - post_type
      - content_category
    refresh_frequency: daily

  twitter:
    metrics:
      - impressions
      - engagements
      - followers
      - retweets
      - likes
    refresh_frequency: 4h
```

## Data Transformation

### Metric Calculations

```yaml
derived_metrics:
  ctr:
    formula: clicks / impressions
    format: percentage
    description: Click-through rate

  cpc:
    formula: cost / clicks
    format: currency
    description: Cost per click

  cpm:
    formula: (cost / impressions) * 1000
    format: currency
    description: Cost per thousand impressions

  cpa:
    formula: cost / conversions
    format: currency
    description: Cost per acquisition

  roas:
    formula: revenue / cost
    format: ratio
    description: Return on ad spend

  conversion_rate:
    formula: conversions / clicks
    format: percentage
    description: Conversion rate

  engagement_rate:
    formula: engagements / impressions
    format: percentage
    description: Engagement rate
```

### Attribution Models

```yaml
attribution_models:
  last_click:
    description: 100% credit to last touchpoint
    use_case: Bottom-funnel optimization

  first_click:
    description: 100% credit to first touchpoint
    use_case: Top-funnel optimization

  linear:
    description: Equal credit across touchpoints
    use_case: Multi-touch awareness

  time_decay:
    description: More credit to recent touchpoints
    use_case: Typical purchase journey

  position_based:
    description: 40% first, 40% last, 20% middle
    use_case: Balanced attribution

  data_driven:
    description: ML-based credit assignment
    use_case: Advanced optimization
```

## Pipeline Configuration

```yaml
pipeline_config:
  name: marketing-data-pipeline
  schedule: "0 */4 * * *"  # Every 4 hours

  sources:
    - name: google_ads
      credentials: .aiwg/marketing/config/google-ads-creds.json
      date_range: last_30_days

    - name: google_analytics
      credentials: .aiwg/marketing/config/ga4-creds.json
      property_id: "123456789"

    - name: meta_ads
      credentials: .aiwg/marketing/config/meta-creds.json
      ad_account_id: "act_123456"

  transformations:
    - name: normalize_naming
      rules:
        - source: google_ads
          campaign_pattern: "^GA_"
        - source: meta_ads
          campaign_pattern: "^META_"

    - name: calculate_metrics
      metrics: [ctr, cpc, cpa, roas]

    - name: apply_attribution
      model: position_based
      lookback_window: 30

  output:
    - type: json
      path: .aiwg/marketing/data/
    - type: csv
      path: .aiwg/marketing/reports/
    - type: dashboard
      tool: internal

  alerts:
    - name: spend_anomaly
      condition: daily_spend > avg_spend * 1.5
      notify: [marketing-team]

    - name: conversion_drop
      condition: daily_conversions < avg_conversions * 0.5
      notify: [marketing-team, analytics]
```

## Data Quality Checks

```yaml
quality_checks:
  completeness:
    - all_platforms_reporting: true
    - date_gaps: none_allowed
    - metric_nulls: <5%

  consistency:
    - cross_platform_totals: ±5% variance
    - historical_trend: ±20% from avg
    - attribution_sum: 100%

  freshness:
    - max_age: 24h
    - preferred_age: 4h
    - alert_threshold: 12h

  anomaly_detection:
    - z_score_threshold: 3
    - min_data_points: 14
    - metrics_to_monitor:
      - spend
      - conversions
      - ctr
      - cpc
```

## Pipeline Report Format

```markdown
# Marketing Data Pipeline Report

**Run ID**: PIPE-2025-12-08-1400
**Status**: Completed with Warnings
**Duration**: 4m 32s
**Date Range**: 2025-11-08 to 2025-12-08

## Data Collection Summary

| Source | Status | Records | Freshness |
|--------|--------|---------|-----------|
| Google Ads | ✅ Success | 45,231 | 2h ago |
| Meta Ads | ✅ Success | 32,156 | 3h ago |
| Google Analytics | ✅ Success | 128,459 | 1h ago |
| Mailchimp | ⚠️ Partial | 5,234 | 6h ago |
| Instagram | ✅ Success | 1,847 | 4h ago |

## Data Quality

| Check | Status | Details |
|-------|--------|---------|
| Completeness | ✅ Pass | All platforms reporting |
| Consistency | ⚠️ Warning | GA vs Ads conversion ±8% |
| Freshness | ✅ Pass | All data <12h old |
| Anomaly | ✅ Pass | No anomali