""" core/tu_rating.py — Traders-Union-Analyse (tradersunion.com) ============================================================= Holt die technische Analyse für WTI von der öffentlichen Traders-Union-API (quotes.tradersunion.com) — dieselben Daten, die der Tacho auf https://tradersunion.com/currencies/forecast/wti-crude-oil/signals/ anzeigt. API: GET /api/v3/informer/technical-analysis/detailed/?symbol=WTI/USD Liefert pro Zeitebene (m5, m15, m30, h1, h4, d1, w1): - forecast → Gesamt-Verdikt ("Strong Sell" … "Strong Buy") - ta → Oszillator-Zähler (buy/sell/neutral, 13 Indikatoren) - ma → Moving-Average-Zähler (MA5–MA200, SMA+EMA) - indicators → 14 Einzelindikatoren mit Werten Auto-Trader-Signal (trade_signal): Verdikt je TF → Score (-2 Strong Sell … +2 Strong Buy) LONG wenn m15 ≥ +1 UND m30 ≥ +1 UND h1 nicht dagegen (≥ 0) SHORT spiegelbildlich. Konfidenz steigt mit Strong-Verdikten und H1-/M5-Bestätigung. Daten älter als 5 min → WARTEN (kein Stale-Trading). Kein API-Key nötig. fetch() blockiert (HTTP) — im Hintergrund-Thread rufen; snapshot()/trade_signal() liefern den letzten Stand sofort. """ from __future__ import annotations import threading import time from core.logger import get_logger log = get_logger("turating") _API_URL = ("https://quotes.tradersunion.com/api/v3/" "informer/technical-analysis/detailed/") _UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/125.0 Safari/537.36") # Zeitebenen wie auf der Website: Anzeige-Key → API-Key INTERVALS: dict[str, str] = { "5m": "m5", "15m": "m15", "30m": "m30", "1h": "h1", "4h": "h4", "1d": "d1", "1w": "w1", } # Verdikt → numerischer Score (für Nadel + Trading-Logik) _VERDICT_SCORE = { "strong sell": -2, "sell": -1, "neutral": 0, "buy": 1, "strong buy": 2, } _STALE_S = 300 # Daten älter als 5 min → kein Trading-Signal def _vscore(verdict: str | None) -> int: return _VERDICT_SCORE.get((verdict or "").strip().lower(), 0) def _counts_rating(c: dict) -> float | None: """Nadel-Position -1..+1 aus Buy/Sell/Neutral-Zählern.""" total = (c.get("buy") or 0) + (c.get("sell") or 0) + (c.get("neutral") or 0) if not total: return None return ((c.get("buy") or 0) - (c.get("sell") or 0)) / total class TradersUnionProvider: """Holt + cached die Traders-Union-Analyse für ein Symbol.""" def __init__(self, symbol: str = "WTI/USD"): self.symbol = symbol self._lock = threading.Lock() self._busy = False self._data: dict = {} # {interval_key: {...}} self._error: str | None = None self._last_update: float | None = None # ── HTTP-Fetch (blockierend — im Hintergrund-Thread aufrufen) ─────────── def fetch(self) -> bool: with self._lock: if self._busy: return False self._busy = True try: import requests resp = requests.get( _API_URL, params={"symbol": self.symbol}, headers={"User-Agent": _UA, "Accept": "application/json"}, timeout=15) resp.raise_for_status() payload = resp.json().get("data") or {} data = {} for key, api_key in INTERVALS.items(): tf = payload.get(api_key) if not isinstance(tf, dict): continue osc = tf.get("ta") or {} ma = tf.get("ma") or {} c_osc = {"buy": osc.get("buy", 0), "sell": osc.get("sell", 0), "neutral": osc.get("neutral", 0)} c_ma = {"buy": ma.get("buy", 0), "sell": ma.get("sell", 0), "neutral": ma.get("neutral", 0)} total = {k: c_osc[k] + c_ma[k] for k in c_osc} data[key] = { "verdict": tf.get("forecast") or "—", "verdict_osc": osc.get("forecast") or "—", "verdict_ma": ma.get("forecast") or "—", "rating": _counts_rating(total), "rating_osc": _counts_rating(c_osc), "rating_ma": _counts_rating(c_ma), "counts": total, "counts_osc": c_osc, "counts_ma": c_ma, "score": _vscore(tf.get("forecast")), } if not data: raise ValueError(f"keine TF-Daten für {self.symbol}") with self._lock: self._data = data self._error = None self._last_update = time.time() log.info(f"{self.symbol}: " + " ".join( f"{k}={d['verdict']}" for k, d in data.items())) return True except Exception as e: with self._lock: self._error = str(e)[:120] log.warning(f"TU-Analyse Fetch fehlgeschlagen: {e}") return False finally: with self._lock: self._busy = False # ── Snapshot für UI (Tachos) ───────────────────────────────────────────── def snapshot(self) -> dict: with self._lock: return { "symbol": self.symbol, "intervals": dict(self._data), "error": self._error, "last_update": self._last_update, "busy": self._busy, } # ── Trading-Signal für den Auto-Trader ─────────────────────────────────── def trade_signal(self) -> dict: """ Konsens-Signal aus der Traders-Union-Analyse: LONG: m15 ≥ +1 und m30 ≥ +1 und h1 ≥ 0 SHORT: m15 ≤ −1 und m30 ≤ −1 und h1 ≤ 0 Konfidenz: 60 % Basis, +10 je Strong-Verdikt (m15/m30), +10 bei H1-Bestätigung, +5 bei M5-Bestätigung (max 90). """ with self._lock: data = dict(self._data) ts = self._last_update base = {"signal": "WARTEN", "conf_pct": 0, "score": 0.0, "setup": "TU_KONSENS", "regime": None, "rsi": None, "reasons": []} if not data: base["reasons"] = ["keine TU-Daten"] return base if not ts or time.time() - ts > _STALE_S: base["reasons"] = ["TU-Daten veraltet"] return base sc = {k: data.get(k, {}).get("score", 0) for k in ("5m", "15m", "30m", "1h")} reasons = [f"{k}: {data[k]['verdict']}" for k in ("5m", "15m", "30m", "1h", "4h", "1d") if k in data] signal = "WARTEN" if sc["15m"] >= 1 and sc["30m"] >= 1 and sc["1h"] >= 0: signal = "LONG" elif sc["15m"] <= -1 and sc["30m"] <= -1 and sc["1h"] <= 0: signal = "SHORT" conf = 0 if signal != "WARTEN": d = 1 if signal == "LONG" else -1 # Basis 60: mit H1-Bestätigung (+10) erreicht ein sauberer # Konsens die Auto-Schwelle (70) auch ohne Strong-Verdikt conf = 60 conf += 10 * sum(1 for k in ("15m", "30m") if sc[k] * d >= 2) if sc["1h"] * d >= 1: conf += 10 if sc["5m"] * d >= 1: conf += 5 conf = min(conf, 90) # Score -1..+1 (Mittel der Trading-TFs, normiert auf ±2) score = (sc["15m"] + sc["30m"] + sc["1h"]) / 6.0 return {"signal": signal, "conf_pct": conf, "score": score, "setup": "TU_KONSENS", "regime": None, "rsi": None, "reasons": reasons}