uv venv
source .venv/bin/activate
uv pip install -e ".[dev]"
cp .env.example .env # then edit with your API keysuv run ruff format . # format code
uv run ruff check --fix . # lint + autofix
uv run pytest tests/ # run testsEach protocol lives in its own directory with a main.py entry point:
protocol-name/
main.py # entry point
abi/ # contract ABIs (if needed)
README.md # protocol-specific docs
Shared utilities live in utils/:
| Module | Purpose |
|---|---|
utils/logging.py |
Structured logging via get_logger(name) |
utils/telegram.py |
Telegram alert delivery |
utils/cache.py |
File-based key:value persistence |
utils/web3_wrapper.py |
Web3 connection management (ChainManager) |
utils/config.py |
Environment config (Config) |
utils/formatting.py |
Number formatting helpers (format_usd, format_token_amount) |
utils/http.py |
HTTP request helper (fetch_json) |
utils/chains.py |
Chain enum and explorer URLs |
utils/abi.py |
ABI loader |
utils/gauntlet.py |
Gauntlet risk parameter helpers |
- Line length: 120 characters (configured in
pyproject.toml) - Python version: 3.10+
- Formatter/linter: ruff (run both format and check before committing)
- Naming:
snake_casefor functions/variables,UPPER_CASEfor constants,PascalCasefor classes
Use structured logging everywhere — never use print().
from utils.logging import get_logger
logger = get_logger("protocol_name")
# Use %s formatting (not f-strings) for lazy evaluation
logger.info("Processing %s assets on %s", len(assets), chain.name)
logger.error("Failed to fetch data: %s", error)Each monitor defines a lowercase PROTOCOL constant used for telegram credential lookup and cache keys:
PROTOCOL = "aave" # lowercase — telegram.py calls .upper() internally
logger = get_logger(PROTOCOL)Add type annotations to all function signatures:
def process_assets(chain: Chain, threshold: float = 0.99) -> None:
...
def get_price(token_address: str) -> float | None:
...Use send_telegram_message from utils/telegram.py. It looks up TELEGRAM_BOT_TOKEN_{PROTOCOL.upper()} and TELEGRAM_CHAT_ID_{PROTOCOL.upper()} from environment variables:
from utils.telegram import send_telegram_message
send_telegram_message("Alert text here", PROTOCOL)Use ChainManager for connections and batch requests whenever possible:
from utils.chains import Chain
from utils.web3_wrapper import ChainManager
client = ChainManager.get_client(Chain.MAINNET)
with client.batch_requests() as batch:
batch.add(contract.functions.totalSupply())
batch.add(contract.functions.balanceOf(address))
responses = client.execute_batch(batch)Use utils/cache.py for persisting state between runs (e.g. last processed timestamp or proposal ID):
from utils.cache import cache_filename, get_last_value_for_key_from_file, write_last_value_to_file
value = get_last_value_for_key_from_file(cache_filename, "MY_KEY")
write_last_value_to_file(cache_filename, "MY_KEY", new_value)- Create
protocol-name/main.pyfollowing the pattern above - Add a
protocol-name/README.mddescribing what it monitors - Add
"protocol-name"to thepackageslist inpyproject.tomlunder[tool.setuptools] - Add the corresponding
TELEGRAM_BOT_TOKEN_*andTELEGRAM_CHAT_ID_*entries to.env.example - Add the protocol to the CI workflow if it should run on a schedule
Tests live in tests/. Run them with:
uv run pytest tests/When writing tests, patch utils.telegram.logger (not builtins.print) for assertion on log output.