บทนำ
ในโลกของการซื้อขายความถี่สูงหรือ HFT (High-Frequency Trading) ความเร็วคือทุกสิ่ง การทดสอบย้อนกลับหรือ Backtesting กลยุทธ์การซื้อขายบนข้อมูลประวัติศาสตร์ที่มีความละเอียดระดับ Tick-by-Tick จำเป็นต้องอาศัยระบบจัดเก็บข้อมูลที่มีประสิทธิภาพสูงและการ Query ที่ได้รับการ Optimize อย่างดี บทความนี้จะพาคุณไปทำความรู้จักกับรูปแบบ Tardis Parquet สำหรับเก็บ Order Book ประวัติศาสตร์ และเทคนิคการ Optimize SQL Query ที่จะช่วยลดเวลาการ Backtest จากหลายชั่วโมงให้เหลือเพียงไม่กี่นาที
กรณีศึกษา: ทีม Quant จากกรุงเทพฯ
บริบทธุรกิจ: ทีม Quant Developer ที่ประกอบด้วยอดีตวิศวกรจากสถาบันการเงินชั้นนำในกรุงเทพฯ ต้องการพัฒนาระบบ Backtesting สำหรับกลยุทธ์ Market Making และ Statistical Arbitrage บนตลาดหุ้นไทยและตลาด Digital Assets ทีมมีข้อมูล Order Book History ขนาด 2TB จาก 3 ปีย้อนหลัง โดยแต่ละวันมีข้อมูลประมาณ 500GB ต่อตลาด
จุดเจ็บปวดกับผู้ให้บริการเดิม: ระบบ PostgreSQL เดิมที่ใช้อยู่เกิดปัญหาหลายประการ ประการแรกคือประสิทธิภาพ Query ตกต่ำลง drasticaly เมื่อข้อมูลมีขนาดใหญ่ขึ้น การ Query ข้อมูล Order Book เพียง 1 วันใช้เวลามากกว่า 45 นาที ประการที่สองคือต้นทุน Infrastructure สูงมาก เครื่อง Database Server ระดับ High-End ราคาเกือบ $15,000 ต่อเดือน ประการที่สามคือระบบมี Downtime บ่อยครั้ง ทำให้การทดสอบไม่ต่อเนื่อง และประการสุดท้ายคือขาดความสามารถในการรองรับ Parquet Format ที่เป็นมาตรฐานในอุตสาหกรรม
เหตุผลที่เลือก HolySheep AI: ทีมได้ทดสอบ HolySheep AI และพบว่าระบบสามารถรองรับ Parquet Format ได้อย่างมีประสิทธิภาพ มีค่า Latency เฉลี่ยต่ำกว่า 50ms ซึ่งเหมาะสมกับงาน Backtesting ที่ต้องการความเร็วสูง ที่สำคัญคือราคาถูกกว่าผู้ให้บริการเดิมถึง 85% ด้วยอัตราแลกเปลี่ยนที่คุ้มค่า
ขั้นตอนการย้ายระบบ: การย้ายระบบเริ่มจากการ Export ข้อมูลจาก PostgreSQL เดิมเป็น Parquet Format โดยใช้ Apache Arrow จากนั้นทำการ Convert เป็น Tardis Parquet Schema ที่เหมาะสมกับโครงสร้าง Order Book จากนั้นอัปโหลดข้อมูลขึ้น HolySheep Storage และสุดท้ายคือการทำ Canary Deployment เพื่อทดสอบระบบใหม่ควบคู่กับระบบเดิม
ตัวชี้วัดหลังย้าย 30 วัน: เวลา Query เฉลี่ยลดลงจาก 45 นาที เหลือเพียง 3 นาที คิดเป็นการปรับปรุงประสิทธิภาพ 93% Latency เฉลือนเหลือ 180ms จากเดิมที่มากกว่า 420ms และต้นทุนรายเดือนลดลงจาก $4,200 เหลือเพียง $680 ประหยัดได้ถึง $3,520 ต่อเดือน หรือกว่า $42,000 ต่อปี
Tardis Parquet Format คืออะไร
Tardis Parquet Format เป็น Schema ที่ออกแบบมาเพื่อจัดเก็บข้อมูล Order Book ประวัติศาสตร์โดยเฉพาะ ชื่อ "Tardis" มาจากความสามารถในการเก็บข้อมูลที่มี "มิติเวลาย้อนกลับได้" (Time-Travel) เหมือนกับเครื่อง Tardis ในนิยาย Doctor Who ที่สามารถเดินทางข้ามเวลาได้
โครงสร้างหลักของ Tardis Parquet ประกอบด้วย Time Dimension สำหรับเก็บ Timestamps ระดับ Nanosecond พร้อมกับ Version ID สำหรับ Order Book Versioning ข้อมูลระดับ Price Level และ Individual Order ที่มีการ Compress ด้วย Dictionary Encoding และสุดท้ายคือ Snapshot Delta สำหรับเก็บเฉพาะส่วนที่เปลี่ยนแปลงจาก Snapshot ก่อนหน้า
{
"schema_version": "2.1",
"compression": {
"codec": "zstd",
"compression_level": 3,
"dictionary_encoding": true
},
"columns": {
"timestamp_ns": "uint64",
"version_id": "uint32",
"order_book_snapshot": {
"bids": [
{
"price": "decimal(18,8)",
"quantity": "decimal(18,8)",
"order_count": "uint32"
}
],
"asks": [
{
"price": "decimal(18,8)",
"quantity": "decimal(18,8)",
"order_count": "uint32"
}
]
}
}
}
ข้อได้เปรียบหลักของรูปแบบนี้คือการประหยัดพื้นที่จัดเก็บถึง 70% เมื่อเทียบกับ JSON หรือ CSV ธรรมดา การ Query ด้วย Timestamp Range ที่เร็วกว่า Columnar Database ทั่วไป 3-5 เท่า และการรองรับ Push-down Predicate ที่ช่วยลด I/O อย่างมาก
การตั้งค่า Environment และติดตั้ง Dependencies
ก่อนเริ่มการ Optimize SQL Query คุณต้องตั้งค่า Environment ให้พร้อมก่อน ระบบ Backtesting ที่มีประสิทธิภาพต้องการ Python 3.10 ขึ้นไป Apache Arrow และ PyArrow สำหรับจัดการ Parquet, DuckDB หรือ Apache Spark สำหรับ SQL Query Engine และ HolySheep SDK สำหรับเชื่อมต่อ API
# ติดตั้ง Dependencies ที่จำเป็น
pip install pyarrow==14.0.1
pip install duckdb==0.9.2
pip install pandas==2.1.4
pip install sqlalchemy==2.0.23
pip install holysheep-sdk==1.2.0
ตรวจสอบ Version
python --version # ควรเป็น 3.10 ขึ้นไป
pyarrow.__version__ # ควรเป็น 14.0.1
duckdb.__version__ # ควรเป็น 0.9.2
# การเชื่อมต่อ HolySheep API สำหรับ Query Parquet Data
import os
from holysheep import HolySheepClient
ตั้งค่า API Key
os.environ['HOLYSHEEP_API_KEY'] = 'YOUR_HOLYSHEEP_API_KEY'
สร้าง Client
client = HolySheepClient(
base_url='https://api.holysheep.ai/v1',
api_key=os.environ['HOLYSHEEP_API_KEY']
)
ตรวจสอบการเชื่อมต่อ
health = client.health_check()
print(f"HolySheep Status: {health.status}")
print(f"Latency: {health.latency_ms}ms")
การสร้าง Tardis Parquet และ Optimize Schema
การออกแบบ Schema ที่ดีเป็นรากฐานของการ Query ที่เร็ว สำหรับข้อมูล Order Book คุณควรแบ่งข้อมูลตาม Partition เป็น Year/Month/Day เพื่อให้สามารถ Prune Partition ได้อย่างมีประสิทธิภาพ ใช้ Column Order ที่เหมาะสมโดยให้ Timestamp อยู่เป็น Column แรกเสมอ เนื่องจาก DuckDB และ Spark จะทำ Column Pruning ตามลำดับ กำหนด Data Type ให้เหมาะสม ใช้ UInt64 สำหรับ Timestamp แทน String จะช่วยประหยัดพื้นที่และเร็วกว่า และสุดท้ายคือ Enable Dictionary Encoding สำหรับ Categorical Columns เช่น Symbol และ Exchange
import pyarrow as pa
import pyarrow.parquet as pq
from pathlib import Path
import pandas as pd
def create_tardis_parquet_schema():
"""
สร้าง Tardis Parquet Schema สำหรับ Order Book History
Optimize สำหรับ High-Frequency Backtesting
"""
# กำหนด Schema หลัก
schema = pa.schema([
# Dimension Columns (ควรอยู่ด้านบนสำหรับ Pruning)
('timestamp_ns', pa.uint64()), # Nanosecond timestamp
('version_id', pa.uint32()), # Order book version
('exchange', pa.dictionary(pa.string())), # Dictionary encoded
('symbol', pa.dictionary(pa.string())), # Dictionary encoded
# Bid Side
('bid_price_0', pa.float64()), # Best bid price
('bid_price_1', pa.float64()),
('bid_price_2', pa.float64()),
('bid_price_3', pa.float64()),
('bid_price_4', pa.float64()),
('bid_qty_0', pa.float64()), # Best bid quantity
('bid_qty_1', pa.float64()),
('bid_qty_2', pa.float64()),
('bid_qty_3', pa.float64()),
('bid_qty_4', pa.float64()),
# Ask Side
('ask_price_0', pa.float64()), # Best ask price
('ask_price_1', pa.float64()),
('ask_price_2', pa.float64()),
('ask_price_3', pa.float64()),
('ask_price_4', pa.float64()),
('ask_qty_0', pa.float64()), # Best ask quantity
('ask_qty_1', pa.float64()),
('ask_qty_2', pa.float64()),
('ask_qty_3', pa.float64()),
('ask_qty_4', pa.float64()),
# Metadata
('mid_price', pa.float64()), # (best_bid + best_ask) / 2
('spread', pa.float64()), # best_ask - best_bid
('imbalance', pa.float64()), # (bid_vol - ask_vol) / (bid_vol + ask_vol)
])
return schema
def optimize_parquet_file(input_path, output_path, row_group_size=50000):
"""
Optimize Parquet File ด้วย Compression และ Encoding ที่เหมาะสม
"""
# อ่านข้อมูล
table = pq.read_table(input_path)
# กำหนด Parquet Write Options
writer_options = pq.ParquetWriterOptions(
compression='zstd', # Zstandard compression
compression_level=3, # Balance ระหว่าง Speed และ Size
use_dictionary=True, # Enable Dictionary Encoding
write_statistics=True, # Write Statistics สำหรับ Predicate Pushdown
data_page_version='2.0' # V2 Data Pages สำหรับประสิทธิภาพดีกว่า
)
# กำหนด Row Group Size (50,000 rows เหมาะสำหรับ Memory และ Parallelism)
writer = pq.ParquetWriter(
output_path,
table.schema,
version='2.6',
use_dictionary=True,
compression='zstd'
)
# เขียนข้อมูลทีละ Row Group
total_rows = table.num_rows
for start in range(0, total_rows, row_group_size):
end = min(start + row_group_size, total_rows)
row_group = table.slice(start, end - start)
writer.write(row_group)
writer.close()
# ตรวจสอบ File Size หลัง Optimize
original_size = Path(input_path).stat().st_size
optimized_size = Path(output_path).stat().st_size
print(f"Original: {original_size / 1024**3:.2f} GB")
print(f"Optimized: {optimized_size / 1024**3:.2f} GB")
print(f"Compression Ratio: {original_size / optimized_size:.2f}x")
ตัวอย่างการใช้งาน
schema = create_tardis_parquet_schema()
print("Schema Created Successfully")
print(schema)
การเขียน SQL Query ที่ Optimize สำหรับ Tardis Parquet
การเขียน SQL Query สำหรับ Parquet Files ต้องคำนึงถึงหลายปัจจัยเพื่อให้ได้ประสิทธิภาพสูงสุด ประการแรกคือ Push-down Predicate ควรใช้ Filter Conditions ใน WHERE clause เพื่อให้ Engine สามารถ Prune Data Pages ที่ไม่จำเป็นออกได้ตั้งแต่ต้น ประการที่สองคือ Column Pruning ควร Select เฉพาะ Columns ที่จำเป็นเท่านั้น ไม่ควรใช้ SELECT * เพราะจะทำให้ต้องอ่านข้อมูลทั้งหมด ประการที่สามคือ Partition Pruning หากข้อมูลถูก Partition แล้ว ควรใช้ Partition Columns ใน WHERE clause เสมอ
import duckdb
from pathlib import Path
class TardisQueryOptimizer:
"""
Query Optimizer สำหรับ Tardis Parquet Order Book Data
ใช้ DuckDB สำหรับ SQL Query บน Parquet Files
"""
def __init__(self, parquet_path):
self.parquet_path = parquet_path
self.con = duckdb.connect(database=':memory:')
self.con.execute("SET threads TO 8")
self.con.execute("SET memory_limit TO '16GB'")
def execute_query(self, query, explain=False):
"""Execute SQL Query พร้อม Option สำหรับดู Execution Plan"""
if explain:
result = self.con.execute(f"EXPLAIN ANALYZE {query}").fetchall()
for row in result:
print(row[0])
else:
return self.con.execute(query).fetchdf()
def get_market_making_metrics(self, exchange, symbol, start_ts, end_ts):
"""
คำนวณ Market Making Metrics จาก Order Book Data
Metrics ที่คำนวณ:
- Spread Statistics (Half-spread, Effective Spread)
- Order Imbalance
- Depth at Price Levels
- Price Impact
"""
query = f"""
WITH filtered_data AS (
-- กรองข้อมูลเฉพาะที่ต้องการด้วย Predicate Pushdown
SELECT
timestamp_ns,
bid_price_0,
bid_qty_0,
ask_price_0,
ask_qty_0,
bid_price_1,
bid_price_2,
bid_price_3,
bid_price_4,
ask_price_1,
ask_price_2,
ask_price_3,
ask_price_4,
bid_qty_1 + bid_qty_2 + bid_qty_3 + bid_qty_4 AS total_bid_qty_deep,
ask_qty_1 + ask_qty_2 + ask_qty_3 + ask_qty_4 AS total_ask_qty_deep,
bid_qty_0 + bid_qty_1 + bid_qty_2 + bid_qty_3 + bid_qty_4 AS total_bid_qty,
ask_qty_0 + ask_qty_1 + ask_qty_2 + ask_qty_3 + ask_qty_4 AS total_ask_qty
FROM '{self.parquet_path}'
WHERE
exchange = '{exchange}'
AND symbol = '{symbol}'
AND timestamp_ns >= {start_ts}
AND timestamp_ns < {end_ts}
),
metrics AS (
SELECT
-- Convert timestamp_ns to readable format
CAST(timestamp_ns / 1e9 AS TIMESTAMP) AS timestamp,
-- Spread Metrics
(ask_price_0 - bid_price_0) / ((ask_price_0 + bid_price_0) / 2) AS relative_spread,
(ask_price_0 - bid_price_0) / 2 AS half_spread_bps,
-- Imbalance Metrics
CASE
WHEN (total_bid_qty + total_ask_qty) > 0
THEN (total_bid_qty - total_ask_qty) / (total_bid_qty + total_ask_qty)
ELSE 0
END AS order_imbalance,
-- Depth Metrics (Volume Weighted Average Price at each level)
(bid_price_0 * bid_qty_0 + bid_price_1 * bid_qty_1 +
bid_price_2 * bid_qty_2 + bid_price_3 * bid_qty_3 +
bid_price_4 * bid_qty_4) /
NULLIF(bid_qty_0 + bid_qty_1 + bid_qty_2 + bid_qty_3 + bid_qty_4, 0) AS vwap_bid,
(ask_price_0 * ask_qty_0 + ask_price_1 * ask_qty_1 +
ask_price_2 * ask_qty_2 + ask_price_3 * ask_qty_3 +
ask_price_4 * ask_qty_4) /
NULLIF(ask_qty_0 + ask_qty_1 + ask_qty_2 + ask_qty_3 + ask_qty_4, 0) AS vwap_ask
FROM filtered_data
)
SELECT
-- Time-based Aggregations
DATE_TRUNC('minute', timestamp) AS minute_bucket,
-- Summary Statistics
AVG(relative_spread) * 10000 AS avg_spread_bps,
STDDEV(relative_spread) * 10000 AS std_spread_bps,
AVG(order_imbalance) AS avg_imbalance,
STDDEV(order_imbalance) AS std_imbalance,
MAX(order_imbalance) AS max_imbalance,
MIN(order_imbalance) AS min_imbalance,
COUNT(*) AS data_points
FROM metrics
GROUP BY minute_bucket
ORDER BY minute_bucket
"""
return self.execute_query(query)
def backtest_market_making_strategy(
self,
exchange,
symbol,
start_ts,
end_ts,
inventory_skew=0,
max_position=1000
):
"""
Backtest Simple Market Making Strategy
Strategy Logic:
- Place bid at mid - half_spread
- Place ask at mid + half_spread
- Adjust quotes based on order imbalance
- Inventory skew management
"""
query = f"""
WITH orderbook AS (
SELECT
timestamp_ns,
bid_price_0 AS bid,
ask_price_0 AS ask,
(bid_price_0 + ask_price_0) / 2 AS mid,
(ask_price_0 - bid_price_0) AS spread,
bid_qty_0 AS bid_depth,
ask_qty_0 AS ask_depth,
bid_qty_0 / NULLIF(bid_qty_0 + ask_qty_0, 0) AS bid_ratio,
CASE
WHEN bid_qty_0 + ask_qty_0 > 0
THEN (bid_qty_0 - ask_qty_0) / (bid_qty_0 + ask_qty_0)
ELSE 0
END AS imbalance
FROM '{self.parquet_path}'
WHERE
exchange = '{exchange}'
AND symbol = '{symbol}'
AND timestamp_ns >= {start_ts}
AND timestamp_ns < {end_ts}
),
quotes AS (
SELECT
*,
-- Inventory-adjusted quotes
bid * (1 - imbalance * {inventory_skew}) AS adj_bid,
ask * (1 + imbalance * {inventory_skew}) AS adj_ask,
-- Half spread in price terms
spread / 2 AS half_spread
FROM orderbook
),
trades AS (
SELECT
*,
-- Simulate trade execution
CASE
WHEN bid_ratio > 0.6 AND bid_ratio < 0.7 THEN 'hit_bid'
WHEN bid_ratio > 0.3 AND bid_ratio < 0.4 THEN 'lift_ask'
ELSE 'no_trade'
END AS trade_side,
-- Simulated PnL
CASE
WHEN bid_ratio > 0.6 AND bid_ratio < 0.7 THEN -half_spread
WHEN bid_ratio > 0.3 AND bid_ratio < 0.4 THEN half_spread
ELSE 0
END AS pnl_per_trade
FROM quotes
)
SELECT
DATE_TRUNC('minute', CAST(timestamp_ns / 1e9 AS TIMESTAMP)) AS minute_ts,
SUM(CASE WHEN trade_side = 'hit_bid' THEN 1 ELSE 0 END) AS bid_hits,
SUM(CASE WHEN trade_side = 'lift_ask' THEN 1 ELSE 0 END) AS ask_lifts,
SUM(pnl_per_trade) AS minute_pnl,
AVG(spread) AS avg_spread,
AVG(imbalance) AS avg_imbalance
FROM trades
GROUP BY minute_ts
ORDER BY minute_ts
"""
return self.execute_query(query)
ตัวอย่างการใช้งาน
optimizer = TardisQueryOptimizer('/data/orderbook/tardis_parquet/')
คำนวณ Metrics สำหรับ BTC/USDT
result = optimizer.get_market_making_metrics(
exchange='binance',
symbol='BTCUSDT',
start_ts=1704067200000000000, # 2024-01-01 00:00:00
end_ts=1704153600000000000 # 2024-01-02 00:00:00
)
print(f"Query completed. Rows returned: {len(result)}")
print(result.head())
เทคนิค Advanced SQL Optimization
สำหรับการทำ Backtesting ที่ซับซ้อนมากขึ้น มีเทคนิค Advanced หลายอย่างที่จะช่วยเพิ่มประสิทธิภาพได้อีกมาก
การใช้ Window Functions อย่างมีประสิทธิภาพ: Window Functions เช่น LAG, LEAD,
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง