momentum_strategy_analysis_and_simulation_tool.py

"""
Cryptocurrency Momentum Strategy Analysis and Simulation Tool

WARNING: THIS SCRIPT IS PROVIDED FOR EDUCATIONAL PURPOSES ONLY.
THIS IS NOT FINANCIAL ADVICE. CRYPTOCURRENCY INVESTMENTS ARE HIGHLY
VOLATILE AND INVOLVE SIGNIFICANT RISKS. ALWAYS CONDUCT YOUR OWN
RESEARCH AND CONSULT A QUALIFIED FINANCIAL ADVISOR BEFORE MAKING
ANY INVESTMENT DECISIONS.

Description:
This script enables users to:
- Analyze historical cryptocurrency performance
- Implement an advanced momentum strategy
- Compare different strategy configurations
- Visualize performance results

Key Features:
- Historical data download via yfinance
- Momentum calculation across different periods
- Selection of best-performing cryptocurrencies based on momentum
- Portfolio simulation with annual rebalancing
- Comparative performance visualization

Dependencies:
- numpy
- pandas
- matplotlib
- yfinance
- tabulate
"""

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import yfinance as yf
from tabulate import tabulate

class EnhancedMomentumStrategy:
    def __init__(self, tickers, start_date, end_date, initial_investment=100000,
                 momentum_window=12, top_n_cryptos=3):
        self.tickers = tickers
        self.start_date = start_date
        self.end_date = end_date
        self.initial_investment = initial_investment
        self.momentum_window = momentum_window
        self.top_n_cryptos = top_n_cryptos

    def download_crypto_data(self):
        """
        Downloads historical cryptocurrency data with robust error handling
        """
        try:
            data = yf.download(self.tickers, start=self.start_date, end=self.end_date)['Adj Close']
            data = data.ffill().bfill()
            return data
        except Exception as e:
            print(f"Error downloading data: {e}")
            return None

    def calculate_momentum(self, data):
        """
        Calculates momentum with proper handling of missing data
        """
        if data is None or data.empty:
            return None

        # Calculate momentum as percentage change over specified window
        momentum = data.pct_change(periods=self.momentum_window) * 100
        momentum = momentum.fillna(0)

        return momentum

    def select_top_momentum_cryptos(self, momentum):
        """
        Selects top cryptocurrencies based on momentum
        """
        if momentum is None or momentum.empty:
            return None

        # Use last row of momentum data, sort and select top N
        try:
            return momentum.iloc[-1].sort_values(ascending=False).head(self.top_n_cryptos)
        except Exception as e:
            print(f"Error selecting top cryptos: {e}")
            return None

    def simulate_portfolio(self):
        """
        Simulates portfolio strategy with comprehensive error handling
        """
        # Download price data
        data = self.download_crypto_data()

        if data is None or data.empty:
            print("No data available for simulation.")
            return None

        portfolio_value = self.initial_investment
        portfolio_history = [portfolio_value]

        # Simulate for each year
        years_to_simulate = range(int(self.start_date[:4]), min(int(self.end_date[:4]) + 1, 2025))

        for year in years_to_simulate:
            year_start = f"{year}-01-01"
            year_end = f"{year}-12-31"

            # Calculate momentum and select top cryptos
            yearly_data = data.loc[:year_end]
            yearly_momentum = self.calculate_momentum(yearly_data)

            if yearly_momentum is None:
                print(f"Unable to calculate momentum for {year}")
                continue

            top_cryptos = self.select_top_momentum_cryptos(yearly_momentum)

            if top_cryptos is None or len(top_cryptos) == 0:
                print(f"No top cryptos found for {year}")
                continue

            # Allocate portfolio equally among top cryptos
            allocation_per_crypto = portfolio_value / len(top_cryptos)

            # Calculate year-end portfolio value
            year_end_values = {}
            for crypto in top_cryptos.index:
                try:
                    # Use first available price of the year
                    current_price = yearly_data.loc[year_start:, crypto].iloc[0]
                    crypto_shares = allocation_per_crypto / current_price

                    # Calculate year-end value
                    year_end_price = yearly_data.loc[year_start:year_end, crypto].iloc[-1]
                    year_end_values[crypto] = crypto_shares * year_end_price
                except Exception as e:
                    print(f"Unable to calculate value for {crypto}: {e}")

            # Update portfolio value
            if year_end_values:
                portfolio_value = sum(year_end_values.values())
                portfolio_history.append(portfolio_value)

        # Performance calculations
        if len(portfolio_history) > 1:
            total_return = (portfolio_history[-1] - self.initial_investment) / self.initial_investment * 100

            return {
                'initial_investment': self.initial_investment,
                'final_value': portfolio_history[-1],
                'total_return_percent': total_return,
                'top_cryptos': list(top_cryptos.index)
            }
        else:
            print("Insufficient data for performance calculation")
            return None

def compare_momentum_strategies(strategies_params):
    """
    Compares multiple momentum strategy configurations
    """
    comparison_results = []

    for params in strategies_params:
        strategy = EnhancedMomentumStrategy(**params)
        performance = strategy.simulate_portfolio()

        if performance:
            result = {
                'Momentum Window': params.get('momentum_window', 'N/A'),
                'Top N Cryptos': params.get('top_n_cryptos', 'N/A'),
                'Initial Investment': f"${performance['initial_investment']:,.2f}",
                'Final Value': f"${performance['final_value']:,.2f}",
                'Total Return (%)': f"{performance['total_return_percent']:.2f}%",
                'Top Cryptocurrencies': ', '.join(performance['top_cryptos'])
            }
            comparison_results.append(result)

    return comparison_results

def main():
    # Complete list of cryptocurrencies
    cryptos = [
        'BTC-USD', 'ETH-USD', 'BNB-USD', 'ADA-USD', 'DOT-USD',
        'LTC-USD', 'SOL-USD', 'LINK-USD', 'DOGE-USD', 'XRP-USD',
        'AVAX-USD', 'UNI-USD', 'ATOM-USD', 'ALGO-USD', 'MANA-USD'
    ]

    # Define different strategy parameter configurations
    strategies_params = [
        {
            'tickers': cryptos,
            'start_date': '2020-01-01',
            'end_date': '2024-12-31',
            'initial_investment': 100000,
            'momentum_window': window,
            'top_n_cryptos': top_n
        }
        for window in [6, 12, 18]
        for top_n in [3, 5, 7]
    ]

    # Compare strategies
    results = compare_momentum_strategies(strategies_params)

    # Display results
    print(tabulate(results, headers="keys", tablefmt="pretty"))

    # Performance visualization
    if results:
        plt.figure(figsize=(12, 6))
        returns = [float(result['Total Return (%)'].strip('%')) for result in results]
        labels = [f"MW{result['Momentum Window']}\nTop{result['Top N Cryptos']}" for result in results]

        plt.bar(labels, returns)
        plt.title('Comparison of Momentum Strategy Performances')
        plt.ylabel('Total Return (%)')
        plt.xlabel('Strategy Configuration')
        plt.xticks(rotation=45, ha='right')
        plt.tight_layout()
        plt.show()

if __name__ == "__main__":
    main()