feat: Add parallel run system with API key distribution

- Add predix_parallel.py: Run multiple factor experiments concurrently * python predix_parallel.py --runs 5 --api-keys 2 -m openrouter * Round-robin API key distribution across available keys * Rich live dashboard with per-run status, elapsed time, exit codes * Graceful shutdown (Ctrl+C kills all children cleanly) - Add --run-id parameter to predix.py for isolated single runs * Separate log files: fin_quant_run{N}.log * Separate results: results/runs/run{N}/ * Separate workspace: RD-Agent_workspace_run{N}/ * Separate databases per run - Modify CoSTEER and FactorRunner for PARALLEL_RUN_ID isolation * _save_intermediate_results uses run-specific directories * _save_result_to_database and _write_run_log isolated per run * _ensure_results_dirs creates run-specific paths - Reduce max_loop from 10 to 3 for faster iterations - Add docs/parallel_runs.md with full documentation Tests: 103 passed
2026-08-09 04:57:44 +00:00 · 2026-04-04 09:39:12 +02:00
parent 54073da2b0
commit 8b7eb87546
8 changed files with 1062 additions and 16 deletions
@@ -90,13 +90,99 @@ class QuantRDLoop(RDLoop):
            await asyncio.sleep(1)

    def coding(self, prev_out: dict[str, Any]):
-        if prev_out["direct_exp_gen"]["propose"].action == "factor":
-            exp = self.factor_coder.develop(prev_out["direct_exp_gen"]["exp_gen"])
-        elif prev_out["direct_exp_gen"]["propose"].action == "model":
-            exp = self.model_coder.develop(prev_out["direct_exp_gen"]["exp_gen"])
-        logger.log_object(exp, tag="coder result")
+        exp = None
+        try:
+            if prev_out["direct_exp_gen"]["propose"].action == "factor":
+                exp = self.factor_coder.develop(prev_out["direct_exp_gen"]["exp_gen"])
+            elif prev_out["direct_exp_gen"]["propose"].action == "model":
+                exp = self.model_coder.develop(prev_out["direct_exp_gen"]["exp_gen"])
+            logger.log_object(exp, tag="coder result")
+        except (FactorEmptyError, ModelEmptyError) as e:
+            logger.warning(f"Coding failed with {type(e).__name__}: {e}")
+            raise
+        except Exception as e:
+            logger.error(f"Unexpected coding error: {e}")
+            raise
+        finally:
+            # Always save results, even on partial failure
+            if exp is not None:
+                self._save_coder_results(exp)
+
        return exp

+    def _save_coder_results(self, exp) -> None:
+        """
+        Save CoSTEER-generated code and evaluation to results/ directory.
+
+        This ensures we have a record of generated factors even if
+        the full Qlib backtest pipeline fails or is skipped.
+
+        Parameters
+        ----------
+        exp : Experiment
+            The experiment with generated code
+        """
+        import json
+        from datetime import datetime
+        from pathlib import Path
+
+        try:
+            project_root = Path(__file__).parent.parent.parent.parent
+            results_dir = project_root / "results" / "runs"
+            results_dir.mkdir(parents=True, exist_ok=True)
+
+            # Build result summary
+            summary = {
+                "timestamp": datetime.now().isoformat(),
+                "hypothesis": None,
+                "factors": [],
+                "status": "generated",
+            }
+
+            if hasattr(exp, "hypothesis") and exp.hypothesis is not None:
+                summary["hypothesis"] = getattr(exp.hypothesis, "hypothesis", None)
+
+            # Extract generated code from sub_workspace_list
+            if hasattr(exp, "sub_workspace_list") and exp.sub_workspace_list:
+                for i, ws in enumerate(exp.sub_workspace_list):
+                    factor_info = {
+                        "index": i,
+                        "code": None,
+                        "file_count": 0,
+                    }
+                    if hasattr(ws, "file_dict") and ws.file_dict:
+                        factor_info["file_count"] = len(ws.file_dict)
+                        factor_info["code"] = ws.file_dict.get("factor.py", None)
+                    summary["factors"].append(factor_info)
+
+            # Check if experiment was accepted or rejected
+            if hasattr(exp, "accepted_tasks"):
+                accepted = getattr(exp, "accepted_tasks", [])
+                summary["accepted_count"] = len(accepted)
+                summary["status"] = "accepted" if accepted else "rejected"
+
+            # Write JSON summary
+            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+            safe_name = (summary["hypothesis"] or "unknown_factor")[:80].replace("/", "_").replace(" ", "_")
+            json_path = results_dir / f"{timestamp}_{safe_name}.json"
+
+            with open(json_path, "w", encoding="utf-8") as f:
+                json.dump(summary, f, ensure_ascii=False, indent=2, default=str)
+
+            logger.info(f"CoSTEER result saved to {json_path}")
+
+            # Also write a consolidated log entry
+            log_dir = project_root / "results" / "logs"
+            log_dir.mkdir(parents=True, exist_ok=True)
+            today = datetime.now().strftime("%Y-%m-%d")
+            log_file = log_dir / f"coder_runs_{today}.jsonl"
+
+            with open(log_file, "a", encoding="utf-8") as f:
+                f.write(json.dumps(summary, ensure_ascii=False, default=str) + "\n")
+
+        except Exception as e:
+            logger.warning(f"Failed to save CoSTEER results: {e}")
+
    def running(self, prev_out: dict[str, Any]):
        if prev_out["direct_exp_gen"]["propose"].action == "factor":
            exp = self.factor_runner.develop(prev_out["coding"])
@@ -114,7 +114,13 @@ class CoSTEER(Developer[Experiment]):
        reached_max_seconds = False

        evo_fb = None
+        iteration_count = 0
+
+        # Save initial state before first iteration
+        self._save_intermediate_results(evo_exp, None, 0, start_datetime)
+
        for evo_exp in self.evolve_agent.multistep_evolve(evo_exp, self.evaluator):
+            iteration_count += 1
            assert isinstance(evo_exp, Experiment)  # multiple inheritance
            evo_fb = self._get_last_fb()
            update_fallback = self.should_use_new_evo(
@@ -129,6 +135,10 @@ class CoSTEER(Developer[Experiment]):
            logger.log_object(evo_exp.sub_workspace_list, tag="evolving code")
            for sw in evo_exp.sub_workspace_list:
                logger.info(f"evolving workspace: {sw}")
+
+            # Save intermediate results after each iteration
+            self._save_intermediate_results(evo_exp, evo_fb, iteration_count, start_datetime)
+
            if max_seconds is not None and (datetime.now() - start_datetime).total_seconds() > max_seconds:
                logger.info(f"Reached max time limit {max_seconds} seconds, stop evolving")
                reached_max_seconds = True
@@ -154,6 +164,100 @@ class CoSTEER(Developer[Experiment]):
        exp.experiment_workspace = evo_exp.experiment_workspace
        return exp

+    def _save_intermediate_results(self, evo_exp, evo_fb, iteration: int, start_datetime) -> None:
+        """
+        Save intermediate CoSTEER results to results/ directory after each iteration.
+
+        This ensures results are visible even if CoSTEER takes a long time
+        or ultimately fails.
+
+        Parameters
+        ----------
+        evo_exp : EvolvingItem
+            Current evolving experiment
+        evo_fb : CoSTEERMultiFeedback
+            Feedback from the evaluator
+        iteration : int
+            Current iteration number
+        start_datetime : datetime
+            When the develop process started
+        """
+        import json as _json
+        import os as _os
+        from datetime import datetime as _dt
+
+        try:
+            # Go up from rdagent/components/coder/CoSTEER/ to project root (5 levels)
+            project_root = Path(__file__).parent.parent.parent.parent.parent
+
+            # Parallel run isolation: use run-specific directory if PARALLEL_RUN_ID is set
+            parallel_run_id = _os.getenv("PARALLEL_RUN_ID", "0")
+            if parallel_run_id != "0":
+                results_dir = project_root / "results" / "runs" / f"run{parallel_run_id}" / "costeer"
+            else:
+                results_dir = project_root / "results" / "runs"
+            results_dir.mkdir(parents=True, exist_ok=True)
+
+            # Build summary
+            summary = {
+                "timestamp": _dt.now().isoformat(),
+                "iteration": iteration,
+                "elapsed_seconds": (_dt.now() - start_datetime).total_seconds(),
+                "factors": [],
+            }
+
+            # Extract factor info from sub_workspace_list
+            if hasattr(evo_exp, "sub_workspace_list") and evo_exp.sub_workspace_list:
+                for i, sw in enumerate(evo_exp.sub_workspace_list):
+                    factor = {"index": i, "file_count": 0, "code_preview": None}
+                    if hasattr(sw, "file_dict") and sw.file_dict:
+                        factor["file_count"] = len(sw.file_dict)
+                        code = sw.file_dict.get("factor.py", "")
+                        if code:
+                            # First 200 chars as preview
+                            factor["code_preview"] = code[:200]
+                    summary["factors"].append(factor)
+
+            # Extract feedback info
+            if evo_fb is not None:
+                summary["feedback_count"] = len(evo_fb) if hasattr(evo_fb, "__len__") else 0
+                accepted = 0
+                rejected = 0
+                for fb in evo_fb:
+                    if fb is not None:
+                        if fb.is_acceptable():
+                            accepted += 1
+                        else:
+                            rejected += 1
+                summary["accepted"] = accepted
+                summary["rejected"] = rejected
+                summary["status"] = "accepted" if accepted > 0 else "rejected"
+            else:
+                summary["feedback_count"] = 0
+                summary["accepted"] = 0
+                summary["rejected"] = 0
+                summary["status"] = "initialized"
+
+            # Write JSON file
+            ts = _dt.now().strftime("%Y%m%d_%H%M%S")
+            if parallel_run_id != "0":
+                json_path = results_dir / f"costeer_run{parallel_run_id}_iter{iteration:02d}_{ts}.json"
+            else:
+                json_path = results_dir / f"costeer_iter{iteration:02d}_{ts}.json"
+
+            with open(json_path, "w", encoding="utf-8") as f:
+                _json.dump(summary, f, ensure_ascii=False, indent=2, default=str)
+
+            logger.info(
+                f"CoSTEER iteration {iteration}: "
+                f"accepted={summary.get('accepted', 0)}, "
+                f"rejected={summary.get('rejected', 0)}, "
+                f"saved to {json_path.name}"
+            )
+
+        except Exception as e:
+            logger.warning(f"Failed to save intermediate CoSTEER results: {e}")
+
    def _exp_postprocess_by_feedback(self, evo: Experiment, feedback: CoSTEERMultiFeedback) -> Experiment:
        """
        Responsibility:
@@ -12,8 +12,8 @@ class CoSTEERSettings(ExtendedBaseSettings):
    coder_use_cache: bool = False
    """Indicates whether to use cache for the coder"""

-    max_loop: int = 10
-    """Maximum number of task implementation loops"""
+    max_loop: int = 3
+    """Maximum number of task implementation loops (reduced from 10 for faster iterations)"""

    fail_task_trial_limit: int = 20

@@ -513,6 +513,14 @@ class QlibFactorRunner(CachedRunner[QlibFactorExperiment]):
            db_path.mkdir(parents=True, exist_ok=True)
            db_file = db_path / "backtest_results.db"

+            # Parallel run isolation: use run-specific subdirectory if PARALLEL_RUN_ID is set
+            run_id = os.getenv("PARALLEL_RUN_ID", "0")
+            if run_id != "0":
+                # For parallel runs, save to isolated results directory
+                isolated_db_path = project_root / "results" / "runs" / f"run{run_id}" / "db"
+                isolated_db_path.mkdir(parents=True, exist_ok=True)
+                db_file = isolated_db_path / "backtest_results.db"
+
            # Save to database
            db = ResultsDatabase(db_path=str(db_file))
            run_id = db.add_backtest(factor_name=factor_name[:100], metrics=metrics)
@@ -547,13 +555,20 @@ class QlibFactorRunner(CachedRunner[QlibFactorExperiment]):
            Database run ID
        """
        import json
+        import os as _os
        from datetime import datetime
        from pathlib import Path

        try:
            # Ensure factors directory exists (5 levels up to project root)
            project_root = Path(__file__).parent.parent.parent.parent.parent
-            factors_dir = project_root / "results" / "factors"
+
+            # Parallel run isolation: use run-specific directory if PARALLEL_RUN_ID is set
+            parallel_run_id = _os.getenv("PARALLEL_RUN_ID", "0")
+            if parallel_run_id != "0":
+                factors_dir = project_root / "results" / "runs" / f"run{parallel_run_id}" / "factors"
+            else:
+                factors_dir = project_root / "results" / "factors"
            factors_dir.mkdir(parents=True, exist_ok=True)

            # Sanitize factor name for filename
@@ -777,7 +792,13 @@ class QlibFactorRunner(CachedRunner[QlibFactorExperiment]):
        # Write to results/logs/
        try:
            project_root = Path(__file__).parent.parent.parent.parent.parent
-            log_dir = project_root / "results" / "logs"
+
+            # Parallel run isolation: use run-specific log directory if PARALLEL_RUN_ID is set
+            parallel_run_id = os.getenv("PARALLEL_RUN_ID", "0")
+            if parallel_run_id != "0":
+                log_dir = project_root / "results" / "runs" / f"run{parallel_run_id}" / "logs"
+            else:
+                log_dir = project_root / "results" / "logs"
            log_dir.mkdir(parents=True, exist_ok=True)

            # One file per day
@@ -798,5 +819,15 @@ class QlibFactorRunner(CachedRunner[QlibFactorExperiment]):
        """Ensure all results directories exist."""
        from pathlib import Path
        project_root = Path(__file__).parent.parent.parent.parent.parent
-        for subdir in ["results/runs", "results/factors", "results/logs", "results/backtests", "results/db"]:
-            (project_root / subdir).mkdir(parents=True, exist_ok=True)
+
+        # Parallel run isolation: create run-specific directories if PARALLEL_RUN_ID is set
+        parallel_run_id = os.getenv("PARALLEL_RUN_ID", "0")
+        if parallel_run_id != "0":
+            # Isolated run directories
+            run_base = project_root / "results" / "runs" / f"run{parallel_run_id}"
+            for subdir in ["factors", "logs", "db"]:
+                (run_base / subdir).mkdir(parents=True, exist_ok=True)
+        else:
+            # Standard shared directories
+            for subdir in ["results/runs", "results/factors", "results/logs", "results/backtests", "results/db"]:
+                (project_root / subdir).mkdir(parents=True, exist_ok=True)