Feat/1430 authoritative overlay

nodes/src/nodes/authoritative_overlay/IGlobal.py

@@ -0,0 +1,38 @@
+# =============================================================================
+# MIT License
+# Copyright (c) 2026 Aparavi Software AG
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+# =============================================================================
+
+from rocketlib import IGlobalBase, debug, warning
+from ai.common.config import Config
+import json
+
+class IGlobal(IGlobalBase):
+    def __init__(self):
+        super().__init__()
+        self.regulator_type = 'sec'
+
+    def beginGlobal(self):
+        """Initialize the global authoritative overlay configuration."""
+        raw = Config.getNodeConfig(self.glb.logicalType, self.glb.connConfig) or {}
+        self.regulator_type = str(raw.get('regulator_type', 'sec')).strip()
+        self.cik = str(raw.get('cik', '')).strip().zfill(10)
+        debug(f'Initialized Authoritative Overlay with regulator: {self.regulator_type}, CIK: {self.cik}')

nodes/src/nodes/authoritative_overlay/IInstance.py

@@ -0,0 +1,175 @@
+# =============================================================================
+# MIT License
+# Copyright (c) 2026 Aparavi Software AG
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+# =============================================================================
+
+from rocketlib import IInstanceBase, Entry, warning, debug
+from ai.common.schema import Answer
+from .IGlobal import IGlobal
+
+from .connectors.sec import query_sec
+
+import re
+import json
+import math
+
+def _normalize_number(value_str: str) -> float | None:
+    """Strip currency symbols, commas, and spaces, then parse to float.
+    Handles parenthesized negatives and scale suffixes (M, k, B, in thousands).
+    """
+    clean_str = value_str.strip().lower()
+
+    # Handle "in thousands", "in millions", etc.
+    scale = 1.0
+    if 'in thousands' in clean_str:
+        scale = 1_000.0
+        clean_str = clean_str.replace('in thousands', '')
+    elif 'in millions' in clean_str:
+        scale = 1_000_000.0
+        clean_str = clean_str.replace('in millions', '')
+    elif 'in billions' in clean_str:
+        scale = 1_000_000_000.0
+        clean_str = clean_str.replace('in billions', '')
+
+    # Remove currency, commas, and spaces
+    clean_str = re.sub(r'[\$€£¥,\s]', '', clean_str)
+
+    # Handle parenthesized negatives: (1.5) -> -1.5
+    if clean_str.startswith('(') and clean_str.endswith(')'):
+        clean_str = '-' + clean_str[1:-1]
+
+    # Handle k, m, b suffixes
+    if clean_str.endswith('k'):
+        scale *= 1_000.0
+        clean_str = clean_str[:-1]
+    elif clean_str.endswith('m'):
+        scale *= 1_000_000.0
+        clean_str = clean_str[:-1]
+    elif clean_str.endswith('b'):
+        scale *= 1_000_000_000.0
+        clean_str = clean_str[:-1]
+
+    try:
+        return float(clean_str) * scale
+    except ValueError:
+        return None
+
+class IInstance(IInstanceBase):
+    IGlobal: IGlobal
+
+    def __init__(self):
+        super().__init__()
+
+    def writeText(self, text: str):
+        """Handle raw text input (used by the test framework).
+
+        The test framework sends data on the ``text`` lane as a plain
+        string.  This handler parses the JSON payload, constructs a
+        proper ``Answer`` object, and delegates to ``writeAnswers``.
+        This serves as the main entry point for the text lane.
+        """
+        text = text.strip()
+        if not text:
+            warning('Abstaining: empty text received')
+            self.preventDefault()
+            return
+
+        # Build a real Answer so the rest of the pipeline sees the same type
+        answer = Answer()
+        answer.setAnswer(text)
+        self.writeAnswers(answer)
+
+    def writeAnswers(self, answer: Answer):
+        """Run authoritative cross-check on the answer.
+
+        Extracts the answer text and attempts to verify it against the
+        configured regulator database. If the official data does not match,
+        the node abstains (drops the answer).
+        """
+        regulator_type = self.IGlobal.regulator_type
+
+        # We expect a JSON answer with a 'concept' and a 'value'
+        try:
+            payload = None
+            text_val = answer.getText()
+
+            if answer.isJson():
+                payload = answer.getJson()
+            else:
+                if isinstance(text_val, dict):
+                    payload = text_val
+                else:
+                    payload = json.loads(text_val)
+
+            if not isinstance(payload, dict):
+                raise ValueError(f'Payload is not a dictionary, it is {type(payload)}')
+
+            concept = str(payload.get('concept', ''))
+            text = str(payload.get('value', ''))
+        except Exception as e:
+            warning(f'Abstaining: Expected JSON answer with \'concept\' and \'value\'. Error: {e}')
+            self.preventDefault()
+            return
+
+        text = text.strip()
+        concept = concept.strip()
+
+        if not text or not concept:
+            warning('Abstaining: Missing concept or value in answer')
+            self.preventDefault()
+            return
+
+        normalized_text = _normalize_number(text)
+        if normalized_text is None:
+            warning(f"Abstaining: Could not normalize extracted text '{text}' into a number.")
+            self.preventDefault()
+            return
+
+        # Query the appropriate regulator for the official number
+        official_data = None
+
+        try:
+            if regulator_type == 'sec':
+                official_data = query_sec(concept, text, cik=self.IGlobal.cik)
+            else:
+                warning(f"Unknown regulator type: {regulator_type}")
+                # Fail closed: abstain rather than forward an unverified answer
+                self.preventDefault()
+                return
+        except Exception as e:
+            warning(f"Failed to query {regulator_type} connector: {str(e)}")
+            self.preventDefault()
+            return
+
+        if not official_data:
+            warning(f"Abstaining: Extracted value '{text}' not found in {regulator_type} authoritative data.")
+            self.preventDefault()
+            return
+
+        # official_data is now expected to be a list of floats
+        if not any(math.isclose(normalized_text, v, rel_tol=1e-9, abs_tol=1e-6) for v in official_data):
+            warning(f"Abstaining: Value mismatch. Extracted '{text}' (normalized: {normalized_text}) does not match official data from {regulator_type}.")
+            self.preventDefault()
+            return
+
+        debug(f"Authoritative Match: '{text}' verified against {regulator_type} database.")
+        # Forward the answer downstream
+        self.instance.writeAnswers(answer)

nodes/src/nodes/authoritative_overlay/README.md

@@ -0,0 +1,27 @@
+# Authoritative Overlay Node
+
+Part of the **Audit-grade financial extraction node suite**.
+
+## Overview
+The `authoritative_overlay` node is a pipeline filter designed to cross-check extracted financial numbers against official regulator data. 
+
+Currently supported regulators include:
+- US SEC (EDGAR)
+
+## Behavior
+
+- **Input**: Pipeline `answers` (extracted financial data/numbers).
+- **Process**: The node queries the live SEC EDGAR API dynamically using the configured company CIK for the official recorded value.
+- **Output**: 
+  - If the extracted number matches the official data, the answer is forwarded downstream.
+  - If there is a mismatch or no record is found, the node **abstains** by dropping the answer and logging a warning.
+
+> [!NOTE]
+> **Strict Matching**: The node uses exact matching (`math.isclose` with a high strictness tolerance `rel_tol=1e-9`). This ensures that only exact financial figures are passed through. Filings restated to the nearest thousand will not match a value extracted to the unit.
+
+## Configuration
+
+See `services.json` for node configuration schemas.
+
+<!-- ROCKETRIDE:GENERATED:PARAMS START -->
+<!-- ROCKETRIDE:GENERATED:PARAMS END -->

nodes/src/nodes/authoritative_overlay/__init__.py

@@ -0,0 +1,27 @@
+# =============================================================================
+# MIT License
+# Copyright (c) 2026 Aparavi Software AG
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+# =============================================================================
+
+from .IGlobal import IGlobal
+from .IInstance import IInstance
+
+__all__ = ['IGlobal', 'IInstance']

nodes/src/nodes/authoritative_overlay/connectors/__init__.py

@@ -0,0 +1,50 @@
+# =============================================================================
+# MIT License
+# Copyright (c) 2026 Aparavi Software AG
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+# =============================================================================
+
+# Connectors for authoritative overlay
+
+import json
+import os
+from rocketlib import debug
+
+def _load_statements_snapshot(concept: str, snapshot_filename: str, log_name: str) -> list[float] | None:
+    """Helper to load a specific concept from a statements snapshot."""
+    snapshot_path = os.path.join(os.path.dirname(__file__), '..', '..', '..', '..', 'test', 'authoritative_overlay', 'testdata', snapshot_filename)
+    try:
+        with open(snapshot_path, 'r', encoding='utf-8') as f:
+            data = json.load(f)
+            values = []
+
+            # Scope the lookup to the requested concept
+            statements = data.get('statements', {})
+            val = statements.get(concept)
+
+            if val is not None:
+                try:
+                    values.append(float(val))
+                except (ValueError, TypeError):
+                    pass
+            return values
+    except Exception as e:
+        debug(f'{log_name} snapshot not available: {e}')
+        return None

nodes/src/nodes/authoritative_overlay/connectors/sec.py

@@ -0,0 +1,64 @@
+# =============================================================================
+# MIT License
+# Copyright (c) 2026 Aparavi Software AG
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+# =============================================================================
+
+import json
+import requests
+from rocketlib import debug, warning
+
+def query_sec(concept: str, extracted_text: str, cik: str):
+    """
+    Queries the US SEC EDGAR API for a specific concept within us-gaap.
+    Returns a list of values found, or None if the query fails.
+    """
+    if not cik:
+        warning('SEC EDGAR requires a CIK.')
+        return None
+
+    url = f"https://data.sec.gov/api/xbrl/companyconcept/CIK{cik}/us-gaap/{concept}.json"
+    headers = {
+        'User-Agent': 'RocketRide-Authoritative-Overlay/1.0'
+    }
+
+    try:
+        response = requests.get(url, headers=headers, timeout=10)
+
+        if response.status_code == 404:
+            debug(f'US SEC concept {concept} not found for CIK {cik}')
+            return None
+
+        response.raise_for_status()
+        data = response.json()
+
+        values = []
+        for unit, measurements in data.get('units', {}).items():
+            for measurement in measurements:
+                val = measurement.get('val')
+                if val is not None:
+                    try:
+                        values.append(float(val))
+                    except (ValueError, TypeError):
+                        pass
+        return values
+    except requests.exceptions.RequestException as e:
+        warning(f'US SEC API query failed: {str(e)}')
+        return None

nodes/src/nodes/authoritative_overlay/requirements.txt

@@ -0,0 +1,2 @@
+# Authoritative Overlay Dependencies
+# Standard libraries are used for snapshot parsing, no extra pip dependencies required at this time.