mirror of
https://github.com/chrisnov-it/quantumbotx.git
synced 2026-07-27 18:57:47 +00:00
feat: Major v2.0 enhancements and new features
🔧 Core System Improvements: - Enhanced backtesting engine with realistic spread modeling and ATR-based risk management - Improved bot controller with better error handling and status tracking - Optimized MT5 integration with symbol verification and market watch integration - Strengthened database queries with better performance and reliability 🎯 New Strategy Features: - Added index strategies (Index Momentum, Index Breakout Pro) for stock market trading - Implemented market condition detector for dynamic strategy adaptation - Created performance scorer for strategy evaluation and ranking - Added strategy switcher system for automatic strategy optimization 📚 Educational Framework: - New beginner guide documentation for newcomer onboarding - Enhanced FAQ section with common trading questions - Quick start guide for rapid setup and deployment - Improved AI mentor integration with personalized guidance 🌍 Multi-Asset Expansion: - Extended data collection for 20+ trading instruments (Forex, Crypto, Indices) - Enhanced broker compatibility with FBS and other platforms - Improved symbol migration system for seamless broker switching - Added holiday integration for culturally-aware trading automation 🧪 Testing & Validation: - Added comprehensive index strategy testing suite - Enhanced holiday integration validation - Dynamic strategy signal testing for improved reliability - EURUSD optimization testing with London session focus ⚡ Performance & UI: - Frontend JavaScript optimizations for better trading bot management - Enhanced templates with improved user experience - Database migration system for smooth version upgrades - Optimized data download scripts for better efficiency 📊 Analytics & Monitoring: - Strengthened Flask application architecture with better routing - Improved logging system for production deployment - Enhanced error handling across all components - Better API response handling and status reporting
This commit is contained in:
Reynov Christian
@@ -14,6 +14,15 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
# Load environment
|
||||
load_dotenv()
|
||||
|
||||
# Initialize default values for imports
|
||||
mt5 = None
|
||||
initialize_mt5 = None
|
||||
find_mt5_symbol = None
|
||||
queries = None
|
||||
hentikan_bot = None
|
||||
mulai_bot = None
|
||||
active_bots = {}
|
||||
|
||||
try:
|
||||
import MetaTrader5 as mt5
|
||||
from core.utils.mt5 import initialize_mt5, find_mt5_symbol
|
||||
@@ -28,7 +37,7 @@ except ImportError as e:
|
||||
def detect_current_broker():
|
||||
"""Detect current broker and return standardized name"""
|
||||
try:
|
||||
account_info = mt5.account_info()
|
||||
account_info = mt5.account_info() # pyright: ignore
|
||||
if not account_info:
|
||||
return "Unknown"
|
||||
|
||||
@@ -96,6 +105,10 @@ def get_broker_preferred_symbols():
|
||||
|
||||
def analyze_current_bots():
|
||||
"""Analyze current bot configurations and symbol availability"""
|
||||
if not MT5_AVAILABLE or queries is None:
|
||||
print("❌ Cannot analyze bots: Required modules not available")
|
||||
return "Unknown", []
|
||||
|
||||
print("🔍 Analyzing Current Bot Configurations")
|
||||
print("=" * 45)
|
||||
|
||||
@@ -119,7 +132,7 @@ def analyze_current_bots():
|
||||
print(f" Current Market: {current_market}")
|
||||
|
||||
# Test if current symbol works
|
||||
resolved_symbol = find_mt5_symbol(current_market)
|
||||
resolved_symbol = find_mt5_symbol(current_market) if find_mt5_symbol else None
|
||||
if resolved_symbol:
|
||||
print(f" ✅ Symbol resolved to: {resolved_symbol}")
|
||||
if resolved_symbol != current_market:
|
||||
@@ -137,7 +150,7 @@ def analyze_current_bots():
|
||||
# Try to find broker-preferred alternative
|
||||
if current_market.upper() in preferred_symbols:
|
||||
preferred = preferred_symbols[current_market.upper()]
|
||||
test_symbol = find_mt5_symbol(preferred)
|
||||
test_symbol = find_mt5_symbol(preferred) if find_mt5_symbol else None
|
||||
if test_symbol:
|
||||
print(f" 💡 Broker prefers: {preferred} -> resolves to: {test_symbol}")
|
||||
symbol_issues.append({
|
||||
@@ -169,6 +182,10 @@ def analyze_current_bots():
|
||||
|
||||
def migrate_bot_symbols(symbol_issues):
|
||||
"""Migrate bot symbols to correct broker-specific symbols"""
|
||||
if not MT5_AVAILABLE or queries is None:
|
||||
print("❌ Cannot migrate symbols: Required modules not available")
|
||||
return
|
||||
|
||||
print("\\n🔄 SYMBOL MIGRATION")
|
||||
print("=" * 25)
|
||||
|
||||
@@ -208,7 +225,7 @@ def migrate_bot_symbols(symbol_issues):
|
||||
continue
|
||||
|
||||
# Stop bot if running
|
||||
if bot_id in active_bots:
|
||||
if bot_id in active_bots and hentikan_bot:
|
||||
print(f"🛑 Stopping bot {bot_id} for migration...")
|
||||
hentikan_bot(bot_id)
|
||||
|
||||
@@ -231,7 +248,7 @@ def migrate_bot_symbols(symbol_issues):
|
||||
migrated += 1
|
||||
|
||||
# Restart if it was running
|
||||
if bot_id in active_bots:
|
||||
if bot_id in active_bots and mulai_bot:
|
||||
print(f"🚀 Restarting bot {bot_id}...")
|
||||
mulai_bot(bot_id)
|
||||
else:
|
||||
@@ -244,6 +261,10 @@ def migrate_bot_symbols(symbol_issues):
|
||||
|
||||
def create_broker_config_backup():
|
||||
"""Create a backup of current broker configuration"""
|
||||
if not MT5_AVAILABLE or queries is None:
|
||||
print("❌ Cannot create backup: Required modules not available")
|
||||
return None
|
||||
|
||||
current_broker = detect_current_broker()
|
||||
|
||||
backup_data = {
|
||||
@@ -282,11 +303,19 @@ def main():
|
||||
|
||||
# Connect to MT5
|
||||
try:
|
||||
ACCOUNT = int(os.getenv('MT5_LOGIN'))
|
||||
PASSWORD = os.getenv('MT5_PASSWORD')
|
||||
SERVER = os.getenv('MT5_SERVER')
|
||||
mt5_login = os.getenv('MT5_LOGIN')
|
||||
mt5_password = os.getenv('MT5_PASSWORD')
|
||||
mt5_server = os.getenv('MT5_SERVER')
|
||||
|
||||
if not initialize_mt5(ACCOUNT, PASSWORD, SERVER):
|
||||
if not mt5_login or not mt5_password or not mt5_server:
|
||||
print("❌ MT5 credentials not found in environment")
|
||||
return
|
||||
|
||||
ACCOUNT = int(mt5_login)
|
||||
PASSWORD = mt5_password
|
||||
SERVER = mt5_server
|
||||
|
||||
if not initialize_mt5 or not initialize_mt5(ACCOUNT, PASSWORD, SERVER):
|
||||
print("❌ Failed to connect to MT5")
|
||||
return
|
||||
except Exception as e:
|
||||
@@ -294,7 +323,7 @@ def main():
|
||||
return
|
||||
|
||||
# Create backup
|
||||
backup_file = create_broker_config_backup()
|
||||
create_broker_config_backup()
|
||||
|
||||
# Analyze current configuration
|
||||
current_broker, symbol_issues = analyze_current_bots()
|
||||
@@ -305,13 +334,14 @@ def main():
|
||||
else:
|
||||
print("\\n✅ All bots are properly configured for current broker!")
|
||||
|
||||
print(f"\\n💡 TIPS FOR FUTURE BROKER SWITCHES:")
|
||||
print("\\n💡 TIPS FOR FUTURE BROKER SWITCHES:")
|
||||
print("1. Run this script after connecting to a new broker")
|
||||
print("2. Keep backup files for easy rollback")
|
||||
print("3. Test bot functionality after migration")
|
||||
print("4. The enhanced find_mt5_symbol() will auto-detect most symbols")
|
||||
|
||||
mt5.shutdown()
|
||||
if mt5:
|
||||
mt5.shutdown() # pyright: ignore
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,211 @@
|
||||
#!/usr/bin/env python3
|
||||
# debug_index_strategy.py - Debug the INDEX_BREAKOUT_PRO strategy with US500 data
|
||||
|
||||
import sys
|
||||
import os
|
||||
import pandas as pd
|
||||
import logging
|
||||
|
||||
# Add project root to path
|
||||
project_root = os.path.dirname(os.path.abspath(__file__))
|
||||
sys.path.append(project_root)
|
||||
|
||||
# Set up logging
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def test_index_breakout_strategy():
|
||||
"""Test the INDEX_BREAKOUT_PRO strategy with US500 data"""
|
||||
|
||||
print("🔍 Debugging INDEX_BREAKOUT_PRO Strategy with US500 Data")
|
||||
print("=" * 70)
|
||||
|
||||
try:
|
||||
# Import required modules
|
||||
from core.backtesting.enhanced_engine import run_enhanced_backtest
|
||||
from core.strategies.strategy_map import STRATEGY_MAP
|
||||
|
||||
# Check if strategy exists
|
||||
strategy_id = 'INDEX_BREAKOUT_PRO'
|
||||
if strategy_id not in STRATEGY_MAP:
|
||||
print(f"❌ Strategy {strategy_id} not found in STRATEGY_MAP")
|
||||
print(f"Available strategies: {list(STRATEGY_MAP.keys())}")
|
||||
return False
|
||||
|
||||
strategy_class = STRATEGY_MAP[strategy_id]
|
||||
print(f"✅ Strategy found: {strategy_class}")
|
||||
print(f"Strategy name: {getattr(strategy_class, 'name', 'Unknown')}")
|
||||
print(f"Strategy description: {getattr(strategy_class, 'description', 'No description')}")
|
||||
|
||||
# Load US500 data
|
||||
csv_file = 'lab/backtest_data/US500_H1_data.csv'
|
||||
if not os.path.exists(csv_file):
|
||||
print(f"❌ Data file not found: {csv_file}")
|
||||
return False
|
||||
|
||||
print(f"\n📊 Loading data from: {csv_file}")
|
||||
df = pd.read_csv(csv_file, parse_dates=['time'])
|
||||
print(f"✅ Loaded {len(df)} rows of data")
|
||||
print(f"Date range: {df['time'].min()} to {df['time'].max()}")
|
||||
print(f"Columns: {list(df.columns)}")
|
||||
print("Sample data:")
|
||||
print(df.head())
|
||||
|
||||
# Test strategy parameters
|
||||
print("\n⚙️ Testing strategy parameters...")
|
||||
if hasattr(strategy_class, 'get_definable_params'):
|
||||
params_def = strategy_class.get_definable_params()
|
||||
print(f"✅ Strategy has {len(params_def)} definable parameters:")
|
||||
for param in params_def:
|
||||
name = param.get('name', 'Unknown')
|
||||
display_name = param.get('display_name', param.get('label', 'Unknown'))
|
||||
default = param.get('default', 'No default')
|
||||
print(f" - {name} ({display_name}): {default}")
|
||||
else:
|
||||
print("❌ Strategy has no get_definable_params method")
|
||||
return False
|
||||
|
||||
# Test strategy instantiation
|
||||
print("\n🧪 Testing strategy instantiation...")
|
||||
try:
|
||||
# Create a mock bot instance
|
||||
class MockBot:
|
||||
def __init__(self):
|
||||
self.market_for_mt5 = 'US500'
|
||||
self.status = 'Testing'
|
||||
|
||||
mock_bot = MockBot()
|
||||
strategy_instance = strategy_class(mock_bot, {})
|
||||
print("✅ Strategy instantiated successfully")
|
||||
|
||||
# Test analyze_df method
|
||||
print("\n🔬 Testing analyze_df method...")
|
||||
|
||||
# Use a smaller subset for testing
|
||||
test_df = df.tail(500).copy() # Last 500 rows
|
||||
print(f"Testing with {len(test_df)} rows")
|
||||
|
||||
result_df = strategy_instance.analyze_df(test_df)
|
||||
print("✅ analyze_df completed")
|
||||
print(f"Result columns: {list(result_df.columns)}")
|
||||
|
||||
# Check for signals
|
||||
if 'signal' in result_df.columns:
|
||||
signals = result_df['signal'].value_counts()
|
||||
print(f"Signal distribution: {signals.to_dict()}")
|
||||
|
||||
# Count non-HOLD signals
|
||||
non_hold_signals = result_df[result_df['signal'] != 'HOLD']
|
||||
print(f"Non-HOLD signals: {len(non_hold_signals)}")
|
||||
|
||||
if len(non_hold_signals) > 0:
|
||||
print("Sample signals:")
|
||||
print(non_hold_signals[['time', 'signal', 'explanation']].head(10) if 'time' in result_df.columns else non_hold_signals[['signal', 'explanation']].head(10))
|
||||
else:
|
||||
print("❌ No trading signals generated!")
|
||||
print("Sample explanations:")
|
||||
print(result_df['explanation'].tail(10).tolist())
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Strategy testing failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
# Test full backtesting
|
||||
print("\n🚀 Testing full backtest...")
|
||||
|
||||
# Simulate web interface parameters
|
||||
web_params = {
|
||||
'breakout_period': 20,
|
||||
'volume_surge_multiplier': 2.0,
|
||||
'confirmation_candles': 2,
|
||||
'atr_multiplier_sl': 2.0,
|
||||
'atr_multiplier_tp': 4.0
|
||||
}
|
||||
|
||||
# Enhanced parameters (like API mapping)
|
||||
enhanced_params = web_params.copy()
|
||||
enhanced_params['risk_percent'] = 1.0 # Conservative for index
|
||||
enhanced_params['sl_atr_multiplier'] = web_params.get('atr_multiplier_sl', 2.0)
|
||||
enhanced_params['tp_atr_multiplier'] = web_params.get('atr_multiplier_tp', 4.0)
|
||||
|
||||
print(f"Parameters: {enhanced_params}")
|
||||
|
||||
# Engine configuration
|
||||
engine_config = {
|
||||
'enable_spread_costs': True,
|
||||
'enable_slippage': True,
|
||||
'enable_realistic_execution': True
|
||||
}
|
||||
|
||||
# Extract symbol name (like API does)
|
||||
symbol_name = 'US500'
|
||||
print(f"Symbol: {symbol_name}")
|
||||
|
||||
# Use smaller dataset for testing
|
||||
test_df = df.tail(1000).copy() # Last 1000 rows for faster testing
|
||||
|
||||
results = run_enhanced_backtest(
|
||||
strategy_id,
|
||||
enhanced_params,
|
||||
test_df,
|
||||
symbol_name=symbol_name,
|
||||
engine_config=engine_config
|
||||
)
|
||||
|
||||
if 'error' in results:
|
||||
print(f"❌ Backtest error: {results['error']}")
|
||||
return False
|
||||
|
||||
print("✅ Backtest completed successfully!")
|
||||
print("\n📈 Results Summary:")
|
||||
print(f" Strategy: {results.get('strategy_name', 'Unknown')}")
|
||||
print(f" Total Trades: {results.get('total_trades', 0)}")
|
||||
print(f" Wins: {results.get('wins', 0)}")
|
||||
print(f" Losses: {results.get('losses', 0)}")
|
||||
print(f" Win Rate: {results.get('win_rate_percent', 0):.1f}%")
|
||||
print(f" Total Profit USD: ${results.get('total_profit_usd', 0):.2f}")
|
||||
print(f" Max Drawdown: {results.get('max_drawdown_percent', 0):.1f}%")
|
||||
print(f" Final Capital: ${results.get('final_capital', 0):.2f}")
|
||||
|
||||
if results.get('total_trades', 0) == 0:
|
||||
print("\n❌ PROBLEM: No trades generated!")
|
||||
print("This could be why the web interface shows empty results.")
|
||||
|
||||
# Debug signal generation
|
||||
print("\n🔍 Debugging signal generation...")
|
||||
strategy_instance = strategy_class(MockBot(), enhanced_params)
|
||||
debug_df = test_df.tail(100).copy()
|
||||
debug_result = strategy_instance.analyze_df(debug_df)
|
||||
|
||||
if 'signal' in debug_result.columns:
|
||||
signals = debug_result['signal'].value_counts()
|
||||
print(f"Signal counts in last 100 rows: {signals.to_dict()}")
|
||||
|
||||
if 'BUY' in signals or 'SELL' in signals:
|
||||
print("✅ Signals are being generated by strategy")
|
||||
print("❓ Problem might be in the backtesting engine")
|
||||
else:
|
||||
print("❌ Strategy is not generating BUY/SELL signals")
|
||||
print("Sample explanations:")
|
||||
sample_explanations = debug_result['explanation'].tail(10).tolist()
|
||||
for i, exp in enumerate(sample_explanations):
|
||||
print(f" {i+1}: {exp}")
|
||||
else:
|
||||
print("✅ Trades were generated successfully!")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Test failed with exception: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
success = test_index_breakout_strategy()
|
||||
if success:
|
||||
print("\n✅ Debug completed successfully")
|
||||
else:
|
||||
print("\n❌ Debug revealed issues that need fixing")
|
||||
@@ -0,0 +1,293 @@
|
||||
#!/usr/bin/env python3
|
||||
# test_eurusd_london_session.py - EURUSD London Session Optimization
|
||||
|
||||
import sys
|
||||
import os
|
||||
import pandas as pd
|
||||
|
||||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
def test_eurusd_strategies():
|
||||
"""Test multiple strategies optimized for EURUSD London session"""
|
||||
|
||||
print("🇪🇺 EURUSD London Session Strategy Testing")
|
||||
print("=" * 70)
|
||||
print("🕐 Perfect timing: 1:19 PM London = Peak EURUSD activity!")
|
||||
|
||||
try:
|
||||
from core.backtesting.enhanced_engine import run_enhanced_backtest
|
||||
from core.strategies.strategy_map import STRATEGY_MAP
|
||||
|
||||
# Load EURUSD data
|
||||
csv_file = 'lab/backtest_data/EURUSD_H1_data.csv'
|
||||
if not os.path.exists(csv_file):
|
||||
print(f"❌ EURUSD data file not found: {csv_file}")
|
||||
return False
|
||||
|
||||
print(f"📊 Loading EURUSD H1 data...")
|
||||
df = pd.read_csv(csv_file, parse_dates=['time'])
|
||||
print(f"✅ Loaded {len(df)} rows of EURUSD data")
|
||||
print(f"Date range: {df['time'].min()} to {df['time'].max()}")
|
||||
|
||||
# Use recent 3000 rows for comprehensive testing
|
||||
test_df = df.tail(3000).copy()
|
||||
print(f"Testing with recent {len(test_df)} rows")
|
||||
print(f"Recent period: {test_df['time'].min()} to {test_df['time'].max()}")
|
||||
|
||||
# EURUSD-optimized strategy configurations
|
||||
eurusd_strategies = [
|
||||
{
|
||||
'name': 'MA_CROSSOVER - Conservative EURUSD',
|
||||
'strategy_id': 'MA_CROSSOVER',
|
||||
'params': {
|
||||
'fast_period': 12,
|
||||
'slow_period': 26,
|
||||
'risk_percent': 0.8, # Conservative for major pair
|
||||
'sl_atr_multiplier': 1.8,
|
||||
'tp_atr_multiplier': 3.6
|
||||
}
|
||||
},
|
||||
{
|
||||
'name': 'MA_CROSSOVER - Aggressive London Session',
|
||||
'strategy_id': 'MA_CROSSOVER',
|
||||
'params': {
|
||||
'fast_period': 8,
|
||||
'slow_period': 21,
|
||||
'risk_percent': 1.2, # More aggressive for volatility
|
||||
'sl_atr_multiplier': 2.0,
|
||||
'tp_atr_multiplier': 4.0
|
||||
}
|
||||
},
|
||||
{
|
||||
'name': 'RSI_CROSSOVER - EURUSD Momentum',
|
||||
'strategy_id': 'RSI_CROSSOVER',
|
||||
'params': {
|
||||
'rsi_period': 14,
|
||||
'rsi_ma_period': 7,
|
||||
'trend_filter_period': 30,
|
||||
'risk_percent': 1.0,
|
||||
'sl_atr_multiplier': 2.0,
|
||||
'tp_atr_multiplier': 4.0
|
||||
}
|
||||
},
|
||||
{
|
||||
'name': 'TURTLE_BREAKOUT - EURUSD Institutional',
|
||||
'strategy_id': 'TURTLE_BREAKOUT',
|
||||
'params': {
|
||||
'entry_period': 15, # Shorter for EURUSD
|
||||
'exit_period': 8,
|
||||
'risk_percent': 1.0,
|
||||
'sl_atr_multiplier': 2.0,
|
||||
'tp_atr_multiplier': 4.5
|
||||
}
|
||||
}
|
||||
]
|
||||
|
||||
# Engine configuration for realistic EURUSD trading
|
||||
engine_config = {
|
||||
'enable_spread_costs': True,
|
||||
'enable_slippage': True,
|
||||
'enable_realistic_execution': True
|
||||
}
|
||||
|
||||
results = []
|
||||
|
||||
for strategy_config in eurusd_strategies:
|
||||
print(f"\n🚀 Testing: {strategy_config['name']}")
|
||||
print("-" * 50)
|
||||
|
||||
strategy_id = strategy_config['strategy_id']
|
||||
params = strategy_config['params']
|
||||
|
||||
# Check if strategy exists
|
||||
if strategy_id not in STRATEGY_MAP:
|
||||
print(f"❌ Strategy {strategy_id} not found")
|
||||
continue
|
||||
|
||||
print(f"Parameters: {params}")
|
||||
|
||||
# Run backtest
|
||||
result = run_enhanced_backtest(
|
||||
strategy_id,
|
||||
params,
|
||||
test_df,
|
||||
symbol_name='EURUSD',
|
||||
engine_config=engine_config
|
||||
)
|
||||
|
||||
if 'error' in result:
|
||||
print(f"❌ Backtest error: {result['error']}")
|
||||
continue
|
||||
|
||||
# Extract key metrics
|
||||
total_trades = result.get('total_trades', 0)
|
||||
gross_profit = result.get('total_profit_usd', 0)
|
||||
spread_costs = result.get('total_spread_costs', 0)
|
||||
net_profit = result.get('net_profit_after_costs', 0)
|
||||
win_rate = result.get('win_rate_percent', 0)
|
||||
max_drawdown = result.get('max_drawdown_percent', 0)
|
||||
wins = result.get('wins', 0)
|
||||
losses = result.get('losses', 0)
|
||||
|
||||
print(f"📈 Results:")
|
||||
print(f" Total Trades: {total_trades}")
|
||||
print(f" Wins/Losses: {wins}/{losses}")
|
||||
print(f" Win Rate: {win_rate:.1f}%")
|
||||
print(f" Gross Profit: ${gross_profit:.2f}")
|
||||
print(f" Spread Costs: ${spread_costs:.2f}")
|
||||
print(f" Net Profit: ${net_profit:.2f}")
|
||||
print(f" Max Drawdown: {max_drawdown:.2f}%")
|
||||
|
||||
# Performance assessment
|
||||
if total_trades > 0:
|
||||
cost_ratio = (abs(spread_costs) / abs(gross_profit)) * 100 if gross_profit != 0 else 0
|
||||
print(f" Spread Cost Ratio: {cost_ratio:.1f}%")
|
||||
|
||||
# Quality indicators
|
||||
indicators = []
|
||||
if total_trades >= 20:
|
||||
indicators.append("✅ Good sample size")
|
||||
if win_rate >= 35:
|
||||
indicators.append("✅ Decent win rate")
|
||||
if net_profit > 0:
|
||||
indicators.append("✅ Profitable")
|
||||
if max_drawdown < 20:
|
||||
indicators.append("✅ Controlled risk")
|
||||
if cost_ratio < 30:
|
||||
indicators.append("✅ Reasonable costs")
|
||||
|
||||
if indicators:
|
||||
print(f" Quality: {' | '.join(indicators)}")
|
||||
|
||||
# Overall assessment
|
||||
if net_profit > 0 and max_drawdown < 30 and total_trades >= 10:
|
||||
print(f" 🎯 ASSESSMENT: EXCELLENT for EURUSD!")
|
||||
elif net_profit > -50 and max_drawdown < 50:
|
||||
print(f" ⚠️ ASSESSMENT: Acceptable, needs tuning")
|
||||
else:
|
||||
print(f" ❌ ASSESSMENT: Poor performance, avoid")
|
||||
|
||||
else:
|
||||
print(f" ❌ No trades generated - strategy too conservative")
|
||||
|
||||
# Store results for comparison
|
||||
results.append({
|
||||
'name': strategy_config['name'],
|
||||
'strategy_id': strategy_id,
|
||||
'trades': total_trades,
|
||||
'net_profit': net_profit,
|
||||
'win_rate': win_rate,
|
||||
'max_drawdown': max_drawdown,
|
||||
'spread_costs': spread_costs
|
||||
})
|
||||
|
||||
# Final comparison and recommendations
|
||||
print(f"\n🏆 EURUSD Strategy Performance Ranking")
|
||||
print("=" * 70)
|
||||
|
||||
# Sort by net profit
|
||||
results.sort(key=lambda x: x['net_profit'], reverse=True)
|
||||
|
||||
for i, result in enumerate(results):
|
||||
rank_emoji = ["🥇", "🥈", "🥉", "4️⃣", "5️⃣"][min(i, 4)]
|
||||
print(f"{rank_emoji} {result['name']}")
|
||||
print(f" Net Profit: ${result['net_profit']:.2f} | Win Rate: {result['win_rate']:.1f}% | Trades: {result['trades']} | Drawdown: {result['max_drawdown']:.1f}%")
|
||||
|
||||
# Best strategy recommendation
|
||||
if results and results[0]['net_profit'] > 0:
|
||||
best_strategy = results[0]
|
||||
print(f"\n🎯 RECOMMENDED FOR EURUSD LONDON SESSION:")
|
||||
print(f" Strategy: {best_strategy['name']}")
|
||||
print(f" Expected Performance: ${best_strategy['net_profit']:.2f} net profit")
|
||||
print(f" Win Rate: {best_strategy['win_rate']:.1f}%")
|
||||
print(f" Risk Level: {best_strategy['max_drawdown']:.1f}% max drawdown")
|
||||
|
||||
print(f"\n💡 LONDON SESSION ADVANTAGES:")
|
||||
print(f" • High liquidity during 1-4 PM London time")
|
||||
print(f" • Institutional activity creates clear trends")
|
||||
print(f" • Lower spreads during peak hours")
|
||||
print(f" • Perfect timing for breakout strategies")
|
||||
|
||||
else:
|
||||
print(f"\n⚠️ All strategies showed losses - consider:")
|
||||
print(f" • Using different timeframes (M15 or M30)")
|
||||
print(f" • Adjusting risk parameters")
|
||||
print(f" • Testing during different market conditions")
|
||||
|
||||
return len([r for r in results if r['net_profit'] > 0]) > 0
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Test failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
def test_london_session_timing():
|
||||
"""Test specific London session timing optimization"""
|
||||
|
||||
print(f"\n🕐 London Session Timing Analysis")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
# Load EURUSD data
|
||||
csv_file = 'lab/backtest_data/EURUSD_H1_data.csv'
|
||||
df = pd.read_csv(csv_file, parse_dates=['time'])
|
||||
|
||||
# Extract hour from timestamp
|
||||
df['hour'] = df['time'].dt.hour
|
||||
|
||||
# London session hours (UTC) - 8 AM to 4 PM London = 7 AM to 3 PM UTC
|
||||
london_hours = list(range(7, 16)) # 7 AM to 3 PM UTC
|
||||
|
||||
london_session_data = df[df['hour'].isin(london_hours)]
|
||||
other_session_data = df[~df['hour'].isin(london_hours)]
|
||||
|
||||
print(f"📊 Session Analysis:")
|
||||
print(f" London Session (7AM-3PM UTC): {len(london_session_data)} bars")
|
||||
print(f" Other Sessions: {len(other_session_data)} bars")
|
||||
|
||||
# Calculate volatility for each session
|
||||
london_volatility = london_session_data['close'].pct_change().std() * 100
|
||||
other_volatility = other_session_data['close'].pct_change().std() * 100
|
||||
|
||||
print(f"\n📈 Volatility Comparison:")
|
||||
print(f" London Session: {london_volatility:.4f}% per hour")
|
||||
print(f" Other Sessions: {other_volatility:.4f}% per hour")
|
||||
print(f" London Advantage: {(london_volatility/other_volatility-1)*100:.1f}% higher volatility")
|
||||
|
||||
# Current time analysis
|
||||
current_hour_utc = 12 # Approximate 1:19 PM London = 12:19 PM UTC
|
||||
print(f"\n🎯 Current Time Analysis (12:19 PM UTC):")
|
||||
if current_hour_utc in london_hours:
|
||||
print(f" ✅ PERFECT TIMING! You're in prime London session")
|
||||
print(f" ✅ Expected high liquidity and clear trends")
|
||||
print(f" ✅ Optimal for MA crossover and breakout strategies")
|
||||
else:
|
||||
print(f" ⚠️ Outside optimal London hours")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Timing analysis failed: {e}")
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("🚀 EURUSD London Session Comprehensive Testing")
|
||||
print("=" * 80)
|
||||
print("💡 Testing EURUSD strategies during peak London session activity")
|
||||
print("🕐 Current market timing: Perfect for institutional breakouts!")
|
||||
|
||||
success1 = test_london_session_timing()
|
||||
success2 = test_eurusd_strategies()
|
||||
|
||||
if success1 and success2:
|
||||
print(f"\n✅ EURUSD LONDON SESSION TESTING COMPLETE!")
|
||||
print(f"\n🎉 Key Takeaways:")
|
||||
print(f" 1. ✅ London session provides optimal EURUSD volatility")
|
||||
print(f" 2. ✅ Current timing (1:19 PM) is PERFECT for trading")
|
||||
print(f" 3. ✅ MA_CROSSOVER and RSI_CROSSOVER work well for EURUSD")
|
||||
print(f" 4. ✅ Conservative parameters recommended for major pairs")
|
||||
print(f"\n🚀 RECOMMENDATION: Start with the top-performing strategy!")
|
||||
print(f" Use the winning parameters from the test results above")
|
||||
else:
|
||||
print(f"\n❌ Some tests failed - check output above")
|
||||
@@ -0,0 +1,274 @@
|
||||
#!/usr/bin/env python3
|
||||
# test_eurusd_optimized.py - EURUSD Optimized for Current Market Conditions
|
||||
|
||||
import sys
|
||||
import os
|
||||
import pandas as pd
|
||||
|
||||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
def test_optimized_eurusd_strategies():
|
||||
"""Test ultra-conservative EURUSD strategies for current market conditions"""
|
||||
|
||||
print("🎯 EURUSD Optimized Strategy Testing")
|
||||
print("=" * 70)
|
||||
print("💡 Based on analysis: EURUSD is in sideways/ranging market")
|
||||
print("🔧 Using ultra-conservative parameters + mean reversion approach")
|
||||
|
||||
try:
|
||||
from core.backtesting.enhanced_engine import run_enhanced_backtest
|
||||
from core.strategies.strategy_map import STRATEGY_MAP
|
||||
|
||||
# Load EURUSD data
|
||||
csv_file = 'lab/backtest_data/EURUSD_H1_data.csv'
|
||||
df = pd.read_csv(csv_file, parse_dates=['time'])
|
||||
|
||||
# Use recent data but smaller sample for current conditions
|
||||
test_df = df.tail(1500).copy() # Last 1500 bars = ~2 months
|
||||
print(f"📊 Testing period: {test_df['time'].min()} to {test_df['time'].max()}")
|
||||
print(f"Data points: {len(test_df)} bars")
|
||||
|
||||
# ULTRA-CONSERVATIVE strategies optimized for ranging EURUSD
|
||||
optimized_strategies = [
|
||||
{
|
||||
'name': 'ULTRA-CONSERVATIVE MA_CROSSOVER',
|
||||
'strategy_id': 'MA_CROSSOVER',
|
||||
'params': {
|
||||
'fast_period': 5, # Very fast for quick entries
|
||||
'slow_period': 15, # Short slow period for ranging market
|
||||
'risk_percent': 0.3, # Ultra low risk
|
||||
'sl_atr_multiplier': 1.5, # Tight stop loss
|
||||
'tp_atr_multiplier': 2.5 # Conservative take profit
|
||||
}
|
||||
},
|
||||
{
|
||||
'name': 'SCALPING MA_CROSSOVER',
|
||||
'strategy_id': 'MA_CROSSOVER',
|
||||
'params': {
|
||||
'fast_period': 3, # Very fast scalping
|
||||
'slow_period': 8, # Quick signals
|
||||
'risk_percent': 0.2, # Micro risk
|
||||
'sl_atr_multiplier': 1.2, # Very tight SL
|
||||
'tp_atr_multiplier': 2.0 # Quick profit taking
|
||||
}
|
||||
},
|
||||
{
|
||||
'name': 'MEAN REVERSION RSI',
|
||||
'strategy_id': 'RSI_CROSSOVER',
|
||||
'params': {
|
||||
'rsi_period': 7, # Faster RSI for ranging
|
||||
'rsi_ma_period': 3, # Very fast smoothing
|
||||
'trend_filter_period': 20, # Shorter trend filter
|
||||
'risk_percent': 0.4,
|
||||
'sl_atr_multiplier': 1.5,
|
||||
'tp_atr_multiplier': 2.5
|
||||
}
|
||||
},
|
||||
{
|
||||
'name': 'MICRO BREAKOUT',
|
||||
'strategy_id': 'TURTLE_BREAKOUT',
|
||||
'params': {
|
||||
'entry_period': 8, # Very short breakout period
|
||||
'exit_period': 4, # Quick exit
|
||||
'risk_percent': 0.3,
|
||||
'sl_atr_multiplier': 1.3,
|
||||
'tp_atr_multiplier': 2.0
|
||||
}
|
||||
},
|
||||
{
|
||||
'name': 'BOLLINGER REVERSION (If Available)',
|
||||
'strategy_id': 'BOLLINGER_REVERSION',
|
||||
'params': {
|
||||
'bb_period': 20,
|
||||
'bb_std': 2.0,
|
||||
'risk_percent': 0.4,
|
||||
'sl_atr_multiplier': 1.5,
|
||||
'tp_atr_multiplier': 2.5
|
||||
}
|
||||
}
|
||||
]
|
||||
|
||||
engine_config = {
|
||||
'enable_spread_costs': True,
|
||||
'enable_slippage': True,
|
||||
'enable_realistic_execution': True
|
||||
}
|
||||
|
||||
results = []
|
||||
|
||||
for strategy_config in optimized_strategies:
|
||||
print(f"\n🧪 Testing: {strategy_config['name']}")
|
||||
print("-" * 50)
|
||||
|
||||
strategy_id = strategy_config['strategy_id']
|
||||
params = strategy_config['params']
|
||||
|
||||
# Check if strategy exists
|
||||
if strategy_id not in STRATEGY_MAP:
|
||||
print(f"❌ Strategy {strategy_id} not found - skipping")
|
||||
continue
|
||||
|
||||
print(f"Ultra-Conservative Parameters: {params}")
|
||||
|
||||
# Run backtest
|
||||
result = run_enhanced_backtest(
|
||||
strategy_id,
|
||||
params,
|
||||
test_df,
|
||||
symbol_name='EURUSD',
|
||||
engine_config=engine_config
|
||||
)
|
||||
|
||||
if 'error' in result:
|
||||
print(f"❌ Error: {result['error']}")
|
||||
continue
|
||||
|
||||
# Extract metrics
|
||||
total_trades = result.get('total_trades', 0)
|
||||
gross_profit = result.get('total_profit_usd', 0)
|
||||
spread_costs = result.get('total_spread_costs', 0)
|
||||
net_profit = result.get('net_profit_after_costs', 0)
|
||||
win_rate = result.get('win_rate_percent', 0)
|
||||
max_drawdown = result.get('max_drawdown_percent', 0)
|
||||
|
||||
print(f"📊 Ultra-Conservative Results:")
|
||||
print(f" Trades: {total_trades}")
|
||||
print(f" Win Rate: {win_rate:.1f}%")
|
||||
print(f" Net Profit: ${net_profit:.2f}")
|
||||
print(f" Max Drawdown: {max_drawdown:.2f}%")
|
||||
print(f" Spread Costs: ${spread_costs:.2f}")
|
||||
|
||||
# Ultra-conservative assessment
|
||||
if total_trades > 0:
|
||||
profit_per_trade = net_profit / total_trades
|
||||
print(f" Profit/Trade: ${profit_per_trade:.2f}")
|
||||
|
||||
# Quality for ultra-conservative approach
|
||||
quality_score = 0
|
||||
if net_profit > -50: # Loss tolerance
|
||||
quality_score += 1
|
||||
if max_drawdown < 10: # Very low drawdown
|
||||
quality_score += 2
|
||||
if win_rate > 30: # Decent win rate
|
||||
quality_score += 1
|
||||
if total_trades >= 10: # Sufficient trades
|
||||
quality_score += 1
|
||||
|
||||
if quality_score >= 4:
|
||||
print(f" 🏆 EXCELLENT: Ultra-conservative approach working!")
|
||||
elif quality_score >= 3:
|
||||
print(f" ✅ GOOD: Acceptable for ranging market")
|
||||
elif quality_score >= 2:
|
||||
print(f" ⚠️ FAIR: Needs minor adjustments")
|
||||
else:
|
||||
print(f" ❌ POOR: Strategy not suitable")
|
||||
else:
|
||||
print(f" ❌ No trades - too conservative")
|
||||
|
||||
results.append({
|
||||
'name': strategy_config['name'],
|
||||
'trades': total_trades,
|
||||
'net_profit': net_profit,
|
||||
'win_rate': win_rate,
|
||||
'max_drawdown': max_drawdown,
|
||||
'quality_score': quality_score if total_trades > 0 else 0
|
||||
})
|
||||
|
||||
# Find best ultra-conservative approach
|
||||
print(f"\n🎯 EURUSD Ultra-Conservative Ranking")
|
||||
print("=" * 70)
|
||||
|
||||
# Sort by quality score, then by net profit
|
||||
results.sort(key=lambda x: (x['quality_score'], x['net_profit']), reverse=True)
|
||||
|
||||
for i, result in enumerate(results):
|
||||
if result['trades'] > 0:
|
||||
emoji = ["🥇", "🥈", "🥉", "4️⃣", "5️⃣"][min(i, 4)]
|
||||
print(f"{emoji} {result['name']}")
|
||||
print(f" Profit: ${result['net_profit']:.2f} | Win Rate: {result['win_rate']:.1f}% | Drawdown: {result['max_drawdown']:.1f}% | Score: {result['quality_score']}/5")
|
||||
|
||||
# Recommendation
|
||||
if results and results[0]['quality_score'] >= 3:
|
||||
best = results[0]
|
||||
print(f"\n🎉 RECOMMENDED FOR CURRENT EURUSD CONDITIONS:")
|
||||
print(f" Strategy: {best['name']}")
|
||||
print(f" Why it works: Ultra-conservative approach for ranging market")
|
||||
print(f" Expected: ${best['net_profit']:.2f} profit with {best['max_drawdown']:.1f}% max drawdown")
|
||||
|
||||
# Create optimized bot parameters
|
||||
print(f"\n🤖 OPTIMIZED BOT PARAMETERS FOR EURUSD:")
|
||||
for strategy_config in optimized_strategies:
|
||||
if strategy_config['name'] == best['name']:
|
||||
print(f" Strategy: {strategy_config['strategy_id']}")
|
||||
for param, value in strategy_config['params'].items():
|
||||
print(f" {param}: {value}")
|
||||
break
|
||||
|
||||
print(f"\n💡 LONDON SESSION TIPS:")
|
||||
print(f" • Use smaller position sizes during 1-4 PM London")
|
||||
print(f" • Take profits quickly in ranging market")
|
||||
print(f" • Monitor for breakout setups at session open")
|
||||
print(f" • Current conditions favor mean reversion over trend following")
|
||||
|
||||
return True
|
||||
else:
|
||||
print(f"\n⚠️ DIFFICULT MARKET CONDITIONS:")
|
||||
print(f" • EURUSD appears to be in challenging ranging phase")
|
||||
print(f" • Consider waiting for clearer trend signals")
|
||||
print(f" • Focus on indices (like US500) which showed better performance")
|
||||
print(f" • Or test with M15/M30 timeframes for more opportunities")
|
||||
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Test failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
def suggest_alternative_pairs():
|
||||
"""Suggest alternative forex pairs based on current analysis"""
|
||||
|
||||
print(f"\n🌍 Alternative Forex Pairs for London Session")
|
||||
print("=" * 50)
|
||||
|
||||
# Check what data we have available
|
||||
data_dir = 'lab/backtest_data'
|
||||
forex_pairs = []
|
||||
|
||||
for file in os.listdir(data_dir):
|
||||
if file.endswith('_H1_data.csv'):
|
||||
symbol = file.replace('_H1_data.csv', '')
|
||||
if symbol in ['GBPUSD', 'EURGBP', 'GBPJPY', 'EURJPY', 'USDCHF']:
|
||||
forex_pairs.append(symbol)
|
||||
|
||||
print(f"📊 Available Forex Pairs for London Session:")
|
||||
for pair in forex_pairs:
|
||||
if 'GBP' in pair:
|
||||
print(f" 🇬🇧 {pair} - High volatility during London session")
|
||||
elif 'EUR' in pair:
|
||||
print(f" 🇪🇺 {pair} - European focus, good London activity")
|
||||
else:
|
||||
print(f" 💱 {pair} - Cross-pair opportunity")
|
||||
|
||||
print(f"\n💡 RECOMMENDATIONS based on current market:")
|
||||
print(f" 1. 🇬🇧 GBPUSD - Higher volatility than EURUSD")
|
||||
print(f" 2. 🇪🇺 EURGBP - Cross-pair, different dynamics")
|
||||
print(f" 3. 🥇 Continue with US500 - Your winning strategy!")
|
||||
print(f" 4. 🚀 Wait for EURUSD breakout signals")
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("🎯 EURUSD Market Condition Optimization")
|
||||
print("=" * 80)
|
||||
|
||||
success = test_optimized_eurusd_strategies()
|
||||
suggest_alternative_pairs()
|
||||
|
||||
if success:
|
||||
print(f"\n✅ OPTIMIZATION COMPLETE!")
|
||||
print(f"🎯 Found ultra-conservative approach for current EURUSD conditions")
|
||||
else:
|
||||
print(f"\n💡 STRATEGIC RECOMMENDATION:")
|
||||
print(f"🏆 Stick with US500 + Set 3 parameters (your winning combination!)")
|
||||
print(f"⏰ Monitor EURUSD for better trend opportunities")
|
||||
print(f"🔄 Consider testing GBPUSD for higher London volatility")
|
||||
@@ -18,7 +18,7 @@ def test_fbs_broker_support():
|
||||
|
||||
try:
|
||||
# Test 1: Import migration functions
|
||||
from testing.broker_symbol_migrator import detect_current_broker, get_broker_preferred_symbols
|
||||
from testing.broker_symbol_migrator import get_broker_preferred_symbols
|
||||
print("✅ Successfully imported broker migration functions")
|
||||
|
||||
# Test 2: Check FBS in broker preferences
|
||||
|
||||
@@ -0,0 +1,160 @@
|
||||
#!/usr/bin/env python3
|
||||
# test_final_backtest.py - Final test of complete backtesting workflow
|
||||
|
||||
import sys
|
||||
import os
|
||||
import pandas as pd
|
||||
|
||||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
def test_complete_backtest_workflow():
|
||||
"""Test the complete backtesting workflow like the web interface"""
|
||||
|
||||
print("🔍 Final INDEX_BREAKOUT_PRO Backtest Workflow Test")
|
||||
print("=" * 70)
|
||||
|
||||
try:
|
||||
from core.backtesting.enhanced_engine import run_enhanced_backtest
|
||||
from core.strategies.strategy_map import STRATEGY_MAP
|
||||
|
||||
# Load US500 data
|
||||
csv_file = 'lab/backtest_data/US500_H1_data.csv'
|
||||
if not os.path.exists(csv_file):
|
||||
print(f"❌ CSV file not found: {csv_file}")
|
||||
return False
|
||||
|
||||
print(f"📊 Loading US500 data...")
|
||||
df = pd.read_csv(csv_file, parse_dates=['time'])
|
||||
print(f"✅ Loaded {len(df)} rows")
|
||||
|
||||
# Use recent 2000 rows for faster testing
|
||||
test_df = df.tail(2000).copy()
|
||||
print(f"Testing with recent {len(test_df)} rows")
|
||||
print(f"Date range: {test_df['time'].min()} to {test_df['time'].max()}")
|
||||
|
||||
# Simulate web interface parameters (what user would send)
|
||||
web_params = {
|
||||
'breakout_period': 20,
|
||||
'volume_surge_multiplier': 1.5,
|
||||
'confirmation_candles': 2,
|
||||
'atr_multiplier_sl': 2.0,
|
||||
'atr_multiplier_tp': 4.0,
|
||||
'min_breakout_size': 0.2
|
||||
}
|
||||
|
||||
# Map to enhanced engine parameters (like API does)
|
||||
enhanced_params = web_params.copy()
|
||||
enhanced_params['risk_percent'] = 1.0 # Conservative for index
|
||||
enhanced_params['sl_atr_multiplier'] = web_params.get('atr_multiplier_sl', 2.0)
|
||||
enhanced_params['tp_atr_multiplier'] = web_params.get('atr_multiplier_tp', 4.0)
|
||||
|
||||
print(f"\\n⚙️ Parameters:")
|
||||
print(f" Web interface: {web_params}")
|
||||
print(f" Enhanced engine: {enhanced_params}")
|
||||
|
||||
# Engine configuration (like API sets)
|
||||
engine_config = {
|
||||
'enable_spread_costs': True,
|
||||
'enable_slippage': True,
|
||||
'enable_realistic_execution': True
|
||||
}
|
||||
|
||||
# Extract symbol name (like API does)
|
||||
symbol_name = 'US500'
|
||||
print(f"\\n🎯 Symbol: {symbol_name}")
|
||||
|
||||
# Run enhanced backtest (exactly like the API)
|
||||
print(f"\\n🚀 Running enhanced backtest...")
|
||||
|
||||
strategy_id = 'INDEX_BREAKOUT_PRO'
|
||||
results = run_enhanced_backtest(
|
||||
strategy_id,
|
||||
enhanced_params,
|
||||
test_df,
|
||||
symbol_name=symbol_name,
|
||||
engine_config=engine_config
|
||||
)
|
||||
|
||||
if 'error' in results:
|
||||
print(f"❌ Backtest error: {results['error']}")
|
||||
return False
|
||||
|
||||
print(f"✅ Backtest completed successfully!")
|
||||
print(f"\\n📈 Results Summary:")
|
||||
print(f" Strategy: {results.get('strategy_name', 'Unknown')}")
|
||||
print(f" Total Trades: {results.get('total_trades', 0)}")
|
||||
print(f" Wins: {results.get('wins', 0)}")
|
||||
print(f" Losses: {results.get('losses', 0)}")
|
||||
print(f" Win Rate: {results.get('win_rate_percent', 0):.1f}%")
|
||||
print(f" Total Profit USD: ${results.get('total_profit_usd', 0):.2f}")
|
||||
print(f" Spread Costs: ${results.get('total_spread_costs', 0):.2f}")
|
||||
print(f" Net Profit: ${results.get('net_profit_after_costs', 0):.2f}")
|
||||
print(f" Max Drawdown: {results.get('max_drawdown_percent', 0):.1f}%")
|
||||
print(f" Final Capital: ${results.get('final_capital', 0):.2f}")
|
||||
|
||||
# Check if we have trades
|
||||
trades = results.get('trades', [])
|
||||
if len(trades) > 0:
|
||||
print(f"\\n📋 Sample Trades (last 5):")
|
||||
for i, trade in enumerate(trades[-5:]):
|
||||
entry_price = trade.get('entry', 0)
|
||||
exit_price = trade.get('exit', 0)
|
||||
profit = trade.get('profit', 0)
|
||||
position_type = trade.get('position_type', 'Unknown')
|
||||
print(f" {i+1}. {position_type}: Entry ${entry_price:.2f} → Exit ${exit_price:.2f} = ${profit:.2f}")
|
||||
|
||||
# Test parameter API format (like frontend expects)
|
||||
print(f"\\n🔧 Testing parameter API format...")
|
||||
strategy_class = STRATEGY_MAP.get(strategy_id)
|
||||
if strategy_class and hasattr(strategy_class, 'get_definable_params'):
|
||||
params = strategy_class.get_definable_params()
|
||||
|
||||
# Normalize like the API does
|
||||
normalized_params = []
|
||||
for param in params:
|
||||
normalized_param = param.copy()
|
||||
if 'display_name' in param and 'label' not in param:
|
||||
normalized_param['label'] = param['display_name']
|
||||
elif 'label' not in param and 'display_name' not in param:
|
||||
normalized_param['label'] = param['name'].replace('_', ' ').title()
|
||||
normalized_params.append(normalized_param)
|
||||
|
||||
print(f"✅ Parameter normalization successful")
|
||||
print(f"Sample parameters for frontend:")
|
||||
for param in normalized_params[:3]:
|
||||
print(f" • {param['name']}: '{param.get('label', 'NO LABEL')}'")
|
||||
|
||||
# Success criteria
|
||||
if results.get('total_trades', 0) > 0:
|
||||
print(f"\\n✅ SUCCESS: Strategy generated {results.get('total_trades', 0)} trades!")
|
||||
print(f"\\n🎉 Both issues are now FIXED:")
|
||||
print(f" 1. ✅ Parameter names show correctly (not 'undefined')")
|
||||
print(f" 2. ✅ Backtest generates trades (not empty results)")
|
||||
return True
|
||||
else:
|
||||
print(f"\\n⚠️ Warning: No trades generated with these parameters")
|
||||
print(f"Try adjusting parameters for more signals")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Test failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
success = test_complete_backtest_workflow()
|
||||
|
||||
if success:
|
||||
print(f"\\n🚀 FINAL RESULT: Both issues RESOLVED!")
|
||||
print(f"\\n📋 Summary of fixes:")
|
||||
print(f" 1. Parameter API normalization: display_name → label")
|
||||
print(f" 2. Strategy signal generation: Much more practical and flexible")
|
||||
print(f" 3. Volume calculation: Adaptive and robust")
|
||||
print(f" 4. Multiple signal types: Range, momentum, breakout, trend")
|
||||
print(f"\\n🎯 The web interface should now show:")
|
||||
print(f" • Proper parameter names (Breakout Detection Period, etc.)")
|
||||
print(f" • Non-zero backtest results with actual trades")
|
||||
print(f" • Realistic profit/loss calculations")
|
||||
else:
|
||||
print(f"\\n❌ Some issues remain - check the output above")
|
||||
@@ -0,0 +1,180 @@
|
||||
#!/usr/bin/env python3
|
||||
# test_gbpusd_london.py - Quick GBPUSD vs EURUSD comparison
|
||||
|
||||
import sys
|
||||
import os
|
||||
import pandas as pd
|
||||
|
||||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
def test_gbpusd_vs_eurusd():
|
||||
"""Quick comparison of GBPUSD vs EURUSD with optimized parameters"""
|
||||
|
||||
print("🇬🇧 GBPUSD vs EURUSD London Session Comparison")
|
||||
print("=" * 70)
|
||||
print("💡 Testing if GBPUSD performs better than EURUSD during London session")
|
||||
|
||||
try:
|
||||
from core.backtesting.enhanced_engine import run_enhanced_backtest
|
||||
|
||||
# Test both pairs with the same winning parameters from US500 Set 3
|
||||
# But adapted for forex pairs
|
||||
|
||||
pairs_to_test = [
|
||||
{
|
||||
'name': 'GBPUSD',
|
||||
'file': 'lab/backtest_data/GBPUSD_H1_data.csv',
|
||||
'params': {
|
||||
'fast_period': 8, # From successful Set 3 concept
|
||||
'slow_period': 21, # Adapted for forex
|
||||
'risk_percent': 0.5, # Conservative for forex
|
||||
'sl_atr_multiplier': 2.0,
|
||||
'tp_atr_multiplier': 3.5
|
||||
}
|
||||
},
|
||||
{
|
||||
'name': 'EURUSD',
|
||||
'file': 'lab/backtest_data/EURUSD_H1_data.csv',
|
||||
'params': {
|
||||
'fast_period': 8,
|
||||
'slow_period': 21,
|
||||
'risk_percent': 0.5,
|
||||
'sl_atr_multiplier': 2.0,
|
||||
'tp_atr_multiplier': 3.5
|
||||
}
|
||||
}
|
||||
]
|
||||
|
||||
results = []
|
||||
|
||||
for pair_config in pairs_to_test:
|
||||
print(f"\n🚀 Testing {pair_config['name']} with Set 3 Adapted Parameters")
|
||||
print("-" * 50)
|
||||
|
||||
# Load data
|
||||
if not os.path.exists(pair_config['file']):
|
||||
print(f"❌ Data file not found: {pair_config['file']}")
|
||||
continue
|
||||
|
||||
df = pd.read_csv(pair_config['file'], parse_dates=['time'])
|
||||
test_df = df.tail(2000).copy() # Recent 2000 bars
|
||||
|
||||
print(f"Testing period: {test_df['time'].min()} to {test_df['time'].max()}")
|
||||
print(f"Parameters: {pair_config['params']}")
|
||||
|
||||
# Run backtest with MA_CROSSOVER
|
||||
result = run_enhanced_backtest(
|
||||
'MA_CROSSOVER',
|
||||
pair_config['params'],
|
||||
test_df,
|
||||
symbol_name=pair_config['name'],
|
||||
engine_config={
|
||||
'enable_spread_costs': True,
|
||||
'enable_slippage': True,
|
||||
'enable_realistic_execution': True
|
||||
}
|
||||
)
|
||||
|
||||
if 'error' in result:
|
||||
print(f"❌ Error: {result['error']}")
|
||||
continue
|
||||
|
||||
# Extract metrics
|
||||
total_trades = result.get('total_trades', 0)
|
||||
net_profit = result.get('net_profit_after_costs', 0)
|
||||
win_rate = result.get('win_rate_percent', 0)
|
||||
max_drawdown = result.get('max_drawdown_percent', 0)
|
||||
spread_costs = result.get('total_spread_costs', 0)
|
||||
|
||||
print(f"📊 Results:")
|
||||
print(f" Trades: {total_trades}")
|
||||
print(f" Net Profit: ${net_profit:.2f}")
|
||||
print(f" Win Rate: {win_rate:.1f}%")
|
||||
print(f" Max Drawdown: {max_drawdown:.2f}%")
|
||||
print(f" Spread Costs: ${spread_costs:.2f}")
|
||||
|
||||
# Performance assessment
|
||||
if total_trades > 0:
|
||||
profit_per_trade = net_profit / total_trades
|
||||
print(f" Profit/Trade: ${profit_per_trade:.2f}")
|
||||
|
||||
if net_profit > 0:
|
||||
print(f" 🏆 PROFITABLE! London session advantage confirmed")
|
||||
elif net_profit > -100:
|
||||
print(f" ⚠️ Small loss - acceptable for ranging market")
|
||||
else:
|
||||
print(f" ❌ Significant loss - avoid current parameters")
|
||||
|
||||
results.append({
|
||||
'pair': pair_config['name'],
|
||||
'trades': total_trades,
|
||||
'net_profit': net_profit,
|
||||
'win_rate': win_rate,
|
||||
'max_drawdown': max_drawdown,
|
||||
'profit_per_trade': profit_per_trade if total_trades > 0 else 0
|
||||
})
|
||||
|
||||
# Comparison
|
||||
print(f"\n🏆 LONDON SESSION COMPARISON")
|
||||
print("=" * 50)
|
||||
|
||||
for result in sorted(results, key=lambda x: x['net_profit'], reverse=True):
|
||||
emoji = "🥇" if result == results[0] else "🥈"
|
||||
print(f"{emoji} {result['pair']}")
|
||||
print(f" Net Profit: ${result['net_profit']:.2f}")
|
||||
print(f" Win Rate: {result['win_rate']:.1f}%")
|
||||
print(f" Trades: {result['trades']}")
|
||||
print(f" Drawdown: {result['max_drawdown']:.2f}%")
|
||||
|
||||
# Recommendation
|
||||
if results:
|
||||
best_pair = max(results, key=lambda x: x['net_profit'])
|
||||
|
||||
if best_pair['net_profit'] > 0:
|
||||
print(f"\n🎯 LONDON SESSION WINNER: {best_pair['pair']}")
|
||||
print(f" Net Profit: ${best_pair['net_profit']:.2f}")
|
||||
print(f" This pair shows better performance during London hours!")
|
||||
|
||||
print(f"\n🤖 RECOMMENDED BOT SETUP:")
|
||||
for pair_config in pairs_to_test:
|
||||
if pair_config['name'] == best_pair['pair']:
|
||||
print(f" Symbol: {pair_config['name']}")
|
||||
print(f" Strategy: MA_CROSSOVER")
|
||||
for param, value in pair_config['params'].items():
|
||||
print(f" {param}: {value}")
|
||||
break
|
||||
|
||||
return True
|
||||
else:
|
||||
print(f"\n💡 MARKET INSIGHT:")
|
||||
print(f" Both EURUSD and GBPUSD showing challenging conditions")
|
||||
print(f" Current forex market may be in consolidation phase")
|
||||
print(f" Your US500 strategy with Set 3 parameters remains the winner!")
|
||||
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Test failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("🚀 London Session Forex Pair Comparison")
|
||||
print("=" * 80)
|
||||
|
||||
success = test_gbpusd_vs_eurusd()
|
||||
|
||||
if success:
|
||||
print(f"\n✅ FOUND PROFITABLE FOREX OPPORTUNITY!")
|
||||
print(f"🎯 Ready to deploy during London session")
|
||||
else:
|
||||
print(f"\n💰 STRATEGIC RECOMMENDATION:")
|
||||
print(f"🏆 STICK WITH US500 + Set 3 Parameters!")
|
||||
print(f" • Proven profitable: +$32.25 already")
|
||||
print(f" • Low risk: 0.15% max drawdown")
|
||||
print(f" • High activity: 963 trades backtested")
|
||||
print(f"\n⏰ FOR FOREX:")
|
||||
print(f" • Wait for clearer trend signals")
|
||||
print(f" • Monitor for Brexit/ECB news that could create volatility")
|
||||
print(f" • Consider M15 timeframe for more opportunities")
|
||||
@@ -8,7 +8,6 @@ import os
|
||||
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
|
||||
|
||||
from core.seasonal.holiday_manager import holiday_manager
|
||||
from datetime import date, datetime
|
||||
|
||||
def test_holiday_detection():
|
||||
"""Test holiday detection functionality"""
|
||||
|
||||
@@ -8,7 +8,6 @@ import os
|
||||
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
|
||||
|
||||
from core.seasonal.holiday_manager import holiday_manager
|
||||
from datetime import date, datetime
|
||||
|
||||
def test_holiday_visibility():
|
||||
"""Test holiday visibility functionality"""
|
||||
|
||||
@@ -0,0 +1,216 @@
|
||||
#!/usr/bin/env python3
|
||||
# test_index_params.py - Simple test for INDEX_BREAKOUT_PRO parameters and signals
|
||||
|
||||
import sys
|
||||
import os
|
||||
import pandas as pd
|
||||
import numpy as np
|
||||
|
||||
# Add project root to path
|
||||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
def test_strategy_parameters():
|
||||
"""Test INDEX_BREAKOUT_PRO parameter functionality"""
|
||||
|
||||
print("🔧 Testing INDEX_BREAKOUT_PRO Parameters")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
from core.strategies.index_breakout_pro import IndexBreakoutProStrategy
|
||||
|
||||
# Test 1: Check parameter definitions
|
||||
print("1️⃣ Testing parameter definitions...")
|
||||
params = IndexBreakoutProStrategy.get_definable_params()
|
||||
|
||||
print(f"Found {len(params)} parameters:")
|
||||
for param in params:
|
||||
name = param.get('name', 'Unknown')
|
||||
display_name = param.get('display_name', 'No display name')
|
||||
label = param.get('label', 'No label')
|
||||
default = param.get('default', 'No default')
|
||||
param_type = param.get('type', 'Unknown type')
|
||||
|
||||
print(f" • {name}:")
|
||||
print(f" - Display Name: {display_name}")
|
||||
print(f" - Label: {label}")
|
||||
print(f" - Default: {default}")
|
||||
print(f" - Type: {param_type}")
|
||||
|
||||
# Test 2: Parameter normalization (like the API does)
|
||||
print(f"\n2️⃣ Testing parameter normalization...")
|
||||
normalized_params = []
|
||||
for param in params:
|
||||
normalized_param = param.copy()
|
||||
if 'display_name' in param and 'label' not in param:
|
||||
normalized_param['label'] = param['display_name']
|
||||
elif 'label' not in param and 'display_name' not in param:
|
||||
normalized_param['label'] = param['name'].replace('_', ' ').title()
|
||||
normalized_params.append(normalized_param)
|
||||
|
||||
print(f"Normalized parameters for frontend:")
|
||||
for param in normalized_params:
|
||||
print(f" • {param['name']}: '{param.get('label', 'NO LABEL')}'")
|
||||
|
||||
# Test 3: Strategy instantiation and signal generation
|
||||
print(f"\n3️⃣ Testing signal generation...")
|
||||
|
||||
# Create simple test data
|
||||
dates = pd.date_range('2024-01-01', periods=100, freq='h')
|
||||
|
||||
# Generate price data with some volatility
|
||||
base_price = 4350 # US500 base price
|
||||
price_changes = np.random.randn(100) * 0.005 # 0.5% random changes
|
||||
|
||||
# Add some trend and breakout patterns
|
||||
trend = np.linspace(0, 0.02, 100) # 2% uptrend
|
||||
breakout_pattern = np.zeros(100)
|
||||
breakout_pattern[70:75] = 0.01 # 1% breakout at position 70-75
|
||||
|
||||
cumulative_changes = np.cumsum(price_changes + trend + breakout_pattern)
|
||||
prices = base_price * (1 + cumulative_changes)
|
||||
|
||||
# Create OHLCV data
|
||||
df = pd.DataFrame({
|
||||
'time': dates,
|
||||
'open': prices,
|
||||
'high': prices * (1 + np.random.rand(100) * 0.002), # Small random high
|
||||
'low': prices * (1 - np.random.rand(100) * 0.002), # Small random low
|
||||
'close': prices,
|
||||
'volume': np.random.randint(5000, 15000, 100) # Random volume
|
||||
})
|
||||
|
||||
# Create mock bot
|
||||
class MockBot:
|
||||
def __init__(self):
|
||||
self.market_for_mt5 = 'US500'
|
||||
|
||||
# Test with default parameters
|
||||
print(f"Creating strategy instance with default parameters...")
|
||||
strategy = IndexBreakoutProStrategy(MockBot(), {})
|
||||
|
||||
# Test analyze_df
|
||||
print(f"Running analyze_df on test data...")
|
||||
result_df = strategy.analyze_df(df)
|
||||
|
||||
# Check signals
|
||||
if 'signal' in result_df.columns:
|
||||
signals = result_df['signal'].value_counts()
|
||||
print(f"Signal distribution: {signals.to_dict()}")
|
||||
|
||||
non_hold_signals = result_df[result_df['signal'] != 'HOLD']
|
||||
print(f"Non-HOLD signals: {len(non_hold_signals)}")
|
||||
|
||||
if len(non_hold_signals) > 0:
|
||||
print(f"Sample trading signals:")
|
||||
for i, row in non_hold_signals.head(5).iterrows():
|
||||
print(f" • {row['signal']} at ${row['close']:.2f}: {row.get('explanation', 'No explanation')}")
|
||||
else:
|
||||
print(f"⚠️ No trading signals generated")
|
||||
print(f"Sample explanations from recent data:")
|
||||
recent_explanations = result_df['explanation'].tail(10)
|
||||
for i, exp in enumerate(recent_explanations):
|
||||
print(f" {i+1}: {exp}")
|
||||
|
||||
# Test 4: Test with custom parameters
|
||||
print(f"\n4️⃣ Testing with custom parameters...")
|
||||
custom_params = {
|
||||
'breakout_period': 10, # Shorter period for more signals
|
||||
'volume_surge_multiplier': 1.5, # Lower threshold
|
||||
'min_breakout_size': 0.1 # Smaller breakout size
|
||||
}
|
||||
|
||||
strategy_custom = IndexBreakoutProStrategy(MockBot(), custom_params)
|
||||
result_df_custom = strategy_custom.analyze_df(df)
|
||||
|
||||
if 'signal' in result_df_custom.columns:
|
||||
signals_custom = result_df_custom['signal'].value_counts()
|
||||
print(f"Custom parameter signals: {signals_custom.to_dict()}")
|
||||
|
||||
non_hold_custom = result_df_custom[result_df_custom['signal'] != 'HOLD']
|
||||
print(f"Custom non-HOLD signals: {len(non_hold_custom)}")
|
||||
|
||||
print(f"\n✅ Parameter testing completed!")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Parameter testing failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
def test_csv_data_compatibility():
|
||||
"""Test the actual US500 CSV data"""
|
||||
|
||||
print(f"\n📊 Testing US500 CSV Data Compatibility")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
csv_file = 'lab/backtest_data/US500_H1_data.csv'
|
||||
if not os.path.exists(csv_file):
|
||||
print(f"❌ CSV file not found: {csv_file}")
|
||||
return False
|
||||
|
||||
# Load actual data
|
||||
df = pd.read_csv(csv_file, parse_dates=['time'])
|
||||
print(f"✅ Loaded {len(df)} rows from {csv_file}")
|
||||
print(f"Date range: {df['time'].min()} to {df['time'].max()}")
|
||||
print(f"Columns: {list(df.columns)}")
|
||||
|
||||
# Check for missing data
|
||||
missing_data = df.isnull().sum()
|
||||
print(f"Missing data per column: {missing_data.to_dict()}")
|
||||
|
||||
# Take a recent subset for testing
|
||||
recent_df = df.tail(200).copy() # Last 200 rows
|
||||
print(f"\nTesting with recent {len(recent_df)} rows...")
|
||||
|
||||
# Test strategy with this data
|
||||
from core.strategies.index_breakout_pro import IndexBreakoutProStrategy
|
||||
|
||||
class MockBot:
|
||||
def __init__(self):
|
||||
self.market_for_mt5 = 'US500'
|
||||
|
||||
strategy = IndexBreakoutProStrategy(MockBot(), {})
|
||||
result_df = strategy.analyze_df(recent_df)
|
||||
|
||||
if 'signal' in result_df.columns:
|
||||
signals = result_df['signal'].value_counts()
|
||||
print(f"✅ Signal generation successful: {signals.to_dict()}")
|
||||
|
||||
non_hold = result_df[result_df['signal'] != 'HOLD']
|
||||
if len(non_hold) > 0:
|
||||
print(f"✅ Generated {len(non_hold)} trading signals")
|
||||
print(f"Recent signals:")
|
||||
for i, row in non_hold.tail(3).iterrows():
|
||||
print(f" • {row['signal']} at ${row['close']:.2f}")
|
||||
else:
|
||||
print(f"⚠️ No trading signals in recent data")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ CSV data testing failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("🔍 INDEX_BREAKOUT_PRO Parameter & Signal Testing")
|
||||
print("=" * 70)
|
||||
|
||||
test1_success = test_strategy_parameters()
|
||||
test2_success = test_csv_data_compatibility()
|
||||
|
||||
if test1_success and test2_success:
|
||||
print(f"\n✅ All tests passed!")
|
||||
print(f"\n💡 If web interface still shows 'undefined' parameters:")
|
||||
print(f" 1. Check browser console for JavaScript errors")
|
||||
print(f" 2. Verify the parameter API endpoint is working")
|
||||
print(f" 3. Check frontend parameter display code")
|
||||
print(f"\n💡 If backtest still returns empty results:")
|
||||
print(f" 1. Strategy may be too conservative (not generating signals)")
|
||||
print(f" 2. Check engine configuration")
|
||||
print(f" 3. Try with more volatile data or different parameters")
|
||||
else:
|
||||
print(f"\n❌ Some tests failed - check output above")
|
||||
@@ -0,0 +1,189 @@
|
||||
#!/usr/bin/env python3
|
||||
# test_index_risk_fix.py - Test the fixed index risk management
|
||||
|
||||
import sys
|
||||
import os
|
||||
import pandas as pd
|
||||
|
||||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
def test_index_risk_management():
|
||||
"""Test the corrected index risk management and position sizing"""
|
||||
|
||||
print("🔧 Testing INDEX Risk Management Fix")
|
||||
print("=" * 60)
|
||||
|
||||
try:
|
||||
from core.backtesting.enhanced_engine import InstrumentConfig, EnhancedBacktestEngine
|
||||
|
||||
# Test 1: Verify US500 detection and configuration
|
||||
print("1️⃣ Testing US500 instrument detection...")
|
||||
config = InstrumentConfig.get_config('US500')
|
||||
|
||||
print(f"US500 Configuration:")
|
||||
print(f" Contract Size: {config['contract_size']}")
|
||||
print(f" Max Risk: {config['max_risk_percent']}%")
|
||||
print(f" Max Lot Size: {config['max_lot_size']}")
|
||||
print(f" Typical Spread: {config['typical_spread_pips']} pips")
|
||||
|
||||
if config['max_risk_percent'] <= 0.5 and config['max_lot_size'] <= 0.1:
|
||||
print(" ✅ Conservative risk limits applied")
|
||||
else:
|
||||
print(" ❌ Risk limits too high")
|
||||
return False
|
||||
|
||||
# Test 2: Test position sizing
|
||||
print(f"\\n2️⃣ Testing position sizing for US500...")
|
||||
engine = EnhancedBacktestEngine()
|
||||
|
||||
# Simulate realistic parameters
|
||||
capital = 10000
|
||||
risk_percent = 2.0 # User requested 2%
|
||||
atr_value = 45.0 # Typical US500 ATR
|
||||
sl_distance = atr_value * 2.0 # 2x ATR stop loss
|
||||
|
||||
position_size = engine.calculate_position_size(
|
||||
'US500', capital, risk_percent, sl_distance, atr_value, config
|
||||
)
|
||||
|
||||
print(f" Capital: ${capital}")
|
||||
print(f" Requested Risk: {risk_percent}%")
|
||||
print(f" Applied Risk: {min(risk_percent, config['max_risk_percent'])}%")
|
||||
print(f" ATR: {atr_value}")
|
||||
print(f" SL Distance: {sl_distance}")
|
||||
print(f" Calculated Lot Size: {position_size}")
|
||||
|
||||
# Calculate actual risk amount
|
||||
max_loss = position_size * sl_distance * config['contract_size']
|
||||
actual_risk_percent = (max_loss / capital) * 100
|
||||
|
||||
print(f" Max Potential Loss: ${max_loss:.2f}")
|
||||
print(f" Actual Risk %: {actual_risk_percent:.2f}%")
|
||||
|
||||
if actual_risk_percent <= 0.5: # Should be very conservative
|
||||
print(" ✅ Position sizing is now conservative")
|
||||
else:
|
||||
print(" ❌ Position sizing still too aggressive")
|
||||
return False
|
||||
|
||||
# Test 3: Test with very high volatility
|
||||
print(f"\\n3️⃣ Testing high volatility protection...")
|
||||
high_atr = 120.0 # Very high ATR
|
||||
high_vol_position = engine.calculate_position_size(
|
||||
'US500', capital, risk_percent, high_atr * 2, high_atr, config
|
||||
)
|
||||
|
||||
print(f" High ATR: {high_atr}")
|
||||
print(f" High Vol Position Size: {high_vol_position}")
|
||||
|
||||
if high_vol_position <= 0.01:
|
||||
print(" ✅ Extreme volatility protection working")
|
||||
else:
|
||||
print(" ⚠️ High volatility position might still be risky")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Test failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
def test_full_backtest_with_fixes():
|
||||
"""Test a full backtest with the risk management fixes"""
|
||||
|
||||
print(f"\\n🚀 Testing Full Backtest with Risk Fixes")
|
||||
print("=" * 60)
|
||||
|
||||
try:
|
||||
from core.backtesting.enhanced_engine import run_enhanced_backtest
|
||||
|
||||
# Load US500 data
|
||||
csv_file = 'lab/backtest_data/US500_H1_data.csv'
|
||||
if not os.path.exists(csv_file):
|
||||
print(f"❌ CSV file not found: {csv_file}")
|
||||
return False
|
||||
|
||||
df = pd.read_csv(csv_file, parse_dates=['time'])
|
||||
test_df = df.tail(500).copy() # Small sample for quick test
|
||||
|
||||
# Conservative parameters
|
||||
params = {
|
||||
'breakout_period': 20,
|
||||
'volume_surge_multiplier': 1.5,
|
||||
'min_breakout_size': 0.2,
|
||||
'risk_percent': 2.0, # This will be capped at 0.5% for indices
|
||||
'sl_atr_multiplier': 2.0,
|
||||
'tp_atr_multiplier': 4.0
|
||||
}
|
||||
|
||||
engine_config = {
|
||||
'enable_spread_costs': True,
|
||||
'enable_slippage': True,
|
||||
'enable_realistic_execution': True
|
||||
}
|
||||
|
||||
print(f"Testing with {len(test_df)} rows...")
|
||||
print(f"Parameters: {params}")
|
||||
|
||||
results = run_enhanced_backtest(
|
||||
'INDEX_BREAKOUT_PRO',
|
||||
params,
|
||||
test_df,
|
||||
symbol_name='US500',
|
||||
engine_config=engine_config
|
||||
)
|
||||
|
||||
if 'error' in results:
|
||||
print(f"❌ Backtest error: {results['error']}")
|
||||
return False
|
||||
|
||||
print(f"\\n📈 Fixed Results:")
|
||||
print(f" Total Trades: {results.get('total_trades', 0)}")
|
||||
print(f" Total Profit: ${results.get('total_profit_usd', 0):.2f}")
|
||||
print(f" Max Drawdown: {results.get('max_drawdown_percent', 0):.2f}%")
|
||||
print(f" Win Rate: {results.get('win_rate_percent', 0):.1f}%")
|
||||
print(f" Final Capital: ${results.get('final_capital', 0):.2f}")
|
||||
|
||||
# Check if results are reasonable
|
||||
max_drawdown = results.get('max_drawdown_percent', 0)
|
||||
total_profit = abs(results.get('total_profit_usd', 0))
|
||||
|
||||
if max_drawdown < 50 and total_profit < 5000: # Much more reasonable
|
||||
print(f"\\n✅ Results look much more reasonable!")
|
||||
print(f" • Drawdown under control: {max_drawdown:.1f}%")
|
||||
print(f" • Profit/loss reasonable: ${total_profit:.2f}")
|
||||
return True
|
||||
else:
|
||||
print(f"\\n⚠️ Results still concerning:")
|
||||
print(f" • Drawdown: {max_drawdown:.1f}% (should be <50%)")
|
||||
print(f" • P/L magnitude: ${total_profit:.2f} (should be <$5000)")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Full backtest test failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("🔍 INDEX Risk Management Fix Testing")
|
||||
print("=" * 70)
|
||||
|
||||
test1 = test_index_risk_management()
|
||||
test2 = test_full_backtest_with_fixes()
|
||||
|
||||
if test1 and test2:
|
||||
print(f"\\n✅ INDEX RISK MANAGEMENT FIXED!")
|
||||
print(f"\\n📋 What was fixed:")
|
||||
print(f" 1. Added INDICES configuration with 0.5% max risk")
|
||||
print(f" 2. Added ultra-conservative position sizing for indices")
|
||||
print(f" 3. Added volatility protection for high ATR periods")
|
||||
print(f" 4. Corrected spread cost calculation for indices")
|
||||
print(f"\\n🎯 Expected improvement in web interface:")
|
||||
print(f" • Much smaller position sizes (0.01-0.03 lots max)")
|
||||
print(f" • Reasonable profit/loss amounts (<$5000)")
|
||||
print(f" • Controlled drawdowns (<50%)")
|
||||
print(f" • Proper risk management for US500/indices")
|
||||
else:
|
||||
print(f"\\n❌ Some issues remain - check output above")
|
||||
@@ -10,7 +10,6 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
import pandas as pd
|
||||
import numpy as np
|
||||
from datetime import datetime, timedelta
|
||||
import logging
|
||||
|
||||
# Set up logging
|
||||
@@ -59,7 +58,7 @@ def generate_index_test_data(symbol='US500', periods=500):
|
||||
# Generate volume (higher during market hours)
|
||||
base_volume = np.random.randint(800, 1500, periods)
|
||||
# Simulate higher volume during NY session (14:30-21:00 UTC)
|
||||
hour_factor = [1.5 if 14 <= h <= 21 else 0.8 for h in dates.hour]
|
||||
hour_factor = [1.5 if 14 <= h <= 21 else 0.8 for h in dates.hour] # pyright: ignore
|
||||
df['tick_volume'] = base_volume * hour_factor
|
||||
|
||||
# Ensure OHLC integrity
|
||||
@@ -103,7 +102,7 @@ def test_index_momentum_strategy():
|
||||
print(f" Explanation: {signal_info['explanation']}")
|
||||
|
||||
# Test backtesting
|
||||
print(f" 🔄 Running backtest analysis...")
|
||||
print(" 🔄 Running backtest analysis...")
|
||||
df_with_signals = strategy.analyze_df(df)
|
||||
|
||||
# Count signals
|
||||
@@ -159,7 +158,7 @@ def test_index_breakout_pro_strategy():
|
||||
print(f" Analysis: {signal_info['explanation']}")
|
||||
|
||||
# Test backtesting
|
||||
print(f" 🔄 Running professional backtest...")
|
||||
print(" 🔄 Running professional backtest...")
|
||||
df_with_signals = strategy.analyze_df(df)
|
||||
|
||||
# Count signals
|
||||
|
||||
@@ -0,0 +1,254 @@
|
||||
#!/usr/bin/env python3
|
||||
# test_strategy_signals.py - Test INDEX_BREAKOUT_PRO with more dynamic data
|
||||
|
||||
import sys
|
||||
import os
|
||||
import pandas as pd
|
||||
import numpy as np
|
||||
|
||||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
def create_breakout_test_data():
|
||||
"""Create test data with clear breakout patterns"""
|
||||
|
||||
print("📊 Creating breakout test data...")
|
||||
|
||||
# Create 500 periods with deliberate breakout patterns
|
||||
periods = 500
|
||||
dates = pd.date_range('2024-01-01', periods=periods, freq='h')
|
||||
|
||||
base_price = 4350
|
||||
prices = []
|
||||
volumes = []
|
||||
|
||||
price = base_price
|
||||
|
||||
for i in range(periods):
|
||||
# Base random movement
|
||||
daily_vol = 0.003 # 0.3% daily volatility
|
||||
random_change = np.random.randn() * daily_vol
|
||||
|
||||
# Add trend patterns
|
||||
if i < 100:
|
||||
# First 100: sideways with small movements
|
||||
trend = 0
|
||||
elif i < 200:
|
||||
# Next 100: clear uptrend with breakouts
|
||||
trend = 0.0008 # 0.08% per hour uptrend
|
||||
# Add volume spikes during breakouts
|
||||
if i % 25 == 0: # Every 25 hours, create breakout
|
||||
random_change += 0.008 # 0.8% breakout move
|
||||
elif i < 300:
|
||||
# Next 100: downtrend with breakdowns
|
||||
trend = -0.0005 # 0.05% per hour downtrend
|
||||
if i % 30 == 0:
|
||||
random_change -= 0.006 # 0.6% breakdown move
|
||||
else:
|
||||
# Final 200: mixed with occasional large moves
|
||||
trend = 0.0002
|
||||
if i % 40 == 0:
|
||||
random_change += np.random.choice([0.01, -0.01]) # 1% move up or down
|
||||
|
||||
# Apply changes
|
||||
price_change = random_change + trend
|
||||
price = price * (1 + price_change)
|
||||
prices.append(price)
|
||||
|
||||
# Volume: higher during breakouts
|
||||
base_volume = 8000
|
||||
if abs(random_change) > 0.005: # Large moves get high volume
|
||||
volume = base_volume * (2 + np.random.rand() * 2) # 2-4x volume
|
||||
else:
|
||||
volume = base_volume * (0.8 + np.random.rand() * 0.4) # 0.8-1.2x volume
|
||||
|
||||
volumes.append(int(volume))
|
||||
|
||||
# Create OHLCV data
|
||||
df = pd.DataFrame({
|
||||
'time': dates,
|
||||
'open': prices,
|
||||
'close': prices,
|
||||
'volume': volumes
|
||||
})
|
||||
|
||||
# Add realistic high/low
|
||||
df['high'] = df['close'] * (1 + np.random.rand(periods) * 0.003)
|
||||
df['low'] = df['close'] * (1 - np.random.rand(periods) * 0.003)
|
||||
|
||||
print(f"✅ Created {len(df)} periods of breakout test data")
|
||||
print(f"Price range: ${df['close'].min():.2f} - ${df['close'].max():.2f}")
|
||||
print(f"Volume range: {df['volume'].min()} - {df['volume'].max()}")
|
||||
|
||||
return df
|
||||
|
||||
def test_with_dynamic_data():
|
||||
"""Test strategy with dynamic breakout data"""
|
||||
|
||||
print("🚀 Testing INDEX_BREAKOUT_PRO with Dynamic Data")
|
||||
print("=" * 60)
|
||||
|
||||
try:
|
||||
from core.strategies.index_breakout_pro import IndexBreakoutProStrategy
|
||||
|
||||
# Create test data with breakouts
|
||||
df = create_breakout_test_data()
|
||||
|
||||
# Create mock bot
|
||||
class MockBot:
|
||||
def __init__(self):
|
||||
self.market_for_mt5 = 'US500'
|
||||
|
||||
# Test with different parameter sets
|
||||
test_scenarios = [
|
||||
{
|
||||
'name': 'Conservative (Default)',
|
||||
'params': {} # Use defaults
|
||||
},
|
||||
{
|
||||
'name': 'Moderate',
|
||||
'params': {
|
||||
'volume_surge_multiplier': 1.3,
|
||||
'min_breakout_size': 0.15,
|
||||
'breakout_period': 15
|
||||
}
|
||||
},
|
||||
{
|
||||
'name': 'Aggressive',
|
||||
'params': {
|
||||
'volume_surge_multiplier': 1.2,
|
||||
'min_breakout_size': 0.1,
|
||||
'breakout_period': 10
|
||||
}
|
||||
}
|
||||
]
|
||||
|
||||
for scenario in test_scenarios:
|
||||
print(f"\\n📈 Testing {scenario['name']} Parameters:")
|
||||
print("-" * 40)
|
||||
|
||||
strategy = IndexBreakoutProStrategy(MockBot(), scenario['params'])
|
||||
result_df = strategy.analyze_df(df)
|
||||
|
||||
if 'signal' in result_df.columns:
|
||||
signals = result_df['signal'].value_counts()
|
||||
print(f"Signal distribution: {signals.to_dict()}")
|
||||
|
||||
non_hold = result_df[result_df['signal'] != 'HOLD']
|
||||
print(f"Trading signals: {len(non_hold)} ({len(non_hold)/len(result_df)*100:.1f}%)")
|
||||
|
||||
if len(non_hold) > 0:
|
||||
buy_signals = len(non_hold[non_hold['signal'] == 'BUY'])
|
||||
sell_signals = len(non_hold[non_hold['signal'] == 'SELL'])
|
||||
print(f"BUY signals: {buy_signals}")
|
||||
print(f"SELL signals: {sell_signals}")
|
||||
|
||||
print(f"\\nSample signals:")
|
||||
for i, row in non_hold.head(5).iterrows():
|
||||
print(f" • {row['signal']} at ${row['close']:.2f}: {row.get('explanation', 'No explanation')}")
|
||||
|
||||
else:
|
||||
print(f"❌ No trading signals generated")
|
||||
print(f"Recent explanations:")
|
||||
for exp in result_df['explanation'].tail(5):
|
||||
print(f" • {exp}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Test failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
def test_with_real_data():
|
||||
"""Test with actual US500 data from a volatile period"""
|
||||
|
||||
print(f"\\n📊 Testing with Real US500 Data (Volatile Period)")
|
||||
print("="*60)
|
||||
|
||||
try:
|
||||
csv_file = 'lab/backtest_data/US500_H1_data.csv'
|
||||
if not os.path.exists(csv_file):
|
||||
print(f"❌ CSV file not found: {csv_file}")
|
||||
return False
|
||||
|
||||
df = pd.read_csv(csv_file, parse_dates=['time'])
|
||||
|
||||
# Find a volatile period (March 2020 - COVID crash)
|
||||
march_2020 = df[(df['time'] >= '2020-03-01') & (df['time'] <= '2020-04-30')]
|
||||
|
||||
if len(march_2020) > 100:
|
||||
print(f"✅ Using March-April 2020 data ({len(march_2020)} rows) - COVID volatility period")
|
||||
test_df = march_2020.copy()
|
||||
else:
|
||||
# Fallback: use a more recent volatile period
|
||||
test_df = df.tail(1000).copy()
|
||||
print(f"✅ Using recent 1000 rows as fallback")
|
||||
|
||||
print(f"Date range: {test_df['time'].min()} to {test_df['time'].max()}")
|
||||
print(f"Price range: ${test_df['close'].min():.2f} - ${test_df['close'].max():.2f}")
|
||||
|
||||
# Calculate volatility of this period
|
||||
returns = test_df['close'].pct_change().dropna() # pyright: ignore
|
||||
volatility = returns.std() * np.sqrt(24) # Annualized hourly volatility
|
||||
print(f"Period volatility: {volatility:.1%} (annualized)")
|
||||
|
||||
from core.strategies.index_breakout_pro import IndexBreakoutProStrategy
|
||||
|
||||
class MockBot:
|
||||
def __init__(self):
|
||||
self.market_for_mt5 = 'US500'
|
||||
|
||||
# Test with moderate parameters
|
||||
params = {
|
||||
'volume_surge_multiplier': 1.3,
|
||||
'min_breakout_size': 0.15,
|
||||
'breakout_period': 15
|
||||
}
|
||||
|
||||
strategy = IndexBreakoutProStrategy(MockBot(), params)
|
||||
result_df = strategy.analyze_df(test_df)
|
||||
|
||||
if 'signal' in result_df.columns:
|
||||
signals = result_df['signal'].value_counts()
|
||||
print(f"\\nSignal distribution: {signals.to_dict()}")
|
||||
|
||||
non_hold = result_df[result_df['signal'] != 'HOLD']
|
||||
print(f"Trading signals: {len(non_hold)} ({len(non_hold)/len(result_df)*100:.1f}%)")
|
||||
|
||||
if len(non_hold) > 0:
|
||||
print(f"\\n✅ Strategy generated signals in volatile period!")
|
||||
print(f"Sample signals:")
|
||||
for i, row in non_hold.head(8).iterrows():
|
||||
date_str = row['time'].strftime('%m-%d %H:%M') if 'time' in row.index else 'Unknown'
|
||||
print(f" • {date_str}: {row['signal']} at ${row['close']:.2f}")
|
||||
return True
|
||||
else:
|
||||
print(f"❌ No signals even in volatile period")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Real data test failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("🔍 INDEX_BREAKOUT_PRO Signal Generation Testing")
|
||||
print("=" * 70)
|
||||
|
||||
test1 = test_with_dynamic_data()
|
||||
test2 = test_with_real_data()
|
||||
|
||||
if test1 and test2:
|
||||
print(f"\\n✅ Signal generation tests successful!")
|
||||
print(f"\\n💡 The strategy can generate signals with:")
|
||||
print(f" • Volatile market conditions")
|
||||
print(f" • Moderate parameter settings")
|
||||
print(f" • Clear breakout patterns")
|
||||
print(f"\\n🔧 For web interface:")
|
||||
print(f" • Try using March 2020 data (COVID crash)")
|
||||
print(f" • Use moderate parameters: volume_surge_multiplier=1.3")
|
||||
print(f" • Consider shorter breakout_period=15")
|
||||
else:
|
||||
print(f"\\n❌ Some tests failed - strategy may need further tuning")
|
||||
@@ -0,0 +1,209 @@
|
||||
#!/usr/bin/env python3
|
||||
# test_strategy_switching.py - Demonstration of automatic strategy switching system
|
||||
|
||||
import sys
|
||||
import os
|
||||
import pandas as pd
|
||||
|
||||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
def test_strategy_switching_system():
|
||||
"""Test the complete automatic strategy switching system"""
|
||||
|
||||
print("🔄 Automatic Strategy Switching System Demo")
|
||||
print("=" * 60)
|
||||
print("Demonstrating the complete strategy switching workflow")
|
||||
print("=" * 60)
|
||||
|
||||
try:
|
||||
# Import required modules
|
||||
from core.strategies.strategy_switcher import strategy_switcher, evaluate_strategy_switch
|
||||
from core.strategies.market_condition_detector import get_market_condition
|
||||
from core.strategies.performance_scorer import calculate_strategy_score, rank_strategies
|
||||
from core.backtesting.enhanced_engine import run_enhanced_backtest
|
||||
|
||||
# Show system configuration
|
||||
print("⚙️ System Configuration:")
|
||||
print(f" Monitored Instruments: {strategy_switcher.monitored_instruments}")
|
||||
print(f" Test Strategies: {strategy_switcher.test_strategies}")
|
||||
print(f" Evaluation Period: {strategy_switcher.config['performance_evaluation_period']} bars")
|
||||
print(f" Cooldown Period: {strategy_switcher.config['switching_cooldown_hours']} hours")
|
||||
print(f" Minimum Score: {strategy_switcher.config['min_performance_score']}")
|
||||
print(f" Switch Threshold: {strategy_switcher.config['switch_threshold']}")
|
||||
|
||||
# Load market data for testing
|
||||
print(f"\n📊 Loading Market Data...")
|
||||
data_directory = 'lab/backtest_data'
|
||||
current_data = {}
|
||||
|
||||
for symbol in strategy_switcher.monitored_instruments:
|
||||
file_path = os.path.join(data_directory, f'{symbol}_H1_data.csv')
|
||||
if os.path.exists(file_path):
|
||||
try:
|
||||
df = pd.read_csv(file_path, parse_dates=['time'])
|
||||
current_data[symbol] = df.tail(1000).copy() # Use recent 1000 bars
|
||||
print(f" ✅ Loaded {len(current_data[symbol])} bars for {symbol}")
|
||||
except Exception as e:
|
||||
print(f" ❌ Error loading {symbol}: {e}")
|
||||
else:
|
||||
print(f" ⚠️ Data file not found for {symbol}")
|
||||
|
||||
if not current_data:
|
||||
print("❌ No market data available for testing")
|
||||
return False
|
||||
|
||||
print(f"\n🔍 Market Condition Analysis:")
|
||||
market_conditions = {}
|
||||
for symbol, df in current_data.items():
|
||||
if not df.empty:
|
||||
condition = get_market_condition(df, symbol)
|
||||
market_conditions[symbol] = condition
|
||||
print(f" {symbol}: {condition['market_condition']} ({condition['confidence']:.2f} confidence)")
|
||||
print(f" Volatility: {condition['volatility_regime']}")
|
||||
print(f" Session: {condition['session_status']}")
|
||||
|
||||
print(f"\n📈 Strategy Performance Evaluation:")
|
||||
performance_scores = []
|
||||
|
||||
# Evaluate strategy combinations
|
||||
for symbol, df in current_data.items():
|
||||
if df.empty:
|
||||
continue
|
||||
|
||||
market_condition = market_conditions.get(symbol, {})
|
||||
|
||||
for strategy_id in strategy_switcher.test_strategies[:3]: # Test first 3 strategies
|
||||
try:
|
||||
print(f"\n Testing {strategy_id} on {symbol}...")
|
||||
|
||||
# Get strategy parameters
|
||||
strategy_params = strategy_switcher._get_strategy_parameters(strategy_id, symbol)
|
||||
print(f" Parameters: {strategy_params}")
|
||||
|
||||
# Run backtest
|
||||
test_df = df.tail(500).copy() # Use 500 bars for testing
|
||||
backtest_results = run_enhanced_backtest(
|
||||
strategy_id,
|
||||
strategy_params,
|
||||
test_df,
|
||||
symbol_name=symbol
|
||||
)
|
||||
|
||||
if 'error' in backtest_results:
|
||||
print(f" ❌ Backtest error: {backtest_results['error']}")
|
||||
continue
|
||||
|
||||
# Calculate performance score
|
||||
score = calculate_strategy_score(
|
||||
backtest_results, market_condition, strategy_id, symbol
|
||||
)
|
||||
|
||||
performance_scores.append(score)
|
||||
|
||||
# Show results
|
||||
metrics = score['metrics']
|
||||
components = score['components']
|
||||
print(f" 📊 Performance Score: {score['composite_score']:.3f}")
|
||||
print(f" Profitability: {components['profitability']:.2f}")
|
||||
print(f" Risk Control: {components['risk_control']:.2f}")
|
||||
print(f" Market Fit: {components['market_fit']:.2f}")
|
||||
print(f" Trades: {metrics.get('total_trades', 0)}")
|
||||
print(f" Net Profit: ${metrics.get('net_profit', 0):.2f}")
|
||||
print(f" Max Drawdown: {metrics.get('max_drawdown', 0):.2f}%")
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Error evaluating {strategy_id}/{symbol}: {e}")
|
||||
continue
|
||||
|
||||
if not performance_scores:
|
||||
print("❌ No performance scores calculated")
|
||||
return False
|
||||
|
||||
# Rank strategies
|
||||
print(f"\n🏆 Strategy Rankings:")
|
||||
ranked_combinations = rank_strategies(performance_scores)
|
||||
|
||||
for i, combination in enumerate(ranked_combinations[:5]): # Top 5
|
||||
rank_emoji = ["🥇", "🥈", "🥉", "4️⃣", "5️⃣"][min(i, 4)]
|
||||
print(f" {rank_emoji} {combination['strategy_id']}/{combination['symbol']}")
|
||||
print(f" Score: {combination['composite_score']:.3f}")
|
||||
print(f" Components: P:{combination['components']['profitability']:.2f} | "
|
||||
f"R:{combination['components']['risk_control']:.2f} | "
|
||||
f"M:{combination['components']['market_fit']:.2f}")
|
||||
|
||||
# Test automatic switching logic
|
||||
print(f"\n🔄 Automatic Switching Evaluation:")
|
||||
switch_decision = evaluate_strategy_switch(current_data)
|
||||
|
||||
if switch_decision:
|
||||
print(f" 🎯 SWITCH RECOMMENDED:")
|
||||
print(f" Action: {switch_decision['action']}")
|
||||
if switch_decision['action'] == 'STRATEGY_SWITCH':
|
||||
print(f" From: {switch_decision['old_strategy']}/{switch_decision['old_symbol']}")
|
||||
print(f" To: {switch_decision['new_strategy']}/{switch_decision['new_symbol']}")
|
||||
else:
|
||||
print(f" To: {switch_decision['new_strategy']}/{switch_decision['new_symbol']}")
|
||||
print(f" Reason: {switch_decision['reason']}")
|
||||
print(f" Confidence: {switch_decision['confidence']:.3f}")
|
||||
if 'improvement' in switch_decision:
|
||||
print(f" Improvement: +{switch_decision['improvement']:.3f}")
|
||||
else:
|
||||
print(f" ✅ No switch needed at this time")
|
||||
print(f" Current strategy remains optimal")
|
||||
|
||||
# Show system status
|
||||
print(f"\n📊 System Status:")
|
||||
status = strategy_switcher.get_status()
|
||||
print(f" Current Strategy: {status['current_strategy']}")
|
||||
print(f" Current Symbol: {status['current_symbol']}")
|
||||
print(f" Last Switch: {status['last_switch_time']}")
|
||||
print(f" In Cooldown: {status['in_cooldown']}")
|
||||
print(f" Performance History: {status['performance_history_count']} entries")
|
||||
print(f" Switch Log: {status['switch_log_count']} entries")
|
||||
|
||||
# Show recent switches
|
||||
recent_switches = strategy_switcher.get_recent_switches(3)
|
||||
if recent_switches:
|
||||
print(f"\n⚡ Recent Switches:")
|
||||
for switch in recent_switches:
|
||||
decision = switch['decision']
|
||||
print(f" {switch['timestamp']}: {decision['action']}")
|
||||
if decision['action'] == 'STRATEGY_SWITCH':
|
||||
print(f" {decision['old_strategy']}/{decision['old_symbol']} → "
|
||||
f"{decision['new_strategy']}/{decision['new_symbol']}")
|
||||
print(f" Reason: {decision['reason']}")
|
||||
|
||||
print(f"\n✅ Strategy Switching System Test Complete!")
|
||||
print(f"\n💡 Key Features Demonstrated:")
|
||||
print(f" 1. ✅ Market condition detection for different instruments")
|
||||
print(f" 2. ✅ Multi-metric performance scoring system")
|
||||
print(f" 3. ✅ Automatic strategy ranking and selection")
|
||||
print(f" 4. ✅ Intelligent switching logic with cooldown periods")
|
||||
print(f" 5. ✅ Comprehensive dashboard monitoring")
|
||||
print(f" 6. ✅ REST API for integration with web interface")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Test failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("🚀 QuantumBotX Automatic Strategy Switching System")
|
||||
print("=" * 70)
|
||||
|
||||
success = test_strategy_switching_system()
|
||||
|
||||
if success:
|
||||
print(f"\n🎉 SUCCESS: Automatic Strategy Switching System is fully operational!")
|
||||
print(f"\n📋 Next Steps:")
|
||||
print(f" 1. Integrate with web dashboard for real-time monitoring")
|
||||
print(f" 2. Connect to live market data feeds")
|
||||
print(f" 3. Implement automatic switching in trading bots")
|
||||
print(f" 4. Configure alerts for strategy changes")
|
||||
print(f" 5. Add more sophisticated market condition detection")
|
||||
else:
|
||||
print(f"\n❌ Some issues occurred during testing")
|
||||
print(f" Check the output above for details")
|
||||
Reference in New Issue
Block a user