Yfinance

def get_yfin_data_online(
    symbol: Annotated[str, "ticker symbol of the company"],
    start_date: Annotated[str, "Start date in yyyy-mm-dd format"],
    end_date: Annotated[str, "End date in yyyy-mm-dd format"],
) -> str:
    """Fetch OHLCV stock data online via yfinance and return as CSV string.

    Args:
        symbol (str): Ticker symbol of the company.
        start_date (str): Start date in YYYY-MM-DD format.
        end_date (str): End date in YYYY-MM-DD format.

    Returns:
        str: CSV string containing stock data with header info.

    Raises:
        ValueError: If `start_date` or `end_date` does not match YYYY-MM-DD, or
            if the ticker symbol is empty.
    """
    _, end_dt = _validate_date_range(start_date, end_date)
    download_end_date = (end_dt + timedelta(days=1)).strftime("%Y-%m-%d")

    candidates = get_yfinance_symbol_candidates(symbol)

    data = pd.DataFrame()
    resolved_symbol = candidates[0]
    last_error: Exception | None = None
    fetched_any_candidate = False
    for candidate in candidates:
        try:
            ticker = yf.Ticker(candidate)
            candidate_data = ticker.history(
                start=start_date, end=download_end_date, auto_adjust=False
            )
        except Exception as exc:
            logger.debug("Failed to fetch history for %s", candidate, exc_info=True)
            last_error = exc
            continue
        fetched_any_candidate = True
        if not candidate_data.empty:
            data = candidate_data
            resolved_symbol = candidate
            break

    # Check if data is empty
    if data.empty:
        tried = describe_symbol_candidates(symbol, candidates)
        if not fetched_any_candidate and last_error is not None:
            raise RuntimeError(
                f"Failed to fetch market data for symbol '{symbol}' (tried: {tried})"
            ) from last_error
        return f"No data found for symbol '{symbol}' (tried: {tried}) between {start_date} and {end_date}"

    # Remove timezone info from index for cleaner output
    if data.index.tz is not None:
        data.index = data.index.tz_localize(None)

    # Round numerical values to 2 decimal places for cleaner display
    numeric_columns = ["Open", "High", "Low", "Close", "Adj Close"]
    for col in numeric_columns:
        if col in data.columns:
            data[col] = data[col].round(2)

    # Convert DataFrame to CSV string
    csv_string = data.to_csv()

    # Add header information
    header = f"# Stock data for {resolved_symbol} from {start_date} to {end_date}\n"
    header += f"# Total records: {len(data)}\n"
    header += f"# Data retrieved on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n"

    return header + csv_string

def get_stock_stats_indicators_window(
    symbol: Annotated[str, "ticker symbol of the company"],
    indicator: Annotated[str, "technical indicator to get the analysis and report of"],
    curr_date: Annotated[str, "The current trading date you are trading on, YYYY-mm-dd"],
    look_back_days: Annotated[int, "how many days to look back"],
) -> str:
    """Calculate and format a specific technical indicator over a window of days.

    Args:
        symbol (str): Ticker symbol of the company.
        indicator (str): Technical indicator to calculate.
        curr_date (str): Current trading date in YYYY-MM-DD format.
        look_back_days (int): Number of days to look back.

    Returns:
        str: Formatted string containing indicator values and a description.

    Raises:
        ValueError: If the requested indicator is not supported, `curr_date`
            does not match YYYY-MM-DD, the symbol is empty, or no market data is
            available.
    """
    best_ind_params = {
        # Moving Averages
        "close_50_sma": (
            "50 SMA: A medium-term trend indicator. "
            "Usage: Identify trend direction and serve as dynamic support/resistance. "
            "Tips: It lags price; combine with faster indicators for timely signals."
        ),
        "close_200_sma": (
            "200 SMA: A long-term trend benchmark. "
            "Usage: Confirm overall market trend and identify golden/death cross setups. "
            "Tips: It reacts slowly; best for strategic trend confirmation rather than frequent trading entries."
        ),
        "close_10_ema": (
            "10 EMA: A responsive short-term average. "
            "Usage: Capture quick shifts in momentum and potential entry points. "
            "Tips: Prone to noise in choppy markets; use alongside longer averages for filtering false signals."
        ),
        # MACD Related
        "macd": (
            "MACD: Computes momentum via differences of EMAs. "
            "Usage: Look for crossovers and divergence as signals of trend changes. "
            "Tips: Confirm with other indicators in low-volatility or sideways markets."
        ),
        "macds": (
            "MACD Signal: An EMA smoothing of the MACD line. "
            "Usage: Use crossovers with the MACD line to trigger trades. "
            "Tips: Should be part of a broader strategy to avoid false positives."
        ),
        "macdh": (
            "MACD Histogram: Shows the gap between the MACD line and its signal. "
            "Usage: Visualize momentum strength and spot divergence early. "
            "Tips: Can be volatile; complement with additional filters in fast-moving markets."
        ),
        # Momentum Indicators
        "rsi": (
            "RSI: Measures momentum to flag overbought/oversold conditions. "
            "Usage: Apply 70/30 thresholds and watch for divergence to signal reversals. "
            "Tips: In strong trends, RSI may remain extreme; always cross-check with trend analysis."
        ),
        # Volatility Indicators
        "boll": (
            "Bollinger Middle: A 20 SMA serving as the basis for Bollinger Bands. "
            "Usage: Acts as a dynamic benchmark for price movement. "
            "Tips: Combine with the upper and lower bands to effectively spot breakouts or reversals."
        ),
        "boll_ub": (
            "Bollinger Upper Band: Typically 2 standard deviations above the middle line. "
            "Usage: Signals potential overbought conditions and breakout zones. "
            "Tips: Confirm signals with other tools; prices may ride the band in strong trends."
        ),
        "boll_lb": (
            "Bollinger Lower Band: Typically 2 standard deviations below the middle line. "
            "Usage: Indicates potential oversold conditions. "
            "Tips: Use additional analysis to avoid false reversal signals."
        ),
        "atr": (
            "ATR: Averages true range to measure volatility. "
            "Usage: Set stop-loss levels and adjust position sizes based on current market volatility. "
            "Tips: It's a reactive measure, so use it as part of a broader risk management strategy."
        ),
        # Volume-Based Indicators
        "vwma": (
            "VWMA: A moving average weighted by volume. "
            "Usage: Confirm trends by integrating price action with volume data. "
            "Tips: Watch for skewed results from volume spikes; use in combination with other volume analyses."
        ),
        "mfi": (
            "MFI: The Money Flow Index is a momentum indicator that uses both price and volume to measure buying and selling pressure. "
            "Usage: Identify overbought (>80) or oversold (<20) conditions and confirm the strength of trends or reversals. "
            "Tips: Use alongside RSI or MACD to confirm signals; divergence between price and MFI can indicate potential reversals."
        ),
    }

    if indicator not in best_ind_params:
        raise ValueError(
            f"Indicator {indicator} is not supported. Please choose from: {list(best_ind_params.keys())}"
        )

    if look_back_days < 0:
        raise ValueError("look_back_days must be >= 0.")

    end_date = curr_date
    curr_date_dt = _parse_yyyy_mm_dd(curr_date, "curr_date")
    before = curr_date_dt - relativedelta(days=look_back_days)

    indicator_data = _get_stock_stats_bulk(symbol, indicator, curr_date)

    current_dt = curr_date_dt
    ind_string = ""
    while current_dt >= before:
        date_str = current_dt.strftime("%Y-%m-%d")
        value = indicator_data.get(date_str, "N/A: Not a trading day (weekend or holiday)")
        ind_string += f"{date_str}: {value}\n"
        current_dt = current_dt - relativedelta(days=1)

    return (
        f"## {indicator} values from {before.strftime('%Y-%m-%d')} to {end_date}:\n\n"
        + ind_string
        + "\n\n"
        + best_ind_params.get(indicator, "No description available.")
    )

def get_fundamentals(
    ticker: Annotated[str, "ticker symbol of the company"],
    curr_date: Annotated[str | None, "current trading date in YYYY-MM-DD format"] = None,
) -> str:
    """Get company fundamentals overview from yfinance.

    Args:
        ticker (str): Ticker symbol of the company.
        curr_date (str | None, optional): Current date used as a point-in-time
            boundary. Current yfinance snapshot metrics are omitted for
            historical dates because Yahoo Finance does not expose their
            historical availability through this endpoint. Defaults to None.

    Returns:
        str: Formatted string containing fundamentals overview.
    """
    _as_of_datetime(curr_date)

    candidates = get_yfinance_symbol_candidates(ticker)
    info = {}
    resolved_ticker = candidates[0]
    last_error: Exception | None = None
    fetched_any_candidate = False

    for candidate in candidates:
        try:
            ticker_obj = yf.Ticker(candidate)
            candidate_info = ticker_obj.info
        except Exception as exc:
            logger.debug("Failed to fetch fundamentals for %s", candidate, exc_info=True)
            last_error = exc
            continue
        fetched_any_candidate = True
        if _has_meaningful_ticker_info(candidate_info):
            info = candidate_info
            resolved_ticker = candidate
            break

    if not info:
        tried = describe_symbol_candidates(ticker, candidates)
        if not fetched_any_candidate and last_error is not None:
            raise RuntimeError(
                f"Failed to fetch fundamentals for symbol '{ticker}' (tried: {tried})"
            ) from last_error
        return f"No fundamentals data found for symbol '{ticker}' (tried: {tried})"

    profile_fields = [
        ("Name", info.get("longName")),
        ("Sector", info.get("sector")),
        ("Industry", info.get("industry")),
    ]
    snapshot_fields = [
        ("Name", info.get("longName")),
        ("Sector", info.get("sector")),
        ("Industry", info.get("industry")),
        ("Market Cap", info.get("marketCap")),
        ("PE Ratio (TTM)", info.get("trailingPE")),
        ("Forward PE", info.get("forwardPE")),
        ("PEG Ratio", info.get("pegRatio")),
        ("Price to Book", info.get("priceToBook")),
        ("EPS (TTM)", info.get("trailingEps")),
        ("Forward EPS", info.get("forwardEps")),
        ("Dividend Yield", info.get("dividendYield")),
        ("Beta", info.get("beta")),
        ("52 Week High", info.get("fiftyTwoWeekHigh")),
        ("52 Week Low", info.get("fiftyTwoWeekLow")),
        ("50 Day Average", info.get("fiftyDayAverage")),
        ("200 Day Average", info.get("twoHundredDayAverage")),
        ("Revenue (TTM)", info.get("totalRevenue")),
        ("Gross Profit", info.get("grossProfits")),
        ("EBITDA", info.get("ebitda")),
        ("Net Income", info.get("netIncomeToCommon")),
        ("Profit Margin", info.get("profitMargins")),
        ("Operating Margin", info.get("operatingMargins")),
        ("Return on Equity", info.get("returnOnEquity")),
        ("Return on Assets", info.get("returnOnAssets")),
        ("Debt to Equity", info.get("debtToEquity")),
        ("Current Ratio", info.get("currentRatio")),
        ("Book Value", info.get("bookValue")),
        ("Free Cash Flow", info.get("freeCashflow")),
    ]

    fields = profile_fields if _is_historical_date(curr_date) else snapshot_fields
    lines = []
    for label, value in fields:
        if value is not None:
            lines.append(f"{label}: {value}")

    header = f"# Company Fundamentals for {resolved_ticker}\n"
    if curr_date is not None:
        header += f"# Current trading date: {curr_date}\n"
    header += f"# Data retrieved on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
    if _is_historical_date(curr_date):
        header += (
            "# Snapshot valuation/market metrics omitted: yfinance.info only "
            "provides current values, not point-in-time historical values.\n"
        )
    header += "\n"

    return header + "\n".join(lines)

def get_balance_sheet(
    ticker: Annotated[str, "ticker symbol of the company"],
    freq: Annotated[str, "frequency of data: 'annual' or 'quarterly'"] = "quarterly",
    curr_date: Annotated[str | None, "current trading date in YYYY-MM-DD format"] = None,
) -> str:
    """Get balance sheet data from yfinance.

    Args:
        ticker (str): Ticker symbol of the company.
        freq (str, optional): Frequency of data ('annual' or 'quarterly'). Defaults to "quarterly".
        curr_date (str | None, optional): Current date used as a point-in-time boundary. Defaults to None.

    Returns:
        str: CSV string containing balance sheet data.
    """
    freq = _normalize_freq(freq)
    _as_of_datetime(curr_date)
    candidates = get_yfinance_symbol_candidates(ticker)
    data = pd.DataFrame()
    resolved_ticker = candidates[0]
    last_error: Exception | None = None
    fetched_any_candidate = False

    for candidate in candidates:
        try:
            ticker_obj = yf.Ticker(candidate)
            candidate_data = (
                ticker_obj.quarterly_balance_sheet
                if freq == "quarterly"
                else ticker_obj.balance_sheet
            )
        except Exception as exc:
            logger.debug("Failed to fetch balance sheet for %s", candidate, exc_info=True)
            last_error = exc
            continue
        fetched_any_candidate = True
        if candidate_data is None or candidate_data.empty:
            continue
        candidate_data = _filter_statement_as_of(candidate_data, curr_date, freq)
        if not candidate_data.empty:
            data = candidate_data
            resolved_ticker = candidate
            break

    if data.empty:
        tried = describe_symbol_candidates(ticker, candidates)
        if not fetched_any_candidate and last_error is not None:
            raise RuntimeError(
                f"Failed to fetch balance sheet for symbol '{ticker}' (tried: {tried})"
            ) from last_error
        return (
            f"No balance sheet data found for symbol '{ticker}' (tried: {tried}) "
            f"as of {curr_date or 'latest'}"
        )

    csv_string = data.to_csv()

    header = f"# Balance Sheet data for {resolved_ticker} ({freq})\n"
    header += _statement_as_of_note(curr_date, freq)
    header += f"# Data retrieved on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n"

    return header + csv_string

def get_cashflow(
    ticker: Annotated[str, "ticker symbol of the company"],
    freq: Annotated[str, "frequency of data: 'annual' or 'quarterly'"] = "quarterly",
    curr_date: Annotated[str | None, "current trading date in YYYY-MM-DD format"] = None,
) -> str:
    """Get cash flow data from yfinance.

    Args:
        ticker (str): Ticker symbol of the company.
        freq (str, optional): Frequency of data ('annual' or 'quarterly'). Defaults to "quarterly".
        curr_date (str | None, optional): Current date used as a point-in-time boundary. Defaults to None.

    Returns:
        str: CSV string containing cash flow data.
    """
    freq = _normalize_freq(freq)
    _as_of_datetime(curr_date)
    candidates = get_yfinance_symbol_candidates(ticker)
    data = pd.DataFrame()
    resolved_ticker = candidates[0]
    last_error: Exception | None = None
    fetched_any_candidate = False

    for candidate in candidates:
        try:
            ticker_obj = yf.Ticker(candidate)
            candidate_data = (
                ticker_obj.quarterly_cashflow if freq == "quarterly" else ticker_obj.cashflow
            )
        except Exception as exc:
            logger.debug("Failed to fetch cash flow for %s", candidate, exc_info=True)
            last_error = exc
            continue
        fetched_any_candidate = True
        if candidate_data is None or candidate_data.empty:
            continue
        candidate_data = _filter_statement_as_of(candidate_data, curr_date, freq)
        if not candidate_data.empty:
            data = candidate_data
            resolved_ticker = candidate
            break

    if data.empty:
        tried = describe_symbol_candidates(ticker, candidates)
        if not fetched_any_candidate and last_error is not None:
            raise RuntimeError(
                f"Failed to fetch cash flow for symbol '{ticker}' (tried: {tried})"
            ) from last_error
        return (
            f"No cash flow data found for symbol '{ticker}' (tried: {tried}) "
            f"as of {curr_date or 'latest'}"
        )

    csv_string = data.to_csv()

    header = f"# Cash Flow data for {resolved_ticker} ({freq})\n"
    header += _statement_as_of_note(curr_date, freq)
    header += f"# Data retrieved on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n"

    return header + csv_string

def get_income_statement(
    ticker: Annotated[str, "ticker symbol of the company"],
    freq: Annotated[str, "frequency of data: 'annual' or 'quarterly'"] = "quarterly",
    curr_date: Annotated[str | None, "current trading date in YYYY-MM-DD format"] = None,
) -> str:
    """Get income statement data from yfinance.

    Args:
        ticker (str): Ticker symbol of the company.
        freq (str, optional): Frequency of data ('annual' or 'quarterly'). Defaults to "quarterly".
        curr_date (str | None, optional): Current date used as a point-in-time boundary. Defaults to None.

    Returns:
        str: CSV string containing income statement data.
    """
    freq = _normalize_freq(freq)
    _as_of_datetime(curr_date)
    candidates = get_yfinance_symbol_candidates(ticker)
    data = pd.DataFrame()
    resolved_ticker = candidates[0]
    last_error: Exception | None = None
    fetched_any_candidate = False

    for candidate in candidates:
        try:
            ticker_obj = yf.Ticker(candidate)
            candidate_data = (
                ticker_obj.quarterly_income_stmt if freq == "quarterly" else ticker_obj.income_stmt
            )
        except Exception as exc:
            logger.debug("Failed to fetch income statement for %s", candidate, exc_info=True)
            last_error = exc
            continue
        fetched_any_candidate = True
        if candidate_data is None or candidate_data.empty:
            continue
        candidate_data = _filter_statement_as_of(candidate_data, curr_date, freq)
        if not candidate_data.empty:
            data = candidate_data
            resolved_ticker = candidate
            break

    if data.empty:
        tried = describe_symbol_candidates(ticker, candidates)
        if not fetched_any_candidate and last_error is not None:
            raise RuntimeError(
                f"Failed to fetch income statement for symbol '{ticker}' (tried: {tried})"
            ) from last_error
        return (
            f"No income statement data found for symbol '{ticker}' (tried: {tried}) "
            f"as of {curr_date or 'latest'}"
        )

    csv_string = data.to_csv()

    header = f"# Income Statement data for {resolved_ticker} ({freq})\n"
    header += _statement_as_of_note(curr_date, freq)
    header += f"# Data retrieved on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n"

    return header + csv_string

def get_insider_transactions(
    ticker: Annotated[str, "ticker symbol of the company"],
    curr_date: Annotated[str | None, "current trading date in YYYY-MM-DD format"] = None,
) -> str:
    """Get insider transactions data from yfinance.

    Args:
        ticker (str): Ticker symbol of the company.
        curr_date (str | None, optional): Current date used as a point-in-time
            boundary when transaction date columns are available. Defaults to None.

    Returns:
        str: CSV string containing insider transactions data.
    """
    as_of = _as_of_datetime(curr_date)
    candidates = get_yfinance_symbol_candidates(ticker)
    data = pd.DataFrame()
    resolved_ticker = candidates[0]
    last_error: Exception | None = None
    fetched_any_candidate = False

    for candidate in candidates:
        try:
            ticker_obj = yf.Ticker(candidate)
            candidate_data = ticker_obj.insider_transactions
        except Exception as exc:
            logger.debug("Failed to fetch insider transactions for %s", candidate, exc_info=True)
            last_error = exc
            continue
        fetched_any_candidate = True
        if candidate_data is None or candidate_data.empty:
            continue
        if as_of is not None:
            date_columns = [
                column for column in candidate_data.columns if "date" in str(column).lower()
            ]
            if date_columns:
                dates = pd.to_datetime(candidate_data[date_columns[0]], errors="coerce")
                candidate_data = candidate_data.loc[dates <= pd.Timestamp(as_of)]
        if not candidate_data.empty:
            data = candidate_data
            resolved_ticker = candidate
            break

    if data.empty:
        tried = describe_symbol_candidates(ticker, candidates)
        if not fetched_any_candidate and last_error is not None:
            raise RuntimeError(
                f"Failed to fetch insider transactions for symbol '{ticker}' (tried: {tried})"
            ) from last_error
        return (
            f"No insider transactions data found for symbol '{ticker}' (tried: {tried}) "
            f"as of {curr_date or 'latest'}"
        )

    csv_string = data.to_csv()

    header = f"# Insider Transactions data for {resolved_ticker}\n"
    if curr_date is not None:
        header += f"# Current trading date: {curr_date}\n"
    header += f"# Data retrieved on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n"

    return header + csv_string