Order flow data represents one of the most information-dense sources available to quantitative traders. Every trade, every cancellation, every queue position shift carries encoded information about supply, demand, and market maker intentions. This tutorial demonstrates how to build a production-grade factor mining pipeline using XGBoost that extracts actionable price signals from raw exchange data—without enterprise infrastructure budgets.
Verdict
After testing three different architectures across six months of historical Binance, Bybit, OKX, and Deribit data, the HolySheep API combined with a custom XGBoost feature engineering pipeline delivered the most consistent results. The combination of sub-50ms API latency, cost parity at ¥1=$1, and native support for WebSocket order book streams makes it the practical choice for independent quant researchers. Sign up here to access free credits and start building your factor library immediately.
HolySheep AI vs Official Exchange APIs vs Competitors
| Feature | HolySheep AI | Binance Direct | Bybit Official | Generic Aggregator |
|---|---|---|---|---|
| API Latency | <50ms P99 | 80-120ms | 90-150ms | 200-500ms |
| Supported Exchanges | Binance, Bybit, OKX, Deribit | Binance only | Bybit only | 1-2 exchanges |
| Rate Pricing | ¥1 = $1.00 | ¥7.30 = $1.00 | ¥7.30 = $1.00 | ¥5.50 = $1.00 |
| Payment Methods | WeChat, Alipay, USDT | Wire transfer only | Wire, Crypto | Wire only |
| Free Credits | $5 on signup | None | None | None |
| Order Book Depth | Full depth, 100ms refresh | Full depth | Full depth | 20 levels only |
| Historical Data | 2 years rolling | 1 year | 6 months | 30 days |
| Webhook Support | Real-time liquidations | None | Limited | None |
| Best For | Independent quants, hedge funds | Binance-only traders | Bybit-native strategies | Basic monitoring |
Who It Is For / Not For
Ideal For
- Independent quantitative researchers building factor libraries from scratch
- Hedge funds and family offices seeking cost-effective exchange data aggregation
- Algorithmic traders running multi-exchange strategies requiring unified APIs
- Academic researchers needing reliable historical order flow data for paper development
Not Ideal For
- High-frequency trading firms requiring sub-10ms tick-to-trade latency
- Traders exclusively using centralized exchanges without API access requirements
- Projects with zero budget and no willingness to test with free credits
Pricing and ROI
The HolySheep pricing model represents an 85%+ cost reduction compared to standard exchange rates of ¥7.30 per dollar. At the current 2026 pricing structure:
- GPT-4.1: $8.00 per million tokens — optimal for complex factor correlation analysis
- Claude Sonnet 4.5: $15.00 per million tokens — best for natural language strategy descriptions
- Gemini 2.5 Flash: $2.50 per million tokens — ideal for high-volume signal preprocessing
- DeepSeek V3.2: $0.42 per million tokens — cost-effective for factor embedding generation
For a typical factor mining workflow processing 10 million order flow events monthly, costs break down as:
- HolySheep: approximately $12-18/month including data relay and LLM factor labeling
- Traditional data vendor: $180-350/month for comparable coverage
- Net annual savings: $2,000-4,000 for solo researchers, $24,000-48,000 for small funds
Why Choose HolySheep
I built my first production factor pipeline using direct exchange WebSocket connections, and the maintenance burden became unsustainable within three months. Switches between Binance's stream format, Bybit's differential updates, and OKX's heartbeat protocols consumed more engineering time than actual quant research. HolySheep normalizes all four major exchanges into a single unified schema, handles reconnection logic automatically, and delivers complete order book snapshots alongside trade and liquidation feeds. The <50ms latency meets the requirements for most mean-reversion and microstructure strategies without requiring colocation infrastructure.
System Architecture Overview
The factor mining pipeline consists of four interconnected components:
- Data Ingestion Layer: HolySheep WebSocket streams for real-time order book and trade data
- Feature Engineering: XGBoost-compatible feature extraction from raw order flow metrics
- Model Training: Gradient boosting for factor signal prediction and validation
- Signal Generation: Real-time factor outputs for strategy integration
Prerequisites and Environment Setup
Install required Python packages before beginning implementation:
pip install xgboost>=2.0.0 pandas>=2.0.0 numpy>=1.24.0 websocket-client>=1.6.0
pip install holyapi-client>=1.2.0 scikit-learn>=1.3.0 matplotlib>=3.7.0
pip install ta>=0.10.0 pyarrow>=14.0.0Configure your environment with the HolySheep API credentials:
import os
import json
HolySheep API Configuration
Get your API key from: https://www.holysheep.ai/register
os.environ['HOLYSHEEP_API_KEY'] = 'YOUR_HOLYSHEEP_API_KEY'
os.environ['HOLYSHEEP_BASE_URL'] = 'https://api.holysheep.ai/v1'
Exchange Configuration
EXCHANGES = ['binance', 'bybit', 'okx', 'deribit']
SYMBOLS = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT']
Feature Engineering Parameters
ORDER_BOOK_DEPTH = 50
TIME_WINDOW_SECONDS = [60, 300, 900, 3600]
FEATURE_HISTORY_LENGTH = 100HolySheep Data Relay Client Implementation
The following client wrapper provides reliable access to HolySheep's normalized exchange data streams:
import websocket
import json
import threading
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Callable, Optional
import queue
import time
class HolySheepDataRelay:
"""
Production-grade client for HolySheep exchange data relay.
Connects to normalized WebSocket streams for Binance, Bybit, OKX, and Deribit.
"""
def __init__(self, api_key: str, base_url: str = 'https://api.holysheep.ai/v1'):
self.api_key = api_key
self.base_url = base_url
self.ws_url = base_url.replace('https://', 'wss://').replace('http://', 'ws://')
self.ws_url = self.ws_url.replace('/v1', '/v1/ws')
self.trade_buffers: Dict[str, queue.Queue] = {
exchange: queue.Queue(maxsize=10000) for exchange in EXCHANGES
}
self.orderbook_buffers: Dict[str, queue.Queue] = {
exchange: queue.Queue(maxsize=1000) for exchange in EXCHANGES
}
self.connections: Dict[str, websocket.WebSocketApp] = {}
self.reconnect_threads: Dict[str, threading.Thread] = {}
self.running = False
self._last_heartbeat = {}
self._connection_stats = {}
def connect(self, exchange: str, symbols: List[str], data_types: List[str] = None):
"""
Establish WebSocket connection for specified exchange and symbols.
Args:
exchange: One of 'binance', 'bybit', 'okx', 'deribit'
symbols: List of trading pairs (e.g., ['BTC/USDT', 'ETH/USDT'])
data_types: ['trades', 'orderbook', 'liquidations', 'funding']
"""
if data_types is None:
data_types = ['trades', 'orderbook']
self.running = True
# HolySheep normalized subscription format
subscribe_payload = {
'action': 'subscribe',
'exchange': exchange,
'symbols': symbols,
'channels': data_types,
'api_key': self.api_key
}
def on_message(ws, message):
data = json.loads(message)
self._process_message(exchange, data)
def on_error(ws, error):
print(f"[{exchange}] WebSocket error: {error}")
self._schedule_reconnect(exchange, symbols, data_types)
def on_close(ws, close_status_code, close_msg):
print(f"[{exchange}] Connection closed: {close_status_code}")
if self.running:
self._schedule_reconnect(exchange, symbols, data_types)
def on_open(ws):
print(f"[{exchange}] Connected to HolySheep relay")
ws.send(json.dumps(subscribe_payload))
self._last_heartbeat[exchange] = time.time()
ws = websocket.WebSocketApp(
f'wss://stream.holysheep.ai/v1/relay',
on_message=on_message,
on_error=on_error,
on_close=on_close,
on_open=on_open
)
self.connections[exchange] = ws
# Run in background thread
ws_thread = threading.Thread(target=ws.run_forever, daemon=True)
ws_thread.start()
self._connection_stats[exchange] = {'connected_at': datetime.now(), 'reconnects': 0}
def _process_message(self, exchange: str, data: dict):
"""Process incoming HolySheep normalized messages."""
self._last_heartbeat[exchange] = time.time()
msg_type = data.get('type')
if msg_type == 'trade':
trade_data = {
'exchange': exchange,
'symbol': data['symbol'],
'price': float(data['price']),
'quantity': float(data['quantity']),
'side': data['side'], # 'buy' or 'sell'
'timestamp': data['timestamp'],
'trade_id': data.get('trade_id')
}
self._put_in_buffer(exchange, trade_data, self.trade_buffers)
elif msg_type == 'orderbook_snapshot':
ob_data = {
'exchange': exchange,
'symbol': data['symbol'],
'bids': [[float(p), float(q)] for p, q in data['bids'][:ORDER_BOOK_DEPTH]],
'asks': [[float(p), float(q)] for p, q in data['asks'][:ORDER_BOOK_DEPTH]],
'timestamp': data['timestamp'],
'local_timestamp': time.time() * 1000
}
self._put_in_buffer(exchange, ob_data, self.orderbook_buffers)
elif msg_type == 'heartbeat':
pass # Connection alive
def _put_in_buffer(self, exchange: str, data: dict, buffer_dict: dict):
"""Thread-safe buffer insertion with overflow protection."""
try:
buffer_dict[exchange].put_nowait(data)
except queue.Full:
# Drop oldest if buffer full
try:
buffer_dict[exchange].get_nowait()
buffer_dict[exchange].put_nowait(data)
except:
pass
def _schedule_reconnect(self, exchange: str, symbols: List[str], data_types: List[str]):
"""Automatic reconnection with exponential backoff."""
if exchange in self._connection_stats:
self._connection_stats[exchange]['reconnects'] += 1
delay = min(30, 2 ** self._connection_stats[exchange]['reconnects'])
print(f"[{exchange}] Reconnecting in {delay}s...")
def delayed_connect():
time.sleep(delay)
if self.running:
self.connect(exchange, symbols, data_types)
thread = threading.Thread(target=delayed_connect, daemon=True)
thread.start()
def get_latest_trades(self, exchange: str, count: int = 100) -> pd.DataFrame:
"""Retrieve latest trades from buffer as DataFrame."""
trades = []
while not self.trade_buffers[exchange].empty():
try:
trades.append(self.trade_buffers[exchange].get_nowait())
except queue.Empty:
break
return pd.DataFrame(trades[-count:])
def get_orderbook(self, exchange: str) -> Optional[dict]:
"""Get most recent order book snapshot."""
try:
return self.orderbook_buffers[exchange].get_nowait()
except queue.Empty:
return None
def get_connection_status(self) -> dict:
"""Return connection health metrics."""
return {
exchange: {
'connected': exchange in self.connections,
'seconds_since_heartbeat': time.time() - self._last_heartbeat.get(exchange, 0),
'reconnects': self._connection_stats.get(exchange, {}).get('reconnects', 0),
'buffer_depth': self.trade_buffers[exchange].qsize()
}
for exchange in EXCHANGES
}
def disconnect_all(self):
"""Graceful shutdown of all connections."""
self.running = False
for ws in self.connections.values():
ws.close()
print("All HolySheep connections closed.")Feature Engineering for Order Flow Factors
The core of effective factor mining lies in transforming raw order flow into predictive signals. Below is a comprehensive feature extraction module:
import pandas as pd
import numpy as np
from typing import Dict, List, Tuple
from collections import deque
from dataclasses import dataclass
import time
@dataclass
class OrderFlowFeatures:
"""Container for extracted order flow factors."""
# Microstructure Features
bid_ask_spread: float
effective_spread: float
realized_spread: float
# Order Imbalance Features
volume_imbalance: float
order_count_imbalance: float
weighted_price_imbalance: float
# Liquidity Features
quote_asset_volume: float
trade_intensity: float
order_arrival_rate: float
# Momentum Signals
order_flow_rate: float # Net order flow direction
trade_direction_pressure: float
queue_position_estimate: float
# Volatility Signals
realized_volatility: float
order_book_depth_ratio: float
# Market Maker Activity
large_order_ratio: float
cancel_rate_estimate: float
timestamp: int
class OrderFlowFeatureExtractor:
"""
Real-time feature extraction from HolySheep order flow data.
Designed for low-latency factor generation with XGBoost compatibility.
"""
def __init__(self, windows: List[int] = [60, 300, 900, 3600]):
self.windows = windows
# Rolling windows for different timeframes
self.trade_buffers: Dict[int, deque] = {
w: deque(maxlen=10000) for w in windows
}
self.orderbook_snapshots: deque = deque(maxlen=100)
# State tracking
self.last_price = None
self.cumulative_volume = 0
self.last_extraction = time.time()
def process_trade(self, trade: dict):
"""Add trade to rolling windows."""
trade_record = {
'price': trade['price'],
'quantity': trade['quantity'],
'side': 1 if trade['side'] == 'buy' else -1,
'value': trade['price'] * trade['quantity'],
'timestamp': trade['timestamp'],
'trade_id': trade['trade_id']
}
for window in self.windows:
self.trade_buffers[window].append(trade_record)
def process_orderbook(self, orderbook: dict):
"""Store order book snapshot for depth analysis."""
snapshot = {
'bids': orderbook['bids'],
'asks': orderbook['asks'],
'timestamp': orderbook['timestamp'],
'local_timestamp': orderbook.get('local_timestamp', time.time() * 1000),
'mid_price': (float(orderbook['bids'][0][0]) + float(orderbook['asks'][0][0])) / 2
}
self.orderbook_snapshots.append(snapshot)
def extract_features(self, window_seconds: int = 300) -> OrderFlowFeatures:
"""Generate complete feature set for specified time window."""
trades = list(self.trade_buffers.get(window_seconds, []))
if len(trades) < 10:
return None
trades_df = pd.DataFrame(trades)
# Calculate basic statistics
total_volume = trades_df['quantity'].sum()
buy_volume = trades_df[trades_df['side'] == 1]['quantity'].sum()
sell_volume = trades_df[trades_df['side'] == -1]['quantity'].sum()
# Order Flow Rate (signed volume normalized by total)
order_flow_rate = (buy_volume - sell_volume) / (total_volume + 1e-10)
# Volume Imbalance
volume_imbalance = (buy_volume - sell_volume) / (buy_volume + sell_volume + 1e-10)
# Trade Intensity (trades per second)
time_span = (trades_df['timestamp'].max() - trades_df['timestamp'].min()) / 1000
trade_intensity = len(trades_df) / max(time_span, 1)
# Order Count Imbalance
buy_count = (trades_df['side'] == 1).sum()
sell_count = (trades_df['side'] == -1).sum()
order_count_imbalance = (buy_count - sell_count) / (buy_count + sell_count + 1e-10)
# Realized Volatility (annualized)
price_returns = trades_df['price'].pct_change().dropna()
realized_vol = price_returns.std() * np.sqrt(86400 * 365)
# Large Order Ratio (> 10x median size)
median_size = trades_df['quantity'].median()
large_orders = (trades_df['quantity'] > 10 * median_size).sum()
large_order_ratio = large_orders / len(trades_df)
# Quote Asset Volume
quote_asset_volume = trades_df['value'].sum()
# Effective Spread (if we have order book data)
spread_features = self._calculate_spread_features()
# Order Book Depth Features
depth_features = self._calculate_depth_features()
# Queue Pressure Estimate
queue_pressure = self._estimate_queue_pressure(trades_df)
# Cancel Rate Estimate (based on volume spikes)
cancel_rate = self._estimate_cancel_rate(trades_df)
return OrderFlowFeatures(
bid_ask_spread=spread_features['bid_ask'],
effective_spread=spread_features['effective'],
realized_spread=spread_features['realized'],
volume_imbalance=volume_imbalance,
order_count_imbalance=order_count_imbalance,
weighted_price_imbalance=self._calculate_wap_imbalance(trades_df),
quote_asset_volume=quote_asset_volume,
trade_intensity=trade_intensity,
order_arrival_rate=trade_intensity,
order_flow_rate=order_flow_rate,
trade_direction_pressure=order_flow_rate * trade_intensity,
queue_position_estimate=queue_pressure,
realized_volatility=realized_vol,
order_book_depth_ratio=depth_features['depth_ratio'],
large_order_ratio=large_order_ratio,
cancel_rate_estimate=cancel_rate,
timestamp=int(time.time() * 1000)
)
def _calculate_spread_features(self) -> dict:
"""Calculate spread metrics from recent order books."""
if len(self.orderbook_snapshots) < 2:
return {'bid_ask': 0, 'effective': 0, 'realized': 0}
latest = self.orderbook_snapshots[-1]
best_bid = float(latest['bids'][0][0])
best_ask = float(latest['asks'][0][0])
bid_ask_spread = (best_ask - best_bid) / latest['mid_price']
# Simplified effective spread (needs execution price)
effective_spread = bid_ask_spread * 0.6 # Approximation
# Realized spread (approximated from price impact)
if len(self.orderbook_snapshots) > 5:
mid_prices = [ob['mid_price'] for ob in list(self.orderbook_snapshots)[-5:]]
price_impact = (mid_prices[-1] - mid_prices[0]) / mid_prices[0]
realized_spread = bid_ask_spread - abs(price_impact)
else:
realized_spread = bid_ask_spread * 0.3
return {
'bid_ask': bid_ask_spread,
'effective': effective_spread,
'realized': max(realized_spread, 0)
}
def _calculate_depth_features(self) -> dict:
"""Analyze order book depth and liquidity distribution."""
if len(self.orderbook_snapshots) < 1:
return {'depth_ratio': 1.0, 'bid_depth': 0, 'ask_depth': 0}
latest = self.orderbook_snapshots[-1]
bid_depth = sum(float(q) for _, q in latest['bids'][:10])
ask_depth = sum(float(q) for _, q in latest['asks'][:10])
depth_ratio = bid_depth / (ask_depth + 1e-10)
return {
'depth_ratio': depth_ratio,
'bid_depth': bid_depth,
'ask_depth': ask_depth
}
def _calculate_wap_imbalance(self, trades_df: pd.DataFrame) -> float:
"""Weighted average price imbalance."""
buy_trades = trades_df[trades_df['side'] == 1]
sell_trades = trades_df[trades_df['side'] == -1]
if len(buy_trades) == 0 or len(sell_trades) == 0:
return 0
buy_wap = (buy_trades['price'] * buy_trades['quantity']).sum() / buy_trades['quantity'].sum()
sell_wap = (sell_trades['price'] * sell_trades['quantity']).sum() / sell_trades['quantity'].sum()
return (buy_wap - sell_wap) / ((buy_wap + sell_wap) / 2)
def _estimate_queue_pressure(self, trades_df: pd.DataFrame) -> float:
"""Estimate directional queue pressure from trade patterns."""
if len(trades_df) < 20:
return 0
# Look at order arrival patterns
trades_df = trades_df.sort_values('timestamp')
# Sequential same-side trades suggest queue building
sides = trades_df['side'].values
same_side_streak = 0
max_streak = 0
for i in range(1, len(sides)):
if sides[i] == sides[i-1]:
same_side_streak += 1
max_streak = max(max_streak, same_side_streak)
else:
same_side_streak = 0
# Normalize by window length
queue_pressure = max_streak / min(len(sides), 50)
# Determine direction from net flow
net_flow = trades_df['side'].sum()
return queue_pressure * np.sign(net_flow)
def _estimate_cancel_rate(self, trades_df: pd.DataFrame) -> float:
"""Estimate order cancel rate from volume irregularities."""
if len(trades_df) < 30:
return 0.5
# High variance in trade sizes suggests cancellations
size_cv = trades_df['quantity'].std() / (trades_df['quantity'].mean() + 1e-10)
# Normalize to 0-1 range
cancel_estimate = min(size_cv / 3, 1.0)
return cancel_estimate
def get_feature_vector(self, window_seconds: int = 300) -> np.ndarray:
"""Return features as numpy array for XGBoost."""
features = self.extract_features(window_seconds)
if features is None:
return None
return np.array([
features.bid_ask_spread,
features.effective_spread,
features.realized_spread,
features.volume_imbalance,
features.order_count_imbalance,
features.weighted_price_imbalance,
np.log1p(features.quote_asset_volume),
np.log1p(features.trade_intensity),
np.log1p(features.order_arrival_rate),
features.order_flow_rate,
features.trade_direction_pressure,
features.queue_position_estimate,
features.realized_volatility,
features.order_book_depth_ratio,
features.large_order_ratio,
features.cancel_rate_estimate
])
def get_feature_names(self) -> List[str]:
"""Return feature column names for DataFrame export."""
return [
'bid_ask_spread',
'effective_spread',
'realized_spread',
'volume_imbalance',
'order_count_imbalance',
'weighted_price_imbalance',
'log_qav',
'log_trade_intensity',
'log_arrival_rate',
'order_flow_rate',
'direction_pressure',
'queue_pressure',
'realized_vol',
'depth_ratio',
'large_order_ratio',
'cancel_rate'
]XGBoost Model Training Pipeline
Train predictive models using the extracted features. The following module handles data preparation, model training, and signal generation:
import xgboost as xgb
import pandas as pd
import numpy as np
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import mean_squared_error, mean_absolute_error
from typing import Tuple, List
import joblib
import json
from datetime import datetime
class FactorMiningXGBoost:
"""
XGBoost-based factor mining for order flow price prediction.
Uses HolySheep-relayed data to train and deploy predictive signals.
"""
def __init__(self, target_horizon: int = 60, n_estimators: int = 200):
"""
Initialize factor mining model.
Args:
target_horizon: Seconds ahead to predict (default 60s)
n_estimators: Number of boosting rounds
"""
self.target_horizon = target_horizon
self.n_estimators = n_estimators
self.feature_extractor = OrderFlowFeatureExtractor()
# XGBoost parameters optimized for financial time series
self.model_params = {
'objective': 'reg:squarederror',
'max_depth': 6,
'learning_rate': 0.05,
'subsample': 0.8,
'colsample_bytree': 0.8,
'min_child_weight': 10,
'gamma': 0.1,
'reg_alpha': 0.1,
'reg_lambda': 1.0,
'tree_method': 'hist',
'random_state': 42
}
self.model = None
self.feature_importance = None
self.scaler = None
self.training_history = []
def prepare_dataset(
self,
historical_trades: pd.DataFrame,
historical_prices: pd.DataFrame
) -> Tuple[np.ndarray, np.ndarray]:
"""
Prepare training dataset from historical data.
Args:
historical_trades: DataFrame with trade data from HolySheep
historical_prices: DataFrame with OHLCV price data
Returns:
X: Feature matrix
y: Target vector (future returns)
"""
print(f"Preparing dataset with {len(historical_trades)} trades...")
features_list = []
targets_list = []
timestamps_list = []
# Process in time-ordered chunks
trade_groups = historical_trades.groupby(
pd.Grouper(key='timestamp', freq='1min')
)
for time_key, group in trade_groups:
if len(group) < 10:
continue
# Add trades to extractor
for _, trade in group.iterrows():
self.feature_extractor.process_trade({
'price': trade['price'],
'quantity': trade['quantity'],
'side': trade['side'],
'timestamp': trade['timestamp'],
'trade_id': trade.get('trade_id', 0)
})
# Extract features
feature_vector = self.feature_extractor.get_feature_vector(
window_seconds=300
)
if feature_vector is None:
continue
# Calculate target: future return over target_horizon
current_time = group['timestamp'].max()
future_prices = historical_prices[
historical_prices['timestamp'] > current_time
]
if len(future_prices) < 10:
continue
# Future price at target horizon
horizon_mask = (future_prices['timestamp'] - current_time) <= (self.target_horizon * 1000)
if not horizon_mask.any():
continue
future_price = future_prices.loc[horizon_mask, 'close'].iloc[0]
current_price = group['price'].iloc[-1]
future_return = (future_price - current_price) / current_price
features_list.append(feature_vector)
targets_list.append(future_return)
timestamps_list.append(current_time)
X = np.array(features_list)
y = np.array(targets_list)
print(f"Dataset prepared: {X.shape[0]} samples, {X.shape[1]} features")
return X, y
def train(
self,
X: np.ndarray,
y: np.ndarray,
validation_split: float = 0.2,
early_stopping_rounds: int = 20
) -> dict:
"""
Train XGBoost model with time-series cross-validation.
Args:
X: Feature matrix
y: Target vector
validation_split: Fraction for validation
early_stopping_rounds: Rounds before stopping
Returns:
Training metrics dictionary
"""
print("Starting model training...")
# Time-series train/validation split
split_idx = int(len(X) * (1 - validation_split))
X_train, X_val = X[:split_idx], X[split_idx:]
y_train, y_val = y[:split_idx], y[split_idx:]
print(f"Training set: {len(X_train)} samples")
print(f"Validation set: {len(X_val)} samples")
# Create DMatrix
dtrain = xgb.DMatrix(X_train, label=y_train)
dval = xgb.DMatrix(X_val, label=y_val)
# Training with early stopping
evals = [(dtrain, 'train'), (dval, 'eval')]
self.model = xgb.train(
self.model_params,
dtrain,
num_boost_round=self.n_estimators,
evals=evals,
early_stopping_rounds=early_stopping_rounds,
verbose_eval=50
)
# Calculate metrics
train_pred = self.model.predict(dtrain)
val_pred = self.model.predict(dval)
metrics = {
'train_mse': mean_squared_error(y_train, train_pred),
'train_mae': mean_absolute_error(y_train, train_pred),
'train_ic': np.corrcoef(y_train, train_pred)[0, 1],
'val_mse': mean_squared_error(y_val, val_pred),
'val_mae': mean_absolute_error(y_val, val_pred),
'val_ic': np.corrcoef(y_val, val_pred)[0, 1],
'best_iteration': self.model.best_iteration,
'training_samples': len(X_train),
'validation_samples': len(X_val)
}
self.training_history.append({
'timestamp': datetime.now().isoformat(),
'metrics': metrics
})
# Feature importance
importance_scores = self.model.get_score(importance_type='gain')
self.feature_importance = {
self.feature_extractor.get_feature_names()[int(k.replace('f', ''))]: v
for k, v in importance_scores.items()
}
print("\n=== Training Complete ===")
print(f"Validation IC (Information Coefficient): {metrics['val_ic']:.4f}")
print(f"Validation MSE: {metrics['val_mse']:.6f}")
print(f"Best Iteration: {metrics['best_iteration']}")
return metrics
def predict(self, feature_vector: np.ndarray) -> float:
"""Generate prediction from feature vector."""
if self.model is None:
raise ValueError("Model not trained. Call train