スコープ 1、2、および 3: ESG レポート ソフトウェアのデータ モデリング
Il GHGプロトコル （温室効果ガスプロトコル）は事実上の世界共通の枠組みとなった 企業の温室効果ガス排出量を測定し報告する。世界資源研究所 (WRI) によって開発されました。 持続可能な開発のための世界経済人会議 (WBCSD) は、排出量の 3 つの「範囲」を定義しています。 組織のバリューチェーン全体をカバーします。
の発効に伴い、 CSRD (企業持続可能性報告指令) 連合の ヨーロッパと標準 ESRS E1、企業を含む数千の企業 ソフトウェアと技術 — 独自の排出量データを収集、計算、公開することが義務付けられています 厳密な方法論に従って。 2024 年の時点で、従業員 500 人を超える大企業は、 2025年に最初の報告義務を伴うデータ収集を開始する。従業員250人以上の企業 2026年から続くことになる。
しかし、なぜ私は 開発者 スコープ 1、2、3 を理解する必要がありますか?なぜシステムを構築するのか 信頼性の高い ESG はサステナビリティ管理者だけの問題ではなく、 データエンジニアリング。一貫したデータ モデル、正確な計算パイプライン、API が必要です。 アクティビティ収集、および監査対応レポート。これらのシステムを設計および実装するのは誰か すべての ESG レポートの品質、ひいては信頼性。
この記事では、 ESGレポートのためのデータモデリングシステム、 スコープ定義からデータベース テーブルまで、Python 計算から FastAPI エンドポイントまで、 従業員 200 人のソフトウェア中小企業の完全なケーススタディまで。
何を学ぶか
- GHG プロトコルの構造: 正確な定義を含むスコープ 1、2、および 3
- 直接、間接、バリューチェーン排出量のエンティティ関係モデル
- スコープ 2 の場所ベースと市場ベース: 違いとデータへの影響
- 15 のスコープ 3 カテゴリとソフトウェア会社に関連するカテゴリ
- 完全な SQLAlchemy スキーマ: 組織、施設、排出元、排出係数、アクティビティデータ
- アクティビティベースの見積もりと支出ベースの見積もりを使用した Python 計算
- 集計、前年比比較、SBTi目標との整合性
- タスクの送信、排出量の計算、レポートの生成のための FastAPI を備えた REST API
- pytest を使用した GHG 計算の単体テストと排出係数の検証
- エンドツーエンドのケーススタディ: ソフトウェア中小企業従業員 200 名、スコープ 1+2+3 完了
グリーン ソフトウェア シリーズ — 10 件の記事
| # | タイトル | Stato |
|---|---|---|
| 1 | グリーン ソフトウェア エンジニアリングと SCI の原則 | 発行済み |
| 2 | CodeCarbon: Python コードエミッションの測定 | 発行済み |
| 3 | Carbon Aware SDK: ワークロードシフトとタイムシフト | 発行済み |
| 4 | Climatiq API: 排出係数と炭素の計算 | 発行済み |
| 5 | スコープ 1、2、3: ESG レポートのためのデータ モデリング | 現在の記事 |
| 6 | ESG と CSRD: テクノロジー企業に対する欧州の義務 | すぐ |
| 7 | 持続可能なソフトウェア パターン: 影響の少ないアーキテクチャ | すぐ |
| 8 | GreenOps: 持続可能性のためのクラウドの最適化 | すぐ |
| 9 | スコープ 3 パイプライン: バリュー チェーンを自動化する | すぐ |
| 10 | AI カーボン フットプリント: LLM、トレーニングと推論 | すぐ |
GHG プロトコル: 排出に関する普遍的な枠組み
2001 年に発行され、2004 年に更新された GHG プロトコル企業基準は、 ほぼすべての気候報告フレームワーク: CDP、TCFD、CSRD/ESRS E1、SBTi、その他多数 彼らはそれを直接参照します。その強みは概念的な明快さにあります。 組織は、相互に排他的ではありますが、全体的に網羅的な 3 つのカテゴリに分類されます。
このフレームワークでは次の概念が使用されます。 組織の境界線 (組織の境界) どの排出量を計算に含めるかを決定します。 2 つのアプローチがあります。株式シェアアプローチ （財政参加率に基づく）および コントロールアプローチ (制御に基づく 運営上または財務上）。ほとんどの企業は、運用管理を主要な基準として採用しています。
3 つのスコープ: 概要
| ほうき | 意味 | 代表的な例 | ESRS E1 必須 |
|---|---|---|---|
| スコープ1 | 組織が所有または管理する発生源からの直接排出 | 天然ガスボイラー、自社車両、ディーゼル発電機、産業プロセス | 義務的 |
| スコープ2 | エネルギーの購入による間接排出（電気、蒸気、熱、冷気） | オフィスの電力、データセンターの冷房、地域暖房 | 必須（両方の方法） |
| スコープ3 | バリューチェーン（上流および下流）におけるその他すべての間接排出 | 出張、通勤、商品・サービスの購入、販売商品の利用 | 必須（材料カテゴリ） |
スコープ 1: 直接排出のデータモデル
スコープ 1 の排出は、企業が直接管理する物理的発生源に由来します。彼らのために ソフトウェア会社の場合、これらは通常小規模ではありますが、無視できるものではありません。暖房も含まれます。 オフィス (天然ガス)、バックアップ発電機、および会社の車両 (社用車) またはハードウェア配送バン）。
GHG プロトコルでは、スコープ 1 排出の 4 つのカテゴリが特定されています。
- 定常燃焼: 化石燃料を燃やすボイラー、ヒーター、発電機
- 移動式燃焼: ガソリン、ディーゼル、LPG を燃料とする社用車
- プロセス排出量: 化学反応または生物学的反応 (ソフトウェアにはほとんど関係ありません)
- 逃亡放出: HVAC システムからの冷媒 (R-410A、R-134a) の漏れ
スコープ 1 の計算の基本式は次のとおりです。
排出量 (tCO₂e) = 活動データ × 排出係数 × GWP
どこ：
- 与えられたアクティビティ: 燃料消費量 (リットル、m3、kWh)
- 排出係数: 燃料単位あたり CO₂ kg (出典: IPCC、DEFRA、IEA)
- GWP: 地球温暖化係数 (CO₂=1、CH₄=28、N₂O=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: 場所ベースと市場ベース
スコープ 2 はテクノロジー企業にとって特に複雑です。 オフィスやデータセンターからの 間接排出の主な発生源。 GHG プロトコルのスコープ 2 ガイダンス (2015) では、報告する義務が導入されました。 両方 方法は、場所ベースと市場ベースです。
ロケーションベースとマーケットベース: 主な違い
| 待ってます | ロケーションベース | 市場ベース |
|---|---|---|
| 意味 | 消費が発生する地域の電力網の平均強度 | 契約によりエネルギーを購入する発電機からの排出量 |
| 排出係数 | 国/地域別のグリッド平均排出係数 (gCO₂/kWh) | サプライヤー固有の係数または残留混合係数 |
| 契約文書 | 適用できない | GO (原産地保証)、REC (再生可能エネルギー証明書)、PPA |
| 例 イタリア 2024 | ~310 gCO₂/kWh (AIB 残留混合データ) | 100% 検証済みの再生可能 GO を購入した場合、0 gCO₂/kWh |
| GHGプロトコル2025提案 | 時間精度を備えた新しい因子階層 | 再生可能エネルギーの 1 時間ごとのマッチングをリクエストする |
| ESRS E1での使用 | 開示義務（E1-6） | 開示義務（E1-6） |
データベース内のスコープ 2 を正しくモデル化するには、 エネルギー証明書 (GO/REC) が「消費」される別個のエンティティとして 企業の電力消費量に対して、市場ベースの係数をゼロに向けて引き下げます。
-- 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 プロトコルの計算に関する技術ガイダンス スコープ 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 （従業員の通勤） 猫。 11 (ユーザーが消費するエネルギー ソフトウェアを実行します)。後者は過小評価されることが多いですが、数百万規模のエンタープライズ SaaS にとっては ユーザーの数は膨大になる可能性があります。
SQLAlchemy スキーマ: 完全なモデル
それでは、3 つのスコープすべてをサポートする完全なデータベース スキーマを 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 つの計算方法:
- サプライヤー固有の方法: ベンダー直接データ (最も正確)
- ハイブリッド方式: 一次データと二次データの組み合わせ
- 平均データ法: 物理単位あたりの平均排出係数
- 支出ベースの方法: 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年次集計と前年比
各アクティビティレコードの排出量を計算したら、期間ごとに集計する必要があります。 レポートを作成し、前年と比較し、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.
| パラメータ | 価値 |
|---|---|
| 従業員 | 200 FTE (ミラノ 70%、ローマ 20%、フルリモート 10%) |
| 販売 | 1,500万ユーロ |
| オフィス | 賃貸2ヶ所（計2,400㎡） |
| 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 年)
| サイズ | tCO₂e/依存 | % スコープ 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% 削減 (-24 tCO2e)。 単品で一番大きいのは通勤です。
- 両方の場所で 100% GO: スコープ 2 市場ベースでは約 0 tCO2e まで低下 (-55 tCO2e)。 GO の推定コスト: 年間 3,000 ～ 5,000 ユーロ。
- 旅行の炭素予算ポリシー: 短いフライト（3 時間未満）を電車に置き換えます。 猫。 6 約 40% 削減 (-6.5 tCO2e)。
- AWS 二酸化炭素排出量ツール: 費用ベースの見積もりから特定のプロバイダーに切り替える 猫を減らします。 1 クラウドでは最大 40% (-12 tCO2e) 削減され、データ品質が向上します。
TechFlow 脱炭素化ロードマップ — 目標 SBTi 1.5°C
| Anno | 目標（tCO₂e） | 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 レポートのデータ モデリングは、単純なコンプライアンスの実践ではありません。 それは、信頼性があり検証可能な脱炭素戦略を構築するための基礎となります。 企業のバリューチェーン全体をカバーするシステムを構築する方法を見てきました。 ソフトウェア、暖房ボイラー (スコープ 1) から民間航空機旅行 (スコープ 3 カテゴリ 6) まで ユーザーが製品を実行するために消費するエネルギーまで (スコープ 3 カテゴリ 11)。
持っていくべき重要なポイント:
- スコープ 3 が優勢: ソフトウェア企業の場合、排出量の 55 ～ 75% がスコープ 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コネクタを使用したバリューチェーンからのデータ収集