Spaces:
Running
Running
| # -*- coding: utf-8 -*- | |
| """Quantitative evidence preparation for SmartPaper's AMJ-oriented track.""" | |
| from __future__ import annotations | |
| import json | |
| import logging | |
| import re | |
| import numpy as np | |
| import pandas as pd | |
| from i18n import tr | |
| from utils.file_parser import extract_data_from_files | |
| class QuantitativeEngine: | |
| """Prepare one frozen quantitative evidence packet before manuscript writing.""" | |
| QUANTITATIVE_EVIDENCE_LEDGER_VERSION = "quantitative-evidence-ledger-v2" | |
| QUANTITATIVE_TABLE_MANIFEST_VERSION = "quantitative-table-manifest-v1" | |
| HYPOTHESIS_LEDGER_VERSION = "quantitative-hypothesis-ledger-v1" | |
| 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 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 _causal_phrase_is_explicitly_disclaimed(sentence, match_start, match_end): | |
| """Return True only when a causal phrase is explicitly scoped by a limitation. | |
| This is intentionally local to the matched phrase rather than sentence-wide so a | |
| sentence can safely say that causal identification is unavailable while still | |
| failing closed if a separate clause makes an affirmative causal claim. | |
| """ | |
| value = str(sentence or "") | |
| left_floor = max(0, int(match_start) - 180) | |
| left = value[left_floor:int(match_start)] | |
| # Do not let a disclaimer in an earlier independent/contrast clause shield a later | |
| # affirmative causal assertion. Keep the closest clause-like segment only. | |
| boundaries = list(re.finditer( | |
| r"(?i)(?:[;:,]|\bbut\b|\bhowever\b|\bwhereas\b|\byet\b|\bnevertheless\b)", | |
| left, | |
| )) | |
| if boundaries: | |
| left = left[boundaries[-1].end():] | |
| left_folded = re.sub(r"\s+", " ", left.casefold()).strip() | |
| right = value[int(match_end):min(len(value), int(match_end) + 140)] | |
| right_folded = re.sub(r"\s+", " ", right.casefold()).strip() | |
| pre_disclaimer_patterns = ( | |
| r"\b(?:cannot|can't|could not|couldn't|does not|doesn't|do not|don't|did not|didn't)\s+(?:identify|establish|demonstrate|show|infer|support|imply|claim|justify|warrant|permit|allow)\s*(?:(?:an?|the)\s*)?$", | |
| r"\b(?:is not|isn't|are not|aren't|was not|wasn't|were not|weren't)\s+(?:able\s+to\s+)?(?:identify|establish|demonstrate|show|infer|support|justify|warrant)\s*(?:(?:an?|the)\s*)?$", | |
| r"\b(?:unable|insufficient)\s+to\s+(?:identify|establish|demonstrate|show|infer|support|justify|warrant)\s*(?:(?:an?|the)\s*)?$", | |
| r"\b(?:inability|lack\s+of\s+(?:an?\s+)?ability)\s+to\s+(?:identify|establish|demonstrate|show|infer|support|justify|warrant)\s*(?:(?:an?|the)\s*)?$", | |
| r"\bwithout\s+(?:directly\s+)?(?:identifying|establishing|demonstrating|showing|inferring|supporting|justifying|warranting)(?:\s+\w+){0,8}\s*(?:(?:an?|the)\s*)?$", | |
| r"\b(?:not\s+designed|not\s+intended)\s+to\s+(?:identify|establish|demonstrate|show|infer|support|justify|warrant)\s*(?:(?:an?|the)\s*)?$", | |
| r"\b(?:should not|shouldn't|must not|cannot|can't|could not|couldn't)\s+be\s+interpreted\s+as\s*(?:(?:an?|the)\s*)?$", | |
| r"\b(?:fails?|failed)\s+to\s+(?:identify|establish|demonstrate|show|infer|support|justify|warrant)\s*(?:(?:an?|the)\s*)?$", | |
| r"\bno\s+(?:evidence|basis|support)\s+(?:for|of)\s*(?:(?:an?|the)\s*)?$", | |
| r"\bnot\s+(?:evidence|proof)\s+(?:for|of)\s*(?:(?:an?|the)\s*)?$", | |
| r"\bno\s*$", | |
| ) | |
| if any(re.search(pattern, left_folded) for pattern in pre_disclaimer_patterns): | |
| return True | |
| post_disclaimer_patterns = ( | |
| r"^(?:\s*(?:is|are|was|were|has|have|had))?\s*(?:not|never)\s+(?:identified|established|demonstrated|shown|inferred|supported|claimed|justified|warranted)\b", | |
| r"^\s*(?:cannot|can't|could not|couldn't|should not|shouldn't|must not|does not|doesn't|do not|don't)\s+(?:be\s+)?(?:identified|established|demonstrated|shown|inferred|supported|claimed|interpreted|justified|warranted)\b", | |
| r"^\s*(?:claim|claims|interpretation|inference)s?\s+(?:is|are)\s+(?:not|unsupported|unwarranted)\b", | |
| ) | |
| return any(re.search(pattern, right_folded) for pattern in post_disclaimer_patterns) | |
| def inspect_quantitative_study_fact_contract(cls, text, metadata, section_name=""): | |
| """Pure AMJ quantitative Evidence Ledger inspection; never rewrites prose.""" | |
| value = str(text or "") | |
| meta = dict(metadata or {}) | |
| ledger = dict(meta.get("quantitative_evidence_ledger") or {}) | |
| if not ledger: | |
| ledger = cls.build_quantitative_evidence_ledger(meta) | |
| sample_size = ledger.get("sample_size") | |
| scale_points = ledger.get("scale_points") | |
| strict = bool(ledger.get("strict")) | |
| constructs = dict(ledger.get("constructs") or {}) | |
| results = dict(ledger.get("association_results") or {}) | |
| uploaded = dict(ledger.get("uploaded_metric_anchors") or {}) | |
| known_constructs = [str(v) for v in constructs.values() if str(v)] | |
| construct_folded = [c.casefold() for c in known_constructs] | |
| forbidden = [str(x) for x in ledger.get("forbidden_unowned_methods") or []] | |
| role_patterns = { | |
| "predictor": r"(?:predictor|independent\s+variable|predictor\s+construct)", | |
| "mediator": r"(?:mediator|mediating\s+variable|mediating\s+construct)", | |
| "outcome": r"(?:outcome|dependent\s+variable|outcome\s+construct)", | |
| } | |
| causal_patterns = ( | |
| r"(?i)\bhad\s+a\s+causal\s+effect\s+on\b", | |
| r"(?i)\bexerted\s+a\s+causal\s+effect\s+on\b", | |
| r"(?i)\bcausal\s+effect\s+on\b", | |
| r"(?i)\b(?:an|a|the)\s+causal\s+effect\b", | |
| r"(?i)\bcausal\s+effects\b", | |
| r"(?i)\bcausal\s+effect\b", | |
| r"(?i)\bcausal\s+mediation\b", | |
| ) | |
| violations = [] | |
| def add_metric_violations(sentence, pattern, expected, *, dtype="float", label="metric"): | |
| if expected is None: | |
| return | |
| regex = re.compile(pattern, re.I) | |
| for match in regex.finditer(sentence): | |
| observed = match.group(2) | |
| try: | |
| same = int(float(observed)) == int(expected) if dtype == "int" else abs(float(observed) - float(expected)) < 0.0006 | |
| except Exception: | |
| same = False | |
| if not same: | |
| expected_text = str(int(expected)) if dtype == "int" else f"{float(expected):.3f}".rstrip("0").rstrip(".") | |
| violations.append({ | |
| "type":"quantitative_metric_mismatch", "metric":label, | |
| "observed":observed, "expected":expected_text, | |
| "preview":str(sentence).strip()[:180], | |
| }) | |
| for _start, _end, sentence in cls._sentence_spans(value): | |
| ownership = cls._quantitative_sentence_ownership(sentence, section_name, known_constructs) | |
| if ownership not in {"study", "result"}: | |
| continue | |
| if sample_size: | |
| for match in re.finditer(r"(?i)(\b(?:sample(?:\s+size)?(?:\s+of)?|N)\s*(?:=|:)?\s*)(\d{2,6})\b", sentence): | |
| observed = int(match.group(2)) | |
| if observed != int(sample_size): | |
| violations.append({"type":"sample_size_mismatch","observed":observed,"expected":int(sample_size),"preview":sentence.strip()[:180]}) | |
| for match in re.finditer(r"(?i)\b(\d{2,6})\s+(respondents|participants|observations|cases)\b", sentence): | |
| observed = int(match.group(1)) | |
| if observed != int(sample_size): | |
| violations.append({"type":"sample_count_mismatch","observed":observed,"expected":int(sample_size),"preview":sentence.strip()[:180]}) | |
| if scale_points: | |
| for match in re.finditer(r"(?i)\b([2-9]|1[0-1])[- ]point\s+(Likert(?:-type)?\s+scale)", sentence): | |
| observed = int(match.group(1)) | |
| if observed != int(scale_points): | |
| violations.append({"type":"scale_points_mismatch","observed":observed,"expected":int(scale_points),"preview":sentence.strip()[:180]}) | |
| for role, role_pattern in role_patterns.items(): | |
| expected = str(constructs.get(role) or "") | |
| if not expected: | |
| continue | |
| for candidate in known_constructs: | |
| if candidate.casefold() == expected.casefold(): | |
| continue | |
| pattern = re.compile(rf"(?i)({role_pattern}\s*(?:was|is|:|=)?\s*){re.escape(candidate)}") | |
| if pattern.search(sentence): | |
| violations.append({"type":"construct_role_mismatch","role":role,"observed":candidate,"expected":expected,"preview":sentence.strip()[:180]}) | |
| if strict: | |
| for pattern in causal_patterns: | |
| match = re.search(pattern, sentence) | |
| if match and not cls._causal_phrase_is_explicitly_disclaimed( | |
| sentence, match.start(), match.end() | |
| ): | |
| violations.append({"type":"causal_language_violation","preview":sentence.strip()[:180]}) | |
| break | |
| if (any(c in sentence.casefold() for c in construct_folded) if construct_folded else False) and re.search(r"(?i)\bp\s*(?:=|<|>|≤|≥)\s*(?:0?\.\d+|\d+(?:\.\d+)?)", sentence): | |
| violations.append({"type":"unfrozen_p_value","preview":sentence.strip()[:180]}) | |
| if ownership == "study": | |
| bad = next((term for term in forbidden if re.search(rf"(?i)(?<![A-Za-z]){re.escape(term)}(?![A-Za-z])", sentence)), "") | |
| if bad: | |
| violations.append({"type":"unsupported_quantitative_method","method":bad,"preview":sentence.strip()[:180]}) | |
| vif_expected = results.get("predictor_mediator_vif") if strict else uploaded.get("VIF") | |
| boot_expected = results.get("bootstrap_iterations") if strict else uploaded.get("bootstrap_iterations") | |
| add_metric_violations(sentence, r"(\bVIF\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", vif_expected, label="VIF") | |
| add_metric_violations(sentence, r"(\bbootstrap\s+(?:resamples?|iterations?)\s*(?:=|:)?\s*)(\d{2,6})", boot_expected, dtype="int", label="bootstrap_iterations") | |
| if strict and results: | |
| for role, key in (("mediator", "r2_mediator"), ("outcome", "r2_outcome")): | |
| construct = str(constructs.get(role) or "") | |
| if construct and results.get(key) is not None: | |
| add_metric_violations(sentence, rf"(R(?:²|\^2)\s*(?:for|of)?\s*{re.escape(construct)}\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", results.get(key), label=key) | |
| lower = sentence.casefold() | |
| selected = None | |
| specs = [ | |
| ("indirect", results.get("indirect"), [constructs.get("predictor"), constructs.get("mediator"), constructs.get("outcome")]), | |
| ("direct", results.get("direct"), [constructs.get("predictor"), constructs.get("outcome")]), | |
| ("a", results.get("a"), [constructs.get("predictor"), constructs.get("mediator")]), | |
| ("b", results.get("b"), [constructs.get("mediator"), constructs.get("outcome")]), | |
| ] | |
| if "indirect" in lower and all(str(x or "").casefold() in lower for x in specs[0][2] if x): | |
| selected = specs[0] | |
| elif f"controlling {str(constructs.get('mediator') or '').casefold()}" in lower and str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower: | |
| selected = specs[1] | |
| elif f"controlling {str(constructs.get('predictor') or '').casefold()}" in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower: | |
| selected = specs[3] | |
| elif str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() not in lower: | |
| selected = specs[2] | |
| elif str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower and str(constructs.get("predictor") or "").casefold() not in lower: | |
| selected = specs[3] | |
| if selected and selected[1] is not None: | |
| add_metric_violations(sentence, r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", selected[1], label=selected[0]) | |
| elif not strict: | |
| for label, expected in dict(uploaded.get("r2_by_label") or {}).items(): | |
| if label and label in sentence.casefold(): | |
| add_metric_violations(sentence, rf"(R(?:²|\^2)\s*(?:for|of)\s*{re.escape(label)}\s*(?:=|:)\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_R2:{label}") | |
| for path, expected in dict(uploaded.get("beta_by_path") or {}).items(): | |
| left, _, right = path.partition(" -> ") | |
| if left and right and left in sentence.casefold() and right in sentence.casefold(): | |
| add_metric_violations(sentence, r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_beta:{path}") | |
| return violations | |
| def repair_quantitative_study_fact_contract(cls, text, metadata, section_name=""): | |
| """Repair deterministic AMJ drift only in study/result-owned prose.""" | |
| value = str(text or "") | |
| meta = dict(metadata or {}) | |
| ledger = dict(meta.get("quantitative_evidence_ledger") or {}) | |
| if not ledger: | |
| ledger = cls.build_quantitative_evidence_ledger(meta) | |
| diagnostics = [] | |
| sample_size = ledger.get("sample_size") | |
| scale_points = ledger.get("scale_points") | |
| strict = bool(ledger.get("strict")) | |
| constructs = dict(ledger.get("constructs") or {}) | |
| results = dict(ledger.get("association_results") or {}) | |
| uploaded = dict(ledger.get("uploaded_metric_anchors") or {}) | |
| known_constructs = [str(v) for v in constructs.values() if str(v)] | |
| construct_folded = [c.casefold() for c in known_constructs] | |
| forbidden = [str(x) for x in ledger.get("forbidden_unowned_methods") or []] | |
| role_patterns = { | |
| "predictor": r"(?:predictor|independent\s+variable|predictor\s+construct)", | |
| "mediator": r"(?:mediator|mediating\s+variable|mediating\s+construct)", | |
| "outcome": r"(?:outcome|dependent\s+variable|outcome\s+construct)", | |
| } | |
| causal_replacements = ( | |
| (r"(?i)\bhad\s+a\s+causal\s+effect\s+on\b", "was associated with"), | |
| (r"(?i)\bexerted\s+a\s+causal\s+effect\s+on\b", "was associated with"), | |
| (r"(?i)\bcausal\s+effect\s+on\b", "association with"), | |
| (r"(?i)\ban\s+causal\s+effect\b", "an association"), | |
| (r"(?i)\ba\s+causal\s+effect\b", "an association"), | |
| (r"(?i)\bthe\s+causal\s+effect\b", "the association"), | |
| (r"(?i)\bcausal\s+effects\b", "associations"), | |
| (r"(?i)\bcausal\s+effect\b", "association"), | |
| (r"(?i)\bcausal\s+mediation\b", "indirect association"), | |
| ) | |
| replacements = [] | |
| for start, end, sentence in cls._sentence_spans(value): | |
| ownership = cls._quantitative_sentence_ownership(sentence, section_name, known_constructs) | |
| if ownership not in {"study", "result"}: | |
| continue | |
| revised = sentence | |
| if sample_size: | |
| sample_pattern = re.compile(r"(?i)(\b(?:sample(?:\s+size)?(?:\s+of)?|N)\s*(?:=|:)?\s*)(\d{2,6})\b") | |
| def sample_repl(match): | |
| observed = int(match.group(2)) | |
| if observed == int(sample_size): | |
| return match.group(0) | |
| diagnostics.append({"type":"sample_size_repaired","observed":observed,"expected":int(sample_size)}) | |
| return match.group(1) + str(int(sample_size)) | |
| revised = sample_pattern.sub(sample_repl, revised) | |
| count_pattern = re.compile(r"(?i)\b(\d{2,6})\s+(respondents|participants|observations|cases)\b") | |
| def count_repl(match): | |
| observed = int(match.group(1)) | |
| if observed == int(sample_size): | |
| return match.group(0) | |
| diagnostics.append({"type":"sample_count_repaired","observed":observed,"expected":int(sample_size)}) | |
| return f"{int(sample_size)} {match.group(2)}" | |
| revised = count_pattern.sub(count_repl, revised) | |
| if scale_points: | |
| scale_pattern = re.compile(r"(?i)\b([2-9]|1[0-1])[- ]point\s+(Likert(?:-type)?\s+scale)") | |
| def scale_repl(match): | |
| observed = int(match.group(1)) | |
| if observed == int(scale_points): | |
| return match.group(0) | |
| diagnostics.append({"type":"scale_points_repaired","observed":observed,"expected":int(scale_points)}) | |
| return f"{int(scale_points)}-point {match.group(2)}" | |
| revised = scale_pattern.sub(scale_repl, revised) | |
| for role, role_pattern in role_patterns.items(): | |
| expected = str(constructs.get(role) or "") | |
| if not expected: | |
| continue | |
| for candidate in known_constructs: | |
| if candidate.casefold() == expected.casefold(): | |
| continue | |
| pattern = re.compile(rf"(?i)({role_pattern}\s*(?:was|is|:|=)?\s*){re.escape(candidate)}") | |
| revised, count = pattern.subn(lambda m: m.group(1) + expected, revised) | |
| if count: | |
| diagnostics.append({"type":"construct_role_repaired","role":role,"observed":candidate,"expected":expected,"count":count}) | |
| if strict: | |
| for pattern, replacement in causal_replacements: | |
| regex = re.compile(pattern) | |
| repaired_count = 0 | |
| def causal_repl(match): | |
| nonlocal repaired_count | |
| if cls._causal_phrase_is_explicitly_disclaimed( | |
| revised, match.start(), match.end() | |
| ): | |
| return match.group(0) | |
| repaired_count += 1 | |
| return replacement | |
| revised = regex.sub(causal_repl, revised) | |
| if repaired_count: | |
| diagnostics.append({ | |
| "type":"causal_language_repaired", | |
| "replacement":replacement, | |
| "count":repaired_count, | |
| }) | |
| if any(c in revised.casefold() for c in construct_folded) if construct_folded else False: | |
| revised, p_count = re.subn(r"(?i)\s*[,;]?\s*p\s*(?:=|<|>|≤|≥)\s*(?:0?\.\d+|\d+(?:\.\d+)?)", "", revised) | |
| if p_count: | |
| diagnostics.append({"type":"unfrozen_p_value_removed","count":p_count}) | |
| if ownership == "study": | |
| bad = next((term for term in forbidden if re.search(rf"(?i)(?<![A-Za-z]){re.escape(term)}(?![A-Za-z])", revised)), "") | |
| if bad: | |
| diagnostics.append({"type":"unsupported_quantitative_method_removed","method":bad,"preview":sentence.strip()[:180]}) | |
| revised = "" | |
| def replace_local_metric(pattern, expected, *, dtype="float", label="metric"): | |
| nonlocal revised | |
| if expected is None: | |
| return | |
| expected_text = str(int(expected)) if dtype == "int" else f"{float(expected):.3f}".rstrip("0").rstrip(".") | |
| regex = re.compile(pattern, re.I) | |
| def repl(match): | |
| observed = match.group(2) | |
| try: | |
| same = int(float(observed)) == int(expected) if dtype == "int" else abs(float(observed) - float(expected)) < 0.0006 | |
| except Exception: | |
| same = False | |
| if same: | |
| return match.group(0) | |
| diagnostics.append({"type":"quantitative_metric_repaired","metric":label,"observed":observed,"expected":expected_text}) | |
| return match.group(1) + expected_text | |
| revised = regex.sub(repl, revised) | |
| vif_expected = results.get("predictor_mediator_vif") if strict else uploaded.get("VIF") | |
| boot_expected = results.get("bootstrap_iterations") if strict else uploaded.get("bootstrap_iterations") | |
| replace_local_metric(r"(\bVIF\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", vif_expected, label="VIF") | |
| replace_local_metric(r"(\bbootstrap\s+(?:resamples?|iterations?)\s*(?:=|:)?\s*)(\d{2,6})", boot_expected, dtype="int", label="bootstrap_iterations") | |
| if strict and results: | |
| for role, key in (("mediator", "r2_mediator"), ("outcome", "r2_outcome")): | |
| construct = str(constructs.get(role) or "") | |
| if construct and results.get(key) is not None: | |
| metric_pattern = rf"(R(?:²|\^2)\s*(?:for|of)?\s*{re.escape(construct)}\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)" | |
| replace_local_metric(metric_pattern, results.get(key), label=key) | |
| lower = revised.casefold() | |
| selected = None | |
| specs = [ | |
| ("indirect", results.get("indirect"), [constructs.get("predictor"), constructs.get("mediator"), constructs.get("outcome")]), | |
| ("direct", results.get("direct"), [constructs.get("predictor"), constructs.get("outcome")]), | |
| ("a", results.get("a"), [constructs.get("predictor"), constructs.get("mediator")]), | |
| ("b", results.get("b"), [constructs.get("mediator"), constructs.get("outcome")]), | |
| ] | |
| if "indirect" in lower and all(str(x or "").casefold() in lower for x in specs[0][2] if x): | |
| selected = specs[0] | |
| elif f"controlling {str(constructs.get('mediator') or '').casefold()}" in lower and str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower: | |
| selected = specs[1] | |
| elif f"controlling {str(constructs.get('predictor') or '').casefold()}" in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower: | |
| selected = specs[3] | |
| elif str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() not in lower: | |
| selected = specs[2] | |
| elif str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower and str(constructs.get("predictor") or "").casefold() not in lower: | |
| selected = specs[3] | |
| if selected and selected[1] is not None: | |
| replace_local_metric(r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", selected[1], label=selected[0]) | |
| elif not strict: | |
| r2_map = dict(uploaded.get("r2_by_label") or {}) | |
| for label, expected in r2_map.items(): | |
| if label and label in revised.casefold(): | |
| replace_local_metric(rf"(R(?:²|\^2)\s*(?:for|of)\s*{re.escape(label)}\s*(?:=|:)\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_R2:{label}") | |
| beta_map = dict(uploaded.get("beta_by_path") or {}) | |
| for path, expected in beta_map.items(): | |
| left, _, right = path.partition(" -> ") | |
| if left and right and left in revised.casefold() and right in revised.casefold(): | |
| replace_local_metric(r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_beta:{path}") | |
| if revised != sentence: | |
| replacements.append((start, end, revised)) | |
| for start, end, replacement in sorted(replacements, reverse=True): | |
| value = value[:start] + replacement + value[end:] | |
| value = re.sub(r"[ \t]+\n", "\n", value) | |
| value = re.sub(r"\n{3,}", "\n\n", value) | |
| value = re.sub(r" {2,}", " ", value) | |
| return value.strip(), diagnostics | |
| 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["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() | |