Trong thị trường crypto, dữ liệu liquidation là vàng - nó phản ánh tâm lý đòn bẩy, điểm uốn thị trường và cơ hội alpha. Tuy nhiên, việc tái tạo chính xác từng sự kiện强平 từ dữ liệu thô gặp nhiều thách thức về latency, độ chính xác tick-by-tick và chi phí API. Bài viết này sẽ hướng dẫn bạn xây dựng pipeline hoàn chỉnh để thu thập, tái tạo và khai thác yếu tố từ dữ liệu liquidation.
Bảng so sánh: HolySheep vs API chính thức vs Dịch vụ Relay
| Tiêu chí | HolySheep AI | API chính thức (Binance) | Dịch vụ Relay khác | |
|---|---|---|---|---|
| Chi phí/1M token | $0.42 - $15 | $2.50 - $60 | $1.50 - $25 | Tiết kiệm |
| Độ trễ trung bình | <50ms | 80-200ms | 100-300ms | |
| Tỷ giá | ¥1 = $1 | Tỷ giá thị trường | Tỷ giá thị trường | |
| Hỗ trợ thanh toán | WeChat/Alipay/Thẻ | Chỉ thẻ quốc tế | Hạn chế | |
| Tín dụng miễn phí | ✅ Có khi đăng ký | ❌ Không | ❌ Không | |
| Stream liquidation | ✅ Real-time WebSocket | ✅ Có | ⚠️ Hạn chế |
Kết luận nhanh: Đăng ký tại đây HolySheep AI cung cấp chi phí thấp nhất với latency tốt nhất, đặc biệt phù hợp cho việc xây dựng hệ thống phân tích liquidation real-time.
Giới thiệu về Dữ liệu Liquidation trong Crypto
Dữ liệu liquidation (强平) là thông tin về các vị thế bị thanh lý cưỡng bức khi margin không đủ. Đây là chỉ báo quan trọng vì:
- Tâm lý thị trường: Số lượng liquidation lớn cho thấy đà giảm mạnh hoặc FOMO quá mức
- Điểm uốn: Liquidation cliff thường là vùng hỗ trợ/kháng cự quan trọng
- Volatility clustering: Liquidation cascade tạo ra biến động theo cụm
- Funding rate signal: Liquidation data giúp xác nhận funding rate divergence
Pipeline Xây dựng Hệ thống Thu thập Liquidation
1. Kết nối WebSocket Real-time
Đầu tiên, ta cần thiết lập kết nối WebSocket để nhận liquidation event theo thời gian thực. Sử dụng HolySheep AI với base_url được cấu hình đúng:
#!/usr/bin/env python3
"""
Liquidation Stream Collector - Sử dụng HolySheep AI WebSocket
Author: HolySheep AI Technical Team
"""
import asyncio
import json
import websockets
from datetime import datetime
from typing import List, Dict, Optional
import sqlite3
import hashlib
class LiquidationCollector:
"""
Bộ thu thập dữ liệu liquidation real-time
- Kết nối WebSocket đến exchange
- Parse liquidation events
- Lưu vào SQLite database
"""
def __init__(self, db_path: str = "liquidation.db"):
self.db_path = db_path
self.connection = None
self.liquidation_cache: List[Dict] = []
self.batch_size = 100
self.last_flush = datetime.now()
def init_database(self):
"""Khởi tạo bảng SQLite cho liquidation data"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS liquidations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
event_id TEXT UNIQUE NOT NULL,
symbol TEXT NOT NULL,
side TEXT NOT NULL,
price REAL NOT NULL,
quantity REAL NOT NULL,
quote_quantity REAL NOT NULL,
time_exchange INTEGER NOT NULL,
time_received INTEGER NOT NULL,
latency_ms REAL,
is_adverse_fill INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_symbol_time
ON liquidations(symbol, time_exchange)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_event_id
ON liquidations(event_id)
""")
conn.commit()
return conn
def generate_event_id(self, symbol: str, price: float,
quantity: float, timestamp: int) -> str:
"""Tạo event_id duy nhất để deduplicate"""
raw = f"{symbol}:{price}:{quantity}:{timestamp}"
return hashlib.sha256(raw.encode()).hexdigest()[:16]
async def process_liquidation(self, data: Dict):
"""Xử lý một liquidation event"""
try:
event = {
'event_id': self.generate_event_id(
data['symbol'],
float(data['price']),
float(data['quantity']),
data['time']
),
'symbol': data['symbol'].replace('USDT', ''),
'side': 'LONG' if data['side'] == 'BUY' else 'SHORT',
'price': float(data['price']),
'quantity': float(data['quantity']),
'quote_quantity': float(data['price']) * float(data['quantity']),
'time_exchange': data['time'],
'time_received': int(datetime.now().timestamp() * 1000),
'is_adverse_fill': 1 if data.get('type') == 'ADVERSE' else 0
}
event['latency_ms'] = event['time_received'] - event['time_exchange']
self.liquidation_cache.append(event)
# Flush khi đủ batch_size hoặc sau 1 giây
if (len(self.liquidation_cache) >= self.batch_size or
(datetime.now() - self.last_flush).total_seconds() > 1):
await self.flush_cache()
except Exception as e:
print(f"Error processing liquidation: {e}")
async def flush_cache(self):
"""Đẩy cache vào database"""
if not self.liquidation_cache:
return
conn = self.init_database()
cursor = conn.cursor()
cursor.executemany("""
INSERT OR IGNORE INTO liquidations
(event_id, symbol, side, price, quantity, quote_quantity,
time_exchange, time_received, latency_ms, is_adverse_fill)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", [(e['event_id'], e['symbol'], e['side'], e['price'],
e['quantity'], e['quote_quantity'], e['time_exchange'],
e['time_received'], e['latency_ms'], e['is_adverse_fill'])
for e in self.liquidation_cache])
conn.commit()
conn.close()
print(f"Flushed {len(self.liquidation_cache)} liquidations")
self.liquidation_cache = []
self.last_flush = datetime.now()
async def run(self, symbols: List[str] = None):
"""Chạy collector với HolySheep AI WebSocket"""
symbols = symbols or ['BTCUSDT', 'ETHUSDT']
conn = self.init_database()
try:
# Kết nối WebSocket - thay thế bằng endpoint HolySheep
ws_url = "wss://stream.holysheep.ai/ws/liquidation"
async with websockets.connect(ws_url) as ws:
# Subscribe to symbols
subscribe_msg = {
'method': 'SUBSCRIBE',
'params': [f"{s}@liquidation" for s in symbols],
'id': 1
}
await ws.send(json.dumps(subscribe_msg))
print(f"Connected to HolySheep WS, subscribed to {symbols}")
async for message in ws:
data = json.loads(message)
if 'data' in data:
for liquidation in data['data']:
await self.process_liquidation(liquidation)
except Exception as e:
print(f"WebSocket error: {e}")
finally:
await self.flush_cache()
conn.close()
Chạy collector
if __name__ == "__main__":
collector = LiquidationCollector("liquidation_realtime.db")
asyncio.run(collector.run(['BTCUSDT', 'ETHUSDT', 'SOLUSDT']))2. Tái tạo Liquidation Events từ Raw Ticker Data
Đôi khi WebSocket liquidation không có sẵn, ta cần tái tạo từ dữ liệu ticker và trade. Thuật toán sau sử dụng HolySheep AI API để xử lý:
#!/usr/bin/env python3
"""
Liquidation Event Reconstruction Engine
Tái tạo từng sự kiện强平 từ OHLCV + Trade data
"""
import asyncio
import aiohttp
import json
from datetime import datetime, timedelta
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass
import numpy as np
@dataclass
class LiquidationEvent:
"""Cấu trúc một liquidation event"""
symbol: str
timestamp: int
price: float
quantity: float
side: str # LONG or SHORT
side liquidation_price: float
leverage: int
is_auto_liquidate: bool
reconstructed: bool = False
confidence: float = 0.0
class LiquidationReconstructor:
"""
Engine tái tạo liquidation events từ historical data
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session: Optional[aiohttp.ClientSession] = None
self.cache = {}
async def get_session(self) -> aiohttp.ClientSession:
if self.session is None:
self.session = aiohttp.ClientSession(
headers={"Authorization": f"Bearer {self.api_key}"}
)
return self.session
async def analyze_price_impact(self, trades: List[Dict]) -> Dict:
"""
Phân tích impact của trades lên giá
Dùng HolySheep AI để xử lý pattern recognition
"""
session = await self.get_session()
# Chuẩn bị prompt cho AI
prompt = f"""
Analyze these trades for liquidation cascade patterns:
- Count buy vs sell orders
- Calculate volume-weighted price impact
- Identify rapid price movements >2% in 1 minute
- Detect adverse selection patterns
Trades sample:
{json.dumps(trades[:50], indent=2)}
Return JSON with:
- has_liquidation_cascade: bool
- cascade_direction: "up" | "down" | "none"
- impact_percentage: float
- confidence_score: float
"""
payload = {
"model": "deepseek-v3.2", # Model rẻ nhất, phù hợp cho data analysis
"messages": [
{"role": "system", "content": "You are a crypto liquidation analyst."},
{"role": "user", "content": prompt}
],
"temperature": 0.1,
"max_tokens": 500
}
async with session.post(
f"{self.BASE_URL}/chat/completions",
json=payload
) as resp:
result = await resp.json()
if 'choices' in result:
content = result['choices'][0]['message']['content']
return json.loads(content)
return {"has_liquidation_cascade": False}
async def reconstruct_from_funding(self, symbol: str,
start_ts: int,
end_ts: int) -> List[LiquidationEvent]:
"""
Tái tạo liquidation dựa trên funding rate và price
Funding rate tăng đột ngột → có thể có liquidation cascade
"""
events = []
# Lấy funding rate history
funding_data = await self.get_funding_rate(symbol, start_ts, end_ts)
for i, funding in enumerate(funding_data):
if abs(funding['rate']) > 0.001: # Funding > 0.1%
# Tính toán liquidation zones
price_at_funding = funding['price']
# Estimate liquidation zones dựa trên funding
if funding['rate'] > 0:
# Positive funding = long pay short → potential short squeeze
liq_zone = price_at_funding * (1 + funding['rate'] * 10)
else:
# Negative funding = short pay long → potential long cascade
liq_zone = price_at_funding * (1 + funding['rate'] * 10)
event = LiquidationEvent(
symbol=symbol,
timestamp=funding['timestamp'],
price=liq_zone,
quantity=0, # Cần trade data để estimate
side='SHORT' if funding['rate'] > 0 else 'LONG',
liquidation_price=liq_zone,
leverage=20, # Assumed average leverage
is_auto_liquidate=True,
reconstructed=True,
confidence=0.6
)
events.append(event)
return events
async def get_funding_rate(self, symbol: str,
start_ts: int,
end_ts: int) -> List[Dict]:
"""Lấy funding rate history"""
session = await self.get_session()
# Demo endpoint - thực tế cần endpoint đúng
async with session.get(
f"{self.BASE_URL}/funding",
params={
"symbol": symbol,
"startTime": start_ts,
"endTime": end_ts
}
) as resp:
return await resp.json()
def calculate_liquidation_price(self, entry_price: float,
leverage: int,
side: str,
maintenance_margin: float = 0.005) -> float:
"""
Tính liquidation price
Args:
entry_price: Giá vào lệnh
leverage: Đòn bẩy (1-125)
side: LONG hoặc SHORT
maintenance_margin: Tỷ lệ margin bảo trì (0.5%)
Returns:
Liquidation price
"""
if side == 'LONG':
# Liquidation price cho LONG position
liq_price = entry_price * (1 - (1 / leverage) + maintenance_margin)
else:
# Liquidation price cho SHORT position
liq_price = entry_price * (1 + (1 / leverage) - maintenance_margin)
return liq_price
async def run_backfill(self, symbol: str,
start_date: datetime,
end_date: datetime) -> List[LiquidationEvent]:
"""
Chạy backfill để tái tạo liquidation history
"""
all_events = []
current_date = start_date
while current_date <= end_date:
start_ts = int(current_date.timestamp() * 1000)
end_ts = int((current_date + timedelta(days=1)).timestamp() * 1000)
# Tái tạo từ funding
funding_events = await self.reconstruct_from_funding(
symbol, start_ts, end_ts
)
all_events.extend(funding_events)
current_date += timedelta(days=1)
print(f"Reconstructed {len(all_events)} potential liquidation events")
return all_events
Sử dụng
async def main():
reconstructor = LiquidationReconstructor(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
events = await reconstructor.run_backfill(
symbol="BTCUSDT",
start_date=datetime(2024, 1, 1),
end_date=datetime(2024, 1, 7)
)
# Phân tích cluster
print(f"Total events: {len(events)}")
for event in events[:10]:
print(f"{event.symbol} @ {event.price} - {event.side}")
if __name__ == "__main__":
asyncio.run(main())Chiến lược Factor Mining từ Liquidation Data
Sau khi thu thập đủ dữ liệu, bước tiếp theo là khai thác các yếu tố (factors) có giá trị dự báo:
#!/usr/bin/env python3
"""
Liquidation Factor Mining Engine
Khai thác factors từ dữ liệu liquidation history
"""
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Tuple
import sqlite3
class LiquidationFactorEngine:
"""
Engine khai thác factors từ liquidation data
"""
def __init__(self, db_path: str):
self.db_path = db_path
def load_data(self, symbol: str,
start_date: datetime,
end_date: datetime) -> pd.DataFrame:
"""Load liquidation data từ database"""
conn = sqlite3.connect(self.db_path)
query = """
SELECT
symbol,
side,
price,
quantity,
quote_quantity,
time_exchange,
latency_ms,
is_adverse_fill,
created_at
FROM liquidations
WHERE symbol = ?
AND created_at BETWEEN ? AND ?
ORDER BY time_exchange
"""
df = pd.read_sql_query(
query,
conn,
params=(
symbol,
start_date.strftime('%Y-%m-%d'),
end_date.strftime('%Y-%m-%d')
)
)
conn.close()
if not df.empty:
df['created_at'] = pd.to_datetime(df['created_at'])
df['time_exchange'] = pd.to_datetime(
df['time_exchange'], unit='ms'
)
return df
def calculate_factors(self, df: pd.DataFrame,
window: str = '1H') -> pd.DataFrame:
"""
Tính toán các liquidation factors
Factors được tính:
1. Liquidation Volume Ratio (LVR)
2. Long/Short Imbalance (LSI)
3. Liquidation Cluster Density (LCD)
4. Cascade Probability Score (CPS)
5. Adverse Selection Ratio (ASR)
"""
if df.empty:
return pd.DataFrame()
# Resample theo window
df.set_index('created_at', inplace=True)
# Factor 1: Liquidation Volume Ratio
# Tổng USD liquidation / tổng volume giao dịch
lvr = df['quote_quantity'].resample(window).sum() / \
df['quote_quantity'].resample(window).sum().shift(1)
lvr = lvr.fillna(0).clip(0, 10) # Cap extreme values
# Factor 2: Long/Short Imbalance
# (Long liq - Short liq) / (Long liq + Short liq)
long_liq = df[df['side'] == 'LONG']['quote_quantity'].resample(window).sum()
short_liq = df[df['side'] == 'SHORT']['quote_quantity'].resample(window).sum()
lsi = (long_liq - short_liq) / (long_liq + short_liq + 1e-10)
lsi = lsi.fillna(0).clip(-1, 1)
# Factor 3: Liquidation Cluster Density
# Số liquidation events / thời gian trong window
event_count = df['quote_quantity'].resample(window).count()
lcd = event_count / event_count.rolling(24).mean() # So với MA 24 windows
lcd = lcd.fillna(1).clip(0, 10)
# Factor 4: Cascade Probability Score
# Xác suất cascade dựa trên:
# - Tốc độ tăng liquidation
# - Concentration của liquidation size
# - Adverse fill ratio
lvr_change = lvr.pct_change().clip(-1, 1)
concentration = df.groupby(df.index)['quote_quantity'].apply(
lambda x: x.max() / (x.sum() + 1e-10)
).resample(window).mean()
concentration = concentration.fillna(0).clip(0, 1)
adverse_ratio = df['is_adverse_fill'].resample(window).mean()
adverse_ratio = adverse_ratio.fillna(0)
# CPS = f(LVR_change, concentration, adverse_ratio)
cps = (
0.4 * lvr_change.abs() +
0.3 * concentration +
0.3 * adverse_ratio
)
cps = cps.fillna(0).clip(0, 1)
# Factor 5: Average Latency Spike
# Latency tăng đột ngột = congestion = potential cascade
avg_latency = df['latency_ms'].resample(window).mean()
latency_std = df['latency_ms'].resample(window).std()
latency_spike = (avg_latency - avg_latency.rolling(24).mean()) / \
(latency_std + 1)
latency_spike = latency_spike.fillna(0).clip(-3, 3) / 3 # Normalize
# Tổng hợp factors
factors = pd.DataFrame({
'LVR': lvr,
'LSI': lsi,
'LCD': lcd,
'CPS': cps,
'LATENCY_SPIKE': latency_spike,
'LONG_LIQ': long_liq,
'SHORT_LIQ': short_liq,
'EVENT_COUNT': event_count
})
return factors
def detect_liquidation_zones(self, df: pd.DataFrame,
price_data: pd.DataFrame,
threshold: float = 1000000) -> List[Dict]:
"""
Phát hiện các vùng liquidation zones quan trọng
Vùng có tổng liquidation > threshold USD
"""
zones = []
# Nhóm theo price bins
df['price_bin'] = pd.cut(df['price'], bins=50)
zone_summary = df.groupby(['price_bin', 'side'])['quote_quantity'].sum()
for (price_range, side), total_liq in zone_summary.items():
if total_liq >= threshold:
zone = {
'price_min': price_range.left,
'price_max': price_range.right,
'side': side,
'total_liquidation': total_liq,
'event_count': len(df[
(df['price_bin'] == price_range) &
(df['side'] == side)
])
}
zones.append(zone)
# Sắp xếp theo total liquidation
zones.sort(key=lambda x: x['total_liquidation'], reverse=True)
return zones
def calculate_liquidation_heatmap(self, df: pd.DataFrame,
bins: int = 100) -> np.ndarray:
"""
Tạo heatmap 2D: Price vs Time của liquidation
Dùng cho visualization
"""
if df.empty:
return np.zeros((bins, bins))
# Price bins
price_min, price_max = df['price'].min(), df['price'].max()
df['price_bin_idx'] = pd.cut(
df['price'],
bins=bins,
labels=range(bins)
).astype(int)
# Time bins (theo giờ)
df['hour'] = df['created_at'].dt.hour
df['date'] = df['created_at'].dt.date
# Tạo heatmap
heatmap = df.pivot_table(
values='quote_quantity',
index='price_bin_idx',
columns='date',
aggfunc='sum',
fill_value=0
).values
return heatmap
def generate_signals(self, factors: pd.DataFrame,
lvr_threshold: float = 2.0,
lsi_threshold: float = 0.7,
cps_threshold: float = 0.6) -> pd.DataFrame:
"""
Generate trading signals từ factors
Signal types:
- LONG_SQUEEZE: LSI < -threshold AND LVR > threshold
- SHORT_SQUEEZE: LSI > threshold AND LVR > threshold
- CASCADE_RISK: CPS > threshold
- LIQUIDATION_CLUSTER: LCD > 3
"""
signals = pd.DataFrame(index=factors.index)
signals['LONG_SQUEEZE'] = (
(factors['LSI'] < -lsi_threshold) &
(factors['LVR'] > lvr_threshold)
).astype(int)
signals['SHORT_SQUEEZE'] = (
(factors['LSI'] > lsi_threshold) &
(factors['LVR'] > lvr_threshold)
).astype(int)
signals['CASCADE_RISK'] = (factors['CPS'] > cps_threshold).astype(int)
signals['CLUSTER'] = (factors['LCD'] > 3).astype(int)
# Combined signal strength
signals['STRENGTH'] = (
signals['LONG_SQUEEZE'] * -1 +
signals['SHORT_SQUEEZE'] * 1 +
signals['CASCADE_RISK'] * 0.5 +
signals['CLUSTER'] * 0.3
)
return signals
Sử dụng
if __name__ == "__main__":
engine = LiquidationFactorEngine("liquidation.db")
df = engine.load_data(
symbol="BTCUSDT",
start_date=datetime(2024, 1, 1),
end_date=datetime(2024, 1, 7)
)
factors = engine.calculate_factors(df, window='1H')
signals = engine.generate_signals(factors)
print("Factor Summary:")
print(factors.tail(20))
print("\nSignal Summary:")
print(signals.tail(20))Ứng dụng Thực tế: Market Regime Detection
Kết hợp liquidation factors với các chỉ báo khác để phát hiện market regimes:
#!/usr/bin/env python3
"""
Market Regime Detector sử dụng Liquidation Factors
"""
import pandas as pd
import numpy as np
from typing import Dict, List
from enum import Enum
class MarketRegime(Enum):
NORMAL = "normal"
HIGH_VOL = "high_volatility"
LONG_SQUEEZE = "long_squeeze"
SHORT_SQUEEZE = "short_squeeze"
CASCADE = "cascade"
ACCUMULATION = "accumulation"
class RegimeDetector:
"""
Phát hiện market regime dựa trên liquidation data
"""
def __init__(self):
self.lookback_windows = [6, 24, 72] # hours
def detect_regime(self, factors: pd.DataFrame) -> pd.DataFrame:
"""
Detect market regime từ factors
"""
regimes = pd.DataFrame(index=factors.index)
# Rolling statistics
for window in self.lookback_windows:
factors[f'LVR_MA{window}'] = factors['LVR'].rolling(window).mean()
factors[f'LSI_MA{window}'] = factors['LSI'].rolling(window).mean()
factors[f'CPS_MA{window}'] = factors['CPS'].rolling(window).mean()
# Regime detection logic
conditions = {
MarketRegime.LONG_SQUEEZE: (
(factors['LSI'] < -0.6) &
(factors['LVR'] > factors['LVR_MA6'] * 1.5) &
(factors['CPS'] > 0.5)
),
MarketRegime.SHORT_SQUEEZE: (
(factors['LSI'] > 0.6) &
(factors['LVR'] > factors['LVR_MA6'] * 1.5) &
(factors['CPS'] > 0.5)
),
MarketRegime.CASCADE: (
(factors['CPS'] > 0.7) &
(factors['LCD'] > 3) &
(factors['LATENCY_SPIKE'] > 0.5)
),
MarketRegime.HIGH_VOL: (
(factors['LVR_MA24'] > 2) &
(factors['EVENT_COUNT'] > factors['EVENT_COUNT'].quantile(0.75))
),
MarketRegime.ACCUMULATION: (
(factors['LSI'].abs() < 0.2) &
(factors['CPS'] < 0.3) &
(factors['LVR'] < 1.5)
)
}
for regime, condition in conditions.items():
regimes[regime.value] = condition.astype(int)
# Default to NORMAL if no specific regime detected
regimes[MarketRegime.NORMAL.value] = (
regimes.sum(axis=1) == 0
).astype(int)
# Final regime assignment
regimes['REGIME'] = regimes.idxmax(axis=1)
return regimes
def backtest_regime_strategy(self,
regimes: pd.DataFrame,
returns: pd.Series,
position_size: float = 1.0) -> pd.DataFrame:
"""
Backtest chiến lược giao dịch dựa trên regime
"""
trades = pd.DataFrame(index=regimes.index)
# Long khi có short squeeze (muốn bắt reversal)
trades['signal'] = 0
trades.loc[regimes['short_squeeze'] == 1, 'signal'] = 1
trades.loc[regimes['long_squeeze'] == 1, 'signal'] = -1
# Calculate returns
trades['returns'] = returns
trades['strategy_returns'] = trades['signal'].shift(1) * trades['returns']
# Performance metrics
total_return = (1 + trades['strategy_returns']).prod() - 1
sharpe = trades['strategy_returns'].mean() / \
trades['strategy_returns'].std() * np.sqrt(365)
max_dd = (trades['strategy_returns'].cumsum() -
trades['strategy_returns'].cumsum().cummax()).min()
return {
'total_return': total_return,
'sharpe_ratio': sharpe,
'max_drawdown': max_dd,
'trade_count': (trades['signal'].diff() != 0).sum()
}
Demo usage
if __name__ == "__main__":
detector = RegimeDetector()
# Load factors từ example trước
# regimes = detector.detect_regime(factors)
# stats = detector.backtest_regime_strategy(regimes, price_returns)
# print(stats)Phù hợp / Không phù hợp với ai
| ✅ PHÙ HỢP với | ❌ KHÔNG PHÙ HỢP với |
|---|---|
|
Market makers & Liquidity providers Cần real-time liquidation data để đặt spread chính xác |
Người mới bắt đầu Cần kiến thức nền tảng về derivatives trước |