import os import csv import argparse import pandas as pd import pandas_ta as ta from dotenv import load_dotenv from datetime import datetime, timedelta # Backtesting-Spezialisten from backtesting import Backtest, Strategy from backtesting.lib import crossover # Alpaca API from alpaca.data.historical import StockHistoricalDataClient from alpaca.data.requests import StockBarsRequest from alpaca.data.timeframe import TimeFrame # --- CONFIG --- load_dotenv() API_KEY = os.getenv('ALPACA_API_KEY') SECRET_KEY = os.getenv('ALPACA_SECRET_KEY') PATH_TO_OUTPUT = "output/" # Ensure output folder exists os.makedirs(PATH_TO_OUTPUT, exist_ok=True) # --- KLASSE: DATEN-ENGINE --- class DataEngine: @staticmethod def get_alpaca_data(symbol, days=365): client = StockHistoricalDataClient(API_KEY, SECRET_KEY) start_date = datetime.now() - timedelta(days=days) request = StockBarsRequest( symbol_or_symbols=[symbol], timeframe=TimeFrame.Hour, start=start_date ) df = client.get_stock_bars(request).df df = df.reset_index(level=0, drop=True) # Formatierung für Backtesting.py (Spalten müssen groß geschrieben sein) df.columns = [c.capitalize() for c in df.columns] # Zeitzonen entfernen (wichtig für Excel!) df.index = df.index.tz_localize(None) return df # --- KLASSE: STRATEGIE (RSI) --- class MyRsiStrategy(Strategy): # Parameter - diese können später optimiert werden rsi_period = 29 rsi_low = 35 rsi_high = 75 # Position sizing: fraction of equity to allocate per new trade (0-1) position_size_pct = 0.1 # Fallback fixed minimum shares to buy if computed shares is 0 min_shares = 1 # ATR-based stop-loss parameters atr_period = 18 # Stop-loss multiplier applied to ATR (e.g. 3 -> stop = entry - 3 * ATR) stop_loss_atr_multiplier = 4.0 def init(self): # Indikator berechnen (self.I stellt sicher, dass er im Chart erscheint) self.rsi = self.I(ta.rsi, pd.Series(self.data.Close), length=self.rsi_period) # Average True Range for volatility-based stops self.atr = self.I( ta.atr, pd.Series(self.data.High), pd.Series(self.data.Low), pd.Series(self.data.Close), length=self.atr_period, ) def next(self): # KAUFEN: Wenn RSI die untere Grenze von unten nach oben kreuzt if crossover(self.rsi, self.rsi_low): price = float(self.data.Close[-1]) # Use a conservative fraction of current equity to size the position max_invest = max(0, self.equity * self.position_size_pct) shares = int(max_invest // price) if shares < self.min_shares: shares = self.min_shares # Compute ATR-based stop-loss price below entry atr_value = float(self.atr[-1]) if not pd.isna(self.atr[-1]) else None if atr_value and atr_value > 0: stop_price = round(price - (self.stop_loss_atr_multiplier * atr_value), 6) # Ensure stop is below price if stop_price >= price: stop_price = round(price * 0.99, 6) else: # Fallback to a small percent-based stop if ATR not ready stop_price = round(price * 0.98, 6) # Place an integer-sized buy order with SL (avoids relative-size cancellation) self.buy(size=shares, sl=stop_price) # VERKAUFEN: Wenn RSI die obere Grenze von oben nach unten kreuzt elif crossover(self.rsi_high, self.rsi): if self.position: self.position.close() # --- HAUPTPROGRAMM --- def run_backtest(symbol="AAPL", days=365, cash=1000, commission=0.002): # 1. Daten laden print(f"Lade Daten für {symbol}...") data = DataEngine.get_alpaca_data(symbol, days=days) # 2. Backtest initialisieren bt = Backtest(data, MyRsiStrategy, cash=cash, commission=commission, finalize_trades=True) # 3. Backtest ausführen stats = bt.run() # --- AUSWERTUNG --- print("\n" + "="*30) print(f"BACKTEST ERGEBNISSE FÜR {symbol}") print("="*30) print(f"Startkapital: {cash}$") print(f"Endkapital: {stats['Equity Final [$]']:.2f}$") print(f"Rendite [%]: {stats['Return [%]']:.2f}%") print(f"Max. Drawdown: {stats['Max. Drawdown [%]']:.2f}%") print(f"Anzahl Trades: {stats['# Trades']}") print(f"Win Rate [%]: {stats['Win Rate [%]']:.2f}%") print("="*30) result = { 'symbol': symbol, 'start_cash': cash, 'end_equity': stats.get('Equity Final [$]', None), 'return_pct': stats.get('Return [%]', None), 'max_drawdown_pct': stats.get('Max. Drawdown [%]', None), 'n_trades': stats.get('# Trades', None), 'win_rate_pct': stats.get('Win Rate [%]', None), } # 4. EXPORT: Trades nach Excel trades = stats.get('_trades') if trades is not None and not trades.empty: trades['Duration'] = trades['ExitTime'] - trades['EntryTime'] excel_name = os.path.join(PATH_TO_OUTPUT, f"Backtest_Trades_{symbol}.xlsx") trades.to_excel(excel_name) print(f"✅ Excel-Liste gespeichert: {excel_name}") result['trades_file'] = excel_name # 5. EXPORT: Interaktiver HTML Report report_name = os.path.join(PATH_TO_OUTPUT, f"Backtest_Report_{symbol}.html") bt.plot(filename=report_name, open_browser=False) print(f"✅ Interaktiver Chart gespeichert: {report_name}") result['report_file'] = report_name return result def run_backtests(symbols, days=365, cash=1000, commission=0.002, summary_name=None): summary = [] for sym in symbols: try: res = run_backtest(symbol=sym, days=days, cash=cash, commission=commission) summary.append(res) except Exception as e: print(f"⚠️ Fehler beim Backtest für {sym}: {e}") summary.append({'symbol': sym, 'error': str(e)}) # Write summary CSV if summary_name is None: ts = datetime.now().strftime('%Y%m%d_%H%M%S') summary_name = os.path.join(PATH_TO_OUTPUT, f"backtest_summary_{ts}.csv") keys = set() for row in summary: keys.update(row.keys()) keys = list(keys) with open(summary_name, 'w', newline='', encoding='utf-8') as csvfile: writer = csv.DictWriter(csvfile, fieldnames=keys) writer.writeheader() for row in summary: writer.writerow(row) print(f"\n✅ Zusammenfassung gespeichert: {summary_name}") return summary_name if __name__ == "__main__": parser = argparse.ArgumentParser(description='Batch backtester for multiple stocks') parser.add_argument('--tickers', type=str, help='Comma-separated list of tickers, e.g. AAPL,MSFT,TSLA') parser.add_argument('--file', type=str, help='Path to a file containing one ticker per line') parser.add_argument('--days', type=int, default=365, help='Days of historical data to fetch') parser.add_argument('--cash', type=float, default=1000, help='Starting cash per backtest') parser.add_argument('--commission', type=float, default=0.002, help='Commission (fraction), e.g. 0.002') args = parser.parse_args() symbols = [] if args.tickers: symbols = [s.strip().upper() for s in args.tickers.split(',') if s.strip()] elif args.file: if os.path.exists(args.file): with open(args.file, 'r', encoding='utf-8') as f: symbols = [line.strip().upper() for line in f if line.strip()] else: print(f"Ticker-Datei nicht gefunden: {args.file}") raise SystemExit(1) else: symbols = ["AAPL"] print(f"Starte Batch-Backtest für: {symbols}") summary_csv = run_backtests(symbols, days=args.days, cash=args.cash, commission=args.commission) print(f"Fertig. Übersicht: {summary_csv}")