Spaces:

ResearchEngineering
/

financial_news_bot

Running

financial_news_bot / src /transformations /wavelets /analyzer.py

Dmitry Beresnev

fix trading/portfolio/wavelets/predictor bugs; refactor markov chains to modular pipeline

5784bb6 4 months ago

9.43 kB

"""WaveletAnalyzer — main entry point for wavelet analysis.
Orchestrates:
  1. Data fetching (yfinance, cached)
  2. Full MODWT detail decomposition for visualization
  3. Walk-forward backtest
  4. Signal snapshot (current mid-band signal for the ticker)
  5. Returns an Analysis dataclass ready for formatting
Usage:
    from src.transformations.wavelets.analyzer import WaveletAnalyzer
    result = await WaveletAnalyzer.analyze("SPY", "1d", equity=10_000)
"""
from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Optional

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

from .backtest import run_backtest, LOOKBACK, WAVELET, LEVEL, SIG_LEVELS, SLOPE_WINDOW
from .modwt import modwt_details_causal, reconstruct_midband, trim_boundary
from .signal import (
    _linear_slope,
    compute_midband_series,
    compute_all_details_series,
    volatility_target,
)
from .stats import Stats, perf

logger = logging.getLogger(__name__)

_MIN_BARS_FOR_BACKTEST = LOOKBACK + 20
_MIN_BARS_FOR_SIGNAL = 256


@dataclass
class SignalSnapshot:
    """Current state of the MODWT signal for a single ticker."""
    ticker: str
    timeframe: str
    current_price: float
    last_bar_time: pd.Timestamp
    raw_signal: float            # +1.0, -1.0, or 0.0
    sized_position: float        # after vol targeting
    realized_vol_ann: float      # 60-day annualized realized vol
    midband_slope: float         # the underlying slope value (not just sign)
    midband_last: float          # mid-band value at the last safe point
    bars_used: int


@dataclass
class WaveletAnalysis:
    """Full output of WaveletAnalyzer.analyze()."""
    ticker: str
    timeframe: str
    timestamp: datetime

    signal: SignalSnapshot
    sig_levels: list[int] = field(default_factory=lambda: list(SIG_LEVELS))

    # Backtest stats (None if price history too short)
    strategy_stats: Optional[Stats] = None
    bh_stats: Optional[Stats] = None
    sma_stats: Optional[Stats] = None

    # Mid-band reconstruction series (for visualization, full history)
    midband_series: Optional[pd.Series] = None

    # Per-level detail series (for visualization)
    detail_series: Optional[dict[int, pd.Series]] = None

    # Backtest equity curves
    equity_curve: Optional[pd.Series] = None
    bh_equity: Optional[pd.Series] = None
    sma_equity: Optional[pd.Series] = None

    # Warnings
    warnings: list[str] = field(default_factory=list)


def _load_prices(ticker: str, timeframe: str, bars: int) -> pd.Series:
    """Fetch OHLCV via yfinance and return Close series."""
    period_map = {
        "1d": f"{max(bars * 2, 365)}d",
        "1wk": f"{max(bars * 8, 365 * 2)}d",
    }
    period = period_map.get(timeframe, f"{bars * 2}d")

    df = yf.download(
        ticker,
        period=period,
        interval=timeframe,
        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)

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


class WaveletAnalyzer:
    """Stateless wavelet analysis for a ticker + timeframe."""

    @staticmethod
    async def analyze(
        ticker: str,
        timeframe: str = "1d",
        equity: float = 10_000.0,
        run_full_backtest: bool = True,
        wavelet: str = WAVELET,
        level: int = LEVEL,
        sig_levels: list[int] | None = None,
    ) -> WaveletAnalysis:
        """Full analysis: signal snapshot + optional walk-forward backtest.
        Runs in executor to avoid blocking the async event loop.
        """
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(
            None,
            lambda: WaveletAnalyzer._analyze_sync(
                ticker, timeframe, equity, run_full_backtest, wavelet, level, sig_levels
            ),
        )

    @staticmethod
    def _analyze_sync(
        ticker: str,
        timeframe: str,
        equity: float,
        run_full_backtest: bool,
        wavelet: str,
        level: int,
        sig_levels: list[int] | None,
    ) -> WaveletAnalysis:
        if sig_levels is None:
            sig_levels = SIG_LEVELS

        warnings: list[str] = []
        max_sig_level = max(sig_levels)

        # ── 1. Fetch data ─────────────────────────────────────────────────────
        bars_to_fetch = _MIN_BARS_FOR_BACKTEST + 50 if run_full_backtest else _MIN_BARS_FOR_SIGNAL + 50
        try:
            prices = _load_prices(ticker, timeframe, bars_to_fetch)
        except Exception as e:
            raise ValueError(f"Data fetch failed for {ticker}: {e}") from e

        if len(prices) < _MIN_BARS_FOR_SIGNAL:
            raise ValueError(
                f"Only {len(prices)} bars available for {ticker} — need {_MIN_BARS_FOR_SIGNAL}+"
            )

        log_prices = np.log(prices.values)
        daily_ret = np.diff(log_prices, prepend=log_prices[0])

        # ── 2. Current signal snapshot ────────────────────────────────────────
        # ONE decomposition drives signal, slope, and mid-band value — the
        # sign of the reported slope IS the traded signal, so they can never
        # disagree (previously compute_signal ran an identical second MODWT).
        signal_window_size = min(LOOKBACK, len(prices))
        signal_window = log_prices[-signal_window_size:]

        try:
            details = modwt_details_causal(signal_window, wavelet=wavelet, level=level)
            midband = reconstruct_midband(details, sig_levels)
            safe_midband = trim_boundary(midband, max_level=max_sig_level)
        except Exception:
            safe_midband = np.zeros(1)

        if len(safe_midband) >= SLOPE_WINDOW:
            slope = _linear_slope(safe_midband[-SLOPE_WINDOW:])
        else:
            slope = 0.0
        raw_sig = float(np.sign(slope))
        midband_last = float(safe_midband[-1]) if len(safe_midband) > 0 else 0.0

        # Realized vol from past 60 bars
        vol_window = min(60, len(daily_ret) - 1)
        realized_vol = float(np.std(daily_ret[-vol_window:]) * np.sqrt(252))

        sized_pos = volatility_target(raw_sig, realized_vol)

        snapshot = SignalSnapshot(
            ticker=ticker,
            timeframe=timeframe,
            current_price=float(prices.iloc[-1]),
            last_bar_time=prices.index[-1],
            raw_signal=raw_sig,
            sized_position=sized_pos,
            realized_vol_ann=realized_vol,
            midband_slope=slope,
            midband_last=midband_last,
            bars_used=signal_window_size,
        )

        # ── 3. Full decomposition for visualization ───────────────────────────
        midband_vis = None
        detail_vis = None
        try:
            midband_vis = compute_midband_series(
                pd.Series(log_prices, index=prices.index),
                wavelet=wavelet, level=level, sig_levels=sig_levels,
            )
            detail_vis = compute_all_details_series(
                pd.Series(log_prices, index=prices.index),
                wavelet=wavelet, level=level,
            )
        except Exception as e:
            warnings.append(f"Full decomposition failed: {e}")

        # ── 4. Walk-forward backtest ──────────────────────────────────────────
        strat_stats = bh_stats = sma_stats = None
        equity_curve = bh_equity = sma_equity = None

        if run_full_backtest:
            if len(prices) < _MIN_BARS_FOR_BACKTEST:
                warnings.append(
                    f"Only {len(prices)} bars — backtest requires {_MIN_BARS_FOR_BACKTEST}+. "
                    "Showing signal snapshot only."
                )
            else:
                try:
                    bt = run_backtest(
                        prices,
                        wavelet=wavelet,
                        level=level,
                        sig_levels=sig_levels,
                    )
                    strat_stats = bt["strategy_stats"]
                    bh_stats = bt["bh_stats"]
                    sma_stats = bt["sma_stats"]
                    equity_curve = bt["equity_curve"]
                    bh_equity = bt["bh_equity"]
                    sma_equity = bt["sma_equity"]
                except Exception as e:
                    warnings.append(f"Backtest failed: {e}")

        return WaveletAnalysis(
            ticker=ticker,
            timeframe=timeframe,
            timestamp=datetime.now(timezone.utc),
            signal=snapshot,
            sig_levels=sig_levels,
            strategy_stats=strat_stats,
            bh_stats=bh_stats,
            sma_stats=sma_stats,
            midband_series=midband_vis,
            detail_series=detail_vis,
            equity_curve=equity_curve,
            bh_equity=bh_equity,
            sma_equity=sma_equity,
            warnings=warnings,
        )

	"""WaveletAnalyzer — main entry point for wavelet analysis.

	Orchestrates:
	1. Data fetching (yfinance, cached)
	2. Full MODWT detail decomposition for visualization
	3. Walk-forward backtest
	4. Signal snapshot (current mid-band signal for the ticker)
	5. Returns an Analysis dataclass ready for formatting

	Usage:
	from src.transformations.wavelets.analyzer import WaveletAnalyzer
	result = await WaveletAnalyzer.analyze("SPY", "1d", equity=10_000)
	"""
	from __future__ import annotations
	import asyncio
	import logging
	from dataclasses import dataclass, field
	from datetime import datetime, timezone
	from typing import Optional

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

	from .backtest import run_backtest, LOOKBACK, WAVELET, LEVEL, SIG_LEVELS, SLOPE_WINDOW
	from .modwt import modwt_details_causal, reconstruct_midband, trim_boundary
	from .signal import (
	_linear_slope,
	compute_midband_series,
	compute_all_details_series,
	volatility_target,
	)
	from .stats import Stats, perf

	logger = logging.getLogger(__name__)

	_MIN_BARS_FOR_BACKTEST = LOOKBACK + 20
	_MIN_BARS_FOR_SIGNAL = 256


	@dataclass
	class SignalSnapshot:
	"""Current state of the MODWT signal for a single ticker."""
	ticker: str
	timeframe: str
	current_price: float
	last_bar_time: pd.Timestamp
	raw_signal: float # +1.0, -1.0, or 0.0
	sized_position: float # after vol targeting
	realized_vol_ann: float # 60-day annualized realized vol
	midband_slope: float # the underlying slope value (not just sign)
	midband_last: float # mid-band value at the last safe point
	bars_used: int


	@dataclass
	class WaveletAnalysis:
	"""Full output of WaveletAnalyzer.analyze()."""
	ticker: str
	timeframe: str
	timestamp: datetime

	signal: SignalSnapshot
	sig_levels: list[int] = field(default_factory=lambda: list(SIG_LEVELS))

	# Backtest stats (None if price history too short)
	strategy_stats: Optional[Stats] = None
	bh_stats: Optional[Stats] = None
	sma_stats: Optional[Stats] = None

	# Mid-band reconstruction series (for visualization, full history)
	midband_series: Optional[pd.Series] = None

	# Per-level detail series (for visualization)
	detail_series: Optional[dict[int, pd.Series]] = None

	# Backtest equity curves
	equity_curve: Optional[pd.Series] = None
	bh_equity: Optional[pd.Series] = None
	sma_equity: Optional[pd.Series] = None

	# Warnings
	warnings: list[str] = field(default_factory=list)


	def _load_prices(ticker: str, timeframe: str, bars: int) -> pd.Series:
	"""Fetch OHLCV via yfinance and return Close series."""
	period_map = {
	"1d": f"{max(bars * 2, 365)}d",
	"1wk": f"{max(bars * 8, 365 * 2)}d",
	}
	period = period_map.get(timeframe, f"{bars * 2}d")

	df = yf.download(
	ticker,
	period=period,
	interval=timeframe,
	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)

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


	class WaveletAnalyzer:
	"""Stateless wavelet analysis for a ticker + timeframe."""

	@staticmethod
	async def analyze(
	ticker: str,
	timeframe: str = "1d",
	equity: float = 10_000.0,
	run_full_backtest: bool = True,
	wavelet: str = WAVELET,
	level: int = LEVEL,
	sig_levels: list[int] \| None = None,
	) -> WaveletAnalysis:
	"""Full analysis: signal snapshot + optional walk-forward backtest.

	Runs in executor to avoid blocking the async event loop.
	"""
	loop = asyncio.get_running_loop()
	return await loop.run_in_executor(
	None,
	lambda: WaveletAnalyzer._analyze_sync(
	ticker, timeframe, equity, run_full_backtest, wavelet, level, sig_levels
	),
	)

	@staticmethod
	def _analyze_sync(
	ticker: str,
	timeframe: str,
	equity: float,
	run_full_backtest: bool,
	wavelet: str,
	level: int,
	sig_levels: list[int] \| None,
	) -> WaveletAnalysis:
	if sig_levels is None:
	sig_levels = SIG_LEVELS

	warnings: list[str] = []
	max_sig_level = max(sig_levels)

	# ── 1. Fetch data ─────────────────────────────────────────────────────
	bars_to_fetch = _MIN_BARS_FOR_BACKTEST + 50 if run_full_backtest else _MIN_BARS_FOR_SIGNAL + 50
	try:
	prices = _load_prices(ticker, timeframe, bars_to_fetch)
	except Exception as e:
	raise ValueError(f"Data fetch failed for {ticker}: {e}") from e

	if len(prices) < _MIN_BARS_FOR_SIGNAL:
	raise ValueError(
	f"Only {len(prices)} bars available for {ticker} — need {_MIN_BARS_FOR_SIGNAL}+"
	)

	log_prices = np.log(prices.values)
	daily_ret = np.diff(log_prices, prepend=log_prices[0])

	# ── 2. Current signal snapshot ────────────────────────────────────────
	# ONE decomposition drives signal, slope, and mid-band value — the
	# sign of the reported slope IS the traded signal, so they can never
	# disagree (previously compute_signal ran an identical second MODWT).
	signal_window_size = min(LOOKBACK, len(prices))
	signal_window = log_prices[-signal_window_size:]

	try:
	details = modwt_details_causal(signal_window, wavelet=wavelet, level=level)
	midband = reconstruct_midband(details, sig_levels)
	safe_midband = trim_boundary(midband, max_level=max_sig_level)
	except Exception:
	safe_midband = np.zeros(1)

	if len(safe_midband) >= SLOPE_WINDOW:
	slope = _linear_slope(safe_midband[-SLOPE_WINDOW:])
	else:
	slope = 0.0
	raw_sig = float(np.sign(slope))
	midband_last = float(safe_midband[-1]) if len(safe_midband) > 0 else 0.0

	# Realized vol from past 60 bars
	vol_window = min(60, len(daily_ret) - 1)
	realized_vol = float(np.std(daily_ret[-vol_window:]) * np.sqrt(252))

	sized_pos = volatility_target(raw_sig, realized_vol)

	snapshot = SignalSnapshot(
	ticker=ticker,
	timeframe=timeframe,
	current_price=float(prices.iloc[-1]),
	last_bar_time=prices.index[-1],
	raw_signal=raw_sig,
	sized_position=sized_pos,
	realized_vol_ann=realized_vol,
	midband_slope=slope,
	midband_last=midband_last,
	bars_used=signal_window_size,
	)

	# ── 3. Full decomposition for visualization ───────────────────────────
	midband_vis = None
	detail_vis = None
	try:
	midband_vis = compute_midband_series(
	pd.Series(log_prices, index=prices.index),
	wavelet=wavelet, level=level, sig_levels=sig_levels,
	)
	detail_vis = compute_all_details_series(
	pd.Series(log_prices, index=prices.index),
	wavelet=wavelet, level=level,
	)
	except Exception as e:
	warnings.append(f"Full decomposition failed: {e}")

	# ── 4. Walk-forward backtest ──────────────────────────────────────────
	strat_stats = bh_stats = sma_stats = None
	equity_curve = bh_equity = sma_equity = None

	if run_full_backtest:
	if len(prices) < _MIN_BARS_FOR_BACKTEST:
	warnings.append(
	f"Only {len(prices)} bars — backtest requires {_MIN_BARS_FOR_BACKTEST}+. "
	"Showing signal snapshot only."
	)
	else:
	try:
	bt = run_backtest(
	prices,
	wavelet=wavelet,
	level=level,
	sig_levels=sig_levels,
	)
	strat_stats = bt["strategy_stats"]
	bh_stats = bt["bh_stats"]
	sma_stats = bt["sma_stats"]
	equity_curve = bt["equity_curve"]
	bh_equity = bt["bh_equity"]
	sma_equity = bt["sma_equity"]
	except Exception as e:
	warnings.append(f"Backtest failed: {e}")

	return WaveletAnalysis(
	ticker=ticker,
	timeframe=timeframe,
	timestamp=datetime.now(timezone.utc),
	signal=snapshot,
	sig_levels=sig_levels,
	strategy_stats=strat_stats,
	bh_stats=bh_stats,
	sma_stats=sma_stats,
	midband_series=midband_vis,
	detail_series=detail_vis,
	equity_curve=equity_curve,
	bh_equity=bh_equity,
	sma_equity=sma_equity,
	warnings=warnings,
	)