Skip to content

graphld.ldsc_io

I/O functions for LDSC format files (summary statistics and annotations).

For broader loading workflows, see the I/O and Merging guide.

ldsc_io

Functions for reading LDSC sumstats files.

read_ldsc_snplist 

read_ldsc_snplist(file: Union[str, Path], add_positions: bool = True, positions_file: str = POSITIONS_FILE) -> pl.DataFrame

Read LDSC snplist file format.

Source code in src/graphld/ldsc_io.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def read_ldsc_snplist(
    file: Union[str, Path],
    add_positions: bool = True,
    positions_file: str = POSITIONS_FILE,
) -> pl.DataFrame:
    """Read LDSC snplist file format."""
    print(f"Reading LDSC snplist file: {file}")
    df = pl.read_csv(
        file,
        separator='\t',
        columns=['SNP', 'A1', 'A2'],
        has_header=True,
    )

    if add_positions:
        df = _add_positions_to_df(df, positions_file)

    return df

read_ldsc_sumstats 

read_ldsc_sumstats(file: Union[str, Path], add_positions: bool = True, positions_file: str = POSITIONS_FILE, maximum_missingness: float = 1.0) -> pl.DataFrame

Read LDSC sumstats file format.

Parameters:

Name Type Description Default
file Union[str, Path]

Path to LDSC sumstats file

 required
add_positions bool

If True, merge with external file to add positions

 True
positions_file str

File containing RSIDs and positions, defaults to data/rsid_position.csv

 POSITIONS_FILE
maximum_missingness float

Maximum fraction of missing samples allowed. Variants with N below (1 - maximum_missingness) * max(N) are removed.

 1.0

Returns:

Type Description
DataFrame

DataFrame with columns: SNP, N, Z, ALT, REF
DataFrame

If add_positions=True, also includes: CHR, POS
Source code in src/graphld/ldsc_io.py 
 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
def read_ldsc_sumstats(
    file: Union[str, Path],
    add_positions: bool = True,
    positions_file: str = POSITIONS_FILE,
    maximum_missingness: float = 1.0,
) -> pl.DataFrame:
    """Read LDSC sumstats file format.

    Args:
        file: Path to LDSC sumstats file
        add_positions: If True, merge with external file to add positions
        positions_file: File containing RSIDs and positions, defaults to data/rsid_position.csv
        maximum_missingness: Maximum fraction of missing samples allowed.
            Variants with N below (1 - maximum_missingness) * max(N) are removed.

    Returns:
        DataFrame with columns: SNP, N, Z, ALT, REF
        If add_positions=True, also includes: CHR, POS
    """
    print(f"Reading LDSC sumstats file: {file} with max missingness: {maximum_missingness}")
    # Dynamically detect columns
    with open(file, 'r') as f:
        header = f.readline().strip().split('\t')

    # Determine schema overrides based on available columns
    if 'Z' in header:
        schema_overrides = {'SNP': pl.Utf8, 'N': pl.Float64, 'Z': pl.Float64, 'A1': pl.Utf8, 'A2': pl.Utf8}
    elif 'Beta' in header and 'se' in header:
        schema_overrides = {'SNP': pl.Utf8, 'N': pl.Float64, 'Beta': pl.Float64, 'se': pl.Float64, 'A1': pl.Utf8, 'A2': pl.Utf8}
    else:
        raise ValueError("Unsupported LDSC sumstats format")

    # Read the file using polars
    df = pl.read_csv(
        file,
        separator='\t',
        schema_overrides=schema_overrides,
    )

    # Compute Z score if needed
    if 'Beta' in df.columns and 'Z' not in df.columns:
        df = df.with_columns(
            (pl.col('Beta') / pl.col('se')).alias('Z')
        ).drop(['Beta', 'se'])

    # Validate required columns
    required_cols = {'SNP', 'N', 'Z', 'A1', 'A2'}
    missing_cols = required_cols - set(df.columns)
    if missing_cols:
        raise ValueError(f"Missing required columns: {missing_cols}")

    # Rename A1, A2 to ALT, REF
    df = df.with_columns(
        pl.col('A1').alias('ALT'),
        pl.col('A2').alias('REF')
    ).drop(['A1', 'A2'])

    if add_positions:
        df = _add_positions_to_df(df, positions_file)

    # Drop variants with N < (1 - maximum_missingness) * max(N)
    print(f"Num variants before filtering by missingness: {len(df)}")
    df = df.filter(pl.col('N') >= (1 - maximum_missingness) * pl.col('N').max())
    print(f"Num variants after filtering by missingness: {len(df)}")

    return df