stefan-jansen--machine-learning-for-trading/data/exceptions.py

"""Custom exceptions for ML4T data infrastructure.

Provides clear, actionable error messages when data is missing or unavailable.
All exceptions point readers to the appropriate documentation.
"""

from pathlib import Path


class DataNotFoundError(FileNotFoundError):
    """Raised when a required dataset is not found.

    Provides clear instructions for obtaining the missing data.

    Args:
        dataset_name: Human-readable name of the dataset (e.g., "ETF Universe")
        path: Path where the data was expected
        download_script: Name of the download script (e.g., "yfinance_etfs.py")
        requires_api_key: Name of required API key if applicable
        instructions: Custom multi-line instructions (overrides download_script template)
        download_url: Hosted-download URL (e.g., AlgoSeek S3 bucket)
        derivation_notebook: Path/stem of a notebook that derives this dataset,
            for derived datasets where readers may want to follow the computation
        readme: Repo-relative path to the dataset's local README with full
            download instructions (e.g., "data/equities/fundamentals/README.md").
            Falls back to "data/README.md" when not set.

    Example:
        >>> raise DataNotFoundError(
        ...     dataset_name="ETF Universe",
        ...     path=Path("/data/etfs/market/etf_universe.parquet"),
        ...     download_script="yfinance_etfs.py"
        ... )
    """

    def __init__(
        self,
        dataset_name: str,
        path: Path | str,
        download_script: str | None = None,
        requires_api_key: str | None = None,
        instructions: str | None = None,
        download_url: str | None = None,
        derivation_notebook: str | None = None,
        readme: str | None = None,
    ):
        self.dataset_name = dataset_name
        self.path = Path(path)
        self.download_script = download_script
        self.requires_api_key = requires_api_key
        self.instructions = instructions
        self.download_url = download_url
        self.derivation_notebook = derivation_notebook
        self.readme = readme

        message = self._build_message()
        super().__init__(message)

    def _build_message(self) -> str:
        lines = [
            "",
            "=" * 70,
            f"DATA NOT FOUND: {self.dataset_name}",
            "=" * 70,
            "",
            f"Expected location: {self.path}",
            "",
        ]

        if self.instructions:
            lines.extend([self.instructions, ""])
        elif self.download_url:
            lines.extend(
                [
                    "Download from:",
                    f"  {self.download_url}",
                    "",
                    f"Extract to: {self.path.parent}",
                    "",
                ]
            )
        elif self.download_script:
            lines.extend(
                [
                    "To download this dataset (from repo root):",
                    f"  uv run python {self.download_script}",
                    "",
                ]
            )

        if self.derivation_notebook:
            lines.extend(
                [
                    f"How this dataset is built: {self.derivation_notebook}",
                    "",
                ]
            )

        if self.requires_api_key:
            lines.extend(
                [
                    f"Note: Requires {self.requires_api_key} in .env file",
                    "",
                ]
            )

        readme_path = self.readme or "data/README.md"
        lines.extend(
            [
                f"For complete instructions, see: {readme_path}",
                "=" * 70,
            ]
        )

        return "\n".join(lines)


class DownloadError(RuntimeError):
    """Raised when a data download fails.

    Used by download scripts to provide clear failure messages instead of
    silently succeeding with empty or incomplete data.

    Args:
        dataset_name: Human-readable name of the dataset
        reason: Explanation of why the download failed
        suggestion: Optional suggestion for resolving the issue
    """

    def __init__(
        self,
        dataset_name: str,
        reason: str,
        suggestion: str | None = None,
    ):
        self.dataset_name = dataset_name
        self.reason = reason
        self.suggestion = suggestion

        message = self._build_message()
        super().__init__(message)

    def _build_message(self) -> str:
        lines = [
            "",
            "=" * 70,
            f"DOWNLOAD FAILED: {self.dataset_name}",
            "=" * 70,
            "",
            f"Reason: {self.reason}",
            "",
        ]

        if self.suggestion:
            lines.extend(
                [
                    f"Suggestion: {self.suggestion}",
                    "",
                ]
            )

        lines.extend(
            [
                "For help, see: data/README.md",
                "=" * 70,
            ]
        )

        return "\n".join(lines)


class MissingDependencyError(ImportError):
    """Raised when a required dependency is not installed.

    Provides clear installation instructions.

    Args:
        package: Name of the missing package
        install_command: Command to install the package
        purpose: What the package is needed for
    """

    def __init__(
        self,
        package: str,
        install_command: str | None = None,
        purpose: str | None = None,
    ):
        self.package = package
        self.install_command = install_command or f"pip install {package}"
        self.purpose = purpose

        message = self._build_message()
        super().__init__(message)

    def _build_message(self) -> str:
        lines = [
            "",
            f"Missing dependency: {self.package}",
        ]

        if self.purpose:
            lines.append(f"Required for: {self.purpose}")

        lines.extend(
            [
                "",
                f"Install with: {self.install_command}",
            ]
        )

        return "\n".join(lines)