utils.py

"""Utility helper functions used across the project.

This module centralizes small helpers such as writing generated code to file,
converting values to floats with defaults, and saving dataframe objects to CSV.
"""
from __future__ import annotations

import os
import tempfile
from datetime import datetime
from typing import Any, Optional, Tuple

__all__ = [
    "write_output",
    "_to_float_or_default",
    "save_df_to_csv",
    "resolve_market_to_ticker",
    "validate_date_range",
    "validate_ticker_symbol",
]


def write_output(code: str, filename: str = "bt_strategy.py") -> None:
    """Write a string of code to a file.

    Args:
        code: The string content to write to disk.
        filename: The filename to use (defaults to `code.py`).
    """
    with open(filename, "w") as f:
        f.write(code)


def _to_float_or_default(value: Any, default: Any) -> float:
    """Convert a possibly-empty value to float or return a default.

    This mirrors behavior in the application code: if the input is None or an
    empty string it returns the provided default cast to float.
    """
    try:
        if value is None or value == "":
            raise ValueError
        return float(value)
    except (ValueError, TypeError):
        return float(default)


def resolve_market_to_ticker(market_name: str, market_to_ticker: dict) -> str:
    """Return a ticker symbol for a market name or ticker string.

    If market_name is a friendly name (in market_to_ticker), returns the mapped
    ticker symbol; otherwise assumes the input is a ticker symbol and returns
    the uppercased stripped version.
    """
    if market_name in list(market_to_ticker.keys()):
        return market_to_ticker[market_name]
    return market_name.upper().strip()


def validate_date_range(selected_start: str, selected_end: str, current_date_str: str) -> Tuple[str, str]:
    """Validate date strings in YYYY-MM-DD format and ensure start <= end <= today.

    Returns a tuple (start, end) as strings. Raises ValueError with a user-facing
    message on invalid input.
    """
    try:
        start_dt = datetime.strptime(selected_start, "%Y-%m-%d")
    except ValueError:
        raise ValueError(f"❌ Error: Start Date must be in format YYYY-MM-DD. Got '{selected_start}'.")
    try:
        end_dt = datetime.strptime(selected_end, "%Y-%m-%d")
    except ValueError:
        raise ValueError(f"❌Error: End Date must be in format YYYY-MM-DD. Got '{selected_end}'.")
    today_dt = datetime.strptime(current_date_str, "%Y-%m-%d")
    if start_dt > end_dt:
        raise ValueError("❌ Error: Start Date must be earlier than or equal to End Date.")
    if end_dt > today_dt:
        raise ValueError(f"❌ Error: End Date cannot be in the future. Today is {current_date_str}.")
    return selected_start, selected_end


def validate_ticker_symbol(tckr_symbl: str) -> str:
    """Validate a ticker symbol using yfinance and return a human-friendly name.

    Raises ValueError on invalid tickers or if yfinance fails to provide info.
    """
    try:
        import yfinance as yf

        ticker_data = yf.Ticker(tckr_symbl)
        info = ticker_data.info
        if not info or "symbol" not in info:
            raise ValueError(f"❌ Error: Invalid ticker symbol '{tckr_symbl}'.")
        # Prefer longName or shortName if available
        return info.get("longName") or info.get("shortName") or tckr_symbl
    except ValueError:
        raise
    except Exception:
        raise ValueError(f"❌ Error: Unable to validate ticker symbol '{tckr_symbl}'.")


def save_df_to_csv(df: Optional[Any], filename: str) -> Optional[str]:
    """Save a pandas DataFrame (or df-like object) to CSV in the system temp dir.

    Returns the full path to the saved file or None when the input could not be
    converted into a non-empty DataFrame.
    """
    if df is None:
        return None

    try:
        import pandas as pd
    except Exception:
        # pandas is required for this helper; if it's not installed return None to
        # fail gracefully so callers can handle the absence (e.g. in limited
        # environments where CSV exports aren't available).
        return None

    if isinstance(df, pd.DataFrame):
        df_to_save = df
    else:
        try:
            df_to_save = pd.DataFrame(df)
        except Exception:
            return None

    if df_to_save.empty:
        return None

    out_dir = tempfile.gettempdir()
    os.makedirs(out_dir, exist_ok=True)
    filepath = os.path.join(out_dir, f"{filename}.csv")
    df_to_save.to_csv(filepath, index=False)
    return filepath