加密货币量化数据管道：AI 大模型实时聚合多交易所数据清洗

การใช้ DeepSeek V3.2 สำหรับ Data Cleansing

"""
Data Cleansing Module ใช้ HolySheep AI
"""
import numpy as np
from scipy import stats
import json

class DataCleansingPipeline:
    """
    ระบบทำความสะอาดข้อมูล crypto แบบอัตโนมัติ
    รองรับ: outlier detection, missing data imputation, time sync
    """
    
    def __init__(self, api_key: str):
        self.holy_sheep_client = HolySheepAIClient(api_key)
        self.outlier_z_threshold = 3.0
        self.missing_threshold = 0.1  # 10% missing allowed
        
    async def clean_cross_exchange_data(self, data_points: List[Dict]) -> Dict:
        """
        ทำความสะอาดข้อมูลจากหลาย Exchange
        และสร้าง unified price feed
        """
        # ขั้นตอนที่ 1: ตรวจจับ outlier เบื้องต้น
        prices = [d['price'] for d in data_points]
        z_scores = stats.zscore(prices)
        
        initial_clean = [
            d for d, z in zip(data_points, z_scores) 
            if abs(z) < self.outlier_z_threshold
        ]
        
        # ขั้นตอนที่ 2: ใช้ AI วิเคราะห์ความผิดปกติเชิงลึก
        ai_analysis = await self.holy_sheep_client.analyze_anomalies(
            initial_clean,
            context="cross-exchange arbitrage detection"
        )
        
        # ขั้นตอนที่ 3: คำนวณ weighted average
        final_price = self.calculate_weighted_price(
            initial_clean,
            ai_analysis.get('confidence_scores', {})
        )
        
        # ขั้นตอนที่ 4: ตรวจสอบ missing data
        completeness = self.check_completeness(initial_clean)
        
        return {
            'symbol': data_points[0]['symbol'],
            'cleaned_price': final_price,
            'sources': len(data_points),
            'completeness': completeness,
            'anomalies_detected': len(data_points) - len(initial_clean),
            'timestamp': datetime.utcnow().isoformat()
        }
    
    async def handle_missing_data(self, symbol: str, timeframe: str) -> Optional[float]:
        """เติมข้อมูลที่หายไปด้วย AI prediction"""
        
        # ดึงข้อมูลรอบข้างเพื่อใช้ในการ predict
        context_data = await self.get_context_data(symbol, timeframe)
        
        if not context_data:
            return None
            
        prompt = f"""
        Based on this historical price data for {symbol}:
        {json.dumps(context_data, indent=2)}
        
        Predict the missing price value. Return only the numeric value.
        Consider: trend, volatility, and recent patterns.
        """
        
        predicted_price = await self.holy_sheep_client.predict(
            prompt,
            model="deepseek-v3.2"
        )
        
        return float(predicted_price)
    
    def calculate_weighted_price(self, data: List[Dict], confidence: Dict) -> float:
        """คำนวณราคาเฉลี่ยถ่วงน้ำหนักตาม confidence score"""
        
        total_weight = 0
        weighted_sum = 0
        
        for d in data:
            source = d['exchange']
            weight = confidence.get(source, 1.0)
            weighted_sum += d['price'] * weight
            total_weight += weight
            
        return weighted_sum / total_weight if total_weight > 0 else 0
    
    async def sync_timestamps(self, data_points: List[Dict]) -> List[Dict]:
        """ซิงโครไนซ์ timestamp จาก Exchange ต่างๆ"""
        
        # รวบรวม timestamp ทั้งหมด
        timestamps = [d['timestamp'] for d in data_points]
        
        # หา median timestamp เป็น reference
        sorted_times = sorted(timestamps)
        median_time = sorted_times[len(sorted_times) // 2]
        
        # ปรับ timestamp ให้ตรงกับ median
        adjusted_data = []
        for d in data_points:
            time_diff = median_time - d['timestamp']
            adjusted = d.copy()
            adjusted['time_offset'] = time_diff
            adjusted_data.append(adjusted)
            
        return adjusted_data


class HolySheepAIClient:
    """Client สำหรับเรียกใช้ HolySheep AI API"""
    
    def __init__(self, api_key: str):
        self.base_url = "https://api.holysheep.ai/v1"
        self.api_key = api_key
        
    async def analyze_anomalies(self, data: List[Dict], context: str) -> Dict:
        """วิเคราะห์ความผิดปกติของข้อมูลด้วย AI"""
        
        async with httpx.AsyncClient(timeout=60.0) as client:
            response = await client.post(
                f"{self.base_url}/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "deepseek-v3.2",
                    "messages": [
                        {
                            "role": "system",
                            "content": """You are a cryptocurrency data quality expert. 
                            Analyze price data for anomalies considering:
                            1. Price deviation from market average
                            2. Unusual volume patterns  
                            3. Exchange-specific anomalies
                            Return JSON with confidence scores for each source."""
                        },
                        {
                            "role": "user",
                            "content": f"Analyze this data for {context}:\n{json.dumps(data, indent=2)}"
                        }
                    ],
                    "temperature": 0.1
                }
            )
            
            result = response.json()
            return json.loads(result['choices'][0]['message']['content'])
    
    async def predict(self, prompt: str, model: str = "deepseek-v3.2") -> str:
        """ทำนายค่าที่หายไปด้วย AI"""
        
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(
                f"{self.base_url}/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": model,
                    "messages": [{"role": "user", "content": prompt}],
                    "temperature": 0.3
                }
            )
            
            return response.json()['choices'][0]['message']['content']

# ❌ วิธีที่ผิด: ไม่มี reconnection logic
async def bad_connect():
    async with websockets.connect(uri) as ws:
        async for msg in ws:
            process(msg)
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
向量数据库选型实战：Pinecone vs Milvus vs Weaviate บทวิเคราะห์เปรียบเท
DeepSeek R1 vs o1-mini: คู่มือฉบับเข้าใจลึกสำหรับวิศวกร AI A
Kimi K2 Long Text Processing: รีวิวเชิงลึก 200K Context พร้อ