有没有办法在Python中向量化这个逻辑？

Question

我正在研究一个金融模拟问题，我有一段非常简单的代码，它利用 for 循环来获得结果。 我认为应该可以对其进行矢量化，但我也认为它可能需要多个矢量运算和一些巧妙的数学来实现矢量化，而我尝试的所有解决方案都未能产生预期的结果。

# for size: 10000000 3.8 seconds
import numpy as np
negative_threshold = -50
positive_threshold = 30

LOWER_BOND_RISK = -10000
UPPER_BOUND_RISK = 200000
SIZE_TO_RUN = 100000

no_hedges_flows_vector = np.random.randint(LOWER_BOND_RISK, UPPER_BOUND_RISK, size=SIZE_TO_RUN)
no_hedges_flows_vector

hedges_vector = np.zeros(len(no_hedges_flows_vector))

position = 0 # We start at 0 risk
for i,no_hedges_flow in enumerate(no_hedges_flows_vector):
    previous_position=position
    position += no_hedges_flow
    if ((position < negative_threshold)| (position > positive_threshold)): # we hedge
        hedges_vector[i] = -position 
        position = 0


hedged_positions_vector = hedges_vector.cumsum()  + no_hedges_flows_vector.cumsum()

assert np.all(hedged_positions_vector >= negative_threshold)
assert np.all(hedged_positions_vector <= positive_threshold)
assert np.all(hedges_vector >= -(UPPER_BOUND_RISK + positive_threshold))
assert np.all(hedges_vector <= -(LOWER_BOND_RISK + negative_threshold))

上面的代码的本质是： 给定货币中的风险变化向量，在每次迭代中它都会决定是否对冲，其中“对冲”仅意味着向“hedges_vector”添加相等且相反的风险变化，并将风险减少“对冲”金额。对冲的决定取决于当前头寸（作为迄今为止所有风险变化（包括对冲）的总和）是否在可接受的范围内

我自己尝试了一些案例，但所有失败的断言和回想起来都是逻辑上错误的结构。我也尝试了 ChatGPT 的运气，但这根本没有帮助（失败的断言并尝试更改这些而不是代码，即使被要求不要这样做。https://chat.openai.com/share/248403f1 -12a6-45d1-ab1d-904a0344490b）

上面的代码并不是超级慢，但是需要矢量化版本，因为这对性能至关重要

Answer 1

在这种情况下，我会使用 numba 来加快处理速度：

import numba


@numba.njit
def func_numba(no_hedges_flows_vector):
    hedges_vector = np.zeros_like(no_hedges_flows_vector)

    position = 0  # We start at 0 risk
    for i, no_hedges_flow in enumerate(no_hedges_flows_vector):
        position += no_hedges_flow
        if (position < negative_threshold) | (
            position > positive_threshold
        ):  # we hedge
            hedges_vector[i] = -position
            position = 0

    hedged_positions_vector = hedges_vector.cumsum() + no_hedges_flows_vector.cumsum()
    return hedged_positions_vector

---

基准：

from timeit import timeit

import numba
import numpy as np

negative_threshold = -50
positive_threshold = 30

LOWER_BOND_RISK = -10000
UPPER_BOUND_RISK = 200000
SIZE_TO_RUN = 10000000

np.random.seed(0)

no_hedges_flows_vector = np.random.randint(
    LOWER_BOND_RISK, UPPER_BOUND_RISK, size=SIZE_TO_RUN
)


def func_normal(no_hedges_flows_vector):
    hedges_vector = np.zeros(len(no_hedges_flows_vector))

    position = 0  # We start at 0 risk
    for i, no_hedges_flow in enumerate(no_hedges_flows_vector):
        position += no_hedges_flow
        if (position < negative_threshold) | (
            position > positive_threshold
        ):  # we hedge
            hedges_vector[i] = -position
            position = 0

    hedged_positions_vector = hedges_vector.cumsum() + no_hedges_flows_vector.cumsum()
    return hedged_positions_vector


import numba


@numba.njit
def func_numba(no_hedges_flows_vector):
    hedges_vector = np.zeros_like(no_hedges_flows_vector)

    position = 0  # We start at 0 risk
    for i, no_hedges_flow in enumerate(no_hedges_flows_vector):
        position += no_hedges_flow
        if (position < negative_threshold) | (
            position > positive_threshold
        ):  # we hedge
            hedges_vector[i] = -position
            position = 0

    hedged_positions_vector = hedges_vector.cumsum() + no_hedges_flows_vector.cumsum()
    return hedged_positions_vector


r1 = func_normal(no_hedges_flows_vector)
r2 = func_numba(no_hedges_flows_vector)

assert np.allclose(r1, r2)

assert np.all(r1 >= negative_threshold)
assert np.all(r2 <= positive_threshold)

t1 = timeit("func_normal(no_hedges_flows_vector)", number=1, globals=globals())
t2 = timeit("func_numba(no_hedges_flows_vector)", number=1, globals=globals())

print(t1)
print(t2)

在我的机器上打印（AMD 5700x）：

2.648123851045966
0.1116715909447521

加速约 23 倍