Backtesting

def backtesting(ticker, start, end): df = yf.download(ticker, start=start, end=end, auto_adjust=True) df = df[["Close"]].copy() # Segnali df["sma20"] = df.Close.rolling(20).mean() df["sma50"] = df.Close.rolling(50).mean() df["signal"] = (df.sma20. > df.sma50).astype(int) df = df.dropna() # Rendimenti giornalieri (con lag = no look-ahead bias) df["ret_mkt"] = df.Close.pct_change() df["ret_strat"] = df.ret_mkt * df.signal.shift(1) # Equity curve cumulativa df["eq_s"] = (1 + df.ret_strat).cumprod() df["eq_bh"] = (1 + df.ret_mkt).cumprod() n_anni = len(df) / 252 cagr = df.eq_s.iloc[-1] ** (1/n_anni) - 1 return {"ticker": ticker, "CAGR": (cagr*100).round(1), "Rend_tot": ((df.eq_s.iloc[-1]-1)*100).round(1), "Buy&Hold": ((df.eq_bh.iloc[-1]-1)*100).round(1)} tickers = ["AAPL", "MSFT", "GOOGL", "AMZN", "META", "TSLA"] start, end ="2019-01-01", "2024-01-01" ris = [] for ticker in tickers: ris.append(backtesting(ticker, start, end)) pd.DataFrame(ris)

def backtesting(ticker, start, end): df = yf.download(ticker, start=start, end=end, auto_adjust=True) df = df[["Close"]].copy() # Segnali df["sma20"] = df.Close.rolling(20).mean() df["sma50"] = df.Close.rolling(50).mean() df["signal"] = (df.sma20. > df.sma50).astype(int) df = df.dropna() # Rendimenti giornalieri (con lag = no look-ahead bias) df["ret_mkt"] = df.Close.pct_change() df["ret_strat"] = df.ret_mkt * df.signal.shift(1) # Equity curve cumulativa df["eq_s"] = (1 + df.ret_strat).cumprod() df["eq_bh"] = (1 + df.ret_mkt).cumprod() n_anni = len(df) / 252 cagr = df["eq_s"].iloc[-1] ** (1/n_anni) - 1 r = df.ret_strat.dropna() rf = 0.03 # rendimento free risk exc = r - rf/252 eq = df.eq_s.dropna() return {"ticker": ticker, "CAGR": (cagr*100).round(1), "Rend_tot": ((df["eq_s"].iloc[-1]-1)*100).round(1), "Buy&Hold": ((df['eq_bh'].iloc[-1]-1)*100).round(1), "Sharpe": (np.sqrt(252) * exc.mean() / exc.std()).round(2), "max_dd": round((((eq - eq.cummax()) / eq.cummax()).min()), 2), "PF": (r[r>0].sum() / r[r<0].abs().sum()).round(2) } tickers = ["AAPL", "MSFT", "GOOGL", "AMZN", "META", "TSLA"] start, end ="2019-01-01", "2024-01-01" ris = [] for ticker in tickers: ris.append(backtesting(ticker, start, end)) pd.DataFrame(ris)

import backtrader as bt class SMACross(bt.Strategy): params = (('fast', 20), ('slow', 50),) def __init__(self): self.fast = bt.ind.SMA(self.data, period=self.p.fast) self.slow = bt.ind.SMA(self.data, period=self.p.slow) self.up = bt.ind.CrossUp(self.fast, self.slow) self.down = bt.ind.CrossDown(self.fast, self.slow) self.crossover = bt.ind.CrossOver(self.fast, self.slow) self.order = None # To keep track of pending orders def next(self): if self.order: # Check if an order is pending return if not self.position: # Not in the market if self.crossover > 0: # fast crosses slow upwards # self.log.info(f'BUY CREATE: {self.data.datetime.date(0)} - Close: {self.data.close[0]:.2f}') self.order = self.buy(size=100) # Buy 100 shares else: # In the market if self.crossover < 0: # fast crosses slow downwards # self.log.info(f'SELL CREATE: {self.data.datetime.date(0)} - Close: {self.data.close[0]:.2f}') self.order = self.close() # Close existing position def notify_order(self, order): if order.status in [order.Submitted, order.Accepted]: # Order has been submitted/accepted - no notification on this return # Check if an order has been completed if order.status in [order.Completed]: if order.isbuy(): # self.log.info(f'BUY EXECUTED, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}, Comm: {order.executed.comm:.2f}') pass elif order.issell(): # self.log.info(f'SELL EXECUTED, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}, Comm: {order.executed.comm:.2f}') pass self.bar_executed = len(self) elif order.status in [order.Canceled, order.Margin, order.Rejected]: # self.log.info('Order Canceled/Margin/Rejected') pass # Clear pending order self.order = None cerebro = bt.Cerebro() cerebro.addstrategy(SMACross) cerebro.broker.set_cash(10_000) cerebro.broker.setcommission(commission=0.001) raw = yf.download("AAPL", start="2019-01-01", end="2024-01-01") # Flatten MultiIndex columns to single-level strings expected by backtrader # Based on the kernel state, the columns are already transformed to 'close_aapl', 'high_aapl', etc. # This block correctly transforms the column names from MultiIndex to single-level, but `backtrader` still needs explicit mapping. new_columns = [] for col_tuple in raw.columns: if col_tuple[0] == 'Price': new_columns.append(col_tuple[1].lower()) # Convert to lowercase (e.g., 'Close' -> 'close') elif col_tuple[0] == 'Volume': new_columns.append('volume') # Convert to lowercase else: new_columns.append('_'.join(col_tuple).lower()) # Fallback for unexpected columns raw.columns = new_columns cerebro.adddata( bt.feeds.PandasData( dataname=raw, datetime=None, # Index is datetime open='open_aapl', high='high_aapl', low='low_aapl', close='close_aapl', volume='volume', openinterest=None # Not available in yfinance data ) ) cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe') cerebro.addanalyzer(bt.analyzers.DrawDown, _name='dd') print(f"Capitale iniziale: ${cerebro.broker.getvalue():,.0f}") res = cerebro.run()[0] print(f"Capitale finale: ${cerebro.broker.getvalue():,.0f}") # Check if sharpe ratio was calculated before printing sharpe_ratio_analysis = res.analyzers.sharpe.get_analysis() if 'sharperatio' in sharpe_ratio_analysis and sharpe_ratio_analysis['sharperatio'] is not None: print(f"Sharpe: {sharpe_ratio_analysis['sharperatio']:.2f}") else: print("Sharpe: N/A (No trades or insufficient data)") max_dd_analysis = res.analyzers.dd.get_analysis() if max_dd_analysis and max_dd_analysis.max.drawdown is not None: print(f"Max DD: {max_dd_analysis.max.drawdown:.1f}%") else: print("Max DD: N/A (No trades or insufficient data)") figs = cerebro.plot(iplot=False, style='candlestick') plt.savefig('backtest.png', dpi=100, bbox_inches='tight') from IPython.display import Image Image('backtest.png')

df = yf.download("AAPL", start="2015-01-01", end="2024-01-01", auto_adjust=True) df.columns = [col[0] for col in df.columns] df["ret"] = df.Close.pct_change() def backtest(data, f, s): d = data.copy() # Segnale: 1 quando fast MA > slow MA (trend rialzista) 0 altrimenti (non investi) d["sig"] = (d.Close.rolling(f).mean() > d.Close.rolling(s).mean()).astype(int) # Ritorna strategia: segnale shiftato * rendimento, poi annualizzato return (d.ret * d.sig.shift()).dropna().mean() * 252 candidati = [(f,s) for f in [10,20,30] for s in [50,100,200] if f < s] for a in range(2017, 2024): train = df[f"{a-2}-01-01":f"{a}-01-01"] # due anni prima test = df[f"{a}-01-01":f"{a}-07-01"] # sei mesi dopo best = max(candidati, key=lambda p: backtest(train, *p)) oos = backtest(test, *best) print(f"anno: {a}, best params: {best}, oos_cagr: {round(oos*100,1)}")

from collections import Counter # Scarica dati df = yf.download("AAPL", start="2015-01-01", end="2024-01-01", auto_adjust=True) df.columns = [col[0] for col in df.columns] df["ret"] = df.Close.pct_change() def backtest(data, f, s): """Versione base: ritorna CAGR annualizzato""" d = data.copy() d["sig"] = (d.Close.rolling(f).mean() > d.Close.rolling(s).mean()).astype(int) return (d.ret * d.sig.shift()).dropna().mean() * 252 def backtest_avanzato(data, f, s): """Versione con metriche multiple""" d = data.copy() d["sig"] = (d.Close.rolling(f).mean() > d.Close.rolling(s).mean()).astype(int) d["strat_ret"] = d.ret * d.sig.shift() returns = d["strat_ret"].dropna() if len(returns) == 0: return {'cagr': -999, 'sharpe': -999, 'trades': 0} cagr = returns.mean() * 252 sharpe = returns.mean() / returns.std() * np.sqrt(252) if returns.std() > 0 else 0 trades = (d["sig"].diff() != 0).sum() return {'cagr': cagr, 'sharpe': sharpe, 'trades': trades} def score_composito(metriche): """Metrica composita per selezionare il miglior modello""" # Preferiamo Sharpe alto, ma penalizziamo troppi trades return metriche['sharpe'] - (metriche['trades'] / 100) * 0.1 # Griglia parametri candidati = [(f,s) for f in [10,20,30] for s in [50,100,200] if f < s] # Walk-forward analysis risultati = [] print("="*70) print("WALK-FORWARD ANALYSIS") print("="*70) for a in range(2017, 2024): train = df[f"{a-2}-01-01":f"{a}-01-01"] test = df[f"{a}-01-01":f"{a}-07-01"] if len(train) == 0 or len(test) == 0: continue # Ottimizzazione su train (usa metrica composita) best_params = None best_score = -np.inf for f,s in candidati: metriche = backtest_avanzato(train, f, s) score = score_composito(metriche) if score > best_score: best_score = score best_params = (f, s) # Test su out-of-sample perf_oos = backtest_avanzato(test, *best_params) risultati.append({ 'anno': a, 'params': best_params, 'oos_cagr': perf_oos['cagr'], 'oos_sharpe': perf_oos['sharpe'], 'trades': perf_oos['trades'] }) print(f"{a}: params={best_params}, cagr={perf_oos['cagr']*100:5.1f}%, sharpe={perf_oos['sharpe']:.2f}, trades={perf_oos['trades']}") # Analisi finale print("\n" + "="*70) print("ANALISI FINALE") print("="*70) df_ris = pd.DataFrame(risultati) print(f"CAGR medio OOS: {df_ris['oos_cagr'].mean()*100:.1f}%") print(f"CAGR std OOS: {df_ris['oos_cagr'].std()*100:.1f}%") print(f"Sharpe medio OOS: {df_ris['oos_sharpe'].mean():.2f}") print(f"Finestre positive: {(df_ris['oos_cagr'] > 0).sum()}/{len(df_ris)}") # Stabilità parametri params_più_comuni = Counter([str(p) for p in df_ris['params']]).most_common(3) print(f"\nParametri più stabili nel tempo:") for params, count in params_più_comuni: print(f" {params}: {count} finestre") # Verdetto if df_ris['oos_cagr'].mean() > 0 and (df_ris['oos_cagr'] > 0).sum() > len(df_ris)*0.7: print("\n✅ VERDETTO: Sistema ROBUSTO") elif df_ris['oos_cagr'].mean() > 0: print("\n🟡 VERDETTO: Sistema INSTABILE ma potenzialmente profittevole") else: print("\n🔴 VERDETTO: Sistema NON ROBUSTO - evita di usarlo")