Zakres 1, 2 i 3: Modelowanie danych dla oprogramowania do raportowania ESG
Il Protokół dotyczący gazów cieplarnianych (Protokół w sprawie gazów cieplarnianych) stał się de facto uniwersalnymi ramami do pomiaru i raportowania emisji gazów cieplarnianych przez przedsiębiorstwa. Opracowany przez Światowy Instytut Zasobów (WRI) oraz Światowa Rada Biznesu na rzecz Zrównoważonego Rozwoju (WBCSD) definiuje trzy „zakresy” emisji, które obejmują cały łańcuch wartości organizacji.
Wraz z wejściem w życie CSRD (dyrektywa w sprawie raportowania zrównoważonego rozwoju przedsiębiorstwa) Unii Europejski i standardowy ESRS E1, tysiące przedsiębiorstw — w tym korporacje oprogramowanie i technologie – mają obecnie obowiązek gromadzenia, obliczania i publikowania własnych danych na temat emisji według rygorystycznych metodologii. Od 2024 r. muszą to robić duże firmy zatrudniające ponad 500 pracowników rozpocząć zbieranie danych, z pierwszym obowiązkiem sprawozdawczym w 2025 r.; firmy zatrudniające powyżej 250 pracowników będą one stosowane począwszy od 2026 r.
Ale dlaczego ja wywoływacz czy muszą rozumieć zakres 1, 2 i 3? Po co budować system Niezawodny ESG to nie tylko problem menedżerów ds. zrównoważonego rozwoju: to wyzwanie inżynieria danych. Wymaga spójnego modelu danych, dokładnych potoków obliczeniowych, interfejsów API dla zbieranie działań i raporty gotowe do audytu. Decyduje kto projektuje i wdraża te systemy jakość – a co za tym idzie – wiarygodność – całej sprawozdawczości ESG.
W tym artykule zbudujemy system modelowania danych do raportowania ESG, od definicji zakresów po tabele bazy danych, od obliczeń w Pythonie po punkty końcowe FastAPI, aż do kompletnego studium przypadku dla MŚP zajmującego się oprogramowaniem zatrudniającego 200 pracowników.
Czego się nauczysz
- Struktura Protokołu GHG: Zakres 1, 2 i 3 z dokładnymi definicjami
- Modele relacji podmiotowych dla emisji bezpośrednich, pośrednich i łańcuchów wartości
- Oparte na lokalizacji a rynkowe dla Zakresu 2: różnice i wpływ na dane
- 15 kategorii z zakresu 3 i które z nich są istotne dla producentów oprogramowania
- Kompletny schemat SQLAlchemy: Organizacja, Obiekt, EmissionSource, EmissionFactor, ActivityData
- Obliczenia w Pythonie z estymacją opartą na aktywności i estymacją opartą na wydatkach
- Agregacja, porównanie rok do roku i dostosowanie do celów SBTi
- REST API z FastAPI do przesyłania zadań, obliczania emisji i generowania raportów
- Testy jednostkowe do obliczeń GHG z pytestem i walidacją współczynników emisji
- Kompleksowe studium przypadku: Oprogramowanie dla MŚP 200 pracowników, zakres 1+2+3 ukończony
Seria Green Software — 10 artykułów
| # | Tytuł | Państwo |
|---|---|---|
| 1 | Zasady zielonej inżynierii oprogramowania i SCI | Opublikowany |
| 2 | CodeCarbon: Pomiar emisji kodu Pythona | Opublikowany |
| 3 | Zestaw SDK Carbon Aware: przesunięcie obciążenia i przesunięcie w czasie | Opublikowany |
| 4 | Climatiq API: współczynniki emisji i obliczanie emisji dwutlenku węgla | Opublikowany |
| 5 | Zakres 1, 2 i 3: Modelowanie danych na potrzeby raportowania ESG | Aktualny artykuł |
| 6 | ESG i CSRD: Europejskie obowiązki firm technologicznych | Wkrótce |
| 7 | Zrównoważone wzorce oprogramowania: architektury o niskim wpływie | Wkrótce |
| 8 | GreenOps: Optymalizacja chmury pod kątem zrównoważonego rozwoju | Wkrótce |
| 9 | Zakres 3 Pipeline: Automatyzacja łańcucha wartości | Wkrótce |
| 10 | Ślad węglowy AI: LLM, szkolenie i wnioskowanie | Wkrótce |
Protokół dotyczący gazów cieplarnianych: uniwersalne ramy dotyczące emisji
Standard korporacyjny dotyczący protokołu GHG, opublikowany w 2001 r. i zaktualizowany w 2004 r., stanowi podstawę prawie wszystkie ramy raportowania klimatycznego: CDP, TCFD, CSRD/ESRS E1, SBTi i wiele innych odnoszą się bezpośrednio do tego. Jego siła leży w przejrzystości pojęciowej: wszystkie emisje organizację dzieli się na trzy wzajemnie wykluczające się, ale ogólnie wyczerpujące kategorie.
Framework wykorzystuje koncepcję granice organizacyjne (granica organizacyjna) dla określić, które emisje należy uwzględnić w obliczeniach. Istnieją dwa podejścia:metoda udziałów kapitałowych (w oparciu o wskaźnik partycypacji finansowej) oraz podejście kontrolne (w oparciu o kontrolę operacyjne lub finansowe). Większość firm przyjmuje kontrolę operacyjną jako główne kryterium.
Trzy zakresy: przegląd
| Miotły | Definicja | Typowe przykłady | ESRS E1 obowiązkowy |
|---|---|---|---|
| Zakres 1 | Bezpośrednie emisje ze źródeł będących własnością organizacji lub przez nią kontrolowanych | Kotły na gaz ziemny, flota firmowa, generatory diesla, procesy przemysłowe | Obowiązkowy |
| Zakres 2 | Emisje pośrednie z zakupu energii (prądu, pary, ciepła, chłodu) | Energia elektryczna dla biura, chłodzenie centrum danych, ciepłownictwo miejskie | Obowiązkowe (obie metody) |
| Zakres 3 | Wszystkie inne emisje pośrednie w łańcuchu wartości (na wyższym i niższym szczeblu łańcucha dostaw) | Podróże służbowe, dojazdy do pracy, zakup towarów/usług, korzystanie ze sprzedanych produktów | Obowiązkowe (kategorie materiałów) |
Zakres 1: Model danych dla emisji bezpośrednich
Emisje z zakresu 1 pochodzą ze źródeł fizycznych bezpośrednio kontrolowanych przez firmę. Dla nich firmy zajmujące się oprogramowaniem, są to zazwyczaj małe, ale nie bez znaczenia: obejmują one ogrzewanie biur (gaz ziemny), ewentualnych generatorów rezerwowych oraz floty firmowej (samochody służbowe lub samochody dostawcze sprzętu).
Protokół GHG identyfikuje cztery kategorie emisji Zakresu 1:
- Spalanie stacjonarne: Kotły, grzejniki, generatory spalające paliwa kopalne
- Spalanie mobilne: Pojazdy firmowe zasilane benzyną, olejem napędowym lub LPG
- Emisje procesowe: Reakcje chemiczne lub biologiczne (rzadko związane z oprogramowaniem)
- Emisje ulotne: Wycieki czynników chłodniczych (R-410A, R-134a) z instalacji HVAC
Podstawowy wzór na obliczenie Zakresu 1 to:
Emisje (tCO₂e) = Dane dotyczące działalności × Współczynnik emisji × GWP
Gdzie:
- Podana aktywność: ilość zużytego paliwa (litry, m³, kWh)
- Współczynnik emisji: kg CO₂ na jednostkę paliwa (źródło: IPCC, DEFRA, IEA)
- GWP: Potencjał globalnego ocieplenia (CO₂=1, CH₄=28, N₂O=265, HFC zmienia się)
Diagram relacji encji dla Zakresu 1 obejmuje następujące główne encje:
-- 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_byZakres 2: oparte na lokalizacji a oparte na rynku
Zakres 2 jest szczególnie złożony dla firm technologicznych ze względu na zużytą energię elektryczną z biur i centrów danych jest często głównym źródłem emisji pośrednich. Wytyczne dotyczące zakresu 2 protokołu GHG (2015) wprowadziły obowiązek raportowania Zarówno metody: lokalizacyjna i rynkowa.
Oparte na lokalizacji a oparte na rynku: kluczowe różnice
| Czekam | Oparte na lokalizacji | Oparte na rynku |
|---|---|---|
| Definicja | Średnie natężenie lokalnej sieci elektroenergetycznej, w której następuje pobór | Emisje z generatorów, od których energia jest kupowana umownie |
| Współczynnik emisji | Średni współczynnik emisji sieci (gCO₂/kWh) według kraju/regionu | Czynnik specyficzny dla dostawcy lub współczynnik mieszanki resztkowej |
| Instrumenty umowne | Nie dotyczy | GO (Gwarancje Pochodzenia), REC (Certyfikaty Energii Odnawialnej), PPA |
| Przykład Włochy 2024 | ~310 gCO₂/kWh (dane AIB dotyczące mieszanki resztkowej) | 0 gCO₂/kWh w przypadku zakupu w 100% zweryfikowanych odnawialnych źródeł energii |
| Propozycja protokołu dotyczącego gazów cieplarnianych 2025 | Nowa hierarchia czynników z precyzją czasową | Poproś o dopasowanie godzinowe dla odnawialnych źródeł energii |
| Zastosowanie w ESRS E1 | Obowiązkowe ujawnianie informacji (E1-6) | Obowiązkowe ujawnianie informacji (E1-6) |
Aby poprawnie modelować Scope 2 w bazie danych, musisz zarządzać świadectwa energetyczne (GO/REC) jako odrębne byty, które są „konsumowane” względem zużycia energii elektrycznej przez przedsiębiorstwo, obniżając współczynnik rynkowy do zera:
-- 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_valueZakres 3: 15 kategorii łańcucha wartości
Zakres 3 to zazwyczaj główne źródło emisji dla producentów oprogramowania: może stanowić ponad 70-80% całości. Wytyczne techniczne dotyczące obliczeń protokołu GHG Zakres 3 Emisje definiuje 15 kategorii z podziałem na wydobycie (związane z zakupem towarów i usług) i downstream (związany z wykorzystaniem sprzedanych produktów).
15 kategorii z zakresu 3 dla producentów oprogramowania
| # | Kategoria | Typ | Znaczenie oprogramowania | Metoda obliczeniowa |
|---|---|---|---|---|
| 1 | Zakupione towary i usługi | Pod prąd | Wysoki (SaaS, chmura, sprzęt) | Oparte na wydatkach lub specyficzne dla dostawcy |
| 2 | Dobra kapitałowe | Pod prąd | Media (laptopy, serwery, meble) | Jednostka oparta na wydatkach lub jednostka fizyczna |
| 3 | Działalność związana z paliwami i energią | Pod prąd | Średnia (straty T&D, paliwo na wydobyciu) | Oparte na aktywności |
| 4 | Transport i dystrybucja wyższego szczebla | Pod prąd | Niski (dostawa sprzętu) | Na podstawie wydatków lub odległości |
| 5 | Odpady powstające w trakcie działalności | Pod prąd | Niski | Specyficzne dla rodzaju odpadów |
| 6 | Podróże biznesowe | Pod prąd | Wysoki (loty, hotele, pociągi) | Na podstawie odległości lub wydatków |
| 7 | Dojazdy pracowników | Pod prąd | Wysoki (dojazdy, praca zdalna) | Ankieta + oparta na odległości |
| 8 | Wydobywcze aktywa dzierżawione | Pod prąd | Średni (biura do wynajęcia) | Specyficzne dla aktywów |
| 9 | Transport dolny | W dół rzeki | Niski | Oparte na odległości |
| 10 | Przetwarzanie sprzedanych produktów | W dół rzeki | Nie dotyczy (oprogramowanie) | Oparte na aktywności |
| 11 | Korzystanie ze sprzedanych produktów | W dół rzeki | Wysoki (energia do uruchomienia oprogramowania) | Zużycie energii przez całe życie |
| 12 | Leczenie końca życia | W dół rzeki | Niski (urządzenia użytkownika) | Specyficzne dla rodzaju odpadów |
| 13 | Aktywa dzierżawione na niższym szczeblu łańcucha dostaw | W dół rzeki | Niski | Specyficzne dla aktywów |
| 14 | Franczyzy | W dół rzeki | Nie dotyczy | Nie dotyczy |
| 15 | Inwestycje | W dół rzeki | Media (portfolio VC/startupowe) | Oparte na portfelu |
W przypadku producentów oprogramowania najodpowiedniejsze kategorie to zazwyczaj: Kot. 1 (zakupiono chmurę obliczeniową), Kot. 6 (podróże służbowe), Kot. 7 (dojazdy pracowników) e Kot. 11 (energia zużywana przez użytkowników do uruchomienia oprogramowania). Ten ostatni jest często niedoceniany, ale w przypadku korporacyjnego SaaS z milionami użytkowników może być ogromna.
Schemat SQLAlchemy: kompletny model
Zbudujmy teraz kompletny schemat bazy danych w SQLAlchemy 2.0, który obsługuje wszystkie trzy zakresy. Projekt opiera się na zasadach wysokiej spójności i niskiego sprzężenia, z osobnymi tabelami dla każdego koncepcja dominacji i jasnych relacji.
# 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")Obliczenia GHG w Pythonie: oparte na działaniach i oparte na wydatkach
Mając zdefiniowany model danych, możemy przystąpić do obliczeń. Protokół GHG uznaje cztery metody obliczeń w malejącej kolejności dokładności:
- Metoda specyficzna dla dostawcy: Dane bezpośrednie dostawcy (najdokładniejsze)
- Metoda hybrydowa: Połączenie danych pierwotnych i wtórnych
- Metoda danych średnich: Średnie współczynniki emisji na jednostkę fizyczną
- Metoda oparta na wydatkach: Współczynniki na jednostkę wydatków w EUR/USD (mniej dokładne)
# 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_kgAgregacja roczna i porównanie rok do roku
Po obliczeniu emisji dla każdego rekordu działania należy je zagregować według okresu sprawozdawczości, porównać je z rokiem poprzednim i zweryfikować zgodność z celami 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
)
}REST API dla ESG: Punkty końcowe FastAPI
Korporacyjny system ESG wymaga solidnych interfejsów API, aby umożliwić wysyłanie wiadomości różnym zespołom dane dotyczące aktywności, obliczenia wyzwalaczy i raporty dostępu. FastAPI to wybór idealny ze względu na bezpieczeństwo typu, automatyczną dokumentację i wydajność.
# 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)
)Testowanie: Test jednostkowy do obliczeń gazów cieplarnianych
Obliczenia GHG muszą zostać rygorystycznie przetestowane: błędy we współczynnikach emisji lub formuły prowadzą do nieprawidłowego raportowania ESG, z potencjalnymi konsekwencjami prawnymi (greenwashing) i sankcje na mocy CSRD. Testowanie nie jest opcjonalne.
# 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")Uwaga: czynniki dotyczące problemu i wersji
Corocznie aktualizowane są współczynniki emisji (DEFRA, IEA, EPA). System ESG poprawne musi:
- Wersja czynników w bazie danych z wyraźnym rokiem i źródłem
- Użyj współczynnika roku sprawozdawczego (nie najnowszego)
- Zezwól na ponowne obliczenia historyczne, jeśli czynniki zostaną poprawione
- Zachowaj ścieżkę audytu: który czynnik był aktywny w momencie obliczeń
- Nigdy nie aktualizuj lokalnych obliczeń GHG: zawsze twórz nową wersję
Studium przypadku: Oprogramowanie dla MŚP zatrudniające 200 pracowników
Zobaczmy konkretny, kompleksowy przykład. TechFlow S.r.l. jest MŚP Włoskie oprogramowanie zatrudniające 200 pracowników, główne biuro w Mediolanie (wynajmowane), drugie biuro w Rzymie brak własnego centrum danych (wszystkie na AWS eu-west-1 i Azure westeurope).
Profil firmy TechFlow S.r.l.
| Parametr | Wartość |
|---|---|
| Pracownicy | 200 FTE (70% Mediolan, 20% Rzym, 10% całkowicie zdalny) |
| Obroty | 15 milionów euro |
| Biura | 2 lokalizacje do wynajęcia (łącznie 2400 m²) |
| Infrastruktura informatyczna | 100% chmura (AWS + Azure), zero serwerów fizycznych |
| Flota | 3 samochody służbowe z silnikiem Diesla, 1 samochód dostawczy z silnikiem Diesla |
| Podróże służbowe | ~120 lotów rocznie, 40 noclegów hotelowych, 200 tras pociągów |
# 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/dipendentePunkt odniesienia: intensywność emisji dwutlenku węgla dla producentów oprogramowania (2024)
| Rozmiar | tCO₂e/zależne | % Zakres 3 | Wpis główny S3 |
|---|---|---|---|
| Startupy (<50 pracowników) | 0,8 - 1,2 | 65-75% | Podróże służbowe |
| MŚP zajmujące się oprogramowaniem (50-500) | 1,0 - 1,8 | 55-70% | Dojazdy + chmura |
| Technologia dla przedsiębiorstw (500+) | 1,5 - 3,0 | 70-85% | Chmura + korzystanie z produktów |
| Big Tech (hiperskalery) | 3,0 - 8,0 | 75-90% | Łańcuch dostaw sprzętu + centra danych |
| TechFlow (studium przypadku) | 1.18 | 58% | Dojazdy |
Możliwości redukcji w TechFlow
Z analizy Inwentaryzacji GHG 2024 wynika, że najskuteczniejsze dźwignie redukcyjne to:
- Inteligentna praca rozszerzona do 50%: Kot. 7 redukcja o ~35% (-24 tCO2e). Dojazdy do pracy to największy pojedynczy element.
- 100% GO dla obu lokalizacji: Rynkowy zakres 2 spada do ~0 tCO2e (-55 tCO2e). Szacunkowy koszt GO: 3 000-5 000 EUR/rok.
- Polityka budżetu dotycząca emisji dwutlenku węgla w podróżach: Zastąpienie krótkich lotów (<3h) pociągiem. Kot. 6 redukcja o ~40% (-6,5 tCO2e).
- Narzędzie śladu węglowego AWS: Przełącz się z szacunków opartych na wydatkach na danego dostawcę zmniejsza kat. 1 chmura o ~40% (-12 tCO2e) i poprawia jakość danych.
Plan dekarbonizacji TechFlow — docelowa temperatura SBTi 1,5°C
| Rok | Wartość docelowa (tCO₂e) | Redukcja vs 2024 | Główne działania |
|---|---|---|---|
| 2024 (wartość bazowa) | 235,8 | - | Pierwszy raport CSRD |
| 2025 | 211,0 | -10,5% | GO 100%, inteligentna praca 50% |
| 2027 | 165,0 | -30% | Flota elektryczna, polityka podróżnicza |
| 2030 | 118,0 | -50% (SBTi krótkoterminowo) | Ekologiczni, zdalni dostawcy usług w chmurze |
| 2050 | <24 | -90% (SBTi zero netto) | Zneutralizowana pozostałość, bez kompensacji |
Wnioski i dalsze kroki
Modelowanie danych na potrzeby raportowania ESG nie jest prostym ćwiczeniem dotyczącym zgodności: stanowi podstawę do zbudowania wiarygodnej i weryfikowalnej strategii dekarbonizacji. Widzieliśmy, jak skonstruować system obejmujący cały łańcuch wartości firmy oprogramowania, od kotła grzewczego (Zakres 1) do lotu samolotem komercyjnym (Zakres 3 Kat. 6) aż do energii zużytej przez użytkowników do uruchomienia produktu (Zakres 3 Kat. 11).
Najważniejsze punkty, które należy zabrać ze sobą:
- Dominuje zakres 3: W przypadku producentów oprogramowania 55–75% emisji mieści się w zakresie 3. Nieprawidłowe modelowanie oznacza zniekształcony obraz śladu węglowego.
- Podwójny raport Zakres 2: ESRS E1 wymaga obu metod (opartej na lokalizacji i rynkowe). Rynek zachęca do zakupu certyfikowanej energii odnawialnej.
- Jakość danych to wszystko: Hierarchia oparta na działaniach > oparta na wydatkach musi kierować wyborami dotyczącymi gromadzenia danych. Zacznij od szacunków opartych na wydatkach i poprawiaj je z roku na rok.
- Niezmienność obliczeń: Nigdy nie aktualizuj kalkulacji GHG na miejscu. Wersjonuj i utrzymuj pełną ścieżkę audytu do przeglądu audytora ESG.
- SBTi jako gwiazda północna: Cel oparty na nauce (-50% do 2030 r., -90% do 2050 r.) musi wyznaczać priorytety redukcji, a nie minimalną zgodność z dyrektywą CSRD.
Następne artykuły z serii
W kolejnym artykule z serii pt. „ESG i CSRD: Europejskie obowiązki firm technologicznych”, przyjrzyjmy się pełnym ramom regulacyjnym: kto podlega CSRD, terminy w przypadku MŚP koncepcja podwójnej oceny istotności oraz takie jak system modelowania danych skonstruowane w tym artykule przekładają się na ujawnienia ESRS E1 wymagane od Urzędu.
Zobacz także powiązane artykuły:
- Seria MLOps: Jak zintegrować śledzenie emisji dwutlenku węgla z rurociągami ML (metryki zrównoważonego rozwoju w rejestrze modeli MLflow)
- Seria inżynierii AI: Ślad węglowy systemów RAG i LLM w produkcji
- Artykuł 9 (ta seria): Potok Zakres 3 — Automatyzacja zbieranie danych z łańcucha wartości wraz z łącznikami API dla dostawców