← Back to all products

Depeg Detector

$10

Real-time stablecoin depeg detection with automated alerts and coverage eligibility checks.

📁 12 files
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

SymbolPegWarningCritical

... 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 ...
Buy Now — $10 Back to Products