Skip to content

mecfs_bio.build_system.task.fast_ct_ldsc.trait_matrix

Stack the aligned traits of a fast CT-LDSC study into (S, T) matrices over the shared context.

Glossary: S is the number of context (regression) SNPs and T the number of traits. Column t of z and n holds trait t's aligned z-score and per-SNP sample size (NaN where the trait lacks the variant), in context row order.

Classes:

  • TraitMatrix –

    The study's traits over the shared context: z and n are (S, T); aliases, diagnostics and

Functions:

TraitMatrix

The study's traits over the shared context: z and n are (S, T); aliases, diagnostics and sample_infos are length T, in column order.

Attributes:

aliases instance-attribute 

aliases: tuple[str, ...]

context instance-attribute 

context: PppLdscContext

diagnostics instance-attribute 

diagnostics: tuple[AlignmentDiagnostics, ...]

n instance-attribute 

n: ndarray

sample_infos instance-attribute 

sample_infos: tuple[PhenotypeInfo, ...]

z instance-attribute 

z: ndarray

assert_valid_study 

assert_valid_study(
    aligned_tasks: Sequence[FastCtLdscAlignedTraitTask],
) -> None

A study needs at least two traits with unique aliases, ordered by lower-cased alias.

Source code in mecfs_bio/build_system/task/fast_ct_ldsc/trait_matrix.py 
108
109
110
111
112
113
114
115
def assert_valid_study(aligned_tasks: Sequence[FastCtLdscAlignedTraitTask]) -> None:
    """A study needs at least two traits with unique aliases, ordered by lower-cased alias."""
    aliases = [task.alias for task in aligned_tasks]
    assert len(aliases) >= 2, "an all-pairs study needs at least two traits"
    assert len(set(aliases)) == len(aliases), f"duplicate trait aliases in {aliases}"
    assert aliases == sorted(aliases, key=str.lower), (
        f"aligned tasks must be ordered by lower-cased alias, got {aliases}"
    )

estimate_trait_contexts 

estimate_trait_contexts(
    matrix: TraitMatrix, *, n_blocks: int
) -> list[TraitLdscContext]

Each trait's own-set LDSC heritability fit (estimate_trait_context), in column order.

Source code in mecfs_bio/build_system/task/fast_ct_ldsc/trait_matrix.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def estimate_trait_contexts(
    matrix: TraitMatrix, *, n_blocks: int
) -> list[TraitLdscContext]:
    """Each trait's own-set LDSC heritability fit (estimate_trait_context), in column order."""
    return [
        estimate_trait_context(
            z_trait=matrix.z[:, t],
            n_trait=matrix.n[:, t],
            ld=matrix.context.ld,
            m=matrix.context.m,
            n_blocks=n_blocks,
        )
        for t in range(len(matrix.aliases))
    ]

load_trait_matrix 

load_trait_matrix(
    index_task: FastCtLdscVariantIndexTask,
    aligned_tasks: Sequence[FastCtLdscAlignedTraitTask],
    fetch: Fetch,
) -> TraitMatrix

Read every aligned trait and stack it, asserting each was aligned to this context.

Source code in mecfs_bio/build_system/task/fast_ct_ldsc/trait_matrix.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def load_trait_matrix(
    index_task: FastCtLdscVariantIndexTask,
    aligned_tasks: Sequence[FastCtLdscAlignedTraitTask],
    fetch: Fetch,
) -> TraitMatrix:
    """Read every aligned trait and stack it, asserting each was aligned to this context."""
    indexed = load_indexed_ldsc_context(index_task, fetch)
    context_rsids = indexed.variants[GWASLAB_RSID_COL]
    arrays = [read_aligned_trait(task, fetch) for task in aligned_tasks]
    for task, aligned in zip(aligned_tasks, arrays):
        assert aligned.rsid.equals(context_rsids), (
            f"{task.alias}: aligned rsIDs differ from the context's; the aligned trait was "
            "built against a different variant index"
        )
    return TraitMatrix(
        context=indexed.context,
        aliases=tuple(task.alias for task in aligned_tasks),
        z=np.column_stack([a.z for a in arrays]),
        n=np.column_stack([a.n for a in arrays]),
        diagnostics=tuple(a.diagnostics for a in arrays),
        sample_infos=tuple(task.source.sample_info for task in aligned_tasks),
    )

multi_trait_table_meta 

multi_trait_table_meta(asset_id: str) -> ResultTableMeta
Source code in mecfs_bio/build_system/task/fast_ct_ldsc/trait_matrix.py 
125
126
127
128
129
130
131
132
133
def multi_trait_table_meta(asset_id: str) -> ResultTableMeta:
    return ResultTableMeta(
        id=asset_id,
        trait=MULTI_TRAIT,
        project="genetic_correlation",
        sub_dir=PurePath("fast_ct_ldsc"),
        extension=".parquet",
        read_spec=DataFrameReadSpec(DataFrameParquetFormat()),
    )

study_deps 

study_deps(
    index_task: FastCtLdscVariantIndexTask,
    aligned_tasks: Sequence[FastCtLdscAlignedTraitTask],
) -> list[Task]
Source code in mecfs_bio/build_system/task/fast_ct_ldsc/trait_matrix.py 
118
119
120
121
122
def study_deps(
    index_task: FastCtLdscVariantIndexTask,
    aligned_tasks: Sequence[FastCtLdscAlignedTraitTask],
) -> list[Task]:
    return [index_task, index_task.ld_scores_task, *aligned_tasks]