docstr: A dataset for meta model.
  
  Parameters
  ----------
  task_tpl : Union[dict, list]
      Decide what tasks are used.
      - dict : the task template, the prepared task is generated with `step`, `trunc_days` and `RollingGen`
      - list : when list, use the list of tasks directly
               the list is supposed to be sorted according timeline
  step : int
      the rolling step
  trunc_days: int
      days to be truncated based on the test start
  rolling_ext_days: int
      sometimes users want to train meta models for a longer test period but with smaller rolling steps for more task samples.
      the total length of test periods will be `step + rolling_ext_days`
  
  exp_name : Union[str, InternalData]
      Decide what meta_info are used for prediction.
      - str: the name of the experiment to store the performance of data
      - InternalData: a prepared internal data
  segments: Union[Dict[Text, Tuple], float]
      the segments to divide data
      both left and right
      if segments is a float:
          the float represents the percentage of data for training
  hist_step_n: int
      length of historical steps for the meta infomation
  task_mode : str
      Please refer to the docs of MetaTask
docstr_tok: ["A","dataset","for","meta","model",".","Parameters","--","--","--","--","--","task_tpl",":","Union","[","dict",",","list","]","Decide","what","tasks","are","used",".","-","dict",":","the","task","template",",","the","prepared","task","is","generated","with","`","step","`",",","`","trunc_days","`","and","`","RollingGen","`","-","list",":","when","list",",","use","the","list","of","tasks","directly","the","list","is","supposed","to","be","sorted","according","timeline","step",":","int","the","rolling","step","trunc_days",":","int","days","to","be","truncated","based","on","the","test","start","rolling_ext_days",":","int","sometimes","users","want","to","train","meta","models","for","a","longer","test","period","but","with","smaller","rolling","steps","for","more","task","samples",".","the","total","length","of","test","periods","will","be","`","step","+","rolling_ext_days","`","exp_name",":","Union","[","str",",","InternalData","]","Decide","what","meta_info","are","used","for","prediction",".","-","str",":","the","name","of","the","experiment","to","store","the","performance","of","data","-","InternalData",":","a","prepared","internal","data","segments",":","Union","[","Dict","[","Text",",","Tuple","]",",","float","]","the","segments","to","divide","data","both","left","and","right","if","segments","is","a","float",":","the","float","represents","the","percentage","of","data","for","training","hist_step_n",":","int","length","of","historical","steps","for","the","meta","infomation","task_mode",":","str","Please","refer","to","the","docs","of","MetaTask"]
code: def __init__(
      self,
      *,
      task_tpl: Union[dict, list],
      step: int,
      trunc_days: int = None,
      rolling_ext_days: int = 0,
      exp_name: Union[str, InternalData],
      segments: Union[Dict[Text, Tuple], float],
      hist_step_n: int = 10,
      task_mode: str = MetaTask.PROC_MODE_FULL,
      fill_method: str = "max",
  ):
      """
      A dataset for meta model.
  
      Parameters
      ----------
      task_tpl : Union[dict, list]
          Decide what tasks are used.
          - dict : the task template, the prepared task is generated with `step`, `trunc_days` and `RollingGen`
          - list : when list, use the list of tasks directly
                   the list is supposed to be sorted according timeline
      step : int
          the rolling step
      trunc_days: int
          days to be truncated based on the test start
      rolling_ext_days: int
          sometimes users want to train meta models for a longer test period but with smaller rolling steps for more task samples.
          the total length of test periods will be `step + rolling_ext_days`
  
      exp_name : Union[str, InternalData]
          Decide what meta_info are used for prediction.
          - str: the name of the experiment to store the performance of data
          - InternalData: a prepared internal data
      segments: Union[Dict[Text, Tuple], float]
          the segments to divide data
          both left and right
          if segments is a float:
              the float represents the percentage of data for training
      hist_step_n: int
          length of historical steps for the meta infomation
      task_mode : str
          Please refer to the docs of MetaTask
      """
      super().__init__(segments=segments)
      if isinstance(exp_name, InternalData):
          self.internal_data = exp_name
      else:
          self.internal_data = InternalData(
              task_tpl, step=step, exp_name=exp_name
          )
          self.internal_data.setup()
      self.task_tpl = deepcopy(
          task_tpl
      )  # FIXME: if the handler is shared, how to avoid the explosion of the memroy.
      self.trunc_days = trunc_days
      self.hist_step_n = hist_step_n
      self.step = step
  
      if isinstance(task_tpl, dict):
          rg = RollingGen(
              step=step,
              trunc_days=trunc_days,
              task_copy_func=deepcopy_basic_type,
          )  # NOTE: trunc_days is very important !!!!
          task_iter = rg(task_tpl)
          if rolling_ext_days > 0:
              self.ta = TimeAdjuster(future=True)
              for t in task_iter:
                  t["dataset"]["kwargs"]["segments"][
                      "test"
                  ] = self.ta.shift(
                      t["dataset"]["kwargs"]["segments"]["test"],
                      step=rolling_ext_days,
                      rtype=RollingGen.ROLL_EX,
                  )
          if task_mode == MetaTask.PROC_MODE_FULL:
              # Only pre initializing the task when full task is req
              # initializing handler and share it.
              init_task_handler(task_tpl)
      else:
          assert isinstance(task_tpl, list)
          task_iter = task_tpl
  
      self.task_list = []
      self.meta_task_l = []
      logger = get_module_logger("MetaDatasetDS")
      logger.info(
          f"Example task for training meta model: {task_iter[0]}"
      )
      for t in tqdm(task_iter, desc="creating meta tasks"):
          try:
              self.meta_task_l.append(
                  MetaTaskDS(
                      t,
                      meta_info=self._prepare_meta_ipt(t),
                      mode=task_mode,
                      fill_method=fill_method,
                  )
              )
              self.task_list.append(t)
          except ValueError as e:
              logger.warning(f"ValueError: {e}")
      assert (
          len(self.meta_task_l) > 0
      ), "No meta tasks found. Please check the data and setting"
  
imports: import pandas
  import numpy
  from copy import deepcopy
  from joblib import Parallel, delayed
  from typing import Dict, List, Union, Text, Tuple
  from qlib.data.dataset.utils import init_task_handler
  from qlib.data.dataset import DatasetH
  from qlib.contrib.torch import data_to_tensor
  from qlib.model.meta.task import MetaTask
  from qlib.model.meta.dataset import MetaTaskDataset
  from qlib.model.trainer import TrainerR
  from qlib.log import get_module_logger
  from qlib.utils import auto_filter_kwargs, get_date_by_shift, init_instance_by_config
  from qlib.utils.data import deepcopy_basic_type
  from qlib.workflow import R
  from qlib.workflow.task.gen import RollingGen, task_generator
  from qlib.workflow.task.utils import TimeAdjuster
  from tqdm.auto import tqdm
label_desc: Use image node_id 1 to create a new MetaDatasetDS object from inherited base classes: MetaTaskDataset with example: obj = MetaDatasetDS()