diff --git a/src/polymarket_insider_tracker/profiler/funding.py b/src/polymarket_insider_tracker/profiler/funding.py index a688c2b..3782fc7 100644 --- a/src/polymarket_insider_tracker/profiler/funding.py +++ b/src/polymarket_insider_tracker/profiler/funding.py @@ -26,8 +26,46 @@ logger = logging.getLogger(__name__) USDC_BRIDGED = "0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174" USDC_NATIVE = "0x3c499c542cEF5E3811e1192ce70d8cC03d5c3359" -# ERC20 Transfer event signature +# ERC20 Transfer event signature. ``HexBytes.hex()`` returns a *bare* hex +# string without the ``0x`` prefix; publicnode tolerates that, but stricter +# providers (e.g. drpc — which we use as the fallback) reject it with +# ``invalid argument 0: hex string without 0x prefix``. Always pass the +# 0x-prefixed form to ``eth_getLogs``. TRANSFER_EVENT_SIGNATURE = AsyncWeb3.keccak(text="Transfer(address,address,uint256)") +TRANSFER_EVENT_TOPIC = "0x" + TRANSFER_EVENT_SIGNATURE.hex().removeprefix("0x") + +# eth_getLogs block-range chunking. Most public Polygon RPCs (publicnode, ankr, +# llamarpc) cap the range at 10_000 blocks per call; pick a window slightly +# under the cap so off-by-one differences between providers don't trip us up. +DEFAULT_CHUNK_SIZE_BLOCKS = 9_000 +# Polygon block time is ~2.0s. publicnode (the most common free RPC) prunes +# log history aggressively — empirically only ~100k blocks (~55 hours) are +# served before requests start returning "History has been pruned". We default +# to 80k blocks (~44 hours), which is more than enough for fresh-wallet +# funding traces (those wallets are by definition new) and fits comfortably +# inside what most public providers retain. +DEFAULT_MAX_LOOKBACK_BLOCKS = 80_000 + +# Substrings that, when present in an RPC error, indicate the chunk we just +# asked for is outside the provider's archive horizon. Walking further back +# is futile, so we stop the trace early instead of hammering every chunk. +_PRUNED_HISTORY_MARKERS: tuple[str, ...] = ( + "history has been pruned", + "missing trie node", + "older than", +) + + +def _is_pruned_history_error(err: BaseException) -> bool: + """Return True if the RPC error indicates pruned history. + + Public Polygon nodes only retain a recent slice of log history. When we + walk back through that slice in chunks and hit the cutoff, every further + chunk will fail with the same message — so we stop early instead of + burning quota on guaranteed failures. + """ + text = str(err).lower() + return any(marker in text for marker in _PRUNED_HISTORY_MARKERS) class FundingTracer: @@ -50,6 +88,8 @@ class FundingTracer: *, max_hops: int = 3, usdc_addresses: list[str] | None = None, + chunk_size_blocks: int = DEFAULT_CHUNK_SIZE_BLOCKS, + max_lookback_blocks: int = DEFAULT_MAX_LOOKBACK_BLOCKS, ) -> None: """Initialize the funding tracer. @@ -58,6 +98,11 @@ class FundingTracer: entity_registry: Registry for entity classification. Creates default if None. max_hops: Maximum hops to trace back (default 3). usdc_addresses: USDC contract addresses to track. Uses defaults if None. + chunk_size_blocks: Block window size per eth_getLogs call. Public + Polygon RPCs cap at 10_000 blocks; default leaves a safety margin. + max_lookback_blocks: How far back to scan when caller passes + ``from_block=0``. Default ~44 hours at 2s block time, which + fits inside the pruned-history horizon of most public RPCs. """ self.polygon_client = polygon_client self.entity_registry = entity_registry or EntityRegistry() @@ -65,6 +110,8 @@ class FundingTracer: self._usdc_addresses = [ addr.lower() for addr in (usdc_addresses or [USDC_BRIDGED, USDC_NATIVE]) ] + self._chunk_size_blocks = chunk_size_blocks + self._max_lookback_blocks = max_lookback_blocks async def trace( self, @@ -209,46 +256,144 @@ class FundingTracer: ) -> list[dict[str, Any]]: """Get ERC20 Transfer event logs. + Public Polygon RPCs (publicnode, ankr, llamarpc) cap ``eth_getLogs`` at + 10_000 blocks per call. To work around this we resolve the requested + range into a concrete block window (defaulting to the last + ``max_lookback_blocks`` when caller passes ``from_block=0``) and walk + the window in chunks of ``chunk_size_blocks``, oldest-first, stopping + once ``limit`` matches are collected. Walking oldest-first preserves + the "first transfer" semantics expected by the funding chain tracer. + + If a chunk comes back with a "history has been pruned" style error + the rest of the walk is short-circuited — every subsequent chunk + would hit the same archive cutoff and there's no point burning quota + on guaranteed failures. + Args: to_address: Filter by recipient address. token_address: ERC20 token contract address. limit: Maximum logs to return. - from_block: Starting block number. - to_block: Ending block number. + from_block: Starting block number (0 means + ``latest - max_lookback_blocks``). + to_block: Ending block number ("latest" resolves to current head). Returns: - List of log dictionaries. + List of log dictionaries, oldest first, capped at ``limit``. """ # Pad address to 32 bytes for topic filter padded_to = "0x" + to_address.lower().replace("0x", "").zfill(64) + topics = [ + TRANSFER_EVENT_TOPIC, # Transfer event (must be 0x-prefixed for drpc) + None, # from (any) + padded_to, # to (target address) + ] + contract_address = AsyncWeb3.to_checksum_address(token_address) + start_block, end_block = await self._resolve_block_range(from_block, to_block) + if start_block > end_block: + return [] + + results: list[dict[str, Any]] = [] + chunk_size = max(1, self._chunk_size_blocks) + chunk_start = start_block + + while chunk_start <= end_block: + chunk_end = min(chunk_start + chunk_size - 1, end_block) + try: + chunk_logs = await self._fetch_logs_chunk( + contract_address=contract_address, + topics=topics, + from_block=chunk_start, + to_block=chunk_end, + ) + except Exception as e: + if _is_pruned_history_error(e): + # The provider has dropped this slice of history. Walking + # further back will hit the same wall on every chunk; + # stop now and return what we already have. + logger.info( + "eth_getLogs chunk %d-%d outside archive horizon for %s; stopping trace", + chunk_start, + chunk_end, + to_address, + ) + break + logger.warning( + "eth_getLogs chunk %d-%d failed for %s: %s", + chunk_start, + chunk_end, + to_address, + e, + ) + # Skip this window and keep walking — partial data is better + # than aborting the whole trace on a single flaky chunk. + chunk_start = chunk_end + 1 + continue + + for log in chunk_logs: + results.append(dict(log)) + if len(results) >= limit: + return results + + chunk_start = chunk_end + 1 + + return results + + async def _resolve_block_range( + self, + from_block: int | str, + to_block: int | str, + ) -> tuple[int, int]: + """Resolve symbolic block params to concrete numeric bounds. + + ``from_block=0`` (the historical default) is rewritten to + ``latest - max_lookback_blocks`` so we don't try to scan all of Polygon. + """ + w3 = self._select_w3() + + if isinstance(to_block, str): + await self.polygon_client._rate_limiter.acquire() + head = int(await w3.eth.block_number) + end = head + else: + end = int(to_block) + + if isinstance(from_block, str): + # Treat any symbolic from-block (e.g. "earliest") as "go back + # max_lookback_blocks from end"; that's what callers actually want. + start = max(0, end - self._max_lookback_blocks) + elif from_block == 0: + start = max(0, end - self._max_lookback_blocks) + else: + start = int(from_block) + + return start, end + + async def _fetch_logs_chunk( + self, + contract_address: str, + topics: list[Any], + from_block: int, + to_block: int, + ) -> list[Any]: + """Issue a single bounded ``eth_getLogs`` call.""" await self.polygon_client._rate_limiter.acquire() - - # Use the web3 instance from polygon client - w3 = ( - self.polygon_client._w3 - if self.polygon_client._primary_healthy - else (self.polygon_client._w3_fallback or self.polygon_client._w3) - ) - - # Get logs with Transfer event filtering by recipient + w3 = self._select_w3() # Note: web3 typing is overly restrictive for block params - logs = await w3.eth.get_logs( + return await w3.eth.get_logs( { - "address": AsyncWeb3.to_checksum_address(token_address), - "topics": [ - TRANSFER_EVENT_SIGNATURE.hex(), # Transfer event - None, # from (any) - padded_to, # to (target address) - ], + "address": contract_address, + "topics": topics, "fromBlock": from_block, # type: ignore[typeddict-item] "toBlock": to_block, # type: ignore[typeddict-item] } ) - # Convert to list of dicts and limit - result = [dict(log) for log in logs[:limit]] - return result + def _select_w3(self) -> AsyncWeb3: + """Pick primary or fallback web3 instance based on health.""" + if self.polygon_client._primary_healthy: + return self.polygon_client._w3 + return self.polygon_client._w3_fallback or self.polygon_client._w3 async def _log_to_funding_transfer( self, diff --git a/tests/profiler/test_funding.py b/tests/profiler/test_funding.py index f1c2ccf..d1aa76d 100644 --- a/tests/profiler/test_funding.py +++ b/tests/profiler/test_funding.py @@ -327,6 +327,10 @@ class TestGetTransferLogs: await funding_tracer._get_transfer_logs( to_address=TEST_WALLET, token_address=USDC_BRIDGED, + # Explicit numeric range so we stay inside one chunk and skip + # the "latest" → block_number resolution path. + from_block=1, + to_block=8_000, ) mock_w3.eth.get_logs.assert_called_once() @@ -334,10 +338,15 @@ class TestGetTransferLogs: # Verify topics structure assert len(call_args["topics"]) == 3 - assert call_args["topics"][0] == TRANSFER_EVENT_SIGNATURE.hex() + # The Transfer event topic must be 0x-prefixed; drpc rejects bare hex. + assert call_args["topics"][0] == "0x" + TRANSFER_EVENT_SIGNATURE.hex().removeprefix("0x") + assert call_args["topics"][0].startswith("0x") assert call_args["topics"][1] is None # from (any) # to address should be padded to 32 bytes assert call_args["topics"][2].endswith(TEST_WALLET.lower().replace("0x", "")) + # And the chunk bounds match what we asked for. + assert call_args["fromBlock"] == 1 + assert call_args["toBlock"] == 8_000 @pytest.mark.asyncio async def test_get_transfer_logs_respects_limit( @@ -355,6 +364,8 @@ class TestGetTransferLogs: to_address=TEST_WALLET, token_address=USDC_BRIDGED, limit=3, + from_block=1, + to_block=8_000, ) assert len(result) == 3 @@ -374,10 +385,242 @@ class TestGetTransferLogs: await funding_tracer._get_transfer_logs( to_address=TEST_WALLET, token_address=USDC_BRIDGED, + from_block=1, + to_block=8_000, ) mock_fallback.eth.get_logs.assert_called_once() + @pytest.mark.asyncio + async def test_get_transfer_logs_chunks_large_ranges( + self, + funding_tracer: FundingTracer, + mock_polygon_client: MagicMock, + ) -> None: + """Ranges wider than chunk_size are split into multiple eth_getLogs calls. + + This is the regression guard for the publicnode 10_000-block cap that + was rejecting every funding trace before chunking landed. + """ + mock_w3 = MagicMock() + mock_w3.eth.get_logs = AsyncMock(return_value=[]) + mock_polygon_client._w3 = mock_w3 + + # 25_000 blocks at 9_000-per-chunk → 3 calls (9000 + 9000 + 7001). + await funding_tracer._get_transfer_logs( + to_address=TEST_WALLET, + token_address=USDC_BRIDGED, + from_block=1_000_000, + to_block=1_025_000, + ) + + assert mock_w3.eth.get_logs.call_count == 3 + windows = [call[0][0] for call in mock_w3.eth.get_logs.call_args_list] + assert windows[0]["fromBlock"] == 1_000_000 + assert windows[0]["toBlock"] == 1_008_999 + assert windows[1]["fromBlock"] == 1_009_000 + assert windows[1]["toBlock"] == 1_017_999 + assert windows[2]["fromBlock"] == 1_018_000 + assert windows[2]["toBlock"] == 1_025_000 + # No window exceeds the chunk size — that's what RPC providers reject. + for win in windows: + assert win["toBlock"] - win["fromBlock"] + 1 <= 9_000 + + @pytest.mark.asyncio + async def test_get_transfer_logs_stops_when_limit_hit_mid_walk( + self, + funding_tracer: FundingTracer, + mock_polygon_client: MagicMock, + ) -> None: + """Walking should stop as soon as ``limit`` matches are gathered.""" + mock_w3 = MagicMock() + # First chunk yields 5 logs, more than the limit, so subsequent chunks + # must not be queried. + mock_w3.eth.get_logs = AsyncMock(return_value=[MagicMock() for _ in range(5)]) + mock_polygon_client._w3 = mock_w3 + + result = await funding_tracer._get_transfer_logs( + to_address=TEST_WALLET, + token_address=USDC_BRIDGED, + limit=2, + from_block=1_000_000, + to_block=1_025_000, + ) + + assert len(result) == 2 + mock_w3.eth.get_logs.assert_called_once() + + @pytest.mark.asyncio + async def test_get_transfer_logs_skips_failing_chunk( + self, + funding_tracer: FundingTracer, + mock_polygon_client: MagicMock, + ) -> None: + """A flaky chunk must not abort the whole trace — we move on.""" + mock_w3 = MagicMock() + good_log = MagicMock() + responses: list[Any] = [ + RuntimeError("RPC hiccup"), + [good_log], + ] + + async def fake_get_logs(_params: dict[str, Any]) -> list[Any]: + outcome = responses.pop(0) + if isinstance(outcome, BaseException): + raise outcome + return outcome + + mock_w3.eth.get_logs = AsyncMock(side_effect=fake_get_logs) + mock_polygon_client._w3 = mock_w3 + + result = await funding_tracer._get_transfer_logs( + to_address=TEST_WALLET, + token_address=USDC_BRIDGED, + from_block=1_000_000, + to_block=1_018_000, # forces 3 chunks; we exercise chunks 1+2 + ) + + # The error chunk is skipped; the second chunk contributes one log. + assert result == [dict(good_log)] + assert mock_w3.eth.get_logs.call_count >= 2 + + @pytest.mark.asyncio + async def test_get_transfer_logs_resolves_latest_via_block_number( + self, + funding_tracer: FundingTracer, + mock_polygon_client: MagicMock, + ) -> None: + """``to_block='latest'`` should resolve via ``eth.block_number``. + + And ``from_block=0`` should not become a full-history scan — it must + be clamped to ``latest - max_lookback_blocks``. + """ + + async def _block_number_coro() -> int: + return 5_000 + + mock_eth = MagicMock() + mock_eth.get_logs = AsyncMock(return_value=[]) + # Property-style awaitable: web3.py exposes block_number as a property + # returning a coroutine, so each access must yield a fresh awaitable. + type(mock_eth).block_number = property( # type: ignore[misc] + lambda _self: _block_number_coro() + ) + mock_w3 = MagicMock() + mock_w3.eth = mock_eth + mock_polygon_client._w3 = mock_w3 + + await funding_tracer._get_transfer_logs( + to_address=TEST_WALLET, + token_address=USDC_BRIDGED, + ) + + # block_number=5000 < chunk_size, so it's one chunk that bottoms at 0. + mock_eth.get_logs.assert_called_once() + call_args = mock_eth.get_logs.call_args[0][0] + assert call_args["fromBlock"] == 0 + assert call_args["toBlock"] == 5_000 + + @pytest.mark.asyncio + async def test_get_transfer_logs_breaks_on_pruned_history( + self, + funding_tracer: FundingTracer, + mock_polygon_client: MagicMock, + ) -> None: + """A pruned-history error must short-circuit the whole walk. + + Public Polygon RPCs prune log history. Once we walk past the cutoff, + every subsequent chunk will raise the same error — keep walking and + we just burn quota on guaranteed failures. The first such error must + end the walk and return whatever we already collected. + """ + mock_w3 = MagicMock() + good_log = MagicMock() + + responses: list[Any] = [ + [good_log], + RuntimeError( + "{'code': -32701, 'message': 'History has been pruned for " + "this block. To remove restrictions, order a dedicated full " + "node here: https://www.allnodes.com/pol/host'}" + ), + # If the early-break logic is missing, this third chunk would + # also be requested. The test asserts it isn't. + [MagicMock()], + ] + + async def fake_get_logs(_params: dict[str, Any]) -> list[Any]: + outcome = responses.pop(0) + if isinstance(outcome, BaseException): + raise outcome + return outcome + + mock_w3.eth.get_logs = AsyncMock(side_effect=fake_get_logs) + mock_polygon_client._w3 = mock_w3 + + # 3 chunks total. The pruned error fires on chunk #2; chunk #3 must + # never be issued. + result = await funding_tracer._get_transfer_logs( + to_address=TEST_WALLET, + token_address=USDC_BRIDGED, + from_block=1_000_000, + to_block=1_027_000, + ) + + assert result == [dict(good_log)] + assert mock_w3.eth.get_logs.call_count == 2 + + @pytest.mark.asyncio + async def test_get_transfer_logs_default_lookback_fits_pruned_horizon( + self, + ) -> None: + """Default ``max_lookback_blocks`` must stay inside what public RPCs serve. + + publicnode prunes after ~100k blocks. If we default to 1.3M, every + funding trace blows through the archive horizon and produces nothing + but pruned-history warnings. Pin the default at <= 100k as a + regression guard. + """ + from polymarket_insider_tracker.profiler.funding import ( + DEFAULT_MAX_LOOKBACK_BLOCKS, + ) + + assert DEFAULT_MAX_LOOKBACK_BLOCKS <= 100_000 + + @pytest.mark.asyncio + async def test_get_transfer_logs_topic_is_0x_prefixed( + self, + funding_tracer: FundingTracer, + mock_polygon_client: MagicMock, + ) -> None: + """The Transfer event topic passed to ``eth_getLogs`` must begin with ``0x``. + + ``HexBytes.hex()`` returns a bare hex string. publicnode tolerates + that, but stricter providers like drpc (our fallback) reject it with + ``invalid argument 0: hex string without 0x prefix`` and every chunk + in the trace fails. This guards against regressing back to the + bare-hex form. + """ + mock_w3 = MagicMock() + mock_w3.eth.get_logs = AsyncMock(return_value=[]) + mock_polygon_client._w3 = mock_w3 + + await funding_tracer._get_transfer_logs( + to_address=TEST_WALLET, + token_address=USDC_BRIDGED, + from_block=1, + to_block=8_000, + ) + + topics = mock_w3.eth.get_logs.call_args[0][0]["topics"] + assert topics[0].startswith("0x") + # And the topic also has to be 32 bytes (64 hex chars) as required by + # the JSON-RPC spec. + assert len(topics[0]) == 2 + 64 + # The padded `to` topic was already 0x-prefixed; double-check that + # didn't regress either. + assert topics[2].startswith("0x") + class TestLogToFundingTransfer: """Tests for _log_to_funding_transfer method."""