"""Statistical utility functions for data analysis.""" from collections import defaultdict, Counter from typing import List, Dict, Any, Union import logging logger = logging.getLogger(__name__) def aggregate_counters(counters: List[Dict[str, int]]) -> Dict[str, int]: """Aggregate multiple counter dictionaries. Args: counters: List of counter dictionaries Returns: Aggregated counter dictionary """ total = defaultdict(int) for counter in counters: for key, value in counter.items(): total[key] += value return dict(total) def merge_distributions(distributions: List[List[float]]) -> List[float]: """Merge multiple distribution lists into one. Args: distributions: List of value lists Returns: Merged flat list of all values """ merged = [] for dist in distributions: merged.extend(dist) return merged def calculate_percentiles( values: List[float], percentiles: List[float] = [25, 50, 75, 90, 95, 99] ) -> Dict[str, float]: """Calculate percentiles for a list of values. Args: values: List of numeric values percentiles: List of percentiles to calculate (0-100) Returns: Dictionary mapping percentile names to values """ if not values: return {} sorted_values = sorted(values) n = len(sorted_values) result = {} for p in percentiles: idx = int(n * p / 100) idx = min(idx, n - 1) result[f"p{p}"] = sorted_values[idx] return result def merge_value_lists(value_lists: List[Dict[str, List]]) -> Dict[str, List]: """Merge dictionaries of value lists. Args: value_lists: List of dictionaries mapping keys to value lists Returns: Merged dictionary with concatenated value lists """ merged = defaultdict(list) for vl in value_lists: for key, values in vl.items(): merged[key].extend(values) return dict(merged) def analyze_numeric_values(values: List[Union[int, float]]) -> Dict[str, Any]: """Analyze numeric values and return statistics. Args: values: List of numeric values Returns: Dictionary with statistical analysis """ if not values: return {"count": 0, "exists": False} numeric_values = [] for v in values: try: numeric_values.append(float(v)) except (ValueError, TypeError): continue if not numeric_values: return {"count": len(values), "exists": True, "all_non_numeric": True} sorted_vals = sorted(numeric_values) n = len(sorted_vals) stats = { "count": n, "exists": True, "min": sorted_vals[0], "max": sorted_vals[-1], "mean": sum(sorted_vals) / n, "median": sorted_vals[n // 2], } # Add percentiles if enough data if n >= 4: stats["p25"] = sorted_vals[n // 4] stats["p75"] = sorted_vals[3 * n // 4] if n >= 20: stats["p5"] = sorted_vals[n // 20] stats["p95"] = sorted_vals[19 * n // 20] if n >= 100: stats["p1"] = sorted_vals[n // 100] stats["p99"] = sorted_vals[99 * n // 100] return stats def analyze_string_values(values: List[str], top_n: int = 20) -> Dict[str, Any]: """Analyze string values and return frequency statistics. Args: values: List of string values top_n: Number of top values to return Returns: Dictionary with string analysis """ if not values: return {"count": 0, "unique_count": 0, "top_values": []} counter = Counter(values) return { "count": len(values), "unique_count": len(counter), "top_values": counter.most_common(top_n), "singleton_count": sum(1 for count in counter.values() if count == 1), } def merge_id_sets(id_sets: List[set]) -> set: """Merge multiple ID sets. Args: id_sets: List of ID sets Returns: Merged set of all IDs """ merged = set() for id_set in id_sets: merged.update(id_set) return merged def calculate_overlap(set1: set, set2: set) -> Dict[str, Any]: """Calculate overlap statistics between two sets. Args: set1: First set set2: Second set Returns: Dictionary with overlap statistics """ intersection = set1 & set2 union = set1 | set2 return { "set1_size": len(set1), "set2_size": len(set2), "intersection_size": len(intersection), "union_size": len(union), "set1_only": len(set1 - set2), "set2_only": len(set2 - set1), "jaccard_similarity": len(intersection) / len(union) if union else 0, "overlap_ratio_set1": len(intersection) / len(set1) if set1 else 0, "overlap_ratio_set2": len(intersection) / len(set2) if set2 else 0, }