Emisyon Boru Hattı Kapsam 3 Değer Zinciri: Ham Verilerden Denetim Takibine
Il Toplam emisyonların %70-90'ı bir teknoloji veya yazılım şirketinin kendi şirketinin dışında saklanması sınırlar: hizmet olarak satın alınan bulut sunucularında, çalışanların dizüstü bilgisayarlarında, iş uçuşlarında, kodda Müşterilerin cihazlarında çalıştırdığı. Bunlar emisyonlar Kapsam 3ve çoğunlukla Dijital kuruluşların bir kısmı, var olan en karmaşık ölçüm problemini temsil etmektedir. ESG alanı.
Kapsam 1 (yakıtı yakarız) ve Kapsam 2'den (elektriği satın alırız) farklı olarak, Kapsam 3, yüzlerce tedarikçiden veri toplanmasını, heterojen emisyon faktörlerinin uygulanmasını, çok yüksek derecede belirsizliği yönetin ve dış denetçiler tarafından doğrulanabilen bir denetim izi oluşturun. ile CSRD/ESRS E1 Bu da büyük şirketler için Kapsam 3 raporlamasını zorunlu kılmaktadır. 2025-2026 ve 2028 itibarıyla hedefte KOBİ'ler varken, soru artık akademik değil: mühendislik.
Bu yazıda bir tane inşa ediyoruz komple boru hattı Kapsam 3 emisyonlarının hesaplanması için değer zinciri: tedarikçi veri toplamaya yönelik ETL mimarisinden CDP gibi platformlarla entegrasyona kadar ve EcoVadis, aktivite bazlı ve harcama bazlı hesaplamalardan otomasyon için Airflow DAG'a kadar ve doğrulayıcılar için değişmez denetim takibine. Her bölüm çalışan Python kodunu ve en iyi uygulamaları içerir gerçek bağlamlarda operasyonel olarak test edilmiştir.
Ne Öğreneceksiniz
- 15 Sera Gazı Protokolü Kapsam 3 kategorisi ve hangilerinin yazılım/SaaS şirketleriyle ilgili olduğu
- Tedarikçi verilerinin toplanması için ETL/ELT mimarisi: anketler, CDP, EcoVadis ve doğrudan API'ler
- Etkinliğe dayalı ve harcamaya dayalı: formüller, doğruluk ve hangi yaklaşımın ne zaman kullanılacağı
- Apache Airflow ile Python ardışık düzeni: Otomatik ve ölçeklenebilir Kapsam 3 hesaplaması için DAG
- Veri kalitesi puanlaması ve tahminlerde istatistiksel belirsizliğin yayılması
- Harici doğrulayıcının izlenebilirliği için SHA-256 karma zincirine sahip değişmez denetim izi
- Değer zinciri ve ısı haritası öncelik kategorileri için Sankey diyagramı görselleştirmesi
- CSRD/ESRS E1 gereksinimleri: neleri ve hangi ayrıntı düzeyinde açıklamanız gerekir
- Tam örnek olay incelemesi: 50 tedarikçili SaaS şirketi, uçtan uca Kapsam 3 hesaplaması
- Emisyon faktörleri için EcoVadis Karbon Veri Ağı ve Climatiq API ile entegrasyon
Yeşil Yazılım Serisi — 10 Makale
| # | Öğe | Ders |
|---|---|---|
| 1 | Yeşil Yazılımın Temel İlkeleri | Karbon verimliliği, GSF, SCI |
| 2 | CodeCarbon: Kodu Ölçme | Ölçüm, kontrol paneli, optimizasyon |
| 3 | Climatiq API: Karbon Hesaplamaları | REST API, Sera Gazı Protokolü, Kapsam 1-3 |
| 4 | Karbon Bilinçli SDK | Zamanın değişmesi, yerin değişmesi |
| 5 | Kapsam 1-2-3: ÇSY Veri Modellemesi | Veri yapısı, hesaplamalar, toplama |
| 6 | GreenOps: Karbon Bilinçli Kubernetes | Planlama, ölçeklendirme, izleme |
| 7 | Emisyon Boru Hattı Kapsam 3 Değer Zinciri | Bu makale |
| 8 | ÇSY Raporlama API'si: CSRD | API, iş akışı, uyumluluk |
| 9 | Sürdürülebilir Mimari Desenler | Depolama, önbelleğe alma, toplu işlem |
| 10 | Yapay Zeka ve Karbon: ML Eğitimi | ML eğitimi, optimizasyon, Yeşil AI |
15 Sera Gazı Protokolü Kapsam 3 Kategorisi
Il Sera Gazı Protokolü Kurumsal Değer Zinciri (Kapsam 3) Standardı bu uluslararası çerçevedir Referans 2011'de yayınlandı ve şu anda 2026'da beklenen güncellemelerle revize ediliyor. değer zincirinin dolaylı emisyonları 15 farklı kategoriiki şekilde organize edilmiş makrogruplar: yukarı akış (üretim/hizmet sunumu öncesi faaliyetler) e aşağı akış (müşteriye satış sonrası faaliyetler).
15 Kapsam 3 Kategorileri: Yukarı ve Aşağı Akım
| Kedi. | İsim | Akış | SaaS/Teknoloji İlişkisi |
|---|---|---|---|
| 1 | Satın alınan mal ve hizmetler | Yukarı akış | YÜKSEK: sunucu donanımı, yazılım lisansları, danışmanlık hizmetleri |
| 2 | Sermaye malları | Yukarı akış | MEDYA: veri merkezi ekipmanları, dizüstü bilgisayarlar, şirket telefonları |
| 3 | Yakıt ve enerji ile ilgili faaliyetler | Yukarı akış | ORTALAMA: satın alınan enerji üretiminden kaynaklanan emisyonlar (yukarı yönde Kapsam 2) |
| 4 | Yukarı yönde taşıma ve dağıtım | Yukarı akış | DÜŞÜK: ofislere ve veri merkezlerine donanım sevkiyatı |
| 5 | Operasyonlarda oluşan atıklar | Yukarı akış | DÜŞÜK: WEEE, kağıt, ofis atığı |
| 6 | İş seyahati | Yukarı akış | YÜKSEK: dağıtılmış ekipler için uçuşlar, oteller, trenler |
| 7 | Çalışanların işe gidiş gelişleri | Yukarı akış | YÜKSEK: Ev-ofis seyahati, özellikle hibrit takımlar için |
| 8 | Yukarı yönde kiralanan varlıklar | Yukarı akış | MEDYA: kiralık ofisler (Kapsam 1/2 kapsamında yer almıyorsa) |
| 9 | Aşağı yönde taşıma ve dağıtım | Aşağı akış | DÜŞÜK: fiziksel ortamda yazılım dağıtımı (nadir) |
| 10 | Satılan ürünlerin işlenmesi | Aşağı akış | Yok: Salt yazılım için geçerli değildir |
| 11 | Satılan ürünlerin kullanımı | Aşağı akış | ÇOK YÜKSEK: SaaS kullanan müşterilerin tükettiği enerji |
| 12 | Satılan ürünlerin kullanım ömrü sonu işlemleri | Aşağı akış | DÜŞÜK: kullanım ömrünün sonundaki kullanıcı cihazları |
| 13 | Aşağı yöndeki kiralanan varlıklar | Aşağı akış | MEDYA: Müşterilere kiralanan donanım |
| 14 | Bayilikler | Aşağı akış | Uygulanamaz: Geçerli değil |
| 15 | Yatırımlar | Aşağı akış | YÜKSEK: kurumsal portföy, yeni kurulan şirketlerdeki sermaye yatırımları |
Bir SaaS veya yazılım geliştirme şirketi için en alakalı kategoriler genellikle şunlardır: Kedi. 1 (satın alınan mal ve hizmetler, genellikle en büyük kalemdir), Kedi. 6 (iş seyahati), Kedi. 7 (çalışanların işe gidiş gelişleri) e Kedi. 11 (satılan ürünlerin kullanımı). Orada çifte önemlilik istek CSRD, hem etki açısından hem de hangi kategorilerin öncelikli olduğunun belirlenmesini gerektirir şirket için çevresel ve finansal risk.
Yaygın Hata: Cat'i Atlamak. SaaS için 11
Birçok yazılım şirketi, geçerli olmadığı varsayımıyla Kategori 11'i ("Satılan ürünlerin kullanımı") hariç tutar. Gerçekte, müşterilerin sizinkini çalıştırmak için tükettiği her API çağrısı, her sorgu, her watt yazılım Kapsam 3 Kat. 11 emisyon sizin sorumluluğunuzdadır. Milyonların bulunduğu bir SaaS için kullanıcıların baskın kategorisi bu olabilir. Hesaplama yöntemi şunları kullanır: karbon yoğunluğu (SCI) yazılımı sağlanan fonksiyonel birimlerle çarpılır.
Veri Toplama Boru Hattının Mimarisi
Değer zinciri boyunca güvenilir veri toplamak her işletme için bir numaralı darboğazdır Kapsam 3 projesi. Boru hattı heterojen kaynakları yönetmelidir: manuel anketler, üçüncü taraf ESG platformları parçalar, tedarikçilerle doğrudan API'ler, e-posta yoluyla gönderilen CSV dosyaları, dahili ERP verileri. Aşağıdaki mimari bir model benimsemek Üç katmanlı ETL (Bronz/Gümüş/Altın) Lakehouse'dan esinlenilmiştir.
Boru Hattı Mimarisi Kapsam 3: Bronz / Gümüş / Altın
| Katmanlar | İçerik | Teknoloji | Kapsam |
|---|---|---|---|
| Bronz (Ham) | Tedarikçilerden alınan değişmez ham veriler | S3/GCS, Delta Gölü | Denetim izi, tekrar, gerçeğin kaynağı |
| Gümüş (Standartlaştırılmış) | Birim ve para birimine göre normalleştirilmiş veriler | dbt, Kıvılcım, Pandalar | Emisyon hesaplaması, emisyon faktörleriyle birleştirme |
| Altın (Raporlama) | Sera gazı kategorisine göre toplam emisyonlar | PostgreSQL, BigQuery | Gösterge tabloları, CSRD raporları, doğrulayıcılar |
Bronz katman önemlidir: alınan tüm veriler kaydedilir olduğu gibi zaman damgalı alımı, içeriğin SHA-256 karması ve kaynak meta verileri. Bu olasılığı garanti eder Emisyon faktörleri veya metodoloji değişirse, kayıp olmadan tüm boru hattını yeniden işlemek orijinal veriler.
# models/scope3_pipeline.py
# Struttura dati per la pipeline Scope 3
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional
import hashlib
import json
class DataSource(Enum):
SUPPLIER_QUESTIONNAIRE = "supplier_questionnaire"
CDP_API = "cdp_api"
ECOVADIS_API = "ecovadis_api"
ERP_EXPORT = "erp_export"
MANUAL_UPLOAD = "manual_upload"
CLIMATIQ_API = "climatiq_api"
class CalculationMethod(Enum):
ACTIVITY_BASED = "activity_based"
SPEND_BASED = "spend_based"
HYBRID = "hybrid"
SUPPLIER_SPECIFIC = "supplier_specific"
class DataQualityTier(Enum):
TIER_1 = "primary_data" # Dati primari dal supplier
TIER_2 = "secondary_sector" # Fattori settoriali
TIER_3 = "spend_estimated" # Stima basata su spesa
@dataclass
class RawSupplierData:
"""Layer Bronze: dato grezzo immutabile"""
supplier_id: str
source: DataSource
raw_payload: dict
received_at: datetime
content_hash: str = field(init=False)
def __post_init__(self):
payload_str = json.dumps(self.raw_payload, sort_keys=True)
self.content_hash = hashlib.sha256(
payload_str.encode()
).hexdigest()
@dataclass
class StandardizedActivity:
"""Layer Silver: attività normalizzata"""
activity_id: str
supplier_id: str
scope3_category: int # 1-15
activity_type: str # es. "freight_transport"
quantity: float
unit: str # es. "tonne.km"
reporting_period_start: datetime
reporting_period_end: datetime
source: DataSource
quality_tier: DataQualityTier
emission_factor_id: Optional[str] = None
uncertainty_pct: float = 0.0
raw_data_hash: str = "" # Ref al Bronze layer
@dataclass
class EmissionResult:
"""Layer Gold: emissione calcolata"""
result_id: str
activity_id: str
scope3_category: int
co2e_tonnes: float
calculation_method: CalculationMethod
emission_factor_source: str # es. "climatiq:IPCC_2021"
emission_factor_value: float
quality_tier: DataQualityTier
uncertainty_pct: float
calculated_at: datetime
pipeline_version: str
audit_hash: str = field(init=False)
def __post_init__(self):
audit_data = {
"result_id": self.result_id,
"activity_id": self.activity_id,
"co2e_tonnes": self.co2e_tonnes,
"emission_factor_source": self.emission_factor_source,
"calculated_at": self.calculated_at.isoformat(),
"pipeline_version": self.pipeline_version,
}
self.audit_hash = hashlib.sha256(
json.dumps(audit_data, sort_keys=True).encode()
).hexdigest()Tedarikçi Veri Entegrasyonu: CDP, EcoVadis ve Direct API
Tedarikçilerden veri toplanması, kalite ve kalite seviyelerine sahip birden fazla kanal aracılığıyla gerçekleşir. çok farklı bir otomasyon. Karbon Saydamlık Projesi (CDP) gelen verileri toplar 24.000'den fazla şirket ve doğrulanmış raporlara erişim için bir API sunuyor. EcoVadis 2025 yılında 48.000'den fazla sera gazı muhabirinin bir şekilde veri paylaşmasıyla Karbon Veri Ağı'nı başlattı standartlaştırılmış. Son olarak, birçok büyük satıcı doğrudan paylaşım için özel API'leri kullanıma sunuyor ayak izlerinden.
# collectors/supplier_collector.py
# Integrazione con fonti dati supplier
import httpx
import asyncio
from typing import AsyncGenerator
from datetime import datetime
from models.scope3_pipeline import RawSupplierData, DataSource
class ClimatiqEmissionFactors:
"""Client per Climatiq API - emission factors database"""
BASE_URL = "https://beta3.api.climatiq.io"
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
async def get_emission_factor(
self,
activity_id: str,
year: int = 2024,
region: str = "IT"
) -> dict:
"""Recupera fattore di emissione per attività specifica"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.BASE_URL}/estimate",
headers=self.headers,
json={
"emission_factor": {
"activity_id": activity_id,
"data_version": "^21",
"year": year,
"region": region
},
"parameters": {
"money": 1.0,
"money_unit": "eur"
}
}
)
response.raise_for_status()
return response.json()
async def batch_estimate(
self,
activities: list[dict]
) -> list[dict]:
"""Stima batch per multiple attività - ottimizza le API call"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.BASE_URL}/batch",
headers=self.headers,
json={"batch": activities},
timeout=30.0
)
response.raise_for_status()
return response.json().get("results", [])
class EcoVadisCollector:
"""Raccoglie dati Scope 3 dalla piattaforma EcoVadis"""
def __init__(self, api_key: str, base_url: str):
self.api_key = api_key
self.base_url = base_url
async def fetch_supplier_carbon_data(
self,
supplier_ecovadis_id: str,
reporting_year: int
) -> RawSupplierData:
"""Recupera dati carbonio per un supplier dalla Carbon Data Network"""
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.base_url}/v1/suppliers/{supplier_ecovadis_id}/carbon",
headers={"X-API-Key": self.api_key},
params={"year": reporting_year},
timeout=15.0
)
if response.status_code == 404:
# Supplier non ha condiviso dati primari
return self._create_no_data_record(supplier_ecovadis_id)
response.raise_for_status()
payload = response.json()
return RawSupplierData(
supplier_id=supplier_ecovadis_id,
source=DataSource.ECOVADIS_API,
raw_payload=payload,
received_at=datetime.utcnow()
)
def _create_no_data_record(self, supplier_id: str) -> RawSupplierData:
return RawSupplierData(
supplier_id=supplier_id,
source=DataSource.ECOVADIS_API,
raw_payload={"status": "no_data", "supplier_id": supplier_id},
received_at=datetime.utcnow()
)
class CDPCollector:
"""Raccoglie dati da CDP (Carbon Disclosure Project)"""
CDP_API_URL = "https://api.cdp.net/v1"
def __init__(self, api_token: str):
self.api_token = api_token
async def search_supplier(
self,
company_name: str,
year: int = 2024
) -> RawSupplierData | None:
"""Cerca un supplier nel database CDP e recupera dati GHG"""
async with httpx.AsyncClient() as client:
# Ricerca azienda
search_resp = await client.get(
f"{self.CDP_API_URL}/companies/search",
headers={"Authorization": f"Bearer {self.api_token}"},
params={"q": company_name, "year": year}
)
if not search_resp.json().get("results"):
return None
company_id = search_resp.json()["results"][0]["id"]
# Recupera dati GHG disclosure
ghg_resp = await client.get(
f"{self.CDP_API_URL}/companies/{company_id}/ghg-emissions",
headers={"Authorization": f"Bearer {self.api_token}"},
params={"year": year}
)
if ghg_resp.status_code != 200:
return None
return RawSupplierData(
supplier_id=company_name,
source=DataSource.CDP_API,
raw_payload=ghg_resp.json(),
received_at=datetime.utcnow()
)
class BronzeLayerStorage:
"""Salvataggio immutabile nel layer Bronze"""
def __init__(self, storage_client, bucket: str):
self.storage = storage_client
self.bucket = bucket
async def store(self, raw_data: RawSupplierData) -> str:
"""Salva dato grezzo con path deterministico basato su hash"""
path = (
f"scope3/bronze/"
f"{raw_data.received_at.year}/"
f"{raw_data.received_at.month:02d}/"
f"{raw_data.supplier_id}/"
f"{raw_data.content_hash}.json"
)
await self.storage.upload_json(
bucket=self.bucket,
path=path,
data={
"supplier_id": raw_data.supplier_id,
"source": raw_data.source.value,
"received_at": raw_data.received_at.isoformat(),
"content_hash": raw_data.content_hash,
"payload": raw_data.raw_payload
}
)
return pathFaaliyet Tabanlı vs Harcama Tabanlı: Doğru Yöntemi Seçmek
Sera Gazı Protokolü Kapsam 3 için dört hesaplama yöntemi tanımlar; bunlar pratikte evet iki temel yaklaşıma indirgeyin: aktivite bazlı e harcamaya dayalı. Seçim, verilerin kullanılabilirliğine ve kategorinin önemliliğine bağlıdır. ve tedarikçiyle ilişkinin olgunluğu.
Metodolojik Karşılaştırma: Etkinliğe Dayalı ve Harcamaya Dayalı
| Boyut | Etkinlik Tabanlı | Harcamaya Dayalı |
|---|---|---|
| Formül | Miktar × Emisyon Faktörü (birim/kg CO2e) | Harcama (EUR) × EEIO Faktörü (kg CO2e/EUR) |
| Kesinlik | Yüksek (birincil verilerle ±%5-15) | Düşük-Orta (±50-100%) |
| Talep edilen veriler | Fiziksel büyüklükler (kg, km, kWh, t) | Yalnızca muhasebe faturaları (EUR, USD) |
| Kaynak EF | Climatiq, IPCC, DEFRA, ecoinvent | USEEIO, EXIOBASE, WIOD |
| Ne zaman kullanılmalı? | Malzeme kategorileri, büyük tedarikçiler | Başlangıç, küçük tedarikçiler, Cat. <%1 |
| Toplama çabası | Yüksek: tedarikçi işbirliği gerektirir | Düşük: Veriler zaten ERP/SAP'ta |
| CSRD kabul edilebilirliği | Malzeme kategorileri için favori | İlk vekil olarak kabul edildi |
En uygun strateji bir yaklaşımdır ilerici hibrit: başlayalım değer zincirinin tamamında hızlı bir temele sahip olmak için harcamaya dayalıdır, daha sonra aşamalı olarak taşınır belirlenen malzeme kategorileri için faaliyet bazlı olmaya doğru. Sera Gazı Protokolü üç seviyeyi tanımlar Bu ilerlemeye tam olarak karşılık gelen veri kalitesi (Kademe 1, 2, 3).
# calculators/emission_calculator.py
# Calcolo emissioni activity-based e spend-based
from dataclasses import dataclass
from typing import Optional
import math
# ============================================================
# EMISSION FACTORS DATABASE (simplified)
# In produzione: usa Climatiq API o database ecoinvent
# ============================================================
EMISSION_FACTORS: dict[str, dict] = {
# Cat. 1: Purchased goods & services
"cloud_compute_kwh": {
"value": 0.233, # kg CO2e/kWh (IT grid mix 2024)
"unit": "kWh",
"source": "IEA 2024",
"uncertainty_pct": 10.0
},
"hardware_laptop": {
"value": 350.0, # kg CO2e/unit (embodied carbon)
"unit": "unit",
"source": "Dell 2024 PCF",
"uncertainty_pct": 20.0
},
# Cat. 6: Business travel
"flight_economy_short": {
"value": 0.255, # kg CO2e/passenger.km
"unit": "passenger.km",
"source": "DEFRA 2024",
"uncertainty_pct": 15.0
},
"flight_economy_long": {
"value": 0.195,
"unit": "passenger.km",
"source": "DEFRA 2024",
"uncertainty_pct": 15.0
},
# Cat. 7: Employee commuting
"car_average": {
"value": 0.170, # kg CO2e/km
"unit": "km",
"source": "DEFRA 2024",
"uncertainty_pct": 12.0
},
"public_transport_it": {
"value": 0.048,
"unit": "passenger.km",
"source": "Ispra 2024",
"uncertainty_pct": 18.0
},
}
# EEIO Spend-based factors (EXIOBASE 3.8)
# kg CO2e per EUR di spesa per categoria merceologica
EEIO_FACTORS: dict[str, float] = {
"it_services": 0.312, # IT e telecomunicazioni
"professional_services": 0.198, # Consulenza, legale, etc.
"office_supplies": 0.445,
"cloud_hosting": 0.287,
"marketing": 0.231,
"utilities": 0.892,
"hr_services": 0.167,
"travel_accommodation": 0.521,
}
def calculate_activity_based(
activity_type: str,
quantity: float,
custom_ef: Optional[float] = None
) -> tuple[float, float]:
"""
Calcola emissioni con metodo activity-based.
Returns:
(co2e_kg, uncertainty_pct)
"""
if custom_ef is not None:
return quantity * custom_ef, 30.0 # alta incertezza EF custom
ef_data = EMISSION_FACTORS.get(activity_type)
if not ef_data:
raise ValueError(f"Emission factor non trovato: {activity_type}")
co2e_kg = quantity * ef_data["value"]
uncertainty = ef_data["uncertainty_pct"]
return co2e_kg, uncertainty
def calculate_spend_based(
spend_eur: float,
procurement_category: str,
inflation_correction: float = 1.0
) -> tuple[float, float]:
"""
Calcola emissioni con metodo spend-based (EEIO).
Args:
spend_eur: importo in EUR
procurement_category: categoria merceologica EEIO
inflation_correction: fattore per correggere inflazione vs anno base EEIO
Returns:
(co2e_kg, uncertainty_pct)
"""
eeio_factor = EEIO_FACTORS.get(procurement_category)
if not eeio_factor:
raise ValueError(f"EEIO factor non trovato: {procurement_category}")
# Corregge per inflazione (EEIO factors spesso in EUR 2015)
adjusted_spend = spend_eur / inflation_correction
co2e_kg = adjusted_spend * eeio_factor
# Lo spend-based ha incertezza intrinsecamente alta
uncertainty = 75.0
return co2e_kg, uncertainty
def propagate_uncertainty(
values: list[float],
uncertainties_pct: list[float]
) -> float:
"""
Propagazione incertezza quadratica (somma in quadratura).
Valida quando le incertezze sono indipendenti.
Returns:
uncertainty_pct sul totale
"""
weighted_variance_sum = sum(
(v * u/100) ** 2
for v, u in zip(values, uncertainties_pct)
)
total = sum(values)
if total == 0:
return 0.0
combined_std = math.sqrt(weighted_variance_sum)
return (combined_std / total) * 100
def calculate_category_total(
activities: list[dict]
) -> dict:
"""
Calcola totale categoria Scope 3 con propagazione incertezza.
activities: lista di {method, value_kg, uncertainty_pct}
"""
if not activities:
return {"total_co2e_kg": 0.0, "uncertainty_pct": 0.0}
values = [a["value_kg"] for a in activities]
uncertainties = [a["uncertainty_pct"] for a in activities]
total_co2e = sum(values)
combined_uncertainty = propagate_uncertainty(values, uncertainties)
# Qualità aggregata: peggiore del gruppo determina il tier
quality_tiers = [a.get("quality_tier", "TIER_3") for a in activities]
dominant_tier = min(quality_tiers) # TIER_1 < TIER_2 < TIER_3 lexicograficamente
return {
"total_co2e_kg": total_co2e,
"total_co2e_tonnes": total_co2e / 1000,
"uncertainty_pct": combined_uncertainty,
"uncertainty_kg": total_co2e * combined_uncertainty / 100,
"dominant_quality_tier": dominant_tier,
"activity_count": len(activities)
}Boru Hattı Hava Akışı: Otomatik Kapsam 3 Hesaplaması için DAG
Kapsam 3 işlem hattı orkestrasyonu, iyi yapılandırılmış bir DAG gerektirir. tam yaşam döngüsü: veri toplama, standardizasyon, emisyon hesaplaması, kalite kontrolü ve Altın katmanında yayınlama. DAG olmalıdır güçsüz (yürütülebilir yan etki olmaksızın birkaç kez) e sıfırlanabilir durumunda kısmi başarısızlık.
# dags/scope3_pipeline_dag.py
# Apache Airflow DAG per pipeline emissioni Scope 3
from datetime import datetime, timedelta
from airflow import DAG
from airflow.decorators import task, task_group
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.models import Variable
import json
import logging
logger = logging.getLogger(__name__)
# ============================================================
# CONFIGURAZIONE DAG
# ============================================================
SCOPE3_DAG_CONFIG = {
"reporting_year": 2024,
"companies": [
{"id": "S001", "name": "AWS", "tier": "TIER_1", "source": "ecovadis"},
{"id": "S002", "name": "Microsoft Azure", "tier": "TIER_1", "source": "cdp"},
{"id": "S003", "name": "Supplier_XYZ", "tier": "TIER_2", "source": "questionnaire"},
# ... altri supplier
],
"categories_enabled": [1, 2, 3, 6, 7, 11, 15],
"quality_threshold_pct": 80.0,
"alert_email": "esg-team@company.com"
}
default_args = {
"owner": "esg-team",
"depends_on_past": False,
"email_on_failure": True,
"email": [SCOPE3_DAG_CONFIG["alert_email"]],
"retries": 3,
"retry_delay": timedelta(minutes=5),
}
with DAG(
dag_id="scope3_emissions_pipeline",
default_args=default_args,
description="Pipeline calcolo emissioni Scope 3 value chain",
schedule_interval="@quarterly", # Esecuzione trimestrale
start_date=datetime(2024, 1, 1),
catchup=False,
tags=["emissions", "scope3", "esg", "ghg-protocol"],
max_active_runs=1, # Serializza: mai due calcoli in parallelo
) as dag:
# ============================================================
# FASE 1: RACCOLTA DATI SUPPLIER (in parallelo per supplier)
# ============================================================
@task_group(group_id="data_collection")
def collect_supplier_data():
@task(task_id="fetch_ecovadis_suppliers")
def fetch_ecovadis() -> list[dict]:
"""Raccoglie dati da EcoVadis Carbon Data Network"""
from collectors.supplier_collector import EcoVadisCollector
import asyncio
api_key = Variable.get("ECOVADIS_API_KEY", deserialize_json=False)
collector = EcoVadisCollector(api_key, "https://api.ecovadis.com")
suppliers_ecovadis = [
s for s in SCOPE3_DAG_CONFIG["companies"]
if s["source"] == "ecovadis"
]
results = []
for supplier in suppliers_ecovadis:
raw = asyncio.run(
collector.fetch_supplier_carbon_data(
supplier["id"],
SCOPE3_DAG_CONFIG["reporting_year"]
)
)
results.append({
"supplier_id": raw.supplier_id,
"content_hash": raw.content_hash,
"status": "fetched",
"has_data": raw.raw_payload.get("status") != "no_data"
})
logger.info(f"EcoVadis - Supplier {supplier['id']}: fetched")
return results
@task(task_id="fetch_cdp_suppliers")
def fetch_cdp() -> list[dict]:
"""Raccoglie dati verificati da CDP"""
from collectors.supplier_collector import CDPCollector
import asyncio
api_token = Variable.get("CDP_API_TOKEN")
collector = CDPCollector(api_token)
suppliers_cdp = [
s for s in SCOPE3_DAG_CONFIG["companies"]
if s["source"] == "cdp"
]
results = []
for supplier in suppliers_cdp:
raw = asyncio.run(
collector.search_supplier(
supplier["name"],
SCOPE3_DAG_CONFIG["reporting_year"]
)
)
if raw:
results.append({
"supplier_id": supplier["id"],
"content_hash": raw.content_hash,
"status": "fetched",
"has_data": True
})
else:
results.append({
"supplier_id": supplier["id"],
"status": "not_found",
"has_data": False
})
return results
@task(task_id="process_manual_questionnaires")
def process_questionnaires() -> list[dict]:
"""Processa questionari manuali caricati in S3"""
# In produzione: legge da bucket S3 o SharePoint
# Qui restituiamo dati di esempio
return [{
"supplier_id": "S003",
"status": "processed",
"has_data": True,
"scope3_cat1_tco2e": 45.2,
"scope3_cat6_tco2e": 12.8
}]
ev = fetch_ecovadis()
cdp = fetch_cdp()
q = process_questionnaires()
return [ev, cdp, q]
# ============================================================
# FASE 2: STANDARDIZZAZIONE E CALCOLO EMISSIONI
# ============================================================
@task(task_id="standardize_activities")
def standardize_activities(collection_results: list) -> list[dict]:
"""Normalizza tutti i dati in unità fisiche standard"""
from normalizers.activity_normalizer import ActivityNormalizer
normalizer = ActivityNormalizer()
standardized = []
for batch in collection_results:
for result in batch:
if result.get("has_data"):
activities = normalizer.normalize(result)
standardized.extend(activities)
logger.info(f"Standardizzate {len(standardized)} attività")
return standardized
@task(task_id="calculate_emissions")
def calculate_emissions(activities: list[dict]) -> list[dict]:
"""Calcola emissioni CO2e per ogni attività standardizzata"""
from calculators.emission_calculator import (
calculate_activity_based,
calculate_spend_based
)
from models.scope3_pipeline import CalculationMethod
results = []
for activity in activities:
if activity["method"] == "activity_based":
co2e_kg, uncertainty = calculate_activity_based(
activity["activity_type"],
activity["quantity"]
)
method = CalculationMethod.ACTIVITY_BASED
else:
co2e_kg, uncertainty = calculate_spend_based(
activity["spend_eur"],
activity["procurement_category"]
)
method = CalculationMethod.SPEND_BASED
results.append({
**activity,
"co2e_kg": co2e_kg,
"co2e_tonnes": co2e_kg / 1000,
"uncertainty_pct": uncertainty,
"calculation_method": method.value,
"calculated_at": datetime.utcnow().isoformat()
})
return results
# ============================================================
# FASE 3: DATA QUALITY CHECK
# ============================================================
@task(task_id="data_quality_check")
def data_quality_check(results: list[dict]) -> dict:
"""Verifica qualità dati e genera score per categoria"""
from quality.data_quality_scorer import DataQualityScorer
scorer = DataQualityScorer()
quality_report = scorer.score_results(results)
if quality_report["overall_score"] < SCOPE3_DAG_CONFIG["quality_threshold_pct"]:
logger.warning(
f"Quality score sotto soglia: {quality_report['overall_score']}%"
)
return quality_report
# ============================================================
# FASE 4: AUDIT TRAIL E PUBBLICAZIONE GOLD LAYER
# ============================================================
@task(task_id="create_audit_trail")
def create_audit_trail(
results: list[dict],
quality_report: dict
) -> str:
"""Crea audit trail immutabile con hash chain"""
from audit.hash_chain import HashChain
chain = HashChain()
chain_id = chain.create_chain(
calculation_results=results,
quality_report=quality_report,
pipeline_version="2.1.0",
methodology="GHG_Protocol_Scope3_2011",
reporting_standard="CSRD_ESRS_E1"
)
logger.info(f"Audit trail creato: {chain_id}")
return chain_id
@task(task_id="publish_gold_layer")
def publish_gold_layer(
results: list[dict],
audit_chain_id: str
) -> None:
"""Pubblica dati aggregati nel Gold layer (PostgreSQL)"""
hook = PostgresHook(postgres_conn_id="emissions_db")
for result in results:
hook.run(
"""
INSERT INTO scope3_emissions_gold (
supplier_id, scope3_category, co2e_tonnes,
calculation_method, uncertainty_pct,
quality_tier, audit_chain_id,
reporting_year, published_at
) VALUES (
%(supplier_id)s, %(scope3_category)s, %(co2e_tonnes)s,
%(calculation_method)s, %(uncertainty_pct)s,
%(quality_tier)s, %(audit_chain_id)s,
%(reporting_year)s, NOW()
)
ON CONFLICT (supplier_id, scope3_category, reporting_year)
DO UPDATE SET
co2e_tonnes = EXCLUDED.co2e_tonnes,
updated_at = NOW()
""",
parameters={
**result,
"audit_chain_id": audit_chain_id,
"reporting_year": SCOPE3_DAG_CONFIG["reporting_year"]
}
)
logger.info(f"Pubblicati {len(results)} record nel Gold layer")
# ============================================================
# WIRING DEL DAG
# ============================================================
collection_results = collect_supplier_data()
standardized = standardize_activities(collection_results)
emission_results = calculate_emissions(standardized)
quality = data_quality_check(emission_results)
chain_id = create_audit_trail(emission_results, quality)
publish_gold_layer(emission_results, chain_id)Veri Kalitesi Puanlaması ve Belirsizliğin Yayılması
Sera Gazı Protokolü Kapsam 3 Standardı bu kediyi açıkça tanır. 1-15 değil hiçbir zaman kesin olarak bilinemez. Kalite raporlaması bir tahmin içermelidir arasındaniceliksel belirsizlik her kategoriyle ilişkilendirilir. IPCC resmileşti İyi Uygulama Kılavuzunda belirsizlik yayılma yöntemi.
# quality/data_quality_scorer.py
# Scoring qualità dati Scope 3
from dataclasses import dataclass
from enum import Enum
from typing import Optional
import math
from datetime import datetime, timedelta
class QualityDimension(Enum):
COMPLETENESS = "completeness"
ACCURACY = "accuracy"
TIMELINESS = "timeliness"
VERIFICATION = "verification"
GRANULARITY = "granularity"
@dataclass
class QualityScore:
dimension: QualityDimension
score: float # 0-100
weight: float # peso nel calcolo aggregato
notes: str = ""
def score_supplier_data_quality(
supplier: dict,
reference_date: datetime = None
) -> dict[str, float]:
"""
Calcola score qualità multi-dimensionale per i dati di un supplier.
Basato su GHG Protocol Data Quality Guidance.
"""
if reference_date is None:
reference_date = datetime.utcnow()
scores = []
# 1. COMPLETENESS: quante delle categorie richieste sono presenti?
required_fields = [
"scope3_cat1_tco2e", "scope3_cat6_tco2e", "scope3_cat7_tco2e"
]
present = sum(1 for f in required_fields if supplier.get(f) is not None)
completeness_score = (present / len(required_fields)) * 100
scores.append(QualityScore(
dimension=QualityDimension.COMPLETENESS,
score=completeness_score,
weight=0.30
))
# 2. TIMELINESS: quanto sono recenti i dati?
data_year = supplier.get("reporting_year", 2020)
current_year = reference_date.year
age_years = current_year - data_year
if age_years <= 1:
timeliness_score = 100.0
elif age_years == 2:
timeliness_score = 75.0
elif age_years == 3:
timeliness_score = 50.0
else:
timeliness_score = 20.0
scores.append(QualityScore(
dimension=QualityDimension.TIMELINESS,
score=timeliness_score,
weight=0.20
))
# 3. VERIFICATION: i dati sono stati verificati da terze parti?
verification_level = supplier.get("verification", "none")
verification_score = {
"independent_assured": 100.0, # GHG verificato da auditor indipendente
"limited_assurance": 80.0, # Limited assurance
"internal_reviewed": 60.0, # Solo review interna
"supplier_declared": 40.0, # Auto-dichiarazione
"estimated": 20.0, # Stima spend-based
"none": 0.0
}.get(verification_level, 20.0)
scores.append(QualityScore(
dimension=QualityDimension.VERIFICATION,
score=verification_score,
weight=0.30
))
# 4. GRANULARITY: attività-specifico o aggregato?
data_type = supplier.get("data_type", "aggregated")
granularity_score = {
"site_specific": 100.0, # Dati per sito produttivo
"product_specific": 90.0, # PCF per prodotto/servizio
"supplier_specific": 70.0, # Dato totale supplier
"sector_average": 40.0, # Media settoriale
"aggregated": 20.0
}.get(data_type, 20.0)
scores.append(QualityScore(
dimension=QualityDimension.GRANULARITY,
score=granularity_score,
weight=0.20
))
# Calcolo score aggregato ponderato
overall = sum(s.score * s.weight for s in scores)
# Mappa score a Tier GHG Protocol
if overall >= 80:
tier = "TIER_1"
uncertainty_band_pct = 15.0
elif overall >= 50:
tier = "TIER_2"
uncertainty_band_pct = 40.0
else:
tier = "TIER_3"
uncertainty_band_pct = 75.0
return {
"overall_score": round(overall, 1),
"tier": tier,
"uncertainty_band_pct": uncertainty_band_pct,
"dimension_scores": {
s.dimension.value: round(s.score, 1)
for s in scores
}
}
def monte_carlo_uncertainty(
base_estimate_tco2e: float,
uncertainty_pct: float,
n_simulations: int = 10_000
) -> dict:
"""
Stima intervallo di confidenza con simulazione Monte Carlo.
Per reporting CSRD si raccomanda almeno 1.000 simulazioni.
"""
import random
# Distribuzione log-normale (emissioni non possono essere negative)
sigma = math.log(1 + (uncertainty_pct / 100) ** 2) ** 0.5
mu = math.log(base_estimate_tco2e) - sigma ** 2 / 2
simulated = [
math.exp(random.gauss(mu, sigma))
for _ in range(n_simulations)
]
simulated_sorted = sorted(simulated)
p05 = simulated_sorted[int(n_simulations * 0.05)]
p50 = simulated_sorted[int(n_simulations * 0.50)]
p95 = simulated_sorted[int(n_simulations * 0.95)]
return {
"base_estimate_tco2e": base_estimate_tco2e,
"p05_tco2e": round(p05, 2),
"p50_tco2e": round(p50, 2),
"p95_tco2e": round(p95, 2),
"confidence_interval_90pct": {
"lower": round(p05, 2),
"upper": round(p95, 2)
},
"coefficient_of_variation": round(
(p95 - p05) / (2 * p50) * 100, 1
)
}Hash Chain ile Değişmez Denetim Yolu
La uçtan uca izlenebilirlik için en kritik gereksinimlerden biridir. doğrulanabilir Kapsam 3 raporlaması. Dış denetçiler rapordaki her sayıyı geriye doğru izleyebilmelidir Verinin birincil kaynağına kadar tüm dönüşüm adımlarından geçerek son halini alır. A hash zincirleri Blockchain teknolojisinden ilham aldı (fakat karmaşıklık olmadan) dağıtılmış) denetim izinin değişmezliğini garanti eder.
# audit/hash_chain.py
# Audit trail immutabile per emissioni Scope 3
import hashlib
import json
import uuid
from datetime import datetime
from typing import Optional
import logging
logger = logging.getLogger(__name__)
class AuditRecord:
"""Singolo record nell'audit chain"""
def __init__(
self,
record_type: str,
payload: dict,
previous_hash: str,
chain_id: str,
sequence: int
):
self.record_id = str(uuid.uuid4())
self.record_type = record_type
self.payload = payload
self.previous_hash = previous_hash
self.chain_id = chain_id
self.sequence = sequence
self.created_at = datetime.utcnow().isoformat()
self.record_hash = self._compute_hash()
def _compute_hash(self) -> str:
"""SHA-256 hash di tutti i campi del record (eccetto il hash stesso)"""
data = {
"record_id": self.record_id,
"record_type": self.record_type,
"chain_id": self.chain_id,
"sequence": self.sequence,
"previous_hash": self.previous_hash,
"created_at": self.created_at,
"payload_hash": hashlib.sha256(
json.dumps(self.payload, sort_keys=True, default=str).encode()
).hexdigest()
}
return hashlib.sha256(
json.dumps(data, sort_keys=True).encode()
).hexdigest()
def to_dict(self) -> dict:
return {
"record_id": self.record_id,
"record_type": self.record_type,
"chain_id": self.chain_id,
"sequence": self.sequence,
"previous_hash": self.previous_hash,
"record_hash": self.record_hash,
"created_at": self.created_at,
"payload": self.payload
}
class HashChain:
"""
Hash chain per audit trail immutabile emissioni Scope 3.
Ogni record contiene l'hash del record precedente,
rendendo impossibile modificare un record senza invalidare
tutti i record successivi.
"""
GENESIS_HASH = "0" * 64 # Hash del primo record della chain
def __init__(self, db_client=None):
self.db = db_client
self.records: list[AuditRecord] = []
def create_chain(
self,
calculation_results: list[dict],
quality_report: dict,
pipeline_version: str,
methodology: str,
reporting_standard: str
) -> str:
"""Crea una nuova chain per un calcolo Scope 3 completo"""
chain_id = str(uuid.uuid4())
previous_hash = self.GENESIS_HASH
# Record 1: Metadati della pipeline
pipeline_record = AuditRecord(
record_type="PIPELINE_METADATA",
payload={
"version": pipeline_version,
"methodology": methodology,
"reporting_standard": reporting_standard,
"calculation_timestamp": datetime.utcnow().isoformat(),
"total_activities": len(calculation_results)
},
previous_hash=previous_hash,
chain_id=chain_id,
sequence=0
)
self.records.append(pipeline_record)
previous_hash = pipeline_record.record_hash
# Record 2: Quality report
quality_record = AuditRecord(
record_type="QUALITY_ASSESSMENT",
payload=quality_report,
previous_hash=previous_hash,
chain_id=chain_id,
sequence=1
)
self.records.append(quality_record)
previous_hash = quality_record.record_hash
# Record 3..N: Singoli risultati di emissione
for i, result in enumerate(calculation_results):
emission_record = AuditRecord(
record_type="EMISSION_CALCULATION",
payload={
"supplier_id": result.get("supplier_id"),
"scope3_category": result.get("scope3_category"),
"co2e_tonnes": result.get("co2e_tonnes"),
"calculation_method": result.get("calculation_method"),
"emission_factor_source": result.get("emission_factor_source"),
"uncertainty_pct": result.get("uncertainty_pct"),
"quality_tier": result.get("quality_tier")
},
previous_hash=previous_hash,
chain_id=chain_id,
sequence=2 + i
)
self.records.append(emission_record)
previous_hash = emission_record.record_hash
# Persist su DB (o storage immutabile)
if self.db:
self._persist_chain(chain_id)
logger.info(
f"Chain {chain_id} creata con {len(self.records)} record. "
f"Final hash: {previous_hash[:16]}..."
)
return chain_id
def verify_chain_integrity(self, chain_id: str) -> bool:
"""
Verifica che nessun record sia stato alterato.
Percorre la chain ricomputando ogni hash.
"""
records = self._load_chain(chain_id)
if not records:
return False
expected_previous = self.GENESIS_HASH
for record_dict in records:
# Ricomputa hash
record = AuditRecord(
record_type=record_dict["record_type"],
payload=record_dict["payload"],
previous_hash=record_dict["previous_hash"],
chain_id=record_dict["chain_id"],
sequence=record_dict["sequence"]
)
if record_dict["previous_hash"] != expected_previous:
logger.error(
f"Chain corrotta al record {record_dict['sequence']}: "
f"previous_hash non corrisponde"
)
return False
expected_previous = record.record_hash
return True
def _persist_chain(self, chain_id: str) -> None:
"""Salva tutti i record della chain nel DB"""
for record in self.records:
self.db.insert("scope3_audit_chain", record.to_dict())
def _load_chain(self, chain_id: str) -> list[dict]:
"""Carica i record della chain dal DB in ordine di sequenza"""
if not self.db:
return [r.to_dict() for r in self.records]
return self.db.query(
"SELECT * FROM scope3_audit_chain WHERE chain_id = %s ORDER BY sequence",
[chain_id]
)Görselleştirme: Sankey Diyagramı ve Isı Haritası Kategorileri
İyi oluşturulmuş bir Kapsam 3 işlem hattı aynı zamanda verileri işleyen görselleştirmeler de üretmelidir. Teknik ve teknik olmayan paydaşlar için anlaşılabilir. Sankey diyagramı öyle Değer zinciri boyunca emisyon akışlarını göstermek için ideal bir araç. ısı haritası en önemli kategorileri hızlı bir şekilde tanımlamanıza olanak tanır ve daha düşük veri kalitesine sahip olanlar.
# visualizations/scope3_charts.py
# Generazione Sankey diagram e heatmap Scope 3
import plotly.graph_objects as go
import plotly.express as px
import pandas as pd
from typing import Optional
def create_scope3_sankey(
emission_data: list[dict],
title: str = "Scope 3 Value Chain Emissions"
) -> go.Figure:
"""
Crea Sankey diagram per visualizzare flussi emissioni Scope 3.
Struttura: Supplier -> Categoria S3 -> Totale Scope 3
"""
# Raccoglie nodi unici
suppliers = list(set(d["supplier_id"] for d in emission_data))
categories = list(set(f"Cat. {d['scope3_category']}" for d in emission_data))
all_nodes = suppliers + categories + ["Scope 3 Total"]
node_index = {node: i for i, node in enumerate(all_nodes)}
# Costruisce link source->target->value
source_indices = []
target_indices = []
values = []
link_labels = []
for record in emission_data:
supplier = record["supplier_id"]
category = f"Cat. {record['scope3_category']}"
tco2e = record["co2e_tonnes"]
# Supplier -> Categoria
source_indices.append(node_index[supplier])
target_indices.append(node_index[category])
values.append(tco2e)
link_labels.append(f"{tco2e:.1f} tCO2e")
# Categoria -> Total
for cat in categories:
cat_total = sum(
d["co2e_tonnes"]
for d in emission_data
if f"Cat. {d['scope3_category']}" == cat
)
source_indices.append(node_index[cat])
target_indices.append(node_index["Scope 3 Total"])
values.append(cat_total)
link_labels.append(f"{cat_total:.1f} tCO2e")
# Colori nodi
node_colors = (
["#2196F3"] * len(suppliers) + # Blu per supplier
["#FF9800"] * len(categories) + # Arancione per categorie
["#4CAF50"] # Verde per totale
)
fig = go.Figure(go.Sankey(
arrangement="snap",
node=dict(
pad=20,
thickness=20,
line=dict(color="white", width=0.5),
label=all_nodes,
color=node_colors,
hovertemplate="{label}
tCO2e: {value:.1f}<extra></extra>"
),
link=dict(
source=source_indices,
target=target_indices,
value=values,
label=link_labels,
color="rgba(100,100,100,0.3)"
)
))
fig.update_layout(
title_text=title,
font_size=12,
height=600,
paper_bgcolor="rgba(0,0,0,0)",
plot_bgcolor="rgba(0,0,0,0)"
)
return fig
def create_category_heatmap(
emission_data: list[dict]
) -> go.Figure:
"""
Heatmap: asse X = categoria Scope 3, asse Y = qualità dato.
Colore = tCO2e. Aiuta a prioritizzare effort raccolta dati.
"""
df = pd.DataFrame(emission_data)
# Aggrega per categoria e tier qualità
pivot = df.pivot_table(
values="co2e_tonnes",
index="quality_tier",
columns="scope3_category",
aggfunc="sum",
fill_value=0
)
# Ordina tier (TIER_1 migliore in alto)
tier_order = ["TIER_1", "TIER_2", "TIER_3"]
pivot = pivot.reindex(
[t for t in tier_order if t in pivot.index]
)
fig = go.Figure(go.Heatmap(
z=pivot.values,
x=[f"Cat. {c}" for c in pivot.columns],
y=list(pivot.index),
colorscale="RdYlGn_r", # Rosso = alta emissione (critico)
text=pivot.values.round(1),
texttemplate="%{text} t",
textfont={"size": 11},
hovertemplate="Categoria: %{x}
Tier: %{y}
%{z:.1f} tCO2e<extra></extra>",
colorbar=dict(title="tCO2e")
))
fig.update_layout(
title="Heatmap Scope 3: Emissioni per Categoria e Qualità Dato",
xaxis_title="Categoria GHG Protocol",
yaxis_title="Tier Qualità Dato",
height=350,
margin=dict(l=80, r=20, t=60, b=60)
)
return fig
def generate_scope3_dashboard_html(
emission_data: list[dict],
output_path: str
) -> None:
"""Genera report HTML standalone con tutti i grafici"""
sankey = create_scope3_sankey(emission_data)
heatmap = create_category_heatmap(emission_data)
total_tco2e = sum(d["co2e_tonnes"] for d in emission_data)
by_category = {}
for d in emission_data:
cat = d["scope3_category"]
by_category[cat] = by_category.get(cat, 0) + d["co2e_tonnes"]
top_category = max(by_category, key=by_category.get)
html_content = f"""
<!DOCTYPE html>
<html>
<head>
<title>Scope 3 Emissions Report</title>
<meta charset="utf-8">
</head>
<body>
<h1>Scope 3 Value Chain Emissions Report</h1>
<p>Totale: <strong>{total_tco2e:.1f} tCO2e</strong></p>
<p>Categoria più materiale: Cat. {top_category}
({by_category[top_category]:.1f} tCO2e)</p>
{sankey.to_html(full_html=False)}
{heatmap.to_html(full_html=False)}
</body>
</html>
"""
with open(output_path, "w") as f:
f.write(html_content)Raporlama Kapsamı 3 için CSRD/ESRS E1 Gereksinimleri
La Kurumsal Sürdürülebilirlik Raporlaması Direktifi (CSRD) ve ilgili standart ESRS E1 (İklim Değişikliği) Kapsam 3 raporlamasını gönüllüden raporlamaya dönüştürüyorlar Binlerce Avrupalı şirket için zorunludur. Uygulama takvimi kademelidir ve zaten devam ediyor.
CSRD Kapsam 3 Zaman Çizelgesi Gerekli
| FY | Rapor ver içeri gireceğim | Konular | Notlar |
|---|---|---|---|
| 2024 | 2025 başı | Halihazırda NFRD'ye tabi olan büyük KAYİK'ler (>500 çalışan) | İlk dalga: ~12.000 AB şirketi |
| 2025 | 2026 Başı | Tüm büyük şirketler (>250 borç veya >40 milyon Avro) | ~50.000 AB şirketi |
| 2026 | 2027 başı | Listelenen KOBİ'ler | Basitleştirilmiş ESRS standardı |
| 2028 | 2029 başı | AB iştirakleri olan AB dışı şirketler | Önemli küresel etki |
ESRS E1 Kapsam 3 emisyonları için özellikle şunları gerektirir:
- Tüm malzeme kategorilerinin açıklanması: önemlilik belirlenmelidir ikili önemlilik analizi yoluyla (etki + finansal risk). Çoğunlukla Teknoloji şirketlerinin en az 4-6 kategorisi maddidir.
- Kategoriye göre döküm: değerler tek bir toplam olarak raporlanamaz agrega; her malzeme kategorisinin tCO2e cinsinden kendi verileri bulunmalıdır.
- Açık metodoloji: her kategori için hesaplama yöntemi beyan edilmelidir (faaliyet bazlı, harcama bazlı, tedarikçiye özel), emisyon faktörlerinin kaynağı ve kademesi veri kalitesi.
- Karbon kredileriyle netleştirme yok: Brüt emisyonlar rapor edilmelidir satın alınan herhangi bir tazminattan veya karbon dengelemesinden ayrı olarak.
- Zorunlu sigorta: Başlangıçta sınırlı güvence gelecekte makul güvenceye geçin. Bu makalede açıklanan denetim takibi eleştirmenlerin tam olarak talep edeceği şey budur.
- Hedef ve geçiş planı: şirketler hedeflerini beyan etmelidir azaltma, ara kilometre taşlarıyla 1,5°C'ye (tercihen SBTi tarafından doğrulanmıştır) hizalandı.
Dikkat: Kapsam 3 ve Çifte Önemlilik
ESRS E1, 15 Kapsam 3 kategorisinin tümünün raporlanmasını gerektirmez, yalnızca tanımlananların raporlanmasını gerektirir nasıl malzemeler ikili önemliliğin analizinde. Ancak süreç Önemliliğin tespiti belgelenmeli ve denetlenebilir olmalıdır. Birini hariç tut "Veri eksikliği nedeniyle" kategorisi kabul edilebilir bir gerekçe değildir: kanıtlanması gerekir bu kategorinin belirli bir iş için önemli olmadığını.
Örnek Olay: 50 Tedarikçili SaaS Şirketi
Her şeyi somut bir örneğe çevirelim: 150 çalışanı olan İtalyan orta ölçekli bir SaaS, 15 milyon Euro gelir, AWS altyapısı ve 50 aktif tedarikçi. Yönetim karar verdi Kapsam 3 hesaplamasını CSRD'den önce başlatın ve teslim etmek için 3 aylık bir süreye sahip olun denetçiniz tarafından doğrulanabilen veriler.
Şirket Profili: SaaS Italia S.r.l.
| Parametre | Değer |
|---|---|
| Çalışanlar | 150 (%70 akıllı çalışma) |
| Konumlar | Milano Merkez + Roma ofisi |
| Altyapı | AWS eu-west-1 (birincil), GCP europe-west1 (yedek) |
| Aktif tedarikçiler | 50 (8 büyük, 42 küçük/orta) |
| Tedarik gideri | ~4,2 milyon Avro/yıl |
| Yıllık uçuşlar | ~380 uçuş (konferanslar + müşteriler) |
Aşama 1 – Önceliklendirme Analizi (1-2. Hafta): ESG ekibi Kapsam 3 malzeme kategorilerini belirlemek için hızlı bir analiz gerçekleştirdi. Şuradaki veriler kullanılıyor: EEIO faktörlerinin başlangıç göstergesi olarak ERP'den (SAP) yapılan harcamalar dikkate alındığında, şu tahmini elde ettiler: "tarama":
Kapsam 3 Önemlilik Taraması — SaaS Italia S.r.l.
| Kedi. | Tanım | Harcamaya dayalı tahmin (tCO2e) | Toplamın %'si | Karar |
|---|---|---|---|---|
| 1 | Satın alınan ürün ve hizmetler (bulut, SW) | 342 | %54 | MALZEME → Faaliyet bazlı |
| 6 | İş seyahati | 98 | %15 | MALZEME → Faaliyet bazlı |
| 7 | Çalışanların işe gidiş gelişleri | 87 | %14 | MALZEME → Çalışan anketleri |
| 11 | Satılan ürünlerin kullanımı | 76 | %12 | MALZEME → SKI ölçümü |
| 2 | Sermaye malları (dizüstü bilgisayarlar, donanım) | 28 | 4% | MALZEME → Satıcı PCF'si |
| Diğer | Kedi. 3, 5, 8, 15 | 6 | 1% | Maddi olmayan → Harcamaya dayalı |
Aşama 2 – Veri Toplama (2-8. Hafta):
- Kedi. 1 (Bulut): AWS Müşteri Karbon Ayak İzi Aracı ve GCP Karbon Ayak İzi hesap başına aylık düzenleme verileri sağlarlar. API aracılığıyla çıkarılan ve Bronze'a yüklenen veriler katmanlar. Kalite: TIER 1 (tedarikçiye özel, AWS onaylı).
- Kedi. 1 (Yazılım ve hizmetler): 8 büyük tedarikçiyle (>50.000 EUR/yıl) iletişime geçildi yapılandırılmış bir anket ile. 5 tanesi birincil verilerle yanıt verdi (Microsoft ERP, Slack dahil), Satış gücü). 3'ünde veri yok → EEIO ile harcamaya dayalı.
- Kedi. 6 (İş seyahati): seyahat acentesinden alınan veriler (Carlson Wagonlit) API aracılığıyla: Yönlendirme ve sınıfla birlikte 380 uçuş. DEFRA 2024 ile aktivite bazlı hesaplama.
- Kedi. 7 (İşe gidip gelme): 150 çalışanın tamamına anonim anket (%82 yanıt oranı). Ulaşım araçları, ortalama mesafe, ofiste haftanın günleri.
- Kedi. 11 (Satılan ürünlerin kullanımı): Hesaplanan SCI (Yazılım Karbon Yoğunluğu) Üretim altyapısında CodeCarbon ile. Aktif oturum/ay sayısıyla çarpılır.
# case_study/saas_italia_scope3.py
# Calcolo completo Scope 3 per SaaS Italia S.r.l.
from calculators.emission_calculator import (
calculate_activity_based,
calculate_spend_based,
calculate_category_total
)
def calculate_cat1_cloud() -> dict:
"""Cat. 1: Emissioni cloud AWS + GCP (dati primari vendor)"""
# Dati estratti dall'AWS Customer Carbon Footprint API
aws_kwh_year = 187_500 # kWh totali 2024
gcp_kwh_year = 12_300
aws_ef = 0.233 # kg CO2e/kWh IT grid (AWS eu-west-1)
gcp_ef = 0.198 # kg CO2e/kWh GCP europe-west1
aws_co2, aws_unc = calculate_activity_based("cloud_compute_kwh", aws_kwh_year, aws_ef)
gcp_co2, gcp_unc = calculate_activity_based("cloud_compute_kwh", gcp_kwh_year, gcp_ef)
activities = [
{"value_kg": aws_co2, "uncertainty_pct": 8.0, "quality_tier": "TIER_1"},
{"value_kg": gcp_co2, "uncertainty_pct": 10.0, "quality_tier": "TIER_1"},
]
result = calculate_category_total(activities)
result["category"] = 1
result["sub_category"] = "cloud_infrastructure"
return result
def calculate_cat6_business_travel() -> dict:
"""Cat. 6: Business travel (dati agenzia viaggi)"""
# 380 voli totali anno 2024
# 60% corto raggio (<1500km), 40% lungo raggio
short_haul_pkm = 380 * 0.6 * 850 # 850km avg corto raggio
long_haul_pkm = 380 * 0.4 * 3200 # 3200km avg lungo raggio
short_co2, short_unc = calculate_activity_based(
"flight_economy_short", short_haul_pkm
)
long_co2, long_unc = calculate_activity_based(
"flight_economy_long", long_haul_pkm
)
# Radiative forcing factor x1.9 per quota alta
rf_factor = 1.9
short_co2 *= rf_factor
long_co2 *= rf_factor
activities = [
{"value_kg": short_co2, "uncertainty_pct": 20.0, "quality_tier": "TIER_2"},
{"value_kg": long_co2, "uncertainty_pct": 20.0, "quality_tier": "TIER_2"},
]
result = calculate_category_total(activities)
result["category"] = 6
return result
def calculate_cat7_commuting() -> dict:
"""Cat. 7: Employee commuting (survey 123/150 dipendenti)"""
# Risultati survey (valori medi per dipendente/anno)
commuters = {
"car_solo": {"count": 38, "km_day": 28, "days_year": 120},
"car_shared": {"count": 12, "km_day": 22, "days_year": 110},
"public_transport": {"count": 52, "km_day": 35, "days_year": 140},
"cycling_walking": {"count": 21, "km_day": 4, "days_year": 150},
}
activities = []
# Auto privata
car_pkm = (
commuters["car_solo"]["count"] *
commuters["car_solo"]["km_day"] *
commuters["car_solo"]["days_year"]
)
co2_car, unc = calculate_activity_based("car_average", car_pkm)
activities.append({"value_kg": co2_car, "uncertainty_pct": 15.0, "quality_tier": "TIER_2"})
# Trasporto pubblico
pt_pkm = (
commuters["public_transport"]["count"] *
commuters["public_transport"]["km_day"] *
commuters["public_transport"]["days_year"]
)
co2_pt, unc = calculate_activity_based("public_transport_it", pt_pkm)
activities.append({"value_kg": co2_pt, "uncertainty_pct": 20.0, "quality_tier": "TIER_2"})
# Ciclismo/piedi: zero emissioni dirette
activities.append({"value_kg": 0.0, "uncertainty_pct": 0.0, "quality_tier": "TIER_1"})
result = calculate_category_total(activities)
result["category"] = 7
result["survey_response_rate"] = 82.0
return result
def calculate_cat11_use_of_products() -> dict:
"""Cat. 11: Energia consumata dai clienti usando il SaaS"""
# SCI = 0.045 gCO2e per ogni API call (misurato con CodeCarbon)
sci_gco2e_per_call = 0.045
avg_calls_per_month = 48_500_000 # 48.5M calls/mese (dati produzione)
months = 12
total_calls = avg_calls_per_month * months
co2e_grams = total_calls * sci_gco2e_per_call
co2e_kg = co2e_grams / 1000
activities = [
{"value_kg": co2e_kg, "uncertainty_pct": 25.0, "quality_tier": "TIER_2"}
]
result = calculate_category_total(activities)
result["category"] = 11
result["metric"] = "API calls"
result["total_calls"] = total_calls
return result
def run_full_scope3_calculation() -> dict:
"""Esegue il calcolo completo Scope 3 per SaaS Italia S.r.l."""
results = {
"cat_1_cloud": calculate_cat1_cloud(),
"cat_6_travel": calculate_cat6_business_travel(),
"cat_7_commuting": calculate_cat7_commuting(),
"cat_11_use": calculate_cat11_use_of_products(),
}
# Categoria residuale (spend-based per tutto il resto)
residual_spend_eur = 210_000 # ~5% della spesa totale
residual_co2_kg, res_unc = calculate_spend_based(
residual_spend_eur, "it_services"
)
results["cat_residual"] = {
"total_co2e_tonnes": residual_co2_kg / 1000,
"uncertainty_pct": res_unc,
"category": "other",
"dominant_quality_tier": "TIER_3"
}
# Totale Scope 3
total_tco2e = sum(
v["total_co2e_tonnes"] for v in results.values()
)
from calculators.emission_calculator import propagate_uncertainty
all_values = [v["total_co2e_tonnes"] for v in results.values()]
all_uncertainties = [v["uncertainty_pct"] for v in results.values()]
overall_uncertainty = propagate_uncertainty(all_values, all_uncertainties)
return {
"company": "SaaS Italia S.r.l.",
"reporting_year": 2024,
"methodology": "GHG Protocol Corporate Value Chain Standard",
"scope3_total_tco2e": round(total_tco2e, 1),
"overall_uncertainty_pct": round(overall_uncertainty, 1),
"categories": results,
"notes": "Cat. 11 include radiative forcing factor per aviation"
}
if __name__ == "__main__":
import json
report = run_full_scope3_calculation()
print("=" * 50)
print(f"SCOPE 3 TOTALE: {report['scope3_total_tco2e']} tCO2e")
print(f"Incertezza: +/- {report['overall_uncertainty_pct']}%")
print("=" * 50)
for name, cat in report["categories"].items():
tco2e = cat.get("total_co2e_tonnes", 0)
unc = cat.get("uncertainty_pct", 0)
pct = tco2e / report["scope3_total_tco2e"] * 100
print(f" {name:25s} {tco2e:6.1f} tCO2e ({pct:.0f}%) ±{unc:.0f}%")SaaS Italia S.r.l. için hesaplama sonucu. üretir:
Kapsam 3 Nihai Sonuçları — SaaS Italia S.r.l. (2024 Mali Yılı)
| Kategori | tCO2e | % Toplam | Belirsizlik | Katmanlar |
|---|---|---|---|---|
| Kedi. 1 – Bulut ve Hizmetler | 43.7 | %36 | ±%9 | 1. KADEME |
| Kedi. 6 – İş Seyahati | 33.5 | %28 | ±%20 | 2. KADEME |
| Kedi. 7 – İşe gidip gelme | 23.8 | %20 | ±%17 | 2. KADEME |
| Kedi. 11 – Ürünlerin Kullanımı | 26.2 | %22 | ±%25 | 2. KADEME |
| Artık Kedi | 6.5 | 5% | ±75% | 3. KADEME |
| TOPLAM KAPSAM 3 | 133.7 | %100 | ±%13 | 2. KADEME |
Kapsam 1 (HQ kazanından ~8 tCO2e) ve Kapsam 2 (elektrikten ~12 tCO2e) eklenmesi ofislerden), bir ~154 tCO2e toplam ayak izi 2024 için %87'si Kapsam 3'tür. Tam olarak yazılım şirketlerinin tipik modeli.
Kapsam 3 İşlem Hattında En İyi Uygulamalar ve Anti-Modeller
Kapsam 3 Boru Hattı Uygulama Kontrol Listesi
| Alan | En İyi Uygulamalar | Kaçınılması Gereken Anti-Desenler |
|---|---|---|
| Veri | Alınan her veri için değişmez bronz katman | Düzeltilmiş sürümlerle ham verilerin üzerine yazın |
| Hesaplama | Kullanılan emisyon faktörlerinin versiyonu | EF'yi yıl ve kaynak belirtmeden kullanın |
| Belirsizlik | Her zaman belirsizliği her kategoriye yayın | Aralık olmadan yalnızca kesin değeri bildirin |
| Kalite | Açık ve belgelenmiş kalite puanı | TIER 1 ve TIER 3'ü ayrım yapmadan karıştırın |
| Denetimler | Her hesaplama için hash zinciri, zincir dışı doğrulanabilir | Excel raporunun sürümü yok ve izlenebilir değil |
| Tedarikçi | Harcama/emisyonlara göre ilk 20 tedarikçiye öncelik verin | 50 tedarikçinin hepsine aynı şekilde davranın |
| Güncelleme | Verilen yıllık kalite iyileştirme planı | Harcamayı kalıcı bir çözüm olarak kabul edin |
| Süpürgeler | Gerekçeli hariç tutmaları açıkça belgeleyin | Resmi gerekçe olmaksızın kategorileri hariç tutun |
Aşamalı İyileştirme Planı (Veri Olgunluğu Yol Haritası)
Sera Gazı Protokolü açıkça ilerici bir yaklaşımı teşvik etmektedir: günümüzün hiçbir şeyi olmayan düşük kaliteli harcamaya dayalı verileri. Amaç geliştirmektir her yıl malzeme kategorilerinin kalite kademesi:
- 1. Yıl (temel): %100 harcamaya dayalı, hata ±%75, TIER 3
- 2. Yıl: Faaliyet bazlı, ±%40, TIER 2 ile ilk 10 tedarikçi
- 3. Yıl: Doğrulanmış birincil verilere sahip ilk 20 tedarikçi, ±%20, TIER 2
- 4+ Yıl: Tüm tedarikçiler için EcoVadis/CDP entegrasyonu, ±%10, TIER 1
Bu ilerici gelişme CSRD raporunda şu şekilde belgelenebilir: "metodoloji evrimi" ve eleştirmenler tarafından olumlu değerlendiriliyor.
Altın Katman için Veritabanı Şeması
Altın katmanı, hızlı toplu sorguları desteklemek için tasarlanmış bir şema gerektirir CSRD raporlaması için denetim zincirinde izlenebilirliğin sağlanması.
-- schema/scope3_gold.sql
-- Schema PostgreSQL per il Gold Layer Scope 3
-- Tabella principale: emissioni aggregate per categoria
CREATE TABLE scope3_emissions_gold (
id BIGSERIAL PRIMARY KEY,
company_id VARCHAR(50) NOT NULL,
reporting_year INTEGER NOT NULL,
scope3_category INTEGER NOT NULL CHECK (scope3_category BETWEEN 1 AND 15),
supplier_id VARCHAR(100),
-- Valori emissioni
co2e_tonnes DECIMAL(12, 3) NOT NULL,
co2_tonnes DECIMAL(12, 3),
ch4_tonnes_co2e DECIMAL(12, 3),
n2o_tonnes_co2e DECIMAL(12, 3),
-- Metodologia e qualità
calculation_method VARCHAR(30) NOT NULL, -- activity_based, spend_based, etc.
emission_factor_source VARCHAR(100) NOT NULL,
emission_factor_value DECIMAL(10, 6),
quality_tier VARCHAR(10) NOT NULL, -- TIER_1, TIER_2, TIER_3
uncertainty_pct DECIMAL(5, 1) NOT NULL,
uncertainty_tonnes DECIMAL(12, 3) GENERATED ALWAYS AS
(co2e_tonnes * uncertainty_pct / 100) STORED,
-- Tracciabilità
audit_chain_id UUID NOT NULL REFERENCES scope3_audit_chain(chain_id),
pipeline_version VARCHAR(20) NOT NULL,
reporting_standard VARCHAR(50) DEFAULT 'GHG_Protocol_Scope3_2011',
-- Timestamps
published_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-- Unicità per reporting period
CONSTRAINT uq_emission_period
UNIQUE (company_id, reporting_year, scope3_category, supplier_id)
);
-- Indici per performance query CSRD report
CREATE INDEX idx_scope3_company_year
ON scope3_emissions_gold (company_id, reporting_year);
CREATE INDEX idx_scope3_category
ON scope3_emissions_gold (scope3_category);
CREATE INDEX idx_scope3_quality
ON scope3_emissions_gold (quality_tier, uncertainty_pct);
-- View per report aggregato CSRD
CREATE VIEW v_scope3_csrd_report AS
SELECT
company_id,
reporting_year,
scope3_category,
SUM(co2e_tonnes) AS total_co2e_tonnes,
-- Propagazione incertezza quadratica
SQRT(SUM(POWER(co2e_tonnes * uncertainty_pct / 100, 2))) /
NULLIF(SUM(co2e_tonnes), 0) * 100 AS combined_uncertainty_pct,
-- Qualità aggregata (tier peggiore nella categoria)
MIN(quality_tier) AS data_quality_tier,
-- Metodo più usato
MODE() WITHIN GROUP (ORDER BY calculation_method) AS primary_method,
COUNT(DISTINCT supplier_id) AS supplier_count,
MAX(updated_at) AS last_updated
FROM scope3_emissions_gold
GROUP BY company_id, reporting_year, scope3_category
ORDER BY company_id, reporting_year, scope3_category;
-- Tabella audit chain
CREATE TABLE scope3_audit_chain (
chain_id UUID PRIMARY KEY,
record_id UUID NOT NULL DEFAULT gen_random_uuid(),
sequence INTEGER NOT NULL,
record_type VARCHAR(50) NOT NULL,
previous_hash CHAR(64) NOT NULL,
record_hash CHAR(64) NOT NULL UNIQUE,
payload JSONB NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
CONSTRAINT uq_chain_sequence UNIQUE (chain_id, sequence)
);
-- Indice per verifica integrità chain
CREATE INDEX idx_audit_chain_id_seq
ON scope3_audit_chain (chain_id, sequence);Sonuçlar ve Sonraki Adımlar
Kapsam 3 emisyonları için sağlam bir boru hattı oluşturmak akademik bir çalışma değildir: bu birkritik veri altyapısı için zorunlu hale gelecek 2028 yılına kadar binlerce Avrupalı şirket. Bu makalede gördüğümüz temel ilkeler şirket büyüklüğüne bakılmaksızın geçerlidir:
- Ham verilerin değişmezliği: SHA-256 hash garantili bronz katman her incelemecinin yıllar sonra bile verileri her zaman orijinal kaynağa kadar takip edebilmesi.
- Metodolojik ilerleme: Harcamaya dayalı olarak başlayın ve şuraya geçin: Malzeme kategorileri için faaliyete dayalı yaklaşım, Sera Gazı Protokolü tarafından önerilen yaklaşımdır. kendisi, bir kısayol değil.
- Belirsizliğin ölçülmesi: Emisyonları aralıksız raporlayın güven eksik bilgidir. Belirsizliğin ikinci dereceden yayılımı Uygulanması basittir ve raporun güvenilirliği açısından temeldir.
- Doğrulanabilir denetim izi: Karma zinciri harici bir doğrulayıcıya izin verir Hesaplama sonrasında hiçbir verinin değiştirilmediğini matematiksel olarak doğrulamak için.
- Ekosistem entegrasyonu: EcoVadis Karbon Veri Ağı gibi platformlar ve CDP, özellikle büyük tedarik zincirleri için veri toplama yükünü büyük ölçüde azaltır.
SaaS Italia S.r.l.'nin örnek olay incelemesi orta ölçekli bir şirketin bile bunu yapabileceğini gösteriyor 2-3 kişilik bir ekiple 3 ayda Kapsam 3 CSRD uyumlu bir rapor hazırlamak, malzeme kategorileri için birincil veriler ve kalanlar için harcamaya dayalı veriler. Anahtar şu: önceliklendirme: Her yerde mükemmeli aramayın, çabanızı yoğunlaştırın Emisyonların en yüksek olduğu yer.
Yararlı Kaynaklar
- Sera Gazı Protokolü Kapsam 3 Standardı: ghgprotocol.org/corporate-value-chain-scope-3-standard
- İklimlendirme API'si (emisyon faktörleri veritabanı): iklim.io
- EcoVadis Karbon Veri Ağı: ecovadis.com/solutions/karbon
- ESRS E1 İklim Değişikliği (resmi AB metni): EFRAG ESRS E1
- EXIOBASE 3.8 (EEIO harcamaya dayalı faktörler): exiobase.eu
Serideki Sonraki Makale
Bir sonraki makalede ESG Raporlama API'si: CSRD İş Akışı ile Entegrasyon bu makalede hesaplanan Kapsam 3 verilerinin üzerine bir REST API katmanı oluşturacağız. Avrupa direktifinin gerektirdiği formatlarla uyumlu ve iş akışını entegre eden uç noktalar denetçinin dijital imzasıyla rapor onayı.
Ayrıca verileri formatta nasıl ortaya çıkaracağımızı da göreceğiz. XBRL/iXBRL teslim için CSRD raporları için zorunlu format olan ESEF'e (Avrupa Tek Elektronik Formatı) Avrupa borsasında işlem görüyor.