Forex Trading Strategy with Ollama: AI-Powered Currency Pair Analysis and Prediction

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

# Verify installation
ollama --version

# Download general analysis model
ollama pull llama2

# Download code generation model for trading scripts
ollama pull codellama

# Download specialized finance model (if available)
ollama pull finance-llm

import yfinance as yf
import pandas as pd
import requests
import json
from datetime import datetime, timedelta

class ForexDataCollector:
    def __init__(self):
        self.currency_pairs = ['EURUSD=X', 'GBPUSD=X', 'USDJPY=X', 'AUDUSD=X']
    
    def fetch_currency_data(self, pair, period='1mo'):
        """
        Fetch historical forex data for specified currency pair
        Returns: DataFrame with OHLCV data
        """
        ticker = yf.Ticker(pair)
        data = ticker.history(period=period)
        return data
    
    def get_economic_indicators(self):
        """
        Gather economic indicators affecting currency movements
        Returns: Dictionary with economic data
        """
        # Placeholder for economic data API integration
        indicators = {
            'fed_rate': 5.25,
            'ecb_rate': 4.50,
            'inflation_us': 3.2,
            'inflation_eu': 2.8
        }
        return indicators

# Initialize data collector
collector = ForexDataCollector()
eur_usd_data = collector.fetch_currency_data('EURUSD=X')
print("Data collected successfully:", len(eur_usd_data), "records")

import subprocess
import json

class OllamaForexAnalyzer:
    def __init__(self, model='llama2'):
        self.model = model
    
    def analyze_currency_trend(self, data, pair):
        """
        Use Ollama to analyze currency pair trends
        Returns: Trading signal and confidence level
        """
        # Prepare data summary for Ollama
        recent_prices = data.tail(10)['Close'].tolist()
        price_change = ((data['Close'].iloc[-1] - data['Close'].iloc[-5]) / data['Close'].iloc[-5]) * 100
        
        prompt = f"""
        Analyze this forex data for {pair}:
        
        Recent closing prices: {recent_prices}
        5-day price change: {price_change:.2f}%
        Current price: {data['Close'].iloc[-1]:.4f}
        
        Provide:
        1. Trend direction (bullish/bearish/neutral)
        2. Confidence level (1-10)
        3. Key support/resistance levels
        4. Trading recommendation (buy/sell/hold)
        5. Risk assessment
        
        Format as JSON.
        """
        
        # Call Ollama API
        result = subprocess.run([
            'ollama', 'run', self.model, prompt
        ], capture_output=True, text=True)
        
        return result.stdout

# Initialize analyzer
analyzer = OllamaForexAnalyzer()
analysis = analyzer.analyze_currency_trend(eur_usd_data, 'EUR/USD')
print("Ollama Analysis:", analysis)

import talib
import numpy as np

class TechnicalAnalysisAI:
    def __init__(self):
        self.analyzer = OllamaForexAnalyzer()
    
    def calculate_indicators(self, data):
        """
        Calculate technical indicators for Ollama analysis
        Returns: Dictionary with indicator values
        """
        close_prices = data['Close'].values
        
        indicators = {
            'rsi': talib.RSI(close_prices, timeperiod=14)[-1],
            'macd': talib.MACD(close_prices)[0][-1],
            'bollinger_upper': talib.BBANDS(close_prices)[0][-1],
            'bollinger_lower': talib.BBANDS(close_prices)[2][-1],
            'ema_20': talib.EMA(close_prices, timeperiod=20)[-1],
            'ema_50': talib.EMA(close_prices, timeperiod=50)[-1]
        }
        
        return indicators
    
    def generate_trading_signal(self, data, pair):
        """
        Combine technical indicators with Ollama analysis
        Returns: Comprehensive trading signal
        """
        indicators = self.calculate_indicators(data)
        current_price = data['Close'].iloc[-1]
        
        prompt = f"""
        Advanced forex analysis for {pair}:
        
        Current Price: {current_price:.4f}
        RSI: {indicators['rsi']:.2f}
        MACD: {indicators['macd']:.4f}
        Bollinger Upper: {indicators['bollinger_upper']:.4f}
        Bollinger Lower: {indicators['bollinger_lower']:.4f}
        EMA 20: {indicators['ema_20']:.4f}
        EMA 50: {indicators['ema_50']:.4f}
        
        Generate trading signal:
        1. Signal: BUY/SELL/HOLD
        2. Entry price
        3. Stop loss level
        4. Take profit targets
        5. Position size (risk 2% of account)
        6. Confidence score (1-100)
        
        Consider overbought/oversold conditions, trend direction, and support/resistance levels.
        """
        
        result = subprocess.run([
            'ollama', 'run', 'llama2', prompt
        ], capture_output=True, text=True)
        
        return result.stdout

# Initialize technical analyzer
tech_analyzer = TechnicalAnalysisAI()
signal = tech_analyzer.generate_trading_signal(eur_usd_data, 'EUR/USD')
print("Trading Signal:", signal)

class MultiCurrencyStrategy:
    def __init__(self):
        self.pairs = ['EURUSD=X', 'GBPUSD=X', 'USDJPY=X', 'AUDUSD=X']
        self.collector = ForexDataCollector()
        self.analyzer = OllamaForexAnalyzer()
    
    def scan_all_pairs(self):
        """
        Scan all currency pairs for trading opportunities
        Returns: List of trading signals
        """
        signals = []
        
        for pair in self.pairs:
            data = self.collector.fetch_currency_data(pair)
            analysis = self.analyzer.analyze_currency_trend(data, pair)
            
            signals.append({
                'pair': pair,
                'analysis': analysis,
                'timestamp': datetime.now()
            })
        
        return signals
    
    def rank_opportunities(self, signals):
        """
        Use Ollama to rank trading opportunities by potential
        Returns: Ranked list of currency pairs
        """
        prompt = f"""
        Rank these forex trading opportunities from best to worst:
        
        {json.dumps(signals, indent=2, default=str)}
        
        Consider:
        1. Risk-reward ratio
        2. Market volatility
        3. Economic fundamentals
        4. Technical strength
        
        Provide ranking with explanations.
        """
        
        result = subprocess.run([
            'ollama', 'run', 'llama2', prompt
        ], capture_output=True, text=True)
        
        return result.stdout

# Execute multi-currency analysis
strategy = MultiCurrencyStrategy()
all_signals = strategy.scan_all_pairs()
rankings = strategy.rank_opportunities(all_signals)
print("Opportunity Rankings:", rankings)

class RiskManager:
    def __init__(self, account_balance=10000):
        self.account_balance = account_balance
        self.max_risk_per_trade = 0.02  # 2% risk per trade
    
    def calculate_position_size(self, entry_price, stop_loss, pair):
        """
        Calculate optimal position size using Ollama
        Returns: Position size and risk metrics
        """
        risk_amount = self.account_balance * self.max_risk_per_trade
        price_difference = abs(entry_price - stop_loss)
        
        prompt = f"""
        Calculate forex position size:
        
        Account Balance: ${self.account_balance}
        Risk per Trade: {self.max_risk_per_trade * 100}%
        Entry Price: {entry_price}
        Stop Loss: {stop_loss}
        Currency Pair: {pair}
        
        Calculate:
        1. Position size in units
        2. Risk amount in dollars
        3. Potential loss percentage
        4. Leverage required
        5. Margin requirement
        
        Show calculations step by step.
        """
        
        result = subprocess.run([
            'ollama', 'run', 'llama2', prompt
        ], capture_output=True, text=True)
        
        return result.stdout

# Initialize risk manager
risk_mgr = RiskManager(account_balance=50000)
position_calc = risk_mgr.calculate_position_size(1.0950, 1.0900, 'EUR/USD')
print("Position Sizing:", position_calc)

class StrategyBacktester:
    def __init__(self):
        self.analyzer = OllamaForexAnalyzer()
        self.initial_balance = 10000
    
    def backtest_strategy(self, pair, start_date, end_date):
        """
        Backtest trading strategy using historical data
        Returns: Performance metrics
        """
        # Fetch historical data
        ticker = yf.Ticker(pair)
        data = ticker.history(start=start_date, end=end_date)
        
        trades = []
        balance = self.initial_balance
        
        # Simulate trading decisions
        for i in range(20, len(data)):
            window_data = data.iloc[i-20:i]
            signal = self.simulate_ollama_signal(window_data, pair)
            
            if signal['action'] in ['BUY', 'SELL']:
                trade_result = self.execute_simulated_trade(
                    signal, data.iloc[i], balance
                )
                trades.append(trade_result)
                balance = trade_result['new_balance']
        
        return self.calculate_performance_metrics(trades, balance)
    
    def simulate_ollama_signal(self, data, pair):
        """
        Simulate Ollama trading signal for backtesting
        Returns: Trading signal dictionary
        """
        # Simplified signal generation for backtesting
        price_change = ((data['Close'].iloc[-1] - data['Close'].iloc[-5]) / data['Close'].iloc[-5]) * 100
        
        if price_change > 1.5:
            return {'action': 'BUY', 'confidence': 0.8}
        elif price_change < -1.5:
            return {'action': 'SELL', 'confidence': 0.8}
        else:
            return {'action': 'HOLD', 'confidence': 0.5}
    
    def calculate_performance_metrics(self, trades, final_balance):
        """
        Calculate comprehensive performance statistics
        Returns: Performance dictionary
        """
        total_return = ((final_balance - self.initial_balance) / self.initial_balance) * 100
        winning_trades = len([t for t in trades if t['profit'] > 0])
        total_trades = len(trades)
        win_rate = (winning_trades / total_trades) * 100 if total_trades > 0 else 0
        
        return {
            'total_return': total_return,
            'final_balance': final_balance,
            'total_trades': total_trades,
            'win_rate': win_rate,
            'profitable_trades': winning_trades
        }

# Run backtest
backtester = StrategyBacktester()
performance = backtester.backtest_strategy('EURUSD=X', '2024-01-01', '2024-12-31')
print("Backtest Results:", performance)

import time
import logging
from datetime import datetime

class LiveTradingMonitor:
    def __init__(self):
        self.analyzer = OllamaForexAnalyzer()
        self.risk_manager = RiskManager()
        self.is_running = False
        
        # Configure logging
        logging.basicConfig(
            filename='trading_log.txt',
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
    
    def start_monitoring(self, pairs, interval=300):
        """
        Start live market monitoring
        interval: seconds between analysis cycles
        """
        self.is_running = True
        logging.info("Live trading monitor started")
        
        while self.is_running:
            for pair in pairs:
                try:
                    # Fetch real-time data
                    data = self.get_realtime_data(pair)
                    
                    # Generate trading signal
                    signal = self.analyzer.analyze_currency_trend(data, pair)
                    
                    # Log analysis
                    logging.info(f"Analysis for {pair}: {signal}")
                    
                    # Check for trading opportunities
                    if self.is_strong_signal(signal):
                        self.alert_trading_opportunity(pair, signal)
                    
                except Exception as e:
                    logging.error(f"Error analyzing {pair}: {str(e)}")
            
            # Wait before next analysis cycle
            time.sleep(interval)
    
    def get_realtime_data(self, pair):
        """
        Fetch real-time forex data
        Returns: Recent price data
        """
        ticker = yf.Ticker(pair)
        data = ticker.history(period='5d', interval='1h')
        return data
    
    def is_strong_signal(self, signal):
        """
        Determine if signal meets trading criteria
        Returns: Boolean
        """
        # Parse signal strength from Ollama response
        # This would need more sophisticated parsing
        return 'BUY' in signal or 'SELL' in signal
    
    def alert_trading_opportunity(self, pair, signal):
        """
        Alert trader of potential opportunity
        """
        alert_message = f"""
        TRADING OPPORTUNITY DETECTED
        
        Pair: {pair}
        Time: {datetime.now()}
        Signal: {signal}
        
        Review and consider manual execution.
        """
        
        print(alert_message)
        logging.info(alert_message)

# Initialize live monitor
monitor = LiveTradingMonitor()
# monitor.start_monitoring(['EURUSD=X', 'GBPUSD=X'])  # Uncomment to start

class ForexModelTrainer:
    def __init__(self):
        self.training_data = []
    
    def prepare_training_data(self):
        """
        Prepare forex-specific training data for Ollama
        Returns: Formatted training dataset
        """
        # Collect historical forex data with outcomes
        training_examples = [
            {
                "input": "EUR/USD showing strong bullish momentum, RSI at 65, breaking resistance at 1.0950",
                "output": "BUY signal with high confidence. Target 1.1000, stop loss 1.0920"
            },
            {
                "input": "GBP/USD oversold conditions, RSI at 25, approaching strong support at 1.2500",
                "output": "Potential BUY setup forming. Watch for reversal confirmation at support level"
            }
        ]
        
        return training_examples
    
    def create_modelfile(self):
        """
        Create Modelfile for forex-specific Ollama model
        Returns: Modelfile content
        """
        modelfile_content = """
FROM llama2

# Set forex trading parameters
PARAMETER temperature 0.3
PARAMETER top_k 20
PARAMETER top_p 0.9

# Forex trading system prompt
SYSTEM You are an expert forex trading assistant specialized in currency pair analysis and prediction. 

You analyze market data and provide:
1. Clear BUY/SELL/HOLD signals
2. Entry and exit points
3. Risk management advice
4. Confidence levels

Always consider:
- Technical indicators (RSI, MACD, Bollinger Bands)
- Support and resistance levels
- Market sentiment
- Economic factors
- Risk-reward ratios

Provide concise, actionable advice in a structured format.
"""
        return modelfile_content

# Create custom forex model
trainer = ForexModelTrainer()
modelfile = trainer.create_modelfile()

# Save and build custom model
with open('Modelfile.forex', 'w') as f:
    f.write(modelfile)

print("Run: ollama create forex-expert -f Modelfile.forex")

import feedparser
import re

class NewsSentimentAnalyzer:
    def __init__(self):
        self.news_sources = [
            'https://feeds.finance.yahoo.com/rss/2.0/headline',
            'https://www.forexfactory.com/rss.php'
        ]
    
    def fetch_forex_news(self):
        """
        Fetch recent forex-related news
        Returns: List of news articles
        """
        news_items = []
        
        for source in self.news_sources:
            try:
                feed = feedparser.parse(source)
                for entry in feed.entries[:10]:  # Latest 10 articles
                    if self.is_forex_related(entry.title + ' ' + entry.summary):
                        news_items.append({
                            'title': entry.title,
                            'summary': entry.summary,
                            'published': entry.published,
                            'link': entry.link
                        })
            except Exception as e:
                print(f"Error fetching from {source}: {e}")
        
        return news_items
    
    def is_forex_related(self, text):
        """
        Check if news article is forex-related
        Returns: Boolean
        """
        forex_keywords = [
            'EUR', 'USD', 'GBP', 'JPY', 'forex', 'currency',
            'central bank', 'interest rate', 'inflation',
            'Federal Reserve', 'ECB', 'BOJ'
        ]
        
        return any(keyword.lower() in text.lower() for keyword in forex_keywords)
    
    def analyze_sentiment_impact(self, news_items, currency_pair):
        """
        Use Ollama to analyze news sentiment impact on currency pair
        Returns: Sentiment analysis
        """
        news_summary = '\n'.join([
            f"- {item['title']}: {item['summary'][:100]}..."
            for item in news_items[:5]
        ])
        
        prompt = f"""
        Analyze the sentiment impact of these recent news items on {currency_pair}:
        
        {news_summary}
        
        Provide:
        1. Overall sentiment (bullish/bearish/neutral)
        2. Specific impact on {currency_pair}
        3. Time horizon for impact (short/medium/long term)
        4. Confidence level (1-10)
        5. Key factors to monitor
        
        Consider central bank policies, economic indicators, and geopolitical events.
        """
        
        result = subprocess.run([
            'ollama', 'run', 'llama2', prompt
        ], capture_output=True, text=True)
        
        return result.stdout

# Analyze news sentiment
sentiment_analyzer = NewsSentimentAnalyzer()
recent_news = sentiment_analyzer.fetch_forex_news()
sentiment_impact = sentiment_analyzer.analyze_sentiment_impact(recent_news, 'EUR/USD')
print("News Sentiment Analysis:", sentiment_impact)

import psutil
import time
from datetime import datetime

class SystemMonitor:
    def __init__(self):
        self.start_time = datetime.now()
        self.performance_log = []
    
    def monitor_ollama_performance(self):
        """
        Monitor Ollama system performance
        Returns: Performance metrics
        """
        # Get system metrics
        cpu_usage = psutil.cpu_percent(interval=1)
        memory_usage = psutil.virtual_memory().percent
        
        # Time Ollama response
        start_time = time.time()
        
        test_prompt = "Analyze EUR/USD trend: price 1.0950, RSI 65, trending up"
        result = subprocess.run([
            'ollama', 'run', 'llama2', test_prompt
        ], capture_output=True, text=True)
        
        response_time = time.time() - start_time
        
        metrics = {
            'timestamp': datetime.now(),
            'cpu_usage': cpu_usage,
            'memory_usage': memory_usage,
            'response_time': response_time,
            'response_length': len(result.stdout)
        }
        
        self.performance_log.append(metrics)
        return metrics
    
    def generate_performance_report(self):
        """
        Generate comprehensive performance report
        Returns: Performance analysis
        """
        if not self.performance_log:
            return "No performance data available"
        
        avg_response_time = sum(m['response_time'] for m in self.performance_log) / len(self.performance_log)
        avg_cpu = sum(m['cpu_usage'] for m in self.performance_log) / len(self.performance_log)
        avg_memory = sum(m['memory_usage'] for m in self.performance_log) / len(self.performance_log)
        
        report = f"""
        Ollama Forex Trading System Performance Report
        ============================================
        
        Monitoring Period: {self.start_time} to {datetime.now()}
        Total Analyses: {len(self.performance_log)}
        
        Performance Metrics:
        - Average Response Time: {avg_response_time:.2f} seconds
        - Average CPU Usage: {avg_cpu:.1f}%
        - Average Memory Usage: {avg_memory:.1f}%
        
        Recommendations:
        {'- Consider GPU acceleration for faster processing' if avg_response_time > 5 else '- Performance is optimal'}
        {'- Monitor memory usage closely' if avg_memory > 80 else '- Memory usage is within normal range'}
        """
        
        return report

# Monitor system performance
monitor = SystemMonitor()
performance = monitor.monitor_ollama_performance()
print("Current Performance:", performance)

import logging
from functools import wraps
import time

def retry_on_failure(max_retries=3, delay=1):
    """
    Decorator to retry failed Ollama calls
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    logging.warning(f"Attempt {attempt + 1} failed: {str(e)}")
                    if attempt == max_retries - 1:
                        raise e
                    time.sleep(delay * (attempt + 1))
            return None
        return wrapper
    return decorator

class RobustForexAnalyzer:
    def __init__(self):
        self.fallback_strategies = []
    
    @retry_on_failure(max_retries=3)
    def analyze_with_fallback(self, data, pair):
        """
        Analyze currency pair with fallback strategies
        Returns: Analysis or fallback result
        """
        try:
            # Primary Ollama analysis
            return self.analyzer.analyze_currency_trend(data, pair)
        except Exception as e:
            logging.error(f"Primary analysis failed: {str(e)}")
            return self.execute_fallback_analysis(data, pair)
    
    def execute_fallback_analysis(self, data, pair):
        """
        Execute fallback analysis when Ollama fails
        Returns: Basic technical analysis
        """
        current_price = data['Close'].iloc[-1]
        price_change = ((current_price - data['Close'].iloc[-5]) / data['Close'].iloc[-5]) * 100
        
        if price_change > 2:
            return f"FALLBACK: {pair} showing strong upward momentum (+{price_change:.2f}%) - Consider BUY"
        elif price_change < -2:
            return f"FALLBACK: {pair} showing strong downward momentum ({price_change:.2f}%) - Consider SELL"
        else:
            return f"FALLBACK: {pair} in consolidation phase ({price_change:.2f}%) - HOLD position"

# Initialize robust analyzer
robust_analyzer = RobustForexAnalyzer()

# Dockerfile for Ollama Forex Trading System
FROM python:3.9-slim

# Install system dependencies
RUN apt-get update && apt-get install -y \
    curl \
    && rm -rf /var/lib/apt/lists/*

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

# Set working directory
WORKDIR /app

# Copy requirements and install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . .

# Expose port for API
EXPOSE 8000

# Start Ollama service and application
CMD ["python", "main.py"]

# docker-compose.yml for production deployment
version: '3.8'

services:
  ollama-forex:
    build: .
    container_name: forex-trading-system
    restart: unless-stopped
    ports:
      - "8000:8000"
    volumes:
      - ./data:/app/data
      - ./logs:/app/logs
    environment:
      - ENVIRONMENT=production
      - LOG_LEVEL=INFO
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  redis:
    image: redis:alpine
    container_name: forex-cache
    restart: unless-stopped
    ports:
      - "6379:6379"

  monitoring:
    image: grafana/grafana:latest
    container_name: forex-monitoring
    restart: unless-stopped
    ports:
      - "3000:3000"
    volumes:
      - grafana-storage:/var/lib/grafana

volumes:
  grafana-storage: