Backtesting Your Rules

# ta-config.yaml — backtesting section
# OpenClaw reads these settings when running ta_backtest.py

watchlist:
  tickers:
    - AAPL
    - SPY
    - NVDA

backtest:
  strategy: "rsi_mean_reversion"  # Which strategy to test
  period: "5y"                    # How much history (5y = 5 years)

  # RSI Mean Reversion parameters
  rsi_buy:    30    # Buy when RSI drops below this
  rsi_sell:   60    # Sell when RSI rises above this
  max_hold:   30    # Sell after this many days if RSI hasn't recovered

# ta_backtest.py — OpenClaw Backtesting Module
# Tests your signal rules against historical price data

import yaml
import yfinance as yf
import pandas_ta as ta
import pandas as pd
from datetime import datetime

# ── Load config ───────────────────────────────────────────────────
def load_config(path: str = "ta-config.yaml") -> dict:
    """Load settings from ta-config.yaml."""
    with open(path) as f:
        return yaml.safe_load(f)

cfg = load_config()
BT  = cfg["backtest"]

# ── Download data ─────────────────────────────────────────────────
def get_data(ticker: str) -> pd.DataFrame:
    """Download and prepare daily OHLCV data."""
    df = yf.download(ticker, period=BT["period"], interval="1d",
                     progress=False)
    df.columns = [c.lower() for c in df.columns]
    df = df[["open","high","low","close","volume"]].dropna()
    df["rsi"] = ta.rsi(df["close"], 14)
    return df.dropna()

# ── Run the backtest ──────────────────────────────────────────────
def run_backtest(ticker: str) -> list:
    """
    Simulates the RSI mean reversion strategy on historical data.
    Returns a list of completed trades.

    IMPORTANT: We use next-day's open price for all entries/exits.
    This ensures we're not using data that wasn't available yet.
    """
    df = get_data(ticker)
    trades = []
    in_trade = False
    entry_price = None
    entry_date  = None
    entry_idx   = None

    for i in range(1, len(df) - 1):
        rsi_yesterday = df.iloc[i - 1]["rsi"]

        if not in_trade:
            # Entry: RSI was oversold yesterday, buy at today's open
            if rsi_yesterday < BT["rsi_buy"]:
                entry_price = df.iloc[i]["open"]
                entry_date  = df.index[i]
                entry_idx   = i
                in_trade    = True

        else:
            days_held = i - entry_idx
            # Exit: RSI recovered OR we've held too long
            if (rsi_yesterday > BT["rsi_sell"] or
                days_held >= BT["max_hold"]):
                exit_price = df.iloc[i]["open"]
                exit_date  = df.index[i]
                return_pct = (exit_price - entry_price) / entry_price * 100

                trades.append({
                    "entry_date":   entry_date.date(),
                    "exit_date":    exit_date.date(),
                    "entry_price":  round(entry_price, 2),
                    "exit_price":   round(exit_price, 2),
                    "days_held":    days_held,
                    "return_pct":   round(return_pct, 2),
                    "win":          return_pct > 0,
                    "exit_reason":  ("RSI_SELL" if rsi_yesterday > BT["rsi_sell"]
                                     else "MAX_HOLD"),
                })
                in_trade = False

    return trades

# ── Analyse and print results ─────────────────────────────────────
def print_results(ticker: str, trades: list) -> None:
    """Print backtest results in a human-readable format."""
    if not trades:
        print(f"\n{ticker}: No trades — try widening RSI thresholds.")
        return

    returns = [t["return_pct"] for t in trades]
    wins    = [r for r in returns if r > 0]
    losses  = [r for r in returns if r <= 0]
    win_rate = len(wins) / len(returns) * 100
    avg_return = sum(returns) / len(returns)
    avg_days = sum(t["days_held"] for t in trades) / len(trades)

    # Max drawdown: worst peak-to-trough on cumulative returns
    cumulative, peak, max_dd = 0, 0, 0
    for r in returns:
        cumulative += r
        peak = max(peak, cumulative)
        max_dd = max(max_dd, peak - cumulative)

    print(f"\n{'🦞 OpenClaw Backtest Results':=^55}")
    print(f"  Ticker:          {ticker}")
    print(f"  Strategy:        RSI Mean Reversion")
    print(f"  Parameters:      Buy < {BT['rsi_buy']} | Sell > {BT['rsi_sell']} | "
          f"Max hold {BT['max_hold']}d")
    print(f"  Period:          {BT['period']}")
    print(f"{'─'*55}")
    print(f"  Total trades:    {len(trades)}")
    print(f"  Win rate:        {win_rate:.1f}%")
    print(f"  Avg return:      {avg_return:+.2f}% per trade")
    print(f"  Best trade:      {max(returns):+.2f}%")
    print(f"  Worst trade:     {min(returns):+.2f}%")
    print(f"  Max drawdown:    -{max_dd:.2f}%")
    print(f"  Avg hold time:   {avg_days:.1f} days")
    print(f"{'─'*55}")
    print(f"\n  Last 5 trades:")
    print(f"  {'Entry':<12} {'Exit':<12} {'Return':>8} {'Days':>5} "
          f"{'Exit Reason':<12}")
    for t in trades[-5:]:
        icon = "✓" if t["win"] else "✗"
        print(f"  {str(t['entry_date']):<12} {str(t['exit_date']):<12} "
              f"{t['return_pct']:>+7.2f}% {t['days_held']:>5}  "
              f"{icon} {t['exit_reason']}")

# ── Run for all tickers in config ────────────────────────────────
for ticker in cfg["watchlist"]["tickers"]:
    trades = run_backtest(ticker)
    print_results(ticker, trades)