Function run_flywheel_grid

python/openquant/research.py:187–237

Execute multiple flywheel configs and return a ranked leaderboard.

(
    dataset: ResearchDataset,
    configs: list[dict[str, Any]],
    *,
    run_names: list[str] | None = None,
)

def run_flywheel_grid(
    dataset: ResearchDataset,
    configs: list[dict[str, Any]],
    *,
    run_names: list[str] | None = None,
) -> dict[str, Any]:
    """Execute multiple flywheel configs and return a ranked leaderboard."""
    if not configs:
        raise ValueError("configs cannot be empty")
    if run_names is not None and len(run_names) != len(configs):
        raise ValueError("run_names/configs length mismatch")

    rows: list[dict[str, Any]] = []
    runs: list[dict[str, Any]] = []

    for idx, cfg in enumerate(configs):
        run_name = run_names[idx] if run_names is not None else f"run_{idx:02d}"
        out = run_flywheel_iteration(dataset, config=cfg)
        summary = out["summary"].row(0, named=True)
        promotion = out["promotion"]
        costs = out["costs"]

        row = {
            "run_name": run_name,
            "run_index": idx,
            "config_digest": research_run_manifest(cfg)["config_digest"],
            "portfolio_sharpe": float(summary["portfolio_sharpe"]),
            "realized_sharpe": float(summary["realized_sharpe"]),
            "net_sharpe": float(summary["net_sharpe"]),
            "gross_total_return": float(costs["gross_total_return"]),
            "net_total_return": float(costs["net_total_return"]),
            "turnover": float(costs["turnover"]),
            "estimated_cost": float(costs["estimated_total_cost"]),
            "passed_realized_sharpe": bool(promotion["passed_realized_sharpe"]),
            "passed_net_sharpe": bool(promotion["passed_net_sharpe"]),
            "passed_alignment_guard": bool(promotion["passed_alignment_guard"]),
            "passed_event_order_guard": bool(promotion["passed_event_order_guard"]),
            "promote_candidate": bool(promotion["promote_candidate"]),
        }
        rows.append(row)
        runs.append({"run_name": run_name, "config": cfg, "output": out})

    leaderboard = pl.DataFrame(rows).sort(
        by=["promote_candidate", "net_sharpe", "realized_sharpe", "run_index"],
        descending=[True, True, True, False],
    )
    return {
        "leaderboard": leaderboard,
        "records": leaderboard.to_dicts(),
        "runs": runs,
    }


def _turnover(positions: list[float]) -> float: