models/base_model.py

import numpy as np
import pandas as pd
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Union


def validate_history_and_horizon(history, horizon: int, model_name: str) -> None:
    if horizon < 1:
        raise ValueError(f'{model_name}: horizon must be >= 1, got {horizon}.')
    values = np.asarray(pd.Series(history).astype(float), dtype=float)
    if not np.all(np.isfinite(values)):
        raise ValueError(f'{model_name}: price history contains NaN/inf. This usually means a data glitch left a missing or malformed candle -- re-fetch the history and try again.')


@dataclass
class Signal:
    """Unified prediction contract shared by every model in the project.

    This is the single interface the backtester and the live engine consume, so
    that price-forecasting models (ARIMA, Auto-ARIMA, ARIMA-GARCH, Moirai,
    TimesFM) and the MAYTHOS decision engine are driven through exactly one code
    path. See MERGE_NOTES.md (Decision 1).

    Attributes:
        direction: 'up' or 'down' -- the predicted move of the next close.
        confidence: model confidence in [0, 1].
        predicted_close: the forecast price for the final horizon step, or None
            for models (like MAYTHOS) that emit a directional decision rather
            than a price level.
        horizon: number of steps ahead this signal describes.
        meta: model-specific diagnostic fields (market state, execution
            suitability, etc.). Empty for plain forecasters.
    """
    direction: str
    confidence: float
    predicted_close: Optional[float] = None
    horizon: int = 1
    meta: Dict[str, Any] = field(default_factory=dict)


class BaseModel(ABC):
    """Root of the model hierarchy. Every model produces a :class:`Signal`."""

    name = 'base'

    @abstractmethod
    def predict_signal(self, window: pd.DataFrame, horizon: int = 1, features: pd.DataFrame = None) -> Signal:
        raise NotImplementedError


def _confidence_from_forecast(history: np.ndarray, predicted_close: float) -> float:
    """Map a numeric forecast to a [0, 1] confidence for the unified Signal.

    The size of the predicted move is compared against the recent realised
    volatility (standard deviation of first differences). A move of two standard
    deviations or more saturates confidence at 1.0. This is a documented,
    deterministic heuristic so that forecast-only models still populate the same
    confidence field MAYTHOS reports natively. See MERGE_NOTES.md (Decision 1).
    """
    arr = np.asarray(history, dtype=float)
    current = float(arr[-1])
    move = abs(float(predicted_close) - current)
    diffs = np.diff(arr)
    scale = float(np.std(diffs)) if diffs.size else 0.0
    if scale <= 0.0:
        return 1.0 if move > 0.0 else 0.0
    return float(min(1.0, move / (2.0 * scale)))


class BaseForecastModel(BaseModel):
    """Base class for models that forecast a price path.

    Subclasses implement :meth:`predict` (a list of future closes). The concrete
    :meth:`predict_signal` here turns that path into the unified Signal, so every
    forecaster automatically speaks the same contract as MAYTHOS without any
    adapter or bridge layer.
    """

    name = 'base'

    @abstractmethod
    def predict(self, history: pd.Series, horizon: int = 1, features: pd.DataFrame = None) -> list:
        raise NotImplementedError

    @staticmethod
    def _history_series(window: Union[pd.DataFrame, pd.Series, Any]) -> pd.Series:
        if isinstance(window, pd.DataFrame):
            if 'Close' not in window.columns:
                raise ValueError("forecast window DataFrame must contain a 'Close' column.")
            return window['Close']
        return pd.Series(window)

    def predict_signal(self, window: pd.DataFrame, horizon: int = 1, features: pd.DataFrame = None) -> Signal:
        history = self._history_series(window)
        forecast: List[float] = list(self.predict(history, horizon=horizon, features=features))
        if len(forecast) < 1:
            raise ValueError(f'{self.name}: predict() returned an empty forecast.')
        predicted_close = float(forecast[-1])
        current_close = float(pd.Series(history).astype(float).iloc[-1])
        direction = 'up' if predicted_close > current_close else 'down'
        confidence = _confidence_from_forecast(pd.Series(history).astype(float).to_numpy(), predicted_close)
        return Signal(direction=direction, confidence=confidence, predicted_close=predicted_close, horizon=horizon, meta={})