优化SciPy自定义连续分布：高效预计算与缓存常数

from scipy.stats import rv_continuous

# 假设 N(a, b) 和 C(a, b) 是耗时函数
# f(x, a, b) 是非归一化PDF，F(x, a, b) 是其反导数

class Example_gen(rv_continuous):

    def _norm(self, a, b):
        """耗时的归一化常数计算占位符"""
        # 实际实现中会调用 N(a, b)
        print(f"Executing N({a}, {b}) (expensive)") # 模拟耗时操作
        import time; time.sleep(0.1)
        return a + b + 0.5 # 占位符，实际应为 N(a,b)

    def _C(self, a, b):
        """耗时的积分常数计算占位符"""
        # 实际实现中会调用 C(a, b)
        print(f"Executing C({a}, {b}) (expensive)") # 模拟耗时操作
        import time; time.sleep(0.1)
        return a * b - 0.1 # 占位符，实际应为 C(a,b)

    def _pdf(self, x, a, b):
        # 假设 f(x, a, b) 是一个简单的函数
        return (x * a + b) / self._norm(a, b)

    def _cdf(self, x, a, b):
        # 假设 F(x, a, b) 是一个简单的函数
        return (0.5 * x**2 * a + x * b + self._C(a, b)) / self._norm(a, b)

Example = Example_gen()

# 示例使用：每次调用都会重新计算常数
# rv = Example(a=1, b=2)
# rv.pdf(0.5) # 会触发 _norm 和 _C 的计算
# rv.cdf(0.5) # 再次触发 _norm 和 _C 的计算

from scipy.stats import rv_continuous
import time

# 模拟耗时函数
def N_expensive(a, b):
    """模拟耗时归一化常数计算"""
    time.sleep(0.1) # 模拟计算延迟
    return a + b + 0.5

def C_expensive(a, b):
    """模拟耗时积分常数计算"""
    time.sleep(0.1) # 模拟计算延迟
    return a * b - 0.1

class Example_gen_cached(rv_continuous):

    # 定义类级别缓存字典
    _n_cache = {}
    _C_cache = {}

    def _norm(self, a, b):
        """带有缓存的归一化常数计算"""
        # 使用参数元组作为缓存键。对浮点数进行四舍五入以处理精度问题。
        key = (round(a, 5), round(b, 5))

        # 尝试从缓存中获取值
        v = self._n_cache.get(key)

        if v is None:
            # 如果缓存中没有，则进行昂贵计算并存储
            print(f"  Calculating N({a}, {b})...")
            v = N_expensive(a, b) # 调用实际的耗时函数
            self._n_cache[key] = v
        else:
            print(f"  Retrieving N({a}, {b}) from cache.")
        return v

    def _C(self, a, b):
        """带有缓存的积分常数计算"""
        key = (round(a, 5), round(b, 5))
        v = self._C_cache.get(key)

        if v is None:
            print(f"  Calculating C({a}, {b})...")
            v = C_expensive(a, b) # 调用实际的耗时函数
            self._C_cache[key] = v
        else:
            print(f"  Retrieving C({a}, {b}) from cache.")
        return v

    def _pdf(self, x, a, b):
        # 假设 f(x, a, b) 是一个简单的函数
        return (x * a + b) / self._norm(a, b)

    def _cdf(self, x, a, b):
        # 假设 F(x, a, b) 是一个简单的函数
        return (0.5 * x**2 * a + x * b + self._C(a, b)) / self._norm(a, b)

Example_cached = Example_gen_cached()

# 示例使用：
print("--- 首次计算 (a=1, b=2) ---")
rv1 = Example_cached(a=1, b=2)
_ = rv1.pdf(0.5)
_ = rv1.cdf(0.5)

print("\n--- 再次计算 (a=1, b=2) ---")
rv2 = Example_cached(a=1, b=2) # 即使是新实例，参数相同也会使用缓存
_ = rv2.pdf(0.7)
_ = rv2.cdf(0.7)

print("\n--- 改变参数 (a=1.1, b=2.2) ---")
rv3 = Example_cached(a=1.1, b=2.2)
_ = rv3.pdf(0.3)
_ = rv3.cdf(0.3)