탄소 회계 소프트웨어 아키텍처: ESG 플랫폼
2025년에는 탄소 보고가 더 이상 대기업의 자발적 선택이 아닙니다. 유럽 국가: 이는 법적 의무입니다. 거기 기업 지속가능성 보고 지침(CSRD), 입장 이탈리아 입법령 125/2024에 따라 ESG 보고 환경을 근본적으로 변화시켰습니다. 기업 커뮤니케이션 영역에서 규정 준수 영역까지의 규율을 재무회계시스템의 운영적, 기술적 영향.
글로벌 탄소회계 소프트웨어 시장이 따라잡았다 2025년에는 18억 달러 CAGR은 27.4%로 2030년까지 60억 달러를 초과할 것으로 예상됩니다. 단순히 구매만 하는 것이 아니다 SaaS 플랫폼: 강력한 탄소 회계 시스템을 설계하려면 다음과 같은 기술이 필요합니다. 대기화학부터 소프트웨어 공학까지, 통계부터 공동체법까지. 엔지니어링 팀 이러한 플랫폼을 구축하는 사람들은 온실가스 프로토콜, gli 유럽 지속가능성 보고 표준(ESRS), 배출 계수 데이터베이스 및 수백 개의 데이터를 수집, 계산 및 인증할 수 있는 분산 시스템 아키텍처 다른 출처에서.
이 문서는 GHG 프로토콜 엔터티를 사용하여 도메인을 모델링하는 것부터, FastAPI를 사용하여 Python에서 계산 엔진 구현, CSRD/CDP/GRI 보고서 자동화까지 실시간 배출계수를 위한 Climatiq API와의 통합. 플랫폼을 구축한다면 내부 ESG, 엔터프라이즈 솔루션 평가 또는 단순히 이러한 아키텍처의 작동 방식을 이해하고 싶은 경우 내부적으로는 올바른 위치에 있습니다.
이 기사에서 배울 내용
- GHG 프로토콜: 범위 1, 2, 3과 공급망에서 발생하는 간접 배출의 15가지 범주
- 탄소회계 시스템의 데이터 영역을 모델링하는 방법(조직, 시설, 배출원, 활동)
- 배출계수 주요 데이터베이스: DEFRA, EPA, ecoinvent, Climatiq API
- 마이크로서비스 아키텍처: 데이터 수집, 계산 엔진, 보고, 감사 추적
- 배출량 계산을 위해 FastAPI 및 Pandas를 사용한 Python 구현
- 범위 3 자동화: SAP ERP, 조달 데이터, 여행 데이터와의 통합
- CSRD/ESRS, CDP 및 GRI 표준 보고서 자동 생성
- 감사 추적 및 데이터 계보: 계산의 완벽한 추적성
- 플랫폼 비교: Persefoni, Watershed, Sphera, Plan A
- 사례 연구: 2025 CSRD 보고를 갖춘 이탈리아 제조 회사
- 업데이트된 법안: CSRD 옴니버스 2025, EU 분류법, 입법 법령 125/2024
EnergyTech 시리즈에서의 위치
| # | Articolo | 상태 |
|---|---|---|
| 1 | MQTT 및 InfluxDB: 에너지 데이터의 시계열 | 게시됨 |
| 2 | IEC 61850: 지능형 전기 네트워크용 표준 프로토콜 | 게시됨 |
| 3 | DERMS: 분산 에너지 자원 관리 | 게시됨 |
| 4 | 빌딩 관리 시스템: AI 에너지 최적화 | 게시됨 |
| 5 | 재생 가능 에너지 예측: 태양광 및 풍력을 위한 ML | 게시됨 |
| 6 | EV 부하 분산: 스마트 충전 및 차량-그리드(Vehicle-to-Grid) | 게시됨 |
| 7 | 블록체인 P2P 에너지 거래: 분산형 에너지 시장 | 게시됨 |
| 8 | 현재 위치 - 탄소 회계 소프트웨어 아키텍처: ESG 플랫폼 | 현재의 |
| 9 | 에너지 디지털 트윈: 시뮬레이션 및 최적화 | 다음 |
| 10 | OCPP 및 EV 인프라: 표준 및 구현 | 곧 출시 예정 |
규제 상황: CSRD, ESRS 및 법령 125/2024
법률을 이해하는 것은 관료적 전제조건이 아닙니다. 이는 전체 아키텍처가 설계되는 기초입니다. 시스템의. 모든 기술 요구 사항, 모든 데이터베이스 필드, 모든 API 엔드포인트는 의무에서 비롯됩니다. 구체적인 공개.
CSRD 및 애플리케이션 웨이브
La 기업 지속가능성 보고 지침 세 가지 적용 물결을 정의했습니다. 유럽 기업의 경우. 첫 번째 물결(2024년 회계연도, 2025년에 발표된 보고서)에는 기업이 참여했습니다. 이미 이전 NFRD의 적용을 받는 기업: 상장회사, 직원 500명 이상의 은행, 보험회사. 당초 2025년과 2026년으로 예정됐던 2·3차 웨이브가 2년 연기됐다. 에서 시계 중지 지시문 2025년 4월 16일 EU 공식 저널에 게재됨.
가장 중요한 변화는 2025년 12월 패키지 승인으로 이루어졌습니다. 옴니버스 I: 필수 신청 기준이 다음으로 상향 조정되었습니다. 직원 1,000명, 매출 4억 5천만 유로, 숫자가 약 80% 감소합니다. CSRD 의무화 대상 기업. ESRS 표준은 축소되어 개정되고 있습니다. 필수 데이터 포인트 중 61%(약 1,100~430개), 상반기에 채택 예정 2026년부터 적용, 2027년부터 적용.
업데이트된 CSRD 타임라인(Omnibus 2025 이후)
| 파도 | 과목 | 첫 번째 보고서 | 상태 |
|---|---|---|---|
| 웨이브 1 | 이미 NFRD 회사(500개 이상의 직원, 상장/은행/보험) | 2025년 보고서(2024년) | 진행중 |
| 웨이브 2 | 대기업: 직원 1,000명 이상, 매출 4억 5천만 유로 이상 | 보고서 2027(FY 2026) - 연기됨 | 연기됨 |
| 웨이브 3 | 규제된 EU 시장에 상장된 중소기업 | 보고서 2028(FY 2027) - 연기됨 | 연기됨 |
| ESRS 개정. | 모든 CSRD 주제, 단순화된 표준 | 2027년부터 | 상담중 |
입법령 125/2024: 이탈리아 전치
이탈리아는 CSRD를 시행했습니다. 2024년 9월 6일 입법령 125, 2024년 9월 25일에 발효되었습니다. 이 법령은 이전 입법 법령 254/2016을 폐지했습니다. NFRD를 전치했습니다. 이탈리아 기업의 주요 혁신은 다음과 같습니다. 제한된 보증 (제한적 검증) 지속가능성 보고서에 대한 자격을 갖춘 감사인, 지속 가능성 보고서를 경영 보고서에 통합, 회사 등록부의 해당 섹션에 게시됩니다.
EU 분류: 연결
탄소회계는 다음과 직접적으로 교차합니다. EU 분류법 규정 (EU 규정 852/2020), 이는 6가지 환경 목표를 기반으로 경제 활동을 "지속 가능한" 활동으로 분류합니다. CSRD 회사는 분류 체계 정렬 KPI(매출 지분, CapEx, OpEx)를 보고해야 합니다. "정렬" 및 "적격"). 2025년부터 단순화로 직원 1,000명 미만 기업은 분류 보고에서 제외되며 나머지는 그들이 대표하는 활동으로 제한될 수 있습니다. 매출, CapEx 또는 OpEx의 최소 10%.
GHG 프로토콜: 참조 프레임워크
Il 온실가스 프로토콜 온실가스 배출 회계기준 세계자원연구소(WRI)와 세계비즈니스협의회(WRI)가 개발한 세계에서 가장 널리 퍼진 것 지속 가능한 개발(WBCSD)을 위해. 거의 모든 보고 프레임워크(CSRD/ESRS, CDP, GRI, ISO 14064)은 GHG 프로토콜을 참조하거나 이를 기반으로 합니다.
세 가지 범위: 정확한 정의
세 가지 "범위"로 구분하면 배출을 명확하게 파악할 수 있습니다. 공급망 내 서로 다른 행위자 간의 계산 중복을 방지합니다.
범위 1: 직접 배출
조직이 소유하거나 통제하는 배출원에서 배출됩니다. 여기에는 보일러, 오븐, 회사 소유 차량에서 화석 연료를 태우는 행위, 공정 배출(예: 화학 반응으로 인한 CO2, 가축에서 발생하는 CH4) 비산 배출 (냉매 누출, 시스템에서 가스 누출). 관련 GHG 가스는 다음 7가지입니다. 교토 의정서: CO2, CH4, N2O, HFC, PFC, SF6, NF3, 모두 변환됨 CO2 상당(CO2e) IPCC 지구 온난화 지수(GWP)를 사용합니다.
범위 2: 구매한 에너지의 간접 배출
전기, 열, 증기 또는 냉각의 생성과 관련된 배출 조직에서 구매하고 소비합니다. GHG 프로토콜 범위 2 지침(2015)에서는 다음을 요구합니다. 보고 두 가지 별개의 방법: 방법 위치 기반 (위치에 있는 전력망의 배출 계수를 사용하십시오. 소비, 예를 들어 이탈리아 네트워크의 평균 요소) 및 방법 시장 기반 (차액계약, 인증서와 같은 시장 도구의 요소를 사용합니다. 재생에너지/GO, 전력 구매 계약). 둘 다 신고해야 합니다.
범위 3: 기타 간접 배출
가치 사슬의 배출량은 15개의 업스트림 및 다운스트림 범주로 구분됩니다. 이는 일반적으로 가장 중요하며(전체 설치 공간의 평균 70-80%) 가장 측정하기 어렵다. ESRS는 범위 3 범주에 대한 보고를 요구합니다. 이중 중요성 평가를 통해 식별된 "재료"입니다.
15가지 범위 3 카테고리
| # | 범주 | 유형 | 일반적인 대상 |
|---|---|---|---|
| 1 | 구매한 상품 및 서비스 | 업스트림 | 모든 부문 |
| 2 | 자본재 | 업스트림 | 제조, 건설 |
| 3 | 연료 및 에너지 관련 활동 | 업스트림 | 모든 부문 |
| 4 | 업스트림 운송 및 유통 | 업스트림 | 소매, 제조 |
| 5 | 운영 중 발생하는 폐기물 | 업스트림 | 제조업, 식품 |
| 6 | 출장 | 업스트림 | 서비스, 기술 |
| 7 | 직원 통근 | 업스트림 | 모든 부문 |
| 8 | 업스트림 임대 자산 | 업스트림 | 부동산, 소매 |
| 9 | 하류 운송 | 하류 | 제조, FMCG |
| 10 | 판매된 제품의 처리 | 하류 | 원료, 화학 |
| 11 | 판매된 제품의 사용 | 하류 | 자동차, 전자 |
| 12 | 임종치료 | 하류 | 포장, 내구재 |
| 13 | 다운스트림 임대 자산 | 하류 | 부동산 |
| 14 | 프랜차이즈 | 하류 | 식음료, 소매 |
| 15 | 투자 | 하류 | 은행, 투자 자금 |
데이터 모델: 탄소 회계 영역 모델링
모든 탄소회계 플랫폼의 핵심은 탄소정보를 충실히 반영하는 강력한 데이터 모델입니다. GHG 프로토콜의 개념. 주요 엔터티의 속성과 관계를 살펴보겠습니다.
주요 엔터티
전체 시스템에는 조직, 시설, EmissionSource, EmissionFactor, 활동 및 계산. 다음은 SQLAlchemy를 사용한 Python 모델입니다.
# 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")배출계수 데이터베이스: DEFRA, EPA, ecoinvent, Climatiq
배출계수는 탄소 계산의 수학적 핵심입니다. 배출계수는 수량을 변환합니다. 활동량(디젤 리터, 소비 kWh, 이동 거리, 소비 유로)을 CO2e 톤으로 표시합니다. 요인의 품질과 업데이트는 보고서의 품질을 직접적으로 결정합니다.
주요 데이터베이스
| 데이터베이스 | 관리자 | 적용 범위 | 업데이트 | 입장 |
|---|---|---|---|---|
| 데프라/BEIS | 영국 정부(DESNZ) | 영국 + 국제, 모든 범위 | 연간(7월) | 무료(엑셀) |
| EPA 온실가스 허브 | 미국 EPA | 미국, 모빌리티, 에너지, Scope 3 | 연간(1월) | 무료(엑셀) |
| 에코벤트 | 에코발명협회 | 글로벌, 전체 LCA, 18,000개 이상의 데이터세트 | 반년마다 | 유료 라이센스 |
| IPCC EF 데이터베이스 | IPCC | 글로벌, 국가 재고 | 모든 평가 보고서에는 | 무료 |
| Climatiq API | 기후 | 다중 소스, 50,000개 이상의 요소 | 연속(실시간) | API(부분유료) |
| AIB 잔여 믹스 | 발행 기관 협회 | 유럽, Scope 2 시장 기반 | 연간 | 무료 |
| IEA 전력 | 국제에너지기구 | 글로벌 전력망 요인 | 연간 | 부분적으로 무료 |
Climatiq API 통합
Climatiq은 50,000개 이상의 검증된 배출 계수, 적용 범위를 갖춘 REST API를 제공합니다. ISO 14067 및 GHG 프로토콜이며 계산 통합에 가장 많이 사용되는 솔루션 중 하나입니다. 프로그래밍 방식으로 배출. API는 활동 기반 방법으로 범위 1, 2, 3을 지원합니다. 그리고 지출 기반.
# 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.co2e탄소 회계를 위한 마이크로서비스 아키텍처
기업 탄소 회계 플랫폼은 이기종 데이터 흐름을 관리해야 합니다. (에너지 요금, ERP 데이터, 비용 보고서, 공급업체 데이터), 복잡하고 감사 가능한 계산, 다양한 형식으로 보고서를 생성합니다. 마이크로서비스 아키텍처 및 선택 이러한 요구 사항에는 당연합니다.
4가지 핵심 서비스
1. 데이터 수집 서비스
ERP API(SAP, Oracle), 공과금 청구서 등 다양한 소스로부터 데이터 수집을 담당합니다. OCR/파서, 출장 관리 도구(Concur, TravelPerk), 조달 데이터를 통해 CSV 수동 업로드, 공급업체 API. 수집을 위한 엔드포인트를 노출하고 관리합니다. 측정 단위의 정규화.
2. 계산 엔진
계산 심장: 정규화된 활동을 수신하고 배출 요인을 선택합니다. 적절하게(로컬 또는 Climatiq API를 통해) GHG 프로토콜 공식을 적용하고 감사에 대한 명시적인 계산 공식을 사용하면 결과를 변경할 수 없습니다. 지원 요인이 업데이트되면 과거 재계산이 수행됩니다.
3. 신고 서비스
CSRD/ESRS(XBRL 태깅 포함) 프레임워크에 필요한 형식으로 보고서를 생성합니다. CDP 설문지(JSON/XML 형식), GRI Standards 공개, 내부 보고서 대시보드용. 보증을 위해 보고서 버전 관리 및 디지털 서명을 관리합니다.
4. 감사 추적 서비스
누가 데이터를 입력했는지, 어떤 요소를 입력했는지 등 모든 작업에 대한 불변의 로그를 유지합니다. 계산할 때 누가 승인했는지 사용되었습니다. 데이터 계보 지원 완료: 소스 데이터에서 최종 보고서의 숫자까지. 보증에 필수 감사관에 의해.
아키텍처 다이어그램
┌─────────────────────────────────────────────────────────────────┐
│ 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)
계산 엔진: FastAPI를 사용한 Python 구현
계산 엔진이자 가장 중요한 구성 요소입니다. 정확하고 감사 가능해야 하며, 버전이 지정되어 있으며 수천 개의 계산을 병렬로 처리할 수 있습니다. 구현은 다음과 같습니다. 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
pass범위 3 자동화: ERP, 조달 및 여행
Scope 3과 탄소회계의 가장 큰 과제: 수십 개의 데이터에 분산된 데이터 시스템, 다양한 성숙도를 지닌 공급업체, 다양한 계산 방법론 카테고리. 자동화는 운영상 지속 가능하게 만드는 유일한 방법입니다.
범위 3 범주 1(구매한 상품)에 대한 SAP 통합
# 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)
}자동 보고: CSRD/ESRS, CDP 및 GRI
규제 보고서를 생성하는 것은 탄소 계산에서 가장 시간이 많이 걸리는 프로세스인 경우가 많습니다. 자동화로 시간이 몇 주에서 몇 시간으로 대폭 단축되고 일관성이 향상됩니다. 그리고 각 그림의 출처를 추적할 수 있는지 확인합니다.
CSRD/ESRS E1 보고서 구조(기후 변화)
ESRS E1 기후 표준은 다음에 대한 공개를 요구합니다: 기후 거버넌스, 전략 및 시나리오 분석, 위험 관리, 지표 및 목표. 측정항목 배출량은 ESRS E1-6에 정의되어 있으며 세 가지 범위 모두에 대한 데이터가 필요합니다.
# 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"
}
}감사 추적 및 데이터 계보
자격을 갖춘 감사자의 제한적인 보증을 요구하는 CSRD에서는 감사 추적이 불가능합니다. 선택적 요구 사항: 플랫폼의 백본. 보고서의 각 호 명시적인 계산 공식을 사용하여 원래 소스를 다시 추적할 수 있어야 합니다.
# 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 lineage엔터프라이즈 플랫폼 비교: 구축과 구매
맞춤형 플랫폼을 구축하기 전에 솔루션을 평가하는 것이 필수적입니다. 기업 가능. 탄소 회계 소프트웨어 시장은 매우 성숙해졌습니다. 선도적인 플랫폼은 대부분의 표준 사용 사례를 포괄합니다.
| 플랫폼 | 강점 | 대상분야 | 참고 가격 | CSRD 준비 |
|---|---|---|---|---|
| 페르세포네 | 투자자급 보고, XBRL 태깅, SEC 중심 | 금융, 기업 | $50K-500K/년 | Si |
| 유역 | 보고 속도, 고급 Scope 3 도구 | 기술, 기업 | $100K-1M/년 | Si |
| 스피라 | 통합 LCA, 산업 규정 준수, 위험 관리 | 제조, 에너지, 화학 | 요청 시 | Si |
| 플랜 A | 간단한 UX, 빠른 구현 | 중소기업, 중소기업 | $10K-100K/년 | 부분 |
| IBM 엔비지 | 40,000개 이상의 배출계수, ERP 통합 | 기업, 유틸리티 | 요청 시 | Si |
| 맞춤형 빌드 | 완전한 유연성, 내부 시스템의 기본 통합 | 특정 요구 사항이 있는 대기업 | $500K-5M 개발 | 의존한다 |
맞춤 제작과 구매 시기
다음과 같은 경우 플랫폼을 구매하세요. 표준 요구 사항이 있고 라이브로 전환하고 싶습니다. 3~6개월 후에는 지속 가능성을 전담하는 엔지니어링 팀이 없어지고 데이터의 양이 늘어나게 됩니다. 그리고 관리 가능합니다.
다음과 같은 경우 맞춤 제작하세요. 데이터를 사용하는 매우 구체적인 생산 프로세스가 있습니다. 표준 플랫폼으로는 관리할 수 없고 레거시 시스템과 기본적으로 통합하기를 원합니다. SaaS 또는 데이터 볼륨과 호환되지 않는 데이터 상주 요구 사항이 있으며 연간 레코드 수는 수백만 건입니다(SaaS 비용은 엄청납니다).
사례 연구: CSRD 2025를 적용한 이탈리아 제조업 중소기업
1,200명의 직원을 보유한 이탈리아의 정밀 엔지니어링 제조 회사로, 매출액은 3억 5천만 유로이며 토리노, 밀라노, 브레시아에 공장이 위치하고 있습니다. CSRD 1차 조사에 참여하고 2024 회계연도에 대한 첫 번째 보고서를 제출해야 합니다. 2025년 6월까지.
2024년 배출 인벤토리
# case_study/inventory_2024.py - Esempio inventario GHG realistico
# Dati rappresentativi per PMI manifatturiera italiana
INVENTORY_2024 = {
"organization": "MeccanicaPrecisione SpA",
"reporting_year": 2024,
"boundary": "Operational Control",
"currency": "EUR",
"scope_1": {
"natural_gas_combustion": {
"consumption_m3": 485_000, # caldaie industriali
"consumption_kwh": 5_116_750, # conversione: 1 m3 NG = 10.55 kWh
"emission_factor_kgco2e_kwh": 0.18306, # DEFRA 2024
"co2e_tonnes": 937.0,
"source": "DEFRA 2024 - Natural Gas Gross CV"
},
"diesel_mobile": {
"consumption_liters": 42_000, # carrelli elevatori, mezzi operativi
"emission_factor_kgco2e_liter": 2.56, # DEFRA 2024
"co2e_tonnes": 107.5,
"source": "DEFRA 2024 - Diesel"
},
"fugitive_refrigerants": {
"substance": "R-410A",
"kg_recharged": 45,
"gwp_ar6": 2088,
"co2e_tonnes": 94.0,
"source": "IPCC AR6 GWP100"
},
"total_co2e_tonnes": 1138.5
},
"scope_2": {
"purchased_electricity": {
"consumption_kwh": 8_250_000,
"location_based": {
"emission_factor": 0.233, # IEA Italy 2024 kgCO2e/kWh
"co2e_tonnes": 1922.3,
"source": "IEA 2024 Italy Grid"
},
"market_based": {
"go_certificates_kwh": 4_125_000, # 50% rinnovabili con GO
"residual_mix_factor": 0.395, # AIB Italy residual mix 2024
"co2e_tonnes": 1629.4, # solo su 50% non coperto da GO
"source": "AIB Italy Residual Mix 2024"
}
}
},
"scope_3": {
"cat_1_purchased_goods": {
"total_spend_eur": 145_000_000,
"method": "spend_based + supplier_specific (top 20 suppliers)",
"co2e_tonnes": 28_450.0,
"data_quality": "mix: 35% supplier-specific, 65% spend-based"
},
"cat_4_upstream_transport": {
"tonne_km": 3_200_000,
"emission_factor": 0.089, # kgCO2e/tonne-km road freight DEFRA
"co2e_tonnes": 284.8,
},
"cat_5_waste": {
"waste_tonnes": 1_850,
"co2e_tonnes": 185.0,
"method": "waste-type specific factors"
},
"cat_6_business_travel": {
"total_km_air": 1_250_000,
"total_km_rail": 320_000,
"co2e_tonnes": 312.5,
"source": "DEFRA 2024 with RFI=1.9"
},
"cat_7_employee_commuting": {
"employees": 1_200,
"avg_km_per_day": 22,
"working_days": 220,
"mode_split": {"car_solo": 0.65, "car_sharing": 0.05,
"public_transport": 0.25, "cycling": 0.05},
"co2e_tonnes": 892.5
},
"cat_11_use_of_products": {
"units_sold": 45_000,
"avg_energy_use_kwh_per_unit_per_year": 850,
"product_lifetime_years": 15,
"co2e_tonnes_per_year": 2_250.0,
"note": "Calcolato per anno di uso, non lifetime"
},
"total_co2e_tonnes": 32_374.8
},
"summary": {
"scope_1_tco2e": 1_138.5,
"scope_2_location_tco2e": 1_922.3,
"scope_2_market_tco2e": 1_629.4,
"scope_3_tco2e": 32_374.8,
"total_location_based_tco2e": 35_435.6,
"total_market_based_tco2e": 35_142.7,
"intensity_tco2e_per_meur_revenue": 101.3,
"intensity_tco2e_per_fte": 29.5,
"scope_3_percentage_of_total": 91.4, # tipico nel manifatturiero
}
}사례 연구에 대한 관찰
가장 중요한 결과는 Scope 3이 배출량의 91.4% 합계, 카테고리 1(원자재 및 부품)만 가격의 80% 가치 범위 3. 이는 제조 부문의 전형적인 현상이며 CSRD가 요구하는 이유를 설명합니다. 공급망 보고: Scope 3이 없으면 탄소 회계 처리 범위는 다음보다 적습니다. 실제 효과의 10%입니다.
Scope 2 위치 기반(1,922tCO2e)과 시장 기반(1,629tCO2e)의 차이는 다음과 같습니다. 50%에 대한 재생 에너지 인증서(원산지 보증 - GO) 구매 전기 소비. 2024년 이탈리아 잔여 혼합량(0.395kgCO2e/kWh)은 해당 계수보다 높습니다. 네트워크 평균(0.233kgCO2e/kWh): 이는 직관에 어긋나지만 방법론적으로는 정확합니다. 잔여 혼합에서는 이미 GO로 인증된 에너지를 제외하고 따라서 할당량을 "포함"하기 때문입니다. 고강도 소스보다 높습니다.
테스트: 계산 검증
배출량 계산은 금융 규정과 동일하게 엄격하게 테스트되어야 합니다. 하나의 변환 계수에 오류가 있으면 크기 순서대로 오류가 발생할 수 있습니다. 수백 톤의 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.01과학 기반 목표(SBTi) 및 대시보드
Il 과학 기반 목표 계획(SBTi) 정확한 기준을 정했습니다 파리 협정에 따른 배출 감소 목표를 위해. SBTi 기업 표준은 다음을 요구합니다: 2030년까지 Scope 1+2 42% 감소 (2020년 기준) 1.5°C 시나리오의 경우, 범위 3 적용 범위(해당되는 경우) 총 배출량의 40% 이상(제조업에서는 거의 항상 그렇습니다).
플랫폼 대시보드에는 현재 배출량뿐만 아니라 하지만 목표로 가는 길: SBTi에 필요한 감소 곡선, 연도별 실제 배출량 및 감축 계획에 따른 예측 계획됨(재생 에너지, 차량 전기화, 프로세스 효율성을 위한 PPA).
# 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 riduzione모범 사례 및 안티 패턴
모범 사례
- 계산의 불변성: 계산되고 저장되면 방출량은 절대 수정하면 안 됩니다. 요인이 변경되면 새 버전의 계산이 생성됩니다. 감사 추적에는 두 버전이 모두 표시되어야 합니다.
- Scope 2 방법을 모두 보고합니다. 위치 기반과 시장 기반이 있습니다. ESRS 및 GHG 프로토콜에 따라 둘 다 필수입니다. 단순히 시장 기반으로 보고하지 마세요. 더 낮기 때문입니다.
- 문서 범위 3 중요성: 15개 범주를 모두 계산할 필요는 없습니다. 범위 3. 그러나 포함/제외 이유를 중요성 분석을 통해 입증해야 합니다. 각 카테고리. 이는 명시적인 ESRS 요구 사항입니다.
- 배출계수 버전 관리: 요인은 매년 변합니다. 각 계산에 사용된 요소의 스냅샷을 유지하세요. 업데이트하지 않음 새 버전으로 명시적인 재계산을 생성하지 않고 소급하여 사용할 수 있습니다.
- 데이터 품질 플래그: 각 데이터를 품질별로 분류(측정, 계산, 추정, 지출 기반). ESRS에는 품질 선언이 필요합니다 Scope 3 데이터. 낮은 품질의 데이터가 사용될 수 있지만 반드시 선언된다.
- 경계 문서: 포함된 엔터티를 명시적으로 문서화 보고 범위 내 및 제외된 범위 내에서 정당한 사유가 있는 경우 (임계값 <5%, 데이터 없음 등).
피해야 할 안티패턴
- 시장 기반 범위 2만 사용하십시오. GHG 프로토콜은 두 가지 모두를 요구합니다. 없이 시장 기반(일반적으로 GO/PPA 덕분에 더 낮음)만 보고합니다. 위치 기반이며 방법론적으로 부정확하고 잠재적인 greenwashing입니다.
- 업데이트되지 않은 배출계수: 5년 이상 된 요소를 사용하세요. 특히 변화하는 전력망에 심각한 오류가 발생합니다. 매년 재생에너지의 성장과 함께.
- 통합 시 중복 배출: 자회사인 경우 배출량을 계산하고 모회사는 이를 연결에 포함합니다. 지분 지분이나 운영 통제 방법을 일관되게 사용해야 합니다.
- 불확실성을 무시하십시오: 모든 배출량 계산에는 불확실성, 특히 범위 3. 없이 숫자를 제시하지 마십시오. 신뢰 범위와 사용된 방법을 나타냅니다.
- 보증이 없는 보고서: CSRD를 사용하면 제한된 보증 없이 보고할 수 있습니다. Wave 1에 대한 규제 요구 사항을 충족하지 않습니다. 보증 계획 나중에 생각한 것이 아니라 첫해부터.
결론 및 구현 로드맵
엔터프라이즈급 탄소 회계 플랫폼을 설계하는 것은 복잡한 프로젝트입니다. 다학제적 기술이 필요한 일이다. 그러나 개념적 구조는 명확합니다. 온실가스 프로토콜 회계 프레임워크, 데이터베이스를 제공합니다. 배출계수 (DEFRA, EPA, Climatiq)은 계수를 제공합니다. 는마이크로서비스 아키텍처 확장성과 유지관리성을 보장합니다. 그리고불변의 감사 추적 CSRD에 대한 신뢰성을 보장합니다.
2025~2026년에 CSRD에 접근해야 하는 SME의 경우 실질적인 권장 사항은 다음과 같습니다. i용 성숙한 SaaS 플랫폼(Persefoni, Plan A 또는 Watershed)으로 시작 처음 2년 동안은 실제 데이터를 수집하고 데이터 격차를 이해한 다음 맞춤형 솔루션을 구축할지 아니면 SaaS를 유지할지 평가하세요. 조직의 90% 처음부터 빌드할 필요가 없습니다.
대신 ESG 플랫폼을 제품으로 사용하는 엔지니어링 팀의 경우 개념 이 기사에 설명된 - GHG 프로토콜 데이터 모델, 감사 가능한 계산 엔진, 배출계수 API와 통합 및 규제 보고서 생성 - sono 구축할 기초. 탄소회계 소프트웨어 시장은 성장할 것이다 2030년까지 매년 27%씩 증가: 특히 전문 솔루션을 위한 여지가 있습니다. 플랫폼에서 다루지 않는 프로세스 데이터 요구사항이 있는 산업 부문용 일반주의자.
권장 구현 로드맵
| 단계 | 지속 | 목적 | 출력 |
|---|---|---|---|
| 1단계 | 1~2개월 | 이중 중요성 평가 + 경계 | 범위 3 물질 범주 목록, 보고 경계 |
| 2단계 | 2~3개월 | 범위 1 및 2 데이터 수집 | 청구서, SAP, SCADA를 포함한 자동화된 파이프라인 |
| 3단계 | 3~4개월 | 범위 3 자재 범주(지출 기반) | 고양이에 대한 범위 3 계산. 1, 4, 6, 7, 11 |
| 4단계 | 1~2개월 | CSRD/ESRS E1 보고 및 감사 | 제한된 보증을 위한 보고서 준비됨 |
| 5단계 | 마디 없는 | 데이터 품질 및 공급업체 참여 개선 | 지출 기반 감소, 공급업체별 Scope 3 증가 |
필수 리소스
- GHG 프로토콜 기업 표준: gghgprotocol.org - 표준 참조, 무료 및 다운로드 가능
- DEFRA 배출계수 2024: gov.uk/government/publications/ 온실가스 보고 변환계수-2024
- Climatiq API: climatiq.io/docs - 전체 문서 e 무료 빠른 시작
- ESRS E1 표준: efrag.org - 유럽 재무 보고 자문 그룹, 최종 ESRS 표준
- SBTi 기업 매뉴얼: sciencebasedtargets.org - 가이드 과학적 기반의 목표 정의
- 에코인벤트 데이터베이스: ecoinvent.org - LCA 데이터베이스 상세한 프로세스 데이터가 포함된 Scope 3
EnergyTech 시리즈의 향후 기사
다음 기사에서는 디지털 에너지 트윈: 만드는 방법 배출 감소 시나리오를 시뮬레이션하기 위한 산업 플랜트의 가상 복제본 실제 세계에서 구현하기 전에. 점점 더 중심이 되는 기술 대기업의 탈탄소화 전략.
데이터 비즈니스에 적용되는 AI 기술에 대해 자세히 알아보려면 다음을 참조하세요. 시리즈 데이터 웨어하우스, AI 및 디지털 혁신, 특히 기사에 안정적인 AI를 위한 데이터 거버넌스 및 데이터 품질 그리고 위로 비즈니스용 MLOps, 계산모델 관리에 직접 적용할 수 있는 주제를 다룬다. 생산 중 배출.