Code Examples - Causal Inference Based Attribution

API Examples

Copy-paste ready code examples for calling the Causal Attribution API in your preferred programming language.

cURL

Basic command-line example for testing the API:

curl -X POST https://api.causalmma.com/api/v1/attribution \
  -H "X-API-Key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "touchpoints": [
      {
        "timestamp": "2025-01-01T10:00:00Z",
        "channel": "facebook_ad",
        "cost": 2.50,
        "user_id": "user123",
        "campaign": "summer_sale"
      },
      {
        "timestamp": "2025-01-02T15:30:00Z",
        "channel": "google_search",
        "cost": 3.75,
        "user_id": "user123",
        "campaign": "brand_awareness"
      },
      {
        "timestamp": "2025-01-03T18:45:00Z",
        "channel": "email",
        "cost": 0.10,
        "user_id": "user123",
        "campaign": "retargeting"
      }
    ],
    "conversion": {
      "timestamp": "2025-01-03T19:00:00Z",
      "value": 150.00,
      "user_id": "user123",
      "type": "purchase"
    },
    "model": "data_driven",
    "options": {
      "return_causal_graph": true,
      "confidence_interval": 0.95
    }
  }'

Python

Using the popular requests library:

import requests
import json

class CausalAttributionAPI:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.causalmma.com/api/v1"
        self.headers = {
            "X-API-Key": f"{api_key}",
            "Content-Type": "application/json"
        }

    def analyze_attribution(self, touchpoints, conversion, model="data_driven", options=None):
        """
        Analyze attribution for a customer journey.

        Args:
            touchpoints: List of touchpoint dictionaries
            conversion: Conversion dictionary
            model: Attribution model to use (default: data_driven)
            options: Additional options dictionary

        Returns:
            Attribution results dictionary
        """
        endpoint = f"{self.base_url}/attribution"

        payload = {
            "touchpoints": touchpoints,
            "conversion": conversion,
            "model": model
        }

        if options:
            payload["options"] = options

        response = requests.post(endpoint, headers=self.headers, json=payload)
        response.raise_for_status()

        return response.json()

    def batch_analyze(self, journeys):
        """
        Analyze multiple customer journeys in a single request.

        Args:
            journeys: List of journey dictionaries

        Returns:
            Batch attribution results
        """
        endpoint = f"{self.base_url}/batch-attribution"

        payload = {"journeys": journeys}

        response = requests.post(endpoint, headers=self.headers, json=payload)
        response.raise_for_status()

        return response.json()

# Example usage
if __name__ == "__main__":
    api = CausalAttributionAPI("YOUR_API_KEY")

    touchpoints = [
        {
            "timestamp": "2025-01-01T10:00:00Z",
            "channel": "facebook_ad",
            "cost": 2.50,
            "user_id": "user123"
        },
        {
            "timestamp": "2025-01-02T15:30:00Z",
            "channel": "google_search",
            "cost": 3.75,
            "user_id": "user123"
        }
    ]

    conversion = {
        "timestamp": "2025-01-03T19:00:00Z",
        "value": 150.00,
        "user_id": "user123"
    }

    results = api.analyze_attribution(
        touchpoints=touchpoints,
        conversion=conversion,
        model="data_driven",
        options={"return_causal_graph": True}
    )

    print(json.dumps(results, indent=2))

JavaScript (Node.js)

Using native fetch API:

const fetch = require('node-fetch');

class CausalAttributionAPI {
  constructor(apiKey) {
    this.apiKey = apiKey;
    this.baseURL = "https://api.causalmma.com/api/v1";
    this.headers = {
      'X-API-Key': `${apiKey}`,
      'Content-Type': 'application/json'
    };
  }

  async analyzeAttribution(touchpoints, conversion, model = 'data_driven', options = {}) {
    const endpoint = `${this.baseURL}/attribute`;

    const payload = {
      touchpoints,
      conversion,
      model,
      ...options
    };

    try {
      const response = await fetch(endpoint, {
        method: 'POST',
        headers: this.headers,
        body: JSON.stringify(payload)
      });

      if (!response.ok) {
        throw new Error(`API error: ${response.status} ${response.statusText}`);
      }

      return await response.json();
    } catch (error) {
      console.error('Attribution API Error:', error);
      throw error;
    }
  }

  async batchAnalyze(journeys) {
    const endpoint = `${this.baseURL}/batch-attribute`;

    try {
      const response = await fetch(endpoint, {
        method: 'POST',
        headers: this.headers,
        body: JSON.stringify({ journeys })
      });

      if (!response.ok) {
        throw new Error(`API error: ${response.status}`);
      }

      return await response.json();
    } catch (error) {
      console.error('Batch Attribution Error:', error);
      throw error;
    }
  }
}

// Example usage
(async () => {
  const api = new CausalAttributionAPI('YOUR_API_KEY');

  const touchpoints = [
    {
      timestamp: '2025-01-01T10:00:00Z',
      channel: 'facebook_ad',
      cost: 2.50,
      user_id: 'user123'
    },
    {
      timestamp: '2025-01-02T15:30:00Z',
      channel: 'google_search',
      cost: 3.75,
      user_id: 'user123'
    }
  ];

  const conversion = {
    timestamp: '2025-01-03T19:00:00Z',
    value: 150.00,
    user_id: 'user123'
  };

  const results = await api.analyzeAttribution(touchpoints, conversion);
  console.log(JSON.stringify(results, null, 2));
})();

PHP

Using cURL extension:

<?php

class CausalAttributionAPI {
    private $apiKey;
    private $baseURL = "https://api.causalmma.com/api/v1";

    public function __construct($apiKey) {
        $this->apiKey = $apiKey;
    }

    public function analyzeAttribution($touchpoints, $conversion, $model = 'data_driven', $options = []) {
        $endpoint = $this->baseURL . '/attribute';

        $payload = [
            'touchpoints' => $touchpoints,
            'conversion' => $conversion,
            'model' => $model
        ];

        if (!empty($options)) {
            $payload['options'] = $options;
        }

        $ch = curl_init($endpoint);

        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
        curl_setopt($ch, CURLOPT_POST, true);
        curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($payload));
        curl_setopt($ch, CURLOPT_HTTPHEADER, [
            'X-API-Key: ' . $this->apiKey,
            'Content-Type: application/json'
        ]);

        $response = curl_exec($ch);
        $httpCode = curl_getinfo($ch, CURLINFO_HTTP_CODE);

        curl_close($ch);

        if ($httpCode !== 200) {
            throw new Exception("API error: HTTP $httpCode");
        }

        return json_decode($response, true);
    }
}

// Example usage
$api = new CausalAttributionAPI('YOUR_API_KEY');

$touchpoints = [
    [
        'timestamp' => '2025-01-01T10:00:00Z',
        'channel' => 'facebook_ad',
        'cost' => 2.50,
        'user_id' => 'user123'
    ],
    [
        'timestamp' => '2025-01-02T15:30:00Z',
        'channel' => 'google_search',
        'cost' => 3.75,
        'user_id' => 'user123'
    ]
];

$conversion = [
    'timestamp' => '2025-01-03T19:00:00Z',
    'value' => 150.00,
    'user_id' => 'user123'
];

$results = $api->analyzeAttribution($touchpoints, $conversion);
echo json_encode($results, JSON_PRETTY_PRINT);

?>

Ruby

Using the net/http standard library:

require 'net/http'
require 'json'
require 'uri'

class CausalAttributionAPI
  def initialize(api_key)
    @api_key = api_key
    @base_url = "https://api.causalmma.com/api/v1"
  end

  def analyze_attribution(touchpoints:, conversion:, model: 'data_driven', options: {})
    uri = URI("#{@base_url}/attribution")

    payload = {
      touchpoints: touchpoints,
      conversion: conversion,
      model: model
    }

    payload[:options] = options unless options.empty?

    http = Net::HTTP.new(uri.host, uri.port)
    http.use_ssl = true

    request = Net::HTTP::Post.new(uri.path)
    request["X-API-Key"] = "#{@api_key}"
    request['Content-Type'] = 'application/json'
    request.body = payload.to_json

    response = http.request(request)

    raise "API error: #{response.code}" unless response.code == '200'

    JSON.parse(response.body)
  end
end

# Example usage
api = CausalAttributionAPI.new('YOUR_API_KEY')

touchpoints = [
  {
    timestamp: '2025-01-01T10:00:00Z',
    channel: 'facebook_ad',
    cost: 2.50,
    user_id: 'user123'
  },
  {
    timestamp: '2025-01-02T15:30:00Z',
    channel: 'google_search',
    cost: 3.75,
    user_id: 'user123'
  }
]

conversion = {
  timestamp: '2025-01-03T19:00:00Z',
  value: 150.00,
  user_id: 'user123'
}

results = api.analyze_attribution(
  touchpoints: touchpoints,
  conversion: conversion,
  model: 'data_driven'
)

puts JSON.pretty_generate(results)

R

Using the httr package:

library(httr)
library(jsonlite)

CausalAttributionAPI <- R6::R6Class(
  "CausalAttributionAPI",

  private = list(
    api_key = NULL,
    base_url = "https://api.causalmma.com/api/v1"
  ),

  public = list(
    initialize = function(api_key) {
      private$api_key <- api_key
    },

    analyze_attribution = function(touchpoints, conversion, model = "data_driven", options = list()) {
      endpoint <- paste0(private$base_url, "/attribution")

      payload <- list(
        touchpoints = touchpoints,
        conversion = conversion,
        model = model
      )

      if (length(options) > 0) {
        payload$options <- options
      }

      response <- POST(
        endpoint,
        add_headers(
          "X-API-Key" = private$api_key),
          "Content-Type" = "application/json"
        ),
        body = toJSON(payload, auto_unbox = TRUE),
        encode = "json"
      )

      if (status_code(response) != 200) {
        stop(paste("API error:", status_code(response)))
      }

      content(response, as = "parsed")
    }
  )
)

# Example usage
api <- CausalAttributionAPI$new("YOUR_API_KEY")

touchpoints <- list(
  list(
    timestamp = "2025-01-01T10:00:00Z",
    channel = "facebook_ad",
    cost = 2.50,
    user_id = "user123"
  ),
  list(
    timestamp = "2025-01-02T15:30:00Z",
    channel = "google_search",
    cost = 3.75,
    user_id = "user123"
  )
)

conversion <- list(
  timestamp = "2025-01-03T19:00:00Z",
  value = 150.00,
  user_id = "user123"
)

results <- api$analyze_attribution(touchpoints, conversion)
print(results)

Common Use Cases

Multi-Channel Campaign Analysis

Analyze attribution across multiple marketing channels:

# Python example
touchpoints = [
    {"timestamp": "2025-01-01T09:00:00Z", "channel": "facebook_ad", "cost": 5.00, "user_id": "user456"},
    {"timestamp": "2025-01-01T14:00:00Z", "channel": "instagram_ad", "cost": 3.50, "user_id": "user456"},
    {"timestamp": "2025-01-02T10:30:00Z", "channel": "google_search", "cost": 4.25, "user_id": "user456"},
    {"timestamp": "2025-01-02T16:00:00Z", "channel": "email", "cost": 0.15, "user_id": "user456"},
    {"timestamp": "2025-01-03T11:00:00Z", "channel": "retargeting", "cost": 2.75, "user_id": "user456"}
]

conversion = {
    "timestamp": "2025-01-03T12:00:00Z",
    "value": 299.00,
    "user_id": "user456",
    "type": "purchase"
}

results = api.analyze_attribution(touchpoints, conversion, model="data_driven")

Batch Processing

Process multiple customer journeys efficiently:

# Python example
journeys = [
    {
        "touchpoints": [...],  # Journey 1
        "conversion": {...},
        "journey_id": "journey_001"
    },
    {
        "touchpoints": [...],  # Journey 2
        "conversion": {...},
        "journey_id": "journey_002"
    }
]

batch_results = api.batch_analyze(journeys)

Ready to Start Building?

Get your API key and start integrating causal attribution today.

Get API Key View Full Docs