YieldStreet Alternative Assets: Building Diversified Yield Strategies with Code

# requirements.txt
import pandas as pd
import numpy as np
import requests
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta
import yfinance as yf  # For market comparison data
import json

# Install dependencies
# pip install pandas numpy requests matplotlib seaborn yfinance

class YieldStreetAnalyzer:
    def __init__(self):
        self.investments = []
        self.portfolio_value = 0
        self.target_yield = 0.10  # 10% target yield
    
    def add_investment(self, investment_data):
        """
        Add a YieldStreet investment to tracking portfolio
        
        Args:
            investment_data (dict): Investment details including:
                - name: Investment name
                - category: Asset category (real estate, art, etc.)
                - amount: Investment amount in USD
                - target_yield: Expected annual yield percentage
                - term: Investment term in months
                - risk_level: Risk rating (1-5 scale)
        """
        investment = {
            'name': investment_data['name'],
            'category': investment_data['category'],
            'amount': investment_data['amount'],
            'target_yield': investment_data['target_yield'],
            'term_months': investment_data['term'],
            'risk_level': investment_data['risk_level'],
            'date_added': datetime.now(),
            'expected_return': investment_data['amount'] * investment_data['target_yield']
        }
        
        self.investments.append(investment)
        self.portfolio_value += investment_data['amount']
        print(f"Added {investment['name']} - ${investment['amount']:,} at {investment['target_yield']*100:.1f}% yield")
    
    def calculate_portfolio_metrics(self):
        """Calculate key portfolio performance metrics"""
        if not self.investments:
            return None
            
        df = pd.DataFrame(self.investments)
        
        metrics = {
            'total_invested': df['amount'].sum(),
            'weighted_avg_yield': (df['amount'] * df['target_yield']).sum() / df['amount'].sum(),
            'avg_risk_level': df['risk_level'].mean(),
            'category_allocation': df.groupby('category')['amount'].sum() / df['amount'].sum(),
            'expected_annual_income': df['expected_return'].sum(),
            'portfolio_count': len(df)
        }
        
        return metrics
    
    def diversification_score(self):
        """Calculate portfolio diversification using Herfindahl index"""
        if not self.investments:
            return 0
            
        df = pd.DataFrame(self.investments)
        category_weights = df.groupby('category')['amount'].sum() / df['amount'].sum()
        herfindahl_index = (category_weights ** 2).sum()
        
        # Convert to diversification score (1 = perfectly diversified, 0 = concentrated)
        max_categories = len(category_weights)
        diversification_score = (1 - herfindahl_index) / (1 - 1/max_categories) if max_categories > 1 else 0
        
        return diversification_score

def optimize_yield_strategy(available_capital, risk_tolerance=3):
    """
    Create optimized allocation strategy for YieldStreet investments
    
    Args:
        available_capital (float): Total investment capital
        risk_tolerance (int): Risk tolerance 1-5 (1=conservative, 5=aggressive)
    
    Returns:
        dict: Recommended allocation by category
    """
    
    # YieldStreet asset categories with typical yields and risk levels
    asset_categories = {
        'real_estate': {'yield': 0.10, 'risk': 2, 'min_investment': 1000},
        'art_collectibles': {'yield': 0.12, 'risk': 4, 'min_investment': 2500},
        'legal_finance': {'yield': 0.15, 'risk': 5, 'min_investment': 5000},
        'marine_finance': {'yield': 0.11, 'risk': 3, 'min_investment': 1000},
        'consumer_finance': {'yield': 0.08, 'risk': 2, 'min_investment': 500}
    }
    
    # Filter categories based on risk tolerance
    suitable_categories = {
        category: data for category, data in asset_categories.items() 
        if data['risk'] <= risk_tolerance
    }
    
    # Calculate optimal allocation weights
    total_yield_score = sum(data['yield'] / data['risk'] for data in suitable_categories.values())
    
    allocation = {}
    for category, data in suitable_categories.items():
        # Weight by yield-to-risk ratio
        weight = (data['yield'] / data['risk']) / total_yield_score
        allocated_amount = available_capital * weight
        
        # Ensure minimum investment requirements are met
        if allocated_amount >= data['min_investment']:
            allocation[category] = {
                'amount': allocated_amount,
                'percentage': weight * 100,
                'expected_yield': data['yield'],
                'risk_level': data['risk']
            }
    
    return allocation

# Example usage
recommended_allocation = optimize_yield_strategy(50000, risk_tolerance=3)

print("Recommended Portfolio Allocation:")
print("-" * 40)
for category, details in recommended_allocation.items():
    print(f"{category.replace('_', ' ').title()}: ${details['amount']:,.0f} ({details['percentage']:.1f}%)")
    print(f"  Expected Yield: {details['expected_yield']*100:.1f}% | Risk Level: {details['risk_level']}/5")
    print()

def create_performance_dashboard(analyzer):
    """Generate visual dashboard for YieldStreet portfolio performance"""
    
    metrics = analyzer.calculate_portfolio_metrics()
    if not metrics:
        print("No investments to analyze")
        return
    
    # Create subplot figure
    fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(15, 10))
    fig.suptitle('YieldStreet Alternative Assets Portfolio Dashboard', fontsize=16, fontweight='bold')
    
    # 1. Category Allocation Pie Chart
    category_data = metrics['category_allocation']
    ax1.pie(category_data.values, labels=category_data.index, autopct='%1.1f%%', startangle=90)
    ax1.set_title('Portfolio Allocation by Category')
    
    # 2. Yield vs Risk Scatter Plot
    df = pd.DataFrame(analyzer.investments)
    scatter = ax2.scatter(df['risk_level'], df['target_yield']*100, 
                         s=df['amount']/1000, alpha=0.6, c=df['risk_level'], cmap='RdYlGn_r')
    ax2.set_xlabel('Risk Level (1-5)')
    ax2.set_ylabel('Target Yield (%)')
    ax2.set_title('Risk vs Yield Distribution')
    ax2.grid(True, alpha=0.3)
    
    # 3. Expected Monthly Income Timeline
    monthly_income = metrics['expected_annual_income'] / 12
    months = range(1, 13)
    cumulative_income = [monthly_income * month for month in months]
    ax3.plot(months, cumulative_income, marker='o', linewidth=2)
    ax3.set_xlabel('Month')
    ax3.set_ylabel('Cumulative Income ($)')
    ax3.set_title(f'Projected Annual Income: ${metrics["expected_annual_income"]:,.0f}')
    ax3.grid(True, alpha=0.3)
    
    # 4. Diversification Metrics
    diversification = analyzer.diversification_score()
    risk_metrics = ['Portfolio Count', 'Diversification Score', 'Avg Risk Level', 'Weighted Yield']
    values = [metrics['portfolio_count'], diversification, 
              metrics['avg_risk_level'], metrics['weighted_avg_yield']*100]
    
    bars = ax4.bar(range(len(risk_metrics)), values, color=['skyblue', 'lightgreen', 'orange', 'lightcoral'])
    ax4.set_xticks(range(len(risk_metrics)))
    ax4.set_xticklabels(risk_metrics, rotation=45, ha='right')
    ax4.set_title('Portfolio Quality Metrics')
    
    # Add value labels on bars
    for i, (bar, value) in enumerate(zip(bars, values)):
        height = bar.get_height()
        if i == 1:  # Diversification score
            ax4.text(bar.get_x() + bar.get_width()/2., height + 0.01, f'{value:.2f}',
                    ha='center', va='bottom')
        elif i == 3:  # Weighted yield
            ax4.text(bar.get_x() + bar.get_width()/2., height + 0.5, f'{value:.1f}%',
                    ha='center', va='bottom')
        else:
            ax4.text(bar.get_x() + bar.get_width()/2., height + 0.1, f'{value:.1f}',
                    ha='center', va='bottom')
    
    plt.tight_layout()
    plt.show()
    
    # Print summary statistics
    print("\n" + "="*50)
    print("PORTFOLIO PERFORMANCE SUMMARY")
    print("="*50)
    print(f"Total Invested: ${metrics['total_invested']:,}")
    print(f"Weighted Average Yield: {metrics['weighted_avg_yield']*100:.2f}%")
    print(f"Expected Annual Income: ${metrics['expected_annual_income']:,}")
    print(f"Portfolio Diversification Score: {diversification:.2f}/1.00")
    print(f"Average Risk Level: {metrics['avg_risk_level']:.1f}/5.0")

# Initialize the analyzer
portfolio = YieldStreetAnalyzer()

# Add sample YieldStreet investments
sample_investments = [
    {
        'name': 'Brooklyn Rental Properties Fund',
        'category': 'real_estate',
        'amount': 15000,
        'target_yield': 0.095,
        'term': 24,
        'risk_level': 2
    },
    {
        'name': 'Contemporary Art Collection',
        'category': 'art_collectibles', 
        'amount': 10000,
        'target_yield': 0.12,
        'term': 36,
        'risk_level': 4
    },
    {
        'name': 'Commercial Litigation Fund',
        'category': 'legal_finance',
        'amount': 8000,
        'target_yield': 0.16,
        'term': 18,
        'risk_level': 5
    },
    {
        'name': 'Container Shipping Fund',
        'category': 'marine_finance',
        'amount': 12000,
        'target_yield': 0.11,
        'term': 30,
        'risk_level': 3
    },
    {
        'name': 'Consumer Credit Portfolio',
        'category': 'consumer_finance',
        'amount': 5000,
        'target_yield': 0.085,
        'term': 12,
        'risk_level': 2
    }
]

# Add investments to portfolio
for investment in sample_investments:
    portfolio.add_investment(investment)

# Generate performance dashboard
create_performance_dashboard(portfolio)

def implement_dca_strategy(monthly_budget, target_categories, months=12):
    """
    Implement dollar-cost averaging for YieldStreet investments
    
    Args:
        monthly_budget (float): Monthly investment budget
        target_categories (list): Preferred asset categories
        months (int): Investment timeline in months
    """
    
    monthly_allocations = []
    
    for month in range(1, months + 1):
        # Rotate through categories to maintain diversification
        category_index = (month - 1) % len(target_categories)
        selected_category = target_categories[category_index]
        
        allocation = {
            'month': month,
            'category': selected_category,
            'amount': monthly_budget,
            'cumulative_invested': monthly_budget * month
        }
        
        monthly_allocations.append(allocation)
    
    return monthly_allocations

# Example DCA strategy
dca_plan = implement_dca_strategy(
    monthly_budget=2000,
    target_categories=['real_estate', 'marine_finance', 'consumer_finance'],
    months=12
)

print("Dollar-Cost Averaging Plan:")
for allocation in dca_plan[:6]:  # Show first 6 months
    print(f"Month {allocation['month']}: ${allocation['amount']} → {allocation['category']}")

def calculate_sharpe_ratio(portfolio_return, risk_free_rate=0.05, portfolio_volatility=0.15):
    """
    Calculate Sharpe ratio for alternative asset portfolio
    
    Args:
        portfolio_return (float): Expected portfolio return
        risk_free_rate (float): Risk-free rate (typically treasury bonds)
        portfolio_volatility (float): Portfolio standard deviation
    """
    
    excess_return = portfolio_return - risk_free_rate
    sharpe_ratio = excess_return / portfolio_volatility
    
    return sharpe_ratio

# Calculate for our sample portfolio
portfolio_metrics = portfolio.calculate_portfolio_metrics()
sharpe = calculate_sharpe_ratio(portfolio_metrics['weighted_avg_yield'])

print(f"Portfolio Sharpe Ratio: {sharpe:.2f}")
print("Interpretation:")
if sharpe > 1.0:
    print("✅ Excellent risk-adjusted returns")
elif sharpe > 0.5:
    print("✅ Good risk-adjusted returns") 
else:
    print("⚠️ Consider reducing risk or finding higher yields")

def optimize_tax_efficiency(investments, tax_bracket=0.24):
    """
    Analyze tax implications of YieldStreet portfolio
    
    Args:
        investments (list): List of investment dictionaries
        tax_bracket (float): Marginal tax rate
    """
    
    df = pd.DataFrame(investments)
    
    # Different asset classes have different tax treatments
    tax_rates = {
        'real_estate': 0.20,  # Long-term capital gains rate
        'art_collectibles': 0.28,  # Collectibles tax rate
        'legal_finance': tax_bracket,  # Ordinary income
        'marine_finance': tax_bracket,  # Ordinary income  
        'consumer_finance': tax_bracket  # Ordinary income
    }
    
    df['applicable_tax_rate'] = df['category'].map(tax_rates)
    df['after_tax_yield'] = df['target_yield'] * (1 - df['applicable_tax_rate'])
    df['tax_burden'] = df['expected_return'] * df['applicable_tax_rate']
    
    return df[['name', 'category', 'target_yield', 'after_tax_yield', 'tax_burden']]

# Analyze tax efficiency
tax_analysis = optimize_tax_efficiency(sample_investments)
print("Tax-Adjusted Performance Analysis:")
print(tax_analysis.to_string(index=False, float_format='%.3f'))

def rebalance_portfolio(current_portfolio, target_allocation, rebalance_threshold=0.05):
    """
    Determine if portfolio rebalancing is needed
    
    Args:
        current_portfolio (dict): Current asset allocation percentages
        target_allocation (dict): Target asset allocation percentages  
        rebalance_threshold (float): Trigger threshold for rebalancing
    """
    
    rebalancing_needed = False
    rebalancing_actions = []
    
    for category in target_allocation:
        current_weight = current_portfolio.get(category, 0)
        target_weight = target_allocation[category]
        difference = abs(current_weight - target_weight)
        
        if difference > rebalance_threshold:
            rebalancing_needed = True
            action = "INCREASE" if current_weight < target_weight else "DECREASE"
            rebalancing_actions.append({
                'category': category,
                'action': action,
                'current_weight': current_weight,
                'target_weight': target_weight,
                'difference': difference
            })
    
    return rebalancing_needed, rebalancing_actions

# Example rebalancing check
current_allocation = {
    'real_estate': 0.35,
    'art_collectibles': 0.15,
    'legal_finance': 0.20,
    'marine_finance': 0.20,
    'consumer_finance': 0.10
}

target_allocation = {
    'real_estate': 0.30,
    'art_collectibles': 0.20,
    'legal_finance': 0.15,
    'marine_finance': 0.25,
    'consumer_finance': 0.10
}

needs_rebalancing, actions = rebalance_portfolio(current_allocation, target_allocation)

if needs_rebalancing:
    print("Portfolio Rebalancing Required:")
    for action in actions:
        print(f"{action['action']} {action['category']}: {action['current_weight']:.1%} → {action['target_weight']:.1%}")

def benchmark_performance(portfolio_yield, benchmark_symbol='SPY', period='1y'):
    """
    Compare YieldStreet portfolio performance against market benchmarks
    """
    
    # Fetch benchmark performance
    benchmark = yf.Ticker(benchmark_symbol)
    hist = benchmark.history(period=period)
    benchmark_return = (hist['Close'][-1] / hist['Close'][0]) - 1
    
    # Add dividend yield for total return
    info = benchmark.info
    dividend_yield = info.get('dividendYield', 0) or 0
    total_benchmark_return = benchmark_return + dividend_yield
    
    print(f"Performance Comparison ({period}):")
    print(f"YieldStreet Portfolio: {portfolio_yield*100:.2f}%")
    print(f"{benchmark_symbol} Total Return: {total_benchmark_return*100:.2f}%")
    print(f"Outperformance: {(portfolio_yield - total_benchmark_return)*100:+.2f}%")
    
    return portfolio_yield > total_benchmark_return

# Benchmark our portfolio
portfolio_metrics = portfolio.calculate_portfolio_metrics()
outperforming = benchmark_performance(portfolio_metrics['weighted_avg_yield'])

def check_concentration_risk(portfolio, max_single_category=0.40):
    """Check for over-concentration in any single category"""
    
    metrics = portfolio.calculate_portfolio_metrics()
    category_allocation = metrics['category_allocation']
    
    warnings = []
    for category, allocation in category_allocation.items():
        if allocation > max_single_category:
            warnings.append(f"⚠️ Over-concentrated in {category}: {allocation:.1%}")
    
    return warnings

# Check for concentration risk
concentration_warnings = check_concentration_risk(portfolio)
for warning in concentration_warnings:
    print(warning)

def create_liquidity_ladder(investments, emergency_fund_months=6):
    """
    Analyze investment liquidity and suggest improvements
    """
    
    df = pd.DataFrame(investments)
    
    # Calculate liquidity metrics
    total_invested = df['amount'].sum()
    monthly_expenses = total_invested * 0.04 / 12  # Assume 4% expense ratio
    emergency_fund_needed = monthly_expenses * emergency_fund_months
    
    # Sort by term length
    df_sorted = df.sort_values('term_months')
    
    print("Liquidity Ladder Analysis:")
    print(f"Emergency Fund Needed: ${emergency_fund_needed:,.0f}")
    print("\nInvestment Maturity Schedule:")
    
    cumulative_maturing = 0
    for _, investment in df_sorted.iterrows():
        cumulative_maturing += investment['amount']
        print(f"Month {investment['term_months']:2d}: ${investment['amount']:,} (Total: ${cumulative_maturing:,})")

create_liquidity_ladder(sample_investments)