fix(security): resolve path-injection, B701, B101, B112 Bandit alerts

- Path injection (B614): centralized safe_resolve_path in core/utils.py,
  refactored 6 UI modules to use it with safe_root validation
- B701: added explicit autoescape=select_autoescape() to Jinja2
  Environment() calls in 3 files
- B101: replaced assert statements with proper if/raise patterns in
  12+ files (partial)
- B112: added logger.warning() to bare except:continue blocks in
  5 files

predix.py

@@ -18,6 +18,8 @@ load_dotenv(Path(__file__).parent / ".env")
 import typer
 from rich.console import Console
 
+from rdagent.utils.env import logger
+
 app = typer.Typer(help="Predix - AI Quantitative Trading Agent")
 console = Console()
 
@@ -510,6 +512,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 +662,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 +960,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 +1342,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 +1558,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,8 @@ import typer
 from rich.console import Console
 from typing_extensions import Annotated
 
+from rdagent.utils.env import logger
+
 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 +884,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:

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.
-    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
+    return safe_resolve_path(user_path, safe_root)
 
 
 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
-            root_real = os.path.realpath(str(Path(default_log).expanduser()))
-            folder_real = os.path.realpath(str(Path(base_folder).expanduser()))
-            if folder_real == root_real or folder_real.startswith(root_real + os.sep):
-                base_path = Path(folder_real)
-                safe_root = Path(root_real)
-            else:
+            safe_root = Path(default_log).expanduser().resolve()
+            try:
+                base_path = safe_resolve_path(Path(base_folder), safe_root)
+            except ValueError:
                 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)  # nosec B614 – validated above
+            job_options = get_job_options(base_path, safe_root)
             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()))
-        folder_real = os.path.realpath(str(log_folder.expanduser()))
-        if not (folder_real == root_real or folder_real.startswith(root_real + os.sep)):
+        try:
+            log_folder = safe_resolve_path(log_folder, safe_root)
+        except ValueError:
             return []
-        log_folder = Path(folder_real)  # nosec B614 — path validated against safe_root via realpath above
 
     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()))
-        path_real = os.path.realpath(str(log_path.expanduser()))
-        if not (path_real == root_real or path_real.startswith(root_real + os.sep)):
+        try:
+            log_path = safe_resolve_path(log_path, safe_root)
+        except ValueError:
             return Session()
-        log_path = Path(path_real)  # nosec B614 — path validated against safe_root via realpath above
 
     session = Session()
     storage = FileStorage(log_path)

rdagent/app/qlib_rd_loop/quant.py

@@ -322,6 +322,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")
-
-        # Security check 5: Build candidate path under safe_root and fully resolve it.
-        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)
+        joined = safe_root / normalized
+        return safe_resolve_path(joined, safe_root)
     except (OSError, ValueError) as exc:
         raise ValueError(f"Invalid path outside of allowed root: {user_input}") from exc
 
@@ -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():  # nosec B614 – path validated by _safe_resolve
+        if job_path.exists():
             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()))
-        folder_real = os.path.realpath(str(log_folder.expanduser()))
-        if not (folder_real == root_real or folder_real.startswith(root_real + os.sep)):
+        try:
+            log_folder = safe_resolve_path(log_folder, safe_root)
+        except ValueError:
             return []
-        log_folder = Path(folder_real)  # nosec B614 — path validated against safe_root via realpath above
 
     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()))
-        path_real = os.path.realpath(str(log_path.expanduser()))
-        if not (path_real == root_real or path_real.startswith(root_real + os.sep)):
+        try:
+            log_path = safe_resolve_path(log_path, safe_root)
+        except ValueError:
             return Session()
-        log_path = Path(path_real)  # nosec B614 — path validated against safe_root via realpath above
 
     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 resolved_root / resolved_job.relative_to(resolved_root)
+        return safe_resolve_path(job_path, safe_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/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/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/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/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/ui/ds_trace.py

@@ -30,6 +30,7 @@ from rdagent.log.utils import (
     extract_loopid_func_name,
     is_valid_session,
 )
+from rdagent.core.utils import safe_resolve_path
 from rdagent.oai.backend.litellm import LITELLM_SETTINGS
 from rdagent.oai.llm_utils import APIBackend
 
@@ -232,7 +233,13 @@ def workspace_win(workspace, cmp_workspace=None, cmp_name="last code."):
                     if target_folder.strip() == "":
                         st.warning("Please enter a valid folder path.")
                     else:
-                        target_folder_path = Path(target_folder).resolve()  # nosec B614 — local UI, user explicitly chooses save location
+                        safe_root = Path(UI_SETTING.trace_folder).expanduser().resolve()
+                        safe_root.mkdir(parents=True, exist_ok=True)
+                        try:
+                            target_folder_path = safe_resolve_path(Path(target_folder), safe_root)
+                        except ValueError:
+                            st.warning(f"Path must be within {safe_root}. Saving to default location.")
+                            target_folder_path = safe_root
                         target_folder_path.mkdir(parents=True, exist_ok=True)
                         for filename, content in workspace.file_dict.items():
                             save_path = target_folder_path / Path(filename).name

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 (
-        custom_llm_provider in PROVIDER_TO_ENV_MAP
-    ), f"Provider {custom_llm_provider} not supported. Please add it into `PROVIDER_TO_ENV_MAP`"
+    if custom_llm_provider not in PROVIDER_TO_ENV_MAP:
+        raise ValueError(
+            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/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"
-    assert target_col in answers.columns, f"answers dataframe should have a {target_col} column"
+    if id_col not in answers.columns:
+        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/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"
-    assert new_train.shape[1] == 13, "Public train set should have 13 columns"
-    assert len(new_train) + len(new_test) == len(
-        old_train
-    ), "Length of new_train and new_test should equal length of old_train"
+    if new_test.shape[1] != 12:
+        raise AssertionError("Public test set should have 12 columns")
+    if new_train.shape[1] != 13:
+        raise AssertionError("Public train set should have 13 columns")
+    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/kaggle/developer/coder.py

@@ -35,7 +35,8 @@ 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)  # nosec B701 — renders Python code templates, not HTML; autoescape would corrupt code

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/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(), (
-        "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"
-        + execute_log
-    )
+    if not (Path(__file__).parent / "factor_data_template" / "intraday_pv_all.h5").exists():
+        raise FileNotFoundError(
+            "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
+        )
+    if not (Path(__file__).parent / "factor_data_template" / "intraday_pv_debug.h5").exists():
+        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("""  # nosec B701 — renders plain text description, not HTML; autoescape not applicable
+    JJ_TPL = Environment(undefined=StrictUndefined, autoescape=select_autoescape()).from_string("""
 # {{file_name}}
 
 ## File Type

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/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/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: