← Back to all products
$8
Deviation Alerter
Alert system for abnormal price deviations, stale feeds, and oracle malfunction detection.
TOMLPythonConfigMarkdownYAML
📁 File Structure 16 files
deviation-alerter/
├── LICENSE
├── README.md
├── config/
│ └── alerter.example.yaml
├── pyproject.toml
├── security-notes.md
├── src/
│ └── deviation_alerter/
│ ├── __init__.py
│ ├── alerts/
│ │ ├── __init__.py
│ │ ├── channels.py
│ │ └── dispatcher.py
│ ├── monitors/
│ │ ├── __init__.py
│ │ ├── correlation.py
│ │ └── price_monitor.py
│ └── types.py
└── tests/
├── test_correlation.py
└── test_price_monitor.py
📖 Documentation Preview README excerpt
Deviation Alerter
Real-time oracle price deviation monitoring and alerting system. Watches multiple oracle feeds for abnormal price movements, cross-feed divergence, and correlation breakdowns, then dispatches alerts via Telegram, Discord, PagerDuty, and email.
Features
- Price deviation monitoring — Configurable thresholds per-feed (absolute + percentage)
- Cross-feed correlation — Detect when normally-correlated feeds diverge
- Multi-channel alerts — Telegram, Discord webhook, PagerDuty, SMTP email
- Cooldown management — Prevent alert spam with configurable cooldowns
- Historical tracking — Store deviation events for post-mortem analysis
- Async architecture — Non-blocking monitoring via asyncio + aiohttp
- Structured logging — Full observability via structlog
Architecture
src/deviation_alerter/
├── __init__.py Package init + version
├── types.py Pydantic models (FeedConfig, Alert, etc.)
├── monitors/
│ ├── __init__.py Monitor exports
│ ├── price_monitor.py Per-feed deviation detector
│ └── correlation.py Cross-feed correlation tracker
├── alerts/
│ ├── __init__.py Alert exports
│ ├── dispatcher.py Alert routing + cooldown logic
│ └── channels.py Channel implementations (Telegram, Discord, etc.)
tests/
├── test_price_monitor.py Price monitor unit tests
└── test_correlation.py Correlation tracker unit tests
config/
├── alerter.example.yaml Example configuration
└── .env.example Environment variable template
Quick Start
pip install -e .
# Configure feeds and thresholds
cp config/alerter.example.yaml config/alerter.yaml
cp config/.env.example config/.env
# Run
python -m deviation_alerter --config config/alerter.yaml
Configuration
feeds:
- name: "ETH/USD Chainlink"
source: "chainlink"
address: "0x5f4eC3Df9cbd43714FE2740f5E3616155c5b8419"
deviation_pct: 2.0
deviation_abs: 50.0
poll_interval: 30
*... continues with setup instructions, usage examples, and more.*
📄 Code Sample .py preview
src/deviation_alerter/types.py
# ═══════════════════════════════════════════════════════════════════════════
# types.py — Pydantic models for deviation alerter
# ═══════════════════════════════════════════════════════════════════════════
#
# Core data types used throughout the deviation alerter system.
# All models use Pydantic v2 for validation and serialization.
#
# Author: CryptoForge Team
# License: MIT
# ═══════════════════════════════════════════════════════════════════════════
from __future__ import annotations
from datetime import datetime
from enum import Enum
from typing import Any
from pydantic import BaseModel, Field, field_validator
class AlertSeverity(str, Enum):
"""Alert severity levels."""
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
EMERGENCY = "emergency"
class FeedSource(str, Enum):
"""Supported oracle feed sources."""
CHAINLINK = "chainlink"
PYTH = "pyth"
UNISWAP_V3 = "uniswap_v3"
CUSTOM = "custom"
class FeedConfig(BaseModel):
"""Configuration for a single oracle feed to monitor."""
name: str = Field(..., description="Human-readable feed name (e.g. 'ETH/USD Chainlink')")
source: FeedSource = Field(..., description="Oracle source type")
address: str = Field(..., description="Feed contract address")
chain_id: int = Field(default=1, description="Chain ID (1 = Ethereum mainnet)")
decimals: int = Field(default=8, ge=0, le=18, description="Feed decimal precision")
deviation_pct: float = Field(
default=2.0, gt=0, description="Percentage deviation threshold to trigger alert"
)
deviation_abs: float = Field(
# ... 134 more lines ...