Domeniul 1, 2 și 3: Modelarea datelor pentru software-ul de raportare ESG
Il Protocolul GHG (Protocolul privind gazele cu efect de seră) a devenit de facto cadrul universal pentru a măsura și raporta emisiile de gaze cu efect de seră ale companiilor. Dezvoltat de World Resources Institute (WRI) și de către Consiliul Mondial de Afaceri pentru Dezvoltare Durabilă (WBCSD), definește trei „domeni” de emisii care ele acoperă întregul lanț valoric al unei organizații.
Odată cu intrarea în vigoare a CSRD (Directiva de raportare a durabilității corporative) a Uniunii european și standard ESRS E1, mii de afaceri — inclusiv corporații software și tehnologie — acum trebuie să colecteze, să calculeze și să publice propriile date privind emisiile conform metodologiilor riguroase. Începând cu 2024, companiile mari cu peste 500 de angajați trebuie să facă acest lucru începe colectarea datelor, cu prima obligație de raportare în 2025; companii cu peste 250 de angajati vor urma începând cu 2026.
Dar de ce eu dezvoltator trebuie să înțeleagă Scopul 1, 2 și 3? De ce să construim un sistem ESG de încredere nu este doar o problemă a managerilor de sustenabilitate: este o provocare pentru ingineria datelor. Necesită un model de date consistent, conducte de calcul precise, API-uri pentru colectarea activităților și rapoarte gata de audit. Cine proiectează și implementează aceste sisteme determină calitatea – și, prin urmare, credibilitatea – a tuturor raportărilor ESG.
În acest articol vom construi un sistem de modelare a datelor pentru raportarea ESG, de la definițiile domeniului la tabelele bazei de date, de la calculele Python la punctele finale FastAPI, până la un studiu de caz complet pentru un IMM de software cu 200 de angajați.
Ce vei învăța
- Structura Protocolului GHG: Scopul 1, 2 și 3 cu definiții precise
- Modele entitate-relație pentru emisiile directe, indirecte și în lanțul valoric
- Bazat pe locație vs pe piață pentru Scopul 2: diferențe și impact asupra datelor
- Cele 15 categorii Scope 3 și care sunt relevante pentru companiile de software
- Schema SQLAlchemy completă: Organizație, Facilitate, EmissionSource, EmissionFactor, ActivityData
- Calcule Python cu estimare bazată pe activitate și estimare bazată pe cheltuieli
- Agregare, comparație de la an la an și alinierea la obiectivele SBTi
- API REST cu FastAPI pentru trimiterea sarcinilor, calcularea emisiilor și generarea de rapoarte
- Testare unitară pentru calculele GES cu pytest și validarea factorilor de emisie
- Studiu de caz de la capăt la capăt: Software IMM-uri 200 de angajați, Scop 1+2+3 complet
Green Software Series — 10 articole
| # | Titlu | Stat |
|---|---|---|
| 1 | Green Software Engineering și principiile SCI | Publicat |
| 2 | CodeCarbon: Măsurarea emisiilor de cod Python | Publicat |
| 3 | Carbon Aware SDK: schimbarea sarcinii de lucru și schimbarea timpului | Publicat |
| 4 | Climatiq API: factori de emisie și calcul de carbon | Publicat |
| 5 | Domeniul 1, 2 și 3: Modelarea datelor pentru raportarea ESG | Articolul curent |
| 6 | ESG și CSRD: Obligații europene pentru companiile de tehnologie | Curând |
| 7 | Modele software sustenabile: arhitecturi cu impact redus | Curând |
| 8 | GreenOps: Optimizare cloud pentru durabilitate | Curând |
| 9 | Conducta Scope 3: Automatizați lanțul valoric | Curând |
| 10 | Amprenta de carbon AI: LLM, Training and Inference | Curând |
Protocolul GHG: cadru universal pentru emisii
Standardul corporativ GHG Protocol, publicat în 2001 și actualizat în 2004, este fundamentul aproape toate cadrele de raportare climatică: CDP, TCFD, CSRD/ESRS E1, SBTi și multe altele se referă direct la el. Puterea sa constă în claritatea conceptuală: toate emisiile de o organizație sunt clasificate în trei categorii care se exclud reciproc, dar exhaustive în general.
Cadrul folosește conceptul de limitele organizatorice (limită organizațională) pentru determinați ce emisii să includeți în calcul. Există două abordări: celabordarea acțiunilor (pe baza ratei de participare financiară) și abordarea controlului (bazat pe control operaționale sau financiare). Majoritatea companiilor adoptă controlul operațional drept criteriu principal.
Cele trei domenii: prezentare generală
| Mături | Definiţie | Exemple tipice | ESRS E1 obligatoriu |
|---|---|---|---|
| Domeniul de aplicare 1 | Emisii directe din surse deținute sau controlate de organizație | Cazane pe gaz natural, parcul companiei, generatoare diesel, procese industriale | Obligatoriu |
| Domeniul de aplicare 2 | Emisii indirecte din achiziția de energie (electricitate, abur, căldură, frig) | Electricitate de birou, răcire centru de date, termoficare | Obligatoriu (ambele metode) |
| Domeniul de aplicare 3 | Toate celelalte emisii indirecte din lanțul valoric (în amonte și în aval) | Călătorii de afaceri, naveta, achiziționarea de bunuri/servicii, utilizarea produselor vândute | Obligatoriu (categorii de materiale) |
Domeniul de aplicare 1: Model de date pentru emisiile directe
Emisiile Scopul 1 provin din surse fizice controlate direct de companie. Pentru ei companii de software, acestea sunt de obicei mici, dar deloc neglijabile: includ încălzirea a birourilor (gaze naturale), a eventualelor generatoare de rezervă și a parcului companiei (mașini de serviciu sau furgonete de livrare hardware).
Protocolul GHG identifică patru categorii de emisii Scope 1:
- Combustie staționară: Cazane, încălzitoare, generatoare care ard combustibili fosili
- Combustie mobilă: Vehicule de companie alimentate cu benzină, diesel sau GPL
- Emisii de proces: Reacții chimice sau biologice (rareori relevante pentru software)
- Emisii fugitive: Scurgeri de agenți frigorifici (R-410A, R-134a) din sistemele HVAC
Formula de bază pentru calculul Scopului 1 este:
Emisii (tCO₂e) = Date de activitate × Factorul de emisie × GWP
Unde:
- Activitate dată: cantitatea de combustibil consumată (litri, m³, kWh)
- Factorul de emisie: kg CO₂ per unitate de combustibil (sursa: IPCC, DEFRA, IEA)
- GWP: Potenţial de încălzire globală (CO₂=1, CH₄=28, N₂O=265, HFC variază)
Diagrama entitate-relație pentru Scopul 1 include următoarele entități principale:
-- Entity-Relationship: Scope 1 Emissioni Dirette
ORGANIZATION (1) ---< FACILITY (>1)
|id, name, tax_id |id, org_id, name, type, country, area_m2
FACILITY (1) ---< EMISSION_SOURCE (>1)
|id, facility_id, source_type, description
| source_type: STATIONARY | MOBILE | FUGITIVE | PROCESS
EMISSION_SOURCE (1) ---< ACTIVITY_DATA (>1)
|id, source_id, period_start, period_end
|quantity, unit, data_quality, evidence_url
ACTIVITY_DATA (N) ---> EMISSION_FACTOR (1)
|id, gas, fuel_type, unit_from, unit_to
|factor_value, gwp_ar5, source, year, region
ACTIVITY_DATA (1) ---< GHG_CALCULATION (1)
|id, activity_id, scope, co2_kg, ch4_kg
|n2o_kg, hfc_kg, co2e_total_kg, method
|calculated_at, calculated_byDomeniul 2: bazat pe locație vs bazat pe piață
Scopul 2 este deosebit de complex pentru companiile de tehnologie, deoarece energia electrică consumată de la birouri și centre de date este adesea cel sursa principală de emisii indirecte. Ghidul GHG Protocol Scope 2 (2015) a introdus obligația de a raporta cu ambele metodele: bazate pe locație și pe piață.
Bazat pe locație vs bazat pe piață: diferențe cheie
| astept | Bazat pe locație | Bazat pe piață |
|---|---|---|
| Definiţie | Intensitatea medie a rețelei locale de energie electrică unde are loc consumul | Emisii de la generatoarele de la care se achiziționează energie prin contract |
| Factorul de emisie | Factorul de emisie mediu al rețelei (gCO₂/kWh) pe țară/regiune | Factorul specific furnizorului sau factorul de amestec rezidual |
| Instrumente contractuale | Nu se aplică | GO (Garanții de origine), REC (Certificate de energie regenerabilă), PPA |
| Exemplu Italia 2024 | ~310 gCO₂/kWh (date privind amestecul rezidual AIB) | 0 gCO₂/kWh dacă achiziționați GO din surse regenerabile verificate 100%. |
| Propunerea GHG Protocol 2025 | Noua ierarhie a factorilor cu precizie de timp | Solicitați potrivirea orară pentru surse regenerabile |
| Utilizați în ESRS E1 | Dezvăluirea obligatorie (E1-6) | Dezvăluirea obligatorie (E1-6) |
Pentru a modela corect Scope 2 în baza de date, trebuie să gestionați certificate energetice (GO/REC) ca entități separate care sunt „consumate” împotriva consumului de energie electrică al companiei, scăzând factorul bazat pe piață la zero:
-- Modello Scope 2: Energia e Certificati
ELECTRICITY_CONSUMPTION
id UUID PRIMARY KEY
facility_id UUID REFERENCES FACILITY
period_start DATE
period_end DATE
kwh_consumed DECIMAL(15,3)
meter_id VARCHAR(50)
-- Location-based
grid_region VARCHAR(20) -- e.g. 'IT', 'DE-WEST', 'FR'
grid_ef_g_kwh DECIMAL(8,3) -- gCO2e/kWh from IEA/AIB
lb_co2e_kg DECIMAL(15,3) GENERATED ALWAYS AS
(kwh_consumed * grid_ef_g_kwh / 1000)
-- Market-based
supplier_ef_g_kwh DECIMAL(8,3) DEFAULT NULL
residual_mix_ef DECIMAL(8,3) DEFAULT NULL
mb_co2e_kg DECIMAL(15,3) -- calcolato dopo allocazione GO/REC
RENEWABLE_CERTIFICATE
id UUID PRIMARY KEY
cert_type VARCHAR(10) -- 'GO' | 'REC' | 'I-REC'
cert_id VARCHAR(100) UNIQUE
technology VARCHAR(30) -- 'WIND' | 'SOLAR' | 'HYDRO'
country VARCHAR(2)
production_date DATE
kwh_value DECIMAL(15,3)
issuer VARCHAR(100)
status VARCHAR(20) -- 'ACTIVE' | 'CANCELLED' | 'EXPIRED'
CERTIFICATE_ALLOCATION
id UUID PRIMARY KEY
cert_id UUID REFERENCES RENEWABLE_CERTIFICATE
consumption_id UUID REFERENCES ELECTRICITY_CONSUMPTION
kwh_allocated DECIMAL(15,3)
allocation_date DATE
-- Vincolo: kwh_allocated <= RENEWABLE_CERTIFICATE.kwh_valueDomeniul 3: Cele 15 categorii ale lanțului valoric
Scopul 3 este de obicei sursa principala de emisii pentru companiile de software: poate reprezenta peste 70-80% din total. Ghidul tehnic pentru calcul al protocolului GHG Scopul 3 Emisii definește 15 categorii, împărțite în amonte (legate de achiziționarea de bunuri și servicii) și în aval (legat de utilizarea produselor vândute).
Cele 15 categorii Scope 3 pentru companiile de software
| # | Categorie | Tip | Relevanța software | Metoda de calcul |
|---|---|---|---|---|
| 1 | Bunuri și servicii achiziționate | în amonte | Ridicat (SaaS, cloud, hardware) | Bazat pe cheltuieli sau specific furnizorului |
| 2 | Bunuri de capital | în amonte | Media (laptop-uri, servere, mobilier) | Unitate fizică sau bazată pe cheltuieli |
| 3 | Activități legate de combustibil și energie | în amonte | Medie (pierderi T&D, combustibil în amonte) | Bazat pe activitate |
| 4 | Transport și distribuție în amonte | în amonte | Scăzut (livrare hardware) | Bazat pe cheltuieli sau pe distanță |
| 5 | Deșeuri generate în operațiuni | în amonte | Scăzut | Specific tip deșeu |
| 6 | Călătorii de afaceri | în amonte | Ridicat (zboruri, hoteluri, trenuri) | Bazat pe distanță sau pe bază de cheltuieli |
| 7 | Naveta angajaților | în amonte | Ridicat (naveta, lucru la distanță) | Sondaj + bazat pe distanță |
| 8 | Activele închiriate în amonte | în amonte | Mediu (birouri de inchiriat) | Specific pentru bunuri |
| 9 | Transport în aval | în aval | Scăzut | Bazat pe distanță |
| 10 | Prelucrarea produselor vândute | în aval | Nu se aplică (software) | Bazat pe activitate |
| 11 | Utilizarea produselor vândute | în aval | Ridicat (energie pentru a rula software-ul) | Consumul de energie pe viață |
| 12 | Tratamentul de sfârșit de viață | în aval | Scăzut (dispozitive de utilizator) | Specific tip deșeu |
| 13 | Activele închiriate în aval | în aval | Scăzut | Specific pentru bunuri |
| 14 | Francize | în aval | Nu se aplică | N / A |
| 15 | Investiții | în aval | Media (portofoliu VC/startup) | Bazat pe portofoliu |
Pentru companiile de software, cele mai relevante categorii sunt de obicei: Pisică. 1 (computing cloud achiziționat), Pisică. 6 (călătorii de afaceri), Pisică. 7 (naveta angajatului) e Pisică. 11 (energie consumată de utilizatori pentru a rula software-ul). Acesta din urmă este adesea subestimat, dar pentru întreprinderi SaaS cu milioane de utilizatori poate fi imens.
Schema SQLAlchemy: Model complet
Să construim acum o schemă completă a bazei de date în SQLAlchemy 2.0 care acceptă toate cele trei domenii. Designul urmează principiile de coeziune ridicată și cuplare scăzută, cu tabele separate pentru fiecare conceptul de dominanță și relații clare.
# models/base.py
from sqlalchemy import Column, String, DateTime, func
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import DeclarativeBase
import uuid
class Base(DeclarativeBase):
pass
class TimestampMixin:
"""Mixin per created_at e updated_at automatici."""
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())# models/organization.py
from sqlalchemy import Column, String, Integer, Boolean, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import uuid
from .base import Base, TimestampMixin
class Organization(Base, TimestampMixin):
"""Azienda che produce il report ESG."""
__tablename__ = "organizations"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
name = Column(String(200), nullable=False)
legal_name = Column(String(200))
tax_id = Column(String(50)) # Partita IVA o VAT number
country = Column(String(2), nullable=False) # ISO 3166-1 alpha-2
industry_code = Column(String(10)) # NACE Rev.2
employee_count = Column(Integer)
fiscal_year_end = Column(String(5), default="12-31") # MM-DD
reporting_currency = Column(String(3), default="EUR")
boundary_approach = Column(String(20), default="OPERATIONAL_CONTROL")
# OPERATIONAL_CONTROL | FINANCIAL_CONTROL | EQUITY_SHARE
active = Column(Boolean, default=True)
facilities = relationship("Facility", back_populates="organization")
reporting_periods = relationship("ReportingPeriod", back_populates="organization")
class Facility(Base, TimestampMixin):
"""Sede fisica (ufficio, data center, magazzino)."""
__tablename__ = "facilities"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
organization_id = Column(UUID(as_uuid=True), ForeignKey("organizations.id"), nullable=False)
name = Column(String(200), nullable=False)
facility_type = Column(String(30))
# OFFICE | DATA_CENTER | WAREHOUSE | LAB | REMOTE
address = Column(String(500))
city = Column(String(100))
country = Column(String(2), nullable=False)
grid_region = Column(String(30)) # Per fattore emissione location-based
area_sqm = Column(Integer) # Per calcoli per m2
employee_fte = Column(Integer) # Full-time equivalent
organization = relationship("Organization", back_populates="facilities")
emission_sources = relationship("EmissionSource", back_populates="facility")# models/emission_source.py
from sqlalchemy import Column, String, Enum, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import enum, uuid
from .base import Base, TimestampMixin
class SourceType(str, enum.Enum):
# Scope 1
STATIONARY_COMBUSTION = "STATIONARY_COMBUSTION"
MOBILE_COMBUSTION = "MOBILE_COMBUSTION"
FUGITIVE = "FUGITIVE"
PROCESS = "PROCESS"
# Scope 2
ELECTRICITY = "ELECTRICITY"
STEAM = "STEAM"
HEATING = "HEATING"
COOLING = "COOLING"
# Scope 3 (le 15 categorie)
S3_PURCHASED_GOODS = "S3_PURCHASED_GOODS" # Cat. 1
S3_CAPITAL_GOODS = "S3_CAPITAL_GOODS" # Cat. 2
S3_FUEL_ENERGY = "S3_FUEL_ENERGY" # Cat. 3
S3_UPSTREAM_TRANSPORT = "S3_UPSTREAM_TRANSPORT" # Cat. 4
S3_WASTE = "S3_WASTE" # Cat. 5
S3_BUSINESS_TRAVEL = "S3_BUSINESS_TRAVEL" # Cat. 6
S3_EMPLOYEE_COMMUTE = "S3_EMPLOYEE_COMMUTE" # Cat. 7
S3_UPSTREAM_LEASED = "S3_UPSTREAM_LEASED" # Cat. 8
S3_DOWNSTREAM_TRANSPORT = "S3_DOWNSTREAM_TRANSPORT" # Cat. 9
S3_USE_OF_PRODUCTS = "S3_USE_OF_PRODUCTS" # Cat. 11
S3_END_OF_LIFE = "S3_END_OF_LIFE" # Cat. 12
S3_INVESTMENTS = "S3_INVESTMENTS" # Cat. 15
class EmissionSource(Base, TimestampMixin):
"""Sorgente specifica di emissione (es. 'Caldaia edificio A')."""
__tablename__ = "emission_sources"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
facility_id = Column(UUID(as_uuid=True), ForeignKey("facilities.id"), nullable=True)
organization_id = Column(UUID(as_uuid=True), ForeignKey("organizations.id"), nullable=False)
source_type = Column(String(40), nullable=False) # SourceType enum value
scope = Column(Integer, nullable=False) # 1, 2, o 3
scope3_category = Column(Integer, nullable=True) # 1-15 se scope=3
description = Column(String(500))
unit_of_measure = Column(String(20)) # 'kWh', 'litri', 'km', 'EUR', 'kg'
active = Column(Boolean, default=True)
facility = relationship("Facility", back_populates="emission_sources")
activity_records = relationship("ActivityRecord", back_populates="source")# models/emission_factor.py
from sqlalchemy import Column, String, Numeric, Integer, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import uuid
from .base import Base, TimestampMixin
class EmissionFactor(Base, TimestampMixin):
"""
Fattore di emissione da fonti ufficiali.
Converte un'attività (es. kWh) in kg CO2e.
"""
__tablename__ = "emission_factors"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
name = Column(String(200), nullable=False)
category = Column(String(50)) # 'electricity' | 'fuel' | 'transport' | ...
fuel_type = Column(String(50)) # 'natural_gas' | 'diesel' | 'petrol' | ...
technology = Column(String(50)) # 'gas_boiler' | 'bev' | ...
# Unità
activity_unit = Column(String(20), nullable=False) # 'kWh', 'litre', 'km', ...
emission_unit = Column(String(20), default="kg_co2e") # Output sempre in kg CO2e
# Valori per singolo gas (kg/unità)
co2_factor = Column(Numeric(18, 8))
ch4_factor = Column(Numeric(18, 8))
n2o_factor = Column(Numeric(18, 8))
hfc_factor = Column(Numeric(18, 8))
# GWP (AR5 100-year, IPCC 2014)
gwp_co2 = Column(Numeric(6, 2), default=1.0)
gwp_ch4 = Column(Numeric(6, 2), default=28.0)
gwp_n2o = Column(Numeric(6, 2), default=265.0)
# Metadati fonte
source = Column(String(100)) # 'DEFRA_2024' | 'IEA_2024' | 'IPCC_AR6'
year = Column(Integer)
region = Column(String(10)) # 'IT' | 'EU' | 'UK' | 'GLOBAL'
version = Column(String(20))
activity_records = relationship("ActivityRecord", back_populates="emission_factor")# models/activity_record.py
from sqlalchemy import Column, String, Numeric, Date, ForeignKey, Text
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import uuid
from .base import Base, TimestampMixin
class ActivityRecord(Base, TimestampMixin):
"""
Dato di attività misurato o stimato per una fonte di emissione.
Es: 'Ufficio Milano ha consumato 15.432 kWh a novembre 2024'.
"""
__tablename__ = "activity_records"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
source_id = Column(UUID(as_uuid=True), ForeignKey("emission_sources.id"), nullable=False)
factor_id = Column(UUID(as_uuid=True), ForeignKey("emission_factors.id"), nullable=False)
reporting_period_id = Column(UUID(as_uuid=True), ForeignKey("reporting_periods.id"))
# Periodo
period_start = Column(Date, nullable=False)
period_end = Column(Date, nullable=False)
# Quantità
quantity = Column(Numeric(18, 4), nullable=False)
unit = Column(String(20), nullable=False)
# Qualità del dato
data_quality = Column(String(20), default="MEASURED")
# MEASURED | ESTIMATED | CALCULATED | MODELED
uncertainty_pct = Column(Numeric(5, 2)) # Es: 10.00 = +-10%
evidence_url = Column(Text) # Link a fattura, report, foglio elettrico
# Note
notes = Column(Text)
reviewer = Column(String(100))
source = relationship("EmissionSource", back_populates="activity_records")
emission_factor = relationship("EmissionFactor", back_populates="activity_records")
calculation = relationship("GHGCalculation", back_populates="activity_record", uselist=False)
class GHGCalculation(Base, TimestampMixin):
"""
Risultato del calcolo GHG per un ActivityRecord.
Separato per permettere ricalcoli senza perdere il dato sorgente.
"""
__tablename__ = "ghg_calculations"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
activity_record_id = Column(UUID(as_uuid=True),
ForeignKey("activity_records.id"), nullable=False, unique=True)
# Emissioni per gas (kg)
co2_kg = Column(Numeric(18, 4), default=0)
ch4_kg = Column(Numeric(18, 4), default=0)
n2o_kg = Column(Numeric(18, 4), default=0)
hfc_kg = Column(Numeric(18, 4), default=0)
# Totale in CO2e (kg e tonnellate)
co2e_kg = Column(Numeric(18, 4), nullable=False)
co2e_tonnes = Column(Numeric(18, 6), nullable=False)
# Metodo di calcolo
calculation_method = Column(String(50))
# ACTIVITY_BASED | SPEND_BASED | AVERAGE_DATA | HYBRID
# Scope 2 specifico
is_location_based = Column(Boolean)
is_market_based = Column(Boolean)
calculated_at = Column(DateTime(timezone=True), server_default=func.now())
calculator_version = Column(String(20), default="1.0.0")
activity_record = relationship("ActivityRecord", back_populates="calculation")Calcule GHG în Python: bazate pe activitate și bazate pe cheltuieli
Cu modelul de date definit, putem implementa calculele. Protocolul GHG recunoaște patru metode de calcul în ordinea descrescătoare a preciziei:
- Metoda specifică furnizorului: Date directe către furnizor (cele mai precise)
- Metoda hibridă: Combinație de date primare și secundare
- Metoda medie a datelor: Factori de emisie medii pe unitate fizică
- Metoda bazată pe cheltuieli: Factori per unitate de cheltuieli în EUR/USD (mai puțin precis)
# calculators/ghg_calculator.py
from decimal import Decimal
from dataclasses import dataclass
from typing import Optional
from models import ActivityRecord, EmissionFactor, GHGCalculation
@dataclass
class CalculationResult:
co2_kg: Decimal
ch4_kg: Decimal
n2o_kg: Decimal
hfc_kg: Decimal
co2e_kg: Decimal
co2e_tonnes: Decimal
method: str
def activity_based_calculation(
quantity: Decimal,
factor: EmissionFactor
) -> CalculationResult:
"""
Calcolo activity-based: quantità * fattore emissione.
Formula GHG Protocol: E = AD * EF * GWP
"""
co2_kg = Decimal(str(quantity)) * Decimal(str(factor.co2_factor or 0))
ch4_kg = Decimal(str(quantity)) * Decimal(str(factor.ch4_factor or 0))
n2o_kg = Decimal(str(quantity)) * Decimal(str(factor.n2o_factor or 0))
hfc_kg = Decimal(str(quantity)) * Decimal(str(factor.hfc_factor or 0))
# Converti in CO2e usando GWP AR5
co2e_kg = (
co2_kg * Decimal(str(factor.gwp_co2))
+ ch4_kg * Decimal(str(factor.gwp_ch4))
+ n2o_kg * Decimal(str(factor.gwp_n2o))
+ hfc_kg * Decimal("1300") # GWP medio HFC-134a
)
return CalculationResult(
co2_kg=co2_kg,
ch4_kg=ch4_kg,
n2o_kg=n2o_kg,
hfc_kg=hfc_kg,
co2e_kg=co2e_kg,
co2e_tonnes=co2e_kg / Decimal("1000"),
method="ACTIVITY_BASED"
)
def spend_based_calculation(
spend_eur: Decimal,
spend_factor_kg_per_eur: Decimal
) -> CalculationResult:
"""
Calcolo spend-based per Cat. 1 Scope 3 quando non si hanno
dati fisici. Usa EXIOBASE o fattori DEFRA per settore NACE.
Meno accurato ma utile come stima iniziale.
"""
co2e_kg = spend_eur * spend_factor_kg_per_eur
return CalculationResult(
co2_kg=co2e_kg, # Semplificazione: tutto attribuito a CO2
ch4_kg=Decimal("0"),
n2o_kg=Decimal("0"),
hfc_kg=Decimal("0"),
co2e_kg=co2e_kg,
co2e_tonnes=co2e_kg / Decimal("1000"),
method="SPEND_BASED"
)
def business_travel_calculation(
distance_km: Decimal,
transport_mode: str,
cabin_class: Optional[str] = None
) -> CalculationResult:
"""
Calcolo Cat. 6 (Business Travel) con fattori DEFRA 2024.
Fattori in kgCO2e per km per passeggero.
"""
# Fattori DEFRA 2024 (kgCO2e/km/passeggero)
FACTORS = {
"SHORT_HAUL_ECONOMY": Decimal("0.15342"), # Volo <1000km economy
"SHORT_HAUL_BUSINESS": Decimal("0.23013"),
"LONG_HAUL_ECONOMY": Decimal("0.19085"), # Volo >1000km economy
"LONG_HAUL_BUSINESS": Decimal("0.57256"), # Business = 3x economy
"LONG_HAUL_FIRST": Decimal("0.76340"),
"RAIL_NATIONAL": Decimal("0.03549"), # Treno nazionale UK
"RAIL_EUROSTAR": Decimal("0.00416"), # Eurostar molto efficiente
"CAR_AVERAGE": Decimal("0.17100"), # Auto media
"CAR_ELECTRIC": Decimal("0.05280"),
"HOTEL_NIGHT": Decimal("20.600"), # Per notte, non per km
}
key = transport_mode
if cabin_class and f"{transport_mode}_{cabin_class}" in FACTORS:
key = f"{transport_mode}_{cabin_class}"
factor = FACTORS.get(key, Decimal("0"))
co2e_kg = distance_km * factor
return CalculationResult(
co2_kg=co2e_kg,
ch4_kg=Decimal("0"),
n2o_kg=Decimal("0"),
hfc_kg=Decimal("0"),
co2e_kg=co2e_kg,
co2e_tonnes=co2e_kg / Decimal("1000"),
method="ACTIVITY_BASED"
)# calculators/scope2_calculator.py
from decimal import Decimal
from dataclasses import dataclass
from typing import Optional
@dataclass
class Scope2Result:
location_based_co2e_kg: Decimal
market_based_co2e_kg: Decimal
renewable_kwh_matched: Decimal
grid_ef_g_kwh: Decimal
market_ef_g_kwh: Decimal
def scope2_dual_calculation(
kwh_consumed: Decimal,
grid_region: str,
go_certificates_kwh: Decimal = Decimal("0"),
supplier_ef_g_kwh: Optional[Decimal] = None
) -> Scope2Result:
"""
Calcolo Scope 2 con doppio metodo obbligatorio (GHG Protocol S2 Guidance).
location_based: usa grid average emission factor (IEA/AIB)
market_based: usa supplier factor o residual mix, dedotti GO/REC
"""
# Grid emission factors per regione (gCO2e/kWh) - IEA 2024
GRID_FACTORS = {
"IT": Decimal("310.0"), # Italia: mix gas + rinnovabili
"DE": Decimal("364.0"), # Germania: ancora carbone
"FR": Decimal("52.0"), # Francia: nucleare dominante
"ES": Decimal("198.0"),
"NL": Decimal("289.0"),
"EU_AVERAGE": Decimal("231.0"),
"UK": Decimal("209.0"),
"US_AVERAGE": Decimal("386.0"),
}
grid_ef = GRID_FACTORS.get(grid_region, GRID_FACTORS["EU_AVERAGE"])
# Location-based (sempre con grid average)
lb_co2e_kg = kwh_consumed * grid_ef / Decimal("1000")
# Market-based:
# 1. Kwh coperti da GO/REC = 0 emissioni
# 2. Kwh residui: usa supplier factor o residual mix
covered_by_go = min(go_certificates_kwh, kwh_consumed)
residual_kwh = kwh_consumed - covered_by_go
# Residual mix factor (tipicamente > grid average perché
# le energie rinnovabili sono "estratte" dal grid average)
# Fonte: AIB European Residual Mixes 2023
RESIDUAL_MIX = {
"IT": Decimal("414.0"), # Residual mix IT più alto del grid
"DE": Decimal("542.0"),
"FR": Decimal("73.0"),
"EU_AVERAGE": Decimal("395.0"),
}
market_ef = supplier_ef_g_kwh or RESIDUAL_MIX.get(grid_region, Decimal("400"))
mb_co2e_kg = residual_kwh * market_ef / Decimal("1000")
return Scope2Result(
location_based_co2e_kg=lb_co2e_kg,
market_based_co2e_kg=mb_co2e_kg,
renewable_kwh_matched=covered_by_go,
grid_ef_g_kwh=grid_ef,
market_ef_g_kwh=market_ef
)# calculators/scope3_commute.py
from decimal import Decimal
from dataclasses import dataclass
from typing import Dict
@dataclass
class CommuteData:
"""Dati raccolti tramite survey annuale dipendenti."""
employee_count: int
avg_working_days: int # Es: 220
remote_work_days_pct: Decimal # Es: 0.40 = 40% remote
transport_split: Dict[str, Decimal]
# {'car_solo': 0.45, 'car_pool': 0.10, 'public_transit': 0.30,
# 'bicycle': 0.10, 'walk': 0.05}
avg_commute_km_oneway: Decimal # km sola andata
def employee_commute_calculation(data: CommuteData) -> Decimal:
"""
Cat. 7 Scope 3: Pendolarismo dipendenti.
Formula: n_dipendenti * giorni_ufficio * km_totali * EF * split_modale
"""
# Fattori emissione (kgCO2e/km/passeggero) - DEFRA 2024
MODE_FACTORS = {
"car_solo": Decimal("0.171"), # Auto singola
"car_pool": Decimal("0.0855"), # Carpooling 2 persone
"public_transit": Decimal("0.089"), # Mix metropolitana/bus
"bicycle": Decimal("0.0"),
"walk": Decimal("0.0"),
"electric_car": Decimal("0.053"),
"motorcycle": Decimal("0.114"),
}
# Giorni effettivi in ufficio
office_days_pct = Decimal("1") - data.remote_work_days_pct
annual_office_days = Decimal(str(data.avg_working_days)) * office_days_pct
km_per_day = data.avg_commute_km_oneway * Decimal("2") # A/R
total_co2e_kg = Decimal("0")
for mode, share in data.transport_split.items():
ef = MODE_FACTORS.get(mode, Decimal("0.1"))
mode_co2e = (
Decimal(str(data.employee_count))
* annual_office_days
* km_per_day
* share
* ef
)
total_co2e_kg += mode_co2e
return total_co2e_kgAgregare anuală și comparație de la an la an
Odată ce emisiile au fost calculate pentru fiecare ActivityRecord, acestea trebuie să fie agregate pe perioadă raportarea, compararea acestora cu anul precedent și verificarea alinierii la țintele SBTi.
# services/reporting_service.py
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from decimal import Decimal
from typing import Dict
from models import GHGCalculation, ActivityRecord, EmissionSource, ReportingPeriod
@dataclass
class AnnualReport:
organization_id: str
year: int
scope1_co2e_tonnes: Decimal
scope2_lb_co2e_tonnes: Decimal # Location-based
scope2_mb_co2e_tonnes: Decimal # Market-based
scope3_co2e_tonnes: Decimal
scope3_by_category: Dict[int, Decimal]
total_co2e_tonnes: Decimal # S1 + S2(MB) + S3
intensity_per_employee: Decimal # tCO2e / FTE
intensity_per_revenue: Decimal # tCO2e / M EUR
def generate_annual_report(
session: Session,
organization_id: str,
year: int,
employee_fte: int,
revenue_million_eur: Decimal
) -> AnnualReport:
"""Aggrega tutti i calcoli GHG per anno fiscale."""
# Query aggregata per scope
stmt = (
select(
EmissionSource.scope,
EmissionSource.scope3_category,
GHGCalculation.is_location_based,
func.sum(GHGCalculation.co2e_tonnes).label("total_tonnes")
)
.join(ActivityRecord, GHGCalculation.activity_record_id == ActivityRecord.id)
.join(EmissionSource, ActivityRecord.source_id == EmissionSource.id)
.where(
EmissionSource.organization_id == organization_id,
func.extract("year", ActivityRecord.period_start) == year
)
.group_by(
EmissionSource.scope,
EmissionSource.scope3_category,
GHGCalculation.is_location_based
)
)
rows = session.execute(stmt).all()
scope1 = Decimal("0")
scope2_lb = Decimal("0")
scope2_mb = Decimal("0")
scope3_by_cat: Dict[int, Decimal] = {}
for row in rows:
tonnes = Decimal(str(row.total_tonnes or 0))
if row.scope == 1:
scope1 += tonnes
elif row.scope == 2:
if row.is_location_based:
scope2_lb += tonnes
else:
scope2_mb += tonnes
elif row.scope == 3:
cat = row.scope3_category or 0
scope3_by_cat[cat] = scope3_by_cat.get(cat, Decimal("0")) + tonnes
scope3_total = sum(scope3_by_cat.values())
total = scope1 + scope2_mb + scope3_total
return AnnualReport(
organization_id=organization_id,
year=year,
scope1_co2e_tonnes=scope1,
scope2_lb_co2e_tonnes=scope2_lb,
scope2_mb_co2e_tonnes=scope2_mb,
scope3_co2e_tonnes=scope3_total,
scope3_by_category=scope3_by_cat,
total_co2e_tonnes=total,
intensity_per_employee=total / Decimal(str(employee_fte)) if employee_fte else Decimal("0"),
intensity_per_revenue=total / revenue_million_eur if revenue_million_eur else Decimal("0")
)
def check_sbti_alignment(
report: AnnualReport,
baseline_year: int,
baseline_emissions: Decimal,
target_year: int = 2030,
reduction_target_pct: Decimal = Decimal("0.50") # SBTi: -50% by 2030
) -> Dict:
"""
Verifica allineamento con target SBTi.
Near-term: -50% entro 2030 (1.5°C pathway).
Net-zero: -90-95% entro 2050.
"""
current_year_progress = report.year - baseline_year
total_years = target_year - baseline_year
linear_target = baseline_emissions * (
1 - reduction_target_pct * Decimal(str(current_year_progress)) / Decimal(str(total_years))
)
actual_reduction_pct = (baseline_emissions - report.total_co2e_tonnes) / baseline_emissions * 100
return {
"baseline_emissions_tonnes": float(baseline_emissions),
"current_emissions_tonnes": float(report.total_co2e_tonnes),
"linear_target_tonnes": float(linear_target),
"actual_reduction_pct": float(actual_reduction_pct),
"on_track": report.total_co2e_tonnes <= linear_target,
"gap_tonnes": float(report.total_co2e_tonnes - linear_target),
"required_annual_reduction_pct": float(
reduction_target_pct / Decimal(str(total_years)) * 100
)
}API REST pentru ESG: puncte finale FastAPI
Un sistem ESG de întreprindere are nevoie de API-uri solide pentru a permite diferitelor echipe să trimită date de activitate, calcule de declanșare și rapoarte de acces. FastAPI este alegerea ideal pentru tipul său de siguranță, documentare automată și performanță.
# api/routers/activities.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from pydantic import BaseModel, validator
from typing import Optional
from datetime import date
from decimal import Decimal
import uuid
router = APIRouter(prefix="/api/v1", tags=["activities"])
class ActivitySubmitRequest(BaseModel):
source_id: uuid.UUID
period_start: date
period_end: date
quantity: Decimal
unit: str
data_quality: str = "MEASURED"
evidence_url: Optional[str] = None
notes: Optional[str] = None
@validator("data_quality")
def validate_quality(cls, v: str) -> str:
allowed = ["MEASURED", "ESTIMATED", "CALCULATED", "MODELED"]
if v not in allowed:
raise ValueError(f"data_quality deve essere uno di: {allowed}")
return v
@validator("period_end")
def validate_dates(cls, v: date, values: dict) -> date:
if "period_start" in values and v < values["period_start"]:
raise ValueError("period_end deve essere >= period_start")
return v
class ActivitySubmitResponse(BaseModel):
activity_id: uuid.UUID
co2e_tonnes: Decimal
calculation_method: str
message: str
@router.post("/activities", response_model=ActivitySubmitResponse, status_code=201)
async def submit_activity(
req: ActivitySubmitRequest,
session: Session = Depends(get_db),
current_org: Organization = Depends(get_current_org)
) -> ActivitySubmitResponse:
"""
Invia un dato di attività e calcola le emissioni GHG.
Validazione automatica: unità coerente con la sorgente.
"""
source = session.get(EmissionSource, req.source_id)
if not source or source.organization_id != current_org.id:
raise HTTPException(status_code=404, detail="Sorgente emissione non trovata")
# Trova il fattore di emissione più recente per questa sorgente
factor = get_best_emission_factor(session, source, req.unit, req.period_start.year)
if not factor:
raise HTTPException(
status_code=422,
detail=f"Nessun fattore emissione trovato per {source.source_type} in {req.unit}"
)
# Calcola emissioni
result = activity_based_calculation(req.quantity, factor)
# Salva ActivityRecord e GHGCalculation
record = ActivityRecord(
source_id=req.source_id,
factor_id=factor.id,
period_start=req.period_start,
period_end=req.period_end,
quantity=req.quantity,
unit=req.unit,
data_quality=req.data_quality,
evidence_url=req.evidence_url,
notes=req.notes
)
session.add(record)
session.flush() # Get the ID
calc = GHGCalculation(
activity_record_id=record.id,
co2_kg=result.co2_kg,
ch4_kg=result.ch4_kg,
n2o_kg=result.n2o_kg,
co2e_kg=result.co2e_kg,
co2e_tonnes=result.co2e_tonnes,
calculation_method=result.method
)
session.add(calc)
session.commit()
return ActivitySubmitResponse(
activity_id=record.id,
co2e_tonnes=result.co2e_tonnes,
calculation_method=result.method,
message=f"Emissioni calcolate: {result.co2e_tonnes:.4f} tCO2e"
)# api/routers/reports.py
from fastapi import APIRouter, Query
from pydantic import BaseModel
from typing import Dict
from decimal import Decimal
router = APIRouter(prefix="/api/v1", tags=["reports"])
class ESGReportResponse(BaseModel):
organization_id: str
year: int
scope1_co2e_tonnes: float
scope2_location_based_tonnes: float
scope2_market_based_tonnes: float
scope3_co2e_tonnes: float
scope3_breakdown: Dict[int, float]
total_co2e_tonnes: float
intensity_per_employee: float
intensity_per_revenue_m_eur: float
sbti_alignment: Dict
class YoYComparisonResponse(BaseModel):
year_current: int
year_previous: int
total_change_pct: float
scope1_change_pct: float
scope2_change_pct: float
scope3_change_pct: float
top_increasing_categories: list
top_decreasing_categories: list
@router.get("/reports/annual/{year}", response_model=ESGReportResponse)
async def get_annual_report(
year: int,
session: Session = Depends(get_db),
current_org: Organization = Depends(get_current_org)
) -> ESGReportResponse:
"""
Genera report ESG annuale con tutte le metriche GHG Protocol.
Include verifica allineamento SBTi.
"""
employee_fte = get_employee_fte(session, current_org.id, year)
revenue = get_annual_revenue(session, current_org.id, year)
report = generate_annual_report(
session, str(current_org.id), year, employee_fte, revenue
)
baseline = get_baseline_emissions(session, str(current_org.id))
sbti = check_sbti_alignment(report, baseline["year"], baseline["tonnes"])
return ESGReportResponse(
organization_id=str(current_org.id),
year=year,
scope1_co2e_tonnes=float(report.scope1_co2e_tonnes),
scope2_location_based_tonnes=float(report.scope2_lb_co2e_tonnes),
scope2_market_based_tonnes=float(report.scope2_mb_co2e_tonnes),
scope3_co2e_tonnes=float(report.scope3_co2e_tonnes),
scope3_breakdown={k: float(v) for k, v in report.scope3_by_category.items()},
total_co2e_tonnes=float(report.total_co2e_tonnes),
intensity_per_employee=float(report.intensity_per_employee),
intensity_per_revenue_m_eur=float(report.intensity_per_revenue),
sbti_alignment=sbti
)
@router.get("/reports/yoy/{year}", response_model=YoYComparisonResponse)
async def year_over_year(year: int, ...) -> YoYComparisonResponse:
"""Confronto Year-over-Year con identificazione trend per categoria."""
curr = generate_annual_report(session, org_id, year, fte, rev)
prev = generate_annual_report(session, org_id, year - 1, fte_prev, rev_prev)
def pct_change(curr_val: Decimal, prev_val: Decimal) -> float:
if prev_val == 0:
return 0.0
return float((curr_val - prev_val) / prev_val * 100)
return YoYComparisonResponse(
year_current=year,
year_previous=year - 1,
total_change_pct=pct_change(curr.total_co2e_tonnes, prev.total_co2e_tonnes),
scope1_change_pct=pct_change(curr.scope1_co2e_tonnes, prev.scope1_co2e_tonnes),
scope2_change_pct=pct_change(curr.scope2_mb_co2e_tonnes, prev.scope2_mb_co2e_tonnes),
scope3_change_pct=pct_change(curr.scope3_co2e_tonnes, prev.scope3_co2e_tonnes),
top_increasing_categories=get_top_changes(curr, prev, ascending=False),
top_decreasing_categories=get_top_changes(curr, prev, ascending=True)
)Testare: Test unitar pentru calculele GES
Calculele GES trebuie testate riguros: erori în factorii de emisie sau formulele conduc la raportare incorectă ESG, cu potențiale implicații legale (greenwashing) și sancțiuni în temeiul CSRD. Testarea nu este opțională.
# tests/test_ghg_calculator.py
import pytest
from decimal import Decimal
from calculators.ghg_calculator import (
activity_based_calculation,
spend_based_calculation,
business_travel_calculation
)
from calculators.scope2_calculator import scope2_dual_calculation
from calculators.scope3_commute import employee_commute_calculation, CommuteData
from models import EmissionFactor
@pytest.fixture
def natural_gas_factor() -> EmissionFactor:
"""Fattore gas naturale DEFRA 2024: 2.04264 kgCO2e/m3."""
factor = EmissionFactor()
factor.co2_factor = Decimal("1.88900") # kgCO2/m3
factor.ch4_factor = Decimal("0.00011") # kgCH4/m3
factor.n2o_factor = Decimal("0.00003") # kgN2O/m3
factor.hfc_factor = Decimal("0")
factor.gwp_co2 = Decimal("1.0")
factor.gwp_ch4 = Decimal("28.0")
factor.gwp_n2o = Decimal("265.0")
return factor
class TestActivityBasedCalculation:
def test_natural_gas_calculation(self, natural_gas_factor):
"""Test: 1000 m3 gas naturale = ~2.04 tCO2e."""
result = activity_based_calculation(Decimal("1000"), natural_gas_factor)
assert result.method == "ACTIVITY_BASED"
# CO2: 1000 * 1.889 = 1889 kg
assert abs(result.co2_kg - Decimal("1889.00")) < Decimal("0.01")
# CH4: 1000 * 0.00011 * 28 = 3.08 kgCO2e
expected_co2e = (
Decimal("1889.00") * Decimal("1.0") # CO2
+ Decimal("0.11") * Decimal("28.0") # CH4
+ Decimal("0.03") * Decimal("265.0") # N2O
)
assert abs(result.co2e_kg - expected_co2e) < Decimal("1.0")
assert result.co2e_tonnes == result.co2e_kg / Decimal("1000")
def test_zero_consumption(self, natural_gas_factor):
"""Zero consumo = zero emissioni."""
result = activity_based_calculation(Decimal("0"), natural_gas_factor)
assert result.co2e_kg == Decimal("0")
def test_precision_decimal(self, natural_gas_factor):
"""Verifica che i calcoli usino Decimal per evitare floating point errors."""
result = activity_based_calculation(Decimal("333.333"), natural_gas_factor)
# Non deve avere floating point drift
assert isinstance(result.co2e_kg, Decimal)
class TestScope2DualMethod:
def test_italy_location_based(self):
"""Test location-based Italia: 10.000 kWh * 310 gCO2e/kWh = 3100 kg."""
result = scope2_dual_calculation(
kwh_consumed=Decimal("10000"),
grid_region="IT"
)
assert abs(result.location_based_co2e_kg - Decimal("3100.0")) < Decimal("1.0")
def test_full_go_coverage_zeroes_market_based(self):
"""Con GO = consumo totale, market-based deve essere 0."""
result = scope2_dual_calculation(
kwh_consumed=Decimal("10000"),
grid_region="IT",
go_certificates_kwh=Decimal("10000")
)
assert result.market_based_co2e_kg == Decimal("0")
assert result.renewable_kwh_matched == Decimal("10000")
def test_partial_go_coverage(self):
"""Con GO parziali, market-based usa residual mix sul rimanente."""
result = scope2_dual_calculation(
kwh_consumed=Decimal("10000"),
grid_region="IT",
go_certificates_kwh=Decimal("6000")
)
# 4000 kWh residui * 414 gCO2e/kWh (residual mix IT) / 1000
expected = Decimal("4000") * Decimal("414") / Decimal("1000")
assert abs(result.market_based_co2e_kg - expected) < Decimal("0.01")
class TestEmployeeCommute:
def test_standard_commute(self):
"""Test pendolarismo con mix modale realistico."""
data = CommuteData(
employee_count=200,
avg_working_days=220,
remote_work_days_pct=Decimal("0.40"),
transport_split={
"car_solo": Decimal("0.45"),
"public_transit": Decimal("0.40"),
"bicycle": Decimal("0.10"),
"walk": Decimal("0.05")
},
avg_commute_km_oneway=Decimal("15")
)
result = employee_commute_calculation(data)
assert result > Decimal("0")
# Ordine di grandezza: ~50-150 tCO2e per 200 dipendenti
assert Decimal("40000") < result < Decimal("200000") # in kg
def test_full_remote_work_zero(self):
"""100% remote work = 0 emissioni pendolarismo."""
data = CommuteData(
employee_count=200,
avg_working_days=220,
remote_work_days_pct=Decimal("1.0"),
transport_split={"car_solo": Decimal("1.0")},
avg_commute_km_oneway=Decimal("20")
)
result = employee_commute_calculation(data)
assert result == Decimal("0")Atenție: Factori de problemă și de versiune
Factorii de emisie sunt actualizați în fiecare an (DEFRA, IEA, EPA). Un sistem ESG corect trebuie:
- Versiune factorii din baza de date cu anul și sursa explicite
- Utilizați factorul anul de raportare (nu cel mai recent)
- Permiteți recalculări istorice dacă factorii sunt corectați
- Menține urmărirea de audit: care factor era activ la momentul calculului
- Nu actualizați niciodată un calcul GHGCalculator în loc: creați întotdeauna o versiune nouă
Studiu de caz: Software pentru IMM-uri cu 200 de angajați
Să vedem un exemplu concret de la capăt la capăt. TechFlow S.r.l. este un IMM Software italian cu 200 de angajați, sediu principal în Milano (închiriat), un sediu secundar în Roma, fără centru de date propriu (toate pe AWS eu-west-1 și Azure westeurope).
Profilul companiei TechFlow S.r.l.
| Parametru | Valoare |
|---|---|
| Angajatii | 200 FTE (70% Milano, 20% Roma, 10% full-remote) |
| Vânzări | 15 milioane EUR |
| Birouri | 2 locatii de inchiriat (total 2.400 m²) |
| infrastructura IT | 100% cloud (AWS + Azure), zero servere fizice |
| Flota | 3 masini de companie diesel, 1 autoutilitara diesel |
| Călătorii de afaceri | ~120 de zboruri/an, 40 de nopți de hotel, 200 de rute de tren |
# Esempio calcolo completo TechFlow S.r.l. - Anno 2024
from decimal import Decimal
from calculators.ghg_calculator import activity_based_calculation, business_travel_calculation
from calculators.scope2_calculator import scope2_dual_calculation
from calculators.scope3_commute import employee_commute_calculation, CommuteData
# ---- SCOPE 1 ----
# Gas naturale: riscaldamento uffici (caldaie condominiali, incluse nel canone)
# TechFlow ha controllo operativo: include le caldaie
scope1_gas_heating = activity_based_calculation(
Decimal("12500"), # m3 gas naturale annui (entrambe le sedi)
NATURAL_GAS_FACTOR_IT_2024
)
# Risultato atteso: ~25.5 tCO2e
# Flotta aziendale: 4 veicoli diesel
scope1_fleet_km = activity_based_calculation(
Decimal("85000"), # km totali flotta annui
DIESEL_MOBILE_FACTOR_2024 # 0.16844 kgCO2e/km
)
# Risultato atteso: ~14.3 tCO2e
# Refrigeranti HVAC: perdita stimata 2% del carico R-410A
scope1_refrigerants = activity_based_calculation(
Decimal("1.5"), # kg R-410A persi
R410A_FACTOR # GWP=2088 kgCO2e/kg
)
# Risultato atteso: ~3.1 tCO2e
scope1_total = (
scope1_gas_heating.co2e_tonnes
+ scope1_fleet_km.co2e_tonnes
+ scope1_refrigerants.co2e_tonnes
)
print(f"Scope 1 totale: {scope1_total:.2f} tCO2e")
# Output: Scope 1 totale: 42.90 tCO2e
# ---- SCOPE 2 ----
# Elettricità Milano: 320.000 kWh, GO acquistate: 200.000 kWh
scope2_milano = scope2_dual_calculation(
kwh_consumed=Decimal("320000"),
grid_region="IT",
go_certificates_kwh=Decimal("200000")
)
# Elettricità Roma: 95.000 kWh, nessuna GO
scope2_roma = scope2_dual_calculation(
kwh_consumed=Decimal("95000"),
grid_region="IT",
go_certificates_kwh=Decimal("0")
)
scope2_lb_total = scope2_milano.location_based_co2e_kg + scope2_roma.location_based_co2e_kg
scope2_mb_total = scope2_milano.market_based_co2e_kg + scope2_roma.market_based_co2e_kg
print(f"Scope 2 Location-Based: {scope2_lb_total/1000:.2f} tCO2e")
# Output: Scope 2 Location-Based: 128.45 tCO2e
print(f"Scope 2 Market-Based: {scope2_mb_total/1000:.2f} tCO2e")
# Output: Scope 2 Market-Based: 55.12 tCO2e (ridotto grazie alle GO)
# ---- SCOPE 3 ----
# Cat. 1: Cloud computing (AWS+Azure) - spend-based
# Spesa totale cloud: EUR 480.000/anno
# Fattore EXIOBASE settore "Computer Services" IT: ~0.062 kgCO2e/EUR
scope3_cat1_cloud = spend_based_calculation(
spend_eur=Decimal("480000"),
spend_factor_kg_per_eur=Decimal("0.062")
)
# Risultato atteso: ~29.8 tCO2e
# NOTA: Provider-specific è più accurato se disponibile
# AWS Carbon Footprint: ~18 tCO2e (più basso per energia rinnovabile AWS)
# Cat. 2: Capital goods - laptop/monitor (life cycle amortized)
# 40 laptop nuovi a EUR 1.200 cad, ammortizzati su 4 anni
scope3_cat2_laptops = spend_based_calculation(
spend_eur=Decimal("40") * Decimal("1200") / Decimal("4"),
spend_factor_kg_per_eur=Decimal("0.35") # Electronic equipment
)
# Risultato atteso: ~4.2 tCO2e
# Cat. 6: Business travel
scope3_cat6_flights = business_travel_calculation(
distance_km=Decimal("95000"), # Totale km voli
transport_mode="SHORT_HAUL_ECONOMY" # Mix short/medium haul
)
scope3_cat6_rail = business_travel_calculation(
distance_km=Decimal("18000"), # Totale km treno
transport_mode="RAIL_NATIONAL"
)
scope3_cat6_hotel = business_travel_calculation(
distance_km=Decimal("40"), # 40 notti hotel
transport_mode="HOTEL_NIGHT"
)
scope3_cat6_total = (
scope3_cat6_flights.co2e_tonnes
+ scope3_cat6_rail.co2e_tonnes
+ scope3_cat6_hotel.co2e_tonnes
)
# Risultato atteso: ~16.3 tCO2e
# Cat. 7: Employee commuting
commute_data = CommuteData(
employee_count=180, # Esclude 20 full-remote
avg_working_days=220,
remote_work_days_pct=Decimal("0.35"), # Media 35% smart working
transport_split={
"car_solo": Decimal("0.38"),
"public_transit": Decimal("0.45"),
"bicycle": Decimal("0.08"),
"walk": Decimal("0.06"),
"car_pool": Decimal("0.03")
},
avg_commute_km_oneway=Decimal("12")
)
scope3_cat7_kg = employee_commute_calculation(commute_data)
# Risultato atteso: ~68.400 kg = 68.4 tCO2e
# ---- RIEPILOGO FINALE ----
scope3_total = (
scope3_cat1_cloud.co2e_tonnes
+ scope3_cat2_laptops.co2e_tonnes
+ scope3_cat6_total
+ scope3_cat7_kg / Decimal("1000")
# + Cat. 11 Use of sold products (SaaS): stimato ~15 tCO2e
# + Cat. 3 T&D losses: ~8.2 tCO2e
)
grand_total = scope1_total + scope2_mb_total / Decimal("1000") + scope3_total
print("\n=== TECHFLOW S.R.L. - GHG INVENTORY 2024 ===")
print(f"Scope 1: {float(scope1_total):8.1f} tCO2e ( {float(scope1_total/grand_total)*100:4.1f}%)")
print(f"Scope 2: {float(scope2_mb_total/1000):8.1f} tCO2e ( {float(scope2_mb_total/1000/grand_total)*100:4.1f}%)")
print(f"Scope 3: {float(scope3_total):8.1f} tCO2e ( {float(scope3_total/grand_total)*100:4.1f}%)")
print(f"TOTALE: {float(grand_total):8.1f} tCO2e")
print(f"Intensità: {float(grand_total/200):5.2f} tCO2e/dipendente")
# === TECHFLOW S.R.L. - GHG INVENTORY 2024 ===
# Scope 1: 42.9 tCO2e ( 18.2%)
# Scope 2: 55.1 tCO2e ( 23.4%)
# Scope 3: 137.8 tCO2e ( 58.4%)
# TOTALE: 235.8 tCO2e
# Intensità: 1.18 tCO2e/dipendenteBenchmark: Intensitatea carbonului pentru companiile de software (2024)
| Dimensiune | tCO₂e/dependent | % Domeniul de aplicare 3 | Intrarea principală S3 |
|---|---|---|---|
| Startup-uri (<50 de angajați) | 0,8 - 1,2 | 65-75% | Călătorii de afaceri |
| IMM-uri de software (50-500) | 1,0 - 1,8 | 55-70% | Naveta + nor |
| Tehnologia companiei (500+) | 1,5 - 3,0 | 70-85% | Cloud + utilizarea produselor |
| Big Tech (hiperscaler) | 3,0 - 8,0 | 75-90% | Lanț de aprovizionare hardware + centre de date |
| TechFlow (studiu de caz) | 1.18 | 58% | Naveta |
Oportunități de reducere pentru TechFlow
Din analiza Inventarului de GES 2024, reies cele mai eficiente pârghii de reducere:
- Lucrul inteligent extins la 50%: Pisică. 7 reducere de ~35% (-24 tCO2e). Naveta este cel mai mare element.
- 100% GO pentru ambele locații: Scopul 2 bazat pe piață scade la ~0 tCO2e (-55 tCO2e). Cost estimat GO: 3.000-5.000 EUR/an.
- Politica bugetului de carbon pentru călătorii: Înlocuirea zborurilor scurte (<3h) cu trenul. Pisică. 6 reducere de ~40% (-6,5 tCO2e).
- Instrumentul AWS pentru amprenta de carbon: Treceți de la estimarea bazată pe cheltuieli la un anumit furnizor reduce Cat. 1 nor cu ~40% (-12 tCO2e) și îmbunătățește calitatea datelor.
Foaia de parcurs de decarbonizare TechFlow — SBTi țintă 1,5°C
| An | Țintă (tCO₂e) | Reducere față de 2024 | Principalele acțiuni |
|---|---|---|---|
| 2024 (linie de referință) | 235,8 | - | Prima raportare CSRD |
| 2025 | 211,0 | -10,5% | GO 100%, lucru inteligent 50% |
| 2027 | 165,0 | -30% | Flota electrica, politica de calatorie |
| 2030 | 118,0 | -50% (SBTi pe termen scurt) | Furnizori de cloud verzi, la distanță mai întâi |
| 2050 | <24 | -90% (SBTi net-zero) | Reziduu neutralizat, fără compensare |
Concluzii și pașii următori
Modelarea datelor pentru raportarea ESG nu este un simplu exercițiu de conformitate: este baza pe care să construim o strategie de decarbonizare credibilă și verificabilă. Am văzut cum se structurează un sistem care acoperă întregul lanț valoric al unei companii software, de la cazanul de încălzire (Scope 1) până la călătoria cu avionul comercial (Scope 3 Cat. 6) până la energia consumată de utilizatori pentru a rula produsul (Scope 3 Cat. 11).
Puncte cheie de luat cu tine:
- Domeniul 3 domină: Pentru companiile de software, 55-75% din emisii sunt Scope 3. A nu-l modela corect înseamnă a avea o imagine distorsionată a amprentei tale de carbon.
- Raportare dublă Domeniul 2: ESRS E1 necesită ambele metode (bazate pe locație și bazate pe piata). Bazat pe piață încurajează achiziționarea de energie regenerabilă certificată.
- Calitatea datelor este totul: Ierarhia bazată pe activitate > bazată pe cheltuieli trebuie să ghideze alegerile de colectare a datelor. Începeți cu estimări bazate pe cheltuieli și îmbunătățiți-le an de an.
- Imuabilitatea calculelor: Nu actualizați niciodată un calcul GHGC. Factori de versiune și menținerea unei piste de audit complete pentru revizuirea auditorului ESG.
- SBTi ca stea nordică: Ținta bazată pe știință (-50% până în 2030, -90% până în 2050) trebuie să conducă prioritățile de reducere, nu respectarea minimă CSRD.
Următoarele articole din serie
În următorul articol al seriei, „ESG și CSRD: obligații europene pentru companiile din domeniul tehnologiei”, haideți să pătrundem în cadrul de reglementare complet: cine face obiectul CSRD, termenele pentru IMM-uri, conceptul de evaluare a dublei semnificații, precum și sistemul de modelare a datelor construit în acest articol se traduce în dezvăluirile ESRS E1 solicitate de la Autoritate.
Vedeți și articolele conexe:
- Seria MLOps: Cum se integrează urmărirea carbonului în conductele ML (metrici de sustenabilitate în registrul modelului MLflow)
- Seria de inginerie AI: Amprenta de carbon a sistemelor RAG și LLM în producție
- Articolul 9 (această serie): Conducta Scope 3 — Automatizați colectarea datelor din lanțul valoric cu conectori API pentru furnizori