Axel HocksandClaude Opus 4.8
75d28827e8
Headless FastAPI-Backend (server.py + core/engine.py) mit Mobile-PWA (web/), Strategie-/Backtest-Suite und Doku. Secrets, DB, Logs und Laufzeit-State sind via .gitignore ausgeschlossen; Config-Vorlage: oil_widget_config.ini.example. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
344 lines
13 KiB
Python
344 lines
13 KiB
Python
"""
|
||
core/ai.py — KI-Marktanalyse via OpenAI ChatGPT
|
||
=================================================
|
||
Nutzt `gpt-4o-mini-search-preview` (oder das größere `gpt-4o-search-preview`)
|
||
für eine Crude-Oil-Marktanalyse (WTI + Brent) mit Web-Suche.
|
||
|
||
Parsed eine strukturierte Antwort:
|
||
SENTIMENT: bullish|bearish|neutral
|
||
CONFIDENCE: 0-100
|
||
SUMMARY: <2-3 Sätze>
|
||
DRIVERS: <Treiber1> | <Treiber2> | <Treiber3>
|
||
|
||
Loggt jede Analyse in die History-DB (falls injiziert).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
import threading
|
||
import time
|
||
|
||
from core.logger import get_logger
|
||
|
||
log_ai = get_logger("ai")
|
||
log_hist = get_logger("hist")
|
||
|
||
|
||
class OpenAIAnalyzer:
|
||
"""
|
||
Ruft die OpenAI Chat-Completions API auf.
|
||
|
||
Web-Search ist bei *-search-preview-Modellen automatisch aktiv,
|
||
konfigurierbar über `search_context` (low|medium|high).
|
||
"""
|
||
|
||
PROMPT = (
|
||
"Du bist ein professioneller Rohstoff-Marktanalyst. Suche im Web nach den "
|
||
"wichtigsten Crude-Oil-News der letzten 24 Stunden — sowohl für "
|
||
"WTI (US Crude) als auch Brent (OPEC, US-Lagerbestände/EIA, "
|
||
"Geopolitik Naher Osten, Förderdaten, Nachfrage-Indikatoren, "
|
||
"Pipelines, Raffinerien).\n\n"
|
||
"Antworte AUSSCHLIESSLICH in genau diesem Format (deutsche Sprache):\n"
|
||
"SENTIMENT: bullish|bearish|neutral\n"
|
||
"CONFIDENCE: <0-100>\n"
|
||
"SUMMARY: <2-3 Sätze, prägnant, nur preisbewegende Faktoren>\n"
|
||
"DRIVERS: <Treiber1> | <Treiber2> | <Treiber3>\n\n"
|
||
"Sei knapp und konkret. Keine Disclaimer, keine Einleitung."
|
||
)
|
||
|
||
# Preis pro Mio Token (USD) – grobe Schätzung für Cost-Anzeige
|
||
PRICES = {
|
||
"gpt-4o-mini-search-preview": (0.15, 0.60, 25.0),
|
||
"gpt-4o-search-preview": (2.50, 10.00, 30.0),
|
||
"_default": (0.15, 0.60, 25.0),
|
||
}
|
||
|
||
def __init__(self, api_key: str, model: str, search_context: str = "low"):
|
||
self.api_key = api_key.strip()
|
||
self.model = model.strip() or "gpt-4o-mini-search-preview"
|
||
self.search_context = search_context.strip().lower() or "low"
|
||
self.sentiment = None
|
||
self.confidence = None
|
||
self.summary = ""
|
||
self.drivers = []
|
||
self.last_update = None
|
||
self.last_cost = None
|
||
self.error = None
|
||
self.busy = False
|
||
self._lock = threading.Lock()
|
||
# History-Logger wird vom main() injiziert
|
||
self.history = None
|
||
|
||
def is_configured(self) -> bool:
|
||
return bool(self.api_key) and self.api_key.startswith(("sk-", "sk-proj-"))
|
||
|
||
def _estimate_cost(self, in_tok: int, out_tok: int) -> float:
|
||
in_p, out_p, search_p = self.PRICES.get(self.model, self.PRICES["_default"])
|
||
token_cost = (in_tok * in_p + out_tok * out_p) / 1_000_000
|
||
search_cost = search_p / 1000.0
|
||
return token_cost + search_cost
|
||
|
||
def analyze(self):
|
||
if not self.is_configured():
|
||
with self._lock:
|
||
self.error = "Kein API-Key (config.ini)"
|
||
return
|
||
try:
|
||
from openai import OpenAI
|
||
except ImportError:
|
||
with self._lock:
|
||
self.error = "pip install openai"
|
||
return
|
||
|
||
with self._lock:
|
||
if self.busy:
|
||
return
|
||
self.busy = True
|
||
self.error = None
|
||
|
||
try:
|
||
client = OpenAI(api_key=self.api_key)
|
||
kwargs = {
|
||
"model": self.model,
|
||
"messages": [{"role": "user", "content": self.PROMPT}],
|
||
}
|
||
if "search-preview" in self.model:
|
||
kwargs["web_search_options"] = {
|
||
"search_context_size": self.search_context,
|
||
}
|
||
else:
|
||
kwargs["max_tokens"] = 700
|
||
|
||
res = client.chat.completions.create(**kwargs)
|
||
text = (res.choices[0].message.content or "").strip()
|
||
|
||
# Strukturierte Antwort parsen
|
||
sentiment = "neutral"
|
||
confidence = 50
|
||
summary = text[:300]
|
||
drivers = []
|
||
for raw in text.split("\n"):
|
||
line = raw.strip()
|
||
low = line.lower()
|
||
if low.startswith("sentiment:"):
|
||
v = low.split(":", 1)[1].strip()
|
||
sentiment = ("bullish" if "bull" in v
|
||
else "bearish" if "bear" in v else "neutral")
|
||
elif low.startswith("confidence:"):
|
||
digits = "".join(c for c in line.split(":", 1)[1] if c.isdigit())[:3]
|
||
if digits:
|
||
confidence = max(0, min(100, int(digits)))
|
||
elif low.startswith("summary:"):
|
||
summary = line.split(":", 1)[1].strip()
|
||
elif low.startswith("drivers:"):
|
||
parts = line.split(":", 1)[1]
|
||
drivers = [d.strip() for d in parts.split("|") if d.strip()][:4]
|
||
|
||
# Token-Verbrauch & Kosten
|
||
usage = getattr(res, "usage", None)
|
||
cost_est = None
|
||
if usage:
|
||
in_tok = getattr(usage, "prompt_tokens", 0) or 0
|
||
out_tok = getattr(usage, "completion_tokens", 0) or 0
|
||
cost_est = self._estimate_cost(in_tok, out_tok)
|
||
|
||
with self._lock:
|
||
self.sentiment = sentiment
|
||
self.confidence = confidence
|
||
self.summary = summary
|
||
self.drivers = drivers
|
||
self.last_update = time.time()
|
||
self.last_cost = cost_est
|
||
self.error = None
|
||
|
||
log_ai.info(f"{sentiment.upper()} ({confidence}%) — {summary[:80]}")
|
||
if cost_est:
|
||
log_ai.info(f"Geschätzte Kosten: ${cost_est:.4f}")
|
||
|
||
if self.history:
|
||
try:
|
||
self.history.log_ai(
|
||
sentiment = sentiment,
|
||
confidence = confidence,
|
||
summary = summary,
|
||
drivers = drivers,
|
||
cost_estimate= cost_est,
|
||
model = self.model,
|
||
)
|
||
except Exception as e:
|
||
log_hist.error(f"log_ai: {e}")
|
||
except Exception as e:
|
||
err = str(e)
|
||
with self._lock:
|
||
self.error = err[:120]
|
||
log_ai.error(f"Fehler: {err}")
|
||
finally:
|
||
with self._lock:
|
||
self.busy = False
|
||
|
||
def snapshot(self):
|
||
with self._lock:
|
||
return {
|
||
"sentiment": self.sentiment,
|
||
"confidence": self.confidence,
|
||
"summary": self.summary,
|
||
"drivers": list(self.drivers),
|
||
"last_update": self.last_update,
|
||
"last_cost": self.last_cost,
|
||
"error": self.error,
|
||
"busy": self.busy,
|
||
"configured": self.is_configured(),
|
||
"model": self.model,
|
||
}
|
||
|
||
|
||
class GeminiAnalyzer:
|
||
"""
|
||
Ruft die Google Gemini API auf (REST, via requests).
|
||
Nutzt Google Search Grounding für aktuelle Web-Daten.
|
||
"""
|
||
|
||
PROMPT = (
|
||
"Du bist ein professioneller Rohstoff-Marktanalyst. Suche im Web nach den "
|
||
"wichtigsten Crude-Oil-News der letzten 24 Stunden — sowohl für "
|
||
"WTI (US Crude) als auch Brent (OPEC, US-Lagerbestände/EIA, "
|
||
"Geopolitik Naher Osten, Förderdaten, Nachfrage-Indikatoren, "
|
||
"Pipelines, Raffinerien).\n\n"
|
||
"Antworte AUSSCHLIESSLICH in genau diesem Format (deutsche Sprache):\n"
|
||
"SENTIMENT: bullish|bearish|neutral\n"
|
||
"CONFIDENCE: <0-100>\n"
|
||
"SUMMARY: <2-3 Sätze, prägnant, nur preisbewegende Faktoren>\n"
|
||
"DRIVERS: <Treiber1> | <Treiber2> | <Treiber3>\n\n"
|
||
"Sei knapp und konkret. Keine Disclaimer, keine Einleitung."
|
||
)
|
||
|
||
_API_BASE = "https://generativelanguage.googleapis.com/v1beta/models"
|
||
|
||
def __init__(self, api_key: str, model: str = "gemini-2.0-flash"):
|
||
self.api_key = api_key.strip()
|
||
self.model = model.strip() or "gemini-2.0-flash"
|
||
self.sentiment = None
|
||
self.confidence = None
|
||
self.summary = ""
|
||
self.drivers = []
|
||
self.last_update = None
|
||
self.error = None
|
||
self.busy = False
|
||
self._lock = threading.Lock()
|
||
self.history = None
|
||
|
||
def is_configured(self) -> bool:
|
||
return bool(self.api_key) and self.api_key.startswith("AIza")
|
||
|
||
def analyze(self):
|
||
if not self.is_configured():
|
||
with self._lock:
|
||
self.error = "Kein Gemini API-Key (config.ini)"
|
||
return
|
||
try:
|
||
import requests as _req
|
||
except ImportError:
|
||
with self._lock:
|
||
self.error = "pip install requests"
|
||
return
|
||
|
||
with self._lock:
|
||
if self.busy:
|
||
return
|
||
self.busy = True
|
||
self.error = None
|
||
|
||
try:
|
||
url = f"{self._API_BASE}/{self.model}:generateContent"
|
||
payload = {
|
||
"contents": [{"role": "user", "parts": [{"text": self.PROMPT}]}],
|
||
"tools": [{"google_search": {}}],
|
||
"generationConfig": {
|
||
"temperature": 0.3,
|
||
"maxOutputTokens": 600,
|
||
},
|
||
}
|
||
# Key im Header statt als URL-Parameter — sonst landet er bei
|
||
# jedem HTTP-Fehler im Klartext in der Log-Fehlermeldung (URL)
|
||
resp = _req.post(
|
||
url,
|
||
headers={"x-goog-api-key": self.api_key},
|
||
json=payload,
|
||
timeout=60,
|
||
)
|
||
resp.raise_for_status()
|
||
result = resp.json()
|
||
|
||
text = ""
|
||
try:
|
||
text = result["candidates"][0]["content"]["parts"][0]["text"].strip()
|
||
except (KeyError, IndexError):
|
||
text = str(result)[:300]
|
||
|
||
sentiment = "neutral"
|
||
confidence = 50
|
||
summary = text[:300]
|
||
drivers = []
|
||
for raw in text.split("\n"):
|
||
line = raw.strip()
|
||
low = line.lower()
|
||
if low.startswith("sentiment:"):
|
||
v = low.split(":", 1)[1].strip()
|
||
sentiment = ("bullish" if "bull" in v
|
||
else "bearish" if "bear" in v else "neutral")
|
||
elif low.startswith("confidence:"):
|
||
digits = "".join(c for c in line.split(":", 1)[1] if c.isdigit())[:3]
|
||
if digits:
|
||
confidence = max(0, min(100, int(digits)))
|
||
elif low.startswith("summary:"):
|
||
summary = line.split(":", 1)[1].strip()
|
||
elif low.startswith("drivers:"):
|
||
parts = line.split(":", 1)[1]
|
||
drivers = [d.strip() for d in parts.split("|") if d.strip()][:4]
|
||
|
||
with self._lock:
|
||
self.sentiment = sentiment
|
||
self.confidence = confidence
|
||
self.summary = summary
|
||
self.drivers = drivers
|
||
self.last_update = time.time()
|
||
self.error = None
|
||
|
||
log_ai.info(f"[Gemini] {sentiment.upper()} ({confidence}%) — {summary[:80]}")
|
||
|
||
if self.history:
|
||
try:
|
||
self.history.log_ai(
|
||
sentiment = sentiment,
|
||
confidence = confidence,
|
||
summary = summary,
|
||
drivers = drivers,
|
||
cost_estimate= None,
|
||
model = self.model,
|
||
)
|
||
except Exception as e:
|
||
log_hist.error(f"[Gemini] log_ai: {e}")
|
||
|
||
except Exception as e:
|
||
err = str(e)
|
||
with self._lock:
|
||
self.error = err[:120]
|
||
log_ai.error(f"[Gemini] Fehler: {err}")
|
||
finally:
|
||
with self._lock:
|
||
self.busy = False
|
||
|
||
def snapshot(self):
|
||
with self._lock:
|
||
return {
|
||
"sentiment": self.sentiment,
|
||
"confidence": self.confidence,
|
||
"summary": self.summary,
|
||
"drivers": list(self.drivers),
|
||
"last_update": self.last_update,
|
||
"last_cost": None,
|
||
"error": self.error,
|
||
"busy": self.busy,
|
||
"configured": self.is_configured(),
|
||
"model": self.model,
|
||
}
|