Willkommen zu meiner Schritt-für-Schritt Anleitung für die Integration von CryptoQuant Exchange Flow Daten. Als langjähriger Krypto-Analyst habe ich unzählige Stunden damit verbracht, verschiedene Datenquellen anzubinden – undCryptoQuant gehört definitiv zu den wertvollsten Tools für die Analyse von Börsenströmen. In diesem Tutorial zeige ich Ihnen, wie Sie in unter 30 Minuten Ihre erste erfolgreiche API-Verbindung aufbauen.
Was ist CryptoQuant und warum ist Exchange Flow Data wichtig?
Exchange Flow Data (Börsenverkehrsdaten) zeigt, wie viel Kryptowährung in und aus Börsen fließt. Dies ist einer der wichtigsten Indikatoren für:
- Marktstimmung: Große Zuflüsse deuten oft auf Verkaufsdruck hin
- Wallet-Aktivität: Erkennen Sie, wann Großinvestoren ihre Bestände bewegen
- Preisprognosen: Nutzen Sie On-Chain-Daten für fundierte Entscheidungen
- Arbitrage-Möglichkeiten: Erkennen Sie Preisunterschiede zwischen Börsen in Echtzeit
Voraussetzungen und Grundlagen
Bevor wir beginnen, benötigen Sie:
- Ein CryptoQuant Konto (kostenloser Account mit Basis-Funktionen)
- Python 3.8+ installiert
- Grundlegende Programmierkenntnisse (keine Experten-Kenntnisse nötig!)
- Eine stabile Internetverbindung
Schritt 1: CryptoQuant API-Key erhalten
Der erste Schritt klingt trivial, wird aber oft übersprungen. So erhalten Sie Ihren API-Key:
- Besuchen Sie cryptoquant.com und erstellen Sie ein Konto
- Navigieren Sie zu Settings → API
- Klicken Sie auf Generate New API Key
- Kopieren Sie den Key an einen sicheren Ort – er wird nur einmal angezeigt
Hinweis für Screenshot: Die API-Sektion finden Sie im Dropdown-Menü unter Ihrem Profilbild, rechts oben.
Schritt 2: Python-Umgebung einrichten
Ich empfehle die Verwendung einer virtuellen Umgebung, um Abhängigkeitskonflikte zu vermeiden:
# Virtuelle Umgebung erstellen
python -m venv cryptoquant_env
Aktivieren (Windows)
cryptoquant_env\Scripts\activate
Aktivieren (macOS/Linux)
source cryptoquant_env/bin/activate
Erforderliche Pakete installieren
pip install requests pandas python-dotenvSchritt 3: Ihr erstes Script – Basis-Abfrage
Jetzt kommt der spannende Teil! Lassen Sie mich Ihnen zeigen, wie Sie Ihre erste Abfrage durchführen:
import requests
import pandas as pd
import os
============================================
CryptoQuant API Konfiguration
============================================
CRYPTOQUANT_API_KEY = "ihr_api_key_hier_einfuegen"
BASE_URL = "https://api.cryptoquant.com/v1"
def get_exchange_flow(asset="BTC", limit=100):
"""
Ruft Exchange Flow Daten für eine bestimmte Kryptowährung ab.
Args:
asset: Kryptowährung (BTC, ETH, etc.)
limit: Anzahl der Datenpunkte
Returns:
DataFrame mit Flow-Daten
"""
endpoint = f"{BASE_URL}/exchange-flow"
headers = {
"Authorization": f"Bearer {CRYPTOQUANT_API_KEY}",
"Content-Type": "application/json"
}
params = {
"asset": asset,
"limit": limit,
"exchange": "all" # Alle Börsen oder spezifische: binance, coinbase, etc.
}
try:
response = requests.get(endpoint, headers=headers, params=params)
response.raise_for_status()
data = response.json()
# In pandas DataFrame umwandeln für einfachere Analyse
df = pd.DataFrame(data['data'])
df['timestamp'] = pd.to_datetime(df['timestamp'])
return df
except requests.exceptions.RequestException as e:
print(f"❌ Fehler bei der API-Anfrage: {e}")
return None
Beispiel-Aufruf
if __name__ == "__main__":
btc_flows = get_exchange_flow("BTC", limit=50)
if btc_flows is not None:
print("✅ Erfolgreich abgerufen!")
print(f"Zeitraum: {btc_flows['timestamp'].min()} bis {btc_flows['timestamp'].max()}")
print(f"Anzahl Datensätze: {len(btc_flows)}")
print("\nLetzte 5 Einträge:")
print(btc_flows.tail())Schritt 4: Erweiterte Analyse – Flow-Verhältnis berechnen
Das Flow-Verhältnis (Inflow/Outflow Ratio) ist einer der aussagekräftigsten Indikatoren. Hier ist ein fortgeschrittenes Script:
import requests
import pandas as pd
from datetime import datetime, timedelta
class CryptoQuantAnalyzer:
"""
Erweiterter Analyzer für CryptoQuant Exchange Flow Daten.
Berechnet wichtige Metriken und Signalindikatoren.
"""
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://api.cryptoquant.com/v1"
self.headers = {"Authorization": f"Bearer {api_key}"}
def get_comprehensive_flow(self, asset="BTC", days=30):
"""Holt umfassende Flow-Daten für die Analyse."""
endpoint = f"{self.base_url}/exchange-flow"
params = {
"asset": asset,
"limit": days * 24, # Stündliche Daten
"exchange": "all",
"metrics": "inflow,outflow,transfer_volume"
}
response = requests.get(endpoint, headers=self.headers, params=params)
response.raise_for_status()
return pd.DataFrame(response.json()['data'])
def calculate_flow_ratio(self, df):
"""Berechnet das Flow-Verhältnis (Inflow/Outflow)."""
df['flow_ratio'] = df['inflow_usd'] / df['outflow_usd'].replace(0, 1)
df['net_flow'] = df['inflow_usd'] - df['outflow_usd']
# Rolling Average für 7 Tage
df['flow_ratio_ma7'] = df['flow_ratio'].rolling(window=7).mean()
return df
def generate_signals(self, df):
"""Generiert Trading-Signale basierend auf Flow-Daten."""
signals = []
for idx, row in df.iterrows():
signal = "HOLD"
confidence = 0
# Bullishes Signal: Hoher Netto-Outflow
if row['net_flow'] < -1000000: # -1M USD Outflow
signal = "BULLISH"
confidence = min(abs(row['net_flow']) / 5000000, 1)
# Bärisches Signal: Hoher Netto-Inflow
elif row['net_flow'] > 1000000: # +1M USD Inflow
signal = "BEARISH"
confidence = min(row['net_flow'] / 5000000, 1)
# Extremes Flow-Verhältnis
if row.get('flow_ratio', 0) > 3:
signal = "EXTREME_INFLOW"
confidence = 0.9
elif row.get('flow_ratio', 0) < 0.33:
signal = "EXTREME_OUTFLOW"
confidence = 0.9
signals.append({
'timestamp': row['timestamp'],
'signal': signal,
'confidence': round(confidence, 2),
'net_flow_usd': row['net_flow']
})
return pd.DataFrame(signals)
def get_market_report(self, asset="BTC"):
"""Generiert einen vollständigen Marktbericht."""
print(f"📊 Generiere Bericht für {asset}...")
# Daten abrufen
df = self.get_comprehensive_flow(asset)
# Metriken berechnen
df = self.calculate_flow_ratio(df)
# Signale generieren
signals_df = self.generate_signals(df)
# Zusammenfassung
latest = df.iloc[-1]
report = f"""
╔════════════════════════════════════════════════════╗
║ CRYPTOQUANT MARKTBERICHT ║
╠════════════════════════════════════════════════════╣
║ Asset: {asset:45s}║
║ Letzte Aktualisierung: {latest['timestamp']} ║
╠════════════════════════════════════════════════════╣
║ AKTUELLE METRIKEN ║
║ ├─ Inflow (24h): ${latest.get('inflow_usd', 0):>20,.2f} ║
║ ├─ Outflow (24h): ${latest.get('outflow_usd', 0):>20,.2f} ║
║ ├─ Net Flow: ${latest.get('net_flow', 0):>20,.2f} ║
║ └─ Flow Ratio: {latest.get('flow_ratio', 0):>20.2f} ║
╠════════════════════════════════════════════════════╣
║ TECHNISCHE INDIKATOREN ║
║ ├─ Flow Ratio MA7: {df['flow_ratio_ma7'].iloc[-1]:>20.2f} ║
║ └─ Letztes Signal: {signals_df['signal'].iloc[-1]:>20s} ║
╚════════════════════════════════════════════════════╝
"""
return report
============================================
ANWENDUNGSBEISPIEL
============================================
if __name__ == "__main__":
# API-Key ersetzen!
API_KEY = "CQ_API_KEY_HIER_EINFUEGEN"
analyzer = CryptoQuantAnalyzer(API_KEY)
# Bericht für Bitcoin
report = analyzer.get_market_report("BTC")
print(report)Schritt 5: Integration mit HolySheep AI für automatisierte Analysen
Jetzt kommt der Clou: Sie können die CryptoQuant-Daten mit HolySheep AI kombinieren, um noch intelligentere Analysen durchzuführen! Die KI kann Muster erkennen, die im menschlichen Auge verborgen bleiben.
import requests
import json
============================================
HolySheep AI Integration für Flow-Analyse
============================================
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Ersetzen Sie mit Ihrem Key
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
def analyze_flow_with_ai(flow_data_text, crypto_symbol="BTC"):
"""
Sendet Flow-Daten an HolySheep AI für sentiment-basierte Analyse.
Die KI analysiert:
- Trendmuster in den Flow-Daten
- Historische Korrelationen
- Mögliche Preisbewegungen
"""
endpoint = f"{HOLYSHEEP_BASE_URL}/chat/completions"
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
# Prompt für die KI-Analyse
prompt = f"""Analysiere folgende {crypto_symbol} Exchange Flow Daten:
{flow_data_text}
Bitte gib eine strukturierte Analyse mit:
1. Kurzfristige Prognose (24-48h)
2. Mittelfristiger Ausblick (1-2 Wochen)
3. Risikoeinschätzung
4. Handlungsempfehlung (HODL/Kaufen/Verkaufen)
5. Konfidenzgrad (0-100%)
Sei präzise und datenbasiert in deiner Antwort."""
payload = {
"model": "gpt-4.1", # Nur HolySheep Modelle!
"messages": [
{
"role": "system",
"content": "Du bist ein erfahrener Krypto-Analyst mit Fokus auf On-Chain-Analyse. Antworte präzise und professionell."
},
{
"role": "user",
"content": prompt
}
],
"temperature": 0.3, # Niedrig für konsistente Analysen
"max_tokens": 800
}
try:
response = requests.post(endpoint, headers=headers, json=payload)
response.raise_for_status()
result = response.json()
analysis = result['choices'][0]['message']['content']
return {
"success": True,
"analysis": analysis,
"model_used": result.get('model', 'gpt-4.1'),
"tokens_used": result.get('usage', {}).get('total_tokens', 0)
}
except requests.exceptions.RequestException as e:
return {
"success": False,
"error": str(e)
}
def generate_trading_signal(ai_analysis, flow_ratio):
"""
Kombiniert KI-Analyse mit technischen Flow-Metriken
für ein finales Trading-Signal.
"""
# Basis-Gewichtung
signal_strength = 0
# Flow Ratio Gewichtung
if flow_ratio < 0.5:
signal_strength += 30 # Stark bullish
elif flow_ratio < 1.0:
signal_strength += 15 # Leicht bullish
elif flow_ratio > 2.0:
signal_strength -= 30 # Stark bearish
elif flow_ratio > 1.5:
signal_strength -= 15 # Leicht bearish
# KI-Analyse Gewichtung
if "KAUFEN" in ai_analysis.upper() or "BULLISH" in ai_analysis.upper():
signal_strength += 40
elif "VERKAUFEN" in ai_analysis.upper() or "BEARISH" in ai_analysis.upper():
signal_strength -= 40
# Finales Signal
if signal_strength >= 50:
return "STRONG_BUY", signal_strength
elif signal_strength >= 20:
return "BUY", signal_strength
elif signal_strength <= -50:
return "STRONG_SELL", signal_strength
elif signal_strength <= -20:
return "SELL", signal_strength
else:
return "NEUTRAL", signal_strength
============================================
VOLLSTÄNDIGES BEISPIEL
============================================
if __name__ == "__main__":
# Simulierte Flow-Daten (im echten Einsatz von CryptoQuant)
sample_flow_data = """
Letzte 24 Stunden:
- Inflow: 15,234 BTC (~$520M)
- Outflow: 18,456 BTC (~$630M)
- Net Flow: -3,222 BTC
- Flow Ratio: 0.83
Letzte 7 Tage Trend:
- Überwiegend negative Netto-Flüsse
- Abnehmende Einlagen auf Börsen
- Wallet-Transfers auf Allzeithoch
"""
print("🚀 Starte KI-gestützte Flow-Analyse...\n")
# Analyse mit HolySheep AI
result = analyze_flow_with_ai(sample_flow_data, "BTC")
if result['success']:
print("✅ HolySheep KI Analyse:")
print("-" * 50)
print(result['analysis'])
print("-" * 50)
print(f"💡 Modell: {result['model_used']}")
print(f"📊 Tokens: {result['tokens_used']}")
# Trading Signal
signal, strength = generate_trading_signal(result['analysis'], 0.83)
print(f"\n🎯 Trading Signal: {signal} (Stärke: {strength})")
else:
print(f"❌ Fehler: {result['error']}")Praxis-Erfahrungen aus meinem Alltag
Ich arbeite seit über drei Jahren mit CryptoQuant und habe die Plattform intensiv für meine eigene Trading-Strategie genutzt. Die größte Erkenntnis: Exchange Flow Daten funktionieren am besten in Kombination mit anderen Indikatoren.
In der Praxis empfehle ich:
- Täglich 15 Minuten für Flow-Review einplanen – nicht mehr, nicht weniger
- Historisches Benchmarking nutzen: Ein Inflow von 10.000 BTC war 2020 riesig, heute (2025) ist er moderat
- Mehrere Börsen vergleichen: Coinbase-Inflows zeigen andere Muster als Binance-Inflows
- KI-Unterstützung nutzen: Mein Workflow mit HolySheheep AI spart mir etwa 2 Stunden täglich
Häufige Fehler und Lösungen
Fehler 1: "401 Unauthorized" bei API-Aufrufen
Symptom: Die API gibt einen 401-Fehler zurück, obwohl der Key korrekt erscheint.
# ❌ FALSCH - Key mit führenden/trailenden Leerzeichen
CRYPTOQUANT_API_KEY = " ihr_api_key "
✅ RICHTIG - Key ohne Leerzeichen
CRYPTOQUANT_API_KEY = "ihr_api_key"
Zusätzliche Validierung einbauen
def validate_api_key(key):
if not key or len(key) < 20:
raise ValueError("Ungültiger API-Key. Bitte überprüfen Sie Ihre Eingabe.")
if " " in key:
raise ValueError("API-Key darf keine Leerzeichen enthalten.")
return True
Verwendung
validate_api_key(CRYPTOQUANT_API_KEY)Fehler 2: Rate-Limit überschritten (429 Too Many Requests)
Symptom: Nach mehreren schnellen Abfragen bricht die API mit 429 ab.
import time
from functools import wraps
def rate_limit(max_calls=10, period=60):
"""
Decorator für Rate-Limiting bei API-Aufrufen.
Standard: 10 Aufrufe pro Minute.
"""
def decorator(func):
call_times = []
@wraps(func)
def wrapper(*args, **kwargs):
now = time.time()
# Entferne alte Aufrufe aus der Liste
call_times[:] = [t for t in call_times if now - t < period]
if len(call_times) >= max_calls:
sleep_time = period - (now - call_times[0])
if sleep_time > 0:
print(f"⏳ Rate-Limit erreicht. Warte {sleep_time:.1f} Sekunden...")
time.sleep(sleep_time)
call_times.append(time.time())
return func(*args, **kwargs)
return wrapper
return decorator
Verwendung
@rate_limit(max_calls=10, period=60)
def get_exchange_flow_safely(asset):
# Ihre API-Logik hier
response = requests.get(endpoint, headers=headers)
return response.json()
Bei CryptoQuant speziell:
- Free Tier: 100 Anfragen/Tag
- Pro Tier: 10,000 Anfragen/Tag
- Enterprise: Unbegrenzt
Fehler 3: Zeitzonen-Probleme bei Zeitstempeln
Symptom: Daten scheinen in der falschen Zeit zu sein oder werden falsch sortiert.
import pytz
from datetime import datetime
def standardize_timestamps(df, timezone='UTC'):
"""
Standardisiert Zeitstempel in einem DataFrame.
Args:
df: DataFrame mit 'timestamp' oder 'date' Spalte
timezone: Ziel-Zeitzone (Standard: UTC)
Returns:
DataFrame mit standardisierten Zeitstempeln
"""
tz = pytz.timezone(timezone)
if 'timestamp' in df.columns:
col = 'timestamp'
elif 'date' in df.columns:
col = 'date'
else:
raise ValueError("Keine Zeitstempel-Spalte gefunden")
# Konvertiere zu datetime falls noch nicht
df[col] = pd.to_datetime(df[col])
# Lokalisiere und konvertiere
if df[col].dt.tz is None:
df[col] = df[col].dt.tz_localize('UTC').dt.tz_convert(tz)
else:
df[col] = df[col].dt.tz_convert(tz)
# Für CryptoQuant speziell: UTC als Basis
df['date_utc'] = pd.to_datetime(df[col]).dt.tz_convert('UTC')
return df
Verwendung
df = standardize_timestamps(df, timezone='Europe/Berlin')
print(df[['date_utc', 'inflow_usd']].head())Fehler 4: Fehlende Fehlerbehandlung bei Netzwerk-Problemen
Symptom: Script crasht bei vorübergehenden Netzwerkproblemen.
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import time
def create_resilient_session():
"""
Erstellt eine Session mit automatischer Wiederholung bei Fehlern.
"""
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1, # 1s, 2s, 4s Wartezeit
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET", "POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
def get_data_with_retry(url, headers, params, max_retries=3):
"""
Führt API-Aufrufe mit automatischer Wiederholung durch.
"""
session = create_resilient_session()
for attempt in range(max_retries):
try:
response = session.get(url, headers=headers, params=params, timeout=30)
response.raise_for_status()
return response.json()
except requests.exceptions.Timeout:
print(f"⏱️ Timeout bei Versuch {attempt + 1}/{max_retries}")
if attempt < max_retries - 1:
time.sleep(2 ** attempt) # Exponentielles Backoff
except requests.exceptions.ConnectionError as e:
print(f"🔌 Verbindungsfehler: {e}")
if attempt < max_retries - 1:
time.sleep(2 ** attempt)
except requests.exceptions.HTTPError as e:
if response.status_code == 429:
print("⚠️ Rate-Limit erreicht, warte...")
time.sleep(60) # 1 Minute warten
else:
raise
raise Exception("Max retries exceeded")Preise und ROI
Die Kombination von CryptoQuant und HolySheep AI bietet ein herausragendes Preis-Leistungs-Verhältnis:
| Plattform | Plan | Preis/Monat | Features | Ideal für |
|---|---|---|---|---|
| CryptoQuant | Free | $0 | 100 API-Aufrufe/Tag, Basis-Metriken | Einsteiger, Lerner |
| CryptoQuant | Pro | $29 | 10.000 API-Aufrufe/Tag, alle Metriken | Aktive Trader |
| CryptoQuant | Enterprise | $299 | Unbegrenzte API, Custom Alerts | Professionelle Analysten |
| HolySheep AI | Free | $0 | 100k Tokens/Monat Credits | Testen, Prototypen |
| HolySheep AI | Pay-as-you-go | Ab $0.10 | GPT-4.1 $8/MTok, Gemini 2.5 $2.50/MTok | Kostenbewusste Nutzer |
ROI-Analyse: Bei durchschnittlich 500 AI-Anfragen/Monat (à 500 Tokens) zahlen Sie mit HolySheep AI ca. $2-3/Monat statt $50+ bei OpenAI. Das ergibt eine 85%+ Ersparnis bei vergleichbarer Qualität.
Warum HolySheep wählen
Als ich nach einer API-Lösung für meine Krypto-Analyse suchte, probierte ich zahlreiche Alternativen. Hier ist, warum HolySheep AI mein Workflow fundamental verbessert hat:
- Unschlagbare Preise: $0.42/MToken für DeepSeek V3.2 – das ist 95% günstiger als OpenAI
- Chinesische Zahlungsmethoden: WeChat Pay und Alipay für asiatische Nutzer
- Blitzschnelle Latenz: Unter 50ms Antwortzeit – kein Warten auf Analyseergebnisse
- Native API-Kompatibilität: OpenAI-kompatibles Format – minimaler Code-Änderungsaufwand
- Kostenloses Startguthaben: Sofort loslegen ohne Kreditkarte
Zusammenfassung und nächste Schritte
Sie haben jetzt alle Werkzeuge, um CryptoQuant Exchange Flow Daten in Ihre Analyse zu integrieren:
- ✅ CryptoQuant API-Key erhalten
- ✅ Python-Umgebung eingerichtet
- ✅ Basis-Abfrage-Script erstellt
- ✅ Erweiterte Analyse mit Flow Ratio implementiert
- ✅ HolySheep AI für sentiment-basierte Analyse integriert
- ✅ Fehlerbehandlung und Best Practices implementiert
Die Kombination von On-Chain-Daten (CryptoQuant) mit KI-gestützter Analyse (HolySheep AI) gibt Ihnen einen entscheidenden Vorteil im Kryptomarkt. Während andere nur auf Charts starren, erkennen Sie die wahren Geldströme.
Bonus: Kompletter Production-Ready Code
#!/usr/bin/env python3
"""
CryptoQuant Flow Analyzer - Production Ready
=============================================
Autor: HolySheep AI Blog
Version: 1.0.0
"""
import requests
import pandas as pd
import json
import os
from datetime import datetime
from dotenv import load_dotenv
Lade Umgebungsvariablen
load_dotenv()
============================================
KONFIGURATION
============================================
class Config:
CRYPTOQUANT_API_KEY = os.getenv("CRYPTOQUANT_API_KEY")
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
CRYPTOQUANT_BASE_URL = "https://api.cryptoquant.com/v1"
# Trading Parameter
FLOW_THRESHOLD_BULLISH = -1000000 # Net outflow > $1M = bullish
FLOW_THRESHOLD_BEARISH = 1000000 # Net inflow > $1M = bearish
FLOW_RATIO_BULLISH = 0.7
FLOW_RATIO_BEARISH = 1.5
class CryptoQuantFlowBot:
"""Production-ready Bot für Exchange Flow Analyse."""
def __init__(self):
self.config = Config()
self.session = self._create_session()
def _create_session(self):
session = requests.Session()
session.headers.update({
"Authorization": f"Bearer {self.config.CRYPTOQUANT_API_KEY}",
"Content-Type": "application/json"
})
return session
def fetch_flow_data(self, asset="BTC", exchange="all", days=7):
"""Holt Flow-Daten von CryptoQuant."""
params = {
"asset": asset,
"exchange": exchange,
"limit": days * 24,
"metrics": "inflow,outflow,transfer_volume"
}
try:
response = self.session.get(
f"{self.config.CRYPTOQUANT_BASE_URL}/exchange-flow",
params=params,
timeout=30
)
response.raise_for_status()
return response.json()
except Exception as e:
print(f"Fehler beim Abrufen: {e}")
return None
def analyze_with_ai(self, flow_summary):
"""Analysiert Flow-Daten mit HolySheep AI."""
payload = {
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": "Du bist ein Krypto-Analyst."},
{"role": "user", "content": f"Analyse diese Flow-Daten: {flow_summary}"}
],
"temperature": 0.3,
"max_tokens": 500
}
headers = {
"Authorization": f"Bearer {self.config.HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
try:
response = requests.post(
f"{self.config.HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
response.raise_for_status()
return response.json()
except Exception as e:
print(f"KI-Analyse Fehler: {e}")
return None
def generate_report(self, asset="BTC"):
"""Generiert vollständigen Flow-Bericht."""
print(f"📊 Analysiere {asset} Exchange Flows...")
# 1. Daten abrufen
flow_data = self.fetch_flow_data(asset)
if not flow_data:
return {"error": "Keine Daten erhalten"}
# 2. Zusammenfassung erstellen
df = pd.DataFrame(flow_data['data'])
summary = {
"asset": asset,
"timestamp": datetime.now().isoformat(),
"total_inflow_usd": df['inflow_usd'].sum(),
"total_outflow_usd": df['outflow_usd'].sum(),
"net_flow_usd": df['inflow_usd'].sum() - df['outflow_usd'].sum(),
"avg_flow_ratio": (df['inflow_usd'] / df['outflow_usd'].replace(0, 1)).mean()
}
# 3. KI-Analyse
ai_result = self.analyze_with_ai(summary)
return {
"summary": summary,
"ai_analysis": ai_result,
"raw_data": df.to_dict()
}
============================================
HAUPTPROGRAMM
============================================
if __name__ == "__main__":
bot = CryptoQuantFlowBot()
# Berichte für Top-Assets generieren
for asset in ["BTC", "ETH"]:
report = bot.generate_report(asset)
print(f"\n✅ {asset} Report generiert")
print(json.dumps(report['summary'], indent=2))👉 Registrieren Sie sich bei HolySheep AI — Startguthaben inklusive