ในยุคที่ข้อมูลคริปโตมีปริมาณมหาศาล การเขียน SQL query ด้วยตนเองทุกครั้งที่ต้องการวิเคราะห์กราฟราคา ปริมาณการซื้อขาย หรือ on-chain metrics อาจเป็นงานที่ใช้เวลามากเกินจำเป็น บทความนี้จะพาคุณสร้าง Text-to-SQL Crypto Assistant ที่แปลงคำถามภาษาธรรมชาติเป็น SQL query ที่ทำงานกับ Tardis API ได้ทันที โดยใช้ HolySheep AI เป็นหัวใจหลักของระบบ
ทำความรู้จัก Tardis API และข้อมูลสกุลเงินดิจิทัล
Tardis API เป็นบริการที่รวบรวมข้อมูลการซื้อขายสกุลเงินดิจิทัลจากหลาย Exchange ไม่ว่าจะเป็น Binance, Coinbase, Kraken หรือ Bybit ฐานข้อมูลประกอบด้วยตารางหลัก 3 ตารางที่เราจะใช้บ่อย:
- trades - ข้อมูลการซื้อขายรายวินาที ประกอบด้วย timestamp, price, amount, side, exchange
- orderbook_snapshot - ภาพรวมของออเดอร์ที่รอดำเนินการ ณ ช่วงเวลาใดเวลาหนึ่ง
- ohlcv - ข้อมูลราคาเปิด สูงสุด ต่ำสุด ปิด พร้อมปริมาณการซื้อขายรายชั่วโมง/วัน
สถาปัตยกรรม Text-to-SQL Crypto Assistant
ระบบที่เราจะสร้างประกอบด้วย 3 ส่วนหลัก ระบบนี้ใช้หลักการ Chain-of-Thought Prompting เพื่อให้ LLM วิเคราะห์คำถามทีละขั้นตอนก่อนสร้าง SQL:
┌─────────────────────────────────────────────────────────────┐
│ Text-to-SQL Pipeline │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. User Input: "ราคา BTC เฉลี่ย 7 วันที่ผ่านมาเท่าไร?" │
│ │ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ Schema Analyzer │ ← ดึง schema ของฐานข้อมูล Tardis │
│ └────────┬────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ Intent Classifier│ ← จำแนกประเภทคำถาม (price/query) │
│ └────────┬────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ SQL Generator │ ← สร้าง SQL จาก prompt + schema │
│ └────────┬────────┘ │
│ │ │
│ ▼ │
│ 2. Generated SQL: "SELECT AVG(close) FROM ohlcv WHERE..." │
│ │ │
│ ▼ │
│ 3. Execute on Tardis → Return Results │
│ │
└─────────────────────────────────────────────────────────────┘การตั้งค่า HolySheep API Client
ก่อนเริ่มต้นเขียนโค้ด คุณต้องมี API key จาก สมัคร HolySheep AI ฟรี ก่อน ระบบนี้รองรับทั้ง GPT-4.1, Claude Sonnet 4.5 และ DeepSeek V3.2 โดยสำหรับงาน Text-to-SQL ที่ต้องการความแม่นยำสูง เราแนะนำให้ใช้ DeepSeek V3.2 เพราะราคาถูกกว่าถึง 20 เท่าเมื่อเทียบกับ Claude
import os
import httpx
import json
from typing import Optional, List, Dict, Any
from dataclasses import dataclass
from datetime import datetime, timedelta
กำหนดค่า base_url ตามที่กำหนด
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
@dataclass
class TardisSchema:
"""Schema ของฐานข้อมูล Tardis สำหรับ Text-to-SQL"""
tables: List[Dict[str, Any]]
def to_prompt_context(self) -> str:
"""แปลง schema เป็น text สำหรับใส่ใน prompt"""
context = "📊 DATABASE SCHEMA:\n\n"
for table in self.tables:
context += f"Table: {table['name']}\n"
context += f"Description: {table.get('description', 'N/A')}\n"
context += "Columns:\n"
for col in table.get('columns', []):
context += f" - {col['name']}: {col['type']} ({col.get('description', 'N/A')})\n"
context += "\n"
return context
Schema ตัวอย่างสำหรับ Tardis
TARDIS_SCHEMA = TardisSchema(
tables=[
{
"name": "ohlcv",
"description": "ราคา OHLCV รายนาที/ชั่วโมง/วัน",
"columns": [
{"name": "symbol", "type": "VARCHAR", "description": "ชื่อคู่เทรด เช่น BTCUSDT"},
{"name": "timestamp", "type": "TIMESTAMP", "description": "เวลาของแท่งเทียน"},
{"name": "open", "type": "DECIMAL(18,8)", "description": "ราคาเปิด"},
{"name": "high", "type": "DECIMAL(18,8)", "description": "ราคาสูงสุด"},
{"name": "low", "type": "DECIMAL(18,8)", "description": "ราคาต่ำสุด"},
{"name": "close", "type": "DECIMAL(18,8)", "description": "ราคาปิด"},
{"name": "volume", "type": "DECIMAL(18,8)", "description": "ปริมาณการซื้อขาย"},
{"name": "exchange", "type": "VARCHAR", "description": "ชื่อ Exchange"},
]
},
{
"name": "trades",
"description": "รายการซื้อขายรายวินาที",
"columns": [
{"name": "id", "type": "BIGINT", "description": "Trade ID"},
{"name": "symbol", "type": "VARCHAR", "description": "ชื่อคู่เทรด"},
{"name": "price", "type": "DECIMAL(18,8)", "description": "ราคาที่ซื้อขาย"},
{"name": "amount", "type": "DECIMAL(18,8)", "description": "จำนวนที่ซื้อขาย"},
{"name": "side", "type": "VARCHAR", "description": "buy หรือ sell"},
{"name": "timestamp", "type": "TIMESTAMP", "description": "เวลาที่เกิด trade"},
{"name": "exchange", "type": "VARCHAR", "description": "Exchange ที่เกิด trade"},
]
}
]
)
class HolySheepLLM:
"""Client สำหรับเรียก HolySheep API"""
def __init__(self, api_key: str, model: str = "deepseek-v3.2"):
self.api_key = api_key
self.model = model
self.base_url = HOLYSHEEP_BASE_URL
def chat_completion(
self,
messages: List[Dict[str, str]],
temperature: float = 0.3,
max_tokens: int = 2000
) -> str:
"""เรียก HolySheep API สำหรับ chat completion"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
with httpx.Client(timeout=30.0) as client:
response = client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
result = response.json()
return result["choices"][0]["message"]["content"]
print("✅ HolySheep LLM Client initialized")สร้าง Text-to-SQL Engine สำหรับ Crypto Analysis
ต่อไปคือหัวใจหลักของระบบ - Text-to-SQL Engine ที่ทำหน้าที่แปลงคำถามภาษาธรรมชาติเป็น SQL query โดยใช้ Chain-of-Thought reasoning เพื่อเพิ่มความแม่นยำ
import re
from typing import Tuple, Optional
class CryptoTextToSQL:
"""Text-to-SQL Engine สำหรับวิเคราะห์ข้อมูลคริปโต"""
def __init__(self, llm_client: HolySheepLLM, schema: TardisSchema):
self.llm = llm_client
self.schema = schema
def generate_sql(self, user_question: str, context: Optional[str] = None) -> str:
"""แปลงคำถามภาษาธรรมชาติเป็น SQL query"""
system_prompt = f"""คุณเป็นผู้เชี่ยวชาญ SQL สำหรับฐานข้อมูล Tardis ที่เก็บข้อมูลสกุลเงินดิจิทัล
หน้าที่ของคุณคือแปลงคำถามภาษาธรรมชาติเป็น SQL query ที่ถูกต้อง
{self.schema.to_prompt_context()}
กฎสำคัญ:
1. ตอบกลับเฉพาะ SQL query เท่านั้น โดยอยู่ใน code block
2. ใช้ timestamp รูปแบบ ISO 8601 สำหรับการกรองวันที่
3. สำหรับคำถามเกี่ยวกับ "ราคาเฉลี่ย" ใช้ AVG(close)
4. สำหรับคำถามเกี่ยวกับ "ราคาสูงสุด/ต่ำสุด" ใช้ MAX/MIN
5. Symbol ต้องเป็นตัวพิมพ์ใหญ่ เช่น BTCUSDT, ETHUSDT
ตัวอย่าง:
- คำถาม: "ราคา BTC ล่าสุด"
- SQL: SELECT close as price, timestamp FROM ohlcv
WHERE symbol = 'BTCUSDT' ORDER BY timestamp DESC LIMIT 1
- คำถาม: "ปริมาณ ETH 7 วันที่แล้ว"
- SQL: SELECT SUM(volume) as total_volume FROM ohlcv
WHERE symbol = 'ETHUSDT' AND timestamp >= NOW() - INTERVAL '7 days'"""
user_prompt = f"""คำถาม: {user_question}
{('บริบทเพิ่มเติม: ' + context) if context else ''}
ดำเนินการ:
1. วิเคราะห์คำถามเพื่อหาว่าต้องการข้อมูลอะไร
2. ระบุ table ที่ต้องใช้
3. เขียน SQL query ที่ถูกต้อง
4. ตอบกลับเฉพาะ SQL query ใน code block"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
]
response = self.llm.chat_completion(messages, temperature=0.2)
# Extract SQL from code block
sql = self._extract_sql(response)
return sql
def _extract_sql(self, response: str) -> str:
"""ดึง SQL ออกจาก response ที่อยู่ใน code block"""
# ลองหาใน code block ก่อน
code_match = re.search(r'``sql\s*(.*?)\s*``', response, re.DOTALL | re.IGNORECASE)
if code_match:
return code_match.group(1).strip()
# ถ้าไม่เจอ ลองหา code block ทั่วไป
code_match = re.search(r'``\s*(.*?)\s*``', response, re.DOTALL)
if code_match:
return code_match.group(1).strip()
# ถ้าไม่เจอ ส่งคืนทั้งหมดโดยตรง
return response.strip()
def execute_query(self, sql: str, tardis_client) -> List[Dict]:
"""Execute SQL query บน Tardis API"""
# ตัวอย่างการใช้ Tardis Query API
query_payload = {
"sql": sql,
"timeout": 30
}
# ที่นี่จะมีการเรียก Tardis API จริง
# response = tardis_client.query(query_payload)
# return response["results"]
# Mock response สำหรับ demo
print(f"📤 Executing SQL: {sql}")
return []
ทดสอบระบบ
llm_client = HolySheepLLM(HOLYSHEEP_API_KEY)
text_to_sql = CryptoTextToSQL(llm_client, TARDIS_SCHEMA)
ทดสอบคำถาม
test_questions = [
"ราคา Bitcoin วันนี้เท่าไร?",
"ปริมาณการซื้อขาย ETH ใน 24 ชั่วโมงที่ผ่านมา",
"ราคาเฉลี่ย SOL 7 วันที่ผ่านมา"
]
for q in test_questions:
sql = text_to_sql.generate_sql(q)
print(f"\n❓ {q}")
print(f"📝 {sql}\n")
print("✅ Text-to-SQL Engine ready")สร้าง Chat Interface สำหรับ Crypto Analyst
หลังจากมี Text-to-SQL Engine แล้ว ต่อไปคือการสร้าง Chat Interface ที่ใช้งานง่าย รองรับการสนทนาต่อเนื่องและการแปลงผลลัพธ์เป็นภาษาธรรมชาติ
import pandas as pd
from datetime import datetime
class CryptoAnalystAssistant:
"""Chat Assistant สำหรับวิเคราะห์ข้อมูลคริปโต"""
def __init__(self, llm_client: HolySheepLLM, text_to_sql: CryptoTextToSQL):
self.llm = llm_client
self.text_to_sql = text_to_sql
self.conversation_history = []
def ask(self, question: str, execute: bool = True) -> Dict[str, Any]:
"""ถามคำถามเกี่ยวกับข้อมูลคริปโต"""
# เพิ่มคำถามใน history
self.conversation_history.append({
"role": "user",
"content": question
})
# สร้าง SQL
sql = self.text_to_sql.generate_sql(question)
result = {
"question": question,
"generated_sql": sql,
"data": None,
"explanation": None
}
if execute:
# Execute query (ต้องต่อ Tardis API จริง)
# data = self.text_to_sql.execute_query(sql, tardis_client)
# result["data"] = data
# Mock data สำหรับ demo
result["data"] = self._mock_execute(sql)
# สร้างคำอธิบายผลลัพธ์
result["explanation"] = self._explain_result(question, sql, result["data"])
# เพิ่มคำตอบใน history
self.conversation_history.append({
"role": "assistant",
"content": result["explanation"]
})
return result
def _mock_execute(self, sql: str) -> pd.DataFrame:
"""Mock data สำหรับ demo"""
if "BTCUSDT" in sql.upper():
return pd.DataFrame([{
"symbol": "BTCUSDT",
"price": 67245.50,
"timestamp": datetime.now().isoformat()
}])
elif "ETHUSDT" in sql.upper():
return pd.DataFrame([{
"symbol": "ETHUSDT",
"total_volume": 1250000.45,
"timestamp": datetime.now().isoformat()
}])
elif "SOLUSDT" in sql.upper():
return pd.DataFrame([{
"symbol": "SOLUSDT",
"avg_close": 148.23,
"period": "7 days"
}])
return pd.DataFrame()
def _explain_result(self, question: str, sql: str, data: pd.DataFrame) -> str:
"""แปลงผลลัพธ์ SQL เป็นคำอธิบายภาษาธรรมชาติ"""
system_prompt = """คุณเป็นผู้เชี่ยวชาญการวิเคราะห์สกุลเงินดิจิทัล
จงแปลงผลลัพธ์จากการ query เป็นคำอธิบายที่เข้าใจง่ายในภาษาไทย
กฎ:
1. ตอบเป็นภาษาไทยที่เป็นธรรมชาติ
2. ใส่ตัวเลขที่เป็นรูปธรรม
3. เพิ่ม context เล็กน้อย เช่น เปรียบเทียบกับเมื่อวานหรือสัปดาห์ก่อน"""
user_prompt = f"""คำถาม: {question}
SQL ที่ใช้:
{sql}
ผลลัพธ์:
{data.to_string()}
จงอธิบายผลลัพธ์ให้เข้าใจง่าย"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
]
return self.llm.chat_completion(messages, temperature=0.4)
ทดสอบระบบ
assistant = CryptoAnalystAssistant(llm_client, text_to_sql)
ถามคำถาม
print("=" * 60)
print("🪙 Crypto Analyst Assistant - Demo")
print("=" * 60)
result = assistant.ask("ราคา Bitcoin ล่าสุดเท่าไร?")
print(f"\n❓ คำถาม: {result['question']}")
print(f"📝 SQL: {result['generated_sql']}")
print(f"📊 ผลลัพธ์:\n{result['data']}")
print(f"\n💬 คำอธิบาย: {result['explanation']}")
print("\n" + "=" * 60)เหมาะกับใคร / ไม่เหมาะกับใคร
| เหมาะกับใคร ✅ | ไม่เหมาะกับใคร ❌ |
|---|---|
|
|
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
กรณีที่ 1: SQL Syntax Error จาก LLM
ปัญหา: LLM สร้าง SQL ที่มี syntax ผิดพลาด เช่น ใช้ชื่อ column ผิด หรือฟังก์ชันไม่ถูกต้อง
# วิธีแก้ไข: เพิ่ม Validation Layer
def validate_and_fix_sql(self, sql: str, schema: TardisSchema) -> str:
"""ตรวจสอบและแก้ไข SQL ก่อน execute"""
# 1. ดึง list column ที่ถูกต้องจาก schema
valid_columns = set()
valid_tables = set()
for table in schema.tables:
valid_tables.add(table['name'].lower())
for col in table['columns']:
valid_columns.add(col['name'].lower())
# 2. ตรวจสอบว่า tables ที่ใช้มีอยู่จริง
for table_name in valid_tables:
if table_name not in sql.lower():
continue
# ถ้าพบ table ที่ไม่มีใน schema
if not any(t in sql.lower() for t in valid_tables):
raise ValueError(f"Invalid table in SQL: {sql}")
# 3. ถ้า SQL มีปัญหา ให้ลองให้ LLM fix
if self._has_syntax_issues(sql):
fix_prompt = f"""แก้ไข SQL นี้ให้ถูกต้อง:
{sql}
Schema ที่ถูกต้อง:
{schema.to_prompt_context()}
กฎ: ตอบเฉพาะ SQL ที่แก้ไขแล้วใน code block"""
messages = [{"role": "user", "content": fix_prompt}]
fixed_sql = self.llm.chat_completion(messages, temperature=0.1)
return self._extract_sql(fixed_sql)
return sql
def _has_syntax_issues(self, sql: str) -> bool:
"""ตรวจสอบเบื้องต้นว่า SQL อาจมีปัญหา"""
# ตรวจสอบว่ามี keywords พื้นฐานครบ
required = ['SELECT', 'FROM']
return not all(k in sql.upper() for k in required)กรณีที่ 2: Token Limit หมดเมื่อ Query ซับซ้อน
ปัญหา: เมื่อ schema มีขนาดใหญ่มาก หรือ prompt ยาวเกินไปจะทำให้เกิด context overflow
# วิธีแก้ไข: ใช้ Schema Pruning + Caching
from functools import lru_cache
class OptimizedTextToSQL(CryptoTextToSQL):
"""Text-to-SQL ที่ปรับปรุงสำหรับ memory efficiency"""
def __init__(self, llm_client: HolySheepLLM, schema: TardisSchema):
super().__init__(llm_client, schema)
self.schema_cache = {}
@lru_cache(maxsize=100)
def