504 lines
21 KiB
Python
504 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Strategy Comparison for 2025 Q1 Data
|
|
|
|
This script runs both the original DefaultStrategy and incremental IncMetaTrendStrategy
|
|
on the same timeframe (2025-01-01 to 2025-05-01) and creates comprehensive
|
|
side-by-side comparison plots and analysis.
|
|
"""
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
import matplotlib.pyplot as plt
|
|
import matplotlib.dates as mdates
|
|
import seaborn as sns
|
|
import logging
|
|
from typing import Dict, List, Tuple, Optional
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timedelta
|
|
import json
|
|
|
|
# Add project root to path
|
|
sys.path.insert(0, os.path.abspath('..'))
|
|
|
|
from cycles.strategies.default_strategy import DefaultStrategy
|
|
from cycles.IncStrategies.metatrend_strategy import IncMetaTrendStrategy
|
|
from cycles.IncStrategies.inc_backtester import IncBacktester, BacktestConfig
|
|
from cycles.IncStrategies.inc_trader import IncTrader
|
|
from cycles.utils.storage import Storage
|
|
from cycles.backtest import Backtest
|
|
from cycles.market_fees import MarketFees
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Set style for better plots
|
|
plt.style.use('default')
|
|
sns.set_palette("husl")
|
|
|
|
|
|
class StrategyComparison2025:
|
|
"""Comprehensive comparison between original and incremental strategies for 2025 data."""
|
|
|
|
def __init__(self, start_date: str = "2025-01-01", end_date: str = "2025-05-01"):
|
|
"""Initialize the comparison."""
|
|
self.start_date = start_date
|
|
self.end_date = end_date
|
|
self.market_fees = MarketFees()
|
|
|
|
# Data storage
|
|
self.test_data = None
|
|
self.original_results = None
|
|
self.incremental_results = None
|
|
|
|
# Results storage
|
|
self.original_trades = []
|
|
self.incremental_trades = []
|
|
self.original_portfolio = []
|
|
self.incremental_portfolio = []
|
|
|
|
def load_data(self) -> pd.DataFrame:
|
|
"""Load test data for the specified date range."""
|
|
logger.info(f"Loading data from {self.start_date} to {self.end_date}")
|
|
|
|
try:
|
|
# Load data directly from CSV file
|
|
data_file = "../data/btcusd_1-min_data.csv"
|
|
logger.info(f"Loading data from: {data_file}")
|
|
|
|
# Read CSV file
|
|
df = pd.read_csv(data_file)
|
|
|
|
# Convert timestamp column
|
|
df['timestamp'] = pd.to_datetime(df['Timestamp'], unit='s')
|
|
|
|
# Rename columns to match expected format
|
|
df = df.rename(columns={
|
|
'Open': 'open',
|
|
'High': 'high',
|
|
'Low': 'low',
|
|
'Close': 'close',
|
|
'Volume': 'volume'
|
|
})
|
|
|
|
# Filter by date range
|
|
start_dt = pd.to_datetime(self.start_date)
|
|
end_dt = pd.to_datetime(self.end_date)
|
|
|
|
df = df[(df['timestamp'] >= start_dt) & (df['timestamp'] < end_dt)]
|
|
|
|
if df.empty:
|
|
raise ValueError(f"No data found for the specified date range: {self.start_date} to {self.end_date}")
|
|
|
|
# Keep only required columns
|
|
df = df[['timestamp', 'open', 'high', 'low', 'close', 'volume']]
|
|
|
|
self.test_data = df
|
|
|
|
logger.info(f"Loaded {len(df)} data points")
|
|
logger.info(f"Date range: {df['timestamp'].min()} to {df['timestamp'].max()}")
|
|
logger.info(f"Price range: ${df['close'].min():.0f} - ${df['close'].max():.0f}")
|
|
|
|
return df
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to load test data: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
raise
|
|
|
|
def run_original_strategy(self, initial_usd: float = 10000) -> Dict:
|
|
"""Run the original DefaultStrategy and extract results."""
|
|
logger.info("🔄 Running Original DefaultStrategy...")
|
|
|
|
try:
|
|
# Create indexed DataFrame for original strategy
|
|
indexed_data = self.test_data.set_index('timestamp')
|
|
|
|
# Use all available data (not limited to 200 points)
|
|
logger.info(f"Original strategy processing {len(indexed_data)} data points")
|
|
|
|
# Run original backtest with correct parameters
|
|
backtest = Backtest(
|
|
initial_balance=initial_usd,
|
|
strategies=[DefaultStrategy(weight=1.0, params={
|
|
"stop_loss_pct": 0.03,
|
|
"timeframe": "1min"
|
|
})],
|
|
market_fees=self.market_fees
|
|
)
|
|
|
|
# Run backtest
|
|
results = backtest.run(indexed_data)
|
|
|
|
# Extract trades and portfolio history
|
|
trades = results.get('trades', [])
|
|
portfolio_history = results.get('portfolio_history', [])
|
|
|
|
# Convert trades to standardized format
|
|
standardized_trades = []
|
|
for trade in trades:
|
|
standardized_trades.append({
|
|
'timestamp': trade.get('entry_time', trade.get('timestamp')),
|
|
'type': 'BUY' if trade.get('action') == 'buy' else 'SELL',
|
|
'price': trade.get('entry_price', trade.get('price')),
|
|
'exit_time': trade.get('exit_time'),
|
|
'exit_price': trade.get('exit_price'),
|
|
'profit_pct': trade.get('profit_pct', 0),
|
|
'source': 'original'
|
|
})
|
|
|
|
self.original_trades = standardized_trades
|
|
self.original_portfolio = portfolio_history
|
|
|
|
# Calculate performance metrics
|
|
final_value = results.get('final_balance', initial_usd)
|
|
total_return = (final_value - initial_usd) / initial_usd * 100
|
|
|
|
performance = {
|
|
'strategy_name': 'Original DefaultStrategy',
|
|
'initial_value': initial_usd,
|
|
'final_value': final_value,
|
|
'total_return': total_return,
|
|
'num_trades': len(trades),
|
|
'trades': standardized_trades,
|
|
'portfolio_history': portfolio_history
|
|
}
|
|
|
|
logger.info(f"✅ Original strategy completed: {len(trades)} trades, {total_return:.2f}% return")
|
|
|
|
self.original_results = performance
|
|
return performance
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Error running original strategy: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return None
|
|
|
|
def run_incremental_strategy(self, initial_usd: float = 10000) -> Dict:
|
|
"""Run the incremental strategy using the backtester."""
|
|
logger.info("🔄 Running Incremental Strategy...")
|
|
|
|
try:
|
|
# Create strategy instance
|
|
strategy = IncMetaTrendStrategy("metatrend", weight=1.0, params={
|
|
"timeframe": "1min",
|
|
"enable_logging": False
|
|
})
|
|
|
|
# Create backtest configuration
|
|
config = BacktestConfig(
|
|
initial_usd=initial_usd,
|
|
stop_loss_pct=0.03,
|
|
take_profit_pct=None
|
|
)
|
|
|
|
# Create backtester
|
|
backtester = IncBacktester()
|
|
|
|
# Run backtest
|
|
results = backtester.run_single_strategy(
|
|
strategy=strategy,
|
|
data=self.test_data,
|
|
config=config
|
|
)
|
|
|
|
# Extract results
|
|
trades = results.get('trades', [])
|
|
portfolio_history = results.get('portfolio_history', [])
|
|
|
|
# Convert trades to standardized format
|
|
standardized_trades = []
|
|
for trade in trades:
|
|
standardized_trades.append({
|
|
'timestamp': trade.entry_time,
|
|
'type': 'BUY',
|
|
'price': trade.entry_price,
|
|
'exit_time': trade.exit_time,
|
|
'exit_price': trade.exit_price,
|
|
'profit_pct': trade.profit_pct,
|
|
'source': 'incremental'
|
|
})
|
|
|
|
# Add sell signal
|
|
if trade.exit_time:
|
|
standardized_trades.append({
|
|
'timestamp': trade.exit_time,
|
|
'type': 'SELL',
|
|
'price': trade.exit_price,
|
|
'exit_time': trade.exit_time,
|
|
'exit_price': trade.exit_price,
|
|
'profit_pct': trade.profit_pct,
|
|
'source': 'incremental'
|
|
})
|
|
|
|
self.incremental_trades = standardized_trades
|
|
self.incremental_portfolio = portfolio_history
|
|
|
|
# Calculate performance metrics
|
|
final_value = results.get('final_balance', initial_usd)
|
|
total_return = (final_value - initial_usd) / initial_usd * 100
|
|
|
|
performance = {
|
|
'strategy_name': 'Incremental MetaTrend',
|
|
'initial_value': initial_usd,
|
|
'final_value': final_value,
|
|
'total_return': total_return,
|
|
'num_trades': len([t for t in trades if t.exit_time]),
|
|
'trades': standardized_trades,
|
|
'portfolio_history': portfolio_history
|
|
}
|
|
|
|
logger.info(f"✅ Incremental strategy completed: {len(trades)} trades, {total_return:.2f}% return")
|
|
|
|
self.incremental_results = performance
|
|
return performance
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Error running incremental strategy: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return None
|
|
|
|
def create_side_by_side_comparison(self, save_path: str = "../results/strategy_comparison_2025.png"):
|
|
"""Create comprehensive side-by-side comparison plots."""
|
|
logger.info("📊 Creating side-by-side comparison plots...")
|
|
|
|
# Create figure with subplots
|
|
fig = plt.figure(figsize=(24, 16))
|
|
|
|
# Create grid layout
|
|
gs = fig.add_gridspec(3, 2, height_ratios=[2, 2, 1], hspace=0.3, wspace=0.2)
|
|
|
|
# Plot 1: Original Strategy Price + Signals
|
|
ax1 = fig.add_subplot(gs[0, 0])
|
|
self._plot_strategy_signals(ax1, self.original_results, "Original DefaultStrategy", 'blue')
|
|
|
|
# Plot 2: Incremental Strategy Price + Signals
|
|
ax2 = fig.add_subplot(gs[0, 1])
|
|
self._plot_strategy_signals(ax2, self.incremental_results, "Incremental MetaTrend", 'red')
|
|
|
|
# Plot 3: Portfolio Value Comparison
|
|
ax3 = fig.add_subplot(gs[1, :])
|
|
self._plot_portfolio_comparison(ax3)
|
|
|
|
# Plot 4: Performance Summary Table
|
|
ax4 = fig.add_subplot(gs[2, :])
|
|
self._plot_performance_table(ax4)
|
|
|
|
# Overall title
|
|
fig.suptitle(f'Strategy Comparison: {self.start_date} to {self.end_date}',
|
|
fontsize=20, fontweight='bold', y=0.98)
|
|
|
|
# Save plot
|
|
plt.savefig(save_path, dpi=300, bbox_inches='tight')
|
|
plt.show()
|
|
|
|
logger.info(f"📈 Comparison plot saved to: {save_path}")
|
|
|
|
def _plot_strategy_signals(self, ax, results: Dict, title: str, color: str):
|
|
"""Plot price data with trading signals for a single strategy."""
|
|
if not results:
|
|
ax.text(0.5, 0.5, f"No data for {title}", ha='center', va='center', transform=ax.transAxes)
|
|
return
|
|
|
|
# Plot price data
|
|
ax.plot(self.test_data['timestamp'], self.test_data['close'],
|
|
color='black', linewidth=1, alpha=0.7, label='BTC Price')
|
|
|
|
# Plot trading signals
|
|
trades = results['trades']
|
|
buy_signals = [t for t in trades if t['type'] == 'BUY']
|
|
sell_signals = [t for t in trades if t['type'] == 'SELL']
|
|
|
|
if buy_signals:
|
|
buy_times = [t['timestamp'] for t in buy_signals]
|
|
buy_prices = [t['price'] for t in buy_signals]
|
|
ax.scatter(buy_times, buy_prices, color='green', marker='^',
|
|
s=100, label=f'Buy ({len(buy_signals)})', zorder=5, alpha=0.8)
|
|
|
|
if sell_signals:
|
|
sell_times = [t['timestamp'] for t in sell_signals]
|
|
sell_prices = [t['price'] for t in sell_signals]
|
|
|
|
# Separate profitable and losing sells
|
|
profitable_sells = [t for t in sell_signals if t.get('profit_pct', 0) > 0]
|
|
losing_sells = [t for t in sell_signals if t.get('profit_pct', 0) <= 0]
|
|
|
|
if profitable_sells:
|
|
profit_times = [t['timestamp'] for t in profitable_sells]
|
|
profit_prices = [t['price'] for t in profitable_sells]
|
|
ax.scatter(profit_times, profit_prices, color='blue', marker='v',
|
|
s=100, label=f'Profitable Sell ({len(profitable_sells)})', zorder=5, alpha=0.8)
|
|
|
|
if losing_sells:
|
|
loss_times = [t['timestamp'] for t in losing_sells]
|
|
loss_prices = [t['price'] for t in losing_sells]
|
|
ax.scatter(loss_times, loss_prices, color='red', marker='v',
|
|
s=100, label=f'Loss Sell ({len(losing_sells)})', zorder=5, alpha=0.8)
|
|
|
|
ax.set_title(title, fontsize=14, fontweight='bold')
|
|
ax.set_ylabel('Price (USD)', fontsize=12)
|
|
ax.legend(loc='upper left', fontsize=10)
|
|
ax.grid(True, alpha=0.3)
|
|
ax.yaxis.set_major_formatter(plt.FuncFormatter(lambda x, p: f'${x:,.0f}'))
|
|
|
|
# Format x-axis
|
|
ax.xaxis.set_major_locator(mdates.DayLocator(interval=7)) # Every 7 days (weekly)
|
|
ax.xaxis.set_major_formatter(mdates.DateFormatter('%m-%d'))
|
|
plt.setp(ax.xaxis.get_majorticklabels(), rotation=45)
|
|
|
|
def _plot_portfolio_comparison(self, ax):
|
|
"""Plot portfolio value comparison between strategies."""
|
|
# Plot initial value line
|
|
ax.axhline(y=10000, color='gray', linestyle='--', alpha=0.7, label='Initial Value ($10,000)')
|
|
|
|
# Plot original strategy portfolio
|
|
if self.original_results and self.original_results.get('portfolio_history'):
|
|
portfolio = self.original_results['portfolio_history']
|
|
if portfolio:
|
|
times = [p.get('timestamp', p.get('time')) for p in portfolio]
|
|
values = [p.get('portfolio_value', p.get('value', 10000)) for p in portfolio]
|
|
ax.plot(times, values, color='blue', linewidth=2,
|
|
label=f"Original ({self.original_results['total_return']:+.1f}%)", alpha=0.8)
|
|
|
|
# Plot incremental strategy portfolio
|
|
if self.incremental_results and self.incremental_results.get('portfolio_history'):
|
|
portfolio = self.incremental_results['portfolio_history']
|
|
if portfolio:
|
|
times = [p.get('timestamp', p.get('time')) for p in portfolio]
|
|
values = [p.get('portfolio_value', p.get('value', 10000)) for p in portfolio]
|
|
ax.plot(times, values, color='red', linewidth=2,
|
|
label=f"Incremental ({self.incremental_results['total_return']:+.1f}%)", alpha=0.8)
|
|
|
|
ax.set_title('Portfolio Value Comparison', fontsize=14, fontweight='bold')
|
|
ax.set_ylabel('Portfolio Value (USD)', fontsize=12)
|
|
ax.set_xlabel('Date', fontsize=12)
|
|
ax.legend(loc='upper left', fontsize=12)
|
|
ax.grid(True, alpha=0.3)
|
|
ax.yaxis.set_major_formatter(plt.FuncFormatter(lambda x, p: f'${x:,.0f}'))
|
|
|
|
# Format x-axis
|
|
ax.xaxis.set_major_locator(mdates.DayLocator(interval=7)) # Every 7 days (weekly)
|
|
ax.xaxis.set_major_formatter(mdates.DateFormatter('%m-%d'))
|
|
plt.setp(ax.xaxis.get_majorticklabels(), rotation=45)
|
|
|
|
def _plot_performance_table(self, ax):
|
|
"""Create performance comparison table."""
|
|
ax.axis('off')
|
|
|
|
if not self.original_results or not self.incremental_results:
|
|
ax.text(0.5, 0.5, "Performance data not available", ha='center', va='center',
|
|
transform=ax.transAxes, fontsize=14)
|
|
return
|
|
|
|
# Create comparison table
|
|
orig = self.original_results
|
|
incr = self.incremental_results
|
|
|
|
comparison_text = f"""
|
|
PERFORMANCE COMPARISON - {self.start_date} to {self.end_date}
|
|
{'='*80}
|
|
|
|
{'Metric':<25} {'Original':<20} {'Incremental':<20} {'Difference':<15}
|
|
{'-'*80}
|
|
{'Initial Value':<25} ${orig['initial_value']:>15,.0f} ${incr['initial_value']:>17,.0f} ${incr['initial_value'] - orig['initial_value']:>12,.0f}
|
|
{'Final Value':<25} ${orig['final_value']:>15,.0f} ${incr['final_value']:>17,.0f} ${incr['final_value'] - orig['final_value']:>12,.0f}
|
|
{'Total Return':<25} {orig['total_return']:>15.2f}% {incr['total_return']:>17.2f}% {incr['total_return'] - orig['total_return']:>12.2f}%
|
|
{'Number of Trades':<25} {orig['num_trades']:>15} {incr['num_trades']:>17} {incr['num_trades'] - orig['num_trades']:>12}
|
|
|
|
ANALYSIS:
|
|
• Data Period: {len(self.test_data):,} minute bars ({(len(self.test_data) / 1440):.1f} days)
|
|
• Price Range: ${self.test_data['close'].min():,.0f} - ${self.test_data['close'].max():,.0f}
|
|
• Both strategies use identical MetaTrend logic with 3% stop loss
|
|
• Differences indicate implementation variations or data processing differences
|
|
"""
|
|
|
|
ax.text(0.05, 0.95, comparison_text, transform=ax.transAxes, fontsize=11,
|
|
verticalalignment='top', fontfamily='monospace',
|
|
bbox=dict(boxstyle="round,pad=0.5", facecolor="lightblue", alpha=0.9))
|
|
|
|
def save_results(self, output_dir: str = "../results"):
|
|
"""Save detailed results to files."""
|
|
logger.info("💾 Saving detailed results...")
|
|
|
|
os.makedirs(output_dir, exist_ok=True)
|
|
|
|
# Save original strategy trades
|
|
if self.original_results:
|
|
orig_trades_df = pd.DataFrame(self.original_results['trades'])
|
|
orig_file = f"{output_dir}/original_trades_2025.csv"
|
|
orig_trades_df.to_csv(orig_file, index=False)
|
|
logger.info(f"Original trades saved to: {orig_file}")
|
|
|
|
# Save incremental strategy trades
|
|
if self.incremental_results:
|
|
incr_trades_df = pd.DataFrame(self.incremental_results['trades'])
|
|
incr_file = f"{output_dir}/incremental_trades_2025.csv"
|
|
incr_trades_df.to_csv(incr_file, index=False)
|
|
logger.info(f"Incremental trades saved to: {incr_file}")
|
|
|
|
# Save performance summary
|
|
summary = {
|
|
'timeframe': f"{self.start_date} to {self.end_date}",
|
|
'data_points': len(self.test_data) if self.test_data is not None else 0,
|
|
'original_strategy': self.original_results,
|
|
'incremental_strategy': self.incremental_results
|
|
}
|
|
|
|
summary_file = f"{output_dir}/strategy_comparison_2025.json"
|
|
with open(summary_file, 'w') as f:
|
|
json.dump(summary, f, indent=2, default=str)
|
|
logger.info(f"Performance summary saved to: {summary_file}")
|
|
|
|
def run_full_comparison(self, initial_usd: float = 10000):
|
|
"""Run the complete comparison workflow."""
|
|
logger.info("🚀 Starting Full Strategy Comparison for 2025 Q1")
|
|
logger.info("=" * 60)
|
|
|
|
try:
|
|
# Load data
|
|
self.load_data()
|
|
|
|
# Run both strategies
|
|
self.run_original_strategy(initial_usd)
|
|
self.run_incremental_strategy(initial_usd)
|
|
|
|
# Create comparison plots
|
|
self.create_side_by_side_comparison()
|
|
|
|
# Save results
|
|
self.save_results()
|
|
|
|
# Print summary
|
|
if self.original_results and self.incremental_results:
|
|
logger.info("\n📊 COMPARISON SUMMARY:")
|
|
logger.info(f"Original Strategy: ${self.original_results['final_value']:,.0f} ({self.original_results['total_return']:+.2f}%)")
|
|
logger.info(f"Incremental Strategy: ${self.incremental_results['final_value']:,.0f} ({self.incremental_results['total_return']:+.2f}%)")
|
|
logger.info(f"Difference: ${self.incremental_results['final_value'] - self.original_results['final_value']:,.0f} ({self.incremental_results['total_return'] - self.original_results['total_return']:+.2f}%)")
|
|
|
|
logger.info("✅ Full comparison completed successfully!")
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Error during comparison: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
|
|
def main():
|
|
"""Main function to run the strategy comparison."""
|
|
# Create comparison instance
|
|
comparison = StrategyComparison2025(
|
|
start_date="2025-01-01",
|
|
end_date="2025-05-01"
|
|
)
|
|
|
|
# Run full comparison
|
|
comparison.run_full_comparison(initial_usd=10000)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |