Create app/arbitrage_analyzer.py

app/arbitrage_analyzer.py

@@ -0,0 +1,63 @@
+"""
+The Arbitrage Analyzer Engine.
+
+Uses Gemini to provide risk assessment and strategic plans for
+arbitrage opportunities detected by the PriceFetcher.
+"""
+import os
+import logging
+from typing import Optional, Dict
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+class ArbitrageAnalyzer:
+    API_URL = "https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-pro-latest:generateContent"
+
+    def __init__(self, client: httpx.AsyncClient, api_key: Optional[str] = None):
+        self.client = client
+        self.api_key = api_key or os.getenv("GEMINI_API_KEY")
+        if not self.api_key:
+            raise ValueError("GEMINI_API_KEY is not set.")
+        self.params = {"key": self.api_key}
+        self.headers = {"Content-Type": "application/json"}
+
+    def _build_prompt(self, opportunity: Dict) -> dict:
+        return {
+            "contents": [{
+                "parts": [{
+                    "text": f"""
+                    You are a quantitative analyst for a high-frequency trading firm.
+                    An arbitrage opportunity has been detected for Bitcoin (BTC).
+                    Provide a concise "Alpha Briefing" as a single, minified JSON object with NO markdown formatting.
+
+                    The JSON object must have these exact keys: "risk", "strategy", "profit_usd".
+
+                    - "risk": Assess the execution risk. MUST be one of "LOW", "MEDIUM", "HIGH". Consider exchange reliability, withdrawal times, and market volatility.
+                    - "strategy": A very brief, one-sentence action plan.
+                    - "profit_usd": Calculate the estimated net profit in USD for trading 1 BTC, assuming a total of 0.2% in fees (0.1% per trade).
+
+                    OPPORTUNITY DETAILS:
+                    - Buy Exchange: {opportunity['buy_exchange']}
+                    - Buy Price: ${opportunity['buy_price']:,.2f}
+                    - Sell Exchange: {opportunity['sell_exchange']}
+                    - Sell Price: ${opportunity['sell_price']:,.2f}
+                    """
+                }]
+            }]
+        }
+
+    async def get_alpha_briefing(self, opportunity: Dict) -> Optional[Dict]:
+        prompt = self._build_prompt(opportunity)
+        try:
+            response = await self.client.post(self.API_URL, json=prompt, params=self.params, headers=self.headers, timeout=20)
+            response.raise_for_status()
+            content = response.json()["candidates"][0]["content"]["parts"][0]["text"]
+            # A simple trick to remove markdown ```json ... ``` wrappers
+            if content.startswith("```"):
+                content = content.strip("```json\n")
+            return json.loads(content)
+        except Exception as e:
+            logger.error(f"❌ Gemini Alpha Briefing Error: {e}")
+            return None