ในโลกของการเทรดและการวิเคราะห์ทางเทคนิค การรวมข้อมูลประวัติจากหลายแหล่งเข้าด้วยกันเป็นสิ่งจำเป็นอย่างยิ่ง บทความนี้จะพาคุณเรียนรู้การดึงข้อมูลจาก Tardis (แพลตฟอร์ม API สำหรับข้อมูล crypto exchange) และนำมาคำนวณ technical indicators ด้วย pandas-ta พร้อมแนะนำโซลูชัน AI ที่ช่วยเพิ่มประสิทธิภาพในการวิเคราะห์
สถานการณ์ข้อผิดพลาดจริงที่พบบ่อย
ฉันเคยเจอปัญหาที่ทำให้หยุดงานไปหลายวัน: เมื่อดึงข้อมูล OHLCV จาก Tardis API แล้วพยายามคำนวณ EMA, RSI, MACD แต่ได้รับข้อผิดพลาดแบบนี้:
ValueError: operands could not be broadcast together with shapes (100,) (99,)
IndexError: array index out of range when calculating Bollinger Bandsปัญหาเกิดจากคอลัมน์ว่างเปล่าในข้อมูล หรือข้อมูลไม่เรียงลำดับเวลาอย่างถูกต้อง มาดูวิธีแก้ไขอย่างละเอียด
Tardis API คืออะไร
Tardis เป็น API ที่ให้บริการข้อมูล historical data จาก crypto exchanges หลายตัว เช่น Binance, Bybit, OKX โดยรองรับข้อมูล trades, quotes, OHLCV และอื่นๆ ราคาเริ่มต้นที่ $49/เดือน สำหรับผู้ที่ต้องการทดลองใช้งาน
การติดตั้งและเตรียม Environment
# สร้าง virtual environment และติดตั้ง dependencies
python -m venv trading_env
source trading_env/bin/activate # Windows: trading_env\Scripts\activate
ติดตั้ง packages ที่จำเป็น
pip install pandas pandas-ta requests python-dotenvการดึงข้อมูลจาก Tardis API
import requests
import pandas as pd
from datetime import datetime, timedelta
import time
class TardisClient:
"""Client สำหรับดึงข้อมูล historical จาก Tardis API"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.tardis.dev/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def get_ohlcv(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime,
timeframe: str = "1m"
) -> pd.DataFrame:
"""
ดึงข้อมูล OHLCV จาก exchange ที่ต้องการ
Args:
exchange: ชื่อ exchange เช่น 'binance', 'bybit'
symbol: สัญลักษณ์ เช่น 'BTCUSDT'
start_date: วันที่เริ่มต้น
end_date: วันที่สิ้นสุด
timeframe: Timeframe เช่น '1m', '5m', '1h', '1d'
"""
# แปลง symbol format สำหรับ Tardis
formatted_symbol = f"{exchange}:{symbol}"
url = f"{self.base_url}/historical/{exchange}/{timeframe}"
params = {
"symbols": formatted_symbol,
"from": int(start_date.timestamp()),
"to": int(end_date.timestamp()),
"limit": 1000 # จำนวน records ต่อ request
}
all_data = []
current_start = start_date
while current_start < end_date:
params["from"] = int(current_start.timestamp())
try:
response = requests.get(
url,
headers=self.headers,
params=params,
timeout=30
)
response.raise_for_status()
data = response.json()
if data and len(data) > 0:
records = data[0].get("data", [])
all_data.extend(records)
# ไปยังช่วงเวลาถัดไป
if records:
last_timestamp = records[-1][0]
current_start = datetime.fromtimestamp(last_timestamp / 1000)
else:
break
time.sleep(0.5) # รอเพื่อไม่ให้เกิน rate limit
except requests.exceptions.Timeout:
print(f"Timeout error: ลองดึงข้อมูลใหม่ในอีก 5 วินาที...")
time.sleep(5)
continue
except requests.exceptions.HTTPError as e:
if e.response.status_code == 401:
raise ValueError("API Key ไม่ถูกต้อง กรุณาตรวจสอบ API Key ของคุณ")
elif e.response.status_code == 429:
print("Rate limit exceeded: รอ 60 วินาที...")
time.sleep(60)
continue
else:
raise
# แปลงข้อมูลเป็น DataFrame
df = pd.DataFrame(all_data, columns=["timestamp", "open", "high", "low", "close", "volume"])
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
df = df.set_index("timestamp")
df = df.sort_index() # เรียงลำดับตามเวลา
# ตรวจสอบและลบข้อมูลที่ผิดพลาด
df = self._clean_data(df)
return df
def _clean_data(self, df: pd.DataFrame) -> pd.DataFrame:
"""ทำความสะอาดข้อมูล ลบค่าว่างและ outliers"""
# ลบ rows ที่มีค่า NaN
df = df.dropna()
# ลบ duplicates
df = df[~df.index.duplicated(keep="first")]
# ตรวจสอบค่าผิดปกติ
for col in ["open", "high", "low", "close", "volume"]:
if col in df.columns:
# high ต้องมากกว่า low
mask = df["high"] < df["low"]
if mask.any():
print(f"พบ {mask.sum()} rows ที่ high < low จะถูกลบ")
df = df[~mask]
# ลบ outliers โดยใช้ IQR method
Q1 = df[col].quantile(0.01)
Q99 = df[col].quantile(0.99)
df = df[(df[col] >= Q1) & (df[col] <= Q99)]
return df
ตัวอย่างการใช้งาน
if __name__ == "__main__":
API_KEY = "YOUR_TARDIS_API_KEY"
client = TardisClient(API_KEY)
df = client.get_ohlcv(
exchange="binance",
symbol="BTCUSDT",
start_date=datetime(2024, 1, 1),
end_date=datetime(2024, 1, 31),
timeframe="5m"
)
print(f"ดึงข้อมูลสำเร็จ: {len(df)} rows")
print(df.tail())การคำนวณ Technical Indicators ด้วย pandas-ta
import pandas_ta as ta
import pandas as pd
import numpy as np
class TechnicalAnalyzer:
"""Class สำหรับคำนวณ Technical Indicators"""
def __init__(self, df: pd.DataFrame):
"""
รับ DataFrame ที่มี OHLCV columns
Args:
df: DataFrame ที่มี columns: open, high, low, close, volume
"""
self.df = df.copy()
self._validate_data()
def _validate_data(self):
"""ตรวจสอบความถูกต้องของข้อมูล"""
required_cols = ["open", "high", "low", "close"]
for col in required_cols:
if col not in self.df.columns:
raise ValueError(f"Missing required column: {col}")
# ตรวจสอบว่า index เป็น datetime
if not isinstance(self.df.index, pd.DatetimeIndex):
self.df.index = pd.to_datetime(self.df.index)
# ตรวจสอบข้อมูลว่าง
if self.df[required_cols].isnull().all().any():
raise ValueError("มี columns ที่ว่างเปล่าทั้งหมด")
def add_indicators(self, indicators_config: dict = None) -> pd.DataFrame:
"""
เพิ่ม Technical Indicators ตาม config
Args:
indicators_config: dict ของ indicators ที่ต้องการ
"""
if indicators_config is None:
indicators_config = self._default_config()
# Trend Indicators
if indicators_config.get("ema"):
periods = indicators_config["ema"].get("periods", [20, 50, 200])
for period in periods:
self.df[f"EMA_{period}"] = ta.ema(
self.df["close"],
length=period
)
if indicators_config.get("sma"):
periods = indicators_config["sma"].get("periods", [20, 50, 200])
for period in periods:
self.df[f"SMA_{period}"] = ta.sma(
self.df["close"],
length=period
)
# Momentum Indicators
if indicators_config.get("rsi"):
period = indicators_config["rsi"].get("period", 14)
self.df["RSI_14"] = ta.rsi(
self.df["close"],
length=period
)
if indicators_config.get("macd"):
self.df.ta.macd(
fast=indicators_config["macd"].get("fast", 12),
slow=indicators_config["macd"].get("slow", 26),
signal=indicators_config["macd"].get("signal", 9),
append=True
)
# Volatility Indicators
if indicators_config.get("bbands"):
self.df.ta.bbands(
length=indicators_config["bbands"].get("length", 20),
std=indicators_config["bbands"].get("std", 2),
append=True
)
if indicators_config.get("atr"):
period = indicators_config["atr"].get("period", 14)
self.df["ATR_14"] = ta.atr(
self.df["high"],
self.df["low"],
self.df["close"],
length=period
)
# Volume Indicators
if indicators_config.get("volume_profile"):
self.df["Volume_SMA_20"] = ta.sma(
self.df["volume"],
length=20
)
self.df["Volume_Ratio"] = self.df["volume"] / self.df["Volume_SMA_20"]
# Stochastic
if indicators_config.get("stoch"):
self.df.ta.stoch(
k=indicators_config["stoch"].get("k", 14),
d=indicators_config["stoch"].get("d", 3),
append=True
)
return self.df
def _default_config(self) -> dict:
"""Config เริ่มต้นสำหรับ indicators"""
return {
"ema": {"periods": [20, 50, 200]},
"rsi": {"period": 14},
"macd": {"fast": 12, "slow": 26, "signal": 9},
"bbands": {"length": 20, "std": 2},
"atr": {"period": 14},
"volume_profile": True,
"stoch": {"k": 14, "d": 3}
}
def get_signals(self) -> pd.DataFrame:
"""สร้าง trading signals จาก indicators"""
signals = pd.DataFrame(index=self.df.index)
# RSI Signals
signals["RSI_Signal"] = "hold"
signals.loc[self.df["RSI_14"] < 30, "RSI_Signal"] = "oversold"
signals.loc[self.df["RSI_14"] > 70, "RSI_Signal"] = "overbought"
# MACD Signals
signals["MACD_Signal"] = "hold"
signals.loc[self.df["MACDs_12_26_9"] > self.df["MACD_12_26_9"], "MACD_Signal"] = "bullish"
signals.loc[self.df["MACDs_12_26_9"] < self.df["MACD_12_26_9"], "MACD_Signal"] = "bearish"
# EMA Crossover Signals
signals["EMA_Signal"] = "hold"
signals.loc[
(self.df["EMA_20"] > self.df["EMA_50"]) &
(self.df["EMA_20"].shift(1) <= self.df["EMA_50"].shift(1)),
"EMA_Signal"
] = "golden_cross"
signals.loc[
(self.df["EMA_20"] < self.df["EMA_50"]) &
(self.df["EMA_20"].shift(1) >= self.df["EMA_50"].shift(1)),
"EMA_Signal"
] = "death_cross"
return signals
ตัวอย่างการใช้งาน
if __name__ == "__main__":
# สร้าง sample data
dates = pd.date_range("2024-01-01", periods=500, freq="5min")
np.random.seed(42)
sample_df = pd.DataFrame({
"open": 45000 + np.cumsum(np.random.randn(500) * 50),
"high": 45000 + np.cumsum(np.random.randn(500) * 50) + 100,
"low": 45000 + np.cumsum(np.random.randn(500) * 50) - 100,
"close": 45000 + np.cumsum(np.random.randn(500) * 50),
"volume": np.random.randint(100, 1000, 500)
}, index=dates)
# คำนวณ indicators
analyzer = TechnicalAnalyzer(sample_df)
df_with_indicators = analyzer.add_indicators()
print("Indicators ที่เพิ่มแล้ว:")
print(df_with_indicators.tail(10))
# ดู signals
signals = analyzer.get_signals()
print("\nTrading Signals:")
print(signals[signals["EMA_Signal"] != "hold"].tail())Pipeline สมบูรณ์: Tardis + pandas-ta + AI Analysis
import requests
import pandas as pd
import pandas_ta as ta
from datetime import datetime
import os
from dotenv import load_dotenv
class TradingDataPipeline:
"""
Pipeline สมบูรณ์สำหรับดึงข้อมูล คำนวณ indicators
และวิเคราะห์ด้วย AI
"""
def __init__(self, tardis_key: str, holySheep_key: str):
self.tardis_key = tardis_key
self.holySheep_key = holySheep_key
self.holySheep_base_url = "https://api.holysheep.ai/v1"
def run_analysis(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime,
timeframe: str = "1h"
) -> dict:
"""
Run analysis pipeline ทั้งหมด
"""
# Step 1: ดึงข้อมูลจาก Tardis
print("กำลังดึงข้อมูลจาก Tardis...")
df = self._fetch_tardis_data(exchange, symbol, start_date, end_date, timeframe)
# Step 2: คำนวณ Technical Indicators
print("กำลังคำนวณ Technical Indicators...")
df_analyzed = self._calculate_indicators(df)
# Step 3: วิเคราะห์ด้วย AI
print("กำลังวิเคราะห์ด้วย AI...")
analysis = self._analyze_with_ai(df_analyzed, symbol)
return {
"data": df_analyzed,
"analysis": analysis,
"summary": self._generate_summary(df_analyzed)
}
def _fetch_tardis_data(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime,
timeframe: str
) -> pd.DataFrame:
"""ดึงข้อมูลจาก Tardis API"""
url = f"https://api.tardis.dev/v1/historical/{exchange}/{timeframe}"
headers = {"Authorization": f"Bearer {self.tardis_key}"}
params = {
"symbols": f"{exchange}:{symbol}",
"from": int(start_date.timestamp()),
"to": int(end_date.timestamp()),
"limit": 1000
}
response = requests.get(url, headers=headers, params=params, timeout=30)
response.raise_for_status()
data = response.json()
if not data:
raise ValueError("ไม่พบข้อมูลจาก Tardis API")
records = data[0].get("data", [])
df = pd.DataFrame(records, columns=["timestamp", "open", "high", "low", "close", "volume"])
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
df = df.set_index("timestamp").sort_index()
return df
def _calculate_indicators(self, df: pd.DataFrame) -> pd.DataFrame:
"""คำนวณ Technical Indicators"""
df = df.copy()
# Trend
df["EMA_20"] = ta.ema(df["close"], length=20)
df["EMA_50"] = ta.ema(df["close"], length=50)
df["EMA_200"] = ta.ema(df["close"], length=200)
# Momentum
df["RSI_14"] = ta.rsi(df["close"], length=14)
df.ta.macd(append=True)
# Volatility
df.ta.bbands(append=True)
df["ATR_14"] = ta.atr(df["high"], df["low"], df["close"], length=14)
# Volume
df["Volume_SMA_20"] = ta.sma(df["volume"], length=20)
return df.dropna()
def _analyze_with_ai(self, df: pd.DataFrame, symbol: str) -> str:
"""
วิเคราะห์ข้อมูลด้วย AI โดยใช้ HolySheep AI
HolySheep AI มี latency <50ms และราคาประหยัดกว่า 85%+
เมื่อเทียบกับ OpenAI API
"""
# เตรียมข้อมูลสรุป
recent_data = df.tail(100).copy()
summary_prompt = f"""Analyze the following technical analysis data for {symbol}:
Latest 5 candles:
{recent_data[['open', 'high', 'low', 'close', 'volume']].tail().to_string()}
Current Indicators:
- RSI(14): {recent_data['RSI_14'].iloc[-1]:.2f}
- MACD: {recent_data['MACD_12_26_9'].iloc[-1]:.2f}
- MACD Signal: {recent_data['MACDs_12_26_9'].iloc[-1]:.2f}
- EMA 20: {recent_data['EMA_20'].iloc[-1]:.2f}
- EMA 50: {recent_data['EMA_50'].iloc[-1]:.2f}
- Price vs EMA200: {'Bullish' if recent_data['close'].iloc[-1] > recent_data['EMA_200'].iloc[-1] else 'Bearish'}
- Bollinger Bands Position: {((recent_data['close'].iloc[-1] - recent_data['BBL_20_2.0'].iloc[-1]) / (recent_data['BBU_20_2.0'].iloc[-1] - recent_data['BBL_20_2.0'].iloc[-1]) * 100):.1f}%
Please provide:
1. Trend analysis (bullish/bearish/neutral)
2. Key support and resistance levels
3. Entry signals (if any)
4. Risk assessment
Respond in Thai language."""
payload = {
"model": "gpt-4.1",
"messages": [
{"role": "user", "content": summary_prompt}
],
"temperature": 0.3
}
headers = {
"Authorization": f"Bearer {self.holySheep_key}",
"Content-Type": "application/json"
}
response = requests.post(
f"{self.holySheep_base_url}/chat/completions",
json=payload,
headers=headers,
timeout=10
)
if response.status_code == 200:
result = response.json()
return result["choices"][0]["message"]["content"]
elif response.status_code == 401:
raise ValueError("HolySheep API Key ไม่ถูกต้อง กรุณาตรวจสอบที่ https://www.holysheep.ai")
else:
raise Exception(f"AI Analysis failed: {response.status_code} - {response.text}")
def _generate_summary(self, df: pd.DataFrame) -> dict:
"""สร้าง summary ของข้อมูล"""
return {
"total_records": len(df),
"date_range": f"{df.index.min()} to {df.index.max()}",
"price_range": f"{df['low'].min():.2f} - {df['high'].max():.2f}",
"avg_volume": df['volume'].mean(),
"current_rsi": df['RSI_14'].iloc[-1],
"trend": "Bullish" if df['EMA_20'].iloc[-1] > df['EMA_50'].iloc[-1] else "Bearish"
}
ตัวอย่างการใช้งาน
if __name__ == "__main__":
load_dotenv()
# ดึง API keys จาก environment
TARDIS_API_KEY = os.getenv("TARDIS_API_KEY")
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")
# สร้าง pipeline
pipeline = TradingDataPipeline(TARDIS_API_KEY, HOLYSHEEP_API_KEY)
# Run analysis
result = pipeline.run_analysis(
exchange="binance",
symbol="BTCUSDT",
start_date=datetime(2024, 6, 1),
end_date=datetime(2024, 6, 30),
timeframe="1h"
)
print("=" * 50)
print("ANALYSIS SUMMARY")
print("=" * 50)
for key, value in result["summary"].items():
print(f"{key}: {value}")
print("\n" + "=" * 50)
print("AI ANALYSIS")
print("=" * 50)
print(result["analysis"])ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
| ข้อผิดพลาด | สาเหตุ | วิธีแก้ไข |
|---|---|---|
ValueError: operands could not be broadcast together with shapes (100,) (99,) |
ข้อมูลมี NaN หรือ index ไม่เรียงลำดับ ทำให้ indicator บางตัวคำนวณได้ไม่ครบ | |
401 Unauthorized |
API Key ไม่ถูกต้อง หมดอายุ หรือไม่มีสิทธิ์เข้าถึง | |
IndexError: array index out of range |
ข้อมูลมีขนาดน้อยกว่า period ของ indicator ที่กำหนด | |
429 Rate Limit Exceeded |
ส่ง request เร็วเกินไป เกิน rate limit ของ API | |
ConnectionError: timeout |
เครือข่ายช้า หรือ API server มีปัญหา | |