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'
' \ f'' \ f'
' 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="""

🪙 CryptoQuantum Analyst Pro

Advanced AI-powered Cryptocurrency Technical Analysis System

📈 Multi-Timeframe Analysis
🔍 Deep Pattern Recognition
🤖 Neural Market Forecasting
""", theme="Nymbo/Nymbo_Theme", textbox=gr.Textbox( label="Ask Technical Questions", placeholder="Enter your crypto analysis questions...", container=False ) ) if __name__ == "__main__": demo.launch()