feat: group Dune alert transfers by tx

protocols/stables/dune_large_transfers.py

@@ -82,6 +82,21 @@ def _short_tx_hash(tx_hash: str) -> str:
     return f"{tx_hash[:10]}…{tx_hash[-8:]}"
 
 
+def _tx_group_key(row: dict[str, Any]) -> str:
+    tx_hash = _as_str(row.get("tx_hash")).lower()
+    chain = _as_str(row.get("blockchain")).lower()
+    if tx_hash:
+        return f"{chain}|{tx_hash}"
+    return _row_key(row)
+
+
+def _group_rows_by_tx(rows: list[dict[str, Any]]) -> list[list[dict[str, Any]]]:
+    grouped: dict[str, list[dict[str, Any]]] = {}
+    for row in rows:
+        grouped.setdefault(_tx_group_key(row), []).append(row)
+    return list(grouped.values())
+
+
 def _row_key(row: dict[str, Any]) -> str:
     tx_hash = _as_str(row.get("tx_hash")).lower()
     contract = _as_str(row.get("contract_address")).lower()
@@ -98,7 +113,7 @@ def _route_for_row(row: dict[str, Any]) -> tuple[str, str] | None:
     return TOKEN_ROUTE.get((chain, addr))
 
 
-def _build_row_line(row: dict[str, Any], position: int = 1) -> str:
+def _build_row_line(row: dict[str, Any], matched_transfers_in_tx: int = 1) -> str:
     chain = _as_str(row.get("blockchain"))
     chain_name = escape_markdown(chain.replace("_", " ").title() or "Unknown")
     symbol = escape_markdown(_as_str(row.get("symbol")) or "unknown")
@@ -108,22 +123,25 @@ def _build_row_line(row: dict[str, Any], position: int = 1) -> str:
     link = _tx_link(chain, tx_hash)
     tx_label = escape_markdown(_short_tx_hash(tx_hash) or "unknown")
     tx_line = f"🔗 Transaction: [{tx_label}]({link})" if link != tx_hash else f"🔗 Transaction: {tx_label}"
-    return (
-        f"*Transfer {position}*\n"
-        f"🌐 Network: {chain_name}\n"
-        f"💰 Amount: {amount} {symbol}\n"
-        f"💵 Value: ${amount_usd}\n"
-        f"{tx_line}"
-    )
+    lines = [
+        f"🌐 Network: {chain_name}",
+        f"💰 Amount: {amount} {symbol}",
+        f"💵 Value: ${amount_usd}",
+    ]
+    if matched_transfers_in_tx > 1:
+        lines.append(f"🧾 Matched transfers in tx: {matched_transfers_in_tx}")
+    lines.append(tx_line)
+    return "\n".join(lines)
 
 
 def _build_protocol_lines(protocol_rows: list[dict[str, Any]], query_id: int) -> list[str]:
-    included_rows = protocol_rows[:MAX_ROWS_PER_PROTOCOL_ALERT]
-    lines = [_build_row_line(row, position) for position, row in enumerate(included_rows, start=1)]
+    tx_groups = _group_rows_by_tx(protocol_rows)
+    included_groups = tx_groups[:MAX_ROWS_PER_PROTOCOL_ALERT]
+    lines = [_build_row_line(tx_rows[0], len(tx_rows)) for tx_rows in included_groups]
 
-    omitted_count = len(protocol_rows) - len(included_rows)
+    omitted_count = len(tx_groups) - len(included_groups)
     if omitted_count > 0:
-        lines.append(f"…and {omitted_count} more. See Dune query {query_id} for the full result.")
+        lines.append(f"…and {omitted_count} more transactions. See Dune query {query_id} for the full result.")
 
     return lines
 
@@ -137,19 +155,11 @@ def _build_alert_message(
     route = _route_for_row(protocol_rows[0])
     symbol = route[0] if route else (_as_str(protocol_rows[0].get("symbol")) or "unknown")
     symbol = escape_markdown(symbol)
-    protocol_name = escape_markdown(protocol.replace("_", " ").title())
-    transfer_word = "transfer" if len(protocol_rows) == 1 else "transfers"
-    match_summary = str(len(protocol_rows))
-    if total_rows != len(protocol_rows):
-        match_summary = f"{len(protocol_rows)} for {protocol_name} ({total_rows} total)"
+    tx_count = len(_group_rows_by_tx(protocol_rows))
+    transfer_word = "transfer" if tx_count == 1 else "transfers"
 
     lines = _build_protocol_lines(protocol_rows, query_id)
-    return (
-        f"*Large {symbol} {transfer_word} detected*\n\n"
-        f"🏦 Protocol: {protocol_name}\n"
-        f"📦 New matches: {match_summary}\n"
-        f"📊 Dune query: {query_id}\n\n" + "\n\n".join(lines)
-    )
+    return f"*Large {symbol} {transfer_word} detected*\n" + "\n—————\n".join(lines)
 
 
 def _group_rows_by_protocol(rows: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:

tests/test_dune_large_transfers.py

@@ -81,7 +81,21 @@ def test_build_protocol_lines_appends_truncation_notice():
     lines = monitor._build_protocol_lines(rows, query_id=1234567)
 
     assert len(lines) == monitor.MAX_ROWS_PER_PROTOCOL_ALERT + 1
-    assert lines[-1] == "…and 2 more. See Dune query 1234567 for the full result."
+    assert lines[-1] == "…and 2 more transactions. See Dune query 1234567 for the full result."
+
+
+def test_build_protocol_lines_shows_one_entry_per_transaction():
+    rows = [
+        _row(tx_hash="0xSAME", log_index=1, amount="5139554.464867114", amount_usd="5139554.464867114"),
+        _row(tx_hash="0xSAME", log_index=2, amount="5139554.464867114", amount_usd="5139554.464867114"),
+    ]
+
+    lines = monitor._build_protocol_lines(rows, query_id=1234567)
+
+    assert len(lines) == 1
+    assert lines[0].startswith("🌐 Network: Ethereum")
+    assert "🧾 Matched transfers in tx: 2" in lines[0]
+    assert "*Transaction " not in lines[0]
 
 
 def test_build_alert_message_is_readable_and_formats_large_values():
@@ -97,18 +111,72 @@ def test_build_alert_message_is_readable_and_formats_large_values():
     message = monitor._build_alert_message("infinifi", [row], query_id=7558262, total_rows=1)
 
     assert message == (
-        "*Large iUSD transfer detected*\n\n"
-        "🏦 Protocol: Infinifi\n"
-        "📦 New matches: 1\n"
-        "📊 Dune query: 7558262\n\n"
-        "*Transfer 1*\n"
+        "*Large iUSD transfer detected*\n"
         "🌐 Network: Ethereum\n"
         "💰 Amount: 5,139,554.46 iUSD\n"
         "💵 Value: $5,139,554.46\n"
         f"🔗 Transaction: [0xbcd224d8…3c0c1bfd](https://etherscan.io/tx/{tx_hash})"
     )
 
 
+def test_build_alert_message_counts_duplicate_rows_inside_same_tx_once():
+    tx_hash = "0xbcd224d842f47167ec6339c47ac473ba751b73afbce36ed82142d8603c0c1bfd"
+    rows = [
+        _row(
+            contract_address="0x48f9e38f3070ad8945dfeae3fa70987722e3d89c",
+            symbol="iUSD",
+            amount="5139554.464867114",
+            amount_usd="5139554.464867114",
+            tx_hash=tx_hash,
+            log_index=1,
+        ),
+        _row(
+            contract_address="0x48f9e38f3070ad8945dfeae3fa70987722e3d89c",
+            symbol="iUSD",
+            amount="5139554.464867114",
+            amount_usd="5139554.464867114",
+            tx_hash=tx_hash,
+            log_index=2,
+        ),
+    ]
+
+    message = monitor._build_alert_message("infinifi", rows, query_id=7558262, total_rows=2)
+
+    assert "📦 New transactions" not in message
+    assert "🏦 Protocol" not in message
+    assert "📊 Dune query" not in message
+    assert "*Transaction " not in message
+    assert "🧾 Matched transfers in tx: 2" in message
+
+
+def test_build_alert_message_separates_multiple_transactions_without_repeated_headings():
+    first_tx_hash = "0xbcd224d842f47167ec6339c47ac473ba751b73afbce36ed82142d8603c0c1bfd"
+    second_tx_hash = "0xaaaaaaaa42f47167ec6339c47ac473ba751b73afbce36ed82142d8603c0c1bfd"
+    rows = [
+        _row(
+            contract_address="0x48f9e38f3070ad8945dfeae3fa70987722e3d89c",
+            symbol="iUSD",
+            amount="5139554.464867114",
+            amount_usd="5139554.464867114",
+            tx_hash=first_tx_hash,
+        ),
+        _row(
+            contract_address="0x48f9e38f3070ad8945dfeae3fa70987722e3d89c",
+            symbol="iUSD",
+            amount="7000000",
+            amount_usd="7000000",
+            tx_hash=second_tx_hash,
+        ),
+    ]
+
+    message = monitor._build_alert_message("infinifi", rows, query_id=7558262, total_rows=2)
+
+    assert message.startswith("*Large iUSD transfers detected*\n")
+    assert message.count("🌐 Network: Ethereum") == 2
+    assert message.count("\n—————\n") == 1
+    assert "*Transaction " not in message
+
+
 def test_main_sends_pretty_alert_with_markdown_enabled(monkeypatch):
     row = _row()
     result = Mock()