Karbon Muhasebe Yazılım Mimarisi: ESG Platformları
2025'te karbon raporlaması artık büyük şirketler için gönüllü bir tercih değil Avrupa ülkeleri: Yasal bir zorunluluktur. Orada Kurumsal Sürdürülebilirlik Raporlaması Direktifi (CSRD), giriş İtalya'da 125/2024 sayılı Kanun Hükmünde Kararname ile yürürlükte olan bu rapor, ESG raporlama ortamını kökten değiştirdi ve Kurumsal iletişim alanından mevzuata uyum alanına kadar uzanan disiplin, aynı Finansal muhasebe sisteminin operasyonel ve teknik etkileri.
Küresel karbon muhasebesi yazılım pazarı yakalandı 2025'te 1,8 milyar dolar ve 2030 yılına kadar %27,4'lük bir Bileşik Büyüme Oranı ile 6 milyarı aşması bekleniyor. Bu sadece satın almakla ilgili değil Bir SaaS platformu: Sağlam bir karbon muhasebesi sistemi tasarlamak, aşağıdakiler arasında değişen beceriler gerektirir: atmosfer kimyasından yazılım mühendisliğine, istatistikten topluluk hukukuna kadar. Mühendislik ekipleri Kendilerini bu platformları inşa ederken bulanların şunu anlaması gerekir: Sera Gazı Protokolü, gli Avrupa Sürdürülebilirlik Raporlama Standartları (ESRS), emisyon faktörü veritabanları ve yüzlerce veriden veri toplama, hesaplama ve sertifikalandırma kapasitesine sahip dağıtılmış sistemlerin mimarisi farklı kaynaklardan.
Bu makale eksiksiz bir teknik kılavuzdur: etki alanının GHG Protokolü varlıklarıyla modellenmesinden, FastAPI ile Python'da bir hesaplama motorunun uygulanmasına ve CSRD/CDP/GRI raporlarının otomasyonuna kadar ve gerçek zamanlı emisyon faktörleri için Climatiq API ile entegrasyon. Bir platform inşa ediyorsanız Kurumsal çözümleri değerlendiren dahili ESG veya yalnızca bu mimarilerin nasıl çalıştığını anlamak istiyorsanız kaputun altında doğru yerdesiniz.
Bu Makalede Neler Öğreneceksiniz?
- Sera Gazı Protokolü: Kapsam 1, 2 ve 3 ile tedarik zincirinden kaynaklanan 15 dolaylı emisyon kategorisi
- Bir karbon muhasebe sisteminin veri alanı nasıl modellenir (Organizasyon, Tesis, EmisyonKaynağı, Faaliyet)
- Emisyon faktörlerinin ana veritabanları: DEFRA, EPA, ecoinvent, Climatiq API
- Mikro hizmet mimarisi: Veri Toplama, Hesaplama Motoru, Raporlama, Denetim İzi
- Emisyon hesaplaması için FastAPI ve pandalarla Python uygulaması
- Kapsam 3 otomasyonu: SAP ERP ile entegrasyon, satın alma verileri, seyahat verileri
- CSRD/ESRS, CDP ve GRI Standartları raporlarının otomatik oluşturulması
- Denetim takibi ve veri kökeni: hesaplamaların tam izlenebilirliği
- Platform karşılaştırması: Persefoni, Watershed, Sphera, Plan A
- Örnek olay: 2025 CSRD raporlamasına sahip İtalyan imalat şirketi
- Güncellenen mevzuat: CSRD Omnibus 2025, AB Taksonomisi, Kanun Hükmünde Kararname 125/2024
EnergyTech Serisindeki Konumu
| # | Öğe | Durum |
|---|---|---|
| 1 | MQTT ve InfluxDB: Enerji Verilerine İlişkin Zaman Serileri | Yayınlandı |
| 2 | IEC 61850: Akıllı Elektrik Ağları için Standart Protokol | Yayınlandı |
| 3 | DERMS: Dağıtılmış Enerji Kaynağı Yönetimi | Yayınlandı |
| 4 | Bina Yönetim Sistemi: Yapay Zeka Enerji Optimizasyonu | Yayınlandı |
| 5 | Yenilenebilir Enerji Tahmini: Fotovoltaik ve Rüzgar için ML | Yayınlandı |
| 6 | EV Yük Dengeleme: Akıllı Şarj ve Araçtan Şebekeye | Yayınlandı |
| 7 | Blockchain P2P Enerji Ticareti: Merkezi Olmayan Enerji Piyasaları | Yayınlandı |
| 8 | Buradasınız - Karbon Muhasebe Yazılım Mimarisi: ESG Platformları | Akım |
| 9 | Energy Digital Twin: Simülasyon ve Optimizasyon | Sonraki |
| 10 | OCPP ve EV Altyapısı: Standartlar ve Uygulama | Yakında gelecek |
Düzenleyici Bağlam: CSRD, ESRS ve 125/2024 sayılı Kanun Hükmünde Kararname
Mevzuatı anlamak bürokratik bir ön koşul değildir: tüm mimarinin üzerinde tasarlandığı temeldir sistemin. Her teknik gereksinim, her veritabanı alanı, her API uç noktası bir zorunluluktan kaynaklanır özel açıklama.
CSRD ve Uygulama Dalgaları
La Kurumsal Sürdürülebilirlik Raporlaması Yönergesi tanımlanmış üç uygulama dalgası Avrupalı şirketler için. İlk dalga (2024 Mali Yılı, 2025'te yayınlanan raporlar) şirketleri içeriyordu halihazırda önceki NFRD'ye tabidir: 500'den fazla çalışanı olan borsaya kayıtlı şirketler, bankalar ve sigorta şirketleri. Başlangıçta 2025 ve 2026 için planlanan ikinci ve üçüncü dalgalar iki yıl ertelendi itibaren Saati Durdurma Direktifi 16 Nisan 2025 tarihinde AB Resmi Gazetesinde yayınlandı.
En önemli değişiklik ise Aralık 2025'te paketin onaylanmasıyla geldi. Omnibus I: Zorunlu başvuru eşiği yükseltildi 1.000 çalışan ve 450 milyon euro cirosayısını yaklaşık %80 oranında azalttı. Zorunlu CSRD'ye tabi şirketlerin sayısı. ESRS standartları azaltılarak revize ediliyor Zorunlu veri noktalarının %61'i (yaklaşık 1.100'den yaklaşık 430'a), ilk yarıda benimsenmesi bekleniyor 2026 ve başvuru 2027'den itibaren.
Güncellenmiş CSRD Zaman Çizelgesi (Omnibus 2025 sonrası)
| Dalga | Konular | İlk Rapor | Durum |
|---|---|---|---|
| 1. Dalga | Halihazırda NFRD'ye sahip şirketler (>500 istihdam, borsada işlem gören/bankalar/sigorta) | Rapor 2025 (2024 Mali Yılı) | Devam etmekte |
| 2. Dalga | Büyük şirketler >1.000 çalışan ve >450 milyon Avro ciro | Rapor 2027 (2026 Mali Yılı) - ertelendi | Ertelendi |
| 3. Dalga | Düzenlenmiş AB pazarlarında listelenen KOBİ'ler | Rapor 2028 (2027 Mali Yılı) - ertelendi | Ertelendi |
| ESRS rev. | Tüm CSRD konuları, basitleştirilmiş standartlar | 2027 Mali Yılı'ndan itibaren | istişare halinde |
Kanun Hükmünde Kararname 125/2024: İtalyan aktarımı
İtalya CSRD'yi şu şekilde uygulamıştır: 6 Eylül 2024 Tarihli ve 125 Sayılı Kanun Hükmünde Kararname, 25 Eylül 2024 tarihinde yürürlüğe girmiştir. Kararname, önceki Kanun Hükmünde Kararname olan 254/2016'yı yürürlükten kaldırmıştır. NFRD'yi aktardı. İtalyan şirketlerine yönelik başlıca yenilikler şunlardır: sınırlı güvence (sınırlı güvence) tarafından sürdürülebilirlik raporuna ilişkin nitelikli denetçi, sürdürülebilirlik raporunun yönetim raporuna entegrasyonu, ve Şirket Sicili'nin özel bölümünde yayımlanması.
AB Taksonomisi: Bağlantı
Karbon muhasebesi doğrudan kesişiyor AB Taksonomi Yönetmeliği (AB Tüzüğü 852/2020), Ekonomik faaliyetleri altı çevresel hedefe dayalı olarak "sürdürülebilir" olarak sınıflandıran kuruluştur. CSRD şirketleri, Sınıflandırma uyumu KPI'larını (ciro payları, Sermaye Harcamaları, OpEx) raporlamalıdır "uyumlu" ve "uygun"). 2025 yılı sadeleştirmeleri ile 1.000'den az çalışanı olan şirketler Sınıflandırma raporlamasından muaftır ve geri kalanı temsil ettikleri faaliyetlerle sınırlı olabilir Cironun en az %10'u, CapEx veya OpEx.
Sera Gazı Protokolü: Referans Çerçevesi
Il Sera Gazı Protokolü ve sera gazı emisyonlarına ilişkin muhasebe standardı Dünya Kaynakları Enstitüsü (WRI) ve Dünya İş Konseyi tarafından geliştirilen, dünyada en yaygın olanı Sürdürülebilir Kalkınma için (WBCSD). Hemen hemen tüm raporlama çerçeveleri (CSRD/ESRS, CDP, GRI, ISO 14064) GHG Protokolüne atıfta bulunur veya bu Protokolü temel alır.
Üç Kapsam: Kesin Tanımlar
Üç "kapsam"a yapılan ayrım, emisyonların açıkça atfedilmesine olanak sağlar. Tedarik zincirindeki farklı aktörler arasında sayımlarda tekrarların önlenmesi.
Kapsam 1: Doğrudan Emisyonlar
Kuruluşun sahip olduğu veya kontrol ettiği kaynaklardan kaynaklanan emisyonlar. Bunlar şunları içerir: fosil yakıtların kazanlarda, fırınlarda, şirkete ait araçlarda yakılması; proses emisyonları (örneğin kimyasal reaksiyonlardan kaynaklanan CO2, hayvancılıktan kaynaklanan CH4); kaçak emisyonlar (soğutucu akışkan sızıntıları, sistemlerden gaz sızıntıları). İlgili sera gazı gazları yedi tanesidir: Kyoto Protokolü: CO2, CH4, N2O, HFC, PFC, SF6, NF3, hepsi dönüştürülür CO2 eşdeğeri (CO2e) IPCC Küresel Isınma Potansiyellerini (GWP'ler) kullanarak.
Kapsam 2: Satın Alınan Enerjiden Kaynaklanan Dolaylı Emisyonlar
Elektrik, ısı, buhar veya soğutma üretimiyle ilişkili emisyonlar kuruluş tarafından satın alınır ve tüketilir. Sera Gazı Protokolü Kapsam 2 Rehberi (2015) şunları gerektirir: ile raporlama iki farklı yöntem: yöntem konuma dayalı (bulunduğunuz yerdeki elektrik şebekesinin emisyon faktörünü kullanın) tüketim, ör. İtalyan ağının ortalama faktörü) ve yöntem pazar bazlı (Fark Sözleşmeleri, Sertifikalar gibi piyasa araçlarının faktörlerini kullanır Yenilenebilir Enerji/GO, Enerji Satın Alma Anlaşmaları). Her ikisinin de bildirilmesi gerekiyor.
Kapsam 3: Diğer Dolaylı Emisyonlar
Değer zincirinin emisyonları, 15 üst ve alt kategoriye ayrılmıştır. Bunlar genellikle en önemlileridir (toplam ayak izinin ortalama %70-80'i) ve en ölçülmesi zordur. ESRS, Kapsam 3 kategorilerinin raporlanmasını gerektirir Çifte Önemlilik Değerlendirmesi aracılığıyla belirlenen "materyaller".
15 Kapsam 3 Kategorisi
| # | Kategori | Tip | için tipik |
|---|---|---|---|
| 1 | Satın alınan mal ve hizmetler | Yukarı akış | Tüm sektörler |
| 2 | Sermaye malları | Yukarı akış | İmalat, inşaat |
| 3 | Yakıt ve enerji ile ilgili faaliyetler | Yukarı akış | Tüm sektörler |
| 4 | Yukarı yönde taşıma ve dağıtım | Yukarı akış | Perakende, imalat |
| 5 | Operasyonlarda oluşan atıklar | Yukarı akış | Üretim, gıda |
| 6 | İş seyahati | Yukarı akış | Hizmetler, teknoloji |
| 7 | Çalışanların işe gidiş gelişleri | Yukarı akış | Tüm sektörler |
| 8 | Yukarı yönde kiralanan varlıklar | Yukarı akış | Gayrimenkul, perakende |
| 9 | Aşağı yönde ulaşım | Aşağı akış | İmalat, FMCG |
| 10 | Satılan ürünlerin işlenmesi | Aşağı akış | Hammaddeler, kimyasal |
| 11 | Satılan ürünlerin kullanımı | Aşağı akış | Otomotiv, elektronik |
| 12 | Yaşam sonu tedavisi | Aşağı akış | Ambalaj, dayanıklı tüketim malları |
| 13 | Aşağı yöndeki kiralanan varlıklar | Aşağı akış | Gayrimenkul |
| 14 | Bayilikler | Aşağı akış | Yiyecek ve içecek, perakende |
| 15 | Yatırımlar | Aşağı akış | Bankalar, yatırım fonları |
Veri Modeli: Karbon Muhasebesi Etki Alanının Modellenmesi
Her karbon muhasebesi platformunun kalbi, aslına sadık kalarak yansıtan sağlam bir veri modelidir. Sera Gazı Protokolü kavramları. Ana varlıkları nitelikleri ve ilişkileriyle görelim.
Ana Kuruluşlar
Eksiksiz bir sistem en az altı temel varlık içerir: Organizasyon, Tesis, Emisyon Kaynağı, Emisyon Faktörü, Faaliyet ve Hesaplama. İşte SQLAlchemy ile Python'daki model:
# models/core.py - Data model GHG Protocol completo
from sqlalchemy import Column, String, Float, Enum, DateTime, ForeignKey, JSON, Integer
from sqlalchemy.orm import relationship, DeclarativeBase
from datetime import datetime
from enum import Enum as PyEnum
import uuid
class Base(DeclarativeBase):
pass
class ScopeType(PyEnum):
SCOPE_1 = "scope_1"
SCOPE_2_LOCATION = "scope_2_location"
SCOPE_2_MARKET = "scope_2_market"
SCOPE_3 = "scope_3"
class Scope3Category(PyEnum):
CAT_1_PURCHASED_GOODS = "cat_1"
CAT_2_CAPITAL_GOODS = "cat_2"
CAT_3_FUEL_ENERGY = "cat_3"
CAT_4_UPSTREAM_TRANSPORT = "cat_4"
CAT_5_WASTE = "cat_5"
CAT_6_BUSINESS_TRAVEL = "cat_6"
CAT_7_EMPLOYEE_COMMUTING = "cat_7"
CAT_8_UPSTREAM_LEASED = "cat_8"
CAT_9_DOWNSTREAM_TRANSPORT = "cat_9"
CAT_10_PROCESSING = "cat_10"
CAT_11_USE_OF_PRODUCTS = "cat_11"
CAT_12_END_OF_LIFE = "cat_12"
CAT_13_DOWNSTREAM_LEASED = "cat_13"
CAT_14_FRANCHISES = "cat_14"
CAT_15_INVESTMENTS = "cat_15"
class Organization(Base):
__tablename__ = "organizations"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
name = Column(String(255), nullable=False)
legal_entity_id = Column(String(100)) # LEI o P.IVA
country_code = Column(String(3), nullable=False) # ISO 3166-1 alpha-3
nace_code = Column(String(10)) # Codice settore NACE Rev.2
reporting_year = Column(Integer, nullable=False)
consolidation_approach = Column(String(50)) # equity_share, financial_control, operational_control
base_year = Column(Integer) # anno di riferimento per i target
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
facilities = relationship("Facility", back_populates="organization")
calculations = relationship("Calculation", back_populates="organization")
class Facility(Base):
__tablename__ = "facilities"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
organization_id = Column(String, ForeignKey("organizations.id"), nullable=False)
name = Column(String(255), nullable=False)
facility_type = Column(String(50)) # manufacturing_plant, office, warehouse, data_center
address = Column(String(500))
country_code = Column(String(3), nullable=False)
latitude = Column(Float)
longitude = Column(Float)
grid_region = Column(String(50)) # es. "IT_NORD", "DE_TENNET" per Scope 2
floor_area_sqm = Column(Float)
is_owned = Column(String(10)) # owned, leased, operated
operational_start = Column(DateTime)
organization = relationship("Organization", back_populates="facilities")
emission_sources = relationship("EmissionSource", back_populates="facility")
class EmissionSource(Base):
"""Fonte di emissione specifica: caldaia, veicolo, processo, acquisto energia"""
__tablename__ = "emission_sources"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
facility_id = Column(String, ForeignKey("facilities.id"), nullable=False)
name = Column(String(255), nullable=False)
source_type = Column(String(100)) # natural_gas_boiler, diesel_generator, company_car, electricity
scope = Column(Enum(ScopeType), nullable=False)
scope_3_category = Column(Enum(Scope3Category)) # solo per Scope 3
fuel_type = Column(String(50)) # natural_gas, diesel, petrol, LPG
unit_of_measure = Column(String(20)) # kWh, liters, kg, km, tonne
description = Column(String(1000))
is_active = Column(String(10), default="true")
facility = relationship("Facility", back_populates="emission_sources")
activities = relationship("Activity", back_populates="emission_source")
class EmissionFactor(Base):
"""Fattore di emissione da database certificato"""
__tablename__ = "emission_factors"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
source_database = Column(String(50), nullable=False) # DEFRA, EPA, ecoinvent, Climatiq, IPCC
activity_id = Column(String(200)) # ID specifico del database sorgente
name = Column(String(500), nullable=False)
category = Column(String(100))
region = Column(String(50)) # country/region code
year = Column(Integer, nullable=False)
unit_type = Column(String(50)) # kgCO2e/kWh, kgCO2e/liter, kgCO2e/km, kgCO2e/tonne
co2e_factor = Column(Float, nullable=False) # valore principale in kgCO2e
co2_factor = Column(Float) # CO2 separato
ch4_factor = Column(Float) # CH4 separato
n2o_factor = Column(Float) # N2O separato
gwp_version = Column(String(20), default="AR6") # AR5, AR6
lca_activity = Column(String(50)) # upstream, combustion, downstream
source_url = Column(String(500))
last_updated = Column(DateTime)
class Activity(Base):
"""Registrazione di attivita con consumo misurato"""
__tablename__ = "activities"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
emission_source_id = Column(String, ForeignKey("emission_sources.id"), nullable=False)
emission_factor_id = Column(String, ForeignKey("emission_factors.id"))
period_start = Column(DateTime, nullable=False)
period_end = Column(DateTime, nullable=False)
quantity = Column(Float, nullable=False)
unit = Column(String(20), nullable=False)
data_quality = Column(String(20), default="measured") # measured, estimated, calculated, supplier
data_source = Column(String(200)) # nome sistema sorgente: SAP, utility_bill, travel_tool
raw_data = Column(JSON) # dati originali non elaborati
notes = Column(String(1000))
created_by = Column(String(100))
created_at = Column(DateTime, default=datetime.utcnow)
emission_source = relationship("EmissionSource", back_populates="activities")
emission_factor = relationship("EmissionFactor")
calculation = relationship("Calculation", back_populates="activity", uselist=False)
class Calculation(Base):
"""Risultato del calcolo emissioni, immutabile una volta creato"""
__tablename__ = "calculations"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
organization_id = Column(String, ForeignKey("organizations.id"), nullable=False)
activity_id = Column(String, ForeignKey("activities.id"), nullable=False)
scope = Column(Enum(ScopeType), nullable=False)
scope_3_category = Column(Enum(Scope3Category))
co2e_tonnes = Column(Float, nullable=False) # risultato in tonnellate CO2e
co2_tonnes = Column(Float)
ch4_tonnes = Column(Float)
n2o_tonnes = Column(Float)
calculation_method = Column(String(50)) # spend_based, activity_based, hybrid, supplier_specific
emission_factor_value = Column(Float) # snapshot del fattore usato
emission_factor_unit = Column(String(50))
emission_factor_source = Column(String(100))
calculation_formula = Column(String(500)) # formula usata, per audit
calculated_at = Column(DateTime, default=datetime.utcnow)
calculated_by = Column(String(100)) # user o sistema automatico
version = Column(Integer, default=1) # versioning per ricertificazione
is_verified = Column(String(10), default="false")
organization = relationship("Organization", back_populates="calculations")
activity = relationship("Activity", back_populates="calculation")Emisyon Faktörleri Veritabanı: DEFRA, EPA, ecoinvent, Climatiq
Emisyon faktörleri karbon muhasebesinin matematiksel kalbidir: bir miktarı dönüştürürler ton CO2e cinsinden aktivite (litre dizel, tüketilen kWh, kat edilen km, harcanan euro). Faktörlerin kalitesi ve güncelliği raporun kalitesini doğrudan belirler.
Ana Veritabanları
| Veritabanları | Müdür | Kapsam | Güncelleme | Erişim |
|---|---|---|---|---|
| DEFRA/BEIS | Birleşik Krallık Hükümeti (DESNZ) | Birleşik Krallık + uluslararası, tüm kapsamlar | Yıllık (Temmuz) | Ücretsiz (Excel) |
| EPA Sera Gazı Merkezi | ABD EPA | ABD, mobilite, enerji, Kapsam 3 | Yıllık (Ocak) | Ücretsiz (Excel) |
| eko icat | eko icat Derneği | Küresel, tam LCA, 18.000'den fazla veri kümesi | Altı ayda bir | Ücretli lisans |
| IPCC EF Veritabanı | IPCC | Küresel, ulusal envanterler | Her Değerlendirme Raporuyla | Özgür |
| İklimlendirme API'si | İklim | Çoklu kaynak, 50.000+ faktör | Sürekli (gerçek zamanlı) | API (ücretsiz) |
| AIB Artık Karışımı | Düzenleyen Kuruluşlar Birliği | Avrupa, Kapsam 2 pazar bazlı | Yıllık | Özgür |
| IEA Elektrik | Uluslararası Enerji Ajansı | Küresel, elektrik şebekesi faktörleri | Yıllık | Kısmen ücretsiz |
Climatiq API entegrasyonu
Climatiq, 50.000'den fazla doğrulanmış emisyon faktörünü, kapsamını içeren bir REST API sağlar ISO 14067 ve GHG Protokolü ile entegre hesaplamalar için en çok kullanılan çözümlerden biridir. Emisyonların programlı olarak API, aktiviteye dayalı yöntemlerle Kapsam 1, 2 ve 3'ü destekler ve harcamaya dayalı.
# services/climatiq_client.py - Integrazione Climatiq API
import httpx
import os
from dataclasses import dataclass
from typing import Optional
from functools import lru_cache
CLIMATIQ_BASE_URL = "https://api.climatiq.io"
@dataclass
class EmissionEstimate:
co2e: float
co2e_unit: str
co2e_calculation_method: str
co2e_calculation_origin: str
emission_factor_name: str
emission_factor_id: str
source: str
year: int
region: str
@dataclass
class ActivityData:
activity_id: str
data: dict # { "energy": { "value": 1000, "energy_unit": "kWh" } }
region: Optional[str] = None
year: Optional[int] = None
class ClimatiqClient:
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key or os.environ.get("CLIMATIQ_API_KEY")
if not self.api_key:
raise ValueError("CLIMATIQ_API_KEY not set")
self.client = httpx.AsyncClient(
base_url=CLIMATIQ_BASE_URL,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=30.0
)
async def estimate_emission(self, activity: ActivityData) -> EmissionEstimate:
"""Calcola emissioni per una singola attivita"""
payload = {
"emission_factor": {
"activity_id": activity.activity_id,
},
**activity.data
}
if activity.region:
payload["emission_factor"]["region"] = activity.region
if activity.year:
payload["emission_factor"]["year"] = activity.year
response = await self.client.post("/estimate", json=payload)
response.raise_for_status()
result = response.json()
return EmissionEstimate(
co2e=result["co2e"],
co2e_unit=result["co2e_unit"],
co2e_calculation_method=result["co2e_calculation_method"],
co2e_calculation_origin=result.get("co2e_calculation_origin", ""),
emission_factor_name=result["emission_factor"]["name"],
emission_factor_id=result["emission_factor"]["activity_id"],
source=result["emission_factor"]["source"],
year=result["emission_factor"]["year"],
region=result["emission_factor"].get("region", "")
)
async def estimate_batch(self, activities: list[ActivityData]) -> list[EmissionEstimate]:
"""Calcolo batch fino a 100 attivita"""
payload = {
"requests": [
{
"emission_factor": {"activity_id": a.activity_id},
**a.data
}
for a in activities
]
}
response = await self.client.post("/batch", json=payload)
response.raise_for_status()
results = response.json()["results"]
return [
EmissionEstimate(
co2e=r["co2e"],
co2e_unit=r["co2e_unit"],
co2e_calculation_method=r["co2e_calculation_method"],
co2e_calculation_origin=r.get("co2e_calculation_origin", ""),
emission_factor_name=r["emission_factor"]["name"],
emission_factor_id=r["emission_factor"]["activity_id"],
source=r["emission_factor"]["source"],
year=r["emission_factor"]["year"],
region=r["emission_factor"].get("region", "")
)
for r in results
]
async def search_emission_factors(
self,
query: str,
region: Optional[str] = None,
year: Optional[int] = None,
source: Optional[str] = None
) -> list[dict]:
"""Ricerca fattori di emissione nel database Climatiq"""
params = {"query": query, "page": 1, "page_size": 20}
if region:
params["region"] = region
if year:
params["year"] = year
if source:
params["source"] = source
response = await self.client.get("/search", params=params)
response.raise_for_status()
return response.json()["results"]
async def __aenter__(self):
return self
async def __aexit__(self, *args):
await self.client.aclose()
# Esempio utilizzo - calcolo Scope 2 location-based per impianto italiano
async def calculate_scope2_italy(kwh_consumed: float) -> float:
"""
Calcola emissioni Scope 2 location-based per consumo elettrico in Italia.
Fattore IEA 2024 per l'Italia: ~0.233 kgCO2e/kWh
"""
async with ClimatiqClient() as client:
activity = ActivityData(
activity_id="electricity-supply_grid-source_supplier_mix",
data={
"energy": {
"value": kwh_consumed,
"energy_unit": "kWh"
}
},
region="IT", # Italia
year=2024
)
estimate = await client.estimate_emission(activity)
# Converti kgCO2e in tonnellate CO2e
return estimate.co2e / 1000 if estimate.co2e_unit == "kg" else estimate.co2eKarbon Muhasebesi için Mikro Hizmet Mimarisi
Bir kurumsal karbon muhasebesi platformunun heterojen veri akışlarını yönetmesi gerekir (enerji faturaları, ERP verileri, gider raporları, tedarikçi verileri), karmaşık ve denetlenebilir hesaplamalar, ve farklı formatlarda raporlar oluşturmak. Mikro hizmet mimarisi ve seçimi bu gereksinimler için doğaldır.
Dört Temel Hizmet
1. Veri Toplama Hizmeti
Farklı kaynaklardan veri toplanmasından sorumlu: ERP API (SAP, Oracle), hizmet faturaları OCR/ayrıştırıcı, seyahat yönetimi araçları (Concur, TravelPerk), satın alma verileri aracılığıyla, CSV manuel yüklemeleri, tedarikçi API'leri. Uç noktaları alıma sunar ve yönetir ölçü birimlerinin normalleştirilmesi.
2. Hesaplama Motoru
Hesaplamalı kalp: normalleştirilmiş aktiviteleri alır, emisyon faktörlerini seçer uygun (yerel veya Climatiq API aracılığıyla), Sera Gazı Protokolü formüllerini uygular, üretir Denetim için açık hesaplama formülü ile değişmez sonuçlar. Destek Faktörler güncellendiğinde tarihsel yeniden hesaplamalar.
3. Raporlama Hizmeti
Çerçevelerin gerektirdiği formatlarda raporlar oluşturun: CSRD/ESRS (XBRL etiketlemeyle), CDP anketi (JSON/XML formatı), GRI Standartları açıklamaları, dahili raporlar kontrol panelleri için. Güvence için rapor sürüm oluşturmayı ve dijital imzayı yönetir.
4. Denetim Takip Hizmeti
Her işlemin değişmez bir kaydını tutar: verileri kim girdi, hangi faktör Hesaplandığında kimin onayladığı kullanıldı. Veri kökenini destekler tamamlandı: kaynak verilerden nihai rapordaki sayıya kadar. Güvence için gerekli denetçiler tarafından.
Mimari Şema
┌─────────────────────────────────────────────────────────────────┐
│ CARBON ACCOUNTING PLATFORM │
├─────────────────────────────────────────────────────────────────┤
│ FRONTEND │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐ │
│ │ Dashboard │ │ Data Input │ │ Report Generator │ │
│ │ (Angular) │ │ Wizard │ │ (CSRD/CDP/GRI) │ │
│ └──────┬──────┘ └──────┬──────┘ └─────────────┬───────────┘ │
├─────────┼────────────────┼─────────────────────────┼────────────┤
│ API GATEWAY (FastAPI/Kong) │
│ │ │ │ │
├─────────┼────────────────┼─────────────────────────┼────────────┤
│ MICROSERVICES │
│ ┌──────▼──────┐ ┌──────▼──────┐ ┌──────────────▼─────────┐ │
│ │ Data │ │ Calculation │ │ Reporting │ │
│ │ Collection │→ │ Engine │→ │ Service │ │
│ │ Service │ │ (Python) │ │ (PDF/XBRL/JSON) │ │
│ └──────┬──────┘ └──────┬──────┘ └────────────────────────┘ │
│ │ │ │
│ ┌──────▼──────┐ ┌──────▼──────┐ ┌────────────────────────┐ │
│ │ SAP │ │ Climatiq │ │ Audit Trail │ │
│ │ Connector │ │ API Client │ │ Service │ │
│ └─────────────┘ └─────────────┘ └────────────────────────┘ │
├─────────────────────────────────────────────────────────────────┤
│ DATA LAYER │
│ ┌──────────────┐ ┌─────────────┐ ┌────────────────────────┐ │
│ │ PostgreSQL │ │ TimeSeries │ │ Object Storage │ │
│ │ (Core data) │ │ (InfluxDB) │ │ (S3 - documents) │ │
│ └──────────────┘ └─────────────┘ └────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
INTEGRAZIONI ESTERNE:
- SAP S/4HANA (procurement, energy data)
- Oracle NetSuite (financials per spend-based)
- TravelPerk/Concur (travel data - Cat.6)
- Climatiq API (emission factors)
- AIB Registry (GO certificates - Scope 2 market-based)
- Utility providers API (bollette energia)
Hesaplama Motoru: FastAPI ile Python uygulaması
Hesaplama motoru ve en kritik bileşen. Kesin olmalı, denetlenebilir olmalı, versiyonlanmış ve binlerce hesaplamayı paralel olarak gerçekleştirebilme kapasitesine sahiptir. İşte bir uygulama FastAPI ile tamamlandı.
# calculation_engine/main.py - FastAPI Calculation Engine
from fastapi import FastAPI, HTTPException, BackgroundTasks, Depends
from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetime
from enum import Enum
import pandas as pd
import uuid
import asyncio
from decimal import Decimal, ROUND_HALF_UP
app = FastAPI(
title="Carbon Calculation Engine",
description="GHG Protocol-compliant emission calculation service",
version="2.1.0"
)
# ── Pydantic Models ──────────────────────────────────────────────
class ScopeEnum(str, Enum):
scope_1 = "scope_1"
scope_2_location = "scope_2_location"
scope_2_market = "scope_2_market"
scope_3 = "scope_3"
class CalculationMethod(str, Enum):
activity_based = "activity_based" # quantità * fattore emissione
spend_based = "spend_based" # spesa in EUR * fattore intensità
average_data = "average_data" # media su periodi
supplier_specific = "supplier_specific" # dati diretti da fornitori
class ActivityInput(BaseModel):
activity_id: str = Field(..., description="ID univoco attivita sorgente")
emission_source_id: str
scope: ScopeEnum
scope_3_category: Optional[str] = None
period_start: datetime
period_end: datetime
quantity: float = Field(..., gt=0, description="Quantità attivita")
unit: str = Field(..., description="Unita di misura (kWh, liter, km, tonne, EUR)")
emission_factor_id: Optional[str] = None # se None, il motore cerca automaticamente
calculation_method: CalculationMethod = CalculationMethod.activity_based
region: Optional[str] = None
data_quality: str = "measured"
@validator("quantity")
def quantity_must_be_positive(cls, v):
if v <= 0:
raise ValueError("La quantità deve essere positiva")
return v
class CalculationResult(BaseModel):
calculation_id: str
activity_id: str
scope: ScopeEnum
co2e_tonnes: float
co2_tonnes: Optional[float]
ch4_tonnes: Optional[float]
n2o_tonnes: Optional[float]
emission_factor_value: float
emission_factor_unit: str
emission_factor_source: str
emission_factor_year: int
calculation_formula: str
calculation_method: str
uncertainty_percentage: Optional[float]
calculated_at: datetime
audit_reference: str
class BatchCalculationRequest(BaseModel):
organization_id: str
reporting_year: int
activities: List[ActivityInput] = Field(..., max_items=500)
class ScopeAggregation(BaseModel):
scope_1_tonnes: float
scope_2_location_tonnes: float
scope_2_market_tonnes: float
scope_3_tonnes: float
scope_3_by_category: dict
total_location_based: float
total_market_based: float
reporting_year: int
organization_id: str
# ── Calculation Logic ────────────────────────────────────────────
class EmissionCalculator:
"""
Implementazione GHG Protocol Corporate Standard.
Formula base: Emissioni (kgCO2e) = Attivita x Fattore di Emissione
"""
# Fattori GWP AR6 (IPCC Sixth Assessment Report, 2021)
GWP_AR6 = {
"CO2": 1.0,
"CH4": 27.9, # CH4 fossil
"CH4_bio": 27.9,
"N2O": 273.0,
"SF6": 25200.0,
"NF3": 17400.0,
"HFC134a": 1526.0,
"HFC32": 771.0,
}
# Fattori di incertezza per qualità dato (DEFRA methodology)
UNCERTAINTY_BY_DATA_QUALITY = {
"measured": 5.0, # dati misurati diretti
"calculated": 10.0, # calcolati da misure
"estimated": 20.0, # stime con dati proxy
"spend_based": 35.0, # spend-based (meno preciso)
"default": 25.0,
}
def calculate_activity_based(
self,
quantity: float,
unit: str,
emission_factor: dict,
gwp_version: str = "AR6"
) -> dict:
"""
Calcolo activity-based:
CO2e (kg) = Quantità (unita) x EF (kgCO2e/unita)
"""
ef_value = emission_factor["co2e_factor"]
ef_unit = emission_factor["unit_type"]
# Verifica compatibilità unita
if not self._units_compatible(unit, ef_unit):
raise ValueError(
f"Unita incompatibili: attivita in {unit}, "
f"fattore in {ef_unit}"
)
co2e_kg = quantity * ef_value
# Calcola componenti separate se disponibili
co2_kg = quantity * emission_factor.get("co2_factor", 0)
ch4_kg = quantity * emission_factor.get("ch4_factor", 0)
n2o_kg = quantity * emission_factor.get("n2o_factor", 0)
return {
"co2e_tonnes": round(co2e_kg / 1000, 6),
"co2_tonnes": round(co2_kg / 1000, 6) if co2_kg else None,
"ch4_tonnes": round(ch4_kg / 1000, 6) if ch4_kg else None,
"n2o_tonnes": round(n2o_kg / 1000, 6) if n2o_kg else None,
"formula": (
f"{quantity} {unit} x {ef_value} kgCO2e/{unit} "
f"= {co2e_kg:.4f} kgCO2e "
f"= {co2e_kg/1000:.6f} tCO2e"
)
}
def calculate_spend_based(
self,
spend_eur: float,
emission_intensity: float, # kgCO2e/EUR
currency: str = "EUR",
exchange_rate: float = 1.0
) -> dict:
"""
Calcolo spend-based (Scope 3 Cat.1 quando non si hanno dati attivita):
CO2e (kg) = Spesa (EUR) x Intensità (kgCO2e/EUR)
"""
spend_normalized = spend_eur * exchange_rate
co2e_kg = spend_normalized * emission_intensity
return {
"co2e_tonnes": round(co2e_kg / 1000, 6),
"co2_tonnes": None,
"ch4_tonnes": None,
"n2o_tonnes": None,
"formula": (
f"{spend_normalized:.2f} EUR x {emission_intensity} kgCO2e/EUR "
f"= {co2e_kg:.4f} kgCO2e"
)
}
def _units_compatible(self, activity_unit: str, ef_unit: str) -> bool:
"""Verifica compatibilità unita di misura"""
# Normalizza: il fattore e tipicamente espresso come kgCO2e/[unita_attivita]
ef_denominator = ef_unit.split("/")[-1].strip().lower() if "/" in ef_unit else ef_unit
return activity_unit.lower() == ef_denominator
def aggregate_by_scope(
self,
calculations: List[CalculationResult]
) -> ScopeAggregation:
"""Aggrega i calcoli per scope - usa pandas per performance"""
df = pd.DataFrame([c.dict() for c in calculations])
scope_1 = df[df["scope"] == "scope_1"]["co2e_tonnes"].sum()
scope_2_loc = df[df["scope"] == "scope_2_location"]["co2e_tonnes"].sum()
scope_2_mkt = df[df["scope"] == "scope_2_market"]["co2e_tonnes"].sum()
scope_3_df = df[df["scope"] == "scope_3"]
scope_3_total = scope_3_df["co2e_tonnes"].sum()
# Aggregazione Scope 3 per categoria
scope_3_by_cat = {}
if not scope_3_df.empty and "scope_3_category" in scope_3_df.columns:
scope_3_by_cat = (
scope_3_df.groupby("scope_3_category")["co2e_tonnes"]
.sum()
.to_dict()
)
return ScopeAggregation(
scope_1_tonnes=round(scope_1, 3),
scope_2_location_tonnes=round(scope_2_loc, 3),
scope_2_market_tonnes=round(scope_2_mkt, 3),
scope_3_tonnes=round(scope_3_total, 3),
scope_3_by_category=scope_3_by_cat,
total_location_based=round(scope_1 + scope_2_loc + scope_3_total, 3),
total_market_based=round(scope_1 + scope_2_mkt + scope_3_total, 3),
reporting_year=2024,
organization_id=""
)
calculator = EmissionCalculator()
# ── API Endpoints ────────────────────────────────────────────────
@app.post("/v1/calculate/single", response_model=CalculationResult)
async def calculate_single(activity: ActivityInput):
"""Calcola emissioni per una singola attivita"""
# In produzione: cerca emission factor dal DB o Climatiq
# Qui mock per illustrazione
mock_ef = {
"co2e_factor": 0.233, # kgCO2e/kWh - rete italiana 2024
"co2_factor": 0.228,
"ch4_factor": 0.002,
"n2o_factor": 0.001,
"unit_type": "kgCO2e/kWh",
"source": "IEA 2024",
"year": 2024
}
result = calculator.calculate_activity_based(
quantity=activity.quantity,
unit=activity.unit,
emission_factor=mock_ef
)
calculation_id = str(uuid.uuid4())
uncertainty = calculator.UNCERTAINTY_BY_DATA_QUALITY.get(
activity.data_quality, 25.0
)
return CalculationResult(
calculation_id=calculation_id,
activity_id=activity.activity_id,
scope=activity.scope,
co2e_tonnes=result["co2e_tonnes"],
co2_tonnes=result["co2_tonnes"],
ch4_tonnes=result["ch4_tonnes"],
n2o_tonnes=result["n2o_tonnes"],
emission_factor_value=mock_ef["co2e_factor"],
emission_factor_unit=mock_ef["unit_type"],
emission_factor_source=mock_ef["source"],
emission_factor_year=mock_ef["year"],
calculation_formula=result["formula"],
calculation_method=activity.calculation_method.value,
uncertainty_percentage=uncertainty,
calculated_at=datetime.utcnow(),
audit_reference=f"CALC-{calculation_id[:8].upper()}"
)
@app.post("/v1/calculate/batch", response_model=List[CalculationResult])
async def calculate_batch(request: BatchCalculationRequest):
"""Calcola emissioni per un batch di attivita (max 500)"""
tasks = [calculate_single(activity) for activity in request.activities]
results = await asyncio.gather(*tasks, return_exceptions=True)
successful = [r for r in results if isinstance(r, CalculationResult)]
failed = [r for r in results if isinstance(r, Exception)]
if failed:
# Log errori ma non blocca il batch
pass
return successful
@app.get("/v1/organizations/{org_id}/summary", response_model=ScopeAggregation)
async def get_emissions_summary(org_id: str, year: int = 2024):
"""Riepilogo emissioni per scope per un'organizzazione"""
# In produzione: query dal DB
passKapsam 3 otomasyonu: ERP, Tedarik ve Seyahat
Kapsam 3 ve karbon muhasebesinin en büyük zorluğu: düzinelerce alana dağıtılan veriler sistemler, farklı olgunluk seviyelerindeki tedarikçiler, farklı hesaplama metodolojileri kategori. Otomasyon, onu operasyonel açıdan sürdürülebilir kılmanın tek yoludur.
Kapsam 3 Kategori 1 (Satın Alınan Mallar) için SAP entegrasyonu
# integrations/sap_connector.py - Estrazione dati procurement da SAP
import httpx
from dataclasses import dataclass
from typing import List, Optional
from datetime import date
import pandas as pd
@dataclass
class ProcurementRecord:
purchase_order_id: str
vendor_id: str
vendor_name: str
vendor_country: str
material_code: str
material_description: str
quantity: float
unit: str
amount_eur: float
nace_code: Optional[str] # classificazione settore fornitore
delivery_date: date
class SAPConnector:
"""
Connettore SAP S/4HANA via OData API.
Estrae dati procurement per Scope 3 Cat.1 e Cat.4.
"""
def __init__(self, base_url: str, client_id: str, client_secret: str):
self.base_url = base_url
self.client_id = client_id
self.client_secret = client_secret
self._token: Optional[str] = None
async def authenticate(self):
"""OAuth 2.0 client credentials per SAP"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/oauth/token",
data={
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret,
}
)
response.raise_for_status()
self._token = response.json()["access_token"]
async def get_purchase_orders(
self,
year: int,
cost_center: Optional[str] = None
) -> List[ProcurementRecord]:
"""
Recupera ordini di acquisto dal modulo MM di SAP.
Endpoint OData: /sap/opu/odata/sap/MM_PUR_PO_MANAGE_SRV/
"""
if not self._token:
await self.authenticate()
params = {
"$filter": f"PostingDate ge datetime'{year}-01-01T00:00:00' and "
f"PostingDate le datetime'{year}-12-31T23:59:59'",
"$select": "PurchaseOrder,Supplier,SupplierName,OrderedQuantity,"
"PurchaseOrderQuantityUnit,NetPriceAmount,Currency",
"$format": "json",
"$top": 5000
}
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.base_url}/sap/opu/odata/sap/MM_PUR_PO_MANAGE_SRV/"
"A_PurchaseOrder",
headers={"Authorization": f"Bearer {self._token}"},
params=params
)
response.raise_for_status()
data = response.json()["d"]["results"]
return [
ProcurementRecord(
purchase_order_id=item["PurchaseOrder"],
vendor_id=item["Supplier"],
vendor_name=item["SupplierName"],
vendor_country="IT", # da arricchire con master data
material_code=item.get("Material", ""),
material_description=item.get("MaterialName", ""),
quantity=float(item["OrderedQuantity"]),
unit=item["PurchaseOrderQuantityUnit"],
amount_eur=float(item["NetPriceAmount"]),
nace_code=None, # da mappare con codifica NACE
delivery_date=date.fromisoformat(item.get("ScheduleLine", f"{year}-12-31")[:10])
)
for item in data
]
def calculate_scope3_cat1_spend_based(
self,
records: List[ProcurementRecord],
emission_intensities: dict # { nace_code: kgCO2e_per_EUR }
) -> pd.DataFrame:
"""
Calcolo Scope 3 Cat.1 con metodo spend-based.
Da usare quando non si hanno dati attivita dai fornitori.
Fattori da EXIOBASE o WIOD (World Input-Output Database).
"""
df = pd.DataFrame([vars(r) for r in records])
# Mappa NACE -> intensità emissioni
default_intensity = emission_intensities.get("DEFAULT", 0.35) # kgCO2e/EUR
df["emission_intensity"] = df["nace_code"].map(emission_intensities).fillna(default_intensity)
df["co2e_kg"] = df["amount_eur"] * df["emission_intensity"]
df["co2e_tonnes"] = df["co2e_kg"] / 1000
# Aggregazione per fornitore
summary = df.groupby(["vendor_id", "vendor_name"]).agg(
total_spend_eur=("amount_eur", "sum"),
total_co2e_tonnes=("co2e_tonnes", "sum"),
transaction_count=("purchase_order_id", "count")
).reset_index()
summary["avg_intensity"] = summary["total_co2e_tonnes"] * 1000 / summary["total_spend_eur"]
return summary.sort_values("total_co2e_tonnes", ascending=False)
# integrations/travel_connector.py - Dati viaggi per Scope 3 Cat.6
@dataclass
class TravelRecord:
employee_id: str
travel_date: date
origin_iata: str # codice aeroporto IATA
destination_iata: str
transport_mode: str # air, rail, car, ferry
distance_km: float
travel_class: str # economy, business, first
booking_amount_eur: float
class TravelDataProcessor:
"""Processa dati di business travel per Scope 3 Cat.6"""
# Fattori emissione voli (kgCO2e/pkm) - DEFRA 2024
DEFRA_FLIGHT_FACTORS = {
("short_haul", "economy"): 0.151,
("short_haul", "business"): 0.227,
("medium_haul", "economy"): 0.131,
("medium_haul", "business"): 0.262,
("long_haul", "economy"): 0.195,
("long_haul", "business"): 0.585,
("long_haul", "first"): 0.780,
}
# Fattori treno (kgCO2e/pkm) - media europea
RAIL_FACTORS = {
"IT": 0.004, # Trenitalia (alta % rinnovabili)
"DE": 0.006,
"FR": 0.002,
"DEFAULT": 0.041,
}
def classify_flight(self, distance_km: float) -> str:
if distance_km < 1500:
return "short_haul"
elif distance_km < 4000:
return "medium_haul"
else:
return "long_haul"
def calculate_flight_emissions(self, record: TravelRecord) -> float:
"""
Calcolo emissioni volo con Radiative Forcing Index (RFI = 1.9)
per effetti non-CO2 ad alta quota (metodo DEFRA con uplift factor)
"""
haul = self.classify_flight(record.distance_km)
travel_class = record.travel_class.lower() if record.travel_class else "economy"
key = (haul, travel_class)
base_factor = self.DEFRA_FLIGHT_FACTORS.get(
key,
self.DEFRA_FLIGHT_FACTORS[(haul, "economy")]
)
# Applica Radiative Forcing (RFI) per effetti non-CO2 in alta quota
rfi_factor = 1.9
co2e_kg = record.distance_km * base_factor * rfi_factor
return co2e_kg / 1000 # in tonnellate
def process_travel_data(self, records: List[TravelRecord]) -> dict:
"""Elabora tutti i dati travel e restituisce riepilogo per Cat.6"""
results = []
for record in records:
if record.transport_mode == "air":
co2e = self.calculate_flight_emissions(record)
method = "DEFRA 2024 with RFI=1.9"
elif record.transport_mode == "rail":
country = record.origin_iata[:2] # approssimazione
factor = self.RAIL_FACTORS.get(country, self.RAIL_FACTORS["DEFAULT"])
co2e = record.distance_km * factor / 1000
method = f"Rail factor {country}"
else:
co2e = record.distance_km * 0.171 / 1000 # car average
method = "DEFRA car average"
results.append({
"employee_id": record.employee_id,
"travel_date": record.travel_date,
"transport_mode": record.transport_mode,
"distance_km": record.distance_km,
"co2e_tonnes": co2e,
"method": method
})
df = pd.DataFrame(results)
return {
"total_co2e_tonnes": df["co2e_tonnes"].sum(),
"by_mode": df.groupby("transport_mode")["co2e_tonnes"].sum().to_dict(),
"by_month": df.groupby(df["travel_date"].apply(lambda x: x.month))["co2e_tonnes"].sum().to_dict(),
"total_km": df["distance_km"].sum(),
"record_count": len(df)
}Otomatik Raporlama: CSRD/ESRS, CDP ve GRI
Düzenleyici raporların oluşturulması genellikle karbon muhasebesinin en çok zaman alan sürecidir. Otomasyon, süreyi önemli ölçüde haftalardan saatlere indirir, tutarlılığı artırır ve her rakamın kaynağına kadar izlenebilir olmasını sağlar.
CSRD/ESRS E1 Rapor Yapısı (İklim Değişikliği)
ESRS E1 iklim standardı aşağıdaki konularda açıklama yapılmasını gerektirir: iklim yönetişimi, strateji ve senaryo analizi, risk yönetimi, ölçümler ve hedefler. Metrikler Emisyonların miktarı ESRS E1-6'da tanımlanmıştır ve her üç kapsam için de veri gerektirir.
# reporting/csrd_generator.py - Generazione report CSRD/ESRS E1
from dataclasses import dataclass, asdict
from typing import List, Optional, Dict
from datetime import datetime
import json
@dataclass
class ESRS_E1_6_Disclosure:
"""
ESRS E1-6: Gross Scopes 1, 2 and 3 greenhouse gas emissions
Disclosure requirements per ESRS E1 (climate change)
"""
organization_name: str
legal_entity_identifier: str # LEI
reporting_period: str # es. "FY2024"
reporting_standard: str = "ESRS E1 - Climate Change"
# Scope 1 - Emissioni dirette
scope_1_total_gross_tco2e: float = 0.0
scope_1_breakdown_by_ghg: Dict[str, float] = None # { "CO2": x, "CH4": y ... }
scope_1_breakdown_by_source: Dict[str, float] = None
# Scope 2 - Emissioni indirette energia
scope_2_location_based_tco2e: float = 0.0
scope_2_market_based_tco2e: float = 0.0
scope_2_purchased_electricity_kwh: float = 0.0
scope_2_renewable_electricity_percentage: float = 0.0
# Scope 3 - Emissioni catena del valore
scope_3_total_tco2e: float = 0.0
scope_3_upstream_tco2e: float = 0.0
scope_3_downstream_tco2e: float = 0.0
scope_3_by_category: Dict[str, float] = None
# Intensità emissioni
revenue_intensity_tco2e_per_meur: Optional[float] = None # tCO2e/M EUR
employee_intensity_tco2e_per_fte: Optional[float] = None
# Confronto anno precedente e target
base_year: int = 2020
scope_1_2_reduction_vs_base: Optional[float] = None # percentuale
sbti_target: Optional[str] = None # es. "1.5°C aligned, -42% by 2030"
# Metodologia
ghg_accounting_standard: str = "GHG Protocol Corporate Standard"
emission_factor_sources: List[str] = None
data_quality_notes: str = ""
assurance_level: str = "limited_assurance"
assurance_provider: str = ""
def __post_init__(self):
if self.scope_3_by_category is None:
self.scope_3_by_category = {}
if self.scope_1_breakdown_by_ghg is None:
self.scope_1_breakdown_by_ghg = {}
if self.emission_factor_sources is None:
self.emission_factor_sources = []
@property
def total_ghg_location_based(self) -> float:
return (self.scope_1_total_gross_tco2e +
self.scope_2_location_based_tco2e +
self.scope_3_total_tco2e)
@property
def total_ghg_market_based(self) -> float:
return (self.scope_1_total_gross_tco2e +
self.scope_2_market_based_tco2e +
self.scope_3_total_tco2e)
class CSRDReportGenerator:
def generate_esrs_e1_json(self, disclosure: ESRS_E1_6_Disclosure) -> str:
"""Genera disclosure ESRS E1-6 in formato JSON strutturato"""
report = {
"metadata": {
"standard": disclosure.reporting_standard,
"generated_at": datetime.utcnow().isoformat(),
"reporting_period": disclosure.reporting_period,
"organization": disclosure.organization_name,
"lei": disclosure.legal_entity_identifier,
},
"ESRS_E1-6": {
"gross_scope_1_tco2e": disclosure.scope_1_total_gross_tco2e,
"scope_1_breakdown_by_ghg": disclosure.scope_1_breakdown_by_ghg,
"scope_2_location_based_tco2e": disclosure.scope_2_location_based_tco2e,
"scope_2_market_based_tco2e": disclosure.scope_2_market_based_tco2e,
"scope_2_purchased_electricity_kwh": disclosure.scope_2_purchased_electricity_kwh,
"scope_2_renewable_pct": disclosure.scope_2_renewable_electricity_percentage,
"scope_3_total_tco2e": disclosure.scope_3_total_tco2e,
"scope_3_by_category": disclosure.scope_3_by_category,
"total_ghg_location_based_tco2e": disclosure.total_ghg_location_based,
"total_ghg_market_based_tco2e": disclosure.total_ghg_market_based,
"ghg_intensity_revenue": disclosure.revenue_intensity_tco2e_per_meur,
"ghg_intensity_employee": disclosure.employee_intensity_tco2e_per_fte,
},
"methodology": {
"accounting_standard": disclosure.ghg_accounting_standard,
"emission_factor_sources": disclosure.emission_factor_sources,
"base_year": disclosure.base_year,
"consolidation_approach": "operational_control",
"data_quality": disclosure.data_quality_notes,
},
"assurance": {
"level": disclosure.assurance_level,
"provider": disclosure.assurance_provider,
},
"targets": {
"sbti_commitment": disclosure.sbti_target,
"scope_1_2_reduction_vs_base": disclosure.scope_1_2_reduction_vs_base,
}
}
return json.dumps(report, indent=2, ensure_ascii=False)
def generate_cdp_questionnaire_c6(self, disclosure: ESRS_E1_6_Disclosure) -> dict:
"""
Genera risposte per CDP questionnaire - sezione C6 (Emissions Data).
CDP condivide molti requisiti con CSRD, il che riduce il double reporting.
"""
return {
"C6.1": {
"question": "Provide total gross global Scope 1 emissions in metric tons CO2e",
"response": disclosure.scope_1_total_gross_tco2e,
"unit": "metric tons CO2e"
},
"C6.2": {
"question": "Describe Scope 1 emissions by constituent gases",
"response": disclosure.scope_1_breakdown_by_ghg
},
"C6.3": {
"question": "Provide total gross global Scope 2 emissions",
"response": {
"location_based": disclosure.scope_2_location_based_tco2e,
"market_based": disclosure.scope_2_market_based_tco2e,
}
},
"C6.5": {
"question": "Account for Scope 3 emissions",
"response": {
"total": disclosure.scope_3_total_tco2e,
"categories": disclosure.scope_3_by_category
}
},
"C6.10": {
"question": "Describe Scope 1 and 2 GHG emissions by location",
"note": "See facility-level breakdown in supporting data"
}
}Denetim İzi ve Veri Bağlantısı
CSRD'nin nitelikli bir denetçiden sınırlı güvence talep etmesi nedeniyle, denetim takibi isteğe bağlı bir gereklilik: ve platformun omurgası. Raporun her sayısında açık hesaplama formülüyle orijinal kaynağa kadar izlenebilir olmalıdır.
# audit/audit_trail.py - Sistema di audit immutabile
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Dict, Optional
import hashlib
import json
import uuid
@dataclass(frozen=True) # immutabile
class AuditEvent:
"""
Evento di audit immutabile. Non si modifica mai, solo si aggiungono nuovi eventi.
Il hash SHA256 garantisce l'integrita della catena.
"""
event_id: str
event_type: str # DATA_INGESTED, FACTOR_SELECTED, CALCULATION_PERFORMED, REPORT_GENERATED, APPROVED
entity_type: str # Activity, Calculation, EmissionFactor, Report
entity_id: str
actor_id: str # user_id o service_name per azioni automatiche
actor_type: str # human, system
timestamp: str # ISO 8601 UTC
data_before: Optional[str] # JSON snapshot before change
data_after: Optional[str] # JSON snapshot after change
metadata: str # JSON { formula, ef_source, notes ... }
previous_hash: str # hash dell'evento precedente (blockchain-like)
event_hash: str # SHA256 di tutti i campi
@classmethod
def create(
cls,
event_type: str,
entity_type: str,
entity_id: str,
actor_id: str,
actor_type: str = "human",
data_before: Optional[Dict] = None,
data_after: Optional[Dict] = None,
metadata: Optional[Dict] = None,
previous_hash: str = "GENESIS"
) -> "AuditEvent":
event_id = str(uuid.uuid4())
timestamp = datetime.utcnow().isoformat() + "Z"
data_before_str = json.dumps(data_before, sort_keys=True) if data_before else None
data_after_str = json.dumps(data_after, sort_keys=True) if data_after else None
metadata_str = json.dumps(metadata or {}, sort_keys=True)
# Calcola hash dell'evento per garanzia di integrita
hash_content = "|".join([
event_id, event_type, entity_type, entity_id,
actor_id, timestamp, data_after_str or "", previous_hash
])
event_hash = hashlib.sha256(hash_content.encode()).hexdigest()
return cls(
event_id=event_id,
event_type=event_type,
entity_type=entity_type,
entity_id=entity_id,
actor_id=actor_id,
actor_type=actor_type,
timestamp=timestamp,
data_before=data_before_str,
data_after=data_after_str,
metadata=metadata_str,
previous_hash=previous_hash,
event_hash=event_hash
)
class DataLineageTracker:
"""
Traccia il percorso completo di un dato dalla sorgente al report.
Permette ai revisori di verificare ogni numero del report finale.
"""
def __init__(self, db_session):
self.db = db_session
def trace_calculation(self, calculation_id: str) -> dict:
"""
Ricostruisce il lineage completo di un calcolo:
Report -> Calculation -> Activity -> EmissionFactor -> RawData -> Source System
"""
# In produzione: query al DB per tutti gli eventi correlati
lineage = {
"calculation_id": calculation_id,
"trace": [
{
"step": 1,
"description": "Raw data ingested from SAP S/4HANA",
"system": "SAP_CONNECTOR",
"timestamp": "2025-01-15T08:30:00Z",
"data_summary": "Natural gas consumption: 5,420 m3, Jan 2024",
"audit_event_id": "AE-001"
},
{
"step": 2,
"description": "Unit conversion applied: m3 -> kWh",
"formula": "5,420 m3 x 10.55 kWh/m3 = 57,181 kWh",
"timestamp": "2025-01-15T08:31:00Z",
"audit_event_id": "AE-002"
},
{
"step": 3,
"description": "Emission factor selected from DEFRA 2024",
"emission_factor": "Natural gas - Gross CV: 0.18306 kgCO2e/kWh",
"factor_id": "DEFRA_2024_NG_GROSS",
"timestamp": "2025-01-15T08:31:05Z",
"audit_event_id": "AE-003"
},
{
"step": 4,
"description": "Emission calculation performed",
"formula": "57,181 kWh x 0.18306 kgCO2e/kWh = 10,467 kgCO2e = 10.467 tCO2e",
"result_tco2e": 10.467,
"scope": "Scope 1",
"timestamp": "2025-01-15T08:31:06Z",
"audit_event_id": "AE-004"
},
{
"step": 5,
"description": "Calculation included in FY2024 CSRD Report - ESRS E1-6",
"report_id": "RPT-CSRD-2024-001",
"approved_by": "CFO - Mario Rossi",
"timestamp": "2025-03-10T14:00:00Z",
"audit_event_id": "AE-089"
}
],
"data_quality_flag": "HIGH",
"uncertainty_pct": 5.0,
"verification_status": "verified_by_auditor"
}
return lineageKurumsal Platform Karşılaştırması: Oluştur vs Satın Al
Özel bir platform oluşturmadan önce çözümleri değerlendirmek önemlidir. işletme mevcuttur. Karbon muhasebesi yazılım pazarı oldukça olgun ve önde gelen platformlar standart kullanım durumlarının çoğunu kapsar.
| platformu | Güç noktası | Hedef Sektör | Gösterge fiyatı | CSRD'ye Hazır |
|---|---|---|---|---|
| Persephones | Yatırımcı düzeyinde raporlama, XBRL etiketleme, SEC odaklı | Finans, Kurumsal | 50.000-500.000$/yıl | Si |
| Havza | Raporlama hızı, gelişmiş Kapsam 3 araçları | Teknoloji, Kurumsal | 100 bin-1 milyon dolar/yıl | Si |
| Küre | Entegre LCA, endüstriyel uyumluluk, risk yönetimi | İmalat, Enerji, Kimya | Talep üzerine | Si |
| A Planı | Basit UX, hızlı uygulama | KOBİ'ler, orta ölçekli işletmeler | 10.000-100.000$/yıl | Kısmi |
| IBM Envizi | 40.000'den fazla emisyon faktörü, ERP entegrasyonu | Kurumsal, Yardımcı Programlar | Talep üzerine | Si |
| Özel Yapı | Tam esneklik, dahili sistemlerin yerel entegrasyonu | Özel gereksinimleri olan büyük şirketler | 500 bin-5 milyon dolarlık geliştirme | Bağlı olmak |
Ne Zaman Özel Oluşturulmalı ve Ne Zaman Satın Almalı?
Aşağıdaki durumlarda bir platform satın alın: standart gereksinimleriniz var, canlı yayına geçmek istiyorsunuz 3-6 ay içinde sürdürülebilirliğe adanmış bir mühendislik ekibiniz kalmayacak, veri hacmi artacak ve yönetilebilir.
Aşağıdaki durumlarda özel oluşturun: veriler içeren çok spesifik üretim süreçleriniz var Standart platformlar tarafından yönetilemiyor, eski sistemlerle yerel olarak entegre olmak istiyorsunuz, SaaS ile uyumlu olmayan veri yerleşimi gereksinimleriniz veya veri hacminiz varsa ve milyonlarca kayıt/yıl (SaaS maliyetleri fahiş hale gelir).
Örnek Olay İncelemesi: CSRD 2025 ile İtalyan İmalat KOBİ'leri
1.200 çalışana sahip bir İtalyan hassas mühendislik üretim şirketi, 350 milyon Euro cirosu ve Torino, Milano ve Brescia'daki fabrikaları bulunuyor. CSRD'nin 1. Dalgasındadır ve 2024 mali yılı için ilk raporu sunmalıdır Haziran 2025'e kadar.
Emisyon Envanteri 2024 Mali Yılı
# case_study/inventory_2024.py - Esempio inventario GHG realistico
# Dati rappresentativi per PMI manifatturiera italiana
INVENTORY_2024 = {
"organization": "MeccanicaPrecisione SpA",
"reporting_year": 2024,
"boundary": "Operational Control",
"currency": "EUR",
"scope_1": {
"natural_gas_combustion": {
"consumption_m3": 485_000, # caldaie industriali
"consumption_kwh": 5_116_750, # conversione: 1 m3 NG = 10.55 kWh
"emission_factor_kgco2e_kwh": 0.18306, # DEFRA 2024
"co2e_tonnes": 937.0,
"source": "DEFRA 2024 - Natural Gas Gross CV"
},
"diesel_mobile": {
"consumption_liters": 42_000, # carrelli elevatori, mezzi operativi
"emission_factor_kgco2e_liter": 2.56, # DEFRA 2024
"co2e_tonnes": 107.5,
"source": "DEFRA 2024 - Diesel"
},
"fugitive_refrigerants": {
"substance": "R-410A",
"kg_recharged": 45,
"gwp_ar6": 2088,
"co2e_tonnes": 94.0,
"source": "IPCC AR6 GWP100"
},
"total_co2e_tonnes": 1138.5
},
"scope_2": {
"purchased_electricity": {
"consumption_kwh": 8_250_000,
"location_based": {
"emission_factor": 0.233, # IEA Italy 2024 kgCO2e/kWh
"co2e_tonnes": 1922.3,
"source": "IEA 2024 Italy Grid"
},
"market_based": {
"go_certificates_kwh": 4_125_000, # 50% rinnovabili con GO
"residual_mix_factor": 0.395, # AIB Italy residual mix 2024
"co2e_tonnes": 1629.4, # solo su 50% non coperto da GO
"source": "AIB Italy Residual Mix 2024"
}
}
},
"scope_3": {
"cat_1_purchased_goods": {
"total_spend_eur": 145_000_000,
"method": "spend_based + supplier_specific (top 20 suppliers)",
"co2e_tonnes": 28_450.0,
"data_quality": "mix: 35% supplier-specific, 65% spend-based"
},
"cat_4_upstream_transport": {
"tonne_km": 3_200_000,
"emission_factor": 0.089, # kgCO2e/tonne-km road freight DEFRA
"co2e_tonnes": 284.8,
},
"cat_5_waste": {
"waste_tonnes": 1_850,
"co2e_tonnes": 185.0,
"method": "waste-type specific factors"
},
"cat_6_business_travel": {
"total_km_air": 1_250_000,
"total_km_rail": 320_000,
"co2e_tonnes": 312.5,
"source": "DEFRA 2024 with RFI=1.9"
},
"cat_7_employee_commuting": {
"employees": 1_200,
"avg_km_per_day": 22,
"working_days": 220,
"mode_split": {"car_solo": 0.65, "car_sharing": 0.05,
"public_transport": 0.25, "cycling": 0.05},
"co2e_tonnes": 892.5
},
"cat_11_use_of_products": {
"units_sold": 45_000,
"avg_energy_use_kwh_per_unit_per_year": 850,
"product_lifetime_years": 15,
"co2e_tonnes_per_year": 2_250.0,
"note": "Calcolato per anno di uso, non lifetime"
},
"total_co2e_tonnes": 32_374.8
},
"summary": {
"scope_1_tco2e": 1_138.5,
"scope_2_location_tco2e": 1_922.3,
"scope_2_market_tco2e": 1_629.4,
"scope_3_tco2e": 32_374.8,
"total_location_based_tco2e": 35_435.6,
"total_market_based_tco2e": 35_142.7,
"intensity_tco2e_per_meur_revenue": 101.3,
"intensity_tco2e_per_fte": 29.5,
"scope_3_percentage_of_total": 91.4, # tipico nel manifatturiero
}
}Vaka Çalışmasına İlişkin Gözlemler
En önemli sonuç Kapsam 3'ün aşağıdakileri temsil etmesidir: Emisyonların %91,4'ü toplamlar, Kategori 1 (hammaddeler ve bileşenler) tek başına fiyatın %80'i değerindedir Kapsam 3. Bu, imalat sektörünün tipik bir örneğidir ve CSRD'nin neden aşağıdakileri gerektirdiğini açıklamaktadır: Tedarik zinciri raporlaması: Kapsam 3 olmasaydı, karbon muhasebesi şu orandan daha azını kapsayacaktı: Gerçek etkinin %10'u.
Kapsam 2 konuma dayalı (1.922 tCO2e) ile pazara dayalı (1.629 tCO2e) arasındaki fark, %50'si karşılığında yenilenebilir enerji sertifikalarının (Menşe Garantileri - GO) satın alınmasına elektrik tüketimi. 2024 İtalya'nın artık karışımı (0,395 kgCO2e/kWh) faktörden daha yüksektir ağın ortalaması (0,233 kgCO2e/kWh): bu mantık dışıdır ancak metodolojik olarak doğrudur, çünkü artık karışım halihazırda GO sertifikasına sahip olan enerjileri hariç tutuyor ve bu nedenle bir kota "içeriyor" yüksek yoğunluklu kaynaklardan daha yüksektir.
Test: Hesaplamaların Doğrulanması
Emisyon hesaplamaları mali kanunla aynı titizlikle test edilmelidir. Bir dönüşüm faktöründeki hata, büyüklük sırasına göre hatalara yol açabilir yüzlerce ton CO2e.
# tests/test_calculation_engine.py
import pytest
from decimal import Decimal
from calculation_engine.main import EmissionCalculator
@pytest.fixture
def calculator():
return EmissionCalculator()
class TestScope1Calculations:
def test_natural_gas_combustion_scope1(self, calculator):
"""
Test calcolo gas naturale con fattore DEFRA 2024.
Valore atteso: 5000 kWh x 0.18306 kgCO2e/kWh = 915.3 kgCO2e = 0.9153 tCO2e
"""
emission_factor = {
"co2e_factor": 0.18306,
"co2_factor": 0.18207,
"ch4_factor": 0.000291,
"n2o_factor": 0.000087,
"unit_type": "kgCO2e/kWh",
"source": "DEFRA 2024",
"year": 2024
}
result = calculator.calculate_activity_based(
quantity=5000,
unit="kWh",
emission_factor=emission_factor
)
assert abs(result["co2e_tonnes"] - 0.9153) < 0.001, (
f"Scope 1 gas naturale: atteso 0.9153, ottenuto {result['co2e_tonnes']}"
)
assert result["co2_tonnes"] is not None
assert "formula" in result
def test_diesel_combustion_scope1(self, calculator):
"""
Test calcolo gasolio.
Fattore DEFRA 2024: 2.56179 kgCO2e/liter
1000 litri -> 2.56179 kgCO2e -> 0.00256 tCO2e (arrotondato a 6 decimali)
"""
emission_factor = {
"co2e_factor": 2.56179,
"co2_factor": 2.51476,
"ch4_factor": 0.00179,
"n2o_factor": 0.00524,
"unit_type": "kgCO2e/liter",
"source": "DEFRA 2024",
"year": 2024
}
result = calculator.calculate_activity_based(
quantity=1000,
unit="liter",
emission_factor=emission_factor
)
expected = 2561.79 / 1000
assert abs(result["co2e_tonnes"] - expected) < 0.001
def test_unit_incompatibility_raises_error(self, calculator):
"""Test che unita incompatibili generino errore"""
emission_factor = {
"co2e_factor": 0.18306,
"unit_type": "kgCO2e/kWh",
}
with pytest.raises(ValueError, match="incompatibili"):
calculator.calculate_activity_based(
quantity=1000,
unit="liter", # incompatibile con kWh
emission_factor=emission_factor
)
class TestScope2Calculations:
def test_scope2_location_based_italy_2024(self, calculator):
"""
Fattore rete elettrica italiana IEA 2024: 0.233 kgCO2e/kWh
100.000 kWh -> 23.300 kgCO2e -> 23.3 tCO2e
"""
emission_factor = {
"co2e_factor": 0.233,
"unit_type": "kgCO2e/kWh",
"source": "IEA 2024 Italy",
"year": 2024
}
result = calculator.calculate_activity_based(
quantity=100_000,
unit="kWh",
emission_factor=emission_factor
)
assert abs(result["co2e_tonnes"] - 23.3) < 0.01
def test_scope2_market_based_with_go_certificates(self, calculator):
"""
Con GO certificates per energia rinnovabile: fattore = 0 per la quota coperta.
Residual mix IT 2024: 0.395 kgCO2e/kWh per la quota non coperta.
50% rinnovabili (GO): 50.000 kWh x 0 = 0
50% residual mix: 50.000 kWh x 0.395 = 19.750 kgCO2e = 19.75 tCO2e
"""
# Calcolo solo sulla quota non coperta da GO
emission_factor_residual = {
"co2e_factor": 0.395,
"unit_type": "kgCO2e/kWh",
"source": "AIB Italy Residual Mix 2024",
"year": 2024
}
non_go_kwh = 50_000 # 50% non coperto da GO
result = calculator.calculate_activity_based(
quantity=non_go_kwh,
unit="kWh",
emission_factor=emission_factor_residual
)
assert abs(result["co2e_tonnes"] - 19.75) < 0.01
class TestScope3Calculations:
def test_business_travel_air_short_haul_economy(self):
"""Test emissioni volo breve raggio con RFI"""
from integrations.travel_connector import TravelDataProcessor, TravelRecord
from datetime import date
processor = TravelDataProcessor()
record = TravelRecord(
employee_id="EMP001",
travel_date=date(2024, 3, 15),
origin_iata="LIN",
destination_iata="FCO",
transport_mode="air",
distance_km=490,
travel_class="economy",
booking_amount_eur=180
)
# 490 km x 0.151 kgCO2e/pkm x 1.9 RFI = 140.531 kgCO2e = 0.14053 tCO2e
co2e = processor.calculate_flight_emissions(record)
expected = 490 * 0.151 * 1.9 / 1000
assert abs(co2e - expected) < 0.001
def test_spend_based_with_nace_intensity(self, calculator):
"""Test calcolo spend-based per Category 1"""
result = calculator.calculate_spend_based(
spend_eur=100_000,
emission_intensity=0.35 # kgCO2e/EUR - manifattura metalli
)
# 100.000 EUR x 0.35 kgCO2e/EUR = 35.000 kgCO2e = 35.0 tCO2e
assert abs(result["co2e_tonnes"] - 35.0) < 0.01Bilime Dayalı Hedefler (SBTi) ve Kontrol Paneli
Il Bilime Dayalı Hedefler girişimi (SBTi) kesin kriterleri tanımladı Paris Anlaşması ile uyumlu emisyon azaltma hedefleri için. SBTi kurumsal standardı şunları gerektirir: Kapsam 1+2'nin 2030'a kadar %42 azaltılması (temel 2020) 1,5°C senaryosu için ve eğer temsil ediyorlarsa Kapsam 3 kapsamı toplam emisyonların %40'ından fazlası (üretimde neredeyse her zaman doğrudur).
Platform kontrol paneli yalnızca mevcut emisyonları göstermemeli, ama hedefe giden yol: SBTi'nin gerektirdiği azaltma eğrisi, yıldan yıla gerçek emisyonlar ve azaltma girişimlerine dayalı tahminler planlanmıştır (yenilenebilir enerji için PPA, filo elektrifikasyonu, süreç verimliliği).
# dashboard/sbti_tracker.py - Tracking verso target SBTi
from dataclasses import dataclass
from typing import List, Dict
import pandas as pd
import numpy as np
@dataclass
class SBTiTarget:
organization_id: str
base_year: int
base_year_scope_1_2_tco2e: float
base_year_scope_3_tco2e: float
target_year: int = 2030
scope_12_reduction_pct: float = 42.0 # % riduzione vs base year
scope_3_reduction_pct: float = 25.0 # % riduzione vs base year
scenario: str = "1.5C"
@property
def scope_12_target_tco2e(self) -> float:
return self.base_year_scope_1_2_tco2e * (1 - self.scope_12_reduction_pct / 100)
@property
def annual_reduction_rate(self) -> float:
"""Tasso di riduzione annuale lineare necessario"""
years = self.target_year - self.base_year
return self.scope_12_reduction_pct / years
def build_sbti_trajectory(target: SBTiTarget, actuals: Dict[int, float]) -> pd.DataFrame:
"""
Costruisce tabella confronto tra traiettoria SBTi e emissioni effettive.
actuals: { anno: tCO2e Scope1+2 effettivo }
"""
years = range(target.base_year, target.target_year + 1)
trajectory = []
for year in years:
years_elapsed = year - target.base_year
total_years = target.target_year - target.base_year
# Riduzione lineare richiesta da SBTi
required_reduction_pct = (years_elapsed / total_years) * target.scope_12_reduction_pct
sbti_budget = target.base_year_scope_1_2_tco2e * (1 - required_reduction_pct / 100)
actual = actuals.get(year)
on_track = actual <= sbti_budget if actual is not None else None
trajectory.append({
"year": year,
"sbti_budget_tco2e": round(sbti_budget, 1),
"actual_tco2e": actual,
"gap_tco2e": round(actual - sbti_budget, 1) if actual else None,
"on_track": on_track
})
df = pd.DataFrame(trajectory)
return df
# Utilizzo per MeccanicaPrecisione SpA
target = SBTiTarget(
organization_id="meccanica-precisione-spa",
base_year=2020,
base_year_scope_1_2_tco2e=3_850.0, # baseline 2020
base_year_scope_3_tco2e=35_000.0,
)
actuals_scope_12 = {
2020: 3850.0,
2021: 3720.0,
2022: 3650.0,
2023: 3320.0,
2024: 3060.8, # scope_1 + scope_2_market dal nostro inventory
}
trajectory = build_sbti_trajectory(target, actuals_scope_12)
print(trajectory.to_string(index=False))
# Output:
# year sbti_budget_tco2e actual_tco2e gap_tco2e on_track
# 2020 3850.0 3850.0 0.0 True
# 2021 3608.6 3720.0 111.4 False
# 2022 3367.2 3650.0 282.8 False
# 2023 3125.8 3320.0 194.2 False
# 2024 2884.4 3060.8 176.4 False
# -> L'azienda e fuori traiettoria: serve accelerare le iniziative di riduzioneEn İyi Uygulamalar ve Anti-Kalıplar
En İyi Uygulamalar
- Hesaplamaların değişmezliği: Bir kez hesaplanıp kaydedildikten sonra, bir emisyon asla değiştirilmemelidir. Faktörlerin değişmesi durumunda hesaplamanın yeni bir versiyonu oluşturulur. Denetim takibi her iki sürümü de göstermelidir.
- Her iki Kapsam 2 yöntemini de raporlayın: Lokasyon bazlı ve pazar bazlı ESRS ve GHG Protokolüne göre her ikisi de zorunludur. Sadece pazar bazlı raporlama yapmayın çünkü daha düşüktür.
- Belge Kapsamı 3 önemliliği: 15 kategorinin tamamını hesaplamanıza gerek yok Kapsam 3. Ancak neden dahil ettiğinizi/hariç tuttuğunuzu bir önceliklendirme analizi ile göstermelisiniz. her kategori. Bu açık bir ESRS gerekliliğidir.
- Emisyon faktörlerinin versiyonlanması: Faktörler her yıl değişmektedir. Her hesaplama için kullanılan faktörün anlık görüntüsünü tutun. Güncelleme yeni sürümle açık bir yeniden hesaplama yapılmadan geriye dönük olarak.
- Veri kalitesi işaretleri: Her veriyi kaliteye göre sınıflandırın (ölçülen, hesaplanmış, tahmin edilmiş, harcamaya dayalı). ESRS kalitenin beyan edilmesini gerektirir Kapsam 3 verileri. Daha düşük kaliteli veriler kullanılabilir ancak beyan edilmelidir.
- Sınır belgeleri: Dahil edilen varlıkları açıkça belgeleyin gerekçeli olarak raporlama sınırları dahilinde olan ve hariç tutulanlar (eşik <%5, veriler mevcut değil, vb.).
Kaçınılması Gereken Anti-Desenler
- Yalnızca pazara dayalı Kapsam 2'yi kullanın: Sera Gazı Protokolü her ikisini de gerektirir. Yalnızca pazara dayalı (GO/PPA sayesinde genellikle daha düşük) olanları raporlayın. lokasyon bazlı ve metodolojik olarak yanlış ve potansiyel yeşil aklama.
- Emisyon faktörleri güncellenmedi: 5+ yaş faktörlerini kullanın özellikle değişen elektrik şebekesi için önemli hatalara neden olur yenilenebilir enerji kaynaklarının büyümesiyle her yıl.
- Konsolidasyonda mükerrer emisyonlar: Bir yan kuruluş ise Emisyonlarını hesaplıyor ve ana şirket bunları konsolidasyona dahil ediyor, özsermaye payı veya operasyonel kontrol yöntemi tutarlı bir şekilde kullanılmalıdır.
- Belirsizliği göz ardı edin: Tüm emisyon hesaplamaları bir belirsizlik, özellikle Kapsam 3. Rakamları, olmadan sunmayın. güven aralığını ve kullanılan yöntemi belirtin.
- Güvencesi olmayan raporlar: CSRD ile sınırlı güvence gerektirmeyen raporlar 1. Dalga için düzenleyici gereksinimleri karşılamıyor. Güvence planı ilk yıldan itibaren, sonradan düşünüldüğü gibi değil.
Sonuçlar ve Uygulama Yol Haritası
Kurumsal düzeyde bir karbon muhasebesi platformu tasarlamak karmaşık bir projedir multidisipliner beceriler gerektirir. Ancak kavramsal yapı açıktır: Sera Gazı Protokolü muhasebe çerçevesini ve veritabanlarını sağlar. emisyon faktörleri (DEFRA, EPA, Climatiq) katsayılarını sağlar, themikro hizmet mimarisi ölçeklenebilirliği ve sürdürülebilirliği garanti eder, vedeğişmez denetim izi CSRD'nin güvenilirliğini sağlar.
2025-2026'da CSRD'ye başvurması gereken bir KOBİ için pratik öneri şudur: için olgun bir SaaS platformuyla (Persefoni, Plan A veya Watershed) başlamak İlk iki yıl, gerçek verileri toplayın, veri eksikliklerinizi anlayın ve ancak o zaman Özel bir çözüm oluşturup oluşturmayacağınızı veya SaaS'ta mı kalacağınızı değerlendirin. Kuruluşların %90'ı sıfırdan inşa etmeye gerek yok.
Bunun yerine ESG platformlarında ürün olarak çalışan mühendislik ekipleri için konseptler bu makalede gösterilen - Sera Gazı Protokolü veri modeli, denetlenebilir hesaplama motoru, emisyon faktörleri API'si ile entegrasyon ve düzenleyici raporların oluşturulması - sono üzerine inşa edilecek temeller. Karbon muhasebesi yazılım pazarı büyüyecek 2030'a kadar yılda %27 artış: özel çözümlere yer var, özellikle platformların kapsamadığı proses verisi gereksinimlerine sahip endüstriyel sektörler için genelciler.
Önerilen Uygulama Yol Haritası
| Faz | Süre | Amaç | Çıkışlar |
|---|---|---|---|
| Aşama 1 | 1-2 ay | Çifte Önemlilik Değerlendirmesi + sınır | Kapsam 3 malzeme kategorilerinin listesi, raporlama sınırı |
| Aşama 2 | 2-3 ay | Kapsam 1 ve 2 veri toplama | Faturalar, SAP, SCADA ile otomatik boru hattı |
| Aşama 3 | 3-4 ay | Kapsam 3 malzeme kategorileri (harcamaya dayalı) | Kedi için Kapsam 3 hesaplamaları. 1, 4, 6, 7, 11 |
| Aşama 4 | 1-2 ay | CSRD/ESRS E1 raporu ve denetimi | Sınırlı güvence için rapor hazır |
| Aşama 5 | Sürekli | Veri kalitesinin ve tedarikçi katılımının iyileştirilmesi | Harcama bazlı azalma, tedarikçiye özel Kapsam 3 artışı |
Temel Kaynaklar
- Sera Gazı Protokolü Kurumsal Standardı: ghgprotocol.org - standart referans, ücretsiz ve indirilebilir
- DEFRA Emisyon Faktörleri 2024: gov.uk/government/publications/ sera gazı raporlama-dönüşüm-faktörleri-2024
- İklimlendirme API'si: climatiq.io/docs - tüm dokümantasyon e ücretsiz hızlı başlangıç
- ESRS E1 Standardı: efrag.org - Avrupa Finansal Raporlama Danışma Grubu, nihai ESRS standartları
- SBTi Kurumsal Kılavuzu: science Basedtargets.org - kılavuz bilimsel temelli hedeflerin tanımına
- ecoinvent Veritabanı: ecoinvent.org - LCA veritabanı Ayrıntılı proses verileriyle Kapsam 3
EnergyTech Serisinde Gelecek Makaleler
Bir sonraki makale konuyu inceleyecek Dijital Enerji İkizleri: nasıl oluşturulur Emisyon azaltma senaryolarını simüle etmek için endüstriyel tesislerin sanal kopyaları bunları fiziksel dünyada uygulamadan önce. Giderek daha merkezi hale gelen teknoloji Büyük endüstrilerin karbondan arındırma stratejileri.
Veri işine uygulanan yapay zeka teknolojileri hakkında daha fazla bilgi edinmek için ayrıca bkz. seri Veri Ambarı, Yapay Zeka ve Dijital Dönüşüm, özellikle makaleler Güvenilir Yapay Zeka için Veri Yönetişimi ve Veri Kalitesi ve yukarı İşletmeler için MLOps, hesaplama modellerinin yönetimine doğrudan uygulanabilir konuları ele alan Üretimdeki emisyonlar.