Data mining (#103)

* scen

* scen2

* app

* fix

* Simplify workflow

* We can share more code in new scenarios

* rename model to rd loop

* Optimize data path

* Update rdagent/app/data_mining/model.py

* Add TODO

* Support GPU

* gpu

---------

Co-authored-by: SH-Src <suhan.c@outlook.com>

rdagent/utils/env.py

@@ -3,8 +3,8 @@ The motiviation of the utils is for environment management
 
 Tries to create uniform environment for the agent to run;
 - All the code and data is expected included in one folder
-
 """
+# TODO: move the scenario specific docker env into other folders.
 
 import os
 import subprocess
@@ -125,6 +125,7 @@ class DockerConf(BaseSettings):
     # So we just want to download it once.
     network: str | None = "bridge"  # the network mode for the docker
     shm_size: str | None = None
+    enable_gpu: bool = True  # because we will automatically disable GPU if not available. So we enable it by default.
 
 
 class QlibDockerConf(DockerConf):
@@ -141,6 +142,19 @@ class QlibDockerConf(DockerConf):
     enable_gpu: bool = True
 
 
+class DMDockerConf(DockerConf):
+    class Config:
+        env_prefix = "DM_DOCKER_"  
+
+    build_from_dockerfile: bool = True
+    dockerfile_folder_path: Path = Path(__file__).parent.parent / "scenarios" / "data_mining" / "docker"
+    image: str = "local_dm:latest"
+    mount_path: str = "/workspace/dm_workspace/"
+    default_entry: str = "python train.py"
+    extra_volumes: dict = {Path("~/.rdagent/.data/physionet.org/files/mimic-eicu-fiddle-feature/1.0.0/FIDDLE_mimic3/").expanduser().resolve(): "/root/.data/"}
+    shm_size: str | None = "16g"
+
+# physionet.org/files/mimic-eicu-fiddle-feature/1.0.0/FIDDLE_mimic3
 class DockerEnv(Env[DockerConf]):
     # TODO: Save the output into a specific file
 
@@ -243,3 +257,23 @@ class QTDockerEnv(DockerEnv):
             self.run(entry=cmd)
         else:
             logger.info("Data already exists. Download skipped.")
+
+
+class DMDockerEnv(DockerEnv):
+    """Qlib Torch Docker"""
+
+    def __init__(self, conf: DockerConf = DMDockerConf()):
+        super().__init__(conf)
+
+    def prepare(self, username: str, password: str):
+        """
+        Download image & data if it doesn't exist
+        """
+        super().prepare()
+        data_path = next(iter(self.conf.extra_volumes.keys()))
+        if not (Path(data_path)).exists():
+            logger.info("We are downloading!")
+            cmd = 'wget -r -N -c -np --user={} --password={} -P ~/.rdagent/.data/ https://physionet.org/files/mimic-eicu-fiddle-feature/1.0.0/'.format(username, password)
+            os.system(cmd)
+        else:
+            logger.info("Data already exists. Download skipped.")

rdagent/utils/workflow.py

@@ -13,7 +13,7 @@ from tqdm.auto import tqdm
 
 
 from collections import defaultdict
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 import datetime
 from typing import Callable
 from rdagent.log import rdagent_logger as logger
@@ -21,15 +21,23 @@ from rdagent.log import rdagent_logger as logger
 
 class LoopMeta(type):
 
-    def __new__(cls, clsname, bases, attrs):
-
-        # move custommized steps into steps
+    @staticmethod
+    def _get_steps(bases):
+        """
+        get all the `steps` of base classes and combine them to a single one.
+        """
         steps = []
-        for name in attrs.keys():
-            if not name.startswith("__"):
+        for base in bases:
+            steps.extend(LoopMeta._get_steps(base.__bases__) + getattr(base,"steps", []))
+        return steps
+
+    def __new__(cls, clsname, bases, attrs):
+        # move custommized steps into steps
+        steps = LoopMeta._get_steps(bases)  # all the base classes of parents
+        for name, attr in attrs.items():
+            if not name.startswith("__") and isinstance(attr, Callable):
                 steps.append(name)
         attrs["steps"] = steps
-
         return super().__new__(cls, clsname, bases, attrs)
 
 
@@ -44,6 +52,8 @@ class LoopBase:
     steps: list[Callable]  # a list of steps to work on
     loop_trace: dict[int, list[LoopTrace]]
 
+    skip_loop_error: tuple[Exception]  = field(default_factory=tuple)  # you can define a list of error that will skip current loop
+
     def __init__(self):
         self.loop_idx = 0 # current loop index
         self.step_idx = 0 # the index of next step to be run
@@ -73,7 +83,14 @@ class LoopBase:
 
                 name = self.steps[si]
                 func = getattr(self, name)
-                self.loop_prev_out[name] = func(self.loop_prev_out)
+                try:
+                    self.loop_prev_out[name] = func(self.loop_prev_out)
+                    # TODO: Fix the error logger.exception(f"Skip loop {li} due to {e}")
+                except self.skip_loop_error as e:
+                    logger.warning(f"Skip loop {li} due to {e}")
+                    self.loop_idx += 1
+                    self.step_index = 0
+                    continue
 
                 end = datetime.datetime.now(datetime.timezone.utc)