crypto_portfolio_optimizer

# Cryptocurrency Portfolio Optimizer
# Version: 1.0
# Filename: crypto_portfolio_optimizer_en_v1.0.py
# Author: Sunu Xaliss
# Date: December 2024
# Description: Cryptocurrency Portfolio Optimization Tool

"""
Cryptocurrency Portfolio Allocation and Analysis Tool

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

Description:
This script allows users to:
- Retrieve historical cryptocurrency market data
- Optimize a cryptocurrency portfolio using Modern Portfolio Theory
- Analyze portfolio performance metrics
- Visualize the efficient frontier of cryptocurrency investments

Key Features:
- Fetches historical price data for multiple cryptocurrencies
- Performs Monte Carlo simulation for portfolio optimization
- Calculates portfolio return, volatility, and Sharpe ratio
- Generates an efficient frontier visualization
- Provides transparent ethical disclosure of methodology

Dependencies:
- numpy
- pandas
- yfinance
- matplotlib
- seaborn
- scipy
"""

# Required imports
!pip install numpy pandas yfinance matplotlib seaborn scipy
import numpy as np
import pandas as pd
import yfinance as yf
from datetime import datetime, timedelta
import logging
import matplotlib.pyplot as plt
import seaborn as sns
from scipy.optimize import minimize

# Configuration
CRYPTOS_TO_ANALYZE = [
    'BTC-USD',   # Bitcoin
    'ETH-USD',   # Ethereum
    'XRP-USD',   # Ripple
    'DASH-USD',  # Dash
    'DOGE-USD',  # Dogecoin
    'XLM-USD',   # Stellar
    'LTC-USD',   # Litecoin
    'ETC-USD',   # Ethereum Classic
    'BCH-USD'    # Bitcoin Cash
]

# Logging configuration
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class CryptoPortfolioOptimizer:
    def __init__(
        self,
        cryptos=CRYPTOS_TO_ANALYZE,
        start_date='2014-01-01',
        end_date=datetime.now().strftime('%Y-%m-%d'),
        initial_capital=10000
    ):
        """
        Initialize the cryptocurrency portfolio optimizer

        Args:
            cryptos (list): List of cryptocurrency symbols
            start_date (str): Start date for historical analysis
            end_date (str): End date for historical analysis
            initial_capital (float): Initial investment capital
        """
        self.cryptos = cryptos
        self.start_date = start_date
        self.end_date = end_date
        self.initial_capital = initial_capital

        # Initialization of attributes
        self.historical_data = None
        self.returns = None
        self.covariance_matrix = None

    def fetch_historical_data(self):
        """
        Retrieve historical data for cryptocurrencies

        Returns:
            pd.DataFrame: Consolidated historical data
        """
        all_crypto_data = []

        for crypto in self.cryptos:
            try:
                ticker = yf.Ticker(crypto)
                data = ticker.history(start=self.start_date, end=self.end_date)

                if not data.empty:
                    data['Symbol'] = crypto.split('-')[0]
                    data = data.reset_index().rename(columns={
                        'Date': 'date',
                        'Close': 'close',
                        'Volume': 'volume'
                    })

                    all_crypto_data.append(data[['date', 'Symbol', 'close', 'volume']])
                else:
                    logger.warning(f"No data found for {crypto}")

            except Exception as e:
                logger.error(f"Error retrieving data for {crypto}: {e}")

        self.historical_data = pd.concat(all_crypto_data, ignore_index=True)
        return self.historical_data

    def preprocess_data(self):
        """
        Preprocess historical data
        - Clean missing data
        - Calculate daily returns
        - Normalize data
        """
        if self.historical_data is None:
            raise ValueError("No historical data loaded. Run fetch_historical_data() first.")

        # Pivot data to create price matrix
        prices_pivot = self.historical_data.pivot(
            index='date',
            columns='Symbol',
            values='close'
        )

        # Clean missing data
        prices_pivot.dropna(inplace=True)

        # Calculate daily returns
        self.returns = prices_pivot.pct_change().dropna()

        # Calculate covariance matrix
        self.covariance_matrix = self.returns.cov()

        return self.returns

    def calculate_portfolio_metrics(self, weights):
        """
        Calculate portfolio metrics

        Args:
            weights (np.array): Asset weights

        Returns:
            tuple: Portfolio return, volatility, Sharpe ratio
        """
        # Annualized portfolio return
        portfolio_return = np.sum(self.returns.mean() * weights) * 252

        # Annualized portfolio volatility
        portfolio_volatility = np.sqrt(
            np.dot(weights.T, np.dot(self.covariance_matrix * 252, weights))
        )

        # Sharpe ratio (assuming a risk-free rate of 2%)
        sharpe_ratio = (portfolio_return - 0.02) / portfolio_volatility

        return portfolio_return, portfolio_volatility, sharpe_ratio

    def optimize_portfolio(self, num_portfolios=100000):
        """
        Portfolio optimization using Monte Carlo simulation

        Args:
            num_portfolios (int): Number of portfolios to simulate

        Returns:
            dict: Optimization results
        """
        # Ensure data is preprocessed
        if self.returns is None:
            self.preprocess_data()

        # Number of assets
        num_assets = len(self.cryptos)

        # Initialize results storage
        results = np.zeros((3, num_portfolios))
        weights_record = np.zeros((num_portfolios, num_assets))

        for i in range(num_portfolios):
            # Generate random weights with constraint sum = 1
            weights = np.random.random(num_assets)
            weights /= np.sum(weights)

            # Calculate metrics
            portfolio_return, portfolio_volatility, sharpe_ratio = self.calculate_portfolio_metrics(weights)

            # Store results
            results[0, i] = portfolio_volatility
            results[1, i] = portfolio_return
            results[2, i] = sharpe_ratio
            weights_record[i, :] = weights

        # Identify optimal portfolio (max Sharpe)
        max_sharpe_idx = np.argmax(results[2, :])
        optimal_weights = weights_record[max_sharpe_idx, :]

        return {
            'volatilities': results[0, :],
            'returns': results[1, :],
            'sharpe_ratios': results[2, :],
            'optimal_weights': dict(zip(self.cryptos, optimal_weights)),
            'optimal_metrics': {
                'volatility': results[0, max_sharpe_idx],
                'return': results[1, max_sharpe_idx],
                'sharpe_ratio': results[2, max_sharpe_idx]
            }
        }

    def visualize_efficient_frontier(self, optimization_results):
        """
        Visualize the efficient frontier

        Args:
            optimization_results (dict): Optimization results
        """
        plt.figure(figsize=(12, 8))
        plt.scatter(
            optimization_results['volatilities'],
            optimization_results['returns'],
            c=optimization_results['sharpe_ratios'],
            cmap='viridis'
        )
        plt.colorbar(label='Sharpe Ratio')
        plt.xlabel('Volatility (Risk)')
        plt.ylabel('Expected Return')
        plt.title('Efficient Frontier - Crypto Portfolio')
        plt.show()

    def ethical_disclosure(self):
        """
        Transparent disclosure of assumptions and limitations
        """
        disclosure = f"""
        Transparency Disclosure for Cryptocurrency Portfolio Optimization

        Analysis Period: {self.start_date} - {self.end_date}
        Cryptocurrencies Analyzed: {', '.join(self.cryptos)}

        Assumptions:
        1. Use of historical price data
        2. Assumption of normal distribution of returns
        3. Reinvestment of returns
        4. Risk-free rate approximated at 2%
        5. No transaction fees included

        Limitations:
        - Past performance does not guarantee future results
        - Extreme volatility of cryptocurrency market
        - Regulatory risk not accounted for
        - Variable asset liquidity
        """
        print(disclosure)
        return disclosure

def main():
    # Initialize optimizer
    optimizer = CryptoPortfolioOptimizer()

    try:
        # Load historical data
        optimizer.fetch_historical_data()

        # Preprocess data
        optimizer.preprocess_data()

        # Optimize portfolio
        optimization_results = optimizer.optimize_portfolio()

        # Visualize efficient frontier
        optimizer.visualize_efficient_frontier(optimization_results)

        # Display results
        print("\nOptimal Portfolio:")
        for crypto, weight in optimization_results['optimal_weights'].items():
            print(f"{crypto}: {weight*100:.2f}%")

        print("\nOptimal Portfolio Metrics:")
        print(f"Return: {optimization_results['optimal_metrics']['return']*100:.2f}%")
        print(f"Volatility: {optimization_results['optimal_metrics']['volatility']*100:.2f}%")
        print(f"Sharpe Ratio: {optimization_results['optimal_metrics']['sharpe_ratio']:.2f}")

        # Ethical disclosure
        optimizer.ethical_disclosure()

    except Exception as e:
        logger.error(f"Execution error: {e}")

if __name__ == "__main__":
    main()