code: def add_data(tick_date, doc_type, stock_name_dict):
      pid = os.getpid()
  
      if doc_type not in DOC_TYPE:
          print("doc_type not in {}".format(DOC_TYPE))
          return
      try:
          begin_time = time.time()
          os.system(
              f"cp {DATABASE_PATH}/{tick_date + '_{}.tar.gz'.format(doc_type)} {DATA_PATH}/"
          )
  
          os.system(
              f"tar -xvzf {DATA_PATH}/{tick_date + '_{}.tar.gz'.format(doc_type)} -C {DATA_PATH}/ {tick_date + '_' + doc_type}/SH"
          )
          os.system(
              f"tar -xvzf {DATA_PATH}/{tick_date + '_{}.tar.gz'.format(doc_type)} -C {DATA_PATH}/ {tick_date + '_' + doc_type}/SZ"
          )
          os.system(f"chmod 777 {DATA_PATH}")
          os.system(
              f"chmod 777 {DATA_PATH}/{tick_date + '_' + doc_type}"
          )
          os.system(
              f"chmod 777 {DATA_PATH}/{tick_date + '_' + doc_type}/SH"
          )
          os.system(
              f"chmod 777 {DATA_PATH}/{tick_date + '_' + doc_type}/SZ"
          )
          os.system(
              f"chmod 777 {DATA_PATH}/{tick_date + '_' + doc_type}/SH/{tick_date}"
          )
          os.system(
              f"chmod 777 {DATA_PATH}/{tick_date + '_' + doc_type}/SZ/{tick_date}"
          )
  
          print("tick_date={}".format(tick_date))
  
          temp_data_path_sh = os.path.join(
              DATA_PATH, tick_date + "_" + doc_type, "SH", tick_date
          )
          temp_data_path_sz = os.path.join(
              DATA_PATH, tick_date + "_" + doc_type, "SZ", tick_date
          )
          is_files_exist = {
              "sh": os.path.exists(temp_data_path_sh),
              "sz": os.path.exists(temp_data_path_sz),
          }
  
          sz_files = (
              (
                  set(
                      [
                          i.split(".csv")[0]
                          for i in os.listdir(temp_data_path_sz)
                          if i[:2] == "30" or i[0] == "0"
                      ]
                  )
                  & set(stock_name_dict["SZ"])
              )
              if is_files_exist["sz"]
              else set()
          )
          sz_file_nums = len(sz_files) if is_files_exist["sz"] else 0
          sh_files = (
              (
                  set(
                      [
                          i.split(".csv")[0]
                          for i in os.listdir(temp_data_path_sh)
                          if i[0] == "6"
                      ]
                  )
                  & set(stock_name_dict["SH"])
              )
              if is_files_exist["sh"]
              else set()
          )
          sh_file_nums = len(sh_files) if is_files_exist["sh"] else 0
          print(
              "sz_file_nums:{}, sh_file_nums:{}".format(
                  sz_file_nums, sh_file_nums
              )
          )
  
          f = (
              DATA_INFO_PATH
              / "data_info_log_{}_{}".format(doc_type, tick_date)
          ).open("w+")
          f.write(
              "sz:{}, sh:{}, date:{}:".format(
                  sz_file_nums, sh_file_nums, tick_date
              )
              + "\n"
          )
          f.close()
  
          if sh_file_nums > 0:
              # write is not thread-safe, update may be thread-safe
              Parallel(n_jobs=N_JOBS)(
                  delayed(add_one_stock_daily_data_wrapper)(
                      os.path.join(temp_data_path_sh, name + ".csv"),
                      doc_type,
                      "SH",
                      index,
                      tick_date,
                  )
                  for index, name in enumerate(list(sh_files))
              )
          if sz_file_nums > 0:
              # write is not thread-safe, update may be thread-safe
              Parallel(n_jobs=N_JOBS)(
                  delayed(add_one_stock_daily_data_wrapper)(
                      os.path.join(temp_data_path_sz, name + ".csv"),
                      doc_type,
                      "SZ",
                      index,
                      tick_date,
                  )
                  for index, name in enumerate(list(sz_files))
              )
  
          os.system(
              f"rm -f {DATA_PATH}/{tick_date + '_{}.tar.gz'.format(doc_type)}"
          )
          os.system(f"rm -rf {DATA_PATH}/{tick_date + '_' + doc_type}")
          total_time = time.time() - begin_time
          f = (
              DATA_FINISH_INFO_PATH
              / "data_info_finish_log_{}_{}".format(doc_type, tick_date)
          ).open("w+")
          f.write(
              "finish: date:{}, consume_time:{}, end_time: {}".format(
                  tick_date, total_time, time.time()
              )
              + "\n"
          )
          f.close()
  
      except Exception as e:
          info = traceback.format_exc()
          print("date error:" + str(e))
          f = open(
              os.path.join(
                  LOG_FILE_PATH,
                  "temp_fail_{0}_{1}_{2}.txt".format(
                      pid, tick_date, doc_type
                  ),
              ),
              "a+",
          )
          f.write(
              "fail:"
              + str(tick_date)
              + "\n"
              + str(e)
              + "\n"
              + str(info)
              + "\n"
          )
          f.close()
  
imports: from datetime import date, datetime
  import os
  from pathlib import Path
  import random
  import shutil
  import time
  import traceback
  from arctic import Arctic, chunkstore
  import arctic
  from arctic import Arctic, CHUNK_STORE
  from arctic.chunkstore.chunkstore import CHUNK_SIZE
  import fire
  from joblib import Parallel, delayed, parallel
  import numpy
  import pandas
  from pandas import DataFrame
  from pandas.core.indexes.datetimes import date_range
  from pymongo.mongo_client import MongoClient
label_desc: Use image node_id 5 for calling a global function with example usage: add_data(tick_date, doc_type, stock_name_dict) without return types