import gradio as gr import openai import os import base64 from functools import lru_cache from PIL import Image import cv2 import numpy as np import datetime import uuid import requests from reportlab.lib.pagesizes import letter from reportlab.platypus import SimpleDocTemplate, Image as RLImage, Paragraph, Spacer from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle from reportlab.lib.enums import TA_JUSTIFY from reportlab.lib import colors # OpenAI ve GitHub Konfigürasyonları openai.api_key = os.getenv("OPENAI_API_KEY") GITHUB_TOKEN = os.getenv("GITHUB_TOKEN") REPO_OWNER = os.getenv("GITHUB_REPO_OWNER") REPO_NAME = os.getenv("GITHUB_REPO_NAME") # Sabitler ANALYSIS_MODEL = "gpt-4o" MAX_TOKENS = 4096 PDF_DIR = "reports" os.makedirs(PDF_DIR, exist_ok=True) # Persona Tanımları PERSONAS = { "Aggressive Trader": { "description": "High-risk, short-term gains, leverages volatile market movements.", "prompt": "Focus on high-risk strategies, short-term gains, and leverage opportunities. Suggest aggressive entry and exit points.", "color": colors.red }, "Conservative Trader": { "description": "Low-risk, long-term investments, prioritizes capital preservation.", "prompt": "Focus on low-risk strategies, long-term investments, and capital preservation. Suggest safe entry points and strict stop-loss levels.", "color": colors.blue }, "Neutral Trader": { "description": "Balanced approach, combines short and long-term strategies.", "prompt": "Focus on balanced strategies, combining short and long-term opportunities. Suggest moderate risk levels and trend-following approaches.", "color": colors.green }, "Reactive Trader": { "description": "Quick decisions based on market news and social media trends.", "prompt": "Focus on quick decision-making, momentum trading, and reacting to market news. Suggest strategies based on current trends and FOMO opportunities.", "color": colors.orange }, "Systematic Trader": { "description": "Algorithm-based, rule-driven, and emotionless trading.", "prompt": "Focus on algorithmic strategies, backtested rules, and quantitative analysis. Suggest data-driven entry and exit points.", "color": colors.purple } } # Sistem Prompt'u SYSTEM_PROMPT = """Professional Crypto Technical Analyst: 1. Identify all technical patterns in the chart 2. Determine key support/resistance levels 3. Analyze volume and momentum indicators 4. Calculate risk/reward ratios 5. Provide clear trading recommendations 6. Include specific price targets 7. Assess market sentiment 8. Evaluate trend strength 9. Identify potential breakout/breakdown levels 10. Provide time-based projections""" class ChartAnalyzer: def __init__(self): self.last_optimized_path = "" def validate_image(self, image_path: str) -> bool: try: with Image.open(image_path) as img: img.verify() img = cv2.imread(image_path) if img is None: return False gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) edges = cv2.Canny(gray, 50, 150) return np.sum(edges) >= 1000 except Exception: return False def optimize_image(self, image_path: str) -> str: try: img = Image.open(image_path) original_width, original_height = img.size max_size = 1024 if original_width > max_size or original_height > max_size: ratio = min(max_size/original_width, max_size/original_height) new_size = (int(original_width * ratio), int(original_height * ratio)) img = img.resize(new_size, Image.LANCZOS) unique_id = uuid.uuid4().hex optimized_path = f"{PDF_DIR}/optimized_chart_{unique_id}.png" img.save(optimized_path, "PNG", optimize=True, quality=85) return optimized_path except Exception as e: print(f"Image optimization error: {str(e)}") return image_path def encode_image(self, image_path: str) -> str: if not os.path.exists(image_path): raise FileNotFoundError("File not found") if os.path.getsize(image_path) > 5 * 1024 * 1024: raise ValueError("Maximum file size is 5MB") with open(image_path, "rb") as image_file: return base64.b64encode(image_file.read()).decode('utf-8') @lru_cache(maxsize=100) def analyze_chart(self, image_path: str, persona: str) -> tuple: try: if not self.validate_image(image_path): return "Error: Invalid or low-quality image", None optimized_path = self.optimize_image(image_path) base64_image = self.encode_image(optimized_path) persona_prompt = PERSONAS.get(persona, {}).get("prompt", "") full_system_prompt = f"{SYSTEM_PROMPT}\n\n{persona_prompt}" response = openai.ChatCompletion.create( model=ANALYSIS_MODEL, messages=[ {"role": "system", "content": full_system_prompt}, {"role": "user", "content": [ {"type": "text", "text": "Perform detailed technical analysis of this chart:"}, {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"} } ]} ], max_tokens=MAX_TOKENS ) analysis_text = response.choices[0].message.content self.last_optimized_path = optimized_path return analysis_text, optimized_path except Exception as e: return f"Error: {str(e)}", None def create_pdf_styles(): styles = getSampleStyleSheet() styles.add(ParagraphStyle( 'Justify', parent=styles['BodyText'], alignment=TA_JUSTIFY, spaceAfter=6 )) styles.add(ParagraphStyle( 'PersonaTitle', fontSize=14, textColor=colors.white, backColor=colors.darkblue, alignment=1, spaceAfter=12 )) return styles def generate_pdf(optimized_image_path: str, analysis_text: str, persona: str) -> str: try: # Validasyonlar if not optimized_image_path: raise ValueError("Optimized image path is missing") if not os.path.exists(optimized_image_path): raise FileNotFoundError(f"Optimized image not found at {optimized_image_path}") if not analysis_text or not analysis_text.strip(): raise ValueError("Analysis text cannot be empty") # PDF dosya yolu oluştur timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S") filename = f"{PDF_DIR}/report_{timestamp}_{uuid.uuid4().hex[:6]}.pdf" # PDF içeriğini oluştur doc = SimpleDocTemplate(filename, pagesize=letter) story = [] # Resim ekleme try: img = Image.open(optimized_image_path) img_width, img_height = img.size aspect = img_height / float(img_width) target_width = 400 target_height = target_width * aspect if target_height > 600: target_height = 600 target_width = target_height / aspect story.append(RLImage(optimized_image_path, width=target_width, height=target_height)) story.append(Spacer(1, 20)) except Exception as e: print(f"PDF image error: {str(e)}") raise # Persona bilgisi if persona in PERSONAS: persona_color = PERSONAS[persona]["color"] story.append(Paragraph( f"Persona: {persona}", ParagraphStyle( 'PersonaTitle', fontSize=14, textColor=colors.white, backColor=persona_color, alignment=1 ) )) story.append(Spacer(1, 20)) # Analiz metni styles = create_pdf_styles() analysis_style = styles['Justify'] cleaned_text = analysis_text.replace('•', '•') for line in cleaned_text.split('\n'): if line.strip(): p = Paragraph(line, analysis_style) story.append(p) story.append(Spacer(1, 12)) # PDF'i oluştur doc.build(story) if not os.path.exists(filename): raise RuntimeError("PDF file creation failed") return filename except Exception as e: print(f"[PDF Generation Error] {str(e)}") return None def upload_to_github(file_path: str) -> str: try: if not file_path or file_path == "None": return "⛔ Error: PDF generation failed in previous step" if not os.path.exists(file_path): return f"⛔ File not found: {file_path}" # GitHub yükleme işlemleri file_name = os.path.basename(file_path) url = f"https://api.github.com/repos/{REPO_OWNER}/{REPO_NAME}/contents/{PDF_DIR}/{file_name}" with open(file_path, "rb") as f: content = base64.b64encode(f.read()).decode("utf-8") headers = { "Authorization": f"token {GITHUB_TOKEN}", "Accept": "application/vnd.github.v3+json" } # Dosya varlık kontrolü response = requests.get(url, headers=headers) sha = response.json().get("sha") if response.status_code == 200 else None data = { "message": f"Add report {file_name}", "content": content, "branch": "main" } if sha: data["sha"] = sha response = requests.put(url, headers=headers, json=data) if response.status_code in [200, 201]: return f"✅ Report successfully uploaded to GitHub!\nURL: {response.json()['content']['html_url']}" return f"❌ GitHub upload failed ({response.status_code}): {response.text}" except Exception as e: return f"⚠️ Upload error: {str(e)}" with gr.Blocks(theme=gr.themes.Soft()) as demo: analyzer = ChartAnalyzer() with gr.Column(): gr.Markdown("# 🚀 CryptoVision Pro") with gr.Row(): with gr.Column(): chart_input = gr.Image(type="filepath", label="Upload Chart", sources=["upload"]) persona_dropdown = gr.Dropdown( list(PERSONAS.keys()), label="Select Trading Persona", value="Neutral Trader", info="Choose your trading style" ) analyze_btn = gr.Button("Analyze Chart", variant="primary") with gr.Column(): analysis_output = gr.Markdown("## Analysis Results\n*Your analysis will appear here*") pdf_status = gr.HTML() # Gizli bileşenler optimized_image_path = gr.Text(visible=False) pdf_file = gr.Text(visible=False) # İşlem Zinciri analyze_btn.click( lambda: [ gr.Markdown(visible=False), gr.HTML(value=""), None ], outputs=[analysis_output, pdf_status, pdf_file], queue=False ).then( analyzer.analyze_chart, inputs=[chart_input, persona_dropdown], outputs=[analysis_output, optimized_image_path] ).then( generate_pdf, inputs=[optimized_image_path, analysis_output, persona_dropdown], outputs=pdf_file ).then( upload_to_github, inputs=pdf_file, outputs=pdf_status ) if __name__ == "__main__": demo.launch()