mirror of
https://github.com/B-Wear/QuantumEdge.git
synced 2026-07-27 15:37:46 +00:00
B-Wear
266 lines
9.6 KiB
Python
266 lines
9.6 KiB
Python
import pandas as pd
|
|
import numpy as np
|
|
from datetime import datetime, timedelta
|
|
import logging
|
|
from typing import Dict, List, Optional
|
|
import ccxt
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
import json
|
|
import os
|
|
|
|
from .trading_strategy import TradingStrategy
|
|
from .config import load_config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class Backtester:
|
|
def __init__(self, config_path: str):
|
|
# Load configuration
|
|
self.config = load_config(config_path)
|
|
|
|
# Initialize exchange
|
|
self.exchange = self._initialize_exchange()
|
|
|
|
# Initialize trading strategy
|
|
self.strategy = TradingStrategy(self.config)
|
|
|
|
# Initialize results storage
|
|
self.results = {
|
|
'trades': [],
|
|
'equity_curve': [],
|
|
'performance_metrics': {}
|
|
}
|
|
|
|
def _initialize_exchange(self) -> ccxt.Exchange:
|
|
"""Initialize the cryptocurrency exchange"""
|
|
try:
|
|
exchange_class = getattr(ccxt, self.config['exchange']['name'])
|
|
exchange = exchange_class({
|
|
'apiKey': self.config['exchange']['api_key'],
|
|
'secret': self.config['exchange']['api_secret'],
|
|
'enableRateLimit': True
|
|
})
|
|
|
|
# Test connection
|
|
exchange.load_markets()
|
|
logger.info(f"Successfully connected to {self.config['exchange']['name']}")
|
|
|
|
return exchange
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error initializing exchange: {str(e)}")
|
|
raise
|
|
|
|
def fetch_historical_data(self,
|
|
symbol: str,
|
|
timeframe: str,
|
|
start_date: datetime,
|
|
end_date: datetime) -> pd.DataFrame:
|
|
"""Fetch historical market data"""
|
|
try:
|
|
# Convert dates to timestamps
|
|
start_timestamp = int(start_date.timestamp() * 1000)
|
|
end_timestamp = int(end_date.timestamp() * 1000)
|
|
|
|
# Fetch OHLCV data
|
|
ohlcv = self.exchange.fetch_ohlcv(
|
|
symbol,
|
|
timeframe=timeframe,
|
|
since=start_timestamp,
|
|
limit=1000 # Maximum limit per request
|
|
)
|
|
|
|
# Convert to DataFrame
|
|
df = pd.DataFrame(
|
|
ohlcv,
|
|
columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
|
|
)
|
|
|
|
# Convert timestamp to datetime
|
|
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
|
|
df.set_index('timestamp', inplace=True)
|
|
|
|
return df
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error fetching historical data for {symbol}: {str(e)}")
|
|
return None
|
|
|
|
def run_backtest(self,
|
|
symbol: str,
|
|
timeframe: str,
|
|
start_date: datetime,
|
|
end_date: datetime,
|
|
initial_capital: float = 50.0) -> Dict:
|
|
"""Run backtest for a given period"""
|
|
try:
|
|
logger.info(f"Starting backtest for {symbol} from {start_date} to {end_date}")
|
|
|
|
# Fetch historical data
|
|
df = self.fetch_historical_data(symbol, timeframe, start_date, end_date)
|
|
if df is None:
|
|
raise ValueError("Failed to fetch historical data")
|
|
|
|
# Initialize variables
|
|
current_capital = initial_capital
|
|
position = 0
|
|
entry_price = 0
|
|
trades = []
|
|
equity_curve = []
|
|
|
|
# Iterate through data
|
|
for i in range(len(df)):
|
|
current_data = df.iloc[:i+1]
|
|
current_price = current_data['close'].iloc[-1]
|
|
|
|
# Generate trading signal
|
|
signal = self.strategy.analyze_market(symbol, current_data)
|
|
|
|
if signal:
|
|
# Execute trade if conditions are met
|
|
if signal.action == 'buy' and position <= 0:
|
|
position = 1
|
|
entry_price = current_price
|
|
trades.append({
|
|
'timestamp': current_data.index[-1],
|
|
'action': 'buy',
|
|
'price': current_price,
|
|
'position_size': signal.position_size
|
|
})
|
|
|
|
elif signal.action == 'sell' and position >= 0:
|
|
position = -1
|
|
entry_price = current_price
|
|
trades.append({
|
|
'timestamp': current_data.index[-1],
|
|
'action': 'sell',
|
|
'price': current_price,
|
|
'position_size': signal.position_size
|
|
})
|
|
|
|
# Update position PnL
|
|
if position != 0:
|
|
pnl = position * (current_price - entry_price) * signal.position_size
|
|
current_capital += pnl
|
|
|
|
# Record equity
|
|
equity_curve.append({
|
|
'timestamp': current_data.index[-1],
|
|
'equity': current_capital
|
|
})
|
|
|
|
# Calculate performance metrics
|
|
self.results['trades'] = trades
|
|
self.results['equity_curve'] = equity_curve
|
|
self.results['performance_metrics'] = self._calculate_performance_metrics(trades)
|
|
|
|
return self.results
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error running backtest: {str(e)}")
|
|
return None
|
|
|
|
def _calculate_performance_metrics(self, trades: List[Dict]) -> Dict:
|
|
"""Calculate performance metrics from trades"""
|
|
if not trades:
|
|
return {}
|
|
|
|
# Convert trades to DataFrame
|
|
df = pd.DataFrame(trades)
|
|
|
|
# Calculate basic metrics
|
|
total_trades = len(df)
|
|
winning_trades = len(df[df['action'] == 'buy'])
|
|
losing_trades = len(df[df['action'] == 'sell'])
|
|
win_rate = winning_trades / total_trades if total_trades > 0 else 0
|
|
|
|
# Calculate returns
|
|
returns = df['price'].pct_change()
|
|
total_return = (1 + returns).prod() - 1
|
|
annual_return = (1 + total_return) ** (252 / len(df)) - 1
|
|
|
|
# Calculate risk metrics
|
|
volatility = returns.std() * np.sqrt(252)
|
|
sharpe_ratio = annual_return / volatility if volatility > 0 else 0
|
|
|
|
# Calculate drawdown
|
|
cumulative_returns = (1 + returns).cumprod()
|
|
rolling_max = cumulative_returns.expanding().max()
|
|
drawdowns = (cumulative_returns - rolling_max) / rolling_max
|
|
max_drawdown = drawdowns.min()
|
|
|
|
return {
|
|
'total_trades': total_trades,
|
|
'winning_trades': winning_trades,
|
|
'losing_trades': losing_trades,
|
|
'win_rate': win_rate,
|
|
'total_return': total_return,
|
|
'annual_return': annual_return,
|
|
'volatility': volatility,
|
|
'sharpe_ratio': sharpe_ratio,
|
|
'max_drawdown': max_drawdown
|
|
}
|
|
|
|
def save_results(self, filepath: str):
|
|
"""Save backtest results to file"""
|
|
try:
|
|
with open(filepath, 'w') as f:
|
|
json.dump(self.results, f, default=str)
|
|
logger.info(f"Backtest results saved to {filepath}")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Error saving results: {str(e)}")
|
|
return False
|
|
|
|
def load_results(self, filepath: str):
|
|
"""Load backtest results from file"""
|
|
try:
|
|
with open(filepath, 'r') as f:
|
|
self.results = json.load(f)
|
|
logger.info(f"Backtest results loaded from {filepath}")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Error loading results: {str(e)}")
|
|
return False
|
|
|
|
def main():
|
|
# Load configuration
|
|
config_path = os.path.join(os.path.dirname(__file__), '..', 'config', 'config.json')
|
|
|
|
# Create backtester
|
|
backtester = Backtester(config_path)
|
|
|
|
# Define backtest parameters
|
|
symbol = "BTC/USDT"
|
|
timeframe = "1h"
|
|
start_date = datetime.now() - timedelta(days=30)
|
|
end_date = datetime.now()
|
|
initial_capital = 50.0
|
|
|
|
# Run backtest
|
|
results = backtester.run_backtest(
|
|
symbol=symbol,
|
|
timeframe=timeframe,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
initial_capital=initial_capital
|
|
)
|
|
|
|
if results:
|
|
# Print performance metrics
|
|
metrics = results['performance_metrics']
|
|
print("\nBacktest Results:")
|
|
print(f"Total Trades: {metrics['total_trades']}")
|
|
print(f"Win Rate: {metrics['win_rate']:.2%}")
|
|
print(f"Total Return: {metrics['total_return']:.2%}")
|
|
print(f"Annual Return: {metrics['annual_return']:.2%}")
|
|
print(f"Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")
|
|
print(f"Max Drawdown: {metrics['max_drawdown']:.2%}")
|
|
|
|
# Save results
|
|
backtester.save_results('backtest_results.json')
|
|
else:
|
|
print("Backtest failed to complete")
|
|
|
|
if __name__ == "__main__":
|
|
main() |