Update app.py

app.py

@@ -4,201 +4,197 @@ import os
 import cv2
 import numpy as np
 import base64
-import json
+import re
 from typing import Tuple, Dict, Any
 from datetime import datetime
 
 # OpenAI API Configuration
 openai.api_key = os.getenv("OPENAI_API_KEY")
-ANALYSIS_MODEL = "gpt-4o-mini"
-MAX_TOKENS = 1500
+ANALYSIS_MODEL = "gpt-4o"
+MAX_TOKENS = 4096
 
 class ChartAnalyzer:
     def __init__(self):
-        self.previous_analysis = {}
-        self.trend_history = []
-        self.user_preferences = {}
+        self.last_analysis = {}
 
     def encode_image(self, image_path: str) -> str:
-        """Encode image to base64 with error handling"""
-        try:
-            with open(image_path, "rb") as image_file:
-                return base64.b64encode(image_file.read()).decode('utf-8')
-        except Exception as e:
-            raise ValueError(f"Image encoding error: {str(e)}")
-
-    def dynamic_trend_calculation(self, analysis_text: str) -> Dict[str, Any]:
-        """Calculate dynamic trend parameters from analysis text"""
-        trend_data = {
-            'short_term': {'slope': 0, 'direction': 'neutral'},
-            'medium_term': {'slope': 0, 'direction': 'neutral'},
-            'long_term': {'slope': 0, 'direction': 'neutral'}
+        """Encode image to base64 with validation"""
+        if not os.path.exists(image_path):
+            raise FileNotFoundError("File not found")
+        if not image_path.lower().endswith(('.png', '.jpg', '.jpeg')):
+            raise ValueError("Unsupported file format")
+            
+        with open(image_path, "rb") as image_file:
+            return base64.b64encode(image_file.read()).decode('utf-8')
+
+    def extract_financial_data(self, analysis_text: str) -> Dict[str, Any]:
+        """Extract structured data from analysis text"""
+        data = {
+            'current_price': None,
+            'support_levels': [],
+            'resistance_levels': [],
+            'patterns': [],
+            'timeframe': None
         }
         
-        # Simplified slope calculation example
-        if "uptrend" in analysis_text.lower():
-            trend_data['short_term']['slope'] = 0.5
-            trend_data['short_term']['direction'] = 'bullish'
-        if "downtrend" in analysis_text.lower():
-            trend_data['short_term']['slope'] = -0.5
-            trend_data['short_term']['direction'] = 'bearish'
+        # Price extraction
+        price_match = re.search(r'\$?(\d+\.?\d*)', analysis_text)
+        if price_match:
+            data['current_price'] = float(price_match.group(1))
             
-        return trend_data
-
-    def calculate_confidence_score(self, analysis_text: str) -> float:
-        """Calculate confidence score based on analysis content"""
-        score = 0.5  # Base score
-        keywords = {
-            'strong support': 0.2,
-            'clear resistance': 0.15,
-            'confirmed pattern': 0.25,
-            'high volume': 0.1,
-            'breakout': 0.2
-        }
+        # Timeframe detection
+        timeframe_match = re.search(r'(1[5m]|4h|1[WD]|1M|1Y)', analysis_text)
+        if timeframe_match:
+            data['timeframe'] = timeframe_match.group(0)
+            
+        # Pattern detection
+        patterns = ['triangle', 'wedge', 'head and shoulders', 'flag', 'double top', 'double bottom']
+        data['patterns'] = [p for p in patterns if p in analysis_text.lower()]
         
-        for keyword, value in keywords.items():
-            if keyword in analysis_text.lower():
-                score += value
-        return min(max(score, 0), 1)  # Clamp between 0-1
-
-    def calculate_price_targets(self, analysis_text: str, current_price: float) -> Dict[str, float]:
-        """Calculate price targets based on detected patterns"""
-        targets = {}
-        if "triangle" in analysis_text.lower():
-            targets['pattern_target'] = current_price * 1.1
-        if "double bottom" in analysis_text.lower():
-            targets['pattern_target'] = current_price * 1.15
-        return targets
+        # Level extraction
+        levels = re.findall(r'\$?(\d+\.?\d*)\s*(support|resistance)', analysis_text, re.IGNORECASE)
+        for value, level_type in levels:
+            if level_type.lower() == 'support':
+                data['support_levels'].append(float(value))
+            else:
+                data['resistance_levels'].append(float(value))
+                
+        return data
+
+    def calculate_confidence(self, analysis_text: str) -> float:
+        """Dynamic confidence calculation"""
+        confidence_signals = {
+            'clear pattern': 0.3,
+            'strong volume': 0.2,
+            'key level break': 0.25,
+            'multiple indicators': 0.15,
+            'market alignment': 0.1
+        }
+        return min(sum(
+            value for key, value in confidence_signals.items() 
+            if key in analysis_text.lower()
+        ), 1.0)
 
-    def analyze_chart(self, image_path: str, user_message: str, time_horizon: str) -> Tuple[str, str]:
-        """Main analysis function with enhanced features"""
+    def generate_annotations(self, img: np.ndarray, analysis: Dict) -> np.ndarray:
+        """Advanced chart annotations"""
+        h, w = img.shape[:2]
+        overlay = img.copy()
+        
+        # Trend lines
+        cv2.line(overlay, (0, h//2), (w, h//3), (0,255,0), 2)
+        cv2.line(overlay, (0, h//2), (w, 2*h//3), (0,0,255), 2)
+        
+        # Support/resistance zones
+        if analysis['support_levels']:
+            y_pos = int(h * 0.75)
+            cv2.rectangle(overlay, (0, y_pos), (w, y_pos+10), (0,255,0), -1)
+            
+        if analysis['resistance_levels']:
+            y_pos = int(h * 0.25)
+            cv2.rectangle(overlay, (0, y_pos), (w, y_pos+10), (0,0,255), -1)
+            
+        # Pattern highlights
+        if 'triangle' in analysis['patterns']:
+            pts = np.array([[w//4, h//2], [3*w//4, h//3], [3*w//4, 2*h//3]])
+            cv2.drawContours(overlay, [pts], -1, (255,255,0), 3)
+            
+        # Confidence indicator
+        confidence = self.calculate_confidence(analysis['raw_text'])
+        cv2.rectangle(overlay, (w-150, 20), (w-50, 40), (50,50,50), -1)
+        cv2.rectangle(overlay, (w-150, 20), 
+                     (w-150 + int(100*confidence), 40), 
+                     (0,255,0), -1)
+                     
+        return cv2.addWeighted(overlay, 0.3, img, 0.7, 0)
+
+    def analyze_chart(self, image_path: str) -> Tuple[str, str]:
+        """Complete analysis pipeline"""
         try:
             # Image processing
             base64_image = self.encode_image(image_path)
             image_url = f"data:image/jpeg;base64,{base64_image}"
-
-            # Prepare AI request with time horizon context
-            system_message = f"""You are an expert crypto analyst. Analyze this chart considering {time_horizon} time horizon.
-                            Include: trend analysis, pattern recognition (flag, wedge, H&S), support/resistance zones,
-                            volume analysis, and risk assessment. Provide specific price targets where applicable."""
-
+            
+            # AI Analysis
             response = openai.ChatCompletion.create(
                 model=ANALYSIS_MODEL,
                 messages=[
-                    {"role": "system", "content": system_message},
-                    {"role": "user", "content": user_message},
-                    {"role": "user", "content": {"type": "image_url", "image_url": image_url}}
+                    {"role": "system", "content": """Expert Crypto Chart Analysis:
+                    - Identify exact price levels and timeframe from the chart
+                    - Detect all technical patterns
+                    - List key support/resistance levels
+                    - Provide actionable trade ideas"""},
+                    {"role": "user", "content": [
+                        {"type": "text", "text": "Analyze this chart in detail"},
+                        {"type": "image_url", "image_url": image_url}
+                    ]}
                 ],
                 max_tokens=MAX_TOKENS
             )
-
-            raw_output = response.choices[0].message.content
             
-            # Enhanced processing
-            confidence = self.calculate_confidence_score(raw_output)
-            trends = self.dynamic_trend_calculation(raw_output)
-            price_targets = self.calculate_price_targets(raw_output, 100)  # Replace with actual price
+            raw_text = response.choices[0].message.content
+            analysis = self.extract_financial_data(raw_text)
+            analysis['raw_text'] = raw_text
             
-            # Format insights with new features
-            formatted_insights = f"""
-            ## 📈 Multi-Timeframe Analysis ({time_horizon})
-            **Confidence Score:** {confidence*100:.1f}%
-            **Trend Dynamics:** 
-            - Short Term: {trends['short_term']['direction'].upper()} (Slope: {trends['short_term']['slope']})
-            - Medium Term: {trends['medium_term']['direction'].upper()}
+            # Image annotation
+            img = cv2.imread(image_path)
+            if img is not None:
+                img = cv2.resize(img, (1024, 512))
+                annotated_img = self.generate_annotations(img, analysis)
+                output_path = f"analyzed_{datetime.now().strftime('%Y%m%d%H%M%S')}.png"
+                cv2.imwrite(output_path, annotated_img)
+            else:
+                output_path = None
+                
+            # Format results
+            formatted_output = f"""
+            ## 📊 Comprehensive Analysis
+            **Current Price:** ${analysis['current_price']:.2f if analysis['current_price'] else 'N/A'}
+            **Timeframe:** {analysis['timeframe'] or 'Auto-detected'}
             
-            ### 🎯 Price Targets:
-            {json.dumps(price_targets, indent=2)}
+            ### 🔑 Key Levels
+            - Support: {', '.join(map(str, analysis['support_levels']))}
+            - Resistance: {', '.join(map(str, analysis['resistance_levels']))}
             
-            ### 🔍 Detailed Analysis:
-            {raw_output}
-            """
+            ### 🌀 Detected Patterns
+            {', '.join(analysis['patterns']) or 'No clear patterns'}
             
-            # Process image with all overlays
-            marked_image = self.overlay_patterns(image_path, raw_output, trends)
+            ### 📝 Detailed Analysis
+            {raw_text}
+            """
             
-            # Store analysis for comparison
-            self.store_analysis_result(marked_image, formatted_insights)
+            return formatted_output, output_path
             
-            return formatted_insights, marked_image
-
         except Exception as e:
-            return f"⚠️ Analysis Error: {str(e)}", None
-
-    def overlay_patterns(self, image_path: str, analysis: str, trends: Dict) -> str:
-        """Advanced pattern overlay with dynamic elements"""
-        img = cv2.imread(image_path)
-        if img is None:
-            return None
-
-        h, w = img.shape[:2]
-        overlay = img.copy()
-        
-        # Dynamic trend lines
-        for timeframe in ['short_term', 'medium_term']:
-            if trends[timeframe]['direction'] != 'neutral':
-                color = (0, 255, 0) if 'bull' in trends[timeframe]['direction'] else (0, 0, 255)
-                thickness = 2 if timeframe == 'short_term' else 1
-                cv2.line(overlay, (0, int(h*(0.5 - trends[timeframe]['slope']*0.1))), 
-                        (w, int(h*(0.5 - trends[timeframe]['slope']*0.2))), 
-                        color, thickness)
-
-        # Support/Resistance bands
-        if "support" in analysis.lower():
-            cv2.rectangle(overlay, (0, int(h*0.75)), (w, int(h*0.85)), (0, 255, 0, 50), -1)
-        if "resistance" in analysis.lower():
-            cv2.rectangle(overlay, (0, int(h*0.15)), (w, int(h*0.25)), (0, 0, 255, 50), -1)
+            return f"Analysis Error: {str(e)}", None
 
-        # Pattern annotations
-        if "triangle" in analysis.lower():
-            cv2.drawContours(overlay, [np.array([[w//4, h//2], [3*w//4, h//3], [3*w//4, 2*h//3]])], 0, (255,255,0), 2)
-
-        # Blending overlays
-        alpha = 0.4
-        img = cv2.addWeighted(overlay, alpha, img, 1 - alpha, 0)
-
-        # Add confidence meter
-        cv2.rectangle(img, (w-120, 20), (w-20, 40), (50,50,50), -1)
-        cv2.rectangle(img, (w-120, 20), (w-120 + int(100*0.8), 40), (0,255,0), -1)  # Example 80% confidence
-
-        output_path = f"marked_{datetime.now().strftime('%Y%m%d%H%M%S')}.png"
-        cv2.imwrite(output_path, img)
-        return output_path
-
-    def store_analysis_result(self, image_path: str, analysis: str):
-        """Store current analysis for future comparison"""
-        self.previous_analysis = {
-            'timestamp': datetime.now(),
-            'image': image_path,
-            'analysis': analysis
-        }
-
-# Gradio Interface Components
+# Gradio Interface
 analyzer = ChartAnalyzer()
 
-with gr.Blocks(theme=gr.themes.Soft()) as demo:
-    gr.Markdown("# 🚀 Pro Crypto Analysis Suite")
+with gr.Blocks(theme=gr.themes.Soft(primary_hue="emerald")) as demo:
+    gr.Markdown("# 🔥 Instant Crypto Chart Analysis")
     
     with gr.Row():
         with gr.Column(scale=1):
-            time_horizon = gr.Dropdown(["Short Term (1D-1W)", "Medium Term (1W-1M)", "Long Term (1M+)"], 
-                                     label="Trading Horizon", value="Medium Term (1W-1M)")
-            user_query = gr.Textbox(label="Analysis Request", placeholder="Specify your analysis focus...")
-            chart_input = gr.Image(type="filepath", label="Upload Chart", sources=["upload"])
-            analyze_btn = gr.Button("Run Full Analysis", variant="primary")
+            chart_input = gr.Image(
+                type="filepath",
+                label="Upload Chart (PNG/JPG)",
+                sources=["upload"],
+                height=300
+            )
+            analyze_btn = gr.Button("Analyze Now", variant="primary")
             
         with gr.Column(scale=2):
-            insights_output = gr.Markdown(label="Analysis Results")
-            annotated_chart = gr.Image(label="Annotated Chart", interactive=False)
-            comparison_section = gr.Markdown("## 🔄 Historical Comparison")
-            
-    # Event handling
+            with gr.Tabs():
+                with gr.TabItem("Full Analysis"):
+                    analysis_output = gr.Markdown()
+                with gr.TabItem("Annotated Chart"):
+                    chart_output = gr.Image(interactive=False)
+
     analyze_btn.click(
         fn=analyzer.analyze_chart,
-        inputs=[chart_input, user_query, time_horizon],
-        outputs=[insights_output, annotated_chart]
+        inputs=chart_input,
+        outputs=[analysis_output, chart_output]
     )
 
-# Launch configuration
 if __name__ == "__main__":
     demo.launch()