feat: add a web UI server (#1345)

* update rdagent cmd

* fix log error message

* use multiProcessing.Process instead of subprocess.Popen

* add traces to gitignore

* add user interactor in RDLoop (finance scenarios)

* add interactor (feedback, hypothesis) for quant scens

* fix the test_end in qlib conf

* add features init config, general instruction to qlib scenarios

* set base features for based exp

* fix bug when combine factors

* move traces folder to git_ignore_folder

* fix bug in features init

* fix quant interact bug

* fix logger warning error

* bug fixes

* modify rdagent logger, now it can set file output

* adjust cli functions and fix logger bug

* fix server port transport problem

* update server_ui in cli

* add web code

* fix CI problem

* black fix

* update web ui README

* update README

* update readme

rdagent/app/cli.py

@@ -16,10 +16,13 @@ load_dotenv(".env")
 
 import subprocess
 from importlib.resources import path as rpath
+from typing import Optional
 
 import typer
+from typing_extensions import Annotated
 
 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 (
     extract_models_and_implement as general_model,
 )
@@ -33,6 +36,11 @@ from rdagent.log.mle_summary import grade_summary as grade_summary
 
 app = typer.Typer()
 
+CheckoutOption = Annotated[bool, typer.Option("--checkout/--no-checkout", "-c/-C")]
+CheckEnvOption = Annotated[bool, typer.Option("--check-env/--no-check-env", "-e/-E")]
+CheckDockerOption = Annotated[bool, typer.Option("--check-docker/--no-check-docker", "-d/-D")]
+CheckPortsOption = Annotated[bool, typer.Option("--check-ports/--no-check-ports", "-p/-P")]
+
 
 def ui(port=19899, log_dir="", debug: bool = False, data_science: bool = False):
     """
@@ -56,9 +64,11 @@ def ui(port=19899, log_dir="", debug: bool = False, data_science: bool = False):
 
 def server_ui(port=19899):
     """
-    start web app to show the log traces in real time
+    start the Flask log server in real time
     """
-    subprocess.run(["python", "rdagent/log/server/app.py", f"--port={port}"])
+    from rdagent.log.server.app import main as log_server_main
+
+    log_server_main(port=port)
 
 
 def ds_user_interact(port=19900):
@@ -69,17 +79,123 @@ def ds_user_interact(port=19900):
     subprocess.run(commands)
 
 
-app.command(name="fin_factor")(fin_factor)
-app.command(name="fin_model")(fin_model)
-app.command(name="fin_quant")(fin_quant)
-app.command(name="fin_factor_report")(fin_factor_report)
-app.command(name="general_model")(general_model)
-app.command(name="data_science")(data_science)
-app.command(name="grade_summary")(grade_summary)
+@app.command(name="fin_factor")
+def fin_factor_cli(
+    path: Optional[str] = None,
+    step_n: Optional[int] = None,
+    loop_n: Optional[int] = None,
+    all_duration: Optional[str] = None,
+    checkout: CheckoutOption = True,
+):
+    fin_factor(path=path, step_n=step_n, loop_n=loop_n, all_duration=all_duration, checkout=checkout)
+
+
+@app.command(name="fin_model")
+def fin_model_cli(
+    path: Optional[str] = None,
+    step_n: Optional[int] = None,
+    loop_n: Optional[int] = None,
+    all_duration: Optional[str] = None,
+    checkout: CheckoutOption = True,
+):
+    fin_model(path=path, step_n=step_n, loop_n=loop_n, all_duration=all_duration, checkout=checkout)
+
+
+@app.command(name="fin_quant")
+def fin_quant_cli(
+    path: Optional[str] = None,
+    step_n: Optional[int] = None,
+    loop_n: Optional[int] = None,
+    all_duration: Optional[str] = None,
+    checkout: CheckoutOption = True,
+):
+    fin_quant(path=path, step_n=step_n, loop_n=loop_n, all_duration=all_duration, checkout=checkout)
+
+
+@app.command(name="fin_factor_report")
+def fin_factor_report_cli(
+    report_folder: Optional[str] = None,
+    path: Optional[str] = None,
+    all_duration: Optional[str] = None,
+    checkout: CheckoutOption = True,
+):
+    fin_factor_report(report_folder=report_folder, path=path, all_duration=all_duration, checkout=checkout)
+
+
+@app.command(name="general_model")
+def general_model_cli(report_file_path: str):
+    general_model(report_file_path)
+
+
+@app.command(name="data_science")
+def data_science_cli(
+    path: Optional[str] = None,
+    checkout: CheckoutOption = True,
+    step_n: Optional[int] = None,
+    loop_n: Optional[int] = None,
+    timeout: Optional[str] = None,
+    competition: Optional[str] = None,
+):
+    data_science(
+        path=path,
+        checkout=checkout,
+        step_n=step_n,
+        loop_n=loop_n,
+        timeout=timeout,
+        competition=competition,
+    )
+
+
+@app.command(name="llm_finetune")
+def llm_finetune_cli(
+    path: Optional[str] = None,
+    checkout: CheckoutOption = True,
+    benchmark: Optional[str] = None,
+    benchmark_description: Optional[str] = None,
+    dataset: Optional[str] = None,
+    base_model: Optional[str] = None,
+    upper_data_size_limit: Optional[int] = None,
+    step_n: Optional[int] = None,
+    loop_n: Optional[int] = None,
+    timeout: Optional[str] = None,
+):
+    llm_finetune(
+        path=path,
+        checkout=checkout,
+        benchmark=benchmark,
+        benchmark_description=benchmark_description,
+        dataset=dataset,
+        base_model=base_model,
+        upper_data_size_limit=upper_data_size_limit,
+        step_n=step_n,
+        loop_n=loop_n,
+        timeout=timeout,
+    )
+
+
+@app.command(name="grade_summary")
+def grade_summary_cli(log_folder: str):
+    grade_summary(log_folder)
+
+
 app.command(name="ui")(ui)
 app.command(name="server_ui")(server_ui)
-app.command(name="health_check")(health_check)
-app.command(name="collect_info")(collect_info)
+
+
+@app.command(name="health_check")
+def health_check_cli(
+    check_env: CheckEnvOption = True,
+    check_docker: CheckDockerOption = True,
+    check_ports: CheckPortsOption = True,
+):
+    health_check(check_env=check_env, check_docker=check_docker, check_ports=check_ports)
+
+
+@app.command(name="collect_info")
+def collect_info_cli():
+    collect_info()
+
+
 app.command(name="ds_user_interact")(ds_user_interact)
 
 

rdagent/app/data_science/loop.py

@@ -3,8 +3,6 @@ from pathlib import Path
 from typing import Optional
 
 import fire
-import typer
-from typing_extensions import Annotated
 
 from rdagent.app.data_science.conf import DS_RD_SETTING
 from rdagent.core.utils import import_class
@@ -14,7 +12,7 @@ from rdagent.scenarios.data_science.loop import DataScienceRDLoop
 
 def main(
     path: Optional[str] = None,
-    checkout: Annotated[bool, typer.Option("--checkout/--no-checkout", "-c/-C")] = True,
+    checkout: bool = True,
     checkout_path: Optional[str] = None,
     step_n: Optional[int] = None,
     loop_n: Optional[int] = None,

rdagent/app/finetune/llm/loop.py

@@ -7,8 +7,7 @@ Standard RDLoop entry point for LLM fine-tuning, consistent with data science im
 import asyncio
 from typing import Optional, cast
 
-import typer
-from typing_extensions import Annotated
+import fire
 
 from rdagent.app.finetune.llm.conf import FT_RD_SETTING
 from rdagent.log import rdagent_logger as logger
@@ -17,7 +16,7 @@ from rdagent.scenarios.finetune.loop import LLMFinetuneRDLoop
 
 def main(
     path: Optional[str] = None,
-    checkout: Annotated[bool, typer.Option("--checkout/--no-checkout", "-c/-C")] = True,
+    checkout: bool = True,
     user_target_scenario: Optional[str] = None,
     benchmark: Optional[str] = None,
     benchmark_description: Optional[str] = None,
@@ -98,4 +97,4 @@ def main(
 
 
 if __name__ == "__main__":
-    typer.run(main)
+    fire.Fire(main)

rdagent/app/qlib_rd_loop/conf.py

@@ -45,7 +45,7 @@ class ModelBasePropSetting(BasePropSetting):
     test_start: str = "2017-01-01"
     """Start date of the test / backtest segment"""
 
-    test_end: Optional[str] = None
+    test_end: Optional[str] = "2020-08-01"
     """End date of the test / backtest segment"""
 
 
@@ -89,7 +89,7 @@ class FactorBasePropSetting(BasePropSetting):
     test_start: str = "2017-01-01"
     """Start date of the test / backtest segment"""
 
-    test_end: Optional[str] = None
+    test_end: Optional[str] = "2020-08-01"
     """End date of the test / backtest segment"""
 
 
@@ -102,10 +102,10 @@ class FactorFromReportPropSetting(FactorBasePropSetting):
     report_result_json_file_path: str = "git_ignore_folder/report_list.json"
     """Path to the JSON file listing research reports for factor extraction"""
 
-    max_factors_per_exp: int = 10000
+    max_factors_per_exp: int = 6
     """Maximum number of factors implemented per experiment"""
 
-    report_limit: int = 10000
+    report_limit: int = 20
     """Maximum number of reports to process"""
 
 
@@ -166,7 +166,7 @@ class QuantBasePropSetting(BasePropSetting):
     test_start: str = "2017-01-01"
     """Start date of the test / backtest segment"""
 
-    test_end: Optional[str] = None
+    test_end: Optional[str] = "2020-08-01"
     """End date of the test / backtest segment"""
 
 

rdagent/app/qlib_rd_loop/factor.py

@@ -7,8 +7,6 @@ from pathlib import Path
 from typing import Any, Optional
 
 import fire
-import typer
-from typing_extensions import Annotated
 
 from rdagent.app.qlib_rd_loop.conf import FACTOR_PROP_SETTING
 from rdagent.components.workflow.rd_loop import RDLoop
@@ -34,8 +32,10 @@ def main(
     step_n: Optional[int] = None,
     loop_n: Optional[int] = None,
     all_duration: str | None = None,
-    checkout: Annotated[bool, typer.Option("--checkout/--no-checkout", "-c/-C")] = True,
+    checkout: bool = True,
     checkout_path: Optional[str] = None,
+    base_features_path: Optional[str] = None,
+    **kwargs,
 ):
     """
     Auto R&D Evolving loop for fintech factors.
@@ -51,10 +51,15 @@ def main(
         checkout = Path(checkout_path)
 
     if path is None:
-        model_loop = FactorRDLoop(FACTOR_PROP_SETTING)
+        factor_loop = FactorRDLoop(FACTOR_PROP_SETTING)
     else:
-        model_loop = FactorRDLoop.load(path, checkout=checkout)
-    asyncio.run(model_loop.run(step_n=step_n, loop_n=loop_n, all_duration=all_duration))
+        factor_loop = FactorRDLoop.load(path, checkout=checkout)
+
+    factor_loop._init_base_features(base_features_path)
+    if "user_interaction_queues" in kwargs and kwargs["user_interaction_queues"] is not None:
+        factor_loop._set_interactor(*kwargs["user_interaction_queues"])
+        factor_loop._interact_init_params()
+    asyncio.run(factor_loop.run(step_n=step_n, loop_n=loop_n, all_duration=all_duration))
 
 
 if __name__ == "__main__":

rdagent/app/qlib_rd_loop/factor_from_report.py

@@ -126,6 +126,9 @@ class FactorReportLoop(FactorRDLoop, metaclass=LoopMeta):
                 ]
                 exp.sub_workspace_list = exp.sub_workspace_list[: FACTOR_FROM_REPORT_PROP_SETTING.max_factors_per_exp]
                 exp.sub_tasks = exp.sub_tasks[: FACTOR_FROM_REPORT_PROP_SETTING.max_factors_per_exp]
+                exp.base_features = self.plan["features"]
+                if exp.based_experiments:
+                    exp.based_experiments[-1].base_features = self.plan["features"]
                 logger.log_object(exp.hypothesis, tag="hypothesis generation")
                 logger.log_object(exp.sub_tasks, tag="experiment generation")
                 return exp

rdagent/app/qlib_rd_loop/model.py

@@ -21,6 +21,8 @@ def main(
     loop_n: int | None = None,
     all_duration: str | None = None,
     checkout: bool = True,
+    base_features_path: str | None = None,
+    **kwargs,
 ):
     """
     Auto R&D Evolving loop for fintech models
@@ -36,6 +38,10 @@ def main(
         model_loop = ModelRDLoop(MODEL_PROP_SETTING)
     else:
         model_loop = ModelRDLoop.load(path, checkout=checkout)
+    model_loop._init_base_features(base_features_path)
+    if "user_interaction_queues" in kwargs and kwargs["user_interaction_queues"] is not None:
+        model_loop._set_interactor(*kwargs["user_interaction_queues"])
+        model_loop._interact_init_params()
     asyncio.run(model_loop.run(step_n=step_n, loop_n=loop_n, all_duration=all_duration))
 
 

rdagent/app/qlib_rd_loop/quant.py

@@ -15,6 +15,7 @@ from rdagent.core.developer import Developer
 from rdagent.core.exception import FactorEmptyError, ModelEmptyError
 from rdagent.core.proposal import (
     Experiment2Feedback,
+    ExperimentPlan,
     Hypothesis2Experiment,
     HypothesisFeedback,
     HypothesisGen,
@@ -23,6 +24,7 @@ from rdagent.core.scenario import Scenario
 from rdagent.core.utils import import_class
 from rdagent.log import rdagent_logger as logger
 from rdagent.scenarios.qlib.proposal.quant_proposal import QuantTrace
+from rdagent.utils.qlib import ALPHA20
 
 
 class QuantRDLoop(RDLoop):
@@ -62,6 +64,10 @@ class QuantRDLoop(RDLoop):
         self.model_summarizer: Experiment2Feedback = import_class(PROP_SETTING.model_summarizer)(scen)
         logger.log_object(self.model_summarizer, tag="model summarizer")
 
+        self.plan: ExperimentPlan = {
+            "features": ALPHA20,
+            "feature_codes": {},
+        }  # for user interaction
         self.trace = QuantTrace(scen=scen)
         super(RDLoop, self).__init__()
 
@@ -75,6 +81,11 @@ class QuantRDLoop(RDLoop):
                 else:
                     exp = self.model_hypothesis2experiment.convert(hypo, self.trace)
                 logger.log_object(exp.sub_tasks, tag="experiment generation")
+                exp.base_features = self.plan["features"]
+                exp.base_feature_codes = self.plan["feature_codes"]
+                if exp.based_experiments:
+                    exp.based_experiments[-1].base_features = self.plan["features"]
+                    exp.based_experiments[-1].base_feature_codes = self.plan["feature_codes"]
                 return {"propose": hypo, "exp_gen": exp}
             await asyncio.sleep(1)
 
@@ -112,6 +123,7 @@ class QuantRDLoop(RDLoop):
                 feedback = self.factor_summarizer.generate_feedback(prev_out["running"], self.trace)
             elif prev_out["direct_exp_gen"]["propose"].action == "model":
                 feedback = self.model_summarizer.generate_feedback(prev_out["running"], self.trace)
+        feedback = self._interact_feedback(feedback)
         logger.log_object(feedback, tag="feedback")
         return feedback
 
@@ -122,6 +134,8 @@ def main(
     loop_n: int | None = None,
     all_duration: str | None = None,
     checkout: bool = True,
+    base_features_path: str | None = None,
+    **kwargs,
 ):
     """
     Auto R&D Evolving loop for fintech factors.
@@ -133,6 +147,10 @@ def main(
         quant_loop = QuantRDLoop(QUANT_PROP_SETTING)
     else:
         quant_loop = QuantRDLoop.load(path, checkout=checkout)
+    quant_loop._init_base_features(base_features_path)
+    if "user_interaction_queues" in kwargs and kwargs["user_interaction_queues"] is not None:
+        quant_loop._set_interactor(*kwargs["user_interaction_queues"])
+        quant_loop._interact_init_params()
 
     asyncio.run(quant_loop.run(step_n=step_n, loop_n=loop_n, all_duration=all_duration))
 

rdagent/app/utils/health_check.py

@@ -4,10 +4,8 @@ import socket
 import docker
 import fire
 import litellm
-import typer
 from litellm import completion, embedding
 from litellm.utils import ModelResponse
-from typing_extensions import Annotated
 
 from rdagent.log import rdagent_logger as logger
 from rdagent.utils.env import cleanup_container
@@ -44,10 +42,10 @@ def check_and_list_free_ports(start_port=19899, max_ports=10) -> None:
             if not is_port_in_use(port):
                 free_ports.append(port)
         logger.warning(
-            f"Port 19899 is occupied, please replace it with an available port when running the `rdagent ui` command. Available ports: {free_ports}"
+            f"Port 19899 is occupied, please replace it with an available port when running the `rdagent ui/server_ui` command. Available ports: {free_ports}"
         )
     else:
-        logger.info(f"Port 19899 is not occupied, you can run the `rdagent ui` command")
+        logger.info(f"Port 19899 is not occupied, you can run the `rdagent ui/server_ui` command")
 
 
 def test_chat(chat_model, chat_api_key, chat_api_base):
@@ -135,9 +133,9 @@ def env_check():
 
 
 def health_check(
-    check_env: Annotated[bool, typer.Option("--check-env/--no-check-env", "-e/-E")] = True,
-    check_docker: Annotated[bool, typer.Option("--check-docker/--no-check-docker", "-d/-D")] = True,
-    check_ports: Annotated[bool, typer.Option("--check-ports/--no-check-ports", "-p/-P")] = True,
+    check_env: bool = True,
+    check_docker: bool = True,
+    check_ports: bool = True,
 ):
     """
     Run the RD-Agent health check:
@@ -167,4 +165,4 @@ def health_check(
 
 
 if __name__ == "__main__":
-    typer.run(health_check)
+    fire.Fire(health_check)

rdagent/components/proposal/__init__.py

@@ -42,6 +42,7 @@ class LLMHypothesisGen(HypothesisGen):
             ),
             hypothesis_output_format=context_dict["hypothesis_output_format"],
             hypothesis_specification=context_dict["hypothesis_specification"],
+            user_instruction=plan.get("user_instruction", None) if plan is not None else None,
         )
         user_prompt = T(".prompts:hypothesis_gen.user_prompt").r(
             targets=self.targets,

rdagent/components/proposal/prompts.yaml

@@ -3,10 +3,17 @@ hypothesis_gen:
     The user is working on generating new hypotheses for the {{ targets }} in a data-driven research and development process. 
     The {{ targets }} are used in the following scenario:
     {{ scenario }}
+
+    {% if user_instruction %}
+    **User's overall instruction:**
+    {{ user_instruction }}
+    {% endif %}
+
     The user has already proposed several hypotheses and conducted evaluations on them. This information will be provided to you. Your task is to analyze previous experiments, reflect on the decision made in each experiment, and consider why experiments with a decision of true were successful while those with a decision of false failed. Then, think about how to improve further — either by refining the existing approach or by exploring an entirely new direction.
     If one exists and you agree with it, feel free to use it. If you disagree, please generate an improved version.
     {% if hypothesis_specification %}
-    To assist you in formulating new hypotheses, the user has provided some additional information: {{ hypothesis_specification }}.
+    To assist you in formulating new hypotheses, the user has provided some additional information:
+    {{ hypothesis_specification }}
     **Important:** If the hypothesis_specification outlines the next steps you need to follow, ensure you adhere to those instructions.
     {% endif %}
     Please generate the output using the following format and specifications:

rdagent/components/workflow/rd_loop.py

@@ -4,6 +4,9 @@ It is from `rdagent/app/qlib_rd_loop/model.py` and try to replace `rdagent/app/q
 """
 
 import asyncio
+import json
+from multiprocessing import Queue
+from pathlib import Path
 from typing import Any
 
 from rdagent.components.workflow.conf import BasePropSetting
@@ -11,6 +14,7 @@ from rdagent.core.conf import RD_AGENT_SETTINGS
 from rdagent.core.developer import Developer
 from rdagent.core.proposal import (
     Experiment2Feedback,
+    ExperimentPlan,
     Hypothesis,
     Hypothesis2Experiment,
     HypothesisFeedback,
@@ -20,6 +24,7 @@ from rdagent.core.proposal import (
 from rdagent.core.scenario import Scenario
 from rdagent.core.utils import import_class
 from rdagent.log import rdagent_logger as logger
+from rdagent.utils.qlib import ALPHA20, validate_qlib_features
 from rdagent.utils.workflow import LoopBase, LoopMeta
 
 
@@ -36,6 +41,11 @@ class RDLoop(LoopBase, metaclass=LoopMeta):
             else None
         )
 
+        self.plan: ExperimentPlan = {
+            "features": ALPHA20,
+            "feature_codes": {},
+        }  # for user interaction
+
         self.hypothesis2experiment: Hypothesis2Experiment = (
             import_class(PROP_SETTING.hypothesis2experiment)()
             if hasattr(PROP_SETTING, "hypothesis2experiment") and PROP_SETTING.hypothesis2experiment
@@ -58,8 +68,125 @@ class RDLoop(LoopBase, metaclass=LoopMeta):
         super().__init__()
 
     # excluded steps
+    def _set_interactor(self, user_request_q: Queue, user_response_q: Queue):
+        self.user_request_q = user_request_q
+        self.user_response_q = user_response_q
+
+    def _init_base_features(self, base_features_path: str | None):
+        if base_features_path is not None:
+            try:
+                base_dir = Path(base_features_path)
+                base_factors_file = base_dir / "base_factors.json"
+
+                feature_codes: dict[str, str] = {}
+                for py_file in sorted(base_dir.glob("*.py")):
+                    feature_codes[py_file.name] = py_file.read_text()
+                self.plan["feature_codes"] = feature_codes
+
+                if not base_factors_file.exists():
+                    logger.info(f"No base_factors.json found under {base_dir}. Keeping default base features.")
+                    logger.info(f"{len(feature_codes)} feature code files loaded from {base_dir}.")
+                else:
+                    with base_factors_file.open("r") as f:
+                        features = json.load(f)
+
+                    if not isinstance(features, dict):
+                        raise ValueError(
+                            "`base_factors.json` must contain a JSON object of feature_name -> expression."
+                        )
+
+                    if validate_qlib_features(list(features.values())):
+                        self.plan["features"] = features
+                        logger.info(
+                            f"Loaded base features from {base_factors_file}. {len(features)} features loaded and {len(feature_codes)} feature code files loaded."
+                        )
+                    else:
+                        logger.warning(
+                            f"Base feature validation failed for features loaded from {base_factors_file}. Using default features."
+                        )
+            except Exception as e:
+                logger.warning(f"Failed to load base features from {base_features_path}: {e}. Using default features.")
+        else:
+            logger.info("No base features path provided. Using default features.")
+
+    def _interact_init_params(self) -> None:
+        if not (hasattr(self, "user_request_q") and hasattr(self, "user_response_q")):
+            return
+
+        logger.info("Waiting for user interaction on initial parameters...")
+        try:
+            self.user_request_q.put(
+                {
+                    "user_instruction": None,
+                }
+            )
+            res_dict = self.user_response_q.get()
+            logger.info("Received user instruction response.")
+            self.plan.update(res_dict)
+
+            if "feature_codes" not in self.plan:
+                self.plan[
+                    "user_instruction"
+                ] += f"\n\n{str(list(self.plan['feature_codes'].keys()))} has been configured as the base factor; do not generate duplicate factors."
+            fea_valid_msg = ""
+            while True:
+                logger.info("Requesting base feature configuration from user.")
+                self.user_request_q.put(
+                    {
+                        "features": self.plan["features"],
+                        "feature_validation_msg": fea_valid_msg,
+                    }
+                )
+                self.plan["features"] = self.user_response_q.get()
+                logger.info("Received base feature configuration response.")
+                if validate_qlib_features(list(self.plan["features"].values())):
+                    logger.info(f"Base feature validation passed. {len(self.plan['features'])} features selected.")
+                    break
+                else:
+                    logger.info("Base feature validation failed. Asking user to revise.")
+                    fea_valid_msg = "Some features are invalid, please revise."
+
+        except (EOFError, OSError):
+            logger.info("User interaction failed, using default initial parameters.")
+            return
+        logger.info("Received user interaction on initial parameters.")
+
+    def _interact_hypo(self, hypo: Hypothesis) -> Hypothesis:
+        if not (hasattr(self, "user_request_q") and hasattr(self, "user_response_q")):
+            return hypo
+
+        logger.info("Waiting for user interaction on hypothesis...")
+        try:
+            self.user_request_q.put(hypo.__dict__)
+            res_dict = self.user_response_q.get()
+            modified_hypo = type(hypo)(**res_dict)
+        except (EOFError, OSError, TypeError):
+            logger.info("User interaction failed, using original hypothesis.")
+            return hypo
+        logger.info("Received user interaction on hypothesis.")
+        return modified_hypo
+
+    def _interact_feedback(self, feedback: HypothesisFeedback) -> HypothesisFeedback:
+        if not (hasattr(self, "user_request_q") and hasattr(self, "user_response_q")):
+            return feedback
+
+        logger.info("Waiting for user interaction on feedback...")
+        try:
+            self.user_request_q.put(feedback.__dict__)
+            res_dict = self.user_response_q.get()
+            modified_feedback = HypothesisFeedback(**res_dict)
+        except (EOFError, OSError, TypeError):
+            logger.info("User interaction failed, using original feedback.")
+            return feedback
+        logger.info("Received user interaction on feedback.")
+        return modified_feedback
+
     def _propose(self):
-        hypothesis = self.hypothesis_gen.gen(self.trace)
+        hypothesis = self.hypothesis_gen.gen(self.trace, self.plan)
+
+        # user can change the hypothesis here
+        hypothesis = self._interact_hypo(hypothesis)
+
         logger.log_object(hypothesis, tag="hypothesis generation")
         return hypothesis
 
@@ -74,6 +201,11 @@ class RDLoop(LoopBase, metaclass=LoopMeta):
             if self.get_unfinished_loop_cnt(self.loop_idx) < RD_AGENT_SETTINGS.get_max_parallel():
                 hypo = self._propose()
                 exp = self._exp_gen(hypo)
+                exp.base_features = self.plan["features"]
+                exp.base_feature_codes = self.plan["feature_codes"]
+                if exp.based_experiments:
+                    exp.based_experiments[-1].base_features = self.plan["features"]
+                    exp.based_experiments[-1].base_feature_codes = self.plan["feature_codes"]
                 return {"propose": hypo, "exp_gen": exp}
             await asyncio.sleep(1)
 
@@ -99,6 +231,7 @@ class RDLoop(LoopBase, metaclass=LoopMeta):
             )
         else:
             feedback = self.summarizer.generate_feedback(prev_out["running"], self.trace)
+        feedback = self._interact_feedback(feedback)
         logger.log_object(feedback, tag="feedback")
         return feedback
 

rdagent/core/proposal.py

@@ -105,12 +105,14 @@ class HypothesisFeedback(ExperimentFeedback):
         new_hypothesis: str | None = None,
         eda_improvement: str | None = None,
         acceptable: bool | None = None,
+        exception: Exception | None = None,
     ) -> None:
         super().__init__(
             reason,
             decision=decision,
             code_change_summary=code_change_summary,
             eda_improvement=eda_improvement,
+            exception=exception,
         )
         self.observations = observations
         self.hypothesis_evaluation = hypothesis_evaluation

rdagent/log/conf.py

@@ -19,9 +19,16 @@ class LogSettings(ExtendedBaseSettings):
 
     storages: dict[str, list[int | str]] = {}
 
+    def set_ui_server_port(self, port: int | None) -> None:
+        self.ui_server_port = port
+        if port is None:
+            self.storages.pop("rdagent.log.ui.storage.WebStorage", None)
+            return
+
+        self.storages["rdagent.log.ui.storage.WebStorage"] = [port, self.trace_path]
+
     def model_post_init(self, _context: Any, /) -> None:
-        if self.ui_server_port is not None:
-            self.storages["rdagent.log.ui.storage.WebStorage"] = [self.ui_server_port, self.trace_path]
+        self.set_ui_server_port(self.ui_server_port)
 
 
 LOG_SETTINGS = LogSettings()

rdagent/log/logger.py

@@ -7,18 +7,12 @@ from pathlib import Path
 from typing import Generator
 
 from loguru import logger
-
-from .conf import LOG_SETTINGS
-
-if LOG_SETTINGS.format_console is not None:
-    logger.remove()
-    logger.add(sys.stdout, format=LOG_SETTINGS.format_console)
-
 from psutil import Process
 
 from rdagent.core.utils import SingletonBaseClass, import_class
 
 from .base import Storage
+from .conf import LOG_SETTINGS
 from .storage import FileStorage
 from .utils import get_caller_info
 
@@ -48,6 +42,18 @@ class RDAgentLog(SingletonBaseClass):
 
     # Thread-/coroutine-local tag;  In Linux forked subprocess, it will be copied to the subprocess.
     _tag_ctx: ContextVar[str] = ContextVar("_tag_ctx", default="")
+    _raw_log_key = "_rdagent_raw"
+
+    @classmethod
+    def _configure_console_sinks(cls) -> None:
+        raw_filter = lambda record: bool(record["extra"].get(cls._raw_log_key, False))
+        normal_filter = lambda record: not raw_filter(record)
+
+        if LOG_SETTINGS.format_console is not None:
+            logger.add(sys.stdout, format=LOG_SETTINGS.format_console, filter=normal_filter)
+        else:
+            logger.add(sys.stdout, filter=normal_filter)
+        logger.add(sys.stdout, format="{message}", filter=raw_filter)
 
     @property
     def _tag(self) -> str:  # Get current tag
@@ -58,13 +64,29 @@ class RDAgentLog(SingletonBaseClass):
         self._tag_ctx.set(value)
 
     def __init__(self) -> None:
+        logger.remove()
+        self._configure_console_sinks()
+
         self.storage = FileStorage(LOG_SETTINGS.trace_path)
         self.other_storages: list[Storage] = []
+        self.refresh_storages_from_settings()
+
+        self.main_pid = os.getpid()
+
+    def refresh_storages_from_settings(self) -> None:
+        self.other_storages = []
         for storage, args in LOG_SETTINGS.storages.items():
             storage_cls = import_class(storage)
             self.other_storages.append(storage_cls(*args))
 
-        self.main_pid = os.getpid()
+    def rebind_console_to_current_streams(self) -> None:
+        """Rebind loguru sinks to the current stdio objects.
+
+        This is needed in forked/spawned subprocesses after stdout/stderr have been
+        redirected, because loguru keeps references to the original stream objects.
+        """
+        logger.remove()
+        self._configure_console_sinks()
 
     @contextmanager
     def tag(self, tag: str) -> Generator[None, None, None]:
@@ -82,6 +104,8 @@ class RDAgentLog(SingletonBaseClass):
             self._tag_ctx.reset(token)
 
     def set_storages_path(self, path: str | Path) -> None:
+        if isinstance(path, str):
+            path = Path(path)
         for storage in [self.storage] + self.other_storages:
             if hasattr(storage, "path"):
                 storage.path = path
@@ -115,17 +139,10 @@ class RDAgentLog(SingletonBaseClass):
         caller_info = get_caller_info(level=3)
         tag = f"{self._tag}.{tag}.{self.get_pids()}".strip(".")
 
-        if raw:
-            logger.remove()
-            logger.add(sys.stderr, format=lambda r: "{message}")
-
-        log_func = getattr(logger.patch(lambda r: r.update(caller_info)), level)
+        patched_logger = logger.patch(lambda r: r.update(caller_info)).bind(**{self._raw_log_key: raw}).opt(raw=raw)
+        log_func = getattr(patched_logger, level)
         log_func(msg)
 
-        if raw:
-            logger.remove()
-            logger.add(sys.stderr)
-
     def info(self, msg: str, *, tag: str = "", raw: bool = False) -> None:
         self._log("info", msg, tag=tag, raw=raw)
 

rdagent/log/server/app.py

@@ -1,69 +1,287 @@
+import logging
 import os
 import random
-import signal
-import subprocess
+import traceback
 from collections import defaultdict
+from contextlib import redirect_stderr, redirect_stdout
 from datetime import datetime, timezone
+from multiprocessing import Process, Queue
 from pathlib import Path
+from queue import Empty
 
 import randomname
 import typer
-from flask import Flask, jsonify, request, send_from_directory
+from flask import Flask, jsonify, request, send_file, send_from_directory
 from flask_cors import CORS
 from werkzeug.utils import secure_filename
 
 from rdagent.log.storage import FileStorage
 from rdagent.log.ui.conf import UI_SETTING
 from rdagent.log.ui.storage import WebStorage
-from rdagent.log.utils import is_valid_session
 
-app = Flask(__name__, static_folder=UI_SETTING.static_path)
+app = Flask(__name__, static_folder=str(Path(UI_SETTING.static_path).resolve()))
 CORS(app)
+app.config["UI_SERVER_PORT"] = 19899
 
-rdagent_processes = defaultdict()
-server_port = 19899
+_YELLOW = "\033[33m"
+_RESET = "\033[0m"
+
+
+class _YellowWarningFormatter(logging.Formatter):
+    def format(self, record: logging.LogRecord) -> str:
+        if record.levelno == logging.WARNING:
+            record.levelname = f"{_YELLOW}{record.levelname}{_RESET}"
+        return super().format(record)
+
+
+def _configure_app_logger() -> None:
+    formatter = _YellowWarningFormatter(
+        fmt="[%(asctime)s] %(levelname)s in %(module)s: %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+    for handler in app.logger.handlers:
+        handler.setFormatter(formatter)
+
+
+_configure_app_logger()
+
+
+_TARGETS_WITHOUT_USER_INTERACTION = {"general_model", "fin_factor_report"}
+
+
+class RDAgentTask:
+    def __init__(
+        self,
+        target_name: str,
+        kwargs: dict,
+        stdout_path: str,
+        log_trace_path: str,
+        scenario: str,
+        trace_name: str,
+        ui_server_port: int | None = None,
+        create_process: bool = True,
+    ) -> None:
+        self.target_name = target_name
+        self.kwargs = kwargs
+        self.stdout_path = stdout_path
+        self.log_trace_path = log_trace_path
+        self.scenario = scenario
+        self.trace_name = trace_name
+        self.ui_server_port = ui_server_port
+        self.process: Process | None = None
+
+        # Two IPC queues for user interaction.
+        # - `user_request_q`: rdagent subprocess -> server (dicts to render on frontend)
+        # - `user_response_q`: server -> rdagent subprocess (user input dicts)
+        # NOTE: Use multiprocessing.Queue because rdagent is started as a separate process.
+        self.user_request_q: Queue = Queue(maxsize=1024)
+        self.user_response_q: Queue = Queue(maxsize=1024)
+
+        if create_process:
+            self.process = Process(
+                target=self._run,
+                name=f"rdagent:{self.scenario}:{self.trace_name}",
+            )
+        self.messages: list[dict] = []
+        self.pointers: defaultdict[str, int] = defaultdict(int)
+
+    def start(self) -> None:
+        if self.process is not None:
+            self.process.start()
+
+    def is_alive(self) -> bool:
+        return self.process is not None and self.process.is_alive()
+
+    def get_end_code(self) -> int:
+        if self.process is None or self.process.exitcode is None:
+            return 0
+        return self.process.exitcode
+
+    def stop(self) -> None:
+        if self.process is not None and self.process.is_alive():
+            self.process.terminate()
+            self.process.join()
+
+        # Best-effort cleanup for IPC queues.
+        for q in (self.user_request_q, self.user_response_q):
+            try:
+                q.cancel_join_thread()
+            except Exception:
+                pass
+            try:
+                q.close()
+            except Exception:
+                pass
+
+    def _run(self) -> None:
+        from rdagent.log.conf import LOG_SETTINGS
+
+        LOG_SETTINGS.set_ui_server_port(self.ui_server_port)
+
+        from rdagent.log import rdagent_logger
+
+        rdagent_logger.refresh_storages_from_settings()
+        rdagent_logger.set_storages_path(self.log_trace_path)
+        Path(self.stdout_path).parent.mkdir(parents=True, exist_ok=True)
+        with open(self.stdout_path, "w") as log_file:
+            with redirect_stdout(log_file), redirect_stderr(log_file):
+                rdagent_logger.rebind_console_to_current_streams()
+                try:
+                    # Only interactive targets should receive IPC queues.
+                    if self.target_name not in _TARGETS_WITHOUT_USER_INTERACTION:
+                        self.kwargs.setdefault(
+                            "user_interaction_queues",
+                            (self.user_request_q, self.user_response_q),
+                        )
+
+                    if self.target_name == "data_science":
+                        from rdagent.app.data_science.loop import main as data_science
+
+                        data_science(**self.kwargs)
+                    elif self.target_name == "general_model":
+                        from rdagent.app.general_model.general_model import (
+                            extract_models_and_implement as general_model,
+                        )
+
+                        general_model(**self.kwargs)
+                    elif self.target_name == "fin_factor":
+                        from rdagent.app.qlib_rd_loop.factor import main as fin_factor
+
+                        fin_factor(**self.kwargs)
+                    elif self.target_name == "fin_factor_report":
+                        from rdagent.app.qlib_rd_loop.factor_from_report import (
+                            main as fin_factor_report,
+                        )
+
+                        fin_factor_report(**self.kwargs)
+                    elif self.target_name == "fin_model":
+                        from rdagent.app.qlib_rd_loop.model import main as fin_model
+
+                        fin_model(**self.kwargs)
+                    elif self.target_name == "fin_quant":
+                        from rdagent.app.qlib_rd_loop.quant import main as fin_quant
+
+                        fin_quant(**self.kwargs)
+                    else:
+                        raise ValueError(f"Unknown target: {self.target_name}")
+                except Exception:
+                    traceback.print_exc()
+
+
+rdagent_processes: dict[str, RDAgentTask] = {}
 log_folder_path = Path(UI_SETTING.trace_folder).absolute()
 
 
+def _drain_user_requests_into_messages(task: RDAgentTask) -> None:
+    """Move a single pending user-interaction request into `task.messages`.
+
+    Assumption: each rdagent process only has one active request at a time.
+    """
+
+    try:
+        req = task.user_request_q.get_nowait()
+    except Empty:
+        return
+    except Exception:
+        return
+
+    # Standardize the message shape for the frontend.
+    # The agent can send either a full message dict, or a raw content dict.
+    if isinstance(req, dict) and {"tag", "timestamp", "content"}.issubset(req.keys()):
+        msg = req
+    else:
+        msg = {
+            "tag": "user_interaction.request",
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+            "content": req,
+        }
+    task.messages.append(msg)
+
+
 @app.route("/favicon.ico")
 def favicon():
     return send_from_directory(app.static_folder, "favicon.ico", mimetype="image/vnd.microsoft.icon")
 
 
-msgs_for_frontend = defaultdict(list)
-pointers = defaultdict(lambda: defaultdict(int))  # pointers[trace_id][user_ip]
+def _normalize_static_request_path(fn: str) -> str:
+    static_prefix = UI_SETTING.static_path.strip("./")
+    if static_prefix and fn.startswith(f"{static_prefix}/"):
+        return fn[len(static_prefix) + 1 :]
+    return fn
+
+
+def _get_or_create_task(trace_id: str) -> RDAgentTask:
+    task = rdagent_processes.get(trace_id)
+    if task is None:
+        task = RDAgentTask(
+            target_name="",
+            kwargs={},
+            stdout_path="",
+            log_trace_path=trace_id,
+            scenario="",
+            trace_name="",
+            ui_server_port=None,
+            create_process=False,
+        )
+        rdagent_processes[trace_id] = task
+    return task
+
+
+def _resolve_stdout_path(trace_id: str) -> Path | None:
+    normalized_trace_id = str(trace_id or "").strip()
+    if not normalized_trace_id:
+        return None
+
+    task = rdagent_processes.get(str(log_folder_path / normalized_trace_id))
+    if task is None or not task.stdout_path:
+        return None
+
+    stdout_path = Path(task.stdout_path).resolve()
+
+    try:
+        if os.path.commonpath([str(stdout_path), str(log_folder_path)]) != str(log_folder_path):
+            return None
+    except ValueError:
+        return None
+
+    return stdout_path
 
 
 def read_trace(log_path: Path, id: str = "") -> None:
     fs = FileStorage(log_path)
     ws = WebStorage(port=1, path=log_path)
-    msgs_for_frontend[id] = []
+    task = _get_or_create_task(id)
+    task.messages = []
     last_timestamp = None
     for msg in fs.iter_msg():
         data = ws._obj_to_json(obj=msg.content, tag=msg.tag, id=id, timestamp=msg.timestamp.isoformat())
         if data:
             if isinstance(data, list):
                 for d in data:
-                    msgs_for_frontend[id].append(d["msg"])
+                    task.messages.append(d["msg"])
                     last_timestamp = msg.timestamp
             else:
-                msgs_for_frontend[id].append(data["msg"])
+                task.messages.append(data["msg"])
                 last_timestamp = msg.timestamp
 
     now = datetime.now(timezone.utc)
     if last_timestamp and (now - last_timestamp).total_seconds() > 1800:
-        msgs_for_frontend[id].append({"tag": "END", "timestamp": now.isoformat(), "content": {}})
+        task.messages.append(
+            {
+                "tag": "END",
+                "timestamp": now.isoformat(),
+                "content": {"error_msg": "Trace session has ended.", "end_code": 0},
+            }
+        )
 
 
 # load all traces from the log folder
-for p in log_folder_path.glob("*/*/"):
-    if is_valid_session(p):
-        read_trace(p, id=str(p))
+# for p in log_folder_path.glob("*/*/"):
+#     read_trace(p, id=str(p))
 
 
 @app.route("/trace", methods=["POST"])
 def update_trace():
-    global pointers, msgs_for_frontend
     data = request.get_json()
     trace_id = data.get("id")
     return_all = data.get("all")
@@ -75,28 +293,64 @@ def update_trace():
         return jsonify({"error": "Trace ID is required"}), 400
     trace_id = str(log_folder_path / trace_id)
 
+    task = _get_or_create_task(trace_id)
+
+    # Make sure any pending user-interaction requests are visible to the frontend.
+    _drain_user_requests_into_messages(task)
+
+    if task.process is not None and not task.is_alive():
+        if not task.messages or task.messages[-1].get("tag") != "END":
+            task.messages.append(
+                {
+                    "tag": "END",
+                    "timestamp": datetime.now(timezone.utc).isoformat(),
+                    "content": {
+                        "error_msg": "RD-Agent process has completed.",
+                        "end_code": task.get_end_code(),
+                    },
+                }
+            )
+            app.logger.warning(f"Process for {trace_id} has ended.")
+
     user_ip = request.remote_addr
 
     if reset:
-        pointers[trace_id][user_ip] = 0
+        task.pointers[user_ip] = 0
 
-    start_pointer = pointers[trace_id][user_ip]
+    start_pointer = task.pointers[user_ip]
     end_pointer = start_pointer + msg_num
-    if end_pointer > len(msgs_for_frontend[trace_id]) or return_all:
-        end_pointer = len(msgs_for_frontend[trace_id])
+    if end_pointer > len(task.messages) or return_all:
+        end_pointer = len(task.messages)
 
-    returned_msgs = msgs_for_frontend[trace_id][start_pointer:end_pointer]
-
-    pointers[trace_id][user_ip] = end_pointer
+    returned_msgs = task.messages[start_pointer:end_pointer]
+    task.pointers[user_ip] = end_pointer
     if returned_msgs:
         app.logger.info([msg["tag"] for msg in returned_msgs])
     return jsonify(returned_msgs), 200
 
 
+@app.route("/stdout", methods=["GET"])
+def download_stdout_file():
+    trace_id = request.args.get("id", "")
+    stdout_path = _resolve_stdout_path(trace_id)
+
+    if stdout_path is None:
+        return jsonify({"error": "Trace ID is required or invalid"}), 400
+    if not stdout_path.exists() or not stdout_path.is_file():
+        return jsonify({"error": "Stdout file not found"}), 404
+
+    return send_file(
+        stdout_path,
+        as_attachment=True,
+        download_name=stdout_path.name,
+        mimetype="text/plain",
+    )
+
+
 @app.route("/upload", methods=["POST"])
 def upload_file():
     # 获取请求体中的字段
-    global rdagent_processes, server_port
+    global rdagent_processes
     scenario = request.form.get("scenario")
     files = request.files.getlist("files")
     competition = request.form.get("competition")
@@ -109,21 +363,19 @@ def upload_file():
         trace_name = f"{competition}-{randomname.get_name()}"
     else:
         trace_name = randomname.get_name()
-    trace_files_path = log_folder_path / scenario / "uploads" / trace_name
+    trace_files_path = log_folder_path / "uploads" / scenario / trace_name
 
     log_trace_path = (log_folder_path / scenario / trace_name).absolute()
-    stdout_path = log_folder_path / scenario / f"{trace_name}.stdout"
+    stdout_path = log_folder_path / scenario / f"{trace_name}.log"
     if not stdout_path.exists():
         stdout_path.parent.mkdir(parents=True, exist_ok=True)
 
     # save files
     for file in files:
         if file:
-            p = (log_folder_path / scenario / "uploads" / trace_name).resolve()
+            p = (log_folder_path / "uploads" / scenario / trace_name).resolve()
             sanitized_filename = secure_filename(file.filename)  # Sanitize filename
             target_path = (p / sanitized_filename).resolve()  # Normalize target path
-            if not sanitized_filename.lower().endswith(".pdf"):
-                return jsonify({"error": "Invalid file type"}), 400
             # Ensure target_path is within the allowed base directory
             if os.path.commonpath([str(target_path), str(p)]) == str(p) and target_path.is_file() == False:
                 if not p.exists():
@@ -132,42 +384,62 @@ def upload_file():
             else:
                 return jsonify({"error": "Invalid file path"}), 400
 
+    target_name = None
+    kwargs = {}
+    loop_n_val = int(loop_n) if loop_n else None
+    all_duration_val = f"{all_duration}h" if all_duration else None
+
     if scenario == "Finance Data Building":
-        cmds = ["rdagent", "fin_factor"]
-    if scenario == "Finance Data Building (Reports)":
-        cmds = ["rdagent", "fin_factor_report", "--report_folder", str(trace_files_path)]
+        target_name = "fin_factor"
+        kwargs = {
+            "loop_n": loop_n_val,
+            "all_duration": all_duration_val,
+            "base_features_path": str(trace_files_path),
+        }
     if scenario == "Finance Model Implementation":
-        cmds = ["rdagent", "fin_model"]
+        target_name = "fin_model"
+        kwargs = {
+            "loop_n": loop_n_val,
+            "all_duration": all_duration_val,
+            "base_features_path": str(trace_files_path),
+        }
+    if scenario == "Finance Whole Pipeline":
+        target_name = "fin_quant"
+        kwargs = {
+            "loop_n": loop_n_val,
+            "all_duration": all_duration_val,
+            "base_features_path": str(trace_files_path),
+        }
+    if scenario == "Finance Data Building (Reports)":
+        target_name = "fin_factor_report"
+        kwargs = {"report_folder": str(trace_files_path), "all_duration": all_duration_val}
     if scenario == "General Model Implementation":
         if len(files) == 0:  # files is one link
             rfp = request.form.get("files")[0]
         else:  # one file is uploaded
             rfp = str(trace_files_path / files[0].filename)
-        cmds = ["rdagent", "general_model", "--report_file_path", rfp]
-    if scenario == "Finance Whole Pipeline":
-        cmds = ["rdagent", "fin_quant"]
+        target_name = "general_model"
+        kwargs = {"report_file_path": rfp}
     if scenario == "Data Science":
-        cmds = ["rdagent", "data_science", "--competition", competition]
+        target_name = "data_science"
+        kwargs = {"competition": competition, "loop_n": loop_n_val, "timeout": all_duration_val}
 
-    # time control parameters
-    if scenario != "Finance Data Building (Reports)":
-        if loop_n:
-            cmds += ["--loop_n", loop_n]
-    if all_duration:
-        cmds += ["--timeout", f"{all_duration}h"]
+    if target_name is None:
+        return jsonify({"error": "Unknown scenario"}), 400
 
-    app.logger.info(f"Started process for {log_trace_path} with parameters: {cmds}")
-    with stdout_path.open("w") as log_file:
-        rdagent_processes[str(log_trace_path)] = subprocess.Popen(
-            cmds,
-            stdout=log_file,
-            stderr=log_file,
-            env={
-                **os.environ,
-                "LOG_TRACE_PATH": str(log_trace_path),
-                "LOG_UI_SERVER_PORT": str(server_port),
-            },
-        )
+    app.logger.info(f"Started process for {log_trace_path} with target: {target_name}, kwargs: {kwargs}")
+    task = RDAgentTask(
+        target_name=target_name,
+        kwargs=kwargs,
+        stdout_path=str(stdout_path),
+        log_trace_path=str(log_trace_path),
+        scenario=scenario,
+        trace_name=trace_name,
+        ui_server_port=app.config["UI_SERVER_PORT"],
+    )
+    task.start()
+    app.logger.warning(f"Task {log_trace_path} started.")
+    rdagent_processes[str(log_trace_path)] = task
     return (
         jsonify(
             {
@@ -182,7 +454,6 @@ def upload_file():
 def receive_msgs():
     try:
         data = request.get_json()
-        # app.logger.info(data["msg"]["tag"])
         if not data:
             return jsonify({"error": "No JSON data received"}), 400
     except Exception as e:
@@ -190,16 +461,41 @@ def receive_msgs():
 
     if isinstance(data, list):
         for d in data:
-            msgs_for_frontend[d["id"]].append(d["msg"])
+            task = _get_or_create_task(d["id"])
+            task.messages.append(d["msg"])
     else:
-        msgs_for_frontend[data["id"]].append(data["msg"])
+        task = _get_or_create_task(data["id"])
+        task.messages.append(data["msg"])
+
+    return jsonify({"status": "success"}), 200
+
+
+@app.route("/user_interaction/submit", methods=["POST"])
+def submit_user_interaction_response():
+    """Frontend submits a user response; server forwards it to the rdagent subprocess via IPC queue."""
+    data = request.get_json(silent=True) or {}
+    trace_id = data.get("id")
+    payload = data.get("payload")
+
+    if not trace_id:
+        return jsonify({"error": "Trace ID is required"}), 400
+    if payload is None:
+        return jsonify({"error": "Missing 'payload'"}), 400
+
+    trace_id = str(log_folder_path / trace_id)
+    task = _get_or_create_task(trace_id)
+
+    try:
+        task.user_response_q.put(payload, block=False)
+    except Exception as e:
+        return jsonify({"error": f"Failed to enqueue user response: {e}"}), 500
 
     return jsonify({"status": "success"}), 200
 
 
 @app.route("/control", methods=["POST"])
 def control_process():
-    global rdagent_processes, msgs_for_frontend
+    global rdagent_processes
     data = request.get_json()
     app.logger.info(data)
     if not data or "id" not in data or "action" not in data:
@@ -208,32 +504,31 @@ def control_process():
     id = str(log_folder_path / data["id"])
     action = data["action"]
 
+    if action != "stop":
+        return jsonify({"error": "Only 'stop' action is supported"}), 400
+
     if id not in rdagent_processes or rdagent_processes[id] is None:
         return jsonify({"error": "No running process for given id"}), 400
 
-    process = rdagent_processes[id]
+    task = rdagent_processes[id]
 
-    if process.poll() is not None:
-        msgs_for_frontend[id].append({"tag": "END", "timestamp": datetime.now(timezone.utc).isoformat(), "content": {}})
-        return jsonify({"error": "Process has already terminated"}), 400
+    if task.process is None:
+        return jsonify({"error": "No running process for given id"}), 400
 
     try:
-        if action == "pause":
-            os.kill(process.pid, signal.SIGSTOP)
-            return jsonify({"status": "paused"}), 200
-        elif action == "resume":
-            os.kill(process.pid, signal.SIGCONT)
-            return jsonify({"status": "resumed"}), 200
-        elif action == "stop":
-            process.terminate()
-            process.wait()
-            del rdagent_processes[id]
-            msgs_for_frontend[id].append(
-                {"tag": "END", "timestamp": datetime.now(timezone.utc).isoformat(), "content": {}}
+        if task.is_alive():
+            task.stop()
+
+        if not task.messages or task.messages[-1].get("tag") != "END":
+            task.messages.append(
+                {
+                    "tag": "END",
+                    "timestamp": datetime.now(timezone.utc).isoformat(),
+                    "content": {"error_msg": "RD-Agent process was stopped by user.", "end_code": -1},
+                }
             )
-            return jsonify({"status": "stopped"}), 200
-        else:
-            return jsonify({"error": "Unknown action"}), 400
+            app.logger.warning(f"Process for {id} has been stopped.")
+        return jsonify({"status": "stopped"}), 200
     except Exception as e:
         return jsonify({"error": f"Failed to {action} process, {e}"}), 500
 
@@ -241,9 +536,8 @@ def control_process():
 @app.route("/test", methods=["GET"])
 def test():
     # return 'Hello, World!'
-    global msgs_for_frontend, pointers
-    msgs = {k: [i["tag"] for i in v] for k, v in msgs_for_frontend.items()}
-    pointers = pointers
+    msgs = {k: [i["tag"] for i in task.messages] for k, task in rdagent_processes.items()}
+    pointers = {k: dict(task.pointers) for k, task in rdagent_processes.items()}
     return jsonify({"msgs": msgs, "pointers": pointers}), 200
 
 
@@ -256,12 +550,11 @@ def index():
 
 @app.route("/<path:fn>", methods=["GET"])
 def server_static_files(fn):
-    return send_from_directory(app.static_folder, fn)
+    return send_from_directory(app.static_folder, _normalize_static_request_path(fn))
 
 
 def main(port: int = 19899):
-    global server_port
-    server_port = port
+    app.config["UI_SERVER_PORT"] = port
     app.run(debug=False, host="0.0.0.0", port=port)
 
 

rdagent/log/ui/conf.py

@@ -16,7 +16,7 @@ class UIBasePropSetting(ExtendedBaseSettings):
 
     static_path: str = "./git_ignore_folder/static"
 
-    trace_folder: str = "./traces"
+    trace_folder: str = "./git_ignore_folder/traces"
 
     enable_cache: bool = True
 

rdagent/log/ui/storage.py

@@ -33,10 +33,11 @@ class WebStorage(Storage):
     def log(self, obj: object, tag: str, timestamp: datetime | None = None, **kwargs: Any) -> str | Path:
         timestamp = gen_datetime(timestamp)
         if "pdf_image" in tag or "load_pdf_screenshot" in tag:
-            obj.save(f"{UI_SETTING.static_path}/{timestamp.isoformat()}.jpg")
+            Path(f"{UI_SETTING.static_path}/pdf_images").mkdir(parents=True, exist_ok=True)
+            obj.save(f"{UI_SETTING.static_path}/pdf_images/{timestamp.isoformat()}.jpg")
 
         try:
-            data = self._obj_to_json(obj=obj, tag=tag, id=self.path, timestamp=timestamp.isoformat())
+            data = self._obj_to_json(obj=obj, tag=tag, id=str(self.path), timestamp=timestamp.isoformat())
             if not data:
                 return "Normal log, skipped"
             if isinstance(data, list):
@@ -48,7 +49,7 @@ class WebStorage(Storage):
             resp = requests.post(f"{self.url}/receive", json=data, headers=headers, timeout=1)
             return f"{resp.status_code} {resp.text}"
         except (requests.ConnectionError, requests.Timeout) as e:
-            pass
+            print(f"Failed to connect to the web storage server at {self.url}: {e}")
 
     def truncate(self, time: datetime) -> None:
         self.msgs = [m for m in self.msgs if datetime.fromisoformat(m["msg"]["timestamp"]) <= time]
@@ -100,7 +101,7 @@ class WebStorage(Storage):
                     "tag": "research.pdf_image",
                     "timestamp": timestamp,
                     "loop_id": li,
-                    "content": {"image": f"{timestamp}.jpg"},
+                    "content": {"image": f"pdf_images/{timestamp}.jpg"},
                 },
             }
         elif "experiment generation" in tag or "load_experiment" in tag:

rdagent/log/utils/__init__.py

@@ -92,7 +92,7 @@ def extract_loopid_func_name(tag: str) -> tuple[str, str] | tuple[None, None]:
 
 def extract_evoid(tag: str) -> str | None:
     """extract evo id from the tag in Message"""
-    match = re.search(r"\.evo_loop_(\d+)\.", tag)
+    match = re.search(r"evo_loop_(\d+)\.", tag)
     return cast(str, match.group(1)) if match else None
 
 

rdagent/scenarios/qlib/developer/factor_runner.py

@@ -19,17 +19,6 @@ from rdagent.scenarios.qlib.experiment.model_experiment import QlibModelExperime
 DIRNAME = Path(__file__).absolute().resolve().parent
 DIRNAME_local = Path.cwd()
 
-# class QlibFactorExpWorkspace:
-
-#     def prepare():
-#         # create a folder;
-#         # copy template
-#         # place data inside the folder `combined_factors`
-#         #
-#     def execute():
-#         de = DockerEnv()
-#         de.run(local_path=self.ws_path, entry="qrun conf_baseline.yaml")
-
 # TODO: supporting multiprocessing and keep previous results
 
 
@@ -89,6 +78,8 @@ class QlibFactorRunner(CachedRunner[QlibFactorExperiment]):
             "valid_start": fbps.valid_start,
             "valid_end": fbps.valid_end,
             "test_start": fbps.test_start,
+            "feature_names": str(list(exp.base_features.keys())),
+            "feature_expressions": str(list(exp.base_features.values())),
         }
         if fbps.test_end is not None:
             env_to_use.update({"test_end": fbps.test_end})
@@ -103,8 +94,8 @@ class QlibFactorRunner(CachedRunner[QlibFactorExperiment]):
                 logger.info(f"SOTA factor processing ...")
                 SOTA_factor = process_factor_data(sota_factor_experiments_list)
 
-            logger.info(f"New factor processing ...")
             # Process the new factors data
+            logger.info(f"New factor processing ...")
             new_factors = process_factor_data(exp)
 
             if new_factors.empty:
@@ -126,9 +117,10 @@ class QlibFactorRunner(CachedRunner[QlibFactorExperiment]):
             combined_factors = combined_factors.loc[:, ~combined_factors.columns.duplicated(keep="last")]
             new_columns = pd.MultiIndex.from_product([["feature"], combined_factors.columns])
             combined_factors.columns = new_columns
-            num_features = RD_AGENT_SETTINGS.initial_fator_library_size + len(combined_factors.columns)
             logger.info(f"Factor data processing completed.")
 
+            num_features = len(exp.base_features) + len(combined_factors.columns)
+
             # Due to the rdagent and qlib docker image in the numpy version of the difference,
             # the `combined_factors_df.pkl` file could not be loaded correctly in qlib dokcer,
             # so we changed the file type of `combined_factors_df` from pkl to parquet.
@@ -175,19 +167,30 @@ class QlibFactorRunner(CachedRunner[QlibFactorExperiment]):
             else:
                 # LGBM + combined factors
                 result, stdout = exp.experiment_workspace.execute(
-                    qlib_config_name=(
-                        f"conf_baseline.yaml" if len(exp.based_experiments) == 0 else "conf_combined_factors.yaml"
-                    ),
+                    qlib_config_name="conf_combined_factors.yaml",
                     run_env=env_to_use,
                 )
         else:
             logger.info(f"Experiment execution ...")
-            result, stdout = exp.experiment_workspace.execute(
-                qlib_config_name=(
-                    f"conf_baseline.yaml" if len(exp.based_experiments) == 0 else "conf_combined_factors.yaml"
-                ),
-                run_env=env_to_use,
-            )
+            if exp.base_feature_codes:
+                factors = process_factor_data(exp)
+                factors = factors.sort_index()
+                factors = factors.loc[:, ~factors.columns.duplicated(keep="last")]
+                new_columns = pd.MultiIndex.from_product([["feature"], factors.columns])
+                factors.columns = new_columns
+                target_path = exp.experiment_workspace.workspace_path / "combined_factors_df.parquet"
+                # Save the combined factors to the workspace
+                factors.to_parquet(target_path, engine="pyarrow")
+                logger.info(f"Factor data processing completed.")
+                result, stdout = exp.experiment_workspace.execute(
+                    qlib_config_name="conf_combined_factors.yaml",
+                    run_env=env_to_use,
+                )
+            else:
+                result, stdout = exp.experiment_workspace.execute(
+                    qlib_config_name="conf_baseline.yaml",
+                    run_env=env_to_use,
+                )
 
         if result is None:
             logger.error(f"Failed to run this experiment, because {stdout}")

rdagent/scenarios/qlib/developer/model_runner.py

@@ -47,7 +47,7 @@ class QlibModelRunner(CachedRunner[QlibModelExperiment]):
                 combined_factors = combined_factors.loc[:, ~combined_factors.columns.duplicated(keep="last")]
                 new_columns = pd.MultiIndex.from_product([["feature"], combined_factors.columns])
                 combined_factors.columns = new_columns
-                num_features = str(RD_AGENT_SETTINGS.initial_fator_library_size + len(combined_factors.columns))
+                num_features = str(len(exp.base_features) + len(combined_factors.columns))
 
                 target_path = exp.experiment_workspace.workspace_path / "combined_factors_df.parquet"
 
@@ -67,6 +67,8 @@ class QlibModelRunner(CachedRunner[QlibModelExperiment]):
             "valid_start": mbps.valid_start,
             "valid_end": mbps.valid_end,
             "test_start": mbps.test_start,
+            "feature_names": str(list(exp.base_features.keys())),
+            "feature_expressions": str(list(exp.base_features.values())),
         }
         if mbps.test_end is not None:
             env_to_use.update({"test_end": mbps.test_end})

rdagent/scenarios/qlib/developer/utils.py

@@ -3,6 +3,7 @@ from typing import List
 import pandas as pd
 
 from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback
+from rdagent.components.coder.factor_coder.factor import FactorFBWorkspace, FactorTask
 from rdagent.core.conf import RD_AGENT_SETTINGS
 from rdagent.core.exception import FactorEmptyError
 from rdagent.core.utils import multiprocessing_wrapper
@@ -10,6 +11,123 @@ from rdagent.log import rdagent_logger as logger
 from rdagent.scenarios.qlib.experiment.factor_experiment import QlibFactorExperiment
 
 
+def _build_base_feature_workspaces(exp: QlibFactorExperiment) -> list[FactorFBWorkspace]:
+    workspaces: list[FactorFBWorkspace] = []
+    for file_name, code in exp.base_feature_codes.items():
+        workspace = FactorFBWorkspace(
+            target_task=FactorTask(
+                factor_name=file_name,
+                factor_description=f"Base feature from {file_name}",
+                factor_formulation="",
+            )
+        )
+        workspace.inject_files(**{"factor.py": code})
+        workspaces.append(workspace)
+    return workspaces
+
+
+def _build_execute_calls(exp: QlibFactorExperiment, base_feature_workspaces: list[FactorFBWorkspace]) -> list[tuple]:
+    execute_calls = []
+
+    if exp.sub_tasks:
+        assert isinstance(exp.prop_dev_feedback, CoSTEERMultiFeedback)
+        execute_calls.extend(
+            (implementation.execute, ("All",))
+            for implementation, feedback in zip(exp.sub_workspace_list, exp.prop_dev_feedback)
+            if implementation and feedback
+        )
+
+    execute_calls.extend((workspace.execute, ("All",)) for workspace in base_feature_workspaces)
+    return execute_calls
+
+
+def _resolve_index_level_values(df: pd.DataFrame, level_name: str) -> pd.Index | None:
+    matching_levels = [idx for idx, name in enumerate(df.index.names) if name == level_name]
+    if not matching_levels:
+        return None
+
+    if len(matching_levels) == 1:
+        return df.index.get_level_values(matching_levels[0])
+
+    candidate_values = [df.index.get_level_values(idx) for idx in matching_levels]
+    first_values = candidate_values[0]
+    if all(first_values.equals(values) for values in candidate_values[1:]):
+        logger.warning(
+            f"Factor dataframe has duplicated '{level_name}' index levels at positions {matching_levels}; "
+            "their values are identical, so the first one is used."
+        )
+        return first_values
+
+    logger.warning(
+        f"Skip factor dataframe because index has ambiguous duplicated '{level_name}' levels at positions "
+        f"{matching_levels}. index names={list(df.index.names)}"
+    )
+    return None
+
+
+def _normalize_factor_index(df: pd.DataFrame) -> pd.DataFrame | None:
+    """Normalize factor index to a 2-level MultiIndex: (datetime, instrument)."""
+    if df is None or df.empty:
+        return None
+
+    index_names = list(df.index.names)
+    if "datetime" not in index_names:
+        return None
+
+    if "instrument" not in index_names:
+        logger.warning(f"Skip factor dataframe because index misses 'instrument'. index names={index_names}")
+        return None
+
+    datetime_values = _resolve_index_level_values(df, "datetime")
+    instrument_values = _resolve_index_level_values(df, "instrument")
+    if datetime_values is None or instrument_values is None:
+        return None
+
+    normalized = df.copy()
+    normalized.index = pd.MultiIndex.from_arrays(
+        [datetime_values, instrument_values],
+        names=["datetime", "instrument"],
+    )
+    return normalized
+
+
+def _format_index_info(df: pd.DataFrame | None) -> str:
+    if df is None:
+        return "df is None"
+    return f"index_type={type(df.index).__name__}, nlevels={df.index.nlevels}, names={list(df.index.names)}"
+
+
+def _process_message_and_df(
+    source_name: str,
+    message: str,
+    df: pd.DataFrame | None,
+    factor_dfs: list[pd.DataFrame],
+    error_message: str,
+) -> str:
+    index_info = _format_index_info(df)
+    if df is None or "datetime" not in df.index.names:
+        logger.warning(f"Factor data from {source_name} has invalid execution output or index: {index_info}")
+        logger.warning(f"Factor data from {source_name} is not generated because of {message}")
+        return (
+            f"{error_message}Factor data from {source_name} is not generated because of {message}. "
+            f"index_info={index_info}. "
+        )
+
+    normalized_df = _normalize_factor_index(df)
+    if normalized_df is None:
+        logger.warning(f"Factor data from {source_name} is skipped due to invalid index structure: {index_info}")
+        return f"{error_message}Factor data from {source_name} is skipped due to invalid index: {index_info}. "
+
+    time_diff = df.index.get_level_values("datetime").to_series().diff().dropna().unique()
+    if pd.Timedelta(minutes=1) in time_diff:
+        logger.warning(f"Factor data from {source_name} is not generated.")
+        return error_message
+
+    factor_dfs.append(normalized_df)
+    logger.info(f"Factor data from {source_name} is successfully generated.")
+    return error_message
+
+
 def process_factor_data(exp_or_list: List[QlibFactorExperiment] | QlibFactorExperiment) -> pd.DataFrame:
     """
     Process and combine factor data from experiment implementations.
@@ -23,44 +141,36 @@ def process_factor_data(exp_or_list: List[QlibFactorExperiment] | QlibFactorExpe
     if isinstance(exp_or_list, QlibFactorExperiment):
         exp_or_list = [exp_or_list]
     factor_dfs = []
+    error_message = ""
 
     # Collect all exp's dataframes
     for exp in exp_or_list:
-        if isinstance(exp, QlibFactorExperiment):
-            if len(exp.sub_tasks) > 0:
-                # if it has no sub_tasks, the experiment is results from template project.
-                # otherwise, it is developed with designed task. So it should have feedback.
-                assert isinstance(exp.prop_dev_feedback, CoSTEERMultiFeedback)
-                # Iterate over sub-implementations and execute them to get each factor data
-                message_and_df_list = multiprocessing_wrapper(
-                    [
-                        (implementation.execute, ("All",))
-                        for implementation, fb in zip(exp.sub_workspace_list, exp.prop_dev_feedback)
-                        if implementation and fb
-                    ],  # only execute successfully feedback
-                    n=RD_AGENT_SETTINGS.multi_proc_n,
-                )
-                error_message = ""
-                for message, df in message_and_df_list:
-                    # Check if factor generation was successful
-                    if df is not None and "datetime" in df.index.names:
-                        time_diff = df.index.get_level_values("datetime").to_series().diff().dropna().unique()
-                        if pd.Timedelta(minutes=1) not in time_diff:
-                            factor_dfs.append(df)
-                            logger.info(
-                                f"Factor data from {exp.hypothesis.concise_justification} is successfully generated."
-                            )
-                        else:
-                            logger.warning(f"Factor data from {exp.hypothesis.concise_justification} is not generated.")
-                    else:
-                        error_message += f"Factor data from {exp.hypothesis.concise_justification} is not generated because of {message}"
-                        logger.warning(
-                            f"Factor data from {exp.hypothesis.concise_justification} is not generated because of {message}"
-                        )
+        if not isinstance(exp, QlibFactorExperiment):
+            continue
+
+        source_name = exp.hypothesis.concise_justification if exp.hypothesis else "BASE factor files"
+        base_feature_workspaces = _build_base_feature_workspaces(exp)
+        execute_calls = _build_execute_calls(exp, base_feature_workspaces)
+        if not execute_calls:
+            continue
+
+        message_and_df_list = multiprocessing_wrapper(execute_calls, n=RD_AGENT_SETTINGS.multi_proc_n)
+        for message, df in message_and_df_list:
+            error_message = _process_message_and_df(source_name, message, df, factor_dfs, error_message)
 
     # Combine all successful factor data
     if factor_dfs:
-        return pd.concat(factor_dfs, axis=1)
+        try:
+            return pd.concat(factor_dfs, axis=1)
+        except Exception as concat_error:
+            concat_index_info = " | ".join([f"df#{i}: {_format_index_info(df)}" for i, df in enumerate(factor_dfs)])
+            logger.warning(
+                f"Failed to concat factor data due to index misalignment. concat_error={concat_error}; collected_index_info={concat_index_info}"
+            )
+            raise FactorEmptyError(
+                "Failed to concat factor data due to index misalignment or incompatible index structure. "
+                f"concat_error={concat_error}; collected_index_info={concat_index_info}; details={error_message}"
+            ) from concat_error
     else:
         raise FactorEmptyError(
             f"No valid factor data found to merge (in process_factor_data) because of {error_message}."

rdagent/scenarios/qlib/experiment/factor_experiment.py

@@ -21,6 +21,10 @@ class QlibFactorExperiment(FactorExperiment[FactorTask, QlibFBWorkspace, FactorF
         super().__init__(*args, **kwargs)
         self.experiment_workspace = QlibFBWorkspace(template_folder_path=Path(__file__).parent / "factor_template")
         self.stdout = ""
+        self.base_features: dict[str, str] = (
+            {}
+        )  # Qlib features in operator form, e.g., "RESI5": "Resi($close, 5)/$close"
+        self.base_feature_codes: dict[str, str] = {}  # Qlib features in code form
 
 
 class QlibFactorScenario(Scenario):

rdagent/scenarios/qlib/experiment/factor_template/conf_baseline.yaml

@@ -8,30 +8,36 @@ benchmark: &benchmark SH000300
 data_handler_config: &data_handler_config
     start_time: {{ train_start | default("2008-01-01", true) }}
     end_time: {{ test_end | default("null", true) }}
-    fit_start_time: {{ train_start | default("2008-01-01", true) }}
-    fit_end_time: {{ train_end | default("2014-12-31", true) }}
     instruments: *market
+    data_loader:
+        class: NestedDataLoader
+        kwargs:
+            dataloader_l:
+                - class: qlib.contrib.data.loader.Alpha158DL
+                  kwargs:
+                    config:
+                        label: 
+                            - ["Ref($close, -2)/Ref($close, -1) - 1"]
+                            - ["LABEL0"]
+                        feature:
+                            - {{ feature_expressions }}
+                            - {{ feature_names }}
+    
     infer_processors:
-        - class: FilterCol
-          kwargs:
-              fields_group: feature
-              col_list: ["RESI5", "WVMA5", "RSQR5", "KLEN", "RSQR10", "CORR5", "CORD5", "CORR10", 
-                            "ROC60", "RESI10", "VSTD5", "RSQR60", "CORR60", "WVMA60", "STD5", 
-                            "RSQR20", "CORD60", "CORD10", "CORR20", "KLOW"
-                        ]
         - class: RobustZScoreNorm
           kwargs:
               fields_group: feature
               clip_outlier: true
+              fit_start_time: {{ train_start | default("2008-01-01", true) }}
+              fit_end_time: {{ train_end | default("2014-12-31", true) }}
         - class: Fillna
           kwargs:
               fields_group: feature
     learn_processors:
         - class: DropnaLabel
-        - class: CSRankNorm
+        - class: CSZScoreNorm
           kwargs:
               fields_group: label
-    label: ["Ref($close, -2) / Ref($close, -1) - 1"] 
 
 port_analysis_config: &port_analysis_config
     strategy:
@@ -52,6 +58,7 @@ port_analysis_config: &port_analysis_config
             open_cost: 0.0005
             close_cost: 0.0015
             min_cost: 5
+
 task:
     model:
         class: LGBModel
@@ -71,7 +78,7 @@ task:
         module_path: qlib.data.dataset
         kwargs:
             handler:
-                class: Alpha158
+                class: DataHandlerLP
                 module_path: qlib.contrib.data.handler
                 kwargs: *data_handler_config
             segments:

rdagent/scenarios/qlib/experiment/factor_template/conf_combined_factors.yaml

@@ -20,17 +20,8 @@ data_handler_config: &data_handler_config
                             - ["Ref($close, -2)/Ref($close, -1) - 1"]
                             - ["LABEL0"]
                         feature:
-                            - ["Resi($close, 5)/$close", "Std(Abs($close/Ref($close, 1)-1)*$volume, 5)/(Mean(Abs($close/Ref($close, 1)-1)*$volume, 5)+1e-12)",
-                               "Rsquare($close, 5)", "($high-$low)/$open", "Rsquare($close, 10)", "Corr($close, Log($volume+1), 5)",
-                               "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 5)", "Corr($close, Log($volume+1), 10)",
-                               "Ref($close, 60)/$close", "Resi($close, 10)/$close", "Std($volume, 5)/($volume+1e-12)",
-                               "Rsquare($close, 60)", "Corr($close, Log($volume+1), 60)", "Std(Abs($close/Ref($close, 1)-1)*$volume, 60)/(Mean(Abs($close/Ref($close, 1)-1)*$volume, 60)+1e-12)",
-                               "Std($close, 5)/$close", "Rsquare($close, 20)", "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 60)",
-                               "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 10)", "Corr($close, Log($volume+1), 20)",
-                               "(Less($open, $close)-$low)/$open"]
-                            - ["RESI5", "WVMA5", "RSQR5", "KLEN", "RSQR10", "CORR5", "CORD5", "CORR10", 
-                               "ROC60", "RESI10", "VSTD5", "RSQR60", "CORR60", "WVMA60", "STD5", 
-                               "RSQR20", "CORD60", "CORD10", "CORR20", "KLOW"]
+                            - {{ feature_expressions }}
+                            - {{ feature_names }}
                 - class: qlib.data.dataset.loader.StaticDataLoader
                   kwargs:
                     config: "combined_factors_df.parquet"

rdagent/scenarios/qlib/experiment/factor_template/conf_combined_factors_sota_model.yaml

@@ -20,17 +20,8 @@ data_handler_config: &data_handler_config
                             - ["Ref($close, -2)/Ref($close, -1) - 1"]
                             - ["LABEL0"]
                         feature:
-                            - ["Resi($close, 5)/$close", "Std(Abs($close/Ref($close, 1)-1)*$volume, 5)/(Mean(Abs($close/Ref($close, 1)-1)*$volume, 5)+1e-12)",
-                               "Rsquare($close, 5)", "($high-$low)/$open", "Rsquare($close, 10)", "Corr($close, Log($volume+1), 5)",
-                               "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 5)", "Corr($close, Log($volume+1), 10)",
-                               "Ref($close, 60)/$close", "Resi($close, 10)/$close", "Std($volume, 5)/($volume+1e-12)",
-                               "Rsquare($close, 60)", "Corr($close, Log($volume+1), 60)", "Std(Abs($close/Ref($close, 1)-1)*$volume, 60)/(Mean(Abs($close/Ref($close, 1)-1)*$volume, 60)+1e-12)",
-                               "Std($close, 5)/$close", "Rsquare($close, 20)", "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 60)",
-                               "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 10)", "Corr($close, Log($volume+1), 20)",
-                               "(Less($open, $close)-$low)/$open"]
-                            - ["RESI5", "WVMA5", "RSQR5", "KLEN", "RSQR10", "CORR5", "CORD5", "CORR10", 
-                               "ROC60", "RESI10", "VSTD5", "RSQR60", "CORR60", "WVMA60", "STD5", 
-                               "RSQR20", "CORD60", "CORD10", "CORR20", "KLOW"]
+                            - {{ feature_expressions }}
+                            - {{ feature_names }}
                 - class: qlib.data.dataset.loader.StaticDataLoader
                   kwargs:
                     config: "combined_factors_df.parquet"

rdagent/scenarios/qlib/experiment/model_experiment.py

@@ -20,6 +20,7 @@ class QlibModelExperiment(ModelExperiment[ModelTask, QlibFBWorkspace, ModelFBWor
         super().__init__(*args, **kwargs)
         self.experiment_workspace = QlibFBWorkspace(template_folder_path=Path(__file__).parent / "model_template")
         self.stdout = ""
+        self.base_features: dict[str, str] = {}
 
 
 class QlibModelScenario(Scenario):

rdagent/scenarios/qlib/experiment/model_template/conf_baseline_factors_model.yaml

@@ -3,33 +3,40 @@ qlib_init:
     region: cn
 market: &market csi300
 benchmark: &benchmark SH000300
+
 data_handler_config: &data_handler_config
     start_time: {{ train_start | default("2008-01-01", true) }}
     end_time: {{ test_end | default("null", true) }}
-    fit_start_time: {{ train_start | default("2008-01-01", true) }}
-    fit_end_time: {{ train_end | default("2014-12-31", true) }}
     instruments: *market
+    data_loader:
+        class: NestedDataLoader
+        kwargs:
+            dataloader_l:
+                - class: qlib.contrib.data.loader.Alpha158DL
+                  kwargs:
+                    config:
+                        label: 
+                            - ["Ref($close, -2)/Ref($close, -1) - 1"]
+                            - ["LABEL0"]
+                        feature:
+                            - {{ feature_expressions }}
+                            - {{ feature_names }}
+    
     infer_processors:
-        - class: FilterCol
-          kwargs:
-              fields_group: feature
-              col_list: ["RESI5", "WVMA5", "RSQR5", "KLEN", "RSQR10", "CORR5", "CORD5", "CORR10", 
-                            "ROC60", "RESI10", "VSTD5", "RSQR60", "CORR60", "WVMA60", "STD5", 
-                            "RSQR20", "CORD60", "CORD10", "CORR20", "KLOW"
-                        ]
         - class: RobustZScoreNorm
           kwargs:
               fields_group: feature
               clip_outlier: true
+              fit_start_time: {{ train_start | default("2008-01-01", true) }}
+              fit_end_time: {{ train_end | default("2014-12-31", true) }}
         - class: Fillna
           kwargs:
               fields_group: feature
     learn_processors:
         - class: DropnaLabel
-        - class: CSRankNorm
+        - class: CSZScoreNorm
           kwargs:
               fields_group: label
-    label: ["Ref($close, -2) / Ref($close, -1) - 1"] 
 
 port_analysis_config: &port_analysis_config
     strategy:
@@ -74,7 +81,7 @@ task:
         module_path: qlib.data.dataset
         kwargs:
             handler:
-                class: Alpha158
+                class: DataHandlerLP
                 module_path: qlib.contrib.data.handler
                 kwargs: *data_handler_config
             segments:

rdagent/scenarios/qlib/experiment/model_template/conf_sota_factors_model.yaml

@@ -20,17 +20,8 @@ data_handler_config: &data_handler_config
                             - ["Ref($close, -2)/Ref($close, -1) - 1"]
                             - ["LABEL0"]
                         feature:
-                            - ["Resi($close, 5)/$close", "Std(Abs($close/Ref($close, 1)-1)*$volume, 5)/(Mean(Abs($close/Ref($close, 1)-1)*$volume, 5)+1e-12)",
-                               "Rsquare($close, 5)", "($high-$low)/$open", "Rsquare($close, 10)", "Corr($close, Log($volume+1), 5)",
-                               "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 5)", "Corr($close, Log($volume+1), 10)",
-                               "Ref($close, 60)/$close", "Resi($close, 10)/$close", "Std($volume, 5)/($volume+1e-12)",
-                               "Rsquare($close, 60)", "Corr($close, Log($volume+1), 60)", "Std(Abs($close/Ref($close, 1)-1)*$volume, 60)/(Mean(Abs($close/Ref($close, 1)-1)*$volume, 60)+1e-12)",
-                               "Std($close, 5)/$close", "Rsquare($close, 20)", "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 60)",
-                               "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 10)", "Corr($close, Log($volume+1), 20)",
-                               "(Less($open, $close)-$low)/$open"]
-                            - ["RESI5", "WVMA5", "RSQR5", "KLEN", "RSQR10", "CORR5", "CORD5", "CORR10", 
-                               "ROC60", "RESI10", "VSTD5", "RSQR60", "CORR60", "WVMA60", "STD5", 
-                               "RSQR20", "CORD60", "CORD10", "CORR20", "KLOW"]
+                            - {{ feature_expressions }}
+                            - {{ feature_names }}
                 - class: qlib.data.dataset.loader.StaticDataLoader
                   kwargs:
                     config: "combined_factors_df.parquet"

rdagent/utils/qlib.py

@@ -0,0 +1,208 @@
+from rdagent.core.experiment import FBWorkspace
+from rdagent.utils.env import QlibCondaConf, QlibCondaEnv
+
+ALPHA20 = {
+    "RESI5": "Resi($close, 5)/$close",
+    "WVMA5": "Std(Abs($close/Ref($close, 1)-1)*$volume, 5)/(Mean(Abs($close/Ref($close, 1)-1)*$volume, 5)+1e-12)",
+    "RSQR5": "Rsquare($close, 5)",
+    "KLEN": "($high-$low)/$open",
+    "RSQR10": "Rsquare($close, 10)",
+    "CORR5": "Corr($close, Log($volume+1), 5)",
+    "CORD5": "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 5)",
+    "CORR10": "Corr($close, Log($volume+1), 10)",
+    "ROC60": "Ref($close, 60)/$close",
+    "RESI10": "Resi($close, 10)/$close",
+    "VSTD5": "Std($volume, 5)/($volume+1e-12)",
+    "RSQR60": "Rsquare($close, 60)",
+    "CORR60": "Corr($close, Log($volume+1), 60)",
+    "WVMA60": "Std(Abs($close/Ref($close, 1)-1)*$volume, 60)/(Mean(Abs($close/Ref($close, 1)-1)*$volume, 60)+1e-12)",
+    "STD5": "Std($close, 5)/$close",
+    "RSQR20": "Rsquare($close, 20)",
+    "CORD60": "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 60)",
+    "CORD10": "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 10)",
+    "CORR20": "Corr($close, Log($volume+1), 20)",
+    "KLOW": "(Less($open, $close)-$low)/$open",
+}
+
+ALPHA158 = {
+    "KMID": "($close-$open)/$open",
+    "KLEN": "($high-$low)/$open",
+    "KMID2": "($close-$open)/($high-$low+1e-12)",
+    "KUP": "($high-Greater($open, $close))/$open",
+    "KUP2": "($high-Greater($open, $close))/($high-$low+1e-12)",
+    "KLOW": "(Less($open, $close)-$low)/$open",
+    "KLOW2": "(Less($open, $close)-$low)/($high-$low+1e-12)",
+    "KSFT": "(2*$close-$high-$low)/$open",
+    "KSFT2": "(2*$close-$high-$low)/($high-$low+1e-12)",
+    "OPEN0": "$open/$close",
+    "HIGH0": "$high/$close",
+    "LOW0": "$low/$close",
+    "VWAP0": "$vwap/$close",
+    "ROC5": "Ref($close, 5)/$close",
+    "ROC10": "Ref($close, 10)/$close",
+    "ROC20": "Ref($close, 20)/$close",
+    "ROC30": "Ref($close, 30)/$close",
+    "ROC60": "Ref($close, 60)/$close",
+    "MA5": "Mean($close, 5)/$close",
+    "MA10": "Mean($close, 10)/$close",
+    "MA20": "Mean($close, 20)/$close",
+    "MA30": "Mean($close, 30)/$close",
+    "MA60": "Mean($close, 60)/$close",
+    "STD5": "Std($close, 5)/$close",
+    "STD10": "Std($close, 10)/$close",
+    "STD20": "Std($close, 20)/$close",
+    "STD30": "Std($close, 30)/$close",
+    "STD60": "Std($close, 60)/$close",
+    "BETA5": "Slope($close, 5)/$close",
+    "BETA10": "Slope($close, 10)/$close",
+    "BETA20": "Slope($close, 20)/$close",
+    "BETA30": "Slope($close, 30)/$close",
+    "BETA60": "Slope($close, 60)/$close",
+    "RSQR5": "Rsquare($close, 5)",
+    "RSQR10": "Rsquare($close, 10)",
+    "RSQR20": "Rsquare($close, 20)",
+    "RSQR30": "Rsquare($close, 30)",
+    "RSQR60": "Rsquare($close, 60)",
+    "RESI5": "Resi($close, 5)/$close",
+    "RESI10": "Resi($close, 10)/$close",
+    "RESI20": "Resi($close, 20)/$close",
+    "RESI30": "Resi($close, 30)/$close",
+    "RESI60": "Resi($close, 60)/$close",
+    "MAX5": "Max($high, 5)/$close",
+    "MAX10": "Max($high, 10)/$close",
+    "MAX20": "Max($high, 20)/$close",
+    "MAX30": "Max($high, 30)/$close",
+    "MAX60": "Max($high, 60)/$close",
+    "MIN5": "Min($low, 5)/$close",
+    "MIN10": "Min($low, 10)/$close",
+    "MIN20": "Min($low, 20)/$close",
+    "MIN30": "Min($low, 30)/$close",
+    "MIN60": "Min($low, 60)/$close",
+    "QTLU5": "Quantile($close, 5, 0.8)/$close",
+    "QTLU10": "Quantile($close, 10, 0.8)/$close",
+    "QTLU20": "Quantile($close, 20, 0.8)/$close",
+    "QTLU30": "Quantile($close, 30, 0.8)/$close",
+    "QTLU60": "Quantile($close, 60, 0.8)/$close",
+    "QTLD5": "Quantile($close, 5, 0.2)/$close",
+    "QTLD10": "Quantile($close, 10, 0.2)/$close",
+    "QTLD20": "Quantile($close, 20, 0.2)/$close",
+    "QTLD30": "Quantile($close, 30, 0.2)/$close",
+    "QTLD60": "Quantile($close, 60, 0.2)/$close",
+    "RANK5": "Rank($close, 5)",
+    "RANK10": "Rank($close, 10)",
+    "RANK20": "Rank($close, 20)",
+    "RANK30": "Rank($close, 30)",
+    "RANK60": "Rank($close, 60)",
+    "RSV5": "($close-Min($low, 5))/(Max($high, 5)-Min($low, 5)+1e-12)",
+    "RSV10": "($close-Min($low, 10))/(Max($high, 10)-Min($low, 10)+1e-12)",
+    "RSV20": "($close-Min($low, 20))/(Max($high, 20)-Min($low, 20)+1e-12)",
+    "RSV30": "($close-Min($low, 30))/(Max($high, 30)-Min($low, 30)+1e-12)",
+    "RSV60": "($close-Min($low, 60))/(Max($high, 60)-Min($low, 60)+1e-12)",
+    "IMAX5": "IdxMax($high, 5)/5",
+    "IMAX10": "IdxMax($high, 10)/10",
+    "IMAX20": "IdxMax($high, 20)/20",
+    "IMAX30": "IdxMax($high, 30)/30",
+    "IMAX60": "IdxMax($high, 60)/60",
+    "IMIN5": "IdxMin($low, 5)/5",
+    "IMIN10": "IdxMin($low, 10)/10",
+    "IMIN20": "IdxMin($low, 20)/20",
+    "IMIN30": "IdxMin($low, 30)/30",
+    "IMIN60": "IdxMin($low, 60)/60",
+    "IMXD5": "(IdxMax($high, 5)-IdxMin($low, 5))/5",
+    "IMXD10": "(IdxMax($high, 10)-IdxMin($low, 10))/10",
+    "IMXD20": "(IdxMax($high, 20)-IdxMin($low, 20))/20",
+    "IMXD30": "(IdxMax($high, 30)-IdxMin($low, 30))/30",
+    "IMXD60": "(IdxMax($high, 60)-IdxMin($low, 60))/60",
+    "CORR5": "Corr($close, Log($volume+1), 5)",
+    "CORR10": "Corr($close, Log($volume+1), 10)",
+    "CORR20": "Corr($close, Log($volume+1), 20)",
+    "CORR30": "Corr($close, Log($volume+1), 30)",
+    "CORR60": "Corr($close, Log($volume+1), 60)",
+    "CORD5": "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 5)",
+    "CORD10": "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 10)",
+    "CORD20": "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 20)",
+    "CORD30": "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 30)",
+    "CORD60": "Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), 60)",
+    "CNTP5": "Mean($close>Ref($close, 1), 5)",
+    "CNTP10": "Mean($close>Ref($close, 1), 10)",
+    "CNTP20": "Mean($close>Ref($close, 1), 20)",
+    "CNTP30": "Mean($close>Ref($close, 1), 30)",
+    "CNTP60": "Mean($close>Ref($close, 1), 60)",
+    "CNTN5": "Mean($close<Ref($close, 1), 5)",
+    "CNTN10": "Mean($close<Ref($close, 1), 10)",
+    "CNTN20": "Mean($close<Ref($close, 1), 20)",
+    "CNTN30": "Mean($close<Ref($close, 1), 30)",
+    "CNTN60": "Mean($close<Ref($close, 1), 60)",
+    "CNTD5": "Mean($close>Ref($close, 1), 5)-Mean($close<Ref($close, 1), 5)",
+    "CNTD10": "Mean($close>Ref($close, 1), 10)-Mean($close<Ref($close, 1), 10)",
+    "CNTD20": "Mean($close>Ref($close, 1), 20)-Mean($close<Ref($close, 1), 20)",
+    "CNTD30": "Mean($close>Ref($close, 1), 30)-Mean($close<Ref($close, 1), 30)",
+    "CNTD60": "Mean($close>Ref($close, 1), 60)-Mean($close<Ref($close, 1), 60)",
+    "SUMP5": "Sum(Greater($close-Ref($close, 1), 0), 5)/(Sum(Abs($close-Ref($close, 1)), 5)+1e-12)",
+    "SUMP10": "Sum(Greater($close-Ref($close, 1), 0), 10)/(Sum(Abs($close-Ref($close, 1)), 10)+1e-12)",
+    "SUMP20": "Sum(Greater($close-Ref($close, 1), 0), 20)/(Sum(Abs($close-Ref($close, 1)), 20)+1e-12)",
+    "SUMP30": "Sum(Greater($close-Ref($close, 1), 0), 30)/(Sum(Abs($close-Ref($close, 1)), 30)+1e-12)",
+    "SUMP60": "Sum(Greater($close-Ref($close, 1), 0), 60)/(Sum(Abs($close-Ref($close, 1)), 60)+1e-12)",
+    "SUMN5": "Sum(Greater(Ref($close, 1)-$close, 0), 5)/(Sum(Abs($close-Ref($close, 1)), 5)+1e-12)",
+    "SUMN10": "Sum(Greater(Ref($close, 1)-$close, 0), 10)/(Sum(Abs($close-Ref($close, 1)), 10)+1e-12)",
+    "SUMN20": "Sum(Greater(Ref($close, 1)-$close, 0), 20)/(Sum(Abs($close-Ref($close, 1)), 20)+1e-12)",
+    "SUMN30": "Sum(Greater(Ref($close, 1)-$close, 0), 30)/(Sum(Abs($close-Ref($close, 1)), 30)+1e-12)",
+    "SUMN60": "Sum(Greater(Ref($close, 1)-$close, 0), 60)/(Sum(Abs($close-Ref($close, 1)), 60)+1e-12)",
+    "SUMD5": "(Sum(Greater($close-Ref($close, 1), 0), 5)-Sum(Greater(Ref($close, 1)-$close, 0), 5))/(Sum(Abs($close-Ref($close, 1)), 5)+1e-12)",
+    "SUMD10": "(Sum(Greater($close-Ref($close, 1), 0), 10)-Sum(Greater(Ref($close, 1)-$close, 0), 10))/(Sum(Abs($close-Ref($close, 1)), 10)+1e-12)",
+    "SUMD20": "(Sum(Greater($close-Ref($close, 1), 0), 20)-Sum(Greater(Ref($close, 1)-$close, 0), 20))/(Sum(Abs($close-Ref($close, 1)), 20)+1e-12)",
+    "SUMD30": "(Sum(Greater($close-Ref($close, 1), 0), 30)-Sum(Greater(Ref($close, 1)-$close, 0), 30))/(Sum(Abs($close-Ref($close, 1)), 30)+1e-12)",
+    "SUMD60": "(Sum(Greater($close-Ref($close, 1), 0), 60)-Sum(Greater(Ref($close, 1)-$close, 0), 60))/(Sum(Abs($close-Ref($close, 1)), 60)+1e-12)",
+    "VMA5": "Mean($volume, 5)/($volume+1e-12)",
+    "VMA10": "Mean($volume, 10)/($volume+1e-12)",
+    "VMA20": "Mean($volume, 20)/($volume+1e-12)",
+    "VMA30": "Mean($volume, 30)/($volume+1e-12)",
+    "VMA60": "Mean($volume, 60)/($volume+1e-12)",
+    "VSTD5": "Std($volume, 5)/($volume+1e-12)",
+    "VSTD10": "Std($volume, 10)/($volume+1e-12)",
+    "VSTD20": "Std($volume, 20)/($volume+1e-12)",
+    "VSTD30": "Std($volume, 30)/($volume+1e-12)",
+    "VSTD60": "Std($volume, 60)/($volume+1e-12)",
+    "WVMA5": "Std(Abs($close/Ref($close, 1)-1)*$volume, 5)/(Mean(Abs($close/Ref($close, 1)-1)*$volume, 5)+1e-12)",
+    "WVMA10": "Std(Abs($close/Ref($close, 1)-1)*$volume, 10)/(Mean(Abs($close/Ref($close, 1)-1)*$volume, 10)+1e-12)",
+    "WVMA20": "Std(Abs($close/Ref($close, 1)-1)*$volume, 20)/(Mean(Abs($close/Ref($close, 1)-1)*$volume, 20)+1e-12)",
+    "WVMA30": "Std(Abs($close/Ref($close, 1)-1)*$volume, 30)/(Mean(Abs($close/Ref($close, 1)-1)*$volume, 30)+1e-12)",
+    "WVMA60": "Std(Abs($close/Ref($close, 1)-1)*$volume, 60)/(Mean(Abs($close/Ref($close, 1)-1)*$volume, 60)+1e-12)",
+    "VSUMP5": "Sum(Greater($volume-Ref($volume, 1), 0), 5)/(Sum(Abs($volume-Ref($volume, 1)), 5)+1e-12)",
+    "VSUMP10": "Sum(Greater($volume-Ref($volume, 1), 0), 10)/(Sum(Abs($volume-Ref($volume, 1)), 10)+1e-12)",
+    "VSUMP20": "Sum(Greater($volume-Ref($volume, 1), 0), 20)/(Sum(Abs($volume-Ref($volume, 1)), 20)+1e-12)",
+    "VSUMP30": "Sum(Greater($volume-Ref($volume, 1), 0), 30)/(Sum(Abs($volume-Ref($volume, 1)), 30)+1e-12)",
+    "VSUMP60": "Sum(Greater($volume-Ref($volume, 1), 0), 60)/(Sum(Abs($volume-Ref($volume, 1)), 60)+1e-12)",
+    "VSUMN5": "Sum(Greater(Ref($volume, 1)-$volume, 0), 5)/(Sum(Abs($volume-Ref($volume, 1)), 5)+1e-12)",
+    "VSUMN10": "Sum(Greater(Ref($volume, 1)-$volume, 0), 10)/(Sum(Abs($volume-Ref($volume, 1)), 10)+1e-12)",
+    "VSUMN20": "Sum(Greater(Ref($volume, 1)-$volume, 0), 20)/(Sum(Abs($volume-Ref($volume, 1)), 20)+1e-12)",
+    "VSUMN30": "Sum(Greater(Ref($volume, 1)-$volume, 0), 30)/(Sum(Abs($volume-Ref($volume, 1)), 30)+1e-12)",
+    "VSUMN60": "Sum(Greater(Ref($volume, 1)-$volume, 0), 60)/(Sum(Abs($volume-Ref($volume, 1)), 60)+1e-12)",
+    "VSUMD5": "(Sum(Greater($volume-Ref($volume, 1), 0), 5)-Sum(Greater(Ref($volume, 1)-$volume, 0), 5))/(Sum(Abs($volume-Ref($volume, 1)), 5)+1e-12)",
+    "VSUMD10": "(Sum(Greater($volume-Ref($volume, 1), 0), 10)-Sum(Greater(Ref($volume, 1)-$volume, 0), 10))/(Sum(Abs($volume-Ref($volume, 1)), 10)+1e-12)",
+    "VSUMD20": "(Sum(Greater($volume-Ref($volume, 1), 0), 20)-Sum(Greater(Ref($volume, 1)-$volume, 0), 20))/(Sum(Abs($volume-Ref($volume, 1)), 20)+1e-12)",
+    "VSUMD30": "(Sum(Greater($volume-Ref($volume, 1), 0), 30)-Sum(Greater(Ref($volume, 1)-$volume, 0), 30))/(Sum(Abs($volume-Ref($volume, 1)), 30)+1e-12)",
+    "VSUMD60": "(Sum(Greater($volume-Ref($volume, 1), 0), 60)-Sum(Greater(Ref($volume, 1)-$volume, 0), 60))/(Sum(Abs($volume-Ref($volume, 1)), 60)+1e-12)",
+}
+
+_TFW = FBWorkspace()  # test feature workspace
+TEST_FEATURE_CODE = """
+import qlib  
+from qlib.data import D  
+
+qlib.init()  
+expressions = {experessions}
+df = D.features(["SH600000"], expressions, start_time="2008-01-01", end_time="2020-08-31")
+"""
+
+
+def validate_qlib_features(expressions: list[str]) -> bool:
+    _TFW.inject_files(**{"test_fea.py": TEST_FEATURE_CODE.format(experessions=str(expressions))})
+
+    qlib_env = QlibCondaEnv(conf=QlibCondaConf())
+    qlib_env.prepare()
+    res = _TFW.run(
+        env=qlib_env,
+        entry="python test_fea.py",
+    )
+    return res.exit_code == 0

rdagent/utils/workflow/loop.py

@@ -11,6 +11,7 @@ Postscripts:
 import asyncio
 import concurrent.futures
 import copy
+import multiprocessing.queues
 import os
 import pickle
 from collections import defaultdict
@@ -528,8 +529,11 @@ class LoopBase:
     def __getstate__(self) -> dict[str, Any]:
         res = {}
         for k, v in self.__dict__.items():
-            if k not in ["queue", "semaphores", "_pbar"]:
-                res[k] = v
+            if k in ["queue", "semaphores", "_pbar"]:
+                continue
+            if isinstance(v, multiprocessing.queues.Queue):  # interaction queues are not picklable
+                continue
+            res[k] = v
         return res
 
     def __setstate__(self, state: dict[str, Any]) -> None: