rams/ftdt-quant-lab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Cartea-Jaimungal, Queue Imbalance, Guéant MM: 3 new quant finance strategies + backtests

Browse Source
This commit is contained in:
ramseshk
2026-08-04 06:19:19 +00:00
parent 2c0750e355
commit 9768bf80cc
8 changed files with 11311 additions and 574 deletions
Download Patch File Download Diff File Expand all files Collapse all files
strategies/cartea_jaimungal.py
+148
View File
@@ -0,0 +1,148 @@
"""
Cartea-Jaimungal HFT Model — Stochastic Control for High-Frequency Trading.
Combines short-term alpha signals with optimal market making and
statistical arbitrage, solving the Hamilton-Jacobi-Bellman (HJB)
equation via stochastic control.
Key equations (Cartea, Jaimungal, Penalva — "Algorithmic and
High-Frequency Trading", Cambridge 2015):
Reservation price:
r = S_t + α_t/(2γσ²) - q·γ·σ²·(T-t)
Optimal spread around reservation:
δ* = γ·σ²·(T-t)/2 + (1/γ)·log(1 + γ/κ)
where:
S_t = mid price
α_t = short-term alpha signal
γ = risk aversion parameter
σ = volatility
q = current inventory
T-t = time remaining
κ = order arrival intensity
The model adjusts quoting aggressively when alpha is strong
and inventory is low, and defensively when inventory is high.
Usage:
from strategies.cartea_jaimungal import CarteaJaimungal
cj = CarteaJaimungal(gamma=0.1, sigma=0.01, kappa=1.5)
bid, ask, res_price = cj.compute_quotes(mid_price, alpha, inventory)
"""
import math
class CarteaJaimungal:
"""HFT model: stochastic control for combined alpha + market making.
Produces optimal bid/ask quotes, reservation price, and position
limits given current market conditions and alpha signal.
"""
def __init__(self, gamma: float = 0.1, sigma: float = 0.01, kappa: float = 1.5,
T: float = 10.0, max_inventory: float = 0.01):
"""
Args:
gamma: risk aversion (higher = more defensive)
sigma: volatility (annualized)
kappa: order arrival intensity (fills per second)
T: time horizon in seconds
max_inventory: maximum absolute position size
"""
self.gamma = gamma
self.sigma = sigma
self.kappa = kappa
self.T = T
self.max_inventory = max_inventory
def compute_quotes(self, mid_price: float, alpha: float,
inventory: float, elapsed: float) -> dict:
"""Compute optimal bid/ask quotes.
Args:
mid_price: current mid price
alpha: short-term alpha signal (drift, in price units/sec)
inventory: current position (+ = long, - = short)
elapsed: time elapsed since start of session
Returns:
dict with bid, ask, reservation_price, half_spread
"""
tau = self.T - elapsed
if tau < 0.01:
tau = 0.01 # Prevent singularity at expiry
sig2 = self.sigma**2
# Reservation price (fair value adjusted for inventory and alpha)
# r = S + α/(2γσ²) - q·γ·σ²·(T-t)
alpha_term = alpha / (2 * self.gamma * sig2) if sig2 > 1e-10 else 0
inventory_penalty = inventory * self.gamma * sig2 * tau
reservation = mid_price + alpha_term - inventory_penalty
# Optimal half-spread
# δ* = γ·σ²·τ/2 + (1/γ)·log(1 + γ/κ)
spread_risk = self.gamma * sig2 * tau / 2.0
if self.kappa > 0 and self.gamma > 0:
log_term = (1.0 / self.gamma) * math.log(1.0 + self.gamma / self.kappa)
else:
log_term = 0.001
half_spread = max(spread_risk + log_term, 0.0001)
# Apply inventory constraints — don't quote beyond max position
max_long = self.max_inventory
max_short = -self.max_inventory
bid = reservation - half_spread
ask = reservation + half_spread
# If at max long, stop buying (no bid)
if inventory >= max_long:
bid = 0
# If at max short, stop selling (no ask → very high ask)
if inventory <= max_short:
ask = float('inf')
return {
"bid": round(bid, 1),
"ask": round(ask, 1),
"reservation": round(reservation, 1),
"half_spread": round(half_spread, 2),
"skew": round(reservation - mid_price, 2),
}
def should_trade(self, mid_price: float, alpha: float,
inventory: float, elapsed: float) -> dict:
"""Determine if we should enter a directional position based on alpha.
Returns dict with side, size, and confidence.
"""
quotes = self.compute_quotes(mid_price, alpha, inventory, elapsed)
# Size: scale with alpha magnitude, capped by inventory remaining
remaining_long = max(0, self.max_inventory - inventory)
remaining_short = max(0, self.max_inventory + inventory)
alpha_strength = abs(alpha)
threshold = self.gamma * self.sigma**2 * 0.1 # Minimum edge
signal = None
size = 0.0
confidence = 0.0
if alpha > threshold and remaining_long > 0:
signal = "BUY"
size = min(remaining_long, alpha_strength * 100)
confidence = min(alpha_strength / threshold / 5, 1.0)
elif alpha < -threshold and remaining_short > 0:
signal = "SELL"
size = min(remaining_short, alpha_strength * 100)
confidence = min(alpha_strength / threshold / 5, 1.0)
return {
"signal": signal,
"size": round(size, 6),
"confidence": round(confidence, 4),
"quotes": quotes,
}
strategies/gueant.py
+175
View File
@@ -0,0 +1,175 @@
"""
Guéant Closed-Form Market Making Model.
Extends Avellaneda-Stoikov with closed-form asymptotic solutions
that are computationally efficient and embed asymmetric information.
Reference: Guéant, Lehalle, Fernandez-Tapia — "Dealing with the
Inventory Risk: A solution to the market making problem" (2012)
Key improvements over standard A-S:
1. Closed-form solutions (no PDE solving needed)
2. Explicit handling of asymmetric information (adverse selection)
3. Explicit dependence on order book shape
4. Better terminal condition handling
Optimal quotes:
δ_a(t,q) = σ²γ(T-t)/2 + (1/γ)log(1 + γ/k)
δ_b(t,q) = δ_a(t,q)
r(t,q) = s - q·σ²γ(T-t) [reservation price]
where:
s = mid price
q = inventory
γ = risk aversion
σ = volatility
k = order arrival intensity
T-t = remaining time
Bid = r(t,q) - δ_b
Ask = r(t,q) + δ_a
The Guéant extension adds:
- Asymmetric spreads when adverse selection detected
- Queue-position dependent fill probabilities
- Better parameter estimation from LOB data
Usage:
from strategies.gueant import GueantMM
mm = GueantMM(gamma=0.1, sigma=0.01)
bid, ask = mm.optimal_quotes(mid, inventory, elapsed, adverse)
"""
import math
class GueantMM:
"""Closed-form market making with asymmetric information handling."""
def __init__(self, gamma: float = 0.1, sigma: float = 0.01,
k: float = 1.5, T: float = 60.0, max_pos: float = 0.005):
"""
Args:
gamma: risk aversion (0.01=v.aggressive, 1.0=v.conservative)
sigma: volatility (annualized)
k: baseline order arrival intensity
T: trading session length in seconds
max_pos: max absolute position
"""
self.gamma = gamma
self.sigma = sigma
self.k = k
self.T = T
self.max_pos = max_pos
def optimal_spread(self, tau: float, adverse_prob: float = 0) -> float:
"""Compute optimal half-spread.
Args:
tau: time remaining (T - elapsed)
adverse_prob: estimated adverse selection probability (0-1)
Returns half-spread δ in price units.
"""
if tau < 0.01:
tau = 0.01
sig2 = self.sigma**2
gamma = self.gamma
# Base Guéant spread: σ²γτ/2 + (1/γ)log(1+γ/k)
base_spread = gamma * sig2 * tau / 2.0
if gamma > 0 and self.k > 0:
log_term = (1.0 / gamma) * math.log(1.0 + gamma / self.k)
else:
log_term = 0.001
half_spread = base_spread + log_term
# Asymmetric information adjustment
# When adverse selection is high, widen spread proportionally
if adverse_prob > 0:
# Guéant extension: adverse selection increases effective spread
# δ_effective = δ_base · (1 + φ·P(adverse))
phi = 2.0 # Sensitivity to adverse selection
half_spread *= (1.0 + phi * adverse_prob)
return max(half_spread, 0.01) # Minimum 1 cent spread
def reservation_price(self, mid_price: float, inventory: float,
tau: float) -> float:
"""Compute reservation price adjusted for inventory risk.
r = s - q·γ·σ²·τ
Long inventory (q > 0): reservation shifts DOWN (want to sell)
Short inventory (q < 0): reservation shifts UP (want to buy)
"""
r = mid_price - inventory * self.gamma * self.sigma**2 * tau
return r
def optimal_quotes(self, mid_price: float, inventory: float,
elapsed: float, adverse_prob: float = 0,
bid_depth: float = 1.0, ask_depth: float = 1.0) -> dict:
"""Compute optimal bid and ask quotes.
Args:
mid_price: current mid price
inventory: current net position
elapsed: time elapsed this session
adverse_prob: estimated probability of adverse selection
bid_depth: relative bid depth (1.0 = normal, >1 = deeper book)
ask_depth: relative ask depth (1.0 = normal, >1 = deeper book)
Returns dict with bid, ask, reservation, half_spread, skew.
"""
tau = self.T - elapsed
if tau < 0.01:
tau = 0.01
r = self.reservation_price(mid_price, inventory, tau)
spread = self.optimal_spread(tau, adverse_prob)
# Adjust spread based on book depth
# Deeper book → tighter spreads (more competition)
# Thinner book → wider spreads (less competition)
bid_spread = spread / max(bid_depth, 0.5)
ask_spread = spread / max(ask_depth, 0.5)
bid = r - bid_spread
ask = r + ask_spread
# Enforce inventory limits
if inventory >= self.max_pos:
bid = 0 # Don't buy more
if inventory <= -self.max_pos:
ask = float('inf') # Don't sell more
return {
"bid": round(bid, 1),
"ask": round(ask, 1),
"reservation": round(r, 1),
"half_spread": round(spread, 2),
"bid_spread": round(bid_spread, 2),
"ask_spread": round(ask_spread, 2),
"skew": round(r - mid_price, 2),
}
def estimated_fill_probability(self, our_price: float,
best_price: float,
is_bid: bool) -> float:
"""Estimate probability our quote gets filled.
Based on distance from best and queue position.
At best (matching): high fill rate
1 tick away: moderate
>2 ticks away: low
"""
dist = abs(our_price - best_price) / best_price if best_price > 0 else 0
if dist < 0.0001: # At the best price level
return 0.30 # ~30% chance per tick
elif dist < 0.0005: # Within 1 tick
return 0.10
elif dist < 0.002: # Within 2 ticks
return 0.03
return 0.01
strategies/queue_imbalance.py
+170
View File
@@ -0,0 +1,170 @@
"""
Queue Imbalance Model — Order Book Dynamics for Short-Term Prediction.
Based on the Stoikov & Sağlam and Cont et al. frameworks.
Analyzes queue position, order flow, and imbalance to predict
short-term price direction.
Key concepts:
1. Queue position estimation — where we sit in the LOB queue
2. Order flow imbalance at each level — net adds vs cancels
3. Probability of mid-price move based on queue dynamics
4. Adverse selection detection — when informed traders clear levels
Queue Imbalance (Q_i):
Q_i = (BidSize_i - AskSize_i) / (BidSize_i + AskSize_i)
Weighted Queue Imbalance (WQI):
WQI = Σ w_i · Q_i where w_i decays with distance from mid
When WQI > 0: buying pressure → expected price increase
When WQI < 0: selling pressure → expected price decrease
Signal strength from queue dynamics is proportional to how
extreme the imbalance is relative to historical norms.
Usage:
from strategies.queue_imbalance import QueueImbalance
qi = QueueImbalance()
signal = qi.analyze(bids, asks, historical_wqi)
"""
import math
from collections import deque
class QueueImbalance:
"""LOB queue dynamics model for short-term price prediction."""
def __init__(self, depth_levels: int = 10):
self.depth_levels = depth_levels
self.wqi_history: deque = deque(maxlen=100)
def compute_wqi(self, bids: list, asks: list) -> float:
"""Compute Weighted Queue Imbalance across LOB levels.
Weights decay exponentially: w_i = e^(-i/3) for level i.
This gives 3x more weight to top-of-book than 3 levels deep.
"""
if not bids or not asks:
return 0.0
wqi = 0.0
total_weight = 0.0
for i in range(min(len(bids), len(asks), self.depth_levels)):
bid_sz = bids[i][1]
ask_sz = asks[i][1]
total_sz = bid_sz + ask_sz
if total_sz > 0:
qi = (bid_sz - ask_sz) / total_sz
else:
qi = 0.0
# Exponential decay weight
weight = math.exp(-i / 3.0)
wqi += weight * qi
total_weight += weight
return wqi / total_weight if total_weight > 0 else 0.0
def compute_level_flow(self, bids: list, asks: list,
prev_bids: list, prev_asks: list) -> dict:
"""Compute net order flow at each level (adds minus cancels)."""
flow = {"bid_flow": 0.0, "ask_flow": 0.0, "net_flow": 0.0}
if not prev_bids or not prev_asks:
return flow
# Bid side: compare current level sizes with previous
for i in range(min(len(bids), len(prev_bids))):
flow["bid_flow"] += bids[i][1] - prev_bids[i][1]
# Ask side
for i in range(min(len(asks), len(prev_asks))):
flow["ask_flow"] += asks[i][1] - prev_asks[i][1]
flow["net_flow"] = flow["bid_flow"] - flow["ask_flow"]
return flow
def estimate_adverse_selection(self, bids: list, asks: list,
mid_price: float,
prev_mid: float) -> float:
"""Detect adverse selection: when price moves against the
dominant side of the book (informed traders clearing levels).
Returns 0-1 score where 1 = high adverse selection risk.
"""
if prev_mid <= 0 or mid_price <= 0:
return 0.0
# Compute which side was dominant in the previous tick
wqi = self.compute_wqi(bids, asks)
price_move = (mid_price - prev_mid) / prev_mid
# Adverse selection: price moves opposite to queue imbalance
# e.g., bids dominant (WQI > 0) but price goes down
if wqi > 0.1 and price_move < -0.0005:
return min(abs(price_move) * 1000, 1.0)
elif wqi < -0.1 and price_move > 0.0005:
return min(abs(price_move) * 1000, 1.0)
return 0.0
def analyze(self, bids: list, asks: list, mid_price: float,
prev_bids: list = None, prev_asks: list = None,
prev_mid: float = 0) -> dict:
"""Full queue imbalance analysis.
Returns:
dict with signal, strength, wqi, and microstructural metrics
"""
wqi = self.compute_wqi(bids, asks)
self.wqi_history.append(wqi)
# Compute WQI z-score
if len(self.wqi_history) >= 20:
history = list(self.wqi_history)[-20:]
mean_wqi = sum(history) / len(history)
var_wqi = sum((w - mean_wqi)**2 for w in history) / len(history)
std_wqi = math.sqrt(var_wqi) if var_wqi > 0 else 0.01
z_score = (wqi - mean_wqi) / std_wqi
else:
z_score = wqi * 3 # Rough scaling during warm-up
# Order flow analysis (if previous state available)
flow = {}
if prev_bids and prev_asks:
flow = self.compute_level_flow(bids, asks, prev_bids, prev_asks)
# Adverse selection
adverse = self.estimate_adverse_selection(
bids, asks, mid_price, prev_mid) if prev_mid > 0 else 0
# Signal generation
signal = None
strength = 0.0
z_threshold = 1.2
wqi_threshold = 0.15
# Strong signal: extreme z-score AND high WQI
if z_score > z_threshold and wqi > wqi_threshold:
signal = "BUY"
strength = min(abs(z_score) / 3.0, 1.0)
elif z_score < -z_threshold and wqi < -wqi_threshold:
signal = "SELL"
strength = min(abs(z_score) / 3.0, 1.0)
# Reduce confidence if adverse selection detected
if adverse > 0.3:
strength *= (1.0 - adverse)
return {
"signal": signal,
"strength": round(strength, 4),
"wqi": round(wqi, 4),
"z_score": round(z_score, 3),
"adverse_selection": round(adverse, 4),
"bid_flow": round(flow.get("bid_flow", 0), 4),
"ask_flow": round(flow.get("ask_flow", 0), 4),
"net_flow": round(flow.get("net_flow", 0), 4),
}