Compare commits

..

2 Commits

Author SHA1 Message Date
a3fda36d43 add database support 2024-06-08 14:16:53 -04:00
6510eab5a8 readability update 2024-06-08 13:15:49 -04:00
6 changed files with 132 additions and 12 deletions

1
.gitignore vendored
View File

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

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,10 @@ 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']
},
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:
@ -34,6 +35,31 @@ class EventMonitor:
time.sleep(interval)
def fetch_token_info(self, token_address):
try:
token_abi = fetch_abi(token_address)
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

@ -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())