炭素会計ソフトウェア アーキテクチャ: ESG プラットフォーム
2025 年には、炭素報告は大企業の自主的な選択ではなくなる ヨーロッパ諸国: それは法的義務です。そこには 企業持続可能性報告指令 (CSRD)、エントリー イタリアの立法 125/2024 とともに発効し、ESG レポートの状況を根本的に変革し、 企業コミュニケーションの領域から規制遵守の領域まで、同じ規律を持ちながら 財務会計システムの運用上および技術的な影響。
世界の炭素会計ソフトウェア市場が追いついてきた 2025年には18億ドル 2030 年までに 60 億を超え、CAGR は 27.4% になると予想されています。ただ買うだけではない SaaS プラットフォーム: 堅牢な炭素会計システムの設計には、以下のようなスキルが必要です。 大気化学からソフトウェア工学、統計からコミュニティ法まで。エンジニアリングチーム これらのプラットフォームを構築している人は、次のことを理解する必要があります。 GHGプロトコル, gli 欧州持続可能性報告基準 (ESRS)、排出係数データベース、および 何百ものデータを収集、計算、認証できる分散システムのアーキテクチャ さまざまな情報源から。
この記事は完全な技術ガイドです。GHG プロトコル エンティティを使用したドメインのモデル化から、 FastAPIを使用したPythonでの計算エンジンの実装から、CSRD/CDP/GRIレポートの自動化まで リアルタイム排出係数のための Climatiq API との統合。プラットフォームを構築している場合 社内 ESG、エンタープライズ ソリューションの評価、または単にこれらのアーキテクチャがどのように機能するかを理解したい場合 ボンネットの下では、あなたは正しい場所にいます。
この記事で学べること
- GHG プロトコル: スコープ 1、2、3 とサプライチェーンからの間接排出の 15 カテゴリー
- 炭素会計システムのデータドメイン（組織、施設、排出源、活動）をモデル化する方法
- 排出係数の主なデータベース: DEFRA、EPA、ecoinvent、Climatiq API
- マイクロサービス アーキテクチャ: データ収集、計算エンジン、レポート作成、監査証跡
- FastAPI とパンダによる排出量計算のための Python 実装
- スコープ 3 の自動化: SAP ERP、調達データ、旅行データとの統合
- CSRD/ESRS、CDP、GRI Standards レポートの自動生成
- 監査証跡とデータ系統: 計算の完全なトレーサビリティ
- プラットフォームの比較: Persefoni、Watershed、Sphera、Plan A
- ケーススタディ: 2025 年の CSRD 報告書を提出したイタリアの製造会社
- 更新された法律: CSRD オムニバス 2025、EU 分類、立法令 125/2024
EnergyTechシリーズにおける位置づけ
| # | アイテム | Stato |
|---|---|---|
| 1 | MQTT と InfluxDB: エネルギー データの時系列 | 発行済み |
| 2 | IEC 61850: インテリジェント電気ネットワークの標準プロトコル | 発行済み |
| 3 | DERMS: 分散型エネルギー リソース管理 | 発行済み |
| 4 | ビル管理システム: AI エネルギー最適化 | 発行済み |
| 5 | 再生可能エネルギーの予測: 太陽光発電と風力発電の ML | 発行済み |
| 6 | EV ロード バランシング: スマート充電と車両から電力網へ | 発行済み |
| 7 | ブロックチェーン P2P エネルギー取引: 分散型エネルギー市場 | 発行済み |
| 8 | 現在位置 - 炭素会計ソフトウェア アーキテクチャ: ESG プラットフォーム | 現在 |
| 9 | エネルギーデジタルツイン: シミュレーションと最適化 | Prossimo |
| 10 | OCPP と EV インフラストラクチャ: 標準と実装 | 近日公開 |
規制の背景: CSRD、ESRS、および立法令 125/2024
法律を理解することは官僚的な前提条件ではありません。それはアーキテクチャ全体を設計するための基礎です。 システムの。すべての技術要件、すべてのデータベース フィールド、すべての API エンドポイントは義務から生まれます。 具体的な開示内容。
CSRD とアプリケーションウェーブ
La 企業の持続可能性報告指令 アプリケーションの 3 つの波を定義 ヨーロッパ企業向け。第 1 波（2024 年度、報告書は 2025 年に発行）には企業が関与しました すでに以前の NFRD の対象となっているのは、従業員 500 名を超える上場企業、銀行、保険会社です。 第2波と第3波は当初2025年と2026年に予定されていたが、2年延期された から 時計を止める指令 2025年4月16日にEU官報に掲載されました。
最も重要な変更は、パッケージの承認とともに 2025 年 12 月に行われました。 オムニバスⅠ: 必須の申請しきい値が次のように引き上げられました。 従業員数 1,000 名、売上高 4 億 5,000 万ユーロ、約80％減少します。 CSRD義務の対象となる企業の割合。 ESRS 基準は改訂され、削減されています 必須データポイントの 61% (約 1,100 から約 430)、前半に採用される予定 2026年から適用され、2027年から適用されます。
CSRD タイムラインを更新 (オムニバス 2025 年以降)
| Ondata | 科目 | 第一報 | Stato |
|---|---|---|---|
| ウェーブ 1 | すでに NFRD を導入している企業 (従業員数 500 人以上、上場/銀行/保険) | レポート2025（2024年度） | 進行中 |
| ウェーブ 2 | 従業員数が 1,000 人を超え、売上高が 4 億 5,000 万ユーロを超える大企業 | レポート 2027 (2026 年度) - 延期 | 延期した |
| ウェーブ 3 | 規制された EU 市場に上場している中小企業 | レポート 2028 (2027 年度) - 延期 | 延期した |
| ESRS 改訂版 | すべてのCSRD主題、簡素化された基準 | 2027年度～ | 相談中 |
立法令 125/2024: イタリア語転置
イタリアはCSRDを実施しており、 2024 年 9 月 6 日の法令 125、 この政令は、以前の立法政令 254/2016 を廃止するもので、2024 年 9 月 25 日に発効しました。 NFRDを置き換えました。イタリア企業の主な革新には次のようなものがあります。 限定的な保証 持続可能性レポートに関する（限定的保証） 資格のある監査人、持続可能性報告書の管理報告書への統合、 会社登記簿の専用セクションに掲載されます。
EU 分類法: つながり
炭素会計は次のものと直接交差します。 EU 分類規則 (EU Reg. 852/2020)、 これは、6 つの環境目標に基づいて経済活動を「持続可能な」ものとして分類します。 CSRD 企業はタクソノミーの整合性に関する KPI (売上高シェア、設備投資、運用コスト) を報告する必要があります。 「調整済み」および「適格」）。 2025 年の簡素化により、従業員数 1,000 人未満の企業は 分類レポートから免除され、残りはそれらが表す活動に限定される場合があります 売上高、CapEx または OpEx の少なくとも 10%。
GHG プロトコル: 参照フレームワーク
Il 温室効果ガスプロトコル 温室効果ガス排出量の会計基準と 世界で最も普及しており、世界資源研究所 (WRI) と世界ビジネス評議会によって開発されました。 持続可能な開発 (WBCSD) のために。事実上すべての報告フレームワーク (CSRD/ESRS、 CDP、GRI、ISO 14064) は、GHG プロトコルを参照するか、GHG プロトコルに基づいています。
3 つのスコープ: 正確な定義
3 つの「範囲」に区別することで、排出量を明確に帰属させることができます。 サプライチェーン内の異なる関係者間でのカウントの重複を回避します。
スコープ 1: 直接排出
組織が所有または管理する発生源からの排出。 これらには、ボイラー、オーブン、社有車両での化石燃料の燃焼が含まれます。 プロセス排出量（例：化学反応からの CO2、家畜からの CH4）。逃亡者排出物 (冷媒漏れ、システムからのガス漏れ)。関連する GHG ガスは次の 7 つです。 京都議定書: CO2、CH4、N2O、HFC、PFC、SF6、NF3、すべて換算 CO2 換算量 (CO2e) IPCC 地球温暖化係数 (GWP) を使用します。
スコープ 2: 購入したエネルギーによる間接排出
電気、熱、蒸気、冷却の生成に伴う排出 組織によって購入および消費されます。 GHG プロトコルのスコープ 2 ガイダンス (2015) では、次のことが求められています。 と報告する 2 つの異なる方法: 方法 ロケーションベースの （所在地の電力網の排出係数を使用） 消費、例:イタリアのネットワークの平均係数) とその方法 市場ベースの (差額契約、証券などの市場手段の要素を使用します) 再生可能エネルギー/GO、電力購入契約）。両方とも報告する必要があります。
スコープ 3: その他の間接排出
バリューチェーンの排出量を上流と下流の 15 のカテゴリーに分類します。 これらは通常、最も重要であり (平均して総フットプリントの 70 ～ 80%)、 測定するのが難しい。 ESRS ではスコープ 3 カテゴリの報告が必要です ダブルマテリアリティ評価により特定された「素材」。
スコープ 3 の 15 のカテゴリ
| # | カテゴリ | タイプ | 典型的なのは |
|---|---|---|---|
| 1 | 購入した商品やサービス | 上流 | 全分野 |
| 2 | 資本財 | 上流 | 製造業、建設業 |
| 3 | 燃料およびエネルギー関連の活動 | 上流 | 全分野 |
| 4 | 上流の輸送と流通 | 上流 | 小売業、製造業 |
| 5 | 事業活動で発生する廃棄物 | 上流 | 製造業、食品 |
| 6 | 出張 | 上流 | サービス、テクノロジー |
| 7 | 従業員の通勤 | 上流 | 全分野 |
| 8 | 上流のリース資産 | 上流 | 不動産、小売業 |
| 9 | 下流輸送 | 下流 | 製造業、日用消費財 |
| 10 | 販売した商品の加工 | 下流 | 原材料、化学品 |
| 11 | 販売した商品の使用 | 下流 | 自動車、エレクトロニクス |
| 12 | 終末期治療 | 下流 | 包装、耐久消費財 |
| 13 | 下流のリース資産 | 下流 | 不動産 |
| 14 | フランチャイズ | 下流 | 飲食、小売 |
| 15 | 投資 | 下流 | 銀行、投資ファンド |
データ モデル: 炭素会計ドメインのモデル化
すべての炭素会計プラットフォームの中心となるのは、忠実に反映する堅牢なデータ モデルです。 GHG プロトコルの概念。主要なエンティティとその属性および関係を見てみましょう。
主要な事業体
完全なシステムには、少なくとも 6 つのコア エンティティ (組織、施設、排出源、 排出係数、活動量および計算。 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、走行距離km、支出されたユーロ）をCO2eトンで表したもの。 要因の品質と更新は、レポートの品質を直接決定します。
主要なデータベース
| データベース | マネージャー | カバレッジ | アップデート | アクセス |
|---|---|---|---|---|
| DEFRA/BEIS | 英国政府 (DESNZ) | 英国 + 国際、すべての範囲 | 毎年恒例（7月） | 無料（エクセル） |
| EPA GHG ハブ | 米国EPA | 米国、モビリティ、エネルギー、スコープ 3 | 毎年恒例（1月） | 無料（エクセル） |
| エコ発明 | エコインベント協会 | グローバルな完全な LCA、18,000 以上のデータセット | 半年ごと | 有料ライセンス |
| IPCC EF データベース | IPCC | 世界規模の国内在庫 | 評価レポートごとに | 無料 |
| Climatiq API | 気候 | マルチソース、50,000 以上の要素 | 連続（リアルタイム） | API（フリーミアム） |
| AIB残留混合物 | 発行機関協会 | ヨーロッパ、スコープ 2 市場ベース | 年間 | 無料 |
| IEA電気 | 国際エネルギー機関 | 世界的な電力網の要因 | 年間 | 部分的に無料 |
Climatiq API の統合
Climatiq は、50,000 を超える検証済みの排出係数、カバレッジを備えた REST API を提供します。 ISO 14067 および GHG プロトコルに準拠しており、計算を統合するために最もよく使用されるソリューションの 1 つです プログラムによる排出量の削減。 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スタンダード開示、内部報告書 ダッシュボード用。レポートのバージョン管理とデジタル署名を管理して保証します。
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、調達、出張
スコープ 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 で定義されており、3 つのスコープすべてのデータが必要です。
# 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 重視 | 金融、企業 | 年間 50,000 ～ 500,000 ドル | Si |
| 流域 | レポートまでのスピード、高度なスコープ 3 ツール | テクノロジー、エンタープライズ | 10万～100万ドル/年 | Si |
| スフェラ | 統合LCA、産業コンプライアンス、リスク管理 | 製造、エネルギー、化学 | リクエストに応じて | Si |
| プランA | シンプルなUX、迅速な実装 | 中小企業・中堅企業 | 10,000 ～ 100,000 ドル/年 | 部分的 |
| IBM Envizi | 40,000 以上の排出係数、ERP 統合 | エンタープライズ、公益事業 | リクエストに応じて | Si |
| カスタムビルド | 完全な柔軟性、内部システムのネイティブ統合 | 特定の要件を持つ大企業 | 50万ドルから500万ドルの開発 | 場合によって異なります |
カスタムを構築する場合と購入する場合
次の場合にプラットフォームを購入してください。 標準的な要件があり、ライブに移行したい 3 ～ 6 か月後には、サステナビリティに特化したエンジニアリング チームがなくなり、データ量が膨大になります。 そして扱いやすい。
次の場合にカスタムを構築します。 データを使用した非常に特殊な生産プロセスがある 標準のプラットフォームでは管理できないため、レガシー システムとネイティブに統合したい、 SaaS と互換性のないデータ常駐要件、またはデータ量があり、 年間数百万レコードのオーダーになります（SaaS のコストが法外に高くなります）。
ケーススタディ: CSRD 2025 を活用したイタリアの製造業中小企業
従業員数 1,200 人のイタリアの精密エンジニアリング製造会社、 売上高は 3 億 5,000 万ユーロで、工場はトリノ、ミラノ、ブレシアにあります。 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
}
}ケーススタディに関する考察
最も重要な結果は、スコープ 3 が 排出量の91.4% 合計、カテゴリ 1 (原材料および成分) だけで価格の 80% の価値がある スコープ 3。これは製造部門に典型的なものであり、CSRD がスコープ 3 を必要とする理由を説明しています。 サプライチェーン報告: スコープ 3 がなければ、炭素会計でカバーできる範囲は以下になります。 実際の影響の 10%。
スコープ 2 の位置ベース (1,922 tCO2e) と市場ベース (1,629 tCO2e) の差は次のとおりです。 50％の再生可能エネルギー証明書（原産地保証 - GO）の購入に 電力消費量。 2024 年のイタリアの残留混合物 (0.395 kgCO2e/kWh) は係数よりも高い ネットワークの平均 (0.233 kgCO2e/kWh): これは直感に反しますが、方法論的には正しいです。 残りの混合には既に GO で認証されているエネルギーが含まれていないため、割り当てが「含まれている」ためです。 高強度のソースよりも高い。
テスト: 計算の検証
排出量の計算は、財務法と同じ厳格さでテストする必要があります。 1 つの変換係数の誤差が桁違いの誤差につながる可能性があります 数百トンの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.01Science Based Targets (SBTi) とダッシュボード
Il Science Based Targets イニシアチブ (SBTi) 正確な基準を定義している パリ協定に沿った排出削減目標に向けて。 SBTi 企業基準では次のことが求められています: 2030 年までにスコープ 1+2 を 42% 削減 1.5℃シナリオの場合は（ベースライン 2020）、スコープ 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ベストプラクティスとアンチパターン
ベストプラクティス
- 計算の不変性: 計算して保存すると、排出量が 決して変更してはなりません。係数が変更されると、新しいバージョンの計算が作成されます。 監査証跡には両方のバージョンが表示される必要があります。
- 両方のスコープ 2 メソッドを報告します。 ロケーションベースとマーケットベースは、 ESRS と GHG プロトコルに従って両方とも必須です。市場ベースのレポートだけではない 低いからです。
- スコープ 3 の重要性を文書化: 15 のカテゴリすべてを計算する必要はありません スコープ 3. ただし、重要性分析を使用して、なぜ含めた/除外したのかを実証する必要があります。 各カテゴリー。これは明示的な ESRS 要件です。
- 排出係数のバージョン管理: 要因は毎年変化します。 各計算に使用される係数のスナップショットを保持します。更新しないでください 新しいバージョンで明示的な再計算を行わずに遡及的に実行できます。
- データ品質フラグ: 各データを品質（測定、 計算、推定、支出ベース）。 ESRS には品質の宣言が必要です スコープ 3 データの。低品質のデータも使用できますが、使用する必要があります。 宣言される。
- 境界文書: 含まれるエンティティを明示的に文書化する 報告範囲内および除外された範囲（正当な理由がある場合） (しきい値 <5%、データが利用できないなど)。
避けるべきアンチパターン
- 市場ベースのスコープ 2 のみを使用します。 GHG プロトコルでは両方が必要です。 市場ベース (通常は GO/PPA のおかげで低い値) のみを報告し、何も含まない 場所に基づいており、方法論的に間違っており、グリーンウォッシングの可能性があります。
- 排出係数が更新されていない: 5 年以上前の要素を使用する 特に変化する電力網に対して重大な誤差が生じる 再生可能エネルギーの成長とともに毎年増加しています。
- 連結における重複排出量: 子会社の場合 排出量を計算し、親会社はそれを連結に含めます。 株式のシェアや運営管理方法は一貫して使用する必要があります。
- 不確実性を無視する: すべての排出量の計算には、 不確実性、特にスコープ 3。 信頼範囲と使用される方法を示します。
- 保証のない報告: CSRD を使用すると、限定的な保証なしでレポートが作成されます Wave 1 の規制要件を満たしていません。保証を計画する 思いつきではなく、最初の年から。
結論と実装ロードマップ
エンタープライズグレードの炭素会計プラットフォームの設計は複雑なプロジェクトです これには多分野のスキルが必要です。しかし、概念的な構造は明らかです。 GHGプロトコル 会計フレームワーク、データベースを提供します。 排出係数 (DEFRA、EPA、Climatiq) が係数を提供します。 のマイクロサービスアーキテクチャ スケーラビリティと保守性を保証し、 そして不変の監査証跡 CSRDの信頼性を確保します。
2025 年から 2026 年にかけて CSRD に取り組む必要がある中小企業にとって、実際的な推奨事項は次のとおりです。 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 | 継続的 | データ品質とサプライヤーの関与の向上 | 支出ベースの削減、サプライヤー固有のスコープ 3 の増加 |
重要なリソース
- GHG プロトコル企業標準: ghgprotocol.org - 標準 リファレンス、無料、ダウンロード可能
- DEFRA 排出係数 2024: gov.uk/government/publications/ 温室効果ガス報告 - 換算係数 - 2024 年
- Climatiq API: climatiq.io/docs - 完全なドキュメント 無料のクイックスタート
- ESRS E1 標準: efrag.org - 欧州財務報告 アドバイザリーグループ、最終的な ESRS 基準
- SBTi 企業マニュアル: sciencebasedtargets.org - ガイド 科学的根拠に基づいた目標の定義へ
- エコインベントデータベース: ecoinvent.org - LCA データベース 詳細なプロセスデータを含むスコープ 3
EnergyTech シリーズの今後の記事
次の記事では、 デジタルエネルギーツイン: 作成方法 排出削減シナリオをシミュレーションするための産業プラントの仮想レプリカ 物理的な世界に実装する前に。ますます中心となるテクノロジー 大規模産業の脱炭素戦略。
データ ビジネスに適用される AI テクノロジーの詳細については、こちらもご覧ください。 シリーズ データ ウェアハウス、AI、デジタル トランスフォーメーション、 特に以下の記事 信頼性の高い AI のためのデータ ガバナンスとデータ品質 そして上へ ビジネス向け MLOps、 計算モデルの管理に直接適用できるトピックを扱います。 生産時の排出量。