Cracking the EBA XBRL Taxonomy with Arelle — a Python Walkthrough

# WSL or Linux terminal — download EBA COREP taxonomy
# EBA publishes taxonomy packages at:
# https://www.eba.europa.eu/risk-analysis-and-data/reporting-frameworks

cd corep-governance-pipeline/data/taxonomy

# EBA Reporting Framework 3.3 (ITS 2024 — valid for submissions from 2024-Q1)
wget https://www.eba.europa.eu/sites/default/documents/files/document_library/\
EBA_taxonomy_3.3.zip -O eba_taxonomy_3.3.zip

unzip eba_taxonomy_3.3.zip -d eba_3.3/

# The taxonomy package structure after extraction:
# eba_3.3/
#   www.eba.europa.eu/
#     eu/fr/xbrl/crr/fws/corep/
#       cor/2024-01-31/
#         mod/
#           c_01.00.xsd   ← C 01.00 Own Funds schema
#           c_02.00.xsd   ← C 02.00 RWA
#           c_03.00.xsd   ← C 03.00 Capital Ratios
#           c_47.00.xsd   ← C 47.00 LCR
#         tab/
#           c_01.00.tab   ← human-readable cell list
#         val/
#           c_01.00_val.xml ← EBA validation rules

# Install Arelle — the open-source XBRL processor
pip install arelle-release==2.30.3

# Verify installation
python -c "from arelle import Cntlr; print('Arelle OK')"

# Add to requirements.txt
# arelle-release==2.30.3    # Day 11 — XBRL generation and validation

# xbrl/explore_taxonomy.py — run this once to verify concept names

from arelle import Cntlr, ModelManager
from arelle.ModelValue import qname
import os

TAXONOMY_ENTRY = os.environ.get(
    "EBA_TAXONOMY_ENTRY",
    "data/taxonomy/eba_3.3/www.eba.europa.eu/eu/fr/xbrl/crr/fws/corep/cor/2024-01-31/mod/corep-full-entry-point.xsd",
)

# ── 1. Initialise Arelle controller ──────────────────────────────
cntlr = Cntlr.Cntlr(logFileName="arelle_explore.log")
cntlr.startLogging(logFileName="arelle_explore.log")
modelMgr = ModelManager.ModelManager(cntlr)

# ── 2. Load the taxonomy (this parses all XSD + linkbase files) ──
print("Loading EBA taxonomy — this takes 20-60 seconds on first load...")
modelXbrl = modelMgr.load(TAXONOMY_ENTRY)
print(f"Taxonomy loaded. Concepts: {len(modelXbrl.qnameConcepts)}")

# ── 3. Look up specific concepts by local name ───────────────────
EBA_NAMESPACE = "http://www.eba.europa.eu/xbrl/crr/dict/con"

CONCEPTS_TO_CHECK = [
    "c0010",  # Own Funds
    "c0020",  # CET1
    "c0030",  # AT1
    "c0040",  # T2
    "c0050",  # CET1 ratio (in C 03.00)
    "c0060",  # Total RWA
    "c0090",  # LCR ratio (in C 47.00)
]

print("\n── Concept Inspection ──────────────────────────────────────")
for local_name in CONCEPTS_TO_CHECK:
    qn = qname(EBA_NAMESPACE, local_name)
    concept = modelXbrl.qnameConcepts.get(qn)
    if concept:
        print(f"\n  {local_name}:")
        print(f"    QName     : {concept.qname}")
        print(f"    TypeName  : {concept.typeQname}")
        print(f"    PeriodType: {concept.periodType}")   # instant or duration
        print(f"    Balance   : {concept.balance}")       # debit / credit
        print(f"    Abstract  : {concept.isAbstract}")    # abstract = not a fact
        # Get the English label from the label linkbase
        labels = concept.label(lang="en", fallbackToQname=True)
        print(f"    Label (EN): {labels}")
    else:
        print(f"  {local_name}: NOT FOUND in taxonomy — check namespace or version")

# ── 4. Inspect the calculation linkbase for C 01.00 ─────────────
# This shows the summation tree: c0010 = c0020 + c0030 + c0040
print("\n── Calculation Linkbase (C 01.00 summations) ───────────────")
for rel in modelXbrl.relationshipSet("http://www.xbrl.org/2003/arcrole/summation-item").modelRelationships:
    parent = rel.fromModelObject
    child  = rel.toModelObject
    if parent and "c_01" in str(rel.linkQname):
        print(f"  {parent.qname.localName} → {child.qname.localName}  weight={rel.weight}")

modelMgr.close()
cntlr.close()

# Expected output from explore_taxonomy.py:

Loading EBA taxonomy — this takes 20-60 seconds on first load...
Taxonomy loaded. Concepts: 18420

── Concept Inspection ──────────────────────────────────────

  c0010:
    QName     : {http://www.eba.europa.eu/xbrl/crr/dict/con}c0010
    TypeName  : xbrli:monetaryItemType
    PeriodType: instant
    Balance   : credit
    Abstract  : False
    Label (EN): Own funds

  c0020:
    QName     : {http://www.eba.europa.eu/xbrl/crr/dict/con}c0020
    TypeName  : xbrli:monetaryItemType
    PeriodType: instant
    Balance   : credit
    Abstract  : False
    Label (EN): Common Equity Tier 1 capital

  c0050:
    QName     : {http://www.eba.europa.eu/xbrl/crr/dict/con}c0050
    TypeName  : xbrli:pureItemType
    PeriodType: instant
    Balance   : None
    Abstract  : False
    Label (EN): CET1 capital ratio

── Calculation Linkbase (C 01.00 summations) ───────────────
  c0010 → c0020  weight=1.0
  c0010 → c0030  weight=1.0
  c0010 → c0040  weight=1.0

# xbrl/read_mart.py — read mart tables via Trino for XBRL generation

from trino.dbapi import connect as trino_connect
import os, logging

log = logging.getLogger(__name__)

TRINO_HOST = os.environ.get("TRINO_HOST", "localhost")
TRINO_PORT = int(os.environ.get("TRINO_PORT", "8080"))
TRINO_USER = os.environ.get("TRINO_XBRL_USER", "corep_reporting_svc")


def fetch_mart_data() -> dict:
    """
    Query all four COREP mart tables via Trino.
    Returns a dict keyed by template code.
    """
    conn = trino_connect(
        host=TRINO_HOST,
        port=TRINO_PORT,
        user=TRINO_USER,
        catalog="postgresql",
        schema="mart",
    )
    cur = conn.cursor()
    mart_data = {}

    # ── C 01.00 — Own Funds ──────────────────────────────────────
    cur.execute("""
        SELECT own_funds, cet1_capital, at1_capital, t2_capital,
               total_rwa, reporting_date
        FROM mart.corep_c0100
        LIMIT 1
    """)
    row = cur.fetchone()
    cols = [d[0] for d in cur.description]
    mart_data["c0100"] = dict(zip(cols, row)) if row else {}

    # ── C 02.00 — RWA by exposure class ──────────────────────────
    cur.execute("""
        SELECT exposure_class, ead, rwa, reporting_date
        FROM mart.corep_c0200
        ORDER BY CASE exposure_class
            WHEN 'central_governments' THEN 1
            WHEN 'institutions'        THEN 2
            WHEN 'corporates'          THEN 3
            WHEN 'retail'              THEN 4
            WHEN 'real_estate'         THEN 5
            WHEN 'equity'              THEN 6
            WHEN 'other'               THEN 7
            ELSE 99
        END
    """)
    rows = cur.fetchall()
    cols = [d[0] for d in cur.description]
    mart_data["c0200"] = [dict(zip(cols, r)) for r in rows]

    # ── C 03.00 — Capital Ratios ──────────────────────────────────
    cur.execute("""
        SELECT cet1_ratio, tier1_ratio, total_capital_ratio,
               leverage_ratio, total_rwa, reporting_date
        FROM mart.corep_c0300
        LIMIT 1
    """)
    row = cur.fetchone()
    cols = [d[0] for d in cur.description]
    mart_data["c0300"] = dict(zip(cols, row)) if row else {}

    # ── C 47.00 — LCR ────────────────────────────────────────────
    cur.execute("""
        SELECT hqla_level1, hqla_level2a, hqla_level2b, hqla_buffer,
               net_outflows, lcr_ratio, reporting_date
        FROM mart.corep_c4700
        LIMIT 1
    """)
    row = cur.fetchone()
    cols = [d[0] for d in cur.description]
    mart_data["c4700"] = dict(zip(cols, row)) if row else {}

    cur.close()
    conn.close()

    # Validate all four templates have data
    empty = [k for k, v in mart_data.items() if not v]
    if empty:
        raise RuntimeError(f"[xbrl_gen] Empty mart tables for templates: {empty}")

    log.info("[xbrl_gen] Mart data fetched for templates: %s", list(mart_data.keys()))
    return mart_data

# modules/xbrl_gen.py — full implementation
"""
modules/xbrl_gen.py — Generate EBA COREP XBRL instance document from mart tables.

Reads mart data via Trino, loads mart_to_xbrl_mapping.yaml,
and produces a valid XBRL instance document for submission.
"""

import logging, os, yaml
from datetime import date, datetime, timezone
from decimal import Decimal, ROUND_HALF_UP
from pathlib import Path
from xml.etree import ElementTree as ET

from modules.base import BaseModule
from xbrl.read_mart import fetch_mart_data

log = logging.getLogger(__name__)

# ── Namespace map ─────────────────────────────────────────────────
NS = {
    "xbrli":    "http://www.xbrl.org/2003/instance",
    "link":     "http://www.xbrl.org/2003/linkbase",
    "xlink":    "http://www.w3.org/1999/xlink",
    "iso4217":  "http://www.xbrl.org/2003/iso4217",
    "ei":       "http://www.eba.europa.eu/xbrl/crr/dict/con",
    "find":     "http://www.eurofiling.info/xbrl/ext/filing-indicators",
}

TAXONOMY_HREF = (
    "http://www.eba.europa.eu/eu/fr/xbrl/crr/fws/corep/cor/2024-01-31/mod/corep-full-entry-point.xsd"
)

MAPPING_FILE = Path(os.environ.get("XBRL_MAPPING_FILE", "mart_to_xbrl_mapping.yaml"))
OUTPUT_DIR   = Path(os.environ.get("XBRL_OUTPUT_DIR",   "output/xbrl"))


class XbrlGenModule(BaseModule):
    MODULE_NAME = "xbrl_gen"

    def input_check(self) -> None:
        if not MAPPING_FILE.exists():
            raise RuntimeError(f"[xbrl_gen] Mapping file not found: {MAPPING_FILE}")
        OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
        log.info("[xbrl_gen] Mapping file present. Output dir: %s", OUTPUT_DIR)

    def _execute(self) -> None:
        mapping  = yaml.safe_load(MAPPING_FILE.read_text())
        mart     = fetch_mart_data()

        # Determine reporting date from mart data
        reporting_date = (
            mart["c0100"].get("reporting_date")
            or date.today().replace(day=31)  # quarter-end fallback
        )
        entity_lei = os.environ.get("ENTITY_LEI", "529900T8BM49AURSDO55")

        # Register all namespaces (prevents ET from generating ns0: prefixes)
        for prefix, uri in NS.items():
            ET.register_namespace(prefix, uri)

        # ── Build the XBRL root element ───────────────────────────────
        xbrl = ET.Element("{http://www.xbrl.org/2003/instance}xbrl")

        # Schema reference
        schema_ref = ET.SubElement(
            xbrl, "{http://www.xbrl.org/2003/linkbase}schemaRef"
        )
        schema_ref.set("{http://www.w3.org/1999/xlink}type", "simple")
        schema_ref.set("{http://www.w3.org/1999/xlink}href", TAXONOMY_HREF)

        # ── Context ──────────────────────────────────────────────────
        ctx_id = f"C_{reporting_date}_instant"
        ctx    = ET.SubElement(xbrl, "{http://www.xbrl.org/2003/instance}context", id=ctx_id)
        entity = ET.SubElement(ctx,  "{http://www.xbrl.org/2003/instance}entity")
        ident  = ET.SubElement(entity, "{http://www.xbrl.org/2003/instance}identifier")
        ident.set("scheme", "http://standards.iso.org/iso/17442")
        ident.text = entity_lei
        period   = ET.SubElement(ctx, "{http://www.xbrl.org/2003/instance}period")
        instant  = ET.SubElement(period, "{http://www.xbrl.org/2003/instance}instant")
        instant.text = str(reporting_date)

        # ── Units ─────────────────────────────────────────────────────
        unit_eur  = ET.SubElement(xbrl, "{http://www.xbrl.org/2003/instance}unit", id="EUR")
        measure_e = ET.SubElement(unit_eur, "{http://www.xbrl.org/2003/instance}measure")
        measure_e.text = "iso4217:EUR"

        unit_pure  = ET.SubElement(xbrl, "{http://www.xbrl.org/2003/instance}unit", id="pure")
        measure_p  = ET.SubElement(unit_pure, "{http://www.xbrl.org/2003/instance}measure")
        measure_p.text = "xbrli:pure"

        # ── Filing indicators ─────────────────────────────────────────
        find_ns = NS["find"]
        fi_root = ET.SubElement(xbrl, f"{{{find_ns}}}fIndicators")
        for tmpl in ["C 01.00", "C 02.00", "C 03.00", "C 47.00"]:
            fi = ET.SubElement(fi_root, f"{{{find_ns}}}filingIndicator")
            fi.set("contextRef", ctx_id)
            fi.text = tmpl

        # ── Generate facts from mapping ───────────────────────────────
        facts_written = 0
        for template_code, template_mapping in mapping.items():
            mart_key = template_code.lower().replace(" ", "").replace(".", "")
            data     = mart.get(mart_key, {})

            if isinstance(data, list):
                # C 02.00 — multi-row template
                for row in data:
                    facts_written += self._write_facts(
                        xbrl, template_mapping, row, ctx_id
                    )
            else:
                facts_written += self._write_facts(
                    xbrl, template_mapping, data, ctx_id
                )

        # ── Write output file ─────────────────────────────────────────
        ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
        out_path = OUTPUT_DIR / f"COREP_{reporting_date}_{ts}.xbrl"
        tree = ET.ElementTree(xbrl)
        ET.indent(tree, space="  ")   # Python 3.9+ pretty-print
        tree.write(str(out_path), xml_declaration=True, encoding="utf-8")
        log.info("[xbrl_gen] Instance written: %s (%d facts)", out_path, facts_written)

        self._upload_to_minio(out_path)
        self._instance_path = out_path   # stored for output_check

    def _write_facts(
        self,
        xbrl: ET.Element,
        template_mapping: list,
        data: dict,
        ctx_id: str,
    ) -> int:
        """
        Write one set of facts from a mapping list + a data dict.
        Returns count of facts written.
        """
        written = 0
        for mapping_entry in template_mapping:
            mart_col    = mapping_entry["mart_column"]
            concept_id  = mapping_entry["xbrl_concept"]   # e.g. "c0020"
            unit        = mapping_entry["unit"]            # "EUR" or "pure"
            decimals    = mapping_entry["decimals"]        # -3 or 4

            value = data.get(mart_col)
            if value is None:
                log.warning("[xbrl_gen] Missing mart value: %s → skipping concept %s", mart_col, concept_id)
                continue

            # Apply decimal rounding per EBA DPM precision rules
            value_str = self._format_value(value, decimals, unit)

            fact = ET.SubElement(
                xbrl,
                f"{{{NS['ei']}}}{concept_id}"
            )
            fact.set("contextRef", ctx_id)
            fact.set("unitRef",    unit)
            fact.set("decimals",   str(decimals))
            fact.text = value_str
            written += 1

        return written

    @staticmethod
    def _format_value(value, decimals: int, unit: str) -> str:
        """
        Format a mart value for XBRL output.

        EBA DPM rules:
          - Monetary (EUR): decimals=-3 means report value in thousands,
            so divide by 1000 and round to nearest integer.
          - Ratios (pure): decimals=4 means 4 significant figures;
            report the raw decimal (e.g. 0.112583).
        """
        d = Decimal(str(value))
        if unit == "EUR":
            # Report in thousands (decimals=-3)
            d_thousands = (d / Decimal("1000")).quantize(
                Decimal("1"), rounding=ROUND_HALF_UP
            )
            return str(d_thousands)
        else:
            # Pure ratio — keep 6 decimal places as produced by dbt ROUND(6)
            return str(d.quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP))

    def _upload_to_minio(self, out_path: Path) -> None:
        try:
            from minio import Minio
            client = Minio(
                os.environ.get("MINIO_ENDPOINT", "minio:9000"),
                access_key=os.environ.get("MINIO_ROOT_USER", "minioadmin"),
                secret_key=os.environ.get("MINIO_ROOT_PASSWORD", "minioadmin"),
                secure=False,
            )
            bucket = "corep-xbrl-output"
            if not client.bucket_exists(bucket):
                client.make_bucket(bucket)
            client.fput_object(
                bucket, out_path.name, str(out_path),
                content_type="application/xbrl+xml",
            )
            log.info("[xbrl_gen] Uploaded → minio://%s/%s", bucket, out_path.name)
        except Exception as exc:
            log.warning("[xbrl_gen] MinIO upload failed (non-fatal): %s", exc)

    def emit_lineage(self) -> None:
        from openlineage.client.run import RunEvent, RunState, Run, Job, Dataset
        from datetime import datetime, timezone

        event = RunEvent(
            eventType=RunState.COMPLETE,
            eventTime=datetime.now(timezone.utc).isoformat(),
            run=Run(runId=self._run_id),
            job=Job(namespace=self._namespace, name="xbrl_gen"),
            inputs=[
                Dataset(namespace="trino://corep", name="mart.corep_c0100"),
                Dataset(namespace="trino://corep", name="mart.corep_c0200"),
                Dataset(namespace="trino://corep", name="mart.corep_c0300"),
                Dataset(namespace="trino://corep", name="mart.corep_c4700"),
            ],
            outputs=[
                Dataset(
                    namespace="minio://corep-xbrl-output",
                    name=str(getattr(self, "_instance_path", "COREP_unknown.xbrl")),
                )
            ],
            producer="https://github.com/your-org/corep-governance-pipeline/modules/xbrl_gen.py",
        )
        try:
            self._ol_client.emit(event)
            log.info("[xbrl_gen] OpenLineage COMPLETE event emitted.")
        except Exception as exc:
            log.warning("[xbrl_gen] OpenLineage emit failed (non-fatal): %s", exc)

    def output_check(self) -> None:
        """Verify the XBRL file exists and is well-formed XML."""
        out_path = getattr(self, "_instance_path", None)
        if not out_path or not Path(out_path).exists():
            raise RuntimeError("[xbrl_gen] Output XBRL file not found.")
        try:
            ET.parse(str(out_path))
            log.info("[xbrl_gen] Output XBRL is well-formed XML: %s", out_path)
        except ET.ParseError as exc:
            raise RuntimeError(f"[xbrl_gen] Output XBRL is malformed XML: {exc}")

# Run just the XBRL generator
python pipeline.py --module xbrl_gen

# Expected output
INFO [xbrl_gen] Mapping file present. Output dir: output/xbrl
INFO [xbrl_gen] Mart data fetched for templates: ['c0100', 'c0200', 'c0300', 'c4700']
INFO [xbrl_gen] Instance written: output/xbrl/COREP_2026-03-31_20260507T081433Z.xbrl (26 facts)
INFO [xbrl_gen] Uploaded → minio://corep-xbrl-output/COREP_2026-03-31_20260507T081433Z.xbrl
INFO [xbrl_gen] OpenLineage COMPLETE event emitted.
INFO [xbrl_gen] Output XBRL is well-formed XML.

# Inspect the generated file
cat output/xbrl/COREP_2026-03-31_*.xbrl | head -60

# Quick fact count check
grep -c "<ei:" output/xbrl/COREP_2026-03-31_*.xbrl
# Expected: 26 (matches the 26 data points in mart_to_xbrl_mapping.yaml)