mcpField
C
carbon-footprint-tracking

carbon-footprint-tracking

github.com/kishorkukreja数据与数据库8456 次安装3 周前更新
代码仓库

关于 carbon-footprint-tracking

面向组织碳核算与供应链减排的技能,帮助确定排放范围、数据需求和报告方式。内容覆盖 Scope 1、2、3,排放因子计算、供应商排放跟踪和减排机会分析。

主要功能

  1. 01确定排放范围和测量边界
  2. 02按活动数据与排放因子计算
  3. 03跟踪供应商排放数据
  4. 04分析减排机会
  5. 05整理碳排放报告

适用场景

  1. 01编制 Scope 1、2、3 排放清单
  2. 02评估供应商数据质量
  3. 03跟踪减排目标与举措进展

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/kishorkukreja/awesome-supply-chain --skill carbon-footprint-tracking

安装后在 Claude Code 中输入/carbon-footprint-tracking确认已加载。

详细介绍

碳足迹追踪

你是碳足迹测量和供应链脱碳领域的专家。你的目标是帮助组织准确测量、追踪、报告和减少供应链运营中的温室气体(GHG)排放。

初步评估

在实施碳排放追踪之前,请了解:

  1. 组织背景

    • 碳排放追踪的驱动因素是什么?(合规、报告、减排目标)
    • 当前碳核算成熟度如何?
    • 是否承诺实现净零排放或减少碳排放?
    • 有哪些监管要求?(CDP、TCFD、SEC 气候规则)
  2. 测量范围

    • 追踪哪些排放范围?(范围 1、2、3)
    • 地理覆盖范围?(设施、区域、全球)
    • 供应链深度?(一级、多级)
    • 产品级还是企业级足迹?
  3. 数据可用性

    • 是否有能源消耗数据?
    • 是否追踪运输数据?
    • 是否能获取供应商排放数据?
    • 活动数据的质量和完整性如何?
  4. 报告要求

    • 内部目标和 KPI?
    • 外部报告框架?(GRI、CDP、SASB)
    • 利益相关方的期望?(投资者、客户、员工)
    • 是否需要核查和鉴证?

GHG Protocol 框架

排放范围

范围 1:直接排放

  • 公司拥有的车辆和设备
  • 现场燃料燃烧
  • 制造流程
  • 逸散排放(制冷剂、泄漏)

范围 2:间接能源排放

  • 外购电力
  • 外购供暖和制冷
  • 外购蒸汽

范围 3:价值链排放

  • 上游:
    • 外购商品和服务
    • 资本货物
    • 运输和配送(上游)
    • 商务旅行
    • 员工通勤
    • 废弃物处理
    • 租赁资产(上游)
  • 下游:
    • 运输和配送(下游)
    • 产品使用
    • 生命周期末端处理
    • 特许经营
    • 投资
排放类别优先级
类别占总量的典型比例测量复杂度优先级
范围 3:外购商品40-70%高关键
范围 3:上游运输10-20%中高
范围 2:电力5-15%低高
范围 1:设施5-10%低中
范围 3:产品使用10-30%高中
范围 3:生命周期末端2-5%中低

碳排放计算方法

排放因子法

基本公式:

CO2e Emissions = Activity Data × Emission Factor

Where:
  Activity Data = Quantity of activity (kWh, liters, kg, tkm, etc.)
  Emission Factor = Emissions per unit (kgCO2e per unit)
  CO2e = Carbon dioxide equivalent (includes all GHGs)

Python 实现:

import pandas as pd
import numpy as np

class CarbonFootprintCalculator:
    """Comprehensive carbon footprint calculator"""

    def __init__(self):
        self.emission_factors = self._load_emission_factors()
        self.gwp_factors = {
            'CO2': 1,
            'CH4': 25,      # Methane (100-year GWP)
            'N2O': 298,     # Nitrous oxide
            'HFCs': 1430,   # Hydrofluorocarbons (avg)
            'PFCs': 7390,   # Perfluorocarbons (avg)
            'SF6': 22800    # Sulfur hexafluoride
        }

    def _load_emission_factors(self):
        """Load standard emission factors database"""
        # Based on EPA, DEFRA, and other sources
        return {
            # Energy (kgCO2e per kWh)
            'electricity_us_grid': 0.417,
            'electricity_eu_grid': 0.295,
            'electricity_renewable': 0.000,
            'natural_gas': 0.202,  # per kWh

            # Fuels (kgCO2e per liter)
            'diesel': 2.68,
            'gasoline': 2.31,
            'jet_fuel': 2.50,

            # Transportation (kgCO2e per tonne-km)
            'truck_full_truckload': 0.062,
            'truck_less_than_truckload': 0.091,
            'rail': 0.022,
            'ocean_shipping': 0.008,
            'air_freight': 0.602,

            # Materials (kgCO2e per kg)
            'steel': 1.85,
            'aluminum': 8.24,
            'plastic_pet': 2.15,
            'cardboard': 0.95,
            'glass': 0.85,
            'concrete': 0.11,

            # Manufacturing (kgCO2e per unit - examples)
            'electronics_assembly': 50,
            'textile_production': 15,
            'food_processing': 2.5
        }

    def calculate_scope1_facilities(self, fuel_consumption):
        """
        Calculate Scope 1 emissions from facility fuel use

        fuel_consumption: dict with fuel types and quantities
        Example: {'diesel_liters': 10000, 'natural_gas_kwh': 50000}
        """
        emissions = 0
        breakdown = []

        # Diesel/gasoline combustion
        if 'diesel_liters' in fuel_consumption:
            diesel_co2 = fuel_consumption['diesel_liters'] * self.emission_factors['diesel']
            emissions += diesel_co2
            breakdown.append({
                'source': 'Diesel combustion',
                'activity': fuel_consumption['diesel_liters'],
                'unit': 'liters',
                'emissions_kgco2e': diesel_co2
            })

        if 'gasoline_liters' in fuel_consumption:
            gas_co2 = fuel_consumption['gasoline_liters'] * self.emission_factors['gasoline']
            emissions += gas_co2
            breakdown.append({
                'source': 'Gasoline combustion',
                'activity': fuel_consumption['gasoline_liters'],
                'unit': 'liters',
                'emissions_kgco2e': gas_co2
            })

        # Natural gas (converted to kWh)
        if 'natural_gas_kwh' in fuel_consumption:
            ng_co2 = fuel_consumption['natural_gas_kwh'] * self.emission_factors['natural_gas']
            emissions += ng_co2
            breakdown.append({
                'source': 'Natural gas',
                'activity': fuel_consumption['natural_gas_kwh'],
                'unit': 'kWh',
                'emissions_kgco2e': ng_co2
            })

        return {
            'scope': 'Scope 1',
            'total_emissions_kgco2e': round(emissions, 2),
            'total_emissions_tco2e': round(emissions / 1000, 2),
            'breakdown': breakdown
        }

    def calculate_scope1_fleet(self, vehicle_data):
        """
        Calculate Scope 1 emissions from company fleet

        vehicle_data: list of dicts with vehicle info
        Example: [{'type': 'diesel', 'distance_km': 50000, 'fuel_efficiency_l_per_100km': 8}]
        """
        emissions = 0
        breakdown = []

        for vehicle in vehicle_data:
            distance = vehicle['distance_km']
            fuel_efficiency = vehicle['fuel_efficiency_l_per_100km']
            fuel_type = vehicle['type']

            # Calculate fuel consumed
            fuel_consumed = (distance / 100) * fuel_efficiency

            # Get emission factor
            if fuel_type in ['diesel']:
                ef = self.emission_factors['diesel']
            elif fuel_type in ['gasoline', 'petrol']:
                ef = self.emission_factors['gasoline']
            else:
                ef = 2.5  # default

            vehicle_emissions = fuel_consumed * ef
            emissions += vehicle_emissions

            breakdown.append({
                'vehicle_id': vehicle.get('id', 'Unknown'),
                'fuel_type': fuel_type,
                'distance_km': distance,
                'fuel_consumed_liters': round(fuel_consumed, 2),
                'emissions_kgco2e': round(vehicle_emissions, 2)
            })

        return {
            'scope': 'Scope 1 - Fleet',
            'total_emissions_kgco2e': round(emissions, 2),
            'total_emissions_tco2e': round(emissions / 1000, 2),
            'breakdown': breakdown
        }

    def calculate_scope2_electricity(self, electricity_consumption, region='us', renewable_pct=0):
        """
        Calculate Scope 2 emissions from electricity

        electricity_consumption: kWh consumed
        region: 'us', 'eu', or custom
        renewable_pct: percentage of renewable energy (0-1)
        """

        # Select emission factor based on region
        if region == 'us':
            ef = self.emission_factors['electricity_us_grid']
        elif region == 'eu':
            ef = self.emission_factors['electricity_eu_grid']
        else:
            ef = 0.40  # global average

        # Adjust for renewable percentage
        grid_electricity = electricity_consumption * (1 - renewable_pct)
        renewable_electricity = electricity_consumption * renewable_pct

        emissions_grid = grid_electricity * ef
        emissions_renewable = renewable_electricity * self.emission_factors['electricity_renewable']

        total_emissions = emissions_grid + emissions_renewable

        return {
            'scope': 'Scope 2',
            'total_electricity_kwh': electricity_consumption,
            'grid_electricity_kwh': grid_electricity,
            'renewable_electricity_kwh': renewable_electricity,
            'emission_factor_kgco2e_per_kwh': ef,
            'total_emissions_kgco2e': round(total_emissions, 2),
            'total_emissions_tco2e': round(total_emissions / 1000, 2)
        }

    def calculate_scope3_transportation(self, shipments):
        """
        Calculate Scope 3 emissions from transportation

        shipments: list of dicts
        Example: [{'mode': 'truck_ftl', 'distance_km': 500, 'weight_tonnes': 20}]
        """
        emissions = 0
        breakdown = []

        for shipment in shipments:
            mode = shipment['mode']
            distance = shipment['distance_km']
            weight = shipment['weight_tonnes']

            # Calculate tonne-kilometers
            tkm = distance * weight

            # Get emission factor
            mode_map = {
                'truck_ftl': 'truck_full_truckload',
                'truck_ltl': 'truck_less_than_truckload',
                'rail': 'rail',
                'ocean': 'ocean_shipping',
                'air': 'air_freight'
            }

            ef_key = mode_map.get(mode, 'truck_full_truckload')
            ef = self.emission_factors[ef_key]

            shipment_emissions = tkm * ef
            emissions += shipment_emissions

            breakdown.append({
                'shipment_id': shipment.get('id', 'Unknown'),
                'mode': mode,
                'distance_km': distance,
                'weight_tonnes': weight,
                'tonne_km': tkm,
                'emission_factor': ef,
                'emissions_kgco2e': round(shipment_emissions, 2)
            })

        return {
            'scope': 'Scope 3 - Transportation',
            'total_emissions_kgco2e': round(emissions, 2),
            'total_emissions_tco2e': round(emissions / 1000, 2),
            'breakdown': breakdown
        }

    def calculate_scope3_materials(self, materials_purchased):
        """
        Calculate Scope 3 emissions from purchased materials

        materials_purchased: dict with material types and quantities (kg)
        Example: {'steel': 10000, 'plastic_pet': 5000}
        """
        emissions = 0
        breakdown = []

        for material, quantity_kg in materials_purchased.items():
            if material in self.emission_factors:
                ef = self.emission_factors[material]
                material_emissions = quantity_kg * ef
                emissions += material_emissions

                breakdown.append({
                    'material': material,
                    'quantity_kg': quantity_kg,
                    'emission_factor': ef,
                    'emissions_kgco2e': round(material_emissions, 2)
                })

        return {
            'scope': 'Scope 3 - Materials',
            'total_emissions_kgco2e': round(emissions, 2),
            'total_emissions_tco2e': round(emissions / 1000, 2),
            'breakdown': breakdown
        }

    def calculate_product_carbon_footprint(self, product_data):
        """
        Calculate product-level carbon footprint (cradle-to-gate)

        product_data: dict with all product lifecycle data
        """
        total_emissions = 0
        lifecycle_breakdown = {}

        # Materials extraction and processing
        if 'materials' in product_data:
            materials_result = self.calculate_scope3_materials(product_data['materials'])
            lifecycle_breakdown['materials'] = materials_result
            total_emissions += materials_result['total_emissions_kgco2e']

        # Manufacturing
        if 'manufacturing' in product_data:
            mfg_energy = product_data['manufacturing'].get('energy_kwh', 0)
            mfg_result = self.calculate_scope2_electricity(
                mfg_energy,
                region=product_data['manufacturing'].get('region', 'us')
            )
            lifecycle_breakdown['manufacturing'] = mfg_result
            total_emissions += mfg_result['total_emissions_kgco2e']

        # Transportation to customer
        if 'transportation' in product_data:
            transport_result = self.calculate_scope3_transportation(
                product_data['transportation']
            )
            lifecycle_breakdown['transportation'] = transport_result
            total_emissions += transport_result['total_emissions_kgco2e']

        # Use phase (if applicable)
        if 'use_phase' in product_data:
            use_emissions = product_data['use_phase'].get('emissions_kgco2e', 0)
            lifecycle_breakdown['use_phase'] = {
                'emissions_kgco2e': use_emissions
            }
            total_emissions += use_emissions

        # End of life
        if 'end_of_life' in product_data:
            eol_emissions = product_data['end_of_life'].get('emissions_kgco2e', 0)
            lifecycle_breakdown['end_of_life'] = {
                'emissions_kgco2e': eol_emissions
            }
            total_emissions += eol_emissions

        return {
            'product_id': product_data.get('product_id', 'Unknown'),
            'total_carbon_footprint_kgco2e': round(total_emissions, 2),
            'lifecycle_breakdown': lifecycle_breakdown,
            'per_unit_emissions': round(total_emissions / product_data.get('units', 1), 2)
        }

    def calculate_emissions_by_scope(self, scope1_data, scope2_data, scope3_data):
        """Generate comprehensive emissions inventory by scope"""

        scope1_result = self.calculate_scope1_facilities(scope1_data['facilities'])
        scope1_fleet = self.calculate_scope1_fleet(scope1_data['fleet'])

        scope2_result = self.calculate_scope2_electricity(
            scope2_data['electricity_kwh'],
            region=scope2_data.get('region', 'us'),
            renewable_pct=scope2_data.get('renewable_pct', 0)
        )

        scope3_transport = self.calculate_scope3_transportation(scope3_data['shipments'])
        scope3_materials = self.calculate_scope3_materials(scope3_data['materials'])

        total_scope1 = (scope1_result['total_emissions_kgco2e'] +
                       scope1_fleet['total_emissions_kgco2e'])
        total_scope2 = scope2_result['total_emissions_kgco2e']
        total_scope3 = (scope3_transport['total_emissions_kgco2e'] +
                       scope3_materials['total_emissions_kgco2e'])

        total_emissions = total_scope1 + total_scope2 + total_scope3

        return {
            'total_emissions_tco2e': round(total_emissions / 1000, 2),
            'scope1_tco2e': round(total_scope1 / 1000, 2),
            'scope2_tco2e': round(total_scope2 / 1000, 2),
            'scope3_tco2e': round(total_scope3 / 1000, 2),
            'scope1_percentage': round(total_scope1 / total_emissions * 100, 1),
            'scope2_percentage': round(total_scope2 / total_emissions * 100, 1),
            'scope3_percentage': round(total_scope3 / total_emissions * 100, 1),
            'detailed_results': {
                'scope1_facilities': scope1_result,
                'scope1_fleet': scope1_fleet,
                'scope2_electricity': scope2_result,
                'scope3_transportation': scope3_transport,
                'scope3_materials': scope3_materials
            }
        }


# Example usage
calculator = CarbonFootprintCalculator()

# Example calculation
scope1_data = {
    'facilities': {
        'diesel_liters': 5000,
        'natural_gas_kwh': 100000
    },
    'fleet': [
        {'id': 'V001', 'type': 'diesel', 'distance_km': 50000, 'fuel_efficiency_l_per_100km': 8},
        {'id': 'V002', 'type': 'diesel', 'distance_km': 30000, 'fuel_efficiency_l_per_100km': 9}
    ]
}

scope2_data = {
    'electricity_kwh': 500000,
    'region': 'us',
    'renewable_pct': 0.20  # 20% renewable
}

scope3_data = {
    'shipments': [
        {'id': 'S001', 'mode': 'truck_ftl', 'distance_km': 800, 'weight_tonnes': 20},
        {'id': 'S002', 'mode': 'ocean', 'distance_km': 5000, 'weight_tonnes': 100},
        {'id': 'S003', 'mode': 'air', 'distance_km': 2000, 'weight_tonnes': 2}
    ],
    'materials': {
        'steel': 50000,
        'plastic_pet': 10000,
        'cardboard': 5000
    }
}

results = calculator.calculate_emissions_by_scope(scope1_data, scope2_data, scope3_data)

print(f"Total Emissions: {results['total_emissions_tco2e']} tCO2e")
print(f"Scope 1: {results['scope1_tco2e']} tCO2e ({results['scope1_percentage']}%)")
print(f"Scope 2: {results['scope2_tco2e']} tCO2e ({results['scope2_percentage']}%)")
print(f"Scope 3: {results['scope3_tco2e']} tCO2e ({results['scope3_percentage']}%)")

供应商排放追踪

一手数据收集
class SupplierEmissionsTracker:
    """Track and manage supplier carbon footprint data"""

    def __init__(self):
        self.suppliers = {}
        self.data_quality_tiers = {
            'tier1_primary': 1.0,      # Supplier-specific verified data
            'tier2_secondary': 0.8,    # Industry averages
            'tier3_estimated': 0.5     # Spend-based estimates
        }

    def add_supplier(self, supplier_id, supplier_name, spend, emissions_data):
        """
        Add supplier with emissions data

        emissions_data: dict with emissions info and data quality
        """
        self.suppliers[supplier_id] = {
            'name': supplier_name,
            'annual_spend': spend,
            'emissions_data': emissions_data,
            'data_quality': emissions_data.get('data_quality', 'tier3_estimated'),
            'last_updated': pd.Timestamp.now()
        }

    def calculate_supplier_emissions(self, supplier_id):
        """Calculate emissions for a specific supplier"""
        supplier = self.suppliers[supplier_id]
        emissions_data = supplier['emissions_data']

        if supplier['data_quality'] == 'tier1_primary':
            # Use supplier-reported data
            return emissions_data.get('reported_emissions_tco2e', 0)

        elif supplier['data_quality'] == 'tier2_secondary':
            # Use industry average emission factors
            category = emissions_data.get('category', 'general_manufacturing')
            revenue = emissions_data.get('revenue', 1000000)

            # Industry emission intensities (tCO2e per $1M revenue)
            intensity_map = {
                'general_manufacturing': 120,
                'electronics': 80,
                'chemicals': 250,
                'textiles': 150,
                'food_processing': 90,
                'logistics': 200
            }

            intensity = intensity_map.get(category, 100)
            emissions = (revenue / 1000000) * intensity
            return emissions

        else:  # tier3_estimated
            # Spend-based estimation
            spend = supplier['annual_spend']

            # Average emission factor: tCO2e per $1000 spend
            spend_emission_factor = 0.5

            emissions = (spend / 1000) * spend_emission_factor
            return emissions

    def generate_supplier_emissions_report(self):
        """Generate comprehensive supplier emissions report"""

        report_data = []

        for supplier_id, supplier in self.suppliers.items():
            emissions = self.calculate_supplier_emissions(supplier_id)

            data_quality = supplier['data_quality']
            quality_score = self.data_quality_tiers[data_quality]

            report_data.append({
                'supplier_id': supplier_id,
                'supplier_name': supplier['name'],
                'annual_spend': supplier['annual_spend'],
                'emissions_tco2e': round(emissions, 2),
                'data_quality': data_quality,
                'quality_score': quality_score,
                'emissions_intensity_per_1k_spend': round(emissions / (supplier['annual_spend'] / 1000), 2)
            })

        df = pd.DataFrame(report_data)

        # Sort by emissions descending
        df = df.sort_values('emissions_tco2e', ascending=False)

        # Calculate cumulative percentage
        total_emissions = df['emissions_tco2e'].sum()
        df['cumulative_pct'] = df['emissions_tco2e'].cumsum() / total_emissions * 100

        # Classify suppliers by emissions contribution
        df['classification'] = df['cumulative_pct'].apply(
            lambda x: 'A (High)' if x <= 80 else ('B (Medium)' if x <= 95 else 'C (Low)')
        )

        return df

    def identify_engagement_priorities(self, threshold_pct=80):
        """Identify suppliers to prioritize for emissions reduction engagement"""

        df = self.generate_supplier_emissions_report()

        # Prioritize suppliers contributing to top 80% of emissions
        priority_suppliers = df[df['cumulative_pct'] <= threshold_pct].copy()

        # Also prioritize poor data quality with high spend
        high_spend_poor_quality = df[
            (df['annual_spend'] > df['annual_spend'].quantile(0.75)) &
            (df['quality_score'] < 0.7)
        ]

        # Combine
        priority_list = pd.concat([priority_suppliers, high_spend_poor_quality]).drop_duplicates()

        return priority_list.sort_values('emissions_tco2e', ascending=False)


# Example usage
tracker = SupplierEmissionsTracker()

# Add suppliers
tracker.add_supplier(
    'SUP001',
    'Acme Manufacturing',
    spend=5000000,
    emissions_data={
        'reported_emissions_tco2e': 8500,
        'data_quality': 'tier1_primary',
        'verification': 'third_party'
    }
)

tracker.add_supplier(
    'SUP002',
    'Beta Chemicals',
    spend=3000000,
    emissions_data={
        'category': 'chemicals',
        'revenue': 50000000,
        'data_quality': 'tier2_secondary'
    }
)

tracker.add_supplier(
    'SUP003',
    'Gamma Logistics',
    spend=1500000,
    emissions_data={
        'data_quality': 'tier3_estimated'
    }
)

# Generate report
report = tracker.generate_supplier_emissions_report()
print(report[['supplier_name', 'emissions_tco2e', 'data_quality', 'classification']])

# Identify priorities
priorities = tracker.identify_engagement_priorities()
print("\nPriority Suppliers for Engagement:")
print(priorities[['supplier_name', 'emissions_tco2e', 'quality_score']])

碳减排策略

减排机会分析
class CarbonReductionPlanner:
    """Identify and prioritize carbon reduction opportunities"""

    def __init__(self, baseline_emissions):
        self.baseline_emissions = baseline_emissions
        self.reduction_initiatives = []

    def add_initiative(self, initiative_name, category, reduction_potential_tco2e,
                      investment_required, timeline_years, complexity='medium'):
        """
        Add carbon reduction initiative

        complexity: 'low', 'medium', 'high'
        """

        # Calculate abatement cost ($ per tCO2e reduced)
        annual_reduction = reduction_potential_tco2e
        abatement_cost = investment_required / (annual_reduction * timeline_years) if annual_reduction > 0 else float('inf')

        self.reduction_initiatives.append({
            'initiative': initiative_name,
            'category': category,
            'reduction_potential_tco2e': reduction_potential_tco2e,
            'reduction_percentage': (reduction_potential_tco2e / self.baseline_emissions) * 100,
            'investment_required': investment_required,
            'timeline_years': timeline_years,
            'complexity': complexity,
            'abatement_cost_per_tco2e': round(abatement_cost, 2)
        })

    def prioritize_initiatives(self):
        """Prioritize initiatives using cost-benefit analysis"""

        df = pd.DataFrame(self.reduction_initiatives)

        # Score each initiative
        # Lower cost, higher reduction, lower complexity = higher score

        # Normalize metrics (0-100 scale)
        if len(df) > 0:
            df['cost_score'] = 100 - (
                (df['abatement_cost_per_tco2e'] - df['abatement_cost_per_tco2e'].min()) /
                (df['abatement_cost_per_tco2e'].max() - df['abatement_cost_per_tco2e'].min() + 0.01) * 100
            )

            df['impact_score'] = (
                (df['reduction_potential_tco2e'] - df['reduction_potential_tco2e'].min()) /
                (df['reduction_potential_tco2e'].max() - df['reduction_potential_tco2e'].min() + 0.01) * 100
            )

            complexity_map = {'low': 100, 'medium': 60, 'high': 30}
            df['complexity_score'] = df['complexity'].map(complexity_map)

            # Overall priority score (weighted)
            df['priority_score'] = (
                df['impact_score'] * 0.4 +
                df['cost_score'] * 0.35 +
                df['complexity_score'] * 0.25
            )

            # Classify priority
            df['priority'] = pd.cut(
                df['priority_score'],
                bins=[0, 40, 70, 100],
                labels=['Low', 'Medium', 'High']
            )

            df = df.sort_values('priority_score', ascending=False)

        return df

    def generate_reduction_roadmap(self, target_reduction_pct, budget_limit):
        """
        Generate optimal roadmap to achieve reduction target within budget

        Uses greedy algorithm to select best initiatives
        """

        df = self.prioritize_initiatives()

        # Sort by priority score
        df = df.sort_values('priority_score', ascending=False)

        target_reduction_tco2e = self.baseline_emissions * (target_reduction_pct / 100)

        selected_initiatives = []
        total_reduction = 0
        total_investment = 0

        for idx, initiative in df.iterrows():
            # Check if we can afford it
            if total_investment + initiative['investment_required'] <= budget_limit:
                # Check if we still need more reduction
                if total_reduction < target_reduction_tco2e:
                    selected_initiatives.append(initiative)
                    total_reduction += initiative['reduction_potential_tco2e']
                    total_investment += initiative['investment_required']

        achievement_pct = (total_reduction / self.baseline_emissions) * 100

        return {
            'selected_initiatives': pd.DataFrame(selected_initiatives),
            'total_reduction_tco2e': round(total_reduction, 2),
            'achievement_percentage': round(achievement_pct, 1),
            'total_investment': total_investment,
            'remaining_budget': budget_limit - total_investment,
            'gap_to_target_tco2e': round(target_reduction_tco2e - total_reduction, 2)
        }


# Example reduction planning
baseline = 10000  # tCO2e
planner = CarbonReductionPlanner(baseline)

# Add various initiatives
planner.add_initiative(
    'Switch to renewable electricity',
    category='Energy',
    reduction_potential_tco2e=2000,
    investment_required=500000,
    timeline_years=10,
    complexity='low'
)

planner.add_initiative(
    'Fleet electrification',
    category='Transportation',
    reduction_potential_tco2e=1500,
    investment_required=2000000,
    timeline_years=5,
    complexity='high'
)

planner.add_initiative(
    'Supplier engagement program',
    category='Supply Chain',
    reduction_potential_tco2e=3000,
    investment_required=200000,
    timeline_years=3,
    complexity='medium'
)

planner.add_initiative(
    'Energy efficiency improvements',
    category='Operations',
    reduction_potential_tco2e=800,
    investment_required=300000,
    timeline_years=7,
    complexity='low'
)

planner.add_initiative(
    'Modal shift to rail',
    category='Transportation',
    reduction_potential_tco2e=600,
    investment_required=150000,
    timeline_years=5,
    complexity='medium'
)

# Prioritize all initiatives
priorities = planner.prioritize_initiatives()
print("All Initiatives (Prioritized):")
print(priorities[['initiative', 'reduction_potential_tco2e', 'abatement_cost_per_tco2e', 'priority']])

# Generate roadmap
roadmap = planner.generate_reduction_roadmap(
    target_reduction_pct=40,  # 40% reduction target
    budget_limit=1500000
)

print(f"\n\nReduction Roadmap:")
print(f"Target: 40% reduction from {baseline} tCO2e")
print(f"Achieved: {roadmap['achievement_percentage']}% ({roadmap['total_reduction_tco2e']} tCO2e)")
print(f"Investment: ${roadmap['total_investment']:,.0f}")
print(f"Gap to target: {roadmap['gap_to_target_tco2e']} tCO2e")
print(f"\nSelected Initiatives:")
print(roadmap['selected_initiatives'][['initiative', 'reduction_potential_tco2e', 'investment_required']])

工具和库

Python 库

碳排放计算:

  • pandas:数据操作和分析
  • numpy:数值计算
  • scipy:统计分析

数据收集:

  • requests:用于集成排放因子的 API
  • beautifulsoup4:用于抓取数据的网页解析
  • sqlalchemy:数据库管理

可视化:

  • matplotlib、seaborn:排放图表
  • plotly:交互式仪表板
  • folium:地理排放地图

专用工具:

  • brightway2:生命周期评估(LCA)
  • openLCA:开源 LCA 框架
  • carbon-footprint:碳排放计算工具
商业软件

碳核算平台:

  • Watershed:企业碳核算
  • Persefoni:气候管理与核算
  • Plan A:碳管理平台
  • Normative:企业碳核算
  • Sweep:碳测量与减排
  • CarbonChain:供应链碳追踪

可持续发展平台:

  • SAP Product Footprint Management:产品级碳核算
  • Enablon:ESG 与可持续发展管理
  • Sphera:可持续发展绩效管理
  • thinkstep (Sphera):LCA 软件

报告平台:

  • Workiva:ESG 报告
  • Emitwise:碳核算与报告
  • FigBytes:净零规划
排放因子数据库
  • EPA Emission Factors:美国 EPA 数据
  • DEFRA:英国政府转换因子
  • Ecoinvent:全球 LCA 数据库
  • GHG Protocol:计算工具与指南
  • Carbon Trust:碳足迹工具
  • GLEC Framework:物流排放

常见挑战与解决方案

挑战:范围 3 数据收集

问题:

  • 供应商不愿分享排放数据
  • 数据质量不一致
  • 对一级供应商以外的情况缺乏可见性

解决方案:

  • 分阶段提升数据质量(基于支出 → 行业平均值 → 一手数据)
  • 通过激励措施开展供应商参与计划
  • 行业协作倡议
  • 第三方数据提供商(CDP Supply Chain、EcoVadis)
  • 在合同中要求披露排放数据
  • 为供应商提供标准化模板和工具
挑战:排放因子的选择

问题:

  • 有多种排放因子来源可用
  • 区域因子与全球因子之间的选择
  • 计算结果存在不确定性
  • 保持因子及时更新

解决方案:

  • 记录方法和因子来源
  • 使用可用的最具体因子(供应商特定 > 行业 > 通用)
  • 为趋势分析保持时间上的一致性
  • 每年审查并更新因子
  • 对关键假设进行敏感性分析
  • 遵循 GHG Protocol 指南
挑战:数据质量与完整性

问题:

  • 活动数据缺失
  • 数据格式不一致
  • 缺乏自动化数据收集
  • 手动流程容易出错

解决方案:

  • 实施数据治理框架
  • 尽可能实现数据收集自动化
  • 与 ERP、TMS、WMS 系统集成
  • 定期进行数据质量审计
  • 使用估算方法填补缺口(记录假设)
  • 优先收集重要排放源的数据
挑战:设定基准线

问题:

  • 无法获取历史数据
  • 业务变化(收购、剥离)
  • 选择基准年
  • 重新计算的触发条件

解决方案:

  • 选择数据质量最佳的基准年
  • 明确记录基准线范围
  • 制定重新计算政策(通常变化超过 5% 时触发重新计算)
  • 同时追踪绝对指标和强度指标
  • 使用标准化指标(每单位收入、每单位产出)
挑战:减排归因

问题:

  • 难以衡量具体举措的影响
  • 业务增长与效率提升之间的区分
  • 外部因素(电网脱碳)

解决方案:

  • 追踪强度指标(单位产出的排放量)
  • 建立测量与验证(M&V)规程
  • 尽可能使用对照组
  • 仔细记录基准线和变化
  • 将有机减排与业务变化分开
  • 定期审查进展,并设定明确的 KPI
挑战:平衡成本与准确性

问题:

  • 完美测量成本高昂
  • 数据质量提升的收益递减
  • 资源有限

解决方案:

  • 运用重要性原则(聚焦重要排放源)
  • 采用分层方法(范围 1 和 2 进行详细核算,部分范围 3 进行估算)
  • 优先提升最大排放源的数据质量
  • 接受对大量小型排放源进行估算
  • 记录数据质量评估
  • 随时间持续改进

输出格式

碳足迹报告

执行摘要:

  • 碳足迹总量(tCO2e)
  • 同比变化
  • 减排目标进展
  • 主要排放源
  • 重大举措及其影响

按范围划分的排放清单:

范围类别排放量(tCO2e)占总量的百分比与上一年的对比
范围 1设施8508.5%-5%
范围 1车队1,20012.0%-2%
范围 2电力1,50015.0%-12%
范围 3采购商品4,20042.0%+3%
范围 3运输1,80018.0%-8%
范围 3其他4504.5%-1%
总计10,000100%-3%

按业务部门划分的排放量:

业务部门排放量(tCO2e)排放强度(tCO2e/百万美元营收)目标状态
制造5,500220200落后
分销2,800180190按计划进行
零售1,7008590超前

供应商排放:

供应商支出(百万美元)排放量(tCO2e)数据质量优先级
供应商 A5.02,100原始数据A
供应商 B3.51,800行业平均值A
供应商 C2.0300估算值B

减排举措:

举措状态减排量(tCO2e/年)投资时间表投资回报期
可再生能源已实施1,200$500K已完成4.2 年
车队优化进行中400$150K2026 年第 3 季度2.8 年
供应商计划规划中800$200K2027 年待定

数据质量评估:

范围覆盖率原始数据占比估算占比质量评分
范围 1100%100%0%高
范围 2100%95%5%高
范围 385%30%70%中

要提出的问题

如果需要更多背景信息:

  1. 碳排放追踪计划的驱动因素是什么?(报告、减排、合规)
  2. 需要测量哪些排放范围?(范围 1、2、3 或特定类别)
  3. 目前有哪些可用数据?(能源账单、运输记录、供应商数据)
  4. 是否已有碳减排目标或承诺?
  5. 必须遵循哪些报告框架?(GHG Protocol、CDP、GRI、TCFD)
  6. 组织边界是什么?(地点、业务部门、地理区域)
  7. 是否需要产品级碳足迹核算?
  8. 期望的数据质量和准确度如何?
  9. 是否有核查或鉴证要求?
  10. 目前使用哪些系统?(ERP、TMS、能源管理系统)

相关技能

  • circular-economy:用于循环闭环和减废策略
  • sustainable-sourcing:用于环境友好型采购
  • compliance-management:用于满足法规要求和报告
  • risk-mitigation:用于应对气候相关的供应链风险
  • supplier-selection:用于在采购中纳入碳排放标准
  • network-design:用于优化供应链配置以减少排放
  • route-optimization:用于减少运输排放
  • demand-forecasting:用于准确规划生产以尽量减少浪费

评论

支持 Markdown