Coverage for tinytroupe / validation / simulation_validator.py: 0%
986 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-02-28 17:48 +0000
1"""
2Simulation experiment empirical validation mechanisms for TinyTroupe.
4This module provides tools to validate simulation experiment results against empirical control data,
5supporting both statistical hypothesis testing and semantic validation approaches.
6This is distinct from LLM-based evaluations, focusing on data-driven validation
7against known empirical benchmarks.
8"""
10from typing import Dict, List, Optional, Union, Any
11import json
12import csv
13from datetime import datetime
14from pathlib import Path
15from pydantic import BaseModel, Field
17import pandas as pd
19from tinytroupe.experimentation.statistical_tests import StatisticalTester
20from tinytroupe.utils.semantics import compute_semantic_proximity
22# TODO Work-in-Progress below
24class SimulationExperimentDataset(BaseModel):
25 """
26 Represents a dataset from a simulation experiment or empirical study.
28 This contains data that can be used for validation, including quantitative metrics
29 and qualitative agent justifications from simulation experiments or empirical studies.
31 Supports both numeric and categorical data. Categorical data (strings) is automatically
32 converted to ordinal values for statistical analysis while preserving the original
33 categories for interpretation.
35 Attributes:
36 name: Optional name for the dataset
37 description: Optional description of the dataset
38 key_results: Map from result names to their values (numbers, proportions, booleans, strings, etc.)
39 result_types: Map indicating whether each result is "aggregate" or "per_agent"
40 data_types: Map indicating the data type for each result ("numeric", "categorical", "ordinal", "ranking", "count", "proportion", "binary")
41 categorical_mappings: Internal mappings from categorical strings to ordinal values
42 ordinal_mappings: Internal mappings for ordinal data with explicit ordering
43 ranking_info: Information about ranking data (items being ranked, ranking direction)
44 agent_names: Optional list of agent names (can be referenced by index in results)
45 agent_justifications: List of justifications (with optional agent references)
46 justification_summary: Optional summary of all agent justifications
47 agent_attributes: Agent attributes for manual inspection only (not used in statistical comparisons)
48 """
49 name: Optional[str] = None
50 description: Optional[str] = None
51 key_results: Dict[str, Union[float, int, bool, str, List[Union[float, int, bool, str, None]], None]] = Field(default_factory=dict)
52 result_types: Dict[str, str] = Field(default_factory=dict, description="Map from result name to 'aggregate' or 'per_agent'")
53 data_types: Dict[str, str] = Field(default_factory=dict, description="Map indicating data type: 'numeric', 'categorical', 'ordinal', 'ranking', 'count', 'proportion', 'binary'")
54 categorical_mappings: Dict[str, Dict[str, int]] = Field(default_factory=dict, description="Internal mappings from categorical strings to ordinal values")
55 ordinal_mappings: Dict[str, Dict[str, int]] = Field(default_factory=dict, description="Internal mappings for ordinal data with explicit ordering")
56 ranking_info: Dict[str, Dict[str, Any]] = Field(default_factory=dict, description="Information about ranking data (items, direction, etc.)")
57 agent_names: Optional[List[Optional[str]]] = Field(None, description="Optional list of agent names for reference (can contain None for unnamed agents)")
58 agent_justifications: List[Union[str, Dict[str, Union[str, int]]]] = Field(
59 default_factory=list,
60 description="List of justifications as strings or dicts with optional 'agent_name'/'agent_index' and 'justification'"
61 )
62 justification_summary: Optional[str] = None
63 agent_attributes: Dict[str, List[Union[str, None]]] = Field(
64 default_factory=dict,
65 description="Agent attributes loaded from CSV but not used in statistical comparisons (e.g., age, gender, etc.)"
66 )
68 class Config:
69 """Pydantic configuration."""
70 extra = "forbid" # Prevent accidental extra fields
71 validate_assignment = True # Validate on assignment after creation
73 def __init__(self, **data):
74 """Initialize with automatic data processing."""
75 super().__init__(**data)
76 self._process_data_types()
78 def _process_data_types(self):
79 """
80 Process different data types and convert them appropriately.
82 Automatically detects and processes:
83 - Categorical data (strings) -> ordinal mapping
84 - Ordinal data (explicit ordering) -> validation of ordering
85 - Ranking data (ranks/positions) -> validation and normalization
86 - Count data (non-negative integers) -> validation
87 - Proportion data (0-1 or 0-100) -> normalization to 0-1
88 - Binary data (boolean/yes-no) -> conversion to 0/1
89 """
90 for metric_name, metric_data in self.key_results.items():
91 data_type = self.data_types.get(metric_name, "auto")
93 if data_type == "auto":
94 # Auto-detect data type
95 data_type = self._detect_data_type(metric_data)
96 self.data_types[metric_name] = data_type
98 # Process based on data type
99 if data_type == "categorical":
100 self._process_categorical_data_for_metric(metric_name, metric_data)
101 elif data_type == "ordinal":
102 self._process_ordinal_data_for_metric(metric_name, metric_data)
103 elif data_type == "ranking":
104 self._process_ranking_data_for_metric(metric_name, metric_data)
105 elif data_type == "count":
106 self._validate_count_data_for_metric(metric_name, metric_data)
107 elif data_type == "proportion":
108 self._process_proportion_data_for_metric(metric_name, metric_data)
109 elif data_type == "binary":
110 self._process_binary_data_for_metric(metric_name, metric_data)
111 # "numeric" requires no special processing
113 def _detect_data_type(self, data: Union[float, int, bool, str, List, None]) -> str:
114 """Auto-detect the data type based on the data content."""
115 if data is None:
116 return "numeric" # Default fallback
118 # Handle single values
119 if not isinstance(data, list):
120 data = [data]
122 # Filter out None values for analysis
123 valid_data = [item for item in data if item is not None]
124 if not valid_data:
125 return "numeric" # Default fallback
127 # Check for string data (categorical) - but only if ALL non-None values are strings
128 string_count = sum(1 for item in valid_data if isinstance(item, str))
129 if string_count > 0:
130 # If we have mixed types (strings + numbers), default to categorical for simplicity
131 # since the string conversion will handle the mixed case
132 return "categorical"
134 # Check for boolean data
135 if all(isinstance(item, bool) for item in valid_data):
136 return "binary"
138 # Check for numeric data
139 numeric_data = [item for item in valid_data if isinstance(item, (int, float))]
140 if len(numeric_data) != len(valid_data):
141 return "numeric" # Mixed types, default to numeric
143 # Check for count data (non-negative integers, including whole number floats)
144 def is_whole_number(x):
145 """Check if a number is a whole number (either int or float with no decimal part)."""
146 return isinstance(x, int) or (isinstance(x, float) and x.is_integer())
148 if all(is_whole_number(item) and item >= 0 for item in numeric_data):
149 # Convert floats to ints for ranking detection
150 int_data = [int(item) for item in numeric_data]
152 # For ranking detection, be more strict:
153 # 1. Must have at least 3 data points
154 # 2. Must have consecutive integers starting from 1
155 # 3. Must have some repetition (indicating actual rankings rather than just sequence)
156 sorted_data = sorted(set(int_data))
157 min_val = min(sorted_data)
158 max_val = max(sorted_data)
160 # Only consider as ranking if:
161 # - Starts from 1
162 # - Has at least 2 different rank values
163 # - Is consecutive (no gaps)
164 # - Has repetition (more data points than unique values) - this is key for rankings
165 if (len(int_data) >= 3 and # At least 3 data points
166 min_val == 1 and # Starts from 1
167 len(sorted_data) >= 2 and # At least 2 different ranks
168 max_val <= 10 and # Reasonable upper limit for rankings
169 sorted_data == list(range(1, max_val + 1)) and # Consecutive
170 len(int_data) > len(sorted_data)): # Has repetition (essential for rankings)
171 return "ranking"
173 # Otherwise, it's count data
174 return "count"
176 # Check for proportion data (0-1 range) - only for floats
177 if all(isinstance(item, (int, float)) and 0 <= item <= 1 for item in numeric_data):
178 # If all values are 0 or 1 integers, it's likely binary
179 if all(isinstance(item, int) and item in [0, 1] for item in numeric_data):
180 return "binary"
181 return "proportion"
183 # Default to numeric
184 return "numeric"
186 def _process_categorical_data_for_metric(self, metric_name: str, metric_data):
187 """Process categorical data for a specific metric."""
188 if self._is_categorical_data(metric_data):
189 # Extract all unique categories
190 categories = self._extract_categories(metric_data)
192 if categories:
193 # Create sorted categorical mapping for consistency
194 sorted_categories = sorted(categories)
195 categorical_mapping = {category: idx for idx, category in enumerate(sorted_categories)}
196 self.categorical_mappings[metric_name] = categorical_mapping
198 # Convert string data to ordinal values
199 self.key_results[metric_name] = self._convert_to_ordinal(metric_data, categorical_mapping)
201 def _process_ordinal_data_for_metric(self, metric_name: str, metric_data):
202 """Process ordinal data for a specific metric."""
203 # For ordinal data, we expect either:
204 # 1. Numeric values that represent ordinal levels (e.g., 1, 2, 3, 4, 5 for Likert)
205 # 2. String values that need explicit ordering (e.g., "Poor", "Fair", "Good", "Excellent")
207 if self._is_categorical_data(metric_data):
208 # String ordinal data - need explicit ordering
209 categories = self._extract_categories(metric_data)
210 if categories:
211 # For string ordinal data, we need to define a meaningful order
212 # This could be enhanced to accept explicit ordering from user
213 sorted_categories = self._order_ordinal_categories(list(categories))
214 ordinal_mapping = {category: idx for idx, category in enumerate(sorted_categories)}
215 self.ordinal_mappings[metric_name] = ordinal_mapping
217 # Convert to ordinal values
218 self.key_results[metric_name] = self._convert_to_ordinal(metric_data, ordinal_mapping)
219 else:
220 # Numeric ordinal data - validate that values are reasonable
221 self._validate_ordinal_numeric_data(metric_name, metric_data)
223 def _process_ranking_data_for_metric(self, metric_name: str, metric_data):
224 """Process ranking data for a specific metric."""
225 # Ranking data should be integers representing positions (1, 2, 3, etc.)
226 valid_data = self._get_valid_numeric_data(metric_data)
228 if valid_data:
229 unique_ranks = sorted(set(valid_data))
230 min_rank = min(unique_ranks)
231 max_rank = max(unique_ranks)
233 # Check if ranking_info already exists (e.g., from ordinal processing)
234 existing_info = self.ranking_info.get(metric_name, {})
236 # Store ranking information, preserving existing keys
237 ranking_info = {
238 "min_rank": min_rank,
239 "max_rank": max_rank,
240 "num_ranks": len(unique_ranks),
241 "rank_values": unique_ranks,
242 "direction": existing_info.get("direction", "ascending") # Preserve existing direction or default
243 }
245 # Preserve any additional keys from existing ranking info (e.g., ordinal-specific data)
246 ranking_info.update({k: v for k, v in existing_info.items()
247 if k not in ranking_info})
249 self.ranking_info[metric_name] = ranking_info
251 # Validate ranking data
252 self._validate_ranking_data(metric_name, metric_data)
254 def _process_proportion_data_for_metric(self, metric_name: str, metric_data):
255 """Process proportion data for a specific metric."""
256 # Normalize proportion data to 0-1 range if needed
257 if isinstance(metric_data, list):
258 normalized_data = []
259 for item in metric_data:
260 if item is None:
261 normalized_data.append(None)
262 elif isinstance(item, (int, float)):
263 # If value > 1, assume it's percentage (0-100), convert to proportion
264 normalized_data.append(item / 100.0 if item > 1 else item)
265 else:
266 normalized_data.append(item) # Keep as-is
267 self.key_results[metric_name] = normalized_data
268 elif isinstance(metric_data, (int, float)) and metric_data > 1:
269 # Single percentage value
270 self.key_results[metric_name] = metric_data / 100.0
272 def _process_binary_data_for_metric(self, metric_name: str, metric_data):
273 """Process binary data for a specific metric."""
274 # Convert boolean/string binary data to 0/1
275 if isinstance(metric_data, list):
276 binary_data = []
277 for item in metric_data:
278 if item is None:
279 binary_data.append(None)
280 else:
281 binary_data.append(self._convert_to_binary(item))
282 self.key_results[metric_name] = binary_data
283 elif metric_data is not None:
284 self.key_results[metric_name] = self._convert_to_binary(metric_data)
286 def _validate_count_data_for_metric(self, metric_name: str, metric_data):
287 """Validate count data for a specific metric."""
288 valid_data = self._get_valid_numeric_data(metric_data)
290 # Check that all values are non-negative integers (including whole number floats)
291 for value in valid_data:
292 # Accept both integers and whole number floats
293 is_whole_number = isinstance(value, int) or (isinstance(value, float) and value.is_integer())
294 if not is_whole_number or value < 0:
295 raise ValueError(f"Count data for metric '{metric_name}' must be non-negative integers, found: {value}")
297 def _order_ordinal_categories(self, categories: List[str]) -> List[str]:
298 """Order ordinal categories in a meaningful way."""
299 # Common ordinal patterns for automatic ordering
300 likert_patterns = {
301 "strongly disagree": 1, "disagree": 2, "neutral": 3, "agree": 4, "strongly agree": 5,
302 "very poor": 1, "poor": 2, "fair": 3, "good": 4, "very good": 5, "excellent": 6,
303 "never": 1, "rarely": 2, "sometimes": 3, "often": 4, "always": 5,
304 "very low": 1, "low": 2, "medium": 3, "high": 4, "very high": 5,
305 "terrible": 1, "bad": 2, "okay": 3, "good": 4, "great": 5, "amazing": 6
306 }
308 # Try to match patterns
309 category_scores = {}
310 for category in categories:
311 normalized_cat = self._normalize_category(category)
312 if normalized_cat in likert_patterns:
313 category_scores[category] = likert_patterns[normalized_cat]
315 # If we found matches for all categories, use that ordering
316 if len(category_scores) == len(categories):
317 return sorted(categories, key=lambda x: category_scores[x])
319 # Otherwise, fall back to alphabetical ordering with a warning
320 return sorted(categories)
322 def _validate_ordinal_numeric_data(self, metric_name: str, metric_data):
323 """Validate numeric ordinal data."""
324 valid_data = self._get_valid_numeric_data(metric_data)
326 if valid_data:
327 unique_values = sorted(set(valid_data))
328 # Check if values are reasonable for ordinal data (consecutive or at least ordered)
329 if len(unique_values) < 2:
330 return # Single value is fine
332 # Store ordinal information
333 self.ordinal_mappings[metric_name] = {
334 "min_value": min(unique_values),
335 "max_value": max(unique_values),
336 "unique_values": unique_values,
337 "num_levels": len(unique_values)
338 }
340 def _validate_ranking_data(self, metric_name: str, metric_data):
341 """Validate ranking data structure."""
342 valid_data = self._get_valid_numeric_data(metric_data)
344 if not valid_data:
345 return
347 unique_ranks = set(valid_data)
348 min_rank = min(unique_ranks)
349 max_rank = max(unique_ranks)
351 # Check for reasonable ranking structure
352 if min_rank < 1:
353 raise ValueError(f"Ranking data for metric '{metric_name}' should start from 1, found minimum: {min_rank}")
355 # Check for gaps in ranking (warning, not error)
356 expected_ranks = set(range(min_rank, max_rank + 1))
357 missing_ranks = expected_ranks - unique_ranks
358 if missing_ranks:
359 # This is often okay in ranking data (tied ranks, incomplete rankings)
360 pass
362 def _get_valid_numeric_data(self, data) -> List[Union[int, float]]:
363 """Get valid numeric data from a metric, handling both single values and lists."""
364 if data is None:
365 return []
367 if not isinstance(data, list):
368 data = [data]
370 return [item for item in data if item is not None and isinstance(item, (int, float))]
372 def _convert_to_binary(self, value) -> int:
373 """Convert various binary representations to 0 or 1."""
374 if isinstance(value, bool):
375 return 1 if value else 0
376 elif isinstance(value, str):
377 normalized = value.lower().strip()
378 true_values = {"true", "yes", "y", "1", "on", "success", "positive"}
379 false_values = {"false", "no", "n", "0", "off", "failure", "negative"}
381 if normalized in true_values:
382 return 1
383 elif normalized in false_values:
384 return 0
385 else:
386 raise ValueError(f"Cannot convert string '{value}' to binary")
387 elif isinstance(value, (int, float)):
388 return 1 if value != 0 else 0
389 else:
390 raise ValueError(f"Cannot convert {type(value)} to binary")
392 def _process_categorical_data(self):
393 """
394 Legacy method for backward compatibility.
395 Process categorical string data by converting to ordinal values.
396 """
397 for metric_name, metric_data in self.key_results.items():
398 if metric_name not in self.data_types: # Only process if data type not explicitly set
399 if self._is_categorical_data(metric_data):
400 self.data_types[metric_name] = "categorical"
401 self._process_categorical_data_for_metric(metric_name, metric_data)
403 def _is_categorical_data(self, data: Union[float, int, bool, str, List, None]) -> bool:
404 """Check if data contains categorical (string) values."""
405 if isinstance(data, str):
406 return True
407 elif isinstance(data, list):
408 return any(isinstance(item, str) for item in data if item is not None)
409 return False
411 def _extract_categories(self, data: Union[float, int, bool, str, List, None]) -> set:
412 """Extract unique string categories from data."""
413 categories = set()
415 if isinstance(data, str):
416 categories.add(self._normalize_category(data))
417 elif isinstance(data, list):
418 for item in data:
419 if isinstance(item, str):
420 categories.add(self._normalize_category(item))
422 return categories
424 def _normalize_category(self, category: str) -> str:
425 """Normalize categorical string (lowercase, strip whitespace)."""
426 return category.lower().strip()
428 def _convert_to_ordinal(self, data: Union[str, List], mapping: Dict[str, int]) -> Union[int, List[Union[int, None]]]:
429 """Convert categorical data to ordinal values using the mapping."""
430 if isinstance(data, str):
431 normalized = self._normalize_category(data)
432 return mapping.get(normalized, 0) # Default to 0 if not found
433 elif isinstance(data, list):
434 converted = []
435 for item in data:
436 if isinstance(item, str):
437 normalized = self._normalize_category(item)
438 converted.append(mapping.get(normalized, 0))
439 elif item is None:
440 converted.append(None) # Preserve None values
441 else:
442 converted.append(item) # Keep numeric values as-is
443 return converted
444 else:
445 return data
447 def get_agent_name(self, index: int) -> Optional[str]:
448 """Get agent name by index, if available."""
449 if self.agent_names and 0 <= index < len(self.agent_names):
450 agent_name = self.agent_names[index]
451 return agent_name if agent_name is not None else None
452 return None
454 def get_agent_data(self, metric_name: str, agent_index: int) -> Optional[Union[float, int, bool]]:
455 """Get a specific agent's data for a given metric. Returns None for missing data."""
456 if metric_name not in self.key_results:
457 return None
459 metric_data = self.key_results[metric_name]
461 # Check if it's per-agent data
462 if self.result_types.get(metric_name) == "per_agent" and isinstance(metric_data, list):
463 if 0 <= agent_index < len(metric_data):
464 return metric_data[agent_index] # This can be None for missing data
466 return None
468 def get_all_agent_data(self, metric_name: str) -> Dict[str, Union[float, int, bool]]:
469 """Get all agents' data for a given metric as a dictionary mapping agent names/indices to values."""
470 if metric_name not in self.key_results:
471 return {}
473 metric_data = self.key_results[metric_name]
474 result = {}
476 # For per-agent data, create mapping
477 if self.result_types.get(metric_name) == "per_agent" and isinstance(metric_data, list):
478 for i, value in enumerate(metric_data):
479 agent_name = self.get_agent_name(i) or f"Agent_{i}"
480 # Only include non-None values in the result
481 if value is not None:
482 result[agent_name] = value
484 # For aggregate data, return single value
485 elif self.result_types.get(metric_name) == "aggregate":
486 result["aggregate"] = metric_data
488 return result
490 def get_valid_agent_data(self, metric_name: str) -> List[Union[float, int, bool]]:
491 """Get only valid (non-None) values for a per-agent metric."""
492 if metric_name not in self.key_results:
493 return []
495 metric_data = self.key_results[metric_name]
497 if self.result_types.get(metric_name) == "per_agent" and isinstance(metric_data, list):
498 return [value for value in metric_data if value is not None]
500 return []
502 def validate_data_consistency(self) -> List[str]:
503 """Validate that per-agent data is consistent across metrics and with agent names."""
504 errors = []
505 warnings = []
507 # Check per-agent metrics have consistent lengths
508 per_agent_lengths = []
509 per_agent_metrics = []
511 for metric_name, result_type in self.result_types.items():
512 if result_type == "per_agent" and metric_name in self.key_results:
513 metric_data = self.key_results[metric_name]
514 if isinstance(metric_data, list):
515 per_agent_lengths.append(len(metric_data))
516 per_agent_metrics.append(metric_name)
517 else:
518 errors.append(f"Metric '{metric_name}' marked as per_agent but is not a list")
520 # Check all per-agent metrics have same length
521 if per_agent_lengths and len(set(per_agent_lengths)) > 1:
522 errors.append(f"Per-agent metrics have inconsistent lengths: {dict(zip(per_agent_metrics, per_agent_lengths))}")
524 # Check agent_names length matches per-agent data length
525 if self.agent_names and per_agent_lengths:
526 agent_count = len(self.agent_names)
527 data_length = per_agent_lengths[0] if per_agent_lengths else 0
528 if agent_count != data_length:
529 errors.append(f"agent_names length ({agent_count}) doesn't match per-agent data length ({data_length})")
531 # Check for None values in agent_names and provide warnings
532 if self.agent_names:
533 none_indices = [i for i, name in enumerate(self.agent_names) if name is None]
534 if none_indices:
535 warnings.append(f"agent_names contains None values at indices: {none_indices}")
537 # Check for None values in per-agent data and provide info
538 for metric_name in per_agent_metrics:
539 if metric_name in self.key_results:
540 metric_data = self.key_results[metric_name]
541 none_indices = [i for i, value in enumerate(metric_data) if value is None]
542 if none_indices:
543 warnings.append(f"Metric '{metric_name}' has missing data (None) at indices: {none_indices}")
545 # Return errors and warnings combined
546 return errors + [f"WARNING: {warning}" for warning in warnings]
548 def get_justification_text(self, justification_item: Union[str, Dict[str, Union[str, int]]]) -> str:
549 """Extract justification text from various formats."""
550 if isinstance(justification_item, str):
551 return justification_item
552 elif isinstance(justification_item, dict):
553 return justification_item.get("justification", "")
554 return ""
556 def get_justification_agent_reference(self, justification_item: Union[str, Dict[str, Union[str, int]]]) -> Optional[str]:
557 """Get agent reference from justification, returning name if available."""
558 if isinstance(justification_item, dict):
559 # Direct agent name
560 if "agent_name" in justification_item:
561 return justification_item["agent_name"]
562 # Agent index reference
563 elif "agent_index" in justification_item:
564 return self.get_agent_name(justification_item["agent_index"])
565 return None
567 def get_categorical_values(self, metric_name: str) -> Optional[List[str]]:
568 """Get the original categorical values for a metric, if it was categorical."""
569 if metric_name in self.categorical_mappings:
570 # Return categories sorted by their ordinal values
571 mapping = self.categorical_mappings[metric_name]
572 return [category for category, _ in sorted(mapping.items(), key=lambda x: x[1])]
573 elif metric_name in self.ordinal_mappings and isinstance(self.ordinal_mappings[metric_name], dict):
574 # Handle string-based ordinal data
575 mapping = self.ordinal_mappings[metric_name]
576 if all(isinstance(k, str) for k in mapping.keys()):
577 return [category for category, _ in sorted(mapping.items(), key=lambda x: x[1])]
578 return None
580 def convert_ordinal_to_categorical(self, metric_name: str, ordinal_value: Union[int, float]) -> Optional[str]:
581 """Convert an ordinal value back to its original categorical string."""
582 # Check categorical mappings first
583 if metric_name in self.categorical_mappings:
584 mapping = self.categorical_mappings[metric_name]
585 # Reverse lookup: find category with this ordinal value
586 for category, value in mapping.items():
587 if value == int(ordinal_value):
588 return category
590 # Check ordinal mappings for string-based ordinal data
591 elif metric_name in self.ordinal_mappings:
592 mapping = self.ordinal_mappings[metric_name]
593 if isinstance(mapping, dict) and all(isinstance(k, str) for k in mapping.keys()):
594 for category, value in mapping.items():
595 if value == int(ordinal_value):
596 return category
598 return None
600 def get_data_type_info(self, metric_name: str) -> Dict[str, Any]:
601 """Get comprehensive information about a metric's data type."""
602 data_type = self.data_types.get(metric_name, "numeric")
603 info = {
604 "data_type": data_type,
605 "result_type": self.result_types.get(metric_name, "unknown")
606 }
608 if data_type == "categorical" and metric_name in self.categorical_mappings:
609 info["categories"] = self.get_categorical_values(metric_name)
610 info["category_mapping"] = self.categorical_mappings[metric_name].copy()
612 elif data_type == "ordinal":
613 if metric_name in self.ordinal_mappings:
614 mapping = self.ordinal_mappings[metric_name]
615 if isinstance(mapping, dict):
616 # Check if this is a string-to-number mapping (categorical ordinal)
617 # vs info dict (numeric ordinal)
618 if "min_value" in mapping or "max_value" in mapping:
619 # Numeric ordinal info
620 info["ordinal_info"] = mapping.copy()
621 elif all(isinstance(k, str) for k in mapping.keys()) and all(isinstance(v, int) for v in mapping.values()):
622 # String-based ordinal - safely sort by values
623 try:
624 info["ordinal_categories"] = [cat for cat, _ in sorted(mapping.items(), key=lambda x: x[1])]
625 info["ordinal_mapping"] = mapping.copy()
626 except TypeError:
627 # Fallback if sorting fails
628 info["ordinal_categories"] = list(mapping.keys())
629 info["ordinal_mapping"] = mapping.copy()
630 else:
631 # Unknown ordinal format, treat as info
632 info["ordinal_info"] = mapping.copy()
634 elif data_type == "ranking" and metric_name in self.ranking_info:
635 info["ranking_info"] = self.ranking_info[metric_name].copy()
637 return info
639 def get_metric_summary(self, metric_name: str) -> Dict[str, Any]:
640 """Get a comprehensive summary of a metric including data type information."""
641 summary = {
642 "metric_name": metric_name,
643 "result_type": self.result_types.get(metric_name, "unknown"),
644 "data_type": self.data_types.get(metric_name, "numeric"),
645 }
647 # Add legacy categorical flag for backward compatibility
648 summary["is_categorical"] = (metric_name in self.categorical_mappings or
649 (metric_name in self.ordinal_mappings and
650 isinstance(self.ordinal_mappings[metric_name], dict) and
651 all(isinstance(k, str) for k in self.ordinal_mappings[metric_name].keys())))
653 if metric_name in self.key_results:
654 data = self.key_results[metric_name]
655 summary["data_type_name"] = type(data).__name__
657 if isinstance(data, list):
658 valid_data = [x for x in data if x is not None]
659 summary["total_values"] = len(data)
660 summary["valid_values"] = len(valid_data)
661 summary["missing_values"] = len(data) - len(valid_data)
663 if valid_data:
664 summary["min_value"] = min(valid_data)
665 summary["max_value"] = max(valid_data)
667 # Add data type specific information
668 data_type_info = self.get_data_type_info(metric_name)
669 summary.update(data_type_info)
671 # Add distribution information for per-agent data
672 if isinstance(data, list) and self.result_types.get(metric_name) == "per_agent":
673 data_type = summary["data_type"]
675 if data_type in ["categorical", "ordinal"] and summary.get("is_categorical"):
676 # Category distribution
677 category_counts = {}
678 for value in data:
679 if value is not None:
680 category = self.convert_ordinal_to_categorical(metric_name, value)
681 if category:
682 category_counts[category] = category_counts.get(category, 0) + 1
683 summary["category_distribution"] = category_counts
685 elif data_type == "ranking":
686 # Ranking distribution
687 rank_counts = {}
688 for value in data:
689 if value is not None:
690 rank_counts[value] = rank_counts.get(value, 0) + 1
691 summary["rank_distribution"] = rank_counts
693 elif data_type == "binary":
694 # Binary distribution
695 true_count = sum(1 for x in data if x == 1)
696 false_count = sum(1 for x in data if x == 0)
697 summary["binary_distribution"] = {"true": true_count, "false": false_count}
699 return summary
701 def is_categorical_metric(self, metric_name: str) -> bool:
702 """Check if a metric contains categorical data (including string-based ordinal)."""
703 return (metric_name in self.categorical_mappings or
704 (metric_name in self.ordinal_mappings and
705 isinstance(self.ordinal_mappings[metric_name], dict) and
706 all(isinstance(k, str) for k in self.ordinal_mappings[metric_name].keys())))
709class SimulationExperimentEmpiricalValidationResult(BaseModel):
710 """
711 Contains the results of a simulation experiment validation against empirical data.
713 This represents the outcome of validating simulation experiment data
714 against empirical benchmarks, using statistical and semantic methods.
716 Attributes:
717 validation_type: Type of validation performed
718 control_name: Name of the control/empirical dataset
719 treatment_name: Name of the treatment/simulation experiment dataset
720 statistical_results: Results from statistical tests (if performed)
721 semantic_results: Results from semantic proximity analysis (if performed)
722 overall_score: Overall validation score (0.0 to 1.0)
723 summary: Summary of validation findings
724 timestamp: When the validation was performed
725 """
726 validation_type: str
727 control_name: str
728 treatment_name: str
729 statistical_results: Optional[Dict[str, Any]] = None
730 semantic_results: Optional[Dict[str, Any]] = None
731 overall_score: Optional[float] = Field(None, ge=0.0, le=1.0, description="Overall validation score between 0.0 and 1.0")
732 summary: str = ""
733 timestamp: str = Field(default_factory=lambda: datetime.now().isoformat())
735 class Config:
736 """Pydantic configuration."""
737 extra = "forbid"
738 validate_assignment = True
741class SimulationExperimentEmpiricalValidator:
742 """
743 A validator for comparing simulation experiment data against empirical control data.
745 This validator performs data-driven validation using statistical hypothesis testing
746 and semantic proximity analysis of agent justifications. It is designed to validate
747 simulation experiment results against known empirical benchmarks, distinct from LLM-based evaluations.
748 """
750 def __init__(self):
751 """Initialize the simulation experiment empirical validator."""
752 pass
754 def validate(self,
755 control: SimulationExperimentDataset,
756 treatment: SimulationExperimentDataset,
757 validation_types: List[str] = ["statistical", "semantic"],
758 statistical_test_type: str = "welch_t_test",
759 significance_level: float = 0.05,
760 output_format: str = "values") -> Union[SimulationExperimentEmpiricalValidationResult, str]:
761 """
762 Validate a simulation experiment dataset against an empirical control dataset.
764 Args:
765 control: The control/empirical reference dataset
766 treatment: The treatment/simulation experiment dataset to validate
767 validation_types: List of validation types to perform ("statistical", "semantic")
768 statistical_test_type: Type of statistical test ("welch_t_test", "ks_test", "mann_whitney", etc.)
769 significance_level: Significance level for statistical tests
770 output_format: "values" for SimulationExperimentEmpiricalValidationResult object, "report" for markdown report
772 Returns:
773 SimulationExperimentEmpiricalValidationResult object or markdown report string
774 """
775 result = SimulationExperimentEmpiricalValidationResult(
776 validation_type=", ".join(validation_types),
777 control_name=control.name or "Control",
778 treatment_name=treatment.name or "Treatment"
779 )
781 # Perform statistical validation
782 if "statistical" in validation_types:
783 result.statistical_results = self._perform_statistical_validation(
784 control, treatment, significance_level, statistical_test_type
785 )
787 # Perform semantic validation
788 if "semantic" in validation_types:
789 result.semantic_results = self._perform_semantic_validation(
790 control, treatment
791 )
793 # Calculate overall score and summary
794 result.overall_score = self._calculate_overall_score(result)
795 result.summary = self._generate_summary(result)
797 if output_format == "report":
798 return self._generate_markdown_report(result, control, treatment)
799 else:
800 return result
802 def _perform_statistical_validation(self,
803 control: SimulationExperimentDataset,
804 treatment: SimulationExperimentDataset,
805 significance_level: float,
806 test_type: str = "welch_t_test") -> Dict[str, Any]:
807 """
808 Perform statistical hypothesis testing on simulation experiment key results.
810 Args:
811 control: Control dataset
812 treatment: Treatment dataset
813 significance_level: Alpha level for statistical tests
814 test_type: Type of statistical test to perform
815 """
816 if not control.key_results or not treatment.key_results:
817 return {"error": "No key results available for statistical testing"}
819 try:
820 # Prepare data for StatisticalTester
821 control_data = {"control": {}}
822 treatment_data = {"treatment": {}}
824 # Convert single values to lists if needed and find common metrics
825 common_metrics = set(control.key_results.keys()) & set(treatment.key_results.keys())
827 for metric in common_metrics:
828 control_value = control.key_results[metric]
829 treatment_value = treatment.key_results[metric]
831 # Convert single values to lists and filter out None values
832 if not isinstance(control_value, list):
833 control_value = [control_value] if control_value is not None else []
834 else:
835 control_value = [v for v in control_value if v is not None]
837 if not isinstance(treatment_value, list):
838 treatment_value = [treatment_value] if treatment_value is not None else []
839 else:
840 treatment_value = [v for v in treatment_value if v is not None]
842 # Only include metrics that have valid data points
843 if len(control_value) > 0 and len(treatment_value) > 0:
844 control_data["control"][metric] = control_value
845 treatment_data["treatment"][metric] = treatment_value
847 if not common_metrics:
848 return {"error": "No common metrics found between control and treatment"}
850 # Run statistical tests
851 tester = StatisticalTester(control_data, treatment_data)
852 test_results = tester.run_test(
853 test_type=test_type,
854 alpha=significance_level
855 )
857 return {
858 "common_metrics": list(common_metrics),
859 "test_results": test_results,
860 "test_type": test_type,
861 "significance_level": significance_level
862 }
864 except Exception as e:
865 return {"error": f"Statistical testing failed: {str(e)}"}
867 def _perform_semantic_validation(self,
868 control: SimulationExperimentDataset,
869 treatment: SimulationExperimentDataset) -> Dict[str, Any]:
870 """Perform semantic proximity analysis on simulation experiment agent justifications."""
871 results = {
872 "individual_comparisons": [],
873 "summary_comparison": None,
874 "average_proximity": None
875 }
877 # Compare individual justifications if available
878 if control.agent_justifications and treatment.agent_justifications:
879 proximities = []
881 for i, control_just in enumerate(control.agent_justifications):
882 for j, treatment_just in enumerate(treatment.agent_justifications):
883 control_text = control.get_justification_text(control_just)
884 treatment_text = treatment.get_justification_text(treatment_just)
886 if control_text and treatment_text:
887 proximity_score = compute_semantic_proximity(
888 control_text,
889 treatment_text,
890 context="Comparing agent justifications from simulation experiments"
891 )
893 # Handle case where LLM call fails or returns invalid data
894 if proximity_score is None or not isinstance(proximity_score, (int, float)):
895 raise ValueError("Invalid semantic proximity score")
897 # Get agent references (names or indices)
898 control_agent_ref = control.get_justification_agent_reference(control_just) or f"Agent_{i}"
899 treatment_agent_ref = treatment.get_justification_agent_reference(treatment_just) or f"Agent_{j}"
901 comparison = {
902 "control_agent": control_agent_ref,
903 "treatment_agent": treatment_agent_ref,
904 "proximity_score": proximity_score,
905 "justification": f"Semantic proximity score: {proximity_score:.3f}"
906 }
908 results["individual_comparisons"].append(comparison)
909 proximities.append(proximity_score)
911 if proximities:
912 results["average_proximity"] = sum(proximities) / len(proximities)
914 # Compare summary justifications if available
915 if control.justification_summary and treatment.justification_summary:
916 summary_proximity_score = compute_semantic_proximity(
917 control.justification_summary,
918 treatment.justification_summary,
919 context="Comparing summary justifications from simulation experiments"
920 )
922 # Handle case where LLM call fails or returns invalid data
923 if summary_proximity_score is None or not isinstance(summary_proximity_score, (int, float)):
924 summary_proximity_score = 0.5 # Default neutral score
926 results["summary_comparison"] = {
927 "proximity_score": summary_proximity_score,
928 "justification": f"Summary semantic proximity score: {summary_proximity_score:.3f}"
929 }
931 return results
933 def _calculate_overall_score(self, result: SimulationExperimentEmpiricalValidationResult) -> float:
934 """Calculate an overall simulation experiment empirical validation score based on statistical and semantic results."""
935 scores = []
937 # Statistical component based on effect sizes
938 if result.statistical_results and "test_results" in result.statistical_results:
939 test_results = result.statistical_results["test_results"]
940 effect_sizes = []
942 for treatment_name, treatment_results in test_results.items():
943 for metric, metric_result in treatment_results.items():
944 # Extract effect size based on test type
945 effect_size = self._extract_effect_size(metric_result)
946 if effect_size is not None:
947 effect_sizes.append(effect_size)
949 if effect_sizes:
950 # Convert effect sizes to similarity scores (closer to 0 = more similar)
951 # Use inverse transformation: similarity = 1 / (1 + |effect_size|)
952 # For very small effect sizes (< 0.1), give even higher scores
953 similarity_scores = []
954 for es in effect_sizes:
955 abs_es = abs(es)
956 if abs_es < 0.1: # Very small effect size
957 similarity_scores.append(0.95 + 0.05 * (1.0 / (1.0 + abs_es)))
958 else:
959 similarity_scores.append(1.0 / (1.0 + abs_es))
961 statistical_score = sum(similarity_scores) / len(similarity_scores)
962 scores.append(statistical_score)
964 # Semantic component
965 if result.semantic_results:
966 semantic_scores = []
968 # Average proximity from individual comparisons
969 if result.semantic_results.get("average_proximity") is not None:
970 semantic_scores.append(result.semantic_results["average_proximity"])
972 # Summary proximity
973 if result.semantic_results.get("summary_comparison"):
974 semantic_scores.append(result.semantic_results["summary_comparison"]["proximity_score"])
976 if semantic_scores:
977 semantic_score = sum(semantic_scores) / len(semantic_scores)
978 scores.append(semantic_score)
980 # If we have both statistical and semantic scores, and the statistical score is very high (>0.9)
981 # indicating statistically equivalent data, weight the statistical component more heavily
982 if len(scores) == 2 and scores[0] > 0.9: # First score is statistical
983 # Weight statistical component at 70%, semantic at 30% for equivalent data
984 return 0.7 * scores[0] + 0.3 * scores[1]
986 return sum(scores) / len(scores) if scores else 0.0
988 def _generate_summary(self, result: SimulationExperimentEmpiricalValidationResult) -> str:
989 """Generate a text summary of the simulation experiment empirical validation results."""
990 summary_parts = []
992 if result.statistical_results:
993 if "error" in result.statistical_results:
994 summary_parts.append(f"Statistical validation: {result.statistical_results['error']}")
995 else:
996 test_results = result.statistical_results.get("test_results", {})
997 effect_sizes = []
998 significant_tests = 0
999 total_tests = 0
1001 for treatment_results in test_results.values():
1002 for metric_result in treatment_results.values():
1003 total_tests += 1
1004 if metric_result.get("significant", False):
1005 significant_tests += 1
1007 # Collect effect sizes
1008 effect_size = self._extract_effect_size(metric_result)
1009 if effect_size is not None:
1010 effect_sizes.append(abs(effect_size))
1012 if effect_sizes:
1013 avg_effect_size = sum(effect_sizes) / len(effect_sizes)
1014 summary_parts.append(
1015 f"Statistical validation: {significant_tests}/{total_tests} tests significant, "
1016 f"average effect size: {avg_effect_size:.3f}"
1017 )
1018 else:
1019 summary_parts.append(
1020 f"Statistical validation: {significant_tests}/{total_tests} tests showed significant differences"
1021 )
1023 if result.semantic_results:
1024 avg_proximity = result.semantic_results.get("average_proximity")
1025 if avg_proximity is not None:
1026 summary_parts.append(
1027 f"Semantic validation: Average proximity score of {avg_proximity:.3f}"
1028 )
1030 summary_comparison = result.semantic_results.get("summary_comparison")
1031 if summary_comparison:
1032 summary_parts.append(
1033 f"Summary proximity: {summary_comparison['proximity_score']:.3f}"
1034 )
1036 if result.overall_score is not None:
1037 summary_parts.append(f"Overall validation score: {result.overall_score:.3f}")
1039 return "; ".join(summary_parts) if summary_parts else "No validation results available"
1041 def _generate_markdown_report(self, result: SimulationExperimentEmpiricalValidationResult,
1042 control: SimulationExperimentDataset = None,
1043 treatment: SimulationExperimentDataset = None) -> str:
1044 """Generate a comprehensive markdown report for simulation experiment empirical validation."""
1045 overall_score_str = f"{result.overall_score:.3f}" if result.overall_score is not None else "N/A"
1047 report = f"""# Simulation Experiment Empirical Validation Report
1049**Validation Type:** {result.validation_type}
1050**Control/Empirical:** {result.control_name}
1051**Treatment/Simulation:** {result.treatment_name}
1052**Timestamp:** {result.timestamp}
1053**Overall Score:** {overall_score_str}
1055## Summary
1057{result.summary}
1059"""
1061 # Add data type information if available
1062 if control or treatment:
1063 data_type_info = self._generate_data_type_info_section(control, treatment)
1064 if data_type_info:
1065 report += data_type_info
1067 # Statistical Results Section
1068 if result.statistical_results:
1069 report += "## Statistical Validation\n\n"
1071 if "error" in result.statistical_results:
1072 report += f"**Error:** {result.statistical_results['error']}\n\n"
1073 else:
1074 stats = result.statistical_results
1075 report += f"**Common Metrics:** {', '.join(stats.get('common_metrics', []))}\n\n"
1076 report += f"**Significance Level:** {stats.get('significance_level', 'N/A')}\n\n"
1078 test_results = stats.get("test_results", {})
1079 if test_results:
1080 report += "### Test Results\n\n"
1082 for treatment_name, treatment_results in test_results.items():
1083 report += f"#### {treatment_name}\n\n"
1085 for metric, metric_result in treatment_results.items():
1086 report += f"**{metric}:**\n\n"
1088 significant = metric_result.get("significant", False)
1089 p_value = metric_result.get("p_value", "N/A")
1090 test_type = metric_result.get("test_type", "N/A")
1091 effect_size = self._extract_effect_size(metric_result)
1093 # Get the appropriate statistic based on test type
1094 statistic = "N/A"
1095 if "t_statistic" in metric_result:
1096 statistic = metric_result["t_statistic"]
1097 elif "u_statistic" in metric_result:
1098 statistic = metric_result["u_statistic"]
1099 elif "f_statistic" in metric_result:
1100 statistic = metric_result["f_statistic"]
1101 elif "chi2_statistic" in metric_result:
1102 statistic = metric_result["chi2_statistic"]
1103 elif "ks_statistic" in metric_result:
1104 statistic = metric_result["ks_statistic"]
1106 status = "✅ Significant" if significant else "❌ Not Significant"
1108 report += f"- **{test_type}:** {status}\n"
1109 report += f" - p-value: {p_value}\n"
1110 report += f" - statistic: {statistic}\n"
1111 if effect_size is not None:
1112 effect_interpretation = self._interpret_effect_size(abs(effect_size), test_type)
1113 report += f" - effect size: {effect_size:.3f} ({effect_interpretation})\n"
1115 report += "\n"
1117 # Semantic Results Section
1118 if result.semantic_results:
1119 report += "## Semantic Validation\n\n"
1121 semantic = result.semantic_results
1123 # Individual comparisons
1124 individual_comps = semantic.get("individual_comparisons", [])
1125 if individual_comps:
1126 report += "### Individual Agent Comparisons\n\n"
1128 for comp in individual_comps:
1129 score = comp["proximity_score"]
1130 control_agent = comp["control_agent"]
1131 treatment_agent = comp["treatment_agent"]
1132 justification = comp["justification"]
1134 report += f"**{control_agent} vs {treatment_agent}:** {score:.3f}\n\n"
1135 report += f"{justification}\n\n"
1137 avg_proximity = semantic.get("average_proximity")
1138 if avg_proximity:
1139 report += f"**Average Proximity Score:** {avg_proximity:.3f}\n\n"
1141 # Summary comparison
1142 summary_comp = semantic.get("summary_comparison")
1143 if summary_comp:
1144 report += "### Summary Comparison\n\n"
1145 report += f"**Proximity Score:** {summary_comp['proximity_score']:.3f}\n\n"
1146 report += f"**Justification:** {summary_comp['justification']}\n\n"
1148 return report
1150 def _generate_data_type_info_section(self, control: SimulationExperimentDataset,
1151 treatment: SimulationExperimentDataset) -> str:
1152 """Generate comprehensive data type information section for the report."""
1153 all_metrics = set()
1155 # Collect all metrics from both datasets
1156 if control:
1157 all_metrics.update(control.key_results.keys())
1158 if treatment:
1159 all_metrics.update(treatment.key_results.keys())
1161 if not all_metrics:
1162 return ""
1164 # Group metrics by data type
1165 data_type_groups = {}
1166 for metric in all_metrics:
1167 for dataset_name, dataset in [("control", control), ("treatment", treatment)]:
1168 if dataset and metric in dataset.data_types:
1169 data_type = dataset.data_types[metric]
1170 if data_type not in data_type_groups:
1171 data_type_groups[data_type] = set()
1172 data_type_groups[data_type].add(metric)
1173 break # Use first available data type
1175 if not data_type_groups:
1176 return ""
1178 report = "## Data Type Information\n\n"
1180 for data_type, metrics in sorted(data_type_groups.items()):
1181 if not metrics:
1182 continue
1184 report += f"### {data_type.title()} Data\n\n"
1186 if data_type == "categorical":
1187 report += "String categories converted to ordinal values for statistical analysis.\n\n"
1188 elif data_type == "ordinal":
1189 report += "Ordered categories or levels with meaningful ranking.\n\n"
1190 elif data_type == "ranking":
1191 report += "Rank positions (1st, 2nd, 3rd, etc.) indicating preference or order.\n\n"
1192 elif data_type == "count":
1193 report += "Non-negative integer counts (frequencies, occurrences, etc.).\n\n"
1194 elif data_type == "proportion":
1195 report += "Values between 0-1 representing proportions or percentages.\n\n"
1196 elif data_type == "binary":
1197 report += "Binary outcomes converted to 0/1 for analysis.\n\n"
1198 elif data_type == "numeric":
1199 report += "Continuous numeric values.\n\n"
1201 for metric in sorted(metrics):
1202 report += f"#### {metric}\n\n"
1204 # Show information from both datasets
1205 for dataset_name, dataset in [("Control", control), ("Treatment", treatment)]:
1206 if not dataset or metric not in dataset.key_results:
1207 continue
1209 data_type_info = dataset.get_data_type_info(metric)
1210 summary = dataset.get_metric_summary(metric)
1212 report += f"**{dataset_name}:**\n"
1214 if data_type == "categorical":
1215 if "categories" in data_type_info:
1216 categories = data_type_info["categories"]
1217 mapping = data_type_info.get("category_mapping", {})
1219 report += f"- Categories: {', '.join(f'`{cat}`' for cat in categories)}\n"
1220 report += f"- Ordinal mapping: {mapping}\n"
1222 if "category_distribution" in summary:
1223 distribution = summary["category_distribution"]
1224 total = sum(distribution.values())
1225 report += "- Distribution: "
1226 dist_items = []
1227 for cat in categories:
1228 count = distribution.get(cat, 0)
1229 pct = (count / total * 100) if total > 0 else 0
1230 dist_items.append(f"`{cat}`: {count} ({pct:.1f}%)")
1231 report += ", ".join(dist_items) + "\n"
1233 elif data_type == "ordinal":
1234 if "ordinal_categories" in data_type_info:
1235 # String-based ordinal
1236 categories = data_type_info["ordinal_categories"]
1237 mapping = data_type_info.get("ordinal_mapping", {})
1238 report += f"- Ordered categories: {' < '.join(f'`{cat}`' for cat in categories)}\n"
1239 report += f"- Ordinal mapping: {mapping}\n"
1240 elif "ordinal_info" in data_type_info:
1241 # Numeric ordinal
1242 info = data_type_info["ordinal_info"]
1243 report += f"- Value range: {info.get('min_value')} to {info.get('max_value')}\n"
1244 report += f"- Unique levels: {info.get('num_levels')} ({info.get('unique_values')})\n"
1246 elif data_type == "ranking":
1247 if "ranking_info" in data_type_info:
1248 info = data_type_info["ranking_info"]
1249 report += f"- Rank range: {info.get('min_rank')} to {info.get('max_rank')}\n"
1250 report += f"- Number of ranks: {info.get('num_ranks')}\n"
1251 report += f"- Direction: {info.get('direction', 'ascending')} (1 = best)\n"
1253 if "rank_distribution" in summary:
1254 distribution = summary["rank_distribution"]
1255 report += "- Distribution: "
1256 rank_items = []
1257 for rank in sorted(distribution.keys()):
1258 count = distribution[rank]
1259 rank_items.append(f"Rank {rank}: {count}")
1260 report += ", ".join(rank_items) + "\n"
1262 elif data_type == "binary":
1263 if "binary_distribution" in summary:
1264 distribution = summary["binary_distribution"]
1265 true_count = distribution.get("true", 0)
1266 false_count = distribution.get("false", 0)
1267 total = true_count + false_count
1268 if total > 0:
1269 true_pct = (true_count / total * 100)
1270 false_pct = (false_count / total * 100)
1271 report += f"- Distribution: True: {true_count} ({true_pct:.1f}%), False: {false_count} ({false_pct:.1f}%)\n"
1273 elif data_type in ["count", "proportion", "numeric"]:
1274 if "min_value" in summary and "max_value" in summary:
1275 report += f"- Range: {summary['min_value']} to {summary['max_value']}\n"
1276 if "valid_values" in summary:
1277 report += f"- Valid values: {summary['valid_values']}/{summary.get('total_values', 'N/A')}\n"
1279 report += "\n"
1281 return report
1283 def _generate_categorical_info_section(self, control: SimulationExperimentDataset,
1284 treatment: SimulationExperimentDataset) -> str:
1285 """
1286 Generate categorical data information section for the report.
1287 This is kept for backward compatibility and now calls the more comprehensive data type method.
1288 """
1289 return self._generate_data_type_info_section(control, treatment)
1291 @classmethod
1292 def read_empirical_data_from_csv(cls,
1293 file_path: Union[str, Path],
1294 experimental_data_type: str = "single_value_per_agent",
1295 agent_id_column: Optional[str] = None,
1296 agent_comments_column: Optional[str] = None,
1297 agent_attributes_columns: Optional[List[str]] = None,
1298 value_column: Optional[str] = None,
1299 ranking_columns: Optional[List[str]] = None,
1300 ordinal_ranking_column: Optional[str] = None,
1301 ordinal_ranking_separator: str = "-",
1302 ordinal_ranking_options: Optional[List[str]] = None,
1303 dataset_name: Optional[str] = None,
1304 dataset_description: Optional[str] = None,
1305 encoding: str = "utf-8") -> 'SimulationExperimentDataset':
1306 """
1307 Read empirical data from a CSV file and convert it to a SimulationExperimentDataset.
1309 Args:
1310 file_path: Path to the CSV file
1311 experimental_data_type: Type of experimental data:
1312 - "single_value_per_agent": Each agent has a single value (e.g., score, rating)
1313 - "ranking_per_agent": Each agent provides rankings for multiple items (separate columns)
1314 - "ordinal_ranking_per_agent": Each agent provides ordinal ranking in single column with separator
1315 agent_id_column: Column name containing agent identifiers (optional)
1316 agent_comments_column: Column name containing agent comments/explanations (optional)
1317 agent_attributes_columns: List of column names containing agent attributes (age, gender, etc.)
1318 value_column: Column name containing the main value for single_value_per_agent mode
1319 ranking_columns: List of column names containing rankings for ranking_per_agent mode
1320 ordinal_ranking_column: Column name containing ordinal rankings for ordinal_ranking_per_agent mode
1321 ordinal_ranking_separator: Separator used in ordinal ranking strings (default: "-")
1322 ordinal_ranking_options: List of options being ranked (if None, auto-detected from data)
1323 dataset_name: Optional name for the dataset
1324 dataset_description: Optional description of the dataset
1325 encoding: File encoding (default: utf-8)
1327 Returns:
1328 SimulationExperimentDataset object populated with the CSV data
1330 Raises:
1331 FileNotFoundError: If the CSV file doesn't exist
1332 ValueError: If required columns are missing or data format is invalid
1333 pandas.errors.EmptyDataError: If the CSV file is empty
1334 """
1335 file_path = Path(file_path)
1337 if not file_path.exists():
1338 raise FileNotFoundError(f"CSV file not found: {file_path}")
1340 try:
1341 # Read CSV with UTF-8 encoding and error handling
1342 df = pd.read_csv(file_path, encoding=encoding, encoding_errors='replace')
1343 except pd.errors.EmptyDataError:
1344 raise pd.errors.EmptyDataError(f"CSV file is empty: {file_path}")
1345 except UnicodeDecodeError as e:
1346 raise ValueError(f"Failed to read CSV file with encoding {encoding}: {e}")
1348 if df.empty:
1349 raise ValueError(f"CSV file contains no data: {file_path}")
1351 # Use common processing method
1352 return cls._process_empirical_data_from_dataframe(
1353 df=df,
1354 experimental_data_type=experimental_data_type,
1355 agent_id_column=agent_id_column,
1356 agent_comments_column=agent_comments_column,
1357 agent_attributes_columns=agent_attributes_columns,
1358 value_column=value_column,
1359 ranking_columns=ranking_columns,
1360 ordinal_ranking_column=ordinal_ranking_column,
1361 ordinal_ranking_separator=ordinal_ranking_separator,
1362 ordinal_ranking_options=ordinal_ranking_options,
1363 dataset_name=dataset_name or f"Empirical_Data_{file_path.stem}",
1364 dataset_description=dataset_description or f"Empirical data loaded from {file_path.name}"
1365 )
1367 @classmethod
1368 def read_empirical_data_from_dataframe(cls,
1369 df: pd.DataFrame,
1370 experimental_data_type: str = "single_value_per_agent",
1371 agent_id_column: Optional[str] = None,
1372 agent_comments_column: Optional[str] = None,
1373 agent_attributes_columns: Optional[List[str]] = None,
1374 value_column: Optional[str] = None,
1375 ranking_columns: Optional[List[str]] = None,
1376 ordinal_ranking_column: Optional[str] = None,
1377 ordinal_ranking_separator: str = "-",
1378 ordinal_ranking_options: Optional[List[str]] = None,
1379 dataset_name: Optional[str] = None,
1380 dataset_description: Optional[str] = None) -> 'SimulationExperimentDataset':
1381 """
1382 Read empirical data from a pandas DataFrame and convert it to a SimulationExperimentDataset.
1384 This method provides the same functionality as read_empirical_data_from_csv but accepts
1385 a pandas DataFrame directly, eliminating the need to save DataFrames to CSV files first.
1387 Args:
1388 df: The pandas DataFrame containing the empirical data
1389 experimental_data_type: Type of experimental data:
1390 - "single_value_per_agent": Each agent has a single value (e.g., score, rating)
1391 - "ranking_per_agent": Each agent provides rankings for multiple items (separate columns)
1392 - "ordinal_ranking_per_agent": Each agent provides ordinal ranking in single column with separator
1393 agent_id_column: Column name containing agent identifiers (optional)
1394 agent_comments_column: Column name containing agent comments/explanations (optional)
1395 agent_attributes_columns: List of column names containing agent attributes (age, gender, etc.)
1396 value_column: Column name containing the main value for single_value_per_agent mode
1397 ranking_columns: List of column names containing rankings for ranking_per_agent mode
1398 ordinal_ranking_column: Column name containing ordinal rankings for ordinal_ranking_per_agent mode
1399 ordinal_ranking_separator: Separator used in ordinal ranking strings (default: "-")
1400 ordinal_ranking_options: List of options being ranked (if None, auto-detected from data)
1401 dataset_name: Optional name for the dataset
1402 dataset_description: Optional description of the dataset
1404 Returns:
1405 SimulationExperimentDataset object populated with the DataFrame data
1407 Raises:
1408 ValueError: If required columns are missing or data format is invalid
1409 TypeError: If df is not a pandas DataFrame
1410 """
1411 # Validate input
1412 if not isinstance(df, pd.DataFrame):
1413 raise TypeError(f"Expected pandas DataFrame, got {type(df)}")
1415 if df.empty:
1416 raise ValueError("DataFrame contains no data")
1418 # Use common processing method
1419 return cls._process_empirical_data_from_dataframe(
1420 df=df,
1421 experimental_data_type=experimental_data_type,
1422 agent_id_column=agent_id_column,
1423 agent_comments_column=agent_comments_column,
1424 agent_attributes_columns=agent_attributes_columns,
1425 value_column=value_column,
1426 ranking_columns=ranking_columns,
1427 ordinal_ranking_column=ordinal_ranking_column,
1428 ordinal_ranking_separator=ordinal_ranking_separator,
1429 ordinal_ranking_options=ordinal_ranking_options,
1430 dataset_name=dataset_name or "Empirical_Data_from_DataFrame",
1431 dataset_description=dataset_description or "Empirical data loaded from pandas DataFrame"
1432 )
1434 @classmethod
1435 def _process_empirical_data_from_dataframe(cls,
1436 df: pd.DataFrame,
1437 experimental_data_type: str,
1438 agent_id_column: Optional[str],
1439 agent_comments_column: Optional[str],
1440 agent_attributes_columns: Optional[List[str]],
1441 value_column: Optional[str],
1442 ranking_columns: Optional[List[str]],
1443 ordinal_ranking_column: Optional[str],
1444 ordinal_ranking_separator: str,
1445 ordinal_ranking_options: Optional[List[str]],
1446 dataset_name: str,
1447 dataset_description: str) -> 'SimulationExperimentDataset':
1448 """
1449 Common processing method for both CSV and DataFrame inputs.
1451 This method contains the shared logic for processing empirical data regardless of input source.
1452 """
1453 # Initialize dataset
1454 dataset = SimulationExperimentDataset(
1455 name=dataset_name,
1456 description=dataset_description
1457 )
1459 # Process based on experimental data type
1460 if experimental_data_type == "single_value_per_agent":
1461 cls._process_single_value_per_agent_csv(df, dataset, value_column,
1462 agent_id_column, agent_comments_column,
1463 agent_attributes_columns)
1464 elif experimental_data_type == "ranking_per_agent":
1465 cls._process_ranking_per_agent_csv(df, dataset, ranking_columns,
1466 agent_id_column, agent_comments_column,
1467 agent_attributes_columns)
1468 elif experimental_data_type == "ordinal_ranking_per_agent":
1469 cls._process_ordinal_ranking_per_agent_csv(df, dataset, ordinal_ranking_column,
1470 ordinal_ranking_separator, ordinal_ranking_options,
1471 agent_id_column, agent_comments_column,
1472 agent_attributes_columns)
1473 else:
1474 raise ValueError(f"Unsupported experimental_data_type: {experimental_data_type}. "
1475 f"Supported types: 'single_value_per_agent', 'ranking_per_agent', 'ordinal_ranking_per_agent'")
1477 # Process data types after all data is loaded
1478 dataset._process_data_types()
1480 return dataset
1482 @classmethod
1483 def _process_single_value_per_agent_csv(cls,
1484 df: pd.DataFrame,
1485 dataset: 'SimulationExperimentDataset',
1486 value_column: Optional[str],
1487 agent_id_column: Optional[str],
1488 agent_comments_column: Optional[str],
1489 agent_attributes_columns: Optional[List[str]]):
1490 """Process CSV data for single value per agent experiments."""
1492 # Auto-detect value column if not specified
1493 if value_column is None:
1494 # Look for common column names that might contain the main value
1495 value_candidates = [col for col in df.columns if any(keyword in col.lower()
1496 for keyword in ['vote', 'score', 'rating', 'value', 'response', 'answer'])]
1498 if len(value_candidates) == 1:
1499 value_column = value_candidates[0]
1500 elif len(value_candidates) > 1:
1501 # Prefer shorter, more specific names
1502 value_column = min(value_candidates, key=len)
1503 else:
1504 # Fall back to first numeric column
1505 numeric_cols = df.select_dtypes(include=['number']).columns.tolist()
1506 if numeric_cols:
1507 value_column = numeric_cols[0]
1508 else:
1509 raise ValueError("No suitable value column found. Please specify value_column parameter.")
1511 if value_column not in df.columns:
1512 raise ValueError(f"Value column '{value_column}' not found in CSV. Available columns: {list(df.columns)}")
1514 # Extract main values (handling mixed types)
1515 values = []
1516 for val in df[value_column]:
1517 if pd.isna(val):
1518 values.append(None)
1519 else:
1520 # Try to convert to numeric if possible, otherwise keep as string
1521 try:
1522 if isinstance(val, str) and val.strip().isdigit():
1523 values.append(int(val.strip()))
1524 elif isinstance(val, str):
1525 try:
1526 float_val = float(val.strip())
1527 # If it's a whole number, convert to int
1528 values.append(int(float_val) if float_val.is_integer() else float_val)
1529 except ValueError:
1530 values.append(val.strip())
1531 else:
1532 values.append(val)
1533 except (AttributeError, ValueError):
1534 values.append(val)
1536 # Store the main experimental result
1537 dataset.key_results[value_column] = values
1538 dataset.result_types[value_column] = "per_agent"
1540 # Process agent IDs/names
1541 agent_names = []
1542 if agent_id_column and agent_id_column in df.columns:
1543 for agent_id in df[agent_id_column]:
1544 if pd.isna(agent_id):
1545 agent_names.append(None)
1546 else:
1547 agent_names.append(str(agent_id))
1548 else:
1549 # Generate default agent names
1550 for i in range(len(df)):
1551 agent_names.append(f"Agent_{i+1}")
1553 dataset.agent_names = agent_names
1555 # Process agent comments/justifications
1556 if agent_comments_column and agent_comments_column in df.columns:
1557 justifications = []
1558 for i, comment in enumerate(df[agent_comments_column]):
1559 # Include all comments, even empty ones, to maintain agent alignment
1560 agent_name = agent_names[i] if i < len(agent_names) else f"Agent_{i+1}"
1561 comment_text = str(comment).strip() if pd.notna(comment) else ""
1562 justifications.append({
1563 "agent_name": agent_name,
1564 "agent_index": i,
1565 "justification": comment_text
1566 })
1567 dataset.agent_justifications = justifications
1569 # Process agent attributes
1570 if agent_attributes_columns:
1571 for attr_col in agent_attributes_columns:
1572 if attr_col in df.columns:
1573 attr_values = []
1574 for val in df[attr_col]:
1575 if pd.isna(val):
1576 attr_values.append(None)
1577 else:
1578 attr_values.append(str(val).strip())
1580 # Store in agent_attributes instead of key_results
1581 dataset.agent_attributes[attr_col] = attr_values
1583 @classmethod
1584 def _process_ranking_per_agent_csv(cls,
1585 df: pd.DataFrame,
1586 dataset: 'SimulationExperimentDataset',
1587 ranking_columns: Optional[List[str]],
1588 agent_id_column: Optional[str],
1589 agent_comments_column: Optional[str],
1590 agent_attributes_columns: Optional[List[str]]):
1591 """Process CSV data for ranking per agent experiments."""
1593 # Auto-detect ranking columns if not specified
1594 if ranking_columns is None:
1595 # Look for columns that might contain rankings
1596 numeric_cols = df.select_dtypes(include=['number']).columns.tolist()
1598 # Exclude agent ID column if specified
1599 if agent_id_column and agent_id_column in numeric_cols:
1600 numeric_cols.remove(agent_id_column)
1602 if len(numeric_cols) < 2:
1603 raise ValueError("No suitable ranking columns found. Please specify ranking_columns parameter.")
1605 ranking_columns = numeric_cols
1607 # Validate ranking columns exist
1608 missing_cols = [col for col in ranking_columns if col not in df.columns]
1609 if missing_cols:
1610 raise ValueError(f"Ranking columns not found in CSV: {missing_cols}. Available columns: {list(df.columns)}")
1612 # Process each ranking column
1613 for rank_col in ranking_columns:
1614 rankings = []
1615 for val in df[rank_col]:
1616 if pd.isna(val):
1617 rankings.append(None)
1618 else:
1619 try:
1620 # Convert to integer rank
1621 rankings.append(int(float(val)))
1622 except (ValueError, TypeError):
1623 rankings.append(None)
1625 dataset.key_results[rank_col] = rankings
1626 dataset.result_types[rank_col] = "per_agent"
1627 dataset.data_types[rank_col] = "ranking"
1629 # Process agent IDs/names (same as single value method)
1630 agent_names = []
1631 if agent_id_column and agent_id_column in df.columns:
1632 for agent_id in df[agent_id_column]:
1633 if pd.isna(agent_id):
1634 agent_names.append(None)
1635 else:
1636 agent_names.append(str(agent_id))
1637 else:
1638 # Generate default agent names
1639 for i in range(len(df)):
1640 agent_names.append(f"Agent_{i+1}")
1642 dataset.agent_names = agent_names
1644 # Process agent comments (same as single value method)
1645 if agent_comments_column and agent_comments_column in df.columns:
1646 justifications = []
1647 for i, comment in enumerate(df[agent_comments_column]):
1648 # Include all comments, even empty ones, to maintain agent alignment
1649 agent_name = agent_names[i] if i < len(agent_names) else f"Agent_{i+1}"
1650 comment_text = str(comment).strip() if pd.notna(comment) else ""
1651 justifications.append({
1652 "agent_name": agent_name,
1653 "agent_index": i,
1654 "justification": comment_text
1655 })
1656 dataset.agent_justifications = justifications
1658 # Process agent attributes (same as single value method)
1659 if agent_attributes_columns:
1660 for attr_col in agent_attributes_columns:
1661 if attr_col in df.columns:
1662 attr_values = []
1663 for val in df[attr_col]:
1664 if pd.isna(val):
1665 attr_values.append(None)
1666 else:
1667 attr_values.append(str(val).strip())
1669 # Store in agent_attributes instead of key_results
1670 dataset.agent_attributes[attr_col] = attr_values
1672 @classmethod
1673 def _process_ordinal_ranking_per_agent_csv(cls,
1674 df: pd.DataFrame,
1675 dataset: 'SimulationExperimentDataset',
1676 ordinal_ranking_column: Optional[str],
1677 ordinal_ranking_separator: str,
1678 ordinal_ranking_options: Optional[List[str]],
1679 agent_id_column: Optional[str],
1680 agent_comments_column: Optional[str],
1681 agent_attributes_columns: Optional[List[str]]):
1682 """Process CSV data for ordinal ranking per agent experiments (single column with separator)."""
1684 # Auto-detect ranking column if not specified
1685 if ordinal_ranking_column is None:
1686 # Look for columns that might contain ordinal rankings
1687 ranking_candidates = [col for col in df.columns if any(keyword in col.lower()
1688 for keyword in ['ranking', 'rank', 'order', 'preference', 'choice'])]
1690 if len(ranking_candidates) == 1:
1691 ordinal_ranking_column = ranking_candidates[0]
1692 elif len(ranking_candidates) > 1:
1693 # Prefer shorter, more specific names
1694 ordinal_ranking_column = min(ranking_candidates, key=len)
1695 else:
1696 # Fall back to first string column that contains separator
1697 string_cols = df.select_dtypes(include=['object']).columns.tolist()
1698 if agent_id_column and agent_id_column in string_cols:
1699 string_cols.remove(agent_id_column)
1700 if agent_comments_column and agent_comments_column in string_cols:
1701 string_cols.remove(agent_comments_column)
1703 # Check which string columns contain the separator
1704 for col in string_cols:
1705 if df[col].astype(str).str.contains(ordinal_ranking_separator, na=False).any():
1706 ordinal_ranking_column = col
1707 break
1709 if ordinal_ranking_column is None:
1710 raise ValueError("No suitable ordinal ranking column found. Please specify ordinal_ranking_column parameter.")
1712 if ordinal_ranking_column not in df.columns:
1713 raise ValueError(f"Ordinal ranking column '{ordinal_ranking_column}' not found in CSV. Available columns: {list(df.columns)}")
1715 # Auto-detect ranking options if not specified
1716 if ordinal_ranking_options is None:
1717 ordinal_ranking_options = cls._auto_detect_ranking_options(df[ordinal_ranking_column], ordinal_ranking_separator)
1719 # Parse ordinal rankings and convert to individual ranking columns
1720 ranking_data = cls._parse_ordinal_rankings(df[ordinal_ranking_column], ordinal_ranking_separator, ordinal_ranking_options)
1722 # Store parsed rankings as separate metrics
1723 for option in ordinal_ranking_options:
1724 option_ranking_key = f"{option}_rank"
1725 dataset.key_results[option_ranking_key] = ranking_data[option]
1726 dataset.result_types[option_ranking_key] = "per_agent"
1727 dataset.data_types[option_ranking_key] = "ranking"
1729 # Store ranking info (always for ordinal ranking data)
1730 valid_ranks = [r for r in ranking_data[option] if r is not None]
1732 # Always store ranking info for ordinal ranking data, regardless of valid ranks
1733 ranking_info = {
1734 "direction": "ascending", # 1 = best, higher = worse
1735 "original_options": ordinal_ranking_options,
1736 "separator": ordinal_ranking_separator,
1737 "source_column": ordinal_ranking_column
1738 }
1740 # Add rank statistics if valid ranks exist
1741 if valid_ranks:
1742 ranking_info.update({
1743 "min_rank": min(valid_ranks),
1744 "max_rank": max(valid_ranks),
1745 "num_ranks": len(set(valid_ranks)),
1746 "rank_values": sorted(set(valid_ranks))
1747 })
1748 else:
1749 # Set reasonable defaults based on options
1750 ranking_info.update({
1751 "min_rank": 1,
1752 "max_rank": len(ordinal_ranking_options),
1753 "num_ranks": 0,
1754 "rank_values": []
1755 })
1757 dataset.ranking_info[option_ranking_key] = ranking_info
1759 # Process agent IDs/names (same as other methods)
1760 agent_names = []
1761 if agent_id_column and agent_id_column in df.columns:
1762 for agent_id in df[agent_id_column]:
1763 if pd.isna(agent_id):
1764 agent_names.append(None)
1765 else:
1766 agent_names.append(str(agent_id))
1767 else:
1768 # Generate default agent names
1769 for i in range(len(df)):
1770 agent_names.append(f"Agent_{i+1}")
1772 dataset.agent_names = agent_names
1774 # Process agent comments (same as other methods)
1775 if agent_comments_column and agent_comments_column in df.columns:
1776 justifications = []
1777 for i, comment in enumerate(df[agent_comments_column]):
1778 # Include all comments, even empty ones, to maintain agent alignment
1779 agent_name = agent_names[i] if i < len(agent_names) else f"Agent_{i+1}"
1780 comment_text = str(comment).strip() if pd.notna(comment) else ""
1781 justifications.append({
1782 "agent_name": agent_name,
1783 "agent_index": i,
1784 "justification": comment_text
1785 })
1786 dataset.agent_justifications = justifications
1788 # Process agent attributes (same as other methods)
1789 if agent_attributes_columns:
1790 for attr_col in agent_attributes_columns:
1791 if attr_col in df.columns:
1792 attr_values = []
1793 for val in df[attr_col]:
1794 if pd.isna(val):
1795 attr_values.append(None)
1796 else:
1797 attr_values.append(str(val).strip())
1799 # Store in agent_attributes instead of key_results
1800 dataset.agent_attributes[attr_col] = attr_values
1802 @classmethod
1803 def _auto_detect_ranking_options(cls, ranking_series: pd.Series, separator: str) -> List[str]:
1804 """Auto-detect the ranking options from ordinal ranking data."""
1805 all_options = set()
1807 for ranking_str in ranking_series.dropna():
1808 if pd.isna(ranking_str):
1809 continue
1811 ranking_str = str(ranking_str).strip()
1812 if separator in ranking_str:
1813 options = [opt.strip() for opt in ranking_str.split(separator)]
1814 all_options.update(options)
1816 if not all_options:
1817 raise ValueError(f"No ranking options found in data using separator '{separator}'")
1819 # Sort options for consistency (could be enhanced to preserve meaningful order)
1820 return sorted(list(all_options))
1822 @classmethod
1823 def _parse_ordinal_rankings(cls, ranking_series: pd.Series, separator: str, options: List[str]) -> Dict[str, List[Optional[int]]]:
1824 """Parse ordinal ranking strings into individual option rankings."""
1825 result = {option: [] for option in options}
1827 for ranking_str in ranking_series:
1828 if pd.isna(ranking_str) or str(ranking_str).strip() == "":
1829 # Handle missing data
1830 for option in options:
1831 result[option].append(None)
1832 continue
1834 ranking_str = str(ranking_str).strip()
1836 if separator not in ranking_str:
1837 # Handle malformed data
1838 for option in options:
1839 result[option].append(None)
1840 continue
1842 # Parse the ranking
1843 ranked_options = [opt.strip() for opt in ranking_str.split(separator)]
1845 # Create rank mapping (position in list = rank, starting from 1)
1846 option_to_rank = {}
1847 for rank, option in enumerate(ranked_options, 1):
1848 if option in options:
1849 option_to_rank[option] = rank
1851 # Fill in ranks for each option
1852 for option in options:
1853 rank = option_to_rank.get(option, None)
1854 result[option].append(rank)
1856 return result
1858 @classmethod
1859 def create_from_csv(cls,
1860 file_path: Union[str, Path],
1861 experimental_data_type: str = "single_value_per_agent",
1862 agent_id_column: Optional[str] = None,
1863 agent_comments_column: Optional[str] = None,
1864 agent_attributes_columns: Optional[List[str]] = None,
1865 value_column: Optional[str] = None,
1866 ranking_columns: Optional[List[str]] = None,
1867 ordinal_ranking_column: Optional[str] = None,
1868 ordinal_ranking_separator: str = "-",
1869 ordinal_ranking_options: Optional[List[str]] = None,
1870 dataset_name: Optional[str] = None,
1871 dataset_description: Optional[str] = None,
1872 encoding: str = "utf-8") -> tuple['SimulationExperimentEmpiricalValidator', 'SimulationExperimentDataset']:
1873 """
1874 Create a validator and load empirical data from CSV in one step.
1876 This is a convenience method that combines validator creation with CSV loading.
1878 Args:
1879 Same as read_empirical_data_from_csv()
1881 Returns:
1882 Tuple of (validator_instance, loaded_dataset)
1883 """
1884 validator = cls()
1885 dataset = cls.read_empirical_data_from_csv(
1886 file_path=file_path,
1887 experimental_data_type=experimental_data_type,
1888 agent_id_column=agent_id_column,
1889 agent_comments_column=agent_comments_column,
1890 agent_attributes_columns=agent_attributes_columns,
1891 value_column=value_column,
1892 ranking_columns=ranking_columns,
1893 ordinal_ranking_column=ordinal_ranking_column,
1894 ordinal_ranking_separator=ordinal_ranking_separator,
1895 ordinal_ranking_options=ordinal_ranking_options,
1896 dataset_name=dataset_name,
1897 dataset_description=dataset_description,
1898 encoding=encoding
1899 )
1900 return validator, dataset
1902 @classmethod
1903 def create_from_dataframe(cls,
1904 df: pd.DataFrame,
1905 experimental_data_type: str = "single_value_per_agent",
1906 agent_id_column: Optional[str] = None,
1907 agent_comments_column: Optional[str] = None,
1908 agent_attributes_columns: Optional[List[str]] = None,
1909 value_column: Optional[str] = None,
1910 ranking_columns: Optional[List[str]] = None,
1911 ordinal_ranking_column: Optional[str] = None,
1912 ordinal_ranking_separator: str = "-",
1913 ordinal_ranking_options: Optional[List[str]] = None,
1914 dataset_name: Optional[str] = None,
1915 dataset_description: Optional[str] = None) -> tuple['SimulationExperimentEmpiricalValidator', 'SimulationExperimentDataset']:
1916 """
1917 Create a validator and load empirical data from a pandas DataFrame in one step.
1919 This is a convenience method that combines validator creation with DataFrame loading.
1921 Args:
1922 Same as read_empirical_data_from_dataframe()
1924 Returns:
1925 Tuple of (validator_instance, loaded_dataset)
1926 """
1927 validator = cls()
1928 dataset = cls.read_empirical_data_from_dataframe(
1929 df=df,
1930 experimental_data_type=experimental_data_type,
1931 agent_id_column=agent_id_column,
1932 agent_comments_column=agent_comments_column,
1933 agent_attributes_columns=agent_attributes_columns,
1934 value_column=value_column,
1935 ranking_columns=ranking_columns,
1936 ordinal_ranking_column=ordinal_ranking_column,
1937 ordinal_ranking_separator=ordinal_ranking_separator,
1938 ordinal_ranking_options=ordinal_ranking_options,
1939 dataset_name=dataset_name,
1940 dataset_description=dataset_description
1941 )
1942 return validator, dataset
1944 def _extract_effect_size(self, metric_result: Dict[str, Any]) -> Optional[float]:
1945 """Extract effect size from statistical test result, regardless of test type."""
1946 # Cohen's d for t-tests (most common)
1947 if "effect_size" in metric_result:
1948 return metric_result["effect_size"]
1950 # For tests that don't provide Cohen's d, calculate standardized effect size
1951 test_type = metric_result.get("test_type", "").lower()
1953 if "t-test" in test_type:
1954 # For t-tests, effect_size should be Cohen's d
1955 return metric_result.get("effect_size", 0.0)
1957 elif "mann-whitney" in test_type:
1958 # For Mann-Whitney, use Common Language Effect Size (CLES)
1959 # Convert CLES to Cohen's d equivalent: d ≈ 2 * Φ^(-1)(CLES)
1960 cles = metric_result.get("effect_size", 0.5)
1961 # Simple approximation: convert CLES to d-like measure
1962 # CLES of 0.5 = no effect, CLES of 0.71 ≈ small effect (d=0.2)
1963 return 2 * (cles - 0.5)
1965 elif "anova" in test_type:
1966 # For ANOVA, use eta-squared and convert to Cohen's d equivalent
1967 eta_squared = metric_result.get("effect_size", 0.0)
1968 # Convert eta-squared to Cohen's d: d = 2 * sqrt(eta^2 / (1 - eta^2))
1969 if eta_squared > 0 and eta_squared < 1:
1970 return 2 * (eta_squared / (1 - eta_squared)) ** 0.5
1971 return 0.0
1973 elif "chi-square" in test_type:
1974 # For Chi-square, use Cramer's V and convert to Cohen's d equivalent
1975 cramers_v = metric_result.get("effect_size", 0.0)
1976 # Rough conversion: d ≈ 2 * Cramer's V
1977 return 2 * cramers_v
1979 elif "kolmogorov-smirnov" in test_type or "ks" in test_type:
1980 # For KS test, the effect size is the KS statistic itself
1981 # It represents the maximum difference between CDFs (0 to 1)
1982 return metric_result.get("effect_size", metric_result.get("ks_statistic", 0.0))
1984 # Fallback: try to calculate from means and standard deviations
1985 if all(k in metric_result for k in ["control_mean", "treatment_mean", "control_std", "treatment_std"]):
1986 control_mean = metric_result["control_mean"]
1987 treatment_mean = metric_result["treatment_mean"]
1988 control_std = metric_result["control_std"]
1989 treatment_std = metric_result["treatment_std"]
1991 # Calculate pooled standard deviation
1992 pooled_std = ((control_std ** 2 + treatment_std ** 2) / 2) ** 0.5
1993 if pooled_std > 0:
1994 return abs(treatment_mean - control_mean) / pooled_std
1996 # If all else fails, return 0 (no effect)
1997 return 0.0
1999 def _interpret_effect_size(self, effect_size: float, test_type: str = "") -> str:
2000 """Provide interpretation of effect size magnitude based on test type."""
2001 test_type_lower = test_type.lower()
2003 # For KS test, use different thresholds since KS statistic ranges 0-1
2004 if "kolmogorov-smirnov" in test_type_lower or "ks" in test_type_lower:
2005 if effect_size < 0.1:
2006 return "negligible difference"
2007 elif effect_size < 0.25:
2008 return "small difference"
2009 elif effect_size < 0.5:
2010 return "medium difference"
2011 else:
2012 return "large difference"
2014 # For other tests, use Cohen's conventions
2015 if effect_size < 0.2:
2016 return "negligible"
2017 elif effect_size < 0.5:
2018 return "small"
2019 elif effect_size < 0.8:
2020 return "medium"
2021 else:
2022 return "large"
2025def validate_simulation_experiment_empirically(control_data: Dict[str, Any],
2026 treatment_data: Dict[str, Any],
2027 validation_types: List[str] = ["statistical", "semantic"],
2028 statistical_test_type: str = "welch_t_test",
2029 significance_level: float = 0.05,
2030 output_format: str = "values") -> Union[SimulationExperimentEmpiricalValidationResult, str]:
2031 """
2032 Convenience function to validate simulation experiment data against empirical control data.
2034 This performs data-driven validation using statistical and semantic methods,
2035 distinct from LLM-based evaluations.
2037 Args:
2038 control_data: Dictionary containing control/empirical data
2039 treatment_data: Dictionary containing treatment/simulation experiment data
2040 validation_types: List of validation types to perform
2041 statistical_test_type: Type of statistical test ("welch_t_test", "ks_test", "mann_whitney", etc.)
2042 significance_level: Significance level for statistical tests
2043 output_format: "values" for SimulationExperimentEmpiricalValidationResult object, "report" for markdown report
2045 Returns:
2046 SimulationExperimentEmpiricalValidationResult object or markdown report string
2047 """
2048 # Use Pydantic's built-in parsing instead of from_dict
2049 control_dataset = SimulationExperimentDataset.model_validate(control_data)
2050 treatment_dataset = SimulationExperimentDataset.model_validate(treatment_data)
2052 validator = SimulationExperimentEmpiricalValidator()
2053 return validator.validate(
2054 control_dataset,
2055 treatment_dataset,
2056 validation_types=validation_types,
2057 statistical_test_type=statistical_test_type,
2058 significance_level=significance_level,
2059 output_format=output_format
2060 )