How to Time Crypto Market Entries with Ollama: DCA vs Lump Sum Strategy

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

# Download financial analysis model
ollama pull llama3.1:8b

# Verify installation
ollama list

# requirements.txt
ollama==0.2.1
ccxt==4.1.45
pandas==2.0.3
numpy==1.24.3
requests==2.31.0
python-dotenv==1.0.0

pip install -r requirements.txt
mkdir crypto-ollama-timing
cd crypto-ollama-timing
touch config.py market_analyzer.py dca_strategy.py lump_sum_strategy.py

# market_analyzer.py
import ccxt
import pandas as pd
import ollama
from datetime import datetime, timedelta
import json

class CryptoMarketAnalyzer:
    def __init__(self, exchange_name='binance'):
        self.exchange = getattr(ccxt, exchange_name)()
        self.client = ollama.Client()
        
    def get_market_data(self, symbol, timeframe='1h', limit=100):
        """Fetch OHLCV data for analysis"""
        try:
            ohlcv = self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
            df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
            df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
            return df
        except Exception as e:
            print(f"Error fetching data: {e}")
            return None
    
    def calculate_technical_indicators(self, df):
        """Calculate RSI, moving averages, and volatility"""
        # Simple Moving Averages
        df['sma_20'] = df['close'].rolling(window=20).mean()
        df['sma_50'] = df['close'].rolling(window=50).mean()
        
        # RSI calculation
        delta = df['close'].diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
        rs = gain / loss
        df['rsi'] = 100 - (100 / (1 + rs))
        
        # Volatility (24-hour)
        df['volatility'] = df['close'].rolling(window=24).std()
        
        return df
    
    def analyze_market_sentiment(self, df, symbol):
        """Use Ollama to analyze market conditions"""
        latest_data = df.tail(5)
        current_price = df['close'].iloc[-1]
        sma_20 = df['sma_20'].iloc[-1]
        sma_50 = df['sma_50'].iloc[-1]
        rsi = df['rsi'].iloc[-1]
        volatility = df['volatility'].iloc[-1]
        
        prompt = f"""
        Analyze this cryptocurrency market data for {symbol}:
        
        Current Price: ${current_price:.2f}
        20-day SMA: ${sma_20:.2f}
        50-day SMA: ${sma_50:.2f}
        RSI: {rsi:.2f}
        24h Volatility: {volatility:.2f}
        
        Price trend (last 5 periods): {latest_data['close'].tolist()}
        
        Provide a market sentiment analysis with:
        1. Current trend direction (bullish/bearish/neutral)
        2. Entry timing recommendation (good/wait/avoid)
        3. Risk level (low/medium/high)
        4. Reasoning in 2-3 sentences
        
        Format as JSON:
        {{
            "trend": "bullish/bearish/neutral",
            "entry_timing": "good/wait/avoid", 
            "risk_level": "low/medium/high",
            "reasoning": "explanation here",
            "confidence": 0.85
        }}
        """
        
        try:
            response = self.client.chat(
                model='llama3.1:8b',
                messages=[{'role': 'user', 'content': prompt}]
            )
            
            # Extract JSON from response
            content = response['message']['content']
            json_start = content.find('{')
            json_end = content.rfind('}') + 1
            json_str = content[json_start:json_end]
            
            return json.loads(json_str)
        except Exception as e:
            print(f"Error in AI analysis: {e}")
            return None

# dca_strategy.py
import schedule
import time
from datetime import datetime
from market_analyzer import CryptoMarketAnalyzer

class DCAStrategy:
    def __init__(self, symbol, investment_amount, frequency='daily'):
        self.symbol = symbol
        self.investment_amount = investment_amount
        self.frequency = frequency
        self.analyzer = CryptoMarketAnalyzer()
        self.purchase_history = []
        
    def execute_dca_purchase(self):
        """Execute a DCA purchase with AI sentiment check"""
        # Get current market analysis
        df = self.analyzer.get_market_data(self.symbol)
        if df is None:
            return False
            
        df_with_indicators = self.analyzer.calculate_technical_indicators(df)
        sentiment = self.analyzer.analyze_market_sentiment(df_with_indicators, self.symbol)
        
        current_price = df['close'].iloc[-1]
        timestamp = datetime.now()
        
        # DCA buys regardless of sentiment, but logs AI opinion
        purchase_amount = self.investment_amount / current_price
        
        purchase_record = {
            'timestamp': timestamp,
            'price': current_price,
            'amount_usd': self.investment_amount,
            'amount_crypto': purchase_amount,
            'ai_sentiment': sentiment,
            'strategy': 'DCA'
        }
        
        self.purchase_history.append(purchase_record)
        
        print(f"DCA Purchase: ${self.investment_amount} of {self.symbol} at ${current_price:.2f}")
        print(f"AI Sentiment: {sentiment['trend']} | Risk: {sentiment['risk_level']}")
        
        return True
    
    def calculate_dca_performance(self, current_price):
        """Calculate total DCA performance"""
        if not self.purchase_history:
            return None
            
        total_invested = sum(p['amount_usd'] for p in self.purchase_history)
        total_crypto = sum(p['amount_crypto'] for p in self.purchase_history)
        current_value = total_crypto * current_price
        
        roi = ((current_value - total_invested) / total_invested) * 100
        avg_price = total_invested / total_crypto
        
        return {
            'total_invested': total_invested,
            'total_crypto': total_crypto,
            'current_value': current_value,
            'roi_percentage': roi,
            'average_price': avg_price,
            'purchase_count': len(self.purchase_history)
        }
    
    def start_automated_dca(self):
        """Start automated DCA schedule"""
        if self.frequency == 'daily':
            schedule.every().day.at("09:00").do(self.execute_dca_purchase)
        elif self.frequency == 'weekly':
            schedule.every().monday.at("09:00").do(self.execute_dca_purchase)
        elif self.frequency == 'monthly':
            schedule.every().month.do(self.execute_dca_purchase)
        
        print(f"Started automated DCA for {self.symbol} - {self.frequency} ${self.investment_amount}")
        
        while True:
            schedule.run_pending()
            time.sleep(3600)  # Check every hour

# Usage example
if __name__ == "__main__":
    # DCA $100 daily into Bitcoin
    dca = DCAStrategy('BTC/USDT', 100, 'daily')
    dca.execute_dca_purchase()  # Manual purchase for testing

# lump_sum_strategy.py
from market_analyzer import CryptoMarketAnalyzer
from datetime import datetime, timedelta
import json

class LumpSumStrategy:
    def __init__(self, symbol, investment_amount, confidence_threshold=0.75):
        self.symbol = symbol
        self.investment_amount = investment_amount
        self.confidence_threshold = confidence_threshold
        self.analyzer = CryptoMarketAnalyzer()
        self.pending_investment = investment_amount
        self.purchase_history = []
        self.analysis_history = []
        
    def should_invest_now(self):
        """AI-powered decision for lump sum timing"""
        df = self.analyzer.get_market_data(self.symbol, timeframe='4h', limit=200)
        if df is None:
            return False, "Data unavailable"
            
        df_with_indicators = self.analyzer.calculate_technical_indicators(df)
        sentiment = self.analyzer.analyze_market_sentiment(df_with_indicators, self.symbol)
        
        # Store analysis for tracking
        self.analysis_history.append({
            'timestamp': datetime.now(),
            'sentiment': sentiment,
            'price': df['close'].iloc[-1]
        })
        
        # Enhanced decision logic
        conditions_met = 0
        total_conditions = 4
        reasons = []
        
        # Condition 1: AI sentiment is positive
        if sentiment['entry_timing'] == 'good':
            conditions_met += 1
            reasons.append("AI recommends good entry timing")
        
        # Condition 2: High AI confidence
        if sentiment['confidence'] >= self.confidence_threshold:
            conditions_met += 1
            reasons.append(f"High AI confidence ({sentiment['confidence']:.2f})")
        
        # Condition 3: Low to medium risk
        if sentiment['risk_level'] in ['low', 'medium']:
            conditions_met += 1
            reasons.append(f"Acceptable risk level ({sentiment['risk_level']})")
        
        # Condition 4: Bullish or neutral trend
        if sentiment['trend'] in ['bullish', 'neutral']:
            conditions_met += 1
            reasons.append(f"Favorable trend ({sentiment['trend']})")
        
        # Require at least 3 out of 4 conditions
        should_invest = conditions_met >= 3
        
        decision_data = {
            'should_invest': should_invest,
            'conditions_met': f"{conditions_met}/{total_conditions}",
            'reasons': reasons,
            'sentiment': sentiment,
            'current_price': df['close'].iloc[-1]
        }
        
        return should_invest, decision_data
    
    def execute_lump_sum_purchase(self, force=False):
        """Execute lump sum purchase if conditions are met"""
        if self.pending_investment <= 0:
            return False, "No pending investment amount"
        
        should_invest, decision_data = self.should_invest_now()
        
        if not should_invest and not force:
            print(f"Waiting for better entry. Conditions: {decision_data['conditions_met']}")
            return False, decision_data
        
        # Execute purchase
        current_price = decision_data['current_price']
        purchase_amount = self.pending_investment / current_price
        
        purchase_record = {
            'timestamp': datetime.now(),
            'price': current_price,
            'amount_usd': self.pending_investment,
            'amount_crypto': purchase_amount,
            'decision_data': decision_data,
            'strategy': 'Lump Sum'
        }
        
        self.purchase_history.append(purchase_record)
        self.pending_investment = 0  # All invested
        
        print(f"Lump Sum Purchase: ${self.investment_amount} of {self.symbol} at ${current_price:.2f}")
        print(f"Decision reasons: {', '.join(decision_data['reasons'])}")
        
        return True, purchase_record
    
    def monitor_for_entry(self, check_interval_hours=4):
        """Continuously monitor for optimal entry point"""
        print(f"Monitoring {self.symbol} for lump sum entry...")
        print(f"Investment amount: ${self.investment_amount}")
        print(f"Confidence threshold: {self.confidence_threshold}")
        
        while self.pending_investment > 0:
            invested, result = self.execute_lump_sum_purchase()
            
            if invested:
                print("Lump sum investment executed!")
                break
            else:
                print(f"Waiting... Next check in {check_interval_hours} hours")
                time.sleep(check_interval_hours * 3600)
    
    def calculate_performance(self, current_price):
        """Calculate lump sum performance"""
        if not self.purchase_history:
            return None
            
        purchase = self.purchase_history[-1]  # Most recent (should be only one)
        invested = purchase['amount_usd']
        crypto_amount = purchase['amount_crypto']
        current_value = crypto_amount * current_price
        
        roi = ((current_value - invested) / invested) * 100
        
        return {
            'invested': invested,
            'current_value': current_value,
            'roi_percentage': roi,
            'entry_price': purchase['price'],
            'current_price': current_price,
            'decision_quality': purchase['decision_data']['conditions_met']
        }

# Usage example
if __name__ == "__main__":
    # Lump sum $5000 into Ethereum with 80% confidence threshold
    lump_sum = LumpSumStrategy('ETH/USDT', 5000, confidence_threshold=0.80)
    
    # Check if now is a good time
    should_invest, decision = lump_sum.should_invest_now()
    print(f"Should invest now: {should_invest}")
    print(f"Decision data: {json.dumps(decision, indent=2, default=str)}")

# strategy_comparison.py
import pandas as pd
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
from market_analyzer import CryptoMarketAnalyzer

class StrategyComparison:
    def __init__(self, symbol, total_investment, dca_frequency='weekly'):
        self.symbol = symbol
        self.total_investment = total_investment
        self.dca_frequency = dca_frequency
        self.analyzer = CryptoMarketAnalyzer()
        
    def backtest_dca(self, start_date, end_date):
        """Backtest DCA strategy over historical period"""
        # Get historical data
        df = self.analyzer.get_market_data(self.symbol, timeframe='1d', limit=365)
        
        # Filter date range
        df = df[(df['timestamp'] >= start_date) & (df['timestamp'] <= end_date)]
        
        # Calculate DCA investment schedule
        if self.dca_frequency == 'daily':
            investment_dates = pd.date_range(start_date, end_date, freq='D')
            per_investment = self.total_investment / len(investment_dates)
        elif self.dca_frequency == 'weekly':
            investment_dates = pd.date_range(start_date, end_date, freq='W')
            per_investment = self.total_investment / len(investment_dates)
        
        dca_purchases = []
        total_crypto = 0
        total_invested = 0
        
        for date in investment_dates:
            # Find closest price data
            closest_data = df[df['timestamp'].dt.date <= date.date()]
            if closest_data.empty:
                continue
                
            price = closest_data['close'].iloc[-1]
            crypto_bought = per_investment / price
            
            dca_purchases.append({
                'date': date,
                'price': price,
                'amount_usd': per_investment,
                'amount_crypto': crypto_bought
            })
            
            total_crypto += crypto_bought
            total_invested += per_investment
        
        # Calculate final value
        final_price = df['close'].iloc[-1]
        final_value = total_crypto * final_price
        dca_roi = ((final_value - total_invested) / total_invested) * 100
        
        return {
            'strategy': 'DCA',
            'purchases': dca_purchases,
            'total_invested': total_invested,
            'total_crypto': total_crypto,
            'final_value': final_value,
            'roi_percentage': dca_roi,
            'average_price': total_invested / total_crypto if total_crypto > 0 else 0
        }
    
    def backtest_lump_sum(self, start_date, end_date, entry_delay_days=30):
        """Backtest lump sum with AI timing simulation"""
        df = self.analyzer.get_market_data(self.symbol, timeframe='1d', limit=365)
        df = df[(df['timestamp'] >= start_date) & (df['timestamp'] <= end_date)]
        df_with_indicators = self.analyzer.calculate_technical_indicators(df)
        
        # Simulate waiting for good entry point
        entry_date = None
        entry_price = None
        
        # Look for first "good" entry point after delay
        search_start = start_date + timedelta(days=entry_delay_days)
        search_df = df_with_indicators[df_with_indicators['timestamp'] >= search_start]
        
        for idx, row in search_df.iterrows():
            # Simple good entry conditions (RSI < 40 and price below 20-day SMA)
            if (row['rsi'] < 40 and 
                row['close'] < row['sma_20'] and 
                not pd.isna(row['rsi']) and 
                not pd.isna(row['sma_20'])):
                entry_date = row['timestamp']
                entry_price = row['close']
                break
        
        # If no "good" entry found, invest at midpoint
        if entry_date is None:
            midpoint_idx = len(search_df) // 2
            entry_date = search_df.iloc[midpoint_idx]['timestamp']
            entry_price = search_df.iloc[midpoint_idx]['close']
        
        # Calculate lump sum performance
        crypto_bought = self.total_investment / entry_price
        final_price = df['close'].iloc[-1]
        final_value = crypto_bought * final_price
        lump_sum_roi = ((final_value - self.total_investment) / self.total_investment) * 100
        
        return {
            'strategy': 'Lump Sum',
            'entry_date': entry_date,
            'entry_price': entry_price,
            'total_invested': self.total_investment,
            'total_crypto': crypto_bought,
            'final_value': final_value,
            'roi_percentage': lump_sum_roi
        }
    
    def run_comparison(self, months_back=6):
        """Run complete comparison between strategies"""
        end_date = datetime.now()
        start_date = end_date - timedelta(days=months_back * 30)
        
        print(f"Comparing DCA vs Lump Sum for {self.symbol}")
        print(f"Period: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}")
        print(f"Total Investment: ${self.total_investment:,.2f}\n")
        
        # Run backtests
        dca_results = self.backtest_dca(start_date, end_date)
        lump_sum_results = self.backtest_lump_sum(start_date, end_date)
        
        # Display results
        print("=== DCA Strategy Results ===")
        print(f"Purchase Count: {len(dca_results['purchases'])}")
        print(f"Average Price: ${dca_results['average_price']:.2f}")
        print(f"Total Crypto: {dca_results['total_crypto']:.6f}")
        print(f"Final Value: ${dca_results['final_value']:,.2f}")
        print(f"ROI: {dca_results['roi_percentage']:.2f}%\n")
        
        print("=== Lump Sum Strategy Results ===")
        print(f"Entry Date: {lump_sum_results['entry_date'].strftime('%Y-%m-%d')}")
        print(f"Entry Price: ${lump_sum_results['entry_price']:.2f}")
        print(f"Total Crypto: {lump_sum_results['total_crypto']:.6f}")
        print(f"Final Value: ${lump_sum_results['final_value']:,.2f}")
        print(f"ROI: {lump_sum_results['roi_percentage']:.2f}%\n")
        
        # Determine winner
        if dca_results['roi_percentage'] > lump_sum_results['roi_percentage']:
            winner = "DCA"
            advantage = dca_results['roi_percentage'] - lump_sum_results['roi_percentage']
        else:
            winner = "Lump Sum"
            advantage = lump_sum_results['roi_percentage'] - dca_results['roi_percentage']
        
        print(f"=== Winner: {winner} Strategy ===")
        print(f"Performance Advantage: {advantage:.2f} percentage points")
        
        return dca_results, lump_sum_results

# Usage example
if __name__ == "__main__":
    comparison = StrategyComparison('BTC/USDT', 10000, 'weekly')
    dca_results, lump_sum_results = comparison.run_comparison(months_back=6)

# risk_manager.py
class RiskManager:
    def __init__(self, max_daily_investment=1000, max_portfolio_allocation=0.1):
        self.max_daily_investment = max_daily_investment
        self.max_portfolio_allocation = max_portfolio_allocation
        self.daily_investment_tracker = {}
        
    def check_investment_limits(self, amount, symbol):
        """Verify investment doesn't exceed risk limits"""
        today = datetime.now().date()
        
        # Check daily limit
        if today not in self.daily_investment_tracker:
            self.daily_investment_tracker[today] = 0
            
        if self.daily_investment_tracker[today] + amount > self.max_daily_investment:
            return False, "Daily investment limit exceeded"
        
        # Update tracker
        self.daily_investment_tracker[today] += amount
        return True, "Investment approved"
    
    def validate_ai_decision(self, sentiment_data):
        """Additional validation of AI recommendations"""
        confidence = sentiment_data.get('confidence', 0)
        risk_level = sentiment_data.get('risk_level', 'high')
        
        # Require high confidence for large investments
        if confidence < 0.7:
            return False, "AI confidence too low"
            
        # Avoid high-risk periods
        if risk_level == 'high':
            return False, "Market risk too high"
            
        return True, "AI decision validated"

# main.py
import logging
from config import EXCHANGE_CONFIG, INVESTMENT_CONFIG
from dca_strategy import DCAStrategy
from lump_sum_strategy import LumpSumStrategy
from risk_manager import RiskManager

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('crypto_timing.log'),
        logging.StreamHandler()
    ]
)

def main():
    """Main execution with error handling"""
    risk_manager = RiskManager(
        max_daily_investment=INVESTMENT_CONFIG['max_daily'],
        max_portfolio_allocation=INVESTMENT_CONFIG['max_allocation']
    )
    
    # Initialize strategies
    dca = DCAStrategy(
        symbol=INVESTMENT_CONFIG['symbol'],
        investment_amount=INVESTMENT_CONFIG['dca_amount'],
        frequency=INVESTMENT_CONFIG['dca_frequency']
    )
    
    lump_sum = LumpSumStrategy(
        symbol=INVESTMENT_CONFIG['symbol'],
        investment_amount=INVESTMENT_CONFIG['lump_sum_amount'],
        confidence_threshold=INVESTMENT_CONFIG['confidence_threshold']
    )
    
    try:
        # Run strategy based on user preference
        if INVESTMENT_CONFIG['strategy'] == 'dca':
            logging.info("Starting DCA strategy...")
            dca.start_automated_dca()
        elif INVESTMENT_CONFIG['strategy'] == 'lump_sum':
            logging.info("Starting lump sum monitoring...")
            lump_sum.monitor_for_entry()
        else:
            logging.info("Running strategy comparison...")
            comparison = StrategyComparison(
                INVESTMENT_CONFIG['symbol'],
                INVESTMENT_CONFIG['comparison_amount']
            )
            comparison.run_comparison()
            
    except Exception as e:
        logging.error(f"Strategy execution failed: {e}")
        
if __name__ == "__main__":
    main()