gavindiaz/polyfill-rs
1
0
Fork 0
mirror of https://github.com/floor-licker/polyfill-rs.git synced 2026-08-14 04:58:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: resolve rustfmt configuration duplicate key error and apply consistent code formatting across all source files

Browse Source
This commit is contained in:
floor-licker
2025-12-05 19:09:06 -05:00
parent 5576d765ee
commit 9993e51c7f
29 changed files with 2540 additions and 1673 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/auth.rs
+39 -49
View File
@@ -54,7 +54,6 @@ sol! {
}
}
/// Get current Unix timestamp in seconds
pub fn get_current_unix_time_secs() -> u64 {
SystemTime::now()
@@ -134,8 +133,10 @@ where
method.to_uppercase(),
request_path,
match body {
Some(b) => serde_json::to_string(b)
.map_err(|e| PolyfillError::parse(format!("Failed to serialize body: {}", e), None))?,
Some(b) => serde_json::to_string(b).map_err(|e| PolyfillError::parse(
format!("Failed to serialize body: {}", e),
None
))?,
None => String::new(),
}
);
@@ -174,7 +175,8 @@ where
let address = encode_prefixed(signer.address().as_slice());
let timestamp = get_current_unix_time_secs();
let hmac_signature = build_hmac_signature(&api_creds.secret, timestamp, method, req_path, body)?;
let hmac_signature =
build_hmac_signature(&api_creds.secret, timestamp, method, req_path, body)?;
Ok(HashMap::from([
(POLY_ADDR_HEADER, address),
@@ -197,26 +199,15 @@ mod tests {
#[test]
fn test_hmac_signature() {
let result = build_hmac_signature::<String>(
"test_secret",
1234567890,
"GET",
"/test",
None,
);
let result =
build_hmac_signature::<String>("test_secret", 1234567890, "GET", "/test", None);
assert!(result.is_ok());
}
#[test]
fn test_hmac_signature_with_body() {
let body = r#"{"test": "data"}"#;
let result = build_hmac_signature(
"test_secret",
1234567890,
"POST",
"/orders",
Some(body),
);
let result = build_hmac_signature("test_secret", 1234567890, "POST", "/orders", Some(body));
assert!(result.is_ok());
let signature = result.unwrap();
assert!(!signature.is_empty());
@@ -228,10 +219,10 @@ mod tests {
let timestamp = 1234567890;
let method = "GET";
let path = "/test";
let sig1 = build_hmac_signature::<String>(secret, timestamp, method, path, None).unwrap();
let sig2 = build_hmac_signature::<String>(secret, timestamp, method, path, None).unwrap();
// Same inputs should produce same signature
assert_eq!(sig1, sig2);
}
@@ -240,11 +231,13 @@ mod tests {
fn test_hmac_signature_different_inputs() {
let secret = "test_secret";
let timestamp = 1234567890;
let sig1 = build_hmac_signature::<String>(secret, timestamp, "GET", "/test", None).unwrap();
let sig2 = build_hmac_signature::<String>(secret, timestamp, "POST", "/test", None).unwrap();
let sig3 = build_hmac_signature::<String>(secret, timestamp, "GET", "/other", None).unwrap();
let sig2 =
build_hmac_signature::<String>(secret, timestamp, "POST", "/test", None).unwrap();
let sig3 =
build_hmac_signature::<String>(secret, timestamp, "GET", "/other", None).unwrap();
// Different inputs should produce different signatures
assert_ne!(sig1, sig2);
assert_ne!(sig1, sig3);
@@ -253,15 +246,15 @@ mod tests {
#[test]
fn test_create_l1_headers() {
use alloy_signer_local::PrivateKeySigner;
use alloy_primitives::U256;
use alloy_signer_local::PrivateKeySigner;
let private_key = "0x1234567890123456789012345678901234567890123456789012345678901234";
let signer: PrivateKeySigner = private_key.parse().expect("Valid private key");
let result = create_l1_headers(&signer, Some(U256::from(12345)));
assert!(result.is_ok());
let headers = result.unwrap();
assert!(headers.contains_key("poly_address"));
assert!(headers.contains_key("poly_signature"));
@@ -271,69 +264,66 @@ mod tests {
#[test]
fn test_create_l1_headers_different_nonces() {
use alloy_signer_local::PrivateKeySigner;
use alloy_primitives::U256;
use alloy_signer_local::PrivateKeySigner;
let private_key = "0x1234567890123456789012345678901234567890123456789012345678901234";
let signer: PrivateKeySigner = private_key.parse().expect("Valid private key");
let headers_1 = create_l1_headers(&signer, Some(U256::from(12345))).unwrap();
let headers_2 = create_l1_headers(&signer, Some(U256::from(54321))).unwrap();
// Different nonces should produce different signatures
assert_ne!(
headers_1.get("poly_signature"),
headers_2.get("poly_signature")
);
// But same address
assert_eq!(
headers_1.get("poly_address"),
headers_2.get("poly_address")
);
assert_eq!(headers_1.get("poly_address"), headers_2.get("poly_address"));
}
#[test]
fn test_create_l2_headers() {
use alloy_signer_local::PrivateKeySigner;
let private_key = "0x1234567890123456789012345678901234567890123456789012345678901234";
let signer: PrivateKeySigner = private_key.parse().expect("Valid private key");
let api_creds = ApiCredentials {
api_key: "test_key".to_string(),
secret: "test_secret".to_string(),
passphrase: "test_passphrase".to_string(),
};
let result = create_l2_headers::<String>(&signer, &api_creds, "/test", "GET", None);
assert!(result.is_ok());
let headers = result.unwrap();
assert!(headers.contains_key("poly_api_key"));
assert!(headers.contains_key("poly_signature"));
assert!(headers.contains_key("poly_timestamp"));
assert!(headers.contains_key("poly_passphrase"));
assert_eq!(headers.get("poly_api_key").unwrap(), "test_key");
assert_eq!(headers.get("poly_passphrase").unwrap(), "test_passphrase");
}
#[test]
fn test_eip712_signature_format() {
use alloy_signer_local::PrivateKeySigner;
use alloy_primitives::U256;
use alloy_signer_local::PrivateKeySigner;
let private_key = "0x1234567890123456789012345678901234567890123456789012345678901234";
let signer: PrivateKeySigner = private_key.parse().expect("Valid private key");
// Test that we can create and sign EIP-712 messages
let result = create_l1_headers(&signer, Some(U256::from(12345)));
assert!(result.is_ok());
let headers = result.unwrap();
let signature = headers.get("poly_signature").unwrap();
// EIP-712 signatures should be hex strings of specific length
assert!(signature.starts_with("0x"));
assert_eq!(signature.len(), 132); // 0x + 130 hex chars = 132 total
@@ -344,10 +334,10 @@ mod tests {
let ts1 = get_current_unix_time_secs();
std::thread::sleep(std::time::Duration::from_millis(1));
let ts2 = get_current_unix_time_secs();
// Timestamps should be increasing
assert!(ts2 >= ts1);
// Should be reasonable current time (after 2020, before 2030)
assert!(ts1 > 1_600_000_000);
assert!(ts1 < 1_900_000_000);
src/book.rs
+272 -181
View File
@@ -3,20 +3,20 @@
use crate::errors::{PolyfillError, Result};
use crate::types::*;
use crate::utils::math;
use chrono::Utc;
use rust_decimal::Decimal;
use std::collections::BTreeMap; // BTreeMap keeps prices sorted automatically - crucial for order books
use std::sync::{Arc, RwLock}; // For thread-safe access across multiple tasks
use tracing::{debug, trace, warn}; // Logging for debugging and monitoring
use chrono::Utc;
/// High-performance order book implementation
///
///
/// This is the core data structure that holds all the live buy/sell orders for a token.
/// The efficiency of this code is critical as the order book is constantly being updated as orders are added and removed.
///
///
/// PERFORMANCE OPTIMIZATION: This struct now uses fixed-point integers internally
/// instead of Decimal for maximum speed. The performance difference is dramatic:
///
///
/// Before (Decimal): ~100ns per operation + memory allocation
/// After (fixed-point): ~5ns per operation, zero allocations
@@ -24,51 +24,51 @@ use chrono::Utc;
pub struct OrderBook {
/// Token ID this book represents (like "123456" for a specific prediction market outcome)
pub token_id: String,
/// Hash of token_id for fast lookups (avoids string comparisons in hot path)
pub token_id_hash: u64,
/// Current sequence number for ordering updates
/// This helps us ignore old/duplicate updates that arrive out of order
pub sequence: u64,
/// Last update timestamp - when we last got new data for this book
pub timestamp: chrono::DateTime<Utc>,
/// Bid side (price -> size, sorted descending) - NOW USING FIXED-POINT!
/// BTreeMap automatically keeps highest bids first, which is what we want
/// Key = price in ticks (like 6500 for $0.65), Value = size in fixed-point units
///
///
/// BEFORE (slow): bids: BTreeMap<Decimal, Decimal>,
/// AFTER (fast): bids: BTreeMap<Price, Qty>,
///
///
/// Why this is faster:
/// - Integer comparisons are ~10x faster than Decimal comparisons
/// - No memory allocation for each price level
/// - Better CPU cache utilization (smaller data structures)
bids: BTreeMap<Price, Qty>,
/// Ask side (price -> size, sorted ascending) - NOW USING FIXED-POINT!
/// BTreeMap keeps lowest asks first - people selling at cheapest prices
///
///
/// BEFORE (slow): asks: BTreeMap<Decimal, Decimal>,
/// AFTER (fast): asks: BTreeMap<Price, Qty>,
asks: BTreeMap<Price, Qty>,
/// Minimum tick size for this market in ticks (like 10 for $0.001 increments)
/// Some markets only allow certain price increments
/// We store this in ticks for fast validation without conversion
tick_size_ticks: Option<Price>,
/// Maximum depth to maintain (how many price levels to keep)
///
///
/// We don't need to track every single price level, just the best ones because:
/// - Trading reality 90% of volume happens in the top 5-10 price levels
/// - Execution priority: Orders get filled from best price first, so deep levels often don't matter
/// - Market efficiency: If you're buying and best ask is $0.67, you'll never pay $0.95
/// - Risk management: Large orders that would hit deep levels are usually broken up
/// - Data freshness: Deep levels often have stale orders from hours/days ago
///
///
/// Typical values: 10-50 for retail, 100-500 for institutional HFT systems
max_depth: usize,
}
@@ -85,7 +85,7 @@ impl OrderBook {
token_id.hash(&mut hasher);
hasher.finish()
};
Self {
token_id,
token_id_hash,
@@ -107,7 +107,7 @@ impl OrderBook {
self.tick_size_ticks = Some(tick_size_ticks);
Ok(())
}
/// Set the tick size directly in ticks (even faster)
/// Use this when you already have the tick size in our internal format
pub fn set_tick_size_ticks(&mut self, tick_size_ticks: Price) {
@@ -116,31 +116,34 @@ impl OrderBook {
/// Get the current best bid (highest price someone is willing to pay)
/// Uses next_back() because BTreeMap sorts ascending, but we want the highest bid
///
///
/// PERFORMANCE: Now returns data in external format but internally uses fast lookups
pub fn best_bid(&self) -> Option<BookLevel> {
// BEFORE (slow, ~50ns + allocation):
// self.bids.iter().next_back().map(|(&price, &size)| BookLevel { price, size })
// AFTER (fast, ~5ns, no allocation for the lookup):
self.bids.iter().next_back().map(|(&price_ticks, &size_units)| {
// Convert from internal fixed-point to external Decimal format
// This conversion only happens at the API boundary
BookLevel {
price: price_to_decimal(price_ticks),
size: qty_to_decimal(size_units),
}
})
self.bids
.iter()
.next_back()
.map(|(&price_ticks, &size_units)| {
// Convert from internal fixed-point to external Decimal format
// This conversion only happens at the API boundary
BookLevel {
price: price_to_decimal(price_ticks),
size: qty_to_decimal(size_units),
}
})
}
/// Get the current best ask (lowest price someone is willing to sell at)
/// Uses next() because BTreeMap sorts ascending, so first item is lowest ask
///
///
/// PERFORMANCE: Now returns data in external format but internally uses fast lookups
pub fn best_ask(&self) -> Option<BookLevel> {
// BEFORE (slow, ~50ns + allocation):
// self.asks.iter().next().map(|(&price, &size)| BookLevel { price, size })
// AFTER (fast, ~5ns, no allocation for the lookup):
self.asks.iter().next().map(|(&price_ticks, &size_units)| {
// Convert from internal fixed-point to external Decimal format
@@ -152,25 +155,27 @@ impl OrderBook {
})
}
/// Get the current best bid in fast internal format
/// Get the current best bid in fast internal format
/// Use this for internal calculations to avoid conversion overhead
pub fn best_bid_fast(&self) -> Option<FastBookLevel> {
self.bids.iter().next_back().map(|(&price, &size)| {
FastBookLevel::new(price, size)
})
self.bids
.iter()
.next_back()
.map(|(&price, &size)| FastBookLevel::new(price, size))
}
/// Get the current best ask in fast internal format
/// Get the current best ask in fast internal format
/// Use this for internal calculations to avoid conversion overhead
pub fn best_ask_fast(&self) -> Option<FastBookLevel> {
self.asks.iter().next().map(|(&price, &size)| {
FastBookLevel::new(price, size)
})
self.asks
.iter()
.next()
.map(|(&price, &size)| FastBookLevel::new(price, size))
}
/// Get the current spread (difference between best ask and best bid)
/// This tells us how "tight" the market is - smaller spread = more liquid market
///
///
/// PERFORMANCE: Now uses fast internal calculations, only converts to Decimal at the end
pub fn spread(&self) -> Option<Decimal> {
// BEFORE (slow, ~100ns + multiple allocations):
@@ -178,7 +183,7 @@ impl OrderBook {
// (Some(bid), Some(ask)) => Some(ask.price - bid.price),
// _ => None,
// }
// AFTER (fast, ~5ns, no allocations):
let (best_bid_ticks, best_ask_ticks) = self.best_prices_fast()?;
let spread_ticks = math::spread_fast(best_bid_ticks, best_ask_ticks)?;
@@ -187,7 +192,7 @@ impl OrderBook {
/// Get the current mid price (halfway between best bid and ask)
/// This is often used as the "fair value" of the market
///
///
/// PERFORMANCE: Now uses fast internal calculations, only converts to Decimal at the end
pub fn mid_price(&self) -> Option<Decimal> {
// BEFORE (slow, ~80ns + allocations):
@@ -195,7 +200,7 @@ impl OrderBook {
// self.best_bid()?.price,
// self.best_ask()?.price,
// )
// AFTER (fast, ~3ns, no allocations):
let (best_bid_ticks, best_ask_ticks) = self.best_prices_fast()?;
let mid_ticks = math::mid_price_fast(best_bid_ticks, best_ask_ticks)?;
@@ -204,7 +209,7 @@ impl OrderBook {
/// Get the spread as a percentage (relative to the bid price)
/// Useful for comparing spreads across different price levels
///
///
/// PERFORMANCE: Now uses fast internal calculations and returns basis points
pub fn spread_pct(&self) -> Option<Decimal> {
let (best_bid_ticks, best_ask_ticks) = self.best_prices_fast()?;
@@ -212,7 +217,7 @@ impl OrderBook {
// Convert basis points back to percentage decimal
Some(Decimal::from(spread_bps) / Decimal::from(100))
}
/// Get best bid and ask prices in fast internal format
/// Helper method to avoid code duplication and minimize conversions
fn best_prices_fast(&self) -> Option<(Price, Price)> {
@@ -220,14 +225,14 @@ impl OrderBook {
let best_ask_ticks = self.asks.iter().next()?.0;
Some((*best_bid_ticks, *best_ask_ticks))
}
/// Get the current spread in fast internal format (PERFORMANCE OPTIMIZED)
/// Returns spread in ticks - use this for internal calculations
pub fn spread_fast(&self) -> Option<Price> {
let (best_bid_ticks, best_ask_ticks) = self.best_prices_fast()?;
math::spread_fast(best_bid_ticks, best_ask_ticks)
}
/// Get the current mid price in fast internal format (PERFORMANCE OPTIMIZED)
/// Returns mid price in ticks - use this for internal calculations
pub fn mid_price_fast(&self) -> Option<Price> {
@@ -237,7 +242,7 @@ impl OrderBook {
/// Get all bids up to a certain depth (top N price levels)
/// Returns them in descending price order (best bids first)
///
///
/// PERFORMANCE: Converts from internal fixed-point to external Decimal format
/// Only call this when you need to return data to external APIs
pub fn bids(&self, depth: Option<usize>) -> Vec<BookLevel> {
@@ -255,7 +260,7 @@ impl OrderBook {
/// Get all asks up to a certain depth (top N price levels)
/// Returns them in ascending price order (best asks first)
///
///
/// PERFORMANCE: Converts from internal fixed-point to external Decimal format
/// Only call this when you need to return data to external APIs
pub fn asks(&self, depth: Option<usize>) -> Vec<BookLevel> {
@@ -269,8 +274,8 @@ impl OrderBook {
})
.collect()
}
/// Get all bids in fast internal format
/// Get all bids in fast internal format
/// Use this for internal calculations to avoid conversion overhead
pub fn bids_fast(&self, depth: Option<usize>) -> Vec<FastBookLevel> {
let depth = depth.unwrap_or(self.max_depth);
@@ -308,7 +313,7 @@ impl OrderBook {
/// Apply a delta update to the book (LEGACY VERSION - for external API compatibility)
/// A "delta" is an incremental change - like "add 100 tokens at $0.65" or "remove all at $0.70"
///
///
/// This method converts the external Decimal delta to our internal fixed-point format
/// and then calls the fast version. Use apply_delta_fast() directly when possible.
pub fn apply_delta(&mut self, delta: OrderDelta) -> Result<()> {
@@ -316,16 +321,16 @@ impl OrderBook {
let tick_size_decimal = self.tick_size_ticks.map(price_to_decimal);
let fast_delta = FastOrderDelta::from_order_delta(&delta, tick_size_decimal)
.map_err(|e| PolyfillError::validation(format!("Invalid delta: {}", e)))?;
// Use the fast internal version
self.apply_delta_fast(fast_delta)
}
/// Apply a delta update to the book
///
///
/// This is the high-performance version that works directly with fixed-point data.
/// It includes tick alignment validation and is much faster than the Decimal version.
///
///
/// Performance improvement: ~50x faster than the old Decimal version!
/// - No Decimal conversions in the hot path
/// - Integer comparisons instead of Decimal comparisons
@@ -334,7 +339,11 @@ impl OrderBook {
// Validate sequence ordering - ignore old updates that arrive late
// This is crucial for maintaining data integrity in real-time systems
if delta.sequence <= self.sequence {
trace!("Ignoring stale delta: {} <= {}", delta.sequence, self.sequence);
trace!(
"Ignoring stale delta: {} <= {}",
delta.sequence,
self.sequence
);
return Ok(());
}
@@ -351,7 +360,7 @@ impl OrderBook {
// if !is_price_tick_aligned(price_to_decimal(delta.price), tick_size_decimal) {
// return Err(...);
// }
// AFTER (fast, ~2ns, pure integer):
if tick_size_ticks > 0 && delta.price % tick_size_ticks != 0 {
// Price is not aligned to tick size - reject the update
@@ -390,7 +399,7 @@ impl OrderBook {
/// Apply a bid-side delta (someone wants to buy) - LEGACY VERSION
/// If size is 0, it means "remove this price level entirely"
/// Otherwise, set the total size at this price level
///
///
/// This converts to fixed-point and calls the fast version
#[allow(dead_code)]
fn apply_bid_delta(&mut self, price: Decimal, size: Decimal) {
@@ -402,7 +411,7 @@ impl OrderBook {
/// Apply an ask-side delta (someone wants to sell) - LEGACY VERSION
/// Same logic as bids - size of 0 means remove the price level
///
///
/// This converts to fixed-point and calls the fast version
#[allow(dead_code)]
fn apply_ask_delta(&mut self, price: Decimal, size: Decimal) {
@@ -411,9 +420,9 @@ impl OrderBook {
let size_units = decimal_to_qty(size).unwrap_or(0);
self.apply_ask_delta_fast(price_ticks, size_units);
}
/// Apply a bid-side delta (someone wants to buy) - FAST VERSION
///
///
/// This is the high-performance version that works directly with fixed-point.
/// Much faster than the Decimal version - pure integer operations.
fn apply_bid_delta_fast(&mut self, price_ticks: Price, size_units: Qty) {
@@ -423,7 +432,7 @@ impl OrderBook {
// } else {
// self.bids.insert(price, size);
// }
// AFTER (fast, ~5ns, no allocation):
if size_units == 0 {
self.bids.remove(&price_ticks); // No more buyers at this price
@@ -433,7 +442,7 @@ impl OrderBook {
}
/// Apply an ask-side delta (someone wants to sell) - FAST VERSION
///
///
/// This is the high-performance version that works directly with fixed-point.
/// Much faster than the Decimal version - pure integer operations.
fn apply_ask_delta_fast(&mut self, price_ticks: Price, size_units: Qty) {
@@ -443,7 +452,7 @@ impl OrderBook {
// } else {
// self.asks.insert(price, size);
// }
// AFTER (fast, ~5ns, no allocation):
if size_units == 0 {
self.asks.remove(&price_ticks); // No more sellers at this price
@@ -454,12 +463,12 @@ impl OrderBook {
/// Trim the book to maintain depth limits
/// We don't want to track every single price level - just the best ones
///
///
/// Why limit depth? Several reasons:
/// 1. Memory efficiency: A popular market might have thousands of price levels,
/// but only the top 10-50 levels are actually tradeable with reasonable size
/// 2. Performance: Fewer levels = faster iteration when calculating market impact
/// 3. Relevance: Deep levels (like bids at $0.01 when best bid is $0.65) are
/// 3. Relevance: Deep levels (like bids at $0.01 when best bid is $0.65) are
/// mostly noise and will never get hit in normal trading
/// 4. Stale data: Deep levels often contain old orders that haven't been cancelled
/// 5. Network bandwidth: Less data to send when streaming updates
@@ -473,7 +482,7 @@ impl OrderBook {
}
}
// For asks, remove the HIGHEST prices (worst asks) if we have too many
// For asks, remove the HIGHEST prices (worst asks) if we have too many
// Example: If best ask is $0.67, we don't care about asks at $0.95
if self.asks.len() > self.max_depth {
let to_remove = self.asks.len() - self.max_depth;
@@ -493,16 +502,16 @@ impl OrderBook {
pub fn calculate_market_impact(&self, side: Side, size: Decimal) -> Option<MarketImpact> {
// PERFORMANCE NOTE: This method still uses Decimal for external compatibility,
// but the internal order book lookups now use our fast fixed-point data structures.
//
//
// BEFORE: Each level lookup involved Decimal operations (~50ns each)
// AFTER: Level lookups use integer operations (~5ns each)
//
//
// For a 10-level impact calculation: 500ns → 50ns (10x speedup)
// Get the levels we'd be trading against
let levels = match side {
Side::BUY => self.asks(None), // If buying, we hit the ask side
Side::SELL => self.bids(None), // If selling, we hit the bid side
Side::BUY => self.asks(None), // If buying, we hit the ask side
Side::SELL => self.bids(None), // If selling, we hit the bid side
};
if levels.is_empty() {
@@ -517,7 +526,7 @@ impl OrderBook {
for level in levels {
let fill_size = std::cmp::min(remaining_size, level.size);
let level_cost = fill_size * level.price;
total_cost += level_cost;
weighted_price += level_cost; // This accumulates the weighted average
remaining_size -= fill_size;
@@ -532,21 +541,21 @@ impl OrderBook {
// This is a perfect example of why we don't need infinite depth:
// If we can't fill your order with the top N levels, you probably
// shouldn't be placing that order anyway - it would move the market too much
return None;
return None;
}
let avg_price = weighted_price / size;
// Calculate how much we moved the market compared to the best price
let impact = match side {
Side::BUY => {
let best_ask = self.best_ask()?.price;
(avg_price - best_ask) / best_ask // How much worse than best ask
}
},
Side::SELL => {
let best_bid = self.best_bid()?.price;
(best_bid - avg_price) / best_bid // How much worse than best bid
}
},
};
Some(MarketImpact {
@@ -572,7 +581,7 @@ impl OrderBook {
Ok(ticks) => ticks,
Err(_) => return Decimal::ZERO, // Invalid price
};
match side {
Side::BUY => {
// How much we can buy at this price (look at asks)
@@ -583,13 +592,18 @@ impl OrderBook {
// How much we can sell at this price (look at bids)
let size_units = self.bids.get(&price_ticks).copied().unwrap_or_default();
qty_to_decimal(size_units)
}
},
}
}
/// Get the total liquidity within a price range
/// Useful for understanding how much depth exists in a certain price band
pub fn liquidity_in_range(&self, min_price: Decimal, max_price: Decimal, side: Side) -> Decimal {
pub fn liquidity_in_range(
&self,
min_price: Decimal,
max_price: Decimal,
side: Side,
) -> Decimal {
// Convert decimal prices to our internal fixed-point representation
let min_price_ticks = match decimal_to_price(min_price) {
Ok(ticks) => ticks,
@@ -599,10 +613,14 @@ impl OrderBook {
Ok(ticks) => ticks,
Err(_) => return Decimal::ZERO, // Invalid price
};
let levels: Vec<_> = match side {
Side::BUY => self.asks.range(min_price_ticks..=max_price_ticks).collect(),
Side::SELL => self.bids.range(min_price_ticks..=max_price_ticks).rev().collect(),
Side::SELL => self
.bids
.range(min_price_ticks..=max_price_ticks)
.rev()
.collect(),
};
// Sum up the sizes, converting from fixed-point back to Decimal
@@ -615,7 +633,7 @@ impl OrderBook {
pub fn is_valid(&self) -> bool {
match (self.best_bid(), self.best_ask()) {
(Some(bid), Some(ask)) => bid.price < ask.price, // Normal market condition
_ => true, // Empty book is technically valid
_ => true, // Empty book is technically valid
}
}
}
@@ -624,20 +642,20 @@ impl OrderBook {
/// This tells you what would happen if you executed a large order
#[derive(Debug, Clone)]
pub struct MarketImpact {
pub average_price: Decimal, // The average price you'd get across all fills
pub impact_pct: Decimal, // How much worse than the best price (as percentage)
pub total_cost: Decimal, // Total amount you'd pay/receive
pub size_filled: Decimal, // How much of your order got filled
pub average_price: Decimal, // The average price you'd get across all fills
pub impact_pct: Decimal, // How much worse than the best price (as percentage)
pub total_cost: Decimal, // Total amount you'd pay/receive
pub size_filled: Decimal, // How much of your order got filled
}
/// Thread-safe order book manager
/// This manages multiple order books (one per token) and handles concurrent access
/// Multiple threads can read/write different books simultaneously
///
///
/// The depth limiting becomes even more critical here because we might be tracking
/// hundreds or thousands of different tokens simultaneously. If each book had
/// unlimited depth, we could easily use gigabytes of RAM for mostly useless data.
///
///
/// Example: 1000 tokens × 1000 price levels × 32 bytes per level = 32MB just for prices
/// With depth limiting: 1000 tokens × 50 levels × 32 bytes = 1.6MB (20x less memory)
#[derive(Debug)]
@@ -659,9 +677,10 @@ impl OrderBookManager {
/// Get or create an order book for a token
/// If we don't have a book for this token yet, create a new empty one
pub fn get_or_create_book(&self, token_id: &str) -> Result<OrderBook> {
let mut books = self.books.write().map_err(|_| {
PolyfillError::internal_simple("Failed to acquire book lock")
})?;
let mut books = self
.books
.write()
.map_err(|_| PolyfillError::internal_simple("Failed to acquire book lock"))?;
if let Some(book) = books.get(token_id) {
Ok(book.clone()) // Return a copy of the existing book
@@ -676,19 +695,18 @@ impl OrderBookManager {
/// Update a book with a delta
/// This is called when we receive real-time updates from the exchange
pub fn apply_delta(&self, delta: OrderDelta) -> Result<()> {
let mut books = self.books.write().map_err(|_| {
PolyfillError::internal_simple("Failed to acquire book lock")
})?;
let mut books = self
.books
.write()
.map_err(|_| PolyfillError::internal_simple("Failed to acquire book lock"))?;
// Find the book for this token (must already exist)
let book = books
.get_mut(&delta.token_id)
.ok_or_else(|| {
PolyfillError::market_data(
format!("No book found for token: {}", delta.token_id),
crate::errors::MarketDataErrorKind::TokenNotFound,
)
})?;
let book = books.get_mut(&delta.token_id).ok_or_else(|| {
PolyfillError::market_data(
format!("No book found for token: {}", delta.token_id),
crate::errors::MarketDataErrorKind::TokenNotFound,
)
})?;
// Apply the update to the specific book
book.apply_delta(delta)
@@ -697,9 +715,10 @@ impl OrderBookManager {
/// Get a book snapshot
/// Returns a copy of the current book state that won't change
pub fn get_book(&self, token_id: &str) -> Result<crate::types::OrderBook> {
let books = self.books.read().map_err(|_| {
PolyfillError::internal_simple("Failed to acquire book lock")
})?;
let books = self
.books
.read()
.map_err(|_| PolyfillError::internal_simple("Failed to acquire book lock"))?;
books
.get(token_id)
@@ -715,9 +734,10 @@ impl OrderBookManager {
/// Get all available books
/// Returns snapshots of every book we're currently tracking
pub fn get_all_books(&self) -> Result<Vec<crate::types::OrderBook>> {
let books = self.books.read().map_err(|_| {
PolyfillError::internal_simple("Failed to acquire book lock")
})?;
let books = self
.books
.read()
.map_err(|_| PolyfillError::internal_simple("Failed to acquire book lock"))?;
Ok(books.values().map(|book| book.snapshot()).collect())
}
@@ -726,9 +746,10 @@ impl OrderBookManager {
/// Cleans up books that haven't been updated recently (probably disconnected)
/// This prevents memory leaks from accumulating dead books
pub fn cleanup_stale_books(&self, max_age: std::time::Duration) -> Result<usize> {
let mut books = self.books.write().map_err(|_| {
PolyfillError::internal_simple("Failed to acquire book lock")
})?;
let mut books = self
.books
.write()
.map_err(|_| PolyfillError::internal_simple("Failed to acquire book lock"))?;
let initial_count = books.len();
books.retain(|_, book| !book.is_stale(max_age)); // Keep only non-stale books
@@ -748,13 +769,13 @@ impl OrderBookManager {
pub struct BookAnalytics {
pub token_id: String,
pub timestamp: chrono::DateTime<Utc>,
pub bid_count: usize, // How many different bid price levels
pub ask_count: usize, // How many different ask price levels
pub total_bid_size: Decimal, // Total size of all bids combined
pub total_ask_size: Decimal, // Total size of all asks combined
pub spread: Option<Decimal>, // Current spread (ask - bid)
pub bid_count: usize, // How many different bid price levels
pub ask_count: usize, // How many different ask price levels
pub total_bid_size: Decimal, // Total size of all bids combined
pub total_ask_size: Decimal, // Total size of all asks combined
pub spread: Option<Decimal>, // Current spread (ask - bid)
pub spread_pct: Option<Decimal>, // Spread as percentage
pub mid_price: Option<Decimal>, // Current mid price
pub mid_price: Option<Decimal>, // Current mid price
pub volatility: Option<Decimal>, // Price volatility (if calculated)
}
@@ -814,7 +835,7 @@ mod tests {
fn test_apply_delta() {
// Test that we can apply order book updates
let mut book = OrderBook::new("test_token".to_string(), 10);
// Create a buy order at $0.50 for 100 tokens
let delta = OrderDelta {
token_id: "test_token".to_string(),
@@ -835,7 +856,7 @@ mod tests {
fn test_spread_calculation() {
// Test that we can calculate the spread between bid and ask
let mut book = OrderBook::new("test_token".to_string(), 10);
// Add a bid at $0.50
book.apply_delta(OrderDelta {
token_id: "test_token".to_string(),
@@ -844,7 +865,8 @@ mod tests {
price: dec!(0.5),
size: dec!(100),
sequence: 1,
}).unwrap();
})
.unwrap();
// Add an ask at $0.52
book.apply_delta(OrderDelta {
@@ -854,7 +876,8 @@ mod tests {
price: dec!(0.52),
size: dec!(100),
sequence: 2,
}).unwrap();
})
.unwrap();
let spread = book.spread().unwrap();
assert_eq!(spread, dec!(0.02)); // $0.52 - $0.50 = $0.02
@@ -864,7 +887,7 @@ mod tests {
fn test_market_impact() {
// Test market impact calculation for a large order
let mut book = OrderBook::new("test_token".to_string(), 10);
// Add multiple ask levels (people selling at different prices)
// $0.50 for 100 tokens, $0.51 for 100 tokens, $0.52 for 100 tokens
for (i, price) in [dec!(0.50), dec!(0.51), dec!(0.52)].iter().enumerate() {
@@ -875,7 +898,8 @@ mod tests {
price: *price,
size: dec!(100),
sequence: i as u64 + 1,
}).unwrap();
})
.unwrap();
}
// Try to buy 150 tokens (will need to hit multiple price levels)
@@ -887,21 +911,27 @@ mod tests {
#[test]
fn test_apply_bid_delta_legacy() {
let mut book = OrderBook::new("test_token".to_string(), 10);
// Test adding a bid
book.apply_bid_delta(Decimal::from_str("0.75").unwrap(), Decimal::from_str("100.0").unwrap());
book.apply_bid_delta(
Decimal::from_str("0.75").unwrap(),
Decimal::from_str("100.0").unwrap(),
);
let best_bid = book.best_bid();
assert!(best_bid.is_some());
let bid = best_bid.unwrap();
assert_eq!(bid.price, Decimal::from_str("0.75").unwrap());
assert_eq!(bid.size, Decimal::from_str("100.0").unwrap());
// Test updating the bid
book.apply_bid_delta(Decimal::from_str("0.75").unwrap(), Decimal::from_str("150.0").unwrap());
book.apply_bid_delta(
Decimal::from_str("0.75").unwrap(),
Decimal::from_str("150.0").unwrap(),
);
let updated_bid = book.best_bid().unwrap();
assert_eq!(updated_bid.size, Decimal::from_str("150.0").unwrap());
// Test removing the bid
book.apply_bid_delta(Decimal::from_str("0.75").unwrap(), Decimal::ZERO);
assert!(book.best_bid().is_none());
@@ -910,21 +940,27 @@ mod tests {
#[test]
fn test_apply_ask_delta_legacy() {
let mut book = OrderBook::new("test_token".to_string(), 10);
// Test adding an ask
book.apply_ask_delta(Decimal::from_str("0.76").unwrap(), Decimal::from_str("50.0").unwrap());
book.apply_ask_delta(
Decimal::from_str("0.76").unwrap(),
Decimal::from_str("50.0").unwrap(),
);
let best_ask = book.best_ask();
assert!(best_ask.is_some());
let ask = best_ask.unwrap();
assert_eq!(ask.price, Decimal::from_str("0.76").unwrap());
assert_eq!(ask.size, Decimal::from_str("50.0").unwrap());
// Test updating the ask
book.apply_ask_delta(Decimal::from_str("0.76").unwrap(), Decimal::from_str("75.0").unwrap());
book.apply_ask_delta(
Decimal::from_str("0.76").unwrap(),
Decimal::from_str("75.0").unwrap(),
);
let updated_ask = book.best_ask().unwrap();
assert_eq!(updated_ask.size, Decimal::from_str("75.0").unwrap());
// Test removing the ask
book.apply_ask_delta(Decimal::from_str("0.76").unwrap(), Decimal::ZERO);
assert!(book.best_ask().is_none());
@@ -933,35 +969,48 @@ mod tests {
#[test]
fn test_liquidity_analysis() {
let mut book = OrderBook::new("test_token".to_string(), 10);
// Build order book using legacy methods
book.apply_bid_delta(Decimal::from_str("0.75").unwrap(), Decimal::from_str("100.0").unwrap());
book.apply_bid_delta(Decimal::from_str("0.74").unwrap(), Decimal::from_str("50.0").unwrap());
book.apply_ask_delta(Decimal::from_str("0.76").unwrap(), Decimal::from_str("80.0").unwrap());
book.apply_ask_delta(Decimal::from_str("0.77").unwrap(), Decimal::from_str("120.0").unwrap());
book.apply_bid_delta(
Decimal::from_str("0.75").unwrap(),
Decimal::from_str("100.0").unwrap(),
);
book.apply_bid_delta(
Decimal::from_str("0.74").unwrap(),
Decimal::from_str("50.0").unwrap(),
);
book.apply_ask_delta(
Decimal::from_str("0.76").unwrap(),
Decimal::from_str("80.0").unwrap(),
);
book.apply_ask_delta(
Decimal::from_str("0.77").unwrap(),
Decimal::from_str("120.0").unwrap(),
);
// Test liquidity at specific price - when buying, we look at ask liquidity
let buy_liquidity = book.liquidity_at_price(Decimal::from_str("0.76").unwrap(), Side::BUY);
assert_eq!(buy_liquidity, Decimal::from_str("80.0").unwrap());
// Test liquidity at specific price - when selling, we look at bid liquidity
let sell_liquidity = book.liquidity_at_price(Decimal::from_str("0.75").unwrap(), Side::SELL);
// Test liquidity at specific price - when selling, we look at bid liquidity
let sell_liquidity =
book.liquidity_at_price(Decimal::from_str("0.75").unwrap(), Side::SELL);
assert_eq!(sell_liquidity, Decimal::from_str("100.0").unwrap());
// Test liquidity in range - when buying, we look at ask liquidity in range
let buy_range_liquidity = book.liquidity_in_range(
Decimal::from_str("0.74").unwrap(),
Decimal::from_str("0.77").unwrap(),
Side::BUY
Side::BUY,
);
// Should include ask liquidity: 80 (0.76 ask) + 120 (0.77 ask) = 200
assert_eq!(buy_range_liquidity, Decimal::from_str("200.0").unwrap());
// Test liquidity in range - when selling, we look at bid liquidity in range
let sell_range_liquidity = book.liquidity_in_range(
Decimal::from_str("0.74").unwrap(),
Decimal::from_str("0.77").unwrap(),
Side::SELL
Side::SELL,
);
// Should include bid liquidity: 50 (0.74 bid) + 100 (0.75 bid) = 150
assert_eq!(sell_range_liquidity, Decimal::from_str("150.0").unwrap());
@@ -970,31 +1019,43 @@ mod tests {
#[test]
fn test_book_validation() {
let mut book = OrderBook::new("test_token".to_string(), 10);
// Empty book should be valid
assert!(book.is_valid());
// Add normal levels
book.apply_bid_delta(Decimal::from_str("0.75").unwrap(), Decimal::from_str("100.0").unwrap());
book.apply_ask_delta(Decimal::from_str("0.76").unwrap(), Decimal::from_str("80.0").unwrap());
book.apply_bid_delta(
Decimal::from_str("0.75").unwrap(),
Decimal::from_str("100.0").unwrap(),
);
book.apply_ask_delta(
Decimal::from_str("0.76").unwrap(),
Decimal::from_str("80.0").unwrap(),
);
assert!(book.is_valid());
// Create crossed book (invalid) - bid higher than ask
book.apply_bid_delta(Decimal::from_str("0.77").unwrap(), Decimal::from_str("50.0").unwrap());
book.apply_bid_delta(
Decimal::from_str("0.77").unwrap(),
Decimal::from_str("50.0").unwrap(),
);
assert!(!book.is_valid());
}
#[test]
fn test_book_staleness() {
let mut book = OrderBook::new("test_token".to_string(), 10);
// Fresh book should not be stale
assert!(!book.is_stale(Duration::from_secs(60))); // 60 second threshold
// Add some data
book.apply_bid_delta(Decimal::from_str("0.75").unwrap(), Decimal::from_str("100.0").unwrap());
book.apply_bid_delta(
Decimal::from_str("0.75").unwrap(),
Decimal::from_str("100.0").unwrap(),
);
assert!(!book.is_stale(Duration::from_secs(60)));
// Note: We can't easily test actual staleness without manipulating time,
// but we can test the method exists and works with fresh data
}
@@ -1002,47 +1063,77 @@ mod tests {
#[test]
fn test_depth_management() {
let mut book = OrderBook::new("test_token".to_string(), 3); // Only 3 levels
// Add multiple levels
book.apply_bid_delta(Decimal::from_str("0.75").unwrap(), Decimal::from_str("100.0").unwrap());
book.apply_bid_delta(Decimal::from_str("0.74").unwrap(), Decimal::from_str("50.0").unwrap());
book.apply_bid_delta(Decimal::from_str("0.73").unwrap(), Decimal::from_str("20.0").unwrap());
book.apply_ask_delta(Decimal::from_str("0.76").unwrap(), Decimal::from_str("80.0").unwrap());
book.apply_ask_delta(Decimal::from_str("0.77").unwrap(), Decimal::from_str("40.0").unwrap());
book.apply_ask_delta(Decimal::from_str("0.78").unwrap(), Decimal::from_str("30.0").unwrap());
book.apply_bid_delta(
Decimal::from_str("0.75").unwrap(),
Decimal::from_str("100.0").unwrap(),
);
book.apply_bid_delta(
Decimal::from_str("0.74").unwrap(),
Decimal::from_str("50.0").unwrap(),
);
book.apply_bid_delta(
Decimal::from_str("0.73").unwrap(),
Decimal::from_str("20.0").unwrap(),
);
book.apply_ask_delta(
Decimal::from_str("0.76").unwrap(),
Decimal::from_str("80.0").unwrap(),
);
book.apply_ask_delta(
Decimal::from_str("0.77").unwrap(),
Decimal::from_str("40.0").unwrap(),
);
book.apply_ask_delta(
Decimal::from_str("0.78").unwrap(),
Decimal::from_str("30.0").unwrap(),
);
// Should have levels on each side
let bids = book.bids(Some(3));
let asks = book.asks(Some(3));
assert!(bids.len() <= 3);
assert!(asks.len() <= 3);
// Best levels should be there
assert_eq!(book.best_bid().unwrap().price, Decimal::from_str("0.75").unwrap());
assert_eq!(book.best_ask().unwrap().price, Decimal::from_str("0.76").unwrap());
assert_eq!(
book.best_bid().unwrap().price,
Decimal::from_str("0.75").unwrap()
);
assert_eq!(
book.best_ask().unwrap().price,
Decimal::from_str("0.76").unwrap()
);
}
#[test]
fn test_fast_operations() {
let mut book = OrderBook::new("test_token".to_string(), 10);
// Test using legacy methods which call fast operations internally
book.apply_bid_delta(Decimal::from_str("0.75").unwrap(), Decimal::from_str("100.0").unwrap());
book.apply_ask_delta(Decimal::from_str("0.76").unwrap(), Decimal::from_str("80.0").unwrap());
book.apply_bid_delta(
Decimal::from_str("0.75").unwrap(),
Decimal::from_str("100.0").unwrap(),
);
book.apply_ask_delta(
Decimal::from_str("0.76").unwrap(),
Decimal::from_str("80.0").unwrap(),
);
let best_bid_fast = book.best_bid_fast();
let best_ask_fast = book.best_ask_fast();
assert!(best_bid_fast.is_some());
assert!(best_ask_fast.is_some());
// Test fast spread and mid price
let spread_fast = book.spread_fast();
let mid_fast = book.mid_price_fast();
assert!(spread_fast.is_some()); // Should have a spread
assert!(mid_fast.is_some()); // Should have a mid price
assert!(mid_fast.is_some()); // Should have a mid price
}
}
}
src/client.rs
+634 -257
View File
File diff suppressed because it is too large Load Diff
src/decode.rs
+41 -32
View File
@@ -31,15 +31,15 @@ pub mod deserializers {
T::deserialize(serde_json::Value::Number(serde_json::Number::from(v)))
.map_err(|_| serde::de::Error::custom("Failed to deserialize number"))
} else if let Some(v) = n.as_f64() {
T::deserialize(serde_json::Value::Number(serde_json::Number::from_f64(v).unwrap()))
.map_err(|_| serde::de::Error::custom("Failed to deserialize number"))
T::deserialize(serde_json::Value::Number(
serde_json::Number::from_f64(v).unwrap(),
))
.map_err(|_| serde::de::Error::custom("Failed to deserialize number"))
} else {
Err(serde::de::Error::custom("Invalid number format"))
}
}
serde_json::Value::String(s) => {
s.parse::<T>().map_err(serde::de::Error::custom)
}
},
serde_json::Value::String(s) => s.parse::<T>().map_err(serde::de::Error::custom),
_ => Err(serde::de::Error::custom("Expected number or string")),
}
}
@@ -62,28 +62,30 @@ pub mod deserializers {
.map(Some)
.map_err(|_| serde::de::Error::custom("Failed to deserialize number"))
} else if let Some(v) = n.as_f64() {
T::deserialize(serde_json::Value::Number(serde_json::Number::from_f64(v).unwrap()))
.map(Some)
.map_err(|_| serde::de::Error::custom("Failed to deserialize number"))
T::deserialize(serde_json::Value::Number(
serde_json::Number::from_f64(v).unwrap(),
))
.map(Some)
.map_err(|_| serde::de::Error::custom("Failed to deserialize number"))
} else {
Err(serde::de::Error::custom("Invalid number format"))
}
}
},
serde_json::Value::String(s) => {
if s.is_empty() {
Ok(None)
} else {
s.parse::<T>()
.map(Some)
.map_err(serde::de::Error::custom)
s.parse::<T>().map(Some).map_err(serde::de::Error::custom)
}
}
},
_ => Err(serde::de::Error::custom("Expected number, string, or null")),
}
}
/// Deserialize DateTime from Unix timestamp
pub fn datetime_from_timestamp<'de, D>(deserializer: D) -> std::result::Result<DateTime<Utc>, D::Error>
pub fn datetime_from_timestamp<'de, D>(
deserializer: D,
) -> std::result::Result<DateTime<Utc>, D::Error>
where
D: Deserializer<'de>,
{
@@ -236,18 +238,25 @@ impl Decoder<Order> for RawOrderResponse {
"FILLED" => OrderStatus::Filled,
"PARTIAL" => OrderStatus::Partial,
"EXPIRED" => OrderStatus::Expired,
_ => return Err(PolyfillError::parse(
format!("Unknown order status: {}", self.status),
None,
)),
_ => {
return Err(PolyfillError::parse(
format!("Unknown order status: {}", self.status),
None,
))
},
};
let created_at = chrono::DateTime::from_timestamp(self.created_at as i64, 0)
.ok_or_else(|| PolyfillError::parse("Invalid created_at timestamp".to_string(), None))?;
let created_at =
chrono::DateTime::from_timestamp(self.created_at as i64, 0).ok_or_else(|| {
PolyfillError::parse("Invalid created_at timestamp".to_string(), None)
})?;
let expiration = if self.expiration > 0 {
Some(chrono::DateTime::from_timestamp(self.expiration as i64, 0)
.ok_or_else(|| PolyfillError::parse("Invalid expiration timestamp".to_string(), None))?)
Some(
chrono::DateTime::from_timestamp(self.expiration as i64, 0).ok_or_else(|| {
PolyfillError::parse("Invalid expiration timestamp".to_string(), None)
})?,
)
} else {
None
};
@@ -344,7 +353,7 @@ impl Decoder<Market> for RawMarketResponse {
/// WebSocket message parsing
pub fn parse_stream_message(raw: &str) -> Result<StreamMessage> {
let value: Value = serde_json::from_str(raw)?;
let msg_type = value["type"]
.as_str()
.ok_or_else(|| PolyfillError::parse("Missing message type".to_string(), None))?;
@@ -354,19 +363,19 @@ pub fn parse_stream_message(raw: &str) -> Result<StreamMessage> {
let data = value["data"].clone();
let delta: OrderDelta = serde_json::from_value(data)?;
Ok(StreamMessage::BookUpdate { data: delta })
}
},
"trade" => {
let data = value["data"].clone();
let raw_trade: RawTradeResponse = serde_json::from_value(data)?;
let fill = raw_trade.decode()?;
Ok(StreamMessage::Trade { data: fill })
}
},
"order_update" => {
let data = value["data"].clone();
let raw_order: RawOrderResponse = serde_json::from_value(data)?;
let order = raw_order.decode()?;
Ok(StreamMessage::OrderUpdate { data: order })
}
},
"heartbeat" => {
let timestamp = value["timestamp"]
.as_str()
@@ -374,7 +383,7 @@ pub fn parse_stream_message(raw: &str) -> Result<StreamMessage> {
.map(|dt| dt.with_timezone(&Utc))
.unwrap_or_else(Utc::now);
Ok(StreamMessage::Heartbeat { timestamp })
}
},
_ => Err(PolyfillError::parse(
format!("Unknown message type: {}", msg_type),
None,
@@ -440,8 +449,8 @@ impl BatchDecoder {
if depth == 0 {
return Some(start + i + 1);
}
}
_ => {}
},
_ => {},
}
}
@@ -512,8 +521,8 @@ mod tests {
fn test_batch_decoder() {
let mut decoder = BatchDecoder::new();
let data = r#"{"test":1}{"test":2}"#.as_bytes();
let results: Vec<serde_json::Value> = decoder.parse_json_stream(data).unwrap();
assert_eq!(results.len(), 2);
}
}
}
src/errors.rs
+81 -91
View File
@@ -4,15 +4,15 @@
//! for clear error handling in trading environments where fast error recovery
//! is critical.
use thiserror::Error;
use std::time::Duration;
use thiserror::Error;
/// Main error type for the Polymarket client
#[derive(Error, Debug)]
pub enum PolyfillError {
/// Network-related errors (retryable)
#[error("Network error: {message}")]
Network {
Network {
message: String,
#[source]
source: Option<Box<dyn std::error::Error + Send + Sync>>,
@@ -20,7 +20,7 @@ pub enum PolyfillError {
/// API errors from Polymarket
#[error("API error ({status}): {message}")]
Api {
Api {
status: u16,
message: String,
error_code: Option<String>,
@@ -28,34 +28,32 @@ pub enum PolyfillError {
/// Authentication/authorization errors
#[error("Auth error: {message}")]
Auth {
Auth {
message: String,
kind: AuthErrorKind,
},
/// Order-related errors
#[error("Order error: {message}")]
Order {
Order {
message: String,
kind: OrderErrorKind,
},
/// Market data errors
#[error("Market data error: {message}")]
MarketData {
MarketData {
message: String,
kind: MarketDataErrorKind,
},
/// Configuration errors
#[error("Config error: {message}")]
Config {
message: String,
},
Config { message: String },
/// Parsing/serialization errors
#[error("Parse error: {message}")]
Parse {
Parse {
message: String,
#[source]
source: Option<Box<dyn std::error::Error + Send + Sync>>,
@@ -63,35 +61,35 @@ pub enum PolyfillError {
/// Timeout errors
#[error("Timeout error: operation timed out after {duration:?}")]
Timeout {
Timeout {
duration: Duration,
operation: String,
},
/// Rate limiting errors
#[error("Rate limit exceeded: {message}")]
RateLimit {
RateLimit {
message: String,
retry_after: Option<Duration>,
},
/// WebSocket/streaming errors
#[error("Stream error: {message}")]
Stream {
Stream {
message: String,
kind: StreamErrorKind,
},
/// Validation errors
#[error("Validation error: {message}")]
Validation {
Validation {
message: String,
field: Option<String>,
},
/// Internal errors (bugs)
#[error("Internal error: {message}")]
Internal {
Internal {
message: String,
#[source]
source: Option<Box<dyn std::error::Error + Send + Sync>>,
@@ -155,7 +153,10 @@ impl PolyfillError {
PolyfillError::Timeout { .. } => true,
PolyfillError::RateLimit { .. } => true,
PolyfillError::Stream { kind, .. } => {
matches!(kind, StreamErrorKind::ConnectionLost | StreamErrorKind::Reconnecting)
matches!(
kind,
StreamErrorKind::ConnectionLost | StreamErrorKind::Reconnecting
)
},
_ => false,
}
@@ -267,7 +268,10 @@ impl PolyfillError {
}
}
pub fn parse(message: impl Into<String>, source: Option<Box<dyn std::error::Error + Send + Sync>>) -> Self {
pub fn parse(
message: impl Into<String>,
source: Option<Box<dyn std::error::Error + Send + Sync>>,
) -> Self {
Self::Parse {
message: message.into(),
source,
@@ -355,7 +359,7 @@ impl From<url::ParseError> for PolyfillError {
impl From<tokio_tungstenite::tungstenite::Error> for PolyfillError {
fn from(err: tokio_tungstenite::tungstenite::Error) -> Self {
use tokio_tungstenite::tungstenite::Error as WsError;
let kind = match &err {
WsError::ConnectionClosed | WsError::AlreadyClosed => StreamErrorKind::ConnectionLost,
WsError::Io(_) => StreamErrorKind::ConnectionFailed,
@@ -371,81 +375,67 @@ impl From<tokio_tungstenite::tungstenite::Error> for PolyfillError {
impl Clone for PolyfillError {
fn clone(&self) -> Self {
match self {
PolyfillError::Network { message, source: _ } => {
PolyfillError::Network {
message: message.clone(),
source: None
}
}
PolyfillError::Api { status, message, error_code } => {
PolyfillError::Api {
status: *status,
message: message.clone(),
error_code: error_code.clone()
}
}
PolyfillError::Auth { message, kind } => {
PolyfillError::Auth {
message: message.clone(),
kind: kind.clone()
}
}
PolyfillError::Order { message, kind } => {
PolyfillError::Order {
message: message.clone(),
kind: kind.clone()
}
}
PolyfillError::MarketData { message, kind } => {
PolyfillError::MarketData {
message: message.clone(),
kind: kind.clone()
}
}
PolyfillError::Config { message } => {
PolyfillError::Config {
message: message.clone()
}
}
PolyfillError::Parse { message, source: _ } => {
PolyfillError::Parse {
message: message.clone(),
source: None
}
}
PolyfillError::Timeout { duration, operation } => {
PolyfillError::Timeout {
duration: *duration,
operation: operation.clone()
}
}
PolyfillError::RateLimit { message, retry_after } => {
PolyfillError::RateLimit {
message: message.clone(),
retry_after: *retry_after
}
}
PolyfillError::Stream { message, kind } => {
PolyfillError::Stream {
message: message.clone(),
kind: kind.clone()
}
}
PolyfillError::Validation { message, field } => {
PolyfillError::Validation {
message: message.clone(),
field: field.clone()
}
}
PolyfillError::Internal { message, source: _ } => {
PolyfillError::Internal {
message: message.clone(),
source: None
}
}
PolyfillError::Network { message, source: _ } => PolyfillError::Network {
message: message.clone(),
source: None,
},
PolyfillError::Api {
status,
message,
error_code,
} => PolyfillError::Api {
status: *status,
message: message.clone(),
error_code: error_code.clone(),
},
PolyfillError::Auth { message, kind } => PolyfillError::Auth {
message: message.clone(),
kind: kind.clone(),
},
PolyfillError::Order { message, kind } => PolyfillError::Order {
message: message.clone(),
kind: kind.clone(),
},
PolyfillError::MarketData { message, kind } => PolyfillError::MarketData {
message: message.clone(),
kind: kind.clone(),
},
PolyfillError::Config { message } => PolyfillError::Config {
message: message.clone(),
},
PolyfillError::Parse { message, source: _ } => PolyfillError::Parse {
message: message.clone(),
source: None,
},
PolyfillError::Timeout {
duration,
operation,
} => PolyfillError::Timeout {
duration: *duration,
operation: operation.clone(),
},
PolyfillError::RateLimit {
message,
retry_after,
} => PolyfillError::RateLimit {
message: message.clone(),
retry_after: *retry_after,
},
PolyfillError::Stream { message, kind } => PolyfillError::Stream {
message: message.clone(),
kind: kind.clone(),
},
PolyfillError::Validation { message, field } => PolyfillError::Validation {
message: message.clone(),
field: field.clone(),
},
PolyfillError::Internal { message, source: _ } => PolyfillError::Internal {
message: message.clone(),
source: None,
},
}
}
}
/// Result type alias for convenience
pub type Result<T> = std::result::Result<T, PolyfillError>;
pub type Result<T> = std::result::Result<T, PolyfillError>;
src/fill.rs
+69 -27
View File
@@ -69,7 +69,7 @@ impl FillEngine {
book: &crate::book::OrderBook,
) -> Result<FillResult> {
let start_time = Utc::now();
// Validate order
self.validate_market_order(order)?;
@@ -81,7 +81,10 @@ impl FillEngine {
if levels.is_empty() {
return Ok(FillResult {
order_id: order.client_id.clone().unwrap_or_else(|| "market_order".to_string()),
order_id: order
.client_id
.clone()
.unwrap_or_else(|| "market_order".to_string()),
fills: Vec::new(),
total_size: Decimal::ZERO,
average_price: Decimal::ZERO,
@@ -105,13 +108,16 @@ impl FillEngine {
let fill_size = std::cmp::min(remaining_size, level.size);
let fill_cost = fill_size * level.price;
// Calculate fee
let fee = self.calculate_fee(fill_cost);
let fill = FillEvent {
id: uuid::Uuid::new_v4().to_string(),
order_id: order.client_id.clone().unwrap_or_else(|| "market_order".to_string()),
order_id: order
.client_id
.clone()
.unwrap_or_else(|| "market_order".to_string()),
token_id: order.token_id.clone(),
side: order.side,
price: level.price,
@@ -136,7 +142,10 @@ impl FillEngine {
slippage, self.max_slippage_pct
);
return Ok(FillResult {
order_id: order.client_id.clone().unwrap_or_else(|| "market_order".to_string()),
order_id: order
.client_id
.clone()
.unwrap_or_else(|| "market_order".to_string()),
fills: Vec::new(),
total_size: Decimal::ZERO,
average_price: Decimal::ZERO,
@@ -166,7 +175,10 @@ impl FillEngine {
let total_fees: Decimal = fills.iter().map(|f| f.fee).sum();
let result = FillResult {
order_id: order.client_id.clone().unwrap_or_else(|| "market_order".to_string()),
order_id: order
.client_id
.clone()
.unwrap_or_else(|| "market_order".to_string()),
fills,
total_size,
average_price,
@@ -178,7 +190,8 @@ impl FillEngine {
// Store fills for tracking
if !result.fills.is_empty() {
self.fills.insert(result.order_id.clone(), result.fills.clone());
self.fills
.insert(result.order_id.clone(), result.fills.clone());
}
info!(
@@ -211,19 +224,22 @@ impl FillEngine {
} else {
false
}
}
},
Side::SELL => {
if let Some(best_bid) = book.best_bid() {
order.price <= best_bid.price
} else {
false
}
}
},
};
if !can_fill {
return Ok(FillResult {
order_id: order.client_id.clone().unwrap_or_else(|| "limit_order".to_string()),
order_id: order
.client_id
.clone()
.unwrap_or_else(|| "limit_order".to_string()),
fills: Vec::new(),
total_size: Decimal::ZERO,
average_price: Decimal::ZERO,
@@ -237,7 +253,10 @@ impl FillEngine {
// Simulate immediate fill
let fill = FillEvent {
id: uuid::Uuid::new_v4().to_string(),
order_id: order.client_id.clone().unwrap_or_else(|| "limit_order".to_string()),
order_id: order
.client_id
.clone()
.unwrap_or_else(|| "limit_order".to_string()),
token_id: order.token_id.clone(),
side: order.side,
price: order.price,
@@ -249,7 +268,10 @@ impl FillEngine {
};
let result = FillResult {
order_id: order.client_id.clone().unwrap_or_else(|| "limit_order".to_string()),
order_id: order
.client_id
.clone()
.unwrap_or_else(|| "limit_order".to_string()),
fills: vec![fill],
total_size: order.size,
average_price: order.price,
@@ -260,7 +282,8 @@ impl FillEngine {
};
// Store fills for tracking
self.fills.insert(result.order_id.clone(), result.fills.clone());
self.fills
.insert(result.order_id.clone(), result.fills.clone());
info!(
"Limit order executed: {} {} @ {}",
@@ -273,7 +296,11 @@ impl FillEngine {
}
/// Calculate slippage for a market order
fn calculate_slippage(&self, order: &MarketOrderRequest, fills: &[FillEvent]) -> Option<Decimal> {
fn calculate_slippage(
&self,
order: &MarketOrderRequest,
fills: &[FillEvent],
) -> Option<Decimal> {
if fills.is_empty() {
return None;
}
@@ -284,11 +311,15 @@ impl FillEngine {
// Get reference price (best bid/ask)
let reference_price = match order.side {
Side::BUY => fills.first()?.price, // Best ask
Side::BUY => fills.first()?.price, // Best ask
Side::SELL => fills.first()?.price, // Best bid
};
Some(math::calculate_slippage(reference_price, average_price, order.side))
Some(math::calculate_slippage(
reference_price,
average_price,
order.side,
))
}
/// Calculate fee for a trade
@@ -307,7 +338,10 @@ impl FillEngine {
if order.amount < self.min_fill_size {
return Err(PolyfillError::order(
format!("Order size {} below minimum {}", order.amount, self.min_fill_size),
format!(
"Order size {} below minimum {}",
order.amount, self.min_fill_size
),
crate::errors::OrderErrorKind::SizeConstraint,
));
}
@@ -333,7 +367,10 @@ impl FillEngine {
if order.size < self.min_fill_size {
return Err(PolyfillError::order(
format!("Order size {} below minimum {}", order.size, self.min_fill_size),
format!(
"Order size {} below minimum {}",
order.size, self.min_fill_size
),
crate::errors::OrderErrorKind::SizeConstraint,
));
}
@@ -424,7 +461,12 @@ impl FillProcessor {
self.cleanup_old_pending();
}
debug!("Processed fill: {} {} @ {}", fill.size, fill.side.as_str(), fill.price);
debug!(
"Processed fill: {} {} @ {}",
fill.size,
fill.side.as_str(),
fill.price
);
Ok(())
}
@@ -517,7 +559,7 @@ mod tests {
#[test]
fn test_market_order_validation() {
let engine = FillEngine::new(dec!(1), dec!(5), 10);
let valid_order = MarketOrderRequest {
token_id: "test".to_string(),
side: Side::BUY,
@@ -547,7 +589,7 @@ mod tests {
#[test]
fn test_fill_processor() {
let mut processor = FillProcessor::new(100);
let fill = FillEvent {
id: "fill1".to_string(),
order_id: "order1".to_string(),
@@ -569,7 +611,7 @@ mod tests {
fn test_fill_engine_advanced_creation() {
// Test that we can create a fill engine with parameters
let _engine = FillEngine::new(dec!(1.0), dec!(0.05), 50); // min_fill_size, max_slippage, fee_rate_bps
// Test basic properties exist (we can't access private fields directly)
// But we can test that the engine was created successfully
// Engine creation successful
@@ -578,7 +620,7 @@ mod tests {
#[test]
fn test_fill_processor_basic_operations() {
let mut processor = FillProcessor::new(100); // max_pending
// Test that we can create a fill event and process it
let fill_event = FillEvent {
id: "fill_1".to_string(),
@@ -592,11 +634,11 @@ mod tests {
taker_address: alloy_primitives::Address::ZERO,
fee: dec!(0.01),
};
let result = processor.process_fill(fill_event);
assert!(result.is_ok());
// Check that the fill was added to pending
assert_eq!(processor.pending_fills.len(), 1);
}
}
}
src/http_config.rs
+18 -36
View File
@@ -1,5 +1,5 @@
//! HTTP client optimization for low-latency trading
//!
//!
//! This module provides optimized HTTP client configurations specifically
//! designed for high-frequency trading environments where every millisecond counts.
@@ -10,7 +10,7 @@ use std::time::Duration;
pub async fn prewarm_connections(client: &Client, base_url: &str) -> Result<(), reqwest::Error> {
// Make a few lightweight requests to establish connections
let endpoints = vec!["/ok", "/time"];
for endpoint in endpoints {
let _ = client
.get(format!("{}{}", base_url, endpoint))
@@ -18,7 +18,7 @@ pub async fn prewarm_connections(client: &Client, base_url: &str) -> Result<(),
.send()
.await;
}
Ok(())
}
@@ -26,30 +26,24 @@ pub async fn prewarm_connections(client: &Client, base_url: &str) -> Result<(),
pub fn create_optimized_client() -> Result<Client, reqwest::Error> {
ClientBuilder::new()
// Connection pooling optimizations
.pool_max_idle_per_host(10) // Keep connections alive
.pool_idle_timeout(Duration::from_secs(30)) // Reuse connections
.pool_max_idle_per_host(10) // Keep connections alive
.pool_idle_timeout(Duration::from_secs(30)) // Reuse connections
// Timeout optimizations - aggressive but safe
.connect_timeout(Duration::from_millis(5000)) // 5s connection timeout
.timeout(Duration::from_millis(30000)) // 30s total timeout
.connect_timeout(Duration::from_millis(5000)) // 5s connection timeout
.timeout(Duration::from_millis(30000)) // 30s total timeout
// TCP optimizations
.tcp_nodelay(true) // Disable Nagle's algorithm for lower latency
.tcp_keepalive(Duration::from_secs(60)) // Keep connections alive
.tcp_nodelay(true) // Disable Nagle's algorithm for lower latency
.tcp_keepalive(Duration::from_secs(60)) // Keep connections alive
// HTTP/2 optimizations
.http2_prior_knowledge() // Use HTTP/2 if server supports it
.http2_prior_knowledge() // Use HTTP/2 if server supports it
.http2_keep_alive_interval(Duration::from_secs(30))
.http2_keep_alive_timeout(Duration::from_secs(10))
.http2_keep_alive_while_idle(true)
// Compression - balance between CPU and network
.gzip(true) // Enable gzip compression
.gzip(true) // Enable gzip compression
// Brotli is enabled by default in reqwest
// User agent for identification
.user_agent("polyfill-rs/0.1.1 (high-frequency-trading)")
.build()
}
@@ -58,29 +52,23 @@ pub fn create_optimized_client() -> Result<Client, reqwest::Error> {
pub fn create_colocated_client() -> Result<Client, reqwest::Error> {
ClientBuilder::new()
// More aggressive connection pooling
.pool_max_idle_per_host(20) // More connections
.pool_idle_timeout(Duration::from_secs(60)) // Longer reuse
.pool_max_idle_per_host(20) // More connections
.pool_idle_timeout(Duration::from_secs(60)) // Longer reuse
// Tighter timeouts for co-located environments
.connect_timeout(Duration::from_millis(1000)) // 1s connection
.timeout(Duration::from_millis(10000)) // 10s total
.connect_timeout(Duration::from_millis(1000)) // 1s connection
.timeout(Duration::from_millis(10000)) // 10s total
// TCP optimizations
.tcp_nodelay(true)
.tcp_keepalive(Duration::from_secs(30))
// HTTP/2 with more aggressive keep-alive
.http2_prior_knowledge()
.http2_keep_alive_interval(Duration::from_secs(10))
.http2_keep_alive_timeout(Duration::from_secs(5))
.http2_keep_alive_while_idle(true)
// Disable compression in co-located environments (CPU vs network tradeoff)
.gzip(false)
.no_brotli() // Disable brotli compression
.no_brotli() // Disable brotli compression
.user_agent("polyfill-rs/0.1.1 (colocated-hft)")
.build()
}
@@ -91,23 +79,17 @@ pub fn create_internet_client() -> Result<Client, reqwest::Error> {
// Conservative connection pooling
.pool_max_idle_per_host(5)
.pool_idle_timeout(Duration::from_secs(90))
// Longer timeouts for internet connections
.connect_timeout(Duration::from_millis(10000)) // 10s connection
.timeout(Duration::from_millis(60000)) // 60s total
.connect_timeout(Duration::from_millis(10000)) // 10s connection
.timeout(Duration::from_millis(60000)) // 60s total
// TCP optimizations
.tcp_nodelay(true)
.tcp_keepalive(Duration::from_secs(120))
// HTTP/1.1 might be more reliable over internet
.http1_title_case_headers()
// Enable compression (gzip and brotli are enabled by default)
.gzip(true)
.user_agent("polyfill-rs/0.1.1 (internet-trading)")
.build()
}
src/lib.rs
+68 -38
View File
@@ -1,7 +1,7 @@
//! Polyfill-rs: High-performance Rust client for Polymarket
//!
//!
//! # Features
//!
//!
//! - **High-performance order book management** with optimized data structures
//! - **Real-time market data streaming** with WebSocket support
//! - **Trade execution simulation** with slippage protection
@@ -9,14 +9,14 @@
//! - **Rate limiting and retry logic** for robust API interactions
//! - **Ethereum integration** with EIP-712 signing support
//! - **Benchmarking tools** for performance analysis
//!
//!
//! # Quick Start
//!
//!
//! ```rust,no_run
//! use polyfill_rs::{ClobClient, OrderArgs, Side};
//! use rust_decimal::Decimal;
//! use std::str::FromStr;
//!
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
//! // Create client (compatible with polymarket-rs-client)
@@ -25,11 +25,11 @@
//! "your_private_key",
//! 137,
//! );
//!
//!
//! // Get API credentials
//! let api_creds = client.create_or_derive_api_key(None).await.unwrap();
//! client.set_api_creds(api_creds);
//!
//!
//! // Create and post order
//! let order_args = OrderArgs::new(
//! "token_id",
@@ -37,40 +37,39 @@
//! Decimal::from_str("100.0").unwrap(),
//! Side::BUY,
//! );
//!
//!
//! let result = client.create_and_post_order(&order_args).await.unwrap();
//! println!("Order posted: {:?}", result);
//!
//!
//! Ok(())
//! }
//! ```
//!
//!
//! # Advanced Usage
//!
//!
//! ```rust,no_run
//! use polyfill_rs::{ClobClient, OrderBookImpl};
//! use rust_decimal::Decimal;
//!
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
//! // Create a basic client
//! let client = ClobClient::new("https://clob.polymarket.com");
//!
//!
//! // Get market data
//! let markets = client.get_sampling_markets(None).await.unwrap();
//! println!("Found {} markets", markets.data.len());
//!
//!
//! // Create an order book for high-performance operations
//! let mut book = OrderBookImpl::new("token_id".to_string(), 100); // 100 levels depth
//! println!("Order book created for token: {}", book.token_id);
//!
//!
//! Ok(())
//! }
//! ```
use tracing::info;
// Global constants
pub const DEFAULT_CHAIN_ID: u64 = 137; // Polygon
pub const DEFAULT_BASE_URL: &str = "https://clob.polymarket.com";
@@ -86,39 +85,70 @@ pub fn init() {
// Re-export main types
pub use crate::types::{
ApiCredentials, Balance, BalanceAllowance, BatchMidpointRequest, BatchMidpointResponse,
BatchPriceRequest, BatchPriceResponse, ClientConfig, FillEvent, MarketSnapshot,
NotificationParams, OpenOrder, OpenOrderParams, Order, OrderBook, OrderDelta,
OrderRequest, OrderStatus, OrderType, Side, StreamMessage, TokenPrice, TradeParams,
WssAuth, WssSubscription, WssChannelType,
ApiCredentials,
// Additional compatibility types
ApiKeysResponse, MidpointResponse, PriceResponse, SpreadResponse, TickSizeResponse,
NegRiskResponse, BookParams, MarketsResponse, SimplifiedMarketsResponse, Market,
SimplifiedMarket, Token, Rewards, ClientResult, OrderBookSummary, OrderSummary,
BalanceAllowanceParams, AssetType,
ApiKeysResponse,
AssetType,
Balance,
BalanceAllowance,
BalanceAllowanceParams,
BatchMidpointRequest,
BatchMidpointResponse,
BatchPriceRequest,
BatchPriceResponse,
BookParams,
ClientConfig,
ClientResult,
FillEvent,
Market,
MarketSnapshot,
MarketsResponse,
MidpointResponse,
NegRiskResponse,
NotificationParams,
OpenOrder,
OpenOrderParams,
Order,
OrderBook,
OrderBookSummary,
OrderDelta,
OrderRequest,
OrderStatus,
OrderSummary,
OrderType,
PriceResponse,
Rewards,
Side,
SimplifiedMarket,
SimplifiedMarketsResponse,
SpreadResponse,
StreamMessage,
TickSizeResponse,
Token,
TokenPrice,
TradeParams,
WssAuth,
WssChannelType,
WssSubscription,
};
// Re-export client
pub use crate::client::{ClobClient, PolyfillClient};
// Re-export compatibility types (for easy migration from polymarket-rs-client)
pub use crate::client::{
OrderArgs,
};
pub use crate::client::OrderArgs;
// Re-export error types
pub use crate::errors::{PolyfillError, Result};
// Re-export advanced components
pub use crate::book::{OrderBook as OrderBookImpl, OrderBookManager};
pub use crate::decode::Decoder;
pub use crate::fill::{FillEngine, FillResult};
pub use crate::stream::{MarketStream, StreamManager, WebSocketStream};
pub use crate::decode::Decoder;
// Re-export utilities
pub use crate::utils::{
crypto, math, retry, time, url, rate_limit,
};
pub use crate::utils::{crypto, math, rate_limit, retry, time, url};
// Module declarations
pub mod auth;
@@ -136,16 +166,16 @@ pub mod utils;
// Benchmarks
#[cfg(test)]
mod benches {
use criterion::{criterion_group, criterion_main};
use crate::{OrderBookManager, OrderDelta, Side};
use rust_decimal::Decimal;
use chrono::Utc;
use criterion::{criterion_group, criterion_main};
use rust_decimal::Decimal;
use std::str::FromStr;
#[allow(dead_code)]
fn order_book_benchmark(c: &mut criterion::Criterion) {
let book_manager = OrderBookManager::new(100);
c.bench_function("apply_order_delta", |b| {
b.iter(|| {
let delta = OrderDelta {
@@ -156,7 +186,7 @@ mod benches {
size: Decimal::from_str("100.0").unwrap(),
sequence: 1,
};
let _ = book_manager.apply_delta(delta);
});
});
@@ -188,7 +218,7 @@ mod tests {
Decimal::from_str("100.0").unwrap(),
Side::BUY,
);
assert_eq!(args.token_id, "test_token");
assert_eq!(args.side, Side::BUY);
}
@@ -201,4 +231,4 @@ mod tests {
assert_eq!(args.size, Decimal::ZERO);
assert_eq!(args.side, Side::BUY);
}
}
}
src/orders.rs
+32 -27
View File
@@ -4,9 +4,9 @@
//! for the Polymarket CLOB, including EIP-712 signature generation.
use crate::auth::sign_order_message;
use crate::errors::{PolyfillError, Result};
use crate::client::OrderArgs;
use crate::types::{ExtraOrderArgs, MarketOrderArgs, OrderOptions, SignedOrderRequest, Side};
use crate::errors::{PolyfillError, Result};
use crate::types::{ExtraOrderArgs, MarketOrderArgs, OrderOptions, Side, SignedOrderRequest};
use alloy_primitives::{Address, U256};
use alloy_signer_local::PrivateKeySigner;
use rand::Rng;
@@ -42,7 +42,6 @@ pub struct ContractConfig {
pub conditional_tokens: String,
}
/// Order builder for creating and signing orders
pub struct OrderBuilder {
signer: PrivateKeySigner,
@@ -177,7 +176,7 @@ impl OrderBuilder {
decimal_to_token_u32(raw_maker_amt),
decimal_to_token_u32(raw_taker_amt),
)
}
},
Side::SELL => {
let raw_maker_amt = size.round_dp_with_strategy(round_config.size, ToZero);
let raw_taker_amt = raw_maker_amt * raw_price;
@@ -187,7 +186,7 @@ impl OrderBuilder {
decimal_to_token_u32(raw_maker_amt),
decimal_to_token_u32(raw_taker_amt),
)
}
},
}
}
@@ -224,9 +223,12 @@ impl OrderBuilder {
return Ok(level.price);
}
}
Err(PolyfillError::order(
format!("Not enough liquidity to create market order with amount {}", amount_to_match),
format!(
"Not enough liquidity to create market order with amount {}",
amount_to_match
),
crate::errors::OrderErrorKind::InsufficientBalance,
))
}
@@ -240,20 +242,20 @@ impl OrderBuilder {
extras: &ExtraOrderArgs,
options: &OrderOptions,
) -> Result<SignedOrderRequest> {
let tick_size = options.tick_size
let tick_size = options
.tick_size
.ok_or_else(|| PolyfillError::validation("Cannot create order without tick size"))?;
let (maker_amount, taker_amount) = self.get_market_order_amounts(
order_args.amount,
price,
&ROUNDING_CONFIG[&tick_size],
);
let neg_risk = options.neg_risk
let (maker_amount, taker_amount) =
self.get_market_order_amounts(order_args.amount, price, &ROUNDING_CONFIG[&tick_size]);
let neg_risk = options
.neg_risk
.ok_or_else(|| PolyfillError::validation("Cannot create order without neg_risk"))?;
let contract_config = get_contract_config(chain_id, neg_risk)
.ok_or_else(|| PolyfillError::config("No contract found with given chain_id and neg_risk"))?;
let contract_config = get_contract_config(chain_id, neg_risk).ok_or_else(|| {
PolyfillError::config("No contract found with given chain_id and neg_risk")
})?;
let exchange_address = Address::from_str(&contract_config.exchange)
.map_err(|e| PolyfillError::config(format!("Invalid exchange address: {}", e)))?;
@@ -279,9 +281,10 @@ impl OrderBuilder {
extras: &ExtraOrderArgs,
options: &OrderOptions,
) -> Result<SignedOrderRequest> {
let tick_size = options.tick_size
let tick_size = options
.tick_size
.ok_or_else(|| PolyfillError::validation("Cannot create order without tick size"))?;
let (maker_amount, taker_amount) = self.get_order_amounts(
order_args.side,
order_args.size,
@@ -289,11 +292,13 @@ impl OrderBuilder {
&ROUNDING_CONFIG[&tick_size],
);
let neg_risk = options.neg_risk
let neg_risk = options
.neg_risk
.ok_or_else(|| PolyfillError::validation("Cannot create order without neg_risk"))?;
let contract_config = get_contract_config(chain_id, neg_risk)
.ok_or_else(|| PolyfillError::config("No contract found with given chain_id and neg_risk"))?;
let contract_config = get_contract_config(chain_id, neg_risk).ok_or_else(|| {
PolyfillError::config("No contract found with given chain_id and neg_risk")
})?;
let exchange_address = Address::from_str(&contract_config.exchange)
.map_err(|e| PolyfillError::config(format!("Invalid exchange address: {}", e)))?;
@@ -387,11 +392,11 @@ mod tests {
// Test zero
let result = decimal_to_token_u32(Decimal::ZERO);
assert_eq!(result, 0);
// Test small decimal
let result = decimal_to_token_u32(Decimal::from_str("0.000001").unwrap());
assert_eq!(result, 1);
// Test large number
let result = decimal_to_token_u32(Decimal::from_str("1000.0").unwrap());
assert_eq!(result, 1_000_000_000);
@@ -402,11 +407,11 @@ mod tests {
// Test Polygon mainnet
let config = get_contract_config(137, false);
assert!(config.is_some());
// Test with neg risk
let config_neg = get_contract_config(137, true);
assert!(config_neg.is_some());
// Test unsupported chain
let config_unsupported = get_contract_config(999, false);
assert!(config_unsupported.is_none());
@@ -415,7 +420,7 @@ mod tests {
#[test]
fn test_seed_generation_uniqueness() {
let mut seeds = std::collections::HashSet::new();
// Generate 1000 seeds and ensure they're all unique
for _ in 0..1000 {
let seed = generate_seed();
src/stream.rs
+172 -91
View File
@@ -5,25 +5,25 @@
use crate::errors::{PolyfillError, Result};
use crate::types::*;
use futures::{Stream, SinkExt, StreamExt};
use chrono::Utc;
use futures::{SinkExt, Stream, StreamExt};
use serde_json::Value;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::sync::mpsc;
use tracing::{debug, error, info, warn};
use chrono::Utc;
/// Trait for market data streams
pub trait MarketStream: Stream<Item = Result<StreamMessage>> + Send + Sync {
/// Subscribe to market data for specific tokens
fn subscribe(&mut self, subscription: Subscription) -> Result<()>;
/// Unsubscribe from market data
fn unsubscribe(&mut self, token_ids: &[String]) -> Result<()>;
/// Check if the stream is connected
fn is_connected(&self) -> bool;
/// Get connection statistics
fn get_stats(&self) -> StreamStats;
}
@@ -33,7 +33,11 @@ pub trait MarketStream: Stream<Item = Result<StreamMessage>> + Send + Sync {
#[allow(dead_code)]
pub struct WebSocketStream {
/// WebSocket connection
connection: Option<tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>>,
connection: Option<
tokio_tungstenite::WebSocketStream<
tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
>,
>,
/// URL for the WebSocket connection
url: String,
/// Authentication credentials
@@ -85,7 +89,7 @@ impl WebSocketStream {
/// Create a new WebSocket stream
pub fn new(url: &str) -> Self {
let (tx, rx) = mpsc::unbounded_channel();
Self {
connection: None,
url: url.to_string(),
@@ -113,8 +117,14 @@ impl WebSocketStream {
/// Connect to the WebSocket
async fn connect(&mut self) -> Result<()> {
let (ws_stream, _) = tokio_tungstenite::connect_async(&self.url).await
.map_err(|e| PolyfillError::stream(format!("WebSocket connection failed: {}", e), crate::errors::StreamErrorKind::ConnectionFailed))?;
let (ws_stream, _) = tokio_tungstenite::connect_async(&self.url)
.await
.map_err(|e| {
PolyfillError::stream(
format!("WebSocket connection failed: {}", e),
crate::errors::StreamErrorKind::ConnectionFailed,
)
})?;
self.connection = Some(ws_stream);
info!("Connected to WebSocket stream at {}", self.url);
@@ -124,16 +134,21 @@ impl WebSocketStream {
/// Send a message to the WebSocket
async fn send_message(&mut self, message: Value) -> Result<()> {
if let Some(connection) = &mut self.connection {
let text = serde_json::to_string(&message)
.map_err(|e| PolyfillError::parse(format!("Failed to serialize message: {}", e), None))?;
let text = serde_json::to_string(&message).map_err(|e| {
PolyfillError::parse(format!("Failed to serialize message: {}", e), None)
})?;
let ws_message = tokio_tungstenite::tungstenite::Message::Text(text);
connection.send(ws_message).await
.map_err(|e| PolyfillError::stream(format!("Failed to send message: {}", e), crate::errors::StreamErrorKind::MessageCorrupted))?;
connection.send(ws_message).await.map_err(|e| {
PolyfillError::stream(
format!("Failed to send message: {}", e),
crate::errors::StreamErrorKind::MessageCorrupted,
)
})?;
self.stats.messages_sent += 1;
}
Ok(())
}
@@ -154,14 +169,16 @@ impl WebSocketStream {
self.send_message(message).await?;
self.subscriptions.push(subscription.clone());
info!("Subscribed to {} channel", subscription.channel_type);
Ok(())
}
/// Subscribe to user channel (orders and trades)
pub async fn subscribe_user_channel(&mut self, markets: Vec<String>) -> Result<()> {
let auth = self.auth.as_ref()
let auth = self
.auth
.as_ref()
.ok_or_else(|| PolyfillError::auth("No authentication provided for WebSocket"))?
.clone();
@@ -177,7 +194,9 @@ impl WebSocketStream {
/// Subscribe to market channel (order book and trades)
pub async fn subscribe_market_channel(&mut self, asset_ids: Vec<String>) -> Result<()> {
let auth = self.auth.as_ref()
let auth = self
.auth
.as_ref()
.ok_or_else(|| PolyfillError::auth("No authentication provided for WebSocket"))?
.clone();
@@ -195,52 +214,54 @@ impl WebSocketStream {
pub async fn unsubscribe_async(&mut self, token_ids: &[String]) -> Result<()> {
// Note: Polymarket WebSocket API doesn't seem to have explicit unsubscribe
// We'll just remove from our local subscriptions
self.subscriptions.retain(|sub| {
match sub.channel_type.as_str() {
self.subscriptions
.retain(|sub| match sub.channel_type.as_str() {
"USER" => {
if let Some(markets) = &sub.markets {
!token_ids.iter().any(|id| markets.contains(id))
} else {
true
}
}
},
"MARKET" => {
if let Some(asset_ids) = &sub.asset_ids {
!token_ids.iter().any(|id| asset_ids.contains(id))
} else {
true
}
}
_ => true
}
});
},
_ => true,
});
info!("Unsubscribed from {} tokens", token_ids.len());
Ok(())
}
/// Handle incoming WebSocket messages
#[allow(dead_code)]
async fn handle_message(&mut self, message: tokio_tungstenite::tungstenite::Message) -> Result<()> {
async fn handle_message(
&mut self,
message: tokio_tungstenite::tungstenite::Message,
) -> Result<()> {
match message {
tokio_tungstenite::tungstenite::Message::Text(text) => {
debug!("Received WebSocket message: {}", text);
// Parse the message according to Polymarket's format
let stream_message = self.parse_polymarket_message(&text)?;
// Send to internal channel
if let Err(e) = self.tx.send(stream_message) {
error!("Failed to send message to internal channel: {}", e);
}
self.stats.messages_received += 1;
self.stats.last_message_time = Some(Utc::now());
}
},
tokio_tungstenite::tungstenite::Message::Close(_) => {
info!("WebSocket connection closed by server");
self.connection = None;
}
},
tokio_tungstenite::tungstenite::Message::Ping(data) => {
// Respond with pong
if let Some(connection) = &mut self.connection {
@@ -249,81 +270,130 @@ impl WebSocketStream {
error!("Failed to send pong: {}", e);
}
}
}
},
tokio_tungstenite::tungstenite::Message::Pong(_) => {
// Handle pong if needed
debug!("Received pong");
}
},
tokio_tungstenite::tungstenite::Message::Binary(_) => {
warn!("Received binary message (not supported)");
}
},
tokio_tungstenite::tungstenite::Message::Frame(_) => {
warn!("Received raw frame (not supported)");
}
},
}
Ok(())
}
/// Parse Polymarket WebSocket message format
#[allow(dead_code)]
fn parse_polymarket_message(&self, text: &str) -> Result<StreamMessage> {
let value: Value = serde_json::from_str(text)
.map_err(|e| PolyfillError::parse(format!("Failed to parse WebSocket message: {}", e), Some(Box::new(e))))?;
let value: Value = serde_json::from_str(text).map_err(|e| {
PolyfillError::parse(
format!("Failed to parse WebSocket message: {}", e),
Some(Box::new(e)),
)
})?;
// Extract message type
let message_type = value.get("type")
.and_then(|v| v.as_str())
.ok_or_else(|| PolyfillError::parse("Missing 'type' field in WebSocket message", None))?;
let message_type = value.get("type").and_then(|v| v.as_str()).ok_or_else(|| {
PolyfillError::parse("Missing 'type' field in WebSocket message", None)
})?;
match message_type {
"book_update" => {
let data = serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
.map_err(|e| PolyfillError::parse(format!("Failed to parse book update: {}", e), Some(Box::new(e))))?;
let data =
serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
.map_err(|e| {
PolyfillError::parse(
format!("Failed to parse book update: {}", e),
Some(Box::new(e)),
)
})?;
Ok(StreamMessage::BookUpdate { data })
}
},
"trade" => {
let data = serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
.map_err(|e| PolyfillError::parse(format!("Failed to parse trade: {}", e), Some(Box::new(e))))?;
let data =
serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
.map_err(|e| {
PolyfillError::parse(
format!("Failed to parse trade: {}", e),
Some(Box::new(e)),
)
})?;
Ok(StreamMessage::Trade { data })
}
},
"order_update" => {
let data = serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
.map_err(|e| PolyfillError::parse(format!("Failed to parse order update: {}", e), Some(Box::new(e))))?;
let data =
serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
.map_err(|e| {
PolyfillError::parse(
format!("Failed to parse order update: {}", e),
Some(Box::new(e)),
)
})?;
Ok(StreamMessage::OrderUpdate { data })
}
},
"user_order_update" => {
let data = serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
.map_err(|e| PolyfillError::parse(format!("Failed to parse user order update: {}", e), Some(Box::new(e))))?;
let data =
serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
.map_err(|e| {
PolyfillError::parse(
format!("Failed to parse user order update: {}", e),
Some(Box::new(e)),
)
})?;
Ok(StreamMessage::UserOrderUpdate { data })
}
},
"user_trade" => {
let data = serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
.map_err(|e| PolyfillError::parse(format!("Failed to parse user trade: {}", e), Some(Box::new(e))))?;
let data =
serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
.map_err(|e| {
PolyfillError::parse(
format!("Failed to parse user trade: {}", e),
Some(Box::new(e)),
)
})?;
Ok(StreamMessage::UserTrade { data })
}
},
"market_book_update" => {
let data = serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
.map_err(|e| PolyfillError::parse(format!("Failed to parse market book update: {}", e), Some(Box::new(e))))?;
let data =
serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
.map_err(|e| {
PolyfillError::parse(
format!("Failed to parse market book update: {}", e),
Some(Box::new(e)),
)
})?;
Ok(StreamMessage::MarketBookUpdate { data })
}
},
"market_trade" => {
let data = serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
.map_err(|e| PolyfillError::parse(format!("Failed to parse market trade: {}", e), Some(Box::new(e))))?;
let data =
serde_json::from_value(value.get("data").unwrap_or(&Value::Null).clone())
.map_err(|e| {
PolyfillError::parse(
format!("Failed to parse market trade: {}", e),
Some(Box::new(e)),
)
})?;
Ok(StreamMessage::MarketTrade { data })
}
},
"heartbeat" => {
let timestamp = value.get("timestamp")
let timestamp = value
.get("timestamp")
.and_then(|v| v.as_u64())
.map(|ts| chrono::DateTime::from_timestamp(ts as i64, 0).unwrap_or_default())
.unwrap_or_else(Utc::now);
Ok(StreamMessage::Heartbeat { timestamp })
}
},
_ => {
warn!("Unknown message type: {}", message_type);
// Return heartbeat as fallback
Ok(StreamMessage::Heartbeat { timestamp: Utc::now() })
}
Ok(StreamMessage::Heartbeat {
timestamp: Utc::now(),
})
},
}
}
@@ -335,38 +405,42 @@ impl WebSocketStream {
while retries < self.reconnect_config.max_retries {
warn!("Attempting to reconnect (attempt {})", retries + 1);
match self.connect().await {
Ok(()) => {
info!("Successfully reconnected");
self.stats.reconnect_count += 1;
// Resubscribe to all previous subscriptions
let subscriptions = self.subscriptions.clone();
for subscription in subscriptions {
self.send_message(serde_json::to_value(subscription)?).await?;
self.send_message(serde_json::to_value(subscription)?)
.await?;
}
return Ok(());
}
},
Err(e) => {
error!("Reconnection attempt {} failed: {}", retries + 1, e);
retries += 1;
if retries < self.reconnect_config.max_retries {
tokio::time::sleep(delay).await;
delay = std::cmp::min(
delay.mul_f64(self.reconnect_config.backoff_multiplier),
self.reconnect_config.max_delay
self.reconnect_config.max_delay,
);
}
}
},
}
}
Err(PolyfillError::stream(
format!("Failed to reconnect after {} attempts", self.reconnect_config.max_retries),
crate::errors::StreamErrorKind::ConnectionFailed
format!(
"Failed to reconnect after {} attempts",
self.reconnect_config.max_retries
),
crate::errors::StreamErrorKind::ConnectionFailed,
))
}
}
@@ -385,17 +459,19 @@ impl Stream for WebSocketStream {
match connection.poll_next_unpin(cx) {
Poll::Ready(Some(Ok(_message))) => {
// Simplified message handling
Poll::Ready(Some(Ok(StreamMessage::Heartbeat { timestamp: Utc::now() })))
}
Poll::Ready(Some(Ok(StreamMessage::Heartbeat {
timestamp: Utc::now(),
})))
},
Poll::Ready(Some(Err(e))) => {
error!("WebSocket error: {}", e);
self.stats.errors += 1;
Poll::Ready(Some(Err(e.into())))
}
},
Poll::Ready(None) => {
info!("WebSocket stream ended");
Poll::Ready(None)
}
},
Poll::Pending => Poll::Pending,
}
} else {
@@ -516,7 +592,7 @@ impl Default for StreamManager {
impl StreamManager {
pub fn new() -> Self {
let (message_tx, message_rx) = mpsc::unbounded_channel();
Self {
streams: Vec::new(),
message_tx,
@@ -537,7 +613,8 @@ impl StreamManager {
}
pub fn broadcast_message(&self, message: StreamMessage) -> Result<()> {
self.message_tx.send(message)
self.message_tx
.send(message)
.map_err(|e| PolyfillError::internal("Failed to broadcast message", e))
}
}
@@ -549,9 +626,11 @@ mod tests {
#[test]
fn test_mock_stream() {
let mut stream = MockStream::new();
// Add some test messages
stream.add_message(StreamMessage::Heartbeat { timestamp: Utc::now() });
stream.add_message(StreamMessage::Heartbeat {
timestamp: Utc::now(),
});
stream.add_message(StreamMessage::BookUpdate {
data: OrderDelta {
token_id: "test".to_string(),
@@ -560,9 +639,9 @@ mod tests {
price: rust_decimal_macros::dec!(0.5),
size: rust_decimal_macros::dec!(100),
sequence: 1,
}
},
});
assert!(stream.is_connected());
assert_eq!(stream.get_stats().messages_received, 2);
}
@@ -572,9 +651,11 @@ mod tests {
let mut manager = StreamManager::new();
let mock_stream = Box::new(MockStream::new());
manager.add_stream(mock_stream);
// Test message broadcasting
let message = StreamMessage::Heartbeat { timestamp: Utc::now() };
let message = StreamMessage::Heartbeat {
timestamp: Utc::now(),
};
assert!(manager.broadcast_message(message).is_ok());
}
}
}
src/types.rs
+65 -80
View File
@@ -5,8 +5,8 @@
use alloy_primitives::{Address, U256};
use chrono::{DateTime, Utc};
use rust_decimal::Decimal;
use rust_decimal::prelude::ToPrimitive;
use rust_decimal::Decimal;
use serde::{Deserialize, Serialize};
// ============================================================================
@@ -34,20 +34,20 @@ use serde::{Deserialize, Serialize};
/// - $0.6543 = 6543 ticks
/// - $1.0000 = 10000 ticks
/// - $0.0001 = 1 tick (minimum price increment)
///
/// Why u32?
///
/// Why u32?
/// - Can represent prices from $0.0001 to $429,496.7295 (way more than needed)
/// - Fits in CPU register for fast operations
/// - No sign bit needed since prices are always positive
pub type Price = u32;
/// Quantity/size represented as fixed-point integer for performance
///
///
/// Each unit represents 0.0001 (1/10,000) of a token
/// Examples:
/// - 100.0 tokens = 1,000,000 units
/// - 0.0001 tokens = 1 unit (minimum size increment)
///
///
/// Why i64?
/// - Can represent quantities from -922,337,203,685.4775 to +922,337,203,685.4775
/// - Signed because we need to handle both buys (+) and sells (-)
@@ -55,7 +55,7 @@ pub type Price = u32;
pub type Qty = i64;
/// Scale factor for converting between Decimal and fixed-point
///
///
/// We use 10,000 (1e4) as our scale factor, giving us 4 decimal places of precision.
/// This is perfect for most prediction markets where prices are between $0.01-$0.99
/// and we need precision to the nearest $0.0001.
@@ -80,11 +80,11 @@ pub const MAX_QTY: Qty = Qty::MAX / 2; // Leave room for intermediate calculatio
// and handle edge cases gracefully.
/// Convert a Decimal price to fixed-point ticks
///
///
/// This is called when we receive price data from the API or user input.
/// We quantize the price to the nearest tick to ensure all prices are
/// aligned to our internal representation.
///
///
/// Examples:
/// - decimal_to_price(Decimal::from_str("0.6543")) = Ok(6543)
/// - decimal_to_price(Decimal::from_str("1.0000")) = Ok(10000)
@@ -92,13 +92,13 @@ pub const MAX_QTY: Qty = Qty::MAX / 2; // Leave room for intermediate calculatio
pub fn decimal_to_price(decimal: Decimal) -> std::result::Result<Price, &'static str> {
// Convert to fixed-point by multiplying by scale factor
let scaled = decimal * Decimal::from(SCALE_FACTOR);
// Round to nearest integer (this handles tick alignment automatically)
let rounded = scaled.round();
// Convert to u64 first to handle the conversion safely
let as_u64 = rounded.to_u64().ok_or("Price too large or negative")?;
// Check bounds
if as_u64 < MIN_PRICE_TICKS as u64 {
return Ok(MIN_PRICE_TICKS); // Clamp to minimum
@@ -106,15 +106,15 @@ pub fn decimal_to_price(decimal: Decimal) -> std::result::Result<Price, &'static
if as_u64 > MAX_PRICE_TICKS as u64 {
return Err("Price exceeds maximum");
}
Ok(as_u64 as Price)
}
/// Convert fixed-point ticks back to Decimal price
///
///
/// This is called when we need to return price data to the API or display to users.
/// It's the inverse of decimal_to_price().
///
///
/// Examples:
/// - price_to_decimal(6543) = Decimal::from_str("0.6543")
/// - price_to_decimal(10000) = Decimal::from_str("1.0000")
@@ -123,28 +123,28 @@ pub fn price_to_decimal(ticks: Price) -> Decimal {
}
/// Convert a Decimal quantity to fixed-point units
///
///
/// Similar to decimal_to_price but handles signed quantities.
/// Quantities can be negative (for sells or position changes).
///
///
/// Examples:
/// - decimal_to_qty(Decimal::from_str("100.0")) = Ok(1000000)
/// - decimal_to_qty(Decimal::from_str("-50.5")) = Ok(-505000)
pub fn decimal_to_qty(decimal: Decimal) -> std::result::Result<Qty, &'static str> {
let scaled = decimal * Decimal::from(SCALE_FACTOR);
let rounded = scaled.round();
let as_i64 = rounded.to_i64().ok_or("Quantity too large")?;
if as_i64.abs() > MAX_QTY {
return Err("Quantity exceeds maximum");
}
Ok(as_i64)
}
/// Convert fixed-point units back to Decimal quantity
///
///
/// Examples:
/// - qty_to_decimal(1000000) = Decimal::from_str("100.0")
/// - qty_to_decimal(-505000) = Decimal::from_str("-50.5")
@@ -153,11 +153,11 @@ pub fn qty_to_decimal(units: Qty) -> Decimal {
}
/// Check if a price is properly tick-aligned
///
///
/// This is used to validate incoming price data. In a well-behaved system,
/// all prices should already be tick-aligned, but we check anyway to catch
/// bugs or malicious data.
///
///
/// A price is tick-aligned if it's an exact multiple of the minimum tick size.
/// Since we use integer ticks internally, this just checks if the price
/// converts cleanly to our internal representation.
@@ -167,19 +167,19 @@ pub fn is_price_tick_aligned(decimal: Decimal, tick_size_decimal: Decimal) -> bo
Ok(ticks) => ticks,
Err(_) => return false,
};
// Convert the price to ticks
let price_ticks = match decimal_to_price(decimal) {
Ok(ticks) => ticks,
Err(_) => return false,
};
// Check if price is a multiple of tick size
// If tick_size_ticks is 0, we consider everything aligned (no restrictions)
if tick_size_ticks == 0 {
return true;
}
price_ticks % tick_size_ticks == 0
}
@@ -256,7 +256,7 @@ pub struct MarketSnapshot {
}
/// Order book level (price/size pair) - EXTERNAL API VERSION
///
///
/// This is what we expose to users and serialize to JSON.
/// It uses Decimal for precision and human readability.
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -268,19 +268,19 @@ pub struct BookLevel {
}
/// Order book level (price/size pair) - INTERNAL HOT PATH VERSION
///
///
/// This is what we use internally for maximum performance.
/// All order book operations use this to avoid Decimal overhead.
///
///
/// The performance difference is huge:
/// - BookLevel: ~50ns per operation (Decimal math + allocation)
/// - FastBookLevel: ~2ns per operation (integer math, no allocation)
///
///
/// That's a 25x speedup on the critical path
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FastBookLevel {
pub price: Price, // Price in ticks (u32)
pub size: Qty, // Size in fixed-point units (i64)
pub price: Price, // Price in ticks (u32)
pub size: Qty, // Size in fixed-point units (i64)
}
impl FastBookLevel {
@@ -288,7 +288,7 @@ impl FastBookLevel {
pub fn new(price: Price, size: Qty) -> Self {
Self { price, size }
}
/// Convert to external BookLevel for API responses
/// This is only called at the edges when we need to return data to users
pub fn to_book_level(self) -> BookLevel {
@@ -297,7 +297,7 @@ impl FastBookLevel {
size: qty_to_decimal(self.size),
}
}
/// Create from external BookLevel (with validation)
/// This is called when we receive data from the API
pub fn from_book_level(level: &BookLevel) -> std::result::Result<Self, &'static str> {
@@ -305,10 +305,10 @@ impl FastBookLevel {
let size = decimal_to_qty(level.size)?;
Ok(Self::new(price, size))
}
/// Calculate notional value (price * size) in fixed-point
/// Returns the result scaled appropriately to avoid overflow
///
///
/// This is much faster than the Decimal equivalent:
/// - Decimal: price.mul(size) -> ~20ns + allocation
/// - Fixed-point: (price as i64 * size) / SCALE_FACTOR -> ~1ns, no allocation
@@ -336,7 +336,7 @@ pub struct OrderBook {
}
/// Order book delta for streaming updates - EXTERNAL API VERSION
///
///
/// This is what we receive from WebSocket streams and REST API calls.
/// It uses Decimal for compatibility with external systems.
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -350,10 +350,10 @@ pub struct OrderDelta {
}
/// Order book delta for streaming updates - INTERNAL HOT PATH VERSION
///
///
/// This is what we use internally for processing order book updates.
/// Converting to this format on ingress gives us massive performance gains.
///
///
/// Why the performance matters:
/// - We might process 10,000+ deltas per second in active markets
/// - Each delta triggers multiple calculations (spread, impact, etc.)
@@ -361,32 +361,35 @@ pub struct OrderDelta {
/// keeping up with the market feed vs falling behind
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FastOrderDelta {
pub token_id_hash: u64, // Hash of token_id for fast lookup (avoids string comparisons)
pub token_id_hash: u64, // Hash of token_id for fast lookup (avoids string comparisons)
pub timestamp: DateTime<Utc>,
pub side: Side,
pub price: Price, // Price in ticks
pub size: Qty, // Size in fixed-point units (0 means remove level)
pub price: Price, // Price in ticks
pub size: Qty, // Size in fixed-point units (0 means remove level)
pub sequence: u64,
}
impl FastOrderDelta {
/// Create from external OrderDelta with validation and tick alignment
///
///
/// This is where we enforce tick alignment - if the incoming price
/// doesn't align to valid ticks, we either reject it or round it.
/// This prevents bad data from corrupting our order book.
pub fn from_order_delta(delta: &OrderDelta, tick_size: Option<Decimal>) -> std::result::Result<Self, &'static str> {
pub fn from_order_delta(
delta: &OrderDelta,
tick_size: Option<Decimal>,
) -> std::result::Result<Self, &'static str> {
// Validate tick alignment if we have a tick size
if let Some(tick_size) = tick_size {
if !is_price_tick_aligned(delta.price, tick_size) {
return Err("Price not aligned to tick size");
}
}
// Convert to fixed-point with validation
let price = decimal_to_price(delta.price)?;
let size = decimal_to_qty(delta.size)?;
// Hash the token_id for fast lookups
// This avoids string comparisons in the hot path
let token_id_hash = {
@@ -396,7 +399,7 @@ impl FastOrderDelta {
delta.token_id.hash(&mut hasher);
hasher.finish()
};
Ok(Self {
token_id_hash,
timestamp: delta.timestamp,
@@ -406,7 +409,7 @@ impl FastOrderDelta {
sequence: delta.sequence,
})
}
/// Convert back to external OrderDelta (for API responses)
/// We need the original token_id since we only store the hash
pub fn to_order_delta(self, token_id: String) -> OrderDelta {
@@ -419,7 +422,7 @@ impl FastOrderDelta {
sequence: self.sequence,
}
}
/// Check if this delta removes a level (size is zero)
pub fn is_removal(self) -> bool {
self.size == 0
@@ -663,39 +666,23 @@ pub struct WssSubscription {
#[serde(tag = "type")]
pub enum StreamMessage {
#[serde(rename = "book_update")]
BookUpdate {
data: OrderDelta,
},
BookUpdate { data: OrderDelta },
#[serde(rename = "trade")]
Trade {
data: FillEvent,
},
Trade { data: FillEvent },
#[serde(rename = "order_update")]
OrderUpdate {
data: Order,
},
OrderUpdate { data: Order },
#[serde(rename = "heartbeat")]
Heartbeat {
timestamp: DateTime<Utc>,
},
Heartbeat { timestamp: DateTime<Utc> },
/// User channel events
#[serde(rename = "user_order_update")]
UserOrderUpdate {
data: Order,
},
UserOrderUpdate { data: Order },
#[serde(rename = "user_trade")]
UserTrade {
data: FillEvent,
},
UserTrade { data: FillEvent },
/// Market channel events
#[serde(rename = "market_book_update")]
MarketBookUpdate {
data: OrderDelta,
},
MarketBookUpdate { data: OrderDelta },
#[serde(rename = "market_trade")]
MarketTrade {
data: FillEvent,
},
MarketTrade { data: FillEvent },
}
/// Subscription parameters for streaming
@@ -757,7 +744,6 @@ pub type OrderId = String;
pub type MarketId = String;
pub type ClientId = String;
/// Parameters for querying open orders
#[derive(Debug, Clone)]
pub struct OpenOrderParams {
@@ -811,19 +797,19 @@ impl TradeParams {
if let Some(x) = &self.market {
params.push(("market", x.clone()));
}
if let Some(x) = &self.maker_address {
params.push(("maker_address", x.clone()));
}
if let Some(x) = &self.before {
params.push(("before", x.to_string()));
}
if let Some(x) = &self.after {
params.push(("after", x.to_string()));
}
params
}
}
@@ -854,7 +840,6 @@ pub struct OpenOrder {
pub created_at: u64,
}
/// Balance allowance information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BalanceAllowance {
@@ -1068,9 +1053,9 @@ pub struct Rewards {
pub type ClientResult<T> = anyhow::Result<T>;
/// Result type used throughout the client
pub type Result<T> = std::result::Result<T, crate::errors::PolyfillError>;
pub type Result<T> = std::result::Result<T, crate::errors::PolyfillError>;
// Type aliases for 100% compatibility with baseline implementation
pub type ApiCreds = ApiCredentials;
pub type CreateOrderOptions = OrderOptions;
pub type OrderArgs = OrderRequest;
pub type OrderArgs = OrderRequest;
src/utils.rs
+45 -45
View File
@@ -4,6 +4,7 @@
//! operations in trading environments.
use crate::errors::{PolyfillError, Result};
use ::url::Url;
use alloy_primitives::{Address, U256};
use base64::{engine::general_purpose::URL_SAFE, Engine};
use chrono::{DateTime, Utc};
@@ -13,7 +14,6 @@ use serde::Serialize;
use sha2::Sha256;
use std::str::FromStr;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use ::url::Url;
type HmacSha256 = Hmac<Sha256>;
@@ -66,8 +66,7 @@ pub mod time {
/// Convert Unix timestamp to DateTime
#[inline]
pub fn secs_to_datetime(timestamp: u64) -> DateTime<Utc> {
DateTime::from_timestamp(timestamp as i64, 0)
.unwrap_or_else(Utc::now)
DateTime::from_timestamp(timestamp as i64, 0).unwrap_or_else(Utc::now)
}
}
@@ -95,12 +94,12 @@ pub mod crypto {
Some(data) => {
let json = serde_json::to_string(data)?;
format!("{timestamp}{method}{path}{json}")
}
},
};
let mut mac = HmacSha256::new_from_slice(&decoded)
.map_err(|e| PolyfillError::internal("HMAC initialization failed", e))?;
mac.update(message.as_bytes());
let result = mac.finalize();
@@ -127,13 +126,13 @@ pub mod crypto {
/// Price and size calculation utilities
pub mod math {
use super::*;
use rust_decimal::prelude::*;
use crate::types::{Price, Qty, SCALE_FACTOR};
use rust_decimal::prelude::*;
// ========================================================================
// LEGACY DECIMAL FUNCTIONS (for backward compatibility)
// ========================================================================
//
//
// These are kept for API compatibility, but internally we should use
// the fixed-point versions below for better performance.
@@ -185,10 +184,10 @@ pub mod math {
// That's a 10-50x speedup on the critical path!
/// Round price to tick size (FAST VERSION)
///
///
/// This is much faster than the Decimal version because it's just
/// integer division and multiplication.
///
///
/// Example: round_to_tick_fast(6543, 10) = 6540 (rounds to nearest 10 ticks)
#[inline]
pub fn round_to_tick_fast(price_ticks: Price, tick_size_ticks: Price) -> Price {
@@ -202,10 +201,10 @@ pub mod math {
}
/// Calculate notional value (price * size) (FAST VERSION)
///
///
/// Returns the result in the same scale as our quantities.
/// This avoids the expensive Decimal multiplication.
///
///
/// Example: notional_fast(6543, 1000000) = 6543000000 (representing $654.30)
#[inline]
pub fn notional_fast(price_ticks: Price, size_units: Qty) -> i64 {
@@ -218,47 +217,47 @@ pub mod math {
}
/// Calculate spread as percentage (FAST VERSION)
///
///
/// Returns the spread as a percentage in basis points (1/100th of a percent).
/// This avoids floating-point arithmetic entirely.
///
///
/// Example: spread_pct_fast(6500, 6700) = Some(307) (representing 3.07%)
#[inline]
pub fn spread_pct_fast(bid_ticks: Price, ask_ticks: Price) -> Option<u32> {
if bid_ticks == 0 || ask_ticks <= bid_ticks {
return None;
}
let spread = ask_ticks - bid_ticks;
// Calculate percentage in basis points (multiply by 10000 for 4 decimal places)
// We use u64 for intermediate calculation to avoid overflow
let spread_bps = ((spread as u64) * 10000) / (bid_ticks as u64);
// Convert back to u32 (should always fit since spreads are typically small)
Some(spread_bps as u32)
}
/// Calculate mid price (FAST VERSION)
///
///
/// Returns the midpoint between bid and ask in ticks.
/// Much faster than the Decimal version.
///
///
/// Example: mid_price_fast(6500, 6700) = Some(6600)
#[inline]
pub fn mid_price_fast(bid_ticks: Price, ask_ticks: Price) -> Option<Price> {
if bid_ticks == 0 || ask_ticks == 0 || ask_ticks <= bid_ticks {
return None;
}
// Use u64 to avoid overflow in addition
let sum = (bid_ticks as u64) + (ask_ticks as u64);
Some((sum / 2) as Price)
}
/// Calculate spread in ticks (FAST VERSION)
///
///
/// Simple subtraction - much faster than Decimal operations.
///
///
/// Example: spread_fast(6500, 6700) = Some(200) (representing $0.02 spread)
#[inline]
pub fn spread_fast(bid_ticks: Price, ask_ticks: Price) -> Option<Price> {
@@ -269,9 +268,9 @@ pub mod math {
}
/// Check if price is within valid range (FAST VERSION)
///
///
/// Much faster than converting to Decimal and back.
///
///
/// Example: is_valid_price_fast(6543, 1, 10000) = true
#[inline]
pub fn is_valid_price_fast(price_ticks: Price, min_tick: Price, max_tick: Price) -> bool {
@@ -310,14 +309,14 @@ pub mod math {
} else {
Decimal::ZERO
}
}
},
crate::types::Side::SELL => {
if executed_price < target_price {
(target_price - executed_price) / target_price
} else {
Decimal::ZERO
}
}
},
}
}
}
@@ -351,10 +350,7 @@ pub mod retry {
}
/// Retry a future with exponential backoff
pub async fn with_retry<F, Fut, T>(
config: &RetryConfig,
mut operation: F,
) -> Result<T>
pub async fn with_retry<F, Fut, T>(config: &RetryConfig, mut operation: F) -> Result<T>
where
F: FnMut() -> Fut,
Fut: Future<Output = Result<T>>,
@@ -367,7 +363,7 @@ pub mod retry {
Ok(result) => return Ok(result),
Err(err) => {
last_error = Some(err.clone());
if !err.is_retryable() || attempt == config.max_attempts - 1 {
return Err(err);
}
@@ -385,14 +381,21 @@ pub mod retry {
// Exponential backoff
delay = std::cmp::min(
Duration::from_nanos((delay.as_nanos() as f64 * config.backoff_factor) as u64),
Duration::from_nanos(
(delay.as_nanos() as f64 * config.backoff_factor) as u64,
),
config.max_delay,
);
}
},
}
}
Err(last_error.unwrap_or_else(|| PolyfillError::internal("Retry loop failed", std::io::Error::other("No error captured"))))
Err(last_error.unwrap_or_else(|| {
PolyfillError::internal(
"Retry loop failed",
std::io::Error::other("No error captured"),
)
}))
}
}
@@ -440,10 +443,7 @@ pub mod url {
}
/// Add query parameters to URL
pub fn add_query_params(
mut url: url::Url,
params: &[(&str, &str)],
) -> url::Url {
pub fn add_query_params(mut url: url::Url, params: &[(&str, &str)]) -> url::Url {
{
let mut query_pairs = url.query_pairs_mut();
for (key, value) in params {
@@ -481,7 +481,7 @@ pub mod rate_limit {
/// Try to consume a token, return true if successful
pub fn try_consume(&self) -> bool {
self.refill();
let mut tokens = self.tokens.lock().unwrap();
if *tokens > 0 {
*tokens -= 1;
@@ -495,7 +495,7 @@ pub mod rate_limit {
let now = SystemTime::now();
let mut last_refill = self.last_refill.lock().unwrap();
let elapsed = now.duration_since(*last_refill).unwrap_or_default();
if elapsed >= self.refill_rate {
let tokens_to_add = elapsed.as_nanos() / self.refill_rate.as_nanos();
let mut tokens = self.tokens.lock().unwrap();
@@ -513,7 +513,7 @@ mod tests {
#[test]
fn test_round_to_tick() {
use math::round_to_tick;
let price = Decimal::from_str("0.567").unwrap();
let tick = Decimal::from_str("0.01").unwrap();
let rounded = round_to_tick(price, tick);
@@ -523,7 +523,7 @@ mod tests {
#[test]
fn test_mid_price() {
use math::mid_price;
let bid = Decimal::from_str("0.50").unwrap();
let ask = Decimal::from_str("0.52").unwrap();
let mid = mid_price(bid, ask).unwrap();
@@ -533,11 +533,11 @@ mod tests {
#[test]
fn test_token_units_conversion() {
use math::{decimal_to_token_units, token_units_to_decimal};
let amount = Decimal::from_str("1.234567").unwrap();
let units = decimal_to_token_units(amount);
assert_eq!(units, 1_234_567);
let back = token_units_to_decimal(units);
assert_eq!(back, amount);
}
@@ -545,11 +545,11 @@ mod tests {
#[test]
fn test_address_validation() {
use address::parse_address;
let valid = "0x1234567890123456789012345678901234567890";
assert!(parse_address(valid).is_ok());
let invalid = "invalid_address";
assert!(parse_address(invalid).is_err());
}
}
}