app.py

import gradio as gr
import cv2
import numpy as np
from PIL import Image, ImageDraw, ImageFont
import warnings
import torch
import base64
import io
import os
import pytesseract
from sklearn.cluster import DBSCAN
from transformers import (
    AutoModelForObjectDetection,
    DetrImageProcessor,
    pipeline
)
from huggingface_hub import InferenceClient
import matplotlib.pyplot as plt
from scipy import stats

warnings.filterwarnings("ignore")

# Constants
MAX_SIZE = 1024
CLAHE_CLIP = 3.0
CANNY_THRESHOLDS = (50, 200)
HOUGH_PARAMS = (50, 30, 50)
DBSCAN_EPS = 10.0
MIN_SAMPLES = 5

# CPU optimizations
os.environ["OMP_NUM_THREADS"] = str(os.cpu_count() or 8)
torch.set_num_threads(os.cpu_count() or 8)

# Model configurations
DETECTION_MODEL = "facebook/detr-resnet-50"
LLM_MODEL_NAME = "meta-llama/Meta-Llama-3-70B-Instruct"
OCR_CONFIG = r'--oem 3 --psm 6 -c tessedit_char_whitelist=0123456789.$€£¥%'

# 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"))

# Enhanced system prompt
SYSTEM_PROMPT = """<|begin_of_text|><|start_header_id|>system<|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}

Price Axis Information:
{price_info}

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|>
"""

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 extract_price_info(image):
    img_np = np.array(image)
    gray = cv2.cvtColor(img_np, cv2.COLOR_RGB2GRAY)
    data = pytesseract.image_to_data(gray, config=OCR_CONFIG, output_type=pytesseract.Output.DICT)
    
    price_levels = []
    price_rects = []
    
    for i, text in enumerate(data['text']):
        if text.strip() and any(c.isdigit() or c in '$€£¥%' for c in text):
            x = data['left'][i]
            y = data['top'][i]
            w = data['width'][i]
            h = data['height'][i]
            price_rects.append((x, y, w, h))
            
            try:
                price = float(text.replace('$','').replace('%','').strip())
                price_levels.append((y + h//2, price))
            except:
                continue
                
    return price_levels, price_rects

def map_y_to_price(y_pos, price_levels):
    if not price_levels:
        return None
    
    y_values = [y for y, _ in price_levels]
    prices = [p for _, p in price_levels]
    
    try:
        slope, intercept, _, _, _ = stats.linregress(y_values, prices)
        return round(intercept + slope * y_pos, 2)
    except:
        return None

def enhance_contrast(img):
    lab = cv2.cvtColor(img, cv2.COLOR_RGB2LAB)
    l, a, b = cv2.split(lab)
    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_chart_elements(image):
    image_np = np.array(image)
    enhanced = enhance_contrast(image_np)
    
    # OCR for price information
    price_levels, price_rects = extract_price_info(image)
    
    # Deep Learning Detection
    inputs = detection_processor(images=Image.fromarray(enhanced), return_tensors="pt")
    with torch.no_grad():
        outputs = detection_model(**inputs)
    
    results = detection_processor.post_process_object_detection(
        outputs, 
        target_sizes=torch.tensor([image.size[::-1]]),
        threshold=0.85
    )[0]
    
    elements = {
        'support_resistance': [],
        'trendlines': [],
        'patterns': [],
        'candlesticks': [],
        'indicators': []
    }
    
    draw = ImageDraw.Draw(image)
    
    # Draw price levels
    for x, y, w, h in price_rects:
        draw.rectangle([x, y, x+w, y+h], outline="#4CAF50", width=1)
    
    # 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()]
        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")
    
    # Trendline and support/resistance detection
    lines = cv2.HoughLinesP(
        cv2.Canny(cv2.cvtColor(enhanced, cv2.COLOR_RGB2GRAY), *CANNY_THRESHOLDS),
        1, np.pi/180, *HOUGH_PARAMS
    )
    
    if lines is not None:
        for line in lines:
            x1, y1, x2, y2 = line[0]
            slope = (y2 - y1) / (x2 - x1) if (x2 - x1) != 0 else np.inf
            
            price1 = map_y_to_price(y1, price_levels)
            price2 = map_y_to_price(y2, price_levels)
            
            if abs(slope) < 0.1:  # Horizontal line
                label = f"Key Level: {price1:.2f}" if price1 else f"Y={y1}"
                elements['support_resistance'].append(label)
                draw.line((x1, y1, x2, y2), fill="#00FF00", width=3)
                draw.text((x1+5, y1+5), label, fill="#00FF00")
            else:  # Trendline
                draw.line((x1, y1, x2, y2), fill="#0000FF", width=3)
                elements['trendlines'].append(f"Trendline ({'Bullish' if slope < 0 else 'Bearish'})")
    
    return image, elements, price_levels

def generate_technical_report(elements, price_levels):
    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']))
    
    if price_levels:
        prices = [p for _, p in price_levels]
        report.append(f"**Detected Price Range**: ${min(prices):.2f} - ${max(prices):.2f}")
    
    return "\n".join(report)

def respond(message, history, image):
    # Handle initial greeting
    if not history:
        return "Merhaba! Hoş geldiniz. Size nasıl yardımcı olabilirim? Crypto analiz için lütfen grafik yükleyin, genel sorularınızı direkt sorabilirsiniz."
    
    try:
        tech_report = ""
        annotated_img = None
        price_info = ""
        
        if image is not None:
            processed_img = adaptive_resize(image)
            annotated_img, elements, price_levels = detect_chart_elements(processed_img)
            tech_report = generate_technical_report(elements, price_levels)
            
            if price_levels:
                prices = [f"${p:.2f}" for _, p in price_levels]
                price_info = f"Detected Price Levels: {', '.join(prices)}"
        
        full_prompt = SYSTEM_PROMPT.format(
            technical_analysis=tech_report,
            price_info=price_info,
            question=message
        )
        
        response = llm_client.text_generation(
            full_prompt,
            max_new_tokens=1500,
            temperature=0.1,
            repetition_penalty=1.1,
            seed=42
        )
        
        if annotated_img:
            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()}"
        
        return response.split('<|assistant|>')[-1].strip()
    
    except Exception as e:
        return f"⚠️ Advanced Analysis Error: {str(e)}"

demo = gr.ChatInterface(
    fn=respond,
    additional_inputs=[gr.Image(label="Upload Crypto Chart", type="pil")],
    chatbot=gr.Chatbot(
        avatar_images=["user.png", "ai.png"],
        show_copy_button=True,
        layout="bubble",
        bubble_full_width=False,
        sanitize_html=False
    ),
    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="Enter your crypto analysis questions...",
        container=False
    )
)

if __name__ == "__main__":
    demo.launch()