Create app.py

app.py

@@ -0,0 +1,452 @@
+import gradio as gr
+import numpy as np
+import pandas as pd
+import matplotlib.pyplot as plt
+import torch
+import io
+import base64
+from PIL import Image
+import plotly.graph_objects as go
+from plotly.subplots import make_subplots
+import time
+import sys
+import os
+
+# Add src to path
+sys.path.append('src')
+
+from src.environments.visual_trading_env import VisualTradingEnvironment
+from src.agents.visual_agent import VisualTradingAgent
+from src.visualizers.chart_renderer import ChartRenderer
+from src.utils.data_loader import DataLoader
+from src.utils.config import TradingConfig
+
+class TradingAIDemo:
+    def __init__(self):
+        self.config = TradingConfig()
+        self.env = None
+        self.agent = None
+        self.current_state = None
+        self.is_training = False
+        self.episode_history = []
+        self.chart_renderer = ChartRenderer()
+        
+    def initialize_environment(self, initial_balance, risk_level, asset_type):
+        """Initialize trading environment"""
+        try:
+            self.env = VisualTradingEnvironment(
+                initial_balance=initial_balance,
+                risk_level=risk_level,
+                asset_type=asset_type
+            )
+            self.agent = VisualTradingAgent(
+                state_dim=self.env.observation_space.shape[0],
+                action_dim=self.env.action_space.n
+            )
+            self.current_state = self.env.reset()
+            return "✅ Environment initialized successfully!"
+        except Exception as e:
+            return f"❌ Error initializing environment: {str(e)}"
+    
+    def run_single_step(self, action_choice):
+        """Run a single step in the environment"""
+        if self.env is None or self.agent is None:
+            return None, "Please initialize environment first!"
+        
+        try:
+            # Use selected action or let agent decide
+            if action_choice == "AI Decision":
+                action = self.agent.select_action(self.current_state)
+            else:
+                action_mapping = {"Buy": 1, "Sell": 2, "Hold": 0, "Close": 3}
+                action = action_mapping[action_choice]
+            
+            # Execute action
+            next_state, reward, done, info = self.env.step(action)
+            self.current_state = next_state
+            
+            # Create visualization
+            fig = self.create_visualization(info, action, reward)
+            
+            # Update history
+            self.episode_history.append({
+                'step': len(self.episode_history),
+                'action': action,
+                'reward': reward,
+                'net_worth': info['net_worth'],
+                'balance': info['balance'],
+                'position': info['position_size']
+            })
+            
+            status = f"Action: {['Hold', 'Buy', 'Sell', 'Close'][action]} | Reward: {reward:.3f} | Net Worth: ${info['net_worth']:.2f}"
+            if done:
+                status += " | Episode Completed!"
+            
+            return fig, status
+            
+        except Exception as e:
+            return None, f"❌ Error during step: {str(e)}"
+    
+    def run_episode(self, num_steps):
+        """Run a complete episode"""
+        if self.env is None or self.agent is None:
+            return None, "Please initialize environment first!"
+        
+        try:
+            self.env.reset()
+            total_reward = 0
+            step_data = []
+            
+            for step in range(num_steps):
+                action = self.agent.select_action(self.current_state)
+                next_state, reward, done, info = self.env.step(action)
+                self.current_state = next_state
+                total_reward += reward
+                
+                step_data.append({
+                    'step': step,
+                    'action': action,
+                    'reward': reward,
+                    'net_worth': info['net_worth'],
+                    'price': info['current_price']
+                })
+                
+                if done:
+                    break
+            
+            # Create episode summary visualization
+            fig = self.create_episode_summary(step_data)
+            summary = f"Episode completed! Total Reward: {total_reward:.2f} | Final Net Worth: ${info['net_worth']:.2f}"
+            
+            return fig, summary
+            
+        except Exception as e:
+            return None, f"❌ Error during episode: {str(e)}"
+    
+    def train_agent(self, num_episodes, learning_rate):
+        """Train the AI agent"""
+        if self.env is None:
+            return "Please initialize environment first!"
+        
+        self.is_training = True
+        progress = []
+        
+        try:
+            for episode in range(num_episodes):
+                state = self.env.reset()
+                episode_reward = 0
+                done = False
+                
+                while not done:
+                    action = self.agent.select_action(state)
+                    next_state, reward, done, info = self.env.step(action)
+                    self.agent.store_transition(state, action, reward, next_state, done)
+                    state = next_state
+                    episode_reward += reward
+                
+                # Update agent
+                loss = self.agent.update()
+                
+                progress.append({
+                    'episode': episode,
+                    'reward': episode_reward,
+                    'net_worth': info['net_worth'],
+                    'loss': loss
+                })
+                
+                yield self.create_training_progress(progress), f"Training... Episode {episode+1}/{num_episodes}"
+            
+            self.is_training = False
+            yield self.create_training_progress(progress), "✅ Training completed!"
+            
+        except Exception as e:
+            self.is_training = False
+            yield None, f"❌ Training error: {str(e)}"
+    
+    def create_visualization(self, info, action, reward):
+        """Create real-time trading visualization"""
+        fig = make_subplots(
+            rows=2, cols=2,
+            subplot_titles=['Price Chart & Actions', 'Portfolio Performance', 
+                          'Action Distribution', 'Reward History'],
+            specs=[[{"secondary_y": True}, {}],
+                   [{}, {}]],
+            vertical_spacing=0.1,
+            horizontal_spacing=0.1
+        )
+        
+        # Add price chart with actions
+        price_data = self.env.get_price_history()
+        fig.add_trace(
+            go.Scatter(x=list(range(len(price_data))), y=price_data,
+                      mode='lines', name='Price', line=dict(color='blue')),
+            row=1, col=1
+        )
+        
+        # Add portfolio value
+        portfolio_history = [h['net_worth'] for h in self.episode_history[-50:]]
+        if portfolio_history:
+            fig.add_trace(
+                go.Scatter(x=list(range(len(portfolio_history))), y=portfolio_history,
+                          mode='lines', name='Portfolio', line=dict(color='green')),
+                row=1, col=2
+            )
+        
+        # Add action distribution
+        if self.episode_history:
+            actions = [h['action'] for h in self.episode_history]
+            action_counts = pd.Series(actions).value_counts().sort_index()
+            fig.add_trace(
+                go.Bar(x=['Hold', 'Buy', 'Sell', 'Close'][:len(action_counts)], 
+                      y=action_counts.values, name='Actions'),
+                row=2, col=1
+            )
+        
+        # Add reward history
+        rewards = [h['reward'] for h in self.episode_history[-20:]]
+        if rewards:
+            fig.add_trace(
+                go.Scatter(x=list(range(len(rewards))), y=rewards,
+                          mode='lines+markers', name='Rewards', line=dict(color='orange')),
+                row=2, col=2
+            )
+        
+        fig.update_layout(
+            height=600,
+            showlegend=True,
+            title_text=f"Trading Dashboard | Action: {['Hold', 'Buy', 'Sell', 'Close'][action]} | Reward: {reward:.3f}"
+        )
+        
+        return fig
+    
+    def create_episode_summary(self, step_data):
+        """Create episode summary visualization"""
+        if not step_data:
+            return go.Figure()
+        
+        df = pd.DataFrame(step_data)
+        
+        fig = make_subplots(
+            rows=2, cols=2,
+            subplot_titles=['Portfolio Value Over Time', 'Cumulative Rewards',
+                          'Action Frequency', 'Price vs Actions'],
+            specs=[[{}, {}], [{}, {}]]
+        )
+        
+        # Portfolio value
+        fig.add_trace(
+            go.Scatter(x=df['step'], y=df['net_worth'], mode='lines',
+                      name='Portfolio Value', line=dict(color='green')),
+            row=1, col=1
+        )
+        
+        # Cumulative rewards
+        df['cumulative_reward'] = df['reward'].cumsum()
+        fig.add_trace(
+            go.Scatter(x=df['step'], y=df['cumulative_reward'], mode='lines',
+                      name='Cumulative Reward', line=dict(color='orange')),
+            row=1, col=2
+        )
+        
+        # Action frequency
+        action_counts = df['action'].value_counts().sort_index()
+        fig.add_trace(
+            go.Bar(x=[['Hold', 'Buy', 'Sell', 'Close'][i] for i in action_counts.index],
+                  y=action_counts.values, name='Actions'),
+            row=2, col=1
+        )
+        
+        # Price with action markers
+        fig.add_trace(
+            go.Scatter(x=df['step'], y=df['price'], mode='lines',
+                      name='Price', line=dict(color='blue')),
+            row=2, col=2
+        )
+        
+        # Add action markers
+        buy_actions = df[df['action'] == 1]
+        sell_actions = df[df['action'] == 2]
+        
+        if not buy_actions.empty:
+            fig.add_trace(
+                go.Scatter(x=buy_actions['step'], y=buy_actions['price'],
+                          mode='markers', name='Buy', marker=dict(color='green', size=10, symbol='triangle-up')),
+                row=2, col=2
+            )
+        
+        if not sell_actions.empty:
+            fig.add_trace(
+                go.Scatter(x=sell_actions['step'], y=sell_actions['price'],
+                          mode='markers', name='Sell', marker=dict(color='red', size=10, symbol='triangle-down')),
+                row=2, col=2
+            )
+        
+        fig.update_layout(height=600, showlegend=True, title_text="Episode Summary")
+        return fig
+    
+    def create_training_progress(self, progress):
+        """Create training progress visualization"""
+        if not progress:
+            return go.Figure()
+        
+        df = pd.DataFrame(progress)
+        
+        fig = make_subplots(
+            rows=2, cols=2,
+            subplot_titles=['Episode Rewards', 'Portfolio Value',
+                          'Training Loss', 'Performance Metrics'],
+            specs=[[{}, {}], [{}, {}]]
+        )
+        
+        # Rewards
+        fig.add_trace(
+            go.Scatter(x=df['episode'], y=df['reward'], mode='lines+markers',
+                      name='Reward', line=dict(color='blue')),
+            row=1, col=1
+        )
+        
+        # Portfolio value
+        fig.add_trace(
+            go.Scatter(x=df['episode'], y=df['net_worth'], mode='lines+markers',
+                      name='Net Worth', line=dict(color='green')),
+            row=1, col=2
+        )
+        
+        # Loss
+        if 'loss' in df.columns:
+            fig.add_trace(
+                go.Scatter(x=df['episode'], y=df['loss'], mode='lines+markers',
+                          name='Loss', line=dict(color='red')),
+                row=2, col=1
+            )
+        
+        # Moving average reward
+        if len(df) > 10:
+            df['ma_reward'] = df['reward'].rolling(window=10).mean()
+            fig.add_trace(
+                go.Scatter(x=df['episode'], y=df['ma_reward'], mode='lines',
+                          name='MA Reward (10)', line=dict(color='orange', dash='dash')),
+                row=2, col=2
+            )
+        
+        fig.update_layout(height=600, showlegend=True, title_text="Training Progress")
+        return fig
+
+# Initialize the demo
+demo = TradingAIDemo()
+
+# Create Gradio interface
+def create_interface():
+    with gr.Blocks(theme=gr.themes.Soft(), title="Visual Trading AI") as interface:
+        gr.Markdown("""
+        # 🚀 Visual Trading AI
+        *Intelligent Trading Agent with Visual Market Analysis*
+        
+        This AI agent learns to trade by analyzing price charts visually using Deep Reinforcement Learning.
+        """)
+        
+        with gr.Row():
+            with gr.Column(scale=1):
+                # Configuration section
+                gr.Markdown("## ⚙️ Configuration")
+                
+                initial_balance = gr.Slider(1000, 50000, value=10000, step=1000, 
+                                          label="Initial Balance ($)")
+                risk_level = gr.Radio(["Low", "Medium", "High"], value="Medium",
+                                    label="Risk Level")
+                asset_type = gr.Radio(["Stock", "Crypto", "Forex"], value="Stock",
+                                    label="Asset Type")
+                
+                init_btn = gr.Button("🚀 Initialize Environment", variant="primary")
+                init_status = gr.Textbox(label="Status", interactive=False)
+            
+            with gr.Column(scale=2):
+                # Visualization output
+                plot_output = gr.Plot(label="Trading Dashboard")
+                status_output = gr.Textbox(label="Step Status", interactive=False)
+        
+        with gr.Row():
+            # Action controls
+            action_choice = gr.Radio(["AI Decision", "Buy", "Sell", "Hold", "Close"],
+                                   value="AI Decision", label="Action Selection")
+            step_btn = gr.Button("▶️ Execute Step", variant="secondary")
+            episode_btn = gr.Button("🎯 Run Episode (50 steps)", variant="secondary")
+        
+        with gr.Row():
+            # Training section
+            gr.Markdown("## 🎓 AI Training")
+            
+            with gr.Column():
+                num_episodes = gr.Slider(10, 1000, value=100, step=10,
+                                       label="Training Episodes")
+                learning_rate = gr.Slider(0.0001, 0.01, value=0.001, step=0.0001,
+                                        label="Learning Rate")
+                train_btn = gr.Button("🤖 Start Training", variant="primary")
+            
+            with gr.Column():
+                training_plot = gr.Plot(label="Training Progress")
+                training_status = gr.Textbox(label="Training Status")
+        
+        with gr.Row():
+            # Information section
+            gr.Markdown("## 📊 Performance Metrics")
+            metrics = gr.DataFrame(
+                headers=["Metric", "Value"],
+                value=[["Total Steps", "0"], ["Total Reward", "0"], 
+                      ["Current Net Worth", "$10,000"], ["Best Action", "Hold"]],
+                row_count=4, col_count=2, interactive=False
+            )
+        
+        # Event handlers
+        init_btn.click(
+            demo.initialize_environment,
+            inputs=[initial_balance, risk_level, asset_type],
+            outputs=[init_status]
+        )
+        
+        step_btn.click(
+            demo.run_single_step,
+            inputs=[action_choice],
+            outputs=[plot_output, status_output]
+        )
+        
+        episode_btn.click(
+            lambda: demo.run_episode(50),
+            outputs=[plot_output, status_output]
+        )
+        
+        train_btn.click(
+            demo.train_agent,
+            inputs=[num_episodes, learning_rate],
+            outputs=[training_plot, training_status]
+        )
+        
+        gr.Markdown("""
+        ## 🔧 How It Works
+        
+        **Architecture:**
+        - **Visual Processing**: CNN analyzes price charts
+        - **Reinforcement Learning**: PPO algorithm learns trading strategies
+        - **Real-time Visualization**: Interactive dashboard shows agent decisions
+        
+        **Features:**
+        - 🎯 Visual market analysis
+        - 🤖 Deep RL-based decision making
+        - 📊 Real-time performance tracking
+        - 🎮 Interactive control
+        - 📈 Professional visualization
+        
+        *Built with PyTorch, Gym, and Gradio*
+        """)
+    
+    return interface
+
+# Create and launch interface
+if __name__ == "__main__":
+    interface = create_interface()
+    interface.launch(
+        share=True,
+        server_name="0.0.0.0",
+        server_port=7860,
+        show_error=True
+    )