Skip to main content

Supply Chain Emissions Tutorial

Learn how to track emissions from your supply chain, including purchased goods, services, and supplier-specific data.
Estimated time: 30 minutesWhat you’ll learn:
  • Track purchased goods and services
  • Use spend-based vs activity-based factors
  • Manage supplier data
  • Apply custom emission factors
  • Calculate Scope 3 Category 1 emissions

Prerequisites

Before starting, ensure you have:
  • Dcycle API credentials (get them here)
  • Basic knowledge of Python or JavaScript
  • Purchase data: invoices, receipts, or procurement records
Using the Dcycle App?You can also track purchases through our web interface:

Understanding Supply Chain Emissions

Supply chain emissions are Scope 3 Category 1 (Purchased Goods and Services) - typically the largest part of an organization’s carbon footprint.

Data Quality Hierarchy

Level 4: Supplier-Specific Data (Highest Accuracy)
β”œβ”€ Supplier EPDs, Product Carbon Footprints
β”œβ”€ Accuracy: Β±5-15%
└─ Use custom emission factors

Level 3: Activity-Based Data
β”œβ”€ Product category + physical quantity
β”œβ”€ Accuracy: Β±20-40%
└─ Use standard emission factors

Level 2: Spend-Based Data (Lowest Accuracy)
β”œβ”€ Only monetary value + category
β”œβ”€ Accuracy: Β±50-100%
└─ Use economic intensity factors

Level 1: No Data
β”œβ”€ Estimate or extrapolate
└─ Accuracy: Very low

Calculation Methods

When to use: You only know how much you spentFormula: Spend Γ— Economic Intensity FactorExample: €10,000 on IT services Γ— 0.15 kg CO2e/€ = 1,500 kg CO2ePros: Easy, covers all purchases Cons: Low accuracy, doesn’t reflect actual products
When to use: You know physical quantitiesFormula: Quantity Γ— Emission FactorExample: 1,000 kg aluminum Γ— 8.5 kg CO2e/kg = 8,500 kg CO2ePros: More accurate, reflects actual products Cons: Requires detailed data
When to use: Supplier provides verified dataFormula: Quantity Γ— Supplier-Specific FactorExample: 1,000 kg recycled aluminum Γ— 2.15 kg CO2e/kg = 2,150 kg CO2ePros: Highest accuracy, enables supplier engagement Cons: Requires supplier collaboration

Step 1: Create Suppliers

Organize purchases by supplier:
import requests
import os

headers = {
    "Authorization": f"Bearer {os.getenv('DCYCLE_API_KEY')}",
    "x-organization-id": os.getenv("DCYCLE_ORG_ID"),
    "x-user-id": os.getenv("DCYCLE_USER_ID")
}

# Create supplier
supplier_data = {
    "name": "Green Materials Co",
    "contact_email": "[email protected]",
    "country": "ES",
    "sector": "Manufacturing",  # Optional
    "supplier_code": "SUP-001"  # Your internal code
}

supplier = requests.post(
    "https://api.dcycle.io/api/v1/suppliers",
    headers=headers,
    json=supplier_data
).json()

supplier_id = supplier['id']
print(f"βœ… Supplier created: {supplier_id}")
print(f"   Name: {supplier['name']}")

List All Suppliers

suppliers = requests.get(
    "https://api.dcycle.io/api/v1/suppliers",
    headers=headers,
    params={"page": 1, "size": 50}
).json()

print("πŸ“‹ Your Suppliers:")
for s in suppliers['items']:
    print(f"   - {s['name']} ({s['country']})")
    if s.get('supplier_code'):
        print(f"     Code: {s['supplier_code']}")

Step 2: Track Purchases (Activity-Based)

When you know physical quantities:
# Get units
units = requests.get(
    "https://api.dcycle.io/api/v1/units",
    headers=headers,
    params={"page": 1, "size": 100}
).json()

kg_unit = next(u for u in units['items'] if u['abbreviation'] == 'kg')

# Track purchase with physical quantity
purchase_data = {
    "name": "Aluminum sheets",
    "category": "metals",  # Product category
    "quantity": 1000,
    "unit_id": kg_unit['id'],
    "purchase_date": "2024-03-15",
    "supplier_id": supplier_id,  # Optional but recommended
    "invoice_number": "INV-2024-001",
    "cost": 8500,  # Optional
    "currency": "EUR"  # Optional
}

purchase = requests.post(
    "https://api.dcycle.io/api/v1/purchases",
    headers=headers,
    json=purchase_data
).json()

print(f"βœ… Purchase tracked")
print(f"   Product: {purchase['name']}")
print(f"   Quantity: {purchase['quantity']} kg")
print(f"   CO2e: {purchase['co2e']:.2f} kg")
print(f"   Emission factor: {purchase['emission_factor_used']}")
print(f"   Method: Activity-based")

Available Product Categories

Get all product categories:
# Categories are organized hierarchically
# Use the most specific category available

categories = [
    "metals",
    "plastics",
    "paper",
    "textiles",
    "chemicals",
    "electronics",
    "wood",
    "food",
    "energy",
    "services_it",
    "services_professional",
    "services_transport",
    "construction_materials",
    # ... many more
]

# For specific categories, see API reference

Step 3: Track Purchases (Spend-Based)

When you only know monetary value:
# Spend-based purchase (less accurate)
spend_purchase = {
    "name": "IT consulting services",
    "category": "services_it",
    "spend_amount": 15000,  # Monetary value
    "spend_currency": "EUR",
    "purchase_date": "2024-03-15",
    "supplier_id": supplier_id,
    "invoice_number": "INV-2024-002"
}

purchase = requests.post(
    "https://api.dcycle.io/api/v1/purchases",
    headers=headers,
    json=spend_purchase
).json()

print(f"βœ… Purchase tracked (spend-based)")
print(f"   Service: {purchase['name']}")
print(f"   Spend: {purchase['spend_amount']} {purchase['spend_currency']}")
print(f"   CO2e: {purchase['co2e']:.2f} kg")
print(f"   Method: Spend-based (lower accuracy)")
Spend-based factors have high uncertainty (Β±50-100%)Try to collect physical quantities whenever possible. Spend-based should be a last resort for:
  • Services without clear units
  • Mixed purchases with unknown breakdown
  • Historical data where only invoices exist

Step 4: Use Supplier-Specific Factors

For highest accuracy, use supplier-provided emission data:

Create Custom Emission Factor

# 1. Create emission group for supplier
group = requests.post(
    "https://api.dcycle.io/api/v1/custom_emission_groups",
    headers=headers,
    json={
        "name": "Green Materials Co - 2024 Product Line",
        "description": "EPD-verified factors from Green Materials Co. All products third-party certified.",
        "category": "purchases",
        "ghg_type": 1  # Fossil
    }
).json()

# 2. Add supplier's recycled aluminum factor
factor = requests.post(
    f"https://api.dcycle.io/api/v1/custom_emission_factors/{group['id']}",
    headers=headers,
    json={
        "ef_name": "Recycled Aluminum Sheet - Green Materials Co",
        "unit_id": kg_unit['id'],
        "factor_uploaded_by": "[email protected]",
        "tag": "advanced",
        "uncertainty_grade": 12,  # Low uncertainty (EPD-verified)
        "factor_start_date": "2024-01-01",
        "factor_end_date": "2024-12-31",
        "additional_docs": "EPD No. GMC-AL-2024, Bureau Veritas verified",
        "emission_factor_values": [
            {"gas_type": "CO2", "value": 2.15},  # Much lower than generic 8.5
            {"gas_type": "CH4", "value": 0.008},
            {"gas_type": "N2O", "value": 0.002}
        ],
        "recycled": True
    }
).json()

factor_id = factor['id']
print(f"βœ… Custom factor created: {factor_id}")

Use in Purchase

# Track purchase using supplier-specific factor
purchase_with_custom_factor = {
    "name": "Recycled aluminum sheets from Green Materials Co",
    "custom_emission_factor_id": factor_id,  # Use supplier's factor
    "quantity": 1000,
    "unit_id": kg_unit['id'],
    "purchase_date": "2024-03-15",
    "supplier_id": supplier_id,
    "invoice_number": "INV-2024-003"
}

purchase = requests.post(
    "https://api.dcycle.io/api/v1/purchases",
    headers=headers,
    json=purchase_with_custom_factor
).json()

print(f"βœ… Purchase tracked with supplier factor")
print(f"   Quantity: {purchase['quantity']} kg")
print(f"   CO2e: {purchase['co2e']:.2f} kg")  # 2,150 kg vs 8,500 kg with generic
print(f"   Savings: {8500 - purchase['co2e']:.2f} kg CO2e (vs generic)")
Learn more about custom emission factors β†’

Step 5: Bulk Upload Purchases

For large datasets from ERP systems:

CSV Format

name,category,quantity,unit_id,purchase_date,supplier_id,invoice_number,cost,currency
Aluminum sheets,metals,1000,kg-uuid,2024-03-01,supplier-uuid-1,INV-001,8500,EUR
Steel beams,metals,5000,kg-uuid,2024-03-05,supplier-uuid-2,INV-002,15000,EUR
Plastic granules,plastics,500,kg-uuid,2024-03-10,supplier-uuid-1,INV-003,2500,EUR
IT consulting,services_it,,,2024-03-15,supplier-uuid-3,INV-004,12000,EUR
Note: Leave quantity and unit_id empty for spend-based (will use cost and currency).

Upload Process

# Get presigned URL
upload_response = requests.post(
    "https://api.dcycle.io/api/v1/purchases/bulk/csv",
    headers=headers
).json()

# Upload CSV
with open('march_2024_purchases.csv', 'rb') as f:
    requests.put(upload_response['upload_url'], data=f)

print("βœ… CSV uploaded, processing...")

# Poll for completion
import time

while True:
    status = requests.get(
        upload_response['status_url'],
        headers=headers
    ).json()

    if status['status'] == 'completed':
        print(f"βœ… Processed {status['records_processed']} purchases")
        print(f"   Total CO2e: {status['total_co2e']:,.2f} kg")
        print(f"   Activity-based: {status['activity_based_count']}")
        print(f"   Spend-based: {status['spend_based_count']}")
        break
    elif status['status'] == 'failed':
        print(f"❌ Error: {status['error']}")
        break

    time.sleep(5)

Step 6: Query and Analyze

Get Supplier Report

supplier_id = "supplier-uuid"

report = requests.get(
    f"https://api.dcycle.io/api/v1/suppliers/{supplier_id}/emissions",
    headers=headers,
    params={
        "start_date": "2024-01-01",
        "end_date": "2024-12-31"
    }
).json()

print(f"πŸ“Š Supplier Report: {report['supplier_name']}")
print(f"   Total purchases: {report['total_purchases']}")
print(f"   Total spend: {report['total_spend']:,.2f} {report['currency']}")
print(f"   Total CO2e: {report['total_co2e']:,.2f} kg")
print(f"   Average per purchase: {report['average_co2e_per_purchase']:.2f} kg")
print(f"\n   By product category:")

for category in report['by_category']:
    print(f"      - {category['category']}: {category['co2e']:,.2f} kg CO2e")

Compare Suppliers

# Get all suppliers
suppliers = requests.get(
    "https://api.dcycle.io/api/v1/suppliers",
    headers=headers
).json()

comparison = []

for supplier in suppliers['items']:
    report = requests.get(
        f"https://api.dcycle.io/api/v1/suppliers/{supplier['id']}/emissions",
        headers=headers,
        params={"start_date": "2024-01-01", "end_date": "2024-12-31"}
    ).json()

    if report['total_purchases'] > 0:
        comparison.append({
            'name': supplier['name'],
            'country': supplier['country'],
            'total_co2e': report['total_co2e'],
            'total_spend': report['total_spend'],
            'intensity': report['total_co2e'] / report['total_spend']  # kg CO2e per €
        })

# Sort by total emissions
comparison.sort(key=lambda x: x['total_co2e'], reverse=True)

print("\nπŸ“Š Supplier Comparison (2024)")
print(f"{'Supplier':<30} {'Country':>10} {'Total CO2e':>15} {'Spend (€)':>15} {'Intensity':>20}")
print("-" * 95)

for s in comparison:
    print(f"{s['name']:<30} {s['country']:>10} {s['total_co2e']:>15,.1f} {s['total_spend']:>15,.2f} {s['intensity']:>20,.3f} kg/€")

Category Analysis

# Get purchases by category
all_purchases = requests.get(
    "https://api.dcycle.io/api/v1/purchases",
    headers=headers,
    params={"page": 1, "size": 1000, "start_date": "2024-01-01", "end_date": "2024-12-31"}
).json()

# Aggregate by category
from collections import defaultdict

by_category = defaultdict(lambda: {'co2e': 0, 'spend': 0, 'count': 0})

for purchase in all_purchases['items']:
    cat = purchase['category']
    by_category[cat]['co2e'] += purchase['co2e']
    by_category[cat]['spend'] += purchase.get('cost', 0)
    by_category[cat]['count'] += 1

# Sort by emissions
categories_sorted = sorted(
    by_category.items(),
    key=lambda x: x[1]['co2e'],
    reverse=True
)

print("\nπŸ“Š Purchases by Category")
for category, data in categories_sorted[:10]:  # Top 10
    print(f"\n{category}:")
    print(f"   Purchases: {data['count']}")
    print(f"   Total CO2e: {data['co2e']:,.2f} kg")
    print(f"   Total spend: €{data['spend']:,.2f}")
    if data['spend'] > 0:
        print(f"   Intensity: {data['co2e']/data['spend']:.3f} kg CO2e/€")

Real-World Example: Supply Chain Management

Complete workflow for managing supply chain emissions:
import requests
import os
from datetime import date
from collections import defaultdict
import csv

class SupplyChainManager:
    def __init__(self):
        self.headers = {
            "Authorization": f"Bearer {os.getenv('DCYCLE_API_KEY')}",
            "x-organization-id": os.getenv("DCYCLE_ORG_ID"),
            "x-user-id": os.getenv("DCYCLE_USER_ID")
        }
        self.base_url = "https://api.dcycle.io"

    def create_supplier(self, **kwargs):
        """Create new supplier"""
        response = requests.post(
            f"{self.base_url}/api/v1/suppliers",
            headers=self.headers,
            json=kwargs
        )
        return response.json()

    def track_purchase(self, **kwargs):
        """Track single purchase"""
        response = requests.post(
            f"{self.base_url}/api/v1/purchases",
            headers=self.headers,
            json=kwargs
        )
        return response.json()

    def bulk_upload_purchases(self, csv_file_path):
        """Bulk upload from ERP export"""
        upload_response = requests.post(
            f"{self.base_url}/api/v1/purchases/bulk/csv",
            headers=self.headers
        ).json()

        with open(csv_file_path, 'rb') as f:
            requests.put(upload_response['upload_url'], data=f)

        return upload_response['status_url']

    def wait_for_upload(self, status_url):
        """Wait for bulk upload completion"""
        import time

        while True:
            status = requests.get(status_url, headers=self.headers).json()

            if status['status'] == 'completed':
                return status
            elif status['status'] == 'failed':
                raise Exception(f"Upload failed: {status.get('error')}")

            time.sleep(5)

    def get_supplier_report(self, supplier_id, start_date, end_date):
        """Get supplier emissions report"""
        response = requests.get(
            f"{self.base_url}/api/v1/suppliers/{supplier_id}/emissions",
            headers=self.headers,
            params={
                "start_date": start_date.isoformat(),
                "end_date": end_date.isoformat()
            }
        )
        return response.json()

    def identify_hotspots(self, start_date, end_date):
        """Identify emission hotspots in supply chain"""
        # Get all purchases
        purchases = requests.get(
            f"{self.base_url}/api/v1/purchases",
            headers=self.headers,
            params={
                "page": 1,
                "size": 1000,
                "start_date": start_date.isoformat(),
                "end_date": end_date.isoformat()
            }
        ).json()

        # Aggregate by supplier and category
        by_supplier = defaultdict(lambda: {'co2e': 0, 'count': 0, 'name': ''})
        by_category = defaultdict(lambda: {'co2e': 0, 'count': 0})

        for p in purchases['items']:
            # By supplier
            if p.get('supplier_id'):
                by_supplier[p['supplier_id']]['co2e'] += p['co2e']
                by_supplier[p['supplier_id']]['count'] += 1
                by_supplier[p['supplier_id']]['name'] = p.get('supplier_name', 'Unknown')

            # By category
            by_category[p['category']]['co2e'] += p['co2e']
            by_category[p['category']]['count'] += 1

        return {
            'by_supplier': dict(by_supplier),
            'by_category': dict(by_category),
            'total_co2e': sum(p['co2e'] for p in purchases['items']),
            'total_purchases': len(purchases['items'])
        }

    def generate_supplier_engagement_report(self, year):
        """Generate report for supplier engagement"""
        hotspots = self.identify_hotspots(
            date(year, 1, 1),
            date(year, 12, 31)
        )

        # Sort suppliers by emissions
        top_suppliers = sorted(
            hotspots['by_supplier'].items(),
            key=lambda x: x[1]['co2e'],
            reverse=True
        )[:20]  # Top 20 suppliers

        # Calculate cumulative percentage
        total_co2e = hotspots['total_co2e']
        cumulative = 0

        report = []
        for supplier_id, data in top_suppliers:
            cumulative += data['co2e']
            cumulative_pct = (cumulative / total_co2e) * 100

            report.append({
                'supplier_id': supplier_id,
                'supplier_name': data['name'],
                'purchases': data['count'],
                'co2e': data['co2e'],
                'percentage': (data['co2e'] / total_co2e) * 100,
                'cumulative_percentage': cumulative_pct,
                'priority': 'HIGH' if cumulative_pct <= 80 else 'MEDIUM'  # Focus on 80%
            })

        return report

# Usage Example
manager = SupplyChainManager()

# 1. Create suppliers
print("Creating suppliers...")
suppliers_data = [
    {"name": "Green Materials Co", "country": "ES", "supplier_code": "GRN-001"},
    {"name": "Tech Components Ltd", "country": "FR", "supplier_code": "TCH-001"},
    {"name": "Logistics Services SA", "country": "IT", "supplier_code": "LOG-001"}
]

supplier_ids = {}
for data in suppliers_data:
    supplier = manager.create_supplier(**data)
    supplier_ids[data['supplier_code']] = supplier['id']
    print(f"βœ… Created: {data['name']}")

# 2. Track purchases from ERP
print("\nUploading Q1 purchases...")
status_url = manager.bulk_upload_purchases("q1_2024_purchases.csv")
result = manager.wait_for_upload(status_url)
print(f"βœ… Processed {result['records_processed']} purchases")
print(f"   Total CO2e: {result['total_co2e']:,.2f} kg")

# 3. Identify hotspots
print("\nAnalyzing emission hotspots...")
hotspots = manager.identify_hotspots(date(2024, 1, 1), date(2024, 3, 31))

print(f"\nπŸ“Š Supply Chain Hotspots (Q1 2024)")
print(f"   Total emissions: {hotspots['total_co2e']:,.2f} kg CO2e")
print(f"   Total purchases: {hotspots['total_purchases']}")

print(f"\n   Top categories by emissions:")
top_categories = sorted(
    hotspots['by_category'].items(),
    key=lambda x: x[1]['co2e'],
    reverse=True
)[:5]

for category, data in top_categories:
    pct = (data['co2e'] / hotspots['total_co2e']) * 100
    print(f"      - {category}: {data['co2e']:,.1f} kg CO2e ({pct:.1f}%)")

# 4. Generate supplier engagement report
print("\nGenerating supplier engagement priorities...")
engagement_report = manager.generate_supplier_engagement_report(2024)

print(f"\nπŸ“‹ Supplier Engagement Priorities")
print(f"{'Supplier':<30} {'Priority':>10} {'CO2e (kg)':>15} {'% of Total':>12} {'Cumulative %':>15}")
print("-" * 90)

for s in engagement_report[:10]:  # Top 10
    print(f"{s['supplier_name']:<30} {s['priority']:>10} {s['co2e']:>15,.1f} {s['percentage']:>12,.1f}% {s['cumulative_percentage']:>15,.1f}%")

# Save full report to CSV
with open('supplier_engagement_2024.csv', 'w', newline='') as f:
    writer = csv.DictWriter(f, fieldnames=engagement_report[0].keys())
    writer.writeheader()
    writer.writerows(engagement_report)

print("\nβœ… Full report saved to supplier_engagement_2024.csv")

Best Practices

1. Prioritize Data Quality

Start with highest-impact categories:
def prioritize_data_collection(hotspots):
    """Identify where to focus data quality improvement"""

    # Focus on categories that are:
    # 1. High emissions (>10% of total)
    # 2. Currently using spend-based (low accuracy)

    priorities = []

    for category, data in hotspots['by_category'].items():
        pct_of_total = (data['co2e'] / hotspots['total_co2e']) * 100

        if pct_of_total > 10:
            priorities.append({
                'category': category,
                'co2e': data['co2e'],
                'percentage': pct_of_total,
                'action': 'Get physical quantities from suppliers'
            })

    return sorted(priorities, key=lambda x: x['co2e'], reverse=True)

2. Engage Suppliers Systematically

Use 80/20 rule:
def identify_key_suppliers(hotspots):
    """Find suppliers representing 80% of emissions"""

    sorted_suppliers = sorted(
        hotspots['by_supplier'].items(),
        key=lambda x: x[1]['co2e'],
        reverse=True
    )

    total_co2e = hotspots['total_co2e']
    cumulative = 0
    key_suppliers = []

    for supplier_id, data in sorted_suppliers:
        cumulative += data['co2e']
        key_suppliers.append(supplier_id)

        if cumulative / total_co2e >= 0.80:
            break

    print(f"πŸ“Š {len(key_suppliers)} suppliers represent 80% of emissions")
    return key_suppliers

# Request EPDs from these key suppliers

3. Track Data Quality Over Time

def calculate_data_quality_score(purchases):
    """Track improvement in data quality"""

    total = len(purchases)
    supplier_specific = sum(1 for p in purchases if p.get('custom_emission_factor_id'))
    activity_based = sum(1 for p in purchases if p.get('quantity') and not p.get('custom_emission_factor_id'))
    spend_based = total - supplier_specific - activity_based

    quality_score = (
        (supplier_specific * 1.0) +
        (activity_based * 0.6) +
        (spend_based * 0.3)
    ) / total * 100

    return {
        'quality_score': quality_score,
        'supplier_specific_pct': supplier_specific / total * 100,
        'activity_based_pct': activity_based / total * 100,
        'spend_based_pct': spend_based / total * 100
    }

# Track quarterly
# Goal: Increase quality score from 30% β†’ 80%

4. Validate Purchase Data

def validate_purchase(purchase):
    """Validate before submission"""

    # Check required fields
    assert purchase.get('name'), "Name required"
    assert purchase.get('category'), "Category required"
    assert purchase.get('purchase_date'), "Date required"

    # Ensure either quantity or spend is provided
    has_quantity = purchase.get('quantity') and purchase.get('unit_id')
    has_spend = purchase.get('spend_amount') and purchase.get('spend_currency')

    assert has_quantity or has_spend, "Must provide either quantity+unit or spend_amount+currency"

    # Sanity checks
    if has_quantity and purchase['quantity'] <= 0:
        return False, "Quantity must be positive"

    if has_spend and purchase['spend_amount'] <= 0:
        return False, "Spend must be positive"

    # Recommend supplier linkage
    if not purchase.get('supplier_id'):
        print("⚠️  Warning: No supplier linked. Supplier tracking recommended.")

    return True, "OK"

Troubleshooting

Issue: Unexpectedly High Emissions

# Check which factor was used
purchase = requests.get(
    f"https://api.dcycle.io/api/v1/purchases/{purchase_id}",
    headers=headers
).json()

print(f"Emission factor used: {purchase['emission_factor_used']}")
print(f"Method: {purchase['calculation_method']}")  # activity-based or spend-based
print(f"CO2e: {purchase['co2e']:.2f} kg")

# If using generic factor, consider requesting EPD from supplier
if not purchase.get('custom_emission_factor_id'):
    print("πŸ’‘ Tip: Request EPD from supplier for more accurate data")

Issue: Can’t Find Appropriate Category

# Use most specific category available
# If unsure, start broad and refine later

# Too broad: "materials"
# Better: "metals"
# Best: "metals_aluminum"

# You can update category later:
requests.patch(
    f"https://api.dcycle.io/api/v1/purchases/{purchase_id}",
    headers=headers,
    json={"category": "metals_aluminum"}
)

Issue: Missing Supplier ID

# Link purchase to supplier retroactively
requests.patch(
    f"https://api.dcycle.io/api/v1/purchases/{purchase_id}",
    headers=headers,
    json={"supplier_id": supplier_id}
)

# Or bulk update via CSV

Next Steps