Skip to content

mecfs_bio.build_system.task.acat_task

Task to combine p-values across collapsing models via the Aggregated Cauchy Association Test (ACAT).

Liu, Yaowu, et al. "ACAT: a fast and powerful p value combination method for rare-variant analysis in sequencing studies." The American Journal of Human Genetics 104.3 (2019): 410-421.

Classes:

Attributes:

ACAT_P_COL module-attribute 

ACAT_P_COL = 'acat_p'

AcatTask

Bases: Task

Methods:

Attributes:

deps property 

deps: list[Task]

excluded_models instance-attribute 

excluded_models: list[str]

group_by instance-attribute 

group_by: list[str]

meta instance-attribute 

meta: Meta

model_col instance-attribute 

model_col: str

p_value_col instance-attribute 

p_value_col: str

source_task instance-attribute 

source_task: Task

create classmethod 

create(
    source_task: Task,
    asset_id: str,
    group_by: Sequence[str],
    p_value_col: str,
    model_col: str,
    excluded_models: Sequence[str] = (),
) -> AcatTask
Source code in mecfs_bio/build_system/task/acat_task.py 
 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
@classmethod
def create(
    cls,
    source_task: Task,
    asset_id: str,
    group_by: Sequence[str],
    p_value_col: str,
    model_col: str,
    excluded_models: Sequence[str] = (),
) -> "AcatTask":
    source_meta = source_task.meta
    assert isinstance(source_meta, (GWASSummaryDataFileMeta, FilteredGWASDataMeta))
    meta = FilteredGWASDataMeta(
        id=AssetId(asset_id),
        trait=source_meta.trait,
        project=source_meta.project,
        sub_dir=PurePath("processed"),
        read_spec=DataFrameReadSpec(DataFrameParquetFormat()),
    )
    return cls(
        source_task=source_task,
        meta=meta,
        group_by=list(group_by),
        p_value_col=p_value_col,
        model_col=model_col,
        excluded_models=list(excluded_models),
    )

execute 

execute(scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset
Source code in mecfs_bio/build_system/task/acat_task.py 
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
def execute(self, scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset:
    asset = fetch(self.source_task.asset_id)
    lf = scan_dataframe_asset(
        asset, self.source_task.meta, parquet_backend="polars"
    )
    pl_df = pl.from_dataframe(lf.collect())

    if self.excluded_models:
        pl_df = pl_df.filter(~pl.col(self.model_col).is_in(self.excluded_models))

    result = (
        pl_df.group_by(self.group_by)
        .agg(
            pl.col(self.model_col).sort().str.join(", ").alias("models_used"),
            pl.col(self.model_col).count().alias("n_models"),
            pl.col(self.p_value_col)
            .map_batches(
                _acat_combine,
                return_dtype=pl.Float64,
                returns_scalar=True,
            )
            .alias(ACAT_P_COL),
        )
        .sort(self.group_by)
    )

    out_path = scratch_dir / "acat_result.parquet"
    result.write_parquet(out_path)
    return FileAsset(out_path)