Compare commits

..

4 Commits

Author SHA1 Message Date
7bd73cae08 add etherscan api key as optional parameter 2024-06-08 14:27:18 -04:00
280a5f830f update README.md 2024-06-08 14:19:09 -04:00
a3fda36d43 add database support 2024-06-08 14:16:53 -04:00
6510eab5a8 readability update 2024-06-08 13:15:49 -04:00
7 changed files with 142 additions and 17 deletions

1
.gitignore vendored
View File

@ -2,4 +2,5 @@
venv/
__pycache__/
*.py[cod]
*.db
config.json

View File

@ -1,5 +1,9 @@
# mbot
Simple Matrix bot that will eventually send Web3 `PairCreation` event alerts to the Juggalol Homeserver.
Uniswap PairCreation monitor that sends alerts to Matrix.
## Requirements
- Infura API URL
- Matrix bot account
## Configuration
```json
@ -11,6 +15,6 @@ Simple Matrix bot that will eventually send Web3 `PairCreation` event alerts to
"room_id": ""
},
"infura_url": "",
"pool_address": ""
"pool_address": "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f"
}
```

49
db.py Normal file
View File

@ -0,0 +1,49 @@
from sqlalchemy import create_engine, Column, String, BigInteger, ForeignKey, TIMESTAMP, UniqueConstraint
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
from datetime import datetime
Base = declarative_base()
class Token(Base):
__tablename__ = 'tokens'
id = Column(String(42), primary_key=True)
name = Column(String(255), nullable=True)
symbol = Column(String(32), nullable=True)
total_supply = Column(String, nullable=True)
class Pair(Base):
__tablename__ = 'pairs'
id = Column(String(42), primary_key=True)
token0 = Column(String(42), ForeignKey('tokens.id'), nullable=False)
token1 = Column(String(42), ForeignKey('tokens.id'), nullable=False)
created_at = Column(TIMESTAMP, default=datetime.utcnow)
# Ensuring the combination of token0, token1, and pool_address is unique
__table_args__ = (UniqueConstraint('token0', 'token1', name='_token_pool_uc'),)
# Relationships
token0_rel = relationship("Token", foreign_keys=[token0])
token1_rel = relationship("Token", foreign_keys=[token1])
# Engine creation
engine = create_engine('sqlite:///uniswap.db', echo=True)
# Create tables
Base.metadata.create_all(engine)
# Create a session factory
SessionLocal = sessionmaker(bind=engine)
# Dependency to get a database session
def get_session():
session = SessionLocal()
try:
yield session
finally:
session.close()

View File

@ -19,7 +19,11 @@ async def main():
queue = Queue()
bot = MatrixBot(config['matrix'])
monitor = EventMonitor(
config={"infura_url": config['infura_url'], "pool_address": config['pool_address']},
config={
"infura_url": config['infura_url'],
"pool_address": config['pool_address'],
"etherscan_key": config['etherscan_key']
},
queue=queue
)
monitor_thread = threading.Thread(target=monitor.log_loop, args=(15,))

View File

@ -3,6 +3,7 @@ from queue import Queue
from loguru import logger
from web3 import Web3
from util import fetch_abi
from db import Token, Pair, get_session
class EventMonitor:
@ -14,7 +15,7 @@ class EventMonitor:
self.web3 = Web3(Web3.HTTPProvider(self.config['infura_url']))
pool_address = self.config['pool_address']
pool_abi = fetch_abi(pool_address)
pool_abi = fetch_abi(pool_address, key=self.config['etherscan_key'])
self.contract = self.web3.eth.contract(
address=pool_address,
abi=pool_abi
@ -34,6 +35,31 @@ class EventMonitor:
time.sleep(interval)
def fetch_token_info(self, token_address):
try:
token_abi = fetch_abi(token_address, key=self.config['etherscan_key'])
except Exception as err:
logger.warning(f"Failed to fetch info for {token_address}: {err}")
return {
'name': None,
'symbol': None,
'total_supply': None
}
# Create the contract instance
token_contract = self.web3.eth.contract(address=token_address, abi=token_abi)
# Fetch the name, symbol, and total supply
name = token_contract.functions.name().call()
symbol = token_contract.functions.symbol().call()
total_supply = token_contract.functions.totalSupply().call()
return {
'name': name,
'symbol': symbol,
'total_supply': str(total_supply)
}
def _check_connection(self):
if self.web3.is_connected():
logger.debug("Connected to Ethereum RPC")
@ -43,16 +69,57 @@ class EventMonitor:
return False
def _handle_event(self, event):
# TODO: Get token names
# TODO: Resolve token names
# TODO: Create data structure for the event
# Get token addresses from the event
token0_address = event.args['token0']
token1_address = event.args['token1']
pair_address = event.args['pair']
token0 = event.args['token0']
token1 = event.args['token1']
# Try to resolve name, symbol, and supply
token0_info = self.fetch_token_info(token0_address)
token1_info = self.fetch_token_info(token1_address)
logger.info(f"New pair: {token0} + {token1}")
self.queue.put(
f"New Uniswap pair: https://etherscan.io/address/{token0} + https://etherscan.io/address/{token1}"
# Create SQLAlchemy objects
new_token0 = Token(
id=token0_address,
name=token0_info['name'],
symbol=token0_info['symbol'],
total_supply=token0_info['total_supply']
)
new_token1 = Token(
id=token1_address,
name=token1_info['name'],
symbol=token1_info['symbol'],
total_supply=token1_info['total_supply']
)
new_pair = Pair(id=pair_address, token0=token0_address, token1=token1_address)
# Add token0
try:
session = next(get_session())
session.add(new_token0)
session.commit()
except Exception as err:
logger.warning(f"Could not add token0: {err}")
# Add token1
try:
session = next(get_session())
session.add(new_token1)
session.commit()
except Exception as err:
logger.warning(f"Could not add token1: {err}")
# Add pair
try:
session = next(get_session())
session.add(new_pair)
session.commit()
except Exception as err:
logger.warning(f"Could not add new pair: {err}")
# Add alert to queue
logger.info(f"New pair: {token0_address} + {token1_address}")
self.queue.put(
f"New Uniswap pair: [{token0_info['name']}](https://etherscan.io/address/{token0_address}) +"
f"[{token1_info['name']}](https://etherscan.io/address/{token1_address})"
)

View File

@ -1,3 +1,4 @@
matrix-nio
loguru
web3
SQLAlchemy

View File

@ -6,8 +6,8 @@ from web3.types import ChecksumAddress
def fetch_abi(address: ChecksumAddress, headers: Optional[Dict[str, Any]] = None,
params: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]:
url = f'https://api.etherscan.io/api?module=contract&action=getabi&address={address}'
params: Optional[Dict[str, Any]] = None, key: Optional[str] = None) -> Optional[Dict[str, Any]]:
url = f'https://api.etherscan.io/api?module=contract&action=getabi&address={address}&apikey={key}'
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
data = response.json()
@ -23,4 +23,3 @@ def fetch_abi(address: ChecksumAddress, headers: Optional[Dict[str, Any]] = None
def load_config(path: str) -> dict:
with open(path, 'r', encoding='utf-8') as f:
return json.loads(f.read())