Blame for core/quantitative_engine.py

# -*- coding: utf-8 -*-
"""Quantitative evidence preparation for SmartPaper's AMJ-oriented track."""

from __future__ import annotations

import json
import logging
import re

import numpy as np
import pandas as pd

from i18n import tr
from utils.file_parser import extract_data_from_files


class QuantitativeEngine:
    """Prepare one frozen quantitative evidence packet before manuscript writing."""

    QUANTITATIVE_EVIDENCE_LEDGER_VERSION = "quantitative-evidence-ledger-v2"

    @staticmethod
    def generate_psd_corr_matrix(n_constructs, max_corr=0.5):
        if n_constructs == 1:
            return np.array([[1.0]])
        base_corr = np.random.uniform(0.2, max_corr - 0.1)
        matrix = np.random.uniform(
            base_corr - 0.1,
            base_corr + 0.1,
            (n_constructs, n_constructs),
        )
        matrix = (matrix + matrix.T) / 2
        np.fill_diagonal(matrix, 1.0)
        min_eig = np.min(np.linalg.eigvals(matrix))
        if min_eig < 0:
            matrix -= (min_eig - 0.01) * np.eye(n_constructs)
        diagonal = np.sqrt(np.diag(matrix))
        return matrix / np.outer(diagonal, diagonal)

    @staticmethod
    def calculate_validity_metrics(df, constructs_info, true_loadings):
        metrics = []
        for construct in constructs_info:
            construct_name = construct["name"]
            items = construct["items"]
            sub_df = df[items]
            var_total = sub_df.sum(axis=1).var(ddof=1)
            sum_var = sub_df.var(ddof=1).sum()
            n_items = len(items)
            alpha = (
                0.0
                if var_total <= 0
                else (n_items / (n_items - 1))
                * (1 - sum_var / var_total)
            )
            loadings = np.array(true_loadings[construct_name])
            sum_loadings_sq = np.sum(loadings) ** 2
            sum_sq_loadings = np.sum(loadings**2)
            sum_error = np.sum(1 - loadings**2)
            cr = sum_loadings_sq / (sum_loadings_sq + sum_error)
            ave = sum_sq_loadings / len(loadings)
            metrics.append(
                {
                    "Construct": construct_name,
                    "Items": n_items,
                    "Cronbach's α": round(alpha, 3),
                    "CR": round(cr, 3),
                    "AVE": round(ave, 3),
                }
            )
        return pd.DataFrame(metrics)

    @staticmethod
    def _safe_constructs(values):
        cleaned = []
        for value in values or []:
            name = re.sub(r"\s+", " ", str(value or "")).strip(" .,:;-_\n\t")
            if not name or len(name) > 80 or name.lower() in {item.lower() for item in cleaned}:
                continue
            cleaned.append(name)
        return cleaned[:3]

    def _derive_system_design(self, agent, topic="", keywords=""):
        """Freeze a parsimonious topic-aligned design without inventing named sites."""
        fallback = {
            "constructs": [
                "Focal Predictor",
                "Focal Mediator",
                "Focal Outcome",
            ],
            "research_context": "respondents with experience relevant to the focal management phenomenon",
        }
        if agent is None:
            return fallback
        prompt = f"""Design the minimum coherent quantitative management-study skeleton for the topic below.
Return ONLY valid JSON:
{{"constructs":["predictor construct","mediating construct","outcome construct"],"research_context":"one generic respondent context"}}
Rules:
1. Exactly three distinct constructs in predictor -> mediator -> outcome order.
2. Use concise construct labels that can appear throughout a journal article.
3. The context must be generic and directly relevant to the topic. Do NOT name or invent a country, university, firm, industry sample frame, software package, time period, or institution.
4. This is a cross-sectional single-wave survey design, so do not imply experimental or causal identification.
5. Do not create any number; sample size and scale are frozen elsewhere.
Topic: {topic}
Keywords: {keywords}"""
        try:
            raw = agent.call_deepseek(
                "You are a rigorous management research-design editor.",
                prompt,
                max_tokens=500,
                temp=0.0,
                model_role="fast",
                thinking=False,
                allow_fallback=True,
            )
            parsed = agent._extract_json_from_text(raw)
            if not isinstance(parsed, dict):
                return fallback
            constructs = self._safe_constructs(parsed.get("constructs"))
            context = re.sub(r"\s+", " ", str(parsed.get("research_context") or "")).strip()
            if len(constructs) != 3:
                return fallback
            if not context or len(context) > 240:
                context = fallback["research_context"]
            # Named-location/site details are deliberately not accepted from this
            # design step. The manuscript may remain generic when no user evidence
            # supplies a real setting.
            if re.search(r"\b(university|company|corporation|inc\.?|ltd\.?|country|china|usa|united states|europe|asia)\b", context, re.I):
                context = fallback["research_context"]
            return {"constructs": constructs, "research_context": context}
        except Exception as exc:
            logging.warning("Quantitative study-design freeze fell back safely: %s", exc)
            return fallback

    def generate_empirical_data_robust(
        self,
        construct_names=None,
        n=300,
        scale_points=7,
    ):
        if not construct_names or len(construct_names) < 2:
            construct_names = [
                "Focal Predictor",
                "Focal Mediator",
                "Focal Outcome",
            ]

        n_constructs = len(construct_names)
        corr = self.generate_psd_corr_matrix(n_constructs, max_corr=0.55)
        latent_scores = np.random.multivariate_normal(
            mean=[0] * n_constructs,
            cov=corr,
            size=n,
        )
        data = {}
        true_loadings = {}
        constructs_info = []
        target_mean = scale_points / 2.0 + 0.5
        target_std = scale_points / 5.0

        for index, construct_name in enumerate(construct_names):
            true_score = latent_scores[:, index]
            n_items = np.random.randint(3, 6)
            loadings = np.random.uniform(0.72, 0.88, n_items)
            true_loadings[construct_name] = loadings
            item_names = []
            safe_prefix = re.sub(r"[^A-Za-z0-9]+", "_", construct_name).strip("_") or f"Construct_{index + 1}"
            for item_index, loading in enumerate(loadings):
                error = np.random.normal(
                    0,
                    np.sqrt(1 - loading**2),
                    n,
                )
                item_score = loading * true_score + error
                raw_scaled = item_score * target_std + target_mean
                item_int = np.clip(
                    np.round(raw_scaled),
                    1,
                    scale_points,
                ).astype(int)
                column_name = f"{safe_prefix}_{item_index + 1}"
                data[column_name] = item_int
                item_names.append(column_name)
            constructs_info.append(
                {"name": construct_name, "items": item_names}
            )

        df = pd.DataFrame(data)
        validity_df = self.calculate_validity_metrics(
            df,
            constructs_info,
            true_loadings,
        )
        mean_scores = pd.DataFrame(
            {
                construct["name"]: df[construct["items"]].mean(axis=1)
                for construct in constructs_info
            }
        )
        corr_matrix = mean_scores.corr()
        fl_matrix = corr_matrix.astype(object)

        for _, row in validity_df.iterrows():
            construct_name = row["Construct"]
            sqrt_ave = np.sqrt(row["AVE"])
            fl_matrix.loc[construct_name, construct_name] = (
                f"({sqrt_ave:.3f})"
            )

        for row_index in range(len(fl_matrix.columns)):
            for column_index in range(row_index + 1, len(fl_matrix.columns)):
                fl_matrix.iloc[row_index, column_index] = ""
            for column_index in range(row_index):
                value = fl_matrix.iloc[row_index, column_index]
                if isinstance(value, (int, float, np.integer, np.floating)):
                    fl_matrix.iloc[row_index, column_index] = f"{float(value):.3f}"

        desc_stats = mean_scores.mean().to_frame("Mean").join(
            mean_scores.std().to_frame("SD")
        )
        return df, validity_df, desc_stats.join(fl_matrix), mean_scores, corr_matrix

    @staticmethod
    def _standardized_ols(y, predictors):
        y = np.asarray(y, dtype=float)
        x = np.column_stack([np.asarray(value, dtype=float) for value in predictors])
        y_sd = y.std(ddof=1)
        x_sd = x.std(axis=0, ddof=1)
        if y_sd <= 0 or np.any(x_sd <= 0):
            raise RuntimeError("quantitative_zero_variance")
        yz = (y - y.mean()) / y_sd
        xz = (x - x.mean(axis=0)) / x_sd
        design = np.column_stack([np.ones(len(yz)), xz])
        coef = np.linalg.lstsq(design, yz, rcond=None)[0]
        fitted = design @ coef
        ss_res = float(np.sum((yz - fitted) ** 2))
        ss_tot = float(np.sum((yz - yz.mean()) ** 2))
        r2 = 0.0 if ss_tot <= 0 else 1.0 - ss_res / ss_tot
        return coef[1:], max(0.0, min(1.0, r2))

    def _path_evidence(self, mean_scores, construct_names, bootstrap_iterations=1500):
        if len(construct_names) < 3:
            return {}, pd.DataFrame(), pd.DataFrame()
        x_name, m_name, y_name = construct_names[0], construct_names[1], construct_names[2]
        x = mean_scores[x_name].to_numpy(dtype=float)
        m = mean_scores[m_name].to_numpy(dtype=float)
        y = mean_scores[y_name].to_numpy(dtype=float)
        a_coef, r2_m = self._standardized_ols(m, [x])
        y_coef, r2_y = self._standardized_ols(y, [x, m])
        a = float(a_coef[0])
        direct = float(y_coef[0])
        b = float(y_coef[1])
        indirect = a * b

        rng = np.random.default_rng(20260809)
        indirect_boot = []
        direct_boot = []
        n = len(mean_scores)
        for _ in range(int(bootstrap_iterations)):
            idx = rng.integers(0, n, n)
            try:
                a_b, _ = self._standardized_ols(m[idx], [x[idx]])
                y_b, _ = self._standardized_ols(y[idx], [x[idx], m[idx]])
                indirect_boot.append(float(a_b[0] * y_b[1]))
                direct_boot.append(float(y_b[0]))
            except Exception:
                continue
        if len(indirect_boot) < 100:
            raise RuntimeError("quantitative_bootstrap_insufficient")
        indirect_ci = np.percentile(indirect_boot, [2.5, 97.5])
        direct_ci = np.percentile(direct_boot, [2.5, 97.5])

        path_table = pd.DataFrame(
            [
                {"Association": f"{x_name} → {m_name}", "Standardized estimate": round(a, 3), "R² of outcome": round(r2_m, 3)},
                {"Association": f"{m_name} → {y_name} (controlling {x_name})", "Standardized estimate": round(b, 3), "R² of outcome": round(r2_y, 3)},
                {"Association": f"{x_name} → {y_name} (controlling {m_name})", "Standardized estimate": round(direct, 3), "R² of outcome": round(r2_y, 3)},
                {"Association": f"Indirect association: {x_name} → {m_name} → {y_name}", "Standardized estimate": round(indirect, 3), "R² of outcome": "—"},
            ]
        )
        bootstrap_table = pd.DataFrame(
            [
                {"Estimate": "Indirect association", "Point estimate": round(indirect, 3), "95% bootstrap CI lower": round(float(indirect_ci[0]), 3), "95% bootstrap CI upper": round(float(indirect_ci[1]), 3)},
                {"Estimate": "Residual direct association", "Point estimate": round(direct, 3), "95% bootstrap CI lower": round(float(direct_ci[0]), 3), "95% bootstrap CI upper": round(float(direct_ci[1]), 3)},
            ]
        )
        x_m_corr = float(np.corrcoef(x, m)[0, 1])
        vif = 1.0 / max(1e-9, 1.0 - x_m_corr**2)
        spearman = mean_scores.corr(method="spearman")
        summary = {
            "predictor": x_name,
            "mediator": m_name,
            "outcome": y_name,
            "a": round(a, 3),
            "b": round(b, 3),
            "direct": round(direct, 3),
            "indirect": round(indirect, 3),
            "r2_mediator": round(r2_m, 3),
            "r2_outcome": round(r2_y, 3),
            "indirect_ci": [round(float(indirect_ci[0]), 3), round(float(indirect_ci[1]), 3)],
            "direct_ci": [round(float(direct_ci[0]), 3), round(float(direct_ci[1]), 3)],
            "predictor_mediator_vif": round(float(vif), 3),
            "bootstrap_iterations": int(bootstrap_iterations),
        }
        return summary, path_table, spearman


    @staticmethod
    def _sentence_spans(text):
        """Return sentence spans without splitting decimal statistics or citations."""
        value = str(text or "")
        spans = []
        start = 0
        round_depth = 0
        square_depth = 0
        index = 0
        abbreviations = ("e.g.", "i.e.", "et al.", "fig.", "dr.", "mr.", "ms.", "prof.", "vs.", "no.", "vol.")
        while index < len(value):
            char = value[index]
            if char == "(": round_depth += 1
            elif char == ")" and round_depth: round_depth -= 1
            elif char == "[": square_depth += 1
            elif char == "]" and square_depth: square_depth -= 1
            boundary = False
            if char == "\n":
                boundary = True
            elif char in ".!?。！？" and round_depth == 0 and square_depth == 0:
                if char == ".":
                    prev = value[index - 1] if index > 0 else ""
                    nxt = value[index + 1] if index + 1 < len(value) else ""
                    # Keep both ordinary decimals (0.31) and leading-decimal
                    # statistical notation (p < .05) inside one sentence.
                    leading_decimal = bool(
                        nxt.isdigit()
                        and re.search(r"(?:[<>=≤≥]|\b(?:p|r|β))\s*$", value[max(0, index - 12):index], flags=re.I)
                    )
                    decimal = (prev.isdigit() and nxt.isdigit()) or leading_decimal
                    prefix = value[max(0, index - 12): index + 1].lower()
                    abbreviation = any(prefix.endswith(item) for item in abbreviations)
                    boundary = not decimal and not abbreviation
                else:
                    boundary = True
            if boundary:
                end = index + (0 if char == "\n" else 1)
                if end > start and value[start:end].strip():
                    spans.append((start, end, value[start:end]))
                start = index + 1
            index += 1
        if start < len(value) and value[start:].strip():
            spans.append((start, len(value), value[start:]))
        return spans

    @staticmethod
    def _has_external_literature_marker(sentence):
        value = str(sentence or "")
        if re.search(r"\bCITE-\d+\b", value, flags=re.I):
            return True
        if re.search(
            r"\((?=[^()\n]*[A-Za-zÀ-ÖØ-öø-ÿ'’.-]+(?:\s+(?:&\s+[A-Za-zÀ-ÖØ-öø-ÿ'’.-]+|et\s+al\.))?\s*,\s*(?:19|20)\d{2}[a-z]?\b)[^()\n]+\)",
            value, flags=re.I,
        ):
            return True
        if re.search(r"\b[A-Z][A-Za-z'’.-]+\s+et\s+al\.\s*\((?:19|20)\d{2}[a-z]?\)", value):
            return True
        return False

    @classmethod
    def _quantitative_sentence_ownership(cls, sentence, section_name="", known_constructs=None):
        """Return external/study/result/unknown for deterministic AMJ repair.

        Only study/result-owned prose may be rewritten. External literature is
        immutable even when it reports the same constructs or statistics.
        """
        value = str(sentence or "")
        stripped = value.strip()
        if not stripped or stripped.startswith(">"):
            return "external" if stripped else "unknown"
        if cls._has_external_literature_marker(value):
            return "external"
        if re.search(r"(?i)\b(?:prior|previous|earlier|existing)\s+(?:research|stud(?:y|ies)|literature)\b", value):
            return "external"
        if re.search(r"(?i)\b(?:we|our\s+study|this\s+study|the\s+present\s+study|our\s+sample|our\s+analysis|our\s+results|the\s+sample|the\s+survey|respondents?|participants?)\b", value):
            return "study"
        section = str(section_name or "").casefold()
        constructs = [str(x or "").casefold() for x in (known_constructs or []) if str(x or "").strip()]
        has_construct = any(item in value.casefold() for item in constructs)
        has_metric = bool(re.search(r"(?i)(?:\bVIF\b|\bbootstrap\b|R(?:²|\^2)|(?:β|standardized\s+estimate)\s*(?:=|:)|\bp\s*(?:=|<|>|≤|≥))", value))
        has_role = bool(re.search(r"(?i)\b(?:predictor|mediator|outcome|independent\s+variable|dependent\s+variable)\b", value))
        if any(token in section for token in ("result", "analysis", "abstract", "final")) and (has_construct or has_metric or has_role):
            return "result"
        if any(token in section for token in ("method", "measure")) and (has_construct or has_role or re.search(r"(?i)\b(?:Likert|sample|survey)\b", value)):
            return "study"
        return "unknown"

    @staticmethod
    def _unique_numeric(pattern, evidence, *, integer=False):
        values = []
        for raw in re.findall(pattern, str(evidence or ""), flags=re.I):
            raw_value = raw[-1] if isinstance(raw, tuple) else raw
            try:
                value = int(float(raw_value)) if integer else float(raw_value)
            except Exception:
                continue
            values.append(value)
        unique = sorted(set(values))
        return unique[0] if len(unique) == 1 else None

    @classmethod
    def build_quantitative_evidence_ledger(cls, metadata, evidence_text=""):
        """Freeze only high-confidence study-owned quantitative facts.

        System-generated evidence has typed metadata and therefore supports a
        strict contract. User-provided evidence is parsed conservatively: only
        uniquely labelled values are promoted to deterministic authority.
        Ambiguous values remain owned by the uploaded evidence text and are not
        guessed or globally substituted.
        """
        meta = dict(metadata or {})
        source_type = str(meta.get("source_type") or "")
        sample_size = meta.get("sample_size")
        scale_points = meta.get("scale_points")
        strict = source_type == "system_generated"
        evidence = str(evidence_text or "")
        uploaded_metric_anchors = {}
        if not strict and evidence:
            sample_candidates = set()
            for pattern in (
                r"(?i)\b(?:sample\s+size|sample|N)\s*(?:=|:|of)?\s*(\d{2,6})\b",
                r"(?i)\b(\d{2,6})\s+(?:respondents|participants|observations|cases)\b",
            ):
                sample_candidates.update(int(x) for x in re.findall(pattern, evidence))
            if len(sample_candidates) == 1:
                sample_size = next(iter(sample_candidates))
            scale_candidates = {int(x) for x in re.findall(r"(?i)\b([2-9]|1[0-1])[- ]point\s+Likert", evidence)}
            if len(scale_candidates) == 1:
                scale_points = next(iter(scale_candidates))

            vif = cls._unique_numeric(r"(?i)\bVIF\s*(?:=|:)?\s*([-+]?\d+(?:\.\d+)?)", evidence)
            bootstrap = cls._unique_numeric(r"(?i)\bbootstrap\s+(?:resamples?|iterations?)\s*(?:=|:)?\s*(\d{2,6})", evidence, integer=True)
            if vif is not None:
                uploaded_metric_anchors["VIF"] = vif
            if bootstrap is not None:
                uploaded_metric_anchors["bootstrap_iterations"] = int(bootstrap)

            r2_by_label = {}
            for match in re.finditer(
                r"(?i)R(?:²|\^2)\s*(?:for|of)\s*([A-Za-z][A-Za-z0-9 _\-/]{1,80}?)\s*(?:=|:)\s*([-+]?\d+(?:\.\d+)?)",
                evidence,
            ):
                label = re.sub(r"\s+", " ", match.group(1)).strip().casefold()
                r2_by_label.setdefault(label, set()).add(float(match.group(2)))
            uploaded_metric_anchors["r2_by_label"] = {
                label: next(iter(values)) for label, values in r2_by_label.items() if len(values) == 1
            }

            beta_by_path = {}
            for match in re.finditer(
                r"(?i)([A-Za-z][A-Za-z0-9 _\-/]{1,60}?)\s*(?:→|->)\s*([A-Za-z][A-Za-z0-9 _\-/]{1,60}?)\s*[,;:]?\s*(?:β|standardized\s+estimate)\s*(?:=|:)\s*([-+]?\d+(?:\.\d+)?)",
                evidence,
            ):
                key = (
                    re.sub(r"\s+", " ", match.group(1)).strip().casefold(),
                    re.sub(r"\s+", " ", match.group(2)).strip().casefold(),
                )
                beta_by_path.setdefault(key, set()).add(float(match.group(3)))
            uploaded_metric_anchors["beta_by_path"] = {
                " -> ".join(key): next(iter(values)) for key, values in beta_by_path.items() if len(values) == 1
            }

        constructs = dict(meta.get("constructs") or {}) if isinstance(meta.get("constructs"), dict) else {}
        results = dict(meta.get("association_results") or {}) if isinstance(meta.get("association_results"), dict) else {}
        return {
            "version": cls.QUANTITATIVE_EVIDENCE_LEDGER_VERSION,
            "source_type": source_type,
            "strict": bool(strict),
            "sample_size": int(sample_size) if isinstance(sample_size, (int, np.integer)) or str(sample_size or "").isdigit() else None,
            "scale_points": int(scale_points) if isinstance(scale_points, (int, np.integer)) or str(scale_points or "").isdigit() else None,
            "study_design": str(meta.get("study_design") or "").strip(),
            "constructs": {k: str(v or "").strip() for k, v in constructs.items() if str(v or "").strip()},
            "association_results": results,
            "uploaded_metric_anchors": uploaded_metric_anchors,
            "causal_identification": "none" if strict else "evidence_owned",
            "allowed_methods": [
                "reliability", "composite reliability", "average variance extracted",
                "Fornell-Larcker", "standardized association", "bootstrap",
                "Spearman", "VIF", "descriptive statistics",
            ] if strict else [],
            "forbidden_unowned_methods": [
                "confirmatory factor analysis", "CFA", "structural equation modeling", "SEM",
                "PLS-SEM", "RMSEA", "CFI", "TLI", "SRMR", "Harman",
                "common method bias", "endogeneity", "instrumental variable",
                "fixed effects", "random effects", "difference-in-differences",
            ] if strict else [],
        }

    @staticmethod
    def _causal_phrase_is_explicitly_disclaimed(sentence, match_start, match_end):
        """Return True only when a causal phrase is explicitly scoped by a limitation.

        This is intentionally local to the matched phrase rather than sentence-wide so a
        sentence can safely say that causal identification is unavailable while still
        failing closed if a separate clause makes an affirmative causal claim.
        """
        value = str(sentence or "")
        left_floor = max(0, int(match_start) - 180)
        left = value[left_floor:int(match_start)]
        # Do not let a disclaimer in an earlier independent/contrast clause shield a later
        # affirmative causal assertion.  Keep the closest clause-like segment only.
        boundaries = list(re.finditer(
            r"(?i)(?:[;:,]|\bbut\b|\bhowever\b|\bwhereas\b|\byet\b|\bnevertheless\b)",
            left,
        ))
        if boundaries:
            left = left[boundaries[-1].end():]
        left_folded = re.sub(r"\s+", " ", left.casefold()).strip()
        right = value[int(match_end):min(len(value), int(match_end) + 140)]
        right_folded = re.sub(r"\s+", " ", right.casefold()).strip()

        pre_disclaimer_patterns = (
            r"\b(?:cannot|can't|could not|couldn't|does not|doesn't|do not|don't|did not|didn't)\s+(?:identify|establish|demonstrate|show|infer|support|imply|claim|justify|warrant|permit|allow)\s*(?:(?:an?|the)\s*)?$",
            r"\b(?:is not|isn't|are not|aren't|was not|wasn't|were not|weren't)\s+(?:able\s+to\s+)?(?:identify|establish|demonstrate|show|infer|support|justify|warrant)\s*(?:(?:an?|the)\s*)?$",
            r"\b(?:unable|insufficient)\s+to\s+(?:identify|establish|demonstrate|show|infer|support|justify|warrant)\s*(?:(?:an?|the)\s*)?$",
            r"\b(?:fails?|failed)\s+to\s+(?:identify|establish|demonstrate|show|infer|support|justify|warrant)\s*(?:(?:an?|the)\s*)?$",
            r"\bno\s+(?:evidence|basis|support)\s+(?:for|of)\s*(?:(?:an?|the)\s*)?$",
            r"\bnot\s+(?:evidence|proof)\s+(?:for|of)\s*(?:(?:an?|the)\s*)?$",
            r"\bno\s*$",
        )
        if any(re.search(pattern, left_folded) for pattern in pre_disclaimer_patterns):
            return True

        post_disclaimer_patterns = (
            r"^(?:\s*(?:is|are|was|were|has|have|had))?\s*(?:not|never)\s+(?:identified|established|demonstrated|shown|inferred|supported|claimed|justified|warranted)\b",
            r"^\s*(?:cannot|can't|could not|couldn't|should not|shouldn't|must not|does not|doesn't|do not|don't)\s+(?:be\s+)?(?:identified|established|demonstrated|shown|inferred|supported|claimed|interpreted|justified|warranted)\b",
        )
        return any(re.search(pattern, right_folded) for pattern in post_disclaimer_patterns)

    @classmethod
    def inspect_quantitative_study_fact_contract(cls, text, metadata, section_name=""):
        """Pure AMJ quantitative Evidence Ledger inspection; never rewrites prose."""
        value = str(text or "")
        meta = dict(metadata or {})
        ledger = dict(meta.get("quantitative_evidence_ledger") or {})
        if not ledger:
            ledger = cls.build_quantitative_evidence_ledger(meta)
        sample_size = ledger.get("sample_size")
        scale_points = ledger.get("scale_points")
        strict = bool(ledger.get("strict"))
        constructs = dict(ledger.get("constructs") or {})
        results = dict(ledger.get("association_results") or {})
        uploaded = dict(ledger.get("uploaded_metric_anchors") or {})
        known_constructs = [str(v) for v in constructs.values() if str(v)]
        construct_folded = [c.casefold() for c in known_constructs]
        forbidden = [str(x) for x in ledger.get("forbidden_unowned_methods") or []]
        role_patterns = {
            "predictor": r"(?:predictor|independent\s+variable|predictor\s+construct)",
            "mediator": r"(?:mediator|mediating\s+variable|mediating\s+construct)",
            "outcome": r"(?:outcome|dependent\s+variable|outcome\s+construct)",
        }
        causal_patterns = (
            r"(?i)\bhad\s+a\s+causal\s+effect\s+on\b",
            r"(?i)\bexerted\s+a\s+causal\s+effect\s+on\b",
            r"(?i)\bcausal\s+effect\s+on\b",
            r"(?i)\b(?:an|a|the)\s+causal\s+effect\b",
            r"(?i)\bcausal\s+effects\b",
            r"(?i)\bcausal\s+effect\b",
            r"(?i)\bcausal\s+mediation\b",
        )
        violations = []

        def add_metric_violations(sentence, pattern, expected, *, dtype="float", label="metric"):
            if expected is None:
                return
            regex = re.compile(pattern, re.I)
            for match in regex.finditer(sentence):
                observed = match.group(2)
                try:
                    same = int(float(observed)) == int(expected) if dtype == "int" else abs(float(observed) - float(expected)) < 0.0006
                except Exception:
                    same = False
                if not same:
                    expected_text = str(int(expected)) if dtype == "int" else f"{float(expected):.3f}".rstrip("0").rstrip(".")
                    violations.append({
                        "type":"quantitative_metric_mismatch", "metric":label,
                        "observed":observed, "expected":expected_text,
                        "preview":str(sentence).strip()[:180],
                    })

        for _start, _end, sentence in cls._sentence_spans(value):
            ownership = cls._quantitative_sentence_ownership(sentence, section_name, known_constructs)
            if ownership not in {"study", "result"}:
                continue
            if sample_size:
                for match in re.finditer(r"(?i)(\b(?:sample(?:\s+size)?(?:\s+of)?|N)\s*(?:=|:)?\s*)(\d{2,6})\b", sentence):
                    observed = int(match.group(2))
                    if observed != int(sample_size):
                        violations.append({"type":"sample_size_mismatch","observed":observed,"expected":int(sample_size),"preview":sentence.strip()[:180]})
                for match in re.finditer(r"(?i)\b(\d{2,6})\s+(respondents|participants|observations|cases)\b", sentence):
                    observed = int(match.group(1))
                    if observed != int(sample_size):
                        violations.append({"type":"sample_count_mismatch","observed":observed,"expected":int(sample_size),"preview":sentence.strip()[:180]})
            if scale_points:
                for match in re.finditer(r"(?i)\b([2-9]|1[0-1])[- ]point\s+(Likert(?:-type)?\s+scale)", sentence):
                    observed = int(match.group(1))
                    if observed != int(scale_points):
                        violations.append({"type":"scale_points_mismatch","observed":observed,"expected":int(scale_points),"preview":sentence.strip()[:180]})
            for role, role_pattern in role_patterns.items():
                expected = str(constructs.get(role) or "")
                if not expected:
                    continue
                for candidate in known_constructs:
                    if candidate.casefold() == expected.casefold():
                        continue
                    pattern = re.compile(rf"(?i)({role_pattern}\s*(?:was|is|:|=)?\s*){re.escape(candidate)}")
                    if pattern.search(sentence):
                        violations.append({"type":"construct_role_mismatch","role":role,"observed":candidate,"expected":expected,"preview":sentence.strip()[:180]})
            if strict:
                for pattern in causal_patterns:
                    match = re.search(pattern, sentence)
                    if match and not cls._causal_phrase_is_explicitly_disclaimed(
                        sentence, match.start(), match.end()
                    ):
                        violations.append({"type":"causal_language_violation","preview":sentence.strip()[:180]})
                        break
                if (any(c in sentence.casefold() for c in construct_folded) if construct_folded else False) and re.search(r"(?i)\bp\s*(?:=|<|>|≤|≥)\s*(?:0?\.\d+|\d+(?:\.\d+)?)", sentence):
                    violations.append({"type":"unfrozen_p_value","preview":sentence.strip()[:180]})
                if ownership == "study":
                    bad = next((term for term in forbidden if re.search(rf"(?i)(?<![A-Za-z]){re.escape(term)}(?![A-Za-z])", sentence)), "")
                    if bad:
                        violations.append({"type":"unsupported_quantitative_method","method":bad,"preview":sentence.strip()[:180]})

            vif_expected = results.get("predictor_mediator_vif") if strict else uploaded.get("VIF")
            boot_expected = results.get("bootstrap_iterations") if strict else uploaded.get("bootstrap_iterations")
            add_metric_violations(sentence, r"(\bVIF\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", vif_expected, label="VIF")
            add_metric_violations(sentence, r"(\bbootstrap\s+(?:resamples?|iterations?)\s*(?:=|:)?\s*)(\d{2,6})", boot_expected, dtype="int", label="bootstrap_iterations")
            if strict and results:
                for role, key in (("mediator", "r2_mediator"), ("outcome", "r2_outcome")):
                    construct = str(constructs.get(role) or "")
                    if construct and results.get(key) is not None:
                        add_metric_violations(sentence, rf"(R(?:²|\^2)\s*(?:for|of)?\s*{re.escape(construct)}\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", results.get(key), label=key)
                lower = sentence.casefold()
                selected = None
                specs = [
                    ("indirect", results.get("indirect"), [constructs.get("predictor"), constructs.get("mediator"), constructs.get("outcome")]),
                    ("direct", results.get("direct"), [constructs.get("predictor"), constructs.get("outcome")]),
                    ("a", results.get("a"), [constructs.get("predictor"), constructs.get("mediator")]),
                    ("b", results.get("b"), [constructs.get("mediator"), constructs.get("outcome")]),
                ]
                if "indirect" in lower and all(str(x or "").casefold() in lower for x in specs[0][2] if x):
                    selected = specs[0]
                elif f"controlling {str(constructs.get('mediator') or '').casefold()}" in lower and str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower:
                    selected = specs[1]
                elif f"controlling {str(constructs.get('predictor') or '').casefold()}" in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower:
                    selected = specs[3]
                elif str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() not in lower:
                    selected = specs[2]
                elif str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower and str(constructs.get("predictor") or "").casefold() not in lower:
                    selected = specs[3]
                if selected and selected[1] is not None:
                    add_metric_violations(sentence, r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", selected[1], label=selected[0])
            elif not strict:
                for label, expected in dict(uploaded.get("r2_by_label") or {}).items():
                    if label and label in sentence.casefold():
                        add_metric_violations(sentence, rf"(R(?:²|\^2)\s*(?:for|of)\s*{re.escape(label)}\s*(?:=|:)\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_R2:{label}")
                for path, expected in dict(uploaded.get("beta_by_path") or {}).items():
                    left, _, right = path.partition(" -> ")
                    if left and right and left in sentence.casefold() and right in sentence.casefold():
                        add_metric_violations(sentence, r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_beta:{path}")
        return violations

    @classmethod
    def repair_quantitative_study_fact_contract(cls, text, metadata, section_name=""):
        """Repair deterministic AMJ drift only in study/result-owned prose."""
        value = str(text or "")
        meta = dict(metadata or {})
        ledger = dict(meta.get("quantitative_evidence_ledger") or {})
        if not ledger:
            ledger = cls.build_quantitative_evidence_ledger(meta)
        diagnostics = []
        sample_size = ledger.get("sample_size")
        scale_points = ledger.get("scale_points")
        strict = bool(ledger.get("strict"))
        constructs = dict(ledger.get("constructs") or {})
        results = dict(ledger.get("association_results") or {})
        uploaded = dict(ledger.get("uploaded_metric_anchors") or {})
        known_constructs = [str(v) for v in constructs.values() if str(v)]
        construct_folded = [c.casefold() for c in known_constructs]
        forbidden = [str(x) for x in ledger.get("forbidden_unowned_methods") or []]
        role_patterns = {
            "predictor": r"(?:predictor|independent\s+variable|predictor\s+construct)",
            "mediator": r"(?:mediator|mediating\s+variable|mediating\s+construct)",
            "outcome": r"(?:outcome|dependent\s+variable|outcome\s+construct)",
        }
        causal_replacements = (
            (r"(?i)\bhad\s+a\s+causal\s+effect\s+on\b", "was associated with"),
            (r"(?i)\bexerted\s+a\s+causal\s+effect\s+on\b", "was associated with"),
            (r"(?i)\bcausal\s+effect\s+on\b", "association with"),
            (r"(?i)\ban\s+causal\s+effect\b", "an association"),
            (r"(?i)\ba\s+causal\s+effect\b", "an association"),
            (r"(?i)\bthe\s+causal\s+effect\b", "the association"),
            (r"(?i)\bcausal\s+effects\b", "associations"),
            (r"(?i)\bcausal\s+effect\b", "association"),
            (r"(?i)\bcausal\s+mediation\b", "indirect association"),
        )

        replacements = []
        for start, end, sentence in cls._sentence_spans(value):
            ownership = cls._quantitative_sentence_ownership(sentence, section_name, known_constructs)
            if ownership not in {"study", "result"}:
                continue
            revised = sentence

            if sample_size:
                sample_pattern = re.compile(r"(?i)(\b(?:sample(?:\s+size)?(?:\s+of)?|N)\s*(?:=|:)?\s*)(\d{2,6})\b")
                def sample_repl(match):
                    observed = int(match.group(2))
                    if observed == int(sample_size):
                        return match.group(0)
                    diagnostics.append({"type":"sample_size_repaired","observed":observed,"expected":int(sample_size)})
                    return match.group(1) + str(int(sample_size))
                revised = sample_pattern.sub(sample_repl, revised)
                count_pattern = re.compile(r"(?i)\b(\d{2,6})\s+(respondents|participants|observations|cases)\b")
                def count_repl(match):
                    observed = int(match.group(1))
                    if observed == int(sample_size):
                        return match.group(0)
                    diagnostics.append({"type":"sample_count_repaired","observed":observed,"expected":int(sample_size)})
                    return f"{int(sample_size)} {match.group(2)}"
                revised = count_pattern.sub(count_repl, revised)
            if scale_points:
                scale_pattern = re.compile(r"(?i)\b([2-9]|1[0-1])[- ]point\s+(Likert(?:-type)?\s+scale)")
                def scale_repl(match):
                    observed = int(match.group(1))
                    if observed == int(scale_points):
                        return match.group(0)
                    diagnostics.append({"type":"scale_points_repaired","observed":observed,"expected":int(scale_points)})
                    return f"{int(scale_points)}-point {match.group(2)}"
                revised = scale_pattern.sub(scale_repl, revised)

            for role, role_pattern in role_patterns.items():
                expected = str(constructs.get(role) or "")
                if not expected:
                    continue
                for candidate in known_constructs:
                    if candidate.casefold() == expected.casefold():
                        continue
                    pattern = re.compile(rf"(?i)({role_pattern}\s*(?:was|is|:|=)?\s*){re.escape(candidate)}")
                    revised, count = pattern.subn(lambda m: m.group(1) + expected, revised)
                    if count:
                        diagnostics.append({"type":"construct_role_repaired","role":role,"observed":candidate,"expected":expected,"count":count})

            if strict:
                for pattern, replacement in causal_replacements:
                    regex = re.compile(pattern)
                    repaired_count = 0

                    def causal_repl(match):
                        nonlocal repaired_count
                        if cls._causal_phrase_is_explicitly_disclaimed(
                            revised, match.start(), match.end()
                        ):
                            return match.group(0)
                        repaired_count += 1
                        return replacement

                    revised = regex.sub(causal_repl, revised)
                    if repaired_count:
                        diagnostics.append({
                            "type":"causal_language_repaired",
                            "replacement":replacement,
                            "count":repaired_count,
                        })
                if any(c in revised.casefold() for c in construct_folded) if construct_folded else False:
                    revised, p_count = re.subn(r"(?i)\s*[,;]?\s*p\s*(?:=|<|>|≤|≥)\s*(?:0?\.\d+|\d+(?:\.\d+)?)", "", revised)
                    if p_count:
                        diagnostics.append({"type":"unfrozen_p_value_removed","count":p_count})
                if ownership == "study":
                    bad = next((term for term in forbidden if re.search(rf"(?i)(?<![A-Za-z]){re.escape(term)}(?![A-Za-z])", revised)), "")
                    if bad:
                        diagnostics.append({"type":"unsupported_quantitative_method_removed","method":bad,"preview":sentence.strip()[:180]})
                        revised = ""

            def replace_local_metric(pattern, expected, *, dtype="float", label="metric"):
                nonlocal revised
                if expected is None:
                    return
                expected_text = str(int(expected)) if dtype == "int" else f"{float(expected):.3f}".rstrip("0").rstrip(".")
                regex = re.compile(pattern, re.I)
                def repl(match):
                    observed = match.group(2)
                    try:
                        same = int(float(observed)) == int(expected) if dtype == "int" else abs(float(observed) - float(expected)) < 0.0006
                    except Exception:
                        same = False
                    if same:
                        return match.group(0)
                    diagnostics.append({"type":"quantitative_metric_repaired","metric":label,"observed":observed,"expected":expected_text})
                    return match.group(1) + expected_text
                revised = regex.sub(repl, revised)

            vif_expected = results.get("predictor_mediator_vif") if strict else uploaded.get("VIF")
            boot_expected = results.get("bootstrap_iterations") if strict else uploaded.get("bootstrap_iterations")
            replace_local_metric(r"(\bVIF\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", vif_expected, label="VIF")
            replace_local_metric(r"(\bbootstrap\s+(?:resamples?|iterations?)\s*(?:=|:)?\s*)(\d{2,6})", boot_expected, dtype="int", label="bootstrap_iterations")

            if strict and results:
                for role, key in (("mediator", "r2_mediator"), ("outcome", "r2_outcome")):
                    construct = str(constructs.get(role) or "")
                    if construct and results.get(key) is not None:
                        metric_pattern = rf"(R(?:²|\^2)\s*(?:for|of)?\s*{re.escape(construct)}\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)"
                        replace_local_metric(metric_pattern, results.get(key), label=key)

                lower = revised.casefold()
                selected = None
                specs = [
                    ("indirect", results.get("indirect"), [constructs.get("predictor"), constructs.get("mediator"), constructs.get("outcome")]),
                    ("direct", results.get("direct"), [constructs.get("predictor"), constructs.get("outcome")]),
                    ("a", results.get("a"), [constructs.get("predictor"), constructs.get("mediator")]),
                    ("b", results.get("b"), [constructs.get("mediator"), constructs.get("outcome")]),
                ]
                if "indirect" in lower and all(str(x or "").casefold() in lower for x in specs[0][2] if x):
                    selected = specs[0]
                elif f"controlling {str(constructs.get('mediator') or '').casefold()}" in lower and str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower:
                    selected = specs[1]
                elif f"controlling {str(constructs.get('predictor') or '').casefold()}" in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower:
                    selected = specs[3]
                elif str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() not in lower:
                    selected = specs[2]
                elif str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower and str(constructs.get("predictor") or "").casefold() not in lower:
                    selected = specs[3]
                if selected and selected[1] is not None:
                    replace_local_metric(r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", selected[1], label=selected[0])
            elif not strict:
                r2_map = dict(uploaded.get("r2_by_label") or {})
                for label, expected in r2_map.items():
                    if label and label in revised.casefold():
                        replace_local_metric(rf"(R(?:²|\^2)\s*(?:for|of)\s*{re.escape(label)}\s*(?:=|:)\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_R2:{label}")
                beta_map = dict(uploaded.get("beta_by_path") or {})
                for path, expected in beta_map.items():
                    left, _, right = path.partition(" -> ")
                    if left and right and left in revised.casefold() and right in revised.casefold():
                        replace_local_metric(r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_beta:{path}")

            if revised != sentence:
                replacements.append((start, end, revised))

        for start, end, replacement in sorted(replacements, reverse=True):
            value = value[:start] + replacement + value[end:]
        value = re.sub(r"[ \t]+\n", "\n", value)
        value = re.sub(r"\n{3,}", "\n\n", value)
        value = re.sub(r" {2,}", " ", value)
        return value.strip(), diagnostics

    @classmethod
    def validate_quantitative_study_fact_contract(cls, text, metadata, section_name=""):
        violations = cls.inspect_quantitative_study_fact_contract(text, metadata, section_name=section_name)
        if violations:
            raise RuntimeError(f"quantitative_study_fact_contract_invalid:{section_name}:{violations[:3]}")
        return True

    def resolve_empirical_data(
        self,
        uploaded_files=None,
        language="zh",
        *,
        topic="",
        keywords="",
        agent=None,
    ):
        if uploaded_files:
            user_md = extract_data_from_files(uploaded_files)
            if user_md:
                evidence_sections = [
                    heading.strip()
                    for heading in re.findall(
                        r"(?m)^#{1,3}\s+(.+?)\s*$",
                        user_md,
                    )
                    if heading.strip()
                ]
                metadata = {
                    "mode": "quantitative",
                    "source_type": "user_provided",
                    "disclosure_required": False,
                    "evidence_sections": evidence_sections[:20]
                    or ["User-provided quantitative evidence"],
                    "evidence_boundary": (
                        "Only values explicitly present in the uploaded evidence are frozen. "
                        "Do not infer absent setting, sample, measures, software, or results."
                    ),
                }
                metadata["quantitative_evidence_ledger"] = self.build_quantitative_evidence_ledger(metadata, user_md)
                return None, user_md, tr(language, "actual_stats_parsed"), metadata

        design = self._derive_system_design(agent, topic, keywords)
        construct_names = design["constructs"]
        df, validity_df, fl_matrix_df, mean_scores, corr_matrix = self.generate_empirical_data_robust(
            construct_names=construct_names
        )
        path_summary, path_table, spearman = self._path_evidence(mean_scores, construct_names)
        context = design["research_context"]
        generated_prompt = f"""### Frozen Quantitative Study Design
| Field | Frozen value |
| :--- | :--- |
| Design | Cross-sectional, single-wave questionnaire study |
| Respondent context | {context} |
| Sample size | {len(df)} |
| Response scale | 7-point Likert-type scale |
| Predictor | {construct_names[0]} |
| Mediator | {construct_names[1]} |
| Outcome | {construct_names[2]} |
| Causal identification | None; report associations and estimated indirect association, not causal effects |

### Table 1: Reliability and Convergent Validity
{validity_df.to_markdown(index=False)}

### Table 2: Descriptive Statistics and Discriminant Validity (Fornell-Larcker Criterion)
*Note: Diagonal elements in parentheses are the square roots of AVE.*
{fl_matrix_df.to_markdown()}

### Table 3: Frozen Association Model Results
{path_table.to_markdown(index=False)}

### Table 4: Rank-Order Robustness Correlations (Spearman)
{spearman.round(3).to_markdown()}

### Frozen Bootstrap and Collinearity Checks
- Bootstrap resamples: {path_summary['bootstrap_iterations']}.
- Indirect association ({construct_names[0]} → {construct_names[1]} → {construct_names[2]}): {path_summary['indirect']}; 95% bootstrap CI [{path_summary['indirect_ci'][0]}, {path_summary['indirect_ci'][1]}].
- Residual direct association ({construct_names[0]} → {construct_names[2]} controlling {construct_names[1]}): {path_summary['direct']}; 95% bootstrap CI [{path_summary['direct_ci'][0]}, {path_summary['direct_ci'][1]}].
- Predictor/mediator VIF in the outcome model: {path_summary['predictor_mediator_vif']}.

[STRICT EMPIRICAL INSTRUCTION]
1. ZERO HALLUCINATION: Every study-specific fact and number must come from this frozen packet. Never invent a country, named organization, sampling frame, response rate, demographic characteristic, measure source, software package, coefficient, p-value, fit index, robustness test, or date.
2. DESIGN DISCIPLINE: This is a cross-sectional observational survey. Use association language. The indirect estimate is not causal mediation identification.
3. MEASUREMENT DEFENSE: State reliability/convergent/discriminant validity only when the displayed values support it.
4. RESULTS DISCIPLINE: Results must report this study's frozen numbers without attaching external citations to the coefficients or sample facts.
5. ROBUSTNESS: The Spearman matrix, bootstrap intervals, and VIF above are the only frozen robustness evidence. Do not add another robustness procedure.
6. CONSISTENCY: The predictor, mediator, outcome, sample size, context, scale, and all numerical results above are the single source of quantitative truth throughout the manuscript.
"""
        metadata = {
            "mode": "quantitative",
            "source_type": "system_generated",
            "disclosure_required": True,
            "sample_size": int(len(df)),
            "scale_points": 7,
            "study_design": "cross-sectional single-wave questionnaire",
            "research_context": context,
            "constructs": {
                "predictor": construct_names[0],
                "mediator": construct_names[1],
                "outcome": construct_names[2],
            },
            "association_results": path_summary,
            "method": (
                "Positive-semidefinite latent correlation generation, indicator loadings, "
                "Likert discretization, reliability/validity summaries, standardized association "
                "models, fixed-seed bootstrap intervals, and Spearman robustness correlations."
            ),
            "generation_parameters": {
                "constructs": construct_names,
                "sample_size": int(len(df)),
                "likert_scale_points": 7,
                "items_per_construct": "random integer from 3 to 5",
                "indicator_loading_range": [0.72, 0.88],
                "latent_correlation_control": "positive-semidefinite matrix with maximum target correlation 0.55",
                "item_generation": "loading × latent score + normally distributed measurement error",
                "discretization": "round to nearest integer and clamp to 1–7",
                "bootstrap_iterations": path_summary["bootstrap_iterations"],
                "bootstrap_seed": 20260809,
                "reported_summaries": [
                    "Cronbach alpha",
                    "composite reliability",
                    "average variance extracted",
                    "descriptive statistics",
                    "Fornell-Larcker matrix",
                    "standardized association estimates",
                    "R-squared",
                    "bootstrap confidence intervals",
                    "Spearman correlations",
                    "VIF",
                ],
            },
            "evidence_sections": [
                "Frozen quantitative study design",
                "Reliability and convergent validity",
                "Descriptive statistics and discriminant validity",
                "Frozen association model results",
                "Rank-order robustness correlations",
                "Frozen bootstrap and collinearity checks",
            ],
        }
        metadata["quantitative_evidence_ledger"] = self.build_quantitative_evidence_ledger(metadata, generated_prompt)
        return (
            df,
            generated_prompt,
            tr(language, "synthetic_stats_generated"),
            metadata,
        )


def get_quantitative_engine():
    return QuantitativeEngine()