求Theano/Aesara带滞后的Adstock函数向量化实现方案

云原生阿岚

2026-8-19

带滞后效应的Adstock函数向量化实现需求

目标函数定义

需要实现的带滞后效应的Adstock函数公式为：
$$y_t = \frac{\sum_{l=0}^{L-1} D^{(l-P)2} x_{t-l}}{\sum_{l=0}^{L-1} D^{(l-P)2}}$$
其中：

$x_t$：原始输入序列
$L$：滞后窗口长度
$P$：效应峰值的延迟步数
$D$：衰减系数（$0 < D < 1$）

当前非向量化实现（速度极慢）

def apply_adstock_with_lag(x, L, P, D):
    """
    params:
    x: original array
    L: length
    P: peak, delay in effect
    D: decay, retain
    """
    x = np.append(np.zeros(L - 1), x)

    weights = [0 for _ in range(L)]
    for l in range(L):
        weight = D ** ((l - P) ** 2)
        weights[L - 1 - l] = weight
    weights = np.array(weights)
    adstocked_x = []
    for i in range(L - 1, len(x)):
        x_array = x[i - L + 1:i + 1]
        xi = sum(x_array * weights) / sum(weights)
        adstocked_x.append(xi)
    adstocked_x = tt.as_tensor_variable(adstocked_x)
    return adstocked_x

参考的简化版Adstock向量化实现（速度更快）

参考的几何衰减Adstock公式为：
$$y_t = x_t + \theta y_{t-1}$$
对应的Theano/PyMC3向量化实现代码：

def adstock_geometric_theano_pymc3(x, theta):
    x = tt.as_tensor_variable(x)

    def adstock_geometric_recurrence_theano(index, input_x, decay_x, theta):
        return tt.set_subtensor(decay_x[index], tt.sum(input_x + theta * decay_x[index - 1]))

    len_observed = x.shape[0]

    x_decayed = tt.zeros_like(x)
    x_decayed = tt.set_subtensor(x_decayed[0], x[0])

    output, _ = theano.scan(
        fn=adstock_geometric_recurrence_theano,
        sequences=[tt.arange(1, len_observed), x[1:len_observed]],
        outputs_info=x_decayed,
        non_sequences=theta,
        n_steps=len_observed - 1
    )

    return output[-1]