import logging
import os
import time
import json
import subprocess
from typing import Annotated

import pandas as pd
import yfinance as yf
from stockstats import wrap
from yfinance.exceptions import YFRateLimitError

from .config import get_config
from .symbol_utils import NoMarketDataError, normalize_symbol
from .utils import safe_ticker_component

logger = logging.getLogger(__name__)

# A vendor's latest OHLCV row this many calendar days before the requested date
# is treated as stale. Generous enough to span long holiday weekends, tight
# enough to catch the year-old frames yfinance occasionally returns (#1021).
MAX_OHLCV_STALE_DAYS = 10

_ssh_config_cache = None

def _get_ssh_config():
    """Load SSH credentials for foreign server from vault (cached)."""
    global _ssh_config_cache
    if _ssh_config_cache is not None:
        return _ssh_config_cache
    result = subprocess.run(
        ["bash", "-c",
         'source <(bash /root/.openclaw/workspace/scripts/vault.sh decrypt '
         '/root/.openclaw/workspace/secrets/credentials.env.enc) '
         '&& echo "$FOREIGN_HOST $FOREIGN_PORT $FOREIGN_PASS"'],
        capture_output=True, text=True, timeout=10
    )
    parts = result.stdout.strip().split()
    if len(parts) != 3:
        raise RuntimeError(f"Failed to load SSH config: {result.stderr}")
    _ssh_config_cache = (parts[0], int(parts[1]), parts[2])
    return _ssh_config_cache


def _fetch_via_ssh(url: str, timeout: int = 30) -> str:
    """Fetch a URL through the foreign server via SSH tunnel."""
    host, port, password = _get_ssh_config()
    cmd = [
        "sshpass", "-p", password,
        "ssh", "-o", "StrictHostKeyChecking=no",
        "-o", "ConnectTimeout=15",
        "-p", str(port), f"root@{host}",
        f'curl -s "{url}" -H "User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"'
    ]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    if result.returncode != 0:
        raise RuntimeError(f"SSH fetch failed: {result.stderr}")
    return result.stdout


def _yf_download_via_proxy(symbol: str, start: str, end: str) -> pd.DataFrame:
    """Download OHLCV data via SSH proxy when local Yahoo Finance is rate-limited."""
    import datetime
    # Convert dates to timestamps
    start_ts = int(pd.Timestamp(start).timestamp())
    end_ts = int(pd.Timestamp(end).timestamp())
    
    url = (f"https://query2.finance.yahoo.com/v8/finance/chart/{symbol}"
           f"?period1={start_ts}&period2={end_ts}&interval=1d")
    
    data_str = _fetch_via_ssh(url)
    parsed = json.loads(data_str)
    
    result = parsed.get("chart", {}).get("result", [])
    if not result:
        return pd.DataFrame()
    
    ts = result[0].get("timestamp", [])
    indicators = result[0].get("indicators", {}).get("quote", [{}])[0]
    
    if not ts:
        return pd.DataFrame()
    
    df = pd.DataFrame({
        "Date": pd.to_datetime(ts, unit="s"),
        "Open": indicators.get("open", []),
        "High": indicators.get("high", []),
        "Low": indicators.get("low", []),
        "Close": indicators.get("close", []),
        "Volume": indicators.get("volume", []),
    })
    
    # Convert None values to NaN
    for col in ["Open", "High", "Low", "Close", "Volume"]:
        df[col] = pd.to_numeric(df[col], errors="coerce")
    
    df = df.dropna(subset=["Close"])
    return df


def yf_retry(func, max_retries=3, base_delay=2.0):
    """Execute a yfinance call with exponential backoff on rate limits.

    yfinance raises YFRateLimitError on HTTP 429 responses but does not
    retry them internally. This wrapper adds retry logic specifically
    for rate limits. Other exceptions propagate immediately.
    """
    # Skip local Yahoo Finance entirely - always use SSH proxy
    raise YFRateLimitError("Local Yahoo Finance disabled, use SSH proxy")


def _ensure_date_column(data: pd.DataFrame) -> pd.DataFrame:
    """Normalize the date column to ``Date``.

    Some yfinance builds leave the index unnamed (so ``reset_index()`` yields
    ``index``) or use ``Datetime`` for intraday data. Rename the first
    date-like column so indicators don't silently drop when it isn't ``Date``.
    """
    if "Date" in data.columns:
        return data
    for candidate in ("index", "Datetime", "date"):
        if candidate in data.columns:
            return data.rename(columns={candidate: "Date"})
    return data


def _clean_dataframe(data: pd.DataFrame) -> pd.DataFrame:
    """Normalize a stock DataFrame for stockstats: parse dates, drop invalid rows, fill price gaps."""
    data = _ensure_date_column(data)
    data["Date"] = pd.to_datetime(data["Date"], errors="coerce")
    data = data.dropna(subset=["Date"])

    price_cols = [c for c in ["Open", "High", "Low", "Close", "Volume"] if c in data.columns]
    data[price_cols] = data[price_cols].apply(pd.to_numeric, errors="coerce")
    data = data.dropna(subset=["Close"])
    data[price_cols] = data[price_cols].ffill().bfill()

    return data


def _coerce_ohlcv_dates(data: pd.DataFrame) -> pd.Series:
    """Return parsed dates from an OHLCV frame, whether Date is a column or the index."""
    if "Date" in data.columns:
        return pd.to_datetime(data["Date"], errors="coerce").dropna()
    # yfinance keeps the dates in the index (a DatetimeIndex, sometimes unnamed).
    if isinstance(data.index, pd.DatetimeIndex):
        return pd.Series(pd.to_datetime(data.index, errors="coerce")).dropna()
    # Fallback: expose the index and look for any date-like column.
    df = data.reset_index()
    for col in ("Date", "Datetime", "date", "index"):
        if col in df.columns:
            parsed = pd.to_datetime(df[col], errors="coerce").dropna()
            if not parsed.empty:
                return parsed
    return pd.Series(dtype="datetime64[ns]")


def _assert_ohlcv_not_stale(
    data: pd.DataFrame,
    curr_date: str,
    symbol: str,
    canonical: str | None = None,
    *,
    max_stale_days: int = MAX_OHLCV_STALE_DAYS,
) -> None:
    """Reject OHLCV whose latest row is far older than curr_date.

    Raises NoMarketDataError (with a stale-specific detail) so the router treats
    it like any other "no usable data from this vendor" — try the next vendor,
    then emit one clear unavailable signal. Empty frames are left to the
    caller's existing no-data handling; this guards only the dangerous case of
    present-but-stale rows (a vendor returning a year-old frame that would
    otherwise feed wrong prices to the agent, #1021).
    """
    if data is None or data.empty:
        return
    requested = pd.to_datetime(curr_date, errors="coerce")
    if pd.isna(requested):
        return
    requested = requested.normalize()
    dates = _coerce_ohlcv_dates(data)
    if dates.empty:
        return
    latest = dates.max().normalize()
    stale_days = (requested - latest).days
    if stale_days > max_stale_days:
        raise NoMarketDataError(
            symbol,
            canonical,
            f"latest row is {latest.date()}, {stale_days} days before the "
            f"requested {requested.date()} (stale) — refusing to use it",
        )


def _load_ohlcv_alpha_vantage(symbol: str, curr_date: str) -> pd.DataFrame:
    """Fetch OHLCV data from Alpha Vantage, filtered to prevent look-ahead bias."""
    from .alpha_vantage_common import _make_api_request
    from io import StringIO

    canonical = normalize_symbol(symbol)
    curr_date_dt = pd.to_datetime(curr_date)
    today_date = pd.Timestamp.today()
    start_date = today_date - pd.DateOffset(years=5)
    start_str = start_date.strftime("%Y-%m-%d")

    # Determine outputsize based on how far back we need
    days_from_today = (today_date - start_date).days
    outputsize = "compact" if days_from_today < 100 else "full"

    params = {
        "symbol": canonical,
        "outputsize": outputsize,
        "datatype": "csv",
    }

    try:
        csv_text = _make_api_request("TIME_SERIES_DAILY", params)
    except Exception as e:
        raise NoMarketDataError(symbol, canonical, f"Alpha Vantage error: {e}") from e

    if not csv_text or not csv_text.strip():
        raise NoMarketDataError(symbol, canonical, "Alpha Vantage returned empty data")

    try:
        df = pd.read_csv(StringIO(csv_text))
    except Exception as e:
        raise NoMarketDataError(symbol, canonical, f"Failed to parse Alpha Vantage CSV: {e}") from e

    if df.empty:
        raise NoMarketDataError(symbol, canonical, "Alpha Vantage returned no rows")

    # Alpha Vantage CSV has 'timestamp' column; normalize to 'Date'
    date_col = df.columns[0]  # usually 'timestamp'
    df = df.rename(columns={date_col: "Date"})

    # Keep only standard OHLCV columns
    keep_cols = ["Date"]
    col_map = {}
    for av_col, std_col in [
        ("open", "Open"), ("high", "High"), ("low", "Low"),
        ("close", "Close"), ("adjusted_close", "Close"), ("volume", "Volume"),
    ]:
        for c in df.columns:
            if c.lower().replace(" ", "_") == av_col:
                if std_col not in keep_cols:
                    keep_cols.append(std_col)
                    col_map[c] = std_col
                break
    df = df.rename(columns=col_map)
    # Prefer adjusted_close over close for "Close"
    if "adjusted_close" in [c.lower().replace(" ", "_") for c in df.columns]:
        for c in df.columns:
            if c.lower().replace(" ", "_") == "adjusted_close":
                df["Close"] = pd.to_numeric(df[c], errors="coerce")
                break

    available = [c for c in ["Date", "Open", "High", "Low", "Close", "Volume"] if c in df.columns]
    df = df[available]

    # Filter to curr_date
    df["Date"] = pd.to_datetime(df["Date"], errors="coerce")
    df = df.dropna(subset=["Date"])
    df = df[df["Date"] <= curr_date_dt].sort_values("Date")

    if df.empty:
        raise NoMarketDataError(symbol, canonical, "No data on or before " + curr_date)

    return df


def load_ohlcv(symbol: str, curr_date: str) -> pd.DataFrame:
    """Fetch OHLCV data with caching, filtered to prevent look-ahead bias.

    Downloads 5 years of data up to today and caches per symbol. On
    subsequent calls the cache is reused. Rows after curr_date are
    filtered out so backtests never see future prices.
    """
    # Resolve broker/forex symbols (XAUUSD+ -> GC=F) to Yahoo's convention,
    # then reject values that would escape the cache directory when
    # interpolated into the cache filename (e.g. ``../../tmp/x``).
    canonical = normalize_symbol(symbol)
    safe_symbol = safe_ticker_component(canonical)

    config = get_config()
    curr_date_dt = pd.to_datetime(curr_date)

    # Check which vendor to use for OHLCV
    vendor_config = config.get("data_vendors", {}).get("core_stock_apis", "yfinance")
    use_alpha_vantage = "alpha_vantage" in vendor_config

    # Cache uses a fixed window (5y to today) so one file per symbol.
    today_date = pd.Timestamp.today()
    start_date = today_date - pd.DateOffset(years=5)
    start_str = start_date.strftime("%Y-%m-%d")
    end_str = (today_date + pd.Timedelta(days=1)).strftime("%Y-%m-%d")

    cache_prefix = "AV" if use_alpha_vantage else "YFin"
    os.makedirs(config["data_cache_dir"], exist_ok=True)
    data_file = os.path.join(
        config["data_cache_dir"],
        f"{safe_symbol}-{cache_prefix}-data-{start_str}-{end_str}.csv",
    )

    # A cached file may be empty if a prior fetch failed. Treat empty cache as miss.
    data = None
    if os.path.exists(data_file):
        cached = pd.read_csv(data_file, on_bad_lines="skip", encoding="utf-8")
        if not cached.empty and "Close" in cached.columns:
            data = cached

    if data is None:
        if use_alpha_vantage:
            downloaded = _load_ohlcv_alpha_vantage(symbol, curr_date)
        else:
            # Skip local Yahoo Finance, go directly through SSH proxy
            logger.info(f"Yahoo Finance: fetching {canonical} via SSH proxy...")
            try:
                downloaded = _yf_download_via_proxy(canonical, start_str, end_str)
            except Exception as proxy_err:
                raise NoMarketDataError(
                    symbol, canonical, f"Yahoo Finance proxy failed: {proxy_err}"
                ) from proxy_err
            if downloaded.empty:
                raise NoMarketDataError(
                    symbol, canonical, f"Yahoo Finance proxy returned no data"
                )
            if downloaded.empty or "Close" not in downloaded.columns:
                raise NoMarketDataError(
                    symbol, canonical, "Yahoo Finance returned no rows"
                )
            downloaded = _clean_dataframe(downloaded)

        downloaded.to_csv(data_file, index=False, encoding="utf-8")
        data = downloaded

    if not use_alpha_vantage:
        data = _clean_dataframe(data)

    # Filter to curr_date to prevent look-ahead bias in backtesting
    data = data[data["Date"] <= curr_date_dt]

    # Reject a stale frame
    _assert_ohlcv_not_stale(data, curr_date, symbol, canonical)

    return data


def filter_financials_by_date(data: pd.DataFrame, curr_date: str) -> pd.DataFrame:
    """Drop financial statement columns (fiscal period timestamps) after curr_date.

    yfinance financial statements use fiscal period end dates as columns.
    Columns after curr_date represent future data and are removed to
    prevent look-ahead bias.
    """
    if not curr_date or data.empty:
        return data
    cutoff = pd.Timestamp(curr_date)
    mask = pd.to_datetime(data.columns, errors="coerce") <= cutoff
    return data.loc[:, mask]


class StockstatsUtils:
    @staticmethod
    def get_stock_stats(
        symbol: Annotated[str, "ticker symbol for the company"],
        indicator: Annotated[
            str, "quantitative indicators based off of the stock data for the company"
        ],
        curr_date: Annotated[
            str, "curr date for retrieving stock price data, YYYY-mm-dd"
        ],
    ):
        data = load_ohlcv(symbol, curr_date)
        df = wrap(data)
        df["Date"] = df["Date"].dt.strftime("%Y-%m-%d")
        curr_date_str = pd.to_datetime(curr_date).strftime("%Y-%m-%d")

        df[indicator]  # trigger stockstats to calculate the indicator
        matching_rows = df[df["Date"].str.startswith(curr_date_str)]

        if not matching_rows.empty:
            indicator_value = matching_rows[indicator].values[0]
            return indicator_value
        else:
            return "N/A: Not a trading day (weekend or holiday)"