Oracle Monitor

$10

Monitor oracle health — heartbeat status, price deviation, gas costs, and node uptime.

📁 16 files

TOMLPythonConfigMarkdownYAML

📁 File Structure 16 files

oracle-monitor/ ├── LICENSE ├── README.md ├── config/ │ └── feeds.example.yaml ├── pyproject.toml ├── security-notes.md ├── src/ │ └── oracle_monitor/ │ ├── __init__.py │ ├── alerts/ │ │ ├── __init__.py │ │ ├── email_alert.py │ │ └── webhook.py │ ├── detectors/ │ │ ├── __init__.py │ │ ├── deviation.py │ │ └── staleness.py │ ├── monitor.py │ └── types.py └── tests/ └── test_monitor.py

📖 Documentation Preview README excerpt

Oracle Monitor

CryptoForge Oracle Forge — Product #3

Real-time oracle feed monitoring with staleness and deviation alerts.

Overview

A Python monitoring service that watches oracle price feeds for staleness, deviation anomalies, and downtime. Configurable alert channels including webhooks (Discord/Slack) and email.

Architecture


Oracle Feeds (on-chain) → PriceMonitor (async polling)
                              ↓
                    StalenessDetector + DeviationDetector
                              ↓
                    AlertDispatcher → Webhook / Email

Install


pip install -e .

# With dev dependencies
pip install -e ".[dev]"

Usage


import asyncio
from oracle_monitor import OracleMonitor
from oracle_monitor.types import FeedConfig, AlertConfig

feeds = [
    FeedConfig(
        address="0x5f4eC3Df9cbd43714FE2740f5E3616155c5b8419",
        name="ETH/USD",
        heartbeat=3600,
        decimals=8,
        deviation_threshold_bps=500,  # 5%
    ),
]

alerts = AlertConfig(
    webhook_url="https://discord.com/api/webhooks/...",
    smtp_host="smtp.gmail.com",
    email_to="alerts@example.com",
)

monitor = OracleMonitor(
    rpc_url="https://eth.llamarpc.com",
    feeds=feeds,
    alerts=alerts,
    poll_interval=60,
)

asyncio.run(monitor.start())

... continues with setup instructions, usage examples, and more.

📄 Code Sample .py preview

src/oracle_monitor/monitor.py # ═══════════════════════════════════════════════════════════════════════ # Oracle Monitor — Main Monitor # CryptoForge Oracle Forge # ═══════════════════════════════════════════════════════════════════════ from __future__ import annotations import asyncio from datetime import datetime import structlog from web3 import AsyncWeb3, AsyncHTTPProvider from .types import ( AlertConfig, AlertEvent, AlertSeverity, AlertType, FeedConfig, FeedSnapshot, MonitorState, ) from .detectors.staleness import StalenessDetector from .detectors.deviation import DeviationDetector from .alerts.webhook import WebhookAlerter from .alerts.email_alert import EmailAlerter logger = structlog.get_logger() # ── Chainlink AggregatorV3 ABI (minimal) ────────────────────────────── AGGREGATOR_ABI = [ { "inputs": [], "name": "latestRoundData", "outputs": [ {"name": "roundId", "type": "uint80"}, {"name": "answer", "type": "int256"}, {"name": "startedAt", "type": "uint256"}, {"name": "updatedAt", "type": "uint256"}, {"name": "answeredInRound", "type": "uint80"}, ], "stateMutability": "view", "type": "function", } ] class OracleMonitor: """Async oracle feed monitor with pluggable detectors and alerters.""" # ... 140 more lines ...

Buy Now — $10 Back to Products