← Back to all products
$10
Depeg Detector
Real-time stablecoin depeg detection with automated alerts and coverage eligibility checks.
TOMLPythonMarkdown
📁 File Structure 12 files
depeg-detector/
├── LICENSE
├── README.md
├── pyproject.toml
├── security-notes.md
├── src/
│ ├── __init__.py
│ ├── alerts.py
│ ├── config.py
│ ├── detector.py
│ ├── feeds.py
│ ├── formatters.py
│ └── models.py
└── tests/
└── test_detector.py
📖 Documentation Preview README excerpt
Depeg Detector — CryptoForge Coverage Vault
Real-time stablecoin depeg monitoring with configurable triggers, multiple price feed sources, and alert dispatching.
Features
- Multi-stablecoin monitoring — USDC, USDT, DAI, FRAX, LUSD, crvUSD out of the box
- Configurable thresholds — Warning and critical severity levels per stablecoin
- Recovery detection — Tracks when depegs resolve and logs duration
- Pluggable price feeds — Chainlink, CoinGecko, mock, and aggregated (median)
- Alert management — Console, webhook, and composite sinks with cooldown deduplication
- Dashboard view — Multi-line terminal dashboard for all monitored stablecoins
Installation
pip install -e .
Quick Start
import asyncio
from depeg_detector import DepegDetector, MonitorConfig, MockFeed, format_dashboard
async def main():
config = MonitorConfig()
feed = MockFeed({"USDC": 0.994, "USDT": 1.0, "DAI": 1.001})
detector = DepegDetector(config, feed)
states = await detector.check_all()
print(format_dashboard(states))
asyncio.run(main())
Configuration
from depeg_detector import StablecoinConfig, MonitorConfig
config = MonitorConfig(
stablecoins=[
StablecoinConfig(
symbol="USDC",
expected_peg=1.0,
warning_threshold_bps=50, # 0.50% deviation
critical_threshold_bps=200, # 2.00% deviation
recovery_threshold_bps=25, # 0.25% to recover
),
],
poll_interval_seconds=10,
alert_cooldown_seconds=300,
)
Supported Stablecoins
| Symbol | Peg | Warning | Critical |
|---|
... continues with setup instructions, usage examples, and more.
📄 Code Sample .py preview
src/alerts.py
"""
Depeg Detector — Alert Dispatching
════════════════════════════════════
Alert sinks and cooldown management.
"""
from __future__ import annotations
import json
import time
from abc import ABC, abstractmethod
from typing import Optional
from .formatters import format_depeg_event, to_json
from .models import DepegEvent
class AlertSink(ABC):
"""Abstract base class for alert destinations."""
@abstractmethod
async def send(self, event: DepegEvent) -> None:
"""Dispatch an alert for a depeg event."""
class ConsoleAlertSink(AlertSink):
"""Prints formatted alerts to stdout."""
async def send(self, event: DepegEvent) -> None:
print(format_depeg_event(event))
class WebhookAlertSink(AlertSink):
"""
POSTs JSON alerts to a webhook URL.
In production, uses aiohttp to POST the alert payload.
"""
def __init__(self, url: str, headers: Optional[dict[str, str]] = None) -> None:
self._url = url
self._headers = headers or {"Content-Type": "application/json"}
async def send(self, event: DepegEvent) -> None:
payload = json.dumps(to_json(event))
# In production:
# async with aiohttp.ClientSession() as session:
# await session.post(self._url, data=payload, headers=self._headers)
#
# For now, log the intent
# ... 77 more lines ...