Build a Crypto Political Sentiment Trading Bot with Ollama: Cabinet Impact Analysis Guide

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

# Pull the Llama 2 model for sentiment analysis
ollama pull llama2:7b

# Verify installation
ollama list

ollama run llama2:7b "Analyze the crypto market sentiment of this news: 'New Treasury Secretary has extensive blockchain experience and supports digital asset innovation.'"

# requirements.txt
import requests
import json
import time
from datetime import datetime
import pandas as pd
import ccxt  # Cryptocurrency exchange library
import feedparser  # RSS feed parsing
import sqlite3  # Local database

pip install requests pandas ccxt feedparser sqlite3

# sentiment_analyzer.py
class OllamaSentimentAnalyzer:
    def __init__(self, model_name="llama2:7b"):
        self.model_name = model_name
        self.ollama_url = "http://localhost:11434/api/generate"
    
    def analyze_political_sentiment(self, news_text, cabinet_position=""):
        """
        Analyze political news sentiment with crypto market focus
        Returns: sentiment score (-1 to 1) and confidence level
        """
        prompt = f"""
        Analyze this political news for cryptocurrency market impact:
        
        News: {news_text}
        Cabinet Position: {cabinet_position}
        
        Consider:
        - Regulatory implications for crypto
        - Historical market reactions to similar appointments
        - Policy positions on blockchain technology
        
        Respond with JSON format:
        {{
            "sentiment_score": [number from -1 to 1],
            "confidence": [number from 0 to 1],
            "key_factors": ["factor1", "factor2"],
            "market_impact": "positive/negative/neutral",
            "timeframe": "immediate/short-term/long-term"
        }}
        """
        
        payload = {
            "model": self.model_name,
            "prompt": prompt,
            "stream": False
        }
        
        try:
            response = requests.post(self.ollama_url, json=payload)
            result = response.json()
            
            # Parse the JSON response from Ollama
            analysis = json.loads(result['response'])
            return analysis
            
        except Exception as e:
            print(f"Sentiment analysis error: {e}")
            return {"sentiment_score": 0, "confidence": 0}

# news_collector.py
class PoliticalNewsCollector:
    def __init__(self):
        self.rss_feeds = [
            "https://feeds.reuters.com/reuters/politicsNews",
            "https://rss.cnn.com/rss/cnn_allpolitics.rss",
            "https://www.politico.com/rss/politico.xml"
        ]
        
        # Cabinet positions that impact crypto markets
        self.target_positions = [
            "Treasury Secretary", "SEC Chair", "CFTC Commissioner",
            "Commerce Secretary", "Attorney General", "Fed Chair"
        ]
    
    def collect_cabinet_news(self):
        """
        Collect recent political news focused on cabinet appointments
        Returns: List of relevant news articles
        """
        relevant_articles = []
        
        for feed_url in self.rss_feeds:
            try:
                feed = feedparser.parse(feed_url)
                
                for entry in feed.entries:
                    # Check if article mentions cabinet positions
                    title_text = entry.title.lower()
                    summary_text = entry.summary.lower() if hasattr(entry, 'summary') else ""
                    
                    for position in self.target_positions:
                        if position.lower() in title_text or position.lower() in summary_text:
                            article_data = {
                                "title": entry.title,
                                "summary": entry.summary if hasattr(entry, 'summary') else "",
                                "link": entry.link,
                                "published": entry.published,
                                "cabinet_position": position
                            }
                            relevant_articles.append(article_data)
                            break
                            
            except Exception as e:
                print(f"Feed collection error for {feed_url}: {e}")
        
        return relevant_articles

# trading_engine.py
class CryptoTradingEngine:
    def __init__(self, exchange_api_key, exchange_secret):
        # Initialize exchange connection (example with Binance)
        self.exchange = ccxt.binance({
            'apiKey': exchange_api_key,
            'secret': exchange_secret,
            'sandbox': True,  # Use sandbox for testing
        })
        
        # Trading thresholds based on sentiment scores
        self.sentiment_thresholds = {
            'strong_buy': 0.7,
            'buy': 0.3,
            'hold': 0.0,
            'sell': -0.3,
            'strong_sell': -0.7
        }
        
        # Position sizes based on confidence levels
        self.position_sizes = {
            'high_confidence': 0.05,  # 5% of portfolio
            'medium_confidence': 0.03,  # 3% of portfolio
            'low_confidence': 0.01   # 1% of portfolio
        }
    
    def calculate_trade_signal(self, sentiment_data):
        """
        Convert sentiment analysis into trading signals
        Returns: Trading action and position size
        """
        sentiment_score = sentiment_data['sentiment_score']
        confidence = sentiment_data['confidence']
        timeframe = sentiment_data.get('timeframe', 'short-term')
        
        # Determine trading action
        if sentiment_score >= self.sentiment_thresholds['strong_buy']:
            action = 'strong_buy'
        elif sentiment_score >= self.sentiment_thresholds['buy']:
            action = 'buy'
        elif sentiment_score <= self.sentiment_thresholds['strong_sell']:
            action = 'strong_sell'
        elif sentiment_score <= self.sentiment_thresholds['sell']:
            action = 'sell'
        else:
            action = 'hold'
        
        # Calculate position size based on confidence
        if confidence >= 0.8:
            confidence_level = 'high_confidence'
        elif confidence >= 0.5:
            confidence_level = 'medium_confidence'
        else:
            confidence_level = 'low_confidence'
        
        position_size = self.position_sizes[confidence_level]
        
        # Adjust for timeframe (longer timeframes = larger positions)
        if timeframe == 'long-term':
            position_size *= 1.5
        elif timeframe == 'immediate':
            position_size *= 0.7
        
        return {
            'action': action,
            'position_size': position_size,
            'confidence_level': confidence_level,
            'timeframe': timeframe
        }
    
    def execute_trade(self, symbol, trade_signal, current_price):
        """
        Execute trading decision with proper risk management
        """
        action = trade_signal['action']
        position_size = trade_signal['position_size']
        
        if action in ['buy', 'strong_buy']:
            # Calculate buy amount
            account_balance = self.get_account_balance('USDT')
            buy_amount = account_balance * position_size
            
            # Place buy order
            order = self.exchange.create_market_buy_order(
                symbol, 
                buy_amount / current_price
            )
            
            return order
            
        elif action in ['sell', 'strong_sell']:
            # Calculate sell amount
            crypto_balance = self.get_account_balance(symbol.split('/')[0])
            sell_amount = crypto_balance * position_size
            
            # Place sell order
            order = self.exchange.create_market_sell_order(
                symbol, 
                sell_amount
            )
            
            return order
        
        return None  # Hold position

# database_manager.py
class TradingDatabase:
    def __init__(self, db_path="crypto_sentiment_trading.db"):
        self.db_path = db_path
        self.init_database()
    
    def init_database(self):
        """Initialize database tables for tracking trades and sentiment"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # News sentiment tracking table
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS sentiment_analysis (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp DATETIME,
                news_title TEXT,
                cabinet_position TEXT,
                sentiment_score REAL,
                confidence REAL,
                market_impact TEXT,
                timeframe TEXT
            )
        ''')
        
        # Trading history table
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS trades (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp DATETIME,
                symbol TEXT,
                action TEXT,
                position_size REAL,
                entry_price REAL,
                exit_price REAL,
                profit_loss REAL,
                sentiment_id INTEGER,
                FOREIGN KEY (sentiment_id) REFERENCES sentiment_analysis (id)
            )
        ''')
        
        conn.commit()
        conn.close()
    
    def log_sentiment_analysis(self, news_data, sentiment_result):
        """Store sentiment analysis results"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO sentiment_analysis 
            (timestamp, news_title, cabinet_position, sentiment_score, 
             confidence, market_impact, timeframe)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        ''', (
            datetime.now(),
            news_data['title'],
            news_data['cabinet_position'],
            sentiment_result['sentiment_score'],
            sentiment_result['confidence'],
            sentiment_result['market_impact'],
            sentiment_result['timeframe']
        ))
        
        sentiment_id = cursor.lastrowid
        conn.commit()
        conn.close()
        
        return sentiment_id

# main_trading_bot.py
import time
import logging
from datetime import datetime, timedelta

class CryptoSentimentTradingBot:
    def __init__(self, config):
        self.sentiment_analyzer = OllamaSentimentAnalyzer()
        self.news_collector = PoliticalNewsCollector()
        self.trading_engine = CryptoTradingEngine(
            config['exchange_api_key'],
            config['exchange_secret']
        )
        self.database = TradingDatabase()
        
        # Trading configuration
        self.trading_pairs = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT']
        self.analysis_interval = 300  # 5 minutes
        self.max_daily_trades = 10
        
        # Performance tracking
        self.daily_trade_count = 0
        self.last_reset_date = datetime.now().date()
        
        # Setup logging
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('trading_bot.log'),
                logging.StreamHandler()
            ]
        )
    
    def run_trading_cycle(self):
        """Execute one complete trading analysis cycle"""
        try:
            # Reset daily trade counter if new day
            if datetime.now().date() > self.last_reset_date:
                self.daily_trade_count = 0
                self.last_reset_date = datetime.now().date()
            
            # Check if daily trade limit reached
            if self.daily_trade_count >= self.max_daily_trades:
                logging.info("Daily trade limit reached. Waiting for next day.")
                return
            
            # Collect recent political news
            logging.info("Collecting political news...")
            news_articles = self.news_collector.collect_cabinet_news()
            
            for article in news_articles:
                # Analyze sentiment for each relevant article
                sentiment_result = self.sentiment_analyzer.analyze_political_sentiment(
                    article['title'] + " " + article['summary'],
                    article['cabinet_position']
                )
                
                # Log sentiment analysis
                sentiment_id = self.database.log_sentiment_analysis(article, sentiment_result)
                
                # Generate trading signals
                trade_signal = self.trading_engine.calculate_trade_signal(sentiment_result)
                
                # Execute trades for each trading pair
                for symbol in self.trading_pairs:
                    if trade_signal['action'] != 'hold':
                        current_price = self.trading_engine.exchange.fetch_ticker(symbol)['last']
                        
                        order = self.trading_engine.execute_trade(
                            symbol, 
                            trade_signal, 
                            current_price
                        )
                        
                        if order:
                            self.daily_trade_count += 1
                            logging.info(f"Executed {trade_signal['action']} for {symbol} based on {article['cabinet_position']} news")
                            
                            # Log trade in database
                            self.database.log_trade(order, sentiment_id)
                
        except Exception as e:
            logging.error(f"Trading cycle error: {e}")
    
    def start_bot(self):
        """Start the automated trading bot"""
        logging.info("Starting Crypto Political Sentiment Trading Bot...")
        
        while True:
            try:
                self.run_trading_cycle()
                time.sleep(self.analysis_interval)
                
            except KeyboardInterrupt:
                logging.info("Bot stopped by user")
                break
            except Exception as e:
                logging.error(f"Unexpected error: {e}")
                time.sleep(60)  # Wait 1 minute before retrying

# Configuration and startup
if __name__ == "__main__":
    config = {
        'exchange_api_key': 'your_api_key_here',
        'exchange_secret': 'your_secret_here'
    }
    
    bot = CryptoSentimentTradingBot(config)
    bot.start_bot()

# risk_management.py
class RiskManager:
    def __init__(self, max_portfolio_risk=0.15, max_single_position=0.05):
        self.max_portfolio_risk = max_portfolio_risk  # 15% max portfolio risk
        self.max_single_position = max_single_position  # 5% max single position
        self.stop_loss_percentage = 0.03  # 3% stop loss
        self.take_profit_percentage = 0.06  # 6% take profit
    
    def validate_trade(self, trade_signal, current_portfolio_value, position_value):
        """
        Validate trade against risk management rules
        Returns: approved_trade_signal or None
        """
        # Check if position size exceeds limits
        position_risk = position_value / current_portfolio_value
        
        if position_risk > self.max_single_position:
            # Reduce position size to maximum allowed
            trade_signal['position_size'] = self.max_single_position
            
        # Check portfolio-wide risk exposure
        total_crypto_exposure = self.calculate_crypto_exposure(current_portfolio_value)
        
        if total_crypto_exposure > self.max_portfolio_risk:
            # Reject trade if it would exceed portfolio risk limits
            return None
            
        return trade_signal
    
    def set_stop_loss_take_profit(self, exchange, symbol, entry_price, action):
        """
        Set automatic stop loss and take profit orders
        """
        if action in ['buy', 'strong_buy']:
            stop_loss_price = entry_price * (1 - self.stop_loss_percentage)
            take_profit_price = entry_price * (1 + self.take_profit_percentage)
            
            # Place stop loss order
            exchange.create_stop_loss_order(symbol, 'sell', None, stop_loss_price)
            
            # Place take profit order
            exchange.create_limit_sell_order(symbol, None, take_profit_price)

# performance_analytics.py
class PerformanceAnalyzer:
    def __init__(self, database):
        self.db = database
    
    def calculate_strategy_performance(self, days=30):
        """
        Analyze trading strategy performance over specified period
        Returns: Performance metrics and insights
        """
        conn = sqlite3.connect(self.db.db_path)
        
        # Get trades from last N days
        query = '''
            SELECT t.*, s.sentiment_score, s.confidence, s.cabinet_position
            FROM trades t
            JOIN sentiment_analysis s ON t.sentiment_id = s.id
            WHERE t.timestamp >= datetime('now', '-{} days')
        '''.format(days)
        
        df = pd.read_sql_query(query, conn)
        conn.close()
        
        if df.empty:
            return {"error": "No trades found in specified period"}
        
        # Calculate performance metrics
        total_trades = len(df)
        profitable_trades = len(df[df['profit_loss'] > 0])
        win_rate = profitable_trades / total_trades
        
        total_profit_loss = df['profit_loss'].sum()
        average_trade_return = df['profit_loss'].mean()
        
        # Analyze by cabinet position impact
        cabinet_performance = df.groupby('cabinet_position').agg({
            'profit_loss': ['sum', 'mean', 'count'],
            'sentiment_score': 'mean'
        }).round(4)
        
        # Analyze by sentiment confidence levels
        df['confidence_range'] = pd.cut(df['confidence'], 
                                       bins=[0, 0.5, 0.8, 1.0], 
                                       labels=['Low', 'Medium', 'High'])
        
        confidence_performance = df.groupby('confidence_range').agg({
            'profit_loss': ['sum', 'mean', 'count']
        }).round(4)
        
        return {
            'total_trades': total_trades,
            'win_rate': win_rate,
            'total_profit_loss': total_profit_loss,
            'average_trade_return': average_trade_return,
            'cabinet_performance': cabinet_performance.to_dict(),
            'confidence_performance': confidence_performance.to_dict()
        }
    
    def generate_optimization_recommendations(self):
        """
        Generate strategy optimization recommendations based on historical performance
        """
        performance = self.calculate_strategy_performance()
        recommendations = []
        
        # Analyze cabinet position effectiveness
        cabinet_perf = performance['cabinet_performance']
        for position, metrics in cabinet_perf.items():
            avg_return = metrics['profit_loss']['mean']
            trade_count = metrics['profit_loss']['count']
            
            if avg_return < 0 and trade_count > 5:
                recommendations.append(f"Consider reducing position size for {position} appointments (negative average return)")
            elif avg_return > 0.02 and trade_count > 3:
                recommendations.append(f"Consider increasing position size for {position} appointments (strong positive return)")
        
        return recommendations

# deployment_config.py
import os
from dataclasses import dataclass

@dataclass
class ProductionConfig:
    # Exchange API credentials
    exchange_api_key: str = os.getenv('EXCHANGE_API_KEY')
    exchange_secret: str = os.getenv('EXCHANGE_SECRET')
    
    # Risk management settings
    max_portfolio_risk: float = 0.10  # Reduced for production
    max_daily_trades: int = 15
    
    # Ollama configuration
    ollama_model: str = "llama2:7b"
    ollama_timeout: int = 30
    
    # Monitoring and alerts
    discord_webhook: str = os.getenv('DISCORD_WEBHOOK')
    email_alerts: bool = True
    
    # Database settings
    database_backup_interval: int = 86400  # 24 hours
    
    def validate_config(self):
        """Validate all required configuration is present"""
        required_fields = [
            'exchange_api_key', 'exchange_secret'
        ]
        
        for field in required_fields:
            if not getattr(self, field):
                raise ValueError(f"Missing required configuration: {field}")

# Alert system for monitoring
class AlertSystem:
    def __init__(self, config):
        self.config = config
    
    def send_trade_alert(self, trade_info):
        """Send alert when significant trade is executed"""
        message = f"""
        🚨 Crypto Sentiment Bot Trade Alert
        Symbol: {trade_info['symbol']}
        Action: {trade_info['action']}
        Amount: ${trade_info['amount']:.2f}
        Reason: {trade_info['cabinet_position']} sentiment
        Confidence: {trade_info['confidence']:.2f}
        """
        
        # Send to Discord webhook
        if self.config.discord_webhook:
            self.send_discord_alert(message)
    
    def send_performance_summary(self, performance_data):
        """Send daily performance summary"""
        message = f"""
        📊 Daily Performance Summary
        Total Trades: {performance_data['total_trades']}
        Win Rate: {performance_data['win_rate']:.2%}
        P&L: ${performance_data['total_profit_loss']:.2f}
        Best Cabinet Signal: {performance_data['best_cabinet']}
        """
        
        if self.config.discord_webhook:
            self.send_discord_alert(message)

# advanced_sentiment.py
class EnsembleSentimentAnalyzer:
    def __init__(self):
        self.models = [
            "llama2:7b",
            "mistral:7b", 
            "codellama:7b"
        ]
        self.weights = [0.4, 0.35, 0.25]  # Model confidence weights
    
    def ensemble_sentiment_analysis(self, news_text, cabinet_position):
        """
        Use multiple models for more accurate sentiment analysis
        Returns: Weighted average sentiment with higher confidence
        """
        model_results = []
        
        for model in self.models:
            analyzer = OllamaSentimentAnalyzer(model)
            result = analyzer.analyze_political_sentiment(news_text, cabinet_position)
            model_results.append(result)
        
        # Calculate weighted average
        weighted_sentiment = sum(
            result['sentiment_score'] * weight 
            for result, weight in zip(model_results, self.weights)
        )
        
        # Calculate ensemble confidence
        confidence_scores = [result['confidence'] for result in model_results]
        ensemble_confidence = sum(
            conf * weight 
            for conf, weight in zip(confidence_scores, self.weights)
        )
        
        # Boost confidence if models agree
        sentiment_agreement = self.calculate_agreement(model_results)
        final_confidence = min(ensemble_confidence * sentiment_agreement, 1.0)
        
        return {
            'sentiment_score': weighted_sentiment,
            'confidence': final_confidence,
            'model_agreement': sentiment_agreement,
            'individual_results': model_results
        }