บทนำ: ทำไมต้องดึงข้อมูล Orderbook จาก Hyperliquid
Hyperliquid เป็น decentralized perpetual futures exchange ที่ได้รับความนิยมอย่างมากในปี 2026 ด้วยปริมาณการซื้อขายที่สูงและความเร็วในการ settlement ที่เป็นเลิศ สำหรับนักพัฒนา AI ที่ต้องการสร้างโมเดล predictive สำหรับตลาด crypto หรือระบบ automated trading การเข้าถึง historical orderbook data เป็นสิ่งจำเป็นอย่างยิ่ง
บทความนี้จะอธิบายวิธีใช้ Tardis Python API เพื่อดึงข้อมูล orderbook ย้อนหลังจาก OKX Hyperliquid อย่างละเอียด พร้อมตัวอย่างโค้ดที่พร้อมใช้งานจริง
Tardis API คืออะไร และทำไมถึงเหมาะกับงานนี้
Tardis เป็นบริการที่รวบรวมข้อมูล market data จาก exchanges หลากหลายรายในรูปแบบ normalized format ทำให้นักพัฒนาสามารถเข้าถึง historical data ได้ง่ายผ่าน unified API
# ติดตั้ง Tardis Python SDK
pip install tardis-python
หรือใช้ poetry
poetry add tardis-python
# ติดตั้ง dependencies ที่จำเป็น
pip install pandas aiohttp asyncio pandas-gbq
การตั้งค่า Credentials และ Authentication
# config.py - การตั้งค่า Tardis API credentials
import os
Tardis credentials
TARDIS_API_KEY = os.environ.get("TARDIS_API_KEY", "your_tardis_api_key")
TARDIS_API_URL = "https://api.tardis.dev/v1"
Exchange configuration
EXCHANGE = "hyperliquid" # หรือ "okx" สำหรับ OKX
เครือข่ายที่ต้องการ (สำหรับ Hyperliquid)
NETWORK = "mainnet"
ช่วงเวลาที่ต้องการดึงข้อมูล
START_DATE = "2026-03-01"
END_DATE = "2026-04-01"
print(f"✅ Configuration loaded for {EXCHANGE}/{NETWORK}")
ดึงข้อมูล Orderbook พร้อม Async Implementation
# hyperliquid_orderbook.py
import asyncio
import aiohttp
import json
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import pandas as pd
class HyperliquidOrderbookFetcher:
"""ดึงข้อมูล historical orderbook จาก Tardis API"""
def __init__(self, api_key: str, exchange: str = "hyperliquid"):
self.api_key = api_key
self.exchange = exchange
self.base_url = "https://api.tardis.dev/v1"
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={"Authorization": f"Bearer {self.api_key}"}
)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def fetch_orderbook_snapshot(
self,
symbol: str,
timestamp_from: datetime,
timestamp_to: datetime,
limit: int = 100
) -> List[Dict]:
"""ดึง orderbook snapshots ในช่วงเวลาที่กำหนด"""
url = f"{self.base_url}/historical/{self.exchange}/orderbook-snapshots"
params = {
"symbol": symbol,
"from": int(timestamp_from.timestamp() * 1000),
"to": int(timestamp_to.timestamp() * 1000),
"limit": limit,
"format": "json"
}
try:
async with self.session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
return data.get("data", [])
elif response.status == 401:
raise Exception("❌ Invalid Tardis API key")
elif response.status == 429:
raise Exception("⏳ Rate limit exceeded. Please wait.")
else:
raise Exception(f"❌ API Error: {response.status}")
except aiohttp.ClientError as e:
raise Exception(f"❌ Network error: {str(e)}")
async def get_orderbook_for_period(
self,
symbol: str,
start: datetime,
end: datetime,
interval_minutes: int = 5
) -> pd.DataFrame:
"""ดึงข้อมูล orderbook เป็น period พร้อมกรองข้อมูล"""
all_snapshots = []
current = start
while current < end:
next_time = min(current + timedelta(minutes=interval_minutes), end)
try:
snapshots = await self.fetch_orderbook_snapshot(
symbol=symbol,
timestamp_from=current,
timestamp_to=next_time
)
all_snapshots.extend(snapshots)
print(f"✅ Fetched {len(snapshots)} snapshots for {current.strftime('%Y-%m-%d %H:%M')}")
except Exception as e:
print(f"⚠️ Error fetching {current}: {e}")
current = next_time
await asyncio.sleep(0.5) # หลีกเลี่ยง rate limit
return self._process_orderbook_data(all_snapshots, symbol)
def _process_orderbook_data(self, snapshots: List[Dict], symbol: str) -> pd.DataFrame:
"""ประมวลผล raw snapshots เป็น structured DataFrame"""
processed = []
for snapshot in snapshots:
timestamp = snapshot.get("timestamp") or snapshot.get("local_timestamp")
if isinstance(timestamp, str):
timestamp = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
record = {
"timestamp": timestamp,
"symbol": symbol,
"bids_count": len(snapshot.get("bids", [])),
"asks_count": len(snapshot.get("asks", [])),
"best_bid": float(snapshot["bids"][0]["price"]) if snapshot.get("bids") else None,
"best_ask": float(snapshot["asks"][0]["price"]) if snapshot.get("asks") else None,
"spread": None,
"mid_price": None,
"bid_volume_total": sum(float(b.get("size", 0)) for b in snapshot.get("bids", [])),
"ask_volume_total": sum(float(a.get("size", 0)) for a in snapshot.get("asks", [])),
"bid_ask_imbalance": None
}
if record["best_bid"] and record["best_ask"]:
record["spread"] = record["best_ask"] - record["best_bid"]
record["mid_price"] = (record["best_bid"] + record["best_ask"]) / 2
total_volume = record["bid_volume_total"] + record["ask_volume_total"]
if total_volume > 0:
record["bid_ask_imbalance"] = (
record["bid_volume_total"] - record["ask_volume_total"]
) / total_volume
processed.append(record)
df = pd.DataFrame(processed)
if not df.empty:
df = df.sort_values("timestamp").reset_index(drop=True)
return df
ตัวอย่างการใช้งาน
async def main():
fetcher = HyperliquidOrderbookFetcher(
api_key="your_tardis_api_key",
exchange="hyperliquid"
)
async with fetcher:
df = await fetcher.get_orderbook_for_period(
symbol="BTC-PERP",
start=datetime(2026, 4, 1, 0, 0),
end=datetime(2026, 4, 1, 12, 0),
interval_minutes=10
)
print(f"\n📊 Total snapshots fetched: {len(df)}")
print(df.head(10))
# บันทึกเป็น CSV
df.to_csv("hyperliquid_orderbook_btc_perp.csv", index=False)
print("💾 Data saved to hyperliquid_orderbook_btc_perp.csv")
if __name__ == "__main__":
asyncio.run(main())
ดึงข้อมูล OKX Hyperliquid (Alternative Exchange)
สำหรับผู้ที่ต้องการข้อมูลจาก OKX exchange ที่รองรับ Hyperliquid perpetual futures:
# okx_hyperliquid_fetcher.py
import asyncio
import pandas as pd
from datetime import datetime
from tardis_client import TardisClient, Exchanges, MarketType
class OKXHyperliquidFetcher:
"""ดึงข้อมูลจาก OKX exchange ผ่าน Tardis"""
def __init__(self, api_key: str):
self.client = TardisClient(api_key=api_key)
async def fetch_trades(self, symbol: str, start: int, end: int) -> pd.DataFrame:
"""ดึงข้อมูล trades (สำหรับ validate orderbook)"""
# Tardis exchange name สำหรับ OKX
exchange = Exchanges.OKX
# Market type สำหรับ perpetual swap
market_type = MarketType.FUTURES
# สัญลักษณ์บน OKX
okx_symbol = f"{symbol}-USDT-SWAP"
messages = self.client.replay(
exchange=exchange,
market_type=market_type,
symbols=[okx_symbol],
from_timestamp=start,
to_timestamp=end
)
trades_data = []
async for message in messages:
if message.type == "trade":
trades_data.append({
"timestamp": message.timestamp,
"symbol": message.symbol,
"price": float(message.price),
"side": message.side,
"size": float(message.size),
"trade_id": message.id
})
return pd.DataFrame(trades_data)
async def analyze_orderbook_depth(self, orderbook_df: pd.DataFrame) -> dict:
"""วิเคราะห์ความลึกของ orderbook"""
if orderbook_df.empty:
return {}
return {
"avg_bid_count": orderbook_df["bids_count"].mean(),
"avg_ask_count": orderbook_df["asks_count"].mean(),
"avg_spread": orderbook_df["spread"].mean(),
"max_imbalance": orderbook_df["bid_ask_imbalance"].abs().max(),
"avg_mid_price": orderbook_df["mid_price"].mean(),
"price_volatility": orderbook_df["mid_price"].std(),
"total_bid_volume": orderbook_df["bid_volume_total"].sum(),
"total_ask_volume": orderbook_df["ask_volume_total"].sum()
}
การใช้งาน
async def main():
client = OKXHyperliquidFetcher(api_key="your_tardis_api_key")
start_ts = int(datetime(2026, 4, 1).timestamp() * 1000)
end_ts = int(datetime(2026, 4, 2).timestamp() * 1000)
trades_df = await client.fetch_trades(
symbol="BTC",
start=start_ts,
end=end_ts
)
print(f"📈 Total trades: {len(trades_df)}")
print(trades_df.describe())
# วิเคราะห์ volume profile
if not trades_df.empty:
trades_df["price_bucket"] = (trades_df["price"] // 100) * 100
volume_profile = trades_df.groupby("price_bucket")["size"].sum()
print("\n📊 Volume Profile (top 10):")
print(volume_profile.sort_values(ascending=False).head(10))
if __name__ == "__main__":
asyncio.run(main())
ผสมผสาน Orderbook Data กับ AI Model ผ่าน HolySheep AI
นี่คือจุดที่ HolySheep AI เข้ามามีบทบาทสำคัญ เมื่อคุณประมวลผล orderbook data แล้ว สามารถใช้ AI models จาก
HolySheep AI เพื่อวิเคราะห์ patterns และทำนาย price movements
# ai_orderbook_analyzer.py
import aiohttp
import json
import pandas as pd
from datetime import datetime
class HolySheepOrderbookAnalyzer:
"""ใช้ AI วิเคราะห์ orderbook patterns ผ่าน HolySheep API"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
async def analyze_orderbook_pattern(
self,
orderbook_df: pd.DataFrame,
market_context: str = "BTC-PERP"
) -> dict:
"""วิเคราะห์ orderbook ด้วย AI"""
# สร้าง summary จาก orderbook data
summary = self._create_orderbook_summary(orderbook_df)
# Prompt สำหรับ AI analysis
prompt = f"""คุณเป็นนักวิเคราะห์ตลาด crypto ที่มีประสบการณ์
วิเคราะห์ orderbook data สำหรับ {market_context} และให้ความเห็น:
{summary}
โปรดวิเคราะห์:
1. ความหนาแน่นของ Orderbook (บอกทิศทางแนวโน้ม)
2. Order Imbalance บ่งชี้อะไร
3. Spread patterns บอกอะไรเกี่ยวกับ liquidity
4. ความเสี่ยงและโอกาสในระยะสั้น
ตอบเป็น JSON format พร้อม sentiment score (1-10)"""
async with aiohttp.ClientSession() as session:
payload = {
"model": "gpt-4.1", # $8/MTok - เหมาะสำหรับ analysis
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 1000
}
async with session.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload
) as response:
if response.status == 200:
result = await response.json()
ai_response = result["choices"][0]["message"]["content"]
# ลอง parse JSON response
try:
analysis = json.loads(ai_response)
except json.JSONDecodeError:
analysis = {
"raw_analysis": ai_response,
"sentiment_score": None
}
return analysis
else:
error = await response.text()
raise Exception(f"❌ HolySheep API Error: {error}")
def _create_orderbook_summary(self, df: pd.DataFrame) -> str:
"""สร้าง summary จาก DataFrame"""
if df.empty:
return "No data available"
latest = df.iloc[-1]
return f"""
Orderbook Summary:
- ช่วงเวลา: {df['timestamp'].min()} ถึง {df['timestamp'].max()}
- จำนวน snapshots: {len(df)}
- Best Bid ล่าสุด: ${latest['best_bid']:,.2f}
- Best Ask ล่าสุด: ${latest['best_ask']:,.2f}
- Spread เฉลี่ย: ${df['spread'].mean():,.4f}
- Mid Price ล่าสุด: ${latest['mid_price']:,.2f}
- Bid/Ask Imbalance เฉลี่ย: {df['bid_ask_imbalance'].mean():.4f}
- Total Bid Volume: {df['bid_volume_total'].sum():,.2f}
- Total Ask Volume: {df['ask_volume_total'].sum():,.2f}
- Bid/Ask Ratio: {df['bid_volume_total'].sum() / df['ask_volume_total'].sum():.4f}
"""
ตัวอย่างการใช้งาน
async def main():
# สมมติว่ามี orderbook_df จาก fetcher
analyzer = HolySheepOrderbookAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
# สร้าง sample data
sample_data = {
"timestamp": pd.date_range("2026-04-01", periods=100, freq="5min"),
"best_bid": 67000 + pd.Series(range(100)).apply(lambda x: 67000 + x * 0.5 + pd.np.random.randn() * 10),
"best_ask": 67010 + pd.Series(range(100)).apply(lambda x: 67010 + x * 0.5 + pd.np.random.randn() * 10),
"spread": [10 + pd.np.random.randn() * 2 for _ in range(100)],
"bid_volume_total": [1000 + pd.np.random.randn() * 200 for _ in range(100)],
"ask_volume_total": [950 + pd.np.random.randn() * 200 for _ in range(100)],
"bid_ask_imbalance": [pd.np.random.randn() * 0.1 for _ in range(100)],
"mid_price": [67005 + x * 0.5 + pd.np.random.randn() * 10 for x in range(100)],
"bids_count": [50 + int(pd.np.random.randn() * 10) for _ in range(100)],
"asks_count": [48 + int(pd.np.random.randn() * 10) for _ in range(100)]
}
orderbook_df = pd.DataFrame(sample_data)
analysis = await analyzer.analyze_orderbook_pattern(
orderbook_df,
market_context="BTC-PERP on Hyperliquid"
)
print("🤖 AI Analysis Result:")
print(json.dumps(analysis, indent=2, ensure_ascii=False))
if __name__ == "__main__":
import pandas as pd
pd.np.random.seed(42)
asyncio.run(main())
บันทึกข้อมูลและส่งออก
# data_exporter.py
import pandas as pd
from datetime import datetime
import json
def export_to_multiple_formats(df: pd.DataFrame, symbol: str, output_dir: str = "./data"):
"""ส่งออกข้อมูลในหลายรูปแบบ"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
base_name = f"{symbol}_{timestamp}"
# 1. CSV - สำหรับ analysis ทั่วไป
csv_path = f"{output_dir}/{base_name}.csv"
df.to_csv(csv_path, index=False)
print(f"💾 Saved CSV: {csv_path}")
# 2. Parquet - สำหรับ large dataset และ performance
parquet_path = f"{output_dir}/{base_name}.parquet"
df.to_parquet(parquet_path, index=False, engine="pyarrow")
print(f"💾 Saved Parquet: {parquet_path}")
# 3. JSON - สำหรับ API responses
json_path = f"{output_dir}/{base_name}.json"
df.to_json(json_path, orient="records", date_format="iso")
print(f"💾 Saved JSON: {json_path}")
# 4. แยกเป็น individual files ตามวัน
if "timestamp" in df.columns:
df["date"] = pd.to_datetime(df["timestamp"]).dt.date
for date, group in df.groupby("date"):
date_str = str(date)
daily_path = f"{output_dir}/{symbol}_{date_str}.csv"
group.drop(columns=["date"]).to_csv(daily_path, index=False)
print(f"📅 Saved daily: {daily_path}")
# 5. Summary statistics
summary_stats = {
"symbol": symbol,
"export_timestamp": timestamp,
"total_records": len(df),
"date_range": {
"start": str(df["timestamp"].min()),
"end": str(df["timestamp"].max())
},
"statistics": {
"avg_spread": float(df["spread"].mean()),
"max_spread": float(df["spread"].max()),
"min_spread": float(df["spread"].min()),
"avg_bid_ask_imbalance": float(df["bid_ask_imbalance"].mean()),
"total_bid_volume": float(df["bid_volume_total"].sum()),
"total_ask_volume": float(df["ask_volume_total"].sum())
}
}
summary_path = f"{output_dir}/{base_name}_summary.json"
with open(summary_path, "w", encoding="utf-8") as f:
json.dump(summary_stats, f, indent=2, ensure_ascii=False)
print(f"📊 Saved summary: {summary_path}")
return {
"csv": csv_path,
"parquet": parquet_path,
"json": json_path,
"summary": summary_path
}
การใช้งาน
if __name__ == "__main__":
# สร้าง sample DataFrame
sample_df = pd.DataFrame({
"timestamp": pd.date_range("2026-04-01", periods=1000, freq="1min"),
"best_bid": 67000 + pd.np.random.randn(1000) * 50,
"best_ask": 67010 + pd.np.random.randn(1000) * 50,
"spread": abs(pd.np.random.randn(1000)) * 5 + 5,
"bid_volume_total": abs(pd.np.random.randn(1000)) * 1000 + 500,
"ask_volume_total": abs(pd.np.random.randn(1000)) * 1000 + 500,
"bid_ask_imbalance": pd.np.random.randn(1000) * 0.2,
"mid_price": 67005 + pd.np.random.randn(1000) * 50,
"bids_count": abs(pd.np.random.randn(1000) * 20 + 50).astype(int),
"asks_count": abs(pd.np.random.randn(1000) * 20 + 48).astype(int)
})
import os
os.makedirs("./data", exist_ok=True)
paths = export_to_multiple_formats(sample_df, "BTC-PERP")
print("\n✅ All exports completed!")
print(f"Files created: {list(paths.keys())}")
ประยุกต์ใช้กับระบบ RAG สำหรับ Trading Analysis
คุณสามารถนำข้อมูล orderbook ที่ประมวลผลแล้วไปใช้ในระบบ RAG (Retrieval-Augmented Generation) เพื่อสร้าง AI assistant ที่เข้าใจบริบทตลาด:
# rag_trading_assistant.py
import aiohttp
import json
import pandas as pd
from datetime import datetime, timedelta
from typing import List, Dict
class TradingRAGAssistant:
"""ระบบ RAG สำหรับวิเคราะห์การซื้อขายด้วย Historical Context"""
def __init__(self, holysheep_api_key: str):
self.holysheep_api_key = holysheep_api_key
self.base_url = "https://api.holysheep.ai/v1"
self.context_window = 24 # ชั่วโมง
self.conversation_history: List[Dict] = []
def prepare_rag_context(
self,
orderbook_df: pd.DataFrame,
current_time: datetime
) -> str:
"""เตรียม context สำหรับ RAG"""
cutoff = current_time - timedelta(hours=self.context_window)
recent_data = orderbook_df[orderbook_df["timestamp"] >= cutoff].copy()
if recent_data.empty:
return "ไม่มีข้อมูล orderbook ในช่วง 24 ชั่วโมงที่ผ่านมา"
# คำนวณ metrics สำคัญ
context_parts = []
context_parts.append("📊 **Orderbook Analysis - 24 Hour Summary**\n")
context_parts.append(f"⏰ ข้อมูลล่าสุด: {recent_data['timestamp'].max()}")
context_parts.append(f"📈 จำนวน snapshots: {len(recent_data):,}")
context_parts.append("\n**💹 Price Movement:**")
context_parts.append(f"- Mid Price ล่าสุด: ${recent_data['mid_price'].iloc[-1]:,.2f}")
context_parts.append(f"- Mid Price สูงสุด: ${recent_data['mid_price'].max():,.2f}")
context_parts.append(f"- Mid Price ต่ำสุด: ${recent_data['mid_price'].min():,.2f}")
context_parts.append(f"- Price Range: ${recent_data['mid_price'].max() - recent_data['mid_price'].min():,.2f}")
context_parts.append("\n**📉 Liquidity Analysis:**")
context_parts.append(f"- Avg Bid Volume: {recent_data['bid_volume_total'].mean():,.2f}")
context_parts.append(f"- Avg Ask Volume: {recent_data['ask_volume_total'].mean():,.2f}")
context_parts.append(f"- Bid/Ask Ratio: {recent_data['bid_volume_total'].mean() / recent_data['ask_volume_total'].mean():.4f}")
context_parts.append("\n**⚖️ Order Imbalance:**")
avg_imbalance = recent_data['bid_ask_imbalance'].mean()
if avg_imbalance > 0.1:
sentiment = "🟢 มีแรงซื้อมากกว่า (Bullish pressure)"
elif avg_imbalance < -0.1:
sentiment = "🔴 มีแรงขายมากกว่า (Bearish pressure)"
else:
sentiment = "⚪ ค่อนข้างสมดุล"
context_parts.append(f"- เฉลี่ย: {avg_imbalance:.4f} {sentiment}")
context_parts.append(f"- สูงสุด: {recent_data['bid_ask_imbalance'].max():.4f}")
context_parts.append(f"- ต่ำสุด: {recent_data['bid_ask_imbalance'].min():.4f}")
# แบ่งตามช่วงเวลา
recent_data["hour"] = pd.to_datetime(recent_data["timestamp"]).dt.hour
hourly_avg = recent_data.groupby("hour")["bid_ask_imbalance"].mean()
context_parts.append("\n**🕐 Hourly Pattern:**")
for hour, imbalance in hourly_avg.items():
bar = "█" * int(abs(imbalance) *
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง