src/transformations/wavelets_lite/analyzer.py

"""Data fetching and async orchestration for wavelets_lite."""
from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone

import numpy as np
import pandas as pd
import yfinance as yf

from .core import signal_snapshot, min_bars_for_clean_signal

logger = logging.getLogger(__name__)

# Timeframe → (yfinance interval, TARGET-timeframe bars per calendar day).
# Used to convert the requested lookback (in target bars) into a fetch period.
_TF_META: dict[str, tuple[str, float]] = {
    "1h":  ("60m",  6.5),
    # 4h is resampled FROM 1h: a 6.5h US session yields ~1.6 four-hour bars.
    # (The old value 6.5 under-fetched 4× — a 512-bar 4h lookback could
    # never be filled.)
    "4h":  ("60m",  1.6),
    "1d":  ("1d",   1.0),
    "1wk": ("1wk",  1.0 / 5),
}

VALID_TIMEFRAMES = set(_TF_META)
DEFAULT_LOOKBACK = 512
DEFAULT_TF = "1d"
SIG_LEVELS = [4, 5]
SLOPE_WINDOW = 40
VOL_WINDOW = 60
DECOMP_LEVELS = 6


@dataclass
class LiteSignal:
    ticker: str
    timeframe: str
    timestamp: datetime
    current_price: float
    last_bar_time: pd.Timestamp

    raw_signal: float         # +1.0, -1.0, 0.0
    sized_position: float     # vol-targeted
    midband_slope: float
    vol_ann: float

    level_slopes: dict[str, float]   # D1..D6, A6
    level_signals: dict[str, float]  # sign of each level slope

    sig_levels: list[int]
    lookback: int
    bars_loaded: int
    warnings: list[str] = field(default_factory=list)


def _resample_4h(df_1h: pd.DataFrame) -> pd.DataFrame:
    return df_1h.resample("4h").agg({
        "Open": "first", "High": "max",
        "Low": "min", "Close": "last", "Volume": "sum",
    }).dropna()


def _fetch_prices(ticker: str, timeframe: str, lookback: int) -> pd.Series:
    yf_interval, bars_per_day = _TF_META[timeframe]

    # Request enough bars to cover lookback with margin
    need_days = int(lookback / bars_per_day * 1.4) + 30
    period = f"{min(need_days, 730)}d" if yf_interval != "1wk" else f"{min(need_days * 7, 3650)}d"

    df = yf.download(
        ticker,
        period=period,
        interval=yf_interval,
        auto_adjust=True,
        progress=False,
        threads=False,
    )
    if df.empty:
        raise ValueError(f"No data returned for {ticker} ({timeframe})")

    if isinstance(df.columns, pd.MultiIndex):
        df.columns = df.columns.get_level_values(0)

    df = df[["Open", "High", "Low", "Close", "Volume"]].dropna()

    if timeframe == "4h":
        df = _resample_4h(df)

    close = df["Close"].dropna()
    close.name = ticker
    return close.tail(lookback)


class WaveletLiteAnalyzer:
    """Stateless lightweight wavelet signal analyzer."""

    @staticmethod
    async def analyze(
        ticker: str,
        timeframe: str = DEFAULT_TF,
        lookback: int = DEFAULT_LOOKBACK,
        sig_levels: list[int] | None = None,
    ) -> LiteSignal:
        """Fetch prices and compute MODWT signal snapshot. Non-blocking."""
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(
            None,
            lambda: WaveletLiteAnalyzer._analyze_sync(
                ticker, timeframe, lookback, sig_levels or SIG_LEVELS
            ),
        )

    @staticmethod
    def _analyze_sync(
        ticker: str,
        timeframe: str,
        lookback: int,
        sig_levels: list[int],
    ) -> LiteSignal:
        warnings: list[str] = []

        if timeframe not in VALID_TIMEFRAMES:
            raise ValueError(
                f"Unsupported timeframe '{timeframe}'. "
                f"Use one of: {', '.join(sorted(VALID_TIMEFRAMES))}"
            )

        # ── fetch ─────────────────────────────────────────────────────────────
        try:
            prices = _fetch_prices(ticker, timeframe, lookback)
        except Exception as e:
            raise ValueError(f"Data fetch failed for {ticker}: {e}") from e

        bars = len(prices)
        if bars < 64:
            raise ValueError(
                f"Only {bars} bars for {ticker} ({timeframe}) — need at least 64."
            )
        if bars < lookback:
            warnings.append(
                f"Only {bars} bars available (requested {lookback}). "
                "Signal quality at deep levels (D5/D6) may be reduced."
            )
        min_clean = min_bars_for_clean_signal(max(sig_levels), SLOPE_WINDOW)
        if bars < min_clean:
            warnings.append(
                f"Only {bars} bars — need {min_clean} for a clean D{max(sig_levels)} signal. "
                "Mid-band slope is in the transient region and may be unreliable."
            )

        # ── data freshness ────────────────────────────────────────────────────
        last_bar_time: pd.Timestamp = prices.index[-1]
        now = datetime.now(timezone.utc)
        lb = last_bar_time
        if lb.tzinfo is None:
            lb = lb.tz_localize("UTC")
        age_min = (now - lb.to_pydatetime()).total_seconds() / 60
        # Thresholds scaled to the timeframe; 1d/1wk must also cover weekends
        # and holidays (a Friday daily bar is 72h old on Sunday — not stale).
        # Values match src/core/levels/analyze.py for consistency.
        _freshness_threshold_min = {"1h": 150, "4h": 600, "1d": 5760, "1wk": 20160}
        stale_threshold = _freshness_threshold_min.get(timeframe, 30)
        if age_min > stale_threshold:
            warnings.append(
                f"Last bar is {age_min:.0f} min old "
                "(yfinance free tier ~15 min lag for US equities)"
            )

        # ── signal ────────────────────────────────────────────────────────────
        log_prices = np.log(prices.values)
        snap = signal_snapshot(
            log_prices,
            sig_levels=sig_levels,
            slope_window=SLOPE_WINDOW,
            vol_window=VOL_WINDOW,
            decomp_levels=DECOMP_LEVELS,
        )

        return LiteSignal(
            ticker=ticker,
            timeframe=timeframe,
            timestamp=datetime.now(timezone.utc),
            current_price=float(prices.iloc[-1]),
            last_bar_time=last_bar_time,
            raw_signal=snap["raw_signal"],
            sized_position=snap["sized_position"],
            midband_slope=snap["midband_slope"],
            vol_ann=snap["vol_ann"],
            level_slopes=snap["level_slopes"],
            level_signals=snap["level_signals"],
            sig_levels=sig_levels,
            lookback=lookback,
            bars_loaded=bars,
            warnings=warnings,
        )