범위 1, 2, 3: ESG 보고 소프트웨어를 위한 데이터 모델링
Il 온실가스 프로토콜 (온실가스 프로토콜)은 사실상 보편적인 프레임워크가 되었습니다. 기업의 온실가스 배출량을 측정하고 보고합니다. 세계자원연구소(WRI)에서 개발 지속가능발전을 위한 세계기업협의회(WBCSD)는 배출의 세 가지 "범위"를 정의합니다. 이는 조직의 전체 가치 사슬을 포괄합니다.
의 발효와 함께 CSRD(기업 지속가능성 보고 지침) 연합의 유럽 및 표준 ESRS E1, 기업을 포함한 수천 개의 기업 소프트웨어 및 기술은 이제 자체 배출량 데이터를 수집, 계산 및 게시해야 합니다. 엄격한 방법론에 따라. 2024년부터 직원 500명 이상 대기업은 2025년 첫 번째 보고 의무와 함께 데이터 수집을 시작합니다. 직원이 250명 이상인 회사 2026년부터 뒤따를 예정이다.
그런데 내가 왜? 개발자 범위 1, 2, 3을 이해해야 합니까? 시스템을 구축하는 이유 신뢰할 수 있는 ESG는 지속가능성 관리자만의 문제가 아닙니다. 데이터 엔지니어링. 일관된 데이터 모델, 정확한 계산 파이프라인, API가 필요합니다. 활동 수집 및 감사 준비 보고서. 누가 이러한 시스템을 설계하고 구현하는지 결정합니다. 모든 ESG 보고의 품질, 즉 신뢰성.
이 글에서 우리는 ESG 보고를 위한 데이터 모델링 시스템, 범위 정의부터 데이터베이스 테이블까지, Python 계산부터 FastAPI 엔드포인트까지, 직원이 200명인 소프트웨어 중소기업에 대한 전체 사례 연구까지.
무엇을 배울 것인가
- GHG 프로토콜의 구조: 정확한 정의가 포함된 범위 1, 2, 3
- 직접, 간접, 가치 사슬 배출을 위한 개체 관계 모델
- 범위 2의 위치 기반과 시장 기반: 데이터에 대한 차이점 및 영향
- 15개의 Scope 3 카테고리 및 소프트웨어 회사와 관련된 카테고리
- 완전한 SQLAlchemy 스키마: Organization, Facility, EmissionSource, EmissionFactor, ActivityData
- 활동 기반 추정 및 지출 기반 추정을 사용한 Python 계산
- 집계, 전년 대비 비교 및 SBTi 목표 조정
- 작업 제출, 배출량 계산 및 보고서 생성을 위한 FastAPI가 포함된 REST API
- pytest 및 배출계수 검증을 통한 GHG 계산을 위한 단위 테스트
- 엔드투엔드 사례 연구: 소프트웨어 SME 직원 200명, 범위 1+2+3 완료
그린 소프트웨어 시리즈 — 10개 기사
| # | 제목 | 상태 |
|---|---|---|
| 1 | 그린소프트웨어공학과 SCI 원칙 | 게시됨 |
| 2 | CodeCarbon: Python 코드 방출 측정 | 게시됨 |
| 3 | Carbon Aware SDK: 워크로드 전환 및 시간 전환 | 게시됨 |
| 4 | Climatiq API: 배출계수 및 탄소 계산 | 게시됨 |
| 5 | 범위 1, 2, 3: ESG 보고를 위한 데이터 모델링 | 현재 기사 |
| 6 | ESG 및 CSRD: 기술 기업에 대한 유럽의 의무 | Prossimamente |
| 7 | 지속 가능한 소프트웨어 패턴: 영향이 적은 아키텍처 | Prossimamente |
| 8 | GreenOps: 지속 가능성을 위한 클라우드 최적화 | Prossimamente |
| 9 | 범위 3 파이프라인: 가치 사슬 자동화 | Prossimamente |
| 10 | AI 탄소 발자국: LLM, 교육 및 추론 | Prossimamente |
GHG 프로토콜: 배출에 대한 보편적 프레임워크
2001년에 발행되고 2004년에 업데이트된 온실가스 프로토콜 기업 표준(GHG Protocol Corporate Standard)은 거의 모든 기후 보고 프레임워크: CDP, TCFD, CSRD/ESRS E1, SBTi 등 그들은 그것을 직접적으로 언급합니다. 그 강점은 개념적 명확성에 있습니다. 조직은 상호 배타적이지만 전체적으로는 철저한 세 가지 범주로 분류됩니다.
프레임워크는 다음 개념을 사용합니다. 조직 경계 (조직 경계) 계산에 포함할 배출량을 결정합니다. 두 가지 접근 방식이 있습니다.지분 공유 접근법 (재정참여율 기준) 및 통제 접근법 (컨트롤을 기준으로 운영 또는 재무). 대부분의 회사는 운영 통제를 주요 기준으로 채택합니다.
세 가지 범위: 개요
| 빗자루 | 정의 | 전형적인 예 | ESRS E1 필수 |
|---|---|---|---|
| 범위 1 | 조직이 소유하거나 통제하는 배출원으로부터의 직접 배출 | 천연 가스 보일러, 회사 차량, 디젤 발전기, 산업 공정 | 의무사항 |
| 범위 2 | 에너지(전기, 증기, 열, 냉기) 구매로 인한 간접 배출 | 사무실 전력, 데이터센터 냉방, 지역난방 | 필수(두 방법 모두) |
| 범위 3 | 가치 사슬(업스트림 및 다운스트림)의 기타 모든 간접 배출 | 출장, 출퇴근, 상품/서비스 구매, 판매상품 이용 | 필수(재료 카테고리) |
범위 1: 직접 배출에 대한 데이터 모델
Scope 1 배출은 회사가 직접 통제하는 물리적 소스에서 발생합니다. 그들을 위해 소프트웨어 회사는 일반적으로 규모가 작지만 무시할 수 없습니다. 여기에는 난방 시설도 포함됩니다. 사무실(천연가스), 백업 발전기, 회사 차량(회사 차량) 또는 하드웨어 배송 밴).
GHG 프로토콜은 Scope 1 배출의 네 가지 범주를 식별합니다.
- 고정연소: 화석 연료를 연소하는 보일러, 히터, 발전기
- 이동 연소: 휘발유, 디젤 또는 LPG로 구동되는 회사 차량
- 공정 배출: 화학적 또는 생물학적 반응(소프트웨어와 거의 관련 없음)
- 비산 배출: HVAC 시스템의 냉매(R-410A, R-134a) 누출
범위 1 계산의 기본 공식은 다음과 같습니다.
배출량(tCO2e) = 활동정보 × 배출계수 × GWP
어디:
- 주어진 활동: 소비된 연료량(리터, m³, kWh)
- 배출계수: 연료 단위당 kg CO2 (출처: IPCC, DEFRA, IEA)
- GWP: 지구 온난화 지수(CO2=1, CH₄=28, N2O=265, HFC는 다양함)
범위 1의 엔터티-관계 다이어그램에는 다음과 같은 주요 엔터티가 포함됩니다.
-- Entity-Relationship: Scope 1 Emissioni Dirette
ORGANIZATION (1) ---< FACILITY (>1)
|id, name, tax_id |id, org_id, name, type, country, area_m2
FACILITY (1) ---< EMISSION_SOURCE (>1)
|id, facility_id, source_type, description
| source_type: STATIONARY | MOBILE | FUGITIVE | PROCESS
EMISSION_SOURCE (1) ---< ACTIVITY_DATA (>1)
|id, source_id, period_start, period_end
|quantity, unit, data_quality, evidence_url
ACTIVITY_DATA (N) ---> EMISSION_FACTOR (1)
|id, gas, fuel_type, unit_from, unit_to
|factor_value, gwp_ar5, source, year, region
ACTIVITY_DATA (1) ---< GHG_CALCULATION (1)
|id, activity_id, scope, co2_kg, ch4_kg
|n2o_kg, hfc_kg, co2e_total_kg, method
|calculated_at, calculated_by범위 2: 위치 기반과 시장 기반
Scope 2는 기술 기업의 경우 특히 복잡합니다. 사무실과 데이터 센터에서 간접 배출의 주요 원인. GHG 프로토콜 범위 2 지침(2015)에는 다음과 같은 보고 의무가 도입되었습니다. 둘 다 방법: 위치 기반 및 시장 기반.
위치 기반과 시장 기반: 주요 차이점
| 나는 기다린다 | 위치 기반 | 시장 기반 |
|---|---|---|
| 정의 | 소비가 발생하는 지역 전력망의 평균 강도 | 계약에 따라 에너지를 구매한 발전기의 배출량 |
| 배출계수 | 국가/지역별 그리드 평균 배출계수(gCO₂/kWh) | 공급업체별 요인 또는 잔여 혼합 요인 |
| 계약 문서 | 해당 없음 | GO(원산지 보증), REC(재생 에너지 인증서), PPA |
| 예시 이탈리아 2024 | ~310 gCO²/kWh(AIB 잔류 혼합 데이터) | 100% 검증된 재생 가능 GO를 구매하는 경우 0gCO²/kWh |
| GHG 프로토콜 2025 제안 | 시간 정밀도를 갖춘 새로운 요인 계층 구조 | 재생 에너지에 대한 시간별 매칭 요청 |
| ESRS E1에서 사용 | 의무공개(E1-6) | 의무공개(E1-6) |
데이터베이스에서 범위 2를 올바르게 모델링하려면 다음을 관리해야 합니다. 에너지 인증서 (GO/REC)를 "소비"되는 별도의 개체로 회사의 전력 소비에 대해 시장 기반 요소를 0으로 낮춥니다.
-- Modello Scope 2: Energia e Certificati
ELECTRICITY_CONSUMPTION
id UUID PRIMARY KEY
facility_id UUID REFERENCES FACILITY
period_start DATE
period_end DATE
kwh_consumed DECIMAL(15,3)
meter_id VARCHAR(50)
-- Location-based
grid_region VARCHAR(20) -- e.g. 'IT', 'DE-WEST', 'FR'
grid_ef_g_kwh DECIMAL(8,3) -- gCO2e/kWh from IEA/AIB
lb_co2e_kg DECIMAL(15,3) GENERATED ALWAYS AS
(kwh_consumed * grid_ef_g_kwh / 1000)
-- Market-based
supplier_ef_g_kwh DECIMAL(8,3) DEFAULT NULL
residual_mix_ef DECIMAL(8,3) DEFAULT NULL
mb_co2e_kg DECIMAL(15,3) -- calcolato dopo allocazione GO/REC
RENEWABLE_CERTIFICATE
id UUID PRIMARY KEY
cert_type VARCHAR(10) -- 'GO' | 'REC' | 'I-REC'
cert_id VARCHAR(100) UNIQUE
technology VARCHAR(30) -- 'WIND' | 'SOLAR' | 'HYDRO'
country VARCHAR(2)
production_date DATE
kwh_value DECIMAL(15,3)
issuer VARCHAR(100)
status VARCHAR(20) -- 'ACTIVE' | 'CANCELLED' | 'EXPIRED'
CERTIFICATE_ALLOCATION
id UUID PRIMARY KEY
cert_id UUID REFERENCES RENEWABLE_CERTIFICATE
consumption_id UUID REFERENCES ELECTRICITY_CONSUMPTION
kwh_allocated DECIMAL(15,3)
allocation_date DATE
-- Vincolo: kwh_allocated <= RENEWABLE_CERTIFICATE.kwh_value범위 3: 가치 사슬의 15가지 범주
범위 3은 일반적으로 주요 소스 소프트웨어 회사의 배출량: 전체의 70~80% 이상을 차지할 수 있습니다. GHG 프로토콜 계산 기술 지침 Scope 3 배출은 업스트림(상품 및 서비스 구매와 연결됨)으로 구분된 15개 범주를 정의합니다. 및 다운스트림(판매된 제품의 사용과 관련됨).
소프트웨어 회사를 위한 15가지 범위 3 범주
| # | 범주 | 유형 | 소프트웨어 관련성 | 계산방법 |
|---|---|---|---|---|
| 1 | 구매한 상품 및 서비스 | 업스트림 | 높은 (SaaS, 클라우드, 하드웨어) | 지출 기반 또는 공급업체별 |
| 2 | 자본재 | 업스트림 | 미디어(노트북, 서버, 가구) | 지출 기반 또는 물리적 단위 |
| 3 | 연료 및 에너지 관련 활동 | 업스트림 | 평균(T&D 손실, 상류 연료) | 활동 기반 |
| 4 | 업스트림 운송 및 유통 | 업스트림 | 낮음(하드웨어 제공) | 지출 기반 또는 거리 기반 |
| 5 | 운영 중 발생하는 폐기물 | 업스트림 | 낮은 | 폐기물 유형별 |
| 6 | 비즈니스 여행 | 업스트림 | 높은 (항공편, 호텔, 기차) | 거리 기반 또는 지출 기반 |
| 7 | 직원 통근 | 업스트림 | 높은 (출퇴근, 원격근무) | 설문조사 + 거리 기반 |
| 8 | 업스트림 임대 자산 | 업스트림 | 중형 (임대 사무실) | 자산별 |
| 9 | 하류 운송 | 하류 | 낮은 | 거리 기반 |
| 10 | 판매된 제품의 처리 | 하류 | 해당 없음(소프트웨어) | 활동 기반 |
| 11 | 판매된 제품의 사용 | 하류 | 높은 (소프트웨어를 실행하는 데 필요한 에너지) | 평생 에너지 사용량 |
| 12 | 임종치료 | 하류 | 낮음(사용자 장치) | 폐기물 유형별 |
| 13 | 다운스트림 임대 자산 | 하류 | 낮은 | 자산별 |
| 14 | 프랜차이즈 | 하류 | 해당 없음 | 해당 없음 |
| 15 | 투자 | 하류 | 미디어(VC/스타트업 포트폴리오) | 포트폴리오 기반 |
소프트웨어 회사의 경우 가장 관련성이 높은 카테고리는 일반적으로 다음과 같습니다. 고양이. 1 (클라우드 컴퓨팅 구입), 고양이. 6 (출장), 고양이. 7 (직원 출퇴근) e 고양이. 11 (사용자가 소비하는 에너지 소프트웨어를 실행합니다). 후자는 종종 과소평가되지만 수백만 달러 규모의 엔터프라이즈 SaaS의 경우 사용자 수가 엄청날 수 있습니다.
SQLAlchemy 스키마: 완전한 모델
이제 세 가지 범위를 모두 지원하는 SQLAlchemy 2.0에서 완전한 데이터베이스 스키마를 구축해 보겠습니다. 디자인은 높은 응집성과 낮은 결합의 원칙을 따르며 각각에 대해 별도의 테이블이 있습니다. 지배력과 명확한 관계의 개념.
# models/base.py
from sqlalchemy import Column, String, DateTime, func
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import DeclarativeBase
import uuid
class Base(DeclarativeBase):
pass
class TimestampMixin:
"""Mixin per created_at e updated_at automatici."""
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())# models/organization.py
from sqlalchemy import Column, String, Integer, Boolean, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import uuid
from .base import Base, TimestampMixin
class Organization(Base, TimestampMixin):
"""Azienda che produce il report ESG."""
__tablename__ = "organizations"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
name = Column(String(200), nullable=False)
legal_name = Column(String(200))
tax_id = Column(String(50)) # Partita IVA o VAT number
country = Column(String(2), nullable=False) # ISO 3166-1 alpha-2
industry_code = Column(String(10)) # NACE Rev.2
employee_count = Column(Integer)
fiscal_year_end = Column(String(5), default="12-31") # MM-DD
reporting_currency = Column(String(3), default="EUR")
boundary_approach = Column(String(20), default="OPERATIONAL_CONTROL")
# OPERATIONAL_CONTROL | FINANCIAL_CONTROL | EQUITY_SHARE
active = Column(Boolean, default=True)
facilities = relationship("Facility", back_populates="organization")
reporting_periods = relationship("ReportingPeriod", back_populates="organization")
class Facility(Base, TimestampMixin):
"""Sede fisica (ufficio, data center, magazzino)."""
__tablename__ = "facilities"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
organization_id = Column(UUID(as_uuid=True), ForeignKey("organizations.id"), nullable=False)
name = Column(String(200), nullable=False)
facility_type = Column(String(30))
# OFFICE | DATA_CENTER | WAREHOUSE | LAB | REMOTE
address = Column(String(500))
city = Column(String(100))
country = Column(String(2), nullable=False)
grid_region = Column(String(30)) # Per fattore emissione location-based
area_sqm = Column(Integer) # Per calcoli per m2
employee_fte = Column(Integer) # Full-time equivalent
organization = relationship("Organization", back_populates="facilities")
emission_sources = relationship("EmissionSource", back_populates="facility")# models/emission_source.py
from sqlalchemy import Column, String, Enum, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import enum, uuid
from .base import Base, TimestampMixin
class SourceType(str, enum.Enum):
# Scope 1
STATIONARY_COMBUSTION = "STATIONARY_COMBUSTION"
MOBILE_COMBUSTION = "MOBILE_COMBUSTION"
FUGITIVE = "FUGITIVE"
PROCESS = "PROCESS"
# Scope 2
ELECTRICITY = "ELECTRICITY"
STEAM = "STEAM"
HEATING = "HEATING"
COOLING = "COOLING"
# Scope 3 (le 15 categorie)
S3_PURCHASED_GOODS = "S3_PURCHASED_GOODS" # Cat. 1
S3_CAPITAL_GOODS = "S3_CAPITAL_GOODS" # Cat. 2
S3_FUEL_ENERGY = "S3_FUEL_ENERGY" # Cat. 3
S3_UPSTREAM_TRANSPORT = "S3_UPSTREAM_TRANSPORT" # Cat. 4
S3_WASTE = "S3_WASTE" # Cat. 5
S3_BUSINESS_TRAVEL = "S3_BUSINESS_TRAVEL" # Cat. 6
S3_EMPLOYEE_COMMUTE = "S3_EMPLOYEE_COMMUTE" # Cat. 7
S3_UPSTREAM_LEASED = "S3_UPSTREAM_LEASED" # Cat. 8
S3_DOWNSTREAM_TRANSPORT = "S3_DOWNSTREAM_TRANSPORT" # Cat. 9
S3_USE_OF_PRODUCTS = "S3_USE_OF_PRODUCTS" # Cat. 11
S3_END_OF_LIFE = "S3_END_OF_LIFE" # Cat. 12
S3_INVESTMENTS = "S3_INVESTMENTS" # Cat. 15
class EmissionSource(Base, TimestampMixin):
"""Sorgente specifica di emissione (es. 'Caldaia edificio A')."""
__tablename__ = "emission_sources"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
facility_id = Column(UUID(as_uuid=True), ForeignKey("facilities.id"), nullable=True)
organization_id = Column(UUID(as_uuid=True), ForeignKey("organizations.id"), nullable=False)
source_type = Column(String(40), nullable=False) # SourceType enum value
scope = Column(Integer, nullable=False) # 1, 2, o 3
scope3_category = Column(Integer, nullable=True) # 1-15 se scope=3
description = Column(String(500))
unit_of_measure = Column(String(20)) # 'kWh', 'litri', 'km', 'EUR', 'kg'
active = Column(Boolean, default=True)
facility = relationship("Facility", back_populates="emission_sources")
activity_records = relationship("ActivityRecord", back_populates="source")# models/emission_factor.py
from sqlalchemy import Column, String, Numeric, Integer, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import uuid
from .base import Base, TimestampMixin
class EmissionFactor(Base, TimestampMixin):
"""
Fattore di emissione da fonti ufficiali.
Converte un'attività (es. kWh) in kg CO2e.
"""
__tablename__ = "emission_factors"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
name = Column(String(200), nullable=False)
category = Column(String(50)) # 'electricity' | 'fuel' | 'transport' | ...
fuel_type = Column(String(50)) # 'natural_gas' | 'diesel' | 'petrol' | ...
technology = Column(String(50)) # 'gas_boiler' | 'bev' | ...
# Unità
activity_unit = Column(String(20), nullable=False) # 'kWh', 'litre', 'km', ...
emission_unit = Column(String(20), default="kg_co2e") # Output sempre in kg CO2e
# Valori per singolo gas (kg/unità)
co2_factor = Column(Numeric(18, 8))
ch4_factor = Column(Numeric(18, 8))
n2o_factor = Column(Numeric(18, 8))
hfc_factor = Column(Numeric(18, 8))
# GWP (AR5 100-year, IPCC 2014)
gwp_co2 = Column(Numeric(6, 2), default=1.0)
gwp_ch4 = Column(Numeric(6, 2), default=28.0)
gwp_n2o = Column(Numeric(6, 2), default=265.0)
# Metadati fonte
source = Column(String(100)) # 'DEFRA_2024' | 'IEA_2024' | 'IPCC_AR6'
year = Column(Integer)
region = Column(String(10)) # 'IT' | 'EU' | 'UK' | 'GLOBAL'
version = Column(String(20))
activity_records = relationship("ActivityRecord", back_populates="emission_factor")# models/activity_record.py
from sqlalchemy import Column, String, Numeric, Date, ForeignKey, Text
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import uuid
from .base import Base, TimestampMixin
class ActivityRecord(Base, TimestampMixin):
"""
Dato di attività misurato o stimato per una fonte di emissione.
Es: 'Ufficio Milano ha consumato 15.432 kWh a novembre 2024'.
"""
__tablename__ = "activity_records"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
source_id = Column(UUID(as_uuid=True), ForeignKey("emission_sources.id"), nullable=False)
factor_id = Column(UUID(as_uuid=True), ForeignKey("emission_factors.id"), nullable=False)
reporting_period_id = Column(UUID(as_uuid=True), ForeignKey("reporting_periods.id"))
# Periodo
period_start = Column(Date, nullable=False)
period_end = Column(Date, nullable=False)
# Quantità
quantity = Column(Numeric(18, 4), nullable=False)
unit = Column(String(20), nullable=False)
# Qualità del dato
data_quality = Column(String(20), default="MEASURED")
# MEASURED | ESTIMATED | CALCULATED | MODELED
uncertainty_pct = Column(Numeric(5, 2)) # Es: 10.00 = +-10%
evidence_url = Column(Text) # Link a fattura, report, foglio elettrico
# Note
notes = Column(Text)
reviewer = Column(String(100))
source = relationship("EmissionSource", back_populates="activity_records")
emission_factor = relationship("EmissionFactor", back_populates="activity_records")
calculation = relationship("GHGCalculation", back_populates="activity_record", uselist=False)
class GHGCalculation(Base, TimestampMixin):
"""
Risultato del calcolo GHG per un ActivityRecord.
Separato per permettere ricalcoli senza perdere il dato sorgente.
"""
__tablename__ = "ghg_calculations"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
activity_record_id = Column(UUID(as_uuid=True),
ForeignKey("activity_records.id"), nullable=False, unique=True)
# Emissioni per gas (kg)
co2_kg = Column(Numeric(18, 4), default=0)
ch4_kg = Column(Numeric(18, 4), default=0)
n2o_kg = Column(Numeric(18, 4), default=0)
hfc_kg = Column(Numeric(18, 4), default=0)
# Totale in CO2e (kg e tonnellate)
co2e_kg = Column(Numeric(18, 4), nullable=False)
co2e_tonnes = Column(Numeric(18, 6), nullable=False)
# Metodo di calcolo
calculation_method = Column(String(50))
# ACTIVITY_BASED | SPEND_BASED | AVERAGE_DATA | HYBRID
# Scope 2 specifico
is_location_based = Column(Boolean)
is_market_based = Column(Boolean)
calculated_at = Column(DateTime(timezone=True), server_default=func.now())
calculator_version = Column(String(20), default="1.0.0")
activity_record = relationship("ActivityRecord", back_populates="calculation")Python의 GHG 계산: 활동 기반 및 지출 기반
데이터 모델이 정의되면 계산을 구현할 수 있습니다. GHG 프로토콜은 다음과 같은 사실을 인정합니다. 정확도가 높은 순서대로 4가지 계산 방법:
- 공급업체별 방법: 공급업체 직접 데이터(가장 정확함)
- 하이브리드 방식: 1차 데이터와 2차 데이터의 결합
- 평균 데이터 방법: 물리적 단위당 평균 배출 계수
- 지출 기반 방법: EUR/USD 단위 지출 단위당 요소(정확도 낮음)
# calculators/ghg_calculator.py
from decimal import Decimal
from dataclasses import dataclass
from typing import Optional
from models import ActivityRecord, EmissionFactor, GHGCalculation
@dataclass
class CalculationResult:
co2_kg: Decimal
ch4_kg: Decimal
n2o_kg: Decimal
hfc_kg: Decimal
co2e_kg: Decimal
co2e_tonnes: Decimal
method: str
def activity_based_calculation(
quantity: Decimal,
factor: EmissionFactor
) -> CalculationResult:
"""
Calcolo activity-based: quantità * fattore emissione.
Formula GHG Protocol: E = AD * EF * GWP
"""
co2_kg = Decimal(str(quantity)) * Decimal(str(factor.co2_factor or 0))
ch4_kg = Decimal(str(quantity)) * Decimal(str(factor.ch4_factor or 0))
n2o_kg = Decimal(str(quantity)) * Decimal(str(factor.n2o_factor or 0))
hfc_kg = Decimal(str(quantity)) * Decimal(str(factor.hfc_factor or 0))
# Converti in CO2e usando GWP AR5
co2e_kg = (
co2_kg * Decimal(str(factor.gwp_co2))
+ ch4_kg * Decimal(str(factor.gwp_ch4))
+ n2o_kg * Decimal(str(factor.gwp_n2o))
+ hfc_kg * Decimal("1300") # GWP medio HFC-134a
)
return CalculationResult(
co2_kg=co2_kg,
ch4_kg=ch4_kg,
n2o_kg=n2o_kg,
hfc_kg=hfc_kg,
co2e_kg=co2e_kg,
co2e_tonnes=co2e_kg / Decimal("1000"),
method="ACTIVITY_BASED"
)
def spend_based_calculation(
spend_eur: Decimal,
spend_factor_kg_per_eur: Decimal
) -> CalculationResult:
"""
Calcolo spend-based per Cat. 1 Scope 3 quando non si hanno
dati fisici. Usa EXIOBASE o fattori DEFRA per settore NACE.
Meno accurato ma utile come stima iniziale.
"""
co2e_kg = spend_eur * spend_factor_kg_per_eur
return CalculationResult(
co2_kg=co2e_kg, # Semplificazione: tutto attribuito a CO2
ch4_kg=Decimal("0"),
n2o_kg=Decimal("0"),
hfc_kg=Decimal("0"),
co2e_kg=co2e_kg,
co2e_tonnes=co2e_kg / Decimal("1000"),
method="SPEND_BASED"
)
def business_travel_calculation(
distance_km: Decimal,
transport_mode: str,
cabin_class: Optional[str] = None
) -> CalculationResult:
"""
Calcolo Cat. 6 (Business Travel) con fattori DEFRA 2024.
Fattori in kgCO2e per km per passeggero.
"""
# Fattori DEFRA 2024 (kgCO2e/km/passeggero)
FACTORS = {
"SHORT_HAUL_ECONOMY": Decimal("0.15342"), # Volo <1000km economy
"SHORT_HAUL_BUSINESS": Decimal("0.23013"),
"LONG_HAUL_ECONOMY": Decimal("0.19085"), # Volo >1000km economy
"LONG_HAUL_BUSINESS": Decimal("0.57256"), # Business = 3x economy
"LONG_HAUL_FIRST": Decimal("0.76340"),
"RAIL_NATIONAL": Decimal("0.03549"), # Treno nazionale UK
"RAIL_EUROSTAR": Decimal("0.00416"), # Eurostar molto efficiente
"CAR_AVERAGE": Decimal("0.17100"), # Auto media
"CAR_ELECTRIC": Decimal("0.05280"),
"HOTEL_NIGHT": Decimal("20.600"), # Per notte, non per km
}
key = transport_mode
if cabin_class and f"{transport_mode}_{cabin_class}" in FACTORS:
key = f"{transport_mode}_{cabin_class}"
factor = FACTORS.get(key, Decimal("0"))
co2e_kg = distance_km * factor
return CalculationResult(
co2_kg=co2e_kg,
ch4_kg=Decimal("0"),
n2o_kg=Decimal("0"),
hfc_kg=Decimal("0"),
co2e_kg=co2e_kg,
co2e_tonnes=co2e_kg / Decimal("1000"),
method="ACTIVITY_BASED"
)# calculators/scope2_calculator.py
from decimal import Decimal
from dataclasses import dataclass
from typing import Optional
@dataclass
class Scope2Result:
location_based_co2e_kg: Decimal
market_based_co2e_kg: Decimal
renewable_kwh_matched: Decimal
grid_ef_g_kwh: Decimal
market_ef_g_kwh: Decimal
def scope2_dual_calculation(
kwh_consumed: Decimal,
grid_region: str,
go_certificates_kwh: Decimal = Decimal("0"),
supplier_ef_g_kwh: Optional[Decimal] = None
) -> Scope2Result:
"""
Calcolo Scope 2 con doppio metodo obbligatorio (GHG Protocol S2 Guidance).
location_based: usa grid average emission factor (IEA/AIB)
market_based: usa supplier factor o residual mix, dedotti GO/REC
"""
# Grid emission factors per regione (gCO2e/kWh) - IEA 2024
GRID_FACTORS = {
"IT": Decimal("310.0"), # Italia: mix gas + rinnovabili
"DE": Decimal("364.0"), # Germania: ancora carbone
"FR": Decimal("52.0"), # Francia: nucleare dominante
"ES": Decimal("198.0"),
"NL": Decimal("289.0"),
"EU_AVERAGE": Decimal("231.0"),
"UK": Decimal("209.0"),
"US_AVERAGE": Decimal("386.0"),
}
grid_ef = GRID_FACTORS.get(grid_region, GRID_FACTORS["EU_AVERAGE"])
# Location-based (sempre con grid average)
lb_co2e_kg = kwh_consumed * grid_ef / Decimal("1000")
# Market-based:
# 1. Kwh coperti da GO/REC = 0 emissioni
# 2. Kwh residui: usa supplier factor o residual mix
covered_by_go = min(go_certificates_kwh, kwh_consumed)
residual_kwh = kwh_consumed - covered_by_go
# Residual mix factor (tipicamente > grid average perché
# le energie rinnovabili sono "estratte" dal grid average)
# Fonte: AIB European Residual Mixes 2023
RESIDUAL_MIX = {
"IT": Decimal("414.0"), # Residual mix IT più alto del grid
"DE": Decimal("542.0"),
"FR": Decimal("73.0"),
"EU_AVERAGE": Decimal("395.0"),
}
market_ef = supplier_ef_g_kwh or RESIDUAL_MIX.get(grid_region, Decimal("400"))
mb_co2e_kg = residual_kwh * market_ef / Decimal("1000")
return Scope2Result(
location_based_co2e_kg=lb_co2e_kg,
market_based_co2e_kg=mb_co2e_kg,
renewable_kwh_matched=covered_by_go,
grid_ef_g_kwh=grid_ef,
market_ef_g_kwh=market_ef
)# calculators/scope3_commute.py
from decimal import Decimal
from dataclasses import dataclass
from typing import Dict
@dataclass
class CommuteData:
"""Dati raccolti tramite survey annuale dipendenti."""
employee_count: int
avg_working_days: int # Es: 220
remote_work_days_pct: Decimal # Es: 0.40 = 40% remote
transport_split: Dict[str, Decimal]
# {'car_solo': 0.45, 'car_pool': 0.10, 'public_transit': 0.30,
# 'bicycle': 0.10, 'walk': 0.05}
avg_commute_km_oneway: Decimal # km sola andata
def employee_commute_calculation(data: CommuteData) -> Decimal:
"""
Cat. 7 Scope 3: Pendolarismo dipendenti.
Formula: n_dipendenti * giorni_ufficio * km_totali * EF * split_modale
"""
# Fattori emissione (kgCO2e/km/passeggero) - DEFRA 2024
MODE_FACTORS = {
"car_solo": Decimal("0.171"), # Auto singola
"car_pool": Decimal("0.0855"), # Carpooling 2 persone
"public_transit": Decimal("0.089"), # Mix metropolitana/bus
"bicycle": Decimal("0.0"),
"walk": Decimal("0.0"),
"electric_car": Decimal("0.053"),
"motorcycle": Decimal("0.114"),
}
# Giorni effettivi in ufficio
office_days_pct = Decimal("1") - data.remote_work_days_pct
annual_office_days = Decimal(str(data.avg_working_days)) * office_days_pct
km_per_day = data.avg_commute_km_oneway * Decimal("2") # A/R
total_co2e_kg = Decimal("0")
for mode, share in data.transport_split.items():
ef = MODE_FACTORS.get(mode, Decimal("0.1"))
mode_co2e = (
Decimal(str(data.employee_count))
* annual_office_days
* km_per_day
* share
* ef
)
total_co2e_kg += mode_co2e
return total_co2e_kg연간 집계 및 전년 대비 비교
각 ActivityRecord에 대해 배출량을 계산한 후에는 기간별로 집계해야 합니다. 보고하고 전년도와 비교하여 SBTi 목표와의 일치 여부를 확인합니다.
# services/reporting_service.py
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from decimal import Decimal
from typing import Dict
from models import GHGCalculation, ActivityRecord, EmissionSource, ReportingPeriod
@dataclass
class AnnualReport:
organization_id: str
year: int
scope1_co2e_tonnes: Decimal
scope2_lb_co2e_tonnes: Decimal # Location-based
scope2_mb_co2e_tonnes: Decimal # Market-based
scope3_co2e_tonnes: Decimal
scope3_by_category: Dict[int, Decimal]
total_co2e_tonnes: Decimal # S1 + S2(MB) + S3
intensity_per_employee: Decimal # tCO2e / FTE
intensity_per_revenue: Decimal # tCO2e / M EUR
def generate_annual_report(
session: Session,
organization_id: str,
year: int,
employee_fte: int,
revenue_million_eur: Decimal
) -> AnnualReport:
"""Aggrega tutti i calcoli GHG per anno fiscale."""
# Query aggregata per scope
stmt = (
select(
EmissionSource.scope,
EmissionSource.scope3_category,
GHGCalculation.is_location_based,
func.sum(GHGCalculation.co2e_tonnes).label("total_tonnes")
)
.join(ActivityRecord, GHGCalculation.activity_record_id == ActivityRecord.id)
.join(EmissionSource, ActivityRecord.source_id == EmissionSource.id)
.where(
EmissionSource.organization_id == organization_id,
func.extract("year", ActivityRecord.period_start) == year
)
.group_by(
EmissionSource.scope,
EmissionSource.scope3_category,
GHGCalculation.is_location_based
)
)
rows = session.execute(stmt).all()
scope1 = Decimal("0")
scope2_lb = Decimal("0")
scope2_mb = Decimal("0")
scope3_by_cat: Dict[int, Decimal] = {}
for row in rows:
tonnes = Decimal(str(row.total_tonnes or 0))
if row.scope == 1:
scope1 += tonnes
elif row.scope == 2:
if row.is_location_based:
scope2_lb += tonnes
else:
scope2_mb += tonnes
elif row.scope == 3:
cat = row.scope3_category or 0
scope3_by_cat[cat] = scope3_by_cat.get(cat, Decimal("0")) + tonnes
scope3_total = sum(scope3_by_cat.values())
total = scope1 + scope2_mb + scope3_total
return AnnualReport(
organization_id=organization_id,
year=year,
scope1_co2e_tonnes=scope1,
scope2_lb_co2e_tonnes=scope2_lb,
scope2_mb_co2e_tonnes=scope2_mb,
scope3_co2e_tonnes=scope3_total,
scope3_by_category=scope3_by_cat,
total_co2e_tonnes=total,
intensity_per_employee=total / Decimal(str(employee_fte)) if employee_fte else Decimal("0"),
intensity_per_revenue=total / revenue_million_eur if revenue_million_eur else Decimal("0")
)
def check_sbti_alignment(
report: AnnualReport,
baseline_year: int,
baseline_emissions: Decimal,
target_year: int = 2030,
reduction_target_pct: Decimal = Decimal("0.50") # SBTi: -50% by 2030
) -> Dict:
"""
Verifica allineamento con target SBTi.
Near-term: -50% entro 2030 (1.5°C pathway).
Net-zero: -90-95% entro 2050.
"""
current_year_progress = report.year - baseline_year
total_years = target_year - baseline_year
linear_target = baseline_emissions * (
1 - reduction_target_pct * Decimal(str(current_year_progress)) / Decimal(str(total_years))
)
actual_reduction_pct = (baseline_emissions - report.total_co2e_tonnes) / baseline_emissions * 100
return {
"baseline_emissions_tonnes": float(baseline_emissions),
"current_emissions_tonnes": float(report.total_co2e_tonnes),
"linear_target_tonnes": float(linear_target),
"actual_reduction_pct": float(actual_reduction_pct),
"on_track": report.total_co2e_tonnes <= linear_target,
"gap_tonnes": float(report.total_co2e_tonnes - linear_target),
"required_annual_reduction_pct": float(
reduction_target_pct / Decimal(str(total_years)) * 100
)
}ESG용 REST API: FastAPI 엔드포인트
엔터프라이즈 ESG 시스템에는 다양한 팀이 전송할 수 있는 강력한 API가 필요합니다. 활동 데이터, 트리거 계산 및 액세스 보고서. FastAPI가 선택입니다 유형 안전, 자동 문서화 및 성능에 이상적입니다.
# api/routers/activities.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from pydantic import BaseModel, validator
from typing import Optional
from datetime import date
from decimal import Decimal
import uuid
router = APIRouter(prefix="/api/v1", tags=["activities"])
class ActivitySubmitRequest(BaseModel):
source_id: uuid.UUID
period_start: date
period_end: date
quantity: Decimal
unit: str
data_quality: str = "MEASURED"
evidence_url: Optional[str] = None
notes: Optional[str] = None
@validator("data_quality")
def validate_quality(cls, v: str) -> str:
allowed = ["MEASURED", "ESTIMATED", "CALCULATED", "MODELED"]
if v not in allowed:
raise ValueError(f"data_quality deve essere uno di: {allowed}")
return v
@validator("period_end")
def validate_dates(cls, v: date, values: dict) -> date:
if "period_start" in values and v < values["period_start"]:
raise ValueError("period_end deve essere >= period_start")
return v
class ActivitySubmitResponse(BaseModel):
activity_id: uuid.UUID
co2e_tonnes: Decimal
calculation_method: str
message: str
@router.post("/activities", response_model=ActivitySubmitResponse, status_code=201)
async def submit_activity(
req: ActivitySubmitRequest,
session: Session = Depends(get_db),
current_org: Organization = Depends(get_current_org)
) -> ActivitySubmitResponse:
"""
Invia un dato di attività e calcola le emissioni GHG.
Validazione automatica: unità coerente con la sorgente.
"""
source = session.get(EmissionSource, req.source_id)
if not source or source.organization_id != current_org.id:
raise HTTPException(status_code=404, detail="Sorgente emissione non trovata")
# Trova il fattore di emissione più recente per questa sorgente
factor = get_best_emission_factor(session, source, req.unit, req.period_start.year)
if not factor:
raise HTTPException(
status_code=422,
detail=f"Nessun fattore emissione trovato per {source.source_type} in {req.unit}"
)
# Calcola emissioni
result = activity_based_calculation(req.quantity, factor)
# Salva ActivityRecord e GHGCalculation
record = ActivityRecord(
source_id=req.source_id,
factor_id=factor.id,
period_start=req.period_start,
period_end=req.period_end,
quantity=req.quantity,
unit=req.unit,
data_quality=req.data_quality,
evidence_url=req.evidence_url,
notes=req.notes
)
session.add(record)
session.flush() # Get the ID
calc = GHGCalculation(
activity_record_id=record.id,
co2_kg=result.co2_kg,
ch4_kg=result.ch4_kg,
n2o_kg=result.n2o_kg,
co2e_kg=result.co2e_kg,
co2e_tonnes=result.co2e_tonnes,
calculation_method=result.method
)
session.add(calc)
session.commit()
return ActivitySubmitResponse(
activity_id=record.id,
co2e_tonnes=result.co2e_tonnes,
calculation_method=result.method,
message=f"Emissioni calcolate: {result.co2e_tonnes:.4f} tCO2e"
)# api/routers/reports.py
from fastapi import APIRouter, Query
from pydantic import BaseModel
from typing import Dict
from decimal import Decimal
router = APIRouter(prefix="/api/v1", tags=["reports"])
class ESGReportResponse(BaseModel):
organization_id: str
year: int
scope1_co2e_tonnes: float
scope2_location_based_tonnes: float
scope2_market_based_tonnes: float
scope3_co2e_tonnes: float
scope3_breakdown: Dict[int, float]
total_co2e_tonnes: float
intensity_per_employee: float
intensity_per_revenue_m_eur: float
sbti_alignment: Dict
class YoYComparisonResponse(BaseModel):
year_current: int
year_previous: int
total_change_pct: float
scope1_change_pct: float
scope2_change_pct: float
scope3_change_pct: float
top_increasing_categories: list
top_decreasing_categories: list
@router.get("/reports/annual/{year}", response_model=ESGReportResponse)
async def get_annual_report(
year: int,
session: Session = Depends(get_db),
current_org: Organization = Depends(get_current_org)
) -> ESGReportResponse:
"""
Genera report ESG annuale con tutte le metriche GHG Protocol.
Include verifica allineamento SBTi.
"""
employee_fte = get_employee_fte(session, current_org.id, year)
revenue = get_annual_revenue(session, current_org.id, year)
report = generate_annual_report(
session, str(current_org.id), year, employee_fte, revenue
)
baseline = get_baseline_emissions(session, str(current_org.id))
sbti = check_sbti_alignment(report, baseline["year"], baseline["tonnes"])
return ESGReportResponse(
organization_id=str(current_org.id),
year=year,
scope1_co2e_tonnes=float(report.scope1_co2e_tonnes),
scope2_location_based_tonnes=float(report.scope2_lb_co2e_tonnes),
scope2_market_based_tonnes=float(report.scope2_mb_co2e_tonnes),
scope3_co2e_tonnes=float(report.scope3_co2e_tonnes),
scope3_breakdown={k: float(v) for k, v in report.scope3_by_category.items()},
total_co2e_tonnes=float(report.total_co2e_tonnes),
intensity_per_employee=float(report.intensity_per_employee),
intensity_per_revenue_m_eur=float(report.intensity_per_revenue),
sbti_alignment=sbti
)
@router.get("/reports/yoy/{year}", response_model=YoYComparisonResponse)
async def year_over_year(year: int, ...) -> YoYComparisonResponse:
"""Confronto Year-over-Year con identificazione trend per categoria."""
curr = generate_annual_report(session, org_id, year, fte, rev)
prev = generate_annual_report(session, org_id, year - 1, fte_prev, rev_prev)
def pct_change(curr_val: Decimal, prev_val: Decimal) -> float:
if prev_val == 0:
return 0.0
return float((curr_val - prev_val) / prev_val * 100)
return YoYComparisonResponse(
year_current=year,
year_previous=year - 1,
total_change_pct=pct_change(curr.total_co2e_tonnes, prev.total_co2e_tonnes),
scope1_change_pct=pct_change(curr.scope1_co2e_tonnes, prev.scope1_co2e_tonnes),
scope2_change_pct=pct_change(curr.scope2_mb_co2e_tonnes, prev.scope2_mb_co2e_tonnes),
scope3_change_pct=pct_change(curr.scope3_co2e_tonnes, prev.scope3_co2e_tonnes),
top_increasing_categories=get_top_changes(curr, prev, ascending=False),
top_decreasing_categories=get_top_changes(curr, prev, ascending=True)
)테스트: GHG 계산을 위한 단위 테스트
GHG 계산은 엄격하게 테스트되어야 합니다. 배출 계수의 오류 또는 공식은 잠재적인 법적 영향(그린워싱)과 함께 잘못된 ESG 보고로 이어집니다. CSRD에 따른 제재. 테스트는 선택 사항이 아닙니다.
# tests/test_ghg_calculator.py
import pytest
from decimal import Decimal
from calculators.ghg_calculator import (
activity_based_calculation,
spend_based_calculation,
business_travel_calculation
)
from calculators.scope2_calculator import scope2_dual_calculation
from calculators.scope3_commute import employee_commute_calculation, CommuteData
from models import EmissionFactor
@pytest.fixture
def natural_gas_factor() -> EmissionFactor:
"""Fattore gas naturale DEFRA 2024: 2.04264 kgCO2e/m3."""
factor = EmissionFactor()
factor.co2_factor = Decimal("1.88900") # kgCO2/m3
factor.ch4_factor = Decimal("0.00011") # kgCH4/m3
factor.n2o_factor = Decimal("0.00003") # kgN2O/m3
factor.hfc_factor = Decimal("0")
factor.gwp_co2 = Decimal("1.0")
factor.gwp_ch4 = Decimal("28.0")
factor.gwp_n2o = Decimal("265.0")
return factor
class TestActivityBasedCalculation:
def test_natural_gas_calculation(self, natural_gas_factor):
"""Test: 1000 m3 gas naturale = ~2.04 tCO2e."""
result = activity_based_calculation(Decimal("1000"), natural_gas_factor)
assert result.method == "ACTIVITY_BASED"
# CO2: 1000 * 1.889 = 1889 kg
assert abs(result.co2_kg - Decimal("1889.00")) < Decimal("0.01")
# CH4: 1000 * 0.00011 * 28 = 3.08 kgCO2e
expected_co2e = (
Decimal("1889.00") * Decimal("1.0") # CO2
+ Decimal("0.11") * Decimal("28.0") # CH4
+ Decimal("0.03") * Decimal("265.0") # N2O
)
assert abs(result.co2e_kg - expected_co2e) < Decimal("1.0")
assert result.co2e_tonnes == result.co2e_kg / Decimal("1000")
def test_zero_consumption(self, natural_gas_factor):
"""Zero consumo = zero emissioni."""
result = activity_based_calculation(Decimal("0"), natural_gas_factor)
assert result.co2e_kg == Decimal("0")
def test_precision_decimal(self, natural_gas_factor):
"""Verifica che i calcoli usino Decimal per evitare floating point errors."""
result = activity_based_calculation(Decimal("333.333"), natural_gas_factor)
# Non deve avere floating point drift
assert isinstance(result.co2e_kg, Decimal)
class TestScope2DualMethod:
def test_italy_location_based(self):
"""Test location-based Italia: 10.000 kWh * 310 gCO2e/kWh = 3100 kg."""
result = scope2_dual_calculation(
kwh_consumed=Decimal("10000"),
grid_region="IT"
)
assert abs(result.location_based_co2e_kg - Decimal("3100.0")) < Decimal("1.0")
def test_full_go_coverage_zeroes_market_based(self):
"""Con GO = consumo totale, market-based deve essere 0."""
result = scope2_dual_calculation(
kwh_consumed=Decimal("10000"),
grid_region="IT",
go_certificates_kwh=Decimal("10000")
)
assert result.market_based_co2e_kg == Decimal("0")
assert result.renewable_kwh_matched == Decimal("10000")
def test_partial_go_coverage(self):
"""Con GO parziali, market-based usa residual mix sul rimanente."""
result = scope2_dual_calculation(
kwh_consumed=Decimal("10000"),
grid_region="IT",
go_certificates_kwh=Decimal("6000")
)
# 4000 kWh residui * 414 gCO2e/kWh (residual mix IT) / 1000
expected = Decimal("4000") * Decimal("414") / Decimal("1000")
assert abs(result.market_based_co2e_kg - expected) < Decimal("0.01")
class TestEmployeeCommute:
def test_standard_commute(self):
"""Test pendolarismo con mix modale realistico."""
data = CommuteData(
employee_count=200,
avg_working_days=220,
remote_work_days_pct=Decimal("0.40"),
transport_split={
"car_solo": Decimal("0.45"),
"public_transit": Decimal("0.40"),
"bicycle": Decimal("0.10"),
"walk": Decimal("0.05")
},
avg_commute_km_oneway=Decimal("15")
)
result = employee_commute_calculation(data)
assert result > Decimal("0")
# Ordine di grandezza: ~50-150 tCO2e per 200 dipendenti
assert Decimal("40000") < result < Decimal("200000") # in kg
def test_full_remote_work_zero(self):
"""100% remote work = 0 emissioni pendolarismo."""
data = CommuteData(
employee_count=200,
avg_working_days=220,
remote_work_days_pct=Decimal("1.0"),
transport_split={"car_solo": Decimal("1.0")},
avg_commute_km_oneway=Decimal("20")
)
result = employee_commute_calculation(data)
assert result == Decimal("0")주의: 문제 및 버전 관리 요소
배출계수는 매년 업데이트됩니다(DEFRA, IEA, EPA). ESG 시스템 정확해야 합니다:
- 명시적인 연도 및 소스를 사용하여 데이터베이스의 요소 버전을 지정합니다.
- 보고 연도 요소 사용(가장 최근 날짜 아님)
- 요인이 수정되면 과거 재계산을 허용합니다.
- 감사 추적 유지: 계산 당시 활성화된 요소
- 내부 GHGCalculation을 업데이트하지 마십시오. 항상 새 버전을 생성하십시오.
사례 연구: 직원이 200명인 중소기업 소프트웨어
구체적인 엔드투엔드 예시를 살펴보겠습니다. TechFlow S.r.l. 중소기업입니다 직원이 200명인 이탈리아 소프트웨어, 밀라노에 본사(임대), 보조 사무실 로마에는 자체 데이터 센터가 없습니다(모두 AWS eu-west-1 및 Azure westeurope에 있음).
회사 프로필 TechFlow S.r.l.
| 매개변수 | Valore |
|---|---|
| 직원 | FTE 200명(밀라노 70%, 로마 20%, 완전 원격 10%) |
| 매상 | 1,500만 유로 |
| 부엌 | 2개소 임대(총 2,400m²) |
| IT 인프라 | 100% 클라우드(AWS + Azure), 물리적 서버 없음 |
| 함대 | 회사 디젤 차량 3대, 디젤 밴 1대 |
| 출장 | 연간 ~120편의 항공편, 40박 호텔, 200개 기차 노선 |
# Esempio calcolo completo TechFlow S.r.l. - Anno 2024
from decimal import Decimal
from calculators.ghg_calculator import activity_based_calculation, business_travel_calculation
from calculators.scope2_calculator import scope2_dual_calculation
from calculators.scope3_commute import employee_commute_calculation, CommuteData
# ---- SCOPE 1 ----
# Gas naturale: riscaldamento uffici (caldaie condominiali, incluse nel canone)
# TechFlow ha controllo operativo: include le caldaie
scope1_gas_heating = activity_based_calculation(
Decimal("12500"), # m3 gas naturale annui (entrambe le sedi)
NATURAL_GAS_FACTOR_IT_2024
)
# Risultato atteso: ~25.5 tCO2e
# Flotta aziendale: 4 veicoli diesel
scope1_fleet_km = activity_based_calculation(
Decimal("85000"), # km totali flotta annui
DIESEL_MOBILE_FACTOR_2024 # 0.16844 kgCO2e/km
)
# Risultato atteso: ~14.3 tCO2e
# Refrigeranti HVAC: perdita stimata 2% del carico R-410A
scope1_refrigerants = activity_based_calculation(
Decimal("1.5"), # kg R-410A persi
R410A_FACTOR # GWP=2088 kgCO2e/kg
)
# Risultato atteso: ~3.1 tCO2e
scope1_total = (
scope1_gas_heating.co2e_tonnes
+ scope1_fleet_km.co2e_tonnes
+ scope1_refrigerants.co2e_tonnes
)
print(f"Scope 1 totale: {scope1_total:.2f} tCO2e")
# Output: Scope 1 totale: 42.90 tCO2e
# ---- SCOPE 2 ----
# Elettricità Milano: 320.000 kWh, GO acquistate: 200.000 kWh
scope2_milano = scope2_dual_calculation(
kwh_consumed=Decimal("320000"),
grid_region="IT",
go_certificates_kwh=Decimal("200000")
)
# Elettricità Roma: 95.000 kWh, nessuna GO
scope2_roma = scope2_dual_calculation(
kwh_consumed=Decimal("95000"),
grid_region="IT",
go_certificates_kwh=Decimal("0")
)
scope2_lb_total = scope2_milano.location_based_co2e_kg + scope2_roma.location_based_co2e_kg
scope2_mb_total = scope2_milano.market_based_co2e_kg + scope2_roma.market_based_co2e_kg
print(f"Scope 2 Location-Based: {scope2_lb_total/1000:.2f} tCO2e")
# Output: Scope 2 Location-Based: 128.45 tCO2e
print(f"Scope 2 Market-Based: {scope2_mb_total/1000:.2f} tCO2e")
# Output: Scope 2 Market-Based: 55.12 tCO2e (ridotto grazie alle GO)
# ---- SCOPE 3 ----
# Cat. 1: Cloud computing (AWS+Azure) - spend-based
# Spesa totale cloud: EUR 480.000/anno
# Fattore EXIOBASE settore "Computer Services" IT: ~0.062 kgCO2e/EUR
scope3_cat1_cloud = spend_based_calculation(
spend_eur=Decimal("480000"),
spend_factor_kg_per_eur=Decimal("0.062")
)
# Risultato atteso: ~29.8 tCO2e
# NOTA: Provider-specific è più accurato se disponibile
# AWS Carbon Footprint: ~18 tCO2e (più basso per energia rinnovabile AWS)
# Cat. 2: Capital goods - laptop/monitor (life cycle amortized)
# 40 laptop nuovi a EUR 1.200 cad, ammortizzati su 4 anni
scope3_cat2_laptops = spend_based_calculation(
spend_eur=Decimal("40") * Decimal("1200") / Decimal("4"),
spend_factor_kg_per_eur=Decimal("0.35") # Electronic equipment
)
# Risultato atteso: ~4.2 tCO2e
# Cat. 6: Business travel
scope3_cat6_flights = business_travel_calculation(
distance_km=Decimal("95000"), # Totale km voli
transport_mode="SHORT_HAUL_ECONOMY" # Mix short/medium haul
)
scope3_cat6_rail = business_travel_calculation(
distance_km=Decimal("18000"), # Totale km treno
transport_mode="RAIL_NATIONAL"
)
scope3_cat6_hotel = business_travel_calculation(
distance_km=Decimal("40"), # 40 notti hotel
transport_mode="HOTEL_NIGHT"
)
scope3_cat6_total = (
scope3_cat6_flights.co2e_tonnes
+ scope3_cat6_rail.co2e_tonnes
+ scope3_cat6_hotel.co2e_tonnes
)
# Risultato atteso: ~16.3 tCO2e
# Cat. 7: Employee commuting
commute_data = CommuteData(
employee_count=180, # Esclude 20 full-remote
avg_working_days=220,
remote_work_days_pct=Decimal("0.35"), # Media 35% smart working
transport_split={
"car_solo": Decimal("0.38"),
"public_transit": Decimal("0.45"),
"bicycle": Decimal("0.08"),
"walk": Decimal("0.06"),
"car_pool": Decimal("0.03")
},
avg_commute_km_oneway=Decimal("12")
)
scope3_cat7_kg = employee_commute_calculation(commute_data)
# Risultato atteso: ~68.400 kg = 68.4 tCO2e
# ---- RIEPILOGO FINALE ----
scope3_total = (
scope3_cat1_cloud.co2e_tonnes
+ scope3_cat2_laptops.co2e_tonnes
+ scope3_cat6_total
+ scope3_cat7_kg / Decimal("1000")
# + Cat. 11 Use of sold products (SaaS): stimato ~15 tCO2e
# + Cat. 3 T&D losses: ~8.2 tCO2e
)
grand_total = scope1_total + scope2_mb_total / Decimal("1000") + scope3_total
print("\n=== TECHFLOW S.R.L. - GHG INVENTORY 2024 ===")
print(f"Scope 1: {float(scope1_total):8.1f} tCO2e ( {float(scope1_total/grand_total)*100:4.1f}%)")
print(f"Scope 2: {float(scope2_mb_total/1000):8.1f} tCO2e ( {float(scope2_mb_total/1000/grand_total)*100:4.1f}%)")
print(f"Scope 3: {float(scope3_total):8.1f} tCO2e ( {float(scope3_total/grand_total)*100:4.1f}%)")
print(f"TOTALE: {float(grand_total):8.1f} tCO2e")
print(f"Intensità: {float(grand_total/200):5.2f} tCO2e/dipendente")
# === TECHFLOW S.R.L. - GHG INVENTORY 2024 ===
# Scope 1: 42.9 tCO2e ( 18.2%)
# Scope 2: 55.1 tCO2e ( 23.4%)
# Scope 3: 137.8 tCO2e ( 58.4%)
# TOTALE: 235.8 tCO2e
# Intensità: 1.18 tCO2e/dipendente벤치마크: 소프트웨어 회사의 탄소 집약도(2024)
| 크기 | tCO2e/종속 | % 범위 3 | 메인 엔트리 S3 |
|---|---|---|---|
| 스타트업(직원 50명 미만) | 0.8 - 1.2 | 65-75% | 출장 |
| 소프트웨어 중소기업(50-500) | 1.0 - 1.8 | 55-70% | 통근+클라우드 |
| 엔터프라이즈 기술(500+) | 1.5 - 3.0 | 70-85% | 클라우드 + 제품 사용 |
| 빅테크(하이퍼스케일러) | 3.0 - 8.0 | 75-90% | 하드웨어 공급망 + 데이터 센터 |
| TechFlow(사례 연구) | 1.18 | 58% | 통근 |
TechFlow의 감소 기회
2024년 GHG 인벤토리 분석에서 가장 효과적인 감축 수단이 나타났습니다.
- 스마트 작업이 50%로 확장되었습니다. 고양이. 7 ~35% 감소(-24tCO2e). 통근은 단일 항목 중 가장 큰 항목이다.
- 두 위치 모두 100% GO: Scope 2 시장 기반이 ~0tCO2e로 감소 (-55tCO2e). GO 예상 비용: EUR 3,000-5,000/년.
- 여행 탄소 예산 정책: 단거리 비행(<3시간)을 기차로 대체합니다. 고양이. 6 ~40% 감소(-6.5tCO2e).
- AWS 탄소 배출량 도구: 지출 기반 견적에서 특정 제공업체로 전환 고양이를 줄입니다. 1 클라우드를 ~40%(-12 tCO2e) 줄이고 데이터 품질을 향상시킵니다.
TechFlow 탈탄소화 로드맵 — 목표 SBTi 1.5°C
| 년도 | 목표(tCO2e) | 2024년 대비 감소 | 주요 활동 |
|---|---|---|---|
| 2024년(기준) | 235.8 | - | 첫 번째 CSRD 보고 |
| 2025년 | 211.0 | -10.5% | GO 100%, 스마트워킹 50% |
| 2027년 | 165.0 | -30% | 전기 함대, 여행 정책 |
| 2030년 | 118.0 | -50% (SBTi 단기) | 친환경 원격 우선 클라우드 제공업체 |
| 2050년 | <24 | -90%(SBTi 순 제로) | 중화된 잔류물, 보상 없음 |
결론 및 다음 단계
ESG 보고를 위한 데이터 모델링은 단순한 규정 준수 활동이 아닙니다. 이는 신뢰할 수 있고 검증 가능한 탈탄소화 전략을 구축하는 기초입니다. 우리는 회사의 전체 가치 사슬을 포괄하는 시스템을 구성하는 방법을 살펴보았습니다. 난방 보일러(Scope 1)부터 상업용 비행기 여행(Scope 3 Cat. 6)까지의 소프트웨어 사용자가 제품을 작동하기 위해 소비하는 최대 에너지(Scope 3 Cat. 11).
가지고 가야 할 주요 사항:
- 범위 3이 지배적입니다. 소프트웨어 회사의 경우 배출량의 55~75%가 Scope 3입니다. 올바르게 모델링하지 않는다는 것은 탄소 배출량에 대한 이미지가 왜곡된다는 의미입니다.
- 이중 보고 범위 2: ESRS E1에는 두 가지 방법(위치 기반 및 시장 기반). 시장 기반은 인증된 재생 가능 에너지의 구매를 장려합니다.
- 데이터 품질이 가장 중요합니다. 활동 기반 > 지출 기반 계층 구조 데이터 수집 선택을 안내해야 합니다. 지출 기반 견적으로 시작하여 해마다 개선해 보세요.
- 계산의 불변성: GHGCalculation을 현재 위치에서 업데이트하지 마십시오. ESG 감사자 검토를 위해 요소를 버전화하고 완전한 감사 추적을 유지합니다.
- 노드 스타로서의 SBTi: 과학 기반 목표(2030년까지 -50%, 2050년까지 -90%) 최소 CSRD 준수가 아닌 감소 우선순위를 추진해야 합니다.
시리즈의 다음 기사
시리즈의 다음 기사에서는 "ESG & CSRD: 기술 기업에 대한 유럽의 의무", 완전한 규제 프레임워크를 살펴보겠습니다. CSRD의 적용을 받는 사람, 마감일 중소기업을 위한 이중 중요성 평가 개념, 데이터 모델링 시스템 등 이 기사에서 구성된 내용은 당국에서 요청한 ESRS E1 공개로 해석됩니다.
관련 기사도 참조하세요:
- MLOps 시리즈: ML 파이프라인에 탄소 추적을 통합하는 방법 (MLflow 모델 레지스트리의 지속 가능성 측정항목)
- AI 엔지니어링 시리즈: 생산 중인 RAG 및 LLM 시스템의 탄소 배출량
- 제9조(본 시리즈): 범위 3 파이프라인 — 자동화 공급업체를 위한 API 커넥터를 사용하여 가치 사슬에서 데이터 수집