บทนำ: ทำไมต้องดึงข้อมูล Orderbook จาก Bitget
ถ้าคุณเป็นนักวิจัยเชิงปริมาณ (Quantitative Researcher) หรือนักพัฒนา Bot Trade คุณคงรู้ว่าข้อมูล "Orderbook" หรือ "ราคาซื้อขาย" ของ Bitget นั้นสำคัญมาก ข้อมูลนี้ช่วยให้เราเข้าใจว่าตลาดมี "ความลึก" (Depth) มากแค่ไหน และถ้าเราซื้อขายจริงจะเสียค่า Slippage เท่าไหร่
ปัญหาคือการดึงข้อมูลเหล่านี้โดยตรงจาก Exchange มักยุ่งยาก ต้องตั้ง Server เอง ดูแลระบบ จ่ายค่า Data Feed แพงๆ แต่วันนี้เราจะมาแนะนำวิธีง่ายๆ โดยใช้
HolySheep AI เป็นตัวกลางดึงข้อมูลจาก Tardis Bitget แทน
ในบทความนี้เราจะครอบคลุม:
- การเชื่อมต่อ API ผ่าน HolySheep สำหรับ Bitget perpetual orderbook
- วิธีเก็บข้อมูล Depth อย่างเป็นระบบ
- การคำนวณ Impact Cost (ค่ากระทบกระเทือนราคา)
- การนำข้อมูลไปใช้ Backtest กลยุทธ์
พื้นฐานที่ควรรู้ก่อนเริ่ม
ก่อนจะลงมือทำ มาทำความเข้าใจคำศัพท์พื้นฐานกันก่อน:
- Orderbook = รายการคำสั่งซื้อ-ขายที่รอการจับคู่ บอกว่าราคานี้มีคนเท่าไหร่รอซื้อ/ขาย
- Depth = ความลึกของ Orderbook หรือปริมาณรวมที่รอซื้อขายในแต่ละราคา
- Impact Cost = ค่าใช้จ่ายที่เกิดขึ้นเพิ่มเติมเมื่อเราซื้อขาย volume ใหญ่ ทำให้ราคาเลื่อน
- Backtest = การทดสอบกลยุทธ์กับข้อมูลในอดีต
ขั้นตอนที่ 1: สมัคร HolySheep AI และรับ API Key
ขั้นแรกเราต้องมีบัญชี HolySheep ก่อน ไปที่
สมัครที่นี่ แล้วทำตามนี้:
- กรอกอีเมลและรหัสผ่าน
- ยืนยันอีเมล
- ไปที่หน้า Dashboard → API Keys
- กดปุ่ม "สร้าง API Key ใหม่"
- ตั้งชื่อ Key เช่น "Bitget-Research"
- คัดลอก Key ที่ได้เก็บไว้ (จะเห็นได้แค่ครั้งเดียว)
💡 เคล็ดลับ: HolySheep มีเครดิตฟรีเมื่อลงทะเบียน ลองใช้ทดลองก่อนตัดสินใจซื้อแพ็กเกจ
ขั้นตอนที่ 2: เตรียม Python Environment
เราจะใช้ Python ในการเขียนโค้ด ถ้ายังไม่มี Python ให้ดาวน์โหลดจาก python.org ก่อน จากนั้นติดตั้ง Library ที่ต้องใช้:
pip install requests pandas numpy matplotlib
ขั้นตอนที่ 3: เชื่อมต่อ HolySheep API กับ Tardis Bitget
ต่อไปเราจะเขียนโค้ดเพื่อดึงข้อมูล Orderbook จาก Bitgetผ่าน HolySheep
import requests
import json
import time
from datetime import datetime
กำหนดค่าพื้นฐาน
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # แทนที่ด้วย Key ของคุณ
Headers สำหรับ Authentication
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
def get_bitget_orderbook(symbol="BTC/USDT:USDT", limit=20):
"""
ดึงข้อมูล Orderbook จาก Bitget ผ่าน HolySheep
symbol: คู่เทรด (default: BTC/USDT perpetual)
limit: จำนวนระดับราคาที่ต้องการ
"""
endpoint = f"{BASE_URL}/market/orderbook"
params = {
"exchange": "bitget",
"symbol": symbol,
"limit": limit,
"depth_type": "snapshot" # หรือ "incremental" สำหรับ streaming
}
try:
response = requests.get(endpoint, headers=headers, params=params, timeout=10)
response.raise_for_status()
data = response.json()
return {
"timestamp": datetime.now().isoformat(),
"bids": data.get("bids", []), # ราคาซื้อ [ราคา, ปริมาณ]
"asks": data.get("asks", []), # ราคาขาย [ราคา, ปริมาณ]
"symbol": symbol
}
except requests.exceptions.RequestException as e:
print(f"❌ เกิดข้อผิดพลาด: {e}")
return None
ทดสอบการเชื่อมต่อ
if __name__ == "__main__":
print("🔄 กำลังดึงข้อมูล Orderbook จาก Bitget...")
result = get_bitget_orderbook()
if result:
print(f"✅ ได้ข้อมูลแล้ว - {result['symbol']}")
print(f"⏰ Timestamp: {result['timestamp']}")
print(f"📊 ราคาซื้อ (Bids) สูงสุด 5 รายการ:")
for i, (price, qty) in enumerate(result['bids'][:5]):
print(f" {i+1}. ราคา {price} | ปริมาณ {qty}")
print(f"📊 ราคาขาย (Asks) สูงสุด 5 รายการ:")
for i, (price, qty) in enumerate(result['asks'][:5]):
print(f" {i+1}. ราคา {price} | ปริมาณ {qty}")
ขั้นตอนที่ 4: เก็บข้อมูล Depth อย่างเป็นระบบ (Depth Archival)
สำหรับงานวิจัย เราต้องเก็บข้อมูล Depth ไว้ใช้ในอนาคต โค้ดด้านล่างจะดึงข้อมูลทุกๆ 5 วินาทีและบันทึกลงไฟล์:
import requests
import json
import time
import pandas as pd
from datetime import datetime
from pathlib import Path
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
def fetch_and_save_depth(symbol="BTC/USDT:USDT", interval_seconds=5, duration_minutes=60):
"""
เก็บข้อมูล Depth ทุก N วินาที และบันทึกลง CSV
Parameters:
- symbol: คู่เทรด
- interval_seconds: ความถี่ในการดึงข้อมูล (วินาที)
- duration_minutes: ระยะเวลาการเก็บข้อมูล (นาที)
"""
records = []
start_time = time.time()
end_time = start_time + (duration_minutes * 60)
print(f"🚀 เริ่มเก็บข้อมูล Depth สำหรับ {symbol}")
print(f" ความถี่: ทุก {interval_seconds} วินาที")
print(f" ระยะเวลา: {duration_minutes} นาที")
iteration = 0
while time.time() < end_time:
iteration += 1
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# ดึงข้อมูล Orderbook
endpoint = f"{BASE_URL}/market/orderbook"
params = {
"exchange": "bitget",
"symbol": symbol,
"limit": 50,
"depth_type": "snapshot"
}
try:
response = requests.get(endpoint, headers=headers, params=params, timeout=10)
data = response.json()
# คำนวณ Depth ของ Bid และ Ask
total_bid_volume = sum(float(qty) for _, qty in data.get("bids", []))
total_ask_volume = sum(float(qty) for _, qty in data.get("asks", []))
# ราคาที่ดีที่สุด
best_bid = float(data["bids"][0][0]) if data.get("bids") else 0
best_ask = float(data["asks"][0][0]) if data.get("asks") else 0
spread = best_ask - best_bid
spread_pct = (spread / best_bid) * 100 if best_bid > 0 else 0
record = {
"timestamp": current_time,
"iteration": iteration,
"best_bid": best_bid,
"best_ask": best_ask,
"spread": spread,
"spread_pct": round(spread_pct, 4),
"total_bid_volume": total_bid_volume,
"total_ask_volume": total_ask_volume,
"mid_price": (best_bid + best_ask) / 2
}
records.append(record)
print(f" [{current_time}] Iteration #{iteration} | Spread: {spread_pct:.4f}% | Bid Vol: {total_bid_volume:.2f}")
except Exception as e:
print(f" ⚠️ Error ใน iteration #{iteration}: {e}")
# รอตาม interval ที่กำหนด
time.sleep(interval_seconds)
# บันทึกลง CSV
df = pd.DataFrame(records)
filename = f"depth_data_{symbol.replace('/', '_').replace(':', '_')}_{datetime.now().strftime('%Y%m%d_%H%M')}.csv"
df.to_csv(filename, index=False)
print(f"\n✅ เก็บข้อมูลเสร็จสิ้น! บันทึก {len(records)} รายการ")
print(f"📁 ไฟล์: {filename}")
return df
รันการเก็บข้อมูล
if __name__ == "__main__":
# ทดสอบเก็บข้อมูล 5 นาที ทุก 3 วินาที
df = fetch_and_save_depth(
symbol="BTC/USDT:USDT",
interval_seconds=3,
duration_minutes=5
)
# แสดงสถิติเบื้องต้น
print("\n📈 สถิติเบื้องต้น:")
print(df.describe())
ขั้นตอนที่ 5: คำนวณ Impact Cost
Impact Cost คือ "ค่ากระทบ" ที่เกิดขึ้นเมื่อเราซื้อขาย volume มาก ยิ่งซื้อมาก ราคายิ่งเลื่อน เป็นต้นว่า ถ้าเราต้องการซื้อ 1 BTC แต่ใน Orderbook มีแค่ 0.3 BTC ที่ราคาที่ต้องการ ส่วนที่เหลือต้องซื้อในราคาที่แพงกว่า
import requests
import pandas as pd
import numpy as np
from datetime import datetime
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
def calculate_impact_cost(orderbook_data, target_volume, side="buy"):
"""
คำนวณ Impact Cost สำหรับ volume ที่กำหนด
Parameters:
- orderbook_data: dict ที่มี 'bids' และ 'asks'
- target_volume: ปริมาณที่ต้องการซื้อ/ขาย (เช่น 1 = 1 BTC)
- side: 'buy' หรือ 'sell'
Returns:
- dict ที่มีรายละเอียดการคำนวณ
"""
if side == "buy":
levels = orderbook_data.get("asks", []) # การซื้อ = ไล่ราคาขายขึ้น
else:
levels = orderbook_data.get("bids", []) # การขาย = ไล่ราคาซื้อลง
remaining_volume = target_volume
total_cost = 0
filled_levels = []
avg_price = 0
# คำนวณราคาเฉลี่ยที่ได้
for price, qty in levels:
price = float(price)
qty = float(qty)
if remaining_volume <= 0:
break
# ปริมาณที่จะซื้อในระดับราคานี้
fill_qty = min(remaining_volume, qty)
total_cost += fill_qty * price
remaining_volume -= fill_qty
filled_levels.append({
"price": price,
"qty": fill_qty,
"cum_qty": target_volume - remaining_volume,
"cum_cost": total_cost
})
# ราคาเฉลี่ยที่ได้
if target_volume > 0:
avg_price = total_cost / target_volume
# เปรียบเทียบกับราคา Best Price
if side == "buy":
best_price = float(levels[0][0]) if levels else 0
else:
best_price = float(levels[0][0]) if levels else 0
# Impact Cost
if best_price > 0:
impact_cost_pct = ((avg_price - best_price) / best_price) * 100
else:
impact_cost_pct = 0
return {
"side": side,
"target_volume": target_volume,
"best_price": best_price,
"avg_price": avg_price,
"total_cost": total_cost,
"filled_volume": target_volume - remaining_volume,
"impact_cost_pct": round(impact_cost_pct, 4),
"impact_cost_abs": round(avg_price - best_price, 2),
"filled_levels": filled_levels
}
def get_live_orderbook_and_analyze(symbol="BTC/USDT:USDT", volumes=[0.1, 0.5, 1.0, 5.0]):
"""
ดึงข้อมูล Orderbook ปัจจุบัน แล้ววิเคราะห์ Impact Cost สำหรับหลายระดับ Volume
"""
endpoint = f"{BASE_URL}/market/orderbook"
params = {
"exchange": "bitget",
"symbol": symbol,
"limit": 100,
"depth_type": "snapshot"
}
print(f"🔄 ดึงข้อมูล Orderbook สดจาก {symbol}...\n")
try:
response = requests.get(endpoint, headers=headers, params=params, timeout=10)
orderbook = response.json()
results = []
for volume in volumes:
# วิเคราะห์การซื้อ
buy_impact = calculate_impact_cost(orderbook, volume, "buy")
# วิเคราะห์การขาย
sell_impact = calculate_impact_cost(orderbook, volume, "sell")
results.append({
"volume": volume,
"buy_impact_pct": buy_impact["impact_cost_pct"],
"buy_impact_abs": buy_impact["impact_cost_abs"],
"sell_impact_pct": sell_impact["impact_cost_pct"],
"sell_impact_abs": sell_impact["impact_cost_abs"],
"mid_price": orderbook["asks"][0][0] # ราคาเฉลี่ยปัจจุบัน
})
# แสดงผล
df = pd.DataFrame(results)
print("=" * 70)
print("📊 ตาราง Impact Cost Analysis")
print("=" * 70)
print(f"Symbol: {symbol}")
print(f"Best Bid: {orderbook['bids'][0][0]} | Best Ask: {orderbook['asks'][0][0]}")
print("-" * 70)
print(df.to_string(index=False))
print("=" * 70)
return df, orderbook
except Exception as e:
print(f"❌ เกิดข้อผิดพลาด: {e}")
return None, None
ทดสอบ
if __name__ == "__main__":
results_df, raw_orderbook = get_live_orderbook_and_analyze(
symbol="BTC/USDT:USDT",
volumes=[0.1, 0.5, 1.0, 2.0, 5.0]
)
if results_df is not None:
# วาดกราฟ Impact Cost
import matplotlib.pyplot as plt
plt.figure(figsize=(10, 6))
plt.plot(results_df['volume'], results_df['buy_impact_pct'], 'r-o', label='Buy Impact %')
plt.plot(results_df['volume'], results_df['sell_impact_pct'], 'g-o', label='Sell Impact %')
plt.xlabel('Volume (BTC)')
plt.ylabel('Impact Cost (%)')
plt.title('Impact Cost vs Volume - Bitget BTC/USDT Perpetual')
plt.legend()
plt.grid(True, alpha=0.3)
plt.savefig('impact_cost_analysis.png', dpi=150)
print("\n📈 บันทึกกราฟ: impact_cost_analysis.png")
ขั้นตอนที่ 6: Backtest กลยุทธ์ง่ายๆ
ตอนนี้เรามีข้อมูล Depth แล้ว มาลองทำ Backtest กลยุทธ์ง่ายๆ กัน ตัวอย่างเช่น กลยุทธ์ "Momentum" ที่ซื้อเมื่อราคาขึ้น และขายเมื่อราคาลง โดยคำนึงถึง Impact Cost จริง
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime
def simple_momentum_backtest_with_impact(df_depth, initial_capital=10000, trade_size=0.1):
"""
Backtest กลยุทธ์ Momentum พร้อมคำนึงถึง Impact Cost จริง
Parameters:
- df_depth: DataFrame จากการเก็บข้อมูล Depth
- initial_capital: ทุนเริ่มต้น (USD)
- trade_size: ปริมาณ BTC ที่ซื้อขายต่อครั้ง
"""
# คำนวณ mid price
df = df_depth.copy()
df['mid_price'] = (df['best_bid'] + df['best_ask']) / 2
# คำนวณ returns
df['returns'] = df['mid_price'].pct_change()
# สร้าง Signal: 1 = Long, -1 = Short, 0 = Hold
# Simple: ถ้า return > threshold = Long, < -threshold = Short
threshold = 0.001 # 0.1%
df['signal'] = 0
df.loc[df['returns'] > threshold, 'signal'] = 1
df.loc[df['returns'] < -threshold, 'signal'] = -1
# สร้าง Impact Cost จาก spread และ volume
# สมมติ impact ~ spread/2 + volume_factor
df['buy_impact'] = (df['spread_pct'] / 2 + 0.02) * trade_size # 2% base impact
df['sell_impact'] = (df['spread_pct'] / 2 + 0.02) * trade_size
# จำลองพอร์ต
capital = initial_capital
position = 0 # จำนวน BTC ที่ถือ
position_value = 0
portfolio_values = []
trades = []
for i, row in df.iterrows():
current_price = row['mid_price']
signal = row['signal']
impact_cost = row['buy_impact'] if signal == 1 else row['sell_impact']
if signal == 1 and position == 0: # ซื้อ
# หัก impact cost
effective_price = current_price * (1 + impact_cost/100)
cost = effective_price * trade_size
if capital >= cost:
position = trade_size
capital -= cost
trades.append({
'timestamp': row['timestamp'],
'type': 'BUY',
'price': effective_price,
'volume': trade_size,
'cost': cost,
'impact_pct': impact_cost
})
elif signal == -1 and position > 0: # ขาย
effective_price = current_price * (1 - impact_cost/100)
revenue = effective_price * position
capital += revenue
trades.append({
'timestamp': row['timestamp'],
'type': 'SELL',
'price': effective_price,
'volume': position,
'revenue': revenue,
'impact_pct': impact_cost
})
position = 0
# คำนวณมูลค่าพอร์ต
portfolio_value = capital + (position * current_price)
portfolio_values.append(portfolio_value)
df['portfolio_value'] = portfolio_values
# คำนวณผลตอบแทน
total_return = (portfolio_values[-1] - initial_capital) / initial_capital * 100
# คำนวณ Sharpe Ratio
df['portfolio_returns'] = df['portfolio_value'].pct_change()
sharpe = df['portfolio_returns'].mean() / df['portfolio_returns'].std() * np.sqrt(288) if df['portfolio_returns'].std() > 0 else 0
# คำนวณ max drawdown
df['cummax'] = df['portfolio_value'].cummax()
df['drawdown'] = (df['portfolio_value'] - df['cummax']) / df['cummax'] * 100
max_drawdown = df['drawdown'].min()
# สรุปผล
print("\n" + "=" * 60)
print("📊 BACKTEST RESULTS - Momentum Strategy with Impact Cost")
print("=" * 60)
print(f"⏰ Period: {df['timestamp'].iloc[0]} to {df['timestamp'].iloc[-1]}")
print(f"💰 Initial Capital: ${initial_capital:,.2f}")
print(f"💵 Final Portfolio: ${portfolio_values[-1]:,.2f}")
print(f"📈 Total Return: {total_return:.2f}%")
print(f"📉 Max Drawdown: {max_drawdown:.2f}%")
print(f"📐 Sharpe Ratio: {sharpe:.2f}")
print(f"🔄 Total Trades: {len(trades)}")
print("-" * 60)
# รายละเอียด Trades
if trades:
trades_df = pd.DataFrame(trades)
print("\n📋 Trade History (5 รายการล่าสุด):")
print(trades_df.tail().to_string(index=False))
# วาดกราฟ
fig, axes = plt.subplots(2, 1, figsize=(12, 8))
# Portfolio Value
axes[0].plot(df['portfolio_value'], 'b-', linewidth=1.5)
axes[0].set_title('Portfolio Value Over Time')
axes[0].set_ylabel('Value (USD)')
axes[0].grid(True, alpha=0.3)
# Drawdown
axes[1].fill_between(df.index, df['drawdown'], 0, color='red', alpha=0.3)
axes[1].set_title('Drawdown Over Time')
axes[1].set_ylabel('Drawdown (%)')
axes[1].set_xlabel('Time')
axes[1].grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig('backtest_results.png', dpi=150)
print("\n📈 บันทึกกราฟ: backtest_results.png")
return df, trades, {
'total_return': total_return,
'max_drawdown': max_drawdown,
'sharpe_ratio': sharpe,
'total_trades': len(trades)
}
ทดสอบกับข้อมูลจริง
if __name__ == "__main__":
try:
# โหลดข้อมูลที่เก็บไว้
import glob
csv_files = glob.glob("depth_data_*.csv")
if csv_files:
latest_file = max(csv_files, key=lambda x: x)
print(f"📂 โหลดข้อมูลจาก: {latest_file}")
df = pd.read_csv(latest_file)
# รัน Backtest
results_df, trades, metrics = simple_momentum_backtest_with_impact(
df,
initial_capital=10000,
trade_size=0.1
)
else:
print("⚠️ ไม
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง