wickra/examples/python/backtest.py

"""Offline backtest example: compute a basket of indicators over a CSV history.

Run with::

    python -m examples.python.backtest path/to/ohlcv.csv

The CSV must have a header row with at least the columns
``timestamp, open, high, low, close, volume``. The script computes a panel of
indicators with the standard parameters used across Wickra's tests and
prints a small summary of the resulting series — enough to verify that the
indicators are wired correctly without pulling in pandas or a charting stack.
"""

from __future__ import annotations

import argparse
import csv
import sys
from dataclasses import dataclass
from typing import List

import numpy as np

import wickra as ta


# Columns the OHLCV layout requires; the CSV header must name every one.
REQUIRED_COLUMNS = ("timestamp", "open", "high", "low", "close", "volume")


@dataclass
class History:
    timestamp: np.ndarray
    open: np.ndarray
    high: np.ndarray
    low: np.ndarray
    close: np.ndarray
    volume: np.ndarray


def read_history(path: str) -> History:
    """Load an OHLCV CSV into typed NumPy columns.

    Raises:
        ValueError: if the file has no header, is missing a required column,
            holds no data rows, or contains a non-numeric value (the message
            pinpoints the offending row and column instead of surfacing an
            opaque ``KeyError`` or NumPy ``ValueError``).
    """
    rows: List[List[str]] = []
    with open(path, newline="") as f:
        reader = csv.DictReader(f)
        if reader.fieldnames is None:
            raise ValueError(f"{path}: CSV has no header row")
        missing = [c for c in REQUIRED_COLUMNS if c not in reader.fieldnames]
        if missing:
            raise ValueError(
                f"{path}: CSV is missing required column(s): "
                f"{', '.join(missing)}; found: {', '.join(reader.fieldnames)}"
            )
        for row in reader:
            rows.append([row[c] for c in REQUIRED_COLUMNS])
    if not rows:
        raise ValueError(f"{path}: CSV has a header but no data rows")
    try:
        arr = np.array(rows, dtype=np.float64)
    except (TypeError, ValueError) as exc:
        # NumPy's own message names neither the row nor the column — locate
        # the first non-numeric cell and report it precisely.
        for line_no, row in enumerate(rows, start=2):
            for col, value in zip(REQUIRED_COLUMNS, row):
                try:
                    float(value)
                except (TypeError, ValueError):
                    raise ValueError(
                        f"{path}: row {line_no} column '{col}' is not "
                        f"numeric: {value!r}"
                    ) from exc
        raise  # could not localise — surface the original error
    return History(
        timestamp=arr[:, 0].astype(np.int64),
        open=arr[:, 1],
        high=arr[:, 2],
        low=arr[:, 3],
        close=arr[:, 4],
        volume=arr[:, 5],
    )


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description=__doc__.splitlines()[0] if __doc__ else None)
    p.add_argument("path", help="Path to an OHLCV CSV file")
    p.add_argument("--rsi", type=int, default=14)
    p.add_argument("--ema", type=int, default=20)
    p.add_argument("--bb-period", type=int, default=20)
    p.add_argument("--bb-mult", type=float, default=2.0)
    return p.parse_args()


def summarize(name: str, values: np.ndarray) -> None:
    valid = values[~np.isnan(values)]
    if valid.size == 0:
        print(f"  {name:<12} (no valid samples — series too short)")
        return
    print(
        f"  {name:<12} mean={valid.mean():>10.4f}  min={valid.min():>10.4f}  "
        f"max={valid.max():>10.4f}  last={valid[-1]:>10.4f}"
    )


def main() -> int:
    args = parse_args()
    history = read_history(args.path)

    rsi = ta.RSI(args.rsi).batch(history.close)
    ema = ta.EMA(args.ema).batch(history.close)
    macd = ta.MACD().batch(history.close)  # shape (n, 3)
    bb = ta.BollingerBands(args.bb_period, args.bb_mult).batch(history.close)  # (n, 4)
    atr = ta.ATR(14).batch(history.high, history.low, history.close)
    adx = ta.ADX(14).batch(history.high, history.low, history.close)  # (n, 3)
    obv = ta.OBV().batch(history.close, history.volume)

    print(f"Backtest summary for {args.path} ({history.close.size} bars)")
    summarize(f"RSI({args.rsi})", rsi)
    summarize(f"EMA({args.ema})", ema)
    summarize("MACD line", macd[:, 0])
    summarize("MACD hist", macd[:, 2])
    summarize(f"BB upper", bb[:, 0])
    summarize(f"BB lower", bb[:, 2])
    summarize("ATR(14)", atr)
    summarize("ADX(14)", adx[:, 2])
    summarize("OBV", obv)

    return 0


if __name__ == "__main__":
    sys.exit(main())