gavindiaz/QuantumEdge
1
0
Fork 0
mirror of https://github.com/B-Wear/QuantumEdge.git synced 2026-07-27 15:37:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add files via upload

Browse Source 
Signed-off-by: B-Wear <Bwear008@gmail.com>
This commit is contained in:
B-Wear
2025-03-22 14:20:21 -04:00
committed by GitHub
parent cadbfbde54
commit bf08d59def
14 changed files with 2882 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+133 -2
View File
@@ -1,2 +1,133 @@
# QuantumEdge
QuantumEdge AI-powered trading system for forex and futures starting with $50. Features self-correcting algorithms, technical analysis, and machine learning with strict risk management. Adapts to market changes while preserving capital and gradually scaling positions through an intuitive monitoring dashboard.RetryClaude
# AI-Powered Trading Bot
A sophisticated trading bot that combines technical analysis, machine learning, sentiment analysis, and risk management to make informed trading decisions.
## Features
- **Technical Analysis**
- Multiple timeframe analysis
- Advanced indicators (SMA, EMA, RSI, MACD, Bollinger Bands)
- Pattern recognition
- Support and resistance levels
- **Machine Learning**
- LSTM-based price prediction
- Reinforcement learning for strategy optimization
- Feature engineering and selection
- Model persistence and retraining
- **Sentiment Analysis**
- News sentiment analysis
- Social media sentiment (Twitter)
- Market sentiment indicators
- Weighted sentiment scoring
- **Risk Management**
- Position sizing based on risk percentage
- Stop-loss and take-profit management
- Maximum drawdown protection
- Performance monitoring
- Risk metrics calculation
- **Web Dashboard**
- Real-time performance monitoring
- Trade history visualization
- Risk metrics display
- Configuration management
## Requirements
- Python 3.8+
- CUDA-capable GPU (recommended for machine learning)
- API keys for:
- Cryptocurrency exchange (e.g., Binance)
- News API
- Twitter API
## Installation
1. Clone the repository:
```bash
git clone https://github.com/yourusername/trading-bot.git
cd trading-bot
```
2. Create a virtual environment:
```bash
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
```
3. Install dependencies:
```bash
pip install -r requirements.txt
```
4. Configure the bot:
- Copy `config/config.json.example` to `config/config.json`
- Update the configuration with your API keys and preferences
## Usage
1. Start the trading bot:
```bash
python -m src.main
```
2. Access the web dashboard:
- Open your browser and navigate to `http://localhost:5000`
- Monitor performance and manage settings
3. Monitor logs:
- Check `trading_bot.log` for detailed information
- Monitor system performance and error messages
## Configuration
The bot can be configured through `config/config.json`. Key settings include:
- Trading pairs and timeframes
- Risk management parameters
- Technical analysis settings
- Machine learning model parameters
- Sentiment analysis weights
- API credentials
## Risk Warning
Trading cryptocurrencies involves significant risk of loss. This bot is for educational purposes only. Always:
- Start with small amounts
- Use testnet for initial testing
- Monitor performance closely
- Implement proper risk management
- Never trade with money you cannot afford to lose
## Contributing
1. Fork the repository
2. Create a feature branch
3. Commit your changes
4. Push to the branch
5. Create a Pull Request
## License
This project is licensed under the MIT License - see the LICENSE file for details.
## Acknowledgments
- Thanks to the open-source community for various libraries used in this project
- Special thanks to contributors and maintainers of key dependencies
- Inspired by various trading strategies and research papers
## Support
For support, please:
1. Check the documentation
2. Search existing issues
3. Create a new issue if needed
## Disclaimer
This trading bot is provided as-is, without any warranties. Use at your own risk. The developers are not responsible for any financial losses incurred through the use of this software.
backtesting.py
+266
View File
@@ -0,0 +1,266 @@
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()
config.json
+105
View File
@@ -0,0 +1,105 @@
{
"exchange": {
"name": "binance",
"api_key": "YOUR_API_KEY",
"api_secret": "YOUR_API_SECRET",
"testnet": true
},
"trading": {
"symbols": ["BTC/USDT", "ETH/USDT"],
"timeframes": ["1m", "5m", "15m", "1h", "4h", "1d"],
"lookback_periods": 100,
"state_file": "trading_state.json",
"initial_capital": 50.0,
"max_positions": 3,
"leverage": 1
},
"risk_management": {
"risk_per_trade": 0.02,
"max_daily_loss": 0.05,
"max_drawdown": 0.15,
"stop_loss_pct": 0.02,
"take_profit_pct": 0.04,
"min_win_rate": 0.45,
"min_profit_factor": 1.2
},
"technical_analysis": {
"indicators": {
"sma": [20, 50, 200],
"ema": [20],
"rsi": [14],
"macd": {
"fast": 12,
"slow": 26,
"signal": 9
},
"bollinger_bands": {
"period": 20,
"std_dev": 2
}
},
"patterns": {
"candlestick": true,
"chart": true
}
},
"machine_learning": {
"model_type": "lstm",
"features": [
"open", "high", "low", "close", "volume",
"sma_20", "sma_50", "sma_200",
"rsi_14", "macd", "macd_signal", "macd_hist",
"bb_upper", "bb_middle", "bb_lower"
],
"train_test_split": 0.2,
"validation_split": 0.1,
"rl": {
"learning_rate": 0.0003,
"n_steps": 2048,
"batch_size": 64,
"n_epochs": 10,
"gamma": 0.99,
"gae_lambda": 0.95,
"clip_range": 0.2,
"ent_coef": 0.01,
"vf_coef": 0.5
}
},
"sentiment_analysis": {
"news_api_key": "YOUR_NEWS_API_KEY",
"twitter_api_key": "YOUR_TWITTER_API_KEY",
"twitter_api_secret": "YOUR_TWITTER_API_SECRET",
"twitter_access_token": "YOUR_TWITTER_ACCESS_TOKEN",
"twitter_access_token_secret": "YOUR_TWITTER_ACCESS_TOKEN_SECRET",
"sentiment_weights": {
"news": 0.4,
"twitter": 0.3,
"market": 0.3
},
"sentiment_thresholds": {
"positive": 0.3,
"negative": -0.3
}
},
"signal_weights": {
"technical": 0.4,
"ml": 0.3,
"sentiment": 0.2,
"risk": 0.1
},
"signal_thresholds": {
"buy": 0.6,
"sell": -0.6
},
"logging": {
"level": "INFO",
"file": "trading_bot.log",
"max_size": 10485760,
"backup_count": 5
},
"web_dashboard": {
"host": "localhost",
"port": 5000,
"debug": false
}
}
config.py
+134
View File
@@ -0,0 +1,134 @@
import os
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Exchange Configuration
EXCHANGE_CONFIG = {
'name': 'binance',
'api_key': os.getenv('BINANCE_API_KEY', ''),
'api_secret': os.getenv('BINANCE_API_SECRET', ''),
'testnet': True # Use testnet for development
}
# Trading Parameters
TRADING_CONFIG = {
'symbols': ['BTC/USDT', 'ETH/USDT', 'EUR/USD'],
'timeframes': ['1h', '4h', '1d'],
'initial_capital': 10000,
'risk_per_trade': 0.02, # 2% risk per trade
'max_positions': 3,
'position_sizing': {
'method': 'fixed_fractional',
'fraction': 0.02 # 2% of capital per trade
}
}
# Technical Analysis Parameters
TECHNICAL_CONFIG = {
'indicators': {
'bollinger_bands': {
'period': 20,
'std_dev': 2
},
'rsi': {
'period': 14,
'overbought': 70,
'oversold': 30
},
'macd': {
'fast_period': 12,
'slow_period': 26,
'signal_period': 9
},
'atr': {
'period': 14
}
},
'patterns': {
'head_and_shoulders': True,
'double_top_bottom': True,
'triangles': True
}
}
# Machine Learning Configuration
ML_CONFIG = {
'model_type': 'lstm', # Options: 'lstm', 'rf', 'xgboost'
'features': [
'open', 'high', 'low', 'close', 'volume',
'sma_20', 'sma_50', 'rsi', 'macd', 'atr',
'bb_upper', 'bb_lower', 'bb_width'
],
'sequence_length': 10,
'prediction_horizon': 1,
'train_test_split': 0.8,
'validation_split': 0.1
}
# Reinforcement Learning Configuration
RL_CONFIG = {
'algorithm': 'PPO',
'learning_rate': 0.0003,
'n_steps': 2048,
'batch_size': 64,
'n_epochs': 10,
'gamma': 0.99,
'gae_lambda': 0.95,
'clip_range': 0.2,
'ent_coef': 0.01,
'vf_coef': 0.5
}
# Risk Management Configuration
RISK_CONFIG = {
'stop_loss': {
'method': 'atr',
'atr_multiplier': 2
},
'take_profit': {
'method': 'risk_reward',
'risk_reward_ratio': 2
},
'trailing_stop': {
'enabled': True,
'activation_percentage': 0.02,
'trail_percentage': 0.01
}
}
# Sentiment Analysis Configuration
SENTIMENT_CONFIG = {
'sources': [
'reuters',
'bloomberg',
'forexfactory'
],
'update_interval': 3600, # 1 hour
'weight': 0.2 # Weight in final decision
}
# Backtesting Configuration
BACKTEST_CONFIG = {
'start_date': '2020-01-01',
'end_date': '2023-12-31',
'initial_capital': 10000,
'commission': 0.001, # 0.1%
'slippage': 0.0001 # 0.01%
}
# Logging Configuration
LOGGING_CONFIG = {
'level': 'INFO',
'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
'file': 'logs/trading_bot.log'
}
# Web Dashboard Configuration
DASHBOARD_CONFIG = {
'host': '0.0.0.0',
'port': 5000,
'debug': False,
'update_interval': 5 # seconds
}
machine_learning.py
+271
View File
@@ -0,0 +1,271 @@
import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
from tensorflow.keras.optimizers import Adam
from stable_baselines3 import PPO
from stable_baselines3.common.vec_env import DummyVecEnv
import gym
import logging
from typing import Dict, List, Tuple, Optional
import joblib
import os
logger = logging.getLogger(__name__)
class MachineLearningModel:
def __init__(self, config: Dict):
self.config = config
self.model_type = config['model_type']
self.features = config['features']
self.scaler = StandardScaler()
self.model = None
self.env = None
def prepare_data(self, df: pd.DataFrame) -> Tuple[np.ndarray, np.ndarray]:
"""
Prepare data for machine learning
"""
# Select features
X = df[self.features].copy()
# Create target variable (price movement direction)
y = np.where(df['close'].shift(-1) > df['close'], 1, 0)
# Drop NaN values
valid_idx = ~np.isnan(y)
X = X[valid_idx]
y = y[valid_idx]
# Scale features
X_scaled = self.scaler.fit_transform(X)
return X_scaled, y
def train_lstm(self, X: np.ndarray, y: np.ndarray):
"""
Train LSTM model
"""
# Reshape data for LSTM [samples, time steps, features]
X_reshaped = np.reshape(X, (X.shape[0], 1, X.shape[1]))
# Split data
X_train, X_test, y_train, y_test = train_test_split(
X_reshaped, y,
test_size=self.config['train_test_split'],
random_state=42
)
# Create model
model = Sequential([
LSTM(50, activation='relu', input_shape=(1, X.shape[1]), return_sequences=True),
Dropout(0.2),
LSTM(50, activation='relu'),
Dropout(0.2),
Dense(1, activation='sigmoid')
])
# Compile model
model.compile(
optimizer=Adam(learning_rate=0.001),
loss='binary_crossentropy',
metrics=['accuracy']
)
# Train model
model.fit(
X_train, y_train,
epochs=50,
batch_size=32,
validation_split=self.config['validation_split'],
verbose=0
)
# Evaluate model
_, accuracy = model.evaluate(X_test, y_test, verbose=0)
logger.info(f"LSTM model accuracy: {accuracy:.4f}")
self.model = model
return model
def train_reinforcement_learning(self, df: pd.DataFrame):
"""
Train reinforcement learning model
"""
# Create custom trading environment
self.env = TradingEnvironment(df)
self.env = DummyVecEnv([lambda: self.env])
# Create and train PPO model
model = PPO(
"MlpPolicy",
self.env,
learning_rate=self.config['rl']['learning_rate'],
n_steps=self.config['rl']['n_steps'],
batch_size=self.config['rl']['batch_size'],
n_epochs=self.config['rl']['n_epochs'],
gamma=self.config['rl']['gamma'],
gae_lambda=self.config['rl']['gae_lambda'],
clip_range=self.config['rl']['clip_range'],
ent_coef=self.config['rl']['ent_coef'],
vf_coef=self.config['rl']['vf_coef'],
verbose=1
)
# Train model
model.learn(total_timesteps=10000)
self.model = model
return model
def predict(self, X: np.ndarray) -> np.ndarray:
"""
Make predictions using the trained model
"""
if self.model is None:
logger.error("Model not trained. Call train() first.")
return None
if self.model_type == 'lstm':
X_reshaped = np.reshape(X, (X.shape[0], 1, X.shape[1]))
predictions = self.model.predict(X_reshaped)
else: # reinforcement learning
predictions = self.model.predict(X)
return predictions
def save_model(self, filepath: str):
"""
Save the trained model
"""
if self.model is None:
logger.error("Model not trained. Call train() first.")
return None
try:
if self.model_type == 'lstm':
self.model.save(filepath)
else:
self.model.save(f"{filepath}_rl")
# Save scaler separately
joblib.dump(self.scaler, f"{filepath}_scaler")
logger.info(f"Model saved to {filepath}")
return True
except Exception as e:
logger.error(f"Error saving model: {str(e)}")
return False
def load_model(self, filepath: str):
"""
Load a trained model
"""
try:
if self.model_type == 'lstm':
self.model = tf.keras.models.load_model(filepath)
else:
self.model = PPO.load(f"{filepath}_rl")
# Load scaler
self.scaler = joblib.load(f"{filepath}_scaler")
logger.info(f"Model loaded from {filepath}")
return True
except Exception as e:
logger.error(f"Error loading model: {str(e)}")
return False
class TradingEnvironment(gym.Env):
"""
Custom trading environment for reinforcement learning
"""
def __init__(self, df: pd.DataFrame):
super(TradingEnvironment, self).__init__()
self.df = df
self.current_step = 0
self.initial_balance = 10000
self.balance = self.initial_balance
self.position = 0
self.position_size = 0
# Define action and observation space
self.action_space = gym.spaces.Discrete(3) # Buy, Sell, Hold
self.observation_space = gym.spaces.Box(
low=-np.inf,
high=np.inf,
shape=(len(df.columns),),
dtype=np.float32
)
def reset(self):
"""
Reset the environment
"""
self.current_step = 0
self.balance = self.initial_balance
self.position = 0
self.position_size = 0
return self._get_observation()
def step(self, action):
"""
Execute one step in the environment
"""
# Get current price
current_price = self.df['close'].iloc[self.current_step]
# Execute action
reward = 0
done = False
if action == 0: # Buy
if self.position <= 0:
self.position = 1
self.position_size = self.balance / current_price
elif action == 1: # Sell
if self.position >= 0:
self.position = -1
self.position_size = self.balance / current_price
# Calculate reward
next_price = self.df['close'].iloc[self.current_step + 1]
price_change = (next_price - current_price) / current_price
if self.position != 0:
reward = self.position * price_change * 100
# Update balance
self.balance *= (1 + reward/100)
# Move to next step
self.current_step += 1
# Check if episode is done
if self.current_step >= len(self.df) - 1:
done = True
return self._get_observation(), reward, done, {}
def _get_observation(self):
"""
Get current observation
"""
return self.df.iloc[self.current_step].values.astype(np.float32)
def render(self):
"""
Render the environment
"""
pass
def close(self):
"""
Clean up resources
"""
pass
main.py
+272
View File
@@ -0,0 +1,272 @@
import os
import sys
import logging
import json
import time
from datetime import datetime
from typing import Dict, List
import pandas as pd
import ccxt
from concurrent.futures import ThreadPoolExecutor
import schedule
import threading
import signal
import queue
from .trading_strategy import TradingStrategy
from .config import load_config
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('trading_bot.log'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
class TradingBot:
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 state
self.is_running = False
self.symbols = self.config['trading']['symbols']
self.timeframes = self.config['trading']['timeframes']
self.data_queue = queue.Queue()
self.signal_queue = queue.Queue()
# Load previous state if exists
self._load_state()
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 _load_state(self):
"""
Load previous trading state
"""
state_file = self.config['trading']['state_file']
if os.path.exists(state_file):
try:
self.strategy.load_state(state_file)
logger.info("Successfully loaded previous trading state")
except Exception as e:
logger.error(f"Error loading state: {str(e)}")
def _save_state(self):
"""
Save current trading state
"""
try:
self.strategy.save_state(self.config['trading']['state_file'])
logger.info("Successfully saved trading state")
except Exception as e:
logger.error(f"Error saving state: {str(e)}")
def fetch_market_data(self, symbol: str, timeframe: str) -> pd.DataFrame:
"""
Fetch market data from exchange
"""
try:
# Get OHLCV data
ohlcv = self.exchange.fetch_ohlcv(
symbol,
timeframe=timeframe,
limit=self.config['trading']['lookback_periods']
)
# 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 market data for {symbol}: {str(e)}")
return None
def process_market_data(self):
"""
Process market data from queue
"""
while self.is_running:
try:
# Get data from queue
data = self.data_queue.get(timeout=1)
symbol, timeframe, df = data
# Generate trading signals
signal = self.strategy.analyze_market(symbol, df)
if signal:
# Add signal to queue
self.signal_queue.put(signal)
except queue.Empty:
continue
except Exception as e:
logger.error(f"Error processing market data: {str(e)}")
def execute_signals(self):
"""
Execute trading signals from queue
"""
while self.is_running:
try:
# Get signal from queue
signal = self.signal_queue.get(timeout=1)
# Execute trade
if self.strategy.execute_trade(signal):
logger.info(f"Executed {signal.action} trade for {signal.symbol}")
# Update positions with current prices
current_prices = {
symbol: self.exchange.fetch_ticker(symbol)['last']
for symbol in self.symbols
}
self.strategy.update_positions(current_prices)
# Save state after trade
self._save_state()
except queue.Empty:
continue
except Exception as e:
logger.error(f"Error executing signals: {str(e)}")
def update_market_data(self):
"""
Update market data for all symbols
"""
with ThreadPoolExecutor(max_workers=len(self.symbols)) as executor:
futures = []
for symbol in self.symbols:
for timeframe in self.timeframes:
future = executor.submit(self.fetch_market_data, symbol, timeframe)
futures.append((symbol, timeframe, future))
for symbol, timeframe, future in futures:
try:
df = future.result()
if df is not None:
self.data_queue.put((symbol, timeframe, df))
except Exception as e:
logger.error(f"Error updating market data for {symbol}: {str(e)}")
def schedule_tasks(self):
"""
Schedule periodic tasks
"""
# Schedule market data updates
for timeframe in self.timeframes:
schedule.every().minute.at(":00").do(self.update_market_data)
# Schedule state saving
schedule.every().hour.at(":00").do(self._save_state)
# Run the scheduler
while self.is_running:
schedule.run_pending()
time.sleep(1)
def start(self):
"""
Start the trading bot
"""
try:
self.is_running = True
# Start threads
threads = [
threading.Thread(target=self.process_market_data),
threading.Thread(target=self.execute_signals),
threading.Thread(target=self.schedule_tasks)
]
for thread in threads:
thread.daemon = True
thread.start()
# Initial market data update
self.update_market_data()
logger.info("Trading bot started successfully")
# Wait for shutdown signal
while self.is_running:
time.sleep(1)
except Exception as e:
logger.error(f"Error starting trading bot: {str(e)}")
self.stop()
def stop(self):
"""
Stop the trading bot
"""
try:
self.is_running = False
self._save_state()
logger.info("Trading bot stopped successfully")
except Exception as e:
logger.error(f"Error stopping trading bot: {str(e)}")
def main():
# Load configuration
config_path = os.path.join(os.path.dirname(__file__), '..', 'config', 'config.json')
# Create trading bot
bot = TradingBot(config_path)
# Handle shutdown signals
def signal_handler(signum, frame):
logger.info("Received shutdown signal")
bot.stop()
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
# Start trading bot
bot.start()
if __name__ == "__main__":
main()
requirements.txt
+18
View File
@@ -0,0 +1,18 @@
numpy>=1.21.0
pandas>=1.3.0
ccxt>=2.0.0
scikit-learn>=0.24.2
tensorflow>=2.6.0
stable-baselines3>=1.5.0
gym>=0.21.0
ta-lib>=0.4.24
textblob>=0.15.3
tweepy>=4.10.0
newsapi-python>=0.2.6
yfinance>=0.1.63
schedule>=1.1.0
joblib>=1.0.2
python-dotenv>=0.19.0
flask>=2.0.1
plotly>=5.3.1
dash>=2.0.0
risk_management.py
+249
View File
@@ -0,0 +1,249 @@
import numpy as np
import pandas as pd
from typing import Dict, Tuple, Optional
import logging
from dataclasses import dataclass
from datetime import datetime
logger = logging.getLogger(__name__)
@dataclass
class RiskMetrics:
"""Data class to store risk metrics"""
daily_drawdown: float
max_drawdown: float
sharpe_ratio: float
sortino_ratio: float
win_rate: float
profit_factor: float
avg_win: float
avg_loss: float
max_consecutive_losses: int
max_consecutive_wins: int
class RiskManager:
def __init__(self, config: Dict):
self.config = config
self.initial_capital = config['initial_capital']
self.current_capital = self.initial_capital
self.risk_per_trade = config['risk_per_trade']
self.max_daily_loss = config['max_daily_loss']
self.max_drawdown = config['max_drawdown']
self.max_positions = config['max_positions']
self.leverage = config['leverage']
self.stop_loss_pct = config['stop_loss_pct']
self.take_profit_pct = config['take_profit_pct']
# Performance tracking
self.trades_history = []
self.daily_pnl = []
self.positions = {}
self.risk_metrics = None
def calculate_position_size(self,
price: float,
stop_loss: float,
account_balance: float) -> Tuple[float, float]:
"""
Calculate position size based on risk parameters
"""
# Calculate risk amount in base currency
risk_amount = account_balance * self.risk_per_trade
# Calculate position size based on stop loss
price_distance = abs(price - stop_loss)
position_size = risk_amount / price_distance
# Apply leverage
position_size *= self.leverage
# Calculate required margin
required_margin = (position_size * price) / self.leverage
# Ensure we don't exceed maximum position size
max_position = account_balance * self.leverage / price
position_size = min(position_size, max_position)
return position_size, required_margin
def validate_trade(self,
symbol: str,
position_size: float,
price: float,
stop_loss: float,
take_profit: float) -> bool:
"""
Validate if a trade meets risk management criteria
"""
# Check if we have too many open positions
if len(self.positions) >= self.max_positions:
logger.warning("Maximum number of positions reached")
return False
# Check if symbol is already in a position
if symbol in self.positions:
logger.warning(f"Position already exists for {symbol}")
return False
# Calculate potential loss
potential_loss = position_size * abs(price - stop_loss)
# Check if potential loss exceeds daily limit
if potential_loss > self.max_daily_loss:
logger.warning("Trade exceeds maximum daily loss limit")
return False
# Check if stop loss is too far
stop_loss_distance = abs(price - stop_loss) / price
if stop_loss_distance > self.stop_loss_pct:
logger.warning("Stop loss distance exceeds maximum allowed")
return False
# Check if take profit is reasonable
take_profit_distance = abs(take_profit - price) / price
if take_profit_distance > self.take_profit_pct:
logger.warning("Take profit distance exceeds maximum allowed")
return False
return True
def update_position(self,
symbol: str,
entry_price: float,
current_price: float,
position_size: float,
position_type: str) -> Optional[float]:
"""
Update position and check for stop loss or take profit
"""
if symbol not in self.positions:
self.positions[symbol] = {
'entry_price': entry_price,
'current_price': current_price,
'position_size': position_size,
'position_type': position_type,
'entry_time': datetime.now()
}
return None
position = self.positions[symbol]
pnl = 0
# Calculate unrealized PnL
if position_type == 'long':
pnl = (current_price - entry_price) * position_size
else:
pnl = (entry_price - current_price) * position_size
# Check stop loss
if position_type == 'long':
if current_price <= entry_price * (1 - self.stop_loss_pct):
pnl = -position_size * entry_price * self.stop_loss_pct
del self.positions[symbol]
return pnl
else:
if current_price >= entry_price * (1 + self.stop_loss_pct):
pnl = -position_size * entry_price * self.stop_loss_pct
del self.positions[symbol]
return pnl
# Check take profit
if position_type == 'long':
if current_price >= entry_price * (1 + self.take_profit_pct):
pnl = position_size * entry_price * self.take_profit_pct
del self.positions[symbol]
return pnl
else:
if current_price <= entry_price * (1 - self.take_profit_pct):
pnl = position_size * entry_price * self.take_profit_pct
del self.positions[symbol]
return pnl
# Update current price
position['current_price'] = current_price
return None
def calculate_risk_metrics(self) -> RiskMetrics:
"""
Calculate various risk metrics
"""
if not self.trades_history:
return None
# Convert trades history to DataFrame
df = pd.DataFrame(self.trades_history)
# Calculate daily returns
daily_returns = df.groupby(df['exit_time'].dt.date)['pnl'].sum()
# Calculate drawdown
cumulative_returns = (1 + daily_returns).cumprod()
rolling_max = cumulative_returns.expanding().max()
drawdowns = (cumulative_returns - rolling_max) / rolling_max
# Calculate metrics
daily_drawdown = drawdowns.min()
max_drawdown = drawdowns.min()
# Calculate Sharpe ratio
risk_free_rate = 0.02 # 2% annual risk-free rate
excess_returns = daily_returns - risk_free_rate/252
sharpe_ratio = np.sqrt(252) * excess_returns.mean() / excess_returns.std()
# Calculate Sortino ratio
downside_returns = excess_returns[excess_returns < 0]
sortino_ratio = np.sqrt(252) * excess_returns.mean() / downside_returns.std()
# Calculate win rate and profit factor
winning_trades = df[df['pnl'] > 0]
losing_trades = df[df['pnl'] < 0]
win_rate = len(winning_trades) / len(df)
profit_factor = abs(winning_trades['pnl'].sum() / losing_trades['pnl'].sum())
# Calculate average win and loss
avg_win = winning_trades['pnl'].mean()
avg_loss = losing_trades['pnl'].mean()
# Calculate consecutive wins and losses
df['consecutive'] = (df['pnl'] > 0).astype(int)
max_consecutive_wins = df['consecutive'].groupby((df['consecutive'] != df['consecutive'].shift()).cumsum()).sum().max()
max_consecutive_losses = df['consecutive'].groupby((df['consecutive'] != df['consecutive'].shift()).cumsum()).sum().min()
self.risk_metrics = RiskMetrics(
daily_drawdown=daily_drawdown,
max_drawdown=max_drawdown,
sharpe_ratio=sharpe_ratio,
sortino_ratio=sortino_ratio,
win_rate=win_rate,
profit_factor=profit_factor,
avg_win=avg_win,
avg_loss=avg_loss,
max_consecutive_losses=max_consecutive_losses,
max_consecutive_wins=max_consecutive_wins
)
return self.risk_metrics
def should_stop_trading(self) -> bool:
"""
Check if trading should be stopped based on risk metrics
"""
if not self.risk_metrics:
return False
# Stop if maximum drawdown is exceeded
if self.risk_metrics.max_drawdown < -self.max_drawdown:
logger.warning("Maximum drawdown exceeded. Stopping trading.")
return True
# Stop if win rate is too low
if self.risk_metrics.win_rate < self.config['min_win_rate']:
logger.warning("Win rate below minimum threshold. Stopping trading.")
return True
# Stop if profit factor is too low
if self.risk_metrics.profit_factor < self.config['min_profit_factor']:
logger.warning("Profit factor below minimum threshold. Stopping trading.")
return True
return False
run_backtest.py
+84
View File
@@ -0,0 +1,84 @@
import os
import sys
from datetime import datetime, timedelta
import logging
from src.backtesting import Backtester
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('backtest.log'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
def run_backtest_scenarios():
# Load configuration
config_path = os.path.join(os.path.dirname(__file__), 'config', 'config.json')
# Create backtester
backtester = Backtester(config_path)
# Define test scenarios
scenarios = [
{
'name': 'BTC/USDT 1h - Last 30 days',
'symbol': 'BTC/USDT',
'timeframe': '1h',
'days': 30
},
{
'name': 'ETH/USDT 4h - Last 60 days',
'symbol': 'ETH/USDT',
'timeframe': '4h',
'days': 60
},
{
'name': 'BTC/USDT 1d - Last 90 days',
'symbol': 'BTC/USDT',
'timeframe': '1d',
'days': 90
}
]
# Run each scenario
for scenario in scenarios:
logger.info(f"\nRunning scenario: {scenario['name']}")
# Calculate date range
end_date = datetime.now()
start_date = end_date - timedelta(days=scenario['days'])
# Run backtest
results = backtester.run_backtest(
symbol=scenario['symbol'],
timeframe=scenario['timeframe'],
start_date=start_date,
end_date=end_date,
initial_capital=50.0
)
if results:
# Print performance metrics
metrics = results['performance_metrics']
print(f"\nResults for {scenario['name']}:")
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
filename = f"backtest_results_{scenario['symbol'].replace('/', '_')}_{scenario['timeframe']}.json"
backtester.save_results(filename)
logger.info(f"Results saved to {filename}")
else:
logger.error(f"Backtest failed for scenario: {scenario['name']}")
if __name__ == "__main__":
run_backtest_scenarios()
sentiment_analysis.py
+267
View File
@@ -0,0 +1,267 @@
import numpy as np
import pandas as pd
from typing import Dict, List, Optional
import logging
from datetime import datetime, timedelta
import requests
from textblob import TextBlob
import tweepy
from newsapi import NewsApiClient
import yfinance as yf
from concurrent.futures import ThreadPoolExecutor
import json
import os
logger = logging.getLogger(__name__)
class SentimentAnalyzer:
def __init__(self, config: Dict):
self.config = config
self.news_api = NewsApiClient(api_key=config['news_api_key'])
self.twitter_auth = tweepy.OAuthHandler(
config['twitter_api_key'],
config['twitter_api_secret']
)
self.twitter_auth.set_access_token(
config['twitter_access_token'],
config['twitter_access_token_secret']
)
self.twitter_api = tweepy.API(self.twitter_auth)
# Initialize sentiment cache
self.sentiment_cache = {}
self.cache_duration = timedelta(hours=1)
def analyze_text(self, text: str) -> float:
"""
Analyze sentiment of a single text using TextBlob
"""
try:
analysis = TextBlob(text)
# Normalize sentiment score to [-1, 1]
return analysis.sentiment.polarity
except Exception as e:
logger.error(f"Error analyzing text: {str(e)}")
return 0.0
def get_news_sentiment(self, symbol: str) -> Optional[float]:
"""
Get sentiment from news articles
"""
try:
# Check cache first
cache_key = f"news_{symbol}"
if cache_key in self.sentiment_cache:
cached_data = self.sentiment_cache[cache_key]
if datetime.now() - cached_data['timestamp'] < self.cache_duration:
return cached_data['sentiment']
# Get news articles
news = self.news_api.get_everything(
q=symbol,
language='en',
from_param=(datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d'),
sort_by='relevancy'
)
if not news['articles']:
return None
# Analyze sentiment of each article
sentiments = []
for article in news['articles']:
title_sentiment = self.analyze_text(article['title'])
if article['description']:
desc_sentiment = self.analyze_text(article['description'])
sentiments.append((title_sentiment + desc_sentiment) / 2)
else:
sentiments.append(title_sentiment)
# Calculate weighted average sentiment
avg_sentiment = np.mean(sentiments)
# Cache the result
self.sentiment_cache[cache_key] = {
'sentiment': avg_sentiment,
'timestamp': datetime.now()
}
return avg_sentiment
except Exception as e:
logger.error(f"Error getting news sentiment: {str(e)}")
return None
def get_twitter_sentiment(self, symbol: str) -> Optional[float]:
"""
Get sentiment from Twitter
"""
try:
# Check cache first
cache_key = f"twitter_{symbol}"
if cache_key in self.sentiment_cache:
cached_data = self.sentiment_cache[cache_key]
if datetime.now() - cached_data['timestamp'] < self.cache_duration:
return cached_data['sentiment']
# Get tweets
tweets = self.twitter_api.search_tweets(
q=f"${symbol}",
lang="en",
count=100
)
if not tweets:
return None
# Analyze sentiment of each tweet
sentiments = []
for tweet in tweets:
sentiment = self.analyze_text(tweet.text)
sentiments.append(sentiment)
# Calculate weighted average sentiment
avg_sentiment = np.mean(sentiments)
# Cache the result
self.sentiment_cache[cache_key] = {
'sentiment': avg_sentiment,
'timestamp': datetime.now()
}
return avg_sentiment
except Exception as e:
logger.error(f"Error getting Twitter sentiment: {str(e)}")
return None
def get_market_sentiment(self, symbol: str) -> Optional[float]:
"""
Get market sentiment indicators
"""
try:
# Check cache first
cache_key = f"market_{symbol}"
if cache_key in self.sentiment_cache:
cached_data = self.sentiment_cache[cache_key]
if datetime.now() - cached_data['timestamp'] < self.cache_duration:
return cached_data['sentiment']
# Get market data
ticker = yf.Ticker(symbol)
info = ticker.info
# Calculate various sentiment indicators
sentiment_indicators = []
# RSI sentiment
if 'RSI' in info:
rsi = info['RSI']
rsi_sentiment = (rsi - 50) / 50 # Normalize to [-1, 1]
sentiment_indicators.append(rsi_sentiment)
# Volume sentiment
if 'volume' in info and 'averageVolume' in info:
volume_ratio = info['volume'] / info['averageVolume']
volume_sentiment = (volume_ratio - 1) / volume_ratio # Normalize to [-1, 1]
sentiment_indicators.append(volume_sentiment)
# Price momentum sentiment
if 'regularMarketChangePercent' in info:
momentum_sentiment = info['regularMarketChangePercent'] / 100
sentiment_indicators.append(momentum_sentiment)
if not sentiment_indicators:
return None
# Calculate weighted average sentiment
avg_sentiment = np.mean(sentiment_indicators)
# Cache the result
self.sentiment_cache[cache_key] = {
'sentiment': avg_sentiment,
'timestamp': datetime.now()
}
return avg_sentiment
except Exception as e:
logger.error(f"Error getting market sentiment: {str(e)}")
return None
def get_combined_sentiment(self, symbol: str) -> Optional[float]:
"""
Get combined sentiment from all sources
"""
try:
# Check cache first
cache_key = f"combined_{symbol}"
if cache_key in self.sentiment_cache:
cached_data = self.sentiment_cache[cache_key]
if datetime.now() - cached_data['timestamp'] < self.cache_duration:
return cached_data['sentiment']
# Get sentiment from all sources
sentiments = []
weights = []
# News sentiment
news_sentiment = self.get_news_sentiment(symbol)
if news_sentiment is not None:
sentiments.append(news_sentiment)
weights.append(self.config['sentiment_weights']['news'])
# Twitter sentiment
twitter_sentiment = self.get_twitter_sentiment(symbol)
if twitter_sentiment is not None:
sentiments.append(twitter_sentiment)
weights.append(self.config['sentiment_weights']['twitter'])
# Market sentiment
market_sentiment = self.get_market_sentiment(symbol)
if market_sentiment is not None:
sentiments.append(market_sentiment)
weights.append(self.config['sentiment_weights']['market'])
if not sentiments:
return None
# Calculate weighted average sentiment
avg_sentiment = np.average(sentiments, weights=weights)
# Cache the result
self.sentiment_cache[cache_key] = {
'sentiment': avg_sentiment,
'timestamp': datetime.now()
}
return avg_sentiment
except Exception as e:
logger.error(f"Error getting combined sentiment: {str(e)}")
return None
def clear_cache(self):
"""
Clear the sentiment cache
"""
self.sentiment_cache.clear()
def get_sentiment_signal(self, symbol: str) -> int:
"""
Convert sentiment to trading signal
"""
sentiment = self.get_combined_sentiment(symbol)
if sentiment is None:
return 0
# Define sentiment thresholds
positive_threshold = self.config['sentiment_thresholds']['positive']
negative_threshold = self.config['sentiment_thresholds']['negative']
if sentiment > positive_threshold:
return 1 # Buy signal
elif sentiment < negative_threshold:
return -1 # Sell signal
else:
return 0 # Neutral signal
technical_analysis.py
+293
View File
@@ -0,0 +1,293 @@
import numpy as np
import pandas as pd
import talib
from typing import Dict, List, Tuple, Optional
import logging
logger = logging.getLogger(__name__)
class TechnicalAnalyzer:
def __init__(self, config: Dict):
self.config = config
self.indicators = {}
self.patterns = {}
def add_indicators(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Add technical indicators to the dataframe
"""
df = df.copy()
# Trend Indicators
df['sma_20'] = talib.SMA(df['close'], timeperiod=20)
df['sma_50'] = talib.SMA(df['close'], timeperiod=50)
df['sma_200'] = talib.SMA(df['close'], timeperiod=200)
df['ema_20'] = talib.EMA(df['close'], timeperiod=20)
# Volatility Indicators
df['upperband'], df['middleband'], df['lowerband'] = talib.BBANDS(
df['close'],
timeperiod=self.config['indicators']['bollinger_bands']['period'],
nbdevup=self.config['indicators']['bollinger_bands']['std_dev'],
nbdevdn=self.config['indicators']['bollinger_bands']['std_dev']
)
df['atr'] = talib.ATR(
df['high'],
df['low'],
df['close'],
timeperiod=self.config['indicators']['atr']['period']
)
# Momentum Indicators
df['rsi'] = talib.RSI(
df['close'],
timeperiod=self.config['indicators']['rsi']['period']
)
df['macd'], df['macdsignal'], df['macdhist'] = talib.MACD(
df['close'],
fastperiod=self.config['indicators']['macd']['fast_period'],
slowperiod=self.config['indicators']['macd']['slow_period'],
signalperiod=self.config['indicators']['macd']['signal_period']
)
# Volume Indicators
df['obv'] = talib.OBV(df['close'], df['volume'])
# Additional Indicators
df['adx'] = talib.ADX(df['high'], df['low'], df['close'], timeperiod=14)
df['cci'] = talib.CCI(df['high'], df['low'], df['close'], timeperiod=14)
df['stoch_k'], df['stoch_d'] = talib.STOCH(
df['high'],
df['low'],
df['close'],
fastk_period=14,
slowk_period=3,
slowd_period=3
)
# Store indicators for reference
self.indicators = {
'trend': ['sma_20', 'sma_50', 'sma_200', 'ema_20'],
'volatility': ['upperband', 'middleband', 'lowerband', 'atr'],
'momentum': ['rsi', 'macd', 'macdsignal', 'macdhist'],
'volume': ['obv'],
'additional': ['adx', 'cci', 'stoch_k', 'stoch_d']
}
return df
def detect_patterns(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Detect candlestick patterns and chart patterns
"""
df = df.copy()
# Candlestick Patterns
if self.config['patterns']['candlestick']:
df['doji'] = talib.CDLDOJI(df['open'], df['high'], df['low'], df['close'])
df['hammer'] = talib.CDLHAMMER(df['open'], df['high'], df['low'], df['close'])
df['engulfing'] = talib.CDLENGULFING(df['open'], df['high'], df['low'], df['close'])
df['morning_star'] = talib.CDLMORNINGSTAR(df['open'], df['high'], df['low'], df['close'])
df['evening_star'] = talib.CDLEVENINGSTAR(df['open'], df['high'], df['low'], df['close'])
# Chart Patterns
if self.config['patterns']['chart']:
df['head_and_shoulders'] = self._detect_head_and_shoulders(df)
df['double_top'] = self._detect_double_top(df)
df['double_bottom'] = self._detect_double_bottom(df)
df['triangle'] = self._detect_triangle(df)
return df
def _detect_head_and_shoulders(self, df: pd.DataFrame) -> pd.Series:
"""
Detect head and shoulders pattern
"""
pattern = pd.Series(0, index=df.index)
for i in range(20, len(df) - 20):
# Find potential left shoulder
left_shoulder = df['high'].iloc[i-20:i].max()
left_shoulder_idx = df['high'].iloc[i-20:i].idxmax()
# Find potential head
head = df['high'].iloc[i-10:i+10].max()
head_idx = df['high'].iloc[i-10:i+10].idxmax()
# Find potential right shoulder
right_shoulder = df['high'].iloc[i:i+20].max()
right_shoulder_idx = df['high'].iloc[i:i+20].idxmax()
# Find neckline
neckline = min(df['low'].iloc[left_shoulder_idx:right_shoulder_idx])
# Check pattern conditions
if (left_shoulder < head and right_shoulder < head and
abs(left_shoulder - right_shoulder) / head < 0.1):
pattern.iloc[i] = 1
return pattern
def _detect_double_top(self, df: pd.DataFrame) -> pd.Series:
"""
Detect double top pattern
"""
pattern = pd.Series(0, index=df.index)
for i in range(20, len(df) - 20):
# Find potential first peak
first_peak = df['high'].iloc[i-20:i].max()
first_peak_idx = df['high'].iloc[i-20:i].idxmax()
# Find potential second peak
second_peak = df['high'].iloc[i:i+20].max()
second_peak_idx = df['high'].iloc[i:i+20].idxmax()
# Find valley between peaks
valley = df['low'].iloc[first_peak_idx:second_peak_idx].min()
# Check pattern conditions
if (abs(first_peak - second_peak) / first_peak < 0.02 and
(first_peak - valley) / first_peak > 0.02):
pattern.iloc[i] = 1
return pattern
def _detect_double_bottom(self, df: pd.DataFrame) -> pd.Series:
"""
Detect double bottom pattern
"""
pattern = pd.Series(0, index=df.index)
for i in range(20, len(df) - 20):
# Find potential first bottom
first_bottom = df['low'].iloc[i-20:i].min()
first_bottom_idx = df['low'].iloc[i-20:i].idxmin()
# Find potential second bottom
second_bottom = df['low'].iloc[i:i+20].min()
second_bottom_idx = df['low'].iloc[i:i+20].idxmin()
# Find peak between bottoms
peak = df['high'].iloc[first_bottom_idx:second_bottom_idx].max()
# Check pattern conditions
if (abs(first_bottom - second_bottom) / first_bottom < 0.02 and
(peak - first_bottom) / first_bottom > 0.02):
pattern.iloc[i] = 1
return pattern
def _detect_triangle(self, df: pd.DataFrame) -> pd.Series:
"""
Detect triangle patterns (ascending, descending, symmetrical)
"""
pattern = pd.Series(0, index=df.index)
for i in range(20, len(df) - 20):
# Get highs and lows for the period
highs = df['high'].iloc[i-20:i]
lows = df['low'].iloc[i-20:i]
# Calculate trend lines
high_slope = np.polyfit(range(len(highs)), highs, 1)[0]
low_slope = np.polyfit(range(len(lows)), lows, 1)[0]
# Classify triangle type
if abs(high_slope) < 0.001 and low_slope > 0.001:
pattern.iloc[i] = 1 # Ascending triangle
elif high_slope < -0.001 and abs(low_slope) < 0.001:
pattern.iloc[i] = 2 # Descending triangle
elif abs(high_slope + low_slope) < 0.001:
pattern.iloc[i] = 3 # Symmetrical triangle
return pattern
def generate_signals(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Generate trading signals based on technical indicators and patterns
"""
df = df.copy()
# Initialize signal column
df['signal'] = 0
# RSI signals
df.loc[df['rsi'] < self.config['indicators']['rsi']['oversold'], 'signal'] += 1
df.loc[df['rsi'] > self.config['indicators']['rsi']['overbought'], 'signal'] -= 1
# MACD signals
df.loc[df['macd'] > df['macdsignal'], 'signal'] += 1
df.loc[df['macd'] < df['macdsignal'], 'signal'] -= 1
# Bollinger Bands signals
df.loc[df['close'] < df['lowerband'], 'signal'] += 1
df.loc[df['close'] > df['upperband'], 'signal'] -= 1
# Trend signals
df.loc[df['close'] > df['sma_20'], 'signal'] += 1
df.loc[df['close'] < df['sma_20'], 'signal'] -= 1
# Pattern signals
if 'head_and_shoulders' in df.columns:
df.loc[df['head_and_shoulders'] == 1, 'signal'] -= 1
if 'double_top' in df.columns:
df.loc[df['double_top'] == 1, 'signal'] -= 1
if 'double_bottom' in df.columns:
df.loc[df['double_bottom'] == 1, 'signal'] += 1
if 'triangle' in df.columns:
df.loc[df['triangle'] == 1, 'signal'] += 1 # Ascending
df.loc[df['triangle'] == 2, 'signal'] -= 1 # Descending
# Normalize signals to -1, 0, 1
df['signal'] = df['signal'].apply(lambda x: 1 if x > 2 else (-1 if x < -2 else 0))
return df
def calculate_support_resistance(self, df: pd.DataFrame, window: int = 20) -> Tuple[pd.Series, pd.Series]:
"""
Calculate support and resistance levels
"""
support = df['low'].rolling(window=window).min()
resistance = df['high'].rolling(window=window).max()
return support, resistance
def calculate_volatility(self, df: pd.DataFrame) -> pd.Series:
"""
Calculate various volatility measures
"""
# ATR-based volatility
atr_volatility = df['atr'] / df['close']
# Bollinger Band width
bb_width = (df['upperband'] - df['lowerband']) / df['middleband']
# Historical volatility
returns = df['close'].pct_change()
hist_volatility = returns.rolling(window=20).std()
return pd.DataFrame({
'atr_volatility': atr_volatility,
'bb_width': bb_width,
'hist_volatility': hist_volatility
})
def get_market_regime(self, df: pd.DataFrame) -> pd.Series:
"""
Determine market regime (trending, ranging, volatile)
"""
regime = pd.Series('unknown', index=df.index)
# Calculate ADX for trend strength
adx = df['adx']
# Calculate volatility
volatility = self.calculate_volatility(df)['atr_volatility']
# Determine regime
regime.loc[adx > 25] = 'trending'
regime.loc[(adx <= 25) & (volatility > volatility.rolling(window=20).mean())] = 'volatile'
regime.loc[(adx <= 25) & (volatility <= volatility.rolling(window=20).mean())] = 'ranging'
return regime
trading_strategy.py
+303
View File
@@ -0,0 +1,303 @@
import numpy as np
import pandas as pd
from typing import Dict, List, Optional, Tuple
import logging
from datetime import datetime
from dataclasses import dataclass
import json
import os
from .technical_analysis import TechnicalAnalyzer
from .machine_learning import MachineLearningModel
from .risk_management import RiskManager
from .sentiment_analysis import SentimentAnalyzer
logger = logging.getLogger(__name__)
@dataclass
class TradeSignal:
"""Data class to store trade signals"""
symbol: str
action: str # 'buy', 'sell', or 'hold'
confidence: float
price: float
stop_loss: float
take_profit: float
position_size: float
timestamp: datetime
technical_score: float
ml_score: float
sentiment_score: float
risk_score: float
class TradingStrategy:
def __init__(self, config: Dict):
self.config = config
self.technical_analyzer = TechnicalAnalyzer(config)
self.ml_model = MachineLearningModel(config)
self.risk_manager = RiskManager(config)
self.sentiment_analyzer = SentimentAnalyzer(config)
# Initialize state
self.active_trades = {}
self.trade_history = []
self.performance_metrics = {}
def analyze_market(self, symbol: str, data: pd.DataFrame) -> TradeSignal:
"""
Analyze market conditions and generate trading signals
"""
try:
# Technical analysis
technical_signals = self.technical_analyzer.generate_signals(data)
technical_score = technical_signals['signal']
# Machine learning prediction
ml_predictions = self.ml_model.predict(data)
ml_score = ml_predictions[-1] if ml_predictions is not None else 0
# Sentiment analysis
sentiment_score = self.sentiment_analyzer.get_combined_sentiment(symbol)
if sentiment_score is None:
sentiment_score = 0
# Risk assessment
risk_score = self.risk_manager.calculate_risk_metrics()
# Combine signals
combined_score = (
technical_score * self.config['signal_weights']['technical'] +
ml_score * self.config['signal_weights']['ml'] +
sentiment_score * self.config['signal_weights']['sentiment'] +
risk_score * self.config['signal_weights']['risk']
)
# Generate trading signal
current_price = data['close'].iloc[-1]
# Calculate stop loss and take profit levels
stop_loss = self._calculate_stop_loss(current_price, combined_score)
take_profit = self._calculate_take_profit(current_price, combined_score)
# Calculate position size
position_size, required_margin = self.risk_manager.calculate_position_size(
current_price,
stop_loss,
self.risk_manager.current_capital
)
# Determine action
if combined_score > self.config['signal_thresholds']['buy']:
action = 'buy'
elif combined_score < self.config['signal_thresholds']['sell']:
action = 'sell'
else:
action = 'hold'
# Create trade signal
signal = TradeSignal(
symbol=symbol,
action=action,
confidence=abs(combined_score),
price=current_price,
stop_loss=stop_loss,
take_profit=take_profit,
position_size=position_size,
timestamp=datetime.now(),
technical_score=technical_score,
ml_score=ml_score,
sentiment_score=sentiment_score,
risk_score=risk_score
)
return signal
except Exception as e:
logger.error(f"Error analyzing market: {str(e)}")
return None
def execute_trade(self, signal: TradeSignal) -> bool:
"""
Execute a trade based on the signal
"""
try:
# Validate trade
if not self.risk_manager.validate_trade(
signal.symbol,
signal.position_size,
signal.price,
signal.stop_loss,
signal.take_profit
):
return False
# Execute trade
if signal.action in ['buy', 'sell']:
# Update position
pnl = self.risk_manager.update_position(
signal.symbol,
signal.price,
signal.price,
signal.position_size,
signal.action
)
if pnl is not None:
# Record trade
self.trade_history.append({
'symbol': signal.symbol,
'action': signal.action,
'entry_price': signal.price,
'exit_price': signal.price,
'position_size': signal.position_size,
'pnl': pnl,
'entry_time': signal.timestamp,
'exit_time': datetime.now()
})
# Update performance metrics
self._update_performance_metrics()
return True
return False
except Exception as e:
logger.error(f"Error executing trade: {str(e)}")
return False
def update_positions(self, current_prices: Dict[str, float]):
"""
Update all open positions with current prices
"""
try:
for symbol, price in current_prices.items():
if symbol in self.risk_manager.positions:
position = self.risk_manager.positions[symbol]
pnl = self.risk_manager.update_position(
symbol,
position['entry_price'],
price,
position['position_size'],
position['position_type']
)
if pnl is not None:
# Record trade
self.trade_history.append({
'symbol': symbol,
'action': 'close',
'entry_price': position['entry_price'],
'exit_price': price,
'position_size': position['position_size'],
'pnl': pnl,
'entry_time': position['entry_time'],
'exit_time': datetime.now()
})
# Update performance metrics
self._update_performance_metrics()
except Exception as e:
logger.error(f"Error updating positions: {str(e)}")
def _calculate_stop_loss(self, price: float, signal: float) -> float:
"""
Calculate stop loss level based on signal strength
"""
# Adjust stop loss distance based on signal strength
base_stop_loss = self.config['stop_loss_pct']
signal_factor = abs(signal)
stop_loss_distance = base_stop_loss * (1 + signal_factor)
if signal > 0: # Buy signal
return price * (1 - stop_loss_distance)
else: # Sell signal
return price * (1 + stop_loss_distance)
def _calculate_take_profit(self, price: float, signal: float) -> float:
"""
Calculate take profit level based on signal strength
"""
# Adjust take profit distance based on signal strength
base_take_profit = self.config['take_profit_pct']
signal_factor = abs(signal)
take_profit_distance = base_take_profit * (1 + signal_factor)
if signal > 0: # Buy signal
return price * (1 + take_profit_distance)
else: # Sell signal
return price * (1 - take_profit_distance)
def _update_performance_metrics(self):
"""
Update performance metrics based on trade history
"""
if not self.trade_history:
return
# Convert trade history to DataFrame
df = pd.DataFrame(self.trade_history)
# Calculate metrics
self.performance_metrics = {
'total_trades': len(df),
'winning_trades': len(df[df['pnl'] > 0]),
'losing_trades': len(df[df['pnl'] < 0]),
'win_rate': len(df[df['pnl'] > 0]) / len(df),
'total_pnl': df['pnl'].sum(),
'avg_pnl': df['pnl'].mean(),
'max_drawdown': self.risk_manager.risk_metrics.max_drawdown if self.risk_manager.risk_metrics else 0,
'sharpe_ratio': self.risk_manager.risk_metrics.sharpe_ratio if self.risk_manager.risk_metrics else 0,
'profit_factor': self.risk_manager.risk_metrics.profit_factor if self.risk_manager.risk_metrics else 0
}
def should_stop_trading(self) -> bool:
"""
Check if trading should be stopped based on risk metrics
"""
return self.risk_manager.should_stop_trading()
def save_state(self, filepath: str):
"""
Save trading strategy state
"""
try:
state = {
'active_trades': self.active_trades,
'trade_history': self.trade_history,
'performance_metrics': self.performance_metrics,
'risk_metrics': self.risk_manager.risk_metrics.__dict__ if self.risk_manager.risk_metrics else None
}
with open(filepath, 'w') as f:
json.dump(state, f, default=str)
logger.info(f"Trading strategy state saved to {filepath}")
return True
except Exception as e:
logger.error(f"Error saving state: {str(e)}")
return False
def load_state(self, filepath: str):
"""
Load trading strategy state
"""
try:
with open(filepath, 'r') as f:
state = json.load(f)
self.active_trades = state['active_trades']
self.trade_history = state['trade_history']
self.performance_metrics = state['performance_metrics']
if state['risk_metrics']:
self.risk_manager.risk_metrics = RiskMetrics(**state['risk_metrics'])
logger.info(f"Trading strategy state loaded from {filepath}")
return True
except Exception as e:
logger.error(f"Error loading state: {str(e)}")
return False
trading_system.py
+318
View File
@@ -0,0 +1,318 @@
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
import logging
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
import json
import os
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('logs/trading_system.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
@dataclass
class Trade:
symbol: str
direction: str
entry_price: float
stop_loss: float
take_profit: float
position_size: float
entry_time: datetime
exit_price: Optional[float] = None
exit_time: Optional[datetime] = None
pnl: Optional[float] = None
status: str = 'open'
class TradingSystem:
def __init__(self, initial_capital: float = 50.0):
self.initial_capital = initial_capital
self.current_capital = initial_capital
self.performance_history: List[Dict] = []
self.active_trades: Dict[str, Trade] = {}
self.trade_history: List[Trade] = []
self.strategy_parameters = self._get_initial_parameters()
self.risk_metrics = self._initialize_risk_metrics()
self.last_recalibration = datetime.now()
# Load configuration
self._load_config()
def _load_config(self):
"""Load configuration from config file"""
try:
with open('config/config.json', 'r') as f:
self.config = json.load(f)
except FileNotFoundError:
logger.warning("Config file not found. Using default parameters.")
self.config = self._get_default_config()
def _get_default_config(self) -> Dict:
"""Get default configuration parameters"""
return {
'risk_per_trade': 0.01, # 1% risk per trade
'max_positions': 2,
'min_win_rate': 0.4,
'recalibration_window': 20,
'max_drawdown': 0.1, # 10% maximum drawdown
'leverage': 1, # No leverage initially
'position_sizing': {
'method': 'fixed_fractional',
'fraction': 0.01 # 1% of capital per trade
}
}
def _get_initial_parameters(self) -> Dict:
"""Get initial strategy parameters"""
return {
'rsi_period': 14,
'rsi_overbought': 70,
'rsi_oversold': 30,
'bb_period': 20,
'bb_std': 2,
'macd_fast': 12,
'macd_slow': 26,
'macd_signal': 9,
'atr_period': 14,
'atr_multiplier': 2
}
def _initialize_risk_metrics(self) -> Dict:
"""Initialize risk metrics tracking"""
return {
'total_trades': 0,
'winning_trades': 0,
'losing_trades': 0,
'win_rate': 0.0,
'profit_factor': 0.0,
'max_drawdown': 0.0,
'current_drawdown': 0.0,
'avg_trade': 0.0,
'largest_win': 0.0,
'largest_loss': 0.0
}
def monitor_performance(self, window_size: int = 20) -> bool:
"""
Monitor recent performance and determine if recalibration is needed
Returns True if recalibration is needed
"""
if len(self.performance_history) < window_size:
return False
recent_performance = self.performance_history[-window_size:]
win_rate = sum(1 for trade in recent_performance if trade['pnl'] > 0) / window_size
# Check various performance metrics
needs_recalibration = False
# Win rate check
if win_rate < self.config['min_win_rate']:
logger.warning(f"Win rate {win_rate:.2%} below threshold {self.config['min_win_rate']:.2%}")
needs_recalibration = True
# Drawdown check
current_drawdown = self._calculate_drawdown()
if current_drawdown > self.config['max_drawdown']:
logger.warning(f"Current drawdown {current_drawdown:.2%} exceeds maximum {self.config['max_drawdown']:.2%}")
needs_recalibration = True
# Profit factor check
profit_factor = self._calculate_profit_factor(recent_performance)
if profit_factor < 1.0:
logger.warning(f"Profit factor {profit_factor:.2f} below 1.0")
needs_recalibration = True
return needs_recalibration
def recalibrate_strategy(self, market_data: pd.DataFrame):
"""
Adjust strategy parameters based on recent market conditions
"""
logger.info("Starting strategy recalibration")
# Analyze market conditions
volatility = self._calculate_volatility(market_data)
trend_strength = self._calculate_trend_strength(market_data)
# Adjust parameters based on market conditions
new_parameters = self.strategy_parameters.copy()
# Adjust RSI levels based on volatility
if volatility > 0.02: # High volatility
new_parameters['rsi_overbought'] = 75
new_parameters['rsi_oversold'] = 25
else: # Low volatility
new_parameters['rsi_overbought'] = 70
new_parameters['rsi_oversold'] = 30
# Adjust ATR multiplier based on trend strength
if trend_strength > 0.7: # Strong trend
new_parameters['atr_multiplier'] = 2.5
else: # Weak trend
new_parameters['atr_multiplier'] = 2.0
# Update parameters
self.strategy_parameters = new_parameters
self.last_recalibration = datetime.now()
logger.info("Strategy recalibration completed")
logger.info(f"New parameters: {new_parameters}")
def calculate_position_size(self, entry_price: float, stop_loss: float) -> float:
"""
Calculate position size based on risk management rules
"""
risk_amount = self.current_capital * self.config['risk_per_trade']
risk_per_unit = abs(entry_price - stop_loss)
if risk_per_unit == 0:
logger.warning("Risk per unit is zero. Cannot calculate position size.")
return 0
position_size = risk_amount / risk_per_unit
# Apply leverage if configured
if self.config['leverage'] > 1:
position_size *= self.config['leverage']
# Ensure position size doesn't exceed maximum allowed
max_position = self.current_capital * self.config['position_sizing']['fraction']
position_size = min(position_size, max_position)
return position_size
def _calculate_volatility(self, data: pd.DataFrame) -> float:
"""Calculate market volatility"""
returns = data['close'].pct_change()
return returns.std()
def _calculate_trend_strength(self, data: pd.DataFrame) -> float:
"""Calculate trend strength using ADX"""
# Implementation would go here
return 0.5 # Placeholder
def _calculate_drawdown(self) -> float:
"""Calculate current drawdown"""
if not self.performance_history:
return 0.0
peak = max(self.performance_history, key=lambda x: x['equity'])['equity']
current = self.performance_history[-1]['equity']
return (peak - current) / peak
def _calculate_profit_factor(self, trades: List[Dict]) -> float:
"""Calculate profit factor from recent trades"""
gross_profit = sum(t['pnl'] for t in trades if t['pnl'] > 0)
gross_loss = abs(sum(t['pnl'] for t in trades if t['pnl'] < 0))
if gross_loss == 0:
return float('inf')
return gross_profit / gross_loss
def update_risk_metrics(self, trade: Trade):
"""Update risk metrics after a trade"""
self.risk_metrics['total_trades'] += 1
if trade.pnl and trade.pnl > 0:
self.risk_metrics['winning_trades'] += 1
self.risk_metrics['largest_win'] = max(
self.risk_metrics['largest_win'],
trade.pnl
)
elif trade.pnl and trade.pnl < 0:
self.risk_metrics['losing_trades'] += 1
self.risk_metrics['largest_loss'] = min(
self.risk_metrics['largest_loss'],
trade.pnl
)
# Update win rate
if self.risk_metrics['total_trades'] > 0:
self.risk_metrics['win_rate'] = (
self.risk_metrics['winning_trades'] /
self.risk_metrics['total_trades']
)
# Update average trade
if trade.pnl:
self.risk_metrics['avg_trade'] = (
(self.risk_metrics['avg_trade'] * (self.risk_metrics['total_trades'] - 1) +
trade.pnl) / self.risk_metrics['total_trades']
)
def save_state(self):
"""Save current system state"""
state = {
'current_capital': self.current_capital,
'strategy_parameters': self.strategy_parameters,
'risk_metrics': self.risk_metrics,
'last_recalibration': self.last_recalibration.isoformat(),
'active_trades': {
symbol: {
'direction': trade.direction,
'entry_price': trade.entry_price,
'stop_loss': trade.stop_loss,
'take_profit': trade.take_profit,
'position_size': trade.position_size,
'entry_time': trade.entry_time.isoformat()
}
for symbol, trade in self.active_trades.items()
}
}
try:
with open('data/system_state.json', 'w') as f:
json.dump(state, f, indent=4)
except Exception as e:
logger.error(f"Error saving system state: {str(e)}")
def load_state(self):
"""Load system state from file"""
try:
with open('data/system_state.json', 'r') as f:
state = json.load(f)
self.current_capital = state['current_capital']
self.strategy_parameters = state['strategy_parameters']
self.risk_metrics = state['risk_metrics']
self.last_recalibration = datetime.fromisoformat(state['last_recalibration'])
# Reconstruct active trades
self.active_trades = {}
for symbol, trade_data in state['active_trades'].items():
self.active_trades[symbol] = Trade(
symbol=symbol,
direction=trade_data['direction'],
entry_price=trade_data['entry_price'],
stop_loss=trade_data['stop_loss'],
take_profit=trade_data['take_profit'],
position_size=trade_data['position_size'],
entry_time=datetime.fromisoformat(trade_data['entry_time'])
)
except FileNotFoundError:
logger.info("No saved state found. Starting fresh.")
except Exception as e:
logger.error(f"Error loading system state: {str(e)}")
def get_system_status(self) -> Dict:
"""Get current system status"""
return {
'current_capital': self.current_capital,
'total_trades': self.risk_metrics['total_trades'],
'win_rate': self.risk_metrics['win_rate'],
'profit_factor': self.risk_metrics['profit_factor'],
'current_drawdown': self.risk_metrics['current_drawdown'],
'active_trades': len(self.active_trades),
'last_recalibration': self.last_recalibration.isoformat(),
'strategy_parameters': self.strategy_parameters
}
visualization.py
+169
View File
@@ -0,0 +1,169 @@
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from typing import Dict, List
import json
import os
def plot_backtest_results(results_file: str, save_path: str = None):
"""
Create interactive plots for backtest results
"""
# Load results
with open(results_file, 'r') as f:
results = json.load(f)
# Convert data to DataFrames
trades_df = pd.DataFrame(results['trades'])
equity_df = pd.DataFrame(results['equity_curve'])
# Create figure with secondary y-axis
fig = make_subplots(
rows=2, cols=1,
shared_xaxes=True,
vertical_spacing=0.03,
subplot_titles=('Price and Trades', 'Equity Curve'),
row_heights=[0.7, 0.3]
)
# Add price line
fig.add_trace(
go.Scatter(
x=trades_df['timestamp'],
y=trades_df['price'],
name='Price',
line=dict(color='blue')
),
row=1, col=1
)
# Add buy trades
buy_trades = trades_df[trades_df['action'] == 'buy']
fig.add_trace(
go.Scatter(
x=buy_trades['timestamp'],
y=buy_trades['price'],
mode='markers',
name='Buy',
marker=dict(color='green', size=10)
),
row=1, col=1
)
# Add sell trades
sell_trades = trades_df[trades_df['action'] == 'sell']
fig.add_trace(
go.Scatter(
x=sell_trades['timestamp'],
y=sell_trades['price'],
mode='markers',
name='Sell',
marker=dict(color='red', size=10)
),
row=1, col=1
)
# Add equity curve
fig.add_trace(
go.Scatter(
x=equity_df['timestamp'],
y=equity_df['equity'],
name='Equity',
line=dict(color='purple')
),
row=2, col=1
)
# Update layout
fig.update_layout(
title='Backtest Results',
xaxis_title='Date',
yaxis_title='Price',
yaxis2_title='Equity',
showlegend=True,
height=800
)
# Save plot if path is provided
if save_path:
fig.write_html(save_path)
print(f"Plot saved to {save_path}")
return fig
def plot_performance_metrics(results_files: List[str], save_path: str = None):
"""
Create comparison plot of performance metrics across different scenarios
"""
metrics_data = []
for file in results_files:
with open(file, 'r') as f:
results = json.load(f)
metrics = results['performance_metrics']
# Extract scenario name from filename
scenario_name = os.path.basename(file).replace('backtest_results_', '').replace('.json', '')
metrics_data.append({
'Scenario': scenario_name,
'Total Return': metrics['total_return'],
'Annual Return': metrics['annual_return'],
'Sharpe Ratio': metrics['sharpe_ratio'],
'Max Drawdown': metrics['max_drawdown'],
'Win Rate': metrics['win_rate']
})
# Create DataFrame
df = pd.DataFrame(metrics_data)
# Create figure
fig = go.Figure()
# Add bars for each metric
metrics = ['Total Return', 'Annual Return', 'Sharpe Ratio', 'Max Drawdown', 'Win Rate']
for metric in metrics:
fig.add_trace(
go.Bar(
name=metric,
x=df['Scenario'],
y=df[metric]
)
)
# Update layout
fig.update_layout(
title='Performance Metrics Comparison',
xaxis_title='Scenario',
yaxis_title='Value',
barmode='group',
height=600
)
# Save plot if path is provided
if save_path:
fig.write_html(save_path)
print(f"Plot saved to {save_path}")
return fig
def main():
# Example usage
results_dir = 'backtest_results'
results_files = [
os.path.join(results_dir, f) for f in os.listdir(results_dir)
if f.startswith('backtest_results_') and f.endswith('.json')
]
# Create plots for each scenario
for file in results_files:
scenario_name = os.path.basename(file).replace('.json', '')
plot_path = os.path.join(results_dir, f'{scenario_name}_plot.html')
plot_backtest_results(file, plot_path)
# Create comparison plot
comparison_path = os.path.join(results_dir, 'performance_comparison.html')
plot_performance_metrics(results_files, comparison_path)
if __name__ == "__main__":
main()