zwitschi
c17f41aaf8
- Introduced new tables for audit events and runtime state snapshots in the database schema. - Created data classes for AuditRecord and RuntimeStateRecord to represent the new entities. - Implemented AuditRepository and RuntimeStateRepository for inserting and retrieving records. - Enhanced the dashboard to include an audit trail section, displaying recent audit events. - Added tests for the new audit repository and runtime lifecycle functionalities. - Updated settings validation to ensure proper configuration for alerting features. - Integrated alert notifications across various components, including execution sequencer and loss limits.
91 lines
2.6 KiB
Python
91 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from arbitrade.risk.loss_limits import LossLimitGuard
|
|
|
|
|
|
class _FakeAlertNotifier:
|
|
def __init__(self) -> None:
|
|
self.events: list[dict[str, Any]] = []
|
|
|
|
async def notify(
|
|
self,
|
|
*,
|
|
category: str,
|
|
severity: str,
|
|
title: str,
|
|
message: str,
|
|
details: dict[str, str] | None = None,
|
|
) -> bool:
|
|
self.events.append(
|
|
{
|
|
"category": category,
|
|
"severity": severity,
|
|
"title": title,
|
|
"message": message,
|
|
"details": details or {},
|
|
}
|
|
)
|
|
return True
|
|
|
|
|
|
def test_loss_limit_guard_tracks_daily_and_cumulative_pnl() -> None:
|
|
guard = LossLimitGuard(daily_loss_limit=100.0, cumulative_loss_limit=200.0)
|
|
t0 = datetime.now(UTC)
|
|
|
|
guard.register_realized_pnl(-40.0, at=t0)
|
|
guard.register_realized_pnl(10.0, at=t0)
|
|
|
|
assert guard.cumulative_pnl == -30.0
|
|
assert guard.daily_pnl(t0.date()) == -30.0
|
|
assert not guard.is_halted
|
|
|
|
|
|
def test_loss_limit_guard_halts_on_daily_limit() -> None:
|
|
guard = LossLimitGuard(daily_loss_limit=50.0)
|
|
t0 = datetime.now(UTC)
|
|
|
|
guard.register_realized_pnl(-30.0, at=t0)
|
|
guard.register_realized_pnl(-25.0, at=t0)
|
|
|
|
assert guard.is_halted
|
|
assert guard.halted_reason == "daily_loss_limit_breached"
|
|
|
|
|
|
def test_loss_limit_guard_halts_on_cumulative_limit_across_days() -> None:
|
|
guard = LossLimitGuard(cumulative_loss_limit=60.0)
|
|
t0 = datetime.now(UTC)
|
|
|
|
guard.register_realized_pnl(-40.0, at=t0)
|
|
guard.register_realized_pnl(-25.0, at=t0 + timedelta(days=1))
|
|
|
|
assert guard.is_halted
|
|
assert guard.halted_reason == "cumulative_loss_limit_breached"
|
|
|
|
|
|
def test_loss_limit_guard_rejects_invalid_limits() -> None:
|
|
with pytest.raises(ValueError, match="daily_loss_limit"):
|
|
LossLimitGuard(daily_loss_limit=0.0)
|
|
|
|
with pytest.raises(ValueError, match="cumulative_loss_limit"):
|
|
LossLimitGuard(cumulative_loss_limit=-1.0)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_loss_limit_guard_emits_alert_on_breach() -> None:
|
|
notifier = _FakeAlertNotifier()
|
|
guard = LossLimitGuard(daily_loss_limit=50.0, alert_notifier=notifier)
|
|
|
|
guard.register_realized_pnl(-60.0, at=datetime.now(UTC))
|
|
await asyncio.sleep(0)
|
|
|
|
assert guard.is_halted
|
|
assert len(notifier.events) == 1
|
|
assert notifier.events[0]["category"] == "threshold"
|
|
assert notifier.events[0]["title"] == "Daily loss limit breached"
|