feat(automations): improve check new feed script

Author: dineshpinto

automations/check_new_feed.py

@@ -1,5 +1,6 @@
 import json
 import logging
+import re
 import sys
 from dataclasses import dataclass
 from pathlib import Path
@@ -24,8 +25,21 @@ def get(cls, name: str) -> "NetworkConfig":
         return cls.by_name[name]
 
 
+# Configuration
+ISSUES_FILE: Final[Path] = Path("issues.md")
+MAX_MARKET_CAP_RANK: Final[int] = 100
+MIN_USD_VOLUME: Final[int] = 100_000_000
+GITHUB_NEUTRAL_EXIT: Final[int] = 78
+HEADER_TEMPLATE: Final[str] = """---
+title: "[auto_req]: Potential New Feeds"
+assignees:
+  - dineshpinto
+labels:
+  - enhancement
+---
+"""
+_VOLUME_RE = re.compile(r"^\$?\s*([\d,]+(?:\.\d{1,2})?)$")
 REGISTRY_ADDRESS: Final[str] = "0xaD67FE66660Fb8dFE9d6b1b4240d8650e30F6019"
-
 NETWORKS: list[NetworkConfig] = [
     NetworkConfig(
         name="FlareMainnet",
@@ -48,20 +62,8 @@ def get(cls, name: str) -> "NetworkConfig":
         explorer_api_url="https://coston-explorer.flare.network/api",
     ),
 ]
-
 NetworkConfig.by_name = {net.name: net for net in NETWORKS}
 
-
-# Configuration
-ISSUES_FILE: Final[Path] = Path("issues.md")
-MAX_MARKET_CAP_RANK: Final[int] = 100
-HEADER_TEMPLATE: Final[str] = """---
-title: "[auto_req]: Potential New Feeds"
-assignees: dineshpinto
-labels: "enhancement"
----
-"""
-
 # Logging
 logging.basicConfig(
     encoding="utf-8",
@@ -76,12 +78,10 @@ class ExplorerError(RuntimeError):
 
 
 @retry(stop=stop_after_attempt(3), wait=wait_exponential_jitter(initial=1, max=8))
-def _get(
-    session: requests.Session, **kwargs: dict[str, str] | str
-) -> requests.Response:
+def _get(session: requests.Session, url: str, **kwargs: Any) -> requests.Response:  # noqa: ANN401
     """`requests.get` with automatic retry and logging."""
-    logger.debug("GET %s", kwargs.get("url") or kwargs.get("url"))
-    resp = session.get(**kwargs, timeout=10)  # pyright: ignore[reportArgumentType]
+    logger.debug("GET %s params=%s", url, kwargs.get("params"))
+    resp = session.get(url, timeout=10, **kwargs)
     resp.raise_for_status()
     return resp
 
@@ -104,83 +104,135 @@ def get_contract_abi(
         raise ExplorerError(msg) from exc
 
 
-def decode_feed_name(feed: bytes) -> str:
+def _decode_feed_name(feed: bytes) -> str:
     """Convert feed bytes (returned by contract) to a plain symbol."""
-    return feed[1:].decode().rstrip("\x00").split("/")[0]
+
+    if not feed:
+        return ""
+    raw = feed[1:]
+    return raw.decode(errors="ignore").rstrip("\x00").split("/")[0]
 
 
 def get_current_feeds(contract: Contract) -> set[str]:
-    """Return a **set** of existing feed symbols."""
     feeds: list[bytes] = contract.functions.fetchAllCurrentFeeds().call()[0]
-    return {decode_feed_name(f) for f in feeds}
+    return {_decode_feed_name(f).lower() for f in feeds}
 
 
 def prettify_coin(coin: dict[str, Any]) -> str:
     """Convert `coin` dict into a readable block of Markdown."""
     d = coin.get("data", {})
-    change_24h = d.get("price_change_percentage_24h", {}).get("usd", "N/A")
+    pct_24 = d.get("price_change_percentage_24h", {}).get("usd")
+    change_24h = f"{round(pct_24, 2)}" if isinstance(pct_24, (int, float)) else "N/A"
     total_vol = d.get("total_volume", "N/A")
 
     return (
-        f"name: {coin.get('name', 'N/A')}\n"
-        f"symbol: {coin.get('symbol', 'N/A')}\n"
-        f"coingecko_id: {coin.get('id', 'N/A')}\n"
-        f"price_change_percentage_24h: {change_24h}\n"
-        f"total_volume: {total_vol}\n"
-        f"coingecko_link: https://www.coingecko.com/en/coins/{coin.get('id', '')}\n"
+        f"Name: {coin.get('name', 'N/A')}\n"
+        f"Symbol: {coin.get('symbol', 'N/A')}\n"
+        f"Coingecko ID: {coin.get('id', 'N/A')}\n"
+        f"24h Price Change: {change_24h}%\n"
+        f"Total Volume: {total_vol}\n"
+        f"Link: https://www.coingecko.com/en/coins/{coin.get('id', '')}\n"
     )
 
 
 def parse_volume(vol_str: str) -> int:
-    """Parse volume string and return it as an integer."""
-    if not vol_str[0] == "$":
-        raise ValueError(f"Invalid volume string: {vol_str}")
-    return int(vol_str.strip("$").replace(",", ""))
+    """Parse a USD volume string like '$123,456.78' -> 123456 (floor)."""
+    m = _VOLUME_RE.match(vol_str.strip())
+    if not m:
+        msg = f"Invalid volume string: {vol_str!r}"
+        raise ValueError(msg)
+    num = m.group(1).replace(",", "")
+    return int(float(num))
 
 
 def write_issue(coins: list[dict[str, Any]]) -> None:
-    """Write the Markdown issue file."""
-    body = ["Coins matching criteria:\n"]
-    body += [f"## {c['name']}\n{prettify_coin(c)}\n" for c in coins]
-    ISSUES_FILE.write_text(HEADER_TEMPLATE + "\n".join(body), encoding="utf-8")
+    lines = [
+        "Coins matching [FIP.08](https://proposals.flare.network/FIP/FIP_8.html) criteria:\n"
+    ]
+    lines += [f"## {c.get('name', 'N/A')}\n{prettify_coin(c)}\n" for c in coins]
+    content = HEADER_TEMPLATE + "\n".join(lines)
+    tmp = ISSUES_FILE.with_suffix(".tmp")
+    tmp.write_text(
+        content if content.endswith("\n") else content + "\n", encoding="utf-8"
+    )
+    tmp.replace(ISSUES_FILE)
     logger.info("Wrote %d coin(s) to %s", len(coins), ISSUES_FILE)
 
 
+@retry(stop=stop_after_attempt(3), wait=wait_exponential_jitter(initial=1, max=8))
+def get_trending(cg: CoinGeckoAPI) -> list[dict[str, Any]]:
+    data = cg.get_search_trending()  # pyright: ignore[reportUnknownMemberType]
+    return data.get("coins", [])
+
+
+def _safe_rank(c: dict[str, Any]) -> float:
+    return c.get("item", {}).get("market_cap_rank") or float("inf")
+
+
+def _safe_vol(c: dict[str, Any]) -> int:
+    v = c.get("item", {}).get("data", {}).get("total_volume")
+    return parse_volume(v) if isinstance(v, str) else 0
+
+
+def filter_candidates(
+    trending: list[dict[str, Any]],
+    current_feeds: set[str],
+    max_mcap_rank: int,
+    min_volume: int,
+) -> list[dict[str, Any]]:
+    return [
+        c["item"]
+        for c in trending
+        if _safe_rank(c) < max_mcap_rank
+        and _safe_vol(c) > min_volume
+        and c.get("item", {}).get("symbol", "").lower() not in current_feeds
+    ]
+
+
 def main() -> None:
     cg = CoinGeckoAPI()
-    session = requests.Session()
 
     rpc_url = NetworkConfig.get("FlareMainnet").rpc_url
-    w3 = Web3(Web3.HTTPProvider(rpc_url))
+    w3 = Web3(Web3.HTTPProvider(rpc_url, request_kwargs={"timeout": 10}))
     if not w3.is_connected():
         msg = f"Cannot reach RPC at {rpc_url}"
         raise ConnectionError(msg)
-    logger.info("Connected to RPC %s", rpc_url)
+    logger.info("Connected to %s", rpc_url, extra={"network": "FlareMainnet"})
 
-    registry = w3.eth.contract(
-        address=Web3.to_checksum_address(REGISTRY_ADDRESS),
-        abi=get_contract_abi(REGISTRY_ADDRESS, session),
-    )
-    updater_addr = registry.functions.getContractAddressByName("FastUpdater").call()
-    fast_updater = w3.eth.contract(
-        address=Web3.to_checksum_address(updater_addr),
-        abi=get_contract_abi(updater_addr, session),
-    )
-    logger.info("FastUpdater contract %s", fast_updater.address)
+    with requests.Session() as session:
+        registry = w3.eth.contract(
+            address=Web3.to_checksum_address(REGISTRY_ADDRESS),
+            abi=get_contract_abi(REGISTRY_ADDRESS, session),
+        )
+        updater_addr = registry.functions.getContractAddressByName("FastUpdater").call()
+        fast_updater = w3.eth.contract(
+            address=Web3.to_checksum_address(updater_addr),
+            abi=get_contract_abi(updater_addr, session),
+        )
+        logger.info("FastUpdater %s", fast_updater.address)
 
     current_feeds = get_current_feeds(fast_updater)
+    logger.info("Existing feeds: %d", len(current_feeds))
+
+    trending = get_trending(cg)
+    logger.info(
+        "Found potential coins %s",
+        [c.get("item", {}).get("name", "N/A") for c in trending],
+    )
+
+    candidates = sorted(
+        filter_candidates(
+            trending,
+            current_feeds,
+            max_mcap_rank=MAX_MARKET_CAP_RANK,
+            min_volume=MIN_USD_VOLUME,
+        ),
+        key=lambda x: (x.get("market_cap_rank") or float("inf"), x.get("name", "")),
+    )
+    logger.info("Filtered candidates: %s", [c["name"] for c in candidates])
 
-    trending = cg.get_search_trending()["coins"]  # pyright: ignore[reportUnknownMemberType]
-    logger.info("Found potential coins %s", [f"{c['item']['name']}" for c in trending])
-    candidates = [
-        c["item"]
-        for c in trending
-        if c["item"].get("market_cap_rank", float("inf")) < MAX_MARKET_CAP_RANK
-        and parse_volume(c["item"]["data"]["total_volume"]) > 100_000_000
-        and c["item"]["symbol"] not in current_feeds
-    ]
     if not candidates:
-        sys.exit(78)  # Skips GitHub workflow 'if' steps
+        sys.exit(GITHUB_NEUTRAL_EXIT)
     write_issue(candidates)