| """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}'.") |
| |
| 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: |
| |
| |
| |
| 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 |
|
|
