Enable scoped Polymarket market-data tape - #9
Conversation
📝 WalkthroughWalkthroughAdds Polymarket market discovery filtering, Gamma response decoding, bounded NDJSON recording, expired-quote filtering, a no-op strategy, Rustls initialization, and a hardened dry-run systemd deployment for selected crypto markets. ChangesPolymarket market tape
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Systemd
participant Runner
participant MarketScanner
participant StrategyRuntime
participant RecordingFeed
participant NoopStrategy
Systemd->>Runner: start dry-run service
Runner->>MarketScanner: discover configured crypto markets
MarketScanner->>StrategyRuntime: emit normalized MarketUpdate
StrategyRuntime->>StrategyRuntime: reject expired event-token quotes
StrategyRuntime->>RecordingFeed: forward active updates
RecordingFeed->>NoopStrategy: forward MarketUpdate
NoopStrategy-->>Runner: emit no trading intent
RecordingFeed-->>Runner: persist permitted NDJSON updates
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@products/ploy/crates/ploy-market-data/src/scanner.rs`:
- Around line 168-170: Update the debug log near the crypto_markets_request call
to remove the hard-coded “next 6 minutes” text and reflect
DEFAULT_DISCOVERY_LOOKAHEAD_MINUTES, keeping the message aligned with the actual
discovery request.
In `@products/ploy/crates/ploy-strategy-bundles/src/feed/recorded.rs`:
- Around line 279-354: Update the EventExpired branch of prepare_recorded_update
to remove each expired token from last_quote_recorded_at alongside event_tokens
and quote_token_end_times. Keep the existing cleanup behavior for both tokens
and preserve quote sampling for active events.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 6734ce73-71e4-4245-8850-6fb0d387540f
⛔ Files ignored due to path filters (1)
products/ploy/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (16)
deployment/aliyun/README.mddeployment/aliyun/polymarket-market-tape.servicedeployment/aliyun/polymarket-market-tape.tomlproducts/ploy/crates/ploy-market-data/src/discovery/crypto.rsproducts/ploy/crates/ploy-market-data/src/gamma_keyset.rsproducts/ploy/crates/ploy-market-data/src/scanner.rsproducts/ploy/crates/ploy-runner-host/Cargo.tomlproducts/ploy/crates/ploy-runner-host/src/lib.rsproducts/ploy/crates/ploy-strategy-bundles/src/config.rsproducts/ploy/crates/ploy-strategy-bundles/src/feed/mod.rsproducts/ploy/crates/ploy-strategy-bundles/src/feed/recorded.rsproducts/ploy/crates/ploy-strategy-bundles/src/strategies/directional.rsproducts/ploy/crates/ploy-strategy-bundles/src/strategies/registry.rsproducts/ploy/crates/ploy-strategy-runtime/src/live.rsproducts/ploy/tasks/todo.mdproducts/ploy/tests/workspace_runtime_retirement.rs
| expire_tracked_events(&tx, &mut tracked, pool.as_ref(), now).await; | ||
|
|
||
| let request = crypto_markets_request(now, 6); | ||
| let request = crypto_markets_request(now, DEFAULT_DISCOVERY_LOOKAHEAD_MINUTES); |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Stale log text after lookahead change.
The discovery lookahead now uses DEFAULT_DISCOVERY_LOOKAHEAD_MINUTES (20 minutes), but the debug log at Line 180 still hard-codes "the next 6 minutes", which will mislead anyone reading logs during the dry-run rollover verification.
🪵 Proposed fix
- if discovered.is_empty() {
- debug!("No active crypto markets in the next 6 minutes");
- }
+ if discovered.is_empty() {
+ debug!(
+ lookahead_minutes = DEFAULT_DISCOVERY_LOOKAHEAD_MINUTES,
+ "No active crypto markets in the discovery lookahead window"
+ );
+ }Also applies to: 180-180
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@products/ploy/crates/ploy-market-data/src/scanner.rs` around lines 168 - 170,
Update the debug log near the crypto_markets_request call to remove the
hard-coded “next 6 minutes” text and reflect
DEFAULT_DISCOVERY_LOOKAHEAD_MINUTES, keeping the message aligned with the actual
discovery request.
| fn prepare_recorded_update(&mut self, update: &MarketUpdate) -> Option<MarketUpdate> { | ||
| if self.policy.event_scoped_quotes { | ||
| match update { | ||
| MarketUpdate::EventDiscovered { | ||
| event_id, | ||
| up_token, | ||
| down_token, | ||
| end_time, | ||
| .. | ||
| } => { | ||
| let tokens = [up_token.to_string(), down_token.to_string()]; | ||
| for token in &tokens { | ||
| self.quote_token_end_times.insert(token.clone(), *end_time); | ||
| } | ||
| self.event_tokens.insert(event_id.to_string(), tokens); | ||
| } | ||
| MarketUpdate::EventExpired { event_id, .. } => { | ||
| if let Some(tokens) = self.event_tokens.remove(event_id.as_ref()) { | ||
| for token in tokens { | ||
| self.quote_token_end_times.remove(&token); | ||
| } | ||
| } | ||
| } | ||
| MarketUpdate::Quote { token_id, ts, .. } => { | ||
| if !self | ||
| .quote_token_end_times | ||
| .get(token_id.as_ref()) | ||
| .is_some_and(|end_time| *ts <= *end_time) | ||
| { | ||
| return None; | ||
| } | ||
| } | ||
| _ => {} | ||
| } | ||
| } | ||
|
|
||
| if !self.policy.include_kinds.is_empty() | ||
| && !self | ||
| .policy | ||
| .include_kinds | ||
| .iter() | ||
| .any(|kind| kind.matches(update)) | ||
| { | ||
| return None; | ||
| } | ||
|
|
||
| let mut recorded = update.clone(); | ||
| if let MarketUpdate::Quote { | ||
| token_id, | ||
| bid_levels, | ||
| ask_levels, | ||
| ts, | ||
| .. | ||
| } = &mut recorded | ||
| { | ||
| if let Some(sample_ms) = self.policy.quote_sample_ms.filter(|value| *value > 0) { | ||
| let sample_ms = i64::try_from(sample_ms).unwrap_or(i64::MAX); | ||
| if self | ||
| .last_quote_recorded_at | ||
| .get(token_id.as_ref()) | ||
| .is_some_and(|last| *ts < *last + chrono::Duration::milliseconds(sample_ms)) | ||
| { | ||
| return None; | ||
| } | ||
| self.last_quote_recorded_at | ||
| .insert(token_id.to_string(), *ts); | ||
| } | ||
|
|
||
| if let Some(depth) = self.policy.quote_depth_levels { | ||
| bid_levels.truncate(depth); | ||
| ask_levels.truncate(depth); | ||
| } | ||
| } | ||
|
|
||
| Some(recorded) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
last_quote_recorded_at is never pruned on event expiry — unbounded growth.
The EventExpired branch cleans up event_tokens and quote_token_end_times but leaves last_quote_recorded_at entries for the expired tokens in place forever. Since RecordingLimits fields are optional (unbounded when unset) and this service is meant to run continuously across a rolling stream of 5-/15-minute crypto events, this map grows without bound for the lifetime of the process.
🧹 Proposed fix
MarketUpdate::EventExpired { event_id, .. } => {
if let Some(tokens) = self.event_tokens.remove(event_id.as_ref()) {
for token in tokens {
self.quote_token_end_times.remove(&token);
+ self.last_quote_recorded_at.remove(&token);
}
}
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| fn prepare_recorded_update(&mut self, update: &MarketUpdate) -> Option<MarketUpdate> { | |
| if self.policy.event_scoped_quotes { | |
| match update { | |
| MarketUpdate::EventDiscovered { | |
| event_id, | |
| up_token, | |
| down_token, | |
| end_time, | |
| .. | |
| } => { | |
| let tokens = [up_token.to_string(), down_token.to_string()]; | |
| for token in &tokens { | |
| self.quote_token_end_times.insert(token.clone(), *end_time); | |
| } | |
| self.event_tokens.insert(event_id.to_string(), tokens); | |
| } | |
| MarketUpdate::EventExpired { event_id, .. } => { | |
| if let Some(tokens) = self.event_tokens.remove(event_id.as_ref()) { | |
| for token in tokens { | |
| self.quote_token_end_times.remove(&token); | |
| } | |
| } | |
| } | |
| MarketUpdate::Quote { token_id, ts, .. } => { | |
| if !self | |
| .quote_token_end_times | |
| .get(token_id.as_ref()) | |
| .is_some_and(|end_time| *ts <= *end_time) | |
| { | |
| return None; | |
| } | |
| } | |
| _ => {} | |
| } | |
| } | |
| if !self.policy.include_kinds.is_empty() | |
| && !self | |
| .policy | |
| .include_kinds | |
| .iter() | |
| .any(|kind| kind.matches(update)) | |
| { | |
| return None; | |
| } | |
| let mut recorded = update.clone(); | |
| if let MarketUpdate::Quote { | |
| token_id, | |
| bid_levels, | |
| ask_levels, | |
| ts, | |
| .. | |
| } = &mut recorded | |
| { | |
| if let Some(sample_ms) = self.policy.quote_sample_ms.filter(|value| *value > 0) { | |
| let sample_ms = i64::try_from(sample_ms).unwrap_or(i64::MAX); | |
| if self | |
| .last_quote_recorded_at | |
| .get(token_id.as_ref()) | |
| .is_some_and(|last| *ts < *last + chrono::Duration::milliseconds(sample_ms)) | |
| { | |
| return None; | |
| } | |
| self.last_quote_recorded_at | |
| .insert(token_id.to_string(), *ts); | |
| } | |
| if let Some(depth) = self.policy.quote_depth_levels { | |
| bid_levels.truncate(depth); | |
| ask_levels.truncate(depth); | |
| } | |
| } | |
| Some(recorded) | |
| } | |
| fn prepare_recorded_update(&mut self, update: &MarketUpdate) -> Option<MarketUpdate> { | |
| if self.policy.event_scoped_quotes { | |
| match update { | |
| MarketUpdate::EventDiscovered { | |
| event_id, | |
| up_token, | |
| down_token, | |
| end_time, | |
| .. | |
| } => { | |
| let tokens = [up_token.to_string(), down_token.to_string()]; | |
| for token in &tokens { | |
| self.quote_token_end_times.insert(token.clone(), *end_time); | |
| } | |
| self.event_tokens.insert(event_id.to_string(), tokens); | |
| } | |
| MarketUpdate::EventExpired { event_id, .. } => { | |
| if let Some(tokens) = self.event_tokens.remove(event_id.as_ref()) { | |
| for token in tokens { | |
| self.quote_token_end_times.remove(&token); | |
| self.last_quote_recorded_at.remove(&token); | |
| } | |
| } | |
| } | |
| MarketUpdate::Quote { token_id, ts, .. } => { | |
| if !self | |
| .quote_token_end_times | |
| .get(token_id.as_ref()) | |
| .is_some_and(|end_time| *ts <= *end_time) | |
| { | |
| return None; | |
| } | |
| } | |
| _ => {} | |
| } | |
| } | |
| if !self.policy.include_kinds.is_empty() | |
| && !self | |
| .policy | |
| .include_kinds | |
| .iter() | |
| .any(|kind| kind.matches(update)) | |
| { | |
| return None; | |
| } | |
| let mut recorded = update.clone(); | |
| if let MarketUpdate::Quote { | |
| token_id, | |
| bid_levels, | |
| ask_levels, | |
| ts, | |
| .. | |
| } = &mut recorded | |
| { | |
| if let Some(sample_ms) = self.policy.quote_sample_ms.filter(|value| *value > 0) { | |
| let sample_ms = i64::try_from(sample_ms).unwrap_or(i64::MAX); | |
| if self | |
| .last_quote_recorded_at | |
| .get(token_id.as_ref()) | |
| .is_some_and(|last| *ts < *last + chrono::Duration::milliseconds(sample_ms)) | |
| { | |
| return None; | |
| } | |
| self.last_quote_recorded_at | |
| .insert(token_id.to_string(), *ts); | |
| } | |
| if let Some(depth) = self.policy.quote_depth_levels { | |
| bid_levels.truncate(depth); | |
| ask_levels.truncate(depth); | |
| } | |
| } | |
| Some(recorded) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@products/ploy/crates/ploy-strategy-bundles/src/feed/recorded.rs` around lines
279 - 354, Update the EventExpired branch of prepare_recorded_update to remove
each expired token from last_quote_recorded_at alongside event_tokens and
quote_token_end_times. Keep the existing cleanup behavior for both tokens and
preserve quote sampling for active events.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: b19beacebd
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| PrivateTmp=true | ||
| ProtectSystem=strict | ||
| ProtectHome=true | ||
| ReadWritePaths=/data/monday/spool/polymarket |
There was a problem hiding this comment.
On a fresh ECS host where /data/monday/spool/polymarket has not already been created, this unit starts the runner under ProtectSystem=strict while only granting write access to the leaf directory. The recorder calls create_dir_all for the parent of market-updates.ndjson, but creating the missing polymarket directory requires write permission on /data/monday/spool, which remains read-only here; the service will fail before recording anything unless deployment pre-creates that exact leaf. Grant the parent spool path or add an explicit pre-start directory creation step.
Useful? React with 👍 / 👎.
| --dry-run | ||
| Restart=always | ||
| RestartSec=5 | ||
| RuntimeMaxSec=21600 |
There was a problem hiding this comment.
When this six-hour RuntimeMaxSec recycle fires, systemd terminates new-ploy-runner; unlike the existing Python archiver, the Rust runner has no SIGTERM/ctrl_c shutdown path, and the NDJSON writer only flushes on lifecycle updates or every 256 records. That means a scheduled restart can drop the last buffered quote samples before the next process rotates the file, creating small holes in the tape. Add graceful signal handling/explicit flushing, or avoid timed restarts for this buffered recorder.
Useful? React with 👍 / 👎.
| ExecStart=/opt/monday/bin/new-ploy-runner \ | ||
| --config /etc/monday/polymarket-market-tape.toml \ | ||
| --deployment-id polymarket-market-data-ecs \ | ||
| --dry-run |
There was a problem hiding this comment.
If /opt/monday/bin/new-ploy-runner is installed from the package's default Cargo features, this dry-run service exits before opening the feeds because the default runner is lean-replay and RuntimeMode::DryRun is only wired when the live/full feature is built. The archived runner docs use --features full, but this new active unit/README path does not pin or verify that artifact, so a normal cargo build -p new-ploy-runner --release deployment produces a binary that cannot run this service. Make the install path build/verify --features full or point the unit at a distinct full-feature binary.
Useful? React with 👍 / 👎.
| if market_start_time.is_some_and(|start_time| start_time > now) { | ||
| return None; | ||
| } |
There was a problem hiding this comment.
When a 5-minute or 15-minute market starts just after a scanner poll, this filter hides it until the next 30-second scanner iteration, so the quote feed is not subscribed for the opening portion of the market and price_to_beat is captured after the true start. For a 5-minute tape that can drop about 10% of the window even though the service is intended to collect active market data; discover the market ahead of time and gate only persisted pre-start quotes instead.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
🧹 Nitpick comments (1)
products/ploy/crates/ploy-strategy-bundles/src/engine.rs (1)
210-212: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win
retired_quote_tokensgrows without bound for the lifetime of the runtime.
quote_token_end_times/event_tokensare cleaned up onEventExpired(Lines 241-242), butretired_quote_tokensonly ever grows — tokens are removed from it solely on rediscovery of the exact same token id (Line 232-233), which won't happen since Polymarket token ids are unique per market outcome. For a continuously-running ECS collector across 4 assets on 5m/15m windows, thisHashSet<String>accumulates roughly two new entries per expired event indefinitely, with no eviction path.Given the PR intent is an always-on dry-run collector, consider bounding this (e.g., store retirement time and periodically sweep entries older than a safety window) instead of an ever-growing set.
♻️ Sketch: bound retired-token memory with a time-based sweep
- let mut retired_quote_tokens: HashSet<String> = HashSet::new(); + let mut retired_quote_tokens: HashMap<String, DateTime<Utc>> = HashMap::new(); ... MarketUpdate::EventExpired { event_id, .. } => { if let Some(tokens) = event_tokens.remove(event_id.as_ref()) { for token in tokens { quote_token_end_times.remove(&token); - retired_quote_tokens.insert(token); + retired_quote_tokens.insert(token, *end_time); } } } MarketUpdate::Quote { token_id, ts, .. } => { - let expired = retired_quote_tokens.contains(token_id.as_ref()) + let expired = retired_quote_tokens.contains_key(token_id.as_ref()) || quote_token_end_times .get(token_id.as_ref()) .is_some_and(|end_time| *ts > *end_time); + // Periodically: retired_quote_tokens.retain(|_, t| *ts - *t < Duration::hours(24));Also applies to: 238-267
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@products/ploy/crates/ploy-strategy-bundles/src/engine.rs` around lines 210 - 212, Bound the lifetime of entries tracked by retired_quote_tokens instead of retaining every retired token indefinitely. Update the retirement tracking and EventExpired cleanup flow around event_tokens and quote_token_end_times to retain each token only for a suitable safety window, periodically evicting older entries while preserving protection against timely rediscovery of recently retired tokens.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@products/ploy/crates/ploy-strategy-bundles/src/engine.rs`:
- Around line 210-212: Bound the lifetime of entries tracked by
retired_quote_tokens instead of retaining every retired token indefinitely.
Update the retirement tracking and EventExpired cleanup flow around event_tokens
and quote_token_end_times to retain each token only for a suitable safety
window, periodically evicting older entries while preserving protection against
timely rediscovery of recently retired tokens.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: c14a96f3-3dc2-46e8-86a9-cba54b9310c9
📒 Files selected for processing (3)
deployment/aliyun/README.mdproducts/ploy/crates/ploy-strategy-bundles/src/engine.rsproducts/ploy/crates/ploy-strategy-bundles/src/strategies/directional.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- deployment/aliyun/README.md
Summary
Safety
Verification
Summary by CodeRabbit
noopstrategy variant that observes data without placing orders.