code: def logistic_regression(
      genotype_df: DataFrame,
      phenotype_df: pd.DataFrame,
      covariate_df: pd.DataFrame = pd.DataFrame({}),
      offset_df: pd.DataFrame = pd.DataFrame({}),
      correction: str = correction_approx_firth,
      pvalue_threshold: float = 0.05,
      contigs: Optional[List[str]] = None,
      add_intercept: bool = True,
      values_column: str = "values",
      dt: type = np.float64,
      intersect_samples: bool = False,
      genotype_sample_ids: Optional[List[str]] = None,
  ) -> DataFrame:
      """
      Uses logistic regression to test for association between genotypes and one or more binary
      phenotypes. This is a distributed version of the method from regenie:
      https://www.nature.com/articles/s41588-021-00870-7
  
      Implementation details:
  
      On the driver node, we fit a logistic regression model based on the covariates for each
      phenotype:
  
      .. math::
          logit(y) \sim C
  
      where :math:`y` is a phenotype vector and :math:`C` is the covariate matrix.
  
      We compute the probability predictions :math:`\hat{y}` and broadcast the residuals (:math:`y - \hat{y}`),
      :math:`\gamma` vectors (where :math:`\gamma = \hat{y} * (1 - \hat{y})`), and
      :math:`(C^\intercal \gamma C)^{-1}` matrices. In each task, we then adjust the new genotypes based on the null fit,
      perform a score test as a fast scan for potentially significant variants, and then test variants with p-values below
      a threshold using a more selective, more expensive test.
  
      Args:
          genotype_df : Spark DataFrame containing genomic data
          phenotype_df : Pandas DataFrame containing phenotypic data
          covariate_df : An optional Pandas DataFrame containing covariates
          offset_df : An optional Pandas DataFrame containing the phenotype offset. This value will be used
                      as an offset in the covariate only and per variant logistic regression models. The ``offset_df`` may
                      have one or two levels of indexing. If one level, the index should be the same as the ``phenotype_df``.
                      If two levels, the level 0 index should be the same as the ``phenotype_df``, and the level 1 index
                      should be the contig name. The two level index scheme allows for per-contig offsets like
                      LOCO predictions from GloWGR.
          correction : Which test to use for variants that meet a significance threshold for the score test. Supported
                       methods are ``none`` and ``approx-firth``.
          pvalue_threshold : Variants with a pvalue below this threshold will be tested using the ``correction`` method.
          contigs : When using LOCO offsets, this parameter indicates the contigs to analyze. You can use this parameter to limit the size of the broadcasted data, which may
                    be necessary with large sample sizes. If this parameter is omitted, the contigs are inferred from
                    the ``offset_df``.
          add_intercept : Whether or not to add an intercept column to the covariate DataFrame
          values_column : A column name or column expression to test with linear regression. If a column name is provided,
                          ``genotype_df`` should have a column with this name and a numeric array type. If a column expression
                          is provided, the expression should return a numeric array type.
          dt : The numpy datatype to use in the linear regression test. Must be ``np.float32`` or ``np.float64``.
  
      Returns:
          A Spark DataFrame that contains
  
          - All columns from ``genotype_df`` except the ``values_column`` and the ``genotypes`` column if one exists
          - ``effect``: The effect size (if approximate Firth correction was applied)
          - ``stderror``: Standard error of the effect size (if approximate Firth correction was applied)
          - ``correctionSucceeded``: Whether the correction succeeded (if the correction test method is not ``none``).
            ``True`` if succeeded, ``False`` if failed, ``null`` if correction was not applied.
          - ``chisq``: The chi squared test statistic according to the score test or the correction method
          - ``pvalue``: p-value estimated from the test statistic
          - ``phenotype``: The phenotype name as determined by the column names of ``phenotype_df``
      """
  
      spark = genotype_df.sql_ctx.sparkSession
      gwas_fx._check_spark_version(spark)
      gwas_fx._validate_covariates_and_phenotypes(
          covariate_df, phenotype_df, is_binary=True
      )
      sql_type = gwas_fx._regression_sql_type(dt)
      genotype_df = gwas_fx._prepare_genotype_df(
          genotype_df, values_column, sql_type
      )
      base_result_fields = [
          StructField("chisq", sql_type),
          StructField("pvalue", sql_type),
          StructField("phenotype", StringType()),
      ]
      if correction == correction_approx_firth:
          result_fields = [
              StructField("effect", sql_type),
              StructField("stderror", sql_type),
              StructField("correctionSucceeded", BooleanType()),
          ] + base_result_fields
      elif correction == correction_none:
          result_fields = base_result_fields
      else:
          raise ValueError(
              f"Only supported correction methods are '{correction_none}' and '{correction_approx_firth}'"
          )
  
      result_struct = gwas_fx._output_schema(
          genotype_df.schema.fields, result_fields
      )
  
      gt_indices_to_drop = None
      if (
          intersect_samples
      ):  # TODO intersect samples accross pdf,covs,and provided genotype_sample_ids
          gt_indices_to_drop = _get_indices_to_drop(
              phenotype_df, genotype_sample_ids
          )
          if not offset_df.empty:
              if offset_df.index.nlevels == 1:  # Indexed by sample id
                  offset_df = offset_df.reindex(phenotype_df.index)
              elif (
                  offset_df.index.nlevels == 2
              ):  # Indexed by sample id and contig
                  offset_df = offset_df[
                      offset_df.index.get_level_values(0).isin(
                          phenotype_df.index
                      )
                  ]
  
      C = covariate_df.to_numpy(dt, copy=True)
      if add_intercept:
          C = gwas_fx._add_intercept(C, phenotype_df.shape[0])
      Y = phenotype_df.to_numpy(dt, copy=True)
      Y_mask = ~(np.isnan(Y))
      np.nan_to_num(Y, copy=False)
  
      if correction == correction_approx_firth:
          Q = np.linalg.qr(C)[0]
      else:
          Q = None
  
      state = _create_log_reg_state(
          spark,
          phenotype_df,
          offset_df,
          sql_type,
          C,
          correction,
          add_intercept,
          contigs,
      )
  
      phenotype_names = phenotype_df.columns.to_series().astype("str")
  
      def map_func(pdf_iterator):
          for pdf in pdf_iterator:
              yield gwas_fx._loco_dispatch(
                  pdf,
                  state,
                  _logistic_regression_inner,
                  C,
                  Y,
                  Y_mask,
                  Q,
                  correction,
                  pvalue_threshold,
                  phenotype_names,
                  gt_indices_to_drop,
              )
  
      return genotype_df.mapInPandas(map_func, result_struct)
  
imports: from pyspark.sql.types import ArrayType, BooleanType, StringType, StructField, DataType, StructType
  from typing import Any, List, Optional, Dict, Union
  import pandas
  import numpy
  from pyspark.sql import DataFrame, SparkSession
  import statsmodels.api
  from dataclasses import dataclass
  from typeguard import typechecked
  from nptyping import Float, NDArray, Int32
  from scipy import stats
  import opt_einsum
  from .None import functions
  from .None import approx_firth
  from .functions import _VALUES_COLUMN_NAME, _get_indices_to_drop
  from ..wgr.wgr_functions import reshape_for_gwas
label_desc: Use image node_id 1 for calling a global function with example usage: logistic_regression(genotype_df, phenotype_df, covariate_df, offset_df, correction, pvalue_threshold, contigs, add_intercept, values_column, dt, intersect_samples, genotype_sample_ids) and returns: genotype_df