""" Visualize AFRES results: QD archive, rubric evolution, factor performance. """ import json import numpy as np import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt import matplotlib.patches as mpatches from matplotlib.gridspec import GridSpec def load_results(path="/app/afres_results.json"): with open(path) as f: return json.load(f) def plot_qd_archive_grid(results, save_path="/app/qd_archive_grid.png"): """Visualize the MAP-Elites archive as a 2D grid (signal_type x time_horizon).""" fig, ax = plt.subplots(figsize=(12, 8)) # Get unique signal types and time horizons signal_types = ["price_based", "volume_based", "fundamental", "technical", "cross_sectional", "time_series"] time_horizons = ["short", "medium", "long"] # Collect all factors with their fitness all_factors = results.get("best_factors", []) + [] # Build fitness heatmap (average across complexity bins) grid = np.zeros((len(signal_types), len(time_horizons))) counts = np.zeros((len(signal_types), len(time_horizons))) for factor in results.get("best_factors", []): st = factor.get("signal_type", "") th = factor.get("time_horizon", "") score = factor.get("overall_score", 0) if st in signal_types and th in time_horizons: i = signal_types.index(st) j = time_horizons.index(th) grid[i, j] += score counts[i, j] += 1 # Average grid = np.where(counts > 0, grid / counts, np.nan) # Plot heatmap im = ax.imshow(grid, cmap='YlOrRd', aspect='auto', vmin=0, vmax=1) ax.set_xticks(range(len(time_horizons))) ax.set_yticks(range(len(signal_types))) ax.set_xticklabels(time_horizons) ax.set_yticklabels(signal_types) ax.set_xlabel("Time Horizon") ax.set_ylabel("Signal Type") ax.set_title("MAP-Elites Archive: Average Factor Quality Score by Behavioral Niche") # Add text annotations for i in range(len(signal_types)): for j in range(len(time_horizons)): if not np.isnan(grid[i, j]): ax.text(j, i, f"{grid[i, j]:.2f}", ha="center", va="center", color="black", fontsize=10) plt.colorbar(im, ax=ax, label="Average Overall Score") plt.tight_layout() plt.savefig(save_path, dpi=150) plt.close() print(f"Saved QD archive grid to {save_path}") def plot_rubric_evolution(results, save_path="/app/rubric_evolution.png"): """Plot how rubric MAE improved over iterations.""" history = results.get("rubric_history", []) if not history: return fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5)) iterations = [h["iteration"] + 1 for h in history] maes = [h["mae"] for h in history] n_items = [h["n_items"] for h in history] # Plot MAE convergence ax1.plot(iterations, maes, 'bo-', linewidth=2, markersize=8) ax1.set_xlabel("Iteration") ax1.set_ylabel("MAE (Mean Absolute Error)") ax1.set_title("Rubric Discovery: MAE Convergence") ax1.grid(True, alpha=0.3) ax1.set_ylim(bottom=0) # Plot number of items ax2.bar(iterations, n_items, color='steelblue', alpha=0.7) ax2.set_xlabel("Iteration") ax2.set_ylabel("Number of Rubric Items") ax2.set_title("Rubric Complexity Growth") ax2.grid(True, alpha=0.3, axis='y') plt.tight_layout() plt.savefig(save_path, dpi=150) plt.close() print(f"Saved rubric evolution to {save_path}") def plot_factor_performance(results, save_path="/app/factor_performance.png"): """Plot top factors ranked by IC and Sharpe.""" factors = results.get("best_factors", []) if not factors: return fig, (ax1, ax2, ax3) = plt.subplots(1, 3, figsize=(18, 5)) names = [f"{f['id']}\n({f['signal_type'][:3]})" for f in factors] ics = [f["ic"] for f in factors] sharpes = [f["sharpe"] for f in factors] scores = [f["overall_score"] for f in factors] # IC plot colors = plt.cm.RdYlGn(np.linspace(0.3, 0.9, len(factors))) bars1 = ax1.barh(range(len(factors)), ics, color=colors) ax1.set_yticks(range(len(factors))) ax1.set_yticklabels(names) ax1.set_xlabel("Information Coefficient (IC)") ax1.set_title("Factor Predictive Power (IC)") ax1.axvline(x=0, color='black', linestyle='-', linewidth=0.5) ax1.invert_yaxis() # Sharpe plot bars2 = ax2.barh(range(len(factors)), sharpes, color=colors) ax2.set_yticks(range(len(factors))) ax2.set_yticklabels(names) ax2.set_xlabel("Sharpe Ratio") ax2.set_title("Factor Sharpe Ratio") ax2.invert_yaxis() # Overall score bars3 = ax3.barh(range(len(factors)), scores, color=colors) ax3.set_yticks(range(len(factors))) ax3.set_yticklabels(names) ax3.set_xlabel("Overall Rubric Score") ax3.set_title("Factor Overall Quality Score") ax3.set_xlim(0, 1) ax3.invert_yaxis() plt.tight_layout() plt.savefig(save_path, dpi=150) plt.close() print(f"Saved factor performance to {save_path}") def plot_factor_radar(results, save_path="/app/factor_radar.png"): """Radar chart showing rubric dimensions for top factor.""" factors = results.get("best_factors", []) if not factors: return top = factors[0] scores = top.get("rubric_scores", {}) if not scores: return categories = list(scores.keys()) values = list(scores.values()) # Close the radar values += values[:1] angles = np.linspace(0, 2 * np.pi, len(categories), endpoint=False).tolist() angles += angles[:1] fig, ax = plt.subplots(figsize=(8, 8), subplot_kw=dict(polar=True)) ax.plot(angles, values, 'o-', linewidth=2, color='#1f77b4') ax.fill(angles, values, alpha=0.25, color='#1f77b4') ax.set_xticks(angles[:-1]) ax.set_xticklabels(categories, size=10) ax.set_ylim(0, 1) ax.set_title(f"Top Factor Rubric Profile\n{top['id']}: {top['expression']}", size=12, pad=20) ax.grid(True) plt.tight_layout() plt.savefig(save_path, dpi=150) plt.close() print(f"Saved factor radar to {save_path}") def create_summary_report(results, save_path="/app/afres_summary.txt"): """Create a text summary report.""" lines = [] lines.append("=" * 70) lines.append("AFRES: AGENTIC FACTOR REVISION AND EVALUATION SYSTEM") lines.append("Summary Report") lines.append("=" * 70) lines.append("") # Rubric summary rubric = results.get("rubric", {}) items = rubric.get("items", []) lines.append(f"DISCOVERED RUBRIC ({len(items)} items)") lines.append("-" * 40) for item in items: lines.append(f" {item['name']:25s} (weight={item['weight']:.2f})") lines.append(f" {item['description']}") lines.append("") # Archive summary archive = results.get("archive_summary", {}) lines.append("QD ARCHIVE SUMMARY") lines.append("-" * 40) lines.append(f" Coverage: {archive.get('coverage', 0):.1%}") lines.append(f" Elite factors: {archive.get('num_elites', 0)}") lines.append(f" Mean fitness: {archive.get('mean_fitness', 0):.3f}") lines.append(f" Max fitness: {archive.get('max_fitness', 0):.3f}") lines.append(f" Min fitness: {archive.get('min_fitness', 0):.3f}") lines.append("") # Top factors lines.append("TOP FACTORS AFTER REVISION") lines.append("-" * 40) for f in results.get("best_factors", [])[:5]: lines.append(f" {f['id']}: {f['expression']}") lines.append(f" IC={f['ic']:.4f} | Sharpe={f['sharpe']:.2f} | Score={f['overall_score']:.3f}") lines.append(f" Type={f['signal_type']} | Horizon={f['time_horizon']} | Gen={f['generation']}") lines.append("") lines.append("=" * 70) lines.append(f"Total factors evaluated: {results.get('total_factors', 0)}") report = "\n".join(lines) with open(save_path, "w") as f: f.write(report) print(f"Saved summary report to {save_path}") return report def main(): results = load_results() plot_qd_archive_grid(results) plot_rubric_evolution(results) plot_factor_performance(results) plot_factor_radar(results) report = create_summary_report(results) print("\n" + report) if __name__ == "__main__": main()