# -*- coding: utf-8 -*- """Quantitative data preparation for SmartPaper's AMJ track.""" from __future__ import annotations 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: """Keep quantitative preparation isolated from the Streamlit UI.""" @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) 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 = [ "Independent_Var", "Mediator_Var", "Dependent_Var", ] 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 = [] 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"{construct_name}_{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 } ) fl_matrix = mean_scores.corr().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] = "" desc_stats = mean_scores.mean().to_frame("Mean").join( mean_scores.std().to_frame("SD") ) return df, validity_df, desc_stats.join(fl_matrix) def resolve_empirical_data(self, uploaded_files=None, language="zh"): 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"], } return None, user_md, tr(language, "actual_stats_parsed"), metadata df, validity_df, fl_matrix_df = self.generate_empirical_data_robust() generated_prompt = f"""### 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()} [STRICT EMPIRICAL INSTRUCTION] 1. ZERO HALLUCINATION: Report the exact metrics provided above. 2. VALIDITY DEFENSE: State convergent validity only when the displayed AVE and CR values support it. 3. DISCRIMINANT DEFENSE: State discriminant validity only when the displayed diagonal and correlation values support it. 4. CONSISTENCY: Use these tables as the single source of numerical truth throughout the paper. """ metadata = { "mode": "quantitative", "source_type": "system_generated", "disclosure_required": True, "sample_size": int(len(df)), "scale_points": 7, "method": ( "Positive-semidefinite latent correlation generation, " "indicator loadings, discretization, and derived " "reliability/validity summaries." ), "generation_parameters": { "constructs": [ "Independent_Var", "Mediator_Var", "Dependent_Var", ], "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", "reported_summaries": [ "Cronbach alpha", "composite reliability", "average variance extracted", "descriptive statistics", "Fornell-Larcker matrix", ], }, "evidence_sections": [ "Reliability and convergent validity", "Descriptive statistics and discriminant validity", ], } return ( df, generated_prompt, tr(language, "synthetic_stats_generated"), metadata, ) def get_quantitative_engine(): return QuantitativeEngine()