liuyu/qlib
1
0
Fork 0
mirror of https://github.com/microsoft/qlib.git synced 2026-07-11 14:56:55 +08:00
Code Activity

replace multi processing with joblib (#477)

Browse Source 
* replace multi processing with joblib

* update class Parallel and data.py

* update class Parallel and data.py

* update class Parallel and data.py

* update class Parallel and data.py

* update class Parallel and data.py

* update class Parallel and data.py

* update class Parallel and data.py

* update class Parallel and data.py

* Fix Parallel support for maxtasksperchild

Co-authored-by: wangw <1666490690@qq.com>
Co-authored-by: zhupr <zhu.pengrong@foxmail.com>
This commit is contained in:
you-n-g
2021-09-14 01:16:03 +08:00
committed by GitHub
parent 6203e4c09e
commit 163e3c6266
4 changed files with 86 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
qlib/utils/paral.py
View File

@@ -1,8 +1,17 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
from joblib import Parallel, delayed
import pandas as pd
from joblib import Parallel, delayed
from joblib._parallel_backends import MultiprocessingBackend
class ParallelExt(Parallel):
def __init__(self, *args, **kwargs):
maxtasksperchild = kwargs.pop("maxtasksperchild", None)
super(ParallelExt, self).__init__(*args, **kwargs)
if isinstance(self._backend, MultiprocessingBackend):
self._backend_args["maxtasksperchild"] = maxtasksperchild
def datetime_groupby_apply(df, apply_func, axis=0, level="datetime", resample_rule="M", n_jobs=-1, skip_group=False):
@@ -31,7 +40,7 @@ def datetime_groupby_apply(df, apply_func, axis=0, level="datetime", resample_ru
return df.groupby(axis=axis, level=level).apply(apply_func)
if n_jobs != 1:
dfs = Parallel(n_jobs=n_jobs)(
dfs = ParallelExt(n_jobs=n_jobs)(
delayed(_naive_group_apply)(sub_df) for idx, sub_df in df.resample(resample_rule, axis=axis, level=level)
)
return pd.concat(dfs, axis=axis).sort_index()