How to Build Bitcoin Treasury ROI Calculator with Ollama: Risk-Return Analysis

# Install Ollama (macOS/Linux)
curl -fsSL https://ollama.ai/install.sh | sh

# Pull the recommended model for financial analysis
ollama pull llama3.1:8b

# Verify installation
ollama list

# requirements.txt
ollama==0.3.1
pandas==2.1.4
numpy==1.24.3
yfinance==0.2.28
plotly==5.17.0
streamlit==1.28.1
python-dotenv==1.0.0

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    """Configuration settings for Bitcoin treasury calculator"""
    
    # Ollama settings
    OLLAMA_HOST = os.getenv('OLLAMA_HOST', 'http://localhost:11434')
    MODEL_NAME = os.getenv('MODEL_NAME', 'llama3.1:8b')
    
    # Bitcoin data sources
    BITCOIN_SYMBOL = 'BTC-USD'
    TREASURY_RATE_SYMBOL = '^TNX'  # 10-year Treasury rate
    
    # Risk parameters
    VAR_CONFIDENCE = 0.05  # 95% confidence level
    CORRELATION_WINDOW = 252  # Trading days for correlation analysis
    VOLATILITY_WINDOW = 30  # Days for volatility calculation

# data_collector.py
import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class BitcoinDataCollector:
    """Collects and processes Bitcoin and treasury data for ROI analysis"""
    
    def __init__(self, config):
        self.config = config
        self.btc_data = None
        self.treasury_data = None
    
    def fetch_bitcoin_data(self, period='2y'):
        """
        Fetch Bitcoin price data with proper error handling
        
        Args:
            period (str): Data period ('1y', '2y', '5y', 'max')
        
        Returns:
            pd.DataFrame: Bitcoin price and volume data
        """
        try:
            btc = yf.Ticker(self.config.BITCOIN_SYMBOL)
            self.btc_data = btc.history(period=period)
            
            # Calculate daily returns
            self.btc_data['Daily_Return'] = self.btc_data['Close'].pct_change()
            
            # Calculate rolling volatility (annualized)
            self.btc_data['Volatility'] = (
                self.btc_data['Daily_Return']
                .rolling(window=self.config.VOLATILITY_WINDOW)
                .std() * np.sqrt(252)
            )
            
            return self.btc_data
            
        except Exception as e:
            raise Exception(f"Failed to fetch Bitcoin data: {str(e)}")
    
    def fetch_treasury_data(self, period='2y'):
        """Fetch 10-year Treasury rate for risk-free rate calculation"""
        try:
            treasury = yf.Ticker(self.config.TREASURY_RATE_SYMBOL)
            self.treasury_data = treasury.history(period=period)
            return self.treasury_data
            
        except Exception as e:
            raise Exception(f"Failed to fetch Treasury data: {str(e)}")
    
    def calculate_sharpe_ratio(self, returns, risk_free_rate=0.02):
        """
        Calculate Sharpe ratio for Bitcoin returns
        
        Args:
            returns (pd.Series): Daily returns
            risk_free_rate (float): Annual risk-free rate
        
        Returns:
            float: Sharpe ratio
        """
        excess_returns = returns - (risk_free_rate / 252)  # Daily risk-free rate
        return np.sqrt(252) * excess_returns.mean() / returns.std()

# risk_analyzer.py
import numpy as np
import pandas as pd
from scipy import stats
import ollama

class BitcoinRiskAnalyzer:
    """Advanced risk analysis for Bitcoin treasury positions"""
    
    def __init__(self, config, data_collector):
        self.config = config
        self.data = data_collector
        self.ollama_client = ollama.Client(host=config.OLLAMA_HOST)
    
    def calculate_var(self, returns, confidence_level=0.05):
        """
        Calculate Value at Risk using historical simulation
        
        Args:
            returns (pd.Series): Daily returns
            confidence_level (float): VaR confidence level
        
        Returns:
            dict: VaR metrics
        """
        # Remove NaN values
        clean_returns = returns.dropna()
        
        # Historical VaR
        var_historical = np.percentile(clean_returns, confidence_level * 100)
        
        # Parametric VaR (assuming normal distribution)
        var_parametric = stats.norm.ppf(confidence_level, 
                                       clean_returns.mean(), 
                                       clean_returns.std())
        
        # Expected Shortfall (Conditional VaR)
        shortfall_returns = clean_returns[clean_returns <= var_historical]
        expected_shortfall = shortfall_returns.mean() if len(shortfall_returns) > 0 else var_historical
        
        return {
            'var_historical': var_historical,
            'var_parametric': var_parametric,
            'expected_shortfall': expected_shortfall,
            'confidence_level': confidence_level
        }
    
    def calculate_maximum_drawdown(self, prices):
        """Calculate maximum drawdown from peak to trough"""
        # Calculate cumulative returns
        cumulative = (1 + prices.pct_change()).cumprod()
        
        # Calculate running maximum
        running_max = cumulative.expanding().max()
        
        # Calculate drawdown
        drawdown = (cumulative - running_max) / running_max
        
        return {
            'max_drawdown': drawdown.min(),
            'max_drawdown_duration': self._calculate_drawdown_duration(drawdown),
            'current_drawdown': drawdown.iloc[-1]
        }
    
    def _calculate_drawdown_duration(self, drawdown):
        """Calculate the longest drawdown duration in days"""
        in_drawdown = drawdown < 0
        drawdown_periods = []
        current_period = 0
        
        for is_dd in in_drawdown:
            if is_dd:
                current_period += 1
            else:
                if current_period > 0:
                    drawdown_periods.append(current_period)
                    current_period = 0
        
        return max(drawdown_periods) if drawdown_periods else 0
    
    def generate_risk_narrative(self, risk_metrics):
        """Use Ollama to generate risk assessment narrative"""
        
        prompt = f"""
        Analyze these Bitcoin treasury risk metrics and provide a concise risk assessment:
        
        Value at Risk (95%): {risk_metrics['var_historical']:.2%}
        Expected Shortfall: {risk_metrics['expected_shortfall']:.2%}
        Maximum Drawdown: {risk_metrics['max_drawdown']:.2%}
        Sharpe Ratio: {risk_metrics.get('sharpe_ratio', 'N/A')}
        
        Provide:
        1. Risk level assessment (Low/Medium/High)
        2. Key concerns for corporate treasury
        3. Recommended position sizing guidance
        
        Keep response under 200 words and focus on actionable insights.
        """
        
        try:
            response = self.ollama_client.generate(
                model=self.config.MODEL_NAME,
                prompt=prompt
            )
            return response['response']
        except Exception as e:
            return f"Risk analysis generation failed: {str(e)}"

# portfolio_optimizer.py
import numpy as np
import pandas as pd
from scipy.optimize import minimize
import ollama

class BitcoinPortfolioOptimizer:
    """Optimize Bitcoin allocation in corporate treasury portfolio"""
    
    def __init__(self, config):
        self.config = config
        self.ollama_client = ollama.Client(host=config.OLLAMA_HOST)
    
    def calculate_optimal_allocation(self, btc_returns, cash_rate=0.02, 
                                   risk_tolerance='medium'):
        """
        Calculate optimal Bitcoin allocation using mean-variance optimization
        
        Args:
            btc_returns (pd.Series): Bitcoin daily returns
            cash_rate (float): Risk-free rate for cash
            risk_tolerance (str): 'low', 'medium', 'high'
        
        Returns:
            dict: Optimization results
        """
        # Define risk tolerance mapping
        risk_mapping = {
            'low': 0.1,      # 10% max allocation
            'medium': 0.25,  # 25% max allocation  
            'high': 0.5      # 50% max allocation
        }
        
        max_allocation = risk_mapping.get(risk_tolerance, 0.25)
        
        # Calculate Bitcoin statistics
        btc_mean_return = btc_returns.mean() * 252  # Annualized
        btc_volatility = btc_returns.std() * np.sqrt(252)  # Annualized
        
        # Objective function: maximize Sharpe ratio
        def objective(weights):
            btc_weight = weights[0]
            cash_weight = 1 - btc_weight
            
            portfolio_return = btc_weight * btc_mean_return + cash_weight * cash_rate
            portfolio_volatility = btc_weight * btc_volatility  # Cash has no volatility
            
            # Sharpe ratio (negative for minimization)
            return -(portfolio_return - cash_rate) / portfolio_volatility if portfolio_volatility > 0 else -999
        
        # Constraints and bounds
        constraints = {'type': 'eq', 'fun': lambda x: 1 - x[0] - (1 - x[0])}  # Weights sum to 1
        bounds = [(0, max_allocation)]  # Bitcoin allocation bounds
        
        # Initial guess
        x0 = [max_allocation / 2]
        
        # Optimize
        result = minimize(objective, x0, method='SLSQP', bounds=bounds, constraints=constraints)
        
        optimal_btc_allocation = result.x[0] if result.success else max_allocation / 2
        
        return {
            'optimal_btc_allocation': optimal_btc_allocation,
            'optimal_cash_allocation': 1 - optimal_btc_allocation,
            'expected_return': optimal_btc_allocation * btc_mean_return + (1 - optimal_btc_allocation) * cash_rate,
            'expected_volatility': optimal_btc_allocation * btc_volatility,
            'optimization_success': result.success,
            'risk_tolerance': risk_tolerance
        }
    
    def calculate_scenario_analysis(self, current_allocation, treasury_size, 
                                  btc_prices, scenarios):
        """
        Perform scenario analysis for different Bitcoin price movements
        
        Args:
            current_allocation (float): Current Bitcoin allocation percentage
            treasury_size (float): Total treasury size in USD
            btc_prices (pd.Series): Historical Bitcoin prices
            scenarios (list): List of scenario dictionaries
        
        Returns:
            pd.DataFrame: Scenario analysis results
        """
        current_btc_price = btc_prices.iloc[-1]
        btc_position_size = treasury_size * current_allocation
        btc_units = btc_position_size / current_btc_price
        
        results = []
        
        for scenario in scenarios:
            price_change = scenario['price_change']
            probability = scenario.get('probability', 1.0)
            
            new_btc_price = current_btc_price * (1 + price_change)
            new_btc_value = btc_units * new_btc_price
            new_total_value = new_btc_value + (treasury_size * (1 - current_allocation))
            
            roi = (new_total_value - treasury_size) / treasury_size
            
            results.append({
                'scenario': scenario['name'],
                'price_change': price_change,
                'probability': probability,
                'new_btc_price': new_btc_price,
                'portfolio_value': new_total_value,
                'roi': roi,
                'weighted_roi': roi * probability
            })
        
        return pd.DataFrame(results)

# app.py
import streamlit as st
import pandas as pd
import plotly.graph_objects as go
import plotly.express as px
from config import Config
from data_collector import BitcoinDataCollector
from risk_analyzer import BitcoinRiskAnalyzer
from portfolio_optimizer import BitcoinPortfolioOptimizer

def main():
    st.set_page_config(
        page_title="Bitcoin Treasury ROI Calculator",
        page_icon="₿",
        layout="wide"
    )
    
    st.title("₿ Bitcoin Treasury ROI Calculator")
    st.markdown("*AI-powered risk-return analysis for corporate Bitcoin strategies*")
    
    # Initialize components
    config = Config()
    collector = BitcoinDataCollector(config)
    risk_analyzer = BitcoinRiskAnalyzer(config, collector)
    optimizer = BitcoinPortfolioOptimizer(config)
    
    # Sidebar inputs
    st.sidebar.header("Treasury Parameters")
    
    treasury_size = st.sidebar.number_input(
        "Total Treasury Size ($M)", 
        min_value=1.0, 
        max_value=10000.0, 
        value=100.0,
        step=10.0
    )
    
    current_allocation = st.sidebar.slider(
        "Current Bitcoin Allocation (%)", 
        min_value=0.0, 
        max_value=50.0, 
        value=5.0,
        step=0.5
    ) / 100
    
    risk_tolerance = st.sidebar.selectbox(
        "Risk Tolerance",
        ["low", "medium", "high"],
        index=1
    )
    
    analysis_period = st.sidebar.selectbox(
        "Analysis Period",
        ["1y", "2y", "5y"],
        index=1
    )
    
    # Load data
    with st.spinner("Loading Bitcoin data..."):
        btc_data = collector.fetch_bitcoin_data(period=analysis_period)
        treasury_data = collector.fetch_treasury_data(period=analysis_period)
    
    # Main dashboard
    col1, col2, col3 = st.columns(3)
    
    with col1:
        current_price = btc_data['Close'].iloc[-1]
        st.metric("Current BTC Price", f"${current_price:,.0f}")
    
    with col2:
        daily_change = btc_data['Daily_Return'].iloc[-1]
        st.metric("24h Change", f"{daily_change:.2%}")
    
    with col3:
        current_volatility = btc_data['Volatility'].iloc[-1]
        st.metric("30-Day Volatility", f"{current_volatility:.1%}")
    
    # Risk Analysis Section
    st.header("📊 Risk Analysis")
    
    # Calculate risk metrics
    returns = btc_data['Daily_Return'].dropna()
    var_metrics = risk_analyzer.calculate_var(returns)
    drawdown_metrics = risk_analyzer.calculate_maximum_drawdown(btc_data['Close'])
    sharpe_ratio = collector.calculate_sharpe_ratio(returns)
    
    # Combine metrics for narrative generation
    risk_metrics = {**var_metrics, **drawdown_metrics, 'sharpe_ratio': sharpe_ratio}
    
    col1, col2 = st.columns(2)
    
    with col1:
        st.subheader("Value at Risk (VaR)")
        st.metric("95% VaR (Daily)", f"{var_metrics['var_historical']:.2%}")
        st.metric("Expected Shortfall", f"{var_metrics['expected_shortfall']:.2%}")
        st.metric("Sharpe Ratio", f"{sharpe_ratio:.2f}")
    
    with col2:
        st.subheader("Drawdown Analysis")
        st.metric("Maximum Drawdown", f"{drawdown_metrics['max_drawdown']:.2%}")
        st.metric("Current Drawdown", f"{drawdown_metrics['current_drawdown']:.2%}")
        st.metric("Max DD Duration", f"{drawdown_metrics['max_drawdown_duration']} days")
    
    # AI Risk Narrative
    st.subheader("🤖 AI Risk Assessment")
    with st.spinner("Generating risk assessment..."):
        risk_narrative = risk_analyzer.generate_risk_narrative(risk_metrics)
        st.write(risk_narrative)
    
    # Portfolio Optimization
    st.header("⚖️ Portfolio Optimization")
    
    optimization_result = optimizer.calculate_optimal_allocation(
        returns, 
        risk_tolerance=risk_tolerance
    )
    
    col1, col2 = st.columns(2)
    
    with col1:
        st.subheader("Optimal Allocation")
        optimal_btc = optimization_result['optimal_btc_allocation']
        st.metric("Recommended BTC Allocation", f"{optimal_btc:.1%}")
        st.metric("Expected Annual Return", f"{optimization_result['expected_return']:.2%}")
        st.metric("Expected Volatility", f"{optimization_result['expected_volatility']:.1%}")
    
    with col2:
        # Allocation comparison chart
        fig = go.Figure(data=[
            go.Bar(name='Current', x=['Bitcoin', 'Cash'], y=[current_allocation*100, (1-current_allocation)*100]),
            go.Bar(name='Optimal', x=['Bitcoin', 'Cash'], y=[optimal_btc*100, (1-optimal_btc)*100])
        ])
        fig.update_layout(title="Current vs Optimal Allocation", yaxis_title="Allocation (%)")
        st.plotly_chart(fig, use_container_width=True)
    
    # Scenario Analysis
    st.header("🎯 Scenario Analysis")
    
    scenarios = [
        {'name': 'Bear Case', 'price_change': -0.5, 'probability': 0.2},
        {'name': 'Base Case', 'price_change': 0.0, 'probability': 0.6},
        {'name': 'Bull Case', 'price_change': 1.0, 'probability': 0.2}
    ]
    
    scenario_results = optimizer.calculate_scenario_analysis(
        current_allocation, 
        treasury_size * 1_000_000,  # Convert to actual dollars
        btc_data['Close'], 
        scenarios
    )
    
    # Scenario results table
    st.subheader("ROI by Scenario")
    display_df = scenario_results[['scenario', 'price_change', 'roi', 'portfolio_value']].copy()
    display_df['price_change'] = display_df['price_change'].apply(lambda x: f"{x:.1%}")
    display_df['roi'] = display_df['roi'].apply(lambda x: f"{x:.2%}")
    display_df['portfolio_value'] = display_df['portfolio_value'].apply(lambda x: f"${x:,.0f}")
    
    st.dataframe(display_df, use_container_width=True)
    
    # Expected ROI calculation
    expected_roi = scenario_results['weighted_roi'].sum()
    st.metric("Expected ROI", f"{expected_roi:.2%}")
    
    # Historical performance chart
    st.header("📈 Historical Performance")
    
    fig = go.Figure()
    fig.add_trace(go.Scatter(
        x=btc_data.index,
        y=btc_data['Close'],
        mode='lines',
        name='Bitcoin Price',
        line=dict(color='orange')
    ))
    
    fig.update_layout(
        title="Bitcoin Price History",
        xaxis_title="Date",
        yaxis_title="Price (USD)",
        hovermode='x unified'
    )
    
    st.plotly_chart(fig, use_container_width=True)
    
    # Export functionality
    st.header("📄 Export Analysis")
    
    if st.button("Generate Executive Summary"):
        # Create summary data
        summary_data = {
            'Treasury Size': f"${treasury_size:.1f}M",
            'Current BTC Allocation': f"{current_allocation:.1%}",
            'Optimal BTC Allocation': f"{optimal_btc:.1%}",
            'Expected Annual Return': f"{optimization_result['expected_return']:.2%}",
            'Value at Risk (95%)': f"{var_metrics['var_historical']:.2%}",
            'Maximum Drawdown': f"{drawdown_metrics['max_drawdown']:.2%}",
            'Sharpe Ratio': f"{sharpe_ratio:.2f}",
            'Expected ROI': f"{expected_roi:.2%}"
        }
        
        # Display as formatted text
        st.subheader("Executive Summary")
        for key, value in summary_data.items():
            st.write(f"**{key}:** {value}")

if __name__ == "__main__":
    main()

# monte_carlo.py
import numpy as np
import pandas as pd

class MonteCarloSimulator:
    """Monte Carlo simulation for Bitcoin treasury scenarios"""
    
    def __init__(self, btc_returns):
        self.returns = btc_returns.dropna()
        self.mean_return = self.returns.mean()
        self.volatility = self.returns.std()
    
    def simulate_paths(self, initial_price, days=252, num_simulations=10000):
        """
        Generate Monte Carlo price paths for Bitcoin
        
        Args:
            initial_price (float): Starting Bitcoin price
            days (int): Number of days to simulate
            num_simulations (int): Number of simulation paths
        
        Returns:
            np.array: Price paths (simulations x days)
        """
        # Generate random daily returns
        random_returns = np.random.normal(
            self.mean_return, 
            self.volatility, 
            (num_simulations, days)
        )
        
        # Calculate price paths
        price_paths = np.zeros((num_simulations, days + 1))
        price_paths[:, 0] = initial_price
        
        for day in range(days):
            price_paths[:, day + 1] = price_paths[:, day] * (1 + random_returns[:, day])
        
        return price_paths
    
    def calculate_portfolio_outcomes(self, price_paths, allocation, treasury_size):
        """Calculate portfolio value outcomes from price simulations"""
        
        initial_btc_price = price_paths[:, 0][0]  # Same for all simulations
        final_btc_prices = price_paths[:, -1]
        
        # Calculate portfolio values
        btc_position_size = treasury_size * allocation
        btc_units = btc_position_size / initial_btc_price
        cash_position = treasury_size * (1 - allocation)
        
        final_btc_values = btc_units * final_btc_prices
        final_portfolio_values = final_btc_values + cash_position
        
        # Calculate returns
        portfolio_returns = (final_portfolio_values - treasury_size) / treasury_size
        
        return {
            'final_values': final_portfolio_values,
            'returns': portfolio_returns,
            'percentiles': {
                '5th': np.percentile(portfolio_returns, 5),
                '25th': np.percentile(portfolio_returns, 25),
                '50th': np.percentile(portfolio_returns, 50),
                '75th': np.percentile(portfolio_returns, 75),
                '95th': np.percentile(portfolio_returns, 95)
            },
            'probability_positive': np.mean(portfolio_returns > 0),
            'probability_loss_10pct': np.mean(portfolio_returns < -0.1)
        }

# Run the Streamlit application
streamlit run app.py

# Test with sample data
python -m pytest tests/ -v

# Docker deployment
docker build -t bitcoin-treasury-calculator .
docker run -p 8501:8501 bitcoin-treasury-calculator

# backtesting.py
def backtest_strategy(btc_data, allocation_strategy, rebalance_frequency='monthly'):
    """
    Backtest Bitcoin treasury allocation strategy
    
    Args:
        btc_data (pd.DataFrame): Historical Bitcoin data
        allocation_strategy (function): Function that returns optimal allocation
        rebalance_frequency (str): 'daily', 'weekly', 'monthly', 'quarterly'
    
    Returns:
        dict: Backtesting results
    """
    # Implementation details for strategy backtesting
    # This would include performance metrics, drawdown analysis, 
    # and comparison to benchmark portfolios
    pass

# compliance_reporting.py
def generate_regulatory_report(portfolio_data, jurisdiction='US'):
    """Generate compliance reports for Bitcoin treasury holdings"""
    
    reports = {
        'US': generate_sec_filing_data,
        'EU': generate_mifid_report,
        'UK': generate_fca_compliance
    }
    
    return reports.get(jurisdiction, generate_generic_report)(portfolio_data)