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 | |
| class QuantitativeEngine: | |
| """Prepare one frozen quantitative evidence packet before manuscript writing.""" | |
| QUANTITATIVE_EVIDENCE_LEDGER_VERSION = "quantitative-evidence-ledger-v2" | |
| QUANTITATIVE_TABLE_MANIFEST_VERSION = "quantitative-table-manifest-v1" | |
| HYPOTHESIS_LEDGER_VERSION = "quantitative-hypothesis-ledger-v1" | |
| TARGET_TABLE_BINDING_VERSION = "target-quantitative-table-binding-v1" | |
| SEMANTIC_CLAIM_FRAME_VERSION = "smartpaper-core-semantic-claim-frame-v1" | |
| _TARGET_TABLE_LOCALIZATION = { | |
| "zh": { | |
| "caption": "表 {number}：{title}", | |
| "titles": { | |
| 1: "信度与收敛效度", | |
| 2: "描述性统计与判别效度（Fornell-Larcker 准则）", | |
| 3: "关联模型结果", | |
| 4: "秩次稳健性相关（Spearman）", | |
| }, | |
| "cells": { | |
| "Construct": "构念", "Items": "题项数", "Cronbach's α": "Cronbach α", | |
| "Mean": "均值", "SD": "SD", "Association": "关联关系", | |
| "Standardized estimate": "标准化估计", "R² of outcome": "结果变量的 R²", | |
| "Path": "路径", "Beta": "Beta", "Estimate": "估计量", | |
| "Point estimate": "点估计", "95% bootstrap CI lower": "95% bootstrap CI 下限", | |
| "95% bootstrap CI upper": "95% bootstrap CI 上限", | |
| }, | |
| "phrases": { | |
| "Indirect association:": "间接关联：", | |
| "Indirect association": "间接关联", | |
| "Residual direct association": "剩余直接关联", | |
| "controlling": "控制", | |
| }, | |
| "notes": { | |
| "Diagonal elements in parentheses are the square roots of AVE.": "括号中的对角线元素为 AVE 的平方根。", | |
| }, | |
| "note_prefix": "注：", | |
| }, | |
| "ru": { | |
| "caption": "Таблица {number}: {title}", | |
| "titles": { | |
| 1: "Надёжность и конвергентная валидность", | |
| 2: "Описательная статистика и дискриминантная валидность (критерий Форнелла–Ларкера)", | |
| 3: "Результаты модели ассоциаций", | |
| 4: "Ранговые корреляции для проверки устойчивости (Спирмен)", | |
| }, | |
| "cells": { | |
| "Construct": "Конструкт", "Items": "Пункты", "Cronbach's α": "α Кронбаха", | |
| "Mean": "Среднее", "SD": "SD", "Association": "Ассоциация", | |
| "Standardized estimate": "Стандартизированная оценка", "R² of outcome": "R² результата", | |
| "Path": "Связь", "Beta": "Beta", "Estimate": "Оценка", | |
| "Point estimate": "Точечная оценка", "95% bootstrap CI lower": "Нижняя граница 95% bootstrap CI", | |
| "95% bootstrap CI upper": "Верхняя граница 95% bootstrap CI", | |
| }, | |
| "phrases": { | |
| "Indirect association:": "Косвенная ассоциация:", | |
| "Indirect association": "Косвенная ассоциация", | |
| "Residual direct association": "Остаточная прямая ассоциация", | |
| "controlling": "при контроле", | |
| }, | |
| "notes": { | |
| "Diagonal elements in parentheses are the square roots of AVE.": "Диагональные элементы в скобках представляют квадратные корни из AVE.", | |
| }, | |
| "note_prefix": "Примечание:", | |
| }, | |
| "kk": { | |
| "caption": "{number}-кесте: {title}", | |
| "titles": { | |
| 1: "Сенімділік және конвергенттік валидтілік", | |
| 2: "Сипаттамалық статистика және дискриминанттық валидтілік (Форнелл–Ларкер критерийі)", | |
| 3: "Байланыстар моделінің нәтижелері", | |
| 4: "Орнықтылықты тексеруге арналған рангтік корреляциялар (Спирмен)", | |
| }, | |
| "cells": { | |
| "Construct": "Конструкт", "Items": "Тармақтар", "Cronbach's α": "Кронбах α", | |
| "Mean": "Орташа", "SD": "SD", "Association": "Байланыс", | |
| "Standardized estimate": "Стандартталған бағалау", "R² of outcome": "Нәтиженің R²", | |
| "Path": "Байланыс", "Beta": "Beta", "Estimate": "Бағалау", | |
| "Point estimate": "Нүктелік бағалау", "95% bootstrap CI lower": "95% bootstrap CI төменгі шегі", | |
| "95% bootstrap CI upper": "95% bootstrap CI жоғарғы шегі", | |
| }, | |
| "phrases": { | |
| "Indirect association:": "Жанама байланыс:", | |
| "Indirect association": "Жанама байланыс", | |
| "Residual direct association": "Қалдық тікелей байланыс", | |
| "controlling": "бақылау кезінде", | |
| }, | |
| "notes": { | |
| "Diagonal elements in parentheses are the square roots of AVE.": "Жақшадағы диагональ элементтер AVE квадрат түбірлерін көрсетеді.", | |
| }, | |
| "note_prefix": "Ескертпе:", | |
| }, | |
| } | |
| def build_hypothesis_ledger(cls, constructs): | |
| """Freeze one result-independent AMJ hypothesis identity contract. | |
| The system-generated quantitative design is observational and does not | |
| justify result-driven directional hypotheses. The three numbered | |
| hypotheses therefore freeze only the construct pairing/association | |
| identity. The indirect association is an analysis target, not H4. | |
| """ | |
| values = [re.sub(r"\s+", " ", str(item or "")).strip() for item in (constructs or [])] | |
| if len(values) != 3 or any(not item for item in values): | |
| return {} | |
| predictor, mediator, outcome = values | |
| items = [ | |
| {"hypothesis_id": "H1", "predictor": predictor, "outcome": outcome, "relation": "association"}, | |
| {"hypothesis_id": "H2", "predictor": predictor, "outcome": mediator, "relation": "association"}, | |
| {"hypothesis_id": "H3", "predictor": mediator, "outcome": outcome, "relation": "association"}, | |
| ] | |
| return { | |
| "version": cls.HYPOTHESIS_LEDGER_VERSION, | |
| "items": items, | |
| "indirect_analysis": { | |
| "predictor": predictor, | |
| "mediator": mediator, | |
| "outcome": outcome, | |
| "numbered_hypothesis": False, | |
| "relation": "estimated_indirect_association", | |
| }, | |
| } | |
| 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 | |
| ): | |
| text = re.sub(re.escape(source), str(target), 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): | |
| cleaned = [] | |
| for value in values or []: | |
| name = re.sub(r"\s+", " ", str(value or "")).strip(" .,:;-_\n\t") | |
| if not name or len(name) > 80 or name.lower() in {item.lower() for item in cleaned}: | |
| continue | |
| cleaned.append(name) | |
| return cleaned[:3] | |
| def _derive_system_design(self, agent, topic="", keywords=""): | |
| """Freeze a parsimonious topic-aligned design without inventing named sites.""" | |
| fallback = { | |
| "constructs": [ | |
| "Focal Predictor", | |
| "Focal Mediator", | |
| "Focal Outcome", | |
| ], | |
| "research_context": "respondents with experience relevant to the focal management phenomenon", | |
| } | |
| if agent is None: | |
| return fallback | |
| prompt = f"""Design the minimum coherent quantitative management-study skeleton for the topic below. | |
| Return ONLY valid JSON: | |
| {{"constructs":["predictor construct","mediating construct","outcome construct"],"research_context":"one generic respondent context"}} | |
| Rules: | |
| 1. Exactly three distinct constructs in predictor -> mediator -> outcome order. | |
| 2. Use concise construct labels that can appear throughout a journal article. | |
| 3. The context must be generic and directly relevant to the topic. Do NOT name or invent a country, university, firm, industry sample frame, software package, time period, or institution. | |
| 4. This is a cross-sectional single-wave survey design, so do not imply experimental or causal identification. | |
| 5. Do not create any number; sample size and scale are frozen elsewhere. | |
| Topic: {topic} | |
| Keywords: {keywords}""" | |
| try: | |
| raw = agent.call_deepseek( | |
| "You are a rigorous management research-design editor.", | |
| prompt, | |
| max_tokens=500, | |
| temp=0.0, | |
| model_role="fast", | |
| thinking=False, | |
| allow_fallback=True, | |
| ) | |
| parsed = agent._extract_json_from_text(raw) | |
| if not isinstance(parsed, dict): | |
| return fallback | |
| constructs = self._safe_constructs(parsed.get("constructs")) | |
| context = re.sub(r"\s+", " ", str(parsed.get("research_context") or "")).strip() | |
| if len(constructs) != 3: | |
| return fallback | |
| if not context or len(context) > 240: | |
| context = fallback["research_context"] | |
| # Named-location/site details are deliberately not accepted from this | |
| # design step. The manuscript may remain generic when no user evidence | |
| # supplies a real setting. | |
| if re.search(r"\b(university|company|corporation|inc\.?|ltd\.?|country|china|usa|united states|europe|asia)\b", context, re.I): | |
| context = fallback["research_context"] | |
| return {"constructs": constructs, "research_context": context} | |
| except Exception as exc: | |
| logging.warning("Quantitative study-design freeze fell back safely: %s", exc) | |
| return fallback | |
| def generate_empirical_data_robust( | |
| self, | |
| construct_names=None, | |
| n=300, | |
| scale_points=7, | |
| ): | |
| if not construct_names or len(construct_names) < 2: | |
| construct_names = [ | |
| "Focal Predictor", | |
| "Focal Mediator", | |
| "Focal Outcome", | |
| ] | |
| n_constructs = len(construct_names) | |
| corr = self.generate_psd_corr_matrix(n_constructs, max_corr=0.55) | |
| latent_scores = np.random.multivariate_normal( | |
| mean=[0] * n_constructs, | |
| cov=corr, | |
| size=n, | |
| ) | |
| data = {} | |
| true_loadings = {} | |
| constructs_info = [] | |
| target_mean = scale_points / 2.0 + 0.5 | |
| target_std = scale_points / 5.0 | |
| for index, construct_name in enumerate(construct_names): | |
| true_score = latent_scores[:, index] | |
| n_items = np.random.randint(3, 6) | |
| loadings = np.random.uniform(0.72, 0.88, n_items) | |
| true_loadings[construct_name] = loadings | |
| item_names = [] | |
| safe_prefix = re.sub(r"[^A-Za-z0-9]+", "_", construct_name).strip("_") or f"Construct_{index + 1}" | |
| for item_index, loading in enumerate(loadings): | |
| error = np.random.normal( | |
| 0, | |
| np.sqrt(1 - loading**2), | |
| n, | |
| ) | |
| item_score = loading * true_score + error | |
| raw_scaled = item_score * target_std + target_mean | |
| item_int = np.clip( | |
| np.round(raw_scaled), | |
| 1, | |
| scale_points, | |
| ).astype(int) | |
| column_name = f"{safe_prefix}_{item_index + 1}" | |
| data[column_name] = item_int | |
| item_names.append(column_name) | |
| constructs_info.append( | |
| {"name": construct_name, "items": item_names} | |
| ) | |
| df = pd.DataFrame(data) | |
| validity_df = self.calculate_validity_metrics( | |
| df, | |
| constructs_info, | |
| true_loadings, | |
| ) | |
| mean_scores = pd.DataFrame( | |
| { | |
| construct["name"]: df[construct["items"]].mean(axis=1) | |
| for construct in constructs_info | |
| } | |
| ) | |
| corr_matrix = mean_scores.corr() | |
| fl_matrix = corr_matrix.astype(object) | |
| for _, row in validity_df.iterrows(): | |
| construct_name = row["Construct"] | |
| sqrt_ave = np.sqrt(row["AVE"]) | |
| fl_matrix.loc[construct_name, construct_name] = ( | |
| f"({sqrt_ave:.3f})" | |
| ) | |
| for row_index in range(len(fl_matrix.columns)): | |
| for column_index in range(row_index + 1, len(fl_matrix.columns)): | |
| fl_matrix.iloc[row_index, column_index] = "" | |
| for column_index in range(row_index): | |
| value = fl_matrix.iloc[row_index, column_index] | |
| if isinstance(value, (int, float, np.integer, np.floating)): | |
| fl_matrix.iloc[row_index, column_index] = f"{float(value):.3f}" | |
| desc_stats = mean_scores.mean().to_frame("Mean").join( | |
| mean_scores.std().to_frame("SD") | |
| ) | |
| return df, validity_df, desc_stats.join(fl_matrix), mean_scores, corr_matrix | |
| 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 _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 _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 {} | |
| results = dict(meta.get("association_results") or {}) if isinstance(meta.get("association_results"), dict) else {} | |
| return { | |
| "version": cls.QUANTITATIVE_EVIDENCE_LEDGER_VERSION, | |
| "source_type": source_type, | |
| "strict": bool(strict), | |
| "sample_size": int(sample_size) if isinstance(sample_size, (int, np.integer)) or str(sample_size or "").isdigit() else None, | |
| "scale_points": int(scale_points) if isinstance(scale_points, (int, np.integer)) or str(scale_points or "").isdigit() else None, | |
| "study_design": str(meta.get("study_design") or "").strip(), | |
| "constructs": {k: str(v or "").strip() for k, v in constructs.items() if str(v or "").strip()}, | |
| "association_results": results, | |
| "uploaded_metric_anchors": uploaded_metric_anchors, | |
| "causal_identification": "none" if strict else "evidence_owned", | |
| "allowed_methods": [ | |
| "reliability", "composite reliability", "average variance extracted", | |
| "Fornell-Larcker", "standardized association", "bootstrap", | |
| "Spearman", "VIF", "descriptive statistics", | |
| ] if strict else [], | |
| "forbidden_unowned_methods": [ | |
| "confirmatory factor analysis", "CFA", "structural equation modeling", "SEM", | |
| "PLS-SEM", "RMSEA", "CFI", "TLI", "SRMR", "Harman", | |
| "common method bias", "endogeneity", "instrumental variable", | |
| "fixed effects", "random effects", "difference-in-differences", | |
| ] if strict else [], | |
| } | |
| def classify_quantitative_method_threat_claim(cls, sentence, term): | |
| """Classify a forbidden quantitative term by *claim role*, not token presence. | |
| R6-R10 separates an actually performed/claimed unowned analysis from a | |
| limitation, explicit non-use statement, threat discussion, or incidental | |
| methodological mention. Only ``PERFORMED_UNOWNED`` is blocking. | |
| """ | |
| text = re.sub(r"\s+", " ", str(sentence or "")).strip() | |
| folded = text.casefold() | |
| label = re.sub(r"\s+", " ", str(term or "")).strip() | |
| term_folded = label.casefold() | |
| if not text or not term_folded: | |
| return {"term": label, "category": "UNKNOWN", "state": "ABSENT"} | |
| match = re.search(rf"(?i)(?<![A-Za-z]){re.escape(label)}(?![A-Za-z])", text) | |
| if not match: | |
| return {"term": label, "category": "UNKNOWN", "state": "ABSENT"} | |
| threat_terms = {"common method bias", "endogeneity"} | |
| fit_terms = {"rmsea", "cfi", "tli", "srmr"} | |
| category = ( | |
| "THREAT" if term_folded in threat_terms else | |
| "FIT_DIAGNOSTIC" if term_folded in fit_terms else | |
| "ANALYTIC_METHOD" | |
| ) | |
| before = folded[max(0, match.start() - 100):match.start()] | |
| after = folded[match.end():min(len(folded), match.end() + 120)] | |
| local = folded[max(0, match.start() - 120):min(len(folded), match.end() + 140)] | |
| scalar_negation = bool(re.search( | |
| r"\bnot\s+(?:only|merely|just|simply)\b", local | |
| )) | |
| explicit_nonuse = (not scalar_negation) and bool( | |
| re.search( | |
| r"\b(?:no|not|never|without|did\s+not|does\s+not|do\s+not|was\s+not|were\s+not|is\s+not|are\s+not|cannot|can\s+not)\b[^.;:]{0,55}$", | |
| before, | |
| ) | |
| or re.search( | |
| r"^.{0,45}\b(?:was\s+not|were\s+not|is\s+not|are\s+not|did\s+not|does\s+not|do\s+not|cannot|can\s+not|not\s+(?:used|conducted|performed|estimated|tested|assessed|evaluated|reported|implemented|applied))\b", | |
| after, | |
| ) | |
| or re.search( | |
| r"\b(?:does\s+not|do\s+not|did\s+not|cannot|can\s+not|is\s+not|are\s+not|was\s+not|were\s+not)\s+(?:address|assess|test|establish|identify|rule\s+out|correct|control|substitute|demonstrate|show|prove|support)\b", | |
| local, | |
| ) | |
| ) | |
| if explicit_nonuse: | |
| return {"term": label, "category": category, "state": "NOT_PERFORMED_OR_LIMITATION"} | |
| performed = bool( | |
| re.search( | |
| r"\b(?:we|the\s+study|this\s+study|the\s+analysis|our\s+analysis|the\s+model|the\s+authors?)\b[^.;:]{0,70}\b(?:use|uses|used|employ|employs|employed|conduct|conducts|conducted|perform|performs|performed|estimate|estimates|estimated|test|tests|tested|assess|assesses|assessed|evaluate|evaluates|evaluated|implement|implements|implemented|apply|applies|applied|ran|run|runs|fit|fits|fitted|model|models|modeled|modelled|control|controls|controlled|correct|corrects|corrected|address|addresses|addressed)\b[^.;:]{0,70}$", | |
| before, | |
| ) | |
| or re.search( | |
| r"^.{0,65}\b(?:was|were|is|are)\s+(?:used|employed|conducted|performed|estimated|tested|assessed|evaluated|implemented|applied|fit|fitted|modeled|modelled|controlled|corrected|addressed|reported)\b", | |
| after, | |
| ) | |
| or re.search( | |
| r"^.{0,65}\b(?:estimated|tested|assessed|evaluated|showed|indicated|yielded|produced|confirmed|supported)\b", | |
| after, | |
| ) | |
| ) | |
| if category == "FIT_DIAGNOSTIC" and re.search( | |
| rf"(?i)(?:{re.escape(label)}\s*(?:=|:|was|were)\s*[-+]?\d)|(?:reported|fit\s+index|fit\s+indices)[^.;:]{{0,45}}{re.escape(label)}", | |
| text, | |
| ): | |
| performed = True | |
| if category == "THREAT": | |
| # Threats may be discussed as limitations without implying a test. | |
| if re.search( | |
| rf"(?i)(?:assess|test|evaluate|examine|address|correct|control)(?:ed|ing|s)?[^.;:]{{0,45}}{re.escape(label)}|{re.escape(label)}[^.;:]{{0,45}}(?:was|were)\s+(?:assessed|tested|evaluated|examined|addressed|corrected|controlled)", | |
| text, | |
| ): | |
| performed = True | |
| elif not performed: | |
| return {"term": label, "category": category, "state": "LIMITATION_OR_THREAT"} | |
| return { | |
| "term": label, | |
| "category": category, | |
| "state": "PERFORMED_UNOWNED" if performed else "MENTION_ONLY", | |
| } | |
| 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 inspect_quantitative_study_fact_contract(cls, text, metadata, section_name=""): | |
| """Pure AMJ quantitative Evidence Ledger inspection; never rewrites prose.""" | |
| value = str(text or "") | |
| meta = dict(metadata or {}) | |
| ledger = dict(meta.get("quantitative_evidence_ledger") or {}) | |
| if not ledger: | |
| ledger = cls.build_quantitative_evidence_ledger(meta) | |
| sample_size = ledger.get("sample_size") | |
| scale_points = ledger.get("scale_points") | |
| strict = bool(ledger.get("strict")) | |
| constructs = dict(ledger.get("constructs") or {}) | |
| results = dict(ledger.get("association_results") or {}) | |
| uploaded = dict(ledger.get("uploaded_metric_anchors") or {}) | |
| known_constructs = [str(v) for v in constructs.values() if str(v)] | |
| construct_folded = [c.casefold() for c in known_constructs] | |
| forbidden = [str(x) for x in ledger.get("forbidden_unowned_methods") or []] | |
| role_patterns = { | |
| "predictor": r"(?:predictor|independent\s+variable|predictor\s+construct)", | |
| "mediator": r"(?:mediator|mediating\s+variable|mediating\s+construct)", | |
| "outcome": r"(?:outcome|dependent\s+variable|outcome\s+construct)", | |
| } | |
| violations = [] | |
| def add_metric_violations(sentence, pattern, expected, *, dtype="float", label="metric"): | |
| if expected is None: | |
| return | |
| regex = re.compile(pattern, re.I) | |
| for match in regex.finditer(sentence): | |
| observed = match.group(2) | |
| try: | |
| same = int(float(observed)) == int(expected) if dtype == "int" else abs(float(observed) - float(expected)) < 0.0006 | |
| except Exception: | |
| same = False | |
| if not same: | |
| expected_text = str(int(expected)) if dtype == "int" else f"{float(expected):.3f}".rstrip("0").rstrip(".") | |
| violations.append({ | |
| "type":"quantitative_metric_mismatch", "metric":label, | |
| "observed":observed, "expected":expected_text, | |
| "preview":str(sentence).strip()[:180], | |
| }) | |
| for _start, _end, sentence in cls._sentence_spans(value): | |
| ownership = cls._quantitative_sentence_ownership(sentence, section_name, known_constructs) | |
| method_scope_unknown = ( | |
| strict and ownership == "unknown" | |
| and bool(re.search(r"(?i)method|measure", str(section_name or ""))) | |
| and cls._first_performed_unowned_quantitative_term(sentence, forbidden) is not None | |
| ) | |
| if ownership not in {"study", "result"} and not method_scope_unknown: | |
| continue | |
| if sample_size: | |
| for match in re.finditer(r"(?i)(\b(?:sample(?:\s+size)?(?:\s+of)?|N)\s*(?:=|:)?\s*)(\d{2,6})\b", sentence): | |
| observed = int(match.group(2)) | |
| if observed != int(sample_size): | |
| violations.append({"type":"sample_size_mismatch","observed":observed,"expected":int(sample_size),"preview":sentence.strip()[:180]}) | |
| for match in re.finditer(r"(?i)\b(\d{2,6})\s+(respondents|participants|observations|cases)\b", sentence): | |
| observed = int(match.group(1)) | |
| if observed != int(sample_size): | |
| violations.append({"type":"sample_count_mismatch","observed":observed,"expected":int(sample_size),"preview":sentence.strip()[:180]}) | |
| if scale_points: | |
| for match in re.finditer(r"(?i)\b([2-9]|1[0-1])[- ]point\s+(Likert(?:-type)?\s+scale)", sentence): | |
| observed = int(match.group(1)) | |
| if observed != int(scale_points): | |
| violations.append({"type":"scale_points_mismatch","observed":observed,"expected":int(scale_points),"preview":sentence.strip()[:180]}) | |
| for role, role_pattern in role_patterns.items(): | |
| expected = str(constructs.get(role) or "") | |
| if not expected: | |
| continue | |
| for candidate in known_constructs: | |
| if candidate.casefold() == expected.casefold(): | |
| continue | |
| pattern = re.compile(rf"(?i)({role_pattern}\s*(?:was|is|:|=)?\s*){re.escape(candidate)}") | |
| if pattern.search(sentence): | |
| violations.append({"type":"construct_role_mismatch","role":role,"observed":candidate,"expected":expected,"preview":sentence.strip()[:180]}) | |
| if strict: | |
| methodological_bias_scope = cls._is_methodological_bias_mechanism_sentence( | |
| sentence, known_constructs | |
| ) | |
| semantic_frame = cls.build_quantitative_semantic_claim_frame( | |
| sentence, meta, section_name | |
| ) | |
| causal_state = str(semantic_frame.get("validation_state") or "NON_CAUSAL") | |
| if causal_state == "AFFIRMATIVE_CAUSAL": | |
| violations.append({ | |
| "type":"causal_language_violation", | |
| "preview":sentence.strip()[:180], | |
| "reason":"The sentence makes an affirmative causal claim that exceeds the cross-sectional observational evidence contract.", | |
| "repair_instruction":( | |
| "Preserve the study meaning but rewrite as association/covariation only. " | |
| "Explicit causal-limit language is allowed, including forms such as " | |
| "'restricted to associations, not causal effects' or 'cannot establish causality'." | |
| ), | |
| }) | |
| if (any(c in sentence.casefold() for c in construct_folded) if construct_folded else False) and re.search(r"(?i)\bp\s*(?:=|<|>|≤|≥)\s*(?:0?\.\d+|\d+(?:\.\d+)?)", sentence): | |
| violations.append({"type":"unfrozen_p_value","preview":sentence.strip()[:180]}) | |
| if ownership == "study" or method_scope_unknown: | |
| method_frame = cls._first_performed_unowned_quantitative_term(sentence, forbidden) | |
| if method_frame: | |
| violations.append({ | |
| "type":"unsupported_quantitative_method", | |
| "method":method_frame.get("term"), | |
| "claim_state":method_frame.get("state"), | |
| "method_category":method_frame.get("category"), | |
| "preview":sentence.strip()[:180], | |
| }) | |
| vif_expected = results.get("predictor_mediator_vif") if strict else uploaded.get("VIF") | |
| boot_expected = results.get("bootstrap_iterations") if strict else uploaded.get("bootstrap_iterations") | |
| add_metric_violations(sentence, r"(\bVIF\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", vif_expected, label="VIF") | |
| add_metric_violations(sentence, r"(\bbootstrap\s+(?:resamples?|iterations?)\s*(?:=|:)?\s*)(\d{2,6})", boot_expected, dtype="int", label="bootstrap_iterations") | |
| if strict and results: | |
| for role, key in (("mediator", "r2_mediator"), ("outcome", "r2_outcome")): | |
| construct = str(constructs.get(role) or "") | |
| if construct and results.get(key) is not None: | |
| add_metric_violations(sentence, rf"(R(?:²|\^2)\s*(?:for|of)?\s*{re.escape(construct)}\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", results.get(key), label=key) | |
| lower = sentence.casefold() | |
| selected = None | |
| specs = [ | |
| ("indirect", results.get("indirect"), [constructs.get("predictor"), constructs.get("mediator"), constructs.get("outcome")]), | |
| ("direct", results.get("direct"), [constructs.get("predictor"), constructs.get("outcome")]), | |
| ("a", results.get("a"), [constructs.get("predictor"), constructs.get("mediator")]), | |
| ("b", results.get("b"), [constructs.get("mediator"), constructs.get("outcome")]), | |
| ] | |
| if "indirect" in lower and all(str(x or "").casefold() in lower for x in specs[0][2] if x): | |
| selected = specs[0] | |
| elif f"controlling {str(constructs.get('mediator') or '').casefold()}" in lower and str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower: | |
| selected = specs[1] | |
| elif f"controlling {str(constructs.get('predictor') or '').casefold()}" in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower: | |
| selected = specs[3] | |
| elif str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() not in lower: | |
| selected = specs[2] | |
| elif str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower and str(constructs.get("predictor") or "").casefold() not in lower: | |
| selected = specs[3] | |
| if selected and selected[1] is not None: | |
| add_metric_violations(sentence, r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", selected[1], label=selected[0]) | |
| elif not strict: | |
| for label, expected in dict(uploaded.get("r2_by_label") or {}).items(): | |
| if label and label in sentence.casefold(): | |
| add_metric_violations(sentence, rf"(R(?:²|\^2)\s*(?:for|of)\s*{re.escape(label)}\s*(?:=|:)\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_R2:{label}") | |
| for path, expected in dict(uploaded.get("beta_by_path") or {}).items(): | |
| left, _, right = path.partition(" -> ") | |
| if left and right and left in sentence.casefold() and right in sentence.casefold(): | |
| add_metric_violations(sentence, r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_beta:{path}") | |
| return violations | |
| def repair_quantitative_study_fact_contract(cls, text, metadata, section_name=""): | |
| """Repair deterministic AMJ drift only in study/result-owned prose.""" | |
| value = str(text or "") | |
| meta = dict(metadata or {}) | |
| ledger = dict(meta.get("quantitative_evidence_ledger") or {}) | |
| if not ledger: | |
| ledger = cls.build_quantitative_evidence_ledger(meta) | |
| diagnostics = [] | |
| sample_size = ledger.get("sample_size") | |
| scale_points = ledger.get("scale_points") | |
| strict = bool(ledger.get("strict")) | |
| constructs = dict(ledger.get("constructs") or {}) | |
| results = dict(ledger.get("association_results") or {}) | |
| uploaded = dict(ledger.get("uploaded_metric_anchors") or {}) | |
| known_constructs = [str(v) for v in constructs.values() if str(v)] | |
| construct_folded = [c.casefold() for c in known_constructs] | |
| forbidden = [str(x) for x in ledger.get("forbidden_unowned_methods") or []] | |
| role_patterns = { | |
| "predictor": r"(?:predictor|independent\s+variable|predictor\s+construct)", | |
| "mediator": r"(?:mediator|mediating\s+variable|mediating\s+construct)", | |
| "outcome": r"(?:outcome|dependent\s+variable|outcome\s+construct)", | |
| } | |
| causal_replacements = ( | |
| (r"(?i)\bhad\s+a\s+causal\s+effect\s+on\b", "was associated with"), | |
| (r"(?i)\bexerted\s+a\s+causal\s+effect\s+on\b", "was associated with"), | |
| (r"(?i)\bcausal\s+effect\s+on\b", "association with"), | |
| (r"(?i)\ban\s+causal\s+effect\b", "an association"), | |
| (r"(?i)\ba\s+causal\s+effect\b", "an association"), | |
| (r"(?i)\bthe\s+causal\s+effect\b", "the association"), | |
| (r"(?i)\bcausal\s+effects\b", "associations"), | |
| (r"(?i)\bcausal\s+effect\b", "association"), | |
| (r"(?i)\bcausal\s+mediation\b", "indirect association"), | |
| ) | |
| replacements = [] | |
| for start, end, sentence in cls._sentence_spans(value): | |
| ownership = cls._quantitative_sentence_ownership(sentence, section_name, known_constructs) | |
| method_scope_unknown = ( | |
| strict and ownership == "unknown" | |
| and bool(re.search(r"(?i)method|measure", str(section_name or ""))) | |
| and cls._first_performed_unowned_quantitative_term(sentence, forbidden) is not None | |
| ) | |
| if ownership not in {"study", "result"} and not method_scope_unknown: | |
| continue | |
| revised = sentence | |
| if sample_size: | |
| sample_pattern = re.compile(r"(?i)(\b(?:sample(?:\s+size)?(?:\s+of)?|N)\s*(?:=|:)?\s*)(\d{2,6})\b") | |
| def sample_repl(match): | |
| observed = int(match.group(2)) | |
| if observed == int(sample_size): | |
| return match.group(0) | |
| diagnostics.append({"type":"sample_size_repaired","observed":observed,"expected":int(sample_size)}) | |
| return match.group(1) + str(int(sample_size)) | |
| revised = sample_pattern.sub(sample_repl, revised) | |
| count_pattern = re.compile(r"(?i)\b(\d{2,6})\s+(respondents|participants|observations|cases)\b") | |
| def count_repl(match): | |
| observed = int(match.group(1)) | |
| if observed == int(sample_size): | |
| return match.group(0) | |
| diagnostics.append({"type":"sample_count_repaired","observed":observed,"expected":int(sample_size)}) | |
| return f"{int(sample_size)} {match.group(2)}" | |
| revised = count_pattern.sub(count_repl, revised) | |
| if scale_points: | |
| scale_pattern = re.compile(r"(?i)\b([2-9]|1[0-1])[- ]point\s+(Likert(?:-type)?\s+scale)") | |
| def scale_repl(match): | |
| observed = int(match.group(1)) | |
| if observed == int(scale_points): | |
| return match.group(0) | |
| diagnostics.append({"type":"scale_points_repaired","observed":observed,"expected":int(scale_points)}) | |
| return f"{int(scale_points)}-point {match.group(2)}" | |
| revised = scale_pattern.sub(scale_repl, revised) | |
| for role, role_pattern in role_patterns.items(): | |
| expected = str(constructs.get(role) or "") | |
| if not expected: | |
| continue | |
| for candidate in known_constructs: | |
| if candidate.casefold() == expected.casefold(): | |
| continue | |
| pattern = re.compile(rf"(?i)({role_pattern}\s*(?:was|is|:|=)?\s*){re.escape(candidate)}") | |
| revised, count = pattern.subn(lambda m: m.group(1) + expected, revised) | |
| if count: | |
| diagnostics.append({"type":"construct_role_repaired","role":role,"observed":candidate,"expected":expected,"count":count}) | |
| if strict: | |
| causal_state = cls.classify_quantitative_causal_sentence(revised) | |
| if causal_state == "AFFIRMATIVE_CAUSAL": | |
| for pattern, replacement in causal_replacements: | |
| regex = re.compile(pattern) | |
| revised, repaired_count = regex.subn(replacement, revised) | |
| if repaired_count: | |
| diagnostics.append({ | |
| "type":"causal_language_repaired", | |
| "replacement":replacement, | |
| "count":repaired_count, | |
| }) | |
| if any(c in revised.casefold() for c in construct_folded) if construct_folded else False: | |
| revised, p_count = re.subn(r"(?i)\s*[,;]?\s*p\s*(?:=|<|>|≤|≥)\s*(?:0?\.\d+|\d+(?:\.\d+)?)", "", revised) | |
| if p_count: | |
| diagnostics.append({"type":"unfrozen_p_value_removed","count":p_count}) | |
| if ownership == "study" or method_scope_unknown: | |
| method_frame = cls._first_performed_unowned_quantitative_term(revised, forbidden) | |
| if method_frame: | |
| diagnostics.append({ | |
| "type":"unsupported_quantitative_method_removed", | |
| "method":method_frame.get("term"), | |
| "claim_state":method_frame.get("state"), | |
| "method_category":method_frame.get("category"), | |
| "preview":sentence.strip()[:180], | |
| }) | |
| revised = "" | |
| def replace_local_metric(pattern, expected, *, dtype="float", label="metric"): | |
| nonlocal revised | |
| if expected is None: | |
| return | |
| expected_text = str(int(expected)) if dtype == "int" else f"{float(expected):.3f}".rstrip("0").rstrip(".") | |
| regex = re.compile(pattern, re.I) | |
| def repl(match): | |
| observed = match.group(2) | |
| try: | |
| same = int(float(observed)) == int(expected) if dtype == "int" else abs(float(observed) - float(expected)) < 0.0006 | |
| except Exception: | |
| same = False | |
| if same: | |
| return match.group(0) | |
| diagnostics.append({"type":"quantitative_metric_repaired","metric":label,"observed":observed,"expected":expected_text}) | |
| return match.group(1) + expected_text | |
| revised = regex.sub(repl, revised) | |
| vif_expected = results.get("predictor_mediator_vif") if strict else uploaded.get("VIF") | |
| boot_expected = results.get("bootstrap_iterations") if strict else uploaded.get("bootstrap_iterations") | |
| replace_local_metric(r"(\bVIF\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", vif_expected, label="VIF") | |
| replace_local_metric(r"(\bbootstrap\s+(?:resamples?|iterations?)\s*(?:=|:)?\s*)(\d{2,6})", boot_expected, dtype="int", label="bootstrap_iterations") | |
| if strict and results: | |
| for role, key in (("mediator", "r2_mediator"), ("outcome", "r2_outcome")): | |
| construct = str(constructs.get(role) or "") | |
| if construct and results.get(key) is not None: | |
| metric_pattern = rf"(R(?:²|\^2)\s*(?:for|of)?\s*{re.escape(construct)}\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)" | |
| replace_local_metric(metric_pattern, results.get(key), label=key) | |
| lower = revised.casefold() | |
| selected = None | |
| specs = [ | |
| ("indirect", results.get("indirect"), [constructs.get("predictor"), constructs.get("mediator"), constructs.get("outcome")]), | |
| ("direct", results.get("direct"), [constructs.get("predictor"), constructs.get("outcome")]), | |
| ("a", results.get("a"), [constructs.get("predictor"), constructs.get("mediator")]), | |
| ("b", results.get("b"), [constructs.get("mediator"), constructs.get("outcome")]), | |
| ] | |
| if "indirect" in lower and all(str(x or "").casefold() in lower for x in specs[0][2] if x): | |
| selected = specs[0] | |
| elif f"controlling {str(constructs.get('mediator') or '').casefold()}" in lower and str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower: | |
| selected = specs[1] | |
| elif f"controlling {str(constructs.get('predictor') or '').casefold()}" in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower: | |
| selected = specs[3] | |
| elif str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() not in lower: | |
| selected = specs[2] | |
| elif str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower and str(constructs.get("predictor") or "").casefold() not in lower: | |
| selected = specs[3] | |
| if selected and selected[1] is not None: | |
| replace_local_metric(r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", selected[1], label=selected[0]) | |
| elif not strict: | |
| r2_map = dict(uploaded.get("r2_by_label") or {}) | |
| for label, expected in r2_map.items(): | |
| if label and label in revised.casefold(): | |
| replace_local_metric(rf"(R(?:²|\^2)\s*(?:for|of)\s*{re.escape(label)}\s*(?:=|:)\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_R2:{label}") | |
| beta_map = dict(uploaded.get("beta_by_path") or {}) | |
| for path, expected in beta_map.items(): | |
| left, _, right = path.partition(" -> ") | |
| if left and right and left in revised.casefold() and right in revised.casefold(): | |
| replace_local_metric(r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_beta:{path}") | |
| if revised != sentence: | |
| replacements.append((start, end, revised)) | |
| for start, end, replacement in sorted(replacements, reverse=True): | |
| value = value[:start] + replacement + value[end:] | |
| value = re.sub(r"[ \t]+\n", "\n", value) | |
| value = re.sub(r"\n{3,}", "\n\n", value) | |
| value = re.sub(r" {2,}", " ", value) | |
| return value.strip(), diagnostics | |
| def validate_quantitative_study_fact_contract(cls, text, metadata, section_name=""): | |
| violations = cls.inspect_quantitative_study_fact_contract(text, metadata, section_name=section_name) | |
| if violations: | |
| raise RuntimeError(f"quantitative_study_fact_contract_invalid:{section_name}:{violations[:3]}") | |
| return True | |
| def resolve_empirical_data( | |
| self, | |
| uploaded_files=None, | |
| language="zh", | |
| *, | |
| topic="", | |
| keywords="", | |
| agent=None, | |
| ): | |
| if uploaded_files: | |
| user_md = extract_data_from_files(uploaded_files) | |
| if user_md: | |
| evidence_sections = [ | |
| heading.strip() | |
| for heading in re.findall( | |
| r"(?m)^#{1,3}\s+(.+?)\s*$", | |
| user_md, | |
| ) | |
| if heading.strip() | |
| ] | |
| metadata = { | |
| "mode": "quantitative", | |
| "source_type": "user_provided", | |
| "disclosure_required": False, | |
| "evidence_sections": evidence_sections[:20] | |
| or ["User-provided quantitative evidence"], | |
| "evidence_boundary": ( | |
| "Only values explicitly present in the uploaded evidence are frozen. " | |
| "Do not infer absent setting, sample, measures, software, or results." | |
| ), | |
| } | |
| metadata["quantitative_evidence_ledger"] = self.build_quantitative_evidence_ledger(metadata, user_md) | |
| return None, user_md, tr(language, "actual_stats_parsed"), metadata | |
| design = self._derive_system_design(agent, topic, keywords) | |
| construct_names = design["constructs"] | |
| df, validity_df, fl_matrix_df, mean_scores, corr_matrix = self.generate_empirical_data_robust( | |
| construct_names=construct_names | |
| ) | |
| path_summary, path_table, spearman = self._path_evidence(mean_scores, construct_names) | |
| context = design["research_context"] | |
| generated_prompt = f"""### Frozen Quantitative Study Design | |
| | Field | Frozen value | | |
| | :--- | :--- | | |
| | Design | Cross-sectional, single-wave questionnaire study | | |
| | Respondent context | {context} | | |
| | Sample size | {len(df)} | | |
| | Response scale | 7-point Likert-type scale | | |
| | Predictor | {construct_names[0]} | | |
| | Mediator | {construct_names[1]} | | |
| | Outcome | {construct_names[2]} | | |
| | Causal identification | None; report associations and estimated indirect association, not causal effects | | |
| ### Table 1: Reliability and Convergent Validity | |
| {validity_df.to_markdown(index=False)} | |
| ### Table 2: Descriptive Statistics and Discriminant Validity (Fornell-Larcker Criterion) | |
| *Note: Diagonal elements in parentheses are the square roots of AVE.* | |
| {fl_matrix_df.to_markdown()} | |
| ### Table 3: Frozen Association Model Results | |
| {path_table.to_markdown(index=False)} | |
| ### Table 4: Rank-Order Robustness Correlations (Spearman) | |
| {spearman.round(3).to_markdown()} | |
| ### Frozen Bootstrap and Collinearity Checks | |
| - Bootstrap resamples: {path_summary['bootstrap_iterations']}. | |
| - Indirect association ({construct_names[0]} → {construct_names[1]} → {construct_names[2]}): {path_summary['indirect']}; 95% bootstrap CI [{path_summary['indirect_ci'][0]}, {path_summary['indirect_ci'][1]}]. | |
| - Residual direct association ({construct_names[0]} → {construct_names[2]} controlling {construct_names[1]}): {path_summary['direct']}; 95% bootstrap CI [{path_summary['direct_ci'][0]}, {path_summary['direct_ci'][1]}]. | |
| - Predictor/mediator VIF in the outcome model: {path_summary['predictor_mediator_vif']}. | |
| [STRICT EMPIRICAL INSTRUCTION] | |
| 1. ZERO HALLUCINATION: Every study-specific fact and number must come from this frozen packet. Never invent a country, named organization, sampling frame, response rate, demographic characteristic, measure source, software package, coefficient, p-value, fit index, robustness test, or date. | |
| 2. DESIGN DISCIPLINE: This is a cross-sectional observational survey. Use association language. The indirect estimate is not causal mediation identification. | |
| 3. MEASUREMENT DEFENSE: State reliability/convergent/discriminant validity only when the displayed values support it. | |
| 4. RESULTS DISCIPLINE: Results must report this study's frozen numbers without attaching external citations to the coefficients or sample facts. | |
| 5. ROBUSTNESS: The Spearman matrix, bootstrap intervals, and VIF above are the only frozen robustness evidence. Do not add another robustness procedure. | |
| 6. CONSISTENCY: The predictor, mediator, outcome, sample size, context, scale, and all numerical results above are the single source of quantitative truth throughout the manuscript. | |
| """ | |
| metadata = { | |
| "mode": "quantitative", | |
| "source_type": "system_generated", | |
| "disclosure_required": True, | |
| "sample_size": int(len(df)), | |
| "scale_points": 7, | |
| "study_design": "cross-sectional single-wave questionnaire", | |
| "research_context": context, | |
| "constructs": { | |
| "predictor": construct_names[0], | |
| "mediator": construct_names[1], | |
| "outcome": construct_names[2], | |
| }, | |
| "association_results": path_summary, | |
| "method": ( | |
| "Positive-semidefinite latent correlation generation, indicator loadings, " | |
| "Likert discretization, reliability/validity summaries, standardized association " | |
| "models, fixed-seed bootstrap intervals, and Spearman robustness correlations." | |
| ), | |
| "generation_parameters": { | |
| "constructs": construct_names, | |
| "sample_size": int(len(df)), | |
| "likert_scale_points": 7, | |
| "items_per_construct": "random integer from 3 to 5", | |
| "indicator_loading_range": [0.72, 0.88], | |
| "latent_correlation_control": "positive-semidefinite matrix with maximum target correlation 0.55", | |
| "item_generation": "loading × latent score + normally distributed measurement error", | |
| "discretization": "round to nearest integer and clamp to 1–7", | |
| "bootstrap_iterations": path_summary["bootstrap_iterations"], | |
| "bootstrap_seed": 20260809, | |
| "reported_summaries": [ | |
| "Cronbach alpha", | |
| "composite reliability", | |
| "average variance extracted", | |
| "descriptive statistics", | |
| "Fornell-Larcker matrix", | |
| "standardized association estimates", | |
| "R-squared", | |
| "bootstrap confidence intervals", | |
| "Spearman correlations", | |
| "VIF", | |
| ], | |
| }, | |
| "evidence_sections": [ | |
| "Frozen quantitative study design", | |
| "Reliability and convergent validity", | |
| "Descriptive statistics and discriminant validity", | |
| "Frozen association model results", | |
| "Rank-order robustness correlations", | |
| "Frozen bootstrap and collinearity checks", | |
| ], | |
| } | |
| metadata["semantic_claim_frame_version"] = self.SEMANTIC_CLAIM_FRAME_VERSION | |
| metadata["hypothesis_ledger"] = self.build_hypothesis_ledger(construct_names) | |
| metadata["quantitative_table_manifest"] = self.build_table_manifest( | |
| validity_df, fl_matrix_df, path_table, spearman | |
| ) | |
| metadata["quantitative_evidence_ledger"] = self.build_quantitative_evidence_ledger(metadata, generated_prompt) | |
| return ( | |
| df, | |
| generated_prompt, | |
| tr(language, "synthetic_stats_generated"), | |
| metadata, | |
| ) | |
| def get_quantitative_engine(): | |
| return QuantitativeEngine() | |