Skip to content

Fix: mirror funds only ever traded in a burst at startup #13

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mpont91 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Fix: mirror funds only ever traded in a burst at startup #13

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 21 additions & 7 deletions ...tegy-service/src/main/java/com/polybot/hft/polymarket/fund/service/FundTradeListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Expand Down Expand Up @@ -133,7 +139,15 @@ private List<TraderSignal> 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
Expand All @@ -147,8 +161,8 @@ private List<TraderSignal> 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
Expand Down Expand Up @@ -200,7 +214,7 @@ private List<RawTrade> queryTrades(List<String> 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);
Expand All @@ -220,8 +234,8 @@ private List<RawTrade> queryTrades(List<String> 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);
Expand Down