Skip to content

mecfs_bio.build_system.task.genome_reference_harmonization.flip

How each summary-statistic column changes when a variant's effect allele is swapped.

Columns are expected to use gwaslab's standard names (the "gwaslab" entry of formatbook.json), plus the regenie binary-trait columns and effective sample size used by datasets in this repo. Every non-allele column must be registered here, or declared through ExtraColumnRule, so that a new allele-dependent statistic can never be silently left unflipped.

All flipped values are computed from the original frame in a single with_columns call, so a confidence bound is never read after being overwritten.

FlipRule is what the type checker sees and the FLIP_* constants are what code uses. An import-time assertion keeps the two in step. A StrEnum cannot replace them, because polars converts Python Enum members into its own Enum dtype.

Classes:

Functions:

  • flip_statistics –

    Apply every flip rule to the rows selected by mask; alleles are not touched.

  • resolve_column_rules –

    Flip rule for every present non-allele column; asserts that none is unregistered.

Attributes:

BOUND_PAIRS module-attribute 

BOUND_PAIRS: tuple[InvertedBoundPair, ...] = (
    InvertedBoundPair(
        lower=GWASLAB_ODDS_RATIO_95L_COL,
        upper=GWASLAB_ODDS_RATIO_95U_COL,
    ),
    InvertedBoundPair(
        lower=GWASLAB_HAZARD_RATIO_95L_COL,
        upper=GWASLAB_HAZARD_RATIO_95U_COL,
    ),
)

COLUMN_FLIP_RULES module-attribute 

COLUMN_FLIP_RULES: Mapping[str, FlipRule] = {
    GWASLAB_BETA_COL: FLIP_NEGATE,
    GWASLAB_Z_COL: FLIP_NEGATE,
    GWASLAB_T_STATISTIC_COL: FLIP_NEGATE,
    GWASLAB_EFFECT_ALLELE_FREQ_COL: FLIP_COMPLEMENT,
    REGENIE_A1FREQ_CASES_COL: FLIP_COMPLEMENT,
    REGENIE_A1FREQ_CONTROLS_COL: FLIP_COMPLEMENT,
    GWASLAB_ODDS_RATIO_COL: FLIP_INVERT,
    GWASLAB_HAZARD_RATIO_COL: FLIP_INVERT,
    GWASLAB_SNPID_COL: FLIP_INVARIANT,
    GWASLAB_RSID_COL: FLIP_INVARIANT,
    GWASLAB_CHROM_COL: FLIP_INVARIANT,
    GWASLAB_POS_COL: FLIP_INVARIANT,
    GWASLAB_SE_COL: FLIP_INVARIANT,
    GWASLAB_P_COL: FLIP_INVARIANT,
    GWASLAB_MLOG10P_COL: FLIP_INVARIANT,
    GWASLAB_CHISQ_COL: FLIP_INVARIANT,
    GWASLAB_F_STATISTIC_COL: FLIP_INVARIANT,
    GWASLAB_P_HET_COL: FLIP_INVARIANT,
    GWASLAB_I2_COL: FLIP_INVARIANT,
    GWASLAB_SNPR2_COL: FLIP_INVARIANT,
    GWASLAB_DOF_COL: FLIP_INVARIANT,
    GWASLAB_SAMPLE_SIZE_COLUMN: FLIP_INVARIANT,
    GWASLAB_N_CASE_COL: FLIP_INVARIANT,
    GWASLAB_N_CONTROL_COL: FLIP_INVARIANT,
    REGENIE_N_CASES_COL: FLIP_INVARIANT,
    REGENIE_N_CONTROLS_COL: FLIP_INVARIANT,
    GWASLAB_EFFECTIVE_SAMPLE_SIZE: FLIP_INVARIANT,
    GWASLAB_INFO_SCORE_COL: FLIP_INVARIANT,
    GWASLAB_MAF_COL: FLIP_INVARIANT,
    REGENIE_TEST_COL: FLIP_INVARIANT,
    REGENIE_EXTRA_COL: FLIP_INVARIANT,
}

DROPPED_COLUMNS module-attribute 

DROPPED_COLUMNS: frozenset[str] = frozenset(
    {GWASLAB_STATUS_COL}
)

FLIP_COMPLEMENT module-attribute 

FLIP_COMPLEMENT: Final[FlipRule] = 'complement'

FLIP_INVARIANT module-attribute 

FLIP_INVARIANT: Final[FlipRule] = 'invariant'

FLIP_INVERT module-attribute 

FLIP_INVERT: Final[FlipRule] = 'invert'

FLIP_NEGATE module-attribute 

FLIP_NEGATE: Final[FlipRule] = 'negate'

FlipRule module-attribute 

FlipRule = Literal[
    "negate", "complement", "invert", "invariant"
]

ExtraColumnRule

Attributes:

column instance-attribute 

column: str

rule instance-attribute 

rule: FlipRule

InvertedBoundPair

Ratio confidence bounds: on a flip, lower becomes 1/upper and upper becomes 1/lower.

Attributes:

lower instance-attribute 

lower: str

upper instance-attribute 

upper: str

flip_statistics 

flip_statistics(
    frame: DataFrame,
    mask: Expr,
    rules: Mapping[str, FlipRule],
) -> pl.DataFrame

Apply every flip rule to the rows selected by mask; alleles are not touched.

Source code in mecfs_bio/build_system/task/genome_reference_harmonization/flip.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def flip_statistics(
    frame: pl.DataFrame, mask: pl.Expr, rules: Mapping[str, FlipRule]
) -> pl.DataFrame:
    """Apply every flip rule to the rows selected by mask; alleles are not touched."""
    updates = [
        pl.when(mask)
        .then(_flipped(column, rule))
        .otherwise(pl.col(column))
        .alias(column)
        for column, rule in rules.items()
        if rule != FLIP_INVARIANT and column in frame.columns
    ]
    for pair in BOUND_PAIRS:
        if pair.lower in frame.columns:
            updates.append(
                pl.when(mask)
                .then(1 / pl.col(pair.upper))
                .otherwise(pl.col(pair.lower))
                .alias(pair.lower)
            )
            updates.append(
                pl.when(mask)
                .then(1 / pl.col(pair.lower))
                .otherwise(pl.col(pair.upper))
                .alias(pair.upper)
            )
    return frame.with_columns(updates) if updates else frame

resolve_column_rules 

resolve_column_rules(
    columns: Sequence[str], extra: Sequence[ExtraColumnRule]
) -> Mapping[str, FlipRule]

Flip rule for every present non-allele column; asserts that none is unregistered.

Source code in mecfs_bio/build_system/task/genome_reference_harmonization/flip.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def resolve_column_rules(
    columns: Sequence[str], extra: Sequence[ExtraColumnRule]
) -> Mapping[str, FlipRule]:
    """Flip rule for every present non-allele column; asserts that none is unregistered."""
    extra_rules = {rule.column: rule.rule for rule in extra}
    assert len(extra_rules) == len(extra), "extra_column_rules names a column twice"
    rules = {**COLUMN_FLIP_RULES, **extra_rules}
    bound_columns = {name for pair in BOUND_PAIRS for name in (pair.lower, pair.upper)}
    unregistered = [
        column
        for column in columns
        if column not in rules
        and column not in bound_columns
        and column not in _ALLELE_COLUMNS
        and column not in DROPPED_COLUMNS
    ]
    assert not unregistered, (
        f"columns without a flip rule: {unregistered}; register them in flip.py or pass "
        "ExtraColumnRule entries"
    )
    for pair in BOUND_PAIRS:
        assert (pair.lower in columns) == (pair.upper in columns), (
            f"confidence bounds {pair.lower} and {pair.upper} must be present together"
        )
    return {column: rules[column] for column in columns if column in rules}