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/.
37 lines
1.1 KiB
Python
37 lines
1.1 KiB
Python
"""
|
|
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
|