Trong thị trường crypto 24/7 với hàng trăm sàn giao dịch, statistical arbitrage là một trong những chiến lược low-latency mang lại lợi nhuận ổn định nhất. Bài viết này tôi sẽ chia sẻ kinh nghiệm xây dựng data pipeline và feature engineering cho statistical arbitrage từ góc nhìn của một kỹ sư đã vận hành hệ thống thực chiến với volume giao dịch 7 con số mỗi tháng.
Tổng Quan Kiến Trúc Statistical Arbitrage
Statistical arbitrage crypto hoạt động dựa trên nguyên lý: khi chênh lệch giá giữa các cặp giao dịch (spread) deviated khỏi historical mean, ta kỳ vọng spread sẽ revert về mean. Hệ thống production-grade cần đáp ứng:
- Latency sub-100ms cho signal generation
- Data consistency cross-exchange (bid-ask spread, order book depth)
- Feature store real-time với sliding window
- Risk management dynamic position sizing
# Kiến trúc tổng quan Statistical Arbitrage Pipeline
┌─────────────┐ ┌──────────────┐ ┌───────────────┐
│ Exchanges │───▶│ Data Ingestion│───▶│ Feature Store │
│ (WebSocket)│ │ (Rust/Python)│ │ (Redis/Polars)│
└─────────────┘ └──────────────┘ └───────┬───────┘
│
┌──────────────┐ │
│ Strategy │◀─────────┘
│ Engine │
└───────┬──────┘
│
┌───────────────┼───────────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Order │ │ Risk │ │ Analytics│
│ Router │ │ Engine │ │ Dashboard│
└──────────┘ └──────────┘ └──────────┘
Data Acquisition: Multi-Exchange WebSocket Architecture
Trong statistical arbitrage, chất lượng data quyết định 80% hiệu suất strategy. Tôi đã thử nghiệm với nhiều phương án và kết luận: WebSocket streaming là lựa chọn tối ưu cho real-time data với latency dưới 50ms.
# Production-grade WebSocket data collector
Sử dụng asyncio cho high-throughput concurrent connections
import asyncio
import websockets
import json
import msgpack
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from collections import deque
import time
from datetime import datetime
import aiohttp
import redis.asyncio as redis
@dataclass
class OrderBookSnapshot:
exchange: str
symbol: str
bids: List[tuple] # [(price, volume)]
asks: List[tuple]
timestamp: int
local_timestamp: float = field(default_factory=time.time)
@property
def mid_price(self) -> float:
return (self.bids[0][0] + self.asks[0][0]) / 2
@property
def spread(self) -> float:
return self.asks[0][0] - self.bids[0][0]
@property
def spread_bps(self) -> float:
return (self.spread / self.mid_price) * 10000
class MultiExchangeCollector:
"""Real-time data collection từ multiple exchanges"""
WS_ENDPOINTS = {
'binance': 'wss://stream.binance.com:9443/ws',
'okx': 'wss://ws.okx.com:8443/ws/v5/public',
'bybit': 'wss://stream.bybit.com/v5/public/spot',
'gate': 'wss://api.gateio.ws/ws/v4/',
'kucoin': 'wss://ws-api.kucoin.com'
}
def __init__(self, redis_client: redis.Redis, pairs: List[str]):
self.redis = redis_client
self.pairs = pairs
self.orderbooks: Dict[str, Dict[str, OrderBookSnapshot]] = {}
self.subscriptions = {}
self.latencies = deque(maxlen=1000)
async def subscribe_binance(self, websocket):
"""Subscribe Binance order book stream"""
symbols = [p.replace('/', '').lower() for p in self.pairs]
subscribe_msg = {
"method": "SUBSCRIBE",
"params": [f"{s}@depth20@100ms" for s in symbols],
"id": 1
}
await websocket.send(json.dumps(subscribe_msg))
async def collect_binance(self):
"""Binance order book collector với latency tracking"""
while True:
try:
async with websockets.connect(
self.WS_ENDPOINTS['binance'],
ping_interval=20
) as ws:
await self.subscribe_binance(ws)
async for raw_msg in ws:
recv_time = time.time()
data = json.loads(raw_msg)
if 'data' not in data:
continue
book_data = data['data']
symbol = book_data['s']
exchange = 'binance'
snapshot = OrderBookSnapshot(
exchange=exchange,
symbol=symbol,
bids=[(float(p), float(q)) for p, q in book_data['b'][:20]],
asks=[(float(p), float(q)) for p, q in book_data['a'][:20]],
timestamp=book_data['E']
)
# Calculate exchange latency
exchange_latency = (recv_time - snapshot.timestamp / 1000) * 1000
self.latencies.append(exchange_latency)
# Store to Redis
await self._store_orderbook(exchange, symbol, snapshot)
except Exception as e:
print(f"Binance connection error: {e}, reconnecting...")
await asyncio.sleep(1)
async def _store_orderbook(self, exchange: str, symbol: str, snapshot: OrderBookSnapshot):
"""Store orderbook to Redis với TTL"""
key = f"ob:{exchange}:{symbol}"
data = {
'mid': snapshot.mid_price,
'spread_bps': snapshot.spread_bps,
'timestamp': snapshot.timestamp,
'bids': snapshot.bids[:5], # Top 5 levels
'asks': snapshot.asks[:5]
}
await self.redis.hset(key, mapping={
'mid': data['mid'],
'spread_bps': data['spread_bps'],
'timestamp': data['timestamp'],
'bids_json': json.dumps(data['bids']),
'asks_json': json.dumps(data['asks'])
})
await self.redis.expire(key, 60)
async def get_spread_opportunities(self) -> List[dict]:
"""Tìm spread opportunities giữa các exchanges"""
opportunities = []
for pair in self.pairs:
symbol_base = pair.replace('/', '').upper()
mid_prices = {}
for exchange in self.WS_ENDPOINTS.keys():
key = f"ob:{exchange}:{symbol_base}"
data = await self.redis.hgetall(key)
if data and b'mid' in data:
mid_prices[exchange] = float(data[b'mid'])
# Calculate pairwise spreads
exchanges = list(mid_prices.keys())
for i in range(len(exchanges)):
for j in range(i + 1, len(exchanges)):
e1, e2 = exchanges[i], exchanges[j]
p1, p2 = mid_prices[e1], mid_prices[e2]
spread_pct = abs(p1 - p2) / ((p1 + p2) / 2) * 100
if spread_pct > 0.1: # >10bps opportunity
opportunities.append({
'pair': pair,
'buy_exchange': e1 if p1 < p2 else e2,
'sell_exchange': e2 if p1 < p2 else e1,
'buy_price': min(p1, p2),
'sell_price': max(p1, p2),
'spread_bps': spread_pct * 100,
'timestamp': int(time.time() * 1000)
})
return opportunities
Usage
async def main():
redis_client = await redis.from_url("redis://localhost:6379")
collector = MultiExchangeCollector(
redis_client,
pairs=['BTC/USDT', 'ETH/USDT', 'SOL/USDT', 'ARB/USDT']
)
# Run collectors concurrently
tasks = [
collector.collect_binance(),
collector.collect_okx(),
collector.collect_bybit()
]
await asyncio.gather(*tasks)
if __name__ == "__main__":
asyncio.run(main())Feature Engineering: Statistical Arbitrage Features
Với statistical arbitrage, các features quan trọng nhất tôi đã validate qua backtesting và live trading:
- Spread features: z-score, half-life, cointegration beta
- Order flow: bid-ask pressure, volume imbalance
- Market microstructure: queue position, cancel rates
- Cross-exchange correlation: lead-lag relationships
# Feature Engineering Pipeline cho Statistical Arbitrage
Sử dụng Polars cho high-performance computation
import polars as pl
import numpy as np
from typing import Dict, Tuple
from dataclasses import dataclass
from collections import deque
import redis
import json
from datetime import datetime, timedelta
import time
@dataclass
class ArbitrageFeatures:
"""Feature container cho arbitrage strategy"""
z_score: float
half_life: float
spread_volatility: float
mean_reversion_speed: float
volume_imbalance: float
correlation: float
cointegration_beta: float
momentum: float
class FeatureEngine:
"""Real-time feature computation cho statistical arbitrage"""
def __init__(self, lookback_minutes: int = 60):
self.lookback = lookback_minutes
self.spread_history: deque = deque(maxlen=lookback_minutes * 60) # 1 tick/sec
self.volume_history: deque = deque(maxlen=1000)
def compute_spread_features(
self,
prices_df: pl.DataFrame,
window_short: int = 20,
window_long: int = 60
) -> pl.DataFrame:
"""Compute spread-based features"""
# Calculate log prices
prices_df = prices_df.with_columns([
pl.col('price').log().alias('log_price')
])
# Spread = log(price_exchange_A) - log(price_exchange_B)
if 'price_A' in prices_df.columns:
prices_df = prices_df.with_columns([
(pl.col('log_price') - pl.col('log_price_B')).alias('spread')
])
else:
# Self-referencing spread (e.g., BTC/USDT across time)
prices_df = prices_df.with_columns([
(pl.col('log_price') - pl.col('log_price').shift(1)).alias('spread')
])
# Rolling statistics
prices_df = prices_df.with_columns([
# Rolling mean
pl.col('spread').rolling_mean(window_long).alias('spread_mean_long'),
pl.col('spread').rolling_mean(window_short).alias('spread_mean_short'),
# Rolling std
pl.col('spread').rolling_std(window_long).alias('spread_std'),
# Z-score
((pl.col('spread') - pl.col('spread_mean_long')) /
pl.col('spread_std')).alias('z_score'),
])
return prices_df
def compute_half_life(self, spread_series: np.ndarray) -> float:
"""
Tính half-life của mean reversion
Sử dụng Ornstein-Uhlenbeck process: dS = lambda * (mu - S) * dt + dW
Half-life = -log(2) / lambda
"""
spread = spread_series[-60:] # Last 60 observations
# Lag spread
spread_lag = np.roll(spread, 1)
spread_lag[0] = spread_lag[1]
# Delta spread
delta_spread = spread - spread_lag
# Fit linear regression: delta = lambda * lag + noise
# lambda < 0 means mean reversion
try:
lambda_coef = np.polyfit(spread_lag[1:], delta_spread[1:], deg=1)[0]
if lambda_coef >= 0:
return float('inf') # No mean reversion
half_life = -np.log(2) / lambda_coef
return max(0.1, min(half_life, 1000)) # Bound between 0.1 and 1000
except:
return 60.0 # Default: 1 minute
def compute_cointegration(
self,
price_A: np.ndarray,
price_B: np.ndarray
) -> Tuple[float, float, float]:
"""
Engle-Granger cointegration test
Returns: beta, residual_std, p_value_approximation
"""
# Hedge ratio (beta) via OLS
beta = np.polyfit(price_A, price_B, deg=1)[0]
# Spread = price_A - beta * price_B
spread = price_A - beta * price_B
# Half-life
half_life = self.compute_half_life(spread)
# Residual volatility (annualized)
residual_std = np.std(spread) * np.sqrt(365 * 24 * 60)
# Simple p-value approximation based on ADF-like statistic
adf_stat = self._adf_statistic(spread)
p_value = self._approximate_p_value(adf_stat, len(spread))
return beta, half_life, residual_std, p_value
def _adf_statistic(self, series: np.ndarray) -> float:
"""Simplified ADF statistic"""
spread = series[-60:]
spread_lag = np.roll(spread, 1)[1:]
delta = np.diff(spread)
try:
coef = np.polyfit(spread_lag, delta, deg=1)[0]
return coef * len(spread) / (np.std(delta) + 1e-10)
except:
return 0.0
def _approximate_p_value(self, adf_stat: float, n: int) -> float:
"""MacKinnon approximate p-value"""
if adf_stat > 0:
return 1.0
# Simplified critical values (for n=60)
if adf_stat < -3.5:
return 0.01
elif adf_stat < -2.9:
return 0.05
elif adf_stat < -2.6:
return 0.10
return 0.5
def compute_volume_features(
self,
trades_df: pl.DataFrame
) -> pl.DataFrame:
"""Volume imbalance and order flow features"""
return trades_df.with_columns([
# Volume imbalance: (buy_vol - sell_vol) / (buy_vol + sell_vol)
((pl.col('buy_volume') - pl.col('sell_volume')) /
(pl.col('buy_volume') + pl.col('sell_volume') + 1e-10)).alias('volume_imbalance'),
# Order flow momentum
pl.col('volume_imbalance').rolling_mean(10).alias('flow_momentum'),
# Trade size distribution
(pl.col('trade_size').rolling_std(50) /
(pl.col('trade_size').rolling_mean(50) + 1e-10)).alias('trade_size_cv'),
])
def build_feature_vector(
self,
symbol: str,
price_data: pl.DataFrame,
trade_data: Optional[pl.DataFrame] = None
) -> ArbitrageFeatures:
"""Build complete feature vector cho model inference"""
# Compute spread features
price_features = self.compute_spread_features(price_data)
latest = price_features.tail(1)
z_score = latest['z_score'].item() if 'z_score' in latest.columns else 0.0
spread_volatility = latest['spread_std'].item() if 'spread_std' in latest.columns else 0.0
# Half-life computation
spread_series = price_features['spread'].to_numpy()
half_life = self.compute_half_life(spread_series)
# Volume features
volume_imbalance = 0.0
if trade_data is not None:
vol_features = self.compute_volume_features(trade_data)
volume_imbalance = vol_features.tail(1)['volume_imbalance'].item()
return ArbitrageFeatures(
z_score=z_score,
half_life=half_life,
spread_volatility=spread_volatility,
mean_reversion_speed=1 / half_life if half_life > 0 else 0,
volume_imbalance=volume_imbalance,
correlation=0.0, # Computed separately
cointegration_beta=0.0, # Computed separately
momentum=0.0
)
Integration với HolySheep AI cho ML-powered feature enhancement
class MLFeatureEnhancer:
"""Use HolySheep AI cho advanced feature generation"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
async def generate_anomaly_features(
self,
historical_spreads: list,
current_features: ArbitrageFeatures
) -> dict:
"""
Sử dụng LLM để analyze market regime và suggest adjustments
"""
import aiohttp
prompt = f"""Analyze the following spread data for statistical arbitrage:
Current Spread Z-Score: {current_features.z_score:.2f}
Spread Volatility: {current_features.spread_volatility:.4f}
Half-Life: {current_features.half_life:.2f} minutes
Volume Imbalance: {current_features.volume_imbalance:.4f}
Historical spreads (last 30): {historical_spreads[-30:]}
Task:
1. Identify if current regime is trending or mean-reverting
2. Suggest position size multiplier (0.0 to 2.0)
3. Identify potential risk factors
4. Confidence score (0.0 to 1.0) for the signal
Return JSON format:
{{"regime": "trending|mean_reverting|mixed", "position_multiplier": float,
"risk_factors": ["string"], "confidence": float, "reasoning": "string"}}
"""
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,
"response_format": {"type": "json_object"}
}
) as resp:
result = await resp.json()
return json.loads(result['choices'][0]['message']['content'])Data Storage: Time-Series Optimization
Với tần suất update 100ms/orderbook, việc chọn đúng storage layer là critical. Tôi recommend:
| Data Type | Storage | Reason | Latency |
|---|---|---|---|
| Order Book (real-time) | Redis Hash | Sub-ms read/write | ~0.5ms |
| Aggregated Features | TimescaleDB | Timeseries + SQL | ~5ms |
| Historical Data | ClickHouse | Columnar, compression | ~50ms |
| ML Features | Feature Store (Feast) | Consistent serving | ~10ms |
# TimescaleDB schema cho OHLCV aggregation
CREATE TABLE arbitrage_ohlcv (
time TIMESTAMPTZ NOT NULL,
symbol TEXT NOT NULL,
exchange TEXT NOT NULL,
open DOUBLE PRECISION,
high DOUBLE PRECISION,
low DOUBLE PRECISION,
close DOUBLE PRECISION,
volume DOUBLE PRECISION,
spread_mean DOUBLE PRECISION,
spread_std DOUBLE PRECISION,
z_score DOUBLE PRECISION
);
SELECT create_hypertable('arbitrage_ohlcv', 'time',
chunk_time_interval => INTERVAL '1 hour');
Continuous aggregate cho 1-minute bars
CREATE MATERIALIZED VIEW arbitrage_1m
WITH (timescaledb.continuous) AS
SELECT time_bucket('1 minute', time) AS bucket,
symbol,
exchange,
AVG(close) as close,
MAX(high) as high,
MIN(low) as low,
FIRST(open, time) as open,
SUM(volume) as volume,
AVG(spread_mean) as spread_mean,
AVG(z_score) as z_score,
STDDEV(z_score) as z_score_vol
FROM arbitrage_ohlcv
GROUP BY bucket, symbol, exchange;
Refresh policy
SELECT add_continuous_aggregate_policy('arbitrage_1m',
start_offset => INTERVAL '3 hours',
end_offset => INTERVAL '1 hour',
schedule_interval => INTERVAL '1 minute');
Query: Get recent arbitrage opportunities
SELECT
a.symbol,
a.exchange as buy_exchange,
b.exchange as sell_exchange,
a.close as buy_price,
b.close as sell_price,
(b.close - a.close) / a.close * 100 as spread_pct,
a.z_score,
b.z_score
FROM arbitrage_1m a
JOIN arbitrage_1m b ON a.symbol = b.symbol AND a.bucket = b.bucket
WHERE a.close < b.close
AND a.bucket = NOW() - INTERVAL '1 minute'
ORDER BY spread_pct DESC
LIMIT 20;Backtesting Framework với Feature Store
Trước khi deploy, backtesting là bắt buộc. Framework của tôi sử dụng vectorized backtesting với realistic slippage và fees:
# Vectorized Backtesting cho Statistical Arbitrage
import polars as pl
import numpy as np
from typing import List, Tuple, Optional
from dataclasses import dataclass
from enum import Enum
class SignalType(Enum):
LONG_SPREAD = 1 # Buy exchange A, sell exchange B
SHORT_SPREAD = -1
FLAT = 0
@dataclass
class Trade:
timestamp: int
symbol: str
direction: SignalType
entry_price: float
exit_price: float
size: float
pnl: float
hold_time: int # milliseconds
class BacktestEngine:
"""Vectorized backtesting với realistic execution model"""
def __init__(
self,
initial_capital: float = 100000,
maker_fee: float = 0.0004,
taker_fee: float = 0.0008,
slippage_bps: float = 1.0,
max_position_size: float = 0.1
):
self.capital = initial_capital
self.maker_fee = maker_fee
self.taker_fee = taker_fee
self.slippage = slippage_bps / 10000
self.max_position = max_position_size
self.trades: List[Trade] = []
def compute_entry_execution(
self,
mid_price: float,
direction: SignalType,
order_book_depth: float = 1.0
) -> Tuple[float, float]:
"""
Tính execution price với slippage
Returns: (entry_price, fees_paid)
"""
if direction == SignalType.LONG_SPREAD:
# Mua ở exchange đắt hơn (ask), bán ở exchange rẻ hơn (bid)
entry = mid_price * (1 + self.slippage * order_book_depth)
elif direction == SignalType.SHORT_SPREAD:
entry = mid_price * (1 - self.slippage * order_book_depth)
else:
entry = mid_price
fees = mid_price * self.taker_fee * 2 # 2 legs
return entry, fees
def run_backtest(
self,
price_data: pl.DataFrame,
z_score_entry: float = 2.0,
z_score_exit: float = 0.5,
z_score_stop: float = 3.5,
hold_limit_ms: int = 60000
) -> dict:
"""
Vectorized backtest với rolling z-score signal
"""
df = price_data.sort('timestamp')
n = len(df)
# Position tracking
position = 0
entry_price = 0.0
entry_time = 0
pnl_list = []
equity_curve = [self.capital]
for i in range(60, n): # Need 60 observations for z-score
current = df.row(i)
timestamp, z_score, spread = current[0], current['z_score'], current['spread']
if position == 0:
# Entry logic
if z_score > z_score_entry:
direction = SignalType.SHORT_SPREAD
entry_price, fees = self.compute_entry_execution(spread, direction)
position = 1
entry_time = timestamp
elif z_score < -z_score_entry:
direction = SignalType.LONG_SPREAD
entry_price, fees = self.compute_entry_execution(spread, direction)
position = 1
entry_time = timestamp
elif position == 1:
# Exit logic
should_exit = False
if direction == SignalType.LONG_SPREAD and z_score > z_score_exit:
should_exit = True
elif direction == SignalType.SHORT_SPREAD and z_score < -z_score_exit:
should_exit = True
elif abs(z_score) > z_score_stop:
should_exit = True
elif timestamp - entry_time > hold_limit_ms:
should_exit = True
if should_exit:
exit_price, fees = self.compute_entry_execution(spread, direction)
if direction == SignalType.LONG_SPREAD:
pnl = (exit_price - entry_price) * self.max_position
else:
pnl = (entry_price - exit_price) * self.max_position
pnl -= fees
self.capital += pnl
pnl_list.append(pnl)
equity_curve.append(self.capital)
position = 0
return self._compute_metrics(pnl_list, equity_curve)
def _compute_metrics(self, pnl_list: List[float], equity: List[float]) -> dict:
"""Compute performance metrics"""
pnl_arr = np.array(pnl_list)
return {
'total_trades': len(pnl_list),
'win_rate': np.sum(pnl_arr > 0) / (len(pnl_list) + 1e-10),
'avg_win': np.mean(pnl_arr[pnl_arr > 0]) if np.any(pnl_arr > 0) else 0,
'avg_loss': np.mean(pnl_arr[pnl_arr < 0]) if np.any(pnl_arr < 0) else 0,
'profit_factor': abs(np.sum(pnl_arr[pnl_arr > 0]) /
(np.sum(pnl_arr[pnl_arr < 0]) + 1e-10)),
'max_drawdown': self._max_drawdown(equity),
'sharpe_ratio': self._sharpe_ratio(pnl_arr),
'final_capital': equity[-1],
'total_return': (equity[-1] - 100000) / 100000
}
def _max_drawdown(self, equity: List[float]) -> float:
peak = equity[0]
max_dd = 0
for value in equity:
if value > peak:
peak = value
dd = (peak - value) / peak
max_dd = max(max_dd, dd)
return max_dd
def _sharpe_ratio(self, returns: np.ndarray, risk_free: float = 0.0) -> float:
excess = returns - risk_free
return np.mean(excess) / (np.std(excess) + 1e-10) * np.sqrt(252 * 24 * 60)
Usage
def optimize_parameters(df: pl.DataFrame):
"""Grid search cho optimal parameters"""
engine = BacktestEngine()
results = []
for entry_z in [1.5, 2.0, 2.5, 3.0]:
for exit_z in [0.3, 0.5, 0.7, 1.0]:
for stop_z in [3.0, 3.5, 4.0]:
metrics = engine.run_backtest(
df,
z_score_entry=entry_z,
z_score_exit=exit_z,
z_score_stop=stop_z
)
results.append({
'entry': entry_z,
'exit': exit_z,
'stop': stop_z,
**metrics
})
return pl.DataFrame(results).sort('profit_factor', descending=True)ML-Powered Regime Detection
Statistical arbitrage hoạt động kém trong trending markets. Tôi sử dụng HolySheep AI để detect market regime và adjust strategy parameters tự động:
# Market Regime Detection với HolySheep AI
API: https://api.holysheep.ai/v1
import aiohttp
import asyncio
import json
from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum
class MarketRegime(Enum):
TRENDING_UP = "trending_up"
TRENDING_DOWN = "trending_down"
MEAN_REVERTING = "mean_reverting"
HIGH_VOLATILITY = "high_volatility"
LOW_LIQUIDITY = "low_liquidity"
UNKNOWN = "unknown"
@dataclass
class RegimeAnalysis:
regime: MarketRegime
confidence: float
position_size_multiplier: float
stop_loss_multiplier: float
reasoning: str
class RegimeDetector:
"""
Sử dụng LLM để analyze market conditions
và suggest optimal strategy parameters
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
async def analyze_regime(
self,
price_data: List[Dict],
orderflow_data: List[Dict],
correlation_data: Dict[str, float]
) -> RegimeAnalysis:
"""
Analyze current market regime sử dụng HolySheep AI
Args:
price_data: Recent price movements
orderflow_data: Order flow imbalance data
correlation_data: Cross-exchange correlations
"""
# Format data for LLM
price_summary = self._summarize_prices(price_data)
flow_summary = self._summarize_orderflow(orderflow_data)
prompt = f"""Analyze the following cryptocurrency market data to determine the optimal
trading regime for a statistical arbitrage strategy.
Price Data Summary (last 60 minutes)
{price_summary}
Order Flow Data
{flow_summary}
Cross-Exchange Correlations
{json.dumps(correlation_data, indent=2)}
Your Task
1. Determine the current market regime
2. Assess regime confidence (0-1)
3. Suggest position size multiplier (0.0-1.5 for lower risk, 1.5-2.0 for higher confidence)
4. Suggest stop-loss multiplier adjustment
5. Provide reasoning
Regimes to choose from:
- trending_up: Strong directional movement, mean-reversion risky
- trending_down: Strong directional movement, mean-reversion risky
- mean_reverting: Ideal for statistical arbitrage
- high_volatility: Increase spreads, reduce size
- low_liquidity: Widen entry criteria, reduce size
Return format (JSON):
{{
"regime": "string",
"confidence": 0.0-1.0,
"position_size_multiplier": 0.0-2.0,
"stop_loss_multiplier": 0.5-2.0,
"reasoning": "string explaining your analysis"
}}
"""
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": "system",
"content": "You are an expert cryptocurrency market analyst specializing in statistical arbitrage. Analyze market data objectively and provide actionable recommendations."
},
{"role": "user", "content": prompt}
],
"temperature": 0.2,
"max_tokens": 500,
"response_format": {"type": "json_object"}
},
timeout=aiohttp.ClientTimeout(total=5)
) as resp:
if resp.status == 200:
result = await resp.json()
analysis = json.loads(result['choices'][0]['message']['content'])
return RegimeAnalysis(
regime=MarketRegime(analysis['regime']),
confidence=analysis['confidence'],
position_size_multiplier=analysis['position_size_multiplier'],
stop_loss_multiplier=analysis['stop_loss_multiplier'],
reasoning=analysis['reasoning']
)
else:
# Fallback to rule-based regime detection
return self._fallback_regime_detection(price_data, orderflow_data)
def _summarize_prices(self, data: List[Dict]) -> str:
"""Create summary statistics for price data"""
if not data: