diff --git a/src/polymarket_insider_tracker/profiler/funding.py b/src/polymarket_insider_tracker/profiler/funding.py index a688c2b..03d64f7 100644 --- a/src/polymarket_insider_tracker/profiler/funding.py +++ b/src/polymarket_insider_tracker/profiler/funding.py @@ -29,6 +29,13 @@ USDC_NATIVE = "0x3c499c542cEF5E3811e1192ce70d8cC03d5c3359" # ERC20 Transfer event signature TRANSFER_EVENT_SIGNATURE = AsyncWeb3.keccak(text="Transfer(address,address,uint256)") +# 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, so 1.3M blocks covers roughly 30 days. +DEFAULT_MAX_LOOKBACK_BLOCKS = 1_300_000 + class FundingTracer: """Traces funding chains to identify wallet funding sources. @@ -50,6 +57,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 +67,10 @@ 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 ~30 days at 2s block time. """ self.polygon_client = polygon_client self.entity_registry = entity_registry or EntityRegistry() @@ -65,6 +78,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 +224,128 @@ 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. + 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_SIGNATURE.hex(), # Transfer event + 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: + 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..79f942c 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() @@ -338,6 +342,9 @@ class TestGetTransferLogs: 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 +362,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 +383,142 @@ 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 + class TestLogToFundingTransfer: """Tests for _log_to_funding_transfer method."""