""" Strategy Comparison Test Framework Comprehensive testing for comparing original incremental strategies from cycles/IncStrategies with new implementations in IncrementalTrader/strategies. This test framework validates: 1. MetaTrend Strategy: IncMetaTrendStrategy vs MetaTrendStrategy 2. Random Strategy: IncRandomStrategy vs RandomStrategy 3. BBRS Strategy: BBRSIncrementalState vs BBRSStrategy Each test validates signal generation, mathematical equivalence, and behavioral consistency. """ import pandas as pd import numpy as np import matplotlib.pyplot as plt import matplotlib.dates as mdates from datetime import datetime import sys from pathlib import Path from typing import Dict, List, Tuple, Any import os # Add project paths sys.path.append(str(Path(__file__).parent.parent)) sys.path.append(str(Path(__file__).parent.parent / "cycles")) sys.path.append(str(Path(__file__).parent.parent / "IncrementalTrader")) # Import original strategies from cycles.IncStrategies.metatrend_strategy import IncMetaTrendStrategy from cycles.IncStrategies.random_strategy import IncRandomStrategy from cycles.IncStrategies.bbrs_incremental import BBRSIncrementalState # Import new strategies from IncrementalTrader.strategies.metatrend import MetaTrendStrategy from IncrementalTrader.strategies.random import RandomStrategy from IncrementalTrader.strategies.bbrs import BBRSStrategy class StrategyComparisonTester: def __init__(self, data_file: str = "data/btcusd_1-min_data.csv"): """Initialize the strategy comparison tester.""" self.data_file = data_file self.data = None self.results_dir = Path("test/results/strategies") self.results_dir.mkdir(parents=True, exist_ok=True) def load_data(self, limit: int = 1000) -> bool: """Load and prepare test data.""" try: print(f"Loading data from {self.data_file}...") self.data = pd.read_csv(self.data_file) # Limit data for testing if limit: self.data = self.data.head(limit) print(f"Loaded {len(self.data)} data points") print(f"Data columns: {list(self.data.columns)}") print(f"Data sample:\n{self.data.head()}") return True except Exception as e: print(f"Error loading data: {e}") return False def compare_metatrend_strategies(self) -> Dict[str, Any]: """Compare IncMetaTrendStrategy vs MetaTrendStrategy.""" print("\n" + "="*80) print("COMPARING METATREND STRATEGIES") print("="*80) try: # Initialize strategies with same parameters original_strategy = IncMetaTrendStrategy() new_strategy = MetaTrendStrategy() # Track signals original_entry_signals = [] new_entry_signals = [] original_exit_signals = [] new_exit_signals = [] combined_original_signals = [] combined_new_signals = [] timestamps = [] # Process data for i, row in self.data.iterrows(): timestamp = pd.Timestamp(row['Timestamp'], unit='s') ohlcv_data = { 'open': row['Open'], 'high': row['High'], 'low': row['Low'], 'close': row['Close'], 'volume': row['Volume'] } # Update original strategy (uses update_minute_data) original_strategy.update_minute_data(timestamp, ohlcv_data) # Update new strategy (uses process_data_point) new_strategy.process_data_point(timestamp, ohlcv_data) # Get signals orig_entry = original_strategy.get_entry_signal() new_entry = new_strategy.get_entry_signal() orig_exit = original_strategy.get_exit_signal() new_exit = new_strategy.get_exit_signal() # Store signals (both use signal_type) original_entry_signals.append(orig_entry.signal_type if orig_entry else "HOLD") new_entry_signals.append(new_entry.signal_type if new_entry else "HOLD") original_exit_signals.append(orig_exit.signal_type if orig_exit else "HOLD") new_exit_signals.append(new_exit.signal_type if new_exit else "HOLD") # Combined signal logic (simplified) orig_signal = "BUY" if orig_entry and orig_entry.signal_type == "ENTRY" else ("SELL" if orig_exit and orig_exit.signal_type == "EXIT" else "HOLD") new_signal = "BUY" if new_entry and new_entry.signal_type == "ENTRY" else ("SELL" if new_exit and new_exit.signal_type == "EXIT" else "HOLD") combined_original_signals.append(orig_signal) combined_new_signals.append(new_signal) timestamps.append(timestamp) # Calculate consistency metrics entry_matches = sum(1 for o, n in zip(original_entry_signals, new_entry_signals) if o == n) exit_matches = sum(1 for o, n in zip(original_exit_signals, new_exit_signals) if o == n) combined_matches = sum(1 for o, n in zip(combined_original_signals, combined_new_signals) if o == n) total_points = len(self.data) entry_consistency = (entry_matches / total_points) * 100 exit_consistency = (exit_matches / total_points) * 100 combined_consistency = (combined_matches / total_points) * 100 results = { 'strategy_name': 'MetaTrend', 'total_points': total_points, 'entry_consistency': entry_consistency, 'exit_consistency': exit_consistency, 'combined_consistency': combined_consistency, 'original_entry_signals': original_entry_signals, 'new_entry_signals': new_entry_signals, 'original_exit_signals': original_exit_signals, 'new_exit_signals': new_exit_signals, 'combined_original_signals': combined_original_signals, 'combined_new_signals': combined_new_signals, 'timestamps': timestamps } print(f"Entry Signal Consistency: {entry_consistency:.2f}%") print(f"Exit Signal Consistency: {exit_consistency:.2f}%") print(f"Combined Signal Consistency: {combined_consistency:.2f}%") return results except Exception as e: print(f"Error comparing MetaTrend strategies: {e}") import traceback traceback.print_exc() return {} def compare_random_strategies(self) -> Dict[str, Any]: """Compare IncRandomStrategy vs RandomStrategy.""" print("\n" + "="*80) print("COMPARING RANDOM STRATEGIES") print("="*80) try: # Initialize strategies with same seed for reproducibility # Original: IncRandomStrategy(weight, params) # New: RandomStrategy(name, weight, params) original_strategy = IncRandomStrategy(weight=1.0, params={"random_seed": 42}) new_strategy = RandomStrategy(name="random", weight=1.0, params={"random_seed": 42}) # Track signals original_signals = [] new_signals = [] timestamps = [] # Process data for i, row in self.data.iterrows(): timestamp = pd.Timestamp(row['Timestamp'], unit='s') ohlcv_data = { 'open': row['Open'], 'high': row['High'], 'low': row['Low'], 'close': row['Close'], 'volume': row['Volume'] } # Update strategies original_strategy.update_minute_data(timestamp, ohlcv_data) new_strategy.process_data_point(timestamp, ohlcv_data) # Get signals orig_signal = original_strategy.get_entry_signal() # Random strategy uses get_entry_signal new_signal = new_strategy.get_entry_signal() # Store signals original_signals.append(orig_signal.signal_type if orig_signal else "HOLD") new_signals.append(new_signal.signal_type if new_signal else "HOLD") timestamps.append(timestamp) # Calculate consistency metrics matches = sum(1 for o, n in zip(original_signals, new_signals) if o == n) total_points = len(self.data) consistency = (matches / total_points) * 100 results = { 'strategy_name': 'Random', 'total_points': total_points, 'consistency': consistency, 'original_signals': original_signals, 'new_signals': new_signals, 'timestamps': timestamps } print(f"Signal Consistency: {consistency:.2f}%") return results except Exception as e: print(f"Error comparing Random strategies: {e}") import traceback traceback.print_exc() return {} def compare_bbrs_strategies(self) -> Dict[str, Any]: """Compare BBRSIncrementalState vs BBRSStrategy.""" print("\n" + "="*80) print("COMPARING BBRS STRATEGIES") print("="*80) try: # Initialize strategies with same configuration # Original: BBRSIncrementalState(config) # New: BBRSStrategy(name, weight, params) original_config = { "timeframe_minutes": 60, "bb_period": 20, "rsi_period": 14, "bb_width": 0.05, "trending": { "bb_std_dev_multiplier": 2.5, "rsi_threshold": [30, 70] }, "sideways": { "bb_std_dev_multiplier": 1.8, "rsi_threshold": [40, 60] }, "SqueezeStrategy": True } new_params = { "timeframe": "1h", "bb_period": 20, "rsi_period": 14, "bb_width_threshold": 0.05, "trending_bb_multiplier": 2.5, "sideways_bb_multiplier": 1.8, "trending_rsi_thresholds": [30, 70], "sideways_rsi_thresholds": [40, 60], "squeeze_strategy": True, "enable_logging": False } original_strategy = BBRSIncrementalState(original_config) new_strategy = BBRSStrategy(name="bbrs", weight=1.0, params=new_params) # Track signals original_signals = [] new_signals = [] timestamps = [] # Process data for i, row in self.data.iterrows(): timestamp = pd.Timestamp(row['Timestamp'], unit='s') ohlcv_data = { 'open': row['Open'], 'high': row['High'], 'low': row['Low'], 'close': row['Close'], 'volume': row['Volume'] } # Update strategies orig_result = original_strategy.update_minute_data(timestamp, ohlcv_data) new_strategy.process_data_point(timestamp, ohlcv_data) # Get signals from original (returns dict with buy_signal/sell_signal) if orig_result and orig_result.get('buy_signal', False): orig_signal = "BUY" elif orig_result and orig_result.get('sell_signal', False): orig_signal = "SELL" else: orig_signal = "HOLD" # Get signals from new strategy new_entry = new_strategy.get_entry_signal() new_exit = new_strategy.get_exit_signal() if new_entry and new_entry.signal_type == "ENTRY": new_signal = "BUY" elif new_exit and new_exit.signal_type == "EXIT": new_signal = "SELL" else: new_signal = "HOLD" # Store signals original_signals.append(orig_signal) new_signals.append(new_signal) timestamps.append(timestamp) # Calculate consistency metrics matches = sum(1 for o, n in zip(original_signals, new_signals) if o == n) total_points = len(self.data) consistency = (matches / total_points) * 100 results = { 'strategy_name': 'BBRS', 'total_points': total_points, 'consistency': consistency, 'original_signals': original_signals, 'new_signals': new_signals, 'timestamps': timestamps } print(f"Signal Consistency: {consistency:.2f}%") return results except Exception as e: print(f"Error comparing BBRS strategies: {e}") import traceback traceback.print_exc() return {} def generate_report(self, results: List[Dict[str, Any]]) -> None: """Generate comprehensive comparison report.""" print("\n" + "="*80) print("GENERATING STRATEGY COMPARISON REPORT") print("="*80) # Create summary report report_file = self.results_dir / "strategy_comparison_report.txt" with open(report_file, 'w', encoding='utf-8') as f: f.write("Strategy Comparison Report\n") f.write("=" * 50 + "\n\n") f.write(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"Data points tested: {results[0]['total_points'] if results else 'N/A'}\n\n") for result in results: if not result: continue f.write(f"Strategy: {result['strategy_name']}\n") f.write("-" * 30 + "\n") if result['strategy_name'] == 'MetaTrend': f.write(f"Entry Signal Consistency: {result['entry_consistency']:.2f}%\n") f.write(f"Exit Signal Consistency: {result['exit_consistency']:.2f}%\n") f.write(f"Combined Signal Consistency: {result['combined_consistency']:.2f}%\n") # Status determination if result['combined_consistency'] >= 95: status = "✅ EXCELLENT" elif result['combined_consistency'] >= 90: status = "✅ GOOD" elif result['combined_consistency'] >= 80: status = "⚠️ ACCEPTABLE" else: status = "❌ NEEDS REVIEW" else: f.write(f"Signal Consistency: {result['consistency']:.2f}%\n") # Status determination if result['consistency'] >= 95: status = "✅ EXCELLENT" elif result['consistency'] >= 90: status = "✅ GOOD" elif result['consistency'] >= 80: status = "⚠️ ACCEPTABLE" else: status = "❌ NEEDS REVIEW" f.write(f"Status: {status}\n\n") print(f"Report saved to: {report_file}") # Generate plots for each strategy for result in results: if not result: continue self.plot_strategy_comparison(result) def plot_strategy_comparison(self, result: Dict[str, Any]) -> None: """Generate comparison plots for a strategy.""" strategy_name = result['strategy_name'] fig, axes = plt.subplots(2, 1, figsize=(15, 10)) fig.suptitle(f'{strategy_name} Strategy Comparison', fontsize=16, fontweight='bold') timestamps = result['timestamps'] if strategy_name == 'MetaTrend': # Plot entry signals axes[0].plot(timestamps, [1 if s == "ENTRY" else 0 for s in result['original_entry_signals']], label='Original Entry', alpha=0.7, linewidth=2) axes[0].plot(timestamps, [1 if s == "ENTRY" else 0 for s in result['new_entry_signals']], label='New Entry', alpha=0.7, linewidth=2, linestyle='--') axes[0].set_title(f'Entry Signals - Consistency: {result["entry_consistency"]:.2f}%') axes[0].set_ylabel('Entry Signal') axes[0].legend() axes[0].grid(True, alpha=0.3) # Plot combined signals signal_map = {"BUY": 1, "SELL": -1, "HOLD": 0} orig_combined = [signal_map[s] for s in result['combined_original_signals']] new_combined = [signal_map[s] for s in result['combined_new_signals']] axes[1].plot(timestamps, orig_combined, label='Original Combined', alpha=0.7, linewidth=2) axes[1].plot(timestamps, new_combined, label='New Combined', alpha=0.7, linewidth=2, linestyle='--') axes[1].set_title(f'Combined Signals - Consistency: {result["combined_consistency"]:.2f}%') axes[1].set_ylabel('Signal (-1=SELL, 0=HOLD, 1=BUY)') else: # For Random and BBRS strategies signal_map = {"BUY": 1, "SELL": -1, "HOLD": 0} orig_signals = [signal_map.get(s, 0) for s in result['original_signals']] new_signals = [signal_map.get(s, 0) for s in result['new_signals']] axes[0].plot(timestamps, orig_signals, label='Original', alpha=0.7, linewidth=2) axes[0].plot(timestamps, new_signals, label='New', alpha=0.7, linewidth=2, linestyle='--') axes[0].set_title(f'Signals - Consistency: {result["consistency"]:.2f}%') axes[0].set_ylabel('Signal (-1=SELL, 0=HOLD, 1=BUY)') # Plot difference diff = [o - n for o, n in zip(orig_signals, new_signals)] axes[1].plot(timestamps, diff, label='Difference (Original - New)', color='red', alpha=0.7) axes[1].set_title('Signal Differences') axes[1].set_ylabel('Difference') axes[1].axhline(y=0, color='black', linestyle='-', alpha=0.3) # Format x-axis for ax in axes: ax.legend() ax.grid(True, alpha=0.3) ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M')) ax.xaxis.set_major_locator(mdates.HourLocator(interval=2)) plt.setp(ax.xaxis.get_majorticklabels(), rotation=45) plt.xlabel('Time') plt.tight_layout() # Save plot plot_file = self.results_dir / f"{strategy_name.lower()}_strategy_comparison.png" plt.savefig(plot_file, dpi=300, bbox_inches='tight') plt.close() print(f"Plot saved to: {plot_file}") def main(): """Main test execution.""" print("Strategy Comparison Test Framework") print("=" * 50) # Initialize tester tester = StrategyComparisonTester() # Load data if not tester.load_data(limit=1000): # Use 1000 points for testing print("Failed to load data. Exiting.") return # Run comparisons results = [] # Compare MetaTrend strategies metatrend_result = tester.compare_metatrend_strategies() if metatrend_result: results.append(metatrend_result) # Compare Random strategies random_result = tester.compare_random_strategies() if random_result: results.append(random_result) # Compare BBRS strategies bbrs_result = tester.compare_bbrs_strategies() if bbrs_result: results.append(bbrs_result) # Generate comprehensive report if results: tester.generate_report(results) print("\n" + "="*80) print("STRATEGY COMPARISON SUMMARY") print("="*80) for result in results: if not result: continue strategy_name = result['strategy_name'] if strategy_name == 'MetaTrend': consistency = result['combined_consistency'] print(f"{strategy_name}: {consistency:.2f}% consistency") else: consistency = result['consistency'] print(f"{strategy_name}: {consistency:.2f}% consistency") if consistency >= 95: status = "✅ EXCELLENT" elif consistency >= 90: status = "✅ GOOD" elif consistency >= 80: status = "⚠️ ACCEPTABLE" else: status = "❌ NEEDS REVIEW" print(f" Status: {status}") print(f"\nDetailed results saved in: {tester.results_dir}") else: print("No successful comparisons completed.") if __name__ == "__main__": main()