From ab2fefe1f60d4e7ae8511fb83154b2b527371b38 Mon Sep 17 00:00:00 2001 From: TPTBusiness Date: Sun, 3 May 2026 08:49:18 +0200 Subject: [PATCH] fix: correct project root paths and subprocess handling in parallel runner and CLI - predix_parallel.py: fix project_root from scripts/ to repo root (parent.parent) - predix_parallel.py: fix .env loading path and API key distribution logic - cli.py: fix project_root depth from 4 to 3 .parent hops (7 locations) - cli.py start_loop: use sys.executable instead of hardcoded python - cli.py start_loop: replace broad pkill with targeted child process management - cli.py parallel: remove hardcoded model=local --- rdagent/app/cli.py | 223 +++++++++++++++++++------------------ scripts/predix_parallel.py | 63 +++++------ 2 files changed, 145 insertions(+), 141 deletions(-) diff --git a/rdagent/app/cli.py b/rdagent/app/cli.py index 92c70dcf..9241a998 100644 --- a/rdagent/app/cli.py +++ b/rdagent/app/cli.py @@ -21,11 +21,10 @@ load_dotenv(".env") import subprocess from importlib.resources import path as rpath -from typing import Dict, Optional +from typing import Annotated import typer from rich.console import Console -from typing_extensions import Annotated try: from rdagent.utils.env import logger @@ -146,10 +145,10 @@ def ds_user_interact(port=19900): @app.command(name="fin_factor") def fin_factor_cli( - path: Optional[str] = None, - step_n: Optional[int] = None, - loop_n: Optional[int] = None, - all_duration: Optional[str] = None, + path: str | None = None, + step_n: int | None = None, + loop_n: int | None = None, + all_duration: str | None = None, checkout: CheckoutOption = True, ): fin_factor(path=path, step_n=step_n, loop_n=loop_n, all_duration=all_duration, checkout=checkout) @@ -157,10 +156,10 @@ def fin_factor_cli( @app.command(name="fin_model") def fin_model_cli( - path: Optional[str] = None, - step_n: Optional[int] = None, - loop_n: Optional[int] = None, - all_duration: Optional[str] = None, + path: str | None = None, + step_n: int | None = None, + loop_n: int | None = None, + all_duration: str | None = None, checkout: CheckoutOption = True, ): fin_model(path=path, step_n=step_n, loop_n=loop_n, all_duration=all_duration, checkout=checkout) @@ -168,10 +167,10 @@ def fin_model_cli( @app.command(name="fin_quant") def fin_quant_cli( - path: Optional[str] = None, - step_n: Optional[int] = None, - loop_n: Optional[int] = None, - all_duration: Optional[str] = None, + path: str | None = None, + step_n: int | None = None, + loop_n: int | None = None, + all_duration: str | None = None, checkout: CheckoutOption = True, with_dashboard: bool = typer.Option(False, "--with-dashboard/-d", help="Start web dashboard automatically"), with_cli_dashboard: bool = typer.Option(False, "--cli-dashboard/-c", help="Show beautiful CLI dashboard"), @@ -231,7 +230,7 @@ def fin_quant_cli( if not api_key: console.print("\n[bold red]❌ OPENROUTER_API_KEY not set in .env[/bold red]") console.print("[yellow]Add your API key to .env and retry:[/yellow]") - console.print(' OPENROUTER_API_KEY=sk-or-your-key-here') + console.print(" OPENROUTER_API_KEY=sk-or-your-key-here") raise typer.Exit(code=1) os.environ["OPENAI_API_KEY"] = api_key @@ -250,8 +249,8 @@ def fin_quant_cli( console.print(f" [dim]Base URL: {os.environ['OPENAI_API_BASE']}[/dim]") # Wait until the llama.cpp server is fully loaded before starting the pipeline - import urllib.request import urllib.error + import urllib.request base_url = os.environ["OPENAI_API_BASE"].removesuffix("/v1").rstrip("/") health_url = f"{base_url}/health" @@ -285,7 +284,7 @@ def fin_quant_cli( subprocess.run( ["python", "web/dashboard_api.py"], cwd=str(Path(__file__).parent.parent.parent), - env={**os.environ, "FLASK_ENV": "development"} + env={**os.environ, "FLASK_ENV": "development"}, ) dashboard_thread = threading.Thread(target=start_web_dashboard, daemon=True) @@ -327,9 +326,9 @@ def fin_quant_cli( @app.command(name="fin_factor_report") def fin_factor_report_cli( - report_folder: Optional[str] = None, - path: Optional[str] = None, - all_duration: Optional[str] = None, + report_folder: str | None = None, + path: str | None = None, + all_duration: str | None = None, checkout: CheckoutOption = True, ): fin_factor_report(report_folder=report_folder, path=path, all_duration=all_duration, checkout=checkout) @@ -342,12 +341,12 @@ def general_model_cli(report_file_path: str): @app.command(name="data_science") def data_science_cli( - path: Optional[str] = None, + path: str | None = None, checkout: CheckoutOption = True, - step_n: Optional[int] = None, - loop_n: Optional[int] = None, - timeout: Optional[str] = None, - competition: Optional[str] = None, + step_n: int | None = None, + loop_n: int | None = None, + timeout: str | None = None, + competition: str | None = None, ): data_science( path=path, @@ -361,16 +360,16 @@ def data_science_cli( @app.command(name="llm_finetune") def llm_finetune_cli( - path: Optional[str] = None, + path: str | None = None, checkout: CheckoutOption = True, - benchmark: Optional[str] = None, - benchmark_description: Optional[str] = None, - dataset: Optional[str] = None, - base_model: Optional[str] = None, - upper_data_size_limit: Optional[int] = None, - step_n: Optional[int] = None, - loop_n: Optional[int] = None, - timeout: Optional[str] = None, + benchmark: str | None = None, + benchmark_description: str | None = None, + dataset: str | None = None, + base_model: str | None = None, + upper_data_size_limit: int | None = None, + step_n: int | None = None, + loop_n: int | None = None, + timeout: str | None = None, ): llm_finetune( path=path, @@ -436,6 +435,7 @@ def rl_trading_cli( rdagent rl_trading --mode backtest --no-with-protections """ from pathlib import Path + import yaml console = Console() @@ -447,18 +447,18 @@ def rl_trading_cli( with open(config_path) as f: config = yaml.safe_load(f) or {} - console.print(f"\n[bold blue]🤖 RL Trading Agent[/bold blue]") + console.print("\n[bold blue]🤖 RL Trading Agent[/bold blue]") console.print(f"Mode: [cyan]{mode}[/cyan]") console.print(f"Algorithm: [cyan]{algorithm.upper()}[/cyan]") console.print(f"Protections: {'[green]Enabled[/green]' if with_protections else '[red]Disabled[/red]'}") try: - from rdagent.components.coder.rl import RLTradingAgent, RLCosteer, TradingEnv + from rdagent.components.coder.rl import RLCosteer, RLTradingAgent, TradingEnv except ImportError as e: - console.print(f"[bold red]Error: RL components not available.[/bold red]") + console.print("[bold red]Error: RL components not available.[/bold red]") console.print(f"Details: {e}") - console.print(f"\n[yellow]Install RL dependencies:[/yellow]") - console.print(f" pip install stable-baselines3 gymnasium") + console.print("\n[yellow]Install RL dependencies:[/yellow]") + console.print(" pip install stable-baselines3 gymnasium") raise typer.Exit(code=1) if mode == "train": @@ -474,8 +474,8 @@ def rl_trading_cli( console.print("[dim]Loading market data...[/dim]") # TODO: Load actual data from config # For now, create mock environment - import numpy as np import gymnasium as gym + import numpy as np # Create simple mock environment for demonstration class MockTradingEnv(gym.Env): @@ -511,7 +511,7 @@ def rl_trading_cli( model_path_out.parent.mkdir(parents=True, exist_ok=True) agent.save(model_path_out) - console.print(f"\n[bold green]✅ Training complete![/bold green]") + console.print("\n[bold green]✅ Training complete![/bold green]") console.print(f"Model saved to: [cyan]{model_path_out}[/cyan]") console.print(f"Algorithm: {result['algorithm']}") console.print(f"Timesteps: {result['total_timesteps']:,}") @@ -537,9 +537,9 @@ def rl_trading_cli( agent = RLTradingAgent(algorithm=algorithm.upper()) # Run backtest - from rdagent.components.backtesting import FactorBacktester - import pandas as pd import numpy as np + import pandas as pd + from rdagent.components.backtesting import FactorBacktester backtester = FactorBacktester() @@ -548,8 +548,8 @@ def rl_trading_cli( n_steps = 500 mock_prices = pd.Series(100 + np.cumsum(np.random.randn(n_steps) * 0.5)) mock_indicators = pd.DataFrame({ - 'rsi': np.random.uniform(30, 70, n_steps), - 'macd': np.random.randn(n_steps) * 0.1, + "rsi": np.random.uniform(30, 70, n_steps), + "macd": np.random.randn(n_steps) * 0.1, }) console.print("[yellow]Running backtest...[/yellow]") @@ -560,7 +560,7 @@ def rl_trading_cli( enable_protections=with_protections, ) - console.print(f"\n[bold green]✅ Backtest complete![/bold green]") + console.print("\n[bold green]✅ Backtest complete![/bold green]") console.print(f" Final Equity: [green]${metrics.get('final_equity', 0):,.2f}[/green]") console.print(f" Sharpe Ratio: {metrics.get('sharpe_ratio', 0):.3f}") console.print(f" Max Drawdown: {metrics.get('max_drawdown', 0):.2%}") @@ -634,7 +634,7 @@ def generate_strategies_cli( rdagent generate_strategies -n 3 -i 10 --optuna-trials 50 # Deep optimization """ from rich.console import Console - from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TimeRemainingColumn + from rich.progress import BarColumn, Progress, SpinnerColumn, TextColumn, TimeRemainingColumn from rich.table import Table console = Console() @@ -653,7 +653,7 @@ def generate_strategies_cli( raise typer.Exit(code=1) console.print(f"\n[bold blue]{'='*60}[/bold blue]") - console.print(f"[bold blue] PREDIX Strategy Generator[/bold blue]") + console.print("[bold blue] PREDIX Strategy Generator[/bold blue]") console.print(f"[bold blue]{'='*60}[/bold blue]") console.print(f" Strategies: [cyan]{count}[/cyan]") console.print(f" Workers: [cyan]{workers}[/cyan]") @@ -680,12 +680,12 @@ def generate_strategies_cli( _slog = _dlog.setup("strategies", **_strat_ctx) try: - from rdagent.components.coder.strategy_orchestrator import StrategyOrchestrator import pandas as pd + from rdagent.components.coder.strategy_orchestrator import StrategyOrchestrator all_results = [] best_strategy = None - best_sharpe = float('-inf') + best_sharpe = float("-inf") # CONTINUOUS OPTIMIZATION LOOP for iteration in range(1, max_iterations + 1): @@ -734,7 +734,7 @@ def generate_strategies_cli( # Track best strategy for r in results: - sharpe = r.get("sharpe_ratio", float('-inf')) + sharpe = r.get("sharpe_ratio", float("-inf")) if sharpe > best_sharpe: best_sharpe = sharpe best_strategy = r @@ -754,7 +754,7 @@ def generate_strategies_cli( rejected = [r for r in results if r.get("status") == "rejected"] console.print(f"\n[bold green]{'='*60}[/bold green]") - console.print(f"[bold green] Strategy Generation Summary[/bold green]") + console.print("[bold green] Strategy Generation Summary[/bold green]") console.print(f"[bold green]{'='*60}[/bold green]") table = Table(show_header=True, header_style="bold magenta", show_lines=True) @@ -783,7 +783,7 @@ def generate_strategies_cli( # Show best strategy details if best_strategy: console.print(f"\n[bold gold1]{'='*60}[/bold gold1]") - console.print(f"[bold gold1] BEST STRATEGY[/bold gold1]") + console.print("[bold gold1] BEST STRATEGY[/bold gold1]") console.print(f"[bold gold1]{'='*60}[/bold gold1]") console.print(f" Name: [cyan]{best_strategy.get('strategy_name', 'Unknown')}[/cyan]") console.print(f" Sharpe: [green]{best_strategy.get('sharpe_ratio', 0):.4f}[/green]") @@ -791,13 +791,13 @@ def generate_strategies_cli( console.print(f" Max DD: [yellow]{best_strategy.get('max_drawdown', 0):.2%}[/yellow]") console.print(f" Win Rate: [cyan]{best_strategy.get('win_rate', 0):.2%}[/cyan]") if best_strategy.get("best_params"): - console.print(f"\n [bold]Optimized Parameters:[/bold]") + console.print("\n [bold]Optimized Parameters:[/bold]") for param, val in best_strategy["best_params"].items(): console.print(f" {param}: [cyan]{val}[/cyan]") console.print(f"[bold gold1]{'='*60}[/bold gold1]") if accepted: - console.print(f"\n[bold]Accepted Strategies:[/bold]") + console.print("\n[bold]Accepted Strategies:[/bold]") acc_table = Table(show_header=True, header_style="bold cyan") acc_table.add_column("#", width=4) acc_table.add_column("Strategy", width=30) @@ -820,13 +820,13 @@ def generate_strategies_cli( ) console.print(acc_table) - console.print(f"\n[bold green]Strategies saved to:[/bold green] [cyan]results/strategies_new/[/cyan]") + console.print("\n[bold green]Strategies saved to:[/bold green] [cyan]results/strategies_new/[/cyan]") console.print(f"[bold blue]{'='*60}[/bold blue]\n") _slog.success(f"Generated {len(all_results)} strategies ({len([r for r in all_results if r.get('status')=='accepted'])} accepted)") except ImportError as e: _slog.error(f"Strategy components not available: {e}") - console.print(f"[bold red]Error: Strategy components not available.[/bold red]") + console.print("[bold red]Error: Strategy components not available.[/bold red]") console.print(f"Details: {e}") raise typer.Exit(code=1) except Exception as e: @@ -862,17 +862,18 @@ def optimize_portfolio_cli( raise typer.Exit(code=1) console.print(f"\n[bold blue]{'='*60}[/bold blue]") - console.print(f"[bold blue] PREDIX Portfolio Optimizer[/bold blue]") + console.print("[bold blue] PREDIX Portfolio Optimizer[/bold blue]") console.print(f"[bold blue]{'='*60}[/bold blue]") console.print(f" Top N: [cyan]{top_n}[/cyan]") console.print(f" Method: [cyan]{method}[/cyan]") console.print(f"[bold blue]{'='*60}[/bold blue]\n") try: - from rdagent.components.backtesting.risk_management import PortfolioOptimizer import json from pathlib import Path + from rdagent.components.backtesting.risk_management import PortfolioOptimizer + project_root = Path(__file__).parent.parent.parent strategies_dir = project_root / "results" / "strategies_new" @@ -1009,14 +1010,15 @@ def strategies_report_cli( rdagent strategies_report -s path/to/strategy.json # Single strategy rdagent strategies_report -o custom/reports/ # Custom output dir """ + from pathlib import Path + from rich.console import Console from rich.progress import Progress, SpinnerColumn, TextColumn - from pathlib import Path console = Console() console.print(f"\n[bold blue]{'='*60}[/bold blue]") - console.print(f"[bold blue] PREDIX Strategy Report Generator[/bold blue]") + console.print("[bold blue] PREDIX Strategy Report Generator[/bold blue]") console.print(f"[bold blue]{'='*60}[/bold blue]\n") project_root = Path(__file__).parent.parent.parent @@ -1068,20 +1070,20 @@ def strategies_report_cli( progress.update(task, completed=1) console.print(f"\n[bold green]{'='*60}[/bold green]") - console.print(f"[bold green] Report Generation Complete[/bold green]") + console.print("[bold green] Report Generation Complete[/bold green]") console.print(f"[bold green]{'='*60}[/bold green]") console.print(f" Reports generated: [cyan]{reports_generated}/{len(strategy_files)}[/cyan]") console.print(f" Output directory: [cyan]{output_dir_path}[/cyan]") console.print(f"[bold green]{'='*60}[/bold green]\n") -def _generate_single_strategy_report(strategy_file: Path, output_dir: Path) -> Dict: +def _generate_single_strategy_report(strategy_file: Path, output_dir: Path) -> dict: """Generate a report for a single strategy.""" import json + import matplotlib matplotlib.use("Agg") # Non-interactive backend import matplotlib.pyplot as plt - import seaborn as sns with open(strategy_file, encoding="utf-8") as f: strategy = json.load(f) @@ -1156,7 +1158,7 @@ if __name__ == "__main__": @app.command(name="start_llama") def start_llama_cli( model: str = typer.Option( - None, "--model", "-m", help="Path to model file" + None, "--model", "-m", help="Path to model file", ), port: int = typer.Option(8081, "--port", "-p", help="Server port"), gpu_layers: int = typer.Option(30, "--gpu-layers", "-g", help="GPU layers"), @@ -1178,8 +1180,6 @@ def start_llama_cli( rdagent start_llama --gpu-layers 40 --ctx-size 4096 rdagent start_llama --reasoning """ - import subprocess - import sys import os model_path = model or os.getenv( @@ -1216,7 +1216,7 @@ def start_llama_cli( if not reasoning: cmd.extend(["--reasoning", "off"]) - print(f"🚀 Starting llama.cpp server...") + print("🚀 Starting llama.cpp server...") print(f" Model: {Path(model_path).name}") print(f" Port: {port}") print(f" GPU Layers: {gpu_layers}") @@ -1249,15 +1249,14 @@ def start_loop_cli( rdagent start_loop rdagent start_loop --target 5 --max-wait 3600 """ - import subprocess - import signal - import sys import os - from datetime import datetime + import signal + import subprocess import time + from datetime import datetime - script_dir = str(Path(__file__).parent.parent.parent.parent) - generator = f"python {script_dir}/scripts/predix_smart_strategy_gen.py" + script_dir = str(Path(__file__).parent.parent.parent) + generator = [sys.executable, f"{script_dir}/scripts/predix_smart_strategy_gen.py"] logfile = f"{script_dir}/results/logs/generator_loop.log" pidfile = "/tmp/predix_loop.pid" # nosec B108 — administrative PID file, single-process daemon @@ -1270,12 +1269,19 @@ def start_loop_cli( with open(logfile, "a") as f: f.write(line + "\n") + child_proc = None # track current child PID for targeted cleanup + def cleanup(signum=None, frame=None): log("Received termination signal. Cleaning up...") - try: - subprocess.run(["pkill", "-f", "predix_smart_strategy_gen.py"], capture_output=True) - except Exception: - pass + if child_proc is not None: + try: + child_proc.terminate() + child_proc.wait(timeout=10) + except Exception: + try: + child_proc.kill() + except Exception: + pass try: os.remove(pidfile) except FileNotFoundError: @@ -1321,26 +1327,32 @@ def start_loop_cli( strat_count = len(list(strat_dir.glob("*.json"))) if strat_dir.exists() else 0 log(f"📁 Existing strategies: {strat_count}") - # Kill stale processes - try: - subprocess.run(["pkill", "-9", "-f", "predix_smart_strategy_gen.py"], capture_output=True) - except Exception: - pass - time.sleep(2) + # Kill stale child from previous iteration + if child_proc is not None: + try: + child_proc.terminate() + child_proc.wait(timeout=10) + except subprocess.TimeoutExpired: + child_proc.kill() + child_proc.wait() + except Exception: + pass + child_proc = None + time.sleep(2) # Start generator log("🤖 Starting generator...") - proc = subprocess.Popen( - generator.split(), + child_proc = subprocess.Popen( + generator, cwd=script_dir, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) - log(f" PID: {proc.pid}") + log(f" PID: {child_proc.pid}") # Monitor progress elapsed = 0 - while proc.poll() is None: + while child_proc.poll() is None: time.sleep(30) elapsed += 30 @@ -1349,11 +1361,12 @@ def start_loop_cli( if elapsed >= max_wait: log(f" ⏰ Timeout after {elapsed}s. Killing...") - proc.kill() + child_proc.kill() break # Check results - exit_code = proc.wait() + exit_code = child_proc.wait() + child_proc = None if exit_code == 0: log("✅ Generator completed successfully") elif exit_code == -9: @@ -1394,24 +1407,24 @@ def parallel_cli( rdagent parallel -n 10 -k 2 """ import subprocess - import sys from pathlib import Path + from rdagent.log import daily_log as _dlog - project_root = Path(__file__).parent.parent.parent.parent + project_root = Path(__file__).parent.parent.parent script = project_root / "scripts" / "predix_parallel.py" if not script.exists(): typer.echo(f"❌ Script not found: {script}") raise typer.Exit(code=1) - cmd = [sys.executable, str(script), "--runs", str(runs), "--api-keys", str(api_keys), "-m", "local"] + cmd = [sys.executable, str(script), "--runs", str(runs), "--api-keys", str(api_keys)] _plog = _dlog.setup("parallel", runs=runs, api_keys=api_keys, model="local") typer.echo(f"🚀 Starting {runs} parallel runs...") typer.echo(f" Script: {script}") typer.echo(f" API Keys: {api_keys}") - typer.echo(f" Model: local (llama.cpp)") + typer.echo(" Model: local (llama.cpp)") try: result = subprocess.run(cmd, cwd=str(project_root)) @@ -1445,11 +1458,11 @@ def eval_all_cli( rdagent eval_all -n 500 -p 8 """ import subprocess - import sys from pathlib import Path + from rdagent.log import daily_log as _dlog - project_root = Path(__file__).parent.parent.parent.parent + project_root = Path(__file__).parent.parent.parent script = project_root / "scripts" / "predix_full_eval.py" if not script.exists(): @@ -1500,10 +1513,9 @@ def batch_backtest_cli( rdagent batch_backtest --all """ import subprocess - import sys from pathlib import Path - project_root = Path(__file__).parent.parent.parent.parent + project_root = Path(__file__).parent.parent.parent script = project_root / "scripts" / "predix_batch_backtest.py" if not script.exists(): @@ -1553,10 +1565,9 @@ def simple_eval_cli( rdagent simple_eval --all """ import subprocess - import sys from pathlib import Path - project_root = Path(__file__).parent.parent.parent.parent + project_root = Path(__file__).parent.parent.parent script = project_root / "scripts" / "predix_simple_eval.py" if not script.exists(): @@ -1586,7 +1597,7 @@ def simple_eval_cli( @app.command(name="rebacktest") def rebacktest_cli( strategies_dir: str = typer.Option( - None, "--strategies-dir", "-d", help="Directory containing strategy JSON files" + None, "--strategies-dir", "-d", help="Directory containing strategy JSON files", ), ): """ @@ -1600,10 +1611,9 @@ def rebacktest_cli( rdagent rebacktest -d results/strategies_new/ """ import subprocess - import sys from pathlib import Path - project_root = Path(__file__).parent.parent.parent.parent + project_root = Path(__file__).parent.parent.parent script = project_root / "scripts" / "predix_rebacktest_strategies.py" if not script.exists(): @@ -1628,10 +1638,10 @@ def rebacktest_cli( @app.command(name="report") def report_cli( strategy_path: str = typer.Option( - None, "--strategy", "-s", help="Path to single strategy JSON (default: all strategies)" + None, "--strategy", "-s", help="Path to single strategy JSON (default: all strategies)", ), output: str = typer.Option( - None, "--output", "-o", help="Output directory (default: results/strategy_reports/)" + None, "--output", "-o", help="Output directory (default: results/strategy_reports/)", ), ): """ @@ -1654,10 +1664,9 @@ def report_cli( rdagent report -o custom/reports/ """ import subprocess - import sys from pathlib import Path - project_root = Path(__file__).parent.parent.parent.parent + project_root = Path(__file__).parent.parent.parent script = project_root / "scripts" / "predix_strategy_report.py" if not script.exists(): diff --git a/scripts/predix_parallel.py b/scripts/predix_parallel.py index 70352353..ec3e0fc0 100644 --- a/scripts/predix_parallel.py +++ b/scripts/predix_parallel.py @@ -19,19 +19,16 @@ import sys import time from datetime import datetime from pathlib import Path -from typing import Dict, List, Optional from dotenv import load_dotenv from rich.console import Console from rich.live import Live +from rich.markdown import Markdown from rich.panel import Panel from rich.table import Table -from rich.markdown import Markdown -from rich.layout import Layout -from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskProgressColumn # Load environment variables from .env file -load_dotenv(Path(__file__).parent / ".env") +load_dotenv(Path(__file__).parent.parent / ".env") console = Console() @@ -43,12 +40,12 @@ class RunState: self.run_id = run_id self.api_key_idx = api_key_idx self.model = model - self.process: Optional[subprocess.Popen] = None + self.process: subprocess.Popen | None = None self.status: str = "pending" # pending, running, success, failed, stopped - self.start_time: Optional[datetime] = None - self.end_time: Optional[datetime] = None - self.exit_code: Optional[int] = None - self.error_message: Optional[str] = None + self.start_time: datetime | None = None + self.end_time: datetime | None = None + self.exit_code: int | None = None + self.error_message: str | None = None self.log_file: str = f"fin_quant_run{run_id}.log" @property @@ -105,8 +102,8 @@ class ParallelRunner: self.num_runs = num_runs self.num_api_keys = num_api_keys self.model = model - self.runs: List[RunState] = [] - self.project_root = Path(__file__).parent + self.runs: list[RunState] = [] + self.project_root = Path(__file__).parent.parent self._shutdown_requested = False # Read API keys from environment @@ -115,10 +112,10 @@ class ParallelRunner: # Validate we have enough API keys if self.model == "openrouter" and len(self.api_keys) < num_api_keys: console.print( - f"[yellow]⚠️ Requested {num_api_keys} API keys, but only {len(self.api_keys)} found in .env[/yellow]" + f"[yellow]⚠️ Requested {num_api_keys} API keys, but only {len(self.api_keys)} found in .env[/yellow]", ) console.print( - f"[dim]Distributing across {len(self.api_keys)} available key(s)[/dim]" + f"[dim]Distributing across {len(self.api_keys)} available key(s)[/dim]", ) self.num_api_keys = len(self.api_keys) @@ -129,7 +126,7 @@ class ParallelRunner: run_state = RunState(run_id=i, api_key_idx=api_key_idx, model=model) self.runs.append(run_state) - def _load_api_keys(self) -> List[str]: + def _load_api_keys(self) -> list[str]: """Load API keys from environment variables.""" keys = [] @@ -149,7 +146,7 @@ class ParallelRunner: return keys - def _build_env(self, run_state: RunState) -> Dict[str, str]: + def _build_env(self, run_state: RunState) -> dict[str, str]: """ Build isolated environment for a subprocess. @@ -174,16 +171,14 @@ class ParallelRunner: env["RD_AGENT_WORKSPACE"] = str(workspace_dir) # Configure API key for this run - if self.model == "openrouter" and run_state.api_key_idx < len(self.api_keys): - api_key = self.api_keys[run_state.api_key_idx] - env["OPENAI_API_KEY"] = api_key - env["OPENAI_API_BASE"] = "https://openrouter.ai/api/v1" - env["CHAT_MODEL"] = os.getenv("OPENROUTER_MODEL", "openrouter/google/gemma-4-26b-a4b-it:free") - - # If we configured multiple API keys AND have enough keys, use load balancing + if self.model == "openrouter": if self.num_api_keys >= 2 and len(self.api_keys) >= 2: env["OPENAI_API_KEY"] = f"{self.api_keys[0]},{self.api_keys[1]}" env["LITELLM_PARALLEL_CALLS"] = "2" + elif run_state.api_key_idx < len(self.api_keys): + env["OPENAI_API_KEY"] = self.api_keys[run_state.api_key_idx] + env["OPENAI_API_BASE"] = "https://openrouter.ai/api/v1" + env["CHAT_MODEL"] = os.getenv("OPENROUTER_MODEL", "openrouter/google/gemma-4-26b-a4b-it:free") elif self.model == "local": env["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "local") env["OPENAI_API_BASE"] = os.getenv("OPENAI_API_BASE", "http://localhost:8081/v1") @@ -191,7 +186,7 @@ class ParallelRunner: return env - def _build_command(self, run_state: RunState) -> List[str]: + def _build_command(self, run_state: RunState) -> list[str]: """ Build the subprocess command to run predix quant. @@ -248,7 +243,7 @@ class ParallelRunner: console.print( f"[dim] ▶️ Run {run_state.run_id} started (PID: {run_state.process.pid}, " - f"API Key: {run_state.api_key_idx + 1}, Model: {run_state.model})[/dim]" + f"API Key: {run_state.api_key_idx + 1}, Model: {run_state.model})[/dim]", ) def _check_run(self, run_state: RunState) -> None: @@ -273,14 +268,14 @@ class ParallelRunner: run_state.status = "success" console.print( f"[bold green] ✅ Run {run_state.run_id} completed " - f"({run_state.elapsed})[/bold green]" + f"({run_state.elapsed})[/bold green]", ) else: run_state.status = "failed" run_state.error_message = f"Exit code: {poll_result}" console.print( f"[bold red] ❌ Run {run_state.run_id} failed " - f"({run_state.elapsed}, exit code: {poll_result})[/bold red]" + f"({run_state.elapsed}, exit code: {poll_result})[/bold red]", ) def _stop_run(self, run_state: RunState) -> None: @@ -386,7 +381,7 @@ class ParallelRunner: if run.status == "running": self._stop_run(run) - def run(self) -> Dict[str, int]: + def run(self) -> dict[str, int]: """ Execute all parallel runs and show live dashboard. @@ -400,7 +395,7 @@ class ParallelRunner: signal.signal(signal.SIGTERM, self._signal_handler) console.print(f"\n[bold cyan]{'=' * 60}[/bold cyan]") - console.print(f"[bold cyan]🔀 Predix Parallel Runner[/bold cyan]") + console.print("[bold cyan]🔀 Predix Parallel Runner[/bold cyan]") console.print(f"[bold cyan]{'=' * 60}[/bold cyan]") console.print(f" Runs: {self.num_runs}") console.print(f" API Keys: {self.num_api_keys} ({len(self.api_keys)} available)") @@ -451,7 +446,7 @@ class ParallelRunner: stopped_count = sum(1 for r in self.runs if r.status == "stopped") console.print(f"\n[bold cyan]{'=' * 60}[/bold cyan]") - console.print(f"[bold cyan]📊 Parallel Run Summary[/bold cyan]") + console.print("[bold cyan]📊 Parallel Run Summary[/bold cyan]") console.print(f"[bold cyan]{'=' * 60}[/bold cyan]") console.print(f" ✅ Success: {success_count}/{self.num_runs}") console.print(f" ❌ Failed: {failed_count}/{self.num_runs}") @@ -463,7 +458,7 @@ class ParallelRunner: if run.start_time and run.end_time: delta = run.end_time - run.start_time console.print( - f" Run #{run.run_id}: {run.status} ({delta.total_seconds():.0f}s)" + f" Run #{run.run_id}: {run.status} ({delta.total_seconds():.0f}s)", ) return { @@ -478,7 +473,7 @@ def main( runs: int = 5, api_keys: int = 2, model: str = "openrouter", -) -> Dict[str, int]: +) -> dict[str, int]: """ Run multiple factor experiments in parallel. @@ -504,7 +499,7 @@ if __name__ == "__main__": import argparse parser = argparse.ArgumentParser( - description="Predix Parallel Runner - Run multiple factor experiments concurrently" + description="Predix Parallel Runner - Run multiple factor experiments concurrently", ) parser.add_argument( "--runs", "-n", @@ -542,7 +537,7 @@ if __name__ == "__main__": elif args.runs > 25: console.print(f"\n[yellow]⚠️ {args.runs} runs - high resource usage expected[/yellow]") console.print(f" Estimated RAM: ~{args.runs * 0.65:.0f} GB") - console.print(f" Use --force to confirm.\n") + console.print(" Use --force to confirm.\n") import time time.sleep(2)