use fallback ERC-20 ABI

README.md

@@ -3,6 +3,7 @@ Uniswap PairCreation monitor that sends alerts to Matrix.
 
 ## Requirements
 - Infura API URL
+- Etherscan API key
 - Matrix bot account
 
 ## Configuration
@@ -15,6 +16,7 @@ Uniswap PairCreation monitor that sends alerts to Matrix.
     "room_id": ""
   },
   "infura_url": "",
-  "pool_address": "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f"
+  "pool_address": "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f",
+  "etherscan_key": ""
 }
 ```

main.py

@@ -11,7 +11,8 @@ from util import load_config
 
 
 def init_logger():
-    logger.add(sys.stdout, level="DEBUG")
+    logger.remove()
+    logger.add(sys.stdout, level="INFO")
 
 
 async def main():
@@ -21,17 +22,25 @@ async def main():
     monitor = EventMonitor(
         config={
             "infura_url": config['infura_url'],
-            "pool_address": config['pool_address']
+            "pool_address": config['pool_address'],
+            "etherscan_key": config['etherscan_key']
         },
         queue=queue
     )
     monitor_thread = threading.Thread(target=monitor.log_loop, args=(15,))
     monitor_thread.start()
 
+    # Send "online message"
+    try:
+        await bot.send_markdown("**MBOT ONLINE**")
+        await bot.logout()
+    except nio.exceptions.LocalProtocolError:
+        pass
+
     while True:
         event = queue.get()
         try:
-            await bot.send_message(event)
+            await bot.send_markdown(event)
             await bot.logout()
             queue.task_done()
         except nio.exceptions.LocalProtocolError:
@@ -39,4 +48,5 @@ async def main():
 
 
 if __name__ == '__main__':
+    init_logger()
     asyncio.get_event_loop().run_until_complete(main())

matrix.py

@@ -1,3 +1,4 @@
+import markdown
 from loguru import logger
 from nio import AsyncClient, LoginResponse
 
@@ -28,6 +29,28 @@ class MatrixBot:
         )
         logger.info("Message sent")
 
+    async def send_markdown(self, message: str):
+        if not self.client.access_token:
+            logged_in = await self.login()
+            if not logged_in:
+                return
+
+        # Convert message to markdown
+        html = markdown.markdown(message)
+
+        # Send markdown formatted message
+        await self.client.room_send(
+            room_id=self.config['room_id'],
+            message_type="m.room.message",
+            content={
+                "msgtype": "m.text",
+                "body": message,
+                "format": "org.matrix.custom.html",
+                "formatted_body": html
+            }
+        )
+        logger.info("Markdown message sent")
+
     async def login(self):
         response = await self.client.login(
             password=self.config['password']

monitor.py

@@ -2,7 +2,7 @@ import time
 from queue import Queue
 from loguru import logger
 from web3 import Web3
-from util import fetch_abi
+from util import fetch_abi, format_matrix_alert
 from db import Token, Pair, get_session
 
 
@@ -15,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
@@ -37,9 +37,12 @@ class EventMonitor:
 
     def fetch_token_info(self, token_address):
         try:
-            token_abi = fetch_abi(token_address)
+            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}")
+            token_abi = None
+
+        if not token_abi:
             return {
                 'name': None,
                 'symbol': None,
@@ -119,7 +122,8 @@ class EventMonitor:
 
         # 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})"
+        formatted_alert = format_matrix_alert(
+            token0={'name': token0_info['name'], 'address': token0_address},
+            token1={'name': token1_info['name'], 'address': token1_address}
         )
+        self.queue.put(formatted_alert)

requirements.txt

@@ -1,4 +1,6 @@
 matrix-nio
 loguru
 web3
+requests
 SQLAlchemy
+markdown

util.py

@@ -4,10 +4,40 @@ from loguru import logger
 from typing import Any, Optional, Dict
 from web3.types import ChecksumAddress
 
+STANDARD_ERC20_ABI = [
+    {
+        "constant": True,
+        "inputs": [],
+        "name": "name",
+        "outputs": [{"name": "", "type": "string"}],
+        "payable": False,
+        "stateMutability": "view",
+        "type": "function",
+    },
+    {
+        "constant": True,
+        "inputs": [],
+        "name": "symbol",
+        "outputs": [{"name": "", "type": "string"}],
+        "payable": False,
+        "stateMutability": "view",
+        "type": "function",
+    },
+    {
+        "constant": True,
+        "inputs": [],
+        "name": "totalSupply",
+        "outputs": [{"name": "", "type": "uint256"}],
+        "payable": False,
+        "stateMutability": "view",
+        "type": "function",
+    }
+]
+
 
 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()
@@ -15,7 +45,8 @@ def fetch_abi(address: ChecksumAddress, headers: Optional[Dict[str, Any]] = None
     # Check for error in the response
     if data['status'] == '0':
         logger.error(f"Error fetching ABI: {data.get('result')}")
-        return None
+        logger.warning(f"Using fallback ABI.")
+        return STANDARD_ERC20_ABI
 
     return json.loads(data['result'])
 
@@ -23,3 +54,11 @@ 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())
+
+
+def format_matrix_alert(token0: dict, token1: dict) -> str:
+    return f"""
+🚨 New Uniswap Pair Alert 🚨\n
+🌐 **Asset Pair:** [{token0['name']}](https://etherscan.io/token/{token0['address']}) + [{token1['name']}](https://etherscan.io/token/{token1['address']})\n
+🏠 **Network:** ETH
+"""