Hướng dẫn验收衍生品历史行情回测 SLA：Tardis tick-level数据延迟、补档与审计留痕

Mở đầu: Khi backtest "như mơ" nhưng production "như ác mộng

Tôi còn nhớ rõ ngày đầu tiên làm việc tại một quỹ proprietary trading ở Thượng Hải — team quantitative của chúng tôi xây một chiến lược arbitrage options với Sharpe ratio 4.2 trên dữ liệu backtest. Mọi thứ hoàn hảo: slippage model 0.5bps, commission 0.2bps, fill rate 98%. Kết quả backtest công bố cho management, ai cũng phấn khích. Sáu tháng sau khi lên production? Hệ thống thua lỗ 340,000 USD trong vòng 3 tuần. Nguyên nhân không phải ở thuật toán — mà ở dữ liệu. Độ trễ thực tế của Tardis tick data lên đến 2.3 giây thay vì 50ms cam kết, có 7.2% ticks bị mất không được补档 (bù), và không có audit trail để trace lỗi. Đó là bài học đắt giá nhất trong sự nghiệp quant của tôi. Bài viết này sẽ hướng dẫn chi tiết và thực chiến cách验收 (nghiệm thu) SLA cho hệ thống backtest dữ liệu phái sinh tick-level, tập trung vào ba trụ cột: độ trễ thực tế, cơ chế补档, và audit trail.

Tardis Tick Data: Tổng quan kiến trúc

Tardis Hồi Quy khác gì so với nguồn khác?

Trước khi đi vào验收 SLA, cần hiểu Tardis là gì và tại sao nó phổ biến trong giới quantitative. Tardis (Tardis.dev) cung cấp historical market data cho crypto derivatives với độ phân giải tick-level — nghĩa là mỗi lần price/khối lượng thay đổi đều được ghi nhận. So với candlestick 1-minute, tick data chính xác hơn 10,000 lần cho việc backtest chiến lược high-frequency.

So sánh các nguồn dữ liệu backtest phổ biến

Nguồn dữ liệu    | Độ phân giải | Độ trễ thực tế | Độ phủ | Chi phí/tháng
-----------------|--------------|----------------|--------|----------------
Tardis Crypto    | Tick-level   | 50ms-2s        | 95%+   | $299-999
Binance History  | Kline/Candle | 1-5 phút       | 99%+   | Miễn phí
Polygon.io       | Tick-level   | 10-100ms       | 98%+   | $200-2000
Tick Data LLC    | Tick-level   | 5-50ms         | 92%+   | $500-5000
HolySheep AI     | Tick-level   | <50ms          | 99.5%+ | $8-15 (MTok)

Lưu ý quan trọng: Tardis cam kết độ trễ 50ms nhưng thực tế trong điều kiện mạng kém hoặc lúc volatility cao (như flash crash), độ trễ có thể lên đến 2-5 giây. Đây là lý do验收 SLA không chỉ là đọc specs mà phải measure thực tế.

Phần 1: Độ trễ dữ liệu — Cách measure và验收

1.1. Cài đặt hệ thống đo độ trễ

Để measure độ trễ Tardis tick data, bạn cần ghi nhận timestamp tại ba điểm:

• T₁: Timestamp server gửi — nằm trong payload của Tardis response
• T₂: Timestamp nhận — khi request hoàn tất (network + processing)
• T₃: Timestamp xử lý — khi tick được đưa vào backtest engine

Độ trễ thực tế = T₃ - T₁

Python script measure độ trễ Tardis tick data
Cài đặt: pip install tardis-dev aiohttp asyncio

import aiohttp
import asyncio
import time
from dataclasses import dataclass
from typing import List
from datetime import datetime

@dataclass
class TickMeasurement:
    exchange: str
    symbol: str
    tardis_timestamp: float
    received_timestamp: float
    processed_timestamp: float
    latency_ms: float
    is_anomaly: bool = False

class TardisLatencyMonitor:
    def __init__(self, api_key: str, threshold_ms: float = 100.0):
        self.api_key = api_key
        self.threshold_ms = threshold_ms
        self.measurements: List[TickMeasurement] = []
        
    async def fetch_ticker_feed(self, exchange: str, symbol: str):
        """Kết nối Tardis real-time feed và measure độ trễ"""
        url = f"https://api.tardis.dev/v1/feeds/{exchange}:{symbol}"
        headers = {"Authorization": f"Bearer {self.api_key}"}
        
        async with aiohttp.ClientSession() as session:
            async with session.get(url, headers=headers) as resp:
                async for line in resp.content:
                    if line:
                        data = line.decode().strip()
                        if data:
                            await self.process_tick(data)
    
    async def process_tick(self, raw_data: str):
        """Xử lý tick và measure độ trễ"""
        import json
        try:
            tick = json.loads(raw_data)
            
            # T₁: Timestamp từ Tardis (trong payload)
            tardis_ts = tick.get('timestamp', tick.get('localTimestamp', time.time()))
            
            # T₂: Timestamp nhận (ngay khi nhận được)
            received_ts = time.time()
            
            # T₃: Timestamp xử lý (sau khi parse và validate)
            processed_ts = time.time()
            
            latency_ms = (processed_ts - tardis_ts) * 1000
            
            measurement = TickMeasurement(
                exchange=tick.get('exchange'),
                symbol=tick.get('symbol'),
                tardis_timestamp=tardis_ts,
                received_timestamp=received_ts,
                processed_timestamp=processed_ts,
                latency_ms=latency_ms,
                is_anomaly=latency_ms > self.threshold_ms
            )
            
            self.measurements.append(measurement)
            
            # Alert nếu vượt ngưỡng
            if measurement.is_anomaly:
                print(f"[ALERT] Latency cao: {latency_ms:.2f}ms - {measurement.exchange}:{measurement.symbol}")
                
        except json.JSONDecodeError:
            pass
    
    def get_sla_report(self) -> dict:
        """Tạo báo cáo SLA dựa trên measurements"""
        if not self.measurements:
            return {}
        
        latencies = [m.latency_ms for m in self.measurements]
        
        return {
            "total_ticks": len(latencies),
            "p50_latency_ms": sorted(latencies)[len(latencies) // 2],
            "p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)],
            "p99_latency_ms": sorted(latencies)[int(len(latencies) * 0.99)],
            "avg_latency_ms": sum(latencies) / len(latencies),
            "max_latency_ms": max(latencies),
            "anomaly_count": sum(1 for m in self.measurements if m.is_anomaly),
            "sla_compliance": (1 - sum(1 for m in self.measurements if m.is_anomaly) / len(self.measurements)) * 100
        }

Chạy monitor
async def main():
    monitor = TardisLatencyMonitor(api_key="YOUR_TARDIS_API_KEY", threshold_ms=100.0)
    
    # Monitor 5 phút = 300 giây
    monitor_task = asyncio.create_task(
        monitor.fetch_ticker_feed("binance", "btcusdt_perpetual_future")
    )
    
    await asyncio.sleep(300)  # 5 phút
    monitor_task.cancel()
    
    # Xuất báo cáo SLA
    report = monitor.get_sla_report()
    print(f"\n=== SLA Report ===")
    print(f"P50: {report['p50_latency_ms']:.2f}ms")
    print(f"P95: {report['p95_latency_ms']:.2f}ms")
    print(f"P99: {report['p99_latency_ms']:.2f}ms")
    print(f"SLA Compliance: {report['sla_compliance']:.2f}%")

asyncio.run(main())

1.2. Tiêu chí验收 SLA độ trễ

Dựa trên kinh nghiệm thực chiến với nhiều dự án, đây là tiêu chí验收 tôi đề xuất:

Chỉ số	Yêu cầu SLA	Nghiêm ngặt (HFT)	Cảnh báo
P50 Latency	≤50ms	≤10ms	>100ms
P95 Latency	≤200ms	≤50ms	>500ms
P99 Latency	≤500ms	≤100ms	>1000ms
Max Latency	≤2000ms	≤500ms	>5000ms
Availability	≥99.5%	≥99.9%	<99%
Data Completeness	≥99%	≥99.9%	<98%

Lưu ý: Đối với backtest chiến lược mean-reversion, P95 ≤ 200ms là đủ. Nhưng với arbitrage lat-sensitive, bạn cần P99 ≤ 100ms và max ≤ 500ms.

Phần 2: Cơ chế补档 (Bù dữ liệu bị thiếu)

2.1. Tại sao dữ liệu bị thiếu?

Dữ liệu tick có thể bị thiếu vì nhiều lý do:

• Network interruption: Mất kết nối 5-30 phút → ticks trong khoảng đó biến mất
• Exchange downtime: Binance maintenance 2-4 giờ mỗi tuần
• API rate limit: Tardis có rate limit, request bị drop khi vượt quota
• Storage gap: Tardis lưu trữ theo partition, có thể có gap ở partition boundaries
• Replay buffer overflow: Khi system bận xử lý, buffer đầy → ticks bị drop

Trong backtest, ngay cả 1% dữ liệu thiếu cũng có thể tạo ra kết quả sai lệch nghiêm trọng. Tôi đã thấy một chiến lược grid trading có drawdown thực tế gấp 3 lần so với backtest chỉ vì 2.3% ticks bị thiếu ở giai đoạn volatility cao.

2.2. Hệ thống补档 tự động

Hệ thống补档 tự động với multi-source fallback
pip install tardis-dev binance-connector sqlalchemy asyncpg

import asyncio
from datetime import datetime, timedelta
from typing import Optional, List, Tuple
from dataclasses import dataclass
from enum import Enum
import aiohttp
from binance.client import Client
import asyncpg

class DataSource(Enum):
    TARDIS = "tardis"
    BINANCE_HISTORY = "binance_history"
    HOLYSHEEP_AI = "holysheep_ai"
    CACHE = "cache"

@dataclass
class GapInfo:
    exchange: str
    symbol: str
    start_time: datetime
    end_time: datetime
    missing_count: int
    filled_count: int = 0
    sources_used: List[DataSource] = None

class DataGapFiller:
    def __init__(self, config: dict):
        self.tardis_api_key = config['tardis_api_key']
        self.binance_client = Client(
            config['binance_api_key'],
            config['binance_secret_key']
        )
        self.pool = None  # asyncpg connection pool
        self.cache = {}  # In-memory cache for filled data
        self.holysheep_base_url = "https://api.holysheep.ai/v1"
        self.holysheep_api_key = config.get('holysheep_api_key')
        
    async def initialize(self):
        """Khởi tạo database connection"""
        self.pool = await asyncpg.create_pool(
            host='localhost',
            port=5432,
            user='trader',
            password='password',
            database='market_data'
        )
    
    async def detect_gaps(self, exchange: str, symbol: str, 
                         start_time: datetime, end_time: datetime,
                         expected_interval_ms: int = 100) -> List[GapInfo]:
        """Phát hiện các khoảng trống dữ liệu"""
        
        query = """
            WITH ticks AS (
                SELECT 
                    timestamp,
                    LAG(timestamp) OVER (ORDER BY timestamp) as prev_timestamp
                FROM ticks_{exchange}_{symbol}
                WHERE timestamp BETWEEN $1 AND $2
            )
            SELECT 
                prev_timestamp,
                timestamp,
                EXTRACT(EPOCH FROM (timestamp - prev_timestamp)) * 1000 as gap_ms
            FROM ticks
            WHERE EXTRACT(EPOCH FROM (timestamp - prev_timestamp)) * 1000 > $3
        """.format(exchange=exchange, symbol=symbol.replace('/', '_'))
        
        async with self.pool.acquire() as conn:
            rows = await conn.fetch(query, start_time, end_time, expected_interval_ms * 3)
        
        gaps = []
        for row in rows:
            gap_info = GapInfo(
                exchange=exchange,
                symbol=symbol,
                start_time=row['prev_timestamp'],
                end_time=row['timestamp'],
                missing_count=int(row['gap_ms'] / expected_interval_ms),
                sources_used=[]
            )
            gaps.append(gap_info)
        
        return gaps
    
    async def fill_gap_from_tardis(self, gap: GapInfo) -> bool:
        """Thử fill từ Tardis historical API"""
        url = f"https://api.tardis.dev/v1/replay"
        params = {
            "exchange": gap.exchange,
            "symbol": gap.symbol,
            "from": int(gap.start_time.timestamp()),
            "to": int(gap.end_time.timestamp()),
            "format": "json"
        }
        headers = {"Authorization": f"Bearer {self.tardis_api_key}"}
        
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(url, params=params, headers=headers) as resp:
                    if resp.status == 200:
                        data = await resp.json()
                        await self._store_ticks(data)
                        gap.filled_count = len(data)
                        gap.sources_used.append(DataSource.TARDIS)
                        return True
        except Exception as e:
            print(f"Tardis fill failed: {e}")
        return False
    
    async def fill_gap_from_binance(self, gap: GapInfo) -> bool:
        """Fallback: Fill từ Binance historical kline"""
        try:
            # Chuyển đổi sang kline data (1-minute)
            klines = self.binance_client.get_historical_klines(
                gap.symbol.replace('_perpetual', ''),
                Client.KLINE_INTERVAL_1MINUTE,
                gap.start_time.strftime("%d %b %Y %H:%M:%S"),
                gap.end_time.strftime("%d %b %Y %H:%M:%S")
            )
            
            # Convert kline to tick-like format
            ticks = []
            for k in klines:
                ticks.append({
                    'timestamp': k[0] / 1000,
                    'price': float(k[4]),  # close price
                    'volume': float(k[5])
                })
            
            await self._store_ticks(ticks)
            gap.filled_count = len(ticks)
            gap.sources_used.append(DataSource.BINANCE_HISTORY)
            return True
        except Exception as e:
            print(f"Binance fill failed: {e}")
        return False
    
    async def fill_gap_from_holysheep(self, gap: GapInfo) -> bool:
        """Fallback với HolySheep AI - chi phí thấp, độ trễ <50ms"""
        # HolySheep cung cấp market data API với độ phủ cao
        url = f"{self.holysheep_base_url}/market-data/historical"
        headers = {
            "Authorization": f"Bearer {self.holysheep_api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "exchange": gap.exchange,
            "symbol": gap.symbol,
            "start_time": gap.start_time.isoformat(),
            "end_time": gap.end_time.isoformat(),
            "resolution": "tick"
        }
        
        try:
            async with aiohttp.ClientSession() as session:
                async with session.post(url, json=payload, headers=headers) as resp:
                    if resp.status == 200:
                        data = await resp.json()
                        await self._store_ticks(data.get('ticks', []))
                        gap.filled_count = len(data.get('ticks', []))
                        gap.sources_used.append(DataSource.HOLYSHEEP_AI)
                        return True
        except Exception as e:
            print(f"HolySheep fill failed: {e}")
        return False
    
    async def _store_ticks(self, ticks: List[dict]):
        """Lưu ticks vào database"""
        values = [
            (
                datetime.fromtimestamp(t['timestamp']),
                t.get('price'),
                t.get('volume', 0)
            )
            for t in ticks
        ]
        
        async with self.pool.acquire() as conn:
            await conn.executemany("""
                INSERT INTO ticks_binance_btcusdt_perpetual (timestamp, price, volume)
                VALUES ($1, $2, $3)
                ON CONFLICT (timestamp) DO UPDATE SET
                    price = EXCLUDED.price,
                    volume = EXCLUDED.volume
            """, values)
    
    async def auto_fill_gaps(self, gaps: List[GapInfo]) -> dict:
        """Tự động fill gaps với multi-source fallback"""
        results = {
            'total_gaps': len(gaps),
            'filled': 0,
            'failed': 0,
            'details': []
        }
        
        for gap in gaps:
            filled = False
            
            # Thứ tự fallback: Tardis → Binance → HolySheep
            if await self.fill_gap_from_tardis(gap):
                filled = True
            elif await self.fill_gap_from_binance(gap):
                filled = True
            elif await self.fill_gap_from_holysheep(gap):
                filled = True
            
            if filled:
                results['filled'] += 1
            else:
                results['failed'] += 1
                results['details'].append({
                    'gap': gap,
                    'status': 'FAILED - manual intervention required'
                })
        
        return results

Sử dụng
async def main():
    filler = DataGapFiller({
        'tardis_api_key': 'YOUR_TARDIS_KEY',
        'binance_api_key': 'YOUR_BINANCE_KEY',
        'binance_secret_key': 'YOUR_BINANCE_SECRET',
        'holysheep_api_key': 'YOUR_HOLYSHEEP_KEY'
    })
    await filler.initialize()
    
    # Phát hiện và fill gaps
    gaps = await filler.detect_gaps(
        exchange='binance',
        symbol='btcusdt_perpetual_future',
        start_time=datetime(2024, 1, 1),
        end_time=datetime(2024, 1, 7)
    )
    
    results = await filler.auto_fill_gaps(gaps)
    print(f"Fill results: {results}")

asyncio.run(main())

2.3.验收补档 completeness

Sau khi fill, cần xác minh completeness:

Script验收补档 completeness
Run sau khi chạy DataGapFiller

import asyncio
import asyncpg
from datetime import datetime, timedelta
from typing import Dict, List

async def validate_completeness(
    pool: asyncpg.Pool,
    exchange: str,
    symbol: str,
    start_time: datetime,
    end_time: datetime
) -> Dict:
    """
    Kiểm tra completeness sau khi fill
    Returns detailed report về data quality
    """
    
    table_name = f"ticks_{exchange}_{symbol.replace('/', '_')}"
    
    queries = {
        # Total ticks expected (assuming avg 10 ticks/sec for BTC perpetual)
        'expected_ticks': f"""
            SELECT 
                EXTRACT(EPOCH FROM ($2::timestamp - $1::timestamp)) as duration_seconds,
                EXTRACT(EPOCH FROM ($2::timestamp - $1::timestamp)) * 10 as expected_ticks
        """,
        
        # Actual ticks in DB
        'actual_ticks': f"""
            SELECT COUNT(*) as count
            FROM {table_name}
            WHERE timestamp BETWEEN $1 AND $2
        """,
        
        # Ticks per hour distribution
        'hourly_distribution': f"""
            SELECT 
                DATE_TRUNC('hour', timestamp) as hour,
                COUNT(*) as tick_count
            FROM {table_name}
            WHERE timestamp BETWEEN $1 AND $2
            GROUP BY DATE_TRUNC('hour', timestamp)
            ORDER BY hour
        """,
        
        # Find suspicious gaps
        'suspicious_gaps': f"""
            WITH tick_gaps AS (
                SELECT 
                    timestamp,
                    LAG(timestamp) OVER (ORDER BY timestamp) as prev_ts,
                    EXTRACT(EPOCH FROM (timestamp - LAG(timestamp) OVER (ORDER BY timestamp))) as gap_sec
                FROM {table_name}
                WHERE timestamp BETWEEN $1 AND $2
            )
            SELECT * FROM tick_gaps
            WHERE gap_sec > 10  -- Gap lớn hơn 10 giây
            ORDER BY gap_sec DESC
            LIMIT 20
        """
    }
    
    async with pool.acquire() as conn:
        # Get expected ticks
        duration_row = await conn.fetchrow(
            queries['expected_ticks'], start_time, end_time
        )
        expected_ticks = int(duration_row['expected_ticks'])
        
        # Get actual ticks
        actual_row = await conn.fetchrow(
            queries['actual_ticks'], start_time, end_time
        )
        actual_ticks = actual_row['count']
        
        # Get hourly distribution
        hourly_rows = await conn.fetch(
            queries['hourly_distribution'], start_time, end_time
        )
        
        # Get suspicious gaps
        gap_rows = await conn.fetch(
            queries['suspicious_gaps'], start_time, end_time
        )
    
    completeness_pct = (actual_ticks / expected_ticks * 100) if expected_ticks > 0 else 0
    
    #验收 criteria
    validation_result = {
        'period': f"{start_time} to {end_time}",
        'expected_ticks': expected_ticks,
        'actual_ticks': actual_ticks,
        'completeness_pct': completeness_pct,
        'missing_ticks': expected_ticks - actual_ticks,
        'hourly_distribution': [
            {'hour': r['hour'], 'count': r['tick_count']}
            for r in hourly_rows
        ],
        'suspicious_gaps': [
            {'timestamp': r['timestamp'], 'gap_seconds': r['gap_sec']}
            for r in gap_rows
        ],
        'validation_status': 'PASS' if completeness_pct >= 99.0 else 'FAIL',
        'sla_compliance': completeness_pct >= 99.0,
        'recommendations': []
    }
    
    # Generate recommendations
    if completeness_pct < 99.0:
        validation_result['recommendations'].append(
            f"Completeness {completeness_pct:.2f}% < 99% SLA - cần investigate gaps"
        )
    
    if len(gap_rows) > 0:
        validation_result['recommendations'].append(
            f"Tìm thấy {len(gap_rows)} suspicious gaps - xem chi tiết suspicious_gaps"
        )
    
    # Check for zero-tick hours
    zero_hours = [h for h in validation_result['hourly_distribution'] if h['count'] == 0]
    if zero_hours:
        validation_result['recommendations'].append(
            f"{len(zero_hours)} giờ không có tick - có thể là exchange downtime"
        )
    
    return validation_result

async def main():
    pool = await asyncpg.create_pool(
        host='localhost', port=5432, user='trader', 
        password='password', database='market_data'
    )
    
    result = await validate_completeness(
        pool=pool,
        exchange='binance',
        symbol='btcusdt_perpetual_future',
        start_time=datetime(2024, 1, 1),
        end_time=datetime(2024, 1, 7)
    )
    
    print(f"\n{'='*50}")
    print(f"VALIDATION REPORT")
    print(f"{'='*50}")
    print(f"Period: {result['period']}")
    print(f"Expected: {result['expected_ticks']:,} ticks")
    print(f"Actual: {result['actual_ticks']:,} ticks")
    print(f"Completeness: {result['completeness_pct']:.4f}%")
    print(f"Status: {result['validation_status']}")
    print(f"SLA Compliance: {result['sla_compliance']}")
    
    if result['recommendations']:
        print(f"\nRecommendations:")
        for rec in result['recommendations']:
            print(f"  - {rec}")
    
    await pool.close()

asyncio.run(main())

Phần 3: Audit Trail — Theo dõi và khả năng trace

3.1. Tại sao audit trail quan trọng?

Trong backtest production, audit trail không phải là optional — nó là REQUIREMENT để:

• Reproducibility: Có thể reproduce lại kết quả backtest bất kỳ lúc nào
• Compliance: Regulators yêu cầu trace source của mọi signal giao dịch
• Debugging: Khi chiến lược thua lỗ, cần trace ngược về data source
• Accountability: Ai thay đổi data hoặc strategy cần được ghi nhận

3.2. Hệ thống audit trail đầy đủ

Audit Trail System cho Backtest
PostgreSQL + Python + HolySheep AI for logging

import asyncio
import asyncpg
import hashlib
import json
from datetime import datetime
from typing import Optional, Dict, Any
from dataclasses import dataclass, asdict
from enum import Enum
import aiohttp

class AuditEventType(Enum):
    DATA_FETCH = "data_fetch"
    DATA_FILL = "data_fill"
    BACKTEST_RUN = "backtest_run"
    PARAMETER_CHANGE = "parameter_change"
    SIGNAL_GENERATED = "signal_generated"
    ORDER_PLACED = "order_placed"
    GAP_DETECTED = "gap_detected"
    ANOMALY_DETECTED = "anomaly_detected"

@dataclass
class AuditEntry:
    event_id: str  # UUID
    timestamp: datetime
    event_type: str
    user_id: Optional[str]
    source: str  # tardis, binance, holysheep, manual
    details: Dict[str, Any]
    data_hash: str  # SHA256 của data liên quan
    parent_event_id: Optional[str]  # Link to parent event
    metadata: Dict[str, Any]
    
    def to_dict(self) -> dict:
        d = asdict(self)
        d['timestamp'] = self.timestamp.isoformat()
        d['event_type'] = self.event_type
        return d

class BacktestAuditTrail:
    def __init__(self, db_config: dict, holysheep_key: str):
        self.pool = None
        self.db_config = db_config
        self.holysheep_key = holysheep_key
        self.holysheep_base_url = "https://api.holysheep.ai/v1"
        
    async def initialize(self):
        """Tạo bảng audit trail"""
        self.pool = await asyncpg.create_pool(**self.db_config)
        
        async with self.pool.acquire() as conn:
            await conn.execute("""
                CREATE TABLE IF NOT EXISTS backtest_audit (
                    event_id UUID PRIMARY KEY,
                    timestamp TIMESTAMPTZ NOT NULL,
                    event_type VARCHAR(50) NOT NULL,
                    user_id VARCHAR(100),
                    source VARCHAR(50) NOT NULL,
                    details JSONB,
                    data_hash VARCHAR(64),
                    parent_event_id UUID REFERENCES backtest_audit(event_id),
                    metadata JSONB,
                    created_at TIMESTAMPTZ DEFAULT NOW()
                )
            """)
            
            # Index cho performance
            await conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_audit_timestamp ON backtest_audit(timestamp)
            """)
            await conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_audit_event_type ON backtest_audit(event_type)
            """)
            await conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_audit_parent ON backtest_audit(parent_event_id)
            """)
    
    def generate_hash(self, data: Any) -> str:
        """Generate SHA256 hash cho data"""
        json_str = json.dumps(data, sort_keys=True, default=str)
        return hashlib.sha256(json_str.encode()).hexdigest()
    
    async def log_event(
        self,
        event_type: AuditEventType,
        source: str,
        details: Dict[str, Any],
        user_id: Optional[str] = None,
        parent_event_id: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None
    ) -> AuditEntry:
        """Log một audit event"""
        
        import uuid
        
        entry = AuditEntry(
            event_id=str(uuid.uuid4()),
            timestamp=datetime.utcnow(),
            event_type=event_type.value,
            user_id=user_id,
            source=source,
            details=details,
            data_hash=self.generate_hash(details),
            parent_event_id=parent_event_id,
            metadata=metadata or {}
        )
        
        async with self.pool.acquire() as conn:
            await conn.execute("""
                INSERT INTO backtest_audit 
                (event_id, timestamp, event_type, user_id, source, details, data_hash, parent_event_id, metadata)
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
            """, 
                entry.event_id
Tài nguyên liên quan
📚 Hướng dẫn AI API
💰 Xem giá
📖 Tài liệu nhà phát triển
🚀 Đăng ký miễn phí
Bài viết liên quan
Gemini 2.5 Pro API: Hướng Dẫn Di Chuyển Sang HolySheep Gatew