modeling_input_data¶

@classmethod
def from_files(
    cls,
    response_path: str,
    predictors_path: str,
    perturbed_tf: str,
    feature_col: str = "target_symbol",
    feature_blacklist_path: str | None = None,
    top_n: int = 600,
) -> "ModelingInputData":
    """
    Load response and predictor data from files. This would be considered an
    overloaded constructor in other languages. The input files must be able to be
    read into objects that satisfy the __init__ method -- see __init__ docs.

    :param response_path: Path to the response file (CSV).
    :param predictors_path: Path to the predictors file (CSV).
    :param perturbed_tf: The perturbed TF.
    :param feature_col: The column name representing features.
    :param feature_blacklist_path: Path to a file containing a list of features to
        exclude.
    :param top_n: Maximum number of features for top-n selection.
    :return: An instance of ModelingInputData.
    :raises FileNotFoundError: If the response or predictor files are missing.

    """
    if not os.path.exists(response_path):
        raise FileNotFoundError(f"Response file '{response_path}' does not exist.")
    if not os.path.exists(predictors_path):
        raise FileNotFoundError(
            f"Predictors file '{predictors_path}' does not exist."
        )

    response_df = pd.read_csv(response_path)
    predictors_df = pd.read_csv(predictors_path)

    # Load feature blacklist if provided
    feature_blacklist: list[str] = []
    if feature_blacklist_path:
        if not os.path.exists(feature_blacklist_path):
            raise FileNotFoundError(
                f"Feature blacklist file '{feature_blacklist_path}' does not exist."
            )
        with open(feature_blacklist_path) as f:
            feature_blacklist = [line.strip() for line in f if line.strip()]

    return cls(
        response_df,
        predictors_df,
        perturbed_tf,
        feature_col,
        feature_blacklist,
        top_n,
    )

def get_modeling_data(
    self,
    formula: str,
    add_row_max: bool = False,
    drop_intercept: bool = False,
    scale_by_std: bool = False,
) -> pd.DataFrame:
    """
    Get the predictors for modeling, optionally adding a row-wise max feature.

    :param formula: The formula to use for modeling.
    :param add_row_max: Whether to add a row-wise max feature to the predictors.
    :param drop_intercept: If `drop_intercept` is True, "-1" will be appended to
        the formula string. This will drop the intercept (constant) term from
        the model matrix output by patsy.dmatrix. Default is `False`.
    :param scale_by_std: If True, scale the design matrix by standard deviation
        using StandardScaler(with_mean=False, with_std=True). The data is NOT
        centered, so the estimator should still fit an intercept
        (fit_intercept=True).
    :return: The design matrix for modeling. self.response_df can be used for the
        response variable.

    :raises ValueError: If the formula is not provided
    :raises PatsyError: If there is an error in creating the model matrix

    """
    if not formula:
        raise ValueError("Formula must be provided for modeling.")

    if drop_intercept:
        logger.info("Dropping intercept from the patsy model matrix")
        formula += " - 1"

    predictors_df = self.predictors_df  # Apply top-n feature mask

    # Add row-wise max feature if requested
    if add_row_max:
        predictors_df["row_max"] = predictors_df.max(axis=1)

    # Create a design matrix using patsy
    try:
        design_matrix = dmatrix(
            formula,
            data=predictors_df,
            return_type="dataframe",
            NA_action="raise",
        )
    except PatsyError as exc:
        logger.error(
            f"Error in creating model matrix with formula '{formula}': {exc}"
        )
        raise

    if scale_by_std:
        logger.info("Center matrix = `False`. Scale matrix = `True`")
        scaler = StandardScaler(with_mean=False, with_std=True)
        scaled_values = scaler.fit_transform(design_matrix)
        design_matrix = pd.DataFrame(
            scaled_values, index=design_matrix.index, columns=design_matrix.columns
        )

    logger.info(f"Design matrix columns: {list(design_matrix.columns)}")

    return design_matrix