quant/research/single_stock_swing.py

"""Single-stock swing trading: adapt V3/V7 concepts to 20 famous stocks.

Strategy: long the stock when trending, cash when not.
- Trend: stock > MA + momentum > 0 + vol < cap + no dd breach
- Position sizing: vol-target overlay
- Risk mgmt: stop-loss + profit-take
- When flat: 100% cash (0% return)

Tests per-stock optimized parameters + a universal parameter set.
"""
from __future__ import annotations

import sys
sys.path.insert(0, ".")

import numpy as np
import pandas as pd
from itertools import product

import data_manager
import metrics

YEARS = 5
CAPITAL = 100_000
TX_COST = 0.002  # 2bp for individual stocks (wider spreads)

STOCKS = [
    "AAPL", "MSFT", "GOOGL", "AMZN", "NVDA", "META", "TSLA",  # Mag 7
    "JPM", "V", "MA",        # Financials
    "JNJ", "UNH", "HD",      # Healthcare / Consumer
    "PG", "KO", "DIS",       # Consumer staples / media
    "NFLX", "AMD", "CRM",    # Tech / growth
    "COST",                   # Retail
]


def swing_backtest(
    prices: pd.Series,
    ma_window: int = 50,
    mom_window: int = 21,
    vol_window: int = 20,
    vol_cap: float = 0.40,
    dd_window: int = 20,
    dd_stop: float = 0.08,
    confirm_days: int = 2,
    min_hold: int = 5,
    stop_loss: float = 0.08,
    profit_take: float = 0.0,
    target_vol: float = 0.25,
    min_scale: float = 0.3,
    tx_cost: float = TX_COST,
) -> tuple[pd.Series, dict]:
    """Backtest a single-stock swing strategy. Returns (equity, stats)."""
    arr = prices.to_numpy().astype(float)
    n = len(arr)
    need = max(ma_window, mom_window, vol_window, dd_window) + 1

    # Precompute indicators
    ma = pd.Series(arr, index=prices.index).rolling(ma_window).mean().to_numpy()
    rets = np.diff(arr, prepend=arr[0]) / np.maximum(np.roll(arr, 1), 1e-12)
    rets[0] = 0.0
    vol = pd.Series(rets).rolling(vol_window, min_periods=10).std().to_numpy() * np.sqrt(252)

    # State machine
    in_position = False
    entry_price = 0.0
    bars_held = 0
    pending_entry = 0
    equity = np.ones(n) * CAPITAL
    n_trades = 0

    for t in range(1, n):
        equity[t] = equity[t - 1]

        if t < need:
            continue

        p = arr[t - 1]  # yesterday's close (PIT-safe)
        p_ma = ma[t - 1]
        p_vol = vol[t - 1]
        p_mom = arr[t - 1] / arr[t - 1 - mom_window] - 1 if arr[t - 1 - mom_window] > 0 else 0
        p_dd = arr[t - 1] / np.max(arr[max(0, t - 1 - dd_window):t]) - 1

        if np.isnan(p) or np.isnan(p_ma):
            continue

        # --- Trend signal ---
        trend_bull = (p > p_ma and p_mom > 0 and
                      (np.isnan(p_vol) or p_vol < vol_cap) and
                      p_dd > -dd_stop)

        if in_position:
            bars_held += 1
            # Apply daily return
            daily_r = arr[t] / arr[t - 1] - 1 if arr[t - 1] > 0 else 0

            # Vol-target scaling
            scale = target_vol / p_vol if p_vol > 0.01 else 1.0
            scale = np.clip(scale, min_scale, 1.0)
            equity[t] = equity[t - 1] * (1 + daily_r * scale)

            # Check exit conditions (using yesterday's close, PIT-safe)
            gain = p / entry_price - 1 if entry_price > 0 else 0

            exit_signal = False
            # Stop-loss
            if gain <= -stop_loss:
                exit_signal = True
            # Profit-take
            if profit_take > 0 and gain >= profit_take:
                exit_signal = True
            # Trend reversal (with min_hold)
            if not trend_bull and bars_held >= min_hold:
                exit_signal = True

            if exit_signal:
                equity[t] -= equity[t] * tx_cost  # exit cost
                in_position = False
                pending_entry = 0
                n_trades += 1
        else:
            # Check entry
            if trend_bull:
                pending_entry += 1
                if pending_entry >= confirm_days:
                    in_position = True
                    entry_price = arr[t]  # enter at today's close
                    bars_held = 0
                    equity[t] -= equity[t] * tx_cost  # entry cost
                    n_trades += 1
            else:
                pending_entry = 0

    eq = pd.Series(equity, index=prices.index)
    total_ret = eq.iloc[-1] / eq.iloc[0] - 1
    days_in = sum(1 for t in range(need, n) if equity[t] != equity[t - 1])
    pct_in = days_in / (n - need) if n > need else 0

    return eq, {
        "total_return": total_ret,
        "n_trades": n_trades,
        "pct_time_in": pct_in,
    }


def optimize_stock(prices: pd.Series, stock: str) -> tuple[dict, dict, pd.Series]:
    """Grid search for best parameters on a single stock."""
    param_grid = {
        "ma_window": [20, 50, 100, 150],
        "mom_window": [10, 21, 42],
        "vol_cap": [0.30, 0.45, 0.60, 999],
        "dd_stop": [0.05, 0.08, 0.12],
        "stop_loss": [0.05, 0.08, 0.12],
        "profit_take": [0.0, 0.15, 0.25],
        "target_vol": [0.20, 0.30, 0.40],
        "min_hold": [3, 5, 10],
        "confirm_days": [1, 2, 3],
    }

    # Fixed params
    fixed = {"vol_window": 20, "dd_window": 20, "min_scale": 0.3}

    best_sharpe = -np.inf
    best_params = {}
    best_eq = None

    keys = list(param_grid.keys())
    values = list(param_grid.values())

    for combo in product(*values):
        params = dict(zip(keys, combo))
        params.update(fixed)
        try:
            eq, stats = swing_backtest(prices, **params)
            m = metrics.raw_summary(eq)
            if m["sharpeRatio"] > best_sharpe and stats["n_trades"] >= 5:
                best_sharpe = m["sharpeRatio"]
                best_params = params.copy()
                best_eq = eq
        except Exception:
            continue

    return best_params, metrics.raw_summary(best_eq) if best_eq is not None else {}, best_eq


def main():
    print("=" * 110)
    print("  SINGLE-STOCK SWING TRADING: 20 FAMOUS STOCKS")
    print("=" * 110)

    # Download data
    print(f"\nDownloading {len(STOCKS)} stocks...")
    data = data_manager.update("swing", STOCKS, with_open=False)
    if isinstance(data, tuple):
        data = data[0]
    cutoff = data.index[-1] - pd.DateOffset(years=YEARS)
    data = data[data.index >= cutoff]
    print(f"Period: {data.index[0].date()} → {data.index[-1].date()}")

    # Buy-and-hold benchmarks
    print(f"\n--- Buy & Hold Returns ({YEARS}y) ---")
    bh_returns = {}
    for stock in STOCKS:
        if stock not in data.columns:
            continue
        s = data[stock].dropna()
        if len(s) < 100:
            continue
        r = s.iloc[-1] / s.iloc[0] - 1
        ann = (1 + r) ** (252 / len(s)) - 1
        bh_returns[stock] = ann
        print(f"  {stock:<6}: {ann*100:>+6.1f}% ann")

    # Universal parameter set (sensible defaults)
    print(f"\n--- Universal Parameters (no per-stock optimization) ---")
    print(f"{'Stock':<7} {'Ann%':>7} {'Vol%':>7} {'Sharpe':>7} {'MaxDD%':>7} {'Trades':>7} {'%In':>6}  {'B&H Ann%':>9}")
    print("-" * 75)

    universal = dict(ma_window=50, mom_window=21, vol_window=20, vol_cap=0.45,
                     dd_window=20, dd_stop=0.08, confirm_days=2, min_hold=5,
                     stop_loss=0.08, profit_take=0.0, target_vol=0.25, min_scale=0.3)

    for stock in STOCKS:
        if stock not in data.columns:
            continue
        prices = data[stock].dropna()
        if len(prices) < 200:
            continue
        eq, stats = swing_backtest(prices, **universal)
        m = metrics.raw_summary(eq)
        bh = bh_returns.get(stock, 0)
        print(f"  {stock:<6} {m['annualizedReturn']*100:>+6.1f}% {m['annualizedVolatility']*100:>6.1f}% "
              f"{m['sharpeRatio']:>7.2f} {m['maxDrawdown']*100:>6.1f}% "
              f"{stats['n_trades']:>7} {stats['pct_time_in']*100:>5.1f}%  {bh*100:>+8.1f}%")

    # Per-stock optimized
    print(f"\n--- Per-Stock Optimized (grid search on Sharpe, min 5 trades) ---")
    print(f"{'Stock':<7} {'Ann%':>7} {'Vol%':>7} {'Sharpe':>7} {'MaxDD%':>7} {'Trades':>7} {'%In':>6}  {'B&H Ann%':>9}  Best params")
    print("-" * 130)

    opt_results = []
    for stock in STOCKS:
        if stock not in data.columns:
            continue
        prices = data[stock].dropna()
        if len(prices) < 200:
            continue
        print(f"  Optimizing {stock}...", end=" ", flush=True)
        params, m, eq = optimize_stock(prices, stock)
        if not m:
            print("FAILED")
            continue
        _, stats = swing_backtest(prices, **params)
        bh = bh_returns.get(stock, 0)
        key_params = f"MA{params.get('ma_window')}/mom{params.get('mom_window')}/SL{int(params.get('stop_loss',0)*100)}%/PT{int(params.get('profit_take',0)*100)}%/VT{int(params.get('target_vol',0)*100)}%"
        print(f"{m['annualizedReturn']*100:>+5.1f}% Sharpe={m['sharpeRatio']:.2f} [{key_params}]")
        opt_results.append((stock, m, stats, params, bh))

    print(f"\n{'Stock':<7} {'Ann%':>7} {'Vol%':>7} {'Sharpe':>7} {'MaxDD%':>7} {'Trades':>7} {'%In':>6}  {'B&H Ann%':>9}  Params")
    print("-" * 130)
    opt_results.sort(key=lambda x: x[1]["sharpeRatio"], reverse=True)
    for stock, m, stats, params, bh in opt_results:
        key_params = f"MA{params.get('ma_window')}/mom{params.get('mom_window')}/SL{int(params.get('stop_loss',0)*100)}%/PT{int(params.get('profit_take',0)*100)}%/VT{int(params.get('target_vol',0)*100)}%"
        beat = "✓" if m["annualizedReturn"] > bh else "✗"
        print(f"  {stock:<6} {m['annualizedReturn']*100:>+6.1f}% {m['annualizedVolatility']*100:>6.1f}% "
              f"{m['sharpeRatio']:>7.2f} {m['maxDrawdown']*100:>6.1f}% "
              f"{stats['n_trades']:>7} {stats['pct_time_in']*100:>5.1f}%  {bh*100:>+8.1f}% {beat}  {key_params}")

    winners = sum(1 for _, m, _, _, bh in opt_results if m["annualizedReturn"] > bh)
    print(f"\n  Beat buy-and-hold: {winners}/{len(opt_results)} stocks")
    avg_sharpe = np.mean([m["sharpeRatio"] for _, m, _, _, _ in opt_results])
    print(f"  Average Sharpe: {avg_sharpe:.2f}")


if __name__ == "__main__":
    main()