inference/debias_methods.py at main · NoaVelner/inference · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
# import numpy as np
# import pandas as pd
# from ci_params_and_config import Methods
# from dists import TruncGauss1D
# from main import compute_single_n, true_value_for_stats, _build_algs_from_methods, rows_from_single_n_results, \
#     append_rows, init_log_file
#
#
# def run_debias_experiment(num_experiments, n_values, distribution, algs, global_params,
#                     log_path=None, overwrite=False, sort=None):
#     init_log_file(log_path, overwrite=overwrite)
#
#     df = []
#     for n in n_values:
#         print(f"n: {n}")
#         results = compute_single_n(n, num_experiments, distribution, algs, global_params, sort=sort)
#         rows = rows_from_single_n_results(results, global_params, n)
#         append_rows(df, rows, log_path=log_path)
#
#     return pd.DataFrame(df)
#
#
# def set_experiment(eps1, eps2, mT1, mT2, num_experiments, n_values, dist, stats, significance, record_mean=False):
#     true_values = true_value_for_stats(dist, stats, n_values)
#     alg1 = _build_algs_from_methods(methods=[Methods.PRIVSUB], stats=stats, sbs_mT_func=mT1,
#                                     sbs_description="")
#     alg2 = _build_algs_from_methods(methods=[Methods.PRIVSUB], stats=stats, sbs_mT_func=mT2,
#                                     sbs_description="")
#
#     g_params1 =  {"domain": dist.get_bounds(), "true_values": true_values, "epsilon": eps1, "delta": 1e-5,
#                      "significance": significance}
#     g_params2 =  {"domain": dist.get_bounds(), "true_values": true_values, "epsilon": eps2, "delta": 1e-5,
#                         "significance": significance}
#     df1 = run_debias_experiment(num_experiments, n_values, dist, alg1, g_params1, record_mean=record_mean)
#     df2 = run_debias_experiment(num_experiments, n_values, dist, alg2, g_params2, record_mean=record_mean)
#     return df1, df2
#
#
# def extract_ci_info(df, method_name, n=None):
#     res = df[df['method'] == method_name]['quantile_approx']
#     if n is not None:
#         res = res[df['n'] == n].values[0]
#     return res
#
#
# if __name__ == '__main__':
#     np.random.seed(50)
#
#     n_values = np.linspace(500, 2000, 10, dtype=int)
#     n_values = [1000]
#     dist = TruncGauss1D(trunc_left=-6, trunc_right=4, sigma=2)
#     eps1, eps2 = 100, 0.5
#     num_experiments = 500
#
#     stats = "median"
#     significance = 0.01
#     mT1 = lambda n: (int(n ** (2 / 3)), 100)
#     mT2 = lambda n: (int(n ** (2 / 3)), 100)
#     # mT2 = lambda n: (int(n ** (1 / 3)), 50)
#
#     df1, df2 = set_experiment(eps1, eps2, mT1, mT2, num_experiments, n_values, dist, stats, significance)
#     df1_ci = extract_ci_info(df1, method_name="PrivSub (ours)", n=1000)
#     df2_ci = extract_ci_info(df2, method_name="PrivSub (ours)", n=1000)
#
#     # coverage for df1_ci:
#     coverage = 0
#     for row in df1_ci:
#         ci_lower, ci_upper = row
#         if ci_lower <= dist.median() <= ci_upper:
#             coverage += 1
#     coverage1 =coverage / len(df1_ci)
#     print("Coverage for eps =", eps1, ":", coverage1)
#
#

#### debias funcs ####

def inv_sens_debias(var_hat, params, dist, median=True):
    """debias function for invSens, tailored to the *median*."""

    if median:
        c = 8
        true_density = dist.pdf(dist.ppf(0.5))
    else:
        #for noise addition (lap)
        c= 2 * ((params.U-params.L)**2)
        true_density = 1

    n, m, T, eps_m, eps_n, scaling_factor = params.n, params.m, params.T, params.lcl_eps, params.center_eps, params.scaling_factor

    lcl_debias_term = (1 - (1 / T)) * (c / ((m * eps_m * true_density) ** 2))
    center_debias_term = (c / ((n * eps_n * true_density) ** 2)) if eps_n > 0 else 0
    var_hat = (scaling_factor ** 2) * (var_hat - lcl_debias_term)
    var_hat += center_debias_term
    return var_hat

def gaussian_debias(var_hat, params):
    lcl_sigma, sigma_cnt = params.sigma_lcl**2, params.sigma_center**2
    print("lcl_sigma:", lcl_sigma, "sigma_cnt:", sigma_cnt)
    return var_hat*(params.scaling_factor ** 2)
    # var_hat = (params.scaling_factor ** 2) * (var_hat - lcl_sigma)
    # var_hat += sigma_cnt
    # return var_hat


def multiplicative(var_hat, params, dist):
    """
    multiplicative function for debaising the variance estimate
    Args:
        var_hat: estimations of subset size m
        params: settings

    Returns: var hat after debias
    """
    n, m, T, eps_m, eps_n, scaling_factor = params.n, params.m, params.T, params.lcl_eps, params.center_eps, params.scaling_factor
    c1, c2 = 0.25, 8

    me_2 = (m * (eps_m ** 2))
    ne_2 = (n * (eps_n ** 2))

    nom = me_2 * (c1 * ne_2 + c2)
    denom = ne_2 * (c1 * me_2 + c2)
    var_hat = (scaling_factor ** 2) * (nom/denom) * var_hat
    return var_hat