# -*- coding: utf-8 -*- """Quantitative evidence preparation for SmartPaper's AMJ-oriented track.""" from __future__ import annotations import hashlib 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" QUANTITATIVE_TABLE_MANIFEST_VERSION = "quantitative-table-manifest-v1" HYPOTHESIS_LEDGER_VERSION = "quantitative-hypothesis-ledger-v1" TARGET_TABLE_BINDING_VERSION = "target-quantitative-table-binding-v1" SEMANTIC_CLAIM_FRAME_VERSION = "smartpaper-core-semantic-claim-frame-v1" _TARGET_TABLE_LOCALIZATION = { "zh": { "caption": "表 {number}：{title}", "titles": { 1: "信度与收敛效度", 2: "描述性统计与判别效度（Fornell-Larcker 准则）", 3: "关联模型结果", 4: "秩次稳健性相关（Spearman）", }, "cells": { "Construct": "构念", "Items": "题项数", "Cronbach's α": "Cronbach α", "Mean": "均值", "SD": "SD", "Association": "关联关系", "Standardized estimate": "标准化估计", "R² of outcome": "结果变量的 R²", "Path": "路径", "Beta": "Beta", "Estimate": "估计量", "Point estimate": "点估计", "95% bootstrap CI lower": "95% bootstrap CI 下限", "95% bootstrap CI upper": "95% bootstrap CI 上限", }, "phrases": { "Indirect association:": "间接关联：", "Indirect association": "间接关联", "Residual direct association": "剩余直接关联", "controlling": "控制", }, "notes": { "Diagonal elements in parentheses are the square roots of AVE.": "括号中的对角线元素为 AVE 的平方根。", }, "note_prefix": "注：", }, "ru": { "caption": "Таблица {number}: {title}", "titles": { 1: "Надёжность и конвергентная валидность", 2: "Описательная статистика и дискриминантная валидность (критерий Форнелла–Ларкера)", 3: "Результаты модели ассоциаций", 4: "Ранговые корреляции для проверки устойчивости (Спирмен)", }, "cells": { "Construct": "Конструкт", "Items": "Пункты", "Cronbach's α": "α Кронбаха", "Mean": "Среднее", "SD": "SD", "Association": "Ассоциация", "Standardized estimate": "Стандартизированная оценка", "R² of outcome": "R² результата", "Path": "Связь", "Beta": "Beta", "Estimate": "Оценка", "Point estimate": "Точечная оценка", "95% bootstrap CI lower": "Нижняя граница 95% bootstrap CI", "95% bootstrap CI upper": "Верхняя граница 95% bootstrap CI", }, "phrases": { "Indirect association:": "Косвенная ассоциация:", "Indirect association": "Косвенная ассоциация", "Residual direct association": "Остаточная прямая ассоциация", "controlling": "при контроле", }, "notes": { "Diagonal elements in parentheses are the square roots of AVE.": "Диагональные элементы в скобках представляют квадратные корни из AVE.", }, "note_prefix": "Примечание:", }, "kk": { "caption": "{number}-кесте: {title}", "titles": { 1: "Сенімділік және конвергенттік валидтілік", 2: "Сипаттамалық статистика және дискриминанттық валидтілік (Форнелл–Ларкер критерийі)", 3: "Байланыстар моделінің нәтижелері", 4: "Орнықтылықты тексеруге арналған рангтік корреляциялар (Спирмен)", }, "cells": { "Construct": "Конструкт", "Items": "Тармақтар", "Cronbach's α": "Кронбах α", "Mean": "Орташа", "SD": "SD", "Association": "Байланыс", "Standardized estimate": "Стандартталған бағалау", "R² of outcome": "Нәтиженің R²", "Path": "Байланыс", "Beta": "Beta", "Estimate": "Бағалау", "Point estimate": "Нүктелік бағалау", "95% bootstrap CI lower": "95% bootstrap CI төменгі шегі", "95% bootstrap CI upper": "95% bootstrap CI жоғарғы шегі", }, "phrases": { "Indirect association:": "Жанама байланыс:", "Indirect association": "Жанама байланыс", "Residual direct association": "Қалдық тікелей байланыс", "controlling": "бақылау кезінде", }, "notes": { "Diagonal elements in parentheses are the square roots of AVE.": "Жақшадағы диагональ элементтер AVE квадрат түбірлерін көрсетеді.", }, "note_prefix": "Ескертпе:", }, } @classmethod def build_hypothesis_ledger(cls, constructs): """Freeze one result-independent AMJ hypothesis identity contract. The system-generated quantitative design is observational and does not justify result-driven directional hypotheses. The three numbered hypotheses therefore freeze only the construct pairing/association identity. The indirect association is an analysis target, not H4. """ values = [re.sub(r"\s+", " ", str(item or "")).strip() for item in (constructs or [])] if len(values) != 3 or any(not item for item in values): return {} predictor, mediator, outcome = values items = [ {"hypothesis_id": "H1", "predictor": predictor, "outcome": outcome, "relation": "association"}, {"hypothesis_id": "H2", "predictor": predictor, "outcome": mediator, "relation": "association"}, {"hypothesis_id": "H3", "predictor": mediator, "outcome": outcome, "relation": "association"}, ] return { "version": cls.HYPOTHESIS_LEDGER_VERSION, "items": items, "indirect_analysis": { "predictor": predictor, "mediator": mediator, "outcome": outcome, "numbered_hypothesis": False, "relation": "estimated_indirect_association", }, } @classmethod def build_table_manifest_from_evidence_markdown(cls, evidence_markdown): """Rebuild the synthetic table manifest from a durable pre-C5 evidence packet.""" value = str(evidence_markdown or "") targets = { 1: "Descriptive Statistics and Measurement Assessment", 2: "Descriptive Statistics and Measurement Assessment", 3: "Hypothesis and Association Tests", 4: "Additional Analyses and Robustness Checks", } rows = [] pattern = re.compile( r"(?ms)^###\s+Table\s+(?P[1-4])\s*:\s*(?P[^\n]+)\n(?P<body>.*?)(?=^###\s+|\Z)" ) for match in pattern.finditer(value): number = int(match.group("number")) body = match.group("body").strip() table_lines = [] note = "" for line in body.splitlines(): stripped = line.strip() if stripped.startswith("|"): table_lines.append(line.rstrip()) elif re.match(r"^\*?Note\s*:", stripped, flags=re.I): note = re.sub(r"^\*?Note\s*:\s*", "", stripped, flags=re.I).strip("* ") elif table_lines: break if len(table_lines) < 2: continue title = re.sub(r"^Frozen\s+", "", match.group("title").strip(), flags=re.I) row = { "number": number, "title": title, "target_subsection": targets[number], "markdown": "\n".join(table_lines), } if note: row["note"] = note rows.append(row) if {row["number"] for row in rows} != {1, 2, 3, 4}: return {} return {"version": cls.QUANTITATIVE_TABLE_MANIFEST_VERSION, "tables": sorted(rows, key=lambda row: row["number"])} @classmethod def build_table_manifest(cls, validity_df, fl_matrix_df, path_table, spearman): """Return the exact reader-facing tables required by the synthetic AMJ path.""" rows = [ { "number": 1, "title": "Reliability and Convergent Validity", "target_subsection": "Descriptive Statistics and Measurement Assessment", "markdown": validity_df.to_markdown(index=False), }, { "number": 2, "title": "Descriptive Statistics and Discriminant Validity (Fornell-Larcker Criterion)", "target_subsection": "Descriptive Statistics and Measurement Assessment", "note": "Diagonal elements in parentheses are the square roots of AVE.", "markdown": fl_matrix_df.to_markdown(), }, { "number": 3, "title": "Association Model Results", "target_subsection": "Hypothesis and Association Tests", "markdown": path_table.to_markdown(index=False), }, { "number": 4, "title": "Rank-Order Robustness Correlations (Spearman)", "target_subsection": "Additional Analyses and Robustness Checks", "markdown": spearman.round(3).to_markdown(), }, ] return {"version": cls.QUANTITATIVE_TABLE_MANIFEST_VERSION, "tables": rows} @staticmethod def _target_table_language_code(target_lang): value = str(target_lang or "").strip().casefold().replace("_", "-") if value.startswith("zh") or "chinese" in value or "中文" in value or "简体" in value: return "zh" if value.startswith("ru") or "russian" in value or "рус" in value: return "ru" if value.startswith("kk") or value.startswith("kz") or "kazakh" in value or "қазақ" in value: return "kk" return "en" @staticmethod def _replace_glossary_terms(value, glossary): text = str(value or "") pairs = [] for source, target in dict(glossary or {}).items(): source = re.sub(r"\s+", " ", str(source or "")).strip() target = re.sub(r"\s+", " ", str(target or "")).strip() if not source or not target or source == target: continue if not re.search(r"[A-Za-z]", source): continue pairs.append((source, target)) for source, target in sorted(pairs, key=lambda row: len(row[0]), reverse=True): pattern = re.compile( rf"(?<![A-Za-z0-9_]){re.escape(source)}(?![A-Za-z0-9_])", flags=re.I, ) text = pattern.sub(lambda _match, replacement=target: replacement, text) return text @classmethod def _localize_quantitative_table_cell(cls, value, target_lang, glossary=None): code = cls._target_table_language_code(target_lang) if code == "en": return cls._replace_glossary_terms(value, glossary) profile = cls._TARGET_TABLE_LOCALIZATION[code] text = str(value or "") stripped = text.strip() exact = dict(profile.get("cells") or {}) if stripped in exact: return text.replace(stripped, str(exact[stripped]), 1) text = cls._replace_glossary_terms(text, glossary) for source, target in sorted( dict(profile.get("phrases") or {}).items(), key=lambda row: len(row[0]), reverse=True ): text = re.sub(re.escape(source), str(target), text, flags=re.I) return text @classmethod def _localize_quantitative_markdown_table(cls, markdown, target_lang, glossary=None): lines = [] for line in str(markdown or "").splitlines(): if not line.lstrip().startswith("|"): lines.append(line) continue raw = line.rstrip() cells = raw.split("|") # Alignment rows contain only punctuation/colons and must remain byte-stable. if all(not re.search(r"[A-Za-z\u0400-\u04FF\u4e00-\u9fff]", cell) for cell in cells): lines.append(raw) continue localized = [cls._localize_quantitative_table_cell(cell, target_lang, glossary) for cell in cells] lines.append("|".join(localized)) return "\n".join(lines).strip() @classmethod def localize_quantitative_table_manifest(cls, manifest, target_lang, glossary=None): record = dict(manifest or {}) rows = [dict(item) for item in (record.get("tables") or []) if isinstance(item, dict)] if {int(row.get("number") or 0) for row in rows} != {1, 2, 3, 4}: return {} code = cls._target_table_language_code(target_lang) profile = cls._TARGET_TABLE_LOCALIZATION.get(code, {}) localized_rows = [] for row in sorted(rows, key=lambda item: int(item.get("number") or 0)): number = int(row.get("number") or 0) title = str(row.get("title") or "").strip() note = str(row.get("note") or "").strip() if code != "en": title = str(dict(profile.get("titles") or {}).get(number) or title) note = str(dict(profile.get("notes") or {}).get(note) or note) localized_rows.append({ **row, "number": number, "title": title, "markdown": cls._localize_quantitative_markdown_table( row.get("markdown"), target_lang, glossary ), **({"note": note} if note else {}), }) return { "version": cls.TARGET_TABLE_BINDING_VERSION, "source_manifest_version": str(record.get("version") or ""), "language": code, "tables": localized_rows, } @staticmethod def _strip_target_quantitative_markdown_tables(text): """Remove any model-rendered table blocks before manifest-owned rebinding.""" lines = str(text or "").splitlines() kept = [] index = 0 caption_pattern = re.compile( r"^\s*(?:Table\s+\d+|表\s*\d+|Таблиц(?:а|ы|е|у|ей|ами|ах)?\s*\d+|Кесте\s*\d+|\d+\s*[-–—]?\s*кесте)\s*[:：.\-–—]", flags=re.I, ) note_pattern = re.compile(r"^\s*\*?(?:Note|注|Примечание|Ескертпе)\s*[:：]", flags=re.I) while index < len(lines): line = lines[index] if caption_pattern.match(line): index += 1 while index < len(lines) and not lines[index].strip(): index += 1 if index < len(lines) and lines[index].lstrip().startswith("|"): while index < len(lines) and lines[index].lstrip().startswith("|"): index += 1 while index < len(lines) and not lines[index].strip(): index += 1 if index < len(lines) and note_pattern.match(lines[index]): index += 1 continue if line.lstrip().startswith("|"): start = index while index < len(lines) and lines[index].lstrip().startswith("|"): index += 1 block = lines[start:index] if len(block) >= 2 and re.search(r"-{3,}", block[1]): continue kept.extend(block) continue kept.append(line) index += 1 return re.sub(r"\n{3,}", "\n\n", "\n".join(kept)).strip() @classmethod def bind_target_language_quantitative_tables(cls, translated_results, manifest, target_lang, glossary=None): """Bind exactly Table 1-4 to one translated Results unit without model ownership. This operation is deterministic and idempotent. Statistical values remain source-manifest owned; only captions, generic statistical labels, notes, and canonical construct surfaces are localized. """ localized = cls.localize_quantitative_table_manifest(manifest, target_lang, glossary) rows = [item for item in (localized.get("tables") or []) if isinstance(item, dict)] if {int(row.get("number") or 0) for row in rows} != {1, 2, 3, 4}: raise RuntimeError("target_quantitative_table_manifest_invalid") code = cls._target_table_language_code(target_lang) profile = cls._TARGET_TABLE_LOCALIZATION.get(code, {}) base = cls._strip_target_quantitative_markdown_tables(translated_results) blocks = [] for row in rows: number = int(row.get("number") or 0) title = str(row.get("title") or "").strip() markdown = str(row.get("markdown") or "").strip() if not number or not title or not markdown: raise RuntimeError(f"target_quantitative_table_manifest_invalid:Table {number}") if code == "zh": caption = f"表 {number}：{title}" elif code == "ru": caption = f"Таблица {number}: {title}" elif code == "kk": caption = f"{number}-кесте: {title}" else: caption = f"Table {number}: {title}" block = caption + "\n" + markdown note = str(row.get("note") or "").strip() if note: prefix = str(profile.get("note_prefix") or "Note:") block += f"\n\n*{prefix} {note}*" blocks.append(block) return (base.rstrip() + "\n\n" + "\n\n".join(blocks)).strip() @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, logical_node='structured_json', ) 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 [], } @classmethod def classify_quantitative_method_threat_claim(cls, sentence, term): """Classify a forbidden quantitative term by *claim role*, not token presence. R6-R10 separates an actually performed/claimed unowned analysis from a limitation, explicit non-use statement, threat discussion, or incidental methodological mention. Only ``PERFORMED_UNOWNED`` is blocking. """ text = re.sub(r"\s+", " ", str(sentence or "")).strip() folded = text.casefold() label = re.sub(r"\s+", " ", str(term or "")).strip() term_folded = label.casefold() if not text or not term_folded: return {"term": label, "category": "UNKNOWN", "state": "ABSENT"} match = re.search(rf"(?i)(?<![A-Za-z]){re.escape(label)}(?![A-Za-z])", text) if not match: return {"term": label, "category": "UNKNOWN", "state": "ABSENT"} threat_terms = {"common method bias", "endogeneity"} fit_terms = {"rmsea", "cfi", "tli", "srmr"} category = ( "THREAT" if term_folded in threat_terms else "FIT_DIAGNOSTIC" if term_folded in fit_terms else "ANALYTIC_METHOD" ) before = folded[max(0, match.start() - 100):match.start()] after = folded[match.end():min(len(folded), match.end() + 120)] local = folded[max(0, match.start() - 120):min(len(folded), match.end() + 140)] scalar_negation = bool(re.search( r"\bnot\s+(?:only|merely|just|simply)\b", local )) explicit_nonuse = (not scalar_negation) and bool( re.search( r"\b(?:no|not|never|without|did\s+not|does\s+not|do\s+not|was\s+not|were\s+not|is\s+not|are\s+not|cannot|can\s+not)\b[^.;:]{0,55}$", before, ) or re.search( r"^.{0,45}\b(?:was\s+not|were\s+not|is\s+not|are\s+not|did\s+not|does\s+not|do\s+not|cannot|can\s+not|not\s+(?:used|conducted|performed|estimated|tested|assessed|evaluated|reported|implemented|applied))\b", after, ) or re.search( r"\b(?:does\s+not|do\s+not|did\s+not|cannot|can\s+not|is\s+not|are\s+not|was\s+not|were\s+not)\s+(?:address|assess|test|establish|identify|rule\s+out|correct|control|substitute|demonstrate|show|prove|support)\b", local, ) ) if explicit_nonuse: return {"term": label, "category": category, "state": "NOT_PERFORMED_OR_LIMITATION"} performed = bool( re.search( r"\b(?:we|the\s+study|this\s+study|the\s+analysis|our\s+analysis|the\s+model|the\s+authors?)\b[^.;:]{0,70}\b(?:use|uses|used|employ|employs|employed|conduct|conducts|conducted|perform|performs|performed|estimate|estimates|estimated|test|tests|tested|assess|assesses|assessed|evaluate|evaluates|evaluated|implement|implements|implemented|apply|applies|applied|ran|run|runs|fit|fits|fitted|model|models|modeled|modelled|control|controls|controlled|correct|corrects|corrected|address|addresses|addressed)\b[^.;:]{0,70}$", before, ) or re.search( r"^.{0,65}\b(?:was|were|is|are)\s+(?:used|employed|conducted|performed|estimated|tested|assessed|evaluated|implemented|applied|fit|fitted|modeled|modelled|controlled|corrected|addressed|reported)\b", after, ) or re.search( r"^.{0,65}\b(?:estimated|tested|assessed|evaluated|showed|indicated|yielded|produced|confirmed|supported)\b", after, ) ) if category == "FIT_DIAGNOSTIC" and re.search( rf"(?i)(?:{re.escape(label)}\s*(?:=|:|was|were)\s*[-+]?\d)|(?:reported|fit\s+index|fit\s+indices)[^.;:]{{0,45}}{re.escape(label)}", text, ): performed = True if category == "THREAT": # Threats may be discussed as limitations without implying a test. if re.search( rf"(?i)(?:assess|test|evaluate|examine|address|correct|control)(?:ed|ing|s)?[^.;:]{{0,45}}{re.escape(label)}|{re.escape(label)}[^.;:]{{0,45}}(?:was|were)\s+(?:assessed|tested|evaluated|examined|addressed|corrected|controlled)", text, ): performed = True elif not performed: return {"term": label, "category": category, "state": "LIMITATION_OR_THREAT"} return { "term": label, "category": category, "state": "PERFORMED_UNOWNED" if performed else "MENTION_ONLY", } @classmethod def _first_performed_unowned_quantitative_term(cls, sentence, forbidden_terms): for term in forbidden_terms or []: frame = cls.classify_quantitative_method_threat_claim(sentence, term) if frame.get("state") == "PERFORMED_UNOWNED": return frame return None @staticmethod def _semantic_frame_hash(payload): canonical = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) return hashlib.sha256(canonical.encode("utf-8")).hexdigest() @classmethod def build_quantitative_semantic_claim_frame(cls, sentence, metadata=None, section_name=""): """Freeze a compact server-side semantic frame for one quantitative claim. R6-R1 does not ask the model to decide this frame. It records the source-side claim role, causal polarity/strength, methodological domain, hypothesis identity, and explicit result direction so downstream validation/translation can inherit one stable semantic authority. """ text = re.sub(r"\s+", " ", str(sentence or "")).strip() folded = text.casefold() meta = dict(metadata or {}) ledger = dict(meta.get("quantitative_evidence_ledger") or {}) if not ledger and meta: try: ledger = cls.build_quantitative_evidence_ledger(meta) except Exception: ledger = {} constructs = dict(ledger.get("constructs") or {}) known_constructs = [str(value) for value in constructs.values() if str(value)] ownership = cls._quantitative_sentence_ownership(text, section_name, known_constructs) if text else "other" relation_specs = ( (r"\b(?:causes?|caused|causing)\b", "CAUSE"), (r"\b(?:leads?|led)\s+to\b", "LEAD_TO"), (r"\b(?:drives?|drove|driven)\b", "DRIVE"), (r"\b(?:determines?|determined)\b", "DETERMINE"), (r"\b(?:produces?|produced)\b", "PRODUCE"), (r"\b(?:results?|resulted)\s+in\b", "RESULT_IN"), (r"\b(?:yields?|yielded)\b", "YIELD"), (r"\b(?:generates?|generated)\b", "GENERATE"), (r"\bcausal(?:ity|\s+(?:effect|effects|mediation|mechanism|pathway|relationship|relationships|identification))\b", "CAUSAL_NOUN"), ) relation = "NONE" for pattern, label in relation_specs: if re.search(pattern, folded, flags=re.I): relation = label break methodological_bias = cls._is_methodological_bias_mechanism_sentence(text, known_constructs) robustness_scope = bool(re.search( r"\b(?:robustness|sensitivity|spearman|rank[- ]order|extreme\s+values?|outliers?|influential\s+(?:cases|observations)|" r"response\s+styles?|common\s+method|same[- ]source|single\s+questionnaire|measurement\s+(?:error|bias))\b", folded, flags=re.I, )) if methodological_bias: domain = "METHODOLOGICAL_BIAS" elif robustness_scope: domain = "METHODOLOGICAL_ROBUSTNESS" elif re.search(r"\b(?:analysis|analyses|bootstrap|regression|correlation|estimation|statistical\s+(?:test|model|analysis)|vif)\b", folded): domain = "ANALYTICAL_REPORTING" elif any(name.casefold() in folded for name in known_constructs): domain = "SUBSTANTIVE_CONSTRUCT" else: domain = "GENERAL" raw_causal_state = cls.classify_quantitative_causal_sentence(text) if text else "NON_CAUSAL" validation_state = "NON_CAUSAL" if methodological_bias else raw_causal_state if raw_causal_state == "EXPLICIT_CAUSAL_LIMITATION": polarity = "NEGATIVE" causal_strength = "CAUSAL_LIMITATION" elif raw_causal_state == "AFFIRMATIVE_CAUSAL": polarity = "AFFIRMATIVE" causal_strength = "CAUSAL_ASSERTION" elif re.search(r"\b(?:association|associations|associated|correlation|correlations|correlated|covariation|covariance)\b", folded): polarity = "NONE" causal_strength = "ASSOCIATIONAL" else: polarity = "NONE" causal_strength = "NON_CAUSAL" if re.search(r"\b(?:may|might|could|appears?\s+to|seems?\s+to|suggests?|consistent\s+with)\b", folded) and not re.search(r"\bcould\s+not\b", folded): modality = "HEDGED" elif raw_causal_state == "EXPLICIT_CAUSAL_LIMITATION": modality = "LIMITATION" else: modality = "ASSERTED" hypothesis_ids = sorted(set(re.findall(r"(?i)\bH\d+[A-Za-z]?\b", text)), key=str.casefold) if re.search(r"\b(?:non[- ]?significant|not\s+significant|did\s+not\s+support|was\s+not\s+supported)\b", folded): result_direction = "NOT_SUPPORTED_OR_NONSIGNIFICANT" elif re.search(r"\b(?:positive|positively)\b[^.;]{0,80}\b(?:association|associated|relationship|correlation|estimate|coefficient)\b", folded): result_direction = "POSITIVE" elif re.search(r"\b(?:negative|negatively)\b[^.;]{0,80}\b(?:association|associated|relationship|correlation|estimate|coefficient)\b", folded): result_direction = "NEGATIVE" else: result_direction = "UNSPECIFIED" if hypothesis_ids: claim_type = "HYPOTHESIS_CLAIM" elif domain in {"METHODOLOGICAL_BIAS", "METHODOLOGICAL_ROBUSTNESS"}: claim_type = "METHODOLOGICAL_CLAIM" elif ownership == "result": claim_type = "RESULT_CLAIM" elif causal_strength == "ASSOCIATIONAL": claim_type = "ASSOCIATION_CLAIM" elif raw_causal_state != "NON_CAUSAL": claim_type = "CAUSAL_SCOPE_CLAIM" else: claim_type = "OTHER" frame = { "version": cls.SEMANTIC_CLAIM_FRAME_VERSION, "mode": "quantitative", "claim_type": claim_type, "ownership": ownership, "domain": domain, "relation": relation, "polarity": polarity, "causal_strength": causal_strength, "causal_state": raw_causal_state, "validation_state": validation_state, "modality": modality, "hypothesis_ids": hypothesis_ids, "result_direction": result_direction, } frame["frame_hash"] = cls._semantic_frame_hash(frame) return frame @classmethod def _classify_quantitative_causal_clause(cls, clause): """Classify one clause by causal *assertion scope*, not token presence. C5.5 freezes a single semantic contract for Writer/Validator/Retry. A causal phrase inside an explicit limitation (for example, ``cannot establish that X causes Y``) is not an affirmative claim. Conversely, an actual causal assertion remains fail-closed even when another part of the sentence contains a disclaimer. """ value = re.sub(r"\s+", " ", str(clause or "")).strip().casefold() if not value: return "NON_CAUSAL" causal_term = ( r"(?:causal(?:ity|\s+(?:effect|effects|mediation|mechanism|pathway|relationship|relationships|identification))|" r"causation|cause(?:s|d|ing)?)" ) direct_patterns = ( (r"\b(?:causes?|caused|causing)\b", "STRONG"), (r"\b(?:leads?|led)\s+to\b", "STRONG"), (r"\b(?:drives?|drove|driven)\b", "STRONG"), (r"\b(?:determines?|determined)\b", "STRONG"), (r"\b(?:produces?|produced)\b", "AMBIGUOUS_REPORTING"), (r"\b(?:results?|resulted)\s+in\b", "AMBIGUOUS_REPORTING"), (r"\b(?:yields?|yielded)\b", "AMBIGUOUS_REPORTING"), (r"\b(?:generates?|generated)\b", "AMBIGUOUS_REPORTING"), (r"\b(?:transmits?|transmitted)\s+(?:an?\s+|the\s+)?causal\s+effect\b", "STRONG"), (r"\bcausally\s+(?:transmits?|mediates?|affects?|influences?)\b", "STRONG"), (r"\b(?:has|have|had|exerts?|exerted)\s+(?:an?\s+|the\s+)?causal\s+effect\b", "STRONG"), ) def local_actor_text(match): """Return the grammatical actor window immediately before a predicate. R3 deliberately classifies the *actor* rather than maintaining an ever-growing whitelist of reporting artefacts. Leading adverbial or prepositional phrases (``In the analysis, ...`` / ``Using Spearman correlations, ...``) are peeled away so that a nearby mention of an analysis cannot shield a later construct-level causal assertion. """ prefix = value[max(0, match.start() - 260):match.start()] actor = re.split(r"[.;!?]", prefix)[-1].strip() lead = re.match( r"^(?:in|using|with|based\s+on|according\s+to|after|before|when|while|because|" r"although|though|given|under|for|as)\b[^,]{0,180},\s*(.+)$", actor, flags=re.I, ) if lead: actor = lead.group(1).strip() return actor[-220:] def actor_type(match): """Classify the predicate subject as analytical, weak-analytical, or other. ``ANALYTICAL_ACTOR`` is intentionally restricted to explicit statistical/diagnostic actors. Generic ``model``/``procedure`` terms are weak because they can also denote the substantive AI system or an intervention; they are exempt only when the predicate target is a recognisable reporting artefact. """ actor = local_actor_text(match) strong_actor = ( r"\b(?:analysis|analyses|" r"bootstrap(?:\s+(?:resampling|procedure|computation|analysis))?|" r"resampling(?:\s+(?:procedure|analysis|computation))?|" r"regression\s+(?:analysis|model)|association\s+(?:analysis|model)|" r"mediation\s+(?:analysis|model)|" r"estimation(?:\s+(?:procedure|method|analysis))?|estimator|" r"statistical\s+(?:analysis|model|procedure|method|test)|" r"robustness\s+(?:check|test|analysis)|" r"sensitivity\s+(?:check|test|analysis)|" r"diagnostic\s+(?:check|test|analysis)|" r"rank[- ]order\s+(?:robustness\s+)?check|" r"(?:spearman|pearson|rank[- ]order)\s+correlations?|" r"correlation\s+(?:analysis|matrix|check|test)|" r"hypothesis\s+test|significance\s+test|" r"variance\s+inflation\s+factor(?:\s+check)?|vif(?:\s+check)?|" r"measurement\s+assessment|descriptive\s+analysis)\b" ) if re.search(strong_actor, actor, flags=re.I): return "ANALYTICAL_ACTOR" # R6-R12 derived statistical-expression actor frame. Phrases such # as ``the product of the two constituent associations`` and # ``multiplying the path coefficients`` denote a deterministic # statistical calculation, not a substantive study construct. # They receive a distinct actor type so ambiguous reporting verbs # (yields/produces/generates/results in) are exempt *only* when the # predicate target is also a recognised statistical reporting # object. This keeps construct-level uses such as ``X yields lower # anxiety`` fail-closed. derived_statistical_actor = ( r"\b(?:" r"(?:the\s+)?product\s+of\s+(?:the\s+)?" r"(?:(?:two|constituent|standardized|standardised|estimated|path)\s+){0,4}" r"(?:associations?|coefficients?|paths?|estimates?)|" r"(?:coefficient|association|path)[- ]product(?:\s+(?:calculation|computation|estimate))?|" r"path[- ]product(?:\s+(?:calculation|computation|estimate))?|" r"multiplying\s+(?:the\s+)?(?:(?:two|constituent|standardized|standardised|path)\s+){0,4}" r"(?:associations?|coefficients?|paths?)|" r"multiplication\s+of\s+(?:the\s+)?(?:(?:two|constituent|standardized|standardised|path)\s+){0,4}" r"(?:associations?|coefficients?|paths?)" r")\b" ) if re.search(derived_statistical_actor, actor, flags=re.I): return "DERIVED_STATISTICAL_ACTOR" if re.search(r"\b(?:model|procedure|method|computation|calculation)\b", actor, flags=re.I): return "WEAK_ANALYTICAL_ACTOR" return "OTHER_ACTOR" def predicate_target_is_reporting_output(match): suffix = value[match.end():min(len(value), match.end() + 220)] target_pattern = ( r"^\s*(?:(?:an?|the|this|that|these|those|following|reported|" r"standardized|standardised|numerical|estimated|95\s*percent|95%)\s+){0,6}" r"(?:confidence\s+intervals?|interval\s+estimates?|" r"coefficient(?:\s+estimates?)?s?|standard\s+errors?|" r"parameter\s+estimates?|point\s+estimates?|indirect\s+estimates?|" r"direct\s+estimates?|effect-size\s+estimates?|" r"indirect\s+associations?|association\s+estimates?|" r"standardized\s+associations?|standardised\s+associations?|" r"correlations?|covariances?|variances?|loadings?|" r"test\s+statistics?|p[- ]?values?|vif(?:s|\s+values)?|" r"cronbach(?:'s)?\s+alpha|composite\s+reliability|" r"average\s+variance\s+extracted|ave|numerical\s+estimates?|estimates?|" r"(?:a\s+)?pattern(?:\s+of\s+(?:associations|correlations|estimates))?|" r"(?:a\s+)?(?:similar|consistent|comparable)\s+pattern|" r"findings?|results?)\b" ) return bool(re.search(target_pattern, suffix, flags=re.I)) def direct_match_is_analytical_reporting(match, predicate_kind): actor = actor_type(match) if predicate_kind == "AMBIGUOUS_REPORTING": # ``analysis/check/test produced/resulted/yielded/generated ...`` # describes what an analytical operation reported. It is not a # study-construct causal assertion even when the following prose is # stylistically loose (e.g. ``the analysis produced higher # engagement``). The CAR layer may flag wording separately; this # causal gate must not terminate the manuscript for it. if actor == "ANALYTICAL_ACTOR": return True if actor in {"WEAK_ANALYTICAL_ACTOR", "DERIVED_STATISTICAL_ACTOR"} and predicate_target_is_reporting_output(match): return True return False # Strong causal predicates remain fail-closed. Preserve only the # narrow statistical-reporting idiom already accepted by R1/R6-R12, # such as ``the analysis led to the following numerical estimate`` # or a derived coefficient product leading to an estimate. return actor in {"ANALYTICAL_ACTOR", "WEAK_ANALYTICAL_ACTOR", "DERIVED_STATISTICAL_ACTOR"} and predicate_target_is_reporting_output(match) def direct_match_is_disclaimed(match): prefix = value[max(0, match.start() - 220):match.start()] suffix = value[match.end():min(len(value), match.end() + 220)] # R6-R1 predicate polarity closure: passive negation is a semantic # limitation ("associations are not driven by outliers"), not an # affirmative causal claim. Do not exempt contrastive constructions # that merely replace one asserted cause with another ("not driven by # X but rather by Y") or scalar forms such as "not only driven". passive_negative = bool(re.search( r"\b(?:is|are|was|were|be|been|being)\s+" r"(?:(?:clearly|necessarily|directly|statistically)\s+){0,2}" r"(?:not|never)\s+" r"(?:(?:clearly|necessarily|directly|statistically)\s+){0,2}$", prefix, flags=re.I, )) if passive_negative: alternate_cause = re.search( r"\b(?:but|rather|instead)\b\s*(?:rather\s+|instead\s+|also\s+)?by\b", suffix, flags=re.I, ) alternate_negated = re.search( r"\b(?:but|rather|instead)\b\s*(?:also\s+)?(?:not|never)\s+by\b", suffix, flags=re.I, ) if alternate_cause and not alternate_negated: return False return True if re.search( r"(?:\b(?:does|do|did|can|could|would|should|must|may|might)\s+not|" r"\bcannot|\bcan't|\bcouldn't|\bnever)\s+(?:\w+\s+){0,2}$", prefix, flags=re.I, ): return True if re.search( r"\b(?:cannot|can't|could not|couldn't|does not|doesn't|do not|don't|did not|didn't|" r"fails? to|failed to|unable to|insufficient to|not able to)\s+" r"(?:identify|establish|demonstrate|show|infer|support|imply|mean|claim|justify|warrant|prove)" r"\b[^.;]{0,180}?(?:\bthat\b|\bwhether\b)[^.;]{0,120}$", prefix, flags=re.I, ): return True if re.search( r"\b(?:should not|shouldn't|must not|cannot|can't|could not|couldn't)\s+be\s+" r"(?:interpreted|read|treated|understood)\s+(?:as|to mean)\b[^.;]{0,140}$", prefix, flags=re.I, ): return True if re.search( r"\bwithout\s+(?:identifying|establishing|demonstrating|showing|inferring|supporting|proving)" r"\b[^.;]{0,180}?(?:\bthat\b|\bwhether\b)[^.;]{0,120}$", prefix, flags=re.I, ): return True return False saw_disclaimed_direct = False for pattern, predicate_kind in direct_patterns: for match in re.finditer(pattern, value, flags=re.I): if direct_match_is_disclaimed(match): saw_disclaimed_direct = True continue if direct_match_is_analytical_reporting(match, predicate_kind): continue return "AFFIRMATIVE_CAUSAL" # R6-R11 causal-noun scope frame. A past participle can be either a # predicate ("the analysis identified a causal effect") or a modifier # inside a causal noun phrase ("rather than an identified causal # effect"). The latter inherits the surrounding negation/contrast # operator and must not be promoted to an affirmative causal claim. causal_noun_participle = ( r"(?:identified|established|demonstrated|shown|inferred|supported|" r"claimed|justified|warranted|proven)" ) causal_noun_adverb = r"(?:empirically|statistically|directly|clearly|credibly|formally|causally)" causal_noun_modifier = ( rf"(?:(?:{causal_noun_adverb}\s+){{0,2}}{causal_noun_participle}\s+)?" ) causal_noun_surface = rf"(?:an?\s+|the\s+)?{causal_noun_modifier}{causal_term}" limitation_patterns = ( rf"\b(?:cannot|can't|could not|couldn't|does not|doesn't|do not|don't|did not|didn't|fails? to|failed to|" rf"is not able to|are not able to|was not able to|were not able to|unable to|insufficient to|not able to)\s+" rf"(?:identify|establish|demonstrate|show|infer|support|imply|mean|claim|justify|warrant|permit|allow|prove)\b[^.;]*\b{causal_term}\b", rf"\b(?:is|are|was|were)\s+not\s+(?:designed|intended)\s+to\s+" rf"(?:identify|establish|demonstrate|show|infer|support|prove)\b[^.;]*\b{causal_term}\b", rf"\bwithout\s+(?:identifying|establishing|demonstrating|showing|inferring|supporting|proving)\b[^.;]*\b{causal_term}\b", rf"\b(?:no|without)\s+(?:evidence|basis|support|identification|proof)\b[^.;]*\b{causal_term}\b", rf"\b(?:is|are|was|were)\s+not\s+(?:evidence|proof)\s+(?:of|for)\s+{causal_noun_surface}\b", rf"\b(?:cannot|can't|could not|couldn't)\s+be\s+used\s+to\s+(?:infer|establish|demonstrate|show|prove)\b[^.;]*\b{causal_term}\b", rf"\b(?:should not|shouldn't|must not|cannot|can't|could not|couldn't)\s+be\s+" rf"(?:interpreted|read|treated|understood)\s+(?:as|to mean)\s+{causal_noun_surface}\b", rf"\b(?:is|are|was|were)?\s*(?:restricted|limited|confined)\s+to\b[^.;]*" rf"\b(?:not|rather than|instead of)\s+(?:to\s+)?{causal_noun_surface}\b", rf"\b(?:association|associations|covariation|covariance|correlation|correlations|indirect association|descriptive pattern|descriptive patterns)\b[^.;]*" rf"\b(?:not(?:\s+as)?|rather than|instead of)\s+{causal_noun_surface}\b", rf"\b{causal_term}\b\s+(?:is|are|was|were)\s+(?:not|never)\s+" rf"{causal_noun_participle}\b", rf"\b(?:not|no)\s+{causal_noun_surface}\b", ) limitation_matches = [ match for pattern in limitation_patterns for match in re.finditer(pattern, value, flags=re.I) ] def noun_match_is_inside_limitation(match): # Only suppress the affirmative interpretation when the matched # predicate/noun phrase is actually inside the frozen limitation # span. A later affirmative clause remains fail-closed. return any( limited.start() <= match.start() and match.end() <= limited.end() for limited in limitation_matches ) affirmative_noun_patterns = ( rf"\b(?:is|are|was|were|represent(?:s|ed)?|constitut(?:es|ed))\s+(?:an?\s+|the\s+)?{causal_term}\b", rf"\b(?:identify|identifies|identified|establish|establishes|established|demonstrate|demonstrates|demonstrated|show|shows|showed|prove|proves|proved|provide|provides|provided)\b[^.;]{{0,80}}\b{causal_term}\b", rf"\b(?:evidence|proof)\s+(?:of|for)\s+(?:an?\s+|the\s+)?{causal_term}\b", rf"\b{causal_term}\b[^.;]{{0,80}}\b(?:is|are|was|were)\s+(?!not\b|never\b)", ) for pattern in affirmative_noun_patterns: for match in re.finditer(pattern, value, flags=re.I): if noun_match_is_inside_limitation(match): continue prefix = value[max(0, match.start() - 80):match.start()] if re.search( r"\b(?:no|not|without|cannot|can't|insufficient|unable|does not|doesn't|do not|don't|did not|didn't|not able)\b", prefix, flags=re.I, ): continue return "AFFIRMATIVE_CAUSAL" if saw_disclaimed_direct or limitation_matches: return "EXPLICIT_CAUSAL_LIMITATION" return "NON_CAUSAL" @classmethod def classify_quantitative_causal_sentence(cls, sentence): """Three-state causal contract shared by inspection, repair, and tests.""" value = str(sentence or "") folded = re.sub(r"\s+", " ", value).strip().casefold() # A negated passive can still assert an alternative cause: "not driven # by X but rather by Y". Detect this before contrast-clause splitting, # otherwise the second fragment loses the causal predicate surface. alternate_cause = re.search( r"\b(?:is|are|was|were|be|been|being)\s+" r"(?:(?:clearly|necessarily|directly|statistically)\s+){0,2}" r"(?:not|never)\s+" r"(?:(?:clearly|necessarily|directly|statistically)\s+){0,2}" r"(?:caused|driven|determined|produced|generated|yielded)\b" r"[^.;]{0,180}\b(?:but|rather|instead)\b\s*" r"(?:rather\s+|instead\s+|also\s+)?by\b", folded, flags=re.I, ) if alternate_cause: return "AFFIRMATIVE_CAUSAL" # Contrast clauses must not let an early disclaimer shield a later assertion. clauses = re.split( r"(?i)(?:;|\bbut\b|\bhowever\b|\bwhereas\b|\byet\b|\bnevertheless\b|\balthough\b|\bthough\b)", value, ) states = [cls._classify_quantitative_causal_clause(part) for part in clauses if str(part).strip()] if "AFFIRMATIVE_CAUSAL" in states: return "AFFIRMATIVE_CAUSAL" if "EXPLICIT_CAUSAL_LIMITATION" in states: return "EXPLICIT_CAUSAL_LIMITATION" return "NON_CAUSAL" @classmethod def _causal_phrase_is_explicitly_disclaimed(cls, sentence, match_start=None, match_end=None): """Compatibility wrapper for older call sites; uses the shared tri-state contract.""" return cls.classify_quantitative_causal_sentence(sentence) == "EXPLICIT_CAUSAL_LIMITATION" @classmethod def _is_methodological_bias_mechanism_sentence(cls, sentence, known_constructs=None): """Recognize measurement/method-bias mechanisms without weakening construct causality. R4 separates methodological statements about how measurement conditions may bias statistical quantities from substantive claims that one study construct causes another. The exemption is intentionally narrow: a measurement/source-bias actor and a statistical/measurement target must both be present. Explicit construct-level causal language still wins. """ value = re.sub(r"\s+", " ", str(sentence or "")).strip().casefold() if not value: return False methodological_actor = re.search( r"\b(?:self[- ]reports?|single\s+questionnaire|same[- ]source|common\s+method(?:\s+bias)?|" r"response\s+styles?|social\s+desirability|measurement\s+error|item\s+wording|" r"common\s+rater|method\s+variance|single[- ]source\s+measurement)\b", value, re.I ) statistical_target = re.search( r"\b(?:correlations?|covariances?|associations?|estimates?|coefficients?|variance|" r"standard\s+errors?|reliability|validity|measurement\s+error|method\s+variance)\b", value, re.I ) mechanism = re.search( r"\b(?:inflate|inflates|inflated|attenuate|attenuates|attenuated|bias|biases|biased|" r"distort|distorts|distorted|increase|increases|increased|decrease|decreases|decreased|" r"contribute|contributes|contributed|induce|induces|induced)\b", value, re.I ) if not (methodological_actor and statistical_target and mechanism): return False # Never let a methodological phrase shield an explicit relation between # named study constructs. constructs = [str(x or '').strip().casefold() for x in (known_constructs or []) if str(x or '').strip()] named_hits = [c for c in constructs if c and c in value] if len(named_hits) >= 2 and re.search( r"\b(?:causes?|leads?\s+to|drives?|determines?|produces?|results?\s+in|yields?|generates?)\b", value, re.I ): return False return True @staticmethod def _normalize_quantitative_construct_surface(value): """Normalize a construct surface for conservative alias matching.""" text = str(value or "").casefold() text = re.sub(r"[^a-z0-9]+", " ", text) return re.sub(r"\s+", " ", text).strip() @classmethod def _quantitative_construct_alias_map(cls, constructs): """Return only unique, conservative aliases for frozen construct names. R6-R14 permits ordinary abstract compression such as ``medical tourism conversion intention`` -> ``conversion intention``. Aliases are suffixes of at least two lexical tokens and are admitted only when they identify exactly one frozen construct role. No semantic model or fuzzy embedding is used, so ambiguous aliases remain unresolved. """ role_values = { str(role): cls._normalize_quantitative_construct_surface(value) for role, value in dict(constructs or {}).items() if str(value or "").strip() } candidates = {} for role, canonical in role_values.items(): tokens = canonical.split() surfaces = {canonical} # Conservative reader-facing shortening: terminal 2-4 token suffixes. # One-word aliases (e.g. "intention") are intentionally forbidden. for width in range(2, min(4, len(tokens)) + 1): if width < len(tokens): alias = " ".join(tokens[-width:]) if len(alias) >= 8: surfaces.add(alias) for surface in surfaces: candidates.setdefault(surface, set()).add(role) unique = {surface: next(iter(roles)) for surface, roles in candidates.items() if len(roles) == 1} by_role = {role: [] for role in role_values} for surface, role in unique.items(): by_role.setdefault(role, []).append(surface) for role in by_role: by_role[role] = sorted(set(by_role[role]), key=lambda x: (-len(x), x)) return by_role @classmethod def _quantitative_construct_roles_in_text(cls, text, constructs): folded = cls._normalize_quantitative_construct_surface(text) roles = set() aliases = cls._quantitative_construct_alias_map(constructs) for role, surfaces in aliases.items(): for surface in surfaces: if re.search(rf"(?<![a-z0-9]){re.escape(surface)}(?![a-z0-9])", folded): roles.add(role) break return roles @staticmethod def _quantitative_metric_local_clause(sentence, metric_start, metric_end): """Bound one metric occurrence to its nearest relation clause. Strong punctuation/conjunction boundaries are preferred. If two metric occurrences remain in one span, an ``and`` between those metric surfaces becomes a boundary. This avoids treating two path estimates in one sentence as if they shared a single owner while not splitting construct names such as ``triage and itinerary planning``. """ text = str(sentence or "") metric_re = re.compile(r"(?i)(?:β|standardized\s+estimate)\s*(?:=|:)?\s*[-+]?\d+(?:\.\d+)?") metrics = list(metric_re.finditer(text)) boundaries = [0, len(text)] for match in re.finditer(r"(?i);|,\s+(?:and|but|while|whereas)\s+|\b(?:however|nevertheless)\b", text): boundaries.extend([match.start(), match.end()]) # Conjunctions between adjacent metric occurrences are relation boundaries. for left, right in zip(metrics, metrics[1:]): between = text[left.end():right.start()] joins = list(re.finditer(r"(?i)\band\b", between)) if joins: join = joins[-1] boundaries.extend([left.end() + join.start(), left.end() + join.end()]) left = max((b for b in boundaries if b <= metric_start), default=0) right = min((b for b in boundaries if b >= metric_end), default=len(text)) clause = text[left:right].strip(" ,;\t\n") return clause or text.strip() @classmethod def _quantitative_metric_owner_for_occurrence(cls, sentence, match, constructs): """Resolve a single standardized estimate to a frozen path owner.""" clause = cls._quantitative_metric_local_clause(sentence, match.start(), match.end()) roles = cls._quantitative_construct_roles_in_text(clause, constructs) folded = cls._normalize_quantitative_construct_surface(clause) has_indirect = bool(re.search(r"\bindirect\b", folded)) has_direct = bool(re.search(r"\b(?:direct|residual direct)\b", folded)) # Specific semantic labels win only when the construct frame is compatible. if has_indirect and {"predictor", "outcome"}.issubset(roles): return "indirect", clause if has_direct and {"predictor", "outcome"}.issubset(roles): return "direct", clause if roles == {"predictor", "mediator"}: return "a", clause if roles == {"mediator", "outcome"}: return "b", clause if roles == {"predictor", "outcome"} and has_direct: return "direct", clause # A three-construct clause without an explicit direct/indirect relation is # genuinely ambiguous and must not be guessed into a/b ownership. return None, clause @classmethod def _strict_standardized_metric_occurrences(cls, sentence, constructs, results): """Yield clause-bound owners for each beta/standardized-estimate occurrence.""" regex = re.compile(r"(?i)((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)") expected = {key: results.get(key) for key in ("a", "b", "direct", "indirect")} for match in regex.finditer(str(sentence or "")): owner, clause = cls._quantitative_metric_owner_for_occurrence(sentence, match, constructs) yield match, owner, expected.get(owner) if owner else None, clause @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)", } 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) method_scope_unknown = ( strict and ownership == "unknown" and bool(re.search(r"(?i)method|measure", str(section_name or ""))) and cls._first_performed_unowned_quantitative_term(sentence, forbidden) is not None ) if ownership not in {"study", "result"} and not method_scope_unknown: 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: methodological_bias_scope = cls._is_methodological_bias_mechanism_sentence( sentence, known_constructs ) semantic_frame = cls.build_quantitative_semantic_claim_frame( sentence, meta, section_name ) causal_state = str(semantic_frame.get("validation_state") or "NON_CAUSAL") if causal_state == "AFFIRMATIVE_CAUSAL": violations.append({ "type":"causal_language_violation", "preview":sentence.strip()[:180], "reason":"The sentence makes an affirmative causal claim that exceeds the cross-sectional observational evidence contract.", "repair_instruction":( "Preserve the study meaning but rewrite as association/covariation only. " "Explicit causal-limit language is allowed, including forms such as " "'restricted to associations, not causal effects' or 'cannot establish causality'." ), }) 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" or method_scope_unknown: method_frame = cls._first_performed_unowned_quantitative_term(sentence, forbidden) if method_frame: violations.append({ "type":"unsupported_quantitative_method", "method":method_frame.get("term"), "claim_state":method_frame.get("state"), "method_category":method_frame.get("category"), "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) # R6-R14 clause-bound metric ownership: validate each occurrence # against its own local construct relation, never one owner per sentence. for metric_match, metric_owner, metric_expected, metric_clause in cls._strict_standardized_metric_occurrences( sentence, constructs, results ): observed = metric_match.group(2) if metric_owner is None or metric_expected is None: violations.append({ "type":"quantitative_metric_owner_unresolved", "observed":observed, "preview":str(metric_clause or sentence).strip()[:180], "reason":"The standardized estimate cannot be uniquely bound to a frozen a/b/direct/indirect construct relation.", }) continue try: same = abs(float(observed) - float(metric_expected)) < 0.0006 except Exception: same = False if not same: expected_text = f"{float(metric_expected):.3f}".rstrip("0").rstrip(".") violations.append({ "type":"quantitative_metric_mismatch", "metric":metric_owner, "observed":observed, "expected":expected_text, "preview":str(metric_clause or sentence).strip()[:180], }) 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) method_scope_unknown = ( strict and ownership == "unknown" and bool(re.search(r"(?i)method|measure", str(section_name or ""))) and cls._first_performed_unowned_quantitative_term(sentence, forbidden) is not None ) if ownership not in {"study", "result"} and not method_scope_unknown: 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: causal_state = cls.classify_quantitative_causal_sentence(revised) if causal_state == "AFFIRMATIVE_CAUSAL": for pattern, replacement in causal_replacements: regex = re.compile(pattern) revised, repaired_count = regex.subn(replacement, 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" or method_scope_unknown: method_frame = cls._first_performed_unowned_quantitative_term(revised, forbidden) if method_frame: diagnostics.append({ "type":"unsupported_quantitative_method_removed", "method":method_frame.get("term"), "claim_state":method_frame.get("state"), "method_category":method_frame.get("category"), "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) # R6-R14 deterministic repair is also occurrence-local. Never # rewrite every metric in a sentence to one selected path value. metric_replacements = [] for metric_match, metric_owner, metric_expected, metric_clause in cls._strict_standardized_metric_occurrences( revised, constructs, results ): if metric_owner is None or metric_expected is None: continue observed = metric_match.group(2) try: same = abs(float(observed) - float(metric_expected)) < 0.0006 except Exception: same = False if same: continue expected_text = f"{float(metric_expected):.3f}".rstrip("0").rstrip(".") diagnostics.append({ "type":"quantitative_metric_repaired", "metric":metric_owner, "observed":observed, "expected":expected_text, "preview":str(metric_clause or revised).strip()[:180], }) metric_replacements.append((metric_match.start(2), metric_match.end(2), expected_text)) for metric_start, metric_end, replacement in sorted(metric_replacements, reverse=True): revised = revised[:metric_start] + replacement + revised[metric_end:] 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["semantic_claim_frame_version"] = self.SEMANTIC_CLAIM_FRAME_VERSION metadata["hypothesis_ledger"] = self.build_hypothesis_ledger(construct_names) metadata["quantitative_table_manifest"] = self.build_table_manifest( validity_df, fl_matrix_df, path_table, spearman ) 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()