[Python] Can anybody tell me what is wrong with this parallelization using...

I am trying to simulate in parallel and it's the first time I use parallelization in Python.

The following code imports a few libraries and defines auxiliary functions. run_simulation is the core function which I wish to pass to each worker to speed-up simulations. The code is executable.

import numpy as np
from scipy.stats import norm
import pandas as pd
import scipy as sp
import sys
from concurrent.futures import ProcessPoolExecutor

def BSFormulaCall(S0, K, T, r, sigma):
x = np.log(S0 / K) + r * T
sig = sigma * np.sqrt(T)
d1 = x / sig + sig / 2
d2 = d1 - sig
pv = np.exp(-r * T)
price = S0 * norm.cdf(d1) - pv * K * norm.cdf(d2)
return price

def BSFormulaPut(S0, K, T, r, sigma):
x = np.log(S0 / K) + r * T
sig = sigma * np.sqrt(T)
d1 = x / sig + sig / 2
d2 = d1 - sig
pv = np.exp(- r * T)
price = S0 * norm.cdf(d1) - pv * K * norm.cdf(d2) + pv * K - S0
return price

def BSImpliedVolCall(S0, K, T, r, C):
nK = 1 if np.ndim(K) == 0 else len(K) # Number of strike prices
sigmaL = np.full(nK, 1e-10) # Lower bound for volatility
CL = BSFormulaCall(S0, K, T, r, sigmaL) # Call option prices for low volatility
sigmaH = np.full(nK, 500) # Upper bound for volatility
CH = BSFormulaCall(S0, K, T, r, sigmaH) # Call option prices for high volatility
while np.mean(sigmaH - sigmaL) > 1e-10:
sigma = (sigmaL + sigmaH) / 2
CM = BSFormulaCall(S0, K, T, r, sigma) # Calculate option price for current sigma
CL = CL + (CM < C) * (CM - CL) # Update lower bound for price
sigmaL = sigmaL + (CM < C) * (sigma - sigmaL) # Adjust lower bound volatility
CH = CH + (CM >= C) * (CM - CH) # Update upper bound for price
sigmaH = sigmaH + (CM >= C) * (sigma - sigmaH) # Adjust upper bound volatility
if nK == 1:
sigma = sigma[0]
return sigma

def BSImpliedVolPut(S0, K, T, r, P):
nK = 1 if np.ndim(K) == 0 else len(K) # Number of strike prices
sigmaL = np.full(nK, 1e-10) # Lower bound for volatility
PL = BSFormulaPut(S0, K, T, r, sigmaL) # Put option prices for low volatility
sigmaH = np.full(nK, 500) # Upper bound for volatility
PH = BSFormulaPut(S0, K, T, r, sigmaH) # Put option prices for high volatility
while np.mean(sigmaH - sigmaL) > 1e-10:
sigma = (sigmaL + sigmaH) / 2
PM = BSFormulaPut(S0, K, T, r, sigma) # Calculate option price for current sigma
PL = PL + (PM < P) * (PM - PL) # Update lower bound for price
sigmaL = sigmaL + (PM < P) * (sigma - sigmaL) # Adjust lower bound volatility
PH = PH + (PM >= P) * (PM - PH) # Update upper bound for price
sigmaH = sigmaH + (PM >= P) * (sigma - sigmaH) # Adjust upper bound volatility
if nK == 1:
sigma = sigma[0]
return sigma

def f_ou(k, t, n = 1):
vWN = np.random.normal(loc = 0, scale = 1, size = int(1e6))
return np.sqrt((1 - np.exp(- 2 * k * t)) / (2 * k)) * vWN


def f_vix_mb(w1, w2, lam, k, t, ksi0, n_int = 300, n_samp = int(1e6)):
d_vix = 30/365.25
grid = np.linspace(start = t + d_vix/n_int, stop = t + d_vix, num = n_int)

# randomness
vOU = f_ou(k, t, n_samp)
varOU = (1 - np.exp(- 2 * k * t))/(2 * k)

# computing instantaneous forward variance
def f_gridpoint(pp):
return ksi0 * (
lam *
np.exp(w1 * np.exp(- k * (grid[pp] - t)) * vOU -
(w1 ** 2) / 2 * np.exp(- 2 * k * (grid[pp] - t)) * varOU) +
(1 - lam) *
np.exp(w2 * np.exp(- k * (grid[pp] - t)) * vOU -
(w2 ** 2) / 2 * np.exp(- 2 * k * (grid[pp] - t)) * varOU)
)
mKsi = np.vstack(list(map(f_gridpoint, range(n_int))))
del vOU
# integrating to come up with VIX
vVIX2 = np.mean(mKsi, axis = 0)
# return mKsi
del mKsi
return np.sqrt(vVIX2)

def f_nsfwdv_rnd(t):
val = - 1
while val <= 0:

r_b0 = [0.025, 0.160]
r_b01 = [0.005, 0.13]
r_b2 = [- 0.15, 0.25]
r_t1 = [0.001, 1.35]
r_t2 = [0.001, 0.125]

b0 = np.random.uniform(low = r_b0[0], high = r_b0[1], size = 1)[0]
b01 = np.random.uniform(low = r_b01[0], high = r_b01[1], size = 1)[0]
b2 = np.random.uniform(low = r_b2[0], high = r_b2[1], size = 1)[0]
t1 = np.random.uniform(low = r_t1[0], high = r_t1[1], size = 1)[0]
t2 = np.random.uniform(low = r_t2[0], high = r_t2[1], size = 1)[0]
b1 = b01 - b0
val = b0 + b1 * np.exp(- t / t1) + b2 * np.exp(t / t2) * np.exp(- t / t2)
return val, b0, b1, b2, t1, t2

def run_simulation(nk, r_w, r_k, r_l, vXM, vYM, vXS, vYS):
w1 = np.random.uniform(low = r_w[0], high = r_w[1], size = 1)[0]
w2 = np.random.uniform(low = r_w[0], high = r_w[1], size = 1)[0]
k = np.random.uniform(low = r_k[0], high = r_k[1], size = 1)[0]
lam = np.random.uniform(low = r_l[0], high = r_l[1], size = 1)[0]
ttm = np.random.choice(a = vXM, size = 1, p = vYM / np.sum(vYM))[0]/365.25
ksi0, b0, b1, b2, t1, t2 = f_nsfwdv_rnd(ttm)
vK = np.random.choice(a = vXS, size = nk, p = vYS / np.sum(vYS))
vType = np.where(vK <= 1, 'put', 'call')
vVIX = f_vix_mb(w1, w2, lam, k, ttm, ksi0, n_samp = 10, n_int = 10) * 100
fwd = np.mean(vVIX)
vStrikes = vK * np.mean(vVIX)

vModelVol = np.zeros(nk)
for i in range(nk):
if vType == "put":
price = np.mean(np.maximum(0, vStrikes - vVIX))
vModelVol = BSImpliedVolPut(fwd, vStrikes, ttm, 0, price)
else:
price = np.mean(np.maximum(0, vVIX - vStrikes))
vModelVol = BSImpliedVolCall(fwd, vStrikes, ttm, 0, price)

vRow = [b0, b1, b2, t1, t2, w1, w2, k, lam, ttm, vK, vModelVol]
vRowFlat = [item for sublist in vRow for item in (sublist.tolist() if isinstance(sublist, np.ndarray) else [sublist])]

return vRowFlat

if __name__ == "__main__":
moneyness = np.random.uniform(0.15, 14.54, 1246)
n = np.random.lognormal(mean=np.log(100), sigma=1.5, size=1246).astype(int)
tStr = pd.DataFrame({'moneyness': moneyness, 'n': n})
vEStr = np.repeat(tStr['moneyness'], tStr['n'])
oKDES = sp.stats.gaussian_kde(vEStr)
ttm = np.random.randint(1, 315, 260)
n = np.random.lognormal(mean=np.log(50), sigma=1.5, size=260).astype(int)
tMat = pd.DataFrame({'ttm': ttm, 'n': n})
vEMat = np.repeat(tMat['ttm'], tMat['n'])
oKDEM = sp.stats.gaussian_kde(vEMat)
nk = 5
iBatchSize = 2
iM = 10
r_w = [0, 10]
r_k = [0, 10]
r_l = [0, 1]
vXS = np.linspace(min(vEStr), max(vEStr), 1000)
vYS = oKDES(vXS)
vXM = np.linspace(min(vEMat), max(vEMat), 1000)
vYM = oKDEM(vXM)
# ProcessPoolExecutor for parallel execution
with ProcessPoolExecutor(max_workers = 2) as executor:
futures = []
future = executor.submit(run_simulation, nk, r_w, r_k,
r_l, vXM, vYM, vXS, vYS)
future.result()


Upon running the code, I get:

BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending.


Process SpawnProcess-1:
Traceback (most recent call last):
File "multiprocessing\process.py", line 315, in _bootstrap
File "multiprocessing\process.py", line 108, in run
File "concurrent\futures\process.py", line 233, in _process_worker
File "multiprocessing\queues.py", line 116, in get
AttributeError: Can't get attribute 'run_simulation' on <module '__main__' (built-in)>

Continue reading...