สวัสดีครับ ผมเป็นนักพัฒนาระบบเทรดอัตโนมัติมากว่า 5 ปี วันนี้จะมาแชร์ประสบการณ์การดึงข้อมูล กราฟเทียน (Candlestick/K-line) จาก Binance API และนำไปทำ การทดสอบย้อนกลับ (Backtesting) อย่างละเอียด ตั้งแต่พื้นฐานจนถึงการประยุกต์ใช้จริงในโปรเจกต์ของผม
บทความนี้เหมาะสำหรับ นักพัฒนา Python, Quant Trader, และ นักวิเคราะห์ข้อมูล ที่ต้องการสร้างระบบเทรดอัตโนมัติด้วยตัวเอง พร้อมแนะนำ เครื่องมือ AI ที่ช่วยประหยัดค่าใช้จ่ายได้ถึง 85%
ทำไมต้องดึงข้อมูลจาก Binance API
Binance เป็นผู้นำตลาดคริปโตที่มี ปริมาณซื้อขายสูงที่สุด และ API ของเขาครบถ้วนทุกฟังก์ชัน รวมถึงข้อมูลประวัติย้อนหลังที่ละเอียดมาก ตั้งแต่ timeframe 1 นาที ไปจนถึงรายเดือน ครอบคลุมทุกคู่เทรด
ข้อดีของ Binance API:
- ฟรี — ไม่มีค่าใช้จ่ายสำหรับข้อมูลประวัติ
- ครอบคลุม — มากกว่า 1,000 คู่เทรด
- เสถียร — API รองรับโหลดสูงได้ดี
- รองรับ WebSocket — สำหรับข้อมูลเรียลไทม์
เตรียมสภาพแวดล้อมการพัฒนา
ก่อนเริ่ม ติดตั้งไลบรารีที่จำเป็น:
# ติดตั้งไลบรารีสำหรับดึงข้อมูลและวิเคราะห์
pip install python-binance pandas numpy matplotlib
ไลบรารีสำหรับ backtesting
pip install backtesting ta-lib
หากติดตั้ง ta-lib มีปัญหา ลองใช้ ta แทน
pip install taดึงข้อมูลกราฟเทียนผ่าน Binance API
นี่คือโค้ดหลักสำหรับดึงข้อมูล K-line จาก Binance ซึ่งผมใช้มาหลายปีแล้ว รันได้เสถียรมาก:
import requests
import pandas as pd
from datetime import datetime
class BinanceDataFetcher:
"""
คลาสสำหรับดึงข้อมูลกราฟเทียนจาก Binance API
รองรับทุก timeframe และคู่เทรด
"""
BASE_URL = "https://api.binance.com"
def __init__(self):
self.session = requests.Session()
def get_klines(self, symbol: str, interval: str,
start_str: str = None, end_str: str = None,
limit: int = 1000) -> pd.DataFrame:
"""
ดึงข้อมูลกราฟเทียน
Parameters:
- symbol: คู่เทรด เช่น 'BTCUSDT', 'ETHBUSD'
- interval: timeframe เช่น '1m', '5m', '1h', '1d'
- start_str: วันเริ่มต้น '2023-01-01'
- end_str: วันสิ้นสุด '2024-01-01'
- limit: จำนวนข้อมูลสูงสุด (1-1000)
Returns:
- DataFrame พร้อมคอลัมน์ OHLCV
"""
endpoint = "/api/v3/klines"
params = {
"symbol": symbol.upper(),
"interval": interval,
"limit": limit
}
if start_str:
# แปลงวันที่เป็น timestamp มิลลิวินาที
start_ts = int(pd.Timestamp(start_str).timestamp() * 1000)
params["startTime"] = start_ts
if end_str:
end_ts = int(pd.Timestamp(end_str).timestamp() * 1000)
params["endTime"] = end_ts
response = self.session.get(
f"{self.BASE_URL}{endpoint}",
params=params,
timeout=30
)
response.raise_for_status()
# แปลงข้อมูลเป็น DataFrame
df = pd.DataFrame(response.json())
# ตั้งชื่อคอลัมน์ตามมาตรฐาน
columns = [
'open_time', 'open', 'high', 'low', 'close', 'volume',
'close_time', 'quote_asset_volume', 'trades',
'taker_buy_base', 'taker_buy_quote', 'ignore'
]
df.columns = columns[:len(df.columns)]
# แปลงประเภทข้อมูล
numeric_cols = ['open', 'high', 'low', 'close', 'volume']
for col in numeric_cols:
df[col] = pd.to_numeric(df[col], errors='coerce')
# แปลง timestamp เป็น datetime
df['open_time'] = pd.to_datetime(df['open_time'], unit='ms')
df['close_time'] = pd.to_datetime(df['close_time'], unit='ms')
return df
ตัวอย่างการใช้งาน
fetcher = BinanceDataFetcher()
ดึงข้อมูล BTCUSDT รายชั่วโมง 1 ปีย้อนหลัง
btc_data = fetcher.get_klines(
symbol='BTCUSDT',
interval='1h',
start_str='2023-01-01',
end_str='2024-01-01',
limit=1000
)
print(f"ดึงข้อมูลสำเร็จ: {len(btc_data)} แท่งเทียน")
print(btc_data.head())ดึงข้อมูลปริมาณมากด้วยการวนลูป
Binance API จำกัดการดึงข้อมูลได้ครั้งละ 1,000 แท่งเทียนเท่านั้น หากต้องการข้อมูลหลายปี ต้องแบ่งดึงเป็นช่วง:
import time
def fetch_all_klines(symbol: str, interval: str,
start_date: str, end_date: str) -> pd.DataFrame:
"""
ดึงข้อมูลทั้งหมดโดยอัตโนมัติ แบ่งเป็นช่วงๆ
"""
fetcher = BinanceDataFetcher()
all_data = []
# แปลงวันที่เป็น timestamp
current_start = pd.Timestamp(start_date)
final_end = pd.Timestamp(end_date)
print(f"กำลังดึงข้อมูล {symbol} {interval}")
print(f"ตั้งแต่ {start_date} ถึง {end_date}")
while current_start < final_end:
# ดึงข้อมูลช่วงละ 1000 แท่ง
df = fetcher.get_klines(
symbol=symbol,
interval=interval,
start_str=str(current_start),
end_str=str(final_end),
limit=1000
)
if df.empty:
break
all_data.append(df)
# ขยับวันเริ่มต้นไปวันถัดไป (สำหรับ timeframe 1d)
# หรือปรับตาม timeframe ที่ใช้
current_start = df['close_time'].max() + pd.Timedelta(minutes=1)
print(f"ดึงได้ {len(df)} แท่ง, รวม {sum(len(d) for d in all_data)} แท่ง")
# หน่วงเวลาเพื่อไม่ให้ถูก rate limit
time.sleep(0.5)
# รวมข้อมูลทั้งหมด
if all_data:
combined_df = pd.concat(all_data, ignore_index=True)
# ลบข้อมูลซ้ำถ้ามี
combined_df = combined_df.drop_duplicates(subset=['open_time'])
combined_df = combined_df.sort_values('open_time').reset_index(drop=True)
return combined_df
return pd.DataFrame()
ตัวอย่าง: ดึงข้อมูล 3 ปี
eth_daily = fetch_all_klines(
symbol='ETHUSDT',
interval='1d',
start_date='2021-01-01',
end_date='2024-01-01'
)
บันทึกเป็นไฟล์ CSV
eth_daily.to_csv('ethusdt_daily.csv', index=False)
print(f"บันทึกสำเร็จ: {len(eth_daily)} แท่งเทียน")เตรียมข้อมูลสำหรับ Backtesting
ข้อมูลดิบจาก API ยังต้องผ่านการประมวลผลก่อนนำไปทดสอบระบบ:
import pandas as pd
import numpy as np
class DataPreprocessor:
"""จัดเตรียมข้อมูลสำหรับการทดสอบย้อนกลับ"""
@staticmethod
def calculate_indicators(df: pd.DataFrame) -> pd.DataFrame:
"""เพิ่ม Technical Indicators ที่จำเป็น"""
df = df.copy()
# Simple Moving Average
df['sma_20'] = df['close'].rolling(window=20).mean()
df['sma_50'] = df['close'].rolling(window=50).mean()
df['sma_200'] = df['close'].rolling(window=200).mean()
# Exponential Moving Average
df['ema_12'] = df['close'].ewm(span=12, adjust=False).mean()
df['ema_26'] = df['close'].ewm(span=26, adjust=False).mean()
# MACD
df['macd'] = df['ema_12'] - df['ema_26']
df['macd_signal'] = df['macd'].ewm(span=9, adjust=False).mean()
df['macd_hist'] = df['macd'] - df['macd_signal']
# RSI
delta = df['close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
df['rsi'] = 100 - (100 / (1 + rs))
# Bollinger Bands
df['bb_middle'] = df['close'].rolling(window=20).mean()
df['bb_std'] = df['close'].rolling(window=20).std()
df['bb_upper'] = df['bb_middle'] + (df['bb_std'] * 2)
df['bb_lower'] = df['bb_middle'] - (df['bb_std'] * 2)
# Volume indicators
df['volume_sma_20'] = df['volume'].rolling(window=20).mean()
df['volume_ratio'] = df['volume'] / df['volume_sma_20']
return df
@staticmethod
def clean_data(df: pd.DataFrame) -> pd.DataFrame:
"""ทำความสะอาดข้อมูล"""
df = df.copy()
# ลบแถวที่มีค่าว่าง
df = df.dropna()
# รีเซ็ต index
df = df.reset_index(drop=True)
# ตรวจสอบ outlier
z_scores = np.abs((df['close'] - df['close'].mean()) / df['close'].std())
df = df[z_scores < 5] # ลบ outlier ที่ห่างจากค่าเฉลี่ยเกิน 5 std
return df
@staticmethod
def split_train_test(df: pd.DataFrame,
train_ratio: float = 0.7) -> tuple:
"""แบ่งข้อมูลสำหรับ train และ test"""
split_idx = int(len(df) * train_ratio)
train = df.iloc[:split_idx].copy()
test = df.iloc[split_idx:].copy()
print(f"Train: {len(train)} แท่ง ({train['open_time'].min()} ถึง {train['open_time'].max()})")
print(f"Test: {len(test)} แท่ง ({test['open_time'].min()} ถึง {test['open_time'].max()})")
return train, test
ใช้งาน
preprocessor = DataPreprocessor()
df_with_indicators = preprocessor.calculate_indicators(eth_daily)
df_clean = preprocessor.clean_data(df_with_indicators)
train_data, test_data = preprocessor.split_train_test(df_clean)สร้างระบบเทรดอัตโนมัติ
หลังจากเตรียมข้อมูลเรียบร้อย มาสร้างกลยุทธ์การซื้อขายกัน:
from dataclasses import dataclass
from typing import Optional
@dataclass
class TradeSignal:
"""สัญญาณซื้อขาย"""
timestamp: pd.Timestamp
action: str # 'BUY' หรือ 'SELL'
price: float
reason: str
confidence: float # 0-1
class TradingStrategy:
"""กลยุทธ์การซื้อขายแบบ Moving Average Crossover + RSI"""
def __init__(self,
fast_ma: int = 20,
slow_ma: int = 50,
rsi_oversold: float = 30,
rsi_overbought: float = 70,
rsi_period: int = 14):
self.fast_ma = fast_ma
self.slow_ma = slow_ma
self.rsi_oversold = rsi_oversold
self.rsi_overbought = rsi_overbought
self.rsi_period = rsi_period
def generate_signals(self, df: pd.DataFrame) -> list:
"""สร้างสัญญาณซื้อขายจากข้อมูล"""
signals = []
position = 0 # 0 = ไม่มีสถานะ, 1 = ถือ long
for i in range(self.slow_ma, len(df)):
row = df.iloc[i]
# เงื่อนไขซื้อ: SMA cross over + RSI oversold
if (row['sma_20'] > row['sma_50'] and
df.iloc[i-1]['sma_20'] <= df.iloc[i-1]['sma_50'] and
row['rsi'] < self.rsi_oversold and
position == 0):
signal = TradeSignal(
timestamp=row['open_time'],
action='BUY',
price=row['close'],
reason=f"SMA Crossover + RSI={row['rsi']:.1f}",
confidence=0.8
)
signals.append(signal)
position = 1
# เงื่อนไขขาย: SMA cross under + RSI overbought
elif (row['sma_20'] < row['sma_50'] and
df.iloc[i-1]['sma_20'] >= df.iloc[i-1]['sma_50'] and
row['rsi'] > self.rsi_overbought and
position == 1):
signal = TradeSignal(
timestamp=row['open_time'],
action='SELL',
price=row['close'],
reason=f"SMA Cross Under + RSI={row['rsi']:.1f}",
confidence=0.8
)
signals.append(signal)
position = 0
return signals
ทดสอบกลยุทธ์
strategy = TradingStrategy(
fast_ma=20,
slow_ma