How to Build Trading Signal Generator with Ollama: Multi-Factor Model Creation

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

# Install Python packages
pip install pandas numpy yfinance requests beautifulsoup4 scikit-learn matplotlib seaborn

# Download Llama 3.1 8B model (recommended for trading analysis)
ollama pull llama3.1:8b

# Alternative: smaller model for testing
ollama pull llama3.1:3b

import yfinance as yf
import pandas as pd
import requests
from datetime import datetime, timedelta
import numpy as np

class TradingDataCollector:
    def __init__(self, symbols=['AAPL', 'GOOGL', 'MSFT', 'TSLA']):
        self.symbols = symbols
        self.data = {}
        
    def fetch_market_data(self, period='1y'):
        """Fetch historical price data for all symbols"""
        for symbol in self.symbols:
            try:
                ticker = yf.Ticker(symbol)
                # Get historical data
                hist = ticker.history(period=period)
                # Get fundamental data
                info = ticker.info
                
                self.data[symbol] = {
                    'price_data': hist,
                    'fundamentals': info
                }
                print(f"✓ Collected data for {symbol}")
            except Exception as e:
                print(f"✗ Failed to collect {symbol}: {e}")
                
    def calculate_technical_indicators(self, symbol):
        """Calculate RSI, MACD, Bollinger Bands"""
        if symbol not in self.data:
            return None
            
        df = self.data[symbol]['price_data'].copy()
        
        # 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))
        
        # MACD Calculation
        df['EMA_12'] = df['Close'].ewm(span=12).mean()
        df['EMA_26'] = df['Close'].ewm(span=26).mean()
        df['MACD'] = df['EMA_12'] - df['EMA_26']
        df['MACD_Signal'] = df['MACD'].ewm(span=9).mean()
        
        # Bollinger Bands
        df['BB_Middle'] = df['Close'].rolling(window=20).mean()
        bb_std = df['Close'].rolling(window=20).std()
        df['BB_Upper'] = df['BB_Middle'] + (bb_std * 2)
        df['BB_Lower'] = df['BB_Middle'] - (bb_std * 2)
        
        self.data[symbol]['technical'] = df
        return df
    
    def fetch_news_sentiment(self, symbol):
        """Fetch recent news for sentiment analysis"""
        # Using Alpha Vantage News API (free tier available)
        # Replace with your preferred news API
        news_data = {
            'headlines': [
                f"Recent positive developments for {symbol}",
                f"Market analysts bullish on {symbol}",
                f"Strong earnings report from {symbol}"
            ],
            'sentiment_score': 0.65  # Placeholder - implement actual sentiment
        }
        return news_data

# Initialize data collector
collector = TradingDataCollector()
collector.fetch_market_data()

# Calculate technical indicators for each symbol
for symbol in collector.symbols:
    collector.calculate_technical_indicators(symbol)

import json
import subprocess
import requests

class OllamaSignalGenerator:
    def __init__(self, model_name='llama3.1:8b'):
        self.model_name = model_name
        self.base_url = 'http://localhost:11434'
        
    def generate_trading_prompt(self, symbol, data):
        """Create comprehensive trading analysis prompt"""
        tech_data = data['technical'].iloc[-1]  # Latest technical indicators
        fundamentals = data['fundamentals']
        
        prompt = f"""
        As an expert quantitative analyst, analyze {symbol} and provide trading signals.
        
        CURRENT TECHNICAL INDICATORS:
        - RSI: {tech_data['RSI']:.2f}
        - MACD: {tech_data['MACD']:.4f}
        - MACD Signal: {tech_data['MACD_Signal']:.4f}
        - Price vs BB Upper: {((tech_data['Close'] - tech_data['BB_Upper']) / tech_data['BB_Upper'] * 100):.2f}%
        - Price vs BB Lower: {((tech_data['Close'] - tech_data['BB_Lower']) / tech_data['BB_Lower'] * 100):.2f}%
        - Current Price: ${tech_data['Close']:.2f}
        
        FUNDAMENTAL METRICS:
        - P/E Ratio: {fundamentals.get('forwardPE', 'N/A')}
        - Debt-to-Equity: {fundamentals.get('debtToEquity', 'N/A')}
        - Revenue Growth: {fundamentals.get('revenueGrowth', 'N/A')}
        - Profit Margin: {fundamentals.get('profitMargins', 'N/A')}
        
        ANALYSIS FRAMEWORK:
        1. Technical Analysis: Evaluate momentum, trend strength, and support/resistance
        2. Fundamental Analysis: Assess valuation and financial health
        3. Risk Assessment: Identify key risk factors
        4. Signal Generation: Provide BUY/SELL/HOLD with confidence score (0-100)
        
        Provide your analysis in this JSON format:
        {{
            "technical_score": 0-100,
            "fundamental_score": 0-100,
            "overall_signal": "BUY/SELL/HOLD",
            "confidence": 0-100,
            "target_price": price_target,
            "stop_loss": stop_loss_price,
            "risk_factors": ["factor1", "factor2"],
            "reasoning": "detailed_explanation"
        }}
        """
        return prompt
    
    def get_ai_signal(self, symbol, data):
        """Get trading signal from Ollama"""
        prompt = self.generate_trading_prompt(symbol, data)
        
        try:
            response = requests.post(
                f"{self.base_url}/api/generate",
                json={
                    "model": self.model_name,
                    "prompt": prompt,
                    "stream": False
                }
            )
            
            if response.status_code == 200:
                result = response.json()
                # Extract JSON from response
                ai_response = result['response']
                
                # Parse JSON from AI response
                try:
                    # Find JSON in response
                    json_start = ai_response.find('{')
                    json_end = ai_response.rfind('}') + 1
                    json_str = ai_response[json_start:json_end]
                    
                    signal_data = json.loads(json_str)
                    return signal_data
                except:
                    # Fallback parsing
                    return self.parse_text_response(ai_response)
            else:
                print(f"Error: {response.status_code}")
                return None
                
        except Exception as e:
            print(f"Ollama request failed: {e}")
            return None
    
    def parse_text_response(self, text):
        """Fallback parser for non-JSON responses"""
        # Simple keyword-based parsing
        signal = "HOLD"
        confidence = 50
        
        if "BUY" in text.upper():
            signal = "BUY"
            confidence = 70
        elif "SELL" in text.upper():
            signal = "SELL"
            confidence = 70
            
        return {
            "overall_signal": signal,
            "confidence": confidence,
            "reasoning": text[:200] + "..."
        }

class MultiFactorSignalAggregator:
    def __init__(self, weights=None):
        # Default weights for different factors
        self.weights = weights or {
            'technical': 0.4,
            'fundamental': 0.3,
            'ai_analysis': 0.2,
            'sentiment': 0.1
        }
    
    def calculate_technical_score(self, data):
        """Calculate technical analysis score (0-100)"""
        latest = data['technical'].iloc[-1]
        score = 50  # Neutral starting point
        
        # RSI analysis
        if latest['RSI'] < 30:
            score += 20  # Oversold - bullish
        elif latest['RSI'] > 70:
            score -= 20  # Overbought - bearish
        
        # MACD analysis
        if latest['MACD'] > latest['MACD_Signal']:
            score += 15  # Bullish crossover
        else:
            score -= 15  # Bearish crossover
        
        # Bollinger Bands analysis
        if latest['Close'] < latest['BB_Lower']:
            score += 10  # Below lower band - potential bounce
        elif latest['Close'] > latest['BB_Upper']:
            score -= 10  # Above upper band - potential pullback
        
        return max(0, min(100, score))
    
    def calculate_fundamental_score(self, fundamentals):
        """Calculate fundamental analysis score (0-100)"""
        score = 50  # Neutral starting point
        
        # P/E ratio analysis
        pe_ratio = fundamentals.get('forwardPE', 20)
        if pe_ratio and pe_ratio < 15:
            score += 15  # Undervalued
        elif pe_ratio and pe_ratio > 25:
            score -= 15  # Overvalued
        
        # Debt-to-equity analysis
        debt_equity = fundamentals.get('debtToEquity', 50)
        if debt_equity and debt_equity < 30:
            score += 10  # Low debt - good
        elif debt_equity and debt_equity > 80:
            score -= 10  # High debt - risky
        
        # Revenue growth
        revenue_growth = fundamentals.get('revenueGrowth', 0)
        if revenue_growth and revenue_growth > 0.1:
            score += 10  # Strong growth
        elif revenue_growth and revenue_growth < -0.05:
            score -= 10  # Declining revenue
        
        return max(0, min(100, score))
    
    def aggregate_signals(self, symbol, data_collector, ollama_generator):
        """Combine all factors into final signal"""
        data = data_collector.data[symbol]
        
        # Calculate individual scores
        technical_score = self.calculate_technical_score(data)
        fundamental_score = self.calculate_fundamental_score(data['fundamentals'])
        
        # Get AI analysis
        ai_signal = ollama_generator.get_ai_signal(symbol, data)
        ai_score = ai_signal['confidence'] if ai_signal else 50
        
        # Placeholder sentiment score
        sentiment_score = 60  # Implement actual sentiment analysis
        
        # Calculate weighted final score
        final_score = (
            technical_score * self.weights['technical'] +
            fundamental_score * self.weights['fundamental'] +
            ai_score * self.weights['ai_analysis'] +
            sentiment_score * self.weights['sentiment']
        )
        
        # Generate signal
        if final_score > 70:
            signal = "BUY"
        elif final_score < 30:
            signal = "SELL"
        else:
            signal = "HOLD"
        
        return {
            'symbol': symbol,
            'final_signal': signal,
            'confidence': final_score,
            'component_scores': {
                'technical': technical_score,
                'fundamental': fundamental_score,
                'ai_analysis': ai_score,
                'sentiment': sentiment_score
            },
            'ai_insights': ai_signal,
            'timestamp': datetime.now().isoformat()
        }

class RiskManager:
    def __init__(self, max_position_size=0.1, max_portfolio_risk=0.02):
        self.max_position_size = max_position_size  # 10% max per position
        self.max_portfolio_risk = max_portfolio_risk  # 2% max portfolio risk
        
    def calculate_position_size(self, signal_data, account_balance, current_price):
        """Calculate optimal position size based on risk"""
        confidence = signal_data['confidence']
        
        # Base position size on confidence
        base_size = account_balance * self.max_position_size
        confidence_multiplier = confidence / 100
        
        # Adjust for risk
        adjusted_size = base_size * confidence_multiplier
        
        # Calculate number of shares
        shares = int(adjusted_size / current_price)
        
        # Apply stop-loss
        stop_loss_price = self.calculate_stop_loss(current_price, signal_data)
        
        return {
            'shares': shares,
            'position_value': shares * current_price,
            'stop_loss': stop_loss_price,
            'risk_amount': abs(current_price - stop_loss_price) * shares
        }
    
    def calculate_stop_loss(self, current_price, signal_data):
        """Calculate stop-loss price"""
        if signal_data['final_signal'] == 'BUY':
            # 5% stop-loss for buy signals
            return current_price * 0.95
        elif signal_data['final_signal'] == 'SELL':
            # 5% stop-loss for sell signals (short positions)
            return current_price * 1.05
        else:
            return current_price

class OllamaTradingSystem:
    def __init__(self, symbols, account_balance=10000):
        self.symbols = symbols
        self.account_balance = account_balance
        self.data_collector = TradingDataCollector(symbols)
        self.ollama_generator = OllamaSignalGenerator()
        self.signal_aggregator = MultiFactorSignalAggregator()
        self.risk_manager = RiskManager()
        self.signals_history = []
        
    def run_analysis(self):
        """Run complete trading analysis"""
        print("🚀 Starting Ollama Trading Analysis...")
        
        # Collect fresh data
        self.data_collector.fetch_market_data()
        
        results = []
        
        for symbol in self.symbols:
            print(f"\n📊 Analyzing {symbol}...")
            
            # Calculate technical indicators
            self.data_collector.calculate_technical_indicators(symbol)
            
            # Generate multi-factor signal
            signal_data = self.signal_aggregator.aggregate_signals(
                symbol, self.data_collector, self.ollama_generator
            )
            
            # Calculate position sizing
            current_price = self.data_collector.data[symbol]['price_data']['Close'].iloc[-1]
            position_info = self.risk_manager.calculate_position_size(
                signal_data, self.account_balance, current_price
            )
            
            # Combine results
            complete_signal = {
                **signal_data,
                'current_price': current_price,
                'position_info': position_info
            }
            
            results.append(complete_signal)
            self.signals_history.append(complete_signal)
            
            # Display results
            self.display_signal(complete_signal)
            
        return results
    
    def display_signal(self, signal):
        """Display trading signal in readable format"""
        print(f"\n{'='*50}")
        print(f"SYMBOL: {signal['symbol']}")
        print(f"SIGNAL: {signal['final_signal']} (Confidence: {signal['confidence']:.1f}%)")
        print(f"CURRENT PRICE: ${signal['current_price']:.2f}")
        
        if signal['final_signal'] != 'HOLD':
            pos_info = signal['position_info']
            print(f"SUGGESTED SHARES: {pos_info['shares']}")
            print(f"POSITION VALUE: ${pos_info['position_value']:.2f}")
            print(f"STOP LOSS: ${pos_info['stop_loss']:.2f}")
            print(f"RISK AMOUNT: ${pos_info['risk_amount']:.2f}")
        
        print(f"\nCOMPONENT SCORES:")
        for component, score in signal['component_scores'].items():
            print(f"  {component.title()}: {score:.1f}")
        
        if signal['ai_insights']:
            print(f"\nAI REASONING: {signal['ai_insights'].get('reasoning', 'N/A')[:100]}...")

# Example usage
if __name__ == "__main__":
    # Initialize trading system
    trading_system = OllamaTradingSystem(['AAPL', 'GOOGL', 'MSFT', 'TSLA'])
    
    # Run analysis
    signals = trading_system.run_analysis()
    
    # Display summary
    print(f"\n{'='*50}")
    print("TRADING SIGNALS SUMMARY")
    print(f"{'='*50}")
    
    for signal in signals:
        print(f"{signal['symbol']}: {signal['final_signal']} ({signal['confidence']:.1f}%)")

import schedule
import time
import logging
from datetime import datetime

class TradingSystemScheduler:
    def __init__(self, trading_system):
        self.trading_system = trading_system
        self.setup_logging()
        
    def setup_logging(self):
        """Setup logging for trading system"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('trading_signals.log'),
                logging.StreamHandler()
            ]
        )
        
    def run_daily_analysis(self):
        """Run daily trading analysis"""
        try:
            logging.info("Starting daily trading analysis")
            signals = self.trading_system.run_analysis()
            
            # Save signals to file
            self.save_signals(signals)
            
            # Send alerts for high-confidence signals
            self.send_alerts(signals)
            
            logging.info(f"Analysis complete. Generated {len(signals)} signals")
            
        except Exception as e:
            logging.error(f"Analysis failed: {e}")
    
    def save_signals(self, signals):
        """Save signals to CSV file"""
        import pandas as pd
        
        df = pd.DataFrame(signals)
        filename = f"signals_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
        df.to_csv(filename, index=False)
        logging.info(f"Signals saved to {filename}")
    
    def send_alerts(self, signals):
        """Send alerts for high-confidence signals"""
        high_confidence_signals = [s for s in signals if s['confidence'] > 80]
        
        if high_confidence_signals:
            logging.info(f"High confidence signals: {len(high_confidence_signals)}")
            # Implement email/SMS alerts here
            
    def start_scheduler(self):
        """Start the scheduled trading system"""
        # Schedule daily analysis at market open
        schedule.every().day.at("09:30").do(self.run_daily_analysis)
        
        # Schedule additional analysis during market hours
        schedule.every().hour.do(self.run_daily_analysis)
        
        logging.info("Trading system scheduler started")
        
        while True:
            schedule.run_pending()
            time.sleep(60)  # Check every minute

# Start automated trading system
# scheduler = TradingSystemScheduler(trading_system)
# scheduler.start_scheduler()

def test_ollama_connection():
    """Test Ollama connection and model availability"""
    try:
        response = requests.get("http://localhost:11434/api/tags")
        if response.status_code == 200:
            models = response.json()
            print(f"✓ Ollama running. Available models: {len(models['models'])}")
            return True
        else:
            print("✗ Ollama not responding")
            return False
    except:
        print("✗ Ollama connection failed")
        return False

def validate_data_quality(data):
    """Validate trading data quality"""
    issues = []
    
    if data.empty:
        issues.append("Empty dataset")
    
    if data.isnull().sum().sum() > len(data) * 0.1:
        issues.append("Too many missing values")
    
    if len(data) < 50:
        issues.append("Insufficient historical data")
    
    return issues