from __future__ import annotations import argparse import dataclasses from pathlib import Path from typing import Optional import pandas as pd from backtester.data_loader import load_csv from backtester.indicators.money_line import MoneyLineInputs from backtester.simulator import TradeConfig, simulate_money_line def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run Money Line backtests against CSV OHLCV data.") parser.add_argument("--csv", type=Path, required=True, help="Path to CSV file with timestamp, open, high, low, close, volume columns") parser.add_argument("--symbol", type=str, required=True, help="Symbol label (e.g., SOL-PERP)") parser.add_argument("--timeframe", type=str, default="5", help="Timeframe label for reference (default: 5)") parser.add_argument("--start", type=str, default=None, help="Optional ISO timestamp for start filter (e.g., 2024-01-01)") parser.add_argument("--end", type=str, default=None, help="Optional ISO timestamp for end filter") parser.add_argument("--position-size", type=float, default=1000.0, dest="position_size", help="Notional size in USD for each simulated trade") parser.add_argument("--max-bars", type=int, default=None, help="Maximum bars to hold a trade before force exit") parser.add_argument("--export-trades", type=Path, default=None, help="Optional path to write detailed trade results as CSV") return parser.parse_args() def main() -> None: args = parse_args() data_slice = load_csv( path=args.csv, symbol=args.symbol, timeframe=args.timeframe, start=args.start, end=args.end, ) df = data_slice.data if not isinstance(df.index, pd.DatetimeIndex): df = df.copy() df.index = pd.to_datetime(df.index) config = TradeConfig(position_size=args.position_size, max_bars_per_trade=args.max_bars) inputs = MoneyLineInputs() result = simulate_money_line(df=df, symbol=data_slice.symbol, inputs=inputs, config=config) print("=== Backtest Summary ===") print(f"Symbol: {data_slice.symbol}") print(f"Rows processed: {len(df)}") print(f"Trades: {len(result.trades)}") print(f"Total PnL: ${result.total_pnl:,.2f}") print(f"Average PnL: ${result.average_pnl:,.2f}") print(f"Win rate: {result.win_rate * 100:.2f}%") print(f"Max drawdown: ${result.max_drawdown:,.2f}") if args.export_trades: trades_df = pd.DataFrame( [ {k: v for k, v in dataclasses.asdict(trade).items() if k != "_exit_index"} for trade in result.trades ] ) trades_df.to_csv(args.export_trades, index=False) print(f"Detailed trades exported to {args.export_trades}") if __name__ == "__main__": main()