112 lines
5.2 KiB
Python
112 lines
5.2 KiB
Python
import os
|
|
import pandas as pd
|
|
import pandas_ta as ta
|
|
from dotenv import load_dotenv
|
|
from datetime import datetime, timedelta
|
|
|
|
from backtesting import Backtest
|
|
from strategies import RsiStrategy
|
|
from alpaca.data.historical import StockHistoricalDataClient
|
|
from alpaca.data.requests import StockBarsRequest
|
|
from alpaca.data.timeframe import TimeFrame
|
|
|
|
# --- 1. DATEN HOLEN ---
|
|
load_dotenv()
|
|
API_KEY = os.getenv('ALPACA_API_KEY')
|
|
SECRET_KEY = os.getenv('ALPACA_SECRET_KEY')
|
|
PATH_TO_OUTPUT = "output/"
|
|
os.makedirs(PATH_TO_OUTPUT, exist_ok=True)
|
|
|
|
def get_data(symbol, days=365):
|
|
client = StockHistoricalDataClient(API_KEY, SECRET_KEY)
|
|
start_date = datetime.now() - timedelta(days=days)
|
|
request_params = StockBarsRequest(symbol_or_symbols=[symbol], timeframe=TimeFrame.Hour, start=start_date)
|
|
df = client.get_stock_bars(request_params).df
|
|
df = df.reset_index(level=0, drop=True)
|
|
df.columns = [c.capitalize() for c in df.columns]
|
|
df.index = df.index.tz_localize(None)
|
|
return df
|
|
|
|
# Strategy classes moved to strategies.py
|
|
|
|
# --- 3. OPTIMIERUNGS-ENGINE ---
|
|
def run_optimized_backtest(symbol, strategy_cls=RsiStrategy, optimize_kwargs=None, report_tag=None):
|
|
data = get_data(symbol)
|
|
bt = Backtest(data, strategy_cls, cash=10000, commission=0.001, finalize_trades=True)
|
|
|
|
print(f"--- Starte Optimierung für {symbol} using {strategy_cls.__name__} ---")
|
|
|
|
# Build common optimization params for strategy if not provided
|
|
if optimize_kwargs is None:
|
|
optimize_kwargs = dict(
|
|
rsi_period=range(7, 30, 2), # Teste Perioden von 7 bis 29 in 2er Schritten
|
|
rsi_lower=range(20, 40, 5), # Teste Kaufsignale von 20 bis 35
|
|
rsi_upper=range(60, 80, 5), # Teste Verkaufsignale von 60 bis 75
|
|
maximize='Return [%]', # Wir wollen den höchsten Gewinn (oder 'Sharpe Ratio')
|
|
constraint=lambda p: p.rsi_upper > p.rsi_lower, # Logik-Check
|
|
)
|
|
# Extend with ATR/stop params if strategy supports them
|
|
if hasattr(strategy_cls, 'atr_period'):
|
|
optimize_kwargs['atr_period'] = range(10, 20, 2)
|
|
if hasattr(strategy_cls, 'stop_loss_atr_multiplier'):
|
|
optimize_kwargs['stop_loss_atr_multiplier'] = [2.0, 2.5, 3.0, 3.5, 4.0]
|
|
|
|
# Run optimization
|
|
stats = bt.optimize(**optimize_kwargs)
|
|
|
|
print("\n--- BESTE PARAMETER GEFUNDEN ---")
|
|
print(stats)
|
|
print("\nDetails der besten Strategie:")
|
|
# Print only attributes that the strategy actually has
|
|
for attr in ('rsi_period', 'rsi_lower', 'rsi_upper', 'short_ema', 'long_ema', 'atr_period', 'stop_loss_atr_multiplier'):
|
|
if hasattr(stats._strategy, attr):
|
|
print(f"{attr.replace('_', ' ').capitalize()}: {getattr(stats._strategy, attr)}")
|
|
# Speichere den Chart der besten Strategie
|
|
tag = f"_{report_tag}" if report_tag else ""
|
|
out_path = os.path.join(PATH_TO_OUTPUT, f"optimized_report_{symbol}{tag}.html")
|
|
bt.plot(filename=out_path, open_browser=False)
|
|
print(f"Optimized report saved: {out_path}")
|
|
|
|
# Build a result dict to return to callers
|
|
result = {
|
|
'symbol': symbol,
|
|
'rsi_period': getattr(stats._strategy, 'rsi_period', None),
|
|
'rsi_lower': getattr(stats._strategy, 'rsi_lower', None),
|
|
'rsi_upper': getattr(stats._strategy, 'rsi_upper', None),
|
|
'short_ema': getattr(stats._strategy, 'short_ema', None),
|
|
'long_ema': getattr(stats._strategy, 'long_ema', None),
|
|
'atr_period': getattr(stats._strategy, 'atr_period', None),
|
|
'stop_loss_atr_multiplier': getattr(stats._strategy, 'stop_loss_atr_multiplier', None),
|
|
'return_pct': stats.get('Return [%]') if hasattr(stats, 'get') else None,
|
|
'equity_final_$': stats.get('Equity Final [$]') if hasattr(stats, 'get') else None,
|
|
'max_drawdown_pct': stats.get('Max. Drawdown [%]') if hasattr(stats, 'get') else None,
|
|
'n_trades': stats.get('# Trades') if hasattr(stats, 'get') else None,
|
|
'win_rate_pct': stats.get('Win Rate [%]') if hasattr(stats, 'get') else None,
|
|
}
|
|
|
|
# Run a final backtest with the best-found parameters to export the full trades list
|
|
best_params = {}
|
|
for attr in ('rsi_period', 'rsi_lower', 'rsi_upper', 'short_ema', 'long_ema', 'atr_period', 'stop_loss_atr_multiplier'):
|
|
if hasattr(stats._strategy, attr):
|
|
best_params[attr] = getattr(stats._strategy, attr)
|
|
|
|
try:
|
|
if best_params:
|
|
print(f"Running final backtest for {symbol} with best params: {best_params}")
|
|
final_stats = bt.run(**best_params)
|
|
trades = final_stats.get('_trades')
|
|
if trades is not None and not trades.empty:
|
|
# add duration column and export
|
|
trades['Duration'] = trades['ExitTime'] - trades['EntryTime']
|
|
tag_name = report_tag if report_tag else strategy_cls.__name__
|
|
trades_file = os.path.join(PATH_TO_OUTPUT, f"Backtest_Trades_{symbol}_{tag_name}.xlsx")
|
|
trades.to_excel(trades_file, index=False)
|
|
print(f"✅ Trades exported: {trades_file}")
|
|
result['trades_file'] = trades_file
|
|
except Exception as e:
|
|
print(f"Failed to run final backtest for trades export: {e}")
|
|
|
|
return result
|
|
|
|
if __name__ == "__main__":
|
|
run_optimized_backtest("GOLD") |