import os
import glob
import zipfile
import requests
import json
import pandas as pd
import shutil
import matplotlib.pyplot as plt
import numpy as np
import random
import cvxpy as cp
from scipy.optimize import minimize, LinearConstraint
def check_evaluation_files(download_cache_sample: bool, download_url):
report = {"cache_directory": None} # Initialize with None
cache_dir_path = "cache"
cache_zip_filename = "cache.zip" # Standard name for the zip file (local or downloaded)
if download_cache_sample:
print("Mode: Download sample cache.")
# Google Drive direct download link from ID: 1uCen9DfMcKuHOyc-Lhrxa_ZeEOuWJfXM
# 1. Clean up existing cache directory and/or target zip file before download
if os.path.isdir(cache_dir_path):
print(f"Removing existing cache directory: '{cache_dir_path}'")
try:
shutil.rmtree(cache_dir_path)
except OSError as e:
print(f"Error removing directory {cache_dir_path}: {e}. Aborting download process.")
return report # report["cache_directory"] is still None
if os.path.isfile(cache_zip_filename):
print(f"Removing existing local zip file named '{cache_zip_filename}' before download.")
try:
os.remove(cache_zip_filename)
except OSError as e:
print(f"Error removing existing zip file {cache_zip_filename}: {e}. Aborting download process.")
return report
# 2. Download the cache zip file
print(f"Downloading cache from Google Drive to '{cache_zip_filename}'...")
downloaded_file_path = None
try:
import gdown # Attempt to import gdown here
# Using fuzzy=True can help with some Google Drive links if the direct one has issues.
downloaded_file_path = gdown.download(download_url, cache_zip_filename, quiet=False, fuzzy=True)
if downloaded_file_path is None or not os.path.isfile(cache_zip_filename):
print(f"Error: Failed to download cache. 'gdown.download' did not confirm success or file '{cache_zip_filename}' not found post-download.")
# Clean up if a partial/empty file was created
if os.path.isfile(cache_zip_filename):
try: os.remove(cache_zip_filename)
except OSError: pass
return report
print(f"Successfully downloaded to '{cache_zip_filename}'.")
except ImportError:
print("Error: The 'gdown' library is not installed. Please install it (e.g., 'pip install gdown') to enable downloading the cache.")
return report
except Exception as e:
print(f"An error occurred during download: {e}")
if os.path.isfile(cache_zip_filename): # Clean up if download failed and left a file
try: os.remove(cache_zip_filename)
except OSError: pass
return report
# 3. Unzip the downloaded file
print(f"Attempting to unzip '{cache_zip_filename}' to '{cache_dir_path}'.")
try:
with zipfile.ZipFile(cache_zip_filename, 'r') as zip_ref:
zip_ref.extractall(cache_dir_path) # Extracts to 'cache/' directory
if os.path.isdir(cache_dir_path):
report["cache_directory"] = cache_dir_path
print(f"Successfully unzipped '{cache_zip_filename}' to '{cache_dir_path}'.")
else:
print(f"Error: Directory '{cache_dir_path}' not found after unzipping.")
except zipfile.BadZipFile:
print(f"Error: Downloaded file '{cache_zip_filename}' is not a valid zip file or is corrupted.")
except Exception as e:
print(f"An error occurred during unzipping: {e}")
finally:
# Clean up the downloaded zip file after attempting to extract
if os.path.isfile(cache_zip_filename):
print(f"Removing downloaded zip file: '{cache_zip_filename}'")
try:
os.remove(cache_zip_filename)
except OSError as e:
print(f"Error removing downloaded zip file '{cache_zip_filename}': {e}")
else: # download_cache_sample is False - check local files
print("Mode: Check local cache.")
if os.path.isdir(cache_dir_path):
report["cache_directory"] = cache_dir_path
print(f"Using existing local cache directory: '{cache_dir_path}'")
elif os.path.isfile(cache_zip_filename):
print(f"Found local cache zip file: '{cache_zip_filename}'. Attempting to unzip.")
# If 'cache' directory exists (e.g. from partial previous attempt), remove it before unzipping.
if os.path.isdir(cache_dir_path):
print(f"Removing existing directory '{cache_dir_path}' before unzipping from local '{cache_zip_filename}'.")
try:
shutil.rmtree(cache_dir_path)
except OSError as e:
print(f"Error removing directory {cache_dir_path}: {e}. Cannot proceed with unzipping local zip.")
return report
# Handle if cache_dir_path is a file (conflicting name)
elif os.path.exists(cache_dir_path) and not os.path.isdir(cache_dir_path):
print(f"Found a file/link at '{cache_dir_path}' which conflicts with the target directory name. Removing it.")
try:
os.remove(cache_dir_path)
except OSError as e:
print(f"Error removing conflicting file/link {cache_dir_path}: {e}. Cannot unzip local zip.")
return report
try:
with zipfile.ZipFile(cache_zip_filename, 'r') as zip_ref:
zip_ref.extractall(cache_dir_path) # Extracts to 'cache/'
if os.path.isdir(cache_dir_path):
report["cache_directory"] = cache_dir_path
print(f"Successfully unzipped local '{cache_zip_filename}' to '{cache_dir_path}'.")
else:
print(f"Error: Failed to create directory '{cache_dir_path}' after unzipping local zip.")
except zipfile.BadZipFile:
print(f"Error: Local '{cache_zip_filename}' is not a valid zip file or is corrupted.")
except Exception as e:
print(f"An error occurred during unzipping local zip: {e}")
else:
print(f"Info: No local cache directory '{cache_dir_path}' or local cache zip file '{cache_zip_filename}' found.")
# Final status message based on whether 'cache_dir_path' ("cache") is set
if report["cache_directory"] == cache_dir_path:
print(f"Final Result: Cache is available at '{report['cache_directory']}'.")
else:
# Ensure report["cache_directory"] is None if not successfully set to cache_dir_path
report["cache_directory"] = None
print("Final Result: No cache directory named 'cache' is available.")
return report
def generate_evaluation_map(file_info):
"""
Generates a map from agent_config.json paths to a list of corresponding episode_*.jsonl paths,
handling nested subdirectories in cache. Now only processes cache.
Args:
file_info (dict): The report from check_evaluation_files.
Returns:
dict: A map where keys are paths to agent_config.json files and
values are lists of paths to corresponding episode_*.jsonl files.
"""
run_map = {}
cache_location = file_info.get("cache_directory")
if cache_location:
print(f"Info: Attempting to generate map from cache location: {cache_location}")
if os.path.isdir(cache_location):
print(f"Info: Processing cache directory: {cache_location}")
for dirpath, dirnames, filenames in os.walk(cache_location):
if "agent_config.json" in filenames:
current_agent_config = os.path.join(dirpath, "agent_config.json")
episode_log_pattern = os.path.join(dirpath, "episode_*.jsonl")
episode_logs = sorted(glob.glob(episode_log_pattern))
if episode_logs:
run_map[current_agent_config] = episode_logs
print(f" Mapped in dir '{dirpath}': '{current_agent_config}' to {len(episode_logs)} log(s)")
elif os.path.isfile(cache_location) and cache_location.endswith(".zip"):
# This part of the logic might need review if cache_location from file_info
# is always expected to be a directory after check_evaluation_files runs.
# For now, keeping the zip processing logic if a .zip path is somehow provided.
print(f"Info: Processing cache zip file: {cache_location}")
files_by_internal_dir = {}
try:
with zipfile.ZipFile(cache_location, 'r') as zip_f:
for item_name_in_zip in zip_f.namelist():
normalized_item_path = item_name_in_zip.replace('\\\\', '/')
if normalized_item_path.endswith('/'):
continue
internal_dir_path = os.path.dirname(normalized_item_path)
internal_filename = os.path.basename(normalized_item_path)
if internal_dir_path not in files_by_internal_dir:
files_by_internal_dir[internal_dir_path] = []
files_by_internal_dir[internal_dir_path].append(internal_filename)
for internal_dir, filenames_in_dir in files_by_internal_dir.items():
if "agent_config.json" in filenames_in_dir:
agent_config_zip_path = f"{internal_dir}/agent_config.json" if internal_dir else "agent_config.json"
episode_logs_zip_paths = []
for fname in filenames_in_dir:
if fname.startswith("episode_") and fname.endswith(".jsonl"):
log_path_in_zip = f"{internal_dir}/{fname}" if internal_dir else fname
episode_logs_zip_paths.append(log_path_in_zip)
if episode_logs_zip_paths:
run_map[agent_config_zip_path] = sorted(episode_logs_zip_paths)
print(f" Mapped in zip (dir '{internal_dir}'): '{agent_config_zip_path}' to {len(episode_logs_zip_paths)} log(s)")
except zipfile.BadZipFile:
print(f"Error: Could not read zip file {cache_location}. It may be corrupted.")
except Exception as e:
print(f"An error occurred while processing zip file {cache_location}: {e}")
if run_map:
print(f"Info: Map generation from cache completed.")
else:
# This warning will trigger if cache_location was valid but no mappings were found.
print(f"Warning: Cache location '{cache_location}' was processed, but no valid agent_config to episode_log mappings were found.")
else:
# This implies file_info did not contain a "cache_directory"
print("Info: No cache directory provided in file_info. No map will be generated from cache.")
if not run_map:
print("Warning: No evaluation map could be generated with the current settings and files.")
return run_map
def _try_parse_json_string_for_function(value, field_name_for_error, log_file_basename, line_num_str):
"""
Helper to parse a field that might be a JSON string.
Used by the process_evaluation_run_map function.
"""
if isinstance(value, str):
try:
return json.loads(value)
except json.JSONDecodeError:
return value # Return original string if parsing fails
return value # Return as is if not a string
def process_evaluation_run_map(evaluation_run_map):
"""
Processes an evaluation_run_map to extract and structure data from
agent_config.json and episode_*.jsonl files.
Args:
evaluation_run_map (dict): A map where keys are paths to agent_config.json
files and values are lists of paths to
corresponding episode_*.jsonl files.
Returns:
dict: A dictionary where keys are tuples (game_name, model_name,
observation_mode, harness_bool) and values are lists of
dictionaries, each representing a processed episode.
Each episode dictionary contains:
'episode_id', 'total_steps', 'total_reward',
'total_perf_score', 'total_time_taken',
'agent_observations' (list), 'infos' (list).
"""
grouped_results = {}
if not evaluation_run_map:
print("Warning: The provided evaluation_run_map is empty. No processing will occur.")
return grouped_results
for agent_config_filepath, episode_log_filepaths in evaluation_run_map.items():
# 1. Read and parse agent_config.json
game_name = None
model_name = None
observation_mode = None
harness_bool = None # Store harness as boolean
if not os.path.isfile(agent_config_filepath):
print(f"Error: Agent config file not found at '{agent_config_filepath}'. Skipping associated logs.")
continue
try:
with open(agent_config_filepath, 'r') as f_agent_config:
agent_config_data = json.load(f_agent_config)
game_name = agent_config_data.get("game_name")
model_name = agent_config_data.get("model_name")
observation_mode = agent_config_data.get("observation_mode")
harness_bool = agent_config_data.get("harness") # This should ideally be boolean (True/False)
if harness_bool is None: # Handle if 'harness' key is missing
print(f"Warning: 'harness' key missing in {agent_config_filepath}. Assuming False.")
harness_bool = False
except json.JSONDecodeError:
print(f"Error: Could not decode JSON from agent config: '{agent_config_filepath}'. Skipping.")
continue
except Exception as e:
print(f"Error reading agent config '{agent_config_filepath}': {e}. Skipping.")
continue
# Create the grouping key
# Ensure all parts of the key are actual values to avoid issues with None in dict keys
if game_name is None or model_name is None or observation_mode is None:
print(f"Warning: Missing one or more key identifiers (game_name, model_name, observation_mode) in '{agent_config_filepath}'. Skipping this config and its logs.")
continue
group_key = (game_name, model_name, observation_mode, harness_bool)
if group_key not in grouped_results:
grouped_results[group_key] = []
# 2. Process corresponding episode logs
if not episode_log_filepaths:
print(f"Info: No episode logs listed for agent config: '{agent_config_filepath}'.")
continue
for episode_log_path in episode_log_filepaths:
if not os.path.isfile(episode_log_path):
print(f"Warning: Episode log file not found: '{episode_log_path}'. Skipping.")
continue
log_basename = os.path.basename(episode_log_path)
try:
episode_id_str = log_basename.replace("episode_", "").replace("_log.jsonl", "")
except Exception:
episode_id_str = log_basename # Fallback
ep_total_steps = 0
ep_total_reward = 0
ep_total_perf_score = 0.0
ep_infos_list = []
ep_agent_observations_list = []
episode_lines_data = []
sum_step_time_taken_s = 0.0
found_step_time_taken_s_in_episode = False
final_total_time_taken_from_last_step_info = None
try:
with open(episode_log_path, 'r') as f_episode_log:
for line_num, line in enumerate(f_episode_log):
try:
step_data = json.loads(line)
episode_lines_data.append(step_data)
except json.JSONDecodeError:
# print(f"Warning: Could not decode JSON from line {line_num + 1} in '{log_basename}'. Skipping line.")
continue
if not episode_lines_data:
# print(f"Warning: No valid JSON lines found in '{log_basename}'. Skipping this log.")
continue
ep_total_steps = len(episode_lines_data)
for step_idx, step_data_dict in enumerate(episode_lines_data):
ep_total_reward += step_data_dict.get("reward", 0)
ep_total_perf_score += step_data_dict.get("perf_score", 0.0)
raw_obs = step_data_dict.get("agent_observation")
parsed_obs = _try_parse_json_string_for_function(raw_obs, "agent_observation", log_basename, str(step_idx+1))
ep_agent_observations_list.append(parsed_obs)
raw_info = step_data_dict.get("info")
parsed_info = _try_parse_json_string_for_function(raw_info, "info", log_basename, str(step_idx+1))
ep_infos_list.append(parsed_info)
if "time_taken_s" in step_data_dict:
try:
sum_step_time_taken_s += float(step_data_dict["time_taken_s"])
found_step_time_taken_s_in_episode = True
except (ValueError, TypeError):
pass
if step_idx == ep_total_steps - 1: # Last step
if isinstance(parsed_info, dict) and "total_time_taken" in parsed_info:
try:
final_total_time_taken_from_last_step_info = float(parsed_info["total_time_taken"])
except (ValueError, TypeError):
pass # Keep it None if not a valid float
ep_total_time_taken = None
if final_total_time_taken_from_last_step_info is not None:
ep_total_time_taken = final_total_time_taken_from_last_step_info
elif found_step_time_taken_s_in_episode:
ep_total_time_taken = sum_step_time_taken_s
episode_data_point = {
"episode_id": episode_id_str,
"total_steps": ep_total_steps,
"total_reward": ep_total_reward,
"total_perf_score": ep_total_perf_score,
"total_time_taken": ep_total_time_taken,
"agent_observations": ep_agent_observations_list,
"infos": ep_infos_list
}
grouped_results[group_key].append(episode_data_point)
except Exception as e:
print(f"Error processing episode log '{episode_log_path}': {e}")
continue # Skip to next episode log file
return grouped_results
def calculate_average_performance(extracted_results_map):
"""
Calculates average performance metrics from the extracted_results_map.
Args:
extracted_results_map (dict): The output from process_evaluation_run_map,
where keys are tuples (game_name, model_name,
observation_mode, harness_bool) and values are
lists of episode data dictionaries.
Returns:
list: A list of dictionaries, where each dictionary represents a group
and its averaged 'total_steps', 'total_reward', 'total_perf_score',
along with the group identifiers and 'num_episodes'.
"""
averaged_data_list = []
if not extracted_results_map:
print("Warning: The provided extracted_results_map is empty. No averages will be calculated.")
return averaged_data_list
for group_key, episodes_list in extracted_results_map.items():
game_name, model_name, observation_mode, harness = group_key
if not episodes_list:
print(f"Info: No episodes found for group: {group_key}. Skipping averaging for this group.")
# Optionally, you could add an entry with 0 episodes and NaN/0 averages
averaged_data_list.append({
"game_name": game_name,
"model_name": model_name,
"observation_mode": observation_mode,
"harness": harness,
"num_episodes": 0,
"avg_total_steps": 0, # or float('nan')
"avg_total_reward": 0, # or float('nan')
"avg_total_perf_score": 0, # or float('nan')
"avg_total_time_taken": 0 # or float('nan')
})
continue
num_episodes = len(episodes_list)
sum_steps = 0
sum_reward = 0
sum_perf_score = 0
sum_time_taken = 0
episodes_with_time = 0
for episode_data in episodes_list:
sum_steps += episode_data.get("total_steps", 0)
sum_reward += episode_data.get("total_reward", 0)
sum_perf_score += episode_data.get("total_perf_score", 0.0)
time_taken = episode_data.get("total_time_taken")
if time_taken is not None:
sum_time_taken += time_taken
episodes_with_time += 1
avg_steps = sum_steps / num_episodes if num_episodes > 0 else 0
avg_reward = sum_reward / num_episodes if num_episodes > 0 else 0
avg_perf_score = sum_perf_score / num_episodes if num_episodes > 0 else 0
avg_time_taken = sum_time_taken / episodes_with_time if episodes_with_time > 0 else None
averaged_data_list.append({
"game_name": game_name,
"model_name": model_name,
"observation_mode": observation_mode,
"harness": harness, # This is already boolean from the group_key
"num_episodes": num_episodes,
"avg_total_steps": avg_steps,
"avg_total_reward": avg_reward,
"avg_total_perf_score": avg_perf_score,
"avg_total_time_taken": avg_time_taken
})
return averaged_data_list
def combine_benchmark_with_updates(df_benchmark, df_local_updates):
"""
Combines a benchmark DataFrame with local update DataFrame.
Updates existing models or adds new ones.
"""
if df_local_updates is None or len(df_local_updates) == 0:
print("No local updates provided. Returning benchmark DataFrame as is.")
return df_benchmark.copy()
if df_benchmark.empty:
print("Benchmark DataFrame is empty. Returning local updates DataFrame as is.")
return df_local_updates.copy()
# Set the multi-index for both DataFrames to easily update and combine
keys = ['model_name', 'harness', 'game_name']
# Ensure keys are present in both dataframes
for key_col in keys:
if key_col not in df_benchmark.columns:
print(f"Warning: Key column '{key_col}' not found in benchmark DataFrame. Returning local updates.")
return df_local_updates.copy()
if key_col not in df_local_updates.columns:
print(f"Warning: Key column '{key_col}' not found in local updates DataFrame. Returning benchmark data.")
return df_benchmark.copy()
# Index the DataFrames by the key columns
df_local_indexed = df_local_updates.set_index(keys)
df_benchmark_indexed = df_benchmark.set_index(keys)
# Update the benchmark data with local updates
df_combined_indexed = df_benchmark_indexed.copy()
df_combined_indexed.update(df_local_indexed)
# Add rows that are entirely new in local updates
new_indices = df_local_indexed.index.difference(df_benchmark_indexed.index)
if not new_indices.empty:
df_combined_indexed = df_combined_indexed.append(df_local_indexed.loc[new_indices])
# Reset the index to return the DataFrame to its original form
df_combined = df_combined_indexed.reset_index()
print(f"Combined {len(df_local_updates)} local updates with benchmark data.")
return df_combined
def update_benchmark_average(df_benchmark_average, df_update_model_average):
"""
Updates benchmark average data with new model averages.
If model exists, update its scores. If not, add it.
"""
if df_update_model_average is None or df_update_model_average.empty:
print("No model update averages provided. Returning benchmark averages as is.")
return df_benchmark_average.copy()
if df_benchmark_average.empty:
print("Benchmark DataFrame is empty. Returning local updates DataFrame as is.")
return df_update_model_average.copy()
# Set the multi-index for both DataFrames to easily update and combine
keys = ['model_name', 'harness']
# Ensure keys are present in both dataframes
for key_col in keys:
if key_col not in df_benchmark_average.columns:
print(f"Warning: Key column '{key_col}' not found in benchmark DataFrame. Returning local updates.")
return df_update_model_average.copy()
if key_col not in df_update_model_average.columns:
print(f"Warning: Key column '{key_col}' not found in local updates DataFrame. Returning benchmark data.")
return df_benchmark_average.copy()
# Index the DataFrames by the key columns
df_local_indexed = df_update_model_average.set_index(keys)
df_benchmark_indexed = df_benchmark_average.set_index(keys)
# Update the benchmark data with local updates
df_combined_indexed = df_benchmark_indexed.copy()
df_combined_indexed.update(df_local_indexed)
# Add rows that are entirely new in local updates
new_indices = df_local_indexed.index.difference(df_benchmark_indexed.index)
if not new_indices.empty:
df_combined_indexed = df_combined_indexed.append(df_local_indexed.loc[new_indices])
# Reset the index to return the DataFrame to its original form
df_combined = df_combined_indexed.reset_index()
print(f"Updated benchmark averages with {len(df_update_model_average)} model averages.")
return df_combined
def visualize_model_scores(score_dict, title="Model Performance Distribution"):
"""
Visualizes model scores distribution in a bar chart.
Args:
score_dict: Dictionary with model names as keys and scores as values
title: Title for the plot
"""
if not score_dict:
print("No scores to visualize.")
return
models = list(score_dict.keys())
scores = list(score_dict.values())
plt.figure(figsize=(12, 6))
bars = plt.bar(models, scores)
# Add value labels on bars
for i, (bar, score) in enumerate(zip(bars, scores)):
if isinstance(score, (int, float)) and not np.isnan(score):
plt.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.5,
f'{score:.2f}', ha='center', va='bottom')
plt.title(title)
plt.xlabel('Models')
plt.ylabel('Score')
plt.xticks(rotation=45, ha='right')
plt.tight_layout()
plt.show()
def polynomial_analysis(x_values, y_values, max_degree=3):
"""
Performs polynomial regression analysis.
Args:
x_values: List of x values
y_values: List of corresponding y values
max_degree: Maximum degree of polynomial to test
Returns:
Dictionary with coefficients and R² scores for each degree
"""
from sklearn.preprocessing import PolynomialFeatures
from sklearn.linear_model import LinearRegression
from sklearn.metrics import r2_score
x_array = np.array(x_values).reshape(-1, 1)
y_array = np.array(y_values)
results = {}
for degree in range(1, max_degree + 1):
# Create polynomial features
poly_features = PolynomialFeatures(degree=degree)
x_poly = poly_features.fit_transform(x_array)
# Fit model
model = LinearRegression()
model.fit(x_poly, y_array)
# Predict and calculate R²
y_pred = model.predict(x_poly)
r2 = r2_score(y_array, y_pred)
results[degree] = {
'coefficients': model.coef_,
'intercept': model.intercept_,
'r2_score': r2
}
return results
def factorize_polynomial(coefficients):
"""
Attempts to factorize a polynomial given its coefficients.
Args:
coefficients: List of polynomial coefficients in ascending order of powers
Returns:
Factored form representation as string
"""
# Remove leading zeros
coeffs = np.array(coefficients)
coeffs = coeffs[coeffs != 0] if len(coeffs) > 1 else coeffs
if len(coeffs) <= 1:
return f"{coeffs[0] if len(coeffs) > 0 else 0}"
# Find roots of the polynomial
try:
roots = np.roots(coeffs[::-1]) # np.roots expects coefficients in descending order
# Check if roots are real
real_roots = []
for root in roots:
if np.isreal(root):
real_roots.append(np.real(root))
if len(real_roots) == len(roots):
# All roots are real, can factorize
factored = f"{coeffs[-1]}" # Leading coefficient
for root in real_roots:
if root >= 0:
factored += f" * (x - {root:.3f})"
else:
factored += f" * (x + {abs(root):.3f})"
return factored
else:
return "Cannot factorize: complex roots present"
except Exception as e:
return f"Factorization failed: {str(e)}"
def normalize_column_safely(column_series):
numeric_series = pd.to_numeric(column_series, errors='coerce')
min_val = numeric_series.min()
max_val = numeric_series.max()
if pd.isna(min_val) or pd.isna(max_val) or min_val == max_val:
result_series = pd.Series([pd.NA] * len(numeric_series), index=numeric_series.index, dtype=float)
valid_numeric_mask = numeric_series.notna()
result_series[valid_numeric_mask] = 1.0 # Default to 1 if no range or all same
return result_series
else:
return (numeric_series - min_val) / (max_val - min_val) * 99 + 1
def hex_to_rgba(hex_color, alpha=0.2):
hex_color = hex_color.lstrip('#')
r = int(hex_color[0:2], 16)
g = int(hex_color[2:4], 16)
b = int(hex_color[4:6], 16)
return f'rgba({r}, {g}, {b}, {alpha})'
def get_random_color(existing_colors_hex: set) -> str:
"""Generates a random hex color string, trying to avoid existing ones."""
attempts = 0
while attempts < 100:
r = random.randint(0, 255)
g = random.randint(0, 255)
b = random.randint(0, 255)
hex_color = f'#{r:02x}{g:02x}{b:02x}'
if hex_color not in existing_colors_hex:
return hex_color
attempts += 1
return f'#{random.randint(0,255):02x}{random.randint(0,255):02x}{random.randint(0,255):02x}'
def prepare_dataframe_for_plots(source_df: pd.DataFrame, selected_games: list, selected_models: list, harness_bool_to_use: bool) -> pd.DataFrame:
if source_df.empty:
print("Input source_df is empty. Cannot prepare for plots.")
return pd.DataFrame()
# Filter by harness, games, and models
df_filtered = source_df[
(source_df['harness'] == harness_bool_to_use) &
(source_df['game_name'].isin(selected_games)) &
(source_df['model_name'].isin(selected_models))
].copy()
if df_filtered.empty:
print(f"Warning: No data found for harness={harness_bool_to_use}, selected games and models.")
return pd.DataFrame()
# Pivot the data
try:
df_pivot = df_filtered.pivot_table(
index='model_name', columns='game_name', values='avg_score_from_rank'
).reset_index().rename(columns={'model_name': 'Player'})
except Exception as e:
print(f"Error during pivoting: {e}")
return pd.DataFrame()
if df_pivot.empty:
print(f"Pivoted DataFrame is empty for harness={harness_bool_to_use}.")
return pd.DataFrame()
# Normalize each game column using the provided function
for game_name in selected_games:
if game_name in df_pivot.columns:
df_pivot[game_name] = normalize_column_safely(df_pivot[game_name])
return df_pivot
def create_comparison_radar_chart(df: pd.DataFrame, model_colors: dict, selected_games: list, harness_status_str: str, highlight_models: list = None):
import plotly.graph_objects as go
if df.empty:
return go.Figure().update_layout(title_text=f"No data for Radar Chart ({harness_status_str})")
# Check available game columns
available_games = [game for game in selected_games if game in df.columns]
if not available_games:
return go.Figure().update_layout(title_text=f"No game columns for Radar Chart ({harness_status_str})")
# Fill NaN values with 1
for game in available_games:
df[game] = df[game].fillna(1)
fig = go.Figure()
if 'Player' not in df.columns:
print("Error: 'Player' column not found for radar chart.")
return fig
sorted_players = sorted(df['Player'].unique())
for player in sorted_players:
player_row_df = df[df['Player'] == player]
if player_row_df.empty:
continue
player_row = player_row_df.iloc[0]
r_values = [player_row.get(game, 1) for game in available_games]
r_values = [val if pd.notna(val) else 1 for val in r_values]
is_highlighted = highlight_models and player in highlight_models
model_color_hex = model_colors.get(player, '#808080')
if not isinstance(model_color_hex, str) or not model_color_hex.startswith('#'):
model_color_hex = '#808080'
line_props = dict(color='red', width=3) if is_highlighted else dict(color=model_color_hex, width=1.5)
marker_props = dict(color='red', size=8, line=dict(color='darkred', width=2)) if is_highlighted else dict(color=model_color_hex, size=4, line=dict(color='#B0B0B0', width=1))
fill_color = 'rgba(255,0,0,0.4)' if is_highlighted else hex_to_rgba(model_color_hex, 0.2)
opacity = 0.9 if is_highlighted else 0.7
fig.add_trace(go.Scatterpolar(
r=r_values + [r_values[0]],
theta=available_games + [available_games[0]],
mode='lines+markers',
name=player,
line=line_props,
marker=marker_props,
fill='toself',
fillcolor=fill_color,
opacity=opacity,
hovertemplate=f'{player}
Game: %{{theta}}
Score: %{{r:.2f}} '
))
fig.update_layout(
title_text=f'Model Performance Radar ({harness_status_str}) - Normalized Scores (1-100 Scale)',
title_x=0.5,
polar=dict(
radialaxis=dict(
visible=True,
range=[1, 100],
gridcolor='lightgray',
tickvals=list(range(10, 101, 10)),
tickformat=".0f"
),
angularaxis=dict(
tickfont=dict(size=10),
gridcolor='lightgray'
)
),
legend=dict(
title="Models",
orientation="v",
yanchor="top",
y=1.02,
xanchor="left",
x=1.05
),
width=800,
height=600,
margin=dict(l=100, r=200, t=100, b=50)
)
return fig
def create_comparison_bar_chart(df: pd.DataFrame, model_colors: dict, selected_games: list, harness_status_str: str, highlight_models: list = None):
import plotly.graph_objects as go
if df.empty:
return go.Figure().update_layout(title_text=f"No data for Bar Chart ({harness_status_str})")
# Check available game columns
available_games = [game for game in selected_games if game in df.columns]
if not available_games:
return go.Figure().update_layout(title_text=f"No game columns for Bar Chart ({harness_status_str})")
# Fill NaN values with 1
for game in available_games:
df[game] = df[game].fillna(1)
fig = go.Figure()
if 'Player' not in df.columns:
print("Error: 'Player' column not found for bar chart.")
return fig
sorted_players = sorted(df['Player'].unique())
for player_name in sorted_players:
player_row_df = df[df['Player'] == player_name]
if player_row_df.empty:
continue
player_row = player_row_df.iloc[0]
y_values = [player_row.get(game, 1) for game in available_games]
y_values = [val if pd.notna(val) else 1 for val in y_values]
model_color_hex = model_colors.get(player_name, '#808080')
if not isinstance(model_color_hex, str) or not model_color_hex.startswith('#'):
model_color_hex = '#808080'
is_highlighted = highlight_models and player_name in highlight_models
opacity = 1.0 if is_highlighted else 0.7
line_width = 2 if is_highlighted else 0
fig.add_trace(go.Bar(
name=player_name,
x=available_games,
y=y_values,
marker_color=model_color_hex,
opacity=opacity,
marker_line_width=line_width,
marker_line_color='red' if is_highlighted else '#333333',
hovertemplate=f'{player_name}
Game: %{{x}}
Score: %{{y:.2f}} '
))
fig.update_layout(
barmode='group',
title_text=f'Model Performance Comparison ({harness_status_str}) - Normalized Scores (1-100 Scale)',
title_x=0.5,
xaxis_title="Game",
yaxis_title="Normalized Score (1-100)",
legend_title_text='Models',
legend=dict(
orientation="v",
yanchor="top",
y=1.02,
xanchor="left",
x=1.02
),
width=max(800, 150 * len(available_games) * min(5, len(sorted_players)) + 250),
height=600,
margin=dict(l=50, r=200, t=100, b=50)
)
fig.update_yaxes(range=[0, 105])
return fig
def create_game_specific_horizontal_bar_charts(df: pd.DataFrame, model_colors: dict, selected_games: list, harness_status_str: str, highlight_models: list = None):
import plotly.graph_objects as go
from plotly.subplots import make_subplots
if df.empty:
print(f"No data for Game-Specific Bar Charts ({harness_status_str})")
return []
# Check available game columns
available_games = [game for game in selected_games if game in df.columns]
if not available_games:
print(f"No game columns for Game-Specific Bar Charts ({harness_status_str})")
return []
# Fill NaN values with 1
for game in available_games:
df[game] = df[game].fillna(1)
if 'Player' not in df.columns:
print("Error: 'Player' column not found for game-specific bar charts.")
return []
figures = []
# Loop through each game to create separate horizontal bar charts
for game_name in available_games:
# Get data for this specific game
game_data = []
for _, row in df.iterrows():
player_name = row['Player']
score = row.get(game_name, 1)
if pd.notna(score):
game_data.append({
'player': player_name,
'score': score
})
if not game_data:
print(f"No data for game: {game_name}")
continue
# Sort by score (high to low)
game_data.sort(key=lambda x: x['score'], reverse=True)
# Create horizontal bar chart for this game
fig = go.Figure()
players = [item['player'] for item in game_data]
scores = [item['score'] for item in game_data]
# Reverse the order so highest scores appear at the top
players_reversed = players[::-1]
scores_reversed = scores[::-1]
# Create bars with colors and highlighting - using the defined model colors
bar_colors = []
bar_opacities = []
bar_line_widths = []
bar_line_colors = []
for player in players_reversed:
# Get the color from the defined model_colors dictionary
model_color_hex = model_colors.get(player, '#808080') # Default gray if not found
# Ensure it's a valid hex color
if not isinstance(model_color_hex, str) or not model_color_hex.startswith('#'):
model_color_hex = '#808080'
is_highlighted = highlight_models and player in highlight_models
bar_colors.append(model_color_hex)
bar_opacities.append(1.0 if is_highlighted else 0.7)
bar_line_widths.append(3 if is_highlighted else 0)
bar_line_colors.append('red' if is_highlighted else model_color_hex) # Use model color for border too
# Debug print to verify colors are being used
# print(f"Player: {player}, Color: {model_color_hex}, Highlighted: {is_highlighted}")
fig.add_trace(go.Bar(
x=scores_reversed,
y=players_reversed,
orientation='h',
marker=dict(
color=bar_colors, # Using the defined colors from model_colors_for_plots
opacity=bar_opacities,
line=dict(
width=bar_line_widths,
color=bar_line_colors
)
),
hovertemplate='%{y}
Score: %{x:.2f}',
showlegend=False
))
fig.update_layout(
title_text=f'{game_name} Performance ({harness_status_str}) - Sorted High to Low',
title_x=0.5,
xaxis_title="Normalized Score (1-100)",
yaxis_title="Models",
width=800,
height=max(400, 40 * len(players) + 100), # Dynamic height based on number of players
margin=dict(l=150, r=50, t=80, b=50),
yaxis=dict(
categoryorder='array',
categoryarray=players_reversed # Reversed order so highest appears at top
)
)
fig.update_xaxes(range=[0, 105])
figures.append(fig)
return figures
def generate_model_performance_plots(final_updated_df: pd.DataFrame, average_results: pd.DataFrame = None, model_colors_url: str = None):
"""
Generates radar and bar charts for model performance comparison.
Args:
final_updated_df: DataFrame with columns ['model_name', 'harness', 'game_name', 'avg_score_from_rank']
average_results: Optional DataFrame with local results to highlight
model_colors_url: Optional URL to fetch model colors JSON
"""
import plotly.graph_objects as go
import requests
import json
import random
# Default model and game names
show_games = list(final_updated_df['game_name'].unique())
show_models = list(final_updated_df['model_name'].unique())
highlight_models = []
if average_results is not None and not average_results.empty:
if 'model_name' in average_results.columns:
highlight_models = list(average_results['model_name'].unique())
# Model colors configuration
model_colors_for_plots = {}
used_colors_from_json = set()
# Load model colors
if model_colors_url:
print(f"Attempting to load model colors from URL...")
try:
response = requests.get(model_colors_url)
response.raise_for_status()
model_colors_from_url = response.json()
model_colors_for_plots.update(model_colors_from_url)
used_colors_from_json.update(model_colors_from_url.values())
print(f"Successfully loaded model colors from URL. {len(model_colors_for_plots)} colors loaded.")
except:
print("Warning: Could not download model colors. Will use random colors for all models.")
# Assign colors to all models
if not final_updated_df.empty:
all_models_in_data = final_updated_df['model_name'].unique()
for model_name_from_data in all_models_in_data:
if model_name_from_data not in model_colors_for_plots:
new_color = get_random_color(used_colors_from_json)
model_colors_for_plots[model_name_from_data] = new_color
used_colors_from_json.add(new_color)
print(f"Models to highlight: {highlight_models}")
print(f"Games to show: {show_games}")
print(f"Models to show: {show_models}")
# Generate and show plots for both harness conditions
for harness_boolean_val, harness_title_str in [(True, "Harness True"), (False, "Harness False")]:
print(f"\n--- Generating plots for {harness_title_str} ---")
df_for_plotting = prepare_dataframe_for_plots(
source_df=final_updated_df,
selected_games=show_games,
selected_models=show_models,
harness_bool_to_use=harness_boolean_val
)
if df_for_plotting.empty:
print(f"No data to plot for {harness_title_str}. Skipping.")
continue
# Create and show radar chart
fig_radar = create_comparison_radar_chart(
df=df_for_plotting,
model_colors=model_colors_for_plots,
selected_games=show_games,
harness_status_str=harness_title_str,
highlight_models=highlight_models
)
print(f"Displaying Radar Chart for {harness_title_str}:")
fig_radar.show()
# Create and show game-specific horizontal bar charts
game_bar_figures = create_game_specific_horizontal_bar_charts(
df=df_for_plotting,
model_colors=model_colors_for_plots,
selected_games=show_games,
harness_status_str=harness_title_str,
highlight_models=highlight_models
)
for i, fig_bar in enumerate(game_bar_figures):
game_name = show_games[i] if i < len(show_games) else f"Game {i+1}"
print(f"Displaying Horizontal Bar Chart for {game_name} ({harness_title_str}):")
fig_bar.show()
else:
print("'final_updated_df' variable not found or is empty. Cannot generate plots.")
def convert_local_averages_to_dataframe(local_averaged_results_list: list) -> pd.DataFrame:
"""
Converts a list of local averaged results (dictionaries) into a Pandas DataFrame
with columns 'model_name', 'harness', 'game_name', and 'avg_score_from_rank'.
The 'avg_score_from_rank' column is populated from the 'avg_total_perf_score'
field of the input dictionaries.
"""
if not local_averaged_results_list:
print("Warning: The local_averaged_results_list is empty. Returning an empty DataFrame.")
return pd.DataFrame(columns=['model_name', 'harness', 'game_name', 'avg_score_from_rank'])
# Create a list of dictionaries with only the desired columns and new key name
formatted_list_for_df = []
for item in local_averaged_results_list:
if not isinstance(item, dict):
print(f"Warning: Skipping an item in local_averaged_results_list as it's not a dictionary: {item}")
continue
# Ensure all necessary keys are present in the item before trying to access them
required_keys = ['model_name', 'harness', 'game_name', 'avg_total_perf_score']
if not all(key in item for key in required_keys):
print(f"Warning: Skipping an item due to missing one or more required keys ({', '.join(required_keys)}): {item}")
continue
formatted_list_for_df.append({
"model_name": item['model_name'],
"harness": item['harness'], # Assuming this is already boolean
"game_name": item['game_name'],
"avg_score_from_rank": item['avg_total_perf_score'] # Mapping local score to this new column name
})
if not formatted_list_for_df:
print("Warning: No valid data could be extracted from local_averaged_results_list. Returning an empty DataFrame.")
return pd.DataFrame(columns=['model_name', 'harness', 'game_name', 'avg_score_from_rank'])
df_local_formatted = pd.DataFrame(formatted_list_for_df)
print(f"Created DataFrame from local averaged results with {len(df_local_formatted)} rows and columns: {list(df_local_formatted.columns)}")
return df_local_formatted
def load_and_average_benchmark_rank_data(url: str) -> pd.DataFrame:
"""
Fetches benchmark rank data from a URL, calculates average scores,
and returns it as a Pandas DataFrame.
The 'harness' column in the output DataFrame will be boolean.
'avg_score_from_rank' is the column with the calculated average from the benchmark file.
"""
benchmark_rank_data_dict = {}
print(f"Attempting to download benchmark data from: {url}")
try:
response = requests.get(url)
response.raise_for_status() # Raise an HTTPError for bad responses (4xx or 5xx)
benchmark_rank_data_dict = response.json() # requests can directly parse to dict
print("Benchmark data successfully downloaded and parsed.")
except requests.exceptions.RequestException as e:
print(f"Error: Could not download data from {url}. Exception: {e}")
return pd.DataFrame() # Return empty DataFrame on error
except json.JSONDecodeError as e: # response.json() can also raise this if content is not valid JSON
print(f"Error: Could not parse JSON from the URL response. Exception: {e}")
return pd.DataFrame() # Return empty DataFrame on error
except Exception as e: # Catch any other unexpected errors
print(f"An unexpected error occurred while fetching or loading benchmark data: {e}")
return pd.DataFrame()
# Calculate averages
rank_averages_list_of_dicts = []
if not benchmark_rank_data_dict:
print("Warning: Empty benchmark_rank_data_dict received. Cannot calculate averages from it.")
return pd.DataFrame()
for model_name, harness_data in benchmark_rank_data_dict.items():
if not isinstance(harness_data, dict):
# print(f"Skipping model '{model_name}' in benchmark data: harness_data is not a dict.")
continue
for harness_status_str, game_data in harness_data.items():
if not isinstance(game_data, dict):
# print(f"Skipping harness '{harness_status_str}' for model '{model_name}': game_data is not a dict.")
continue
# Convert harness status string to boolean for consistent merging
harness_bool = True if harness_status_str == "harness_true" else False
for game_name, scores_list in game_data.items():
avg_score = pd.NA # Use pd.NA for missing numeric data
if scores_list and isinstance(scores_list, list):
# Filter out non-numeric scores before averaging
numeric_scores = [s for s in scores_list if isinstance(s, (int, float))]
if numeric_scores: # Ensure there's at least one numeric score
avg_score = sum(numeric_scores) / len(numeric_scores)
# else:
# print(f"No numeric scores found for {model_name}/{harness_status_str}/{game_name} in benchmark.")
# else:
# print(f"Scores_list is empty or not a list for {model_name}/{harness_status_str}/{game_name} in benchmark.")
rank_averages_list_of_dicts.append({
"model_name": model_name,
"harness": harness_bool,
"game_name": game_name,
"avg_score_from_rank": avg_score # This will be the averaged score from the rank file
})
if not rank_averages_list_of_dicts:
print("No averages could be calculated from the benchmark rank data (list is empty).")
return pd.DataFrame()
df_benchmark_avg = pd.DataFrame(rank_averages_list_of_dicts)
print(f"Created DataFrame from benchmark rank averages with {len(df_benchmark_avg)} rows.")
return df_benchmark_avg
def combine_and_update_averages(df_benchmark: pd.DataFrame, df_local_updates: pd.DataFrame) -> pd.DataFrame:
"""
Combines benchmark averages with local update averages.
Local updates will override benchmark data for the same (model_name, harness, game_name) keys.
All unique entries from both DataFrames will be included.
Args:
df_benchmark: DataFrame with columns ['model_name', 'harness', 'game_name', 'avg_score_from_rank']
from the benchmark data.
df_local_updates: DataFrame with columns ['model_name', 'harness', 'game_name', 'avg_score_from_rank']
from the local averaged data (where 'avg_score_from_rank' holds local scores).
Returns:
A Pandas DataFrame with the combined and updated averages.
"""
if df_local_updates.empty and df_benchmark.empty:
print("Both benchmark and local update DataFrames are empty. Returning an empty DataFrame.")
return pd.DataFrame(columns=['model_name', 'harness', 'game_name', 'avg_score_from_rank'])
if df_local_updates.empty:
print("Local updates DataFrame is empty. Returning benchmark DataFrame as is.")
return df_benchmark.copy() # Return a copy to avoid modifying the original
if df_benchmark.empty:
print("Benchmark DataFrame is empty. Returning local updates DataFrame as is.")
return df_local_updates.copy()
# Set the multi-index for both DataFrames to easily update and combine
keys = ['model_name', 'harness', 'game_name']
# Ensure keys are present in both dataframes
for key_col in keys:
if key_col not in df_benchmark.columns:
print(f"Warning: Key column '{key_col}' not found in benchmark DataFrame. Returning local updates.")
return df_local_updates.copy()
if key_col not in df_local_updates.columns:
print(f"Warning: Key column '{key_col}' not found in local updates DataFrame. Returning benchmark data.")
return df_benchmark.copy()
df_benchmark_indexed = df_benchmark.set_index(keys)
df_local_updates_indexed = df_local_updates.set_index(keys)
# Update the benchmark data with local data.
# Rows in local_updates will overwrite rows in benchmark_indexed if the index matches.
# Rows in local_updates not in benchmark_indexed will be added.
df_combined_indexed = df_benchmark_indexed.copy() # Start with a copy of benchmark
df_combined_indexed.update(df_local_updates_indexed) # Update with local, overwrites existing, does not add new from local
# Add new rows from local_updates_indexed that were not in df_benchmark_indexed
# This requires finding rows in local that are not in benchmark and concatenating.
# A more straightforward way for 'outer join' like behavior with local priority is:
# Re-think: A simpler approach for "local overrides, then add unique from benchmark"
# 1. Take all local data.
# 2. Find benchmark data that is NOT in local (based on keys) and append it.
# Set index on local updates
df_local_final = df_local_updates.set_index(keys)
# Filter benchmark data to get only rows NOT present in local updates
# This uses the index of df_local_final to exclude matching rows from df_benchmark
df_benchmark_only = df_benchmark.set_index(keys)[~df_benchmark.set_index(keys).index.isin(df_local_final.index)]
# Concatenate local data (which has priority) with the unique benchmark data
df_final_combined = pd.concat([df_local_final, df_benchmark_only])
# Reset index to get 'model_name', 'harness', 'game_name' back as columns
df_final_combined = df_final_combined.reset_index()
print(f"Combined DataFrame created. Local updates took priority. Final rows: {len(df_final_combined)}")
return df_final_combined
def run_polynomial_analysis_notebook(
final_updated_df: pd.DataFrame,
OTHER_TASK_RANK_URL: str,
DEFAULT_MODEL_MATCH: list = None,
selected_games: list = None,
selected_models: list = None,
harness_filter: bool = None,
max_degree: int = 3
):
"""
Comprehensive polynomial analysis function that handles model ranking data properly.
Modified to work seamlessly without file I/O.
Args:
final_updated_df: DataFrame with model performance data
OTHER_TASK_RANK_URL: URL to fetch other task ranking data
DEFAULT_MODEL_MATCH: List of default models for analysis
selected_games: List of games to analyze (if None, uses all games)
selected_models: List of models to analyze (if None, uses DEFAULT_MODEL_MATCH)
harness_filter: Boolean to filter by harness (if None, analyzes both)
max_degree: Maximum polynomial degree to test
Returns:
Dictionary with comprehensive polynomial analysis results
"""
from itertools import product
from scipy.stats import rankdata
from sklearn.preprocessing import PolynomialFeatures
from sklearn.linear_model import LinearRegression
from sklearn.metrics import r2_score
import requests
import json
# Use provided DEFAULT_MODEL_MATCH or fallback to predefined list
if DEFAULT_MODEL_MATCH is None:
DEFAULT_MODELS_FOR_OTHER_TASKS = [
"claude-3-5-sonnet-20241022",
"claude-3-7-sonnet-20250219 (thinking)",
"gemini-2.5-pro-preview-05-06",
"llama-4-maverick-17b-128e-instruct-fp8",
"gpt-4o-2024-11-20",
"o1-2024-12-17",
"o3-2025-04-16",
"o4-mini-2025-04-16"
]
else:
DEFAULT_MODELS_FOR_OTHER_TASKS = DEFAULT_MODEL_MATCH
def build_batches(rank_matrices):
batches = []
valid_matrices = [m for m in rank_matrices if isinstance(m, np.ndarray) and m.shape[0] > 0]
if len(valid_matrices) != len(rank_matrices):
print("Warning: Invalid/empty rank matrices for build_batches. Using valid ones.")
if not valid_matrices: return []
rank_matrices = valid_matrices
for row_idxs in product(*(range(m.shape[0]) for m in rank_matrices)):
batches.append(np.stack([m[i] for m, i in zip(rank_matrices, row_idxs)], axis=0).T)
return batches
class RankingPredictor:
def __init__(self, degree: int, lr: float = 1e-4, epochs: int = 1000, shuffle: bool = True, scale: float = 8.0):
self.degree, self.lr, self.epochs, self.shuffle, self.scale = degree, lr, epochs, shuffle, scale
self._phi = PolynomialFeatures(degree=degree, include_bias=True)
self._w = None
def _transform(self, R, fit=False):
return self._phi.fit_transform(R / self.scale) if fit else self._phi.transform(R / self.scale)
def fit(self, R_list, G_list):
if not isinstance(R_list, (list, tuple)): R_list, G_list = [R_list], [G_list]
valid_indices = [i for i, r_matrix in enumerate(R_list) if r_matrix.shape[0] > 0]
if not valid_indices:
print("Error: All R matrices in R_list are empty. Cannot fit PolynomialFeatures.")
self._w = np.zeros(1)
return
R_list_filtered = [R_list[i] for i in valid_indices]
G_list_filtered = [G_list[i] for i in valid_indices]
if not R_list_filtered:
print("Error: R_list_filtered is empty after checking individual matrices. Cannot fit.")
self._w = np.zeros(1)
return
X_list = [self._transform(R_list_filtered[0], fit=True)] + \
[self._transform(R) for R in R_list_filtered[1:]]
self._w = np.zeros(X_list[0].shape[1])
for _ in range(self.epochs):
order = np.random.permutation(len(X_list)) if self.shuffle else range(len(X_list))
for i in order:
X, G = X_list[i], G_list_filtered[i]
if X.shape[0] == 0: continue
self._w -= self.lr * (2.0 * X.T @ (X @ self._w - G))
self._w = np.maximum(self._w, 0.0)
def predict(self, R):
if self._w is None: raise ValueError("Predictor not fitted.")
if R.shape[0] == 0: return np.array([])
return self._transform(R) @ self._w
def evaluate(self, R, G, norm_type='L2', normalization='mean'):
if R.shape[0] == 0 or G.shape[0] == 0:
return (np.nan, 0.0) if self.degree == 1 else np.nan
pred = self.predict(R)
if pred.shape[0] == 0 :
return (np.nan, 0.0) if self.degree == 1 else np.nan
min_len = min(len(pred), len(G))
pred = pred[:min_len]
G = G[:min_len]
if min_len == 0:
return (np.nan, 0.0) if self.degree == 1 else np.nan
if norm_type == 'L1': res = np.sum(np.abs(pred - G))
elif norm_type == 'Linf': res = np.max(np.abs(pred - G))
elif norm_type == 'L0.5': res = np.sum(np.abs(pred - G) ** 0.5) ** 2
else: res = np.linalg.norm(pred - G, ord=2)
eps = np.finfo(float).eps
if normalization == 'mean': factor = np.mean(G) + eps if len(G) > 0 else eps
elif normalization == 'max': factor = np.max(G) + eps if len(G) > 0 else eps
elif normalization == 'std': factor = np.std(G) + eps if len(G) > 1 else eps
elif normalization == 'range': factor = (np.max(G) - np.min(G)) + eps if len(G) > 0 else eps
else: factor = 1
norm_res = res / factor if factor != 0 else res
if self.degree == 1:
if len(pred) < 2 or len(G) < 2: r = np.nan
else:
pred_std, G_std = np.std(pred) + eps, np.std(G) + eps
r = 0.0 if pred_std * G_std == 0 else np.clip(np.corrcoef(pred, G)[0, 1], -1.0, 1.0)
return norm_res, r
return norm_res
def _scores_to_ranks(scores_list, higher_is_better=True):
s = np.array(scores_list, dtype=float); m = np.isnan(s)
s[m] = -np.inf if higher_is_better else np.inf
if np.all(np.isinf(s)) or np.all(np.isnan(s)) or len(s[~np.isinf(s) & ~np.isnan(s)]) == 0 :
return np.full(len(s), np.nan)
ranks = rankdata(-s if higher_is_better else s, method='average')
return ranks
def _get_ranks_for_game(game_name, model_names, model_perf_data, harness_status_str):
scores = []
for model_name in model_names:
try:
harness_key = harness_status_str
game_scores = model_perf_data.get(model_name, {}).get(harness_key, {}).get(game_name, [])
numeric_scores = [s for s in game_scores if isinstance(s, (int, float)) and pd.notna(s)]
scores.append(np.mean(numeric_scores) if numeric_scores else np.nan)
except Exception as e:
scores.append(np.nan)
raw_ranks = _scores_to_ranks(scores, higher_is_better=True)
return raw_ranks
def load_other_task_ranks_from_url(url, model_names_list_for_matching):
try:
response = requests.get(url)
response.raise_for_status()
data = response.json()
except Exception as e:
print(f"Warning: Could not load other task ranks from {url}: {e}")
# Return dummy data for demonstration
num_models = len(model_names_list_for_matching)
return {
"knowledge": np.random.rand(3, num_models),
"math": np.random.rand(3, num_models),
"coding": np.random.rand(3, num_models),
"visual": np.random.rand(3, num_models),
"puzzle": np.random.rand(3, num_models)
}
num_m_expected = len(model_names_list_for_matching)
if num_m_expected == 0:
print("Warning: model_names_list_for_matching is empty in load_other_task_ranks_from_url.")
return {}
if not data: return {}
processed_ranks = {}
for category, rank_vectors in data.items():
if not isinstance(rank_vectors, list): continue
valid_category_vectors = []
for single_rank_vector in rank_vectors:
if isinstance(single_rank_vector, list) and len(single_rank_vector) == num_m_expected:
try:
valid_category_vectors.append([float(x) for x in single_rank_vector])
except (ValueError, TypeError): pass
if valid_category_vectors:
processed_ranks[category] = np.array(valid_category_vectors, dtype=float)
if not processed_ranks:
print(f"Warning: No categories from {url} had rank vectors matching {num_m_expected} models.")
return processed_ranks
def prepare_model_perf_data_from_df(df_input):
model_perf_data = {}
for _, row in df_input.iterrows():
model_name = row['model_name']
harness_status = bool(row['harness'])
game_name = row['game_name']
score = row['avg_score_from_rank']
harness_key = 'harness_true' if harness_status else 'harness_false'
model_perf_data.setdefault(model_name, {})
model_perf_data[model_name].setdefault(harness_key, {})
model_perf_data[model_name][harness_key].setdefault(game_name, [])
if pd.notna(score):
model_perf_data[model_name][harness_key][game_name].append(float(score))
else:
model_perf_data[model_name][harness_key][game_name].append(np.nan)
return model_perf_data
# Main analysis logic
working_df = final_updated_df.copy()
# Set up model order - use selected_models if provided, otherwise use DEFAULT_MODELS_FOR_OTHER_TASKS
if selected_models:
working_df = working_df[working_df['model_name'].isin(selected_models)].copy()
current_model_order = selected_models
else:
current_model_order = DEFAULT_MODELS_FOR_OTHER_TASKS
working_df = working_df[working_df['model_name'].isin(current_model_order)].copy()
if not current_model_order:
print("Error: No models selected for analysis. Exiting.")
return {}
all_model_perf_data = prepare_model_perf_data_from_df(working_df)
all_other_task_ranks = load_other_task_ranks_from_url(OTHER_TASK_RANK_URL, current_model_order)
if not all_model_perf_data:
print("Error: Model performance data is empty. Exiting analysis.")
return {}
if not all_other_task_ranks:
print("Error: Other task ranks data is empty. Exiting analysis.")
return {}
# Set up target games
if selected_games:
target_games_to_process = selected_games
else:
available_games = set()
harness_status_for_G = "harness_true" if harness_filter is None or harness_filter else "harness_false"
for model_name in all_model_perf_data:
if harness_status_for_G in all_model_perf_data.get(model_name, {}):
available_games.update(all_model_perf_data[model_name][harness_status_for_G].keys())
target_games_to_process = sorted(list(available_games)) if available_games else []
# Category combinations
category_combinations = [
{"name": "Comb1_KnowPuzVisMathCode", "keys": ["knowledge", "puzzle", "visual", "math", "coding"]},
{"name": "Comb2_KnowVisMathCode", "keys": ["knowledge", "visual", "math", "coding"]},
{"name": "Comb3_LangPhyVisMathCode", "keys": ["language", "physics", "visual", "math", "coding"]},
{"name": "Comb4_KnowMathCode", "keys": ["knowledge", "math", "coding"]}
]
valid_category_combinations = []
for comb in category_combinations:
if comb["keys"] and all(key in all_other_task_ranks for key in comb["keys"]):
valid_category_combinations.append(comb)
if not valid_category_combinations:
print("Warning: No valid R category combinations. Using available categories.")
available_keys = list(all_other_task_ranks.keys())
if len(available_keys) >= 2:
valid_category_combinations = [{"name": "Available_Categories", "keys": available_keys[:min(4, len(available_keys))]}]
all_polynomial_results = {}
# Process each game
for target_game_name_for_G_loop in target_games_to_process:
harness_status_for_G = "harness_true" if harness_filter is None or harness_filter else "harness_false"
G_current_ranks_float = _get_ranks_for_game(target_game_name_for_G_loop, current_model_order, all_model_perf_data, harness_status_for_G)
if G_current_ranks_float is None or len(G_current_ranks_float) != len(current_model_order):
print(f"Error: G ranks for '{target_game_name_for_G_loop}'. Skipping."); continue
valid_g_indices = ~np.isnan(G_current_ranks_float)
if not np.any(valid_g_indices): continue
G_final_for_fit = G_current_ranks_float[valid_g_indices].astype(int)
models_for_this_G_fit = [current_model_order[i] for i, is_valid in enumerate(valid_g_indices) if is_valid]
all_polynomial_results.setdefault(target_game_name_for_G_loop, {})
for comb_info in valid_category_combinations:
comb_name, comb_keys = comb_info["name"], comb_info["keys"]
R_sources_unfiltered_for_comb = [all_other_task_ranks[k] for k in comb_keys if k in all_other_task_ranks]
if len(R_sources_unfiltered_for_comb) != len(comb_keys): continue
R_sources_for_fit = []
valid_R_source_for_comb = True
for r_matrix_unfiltered in R_sources_unfiltered_for_comb:
if r_matrix_unfiltered.ndim == 1:
if len(r_matrix_unfiltered) == len(current_model_order):
filtered_r = r_matrix_unfiltered[valid_g_indices]
if np.any(~np.isnan(filtered_r)): R_sources_for_fit.append(filtered_r.reshape(-1, 1))
else: valid_R_source_for_comb = False; break
else: valid_R_source_for_comb = False; break
elif r_matrix_unfiltered.ndim == 2:
if r_matrix_unfiltered.shape[1] == len(current_model_order):
filtered_r_matrix = r_matrix_unfiltered[:, valid_g_indices]
if np.any(np.sum(~np.isnan(filtered_r_matrix), axis=1) > 0): R_sources_for_fit.append(filtered_r_matrix)
else: valid_R_source_for_comb = False; break
else: valid_R_source_for_comb = False; break
else: valid_R_source_for_comb = False; break
if not valid_R_source_for_comb or len(R_sources_for_fit) != len(comb_keys): continue
temp_R_sources = []
for r_mat in R_sources_for_fit:
if r_mat.ndim == 1: r_mat = r_mat.reshape(-1,1)
if r_mat.shape[0] > 0: temp_R_sources.append(r_mat)
R_sources_for_fit = temp_R_sources
if len(R_sources_for_fit) != len(comb_keys) : continue
R_batches = build_batches(R_sources_for_fit)
if not R_batches: continue
G_targets_for_fit = [G_final_for_fit] * len(R_batches)
actual_epochs = 1000 # Fixed epochs for simplicity
predictor = RankingPredictor(degree=max_degree, lr=5e-3, epochs=actual_epochs, scale=float(len(models_for_this_G_fit)))
try:
predictor.fit(R_batches, G_targets_for_fit)
except Exception as e_fit:
print(f"Error during predictor.fit for {target_game_name_for_G_loop}/{comb_name}: {e_fit}")
continue
current_run_results = {
"config": {
"poly_degree": max_degree,
"model_order_used": models_for_this_G_fit,
"R_categories_used": comb_keys,
"target_game_name": target_game_name_for_G_loop,
"target_harness_status": harness_status_for_G,
},
"evaluation": {},
"feature_weights": {}
}
eval_results = []
for Rb_idx, Rb_eval in enumerate(R_batches):
if Rb_eval.shape[0] > 0 and G_final_for_fit.shape[0] > 0 and Rb_eval.shape[0] == G_final_for_fit.shape[0]:
eval_results.append(predictor.evaluate(Rb_eval, G_final_for_fit))
if not eval_results:
current_run_results["evaluation"]["avg_residual_error"] = np.nan
if max_degree == 1: current_run_results["evaluation"]["avg_pearson_r"] = np.nan
elif max_degree == 1:
valid_evals = [e for e in eval_results if isinstance(e, tuple) and len(e) == 2 and pd.notna(e[0]) and pd.notna(e[1])]
if valid_evals:
current_run_results["evaluation"]["avg_residual_error"] = np.mean([e[0] for e in valid_evals])
current_run_results["evaluation"]["avg_pearson_r"] = np.mean([e[1] for e in valid_evals])
else: current_run_results["evaluation"].update({"avg_residual_error": np.nan, "avg_pearson_r": np.nan})
else:
valid_evals = [e for e in eval_results if pd.notna(e)]
if valid_evals: current_run_results["evaluation"]["avg_residual_error"] = np.mean(valid_evals)
else: current_run_results["evaluation"]["avg_residual_error"] = np.nan
w_vals = predictor._w
f_names = []
try:
if R_batches and R_batches[0].shape[1] == len(comb_keys):
f_names = predictor._phi.get_feature_names_out(input_features=comb_keys)
elif w_vals is not None: f_names = [f"feat_{j}" for j in range(len(w_vals))]
except Exception:
if w_vals is not None: f_names = [f"feat_{j}" for j in range(len(w_vals))]
if w_vals is not None:
current_run_results["feature_weights"] = {f_names[i]: w_vals[i] for i in range(len(w_vals))}
entry_key = f"{harness_status_for_G}_deg{max_degree}"
all_polynomial_results[target_game_name_for_G_loop].setdefault(comb_name, {})
all_polynomial_results[target_game_name_for_G_loop][comb_name][entry_key] = current_run_results
return all_polynomial_results
def visualize_polynomial_category_weights(polynomial_results: dict, save_plots: bool = False, comb_list: list = None):
"""
Visualizes polynomial analysis results with one bar chart per game within each combination.
Each game's bar chart shows all features (knowledge, physics, math, etc.) on x-axis.
Args:
polynomial_results: Results from run_polynomial_analysis_notebook
save_plots: Whether to save plots (optional, default False for seamless workflow)
comb_list: List of combination names to visualize (default: ["Comb3_LangPhyVisMathCode"])
"""
import matplotlib.pyplot as plt
import numpy as np
# Set default combination list if not provided
if comb_list is None:
comb_list = ["Comb3_LangPhyVisMathCode"]
if not polynomial_results:
print("No polynomial results to visualize")
return
# Organize data by combination first
combination_data = {}
all_games = list(polynomial_results.keys())
for combination_name in comb_list:
combination_data[combination_name] = {}
for game_name in all_games:
game_data = polynomial_results[game_name]
if combination_name in game_data:
config_key = list(game_data[combination_name].keys())[0]
config_data = game_data[combination_name][config_key]
# Extract linear weights only
feature_weights = config_data.get('feature_weights', {})
linear_weights = {}
for feature_name, weight in feature_weights.items():
if feature_name != '1' and '^' not in feature_name and ' ' not in feature_name:
linear_weights[feature_name] = weight
combination_data[combination_name][game_name] = {
'linear_weights': linear_weights,
'error': config_data.get('evaluation', {}).get('avg_residual_error', np.nan),
'categories': config_data.get('config', {}).get('R_categories_used', [])
}
# Filter out combinations with no data
valid_combinations = [comb for comb in comb_list if combination_data.get(comb)]
if not valid_combinations:
print("No valid combinations found with data")
return
print(f"Visualizing {len(valid_combinations)} combinations across {len(all_games)} games")
# Define colors for categories (consistent across all plots)
all_categories = set()
for comb_data in combination_data.values():
for game_data in comb_data.values():
all_categories.update(game_data['linear_weights'].keys())
category_colors = {}
colors = plt.cm.Set3(np.linspace(0, 1, len(all_categories)))
for i, cat in enumerate(sorted(all_categories)):
category_colors[cat] = colors[i]
# Process each combination separately
for combination_name in valid_combinations:
comb_data = combination_data[combination_name]
# Get games with data for this combination
games_with_data = []
for game_name in all_games:
if game_name in comb_data and comb_data[game_name]['linear_weights']:
games_with_data.append(game_name)
if not games_with_data:
print(f"No games with data for {combination_name}")
continue
# Get all features for this combination
all_features_in_comb = set()
for game_name in games_with_data:
all_features_in_comb.update(comb_data[game_name]['linear_weights'].keys())
features_sorted = sorted(all_features_in_comb)
if not features_sorted:
print(f"No features with data for {combination_name}")
continue
n_games = len(games_with_data)
# Create subplots for each game in this combination
if n_games == 1:
fig, axes = plt.subplots(1, 1, figsize=(12, 6))
axes = [axes]
elif n_games == 2:
fig, axes = plt.subplots(1, 2, figsize=(24, 6))
elif n_games <= 4:
fig, axes = plt.subplots(2, 2, figsize=(24, 12))
axes = axes.flatten()
elif n_games <= 6:
fig, axes = plt.subplots(2, 3, figsize=(36, 12))
axes = axes.flatten()
elif n_games <= 9:
fig, axes = plt.subplots(3, 3, figsize=(36, 18))
axes = axes.flatten()
else:
# For more than 9 games, use a 4x4 grid (max 16 games)
fig, axes = plt.subplots(4, 4, figsize=(48, 24))
axes = axes.flatten()
fig.suptitle(f'{combination_name} - Feature Weights by Game', fontsize=16, fontweight='bold')
# Plot each game separately
for game_idx, game_name in enumerate(games_with_data):
if game_idx >= len(axes):
break
ax = axes[game_idx]
game_data = comb_data[game_name]
linear_weights = game_data['linear_weights']
error = game_data['error']
# Prepare data for this game
feature_values = []
feature_colors = []
for feature in features_sorted:
weight = linear_weights.get(feature, 0)
feature_values.append(weight)
feature_colors.append(category_colors.get(feature, '#888888'))
# Create bar chart for this game
x_positions = np.arange(len(features_sorted))
bars = ax.bar(x_positions, feature_values, color=feature_colors, alpha=0.8,
edgecolor='black', linewidth=0.5)
# Add value labels on bars
for bar, value, feature in zip(bars, feature_values, features_sorted):
if abs(value) > 0.001:
height = bar.get_height()
ax.text(bar.get_x() + bar.get_width()/2,
height + (0.02 * max(abs(v) for v in feature_values) if height >= 0 else -0.02 * max(abs(v) for v in feature_values)),
f'{value:.3f}', ha='center',
va='bottom' if height >= 0 else 'top',
fontsize=9, fontweight='bold')
# Customize the plot
ax.set_xticks(x_positions)
ax.set_xticklabels([feature.title() for feature in features_sorted],
fontsize=10, rotation=45, ha='right')
ax.set_ylabel('Linear Weight', fontsize=11)
ax.set_title(f'{game_name}\n(Error: {error:.3f})', fontsize=12, fontweight='bold')
ax.grid(True, alpha=0.3, axis='y')
# Set y-axis limits with some padding
if feature_values:
max_abs_weight = max(abs(v) for v in feature_values)
if max_abs_weight > 0:
ax.set_ylim(-max_abs_weight * 1.2, max_abs_weight * 1.2)
# Add horizontal line at y=0
ax.axhline(y=0, color='black', linestyle='-', alpha=0.3)
# Hide unused subplots
for idx in range(len(games_with_data), len(axes)):
axes[idx].set_visible(False)
plt.tight_layout()
if save_plots:
safe_filename = f'polynomial_weights_{combination_name}'.replace(' ', '_').replace('(', '').replace(')', '')
plt.savefig(f'{safe_filename}.png', dpi=300, bbox_inches='tight')
print(f"Saved plot: {safe_filename}.png")
plt.show()
# Print summary for this combination
print(f"\n{combination_name} Summary:")
print(f" Games: {len(games_with_data)}")
print(f" Features: {', '.join(features_sorted)}")
# Calculate average error across games
errors = [comb_data[game]['error'] for game in games_with_data if not np.isnan(comb_data[game]['error'])]
if errors:
avg_error = np.mean(errors)
print(f" Average error: {avg_error:.4f}")
# Show strongest feature weights across all games
feature_totals = {}
for feature in features_sorted:
total_abs_weight = sum(abs(comb_data[game]['linear_weights'].get(feature, 0)) for game in games_with_data)
feature_totals[feature] = total_abs_weight
if feature_totals:
strongest_feature = max(feature_totals.items(), key=lambda x: x[1])
print(f" Strongest overall feature: {strongest_feature[0]} (total abs weight: {strongest_feature[1]:.3f})")
print("\nPolynomial analysis visualization complete.")
def normalize_game_columns(benchmark_df: pd.DataFrame, columns_to_normalize: list = None) -> pd.DataFrame:
"""
Normalize specified game columns using the formula: (x - x.min()) / (x.max() - x.min()) * 99 + 1
Args:
benchmark_df: DataFrame with benchmark scores
columns_to_normalize: List of column names to normalize.
Defaults to ["Sokoban", "Tetris", "2048", "Candy Crush", "Ace Attorney"]
Returns:
DataFrame with normalized columns
"""
if columns_to_normalize is None:
columns_to_normalize = ["Sokoban", "Tetris", "2048", "Candy Crush", "Ace Attorney"]
normalized_df = benchmark_df.copy()
# Only normalize columns that exist in the DataFrame
existing_columns = [col for col in columns_to_normalize if col in normalized_df.columns]
if existing_columns:
print(f"Normalizing columns: {existing_columns}")
normalized_df[existing_columns] = normalized_df[existing_columns].apply(
lambda x: (x - x.min()) / (x.max() - x.min()) * 99 + 1
)
else:
print("No columns to normalize found in DataFrame")
return normalized_df
def generate_tsne_visualization(benchmark_df: pd.DataFrame, model_names: list, save_plot: bool = False):
"""
Generate t-SNE visualization to explore model clustering based on performance patterns.
Args:
benchmark_df: DataFrame with benchmark scores
model_names: List of model names corresponding to dataframe rows
save_plot: Whether to save the plot
"""
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from sklearn.manifold import TSNE
from sklearn.preprocessing import StandardScaler
print("Generating t-SNE visualization...")
# Prepare data - remove columns with too many NaN values and fill remaining NaNs
data_for_tsne = benchmark_df.copy()
# Remove columns with more than 50% missing values
missing_threshold = 0.5
valid_columns = []
for col in data_for_tsne.columns:
missing_ratio = data_for_tsne[col].isna().sum() / len(data_for_tsne)
if missing_ratio <= missing_threshold:
valid_columns.append(col)
data_for_tsne = data_for_tsne[valid_columns]
print(f"Using {len(valid_columns)} benchmarks for t-SNE (removed {len(benchmark_df.columns) - len(valid_columns)} with >50% missing)")
# Fill remaining NaN values with column mean
data_for_tsne = data_for_tsne.fillna(data_for_tsne.mean())
# Standardize the data
scaler = StandardScaler()
data_scaled = scaler.fit_transform(data_for_tsne)
# Apply t-SNE
tsne = TSNE(n_components=2, random_state=42, perplexity=min(5, len(data_scaled)-1))
tsne_results = tsne.fit_transform(data_scaled)
# Create visualization
fig, ax = plt.subplots(1, 1, figsize=(12, 8))
# Plot points
colors = plt.cm.Set3(np.linspace(0, 1, len(model_names)))
for i, (model_name, color) in enumerate(zip(model_names, colors)):
ax.scatter(tsne_results[i, 0], tsne_results[i, 1],
color=color, s=100, alpha=0.7, edgecolors='black', linewidth=1)
ax.annotate(model_name.replace('-', '-\n'),
(tsne_results[i, 0], tsne_results[i, 1]),
xytext=(5, 5), textcoords='offset points',
fontsize=9, ha='left', va='bottom',
bbox=dict(boxstyle='round,pad=0.3', facecolor=color, alpha=0.3))
ax.set_title('t-SNE Visualization of Model Performance\nAcross Multiple Benchmarks',
fontsize=14, fontweight='bold')
ax.set_xlabel('t-SNE Component 1', fontsize=12)
ax.set_ylabel('t-SNE Component 2', fontsize=12)
ax.grid(True, alpha=0.3)
# Add explanation text
explanation = f"Models clustered by performance similarity across {len(valid_columns)} benchmarks\nCloser models have similar performance patterns"
ax.text(0.02, 0.98, explanation, transform=ax.transAxes,
fontsize=10, va='top', ha='left',
bbox=dict(boxstyle='round,pad=0.5', facecolor='white', alpha=0.8))
plt.tight_layout()
if save_plot:
plt.savefig('tsne_model_comparison.png', dpi=300, bbox_inches='tight')
print("Saved t-SNE plot: tsne_model_comparison.png")
plt.show()
print("t-SNE visualization complete")
def generate_benchmark_correlation_matrix(benchmark_df: pd.DataFrame, save_plot: bool = False):
"""
Generate correlation matrix heatmap for benchmark performance.
Args:
benchmark_df: DataFrame with benchmark scores
save_plot: Whether to save the plot
"""
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np
import pandas as pd
print("Generating benchmark correlation matrix...")
# Prepare data - only use columns with sufficient data
correlation_data = benchmark_df.copy()
# Remove columns with more than 30% missing values for correlation analysis
missing_threshold = 0.3
valid_columns = []
for col in correlation_data.columns:
missing_ratio = correlation_data[col].isna().sum() / len(correlation_data)
if missing_ratio <= missing_threshold:
valid_columns.append(col)
correlation_data = correlation_data[valid_columns]
print(f"Using {len(valid_columns)} benchmarks for correlation analysis")
# Calculate correlation matrix
correlation_matrix = correlation_data.corr()
# Create the heatmap
fig, ax = plt.subplots(1, 1, figsize=(14, 12))
# Generate the heatmap
mask = np.triu(np.ones_like(correlation_matrix, dtype=bool)) # Mask upper triangle
sns.heatmap(correlation_matrix,
mask=mask,
annot=True,
cmap='RdBu_r',
center=0,
square=True,
fmt='.2f',
cbar_kws={"shrink": 0.8},
ax=ax)
ax.set_title('Benchmark Performance Correlation Matrix\n(Lower Triangle Only)',
fontsize=14, fontweight='bold')
# Rotate labels for better readability
ax.set_xticklabels(ax.get_xticklabels(), rotation=45, ha='right')
ax.set_yticklabels(ax.get_yticklabels(), rotation=0)
plt.tight_layout()
if save_plot:
plt.savefig('benchmark_correlation_matrix.png', dpi=300, bbox_inches='tight')
print("Saved correlation matrix: benchmark_correlation_matrix.png")
plt.show()
# Print high correlations
print("\nHighest correlations (>0.7):")
high_corr_pairs = []
for i in range(len(correlation_matrix.columns)):
for j in range(i+1, len(correlation_matrix.columns)):
corr_val = correlation_matrix.iloc[i, j]
if abs(corr_val) > 0.7:
high_corr_pairs.append((correlation_matrix.columns[i],
correlation_matrix.columns[j],
corr_val))
high_corr_pairs.sort(key=lambda x: abs(x[2]), reverse=True)
for bench1, bench2, corr in high_corr_pairs[:10]: # Top 10
print(f" {bench1} ↔ {bench2}: {corr:.3f}")
print("Correlation analysis complete")
def generate_latent_ability_decomposition(benchmark_df: pd.DataFrame, model_names: list, n_components: int = 3, save_plot: bool = False):
"""
Generate latent ability decomposition using PCA to identify underlying model capabilities.
Args:
benchmark_df: DataFrame with benchmark scores
model_names: List of model names corresponding to dataframe rows
n_components: Number of latent components to extract
save_plot: Whether to save the plot
"""
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from sklearn.decomposition import PCA
from sklearn.preprocessing import StandardScaler
print(f"Generating latent ability decomposition with {n_components} components...")
# Prepare data
decomposition_data = benchmark_df.copy()
# Remove columns with more than 40% missing values
missing_threshold = 0.4
valid_columns = []
for col in decomposition_data.columns:
missing_ratio = decomposition_data[col].isna().sum() / len(decomposition_data)
if missing_ratio <= missing_threshold:
valid_columns.append(col)
decomposition_data = decomposition_data[valid_columns]
# Fill NaN values with column mean
decomposition_data = decomposition_data.fillna(decomposition_data.mean())
print(f"Using {len(valid_columns)} benchmarks for decomposition")
# Standardize the data
scaler = StandardScaler()
data_scaled = scaler.fit_transform(decomposition_data)
# Apply PCA
pca = PCA(n_components=n_components)
pca_results = pca.fit_transform(data_scaled)
# Create comprehensive visualization
fig = plt.figure(figsize=(16, 12))
# Plot 1: Model scores on latent components
ax1 = plt.subplot(2, 3, 1)
colors = plt.cm.Set3(np.linspace(0, 1, len(model_names)))
for i, (model_name, color) in enumerate(zip(model_names, colors)):
ax1.scatter(pca_results[i, 0], pca_results[i, 1],
color=color, s=100, alpha=0.7, edgecolors='black')
ax1.annotate(model_name.split('-')[0],
(pca_results[i, 0], pca_results[i, 1]),
xytext=(5, 5), textcoords='offset points',
fontsize=8, ha='left', va='bottom')
ax1.set_title('PCA: Component 1 vs Component 2', fontsize=12, fontweight='bold')
ax1.set_xlabel(f'Component 1 ({pca.explained_variance_ratio_[0]:.1%} variance)')
ax1.set_ylabel(f'Component 2 ({pca.explained_variance_ratio_[1]:.1%} variance)')
ax1.grid(True, alpha=0.3)
# Plot 2: Component loadings heatmap
ax2 = plt.subplot(2, 3, 2)
loadings = pca.components_[:n_components].T
loadings_df = pd.DataFrame(loadings, index=valid_columns,
columns=[f'PC{i+1}' for i in range(n_components)])
im = ax2.imshow(loadings_df.T, cmap='RdBu_r', aspect='auto', vmin=-1, vmax=1)
ax2.set_title('Component Loadings', fontsize=12, fontweight='bold')
ax2.set_yticks(range(n_components))
ax2.set_yticklabels([f'PC{i+1}' for i in range(n_components)])
ax2.set_xticks(range(len(valid_columns)))
ax2.set_xticklabels(valid_columns, rotation=45, ha='right')
# Add colorbar
cbar = plt.colorbar(im, ax=ax2, shrink=0.8)
cbar.set_label('Loading Strength')
# Plot 3: Explained variance
ax3 = plt.subplot(2, 3, 3)
cumvar = np.cumsum(pca.explained_variance_ratio_)
ax3.bar(range(1, n_components+1), pca.explained_variance_ratio_,
alpha=0.7, label='Individual')
ax3.plot(range(1, n_components+1), cumvar, 'ro-', label='Cumulative')
ax3.set_title('Explained Variance', fontsize=12, fontweight='bold')
ax3.set_xlabel('Principal Component')
ax3.set_ylabel('Variance Explained')
ax3.legend()
ax3.grid(True, alpha=0.3)
# Plot 4: Model abilities on each component
ax4 = plt.subplot(2, 3, 4)
model_abilities = pd.DataFrame(pca_results,
index=model_names,
columns=[f'PC{i+1}' for i in range(n_components)])
# Plot PC1 scores as horizontal bar chart
pc1_scores = model_abilities['PC1'].sort_values(ascending=True)
bars = ax4.barh(range(len(pc1_scores)), pc1_scores.values,
color=[colors[model_names.index(name)] for name in pc1_scores.index])
ax4.set_yticks(range(len(pc1_scores)))
ax4.set_yticklabels([name.split('-')[0] for name in pc1_scores.index])
ax4.set_title(f'PC1 Abilities ({pca.explained_variance_ratio_[0]:.1%} var)',
fontsize=12, fontweight='bold')
ax4.set_xlabel('PC1 Score')
ax4.grid(True, alpha=0.3)
# Plot 5: 3D view if we have 3+ components
if n_components >= 3:
ax5 = plt.subplot(2, 3, 5, projection='3d')
for i, (model_name, color) in enumerate(zip(model_names, colors)):
ax5.scatter(pca_results[i, 0], pca_results[i, 1], pca_results[i, 2],
color=color, s=100, alpha=0.7, edgecolors='black')
ax5.set_title('3D PCA View', fontsize=12, fontweight='bold')
ax5.set_xlabel(f'PC1 ({pca.explained_variance_ratio_[0]:.1%})')
ax5.set_ylabel(f'PC2 ({pca.explained_variance_ratio_[1]:.1%})')
ax5.set_zlabel(f'PC3 ({pca.explained_variance_ratio_[2]:.1%})')
# Plot 6: Component interpretation
ax6 = plt.subplot(2, 3, 6)
top_loadings_per_pc = []
for pc in range(min(3, n_components)):
pc_loadings = np.abs(loadings_df[f'PC{pc+1}'])
top_features = pc_loadings.nlargest(3)
top_loadings_per_pc.append(f"PC{pc+1}:\n" + "\n".join([f" {feat}: {val:.2f}"
for feat, val in top_features.items()]))
interpretation_text = "\n\n".join(top_loadings_per_pc)
ax6.text(0.05, 0.95, interpretation_text, transform=ax6.transAxes,
fontsize=10, va='top', ha='left',
bbox=dict(boxstyle='round,pad=0.5', facecolor='lightgray', alpha=0.8))
ax6.set_title('Top Features per Component', fontsize=12, fontweight='bold')
ax6.axis('off')
plt.tight_layout()
if save_plot:
plt.savefig('latent_ability_decomposition.png', dpi=300, bbox_inches='tight')
print("Saved latent ability plot: latent_ability_decomposition.png")
plt.show()
# Print summary
print(f"\nLatent Ability Summary:")
print(f"Total variance explained by {n_components} components: {cumvar[-1]:.1%}")
for i in range(n_components):
print(f" PC{i+1}: {pca.explained_variance_ratio_[i]:.1%}")
print("Latent ability decomposition complete")
return {
'pca_model': pca,
'pca_results': pca_results,
'loadings': loadings_df,
'explained_variance': pca.explained_variance_ratio_,
'valid_columns': valid_columns
}
# Add these functions at the end of the file
def factorize_with_scipy_no_bias(Y, d, lambda_l1=0.01, lambda_l2_m=0.01, lambda_l2_b=0.01):
"""
Low-rank matrix factorization using scipy optimization with improved NaN handling.
"""
import numpy as np
from scipy.optimize import minimize
T, S = Y.shape
def unpack(x):
M = x[:T*d].reshape(T, d)
B = x[T*d:T*d + S*d].reshape(S, d)
return M, B
def loss_fn(x):
M, B = unpack(x)
Y_hat = M @ B.T # change this line to handle null values in Y, so for positions if nan values in Y, don't need to compute the Y_hat
mask = ~np.isnan(Y)
loss = np.sum((Y[mask] - Y_hat[mask])**2)
l1 = lambda_l1 * np.sum(np.abs(B))
l2 = lambda_l2_m * np.sum(M**2) + lambda_l2_b * np.sum(B**2)
# Soft constraint loss: encourage M[2, k] <= M[3, k] for all k
# soft_constraint = np.sum(np.maximum(M[2] - M[3], 0.0)**2)
# pairwise = lambda_pair * soft_constraint
return loss + l1 + l2
x0 = np.random.rand((T + S) * d)
bounds = [(0, None)] * len(x0)
res = minimize(loss_fn, x0, bounds=bounds, method='L-BFGS-B', options={
'maxfun': 50000, # maximum number of function evaluations
'maxiter': 50000, # maximum number of iterations
'disp': True # optional: print progress
})
success, loss = res.success, res.fun
M_opt, B_opt = unpack(res.x)
return M_opt, B_opt, success, loss
def factorize_with_scipy_bias(Y, d, lambda_l1=0.01, lambda_l2_m=0.01, lambda_l2_b=0.01):
"""
Low-rank matrix factorization with bias terms using scipy optimization.
"""
import numpy as np
from scipy.optimize import minimize
T, S = Y.shape
def unpack(x):
M = x[:T*d].reshape(T, d)
B = x[T*d:T*d + S*d].reshape(S, d)
v = x[T*d + S*d:].reshape(S) # Benchmark bias
return M, B, v
def loss_fn(x):
M, B, v = unpack(x)
Y_hat = M @ B.T + v[None, :] # Add benchmark bias
mask = ~np.isnan(Y)
loss = np.sum((Y[mask] - Y_hat[mask])**2)
l1 = lambda_l1 * np.sum(np.abs(B))
l2 = lambda_l2_m * np.sum(M**2) + lambda_l2_b * np.sum(B**2)
return loss + l1 + l2
# Initial values
x0 = np.random.rand((T + S) * d + S)
bounds = [(0, None)] * len(x0)
res = minimize(loss_fn, x0, bounds=bounds, method='L-BFGS-B',options={
'maxfun': 50000, # maximum number of function evaluations
'maxiter': 50000, # maximum number of iterations
'disp': True # optional: print progress
})
success, loss = res.success, res.fun
M_opt, B_opt, v_opt = unpack(res.x)
return M_opt, B_opt, v_opt, success, loss
def generate_simple_tsne_plots(benchmark_data: dict, save_plot: bool = False):
"""
Generate simple t-SNE plots: scores-based and ranks-based side by side.
Args:
benchmark_data: Dictionary with benchmark scores
save_plot: Whether to save the plot
"""
import matplotlib.pyplot as plt
import numpy as np
from sklearn.manifold import TSNE
from scipy.stats import rankdata
print("Generating simple t-SNE plots...")
fig, ax = plt.subplots(1, 2, figsize=(10, 4))
titles = ["t-SNE based on scores", "t-SNE based on ranks"]
# Convert to numpy array, forcing all values to be float (NaNs for invalid values)
values = np.array([np.array(row, dtype=np.float64) for row in benchmark_data.values()], dtype=np.float64)
keys = list(benchmark_data.keys()) # Start with all keys
# Remove rows with any NaN values
mask = ~np.isnan(values).any(axis=1)
values = values[mask]
keys = [keys[i] for i in range(len(keys)) if mask[i]] # Only keep keys for valid rows
# Normalize each column (feature-wise) instead of row-wise
min_vals = np.nanmin(values, axis=0, keepdims=True)
max_vals = np.nanmax(values, axis=0, keepdims=True)
range_vals = np.where(max_vals - min_vals == 0, 1, max_vals - min_vals)
normal_values = (values - min_vals) / range_vals
# Apply t-SNE (reduce to 2D for visualization)
for i in range(2):
tsne = TSNE(n_components=2, perplexity=7, random_state=0)
tsne_result = tsne.fit_transform(normal_values if i == 0 else np.apply_along_axis(rankdata, axis=1, arr=normal_values))
ax[i].scatter(tsne_result[:, 0], tsne_result[:, 1])
for j, key in enumerate(keys):
ax[i].text(tsne_result[j, 0], tsne_result[j, 1], key, fontsize=9)
ax[i].set_title(titles[i])
ax[i].set_xlabel("t-SNE 1")
ax[i].set_ylabel("t-SNE 2")
ax[i].set_xticks([])
ax[i].set_yticks([])
plt.tight_layout()
if save_plot:
plt.savefig('tsne.png', dpi=300, bbox_inches='tight')
print("Saved t-SNE plots: tsne.png")
plt.show()
print("t-SNE visualization complete")
def generate_simple_correlation_matrix(benchmark_df: pd.DataFrame, save_plot: bool = False):
"""
Generate simple correlation matrix heatmap with upper triangle masked.
Args:
benchmark_df: DataFrame with benchmark scores (should be normalized)
save_plot: Whether to save the plot
"""
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np
print("Generating simple correlation matrix...")
corr_matrix = benchmark_df.corr(method='spearman')
mask = np.triu(np.ones_like(corr_matrix, dtype=bool))
plt.figure(figsize=(11, 9))
sns.heatmap(
corr_matrix,
mask=mask, # apply the mask here
annot=False,
fmt=".2f",
cmap="coolwarm",
center=0.4,
linewidths=0.5,
vmin=-0.2,
vmax=1.0
)
plt.title("Benchmark Correlation Matrix")
if save_plot:
plt.savefig('correlation_matrix.png', dpi=300, bbox_inches='tight')
print("Saved correlation matrix: correlation_matrix.png")
plt.show()
print("Correlation matrix complete")
def generate_simple_latent_factorization(benchmark_data: dict, benchmark_df: pd.DataFrame, feature_num: int = 4, save_plot: bool = False):
"""
Generate simple latent ability factorization with horizontal bar charts.
Args:
benchmark_data: Dictionary with benchmark scores
benchmark_df: DataFrame with benchmark scores (for column names)
feature_num: Number of latent features
save_plot: Whether to save the plot
"""
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np
import pandas as pd
from scipy.stats import rankdata
from itertools import chain
print(f"Generating latent factorization with {feature_num} features...")
# Convert scores to ranks
benchmark_ranks = {}
for benchmark, scores in benchmark_data.items():
scores_array = np.array(scores, dtype=np.float64)
valid_mask = ~np.isnan(scores_array)
ranks = np.full_like(scores_array, fill_value=np.nan, dtype=np.float64)
ranks[valid_mask] = rankdata(scores_array[valid_mask], method='average')
benchmark_ranks[benchmark] = ranks.tolist()
benchmark_df_rank = pd.DataFrame(benchmark_ranks)
# Apply factorization
np.random.seed(12)
Y = (benchmark_df_rank * 0.2).to_numpy()
bias = False
M_hat, B_hat, success, loss = factorize_with_scipy_no_bias(Y, feature_num, lambda_l1=0.08, lambda_l2_m=0.01, lambda_l2_b=0.00)
print(f"Factorization: success={success}, loss={loss:.4f}")
np.set_printoptions(suppress=True, precision=2)
# Normalize
normalized = B_hat / B_hat.sum(axis=1, keepdims=True)
print("benchmark:")
for bench, feature_normalized in zip(benchmark_df.columns, normalized):
formatted_normalized = " ".join(f"{x:.2f}" for x in feature_normalized)
print(f"[{formatted_normalized}] {bench}")
feature_names = [f"Feature_{i+1}" for i in range(normalized.shape[1])]
bench_names = list(benchmark_df.columns)
# rows = features, columns = benchmarks
bench_feature_df = pd.DataFrame(
normalized.T, # transpose ⇒ shape (feature_num, n_benchmarks)
index=feature_names, # feature_num feature rows
columns=bench_names # one column per benchmark
)
# Configuration
BASELINE = ['Sokoban', 'Tetris', '2048', 'Candy Crush', 'Ace Attorney']
TOP_K = 3 # highest three
# Color dictionary
model_colors = {
'Multi Challenge': '#66c2a5', 'MMLU-Pro': '#fc8d62', 'GPQA': '#8da0cb',
'Sokoban': '#e78ac3', 'SMB': '#a6d854', 'Tetris': '#ffd92f',
'2048': '#e5c494', 'Candy Crush': '#b3b3b3', 'Ace Attorney': '#1f78b4',
'nyt-connections': '#fb8072', 'Math 500': '#80b1d3', 'AIME 2025': '#fdb462',
'MGSM': '#bebada', 'VISTA': '#ffed6f', 'HLE': '#bc80bd',
'MMMU': '#fccde5', 'EnigmaEval': '#ccebc5', 'HLE(Text)': '#d9d9d9',
'Emma-mini Physics': '#1b9e77', 'PHYBench': '#d95f02', 'chatbot Arena_vision': '#7570b3',
'EMMA-mini Math': '#66a61e', 'LiveBench-Math': '#e6ab02',
'LiveBench-Coding': '#a6761d', 'BigCodeBench Pass@1': '#666666', 'Aider polyglot coding': '#1f78b4',
'LiveBench-Language': '#33a02c'
}
# Build labels + weights for every feature
labels_per_feat = []
weights_per_feat = []
for feat_name, row in bench_feature_df.iterrows():
# pick top‑K non‑baseline benchmarks
available_other = [col for col in row.index if col not in BASELINE and col in row.index]
top_other = (
row[available_other] # only consider non-baseline columns that exist
.sort_values(ascending=False)
.head(TOP_K)
.index.tolist()
)
ordered_labels = top_other + [b for b in BASELINE if b in row.index] # only include existing baseline columns
labels_per_feat.append(ordered_labels)
weights_per_feat.append(row[ordered_labels].values)
# Guarantee a colour for every label
needed = set(chain(*labels_per_feat)) - model_colors.keys()
if needed:
extra_palette = sns.color_palette("tab20", len(needed))
model_colors.update(dict(zip(needed, extra_palette)))
# Plot
n_feat = len(labels_per_feat)
fig, axes = plt.subplots(1, n_feat, figsize=(3.6 * n_feat, 3))
if n_feat == 1:
axes = [axes] # Make it iterable
plt.subplots_adjust(wspace=0.4)
for i, (ax, labels, weights) in enumerate(zip(axes, labels_per_feat, weights_per_feat)):
bar_colors = [model_colors.get(l, '#cccccc') for l in labels]
ax.barh(labels, weights, color=bar_colors)
ax.set_title(f"Feature {i+1}", fontsize=12)
ax.set_xlabel("Weight", fontsize=12)
ax.tick_params(axis='both', labelsize=11)
ax.set_xlim(0, weights.max() * 1.1)
ax.invert_yaxis()
ax.spines[['top', 'right']].set_visible(False)
# dashed line separating "top‑K" and baseline for quick visual cue
baseline_count = len([b for b in BASELINE if b in labels])
if baseline_count > 0:
ax.hlines(y=len(labels) - baseline_count - 0.5,
xmin=0, xmax=1, colors='gray', linestyles='--',
transform=ax.get_yaxis_transform())
plt.tight_layout()
if save_plot:
plt.savefig("low_rank_consistent_colours.png", dpi=300, bbox_inches='tight')
print("Saved latent factorization plot")
plt.show()
print("Latent factorization complete")
return {
'M_hat': M_hat,
'B_hat': B_hat,
'normalized': normalized,
'bench_feature_df': bench_feature_df,
'success': success,
'loss': loss
}