analyze_convergence

Skill: Analyze Convergence

Purpose

Interpret flash algorithm convergence metrics, identify patterns, and produce publication-quality analysis for the Results section.

When to Use

• After running benchmark experiments
• When comparing baseline vs candidate algorithms
• When investigating failure cases

Analysis Procedure

Step 1: Load and Parse Results

import json
import pandas as pd

def load_results(results_dir, algorithm_name):
    """Load JSONL results into DataFrame."""
    records = []
    with open(f"{results_dir}/raw/{algorithm_name}_results.jsonl") as f:
        for line in f:
            records.append(json.loads(line))
    return pd.DataFrame(records)

Step 2: Convergence Rate by Family

def convergence_by_family(df):
    """Calculate convergence rate per fluid family."""
    return df.groupby("family").agg(
        total=("converged", "count"),
        converged=("converged", "sum"),
        rate_pct=("converged", lambda x: round(100 * x.mean(), 2)),
        median_time_ms=("cpu_time_ms", "median")
    ).reset_index()

Step 3: Convergence Maps

Generate 2D convergence maps in (T, P) space:

import matplotlib.pyplot as plt
import numpy as np

def plot_convergence_map(df, family_name, algorithm_name, save_path):
    """Plot convergence success/failure in TP space."""
    fam = df[df["family"] == family_name]

    fig, ax = plt.subplots(figsize=(8, 6))

    conv = fam[fam["converged"] == True]
    fail = fam[fam["converged"] == False]

    ax.scatter(conv["T_K"] - 273.15, conv["P_bara"],
               c="green", alpha=0.3, s=10, label="Converged")
    ax.scatter(fail["T_K"] - 273.15, fail["P_bara"],
               c="red", alpha=0.8, s=20, marker="x", label="Failed")

    ax.set_xlabel("Temperature (°C)")
    ax.set_ylabel("Pressure (bara)")
    ax.set_title(f"Convergence Map — {family_name} — {algorithm_name}")
    ax.legend()
    ax.grid(True, alpha=0.3)
    ax.set_yscale("log")

    plt.tight_layout()
    plt.savefig(save_path, dpi=300, bbox_inches="tight")
    plt.close()