| """ |
| 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)) |
| |
| |
| signal_types = ["price_based", "volume_based", "fundamental", "technical", |
| "cross_sectional", "time_series"] |
| time_horizons = ["short", "medium", "long"] |
| |
| |
| all_factors = results.get("best_factors", []) + [] |
| |
| |
| 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 |
| |
| |
| grid = np.where(counts > 0, grid / counts, np.nan) |
| |
| |
| 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") |
| |
| |
| 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] |
| |
| |
| 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) |
| |
| |
| 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] |
| |
| |
| 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() |
| |
| |
| 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() |
| |
| |
| 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()) |
| |
| |
| 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 = 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 = 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("") |
| |
| |
| 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() |
|
|
