Spaces:
Running
Running
| # -*- 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.""" | |
| 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 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() | |