Tax-Loss Harvesting Bot: Ollama Cryptocurrency Tax Optimization Guide

# Install Ollama
curl -fsSL https://ollama.ai/install.sh | sh

# Pull the Code Llama model for financial analysis
ollama pull codellama:13b

# Install Python dependencies
pip install ccxt pandas numpy ollama python-dotenv

import ccxt
import pandas as pd
import numpy as np
from ollama import Client
from datetime import datetime, timedelta
import json
import logging

class TaxLossHarvestingBot:
    def __init__(self, config_file='config.json'):
        self.config = self.load_config(config_file)
        self.ollama_client = Client(host='http://localhost:11434')
        self.exchanges = self.setup_exchanges()
        self.logger = self.setup_logging()
        
    def load_config(self, config_file):
        """Load exchange API keys and bot parameters"""
        with open(config_file, 'r') as f:
            return json.load(f)
    
    def setup_exchanges(self):
        """Initialize exchange connections"""
        exchanges = {}
        for exchange_name, credentials in self.config['exchanges'].items():
            exchange_class = getattr(ccxt, exchange_name)
            exchanges[exchange_name] = exchange_class({
                'apiKey': credentials['api_key'],
                'secret': credentials['secret'],
                'sandbox': credentials.get('sandbox', True)
            })
        return exchanges

def analyze_tax_harvest_opportunity(self, portfolio_data, market_data):
    """Use Ollama to analyze tax harvesting opportunities"""
    
    prompt = f"""
    Analyze this cryptocurrency portfolio for tax-loss harvesting opportunities:
    
    Portfolio Data:
    {json.dumps(portfolio_data, indent=2)}
    
    Current Market Prices:
    {json.dumps(market_data, indent=2)}
    
    Rules:
    1. Identify positions with unrealized losses > 5%
    2. Check for wash sale violations (30-day rule)
    3. Suggest similar assets to maintain exposure
    4. Calculate tax benefit vs transaction costs
    5. Prioritize by tax savings potential
    
    Return JSON format:
    {{
        "harvest_opportunities": [
            {{
                "symbol": "BTC/USD",
                "current_loss": -1250.00,
                "tax_benefit": 312.50,
                "action": "sell",
                "replacement_asset": "WBTC",
                "reasoning": "Large unrealized loss, no wash sale risk"
            }}
        ],
        "total_potential_savings": 850.00
    }}
    """
    
    response = self.ollama_client.generate(
        model='codellama:13b',
        prompt=prompt,
        stream=False
    )
    
    try:
        # Parse Ollama response
        analysis = json.loads(response['response'])
        return analysis
    except json.JSONDecodeError:
        self.logger.error("Failed to parse Ollama response")
        return {"harvest_opportunities": [], "total_potential_savings": 0}

def get_portfolio_snapshot(self):
    """Fetch current portfolio across all exchanges"""
    consolidated_portfolio = {}
    
    for exchange_name, exchange in self.exchanges.items():
        try:
            balance = exchange.fetch_balance()
            positions = exchange.fetch_positions()
            
            for symbol, position in positions.items():
                if position['contracts'] > 0:  # Only active positions
                    consolidated_portfolio[symbol] = {
                        'quantity': position['contracts'],
                        'entry_price': position['entryPrice'],
                        'current_price': position['markPrice'],
                        'unrealized_pnl': position['unrealizedPnl'],
                        'exchange': exchange_name,
                        'timestamp': datetime.now().isoformat()
                    }
                    
        except Exception as e:
            self.logger.error(f"Error fetching {exchange_name} data: {e}")
            
    return consolidated_portfolio

def calculate_tax_implications(self, portfolio):
    """Calculate potential tax savings from harvesting losses"""
    total_unrealized_losses = 0
    harvest_candidates = []
    
    for symbol, position in portfolio.items():
        unrealized_loss = position['unrealized_pnl']
        
        if unrealized_loss < 0:  # Position is at a loss
            loss_percentage = abs(unrealized_loss) / (position['quantity'] * position['entry_price'])
            
            if loss_percentage >= 0.05:  # 5% loss threshold
                tax_rate = self.config.get('tax_rate', 0.25)  # 25% default
                potential_savings = abs(unrealized_loss) * tax_rate
                
                harvest_candidates.append({
                    'symbol': symbol,
                    'unrealized_loss': unrealized_loss,
                    'loss_percentage': loss_percentage,
                    'potential_tax_savings': potential_savings,
                    'position': position
                })
                
                total_unrealized_losses += abs(unrealized_loss)
    
    return {
        'candidates': sorted(harvest_candidates, key=lambda x: x['potential_tax_savings'], reverse=True),
        'total_losses': total_unrealized_losses,
        'total_potential_savings': total_unrealized_losses * self.config.get('tax_rate', 0.25)
    }

def execute_tax_harvest(self, harvest_plan):
    """Execute tax harvesting trades with safety controls"""
    
    executed_trades = []
    
    for opportunity in harvest_plan['harvest_opportunities']:
        try:
            # Pre-execution validation
            if not self.validate_trade_safety(opportunity):
                continue
                
            # Execute the sell order
            exchange = self.exchanges[opportunity['exchange']]
            
            sell_order = exchange.create_market_sell_order(
                symbol=opportunity['symbol'],
                amount=opportunity['quantity']
            )
            
            # Log the trade
            self.logger.info(f"Executed tax harvest: {opportunity['symbol']} for ${opportunity['current_loss']} loss")
            
            executed_trades.append({
                'symbol': opportunity['symbol'],
                'order_id': sell_order['id'],
                'loss_realized': opportunity['current_loss'],
                'tax_benefit': opportunity['tax_benefit'],
                'timestamp': datetime.now().isoformat()
            })
            
            # Optional: Buy replacement asset to maintain exposure
            if opportunity.get('replacement_asset'):
                self.execute_replacement_trade(opportunity)
                
        except Exception as e:
            self.logger.error(f"Failed to execute harvest for {opportunity['symbol']}: {e}")
    
    return executed_trades

def validate_trade_safety(self, opportunity):
    """Safety checks before executing trades"""
    
    # Check wash sale rule (30 days)
    if self.check_wash_sale_violation(opportunity['symbol']):
        self.logger.warning(f"Wash sale risk for {opportunity['symbol']}")
        return False
    
    # Minimum loss threshold
    if abs(opportunity['current_loss']) < self.config.get('min_loss_amount', 100):
        return False
    
    # Maximum daily trading limit
    daily_volume = self.get_daily_trading_volume()
    if daily_volume > self.config.get('max_daily_volume', 10000):
        return False
        
    return True

def enhanced_tax_analysis(self, portfolio_data):
    """Use multiple Ollama models for comprehensive analysis"""
    
    models = ['codellama:13b', 'mistral:7b', 'llama2:13b']
    analyses = []
    
    for model in models:
        try:
            response = self.ollama_client.generate(
                model=model,
                prompt=self.create_analysis_prompt(portfolio_data),
                stream=False
            )
            analyses.append(json.loads(response['response']))
        except Exception as e:
            self.logger.error(f"Model {model} failed: {e}")
    
    # Consensus-based decision making
    return self.aggregate_model_decisions(analyses)

def aggregate_model_decisions(self, analyses):
    """Combine multiple model outputs for robust decisions"""
    
    consensus_opportunities = []
    
    # Find opportunities agreed upon by majority of models
    for analysis in analyses:
        for opportunity in analysis['harvest_opportunities']:
            vote_count = sum(1 for a in analyses 
                           if any(op['symbol'] == opportunity['symbol'] 
                                 for op in a['harvest_opportunities']))
            
            if vote_count >= len(analyses) // 2 + 1:  # Majority consensus
                consensus_opportunities.append(opportunity)
    
    return {
        'harvest_opportunities': consensus_opportunities,
        'confidence_score': len(consensus_opportunities) / len(analyses[0]['harvest_opportunities']) if analyses else 0
    }

FROM python:3.11-slim

# Install Ollama
RUN curl -fsSL https://ollama.ai/install.sh | sh

# Copy application
COPY . /app
WORKDIR /app

# Install dependencies
RUN pip install -r requirements.txt

# Download models
RUN ollama pull codellama:13b
RUN ollama pull mistral:7b

# Start services
CMD ["python", "tax_harvesting_bot.py"]

def setup_monitoring(self):
    """Configure monitoring and alerting"""
    
    # Prometheus metrics
    from prometheus_client import Counter, Histogram, start_http_server
    
    self.trades_executed = Counter('tax_harvest_trades_total', 'Total tax harvest trades')
    self.tax_savings = Counter('tax_savings_total', 'Total tax savings realized')
    self.processing_time = Histogram('analysis_duration_seconds', 'Time spent on analysis')
    
    # Start metrics server
    start_http_server(8000)

def send_daily_report(self, results):
    """Send daily performance summary"""
    
    report = f"""
    Tax Harvesting Bot Daily Report
    
    Trades Executed: {len(results['executed_trades'])}
    Total Tax Savings: ${results['total_savings']:.2f}
    Portfolio Impact: {results['portfolio_change']:.2%}
    
    Top Opportunities:
    {self.format_opportunities(results['top_opportunities'])}
    """
    
    # Send via email, Slack, or monitoring system
    self.send_notification(report)

def backtest_strategy(self, start_date, end_date):
    """Backtest tax harvesting strategy performance"""
    
    historical_data = self.fetch_historical_data(start_date, end_date)
    simulated_trades = []
    total_savings = 0
    
    for date, market_data in historical_data.items():
        # Simulate Ollama analysis
        opportunities = self.analyze_tax_harvest_opportunity(
            self.simulate_portfolio_state(date),
            market_data
        )
        
        for trade in opportunities['harvest_opportunities']:
            simulated_trades.append({
                'date': date,
                'symbol': trade['symbol'],
                'loss': trade['current_loss'],
                'tax_benefit': trade['tax_benefit']
            })
            total_savings += trade['tax_benefit']
    
    return {
        'total_trades': len(simulated_trades),
        'total_tax_savings': total_savings,
        'average_savings_per_trade': total_savings / len(simulated_trades) if simulated_trades else 0,
        'trades': simulated_trades
    }

def calculate_annual_roi(self, bot_costs, tax_savings):
    """Calculate return on investment for the tax harvesting bot"""
    
    # Bot operational costs
    infrastructure_cost = 1200  # Annual server costs
    development_time = 2000     # Initial development (amortized)
    maintenance_cost = 500      # Annual maintenance
    
    total_costs = infrastructure_cost + development_time + maintenance_cost
    net_savings = tax_savings - total_costs
    roi_percentage = (net_savings / total_costs) * 100
    
    return {
        'total_costs': total_costs,
        'tax_savings': tax_savings,
        'net_benefit': net_savings,
        'roi_percentage': roi_percentage,
        'payback_period_months': (total_costs / (tax_savings / 12)) if tax_savings > 0 else float('inf')
    }

def secure_api_setup(self):
    """Implement secure API key management"""
    
    # Use environment variables
    import os
    from cryptography.fernet import Fernet
    
    # Encrypt API keys at rest
    encryption_key = os.environ.get('ENCRYPTION_KEY')
    cipher_suite = Fernet(encryption_key)
    
    # Read-only trading permissions
    self.exchange_permissions = {
        'read_portfolio': True,
        'place_orders': True,
        'withdraw_funds': False,  # Never allow withdrawals
        'transfer_funds': False
    }

def calculate_market_impact(self, symbol, trade_size):
    """Estimate market impact before executing large trades"""
    
    order_book = self.exchanges['primary'].fetch_order_book(symbol)
    
    # Calculate depth and spread
    bid_depth = sum(order['amount'] for order in order_book['bids'][:10])
    ask_depth = sum(order['amount'] for order in order_book['asks'][:10])
    
    # Estimate slippage
    estimated_slippage = trade_size / bid_depth * 0.1  # Simple model
    
    return {
        'recommended_max_size': bid_depth * 0.1,  # 10% of depth
        'estimated_slippage': estimated_slippage,
        'execute_as_twap': trade_size > bid_depth * 0.05  # TWAP for large orders
    }

def identify_cross_chain_opportunities(self, portfolio):
    """Find arbitrage opportunities across different blockchains"""
    
    prompt = f"""
    Analyze cross-chain arbitrage opportunities for tax harvesting:
    
    Portfolio: {json.dumps(portfolio, indent=2)}
    
    Consider:
    1. Price differences between chains (Ethereum vs BSC vs Polygon)
    2. Bridge costs and timing
    3. Gas fee optimization
    4. Liquidity availability
    
    Suggest profitable harvest-and-rebalance strategies.
    """
    
    response = self.ollama_client.generate(
        model='codellama:13b',
        prompt=prompt
    )
    
    return self.parse_arbitrage_analysis(response['response'])

def integrate_defi_strategies(self, harvested_proceeds):
    """Deploy harvested proceeds into DeFi for additional yield"""
    
    defi_strategies = [
        {
            'protocol': 'Aave',
            'action': 'lend_stablecoins',
            'apy': 0.045,
            'risk_level': 'low'
        },
        {
            'protocol': 'Uniswap',
            'action': 'provide_liquidity',
            'apy': 0.12,
            'risk_level': 'medium'
        }
    ]
    
    # Let Ollama choose optimal strategy
    strategy_prompt = f"""
    Choose optimal DeFi strategy for ${harvested_proceeds} in proceeds:
    
    Available strategies: {json.dumps(defi_strategies, indent=2)}
    
    Current market conditions: {self.get_market_sentiment()}
    
    Prioritize: 1) Capital preservation 2) Tax efficiency 3) Yield
    """
    
    recommendation = self.ollama_client.generate(
        model='mistral:7b',
        prompt=strategy_prompt
    )
    
    return self.execute_defi_strategy(recommendation)