Paper Trading Simulator with Ollama: Risk-Free Strategy Testing Environment

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

# Pull a capable model for trading analysis
ollama pull llama3.1:8b

# Verify installation
ollama list

# Create project directory
mkdir ollama-trading-sim
cd ollama-trading-sim

# Set up virtual environment
python -m venv trading-env
source trading-env/bin/activate  # On Windows: trading-env\Scripts\activate

# Install dependencies
pip install ollama yfinance pandas numpy matplotlib flask requests

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

class MarketDataProvider:
    """Handles real-time and historical market data for simulation"""
    
    def __init__(self):
        self.cache = {}
    
    def get_stock_data(self, symbol, period="1d", interval="1m"):
        """Fetch stock data with caching for performance"""
        cache_key = f"{symbol}_{period}_{interval}"
        
        if cache_key in self.cache:
            return self.cache[cache_key]
        
        try:
            ticker = yf.Ticker(symbol)
            data = ticker.history(period=period, interval=interval)
            
            # Add technical indicators
            data['SMA_20'] = data['Close'].rolling(window=20).mean()
            data['SMA_50'] = data['Close'].rolling(window=50).mean()
            data['RSI'] = self._calculate_rsi(data['Close'])
            
            self.cache[cache_key] = data
            return data
            
        except Exception as e:
            print(f"Error fetching data for {symbol}: {e}")
            return pd.DataFrame()
    
    def _calculate_rsi(self, prices, window=14):
        """Calculate Relative Strength Index"""
        delta = prices.diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=window).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=window).mean()
        rs = gain / loss
        return 100 - (100 / (1 + rs))
    
    def get_market_summary(self, symbol):
        """Generate market summary for AI analysis"""
        data = self.get_stock_data(symbol, period="5d", interval="1h")
        
        if data.empty:
            return "No market data available"
        
        latest = data.iloc[-1]
        previous = data.iloc[-2] if len(data) > 1 else latest
        
        summary = {
            "symbol": symbol,
            "current_price": round(latest['Close'], 2),
            "price_change": round(latest['Close'] - previous['Close'], 2),
            "volume": int(latest['Volume']),
            "rsi": round(latest['RSI'], 2) if 'RSI' in latest else None,
            "trend": "bullish" if latest['Close'] > latest['SMA_20'] else "bearish"
        }
        
        return json.dumps(summary, indent=2)

import ollama
import json
import re

class TradingBrain:
    """AI-powered trading decision engine using Ollama"""
    
    def __init__(self, model_name="llama3.1:8b"):
        self.model = model_name
        self.conversation_history = []
    
    def analyze_market_condition(self, market_data, current_portfolio):
        """Analyze market conditions and suggest trading actions"""
        
        prompt = f"""
        You are an experienced trading analyst. Analyze this market data and provide trading recommendations.
        
        MARKET DATA:
        {market_data}
        
        CURRENT PORTFOLIO:
        Cash: ${current_portfolio.get('cash', 0):,.2f}
        Holdings: {json.dumps(current_portfolio.get('holdings', {}), indent=2)}
        
        Provide a JSON response with:
        1. market_sentiment (bullish/bearish/neutral)
        2. recommended_action (buy/sell/hold)
        3. confidence_level (1-10)
        4. reasoning (brief explanation)
        5. risk_level (low/medium/high)
        
        Response must be valid JSON only.
        """
        
        try:
            response = ollama.chat(
                model=self.model,
                messages=[{'role': 'user', 'content': prompt}]
            )
            
            # Extract JSON from response
            content = response['message']['content']
            json_match = re.search(r'\{.*\}', content, re.DOTALL)
            
            if json_match:
                return json.loads(json_match.group())
            else:
                return self._default_analysis()
                
        except Exception as e:
            print(f"AI analysis error: {e}")
            return self._default_analysis()
    
    def generate_trading_strategy(self, market_conditions, risk_tolerance="medium"):
        """Generate a custom trading strategy based on conditions"""
        
        prompt = f"""
        Create a trading strategy for these market conditions:
        
        CONDITIONS: {json.dumps(market_conditions, indent=2)}
        RISK TOLERANCE: {risk_tolerance}
        
        Generate a strategy with:
        1. entry_criteria (when to buy)
        2. exit_criteria (when to sell)
        3. position_sizing (percentage of portfolio)
        4. stop_loss (percentage)
        5. take_profit (percentage)
        
        Provide JSON response only.
        """
        
        try:
            response = ollama.chat(
                model=self.model,
                messages=[{'role': 'user', 'content': prompt}]
            )
            
            content = response['message']['content']
            json_match = re.search(r'\{.*\}', content, re.DOTALL)
            
            if json_match:
                return json.loads(json_match.group())
            else:
                return self._default_strategy()
                
        except Exception as e:
            print(f"Strategy generation error: {e}")
            return self._default_strategy()
    
    def _default_analysis(self):
        """Fallback analysis when AI fails"""
        return {
            "market_sentiment": "neutral",
            "recommended_action": "hold",
            "confidence_level": 5,
            "reasoning": "Unable to analyze market conditions",
            "risk_level": "medium"
        }
    
    def _default_strategy(self):
        """Fallback strategy when AI fails"""
        return {
            "entry_criteria": "RSI below 30",
            "exit_criteria": "RSI above 70",
            "position_sizing": 10,
            "stop_loss": 5,
            "take_profit": 10
        }

from datetime import datetime
import json

class Portfolio:
    """Manages virtual portfolio for paper trading"""
    
    def __init__(self, initial_cash=100000):
        self.cash = initial_cash
        self.initial_cash = initial_cash
        self.holdings = {}  # {symbol: quantity}
        self.trade_history = []
        self.daily_values = []
    
    def buy_stock(self, symbol, quantity, price, timestamp=None):
        """Execute a buy order"""
        if timestamp is None:
            timestamp = datetime.now()
        
        total_cost = quantity * price
        
        if total_cost > self.cash:
            return {
                "success": False,
                "message": f"Insufficient funds. Need ${total_cost:,.2f}, have ${self.cash:,.2f}"
            }
        
        # Execute trade
        self.cash -= total_cost
        self.holdings[symbol] = self.holdings.get(symbol, 0) + quantity
        
        # Record trade
        trade = {
            "type": "BUY",
            "symbol": symbol,
            "quantity": quantity,
            "price": price,
            "total": total_cost,
            "timestamp": timestamp.isoformat(),
            "remaining_cash": self.cash
        }
        
        self.trade_history.append(trade)
        
        return {
            "success": True,
            "trade": trade,
            "message": f"Bought {quantity} shares of {symbol} at ${price:.2f}"
        }
    
    def sell_stock(self, symbol, quantity, price, timestamp=None):
        """Execute a sell order"""
        if timestamp is None:
            timestamp = datetime.now()
        
        current_holding = self.holdings.get(symbol, 0)
        
        if quantity > current_holding:
            return {
                "success": False,
                "message": f"Insufficient shares. Have {current_holding}, trying to sell {quantity}"
            }
        
        # Execute trade
        total_value = quantity * price
        self.cash += total_value
        self.holdings[symbol] -= quantity
        
        # Remove symbol if no shares left
        if self.holdings[symbol] == 0:
            del self.holdings[symbol]
        
        # Record trade
        trade = {
            "type": "SELL",
            "symbol": symbol,
            "quantity": quantity,
            "price": price,
            "total": total_value,
            "timestamp": timestamp.isoformat(),
            "remaining_cash": self.cash
        }
        
        self.trade_history.append(trade)
        
        return {
            "success": True,
            "trade": trade,
            "message": f"Sold {quantity} shares of {symbol} at ${price:.2f}"
        }
    
    def get_portfolio_value(self, market_data_provider):
        """Calculate total portfolio value"""
        total_value = self.cash
        
        for symbol, quantity in self.holdings.items():
            try:
                data = market_data_provider.get_stock_data(symbol, period="1d", interval="1m")
                if not data.empty:
                    current_price = data['Close'].iloc[-1]
                    total_value += quantity * current_price
            except:
                # Skip if can't get current price
                continue
        
        return total_value
    
    def get_performance_metrics(self, market_data_provider):
        """Calculate portfolio performance statistics"""
        current_value = self.get_portfolio_value(market_data_provider)
        total_return = current_value - self.initial_cash
        return_percentage = (total_return / self.initial_cash) * 100
        
        return {
            "initial_value": self.initial_cash,
            "current_value": current_value,
            "total_return": total_return,
            "return_percentage": return_percentage,
            "cash": self.cash,
            "holdings": self.holdings.copy(),
            "total_trades": len(self.trade_history)
        }
    
    def export_data(self):
        """Export portfolio data for analysis"""
        return {
            "cash": self.cash,
            "initial_cash": self.initial_cash,
            "holdings": self.holdings,
            "trade_history": self.trade_history,
            "daily_values": self.daily_values
        }

class AITradingStrategy:
    """Base class for AI-powered trading strategies"""
    
    def __init__(self, name, brain, risk_level="medium"):
        self.name = name
        self.brain = brain
        self.risk_level = risk_level
        self.active_positions = {}
    
    def should_buy(self, symbol, market_data, portfolio):
        """Determine if we should buy a stock"""
        analysis = self.brain.analyze_market_condition(
            market_data, 
            portfolio.get_performance_metrics(None)
        )
        
        return (
            analysis.get("recommended_action") == "buy" and
            analysis.get("confidence_level", 0) >= 6 and
            analysis.get("risk_level") != "high"
        )
    
    def should_sell(self, symbol, market_data, portfolio):
        """Determine if we should sell a stock"""
        analysis = self.brain.analyze_market_condition(
            market_data, 
            portfolio.get_performance_metrics(None)
        )
        
        return (
            analysis.get("recommended_action") == "sell" or
            analysis.get("risk_level") == "high"
        )
    
    def calculate_position_size(self, portfolio_value, risk_per_trade=0.02):
        """Calculate appropriate position size based on risk"""
        return portfolio_value * risk_per_trade

class MomentumStrategy(AITradingStrategy):
    """AI-powered momentum trading strategy"""
    
    def __init__(self, brain):
        super().__init__("AI Momentum", brain, "medium")
    
    def execute(self, symbol, market_data, portfolio, data_provider):
        """Execute momentum strategy logic"""
        current_data = data_provider.get_market_summary(symbol)
        
        # Get AI recommendation
        if self.should_buy(symbol, current_data, portfolio):
            # Calculate position size
            portfolio_value = portfolio.get_portfolio_value(data_provider)
            position_size = self.calculate_position_size(portfolio_value)
            
            # Get current price
            stock_data = data_provider.get_stock_data(symbol, period="1d", interval="1m")
            if not stock_data.empty:
                current_price = stock_data['Close'].iloc[-1]
                quantity = int(position_size / current_price)
                
                if quantity > 0:
                    return portfolio.buy_stock(symbol, quantity, current_price)
        
        elif symbol in portfolio.holdings and self.should_sell(symbol, current_data, portfolio):
            # Sell current position
            stock_data = data_provider.get_stock_data(symbol, period="1d", interval="1m")
            if not stock_data.empty:
                current_price = stock_data['Close'].iloc[-1]
                quantity = portfolio.holdings[symbol]
                
                return portfolio.sell_stock(symbol, quantity, current_price)
        
        return {"success": False, "message": "No action taken"}

from market_data import MarketDataProvider
from ollama_brain import TradingBrain
from portfolio import Portfolio
from strategies.example_strategies import MomentumStrategy
import time
import json

class TradingSimulator:
    """Main trading simulator orchestrating all components"""
    
    def __init__(self, initial_cash=100000):
        self.data_provider = MarketDataProvider()
        self.brain = TradingBrain()
        self.portfolio = Portfolio(initial_cash)
        self.strategies = {}
        self.simulation_log = []
        
        # Add default strategies
        self.strategies["momentum"] = MomentumStrategy(self.brain)
    
    def add_strategy(self, name, strategy):
        """Add a custom trading strategy"""
        self.strategies[name] = strategy
    
    def simulate_trading_day(self, symbols, strategy_name="momentum"):
        """Simulate one trading day for given symbols"""
        if strategy_name not in self.strategies:
            raise ValueError(f"Strategy '{strategy_name}' not found")
        
        strategy = self.strategies[strategy_name]
        day_log = {
            "timestamp": time.time(),
            "trades": [],
            "portfolio_value": self.portfolio.get_portfolio_value(self.data_provider)
        }
        
        print(f"\n🤖 Starting simulation with {strategy_name} strategy...")
        print(f"💰 Portfolio value: ${day_log['portfolio_value']:,.2f}")
        
        for symbol in symbols:
            print(f"\n📊 Analyzing {symbol}...")
            
            try:
                # Execute strategy
                result = strategy.execute(
                    symbol, 
                    self.data_provider.get_market_summary(symbol),
                    self.portfolio, 
                    self.data_provider
                )
                
                if result.get("success"):
                    print(f"✅ {result['message']}")
                    day_log["trades"].append(result["trade"])
                else:
                    print(f"⏸️  {result['message']}")
                    
            except Exception as e:
                print(f"❌ Error processing {symbol}: {e}")
        
        # Log day results
        day_log["final_portfolio_value"] = self.portfolio.get_portfolio_value(self.data_provider)
        self.simulation_log.append(day_log)
        
        return day_log
    
    def run_backtest(self, symbols, days=5, strategy_name="momentum"):
        """Run multi-day backtesting simulation"""
        print(f"\n🚀 Starting {days}-day backtest simulation...")
        print(f"📈 Symbols: {', '.join(symbols)}")
        
        for day in range(days):
            print(f"\n📅 Day {day + 1}/{days}")
            self.simulate_trading_day(symbols, strategy_name)
            
            # Simulate day passing (in real implementation, wait for market hours)
            time.sleep(1)
        
        # Generate final report
        return self.generate_performance_report()
    
    def generate_performance_report(self):
        """Generate comprehensive performance report"""
        metrics = self.portfolio.get_performance_metrics(self.data_provider)
        
        report = {
            "simulation_summary": {
                "initial_value": metrics["initial_value"],
                "final_value": metrics["current_value"],
                "total_return": metrics["total_return"],
                "return_percentage": metrics["return_percentage"],
                "total_trades": metrics["total_trades"]
            },
            "current_holdings": metrics["holdings"],
            "cash_remaining": metrics["cash"],
            "trade_history": self.portfolio.trade_history[-10:],  # Last 10 trades
            "daily_performance": self.simulation_log
        }
        
        # Pretty print results
        print("\n" + "="*50)
        print("📊 SIMULATION RESULTS")
        print("="*50)
        print(f"Initial Portfolio Value: ${metrics['initial_value']:,.2f}")
        print(f"Final Portfolio Value:   ${metrics['current_value']:,.2f}")
        print(f"Total Return:           ${metrics['total_return']:,.2f}")
        print(f"Return Percentage:      {metrics['return_percentage']:.2f}%")
        print(f"Total Trades:           {metrics['total_trades']}")
        print(f"Cash Remaining:         ${metrics['cash']:,.2f}")
        
        if metrics["holdings"]:
            print("\n📋 Current Holdings:")
            for symbol, quantity in metrics["holdings"].items():
                print(f"  {symbol}: {quantity} shares")
        
        return report

# Example usage and testing
if __name__ == "__main__":
    # Initialize simulator
    sim = TradingSimulator(initial_cash=50000)
    
    # Test symbols (major tech stocks)
    test_symbols = ["AAPL", "GOOGL", "MSFT"]
    
    # Run simulation
    results = sim.run_backtest(test_symbols, days=3)
    
    # Save results
    with open("simulation_results.json", "w") as f:
        json.dump(results, f, indent=2, default=str)

# test_simulation.py
from src.simulator import TradingSimulator

def main():
    # Create simulator with $25,000 virtual cash
    simulator = TradingSimulator(initial_cash=25000)
    
    # Define test portfolio
    symbols = ["AAPL", "TSLA", "NVDA"]
    
    print("🎯 Testing Paper Trading Simulator with Ollama")
    print(f"💵 Starting with ${simulator.portfolio.cash:,.2f}")
    
    # Run single day simulation
    results = simulator.simulate_trading_day(symbols)
    
    # Display results
    print(f"\n📈 Day complete! Portfolio value: ${results['final_portfolio_value']:,.2f}")
    
    if results['trades']:
        print(f"🔄 Executed {len(results['trades'])} trades:")
        for trade in results['trades']:
            print(f"  {trade['type']}: {trade['quantity']} {trade['symbol']} @ ${trade['price']:.2f}")
    else:
        print("⏸️  No trades executed today")

if __name__ == "__main__":
    main()

$ python test_simulation.py

🎯 Testing Paper Trading Simulator with Ollama
💵 Starting with $25,000

🤖 Starting simulation with momentum strategy...
💰 Portfolio value: $25,000.00

📊 Analyzing AAPL...
✅ Bought 14 shares of AAPL at $178.32

📊 Analyzing TSLA...
⏸️  No action taken

📊 Analyzing NVDA...
✅ Bought 2 shares of NVDA at $456.78

📈 Day complete! Portfolio value: $25,003.45
🔄 Executed 2 trades:
  BUY: 14 AAPL @ $178.32
  BUY: 2 NVDA @ $456.78

class NewsBasedStrategy(AITradingStrategy):
    """Strategy that incorporates news sentiment"""
    
    def __init__(self, brain):
        super().__init__("News Sentiment", brain, "medium")
    
    def analyze_news_sentiment(self, symbol):
        """Get AI analysis of recent news impact"""
        prompt = f"""
        Analyze recent market news for {symbol} and provide sentiment score.
        
        Consider:
        - Earnings reports
        - Product announcements  
        - Industry trends
        - Economic indicators
        
        Provide JSON with:
        - sentiment_score (-1 to 1)
        - key_factors (list of important news)
        - market_impact (positive/negative/neutral)
        """
        
        try:
            response = ollama.chat(
                model=self.brain.model,
                messages=[{'role': 'user', 'content': prompt}]
            )
            
            # Parse JSON response
            import re
            content = response['message']['content']
            json_match = re.search(r'\{.*\}', content, re.DOTALL)
            
            if json_match:
                return json.loads(json_match.group())
            
        except Exception as e:
            print(f"News analysis error: {e}")
        
        return {"sentiment_score": 0, "market_impact": "neutral"}

class RiskManager:
    """Advanced risk management for trading strategies"""
    
    def __init__(self, max_position_size=0.1, max_daily_loss=0.05):
        self.max_position_size = max_position_size  # 10% max per position
        self.max_daily_loss = max_daily_loss        # 5% max daily loss
        self.daily_start_value = None
    
    def validate_trade(self, trade_type, quantity, price, portfolio, data_provider):
        """Validate trade against risk parameters"""
        current_value = portfolio.get_portfolio_value(data_provider)
        
        if self.daily_start_value is None:
            self.daily_start_value = current_value
        
        # Check daily loss limit
        daily_loss = (self.daily_start_value - current_value) / self.daily_start_value
        if daily_loss >= self.max_daily_loss:
            return False, "Daily loss limit exceeded"
        
        # Check position size limit
        if trade_type == "BUY":
            trade_value = quantity * price
            position_percentage = trade_value / current_value
            
            if position_percentage > self.max_position_size:
                return False, f"Position size too large: {position_percentage:.1%}"
        
        return True, "Trade approved"
    
    def reset_daily_limits(self, portfolio, data_provider):
        """Reset daily tracking (call at market open)"""
        self.daily_start_value = portfolio.get_portfolio_value(data_provider)

import matplotlib.pyplot as plt
import pandas as pd

class PerformanceAnalyzer:
    """Generate comprehensive performance analytics"""
    
    def __init__(self, portfolio):
        self.portfolio = portfolio
    
    def plot_portfolio_performance(self, save_path="portfolio_performance.png"):
        """Create portfolio performance chart"""
        if not self.portfolio.daily_values:
            print("No daily values to plot")
            return
        
        # Create DataFrame from daily values
        df = pd.DataFrame(self.portfolio.daily_values)
        df['date'] = pd.to_datetime(df['timestamp'], unit='s')
        
        # Plot performance
        plt.figure(figsize=(12, 6))
        plt.plot(df['date'], df['portfolio_value'], linewidth=2, color='#2E86C1')
        plt.axhline(y=self.portfolio.initial_cash, color='red', linestyle='--', alpha=0.7, label='Initial Value')
        
        plt.title('Paper Trading Portfolio Performance', fontsize=16, fontweight='bold')
        plt.xlabel('Date', fontsize=12)
        plt.ylabel('Portfolio Value ($)', fontsize=12)
        plt.legend()
        plt.grid(True, alpha=0.3)
        plt.tight_layout()
        
        # Save chart
        plt.savefig(save_path, dpi=300, bbox_inches='tight')
        print(f"Performance chart saved to {save_path}")
        plt.show()
    
    def generate_trade_analysis(self):
        """Analyze trading patterns and success rates"""
        trades = pd.DataFrame(self.portfolio.trade_history)
        
        if trades.empty:
            return {"message": "No trades to analyze"}
        
        analysis = {
            "total_trades": len(trades),
            "buy_trades": len(trades[trades['type'] == 'BUY']),
            "sell_trades": len(trades[trades['type'] == 'SELL']),
            "most_traded_symbol": trades['symbol'].mode().iloc[0] if not trades.empty else None,
            "average_trade_size": trades['total'].mean(),
            "largest_trade": trades['total'].max(),
            "smallest_trade": trades['total'].min()
        }
        
        return analysis
    
    def calculate_sharpe_ratio(self, risk_free_rate=0.02):
        """Calculate Sharpe ratio for strategy performance"""
        if len(self.portfolio.daily_values) < 2:
            return 0
        
        values = [day['portfolio_value'] for day in self.portfolio.daily_values]
        returns = pd.Series(values).pct_change().dropna()
        
        if returns.std() == 0:
            return 0
        
        excess_return = returns.mean() - (risk_free_rate / 252)  # Daily risk-free rate
        sharpe_ratio = excess_return / returns.std() * (252 ** 0.5)  # Annualized
        
        return sharpe_ratio

import os
from dataclasses import dataclass

@dataclass
class TradingConfig:
    """Configuration settings for production deployment"""
    
    # Ollama settings
    ollama_model: str = "llama3.1:8b"
    ollama_host: str = "localhost:11434"
    
    # Trading parameters
    initial_cash: float = 100000
    max_position_size: float = 0.1
    max_daily_loss: float = 0.05
    
    # Data sources
    market_data_provider: str = "yfinance"
    update_interval_minutes: int = 1
    
    # Risk management
    enable_risk_management: bool = True
    enable_stop_losses: bool = True
    
    # Logging
    log_level: str = "INFO"
    log_file: str = "trading_sim.log"
    
    @classmethod
    def from_env(cls):
        """Load configuration from environment variables"""
        return cls(
            ollama_model=os.getenv("OLLAMA_MODEL", "llama3.1:8b"),
            initial_cash=float(os.getenv("INITIAL_CASH", 100000)),
            max_position_size=float(os.getenv("MAX_POSITION_SIZE", 0.1)),
            # ... other environment variables
        )

# Usage
config = TradingConfig.from_env()

import logging
from datetime import datetime

class TradingMonitor:
    """Monitor trading simulation for issues and opportunities"""
    
    def __init__(self, config):
        self.config = config
        self.setup_logging()
        self.alerts = []
    
    def setup_logging(self):
        """Configure logging for production monitoring"""
        logging.basicConfig(
            level=getattr(logging, self.config.log_level),
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(self.config.log_file),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    
    def check_performance_alerts(self, portfolio, data_provider):
        """Monitor for performance-based alerts"""
        metrics = portfolio.get_performance_metrics(data_provider)
        
        # Alert on significant losses
        if metrics["return_percentage"] < -10:
            alert = {
                "type": "PERFORMANCE_WARNING",
                "message": f"Portfolio down {metrics['return_percentage']:.1f}%",
                "timestamp": datetime.now().isoformat(),
                "severity": "HIGH"
            }
            self.alerts.append(alert)
            self.logger.warning(alert["message"])
        
        # Alert on system errors
        if len(portfolio.trade_history) == 0:
            alert = {
                "type": "SYSTEM_WARNING", 
                "message": "No trades executed recently",
                "timestamp": datetime.now().isoformat(),
                "severity": "MEDIUM"
            }
            self.alerts.append(alert)
            self.logger.warning(alert["message"])