Python pandas + Tardis การสร้างระบบรวมข้อมูล K-Line สำหรับสกุลเงินดิจิทัล

ในยุคที่ข้อมูลคือทองคำ การวิเคราะห์กราฟราคาสกุลเงินดิจิทัลอย่างแม่นยำต้องอาศัยระบบที่รวบรวมข้อมูล K-Line จากหลายแหล่งได้อย่างมีประสิทธิภาพ บทความนี้จะพาคุณสร้าง K-Line Aggregation Engine ด้วย Python pandas และ Tardis API ตั้งแต่เริ่มต้นจนนำไปประยุกต์ใช้งานจริงในองค์กร

Tardis API คืออะไร และทำไมต้องใช้กับ Python pandas

Tardis API เป็นบริการที่ให้ข้อมูลราคาสกุลเงินดิจิทัลแบบเรียลไทม์และย้อนหลัง รองรับการเชื่อมต่อผ่าน WebSocket และ REST API ความแตกต่างจากแหล่งข้อมูลอื่นคือ Tardis ให้ข้อมูลระดับ Tick-by-Tick ที่สามารถนำมาสร้าง K-Line ได้หลาย timeframe พร้อมกัน การใช้ร่วมกับ Python pandas ช่วยให้การประมวลผล DataFrame ขนาดใหญ่ทำได้รวดเร็วด้วยฟังก์ชัน vectorized operations

การติดตั้งสภาพแวดล้อมและขั้นตอนเตรียมความพร้อม

ก่อนเริ่มสร้างระบบ คุณต้องเตรียมสภาพแวดล้อมการพัฒนาให้พร้อม ขั้นตอนต่อไปนี้จะช่วยให้คุณตั้งค่าได้อย่างถูกต้อง

# สร้าง virtual environment แยกต่อ project
python -m venv kline-engine
source kline-engine/bin/activate  # Linux/Mac
kline-engine\Scripts\activate  # Windows

ติดตั้ง dependencies ที่จำเป็น
pip install pandas>=2.0.0 numpy>=1.24.0
pip install tardis-client>=1.0.0 websocket-client>=1.6.0
pip install asyncio-http>=0.1.0 aiofiles>=23.0.0
pip install holy sheep>=1.2.0  # สำหรับเชื่อมต่อ AI API
pip install python-dotenv>=1.0.0

สร้างไฟล์ .env สำหรับเก็บ API keys
cat > .env << 'EOF'
TARDIS_API_KEY=your_tardis_api_key_here
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
EOF

สร้าง Client สำหรับเชื่อมต่อ Tardis API และ HolySheep AI

ต่อไปจะสร้าง Client class ที่จัดการการเชื่อมต่อทั้ง Tardis API สำหรับรับข้อมูลราคา และ HolySheep AI สำหรับวิเคราะห์ข้อมูลด้วย AI โดยการใช้ HolySheep ช่วยให้คุณประมวลผล K-Line patterns ได้เร็วกว่าการใช้ OpenAI ถึง 85% ด้วยอัตรา ¥1=$1

import os
import json
import asyncio
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dotenv import load_dotenv
import aiohttp

load_dotenv()

class HolySheepAIClient:
    """Client สำหรับเชื่อมต่อ HolySheep AI API - ราคาถูกกว่า OpenAI 85%+"""
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.getenv("HOLYSHEEP_API_KEY")
        if not self.api_key:
            raise ValueError("ต้องระบุ HOLYSHEEP_API_KEY")
    
    async def analyze_kline_pattern(
        self, 
        kline_data: List[Dict],
        symbol: str,
        timeframe: str
    ) -> Dict:
        """วิเคราะห์ K-Line pattern ด้วย AI"""
        
        # สร้าง prompt สำหรับวิเคราะห์
        prompt = f"""วิเคราะห์ K-Line data สำหรับ {symbol} timeframe {timeframe}:
        
ข้อมูลล่าสุด:
{json.dumps(kline_data[-20:], indent=2)}       
        
ให้ระบุ:
1. แนวโน้ม (trend) - ขาขึ้น/ขาลง/sideways
2. RSI โดยประมาณ
3. สัญญาณซื้อ/ขาย
4. แนวรับ/แนวต้านสำคัญ"""
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.BASE_URL}/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "gpt-4.1",
                    "messages": [{"role": "user", "content": prompt}],
                    "temperature": 0.3,
                    "max_tokens": 500
                },
                timeout=aiohttp.ClientTimeout(total=10)
            ) as response:
                if response.status != 200:
                    error_text = await response.text()
                    raise Exception(f"HolySheep API Error {response.status}: {error_text}")
                
                result = await response.json()
                return {
                    "symbol": symbol,
                    "timeframe": timeframe,
                    "analysis": result["choices"][0]["message"]["content"],
                    "usage": result.get("usage", {}),
                    "latency_ms": response.headers.get("X-Response-Time", "N/A")
                }
    
    def calculate_ohlcv_indicators(self, df: pd.DataFrame) -> pd.DataFrame:
        """คำนวณ Technical Indicators จาก OHLCV DataFrame"""
        
        # 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))
        
        # Moving Averages
        df['sma_20'] = df['close'].rolling(window=20).mean()
        df['sma_50'] = df['close'].rolling(window=50).mean()
        df['ema_12'] = df['close'].ewm(span=12, adjust=False).mean()
        
        # 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


class TardisKLineAggregator:
    """ระบบรวมข้อมูล K-Line จาก Tardis API"""
    
    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.getenv("TARDIS_API_KEY")
        self.base_url = "https://api.tardis.dev/v1"
        self.exchanges = ["binance", "bybit", "okx", "deribit"]
    
    async def fetch_historical_klines(
        self,
        exchange: str,
        symbol: str,
        start_time: datetime,
        end_time: datetime,
        timeframe: str = "1m"
    ) -> pd.DataFrame:
        """ดึงข้อมูล K-Line ย้อนหลังจาก Tardis"""
        
        params = {
            "exchange": exchange,
            "symbol": symbol,
            "start_time": int(start_time.timestamp() * 1000),
            "end_time": int(end_time.timestamp() * 1000),
            "timeframe": timeframe,
            "limit": 1000
        }
        
        all_data = []
        async with aiohttp.ClientSession() as session:
            while True:
                async with session.get(
                    f"{self.base_url}/klines",
                    params=params,
                    headers={"Authorization": f"Bearer {self.api_key}"}
                ) as response:
                    if response.status != 200:
                        raise Exception(f"Tardis API Error: {await response.text()}")
                    
                    data = await response.json()
                    if not data:
                        break
                    
                    all_data.extend(data)
                    
                    # เตรียมข้อมูลสำหรับ fetch ครั้งถัดไป
                    last_timestamp = data[-1]["timestamp"]
                    if last_timestamp >= params["end_time"]:
                        break
                    params["start_time"] = last_timestamp + 1
                    
                    await asyncio.sleep(0.5)  # Rate limiting
        
        # แปลงเป็น DataFrame และ pivot เพื่อรวมข้อมูลจากทุก exchange
        df = pd.DataFrame(all_data)
        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
        
        return df
    
    def aggregate_klines_multi_timeframe(
        self,
        df: pd.DataFrame,
        timeframes: List[str] = ["1m", "5m", "15m", "1h", "4h", "1d"]
    ) -> Dict[str, pd.DataFrame]:
        """รวม K-Line หลาย timeframe พร้อมกัน"""
        
        result = {}
        
        for tf in timeframes:
            # Map timeframe string เป็น offset
            tf_map = {
                "1m": "1T", "5m": "5T", "15m": "15T",
                "1h": "1H", "4h": "4H", "1d": "1D"
            }
            
            aggregated = df.groupby(pd.Grouper(
                key='timestamp', 
                freq=tf_map.get(tf, "1T")
            )).agg({
                'open': 'first',
                'high': 'max',
                'low': 'min',
                'close': 'last',
                'volume': 'sum'
            }).reset_index()
            
            result[tf] = aggregated
            
        return result


ตัวอย่างการใช้งาน
async def main():
    # เตรียม clients
    holy_sheep = HolySheepAIClient()
    tardis = TardisKLineAggregator()
    
    # ดึงข้อมูล 7 วันย้อนหลัง
    end_time = datetime.now()
    start_time = end_time - timedelta(days=7)
    
    # ดึงข้อมูลจาก Binance
    df = await tardis.fetch_historical_klines(
        exchange="binance",
        symbol="BTC/USDT",
        start_time=start_time,
        end_time=end_time,
        timeframe="1m"
    )
    
    # คำนวณ indicators
    df = holy_sheep.calculate_ohlcv_indicators(df)
    
    # รวมหลาย timeframe
    multi_tf = tardis.aggregate_klines_multi_timeframe(df)
    
    # วิเคราะห์ด้วย AI
    analysis = await holy_sheep.analyze_kline_pattern(
        kline_data=df.tail(20).to_dict('records'),
        symbol="BTC/USDT",
        timeframe="1h"
    )
    
    print(f"Analysis: {analysis['analysis']}")
    print(f"Latency: {analysis['latency_ms']}ms")
    print(f"Cost: ${analysis['usage'].get('total_tokens', 0) * 0.000008:.4f}")

if __name__ == "__main__":
    asyncio.run(main())

โครงสร้างระบบ K-Line Aggregation Engine แบบ Production-Ready

ระบบที่พร้อมใช้งานจริงต้องมีความสามารถในการจัดการข้อผิดพลาด การ retry connection และการเก็บ cache ข้อมูล ส่วนต่อไปนี้จะแสดงโครงสร้างที่สมบูรณ์พร้อม Backtesting Engine ที่ใช้งานได้จริง

import redis
import pickle
from typing import Optional
from dataclasses import dataclass, asdict
from enum import Enum

class SignalType(Enum):
    BUY = "BUY"
    SELL = "SELL"
    HOLD = "HOLD"
    STRONG_BUY = "STRONG_BUY"
    STRONG_SELL = "STRONG_SELL"

@dataclass
class KLineSignal:
    timestamp: datetime
    symbol: str
    timeframe: str
    signal: SignalType
    confidence: float
    entry_price: float
    stop_loss: float
    take_profit: float
    ai_analysis: Optional[str] = None

class KLineEngine:
    """K-Line Aggregation Engine แบบ Production-Ready"""
    
    def __init__(
        self,
        tardis_key: str,
        holy_sheep_key: str,
        redis_url: str = "redis://localhost:6379"
    ):
        self.tardis = TardisKLineAggregator(tardis_key)
        self.ai = HolySheepAIClient(holy_sheep_key)
        self.redis = redis.from_url(redis_url)
        
        # ตั้งค่า cache TTL
        self.cache_ttl = {
            "1m": 60,      # 1 นาที
            "5m": 300,     # 5 นาที
            "15m": 900,    # 15 นาที
            "1h": 3600,    # 1 ชั่วโมง
            "1d": 86400    # 1 วัน
        }
    
    def _get_cache_key(self, symbol: str, timeframe: str) -> str:
        return f"kline:{symbol}:{timeframe}"
    
    async def get_cached_klines(
        self, 
        symbol: str, 
        timeframe: str
    ) -> Optional[pd.DataFrame]:
        """ดึงข้อมูลจาก cache ก่อน"""
        cache_key = self._get_cache_key(symbol, timeframe)
        cached = self.redis.get(cache_key)
        
        if cached:
            return pickle.loads(cached)
        return None
    
    async def set_cached_klines(
        self,
        symbol: str,
        timeframe: str,
        df: pd.DataFrame
    ):
        """เก็บข้อมูลลง cache"""
        cache_key = self._get_cache_key(symbol, timeframe)
        self.redis.setex(
            cache_key,
            self.cache_ttl.get(timeframe, 300),
            pickle.dumps(df)
        )
    
    async def generate_signals(
        self,
        symbol: str,
        timeframe: str = "1h",
        exchanges: List[str] = None
    ) -> List[KLineSignal]:
        """สร้างสัญญาณซื้อ/ขายจากข้อมูล K-Line"""
        
        # ดึงข้อมูลจาก cache หรือ API
        df = await self.get_cached_klines(symbol, timeframe)
        
        if df is None:
            df = await self.tardis.fetch_historical_klines(
                exchange="binance",
                symbol=symbol,
                start_time=datetime.now() - timedelta(days=30),
                end_time=datetime.now(),
                timeframe=timeframe
            )
            await self.set_cached_klines(symbol, timeframe, df)
        
        # คำนวณ indicators
        df = self.ai.calculate_ohlcv_indicators(df)
        
        signals = []
        
        # RSI-based signals
        latest = df.iloc[-1]
        
        if latest['rsi'] < 30:
            signal_type = SignalType.BUY
            confidence = (30 - latest['rsi']) / 30
        elif latest['rsi'] > 70:
            signal_type = SignalType.SELL
            confidence = (latest['rsi'] - 70) / 30
        else:
            signal_type = SignalType.HOLD
            confidence = 0.5
        
        # MACD crossover signals
        if df['macd'].iloc[-1] > df['signal'].iloc[-1] and \
           df['macd'].iloc[-2] <= df['signal'].iloc[-2]:
            signal_type = SignalType.STRONG_BUY
            confidence = 0.9
        
        # วิเคราะห์ด้วย AI
        ai_analysis = await self.ai.analyze_kline_pattern(
            kline_data=df.tail(50).to_dict('records'),
            symbol=symbol,
            timeframe=timeframe
        )
        
        # คำนวณ stop loss และ take profit
        atr = df['high'].rolling(14).max() - df['low'].rolling(14).min()
        atr_value = atr.iloc[-1] if not atr.empty else latest['close'] * 0.02
        
        signal = KLineSignal(
            timestamp=datetime.now(),
            symbol=symbol,
            timeframe=timeframe,
            signal=signal_type,
            confidence=confidence,
            entry_price=latest['close'],
            stop_loss=latest['close'] - (atr_value * 1.5),
            take_profit=latest['close'] + (atr_value * 2),
            ai_analysis=ai_analysis['analysis']
        )
        
        signals.append(signal)
        
        return signals


class BacktestEngine:
    """Engine สำหรับทดสอบกลยุทธ์กับข้อมูลย้อนหลัง"""
    
    def __init__(self, engine: KLineEngine):
        self.engine = engine
        self.trades = []
        self.initial_balance = 10000  # USDT
        self.balance = self.initial_balance
        self.position = 0
    
    async def run_backtest(
        self,
        symbol: str,
        start_date: datetime,
        end_date: datetime,
        timeframe: str = "1h"
    ):
        """ทดสอบกลยุทธ์กับข้อมูลย้อนหลัง"""
        
        # ดึงข้อมูลย้อนหลัง
        df = await self.engine.tardis.fetch_historical_klines(
            exchange="binance",
            symbol=symbol,
            start_time=start_date,
            end_time=end_date,
            timeframe=timeframe
        )
        
        df = self.engine.ai.calculate_ohlcv_indicators(df)
        
        # วน loop ตามข้อมูล
        for i in range(50, len(df)):
            window = df.iloc[:i]
            current = df.iloc[i]
            
            # ตรวจสอบสัญญาณ
            rsi = current['rsi']
            
            if rsi < 30 and self.position == 0:
                # สัญญาณซื้อ
                self.position = self.balance / current['close'] * 0.95
                self.balance = 0
                self.trades.append({
                    'type': 'BUY',
                    'price': current['close'],
                    'time': current['timestamp'],
                    'quantity': self.position
                })
            
            elif rsi > 70 and self.position > 0:
                # สัญญาณขาย
                self.balance = self.position * current['close'] * 0.995
                self.trades.append({
                    'type': 'SELL',
                    'price': current['close'],
                    'time': current['timestamp'],
                    'quantity': self.position,
                    'profit': (current['close'] - self.trades[-1]['price']) * self.position
                })
                self.position = 0
        
        # คำนวณผลลัพธ์
        final_value = self.balance + (self.position * df.iloc[-1]['close'])
        total_return = (final_value - self.initial_balance) / self.initial_balance * 100
        
        return {
            'initial_balance': self.initial_balance,
            'final_value': final_value,
            'total_return': total_return,
            'num_trades': len(self.trades),
            'win_rate': self._calculate_win_rate()
        }
    
    def _calculate_win_rate(self) -> float:
        """คำนวณ win rate จาก trades"""
        if not self.trades:
            return 0
        
        sell_trades = [t for t in self.trades if t['type'] == 'SELL']
        if not sell_trades:
            return 0
        
        wins = sum(1 for t in sell_trades if t.get('profit', 0) > 0)
        return wins / len(sell_trades) * 100

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