app.py

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

# Set matplotlib backend to avoid display issues
plt.switch_backend('Agg')

# Create necessary directories
os.makedirs('src/environments', exist_ok=True)
os.makedirs('src/agents', exist_ok=True)
os.makedirs('src/visualizers', exist_ok=True)
os.makedirs('src/utils', exist_ok=True)

# Create __init__.py files
for dir_path in ['src', 'src/environments', 'src/agents', 'src/visualizers', 'src/utils']:
    init_file = os.path.join(dir_path, '__init__.py')
    with open(init_file, 'w') as f:
        f.write('')

# Now import our custom modules
sys.path.append('src')

# Import our custom modules
from src.environments.visual_trading_env import VisualTradingEnvironment
from src.agents.visual_agent import VisualTradingAgent
from src.visualizers.chart_renderer import ChartRenderer

class TradingAIDemo:
    def __init__(self):
        self.env = None
        self.agent = None
        self.current_state = None
        self.is_training = False
        self.episode_history = []
        self.chart_renderer = ChartRenderer()
        self.initialized = False
        
    def initialize_environment(self, initial_balance, risk_level, asset_type):
        """Initialize trading environment"""
        try:
            print(f"Initializing environment with balance: {initial_balance}, risk: {risk_level}, asset: {asset_type}")
            
            self.env = VisualTradingEnvironment(
                initial_balance=float(initial_balance),
                risk_level=risk_level,
                asset_type=asset_type
            )
            
            # Initialize agent with correct dimensions
            self.agent = VisualTradingAgent(
                state_dim=(84, 84, 4),
                action_dim=4
            )
            
            self.current_state = self.env.reset()
            self.episode_history = []
            self.initialized = True
            
            return "✅ Environment initialized successfully! Ready for trading."
            
        except Exception as e:
            error_msg = f"❌ Error initializing environment: {str(e)}"
            print(error_msg)
            return error_msg
    
    def run_single_step(self, action_choice):
        """Run a single step in the environment"""
        if not self.initialized or self.env is None or self.agent is None:
            return None, None, 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)
                action_source = "AI"
            else:
                action_mapping = {"Hold": 0, "Buy": 1, "Sell": 2, "Close": 3}
                action = action_mapping[action_choice]
                action_source = "Manual"
            
            print(f"Executing action: {action} ({action_source})")
            
            # Execute action
            next_state, reward, done, info = self.env.step(action)
            self.current_state = next_state
            
            # Update history
            history_entry = {
                'step': len(self.episode_history),
                'action': action,
                'reward': reward,
                'net_worth': info['net_worth'],
                'balance': info['balance'],
                'position': info['position_size'],
                'price': info['current_price'],
                'action_source': action_source
            }
            self.episode_history.append(history_entry)
            
            # Create visualizations
            price_chart = self.create_price_chart(info)
            performance_chart = self.create_performance_chart()
            action_chart = self.create_action_chart()
            
            # Create status message
            action_names = ["Hold", "Buy", "Sell", "Close"]
            status = (
                f"✅ Step {info['step']} Completed!\n"
                f"• Action: {action_names[action]} ({action_source})\n"
                f"• Reward: {reward:.3f}\n"
                f"• Net Worth: ${info['net_worth']:.2f}\n"
                f"• Balance: ${info['balance']:.2f}\n"
                f"• Position: {info['position_size']:.4f}\n"
                f"• Current Price: ${info['current_price']:.2f}"
            )
            
            if done:
                status += "\n🎯 Episode Completed!"
            
            return price_chart, performance_chart, action_chart, status
            
        except Exception as e:
            error_msg = f"❌ Error during step execution: {str(e)}"
            print(error_msg)
            return None, None, None, error_msg
    
    def run_episode(self, num_steps=20):
        """Run a complete episode"""
        if not self.initialized or self.env is None or self.agent is None:
            return None, None, None, "⚠️ Please initialize environment first!"
        
        try:
            # Reset environment for new episode
            self.current_state = self.env.reset()
            self.episode_history = []
            total_reward = 0
            
            print(f"Starting episode with {num_steps} steps...")
            
            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
                
                # Store experience for training
                self.agent.store_transition(self.current_state, action, reward, next_state, done)
                
                self.episode_history.append({
                    'step': step,
                    'action': action,
                    'reward': reward,
                    'net_worth': info['net_worth'],
                    'price': info['current_price'],
                    'action_source': 'AI'
                })
                
                # Small delay to make execution visible
                time.sleep(0.05)
                
                if done:
                    break
            
            # Create visualizations
            price_chart = self.create_price_chart(info)
            performance_chart = self.create_performance_chart()
            action_chart = self.create_action_chart()
            
            # Calculate performance metrics
            initial_balance = self.env.initial_balance
            final_net_worth = info['net_worth']
            total_return = 0.0
            if initial_balance > 0:
                total_return = (final_net_worth - initial_balance) / initial_balance * 100
            
            summary = (
                f"🎯 Episode Completed!\n"
                f"• Total Steps: {len(self.episode_history)}\n"
                f"• Total Reward: {total_reward:.2f}\n"
                f"• Final Net Worth: ${final_net_worth:.2f}\n"
                f"• Total Return: {total_return:.2f}%\n"
                f"• Total Trades: {info['total_trades']}"
            )
            
            return price_chart, performance_chart, action_chart, summary
            
        except Exception as e:
            error_msg = f"❌ Error during episode: {str(e)}"
            print(error_msg)
            return None, None, None, error_msg
    
    def train_agent(self, num_episodes, learning_rate):
        """Train the AI agent"""
        if not self.initialized or self.env is None:
            yield None, "⚠️ Please initialize environment first!"
            return
        
        self.is_training = True
        training_history = []
        
        try:
            num_episodes = int(num_episodes)
            for episode in range(num_episodes):
                state = self.env.reset()
                episode_reward = 0.0
                done = False
                steps = 0
                
                while not done and steps < 100:
                    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
                    steps += 1
                
                # Update agent
                loss = self.agent.update()
                
                training_history.append({
                    'episode': episode,
                    'reward': episode_reward,
                    'net_worth': info['net_worth'],
                    'loss': loss,
                    'steps': steps
                })
                
                # Yield progress every 5 episodes or at the end
                if episode % 5 == 0 or episode == num_episodes - 1:
                    progress_chart = self.create_training_progress(training_history)
                    
                    status = (
                        f"🔄 Training Progress: {episode+1}/{num_episodes}\n"
                        f"• Episode Reward: {episode_reward:.2f}\n"
                        f"• Final Net Worth: ${info['net_worth']:.2f}\n"
                        f"• Loss: {loss:.4f}\n"
                        f"• Epsilon: {self.agent.epsilon:.3f}"
                    )
                    yield progress_chart, status
                
                time.sleep(0.01)
            
            self.is_training = False
            
            # Calculate average reward
            rewards = [h['reward'] for h in training_history]
            avg_reward = np.mean(rewards) if rewards else 0.0
            
            final_status = (
                f"✅ Training Completed!\n"
                f"• Total Episodes: {num_episodes}\n"
                f"• Final Epsilon: {self.agent.epsilon:.3f}\n"
                f"• Average Reward: {avg_reward:.2f}"
            )
            yield self.create_training_progress(training_history), final_status
            
        except Exception as e:
            self.is_training = False
            error_msg = f"❌ Training error: {str(e)}"
            print(f"Training error details: {e}")
            yield None, error_msg
    
    def create_price_chart(self, info):
        """Create price chart with actions"""
        if not self.episode_history:
            # Return empty chart with message
            fig = go.Figure()
            fig.update_layout(
                title="Price Chart - No Data Available",
                xaxis_title="Time Step",
                yaxis_title="Price",
                height=300,
                template="plotly_white"
            )
            return fig
        
        prices = [h['price'] for h in self.episode_history]
        actions = [h['action'] for h in self.episode_history]
        
        fig = go.Figure()
        
        # Price line
        fig.add_trace(go.Scatter(
            x=list(range(len(prices))),
            y=prices,
            mode='lines',
            name='Price',
            line=dict(color='blue', width=3)
        ))
        
        # Action markers
        buy_indices = [i for i, action in enumerate(actions) if action == 1]
        sell_indices = [i for i, action in enumerate(actions) if action == 2]
        close_indices = [i for i, action in enumerate(actions) if action == 3]
        
        if buy_indices:
            fig.add_trace(go.Scatter(
                x=buy_indices,
                y=[prices[i] for i in buy_indices],
                mode='markers',
                name='Buy',
                marker=dict(color='green', size=12, symbol='triangle-up', 
                          line=dict(width=2, color='darkgreen'))
            ))
        
        if sell_indices:
            fig.add_trace(go.Scatter(
                x=sell_indices,
                y=[prices[i] for i in sell_indices],
                mode='markers',
                name='Sell',
                marker=dict(color='red', size=12, symbol='triangle-down', 
                          line=dict(width=2, color='darkred'))
            ))
        
        if close_indices:
            fig.add_trace(go.Scatter(
                x=close_indices,
                y=[prices[i] for i in close_indices],
                mode='markers',
                name='Close',
                marker=dict(color='orange', size=10, symbol='x', 
                          line=dict(width=2, color='darkorange'))
            ))
        
        fig.update_layout(
            title="Price Chart with Trading Actions",
            xaxis_title="Step",
            yaxis_title="Price",
            height=350,
            showlegend=True,
            template="plotly_white"
        )
        
        return fig
    
    def create_performance_chart(self):
        """Create portfolio performance chart"""
        if not self.episode_history:
            fig = go.Figure()
            fig.update_layout(
                title="Portfolio Performance - No Data Available",
                height=400
            )
            return fig
        
        net_worth = [h['net_worth'] for h in self.episode_history]
        rewards = [h['reward'] for h in self.episode_history]
        
        fig = make_subplots(
            rows=2, cols=1,
            subplot_titles=['Portfolio Value Over Time', 'Step Rewards'],
            vertical_spacing=0.15
        )
        
        # Portfolio value
        fig.add_trace(go.Scatter(
            x=list(range(len(net_worth))),
            y=net_worth,
            mode='lines+markers',
            name='Net Worth',
            line=dict(color='green', width=3),
            marker=dict(size=4)
        ), row=1, col=1)
        
        # Add initial balance reference line
        if self.env:
            fig.add_hline(y=self.env.initial_balance, line_dash="dash", 
                         line_color="red", annotation_text="Initial Balance",
                         row=1, col=1)
        
        # Rewards as bar chart
        if rewards:
            fig.add_trace(go.Bar(
                x=list(range(len(rewards))),
                y=rewards,
                name='Reward',
                marker_color=['green' if r >= 0 else 'red' for r in rewards],
                opacity=0.7
            ), row=2, col=1)
        
        fig.update_layout(height=500, showlegend=False, template="plotly_white")
        fig.update_yaxes(title_text="Value ($)", row=1, col=1)
        fig.update_yaxes(title_text="Reward", row=2, col=1)
        fig.update_xaxes(title_text="Step", row=2, col=1)
        
        return fig
    
    def create_action_chart(self):
        """Create action distribution chart"""
        if not self.episode_history:
            fig = go.Figure()
            fig.update_layout(
                title="Action Distribution - No Data Available",
                height=300
            )
            return fig
        
        actions = [h['action'] for h in self.episode_history]
        action_names = ['Hold', 'Buy', 'Sell', 'Close']
        action_counts = [actions.count(i) for i in range(4)]
        
        colors = ['blue', 'green', 'red', 'orange']
        
        fig = go.Figure(data=[go.Pie(
            labels=action_names,
            values=action_counts,
            hole=.4,
            marker_colors=colors,
            textinfo='label+percent+value',
            hoverinfo='label+percent+value'
        )])
        
        fig.update_layout(
            title="Action Distribution",
            height=350,
            annotations=[dict(text='Actions', x=0.5, y=0.5, font_size=16, showarrow=False)],
            template="plotly_white"
        )
        
        return fig
    
    def create_training_progress(self, training_history):
        """Create training progress visualization"""
        if not training_history:
            fig = go.Figure()
            fig.update_layout(
                title="Training Progress - No Data Available",
                height=500
            )
            return fig
        
        episodes = [h['episode'] for h in training_history]
        rewards = [h['reward'] for h in training_history]
        net_worths = [h['net_worth'] for h in training_history]
        losses = [h.get('loss', 0) for h in training_history]
        
        fig = make_subplots(
            rows=2, cols=2,
            subplot_titles=['Episode Rewards', 'Portfolio Value',
                          'Training Loss', 'Moving Average Reward (5)'],
            specs=[[{}, {}], [{}, {}]]
        )
        
        # Rewards
        fig.add_trace(go.Scatter(
            x=episodes, y=rewards, mode='lines+markers',
            name='Reward', line=dict(color='blue', width=2),
            marker=dict(size=4)
        ), row=1, col=1)
        
        # Portfolio value
        fig.add_trace(go.Scatter(
            x=episodes, y=net_worths, mode='lines+markers',
            name='Net Worth', line=dict(color='green', width=2),
            marker=dict(size=4)
        ), row=1, col=2)
        
        # Loss
        if any(loss > 0 for loss in losses):
            fig.add_trace(go.Scatter(
                x=episodes, y=losses, mode='lines+markers',
                name='Loss', line=dict(color='red', width=2),
                marker=dict(size=4)
            ), row=2, col=1)
        
        # Moving average reward
        if len(rewards) > 5:
            ma_rewards = []
            for i in range(len(rewards)):
                start_idx = max(0, i - 4)
                ma = np.mean(rewards[start_idx:i+1])
                ma_rewards.append(ma)
            
            fig.add_trace(go.Scatter(
                x=episodes, y=ma_rewards, mode='lines',
                name='MA Reward (5)', line=dict(color='orange', width=3, dash='dash')
            ), row=2, col=2)
        
        fig.update_layout(
            height=600, 
            showlegend=True, 
            title_text="Training Progress Over Episodes",
            template="plotly_white"
        )
        
        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
        **هوش مصنوعی معامله‌گر بصری - تحلیل چارت‌های قیمت با یادگیری تقویتی عمیق**
        
        *این پروژه از شبکه‌های عصبی کانولوشن برای تحلیل بصری نمودارهای قیمت و یادگیری تقویتی برای تصمیم‌گیری معاملاتی استفاده می‌کند.*
        """)
        
        with gr.Row():
            with gr.Column(scale=1):
                # Configuration section
                gr.Markdown("## ⚙️ پیکربندی محیط")
                
                with gr.Row():
                    initial_balance = gr.Slider(
                        minimum=1000, maximum=50000, value=10000, step=1000,
                        label="موجودی اولیه ($)", info="میزان سرمایه اولیه برای معامله"
                    )
                
                with gr.Row():
                    risk_level = gr.Radio(
                        ["Low", "Medium", "High"], 
                        value="Medium",
                        label="سطح ریسک",
                        info="سطح ریسک پذیری در معاملات"
                    )
                
                with gr.Row():
                    asset_type = gr.Radio(
                        ["Stock", "Crypto", "Forex"], 
                        value="Stock",
                        label="نوع دارایی",
                        info="نوع بازار مالی برای شبیه‌سازی"
                    )
                
                with gr.Row():
                    init_btn = gr.Button(
                        "🚀 راه‌اندازی محیط معاملاتی", 
                        variant="primary",
                        size="lg"
                    )
                
                with gr.Row():
                    init_status = gr.Textbox(
                        label="وضعیت راه‌اندازی",
                        interactive=False,
                        placeholder="برای شروع، محیط را راه‌اندازی کنید...",
                        lines=2
                    )
            
            with gr.Column(scale=2):
                # Status output
                gr.Markdown("## 📊 وضعیت معاملات")
                status_output = gr.Textbox(
                    label="وضعیت اجرا",
                    interactive=False,
                    placeholder="وضعیت معاملات اینجا نمایش داده می‌شود...",
                    lines=4
                )
        
        with gr.Row():
            gr.Markdown("## 🎮 کنترل معاملات")
            
        with gr.Row():
            # Action controls
            action_choice = gr.Radio(
                ["AI Decision", "Buy", "Sell", "Hold", "Close"],
                value="AI Decision",
                label="انتخاب اقدام",
                info="AI Decision: تصمیم خودکار هوش مصنوعی"
            )
            
        with gr.Row():
            with gr.Column(scale=1):
                step_btn = gr.Button(
                    "▶️ اجرای یک قدم", 
                    variant="secondary",
                    size="lg"
                )
            
            with gr.Column(scale=1):
                episode_btn = gr.Button(
                    "🎯 اجرای یک اپیزود (20 قدم)", 
                    variant="secondary",
                    size="lg"
                )
        
        with gr.Row():
            # Visualization outputs
            with gr.Column(scale=1):
                price_chart = gr.Plot(
                    label="📈 نمودار قیمت و اقدامات"
                )
            
            with gr.Column(scale=1):
                performance_chart = gr.Plot(
                    label="💰 عملکرد پرتفولیو"
                )
        
        with gr.Row():
            with gr.Column(scale=1):
                action_chart = gr.Plot(
                    label="🎯 توزیع اقدامات"
                )
        
        with gr.Row():
            gr.Markdown("## 🎓 آموزش هوش مصنوعی")
            
        with gr.Row():
            with gr.Column(scale=1):
                num_episodes = gr.Slider(
                    minimum=10, maximum=200, value=50, step=10,
                    label="تعداد اپیزودهای آموزش",
                    info="تعداد دوره‌های آموزشی"
                )
                
                learning_rate = gr.Slider(
                    minimum=0.0001, maximum=0.01, value=0.001, step=0.0001,
                    label="نرخ یادگیری",
                    info="سرعت یادگیری الگوریتم"
                )
                
                train_btn = gr.Button(
                    "🤖 شروع آموزش", 
                    variant="primary",
                    size="lg"
                )
            
            with gr.Column(scale=2):
                training_plot = gr.Plot(
                    label="📊 پیشرفت آموزش"
                )
                
                training_status = gr.Textbox(
                    label="وضعیت آموزش",
                    interactive=False,
                    placeholder="وضعیت آموزش اینجا نمایش داده می‌شود...",
                    lines=3
                )
        
        with gr.Row():
            gr.Markdown("## ℹ️ راهنمای استفاده")
            
        with gr.Row():
            with gr.Column(scale=1):
                gr.Markdown("""
                **🎯 اقدامات ممکن:**
                - **Hold (0)**: حفظ وضعیت فعلی
                - **Buy (1)**: باز کردن پوزیشن خرید
                - **Sell (2)**: افزایش سایز پوزیشن
                - **Close (3)**: بستن پوزیشن فعلی
                
                **📈 معیارهای عملکرد:**
                - **Reward**: امتیاز دریافتی از محیط
                - **Net Worth**: ارزش کل پرتفولیو
                - **Balance**: موجودی نقدی
                - **Position**: سایز پوزیشن فعلی
                """)
            
            with gr.Column(scale=1):
                gr.Markdown("""
                **🔧 نحوه استفاده:**
                1. محیط را راه‌اندازی کنید
                2. اقدامات تکی یا اپیزودها را اجرا کنید
                3. عملکرد را در نمودارها مشاهده کنید
                4. هوش مصنوعی را آموزش دهید
                5. نتایج را تحلیل کنید
                
                **⚠️ توجه:**
                این یک شبیه‌ساز آموزشی است و برای معاملات واقعی طراحی نشده است.
                """)
        
        # 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=[price_chart, performance_chart, action_chart, status_output]
        )
        
        episode_btn.click(
            demo.run_episode,
            inputs=[],
            outputs=[price_chart, performance_chart, action_chart, status_output]
        )
        
        train_btn.click(
            demo.train_agent,
            inputs=[num_episodes, learning_rate],
            outputs=[training_plot, training_status]
        )
        
        gr.Markdown("""
        ## 🏗 معماری فنی
        
        **🎯 هسته هوش مصنوعی:**
        - **پردازش بصری**: شبکه عصبی کانولوشن (CNN) برای تحلیل نمودارهای قیمت
        - **یادگیری تقویتی**: الگوریتم Deep Q-Network (DQN) برای تصمیم‌گیری
        - **تجربه replay**: ذخیره و بازیابی تجربیات برای یادگیری پایدار
        
        **🛠 فناوری‌ها:**
        - **یادگیری عمیق**: PyTorch
        - **محیط شبیه‌سازی**: محیط اختصاصی معاملاتی
        - **رابط کاربری**: Gradio
        - **ویژوالیزیشن**: Plotly, Matplotlib
        - **پردازش داده**: NumPy, Pandas
        
        **📊 ویژگی‌های کلیدی:**
        - تحلیل بصری نمودارهای قیمت
        - یادگیری خودکار استراتژی‌های معاملاتی
        - نمایش زنده عملکرد و تصمیم‌ها
        - کنترل دستی و خودکار
        - آنالیز جامع عملکرد
        
        *توسعه داده شده توسط Omid Sakaki - 2024*
        """)
    
    return interface

# Create and launch interface
if __name__ == "__main__":
    print("🚀 Starting Visual Trading AI Application...")
    print("📊 Initializing components...")
    
    interface = create_interface()
    
    print("✅ Application initialized successfully!")
    print("🌐 Starting server on http://0.0.0.0:7860")
    print("📱 You can now access the application in your browser")
    
    # Launch with better configuration
    interface.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=False,
        show_error=True,
        debug=True
    )