ramseshk
b59dcc3629
Set up the directory structure and wrote placeholder logic for: - Order Book Imbalance: trades on L2 bid/ask skew - Iceberg/TWAP detection: follows whale accumulation patterns - Funding rate arbitrage: delta-neutral carry on perp funding - Pairs trading: BTC/ETH spread mean reversion - Avellaneda-Stoikov market making: optimal bid/ask quoting Also added shared risk manager, portfolio tracker, and a plain-language strategy walkthrough in docs/.
49 lines
1.4 KiB
Python
49 lines
1.4 KiB
Python
"""
|
|
Shared risk manager.
|
|
|
|
Tracks exposure per-strategy and blocks orders that would
|
|
exceed position limits, drawdown limits, or daily trade caps.
|
|
"""
|
|
from dataclasses import dataclass
|
|
|
|
|
|
@dataclass
|
|
class RiskLimits:
|
|
max_position: float = 0.01
|
|
max_drawdown_pct: float = 0.05
|
|
max_daily_trades: int = 50
|
|
max_leverage: float = 2.0
|
|
|
|
|
|
class RiskManager:
|
|
def __init__(self) -> None:
|
|
self.strategy_limits: dict[str, RiskLimits] = {}
|
|
self.daily_trades: dict[str, int] = {}
|
|
self.peak_equity: float = 0.0
|
|
|
|
def register(self, name: str, limits: RiskLimits) -> None:
|
|
self.strategy_limits[name] = limits
|
|
self.daily_trades[name] = 0
|
|
|
|
def can_trade(self, name: str, position: float, equity: float) -> bool:
|
|
limits = self.strategy_limits.get(name)
|
|
if not limits:
|
|
return True
|
|
|
|
if abs(position) >= limits.max_position:
|
|
return False
|
|
if self.daily_trades.get(name, 0) >= limits.max_daily_trades:
|
|
return False
|
|
if self.peak_equity > 0:
|
|
dd = 1 - (equity / self.peak_equity)
|
|
if dd >= limits.max_drawdown_pct:
|
|
return False
|
|
return True
|
|
|
|
def record_trade(self, name: str) -> None:
|
|
self.daily_trades[name] = self.daily_trades.get(name, 0) + 1
|
|
|
|
def update_equity(self, equity: float) -> None:
|
|
if equity > self.peak_equity:
|
|
self.peak_equity = equity
|