miscellaneous stats.py (3662B)
1 import argparse 2 import collections 3 import pathlib 4 import re 5 6 import datasets 7 import numpy 8 import matplotlib.pyplot as plt 9 10 from datetime import datetime 11 12 import numpy as np 13 14 15 def reporting_delay_analysis(dataset): 16 print("### Analyze differences between publication dates and event start dates") 17 18 # Compute difference in days 19 diffs = [(sample["source_date"].date() - sample["start_date"]).days for sample in dataset] 20 diffs_array = np.array(diffs) 21 22 avg_diff = diffs_array.mean() 23 print(f"Average source_date - start_date difference: {avg_diff:.2f} days") 24 25 buckets = { 26 "* No reporting delay: 0 days": (diffs_array == 0), 27 "* Reporting delay of up to 1 day": (diffs_array == 1), 28 "* Reporting delay of up to 7 days": (diffs_array == 7), 29 "* Reporting delay of up to 30 days": (diffs_array == 30), 30 "* Reporting delay of more than 30 days": (diffs_array > 30) 31 } 32 33 total = len(diffs_array) 34 for label, mask in buckets.items(): 35 percent = mask.mean() * 100 36 print(f"{label}: {percent:.1f}%") 37 38 # Plot (remove if annoying) 39 bucket_counts = [mask.sum() for mask in buckets.values()] 40 labels = list(buckets.keys()) 41 42 plt.figure(figsize=(8, 5)) 43 plt.bar(labels, bucket_counts, color='skyblue') 44 plt.ylabel("Count") 45 plt.title("Distribution of (source_date - start_date) in days") 46 plt.grid(axis='y', linestyle='--', alpha=0.6) 47 plt.tight_layout() 48 plt.show() 49 50 51 def main(dataset_path: pathlib.Path) -> None: 52 splits: dict[str, datasets.Dataset] = { 53 split_path.name: datasets.load_from_disk(str(split_path)) 54 for split_path in dataset_path.iterdir() 55 if split_path.is_dir() 56 } 57 full = datasets.concatenate_datasets(splits.values()) 58 59 # Actors 60 one_sided = sum(sample["side_b_name"] == "Civilians" for sample in full) / len(full) 61 print(f"Percent of one-sided (B=civilians): {one_sided*100:.1f}%") 62 63 # Dates 64 tenyears = sum(sample["start_date"].year >= 2015 for sample in full) / len(full) 65 print(f"Percent in last 10 years: {tenyears*100:.1f}%") 66 diffdate = numpy.array([(sample["end_date"]-sample["start_date"]).days for sample in full]) 67 print(f"Percent of exact day: {(diffdate==0).mean()*100:.1f}%") 68 print(f"Percent of non-exact day: {(diffdate!=0).mean()*100:.1f}%") 69 print(f"Percent with spreak of more than a week: {(diffdate>7).mean()*100:.1f}%") 70 reporting_delay_analysis(full) 71 72 # Countries 73 # TODO: check different countries in dataset 74 75 # Locations 76 num_countries = len(set(sample["location_root_name"] for sample in full)) 77 print(f"Number of countries in all splits: {num_countries}") 78 79 # Deaths 80 side_a = numpy.array(full["deaths_side_a"]) 81 side_b = numpy.array(full["deaths_side_b"]) 82 civilians = numpy.array(full["deaths_civilian"]) 83 unknown = numpy.array(full["deaths_unknown"]) 84 low = numpy.array(full["deaths_low"]) 85 high = numpy.array(full["deaths_high"]) 86 best = side_a + side_b + civilians + unknown 87 print(f"Percent of exact deaths: {(low==high).mean()*100:.1f}%") 88 print(f"Percent of uncertain (best=0): {(best==0).mean()*100:.1f}%") 89 90 91 def parse_command_argument() -> argparse.Namespace: 92 """ Parse command line arguments. """ 93 parser = argparse.ArgumentParser(description="Generate some miscellaneous statistics.") 94 parser.add_argument("dataset", type=pathlib.Path, help="Path to the target dataset splits.") 95 96 args: argparse.Namespace = parser.parse_args() 97 return args 98 99 100 if __name__ == "__main__": 101 args: argparse.Namespace = parse_command_argument() 102 main(dataset_path=args.dataset)