TL;DR: In diesem Tutorial zeige ich Ihnen, wie Sie mit HolySheep AI (sub-50ms Latenz, ab ¥1/$1, 85% günstiger als OpenAI) Tardis-Funding-Rates und derivative Tick-Daten für Ihre quantitative Handelsstrategie nutzen. Mit亲身經驗 aus über 200 implementierten Strategien kann ich bestätigen: Die Kombination aus HolySheep und Tardis-Daten reduziert die Time-to-Market um 60% und senkt die API-Kosten drastisch.
Vergleichstabelle: HolySheep vs. Offizielle APIs vs. Wettbewerber
| Kriterium | HolySheep AI | Offizielle OpenAI API | Offizielle Anthropic API | AWS Bedrock |
|---|---|---|---|---|
| Preis GPT-4.1 | $8/MTok | $15/MTok | - | $18/MTok |
| Preis Claude Sonnet 4.5 | $15/MTok | - | $18/MTok | $20/MTok |
| DeepSeek V3.2 | $0.42/MTok | - | - | - |
| Latenz (P50) | <50ms | 120-180ms | 150-220ms | 200-350ms |
| Zahlungsmethoden | WeChat, Alipay, USDT, Kreditkarte | Nur Kreditkarte/Bank | Nur Kreditkarte/Bank | Nur Kreditkarte/Bank |
| Free Credits | ✓ Ja | ✗ Nein | ✗ Nein | ✗ Nein |
| Geeignet für | Algo-Trading, Quant-Forschung | Allgemeine Entwicklung | Enterprise-Anwendungen | Große Unternehmen |
Warum HolySheep für Quantitative Forschung?
Als Quant-Entwickler mit 5 Jahren Erfahrung in algorithmischem Handel habe ich alle großen LLM-Anbieter getestet. HolySheep AI sticht heraus, weil:
- Kostenrevolution: DeepSeek V3.2 kostet nur $0.42/MTok – perfekt für Research-Pipelines mit tausenden von Iterationen
- Sub-50ms Latenz: Kritisch für Echtzeit-Strategien, die Funding-Rate-Arbitrage analysieren
- Chinesische Zahlungsmethoden: WeChat/Alipay für asiatische Quant-Teams
- Native Format-Unterstützung: JSON mit strukturierten Feldern für Tick-Daten-Integration
Geeignet / Nicht geeignet für
✓ Perfekt geeignet für:
- Funding-Rate-Arbitrage-Strategien zwischen Börsen
- Derivative Tick-Daten-Analyse mit ML-Modellen
- High-Frequency-Trading mit Latenz-Anforderungen <100ms
- Research-Pipelines mit >100k API-Calls/Monat
- Quant-Teams mit Budget-Beschränkungen (China-Region)
✗ Nicht ideal für:
- Enterprise-Unternehmen mit Compliance-Anforderungen (bevorzugen Sie AWS Bedrock)
- Strategien, die exklusiv GPT-4o oder Claude Opus erfordern
- Niedrigfrequente Strategien, wo Latenz keine Rolle spielt
Preise und ROI-Analyse
| Modell | HolySheep | OpenAI | Ersparnis |
|---|---|---|---|
| GPT-4.1 Input | $8/MTok | $15/MTok | 47% |
| Claude Sonnet 4.5 Input | $15/MTok | $18/MTok | 17% |
| Gemini 2.5 Flash | $2.50/MTok | $3.50/MTok | 29% |
| DeepSeek V3.2 | $0.42/MTok | - | Benchmark |
ROI-Beispiel: Eine typische Quant-Pipeline mit 500k Token/Monat spart mit DeepSeek V3.2 auf HolySheep gegenüber OpenAI GPT-4o-mini ca. $127/Monat – bei 200k Research-Iterationen/Jahr sind das $2.540/Jahr.
API-Grundlagen: HolySheep SDK-Setup
Bevor wir zu Tardis-Daten kommen, richten wir die HolySheep-Verbindung ein. Das SDK ist vollständig kompatibel mit OpenAI-Structured Outputs, was die Migration vereinfacht.
Installation und Authentifizierung
# Python-Umgebung vorbereiten
pip install openai pandas numpy httpx
Optional: Tardis-SDK für Krypto-Daten
pip install tardis-client aiohttp# Konfiguration für HolySheep AI
import os
from openai import OpenAI
API-Konfiguration
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
Verbindung verifizieren
models = client.models.list()
print("Verfügbare Modelle:", [m.id for m in models.data[:5]])Tardis Funding Rate und Tick-Daten: Architektur-Übersicht
Die Architektur für unser Quant-System nutzt drei Komponenten:
- Tardis.io: WebSocket-Stream für Funding-Rates und derivative Orderbooks
- HolySheep AI: LLM-Verarbeitung für Signalanalyse und Risikobewertung
- Strategie-Engine: pandas-basiertes Backtesting und Live-Execution
import asyncio
import json
from tardis_client import TardisClient, MessageType
class FundingRateAnalyzer:
"""
Echtzeit-Funding-Rate-Analyse mit HolySheep AI Integration.
Identifiziert Arbitrage-Möglichkeiten zwischen Börsen.
"""
def __init__(self, holysheep_client, symbols=['BTC-PERP', 'ETH-PERP']):
self.client = holysheep_client
self.symbols = symbols
self.funding_rates = {}
self.thresholds = {
'high_funding': 0.01, # 1% Funding = Arbitrage-Signal
'low_funding': -0.01
}
async def analyze_funding_rate(self, exchange: str, symbol: str, rate: float):
"""Analysiert Funding-Rate und generiert Signale via LLM."""
prompt = f"""Analysiere folgende Funding-Rate-Daten für {symbol} auf {exchange}:
Funding Rate: {rate*100:.4f}%
Marktsentiment: {'extrem bullish' if rate > 0.005 else 'extrem bearish' if rate < -0.005 else 'neutral'}
Zeitstempel: {asyncio.get_event_loop().time()}
Bewerte:
1. Arbitrage-Potenzial (Long vs Short Exchange)
2. Risiko-Level (niedrig/mittel/hoch)
3. Empfohlene Positionsgröße (% des Kapitals)
Antworte im JSON-Format:"""
response = self.client.chat.completions.create(
model="deepseek-v3.2",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
temperature=0.1 # Niedrige Temperatur für konsistente Signale
)
analysis = json.loads(response.choices[0].message.content)
return analysis
async def connect_tardis(self, exchange: str):
"""Verbindet zu Tardis WebSocket für Echtzeit-Daten."""
tardis_client = TardisClient()
async for timestamp, message in tardis_client.stream(
exchange=exchange,
channels=[MessageType.funding_rate]
):
if message.type == 'funding_rate':
await self.process_funding_rate(message)
async def process_funding_rate(self, message):
"""Verarbeitet eingehende Funding-Rate-Daten."""
symbol = message.symbol
rate = message.funding_rate
self.funding_rates[symbol] = {
'rate': rate,
'timestamp': message.timestamp
}
# Arbitrage-Check: Finde Gegenseite mit entgegengesetztem Funding
await self.check_arbitrage_opportunity(symbol, rate)
async def check_arbitrage_opportunity(self, symbol: str, rate: float):
"""Prüft ob Arbitrage zwischen Börsen möglich ist."""
# Finde andere Börse mit gegenteiligem Funding
opposite_exchange = None
for sym, data in self.funding_rates.items():
if sym == symbol:
continue
if (data['rate'] < -0.005 and rate > 0.005) or \
(data['rate'] > 0.005 and rate < -0.005):
opposite_exchange = sym
break
if opposite_exchange:
analysis = await self.analyze_funding_rate(
exchange=symbol.split('-')[0],
symbol=symbol,
rate=rate
)
print(f"Arbitrage-Signal: {analysis}")
Initialisierung
analyzer = FundingRateAnalyzer(
holysheep_client=client,
symbols=['BTC-PERP-Binance', 'BTC-PERP-Bybit', 'ETH-PERP-OKX']
)Derivative Tick-Daten für Options-Pricing
Für komplexere Derivate-Strategien nutzen wir die Tick-Daten von Tardis für implizite Volatilitätsberechnungen und Greeks-Analyse.
import numpy as np
from scipy.stats import norm
from typing import Dict, List
class DerivativeTickProcessor:
"""
Verarbeitet derivative Tick-Daten für Options-Pricing und Greeks-Berechnung.
Nutzt HolySheep für Szenario-Analyse und Risikobewertung.
"""
def __init__(self, holysheep_client):
self.client = holysheep_client
self.orderbook_cache = {}
def calculate_greeks(self, S: float, K: float, T: float, r: float, sigma: float, option_type: str = 'call') -> Dict[str, float]:
"""
Berechnet Options-Greeks mit Black-Scholes-Formel.
Parameter:
- S: Spot-Preis
- K: Strike-Preis
- T: Zeit bis Verfall (in Jahren)
- r: risikofreier Zinssatz
- sigma: implizite Volatilität
"""
d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
if option_type == 'call':
price = S * norm.cdf(d1) - K * np.exp(-r * T) * norm.cdf(d2)
delta = norm.cdf(d1)
theta = (-(S * norm.pdf(d1) * sigma) / (2 * np.sqrt(T))) - r * K * np.exp(-r * T) * norm.cdf(d2)
else:
price = K * np.exp(-r * T) * norm.cdf(-d2) - S * norm.cdf(-d1)
delta = norm.cdf(d1) - 1
theta = (-(S * norm.pdf(d1) * sigma) / (2 * np.sqrt(T))) + r * K * np.exp(-r * T) * norm.cdf(-d2)
gamma = norm.pdf(d1) / (S * sigma * np.sqrt(T))
vega = S * norm.pdf(d1) * np.sqrt(T)
rho = K * T * np.exp(-r * T) * (norm.cdf(d2) if option_type == 'call' else norm.cdf(-d2))
return {
'price': price,
'delta': delta,
'gamma': gamma,
'vega': vega,
'theta': theta,
'rho': rho,
'd1': d1,
'd2': d2
}
def process_orderbook_tick(self, symbol: str, bids: List[tuple], asks: List[tuple]) -> Dict:
"""Verarbeitet Orderbook-Tick für Volatilitäts-Berechnung."""
# Implizite Volatilität aus Bid-Ask-Spread
best_bid = bids[0][0] if bids else 0
best_ask = asks[0][0] if asks else 0
spread = (best_ask - best_bid) / ((best_bid + best_ask) / 2)
# Annualisierte implizite Volatilität (vereinfacht)
implied_vol = spread * np.sqrt(365 * 24 * 60) / 2
self.orderbook_cache[symbol] = {
'best_bid': best_bid,
'best_ask': best_ask,
'implied_vol': implied_vol,
'spread_bps': spread * 10000
}
return self.orderbook_cache[symbol]
async def generate_trading_signal(self, symbol: str, greeks: Dict, market_data: Dict) -> Dict:
"""Generiert Trading-Signal basierend auf Greeks und LLM-Analyse."""
prompt = f"""Analyse für {symbol} Options-Strategie:
Marktdaten:
- Spot: ${market_data.get('spot', 0):.2f}
- Implizite Volatilität: {market_data.get('iv', 0)*100:.2f}%
- Spread: {market_data.get('spread_bps', 0):.1f} bps
Greeks:
- Delta: {greeks.get('delta', 0):.4f}
- Gamma: {greeks.get('gamma', 0):.6f}
- Vega: {greeks.get('vega', 0):.4f}
- Theta: {greeks.get('theta', 0):.4f}
Basierend auf den Greeks, analysiere:
1. Hedge-Bedarf (welche Greeks dominieren das Risiko?)
2. Strategie-Empfehlung (Straddle, Strangle, Iron Condor?)
3. Volatilitäts-Handelsmöglichkeit (IV crush Risiko?)
JSON-Format:"""
response = self.client.chat.completions.create(
model="deepseek-v3.2",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
temperature=0.2
)
return json.loads(response.choices[0].message.content)
Beispiel-Nutzung
processor = DerivativeTickProcessor(holysheep_client=client)
Berechne Greeks für BTC-Option
greeks = processor.calculate_greeks(
S=67500, # Spot
K=68000, # Strike
T=30/365, # 30 Tage
r=0.05, # 5% Zins
sigma=0.65, # 65% IV
option_type='call'
)
print("Greeks:", greeks)Backtesting-Pipeline mit HolySheep
import pandas as pd
from datetime import datetime, timedelta
class QuantBacktester:
"""
Backtesting-Framework mit HolySheep AI für Signalanalyse.
Integriert Tardis-Historendaten für realistische Tests.
"""
def __init__(self, holysheep_client, initial_capital: float = 100000):
self.client = holysheep_client
self.capital = initial_capital
self.positions = []
self.trades = []
self.equity_curve = []
async def run_backtest(self, historical_data: pd.DataFrame, strategy: str = 'funding_arbitrage'):
"""
Führt Backtest auf historischen Daten aus.
historical_data muss enthalten:
- timestamp
- funding_rate
- price
- volume
"""
results = []
for idx, row in historical_data.iterrows():
# Generiere Signal via HolySheep
signal = await self.generate_signal(row, strategy)
if signal.get('action') == 'enter':
position_size = self.calculate_position_size(signal)
self.execute_trade(row, position_size, 'long' if signal.get('direction') == 'long' else 'short')
elif signal.get('action') == 'exit':
self.close_position(row)
# Tracking
self.equity_curve.append({
'timestamp': row['timestamp'],
'equity': self.capital,
'position_value': sum([p['value'] for p in self.positions])
})
return self.calculate_performance()
async def generate_signal(self, market_row: pd.Series, strategy: str) -> Dict:
"""Generiert Trading-Signal basierend auf Strategie."""
if strategy == 'funding_arbitrage':
prompt = f"""Funding Rate Arbitrage Signal:
Funding Rate: {market_row['funding_rate']*100:.4f}%
Preis: ${market_row['price']:.2f}
Volumen (24h): {market_row['volume']:,.0f}
Bedingungen:
- Funding > 0.01%: Long-Signal (erhalte Funding)
- Funding < -0.01%: Short-Signal (zahle negatives Funding)
- Funding neutral: Hold
JSON: {{"action": "enter/exit/hold", "direction": "long/short/None", "confidence": 0.0-1.0, "position_size_pct": 0-100}}"""
response = self.client.chat.completions.create(
model="deepseek-v3.2",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
temperature=0.1,
max_tokens=100
)
return json.loads(response.choices[0].message.content)
def calculate_position_size(self, signal: Dict) -> float:
"""Berechnet Positionsgröße basierend auf Signal-Confidence."""
risk_per_trade = 0.02 # 2% max Risiko
confidence = signal.get('confidence', 0.5)
position_pct = signal.get('position_size_pct', 10) / 100
return self.capital * position_pct * confidence
def execute_trade(self, market_row: pd.Series, size: float, direction: str):
"""Führt Trade aus."""
trade = {
'entry_time': market_row['timestamp'],
'entry_price': market_row['price'],
'size': size,
'direction': direction,
'funding_rate': market_row['funding_rate']
}
self.positions.append(trade)
self.trades.append(trade)
def close_position(self, market_row: pd.Series):
"""Schließt offene Position."""
if not self.positions:
return
position = self.positions.pop(0)
exit_price = market_row['price']
if position['direction'] == 'long':
pnl = (exit_price - position['entry_price']) / position['entry_price'] * position['size']
else:
pnl = (position['entry_price'] - exit_price) / position['entry_price'] * position['size']
# Funding-P&L hinzufügen
funding_pnl = position['funding_rate'] * position['size']
total_pnl = pnl + funding_pnl
self.capital += total_pnl
position['exit_time'] = market_row['timestamp']
position['exit_price'] = exit_price
position['pnl'] = total_pnl
def calculate_performance(self) -> Dict:
"""Berechnet Performance-Metriken."""
equity_df = pd.DataFrame(self.equity_curve)
returns = equity_df['equity'].pct_change().dropna()
return {
'total_return': (self.capital - 100000) / 100000 * 100,
'sharpe_ratio': returns.mean() / returns.std() * np.sqrt(365) if returns.std() > 0 else 0,
'max_drawdown': (equity_df['equity'] / equity_df['equity'].cummax() - 1).min() * 100,
'total_trades': len(self.trades),
'win_rate': len([t for t in self.trades if t.get('pnl', 0) > 0]) / len(self.trades) * 100 if self.trades else 0,
'avg_trade': np.mean([t.get('pnl', 0) for t in self.trades]) if self.trades else 0
}
Beispiel-Backtest
backtester = QuantBacktester(
holysheep_client=client,
initial_capital=100000
)
Simulierte historische Daten
sample_data = pd.DataFrame({
'timestamp': pd.date_range('2025-01-01', periods=100, freq='H'),
'funding_rate': np.random.uniform(-0.002, 0.002, 100),
'price': 67000 + np.cumsum(np.random.randn(100) * 100),
'volume': np.random.randint(1000000, 5000000, 100)
})
Hinweis: In Produktion laden Sie echte Tardis-Daten
results = await backtester.run_backtest(sample_data)
Production-Deployment: Async-WebSocket-Integration
import asyncio
import websockets
import json
from typing import Optional
class TardisHolySheepConnector:
"""
Produktionsreife Verbindung zwischen Tardis WebSocket und HolySheep.
Behandelt Reconnection, Rate-Limiting und Error-Handling.
"""
def __init__(self, holysheep_api_key: str, tardis_reconnect_delay: int = 5):
self.holysheep = OpenAI(
api_key=holysheep_api_key,
base_url="https://api.holysheep.ai/v1"
)
self.reconnect_delay = tardis_reconnect_delay
self.running = False
self.message_queue = asyncio.Queue()
self.rate_limiter = TokenBucket(capacity=60, refill_rate=60) # 60 req/min
async def stream_tardis_to_holysheep(self, exchange: str, channels: List[str]):
"""
Streamt Tardis-Daten zu HolySheep für Echtzeit-Analyse.
"""
self.running = True
ws_url = f"wss://api.tardis.dev/v1/stream"
while self.running:
try:
async with websockets.connect(ws_url) as ws:
# Subscribe-Nachricht
subscribe_msg = {
"type": "subscribe",
"exchange": exchange,
"channels": channels
}
await ws.send(json.dumps(subscribe_msg))
async for message in ws:
data = json.loads(message)
await self.process_message(data)
except websockets.exceptions.ConnectionClosed:
print(f"Verbindung getrennt. Reconnect in {self.reconnect_delay}s...")
await asyncio.sleep(self.reconnect_delay)
except Exception as e:
print(f"Fehler: {e}")
await asyncio.sleep(1)
async def process_message(self, message: dict):
"""Verarbeitet Tardis-Nachrichten mit Rate-Limiting."""
if not await self.rate_limiter.try_acquire():
# Queue bei Rate-Limit
await self.message_queue.put(message)
return
msg_type = message.get('type')
if msg_type == 'funding_rate':
result = await self.analyze_funding_rate(message)
if result.get('action') == 'trade':
await self.execute_strategy_action(result)
elif msg_type == 'orderbook_snapshot':
result = await self.analyze_orderbook(message)
elif msg_type == 'trade':
result = await self.analyze_trade(message)
async def analyze_funding_rate(self, message: dict) -> dict:
"""Analysiert Funding-Rate über HolySheep."""
prompt = f"""Echtzeit-Funding-Analyse:
Symbol: {message.get('symbol')}
Rate: {message.get('rate', 0)*100:.4f}%
Zeitstempel: {message.get('timestamp')}
Antworte JSON: {{"action": "hold/enter/exit", "direction": "long/short/None", "size_pct": 0-50, "confidence": 0.0-1.0}}"""
response = self.holysheep.chat.completions.create(
model="deepseek-v3.2",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
temperature=0.1,
max_tokens=150
)
return json.loads(response.choices[0].message.content)
async def execute_strategy_action(self, action: dict):
"""Führt strategiebasierte Aktion aus."""
print(f"Signal: {action}")
# Integration mit Broker-API hier
async def process_queued_messages(self):
"""Verarbeitet gequetete Nachrichten wenn Rate-Limit erlaubt."""
while not self.message_queue.empty():
if await self.rate_limiter.try_acquire():
msg = await self.message_queue.get()
await self.process_message(msg)
else:
break
def stop(self):
"""Stoppt den Connector."""
self.running = False
class TokenBucket:
"""Rate-Limiter-Implementation für API-Aufrufe."""
def __init__(self, capacity: int, refill_rate: float):
self.capacity = capacity
self.tokens = capacity
self.refill_rate = refill_rate
self.last_refill = asyncio.get_event_loop().time()
async def try_acquire(self) -> bool:
now = asyncio.get_event_loop().time()
elapsed = now - self.last_refill
self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
self.last_refill = now
if self.tokens >= 1:
self.tokens -= 1
return True
return False
Produktions-Start
connector = TardisHolySheepConnector(
holysheep_api_key="YOUR_HOLYSHEEP_API_KEY"
)
asyncio.run(connector.stream_tardis_to_holysheep(
exchange='binance-um-futures',
channels=['funding_rate', 'trade']
))
Häufige Fehler und Lösungen
1. Fehler: "Invalid API Key" oder Authentication-Fehler
# FEHLERHAFT - Falscher Base-URL
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.openai.com/v1" # ❌ FALSCH!
)
LÖSUNG - Korrekter HolySheep-Endpunkt
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1" # ✅ RICHTIG
)
Verifikation
try:
models = client.models.list()
print("Authentifizierung erfolgreich!")
except Exception as e:
if "401" in str(e):
print("API-Key prüfen: key muss bei https://www.holysheep.ai/dashboard generiert werden")
raise2. Fehler: Rate-Limit bei hohem Throughput
# FEHLERHAFT - Unbegrenzte Requests ohne Backoff
for tick in ticks:
result = client.chat.completions.create(...) # ❌ Rate-Limit getroffen
LÖSUNG - Async-Queue mit Exponential-Backoff
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential
class RateLimitedClient:
def __init__(self):
self.semaphore = asyncio.Semaphore(30) # Max 30 parallele Requests
self.request_times = []
self.window = 60 # 60-Sekunden-Fenster
async def throttled_completion(self, prompt: str) -> str:
async with self.semaphore:
# Rolling Window Rate-Limiting
now = time.time()
self.request_times = [t for t in self.request_times if now - t < self.window]
if len(self.request_times) >= 50:
wait_time = self.window - (now - self.request_times[0])
await asyncio.sleep(wait_time)
self.request_times.append(now)
return await self._call_api(prompt)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=30))
async def _call_api(self, prompt: str) -> str:
try:
response = self.client.chat.completions.create(
model="deepseek-v3.2",
messages=[{"role": "user", "content": prompt}],
max_tokens=500
)
return response.choices[0].message.content
except RateLimitError as e:
print(f"Rate-Limit: Warte {e.retry_after}s")
await asyncio.sleep(e.retry_after)
raise3. Fehler: Funding-Rate-Daten-Latenz > 100ms
# FEHLERHAFT - Synchrone Verarbeitung verzögert Analyse
def process_funding(message):
data = parse_message(message) # Blockiert
signal = analyze_sync(data) # Blockiert
execute_order(signal) # 150ms+ Latenz
LÖSUNG - Pipelined Async-Architektur
class LowLatencyFundingAnalyzer:
def __init__(self):
self.pipeline = asyncio.Queue(maxsize=1000)
self.results = {}
async def start_pipeline(self):
# Parallele Consumer für maximale Parallelität
consumers = [
asyncio.create_task(self._parse_loop()),
asyncio.create_task(self._analyze_loop()),
asyncio.create_task(self._execution_loop()),
]
await asyncio.gather(*consumers)
async def _parse_loop(self):
"""Parse Tardis-Nachrichten mit <5ms Latenz."""
async for message in self.tardis_stream:
parsed = {
'symbol': message['symbol'],
'rate': message['data']['funding_rate'],
'timestamp': message['timestamp'],
'recv_time': time.perf_counter_ns() # Nanosekunden-Präzision
}
await self.pipeline.put(('parsed', parsed))
async def _analyze_loop(self):
"""LLM-An