# -*- 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.""" @staticmethod def generate_psd_corr_matrix(n_constructs, max_corr=0.5): if n_constructs == 1: return np.array([[1.0]]) base_corr = np.random.uniform(0.2, max_corr - 0.1) matrix = np.random.uniform( base_corr - 0.1, base_corr + 0.1, (n_constructs, n_constructs), ) matrix = (matrix + matrix.T) / 2 np.fill_diagonal(matrix, 1.0) min_eig = np.min(np.linalg.eigvals(matrix)) if min_eig < 0: matrix -= (min_eig - 0.01) * np.eye(n_constructs) diagonal = np.sqrt(np.diag(matrix)) return matrix / np.outer(diagonal, diagonal) @staticmethod def calculate_validity_metrics(df, constructs_info, true_loadings): metrics = [] for construct in constructs_info: construct_name = construct["name"] items = construct["items"] sub_df = df[items] var_total = sub_df.sum(axis=1).var(ddof=1) sum_var = sub_df.var(ddof=1).sum() n_items = len(items) alpha = ( 0.0 if var_total <= 0 else (n_items / (n_items - 1)) * (1 - sum_var / var_total) ) loadings = np.array(true_loadings[construct_name]) sum_loadings_sq = np.sum(loadings) ** 2 sum_sq_loadings = np.sum(loadings**2) sum_error = np.sum(1 - loadings**2) cr = sum_loadings_sq / (sum_loadings_sq + sum_error) ave = sum_sq_loadings / len(loadings) metrics.append( { "Construct": construct_name, "Items": n_items, "Cronbach's α": round(alpha, 3), "CR": round(cr, 3), "AVE": round(ave, 3), } ) return pd.DataFrame(metrics) @staticmethod def _safe_constructs(values): cleaned = [] for value in values or []: name = re.sub(r"\s+", " ", str(value or "")).strip(" .,:;-_\n\t") if not name or len(name) > 80 or name.lower() in {item.lower() for item in cleaned}: continue cleaned.append(name) return cleaned[:3] def _derive_system_design(self, agent, topic="", keywords=""): """Freeze a parsimonious topic-aligned design without inventing named sites.""" fallback = { "constructs": [ "Focal Predictor", "Focal Mediator", "Focal Outcome", ], "research_context": "respondents with experience relevant to the focal management phenomenon", } if agent is None: return fallback prompt = f"""Design the minimum coherent quantitative management-study skeleton for the topic below. Return ONLY valid JSON: {{"constructs":["predictor construct","mediating construct","outcome construct"],"research_context":"one generic respondent context"}} Rules: 1. Exactly three distinct constructs in predictor -> mediator -> outcome order. 2. Use concise construct labels that can appear throughout a journal article. 3. The context must be generic and directly relevant to the topic. Do NOT name or invent a country, university, firm, industry sample frame, software package, time period, or institution. 4. This is a cross-sectional single-wave survey design, so do not imply experimental or causal identification. 5. Do not create any number; sample size and scale are frozen elsewhere. Topic: {topic} Keywords: {keywords}""" try: raw = agent.call_deepseek( "You are a rigorous management research-design editor.", prompt, max_tokens=500, temp=0.0, model_role="fast", thinking=False, allow_fallback=True, ) parsed = agent._extract_json_from_text(raw) if not isinstance(parsed, dict): return fallback constructs = self._safe_constructs(parsed.get("constructs")) context = re.sub(r"\s+", " ", str(parsed.get("research_context") or "")).strip() if len(constructs) != 3: return fallback if not context or len(context) > 240: context = fallback["research_context"] # Named-location/site details are deliberately not accepted from this # design step. The manuscript may remain generic when no user evidence # supplies a real setting. if re.search(r"\b(university|company|corporation|inc\.?|ltd\.?|country|china|usa|united states|europe|asia)\b", context, re.I): context = fallback["research_context"] return {"constructs": constructs, "research_context": context} except Exception as exc: logging.warning("Quantitative study-design freeze fell back safely: %s", exc) return fallback def generate_empirical_data_robust( self, construct_names=None, n=300, scale_points=7, ): if not construct_names or len(construct_names) < 2: construct_names = [ "Focal Predictor", "Focal Mediator", "Focal Outcome", ] n_constructs = len(construct_names) corr = self.generate_psd_corr_matrix(n_constructs, max_corr=0.55) latent_scores = np.random.multivariate_normal( mean=[0] * n_constructs, cov=corr, size=n, ) data = {} true_loadings = {} constructs_info = [] target_mean = scale_points / 2.0 + 0.5 target_std = scale_points / 5.0 for index, construct_name in enumerate(construct_names): true_score = latent_scores[:, index] n_items = np.random.randint(3, 6) loadings = np.random.uniform(0.72, 0.88, n_items) true_loadings[construct_name] = loadings item_names = [] safe_prefix = re.sub(r"[^A-Za-z0-9]+", "_", construct_name).strip("_") or f"Construct_{index + 1}" for item_index, loading in enumerate(loadings): error = np.random.normal( 0, np.sqrt(1 - loading**2), n, ) item_score = loading * true_score + error raw_scaled = item_score * target_std + target_mean item_int = np.clip( np.round(raw_scaled), 1, scale_points, ).astype(int) column_name = f"{safe_prefix}_{item_index + 1}" data[column_name] = item_int item_names.append(column_name) constructs_info.append( {"name": construct_name, "items": item_names} ) df = pd.DataFrame(data) validity_df = self.calculate_validity_metrics( df, constructs_info, true_loadings, ) mean_scores = pd.DataFrame( { construct["name"]: df[construct["items"]].mean(axis=1) for construct in constructs_info } ) corr_matrix = mean_scores.corr() fl_matrix = corr_matrix.astype(object) for _, row in validity_df.iterrows(): construct_name = row["Construct"] sqrt_ave = np.sqrt(row["AVE"]) fl_matrix.loc[construct_name, construct_name] = ( f"({sqrt_ave:.3f})" ) for row_index in range(len(fl_matrix.columns)): for column_index in range(row_index + 1, len(fl_matrix.columns)): fl_matrix.iloc[row_index, column_index] = "" for column_index in range(row_index): value = fl_matrix.iloc[row_index, column_index] if isinstance(value, (int, float, np.integer, np.floating)): fl_matrix.iloc[row_index, column_index] = f"{float(value):.3f}" desc_stats = mean_scores.mean().to_frame("Mean").join( mean_scores.std().to_frame("SD") ) return df, validity_df, desc_stats.join(fl_matrix), mean_scores, corr_matrix @staticmethod def _standardized_ols(y, predictors): y = np.asarray(y, dtype=float) x = np.column_stack([np.asarray(value, dtype=float) for value in predictors]) y_sd = y.std(ddof=1) x_sd = x.std(axis=0, ddof=1) if y_sd <= 0 or np.any(x_sd <= 0): raise RuntimeError("quantitative_zero_variance") yz = (y - y.mean()) / y_sd xz = (x - x.mean(axis=0)) / x_sd design = np.column_stack([np.ones(len(yz)), xz]) coef = np.linalg.lstsq(design, yz, rcond=None)[0] fitted = design @ coef ss_res = float(np.sum((yz - fitted) ** 2)) ss_tot = float(np.sum((yz - yz.mean()) ** 2)) r2 = 0.0 if ss_tot <= 0 else 1.0 - ss_res / ss_tot return coef[1:], max(0.0, min(1.0, r2)) def _path_evidence(self, mean_scores, construct_names, bootstrap_iterations=1500): if len(construct_names) < 3: return {}, pd.DataFrame(), pd.DataFrame() x_name, m_name, y_name = construct_names[0], construct_names[1], construct_names[2] x = mean_scores[x_name].to_numpy(dtype=float) m = mean_scores[m_name].to_numpy(dtype=float) y = mean_scores[y_name].to_numpy(dtype=float) a_coef, r2_m = self._standardized_ols(m, [x]) y_coef, r2_y = self._standardized_ols(y, [x, m]) a = float(a_coef[0]) direct = float(y_coef[0]) b = float(y_coef[1]) indirect = a * b rng = np.random.default_rng(20260809) indirect_boot = [] direct_boot = [] n = len(mean_scores) for _ in range(int(bootstrap_iterations)): idx = rng.integers(0, n, n) try: a_b, _ = self._standardized_ols(m[idx], [x[idx]]) y_b, _ = self._standardized_ols(y[idx], [x[idx], m[idx]]) indirect_boot.append(float(a_b[0] * y_b[1])) direct_boot.append(float(y_b[0])) except Exception: continue if len(indirect_boot) < 100: raise RuntimeError("quantitative_bootstrap_insufficient") indirect_ci = np.percentile(indirect_boot, [2.5, 97.5]) direct_ci = np.percentile(direct_boot, [2.5, 97.5]) path_table = pd.DataFrame( [ {"Association": f"{x_name} → {m_name}", "Standardized estimate": round(a, 3), "R² of outcome": round(r2_m, 3)}, {"Association": f"{m_name} → {y_name} (controlling {x_name})", "Standardized estimate": round(b, 3), "R² of outcome": round(r2_y, 3)}, {"Association": f"{x_name} → {y_name} (controlling {m_name})", "Standardized estimate": round(direct, 3), "R² of outcome": round(r2_y, 3)}, {"Association": f"Indirect association: {x_name} → {m_name} → {y_name}", "Standardized estimate": round(indirect, 3), "R² of outcome": "—"}, ] ) bootstrap_table = pd.DataFrame( [ {"Estimate": "Indirect association", "Point estimate": round(indirect, 3), "95% bootstrap CI lower": round(float(indirect_ci[0]), 3), "95% bootstrap CI upper": round(float(indirect_ci[1]), 3)}, {"Estimate": "Residual direct association", "Point estimate": round(direct, 3), "95% bootstrap CI lower": round(float(direct_ci[0]), 3), "95% bootstrap CI upper": round(float(direct_ci[1]), 3)}, ] ) x_m_corr = float(np.corrcoef(x, m)[0, 1]) vif = 1.0 / max(1e-9, 1.0 - x_m_corr**2) spearman = mean_scores.corr(method="spearman") summary = { "predictor": x_name, "mediator": m_name, "outcome": y_name, "a": round(a, 3), "b": round(b, 3), "direct": round(direct, 3), "indirect": round(indirect, 3), "r2_mediator": round(r2_m, 3), "r2_outcome": round(r2_y, 3), "indirect_ci": [round(float(indirect_ci[0]), 3), round(float(indirect_ci[1]), 3)], "direct_ci": [round(float(direct_ci[0]), 3), round(float(direct_ci[1]), 3)], "predictor_mediator_vif": round(float(vif), 3), "bootstrap_iterations": int(bootstrap_iterations), } return summary, path_table, spearman 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." ), } 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", ], } return ( df, generated_prompt, tr(language, "synthetic_stats_generated"), metadata, ) def get_quantitative_engine(): return QuantitativeEngine()