313 lines
12 KiB
Python
313 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Analyze Exit Signal Differences Between Strategies
|
|
=================================================
|
|
|
|
This script examines the exact differences in exit signal logic between
|
|
the original and incremental strategies to understand why the original
|
|
generates so many more exit signals.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import pandas as pd
|
|
import numpy as np
|
|
from datetime import datetime
|
|
import matplotlib.pyplot as plt
|
|
|
|
# Add the parent directory to the path to import cycles modules
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from cycles.utils.storage import Storage
|
|
from cycles.IncStrategies.metatrend_strategy import IncMetaTrendStrategy
|
|
from cycles.strategies.default_strategy import DefaultStrategy
|
|
|
|
|
|
def analyze_exit_conditions():
|
|
"""Analyze the exit conditions in both strategies."""
|
|
print("🔍 ANALYZING EXIT SIGNAL LOGIC")
|
|
print("=" * 80)
|
|
|
|
print("\n📋 ORIGINAL STRATEGY (DefaultStrategy) EXIT CONDITIONS:")
|
|
print("-" * 60)
|
|
print("1. Meta-trend exit: prev_trend != 1 AND curr_trend == -1")
|
|
print(" - Only exits when trend changes TO -1 (downward)")
|
|
print(" - Does NOT exit when trend goes from 1 to 0 (neutral)")
|
|
print("2. Stop loss: Currently DISABLED in signal generation")
|
|
print(" - Code comment: 'skip stop loss checking in signal generation'")
|
|
|
|
print("\n📋 INCREMENTAL STRATEGY (IncMetaTrendStrategy) EXIT CONDITIONS:")
|
|
print("-" * 60)
|
|
print("1. Meta-trend exit: prev_trend != -1 AND curr_trend == -1")
|
|
print(" - Only exits when trend changes TO -1 (downward)")
|
|
print(" - Does NOT exit when trend goes from 1 to 0 (neutral)")
|
|
print("2. Stop loss: Not implemented in this strategy")
|
|
|
|
print("\n🤔 THEORETICAL ANALYSIS:")
|
|
print("-" * 60)
|
|
print("Both strategies have IDENTICAL exit conditions!")
|
|
print("The difference must be in HOW/WHEN they check for exits...")
|
|
|
|
return True
|
|
|
|
|
|
def compare_signal_generation_frequency():
|
|
"""Compare how frequently each strategy checks for signals."""
|
|
print("\n🔍 ANALYZING SIGNAL GENERATION FREQUENCY")
|
|
print("=" * 80)
|
|
|
|
print("\n📋 ORIGINAL STRATEGY SIGNAL CHECKING:")
|
|
print("-" * 60)
|
|
print("• Checks signals at EVERY 15-minute bar")
|
|
print("• Processes ALL historical data points during initialization")
|
|
print("• get_exit_signal() called for EVERY timeframe bar")
|
|
print("• No state tracking - evaluates conditions fresh each time")
|
|
|
|
print("\n📋 INCREMENTAL STRATEGY SIGNAL CHECKING:")
|
|
print("-" * 60)
|
|
print("• Checks signals only when NEW 15-minute bar completes")
|
|
print("• Processes data incrementally as it arrives")
|
|
print("• get_exit_signal() called only on timeframe bar completion")
|
|
print("• State tracking - remembers previous signals to avoid duplicates")
|
|
|
|
print("\n🎯 KEY DIFFERENCE IDENTIFIED:")
|
|
print("-" * 60)
|
|
print("ORIGINAL: Evaluates exit condition at EVERY historical bar")
|
|
print("INCREMENTAL: Evaluates exit condition only on STATE CHANGES")
|
|
|
|
return True
|
|
|
|
|
|
def test_signal_generation_with_sample_data():
|
|
"""Test both strategies with sample data to see the difference."""
|
|
print("\n🧪 TESTING WITH SAMPLE DATA")
|
|
print("=" * 80)
|
|
|
|
# Load a small sample of data
|
|
storage = Storage()
|
|
data_file = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data", "btcusd_1-min_data.csv")
|
|
|
|
# Load just 3 days of data for detailed analysis
|
|
start_date = "2025-01-01"
|
|
end_date = "2025-01-04"
|
|
|
|
print(f"Loading data from {start_date} to {end_date}...")
|
|
data_1min = storage.load_data(data_file, start_date, end_date)
|
|
print(f"Loaded {len(data_1min)} minute-level data points")
|
|
|
|
# Test original strategy
|
|
print("\n🔄 Testing Original Strategy...")
|
|
original_signals = test_original_strategy_detailed(data_1min)
|
|
|
|
# Test incremental strategy
|
|
print("\n🔄 Testing Incremental Strategy...")
|
|
incremental_signals = test_incremental_strategy_detailed(data_1min)
|
|
|
|
# Compare results
|
|
print("\n📊 DETAILED COMPARISON:")
|
|
print("-" * 60)
|
|
|
|
orig_exits = [s for s in original_signals if s['type'] == 'EXIT']
|
|
inc_exits = [s for s in incremental_signals if s['type'] == 'SELL']
|
|
|
|
print(f"Original exit signals: {len(orig_exits)}")
|
|
print(f"Incremental exit signals: {len(inc_exits)}")
|
|
print(f"Difference: {len(orig_exits) - len(inc_exits)} more exits in original")
|
|
|
|
# Show first few exit signals from each
|
|
print(f"\n📋 FIRST 5 ORIGINAL EXIT SIGNALS:")
|
|
for i, signal in enumerate(orig_exits[:5]):
|
|
print(f" {i+1}. {signal['timestamp']} - Price: ${signal['price']:.0f}")
|
|
|
|
print(f"\n📋 FIRST 5 INCREMENTAL EXIT SIGNALS:")
|
|
for i, signal in enumerate(inc_exits[:5]):
|
|
print(f" {i+1}. {signal['timestamp']} - Price: ${signal['price']:.0f}")
|
|
|
|
return original_signals, incremental_signals
|
|
|
|
|
|
def test_original_strategy_detailed(data_1min: pd.DataFrame):
|
|
"""Test original strategy with detailed logging."""
|
|
|
|
# Create mock backtester
|
|
class MockBacktester:
|
|
def __init__(self, data):
|
|
self.original_df = data
|
|
self.strategies = {}
|
|
self.current_position = None
|
|
self.entry_price = None
|
|
|
|
# Initialize strategy
|
|
strategy = DefaultStrategy(
|
|
weight=1.0,
|
|
params={
|
|
"timeframe": "15min",
|
|
"stop_loss_pct": 0.03
|
|
}
|
|
)
|
|
|
|
mock_backtester = MockBacktester(data_1min)
|
|
strategy.initialize(mock_backtester)
|
|
|
|
if not strategy.initialized:
|
|
print(" ❌ Strategy initialization failed")
|
|
return []
|
|
|
|
# Get primary timeframe data
|
|
primary_data = strategy.get_primary_timeframe_data()
|
|
signals = []
|
|
|
|
print(f" Processing {len(primary_data)} timeframe bars...")
|
|
|
|
# Track meta-trend changes for analysis
|
|
meta_trend_changes = []
|
|
|
|
for i in range(len(primary_data)):
|
|
timestamp = primary_data.index[i]
|
|
|
|
# Get current meta-trend value
|
|
if hasattr(strategy, 'meta_trend') and i < len(strategy.meta_trend):
|
|
curr_trend = strategy.meta_trend[i]
|
|
prev_trend = strategy.meta_trend[i-1] if i > 0 else 0
|
|
|
|
if curr_trend != prev_trend:
|
|
meta_trend_changes.append({
|
|
'timestamp': timestamp,
|
|
'prev_trend': prev_trend,
|
|
'curr_trend': curr_trend,
|
|
'index': i
|
|
})
|
|
|
|
# Check for exit signal
|
|
exit_signal = strategy.get_exit_signal(mock_backtester, i)
|
|
if exit_signal and exit_signal.signal_type == "EXIT":
|
|
signals.append({
|
|
'timestamp': timestamp,
|
|
'type': 'EXIT',
|
|
'price': primary_data.iloc[i]['close'],
|
|
'strategy': 'Original',
|
|
'confidence': exit_signal.confidence,
|
|
'metadata': exit_signal.metadata,
|
|
'meta_trend': curr_trend if 'curr_trend' in locals() else 'unknown',
|
|
'prev_meta_trend': prev_trend if 'prev_trend' in locals() else 'unknown'
|
|
})
|
|
|
|
print(f" Found {len(meta_trend_changes)} meta-trend changes")
|
|
print(f" Generated {len([s for s in signals if s['type'] == 'EXIT'])} exit signals")
|
|
|
|
# Show meta-trend changes
|
|
print(f"\n 📈 META-TREND CHANGES:")
|
|
for change in meta_trend_changes[:10]: # Show first 10
|
|
print(f" {change['timestamp']}: {change['prev_trend']} → {change['curr_trend']}")
|
|
|
|
return signals
|
|
|
|
|
|
def test_incremental_strategy_detailed(data_1min: pd.DataFrame):
|
|
"""Test incremental strategy with detailed logging."""
|
|
|
|
# Initialize strategy
|
|
strategy = IncMetaTrendStrategy(
|
|
name="metatrend",
|
|
weight=1.0,
|
|
params={
|
|
"timeframe": "15min",
|
|
"enable_logging": False
|
|
}
|
|
)
|
|
|
|
signals = []
|
|
meta_trend_changes = []
|
|
bars_completed = 0
|
|
|
|
print(f" Processing {len(data_1min)} minute-level data points...")
|
|
|
|
# Process each minute of data
|
|
for i, (timestamp, row) in enumerate(data_1min.iterrows()):
|
|
ohlcv_data = {
|
|
'open': row['open'],
|
|
'high': row['high'],
|
|
'low': row['low'],
|
|
'close': row['close'],
|
|
'volume': row['volume']
|
|
}
|
|
|
|
# Update strategy
|
|
result = strategy.update_minute_data(timestamp, ohlcv_data)
|
|
|
|
# Check if a complete timeframe bar was formed
|
|
if result is not None:
|
|
bars_completed += 1
|
|
|
|
# Track meta-trend changes
|
|
if hasattr(strategy, 'current_meta_trend') and hasattr(strategy, 'previous_meta_trend'):
|
|
if strategy.current_meta_trend != strategy.previous_meta_trend:
|
|
meta_trend_changes.append({
|
|
'timestamp': timestamp,
|
|
'prev_trend': strategy.previous_meta_trend,
|
|
'curr_trend': strategy.current_meta_trend,
|
|
'bar_number': bars_completed
|
|
})
|
|
|
|
# Check for exit signal
|
|
exit_signal = strategy.get_exit_signal()
|
|
if exit_signal and exit_signal.signal_type.upper() == 'EXIT':
|
|
signals.append({
|
|
'timestamp': timestamp,
|
|
'type': 'SELL',
|
|
'price': row['close'],
|
|
'strategy': 'Incremental',
|
|
'confidence': exit_signal.confidence,
|
|
'reason': exit_signal.metadata.get('type', 'EXIT') if exit_signal.metadata else 'EXIT',
|
|
'meta_trend': strategy.current_meta_trend,
|
|
'prev_meta_trend': strategy.previous_meta_trend
|
|
})
|
|
|
|
print(f" Completed {bars_completed} timeframe bars")
|
|
print(f" Found {len(meta_trend_changes)} meta-trend changes")
|
|
print(f" Generated {len([s for s in signals if s['type'] == 'SELL'])} exit signals")
|
|
|
|
# Show meta-trend changes
|
|
print(f"\n 📈 META-TREND CHANGES:")
|
|
for change in meta_trend_changes[:10]: # Show first 10
|
|
print(f" {change['timestamp']}: {change['prev_trend']} → {change['curr_trend']}")
|
|
|
|
return signals
|
|
|
|
|
|
def main():
|
|
"""Main analysis function."""
|
|
print("🔍 ANALYZING WHY ORIGINAL STRATEGY HAS MORE EXIT SIGNALS")
|
|
print("=" * 80)
|
|
|
|
try:
|
|
# Step 1: Analyze exit conditions
|
|
analyze_exit_conditions()
|
|
|
|
# Step 2: Compare signal generation frequency
|
|
compare_signal_generation_frequency()
|
|
|
|
# Step 3: Test with sample data
|
|
original_signals, incremental_signals = test_signal_generation_with_sample_data()
|
|
|
|
print("\n🎯 FINAL CONCLUSION:")
|
|
print("=" * 80)
|
|
print("The original strategy generates more exit signals because:")
|
|
print("1. It evaluates exit conditions at EVERY historical timeframe bar")
|
|
print("2. It doesn't track signal state - treats each bar independently")
|
|
print("3. When meta-trend is -1, it generates exit signal at EVERY bar")
|
|
print("4. The incremental strategy only signals on STATE CHANGES")
|
|
print("\nThis explains the 8x difference in exit signal count!")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"\n❌ Error during analysis: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
success = main()
|
|
sys.exit(0 if success else 1) |