src/strategies.py

"""Strategy presets.

Every strategy is a pure function of price history (and, optionally, stored
model signals) that returns decisions aligned to **bar close**. None of them
shift their own output -- `engine.run_backtest` does that, exactly once, so
next-bar-open execution cannot be bypassed by a strategy.

Every indicator here is causal: it uses `rolling`/`ewm` over past bars only.
`tests/test_engine.py` proves this by perturbation rather than trusting it.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Callable, Protocol

import numpy as np
import pandas as pd

from .engine import StrategyOutput

# --------------------------------------------------------------------------
# Sentiment interface (stubbed for v1, real source lands later)
# --------------------------------------------------------------------------


class SentimentSource(Protocol):
    """Anything that can score sentiment per bar, causally."""

    def score(self, index: pd.DatetimeIndex, asset: str) -> pd.Series:
        """Value in [-1, 1] per bar, using only information available at that bar."""
        ...


class NeutralSentiment:
    """Default source: no opinion. Keeps the gate open so the momentum leg
    behaves as plain momentum until a real feed is wired in."""

    name = "neutral-stub"
    is_stub = True

    def score(self, index: pd.DatetimeIndex, asset: str) -> pd.Series:
        return pd.Series(1.0, index=index, dtype="float64")


class PriceProxySentiment:
    """Deterministic stand-in derived from realised momentum.

    Clearly labelled as a proxy -- it is *not* news sentiment. It exists so the
    Sentiment-Gated preset is demonstrable end to end before the real feed
    exists, and it is causal by construction.
    """

    name = "price-proxy-stub"
    is_stub = True

    def __init__(self, lookback: int = 24):
        self.lookback = lookback

    def score(self, index: pd.DatetimeIndex, asset: str) -> pd.Series:
        return pd.Series(np.nan, index=index, dtype="float64")

    def score_from_prices(self, prices: pd.DataFrame) -> pd.Series:
        ret = prices["close"].pct_change(self.lookback)
        scaled = np.tanh(ret / (ret.rolling(self.lookback * 4).std().replace(0, np.nan) + 1e-12))
        return scaled.fillna(0.0).clip(-1.0, 1.0)


# --------------------------------------------------------------------------
# Indicator helpers (all causal)
# --------------------------------------------------------------------------


def sma(s: pd.Series, n: int) -> pd.Series:
    return s.rolling(int(n), min_periods=int(n)).mean()


def ema(s: pd.Series, n: int) -> pd.Series:
    return s.ewm(span=int(n), adjust=False, min_periods=int(n)).mean()


def rsi(s: pd.Series, n: int = 14) -> pd.Series:
    delta = s.diff()
    gain = delta.clip(lower=0.0)
    loss = -delta.clip(upper=0.0)
    avg_gain = gain.ewm(alpha=1 / int(n), adjust=False, min_periods=int(n)).mean()
    avg_loss = loss.ewm(alpha=1 / int(n), adjust=False, min_periods=int(n)).mean()
    rs = avg_gain / avg_loss.replace(0.0, np.nan)
    return (100.0 - 100.0 / (1.0 + rs)).fillna(50.0)


def bollinger(s: pd.Series, n: int = 20, k: float = 2.0):
    mid = s.rolling(int(n), min_periods=int(n)).mean()
    sd = s.rolling(int(n), min_periods=int(n)).std(ddof=0)
    return mid - k * sd, mid, mid + k * sd


def macd(s: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9):
    line = ema(s, fast) - ema(s, slow)
    sig = line.ewm(span=int(signal), adjust=False, min_periods=int(signal)).mean()
    return line, sig, line - sig


def _cross_up(a: pd.Series, b: pd.Series) -> pd.Series:
    return ((a > b) & (a.shift(1) <= b.shift(1))).astype("boolean").fillna(False).astype(bool)


def _cross_down(a: pd.Series, b: pd.Series) -> pd.Series:
    return ((a < b) & (a.shift(1) >= b.shift(1))).astype("boolean").fillna(False).astype(bool)


def _triggers(index, entries, exits, entry_text: str, exit_text: str) -> pd.Series:
    t = pd.Series("", index=index, dtype="object")
    t[entries] = entry_text
    t[exits] = exit_text
    return t


# --------------------------------------------------------------------------
# Presets
# --------------------------------------------------------------------------


def buy_and_hold(prices: pd.DataFrame, params: dict | None = None,
                 signals: pd.DataFrame | None = None) -> StrategyOutput:
    """Enter on the first bar, never exit. The benchmark every claim is measured against."""
    idx = prices.index
    entries = pd.Series(False, index=idx)
    exits = pd.Series(False, index=idx)
    if len(idx):
        entries.iloc[0] = True
    return StrategyOutput(entries=entries, exits=exits,
                          triggers=_triggers(idx, entries, exits, "buy and hold entry", ""))


def sma_crossover(prices: pd.DataFrame, params: dict | None = None,
                  signals: pd.DataFrame | None = None) -> StrategyOutput:
    p = params or {}
    fast_n, slow_n = int(p.get("fast_ma", 20)), int(p.get("slow_ma", 50))
    close = prices["close"]
    fast, slow = sma(close, fast_n), sma(close, slow_n)
    entries = _cross_up(fast, slow)
    exits = _cross_down(fast, slow)
    return StrategyOutput(
        entries=entries, exits=exits,
        triggers=_triggers(prices.index, entries, exits,
                           f"SMA{fast_n} crossed above SMA{slow_n}",
                           f"SMA{fast_n} crossed below SMA{slow_n}"),
    )


def rsi_mean_reversion(prices: pd.DataFrame, params: dict | None = None,
                       signals: pd.DataFrame | None = None) -> StrategyOutput:
    p = params or {}
    n = int(p.get("rsi_period", 14))
    lo, hi = float(p.get("oversold", 30)), float(p.get("overbought", 70))
    r = rsi(prices["close"], n)
    entries = ((r < lo) & (r.shift(1) >= lo)).astype("boolean").fillna(False).astype(bool)
    exits = ((r > hi) & (r.shift(1) <= hi)).astype("boolean").fillna(False).astype(bool)
    return StrategyOutput(
        entries=entries, exits=exits,
        triggers=_triggers(prices.index, entries, exits,
                           f"RSI({n}) fell below {lo:g}", f"RSI({n}) rose above {hi:g}"),
    )


def bollinger_breakout(prices: pd.DataFrame, params: dict | None = None,
                       signals: pd.DataFrame | None = None) -> StrategyOutput:
    p = params or {}
    n, k = int(p.get("bb_period", 20)), float(p.get("bb_std", 2.0))
    close = prices["close"]
    lower, mid, upper = bollinger(close, n, k)
    entries = ((close > upper) & (close.shift(1) <= upper.shift(1))).astype("boolean").fillna(False).astype(bool)
    exits = ((close < mid) & (close.shift(1) >= mid.shift(1))).astype("boolean").fillna(False).astype(bool)
    return StrategyOutput(
        entries=entries, exits=exits,
        triggers=_triggers(prices.index, entries, exits,
                           f"close broke above the {n}/{k:g}σ upper band",
                           "close fell back through the band midline"),
    )


def macd_momentum(prices: pd.DataFrame, params: dict | None = None,
                  signals: pd.DataFrame | None = None) -> StrategyOutput:
    p = params or {}
    f, s, g = int(p.get("macd_fast", 12)), int(p.get("macd_slow", 26)), int(p.get("macd_signal", 9))
    line, sig, _ = macd(prices["close"], f, s, g)
    entries = _cross_up(line, sig)
    exits = _cross_down(line, sig)
    return StrategyOutput(
        entries=entries, exits=exits,
        triggers=_triggers(prices.index, entries, exits,
                           f"MACD({f},{s}) crossed above its {g}-period signal",
                           f"MACD({f},{s}) crossed below its {g}-period signal"),
    )


def forecast_follower(prices: pd.DataFrame, params: dict | None = None,
                      signals: pd.DataFrame | None = None) -> StrategyOutput:
    """Rule over stored quantiles: go long when the median forecast implies
    enough upside; optionally exit when price breaches the q10 floor.

    The stored forecast at bar `t` was produced from data up to `t`, and the
    engine shifts it before acting, so the earliest possible fill is `t+1`'s open.
    """
    p = params or {}
    threshold = float(p.get("threshold", 0.005))
    use_q10_stop = bool(p.get("use_q10_stop", True))
    exit_threshold = float(p.get("exit_threshold", 0.0))

    idx = prices.index
    close = prices["close"]
    if signals is None or signals.empty or "q50" not in signals.columns:
        false = pd.Series(False, index=idx)
        return StrategyOutput(entries=false, exits=false.copy(),
                              triggers=pd.Series("", index=idx, dtype="object"))

    q50 = signals["q50"].reindex(idx).ffill()
    q10 = signals["q10"].reindex(idx).ffill() if "q10" in signals.columns else None

    edge = (q50 / close) - 1.0
    entries = ((edge > threshold) & (edge.shift(1) <= threshold)).astype("boolean").fillna(False).astype(bool)
    exits = ((edge < exit_threshold) & (edge.shift(1) >= exit_threshold)).astype("boolean").fillna(False).astype(bool)
    if use_q10_stop and q10 is not None:
        breach = (close < q10).astype("boolean").fillna(False).astype(bool)
        prev_breach = breach.astype("boolean").shift(1).fillna(False).astype(bool)
        exits = (exits | (breach & ~prev_breach)).astype("boolean").fillna(False).astype(bool)

    trig = pd.Series("", index=idx, dtype="object")
    trig[entries] = f"forecast median implied >{threshold:.2%} upside"
    trig[exits] = "forecast edge closed or price breached the q10 floor"
    return StrategyOutput(entries=entries, exits=exits, triggers=trig)


def sentiment_gated_momentum(prices: pd.DataFrame, params: dict | None = None,
                             signals: pd.DataFrame | None = None,
                             sentiment: SentimentSource | None = None) -> StrategyOutput:
    """Momentum that only fires while the sentiment gate is open.

    The sentiment input sits behind `SentimentSource`. Until a real feed is
    wired in, the default is a labelled stub -- see `NeutralSentiment`.
    """
    p = params or {}
    fast_n, slow_n = int(p.get("fast_ma", 20)), int(p.get("slow_ma", 50))
    gate = float(p.get("sentiment_gate", 0.40))
    trail = p.get("trail_pct")

    close = prices["close"]
    fast, slow = sma(close, fast_n), sma(close, slow_n)

    src = sentiment or PriceProxySentiment()
    if hasattr(src, "score_from_prices"):
        score = src.score_from_prices(prices)
    else:
        score = src.score(prices.index, "")
    score = score.reindex(prices.index).fillna(0.0)

    gate_open = score >= gate
    entries = (_cross_up(fast, slow) & gate_open).astype("boolean").fillna(False).astype(bool)
    was_open = gate_open.astype("boolean").shift(1).fillna(False).astype(bool)
    exits = (_cross_down(fast, slow) | (~gate_open & was_open)) \
        .astype("boolean").fillna(False).astype(bool)

    trig = pd.Series("", index=prices.index, dtype="object")
    trig[entries] = f"MA cross up with sentiment ≥ {gate:.2f}"
    trig[exits] = "MA cross down or sentiment gate closed"
    if trail:
        trig[entries] = trig[entries] + f" (trailing stop {float(trail):.1%})"
    return StrategyOutput(entries=entries, exits=exits, triggers=trig)


# --------------------------------------------------------------------------
# Registry
# --------------------------------------------------------------------------


@dataclass(frozen=True)
class Preset:
    name: str
    fn: Callable
    needs_signals: bool = False
    available: bool = True
    unavailable_reason: str = ""
    params: tuple[tuple[str, str, float, float, float], ...] = ()
    # (key, label, default, min, max)


PRESETS: dict[str, Preset] = {
    p.name: p
    for p in [
        Preset("Buy & Hold (benchmark)", buy_and_hold),
        Preset("SMA Crossover", sma_crossover, params=(
            ("fast_ma", "Fast MA", 20, 2, 200),
            ("slow_ma", "Slow MA", 50, 3, 400),
        )),
        Preset("RSI Mean Reversion", rsi_mean_reversion, params=(
            ("rsi_period", "RSI period", 14, 2, 100),
            ("oversold", "Oversold", 30, 1, 49),
            ("overbought", "Overbought", 70, 51, 99),
        )),
        Preset("Bollinger Breakout", bollinger_breakout, params=(
            ("bb_period", "Period", 20, 5, 200),
            ("bb_std", "Std devs", 2.0, 0.5, 5.0),
        )),
        Preset("MACD Momentum", macd_momentum, params=(
            ("macd_fast", "Fast EMA", 12, 2, 100),
            ("macd_slow", "Slow EMA", 26, 3, 200),
            ("macd_signal", "Signal", 9, 2, 50),
        )),
        Preset("Chronos Forecast Follower", forecast_follower, needs_signals=True, params=(
            ("threshold", "Entry edge", 0.005, 0.0, 0.2),
            ("exit_threshold", "Exit edge", 0.0, -0.1, 0.1),
        )),
        Preset("Sentiment-Gated Momentum", sentiment_gated_momentum, params=(
            ("fast_ma", "Fast MA", 20, 2, 200),
            ("slow_ma", "Slow MA", 50, 3, 400),
            ("sentiment_gate", "Sentiment gate", 0.40, -1.0, 1.0),
        )),
        # Present in the design; not runnable in v1.
        Preset("Pairs Trading", buy_and_hold, available=False,
               unavailable_reason="Needs a second leg; single-asset runs only in v1."),
        Preset("Custom (code)", buy_and_hold, available=False,
               unavailable_reason="Running user-supplied strategy code is disabled by "
                                  "design — this Space never executes untrusted code."),
    ]
}

PRESET_NAMES = list(PRESETS)


def build(name: str, prices: pd.DataFrame, params: dict | None = None,
          signals: pd.DataFrame | None = None) -> StrategyOutput:
    """Run a preset by name. Unknown or unavailable presets raise."""
    preset = PRESETS.get(name)
    if preset is None:
        raise KeyError(f"unknown strategy preset {name!r}")
    if not preset.available:
        raise ValueError(f"{name} is not available: {preset.unavailable_reason}")
    if preset.needs_signals and (signals is None or signals.empty):
        raise ValueError(f"{name} needs stored model signals for this asset and timeframe")
    return preset.fn(prices, params or {}, signals)


def defaults_for(name: str) -> dict:
    preset = PRESETS.get(name)
    if preset is None:
        return {}
    return {k: d for k, _, d, _, _ in preset.params}