Coverage for adrs/oms/risk.py: 100%

1import logging 

2 

3from decimal import Decimal 

4from collections.abc import Awaitable, Callable 

5 

6from pydantic import BaseModel 

7 

8from adrs.oms.config import ConfigManager 

9from adrs.oms.position import PositionManager, Positions 

10from adrs.oms.rate_limit.rate_limiter import RateLimiter 

11 

12logger = logging.getLogger(__name__) 

13 

14 

15class RiskConfig(BaseModel): 

16 nav_floor: Decimal = Decimal("0.60") # margin_balance / initial kill level 

17 max_exposure: Decimal = Decimal("2.0") # committed notional / initial, 2.0 = 200% 

18 

19 

20class RiskEngine: 

21 """ 

22 Base risk engine: every hook is a no-op. The OMS calls these at fixed 

23 lifecycle points; subclass and override the hooks to add real controls 

24 without touching the OMS flow. 

25 

26 Hooks: 

27 - cap_desired: pre-placement gate. Return desired unchanged to allow, or 

28 a scaled/rewritten mapping to constrain exposure. 

29 - run_risk_checks: monitor, run on a cron. Call self.on_breach() to shut 

30 the OMS down (same path as SIGINT/SIGTERM). 

31 """ 

32 

33 def __init__( 

34 self, 

35 config: ConfigManager, 

36 position: PositionManager, 

37 rate_limiter: RateLimiter, 

38 on_breach: Callable[[], Awaitable[None]], 

39 risk: RiskConfig | None = None, 

40 ): 

41 self.config = config 

42 self.position = position 

43 self.rate_limiter = rate_limiter 

44 self.on_breach = on_breach # OMS shutdown, same path as SIGINT/SIGTERM 

45 self.risk = risk or RiskConfig() 

46 

47 def cap_desired(self, desired: Positions) -> Positions: 

48 """No-op exposure gate; override to constrain desired positions.""" 

49 return desired 

50 

51 async def run_risk_checks(self): 

52 """No-op monitor; override to check NAV/drawdown and trigger on_breach.""" 

53 return