ในโลกของการลงทุนคริปโตเคอเรนซี ข้อมูลประวัติคือทองคำ ไม่ว่าจะเป็นการทำ Technical Analysis, Backtesting กลยุทธ์ หรือพัฒนา Machine Learning Model การมี Data Warehouse ที่เก็บข้อมูลราคา OHLCV (Open-High-Low-Close-Volume) จากหลายตลาดแลกเปลี่ยนอย่างเป็นระบบ จะช่วยให้คุณวิเคราะห์แนวโน้มตลาดได้แม่นยำและรวดเร็วกว่าการดึงข้อมูลแบบ Real-time เพียงอย่างเดียว

ทำไมต้องใช้ ClickHouse สำหรับ Crypto Data Warehouse

ClickHouse เป็น Column-Oriented Database ที่ออกแบบมาเพื่อการ Query ข้อมูลขนาดใหญ่ได้อย่างรวดเร็ว ด้วยความสามารถในการบีบอัดข้อมูลและประมวลผลแบบ Vectorized ทำให้เหมาะอย่างยิ่งสำหรับการวิเคราะห์ข้อมูลคริปโตที่มีปริมาณมากและต้องการความเร็วในการคำนวณ Technical Indicators หลายตัวพร้อมกัน

สถาปัตยกรรมระบบ Crypto Data Warehouse

ระบบที่เราจะสร้างประกอบด้วย 3 ส่วนหลัก ได้แก่ Exchange API Connector สำหรับดึงข้อมูลจากตลาด, ClickHouse Database สำหรับจัดเก็บและ Query ข้อมูล และ Python Script สำหรับการ ETL (Extract-Transform-Load) ที่ทำงานแบบ Scheduled หรือ Event-Driven

การติดตั้งและตั้งค่า ClickHouse

สำหรับผู้เริ่มต้น แนะนำให้ใช้ ClickHouse Cloud หรือ Docker Compose เพื่อความสะดวกในการตั้งค่า ตารางหลักสำหรับเก็บข้อมูล OHLCV จะใช้ MergeTree Engine ซึ่งเหมาะสำหรับการจัดเก็บข้อมูลอนุกรมเวลา

-- สร้าง Database สำหรับเก็บข้อมูลคริปโต
CREATE DATABASE IF NOT EXISTS crypto_warehouse ON CLUSTER 'default';

-- สร้างตาราง OHLCV หลัก
CREATE TABLE IF NOT EXISTS crypto_warehouse.ohlcv_1m
(
    symbol String,
    exchange String,
    timeframe String,
    open_time DateTime64(3),
    open Decimal(18, 8),
    high Decimal(18, 8),
    low Decimal(18, 8),
    close Decimal(18, 8),
    volume Decimal(18, 8),
    quote_volume Decimal(18, 8),
    trades UInt32,
    inserted_at DateTime DEFAULT now()
)
ENGINE = MergeTree()
PARTITION BY (exchange, symbol)
ORDER BY (symbol, exchange, timeframe, open_time)
TTL inserted_at + INTERVAL 2 YEAR;

Connector สำหรับดึงข้อมูลจาก Exchange API

เราจะใช้ Python กับไลบรารี ccxt ที่รองรับ Exchange หลายสิบราย ไม่ว่าจะเป็น Binance, Coinbase, Kraken หรือ Bybit การออกแบบให้รองรับหลาย Exchange จะช่วยให้เราเปรียบเทียบข้อมูลและหา Arbitrage Opportunity ได้

# crypto_data_collector.py
import ccxt
import pandas as pd
from clickhouse_driver import Client
from datetime import datetime, timedelta
import schedule
import time

class CryptoDataCollector:
    def __init__(self, clickhouse_host='localhost', clickhouse_port=9000):
        self.clickhouse = Client(
            host=clickhouse_host,
            port=clickhouse_port,
            database='crypto_warehouse'
        )
        # รองรับ Exchange ยอดนิยม
        self.exchanges = {
            'binance': ccxt.binance(),
            'coinbase': ccxt.coinbase(),
            'kraken': ccxt.kraken(),
            'bybit': ccxt.bybit()
        }
        # คู่เทรดหลักที่ต้องการเก็บข้อมูล
        self.symbols = ['BTC/USDT', 'ETH/USDT', 'BNB/USDT', 'SOL/USDT']
        self.timeframes = ['1m', '5m', '1h', '1d']
    
    def fetch_ohlcv(self, exchange_id, symbol, timeframe, since=None, limit=1000):
        """ดึงข้อมูล OHLCV จาก Exchange API"""
        exchange = self.exchanges.get(exchange_id)
        if not exchange:
            raise ValueError(f"Exchange {exchange_id} ไม่รองรับ")
        
        try:
            ohlcv = exchange.fetch_ohlcv(
                symbol=symbol,
                timeframe=timeframe,
                since=since,
                limit=limit
            )
            return ohlcv
        except ccxt.NetworkError as e:
            print(f"Network Error สำหรับ {exchange_id}: {e}")
            return None
        except ccxt.ExchangeError as e:
            print(f"Exchange Error สำหรับ {exchange_id}: {e}")
            return None
    
    def transform_to_dataframe(self, ohlcv_data, symbol, exchange, timeframe):
        """แปลงข้อมูล OHLCV เป็น DataFrame สำหรับ Insert"""
        if not ohlcv_data:
            return None
        
        df = pd.DataFrame(
            ohlcv_data, 
            columns=['open_time', 'open', 'high', 'low', 'close', 'volume']
        )
        df['symbol'] = symbol.replace('/', '')
        df['exchange'] = exchange
        df['timeframe'] = timeframe
        df['quote_volume'] = df['close'] * df['volume']
        df['trades'] = 0  # ต้องใช้ aggregate function ดึงจาก Exchange
        
        return df
    
    def insert_to_clickhouse(self, df):
        """Insert ข้อมูลลง ClickHouse"""
        if df is None or df.empty:
            return 0
        
        columns = [
            'symbol', 'exchange', 'timeframe', 'open_time',
            'open', 'high', 'low', 'close', 'volume', 
            'quote_volume', 'trades'
        ]
        
        # ใช้ Insert Query พร้อม Ignore Duplicates
        query = f"""
        INSERT INTO crypto_warehouse.ohlcv_1m 
        ({', '.join(columns)})
        VALUES
        """
        
        values = df[columns].values.tolist()
        self.clickhouse.execute(query, values)
        return len(values)
    
    def collect_all_data(self):
        """ฟังก์ชันหลักสำหรับรวบรวมข้อมูลจากทุก Exchange"""
        total_inserted = 0
        
        for exchange_id in self.exchanges.keys():
            for symbol in self.symbols:
                for timeframe in self.timeframes:
                    # ดึงข้อมูล 1000 candles ล่าสุด
                    ohlcv = self.fetch_ohlcv(
                        exchange_id, symbol, timeframe, limit=1000
                    )
                    
                    if ohlcv:
                        df = self.transform_to_dataframe(
                            ohlcv, symbol, exchange_id, timeframe
                        )
                        inserted = self.insert_to_clickhouse(df)
                        total_inserted += inserted
                        print(f"✓ {exchange_id}/{symbol}/{timeframe}: {inserted} records")
        
        print(f"📊 รวม: {total_inserted} records ถูก Insert")
        return total_inserted


รันการเก็บข้อมูลทุก 5 นาที

if __name__ == '__main__':
    collector = CryptoDataCollector()
    
    # รันทันทีเมื่อเริ่มสคริปต์
    collector.collect_all_data()
    
    # ตั้งเวลาให้รันทุก 5 นาที
    schedule.every(5).minutes.do(collector.collect_all_data)
    
    while True:
        schedule.run_pending()
        time.sleep(1)

การสร้าง Materialized View สำหรับ Technical Indicators

หนึ่งในความสามารถเด่นของ ClickHouse คือ Materialized View ที่ช่วยให้เราคำนวณ Technical Indicators ได้ล่วงหน้าและเก็บไว้ในตารางแยก ทำให้ Query ด้วย Grafana หรือ BI Tools รวดเร็วมาก

-- สร้าง Materialized View สำหรับ RSI, MACD, Bollinger Bands
CREATE MATERIALIZED VIEW crypto_warehouse.tech_indicators_1h
ENGINE = SummingMergeTree()
ORDER BY (symbol, exchange, open_time)
AS
SELECT
    symbol,
    exchange,
    open_time,
    -- SMA (Simple Moving Average)
    avgIf(close, rowNumberInBlock() <= 20) OVER (
        PARTITION BY symbol, exchange 
        ORDER BY open_time ROWS BETWEEN 19 PRECEDING AND CURRENT ROW
    ) as sma_20,
    avgIf(close, rowNumberInBlock() <= 50) OVER (
        PARTITION BY symbol, exchange 
        ORDER BY open_time ROWS BETWEEN 49 PRECEDING AND CURRENT ROW
    ) as sma_50,
    -- คำนวณ RSI โดยใช้ Window Functions
    -- (แนะนำใช้ ClickHouse Dictionary หรือ Python คำนวณแล้ว Insert แยก)
FROM crypto_warehouse.ohlcv_1m
WHERE timeframe = '1h';

-- สร้างตารางสำหรับเก็บ VWAP (Volume Weighted Average Price)
CREATE TABLE crypto_warehouse.vwap_1h
(
    symbol String,
    exchange String,
    open_time DateTime64(3),
    vwap Decimal(18, 8),
    PRIMARY KEY (symbol, exchange, open_time)
)
ENGINE = ReplacingMergeTree()
ORDER BY (symbol, exchange, open_time);

-- Insert VWAP โดยคำนวณจาก Python (เนื่องจาก ClickHouse ไม่มีฟังก์ชัน VWAP ในตัว)
-- สคริปต์ Python สำหรับคำนวณ VWAP
vwap_script = """
import pandas as pd
from clickhouse_driver import Client

client = Client('localhost', database='crypto_warehouse')


ดึงข้อมูล OHLCV มาคำนวณ VWAP

query = '''
SELECT symbol, exchange, open_time, open, high, low, close, volume, quote_volume
FROM crypto_warehouse.ohlcv_1h
WHERE open_time >= now() - INTERVAL 24 HOUR
ORDER BY symbol, exchange, open_time
'''
results = client.execute(query)
df = pd.DataFrame(results, columns=[
    'symbol', 'exchange', 'open_time', 'open', 'high', 'low', 'close', 
    'volume', 'quote_volume'
])


คำนวณ VWAP = Cumulative(Typical Price * Volume) / Cumulative(Volume)

df['typical_price'] = (df['high'] + df['low'] + df['close']) / 3
df['pv'] = df['typical_price'] * df['volume']
df['cumulative_pv'] = df.groupby(['symbol', 'exchange'])['pv'].cumsum()
df['cumulative_volume'] = df.groupby(['symbol', 'exchange'])['volume'].cumsum()
df['vwap'] = df['cumulative_pv'] / df['cumulative_volume']


Insert กลับเข้า ClickHouse

client.execute(
    'INSERT INTO crypto_warehouse.vwap_1h VALUES',
    df[['symbol', 'exchange', 'open_time', 'vwap']].values.tolist()
)
print(f"✓ VWAP ถูกคำนวณและบันทึก {len(df)} records")
"""

การใช้ AI วิเคราะห์ข้อมูลคริปโต (พร้อม HolySheep AI)

เมื่อมี Data Warehouse พร้อมแล้ว ขั้นตอนถัดไปคือการนำข้อมูลไปวิเคราะห์ด้วย AI เพื่อหา Pattern, ทำ Sentiment Analysis จากข่าว หรือสร้าง Trading Signals ผมแนะนำให้ใช้ HolySheep AI เพราะมีความหน่วงต่ำกว่า 50ms และรองรับโมเดลหลากหลาย รวมถึง DeepSeek V3.2 ที่ราคาถูกมากเพียง $0.42/MTok

# crypto_analysis_with_holysheep.py
import requests
import pandas as pd
from datetime import datetime, timedelta
from clickhouse_driver import Client
import numpy as np

class CryptoAIAnalyzer:
    def __init__(self, holysheep_api_key):
        self.base_url = "https://api.holysheep.ai/v1"
        self.headers = {
            "Authorization": f"Bearer {holysheep_api_key}",
            "Content-Type": "application/json"
        }
        self.clickhouse = Client('localhost', database='crypto_warehouse')
    
    def get_historical_data(self, symbol, days=30):
        """ดึงข้อมูลประวัติจาก ClickHouse"""
        query = f"""
        SELECT 
            open_time,
            open, high, low, close, volume,
            quote_volume
        FROM crypto_warehouse.ohlcv_1d
        WHERE symbol = '{symbol}'
            AND open_time >= now() - INTERVAL {days} DAY
        ORDER BY open_time
        """
        results = self.clickhouse.execute(query)
        df = pd.DataFrame(results, columns=[
            'open_time', 'open', 'high', 'low', 'close', 'volume', 'quote_volume'
        ])
        return df
    
    def calculate_indicators(self, df):
        """คำนวณ Technical Indicators พื้นฐาน"""
        # 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))
        
        # MACD
        exp1 = df['close'].ewm(span=12, adjust=False).mean()
        exp2 = df['close'].ewm(span=26, adjust=False).mean()
        df['macd'] = exp1 - exp2
        df['signal'] = df['macd'].ewm(span=9, adjust=False).mean()
        
        # 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)
        
        return df
    
    def analyze_with_ai(self, symbol, df):
        """วิเคราะห์ข้อมูลด้วย AI ผ่าน HolySheep"""
        # สร้าง Summary ของข้อมูล
        latest = df.iloc[-1]
        summary = f"""
ข้อมูล {symbol} ล่าสุด:
- ราคาปิด: ${latest['close']:,.2f}
- RSI: {latest['rsi']:.2f}
- MACD: {latest['macd']:.4f}
- Signal: {latest['signal']:.4f}
- Bollinger Upper: ${latest['bb_upper']:,.2f}
- Bollinger Lower: ${latest['bb_lower']:,.2f}

กรุณาวิเคราะห์:
1. Trend ปัจจุบัน (Bullish/Bearish/Sideways)
2. แนะนำการเทรด (Buy/Sell/Hold)
3. Stop Loss และ Take Profit
4. Risk/Reward Ratio
"""
        
        # เรียก HolySheep AI
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=self.headers,
            json={
                "model": "deepseek-v3.2",  # โมเดลราคาถูก คุ้มค่า
                "messages": [
                    {"role": "system", "content": "คุณเป็นผู้เชี่ยวชาญด้านการวิเคราะห์ทางเทคนิคคริปโต"},
                    {"role": "user", "content": summary}
                ],
                "temperature": 0.3,
                "max_tokens": 500
            }
        )
        
        if response.status_code == 200:
            return response.json()['choices'][0]['message']['content']
        else:
            return f"Error: {response.status_code} - {response.text}"
    
    def generate_report(self, symbols=['BTCUSDT', 'ETHUSDT']):
        """สร้างรายงานวิเคราะห์สำหรับหลายเหรียญ"""
        reports = []
        
        for symbol in symbols:
            print(f"📊 กำลังวิเคราะห์ {symbol}...")
            df = self.get_historical_data(symbol, days=30)
            
            if len(df) < 20:
                print(f"⚠ ไม่มีข้อมูลเพียงพอสำหรับ {symbol}")
                continue
            
            df = self.calculate_indicators(df)
            analysis = self.analyze_with_ai(symbol, df)
            
            reports.append({
                'symbol': symbol,
                'analysis': analysis,
                'latest_price': df.iloc[-1]['close'],
                'latest_rsi': df.iloc[-1]['rsi']
            })
        
        return reports


ใช้งาน

if __name__ == '__main__':
    analyzer = CryptoAIAnalyzer(
        holysheep_api_key="YOUR_HOLYSHEEP_API_KEY"  # เปลี่ยนเป็น API Key ของคุณ
    )
    reports = analyzer.generate_report(['BTCUSDT', 'ETHUSDT', 'BNBUSDT'])
    
    for report in reports:
        print(f"\n{'='*50}")
        print(f"📈 {report['symbol']} | ราคา: ${report['latest_price']:,.2f} | RSI: {report['latest_rsi']:.2f}")
        print(f"{'='*50}")
        print(report['analysis'])

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. Error 429: Too Many Requests จาก Exchange API

สาเหตุ: Exchange API มี Rate Limit จำกัด โดยเฉพาะ Binance ที่จำกัด 1200 requests/minute สำหรับ Weighted Request

วิธีแก้ไข: ใช้ Rate Limiter และ Cache ข้อมูลที่ดึงมาแล้ว

# เพิ่ม Rate Limiter ให้กับ Data Collector
import time
from ratelimit import limits, sleep_and_retry

@sleep_and_retry
@limits(calls=10, period=60)  # สูงสุด 10 calls ต่อ 60 วินาที
def fetch_with_rate_limit(exchange, symbol, timeframe, limit=1000):
    """ดึงข้อมูลพร้อม Rate Limiting"""
    while True:
        try:
            return exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
        except ccxt.RateLimitExceeded:
            print("⏳ Rate Limited, รอ 60 วินาที...")
            time.sleep(60)
        except Exception as e:
            print(f"Error: {e}")
            return None

2. Missing Data / Gap ใน Time Series

สาเหตุ: Network Error, API Maintenance หรือ Exchange ปิดให้บริการชั่วคราว ทำให้ข้อมูลบางช่วงหายไป

วิธีแก้ไข: สร้าง Process สำหรับ Fill Gap โดยดึงข้อมูลย้อนหลังเฉพาะช่วงที่ขาด

# Fill Gap Script
def find_and_fill_gaps(clickhouse_client, symbol, exchange, timeframe, max_gap_minutes=10):
    """ค้นหาและเติมช่องว่างในข้อมูล"""
    
    # หาช่วงที่ขาดหายไป
    gap_query = f"""
    WITH gaps AS (
        SELECT 
            open_time as gap_start,
            (SELECT open_time FROM crypto_warehouse.ohlcv_1m 
             WHERE open_time > t.open_time 
             ORDER BY open_time LIMIT 1) as gap_end
        FROM crypto_warehouse.ohlcv_1m t
        WHERE symbol = '{symbol}'
            AND exchange = '{exchange}'
            AND timeframe = '{timeframe}'
    )
    SELECT gap_start, gap_end
    FROM gaps
    WHERE (gap_end - gap_start) > 600000  -- เกิน 10 นาที (milliseconds)
    """
    
    gaps = clickhouse_client.execute(gap_query)
    
    for gap_start, gap_end in gaps:
        print(f"🔧 กำลังเติมช่องว่าง: {gap_start} ถึง {gap_end}")
        # ดึงข้อมูลย้อนหลังสำหรับช่วงที่ขาด
        since_ms = int(gap_start.timestamp() * 1000)
        until_ms = int(gap_end.timestamp() * 1000)
        
        # ปรับ limit ตามช่วงเวลาที่ขาด
        duration_hours = (until_ms - since_ms) / (1000 * 60 * 60)
        limit = int(duration_hours * 60 * 60 / 60)  # 1 candle ต่อ 1 นาที
        
        ohlcv = exchange.fetch_ohlcv(
            symbol=symbol.replace('USDT', '/USDT'),
            timeframe='1m',
            since=since_ms,
            limit=min(limit, 1000)
        )
        # Insert ข้อมูลที่ดึงมา
        insert_gap_data(ohlcv, symbol, exchange)

3. Timezone ไม่ตรงกันระหว่าง Exchange และ ClickHouse

สาเหตุ: แต่ละ Exchange ใช้ Timezone ไม่เหมือนกัน เช่น Binance ใช้ UTC แต่บาง Exchange อาจใช้ Local Time

วิธีแก้ไข: กำหนด Timezone ให้เป็น UTC ทั้งหมดตั้งแต่ต้นทาง

# แก้ไข Timezone ก่อน Insert
from datetime import timezone

def normalize_timezone(dt, target_tz='UTC'):
    """แปลง datetime เป็น UTC ก่อน Insert"""
    if dt.tzinfo is None:
        # ถ้าไม่มี timezone info ถือว่าเป็น UTC
        return dt.replace(tzinfo=timezone.utc)
    else:
        # แปลงเป็น UTC
        return dt.astimezone(timezone.utc)

def insert_ohlcv_normalized(self, ohlcv_data, symbol, exchange, timeframe):
    """Insert ข้อมูลพร้อม Normalize Timezone"""
    df = self.transform_to_dataframe(ohlcv_data, symbol, exchange, timeframe)
    
    if df is not None:
        # แปลง open_time เป็น UTC
        df['open_time'] = pd.to_datetime(df['open_time'], unit='ms', utc=True)
        df['open_time'] = df['open_time'].dt.tz_convert(None)  # Remove timezone เก็บเป็น UTC
        
        self.insert_to_clickhouse(df)

เหมาะกับใคร / ไม่เหมาะกับใคร

กลุ่มผู้ใช้ความเหมาะสมเหตุผล
นักเทรดมืออาชีพที่ต้องการ Backtest กลยุทธ์✓ เหมาะมากมีข้อมูลครบถ้วน รวดเร็วในการ Query หลายปีย้อนหลัง
นักพัฒนา Trading Bot✓ เหมาะมากดึงข้อมูล Real-time และ Historical ได้พร้อมกัน
นักวิเคราะห์ทางเทคนิคที่ใช้หลาย Indicators✓ เหมาะมากMaterialized View ช่วยคำนวณล่วงหน้า
ผู้เริ่มต้นเทรดคริปโต⚠ พอใช้ได้มีความซับซ้อนสูง ต้องมีพื้นฐาน Database และ Programming
ผู้ที่ต้องการแค่ดูราคาแบบ Real-time✗ ไม่เหมาะOverkill ใช้ TradingView หรือ Exchange Dashboard ง่ายกว่า
องค์กรที่ต้องการ Compliance และ Audit Trail✓ เหมาะมากClickHouse มี Write-Ahead Log และ Replication ที่แข็งแกร่ง

ราคาและ ROI

การสร้าง Crypto Data Warehouse ด้วยวิธีนี้มีค่าใช้จ่ายหลัก 2 ส่วน ได้แก่ ค่าโค

แหล่งข้อมูลที่เกี่ยวข้อง

บทความที่เกี่ยวข้อง