การวิเคราะห์ข้อมูลคริปโตเคอร์เรนซี (Cryptocurrency) เป็นหนึ่งในทักษะที่มีความต้องการสูงมากในปัจจุบัน ไม่ว่าจะเป็นนักลงทุนที่ต้องการสร้างโมเดลทำนายราคา หรือนักพัฒนาที่ต้องการสร้างระบบเทรดอัตโนมัติ สิ่งที่ขาดไม่ได้คือข้อมูลอนุกรมเวลา (Time Series Data) ที่มีคุณภาพ ในบทความนี้ผมจะพาทุกท่านเรียนรู้การเตรียมข้อมูลอนุกรมเวลาคริปโตตั้งแต่เริ่มต้น โดยไม่จำเป็นต้องมีประสบการณ์การใช้ API มาก่อนเลย
ทำความรู้จักกับข้อมูลอนุกรมเวลาคริปโต
ก่อนจะเริ่มลงมือทำ เรามาทำความเข้าใจกันก่อนว่าข้อมูลอนุกรมเวลาของคริปโตเคอร์เรนซีคืออะไร และทำไมการเตรียมข้อมูลให้ถูกต้องถึงมีความสำคัญมาก
ข้อมูลอนุกรมเวลาคืออะไร
ข้อมูลอนุกรมเวลาคือชุดข้อมูลที่ถูกจัดเก็บตามลำดับเวลา เช่น ราคา Bitcoin ที่ถูกบันทึกทุกๆ 1 นาที, 1 ชั่วโมง หรือ 1 วัน ตัวอย่างข้อมูลที่เราจะได้รับ ได้แก่ ราคาเปิด (Open), ราคาปิด (Close), ราคาสูงสุด (High), ราคาต่ำสุด (Low) และปริมาณการซื้อขาย (Volume)
ทำไมต้องเตรียมข้อมูลให้ดี
จากประสบการณ์ของผมที่เคยพัฒนาโมเดล Machine Learning สำหรับทำนายราคาคริปโตมาหลายปี พบว่า 80% ของเวลาทั้งหมดจะใช้ไปกับการเตรียมข้อมูล ไม่ใช่การสร้างโมเดล ข้อมูลที่ไม่สมบูรณ์หรือไม่ถูกต้องจะทำให้โมเดลให้ผลลัพธ์ที่ผิดพลาดได้ เช่น โมเดลอาจเรียนรู้จากข้อมูลที่ผิดปกติ (Outliers) และให้การทำนายที่คลาดเคลื่อนมาก
เครื่องมือที่ต้องเตรียมพร้อม
สำหรับผู้เริ่มต้น ผมแนะนำให้ติดตั้ง Python 3.8 ขึ้นไป พร้อมกับไลบรารีสำคัญดังนี้
- pandas - สำหรับจัดการข้อมูลตาราง
- numpy - สำหรับคำนวณทางคณิตศาสตร์
- requests - สำหรับเรียกใช้ API
- matplotlib - สำหรับแสดงผลข้อมูลเป็นกราฟ
# คำสั่งติดตั้งเครื่องมือที่จำเป็น
pip install pandas numpy requests matplotlib
หรือถ้าใช้ Anaconda
conda install pandas numpy requests matplotlibการดึงข้อมูลจากแหล่งที่เชื่อถือได้
การได้ข้อมูลที่ถูกต้องเป็นจุดเริ่มต้นที่สำคัญมาก ผมเคยลองใช้หลายแหล่งข้อมูล ตั้งแต่การดึงจากเว็บไซต์โดยตรง ไปจนถึงการใช้ API ของตลาดต่างๆ พบว่าการใช้ API ที่เสถียรและมีความหน่วงต่ำ (Low Latency) จะให้ข้อมูลที่ทันสมัยและถูกต้องกว่า
วิธีเรียกใช้ HolySheep AI API สำหรับข้อมูลคริปโต
ในการเรียกใช้ HolySheep AI ซึ่งเป็นแพลตฟอร์ม AI ที่มีประสิทธิภาพสูงและราคาประหยัด ท่านสามารถสมัครได้ที่ สมัครที่นี่ เพื่อรับเครดิตฟรีเมื่อลงทะเบียน ระบบรองรับการชำระเงินผ่าน WeChat และ Alipay อีกด้วย
import requests
import json
ตั้งค่าการเชื่อมต่อกับ HolySheep AI API
หมายเหตุ: ค่า Latency ของระบบน้อยกว่า 50 มิลลิวินาที ทำให้การดึงข้อมูลรวดเร็วมาก
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # แทนที่ด้วย API Key ของคุณ
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
ตัวอย่างการดึงข้อมูลราคา Bitcoin
ระบบ HolySheep รองรับการประมวลผลข้อมูลอนุกรมเวลาได้อย่างมีประสิทธิภาพ
response = requests.get(
f"{BASE_URL}/crypto/ohlcv",
headers=headers,
params={
"symbol": "BTC/USDT", # คู่เทรด
"interval": "1h", # ช่วงเวลา 1 ชั่วโมง
"limit": 1000 # จำนวนข้อมูล 1000 จุด
}
)
data = response.json()
print(f"ได้รับข้อมูล {len(data)} จุด")
print(data[:5]) # แสดง 5 จุดแรกโครงสร้างข้อมูลที่ได้รับ
เมื่อเรียกใช้ API สำเร็จ เราจะได้ข้อมูลในรูปแบบ JSON ที่มีโครงสร้างดังนี้
# ตัวอย่างโครงสร้างข้อมูล JSON ที่ได้รับ
sample_data = [
{
"timestamp": 1704067200, # Unix Timestamp
"open": 42350.50,
"high": 42500.00,
"low": 42100.25,
"close": 42450.75,
"volume": 12500.5 # ปริมาณการซื้อขาย
},
{
"timestamp": 1704070800,
"open": 42450.75,
"high": 42680.00,
"low": 42320.50,
"close": 42600.00,
"volume": 15800.3
}
]
การแปลงข้อมูล JSON เป็น DataFrame สำหรับการประมวลผล
import pandas as pd
df = pd.DataFrame(sample_data)
df['datetime'] = pd.to_datetime(df['timestamp'], unit='s')
print(df.head())การทำความสะอาดข้อมูล (Data Cleaning)
หลังจากได้ข้อมูลมาแล้ว ขั้นตอนต่อไปคือการทำความสะอาดข้อมูล ซึ่งเป็นขั้นตอนที่สำคัญมากและต้องใช้ความละเอียด
การจัดการค่าที่หายไป (Missing Values)
ค่าที่หายไปเป็นปัญหาที่พบบ่อยมากในข้อมูลอนุกรมเวลา เกิดจากหลายสาเหตุ เช่น ระบบขัดข้องชั่วคราว หรือตลาดปิดทำการ มีหลายวิธีในการจัดการกับค่าที่หายไป
import pandas as pd
import numpy as np
สร้าง DataFrame ตัวอย่างที่มีค่าที่หายไป
df = pd.DataFrame({
'datetime': pd.date_range('2024-01-01', periods=10, freq='h'),
'close': [42350, 42400, np.nan, 42500, 42450, np.nan, 42600, 42550, 42700, 42650]
})
print("ข้อมูลก่อนทำความสะอาด:")
print(df)
วิธีที่ 1: เติมค่าด้วยค่าเฉลี่ย (ใช้เมื่อมีค่าหายไปน้อย)
df['close_mean'] = df['close'].fillna(df['close'].mean())
วิธีที่ 2: เติมค่าด้วยการ interpolate (แนะนำสำหรับข้อมูลอนุกรมเวลา)
df['close_interpolate'] = df['close'].interpolate(method='linear')
วิธีที่ 3: เติมค่าด้วย Forward Fill (ใช้ค่าก่อนหน้า)
df['close_ffill'] = df['close'].ffill()
วิธีที่ 4: เติมค่าด้วย Backward Fill (ใช้ค่าถัดไป)
df['close_bfill'] = df['close'].bfill()
print("\nข้อมูลหลังทำความสะอาด:")
print(df[['datetime', 'close', 'close_interpolate']].head(10))การตรวจจับและจัดการค่าผิดปกติ (Outliers)
ค่าผิดปกติคือค่าที่แตกต่างจากค่าทั่วไปอย่างมาก อาจเกิดจากข้อผิดพลาดในการบันทึกข้อมูล หรือเหตุการณ์พิเศษในตลาด วิธีตรวจจับที่นิยมใช้คือ IQR Method
def detect_outliers_iqr(data, column, threshold=1.5):
"""
ตรวจจับค่าผิดปกติด้วย IQR Method
threshold: ค่าตัวคูณ IQR (ค่าแนะนำ 1.5 ถึง 3)
"""
Q1 = data[column].quantile(0.25)
Q3 = data[column].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - threshold * IQR
upper_bound = Q3 + threshold * IQR
outliers = data[(data[column] < lower_bound) | (data[column] > upper_bound)]
print(f"Q1: {Q1:.2f}, Q3: {Q3:.2f}, IQR: {IQR:.2f}")
print(f"ขอบเขตล่าง: {lower_bound:.2f}, ขอบเขตบน: {upper_bound:.2f}")
print(f"พบค่าผิดปกติ {len(outliers)} จุด")
return outliers, lower_bound, upper_bound
ตัวอย่างการใช้งาน
sample_prices = [42350, 42400, 50000, 42500, 42450, 49000, 42600, 42550, 42700, 42650]
df_test = pd.DataFrame({'close': sample_prices})
outliers, lower, upper = detect_outliers_iqr(df_test, 'close', threshold=1.5)
วิธีจัดการกับ Outliers
วิธีที่ 1: แทนที่ด้วยค่าขอบเขต
df_test['close_clipped'] = df_test['close'].clip(lower=lower, upper=upper)
วิธีที่ 2: แทนที่ด้วยค่าเฉลี่ย (ไม่รวม outliers)
mask = (df_test['close'] >= lower) & (df_test['close'] <= upper)
mean_value = df_test.loc[mask, 'close'].mean()
df_test['close_mean_replaced'] = df_test['close'].apply(
lambda x: mean_value if x < lower or x > upper else x
)
print("\nผลลัพธ์:")
print(df_test)การสร้างฟีเจอร์สำหรับโมเดล Machine Learning
การสร้างฟีเจอร์ (Feature Engineering) เป็นศาสตร์และศิลป์ในการแปลงข้อมูลดิบให้เป็นข้อมูลที่โมเดลสามารถเรียนรู้ได้ดีขึ้น ผมจะแบ่งปันเทคนิคที่ใช้มาจากประสบการณ์ตรง
ฟีเจอร์พื้นฐานที่ควรมี
import pandas as pd
import numpy as np
def create_basic_features(df, price_col='close'):
"""
สร้างฟีเจอร์พื้นฐานสำหรับข้อมูลอนุกรมเวลาคริปโต
"""
df = df.copy()
# 1. ผลตอบแทน (Returns)
df['returns'] = df[price_col].pct_change()
# 2. ผลตอบแทนสะสม (Cumulative Returns)
df['cumulative_returns'] = (1 + df['returns']).cumprod() - 1
# 3. ความผันผวน (Volatility) - ค่าเบี่ยงเบนมาตรฐานของผลตอบแทน
df['volatility_7d'] = df['returns'].rolling(window=7).std()
df['volatility_30d'] = df['returns'].rolling(window=30).std()
# 4. ค่าเฉลี่ยเคลื่อนที่ (Moving Averages)
df['ma_7'] = df[price_col].rolling(window=7).mean()
df['ma_25'] = df[price_col].rolling(window=25).mean()
df['ma_99'] = df[price_col].rolling(window=99).mean()
# 5. อัตราส่วนราคาต่อ MA (Price to MA Ratio)
df['price_to_ma7'] = df[price_col] / df['ma_7']
df['price_to_ma25'] = df[price_col] / df['ma_25']
# 6. RSI (Relative Strength Index)
delta = df[price_col].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))
# 7. Bollinger Bands
df['bb_middle'] = df[price_col].rolling(window=20).mean()
df['bb_std'] = df[price_col].rolling(window=20).std()
df['bb_upper'] = df['bb_middle'] + (df['bb_std'] * 2)
df['bb_lower'] = df['bb_middle'] - (df['bb_std'] * 2)
df['bb_position'] = (df[price_col] - df['bb_lower']) / (df['bb_upper'] - df['bb_lower'])
# 8. MACD
exp1 = df[price_col].ewm(span=12, adjust=False).mean()
exp2 = df[price_col].ewm(span=26, adjust=False).mean()
df['macd'] = exp1 - exp2
df['macd_signal'] = df['macd'].ewm(span=9, adjust=False).mean()
df['macd_histogram'] = df['macd'] - df['macd_signal']
return df
ตัวอย่างการใช้งาน
สมมติว่ามี DataFrame ชื่อ df ที่มีคอลัมน์ 'close'
df_featured = create_basic_features(df)
print(df_featured.head(30))
ฟีเจอร์ขั้นสูงจากประสบการณ์
def create_advanced_features(df, price_col='close', volume_col='volume'):
"""
สร้างฟีเจอร์ขั้นสูงที่ผมใช้ในโปรเจกต์จริง
"""
df = df.copy()
# 1. Momentum Indicators
# ROC (Rate of Change)
df['roc_10'] = ((df[price_col] - df[price_col].shift(10)) / df[price_col].shift(10)) * 100
df['roc_20'] = ((df[price_col] - df[price_col].shift(20)) / df[price_col].shift(20)) * 100
# Stochastic Oscillator
low_14 = df['low'].rolling(window=14).min()
high_14 = df['high'].rolling(window=14).max()
df['stoch_k'] = 100 * ((df[price_col] - low_14) / (high_14 - low_14))
df['stoch_d'] = df['stoch_k'].rolling(window=3).mean()
# 2. Volume Indicators
# OBV (On Balance Volume)
df['obv'] = (np.sign(df[volume_col].diff()) * df[volume_col]).fillna(0).cumsum()
# Volume MA
df['volume_ma'] = df[volume_col].rolling(window=20).mean()
df['volume_ratio'] = df[volume_col] / df['volume_ma']
# 3. Time-based Features
if 'datetime' in df.columns:
df['hour'] = df['datetime'].dt.hour
df['day_of_week'] = df['datetime'].dt.dayofweek
df['day_of_month'] = df['datetime'].dt.day
df['month'] = df['datetime'].dt.month
df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int)
# 4. Lag Features (ค่าในอดีต)
for lag in [1, 2, 3, 5, 7, 14]:
df[f'returns_lag_{lag}'] = df['returns'].shift(lag)
df[f'close_lag_{lag}'] = df[price_col].shift(lag)
# 5. Rolling Statistics
for window in [7, 14, 30]:
df[f'close_skew_{window}'] = df[price_col].rolling(window=window).skew()
df[f'close_kurt_{window}'] = df[price_col].rolling(window=window).kurt()
# 6. Cross-sectional Features
df['high_low_ratio'] = df['high'] / df['low']
df['close_open_diff'] = df['close'] - df['open']
df['close_open_ratio'] = df['close'] / df['open']
return df
หมายเหตุ: ฟังก์ชันนี้ต้องการข้อมูล 'high' และ 'low' ด้วย
df_advanced = create_advanced_features(df)
print(f"จำนวนฟีเจอร์ทั้งหมด: {len(df_advanced.columns)}")
การปรับขนาดข้อมูล (Data Scaling)
การปรับขนาดข้อมูลเป็นขั้นตอนสำคัญที่ช่วยให้โมเดล Machine Learning ทำงานได้ดีขึ้น เนื่องจากฟีเจอร์ต่างๆ มีมาตราส่วนที่แตกต่างกันมาก เช่น ราคา Bitcoin อยู่ที่หลายหมื่น แต่ RSI อยู่ที่ 0-100
from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler
def scale_features(df, feature_cols, method='standard'):
"""
ปรับขนาดฟีเจอร์ด้วยวิธีต่างๆ
Parameters:
- method: 'standard' (Z-score), 'minmax' (0-1), 'robust' (ใช้ median)
"""
df = df.copy()
# เลือก Scaler ตามวิธีที่ต้องการ
if method == 'standard':
scaler = StandardScaler()
elif method == 'minmax':
scaler