validate_core_functionality.py

#!/usr/bin/env python
"""
CryptoWildfire Core Functionality Validator
-------------------------------------------
This script tests the core components of CryptoWildfire and provides
a comprehensive report on what's working and what needs improvement.
"""
import os
import sys
import json
import logging
import traceback
import importlib
from datetime import datetime
import numpy as np
import pandas as pd

# Configure logging
logger = logging.getLogger(__name__)


# Configure paths
project_root = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, project_root)

# Initialize results dictionary
results = {
    "timestamp": datetime.now().isoformat(),
    "components": {},
    "summary": {
        "passed": 0,
        "failed": 0,
        "skipped": 0,
        "total": 0
    },
    "recommendations": []
}

def colored(text, color):
    """Add color to terminal output."""
    colors = {
        "green": "\033[92m",
        "red": "\033[91m",
        "yellow": "\033[93m",
        "blue": "\033[94m",
        "end": "\033[0m"
    }
    return f"{colors.get(color, '')}{text}{colors['end']}"

def print_header(text):
    """Print a section header."""
    print(f"\n{colored('='*50, 'blue')}")
    print(colored(f" {text.upper()}", "blue"))
    print(colored('='*50, 'blue'))

def print_result(name, status, details=None):
    """Print a test result with appropriate color."""
    if status == "passed":
        status_colored = colored("✓ PASSED", "green")
    elif status == "failed":
        status_colored = colored("✗ FAILED", "red")
    else:
        status_colored = colored("⚠ SKIPPED", "yellow")
    
    print(f"{name}: {status_colored}")
    if details:
        print(f"  {details}")

def import_module(module_path):
    """
    Import a module and return it, or None if import fails.
    """
    try:
        module = importlib.import_module(module_path)
        return module
    except ImportError as e:
        print(colored(f"  Could not import {module_path}: {e}", "yellow"))
        return None

def create_test_market_data():
    """Generate sample market data for testing."""
    # Create 100 days of daily OHLCV data
    dates = pd.date_range(start="2023-01-01", periods=100)
    base = 100
    trend = np.linspace(0, 20, 100)  # Upward trend
    noise = np.random.normal(0, 2, 100)  # Add some noise
    
    prices = base + trend + noise
    
    data = pd.DataFrame({
        "timestamp": dates,
        "open": prices - np.random.uniform(0, 1, 100),
        "high": prices + np.random.uniform(0, 2, 100),
        "low": prices - np.random.uniform(0, 2, 100),
        "close": prices,
        "volume": np.random.uniform(1000, 5000, 100)
    })
    
    data.set_index("timestamp", inplace=True)
    return data

def create_multi_asset_data():
    """Generate multi-asset data for correlation testing."""
    dates = pd.date_range(start="2023-01-01", periods=100)
    
    # Base asset (BTC)
    base = 100
    trend1 = np.linspace(0, 20, 100)
    noise1 = np.random.normal(0, 5, 100)
    asset1 = base + trend1 + noise1
    
    # Correlated asset (ETH)
    base2 = 50
    trend2 = np.linspace(0, 15, 100)
    noise2 = 0.8 * noise1 + 0.6 * np.random.normal(0, 5, 100)
    asset2 = base2 + trend2 + noise2
    
    # Less correlated asset (SOL)
    base3 = 20
    trend3 = np.linspace(0, 10, 100)
    noise3 = 0.5 * noise1 + 0.866 * np.random.normal(0, 5, 100)
    asset3 = base3 + trend3 + noise3
    
    assets = {}
    for name, prices in [("BTC", asset1), ("ETH", asset2), ("SOL", asset3)]:
        assets[name] = pd.DataFrame({"timestamp": dates, "close": prices})
        assets[name].set_index("timestamp", inplace=True)
    
    return assets

def test_risk_assessment():
    """Test the risk assessment module."""
    print_header("Testing Risk Assessment")
    
    # Import risk assessment module
    risk_module = import_module("src.risk.risk_assessment")
    
    if risk_module is None:
        results["components"]["risk_assessment"] = {
            "status": "skipped",
            "reason": "Module could not be imported"
        }
        results["summary"]["skipped"] += 1
        results["summary"]["total"] += 1
        print_result("Risk Assessment Module", "skipped", "Module could not be imported")
        return
    
    test_results = {
        "dynamic_stop_loss": {},
        "trailing_stop": {},
        "chart_stop_levels": {},
        "status": "passed"
    }
    
    # Test DynamicStopLoss class
    try:
        DynamicStopLoss = getattr(risk_module, "DynamicStopLoss")
        stop_loss = DynamicStopLoss(atr_period=14, atr_multiplier=2.0)
        
        # Test basic stop loss calculation
        stop_result = stop_loss.calculate_stop_loss(
            asset="BTC",
            entry_price=100.0,
            current_price=110.0,
            atr_value=5.0,
            position_type="long"
        )
        
        # Log the result for debugging
        logger.info(f"Calculated stop-loss for BTC at {stop_result}")
        
        # Check that the function returns a dictionary with stop_loss key
        if isinstance(stop_result, dict) and 'stop_loss' in stop_result:
            stop_level = stop_result['stop_loss']
            # Check that stop is below current price
            if stop_level < 110.0:
                test_results["dynamic_stop_loss"]["basic"] = {
                    "status": "passed",
                    "stop_level": stop_level
                }
                print_result("Basic Stop Loss", "passed", f"Stop level: {stop_level:.2f}")
            else:
                test_results["dynamic_stop_loss"]["basic"] = {
                    "status": "failed",
                    "reason": "Stop level not below current price",
                    "stop_level": stop_level
                }
                test_results["status"] = "failed"
                print_result("Basic Stop Loss", "failed", f"Stop level {stop_level:.2f} not below current price 110.0")
        else:
            test_results["dynamic_stop_loss"]["basic"] = {
                "status": "failed",
                "reason": "Invalid return format",
                "stop_result": str(stop_result)
            }
            test_results["status"] = "failed"
            print_result("Basic Stop Loss", "failed", "Function should return dict with 'stop_loss' key")
        
        # Test trailing stop functionality if available
        if hasattr(stop_loss, "enable_trailing"):
            try:
                # Enable trailing stop
                stop_loss.enable_trailing(initial_trail_pct=0.05)
                
                # Test trailing stop update
                stop_loss.update_trail(current_price=115.0)
                
                # Get new stop level
                new_stop_result = stop_loss.calculate_stop_loss(
                    asset="BTC",
                    entry_price=100.0,
                    current_price=115.0,
                    atr_value=5.0,
                    position_type="long"
                )
                
                # Extract stop level from result
                new_stop = new_stop_result.get('stop_loss') if isinstance(new_stop_result, dict) else new_stop_result
                stop_level = stop_result.get('stop_loss') if isinstance(stop_result, dict) else stop_level
                
                # Trailing stop should adjust upward with price
                if new_stop > stop_level:
                    test_results["trailing_stop"] = {
                        "status": "passed",
                        "initial_stop": stop_level,
                        "new_stop": new_stop
                    }
                    print_result("Trailing Stop", "passed", f"Stop adjusted from {stop_level:.2f} to {new_stop:.2f}")
                else:
                    test_results["trailing_stop"] = {
                        "status": "failed",
                        "reason": "Trailing stop did not adjust upward",
                        "initial_stop": stop_level,
                        "new_stop": new_stop
                    }
                    test_results["status"] = "failed"
                    print_result("Trailing Stop", "failed", f"Stop did not adjust upward ({stop_level:.2f} to {new_stop:.2f})")
            except Exception as e:
                test_results["trailing_stop"] = {
                    "status": "failed",
                    "reason": str(e)
                }
                test_results["status"] = "failed"
                print_result("Trailing Stop", "failed", str(e))
        else:
            test_results["trailing_stop"] = {
                "status": "skipped",
                "reason": "Trailing stop functionality not available"
            }
            print_result("Trailing Stop", "skipped", "Functionality not available")
            results["recommendations"].append("Implement trailing stop functionality in DynamicStopLoss class")
        
        # Test chart stop levels functionality
        if hasattr(risk_module, "get_stop_levels_for_chart"):
            try:
                market_data = create_test_market_data()
                stop_levels = risk_module.get_stop_levels_for_chart(market_data, multiplier=2.0)
                
                if len(stop_levels) > 0 and isinstance(stop_levels[0], dict):
                    test_results["chart_stop_levels"] = {
                        "status": "passed",
                        "levels_count": len(stop_levels)
                    }
                    print_result("Chart Stop Levels", "passed", f"Generated {len(stop_levels)} stop levels")
                else:
                    test_results["chart_stop_levels"] = {
                        "status": "failed",
                        "reason": "Invalid stop levels format"
                    }
                    test_results["status"] = "failed"
                    print_result("Chart Stop Levels", "failed", "Invalid stop levels format")
            except Exception as e:
                test_results["chart_stop_levels"] = {
                    "status": "failed",
                    "reason": str(e)
                }
                test_results["status"] = "failed"
                print_result("Chart Stop Levels", "failed", str(e))
        else:
            test_results["chart_stop_levels"] = {
                "status": "skipped",
                "reason": "Chart stop levels functionality not available"
            }
            print_result("Chart Stop Levels", "skipped", "Functionality not available")
    
    except Exception as e:
        test_results["status"] = "failed"
        test_results["error"] = str(e)
        print_result("Risk Assessment Module", "failed", str(e))
    
    # Update overall results
    results["components"]["risk_assessment"] = test_results
    if test_results["status"] == "passed":
        results["summary"]["passed"] += 1
        print_result("Risk Assessment Module (Overall)", "passed")
    else:
        results["summary"]["failed"] += 1
        print_result("Risk Assessment Module (Overall)", "failed")
    results["summary"]["total"] += 1

def test_correlation_analysis():
    """Test the correlation analysis module."""
    print_header("Testing Correlation Analysis")
    
    # Import correlation module
    correlation_module = import_module("src.analysis.correlation")
    
    if correlation_module is None:
        results["components"]["correlation"] = {
            "status": "skipped",
            "reason": "Module could not be imported"
        }
        results["summary"]["skipped"] += 1
        results["summary"]["total"] += 1
        print_result("Correlation Module", "skipped", "Module could not be imported")
        return
    
    # Test calculate_portfolio_correlation function
    if hasattr(correlation_module, "calculate_portfolio_correlation"):
        try:
            # Generate test data
            assets = create_multi_asset_data()
            
            # Calculate correlation
            correlation_matrix = correlation_module.calculate_portfolio_correlation(
                assets, window=30, price_col="close"
            )
            
            # Check result structure
            if isinstance(correlation_matrix, dict) and "BTC" in correlation_matrix:
                btc_eth_corr = correlation_matrix.get("BTC", {}).get("ETH")
                if btc_eth_corr is not None and -1 <= btc_eth_corr <= 1:
                    results["components"]["correlation"] = {
                        "status": "passed",
                        "btc_eth_correlation": btc_eth_corr,
                        "matrix_size": len(correlation_matrix)
                    }
                    results["summary"]["passed"] += 1
                    print_result("Correlation Calculation", "passed", 
                                f"BTC-ETH correlation: {btc_eth_corr:.2f}")
                else:
                    results["components"]["correlation"] = {
                        "status": "failed",
                        "reason": "Invalid correlation values",
                        "btc_eth_correlation": btc_eth_corr
                    }
                    results["summary"]["failed"] += 1
                    print_result("Correlation Calculation", "failed", 
                                "Invalid correlation values")
            else:
                results["components"]["correlation"] = {
                    "status": "failed",
                    "reason": "Invalid correlation matrix structure"
                }
                results["summary"]["failed"] += 1
                print_result("Correlation Calculation", "failed", 
                            "Invalid correlation matrix structure")
        except Exception as e:
            results["components"]["correlation"] = {
                "status": "failed",
                "reason": str(e)
            }
            results["summary"]["failed"] += 1
            print_result("Correlation Calculation", "failed", str(e))
    else:
        results["components"]["correlation"] = {
            "status": "skipped",
            "reason": "calculate_portfolio_correlation function not found"
        }
        results["summary"]["skipped"] += 1
        print_result("Correlation Calculation", "skipped", 
                    "calculate_portfolio_correlation function not found")
    
    results["summary"]["total"] += 1

def test_technical_analysis():
    """Test the technical analysis module."""
    print_header("Testing Technical Analysis")
    
    # Import technical analysis module
    ta_module = import_module("src.analysis.technical_analysis")
    
    if ta_module is None:
        results["components"]["technical_analysis"] = {
            "status": "skipped",
            "reason": "Module could not be imported"
        }
        results["summary"]["skipped"] += 1
        results["summary"]["total"] += 1
        print_result("Technical Analysis Module", "skipped", "Module could not be imported")
        return
    
    # Test AdvancedTechnicalAnalyzer class
    if hasattr(ta_module, "AdvancedTechnicalAnalyzer"):
        try:
            # Create analyzer
            analyzer = ta_module.AdvancedTechnicalAnalyzer()
            
            # Generate test data
            market_data = create_test_market_data()
            
            # Analyze data
            analysis_result = analyzer.analyze(market_data)
            
            # Check result structure
            if isinstance(analysis_result, dict) and len(analysis_result) > 0:
                indicators = list(analysis_result.keys())
                results["components"]["technical_analysis"] = {
                    "status": "passed",
                    "indicators_count": len(indicators),
                    "indicators": indicators[:5]  # First 5 indicators
                }
                results["summary"]["passed"] += 1
                print_result("Technical Analysis", "passed", 
                            f"Generated {len(indicators)} indicators")
            else:
                results["components"]["technical_analysis"] = {
                    "status": "failed",
                    "reason": "Invalid analysis result structure"
                }
                results["summary"]["failed"] += 1
                print_result("Technical Analysis", "failed", 
                            "Invalid analysis result structure")
        except Exception as e:
            results["components"]["technical_analysis"] = {
                "status": "failed",
                "reason": str(e)
            }
            results["summary"]["failed"] += 1
            print_result("Technical Analysis", "failed", str(e))
    else:
        results["components"]["technical_analysis"] = {
            "status": "skipped",
            "reason": "AdvancedTechnicalAnalyzer class not found"
        }
        results["summary"]["skipped"] += 1
        print_result("Technical Analysis", "skipped", 
                    "AdvancedTechnicalAnalyzer class not found")
    
    results["summary"]["total"] += 1

def test_model_trainer():
    """Test the model trainer module."""
    print_header("Testing Model Trainer")
    
    # Import model trainer module
    model_module = import_module("src.ml.model_trainer")
    
    if model_module is None:
        results["components"]["model_trainer"] = {
            "status": "skipped",
            "reason": "Module could not be imported"
        }
        results["summary"]["skipped"] += 1
        results["summary"]["total"] += 1
        print_result("Model Trainer Module", "skipped", "Module could not be imported")
        return
    
    # Test create_quantile_model function
    if hasattr(model_module, "create_quantile_model"):
        try:
            # Create quantile models
            q10_model = model_module.create_quantile_model(alpha=0.1)
            q50_model = model_module.create_quantile_model(alpha=0.5)
            q90_model = model_module.create_quantile_model(alpha=0.9)
            
            # Check model parameters
            if (q10_model.get_params()["alpha"] == 0.1 and 
                q50_model.get_params()["alpha"] == 0.5 and
                q90_model.get_params()["alpha"] == 0.9):
                
                results["components"]["model_trainer"] = {
                    "status": "passed",
                    "models": ["q10", "q50", "q90"],
                    "model_type": str(type(q50_model).__name__)
                }
                results["summary"]["passed"] += 1
                print_result("Quantile Model Creation", "passed", 
                            f"Created 3 quantile models ({results['components']['model_trainer']['model_type']})")
            else:
                results["components"]["model_trainer"] = {
                    "status": "failed",
                    "reason": "Incorrect model parameters"
                }
                results["summary"]["failed"] += 1
                print_result("Quantile Model Creation", "failed", 
                            "Incorrect model parameters")
        except Exception as e:
            results["components"]["model_trainer"] = {
                "status": "failed",
                "reason": str(e)
            }
            results["summary"]["failed"] += 1
            print_result("Quantile Model Creation", "failed", str(e))
    else:
        results["components"]["model_trainer"] = {
            "status": "skipped",
            "reason": "create_quantile_model function not found"
        }
        results["summary"]["skipped"] += 1
        print_result("Quantile Model Creation", "skipped", 
                    "create_quantile_model function not found")
    
    results["summary"]["total"] += 1

def test_feature_engineering():
    """Test the feature engineering module."""
    print_header("Testing Feature Engineering")
    
    # Import feature engineering module
    feature_module = import_module("src.ml.feature_engineering")
    
    if feature_module is None:
        results["components"]["feature_engineering"] = {
            "status": "skipped",
            "reason": "Module could not be imported"
        }
        results["summary"]["skipped"] += 1
        results["summary"]["total"] += 1
        print_result("Feature Engineering Module", "skipped", "Module could not be imported")
        return
    
    # Test FeatureEngineer class
    if hasattr(feature_module, "FeatureEngineer"):
        try:
            # Create feature engineer
            engineer = feature_module.FeatureEngineer()
            
            # Check if generate_features method exists
            if hasattr(engineer, "generate_features"):
                # Note: We're not actually running the method as it might have
                # dependencies on external data sources, but we check its presence
                results["components"]["feature_engineering"] = {
                    "status": "passed",
                    "methods": ["generate_features"],
                    "notes": "Method existence verified, actual execution skipped"
                }
                results["summary"]["passed"] += 1
                print_result("Feature Engineering", "passed", 
                            "FeatureEngineer class has generate_features method")
                
                # Add recommendation for testing with real data
                results["recommendations"].append(
                    "Test feature engineering with real market data"
                )
            else:
                results["components"]["feature_engineering"] = {
                    "status": "failed",
                    "reason": "generate_features method not found"
                }
                results["summary"]["failed"] += 1
                print_result("Feature Engineering", "failed", 
                            "generate_features method not found")
        except Exception as e:
            results["components"]["feature_engineering"] = {
                "status": "failed",
                "reason": str(e)
            }
            results["summary"]["failed"] += 1
            print_result("Feature Engineering", "failed", str(e))
    else:
        results["components"]["feature_engineering"] = {
            "status": "skipped",
            "reason": "FeatureEngineer class not found"
        }
        results["summary"]["skipped"] += 1
        print_result("Feature Engineering", "skipped", 
                    "FeatureEngineer class not found")
    
    results["summary"]["total"] += 1

def generate_recommendations():
    """Generate additional recommendations based on test results."""
    # Check for any failed components
    failed_components = [name for name, data in results["components"].items() 
                       if data.get("status") == "failed"]
    
    if failed_components:
        results["recommendations"].append(
            f"Fix issues in failed components: {', '.join(failed_components)}"
        )
    
    # Check for technical analysis component
    if (results["components"].get("technical_analysis", {}).get("status") == "passed" and
        results["components"].get("risk_assessment", {}).get("status") == "passed"):
        results["recommendations"].append(
            "Integrate technical indicators with risk assessment for adaptive stop-loss levels"
        )
    
    # Check for correlation component
    if results["components"].get("correlation", {}).get("status") == "passed":
        results["recommendations"].append(
            "Add portfolio risk metrics based on correlation matrix"
        )
    
    # Check for model trainer component
    if results["components"].get("model_trainer", {}).get("status") == "passed":
        results["recommendations"].append(
            "Implement model retraining schedule and version control"
        )
    
    # General recommendations
    results["recommendations"].append(
        "Develop comprehensive integration tests to validate end-to-end pipeline"
    )
    
    results["recommendations"].append(
        "Add benchmarking to identify performance bottlenecks"
    )

def print_summary():
    """Print summary of test results."""
    print_header("Test Summary")
    
    total = results["summary"]["total"]
    passed = results["summary"]["passed"]
    failed = results["summary"]["failed"]
    skipped = results["summary"]["skipped"]
    
    pass_rate = (passed / total * 100) if total > 0 else 0
    
    print(f"Total Components: {total}")
    print(f"Passed: {colored(str(passed), 'green')} ({pass_rate:.1f}%)")
    print(f"Failed: {colored(str(failed), 'red')}")
    print(f"Skipped: {colored(str(skipped), 'yellow')}")
    
    print_header("Recommendations")
    for i, rec in enumerate(results["recommendations"], 1):
        print(f"{i}. {rec}")

def main():
    """Run all tests and generate report."""
    # Create directory for results
    os.makedirs("test_results", exist_ok=True)
    
    # Run tests
    test_risk_assessment()
    test_correlation_analysis()
    test_technical_analysis()
    test_model_trainer()
    test_feature_engineering()
    
    # Generate recommendations
    generate_recommendations()
    
    # Print summary
    print_summary()
    
    # Save results
    result_path = os.path.join("test_results", f"validation_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json")
    with open(result_path, "w") as f:
        json.dump(results, f, indent=2, default=str)
    
    print(f"\nDetailed results saved to: {result_path}")
    
    # Return exit code based on test results
    if results["summary"]["failed"] > 0:
        return 1
    return 0

if __name__ == "__main__":
    sys.exit(main())