#!/usr/bin/env python3 """ MACD + 200 MA trend filter template. Bring your own OHLC CSV. Expected columns: datetime, open, high, low, close Example: python3 run_backtest.py --csv your_ohlc.csv --cost-bps 2 --rr 1.5 """ from __future__ import annotations import argparse from pathlib import Path import numpy as np import pandas as pd def load_ohlc(path: Path) -> pd.DataFrame: df = pd.read_csv(path) df.columns = [column.lower() for column in df.columns] required = {"open", "high", "low", "close"} missing = required - set(df.columns) if missing: raise SystemExit(f"CSV is missing required columns: {', '.join(sorted(missing))}") if "datetime" in df.columns: df["datetime"] = pd.to_datetime(df["datetime"]) df = df.sort_values("datetime") else: df["datetime"] = pd.RangeIndex(len(df)) for column in ["open", "high", "low", "close"]: df[column] = pd.to_numeric(df[column], errors="coerce") return df.dropna(subset=["open", "high", "low", "close"]).reset_index(drop=True) def add_indicators(df: pd.DataFrame, fast: int, slow: int, signal: int, ma_length: int) -> pd.DataFrame: work = df.copy() fast_ema = work["close"].ewm(span=fast, adjust=False, min_periods=fast).mean() slow_ema = work["close"].ewm(span=slow, adjust=False, min_periods=slow).mean() work["macd"] = fast_ema - slow_ema work["macd_signal"] = work["macd"].ewm(span=signal, adjust=False, min_periods=signal).mean() work["ma"] = work["close"].rolling(ma_length).mean() work["cross_up"] = (work["macd"].shift(1) <= work["macd_signal"].shift(1)) & (work["macd"] > work["macd_signal"]) work["cross_down"] = (work["macd"].shift(1) >= work["macd_signal"].shift(1)) & (work["macd"] < work["macd_signal"]) return work def run_backtest(df: pd.DataFrame, args: argparse.Namespace) -> tuple[pd.DataFrame, dict[str, float]]: work = add_indicators(df, args.fast, args.slow, args.signal, args.ma_length) trades = [] min_risk = args.min_risk_bps / 10000 for i in range(1, len(work)): prev = work.iloc[i - 1] row = work.iloc[i] if pd.isna(prev["ma"]) or pd.isna(prev["macd_signal"]): continue side = 0 if prev["cross_up"] and prev["macd"] < 0 and prev["close"] > prev["ma"]: side = 1 elif prev["cross_down"] and prev["macd"] > 0 and prev["close"] < prev["ma"]: side = -1 if side == 0: continue entry = float(row["open"]) ma_stop = float(prev["ma"]) if side == 1: stop = ma_stop * (1 - args.stop_buffer_bps / 10000) risk = entry - stop if risk <= entry * min_risk: continue target = entry + args.rr * risk else: stop = ma_stop * (1 + args.stop_buffer_bps / 10000) risk = stop - entry if risk <= entry * min_risk: continue target = entry - args.rr * risk exit_price = float(work.iloc[min(i + args.max_hold, len(work) - 1)]["close"]) exit_index = min(i + args.max_hold, len(work) - 1) exit_reason = "time" for j in range(i, min(i + args.max_hold + 1, len(work))): bar = work.iloc[j] if side == 1: hit_stop = float(bar["low"]) <= stop hit_target = float(bar["high"]) >= target else: hit_stop = float(bar["high"]) >= stop hit_target = float(bar["low"]) <= target # Daily OHLC cannot reveal intrabar order, so use stop-first ambiguity. if hit_stop: exit_price = stop exit_index = j exit_reason = "stop" break if hit_target: exit_price = target exit_index = j exit_reason = "target" break gross = side * (exit_price / entry - 1) net = gross - args.cost_bps / 10000 trades.append( { "entry_time": row["datetime"], "exit_time": work.iloc[exit_index]["datetime"], "side": "long" if side == 1 else "short", "entry": entry, "stop": stop, "target": target, "exit": exit_price, "exit_reason": exit_reason, "gross_return": gross, "net_return": net, } ) trades_df = pd.DataFrame(trades) if trades_df.empty: return trades_df, { "trades": 0, "win_rate": 0.0, "avg_trade": 0.0, "median_trade": 0.0, "total_net_return": 0.0, "profit_factor": 0.0, } wins = trades_df.loc[trades_df["net_return"] > 0, "net_return"] losses = trades_df.loc[trades_df["net_return"] < 0, "net_return"] profit_factor = wins.sum() / abs(losses.sum()) if abs(losses.sum()) > 0 else np.inf return trades_df, { "trades": int(len(trades_df)), "win_rate": float((trades_df["net_return"] > 0).mean()), "avg_trade": float(trades_df["net_return"].mean()), "median_trade": float(trades_df["net_return"].median()), "total_net_return": float((1 + trades_df["net_return"]).prod() - 1), "profit_factor": float(profit_factor), } def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--csv", required=True, type=Path, help="Path to your OHLC CSV") parser.add_argument("--out-dir", type=Path, default=Path("macd_200ma_results")) parser.add_argument("--fast", type=int, default=12) parser.add_argument("--slow", type=int, default=26) parser.add_argument("--signal", type=int, default=9) parser.add_argument("--ma-length", type=int, default=200) parser.add_argument("--rr", type=float, default=1.5, help="Reward/risk target") parser.add_argument("--max-hold", type=int, default=60) parser.add_argument("--cost-bps", type=float, default=2.0, help="Round-trip cost in basis points") parser.add_argument("--stop-buffer-bps", type=float, default=0.0) parser.add_argument("--min-risk-bps", type=float, default=5.0) args = parser.parse_args() args.out_dir.mkdir(parents=True, exist_ok=True) df = load_ohlc(args.csv) trades, summary = run_backtest(df, args) trades.to_csv(args.out_dir / "trades.csv", index=False) pd.DataFrame([summary]).to_csv(args.out_dir / "summary.csv", index=False) print(pd.DataFrame([summary]).to_string(index=False)) if __name__ == "__main__": main()