From 58e21a828d0b34654fbd768832df71ff705a319a Mon Sep 17 00:00:00 2001 From: Marc Pont Date: Thu, 27 Aug 2026 19:38:51 +0200 Subject: [PATCH] Fix: mirror funds only ever traded in a burst at startup FundTradeListener advanced its highwater mark to now() on every poll and selected trades whose ts fell in that window. Trades reach ClickHouse minutes after they happen, so by the time one lands the mark is already past its ts and no window can contain it. Only the startup lookback ever saw anything. The window is over ingested_at now, and pollForTrades is synchronized because the multi-fund scheduler overlaps invocations. --- .../fund/service/FundTradeListener.java | 28 ++++++++++++++----- 1 file changed, 21 insertions(+), 7 deletions(-) diff --git a/strategy-service/src/main/java/com/polybot/hft/polymarket/fund/service/FundTradeListener.java b/strategy-service/src/main/java/com/polybot/hft/polymarket/fund/service/FundTradeListener.java index 2ba1c4f..1bd874e 100644 --- a/strategy-service/src/main/java/com/polybot/hft/polymarket/fund/service/FundTradeListener.java +++ b/strategy-service/src/main/java/com/polybot/hft/polymarket/fund/service/FundTradeListener.java @@ -96,7 +96,13 @@ public FundTradeListener( * Runs every second to catch trades quickly. */ @Scheduled(fixedRate = 1000) - public void pollForTrades() { + // Synchronized because the multi-fund scheduler can overlap invocations: + // pollPsiMirrorFunds runs at a fixed 1s rate over every fund, and when a + // round takes longer than that the next one starts on another thread. Two + // threads then read the same lastPollTime, query the same window and queue + // the same trade twice, which was visible in the logs as one signal emitted + // by two scheduling threads a millisecond apart. + public synchronized void pollForTrades() { if (!config.enabled()) { return; } @@ -133,7 +139,15 @@ private List fetchNewTrades() { .map(IndexConstituent::proxyAddress) .toList(); - // Initialize polling window + // Initialize polling window. + // + // The window is over ingested_at, not ts. Trades reach ClickHouse + // several minutes after they happen — the observed lag is a steady + // ~4.5 minutes — so a window over ts that advances to now() each second + // can never contain them: by the time a trade lands, the mark has long + // passed its ts. The only trades ever seen were the ones already in the + // table when the startup lookback ran, which is why the funds fired a + // burst on boot and then went quiet for hours. Instant now = clock.instant(); if (lastPollTime == null) { // Start from 1 hour ago on first poll to catch recent trader activity @@ -147,8 +161,8 @@ private List fetchNewTrades() { tradesPolledCounter.increment(trades.size()); if (!trades.isEmpty()) { - log.info("Found {} new trades from PSI-10 traders (window: {} to {})", - trades.size(), lastPollTime, now); + log.info("Found {} new trades from {} traders (ingest window: {} to {})", + trades.size(), config.indexType(), lastPollTime, now); } // Update highwater mark @@ -200,7 +214,7 @@ private List queryTrades(List addresses, Instant from, Instant // Format timestamps for ClickHouse DateTime64 java.time.format.DateTimeFormatter fmt = java.time.format.DateTimeFormatter - .ofPattern("yyyy-MM-dd HH:mm:ss") + .ofPattern("yyyy-MM-dd HH:mm:ss.SSS") .withZone(java.time.ZoneOffset.UTC); String fromStr = fmt.format(from); String toStr = fmt.format(to); @@ -220,8 +234,8 @@ private List queryTrades(List addresses, Instant from, Instant notional FROM polybot.aware_global_trades_dedup WHERE proxy_address IN (%s) - AND ts > toDateTime('%s') - AND ts <= toDateTime('%s') + AND ingested_at > toDateTime64('%s', 3) + AND ingested_at <= toDateTime64('%s', 3) ORDER BY ts LIMIT 500 """.formatted(addressList, fromStr, toStr);