File size: 4,616 Bytes
725cb3b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | """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
|