File size: 3,494 Bytes
a8951a1 | 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 | """Unlearning methods wrapper."""
import numpy as np
from typing import Dict, Optional, Tuple
from src.graph_utils import build_adjacency, get_deletion_neighborhood, get_user_item_sets_in_radius
def one_step_downdate_poisson_gamma(edges, edge_to_remove, full_params,
N, M, K, a0, b0, c0, d0):
"""One-step local downdate: subtract deleted contribution from seed blocks only."""
import time
t0 = time.time()
i_del, j_del, x_del = edge_to_remove
params = {k: v.copy() for k, v in full_params.items()}
a, b, c, d = params['a'], params['b'], params['c'], params['d']
if x_del > 0:
# Compute responsibility for the deleted edge
from scipy.special import digamma
from src.utils import stable_softmax
log_r = digamma(a[i_del]) - np.log(b[i_del]) + digamma(c[j_del]) - np.log(d[j_del])
r = stable_softmax(log_r)
# Subtract contribution from user i
a[i_del] = np.maximum(a0, a[i_del] - x_del * r)
b[i_del] = np.maximum(b0 * 0.5, b[i_del] - c[j_del] / d[j_del])
# Subtract contribution from item j
c[j_del] = np.maximum(c0, c[j_del] - x_del * r)
d[j_del] = np.maximum(d0 * 0.5, d[j_del] - a[i_del] / b[i_del])
runtime = time.time() - t0
from src.utils import FitResult
return FitResult(
params=params,
objective_trace=[],
n_iterations=1,
converged=True,
runtime_sec=runtime,
model_family='poisson_gamma',
inference_type='vi',
likelihood='poisson',
prior='gamma',
diagnostics={'method': 'one_step_downdate'}
)
def one_step_downdate_gaussian(edges, edge_to_remove, full_params,
N, M, K, sigma_x, sigma_U=None, sigma_V=None,
model_family='gaussian_gaussian'):
"""One-step local downdate for Gaussian models."""
import time
t0 = time.time()
i_del, j_del, x_del = edge_to_remove
params = {k: v.copy() for k, v in full_params.items()}
if model_family == 'gaussian_gaussian':
m_U, s_U = params['m_U'], params['s_U']
m_V, s_V = params['m_V'], params['s_V']
prec_x = 1.0 / (sigma_x ** 2)
# Remove contribution of edge (i,j) from user i
for k in range(K):
old_prec = 1.0 / s_U[i_del, k]
new_prec = old_prec - prec_x * (m_V[j_del, k]**2 + s_V[j_del, k])
new_prec = max(new_prec, 1.0 / (sigma_U**2 if sigma_U else 1.0))
s_U[i_del, k] = 1.0 / new_prec
for k in range(K):
old_prec = 1.0 / s_V[j_del, k]
new_prec = old_prec - prec_x * (m_U[i_del, k]**2 + s_U[i_del, k])
new_prec = max(new_prec, 1.0 / (sigma_V**2 if sigma_V else 1.0))
s_V[j_del, k] = 1.0 / new_prec
elif model_family == 'gaussian_gamma_map':
# For MAP: one gradient step removing contribution
pass
runtime = time.time() - t0
from src.utils import FitResult
return FitResult(
params=params, objective_trace=[], n_iterations=1,
converged=True, runtime_sec=runtime,
model_family=model_family, inference_type='vi' if 'gaussian_gaussian' == model_family else 'map',
likelihood='gaussian', prior='gaussian' if model_family == 'gaussian_gaussian' else 'gamma',
diagnostics={'method': 'one_step_downdate'}
)
|