You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何基于Archimedean Copula由U与theta推导并发变量V?

问题:基于Archimedean Copula和EVA获取并发变量值的高效实现

我有两组变量,其中一组是另一组的函数。先使用copulas包拟合Copula,再对因变量做极值分析(Extreme Value Analysis, EVA)。最终希望通过因变量对应值U和Archimedean Copula的theta参数,获取并发的自变量对应值V,但这一步遇到困难。试过ChatGPT的方案,效率较低,寻求高效解决思路或帮助。

def check_and_coupla_fit(df, ice_dist, sped_dist):
    """
    check the copula and return the 
    """

    ice = ice_dist.cdf(df.jones)
    sped = sped_dist.cdf(df.sped)


    tau_values = []
    rho_values = []
    theta_values = []
    for copula_type in ['clayton', 'frank', 'gumbel']:
        copula = getattr(copulas, copula_type.capitalize() + 'Copula')
        copula_instance = copula()
        copula_instance.fit([ice, sped])
        
        # Compute tau and rho
        u, v = copula_instance.sample(1000)
        tau, rho = compute_tau_and_rho(u, v)
        
        tau_values.append(tau)
        rho_values.append(rho)
        theta_values.append(copula_instance.compute_theta())
        
    # Determine the best copula based on tau and rho
    best_idx = np.argmax(np.array(tau_values) + np.array(rho_values))
    
    # Create the result array
    result = np.full((3, 1), np.nan)
    result[best_idx] = theta_values[best_idx]
    
    return result

def compute_tau_and_rho(u, v):
    """
    Compute Kendall's tau and Spearman's correlation coefficient.
    """
    tau, _ = copulas.stats.kendall_tau(u, v)
    rho, _ = spearmanr(u, v)
    return tau, rho

def get_stat(df):
    """
    gets distribution parameters for both ice accretion and wind speed.
    """
    
    # Extract the year from the 'valid' column
    df['year'] = df['valid'].dt.year

    # Group by year and get the maximum annual ice accretion values from the dataset
    max_ice_values = df.groupby(['year'])['jones'].max().reset_index()

    df_max_with_sped = pd.merge(max_ice_values, df,
                                on = ['jones', 'year'],
                                how = 'inner')[['jones', 'sped']]
    
    c_sped,loc_sped, scale_sped = weibull_min.fit(df_max_with_sped.sped)
    c_ice, loc_ice, scale_ice = genextreme.fit(df_max_with_sped.jones)
    
    return c_ice, loc_ice, scale_ice, c_sped,loc_sped, scale_sped
def inv_gumbel(u, theta):
    """
    Inverse of the Gumbel copula for given u and theta.
    """
    return np.exp(-(-np.log(u))**(1/theta))


def inv_frank(u, theta):
    """
    Inverse of the Frank copula for given u and theta.
    """
    term = np.log(1 + (np.exp(-theta*u) - 1) * (np.exp(-theta) - 1) / (np.exp(-theta*u) - np.exp(-theta) - 1))
    return -1/theta * np.log(1 - term)


def inv_clayton(u, theta):
    """
    Inverse of the Clayton copula for given u and theta.
    """
    return (u**(-theta) + 1)**(-1/theta)
def get_concurrent_wind(arr, U):
    """
    transforms U to V
    arr: vector of thetas
    U: is standardized ice value
    """
    V = 0
    if np.isnan(arr[0]):
        # Clayton copula
        V = inv_clayton(U, arr[0])
        
    elif np.isnan(arr[1]):
        # Frank copula
        V = inv_frank(U, arr[1])
        
    elif np.isnan(arr[2]):
        # Gumbel copula
        V = inv_gumbel(U, arr[2])
        
    else:
        raise ValueError('Size of arr should be 3, not more!');

    return V
stations = df.station.unique()
# unique stations

Results = np.array([])
counter_rows = 0
proceeed_statistical_analysis = True
for counter_rows, station in bar(enumerate(stations), total=203):
    
    station_data = df.loc[df.station == station].copy()
    # EVA for each station
    station_data = station_data[['valid','jones']]
    # only accounts for conditions in which EVA can be implemented! 
    try:
        station_data.set_index('valid',
                               inplace = True)
        
        station_data = station_data.squeeze()
        
        model = EVA(station_data)
        
        model.get_extremes(
            method="BM",
            extremes_type="high",
            block_size="365.2425D",
            errors="ignore",
        )

        # Get max values for different return periods (50 and 500 yrs).
        model.fit_model()
        summary = model.get_summary(
        return_period=[50, 500],
        alpha=0.95,
        n_samples=1_000)
        summary = summary.iloc[:, 0]
        
        jones_50 = summary[50]
        jones_500 = summary[500]    

        
                                       
    except:
        jones_50 = jones_500 = sped_50 = sped_500 = np.NaN
        arr = np.zeros([1, 3])
        proceeed_statistical_analysis = False

    if proceeed_statistical_analysis:
        # Obtaining concurrent wind speed for 50 and 500 years return periods.
        station_data = df.loc[df.station == station].copy()
        # EVA for each station
        station_data = station_data[['valid','jones']]
        
        c_ice, loc_ice, scale_ice, c_sped,loc_sped, scale_sped = get_stat(data)
    
        ice_dist = genextreme(c = c_ice,
                              loc = loc_ice,
                              scale = scale_ice)
    
        sped_dist = weibull_min(c = c_sped,
                                loc = loc_sped,
                                scale = scale_sped)
        
        # acquire theta and the corresponding copula function for the observations in the ith station
        arr = check_and_coupla_fit(df = station_data,
                                   ice_dist = ice_dist,
                                   sped_dist = sped_dist)
    
        # computes concurrent 50 years wind speed
        sped_50 = get_concurrent_wind(arr, sped_dist.cdf(jones_50))
    
        # computes concurrent 500 years wind speed
        sped_500 = get_concurrent_wind(arr, sped_dist.cdf(jones_500))

    Temp1 = df.loc[df.station == station].copy()
    lon, lat,elev = Temp1.iloc[0,2], Temp1.iloc[0,3], Temp1.iloc[0,4]

    Temp = np.zeros(shape = [1, 10])
    Temp[0, 0] = lon
    Temp[0, 1] = lat
    Temp[0, 2] = elev
    Temp[0, 3] = jones_50
    Temp[0, 4] = jones_500
    Temp[0, 5] = sped_50
    Temp[0, 6] = sped_500
    Temp[0, 7:] = arr
    
    if counter_rows == 0:
        Results = Temp.copy()
    elif counter_rows >0:
        
        Results = np.concatenate([Results, Temp],
                                axis=0)
    else:
        raise ValueError('Something happened!');

Results = pd.DataFrame(Results,
                      columns=['lon',
                               'lat',
                               'elev',
                               'jones_50',
                               'jones_500',
                               'sped_50',
                               'sped_500',
                               'Clayton',
                               'Frank',
                               'Gumbel'])

Results = Results.loc[~np.isnan(Results.jones_50)]

更新8-01
尝试多种方法后明白,仅靠theta和U值无法直接得到V。不过从拟合的Copula函数(如Clayton Copula)生成足够多的样本,大概率能得到接近目标值的结果;若定义为优化问题,经若干迭代也能得到合理结果,耗时取决于计算机内存。


内容的提问来源于stack exchange,提问作者matt

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.03 07:34:50