feat: Phase 4 — controlled strategy deployment module + 38 tests
New live/ sub-modules for production-ready market making:
live/filters/toxicity.py (ToxicityFilter):
VPIN-based pre-trade filter. Accumulates buy/sell volume, computes
VPIN via microstructure module, produces quoting decision:
- allow_quoting: bool
- size_multiplier: 0.0–1.0 (graduated reduction approaching alarm)
- granular thresholds (threshold vs alarm) with smooth reduction
live/treasury.py (Treasury):
Central capital/risk management — single source of truth:
- Position tracking per coin (opening, closing, average entry)
- Realized + unrealized PnL computation
- Pre-trade constraint checks (inventory limits, fee estimates)
- Circuit breaker (drawdown, trade count, toxic fill rate, API errors)
- Liquidation distance monitoring
- Automatic cooldown reset after trip expiry
live/makers/hl_btc_eth.py:
HlMaker — per-coin market maker integrating:
- AvellanedaStoikovMaker (Phase 3) for optimal quotes
- ToxicityFilter for pre-trade gating
- Treasury for position/risk checks
HlMakerPool — manages multiple HlMaker instances with shared treasury
and coordinated observe_all()/quote_all()
live/monitors/cross_venue.py (CrossVenueMonitor):
Cross-exchange lead-lag detection via cross-correlation at multiple
lags. Spot premium (basis proxy) computation. Multi-venue summary.
live/monitors/funding_basis.py (FundingBasisMonitor):
Funding regime classification, momentum detection, carry PnL
estimation, basis spread analysis. Uses microstructure/funding.py.
live/monitors/liq_risk.py (LiquidationRiskOverlay):
Per-position liquidation distance monitoring with tiered warnings
(safe/warning/danger/critical). Recommended position reduction.
38 tests across 4 files (all pass):
test_live_filters.py (5)
test_live_maker.py (9)
test_live_monitors.py (11)
test_live_treasury.py (13)
Total test suite: 172 tests, all passing.
This commit is contained in:
@@ -0,0 +1,109 @@
|
||||
"""
|
||||
Liquidation risk overlay.
|
||||
|
||||
Monitors current positions, mark prices, and computes liquidation
|
||||
distance. Warns when positions approach liquidation threshold.
|
||||
|
||||
Integrates with live/treasury.py for position tracking.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Optional
|
||||
|
||||
from live.treasury import Treasury
|
||||
|
||||
|
||||
class LiquidationRiskOverlay:
|
||||
"""Liquidation risk monitor for open positions.
|
||||
|
||||
Usage:
|
||||
overlay = LiquidationRiskOverlay(treasury=treasury)
|
||||
status = overlay.check("BTC")
|
||||
if status["warning"]:
|
||||
# reduce position or add margin
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
treasury: Treasury,
|
||||
warning_threshold_pct: float = 10.0,
|
||||
danger_threshold_pct: float = 5.0,
|
||||
critical_threshold_pct: float = 2.5,
|
||||
):
|
||||
self._treasury = treasury
|
||||
self._warning = warning_threshold_pct
|
||||
self._danger = danger_threshold_pct
|
||||
self._critical = critical_threshold_pct
|
||||
|
||||
def check(self, coin: str) -> dict:
|
||||
"""Check liquidation safety for a specific coin."""
|
||||
distance = self._treasury.liquidation_distance(coin.upper())
|
||||
|
||||
if distance >= self._warning or distance >= 1e9:
|
||||
level = "safe"
|
||||
elif distance >= self._danger:
|
||||
level = "warning"
|
||||
elif distance >= self._critical:
|
||||
level = "danger"
|
||||
else:
|
||||
level = "critical"
|
||||
|
||||
return {
|
||||
"coin": coin.upper(),
|
||||
"level": level,
|
||||
"distance_pct": round(min(distance, 999999), 2),
|
||||
"position": self._treasury.position(coin.upper()),
|
||||
"warning": level in ("warning", "danger", "critical"),
|
||||
"needs_action": level == "critical",
|
||||
}
|
||||
|
||||
def check_all(self) -> dict[str, dict]:
|
||||
positions = self._treasury.all_positions
|
||||
return {coin: self.check(coin) for coin, pos in positions.items() if abs(pos) > 0}
|
||||
|
||||
def pnl_at_liquidation(self, coin: str) -> float:
|
||||
"""Estimate realized PnL if position reaches liquidation price."""
|
||||
pos = self._treasury._positions.get(coin.upper())
|
||||
if not pos:
|
||||
return 0.0
|
||||
|
||||
entry = pos["entry_px"]
|
||||
size = pos["size"]
|
||||
side = pos["side"]
|
||||
|
||||
liq_price = self._treasury._liquidation.liquidation_price(
|
||||
entry, size, side, self._treasury.equity
|
||||
)
|
||||
|
||||
if side == "buy":
|
||||
return size * (liq_price - entry)
|
||||
else:
|
||||
return size * (entry - liq_price)
|
||||
|
||||
def recommended_action(self, coin: str) -> str:
|
||||
"""Recommend action based on liquidation distance."""
|
||||
status = self.check(coin)
|
||||
level = status["level"]
|
||||
|
||||
if level == "safe":
|
||||
return "none"
|
||||
elif level == "warning":
|
||||
return "reduce_position_25pct"
|
||||
elif level == "danger":
|
||||
return "reduce_position_50pct"
|
||||
else:
|
||||
return "close_all"
|
||||
|
||||
def summary(self) -> dict:
|
||||
return {
|
||||
"positions": self.check_all(),
|
||||
"worst_case": min(
|
||||
(self.check(c)["distance_pct"] for c in self._treasury.all_positions),
|
||||
default=float("inf")
|
||||
),
|
||||
"any_critical": any(
|
||||
self.check(c)["level"] == "critical"
|
||||
for c in self._treasury.all_positions
|
||||
),
|
||||
}
|
||||
Reference in New Issue
Block a user