"""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()