Softwarová architektura uhlíkového účetnictví: Platformy ESG
V roce 2025 již není podávání zpráv o uhlíku pro velké společnosti dobrovolnou volbou Evropské země: je to zákonná povinnost. Tam Směrnice Corporate Sustainability Reporting Directive (CSRD), vstup v platnost s legislativním nařízením 125/2024 v Itálii radikálně změnilo prostředí podávání zpráv ESG tím, že disciplína z oblasti podnikové komunikace do oblasti dodržování předpisů, se stejným provozní a technické důsledky systému finančního účetnictví.
Globální trh se softwarem pro uhlíkové účetnictví dohnal 1,8 miliardy dolarů v roce 2025 a očekává se, že do roku 2030 překročí 6 miliard s CAGR 27,4 %. Není to jen o nákupu platforma SaaS: navrhování robustního systému uhlíkového účtování vyžaduje dovednosti v rozsahu od od atmosférické chemie po softwarové inženýrství, od statistiky po komunitární právo. Inženýrské týmy kteří se ocitnou v budování těchto platforem, musí pochopit Protokol GHG, The Evropské standardy pro podávání zpráv o udržitelnosti (ESRS), databáze emisních faktorů a architektura distribuovaných systémů schopná shromažďovat, vypočítávat a certifikovat data ze stovek z různých zdrojů.
Tento článek je úplným technickým průvodcem: od modelování domény pomocí entit protokolu GHG, až po implementaci výpočetního enginu v Pythonu s FastAPI až po automatizaci sestav CSRD/CDP/GRI a integrace s Climatiq API pro emisní faktory v reálném čase. Pokud budujete platformu Interní ESG, hodnocení podnikových řešení nebo prostě chcete pochopit, jak tyto architektury fungují pod kapotou, jste na správném místě.
Co se dozvíte v tomto článku
- Protokol o skleníkových plynech: Rozsah 1, 2 a 3 a 15 kategorií nepřímých emisí z dodavatelského řetězce
- Jak modelovat datovou doménu systému uhlíkového účetnictví (Organizace, Facility, EmissionSource, Activity)
- Hlavní databáze emisních faktorů: DEFRA, EPA, ecoinvent, Climatiq API
- Architektura mikroslužeb: sběr dat, výpočetní modul, výkaznictví, auditní záznam
- Implementace Pythonu s FastAPI a pandy pro výpočet emisí
- Automatizace rozsahu 3: integrace se SAP ERP, data nákupu, cestovní data
- Automatické generování zpráv o standardech CSRD/ESRS, CDP a GRI
- Audit trail a datová linie: úplná sledovatelnost výpočtů
- Srovnání platformy: Persefoni, Watershed, Sphera, Plan A
- Případová studie: Italská výrobní společnost se zprávou CSRD za rok 2025
- Aktualizovaná legislativa: CSRD Omnibus 2025, EU Taxonomy, Legislativní vyhláška 125/2024
Pozice v řadě EnergyTech
| # | Položka | Stát |
|---|---|---|
| 1 | MQTT a InfluxDB: Časové řady pro energetická data | Publikováno |
| 2 | IEC 61850: Standardní protokol pro inteligentní elektrické sítě | Publikováno |
| 3 | DERMS: Distributed Energy Resource Management | Publikováno |
| 4 | Building Management System: AI Energy Optimization | Publikováno |
| 5 | Prognóza obnovitelné energie: ML pro fotovoltaiku a větrnou energii | Publikováno |
| 6 | EV Load Balancing: Smart Charging a Vehicle-to-Grid | Publikováno |
| 7 | Blockchain P2P Energy Trading: Decentralizované energetické trhy | Publikováno |
| 8 | Nacházíte se zde - Architektura softwaru pro uhlíkové účetnictví: Platformy ESG | Proud |
| 9 | Energy Digital Twin: Simulace a optimalizace | Další |
| 10 | Infrastruktura OCPP a EV: standardy a implementace | Již brzy |
Regulační kontext: CSRD, ESRS a legislativní vyhláška 125/2024
Pochopení legislativy není byrokratickým předpokladem: je základem, na kterém je navržena celá architektura systému. Každý technický požadavek, každé pole databáze, každý koncový bod API pochází z povinnosti konkrétního zveřejnění.
CSRD a aplikační vlny
La Směrnice Corporate Sustainability Reporting definoval tři vlny aplikace pro evropské společnosti. První vlna (FY 2024, zprávy zveřejněné v roce 2025) zahrnovala společnosti již podléhají předchozímu NFRD: kotované společnosti, banky a pojišťovny s více než 500 zaměstnanci. Druhá a třetí vlna, původně plánované na roky 2025 a 2026, byly odloženy o dva roky od Směrnice o zastavení hodin zveřejněno v Úředním věstníku EU dne 16. dubna 2025.
Nejvýznamnější změna přišla v prosinci 2025 schválením balíčku Omnibus I: práh povinné aplikace byl zvýšen na 1000 zaměstnanců a obrat 450 milionů eur, což snižuje počet asi o 80 %. společností podléhajících povinnému CSRD. Normy ESRS jsou revidovány s redukcí 61 % povinných datových bodů (od přibližně 1 100 do přibližně 430), přičemž přijetí se očekává v první polovině z roku 2026 a aplikace od roku 2027.
Aktualizovaná časová osa CSRD (po Omnibusu 2025)
| Vlna | Předměty | První zpráva | Stát |
|---|---|---|---|
| Vlna 1 | Společnosti již NFRD (>500 emp., kotované/banky/pojišťovny) | Zpráva 2025 (FY 2024) | Probíhá |
| Vlna 2 | Velké společnosti > 1 000 zaměstnanců a obrat > 450 milionů EUR | Zpráva 2027 (FY 2026) – odloženo | Odloženo |
| Vlna 3 | MSP kotované na regulovaných trzích EU | Zpráva 2028 (FY 2027) – odloženo | Odloženo |
| ESRS rev. | Všechny předměty CSRD, zjednodušené standardy | Od fiskálního roku 2027 | V konzultaci |
Legislativní dekret 125/2024: Italská transpozice
Itálie implementovala CSRD s Legislativní vyhláška 125 ze dne 6. září 2024, nabyla účinnosti dnem 25. září 2024. Vyhláška zrušila předchozí legislativní vyhlášku 254/2016, která transponoval NFRD. Mezi hlavní novinky pro italské firmy patří: povinnost omezené ujištění (omezené ujištění) o zprávě o udržitelnosti od a kvalifikovaný auditor, integrace zprávy o udržitelnosti do zprávy vedení, a zveřejnění ve vyhrazené části obchodního rejstříku.
Taxonomie EU: Spojení
Uhlíkové účetnictví se přímo prolíná s Nařízení EU o taxonomii (EU Nařízení 852/2020), který klasifikuje ekonomické aktivity jako „udržitelné“ na základě šesti environmentálních cílů. Společnosti CSRD musí vykazovat KPI zarovnání taxonomie (akcie z obratu, CapEx, OpEx „zarovnané“ a „způsobilé“). Se zjednodušením roku 2025 společnosti s méně než 1 000 zaměstnanci jsou osvobozeni od hlášení taxonomie a zbytek může být omezen na činnosti, které zastupují alespoň 10 % obratu, CapEx nebo OpEx.
Protokol o skleníkových plynech: Referenční rámec
Il Protokol o skleníkových plynech a účetní standard pro emise skleníkových plynů nejrozšířenější na světě, vyvinuté World Resources Institute (WRI) a World Business Council pro udržitelný rozvoj (WBCSD). Prakticky všechny reportovací rámce (CSRD/ESRS, CDP, GRI, ISO 14064) odkazují nebo jsou založeny na protokolu GHG.
Tři obory: Přesné definice
Rozdělení na tři „rozsahy“ umožňuje jasné přiřazení emisí, zamezení duplicitnímu počítání mezi různými účastníky dodavatelského řetězce.
Rozsah 1: Přímé emise
Emise ze zdrojů vlastněných nebo kontrolovaných organizací. Zahrnují: spalování fosilních paliv v kotlích, pecích, firemních vozidlech; emise z procesů (např. CO2 z chemických reakcí, CH4 z hospodářských zvířat); fugitivní emise (úniky chladiva, úniky plynu ze systémů). Relevantních plynů GHG je sedm Kjótský protokol: CO2, CH4, N2O, HFC, PFC, SF6, NF3, vše převedeno na ekvivalent CO2 (CO2e) pomocí IPCC Global Warming Potentials (GWPs).
Rozsah 2: Nepřímé emise z nakupované energie
Emise spojené s výrobou elektřiny, tepla, páry nebo chlazení zakoupené a spotřebované organizací. Vyžaduje GHG Protocol Scope 2 Guidance (2015). hlášení s dvě odlišné metody: metoda na základě polohy (použijte emisní faktor elektrické sítě v místě spotřeba, např. průměrný faktor italské sítě) a metoda tržní (využívá faktory tržních nástrojů, jako jsou kontrakty na rozdíl, certifikáty obnovitelné energie/GO, smlouvy o nákupu energie). Obojí je třeba nahlásit.
Rozsah 3: Jiné nepřímé emise
Emise hodnotového řetězce, rozdělené do 15 kategorií před a po proudu. Obvykle jsou nejvýznamnější (v průměru 70–80 % celkové stopy) a nejvíce těžko měřitelné. ESRS vyžaduje hlášení kategorií Rozsahu 3 „materiály“ identifikované prostřednictvím dvojitého posouzení významnosti.
15 kategorií Scope 3
| # | Kategorie | Typ | Typické pro |
|---|---|---|---|
| 1 | Nakoupené zboží a služby | Proti proudu | Všechny sektory |
| 2 | Kapitálové statky | Proti proudu | Výroba, konstrukce |
| 3 | Činnosti související s palivy a energií | Proti proudu | Všechny sektory |
| 4 | Upstream doprava a distribuce | Proti proudu | Maloobchod, výroba |
| 5 | Odpad vznikající při provozu | Proti proudu | Výroba, potraviny |
| 6 | Služební cesty | Proti proudu | Služby, tech |
| 7 | Dojíždění zaměstnanců | Proti proudu | Všechny sektory |
| 8 | Upstream pronajatá aktiva | Proti proudu | Nemovitosti, maloobchod |
| 9 | Doprava po proudu | Po proudu | Výroba, FMCG |
| 10 | Zpracování prodávaných výrobků | Po proudu | Suroviny, chemické |
| 11 | Použití prodávaných produktů | Po proudu | Automobilový průmysl, elektronika |
| 12 | Léčba na konci životnosti | Po proudu | Balení, trvanlivé zboží |
| 13 | Následný pronajatý majetek | Po proudu | Nemovitost |
| 14 | Franšízy | Po proudu | Potraviny a nápoje, maloobchod |
| 15 | Investice | Po proudu | Banky, investiční fondy |
Datový model: Modelování domény uhlíkového účetnictví
Srdcem každé platformy pro uhlíkové účetnictví je robustní datový model, který věrně odráží koncepty Protokolu o skleníkových plynech. Podívejme se na hlavní entity s jejich atributy a vztahy.
Hlavní entity
Kompletní systém zahrnuje alespoň šest hlavních entit: organizace, zařízení, zdroj emisí, EmissionFactor, aktivita a výpočet. Zde je model v Pythonu s SQLAlchemy:
# models/core.py - Data model GHG Protocol completo
from sqlalchemy import Column, String, Float, Enum, DateTime, ForeignKey, JSON, Integer
from sqlalchemy.orm import relationship, DeclarativeBase
from datetime import datetime
from enum import Enum as PyEnum
import uuid
class Base(DeclarativeBase):
pass
class ScopeType(PyEnum):
SCOPE_1 = "scope_1"
SCOPE_2_LOCATION = "scope_2_location"
SCOPE_2_MARKET = "scope_2_market"
SCOPE_3 = "scope_3"
class Scope3Category(PyEnum):
CAT_1_PURCHASED_GOODS = "cat_1"
CAT_2_CAPITAL_GOODS = "cat_2"
CAT_3_FUEL_ENERGY = "cat_3"
CAT_4_UPSTREAM_TRANSPORT = "cat_4"
CAT_5_WASTE = "cat_5"
CAT_6_BUSINESS_TRAVEL = "cat_6"
CAT_7_EMPLOYEE_COMMUTING = "cat_7"
CAT_8_UPSTREAM_LEASED = "cat_8"
CAT_9_DOWNSTREAM_TRANSPORT = "cat_9"
CAT_10_PROCESSING = "cat_10"
CAT_11_USE_OF_PRODUCTS = "cat_11"
CAT_12_END_OF_LIFE = "cat_12"
CAT_13_DOWNSTREAM_LEASED = "cat_13"
CAT_14_FRANCHISES = "cat_14"
CAT_15_INVESTMENTS = "cat_15"
class Organization(Base):
__tablename__ = "organizations"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
name = Column(String(255), nullable=False)
legal_entity_id = Column(String(100)) # LEI o P.IVA
country_code = Column(String(3), nullable=False) # ISO 3166-1 alpha-3
nace_code = Column(String(10)) # Codice settore NACE Rev.2
reporting_year = Column(Integer, nullable=False)
consolidation_approach = Column(String(50)) # equity_share, financial_control, operational_control
base_year = Column(Integer) # anno di riferimento per i target
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
facilities = relationship("Facility", back_populates="organization")
calculations = relationship("Calculation", back_populates="organization")
class Facility(Base):
__tablename__ = "facilities"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
organization_id = Column(String, ForeignKey("organizations.id"), nullable=False)
name = Column(String(255), nullable=False)
facility_type = Column(String(50)) # manufacturing_plant, office, warehouse, data_center
address = Column(String(500))
country_code = Column(String(3), nullable=False)
latitude = Column(Float)
longitude = Column(Float)
grid_region = Column(String(50)) # es. "IT_NORD", "DE_TENNET" per Scope 2
floor_area_sqm = Column(Float)
is_owned = Column(String(10)) # owned, leased, operated
operational_start = Column(DateTime)
organization = relationship("Organization", back_populates="facilities")
emission_sources = relationship("EmissionSource", back_populates="facility")
class EmissionSource(Base):
"""Fonte di emissione specifica: caldaia, veicolo, processo, acquisto energia"""
__tablename__ = "emission_sources"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
facility_id = Column(String, ForeignKey("facilities.id"), nullable=False)
name = Column(String(255), nullable=False)
source_type = Column(String(100)) # natural_gas_boiler, diesel_generator, company_car, electricity
scope = Column(Enum(ScopeType), nullable=False)
scope_3_category = Column(Enum(Scope3Category)) # solo per Scope 3
fuel_type = Column(String(50)) # natural_gas, diesel, petrol, LPG
unit_of_measure = Column(String(20)) # kWh, liters, kg, km, tonne
description = Column(String(1000))
is_active = Column(String(10), default="true")
facility = relationship("Facility", back_populates="emission_sources")
activities = relationship("Activity", back_populates="emission_source")
class EmissionFactor(Base):
"""Fattore di emissione da database certificato"""
__tablename__ = "emission_factors"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
source_database = Column(String(50), nullable=False) # DEFRA, EPA, ecoinvent, Climatiq, IPCC
activity_id = Column(String(200)) # ID specifico del database sorgente
name = Column(String(500), nullable=False)
category = Column(String(100))
region = Column(String(50)) # country/region code
year = Column(Integer, nullable=False)
unit_type = Column(String(50)) # kgCO2e/kWh, kgCO2e/liter, kgCO2e/km, kgCO2e/tonne
co2e_factor = Column(Float, nullable=False) # valore principale in kgCO2e
co2_factor = Column(Float) # CO2 separato
ch4_factor = Column(Float) # CH4 separato
n2o_factor = Column(Float) # N2O separato
gwp_version = Column(String(20), default="AR6") # AR5, AR6
lca_activity = Column(String(50)) # upstream, combustion, downstream
source_url = Column(String(500))
last_updated = Column(DateTime)
class Activity(Base):
"""Registrazione di attivita con consumo misurato"""
__tablename__ = "activities"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
emission_source_id = Column(String, ForeignKey("emission_sources.id"), nullable=False)
emission_factor_id = Column(String, ForeignKey("emission_factors.id"))
period_start = Column(DateTime, nullable=False)
period_end = Column(DateTime, nullable=False)
quantity = Column(Float, nullable=False)
unit = Column(String(20), nullable=False)
data_quality = Column(String(20), default="measured") # measured, estimated, calculated, supplier
data_source = Column(String(200)) # nome sistema sorgente: SAP, utility_bill, travel_tool
raw_data = Column(JSON) # dati originali non elaborati
notes = Column(String(1000))
created_by = Column(String(100))
created_at = Column(DateTime, default=datetime.utcnow)
emission_source = relationship("EmissionSource", back_populates="activities")
emission_factor = relationship("EmissionFactor")
calculation = relationship("Calculation", back_populates="activity", uselist=False)
class Calculation(Base):
"""Risultato del calcolo emissioni, immutabile una volta creato"""
__tablename__ = "calculations"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
organization_id = Column(String, ForeignKey("organizations.id"), nullable=False)
activity_id = Column(String, ForeignKey("activities.id"), nullable=False)
scope = Column(Enum(ScopeType), nullable=False)
scope_3_category = Column(Enum(Scope3Category))
co2e_tonnes = Column(Float, nullable=False) # risultato in tonnellate CO2e
co2_tonnes = Column(Float)
ch4_tonnes = Column(Float)
n2o_tonnes = Column(Float)
calculation_method = Column(String(50)) # spend_based, activity_based, hybrid, supplier_specific
emission_factor_value = Column(Float) # snapshot del fattore usato
emission_factor_unit = Column(String(50))
emission_factor_source = Column(String(100))
calculation_formula = Column(String(500)) # formula usata, per audit
calculated_at = Column(DateTime, default=datetime.utcnow)
calculated_by = Column(String(100)) # user o sistema automatico
version = Column(Integer, default=1) # versioning per ricertificazione
is_verified = Column(String(10), default="false")
organization = relationship("Organization", back_populates="calculations")
activity = relationship("Activity", back_populates="calculation")Databáze emisních faktorů: DEFRA, EPA, ecoinvent, Climatiq
Emisní faktory jsou matematickým srdcem uhlíkového účetnictví: převádějí množství aktivity (litry nafty, spotřebovaná kWh, ujeté km, vynaložená eura) v tunách CO2e. Kvalita a aktualizace faktorů přímo určuje kvalitu zprávy.
Hlavní databáze
| Databáze | Manažer | Krytí | Aktualizovat | Přístup |
|---|---|---|---|---|
| DEFRA/BEIS | Vláda Spojeného království (DESNZ) | UK + mezinárodní, všechny rozsahy | Roční (červenec) | Zdarma (Excel) |
| Hub EPA GHG | US EPA | USA, mobilita, energie, rozsah 3 | Roční (leden) | Zdarma (Excel) |
| ekoinvent | sdružení ecoinvent | Globální, plná LCA, 18 000+ datových sad | Pololetně | Placená licence |
| IPCC EF databáze | IPCC | Globální, národní inventury | S každou hodnotící zprávou | Uvolnit |
| Climatiq API | Climatiq | Více zdrojů, více než 50 000 faktorů | Nepřetržitý (v reálném čase) | API (freemium) |
| Reziduální mix AIB | Asociace emisních orgánů | Evropa, rozsah 2 tržní | Výroční | Uvolnit |
| IEA Elektřina | Mezinárodní energetická agentura | Globální faktory elektrické sítě | Výroční | Částečně zdarma |
Integrace Climatiq API
Climatiq poskytuje REST API s více než 50 000 ověřenými emisními faktory, pokrytím ISO 14067 a protokol GHG a je jedním z nejpoužívanějších řešení pro integraci výpočtů emisí programově. API podporuje rozsah 1, 2 a 3 s metodami založenými na aktivitě a na základě výdajů.
# services/climatiq_client.py - Integrazione Climatiq API
import httpx
import os
from dataclasses import dataclass
from typing import Optional
from functools import lru_cache
CLIMATIQ_BASE_URL = "https://api.climatiq.io"
@dataclass
class EmissionEstimate:
co2e: float
co2e_unit: str
co2e_calculation_method: str
co2e_calculation_origin: str
emission_factor_name: str
emission_factor_id: str
source: str
year: int
region: str
@dataclass
class ActivityData:
activity_id: str
data: dict # { "energy": { "value": 1000, "energy_unit": "kWh" } }
region: Optional[str] = None
year: Optional[int] = None
class ClimatiqClient:
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key or os.environ.get("CLIMATIQ_API_KEY")
if not self.api_key:
raise ValueError("CLIMATIQ_API_KEY not set")
self.client = httpx.AsyncClient(
base_url=CLIMATIQ_BASE_URL,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=30.0
)
async def estimate_emission(self, activity: ActivityData) -> EmissionEstimate:
"""Calcola emissioni per una singola attivita"""
payload = {
"emission_factor": {
"activity_id": activity.activity_id,
},
**activity.data
}
if activity.region:
payload["emission_factor"]["region"] = activity.region
if activity.year:
payload["emission_factor"]["year"] = activity.year
response = await self.client.post("/estimate", json=payload)
response.raise_for_status()
result = response.json()
return EmissionEstimate(
co2e=result["co2e"],
co2e_unit=result["co2e_unit"],
co2e_calculation_method=result["co2e_calculation_method"],
co2e_calculation_origin=result.get("co2e_calculation_origin", ""),
emission_factor_name=result["emission_factor"]["name"],
emission_factor_id=result["emission_factor"]["activity_id"],
source=result["emission_factor"]["source"],
year=result["emission_factor"]["year"],
region=result["emission_factor"].get("region", "")
)
async def estimate_batch(self, activities: list[ActivityData]) -> list[EmissionEstimate]:
"""Calcolo batch fino a 100 attivita"""
payload = {
"requests": [
{
"emission_factor": {"activity_id": a.activity_id},
**a.data
}
for a in activities
]
}
response = await self.client.post("/batch", json=payload)
response.raise_for_status()
results = response.json()["results"]
return [
EmissionEstimate(
co2e=r["co2e"],
co2e_unit=r["co2e_unit"],
co2e_calculation_method=r["co2e_calculation_method"],
co2e_calculation_origin=r.get("co2e_calculation_origin", ""),
emission_factor_name=r["emission_factor"]["name"],
emission_factor_id=r["emission_factor"]["activity_id"],
source=r["emission_factor"]["source"],
year=r["emission_factor"]["year"],
region=r["emission_factor"].get("region", "")
)
for r in results
]
async def search_emission_factors(
self,
query: str,
region: Optional[str] = None,
year: Optional[int] = None,
source: Optional[str] = None
) -> list[dict]:
"""Ricerca fattori di emissione nel database Climatiq"""
params = {"query": query, "page": 1, "page_size": 20}
if region:
params["region"] = region
if year:
params["year"] = year
if source:
params["source"] = source
response = await self.client.get("/search", params=params)
response.raise_for_status()
return response.json()["results"]
async def __aenter__(self):
return self
async def __aexit__(self, *args):
await self.client.aclose()
# Esempio utilizzo - calcolo Scope 2 location-based per impianto italiano
async def calculate_scope2_italy(kwh_consumed: float) -> float:
"""
Calcola emissioni Scope 2 location-based per consumo elettrico in Italia.
Fattore IEA 2024 per l'Italia: ~0.233 kgCO2e/kWh
"""
async with ClimatiqClient() as client:
activity = ActivityData(
activity_id="electricity-supply_grid-source_supplier_mix",
data={
"energy": {
"value": kwh_consumed,
"energy_unit": "kWh"
}
},
region="IT", # Italia
year=2024
)
estimate = await client.estimate_emission(activity)
# Converti kgCO2e in tonnellate CO2e
return estimate.co2e / 1000 if estimate.co2e_unit == "kg" else estimate.co2eArchitektura mikroslužeb pro uhlíkové účetnictví
Podniková platforma pro uhlíkové účetnictví musí řídit heterogenní toky dat (účty za energie, data ERP, výkazy výdajů, data dodavatelů), komplexní a kontrolovatelné výpočty, a generování zpráv v různých formátech. Architektura a výběr mikroslužeb přirozené pro tyto požadavky.
Čtyři základní služby
1. Služba sběru dat
Zodpovědnost za získávání dat z různých zdrojů: ERP API (SAP, Oracle), účty za energie přes OCR/analyzátor, nástroje pro správu cestování (Concur, TravelPerk), údaje o nákupu, Ruční nahrání CSV, rozhraní API dodavatele. Vystavuje koncové body pro zpracování a spravuje normalizace měrných jednotek.
2. Výpočetní stroj
Výpočetní srdce: přijímá normalizované aktivity, vybírá emisní faktory vhodné (místní nebo prostřednictvím Climatiq API), aplikuje vzorce protokolu GHG, vyrábí neměnné výsledky s explicitním vzorcem výpočtu pro audit. Podpora historické přepočty při aktualizaci faktorů.
3. Reporting Service
Vytvářejte zprávy ve formátech požadovaných frameworky: CSRD/ESRS (s XBRL tagováním), Dotazník CDP (formát JSON/XML), zveřejnění standardů GRI, interní zprávy pro palubní desky. Spravuje správu verzí a digitální podpis pro zajištění.
4. Služba Audit Trail
Uchovává neměnný protokol o každé operaci: kdo vložil data, jaký faktor Použilo se to, když se to spočítalo, kdo to schválil. Podporuje datovou linii kompletní: od zdrojových údajů po číslo v závěrečné zprávě. Nezbytné pro jistotu ze strany auditorů.
Architektonický diagram
┌─────────────────────────────────────────────────────────────────┐
│ CARBON ACCOUNTING PLATFORM │
├─────────────────────────────────────────────────────────────────┤
│ FRONTEND │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐ │
│ │ Dashboard │ │ Data Input │ │ Report Generator │ │
│ │ (Angular) │ │ Wizard │ │ (CSRD/CDP/GRI) │ │
│ └──────┬──────┘ └──────┬──────┘ └─────────────┬───────────┘ │
├─────────┼────────────────┼─────────────────────────┼────────────┤
│ API GATEWAY (FastAPI/Kong) │
│ │ │ │ │
├─────────┼────────────────┼─────────────────────────┼────────────┤
│ MICROSERVICES │
│ ┌──────▼──────┐ ┌──────▼──────┐ ┌──────────────▼─────────┐ │
│ │ Data │ │ Calculation │ │ Reporting │ │
│ │ Collection │→ │ Engine │→ │ Service │ │
│ │ Service │ │ (Python) │ │ (PDF/XBRL/JSON) │ │
│ └──────┬──────┘ └──────┬──────┘ └────────────────────────┘ │
│ │ │ │
│ ┌──────▼──────┐ ┌──────▼──────┐ ┌────────────────────────┐ │
│ │ SAP │ │ Climatiq │ │ Audit Trail │ │
│ │ Connector │ │ API Client │ │ Service │ │
│ └─────────────┘ └─────────────┘ └────────────────────────┘ │
├─────────────────────────────────────────────────────────────────┤
│ DATA LAYER │
│ ┌──────────────┐ ┌─────────────┐ ┌────────────────────────┐ │
│ │ PostgreSQL │ │ TimeSeries │ │ Object Storage │ │
│ │ (Core data) │ │ (InfluxDB) │ │ (S3 - documents) │ │
│ └──────────────┘ └─────────────┘ └────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
INTEGRAZIONI ESTERNE:
- SAP S/4HANA (procurement, energy data)
- Oracle NetSuite (financials per spend-based)
- TravelPerk/Concur (travel data - Cat.6)
- Climatiq API (emission factors)
- AIB Registry (GO certificates - Scope 2 market-based)
- Utility providers API (bollette energia)
Calculation Engine: Implementace Pythonu s FastAPI
Výpočtový stroj a nejkritičtější součást. Musí být přesné, ověřitelné, verzovaný a schopný paralelně zpracovávat tisíce výpočtů. Zde je implementace kompletní s FastAPI.
# calculation_engine/main.py - FastAPI Calculation Engine
from fastapi import FastAPI, HTTPException, BackgroundTasks, Depends
from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetime
from enum import Enum
import pandas as pd
import uuid
import asyncio
from decimal import Decimal, ROUND_HALF_UP
app = FastAPI(
title="Carbon Calculation Engine",
description="GHG Protocol-compliant emission calculation service",
version="2.1.0"
)
# ── Pydantic Models ──────────────────────────────────────────────
class ScopeEnum(str, Enum):
scope_1 = "scope_1"
scope_2_location = "scope_2_location"
scope_2_market = "scope_2_market"
scope_3 = "scope_3"
class CalculationMethod(str, Enum):
activity_based = "activity_based" # quantità * fattore emissione
spend_based = "spend_based" # spesa in EUR * fattore intensità
average_data = "average_data" # media su periodi
supplier_specific = "supplier_specific" # dati diretti da fornitori
class ActivityInput(BaseModel):
activity_id: str = Field(..., description="ID univoco attivita sorgente")
emission_source_id: str
scope: ScopeEnum
scope_3_category: Optional[str] = None
period_start: datetime
period_end: datetime
quantity: float = Field(..., gt=0, description="Quantità attivita")
unit: str = Field(..., description="Unita di misura (kWh, liter, km, tonne, EUR)")
emission_factor_id: Optional[str] = None # se None, il motore cerca automaticamente
calculation_method: CalculationMethod = CalculationMethod.activity_based
region: Optional[str] = None
data_quality: str = "measured"
@validator("quantity")
def quantity_must_be_positive(cls, v):
if v <= 0:
raise ValueError("La quantità deve essere positiva")
return v
class CalculationResult(BaseModel):
calculation_id: str
activity_id: str
scope: ScopeEnum
co2e_tonnes: float
co2_tonnes: Optional[float]
ch4_tonnes: Optional[float]
n2o_tonnes: Optional[float]
emission_factor_value: float
emission_factor_unit: str
emission_factor_source: str
emission_factor_year: int
calculation_formula: str
calculation_method: str
uncertainty_percentage: Optional[float]
calculated_at: datetime
audit_reference: str
class BatchCalculationRequest(BaseModel):
organization_id: str
reporting_year: int
activities: List[ActivityInput] = Field(..., max_items=500)
class ScopeAggregation(BaseModel):
scope_1_tonnes: float
scope_2_location_tonnes: float
scope_2_market_tonnes: float
scope_3_tonnes: float
scope_3_by_category: dict
total_location_based: float
total_market_based: float
reporting_year: int
organization_id: str
# ── Calculation Logic ────────────────────────────────────────────
class EmissionCalculator:
"""
Implementazione GHG Protocol Corporate Standard.
Formula base: Emissioni (kgCO2e) = Attivita x Fattore di Emissione
"""
# Fattori GWP AR6 (IPCC Sixth Assessment Report, 2021)
GWP_AR6 = {
"CO2": 1.0,
"CH4": 27.9, # CH4 fossil
"CH4_bio": 27.9,
"N2O": 273.0,
"SF6": 25200.0,
"NF3": 17400.0,
"HFC134a": 1526.0,
"HFC32": 771.0,
}
# Fattori di incertezza per qualità dato (DEFRA methodology)
UNCERTAINTY_BY_DATA_QUALITY = {
"measured": 5.0, # dati misurati diretti
"calculated": 10.0, # calcolati da misure
"estimated": 20.0, # stime con dati proxy
"spend_based": 35.0, # spend-based (meno preciso)
"default": 25.0,
}
def calculate_activity_based(
self,
quantity: float,
unit: str,
emission_factor: dict,
gwp_version: str = "AR6"
) -> dict:
"""
Calcolo activity-based:
CO2e (kg) = Quantità (unita) x EF (kgCO2e/unita)
"""
ef_value = emission_factor["co2e_factor"]
ef_unit = emission_factor["unit_type"]
# Verifica compatibilità unita
if not self._units_compatible(unit, ef_unit):
raise ValueError(
f"Unita incompatibili: attivita in {unit}, "
f"fattore in {ef_unit}"
)
co2e_kg = quantity * ef_value
# Calcola componenti separate se disponibili
co2_kg = quantity * emission_factor.get("co2_factor", 0)
ch4_kg = quantity * emission_factor.get("ch4_factor", 0)
n2o_kg = quantity * emission_factor.get("n2o_factor", 0)
return {
"co2e_tonnes": round(co2e_kg / 1000, 6),
"co2_tonnes": round(co2_kg / 1000, 6) if co2_kg else None,
"ch4_tonnes": round(ch4_kg / 1000, 6) if ch4_kg else None,
"n2o_tonnes": round(n2o_kg / 1000, 6) if n2o_kg else None,
"formula": (
f"{quantity} {unit} x {ef_value} kgCO2e/{unit} "
f"= {co2e_kg:.4f} kgCO2e "
f"= {co2e_kg/1000:.6f} tCO2e"
)
}
def calculate_spend_based(
self,
spend_eur: float,
emission_intensity: float, # kgCO2e/EUR
currency: str = "EUR",
exchange_rate: float = 1.0
) -> dict:
"""
Calcolo spend-based (Scope 3 Cat.1 quando non si hanno dati attivita):
CO2e (kg) = Spesa (EUR) x Intensità (kgCO2e/EUR)
"""
spend_normalized = spend_eur * exchange_rate
co2e_kg = spend_normalized * emission_intensity
return {
"co2e_tonnes": round(co2e_kg / 1000, 6),
"co2_tonnes": None,
"ch4_tonnes": None,
"n2o_tonnes": None,
"formula": (
f"{spend_normalized:.2f} EUR x {emission_intensity} kgCO2e/EUR "
f"= {co2e_kg:.4f} kgCO2e"
)
}
def _units_compatible(self, activity_unit: str, ef_unit: str) -> bool:
"""Verifica compatibilità unita di misura"""
# Normalizza: il fattore e tipicamente espresso come kgCO2e/[unita_attivita]
ef_denominator = ef_unit.split("/")[-1].strip().lower() if "/" in ef_unit else ef_unit
return activity_unit.lower() == ef_denominator
def aggregate_by_scope(
self,
calculations: List[CalculationResult]
) -> ScopeAggregation:
"""Aggrega i calcoli per scope - usa pandas per performance"""
df = pd.DataFrame([c.dict() for c in calculations])
scope_1 = df[df["scope"] == "scope_1"]["co2e_tonnes"].sum()
scope_2_loc = df[df["scope"] == "scope_2_location"]["co2e_tonnes"].sum()
scope_2_mkt = df[df["scope"] == "scope_2_market"]["co2e_tonnes"].sum()
scope_3_df = df[df["scope"] == "scope_3"]
scope_3_total = scope_3_df["co2e_tonnes"].sum()
# Aggregazione Scope 3 per categoria
scope_3_by_cat = {}
if not scope_3_df.empty and "scope_3_category" in scope_3_df.columns:
scope_3_by_cat = (
scope_3_df.groupby("scope_3_category")["co2e_tonnes"]
.sum()
.to_dict()
)
return ScopeAggregation(
scope_1_tonnes=round(scope_1, 3),
scope_2_location_tonnes=round(scope_2_loc, 3),
scope_2_market_tonnes=round(scope_2_mkt, 3),
scope_3_tonnes=round(scope_3_total, 3),
scope_3_by_category=scope_3_by_cat,
total_location_based=round(scope_1 + scope_2_loc + scope_3_total, 3),
total_market_based=round(scope_1 + scope_2_mkt + scope_3_total, 3),
reporting_year=2024,
organization_id=""
)
calculator = EmissionCalculator()
# ── API Endpoints ────────────────────────────────────────────────
@app.post("/v1/calculate/single", response_model=CalculationResult)
async def calculate_single(activity: ActivityInput):
"""Calcola emissioni per una singola attivita"""
# In produzione: cerca emission factor dal DB o Climatiq
# Qui mock per illustrazione
mock_ef = {
"co2e_factor": 0.233, # kgCO2e/kWh - rete italiana 2024
"co2_factor": 0.228,
"ch4_factor": 0.002,
"n2o_factor": 0.001,
"unit_type": "kgCO2e/kWh",
"source": "IEA 2024",
"year": 2024
}
result = calculator.calculate_activity_based(
quantity=activity.quantity,
unit=activity.unit,
emission_factor=mock_ef
)
calculation_id = str(uuid.uuid4())
uncertainty = calculator.UNCERTAINTY_BY_DATA_QUALITY.get(
activity.data_quality, 25.0
)
return CalculationResult(
calculation_id=calculation_id,
activity_id=activity.activity_id,
scope=activity.scope,
co2e_tonnes=result["co2e_tonnes"],
co2_tonnes=result["co2_tonnes"],
ch4_tonnes=result["ch4_tonnes"],
n2o_tonnes=result["n2o_tonnes"],
emission_factor_value=mock_ef["co2e_factor"],
emission_factor_unit=mock_ef["unit_type"],
emission_factor_source=mock_ef["source"],
emission_factor_year=mock_ef["year"],
calculation_formula=result["formula"],
calculation_method=activity.calculation_method.value,
uncertainty_percentage=uncertainty,
calculated_at=datetime.utcnow(),
audit_reference=f"CALC-{calculation_id[:8].upper()}"
)
@app.post("/v1/calculate/batch", response_model=List[CalculationResult])
async def calculate_batch(request: BatchCalculationRequest):
"""Calcola emissioni per un batch di attivita (max 500)"""
tasks = [calculate_single(activity) for activity in request.activities]
results = await asyncio.gather(*tasks, return_exceptions=True)
successful = [r for r in results if isinstance(r, CalculationResult)]
failed = [r for r in results if isinstance(r, Exception)]
if failed:
# Log errori ma non blocca il batch
pass
return successful
@app.get("/v1/organizations/{org_id}/summary", response_model=ScopeAggregation)
async def get_emissions_summary(org_id: str, year: int = 2024):
"""Riepilogo emissioni per scope per un'organizzazione"""
# In produzione: query dal DB
passAutomatizace rozsahu 3: ERP, Procurement a Travel
Rozsah 3 a největší výzva uhlíkového účtování: data distribuovaná mezi desítky systémy, dodavatelé s různou úrovní vyspělosti, různé metodiky výpočtu pro kategorie. Automatizace je jediný způsob, jak zajistit udržitelnost provozu.
Integrace SAP pro rozsah 3 kategorie 1 (zakoupené zboží)
# integrations/sap_connector.py - Estrazione dati procurement da SAP
import httpx
from dataclasses import dataclass
from typing import List, Optional
from datetime import date
import pandas as pd
@dataclass
class ProcurementRecord:
purchase_order_id: str
vendor_id: str
vendor_name: str
vendor_country: str
material_code: str
material_description: str
quantity: float
unit: str
amount_eur: float
nace_code: Optional[str] # classificazione settore fornitore
delivery_date: date
class SAPConnector:
"""
Connettore SAP S/4HANA via OData API.
Estrae dati procurement per Scope 3 Cat.1 e Cat.4.
"""
def __init__(self, base_url: str, client_id: str, client_secret: str):
self.base_url = base_url
self.client_id = client_id
self.client_secret = client_secret
self._token: Optional[str] = None
async def authenticate(self):
"""OAuth 2.0 client credentials per SAP"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/oauth/token",
data={
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret,
}
)
response.raise_for_status()
self._token = response.json()["access_token"]
async def get_purchase_orders(
self,
year: int,
cost_center: Optional[str] = None
) -> List[ProcurementRecord]:
"""
Recupera ordini di acquisto dal modulo MM di SAP.
Endpoint OData: /sap/opu/odata/sap/MM_PUR_PO_MANAGE_SRV/
"""
if not self._token:
await self.authenticate()
params = {
"$filter": f"PostingDate ge datetime'{year}-01-01T00:00:00' and "
f"PostingDate le datetime'{year}-12-31T23:59:59'",
"$select": "PurchaseOrder,Supplier,SupplierName,OrderedQuantity,"
"PurchaseOrderQuantityUnit,NetPriceAmount,Currency",
"$format": "json",
"$top": 5000
}
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.base_url}/sap/opu/odata/sap/MM_PUR_PO_MANAGE_SRV/"
"A_PurchaseOrder",
headers={"Authorization": f"Bearer {self._token}"},
params=params
)
response.raise_for_status()
data = response.json()["d"]["results"]
return [
ProcurementRecord(
purchase_order_id=item["PurchaseOrder"],
vendor_id=item["Supplier"],
vendor_name=item["SupplierName"],
vendor_country="IT", # da arricchire con master data
material_code=item.get("Material", ""),
material_description=item.get("MaterialName", ""),
quantity=float(item["OrderedQuantity"]),
unit=item["PurchaseOrderQuantityUnit"],
amount_eur=float(item["NetPriceAmount"]),
nace_code=None, # da mappare con codifica NACE
delivery_date=date.fromisoformat(item.get("ScheduleLine", f"{year}-12-31")[:10])
)
for item in data
]
def calculate_scope3_cat1_spend_based(
self,
records: List[ProcurementRecord],
emission_intensities: dict # { nace_code: kgCO2e_per_EUR }
) -> pd.DataFrame:
"""
Calcolo Scope 3 Cat.1 con metodo spend-based.
Da usare quando non si hanno dati attivita dai fornitori.
Fattori da EXIOBASE o WIOD (World Input-Output Database).
"""
df = pd.DataFrame([vars(r) for r in records])
# Mappa NACE -> intensità emissioni
default_intensity = emission_intensities.get("DEFAULT", 0.35) # kgCO2e/EUR
df["emission_intensity"] = df["nace_code"].map(emission_intensities).fillna(default_intensity)
df["co2e_kg"] = df["amount_eur"] * df["emission_intensity"]
df["co2e_tonnes"] = df["co2e_kg"] / 1000
# Aggregazione per fornitore
summary = df.groupby(["vendor_id", "vendor_name"]).agg(
total_spend_eur=("amount_eur", "sum"),
total_co2e_tonnes=("co2e_tonnes", "sum"),
transaction_count=("purchase_order_id", "count")
).reset_index()
summary["avg_intensity"] = summary["total_co2e_tonnes"] * 1000 / summary["total_spend_eur"]
return summary.sort_values("total_co2e_tonnes", ascending=False)
# integrations/travel_connector.py - Dati viaggi per Scope 3 Cat.6
@dataclass
class TravelRecord:
employee_id: str
travel_date: date
origin_iata: str # codice aeroporto IATA
destination_iata: str
transport_mode: str # air, rail, car, ferry
distance_km: float
travel_class: str # economy, business, first
booking_amount_eur: float
class TravelDataProcessor:
"""Processa dati di business travel per Scope 3 Cat.6"""
# Fattori emissione voli (kgCO2e/pkm) - DEFRA 2024
DEFRA_FLIGHT_FACTORS = {
("short_haul", "economy"): 0.151,
("short_haul", "business"): 0.227,
("medium_haul", "economy"): 0.131,
("medium_haul", "business"): 0.262,
("long_haul", "economy"): 0.195,
("long_haul", "business"): 0.585,
("long_haul", "first"): 0.780,
}
# Fattori treno (kgCO2e/pkm) - media europea
RAIL_FACTORS = {
"IT": 0.004, # Trenitalia (alta % rinnovabili)
"DE": 0.006,
"FR": 0.002,
"DEFAULT": 0.041,
}
def classify_flight(self, distance_km: float) -> str:
if distance_km < 1500:
return "short_haul"
elif distance_km < 4000:
return "medium_haul"
else:
return "long_haul"
def calculate_flight_emissions(self, record: TravelRecord) -> float:
"""
Calcolo emissioni volo con Radiative Forcing Index (RFI = 1.9)
per effetti non-CO2 ad alta quota (metodo DEFRA con uplift factor)
"""
haul = self.classify_flight(record.distance_km)
travel_class = record.travel_class.lower() if record.travel_class else "economy"
key = (haul, travel_class)
base_factor = self.DEFRA_FLIGHT_FACTORS.get(
key,
self.DEFRA_FLIGHT_FACTORS[(haul, "economy")]
)
# Applica Radiative Forcing (RFI) per effetti non-CO2 in alta quota
rfi_factor = 1.9
co2e_kg = record.distance_km * base_factor * rfi_factor
return co2e_kg / 1000 # in tonnellate
def process_travel_data(self, records: List[TravelRecord]) -> dict:
"""Elabora tutti i dati travel e restituisce riepilogo per Cat.6"""
results = []
for record in records:
if record.transport_mode == "air":
co2e = self.calculate_flight_emissions(record)
method = "DEFRA 2024 with RFI=1.9"
elif record.transport_mode == "rail":
country = record.origin_iata[:2] # approssimazione
factor = self.RAIL_FACTORS.get(country, self.RAIL_FACTORS["DEFAULT"])
co2e = record.distance_km * factor / 1000
method = f"Rail factor {country}"
else:
co2e = record.distance_km * 0.171 / 1000 # car average
method = "DEFRA car average"
results.append({
"employee_id": record.employee_id,
"travel_date": record.travel_date,
"transport_mode": record.transport_mode,
"distance_km": record.distance_km,
"co2e_tonnes": co2e,
"method": method
})
df = pd.DataFrame(results)
return {
"total_co2e_tonnes": df["co2e_tonnes"].sum(),
"by_mode": df.groupby("transport_mode")["co2e_tonnes"].sum().to_dict(),
"by_month": df.groupby(df["travel_date"].apply(lambda x: x.month))["co2e_tonnes"].sum().to_dict(),
"total_km": df["distance_km"].sum(),
"record_count": len(df)
}Automatické hlášení: CSRD/ESRS, CDP a GRI
Generování regulačních zpráv je často časově nejnáročnějším procesem uhlíkového účtování. Automatizace dramaticky zkracuje čas z týdnů na hodiny, zvyšuje konzistenci a zajišťuje, že každý údaj je sledovatelný ke zdroji.
Struktura zprávy CSRD/ESRS E1 (změna klimatu)
Klimatická norma ESRS E1 vyžaduje zveřejnění: správy klimatu, analýza strategie a scénářů, řízení rizik, metriky a cíle. Metriky emisí jsou definovány v ESRS E1-6 a vyžadují údaje pro všechny tři rozsahy.
# reporting/csrd_generator.py - Generazione report CSRD/ESRS E1
from dataclasses import dataclass, asdict
from typing import List, Optional, Dict
from datetime import datetime
import json
@dataclass
class ESRS_E1_6_Disclosure:
"""
ESRS E1-6: Gross Scopes 1, 2 and 3 greenhouse gas emissions
Disclosure requirements per ESRS E1 (climate change)
"""
organization_name: str
legal_entity_identifier: str # LEI
reporting_period: str # es. "FY2024"
reporting_standard: str = "ESRS E1 - Climate Change"
# Scope 1 - Emissioni dirette
scope_1_total_gross_tco2e: float = 0.0
scope_1_breakdown_by_ghg: Dict[str, float] = None # { "CO2": x, "CH4": y ... }
scope_1_breakdown_by_source: Dict[str, float] = None
# Scope 2 - Emissioni indirette energia
scope_2_location_based_tco2e: float = 0.0
scope_2_market_based_tco2e: float = 0.0
scope_2_purchased_electricity_kwh: float = 0.0
scope_2_renewable_electricity_percentage: float = 0.0
# Scope 3 - Emissioni catena del valore
scope_3_total_tco2e: float = 0.0
scope_3_upstream_tco2e: float = 0.0
scope_3_downstream_tco2e: float = 0.0
scope_3_by_category: Dict[str, float] = None
# Intensità emissioni
revenue_intensity_tco2e_per_meur: Optional[float] = None # tCO2e/M EUR
employee_intensity_tco2e_per_fte: Optional[float] = None
# Confronto anno precedente e target
base_year: int = 2020
scope_1_2_reduction_vs_base: Optional[float] = None # percentuale
sbti_target: Optional[str] = None # es. "1.5°C aligned, -42% by 2030"
# Metodologia
ghg_accounting_standard: str = "GHG Protocol Corporate Standard"
emission_factor_sources: List[str] = None
data_quality_notes: str = ""
assurance_level: str = "limited_assurance"
assurance_provider: str = ""
def __post_init__(self):
if self.scope_3_by_category is None:
self.scope_3_by_category = {}
if self.scope_1_breakdown_by_ghg is None:
self.scope_1_breakdown_by_ghg = {}
if self.emission_factor_sources is None:
self.emission_factor_sources = []
@property
def total_ghg_location_based(self) -> float:
return (self.scope_1_total_gross_tco2e +
self.scope_2_location_based_tco2e +
self.scope_3_total_tco2e)
@property
def total_ghg_market_based(self) -> float:
return (self.scope_1_total_gross_tco2e +
self.scope_2_market_based_tco2e +
self.scope_3_total_tco2e)
class CSRDReportGenerator:
def generate_esrs_e1_json(self, disclosure: ESRS_E1_6_Disclosure) -> str:
"""Genera disclosure ESRS E1-6 in formato JSON strutturato"""
report = {
"metadata": {
"standard": disclosure.reporting_standard,
"generated_at": datetime.utcnow().isoformat(),
"reporting_period": disclosure.reporting_period,
"organization": disclosure.organization_name,
"lei": disclosure.legal_entity_identifier,
},
"ESRS_E1-6": {
"gross_scope_1_tco2e": disclosure.scope_1_total_gross_tco2e,
"scope_1_breakdown_by_ghg": disclosure.scope_1_breakdown_by_ghg,
"scope_2_location_based_tco2e": disclosure.scope_2_location_based_tco2e,
"scope_2_market_based_tco2e": disclosure.scope_2_market_based_tco2e,
"scope_2_purchased_electricity_kwh": disclosure.scope_2_purchased_electricity_kwh,
"scope_2_renewable_pct": disclosure.scope_2_renewable_electricity_percentage,
"scope_3_total_tco2e": disclosure.scope_3_total_tco2e,
"scope_3_by_category": disclosure.scope_3_by_category,
"total_ghg_location_based_tco2e": disclosure.total_ghg_location_based,
"total_ghg_market_based_tco2e": disclosure.total_ghg_market_based,
"ghg_intensity_revenue": disclosure.revenue_intensity_tco2e_per_meur,
"ghg_intensity_employee": disclosure.employee_intensity_tco2e_per_fte,
},
"methodology": {
"accounting_standard": disclosure.ghg_accounting_standard,
"emission_factor_sources": disclosure.emission_factor_sources,
"base_year": disclosure.base_year,
"consolidation_approach": "operational_control",
"data_quality": disclosure.data_quality_notes,
},
"assurance": {
"level": disclosure.assurance_level,
"provider": disclosure.assurance_provider,
},
"targets": {
"sbti_commitment": disclosure.sbti_target,
"scope_1_2_reduction_vs_base": disclosure.scope_1_2_reduction_vs_base,
}
}
return json.dumps(report, indent=2, ensure_ascii=False)
def generate_cdp_questionnaire_c6(self, disclosure: ESRS_E1_6_Disclosure) -> dict:
"""
Genera risposte per CDP questionnaire - sezione C6 (Emissions Data).
CDP condivide molti requisiti con CSRD, il che riduce il double reporting.
"""
return {
"C6.1": {
"question": "Provide total gross global Scope 1 emissions in metric tons CO2e",
"response": disclosure.scope_1_total_gross_tco2e,
"unit": "metric tons CO2e"
},
"C6.2": {
"question": "Describe Scope 1 emissions by constituent gases",
"response": disclosure.scope_1_breakdown_by_ghg
},
"C6.3": {
"question": "Provide total gross global Scope 2 emissions",
"response": {
"location_based": disclosure.scope_2_location_based_tco2e,
"market_based": disclosure.scope_2_market_based_tco2e,
}
},
"C6.5": {
"question": "Account for Scope 3 emissions",
"response": {
"total": disclosure.scope_3_total_tco2e,
"categories": disclosure.scope_3_by_category
}
},
"C6.10": {
"question": "Describe Scope 1 and 2 GHG emissions by location",
"note": "See facility-level breakdown in supporting data"
}
}Audit Trail a datová linie
Vzhledem k tomu, že CSRD vyžaduje omezené ujištění od kvalifikovaného auditora, auditní stopa tomu tak není volitelný požadavek: a páteř platformy. Každé vydání zprávy musí být dohledatelné zpět k původnímu zdroji s explicitním vzorcem výpočtu.
# audit/audit_trail.py - Sistema di audit immutabile
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Dict, Optional
import hashlib
import json
import uuid
@dataclass(frozen=True) # immutabile
class AuditEvent:
"""
Evento di audit immutabile. Non si modifica mai, solo si aggiungono nuovi eventi.
Il hash SHA256 garantisce l'integrita della catena.
"""
event_id: str
event_type: str # DATA_INGESTED, FACTOR_SELECTED, CALCULATION_PERFORMED, REPORT_GENERATED, APPROVED
entity_type: str # Activity, Calculation, EmissionFactor, Report
entity_id: str
actor_id: str # user_id o service_name per azioni automatiche
actor_type: str # human, system
timestamp: str # ISO 8601 UTC
data_before: Optional[str] # JSON snapshot before change
data_after: Optional[str] # JSON snapshot after change
metadata: str # JSON { formula, ef_source, notes ... }
previous_hash: str # hash dell'evento precedente (blockchain-like)
event_hash: str # SHA256 di tutti i campi
@classmethod
def create(
cls,
event_type: str,
entity_type: str,
entity_id: str,
actor_id: str,
actor_type: str = "human",
data_before: Optional[Dict] = None,
data_after: Optional[Dict] = None,
metadata: Optional[Dict] = None,
previous_hash: str = "GENESIS"
) -> "AuditEvent":
event_id = str(uuid.uuid4())
timestamp = datetime.utcnow().isoformat() + "Z"
data_before_str = json.dumps(data_before, sort_keys=True) if data_before else None
data_after_str = json.dumps(data_after, sort_keys=True) if data_after else None
metadata_str = json.dumps(metadata or {}, sort_keys=True)
# Calcola hash dell'evento per garanzia di integrita
hash_content = "|".join([
event_id, event_type, entity_type, entity_id,
actor_id, timestamp, data_after_str or "", previous_hash
])
event_hash = hashlib.sha256(hash_content.encode()).hexdigest()
return cls(
event_id=event_id,
event_type=event_type,
entity_type=entity_type,
entity_id=entity_id,
actor_id=actor_id,
actor_type=actor_type,
timestamp=timestamp,
data_before=data_before_str,
data_after=data_after_str,
metadata=metadata_str,
previous_hash=previous_hash,
event_hash=event_hash
)
class DataLineageTracker:
"""
Traccia il percorso completo di un dato dalla sorgente al report.
Permette ai revisori di verificare ogni numero del report finale.
"""
def __init__(self, db_session):
self.db = db_session
def trace_calculation(self, calculation_id: str) -> dict:
"""
Ricostruisce il lineage completo di un calcolo:
Report -> Calculation -> Activity -> EmissionFactor -> RawData -> Source System
"""
# In produzione: query al DB per tutti gli eventi correlati
lineage = {
"calculation_id": calculation_id,
"trace": [
{
"step": 1,
"description": "Raw data ingested from SAP S/4HANA",
"system": "SAP_CONNECTOR",
"timestamp": "2025-01-15T08:30:00Z",
"data_summary": "Natural gas consumption: 5,420 m3, Jan 2024",
"audit_event_id": "AE-001"
},
{
"step": 2,
"description": "Unit conversion applied: m3 -> kWh",
"formula": "5,420 m3 x 10.55 kWh/m3 = 57,181 kWh",
"timestamp": "2025-01-15T08:31:00Z",
"audit_event_id": "AE-002"
},
{
"step": 3,
"description": "Emission factor selected from DEFRA 2024",
"emission_factor": "Natural gas - Gross CV: 0.18306 kgCO2e/kWh",
"factor_id": "DEFRA_2024_NG_GROSS",
"timestamp": "2025-01-15T08:31:05Z",
"audit_event_id": "AE-003"
},
{
"step": 4,
"description": "Emission calculation performed",
"formula": "57,181 kWh x 0.18306 kgCO2e/kWh = 10,467 kgCO2e = 10.467 tCO2e",
"result_tco2e": 10.467,
"scope": "Scope 1",
"timestamp": "2025-01-15T08:31:06Z",
"audit_event_id": "AE-004"
},
{
"step": 5,
"description": "Calculation included in FY2024 CSRD Report - ESRS E1-6",
"report_id": "RPT-CSRD-2024-001",
"approved_by": "CFO - Mario Rossi",
"timestamp": "2025-03-10T14:00:00Z",
"audit_event_id": "AE-089"
}
],
"data_quality_flag": "HIGH",
"uncertainty_pct": 5.0,
"verification_status": "verified_by_auditor"
}
return lineagePorovnání podnikové platformy: Build vs. Buy
Před vybudováním vlastní platformy je nezbytné vyhodnotit řešení podnik k dispozici. Trh uhlíkového účetního softwaru je velmi vyspělý a přední platformy pokrývají většinu standardních případů použití.
| Platforma | Pevný bod | Cílový sektor | Orientační cena | CSRD připraven |
|---|---|---|---|---|
| Persefony | Vykazování na úrovni investora, značení XBRL, zaměřené na SEC | Finance, korporace | 50–500 tisíc dolarů ročně | Si |
| Povodí | Rychlost hlášení, pokročilé nástroje Scope 3 | Tech, Enterprise | 100 tisíc až 1 milion dolarů ročně | Si |
| Sphera | Integrovaný LCA, průmyslová shoda, řízení rizik | Výroba, energetika, chemie | NA ZNAMENÍ | Si |
| Plán A | Jednoduché UX, rychlá implementace | MSP, střední podniky | 10-100 tisíc dolarů ročně | Částečný |
| IBM Envizi | 40 000+ emisních faktorů, integrace ERP | Podnik, veřejné služby | NA ZNAMENÍ | Si |
| Vlastní sestavení | Celková flexibilita, nativní integrace interních systémů | Velké společnosti se specifickými požadavky | Vývoj 500 000–5 milionů USD | Závisí |
Kdy stavět na zakázku vs
Kupte si platformu, pokud: máte standardní požadavky, chcete žít za 3–6 měsíců nemáte žádný inženýrský tým, který by se věnoval udržitelnosti, objemu dat a zvládnutelné.
Vytvořit vlastní, pokud: máte velmi specifické výrobní procesy s daty nelze spravovat standardními platformami, chcete se nativně integrovat se staršími systémy, máte požadavky na umístění dat, které nejsou kompatibilní se SaaS, nebo objem dat a v řádu milionů záznamů/rok (náklady na SaaS se stanou neúnosnými).
Případová studie: italské výrobní MSP s CSRD 2025
Italská společnost vyrábějící přesné strojírenství s 1200 zaměstnanci, 350 milionů eur v obratu a továrnách v Turíně, Miláně a Brescii se nachází ve vlně 1 CSRD a musí předložit první zprávu za finanční rok 2024 do června 2025.
Emisní inventura FY2024
# case_study/inventory_2024.py - Esempio inventario GHG realistico
# Dati rappresentativi per PMI manifatturiera italiana
INVENTORY_2024 = {
"organization": "MeccanicaPrecisione SpA",
"reporting_year": 2024,
"boundary": "Operational Control",
"currency": "EUR",
"scope_1": {
"natural_gas_combustion": {
"consumption_m3": 485_000, # caldaie industriali
"consumption_kwh": 5_116_750, # conversione: 1 m3 NG = 10.55 kWh
"emission_factor_kgco2e_kwh": 0.18306, # DEFRA 2024
"co2e_tonnes": 937.0,
"source": "DEFRA 2024 - Natural Gas Gross CV"
},
"diesel_mobile": {
"consumption_liters": 42_000, # carrelli elevatori, mezzi operativi
"emission_factor_kgco2e_liter": 2.56, # DEFRA 2024
"co2e_tonnes": 107.5,
"source": "DEFRA 2024 - Diesel"
},
"fugitive_refrigerants": {
"substance": "R-410A",
"kg_recharged": 45,
"gwp_ar6": 2088,
"co2e_tonnes": 94.0,
"source": "IPCC AR6 GWP100"
},
"total_co2e_tonnes": 1138.5
},
"scope_2": {
"purchased_electricity": {
"consumption_kwh": 8_250_000,
"location_based": {
"emission_factor": 0.233, # IEA Italy 2024 kgCO2e/kWh
"co2e_tonnes": 1922.3,
"source": "IEA 2024 Italy Grid"
},
"market_based": {
"go_certificates_kwh": 4_125_000, # 50% rinnovabili con GO
"residual_mix_factor": 0.395, # AIB Italy residual mix 2024
"co2e_tonnes": 1629.4, # solo su 50% non coperto da GO
"source": "AIB Italy Residual Mix 2024"
}
}
},
"scope_3": {
"cat_1_purchased_goods": {
"total_spend_eur": 145_000_000,
"method": "spend_based + supplier_specific (top 20 suppliers)",
"co2e_tonnes": 28_450.0,
"data_quality": "mix: 35% supplier-specific, 65% spend-based"
},
"cat_4_upstream_transport": {
"tonne_km": 3_200_000,
"emission_factor": 0.089, # kgCO2e/tonne-km road freight DEFRA
"co2e_tonnes": 284.8,
},
"cat_5_waste": {
"waste_tonnes": 1_850,
"co2e_tonnes": 185.0,
"method": "waste-type specific factors"
},
"cat_6_business_travel": {
"total_km_air": 1_250_000,
"total_km_rail": 320_000,
"co2e_tonnes": 312.5,
"source": "DEFRA 2024 with RFI=1.9"
},
"cat_7_employee_commuting": {
"employees": 1_200,
"avg_km_per_day": 22,
"working_days": 220,
"mode_split": {"car_solo": 0.65, "car_sharing": 0.05,
"public_transport": 0.25, "cycling": 0.05},
"co2e_tonnes": 892.5
},
"cat_11_use_of_products": {
"units_sold": 45_000,
"avg_energy_use_kwh_per_unit_per_year": 850,
"product_lifetime_years": 15,
"co2e_tonnes_per_year": 2_250.0,
"note": "Calcolato per anno di uso, non lifetime"
},
"total_co2e_tonnes": 32_374.8
},
"summary": {
"scope_1_tco2e": 1_138.5,
"scope_2_location_tco2e": 1_922.3,
"scope_2_market_tco2e": 1_629.4,
"scope_3_tco2e": 32_374.8,
"total_location_based_tco2e": 35_435.6,
"total_market_based_tco2e": 35_142.7,
"intensity_tco2e_per_meur_revenue": 101.3,
"intensity_tco2e_per_fte": 29.5,
"scope_3_percentage_of_total": 91.4, # tipico nel manifatturiero
}
}Postřehy k případové studii
Nejvýznamnějším výsledkem je, že rozsah 3 představuje 91,4 % emisí součty, pouze kategorie 1 (suroviny a komponenty) v hodnotě 80 % ceny Rozsah 3. To je typické pro výrobní sektor a vysvětluje, proč CSRD vyžaduje hlášení dodavatelského řetězce: bez rozsahu 3 by uhlíkové účtování pokrývalo méně než 10 % skutečného dopadu.
Rozdíl mezi rozsahem 2 založeným na místě (1 922 t CO2e) a tržním (1 629 t CO2e) je způsoben na nákup certifikátů obnovitelné energie (Guarantees of Origin - GO) za 50 % spotřeba elektřiny. Italský zbytkový mix z roku 2024 (0,395 kg CO2e/kWh) je vyšší než faktor průměr sítě (0,233 kgCO2e/kWh): toto je kontraintuitivní, ale metodicky správné, protože zbytková směs vylučuje energie již certifikované GO, a proto „obsahuje“ kvótu vyšší než zdroje s vysokou intenzitou.
Testování: Validace výpočtů
Výpočty emisí musí být testovány se stejnou přísností jako finanční zákoník. Chyba v jednom převodním faktoru může vést k řádovým chybám stovek tun CO2e.
# tests/test_calculation_engine.py
import pytest
from decimal import Decimal
from calculation_engine.main import EmissionCalculator
@pytest.fixture
def calculator():
return EmissionCalculator()
class TestScope1Calculations:
def test_natural_gas_combustion_scope1(self, calculator):
"""
Test calcolo gas naturale con fattore DEFRA 2024.
Valore atteso: 5000 kWh x 0.18306 kgCO2e/kWh = 915.3 kgCO2e = 0.9153 tCO2e
"""
emission_factor = {
"co2e_factor": 0.18306,
"co2_factor": 0.18207,
"ch4_factor": 0.000291,
"n2o_factor": 0.000087,
"unit_type": "kgCO2e/kWh",
"source": "DEFRA 2024",
"year": 2024
}
result = calculator.calculate_activity_based(
quantity=5000,
unit="kWh",
emission_factor=emission_factor
)
assert abs(result["co2e_tonnes"] - 0.9153) < 0.001, (
f"Scope 1 gas naturale: atteso 0.9153, ottenuto {result['co2e_tonnes']}"
)
assert result["co2_tonnes"] is not None
assert "formula" in result
def test_diesel_combustion_scope1(self, calculator):
"""
Test calcolo gasolio.
Fattore DEFRA 2024: 2.56179 kgCO2e/liter
1000 litri -> 2.56179 kgCO2e -> 0.00256 tCO2e (arrotondato a 6 decimali)
"""
emission_factor = {
"co2e_factor": 2.56179,
"co2_factor": 2.51476,
"ch4_factor": 0.00179,
"n2o_factor": 0.00524,
"unit_type": "kgCO2e/liter",
"source": "DEFRA 2024",
"year": 2024
}
result = calculator.calculate_activity_based(
quantity=1000,
unit="liter",
emission_factor=emission_factor
)
expected = 2561.79 / 1000
assert abs(result["co2e_tonnes"] - expected) < 0.001
def test_unit_incompatibility_raises_error(self, calculator):
"""Test che unita incompatibili generino errore"""
emission_factor = {
"co2e_factor": 0.18306,
"unit_type": "kgCO2e/kWh",
}
with pytest.raises(ValueError, match="incompatibili"):
calculator.calculate_activity_based(
quantity=1000,
unit="liter", # incompatibile con kWh
emission_factor=emission_factor
)
class TestScope2Calculations:
def test_scope2_location_based_italy_2024(self, calculator):
"""
Fattore rete elettrica italiana IEA 2024: 0.233 kgCO2e/kWh
100.000 kWh -> 23.300 kgCO2e -> 23.3 tCO2e
"""
emission_factor = {
"co2e_factor": 0.233,
"unit_type": "kgCO2e/kWh",
"source": "IEA 2024 Italy",
"year": 2024
}
result = calculator.calculate_activity_based(
quantity=100_000,
unit="kWh",
emission_factor=emission_factor
)
assert abs(result["co2e_tonnes"] - 23.3) < 0.01
def test_scope2_market_based_with_go_certificates(self, calculator):
"""
Con GO certificates per energia rinnovabile: fattore = 0 per la quota coperta.
Residual mix IT 2024: 0.395 kgCO2e/kWh per la quota non coperta.
50% rinnovabili (GO): 50.000 kWh x 0 = 0
50% residual mix: 50.000 kWh x 0.395 = 19.750 kgCO2e = 19.75 tCO2e
"""
# Calcolo solo sulla quota non coperta da GO
emission_factor_residual = {
"co2e_factor": 0.395,
"unit_type": "kgCO2e/kWh",
"source": "AIB Italy Residual Mix 2024",
"year": 2024
}
non_go_kwh = 50_000 # 50% non coperto da GO
result = calculator.calculate_activity_based(
quantity=non_go_kwh,
unit="kWh",
emission_factor=emission_factor_residual
)
assert abs(result["co2e_tonnes"] - 19.75) < 0.01
class TestScope3Calculations:
def test_business_travel_air_short_haul_economy(self):
"""Test emissioni volo breve raggio con RFI"""
from integrations.travel_connector import TravelDataProcessor, TravelRecord
from datetime import date
processor = TravelDataProcessor()
record = TravelRecord(
employee_id="EMP001",
travel_date=date(2024, 3, 15),
origin_iata="LIN",
destination_iata="FCO",
transport_mode="air",
distance_km=490,
travel_class="economy",
booking_amount_eur=180
)
# 490 km x 0.151 kgCO2e/pkm x 1.9 RFI = 140.531 kgCO2e = 0.14053 tCO2e
co2e = processor.calculate_flight_emissions(record)
expected = 490 * 0.151 * 1.9 / 1000
assert abs(co2e - expected) < 0.001
def test_spend_based_with_nace_intensity(self, calculator):
"""Test calcolo spend-based per Category 1"""
result = calculator.calculate_spend_based(
spend_eur=100_000,
emission_intensity=0.35 # kgCO2e/EUR - manifattura metalli
)
# 100.000 EUR x 0.35 kgCO2e/EUR = 35.000 kgCO2e = 35.0 tCO2e
assert abs(result["co2e_tonnes"] - 35.0) < 0.01Cíle založené na vědě (SBTi) a řídicí panel
Il Iniciativa Science Based Targets (SBTi) má stanovena přesná kritéria pro cíle snížení emisí v souladu s Pařížskou dohodou. Firemní standard SBTi vyžaduje: Snížení rozsahu 1+2 o 42 % do roku 2030 (základní 2020) pro scénář 1,5 °C a pokrytí rozsahu 3, pokud představují více než 40 % celkových emisí (ve výrobě téměř vždy).
Přístrojová deska platformy musí ukazovat nejen aktuální emise, ale cesta k cíli: redukční křivka požadovaná SBTi, skutečné emise každý rok a projekce založené na iniciativách ke snížení plánované (PPA pro obnovitelné zdroje energie, elektrifikace vozového parku, efektivita procesů).
# dashboard/sbti_tracker.py - Tracking verso target SBTi
from dataclasses import dataclass
from typing import List, Dict
import pandas as pd
import numpy as np
@dataclass
class SBTiTarget:
organization_id: str
base_year: int
base_year_scope_1_2_tco2e: float
base_year_scope_3_tco2e: float
target_year: int = 2030
scope_12_reduction_pct: float = 42.0 # % riduzione vs base year
scope_3_reduction_pct: float = 25.0 # % riduzione vs base year
scenario: str = "1.5C"
@property
def scope_12_target_tco2e(self) -> float:
return self.base_year_scope_1_2_tco2e * (1 - self.scope_12_reduction_pct / 100)
@property
def annual_reduction_rate(self) -> float:
"""Tasso di riduzione annuale lineare necessario"""
years = self.target_year - self.base_year
return self.scope_12_reduction_pct / years
def build_sbti_trajectory(target: SBTiTarget, actuals: Dict[int, float]) -> pd.DataFrame:
"""
Costruisce tabella confronto tra traiettoria SBTi e emissioni effettive.
actuals: { anno: tCO2e Scope1+2 effettivo }
"""
years = range(target.base_year, target.target_year + 1)
trajectory = []
for year in years:
years_elapsed = year - target.base_year
total_years = target.target_year - target.base_year
# Riduzione lineare richiesta da SBTi
required_reduction_pct = (years_elapsed / total_years) * target.scope_12_reduction_pct
sbti_budget = target.base_year_scope_1_2_tco2e * (1 - required_reduction_pct / 100)
actual = actuals.get(year)
on_track = actual <= sbti_budget if actual is not None else None
trajectory.append({
"year": year,
"sbti_budget_tco2e": round(sbti_budget, 1),
"actual_tco2e": actual,
"gap_tco2e": round(actual - sbti_budget, 1) if actual else None,
"on_track": on_track
})
df = pd.DataFrame(trajectory)
return df
# Utilizzo per MeccanicaPrecisione SpA
target = SBTiTarget(
organization_id="meccanica-precisione-spa",
base_year=2020,
base_year_scope_1_2_tco2e=3_850.0, # baseline 2020
base_year_scope_3_tco2e=35_000.0,
)
actuals_scope_12 = {
2020: 3850.0,
2021: 3720.0,
2022: 3650.0,
2023: 3320.0,
2024: 3060.8, # scope_1 + scope_2_market dal nostro inventory
}
trajectory = build_sbti_trajectory(target, actuals_scope_12)
print(trajectory.to_string(index=False))
# Output:
# year sbti_budget_tco2e actual_tco2e gap_tco2e on_track
# 2020 3850.0 3850.0 0.0 True
# 2021 3608.6 3720.0 111.4 False
# 2022 3367.2 3650.0 282.8 False
# 2023 3125.8 3320.0 194.2 False
# 2024 2884.4 3060.8 176.4 False
# -> L'azienda e fuori traiettoria: serve accelerare le iniziative di riduzioneOsvědčené postupy a anti-vzorce
Nejlepší postupy
- Neměnnost výpočtů: Po vypočítání a uložení emise nesmí se nikdy upravovat. Pokud se faktory změní, vytvoří se nová verze výpočtu. Auditní záznam musí ukazovat obě verze.
- Nahlásit obě metody rozsahu 2: Lokalizační a tržní jsou oba jsou povinné podle ESRS a protokolu GHG. Nevykazujte pouze na základě trhu protože je nižší.
- Významnost dokumentu Rozsah 3: Nemusíte počítat všech 15 kategorií Rozsah 3. Musíte však pomocí analýzy významnosti prokázat, proč jste zahrnuli/vyloučili každá kategorie. Toto je výslovný požadavek ESRS.
- Verze emisních faktorů: Faktory se mění každý rok. Ponechte si snímek faktoru použitého pro každý výpočet. Neaktualizovat zpětně, aniž by došlo k explicitnímu přepočtu s novou verzí.
- Příznaky kvality dat: Klasifikujte jednotlivá data podle kvality (měřeno, vypočítané, odhadované, založené na výdajích). ESRS vyžaduje deklaraci kvality dat rozsahu 3. Mohou být použita data nižší kvality, ale musí být prohlášena.
- Dokumentace hranic: Explicitně zdokumentujte zahrnuté entity v rámci ohlašovací hranice a vyloučené s odůvodněním (práh <5 %, data nejsou k dispozici atd.).
Anti-vzory, kterým je třeba se vyhnout
- Používejte pouze tržní rozsah 2: Protokol GHG vyžaduje obojí. Reportujte pouze tržní (obvykle nižší díky GO/PPA) bez lokalizovaný a metodicky nesprávný a potenciální greenwashing.
- Emisní faktory nebyly aktualizovány: Použijte faktory staré 5 a více let přináší významné chyby, zejména pro měnící se rozvodnou síť každý rok s růstem obnovitelných zdrojů.
- Duplicitní emise při konsolidaci: Pokud dceřiná společnost vypočítá své emise a mateřská společnost je zahrne do konsolidace, musí být důsledně používána metoda podílu na vlastním kapitálu nebo metoda provozní kontroly.
- Ignorovat nejistotu: Všechny výpočty emisí mají nejistota, zejména rozsah 3. Neuvádějte čísla bez uveďte rozsah spolehlivosti a použitou metodu.
- Hlášení bez ujištění: S CSRD zprávy bez omezeného ujištění nesplňují zákonné požadavky pro Wave 1. Plán pro zajištění od prvního roku, ne jako dodatečně.
Závěry a plán implementace
Návrh platformy pro uhlíkové účetnictví na podnikové úrovni je složitý projekt což vyžaduje multidisciplinární dovednosti. Ale koncepční struktura je jasná: Protokol GHG poskytuje účetní rámec, databáze emisní faktory (DEFRA, EPA, Climatiq) uveďte koeficienty, aarchitektura mikroslužeb zaručuje škálovatelnost a udržovatelnost, aneměnný audit trail zajišťuje důvěryhodnost pro CSRD.
Pro malý a střední podnik, který se potřebuje přiblížit CSRD v letech 2025–2026, je toto praktické doporučení začít s vyspělou platformou SaaS (Persefoni, Plan A nebo Watershed) pro i první dva roky shromažďujte skutečná data, pochopte mezery v datech a teprve potom vyhodnotit, zda vytvořit vlastní řešení nebo zůstat na SaaS. 90 % organizací není třeba stavět od nuly.
Pro inženýrské týmy, které místo toho pracují na platformách ESG jako produkt, koncepty ilustrované v tomto článku – datový model protokolu GHG, auditovatelný výpočetní stroj, integrace s emisními faktory API a generování regulačních hlášení - sono základy, na kterých se dá stavět. Trh uhlíkového účetního softwaru poroste o 27 % ročně do roku 2030: existuje prostor pro specializovaná řešení, zej pro průmyslová odvětví s požadavky na procesní údaje, které platformy nepokrývají generalisté.
Doporučený plán implementace
| Fáze | Trvání | Objektivní | Výstupy |
|---|---|---|---|
| Fáze 1 | 1-2 měsíce | Dvojité posouzení významnosti + hranice | Seznam kategorií materiálů Rozsahu 3, hranice hlášení |
| Fáze 2 | 2-3 měsíce | Rozsah 1 a 2 sběr dat | Automatizované potrubí s účty, SAP, SCADA |
| Fáze 3 | 3-4 měsíce | Rozsah 3 kategorií materiálů (na základě útraty) | Rozsah 3 výpočty pro kat. 1, 4, 6, 7, 11 |
| Fáze 4 | 1-2 měsíce | Zpráva a audit CSRD/ESRS E1 | Zpráva připravena pro omezené ujištění |
| Fáze 5 | Kontinuální | Zlepšení kvality dat a zapojení dodavatelů | Snížení na základě útraty, zvýšení rozsahu 3 specifické pro dodavatele |
Základní zdroje
- Firemní standard protokolu GHG: ghgprotocol.org – standard odkaz, zdarma a ke stažení
- Emisní faktory DEFRA 2024: gov.uk/government/publications/ skleníkové-plyny-vykazování-konverzní-faktory-2024
- Climatiq API: climatiq.io/docs - kompletní dokumentace e rychlý start zdarma
- ESRS E1 standard: efrag.org - Evropské finanční výkaznictví Poradní skupina, konečné standardy ESRS
- Firemní příručka SBTi: sciencebasedtargets.org – průvodce k definici vědecky podložených cílů
- Databáze ecoinvent: ecoinvent.org - databáze LCA pro Rozsah 3 s podrobnými daty procesu
Připravované články ze série EnergyTech
Následující článek prozkoumá Digitální energetická dvojčata: jak vytvořit virtuální repliky průmyslových závodů pro simulaci scénářů snižování emisí před jejich implementací ve fyzickém světě. Stále více centrální technologie v dekarbonizační strategie velkých průmyslových odvětví.
Chcete-li se dozvědět více o technologiích AI aplikovaných na datový obchod, viz také série Datový sklad, AI a digitální transformace, zejména články o Správa dat a kvalita dat pro spolehlivou AI a nahoru MLOps for Business, které se zabývají tématy přímo použitelnými pro správu výpočtových modelů emisí ve výrobě.