app.py

import gradio as gr
import cv2
import numpy as np
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

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"

# 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}

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

def respond(message, history, image):
    if image is None:
        return "Please upload a cryptocurrency chart for analysis."
    
    try:
        processed_img = adaptive_resize(image)
        annotated_img, elements = detect_chart_elements(processed_img)
        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"⚠️ 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()