Architektura oprogramowania do rozliczania emisji dwutlenku węgla: platformy ESG
W 2025 r. raportowanie emisji dwutlenku węgla nie będzie już dobrowolnym wyborem dużych firm Kraje europejskie: jest to obowiązek prawny. Tam Dyrektywa w sprawie raportowania zrównoważonego rozwoju korporacyjnego (CSRD), wpis obowiązujący na mocy dekretu legislacyjnego nr 125/2024 we Włoszech, radykalnie zmienił krajobraz raportowania ESG, wprowadzając dyscyplinę z dziedziny komunikacji korporacyjnej do dziedziny zgodności z przepisami, z tym samym implikacje operacyjne i techniczne systemu rachunkowości finansowej.
Globalny rynek oprogramowania do rozliczania emisji dwutlenku węgla nadrobił zaległości 1,8 miliarda dolarów w 2025 roku i oczekuje się, że do 2030 r. przekroczy 6 miliardów, przy CAGR na poziomie 27,4%. Nie chodzi tylko o kupowanie platforma SaaS: zaprojektowanie solidnego systemu rozliczania emisji dwutlenku węgla wymaga umiejętności obejmujących m.in chemia atmosfery po inżynierię oprogramowania, od statystyki po prawo wspólnotowe. Zespoły inżynieryjne którzy sami budują te platformy, muszą zrozumieć Protokół dotyczący gazów cieplarnianych, gli Europejskie standardy raportowania dotyczącego zrównoważonego rozwoju (ESRS), bazy danych współczynników emisji i architektura systemów rozproszonych zdolnych do gromadzenia, obliczania i certyfikowania danych z setek z różnych źródeł.
Ten artykuł stanowi kompletny przewodnik techniczny: od modelowania domeny za pomocą jednostek GHG Protocol, po wdrożenie silnika obliczeniowego w Pythonie z FastAPI, aż po automatyzację raportów CSRD/CDP/GRI oraz integracja z API Climatiq w celu uzyskania współczynników emisji w czasie rzeczywistym. Jeśli budujesz platformę Wewnętrzny ESG, oceniasz rozwiązania dla przedsiębiorstw lub po prostu chcesz zrozumieć, jak działają te architektury pod maską, jesteś we właściwym miejscu.
Czego dowiesz się w tym artykule
- Protokół GHG: Zakres 1, 2 i 3 oraz 15 kategorii emisji pośrednich z łańcucha dostaw
- Jak modelować domenę danych systemu rozliczania emisji dwutlenku węgla (organizacja, obiekt, źródło emisji, działalność)
- Główne bazy danych wskaźników emisji: DEFRA, EPA, ecoinvent, Climatiq API
- Architektura mikrousług: zbieranie danych, silnik obliczeniowy, raportowanie, ścieżka audytu
- Implementacja Pythona z FastAPI i pandami do obliczania emisji
- Automatyzacja zakresu 3: integracja z SAP ERP, dane zakupowe, dane dotyczące podróży
- Automatyczne generowanie raportów CSRD/ESRS, CDP i GRI Standards
- Ścieżka audytu i pochodzenie danych: pełna identyfikowalność obliczeń
- Porównanie platform: Persefoni, Watershed, Sphera, Plan A
- Studium przypadku: włoska firma produkcyjna posiadająca raporty CSRD za rok 2025
- Zaktualizowane ustawodawstwo: Omnibus CSRD 2025, Taksonomia UE, Dekret legislacyjny 125/2024
Pozycja w serii EnergyTech
| # | Przedmiot | Państwo |
|---|---|---|
| 1 | MQTT i InfluxDB: szeregi czasowe danych dotyczących energii | Opublikowany |
| 2 | IEC 61850: Standardowy protokół dla inteligentnych sieci elektrycznych | Opublikowany |
| 3 | DERMS: Zarządzanie rozproszonymi zasobami energii | Opublikowany |
| 4 | System zarządzania budynkiem: optymalizacja energii AI | Opublikowany |
| 5 | Prognozowanie energii odnawialnej: ML dla fotowoltaiki i wiatru | Opublikowany |
| 6 | Równoważenie obciążenia pojazdów elektrycznych: inteligentne ładowanie i połączenie pojazdu z siecią | Opublikowany |
| 7 | Handel energią w technologii Blockchain P2P: zdecentralizowane rynki energii | Opublikowany |
| 8 | Jesteś tutaj - Architektura oprogramowania do rozliczania emisji dwutlenku węgla: Platformy ESG | Aktualny |
| 9 | Cyfrowy bliźniak energii: symulacja i optymalizacja | Następny |
| 10 | Infrastruktura OCPP i EV: standardy i wdrażanie | Już wkrótce |
Kontekst regulacyjny: CSRD, ESRS i dekret legislacyjny 125/2024
Zrozumienie przepisów nie jest wymogiem biurokratycznym: jest podstawą, na której projektuje się całą architekturę systemu. Każde wymaganie techniczne, każde pole bazy danych, każdy punkt końcowy API wynika z obowiązku konkretnego ujawnienia.
CSRD i fale aplikacji
La Dyrektywa w sprawie sprawozdawczości w zakresie zrównoważonego rozwoju korporacyjnego zdefiniowano trzy fale stosowania dla firm europejskich. Pierwsza fala (rok budżetowy 2024, raporty opublikowane w 2025 r.) dotyczyła przedsiębiorstw objęte już wcześniej NFRD: spółki giełdowe, banki i zakłady ubezpieczeń zatrudniające powyżej 500 pracowników. Druga i trzecia fala, pierwotnie zaplanowane na lata 2025 i 2026, zostały przesunięte o dwa lata od Dyrektywa o zatrzymaniu zegara opublikowane w Dzienniku Urzędowym UE w dniu 16 kwietnia 2025 r.
Najważniejsza zmiana nastąpiła w grudniu 2025 r. wraz z zatwierdzeniem pakietu Omnibus I: podniesiono próg obowiązkowego stosowania 1000 pracowników i 450 milionów euro obrotu, zmniejszając tę liczbę o około 80%. spółek objętych obowiązkowym CSRD. Standardy ESRS są poddawane rewizji z redukcją 61% obowiązkowych punktów danych (od około 1100 do około 430), przy czym przyjęcie spodziewane jest w pierwszej połowie z 2026 r. i stosowanie od 2027 r.
Zaktualizowany harmonogram CSRD (po Omnibusie 2025)
| Fala | Przedmioty | Pierwszy raport | Państwo |
|---|---|---|---|
| Fala 1 | Firmy już NFRD (>500 etatów, notowane na giełdzie/banki/ubezpieczenia) | Raport 2025 (rok budżetowy 2024) | W toku |
| Fala 2 | Duże firmy > 1000 pracowników i > 450 mln EUR obrotu | Raport 2027 (FY 2026) - przełożony | Odłożony |
| Fala 3 | MŚP notowane na regulowanych rynkach UE | Raport 2028 (FY 2027) - przełożony | Odłożony |
| ESRS wer. | Wszystkie tematy CSRD, standardy uproszczone | Od roku budżetowego 2027 | W konsultacji |
Dekret legislacyjny nr 125/2024: Włoska transpozycja
Włochy wdrożyły dyrektywę CSRD za pomocą Dekret legislacyjny nr 125 z dnia 6 września 2024 r, wszedł w życie 25 września 2024 r. Dekret uchylił poprzedni Dekret Ustawodawczy nr 254/2016, który transponował NFRD. Do głównych innowacji dla włoskich firm zalicza się: obowiązek ograniczona pewność (ograniczona pewność) na temat raportu dotyczącego zrównoważonego rozwoju sporządzonego przez a kwalifikowany audytor, włączenie raportu zrównoważonego rozwoju do raportu zarządu, i publikację w wydzielonym dziale Rejestru Spółek.
Taksonomia UE: związek
Rachunek emisji dwutlenku węgla krzyżuje się bezpośrednio z Rozporządzenie UE w sprawie taksonomii (Rozporządzenie UE 852/2020), który klasyfikuje działalność gospodarczą jako „zrównoważoną” w oparciu o sześć celów środowiskowych. Spółki CSRD muszą zgłaszać KPI dostosowania taksonomii (udziały w obrotach, CapEx, OpEx „dopasowane” i „kwalifikowane”). Wraz z uproszczeniami od 2025 r., firmy zatrudniające mniej niż 1000 pracowników są zwolnione z raportowania taksonomii, a pozostała część może ograniczać się do działań, które reprezentują co najmniej 10% obrotu, CapEx lub OpEx.
Protokół GHG: Ramy odniesienia
Il Protokół dotyczący gazów cieplarnianych oraz standard rozliczania emisji gazów cieplarnianych najbardziej rozpowszechniony na świecie, opracowany przez Światowy Instytut Zasobów (WRI) i Światową Radę Biznesu na rzecz Zrównoważonego Rozwoju (WBCSD). Praktycznie wszystkie ramy raportowania (CSRD/ESRS, CDP, GRI, ISO 14064) odnoszą się do Protokołu GHG lub na nim bazują.
Trzy zakresy: precyzyjne definicje
Rozróżnienie na trzy „zakresy” pozwala na jasne przypisanie emisji, unikanie powielania liczenia pomiędzy różnymi uczestnikami łańcucha dostaw.
Zakres 1: Emisje bezpośrednie
Emisje ze źródeł będących własnością organizacji lub przez nią kontrolowanych. Należą do nich: spalanie paliw kopalnych w kotłach, piecach, pojazdach będących własnością firmy; emisje procesowe (np. CO2 z reakcji chemicznych, CH4 z hodowli); emisje ulotne (wycieki czynnika chłodniczego, wycieki gazu z instalacji). Do odpowiednich gazów cieplarnianych zalicza się siedem z nich Protokół z Kioto: CO2, CH4, N2O, HFC, PFC, SF6, NF3, wszystkie przeliczone na Równoważnik CO2 (CO2e) wykorzystując potencjał globalnego ocieplenia (GWP) IPCC.
Zakres 2: Emisje pośrednie z zakupionej energii
Emisje związane z wytwarzaniem energii elektrycznej, ciepła, pary lub chłodzenia zakupione i skonsumowane przez organizację. Wymagane są wytyczne dotyczące zakresu 2 protokołu dotyczącego gazów cieplarnianych (2015). raportowanie z dwie różne metody: metoda oparte na lokalizacji (należy zastosować współczynnik emisyjności sieci elektroenergetycznej w lokalizacji konsumpcja, np. średni współczynnik sieci włoskiej) i metodę oparte na rynku (wykorzystuje współczynniki instrumentów rynkowych takich jak kontrakty różnicowe, certyfikaty Energii Odnawialnej/GO, Umowy Zakupu Energii). Jedno i drugie trzeba zgłosić.
Zakres 3: Inne emisje pośrednie
Emisje łańcucha wartości w podziale na 15 kategorii upstream i downstream. Są one zazwyczaj najbardziej znaczące (średnio 70–80% całkowitego śladu) i największe trudne do zmierzenia. ESRS wymaga raportowania kategorii z zakresu 3 „materiały” zidentyfikowane w ramach oceny podwójnej istotności.
15 kategorii z zakresu 3
| # | Kategoria | Typ | Typowe dla |
|---|---|---|---|
| 1 | Zakupione towary i usługi | Pod prąd | Wszystkie sektory |
| 2 | Dobra kapitałowe | Pod prąd | Produkcja, konstrukcja |
| 3 | Działalność związana z paliwami i energią | Pod prąd | Wszystkie sektory |
| 4 | Transport i dystrybucja wyższego szczebla | Pod prąd | Handel detaliczny, produkcja |
| 5 | Odpady powstające w trakcie działalności | Pod prąd | Produkcja, jedzenie |
| 6 | Podróże służbowe | Pod prąd | Usługi, technika |
| 7 | Dojazdy pracowników | Pod prąd | Wszystkie sektory |
| 8 | Aktywa leasingowane na rynku wyższego szczebla | Pod prąd | Nieruchomości, handel detaliczny |
| 9 | Transport dolny | W dół rzeki | Produkcja, FMCG |
| 10 | Obróbka sprzedanych produktów | W dół rzeki | Surowce, chemia |
| 11 | Wykorzystanie sprzedanych produktów | W dół rzeki | Motoryzacja, elektronika |
| 12 | Leczenie końca życia | W dół rzeki | Opakowania, dobra trwałe |
| 13 | Aktywa leasingowane na niższym szczeblu łańcucha dostaw | W dół rzeki | Nieruchomość |
| 14 | Franczyzy | W dół rzeki | Żywność i napoje, sprzedaż detaliczna |
| 15 | Inwestycje | W dół rzeki | Banki, fundusze inwestycyjne |
Model danych: modelowanie domeny rozliczania emisji dwutlenku węgla
Sercem każdej platformy rozliczania emisji dwutlenku węgla jest solidny model danych, który wiernie odzwierciedla koncepcje Protokołu GHG. Zobaczmy główne byty z ich atrybutami i relacjami.
Główne podmioty
Kompletny system obejmuje co najmniej sześć podstawowych podmiotów: Organizację, Obiekt, Źródło Emisji, Współczynnik emisji, aktywność i obliczenia. Oto model w Pythonie z SQLAlchemy:
# 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")Baza danych współczynników emisji: DEFRA, EPA, ecoinvent, Climatiq
Współczynniki emisji są matematycznym sercem rozliczania emisji dwutlenku węgla: przeliczają ilość aktywności (litry oleju napędowego, zużyte kWh, przejechane km, wydane euro) w tonach CO2e. Jakość i aktualizacja czynników bezpośrednio determinuje jakość raportu.
Główne bazy danych
| Bazy danych | Menedżer | Zasięg | Aktualizacja | Dostęp |
|---|---|---|---|---|
| DEFRA/BEIS | Rząd Wielkiej Brytanii (DESNZ) | Wielka Brytania + międzynarodowy, wszystkie zakresy | Roczne (lipiec) | Bezpłatny (Excel) |
| Centrum gazów cieplarnianych EPA | Amerykańska Agencja Ochrony Środowiska | USA, mobilność, energia, Zakres 3 | Roczne (styczeń) | Bezpłatny (Excel) |
| ekowynalazek | Stowarzyszenie Ecoinvent | Globalne, pełne LCA, ponad 18 000 zbiorów danych | Półrocznie | Licencja płatna |
| Baza danych IPCC EF | IPCC | Inwentarze globalne, krajowe | Z każdym raportem z oceny | Bezpłatny |
| API Climatiqa | Klimat | Wiele źródeł, ponad 50 000 czynników | Ciągłe (w czasie rzeczywistym) | API (bezpłatne) |
| Mieszanka pozostałości AIB | Stowarzyszenie Organów Wydających | Europa, zakres 2 w oparciu o rynek | Coroczny | Bezpłatny |
| MAE Energia elektryczna | Międzynarodowa Agencja Energetyczna | Globalne czynniki sieci elektroenergetycznej | Coroczny | Częściowo za darmo |
Integracja API Climatiq
Climatiq udostępnia interfejs API REST z pokryciem ponad 50 000 zweryfikowanych współczynników emisji ISO 14067 i GHG Protocol i jest jednym z najczęściej stosowanych rozwiązań do obliczeń całkujących emisji programowo. Interfejs API obsługuje zakresy 1, 2 i 3 za pomocą metod opartych na działaniach i oparte na wydatkach.
# 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.co2eArchitektura mikrousług dla rozliczania emisji dwutlenku węgla
Korporacyjna platforma rozliczania emisji dwutlenku węgla musi zarządzać heterogenicznymi przepływami danych (rachunki za energię, dane ERP, raporty wydatków, dane dostawców), kompleksowe i kontrolowalne obliczenia, i generowanie raportów w różnych formatach. Architektura i wybór mikrousług naturalne dla tych wymagań.
Cztery podstawowe usługi
1. Usługa gromadzenia danych
Odpowiedzialny za pozyskiwanie danych z różnych źródeł: ERP API (SAP, Oracle), rachunki za media poprzez OCR/parser, narzędzia do zarządzania podróżami (Concur, TravelPerk), dane dotyczące zamówień, Ręczne przesyłanie plików CSV, interfejsy API dostawców. Udostępnia punkty końcowe do pozyskiwania i zarządza normalizacja jednostek miar.
2. Silnik obliczeniowy
Serce obliczeniowe: odbiera znormalizowane działania, wybiera współczynniki emisji właściwe (lokalnie lub poprzez Climatiq API), stosuje formuły protokołu GHG, produkuje niezmienne wyniki z jednoznacznym wzorem obliczeniowym dla audytu. Wsparcie przeliczenia historyczne w przypadku aktualizacji czynników.
3. Usługa raportowania
Generowanie raportów w formatach wymaganych przez frameworki: CSRD/ESRS (z tagowaniem XBRL), Kwestionariusz CDP (format JSON/XML), ujawnienia GRI Standards, raporty wewnętrzne dla dashboardów. Zarządza wersjonowaniem raportów i podpisem cyfrowym w celu zapewnienia pewności.
4. Usługa ścieżki audytu
Prowadzi niezmienny dziennik każdej operacji: kto wprowadził dane, jaki czynnik Został użyty, kiedy został obliczony, kto go zatwierdził. Obsługuje linię danych kompletne: od danych źródłowych do liczby w raporcie końcowym. Niezbędne dla pewności przez audytorów.
Schemat architektoniczny
┌─────────────────────────────────────────────────────────────────┐
│ 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)
Silnik obliczeniowy: Implementacja Pythona z FastAPI
Silnik obliczeniowy i najbardziej krytyczny komponent. Musi być precyzyjny, możliwy do sprawdzenia, wersjonowane i zdolne do równoległej obsługi tysięcy obliczeń. Oto implementacja w komplecie z FastAPI.
# 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
passAutomatyzacja zakresu 3: ERP, Zakupy i Podróże
Zakres 3 i największe wyzwanie w zakresie rozliczania emisji dwutlenku węgla: dane rozproszone w kilkudziesięciu systemy, dostawcy o różnym poziomie dojrzałości, różne metodologie obliczeń kategoria. Automatyzacja to jedyny sposób, aby zapewnić trwałość operacyjną.
Integracja z SAP dla Zakresu 3, Kategoria 1 (Zakupione Towary)
# 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)
}Automatyczne raportowanie: CSRD/ESRS, CDP i GRI
Generowanie raportów regulacyjnych jest często najbardziej czasochłonnym procesem rozliczania emisji dwutlenku węgla. Automatyzacja radykalnie skraca czas z tygodni do godzin, zwiększa spójność i zapewnia możliwość prześledzenia źródła każdej figury.
Struktura raportu CSRD/ESRS E1 (zmiany klimatyczne)
Standard klimatyczny ESRS E1 wymaga ujawnienia informacji na temat: zarządzania klimatem, analiza strategii i scenariuszy, zarządzanie ryzykiem, metryki i cele. Metryki emisji są zdefiniowane w ESRS E1-6 i wymagają danych dla wszystkich trzech zakresów.
# 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"
}
}Ścieżka audytu i pochodzenie danych
Ponieważ CSRD wymaga od wykwalifikowanego audytora ograniczonej pewności, ścieżka audytu taka nie jest wymóg opcjonalny: i szkielet platformy. Każde wydanie raportu musi być identyfikowalny aż do pierwotnego źródła za pomocą wyraźnego wzoru obliczeniowego.
# 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 lineagePorównanie platform dla przedsiębiorstw: budowanie vs kupowanie
Przed zbudowaniem niestandardowej platformy konieczna jest ocena rozwiązań przedsiębiorstwo dostępne. Rynek oprogramowania do rozliczania emisji dwutlenku węgla jest bardzo dojrzały a wiodące platformy obsługują większość standardowych przypadków użycia.
| Platforma | Punkt siły | Sektor docelowy | Cena orientacyjna | Gotowy do CSRD |
|---|---|---|---|---|
| Persefony | Raportowanie na poziomie inwestorskim, tagowanie XBRL, skupienie się na SEC | Finanse, Korporacyjny | 50-500 tys. dolarów rocznie | Si |
| Dział wodny | Szybkość raportowania, zaawansowane narzędzia Scope 3 | Technologia, przedsiębiorczość | 100 tys.–1 mln dolarów rocznie | Si |
| Sfera | Zintegrowana LCA, zgodność przemysłowa, zarządzanie ryzykiem | Produkcja, Energia, Chemia | Na żądanie | Si |
| Plan A | Prosty UX, szybkie wdrożenie | MŚP, średnie przedsiębiorstwa | 10–100 tys. dolarów rocznie | Częściowy |
| IBM Envizi | Ponad 40 000 czynników emisji, integracja z ERP | Przedsiębiorstwo, narzędzia | Na żądanie | Si |
| Niestandardowa kompilacja | Całkowita elastyczność, natywna integracja systemów wewnętrznych | Duże firmy ze specyficznymi wymaganiami | Rozwój o wartości 500–5 mln USD | Zależy |
Kiedy budować na zamówienie, a kiedy kupować
Kup platformę jeśli: masz standardowe wymagania, chcesz rozpocząć transmisję na żywo w ciągu 3-6 miesięcy nie będziesz mieć zespołu inżynierów zajmującego się zrównoważonym rozwojem, ilością danych i zarządzalne.
Zbuduj niestandardowy, jeśli: masz bardzo specyficzne procesy produkcyjne z danymi niemożliwy do zarządzania przez standardowe platformy, chcesz natywnie zintegrować się ze starszymi systemami, masz wymagania dotyczące miejsca przechowywania danych, które nie są kompatybilne z SaaS, lub ilość danych i rzędu milionów rekordów rocznie (koszty SaaS stają się zaporowe).
Studium przypadku: Włoskie MŚP produkcyjne z CSRD 2025
Włoska firma produkcyjna zajmująca się inżynierią precyzyjną, zatrudniająca 1200 pracowników, Posiada obroty wynoszące 350 milionów euro oraz fabryki w Turynie, Mediolanie i Brescii w Fali 1 CSRD i musi złożyć pierwsze sprawozdanie za rok budżetowy 2024 do czerwca 2025 r.
Inwentaryzacja emisji na rok 2024
# 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
}
}Uwagi na temat studium przypadku
Najbardziej znaczącym wynikiem jest to, że Zakres 3 reprezentuje 91,4% emisji sumy, przy czym sama kategoria 1 (surowce i komponenty) jest warta 80% ceny Zakres 3. Jest to typowe dla sektora produkcyjnego i wyjaśnia, dlaczego CSRD wymaga sprawozdawczość dotycząca łańcucha dostaw: bez zakresu 3 rozliczanie emisji dwutlenku węgla obejmowałoby mniej niż 10% rzeczywistego wpływu.
Różnica między zakresem 2 opartym na lokalizacji (1922 tCO2e) a rynkowym (1629 tCO2e) wynika na zakup świadectw energii odnawialnej (Gwarancje Pochodzenia – GO) za 50%. zużycie energii elektrycznej. Włoski miks resztkowy z 2024 r. (0,395 kgCO2e/kWh) jest wyższy niż współczynnik średnia sieci (0,233 kgCO2e/kWh): jest to sprzeczne z intuicją, ale poprawne metodologicznie, ponieważ mieszanka resztkowa nie obejmuje energii już certyfikowanej w ramach GO i dlatego „zawiera” kwotę wyższe niż źródła o dużej intensywności.
Testowanie: Walidacja obliczeń
Obliczenia emisji muszą być sprawdzane z taką samą rygorystycznością jak kodeks finansowy. Błąd w jednym współczynniku konwersji może prowadzić do błędów rzędu wielkości setek 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.01Cele naukowe (SBTi) i pulpit nawigacyjny
Il Inicjatywa na rzecz celów naukowych (SBTi) określił precyzyjne kryteria celów redukcji emisji zgodnych z Porozumieniem paryskim. Standard korporacyjny SBTi wymaga: Redukcji zakresu 1+2 o 42% do roku 2030 (wartość bazowa na rok 2020) dla scenariusza 1,5°C oraz zakres 3, jeśli reprezentują ponad 40% całkowitej emisji (prawie zawsze dotyczy to produkcji).
Pulpit nawigacyjny platformy musi pokazywać nie tylko aktualne emisje, ale drogę do celu: krzywa redukcji wymagana przez SBTi, rzeczywiste emisje rok po roku oraz prognozy oparte na inicjatywach redukcyjnych planowane (PPA na energię odnawialną, elektryfikacja floty, efektywność procesów).
# 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 riduzioneNajlepsze praktyki i anty-wzorce
Najlepsze praktyki
- Niezmienność obliczeń: Po obliczeniu i zapisaniu emisja nigdy nie wolno go modyfikować. Jeżeli współczynniki ulegną zmianie, tworzona jest nowa wersja obliczenia. Ścieżka audytu musi wykazywać obie wersje.
- Zgłoś obie metody Zakresu 2: Są one oparte na lokalizacji i na rynku oba obowiązkowe zgodnie z protokołem ESRS i GHG. Nie zgłaszaj tylko na podstawie rynku bo jest niższy.
- Istotność dokumentu Zakres 3: Nie musisz obliczać wszystkich 15 kategorii Zakres 3. Należy jednak wykazać za pomocą analizy istotności, dlaczego uwzględniłeś/wykluczyłeś każdą kategorię. Jest to wyraźny wymóg ESRS.
- Wersjonowanie współczynników emisji: Czynniki zmieniają się co roku. Zachowaj migawkę współczynnika użytego do każdego obliczenia. Nie aktualizuj z mocą wsteczną, bez dokonywania wyraźnych przeliczeń w nowej wersji.
- Flagi jakości danych: Klasyfikuj każde dane według jakości (mierzonej, obliczone, szacunkowe, oparte na wydatkach). ESRS wymaga deklarowania jakości danych Zakresu 3. Można, ale należy wykorzystać dane o niższej jakości zostać ogłoszony.
- Dokumentacja graniczna: Jawnie dokumentuj uwzględnione encje mieszczących się w granicach raportowania oraz wyłączonych, wraz z uzasadnieniem (próg <5%, brak danych itp.).
Anty-wzorce, których należy unikać
- Używaj wyłącznie zakresu 2 opartego na zasadach rynkowych: Protokół GHG wymaga obu. Raportuj tylko dane rynkowe (zwykle niższe dzięki GO/PPA) bez lokalizacyjne i niepoprawne metodologicznie oraz potencjalne greenwashing.
- Współczynniki emisji nie zostały zaktualizowane: Użyj czynników starszych niż 5 lat wprowadza istotne błędy, zwłaszcza dla zmieniającej się sieci elektroenergetycznej każdego roku wraz z rozwojem odnawialnych źródeł energii.
- Zduplikowane emisje w konsolidacji: Jeśli spółka zależna oblicza swoje emisje, a spółka dominująca uwzględnia je w konsolidacji, należy konsekwentnie stosować udział w kapitale własnym lub metodę kontroli operacyjnej.
- Ignoruj niepewność: Wszystkie obliczenia emisji mają niepewność, zwłaszcza Zakres 3. Nie przedstawiaj liczb bez wskazać zakres ufności i zastosowaną metodę.
- Raporty bez pewności: W przypadku CSRD raporty nie mają ograniczonej pewności nie spełniają wymogów regulacyjnych dla Fazy 1. Planuj w celu zapewnienia pewności od pierwszego roku, a nie po namyśle.
Wnioski i plan wdrożenia
Zaprojektowanie platformy do rozliczania emisji dwutlenku węgla klasy korporacyjnej jest złożonym projektem co wymaga umiejętności interdyscyplinarnych. Ale struktura pojęciowa jest jasna: Protokół dotyczący gazów cieplarnianych zapewnia ramy księgowe, bazy danych współczynniki emisji (DEFRA, EPA, Climatiq) podają współczynniki, thearchitektura mikrousług gwarantuje skalowalność i łatwość konserwacji, iniezmienna ścieżka audytu zapewnia wiarygodność CSRD.
Dla MŚP, które musi przystąpić do CSRD w latach 2025–2026, praktyczną rekomendacją jest: zacząć od dojrzałej platformy SaaS (Persefoni, Plan A lub Watershed) dla m.in przez pierwsze dwa lata zbierz prawdziwe dane, poznaj luki w danych i dopiero wtedy oceń, czy zbudować rozwiązanie niestandardowe, czy pozostać w SaaS. 90% organizacji nie trzeba budować od zera.
W przypadku zespołów inżynierskich, które zamiast tego pracują na platformach ESG jako produkcie, koncepcje zilustrowany w tym artykule - model danych protokołu GHG, silnik obliczeniowy podlegający audytowi, integracja z API wskaźników emisji i generowanie raportów regulacyjnych - sono fundamenty, na których można budować. Rynek oprogramowania do rozliczania emisji dwutlenku węgla będzie rósł o 27% rocznie do 2030 r.: jest miejsce na rozwiązania specjalistyczne, m.in dla sektorów przemysłowych, w których wymagania dotyczące danych procesowych nie są objęte platformami generaliści.
Zalecany plan wdrożenia
| Faza | Czas trwania | Cel | Wyjścia |
|---|---|---|---|
| Faza 1 | 1-2 miesiące | Podwójna ocena istotności + granica | Lista kategorii materiałowych Zakresu 3, granice raportowania |
| Faza 2 | 2-3 miesiące | Gromadzenie danych w zakresie 1 i 2 | Zautomatyzowany rurociąg z rachunkami, SAP, SCADA |
| Faza 3 | 3-4 miesiące | Kategorie materiałów w zakresie 3 (w oparciu o wydatki) | Obliczenia zakresu 3 dla kat. 1, 4, 6, 7, 11 |
| Faza 4 | 1-2 miesiące | Raport i audyt CSRD/ESRS E1 | Raport gotowy do uzyskania ograniczonej pewności |
| Faza 5 | Ciągły | Poprawa jakości danych i zaangażowania dostawców | Redukcja w oparciu o wydatki, zwiększenie zakresu 3 dla konkretnego dostawcy |
Niezbędne zasoby
- Standard korporacyjny protokołu GHG: ghgprotocol.org - standard referencyjne, bezpłatne i do pobrania
- Współczynniki emisji DEFRA 2024: gov.uk/government/publications/ współczynniki konwersji raportowania gazów cieplarnianych-2024
- API Climatiqa: climatiq.io/docs - pełna dokumentacja np darmowy szybki start
- Norma ESRS E1: efrag.org - Europejska sprawozdawczość finansowa Grupa Doradcza, ostateczne standardy ESRS
- Podręcznik korporacyjny SBTi: sciencebasedtargets.org - przewodnik do określenia celów naukowych
- Baza danych ecoinvent: ecoinvent.org - baza danych LCA dla Zakres 3 ze szczegółowymi danymi procesowymi
Nadchodzące artykuły z serii EnergyTech
W następnym artykule omówimy Cyfrowe bliźniaki energetyczne: jak tworzyć wirtualne repliki zakładów przemysłowych do symulacji scenariuszy redukcji emisji przed wdrożeniem ich w świecie fizycznym. Coraz bardziej centralna technologia w strategie dekarbonizacji dużych gałęzi przemysłu.
Aby dowiedzieć się więcej o technologiach AI stosowanych w biznesie opartym na danych, zobacz także seria Hurtownia danych, sztuczna inteligencja i transformacja cyfrowa, zwłaszcza artykuły dot Zarządzanie danymi i jakość danych dla niezawodnej sztucznej inteligencji i w górę MLOps dla biznesu, które dotyczą tematów bezpośrednio odnoszących się do zarządzania modelami obliczeniowymi emisji w produkcji.