quant/main.py

import argparse

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd

import data_manager
import factor_attribution
import metrics
from strategies.adaptive_momentum import AdaptiveMomentumStrategy
from strategies.buy_and_hold import BuyAndHoldStrategy
from strategies.dual_momentum import DualMomentumStrategy
from strategies.inverse_vol import InverseVolatilityStrategy
from strategies.mean_reversion import MeanReversionStrategy
from strategies.momentum import MomentumStrategy
from strategies.momentum_quality import MomentumQualityStrategy
from strategies.multi_factor import MultiFactorStrategy
from strategies.recovery_momentum import RecoveryMomentumStrategy
from strategies.trend_following import TrendFollowingStrategy
from universe import UNIVERSES


# ---------------------------------------------------------------------------
# Backtest engine
# ---------------------------------------------------------------------------

def backtest(
    strategy,
    data: pd.DataFrame,
    initial_capital: float = 100_000,
    transaction_cost: float = 0.001,
    fixed_fee: float = 0.0,
    open_data: pd.DataFrame | None = None,
) -> pd.Series:
    """
    Vectorized backtest.

    Parameters
    ----------
    strategy : Strategy
        Any class implementing generate_signals(data) → DataFrame of weights.
    data : pd.DataFrame
        Adjusted close prices, one column per asset.
    initial_capital : float
        Starting portfolio value.
    transaction_cost : float
        One-way cost per unit of turnover (e.g. 0.001 = 10 bps).
    fixed_fee : float
        Fixed dollar cost per individual trade (each buy or sell).
    open_data : pd.DataFrame, optional
        Open prices. When provided, enables open-to-close execution mode:
        - Morning: observe open prices → run strategy → decide weights
        - Evening: execute all trades at close prices
        Strategies have an internal shift(1) designed for close prices.
        Since open prices are observable same-day (before close), we undo
        that shift so signals use today's open and execute at today's close.

    Returns
    -------
    pd.Series
        Daily equity curve.
    """
    if open_data is not None:
        # Open-to-close mode:
        # Strategy's shift(1) on open prices gives: weights[t] = f(open_{t-1})
        # But open_t is known at morning of day t, so undo shift to get f(open_t)
        # Then execute at close_t, earning close_t → close_{t+1}
        weights = strategy.generate_signals(open_data)
        weights = weights.shift(-1).fillna(0.0)
        weights = weights.reindex(data.index).fillna(0.0)
        positions = weights

        # Returns earned: close[t] → close[t+1], weighted by positions decided at open[t]
        close_returns = data.pct_change().fillna(0.0)
        portfolio_returns = (close_returns * positions.shift(1).fillna(0.0)).sum(axis=1)
    else:
        # Classic close-to-close mode
        weights = strategy.generate_signals(data)
        weights = weights.reindex(data.index).fillna(0.0)
        positions = weights

        daily_returns = data.pct_change().fillna(0.0)
        portfolio_returns = (daily_returns * positions).sum(axis=1)

    # Turnover cost: sum of absolute weight changes each day
    turnover = positions.diff().abs().sum(axis=1).fillna(0.0)
    portfolio_returns -= turnover * transaction_cost

    # Fixed per-trade fee: count positions with non-zero weight change
    if fixed_fee > 0:
        weight_changes = positions.diff().fillna(0.0)
        n_trades = (weight_changes.abs() > 1e-8).sum(axis=1)
        # Build running equity to convert dollar fees to return impact
        equity_running = (1 + portfolio_returns).cumprod() * initial_capital
        fee_impact = (n_trades * fixed_fee) / equity_running.shift(1).fillna(initial_capital)
        portfolio_returns -= fee_impact

    equity = (1 + portfolio_returns).cumprod() * initial_capital
    return equity


# ---------------------------------------------------------------------------
# Visualization
# ---------------------------------------------------------------------------

def plot_results(results: pd.DataFrame) -> None:
    """Two-panel chart: equity curves (top) + drawdowns (bottom)."""
    # Compute drawdowns
    drawdowns = results.div(results.cummax()) - 1

    fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(14, 9), sharex=True,
                                    gridspec_kw={"height_ratios": [3, 1]})

    for col in results.columns:
        ax1.plot(results.index, results[col], label=col, linewidth=1.5)
    ax1.set_title("Strategy Comparison — Equity Curves", fontsize=14)
    ax1.set_ylabel("Portfolio Value ($)")
    ax1.legend(loc="upper left")
    ax1.grid(True, alpha=0.3)

    for col in drawdowns.columns:
        ax2.plot(drawdowns.index, drawdowns[col] * 100, label=col, linewidth=1.0)
    ax2.set_title("Drawdowns")
    ax2.set_ylabel("Drawdown (%)")
    ax2.set_xlabel("Date")
    ax2.grid(True, alpha=0.3)

    plt.tight_layout()
    plt.show()


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main() -> None:
    parser = argparse.ArgumentParser(description="Run strategy backtest")
    parser.add_argument(
        "--market", choices=UNIVERSES.keys(), default="us",
        help="Market universe to backtest (default: us)",
    )
    parser.add_argument(
        "--capital", type=float, default=None,
        help="Initial capital (default: 10000)",
    )
    parser.add_argument(
        "--top-n", type=int, default=None,
        help="Number of stocks for selective strategies (default: ~10%% of universe)",
    )
    parser.add_argument(
        "--years", type=int, default=None,
        help="Limit backtest to last N years of data",
    )
    parser.add_argument(
        "--no-plot", action="store_true",
        help="Skip plotting charts",
    )
    parser.add_argument(
        "--fixed-fee", type=float, default=0.0,
        help="Fixed dollar cost per trade, e.g. 2.0 means $2 per buy or sell",
    )
    parser.add_argument(
        "--execution", choices=["close", "open-close"], default="close",
        help="Execution mode: 'close' (default, signal & execute on close) or "
             "'open-close' (signal on morning open, execute at close)",
    )
    parser.add_argument(
        "--attribution", action="store_true",
        help="Run factor attribution after performance metrics",
    )
    parser.add_argument(
        "--attribution-model", choices=["capm", "ff5", "ff5plus", "all"], default="all",
        help="Factor model selection for attribution output",
    )
    parser.add_argument(
        "--attribution-export", default=None,
        help="Directory to export factor attribution CSVs",
    )
    args = parser.parse_args()
    initial_capital = args.capital if args.capital is not None else 10_000
    use_open = args.execution == "open-close"

    universe = UNIVERSES[args.market]
    tickers = universe["fetch"]()
    benchmark = universe["benchmark"]
    benchmark_label = universe["benchmark_label"]
    all_tickers = sorted(set(tickers + [benchmark]))

    result = data_manager.update(args.market, all_tickers, with_open=use_open)
    if use_open:
        data, open_data = result
    else:
        data = result
        open_data = None

    if args.years:
        cutoff = data.index[-1] - pd.DateOffset(years=args.years)
        data = data[data.index >= cutoff]
        if open_data is not None:
            open_data = open_data[open_data.index >= cutoff]
        print(f"--- Sliced to last {args.years} years: {data.index[0].date()} to {data.index[-1].date()} ---")

    # Filter tickers to only those in the data
    tickers = [t for t in tickers if t in data.columns]
    print(f"--- Universe: {len(tickers)} stocks + {benchmark} benchmark ---")

    top_n = args.top_n if args.top_n else max(5, len(tickers) // 10)
    print(f"--- Selective strategies will pick top {top_n} stocks ---")
    if args.fixed_fee > 0:
        print(f"--- Fixed fee: ${args.fixed_fee:.2f} per trade ---")
    if use_open:
        print(f"--- Execution: open-close (signal on open, execute at close) ---")

    # Build strategy tuples: (strategy, close_data, open_data_or_None)
    open_tickers = open_data[tickers] if open_data is not None else None
    open_all = open_data if open_data is not None else None

    strategies = {
        # --- Original strategies ---
        "Buy & Hold (EW)":    (BuyAndHoldStrategy(),                                          data[tickers], open_tickers),
        "Momentum":           (MomentumStrategy(lookback=252, skip=21, top_n=top_n),           data[tickers], open_tickers),
        "Inverse Volatility": (InverseVolatilityStrategy(vol_window=20),                       data[tickers], open_tickers),
        "Multi-Factor":       (MultiFactorStrategy(tickers=tickers, benchmark=benchmark, top_n=top_n), data, open_all),
        # --- New strategies ---
        "Mean Reversion":     (MeanReversionStrategy(top_n=top_n),                                data[tickers], open_tickers),
        "Trend Following":    (TrendFollowingStrategy(ma_window=150, momentum_period=126, top_n=top_n), data[tickers], open_tickers),
        "Dual Momentum":      (DualMomentumStrategy(top_n=top_n),                               data[tickers], open_tickers),
        "Momentum+Quality":   (MomentumQualityStrategy(momentum_period=252, skip=21, top_n=top_n), data[tickers], open_tickers),
        "Mom+InvVol":         (AdaptiveMomentumStrategy(top_n=top_n),                          data[tickers], open_tickers),
        "Recovery+Mom Top20": (RecoveryMomentumStrategy(top_n=min(20, top_n)),                  data[tickers], open_tickers),
        "Recovery+Mom Top10": (RecoveryMomentumStrategy(top_n=10),                              data[tickers], open_tickers),
    }

    results: dict[str, pd.Series] = {}
    for name, (strategy, strat_data, strat_open) in strategies.items():
        print(f"\nRunning: {name}")
        results[name] = backtest(strategy, strat_data, initial_capital=initial_capital,
                                  fixed_fee=args.fixed_fee, open_data=strat_open)

    # Add benchmark (normalized to same initial capital)
    bench = data[benchmark].dropna()
    results[benchmark_label] = (bench / bench.iloc[0]) * initial_capital

    results_df = pd.DataFrame(results)

    # --- Performance metrics ---
    for name, equity in results_df.items():
        eq = equity.dropna()
        if len(eq) < 2:
            print(f"\n  {name}: insufficient data, skipping")
            continue
        metrics.summary(eq, name=name)

    if args.attribution:
        summary_df, loadings_df = factor_attribution.attribute_strategies(
            results_df=results_df,
            benchmark_label=benchmark_label,
            benchmark=benchmark,
            price_data=data,
            market=args.market,
            model_selection=args.attribution_model,
        )
        factor_attribution.print_attribution_summary(summary_df)
        if args.attribution_export:
            factor_attribution.export_attribution(summary_df, loadings_df, args.attribution_export)
            print(f"Attribution CSVs written to {args.attribution_export}")

    # --- Visualization ---
    if not args.no_plot:
        plot_results(results_df.dropna())


if __name__ == "__main__":
    main()