Initial project scaffold: five quant strategies for Hyperliquid Testnet

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/.

common/__init__.py

@@ -0,0 +1 @@
+# Common utilities package

common/metrics.py

@@ -0,0 +1,43 @@
+"""
+Performance metrics.
+
+Sharpe ratio, Sortino ratio, max drawdown, win rate.
+Standard toolbox for evaluating a trading strategy.
+"""
+import numpy as np
+
+
+def sharpe(returns: list[float], rf: float = 0.0, periods: int = 365) -> float:
+    if len(returns) < 2:
+        return 0.0
+    excess = np.mean(returns) - rf
+    std = np.std(returns, ddof=1)
+    return (excess / std) * np.sqrt(periods) if std > 0 else 0.0
+
+
+def sortino(returns: list[float], rf: float = 0.0, periods: int = 365) -> float:
+    if len(returns) < 2:
+        return 0.0
+    excess = np.mean(returns) - rf
+    downside = [r for r in returns if r < 0]
+    d_std = np.std(downside, ddof=1) if downside else 0.0
+    return (excess / d_std) * np.sqrt(periods) if d_std > 0 else 0.0
+
+
+def max_drawdown(equity: list[float]) -> float:
+    if not equity:
+        return 0.0
+    peak = equity[0]
+    worst = 0.0
+    for v in equity:
+        if v > peak:
+            peak = v
+        dd = (peak - v) / peak if peak > 0 else 0.0
+        worst = max(worst, dd)
+    return worst
+
+
+def win_rate(trades: list[dict]) -> float:
+    if not trades:
+        return 0.0
+    return sum(1 for t in trades if t.get("pnl", 0) > 0) / len(trades)

common/portfolio.py

@@ -0,0 +1,36 @@
+"""
+Portfolio tracker.
+
+Aggregates positions from all running strategies to prevent
+over-concentration in any single instrument.
+"""
+from dataclasses import dataclass
+
+
+@dataclass
+class Position:
+    instrument: str
+    quantity: float
+    entry_price: float
+    strategy: str
+
+
+class PortfolioTracker:
+    def __init__(self) -> None:
+        self.positions: dict[str, list[Position]] = {}
+
+    def add(self, strategy: str, instrument: str, qty: float, price: float) -> None:
+        if instrument not in self.positions:
+            self.positions[instrument] = []
+        self.positions[instrument].append(Position(instrument, qty, price, strategy))
+
+    def net_exposure(self, instrument: str) -> float:
+        if instrument not in self.positions:
+            return 0.0
+        return sum(p.quantity for p in self.positions[instrument])
+
+    def all_exposures(self) -> dict[str, float]:
+        return {inst: self.net_exposure(inst) for inst in self.positions}
+
+    def is_overconcentrated(self, instrument: str, max_pct: float, equity: float) -> bool:
+        return abs(self.net_exposure(instrument)) > equity * max_pct

common/risk_manager.py

@@ -0,0 +1,48 @@
+"""
+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