Skip to content

mecfs_bio.build_system.task.ppp_ldsc.batched_ldsc_rg

Batched cross-trait LD-score-regression genetic correlation for the UKB-PPP database.

One external trait is correlated against every protein. Because every protein's z-score is defined on the SAME shared SNP set (see ppp_ldsc_context), we process K proteins at once and run the three regressions each genetic correlation needs:

rg = gcov(trait, protein) / sqrt(h2_trait * h2_protein)

Following the GenomicSEM convention (see genomic_sem_ldsc.run_ldsc), each DIAGONAL heritability is estimated on that trait's own kept SNP set, and only the OFF-DIAGONAL covariance on the pair's intersection. So h2_trait -- its point estimate, its per-block leave-one-out delete values, and n_bar -- depends only on the trait and is computed exactly ONCE (estimate_trait_context) and reused for every protein. Only h2_protein and gcov are per-protein.

Per-protein exclusions (missing variants, the chi-square filter, and cis variants) are encoded as ZERO regression weights over the shared row set rather than by dropping rows, so all proteins keep the same contiguous jackknife blocks -- exactly as in batched_ldsc_h2.

Weighting (the easy bug): the univariate h2 weight is omega = het * oc (the PRODUCT), because estimate_h2 multiplies BOTH design and response by sqrt(het * oc). The covariance weight is omega_cov = oc * (sqrt(het_trait) + sqrt(het_protein))^2, from estimate_cov's weights_cov = (initial_w_trait + initial_w_protein) with initial_w = sqrt(het * oc); the sum-to-1 normalization is a common scalar that cancels in the 2x2 slope solve. The heteroskedasticity aggregate-h2 that shapes het is computed over the SAME set the regression runs on: own kept set for a diagonal h2, the pair intersection for the covariance.

The rg standard error is a delete-block jackknife on the ratio: leave-one-block-out slopes for gcov and h2_protein are combined with the once-computed h2_trait deletes at the same block index to form per-block rg values, and rg_se is their jackknife spread. This is the classic ldsc rg SE.

Classes:

  • BatchedRgResult –

    Per-protein cross-trait rg outputs; all arrays are length K (protein order preserved).

  • ExactRgResult –
  • TraitLdscContext –

    The trait-only LDSC quantities, computed once and reused for every protein.

Functions:

  • batched_rg –

    Estimate the trait-vs-protein genetic correlation for K proteins sharing the SNP set.

  • estimate_trait_context –

    Estimate the trait heritability once over the shared SNP set and package the trait-only

  • exact_rg_single –

    Reference single-protein genetic correlation via the repo's exact GenomicSEM port: h2 on

BatchedRgResult

Per-protein cross-trait rg outputs; all arrays are length K (protein order preserved). h2_trait and n_bar_trait are trait-level scalars shared by every protein.

Attributes:

gcov instance-attribute

gcov: ndarray

gcov_intercept instance-attribute

gcov_intercept: ndarray

h2_protein instance-attribute

h2_protein: ndarray

h2_trait instance-attribute

h2_trait: float

n_bar_protein instance-attribute

n_bar_protein: ndarray

n_bar_trait instance-attribute

n_bar_trait: float

n_snps instance-attribute

n_snps: ndarray

rg instance-attribute

rg: ndarray

rg_p instance-attribute

rg_p: ndarray

rg_se instance-attribute

rg_se: ndarray

rg_z instance-attribute

rg_z: ndarray

ExactRgResult

Attributes:

gcov instance-attribute

gcov: float

gcov_intercept instance-attribute

gcov_intercept: float

h2_protein instance-attribute

h2_protein: float

h2_trait instance-attribute

h2_trait: float

rg instance-attribute

rg: float

TraitLdscContext

The trait-only LDSC quantities, computed once and reused for every protein.

z / keep / n are parallel (S,) arrays over the shared context SNP set: z is the signed trait z-score (index-effect-allele oriented, zeroed off the kept set), keep marks the trait's kept SNPs (present and passing the chi-square filter), and n is the per-SNP trait sample size. h2 / n_bar are the trait heritability and its N_bar; h2_delete holds the (n_blocks,) per-block leave-one-out h2_trait values used by the rg jackknife.

Attributes:

h2 instance-attribute

h2: float

h2_delete instance-attribute

h2_delete: ndarray

keep instance-attribute

keep: ndarray

n instance-attribute

n: ndarray

n_bar instance-attribute

n_bar: float

z instance-attribute

z: ndarray

batched_rg

batched_rg(
    trait_ctx: TraitLdscContext,
    z_protein: ndarray,
    ld: ndarray,
    n_protein: ndarray,
    m: float,
    *,
    n_blocks: int = DEFAULT_N_BLOCKS,
    exclude: ndarray | None = None,
) -> BatchedRgResult

Estimate the trait-vs-protein genetic correlation for K proteins sharing the SNP set.

trait_ctx: the once-computed trait LDSC context (estimate_trait_context). z_protein: (S, K) signed protein z-score (= BETA/SE, index-effect-allele oriented), NaN where a protein lacks the variant. ld: (S,) LD score, genome-sorted. n_protein: (K,) per-protein sample size. m: total reference-SNP count. exclude: optional (S, K) boolean; True drops that variant for that protein (e.g. cis).

Source code in mecfs_bio/build_system/task/ppp_ldsc/batched_ldsc_rg.py
def batched_rg(
    trait_ctx: TraitLdscContext,
    z_protein: np.ndarray,
    ld: np.ndarray,
    n_protein: np.ndarray,
    m: float,
    *,
    n_blocks: int = DEFAULT_N_BLOCKS,
    exclude: np.ndarray | None = None,
) -> BatchedRgResult:
    """Estimate the trait-vs-protein genetic correlation for K proteins sharing the SNP set.

    trait_ctx: the once-computed trait LDSC context (estimate_trait_context).
    z_protein: (S, K) signed protein z-score (= BETA/SE, index-effect-allele oriented), NaN where
        a protein lacks the variant.
    ld: (S,) LD score, genome-sorted. n_protein: (K,) per-protein sample size. m: total
        reference-SNP count. exclude: optional (S, K) boolean; True drops that variant for that
        protein (e.g. cis).
    """
    s, k = z_protein.shape
    ldm = np.maximum(ld, 1.0)
    oc = 1.0 / ldm

    finite_p = np.isfinite(z_protein)
    z_p = np.where(finite_p, z_protein, 0.0)
    chi_p = z_p * z_p
    n_p = np.broadcast_to(n_protein.astype(float), (s, k))
    keep_p = finite_p & (chi_p <= _chisq_threshold(n_protein)[None, :])
    if exclude is not None:
        keep_p = keep_p & ~exclude

    # Protein heritability: own-set aggregate-h2, weight het*oc (the univariate PRODUCT).
    tot_agg_p = _aggregate_h2(chi_p, ld, n_p, keep_p, m)
    w_p = _sqrt_het(tot_agg_p, ldm, n_p, m) ** 2 * oc[:, None] * keep_p
    slope_p, _intc_p, slope_p_del = _blocked_wls(w_p, chi_p, ld, n_blocks)
    cnt_p = keep_p.sum(0).astype(float)
    n_bar_p = _column_mean_over_keep(n_p, keep_p, cnt_p)
    h2_p = slope_p / n_bar_p * m
    h2_p_del = slope_p_del / n_bar_p[None, :] * m

    # Trait-protein covariance on the intersection: het weights use intersection aggregate-h2.
    keep_cov = trait_ctx.keep[:, None] & keep_p
    chi_t = (trait_ctx.z * trait_ctx.z)[:, None]
    n_t = trait_ctx.n[:, None]
    zz = np.where(keep_cov, trait_ctx.z[:, None] * z_p, 0.0)
    tot_agg_t_cov = _aggregate_h2(
        np.broadcast_to(chi_t, (s, k)), ld, np.broadcast_to(n_t, (s, k)), keep_cov, m
    )
    tot_agg_p_cov = _aggregate_h2(chi_p, ld, n_p, keep_cov, m)
    sqrt_het_t = _sqrt_het(tot_agg_t_cov, ldm, np.broadcast_to(n_t, (s, k)), m)
    sqrt_het_p = _sqrt_het(tot_agg_p_cov, ldm, n_p, m)
    w_cov = oc[:, None] * (sqrt_het_t + sqrt_het_p) ** 2 * keep_cov
    slope_cov, intercept_cov, slope_cov_del = _blocked_wls(w_cov, zz, ld, n_blocks)

    cnt_cov = keep_cov.sum(0).astype(float)
    mean_n_t = _column_mean_over_keep(np.broadcast_to(n_t, (s, k)), keep_cov, cnt_cov)
    mean_n_p = _column_mean_over_keep(n_p, keep_cov, cnt_cov)
    n_bar_cov = np.sqrt(mean_n_t * mean_n_p)
    gcov = slope_cov / n_bar_cov * m
    gcov_del = slope_cov_del / n_bar_cov[None, :] * m

    # rg and its delete-block jackknife SE on the ratio.
    invalid = (h2_p <= 0.0) | (trait_ctx.h2 <= 0.0)
    with np.errstate(invalid="ignore"):
        rg = np.where(invalid, np.nan, gcov / np.sqrt(trait_ctx.h2 * h2_p))
        denom_del = np.sqrt(trait_ctx.h2_delete[:, None] * h2_p_del)
        rg_del = gcov_del / denom_del
        rg_del_mean = rg_del.mean(0)
        rg_se = np.sqrt(
            (n_blocks - 1) / n_blocks * ((rg_del - rg_del_mean[None, :]) ** 2).sum(0)
        )
        rg_se = np.where(invalid, np.nan, rg_se)
        rg_z = rg / rg_se
    rg_p = 2.0 * norm.sf(np.abs(rg_z))

    return BatchedRgResult(
        rg=rg,
        rg_se=rg_se,
        rg_z=rg_z,
        rg_p=rg_p,
        gcov=gcov,
        gcov_intercept=intercept_cov,
        h2_protein=h2_p,
        n_snps=cnt_cov.astype(np.int64),
        n_bar_protein=n_bar_p,
        h2_trait=trait_ctx.h2,
        n_bar_trait=trait_ctx.n_bar,
    )

estimate_trait_context

estimate_trait_context(
    z_trait: ndarray,
    n_trait: ndarray,
    ld: ndarray,
    m: float,
    *,
    n_blocks: int = DEFAULT_N_BLOCKS,
) -> TraitLdscContext

Estimate the trait heritability once over the shared SNP set and package the trait-only quantities the batched rg kernel reuses for every protein.

z_trait: (S,) signed trait z-score aligned to the context (index) effect allele, NaN where the trait lacks the variant. n_trait: (S,) trait sample size (NaN where absent). ld: (S,) LD score, genome-sorted for contiguous blocks. m: total reference-SNP count.

Source code in mecfs_bio/build_system/task/ppp_ldsc/batched_ldsc_rg.py
def estimate_trait_context(
    z_trait: np.ndarray,
    n_trait: np.ndarray,
    ld: np.ndarray,
    m: float,
    *,
    n_blocks: int = DEFAULT_N_BLOCKS,
) -> TraitLdscContext:
    """Estimate the trait heritability once over the shared SNP set and package the trait-only
    quantities the batched rg kernel reuses for every protein.

    z_trait: (S,) signed trait z-score aligned to the context (index) effect allele, NaN where
        the trait lacks the variant. n_trait: (S,) trait sample size (NaN where absent). ld: (S,)
        LD score, genome-sorted for contiguous blocks. m: total reference-SNP count.
    """
    finite = np.isfinite(z_trait)
    n_col = np.where(finite, n_trait, 0.0)
    chi = np.where(finite, z_trait * z_trait, 0.0)
    # The per-trait chi-square threshold uses the trait's max N (as in genomic_sem_ldsc).
    threshold = _chisq_threshold(float(np.nanmax(n_trait)))
    keep = finite & (chi <= threshold)
    z = np.where(keep, z_trait, 0.0)

    keep_col = keep[:, None]
    n_sk = n_col[:, None]
    ldm = np.maximum(ld, 1.0)
    oc = 1.0 / ldm
    tot_agg = _aggregate_h2(chi[:, None], ld, n_sk, keep_col, m)
    sqrt_het = _sqrt_het(tot_agg, ldm, n_sk, m)
    w = sqrt_het**2 * oc[:, None] * keep_col
    slope, _intercept, slope_delete = _blocked_wls(w, chi[:, None], ld, n_blocks)

    cnt = float(keep.sum())
    n_bar = float(np.where(keep, n_col, 0.0).sum() / max(cnt, 1.0))
    h2 = float(slope[0] / n_bar * m)
    h2_delete = slope_delete[:, 0] / n_bar * m
    return TraitLdscContext(
        z=z.astype(float),
        keep=keep,
        n=n_col.astype(float),
        h2=h2,
        n_bar=n_bar,
        h2_delete=h2_delete,
    )

exact_rg_single

exact_rg_single(
    z_trait: ndarray,
    n_trait: float,
    z_protein: ndarray,
    ld: ndarray,
    n_protein: float,
    m: float,
    *,
    n_blocks: int = DEFAULT_N_BLOCKS,
    exclude: ndarray | None = None,
) -> ExactRgResult

Reference single-protein genetic correlation via the repo's exact GenomicSEM port: h2 on each trait's own kept set and gcov on the intersection (dropping, not zero-weighting, the filtered/cis SNPs). This is what the batched kernel's point estimates are validated against. z_trait/z_protein/ld are genome-sorted (S,) arrays; n_trait/n_protein are scalar.

Source code in mecfs_bio/build_system/task/ppp_ldsc/batched_ldsc_rg.py
def exact_rg_single(
    z_trait: np.ndarray,
    n_trait: float,
    z_protein: np.ndarray,
    ld: np.ndarray,
    n_protein: float,
    m: float,
    *,
    n_blocks: int = DEFAULT_N_BLOCKS,
    exclude: np.ndarray | None = None,
) -> ExactRgResult:
    """Reference single-protein genetic correlation via the repo's exact GenomicSEM port: h2 on
    each trait's own kept set and gcov on the intersection (dropping, not zero-weighting, the
    filtered/cis SNPs). This is what the batched kernel's point estimates are validated against.
    z_trait/z_protein/ld are genome-sorted (S,) arrays; n_trait/n_protein are scalar."""
    keep_t = np.isfinite(z_trait) & (z_trait**2 <= _chisq_threshold(n_trait))
    keep_p = np.isfinite(z_protein) & (z_protein**2 <= _chisq_threshold(n_protein))
    if exclude is not None:
        keep_p = keep_p & ~exclude
    keep_cov = keep_t & keep_p

    est_t = estimate_h2(
        chi=z_trait[keep_t] ** 2,
        ld_raw=ld[keep_t],
        wld_raw=ld[keep_t],
        n=np.full(int(keep_t.sum()), n_trait),
        m=m,
        n_blocks=n_blocks,
    )
    est_p = estimate_h2(
        chi=z_protein[keep_p] ** 2,
        ld_raw=ld[keep_p],
        wld_raw=ld[keep_p],
        n=np.full(int(keep_p.sum()), n_protein),
        m=m,
        n_blocks=n_blocks,
    )
    est_c = estimate_cov(
        zz=z_trait[keep_cov] * z_protein[keep_cov],
        chi1=z_trait[keep_cov] ** 2,
        chi2=z_protein[keep_cov] ** 2,
        ld_raw=ld[keep_cov],
        wld_raw=ld[keep_cov],
        n_x=np.full(int(keep_cov.sum()), n_trait),
        n_y=np.full(int(keep_cov.sum()), n_protein),
        m=m,
        n_blocks=n_blocks,
    )
    rg = est_c.reg_tot / np.sqrt(est_t.reg_tot * est_p.reg_tot)
    return ExactRgResult(
        rg=float(rg),
        gcov=est_c.reg_tot,
        gcov_intercept=est_c.intercept,
        h2_trait=est_t.reg_tot,
        h2_protein=est_p.reg_tot,
    )