Natural Language Processing DeFi: AI News Sentiment Trading in 2025

# Install required dependencies
pip install transformers torch pandas requests web3 ccxt

# Import essential libraries
import pandas as pd
import numpy as np
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import requests
from datetime import datetime, timedelta
import json

class CryptoNewsCollector:
    def __init__(self):
        self.news_sources = [
            'https://api.coindesk.com/v2/articles',
            'https://api.cointelegraph.com/v1/news',
            'https://newsapi.org/v2/everything'
        ]
        self.api_keys = {
            'newsapi': 'YOUR_NEWSAPI_KEY'
        }
    
    def fetch_crypto_news(self, hours_back=24):
        """Collect cryptocurrency news from multiple sources"""
        all_articles = []
        end_time = datetime.now()
        start_time = end_time - timedelta(hours=hours_back)
        
        # Fetch from NewsAPI
        params = {
            'q': 'cryptocurrency OR bitcoin OR ethereum OR DeFi',
            'from': start_time.isoformat(),
            'to': end_time.isoformat(),
            'sortBy': 'publishedAt',
            'apiKey': self.api_keys['newsapi']
        }
        
        response = requests.get(
            'https://newsapi.org/v2/everything', 
            params=params
        )
        
        if response.status_code == 200:
            articles = response.json()['articles']
            for article in articles:
                all_articles.append({
                    'title': article['title'],
                    'content': article['description'],
                    'published_at': article['publishedAt'],
                    'source': article['source']['name'],
                    'url': article['url']
                })
        
        return pd.DataFrame(all_articles)

# Initialize news collector
news_collector = CryptoNewsCollector()
recent_news = news_collector.fetch_crypto_news(hours_back=6)
print(f"Collected {len(recent_news)} news articles")

class CryptoSentimentAnalyzer:
    def __init__(self):
        # Load pre-trained model optimized for financial text
        self.tokenizer = AutoTokenizer.from_pretrained(
            'ProsusAI/finbert'
        )
        self.model = AutoModelForSequenceClassification.from_pretrained(
            'ProsusAI/finbert'
        )
        self.sentiment_pipeline = pipeline(
            'sentiment-analysis',
            model=self.model,
            tokenizer=self.tokenizer
        )
    
    def analyze_sentiment(self, text):
        """Analyze sentiment of cryptocurrency news text"""
        # Clean and prepare text
        cleaned_text = self.preprocess_text(text)
        
        # Get sentiment prediction
        result = self.sentiment_pipeline(cleaned_text)[0]
        
        # Convert to numerical score (-1 to 1)
        sentiment_score = self.convert_to_score(
            result['label'], 
            result['score']
        )
        
        return {
            'sentiment': result['label'],
            'confidence': result['score'],
            'numerical_score': sentiment_score
        }
    
    def preprocess_text(self, text):
        """Clean and prepare text for analysis"""
        if not text:
            return ""
        
        # Remove extra whitespace and normalize
        text = ' '.join(text.split())
        
        # Truncate to model's max length
        if len(text) > 512:
            text = text[:512]
        
        return text
    
    def convert_to_score(self, label, confidence):
        """Convert sentiment label to numerical score"""
        if label == 'positive':
            return confidence
        elif label == 'negative':
            return -confidence
        else:  # neutral
            return 0
    
    def batch_analyze(self, texts):
        """Analyze sentiment for multiple texts efficiently"""
        results = []
        for text in texts:
            sentiment = self.analyze_sentiment(text)
            results.append(sentiment)
        return results

# Initialize sentiment analyzer
sentiment_analyzer = CryptoSentimentAnalyzer()

# Analyze news sentiment
if not recent_news.empty:
    # Combine title and content for better analysis
    recent_news['full_text'] = (
        recent_news['title'] + ' ' + 
        recent_news['content'].fillna('')
    )
    
    # Analyze sentiment for all articles
    sentiments = sentiment_analyzer.batch_analyze(
        recent_news['full_text'].tolist()
    )
    
    # Add sentiment scores to dataframe
    recent_news['sentiment'] = [s['sentiment'] for s in sentiments]
    recent_news['sentiment_score'] = [s['numerical_score'] for s in sentiments]
    recent_news['confidence'] = [s['confidence'] for s in sentiments]
    
    print(f"Average sentiment score: {recent_news['sentiment_score'].mean():.3f}")

from web3 import Web3
import json

class DeFiTrader:
    def __init__(self, private_key, infura_url):
        self.w3 = Web3(Web3.HTTPProvider(infura_url))
        self.account = self.w3.eth.account.from_key(private_key)
        self.address = self.account.address
        
        # Uniswap V3 Router address (Ethereum mainnet)
        self.uniswap_router = '0xE592427A0AEce92De3Edee1F18E0157C05861564'
        
        # Load Uniswap Router ABI
        self.router_abi = self.load_uniswap_abi()
        
        # Create contract instance
        self.router_contract = self.w3.eth.contract(
            address=self.uniswap_router,
            abi=self.router_abi
        )
    
    def load_uniswap_abi(self):
        """Load Uniswap V3 Router ABI"""
        # Simplified ABI for essential functions
        return [
            {
                "inputs": [
                    {"internalType": "bytes", "name": "path", "type": "bytes"},
                    {"internalType": "address", "name": "recipient", "type": "address"},
                    {"internalType": "uint256", "name": "deadline", "type": "uint256"},
                    {"internalType": "uint256", "name": "amountIn", "type": "uint256"},
                    {"internalType": "uint256", "name": "amountOutMinimum", "type": "uint256"}
                ],
                "name": "exactInputSingle",
                "outputs": [{"internalType": "uint256", "name": "amountOut", "type": "uint256"}],
                "stateMutability": "payable",
                "type": "function"
            }
        ]
    
    def get_token_balance(self, token_address):
        """Get ERC20 token balance"""
        if token_address == '0x0000000000000000000000000000000000000000':
            # ETH balance
            return self.w3.eth.get_balance(self.address)
        
        # ERC20 token balance
        token_contract = self.w3.eth.contract(
            address=token_address,
            abi=[{
                "constant": True,
                "inputs": [{"name": "_owner", "type": "address"}],
                "name": "balanceOf",
                "outputs": [{"name": "balance", "type": "uint256"}],
                "type": "function"
            }]
        )
        return token_contract.functions.balanceOf(self.address).call()

class SentimentTradingStrategy:
    def __init__(self, defi_trader, sentiment_analyzer):
        self.trader = defi_trader
        self.analyzer = sentiment_analyzer
        self.trading_pairs = {
            'ETH/USDC': {
                'token0': '0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2',  # WETH
                'token1': '0xA0b86a33E6441e2D5e5F14EF59ba1E0a8bdc6e2f',  # USDC
                'fee': 3000
            }
        }
        
        # Trading thresholds
        self.buy_threshold = 0.3   # Buy when sentiment > 0.3
        self.sell_threshold = -0.3  # Sell when sentiment < -0.3
        self.min_confidence = 0.7   # Minimum confidence for trades
    
    def calculate_position_size(self, sentiment_score, confidence):
        """Calculate trade size based on sentiment strength"""
        # Base position size (percentage of portfolio)
        base_size = 0.1  # 10% of portfolio
        
        # Adjust based on sentiment strength and confidence
        sentiment_multiplier = abs(sentiment_score)
        confidence_multiplier = confidence
        
        position_size = base_size * sentiment_multiplier * confidence_multiplier
        
        # Cap maximum position size
        return min(position_size, 0.25)  # Maximum 25% of portfolio
    
    def should_trade(self, sentiment_data):
        """Determine if conditions are met for trading"""
        avg_sentiment = np.mean([s['numerical_score'] for s in sentiment_data])
        avg_confidence = np.mean([s['confidence'] for s in sentiment_data])
        
        # Check minimum confidence threshold
        if avg_confidence < self.min_confidence:
            return False, 0, "Low confidence"
        
        # Check sentiment thresholds
        if avg_sentiment > self.buy_threshold:
            return True, avg_sentiment, "BUY"
        elif avg_sentiment < self.sell_threshold:
            return True, avg_sentiment, "SELL"
        
        return False, avg_sentiment, "HOLD"
    
    def execute_sentiment_trade(self, sentiment_data):
        """Execute trade based on sentiment analysis"""
        should_trade, sentiment_score, action = self.should_trade(sentiment_data)
        
        if not should_trade:
            print(f"No trade executed. Action: {action}, Score: {sentiment_score:.3f}")
            return None
        
        # Calculate position size
        avg_confidence = np.mean([s['confidence'] for s in sentiment_data])
        position_size = self.calculate_position_size(sentiment_score, avg_confidence)
        
        print(f"Executing {action} trade. Sentiment: {sentiment_score:.3f}, Size: {position_size:.1%}")
        
        # Execute the trade (simplified example)
        trade_result = self.execute_trade(action, position_size)
        
        return {
            'action': action,
            'sentiment_score': sentiment_score,
            'confidence': avg_confidence,
            'position_size': position_size,
            'trade_result': trade_result
        }
    
    def execute_trade(self, action, position_size):
        """Execute the actual trade on DeFi protocol"""
        # This is a simplified example
        # In production, implement proper slippage protection and gas optimization
        
        try:
            if action == "BUY":
                # Buy ETH with USDC based on positive sentiment
                result = self.buy_eth_with_sentiment(position_size)
            elif action == "SELL":
                # Sell ETH for USDC based on negative sentiment
                result = self.sell_eth_with_sentiment(position_size)
            
            return {"status": "success", "details": result}
        
        except Exception as e:
            return {"status": "failed", "error": str(e)}
    
    def buy_eth_with_sentiment(self, position_size):
        """Buy ETH when sentiment is positive"""
        # Get USDC balance
        usdc_balance = self.trader.get_token_balance(
            self.trading_pairs['ETH/USDC']['token1']
        )
        
        # Calculate trade amount
        trade_amount = int(usdc_balance * position_size)
        
        # Execute swap (simplified)
        print(f"Buying ETH with {trade_amount} USDC")
        return {"type": "buy", "amount": trade_amount}
    
    def sell_eth_with_sentiment(self, position_size):
        """Sell ETH when sentiment is negative"""
        # Get ETH balance
        eth_balance = self.trader.get_token_balance(
            self.trading_pairs['ETH/USDC']['token0']
        )
        
        # Calculate trade amount
        trade_amount = int(eth_balance * position_size)
        
        # Execute swap (simplified)
        print(f"Selling {trade_amount} ETH for USDC")
        return {"type": "sell", "amount": trade_amount}

# Example usage
# Initialize components (requires actual private key and Infura URL)
# defi_trader = DeFiTrader(private_key='YOUR_PRIVATE_KEY', infura_url='YOUR_INFURA_URL')
# trading_strategy = SentimentTradingStrategy(defi_trader, sentiment_analyzer)

# Simulate trading decision
if not recent_news.empty and len(sentiments) > 0:
    # Create trading strategy instance (simulation mode)
    class SimulatedStrategy(SentimentTradingStrategy):
        def __init__(self):
            # Initialize without actual DeFi connection for demo
            pass
        
        def execute_trade(self, action, position_size):
            return {"status": "simulated", "action": action, "size": position_size}
    
    sim_strategy = SimulatedStrategy()
    trade_decision = sim_strategy.execute_sentiment_trade(sentiments)
    print(f"Trade Decision: {trade_decision}")

class RiskManager:
    def __init__(self, max_portfolio_risk=0.02, max_position_size=0.25):
        self.max_portfolio_risk = max_portfolio_risk  # 2% max risk per trade
        self.max_position_size = max_position_size    # 25% max position
        self.sentiment_history = []
        self.trade_history = []
    
    def calculate_dynamic_position_size(self, sentiment_data, market_volatility):
        """Calculate position size based on sentiment and market conditions"""
        
        # Base calculation from sentiment
        avg_sentiment = np.mean([s['numerical_score'] for s in sentiment_data])
        avg_confidence = np.mean([s['confidence'] for s in sentiment_data])
        
        # Sentiment strength factor (0 to 1)
        sentiment_strength = abs(avg_sentiment)
        
        # Confidence factor (0 to 1)
        confidence_factor = avg_confidence
        
        # Volatility adjustment (reduce size in high volatility)
        volatility_factor = max(0.1, 1 - market_volatility)
        
        # Calculate base position size
        base_size = (sentiment_strength * confidence_factor * volatility_factor)
        
        # Apply maximum limits
        position_size = min(base_size, self.max_position_size)
        
        return position_size
    
    def assess_sentiment_quality(self, sentiment_data):
        """Evaluate quality of sentiment signals"""
        if not sentiment_data:
            return {"quality": "poor", "score": 0}
        
        # Check for conflicting sentiments
        sentiments = [s['numerical_score'] for s in sentiment_data]
        confidences = [s['confidence'] for s in sentiment_data]
        
        # Calculate consensus
        sentiment_std = np.std(sentiments)
        avg_confidence = np.mean(confidences)
        
        # Quality metrics
        consensus_score = 1 - min(sentiment_std / 0.5, 1)  # Lower std = higher consensus
        confidence_score = avg_confidence
        
        overall_quality = (consensus_score + confidence_score) / 2
        
        if overall_quality > 0.8:
            quality = "excellent"
        elif overall_quality > 0.6:
            quality = "good"
        elif overall_quality > 0.4:
            quality = "fair"
        else:
            quality = "poor"
        
        return {
            "quality": quality,
            "score": overall_quality,
            "consensus": consensus_score,
            "confidence": confidence_score
        }
    
    def should_halt_trading(self, recent_performance):
        """Determine if trading should be halted due to poor performance"""
        if len(recent_performance) < 5:
            return False
        
        # Check win rate in last 10 trades
        recent_trades = recent_performance[-10:]
        win_rate = sum(1 for trade in recent_trades if trade['profit'] > 0) / len(recent_trades)
        
        # Check cumulative losses
        total_loss = sum(trade['profit'] for trade in recent_trades if trade['profit'] < 0)
        
        # Halt conditions
        if win_rate < 0.3:  # Win rate below 30%
            return True, "Low win rate"
        
        if total_loss < -0.1:  # Cumulative loss > 10%
            return True, "Excessive losses"
        
        return False, "Trading OK"

# Initialize risk manager
risk_manager = RiskManager()

# Example risk assessment
if sentiments:
    quality_assessment = risk_manager.assess_sentiment_quality(sentiments)
    print(f"Sentiment Quality: {quality_assessment['quality']} (Score: {quality_assessment['score']:.2f})")

class MultiTimeframeSentimentAnalyzer:
    def __init__(self, news_collector, sentiment_analyzer):
        self.news_collector = news_collector
        self.sentiment_analyzer = sentiment_analyzer
        
    def analyze_multiple_timeframes(self):
        """Analyze sentiment across different timeframes"""
        timeframes = {
            '1h': 1,
            '6h': 6,
            '24h': 24,
            '72h': 72
        }
        
        sentiment_by_timeframe = {}
        
        for timeframe, hours in timeframes.items():
            # Collect news for this timeframe
            news_data = self.news_collector.fetch_crypto_news(hours_back=hours)
            
            if not news_data.empty:
                # Analyze sentiment
                full_texts = (news_data['title'] + ' ' + news_data['content'].fillna(''))
                sentiments = self.sentiment_analyzer.batch_analyze(full_texts.tolist())
                
                # Calculate averages
                avg_sentiment = np.mean([s['numerical_score'] for s in sentiments])
                avg_confidence = np.mean([s['confidence'] for s in sentiments])
                article_count = len(sentiments)
                
                sentiment_by_timeframe[timeframe] = {
                    'sentiment': avg_sentiment,
                    'confidence': avg_confidence,
                    'articles': article_count,
                    'raw_data': sentiments
                }
            else:
                sentiment_by_timeframe[timeframe] = {
                    'sentiment': 0,
                    'confidence': 0,
                    'articles': 0,
                    'raw_data': []
                }
        
        return sentiment_by_timeframe
    
    def get_sentiment_trend(self, multi_timeframe_data):
        """Determine if sentiment is improving or deteriorating"""
        timeframes = ['1h', '6h', '24h', '72h']
        sentiments = [multi_timeframe_data[tf]['sentiment'] for tf in timeframes]
        
        # Calculate trend (positive = improving sentiment)
        if len(sentiments) >= 2:
            recent_sentiment = sentiments[0]  # 1h
            older_sentiment = sentiments[-1]   # 72h
            
            trend = recent_sentiment - older_sentiment
            
            if trend > 0.1:
                return "improving"
            elif trend < -0.1:
                return "deteriorating"
            else:
                return "stable"
        
        return "insufficient_data"

# Example multi-timeframe analysis
multi_analyzer = MultiTimeframeSentimentAnalyzer(news_collector, sentiment_analyzer)
timeframe_sentiments = multi_analyzer.analyze_multiple_timeframes()

print("Sentiment Analysis by Timeframe:")
for timeframe, data in timeframe_sentiments.items():
    print(f"{timeframe}: Sentiment={data['sentiment']:.3f}, Articles={data['articles']}")

trend = multi_analyzer.get_sentiment_trend(timeframe_sentiments)
print(f"Sentiment Trend: {trend}")

import time
import logging
from datetime import datetime
import asyncio

class NLPDeFiTradingBot:
    def __init__(self, config):
        self.config = config
        self.setup_logging()
        
        # Initialize components
        self.news_collector = CryptoNewsCollector()
        self.sentiment_analyzer = CryptoSentimentAnalyzer()
        self.risk_manager = RiskManager()
        
        # Initialize DeFi trader (if keys provided)
        if config.get('private_key') and config.get('infura_url'):
            self.defi_trader = DeFiTrader(
                config['private_key'], 
                config['infura_url']
            )
            self.trading_strategy = SentimentTradingStrategy(
                self.defi_trader, 
                self.sentiment_analyzer
            )
        else:
            self.defi_trader = None
            self.trading_strategy = None
            logging.warning("No trading keys provided - running in simulation mode")
        
        # Trading state
        self.is_running = False
        self.last_analysis_time = None
        self.performance_history = []
    
    def setup_logging(self):
        """Configure logging for the trading bot"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('nlp_trading_bot.log'),
                logging.StreamHandler()
            ]
        )
    
    async def run_trading_cycle(self):
        """Execute one complete trading cycle"""
        try:
            # Step 1: Collect recent news
            logging.info("Collecting cryptocurrency news...")
            news_data = self.news_collector.fetch_crypto_news(
                hours_back=self.config.get('news_lookback_hours', 6)
            )
            
            if news_data.empty:
                logging.warning("No news articles found")
                return
            
            # Step 2: Analyze sentiment
            logging.info(f"Analyzing sentiment for {len(news_data)} articles...")
            full_texts = (news_data['title'] + ' ' + news_data['content'].fillna(''))
            sentiments = self.sentiment_analyzer.batch_analyze(full_texts.tolist())
            
            # Step 3: Assess sentiment quality
            quality_assessment = self.risk_manager.assess_sentiment_quality(sentiments)
            logging.info(f"Sentiment quality: {quality_assessment['quality']}")
            
            # Step 4: Check if trading should be halted
            should_halt, halt_reason = self.risk_manager.should_halt_trading(
                self.performance_history
            )
            
            if should_halt:
                logging.warning(f"Trading halted: {halt_reason}")
                return
            
            # Step 5: Make trading decision
            if self.trading_strategy and quality_assessment['score'] > 0.5:
                trade_result = self.trading_strategy.execute_sentiment_trade(sentiments)
                
                if trade_result:
                    logging.info(f"Trade executed: {trade_result}")
                    self.performance_history.append({
                        'timestamp': datetime.now(),
                        'action': trade_result['action'],
                        'sentiment': trade_result['sentiment_score'],
                        'confidence': trade_result['confidence'],
                        'profit': 0  # To be updated later with actual results
                    })
            
            # Step 6: Update analysis time
            self.last_analysis_time = datetime.now()
            
        except Exception as e:
            logging.error(f"Error in trading cycle: {str(e)}")
    
    async def start(self):
        """Start the trading bot"""
        self.is_running = True
        logging.info("NLP DeFi Trading Bot started")
        
        while self.is_running:
            await self.run_trading_cycle()
            
            # Wait for next cycle
            await asyncio.sleep(self.config.get('cycle_interval_seconds', 300))  # 5 minutes
    
    def stop(self):
        """Stop the trading bot"""
        self.is_running = False
        logging.info("NLP DeFi Trading Bot stopped")
    
    def get_status(self):
        """Get current bot status"""
        return {
            'is_running': self.is_running,
            'last_analysis': self.last_analysis_time,
            'total_trades': len(self.performance_history),
            'mode': 'live' if self.defi_trader else 'simulation'
        }

# Configuration for the trading bot
bot_config = {
    'news_lookback_hours': 6,
    'cycle_interval_seconds': 300,  # 5 minutes
    'private_key': None,  # Set to your private key for live trading
    'infura_url': None    # Set to your Infura URL for live trading
}

# Example usage (simulation mode)
trading_bot = NLPDeFiTradingBot(bot_config)
status = trading_bot.get_status()
print(f"Bot Status: {status}")

# To run the bot (uncomment for actual deployment):
# asyncio.run(trading_bot.start())

class PerformanceTracker:
    def __init__(self):
        self.trades = []
        self.metrics = {}
    
    def add_trade(self, trade_data):
        """Record a completed trade"""
        self.trades.append({
            'timestamp': trade_data['timestamp'],
            'action': trade_data['action'],
            'sentiment_score': trade_data['sentiment'],
            'confidence': trade_data['confidence'],
            'entry_price': trade_data['entry_price'],
            'exit_price': trade_data.get('exit_price'),
            'profit_loss': trade_data.get('profit_loss', 0),
            'size': trade_data['size']
        })
    
    def calculate_metrics(self):
        """Calculate performance metrics"""
        if not self.trades:
            return {}
        
        completed_trades = [t for t in self.trades if t.get('exit_price')]
        
        if not completed_trades:
            return {"status": "no_completed_trades"}
        
        # Basic metrics
        total_trades = len(completed_trades)
        winning_trades = sum(1 for t in completed_trades if t['profit_loss'] > 0)
        win_rate = winning_trades / total_trades
        
        # Profit/Loss metrics
        total_pnl = sum(t['profit_loss'] for t in completed_trades)
        avg_win = np.mean([t['profit_loss'] for t in completed_trades if t['profit_loss'] > 0])
        avg_loss = np.mean([t['profit_loss'] for t in completed_trades if t['profit_loss'] < 0])
        
        # Sentiment correlation
        sentiment_scores = [t['sentiment_score'] for t in completed_trades]
        profits = [t['profit_loss'] for t in completed_trades]
        sentiment_correlation = np.corrcoef(sentiment_scores, profits)[0, 1] if len(profits) > 1 else 0
        
        return {
            'total_trades': total_trades,
            'win_rate': win_rate,
            'total_pnl': total_pnl,
            'avg_win': avg_win if not np.isnan(avg_win) else 0,
            'avg_loss': avg_loss if not np.isnan(avg_loss) else 0,
            'sentiment_correlation': sentiment_correlation
        }
    
    def generate_report(self):
        """Generate a comprehensive performance report"""
        metrics = self.calculate_metrics()
        
        if not metrics:
            return "No trades to analyze"
        
        report = f"""
NLP DeFi Trading Performance Report
==================================

Total Trades: {metrics['total_trades']}
Win Rate: {metrics['win_rate']:.1%}
Total P&L: {metrics['total_pnl']:.4f} ETH
Average Win: {metrics['avg_win']:.4f} ETH
Average Loss: {metrics['avg_loss']:.4f} ETH
Sentiment Correlation: {metrics['sentiment_correlation']:.3f}

Strategy Effectiveness: {"Good" if metrics['win_rate'] > 0.5 else "Needs Improvement"}
Sentiment Signal Quality: {"Strong" if abs(metrics['sentiment_correlation']) > 0.3 else "Weak"}
        """
        
        return report

# Example performance tracking
performance_tracker = PerformanceTracker()

# Simulate some trade data
sample_trades = [
    {
        'timestamp': datetime.now(),
        'action': 'BUY',
        'sentiment': 0.6,
        'confidence': 0.8,
        'entry_price': 2000,
        'exit_price': 2100,
        'profit_loss': 0.05,
        'size': 0.1
    },
    {
        'timestamp': datetime.now(),
        'action': 'SELL',
        'sentiment': -0.4,
        'confidence': 0.7,
        'entry_price': 2100,
        'exit_price': 2050,
        'profit_loss': -0.025,
        'size': 0.1
    }
]

for trade in sample_trades:
    performance_tracker.add_trade(trade)

print(performance_tracker.generate_report())