XBRL Formula Validation — Why Your COREP Numbers Must Add Up Across Templates

# xbrl/validate_instance.py — standalone validation function
# Used by both xbrl_valid.py module and manual testing

from arelle import Cntlr, ModelManager
from arelle.validate import ValidateXbrl
from arelle.ValidateXbrlCalcs import ValidateXbrlCalcs
import logging, os
from pathlib import Path
from dataclasses import dataclass, field
from typing import List

log = logging.getLogger(__name__)

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",
)

# Arelle message severity levels
SEVERITY_ERROR   = "ERROR"
SEVERITY_WARNING = "WARNING"
SEVERITY_INFO    = "INFO"


@dataclass
class ValidationMessage:
    severity:   str
    code:       str    # EBA rule code e.g. "xbrlCalcs:inconsistency"
    message:    str
    template:   str    # e.g. "C 01.00" — extracted from message if possible
    concept:    str    # e.g. "ei:c0010"
    value:      str    # reported value
    expected:   str    # computed expected value (for calc errors)


@dataclass
class ValidationResult:
    passed:       bool
    error_count:  int
    warn_count:   int
    messages:     List[ValidationMessage] = field(default_factory=list)

    @property
    def errors(self):
        return [m for m in self.messages if m.severity == SEVERITY_ERROR]

    @property
    def warnings(self):
        return [m for m in self.messages if m.severity == SEVERITY_WARNING]


def validate_xbrl_instance(instance_path: Path) -> ValidationResult:
    """
    Run full XBRL validation on an instance document:
      1. Structural (schema conformance)
      2. Calculation linkbase (summation consistency)
      3. Formula linkbase (EBA business rules — requires formula plugin)

    Returns ValidationResult with all messages categorised.
    """
    messages: List[ValidationMessage] = []

    # ── Custom log handler to capture Arelle messages ─────────────
    class _ArelleLogHandler(logging.Handler):
        def emit(self, record):
            msg_text  = self.format(record)
            severity  = SEVERITY_ERROR   if record.levelno >= 40 else \
                        SEVERITY_WARNING if record.levelno >= 30 else \
                        SEVERITY_INFO
            code = getattr(record, "messageCode", "")
            messages.append(ValidationMessage(
                severity=severity,
                code=code,
                message=msg_text,
                template=_extract_template(msg_text),
                concept=_extract_concept(msg_text),
                value=_extract_value(msg_text, "reported"),
                expected=_extract_value(msg_text, "expected"),
            ))

    log_handler = _ArelleLogHandler()
    arelle_logger = logging.getLogger("arelle")
    arelle_logger.addHandler(log_handler)

    # ── Run Arelle ────────────────────────────────────────────────
    cntlr    = Cntlr.Cntlr()
    modelMgr = ModelManager.ModelManager(cntlr)

    try:
        log.info("[xbrl_valid] Loading taxonomy...")
        modelXbrl = modelMgr.load(TAXONOMY_ENTRY)

        log.info("[xbrl_valid] Loading instance: %s", instance_path)
        instance  = modelMgr.load(str(instance_path), modelXbrl=modelXbrl)

        # Step 1: Structural XBRL validation
        log.info("[xbrl_valid] Running structural validation...")
        ValidateXbrl.ValidateXbrl(modelMgr).validate(instance)

        # Step 2: Calculation linkbase validation
        log.info("[xbrl_valid] Running calculation validation...")
        ValidateXbrlCalcs(instance).validate()

        # Step 3: Formula linkbase (EBA business rules)
        # Requires Arelle formula plugin — loaded via cntlr if installed
        try:
            from arelle.plugin import formulaXbrl
            formulaXbrl.run(instance)
            log.info("[xbrl_valid] Formula validation complete.")
        except ImportError:
            log.warning("[xbrl_valid] Formula plugin not available — skipping EBA business rules.")

    finally:
        modelMgr.close()
        cntlr.close()
        arelle_logger.removeHandler(log_handler)

    error_count = sum(1 for m in messages if m.severity == SEVERITY_ERROR)
    warn_count  = sum(1 for m in messages if m.severity == SEVERITY_WARNING)

    return ValidationResult(
        passed=error_count == 0,
        error_count=error_count,
        warn_count=warn_count,
        messages=messages,
    )


def _extract_template(msg: str) -> str:
    import re
    m = re.search(rr"C \d{2}\.\d{2}", msg)
    return m.group(0) if m else ""

def _extract_concept(msg: str) -> str:
    import re
    m = re.search(rr"(?:ei:|c)\d{4}", msg)
    return m.group(0) if m else ""

def _extract_value(msg: str, label: str) -> str:
    import re
    m = re.search(rfr"{label}[:\s=]+([0-9.\-]+)", msg, re.IGNORECASE)
    return m.group(1) if m else ""

"""
modules/xbrl_valid.py — Validate the generated XBRL instance against the EBA taxonomy.

Runs three validation passes:
  1. Structural (schema conformance)
  2. Calculation linkbase (summation consistency)
  3. Formula linkbase (EBA business rules — v-codes)

On any ERROR-level finding, raises XbrlValidationError.
Warnings are logged but do not halt the pipeline.
Validation report is uploaded to MinIO as audit evidence.
"""

import json, logging, os
from datetime import datetime, timezone
from pathlib import Path

from modules.base import BaseModule
from xbrl.validate_instance import validate_xbrl_instance, ValidationResult

log = logging.getLogger(__name__)

XBRL_OUTPUT_DIR = Path(os.environ.get("XBRL_OUTPUT_DIR", "output/xbrl"))
REPORT_DIR      = Path(os.environ.get("XBRL_REPORT_DIR", "output/validation_reports"))


class XbrlValidationError(RuntimeError):
    """Raised when the XBRL instance fails EBA validation rules."""
    pass


class XbrlValidModule(BaseModule):
    MODULE_NAME = "xbrl_valid"

    def input_check(self) -> None:
        """Verify at least one XBRL file exists in the output directory."""
        xbrl_files = list(XBRL_OUTPUT_DIR.glob("*.xbrl"))
        if not xbrl_files:
            raise RuntimeError(
                f"[xbrl_valid] No .xbrl files found in {XBRL_OUTPUT_DIR}. "
                "Run xbrl_gen first: python pipeline.py --module xbrl_gen"
            )
        # Validate the most recently generated file
        self._instance_path = max(xbrl_files, key=lambda p: p.stat().st_mtime)
        log.info("[xbrl_valid] Target instance: %s", self._instance_path)
        REPORT_DIR.mkdir(parents=True, exist_ok=True)

    def _execute(self) -> None:
        """Run full Arelle validation and raise on any ERROR-level finding."""
        log.info("[xbrl_valid] Starting validation: %s", self._instance_path)
        result: ValidationResult = validate_xbrl_instance(self._instance_path)

        # ── Log summary ──────────────────────────────────────────────
        log.info(
            "[xbrl_valid] Validation complete: %s | errors=%d | warnings=%d",
            "PASS" if result.passed else "FAIL",
            result.error_count,
            result.warn_count,
        )

        # ── Log individual errors ─────────────────────────────────────
        for msg in result.errors:
            log.error(
                "[xbrl_valid] ERROR [%s] template=%s concept=%s | %s",
                msg.code, msg.template, msg.concept, msg.message
            )

        # ── Log warnings (informational — do not halt) ────────────────
        for msg in result.warnings:
            log.warning(
                "[xbrl_valid] WARN  [%s] %s", msg.code, msg.message
            )

        # ── Write JSON validation report ──────────────────────────────
        ts          = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
        report_path = REPORT_DIR / f"validation_report_{ts}.json"
        report_data = {
            "instance":     str(self._instance_path),
            "validated_at": ts,
            "passed":       result.passed,
            "error_count":  result.error_count,
            "warn_count":   result.warn_count,
            "errors": [
                {"code": m.code, "template": m.template,
                 "concept": m.concept, "reported": m.value,
                 "expected": m.expected, "message": m.message}
                for m in result.errors
            ],
            "warnings": [
                {"code": m.code, "message": m.message}
                for m in result.warnings
            ],
        }
        report_path.write_text(json.dumps(report_data, indent=2))
        log.info("[xbrl_valid] Validation report written: %s", report_path)
        self._report_path = report_path

        self._upload_report_to_minio(report_path)
        self._write_audit(report_data)

        # ── Raise on errors ───────────────────────────────────────────
        if not result.passed:
            error_summary = "; ".join(
                f"{m.code}({m.template})" for m in result.errors[:5]
            )
            raise XbrlValidationError(
                f"[xbrl_valid] {result.error_count} EBA validation error(s): {error_summary}. "
                f"See report: {report_path}"
            )

    def _upload_report_to_minio(self, report_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,
                f"validation_reports/{report_path.name}",
                str(report_path),
                content_type="application/json",
            )
            log.info("[xbrl_valid] Report uploaded → minio://%s/validation_reports/%s", bucket, report_path.name)
        except Exception as exc:
            log.warning("[xbrl_valid] MinIO upload failed (non-fatal): %s", exc)

    def _write_audit(self, report_data: dict) -> None:
        import json
        from modules.base import _pg_conn
        conn = _pg_conn()
        try:
            cur = conn.cursor()
            cur.execute(
                """
                INSERT INTO audit.pipeline_run_log
                    (run_id, module_name, status, metadata, ran_at)
                VALUES (%s, 'xbrl_valid', %s, %s, now())
                """,
                (
                    self._run_id,
                    "PASS" if report_data["passed"] else "FAIL",
                    json.dumps({
                        "error_count": report_data["error_count"],
                        "warn_count":  report_data["warn_count"],
                        "instance":    report_data["instance"],
                        "report":      str(self._report_path),
                    }),
                ),
            )
            conn.commit()
        finally:
            conn.close()

    def emit_lineage(self) -> None:
        # Validation is read-only — input is the XBRL file, no new data written
        log.info("[xbrl_valid] No lineage event — validation is read-only.")

    def output_check(self) -> None:
        """Verify the validation report JSON was written successfully."""
        report_path = getattr(self, "_report_path", None)
        if not report_path or not Path(report_path).exists():
            raise RuntimeError("[xbrl_valid] Validation report file was not written.")
        data = json.loads(Path(report_path).read_text())
        if not data.get("passed"):
            raise RuntimeError(
                f"[xbrl_valid] output_check: validation report shows FAIL with {data.get('error_count')} error(s)."
            )
        log.info("[xbrl_valid] output_check: PASS — %s", report_path)

# Run validation manually (outside the pipeline module)
python pipeline.py --module xbrl_valid

# ── PASS output ──────────────────────────────────────────────────────
INFO  [xbrl_valid] Loading taxonomy...
INFO  [xbrl_valid] Loading instance: output/xbrl/COREP_2026-03-31_20260507T081433Z.xbrl
INFO  [xbrl_valid] Running structural validation...
INFO  [xbrl_valid] Running calculation validation...
INFO  [xbrl_valid] Formula validation complete.
INFO  [xbrl_valid] Validation complete: PASS | errors=0 | warnings=2
WARN  [xbrl_valid] WARN  [xbrlCalcs:insignificantRounding] C 02.00 ei:c0060 ...
INFO  [xbrl_valid] Validation report written: output/validation_reports/validation_report_20260507T081445Z.json
INFO  [xbrl_valid] output_check: PASS

# ── FAIL output — calculation inconsistency ──────────────────────────
ERROR [xbrl_valid] ERROR [xbrlCalcs:inconsistency]
      template=C 01.00 concept=ei:c0010
      Calculation inconsistency in {http://www.eba.europa.eu/xbrl/crr/dict/con}c0010
      reported sum 902000 computed sum 900000
      Difference: 2000 (0.22%)
      Context: C_2026-03-31_instant
# Fix: c0010 in your instance = 902000 but c0020+c0030+c0040 = 900000.
# Check corep_c0100.sql — own_funds column must be the arithmetic sum
# of cet1_capital + at1_capital + t2_capital, not independently computed.

# ── FAIL output — EBA formula rule ───────────────────────────────────
ERROR [xbrl_valid] ERROR [EBA.v4789]
      template=C 03.00
      Formula assertion failed: {ei}c0040 >= {ei}c0050
      tier1_ratio (c0040) = 0.089000 but cet1_ratio (c0050) = 0.112583
# Fix: tier1_ratio < cet1_ratio — mathematically impossible.
# Root cause: corep_c0300.sql reads tier1_ratio from int_capital_by_tier
# which uses a different denominator (total_assets) than cet1_ratio
# which uses total_rwa. Align both ratios to use total_rwa as denominator.

-- Step 1: Find the discrepancy in Trino
-- Compare C 01.00 own_funds against manual sum
SELECT
    own_funds                              AS reported_c0010,
    cet1_capital + at1_capital + t2_capital AS computed_sum,
    own_funds - (cet1_capital + at1_capital + t2_capital) AS diff
FROM mart.corep_c0100;
-- If diff != 0 → fix corep_c0100.sql
-- The own_funds column must be: cet1_capital + at1_capital + t2_capital
-- NOT independently summed from raw.capital_instruments

-- Step 2: Compare C 02.00 row sum against C 03.00 total_rwa
SELECT
    c0200_total.rwa   AS c0200_total_rwa,
    c0300.total_rwa   AS c0300_total_rwa,
    c0200_total.rwa - c0300.total_rwa AS diff
FROM (
    SELECT SUM(rwa) AS rwa
    FROM mart.corep_c0200
    WHERE exposure_class != 'TOTAL'
) c0200_total
CROSS JOIN mart.corep_c0300 c0300;
-- If diff != 0 → fix: c0300.total_rwa must reference int_rwa_by_exposure_class
-- which is the same source as c0200 rows. Use a REF() in dbt, not a duplicate calculation.

-- Fix for EBA.v0001: CET1 ratio must use the EXACT same RWA total as C 02.00
-- In corep_c0300.sql, replace any independent total_rwa calculation with:

WITH rwa_total AS (
    -- This CTE must be IDENTICAL to the TOTAL row computation in corep_c0200.sql
    SELECT SUM(rwa) AS total_rwa
    FROM {{ ref('int_rwa_by_exposure_class') }}
),
capital AS (
    SELECT cet1_capital, at1_capital, t2_capital, own_funds
    FROM {{ ref('corep_c0100') }}
)
SELECT
    capital.own_funds,
    capital.cet1_capital,
    rwa_total.total_rwa,
    -- Ratios: use NULLIF to prevent division by zero, ROUND to 6dp
    ROUND(capital.cet1_capital::numeric / NULLIF(rwa_total.total_rwa, 0), 6) AS cet1_ratio,
    ROUND((capital.cet1_capital + capital.at1_capital)::numeric / NULLIF(rwa_total.total_rwa, 0), 6) AS tier1_ratio,
    ROUND(capital.own_funds::numeric / NULLIF(rwa_total.total_rwa, 0), 6) AS total_capital_ratio
FROM rwa_total
CROSS JOIN capital;

-- Key points:
-- 1. rwa_total CTE uses int_rwa_by_exposure_class — the SAME model as corep_c0200
-- 2. No ROUND() on intermediate values — only terminal ROUND on the ratio
-- 3. NULLIF(rwa_total, 0) prevents divide-by-zero producing NULL instead of error
-- 4. tier1_ratio = (cet1 + at1) / rwa — not independently summed from instruments

# dags/corep_pipeline_dag.py — validation branch logic

def _xbrl_validation_branch(**context) -> str:
    """Return next task based on XBRL validation result stored in XCom."""
    validation_status = context["task_instance"].xcom_pull(
        task_ids="run_xbrl_validation", key="validation_status"
    )
    if validation_status == "PASS":
        return "prepare_submission_package"
    return "quarantine_failed_xbrl"   # archives the XBRL + report to MinIO/quarantine/


def _run_xbrl_validation(**context) -> None:
    """Run xbrl_valid module and push status to XCom."""
    from modules.xbrl_valid import XbrlValidModule, XbrlValidationError
    ti = context["task_instance"]
    try:
        mod = XbrlValidModule()
        mod.run()
        ti.xcom_push(key="validation_status", value="PASS")
    except XbrlValidationError as exc:
        log.error("XBRL validation FAILED: %s", exc)
        ti.xcom_push(key="validation_status", value="FAIL")
        # Do not re-raise — let BranchPythonOperator handle routing


run_xbrl_validation = PythonOperator(
    task_id="run_xbrl_validation",
    python_callable=_run_xbrl_validation,
)

branch_on_xbrl_validation = BranchPythonOperator(
    task_id="branch_on_xbrl_validation",
    python_callable=_xbrl_validation_branch,
)

# Full pipeline DAG chain at this point:
(
    run_ingest
    >> run_quality_layer1
    >> run_dbt_staging >> run_dbt_intermediate >> run_dbt_mart
    >> run_quality_layer2
    >> run_catalog
    >> run_security
    >> run_xbrl_gen
    >> run_xbrl_validation         # ← Day 12
    >> branch_on_xbrl_validation
)
branch_on_xbrl_validation >> prepare_submission_package  # Day 13
branch_on_xbrl_validation >> quarantine_failed_xbrl