Arhitectura software de contabilitate a carbonului: Platforme ESG
În 2025, raportarea carbonului nu mai este o alegere voluntară pentru companiile mari Țările europene: este o obligație legală. Acolo Directiva privind raportarea durabilității corporative (CSRD), intrare în vigoare cu Decretul Legislativ 125/2024 în Italia, a transformat radical peisajul raportării ESG, aducând disciplina de la domeniul comunicarii corporative la cel al conformarii reglementarilor, cu aceasta implicațiile operaționale și tehnice ale unui sistem financiar-contabil.
Piața globală a software-ului de contabilizare a carbonului a ajuns din urmă 1,8 miliarde de dolari în 2025 și este de așteptat să depășească 6 miliarde până în 2030, cu un CAGR de 27,4%. Nu este vorba doar despre cumpărare o platformă SaaS: proiectarea unui sistem robust de contabilizare a carbonului necesită competențe de la chimia atmosferică la inginerie software, de la statistici la dreptul comunitar. Echipele de ingineri care se trezesc construind aceste platforme trebuie să înțeleagă Protocolul GHG, The Standardele europene de raportare de durabilitate (ESRS), bazele de date cu factori de emisie și arhitectura sistemelor distribuite capabile să colecteze, să calculeze și să certifice date de la sute din surse diferite.
Acest articol este un ghid tehnic complet: de la modelarea domeniului cu entități GHG Protocol, la implementarea unui motor de calcul în Python cu FastAPI, până la automatizarea rapoartelor CSRD/CDP/GRI și integrarea cu API-ul Climatiq pentru factorii de emisie în timp real. Dacă construiți o platformă ESG intern, evaluarea soluțiilor de întreprindere sau pur și simplu doriți să înțelegeți cum funcționează aceste arhitecturi sub capotă, ești în locul potrivit.
Ce veți învăța în acest articol
- Protocolul GHG: Scopul 1, 2 și 3 și cele 15 categorii de emisii indirecte din lanțul de aprovizionare
- Cum se modelează domeniul de date al unui sistem de contabilitate a carbonului (organizație, instalație, sursă de emisii, activitate)
- Principalele baze de date cu factori de emisie: DEFRA, EPA, ecoinvent, Climatiq API
- Arhitectura microserviciilor: Colectare date, Motor de calcul, Raportare, Audit Trail
- Implementarea Python cu FastAPI și panda pentru calcularea emisiilor
- Automatizare Scope 3: integrare cu SAP ERP, date de achiziții, date de călătorie
- Generarea automată a rapoartelor CSRD/ESRS, CDP și GRI Standards
- Pista de audit și descendența datelor: trasabilitatea completă a calculelor
- Comparație platformă: Persefoni, Watershed, Sphera, Plan A
- Studiu de caz: Companie de producție italiană cu raportare CSRD 2025
- Legislație actualizată: CSRD Omnibus 2025, Taxonomie UE, Decretul Legislativ 125/2024
Poziție în seria EnergyTech
| # | Articol | Stat |
|---|---|---|
| 1 | MQTT și InfluxDB: Serii temporale pentru date energetice | Publicat |
| 2 | IEC 61850: Protocol standard pentru rețele electrice inteligente | Publicat |
| 3 | DERMS: Managementul resurselor energetice distribuite | Publicat |
| 4 | Sistem de management al clădirii: Optimizarea energiei AI | Publicat |
| 5 | Prognoza energiei regenerabile: ML pentru fotovoltaic și eolian | Publicat |
| 6 | Echilibrarea sarcinii EV: încărcare inteligentă și vehicul la rețea | Publicat |
| 7 | Blockchain P2P Energy Trading: piețe de energie descentralizate | Publicat |
| 8 | Sunteți aici - Carbon Accounting Software Architecture: ESG Platforms | Actual |
| 9 | Energy Digital Twin: Simulare și Optimizare | Următorul |
| 10 | OCPP și infrastructura EV: standarde și implementare | În curând |
Contextul de reglementare: CSRD, ESRS și Decretul Legislativ 125/2024
Înțelegerea legislației nu este o condiție prealabilă birocratică: este baza pe care este proiectată întreaga arhitectură a sistemului. Fiecare cerință tehnică, fiecare câmp de bază de date, fiecare punct final API provine dintr-o obligație de dezvăluire specifică.
CSRD și Application Waves
La Directiva de raportare a durabilității corporative definit trei valuri de aplicare pentru companiile europene. Primul val (FY 2024, rapoarte publicate în 2025) a implicat companii deja supuse NFRD anterioare: companii listate, bănci și companii de asigurări cu peste 500 de angajați. Al doilea și al treilea val, programate inițial pentru 2025 și 2026, au fost amânate cu doi ani din Directiva Stop-the-Clock publicat în Jurnalul Oficial al UE la 16 aprilie 2025.
Cea mai semnificativă schimbare a venit în decembrie 2025 odată cu aprobarea pachetului Omnibus I: a fost ridicat pragul de aplicare obligatoriu la 1.000 de angajați și 450 de milioane de euro cifră de afaceri, reducând numărul cu aproximativ 80%. a firmelor supuse CSRD obligatoriu. Standardele ESRS sunt revizuite cu o reducere de 61% din punctele de date obligatorii (de la aproximativ 1.100 la aproximativ 430), cu adoptare așteptată în prima jumătate a anului din 2026 și aplicare din 2027.
Cronologie CSRD actualizată (post-Omnibus 2025)
| Val | Subiecte | Primul Raport | Stat |
|---|---|---|---|
| Valul 1 | Companii deja NFRD (>500 emp., listate/bănci/asigurări) | Raport 2025 (FY 2024) | În curs |
| Valul 2 | Companii mari >1.000 de angajați și cifră de afaceri >450M EUR | Raport 2027 (FY 2026) - amânat | Amânat |
| Valul 3 | IMM-uri listate pe piețele reglementate din UE | Raport 2028 (FY 2027) - amânat | Amânat |
| ESRS rev. | Toate subiectele CSRD, standarde simplificate | Din anul fiscal 2027 | În consultare |
Decretul legislativ 125/2024: Transpunerea italiană
Italia a implementat CSRD cu Decretul legislativ 125 din 6 septembrie 2024, a intrat în vigoare la 25 septembrie 2024. Decretul a abrogat anterior Decretul Legislativ 254/2016 care a transpus NFRD. Principalele inovații pentru companiile italiene includ: obligația de a asigurare limitată (asigurare limitată) privind raportul de sustenabilitate de către a auditor calificat, integrarea raportului de sustenabilitate în raportul de management, și publicarea în secțiunea dedicată a Registrului Societății.
Taxonomia UE: Conexiunea
Contabilitatea carbonului se intersectează direct cu Regulamentul UE privind taxonomia (Reg. UE 852/2020), care clasifică activitățile economice drept „durabile” pe baza a șase obiective de mediu. Companiile CSRD trebuie să raporteze KPI-uri de aliniere a taxonomiei (acțiuni din cifra de afaceri, CapEx, OpEx „aliniat” și „eligibil”). Odată cu simplificările din 2025, companiile cu mai puțin de 1.000 de angajați sunt scutiți de raportarea taxonomiei, iar restul poate fi limitat la activitățile pe care le reprezintă cel puțin 10% din cifra de afaceri, CapEx sau OpEx.
Protocolul GHG: Cadrul de referință
Il Protocolul privind gazele cu efect de seră și standardul de contabilitate pentru emisiile de gaze cu efect de seră cel mai răspândit în lume, dezvoltat de World Resources Institute (WRI) și World Business Council pentru Dezvoltare Durabilă (WBCSD). Practic toate cadrele de raportare (CSRD/ESRS, CDP, GRI, ISO 14064) se referă sau se bazează pe Protocolul GHG.
Cele trei domenii: definiții precise
Distincția în trei „domeni” permite atribuirea clară a emisiilor, evitând dublarea numărării între diferiți actori din lanțul de aprovizionare.
Domeniul 1: Emisii directe
Emisii din surse deținute sau controlate de organizație. Acestea includ: arderea combustibililor fosili în cazane, cuptoare, vehicule deținute de companie; emisii de proces (de exemplu, CO2 din reacții chimice, CH4 de la animale); emisii fugitive (scurgeri de agent frigorific, scurgeri de gaz din sisteme). Gazele cu efect de seră relevante sunt cele șapte dintre Protocolul Kyoto: CO2, CH4, N2O, HFC, PFC, SF6, NF3, toate convertite în echivalent CO2 (CO2e) folosind potenţialele de încălzire globală (GWP) ale IPCC.
Domeniul de aplicare 2: Emisii indirecte din energia achiziționată
Emisii asociate cu generarea de energie electrică, căldură, abur sau răcire cumpărate și consumate de organizație. Ghidul GHG Protocol Scope 2 (2015) necesită raportarea cu două metode distincte: metoda bazate pe locație (utilizați factorul de emisie al rețelei electrice din locația consum, de ex. factorul mediu al reţelei italiene) şi metoda bazate pe piata (folosește factori ai instrumentelor de piață, cum ar fi contracte pentru diferență, certificate de Energie Regenerabilă/GO, Acordurile de Cumpărare a Energiei). Ambele trebuie raportate.
Domeniul de aplicare 3: Alte emisii indirecte
Emisiile lanțului valoric, împărțite în 15 categorii în amonte și în aval. Acestea sunt de obicei cele mai semnificative (în medie 70-80% din amprenta totală) și cele mai multe greu de măsurat. ESRS necesită raportarea categoriilor Scope 3 „materiale” identificate prin evaluarea dublei semnificații.
Cele 15 categorii Scope 3
| # | Categorie | Tip | Tipic pentru |
|---|---|---|---|
| 1 | Bunuri și servicii achiziționate | în amonte | Toate sectoarele |
| 2 | Bunuri de capital | în amonte | Productie, constructii |
| 3 | Activități legate de combustibil și energie | în amonte | Toate sectoarele |
| 4 | Transport și distribuție în amonte | în amonte | Comert cu amanuntul, productie |
| 5 | Deșeuri generate în operațiuni | în amonte | Producție, alimente |
| 6 | Călătorii de afaceri | în amonte | Servicii, tehnologie |
| 7 | Naveta angajaților | în amonte | Toate sectoarele |
| 8 | Activele închiriate în amonte | în amonte | Imobiliare, comert cu amanuntul |
| 9 | Transport în aval | în aval | Producție, FMCG |
| 10 | Prelucrarea produselor vândute | în aval | Materii prime, chimice |
| 11 | Utilizarea produselor vândute | în aval | Auto, electronice |
| 12 | Tratament la sfârșitul vieții | în aval | Ambalare, bunuri de folosință îndelungată |
| 13 | Activele închiriate în aval | în aval | Imobiliare |
| 14 | Francize | în aval | Alimente și băuturi, comerț cu amănuntul |
| 15 | Investiții | în aval | Bănci, fonduri de investiții |
Model de date: modelarea domeniului de contabilitate a carbonului
Inima fiecărei platforme de contabilizare a carbonului este un model de date robust care reflectă fidel conceptele Protocolului GHG. Să vedem principalele entități cu atributele și relațiile lor.
Entitățile principale
Un sistem complet include cel puțin șase entități de bază: organizație, instalație, sursă de emisii, Factorul de emisie, activitate și calcul. Iată modelul în Python cu 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 de date cu factori de emisie: DEFRA, EPA, ecoinvent, Climatiq
Factorii de emisie sunt inima matematică a contabilizării carbonului: ei convertesc o cantitate de activitate (litri de motorină, kWh consumați, km parcurși, euro cheltuiți) în tone de CO2e. Calitatea și actualizarea factorilor determină în mod direct calitatea raportului.
Bazele de date principale
| Baze de date | Manager | Acoperire | Actualizare | Acces |
|---|---|---|---|---|
| DEFRA/BEIS | Guvernul Regatului Unit (DESNZ) | Marea Britanie + internațional, toate domeniile | Anual (iulie) | Gratuit (Excel) |
| Hub EPA GHG | US EPA | SUA, mobilitate, energie, Scop 3 | Anual (ianuarie) | Gratuit (Excel) |
| ecoinvent | Asociația ecoinvent | LCA global, complet, peste 18.000 de seturi de date | Semianual | Licență plătită |
| Baza de date IPCC EF | IPCC | Inventare globale, naționale | Cu fiecare Raport de evaluare | Gratuit |
| Climatiq API | Climatiq | Multi-surse, peste 50.000 de factori | Continuu (în timp real) | API (freemium) |
| Amestecul rezidual AIB | Asociația Organismelor emitente | Europa, Scope 2 bazat pe piață | Anual | Gratuit |
| IEA Electricitate | Agenția Internațională pentru Energie | Factori globali, rețelei electrice | Anual | Parțial gratuit |
Integrare Climatiq API
Climatiq oferă un API REST cu peste 50.000 de factori de emisie verificați, acoperire ISO 14067 și GHG Protocol și este una dintre cele mai utilizate soluții pentru integrarea calculelor a emisiilor în mod programatic. API-ul acceptă Scope 1, 2 și 3 cu metode bazate pe activitate și bazate pe cheltuieli.
# 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.co2eArhitectura de microservicii pentru contabilizarea carbonului
O platformă de contabilitate a carbonului pentru întreprinderi trebuie să gestioneze fluxuri de date eterogene (facturi de energie, date ERP, rapoarte de cheltuieli, date furnizor), calcule complexe și auditabile, și generarea de rapoarte în diferite formate. Arhitectura și alegerea microserviciilor firesc pentru aceste cerinţe.
Cele patru servicii de bază
1. Serviciul de colectare a datelor
Responsabil pentru achiziția datelor din diferite surse: ERP API (SAP, Oracle), facturi de utilități prin OCR/parser, instrumente de management al călătoriilor (Concur, TravelPerk), date de achiziții, Încărcări manuale CSV, API-uri furnizor. Expune punctele finale pentru ingerare și gestionează normalizarea unităților de măsură.
2. Motor de calcul
Inima de calcul: primește activități normalizate, selectează factori de emisie adecvat (local sau prin Climatiq API), aplică formulele GHG Protocol, produce rezultate imuabile cu formulă de calcul explicită pentru audit. Sprijin recalculări istorice atunci când factorii sunt actualizați.
3. Serviciul de raportare
Generați rapoarte în formatele cerute de cadre: CSRD/ESRS (cu etichetare XBRL), Chestionar CDP (format JSON/XML), dezvăluiri GRI Standards, rapoarte interne pentru tablouri de bord. Gestionează versiunea rapoartelor și semnătura digitală pentru asigurare.
4. Serviciul Audit Trail
Păstrează un jurnal imuabil al fiecărei operațiuni: cine a introdus datele, ce factor S-a folosit, când s-a calculat, cine l-a aprobat. Suportă descendența datelor complet: de la datele sursă la numărul din raportul final. Esențial pentru asigurare de către auditori.
Diagrama arhitecturală
┌─────────────────────────────────────────────────────────────────┐
│ 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)
Motor de calcul: implementare Python cu FastAPI
Motorul de calcul și cea mai critică componentă. Trebuie să fie precis, auditabil, versionat și capabil să gestioneze mii de calcule în paralel. Iată o implementare complet cu 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
passAutomatizare Scope 3: ERP, Achiziții și Călătorii
Scopul 3 și cea mai mare provocare a contabilității carbonului: date distribuite pe zeci de sisteme, furnizori cu diferite niveluri de maturitate, diferite metodologii de calcul pt categorie. Automatizarea este singura modalitate de a-l face durabil din punct de vedere operațional.
Integrare SAP pentru Scope 3 Categoria 1 (Bunuri achiziționate)
# 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)
}Raportare automată: CSRD/ESRS, CDP și GRI
Generarea de rapoarte de reglementare este adesea cel mai consumator de timp proces de contabilizare a carbonului. Automatizarea reduce dramatic timpul de la săptămâni la ore, crește consistența și se asigură că fiecare cifră este trasabilă până la sursă.
Structura raportului CSRD/ESRS E1 (schimbări climatice)
Standardul climatic ESRS E1 necesită dezvăluirea cu privire la: guvernanța climatică, analiza strategiei și scenariilor, managementul riscului, metrici și obiective. Valorile de emisii sunt definite în ESRS E1-6 și necesită date pentru toate cele trei domenii.
# 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"
}
}Pista de audit și linia de date
Având în vedere că CSRD necesită o asigurare limitată din partea unui auditor calificat, pista de audit nu este o cerință opțională: și coloana vertebrală a platformei. Fiecare număr al raportului trebuie să fie urmărită până la sursa originală cu formula de calcul explicită.
# 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 lineageComparație platformă Enterprise: Build vs Buy
Înainte de a construi o platformă personalizată, este esențial să evaluați soluțiile intreprindere disponibila. Piața software-ului de contabilizare a carbonului este foarte matură iar platformele de vârf acoperă majoritatea cazurilor de utilizare standard.
| Platformă | Punct de forță | Sectorul țintă | Pret orientativ | Gata CSRD |
|---|---|---|---|---|
| Persefonele | Raportare pentru investitori, etichetare XBRL, concentrată pe SEC | Finanțe, Corporate | 50.000-500.000 USD/an | Si |
| Bazin de apă | Viteza de raportare, instrumente avansate Scope 3 | Tehnic, Enterprise | 100.000 - 1 milion de dolari/an | Si |
| Sfera | LCA integrat, conformitate industrială, managementul riscului | Producție, energie, chimie | La cerere | Si |
| Planul A | UX simplu, implementare rapidă | IMM-uri, întreprinderi mijlocii | 10.000-100.000 USD/an | Parţial |
| IBM Envizi | Peste 40.000 de factori de emisie, integrare ERP | Întreprindere, Utilități | La cerere | Si |
| Construire personalizată | Flexibilitate totală, integrare nativă a sistemelor interne | Companii mari cu cerințe specifice | Dezvoltare 500.000-5M USD | Depinde |
Când să construiți Custom vs Buy
Cumpărați o platformă dacă: aveți cerințe standard, doriți să intrați în direct în 3-6 luni, nu ai nicio echipă de ingineri dedicată durabilității, volumului de date si manevrabil.
Creați personalizat dacă: ai procese de producție foarte specifice cu date nu poate fi gestionat de platformele standard, doriți să vă integrați nativ cu sistemele vechi, aveți cerințe de rezidență a datelor care nu sunt compatibile cu SaaS sau volumul de date și de ordinul milioanelor de înregistrări/an (costurile SaaS devin prohibitive).
Studiu de caz: IMM-uri de producție italiene cu CSRD 2025
O companie italiană de producție de inginerie de precizie, cu 1.200 de angajați, 350 de milioane de euro cifră de afaceri și fabrici din Torino, Milano și Brescia, se află în Valul 1 al CSRD și trebuie să depună primul raport pentru exercițiul financiar 2024 până în iunie 2025.
Inventarul de emisii FY2024
# 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
}
}Observații asupra studiului de caz
Cel mai semnificativ rezultat este că Scopul 3 reprezintă 91,4% din emisii totaluri, doar cu Categoria 1 (materii prime și componente) în valoare de 80% din preț Domeniul de aplicare 3. Acest lucru este tipic sectorului de producție și explică de ce CSRD solicită Raportarea lanțului de aprovizionare: fără Scopul 3, contabilitatea carbonului ar acoperi mai puțin de 10% din impactul real.
Diferența dintre Scope 2 bazat pe locație (1.922 tCO2e) și bazat pe piață (1.629 tCO2e) se datorează la achiziționarea de certificate de energie regenerabilă (Garanții de origine - GO) pentru 50% din consumul de energie electrică. Mixul rezidual italian din 2024 (0,395 kgCO2e/kWh) este mai mare decât factorul media rețelei (0,233 kgCO2e/kWh): acest lucru este contraintuitiv, dar corect din punct de vedere metodologic, deoarece mixul rezidual exclude energiile deja certificate cu GO și deci „conține” o cotă mai mare decât sursele de intensitate mare.
Testare: Validarea calculelor
Calculele emisiilor trebuie testate cu aceeași rigoare ca și codul financiar. O eroare într-un factor de conversie poate duce la erori de ordinul mărimii de sute de tone de 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.01Obiective bazate pe știință (SBTi) și tablou de bord
Il Inițiativa privind obiectivele bazate pe știință (SBTi) a definit criterii precise pentru obiectivele de reducere a emisiilor aliniate la Acordul de la Paris. Standardul corporativ SBTi necesită: Reducere Scope 1+2 de 42% până în 2030 (linie de referință 2020) pentru un scenariu de 1,5 °C și acoperire Scope 3 dacă reprezintă mai mult de 40% din emisiile totale (aproape întotdeauna adevărat în producție).
Tabloul de bord al platformei trebuie să arate nu numai emisiile curente, dar cel calea către țintă: curba de reducere cerută de SBTi, emisiile reale an de an și proiecția bazată pe inițiative de reducere planificate (PPA pentru energie regenerabilă, electrificare a flotei, eficiență a procesului).
# 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 riduzioneCele mai bune practici și anti-modele
Cele mai bune practici
- Imuabilitatea calculelor: Odată calculată și salvată, o emisie nu trebuie modificată niciodată. Dacă factorii se modifică, se creează o nouă versiune a calculului. Pista de audit trebuie să arate ambele versiuni.
- Raportați ambele metode Scopul 2: Bazate pe locație și pe piață sunt ambele obligatorii conform ESRS și GHG Protocol. Nu raportați doar pe piață deoarece este mai jos.
- Materialitatea documentului Scopul 3: Nu trebuie să calculați toate cele 15 categorii Domeniul de aplicare 3. Dar trebuie să demonstrați printr-o analiză de materialitate de ce ați inclus/exclus fiecare categorie. Aceasta este o cerință ESRS explicită.
- Versiunea factorilor de emisie: Factorii se schimbă în fiecare an. Păstrați un instantaneu al factorului utilizat pentru fiecare calcul. Nu actualizați retroactiv fără a produce o recalculare explicită cu noua versiune.
- Indicatori de calitate a datelor: Clasificați fiecare dată după calitate (măsurată, calculate, estimate, bazate pe cheltuieli). ESRS necesită declararea calității a datelor Scopul 3. Date de calitate inferioară pot fi utilizate, dar trebuie fi declarat.
- Documentație de delimitare: Documentați în mod explicit entitățile incluse în limitele de raportare și cei excluși, cu justificare (prag <5%, date nu sunt disponibile etc.).
Anti-modele de evitat
- Utilizați numai Scopul 2 bazat pe piață: Protocolul GHG cere ambele. Raportați numai cele bazate pe piață (de obicei, mai mici datorită GO/PPA) fără bazată pe locație și metodologic incorectă și potențial greenwashing.
- Factorii de emisie nu sunt actualizați: Utilizați factori care au peste 5 ani introduce erori semnificative, în special pentru rețeaua electrică în schimbare în fiecare an, cu creșterea surselor regenerabile.
- Emisii duplicate în consolidare: Dacă o filială își calculează emisiile și societatea-mamă le include în consolidare, cota de capital sau metoda controlului operațional trebuie utilizată în mod consecvent.
- Ignorați incertitudinea: Toate calculele de emisii au o incertitudine, în special Scopul 3. Nu prezentați numere fără indicați intervalul de încredere și metoda utilizată.
- Rapoarte fara asigurare: Cu CSRD, rapoarte fără asigurare limitată nu îndeplinesc cerințele de reglementare pentru Valul 1. Planifică pentru asigurare din primul an, nu ca după gânduri.
Concluzii și Foaia de parcurs de implementare
Proiectarea unei platforme de contabilizare a carbonului la nivel de întreprindere este un proiect complex care necesită competenţe multidisciplinare. Dar structura conceptuală este clară: the Protocolul GHG oferă cadrul contabil, baze de date ale factori de emisie (DEFRA, EPA, Climatiq) furnizează coeficienții, celarhitectura microserviciilor garantează scalabilitate și întreținere, iar celpistă de audit imuabilă asigură credibilitate CSRD.
Pentru un IMM care trebuie să abordeze CSRD în 2025-2026, recomandarea practică este pentru a începe cu o platformă SaaS matură (Persefoni, Plan A sau Watershed) pentru i primii doi ani, colectați date reale, înțelegeți-vă lipsurile de date și abia apoi evaluați dacă să construiți o soluție personalizată sau să rămâneți pe SaaS. 90% din organizații nu trebuie să construiască de la zero.
Pentru echipele de inginerie care lucrează în schimb pe platforme ESG ca produs, conceptele ilustrat în acest articol - model de date GHG Protocol, motor de calcul auditabil, integrarea cu factorii de emisie API și generarea de rapoarte de reglementare - sono bazele pe care să construim. Piața software-ului de contabilizare a carbonului va crește cu 27% pe an până în 2030: există loc pentru soluții specializate, special pentru sectoarele industriale cu cerințe privind datele de proces care nu sunt acoperite de platforme generalişti.
Foaia de parcurs de implementare recomandată
| Fază | Durată | Obiectiv | Ieșiri |
|---|---|---|---|
| Faza 1 | 1-2 luni | Evaluare dublă materialitate + limită | Lista categoriilor de materiale Scope 3, limita de raportare |
| Faza 2 | 2-3 luni | Colectarea datelor pentru Scopul 1 și 2 | Conductă automată cu facturi, SAP, SCADA |
| Faza 3 | 3-4 luni | Categorii de materiale pentru Scopul 3 (pe baza cheltuielilor) | Calcule Scope 3 pentru cat. 1, 4, 6, 7, 11 |
| Faza 4 | 1-2 luni | Raport și audit CSRD/ESRS E1 | Raport gata pentru asigurare limitată |
| Faza 5 | Continuu | Îmbunătățirea calității datelor și implicarea furnizorilor | Reducere bazată pe cheltuieli, creșterea domeniului 3 specific furnizorilor |
Resurse esențiale
- Standard corporativ al protocolului GHG: ghgprotocol.org - standardul referință, gratuit și descărcabil
- Factori de emisie DEFRA 2024: gov.uk/government/publications/ raportarea-factori-de-conversie-gaze-sere-2024
- Climatiq API: climatiq.io/docs - documentație completă e pornire rapidă gratuită
- Standard ESRS E1: efrag.org - Raportarea financiară europeană Grupul consultativ, standardele finale ESRS
- Manual corporativ SBTi: sciencebasedtargets.org - ghid la definirea obiectivelor bazate științific
- Baza de date ecoinvent: ecoinvent.org - baza de date LCA pentru Scopul 3 cu date detaliate de proces
Articole viitoare din seria EnergyTech
Următorul articol va explora Gemeni cu energie digitală: cum se creează replici virtuale ale fabricilor industriale pentru a simula scenarii de reducere a emisiilor înainte de a le implementa în lumea fizică. O tehnologie din ce în ce mai centrală în strategiile de decarbonizare ale industriilor mari.
Pentru a afla mai multe despre tehnologiile AI aplicate afacerilor de date, consultați și seria Depozit de date, AI și transformare digitală, mai ales articolele despre Guvernarea datelor și calitatea datelor pentru IA de încredere și în sus MLOps pentru afaceri, care tratează subiecte direct aplicabile managementului modelelor de calcul a emisiilor în producţie.