<jupyter_start><jupyter_text>20.3<jupyter_code>def barrier_tourch(r, width=.5):
    t = dict()
    p = np.log((1 + r).cumprod(axis=0))
    for j in range(r.shape[1]):
        for i in range(r.shape[0]):
            if np.abs(p[i][j]) >= width:
                t[j] = i
                continue
    return t
%%time

r = np.random.normal(0, 0.01, size=(1000, 10000))
num_threads = 24
parts = lin_parts(len(r), num_threads)
print(parts)
jobs = []
for i in range(1, len(parts)):
    jobs.append(r[:, parts[i-1]:parts[i]])
pool = mp.Pool(processes=num_threads)
outputs = pool.imap_unordered(barrier_tourch, jobs)
out = []
for out_ in outputs:
    out.append(out_)
pool.close()
pool.join()
%%time

r = np.random.normal(0, 0.01, size=(1000, 10000))
num_threads = 24
parts = lin_parts(len(r), num_threads)
parts = nested_parts(len(r), num_threads)
print(parts)
jobs = []
for i in range(1, len(parts)):
    jobs.append(r[:, parts[i-1]:parts[i]])
pool = mp.Pool(processes=num_threads)
outputs = pool.imap_unordered(barrier_tourch, jobs)
out = []
for out_ in outputs:
    out.append(out_)
pool.close()
pool.join()<jupyter_output>[   0  204  288  353  408  456  500  540  577  612  645  677  707  736
  764  790  816  842  866  890  913  935  957  979 1000]
CPU times: user 413 ms, sys: 62.2 ms, total: 476 ms
Wall time: 870 ms<jupyter_text>20.5<jupyter_code>import time
from datetime import datetime
import sys

def expand_call(kwargs):
    func = kwargs['func']
    del kwargs['func']
    out = func(**kwargs)
    return out

def report_progress(job_idx, num_jobs, time0, task):
    msg = [float(job_idx) / num_jobs, (time.time() - time0) / 60.]
    msg.append(msg[1] * (1/msg[0] - 1))
    time_stamp = str(datetime.fromtimestamp(time.time()))
    msg_ = time_stamp + ' ' +str(round(msg[0]*100, 2))+ '% ' + task + ' done after ' + \
        str(round(msg[1], 2)) + ' minutes. Remaining ' + str(round(msg[2],2)) + ' minutes.'
    if job_idx < num_jobs:
        sys.stderr.write(msg_ + '\r')
    else:
        sys.stderr.write(msg_ + '\n')

def process_jobs(jobs, task=None, num_threads=24):
    if task is None:
        task = jobs[0]['func'].__name__
    pool = mp.Pool(processes=num_threads)
    outputs = pool.imap_unordered(expand_call, jobs)
    out = []
    time0 = time.time()
    # Execute programs here
    for i, out_ in enumerate(outputs, 1):
        out.append(out_)
        report_progress(i, len(jobs), time0, task)
    pool.close()
    pool.join()
    return out

def mp_pandas_obj(func, pd_obj, num_threads=24, mp_batches=1, lin_mols=True, descend=False, **kwargs):
    if lin_mols:
        parts = lin_parts(len(pd_obj[1]), num_threads * mp_batches)
    else:
        parts = nested_parts(len(pd_obj[1]), num_threads * mp_batches, descend)
    jobs = []
    for i in range(1, len(parts)):
        job = {pd_obj[0]: pd_obj[1][parts[i-1]: parts[i]], 'func': func}
        job.update(kwargs)
        jobs.append(job)
    if num_threads == 1:
        out = process_jobs(jobs)
    else:
        out = process_jobs(jobs, num_threads=num_threads)
        
    if isinstance(out[0], pd.Series):
        df0 = pd.Series()
    elif isinstance(out[0], pd.DataFrame):
        df0 = pd.DataFrame()
    else:
        return out
    
    for i in out:
        df0 = df0.append(i)
    df0 = df0.sort_index()
    return df0
%%time

r = np.random.normal(0, 0.01, size=(1000, 10000))
num_threads = 24
pd_obj =  ["r", r]
out = mp_pandas_obj(barrier_tourch, pd_obj, num_threads)
%%time

r = np.random.normal(0, 0.01, size=(1000, 10000))
num_threads = 24
pd_obj =  ["r", r]
out = mp_pandas_obj(barrier_tourch, pd_obj, num_threads=1)<jupyter_output>CPU times: user 344 ms, sys: 147 ms, total: 491 ms
Wall time: 12.6 s
{'fraction_non_alphanumeric': 0.15297297297297296, 'fraction_numerical': 0.05837837837837838, 'mean_word_length': 3.1819209039548024, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '165800', 'license': 'MIT', 'n_tokens_mistral': 1564, 'n_tokens_neox': 1416, 'n_words': 333, 'path': 'finance_ml/examples/miscellaneous/Ch 20.ipynb', 'repo_name': 'mikimaus78/ml_monorepo', 'size': '3700'}