Spaces:
Running
Running
| # -*- 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": "Ескертпе:", | |
| }, | |
| } | |
| 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) | |
| 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) | |
| 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<number>[1-4])\s*:\s*(?P<title>[^\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"])} | |
| 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} | |
| 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" | |
| 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 | |
| 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 | |
| ): | |
| replacement = str(target) | |
| text = re.sub( | |
| re.escape(source), | |
| lambda _match, value=replacement: value, | |
| text, | |
| flags=re.I, | |
| ) | |
| return text | |
| 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() | |
| 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, | |
| } | |
| 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() | |
| 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() | |
| 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) | |
| 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) | |
| 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] | |
| 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="", *, source_topic=None, source_keywords=None | |
| ): | |
| """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, | |
| source_topic=source_topic, source_keywords=source_keywords, | |
| ) | |
| 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, | |
| source_topic=source_topic, source_keywords=source_keywords, | |
| ) | |
| 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, | |
| source_topic=source_topic, source_keywords=source_keywords, | |
| ) | |
| 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, | |
| source_topic=source_topic, source_keywords=source_keywords, | |
| ) | |
| 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, | |
| source_topic=source_topic, source_keywords=source_keywords, | |
| ) | |
| 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 | |
| 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") | |
| 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 | |
| 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 | |
| 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" | |
| def _has_explicit_current_study_anchor(sentence): | |
| """Return True only for an explicit *current manuscript* ownership anchor. | |
| A literature citation does not transfer ownership of a clause such as | |
| ``We tested ... (Author, year)`` to the cited source. This helper is | |
| intentionally narrower than the general sentence-ownership heuristic: | |
| generic phrases such as ``the study`` are excluded because they may | |
| refer to a cited study. | |
| """ | |
| value = str(sentence or "") | |
| return bool(re.search( | |
| r"(?i)\b(?:we|our\s+study|this\s+study|the\s+present\s+study|" | |
| r"our\s+analysis|our\s+model|our\s+sample|our\s+results|" | |
| r"in\s+this\s+study|in\s+our\s+study)\b", | |
| value, | |
| )) | |
| 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 | |
| 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 [], | |
| } | |
| 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"} | |
| # The original local-window classifier can miss a long current-study | |
| # frame such as ``We tested the hypotheses ... using partial least | |
| # squares structural equation modeling (PLS-SEM)`` because the action | |
| # verb may sit more than 70 characters before the forbidden term. A | |
| # citation appended to the same sentence must not make that performed | |
| # method claim externally owned. Detect the explicit current-study | |
| # action across the whole pre-term clause before applying the narrower | |
| # legacy patterns below. | |
| full_before = folded[:match.start()] | |
| if re.search( | |
| r"\b(?:did\s+not|does\s+not|do\s+not|was\s+not|were\s+not|never)\s+" | |
| r"(?!only\b)(?:use|uses|used|employ|employs|employed|conduct|conducts|conducted|" | |
| r"perform|performs|performed|estimate|estimates|estimated|test|tests|tested|" | |
| r"assess|assesses|assessed|evaluate|evaluates|evaluated|implement|implements|" | |
| r"implemented|apply|applies|applied|fit|fits|fitted|model|models|modeled|modelled)\b" | |
| r"[^.;:]{0,240}$", | |
| full_before, | |
| ): | |
| return {"term": label, "category": category, "state": "NOT_PERFORMED_OR_LIMITATION"} | |
| current_study_performed = bool( | |
| cls._has_explicit_current_study_anchor(text) | |
| and re.search( | |
| r"\b(?:use|uses|used|employ|employs|employed|conduct|conducts|conducted|" | |
| r"perform|performs|performed|estimate|estimates|estimated|test|tests|tested|" | |
| r"assess|assesses|assessed|evaluate|evaluates|evaluated|implement|implements|" | |
| r"implemented|apply|applies|applied|ran|run|runs|fit|fits|fitted|model|models|" | |
| r"modeled|modelled|control|controls|controlled|correct|corrects|corrected|" | |
| r"address|addresses|addressed)\b[^.;:]{0,240}$", | |
| full_before, | |
| ) | |
| ) | |
| performed = bool( | |
| current_study_performed | |
| or | |
| 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", | |
| } | |
| 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 | |
| def _semantic_frame_hash(payload): | |
| canonical = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) | |
| return hashlib.sha256(canonical.encode("utf-8")).hexdigest() | |
| 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 | |
| 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" | |
| 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" | |
| 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" | |
| 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 | |
| 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() | |
| 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 | |
| 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 | |
| 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() | |
| 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 | |
| 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 | |
| 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) | |
| current_study_method_frame = ( | |
| cls._first_performed_unowned_quantitative_term(sentence, forbidden) | |
| if strict and cls._has_explicit_current_study_anchor(sentence) | |
| else None | |
| ) | |
| method_scope_hybrid = bool( | |
| strict | |
| and ownership == "external" | |
| and current_study_method_frame is not None | |
| ) | |
| 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 method_scope_hybrid: | |
| violations.append({ | |
| "type":"unsupported_quantitative_method", | |
| "method":current_study_method_frame.get("term"), | |
| "claim_state":current_study_method_frame.get("state"), | |
| "method_category":current_study_method_frame.get("category"), | |
| "preview":sentence.strip()[:180], | |
| }) | |
| # The literature-cited part remains externally owned. Do not | |
| # run unrelated metric/role rewrites across this hybrid | |
| # sentence; the method violation alone is sufficient to force | |
| # bounded regeneration/fail-closed handling. | |
| continue | |
| 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 | |
| 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) | |
| current_study_method_frame = ( | |
| cls._first_performed_unowned_quantitative_term(sentence, forbidden) | |
| if strict and cls._has_explicit_current_study_anchor(sentence) | |
| else None | |
| ) | |
| method_scope_hybrid = bool( | |
| strict | |
| and ownership == "external" | |
| and current_study_method_frame is not None | |
| ) | |
| 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 method_scope_hybrid: | |
| diagnostics.append({ | |
| "type":"unsupported_quantitative_method_removed", | |
| "method":current_study_method_frame.get("term"), | |
| "claim_state":current_study_method_frame.get("state"), | |
| "method_category":current_study_method_frame.get("category"), | |
| "preview":sentence.strip()[:180], | |
| }) | |
| replacements.append((start, end, "")) | |
| continue | |
| 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 | |
| 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="", | |
| internal_topic="", | |
| internal_keywords=None, | |
| 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_topic = str(internal_topic or topic or "").strip() | |
| design_keywords = internal_keywords if internal_keywords not in (None, "", []) else keywords | |
| design_contract = self._derive_system_design( | |
| agent, design_topic, design_keywords, | |
| source_topic=topic, source_keywords=keywords, | |
| ) | |
| issues = validate_quantitative_contract(design_contract) | |
| if issues: | |
| p1_identity_issues = [ | |
| issue for issue in issues | |
| if issue.startswith("internal_english_") or issue.startswith("quantitative_construct_") | |
| ] | |
| if p1_identity_issues: | |
| raise RuntimeError(f"quantitative_construct_identity_corruption:{p1_identity_issues[:4]}") | |
| 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() | |