Fenix Trading Bot Alternative: Building Visual Chart Analysis with Ollama LLaVA

# System requirements
- 16GB RAM minimum (32GB recommended)
- NVIDIA GPU with 8GB VRAM (optional but recommended)
- Python 3.9 or higher
- Docker (for Ollama deployment)

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

# Pull the LLaVA model for visual analysis
ollama pull llava:13b

# Verify installation
ollama list

# requirements.txt
ollama-python==0.1.7
pillow==10.0.0
pandas==2.0.3
numpy==1.24.3
matplotlib==3.7.2
yfinance==0.2.18
fastapi==0.100.1
uvicorn==0.23.2
python-multipart==0.0.6

pip install -r requirements.txt

# chart_generator.py
import matplotlib.pyplot as plt
import pandas as pd
import yfinance as yf
from datetime import datetime, timedelta
import io
import base64

class ChartGenerator:
    def __init__(self):
        self.style_config = {
            'figure.figsize': (12, 8),
            'axes.grid': True,
            'grid.alpha': 0.3
        }
        plt.rcParams.update(self.style_config)
    
    def create_candlestick_chart(self, symbol, period='1mo'):
        """Generate candlestick chart for visual analysis"""
        # Fetch market data
        ticker = yf.Ticker(symbol)
        data = ticker.history(period=period)
        
        if data.empty:
            raise ValueError(f"No data found for symbol {symbol}")
        
        # Create figure and axis
        fig, ax = plt.subplots(figsize=(12, 8))
        
        # Plot candlestick data
        for i, (date, row) in enumerate(data.iterrows()):
            color = 'green' if row['Close'] > row['Open'] else 'red'
            
            # Draw the candlestick
            ax.plot([i, i], [row['Low'], row['High']], color='black', linewidth=1)
            ax.plot([i, i], [row['Open'], row['Close']], color=color, linewidth=4)
        
        # Add moving averages for context
        data['MA20'] = data['Close'].rolling(window=20).mean()
        data['MA50'] = data['Close'].rolling(window=50).mean()
        
        ax.plot(range(len(data)), data['MA20'], label='MA20', alpha=0.7)
        ax.plot(range(len(data)), data['MA50'], label='MA50', alpha=0.7)
        
        # Formatting
        ax.set_title(f'{symbol} - {period} Chart Analysis', fontsize=16)
        ax.set_xlabel('Time Period')
        ax.set_ylabel('Price ($)')
        ax.legend()
        ax.grid(True, alpha=0.3)
        
        # Save to bytes for LLaVA processing
        img_buffer = io.BytesIO()
        plt.savefig(img_buffer, format='png', dpi=150, bbox_inches='tight')
        img_buffer.seek(0)
        
        plt.close(fig)  # Free memory
        
        return img_buffer.getvalue()
    
    def save_chart_image(self, image_data, filename):
        """Save chart image to file"""
        with open(filename, 'wb') as f:
            f.write(image_data)
        return filename

# visual_analyzer.py
import ollama
import base64
from typing import Dict, List, Optional

class VisualChartAnalyzer:
    def __init__(self, model_name='llava:13b'):
        self.model_name = model_name
        self.client = ollama.Client()
        
    def analyze_chart(self, image_path: str, analysis_prompt: str = None) -> Dict:
        """Analyze trading chart using LLaVA vision model"""
        
        if analysis_prompt is None:
            analysis_prompt = self._get_default_trading_prompt()
        
        try:
            # Read and encode image
            with open(image_path, 'rb') as image_file:
                image_data = image_file.read()
            
            # Send to LLaVA for analysis
            response = self.client.generate(
                model=self.model_name,
                prompt=analysis_prompt,
                images=[image_data],
                stream=False
            )
            
            # Parse the response
            analysis_result = self._parse_analysis_response(response['response'])
            
            return {
                'success': True,
                'analysis': analysis_result,
                'confidence': self._calculate_confidence(response['response']),
                'timestamp': datetime.now().isoformat()
            }
            
        except Exception as e:
            return {
                'success': False,
                'error': str(e),
                'timestamp': datetime.now().isoformat()
            }
    
    def _get_default_trading_prompt(self) -> str:
        """Default prompt for trading chart analysis"""
        return """
        Analyze this trading chart and provide a detailed assessment. Focus on:
        
        1. **Trend Direction**: Is the overall trend bullish, bearish, or sideways?
        2. **Support and Resistance**: Identify key price levels
        3. **Pattern Recognition**: Look for triangles, flags, head and shoulders, etc.
        4. **Volume Analysis**: Comment on volume patterns if visible
        5. **Moving Average Signals**: Analyze MA crossovers and price interactions
        6. **Risk Assessment**: Evaluate potential entry and exit points
        
        Provide your analysis in this format:
        - Trend: [Direction and strength]
        - Key Levels: [Support/Resistance prices]
        - Pattern: [Any recognizable patterns]
        - Signal: [Buy/Sell/Hold recommendation]
        - Confidence: [High/Medium/Low]
        - Risk: [Risk level assessment]
        
        Be specific with price levels and reasoning.
        """
    
    def _parse_analysis_response(self, response_text: str) -> Dict:
        """Parse LLaVA response into structured data"""
        analysis = {
            'trend': 'Unknown',
            'key_levels': [],
            'pattern': 'None detected',
            'signal': 'Hold',
            'confidence': 'Medium',
            'risk': 'Medium',
            'raw_analysis': response_text
        }
        
        # Simple parsing logic (enhance based on your needs)
        lines = response_text.lower().split('\n')
        
        for line in lines:
            if 'trend:' in line:
                analysis['trend'] = line.split('trend:')[1].strip()
            elif 'signal:' in line:
                analysis['signal'] = line.split('signal:')[1].strip()
            elif 'confidence:' in line:
                analysis['confidence'] = line.split('confidence:')[1].strip()
            elif 'risk:' in line:
                analysis['risk'] = line.split('risk:')[1].strip()
        
        return analysis
    
    def _calculate_confidence(self, response_text: str) -> float:
        """Calculate confidence score based on response certainty"""
        certainty_indicators = ['clearly', 'definitely', 'strong', 'obvious']
        uncertainty_indicators = ['might', 'possibly', 'unclear', 'uncertain']
        
        text_lower = response_text.lower()
        certainty_count = sum(1 for word in certainty_indicators if word in text_lower)
        uncertainty_count = sum(1 for word in uncertainty_indicators if word in text_lower)
        
        # Simple confidence calculation
        base_confidence = 0.5
        confidence_adjust = (certainty_count - uncertainty_count) * 0.1
        
        return max(0.1, min(0.9, base_confidence + confidence_adjust))

# trading_engine.py
from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime
import json

@dataclass
class TradingSignal:
    symbol: str
    action: str  # 'buy', 'sell', 'hold'
    confidence: float
    price: float
    reasoning: str
    timestamp: datetime
    risk_level: str

class TradingDecisionEngine:
    def __init__(self, risk_tolerance='medium'):
        self.risk_tolerance = risk_tolerance
        self.signal_history = []
        self.active_positions = {}
        
    def process_visual_analysis(self, symbol: str, analysis_result: Dict) -> TradingSignal:
        """Convert visual analysis into trading signal"""
        
        if not analysis_result.get('success'):
            return self._create_hold_signal(symbol, "Analysis failed")
        
        analysis = analysis_result['analysis']
        
        # Extract trading signals from visual analysis
        signal_action = self._determine_action(analysis)
        confidence = self._adjust_confidence_for_risk(
            analysis_result.get('confidence', 0.5)
        )
        
        # Create trading signal
        trading_signal = TradingSignal(
            symbol=symbol,
            action=signal_action,
            confidence=confidence,
            price=self._get_current_price(symbol),
            reasoning=analysis.get('raw_analysis', ''),
            timestamp=datetime.now(),
            risk_level=analysis.get('risk', 'medium')
        )
        
        self.signal_history.append(trading_signal)
        return trading_signal
    
    def _determine_action(self, analysis: Dict) -> str:
        """Determine buy/sell/hold based on analysis"""
        trend = analysis.get('trend', '').lower()
        signal = analysis.get('signal', '').lower()
        confidence = analysis.get('confidence', '').lower()
        
        # Conservative approach - require multiple confirmations
        if 'buy' in signal and 'bullish' in trend and confidence in ['high', 'strong']:
            return 'buy'
        elif 'sell' in signal and 'bearish' in trend and confidence in ['high', 'strong']:
            return 'sell'
        else:
            return 'hold'
    
    def _adjust_confidence_for_risk(self, base_confidence: float) -> float:
        """Adjust confidence based on risk tolerance"""
        risk_multipliers = {
            'low': 0.7,      # More conservative
            'medium': 1.0,    # No adjustment
            'high': 1.3       # More aggressive
        }
        
        multiplier = risk_multipliers.get(self.risk_tolerance, 1.0)
        return min(0.95, base_confidence * multiplier)
    
    def _get_current_price(self, symbol: str) -> float:
        """Get current market price (implement with your data provider)"""
        try:
            ticker = yf.Ticker(symbol)
            data = ticker.history(period='1d', interval='1m')
            return float(data['Close'].iloc[-1])
        except:
            return 0.0
    
    def _create_hold_signal(self, symbol: str, reason: str) -> TradingSignal:
        """Create a hold signal with explanation"""
        return TradingSignal(
            symbol=symbol,
            action='hold',
            confidence=0.0,
            price=self._get_current_price(symbol),
            reasoning=reason,
            timestamp=datetime.now(),
            risk_level='unknown'
        )

# fenix_alternative_bot.py
import asyncio
import schedule
import time
from typing import List, Dict
import logging

class FenixAlternativeBot:
    def __init__(self, symbols: List[str], analysis_interval=300):
        self.symbols = symbols
        self.analysis_interval = analysis_interval  # seconds
        
        # Initialize components
        self.chart_generator = ChartGenerator()
        self.visual_analyzer = VisualChartAnalyzer()
        self.trading_engine = TradingDecisionEngine()
        
        # Setup logging
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
        
    async def run_analysis_cycle(self):
        """Run complete analysis cycle for all symbols"""
        self.logger.info(f"Starting analysis cycle for {len(self.symbols)} symbols")
        
        for symbol in self.symbols:
            try:
                await self._analyze_symbol(symbol)
                await asyncio.sleep(10)  # Rate limiting between symbols
            except Exception as e:
                self.logger.error(f"Error analyzing {symbol}: {e}")
    
    async def _analyze_symbol(self, symbol: str):
        """Analyze single symbol and generate trading signal"""
        self.logger.info(f"Analyzing {symbol}")
        
        # Generate chart
        chart_data = self.chart_generator.create_candlestick_chart(symbol)
        chart_filename = f"charts/{symbol}_analysis.png"
        self.chart_generator.save_chart_image(chart_data, chart_filename)
        
        # Analyze with LLaVA
        analysis_result = self.visual_analyzer.analyze_chart(chart_filename)
        
        # Generate trading signal
        signal = self.trading_engine.process_visual_analysis(symbol, analysis_result)
        
        # Log results
        self.logger.info(f"{symbol} Signal: {signal.action} (Confidence: {signal.confidence:.2f})")
        
        # Execute trading logic (implement based on your broker API)
        await self._execute_signal(signal)
    
    async def _execute_signal(self, signal: TradingSignal):
        """Execute trading signal (placeholder for broker integration)"""
        if signal.action == 'hold' or signal.confidence < 0.6:
            self.logger.info(f"Holding position for {signal.symbol}")
            return
        
        # Implement your broker API calls here
        self.logger.info(f"Would execute {signal.action} for {signal.symbol} at ${signal.price}")
        
        # Example pseudo-code for broker integration:
        # if signal.action == 'buy':
        #     broker_api.place_order(signal.symbol, 'market_buy', quantity)
        # elif signal.action == 'sell':
        #     broker_api.place_order(signal.symbol, 'market_sell', quantity)
    
    def start_scheduled_analysis(self):
        """Start scheduled analysis with specified interval"""
        schedule.every(self.analysis_interval).seconds.do(
            lambda: asyncio.run(self.run_analysis_cycle())
        )
        
        self.logger.info(f"Started scheduled analysis every {self.analysis_interval} seconds")
        
        while True:
            schedule.run_pending()
            time.sleep(60)  # Check every minute

# Usage example
if __name__ == "__main__":
    # Define symbols to analyze
    trading_symbols = ['AAPL', 'TSLA', 'NVDA', 'SPY']
    
    # Create bot instance
    bot = FenixAlternativeBot(
        symbols=trading_symbols,
        analysis_interval=900  # 15 minutes
    )
    
    # Start the bot
    bot.start_scheduled_analysis()

# Set environment variables for GPU usage
export OLLAMA_GPU=1
export CUDA_VISIBLE_DEVICES=0

# Restart Ollama with GPU support
ollama serve

# Add to ChartGenerator class
class ChartGenerator:
    def __init__(self):
        self.image_cache = {}
        self.cache_max_size = 50
    
    def get_cached_chart(self, symbol, period):
        cache_key = f"{symbol}_{period}"
        if cache_key in self.image_cache:
            return self.image_cache[cache_key]
        return None
    
    def cache_chart(self, symbol, period, image_data):
        if len(self.image_cache) >= self.cache_max_size:
            # Remove oldest entry
            oldest_key = next(iter(self.image_cache))
            del self.image_cache[oldest_key]
        
        cache_key = f"{symbol}_{period}"
        self.image_cache[cache_key] = image_data

class MultiTimeframeAnalyzer:
    def __init__(self):
        self.timeframes = ['1d', '1wk', '1mo', '3mo']
        self.visual_analyzer = VisualChartAnalyzer()
        self.chart_generator = ChartGenerator()
    
    def analyze_multiple_timeframes(self, symbol: str) -> Dict:
        """Analyze symbol across multiple timeframes"""
        results = {}
        
        for timeframe in self.timeframes:
            try:
                # Generate chart for timeframe
                chart_data = self.chart_generator.create_candlestick_chart(symbol, timeframe)
                chart_file = f"charts/{symbol}_{timeframe}.png"
                self.chart_generator.save_chart_image(chart_data, chart_file)
                
                # Analyze chart
                analysis = self.visual_analyzer.analyze_chart(chart_file)
                results[timeframe] = analysis
                
            except Exception as e:
                results[timeframe] = {'error': str(e)}
        
        return self._synthesize_timeframe_results(results)
    
    def _synthesize_timeframe_results(self, results: Dict) -> Dict:
        """Combine multiple timeframe analyses into single decision"""
        signals = []
        confidences = []
        
        for timeframe, result in results.items():
            if result.get('success'):
                analysis = result['analysis']
                signals.append(analysis.get('signal', 'hold'))
                confidences.append(result.get('confidence', 0.5))
        
        # Majority vote for final signal
        signal_counts = {}
        for signal in signals:
            signal_counts[signal] = signal_counts.get(signal, 0) + 1
        
        final_signal = max(signal_counts, key=signal_counts.get) if signal_counts else 'hold'
        avg_confidence = sum(confidences) / len(confidences) if confidences else 0.5
        
        return {
            'final_signal': final_signal,
            'confidence': avg_confidence,
            'timeframe_results': results,
            'agreement_strength': max(signal_counts.values()) / len(signals) if signals else 0
        }

# Dockerfile
FROM python:3.11-slim

# Install system dependencies
RUN apt-get update && apt-get install -y \
    curl \
    git \
    && 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 . .

# Create charts directory
RUN mkdir -p charts

# Expose port for API (if implementing web interface)
EXPOSE 8000

# Start script
COPY start.sh .
RUN chmod +x start.sh

CMD ["./start.sh"]

# start.sh
#!/bin/bash

# Start Ollama service
ollama serve &

# Wait for Ollama to start
sleep 10

# Pull required model
ollama pull llava:13b

# Start the trading bot
python fenix_alternative_bot.py

# dashboard.py
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
import uvicorn

app = FastAPI(title="Fenix Alternative Trading Bot Dashboard")
templates = Jinja2Templates(directory="templates")
app.mount("/static", StaticFiles(directory="static"), name="static")

@app.get("/")
async def dashboard(request: Request):
    # Get recent signals from trading engine
    recent_signals = bot.trading_engine.signal_history[-10:]
    
    return templates.TemplateResponse("dashboard.html", {
        "request": request,
        "signals": recent_signals,
        "active_symbols": bot.symbols
    })

@app.get("/api/signals/{symbol}")
async def get_symbol_signals(symbol: str):
    signals = [s for s in bot.trading_engine.signal_history if s.symbol == symbol]
    return {"symbol": symbol, "signals": signals[-5:]}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

# risk_manager.py
class RiskManager:
    def __init__(self, max_daily_loss=1000, max_position_size=0.1):
        self.max_daily_loss = max_daily_loss
        self.max_position_size = max_position_size
        self.daily_pnl = 0.0
        self.position_sizes = {}
    
    def validate_signal(self, signal: TradingSignal) -> bool:
        """Validate trading signal against risk parameters"""
        
        # Check daily loss limit
        if self.daily_pnl <= -self.max_daily_loss:
            self.logger.warning("Daily loss limit reached")
            return False
        
        # Check confidence threshold
        if signal.confidence < 0.7:
            return False
        
        # Check position size limits
        current_exposure = self.position_sizes.get(signal.symbol, 0)
        if current_exposure >= self.max_position_size:
            return False
        
        return True
    
    def update_pnl(self, symbol: str, pnl: float):
        """Update P&L tracking"""
        self.daily_pnl += pnl
        
    def reset_daily_metrics(self):
        """Reset daily tracking (call at market open)"""
        self.daily_pnl = 0.0