run_backtest.py

import argparse
from pathlib import Path

import pandas as pd

from trading_bot.agent import Agent, TradeConfig
from trading_bot.methods import evaluate_model, count_action, summarise_backtest
from trading_bot.utils import format_currency, format_position, format_pct


def _valid_prediction_files(paths):
    out = []
    for p in paths:
        p = Path(p)
        if not p.exists():
            continue
        name = p.name
        if not name.endswith(".csv"):
            continue
        if "_backtest_history" in name or "_scored" in name or "backtest_summary" in name:
            continue
        out.append(p)
    return sorted(out)


def _decision_rows(history_df: pd.DataFrame) -> pd.DataFrame:
    if "row_type" in history_df.columns:
        return history_df[history_df["row_type"] != "terminal"].copy()
    return history_df.copy()


def _default_history_path(pred_path: Path, cfg: TradeConfig) -> Path:
    return pred_path.with_name(f"{pred_path.stem}_{cfg.strategy_tag()}_backtest_history.csv")


def _run_one(pred_path: Path, cfg: TradeConfig, debug: bool, out_history: Path | None = None) -> dict:
    pred_df = pd.read_csv(pred_path)
    agent = Agent(cfg)
    history_df, final_assets = evaluate_model(agent, pred_df, debug=debug)
    decision_df = _decision_rows(history_df)

    summary = summarise_backtest(
        history_df=history_df,
        final_assets=final_assets,
        initial_capital=cfg.initial_capital,
        pred_df=pred_df,
        cost_bps=cfg.cost_bps,
    )

    if out_history is None:
        out_history = _default_history_path(pred_path, cfg)

    out_history.parent.mkdir(parents=True, exist_ok=True)
    history_df.to_csv(out_history, index=False)

    print(f"\n=== {pred_path.name} ===")
    print("Strategy tag:", cfg.strategy_tag())
    if summary.get("backtest_start_date") and summary.get("backtest_end_date"):
        print("Backtest window:", f"{summary['backtest_start_date']} → {summary['backtest_end_date']}")
    print("Final assets:", format_currency(summary["final_assets"]))
    print("Profit:", format_position(summary["profit"]))
    print("Return:", format_pct(summary["return_pct"]))
    print("Buy & Hold Return:", format_pct(summary["buy_hold_return_pct"]))
    print("Alpha vs Buy & Hold:", format_pct(summary["alpha_vs_buy_hold_pct"]))
    print("Max drawdown:", format_pct(summary["max_drawdown_pct"]))
    print("Avg exposure:", format_pct(summary["avg_exposure_pct"]))
    print("Days in market:", format_pct(summary["pct_days_in_market"]))
    print("Turnover / initial capital:", format_pct(summary["turnover_pct"]))
    print(f"Buys: {summary['n_buys']} | Sells: {summary['n_sells']} | SELL_ALL: {summary['n_sell_all']}")

    if not decision_df.empty:
        print("\nActions:")
        print(count_action(decision_df))

    print("Saved history to:", out_history)

    return {
        "prediction_file": pred_path.name,
        "history_file": out_history.name,
        "strategy_tag": cfg.strategy_tag(),
        "initial_capital": cfg.initial_capital,
        "max_position_percent": cfg.max_position_percent,
        "min_expected_return": cfg.min_expected_return,
        "strong_expected_return": cfg.strong_expected_return,
        "rebalance_band": cfg.rebalance_band,
        "cost_bps": cfg.cost_bps,
        "warmup_days": cfg.warmup_days,
        **summary,
    }


def main():
    parser = argparse.ArgumentParser(
        description="Run a horizon-matched daily long/flat backtest on one prediction CSV or a batch."
    )

    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--pred_path", help="Single prediction CSV with columns: date, Real, Predicted")
    group.add_argument("--pred_glob", help='Glob for many prediction CSVs, e.g. "data/predictions/*_w4_pred_*.csv"')

    parser.add_argument("--out_history", default=None, help="Optional output path for single-file history CSV")
    parser.add_argument("--summary_out", default="data/predictions/backtest_summary.csv", help="Output CSV for batch summary")
    parser.add_argument("--debug", action="store_true", help="Show progress bar during backtest")

    # Strategy params
    parser.add_argument("--initial_capital", type=float, default=10_000.0)
    parser.add_argument("--max_position_percent", type=float, default=0.60)
    parser.add_argument("--min_expected_return", type=float, default=0.005)
    parser.add_argument("--strong_expected_return", type=float, default=0.03)
    parser.add_argument("--rebalance_band", type=float, default=0.05)
    parser.add_argument("--cost_bps", type=float, default=5.0)
    parser.add_argument("--warmup_days", type=int, default=0)

    args = parser.parse_args()

    cfg = TradeConfig(
        initial_capital=args.initial_capital,
        max_position_percent=args.max_position_percent,
        min_expected_return=args.min_expected_return,
        strong_expected_return=args.strong_expected_return,
        rebalance_band=args.rebalance_band,
        cost_bps=args.cost_bps,
        warmup_days=args.warmup_days,
    )

    if args.pred_path:
        pred_path = Path(args.pred_path)
        if not pred_path.exists():
            raise FileNotFoundError(f"Prediction file not found: {pred_path}")

        out_history = Path(args.out_history) if args.out_history is not None else None
        _run_one(pred_path=pred_path, cfg=cfg, debug=args.debug, out_history=out_history)
        return

    matched = [Path(p) for p in Path().glob(args.pred_glob)]
    pred_files = _valid_prediction_files(matched)

    if not pred_files:
        raise FileNotFoundError(f"No valid prediction CSVs matched glob: {args.pred_glob}")

    rows = []
    for pred_path in pred_files:
        row = _run_one(pred_path=pred_path, cfg=cfg, debug=args.debug, out_history=None)
        rows.append(row)

    summary_df = pd.DataFrame(rows).sort_values(
        ["return_pct", "alpha_vs_buy_hold_pct", "final_assets"],
        ascending=[False, False, False]
    )

    summary_out = Path(args.summary_out)
    summary_out.parent.mkdir(parents=True, exist_ok=True)
    summary_df.to_csv(summary_out, index=False)

    print("\n=== Batch summary ===")
    print(summary_df[[
        "prediction_file",
        "strategy_tag",
        "backtest_start_date",
        "backtest_end_date",
        "final_assets",
        "profit",
        "return_pct",
        "buy_hold_return_pct",
        "alpha_vs_buy_hold_pct",
        "max_drawdown_pct",
        "avg_exposure_pct",
        "pct_days_in_market",
        "turnover_pct",
        "n_buys",
        "n_sells",
    ]])
    print("\nSaved batch summary to:", summary_out)


if __name__ == "__main__":
    main()