Technical indicators as functions of the past, and a minimal event-driven backtester with costs

"""M04 exercise: technical indicators as functions of the past, and a minimal
event-driven backtester with costs (notes §4.1, §4.2, §4.4).

Implement every function, then run from the repo root:

    cd exercises && .venv/bin/python -m pytest modules/M04-systematic-research-swing/tests/test_m04_backtest.py

Conventions
-----------
* Every indicator value at date t uses data dated <= t only. Rolling windows
  are trailing, never centred.
* Bar timing: the strategy sees the bar of date t after its close and returns a
  target weight w_t (fraction of equity: +1 long, 0 flat, -1 short). The engine
  fills that target at the OPEN of bar t+1. A weight decided at the close of t
  never earns the return of bar t itself.
* Costs: `cost_bps` per side on traded value (all statutory and broker charges
  folded into one number; see M01) plus `slippage_bps`, applied by moving the
  fill price against the trader.
The tests perturb FUTURE prices and check that past targets and equity do not move.
"""

from __future__ import annotations

from typing import Callable

import numpy as np
import pandas as pd

# --------------------------------------------------------------- indicators


def sma(x: pd.Series, n: int) -> pd.Series:
    """Trailing simple moving average SMA_n(x)_t = (1/n) sum_{k=0}^{n-1} x_{t-k}; NaN for the first n-1 values."""
    raise NotImplementedError


def ema(x: pd.Series, n: int) -> pd.Series:
    """EMA with alpha = 2/(n+1): E_t = alpha x_t + (1-alpha) E_{t-1}, E_0 = x_0 (no NaNs)."""
    raise NotImplementedError


def wilder_smooth(x: pd.Series, n: int) -> pd.Series:
    """Wilder's smoother on a differenced series (x_0 is NaN).

    W at position n = mean(x_1..x_n); then W_t = ((n-1) W_{t-1} + x_t) / n. Positions < n are NaN.
    """
    raise NotImplementedError


def rsi(close: pd.Series, n: int = 14) -> pd.Series:
    """Wilder's RSI_t = 100 U_t / (U_t + D_t) (notes Definition 4.4).

    U = wilder_smooth(max(dP, 0)), D = wilder_smooth(max(-dP, 0)), dP_t = P_t - P_{t-1}.
    If U + D = 0 return 50. Identity (Prop. 4.3): RSI = 50 + 50 W(dP) / W(|dP|).
    """
    raise NotImplementedError


def donchian_high(high: pd.Series, n: int) -> pd.Series:
    """max(H_{t-n}, ..., H_{t-1}): highest high of the PREVIOUS n bars (excludes bar t)."""
    raise NotImplementedError


def donchian_low(low: pd.Series, n: int) -> pd.Series:
    """min(L_{t-n}, ..., L_{t-1})."""
    raise NotImplementedError


def breakout(close: pd.Series, high: pd.Series, n: int) -> pd.Series:
    """Boolean close_t > donchian_high(high, n)_t; False where undefined."""
    raise NotImplementedError


def relative_volume(volume: pd.Series, n: int = 50) -> pd.Series:
    """RVOL_t = V_t / mean(V_{t-n}, ..., V_{t-1}); the average excludes today."""
    raise NotImplementedError


def mansfield_rs(price: pd.Series, index: pd.Series, n: int = 252) -> pd.Series:
    """Mansfield relative strength 100 (RS_t / SMA_n(RS)_t - 1), RS = price / index."""
    raise NotImplementedError


def trend_template(close: pd.Series, rs_rank: pd.Series | None = None, rs_min: float = 70.0) -> pd.Series:
    """Minervini Trend Template (notes Definition 4.6) as a boolean Series on daily bars.

    1. C > SMA150 and C > SMA200; 2. SMA150 > SMA200; 3. SMA200_t > SMA200_{t-21};
    4. SMA50 > SMA150 and SMA50 > SMA200; 5. C > SMA50; 6. C >= 1.30 * min(C over last 252 bars incl. t);
    7. C >= 0.75 * max(C over last 252 bars incl. t); 8. if rs_rank given, rs_rank >= rs_min.
    False wherever any input is undefined.
    """
    raise NotImplementedError


def weinstein_stage(weekly_close: pd.Series, n: int = 30, slope_lag: int = 4, flat_tol: float = 0.005) -> pd.Series:
    """Weinstein stage 1-4 (notes Definition 4.5).

    m = SMA_n(close); slope_t = m_t / m_{t-slope_lag} - 1.
    Stage 2 if close > m and slope > flat_tol; Stage 4 if close < m and slope < -flat_tol;
    otherwise Stage 3 if the most recent Stage-2-or-4 label was 2, else Stage 1 (start in "1").
    Float Series, NaN until m and slope are defined.
    """
    raise NotImplementedError


# --------------------------------------------------------------- strategies


def sma_crossover_strategy(fast: int = 50, slow: int = 200) -> Callable[[pd.DataFrame], float]:
    """Return a strategy(history) -> +1 if mean of last `fast` closes > mean of last `slow` closes, else 0.
    Return 0 while fewer than `slow` bars are available."""
    raise NotImplementedError


def tsmom_strategy(lookback: int = 252, skip: int = 21) -> Callable[[pd.DataFrame], float]:
    """Return a strategy(history) -> sign(ln(C_{t-skip} / C_{t-lookback})), with C_t the last close.
    Return 0 while len(history) <= lookback."""
    raise NotImplementedError


# --------------------------------------------------------------- engine


def run_backtest(
    bars: pd.DataFrame,
    strategy: Callable[[pd.DataFrame], float],
    cost_bps: float = 0.0,
    slippage_bps: float = 0.0,
    capital: float = 1.0,
    rebalance: str = "on_change",
) -> pd.DataFrame:
    """Event-driven backtest (notes Algorithm 4.1). bars has columns open, high, low, close, volume.

    State: cash = capital, shares = 0, target = 0 (pending), executed = 0 (last executed target).
    For t = 0..T-1:
      (a) eq_open = cash + shares * open_t.
          If rebalance == "always" or target != executed: desired = target * eq_open / open_t and executed = target;
          else desired = shares.
          trade = desired - shares. If trade != 0: fill = open_t (1 + sign(trade) s), s = slippage_bps/1e4;
          commission = |trade| fill c, c = cost_bps/1e4; cash -= trade * fill + commission;
          cost_t = commission + |trade| open_t s; shares = desired.
          turnover_t = |trade| open_t / eq_open.
      (b) equity_t = cash + shares * close_t.
      (c) target = strategy(bars.iloc[: t + 1])   # the strategy sees bars up to t only
    Raise ValueError for an unknown rebalance mode.
    Return DataFrame(index=bars.index, columns=["target", "shares", "equity", "cost", "turnover"]),
    where target is the value returned in (c) at t and shares is the holding over bar t.
    """
    raise NotImplementedError


def vectorised_backtest(
    bars: pd.DataFrame,
    target: pd.Series,
    cost_bps: float = 0.0,
    slippage_bps: float = 0.0,
    capital: float = 1.0,
) -> pd.Series:
    """Vectorised equivalent of run_backtest(..., rebalance="always") (notes Prop. 4.8).

    a_t = target_{t-1} (a_0 = 0); g_t = open_t / open_{t-1} (g_0 = 1);
    drifted weight d_t = a_{t-1} g_t / (1 + a_{t-1}(g_t - 1)) (a_{-1} = 0);
    E^o_t = E^o_{t-1} (1 + a_{t-1}(g_t - 1)) (1 - k |a_t - d_t|), k = (cost_bps + slippage_bps)/1e4, E^o_{-1} = capital;
    E_t = E^o_t (1 + a_t (close_t / open_t - 1)). Return the Series E_t named "equity".
    """
    raise NotImplementedError


def performance_summary(equity: pd.Series, periods: int = 252) -> dict:
    """{'cagr', 'vol', 'sharpe', 'max_drawdown'} from simple returns of equity (rf = 0).

    cagr = (E_end/E_0)^(1/years) - 1 with years = (number of returns)/periods;
    vol = std(ddof=1) sqrt(periods); sharpe = mean/std sqrt(periods); max_drawdown = min(E / cummax(E) - 1).
    """
    raise NotImplementedError