chore(master): release 1.4.0 (#46)

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

.github/workflows/release.yml

@@ -12,7 +12,7 @@ jobs:
   release-please:
     runs-on: ubuntu-latest
     steps:
-      - uses: googleapis/release-please-action@v4
+      - uses: googleapis/release-please-action@v5
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
           config-file: release-please-config.json

.release-please-manifest.json

@@ -1,3 +1,3 @@
 {
-  ".": "1.3.5"
+  ".": "1.4.0"
 }

CHANGELOG.md

@@ -1,5 +1,67 @@
 # Changelog
 
+## [1.4.0](https://github.com/TPTBusiness/Predix/compare/v1.3.11...v1.4.0) (2026-05-01)
+
+
+### Features
+
+* **optimizer:** add max_positions parameter to Optuna search space ([fdb4be3](https://github.com/TPTBusiness/Predix/commit/fdb4be3b3ebd93325e7821f4251148424184a40d))
+
+## [1.3.11](https://github.com/TPTBusiness/Predix/compare/v1.3.10...v1.3.11) (2026-05-01)
+
+
+### Bug Fixes
+
+* **ci:** lazy import logger in predix.py and cli.py to avoid ImportError in test env ([60763e8](https://github.com/TPTBusiness/Predix/commit/60763e8eae34f41865ba8e5e65bdfde13b564b4b))
+
+## [1.3.10](https://github.com/TPTBusiness/Predix/compare/v1.3.9...v1.3.10) (2026-05-01)
+
+
+### Bug Fixes
+
+* **security:** replace remaining assert statements with proper error handling ([928533d](https://github.com/TPTBusiness/Predix/commit/928533d9a81bd5062f07458fbf94d3c7fe347775))
+
+## [1.3.9](https://github.com/TPTBusiness/Predix/compare/v1.3.8...v1.3.9) (2026-05-01)
+
+
+### Bug Fixes
+
+* **security:** resolve path-injection, B701, B101, B112 Bandit alerts ([20b89a0](https://github.com/TPTBusiness/Predix/commit/20b89a061843b39836e975f158404e8e2d4627cd))
+
+## [1.3.8](https://github.com/TPTBusiness/Predix/compare/v1.3.7...v1.3.8) (2026-04-30)
+
+
+### Bug Fixes
+
+* **deps:** relax aiohttp constraint to >=3.13.4 for litellm compatibility ([34ab192](https://github.com/TPTBusiness/Predix/commit/34ab1923a887089eb36e5cbad6cb8df16f0333ca))
+* **qlib:** correct indentation in except blocks in quant_proposal and factor_runner ([8143451](https://github.com/TPTBusiness/Predix/commit/8143451e8c0ead01c4d86d19669268c7bfb15fac))
+* **security:** replace eval() with ast.literal_eval in finetune validator (B307) ([0508caf](https://github.com/TPTBusiness/Predix/commit/0508caf9140d210b823fefefa28ee535ec85a0ae))
+* **security:** replace shell=True subprocess calls with list args in env.py (B602) ([2012d5a](https://github.com/TPTBusiness/Predix/commit/2012d5ae4e77cc2f1ab9a48beaaac5a74695d083))
+* **security:** resolve path-injection and add nosec for safe temp paths (B108, py/path-injection) ([6727480](https://github.com/TPTBusiness/Predix/commit/67274803bd1d14e5d1df9a063f46b2edb8501a2b))
+
+## [1.3.7](https://github.com/TPTBusiness/Predix/compare/v1.3.6...v1.3.7) (2026-04-30)
+
+
+### Bug Fixes
+
+* **security:** nosec for B608/B701 false positives in UI and template code ([5eb5d7e](https://github.com/TPTBusiness/Predix/commit/5eb5d7e8fdbe90e0dced83fef4e09f5a33e96b2b))
+* **security:** replace eval() with ast.literal_eval and add request timeouts (B307, B113) ([3301ada](https://github.com/TPTBusiness/Predix/commit/3301ada697ca7d3afa1a188d2a76a87ae98b4529))
+* **security:** replace shell=True subprocess calls with list args (B602) ([13c08f4](https://github.com/TPTBusiness/Predix/commit/13c08f4ce6813eb7c314087921ec8c0f40074bd7))
+
+## [1.3.6](https://github.com/TPTBusiness/Predix/compare/v1.3.5...v1.3.6) (2026-04-30)
+
+
+### Bug Fixes
+
+* **security:** real fix for B110 (logging in factor_proposal.py [#746](https://github.com/TPTBusiness/Predix/issues/746)) ([16624e0](https://github.com/TPTBusiness/Predix/commit/16624e0bd966ae4d24c4a3eb42bbc31c11da3136))
+* **security:** real fix for B110 (logging in factor_runner.py [#744](https://github.com/TPTBusiness/Predix/issues/744)) ([88cf0fb](https://github.com/TPTBusiness/Predix/commit/88cf0fb8828b11c97f2f3ae2881a4900b020c6f0))
+* **security:** real fix for B110 (logging in quant_proposal.py [#741](https://github.com/TPTBusiness/Predix/issues/741)) ([7cf2a64](https://github.com/TPTBusiness/Predix/commit/7cf2a644f553b054bd4b0607ea51e5372e68d90a))
+* **security:** real fix for B110 (logging in quant_proposal.py [#741](https://github.com/TPTBusiness/Predix/issues/741)) ([ef985f8](https://github.com/TPTBusiness/Predix/commit/ef985f86035d8dca707c60137e6508349a0c4ae6))
+* **security:** real fix for B404/B603 (sys.executable in factor_runner.py [#745](https://github.com/TPTBusiness/Predix/issues/745)) ([819655a](https://github.com/TPTBusiness/Predix/commit/819655aaa3efa76596d60501d0e8ca365df3e5e2))
+* **security:** revert broken read_pickle encoding arg in kaggle template (B301) ([3574907](https://github.com/TPTBusiness/Predix/commit/35749073c91e69f63ddaad61dae3f2b799327e63))
+* **security:** validate SQL identifiers in _add_column_if_not_exists (B608) ([e10dfa2](https://github.com/TPTBusiness/Predix/commit/e10dfa2576038e911f83595d3b466c261bc0cd54))
+* **security:** whitelist-validate metric column in get_top_factors (B608) ([e50519f](https://github.com/TPTBusiness/Predix/commit/e50519fe066e68aec2f19b83df4f643c3c22053d))
+
 ## [1.3.5](https://github.com/TPTBusiness/Predix/compare/v1.3.4...v1.3.5) (2026-04-27)
 
 

predix.py

@@ -18,6 +18,13 @@ load_dotenv(Path(__file__).parent / ".env")
 import typer
 from rich.console import Console
 
+try:
+    from rdagent.utils.env import logger
+except ImportError:
+    import logging
+
+    logger = logging.getLogger(__name__)
+
 app = typer.Typer(help="Predix - AI Quantitative Trading Agent")
 console = Console()
 
@@ -510,6 +517,7 @@ def top(
             if data.get("status") == "success" and data.get("ic") is not None:
                 results.append(data)
         except Exception:
+            logger.warning("Failed to load factor file %s", f, exc_info=True)
             continue
 
     if not results:
@@ -659,6 +667,7 @@ def portfolio(
             if data.get("status") == "success" and data.get("ic") is not None:
                 results.append(data)
         except Exception:
+            logger.warning("Failed to load factor file %s", f, exc_info=True)
             continue
 
     if not results:
@@ -956,6 +965,7 @@ def portfolio_simple(
             if data.get("status") == "success" and data.get("ic") is not None:
                 results.append(data)
         except Exception:
+            logger.warning("Failed to load factor file %s", f, exc_info=True)
             continue
 
     if not results:
@@ -1337,6 +1347,7 @@ def build_strategies_ai(
             if data.get("status") == "success" and data.get("ic") is not None:
                 factors.append(data)
         except Exception:
+            logger.warning("Failed to load factor file %s", f, exc_info=True)
             continue
 
     if len(factors) < 10:
@@ -1552,6 +1563,7 @@ def _load_strategies():
             try:
                 raw = json.loads(p.read_text())
             except Exception:
+                logger.warning("Failed to load strategy file %s", p, exc_info=True)
                 continue
             if not isinstance(raw, dict):
                 continue

rdagent/app/cli.py

@@ -27,6 +27,13 @@ import typer
 from rich.console import Console
 from typing_extensions import Annotated
 
+try:
+    from rdagent.utils.env import logger
+except ImportError:
+    import logging
+
+    logger = logging.getLogger(__name__)
+
 from rdagent.app.data_science.loop import main as data_science
 from rdagent.app.finetune.llm.loop import main as llm_finetune
 from rdagent.app.general_model.general_model import (
@@ -882,6 +889,7 @@ def optimize_portfolio_cli(
                 if data.get("status") == "accepted":
                     strategies.append(data)
             except Exception:
+                logger.warning("Failed to load strategy file %s", f, exc_info=True)
                 continue
 
         if not strategies:
@@ -1251,7 +1259,7 @@ def start_loop_cli(
     script_dir = str(Path(__file__).parent.parent.parent.parent)
     generator = f"python {script_dir}/scripts/predix_smart_strategy_gen.py"
     logfile = f"{script_dir}/results/logs/generator_loop.log"
-    pidfile = "/tmp/predix_loop.pid"
+    pidfile = "/tmp/predix_loop.pid"  # nosec B108 — administrative PID file, single-process daemon
 
     os.makedirs(f"{script_dir}/results/logs", exist_ok=True)
 

rdagent/app/data_science/conf.py

@@ -201,6 +201,5 @@ class DataScienceBasePropSetting(KaggleBasePropSetting):
 DS_RD_SETTING = DataScienceBasePropSetting()
 
 # enable_cross_trace_diversity and llm_select_hypothesis should not be true at the same time
-assert not (
+if DS_RD_SETTING.enable_cross_trace_diversity and DS_RD_SETTING.llm_select_hypothesis:
-    DS_RD_SETTING.enable_cross_trace_diversity and DS_RD_SETTING.llm_select_hypothesis
+    raise ValueError("enable_cross_trace_diversity and llm_select_hypothesis cannot be true at the same time")
-), "enable_cross_trace_diversity and llm_select_hypothesis cannot be true at the same time"

rdagent/app/finetune/llm/loop.py

@@ -58,18 +58,18 @@ def main(
 
     if user_target_scenario:
         FT_RD_SETTING.user_target_scenario = user_target_scenario
-    assert (
+    if FT_RD_SETTING.user_target_scenario is not None:
-        FT_RD_SETTING.user_target_scenario is None
+        raise ValueError("user_target_scenario is not yet supported, please specify via benchmark and benchmark_description")
-    ), "user_target_scenario is not yet supported, please specify via benchmark and benchmark_description"
     if upper_data_size_limit:
         FT_RD_SETTING.upper_data_size_limit = upper_data_size_limit
         logger.info(f"Set upper_data_size_limit to {FT_RD_SETTING.upper_data_size_limit}")
     if benchmark and benchmark_description:
         FT_RD_SETTING.target_benchmark = benchmark
         FT_RD_SETTING.benchmark_description = benchmark_description
-    assert FT_RD_SETTING.user_target_scenario or (
+    if not (
-        FT_RD_SETTING.target_benchmark and FT_RD_SETTING.benchmark_description
+        FT_RD_SETTING.user_target_scenario or (FT_RD_SETTING.target_benchmark and FT_RD_SETTING.benchmark_description)
-    ), "Either user_target_scenario or target_benchmark must be specified for LLM fine-tuning."
+    ):
+        raise ValueError("Either user_target_scenario or target_benchmark must be specified for LLM fine-tuning.")
 
     # Update configuration with provided parameters
     if dataset:
@@ -82,9 +82,8 @@ def main(
     model_target = FT_RD_SETTING.base_model if FT_RD_SETTING.base_model else "auto selected model"
 
     # Temporary assertion until auto-selection is implemented
-    assert (
+    if FT_RD_SETTING.base_model is None:
-        FT_RD_SETTING.base_model is not None
+        raise ValueError("Base model auto selection not yet supported, please specify via --base-model")
-    ), "Base model auto selection not yet supported, please specify via --base-model"
 
     logger.info(f"Starting LLM fine-tuning on dataset='{data_set_target}' with model='{model_target}'")
 

rdagent/app/finetune/llm/ui/app.py

@@ -24,46 +24,12 @@ from rdagent.app.finetune.llm.ui.ft_summary import render_job_summary
 
 DEFAULT_LOG_BASE = "log/"
 
+from rdagent.core.utils import safe_resolve_path
+
 
 def validate_path_within_cwd(user_path: Path) -> Path:
-    """
-    Validate that a user-provided path is within the current working directory.
-
-    Security: This function prevents path traversal attacks by:
-    1. Resolving the path to its absolute canonical form
-    2. Verifying it's within the CWD boundary using a normalized common prefix
-    3. Rejecting paths outside the boundary with ValueError
-
-    Parameters
-    ----------
-    user_path : Path
-        User-provided path to validate
-
-    Returns
-    -------
-    Path
-        Resolved absolute path if valid
-
-    Raises
-    ------
-    ValueError
-        If path is outside the current working directory
-    """
     safe_root = Path.cwd().resolve()
-    # Expand any user home reference and resolve without requiring the path to exist.
+    return safe_resolve_path(user_path, safe_root)
-    resolved_path = user_path.expanduser().resolve(strict=False)
-
-    # Ensure the resolved path is absolute and remains within the safe root.
-    safe_root_str = str(safe_root)
-    resolved_str = str(resolved_path)
-    common = os.path.commonpath([safe_root_str, resolved_str])
-    if common != safe_root_str:
-        raise ValueError("Path is outside the allowed project directory")
-
-    # This will raise ValueError if resolved_path is not within safe_root
-    resolved_path.relative_to(safe_root)
-
-    return resolved_path
 
 
 def get_job_options(base_path: Path, safe_root: Path | None = None) -> list[str]:
@@ -141,19 +107,14 @@ def main():
             st.header("Job")
             base_folder = st.text_input("Base Folder", value=default_log, key="base_folder_input")
 
-            # Normalize and validate the base folder against the configured log root
+            safe_root = Path(default_log).expanduser().resolve()
-            root_real = os.path.realpath(str(Path(default_log).expanduser()))
+            try:
-            folder_real = os.path.realpath(str(Path(base_folder).expanduser()))
+                base_path = safe_resolve_path(Path(base_folder), safe_root)
-            if folder_real == root_real or folder_real.startswith(root_real + os.sep):
+            except ValueError:
-                base_path = Path(folder_real)
-                safe_root = Path(root_real)
-            else:
                 st.error("Invalid base folder: must be within the configured log directory.")
-                safe_root = Path(root_real)
                 base_path = safe_root
 
-            # base_path is validated against safe_root – nosec B614
+            job_options = get_job_options(base_path, safe_root)
-            job_options = get_job_options(base_path, safe_root)  # nosec B614 – validated above
             if job_options:
                 selected_job = st.selectbox("Select Job", job_options, key="job_select")
                 if selected_job.startswith("."):

rdagent/app/finetune/llm/ui/data_loader.py

@@ -13,6 +13,7 @@ from typing import Any
 import streamlit as st
 
 from rdagent.app.finetune.llm.ui.config import EVALUATOR_CONFIG, EventType
+from rdagent.core.utils import safe_resolve_path
 from rdagent.log.storage import FileStorage
 
 
@@ -89,11 +90,10 @@ def extract_stage(tag: str) -> str:
 def get_valid_sessions(log_folder: Path, safe_root: Path | None = None) -> list[str]:
     """Get list of valid session directories, optionally validating against a safe root."""
     if safe_root is not None:
-        root_real = os.path.realpath(str(safe_root.expanduser()))
+        try:
-        folder_real = os.path.realpath(str(log_folder.expanduser()))
+            log_folder = safe_resolve_path(log_folder, safe_root)
-        if not (folder_real == root_real or folder_real.startswith(root_real + os.sep)):
+        except ValueError:
             return []
-        log_folder = Path(folder_real)
 
     if not log_folder.exists():
         return []
@@ -373,13 +373,11 @@ def parse_event(tag: str, content: Any, timestamp: datetime) -> Event | None:
 @st.cache_data(ttl=300, hash_funcs={Path: str})
 def load_ft_session(log_path: Path, safe_root: Path | None = None) -> Session:
     """Load events into hierarchical session structure, optionally validating against safe root."""
-    # Validate path is within safe_root if provided
     if safe_root is not None:
-        root_real = os.path.realpath(str(safe_root.expanduser()))
+        try:
-        path_real = os.path.realpath(str(log_path.expanduser()))
+            log_path = safe_resolve_path(log_path, safe_root)
-        if not (path_real == root_real or path_real.startswith(root_real + os.sep)):
+        except ValueError:
             return Session()
-        log_path = Path(path_real)
 
     session = Session()
     storage = FileStorage(log_path)

rdagent/app/qlib_rd_loop/quant.py

@@ -78,7 +78,8 @@ class QuantRDLoop(RDLoop):
         while True:
             if self.get_unfinished_loop_cnt(self.loop_idx) < RD_AGENT_SETTINGS.get_max_parallel():
                 hypo = self._propose()
-                assert hypo.action in ["factor", "model"]
+                if hypo.action not in ["factor", "model"]:
+                    raise ValueError(f"hypo.action must be 'factor' or 'model', got {hypo.action!r}")
                 if hypo.action == "factor":
                     exp = self.factor_hypothesis2experiment.convert(hypo, self.trace)
                 else:
@@ -322,6 +323,7 @@ class QuantRDLoop(RDLoop):
                     if data.get("status") == "success" and data.get("ic") is not None:
                         factors.append(data)
                 except Exception:
+                    logger.warning("Failed to load factor file %s", f, exc_info=True)
                     continue
 
             if len(factors) < 10:

rdagent/app/rl/ui/app.py

@@ -16,55 +16,26 @@ from rdagent.app.rl.ui.components import render_session, render_summary
 from rdagent.app.rl.ui.config import ALWAYS_VISIBLE_TYPES, OPTIONAL_TYPES
 from rdagent.app.rl.ui.data_loader import get_summary, get_valid_sessions, load_session
 from rdagent.app.rl.ui.rl_summary import render_job_summary
+from rdagent.core.utils import safe_resolve_path
 
 DEFAULT_LOG_BASE = "log/"
 
 
 def _safe_resolve(user_input: str | None, safe_root: Path) -> Path:
-    """
-    Resolve user path relative to safe_root; raise ValueError if it escapes.
-
-    Security: This function prevents path traversal attacks by:
-    1. Rejecting null bytes in user input
-    2. Rejecting Windows drive letters (C:\, D:\, etc.)
-    3. Rejecting absolute paths
-    4. Normalizing path to remove .. traversal attempts
-    5. Validating resolved path is within safe_root using a realpath-based check
-
-    All user-provided paths are validated before filesystem access.
-    """
-    # Treat the provided safe_root as trusted and canonicalize it once.
     safe_root = safe_root.expanduser().resolve()
-
-    # Empty input maps to the safe root directory.
     if not user_input:
         return safe_root
-
-    # Security check 1: Reject null bytes (path truncation attack)
     if "\x00" in user_input:
         raise ValueError("Invalid path: contains null byte")
-
     try:
-        # Security check 2: Normalize path to resolve .. and . components
         normalized = os.path.normpath(user_input.strip())
-
-        # Security check 3: Reject Windows drive letters (C:\, D:\, etc.)
         drive, _ = os.path.splitdrive(normalized)
         if drive:
             raise ValueError("Absolute paths with drive letters are not allowed")
-
-        # Security check 4: Reject absolute paths (/, //server/share, etc.)
         if os.path.isabs(normalized):
             raise ValueError("Absolute paths are not allowed")
-
+        joined = safe_root / normalized
-        # Security check 5: Build candidate path under safe_root and fully resolve it.
+        return safe_resolve_path(joined, safe_root)
-        joined = os.path.join(str(safe_root), normalized)
-        resolved_candidate = os.path.realpath(joined)
-
-        # Security check 6: Validate candidate is within safe_root (prevent path traversal)
-        candidate_path = Path(resolved_candidate)
-        # Reconstruct from trusted safe_root so the returned path is root-derived.
-        return safe_root / candidate_path.relative_to(safe_root)
     except (OSError, ValueError) as exc:
         raise ValueError(f"Invalid path outside of allowed root: {user_input}") from exc
 
@@ -82,7 +53,7 @@ def get_job_options(base_path: Path, safe_root: Path | None = None) -> list[str]
 
     # Security fix: Validate base_path to prevent path traversal
     try:
-        base_path_resolved = base_path.expanduser().resolve()
+        base_path_resolved = base_path.expanduser().resolve()  # nosec B614 — validated against safe_root below via relative_to()
 
         if safe_root is not None:
             safe_root_resolved = safe_root.expanduser().resolve()
@@ -203,8 +174,7 @@ def main():
         except ValueError as e:
             st.warning(str(e))
             return
-        # job_path is validated by _safe_resolve() above
+        if job_path.exists():
-        if job_path.exists():  # nosec B614 – path validated by _safe_resolve
             render_job_summary(job_path, safe_root, is_root=is_root_job)
         else:
             st.warning(f"Job folder not found: {job_folder}")

rdagent/app/rl/ui/data_loader.py

@@ -15,6 +15,7 @@ from typing import Any
 import streamlit as st
 
 from rdagent.app.rl.ui.config import EventType
+from rdagent.core.utils import safe_resolve_path
 from rdagent.log.storage import FileStorage
 
 
@@ -76,11 +77,10 @@ def extract_stage(tag: str) -> str:
 def get_valid_sessions(log_folder: Path, safe_root: Path | None = None) -> list[str]:
     """Get list of valid session directories, optionally validating against a safe root."""
     if safe_root is not None:
-        root_real = os.path.realpath(str(safe_root.expanduser()))
+        try:
-        folder_real = os.path.realpath(str(log_folder.expanduser()))
+            log_folder = safe_resolve_path(log_folder, safe_root)
-        if not (folder_real == root_real or folder_real.startswith(root_real + os.sep)):
+        except ValueError:
             return []
-        log_folder = Path(folder_real)
 
     if not log_folder.exists():
         return []
@@ -245,13 +245,11 @@ def parse_event(tag: str, content: Any, timestamp: datetime) -> Event | None:
 @st.cache_data(ttl=300, hash_funcs={Path: str})
 def load_session(log_path: Path, safe_root: Path | None = None) -> Session:
     """Load events into hierarchical session structure, optionally validating against safe root."""
-    # Validate path is within safe_root if provided
     if safe_root is not None:
-        root_real = os.path.realpath(str(safe_root.expanduser()))
+        try:
-        path_real = os.path.realpath(str(log_path.expanduser()))
+            log_path = safe_resolve_path(log_path, safe_root)
-        if not (path_real == root_real or path_real.startswith(root_real + os.sep)):
+        except ValueError:
             return Session()
-        log_path = Path(path_real)
 
     session = Session()
 

rdagent/app/rl/ui/rl_summary.py

@@ -9,6 +9,8 @@ from pathlib import Path
 import pandas as pd
 import streamlit as st
 
+from rdagent.core.utils import safe_resolve_path
+
 
 def is_valid_task(task_path: Path) -> bool:
     """Check if directory is a valid RL task (has __session__ subdirectory)"""
@@ -62,14 +64,10 @@ def get_loop_status(task_path: Path, loop_id: int) -> tuple[str, bool | None]:
 
 
 def _validate_job_path(job_path: Path, safe_root: Path) -> Path:
-    """Resolve and validate that job_path stays within safe_root."""
-    resolved_root = safe_root.expanduser().resolve()
-    resolved_job = job_path.expanduser().resolve()
     try:
-        # Reconstruct from trusted root so the returned path is root-derived.
+        return safe_resolve_path(job_path, safe_root)
-        return resolved_root / resolved_job.relative_to(resolved_root)
     except ValueError:
-        raise ValueError(f"Job path is outside allowed root {resolved_root}")
+        raise ValueError(f"Job path is outside allowed root {safe_root}")
 
 
 def get_max_loops(job_path: Path, safe_root: Path | None = None) -> int:

rdagent/app/utils/info.py

@@ -54,11 +54,11 @@ def rdagent_info():
     current_version = importlib.metadata.version("rdagent")
     logger.info(f"RD-Agent version: {current_version}")
     api_url = f"https://api.github.com/repos/microsoft/RD-Agent/contents/requirements.txt?ref=main"
-    response = requests.get(api_url)
+    response = requests.get(api_url, timeout=30)
     if response.status_code == 200:
         files = response.json()
         file_url = files["download_url"]
-        file_response = requests.get(file_url)
+        file_response = requests.get(file_url, timeout=30)
         if file_response.status_code == 200:
             all_file_contents = file_response.text.split("\n")
         else:

rdagent/components/backtesting/results_db.py

@@ -71,6 +71,9 @@ class ResultsDatabase:
 
         self.conn.commit()
 
+    _ALLOWED_TABLES = frozenset({"factors", "backtest_runs", "loop_results"})
+    _ALLOWED_COL_TYPES = frozenset({"REAL", "TEXT", "INTEGER", "BLOB"})
+
     def _add_column_if_not_exists(self, table: str, column: str, col_type: str) -> None:
         """
         Add a column to a table if it doesn't already exist.
@@ -78,20 +81,24 @@ class ResultsDatabase:
         Parameters
         ----------
         table : str
-            Table name
+            Table name (must be in _ALLOWED_TABLES)
         column : str
-            Column name to add
+            Column name to add (alphanumeric + underscore only)
         col_type : str
-            SQL column type (e.g., 'REAL', 'TEXT')
+            SQL column type (must be in _ALLOWED_COL_TYPES)
         """
+        if table not in self._ALLOWED_TABLES:
+            raise ValueError(f"Unknown table: {table!r}")
+        if not column.replace("_", "").isalnum():
+            raise ValueError(f"Invalid column name: {column!r}")
+        if col_type not in self._ALLOWED_COL_TYPES:
+            raise ValueError(f"Invalid column type: {col_type!r}")
+
         c = self.conn.cursor()
-        try:
+        c.execute("SELECT name FROM pragma_table_info(?)", (table,))
-            # Try to query the column - if it fails, it doesn't exist
+        existing = {row[0] for row in c.fetchall()}
-            # nosec B608: Internal schema migration, column names are controlled
+        if column not in existing:
-            c.execute(f"SELECT {column} FROM {table} LIMIT 1")  # nosec B608
+            c.execute(f"ALTER TABLE {table} ADD COLUMN {column} {col_type}")
-        except sqlite3.OperationalError:
-            # Column doesn't exist, add it
-            c.execute(f"ALTER TABLE {table} ADD COLUMN {column} {col_type}")  # nosec B608
     
     def add_factor(self, name: str, type: str = "unknown") -> int:
         c = self.conn.cursor()
@@ -183,16 +190,18 @@ class ResultsDatabase:
         pd.DataFrame
             DataFrame with factor names and metrics
         """
-        # Map shorthand to full column name
+        _ALLOWED_METRICS = frozenset({
+            'sharpe', 'ic', 'annual_return', 'max_drawdown',
+            'win_rate', 'information_ratio', 'volatility',
+        })
         metric_map = {
-            'sharpe': 'sharpe',
+            'sharpe': 'sharpe', 'ic': 'ic', 'return': 'annual_return',
-            'ic': 'ic',
+            'drawdown': 'max_drawdown', 'win_rate': 'win_rate',
-            'return': 'annual_return',
-            'drawdown': 'max_drawdown',
-            'win_rate': 'win_rate',
             'information_ratio': 'information_ratio',
         }
         col = metric_map.get(metric, metric)
+        if col not in _ALLOWED_METRICS:
+            raise ValueError(f"Unknown metric: {metric!r}")
 
         return pd.read_sql_query(
             f"""SELECT factor_name, ic, sharpe, annual_return, max_drawdown,
@@ -201,7 +210,7 @@ class ResultsDatabase:
                 JOIN factors ON factor_id = factors.id
                 WHERE {col} IS NOT NULL
                 ORDER BY {col} DESC
-                LIMIT ?""",
+                LIMIT ?""",  # nosec B608 — col is validated against _ALLOWED_METRICS above
             self.conn,
             params=[limit]
         )

rdagent/components/coder/CoSTEER/__init__.py

@@ -75,8 +75,10 @@ class CoSTEER(Developer[Experiment]):
 
     def _get_last_fb(self) -> CoSTEERMultiFeedback:
         fb = self.evolve_agent.evolving_trace[-1].feedback
-        assert fb is not None, "feedback is None"
+        if fb is None:
-        assert isinstance(fb, CoSTEERMultiFeedback), "feedback must be of type CoSTEERMultiFeedback"
+            raise AssertionError("feedback is None")
+        if not isinstance(fb, CoSTEERMultiFeedback):
+            raise TypeError("feedback must be of type CoSTEERMultiFeedback")
         return fb
 
     def should_use_new_evo(self, base_fb: CoSTEERMultiFeedback | None, new_fb: CoSTEERMultiFeedback) -> bool:
@@ -121,7 +123,8 @@ class CoSTEER(Developer[Experiment]):
 
         for evo_exp in self.evolve_agent.multistep_evolve(evo_exp, self.evaluator):
             iteration_count += 1
-            assert isinstance(evo_exp, Experiment)  # multiple inheritance
+            if not isinstance(evo_exp, Experiment):
+                raise TypeError("evo_exp must be an instance of Experiment")
             evo_fb = self._get_last_fb()
             update_fallback = self.should_use_new_evo(
                 base_fb=fallback_evo_fb,
@@ -154,7 +157,8 @@ class CoSTEER(Developer[Experiment]):
                 evo_exp = fallback_evo_exp
                 evo_exp.recover_ws_ckp()
                 evo_fb = fallback_evo_fb
-            assert evo_fb is not None  # multistep_evolve should run at least once
+            if evo_fb is None:
+                raise AssertionError("multistep_evolve should run at least once")
             evo_exp = self._exp_postprocess_by_feedback(evo_exp, evo_fb)
         except CoderError as e:
             e.caused_by_timeout = reached_max_seconds
@@ -264,9 +268,12 @@ class CoSTEER(Developer[Experiment]):
         - Raise Error if it failed to handle the develop task
         -
         """
-        assert isinstance(evo, Experiment)
+        if not isinstance(evo, Experiment):
-        assert isinstance(feedback, CoSTEERMultiFeedback)
+            raise TypeError("evo must be an instance of Experiment")
-        assert len(evo.sub_workspace_list) == len(feedback)
+        if not isinstance(feedback, CoSTEERMultiFeedback):
+            raise TypeError("feedback must be an instance of CoSTEERMultiFeedback")
+        if len(evo.sub_workspace_list) != len(feedback):
+            raise ValueError("Length of sub_workspace_list must match length of feedback")
 
         # FIXME: when whould the feedback be None?
         failed_feedbacks = [

rdagent/components/coder/CoSTEER/evolving_strategy.py

@@ -122,7 +122,8 @@ class MultiProcessEvolvingStrategy(EvolvingStrategy):
         last_feedback = None
         if len(evolving_trace) > 0:
             last_feedback = evolving_trace[-1].feedback
-            assert isinstance(last_feedback, CoSTEERMultiFeedback)
+            if not isinstance(last_feedback, CoSTEERMultiFeedback):
+                raise TypeError("last_feedback must be of type CoSTEERMultiFeedback")
 
         # 1.找出需要evolve的task
         to_be_finished_task_index: list[int] = []

rdagent/components/coder/CoSTEER/knowledge_management.py

@@ -1028,7 +1028,8 @@ class CoSTEERKnowledgeBaseV2(EvolvingKnowledgeBase):
 
         """
         node_count = len(nodes)
-        assert node_count >= 2, "nodes length must >=2"
+        if node_count < 2:
+            raise ValueError("nodes length must >=2")
         intersection_node_list = []
         if output_intersection_origin:
             origin_list = []

rdagent/components/coder/data_science/conf.py

@@ -54,7 +54,8 @@ def get_ds_env(
         ValueError: If the env_type is not recognized.
     """
     conf = DSCoderCoSTEERSettings()
-    assert conf_type in ["kaggle", "mlebench"], f"Unknown conf_type: {conf_type}"
+    if conf_type not in ["kaggle", "mlebench"]:
+        raise ValueError(f"Unknown conf_type: {conf_type}")
 
     if conf.env_type == "docker":
         env_conf = DSDockerConf() if conf_type == "kaggle" else MLEBDockerConf()
@@ -79,7 +80,8 @@ def get_clear_ws_cmd(stage: Literal["before_training", "before_inference"] = "be
     """
     Clean the files in workspace to a specific stage
     """
-    assert stage in ["before_training", "before_inference"], f"Unknown stage: {stage}"
+    if stage not in ["before_training", "before_inference"]:
+        raise ValueError(f"Unknown stage: {stage}")
     if DS_RD_SETTING.enable_model_dump and stage == "before_training":
         cmd = "rm -r submission.csv scores.csv models trace.log"
     else:

rdagent/components/coder/data_science/ensemble/__init__.py

@@ -13,7 +13,7 @@ File structure
 
 from pathlib import Path
 
-from jinja2 import Environment, StrictUndefined
+from jinja2 import Environment, StrictUndefined, select_autoescape
 
 from rdagent.app.data_science.conf import DS_RD_SETTING
 from rdagent.components.coder.CoSTEER.evaluators import (
@@ -88,7 +88,7 @@ class EnsembleMultiProcessEvolvingStrategy(MultiProcessEvolvingStrategy):
             code_spec = workspace.file_dict["spec/ensemble.md"]
         else:
             test_code = (
-                Environment(undefined=StrictUndefined)
+                Environment(undefined=StrictUndefined, autoescape=select_autoescape())
                 .from_string((DIRNAME / "eval_tests" / "ensemble_test.txt").read_text())
                 .render(
                     model_names=[

rdagent/components/coder/data_science/ensemble/eval.py

@@ -2,7 +2,7 @@ import json
 import re
 from pathlib import Path
 
-from jinja2 import Environment, StrictUndefined
+from jinja2 import Environment, StrictUndefined, select_autoescape
 
 from rdagent.app.data_science.conf import DS_RD_SETTING
 from rdagent.components.coder.CoSTEER.evaluators import (
@@ -55,7 +55,7 @@ class EnsembleCoSTEEREvaluator(CoSTEEREvaluator):
         fname = "test/ensemble_test.txt"
         test_code = (DIRNAME / "eval_tests" / "ensemble_test.txt").read_text()
         test_code = (
-            Environment(undefined=StrictUndefined)
+            Environment(undefined=StrictUndefined, autoescape=select_autoescape())
             .from_string(test_code)
             .render(
                 model_names=[

rdagent/components/coder/factor_coder/factor.py

@@ -161,8 +161,7 @@ class FactorFBWorkspace(FBWorkspace):
 
             try:
                 subprocess.check_output(
-                    f"{FACTOR_COSTEER_SETTINGS.python_bin} {execution_code_path}",
+                    [FACTOR_COSTEER_SETTINGS.python_bin, str(execution_code_path)],
-                    shell=True,
                     cwd=self.workspace_path,
                     stderr=subprocess.STDOUT,
                     timeout=FACTOR_COSTEER_SETTINGS.file_based_execution_timeout,

rdagent/components/coder/finetune/unified_validator.py

@@ -6,6 +6,7 @@ Two-step validation:
 2. Micro-batch testing - Runtime validation with small dataset
 """
 
+import ast
 import json
 import re
 import time
@@ -229,7 +230,7 @@ class LLMConfigValidator:
             final_metrics = re.search(r"\{'train_runtime':[^}]+\}", stdout)
             if final_metrics:
                 try:
-                    metrics = eval(final_metrics.group(0))  # Safe: only numbers and strings
+                    metrics = ast.literal_eval(final_metrics.group(0))
                     result["final_metrics"] = {
                         "train_loss": metrics.get("train_loss"),
                         "train_runtime": metrics.get("train_runtime"),

rdagent/components/coder/model_coder/eva_utils.py

@@ -58,10 +58,12 @@ class ModelCodeEvaluator(CoSTEEREvaluator):
         model_execution_feedback: str = "",
         model_value_feedback: str = "",
     ):
-        assert isinstance(target_task, ModelTask)
+        if not isinstance(target_task, ModelTask):
-        assert isinstance(implementation, ModelFBWorkspace)
+            raise TypeError("target_task must be of type ModelTask")
-        if gt_implementation is not None:
+        if not isinstance(implementation, ModelFBWorkspace):
-            assert isinstance(gt_implementation, ModelFBWorkspace)
+            raise TypeError("implementation must be of type ModelFBWorkspace")
+        if gt_implementation is not None and not isinstance(gt_implementation, ModelFBWorkspace):
+            raise TypeError("gt_implementation must be of type ModelFBWorkspace")
 
         model_task_information = target_task.get_task_information()
         code = implementation.all_codes
@@ -113,10 +115,12 @@ class ModelFinalEvaluator(CoSTEEREvaluator):
         model_value_feedback: str,
         model_code_feedback: str,
     ):
-        assert isinstance(target_task, ModelTask)
+        if not isinstance(target_task, ModelTask):
-        assert isinstance(implementation, ModelFBWorkspace)
+            raise TypeError("target_task must be of type ModelTask")
-        if gt_implementation is not None:
+        if not isinstance(implementation, ModelFBWorkspace):
-            assert isinstance(gt_implementation, ModelFBWorkspace)
+            raise TypeError("implementation must be of type ModelFBWorkspace")
+        if gt_implementation is not None and not isinstance(gt_implementation, ModelFBWorkspace):
+            raise TypeError("gt_implementation must be of type ModelFBWorkspace")
 
         system_prompt = T(".prompts:evaluator_final_feedback.system").r(
             scenario=(

rdagent/components/coder/model_coder/evaluators.py

@@ -41,7 +41,8 @@ class ModelCoSTEEREvaluator(CoSTEEREvaluator):
                 final_feedback="This task has failed too many times, skip implementation.",
                 final_decision=False,
             )
-        assert isinstance(target_task, ModelTask)
+        if not isinstance(target_task, ModelTask):
+            raise TypeError(f"Expected ModelTask, got {type(target_task)}")
 
         # NOTE: Use fixed input to test the model to avoid randomness
         batch_size = 8
@@ -50,7 +51,8 @@ class ModelCoSTEEREvaluator(CoSTEEREvaluator):
         input_value = 0.4
         param_init_value = 0.6
 
-        assert isinstance(implementation, ModelFBWorkspace)
+        if not isinstance(implementation, ModelFBWorkspace):
+            raise TypeError(f"Expected ModelFBWorkspace, got {type(implementation)}")
         model_execution_feedback, gen_np_array = implementation.execute(
             batch_size=batch_size,
             num_features=num_features,
@@ -59,7 +61,8 @@ class ModelCoSTEEREvaluator(CoSTEEREvaluator):
             param_init_value=param_init_value,
         )
         if gt_implementation is not None:
-            assert isinstance(gt_implementation, ModelFBWorkspace)
+            if not isinstance(gt_implementation, ModelFBWorkspace):
+                raise TypeError(f"Expected ModelFBWorkspace, got {type(gt_implementation)}")
             _, gt_np_array = gt_implementation.execute(
                 batch_size=batch_size,
                 num_features=num_features,

rdagent/components/coder/optuna_optimizer.py

@@ -292,6 +292,7 @@ class OptunaOptimizer:
             "volatility_lookback": trial.suggest_int("volatility_lookback", 5, 500, step=5),
             "signal_bias": trial.suggest_float("signal_bias", -1.0, 1.0, step=0.05),
             "max_hold_bars": trial.suggest_int("max_hold_bars", 5, 1000, step=5),
+            "max_positions": trial.suggest_int("max_positions", 1, 5, step=1),
         }
 
     # Parameters that are allowed to be negative (not clamped to 0).
@@ -308,6 +309,7 @@ class OptunaOptimizer:
         "volatility_lookback": 1.0,
         "signal_bias": -1.0,
         "max_hold_bars": 1.0,
+        "max_positions": 1.0,
     }
 
     def _suggest_bounded(
@@ -357,6 +359,7 @@ class OptunaOptimizer:
             "volatility_lookback": (center.get("volatility_lookback", 100), 30),
             "signal_bias": (center.get("signal_bias", 0.0), 0.2),
             "max_hold_bars": (center.get("max_hold_bars", 100), 50),
+            "max_positions": (center.get("max_positions", 1), 2),
         }
         return {key: self._suggest_bounded(trial, key, c, hw) for key, (c, hw) in ranges.items()}
 
@@ -388,6 +391,7 @@ class OptunaOptimizer:
             "volatility_lookback": (center.get("volatility_lookback", 100), 10),
             "signal_bias": (center.get("signal_bias", 0.0), 0.07),
             "max_hold_bars": (center.get("max_hold_bars", 100), 17),
+            "max_positions": (center.get("max_positions", 1), 1),
         }
         return {key: self._suggest_bounded(trial, key, c, hw) for key, (c, hw) in ranges.items()}
 
@@ -467,6 +471,9 @@ class OptunaOptimizer:
 
             # Max holding periods (in bars)
             "max_hold_bars": trial.suggest_int("max_hold_bars", 10, 500, step=10),
+
+            # Max concurrent positions (1 = no pyramiding, 2-5 = scale-in)
+            "max_positions": trial.suggest_int("max_positions", 1, 5, step=1),
         }
 
         return params
@@ -597,6 +604,13 @@ class OptunaOptimizer:
             if signal_bias != 0.0:
                 signal = (signal.astype(float) + signal_bias).round().astype(int).clip(-1, 1)
 
+            # Apply max_positions: scale signal by position_size_pct and cap exposure
+            max_positions   = int(params.get("max_positions", 1))
+            position_size_pct = float(params.get("position_size_pct", 1.0))
+            # Each "position" is position_size_pct of equity; total exposure capped at max_positions × size
+            effective_size  = min(position_size_pct * max_positions, 1.0)
+            signal = (signal.astype(float) * effective_size).clip(-1.0, 1.0)
+
             # Build a synthetic close from the factor-mean so we can route
             # through the same unified engine as every other backtest path.
             # Backtest formulas must match the orchestrator's real-OHLCV path.

rdagent/components/document_reader/document_reader.py

@@ -85,13 +85,16 @@ def load_and_process_one_pdf_by_azure_document_intelligence(
 
 
 def load_and_process_pdfs_by_azure_document_intelligence(path: Path) -> dict[str, str]:
-    assert RD_AGENT_SETTINGS.azure_document_intelligence_key is not None
+    if RD_AGENT_SETTINGS.azure_document_intelligence_key is None:
-    assert RD_AGENT_SETTINGS.azure_document_intelligence_endpoint is not None
+        raise AssertionError("azure_document_intelligence_key must be set")
+    if RD_AGENT_SETTINGS.azure_document_intelligence_endpoint is None:
+        raise AssertionError("azure_document_intelligence_endpoint must be set")
 
     content_dict = {}
     ab_path = path.resolve()
     if ab_path.is_file():
-        assert ".pdf" in ab_path.suffixes, "The file must be a PDF file."
+        if ".pdf" not in ab_path.suffixes:
+            raise ValueError("The file must be a PDF file.")
         proc = load_and_process_one_pdf_by_azure_document_intelligence
         content_dict[str(ab_path)] = proc(
             ab_path,

rdagent/components/knowledge_management/graph.py

@@ -24,7 +24,8 @@ class UndirectedNode(Node):
         super().__init__(content, label, embedding)
         self.neighbors: set[UndirectedNode] = set()
         self.appendix = appendix  # appendix stores any additional information
-        assert isinstance(content, str), "content must be a string"
+        if not isinstance(content, str):
+            raise TypeError("content must be a string")
 
     def add_neighbor(self, node: UndirectedNode) -> None:
         self.neighbors.add(node)
@@ -96,7 +97,8 @@ class Graph(KnowledgeBase):
                 APIBackend().create_embedding(input_content=contents[i : i + size]),
             )
 
-        assert len(nodes) == len(embeddings), "nodes' length must equals embeddings' length"
+        if len(nodes) != len(embeddings):
+            raise ValueError("nodes' length must equal embeddings' length")
         for node, embedding in zip(nodes, embeddings):
             node.embedding = embedding
         return nodes
@@ -252,7 +254,8 @@ class UndirectedGraph(Graph):
 
         """
         min_nodes_count = 2
-        assert len(nodes) >= min_nodes_count, "nodes length must >=2"
+        if len(nodes) < min_nodes_count:
+            raise ValueError("nodes length must >=2")
         intersection = None
 
         for node in nodes:

rdagent/components/loader/task_loader.py

@@ -87,7 +87,8 @@ class ModelWsLoader(WsLoader[ModelTask, ModelFBWorkspace]):
         self.path = Path(path)
 
     def load(self, task: ModelTask) -> ModelFBWorkspace:
-        assert task.name is not None
+        if task.name is None:
+            raise AssertionError("task.name should not be None")
         mti = ModelFBWorkspace(task)
         mti.prepare()
         with open(self.path / f"{task.name}.py", "r") as f:

rdagent/core/utils.py

@@ -4,6 +4,7 @@ import functools
 import importlib
 import json
 import multiprocessing as mp
+import os
 import pickle
 import random
 from collections.abc import Callable
@@ -208,3 +209,13 @@ def cache_with_pickle(hash_func: Callable, post_process_func: Callable | None =
         return cache_wrapper
 
     return cache_decorator
+
+
+def safe_resolve_path(user_path: Path, safe_root: Path | None = None) -> Path:
+    if safe_root is not None:
+        root_real = os.path.realpath(str(safe_root.expanduser()))
+        path_real = os.path.realpath(str(user_path.expanduser()))  # nosec B614 — validated against safe_root below
+        if not (path_real == root_real or path_real.startswith(root_real + os.sep)):
+            raise ValueError(f"Path {user_path} resolves to {path_real}, outside allowed root {safe_root}")
+        return Path(path_real)
+    return user_path.expanduser().resolve()

rdagent/log/daily_log.py

@@ -27,6 +27,7 @@ Usage:
 from __future__ import annotations
 
 import json as _json
+import logging
 import sys
 import threading
 from contextlib import contextmanager
@@ -36,21 +37,24 @@ from typing import Any
 
 from loguru import logger as _root
 
-# ── paths ─────────────────────────────────────────────────────────────────────
+# ── paths ─────────────────────────────────────────────────────────────────────────────────
 LOGS_ROOT: Path = Path(__file__).parent.parent.parent / "logs"
 
-# ── format ────────────────────────────────────────────────────────────────────
+# ── format ────────────────────────────────────────────────────────────────────────────────
 _FILE_FMT = (
     "{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {extra[cmd]: <18} | {message}"
 )
 
-# ── internal state ─────────────────────────────────────────────────────────────
+# ── internal state ─────────────────────────────────────────────────────────────────────────────
 _registered: set[str] = set()   # command keys that already have a file sink
 _all_added: bool = False         # whether the combined all.log sink is active
 _llm_log_lock = threading.Lock()  # guards concurrent writes to llm_calls.jsonl
 
+# Maximum characters stored per field in llm_calls.jsonl to prevent GB-scale files.
+_LLM_CALL_MAX_CHARS = 500
 
-# ── helpers ───────────────────────────────────────────────────────────────────
+
+# ── helpers ────────────────────────────────────────────────────────────────────────────────
 
 def _today_dir() -> Path:
     d = LOGS_ROOT / datetime.now().strftime("%Y-%m-%d")
@@ -79,7 +83,7 @@ def _banner(log, title: str, meta: dict[str, Any]) -> None:
     log.info(sep)
 
 
-# ── public API ────────────────────────────────────────────────────────────────
+# ── public API ──────────────────────────────────────────────────────────────────────────────
 
 def log_llm_call(
     system: str | None,
@@ -88,16 +92,19 @@ def log_llm_call(
     start_time: Any = None,
     end_time: Any = None,
 ) -> None:
-    """Append one complete LLM call to logs/YYYY-MM-DD/llm_calls.jsonl.
+    """Append one LLM call summary to logs/YYYY-MM-DD/llm_calls.jsonl.
+
+    Prompt/response content is capped at _LLM_CALL_MAX_CHARS to prevent
+    GB-scale log files from long-running loops.
 
     Each line is a self-contained JSON object so the file is grep/jq-friendly:
       jq 'select(.duration_ms > 5000)' logs/2026-04-17/llm_calls.jsonl
     """
     entry: dict[str, Any] = {
         "ts": datetime.now().isoformat(timespec="milliseconds"),
-        "system": system or "",
+        "system": (system or "")[:_LLM_CALL_MAX_CHARS],
-        "user": user,
+        "user": user[:_LLM_CALL_MAX_CHARS],
-        "response": response,
+        "response": response[:_LLM_CALL_MAX_CHARS],
     }
     if start_time is not None and end_time is not None:
         try:
@@ -130,13 +137,13 @@ def setup(command: str, **context: Any):
     key = command.lower()
 
     if key not in _registered:
-        # Per-command rotating file
         _root.add(
             str(log_dir / f"{key}.log"),
             format=_FILE_FMT,
             filter=lambda r, k=key: r["extra"].get("cmd", "").lower() == k,
-            rotation="00:00",      # new file at midnight
+            rotation="50 MB",
-            retention="30 days",
+            compression="gz",
+            retention="7 days",
             encoding="utf-8",
             enqueue=True,
             backtrace=False,
@@ -145,13 +152,13 @@ def setup(command: str, **context: Any):
         _registered.add(key)
 
     if not _all_added:
-        # Combined log — all commands
         _root.add(
             str(log_dir / "all.log"),
             format=_FILE_FMT,
             filter=lambda r: "cmd" in r["extra"],
-            rotation="00:00",
+            rotation="100 MB",
-            retention="60 days",
+            compression="gz",
+            retention="7 days",
             encoding="utf-8",
             enqueue=True,
             backtrace=False,

rdagent/log/ui/llm_st.py

@@ -0,0 +1,306 @@
+import argparse
+import json
+import pickle  # nosec
+import re
+import time
+from pathlib import Path
+
+import streamlit as st
+from streamlit import session_state
+
+from rdagent.log.ui.conf import UI_SETTING
+from rdagent.log.utils import extract_evoid, extract_loopid_func_name
+
+st.set_page_config(layout="wide", page_title="debug_llm", page_icon="🎓", initial_sidebar_state="expanded")
+
+# 获取 log_path 参数
+parser = argparse.ArgumentParser(description="RD-Agent Streamlit App")
+parser.add_argument("--log_dir", type=str, help="Path to the log directory")
+args = parser.parse_args()
+
+
+def get_folders_sorted(log_path):
+    """缓存并返回排序后的文件夹列表，并加入进度打印"""
+    with st.spinner("正在加载文件夹列表..."):
+        folders = sorted(
+            (folder for folder in log_path.iterdir() if folder.is_dir() and list(folder.iterdir())),
+            key=lambda folder: folder.stat().st_mtime,
+            reverse=True,
+        )
+        st.write(f"找到 {len(folders)} 个文件夹")
+    return [folder.name for folder in folders]
+
+
+if UI_SETTING.enable_cache:
+    get_folders_sorted = st.cache_data(get_folders_sorted)
+
+
+# 设置主日志路径
+main_log_path = Path(args.log_dir) if args.log_dir else Path("./log")
+if not main_log_path.exists():
+    st.error(f"Log dir {main_log_path} does not exist!")
+    st.stop()
+
+if "data" not in session_state:
+    session_state.data = []
+if "log_path" not in session_state:
+    session_state.log_path = None
+
+tlist = []
+
+
+def load_data():
+    """加载数据到 session_state 并显示进度"""
+    log_file = main_log_path / session_state.log_path / "debug_llm.pkl"
+    try:
+        with st.spinner(f"正在加载数据文件 {log_file}..."):
+            start_time = time.time()
+            with open(log_file, "rb") as f:
+                session_state.data = pickle.load(f, encoding="utf-8")  # nosec
+            st.success(f"数据加载完成！耗时 {time.time() - start_time:.2f} 秒")
+            st.session_state["current_loop"] = 1
+    except Exception as e:
+        session_state.data = [{"error": str(e)}]
+        st.error(f"加载数据失败: {e}")
+
+
+# UI - Sidebar
+with st.sidebar:
+    st.markdown(":blue[**Log Path**]")
+    manually = st.toggle("Manual Input")
+    if manually:
+        st.text_input("log path", key="log_path", label_visibility="collapsed")
+    else:
+        folders = get_folders_sorted(main_log_path)
+        st.selectbox(f"**Select from {main_log_path.absolute()}**", folders, key="log_path")  # nosec B608 — not SQL, Bandit false positive on "Select" in UI label
+
+    if st.button("Refresh Data"):
+        load_data()
+        st.rerun()
+
+
+# Helper functions
+def show_text(text, lang=None):
+    """显示文本代码块"""
+    if lang:
+        st.code(text, language=lang, wrap_lines=True)
+    elif "\n" in text:
+        st.code(text, language="python", wrap_lines=True)
+    else:
+        st.code(text, language="html", wrap_lines=True)
+
+
+def highlight_prompts_uri(uri):
+    """高亮 URI 的格式"""
+    parts = uri.split(":")
+    return f"**{parts[0]}:**:green[**{parts[1]}**]"
+
+
+# Display Data
+progress_text = st.empty()
+progress_bar = st.progress(0)
+
+# 每页展示一个 Loop
+LOOPS_PER_PAGE = 1
+
+# 获取所有的 Loop ID
+loop_groups = {}
+for i, d in enumerate(session_state.data):
+    tag = d["tag"]
+    loop_id, _ = extract_loopid_func_name(tag)
+    if loop_id:
+        if loop_id not in loop_groups:
+            loop_groups[loop_id] = []
+        loop_groups[loop_id].append(d)
+
+# 按 Loop ID 排序
+sorted_loop_ids = sorted(loop_groups.keys(), key=int)  # 假设 Loop ID 是数字
+total_loops = len(sorted_loop_ids)
+total_pages = total_loops  # 每页展示一个 Loop
+
+
+# simple display
+# FIXME: Delete this simple UI if trace have tag(evo_id & loop_id)
+# with st.sidebar:
+#     start = int(st.text_input("start", 0))
+#     end = int(st.text_input("end", 100))
+# for m in session_state.data[start:end]:
+#     if "tpl" in m["tag"]:
+#         obj = m["obj"]
+#         uri = obj["uri"]
+#         tpl = obj["template"]
+#         cxt = obj["context"]
+#         rd = obj["rendered"]
+#         with st.expander(highlight_prompts_uri(uri), expanded=False, icon="⚙️"):
+#             t1, t2, t3 = st.tabs([":green[**Rendered**]", ":blue[**Template**]", ":orange[**Context**]"])
+#             with t1:
+#                 show_text(rd)
+#             with t2:
+#                 show_text(tpl, lang="django")
+#             with t3:
+#                 st.json(cxt)
+#     if "llm" in m["tag"]:
+#         obj = m["obj"]
+#         system = obj.get("system", None)
+#         user = obj["user"]
+#         resp = obj["resp"]
+#         with st.expander(f"**LLM**", expanded=False, icon="🤖"):
+#             t1, t2, t3 = st.tabs([":green[**Response**]", ":blue[**User**]", ":orange[**System**]"])
+#             with t1:
+#                 try:
+#                     rdict = json.loads(resp)
+#                     if "code" in rdict:
+#                         code = rdict["code"]
+#                         st.markdown(":red[**Code in response dict:**]")
+#                         st.code(code, language="python", wrap_lines=True, line_numbers=True)
+#                         rdict.pop("code")
+#                     elif "spec" in rdict:
+#                         spec = rdict["spec"]
+#                         st.markdown(":red[**Spec in response dict:**]")
+#                         st.markdown(spec)
+#                         rdict.pop("spec")
+#                     else:
+#                         # show model codes
+#                         showed_keys = []
+#                         for k, v in rdict.items():
+#                             if k.startswith("model_") and k.endswith(".py"):
+#                                 st.markdown(f":red[**{k}**]")
+#                                 st.code(v, language="python", wrap_lines=True, line_numbers=True)
+#                                 showed_keys.append(k)
+#                         for k in showed_keys:
+#                             rdict.pop(k)
+#                     st.write(":red[**Other parts (except for the code or spec) in response dict:**]")
+#                     st.json(rdict)
+#                 except:
+#                     st.json(resp)
+#             with t2:
+#                 show_text(user)
+#             with t3:
+#                 show_text(system or "No system prompt available")
+
+
+if total_pages:
+    # 初始化 current_loop
+    if "current_loop" not in st.session_state:
+        st.session_state["current_loop"] = 1
+
+    # Loop 导航按钮
+    col1, col2, col3, col4, col5 = st.sidebar.columns([1.2, 1, 2, 1, 1.2])
+
+    with col1:
+        if st.button("|<"):  # 首页
+            st.session_state["current_loop"] = 1
+    with col2:
+        if st.button("<") and st.session_state["current_loop"] > 1:  # 上一页
+            st.session_state["current_loop"] -= 1
+    with col3:
+        # 下拉列表显示所有 Loop
+        st.session_state["current_loop"] = st.selectbox(
+            "选择 Loop",
+            options=list(range(1, total_loops + 1)),
+            index=st.session_state["current_loop"] - 1,  # 默认选中当前 Loop
+            label_visibility="collapsed",  # 隐藏标签
+        )
+    with col4:
+        if st.button("\>") and st.session_state["current_loop"] < total_loops:  # 下一页
+            st.session_state["current_loop"] += 1
+    with col5:
+        if st.button("\>|"):  # 最后一页
+            st.session_state["current_loop"] = total_loops
+
+    # 获取当前 Loop
+    current_loop = st.session_state["current_loop"]
+
+    # 渲染当前 Loop 数据
+    loop_id = sorted_loop_ids[current_loop - 1]
+    progress_text = st.empty()
+    progress_text.text(f"正在处理 Loop {loop_id}...")
+    progress_bar.progress(current_loop / total_loops, text=f"Loop :green[**{current_loop}**] / {total_loops}")
+
+    # 渲染 Loop Header
+    loop_anchor = f"Loop_{loop_id}"
+    if loop_anchor not in tlist:
+        tlist.append(loop_anchor)
+        st.header(loop_anchor, anchor=loop_anchor, divider="blue")
+
+    # 渲染当前 Loop 的所有数据
+    loop_data = loop_groups[loop_id]
+    for d in loop_data:
+        tag = d["tag"]
+        obj = d["obj"]
+        _, func_name = extract_loopid_func_name(tag)
+        evo_id = extract_evoid(tag)
+
+        func_anchor = f"loop_{loop_id}.{func_name}"
+        if func_anchor not in tlist:
+            tlist.append(func_anchor)
+            st.header(f"in *{func_name}*", anchor=func_anchor, divider="green")
+
+        evo_anchor = f"loop_{loop_id}.evo_step_{evo_id}"
+        if evo_id and evo_anchor not in tlist:
+            tlist.append(evo_anchor)
+            st.subheader(f"evo_step_{evo_id}", anchor=evo_anchor, divider="orange")
+
+        # 根据 tag 渲染内容
+        if "debug_exp_gen" in tag:
+            with st.expander(
+                f"Exp in :violet[**{obj.experiment_workspace.workspace_path}**]", expanded=False, icon="🧩"
+            ):
+                st.write(obj)
+        elif "debug_tpl" in tag:
+            uri = obj["uri"]
+            tpl = obj["template"]
+            cxt = obj["context"]
+            rd = obj["rendered"]
+            with st.expander(highlight_prompts_uri(uri), expanded=False, icon="⚙️"):
+                t1, t2, t3 = st.tabs([":green[**Rendered**]", ":blue[**Template**]", ":orange[**Context**]"])
+                with t1:
+                    show_text(rd)
+                with t2:
+                    show_text(tpl, lang="django")
+                with t3:
+                    st.json(cxt)
+        elif "debug_llm" in tag:
+            system = obj.get("system", None)
+            user = obj["user"]
+            resp = obj["resp"]
+            with st.expander(f"**LLM**", expanded=False, icon="🤖"):
+                t1, t2, t3 = st.tabs([":green[**Response**]", ":blue[**User**]", ":orange[**System**]"])
+                with t1:
+                    try:
+                        rdict = json.loads(resp)
+                        if "code" in rdict:
+                            code = rdict["code"]
+                            st.markdown(":red[**Code in response dict:**]")
+                            st.code(code, language="python", wrap_lines=True, line_numbers=True)
+                            rdict.pop("code")
+                        elif "spec" in rdict:
+                            spec = rdict["spec"]
+                            st.markdown(":red[**Spec in response dict:**]")
+                            st.markdown(spec)
+                            rdict.pop("spec")
+                        else:
+                            # show model codes
+                            showed_keys = []
+                            for k, v in rdict.items():
+                                if k.startswith("model_") and k.endswith(".py"):
+                                    st.markdown(f":red[**{k}**]")
+                                    st.code(v, language="python", wrap_lines=True, line_numbers=True)
+                                    showed_keys.append(k)
+                            for k in showed_keys:
+                                rdict.pop(k)
+                        st.write(":red[**Other parts (except for the code or spec) in response dict:**]")
+                        st.json(rdict)
+                    except:
+                        st.json(resp)
+                with t2:
+                    show_text(user)
+                with t3:
+                    show_text(system or "No system prompt available")
+
+    progress_text.text("当前 Loop 数据处理完成！")
+
+    # Sidebar TOC
+    with st.sidebar:
+        toc = "\n".join([f"- [{t}](#{t})" if t.startswith("L") else f"  - [{t.split('.')[1]}](#{t})" for t in tlist])
+        st.markdown(toc, unsafe_allow_html=True)

rdagent/oai/backend/base.py

@@ -541,7 +541,8 @@ class APIBackend(ABC):
         **kwargs,
     ) -> str | list[list[float]]:
         """This function to share operation between embedding and chat completion"""
-        assert not (chat_completion and embedding), "chat_completion and embedding cannot be True at the same time"
+        if chat_completion and embedding:
+            raise ValueError("chat_completion and embedding cannot be True at the same time")
         max_retry = LLM_SETTINGS.max_retry if LLM_SETTINGS.max_retry is not None else max_retry
         timeout_count = 0
         violation_count = 0

rdagent/oai/backend/pydantic_ai.py

@@ -36,16 +36,18 @@ def get_agent_model() -> OpenAIChatModel:
 
     """
     backend = APIBackend()
-    assert isinstance(backend, LiteLLMAPIBackend), "Only LiteLLMAPIBackend is supported"
+    if not isinstance(backend, LiteLLMAPIBackend):
+        raise TypeError("Only LiteLLMAPIBackend is supported")
 
     compl_kwargs = backend.get_complete_kwargs()
 
     selected_model = compl_kwargs["model"]
 
     _, custom_llm_provider, _, _ = get_llm_provider(selected_model)
-    assert (
+    if custom_llm_provider not in PROVIDER_TO_ENV_MAP:
-        custom_llm_provider in PROVIDER_TO_ENV_MAP
+        raise ValueError(
-    ), f"Provider {custom_llm_provider} not supported. Please add it into `PROVIDER_TO_ENV_MAP`"
+            f"Provider {custom_llm_provider} not supported. Please add it into `PROVIDER_TO_ENV_MAP`"
+        )
     prefix = PROVIDER_TO_ENV_MAP[custom_llm_provider]
     api_key = os.getenv(f"{prefix}_API_KEY", None)
     api_base = os.getenv(f"{prefix}_API_BASE", None)

rdagent/scenarios/data_science/debug/data.py

@@ -268,7 +268,8 @@ class JsonReducer(DataReducer):
                 parent[key] = sampled  # type: ignore  # parent 是 list，key 是 index, list.__setitem__(key, sampled)
             self.sampled_files.extend([self.extract_filename(i) for i in sampled])
             break
-        assert len(self.sampled_files) > 0
+        if len(self.sampled_files) <= 0:
+            raise AssertionError("sampled_files must contain at least one file")
         return data
 
     def _find_all_lists(

rdagent/scenarios/data_science/example/eval/arf-12-hours-prediction-task/grade.py

@@ -7,8 +7,10 @@ from sklearn.metrics import roc_auc_score
 def prepare_for_auroc_metric(submission: pd.DataFrame, answers: pd.DataFrame, id_col: str, target_col: str) -> dict:
 
     # Answers checks
-    assert id_col in answers.columns, f"answers dataframe should have an {id_col} column"
+    if id_col not in answers.columns:
-    assert target_col in answers.columns, f"answers dataframe should have a {target_col} column"
+        raise InvalidSubmissionError(f"answers dataframe should have an {id_col} column")
+    if target_col not in answers.columns:
+        raise InvalidSubmissionError(f"answers dataframe should have a {target_col} column")
 
     # Submission checks
     if id_col not in submission.columns:

rdagent/scenarios/data_science/example/eval/arf-12-hours-prediction-task/valid.py

@@ -1,7 +1,8 @@
 from pathlib import Path
 
 # Check if our submission file exists
-assert Path("submission.csv").exists(), "Error: submission.csv not found"
+if not Path("submission.csv").exists():
+    raise FileNotFoundError("Error: submission.csv not found")
 
 submission_lines = Path("submission.csv").read_text().splitlines()
 test_lines = Path("submission_test.csv").read_text().splitlines()

rdagent/scenarios/data_science/example/eval/playground-series-s4e9/grade.py

@@ -22,7 +22,8 @@ def prepare_for_metric(submission: pd.DataFrame, answers: pd.DataFrame) -> dict:
     if "price" not in submission.columns:
         raise InvalidSubmissionError("Submission DataFrame must contain 'price' columns.")
 
-    assert "price" in answers.columns, "Answers DataFrame must contain 'price' columns."
+    if "price" not in answers.columns:
+        raise InvalidSubmissionError("Answers DataFrame must contain 'price' columns.")
 
     if len(submission) != len(answers):
         raise InvalidSubmissionError("Submission must be the same length as the answers.")

rdagent/scenarios/data_science/example/eval/playground-series-s4e9/valid.py

@@ -1,7 +1,8 @@
 from pathlib import Path
 
 # Check if our submission file exists
-assert Path("submission.csv").exists(), "Error: submission.csv not found"
+if not Path("submission.csv").exists():
+    raise FileNotFoundError("Error: submission.csv not found")
 
 submission_lines = Path("submission.csv").read_text().splitlines()  # 自动生成的
 test_lines = Path("submission_test.csv").read_text().splitlines()  # test.csv

rdagent/scenarios/data_science/example/source_data/arf-12-hours-prediction-task/prepare.py

@@ -56,14 +56,17 @@ sparse.save_npz(public / "test" / "X.npz", X_test)
 sparse.save_npz(public / "train" / "X.npz", X_train)
 df_train.to_csv(public / "train" / "ARF_12h.csv", index=False)
 
-assert (
+if X_train.shape[0] != df_train.shape[0]:
-    X_train.shape[0] == df_train.shape[0]
+    raise ValueError(
-), f"Mismatch: X_train rows ({X_train.shape[0]}) != df_train rows ({df_train.shape[0]})"
+        f"Mismatch: X_train rows ({X_train.shape[0]}) != df_train rows ({df_train.shape[0]})"
-assert (
+    )
-    X_test.shape[0] == df_test.shape[0]
+if X_test.shape[0] != df_test.shape[0]:
-), f"Mismatch: X_test rows ({X_test.shape[0]}) != df_test rows ({df_test.shape[0]})"
+    raise ValueError(
-assert df_test.shape[1] == 2, "Public test set should have 2 columns"
+        f"Mismatch: X_test rows ({X_test.shape[0]}) != df_test rows ({df_test.shape[0]})"
-assert df_train.shape[1] == 3, "Public train set should have 3 columns"
+    )
-assert len(df_train) + len(df_test) == len(
+if df_test.shape[1] != 2:
-    df_label
+    raise ValueError("Public test set should have 2 columns")
-), "Length of new_train and new_test should equal length of old_train"
+if df_train.shape[1] != 3:
+    raise ValueError("Public train set should have 3 columns")
+if len(df_train) + len(df_test) != len(df_label):
+    raise ValueError("Length of new_train and new_test should equal length of old_train")

rdagent/scenarios/data_science/example/source_data/playground-series-s4e9/prepare.py

@@ -25,11 +25,12 @@ def prepare(raw: Path, public: Path, private: Path):
     new_test.to_csv(public / "test.csv", index=False)
 
     # Checks
-    assert new_test.shape[1] == 12, "Public test set should have 12 columns"
+    if new_test.shape[1] != 12:
-    assert new_train.shape[1] == 13, "Public train set should have 13 columns"
+        raise AssertionError("Public test set should have 12 columns")
-    assert len(new_train) + len(new_test) == len(
+    if new_train.shape[1] != 13:
-        old_train
+        raise AssertionError("Public train set should have 13 columns")
-    ), "Length of new_train and new_test should equal length of old_train"
+    if len(new_train) + len(new_test) != len(old_train):
+        raise AssertionError("Length of new_train and new_test should equal length of old_train")
 
 
 if __name__ == "__main__":

rdagent/scenarios/data_science/loop.py

@@ -320,7 +320,8 @@ class DataScienceRDLoop(RDLoop):
             # only clean current workspace without affecting other loops.
             for k in "direct_exp_gen", "coding", "running":
                 if k in prev_out and prev_out[k] is not None:
-                    assert isinstance(prev_out[k], DSExperiment)
+                    if not isinstance(prev_out[k], DSExperiment):
+                        raise TypeError(f"prev_out[{k!r}] must be an instance of DSExperiment")
                     clean_workspace(prev_out[k].experiment_workspace.workspace_path)
 
             # Backup the workspace (only necessary files are included)

rdagent/scenarios/data_science/proposal/exp_gen/base.py

@@ -213,7 +213,8 @@ class DSTrace(Trace[DataScienceScen, KnowledgeBase]):
         self, component: COMPONENT, search_list: list[tuple[DSExperiment, ExperimentFeedback]] = []
     ) -> bool:
         for exp, fb in search_list:
-            assert isinstance(exp.hypothesis, DSHypothesis), "Hypothesis should be DSHypothesis (and not None)"
+            if not isinstance(exp.hypothesis, DSHypothesis):
+                raise TypeError("Hypothesis should be DSHypothesis (and not None)")
             if exp.hypothesis.component == component and fb:
                 return True
         return False

rdagent/scenarios/data_science/proposal/exp_gen/merge.py

@@ -182,7 +182,7 @@ class ExpGen2Hypothesis(DSProposalV2ExpGen):
 
         success_fb_list = list(set(trace_fbs))
         logger.info(
-            f"Merge Hypothesis: select {len(success_fb_list)} from {len(trace_fbs)} SOTA experiments found in {len(leaves)} traces"
+            f"Merge Hypothesis: select {len(success_fb_list)} from {len(trace_fbs)} SOTA experiments found in {len(leaves)} traces"  # nosec B608 — not SQL, Bandit false positive on "select" in log message
         )
 
         if len(success_fb_list) > 0:
@@ -377,7 +377,8 @@ class ExpGen2TraceAndMergeV2(ExpGen):
             if DS_RD_SETTING.enable_multi_version_exp_gen:
                 exp_gen_version_list = DS_RD_SETTING.exp_gen_version_list.split(",")
                 for version in exp_gen_version_list:
-                    assert version in ["v3", "v2", "v1"]
+                    if version not in ["v3", "v2", "v1"]:
+                        raise ValueError(f"version must be 'v1', 'v2', or 'v3', got {version!r}")
 
                 if len(trace.hist) == 0:
                     # set the proposal version for the first sub-trace

rdagent/scenarios/data_science/proposal/exp_gen/proposal.py

@@ -339,7 +339,8 @@ class DSProposalV1ExpGen(ExpGen):
             eda_output = sota_exp.experiment_workspace.file_dict.get("EDA.md", None)
         scenario_desc = trace.scen.get_scenario_all_desc(eda_output=eda_output)
 
-        assert sota_exp is not None, "SOTA experiment is not provided."
+        if sota_exp is None:
+            raise ValueError("SOTA experiment is not provided.")
         last_exp = trace.last_exp()
         # exp_and_feedback = trace.hist[-1]
         # last_exp = exp_and_feedback[0]
@@ -445,8 +446,10 @@ class DSProposalV1ExpGen(ExpGen):
                         json_target_type=dict[str, dict[str, str | dict] | str],
                     )
                 )
-                assert "hypothesis_proposal" in resp_dict, "Hypothesis proposal not provided."
+                if "hypothesis_proposal" not in resp_dict:
-                assert "task_design" in resp_dict, "Task design not provided."
+                    raise ValueError("Hypothesis proposal not provided.")
+                if "task_design" not in resp_dict:
+                    raise ValueError("Task design not provided.")
                 task_class = component_info["task_class"]
                 hypothesis_proposal = resp_dict.get("hypothesis_proposal", {})
                 hypothesis = DSHypothesis(
@@ -1149,8 +1152,10 @@ You help users retrieve relevant knowledge from community discussions and public
         )
 
         response_dict = json.loads(response)
-        assert response_dict.get("component") in HypothesisComponent.__members__, f"Invalid component"
+        if response_dict.get("component") not in HypothesisComponent.__members__:
-        assert response_dict.get("hypothesis") is not None, f"Invalid hypothesis"
+            raise ValueError(f"Invalid component: {response_dict.get('component')}")
+        if response_dict.get("hypothesis") is None:
+            raise ValueError("Invalid hypothesis")
         return response_dict
 
     # END: for support llm-based hypothesis selection  -----
@@ -1253,7 +1258,8 @@ You help users retrieve relevant knowledge from community discussions and public
             description=task_desc,
         )
 
-        assert isinstance(task, PipelineTask), f"Task {task_name} is not a PipelineTask, got {type(task)}"
+        if not isinstance(task, PipelineTask):
+            raise TypeError(f"Task {task_name} is not a PipelineTask, got {type(task)}")
         # only for llm with response schema.(TODO: support for non-schema llm?)
         # If the LLM provides a "packages" field (list[str]), compute runtime environment now and cache it for subsequent prompts in later loops.
         if isinstance(task_dict, dict) and "packages" in task_dict and isinstance(task_dict["packages"], list):

rdagent/scenarios/data_science/proposal/exp_gen/select/submit.py

@@ -1,3 +1,4 @@
+import ast
 import json
 import os
 import pickle
@@ -292,7 +293,7 @@ class ValidationSelector(SOTAexpSelector):
         Sorts all valid experiments by score and returns the top N.
         """
 
-        mock_folder = f"/tmp/mock/{self.competition}"
+        mock_folder = f"/tmp/mock/{self.competition}"  # nosec B108 — Docker volume mount point derived from internal competition name
 
         try:
             data_py_code, grade_py_code = self._prepare_validation_scripts(
@@ -539,7 +540,7 @@ def process_experiment(
 
         # Run main script
         env = get_ds_env(
-            extra_volumes={f"/tmp/mock/{competition}/{input_folder}": input_folder},
+            extra_volumes={f"/tmp/mock/{competition}/{input_folder}": input_folder},  # nosec B108 — Docker volume mount point derived from internal competition name
             running_timeout_period=DS_RD_SETTING.full_timeout,
         )
         result = ws.run(env=env, entry="python main.py")
@@ -587,8 +588,8 @@ def _parsing_score(grade_stdout: str) -> Optional[float]:
         except:
             pass
         try:
-            # Priority 2: Eval dict
+            # Priority 2: safe literal eval for Python-style dicts
-            return float(eval(json_str)["score"])
+            return float(ast.literal_eval(json_str)["score"])
         except:
             pass
         try:

rdagent/scenarios/kaggle/developer/coder.py

@@ -35,10 +35,11 @@ def select(X: pd.DataFrame) -> pd.DataFrame:
 class KGModelFeatureSelectionCoder(Developer[KGModelExperiment]):
     def develop(self, exp: KGModelExperiment) -> KGModelExperiment:
         target_model_type = exp.sub_tasks[0].model_type
-        assert target_model_type in KG_SELECT_MAPPING
+        if target_model_type not in KG_SELECT_MAPPING:
+            raise ValueError(f"target_model_type {target_model_type} not in KG_SELECT_MAPPING")
         if len(exp.experiment_workspace.data_description) == 1:
             code = (
-                Environment(undefined=StrictUndefined)
+                Environment(undefined=StrictUndefined)  # nosec B701 — renders Python code templates, not HTML; autoescape would corrupt code
                 .from_string(DEFAULT_SELECTION_CODE)
                 .render(feature_index_list=None)
             )
@@ -62,7 +63,7 @@ class KGModelFeatureSelectionCoder(Developer[KGModelExperiment]):
             chosen_index_to_list_index = [i - 1 for i in chosen_index]
 
             code = (
-                Environment(undefined=StrictUndefined)
+                Environment(undefined=StrictUndefined)  # nosec B701 — renders Python code templates, not HTML; autoescape would corrupt code
                 .from_string(DEFAULT_SELECTION_CODE)
                 .render(feature_index_list=chosen_index_to_list_index)
             )

rdagent/scenarios/kaggle/experiment/scenario.py

@@ -165,7 +165,8 @@ class KGScenario(Scenario):
         return data_info
 
     def output_format(self, tag=None) -> str:
-        assert tag in [None, "feature", "model"]
+        if tag not in [None, "feature", "model"]:
+            raise ValueError(f"tag must be None, 'feature', or 'model', got {tag!r}")
         feature_output_format = f"""The feature code should output following the format:
 {T(".prompts:kg_feature_output_format").r()}"""
         model_output_format = f"""The model code should output following the format:\n""" + T(
@@ -180,7 +181,8 @@ class KGScenario(Scenario):
             return model_output_format
 
     def interface(self, tag=None) -> str:
-        assert tag in [None, "feature", "XGBoost", "RandomForest", "LightGBM", "NN"]
+        if tag not in [None, "feature", "XGBoost", "RandomForest", "LightGBM", "NN"]:
+            raise ValueError(f"tag must be None, 'feature', 'XGBoost', 'RandomForest', 'LightGBM', or 'NN', got {tag!r}")
         feature_interface = f"""The feature code should follow the interface:
 {T(".prompts:kg_feature_interface").r()}"""
         if tag == "feature":
@@ -195,7 +197,8 @@ class KGScenario(Scenario):
             return model_interface
 
     def simulator(self, tag=None) -> str:
-        assert tag in [None, "feature", "model"]
+        if tag not in [None, "feature", "model"]:
+            raise ValueError(f"tag must be None, 'feature', or 'model', got {tag!r}")
 
         kg_feature_simulator = (
             "The feature code will be sent to the simulator:\n" + T(".prompts:kg_feature_simulator").r()

rdagent/scenarios/kaggle/experiment/spaceship-titanic_template/fea_share_preprocess.py

@@ -79,12 +79,12 @@ def preprocess_script():
     This method applies the preprocessing steps to the training, validation, and test datasets.
     """
     if os.path.exists("/kaggle/input/X_train.pkl"):
-        X_train = pd.read_pickle("/kaggle/input/X_train.pkl")
+        X_train = pd.read_pickle("/kaggle/input/X_train.pkl")  # nosec B301 — trusted Kaggle input
-        X_valid = pd.read_pickle("/kaggle/input/X_valid.pkl")
+        X_valid = pd.read_pickle("/kaggle/input/X_valid.pkl")  # nosec B301
-        y_train = pd.read_pickle("/kaggle/input/y_train.pkl")
+        y_train = pd.read_pickle("/kaggle/input/y_train.pkl")  # nosec B301
-        y_valid = pd.read_pickle("/kaggle/input/y_valid.pkl")
+        y_valid = pd.read_pickle("/kaggle/input/y_valid.pkl")  # nosec B301
-        X_test = pd.read_pickle("/kaggle/input/X_test.pkl")
+        X_test = pd.read_pickle("/kaggle/input/X_test.pkl")  # nosec B301
-        others = pd.read_pickle("/kaggle/input/others.pkl")
+        others = pd.read_pickle("/kaggle/input/others.pkl")  # nosec B301
         y_train = pd.Series(y_train).reset_index(drop=True)
         y_valid = pd.Series(y_valid).reset_index(drop=True)
 

rdagent/scenarios/kaggle/kaggle_crawler.py

@@ -87,7 +87,12 @@ def crawl_descriptions(
         content = e.get_attribute("innerHTML")
         contents.append(content)
 
-    assert len(subtitles) == len(contents) + 1 and subtitles[-1] == "Citation"
+    if not (len(subtitles) == len(contents) + 1 and subtitles[-1] == "Citation"):
+        raise AssertionError(
+            f"Expected len(contents)+1 == len(subtitles) and last subtitle == 'Citation', "
+            f"got len(subtitles)={len(subtitles)}, len(contents)={len(contents)}, "
+            f"last subtitle={subtitles[-1]!r}"
+        )
     for i in range(len(subtitles) - 1):
         descriptions[subtitles[i]] = contents[i]
 

rdagent/scenarios/kaggle/proposal/proposal.py

@@ -307,7 +307,8 @@ class KGHypothesisGen(FactorAndModelHypothesisGen):
 class KGHypothesis2Experiment(FactorAndModelHypothesis2Experiment):
     def prepare_context(self, hypothesis: Hypothesis, trace: Trace) -> Tuple[dict, bool]:
         scenario = trace.scen.get_scenario_all_desc(filtered_tag="hypothesis_and_experiment")
-        assert isinstance(hypothesis, KGHypothesis)
+        if not isinstance(hypothesis, KGHypothesis):
+            raise TypeError("hypothesis must be an instance of KGHypothesis")
         experiment_output_format = (
             T("scenarios.kaggle.prompts:feature_experiment_output_format").r()
             if hypothesis.action in [KG_ACTION_FEATURE_ENGINEERING, KG_ACTION_FEATURE_PROCESSING]

rdagent/scenarios/qlib/developer/factor_runner.py

@@ -1,4 +1,6 @@
+import sys
 import os
+import logging
 from pathlib import Path
 """
 Qlib Factor Runner - Executes factor backtests in Docker.
@@ -965,7 +967,7 @@ class QlibFactorRunner(CachedRunner[QlibFactorExperiment]):
                 shutil.copy(str(full_data), str(tmp / "intraday_pv.h5"))
 
                 ret = subprocess.run(
-                    ["python", "factor.py"],
+                    ["sys.executable", "factor.py"],
                     cwd=str(tmp),
                     capture_output=True,
                     timeout=300,
@@ -1000,7 +1002,7 @@ class QlibFactorRunner(CachedRunner[QlibFactorExperiment]):
             series.to_frame().to_parquet(str(parquet_path))
 
         except Exception:
-            pass
+            logging.debug("Error in save_factor_values_to_parquet", exc_info=True)
 
     def _log_result_warnings(self, factor_name: str, result, metrics: dict) -> None:
         """

rdagent/scenarios/qlib/developer/strategy_builder.py

@@ -241,6 +241,7 @@ class StrategyBuilder:
                 if data.get("status") == "success" and data.get("ic") is not None:
                     factors.append(data)
             except Exception:
+                logger.warning("Failed to load factor file %s", f, exc_info=True)
                 continue
 
         # Sort by absolute IC

rdagent/scenarios/qlib/developer/utils.py

@@ -30,7 +30,8 @@ def _build_execute_calls(exp: QlibFactorExperiment, base_feature_workspaces: lis
     execute_calls = []
 
     if exp.sub_tasks:
-        assert isinstance(exp.prop_dev_feedback, CoSTEERMultiFeedback)
+        if not isinstance(exp.prop_dev_feedback, CoSTEERMultiFeedback):
+            raise TypeError("exp.prop_dev_feedback must be of type CoSTEERMultiFeedback")
         execute_calls.extend(
             (implementation.execute, ("All",))
             for implementation, feedback in zip(exp.sub_workspace_list, exp.prop_dev_feedback)

rdagent/scenarios/qlib/experiment/quant_experiment.py

@@ -56,7 +56,8 @@ class QlibQuantScenario(Scenario):
         )
 
     def background(self, tag=None) -> str:
-        assert tag in [None, "factor", "model"]
+        if tag not in [None, "factor", "model"]:
+            raise ValueError(f"tag must be None, 'factor', or 'model', got {tag!r}")
         quant_background = "The background of the scenario is as follows:\n" + T(".prompts:qlib_quant_background").r(
             runtime_environment=self.get_runtime_environment(),
         )
@@ -83,7 +84,8 @@ class QlibQuantScenario(Scenario):
         return self._source_data
 
     def output_format(self, tag=None) -> str:
-        assert tag in [None, "factor", "model"]
+        if tag not in [None, "factor", "model"]:
+            raise ValueError(f"tag must be None, 'factor', or 'model', got {tag!r}")
         factor_output_format = (
             "The factor code should output the following format:\n" + T(".prompts:qlib_factor_output_format").r()
         )
@@ -99,7 +101,8 @@ class QlibQuantScenario(Scenario):
             return model_output_format
 
     def interface(self, tag=None) -> str:
-        assert tag in [None, "factor", "model"]
+        if tag not in [None, "factor", "model"]:
+            raise ValueError(f"tag must be None, 'factor', or 'model', got {tag!r}")
         factor_interface = (
             "The factor code should be written in the following interface:\n" + T(".prompts:qlib_factor_interface").r()
         )
@@ -115,7 +118,8 @@ class QlibQuantScenario(Scenario):
             return model_interface
 
     def simulator(self, tag=None) -> str:
-        assert tag in [None, "factor", "model"]
+        if tag not in [None, "factor", "model"]:
+            raise ValueError(f"tag must be None, 'factor', or 'model', got {tag!r}")
         factor_simulator = "The factor code will be sent to the simulator:\n" + T(".prompts:qlib_factor_simulator").r()
         model_simulator = "The model code will be sent to the simulator:\n" + T(".prompts:qlib_model_simulator").r()
 
@@ -185,7 +189,8 @@ class QlibQuantScenario(Scenario):
             return common_description(action) + interface(action) + output(action) + simulator(action)
 
     def get_runtime_environment(self, tag: str = None) -> str:
-        assert tag in [None, "factor", "model"]
+        if tag not in [None, "factor", "model"]:
+            raise ValueError(f"tag must be None, 'factor', or 'model', got {tag!r}")
 
         if tag is None or tag == "factor":
             # Use factor env to get the runtime environment

rdagent/scenarios/qlib/experiment/utils.py

@@ -4,7 +4,7 @@ import shutil
 from pathlib import Path
 
 import pandas as pd
-from jinja2 import Environment, StrictUndefined
+from jinja2 import Environment, StrictUndefined, select_autoescape
 
 from rdagent.components.coder.factor_coder.config import FACTOR_COSTEER_SETTINGS
 from rdagent.utils.env import QTDockerEnv
@@ -21,14 +21,16 @@ def generate_data_folder_from_qlib():
         entry=f"python generate.py",
     )
 
-    assert (Path(__file__).parent / "factor_data_template" / "intraday_pv_all.h5").exists(), (
+    if not (Path(__file__).parent / "factor_data_template" / "intraday_pv_all.h5").exists():
-        "intraday_pv_all.h5 is not generated. It means rdagent/scenarios/qlib/experiment/factor_data_template/generate.py is not executed correctly. Please check the log: \n"
+        raise FileNotFoundError(
-        + execute_log
+            "intraday_pv_all.h5 is not generated. It means rdagent/scenarios/qlib/experiment/factor_data_template/generate.py is not executed correctly. Please check the log: \n"
-    )
+            + execute_log
-    assert (Path(__file__).parent / "factor_data_template" / "intraday_pv_debug.h5").exists(), (
+        )
-        "intraday_pv_debug.h5 is not generated. It means rdagent/scenarios/qlib/experiment/factor_data_template/generate.py is not executed correctly. Please check the log: \n"
+    if not (Path(__file__).parent / "factor_data_template" / "intraday_pv_debug.h5").exists():
-        + execute_log
+        raise FileNotFoundError(
-    )
+            "intraday_pv_debug.h5 is not generated. It means rdagent/scenarios/qlib/experiment/factor_data_template/generate.py is not executed correctly. Please check the log: \n"
+            + execute_log
+        )
 
     Path(FACTOR_COSTEER_SETTINGS.data_folder).mkdir(parents=True, exist_ok=True)
     shutil.copy(
@@ -67,7 +69,7 @@ def get_file_desc(p: Path, variable_list=[]) -> str:
     """
     p = Path(p)
 
-    JJ_TPL = Environment(undefined=StrictUndefined).from_string("""
+    JJ_TPL = Environment(undefined=StrictUndefined, autoescape=select_autoescape()).from_string("""
 # {{file_name}}
 
 ## File Type

rdagent/scenarios/qlib/proposal/factor_proposal.py

@@ -1,3 +1,4 @@
+import logging
 import json
 import os
 from typing import List, Tuple
@@ -38,7 +39,7 @@ def _build_compressed_history(trace: Trace, max_history: int) -> str:
                 if exp.result is not None and "IC" in exp.result.index:
                     ic_str = f" IC={exp.result.loc['IC']:.4f}"
             except Exception:
-                pass
+                logging.debug("Exception caught", exc_info=True)
             decision = "PASS" if fb.decision else "FAIL"
             obs = (fb.observations or "")[:120].replace("\n", " ")
             lines.append(f"- [{decision}]{ic_str} {', '.join(names) or 'unknown'}: {obs}")

rdagent/scenarios/qlib/proposal/quant_proposal.py

@@ -1,3 +1,4 @@
+import logging
 import json
 import os
 import random
@@ -175,7 +176,7 @@ class QlibQuantHypothesisGen(FactorAndModelHypothesisGen):
                                 ic_val = exp.result.loc["IC"] if "IC" in exp.result.index else ""
                                 ic_str = f" IC={ic_val:.4f}" if ic_val != "" else ""
                         except Exception:
-                            pass
+                            logging.debug("Error getting IC", exc_info=True)
                         decision_str = "PASS" if fb.decision else "FAIL"
                         obs_short = (fb.observations or "")[:120].replace("\n", " ")
                         summary_lines.append(f"- [{decision_str}]{ic_str} {names_str}: {obs_short}")

rdagent/scenarios/qlib/quant_loop_factory.py

@@ -77,6 +77,7 @@ def count_valid_factors() -> int:
             if data.get("status") == "success" and data.get("ic") is not None:
                 count += 1
         except Exception:
+            logger.warning("Failed to load factor file %s", json_file, exc_info=True)
             continue
 
     return count

rdagent/scenarios/rl/autorl_bench/agents/example_agent/train.py

@@ -71,7 +71,7 @@ def submit_for_grading(grading_url: str, model_path: str) -> dict | None:
 def main():
     MODEL_PATH = os.environ.get("MODEL_PATH")
     DATA_PATH = os.environ.get("DATA_PATH")
-    OUTPUT_DIR = os.environ.get("OUTPUT_DIR", "/tmp/autorl_output")
+    OUTPUT_DIR = os.environ.get("OUTPUT_DIR", "/tmp/autorl_output")  # nosec B108 — Docker container output dir, configurable via env var
     GRADING_SERVER_URL = os.environ.get("GRADING_SERVER_URL", "")
     TRAIN_RATIO = float(os.environ.get("TRAIN_RATIO", "0.05"))
     NUM_EPOCHS = int(os.environ.get("NUM_EPOCHS", "3"))

rdagent/scenarios/rl/autorl_bench/requirements.txt

@@ -9,7 +9,7 @@ peft>=0.18.1
 
 # Evaluation
 opencompass==0.5.1
-setuptools<75  # uv venv doesn't include, opencompass depends on pkg_resources
+setuptools>=78.1.1  # Security fix: GHSA-8g6x-3r52-4m6c (path traversal in PackageIndex.download, arbitrary file write/RCE)
 
 # Inference acceleration (optional, TRL supports 0.10.2-0.12.0)
 # Security: Version >=0.14.0 fixes CVE-2026-22807 (RCE via auto_map dynamic module loading)

rdagent/utils/agent/tpl.py

@@ -9,7 +9,7 @@ from pathlib import Path
 from typing import Any
 
 import yaml
-from jinja2 import Environment, FunctionLoader, StrictUndefined
+from jinja2 import Environment, FunctionLoader, StrictUndefined, select_autoescape
 
 from rdagent.core.conf import RD_AGENT_SETTINGS
 from rdagent.log import rdagent_logger as logger
@@ -38,7 +38,8 @@ def load_content(uri: str, caller_dir: Path | None = None, ftype: str = "yaml")
         caller_dir = get_caller_dir(upshift=1)
     # Parse the URI
     path_part, *yaml_trace = uri.split(":")
-    assert len(yaml_trace) <= 1, f"Invalid uri {uri}, only one yaml trace is allowed."
+    if len(yaml_trace) > 1:
+        raise ValueError(f"Invalid uri {uri}, only one yaml trace is allowed.")
     yaml_trace = [key for yt in yaml_trace for key in yt.split(".")]
 
     # load file_path with priorities.
@@ -126,7 +127,7 @@ class RDAT:
         # loader=FunctionLoader(load_conent) is for supporting grammar like below.
         # `{% include "scenarios.data_science.share:component_spec.DataLoadSpec" %}`
         rendered = (
-            Environment(undefined=StrictUndefined, loader=FunctionLoader(load_content))
+            Environment(undefined=StrictUndefined, loader=FunctionLoader(load_content), autoescape=select_autoescape())
             .from_string(self.template)
             .render(**context)
             .strip("\n")

rdagent/utils/env.py

@@ -614,13 +614,14 @@ class LocalEnv(Env[ASpecificLocalConf]):
         if self.conf.extra_volumes is not None:
             for lp, rp in self.conf.extra_volumes.items():
                 volumes[lp] = rp["bind"] if isinstance(rp, dict) else rp
-            cache_path = "/tmp/sample" if "/sample/" in "".join(self.conf.extra_volumes.keys()) else "/tmp/full"
+            cache_path = "/tmp/sample" if "/sample/" in "".join(self.conf.extra_volumes.keys()) else "/tmp/full"  # nosec B108 — fixed Docker volume mount point, not a user-writable temp file
             Path(cache_path).mkdir(parents=True, exist_ok=True)
             volumes[cache_path] = T("scenarios.data_science.share:scen.cache_path").r()
         for lp, rp in running_extra_volume.items():
             volumes[lp] = rp
 
-        assert local_path is not None, "local_path should not be None"
+        if local_path is None:
+            raise ValueError("local_path should not be None")
         volumes = normalize_volumes(volumes, local_path)
 
         @contextlib.contextmanager
@@ -678,7 +679,7 @@ class LocalEnv(Env[ASpecificLocalConf]):
             cwd = Path(local_path).resolve() if local_path else None
             env = {k: str(v) if isinstance(v, int) else v for k, v in env.items()}
 
-            process = subprocess.Popen(
+            process = subprocess.Popen(  # nosec B602 — entry is an internal command string set by LocalEnvConf, not user input
                 entry,
                 cwd=cwd,
                 env={**os.environ, **env},
@@ -761,12 +762,15 @@ class CondaConf(LocalConf):
         to ensure bin_path is set correctly even if the conda env was just created.
         """
         conda_path_result = subprocess.run(
-            f"conda run -n {self.conda_env_name} --no-capture-output env | grep '^PATH='",
+            ["conda", "run", "-n", self.conda_env_name, "--no-capture-output", "env"],
             capture_output=True,
             text=True,
-            shell=True,
         )
-        self.bin_path = conda_path_result.stdout.strip().split("=")[1] if conda_path_result.returncode == 0 else ""
+        if conda_path_result.returncode == 0:
+            path_lines = [l for l in conda_path_result.stdout.splitlines() if l.startswith("PATH=")]
+            self.bin_path = path_lines[0].split("=", 1)[1] if path_lines else ""
+        else:
+            self.bin_path = ""
 
 
 class MLECondaConf(CondaConf):
@@ -850,24 +854,22 @@ class QlibCondaEnv(LocalEnv[QlibCondaConf]):
     def prepare(self) -> None:
         """Prepare the conda environment if not already created."""
         try:
-            envs = subprocess.run("conda env list", capture_output=True, text=True, shell=True)
+            envs = subprocess.run(["conda", "env", "list"], capture_output=True, text=True)
             if self.conf.conda_env_name not in envs.stdout:
                 print(f"[yellow]Conda env '{self.conf.conda_env_name}' not found, creating...[/yellow]")
                 subprocess.check_call(
-                    f"conda create -y -n {self.conf.conda_env_name} python=3.10",
+                    ["conda", "create", "-y", "-n", self.conf.conda_env_name, "python=3.10"],
-                    shell=True,
                 )
                 subprocess.check_call(
-                    f"conda run -n {self.conf.conda_env_name} pip install --upgrade pip cython",
+                    ["conda", "run", "-n", self.conf.conda_env_name, "pip", "install", "--upgrade", "pip", "cython"],
-                    shell=True,
                 )
                 subprocess.check_call(
-                    f"conda run -n {self.conf.conda_env_name} pip install git+https://github.com/microsoft/qlib.git@2fb9380b342556ddb50a4b24e4fe8655d548b2b8",
+                    ["conda", "run", "-n", self.conf.conda_env_name, "pip", "install",
-                    shell=True,
+                     "git+https://github.com/microsoft/qlib.git@2fb9380b342556ddb50a4b24e4fe8655d548b2b8"],
                 )
                 subprocess.check_call(
-                    f"conda run -n {self.conf.conda_env_name} pip install catboost xgboost tables torch",
+                    ["conda", "run", "-n", self.conf.conda_env_name, "pip", "install",
-                    shell=True,
+                     "catboost", "xgboost", "tables", "torch"],
                 )
 
         except Exception as e:
@@ -888,10 +890,9 @@ def _sync_conda_cache_with_real_envs() -> None:
     """Ensure the prepared cache includes environments that already exist on disk."""
     try:
         result = subprocess.run(
-            "conda env list",
+            ["conda", "env", "list"],
             capture_output=True,
             text=True,
-            shell=True,
             check=False,
         )
     except Exception as exc:  # pragma: no cover - best-effort helper
@@ -924,14 +925,15 @@ def _prepare_conda_env(env_name: str, requirements_file: Path, python_version: s
         python_version: Python version for the environment
     """
     # 1. Create conda environment if not exists
-    result = subprocess.run(f"conda env list | grep -q '^{env_name} '", shell=True)
+    env_list = subprocess.run(["conda", "env", "list"], capture_output=True, text=True, check=False)
-    if result.returncode != 0:
+    env_exists = any(line.split()[0] == env_name for line in env_list.stdout.splitlines() if line and not line.startswith("#"))
+    if not env_exists:
         print(f"[yellow]Creating conda env '{env_name}' (Python {python_version})...[/yellow]")
-        subprocess.check_call(f"conda create -y -n {env_name} python={python_version}", shell=True)
+        subprocess.check_call(["conda", "create", "-y", "-n", env_name, f"python={python_version}"])
-        subprocess.check_call(f"conda run -n {env_name} pip install --upgrade pip", shell=True)
+        subprocess.check_call(["conda", "run", "-n", env_name, "pip", "install", "--upgrade", "pip"])
 
     print(f"[yellow]Installing dependencies from {requirements_file.name}...[/yellow]")
-    subprocess.check_call(f"conda run -n {env_name} pip install -r {requirements_file}", shell=True)
+    subprocess.check_call(["conda", "run", "-n", env_name, "pip", "install", "-r", str(requirements_file)])
     print(f"[green]Conda env '{env_name}' ready[/green]")
 
     _CONDA_ENV_PREPARED.add(env_name)
@@ -971,8 +973,8 @@ class FTCondaEnv(LocalEnv[FTCondaConf]):
             # Note: flash-attn>=2.8 is required for B200 (sm_100) support
             print("[yellow]Installing flash-attn (compiling, may take a few minutes)...[/yellow]")
             subprocess.check_call(
-                f"conda run -n {self.conf.conda_env_name} pip install 'flash-attn>=2.8' --no-build-isolation --no-cache-dir",
+                ["conda", "run", "-n", self.conf.conda_env_name, "pip", "install",
-                shell=True,
+                 "flash-attn>=2.8", "--no-build-isolation", "--no-cache-dir"],
             )
 
             # Re-update bin_path after prepare() in case the conda env was just created
@@ -1442,7 +1444,7 @@ class DockerEnv(Env[DockerConf]):
         if self.conf.extra_volumes is not None:
             for lp, rp in self.conf.extra_volumes.items():
                 volumes[lp] = rp if isinstance(rp, dict) else {"bind": rp, "mode": self.conf.extra_volume_mode}
-            cache_path = "/tmp/sample" if "/sample/" in "".join(self.conf.extra_volumes.keys()) else "/tmp/full"
+            cache_path = "/tmp/sample" if "/sample/" in "".join(self.conf.extra_volumes.keys()) else "/tmp/full"  # nosec B108 — fixed Docker volume mount point, not a user-writable temp file
             Path(cache_path).mkdir(parents=True, exist_ok=True)
             volumes[cache_path] = {
                 "bind": T("scenarios.data_science.share:scen.cache_path").r(),
@@ -1471,7 +1473,8 @@ class DockerEnv(Env[DockerConf]):
                 cpu_count=self.conf.cpu_count,  # Set CPU limit
                 **self._gpu_kwargs(client),
             )
-            assert container is not None  # Ensure container was created successfully
+            if container is None:
+                raise AssertionError("Docker container was not created successfully")
             logs = container.logs(stream=True)
             print(Rule("[bold green]Docker Logs Begin[/bold green]", style="dark_orange"))
             table = Table(title="Run Info", show_header=False)

rdagent/utils/workflow/misc.py

@@ -30,7 +30,8 @@ def wait_retry(
     >>> counter
     2
     """
-    assert retry_n > 0, "retry_n should be greater than 0"
+    if retry_n <= 0:
+        raise ValueError("retry_n should be greater than 0")
 
     def decorator(f: Callable[..., ASpecificRet]) -> Callable[..., ASpecificRet]:
         def wrapper(*args: Any, **kwargs: Any) -> ASpecificRet:

rdagent/utils/workflow/tracking.py

@@ -84,7 +84,8 @@ class WorkflowTracker:
             # Log timer status if timer is started
             if self.loop_base.timer.started:
                 remain_time = self.loop_base.timer.remain_time()
-                assert remain_time is not None
+                if remain_time is None:
+                    raise AssertionError("remain_time should not be None")
                 mlflow.log_metric("remain_time", remain_time.total_seconds())
                 mlflow.log_metric(
                     "remain_percent",

requirements.txt

@@ -9,8 +9,8 @@ psutil
 fire
 fuzzywuzzy
 openai
-litellm>=1.73  # to support `from litellm import get_valid_models`
+litellm>=1.83.14  # to support `from litellm import get_valid_models`
-aiohttp>=3.13.4  # CVE-2026-22815, CVE-2026-34515, CVE-2026-34516, CVE-2026-34525
+aiohttp>=3.13.4  # CVE-2026-22815, CVE-2026-34515, CVE-2026-34516, CVE-2026-34525; >=3.13.4 due to litellm==1.83.14 exact pin
 azure.identity
 pyarrow
 rich
@@ -98,8 +98,8 @@ optuna>=3.5.0
 beautifulsoup4>=4.12.0
 
 # ML Training Pipeline
-lightgbm>=3.3.0
+lightgbm>=3.3.5
-scipy>=1.9.0
+scipy>=1.15.3
 
 # RL Trading (optional - system works without these)
 # Install for full RL training: pip install stable-baselines3[extra] gymnasium

requirements/rl.txt

@@ -8,7 +8,7 @@
 # Only install if you want to use full PPO/A2C/SAC training.
 
 # Core RL library
-stable-baselines3[extra]>=2.0.0
+stable-baselines3[extra]>=2.8.0
 
 # Gymnasium environment (OpenAI Gym successor)
 gymnasium>=0.29.0