API Reference

Note: This page is auto-generated using mkdocstrings. If you are viewing this on GitHub, please visit the Online Documentation to see the full API reference.

Core

varframe.VarFrame

Bases: DataFrame

A pandas DataFrame subclass with variable metadata and helper methods.

VarFrame behaves exactly like a pandas DataFrame, but also maintains a registry of variable class definitions. This enables: - Filtering columns by variable type (BaseVariable vs DerivedVariable) - Accessing variable metadata and descriptions - Indexing by variable class (e.g., vf[Gap])

Attributes:

Name	Type	Description
_variables	List[Type]	List of variable classes (stored in _metadata).

Example

vf = VarFrame(df_raw, [Lap, Gap, GapDelta]) vf.head() # Normal DataFrame operations work vf.filter_by_type(DerivedVariable) # Variable-aware filtering vf[Gap] # Access by variable class

Source code in varframe/dataframe.py

class VarFrame(pd.DataFrame):
    """
    A pandas DataFrame subclass with variable metadata and helper methods.

    VarFrame behaves exactly like a pandas DataFrame, but also
    maintains a registry of variable class definitions. This enables:
    - Filtering columns by variable type (BaseVariable vs DerivedVariable)
    - Accessing variable metadata and descriptions
    - Indexing by variable class (e.g., ``vf[Gap]``)

    Attributes:
        _variables (List[Type]): List of variable classes (stored in _metadata).

    Example:
        >>> vf = VarFrame(df_raw, [Lap, Gap, GapDelta])
        >>> vf.head()              # Normal DataFrame operations work
        >>> vf.filter_by_type(DerivedVariable)  # Variable-aware filtering
        >>> vf[Gap]                # Access by variable class
    """

    # Tell pandas to preserve these attributes during operations
    _metadata = ["_variables", "_df_raw", "name"]

    def __init__(
        self,
        df_raw: Optional[pd.DataFrame] = None,
        variables: Optional[VariableList] = None,
        compute: bool = True,
        auto_resolve: bool = True,
        name: str = "varframe",
        **kwargs: Any,
    ) -> None:
        """
        Initialize a VarFrame from raw data.

        Args:
            df_raw: The raw DataFrame to process. Stored internally for
                future resolve() calls. If None, creates an empty DataFrame.
            variables: List of variable classes to compute. If None or empty,
                no variables are computed initially (use resolve() later).
            compute: If True (default), compute the specified variables immediately.
                If False, just store df_raw without computing anything.
            auto_resolve: If True (default) and compute=True, automatically
                resolve all dependencies. If False, variables must be in
                correct dependency order.
            **kwargs: Additional arguments passed to pd.DataFrame.__init__.

        Example:
            >>> # Compute specific variables
            >>> vf = VarFrame(df_raw, [Lap, Gap, GapDelta])
            >>>
            >>> # Auto-resolve: just specify final variables
            >>> vf = VarFrame(df_raw, [PredictedGapDelta])
            >>>
            >>> # Store df_raw, compute nothing (use resolve() later)
            >>> vf = VarFrame(df_raw, compute=False)
            >>> vf.resolve(PredictedGapDelta)
        """
        # Store raw DataFrame
        self._df_raw: Optional[pd.DataFrame] = df_raw
        self._variables: VariableList = []

        # Determine which variables to compute
        vars_to_compute: VariableList = []
        if variables and compute:
            if auto_resolve:
                vars_to_compute = resolve_dependencies(variables)
            else:
                vars_to_compute = list(variables)

            # Filter out lazy variables from initial computation
            vars_to_compute = [
                v
                for v in vars_to_compute
                if not (issubclass(v, DerivedVariable) and v.lazy)
            ]

        # Initialize DataFrame with proper index
        if df_raw is not None:
            super().__init__(index=df_raw.index, **kwargs)
        else:
            super().__init__(**kwargs)

        self.name = name

        # Compute variables if requested
        if vars_to_compute and df_raw is not None:
            for var in vars_to_compute:
                if issubclass(var, BaseVariable):
                    self[var.name] = var.compute(df_raw)
                else:
                    self[var.name] = var.compute(self)

                if var not in self._variables:
                    self._variables.append(var)

            # Register remaining variables (lazy ones or non-computed) that were passed in
            if variables:
                 for var in variables:
                     if var not in self._variables:
                         self._variables.append(var)

        elif variables:
            # Store variable list even if not computing
            self._variables = list(variables) if variables else []
            # Ensure no duplicates if variables were passed
            self._variables = list(dict.fromkeys(self._variables))

    @property
    def _constructor(self) -> Type[VarFrame]:
        """Return the constructor for DataFrame operations to preserve type."""
        return _VarFrameInternal

    @property
    def variables(self) -> VariableList:
        """Get the list of variable classes."""
        return self._variables

    @variables.setter
    def variables(self, value: VariableList) -> None:
        """Set the list of variable classes."""
        self._variables = value

    @property
    def df_raw(self) -> Optional[pd.DataFrame]:
        """Get the raw DataFrame (if stored)."""
        return self._df_raw

    @df_raw.setter
    def df_raw(self, value: Optional[pd.DataFrame]) -> None:
        """Set the raw DataFrame."""
        self._df_raw = value

    def filter_by_type(
        self, variable_type: Type[Union[BaseVariable, DerivedVariable]]
    ) -> VarFrame:
        """
        Filter to include only columns of a specific variable type.

        Args:
            variable_type: The base class to filter by (BaseVariable or DerivedVariable).

        Returns:
            A new VarFrame containing only columns whose variables
            are subclasses of the specified type.

        Example:
            >>> derived_df = vf.filter_by_type(DerivedVariable)
            >>> base_df = vf.filter_by_type(BaseVariable)
        """
        filtered_vars = [v for v in self._variables if issubclass(v, variable_type)]
        filtered_names = [v.name for v in filtered_vars]
        result = self[filtered_names].copy()
        result._variables = filtered_vars
        return result

    def get_variable(self, name: str) -> Optional[VariableType]:
        """
        Retrieve a variable class by its name.

        Args:
            name: The name of the variable to retrieve.

        Returns:
            The variable class if found, None otherwise.
        """
        for var in self._variables:
            if var.name == name:
                return var
        return None

    def list_variables(self) -> List[str]:
        """
        Get a list of all variable names.

        Returns:
            List of variable names in order.
        """
        return [v.name for v in self._variables]

    def describe_variables(self) -> pd.DataFrame:
        """
        Generate a summary DataFrame describing all variables.

        Returns:
            A DataFrame with columns: name, type, dtype, description, etc.
        """
        data = [var.info() for var in self._variables]
        return pd.DataFrame(data)

    def view(
        self,
        include: Optional[List[str]] = None,
        variables: Optional[VariableList] = None,
    ) -> pd.DataFrame:
        """
        Create a DataFrame view containing only specific variables.

        Args:
            include: List of variable categories to include.
                Options: 'base', 'derived', 'lazy', 'model', 'all'.
                Defaults to ['all'] if both include and variables are None.
            variables: Explicit list of variable classes to include.

        Returns:
            A pandas DataFrame with the requested variables. 
            Lazy variables will be computed on-demand for this view.

        Example:
            >>> vf.view(include=['base', 'lazy'])
            >>> vf.view(variables=[LazySum])
        """
        target_vars = []

        # 1. Handle 'variables' argument
        if variables:
            target_vars.extend(variables)

        # 2. Handle 'include' argument
        if include:
            for category in include:
                if category == "all":
                    target_vars.extend(self._variables)
                elif category == "base":
                    target_vars.extend(
                        [v for v in self._variables if issubclass(v, BaseVariable)]
                    )
                elif category == "derived":
                    # Non-lazy derived
                    target_vars.extend(
                        [
                            v
                            for v in self._variables
                            if issubclass(v, DerivedVariable) and not v.lazy
                        ]
                    )
                elif category == "lazy":
                    # Lazy derived
                    target_vars.extend(
                        [
                            v
                            for v in self._variables
                            if issubclass(v, DerivedVariable) and v.lazy
                        ]
                    )
                elif category == "model":
                    target_vars.extend(
                        [
                            v
                            for v in self._variables
                            if hasattr(v, "model_class") and v.model_class
                        ]
                    )

        # Default to 'all' if nothing specified
        if not target_vars:
            target_vars = list(self._variables)

        # Remove duplicates while preserving order
        target_vars = list(dict.fromkeys(target_vars))

        # 3. Construct DataFrame
        data = {}
        for var in target_vars:
            # If standard column, use it (zero copy if possible)
            if var.name in self.columns:
                data[var.name] = self[var.name]
            else:
                # Must be lazy or missing -> Compute it
                data[var.name] = var.compute(self)

        return pd.DataFrame(data, index=self.index)

    def __getitem__(self, key: Union[str, VariableType, List, slice, pd.Series]) -> Any:
        """
        Access columns by name, variable class, or standard DataFrame indexing.

        Extends pandas indexing to support variable classes directly.

        Args:
            key: Column name (str), variable class, list of columns,
                 boolean Series, or slice.

        Returns:
            Series for single column, DataFrame for multiple columns.

        Example:
            >>> vf["gap"]      # By string name
            >>> vf[Gap]        # By variable class
            >>> vf[[Lap, Gap]] # Multiple variable classes
        """
        # Handle variable class
        if isinstance(key, type) and issubclass(key, (BaseVariable, DerivedVariable)):
            # Check if it's lazy and missing (Compute just-in-time)
            if (
                issubclass(key, DerivedVariable)
                and key.lazy
                and key.name not in self.columns
            ):
                 return key.compute(self)

            return super().__getitem__(key.name)

        # Handle list of variable classes
        if isinstance(key, list) and len(key) > 0:
            if isinstance(key[0], type) and issubclass(
                key[0], (BaseVariable, DerivedVariable)
            ):
                names = [k.name for k in key]
                return super().__getitem__(names)

        # Default pandas behavior
        try:
            return super().__getitem__(key)
        except KeyError:
            # Check for lazy variable (string access)
            if isinstance(key, str):
                var = self.get_variable(key)

                # Case 1: Variable is registered but not in columns (Lazy or missing)
                if var and issubclass(var, DerivedVariable) and var.lazy:
                    return var.compute(self)

            # Re-raise if not handled
            raise

    def __repr__(self) -> str:
        """Return a string representation."""
        var_info = f"Variables: {len(self._variables)} ({self.list_variables()})"
        df_repr = super().__repr__()
        return f"{var_info}\n{df_repr}"

    # ------------------- ML Compatibility Methods -------------------

    def to_pandas(self) -> pd.DataFrame:
        """
        Convert to a plain pandas DataFrame.

        Use this before passing to ML libraries that may have issues
        with DataFrame subclasses (e.g., pickle, joblib, some sklearn pipelines).

        Returns:
            A plain pandas DataFrame with the same data (no variable metadata).

        Example:
            >>> plain_df = vf.to_pandas()
            >>> model.fit(plain_df, y)
        """
        return pd.DataFrame(self)

    def to_ml(self) -> pd.DataFrame:
        """
        Alias for to_pandas(). Explicit conversion for ML pipelines.

        Use this to clearly indicate the DataFrame is being prepared
        for machine learning operations.

        Returns:
            A plain pandas DataFrame suitable for ML libraries.

        Example:
            >>> X_train = vf.to_ml()
            >>> model.fit(X_train, y_train)
        """
        return pd.DataFrame(self)

    def to_csv(
        self,
        path_or_buf: Optional[Union[str, Any]] = None,
        *args: Any,
        include: Optional[List[str]] = None,
        variables: Optional[VariableList] = None,
        **kwargs: Any,
    ) -> Optional[str]:
        """
        Write object to a comma-separated values (csv) file.

        Enhancements over pandas.to_csv:
        - Supports `include` and `variables` to compute lazy variables on-the-fly.
        - Warns if registered variables are not included in the export.
        - Defaults to {self.name}.csv if path is not provided.
        """
        from varframe.config import VFConfig

        # 1. Resolve Data
        if include or variables:
            df_to_export = self.view(include=include, variables=variables)
        else:
            df_to_export = self

        # 2. Prevent implicit data loss (Warn if variables are missing)
        # Check which variables are in the export
        exported_cols = set(df_to_export.columns)
        missing_vars = [
            v.name for v in self._variables 
            if v.name not in exported_cols and v.name not in self.columns
        ]

        # Also check for variables present in self but not in export (if view filtered them)
        excluded_present_vars = [
             v.name for v in self._variables
             if v.name in self.columns and v.name not in exported_cols
        ]

        if missing_vars:
             path_str = str(path_or_buf) if path_or_buf else "output"
             VFConfig.warn(
                ImplicitOperation.ADD_VARIABLE_COMPUTE, # Reusing generic warning op
                f"Exporting to {path_str} without computing lazy variables: {missing_vars}. "
                "Use `include=['all']` or `variables=[...]` to compute them during export.",
             )

        # 3. Resolve Path
        if path_or_buf is None:
            path_or_buf = f"{self.name}.csv"

        if hasattr(df_to_export, "to_pandas"):
            return df_to_export.to_pandas().to_csv(path_or_buf, *args, **kwargs)
        else:
            return df_to_export.to_csv(path_or_buf, *args, **kwargs)

    def to_parquet(
        self,
        path: Optional[Union[str, Any]] = None,
        *args: Any,
        include: Optional[List[str]] = None,
        variables: Optional[VariableList] = None,
        **kwargs: Any,
    ) -> Optional[bytes]:
        """
        Write object to a binary parquet file.

        Enhancements over pandas.to_parquet:
        - Supports `include` and `variables` to compute lazy variables on-the-fly.
        - Warns if registered variables are not included in the export.
        - Defaults to {self.name}.parquet if path is not provided.
        - Saves variable names in file metadata.
        """
        import json
        from varframe.config import VFConfig

        # 1. Resolve Data
        if include or variables:
            df_to_export = self.view(include=include, variables=variables)
        else:
            df_to_export = self

        # 2. Prevent implicit data loss
        exported_cols = set(df_to_export.columns)
        missing_vars = [
            v.name for v in self._variables 
            if v.name not in exported_cols and v.name not in self.columns
        ]

        if missing_vars:
             path_str = str(path) if path else "output"
             VFConfig.warn(
                ImplicitOperation.ADD_VARIABLE_COMPUTE,
                f"Exporting to {path_str} without computing lazy variables: {missing_vars}. "
                "Use `include=['all']` or `variables=[...]` to compute them during export.",
             )

        # 3. Resolve Path
        if path is None:
            path = f"{self.name}.parquet"

        # 4. Prepare Metadata (Strategy B)
        # Get existing keyword args or creating new dict
        # pyarrow.Table.from_pandas uses 'preserve_index' etc.
        # to_parquet kwargs are passed to the engine. 
        # For pyarrow engine (default), we can't easily inject metadata via to_parquet directly 
        # because pandas abstracts this. 
        # WORKAROUND: We will use table properties if using pyarrow, 
        # but to keep it simple and pandas-compliant, we might need a distinct approach.
        # Actually, pandas >= 1.0 does not easily support custom metadata in to_parquet 
        # without dropping down to pyarrow directly.

        try:
            import pyarrow as pa
            import pyarrow.parquet as pq

            # Convert to Table
            # Handle VarFrame or plain DataFrame
            if hasattr(df_to_export, "to_pandas"):
                df_plain = df_to_export.to_pandas()
            else:
                df_plain = df_to_export

            table = pa.Table.from_pandas(df_plain)

            # Add Metadata
            custom_meta = {
                "varframe_variables": json.dumps([v.name for v in self._variables])
            }

            # Add Hash Metadata
            try:
                hashes = {v.name: v.get_hash_components() for v in self._variables}
                custom_meta["varframe_hashes"] = json.dumps(hashes)
            except Exception as e:
                VFConfig.warn(
                    ImplicitOperation.ADD_VARIABLE_COMPUTE,
                    f"Failed to compute variable hashes for export: {e}"
                )
            # Combine with existing metadata
            existing_meta = table.schema.metadata or {}
            combined_meta = {**existing_meta, **{k.encode(): v.encode() for k, v in custom_meta.items()}}
            table = table.replace_schema_metadata(combined_meta)

            # Write
            pq.write_table(table, path, *args, **kwargs)
            return None

        except ImportError:
            # Fallback for when pyarrow is not available or user wants different engine
             VFConfig.warn(
                 ImplicitOperation.ADD_VARIABLE_COMPUTE,
                 "Pyarrow not found or failed. Exporting without VarFrame metadata."
             )
             if hasattr(df_to_export, "to_pandas"):
                 return df_to_export.to_pandas().to_parquet(path, *args, **kwargs)
             else:
                 return df_to_export.to_parquet(path, *args, **kwargs)

    def add_variables(
        self,
        *variables: VariableType,
        compute: bool = True,
        suppress_warnings: bool = False,
    ) -> VarFrame:
        """
        Register and optionally compute new variables (in-place).

        Args:
            *variables: Variable classes to add.
            compute: If True (default), compute variables immediately.
                If False, just register them (must already exist in DataFrame).
            suppress_warnings: If True, suppress warnings for this call only.

        Returns:
            Self, for method chaining.

        Raises:
            KeyError: If compute=True and dependencies are missing.
            RuntimeError: If implicit usage is disabled in VFConfig.
            ValueError: If compute=True and adding a BaseVariable (must come from raw).
        """
        for var in variables:
            if compute:
                # --- Compute Mode ---
                if var.name in self.columns:
                    continue  # Already exists

                # Note: Explicit addition via add_variables does not trigger implicit warnings.

                if issubclass(var, BaseVariable):
                    raise ValueError(
                        f"Cannot add BaseVariable '{var.name}' to existing DataFrame. "
                        "BaseVariables can only be computed from raw data."
                    )
                else:
                    self[var.name] = var.compute(self)

            else:
                # --- No Compute Mode (Registration Only) ---
                # Explicit registration - no warning needed.
                pass

            # Common: Add to registry
            if var not in self._variables:
                self._variables.append(var)

        return self

    def add_variable(
        self,
        *variables: VariableType,
        compute: bool = True,
        suppress_warnings: bool = False,
    ) -> VarFrame:
        """
        Alias for add_variables.
        """
        return self.add_variables(
            *variables, compute=compute, suppress_warnings=suppress_warnings
        )

    def resolve(
        self,
        *target_variables: VariableType,
        suppress_warnings: bool = False,
    ) -> VarFrame:
        """
        Resolve and compute target variables with automatic dependency resolution.

        This method automatically determines all missing dependencies for the
        target variables, computes them in the correct DAG order, and adds
        them to the DataFrame.

        Args:
            *target_variables: Variable classes to resolve and compute.
            suppress_warnings: If True, suppress warnings for this call only.

        Returns:
            Self, for method chaining.

        Raises:
            ValueError: If circular dependencies are detected.
            ValueError: If a BaseVariable is needed but df_raw is not available.
            RuntimeError: If implicit operations are disabled in VFConfig.

        Example:
            >>> vf = VarFrame(df_raw, [Lap, Gap])
            >>> vf.resolve(PredictedGapDelta)
        """
        # Resolve all dependencies
        all_vars = resolve_dependencies(list(target_variables))

        # Filter to only missing variables
        # Filter to only missing variables
        missing_vars = [v for v in all_vars if v.name not in self.columns]

        # Split into compute-now vs lazy
        lazy_vars = [
            v for v in missing_vars if issubclass(v, DerivedVariable) and v.lazy
        ]
        missing_vars = [v for v in missing_vars if v not in lazy_vars]

        # Check for missing BaseVariables - need df_raw to compute them
        missing_base = [v for v in missing_vars if issubclass(v, BaseVariable)]
        if missing_base and self._df_raw is None:
            names = [v.name for v in missing_base]
            raise ValueError(
                f"Cannot resolve BaseVariables {names}: no raw DataFrame stored. "
                "Pass df_raw to VarFrame() or set vf.df_raw to enable."
            )

        # Warn about implicit computation of missing variables
        if missing_vars and not suppress_warnings:
            var_names = [v.name for v in missing_vars]
            VFConfig.check_permission(
                ImplicitOperation.ADD_VARIABLE_COMPUTE,
                f"Cannot resolve variables {var_names}. "
                "Set VFConfig.allow_implicit_compute = True to enable.",
            )
            VFConfig.warn(
                ImplicitOperation.ADD_VARIABLE_COMPUTE,
                f"Implicitly computing {len(missing_vars)} variable(s) not in _variables: {var_names}",
                stacklevel=3,
            )

        # Register lazy variables (so get_variable works)
        for var in lazy_vars:
            if var not in self._variables:
                self._variables.append(var)

        if suppress_warnings:
            context = VFConfig.suppress_warnings()
        else:
            context = VFConfig.null_context()

        with context:
            # Compute missing variables in order
            for var in missing_vars:
                if issubclass(var, BaseVariable):
                    self[var.name] = var.compute(self._df_raw)
                else:
                    self[var.name] = var.compute(self)

                if var not in self._variables:
                    self._variables.append(var)

        return self

    def explain_calculation(
        self,
        target_variables: VariableList,
        legend: bool = False,
    ) -> None:
        """
        Explain the calculation plan for the given variables given the current DataFrame state.

        Prints a color-coded list indicating:
        - Variable Type (Base, Derived, Model)
        - Calculation Status (Ready vs Needs Calculation)
        - Warnings (Implicit computation, Auto-training, Inference)

        Args:
            target_variables: List of variable classes to explain.
            legend: If True, print a legend for the warning codes.
        """
        from varframe.variables import BaseVariable
        from varframe.config import VFConfig, ImplicitOperation

        # ANSI color codes
        GREEN = "\033[92m"
        YELLOW = "\033[93m"
        PURPLE = "\033[95m"
        ORANGE = "\033[33m"
        GREY = "\033[90m"
        RESET = "\033[0m"

        resolved = resolve_dependencies(target_variables)

        names = ", ".join(v.name for v in target_variables)
        print(f"Calculation Plan for {names}:")

        used_codes = set()
        # Mapping: Code -> (Description, ConfigCheck)
        warning_defs = {
            "I": ("Implicit Compute", VFConfig.warn_add_variable_compute),
            "T": ("Auto-Train Model", VFConfig.warn_train_model),
            "M": ("Model Inference", VFConfig.warn_infer_model),
        }

        for i, v in enumerate(resolved, 1):
            # 1. Determine Type
            if hasattr(v, "model_class") and v.model_class:
                var_type = "model_prediction"
                color = PURPLE
            elif issubclass(v, BaseVariable):
                var_type = "base"
                color = GREEN
            else:
                var_type = "derived"
                color = YELLOW

            # 2. Determine Status
            exists = v.name in self.columns
            status_icon = "✅" if exists else "⏳"

            # 3. Predict Warnings
            current_codes = []
            if not exists:
                # Implicit Variable Warning
                if v not in self._variables and warning_defs["I"][1]:
                     current_codes.append("I")

                # Model Warnings
                if var_type == "model_prediction" and v.model_class:
                    if not v.model_class.is_trained and warning_defs["T"][1]:
                         current_codes.append("T")
                    if warning_defs["M"][1]:
                        current_codes.append("M")

            used_codes.update(current_codes)

            # Formatting
            line = f"  {i}. {status_icon} {color}{v.name}{RESET} ({var_type})"
            if current_codes:
                codes_str = ",".join(current_codes)
                # Subtle grey for the warning hints
                line += f" {GREY}⚠ [{codes_str}]{RESET}"

            print(line)

        if legend and used_codes:
            print("\nLegend:")
            for code in sorted(used_codes):
                desc = warning_defs[code][0]
                print(f"  {GREY}⚠ [{code}]{RESET} : {desc}")
        print()

    @classmethod
    def from_pandas(
        cls,
        df: pd.DataFrame,
        variables: Optional[VariableList] = None,
        df_raw: Optional[pd.DataFrame] = None,
        name: str = "varframe",
    ) -> VarFrame:
        """
        Create a VarFrame from a plain pandas DataFrame.

        Use this to re-wrap a DataFrame after ML operations or when
        loading data that was previously converted with to_pandas().

        Args:
            df: The pandas DataFrame to convert (already processed data).
            variables: List of variable classes to associate with the DataFrame.
            df_raw: Optional raw DataFrame to store for future resolve() calls.

        Returns:
            A new VarFrame with the given data and variables.

        Example:
            >>> plain_df = pd.read_pickle("data.pkl")
            >>> vf = VarFrame.from_pandas(plain_df, variables=[Lap, Gap])
        """
        vf = cls(df_raw=None, variables=None, compute=False, name=name)
        for col in df.columns:
            vf[col] = df[col].values
        vf.index = df.index
        vf._variables = list(variables) if variables else []
        vf._df_raw = df_raw
        return vf

    @staticmethod
    def _discover_all_variables() -> Dict[str, List[VariableType]]:
        """
        Recursively discover all BaseVariable and DerivedVariable subclasses.

        Returns:
            Dict mapping variable name to a list of class definitions (to handle redefinitions).
        """
        discovered = {}

        def _scan(cls):
            # Add self if it's a concrete variable (has a name)
            if hasattr(cls, "name") and cls.name:
                if cls.name not in discovered:
                    discovered[cls.name] = []
                # Avoid duplicates in the list
                if cls not in discovered[cls.name]:
                     discovered[cls.name].append(cls)

            # Recurse
            for sub in cls.__subclasses__():
                _scan(sub)

        _scan(BaseVariable)
        _scan(DerivedVariable)
        return discovered

    @classmethod
    def load_csv(
        cls, 
        path_or_buf: Union[str, Any], 
        variables: Optional[List[VariableType]] = None,
        exclude: Optional[List[VariableType]] = None,
        discard_unmatched: bool = True,
        ambiguity: Optional[Dict[str, VariableType]] = None,
        **kwargs: Any
    ) -> VarFrame:
        """
        Load a VarFrame from a CSV file, automatically discovering variables.

        This method scans the current Python environment for variable definitions
        that match the columns in the CSV.

        Args:
            path_or_buf: Path to the CSV file.
            variables: Whitelist of variable classes to load. If provided, only 
                these variables will be matched. Acts as implicit disambiguation.
            exclude: Blacklist of variable classes to exclude from loading.
                Cannot be used together with `variables`.
            discard_unmatched: If True (default), columns not matching any known
                variable are dropped. If False, they are kept as plain columns.
            ambiguity: Dict mapping variable names to specific classes for 
                disambiguation when multiple definitions exist with the same name.
            **kwargs: Arguments passed to pd.read_csv.

        Returns:
            A reconstructed VarFrame.

        Raises:
            ValueError: If both `variables` and `exclude` are provided.
            AmbiguityError: If multiple variable definitions match a column name
                and no disambiguation is provided.
        """
        from varframe.exceptions import AmbiguityError

        # Validation
        if variables is not None and exclude is not None:
            raise ValueError("Cannot use both 'variables' and 'exclude' parameters together.")

        df = pd.read_csv(path_or_buf, **kwargs)

        # Auto-discovery
        known_vars = cls._discover_all_variables()
        matched_vars = []

        # Build target names based on whitelist/blacklist
        if variables is not None:
            target_names = {v.name for v in variables}
            # Create a lookup for whitelisted variables by name
            whitelist_by_name: Dict[str, VariableType] = {v.name: v for v in variables}
        elif exclude is not None:
            exclude_names = {v.name for v in exclude}
            target_names = {col for col in df.columns if col not in exclude_names}
            whitelist_by_name = None
        else:
            target_names = set(df.columns)
            whitelist_by_name = None

        for col in df.columns:
            if col not in target_names:
                continue

            if col not in known_vars:
                continue

            candidates = known_vars[col]

            # If whitelist mode, use the whitelisted class directly
            if whitelist_by_name is not None:
                if col in whitelist_by_name:
                    matched_vars.append(whitelist_by_name[col])
                continue

            # Check for ambiguity
            if len(candidates) > 1:
                # Check if disambiguation provided via ambiguity parameter
                if ambiguity and col in ambiguity:
                    matched_vars.append(ambiguity[col])
                else:
                    raise AmbiguityError(
                        f"Ambiguous variable '{col}'. Multiple definitions found: "
                        f"{[c.__name__ for c in candidates]}. "
                        f"Use 'ambiguity' parameter to specify which to use, e.g.: "
                        f"ambiguity={{'{col}': {candidates[0].__name__}}}"
                    )
            else:
                matched_vars.append(candidates[0])

        # Handle unmatched columns
        mapped_names = {v.name for v in matched_vars}
        unmapped_cols = [col for col in df.columns if col not in mapped_names]

        if discard_unmatched:
            # Drop unmatched columns
            cols_to_keep = [v.name for v in matched_vars]
            df = df[[c for c in cols_to_keep if c in df.columns]]
        else:
            # Warn about unmapped columns (old behavior)
            if unmapped_cols:
                from varframe.config import VFConfig, ImplicitOperation
                VFConfig.warn(
                    ImplicitOperation.ADD_VARIABLE_NO_COMPUTE,
                    f"Loaded columns {unmapped_cols} do not match any known Variable class. "
                    "They will be loaded as plain pandas columns."
                )

        # Try to infer name from path
        name = "varframe"
        if isinstance(path_or_buf, str):
            import os
            name = os.path.splitext(os.path.basename(path_or_buf))[0]

        return cls.from_pandas(df, variables=matched_vars, name=name)

    @classmethod
    def load_parquet(
        cls, 
        path: Union[str, Any], 
        variables: Optional[List[VariableType]] = None,
        exclude: Optional[List[VariableType]] = None,
        discard_unmatched: bool = True,
        ambiguity: Optional[Dict[str, VariableType]] = None,
        **kwargs: Any
    ) -> VarFrame:
        """
        Load a VarFrame from a Parquet file, using metadata or auto-discovery.

        1. Checks for 'varframe_variables' metadata in the file.
        2. If found, looks up those variables in the environment.
        3. If not found, falls back to matching column names (like load_csv).

        Args:
            path: Path to the Parquet file.
            variables: Whitelist of variable classes to load. If provided, only 
                these variables will be matched. Acts as implicit disambiguation
                and overrides hash-based resolution.
            exclude: Blacklist of variable classes to exclude from loading.
                Cannot be used together with `variables`.
            discard_unmatched: If True (default), columns not matching any known
                variable are dropped. If False, they are kept as plain columns.
            ambiguity: Dict mapping variable names to specific classes for 
                disambiguation. Overrides hash-based resolution for specified names.
            **kwargs: Arguments passed to pyarrow/pandas read functions.

        Returns:
            A reconstructed VarFrame.

        Raises:
            ValueError: If both `variables` and `exclude` are provided.
        """
        import json

        # Validation
        if variables is not None and exclude is not None:
            raise ValueError("Cannot use both 'variables' and 'exclude' parameters together.")

        # ANSI Colors
        RED = "\033[91m"
        ORANGE = "\033[33m"
        WHITE = "\033[37m"
        RESET = "\033[0m"

        warnings_list = []

        # Try reading metadata first (requires pyarrow)
        file_hashes = {}
        var_names_from_meta = []
        all_columns = []

        try:
            import pyarrow.parquet as pq
            meta = pq.read_metadata(path)
            # Get column names from schema
            all_columns = list(meta.schema.names)
            if meta.metadata:
                if b'varframe_variables' in meta.metadata:
                    var_names_from_meta = json.loads(meta.metadata[b'varframe_variables'])
                if b'varframe_hashes' in meta.metadata:
                    file_hashes = json.loads(meta.metadata[b'varframe_hashes'])
        except ImportError:
            pass

        known_vars = cls._discover_all_variables()

        # Determine which columns to load based on parameters
        if variables is not None:
            # Whitelist mode - use provided classes directly
            target_names = [v.name for v in variables]
            whitelist_by_name: Dict[str, VariableType] = {v.name: v for v in variables}
        elif exclude is not None:
            exclude_names = {v.name for v in exclude}
            source_names = var_names_from_meta if var_names_from_meta else all_columns
            target_names = [n for n in source_names if n not in exclude_names]
            whitelist_by_name = None
        else:
            target_names = None  # Load all
            whitelist_by_name = None

        # Use pyarrow column selection for efficiency
        try:
            import pyarrow.parquet as pq
            if target_names is not None:
                # Only read requested columns
                cols_to_read = [c for c in target_names if c in all_columns]
                table = pq.read_table(path, columns=cols_to_read, **kwargs)
            else:
                table = pq.read_table(path, **kwargs)
            df = table.to_pandas()
        except ImportError:
            # Fallback to pandas (less efficient)
            df = pd.read_parquet(path, **kwargs)
            if target_names is not None:
                available = [c for c in target_names if c in df.columns]
                df = df[available]

        matched_vars = []

        # Determine variable names to look for
        if target_names is not None:
            names_to_resolve = target_names
        else:
            names_to_resolve = var_names_from_meta if var_names_from_meta else df.columns

        for name in names_to_resolve:
            # Handle case where name is in metadata but not in known_vars (not imported)
            if name not in known_vars:
                continue

            # If whitelist mode, use the whitelisted class directly
            if whitelist_by_name is not None:
                if name in whitelist_by_name:
                    matched_vars.append(whitelist_by_name[name])
                continue

            candidates = known_vars[name]

            # Check for multiple candidates (ambiguity)
            if len(candidates) > 1:
                # Require explicit disambiguation
                if ambiguity and name in ambiguity:
                    matched_vars.append(ambiguity[name])
                else:
                    from varframe.exceptions import AmbiguityError
                    raise AmbiguityError(
                        f"Ambiguous variable '{name}'. Multiple definitions found: "
                        f"{[c.__name__ for c in candidates]}. "
                        f"Use 'ambiguity' parameter to specify which to use, e.g.: "
                        f"ambiguity={{'{name}': {candidates[0].__name__}}}"
                    )
            else:
                # Single candidate - use it
                matched_vars.append(candidates[0])

        # Handle unmatched columns
        mapped_names = {v.name for v in matched_vars}
        unmapped_cols = [col for col in df.columns if col not in mapped_names]

        if discard_unmatched:
            # Drop unmatched columns
            cols_to_keep = [v.name for v in matched_vars]
            df = df[[c for c in cols_to_keep if c in df.columns]]
        else:
            # Warn about unmapped columns (old behavior)
            if unmapped_cols:
                from varframe.config import VFConfig, ImplicitOperation
                VFConfig.warn(
                    ImplicitOperation.ADD_VARIABLE_NO_COMPUTE,
                    f"Loaded columns {unmapped_cols} do not match any known Variable class. "
                    "They will be loaded as plain pandas columns."
                )

        # Try to infer name from path
        vf_name = "varframe"
        if isinstance(path, str):
            import os
            vf_name = os.path.splitext(os.path.basename(path))[0]

        # ------------------- Integrity Check -------------------
        # Compare loaded variables against current environment

        if file_hashes:
            for var in matched_vars:
                if var.name not in file_hashes:
                    continue

                stored = file_hashes[var.name]
                current = var.get_hash_components()

                # Check for changes
                diffs = []
                severity = 0 # 0=None, 1=White, 2=Orange, 3=Red

                # 1. Calculation (RED)
                if stored.get("calc") != current["calc"]:
                    diffs.append("Calculation logic changed")
                    severity = max(severity, 3)

                # 2. Dependencies (RED)
                if stored.get("deps") != current["deps"]:
                    diffs.append("Dependencies changed")
                    severity = max(severity, 3)

                # 3. Attributes (ORANGE) - dtype
                if stored.get("attrs") != current["attrs"]:
                    diffs.append("Attributes (dtype) changed")
                    severity = max(severity, 2)

                # 4. Metadata (WHITE) - description, lazy
                if stored.get("meta") != current["meta"]:
                    diffs.append("Metadata (desc/lazy) changed")
                    severity = max(severity, 1)

                if diffs:
                    color = WHITE
                    if severity == 3: color = RED
                    elif severity == 2: color = ORANGE

                    msg = f"{color}Variable '{var.name}': {', '.join(diffs)}{RESET}"
                    warnings_list.append(msg)

        if warnings_list:
            print(f"\n{ORANGE}VarFrame Integrity Check Warnings:{RESET}")
            for w in warnings_list:
                print(f"  {w}")

        return cls.from_pandas(df, variables=matched_vars, name=vf_name)

df_raw property writable

Get the raw DataFrame (if stored).

variables property writable

Get the list of variable classes.

add_variable(*variables, compute=True, suppress_warnings=False)

Alias for add_variables.

Source code in varframe/dataframe.py

def add_variable(
    self,
    *variables: VariableType,
    compute: bool = True,
    suppress_warnings: bool = False,
) -> VarFrame:
    """
    Alias for add_variables.
    """
    return self.add_variables(
        *variables, compute=compute, suppress_warnings=suppress_warnings
    )

add_variables(*variables, compute=True, suppress_warnings=False)

Register and optionally compute new variables (in-place).

Parameters:

Name	Type	Description	Default
*variables	VariableType	Variable classes to add.	()
compute	bool	If True (default), compute variables immediately. If False, just register them (must already exist in DataFrame).	True
suppress_warnings	bool	If True, suppress warnings for this call only.	False

Returns:

Type	Description
VarFrame	Self, for method chaining.

Raises:

Type	Description
KeyError	If compute=True and dependencies are missing.
RuntimeError	If implicit usage is disabled in VFConfig.
ValueError	If compute=True and adding a BaseVariable (must come from raw).

Source code in varframe/dataframe.py

def add_variables(
    self,
    *variables: VariableType,
    compute: bool = True,
    suppress_warnings: bool = False,
) -> VarFrame:
    """
    Register and optionally compute new variables (in-place).

    Args:
        *variables: Variable classes to add.
        compute: If True (default), compute variables immediately.
            If False, just register them (must already exist in DataFrame).
        suppress_warnings: If True, suppress warnings for this call only.

    Returns:
        Self, for method chaining.

    Raises:
        KeyError: If compute=True and dependencies are missing.
        RuntimeError: If implicit usage is disabled in VFConfig.
        ValueError: If compute=True and adding a BaseVariable (must come from raw).
    """
    for var in variables:
        if compute:
            # --- Compute Mode ---
            if var.name in self.columns:
                continue  # Already exists

            # Note: Explicit addition via add_variables does not trigger implicit warnings.

            if issubclass(var, BaseVariable):
                raise ValueError(
                    f"Cannot add BaseVariable '{var.name}' to existing DataFrame. "
                    "BaseVariables can only be computed from raw data."
                )
            else:
                self[var.name] = var.compute(self)

        else:
            # --- No Compute Mode (Registration Only) ---
            # Explicit registration - no warning needed.
            pass

        # Common: Add to registry
        if var not in self._variables:
            self._variables.append(var)

    return self

describe_variables()

Generate a summary DataFrame describing all variables.

Returns:

Type	Description
DataFrame	A DataFrame with columns: name, type, dtype, description, etc.

Source code in varframe/dataframe.py

def describe_variables(self) -> pd.DataFrame:
    """
    Generate a summary DataFrame describing all variables.

    Returns:
        A DataFrame with columns: name, type, dtype, description, etc.
    """
    data = [var.info() for var in self._variables]
    return pd.DataFrame(data)

explain_calculation(target_variables, legend=False)

Explain the calculation plan for the given variables given the current DataFrame state.

Prints a color-coded list indicating: - Variable Type (Base, Derived, Model) - Calculation Status (Ready vs Needs Calculation) - Warnings (Implicit computation, Auto-training, Inference)

Parameters:

Name	Type	Description	Default
target_variables	VariableList	List of variable classes to explain.	required
legend	bool	If True, print a legend for the warning codes.	False

Source code in varframe/dataframe.py

def explain_calculation(
    self,
    target_variables: VariableList,
    legend: bool = False,
) -> None:
    """
    Explain the calculation plan for the given variables given the current DataFrame state.

    Prints a color-coded list indicating:
    - Variable Type (Base, Derived, Model)
    - Calculation Status (Ready vs Needs Calculation)
    - Warnings (Implicit computation, Auto-training, Inference)

    Args:
        target_variables: List of variable classes to explain.
        legend: If True, print a legend for the warning codes.
    """
    from varframe.variables import BaseVariable
    from varframe.config import VFConfig, ImplicitOperation

    # ANSI color codes
    GREEN = "\033[92m"
    YELLOW = "\033[93m"
    PURPLE = "\033[95m"
    ORANGE = "\033[33m"
    GREY = "\033[90m"
    RESET = "\033[0m"

    resolved = resolve_dependencies(target_variables)

    names = ", ".join(v.name for v in target_variables)
    print(f"Calculation Plan for {names}:")

    used_codes = set()
    # Mapping: Code -> (Description, ConfigCheck)
    warning_defs = {
        "I": ("Implicit Compute", VFConfig.warn_add_variable_compute),
        "T": ("Auto-Train Model", VFConfig.warn_train_model),
        "M": ("Model Inference", VFConfig.warn_infer_model),
    }

    for i, v in enumerate(resolved, 1):
        # 1. Determine Type
        if hasattr(v, "model_class") and v.model_class:
            var_type = "model_prediction"
            color = PURPLE
        elif issubclass(v, BaseVariable):
            var_type = "base"
            color = GREEN
        else:
            var_type = "derived"
            color = YELLOW

        # 2. Determine Status
        exists = v.name in self.columns
        status_icon = "✅" if exists else "⏳"

        # 3. Predict Warnings
        current_codes = []
        if not exists:
            # Implicit Variable Warning
            if v not in self._variables and warning_defs["I"][1]:
                 current_codes.append("I")

            # Model Warnings
            if var_type == "model_prediction" and v.model_class:
                if not v.model_class.is_trained and warning_defs["T"][1]:
                     current_codes.append("T")
                if warning_defs["M"][1]:
                    current_codes.append("M")

        used_codes.update(current_codes)

        # Formatting
        line = f"  {i}. {status_icon} {color}{v.name}{RESET} ({var_type})"
        if current_codes:
            codes_str = ",".join(current_codes)
            # Subtle grey for the warning hints
            line += f" {GREY}⚠ [{codes_str}]{RESET}"

        print(line)

    if legend and used_codes:
        print("\nLegend:")
        for code in sorted(used_codes):
            desc = warning_defs[code][0]
            print(f"  {GREY}⚠ [{code}]{RESET} : {desc}")
    print()

filter_by_type(variable_type)

Filter to include only columns of a specific variable type.

Parameters:

Name	Type	Description	Default
variable_type	Type[Union[BaseVariable, DerivedVariable]]	The base class to filter by (BaseVariable or DerivedVariable).	required

Returns:

Type	Description
VarFrame	A new VarFrame containing only columns whose variables
VarFrame	are subclasses of the specified type.

Example

derived_df = vf.filter_by_type(DerivedVariable) base_df = vf.filter_by_type(BaseVariable)

Source code in varframe/dataframe.py

def filter_by_type(
    self, variable_type: Type[Union[BaseVariable, DerivedVariable]]
) -> VarFrame:
    """
    Filter to include only columns of a specific variable type.

    Args:
        variable_type: The base class to filter by (BaseVariable or DerivedVariable).

    Returns:
        A new VarFrame containing only columns whose variables
        are subclasses of the specified type.

    Example:
        >>> derived_df = vf.filter_by_type(DerivedVariable)
        >>> base_df = vf.filter_by_type(BaseVariable)
    """
    filtered_vars = [v for v in self._variables if issubclass(v, variable_type)]
    filtered_names = [v.name for v in filtered_vars]
    result = self[filtered_names].copy()
    result._variables = filtered_vars
    return result

from_pandas(df, variables=None, df_raw=None, name='varframe') classmethod

Create a VarFrame from a plain pandas DataFrame.

Use this to re-wrap a DataFrame after ML operations or when loading data that was previously converted with to_pandas().

Parameters:

Name	Type	Description	Default
df	DataFrame	The pandas DataFrame to convert (already processed data).	required
variables	Optional[VariableList]	List of variable classes to associate with the DataFrame.	None
df_raw	Optional[DataFrame]	Optional raw DataFrame to store for future resolve() calls.	None

Returns:

Type	Description
VarFrame	A new VarFrame with the given data and variables.

Example

plain_df = pd.read_pickle("data.pkl") vf = VarFrame.from_pandas(plain_df, variables=[Lap, Gap])

Source code in varframe/dataframe.py

@classmethod
def from_pandas(
    cls,
    df: pd.DataFrame,
    variables: Optional[VariableList] = None,
    df_raw: Optional[pd.DataFrame] = None,
    name: str = "varframe",
) -> VarFrame:
    """
    Create a VarFrame from a plain pandas DataFrame.

    Use this to re-wrap a DataFrame after ML operations or when
    loading data that was previously converted with to_pandas().

    Args:
        df: The pandas DataFrame to convert (already processed data).
        variables: List of variable classes to associate with the DataFrame.
        df_raw: Optional raw DataFrame to store for future resolve() calls.

    Returns:
        A new VarFrame with the given data and variables.

    Example:
        >>> plain_df = pd.read_pickle("data.pkl")
        >>> vf = VarFrame.from_pandas(plain_df, variables=[Lap, Gap])
    """
    vf = cls(df_raw=None, variables=None, compute=False, name=name)
    for col in df.columns:
        vf[col] = df[col].values
    vf.index = df.index
    vf._variables = list(variables) if variables else []
    vf._df_raw = df_raw
    return vf

get_variable(name)

Retrieve a variable class by its name.

Parameters:

Name	Type	Description	Default
name	str	The name of the variable to retrieve.	required

Returns:

Type	Description
Optional[VariableType]	The variable class if found, None otherwise.

Source code in varframe/dataframe.py

def get_variable(self, name: str) -> Optional[VariableType]:
    """
    Retrieve a variable class by its name.

    Args:
        name: The name of the variable to retrieve.

    Returns:
        The variable class if found, None otherwise.
    """
    for var in self._variables:
        if var.name == name:
            return var
    return None

list_variables()

Get a list of all variable names.

Returns:

Type	Description
List[str]	List of variable names in order.

Source code in varframe/dataframe.py

def list_variables(self) -> List[str]:
    """
    Get a list of all variable names.

    Returns:
        List of variable names in order.
    """
    return [v.name for v in self._variables]

load_csv(path_or_buf, variables=None, exclude=None, discard_unmatched=True, ambiguity=None, **kwargs) classmethod

Load a VarFrame from a CSV file, automatically discovering variables.

This method scans the current Python environment for variable definitions that match the columns in the CSV.

Parameters:

Name	Type	Description	Default
path_or_buf	Union[str, Any]	Path to the CSV file.	required
variables	Optional[List[VariableType]]	Whitelist of variable classes to load. If provided, only these variables will be matched. Acts as implicit disambiguation.	None
exclude	Optional[List[VariableType]]	Blacklist of variable classes to exclude from loading. Cannot be used together with variables.	None
discard_unmatched	bool	If True (default), columns not matching any known variable are dropped. If False, they are kept as plain columns.	True
ambiguity	Optional[Dict[str, VariableType]]	Dict mapping variable names to specific classes for disambiguation when multiple definitions exist with the same name.	None
**kwargs	Any	Arguments passed to pd.read_csv.	{}

Returns:

Type	Description
VarFrame	A reconstructed VarFrame.

Raises:

Type	Description
ValueError	If both variables and exclude are provided.
AmbiguityError	If multiple variable definitions match a column name and no disambiguation is provided.

Source code in varframe/dataframe.py

@classmethod
def load_csv(
    cls, 
    path_or_buf: Union[str, Any], 
    variables: Optional[List[VariableType]] = None,
    exclude: Optional[List[VariableType]] = None,
    discard_unmatched: bool = True,
    ambiguity: Optional[Dict[str, VariableType]] = None,
    **kwargs: Any
) -> VarFrame:
    """
    Load a VarFrame from a CSV file, automatically discovering variables.

    This method scans the current Python environment for variable definitions
    that match the columns in the CSV.

    Args:
        path_or_buf: Path to the CSV file.
        variables: Whitelist of variable classes to load. If provided, only 
            these variables will be matched. Acts as implicit disambiguation.
        exclude: Blacklist of variable classes to exclude from loading.
            Cannot be used together with `variables`.
        discard_unmatched: If True (default), columns not matching any known
            variable are dropped. If False, they are kept as plain columns.
        ambiguity: Dict mapping variable names to specific classes for 
            disambiguation when multiple definitions exist with the same name.
        **kwargs: Arguments passed to pd.read_csv.

    Returns:
        A reconstructed VarFrame.

    Raises:
        ValueError: If both `variables` and `exclude` are provided.
        AmbiguityError: If multiple variable definitions match a column name
            and no disambiguation is provided.
    """
    from varframe.exceptions import AmbiguityError

    # Validation
    if variables is not None and exclude is not None:
        raise ValueError("Cannot use both 'variables' and 'exclude' parameters together.")

    df = pd.read_csv(path_or_buf, **kwargs)

    # Auto-discovery
    known_vars = cls._discover_all_variables()
    matched_vars = []

    # Build target names based on whitelist/blacklist
    if variables is not None:
        target_names = {v.name for v in variables}
        # Create a lookup for whitelisted variables by name
        whitelist_by_name: Dict[str, VariableType] = {v.name: v for v in variables}
    elif exclude is not None:
        exclude_names = {v.name for v in exclude}
        target_names = {col for col in df.columns if col not in exclude_names}
        whitelist_by_name = None
    else:
        target_names = set(df.columns)
        whitelist_by_name = None

    for col in df.columns:
        if col not in target_names:
            continue

        if col not in known_vars:
            continue

        candidates = known_vars[col]

        # If whitelist mode, use the whitelisted class directly
        if whitelist_by_name is not None:
            if col in whitelist_by_name:
                matched_vars.append(whitelist_by_name[col])
            continue

        # Check for ambiguity
        if len(candidates) > 1:
            # Check if disambiguation provided via ambiguity parameter
            if ambiguity and col in ambiguity:
                matched_vars.append(ambiguity[col])
            else:
                raise AmbiguityError(
                    f"Ambiguous variable '{col}'. Multiple definitions found: "
                    f"{[c.__name__ for c in candidates]}. "
                    f"Use 'ambiguity' parameter to specify which to use, e.g.: "
                    f"ambiguity={{'{col}': {candidates[0].__name__}}}"
                )
        else:
            matched_vars.append(candidates[0])

    # Handle unmatched columns
    mapped_names = {v.name for v in matched_vars}
    unmapped_cols = [col for col in df.columns if col not in mapped_names]

    if discard_unmatched:
        # Drop unmatched columns
        cols_to_keep = [v.name for v in matched_vars]
        df = df[[c for c in cols_to_keep if c in df.columns]]
    else:
        # Warn about unmapped columns (old behavior)
        if unmapped_cols:
            from varframe.config import VFConfig, ImplicitOperation
            VFConfig.warn(
                ImplicitOperation.ADD_VARIABLE_NO_COMPUTE,
                f"Loaded columns {unmapped_cols} do not match any known Variable class. "
                "They will be loaded as plain pandas columns."
            )

    # Try to infer name from path
    name = "varframe"
    if isinstance(path_or_buf, str):
        import os
        name = os.path.splitext(os.path.basename(path_or_buf))[0]

    return cls.from_pandas(df, variables=matched_vars, name=name)

load_parquet(path, variables=None, exclude=None, discard_unmatched=True, ambiguity=None, **kwargs) classmethod

Load a VarFrame from a Parquet file, using metadata or auto-discovery.

1. Checks for 'varframe_variables' metadata in the file.
2. If found, looks up those variables in the environment.
3. If not found, falls back to matching column names (like load_csv).

Parameters:

Name	Type	Description	Default
path	Union[str, Any]	Path to the Parquet file.	required
variables	Optional[List[VariableType]]	Whitelist of variable classes to load. If provided, only these variables will be matched. Acts as implicit disambiguation and overrides hash-based resolution.	None
exclude	Optional[List[VariableType]]	Blacklist of variable classes to exclude from loading. Cannot be used together with variables.	None
discard_unmatched	bool	If True (default), columns not matching any known variable are dropped. If False, they are kept as plain columns.	True
ambiguity	Optional[Dict[str, VariableType]]	Dict mapping variable names to specific classes for disambiguation. Overrides hash-based resolution for specified names.	None
**kwargs	Any	Arguments passed to pyarrow/pandas read functions.	{}

Returns:

Type	Description
VarFrame	A reconstructed VarFrame.

Raises:

Type	Description
ValueError	If both variables and exclude are provided.

Source code in varframe/dataframe.py

@classmethod
def load_parquet(
    cls, 
    path: Union[str, Any], 
    variables: Optional[List[VariableType]] = None,
    exclude: Optional[List[VariableType]] = None,
    discard_unmatched: bool = True,
    ambiguity: Optional[Dict[str, VariableType]] = None,
    **kwargs: Any
) -> VarFrame:
    """
    Load a VarFrame from a Parquet file, using metadata or auto-discovery.

    1. Checks for 'varframe_variables' metadata in the file.
    2. If found, looks up those variables in the environment.
    3. If not found, falls back to matching column names (like load_csv).

    Args:
        path: Path to the Parquet file.
        variables: Whitelist of variable classes to load. If provided, only 
            these variables will be matched. Acts as implicit disambiguation
            and overrides hash-based resolution.
        exclude: Blacklist of variable classes to exclude from loading.
            Cannot be used together with `variables`.
        discard_unmatched: If True (default), columns not matching any known
            variable are dropped. If False, they are kept as plain columns.
        ambiguity: Dict mapping variable names to specific classes for 
            disambiguation. Overrides hash-based resolution for specified names.
        **kwargs: Arguments passed to pyarrow/pandas read functions.

    Returns:
        A reconstructed VarFrame.

    Raises:
        ValueError: If both `variables` and `exclude` are provided.
    """
    import json

    # Validation
    if variables is not None and exclude is not None:
        raise ValueError("Cannot use both 'variables' and 'exclude' parameters together.")

    # ANSI Colors
    RED = "\033[91m"
    ORANGE = "\033[33m"
    WHITE = "\033[37m"
    RESET = "\033[0m"

    warnings_list = []

    # Try reading metadata first (requires pyarrow)
    file_hashes = {}
    var_names_from_meta = []
    all_columns = []

    try:
        import pyarrow.parquet as pq
        meta = pq.read_metadata(path)
        # Get column names from schema
        all_columns = list(meta.schema.names)
        if meta.metadata:
            if b'varframe_variables' in meta.metadata:
                var_names_from_meta = json.loads(meta.metadata[b'varframe_variables'])
            if b'varframe_hashes' in meta.metadata:
                file_hashes = json.loads(meta.metadata[b'varframe_hashes'])
    except ImportError:
        pass

    known_vars = cls._discover_all_variables()

    # Determine which columns to load based on parameters
    if variables is not None:
        # Whitelist mode - use provided classes directly
        target_names = [v.name for v in variables]
        whitelist_by_name: Dict[str, VariableType] = {v.name: v for v in variables}
    elif exclude is not None:
        exclude_names = {v.name for v in exclude}
        source_names = var_names_from_meta if var_names_from_meta else all_columns
        target_names = [n for n in source_names if n not in exclude_names]
        whitelist_by_name = None
    else:
        target_names = None  # Load all
        whitelist_by_name = None

    # Use pyarrow column selection for efficiency
    try:
        import pyarrow.parquet as pq
        if target_names is not None:
            # Only read requested columns
            cols_to_read = [c for c in target_names if c in all_columns]
            table = pq.read_table(path, columns=cols_to_read, **kwargs)
        else:
            table = pq.read_table(path, **kwargs)
        df = table.to_pandas()
    except ImportError:
        # Fallback to pandas (less efficient)
        df = pd.read_parquet(path, **kwargs)
        if target_names is not None:
            available = [c for c in target_names if c in df.columns]
            df = df[available]

    matched_vars = []

    # Determine variable names to look for
    if target_names is not None:
        names_to_resolve = target_names
    else:
        names_to_resolve = var_names_from_meta if var_names_from_meta else df.columns

    for name in names_to_resolve:
        # Handle case where name is in metadata but not in known_vars (not imported)
        if name not in known_vars:
            continue

        # If whitelist mode, use the whitelisted class directly
        if whitelist_by_name is not None:
            if name in whitelist_by_name:
                matched_vars.append(whitelist_by_name[name])
            continue

        candidates = known_vars[name]

        # Check for multiple candidates (ambiguity)
        if len(candidates) > 1:
            # Require explicit disambiguation
            if ambiguity and name in ambiguity:
                matched_vars.append(ambiguity[name])
            else:
                from varframe.exceptions import AmbiguityError
                raise AmbiguityError(
                    f"Ambiguous variable '{name}'. Multiple definitions found: "
                    f"{[c.__name__ for c in candidates]}. "
                    f"Use 'ambiguity' parameter to specify which to use, e.g.: "
                    f"ambiguity={{'{name}': {candidates[0].__name__}}}"
                )
        else:
            # Single candidate - use it
            matched_vars.append(candidates[0])

    # Handle unmatched columns
    mapped_names = {v.name for v in matched_vars}
    unmapped_cols = [col for col in df.columns if col not in mapped_names]

    if discard_unmatched:
        # Drop unmatched columns
        cols_to_keep = [v.name for v in matched_vars]
        df = df[[c for c in cols_to_keep if c in df.columns]]
    else:
        # Warn about unmapped columns (old behavior)
        if unmapped_cols:
            from varframe.config import VFConfig, ImplicitOperation
            VFConfig.warn(
                ImplicitOperation.ADD_VARIABLE_NO_COMPUTE,
                f"Loaded columns {unmapped_cols} do not match any known Variable class. "
                "They will be loaded as plain pandas columns."
            )

    # Try to infer name from path
    vf_name = "varframe"
    if isinstance(path, str):
        import os
        vf_name = os.path.splitext(os.path.basename(path))[0]

    # ------------------- Integrity Check -------------------
    # Compare loaded variables against current environment

    if file_hashes:
        for var in matched_vars:
            if var.name not in file_hashes:
                continue

            stored = file_hashes[var.name]
            current = var.get_hash_components()

            # Check for changes
            diffs = []
            severity = 0 # 0=None, 1=White, 2=Orange, 3=Red

            # 1. Calculation (RED)
            if stored.get("calc") != current["calc"]:
                diffs.append("Calculation logic changed")
                severity = max(severity, 3)

            # 2. Dependencies (RED)
            if stored.get("deps") != current["deps"]:
                diffs.append("Dependencies changed")
                severity = max(severity, 3)

            # 3. Attributes (ORANGE) - dtype
            if stored.get("attrs") != current["attrs"]:
                diffs.append("Attributes (dtype) changed")
                severity = max(severity, 2)

            # 4. Metadata (WHITE) - description, lazy
            if stored.get("meta") != current["meta"]:
                diffs.append("Metadata (desc/lazy) changed")
                severity = max(severity, 1)

            if diffs:
                color = WHITE
                if severity == 3: color = RED
                elif severity == 2: color = ORANGE

                msg = f"{color}Variable '{var.name}': {', '.join(diffs)}{RESET}"
                warnings_list.append(msg)

    if warnings_list:
        print(f"\n{ORANGE}VarFrame Integrity Check Warnings:{RESET}")
        for w in warnings_list:
            print(f"  {w}")

    return cls.from_pandas(df, variables=matched_vars, name=vf_name)

resolve(*target_variables, suppress_warnings=False)

Resolve and compute target variables with automatic dependency resolution.

This method automatically determines all missing dependencies for the target variables, computes them in the correct DAG order, and adds them to the DataFrame.

Parameters:

Name	Type	Description	Default
*target_variables	VariableType	Variable classes to resolve and compute.	()
suppress_warnings	bool	If True, suppress warnings for this call only.	False

Returns:

Type	Description
VarFrame	Self, for method chaining.

Raises:

Type	Description
ValueError	If circular dependencies are detected.
ValueError	If a BaseVariable is needed but df_raw is not available.
RuntimeError	If implicit operations are disabled in VFConfig.

Example

vf = VarFrame(df_raw, [Lap, Gap]) vf.resolve(PredictedGapDelta)

Source code in varframe/dataframe.py

def resolve(
    self,
    *target_variables: VariableType,
    suppress_warnings: bool = False,
) -> VarFrame:
    """
    Resolve and compute target variables with automatic dependency resolution.

    This method automatically determines all missing dependencies for the
    target variables, computes them in the correct DAG order, and adds
    them to the DataFrame.

    Args:
        *target_variables: Variable classes to resolve and compute.
        suppress_warnings: If True, suppress warnings for this call only.

    Returns:
        Self, for method chaining.

    Raises:
        ValueError: If circular dependencies are detected.
        ValueError: If a BaseVariable is needed but df_raw is not available.
        RuntimeError: If implicit operations are disabled in VFConfig.

    Example:
        >>> vf = VarFrame(df_raw, [Lap, Gap])
        >>> vf.resolve(PredictedGapDelta)
    """
    # Resolve all dependencies
    all_vars = resolve_dependencies(list(target_variables))

    # Filter to only missing variables
    # Filter to only missing variables
    missing_vars = [v for v in all_vars if v.name not in self.columns]

    # Split into compute-now vs lazy
    lazy_vars = [
        v for v in missing_vars if issubclass(v, DerivedVariable) and v.lazy
    ]
    missing_vars = [v for v in missing_vars if v not in lazy_vars]

    # Check for missing BaseVariables - need df_raw to compute them
    missing_base = [v for v in missing_vars if issubclass(v, BaseVariable)]
    if missing_base and self._df_raw is None:
        names = [v.name for v in missing_base]
        raise ValueError(
            f"Cannot resolve BaseVariables {names}: no raw DataFrame stored. "
            "Pass df_raw to VarFrame() or set vf.df_raw to enable."
        )

    # Warn about implicit computation of missing variables
    if missing_vars and not suppress_warnings:
        var_names = [v.name for v in missing_vars]
        VFConfig.check_permission(
            ImplicitOperation.ADD_VARIABLE_COMPUTE,
            f"Cannot resolve variables {var_names}. "
            "Set VFConfig.allow_implicit_compute = True to enable.",
        )
        VFConfig.warn(
            ImplicitOperation.ADD_VARIABLE_COMPUTE,
            f"Implicitly computing {len(missing_vars)} variable(s) not in _variables: {var_names}",
            stacklevel=3,
        )

    # Register lazy variables (so get_variable works)
    for var in lazy_vars:
        if var not in self._variables:
            self._variables.append(var)

    if suppress_warnings:
        context = VFConfig.suppress_warnings()
    else:
        context = VFConfig.null_context()

    with context:
        # Compute missing variables in order
        for var in missing_vars:
            if issubclass(var, BaseVariable):
                self[var.name] = var.compute(self._df_raw)
            else:
                self[var.name] = var.compute(self)

            if var not in self._variables:
                self._variables.append(var)

    return self

to_csv(path_or_buf=None, *args, include=None, variables=None, **kwargs)

Write object to a comma-separated values (csv) file.

Enhancements over pandas.to_csv: - Supports include and variables to compute lazy variables on-the-fly. - Warns if registered variables are not included in the export. - Defaults to {self.name}.csv if path is not provided.

Source code in varframe/dataframe.py

def to_csv(
    self,
    path_or_buf: Optional[Union[str, Any]] = None,
    *args: Any,
    include: Optional[List[str]] = None,
    variables: Optional[VariableList] = None,
    **kwargs: Any,
) -> Optional[str]:
    """
    Write object to a comma-separated values (csv) file.

    Enhancements over pandas.to_csv:
    - Supports `include` and `variables` to compute lazy variables on-the-fly.
    - Warns if registered variables are not included in the export.
    - Defaults to {self.name}.csv if path is not provided.
    """
    from varframe.config import VFConfig

    # 1. Resolve Data
    if include or variables:
        df_to_export = self.view(include=include, variables=variables)
    else:
        df_to_export = self

    # 2. Prevent implicit data loss (Warn if variables are missing)
    # Check which variables are in the export
    exported_cols = set(df_to_export.columns)
    missing_vars = [
        v.name for v in self._variables 
        if v.name not in exported_cols and v.name not in self.columns
    ]

    # Also check for variables present in self but not in export (if view filtered them)
    excluded_present_vars = [
         v.name for v in self._variables
         if v.name in self.columns and v.name not in exported_cols
    ]

    if missing_vars:
         path_str = str(path_or_buf) if path_or_buf else "output"
         VFConfig.warn(
            ImplicitOperation.ADD_VARIABLE_COMPUTE, # Reusing generic warning op
            f"Exporting to {path_str} without computing lazy variables: {missing_vars}. "
            "Use `include=['all']` or `variables=[...]` to compute them during export.",
         )

    # 3. Resolve Path
    if path_or_buf is None:
        path_or_buf = f"{self.name}.csv"

    if hasattr(df_to_export, "to_pandas"):
        return df_to_export.to_pandas().to_csv(path_or_buf, *args, **kwargs)
    else:
        return df_to_export.to_csv(path_or_buf, *args, **kwargs)

to_ml()

Alias for to_pandas(). Explicit conversion for ML pipelines.

Use this to clearly indicate the DataFrame is being prepared for machine learning operations.

Returns:

Type	Description
DataFrame	A plain pandas DataFrame suitable for ML libraries.

Example

X_train = vf.to_ml() model.fit(X_train, y_train)

Source code in varframe/dataframe.py

def to_ml(self) -> pd.DataFrame:
    """
    Alias for to_pandas(). Explicit conversion for ML pipelines.

    Use this to clearly indicate the DataFrame is being prepared
    for machine learning operations.

    Returns:
        A plain pandas DataFrame suitable for ML libraries.

    Example:
        >>> X_train = vf.to_ml()
        >>> model.fit(X_train, y_train)
    """
    return pd.DataFrame(self)

to_pandas()

Convert to a plain pandas DataFrame.

Use this before passing to ML libraries that may have issues with DataFrame subclasses (e.g., pickle, joblib, some sklearn pipelines).

Returns:

Type	Description
DataFrame	A plain pandas DataFrame with the same data (no variable metadata).

Example

plain_df = vf.to_pandas() model.fit(plain_df, y)

Source code in varframe/dataframe.py

def to_pandas(self) -> pd.DataFrame:
    """
    Convert to a plain pandas DataFrame.

    Use this before passing to ML libraries that may have issues
    with DataFrame subclasses (e.g., pickle, joblib, some sklearn pipelines).

    Returns:
        A plain pandas DataFrame with the same data (no variable metadata).

    Example:
        >>> plain_df = vf.to_pandas()
        >>> model.fit(plain_df, y)
    """
    return pd.DataFrame(self)

to_parquet(path=None, *args, include=None, variables=None, **kwargs)

Write object to a binary parquet file.

Enhancements over pandas.to_parquet: - Supports include and variables to compute lazy variables on-the-fly. - Warns if registered variables are not included in the export. - Defaults to {self.name}.parquet if path is not provided. - Saves variable names in file metadata.

Source code in varframe/dataframe.py

def to_parquet(
    self,
    path: Optional[Union[str, Any]] = None,
    *args: Any,
    include: Optional[List[str]] = None,
    variables: Optional[VariableList] = None,
    **kwargs: Any,
) -> Optional[bytes]:
    """
    Write object to a binary parquet file.

    Enhancements over pandas.to_parquet:
    - Supports `include` and `variables` to compute lazy variables on-the-fly.
    - Warns if registered variables are not included in the export.
    - Defaults to {self.name}.parquet if path is not provided.
    - Saves variable names in file metadata.
    """
    import json
    from varframe.config import VFConfig

    # 1. Resolve Data
    if include or variables:
        df_to_export = self.view(include=include, variables=variables)
    else:
        df_to_export = self

    # 2. Prevent implicit data loss
    exported_cols = set(df_to_export.columns)
    missing_vars = [
        v.name for v in self._variables 
        if v.name not in exported_cols and v.name not in self.columns
    ]

    if missing_vars:
         path_str = str(path) if path else "output"
         VFConfig.warn(
            ImplicitOperation.ADD_VARIABLE_COMPUTE,
            f"Exporting to {path_str} without computing lazy variables: {missing_vars}. "
            "Use `include=['all']` or `variables=[...]` to compute them during export.",
         )

    # 3. Resolve Path
    if path is None:
        path = f"{self.name}.parquet"

    # 4. Prepare Metadata (Strategy B)
    # Get existing keyword args or creating new dict
    # pyarrow.Table.from_pandas uses 'preserve_index' etc.
    # to_parquet kwargs are passed to the engine. 
    # For pyarrow engine (default), we can't easily inject metadata via to_parquet directly 
    # because pandas abstracts this. 
    # WORKAROUND: We will use table properties if using pyarrow, 
    # but to keep it simple and pandas-compliant, we might need a distinct approach.
    # Actually, pandas >= 1.0 does not easily support custom metadata in to_parquet 
    # without dropping down to pyarrow directly.

    try:
        import pyarrow as pa
        import pyarrow.parquet as pq

        # Convert to Table
        # Handle VarFrame or plain DataFrame
        if hasattr(df_to_export, "to_pandas"):
            df_plain = df_to_export.to_pandas()
        else:
            df_plain = df_to_export

        table = pa.Table.from_pandas(df_plain)

        # Add Metadata
        custom_meta = {
            "varframe_variables": json.dumps([v.name for v in self._variables])
        }

        # Add Hash Metadata
        try:
            hashes = {v.name: v.get_hash_components() for v in self._variables}
            custom_meta["varframe_hashes"] = json.dumps(hashes)
        except Exception as e:
            VFConfig.warn(
                ImplicitOperation.ADD_VARIABLE_COMPUTE,
                f"Failed to compute variable hashes for export: {e}"
            )
        # Combine with existing metadata
        existing_meta = table.schema.metadata or {}
        combined_meta = {**existing_meta, **{k.encode(): v.encode() for k, v in custom_meta.items()}}
        table = table.replace_schema_metadata(combined_meta)

        # Write
        pq.write_table(table, path, *args, **kwargs)
        return None

    except ImportError:
        # Fallback for when pyarrow is not available or user wants different engine
         VFConfig.warn(
             ImplicitOperation.ADD_VARIABLE_COMPUTE,
             "Pyarrow not found or failed. Exporting without VarFrame metadata."
         )
         if hasattr(df_to_export, "to_pandas"):
             return df_to_export.to_pandas().to_parquet(path, *args, **kwargs)
         else:
             return df_to_export.to_parquet(path, *args, **kwargs)

view(include=None, variables=None)

Create a DataFrame view containing only specific variables.

Parameters:

Name	Type	Description	Default
include	Optional[List[str]]	List of variable categories to include. Options: 'base', 'derived', 'lazy', 'model', 'all'. Defaults to ['all'] if both include and variables are None.	None
variables	Optional[VariableList]	Explicit list of variable classes to include.	None

Returns:

Type	Description
DataFrame	A pandas DataFrame with the requested variables.
DataFrame	Lazy variables will be computed on-demand for this view.

Example

vf.view(include=['base', 'lazy']) vf.view(variables=[LazySum])

Source code in varframe/dataframe.py

def view(
    self,
    include: Optional[List[str]] = None,
    variables: Optional[VariableList] = None,
) -> pd.DataFrame:
    """
    Create a DataFrame view containing only specific variables.

    Args:
        include: List of variable categories to include.
            Options: 'base', 'derived', 'lazy', 'model', 'all'.
            Defaults to ['all'] if both include and variables are None.
        variables: Explicit list of variable classes to include.

    Returns:
        A pandas DataFrame with the requested variables. 
        Lazy variables will be computed on-demand for this view.

    Example:
        >>> vf.view(include=['base', 'lazy'])
        >>> vf.view(variables=[LazySum])
    """
    target_vars = []

    # 1. Handle 'variables' argument
    if variables:
        target_vars.extend(variables)

    # 2. Handle 'include' argument
    if include:
        for category in include:
            if category == "all":
                target_vars.extend(self._variables)
            elif category == "base":
                target_vars.extend(
                    [v for v in self._variables if issubclass(v, BaseVariable)]
                )
            elif category == "derived":
                # Non-lazy derived
                target_vars.extend(
                    [
                        v
                        for v in self._variables
                        if issubclass(v, DerivedVariable) and not v.lazy
                    ]
                )
            elif category == "lazy":
                # Lazy derived
                target_vars.extend(
                    [
                        v
                        for v in self._variables
                        if issubclass(v, DerivedVariable) and v.lazy
                    ]
                )
            elif category == "model":
                target_vars.extend(
                    [
                        v
                        for v in self._variables
                        if hasattr(v, "model_class") and v.model_class
                    ]
                )

    # Default to 'all' if nothing specified
    if not target_vars:
        target_vars = list(self._variables)

    # Remove duplicates while preserving order
    target_vars = list(dict.fromkeys(target_vars))

    # 3. Construct DataFrame
    data = {}
    for var in target_vars:
        # If standard column, use it (zero copy if possible)
        if var.name in self.columns:
            data[var.name] = self[var.name]
        else:
            # Must be lazy or missing -> Compute it
            data[var.name] = var.compute(self)

    return pd.DataFrame(data, index=self.index)

Variables

varframe.BaseVariable

A declarative variable that maps to a column in a raw DataFrame.

Define subclasses with class attributes to create variable definitions. The class itself represents the variable - no instantiation needed.

Class Attributes

name (str): The name of the variable in the processed DataFrame. raw_column (str): The column name in the raw DataFrame to extract. dtype (str): The pandas dtype to cast the column to. Defaults to 'float'.

Note

Use the class docstring as the variable description.

Example

class LapNumber(BaseVariable): ... '''The current lap number in the race.''' ... name = "lap" ... raw_column = "lap_num" ... dtype = "int" ...

Use the class directly, not an instance

series = LapNumber.compute(raw_dataframe)

Source code in varframe/variables.py

class BaseVariable:
    """
    A declarative variable that maps to a column in a raw DataFrame.

    Define subclasses with class attributes to create variable definitions.
    The class itself represents the variable - no instantiation needed.

    Class Attributes:
        **name (str)**: The name of the variable in the processed DataFrame.
        **raw_column (str)**: The column name in the raw DataFrame to extract.
        **dtype (str)**: The pandas dtype to cast the column to. Defaults to 'float'.

    Note:
        Use the class docstring as the variable description.

    Example:
        >>> class LapNumber(BaseVariable):
        ...     '''The current lap number in the race.'''
        ...     name = "lap"
        ...     raw_column = "lap_num"
        ...     dtype = "int"
        ...
        >>> # Use the class directly, not an instance
        >>> series = LapNumber.compute(raw_dataframe)
    """

    # Class attributes to be overridden by subclasses
    name: ClassVar[str] = ""
    raw_column: ClassVar[str] = ""
    dtype: ClassVar[str] = "float"

    def __init_subclass__(cls, **kwargs: Any) -> None:
        """Validate subclass definition on class creation."""
        super().__init_subclass__(**kwargs)
        # Auto-generate name from class name if not specified
        if not cls.name and cls.__name__ not in ("BaseVariable", "DerivedVariable"):
            cls.name = cls.__name__.lower()

    @classmethod
    def get_description(cls) -> str:
        """
        Get the variable description from the class docstring.

        Returns:
            The first line of the class docstring, or empty string if none.
        """
        if cls.__doc__:
            return cls.__doc__.strip().split("\n")[0]
        return ""

    @classmethod
    def compute(cls, df_raw: pd.DataFrame) -> pd.Series:
        """
        Extract and transform the column from a raw DataFrame.

        Args:
            df_raw: The raw DataFrame containing the source column.

        Returns:
            A pandas Series with the extracted column cast to the specified dtype.

        Raises:
            KeyError: If `raw_column` does not exist in `df_raw`.
            ValueError: If the column cannot be cast to the specified `dtype`.
        """
        return df_raw[cls.raw_column].astype(cls.dtype)

    @classmethod
    def get_hash_components(cls) -> Dict[str, str]:
        """
        Get the hash components of the variable definition.

        Returns:
            Dict with keys: calc, deps, attrs, meta.
        """
        # Calc: raw_column
        calc_hash = hashlib.sha256(cls.raw_column.encode("utf-8")).hexdigest()

        # Deps: None for BaseVariable
        deps_hash = hashlib.sha256(b"").hexdigest()

        # Attrs: dtype
        attrs_hash = hashlib.sha256(cls.dtype.encode("utf-8")).hexdigest()

        # Meta: description
        meta_hash = hashlib.sha256(cls.get_description().encode("utf-8")).hexdigest()

        return {
            "calc": calc_hash,
            "deps": deps_hash,
            "attrs": attrs_hash,
            "meta": meta_hash,
        }

    @classmethod
    def info(cls) -> Dict[str, Any]:
        """
        Get variable metadata as a dictionary.

        Returns:
            Dict with name, type, raw_column, dtype, and description.
        """
        return {
            "name": cls.name,
            "type": "base",
            "raw_column": cls.raw_column,
            "dtype": cls.dtype,
            "description": cls.get_description(),
        }

    def __repr__(self) -> str:
        return f"<{self.__class__.__name__}: {self.name}>"

compute(df_raw) classmethod

Extract and transform the column from a raw DataFrame.

Parameters:

Name	Type	Description	Default
df_raw	DataFrame	The raw DataFrame containing the source column.	required

Returns:

Type	Description
Series	A pandas Series with the extracted column cast to the specified dtype.

Raises:

Type	Description
KeyError	If raw_column does not exist in df_raw.
ValueError	If the column cannot be cast to the specified dtype.

Source code in varframe/variables.py

@classmethod
def compute(cls, df_raw: pd.DataFrame) -> pd.Series:
    """
    Extract and transform the column from a raw DataFrame.

    Args:
        df_raw: The raw DataFrame containing the source column.

    Returns:
        A pandas Series with the extracted column cast to the specified dtype.

    Raises:
        KeyError: If `raw_column` does not exist in `df_raw`.
        ValueError: If the column cannot be cast to the specified `dtype`.
    """
    return df_raw[cls.raw_column].astype(cls.dtype)

get_hash_components() classmethod

Get the hash components of the variable definition.

Returns:

Type	Description
Dict[str, str]	Dict with keys: calc, deps, attrs, meta.

Source code in varframe/variables.py

@classmethod
def get_hash_components(cls) -> Dict[str, str]:
    """
    Get the hash components of the variable definition.

    Returns:
        Dict with keys: calc, deps, attrs, meta.
    """
    # Calc: raw_column
    calc_hash = hashlib.sha256(cls.raw_column.encode("utf-8")).hexdigest()

    # Deps: None for BaseVariable
    deps_hash = hashlib.sha256(b"").hexdigest()

    # Attrs: dtype
    attrs_hash = hashlib.sha256(cls.dtype.encode("utf-8")).hexdigest()

    # Meta: description
    meta_hash = hashlib.sha256(cls.get_description().encode("utf-8")).hexdigest()

    return {
        "calc": calc_hash,
        "deps": deps_hash,
        "attrs": attrs_hash,
        "meta": meta_hash,
    }

info() classmethod

Get variable metadata as a dictionary.

Returns:

Type	Description
Dict[str, Any]	Dict with name, type, raw_column, dtype, and description.

Source code in varframe/variables.py

@classmethod
def info(cls) -> Dict[str, Any]:
    """
    Get variable metadata as a dictionary.

    Returns:
        Dict with name, type, raw_column, dtype, and description.
    """
    return {
        "name": cls.name,
        "type": "base",
        "raw_column": cls.raw_column,
        "dtype": cls.dtype,
        "description": cls.get_description(),
    }

varframe.DerivedVariable

A declarative variable computed from other variables.

Define subclasses with class attributes and a calculate() classmethod to create computed variable definitions. Dependencies are resolved automatically using the DataFrame as the single source of truth.

Class Attributes

name (str): The name of the variable in the processed DataFrame. dependencies (List[Type]): List of variable classes this depends on. dtype (str): The pandas dtype for the result. Defaults to 'float'. lazy (bool): If True, computed on access and not stored in DataFrame. Defaults to False.

Note

• Use the class docstring as the variable description.
• Override the calculate() classmethod to define computation logic.
• Access dependencies directly from df (e.g., df["gap"]), not memo.

Example

class GapDelta(DerivedVariable): ... '''Change in gap from previous measurement.''' ... name = "gap_delta" ... dependencies = [Gap] ... ... @classmethod ... def calculate(cls, df: pd.DataFrame) -> pd.Series: ... return df["gap"] - df["gap"].shift(1)

Source code in varframe/variables.py

class DerivedVariable:
    """
    A declarative variable computed from other variables.

    Define subclasses with class attributes and a `calculate()` classmethod
    to create computed variable definitions. Dependencies are resolved
    automatically using the DataFrame as the single source of truth.

    Class Attributes:
        name (str): The name of the variable in the processed DataFrame.
        dependencies (List[Type]): List of variable classes this depends on.
        dtype (str): The pandas dtype for the result. Defaults to 'float'.
        lazy (bool): If True, computed on access and not stored in DataFrame. Defaults to False.

    Note:
        - Use the class docstring as the variable description.
        - Override the `calculate()` classmethod to define computation logic.
        - Access dependencies directly from df (e.g., `df["gap"]`), not memo.

    Example:
        >>> class GapDelta(DerivedVariable):
        ...     '''Change in gap from previous measurement.'''
        ...     name = "gap_delta"
        ...     dependencies = [Gap]
        ...
        ...     @classmethod
        ...     def calculate(cls, df: pd.DataFrame) -> pd.Series:
        ...         return df["gap"] - df["gap"].shift(1)
    """

    # Class attributes to be overridden by subclasses
    name: ClassVar[str] = ""
    dependencies: ClassVar[List["VariableType"]] = []
    dtype: ClassVar[str] = "float"
    lazy: ClassVar[bool] = False

    def __init_subclass__(cls, **kwargs: Any) -> None:
        """Validate subclass definition and set defaults."""
        super().__init_subclass__(**kwargs)
        # Auto-generate name from class name if not specified
        if not cls.name and cls.__name__ not in ("BaseVariable", "DerivedVariable"):
            cls.name = cls.__name__.lower()
        # Ensure dependencies is a new list (not shared reference)
        if "dependencies" not in cls.__dict__:
            cls.dependencies = []

    @classmethod
    def get_description(cls) -> str:
        """
        Get the variable description from the class docstring.

        Returns:
            The first line of the class docstring, or empty string if none.
        """
        if cls.__doc__:
            return cls.__doc__.strip().split("\n")[0]
        return ""

    @classmethod
    def calculate(cls, df: pd.DataFrame) -> pd.Series:
        """
        Calculate the derived variable's values.

        Override this method in subclasses to define the computation logic.
        Access dependency values directly from df (e.g., `df["gap"]`).

        Args:
            df: The current processed DataFrame containing all computed
                variables so far (including dependencies).

        Returns:
            A pandas Series with the computed values.

        Raises:
            NotImplementedError: If not overridden in subclass.
        """
        raise NotImplementedError(
            f"{cls.__name__} must implement the 'calculate' classmethod"
        )

    @classmethod
    def compute(cls, df: pd.DataFrame) -> pd.Series:
        """
        Compute the derived variable from the DataFrame.

        Dependencies must already exist in df before calling this method.
        Use `VarFrame` to handle dependency ordering automatically.

        Args:
            df: The DataFrame containing all dependency columns.

        Returns:
            A pandas Series containing the computed values.

        Raises:
            KeyError: If a dependency column is missing from df.
        """
        # Verify dependencies exist in df
        for dep in cls.dependencies:
            # If dependency is in columns, we are good
            if dep.name in df.columns:
                continue

            # If dependency is Lazy, we assume it can be computed on-demand by __getitem__
            # so we don't raise KeyError here.
            is_lazy_dep = getattr(dep, "lazy", False)
            if is_lazy_dep:
                continue

            raise KeyError(
                f"Dependency '{dep.name}' not found in DataFrame. "
                f"Ensure dependencies are computed before '{cls.name}'."
            )
        return cls.calculate(df)

    @classmethod
    def get_hash_components(cls) -> Dict[str, str]:
        """
        Get the hash components of the variable definition.

        Returns:
            Dict with keys: calc, deps, attrs, meta.
        """
        # Calc: Source code of calculate method
        try:
            source = inspect.getsource(cls.calculate)
        except (OSError, TypeError):
            source = ""
        calc_hash = hashlib.sha256(source.encode("utf-8")).hexdigest()

        # Deps: Recursive hash of dependencies
        # We combine the full hashes of all dependencies
        dep_hashes = []
        for dep in cls.dependencies:
            # Recursively get hash (we use a flattened version generally, 
            # but here we just need a signature of the dependency state).
            # To avoid infinite recursion in malformed cyclic graphs (though resolve handles that),
            # we rely on the fact that dependencies must be solved variables.
            # Ideally, we want the hash of the dependency variable itself.

            # Note: A dependency change should ripple up. 
            # We use the dependency's class name + its own component hashes
            try:
                d_comps = dep.get_hash_components()
                # Encapsulate strictly functional components for dependency hash
                # We exclude 'meta' so that docstring changes don't invalidate downstream calculations
                functional_comps = {k: v for k, v in d_comps.items() if k != "meta"}
                d_combined = hashlib.sha256(json.dumps(functional_comps, sort_keys=True).encode("utf-8")).hexdigest()
                dep_hashes.append(f"{dep.__name__}:{d_combined}")
            except Exception:
                # If dependency is broken or not a Variable class
                dep_hashes.append(f"{dep}:{str(dep)}")

        deps_str = ",".join(sorted(dep_hashes))
        deps_hash = hashlib.sha256(deps_str.encode("utf-8")).hexdigest()

        # Attrs: dtype
        attrs_hash = hashlib.sha256(cls.dtype.encode("utf-8")).hexdigest()

        # Meta: description, lazy
        meta_str = f"{cls.get_description()}|{cls.lazy}"
        meta_hash = hashlib.sha256(meta_str.encode("utf-8")).hexdigest()

        return {
            "calc": calc_hash,
            "deps": deps_hash,
            "attrs": attrs_hash,
            "meta": meta_hash,
        }

    @classmethod
    def info(cls) -> Dict[str, Any]:
        """
        Get variable metadata as a dictionary.

        Returns:
            Dict with name, type, dependencies, dtype, and description.
        """
        return {
            "name": cls.name,
            "type": "derived",
            "dependencies": [d.name for d in cls.dependencies],
            "dtype": cls.dtype,
            "lazy": cls.lazy,
            "description": cls.get_description(),
        }

    def __repr__(self) -> str:
        return f"<{self.__class__.__name__}: {self.name}>"

calculate(df) classmethod

Calculate the derived variable's values.

Override this method in subclasses to define the computation logic. Access dependency values directly from df (e.g., df["gap"]).

Parameters:

Name	Type	Description	Default
df	DataFrame	The current processed DataFrame containing all computed variables so far (including dependencies).	required

Returns:

Type	Description
Series	A pandas Series with the computed values.

Raises:

Type	Description
NotImplementedError	If not overridden in subclass.

Source code in varframe/variables.py

@classmethod
def calculate(cls, df: pd.DataFrame) -> pd.Series:
    """
    Calculate the derived variable's values.

    Override this method in subclasses to define the computation logic.
    Access dependency values directly from df (e.g., `df["gap"]`).

    Args:
        df: The current processed DataFrame containing all computed
            variables so far (including dependencies).

    Returns:
        A pandas Series with the computed values.

    Raises:
        NotImplementedError: If not overridden in subclass.
    """
    raise NotImplementedError(
        f"{cls.__name__} must implement the 'calculate' classmethod"
    )

compute(df) classmethod

Compute the derived variable from the DataFrame.

Dependencies must already exist in df before calling this method. Use VarFrame to handle dependency ordering automatically.

Parameters:

Name	Type	Description	Default
df	DataFrame	The DataFrame containing all dependency columns.	required

Returns:

Type	Description
Series	A pandas Series containing the computed values.

Raises:

Type	Description
KeyError	If a dependency column is missing from df.

Source code in varframe/variables.py

@classmethod
def compute(cls, df: pd.DataFrame) -> pd.Series:
    """
    Compute the derived variable from the DataFrame.

    Dependencies must already exist in df before calling this method.
    Use `VarFrame` to handle dependency ordering automatically.

    Args:
        df: The DataFrame containing all dependency columns.

    Returns:
        A pandas Series containing the computed values.

    Raises:
        KeyError: If a dependency column is missing from df.
    """
    # Verify dependencies exist in df
    for dep in cls.dependencies:
        # If dependency is in columns, we are good
        if dep.name in df.columns:
            continue

        # If dependency is Lazy, we assume it can be computed on-demand by __getitem__
        # so we don't raise KeyError here.
        is_lazy_dep = getattr(dep, "lazy", False)
        if is_lazy_dep:
            continue

        raise KeyError(
            f"Dependency '{dep.name}' not found in DataFrame. "
            f"Ensure dependencies are computed before '{cls.name}'."
        )
    return cls.calculate(df)

get_hash_components() classmethod

Get the hash components of the variable definition.

Returns:

Type	Description
Dict[str, str]	Dict with keys: calc, deps, attrs, meta.

Source code in varframe/variables.py

@classmethod
def get_hash_components(cls) -> Dict[str, str]:
    """
    Get the hash components of the variable definition.

    Returns:
        Dict with keys: calc, deps, attrs, meta.
    """
    # Calc: Source code of calculate method
    try:
        source = inspect.getsource(cls.calculate)
    except (OSError, TypeError):
        source = ""
    calc_hash = hashlib.sha256(source.encode("utf-8")).hexdigest()

    # Deps: Recursive hash of dependencies
    # We combine the full hashes of all dependencies
    dep_hashes = []
    for dep in cls.dependencies:
        # Recursively get hash (we use a flattened version generally, 
        # but here we just need a signature of the dependency state).
        # To avoid infinite recursion in malformed cyclic graphs (though resolve handles that),
        # we rely on the fact that dependencies must be solved variables.
        # Ideally, we want the hash of the dependency variable itself.

        # Note: A dependency change should ripple up. 
        # We use the dependency's class name + its own component hashes
        try:
            d_comps = dep.get_hash_components()
            # Encapsulate strictly functional components for dependency hash
            # We exclude 'meta' so that docstring changes don't invalidate downstream calculations
            functional_comps = {k: v for k, v in d_comps.items() if k != "meta"}
            d_combined = hashlib.sha256(json.dumps(functional_comps, sort_keys=True).encode("utf-8")).hexdigest()
            dep_hashes.append(f"{dep.__name__}:{d_combined}")
        except Exception:
            # If dependency is broken or not a Variable class
            dep_hashes.append(f"{dep}:{str(dep)}")

    deps_str = ",".join(sorted(dep_hashes))
    deps_hash = hashlib.sha256(deps_str.encode("utf-8")).hexdigest()

    # Attrs: dtype
    attrs_hash = hashlib.sha256(cls.dtype.encode("utf-8")).hexdigest()

    # Meta: description, lazy
    meta_str = f"{cls.get_description()}|{cls.lazy}"
    meta_hash = hashlib.sha256(meta_str.encode("utf-8")).hexdigest()

    return {
        "calc": calc_hash,
        "deps": deps_hash,
        "attrs": attrs_hash,
        "meta": meta_hash,
    }

info() classmethod

Get variable metadata as a dictionary.

Returns:

Type	Description
Dict[str, Any]	Dict with name, type, dependencies, dtype, and description.

Source code in varframe/variables.py

@classmethod
def info(cls) -> Dict[str, Any]:
    """
    Get variable metadata as a dictionary.

    Returns:
        Dict with name, type, dependencies, dtype, and description.
    """
    return {
        "name": cls.name,
        "type": "derived",
        "dependencies": [d.name for d in cls.dependencies],
        "dtype": cls.dtype,
        "lazy": cls.lazy,
        "description": cls.get_description(),
    }

Machine Learning

varframe.BaseModel

A declarative base class for defining ML models.

Define subclasses with class attributes to specify input features, target variable, model type, and hyperparameters.

Class Attributes

name (str): Unique identifier for the model. input_vars (List[VariableType]): Variables used as features (X). target_var (VariableType): Variable to predict (y). model_class (Type): The model class (e.g., RandomForestRegressor). model_params (Dict): Hyperparameters passed to model_class(). model (Any): The trained model instance (set after training). is_trained (bool): Whether the model has been trained.

Example

class GapPredictor(BaseModel): ... name = "gap_predictor" ... input_vars = [Lap, Gap, TireAge] ... target_var = GapDelta ... model_class = RandomForestRegressor ... model_params = {"n_estimators": 100}

Source code in varframe/models.py

class BaseModel:
    """
    A declarative base class for defining ML models.

    Define subclasses with class attributes to specify input features,
    target variable, model type, and hyperparameters.

    Class Attributes:
        name (str): Unique identifier for the model.
        input_vars (List[VariableType]): Variables used as features (X).
        target_var (VariableType): Variable to predict (y).
        model_class (Type): The model class (e.g., RandomForestRegressor).
        model_params (Dict): Hyperparameters passed to model_class().
        model (Any): The trained model instance (set after training).
        is_trained (bool): Whether the model has been trained.

    Example:
        >>> class GapPredictor(BaseModel):
        ...     name = "gap_predictor"
        ...     input_vars = [Lap, Gap, TireAge]
        ...     target_var = GapDelta
        ...     model_class = RandomForestRegressor
        ...     model_params = {"n_estimators": 100}
    """

    name: ClassVar[str] = ""
    input_vars: ClassVar[List[VariableType]] = []
    target_var: ClassVar[Optional[VariableType]] = None
    model_class: ClassVar[Optional[Type[Any]]] = None
    model_params: ClassVar[Dict[str, Any]] = {}
    model: ClassVar[Any] = None
    is_trained: ClassVar[bool] = False

    def __init_subclass__(cls, **kwargs: Any) -> None:
        """Initialize subclass with defaults."""
        super().__init_subclass__(**kwargs)
        if not cls.name and cls.__name__ != "BaseModel":
            cls.name = cls.__name__.lower()
        if "input_vars" not in cls.__dict__:
            cls.input_vars = []
        if "model_params" not in cls.__dict__:
            cls.model_params = {}
        cls.model = None
        cls.is_trained = False

    @classmethod
    def get_description(cls) -> str:
        """Get model description from class docstring."""
        if cls.__doc__:
            return cls.__doc__.strip().split("\n")[0]
        return ""

    @classmethod
    def get_input_names(cls) -> List[str]:
        """Get list of input variable names."""
        return [v.name for v in cls.input_vars]

    @classmethod
    def get_target_name(cls) -> Optional[str]:
        """Get target variable name."""
        return cls.target_var.name if cls.target_var else None

    @classmethod
    def prepare_X(cls, vf: Union["VarFrame", pd.DataFrame]) -> pd.DataFrame:
        """Extract input features from a DataFrame."""
        input_names = cls.get_input_names()
        missing = [n for n in input_names if n not in vf.columns]
        if missing:
            raise KeyError(f"Missing input variables: {missing}")

        X = vf[input_names]
        if hasattr(X, "to_ml"):
            X = X.to_ml()
        return X

    @classmethod
    def prepare_y(cls, vf: Union["VarFrame", pd.DataFrame]) -> pd.Series:
        """Extract target variable from a DataFrame."""
        if cls.target_var is None:
            raise ValueError(f"{cls.__name__} has no target_var defined")

        target_name = cls.get_target_name()
        if target_name not in vf.columns:
            raise KeyError(f"Missing target variable: {target_name}")

        return vf[target_name]

    @classmethod
    def train(
        cls,
        vf: Union["VarFrame", pd.DataFrame],
        **fit_kwargs: Any,
    ) -> None:
        """Train the model on the provided data."""
        if cls.model_class is None:
            raise ValueError(
                f"{cls.__name__} must define 'model_class'. "
                "Example: model_class = RandomForestRegressor"
            )

        cls.model = cls.model_class(**cls.model_params)
        X = cls.prepare_X(vf)
        y = cls.prepare_y(vf)
        cls.model.fit(X, y, **fit_kwargs)
        cls.is_trained = True

    @classmethod
    def train_with_validation(
        cls,
        train_vf: Union["VarFrame", pd.DataFrame],
        val_vf: Union["VarFrame", pd.DataFrame],
        **fit_kwargs: Any,
    ) -> None:
        """Train with a validation set (for XGBoost, LightGBM, etc.)."""
        if cls.model_class is None:
            raise ValueError(f"{cls.__name__} must define 'model_class'.")

        cls.model = cls.model_class(**cls.model_params)
        X_train = cls.prepare_X(train_vf)
        y_train = cls.prepare_y(train_vf)
        X_val = cls.prepare_X(val_vf)
        y_val = cls.prepare_y(val_vf)

        fit_params = fit_kwargs.copy()
        model_name = cls.model_class.__name__.lower()

        if any(name in model_name for name in ["xgb", "lgbm", "lightgbm", "catboost"]):
            fit_params.setdefault("eval_set", [(X_val, y_val)])

        cls.model.fit(X_train, y_train, **fit_params)
        cls.is_trained = True

    @classmethod
    def predict(
        cls,
        vf: Union["VarFrame", pd.DataFrame],
        add_to_df: bool = True,
        column_name: Optional[str] = None,
    ) -> pd.Series:
        """Generate predictions."""
        if not cls.is_trained or cls.model is None:
            raise ValueError(f"{cls.__name__} must be trained before calling predict()")

        X = cls.prepare_X(vf)
        predictions = cls.model.predict(X)

        col_name = column_name or f"{cls.name}_pred"
        result = pd.Series(predictions, index=vf.index, name=col_name)

        if add_to_df and hasattr(vf, "_variables"):
            vf[col_name] = result

        return result

    @classmethod
    def evaluate(
        cls,
        vf: Union["VarFrame", pd.DataFrame],
        metrics: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, float]:
        """Evaluate model performance."""
        from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score

        if metrics is None:
            metrics = {
                "mse": mean_squared_error,
                "mae": mean_absolute_error,
                "r2": r2_score,
            }

        y_true = cls.prepare_y(vf)
        y_pred = cls.predict(vf, add_to_df=False)

        results = {}
        for name, metric_fn in metrics.items():
            mask = ~(y_true.isna() | pd.Series(y_pred).isna())
            results[name] = metric_fn(y_true[mask], y_pred[mask])

        return results

    @classmethod
    def info(cls) -> Dict[str, Any]:
        """Get model metadata as a dictionary."""
        return {
            "name": cls.name,
            "description": cls.get_description(),
            "input_vars": cls.get_input_names(),
            "target_var": cls.get_target_name(),
            "model_class": cls.model_class.__name__ if cls.model_class else None,
            "model_params": cls.model_params,
            "is_trained": cls.is_trained,
        }

    @classmethod
    def save(cls, path: str) -> None:
        """Save the trained model to disk."""
        import joblib

        if not cls.is_trained:
            raise ValueError(f"{cls.__name__} must be trained before saving")

        joblib.dump(
            {
                "model": cls.model,
                "input_vars": cls.get_input_names(),
                "target_var": cls.get_target_name(),
                "name": cls.name,
            },
            path,
        )

    @classmethod
    def load(cls, path: str) -> None:
        """Load a trained model from disk."""
        import joblib

        data = joblib.load(path)
        cls.model = data["model"]
        cls.is_trained = True

    def __repr__(self) -> str:
        return f"<{self.__class__.__name__}: {self.name} (trained={self.is_trained})>"

evaluate(vf, metrics=None) classmethod

Evaluate model performance.

Source code in varframe/models.py

@classmethod
def evaluate(
    cls,
    vf: Union["VarFrame", pd.DataFrame],
    metrics: Optional[Dict[str, Any]] = None,
) -> Dict[str, float]:
    """Evaluate model performance."""
    from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score

    if metrics is None:
        metrics = {
            "mse": mean_squared_error,
            "mae": mean_absolute_error,
            "r2": r2_score,
        }

    y_true = cls.prepare_y(vf)
    y_pred = cls.predict(vf, add_to_df=False)

    results = {}
    for name, metric_fn in metrics.items():
        mask = ~(y_true.isna() | pd.Series(y_pred).isna())
        results[name] = metric_fn(y_true[mask], y_pred[mask])

    return results

get_input_names() classmethod

Get list of input variable names.

Source code in varframe/models.py

@classmethod
def get_input_names(cls) -> List[str]:
    """Get list of input variable names."""
    return [v.name for v in cls.input_vars]

get_target_name() classmethod

Get target variable name.

Source code in varframe/models.py

@classmethod
def get_target_name(cls) -> Optional[str]:
    """Get target variable name."""
    return cls.target_var.name if cls.target_var else None

info() classmethod

Get model metadata as a dictionary.

Source code in varframe/models.py

@classmethod
def info(cls) -> Dict[str, Any]:
    """Get model metadata as a dictionary."""
    return {
        "name": cls.name,
        "description": cls.get_description(),
        "input_vars": cls.get_input_names(),
        "target_var": cls.get_target_name(),
        "model_class": cls.model_class.__name__ if cls.model_class else None,
        "model_params": cls.model_params,
        "is_trained": cls.is_trained,
    }

load(path) classmethod

Load a trained model from disk.

Source code in varframe/models.py

@classmethod
def load(cls, path: str) -> None:
    """Load a trained model from disk."""
    import joblib

    data = joblib.load(path)
    cls.model = data["model"]
    cls.is_trained = True

predict(vf, add_to_df=True, column_name=None) classmethod

Generate predictions.

Source code in varframe/models.py

@classmethod
def predict(
    cls,
    vf: Union["VarFrame", pd.DataFrame],
    add_to_df: bool = True,
    column_name: Optional[str] = None,
) -> pd.Series:
    """Generate predictions."""
    if not cls.is_trained or cls.model is None:
        raise ValueError(f"{cls.__name__} must be trained before calling predict()")

    X = cls.prepare_X(vf)
    predictions = cls.model.predict(X)

    col_name = column_name or f"{cls.name}_pred"
    result = pd.Series(predictions, index=vf.index, name=col_name)

    if add_to_df and hasattr(vf, "_variables"):
        vf[col_name] = result

    return result

save(path) classmethod

Save the trained model to disk.

Source code in varframe/models.py

@classmethod
def save(cls, path: str) -> None:
    """Save the trained model to disk."""
    import joblib

    if not cls.is_trained:
        raise ValueError(f"{cls.__name__} must be trained before saving")

    joblib.dump(
        {
            "model": cls.model,
            "input_vars": cls.get_input_names(),
            "target_var": cls.get_target_name(),
            "name": cls.name,
        },
        path,
    )

train(vf, **fit_kwargs) classmethod

Train the model on the provided data.

Source code in varframe/models.py

@classmethod
def train(
    cls,
    vf: Union["VarFrame", pd.DataFrame],
    **fit_kwargs: Any,
) -> None:
    """Train the model on the provided data."""
    if cls.model_class is None:
        raise ValueError(
            f"{cls.__name__} must define 'model_class'. "
            "Example: model_class = RandomForestRegressor"
        )

    cls.model = cls.model_class(**cls.model_params)
    X = cls.prepare_X(vf)
    y = cls.prepare_y(vf)
    cls.model.fit(X, y, **fit_kwargs)
    cls.is_trained = True

train_with_validation(train_vf, val_vf, **fit_kwargs) classmethod

Train with a validation set (for XGBoost, LightGBM, etc.).

Source code in varframe/models.py

@classmethod
def train_with_validation(
    cls,
    train_vf: Union["VarFrame", pd.DataFrame],
    val_vf: Union["VarFrame", pd.DataFrame],
    **fit_kwargs: Any,
) -> None:
    """Train with a validation set (for XGBoost, LightGBM, etc.)."""
    if cls.model_class is None:
        raise ValueError(f"{cls.__name__} must define 'model_class'.")

    cls.model = cls.model_class(**cls.model_params)
    X_train = cls.prepare_X(train_vf)
    y_train = cls.prepare_y(train_vf)
    X_val = cls.prepare_X(val_vf)
    y_val = cls.prepare_y(val_vf)

    fit_params = fit_kwargs.copy()
    model_name = cls.model_class.__name__.lower()

    if any(name in model_name for name in ["xgb", "lgbm", "lightgbm", "catboost"]):
        fit_params.setdefault("eval_set", [(X_val, y_val)])

    cls.model.fit(X_train, y_train, **fit_params)
    cls.is_trained = True

varframe.ModelVariable

Bases: DerivedVariable

A derived variable computed via model inference.

Class Attributes

name (str): The name of the prediction variable. model_class (Type[BaseModel]): The model class to use for predictions. dependencies: Auto-populated from model's input_vars.

Example

class PredictedGap(ModelVariable): ... name = "predicted_gap" ... model_class = GapPredictor

Source code in varframe/models.py

class ModelVariable(DerivedVariable):
    """
    A derived variable computed via model inference.

    Class Attributes:
        name (str): The name of the prediction variable.
        model_class (Type[BaseModel]): The model class to use for predictions.
        dependencies: Auto-populated from model's input_vars.

    Example:
        >>> class PredictedGap(ModelVariable):
        ...     name = "predicted_gap"
        ...     model_class = GapPredictor
    """

    model_class: ClassVar[Optional[Type[BaseModel]]] = None

    def __init_subclass__(cls, **kwargs: Any) -> None:
        """Initialize and validate ModelVariable subclass."""
        super().__init_subclass__(**kwargs)

        if cls.model_class is not None:
            cls.dependencies = list(cls.model_class.input_vars)

    @classmethod
    def calculate(
        cls,
        df: pd.DataFrame,
        suppress_warnings: bool = False,
    ) -> pd.Series:
        """
        Calculate predictions using the associated model.

        If the model is not trained, it will be automatically trained
        using the available data in the DataFrame (with a warning).
        """
        if cls.model_class is None:
            raise ValueError(f"{cls.__name__} must define 'model_class' attribute")

        model_name = cls.model_class.name

        # Auto-train if model is not trained
        if not cls.model_class.is_trained:
            target_name = (
                cls.model_class.target_var.name
                if cls.model_class.target_var
                else "unknown"
            )

            if cls.model_class.target_var is None:
                raise ValueError(
                    f"Cannot auto-train model '{model_name}': no target_var defined"
                )

            if target_name not in df.columns:
                raise ValueError(
                    f"Cannot auto-train model '{model_name}': "
                    f"target variable '{target_name}' not in DataFrame."
                )

            VFConfig.check_permission(
                ImplicitOperation.TRAIN_MODEL,
                f"Cannot auto-train model '{model_name}'. "
                "Set VFConfig.allow_implicit_train = True to enable.",
            )

            if not suppress_warnings:
                VFConfig.warn(
                    ImplicitOperation.TRAIN_MODEL,
                    f"Auto-training model '{model_name}' (target: {target_name}).",
                    stacklevel=4,
                )

            train_data = df.dropna()
            cls.model_class.train(train_data)

        # Check permission and warn for inference
        VFConfig.check_permission(
            ImplicitOperation.INFER_MODEL,
            f"Cannot perform inference with model '{model_name}'.",
        )

        if not suppress_warnings:
            VFConfig.warn(
                ImplicitOperation.INFER_MODEL,
                f"Performing inference with model '{model_name}'.",
                stacklevel=4,
            )

        return cls.model_class.predict(df, add_to_df=False)

    @classmethod
    def info(cls) -> Dict[str, Any]:
        """Get variable metadata including model info."""
        base_info = super().info()
        base_info["model"] = cls.model_class.name if cls.model_class else None
        base_info["type"] = "model_prediction"
        return base_info

calculate(df, suppress_warnings=False) classmethod

Calculate predictions using the associated model.

If the model is not trained, it will be automatically trained using the available data in the DataFrame (with a warning).

Source code in varframe/models.py

@classmethod
def calculate(
    cls,
    df: pd.DataFrame,
    suppress_warnings: bool = False,
) -> pd.Series:
    """
    Calculate predictions using the associated model.

    If the model is not trained, it will be automatically trained
    using the available data in the DataFrame (with a warning).
    """
    if cls.model_class is None:
        raise ValueError(f"{cls.__name__} must define 'model_class' attribute")

    model_name = cls.model_class.name

    # Auto-train if model is not trained
    if not cls.model_class.is_trained:
        target_name = (
            cls.model_class.target_var.name
            if cls.model_class.target_var
            else "unknown"
        )

        if cls.model_class.target_var is None:
            raise ValueError(
                f"Cannot auto-train model '{model_name}': no target_var defined"
            )

        if target_name not in df.columns:
            raise ValueError(
                f"Cannot auto-train model '{model_name}': "
                f"target variable '{target_name}' not in DataFrame."
            )

        VFConfig.check_permission(
            ImplicitOperation.TRAIN_MODEL,
            f"Cannot auto-train model '{model_name}'. "
            "Set VFConfig.allow_implicit_train = True to enable.",
        )

        if not suppress_warnings:
            VFConfig.warn(
                ImplicitOperation.TRAIN_MODEL,
                f"Auto-training model '{model_name}' (target: {target_name}).",
                stacklevel=4,
            )

        train_data = df.dropna()
        cls.model_class.train(train_data)

    # Check permission and warn for inference
    VFConfig.check_permission(
        ImplicitOperation.INFER_MODEL,
        f"Cannot perform inference with model '{model_name}'.",
    )

    if not suppress_warnings:
        VFConfig.warn(
            ImplicitOperation.INFER_MODEL,
            f"Performing inference with model '{model_name}'.",
            stacklevel=4,
        )

    return cls.model_class.predict(df, add_to_df=False)

info() classmethod

Get variable metadata including model info.

Source code in varframe/models.py

@classmethod
def info(cls) -> Dict[str, Any]:
    """Get variable metadata including model info."""
    base_info = super().info()
    base_info["model"] = cls.model_class.name if cls.model_class else None
    base_info["type"] = "model_prediction"
    return base_info

varframe.ModelRegistry

Central registry for managing multiple models.

Example

registry = ModelRegistry() registry.register(GapPredictor) registry.train_all(training_vf) registry.save_all("models/")

Source code in varframe/models.py

class ModelRegistry:
    """
    Central registry for managing multiple models.

    Example:
        >>> registry = ModelRegistry()
        >>> registry.register(GapPredictor)
        >>> registry.train_all(training_vf)
        >>> registry.save_all("models/")
    """

    def __init__(self) -> None:
        """Initialize an empty registry."""
        self._models: Dict[str, Type[BaseModel]] = {}

    def register(self, model_class: Type[BaseModel]) -> None:
        """Register a model class."""
        self._models[model_class.name] = model_class

    def get(self, name: str) -> Optional[Type[BaseModel]]:
        """Get a model class by name."""
        return self._models.get(name)

    def list_models(self) -> List[str]:
        """Get list of registered model names."""
        return list(self._models.keys())

    def train_all(
        self,
        vf: Union["VarFrame", pd.DataFrame],
        models: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, bool]:
        """Train all registered models."""
        models = models or {}
        results = {}

        for name, model_class in self._models.items():
            try:
                model_class.train(vf)
                results[name] = True
            except Exception as e:
                print(f"Failed to train {name}: {e}")
                results[name] = False

        return results

    def evaluate_all(
        self, vf: Union["VarFrame", pd.DataFrame]
    ) -> Dict[str, Dict[str, float]]:
        """Evaluate all trained models."""
        results = {}
        for name, model_class in self._models.items():
            if model_class.is_trained:
                results[name] = model_class.evaluate(vf)
        return results

    def save_all(self, directory: str) -> None:
        """Save all trained models to a directory."""
        import os

        os.makedirs(directory, exist_ok=True)

        for name, model_class in self._models.items():
            if model_class.is_trained:
                path = os.path.join(directory, f"{name}.joblib")
                model_class.save(path)

    def load_all(self, directory: str) -> None:
        """Load all models from a directory."""
        import os

        for name, model_class in self._models.items():
            path = os.path.join(directory, f"{name}.joblib")
            if os.path.exists(path):
                model_class.load(path)

    def describe(self) -> pd.DataFrame:
        """Generate a summary DataFrame of all registered models."""
        data = [model_class.info() for model_class in self._models.values()]
        return pd.DataFrame(data)

describe()

Generate a summary DataFrame of all registered models.

Source code in varframe/models.py

def describe(self) -> pd.DataFrame:
    """Generate a summary DataFrame of all registered models."""
    data = [model_class.info() for model_class in self._models.values()]
    return pd.DataFrame(data)

evaluate_all(vf)

Evaluate all trained models.

Source code in varframe/models.py

def evaluate_all(
    self, vf: Union["VarFrame", pd.DataFrame]
) -> Dict[str, Dict[str, float]]:
    """Evaluate all trained models."""
    results = {}
    for name, model_class in self._models.items():
        if model_class.is_trained:
            results[name] = model_class.evaluate(vf)
    return results

get(name)

Get a model class by name.

Source code in varframe/models.py

def get(self, name: str) -> Optional[Type[BaseModel]]:
    """Get a model class by name."""
    return self._models.get(name)

list_models()

Get list of registered model names.

Source code in varframe/models.py

def list_models(self) -> List[str]:
    """Get list of registered model names."""
    return list(self._models.keys())

load_all(directory)

Load all models from a directory.

Source code in varframe/models.py

def load_all(self, directory: str) -> None:
    """Load all models from a directory."""
    import os

    for name, model_class in self._models.items():
        path = os.path.join(directory, f"{name}.joblib")
        if os.path.exists(path):
            model_class.load(path)

register(model_class)

Register a model class.

Source code in varframe/models.py

def register(self, model_class: Type[BaseModel]) -> None:
    """Register a model class."""
    self._models[model_class.name] = model_class

save_all(directory)

Save all trained models to a directory.

Source code in varframe/models.py

def save_all(self, directory: str) -> None:
    """Save all trained models to a directory."""
    import os

    os.makedirs(directory, exist_ok=True)

    for name, model_class in self._models.items():
        if model_class.is_trained:
            path = os.path.join(directory, f"{name}.joblib")
            model_class.save(path)

train_all(vf, models=None)

Train all registered models.

Source code in varframe/models.py

def train_all(
    self,
    vf: Union["VarFrame", pd.DataFrame],
    models: Optional[Dict[str, Any]] = None,
) -> Dict[str, bool]:
    """Train all registered models."""
    models = models or {}
    results = {}

    for name, model_class in self._models.items():
        try:
            model_class.train(vf)
            results[name] = True
        except Exception as e:
            print(f"Failed to train {name}: {e}")
            results[name] = False

    return results

Dependencies

varframe.resolve_dependencies(target_variables, include_model_training_deps=True)

Resolve all dependencies for the given variables using DAG traversal.

Performs a topological sort to determine the correct computation order, ensuring all dependencies are computed before their dependents.

Parameters:

Name	Type	Description	Default
target_variables	'VariableList'	List of variable classes to compute.	required
include_model_training_deps	bool	If True, includes dependencies needed for model training (target_var of models).	True

Returns:

Type	Description
'VariableList'	A topologically sorted list of all variables needed to compute
'VariableList'	the target variables, with dependencies before dependents.

Raises:

Type	Description
ValueError	If a circular dependency is detected.

Example

Only specify final variables - dependencies auto-resolved!

all_vars = resolve_dependencies([PredictedGapDelta])

Returns: [Lap, Gap, TireAge, GapDelta, PredictedGapDelta]

Source code in varframe/dependencies.py

def resolve_dependencies(
    target_variables: "VariableList",
    include_model_training_deps: bool = True,
) -> "VariableList":
    """
    Resolve all dependencies for the given variables using DAG traversal.

    Performs a topological sort to determine the correct computation order,
    ensuring all dependencies are computed before their dependents.

    Args:
        target_variables: List of variable classes to compute.
        include_model_training_deps: If True, includes dependencies needed
            for model training (target_var of models).

    Returns:
        A topologically sorted list of all variables needed to compute
        the target variables, with dependencies before dependents.

    Raises:
        ValueError: If a circular dependency is detected.

    Example:
        >>> # Only specify final variables - dependencies auto-resolved!
        >>> all_vars = resolve_dependencies([PredictedGapDelta])
        >>> # Returns: [Lap, Gap, TireAge, GapDelta, PredictedGapDelta]
    """
    # Collect all dependencies recursively
    all_vars: Set["VariableType"] = set()
    visiting: Set["VariableType"] = set()  # For cycle detection

    def collect(var: "VariableType") -> None:
        """Recursively collect all dependencies for a variable."""
        if var in all_vars:
            return
        if var in visiting:
            raise ValueError(f"Circular dependency detected involving '{var.name}'")

        visiting.add(var)

        # Get direct dependencies
        deps: List["VariableType"] = []

        # DerivedVariable dependencies
        if hasattr(var, "dependencies") and var.dependencies:
            deps.extend(var.dependencies)

        # ModelVariable: also needs model's input_vars for prediction
        # and target_var for training (if include_model_training_deps)
        if hasattr(var, "model_class") and var.model_class is not None:
            model_cls = var.model_class
            if hasattr(model_cls, "input_vars") and model_cls.input_vars:
                deps.extend(model_cls.input_vars)
            if include_model_training_deps:
                if hasattr(model_cls, "target_var") and model_cls.target_var:
                    deps.append(model_cls.target_var)

        # Recursively collect dependencies
        for dep in deps:
            collect(dep)

        visiting.remove(var)
        all_vars.add(var)

    # Collect from all target variables
    for var in target_variables:
        collect(var)

    # Topological sort using Kahn's algorithm
    return _topological_sort(all_vars)

Configuration

varframe.VFConfig

Global configuration for VarFrame warnings and implicit operations.

Controls whether implicit operations (like auto-training models or computing variables not in _variables) issue warnings or are blocked entirely.

This is a static configuration class - all attributes are class-level.

Class Attributes

warn_add_variable_no_compute (bool): Warn when adding variable without computing. warn_add_variable_compute (bool): Warn when computing variable not in _variables. warn_train_model (bool): Warn when auto-training a model. warn_infer_model (bool): Warn when inferring with a model implicitly. allow_implicit_train (bool): Allow implicit model training. allow_implicit_infer (bool): Allow implicit model inference. allow_implicit_compute (bool): Allow implicit variable computation. warnings_enabled (bool): Master switch for all warnings.

Example

Disable all warnings

VFConfig.warnings_enabled = False

Block implicit model training (raises RuntimeError)

VFConfig.allow_implicit_train = False

Suppress specific warning type

VFConfig.warn_train_model = False

Context manager for temporary suppression

with VFConfig.suppress_warnings(): ... vf.resolve(PredictedGapDelta)

Reset to defaults

VFConfig.reset()

Source code in varframe/config.py

class VFConfig:
    """
    Global configuration for VarFrame warnings and implicit operations.

    Controls whether implicit operations (like auto-training models or computing
    variables not in _variables) issue warnings or are blocked entirely.

    This is a static configuration class - all attributes are class-level.

    Class Attributes:
        warn_add_variable_no_compute (bool): Warn when adding variable without computing.
        warn_add_variable_compute (bool): Warn when computing variable not in _variables.
        warn_train_model (bool): Warn when auto-training a model.
        warn_infer_model (bool): Warn when inferring with a model implicitly.
        allow_implicit_train (bool): Allow implicit model training.
        allow_implicit_infer (bool): Allow implicit model inference.
        allow_implicit_compute (bool): Allow implicit variable computation.
        warnings_enabled (bool): Master switch for all warnings.

    Example:
        >>> # Disable all warnings
        >>> VFConfig.warnings_enabled = False
        >>>
        >>> # Block implicit model training (raises RuntimeError)
        >>> VFConfig.allow_implicit_train = False
        >>>
        >>> # Suppress specific warning type
        >>> VFConfig.warn_train_model = False
        >>>
        >>> # Context manager for temporary suppression
        >>> with VFConfig.suppress_warnings():
        ...     vf.resolve(PredictedGapDelta)
        >>>
        >>> # Reset to defaults
        >>> VFConfig.reset()
    """

    # Warning flags - control which operations emit warnings
    warn_add_variable_no_compute: ClassVar[bool] = True
    warn_add_variable_compute: ClassVar[bool] = True
    warn_train_model: ClassVar[bool] = True
    warn_infer_model: ClassVar[bool] = True

    # Permission flags - if False, raises error instead of warning
    allow_implicit_train: ClassVar[bool] = True
    allow_implicit_infer: ClassVar[bool] = True
    allow_implicit_compute: ClassVar[bool] = True

    # Master switches
    warnings_enabled: ClassVar[bool] = True

    # Internal state for context manager
    _suppressed: ClassVar[bool] = False

    @classmethod
    def warn(
        cls,
        operation: ImplicitOperation,
        message: str,
        stacklevel: int = 3,
    ) -> None:
        """
        Issue a warning for an implicit operation if configured.

        Args:
            operation: The type of implicit operation.
            message: The warning message.
            stacklevel: Stack level for the warning (default 3 for typical call depth).
        """
        if cls._suppressed or not cls.warnings_enabled:
            return

        should_warn = {
            ImplicitOperation.ADD_VARIABLE_NO_COMPUTE: cls.warn_add_variable_no_compute,
            ImplicitOperation.ADD_VARIABLE_COMPUTE: cls.warn_add_variable_compute,
            ImplicitOperation.TRAIN_MODEL: cls.warn_train_model,
            ImplicitOperation.INFER_MODEL: cls.warn_infer_model,
        }.get(operation, True)

        if should_warn:
            # Use orange/yellow color for visibility in terminal
            colored_msg = f"\033[93m⚠ {message}\033[0m"
            warnings.warn(colored_msg, UserWarning, stacklevel=stacklevel)

    @classmethod
    def check_permission(cls, operation: ImplicitOperation, context: str = "") -> None:
        """
        Check if an implicit operation is allowed.

        Args:
            operation: The type of implicit operation.
            context: Additional context for the error message.

        Raises:
            RuntimeError: If the operation is not allowed.
        """
        permission_map = {
            ImplicitOperation.TRAIN_MODEL: (
                cls.allow_implicit_train,
                "Implicit model training is disabled",
            ),
            ImplicitOperation.INFER_MODEL: (
                cls.allow_implicit_infer,
                "Implicit model inference is disabled",
            ),
            ImplicitOperation.ADD_VARIABLE_COMPUTE: (
                cls.allow_implicit_compute,
                "Implicit variable computation is disabled",
            ),
            ImplicitOperation.ADD_VARIABLE_NO_COMPUTE: (
                cls.allow_implicit_compute,
                "Implicit variable addition is disabled",
            ),
        }

        allowed, base_msg = permission_map.get(operation, (True, ""))
        if not allowed:
            full_msg = f"{base_msg}. {context}" if context else base_msg
            raise RuntimeError(full_msg)

    @classmethod
    def suppress_warnings(cls) -> "_WarningSuppressionContext":
        """
        Context manager for temporarily suppressing all VF warnings.

        Returns:
            A context manager that suppresses warnings while active.

        Example:
            >>> with VFConfig.suppress_warnings():
            ...     vf.resolve(PredictedGapDelta)  # No warnings issued
        """
        return _WarningSuppressionContext()

    @classmethod
    def reset(cls) -> None:
        """Reset all configuration to default values."""
        cls.warn_add_variable_no_compute = True
        cls.warn_add_variable_compute = True
        cls.warn_train_model = True
        cls.warn_infer_model = True
        cls.allow_implicit_train = True
        cls.allow_implicit_infer = True
        cls.allow_implicit_compute = True
        cls.warnings_enabled = True
        cls._suppressed = False

    @classmethod
    def null_context(cls) -> "_NullContext":
        """
        Returns a context manager that does nothing.
        """
        return _NullContext()

check_permission(operation, context='') classmethod

Check if an implicit operation is allowed.

Parameters:

Name	Type	Description	Default
operation	ImplicitOperation	The type of implicit operation.	required
context	str	Additional context for the error message.	''

Raises:

Type	Description
RuntimeError	If the operation is not allowed.

Source code in varframe/config.py

@classmethod
def check_permission(cls, operation: ImplicitOperation, context: str = "") -> None:
    """
    Check if an implicit operation is allowed.

    Args:
        operation: The type of implicit operation.
        context: Additional context for the error message.

    Raises:
        RuntimeError: If the operation is not allowed.
    """
    permission_map = {
        ImplicitOperation.TRAIN_MODEL: (
            cls.allow_implicit_train,
            "Implicit model training is disabled",
        ),
        ImplicitOperation.INFER_MODEL: (
            cls.allow_implicit_infer,
            "Implicit model inference is disabled",
        ),
        ImplicitOperation.ADD_VARIABLE_COMPUTE: (
            cls.allow_implicit_compute,
            "Implicit variable computation is disabled",
        ),
        ImplicitOperation.ADD_VARIABLE_NO_COMPUTE: (
            cls.allow_implicit_compute,
            "Implicit variable addition is disabled",
        ),
    }

    allowed, base_msg = permission_map.get(operation, (True, ""))
    if not allowed:
        full_msg = f"{base_msg}. {context}" if context else base_msg
        raise RuntimeError(full_msg)

null_context() classmethod

Returns a context manager that does nothing.

Source code in varframe/config.py

@classmethod
def null_context(cls) -> "_NullContext":
    """
    Returns a context manager that does nothing.
    """
    return _NullContext()

reset() classmethod

Reset all configuration to default values.

Source code in varframe/config.py

@classmethod
def reset(cls) -> None:
    """Reset all configuration to default values."""
    cls.warn_add_variable_no_compute = True
    cls.warn_add_variable_compute = True
    cls.warn_train_model = True
    cls.warn_infer_model = True
    cls.allow_implicit_train = True
    cls.allow_implicit_infer = True
    cls.allow_implicit_compute = True
    cls.warnings_enabled = True
    cls._suppressed = False

suppress_warnings() classmethod

Context manager for temporarily suppressing all VF warnings.

Returns:

Type	Description
'_WarningSuppressionContext'	A context manager that suppresses warnings while active.

Example

with VFConfig.suppress_warnings(): ... vf.resolve(PredictedGapDelta) # No warnings issued

Source code in varframe/config.py

@classmethod
def suppress_warnings(cls) -> "_WarningSuppressionContext":
    """
    Context manager for temporarily suppressing all VF warnings.

    Returns:
        A context manager that suppresses warnings while active.

    Example:
        >>> with VFConfig.suppress_warnings():
        ...     vf.resolve(PredictedGapDelta)  # No warnings issued
    """
    return _WarningSuppressionContext()

warn(operation, message, stacklevel=3) classmethod

Issue a warning for an implicit operation if configured.

Parameters:

Name	Type	Description	Default
operation	ImplicitOperation	The type of implicit operation.	required
message	str	The warning message.	required
stacklevel	int	Stack level for the warning (default 3 for typical call depth).	3

Source code in varframe/config.py

@classmethod
def warn(
    cls,
    operation: ImplicitOperation,
    message: str,
    stacklevel: int = 3,
) -> None:
    """
    Issue a warning for an implicit operation if configured.

    Args:
        operation: The type of implicit operation.
        message: The warning message.
        stacklevel: Stack level for the warning (default 3 for typical call depth).
    """
    if cls._suppressed or not cls.warnings_enabled:
        return

    should_warn = {
        ImplicitOperation.ADD_VARIABLE_NO_COMPUTE: cls.warn_add_variable_no_compute,
        ImplicitOperation.ADD_VARIABLE_COMPUTE: cls.warn_add_variable_compute,
        ImplicitOperation.TRAIN_MODEL: cls.warn_train_model,
        ImplicitOperation.INFER_MODEL: cls.warn_infer_model,
    }.get(operation, True)

    if should_warn:
        # Use orange/yellow color for visibility in terminal
        colored_msg = f"\033[93m⚠ {message}\033[0m"
        warnings.warn(colored_msg, UserWarning, stacklevel=stacklevel)

varframe.ImplicitOperation

Bases: Enum

Enum for different implicit operations that can be warned or blocked.

These represent operations that happen automatically in the library, which users may want to be notified about or prevent entirely.

Attributes:

Name	Type	Description
ADD_VARIABLE_NO_COMPUTE		Adding a variable to registry without computing it.
ADD_VARIABLE_COMPUTE		Computing a variable not explicitly in the registry.
TRAIN_MODEL		Automatically training a model that hasn't been trained.
INFER_MODEL		Performing inference with a model.

Source code in varframe/config.py

class ImplicitOperation(Enum):
    """
    Enum for different implicit operations that can be warned or blocked.

    These represent operations that happen automatically in the library,
    which users may want to be notified about or prevent entirely.

    Attributes:
        ADD_VARIABLE_NO_COMPUTE: Adding a variable to registry without computing it.
        ADD_VARIABLE_COMPUTE: Computing a variable not explicitly in the registry.
        TRAIN_MODEL: Automatically training a model that hasn't been trained.
        INFER_MODEL: Performing inference with a model.
    """

    ADD_VARIABLE_NO_COMPUTE = auto()
    ADD_VARIABLE_COMPUTE = auto()
    TRAIN_MODEL = auto()
    INFER_MODEL = auto()