Update app.py

app.py

@@ -1,99 +1,130 @@
 import gradio as gr
 import cv2
 import numpy as np
-from transformers import (
-    AutoModelForObjectDetection,
-    DetrImageProcessor
-)
 from PIL import Image, ImageDraw
 import warnings
 import io
 import torch
 import base64
 import os
+import matplotlib.pyplot as plt
+from scipy import stats
+from sklearn.cluster import DBSCAN
+from transformers import (
+    AutoModelForObjectDetection,
+    DetrImageProcessor,
+    pipeline
+)
 from huggingface_hub import InferenceClient
 
-ACCESS_TOKEN = os.getenv("HF_TOKEN")
-
 warnings.filterwarnings("ignore")
 
 # Constants
-MAX_WIDTH = 800
-MAX_HEIGHT = 600
+MAX_SIZE = 1024
+CLAHE_CLIP = 3.0
+CANNY_THRESHOLDS = (50, 200)
+HOUGH_PARAMS = (50, 30, 50)
+DBSCAN_EPS = 10.0
+MIN_SAMPLES = 5
 
-# CPU optimizasyonları
-os.environ["OMP_NUM_THREADS"] = str(os.cpu_count() or 4)
+# CPU optimizations
+os.environ["OMP_NUM_THREADS"] = str(os.cpu_count() or 8)
+torch.set_num_threads(os.cpu_count() or 8)
 
-# Model ve InferenceClient ayarları
-DETECTION_MODEL = "facebook/detr-resnet-50"
-LLM_MODEL_NAME = "meta-llama/Llama-3.1-8B-Instruct"
+# Model configurations
+DETECTION_MODEL = "nickmuchi/yolos-small-finance"
+LLM_MODEL_NAME = "meta-llama/Meta-Llama-3-70B-Instruct"
 
-# Detection model ve processor
+# Initialize models
 detection_processor = DetrImageProcessor.from_pretrained(DETECTION_MODEL)
 detection_model = AutoModelForObjectDetection.from_pretrained(DETECTION_MODEL)
+llm_client = InferenceClient(model=LLM_MODEL_NAME, token=os.getenv("HF_TOKEN"))
 
-# InferenceClient for LLM
-llm_client = InferenceClient(
-    model=LLM_MODEL_NAME,
-    token=ACCESS_TOKEN
-)
-
-# Optimize edilmiş sistem prompt
+# Enhanced system prompt
 SYSTEM_PROMPT = """<|begin_of_text|><|start_header_id|>system<|end_header_id|>
-You are a professional financial analyst specializing in cryptocurrency technical analysis.
-Analyze the following chart elements and provide a detailed report:
-<|eot_id|><|start_header_id|>user<|end_header_id|>
-Chart Elements Detected:
-- Support/Resistance: {support_resistance}
-- Trendlines: {trendlines}
-- Patterns: {patterns}
-- Candlestick formations: {candlesticks}
-User Question: {question}
-Provide analysis covering:
-1. Trend analysis (primary and secondary trends)
-2. Key support/resistance levels
-3. Detected chart patterns
-4. Candlestick formations
-5. Volume analysis
-6. Trading signals with confidence levels
-7. Risk management suggestions
-Format response in markdown with clear sections using professional trading terminology.<|eot_id|><|start_header_id|>assistant<|end_header_id|>
+You are a senior cryptocurrency trading analyst with 15 years experience. Analyze the following comprehensive chart data:
+
+Technical Elements Detected:
+{technical_analysis}
+
+User Query: {question}
+
+Provide detailed professional analysis covering:
+1. Price Action Analysis (Trend Strength, Momentum)
+2. Key Support/Resistance Zones (Cluster Analysis)
+3. Volume-Weighted Price Levels
+4. Pattern Recognition (Continuation/Reversal)
+5. Fibonacci Retracement Levels (if applicable)
+6. Market Structure Analysis
+7. Risk/Reward Ratios
+8. Optimal Trade Entry/Exit Strategies
+
+Include statistical confidence levels for each analysis component. Format response in markdown with mathematical notations where appropriate.<|eot_id|><|start_header_id|>assistant<|end_header_id|>
 """
 
-# Yardımcı fonksiyonlar
-def image_to_base64(img):
-    buffered = io.BytesIO()
-    img.save(buffered, format="PNG")
-    return base64.b64encode(buffered.getvalue()).decode("utf-8")
+def adaptive_resize(image):
+    height, width = image.size
+    scale = MAX_SIZE / max(height, width)
+    return image.resize((int(width*scale), int(height*scale)), Image.LANCZOS)
 
-def preprocess_image(image):
-    img = np.array(image)
-    img = cv2.cvtColor(img, cv2.COLOR_RGB2BGR)
-    
-    # Boyutlandırma
-    height, width = img.shape[:2]
-    if width > MAX_WIDTH or height > MAX_HEIGHT:
-        img = cv2.resize(img, (MAX_WIDTH, MAX_HEIGHT), interpolation=cv2.INTER_AREA)
-    
-    # Kontrast iyileştirme
-    lab = cv2.cvtColor(img, cv2.COLOR_BGR2LAB)
+def enhance_contrast(img):
+    lab = cv2.cvtColor(img, cv2.COLOR_RGB2LAB)
     l, a, b = cv2.split(lab)
-    clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8,8))
-    limg = cv2.merge([clahe.apply(l), a, b])
-    enhanced = cv2.cvtColor(limg, cv2.COLOR_LAB2BGR)
-    
-    return Image.fromarray(cv2.cvtColor(enhanced, cv2.COLOR_BGR2RGB))
+    clahe = cv2.createCLAHE(clipLimit=CLAHE_CLIP, tileGridSize=(8,8))
+    limg = clahe.apply(l)
+    merged = cv2.merge([limg, a, b])
+    return cv2.cvtColor(merged, cv2.COLOR_LAB2RGB)
+
+def detect_lines(image):
+    gray = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
+    edges = cv2.Canny(gray, *CANNY_THRESHOLDS)
+    lines = cv2.HoughLinesP(edges, 1, np.pi/180, *HOUGH_PARAMS)
+    return lines
+
+def cluster_lines(lines):
+    if lines is None:
+        return []
+    points = lines.reshape(-1, 2)
+    clustering = DBSCAN(eps=DBSCAN_EPS, min_samples=MIN_SAMPLES).fit(points)
+    return clustering.labels_
+
+def calculate_slope(line):
+    x1, y1, x2, y2 = line
+    return (y2 - y1) / (x2 - x1) if (x2 - x1) != 0 else np.inf
+
+def detect_key_levels(image):
+    gray = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
+    hist = cv2.calcHist([gray], [0], None, [256], [0,256])
+    peaks, _ = find_peaks(hist.flatten(), distance=10, prominence=50)
+    return [p for p in peaks if 10 < p < 240]
+
+def analyze_volume_profile(image):
+    gray = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
+    return cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY+cv2.THRESH_OTSU)[1]
+
+def detect_candlesticks(image):
+    edges = cv2.Canny(image, 50, 150)
+    contours, _ = cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+    candles = []
+    for cnt in contours:
+        x,y,w,h = cv2.boundingRect(cnt)
+        if 5 < w < 50 and 10 < h < 200:
+            candles.append((x,y,w,h))
+    return candles
 
 def detect_chart_elements(image):
-    inputs = detection_processor(images=image, return_tensors="pt")
+    image_np = np.array(image)
+    enhanced = enhance_contrast(image_np)
+    
+    # Deep Learning Detection
+    inputs = detection_processor(images=Image.fromarray(enhanced), return_tensors="pt")
     with torch.no_grad():
         outputs = detection_model(**inputs)
     
-    target_sizes = torch.tensor([image.size[::-1]])
     results = detection_processor.post_process_object_detection(
         outputs, 
-        target_sizes=target_sizes, 
-        threshold=0.7
+        target_sizes=torch.tensor([image.size[::-1]]),
+        threshold=0.85
     )[0]
     
     elements = {
@@ -101,81 +132,115 @@ def detect_chart_elements(image):
         'trendlines': [],
         'patterns': [],
         'candlesticks': [],
+        'indicators': []
     }
     
     draw = ImageDraw.Draw(image)
+    
+    # Process DL detections
     for score, label, box in zip(results["scores"], results["labels"], results["boxes"]):
         box = [round(i, 2) for i in box.tolist()]
         label_name = detection_model.config.id2label[label.item()]
-        
-        # Çizim işlemleri
-        draw.rectangle(box, outline="red", width=2)
-        draw.text((box[0], box[1]), f"{label_name} ({round(score.item(), 2)})", fill="red")
-        
-        # Element kategorizasyonu
-        if "support" in label_name.lower() or "resistance" in label_name.lower():
-            elements['support_resistance'].append(label_name)
-        elif "trendline" in label_name.lower():
-            elements['trendlines'].append(label_name)
-        elif "pattern" in label_name.lower():
-            elements['patterns'].append(label_name)
-        elif "candlestick" in label_name.lower():
-            elements['candlesticks'].append(label_name)
+        elements['patterns' if 'pattern' in label_name else 'indicators'].append(label_name)
+        draw.rectangle(box, outline="#FF0000", width=3)
+        draw.text((box[0], box[1]), f"{label_name} ({score:.2f})", fill="#FF0000")
     
-    return image, elements
-
-def generate_llm_response(elements, question):
-    prompt = SYSTEM_PROMPT.format(
-        support_resistance=", ".join(elements['support_resistance']) or "None detected",
-        trendlines=", ".join(elements['trendlines']) or "None detected",
-        patterns=", ".join(elements['patterns']) or "None detected",
-        candlesticks=", ".join(elements['candlesticks']) or "None detected",
-        question=question
-    )
+    # Traditional CV Detection
+    lines = detect_lines(enhanced)
+    if lines is not None:
+        clusters = cluster_lines(lines)
+        for i, line in enumerate(lines):
+            x1, y1, x2, y2 = line[0]
+            slope = calculate_slope(line[0])
+            length = np.sqrt((x2-x1)**2 + (y2-y1)**2)
+            
+            if abs(slope) < 0.1 and length > 100:
+                elements['support_resistance'].append(f"Key Level at y={y1}")
+                draw.line((x1, y1, x2, y2), fill="#00FF00", width=3)
+            elif 0.1 < abs(slope) < 5:
+                elements['trendlines'].append(f"Trendline ({'Up' if slope < 0 else 'Down'})")
+                draw.line((x1, y1, x2, y2), fill="#0000FF", width=3)
     
-    response = llm_client.text_generation(
-        prompt,
-        max_new_tokens=1024,
-        temperature=0.2,
-        top_p=0.9,
-        do_sample=True
-    )
+    # Volume Profile Analysis
+    volume_profile = analyze_volume_profile(enhanced)
+    contours, _ = cv2.findContours(volume_profile, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+    for cnt in contours:
+        if cv2.contourArea(cnt) > 1000:
+            x,y,w,h = cv2.boundingRect(cnt)
+            elements['support_resistance'].append(f"Volume Cluster at {y+h//2}")
+            draw.rectangle([x,y,x+w,y+h], outline="#FFA500", width=2)
     
-    return response.split("<|assistant|>")[-1].strip()
+    return image, elements
+
+def generate_technical_report(elements):
+    report = []
+    if elements['support_resistance']:
+        report.append("**Key Levels**: " + ", ".join(elements['support_resistance'][:5]))
+    if elements['trendlines']:
+        report.append("**Trend Analysis**: " + ", ".join(elements['trendlines']))
+    if elements['patterns']:
+        report.append("**Chart Patterns**: " + ", ".join(elements['patterns']))
+    return "\n".join(report)
 
-# Gradio Arayüzü
 def respond(message, history, image):
     if image is None:
-        return "Please upload a chart image first."
+        return "Please upload a cryptocurrency chart for analysis."
     
     try:
-        processed_img = preprocess_image(image)
+        processed_img = adaptive_resize(image)
         annotated_img, elements = detect_chart_elements(processed_img)
-        analysis = generate_llm_response(elements, message)
-        img_base64 = image_to_base64(annotated_img)
-        img_html = f'<img src="data:image/png;base64,{img_base64}" style="max-width: 800px; margin-bottom: 20px;">'
-        return f"{img_html}\n{analysis}"
+        tech_report = generate_technical_report(elements)
+        
+        full_prompt = SYSTEM_PROMPT.format(
+            technical_analysis=tech_report,
+            question=message
+        )
+        
+        response = llm_client.text_generation(
+            full_prompt,
+            max_new_tokens=1500,
+            temperature=0.1,
+            repetition_penalty=1.1,
+            seed=42
+        )
+        
+        img_base64 = base64.b64encode(annotated_img.tobytes()).decode('utf-8')
+        img_html = f'<div style="border: 2px solid #4CAF50; padding: 10px; margin-bottom: 20px;">' \
+                   f'<img src="data:image/png;base64,{img_base64}" style="max-width: 100%;">' \
+                   f'</div>'
+        
+        return f"{img_html}\n{response.split('<|assistant|>')[-1].strip()}"
+    
     except Exception as e:
-        return f"Error processing request: {str(e)}"
+        return f"⚠️ Advanced Analysis Error: {str(e)}"
 
-# Interface
 demo = gr.ChatInterface(
     fn=respond,
-    additional_inputs=[gr.Image(label="Upload Chart", type="pil")],
+    additional_inputs=[gr.Image(label="Upload Crypto Chart", type="pil")],
     chatbot=gr.Chatbot(
+        avatar_images=["🤖", "📊"],
         show_copy_button=True,
-        layout="panel",
+        layout="bubble",
         bubble_full_width=False,
         sanitize_html=False
     ),
-    title="Crypto Trading Assistant Pro",
+    title="CryptoQuantum Analyst Pro",
+    description="""<div style="text-align: center; border-bottom: 3px solid #4CAF50; padding: 20px;">
+                 <h1>🪙 CryptoQuantum Analyst Pro</h1>
+                 <p>Advanced AI-powered Cryptocurrency Technical Analysis System</p>
+                 <div style="display: flex; justify-content: center; gap: 15px; margin-top: 10px;">
+                     <div style="background: #4CAF5050; padding: 10px; border-radius: 5px;">📈 Multi-Timeframe Analysis</div>
+                     <div style="background: #4CAF5050; padding: 10px; border-radius: 5px;">🔍 Deep Pattern Recognition</div>
+                     <div style="background: #4CAF5050; padding: 10px; border-radius: 5px;">🤖 Neural Market Forecasting</div>
+                 </div>
+                 </div>""",
     theme="Nymbo/Nymbo_Theme",
     textbox=gr.Textbox(
         label="Ask Technical Questions",
-        placeholder="Upload chart image and ask analysis questions...",
+        placeholder="Enter your crypto analysis questions...",
         container=False
     )
 )
 
 if __name__ == "__main__":
-    demo.launch()
+    demo.launch()