Multi-Timeframe Analysis with Ollama: Scalping to Swing Trading Strategies

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

# Pull the required model
ollama pull llama2:13b

# Install Python dependencies
pip install ollama pandas numpy matplotlib yfinance ta-lib

# config.py
OLLAMA_CONFIG = {
    "model": "llama2:13b",
    "temperature": 0.1,  # Low temperature for consistent analysis
    "top_p": 0.9,
    "num_predict": 500,
    "stop": ["Human:", "Assistant:"]
}

TIMEFRAMES = {
    "scalping": ["1m", "5m", "15m"],
    "day_trading": ["15m", "1h", "4h"],
    "swing_trading": ["4h", "1d", "1w"]
}

import yfinance as yf
import pandas as pd
from datetime import datetime, timedelta
import ollama

class MultiTimeframeData:
    def __init__(self, symbol, timeframes):
        self.symbol = symbol
        self.timeframes = timeframes
        self.data = {}
    
    def fetch_data(self, period="1y"):
        """Fetch data for all specified timeframes"""
        ticker = yf.Ticker(self.symbol)
        
        for tf in self.timeframes:
            # Map timeframe to yfinance interval
            interval_map = {
                "1m": "1m", "5m": "5m", "15m": "15m",
                "1h": "1h", "4h": "4h", "1d": "1d", "1w": "1wk"
            }
            
            try:
                self.data[tf] = ticker.history(
                    period=period,
                    interval=interval_map[tf]
                )
                print(f"✓ Fetched {tf} data: {len(self.data[tf])} bars")
            except Exception as e:
                print(f"✗ Failed to fetch {tf} data: {e}")
    
    def get_latest_bars(self, timeframe, count=100):
        """Get the most recent bars for analysis"""
        if timeframe in self.data:
            return self.data[timeframe].tail(count)
        return None

import talib
import numpy as np

class TechnicalAnalyzer:
    def __init__(self):
        self.indicators = {}
    
    def calculate_indicators(self, data):
        """Calculate technical indicators for a timeframe"""
        close = data['Close'].values
        high = data['High'].values
        low = data['Low'].values
        volume = data['Volume'].values
        
        indicators = {
            'sma_20': talib.SMA(close, timeperiod=20),
            'sma_50': talib.SMA(close, timeperiod=50),
            'ema_12': talib.EMA(close, timeperiod=12),
            'ema_26': talib.EMA(close, timeperiod=26),
            'rsi': talib.RSI(close, timeperiod=14),
            'macd': talib.MACD(close)[0],
            'bb_upper': talib.BBANDS(close)[0],
            'bb_lower': talib.BBANDS(close)[2],
            'atr': talib.ATR(high, low, close, timeperiod=14),
            'volume_sma': talib.SMA(volume, timeperiod=20)
        }
        
        return indicators
    
    def generate_signals(self, data, indicators):
        """Generate trading signals based on indicators"""
        signals = []
        current_price = data['Close'].iloc[-1]
        
        # Trend signals
        if current_price > indicators['sma_20'][-1]:
            signals.append("Price above SMA20 - bullish trend")
        
        # Momentum signals
        if indicators['rsi'][-1] > 70:
            signals.append("RSI overbought - potential reversal")
        elif indicators['rsi'][-1] < 30:
            signals.append("RSI oversold - potential bounce")
        
        # MACD signals
        if indicators['macd'][-1] > indicators['macd'][-2]:
            signals.append("MACD turning positive - momentum building")
        
        return signals

class OllamaAnalyzer:
    def __init__(self, config):
        self.config = config
        self.client = ollama.Client()
    
    def analyze_timeframe(self, timeframe, data, indicators, signals):
        """Analyze a single timeframe using Ollama"""
        
        # Prepare context for AI analysis
        context = self.prepare_context(timeframe, data, indicators, signals)
        
        prompt = f"""
        You are an expert trading analyst. Analyze the following {timeframe} timeframe data:

        {context}

        Provide analysis in this format:
        1. TREND: (bullish/bearish/neutral with confidence 1-10)
        2. MOMENTUM: (strong/weak/neutral with direction)
        3. SUPPORT/RESISTANCE: (key levels)
        4. SIGNAL STRENGTH: (1-10 scale)
        5. TRADING BIAS: (buy/sell/wait with reasoning)
        
        Be specific and actionable. Focus on the most important factors.
        """
        
        response = self.client.generate(
            model=self.config["model"],
            prompt=prompt,
            options={
                "temperature": self.config["temperature"],
                "top_p": self.config["top_p"],
                "num_predict": self.config["num_predict"]
            }
        )
        
        return response['response']
    
    def prepare_context(self, timeframe, data, indicators, signals):
        """Prepare market context for AI analysis"""
        latest_bars = data.tail(5)
        
        context = f"""
        TIMEFRAME: {timeframe}
        CURRENT PRICE: ${data['Close'].iloc[-1]:.2f}
        PRICE CHANGE: {((data['Close'].iloc[-1] / data['Close'].iloc[-2]) - 1) * 100:.2f}%
        
        RECENT PRICE ACTION:
        {latest_bars[['Open', 'High', 'Low', 'Close', 'Volume']].to_string()}
        
        TECHNICAL INDICATORS:
        - SMA20: ${indicators['sma_20'][-1]:.2f}
        - SMA50: ${indicators['sma_50'][-1]:.2f}
        - RSI: {indicators['rsi'][-1]:.1f}
        - MACD: {indicators['macd'][-1]:.4f}
        - ATR: {indicators['atr'][-1]:.2f}
        
        SIGNALS:
        {chr(10).join(signals)}
        """
        
        return context

class ScalpingStrategy:
    def __init__(self, symbol, analyzer):
        self.symbol = symbol
        self.analyzer = analyzer
        self.timeframes = ["1m", "5m", "15m"]
        self.data_fetcher = MultiTimeframeData(symbol, self.timeframes)
        self.tech_analyzer = TechnicalAnalyzer()
    
    def run_scalping_analysis(self):
        """Execute scalping-focused multi-timeframe analysis"""
        self.data_fetcher.fetch_data(period="1d")  # 1 day for scalping
        
        analyses = {}
        
        for tf in self.timeframes:
            data = self.data_fetcher.get_latest_bars(tf, 200)
            if data is None:
                continue
            
            indicators = self.tech_analyzer.calculate_indicators(data)
            signals = self.tech_analyzer.generate_signals(data, indicators)
            
            # AI analysis
            analysis = self.analyzer.analyze_timeframe(tf, data, indicators, signals)
            analyses[tf] = analysis
        
        return self.generate_scalping_decision(analyses)
    
    def generate_scalping_decision(self, analyses):
        """Generate scalping decision based on multi-timeframe analysis"""
        
        decision_prompt = f"""
        Based on the following multi-timeframe analysis, provide a scalping decision:
        
        1-MINUTE ANALYSIS:
        {analyses.get('1m', 'No data available')}
        
        5-MINUTE ANALYSIS:
        {analyses.get('5m', 'No data available')}
        
        15-MINUTE ANALYSIS:
        {analyses.get('15m', 'No data available')}
        
        Provide a SCALPING DECISION in this format:
        - ACTION: (BUY/SELL/WAIT)
        - CONFIDENCE: (1-10)
        - ENTRY: (specific price level)
        - STOP LOSS: (specific price level)
        - TAKE PROFIT: (specific price level)
        - REASONING: (why this decision)
        - TIMEFRAME ALIGNMENT: (how timeframes support this decision)
        
        Focus on quick entries and exits with tight risk management.
        """
        
        response = self.analyzer.client.generate(
            model=self.analyzer.config["model"],
            prompt=decision_prompt,
            options={"temperature": 0.1}
        )
        
        return response['response']

def filter_scalping_signals(self, raw_signals):
    """Filter scalping signals to reduce noise"""
    
    filtered_signals = []
    
    for signal in raw_signals:
        # Apply filters
        if self.passes_volume_filter(signal):
            if self.passes_volatility_filter(signal):
                if self.passes_trend_filter(signal):
                    filtered_signals.append(signal)
    
    return filtered_signals

def passes_volume_filter(self, signal):
    """Check if signal meets volume requirements"""
    current_volume = signal['volume']
    avg_volume = signal['volume_sma']
    
    return current_volume > (avg_volume * 1.2)  # 20% above average

def passes_volatility_filter(self, signal):
    """Ensure adequate volatility for scalping"""
    atr = signal['atr']
    price = signal['price']
    
    volatility_pct = (atr / price) * 100
    return volatility_pct > 0.5  # Minimum 0.5% volatility

class SwingTradingStrategy:
    def __init__(self, symbol, analyzer):
        self.symbol = symbol
        self.analyzer = analyzer
        self.timeframes = ["4h", "1d", "1w"]
        self.data_fetcher = MultiTimeframeData(symbol, self.timeframes)
        self.tech_analyzer = TechnicalAnalyzer()
    
    def run_swing_analysis(self):
        """Execute swing trading multi-timeframe analysis"""
        self.data_fetcher.fetch_data(period="2y")  # 2 years for swing trading
        
        analyses = {}
        trend_strength = {}
        
        for tf in self.timeframes:
            data = self.data_fetcher.get_latest_bars(tf, 500)
            if data is None:
                continue
            
            indicators = self.tech_analyzer.calculate_indicators(data)
            signals = self.tech_analyzer.generate_signals(data, indicators)
            
            # Calculate trend strength
            trend_strength[tf] = self.calculate_trend_strength(data, indicators)
            
            # AI analysis
            analysis = self.analyzer.analyze_timeframe(tf, data, indicators, signals)
            analyses[tf] = analysis
        
        return self.generate_swing_decision(analyses, trend_strength)
    
    def calculate_trend_strength(self, data, indicators):
        """Calculate trend strength for swing trading"""
        
        # Multiple moving average alignment
        sma_20 = indicators['sma_20'][-1]
        sma_50 = indicators['sma_50'][-1]
        current_price = data['Close'].iloc[-1]
        
        # Trend alignment score
        alignment_score = 0
        
        if current_price > sma_20 > sma_50:
            alignment_score += 3  # Strong bullish alignment
        elif current_price > sma_20:
            alignment_score += 1  # Weak bullish alignment
        elif current_price < sma_20 < sma_50:
            alignment_score -= 3  # Strong bearish alignment
        elif current_price < sma_20:
            alignment_score -= 1  # Weak bearish alignment
        
        # Volume confirmation
        volume_ratio = data['Volume'].iloc[-5:].mean() / data['Volume'].iloc[-20:].mean()
        if volume_ratio > 1.2:
            alignment_score += 1
        
        return alignment_score
    
    def generate_swing_decision(self, analyses, trend_strength):
        """Generate swing trading decision"""
        
        decision_prompt = f"""
        Based on the following multi-timeframe analysis, provide a swing trading decision:
        
        4-HOUR ANALYSIS:
        {analyses.get('4h', 'No data available')}
        Trend Strength: {trend_strength.get('4h', 0)}
        
        DAILY ANALYSIS:
        {analyses.get('1d', 'No data available')}
        Trend Strength: {trend_strength.get('1d', 0)}
        
        WEEKLY ANALYSIS:
        {analyses.get('1w', 'No data available')}
        Trend Strength: {trend_strength.get('1w', 0)}
        
        Provide a SWING TRADING DECISION in this format:
        - ACTION: (BUY/SELL/WAIT)
        - CONFIDENCE: (1-10)
        - ENTRY ZONE: (price range)
        - STOP LOSS: (specific price level)
        - TAKE PROFIT TARGETS: (multiple levels)
        - REASONING: (detailed explanation)
        - TIMEFRAME HIERARCHY: (how higher timeframes influence decision)
        - RISK-REWARD RATIO: (calculated ratio)
        
        Focus on high-probability setups with favorable risk-reward ratios.
        """
        
        response = self.analyzer.client.generate(
            model=self.analyzer.config["model"],
            prompt=decision_prompt,
            options={"temperature": 0.1}
        )
        
        return response['response']

class PatternRecognizer:
    def __init__(self, analyzer):
        self.analyzer = analyzer
        self.patterns = {
            'reversal': ['double_top', 'double_bottom', 'head_shoulders', 'inverse_head_shoulders'],
            'continuation': ['flag', 'pennant', 'triangle', 'rectangle'],
            'breakout': ['cup_handle', 'ascending_triangle', 'descending_triangle']
        }
    
    def detect_patterns(self, data, timeframe):
        """Detect chart patterns using AI analysis"""
        
        # Prepare price action data
        price_action = self.prepare_price_action(data)
        
        pattern_prompt = f"""
        Analyze the following price action data for chart patterns:
        
        TIMEFRAME: {timeframe}
        PRICE ACTION (last 50 bars):
        {price_action}
        
        Identify any significant chart patterns:
        1. PATTERN TYPE: (reversal/continuation/breakout)
        2. PATTERN NAME: (specific pattern)
        3. COMPLETION STATUS: (forming/complete/broken)
        4. RELIABILITY: (1-10 scale)
        5. PRICE TARGETS: (projected levels)
        6. INVALIDATION LEVEL: (pattern failure point)
        
        Focus on well-formed, high-probability patterns.
        """
        
        response = self.analyzer.client.generate(
            model=self.analyzer.config["model"],
            prompt=pattern_prompt,
            options={"temperature": 0.2}
        )
        
        return response['response']
    
    def prepare_price_action(self, data):
        """Prepare price action summary for pattern analysis"""
        recent_data = data.tail(50)
        
        price_summary = []
        for i, row in recent_data.iterrows():
            price_summary.append(f"Bar {len(price_summary)+1}: O:{row['Open']:.2f} H:{row['High']:.2f} L:{row['Low']:.2f} C:{row['Close']:.2f}")
        
        return "\n".join(price_summary)

class RiskManager:
    def __init__(self, analyzer, account_size=10000):
        self.analyzer = analyzer
        self.account_size = account_size
        self.max_risk_per_trade = 0.02  # 2% maximum risk
    
    def calculate_position_size(self, entry_price, stop_loss, confidence_score):
        """Calculate position size based on risk parameters"""
        
        # Risk amount based on confidence
        risk_multiplier = min(confidence_score / 10, 1.0)
        risk_amount = self.account_size * self.max_risk_per_trade * risk_multiplier
        
        # Position size calculation
        risk_per_share = abs(entry_price - stop_loss)
        position_size = risk_amount / risk_per_share
        
        return {
            'shares': int(position_size),
            'risk_amount': risk_amount,
            'risk_per_share': risk_per_share,
            'position_value': position_size * entry_price
        }
    
    def analyze_risk_reward(self, entry, stop_loss, take_profit):
        """Analyze risk-reward ratio"""
        
        risk = abs(entry - stop_loss)
        reward = abs(take_profit - entry)
        
        if risk == 0:
            return None
        
        ratio = reward / risk
        
        analysis_prompt = f"""
        Analyze this risk-reward setup:
        
        ENTRY: ${entry:.2f}
        STOP LOSS: ${stop_loss:.2f}
        TAKE PROFIT: ${take_profit:.2f}
        
        RISK: ${risk:.2f}
        REWARD: ${reward:.2f}
        RATIO: {ratio:.2f}:1
        
        Provide analysis:
        1. SETUP QUALITY: (excellent/good/poor)
        2. RECOMMENDATION: (take/modify/skip)
        3. IMPROVEMENTS: (suggested modifications)
        4. PROBABILITY ASSESSMENT: (success likelihood)
        
        Focus on risk management best practices.
        """
        
        response = self.analyzer.client.generate(
            model=self.analyzer.config["model"],
            prompt=analysis_prompt,
            options={"temperature": 0.1}
        )
        
        return {
            'ratio': ratio,
            'risk': risk,
            'reward': reward,
            'analysis': response['response']
        }

class MultiTimeframeTrader:
    def __init__(self, symbol, strategy_type="swing"):
        self.symbol = symbol
        self.strategy_type = strategy_type
        
        # Initialize components
        self.analyzer = OllamaAnalyzer(OLLAMA_CONFIG)
        self.risk_manager = RiskManager(self.analyzer)
        self.pattern_recognizer = PatternRecognizer(self.analyzer)
        
        # Select strategy
        if strategy_type == "scalping":
            self.strategy = ScalpingStrategy(symbol, self.analyzer)
        else:
            self.strategy = SwingTradingStrategy(symbol, self.analyzer)
    
    def run_complete_analysis(self):
        """Run complete multi-timeframe analysis"""
        
        print(f"🔍 Starting multi-timeframe analysis for {self.symbol}")
        print(f"📊 Strategy: {self.strategy_type.upper()}")
        print("-" * 50)
        
        # Execute strategy analysis
        if self.strategy_type == "scalping":
            decision = self.strategy.run_scalping_analysis()
        else:
            decision = self.strategy.run_swing_analysis()
        
        # Pattern recognition
        data = self.strategy.data_fetcher.get_latest_bars(
            self.strategy.timeframes[0], 100
        )
        patterns = self.pattern_recognizer.detect_patterns(
            data, self.strategy.timeframes[0]
        )
        
        # Generate final recommendation
        final_recommendation = self.generate_final_recommendation(
            decision, patterns
        )
        
        return {
            'decision': decision,
            'patterns': patterns,
            'recommendation': final_recommendation
        }
    
    def generate_final_recommendation(self, decision, patterns):
        """Generate final trading recommendation"""
        
        recommendation_prompt = f"""
        Based on the comprehensive analysis, provide a final trading recommendation:
        
        STRATEGY DECISION:
        {decision}
        
        PATTERN ANALYSIS:
        {patterns}
        
        Provide FINAL RECOMMENDATION:
        1. OVERALL ASSESSMENT: (bullish/bearish/neutral)
        2. RECOMMENDED ACTION: (specific action to take)
        3. CONFIDENCE LEVEL: (1-10 with reasoning)
        4. KEY FACTORS: (most important considerations)
        5. MONITORING POINTS: (what to watch for changes)
        6. ALTERNATIVE SCENARIOS: (what if analysis is wrong)
        
        Be decisive yet acknowledge uncertainty where appropriate.
        """
        
        response = self.analyzer.client.generate(
            model=self.analyzer.config["model"],
            prompt=recommendation_prompt,
            options={"temperature": 0.1}
        )
        
        return response['response']

# Usage example
if __name__ == "__main__":
    # Initialize trader
    trader = MultiTimeframeTrader("AAPL", "swing")
    
    # Run analysis
    results = trader.run_complete_analysis()
    
    # Display results
    print("📋 TRADING ANALYSIS RESULTS")
    print("=" * 50)
    print("\n🎯 STRATEGY DECISION:")
    print(results['decision'])
    print("\n📈 PATTERN ANALYSIS:")
    print(results['patterns'])
    print("\n💡 FINAL RECOMMENDATION:")
    print(results['recommendation'])

class BacktestEngine:
    def __init__(self, strategy, start_date, end_date):
        self.strategy = strategy
        self.start_date = start_date
        self.end_date = end_date
        self.trades = []
        self.performance_metrics = {}
    
    def run_backtest(self):
        """Execute backtesting simulation"""
        
        print(f"🔄 Running backtest from {self.start_date} to {self.end_date}")
        
        # Historical data simulation would go here
        # This is a simplified version
        
        historical_results = self.simulate_historical_trades()
        
        self.calculate_performance_metrics(historical_results)
        
        return self.performance_metrics
    
    def calculate_performance_metrics(self, trades):
        """Calculate performance statistics"""
        
        if not trades:
            return
        
        total_return = sum(trade['profit'] for trade in trades)
        win_rate = len([t for t in trades if t['profit'] > 0]) / len(trades)
        avg_win = np.mean([t['profit'] for t in trades if t['profit'] > 0])
        avg_loss = np.mean([t['profit'] for t in trades if t['profit'] < 0])
        
        self.performance_metrics = {
            'total_trades': len(trades),
            'total_return': total_return,
            'win_rate': win_rate,
            'avg_win': avg_win,
            'avg_loss': avg_loss,
            'profit_factor': abs(avg_win / avg_loss) if avg_loss != 0 else 0
        }

def validate_data_quality(self, data):
    """Validate data quality before analysis"""
    
    issues = []
    
    # Check for missing data
    if data.isnull().any().any():
        issues.append("Missing data points detected")
    
    # Check for unrealistic price movements
    price_changes = data['Close'].pct_change().abs()
    if (price_changes > 0.2).any():  # 20% single-bar moves
        issues.append("Unrealistic price movements detected")
    
    # Check for zero volume
    if (data['Volume'] == 0).any():
        issues.append("Zero volume bars detected")
    
    return issues

def handle_data_issues(self, data, issues):
    """Handle identified data issues"""
    
    cleaned_data = data.copy()
    
    for issue in issues:
        if "Missing data" in issue:
            cleaned_data = cleaned_data.dropna()
        elif "Zero volume" in issue:
            cleaned_data = cleaned_data[cleaned_data['Volume'] > 0]
    
    return cleaned_data

class OllamaOptimizer:
    def __init__(self):
        self.cache = {}
        self.max_cache_size = 100
    
    def optimize_prompts(self, prompt):
        """Optimize prompts for better performance"""
        
        # Cache frequently used analyses
        prompt_hash = hash(prompt)
        if prompt_hash in self.cache:
            return self.cache[prompt_hash]
        
        # Limit prompt length
        if len(prompt) > 2000:
            prompt = prompt[:2000] + "..."
        
        # Store in cache
        if len(self.cache) >= self.max_cache_size:
            # Remove oldest entry
            oldest_key = next(iter(self.cache))
            del self.cache[oldest_key]
        
        return prompt
    
    def batch_analysis(self, analyses):
        """Batch multiple analyses for efficiency"""
        
        batch_prompt = "Analyze the following timeframes:\n\n"
        
        for i, analysis in enumerate(analyses):
            batch_prompt += f"TIMEFRAME {i+1}:\n{analysis}\n\n"
        
        batch_prompt += "Provide consolidated analysis for all timeframes."
        
        return batch_prompt