""" ASReview Stop Criteria Simulation ================================== Authors: C. Kempny, K. Annac, D. Wahidie, Y. Yilmaz-Aslan, P. Brzoska Institution: Lehrstuhl für Versorgungsforschung, Universität Witten/Herdecke Contact: christian.kempny@uni-wh.de Description: This script simulates the ASReview screening process for a single dataset from the SYNERGY repository to evaluate three stop criteria for terminating title and abstract screening. For each combination of initial training sample size (prob1: 1–30%) and 1,000 independent simulation runs, the script records: - Stop criterion 1: estimated vs. actual number of relevant studies - Stop criterion 2: position of first gap exceeding a threshold of consecutive irrelevant studies - Stop criterion 3: position of the last relevant study (as % of dataset) Results are appended row-by-row to a CSV file named after the dataset. Usage: Set DATASET_NAME and OUTPUT_DIR before running. The script was applied separately to each of the five datasets used in the study (Appenzeller-Herzog_2019, Bos_2018, Hall_2012, Menon_2022, Van_der_Waal_2022). Requirements: asreview, synergy-dataset, pandas, pathlib, math, os, shutil """ import math import os import shutil import pandas as pd from pathlib import Path from IPython.display import clear_output from asreview import ASReviewData, ASReviewProject, open_state from asreview.review import ReviewSimulate from asreview.models.classifiers import NaiveBayesClassifier from asreview.models.query import MaxQuery from asreview.models.balance import DoubleBalance from asreview.models.feature_extraction import Tfidf from synergy_dataset import Dataset # ============================================================================= # CONFIGURATION – adjust before running # ============================================================================= DATASET_NAME = "Appenzeller-Herzog_2019" # Name of the SYNERGY dataset OUTPUT_DIR = Path("simulation_data") # Directory for output CSV and temp files N_SIMULATIONS = 1000 # Simulations per sample size SAMPLE_PERCENTAGES = [1, 5, 10, 15, 20, 25, 30] # Initial training sample sizes (%) # ============================================================================= # SETUP # ============================================================================= OUTPUT_DIR.mkdir(exist_ok=True) project_path = OUTPUT_DIR csv_path = str(OUTPUT_DIR / "file.csv") output_csv = str(OUTPUT_DIR / f"{DATASET_NAME}_output.csv") # ============================================================================= # HELPER FUNCTIONS # ============================================================================= def dataset_ready(name_database): """Load a SYNERGY dataset and save it as a CSV for ASReview.""" d = Dataset(name_database) df = d.to_frame() df.to_csv(csv_path, index=False) return df def n_fraction_cases(perc_1, n_cases): """Calculate the number of records for the initial training sample.""" return math.ceil((perc_1 * n_cases) / 100) def random_sample_draw(df, n_required_cases): """ Draw a random sample and count relevant/irrelevant records. If no relevant records are found, n_prior_included is set to 1 to allow ASReview model initialization (fallback for sparse datasets). """ random_samples = df.sample(n_required_cases) counts = random_samples["label_included"].value_counts() n_prior_included = counts.get(1, 0) n_prior_excluded = counts.get(0, 0) if n_prior_included == 0: return 1, n_prior_excluded return n_prior_included, n_prior_excluded def presumed_cases(n_inc_cases, n_frac_cases, n_total_cases): """Extrapolate the estimated total number of relevant studies from the sample.""" return math.ceil((n_inc_cases / n_frac_cases) * n_total_cases) def error_probability(e_prob, n_pres_cases): """ Apply an error tolerance to the estimated number of relevant studies. E.g. e_prob=0.99 means the criterion triggers at 99% of the estimated count. """ return math.ceil(n_pres_cases * e_prob) def stop_x(stop_perc, n_total_cases): """Calculate the absolute number of consecutive irrelevant studies for stop criterion 2.""" return math.ceil((stop_perc * n_total_cases) / 100) def review(n, n_included_sample_rev, n_excluded_sample_rev): """Run a single ASReview simulation and return the ordered screening labels.""" project = ASReviewProject.create( project_path=project_path / f"api_simulation_{n}", project_id=f"api_example_{n}", project_mode="simulate", project_name=f"api_example_{n}", ) project.add_dataset(csv_path) data_obj = ASReviewData.from_file(csv_path) reviewer = ReviewSimulate( as_data=data_obj, model=NaiveBayesClassifier(), query_model=MaxQuery(), balance_model=DoubleBalance(), feature_model=Tfidf(), n_instances=1, project=project, n_prior_included=n_included_sample_rev, n_prior_excluded=n_excluded_sample_rev, stop_if="min", ) project.update_review(status="review") try: reviewer.review() project.mark_review_finished() except Exception as err: project.update_review(status="error") raise err export_path = OUTPUT_DIR / "api_example.asreview" project.export(export_path) with open_state(str(export_path)) as state: labels = state.get_labels(priors=False) return labels def delete_datafiles(n): """Remove temporary ASReview project files after each simulation.""" export_path = OUTPUT_DIR / "api_example.asreview" if export_path.exists(): os.remove(export_path) folder_path = OUTPUT_DIR / f"api_simulation_{n}" if folder_path.exists(): shutil.rmtree(folder_path) def count_included_found(labels): """Count how many relevant studies were found in the simulation.""" return labels.value_counts().get(1, 0) def pos_last_included(labels): """Return the position (index) of the last relevant study in the ranked output.""" indices = labels[labels == 1].index return indices[-1] def check_if_1_stop_crit(count_ones, n_found_in_fraction, n_included_cases_est): """ Evaluate stop criterion 1: difference between estimated and actual number of relevant studies (including those found in the training sample). Positive = overestimated (stopped too late or correctly). Negative = underestimated (stopped too early, relevant studies missed). """ total_found = count_ones + n_found_in_fraction if n_included_cases_est == total_found: return 0 return total_found - n_included_cases_est def check_if_2_stop_crit(stop_after_x, labels): """ Evaluate stop criterion 2: find the first gap larger than stop_after_x consecutive irrelevant studies between two relevant studies. Returns the number of relevant studies missed after that gap, and the position where the gap occurred. """ indices_of_ones = labels[labels == 1].index differences = indices_of_ones[1:] - indices_of_ones[:-1] positions = indices_of_ones[:-1][differences > stop_after_x] if len(positions) == 0: return 0, "N.N" index_in_indices = (indices_of_ones > positions[0]).argmax() missed = len(indices_of_ones) - index_in_indices return missed, positions[0] # ============================================================================= # MAIN SIMULATION LOOP # ============================================================================= # Write CSV header before starting header = pd.DataFrame(columns=[ "Durchgangsnummer", "Anzahl an Studien", "Prozent zum 1. Screen", "Anzahl für 1. Screening", "Trainingsdaten-Anzahl Included", "Trainingsdaten-Anzahl Excluded", "Wahrscheinlich vorhandene Included", "Prozent für Wahrscheinlich Included", "Wahrscheinlich vorhandene Included mit Fehlerwert", "Stopp-Prozent", "Stopp nach X Excluded", "Letzte gefundene Studie bei", "Stop Criterion 1", "Stop Criterion 2 an Position", "Stop Criterion 2 ausgeschlossene Included", "Stop richtig?" ]) header.to_csv(output_csv, index=False) # Load dataset df = dataset_ready(DATASET_NAME) n_cases = len(df) for prob1 in SAMPLE_PERCENTAGES: print(f"Starting prob1 = {prob1}%") for n in range(1, N_SIMULATIONS + 1): n_frac_cases = n_fraction_cases(prob1, n_cases) n_included_sample, n_excluded_sample = random_sample_draw(df, n_frac_cases) labels = review(n, n_included_sample, n_excluded_sample) for prob2 in [0.90, 0.95, 0.99, 1.0]: for prob3 in [1, 5, 10, 15, 20, 25, 30]: n_pres_cases = presumed_cases(n_included_sample, n_frac_cases, n_cases) n_pres_cases_prob = error_probability(prob2, n_pres_cases) stop_after_x = stop_x(prob3, n_cases) n_included_found = count_included_found(labels) pos_last_inc = pos_last_included(labels) crit1 = check_if_1_stop_crit(n_included_found, n_included_sample, n_pres_cases_prob) crit2, crit2_pos = check_if_2_stop_crit(stop_after_x, labels) all_ones = labels[labels == 1].index if crit1 <= 0 and crit2 == 0: result = "Alle Gefunden" elif crit1 > 0: result = "Zu früh abgebrochen nach Stop 1" else: result = "Zu früh abgebrochen nach Stop 2" data_dict = { "Durchgangsnummer": [n], "Anzahl an Studien": [n_cases], "Prozent zum 1. Screen": [prob1], "Anzahl für 1. Screening": [n_frac_cases], "Trainingsdaten-Anzahl Included": [n_included_sample], "Trainingsdaten-Anzahl Excluded": [n_excluded_sample], "Wahrscheinlich vorhandene Included": [n_pres_cases], "Prozent für Wahrscheinlich Included": [prob2], "Wahrscheinlich vorhandene Included mit Fehlerwert": [n_pres_cases_prob], "Stopp-Prozent": [prob3], "Stopp nach X Excluded": [stop_after_x], "Letzte gefundene Studie bei": [pos_last_inc], "Stop Criterion 1": [crit1], "Stop Criterion 2 an Position": [crit2_pos], "Stop Criterion 2 ausgeschlossene Included": [crit2], "Stop richtig?": [result], } # Store positions of all relevant studies as additional columns for i, pos in enumerate(all_ones, start=1): data_dict[i] = pos pd.DataFrame(data_dict).to_csv(output_csv, mode="a", header=False, index=False) delete_datafiles(n) clear_output() print(f"Done. Results saved to {output_csv}")