import csv from pathlib import Path import numpy as np import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt ROOT = Path(__file__).resolve().parents[1] RESULTS_DIR = ROOT / "results" FIGURES_DIR = ROOT / "figures" DEFAULT_SEED = 20260925 N_RECORDS = 10000 N_TOPICS = 16 N_GROUPS = 100 REPEATS = 1000 DEPTHS = (0.05, 0.10, 0.20) METHODS = ("random", "keyword", "semantic_noise") METRICS = ("recall", "precision", "topic_coverage", "shannon_evenness", "effective_topics", "head_share", "normalized_hhi") PARAMETERS = { "seed": DEFAULT_SEED, "n_records": N_RECORDS, "n_topics": N_TOPICS, "n_groups": N_GROUPS, "repeats": REPEATS, "prevalence": 0.02, "topic_alpha": 0.80, "head_fraction": 0.10, "semantic_correlation": 0.50, "boundary_semantic_correlation": 0.0, "relevance_topic_scale": 1.40, "relevance_group_scale": 1.00, "relevance_noise_scale": 0.70, "keyword_topic_scale": 1.50, "keyword_group_scale": 0.45, "keyword_noise_scale": 0.90, "head_group_share_design": 0.20, } PARAMETER_FIELDS = ( "seed", "n_records", "n_topics", "n_groups", "repeats", "prevalence", "topic_alpha", "head_fraction", "semantic_correlation", "boundary_semantic_correlation", "relevance_topic_scale", "relevance_group_scale", "relevance_noise_scale", "keyword_topic_scale", "keyword_group_scale", "keyword_noise_scale", "head_group_share_design", "head_group_count", "target_relevant", "query_target_topic_ids", "depth_grid", "method_order", ) def standardize(values): values = np.asarray(values, dtype=float) centered = values - np.mean(values) scale = np.std(values) if not np.isfinite(scale) or scale <= 0.0: return np.zeros(values.shape, dtype=float) return centered / scale def build_design(seed, topic_count, group_count, head_count): design_rng = np.random.default_rng(np.random.SeedSequence([int(seed), 1729])) target_count = max(2, topic_count // 4) target_topics = np.sort(design_rng.choice(topic_count, size=target_count, replace=False)) topic_relevance = design_rng.normal(0.0, 1.0, topic_count) topic_relevance[target_topics] += 1.65 keyword_topics = design_rng.normal(0.0, 1.0, topic_count) keyword_topics[target_topics] += 1.55 group_relevance = design_rng.normal(0.0, 1.0, group_count) group_relevance[:head_count] += 1.70 keyword_groups = design_rng.normal(0.0, 1.0, group_count) keyword_groups[:head_count] += 0.75 head_per_group = 0.020 tail_per_group = (1.0 - head_count * head_per_group) / (group_count - head_count) group_probabilities = np.full(group_count, tail_per_group, dtype=float) group_probabilities[:head_count] = head_per_group group_probabilities /= np.sum(group_probabilities) return { "target_topics": target_topics, "topic_relevance": topic_relevance, "keyword_topics": keyword_topics, "group_relevance": group_relevance, "keyword_groups": keyword_groups, "group_probabilities": group_probabilities, } def generate_instance(rng, design): topic_probabilities = rng.dirichlet(np.full(N_TOPICS, PARAMETERS["topic_alpha"])) topics = rng.choice(N_TOPICS, size=N_RECORDS, replace=True, p=topic_probabilities) groups = rng.choice(N_GROUPS, size=N_RECORDS, replace=True, p=design["group_probabilities"]) latent_relevance = ( PARAMETERS["relevance_topic_scale"] * design["topic_relevance"][topics] + PARAMETERS["relevance_group_scale"] * design["group_relevance"][groups] + PARAMETERS["relevance_noise_scale"] * rng.normal(0.0, 1.0, N_RECORDS) ) target_relevant = int(round(PARAMETERS["prevalence"] * N_RECORDS)) relevance_order = np.argsort(-latent_relevance, kind="mergesort") labels = np.zeros(N_RECORDS, dtype=np.int8) labels[relevance_order[:target_relevant]] = 1 random_score = rng.random(N_RECORDS) keyword_noise = rng.normal(0.0, 1.0, N_RECORDS) semantic_noise = rng.normal(0.0, 1.0, N_RECORDS) keyword_score = ( PARAMETERS["keyword_topic_scale"] * design["keyword_topics"][topics] + PARAMETERS["keyword_group_scale"] * design["keyword_groups"][groups] + PARAMETERS["keyword_noise_scale"] * standardize(keyword_noise) ) semantic_latent = standardize(latent_relevance) semantic_noise_standardized = standardize(semantic_noise) rho = PARAMETERS["semantic_correlation"] semantic_score = rho * semantic_latent + np.sqrt(1.0 - rho * rho) * semantic_noise_standardized return { "topics": topics, "groups": groups, "labels": labels, "scores": { "random": random_score, "keyword": keyword_score, "semantic_noise": semantic_score, }, "zero_signal_score": semantic_noise_standardized, "semantic_latent": semantic_latent, } def selected_metrics(topics, groups, labels, selected, head_mask, target_relevant): selected_count = selected.size if selected_count <= 0: raise ValueError("selected count must be positive") hits = int(np.count_nonzero(labels[selected])) topic_counts = np.bincount(topics[selected], minlength=N_TOPICS).astype(float) topic_proportions = topic_counts / selected_count positive_proportions = topic_proportions[topic_proportions > 0.0] entropy = float(-np.sum(positive_proportions * np.log(positive_proportions))) evenness = entropy / np.log(N_TOPICS) effective_topics = float(np.exp(entropy)) topic_coverage = float(np.count_nonzero(topic_counts > 0.0) / N_TOPICS) group_counts = np.bincount(groups[selected], minlength=N_GROUPS).astype(float) group_proportions = group_counts / selected_count hhi = float(np.sum(group_proportions * group_proportions)) normalized_hhi = (hhi - 1.0 / N_GROUPS) / (1.0 - 1.0 / N_GROUPS) head_share = float(np.mean(head_mask[groups[selected]])) return np.array( [ hits / target_relevant, hits / selected_count, topic_coverage, evenness, effective_topics, head_share, normalized_hhi, ], dtype=float, ) def rank_indices(scores): order = np.argsort(-scores, kind="mergesort") return [order[: int(round(depth * N_RECORDS))] for depth in DEPTHS] def random_expected_recall(selected_count, target_relevant): if selected_count < target_relevant: return selected_count / target_relevant log_probability = 0.0 for offset in range(target_relevant): log_probability += np.log((selected_count - offset) / (N_RECORDS - offset)) return float(np.exp(log_probability)) def quantile_summary(values): return np.quantile(values, [0.025, 0.5, 0.975], axis=0) def format_float(value): value = float(value) if not np.isfinite(value): raise ValueError("non-finite numeric output") return f"{value:.12g}" def parameter_row(head_count, target_relevant, target_topics): return { "seed": PARAMETERS["seed"], "n_records": PARAMETERS["n_records"], "n_topics": PARAMETERS["n_topics"], "n_groups": PARAMETERS["n_groups"], "repeats": PARAMETERS["repeats"], "prevalence": format_float(PARAMETERS["prevalence"]), "topic_alpha": format_float(PARAMETERS["topic_alpha"]), "head_fraction": format_float(PARAMETERS["head_fraction"]), "semantic_correlation": format_float(PARAMETERS["semantic_correlation"]), "boundary_semantic_correlation": format_float(PARAMETERS["boundary_semantic_correlation"]), "relevance_topic_scale": format_float(PARAMETERS["relevance_topic_scale"]), "relevance_group_scale": format_float(PARAMETERS["relevance_group_scale"]), "relevance_noise_scale": format_float(PARAMETERS["relevance_noise_scale"]), "keyword_topic_scale": format_float(PARAMETERS["keyword_topic_scale"]), "keyword_group_scale": format_float(PARAMETERS["keyword_group_scale"]), "keyword_noise_scale": format_float(PARAMETERS["keyword_noise_scale"]), "head_group_share_design": format_float(PARAMETERS["head_group_share_design"]), "head_group_count": head_count, "target_relevant": target_relevant, "query_target_topic_ids": "|".join(str(int(value)) for value in target_topics), "depth_grid": "|".join(f"{depth:.2f}" for depth in DEPTHS), "method_order": "|".join(METHODS), } def write_csv(path, fieldnames, rows): with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames, lineterminator="\n") writer.writeheader() writer.writerows(rows) def result_rows(samples, head_count, target_relevant, target_topics): rows = [] base = parameter_row(head_count, target_relevant, target_topics) fields = list(PARAMETER_FIELDS) + ["depth", "k", "method", "random_expected_recall"] fields += [f"{metric}_{statistic}" for metric in METRICS for statistic in ("mean", "median", "q025", "q975")] for depth_index, depth in enumerate(DEPTHS): selected_count = int(round(depth * N_RECORDS)) expected_recall = random_expected_recall(selected_count, target_relevant) for method_index, method in enumerate(METHODS): values = samples[:, method_index, depth_index, :] summary = quantile_summary(values) row = dict(base) row.update( { "depth": format_float(depth), "k": selected_count, "method": method, "random_expected_recall": format_float(expected_recall), } ) for metric_index, metric in enumerate(METRICS): row[f"{metric}_mean"] = format_float(np.mean(values[:, metric_index])) row[f"{metric}_median"] = format_float(summary[1, metric_index]) row[f"{metric}_q025"] = format_float(summary[0, metric_index]) row[f"{metric}_q975"] = format_float(summary[2, metric_index]) rows.append(row) return rows, fields def comparison_rows(samples, head_count, target_relevant, target_topics): rows = [] base = parameter_row(head_count, target_relevant, target_topics) pairs = (("keyword", "random"), ("semantic_noise", "random"), ("semantic_noise", "keyword")) fields = list(PARAMETER_FIELDS) + ["depth", "k", "method", "reference", "metric", "paired_difference_mean", "paired_difference_median", "paired_difference_q025", "paired_difference_q975"] for method, reference in pairs: method_index = METHODS.index(method) reference_index = METHODS.index(reference) for depth_index, depth in enumerate(DEPTHS): selected_count = int(round(depth * N_RECORDS)) for metric_index, metric in enumerate(METRICS): differences = samples[:, method_index, depth_index, metric_index] - samples[:, reference_index, depth_index, metric_index] summary = quantile_summary(differences) row = dict(base) row.update( { "depth": format_float(depth), "k": selected_count, "method": method, "reference": reference, "metric": metric, "paired_difference_mean": format_float(np.mean(differences)), "paired_difference_median": format_float(summary[1]), "paired_difference_q025": format_float(summary[0]), "paired_difference_q975": format_float(summary[2]), } ) rows.append(row) return rows, fields def boundary_rows(boundary_correlations, boundary_recalls, head_count, target_relevant, target_topics): rows = [] base = parameter_row(head_count, target_relevant, target_topics) correlation_summary = quantile_summary(boundary_correlations) fields = list(PARAMETER_FIELDS) + ["depth", "k", "zero_signal_correlation_mean", "zero_signal_correlation_median", "zero_signal_correlation_q025", "zero_signal_correlation_q975", "zero_signal_recall_mean", "zero_signal_recall_median", "zero_signal_recall_q025", "zero_signal_recall_q975", "random_expected_recall"] for depth_index, depth in enumerate(DEPTHS): selected_count = int(round(depth * N_RECORDS)) recall_summary = quantile_summary(boundary_recalls[:, depth_index]) row = dict(base) row.update( { "depth": format_float(depth), "k": selected_count, "zero_signal_correlation_mean": format_float(np.mean(boundary_correlations)), "zero_signal_correlation_median": format_float(correlation_summary[1]), "zero_signal_correlation_q025": format_float(correlation_summary[0]), "zero_signal_correlation_q975": format_float(correlation_summary[2]), "zero_signal_recall_mean": format_float(np.mean(boundary_recalls[:, depth_index])), "zero_signal_recall_median": format_float(recall_summary[1]), "zero_signal_recall_q025": format_float(recall_summary[0]), "zero_signal_recall_q975": format_float(recall_summary[2]), "random_expected_recall": format_float(random_expected_recall(selected_count, target_relevant)), } ) rows.append(row) return rows, fields def make_tradeoff_figure(samples): means = np.mean(samples, axis=0) intervals = quantile_summary(samples) metric_index = {metric: index for index, metric in enumerate(METRICS)} colors = {"random": "tab:blue", "keyword": "tab:orange", "semantic_noise": "tab:green"} figure, axes = plt.subplots(1, 2, figsize=(11.0, 4.6), constrained_layout=True) depth_values = np.asarray(DEPTHS) * 100.0 for method_index, method in enumerate(METHODS): color = colors[method] for axis, metric in zip(axes, ("topic_coverage", "normalized_hhi")): x_values = means[method_index, :, metric_index[metric]] y_values = means[method_index, :, metric_index["recall"]] lower = y_values - intervals[0, method_index, :, metric_index["recall"]] upper = intervals[2, method_index, :, metric_index["recall"]] - y_values axis.errorbar( x_values, y_values, yerr=np.vstack([lower, upper]), color=color, marker="o", linewidth=1.8, markersize=5.5, capsize=3, label=method, ) for depth_value, x_value, y_value in zip(depth_values, x_values, y_values): axis.annotate(f"{depth_value:.0f}%", (x_value, y_value), xytext=(4, 4), textcoords="offset points", fontsize=8, color=color) axes[0].set_title("Recall vs topic coverage") axes[0].set_xlabel("Topic coverage") axes[0].set_ylabel("Mean recall") axes[1].set_title("Recall vs group concentration") axes[1].set_xlabel("Normalized HHI") axes[1].set_ylabel("Mean recall") for axis in axes: axis.grid(True, alpha=0.25) axis.set_axisbelow(True) axes[0].legend(loc="best", frameon=False) figure.suptitle("Literature screening: recall and trade-off indicators") figure.savefig(FIGURES_DIR / "literature_screening_tradeoff.png", dpi=180, facecolor="white", metadata={"Software": "literature_screening.py"}) plt.close(figure) def make_depth_figure(samples): means = np.mean(samples, axis=0) intervals = quantile_summary(samples) metric_index = {metric: index for index, metric in enumerate(METRICS)} colors = {"random": "tab:blue", "keyword": "tab:orange", "semantic_noise": "tab:green"} figure, axes = plt.subplots(2, 2, figsize=(11.0, 8.0)) panels = ( (("recall", "precision"), "Recall and precision", "Rate"), (("topic_coverage", "shannon_evenness"), "Topic coverage and evenness", "Index"), (("effective_topics", "head_share"), "Effective topics and head share", "Value"), (("normalized_hhi",), "Group concentration", "Index"), ) depth_values = np.asarray(DEPTHS) * 100.0 for axis, (metrics, title, ylabel) in zip(axes.flat, panels): for method_index, method in enumerate(METHODS): color = colors[method] for metric_position, metric in enumerate(metrics): values = means[method_index, :, metric_index[metric]] lower = intervals[0, method_index, :, metric_index[metric]] upper = intervals[2, method_index, :, metric_index[metric]] label = method if metric_position == 0 else "_nolegend_" axis.plot(depth_values, values, color=color, marker="o", linewidth=1.8, markersize=5.0, label=label) axis.fill_between(depth_values, lower, upper, color=color, alpha=0.10) axis.set_title(title) axis.set_xlabel("Screening depth (%)") axis.set_ylabel(ylabel) axis.set_xticks(depth_values) axis.grid(True, alpha=0.25) axis.set_axisbelow(True) handles, labels = axes.flat[0].get_legend_handles_labels() figure.subplots_adjust(left=0.08, right=0.98, bottom=0.08, top=0.88, wspace=0.25, hspace=0.35) figure.legend(handles, labels, loc="upper center", bbox_to_anchor=(0.5, 0.925), ncol=3, frameon=False) figure.suptitle("Literature screening metrics by fixed depth", y=0.98) figure.savefig(FIGURES_DIR / "literature_screening_depth.png", dpi=180, facecolor="white", metadata={"Software": "literature_screening.py"}) plt.close(figure) def print_summary(samples, boundary_correlations): means = np.mean(samples, axis=0) intervals = quantile_summary(samples) print("status=ok") print( f"main scenario: seed={PARAMETERS['seed']} N={N_RECORDS} T={N_TOPICS} G={N_GROUPS} repeats={REPEATS} prevalence={PARAMETERS['prevalence']:.3f} semantic_correlation={PARAMETERS['semantic_correlation']:.2f}" ) for depth_index, depth in enumerate(DEPTHS): print(f"depth={depth:.2f} k={int(round(depth * N_RECORDS))} random_full_recall_probability={random_expected_recall(int(round(depth * N_RECORDS)), int(round(PARAMETERS['prevalence'] * N_RECORDS))):.3e}") for method_index, method in enumerate(METHODS): values = means[method_index, depth_index, :] low = intervals[0, method_index, depth_index, :] high = intervals[2, method_index, depth_index, :] print( f" {method}: recall={values[0]:.4f} [{low[0]:.4f},{high[0]:.4f}] precision={values[1]:.4f} topic_coverage={values[2]:.4f} evenness={values[3]:.4f} effective_topics={values[4]:.2f} head_share={values[5]:.4f} normalized_hhi={values[6]:.4f}" ) correlation_summary = quantile_summary(boundary_correlations) zero_signal_pass = abs(np.mean(boundary_correlations)) < 0.02 and correlation_summary[0] <= 0.0 <= correlation_summary[2] print( f"zero-signal boundary: correlation_mean={np.mean(boundary_correlations):.6f} q025={correlation_summary[0]:.6f} q975={correlation_summary[2]:.6f} pass={str(zero_signal_pass).lower()}" ) print("limitations:") print("- Synthetic labels and proxy scores do not estimate effects of a real AI or keyword-search product.") print("- Topic, entropy, effective-topic, head-share, and HHI values are descriptive model quantities, not quality, fairness, epistemic-diversity, or policy metrics.") print("- Fixed-depth recall is conditional on this generator; 95% recall is not a universal stopping rule.") print("- Database recall, labeling error, real workload, time, cost, and downstream review validity are not modeled.") print("- The zero-signal check validates a constructed boundary only and does not establish external validity.") def main(): RESULTS_DIR.mkdir(parents=True, exist_ok=True) FIGURES_DIR.mkdir(parents=True, exist_ok=True) head_count = int(round(N_GROUPS * PARAMETERS["head_fraction"])) target_relevant = int(round(PARAMETERS["prevalence"] * N_RECORDS)) design = build_design(PARAMETERS["seed"], N_TOPICS, N_GROUPS, head_count) head_mask = np.arange(N_GROUPS) < head_count replicate_seeds = np.random.SeedSequence(PARAMETERS["seed"]).spawn(REPEATS) samples = np.empty((REPEATS, len(METHODS), len(DEPTHS), len(METRICS)), dtype=float) boundary_recalls = np.empty((REPEATS, len(DEPTHS)), dtype=float) boundary_correlations = np.empty(REPEATS, dtype=float) for replicate_index, replicate_seed in enumerate(replicate_seeds): rng = np.random.default_rng(replicate_seed) instance = generate_instance(rng, design) for method_index, method in enumerate(METHODS): selected = rank_indices(instance["scores"][method]) for depth_index, selected_indices in enumerate(selected): samples[replicate_index, method_index, depth_index, :] = selected_metrics( instance["topics"], instance["groups"], instance["labels"], selected_indices, head_mask, target_relevant, ) zero_selected = rank_indices(instance["zero_signal_score"]) for depth_index, selected_indices in enumerate(zero_selected): boundary_recalls[replicate_index, depth_index] = np.count_nonzero(instance["labels"][selected_indices]) / target_relevant boundary_correlations[replicate_index] = np.dot(instance["semantic_latent"], standardize(instance["zero_signal_score"])) / (N_RECORDS - 1.0) if not np.isfinite(samples).all() or not np.isfinite(boundary_recalls).all() or not np.isfinite(boundary_correlations).all(): raise ValueError("simulation produced non-finite values") results, result_fields = result_rows(samples, head_count, target_relevant, design["target_topics"]) comparisons, comparison_fields = comparison_rows(samples, head_count, target_relevant, design["target_topics"]) boundary, boundary_fields = boundary_rows(boundary_correlations, boundary_recalls, head_count, target_relevant, design["target_topics"]) parameter_path = RESULTS_DIR / "literature_screening_parameters.csv" results_path = RESULTS_DIR / "literature_screening_results.csv" comparison_path = RESULTS_DIR / "literature_screening_comparisons.csv" boundary_path = RESULTS_DIR / "literature_screening_boundary.csv" write_csv(parameter_path, PARAMETER_FIELDS, [parameter_row(head_count, target_relevant, design["target_topics"])]) write_csv(results_path, result_fields, results) write_csv(comparison_path, comparison_fields, comparisons) write_csv(boundary_path, boundary_fields, boundary) make_tradeoff_figure(samples) make_depth_figure(samples) print_summary(samples, boundary_correlations) print(f"outputs: {parameter_path.name}, {results_path.name}, {comparison_path.name}, {boundary_path.name}, literature_screening_tradeoff.png, literature_screening_depth.png") if __name__ == "__main__": main()