Trong hành trình xây dựng hệ thống giao dịch định lượng, việc chuẩn bị dữ liệu chất lượng cao quyết định 85% thành bại của chiến lược. Bài viết này chia sẻ kinh nghiệm thực chiến từ đội ngũ HolySheep AI — những người đã từng vật lộn với latency 2000ms, chi phí API 300$/tháng, và vô số lỗi rate-limit khi xây dựng pipeline backtesting cho quỹ tại Việt Nam. Chúng tôi sẽ hướng dẫn bạn từng bước kết nối OKX API, chuẩn hóa dữ liệu, và tích hợp HolySheep AI để tối ưu chi phí và hiệu suất.
Vì sao cần chuẩn bị dữ liệu OKX chất lượng cao?
OKX là một trong những sàn giao dịch phái sinh lớn nhất thế giới với khối lượng giao dịch Futures và Perpetual vượt 2 tỷ USD mỗi ngày. Tuy nhiên, dữ liệu thô từ OKX API chứa nhiều vấn đề:
- Tick data không đồng nhất: Khi thị trường biến động mạnh, OKX gửi nhiều event cùng timestamp, gây sai lệch khi tính VWAP
- Missing data points: API rate-limit khiến data gap xuất hiện trong các giai đoạn cao điểm
- Clock drift: Server OKX sử dụng timestamp UTC, nhưng nhiều trader Việt Nam quên convert timezone
- Chi phí infrastructure: Lưu trữ tick-level data cho 1 năm tốn 500GB+ disk và nhiều RAM để query nhanh
Kiến trúc hệ thống backtesting với OKX + HolySheep
Đội ngũ HolySheep đã thiết kế pipeline xử lý 10 triệu tick/ngày với chi phí chỉ 12$/tháng (so với 280$ khi dùng các relay khác). Kiến trúc bao gồm:
+------------------+ +------------------+ +------------------+
| OKX WebSocket |---->| Data Collector |---->| PostgreSQL |
| wss://ws.okx.com| | (Rust/Python) | | (TimescaleDB) |
+------------------+ +------------------+ +------------------+
|
v
+------------------+ +------------------+
| HolySheep AI API |<----| Data Enrichment |
| gpt-4.1 / deepseek| | + Signal Gen |
+------------------+ +------------------+
|
v
+------------------+ +------------------+
| Backtesting |<----| Feature Store |
| Engine | | (Vector DB) |
+------------------+ +------------------+
Triển khai bước 1: Kết nối OKX WebSocket API
Đầu tiên, chúng ta cần thiết lập kết nối WebSocket để nhận real-time market data. Dưới đây là implementation production-ready với error handling và reconnection logic.
import asyncio
import json
import hmac
import hashlib
import base64
import time
from datetime import datetime, timezone
from typing import Optional, Callable
import aiohttp
class OKXWebSocketClient:
"""
Production-grade OKX WebSocket client với auto-reconnect
và buffer để xử lý burst traffic khi backtesting.
"""
def __init__(
self,
api_key: str,
api_secret: str,
passphrase: str,
sandbox: bool = False
):
self.api_key = api_key
self.api_secret = api_secret
self.passphrase = passphrase
self.sandbox = sandbox
self.base_url = (
"wss://wspap.okx.com:8443/ws/v5/public"
if not sandbox
else "wss://wspap.okx.com:8443/ws/v5/public?brokerId=99"
)
self._ws: Optional[aiohttp.ClientWebSocketResponse] = None
self._session: Optional[aiohttp.ClientSession] = None
self._connected = False
self._ping_interval = 20 # OKX yêu cầu ping mỗi 20s
def _sign(self, timestamp: str, method: str, path: str, body: str = "") -> str:
"""Tạo signature cho request - bắt buộc cho private channel"""
message = timestamp + method + path + body
mac = hmac.new(
self.api_secret.encode('utf-8'),
message.encode('utf-8'),
hashlib.sha256
)
return base64.b64encode(mac.digest()).decode('utf-8')
async def connect(self):
"""Thiết lập WebSocket connection với handshake"""
self._session = aiohttp.ClientSession()
headers = {
"Content-Type": "application/json"
}
self._ws = await self._session.ws_connect(
self.base_url,
headers=headers,
timeout=aiohttp.ClientTimeout(total=30)
)
# Đăng ký private channel nếu cần
if self.api_key:
await self._login()
self._connected = True
print(f"[{datetime.now(timezone.utc)}] ✅ OKX WebSocket connected")
async def _login(self):
"""Login để truy cập private data như account balance"""
timestamp = str(time.time())
signature = self._sign(timestamp, "GET", "/users/self/verify")
login_args = {
"op": "login",
"args": [{
"apiKey": self.api_key,
"passphrase": self.passphrase,
"timestamp": timestamp,
"sign": signature
}]
}
await self._ws.send_json(login_args)
response = await self._ws.receive_json()
if response.get("code") != "0":
raise ConnectionError(f"OKX login failed: {response}")
print(f"[{datetime.now(timezone.utc)}] 🔐 OKX authentication successful")
async def subscribe(self, channel: str, instId: str = "BTC-USDT-SWAP"):
"""Đăng ký channel - ví dụ: trades, books, candles"""
subscribe_args = {
"op": "subscribe",
"args": [{
"channel": channel,
"instId": instId
}]
}
await self._ws.send_json(subscribe_args)
response = await self._ws.receive_json()
if response.get("code") == "0":
print(f"📊 Subscribed to {channel} for {instId}")
else:
print(f"⚠️ Subscription warning: {response}")
async def fetch_historical_candles(
self,
instId: str,
bar: str = "1m",
after: Optional[int] = None,
before: Optional[int] = None,
limit: int = 100
) -> list:
"""
Lấy historical candlestick data qua REST API.
Rate limit: 20 requests/2s cho public, 20 requests/2s cho private
"""
# Ép rate limit để tránh 429
await asyncio.sleep(0.11) # 20 / 2 = 10 req/s, thêm buffer
base = "https://www.okx.com" if not self.sandbox else "https://www.okx.com"
endpoint = f"{base}/api/v5/market/history-candles"
params = {
"instId": instId,
"bar": bar,
"limit": min(limit, 100) # Max 100 records/request
}
if after:
params["after"] = after
if before:
params["before"] = before
async with self._session.get(endpoint, params=params) as resp:
if resp.status == 429:
print("⚠️ Rate limited! Waiting 2s...")
await asyncio.sleep(2)
return await self.fetch_historical_candles(
instId, bar, after, before, limit
)
data = await resp.json()
if data.get("code") != "0":
raise ValueError(f"OKX API error: {data}")
return data.get("data", [])
async def listen(self, callback: Callable):
"""Listen loop với auto-reconnect và heartbeat"""
while True:
try:
if not self._connected:
await self.connect()
async for msg in self._ws:
if msg.type == aiohttp.WSMsgType.PING:
await self._ws.ping()
elif msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
await callback(data)
elif msg.type == aiohttp.WSMsgType.CLOSED:
print("🔌 Connection closed, reconnecting in 5s...")
await asyncio.sleep(5)
self._connected = False
break
elif msg.type == aiohttp.WSMsgType.ERROR:
print(f"❌ WebSocket error: {msg.data}")
await asyncio.sleep(5)
break
except Exception as e:
print(f"❌ Error in listen loop: {e}")
await asyncio.sleep(5)
self._connected = False
Sử dụng
async def main():
client = OKXWebSocketClient(
api_key="YOUR_OKX_API_KEY", # Thay bằng key thật
api_secret="YOUR_OKX_API_SECRET",
passphrase="YOUR_PASSPHRASE",
sandbox=True # Test trên sandbox trước
)
await client.connect()
# Subscribe real-time trades
await client.subscribe("trades", "BTC-USDT-SWAP")
# Callback xử lý message
async def handle_trade(data):
print(f"Trade: {data}")
await client.listen(handle_trade)
if __name__ == "__main__":
asyncio.run(main())
Triển khai bước 2: Chuẩn hóa và lưu trữ dữ liệu backtesting
Sau khi thu thập data, bước tiếp theo là chuẩn hóa format và lưu vào database tối ưu cho query speed. Đội ngũ HolySheep sử dụng TimescaleDB (PostgreSQL extension) để handle time-series data hiệu quả.
from sqlalchemy import create_engine, Column, Float, Integer, BigInteger, DateTime, String, Index
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime, timezone
from typing import List, Optional
import pandas as pd
Base = declarative_base()
class OHLCV(Base):
"""Bảng lưu trữ candlestick data - partition theo ngày"""
__tablename__ = 'ohlcv_okx'
id = Column(BigInteger, primary_key=True, autoincrement=True)
timestamp = Column(DateTime(timezone=True), nullable=False)
inst_id = Column(String(20), nullable=False) # VD: BTC-USDT-SWAP
interval = Column(String(5), nullable=False) # 1m, 5m, 1h, 1d
# OHLCV fields
open = Column(Float, nullable=False)
high = Column(Float, nullable=False)
low = Column(Float, nullable=False)
close = Column(Float, nullable=False)
volume = Column(Float, nullable=False)
quote_volume = Column(Float) # USDT volume
trades_count = Column(Integer) # Số lượng trades trong candle
# Metadata
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
__table_args__ = (
Index('idx_inst_interval_ts', 'inst_id', 'interval', 'timestamp'),
{'postgresql_partition_by': 'RANGE (timestamp)'}
)
class Trade(Base):
"""Bảng tick-level trades - raw data từ WebSocket"""
__tablename__ = 'trades_okx'
id = Column(BigInteger, primary_key=True, autoincrement=True)
trade_id = Column(String(50), unique=True, nullable=False)
timestamp = Column(DateTime(timezone=True), nullable=False)
inst_id = Column(String(20), nullable=False)
price = Column(Float, nullable=False)
size = Column(Float, nullable=False)
side = Column(String(4), nullable=False) # buy / sell
tick_direction = Column(String(10)) # upTick, downTick
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
__table_args__ = (
Index('idx_trade_inst_ts', 'inst_id', 'timestamp'),
Index('idx_trade_timestamp', 'timestamp'),
)
class DatabaseManager:
"""Quản lý kết nối và operations với PostgreSQL/TimescaleDB"""
def __init__(self, connection_string: str):
self.engine = create_engine(
connection_string,
pool_size=20,
max_overflow=30,
pool_pre_ping=True, # Health check connection
echo=False
)
self.Session = sessionmaker(bind=self.engine)
def create_tables(self):
"""Tạo tables với partitioning cho TimescaleDB"""
Base.metadata.create_all(self.engine)
# Tạo hypertables cho TimescaleDB (nếu dùng TimescaleDB)
with self.engine.connect() as conn:
try:
conn.execute("""
SELECT create_hypertable('ohlcv_okx', 'timestamp',
if_not_exists => TRUE,
migrate_data => TRUE);
""")
conn.execute("""
SELECT create_hypertable('trades_okx', 'timestamp',
if_not_exists => TRUE,
migrate_data => TRUE);
""")
conn.commit()
print("✅ TimescaleDB hypertables created")
except Exception as e:
print(f"ℹ️ Regular PostgreSQL tables created (TimescaleDB not available): {e}")
def insert_ohlcv_batch(self, records: List[dict]):
"""Bulk insert OHLCV data với chunking"""
session = self.Session()
try:
# Chunk 1000 records/request để tránh memory spike
chunk_size = 1000
for i in range(0, len(records), chunk_size):
chunk = records[i:i + chunk_size]
objects = [
OHLCV(
timestamp=datetime.fromtimestamp(int(r[0])/1000, tz=timezone.utc),
inst_id=r[8],
interval=r[9] if len(r) > 9 else '1m',
open=float(r[1]),
high=float(r[2]),
low=float(r[3]),
close=float(r[4]),
volume=float(r[5]),
quote_volume=float(r[6]) if r[6] else None,
trades_count=int(r[7]) if r[7] else 0
)
for r in chunk
]
session.bulk_save_objects(objects)
session.commit()
print(f"✅ Inserted {len(records)} OHLCV records")
except Exception as e:
session.rollback()
print(f"❌ Insert failed: {e}")
raise
finally:
session.close()
def insert_trades_batch(self, records: List[dict]):
"""Bulk insert tick trades"""
session = self.Session()
try:
objects = [
Trade(
trade_id=r[3], # tradeId field
timestamp=datetime.fromtimestamp(int(r[0])/1000, tz=timezone.utc),
inst_id=r[8],
price=float(r[1]),
size=float(r[2]),
side=r[4],
tick_direction=r[5] if len(r) > 5 else None
)
for r in records
]
session.bulk_save_objects(objects)
session.commit()
print(f"✅ Inserted {len(records)} trade records")
except Exception as e:
session.rollback()
print(f"❌ Trade insert failed: {e}")
finally:
session.close()
def get_backtest_data(
self,
inst_id: str,
start_time: datetime,
end_time: datetime,
interval: str = '1m'
) -> pd.DataFrame:
"""Query data cho backtesting với index optimization"""
query = f"""
SELECT
timestamp,
open,
high,
low,
close,
volume,
quote_volume,
trades_count
FROM ohlcv_okx
WHERE inst_id = '{inst_id}'
AND interval = '{interval}'
AND timestamp BETWEEN '{start_time.isoformat()}'
AND '{end_time.isoformat()}'
ORDER BY timestamp ASC
"""
return pd.read_sql(query, self.engine, parse_dates=['timestamp'])
def get_data_stats(self, inst_id: str, days: int = 30) -> dict:
"""Thống kê data quality - phát hiện gaps"""
query = f"""
WITH time_gaps AS (
SELECT
timestamp,
lead(timestamp) OVER (ORDER BY timestamp) as next_ts,
EXTRACT(EPOCH FROM (lead(timestamp) OVER (ORDER BY timestamp) - timestamp)) as gap_seconds
FROM ohlcv_okx
WHERE inst_id = '{inst_id}'
AND timestamp > NOW() - INTERVAL '{days} days'
)
SELECT
COUNT(*) as total_candles,
COUNT(CASE WHEN gap_seconds > 120 THEN 1 END) as large_gaps,
AVG(gap_seconds) as avg_gap,
MAX(gap_seconds) as max_gap,
MIN(timestamp) as earliest,
MAX(timestamp) as latest
FROM time_gaps
"""
with self.engine.connect() as conn:
result = conn.execute(query).fetchone()
return {
"total_candles": result[0],
"large_gaps": result[1],
"avg_gap_seconds": round(result[2], 2) if result[2] else 0,
"max_gap_seconds": result[3],
"earliest": result[4],
"latest": result[5]
}
Sử dụng
db = DatabaseManager("postgresql://user:pass@localhost:5432/crypto_data")
db.create_tables()
Query backtest data
df = db.get_backtest_data(
inst_id="BTC-USDT-SWAP",
start_time=datetime(2024, 1, 1, tzinfo=timezone.utc),
end_time=datetime(2024, 3, 1, tzinfo=timezone.utc),
interval='5m'
)
print(f"📊 Loaded {len(df)} candles for backtesting")
Triển khai bước 3: Tích hợp HolySheep AI cho Feature Engineering
Đây là phần quan trọng nhất — sử dụng HolySheep AI để generate features, detect patterns, và tạo signals. Với chi phí chỉ $0.42/MTok cho DeepSeek V3.2 (tiết kiệm 85%+ so với GPT-4.1), bạn có thể chạy hàng triệu inference requests mà không lo về budget.
import httpx
import json
from typing import List, Dict, Optional
from datetime import datetime, timezone
import asyncio
from dataclasses import dataclass
import pandas as pd
@dataclass
class FeatureConfig:
"""Cấu hình feature generation"""
indicators: List[str] # ['RSI', 'MACD', 'BollingerBands', ...]
lookback_periods: List[int] # [7, 14, 30, 60]
pattern_threshold: float = 0.7
class HolySheepAIClient:
"""
HolySheep AI Client cho feature generation và signal analysis.
base_url: https://api.holysheep.ai/v1
Chi phí: DeepSeek V3.2 chỉ $0.42/MTok (85% tiết kiệm so với GPT-4.1 $8)
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(60.0),
limits=httpx.Limits(max_connections=50, max_keepalive_connections=20)
)
# Pricing tham khảo (cập nhật 2026)
self.pricing = {
"gpt-4.1": 8.0, # $8/MTok
"claude-sonnet-4.5": 15.0, # $15/MTok
"gemini-2.5-flash": 2.50, # $2.50/MTok
"deepseek-v3.2": 0.42 # $0.42/MTok ✅ Recommend
}
async def analyze_market_regime(
self,
ohlcv_data: pd.DataFrame,
timeframe: str = "1h"
) -> Dict:
"""
Sử dụng AI để phân tích market regime và đưa ra trading recommendations.
Sử dụng DeepSeek V3.2 cho cost-efficiency với output chất lượng cao.
"""
# Tóm tắt dữ liệu thành text format
recent_closes = ohlcv_data['close'].tail(100).tolist()
recent_volumes = ohlcv_data['volume'].tail(100).tolist()
summary_prompt = f"""
Bạn là chuyên gia phân tích thị trường crypto. Phân tích dữ liệu sau và trả lời JSON format:
Data Summary (last 100 candles, {timeframe} timeframe):
- Close prices (last 10): {recent_closes[-10:]}
- Volume trend: {'Increasing' if recent_volumes[-1] > sum(recent_volumes[-20:])/20 else 'Decreasing'}
- Price change 24h: {((recent_closes[-1] - recent_closes[-24])/recent_closes[-24]*100):.2f}% nếu có đủ data
Hãy trả về JSON với structure:
{{
"regime": "trending_up|trending_down|ranging|volatile",
"confidence": 0.0-1.0,
"signal": "bullish|bearish|neutral",
"support_levels": [list of prices],
"resistance_levels": [list of prices],
"risk_assessment": "low|medium|high",
"reasoning": "explanation in Vietnamese"
}}
Chỉ trả về JSON, không giải thích gì thêm.
"""
response = await self._call_llm(
model="deepseek-v3.2", # ✅ Recommend: $0.42/MTok
messages=[{"role": "user", "content": summary_prompt}],
temperature=0.3
)
return json.loads(response)
async def generate_trading_signals(
self,
features: pd.DataFrame,
symbols: List[str],
strategy_type: str = "mean_reversion"
) -> List[Dict]:
"""
Generate trading signals dựa trên technical features.
Sử dụng batch processing để tối ưu token usage.
"""
batch_prompt = f"""
Bạn là quantitative analyst. Tạo trading signals cho portfolio:
Strategy Type: {strategy_type}
Symbols: {', '.join(symbols)}
Features Data (sample):
{features.head(5).to_string()}
Trả về JSON array với format:
[
{{
"symbol": "BTC-USDT",
"action": "BUY|SELL|HOLD",
"position_size": 0.0-1.0,
"entry_price": null,
"stop_loss": null,
"take_profit": null,
"confidence": 0.0-1.0,
"reasoning": "explanation"
}}
]
Chỉ trả về JSON array.
"""
response = await self._call_llm(
model="deepseek-v3.2",
messages=[{"role": "user", "content": batch_prompt}],
temperature=0.2
)
return json.loads(response)
async def backtest_strategy_explanation(
self,
backtest_results: Dict,
initial_capital: float = 10000
) -> str:
"""
Giải thích kết quả backtest bằng ngôn ngữ tự nhiên.
Sử dụng Gemini 2.5 Flash cho creative tasks: $2.50/MTok
"""
prompt = f"""
Phân tích kết quả backtest sau và đưa ra insights:
Initial Capital: ${initial_capital:,.2f}
Final Capital: ${backtest_results.get('final_capital', 0):,.2f}
Total Return: {backtest_results.get('total_return', 0):.2f}%
Max Drawdown: {backtest_results.get('max_drawdown', 0):.2f}%
Sharpe Ratio: {backtest_results.get('sharpe_ratio', 0):.2f}
Win Rate: {backtest_results.get('win_rate', 0):.2f}%
Total Trades: {backtest_results.get('total_trades', 0)}
Viết báo cáo executive summary ngắn gọn (200-300 từ) bằng tiếng Việt, bao gồm:
1. Tổng quan hiệu suất
2. Điểm mạnh và điểm yếu của chiến lược
3. Khuyến nghị cải thiện
"""
response = await self._call_llm(
model="gemini-2.5-flash", # ✅ Tốt cho creative tasks
messages=[{"role": "user", "content": prompt}],
temperature=0.5
)
return response
async def _call_llm(
self,
model: str,
messages: List[Dict],
temperature: float = 0.3
) -> str:
"""Internal method để call HolySheep AI API"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": 2048
}
response = await self.client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
if response.status_code != 200:
raise ValueError(f"HolySheep API error: {response.status_code} - {response.text}")
result = response.json()
return result["choices"][0]["message"]["content"]
def estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""Ước tính chi phí cho request"""
price_per_mtok = self.pricing.get(model, 8.0)
total_tokens_mtok = (input_tokens + output_tokens) / 1_000_000
return total_tokens_mtok * price_per_mtok
async def close(self):
await self.client.aclose()
============================================
BACKTESTING PIPELINE VỚI HOLYSHEEP AI
============================================
class BacktestPipeline:
"""Pipeline hoàn chỉnh cho backtesting với AI enhancement"""
def __init__(
self,
db_manager: DatabaseManager,
ai_client: HolySheepAIClient
):
self.db = db_manager
self.ai = ai_client
async def run_backtest(
self,
symbol: str,
start_date: datetime,
end_date: datetime,
strategy_config: Dict
) -> Dict:
"""
Chạy backtest với các bước:
1. Load historical data
2. Generate technical features
3. Get AI signals
4. Simulate trades
5. Calculate metrics
"""
print(f"🚀 Starting backtest for {symbol}")
print(f" Period: {start_date.date()} to {end_date.date()}")
# Step 1: Load data
df = self.db.get_backtest_data(
inst_id=symbol,
start_time=start_date,
end_time=end_date,
interval=strategy_config.get('interval', '5m')
)
if df.empty:
raise ValueError(f"No data found for {symbol}")
print(f" 📊 Loaded {len(df)} candles")
# Step 2: Generate features
df = self._calculate_features(df)
# Step 3: Get AI market analysis (chunk data để tiết kiệm cost)
market_analysis = await self.ai.analyze_market_regime(df)
print(f" 🧠 Market regime: {market_analysis.get('regime')}")
# Step 4: Generate signals
signals = await self.ai.generate_trading_signals(
features=df.tail(100),
symbols=[symbol],
strategy_type=strategy_config.get('type', 'momentum')
)
# Step 5: Simulate trades
trades = self._simulate_trades(df, signals)
# Step 6: Calculate metrics
metrics = self._calculate_metrics(trades, strategy_config.get('initial_capital', 10000))
# Step 7: Generate explanation
explanation = await self.ai.backtest_strategy_explanation(
metrics,
initial_capital=strategy_config.get('initial_capital', 10000)
)
metrics['analysis'] = explanation
return metrics
def _calculate_features(self, df: pd.DataFrame) -> pd.DataFrame:
"""Tính technical indicators"""
# Simple Moving Averages
df['sma_20'] = df['close'].rolling(20).mean()
df['sma_50'] = df['close'].rolling(50).mean()
# RSI
delta = df['close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(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['macd_signal'] = df['macd'].ewm(span=9, adjust=False).mean()
# Bollinger Bands
df['bb_middle'] = df['close'].rolling(20).mean()
df['bb_std'] = df['close'].rolling(20).std()
df['bb_upper'] = df['bb_middle'] + (df['bb_std'] * 2)
df['bb_lower'] = df['bb_middle'] - (df['bb_std'] * 2)
return df.dropna()
def _simulate_trades(self, df: pd.DataFrame, signals: List[Dict]) -> List[Dict]:
"""Simulate trades dựa trên signals"""
trades = []
position = None
for i, signal in enumerate(signals):
action = signal.get('action')
if action == 'BUY' and position is None:
position = {
'entry_time': df.iloc[i]['timestamp'],
'entry_price': df.iloc[i]['close'],
'size': signal.get('position_size', 0.1)
}
elif action == 'SELL' and position is not None:
trades.append({
'entry_time': position['entry_time'],
'entry_price': position['entry_price'],
'exit_time': df.iloc[i]['timestamp'],
'exit_price': df.iloc[i]['close'],
'pnl': (df.iloc[i]['close'] - position['entry_price']) * position['size'],
'pnl_pct': (df.iloc[i]['close'] - position['entry_price']) / position['entry_price']
})
position = None
return trades
def _calculate_metrics(self, trades: List[Dict], initial_capital: float) -> Dict:
"""Calculate backtest performance metrics"""