Khi thị trường tiền mã hóa biến động mạnh vào quý 4 năm 2024, một nhà giao dịch algorithmic của tôi đã mất gần 3 tuần để debug một lỗi đơn giản trong việc parse dữ liệu Order Book từ Tardis.dev. Anh ấy không ngờ rằng chỉ cần thêm một tham số asOfSequenceId là toàn bộ chiến lược market-making của mình sẽ chạy sai lệch đến 23%. Câu chuyện này là lý do tôi viết bài hướng dẫn chi tiết này — để bạn không phải lặp lại sai lầm đó.
Tardis.dev là gì và tại sao cần thiết cho backtesting
Tardis.dev là dịch vụ cung cấp dữ liệu lịch sử chất lượng cao cho thị trường tiền mã hóa, bao gồm full Order Book snapshots, trades, funding rates và liquidations. Khác với các nguồn dữ liệu miễn phí khác, Tardis.dev cung cấp data ở mức exchange-native format, cho phép bạn tái hiện chính xác trạng thái thị trường tại bất kỳ thời điểm nào.
Trong bài viết này, tôi sẽ hướng dẫn bạn xây dựng một backtesting framework hoàn chỉnh sử dụng dữ liệu Order Book từ Tardis.dev, đồng thời tích hợp AI để phân tích kết quả và tối ưu chiến lược. Toàn bộ code sẽ sử dụng HolySheep AI làm backend — với chi phí chỉ từ $0.42/MTok cho DeepSeek V3.2, tiết kiệm đến 85% so với các provider khác.
Kiến trúc hệ thống backtesting
Trước khi đi vào code, hãy hiểu rõ kiến trúc tổng thể của framework mà chúng ta sẽ xây dựng:
+------------------+ +-------------------+ +------------------+
| Tardis.dev API | --> | Data Fetcher | --> | Order Book DB |
| (Historical) | | (Python Async) | | (SQLite/Parquet)|
+------------------+ +-------------------+ +------------------+
|
v
+------------------+ +-------------------+ +------------------+
| Strategy Engine | --> | Backtest Runner | --> | Results Store |
| (Your Logic) | | (Event-based) | | (JSON/CSV) |
+------------------+ +-------------------+ +------------------+
|
v
+------------------+ +-------------------+ +------------------+
| HolySheep AI | --> | Analysis Agent | --> | Report Gen |
| (DeepSeek/Claude)| | (Pattern Detect)| | (Markdown/HTML) |
+------------------+ +-------------------+ +------------------+
Cài đặt môi trường và dependencies
Đầu tiên, bạn cần cài đặt các thư viện cần thiết. Framework này được phát triển và test trên Python 3.10+ với asyncio native support.
# requirements.txt
tardis-client>=1.2.0
pandas>=2.0.0
numpy>=1.24.0
aiohttp>=3.9.0
asyncio-throttle>=1.0.2
sqlalchemy>=2.0.0
pydantic>=2.0.0
httpx>=0.25.0
tenacity>=8.2.0
Install all dependencies
pip install -r requirements.txt
Kết nối Tardis.dev API
Để bắt đầu, bạn cần tạo tài khoản Tardis.dev và lấy API key. Sau đó, tôi sẽ hướng dẫn bạn cách fetch dữ liệu Order Book một cách hiệu quả với rate limiting và error handling.
# config.py
import os
from dataclasses import dataclass
@dataclass
class Config:
# Tardis.dev configuration
TARDIS_API_KEY: str = os.getenv("TARDIS_API_KEY", "")
EXCHANGE: str = "binance" # Supported: binance, bybit, okx, etc.
MARKET: str = "BTC-USDT"
# Backtest settings
START_TIMESTAMP: int = 1704067200000 # 2024-01-01 00:00:00 UTC
END_TIMESTAMP: int = 1706745599000 # 2024-01-31 23:59:59 UTC
# HolySheep AI configuration - Cost effective alternative
HOLYSHEEP_API_KEY: str = os.getenv("HOLYSHEEP_API_KEY", "")
HOLYSHEEP_BASE_URL: str = "https://api.holysheep.ai/v1"
AI_MODEL: str = "deepseek-v3.2" # $0.42/MTok - best cost efficiency
# Rate limiting
MAX_REQUESTS_PER_SECOND: int = 10
MAX_CONCURRENT_REQUESTS: int = 5
config = Config()
# tardis_client.py
import asyncio
import aiohttp
import json
from typing import AsyncIterator, Dict, Any, List
from datetime import datetime
import tenacity
from config import config
class TardisDataFetcher:
"""Async fetcher for Tardis.dev historical data with retry logic"""
BASE_URL = "https://api.tardis.dev/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.semaphore = asyncio.Semaphore(config.MAX_CONCURRENT_REQUESTS)
def _get_headers(self) -> Dict[str, str]:
return {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
@tenacity.retry(
stop=tenacity.stop_after_attempt(3),
wait=tenacity.wait_exponential(multiplier=1, min=2, max=10)
)
async def fetch_orderbook_snapshots(
self,
exchange: str,
market: str,
from_ts: int,
to_ts: int
) -> AsyncIterator[Dict[str, Any]]:
"""
Fetch Order Book snapshots with pagination
IMPORTANT: Tardis.dev returns data in exchange-native format.
For Binance, the sequenceId is crucial for maintaining order.
"""
url = f"{self.BASE_URL}/feeds"
# Query params for Order Book data
params = {
"exchange": exchange,
"symbol": market,
"fromTimestamp": from_ts,
"toTimestamp": to_ts,
"types": "orderbook_snapshot"
}
async with aiohttp.ClientSession() as session:
page = 1
while True:
async with self.semaphore:
response = await session.get(
url,
headers=self._get_headers(),
params={**params, "page": page}
)
if response.status == 429:
# Rate limited, wait and retry
await asyncio.sleep(60)
continue
response.raise_for_status()
data = await response.json()
if not data.get("data"):
break
for item in data["data"]:
yield item
if not data.get("hasMore"):
break
page += 1
await asyncio.sleep(0.1) # Respect rate limits
async def get_available_symbols(self, exchange: str) -> List[str]:
"""List available trading pairs for an exchange"""
url = f"{self.BASE_URL}/exchanges/{exchange}/symbols"
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=self._get_headers()) as response:
response.raise_for_status()
data = await response.json()
return [s["symbol"] for s in data.get("symbols", [])]
Usage example
async def main():
fetcher = TardisDataFetcher(config.TARDIS_API_KEY)
async for snapshot in fetcher.fetch_orderbook_snapshots(
config.EXCHANGE,
config.MARKET,
config.START_TIMESTAMP,
config.END_TIMESTAMP
):
# Process each snapshot
print(f"Sequence {snapshot.get('sequenceId')}: {snapshot.get('timestamp')}")
# Your processing logic here
if __name__ == "__main__":
asyncio.run(main())
Xây dựng Order Book Data Model
Để backtesting chính xác, bạn cần một data model biểu diễn trạng thái Order Book đầy đủ. Dưới đây là implementation với Pydantic cho validation và type safety.
# orderbook.py
from pydantic import BaseModel, Field, field_validator
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass, field
from decimal import Decimal
from sortedcontainers import SortedDict
import heapq
@dataclass
class OrderBookLevel:
"""Single price level in order book"""
price: Decimal
quantity: Decimal
def __lt__(self, other):
return self.price < other.price
class OrderBookSide:
"""
Order Book side (bids or asks) implementation
Using SortedDict for O(log n) insert/delete operations
"""
def __init__(self, reverse: bool = True):
# reverse=True for bids (sort descending by price)
self._levels: Dict[Decimal, Decimal] = {}
self._sorted_keys = SortedDict(reverse=reverse)
def update(self, price: Decimal, quantity: Decimal) -> None:
"""Update a price level"""
if quantity == 0:
self.remove(price)
else:
self._levels[price] = quantity
self._sorted_keys[price] = quantity
def remove(self, price: Decimal) -> None:
"""Remove a price level"""
if price in self._levels:
del self._levels[price]
del self._sorted_keys[price]
def get_best_price(self) -> Optional[Decimal]:
"""Get best bid/ask price"""
if self._sorted_keys:
return self._sorted_keys.peekitem(0)[0]
return None
def get_quantity_at_price(self, price: Decimal) -> Decimal:
"""Get quantity at specific price level"""
return self._levels.get(price, Decimal("0"))
def get_levels(self, depth: int = 10) -> List[Tuple[Decimal, Decimal]]:
"""Get top N price levels"""
items = list(self._sorted_keys.items())[:depth]
return [(price, qty) for price, qty in items]
def get_total_quantity(self, depth: int = 10) -> Decimal:
"""Get total quantity in top N levels"""
levels = self.get_levels(depth)
return sum((qty for _, qty in levels), Decimal("0"))
def __len__(self) -> int:
return len(self._levels)
class OrderBook:
"""
Full Order Book representation with snapshot apply logic
IMPORTANT: Tardis.dev provides full snapshots, not delta updates.
This makes our implementation simpler but requires more bandwidth.
"""
def __init__(self, symbol: str):
self.symbol = symbol
self.bids = OrderBookSide(reverse=True) # Descending
self.asks = OrderBookSide(reverse=False) # Ascending
self.sequence_id: Optional[int] = None
self.timestamp: Optional[int] = None
self.last_update_id: Optional[int] = None
@classmethod
def from_tardis_snapshot(cls, data: Dict) -> "OrderBook":
"""
Parse Order Book from Tardis.dev API response
CRITICAL: The data format varies by exchange!
- Binance: {bids: [[price, qty], ...], asks: [...]}
- Bybit: {b: [[price, qty], ...], a: [...]}
"""
ob = cls(data.get("symbol", "UNKNOWN"))
# Handle Binance format
if "bids" in data and "asks" in data:
for price, qty in data["bids"]:
ob.bids.update(Decimal(str(price)), Decimal(str(qty)))
for price, qty in data["asks"]:
ob.asks.update(Decimal(str(price)), Decimal(str(qty)))
# Handle Bybit format
elif "b" in data and "a" in data:
for price, qty in data["b"]:
ob.bids.update(Decimal(str(price)), Decimal(str(qty)))
for price, qty in data["a"]:
ob.asks.update(Decimal(str(price)), Decimal(str(qty)))
ob.sequence_id = data.get("sequenceId") or data.get("sequence")
ob.timestamp = data.get("timestamp") or data.get("ts")
ob.last_update_id = data.get("lastUpdateId") or data.get("u")
return ob
def get_mid_price(self) -> Optional[Decimal]:
"""Calculate mid price"""
best_bid = self.bids.get_best_price()
best_ask = self.asks.get_best_price()
if best_bid and best_ask:
return (best_bid + best_ask) / 2
return None
def get_spread(self) -> Optional[Decimal]:
"""Calculate bid-ask spread"""
best_bid = self.bids.get_best_price()
best_ask = self.asks.get_best_price()
if best_bid and best_ask:
return best_ask - best_bid
return None
def get_spread_bps(self) -> Optional[Decimal]:
"""Calculate spread in basis points"""
spread = self.get_spread()
mid = self.get_mid_price()
if spread and mid:
return (spread / mid) * 10000
return None
def get_imbalance(self, levels: int = 10) -> Decimal:
"""
Calculate order book imbalance
Imbalance = (Bid Qty - Ask Qty) / (Bid Qty + Ask Qty)
Range: -1 (all asks) to +1 (all bids)
"""
bid_qty = self.bids.get_total_quantity(levels)
ask_qty = self.asks.get_total_quantity(levels)
total = bid_qty + ask_qty
if total == 0:
return Decimal("0")
return (bid_qty - ask_qty) / total
def visualize(self, depth: int = 5) -> str:
"""Generate ASCII visualization of order book"""
bid_levels = self.bids.get_levels(depth)
ask_levels = self.asks.get_levels(depth)
lines = [f"Order Book: {self.symbol}", "-" * 60]
# Display asks (reversed so highest ask is at top)
for price, qty in reversed(ask_levels):
lines.append(f"{float(qty):>12.4f} {float(price):>12.2f} ASK")
if bid_levels and ask_levels:
mid = self.get_mid_price()
spread = self.get_spread()
lines.append(f"{'MID: ' + str(float(mid)):>26} Spread: {spread}")
# Display bids
for price, qty in bid_levels:
lines.append(f"{float(qty):>12.4f} {float(price):>12.2f} BID")
return "\n".join(lines)
Backtesting Engine Implementation
Đây là phần cốt lõi của framework — Backtesting Engine xử lý events, quản lý portfolio và tính toán performance metrics. Engine được thiết kế theo event-driven architecture để đảm bảo simulation chính xác.
# backtest_engine.py
from decimal import Decimal
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable, Any
from enum import Enum
from datetime import datetime
import json
class OrderSide(Enum):
BUY = "BUY"
SELL = "SELL"
class OrderType(Enum):
MARKET = "MARKET"
LIMIT = "LIMIT"
@dataclass
class Order:
order_id: str
side: OrderSide
price: Decimal
quantity: Decimal
order_type: OrderType
filled_quantity: Decimal = Decimal("0")
avg_fill_price: Decimal = Decimal("0")
status: str = "pending"
created_at: int = 0
@dataclass
class Position:
"""Single position tracking"""
symbol: str
quantity: Decimal = Decimal("0")
avg_entry_price: Decimal = Decimal("0")
unrealized_pnl: Decimal = Decimal("0")
realized_pnl: Decimal = Decimal("0")
@dataclass
class Portfolio:
"""Portfolio state manager"""
initial_balance: Decimal
current_balance: Decimal
positions: Dict[str, Position] = field(default_factory=dict)
def get_total_value(self, current_price: Decimal) -> Decimal:
"""Calculate total portfolio value"""
total = self.current_balance
for symbol, pos in self.positions.items():
if pos.quantity > 0:
total += pos.quantity * current_price
return total
def update_position(self, symbol: str, fill_price: Decimal, quantity: Decimal, side: OrderSide):
"""Update position after a fill"""
if symbol not in self.positions:
self.positions[symbol] = Position(symbol)
pos = self.positions[symbol]
if side == OrderSide.BUY:
# Add to position
total_cost = pos.quantity * pos.avg_entry_price + quantity * fill_price
pos.quantity += quantity
pos.avg_entry_price = total_cost / pos.quantity if pos.quantity > 0 else Decimal("0")
else: # SELL
# Reduce or close position
if quantity >= pos.quantity:
# Close position
pnl = (fill_price - pos.avg_entry_price) * pos.quantity
pos.realized_pnl += pnl
self.current_balance += pos.quantity * fill_price
pos.quantity = Decimal("0")
pos.avg_entry_price = Decimal("0")
else:
# Partial close
pnl = (fill_price - pos.avg_entry_price) * quantity
pos.realized_pnl += pnl
self.current_balance += quantity * fill_price
pos.quantity -= quantity
@dataclass
class BacktestResult:
"""Container for backtest results"""
total_trades: int = 0
winning_trades: int = 0
losing_trades: int = 0
total_pnl: Decimal = Decimal("0")
max_drawdown: Decimal = Decimal("0")
sharpe_ratio: float = 0.0
trades: List[Dict] = field(default_factory=list)
equity_curve: List[Dict] = field(default_factory=list)
class BacktestEngine:
"""
Event-driven backtesting engine
Key features:
- Realistic slippage modeling
- Fee simulation (configurable)
- Position management
- Performance metrics calculation
"""
def __init__(
self,
initial_balance: Decimal = Decimal("100000"),
maker_fee: Decimal = Decimal("0.001"),
taker_fee: Decimal = Decimal("0.002"),
slippage_bps: Decimal = Decimal("5")
):
self.initial_balance = initial_balance
self.portfolio = Portfolio(initial_balance, initial_balance)
self.maker_fee = maker_fee
self.taker_fee = taker_fee
self.slippage_bps = slippage_bps
self.pending_orders: Dict[str, Order] = {}
self.trade_history: List[Dict] = []
self.equity_curve: List[Dict] = []
self.current_orderbook: Optional[Any] = None
self.strategies: List[Callable] = []
self.order_id_counter = 0
def register_strategy(self, strategy: Callable):
"""Register a strategy function to be called on each tick"""
self.strategies.append(strategy)
def _calculate_slippage(self, price: Decimal, side: OrderSide) -> Decimal:
"""Calculate slippage based on order side"""
if side == OrderSide.BUY:
return price * (1 + self.slippage_bps / 10000)
else:
return price * (1 - self.slippage_bps / 10000)
def _generate_order_id(self) -> str:
self.order_id_counter += 1
return f"ORDER_{self.current_timestamp}_{self.order_id_counter}"
def process_orderbook_snapshot(self, snapshot: Any, timestamp: int):
"""Process a new order book snapshot"""
self.current_orderbook = snapshot
self.current_timestamp = timestamp
# Run registered strategies
for strategy in self.strategies:
strategy(self)
# Check pending limit orders against current order book
self._check_limit_orders()
def place_order(
self,
side: OrderSide,
quantity: Decimal,
price: Optional[Decimal] = None,
order_type: OrderType = OrderType.MARKET
) -> Optional[Order]:
"""
Place an order
For MARKET orders: immediately filled with slippage
For LIMIT orders: added to pending orders
"""
if price is None and self.current_orderbook:
price = self.current_orderbook.get_mid_price()
if price is None:
return None
order = Order(
order_id=self._generate_order_id(),
side=side,
price=price,
quantity=quantity,
order_type=order_type,
created_at=self.current_timestamp
)
if order_type == OrderType.MARKET:
self._execute_order(order)
else:
self.pending_orders[order.order_id] = order
return order
def _execute_order(self, order: Order):
"""Execute an order with slippage and fees"""
fill_price = self._calculate_slippage(order.price, order.side)
fee = fill_price * order.quantity * self.taker_fee
order.filled_quantity = order.quantity
order.avg_fill_price = fill_price
order.status = "filled"
# Update portfolio
self.portfolio.update_position(
self.current_orderbook.symbol if self.current_orderbook else "BTC-USDT",
fill_price,
order.quantity,
order.side
)
# Deduct fees
self.portfolio.current_balance -= fee
# Record trade
trade = {
"order_id": order.order_id,
"side": order.side.value,
"quantity": float(order.quantity),
"price": float(fill_price),
"fee": float(fee),
"timestamp": self.current_timestamp
}
self.trade_history.append(trade)
def _check_limit_orders(self):
"""Check if any pending limit orders can be filled"""
to_remove = []
for order_id, order in self.pending_orders.items():
if order.side == OrderSide.BUY:
# Buy limit: fill when best ask <= limit price
if self.current_orderbook.asks.get_best_price() <= order.price:
self._execute_order(order)
to_remove.append(order_id)
else:
# Sell limit: fill when best bid >= limit price
if self.current_orderbook.bids.get_best_price() >= order.price:
self._execute_order(order)
to_remove.append(order_id)
for order_id in to_remove:
del self.pending_orders[order_id]
def get_results(self) -> BacktestResult:
"""Calculate and return backtest results"""
# Calculate equity curve
for i, trade in enumerate(self.trade_history):
total_value = self.portfolio.current_balance
for symbol, pos in self.portfolio.positions.items():
if pos.quantity > 0 and self.current_orderbook:
total_value += pos.quantity * self.current_orderbook.get_mid_price()
self.equity_curve.append({
"timestamp": trade["timestamp"],
"equity": float(total_value)
})
# Calculate metrics
total_pnl = self.portfolio.current_balance - self.initial_balance
winning = sum(1 for t in self.trade_history if t.get("pnl", 0) > 0)
losing = len(self.trade_history) - winning
# Max drawdown
peak = self.initial_balance
max_dd = Decimal("0")
for entry in self.equity_curve:
equity = Decimal(str(entry["equity"]))
if equity > peak:
peak = equity
dd = peak - equity
if dd > max_dd:
max_dd = dd
return BacktestResult(
total_trades=len(self.trade_history),
winning_trades=winning,
losing_trades=losing,
total_pnl=total_pnl,
max_drawdown=max_dd,
equity_curve=self.equity_curve
)
Tích hợp HolySheep AI để phân tích kết quả
Sau khi chạy backtest, việc phân tích kết quả là vô cùng quan trọng. Với chi phí chỉ $0.42/MTok cho DeepSeek V3.2 tại HolySheep AI, bạn có thể sử dụng AI để:
- Phân tích pattern của các giao dịch thua lỗ
- Đề xuất cải thiện chiến lược
- Tạo báo cáo chi tiết với visualization
- Tối ưu hóa tham số chiến lược
# analysis_agent.py
import httpx
import json
from typing import Dict, List, Any, Optional
from config import config
class HolySheepAnalyzer:
"""
AI-powered analysis agent using HolySheep AI
HolySheep advantages:
- DeepSeek V3.2 at $0.42/MTok (85% cheaper than alternatives)
- Support for both Chinese and English prompts
- <50ms latency for real-time analysis
"""
SYSTEM_PROMPT = """Bạn là một chuyên gia phân tích giao dịch tiền mã hóa.
Phân tích dữ liệu backtest và đưa ra đề xuất cải tiến chiến lược.
Trả lời bằng tiếng Việt với các mục rõ ràng."""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = config.HOLYSHEEP_BASE_URL
async def analyze_backtest_results(
self,
results: Dict[str, Any],
trades: List[Dict]
) -> str:
"""
Analyze backtest results using HolySheep AI
Args:
results: BacktestResult as dictionary
trades: List of executed trades
Returns:
AI-generated analysis in Vietnamese
"""
# Prepare context for AI
context = self._prepare_context(results, trades)
user_prompt = f"""Phân tích kết quả backtest sau:
**Tổng quan:**
- Tổng số giao dịch: {results.get('total_trades', 0)}
- Giao dịch thắng: {results.get('winning_trades', 0)}
- Giao dịch thua: {results.get('losing_trades', 0)}
- PnL tổng: ${results.get('total_pnl', 0)}
- Max Drawdown: ${results.get('max_drawdown', 0)}
**Top 5 giao dịch thua lỗ:**
{json.dumps(results.get('worst_trades', [])[:5], indent=2)}
**Phân tích:**
1. Nguyên nhân chính của các giao dịch thua lỗ?
2. Có pattern nào trong thời điểm vào lệnh không?
3. Đề xuất 3 cách cải thiện chiến lược?
"""
return await self._call_ai(context, user_prompt)
async def optimize_parameters(
self,
strategy_name: str,
current_params: Dict[str, Any],
backtest_summary: Dict[str, Any]
) -> Dict[str, Any]:
"""
Use AI to suggest optimized parameters
Returns optimized parameters dictionary
"""
user_prompt = f"""Tối ưu hóa tham số cho chiến lược {strategy_name}:
**Tham số hiện tại:**
{json.dumps(current_params, indent=2)}
**Kết quả backtest:**
- Sharpe Ratio: {backtest_summary.get('sharpe_ratio', 0):.2f}
- Win Rate: {backtest_summary.get('win_rate', 0)*100:.1f}%
- Max Drawdown: ${backtest_summary.get('max_drawdown', 0)}
Đề xuất tham số tối ưu với:
1. Giải thích rationale
2. Expected improvement
3. Risk considerations
Trả lời theo format JSON: {{"params": {{...}}, "reasoning": "...", "expected_sharpe": X.X}}"""
response = await self._call_ai("", user_prompt)
# Parse JSON from response
try:
# Extract JSON from response
import re
json_match = re.search(r'\{.*\}', response, re.DOTALL)
if json_match:
return json.loads(json_match.group())
except:
pass
return {"error": "Could not parse optimization results"}
async def generate_report(
self,
results: Dict[str, Any],
strategy_name: str,
timeframe: str
) -> str:
"""Generate comprehensive backtest report"""
user_prompt = f"""Tạo báo cáo backtest chi tiết cho chiến lược {strategy_name}:
**Thông tin:**
- Timeframe: {timeframe}
- Số giao dịch: {results.get('total_trades', 0)}
- Win Rate: {results.get('win_rate', 0)*100:.1f}%
- Profit Factor: {results.get('profit_factor', 0):.2f}
- Sharpe Ratio: {results.get('sharpe_ratio', 0):.2f}
- Max Drawdown: {results.get('max_drawdown_pct', 0)*100:.1f}%
- Total PnL: ${results.get('total_pnl', 0)}
Tạo báo cáo theo cấu trúc:
1. Tóm tắt điều hành
2. Hiệu suất chi tiết
3. Phân tích rủi ro
4. Kết luận và khuyến nghị
5. Bước tiếp theo
Format: Markdown với bảng biểu và bullet points"""
return await self._call_ai("", user_prompt)
async def _call_ai(self, context: str, user_prompt: str) -> str:
"""Make API call to HolySheep AI"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}