# -*- 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 from core.study_design_contract import ( STUDY_DESIGN_CONTRACT_VERSION, build_quantitative_contract, validate_quantitative_contract, compile_quantitative_hypothesis_ledger, build_quantitative_front_matter_contract, build_user_research_intent, ) 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" FRONT_MATTER_CONTRACT_VERSION = "quantitative-front-matter-contract-v2" TARGET_TABLE_BINDING_VERSION = "target-quantitative-table-binding-v1" SEMANTIC_CLAIM_FRAME_VERSION = "smartpaper-core-semantic-claim-frame-v1" STUDY_DESIGN_CONTRACT_VERSION = STUDY_DESIGN_CONTRACT_VERSION _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_front_matter_contract(cls, topic, keywords, constructs): """Build a reader-front-matter semantic contract from adaptive design authority. ``constructs`` accepts either the new Study Design Contract or the legacy three-construct list/dict so old checkpoints remain readable. """ if isinstance(constructs, dict) and constructs.get("version") == cls.STUDY_DESIGN_CONTRACT_VERSION: return build_quantitative_front_matter_contract(constructs, version=cls.FRONT_MATTER_CONTRACT_VERSION) if isinstance(constructs, dict) and "quantitative" in constructs: return build_quantitative_front_matter_contract(constructs, version=cls.FRONT_MATTER_CONTRACT_VERSION) if isinstance(constructs, dict): values = [constructs.get(role) for role in ("predictor", "mediator", "outcome")] else: values = list(constructs or []) candidate = {"constructs": values, "research_context": "respondents with experience relevant to the focal management phenomenon"} contract = build_quantitative_contract(topic, keywords, candidate, source_type="system_generated") return build_quantitative_front_matter_contract(contract, version=cls.FRONT_MATTER_CONTRACT_VERSION) @classmethod def build_hypothesis_ledger(cls, constructs): """Compile H1-Hn from a frozen adaptive graph; preserve legacy H1-H3 order.""" if isinstance(constructs, dict) and constructs.get("version") == cls.STUDY_DESIGN_CONTRACT_VERSION: contract = constructs elif isinstance(constructs, dict) and "quantitative" in constructs: contract = constructs else: values = list(constructs.values()) if isinstance(constructs, dict) else list(constructs or []) contract = build_quantitative_contract("", "", {"constructs": values}, source_type="system_generated") return compile_quantitative_hypothesis_ledger(contract, version=cls.HYPOTHESIS_LEDGER_VERSION) @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): """Legacy helper retained for tests/checkpoints; no longer defines model topology.""" 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[:8] @staticmethod def _contract_construct_labels(contract): return [ str(item.get("label") or "").strip() for item in dict(contract or {}).get("quantitative", {}).get("constructs", []) if isinstance(item, dict) and str(item.get("label") or "").strip() ] def _derive_system_design(self, agent, topic="", keywords=""): """Interpret a parsimonious topic-aligned quantitative graph adaptively. The existing ``structured_json`` logical node is intentionally reused so Standard/Flash and Flagship/Pro ModelPolicy routing remains unchanged. """ user_intent = build_user_research_intent(topic, keywords, mode="quantitative") fallback_candidate = { "constructs": [ {"label":"Focal Predictor","roles":["predictor"]}, {"label":"Focal Mediator","roles":["mediator"]}, {"label":"Focal Outcome","roles":["outcome"]}, ], "relations": [ {"type":"direct_association","source":"Focal Predictor","target":"Focal Outcome","hypothesis_order":1}, {"type":"direct_association","source":"Focal Predictor","target":"Focal Mediator","hypothesis_order":2}, {"type":"direct_association","source":"Focal Mediator","target":"Focal Outcome","hypothesis_order":3}, ], "indirect_effects":[{"path":["Focal Predictor","Focal Mediator","Focal Outcome"],"kind":"simple"}], "research_context":"respondents with experience relevant to the focal management phenomenon", "background_concepts":[], } if agent is None: return build_quantitative_contract(topic, keywords, fallback_candidate) prompt = f"""Design the minimum coherent quantitative management-study graph for the topic below. Return ONLY valid JSON with this schema: {{ "constructs":[{{"label":"construct","roles":["predictor|mediator|moderator|outcome|control|covariate"]}}], "relations":[{{"type":"direct_association|moderation|control_effect","source":"construct label","target":"construct label","moderator":"construct label or null","hypothesis_order":1}}], "indirect_effects":[{{"path":["construct A","construct B","construct C"],"kind":"simple|serial|parallel"}}], "research_context":"one generic respondent context", "background_concepts":["topic concept that is contextual rather than measured/modelled"] }} Rules: 0. USER RESEARCH INTENT AUTHORITY: The user's topic and keywords are deliberate research-design inputs. Never delete, replace, or downgrade an explicitly requested focal construct, mediation, moderation, serial/parallel mediation, control, population, or research context merely because direct literature support may be sparse. Literature evidence constrains claims about PRIOR research; it does not redefine what this study is asked to investigate. 1. Use 2-8 distinct constructs only when the topic actually requires them; do NOT force a mediator or moderator that the user did not request. 2. Preserve explicit mediation, moderation, serial/parallel mediation, multiple antecedents/outcomes, and controls when the topic clearly requests them. When the requested relation is novel or direct prior evidence is sparse, keep the relation in the study graph and let later hypothesis development use cautious author inference rather than inventing citation support. 3. A moderation relation means source × moderator is associated with target. 4. Keep the graph parsimonious and acyclic for ordinary cross-sectional regression interpretation. 5. Use concise construct labels suitable for a journal article. 6. Put contextual topic terms that are not measured/modelled in background_concepts instead of pretending they are focal variables. 7. The respondent context must be generic and directly relevant. Do NOT invent a country, university, firm, named industry sample frame, software package, date, or institution. 8. This is a cross-sectional observational survey; do not imply causal identification. 9. 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=1200, temp=0.0, model_role="fast", thinking=False, allow_fallback=True, logical_node='structured_json', ) parsed = agent._extract_json_from_text(raw) contract = build_quantitative_contract(topic, keywords, parsed if isinstance(parsed, dict) else fallback_candidate) issues = validate_quantitative_contract(contract) if issues: logging.warning("Adaptive quantitative design validation requires intent-preserving fallback: %s", issues[:3]) fallback = build_quantitative_contract(topic, keywords, fallback_candidate) fallback_issues = validate_quantitative_contract(fallback) if fallback_issues: raise RuntimeError(f"quantitative_user_intent_unresolved:{fallback_issues[:3]}") return fallback context = str(dict(contract.get("context") or {}).get("research_context") or "") if re.search(r"\b(university|company|corporation|inc\.?|ltd\.?|country|china|usa|united states|europe|asia)\b", context, re.I): fallback_context = "respondents with experience relevant to the focal management phenomenon" contract = dict(contract) contract["context"] = {"research_context": fallback_context} # Rebuild through deterministic normalizer so the hash stays authoritative. candidate = parsed if isinstance(parsed, dict) else fallback_candidate candidate = dict(candidate) candidate["research_context"] = fallback_context contract = build_quantitative_contract(topic, keywords, candidate) return contract except Exception as exc: logging.warning("Quantitative adaptive study-design freeze fell back safely: %s", exc) return build_quantitative_contract(topic, keywords, fallback_candidate) 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 def _compile_graph_evidence(self, mean_scores, contract, bootstrap_iterations=1500): """Compile graph-owned associations/indirect effects/moderation deterministically.""" quant = dict((contract or {}).get("quantitative") or {}) constructs = { str(item.get("construct_id") or ""): str(item.get("label") or "").strip() for item in (quant.get("constructs") or []) if isinstance(item, dict) } relations = [dict(item) for item in (quant.get("relations") or []) if isinstance(item, dict)] if not constructs or not relations: return {}, pd.DataFrame(), mean_scores.corr(method="spearman") # Build one standardized OLS per target so multiple antecedents/controls are estimated together. direct_like = [r for r in relations if r.get("type") in {"direct_association", "control_effect"}] moderation = [r for r in relations if r.get("type") == "moderation"] relation_results = {} r2_by_construct = {} model_specs = {} for target in sorted({r.get("target") for r in relations if r.get("target")}): predictors = [] predictor_meta = [] for r in direct_like: if r.get("target") == target and r.get("source") in constructs: predictors.append(mean_scores[constructs[r["source"]]].to_numpy(dtype=float)) predictor_meta.append(("relation", r)) for r in moderation: if r.get("target") != target: continue sid, mid = r.get("source"), r.get("moderator") if sid not in constructs or mid not in constructs: continue x = mean_scores[constructs[sid]].to_numpy(dtype=float) m = mean_scores[constructs[mid]].to_numpy(dtype=float) # Main effects are included when not already present in the target model. existing_sources = {meta[1].get("source") for meta in predictor_meta if meta[0] == "relation"} if sid not in existing_sources: predictors.append(x); predictor_meta.append(("main", {"source": sid, "target": target, "label": constructs[sid]})) existing_sources.add(sid) if mid not in existing_sources: predictors.append(m); predictor_meta.append(("main", {"source": mid, "target": target, "label": constructs[mid]})) existing_sources.add(mid) xz = (x - x.mean()) / (x.std(ddof=0) or 1.0) mz = (m - m.mean()) / (m.std(ddof=0) or 1.0) predictors.append(xz * mz) predictor_meta.append(("interaction", r)) if not predictors or target not in constructs: continue y = mean_scores[constructs[target]].to_numpy(dtype=float) coef, r2 = self._standardized_ols(y, predictors) r2_by_construct[target] = round(float(r2), 3) model_specs[target] = (y, predictors, predictor_meta, coef) for index, (kind, meta) in enumerate(predictor_meta): estimate = round(float(coef[index]), 3) if kind == "relation": rid = str(meta.get("relation_id") or "") relation_results[rid] = { "relation_id": rid, "type": meta.get("type"), "source": constructs.get(meta.get("source"), ""), "target": constructs.get(meta.get("target"), ""), "estimate": estimate, "r2_target": round(float(r2), 3), } elif kind == "interaction": rid = str(meta.get("relation_id") or "") relation_results[rid] = { "relation_id": rid, "type": "moderation", "source": constructs.get(meta.get("source"), ""), "moderator": constructs.get(meta.get("moderator"), ""), "target": constructs.get(meta.get("target"), ""), "estimate": estimate, "r2_target": round(float(r2), 3), "simple_slope_low": None, "simple_slope_high": None, } # Bootstrap relation and indirect-effect CIs from the same compiled model topology. rng = np.random.default_rng(20260809) relation_boot = {rid: [] for rid in relation_results} indirect_boot = {str(e.get("effect_id") or ""): [] for e in (quant.get("indirect_effects") or []) if isinstance(e, dict)} n = len(mean_scores) for _ in range(int(bootstrap_iterations)): idx = rng.integers(0, n, n) boot_estimates = {} try: for target, (y, predictors, predictor_meta, _coef) in model_specs.items(): yb = y[idx] xb = [arr[idx] for arr in predictors] coef_b, _ = self._standardized_ols(yb, xb) for j, (kind, meta) in enumerate(predictor_meta): if kind in {"relation", "interaction"}: rid = str(meta.get("relation_id") or "") boot_estimates[rid] = float(coef_b[j]) if rid in relation_boot: relation_boot[rid].append(float(coef_b[j])) for effect in quant.get("indirect_effects") or []: if not isinstance(effect, dict): continue eid = str(effect.get("effect_id") or "") path = list(effect.get("path") or []) product = 1.0 valid = True for a, b in zip(path, path[1:]): rid = next((str(r.get("relation_id") or "") for r in direct_like if r.get("source") == a and r.get("target") == b), "") if not rid or rid not in boot_estimates: valid = False; break product *= boot_estimates[rid] if valid and eid in indirect_boot: indirect_boot[eid].append(float(product)) except Exception: continue for rid, values in relation_boot.items(): if len(values) >= 100: lo, hi = np.percentile(values, [2.5, 97.5]) relation_results[rid]["ci"] = [round(float(lo), 3), round(float(hi), 3)] # Recover simple slopes for moderation from standardized main + interaction terms. for r in moderation: rid = str(r.get("relation_id") or "") result = relation_results.get(rid) if not result: continue target = r.get("target") _y, _preds, meta_rows, coef = model_specs.get(target, (None, None, [], [])) main_beta = None for j, (kind, meta) in enumerate(meta_rows): if kind in {"relation", "main"} and meta.get("source") == r.get("source"): main_beta = float(coef[j]); break if main_beta is not None: interaction_beta = float(result["estimate"]) result["simple_slope_low"] = round(main_beta - interaction_beta, 3) result["simple_slope_high"] = round(main_beta + interaction_beta, 3) indirect_results = {} for effect in quant.get("indirect_effects") or []: if not isinstance(effect, dict): continue eid = str(effect.get("effect_id") or "") path = list(effect.get("path") or []) product = 1.0 labels = [] valid = True for cid in path: labels.append(constructs.get(cid, cid)) for a, b in zip(path, path[1:]): rid = next((str(r.get("relation_id") or "") for r in direct_like if r.get("source") == a and r.get("target") == b), "") if not rid or rid not in relation_results: valid = False; break product *= float(relation_results[rid]["estimate"]) if not valid: continue row = {"effect_id":eid,"path":labels,"estimate":round(product,3),"kind":effect.get("kind") or "simple"} vals = indirect_boot.get(eid) or [] if len(vals) >= 100: lo, hi = np.percentile(vals,[2.5,97.5]); row["ci"]=[round(float(lo),3),round(float(hi),3)] indirect_results[eid] = row path_rows = [] for r in relations: rid = str(r.get("relation_id") or "") rr = relation_results.get(rid) if not rr: continue if r.get("type") == "moderation": label = f"{rr['source']} × {rr['moderator']} → {rr['target']} (interaction)" else: label = f"{rr['source']} → {rr['target']}" path_rows.append({"Association":label,"Standardized estimate":rr["estimate"],"R² of outcome":rr.get("r2_target","—")}) for row in indirect_results.values(): path_rows.append({"Association":"Indirect association: " + " → ".join(row["path"]),"Standardized estimate":row["estimate"],"R² of outcome":"—"}) path_table = pd.DataFrame(path_rows) # Conservative multicollinearity summary across ordinary construct scores. vif_values = [] for target, (_y, predictors, _meta, _coef) in model_specs.items(): if len(predictors) < 2: continue matrix = np.column_stack(predictors) corr = np.corrcoef(matrix, rowvar=False) try: inv = np.linalg.pinv(corr) vif_values.extend(float(x) for x in np.diag(inv)) except Exception: pass summary = { "relation_results": relation_results, "indirect_results": indirect_results, "r2_by_construct": {constructs.get(k,k):v for k,v in r2_by_construct.items()}, "max_vif": round(max(vif_values),3) if vif_values else 1.0, "bootstrap_iterations": int(bootstrap_iterations), "study_design_contract_sha256": str((contract or {}).get("contract_sha256") or ""), } # Exact historical aliases keep legacy validators/checkpoints stable. if bool(quant.get("legacy_three_construct_profile")): p = next((x for x in quant.get("constructs") or [] if "predictor" in (x.get("roles") or [])), {}) m = next((x for x in quant.get("constructs") or [] if "mediator" in (x.get("roles") or [])), {}) o = next((x for x in quant.get("constructs") or [] if "outcome" in (x.get("roles") or [])), {}) pl, ml, ol = p.get("label"), m.get("label"), o.get("label") def est(src,tgt): for rr in relation_results.values(): if rr.get("source") == src and rr.get("target") == tgt and rr.get("type") == "direct_association": return rr.get("estimate") return None a, b, direct = est(pl,ml), est(ml,ol), est(pl,ol) indirect = next(iter(indirect_results.values()), {}) summary.update({ "predictor":pl,"mediator":ml,"outcome":ol, "a":a,"b":b,"direct":direct,"indirect":indirect.get("estimate"), "r2_mediator":summary["r2_by_construct"].get(ml), "r2_outcome":summary["r2_by_construct"].get(ol), "indirect_ci":indirect.get("ci"), "direct_ci":next((rr.get("ci") for rr in relation_results.values() if rr.get("source")==pl and rr.get("target")==ol), None), "predictor_mediator_vif":summary["max_vif"], }) return summary, path_table, mean_scores.corr(method="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 {} construct_items = [dict(item) for item in (meta.get("construct_items") or []) if isinstance(item, dict)] results = dict(meta.get("association_results") or {}) if isinstance(meta.get("association_results"), dict) else {} study_contract = dict(meta.get("study_design_contract") or {}) if isinstance(meta.get("study_design_contract"), dict) else {} analysis_plan = list(dict(study_contract.get("quantitative") or {}).get("analysis_plan") or []) allowed_methods = [ "reliability", "composite reliability", "average variance extracted", "Fornell-Larcker", "standardized association", "bootstrap", "Spearman", "VIF", "descriptive statistics", ] if "interaction_test" in analysis_plan: allowed_methods.extend(["interaction term", "moderation", "simple slopes"]) 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(), "study_design_contract": study_contract, "constructs": {k: str(v or "").strip() for k, v in constructs.items() if str(v or "").strip()}, "construct_items": construct_items, "association_results": results, "uploaded_metric_anchors": uploaded_metric_anchors, "causal_identification": "none" if strict else "evidence_owned", "allowed_methods": allowed_methods 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 legacy and adaptive graph estimates.""" regex = re.compile(r"(?i)((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)") legacy_expected = {key: results.get(key) for key in ("a", "b", "direct", "indirect")} relation_results = dict(results.get("relation_results") or {}) if isinstance(results, dict) else {} indirect_results = dict(results.get("indirect_results") or {}) if isinstance(results, dict) else {} for match in regex.finditer(str(sentence or "")): owner, clause = cls._quantitative_metric_owner_for_occurrence(sentence, match, constructs) if owner and legacy_expected.get(owner) is not None: yield match, owner, legacy_expected.get(owner), clause continue folded = cls._normalize_quantitative_construct_surface(clause) candidates = [] for rid, row in relation_results.items(): if not isinstance(row, dict): continue source = cls._normalize_quantitative_construct_surface(row.get("source")) target = cls._normalize_quantitative_construct_surface(row.get("target")) moderator = cls._normalize_quantitative_construct_surface(row.get("moderator")) if not source or not target or source not in folded or target not in folded: continue if row.get("type") == "moderation": if not moderator or moderator not in folded or not re.search(r"(?i)interaction|moderat|×", clause): continue elif re.search(r"(?i)interaction|moderat|×", clause): continue if row.get("estimate") is not None: candidates.append((f"relation:{rid}", row.get("estimate"))) if re.search(r"(?i)\bindirect\b", clause): for eid, row in indirect_results.items(): if not isinstance(row, dict): continue labels = [cls._normalize_quantitative_construct_surface(x) for x in (row.get("path") or [])] if labels and all(label in folded for label in labels) and row.get("estimate") is not None: candidates.append((f"indirect:{eid}", row.get("estimate"))) if len(candidates) == 1: yield match, candidates[0][0], candidates[0][1], clause else: yield match, None, 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 results.get("predictor_mediator_vif") is not None else results.get("max_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) for construct_label, expected_r2 in dict(results.get("r2_by_construct") or {}).items(): if construct_label and expected_r2 is not None: add_metric_violations(sentence, rf"(R(?:²|\^2)\s*(?:for|of)?\s*{re.escape(str(construct_label))}\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", expected_r2, label=f"r2:{construct_label}") # 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 results.get("predictor_mediator_vif") is not None else results.get("max_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) for construct_label, expected_r2 in dict(results.get("r2_by_construct") or {}).items(): if construct_label and expected_r2 is not None: metric_pattern = rf"(R(?:²|\^2)\s*(?:for|of)?\s*{re.escape(str(construct_label))}\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)" replace_local_metric(metric_pattern, expected_r2, label=f"r2:{construct_label}") # 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_contract = self._derive_system_design(agent, topic, keywords) issues = validate_quantitative_contract(design_contract) if issues: raise RuntimeError(f"adaptive_quantitative_study_design_invalid:{issues[:4]}") quant = dict(design_contract.get("quantitative") or {}) construct_items = [dict(item) for item in (quant.get("constructs") or []) if isinstance(item, dict)] construct_names = [str(item.get("label") or "").strip() for item in construct_items if str(item.get("label") or "").strip()] if len(construct_names) < 2: raise RuntimeError("adaptive_quantitative_study_design_has_too_few_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._compile_graph_evidence(mean_scores, design_contract) context = str(dict(design_contract.get("context") or {}).get("research_context") or "").strip() role_lines = [] for item in construct_items: role_lines.append( f"| {item.get('construct_id')} | {item.get('label')} | {', '.join(item.get('roles') or [])} |" ) relation_lines = [] by_id = {item.get("construct_id"): item.get("label") for item in construct_items} for rel in quant.get("relations") or []: if not isinstance(rel, dict): continue source = by_id.get(rel.get("source"), rel.get("source")) target = by_id.get(rel.get("target"), rel.get("target")) moderator = by_id.get(rel.get("moderator"), rel.get("moderator")) if rel.get("moderator") else "—" relation_lines.append( f"| {rel.get('relation_id')} | {rel.get('type')} | {source} | {target} | {moderator} |" ) bootstrap_lines = [f"- Bootstrap resamples: {path_summary.get('bootstrap_iterations', 1500)}."] for rid, row in dict(path_summary.get("relation_results") or {}).items(): label = f"{row.get('source')} → {row.get('target')}" if row.get("type") == "moderation": label = f"{row.get('source')} × {row.get('moderator')} → {row.get('target')}" ci = row.get("ci") or [] ci_text = f"; 95% bootstrap CI [{ci[0]}, {ci[1]}]" if len(ci) == 2 else "" extra = "" if row.get("type") == "moderation" and row.get("simple_slope_low") is not None: extra = f"; standardized simple slopes at moderator -1 SD/+1 SD: {row.get('simple_slope_low')}/{row.get('simple_slope_high')}" bootstrap_lines.append(f"- {rid} {label}: standardized estimate {row.get('estimate')}{ci_text}{extra}.") for eid, row in dict(path_summary.get("indirect_results") or {}).items(): ci = row.get("ci") or [] ci_text = f"; 95% bootstrap CI [{ci[0]}, {ci[1]}]" if len(ci) == 2 else "" bootstrap_lines.append( f"- {eid} indirect association ({' → '.join(row.get('path') or [])}): {row.get('estimate')}{ci_text}." ) bootstrap_lines.append(f"- Maximum predictor-set VIF across compiled regression models: {path_summary.get('max_vif', 1.0)}.") 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 | | Causal identification | None; report associations, indirect associations, and interaction associations only; do not claim causal effects | | Study design contract | {design_contract.get('contract_id')} | ### Frozen Construct Graph | Construct ID | Construct | Role(s) | | :--- | :--- | :--- | {chr(10).join(role_lines)} ### Frozen Relation Graph | Relation ID | Type | Source | Target | Moderator | | :--- | :--- | :--- | :--- | :--- | {chr(10).join(relation_lines)} ### 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, Interaction, and Collinearity Checks {chr(10).join(bootstrap_lines)} [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: The frozen Study Design Contract and relation graph are the sole authority for what this study measures/models. Never add a mediator, moderator, outcome, control, interaction, indirect path, or tested relation that is absent from the graph. 3. INFERENCE DISCIPLINE: This is a cross-sectional observational survey. Use association language. Indirect estimates are not causal mediation identification; interaction estimates are associational moderation tests. 4. MEASUREMENT DEFENSE: State reliability/convergent/discriminant validity only when the displayed values support it. 5. RESULTS DISCIPLINE: Results must report this study's frozen numbers without attaching external citations to coefficients or sample facts. 6. ROBUSTNESS: The Spearman matrix, frozen bootstrap intervals, simple slopes when present, and VIF above are the only frozen robustness evidence. Do not add another robustness procedure. 7. CONSISTENCY: Construct identities, roles, relation IDs, sample size, context, scale, and all numerical results above are the single source of quantitative truth throughout the manuscript. """ role_map = {} for role in ("predictor", "mediator", "moderator", "outcome", "control", "covariate"): vals = [item.get("label") for item in construct_items if role in (item.get("roles") or [])] if len(vals) == 1: role_map[role] = vals[0] 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, "study_design_contract": design_contract, "constructs": role_map, "construct_items": construct_items, "association_results": path_summary, "method": ( "Positive-semidefinite latent correlation generation, indicator loadings, Likert discretization, " "reliability/validity summaries, Study-Design-Contract-compiled standardized association models, " "fixed-seed bootstrap intervals, interaction/simple-slope estimates when frozen, and Spearman robustness correlations." ), "generation_parameters": { "constructs": construct_names, "study_design_contract_sha256": design_contract.get("contract_sha256"), "analysis_plan": list(quant.get("analysis_plan") or []), "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.get("bootstrap_iterations", 1500), "bootstrap_seed": 20260809, "reported_summaries": [ "Cronbach alpha", "composite reliability", "average variance extracted", "descriptive statistics", "Fornell-Larcker matrix", "compiled standardized association estimates", "R-squared", "bootstrap confidence intervals", "interaction/simple-slope estimates when frozen", "Spearman correlations", "VIF", ], }, "evidence_sections": [ "Frozen quantitative study design", "Frozen construct graph", "Frozen relation graph", "Reliability and convergent validity", "Descriptive statistics and discriminant validity", "Frozen association model results", "Rank-order robustness correlations", "Frozen bootstrap, interaction, and collinearity checks", ], } metadata["semantic_claim_frame_version"] = self.SEMANTIC_CLAIM_FRAME_VERSION metadata["front_matter_contract"] = self.build_front_matter_contract(topic, keywords, design_contract) metadata["hypothesis_ledger"] = self.build_hypothesis_ledger(design_contract) 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()