graphld.clumping

LD clumping for identifying independent variants.

LD clumping identifies independent index variants by iteratively selecting the variant with the highest χ² statistic and pruning all variants in high LD with it. Clumping + thresholding is a popular (though suboptimal) way of computing polygenic scores.

For usage examples, see the LD Clumping guide.

clumping

LD clumping implementation using ParallelProcessor framework.

LDClumper

Bases: ParallelProcessor

Fast LD clumping to find unlinked lead SNPs from GWAS summary statistics.

prepare_block_data classmethod

prepare_block_data(metadata: DataFrame, **kwargs: Any) -> list[tuple]

Split summary statistics into blocks whose positions match the LDGMs.

Parameters:

Name	Type	Description	Default
metadata	DataFrame	DataFrame containing LDGM metadata	required
**kwargs	Any	Additional arguments from run(), including: sumstats: DataFrame containing summary statistics	{}

Returns:

Type	Description
list[tuple]	List of block-specific sumstats DataFrames and their offsets

Source code in src/graphld/clumping.py

@classmethod
def prepare_block_data(cls, metadata: pl.DataFrame, **kwargs: Any) -> list[tuple]:
    """Split summary statistics into blocks whose positions match the LDGMs.

    Args:
        metadata: DataFrame containing LDGM metadata
        **kwargs: Additional arguments from run(), including:
            sumstats: DataFrame containing summary statistics

    Returns:
        List of block-specific sumstats DataFrames and their offsets
    """
    sumstats = kwargs.get('sumstats')
    row_nr_col = cls._temporary_column(sumstats.columns, cls._ROW_NR_COL)
    indexed_sumstats = sumstats.with_row_index(name=row_nr_col)

    # Partition annotations into blocks
    sumstats_blocks: list[pl.DataFrame] = partition_variants(metadata, indexed_sumstats)

    cumulative_num_variants = np.cumsum(np.array([len(df) for df in sumstats_blocks]))
    cumulative_num_variants = [0] + list(cumulative_num_variants[:-1])

    return list(zip(sumstats_blocks, cumulative_num_variants, strict=False))

create_shared_memory staticmethod

create_shared_memory(metadata: DataFrame, block_data: list[tuple], **kwargs: Any) -> SharedData

Create output array with length number of variants in the summary statistics.

Parameters:

Name	Type	Description	Default
metadata	DataFrame	Metadata DataFrame containing block information	required
block_data	list[tuple]	List of block-specific sumstats DataFrames	required
**kwargs	Any	Not used	{}

Returns:

Type	Description
SharedData	SharedData containing arrays for clumping results

Source code in src/graphld/clumping.py

@staticmethod
def create_shared_memory(metadata: pl.DataFrame, block_data: list[tuple], **kwargs: Any) -> SharedData:
    """Create output array with length number of variants in the summary statistics.

    Args:
        metadata: Metadata DataFrame containing block information
        block_data: List of block-specific sumstats DataFrames
        **kwargs: Not used

    Returns:
        SharedData containing arrays for clumping results
    """
    total_variants = sum([len(df) for df, _ in block_data])
    return SharedData({
        'is_index': total_variants,  # Will be converted to boolean later
    })

process_block classmethod

process_block(ldgm: PrecisionOperator, flag: Value, shared_data: SharedData, block_offset: int, block_data: tuple, worker_params: tuple) -> None

Process single block for LD clumping.

Parameters:

Name	Type	Description	Default
ldgm	PrecisionOperator	LDGM object	required
flag	Value	Worker flag	required
shared_data	SharedData	Dictionary-like shared data object	required
block_offset	int	Offset for this block	required
block_data	tuple	Tuple of (sumstats DataFrame, variant_offset)	required
worker_params	tuple	Tuple of (rsq_threshold, chisq_threshold, z_col, match_by_position, variant_id_col)	required

Source code in src/graphld/clumping.py

@classmethod
def process_block(cls, ldgm: PrecisionOperator,
                flag: Value,
                shared_data: SharedData,
                block_offset: int,
                block_data: tuple,
                worker_params: tuple) -> None:
    """Process single block for LD clumping.

    Args:
        ldgm: LDGM object
        flag: Worker flag
        shared_data: Dictionary-like shared data object
        block_offset: Offset for this block
        block_data: Tuple of (sumstats DataFrame, variant_offset)
        worker_params: Tuple of (rsq_threshold, chisq_threshold, z_col, match_by_position, variant_id_col)
    """
    rsq_threshold, chisq_threshold, z_col, match_by_position, variant_id_col = worker_params
    assert isinstance(block_data, tuple), "block_data must be a tuple"
    sumstats, variant_offset = block_data
    num_variants = len(sumstats)
    if num_variants == 0:
        return

    # Merge variants with LDGM variant info and get indices of merged variants
    ldgm, sumstat_indices = merge_snplists(
        ldgm, sumstats,
        match_by_position=match_by_position,
        pos_col='POS',
        variant_id_col=variant_id_col,
        ref_allele_col='REF',
        alt_allele_col='ALT',
        add_allelic_cols=[z_col],  # Z score column
    )

    # Keep only first occurrence of each index
    first_index_mask = pl.Series(ldgm.variant_info.select(pl.col('is_representative')).to_numpy().flatten().astype(bool))
    ldgm.variant_info = ldgm.variant_info.filter(first_index_mask)
    sumstat_indices = sumstat_indices[first_index_mask]

    # Get Z scores and compute chi-square statistics
    z_scores = ldgm.variant_info.select(z_col).to_numpy().flatten()  # Z score column
    chisq = z_scores ** 2

    # Sort variants by chi-square statistic
    sort_idx = np.argsort(chisq)[::-1]  # Descending order

    # Initialize arrays for tracking pruned and index variants
    n = len(z_scores)
    was_pruned = np.zeros(n, dtype=bool)
    is_index = np.zeros(n, dtype=bool)

    # Iterate through variants in order of decreasing chi-square
    for i in sort_idx:
        # Stop if we reach variants below threshold
        if chisq[i] < chisq_threshold:
            break

        # Skip if this variant was already pruned
        if was_pruned[i]:
            continue

        # This is an index variant
        is_index[i] = True

        # Compute LD with all other variants
        indicator = np.zeros(n)
        indicator[i] = 1
        ld = ldgm.solve(indicator)

        # Mark variants in high LD for pruning
        to_prune = (ld ** 2 >= rsq_threshold)
        assert to_prune[i]  # Lead SNP should be in LD with itself
        was_pruned[to_prune] = True

    # Initialize results array with all False
    results = np.zeros(num_variants, dtype=float)  # Use float for shared memory

    # Map results back to original variants using sumstat_indices
    # Only set results for variants that were successfully merged
    results[sumstat_indices.flatten()] = is_index.astype(float)  # Convert to float for shared memory

    # Store results
    block_slice = slice(variant_offset, variant_offset + num_variants)
    shared_data['is_index', block_slice] = results

supervise classmethod

supervise(manager: Union[WorkerManager, SerialManager], shared_data: SharedData, block_data: list, **kwargs: Any) -> pl.DataFrame

Monitor workers and process results.

Parameters:

Name	Type	Description	Default
manager	Union[WorkerManager, SerialManager]	Worker manager for controlling processes	required
shared_data	SharedData	Shared memory data	required
block_data	list	List of block-specific data	required
**kwargs	Any	Additional arguments passed from run()	{}

Returns:

Type	Description
DataFrame	DataFrame with clumping results

Source code in src/graphld/clumping.py

@classmethod
def supervise(cls, manager: Union[WorkerManager, SerialManager],
            shared_data: SharedData,
            block_data: list, **kwargs: Any) -> pl.DataFrame:
    """Monitor workers and process results.

    Args:
        manager: Worker manager for controlling processes
        shared_data: Shared memory data
        block_data: List of block-specific data
        **kwargs: Additional arguments passed from run()

    Returns:
        DataFrame with clumping results
    """
    manager.start_workers()
    manager.await_workers()
    is_index = shared_data['is_index']
    sumstats = kwargs['sumstats']
    row_nr_col = cls._temporary_column(sumstats.columns, cls._ROW_NR_COL)
    result_col = cls._temporary_column(
        sumstats.columns + [row_nr_col], cls._IS_INDEX_COL
    )

    block_results = []
    for df, variant_offset in block_data:
        if len(df) == 0:
            continue
        block_slice = slice(variant_offset, variant_offset + len(df))
        block_results.append(
            df.select(row_nr_col).with_columns(
                pl.Series(result_col, is_index[block_slice].astype(bool))
            )
        )

    indexed_sumstats = sumstats.with_row_index(name=row_nr_col)
    if block_results:
        clump_results = (
            pl.concat(block_results)
            .group_by(row_nr_col)
            .agg(pl.col(result_col).any())
        )
        indexed_sumstats = indexed_sumstats.join(
            clump_results, on=row_nr_col, how='left'
        )
    else:
        indexed_sumstats = indexed_sumstats.with_columns(
            pl.lit(None, dtype=pl.Boolean).alias(result_col)
        )

    return (
        indexed_sumstats
        .sort(row_nr_col)
        .with_columns(
            pl.col(result_col).fill_null(False).cast(pl.Boolean).alias('is_index')
        )
        .drop([row_nr_col, result_col])
    )

clump classmethod

clump(sumstats: DataFrame, ldgm_metadata_path: str = 'data/ldgms/metadata.csv', rsq_threshold: float = 0.1, chisq_threshold: float = 30.0, populations: Optional[Union[str, List[str]]] = None, chromosomes: Optional[Union[int, List[int]]] = None, run_in_serial: bool = False, num_processes: Optional[int] = None, z_col: str = 'Z', match_by_position: bool = True, variant_id_col: str = 'SNP', verbose: bool = False) -> pl.DataFrame

Perform LD clumping on summary statistics.

Parameters:

Name	Type	Description	Default
sumstats	DataFrame	Summary statistics DataFrame containing Z scores	required
ldgm_metadata_path	str	Path to metadata CSV file (default 'data/ldgms/metadata.csv')	'data/ldgms/metadata.csv'
rsq_threshold	float	r² threshold for clumping (default 0.1)	0.1
chisq_threshold	float	χ² threshold for significance (default 30.0)	30.0
populations	Optional[Union[str, List[str]]]	Optional population name(s)	None
chromosomes	Optional[Union[int, List[int]]]	Optional chromosome(s)	None
run_in_serial	bool	Whether to run in serial mode	False
num_processes	Optional[int]	Optional number of processes	None
z_col	str	Name of column containing Z scores	'Z'
match_by_position	bool	Whether to match SNPs by position instead of ID	True
variant_id_col	str	Name of column containing variant IDs if not matching by position	'SNP'
verbose	bool	Whether to print progress information	False

Returns:

Type	Description
DataFrame	Input DataFrame in its original row order with an additional boolean
DataFrame	'is_index' column indicating index variants. Variants outside the
DataFrame	selected LDGM blocks, unmatched variants, and allele-mismatched
DataFrame	variants are retained with 'is_index=False'.

Source code in src/graphld/clumping.py

@classmethod
def clump(cls,
        sumstats: pl.DataFrame,
        ldgm_metadata_path: str = 'data/ldgms/metadata.csv',
        rsq_threshold: float = 0.1,
        chisq_threshold: float = 30.0,
        populations: Optional[Union[str, List[str]]] = None,
        chromosomes: Optional[Union[int, List[int]]] = None,
        run_in_serial: bool = False,
        num_processes: Optional[int] = None,
        z_col: str = 'Z',
        match_by_position: bool = True,
        variant_id_col: str = 'SNP',
        verbose: bool = False,
        ) -> pl.DataFrame:
    """Perform LD clumping on summary statistics.

    Args:
        sumstats: Summary statistics DataFrame containing Z scores
        ldgm_metadata_path: Path to metadata CSV file (default 'data/ldgms/metadata.csv')
        rsq_threshold: r² threshold for clumping (default 0.1)
        chisq_threshold: χ² threshold for significance (default 30.0)
        populations: Optional population name(s)
        chromosomes: Optional chromosome(s)
        run_in_serial: Whether to run in serial mode
        num_processes: Optional number of processes
        z_col: Name of column containing Z scores
        match_by_position: Whether to match SNPs by position instead of ID
        variant_id_col: Name of column containing variant IDs if not matching by position
        verbose: Whether to print progress information

    Returns:
        Input DataFrame in its original row order with an additional boolean
        'is_index' column indicating index variants. Variants outside the
        selected LDGM blocks, unmatched variants, and allele-mismatched
        variants are retained with 'is_index=False'.
    """

    run_fn = cls.run_serial if run_in_serial else cls.run
    result = run_fn(ldgm_metadata_path=ldgm_metadata_path,
            populations=populations,
            chromosomes=chromosomes,
            num_processes=num_processes,
            worker_params=(rsq_threshold, chisq_threshold, z_col, match_by_position, variant_id_col),
            sumstats=sumstats)

    if verbose:
        print(f"Number of variants in summary statistics: {len(result)}")
        nonzero_count = (result['is_index']).sum()
        print(f"Number of index variants: {nonzero_count}")

    return result

run_clump

run_clump(*args: Any, **kwargs: Any) -> pl.DataFrame

Perform LD-based clumping on summary statistics.

Positional and keyword arguments are forwarded to :meth:LDClumper.clump.

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments for :meth:LDClumper.clump.	()
**kwargs	Any	Keyword arguments for :meth:LDClumper.clump.	{}

Returns:

Type	Description
DataFrame	Input DataFrame in its original row order with an additional boolean
DataFrame	'is_index' column. Variants outside the selected LDGM blocks,
DataFrame	unmatched variants, and allele-mismatched variants are retained with
DataFrame	'is_index=False'.

Source code in src/graphld/clumping.py

def run_clump(*args: Any, **kwargs: Any) -> pl.DataFrame:
    """Perform LD-based clumping on summary statistics.

    Positional and keyword arguments are forwarded to :meth:`LDClumper.clump`.

    Args:
        *args: Positional arguments for :meth:`LDClumper.clump`.
        **kwargs: Keyword arguments for :meth:`LDClumper.clump`.

    Returns:
        Input DataFrame in its original row order with an additional boolean
        'is_index' column. Variants outside the selected LDGM blocks,
        unmatched variants, and allele-mismatched variants are retained with
        'is_index=False'.
    """
    return LDClumper.clump(*args, **kwargs)