feat: add fixed-point decimal fields and WebSocket sharding support

Add support for latest Kalshi API changes:

- Add _fp (fixed-point decimal string) fields to models:
  - Market: volume_fp, volume_24h_fp, open_interest_fp
  - Trade: count_fp
  - Candlestick: volume_fp, open_interest_fp
  - Order: fill_count_fp, remaining_count_fp, initial_count_fp
  - Fill: count_fp
  - Position: position_fp, resting_orders_count_fp, total_cost_shares_fp
  - Series: volume, volume_fp
  - Settlement: yes_count_fp, no_count_fp

- Add include_latest_before_start parameter to candlestick queries

- Add WebSocket communications channel sharding support:
  - CommunicationsSharding struct for shard_factor/shard_key config
  - subscribe_communications_sharded() method on KalshiStreamHandle
  - Updated build_subscribe() to include sharding parameters

Author: pbeets

src/batch.rs

@@ -772,8 +772,11 @@ mod tests {
                 yes_price_dollars: None,
                 no_price_dollars: None,
                 fill_count: 0,
+                fill_count_fp: None,
                 remaining_count: 0,
+                remaining_count_fp: None,
                 initial_count: 10,
+                initial_count_fp: None,
                 taker_fees: None,
                 maker_fees: None,
                 taker_fill_cost: None,

src/models/fill.rs

@@ -20,6 +20,9 @@ pub struct Fill {
     pub side: Side,
     pub action: Action,
     pub count: i64,
+    /// Count (fixed-point decimal string).
+    #[serde(default)]
+    pub count_fp: Option<String>,
     /// Price in cents.
     pub yes_price: i64,
     /// Price in cents.

src/models/market.rs

@@ -206,12 +206,21 @@ pub struct Market {
     /// Total contracts traded.
     #[serde(default)]
     pub volume: Option<i64>,
+    /// Total contracts traded (fixed-point decimal string).
+    #[serde(default)]
+    pub volume_fp: Option<String>,
     /// 24-hour trading volume.
     #[serde(default)]
     pub volume_24h: Option<i64>,
+    /// 24-hour trading volume (fixed-point decimal string).
+    #[serde(default)]
+    pub volume_24h_fp: Option<String>,
     /// Contracts outstanding.
     #[serde(default)]
     pub open_interest: Option<i64>,
+    /// Contracts outstanding (fixed-point decimal string).
+    #[serde(default)]
+    pub open_interest_fp: Option<String>,
 
     /// Notional value per contract in dollars.
     #[serde(default)]
@@ -592,6 +601,9 @@ pub struct Trade {
     pub price: Option<f64>,
     /// Contract quantity.
     pub count: i64,
+    /// Contract quantity (fixed-point decimal string).
+    #[serde(default)]
+    pub count_fp: Option<String>,
     /// Yes side price in cents.
     pub yes_price: i64,
     /// No side price in cents.
@@ -798,9 +810,15 @@ pub struct Candlestick {
     /// Trading volume during the period.
     #[serde(default)]
     pub volume: Option<i64>,
+    /// Trading volume during the period (fixed-point decimal string).
+    #[serde(default)]
+    pub volume_fp: Option<String>,
     /// Open interest at end of period.
     #[serde(default)]
     pub open_interest: Option<i64>,
+    /// Open interest at end of period (fixed-point decimal string).
+    #[serde(default)]
+    pub open_interest_fp: Option<String>,
 }
 
 /// Response from GET /series/{series_ticker}/markets/{ticker}/candlesticks.
@@ -837,6 +855,9 @@ pub struct GetCandlesticksParams {
     pub end_ts: i64,
     /// Candlestick period interval.
     pub period_interval: CandlestickPeriod,
+    /// Include synthetic candlestick before start_ts for price continuity.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub include_latest_before_start: Option<bool>,
 }
 
 impl GetCandlesticksParams {
@@ -868,15 +889,27 @@ impl GetCandlesticksParams {
             start_ts,
             end_ts,
             period_interval,
+            include_latest_before_start: None,
         })
     }
 
+    /// Include synthetic candlestick before start_ts for price continuity.
+    #[must_use]
+    pub fn include_latest_before_start(mut self, include: bool) -> Self {
+        self.include_latest_before_start = Some(include);
+        self
+    }
+
     #[must_use]
     pub fn to_query_string(&self) -> String {
         let mut qb = QueryBuilder::new();
         qb.push("start_ts", self.start_ts);
         qb.push("end_ts", self.end_ts);
         qb.push("period_interval", self.period_interval.as_minutes());
+        qb.push_opt(
+            "include_latest_before_start",
+            self.include_latest_before_start,
+        );
         qb.build()
     }
 }
@@ -1091,13 +1124,20 @@ mod tests {
             "status": "active",
             "title": "Will Bitcoin reach $50,000?",
             "volume": 1000,
-            "open_interest": 500
+            "volume_fp": "1000.5",
+            "volume_24h": 500,
+            "volume_24h_fp": "500.25",
+            "open_interest": 250,
+            "open_interest_fp": "250.125"
         }"#;
         let market: Market = serde_json::from_str(json).unwrap();
         assert_eq!(market.ticker, "KXBTC-25JAN10-B50000");
         assert_eq!(market.market_type, MarketType::Binary);
         assert_eq!(market.status, MarketStatus::Active);
         assert_eq!(market.volume, Some(1000));
+        assert_eq!(market.volume_fp, Some("1000.5".to_string()));
+        assert_eq!(market.volume_24h_fp, Some("500.25".to_string()));
+        assert_eq!(market.open_interest_fp, Some("250.125".to_string()));
     }
 
     #[test]
@@ -1152,6 +1192,7 @@ mod tests {
             "trade_id": "abc123",
             "ticker": "KXBTC-25JAN10-B50000",
             "count": 10,
+            "count_fp": "10.5",
             "yes_price": 50,
             "no_price": 50,
             "taker_side": "yes",
@@ -1160,6 +1201,7 @@ mod tests {
         let trade: Trade = serde_json::from_str(json).unwrap();
         assert_eq!(trade.trade_id, "abc123");
         assert_eq!(trade.count, 10);
+        assert_eq!(trade.count_fp, Some("10.5".to_string()));
         assert_eq!(trade.taker_side, TakerSide::Yes);
     }
 

src/models/order.rs

@@ -42,8 +42,17 @@ pub struct Order {
     #[serde(default)]
     pub no_price_dollars: Option<String>,
     pub fill_count: i64,
+    /// Fill count (fixed-point decimal string).
+    #[serde(default)]
+    pub fill_count_fp: Option<String>,
     pub remaining_count: i64,
+    /// Remaining count (fixed-point decimal string).
+    #[serde(default)]
+    pub remaining_count_fp: Option<String>,
     pub initial_count: i64,
+    /// Initial count (fixed-point decimal string).
+    #[serde(default)]
+    pub initial_count_fp: Option<String>,
     /// Fees in cents.
     #[serde(default)]
     pub taker_fees: Option<i64>,

src/models/position.rs

@@ -14,6 +14,9 @@ pub struct MarketPosition {
     pub total_traded_dollars: Option<String>,
     /// Negative = NO position, positive = YES position.
     pub position: i64,
+    /// Position (fixed-point decimal string).
+    #[serde(default)]
+    pub position_fp: Option<String>,
     /// Position cost in cents.
     pub market_exposure: i64,
     #[serde(default)]
@@ -23,6 +26,9 @@ pub struct MarketPosition {
     #[serde(default)]
     pub realized_pnl_dollars: Option<String>,
     pub resting_orders_count: i64,
+    /// Resting orders count (fixed-point decimal string).
+    #[serde(default)]
+    pub resting_orders_count_fp: Option<String>,
     /// In cents.
     pub fees_paid: i64,
     #[serde(default)]
@@ -40,6 +46,9 @@ pub struct EventPosition {
     #[serde(default)]
     pub total_cost_dollars: Option<String>,
     pub total_cost_shares: i64,
+    /// Total cost shares (fixed-point decimal string).
+    #[serde(default)]
+    pub total_cost_shares_fp: Option<String>,
     /// In cents.
     pub event_exposure: i64,
     #[serde(default)]

src/models/series.rs

@@ -100,6 +100,14 @@ pub struct Series {
     pub frequency: String,
     /// Title of the series.
     pub title: String,
+    /// Total contracts traded across all events in this series.
+    /// Only present when `include_volume=true` in the request.
+    #[serde(default)]
+    pub volume: Option<i64>,
+    /// Total contracts traded (fixed-point decimal string).
+    /// Only present when `include_volume=true` in the request.
+    #[serde(default)]
+    pub volume_fp: Option<String>,
 }
 
 /// Response from GET /series/{series_ticker}.

src/models/settlement.rs

@@ -16,10 +16,16 @@ pub struct Settlement {
     pub market_result: MarketResult,
     /// Number of YES contracts held at settlement.
     pub yes_count: i64,
+    /// Number of YES contracts (fixed-point decimal string).
+    #[serde(default)]
+    pub yes_count_fp: Option<String>,
     /// Total cost of YES contracts in cents.
     pub yes_total_cost: i64,
     /// Number of NO contracts held at settlement.
     pub no_count: i64,
+    /// Number of NO contracts (fixed-point decimal string).
+    #[serde(default)]
+    pub no_count_fp: Option<String>,
     /// Total cost of NO contracts in cents.
     pub no_total_cost: i64,
     /// Revenue from settlement in cents.

src/ws.rs

@@ -170,7 +170,7 @@ use std::time::Duration;
 
 pub use channel::Channel;
 pub use client::{KalshiStreamClient, KalshiStreamHandle};
-pub use command::{SubscribeResult, UnsubscribeResult};
+pub use command::{CommunicationsSharding, SubscribeResult, UnsubscribeResult};
 pub use message::{
     Action, FillData, MarketLifecycleData, MarketLifecycleEventType, MarketPositionData,
     MultivariateLookupData, MveLeg, OrderbookDeltaData, OrderbookSnapshotData, Side, StreamMessage,

src/ws/client.rs

@@ -613,11 +613,77 @@ impl KalshiStreamHandle {
 
     // --- Internal methods ---
 
+    /// Subscribe to the communications channel with sharding for high-throughput RFQ/quote traffic.
+    ///
+    /// Sharding allows distributing communications traffic across multiple connections.
+    /// Each connection specifies a shard_factor (total number of shards) and shard_key
+    /// (which shard this connection handles).
+    ///
+    /// # Arguments
+    ///
+    /// * `sharding` - Sharding configuration specifying shard_factor and shard_key.
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    /// # use kalshi_trade_rs::ws::{Channel, KalshiStreamHandle, CommunicationsSharding};
+    /// # async fn example(handle: &mut KalshiStreamHandle) -> Result<(), Box<dyn std::error::Error>> {
+    /// // Set up 4 shards, this connection handles shard 0
+    /// let sharding = CommunicationsSharding::new(4, 0);
+    /// handle.subscribe_communications_sharded(sharding).await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn subscribe_communications_sharded(
+        &mut self,
+        sharding: super::command::CommunicationsSharding,
+    ) -> Result<()> {
+        let result = self
+            .subscribe_raw_with_sharding(&[Channel::Communications], &[], Some(sharding))
+            .await?;
+
+        // Check for failures
+        if !result.failed.is_empty() {
+            let errors: Vec<String> = result
+                .failed
+                .iter()
+                .map(|e| format!("{}: {}", e.code, e.message))
+                .collect();
+            return Err(Error::Api(errors.join("; ")));
+        }
+
+        // Update state if successful
+        if let Some(sub) = result.successful.first() {
+            let mut subs = self
+                .subscriptions
+                .write()
+                .expect("subscription lock poisoned");
+            subs.entry(Channel::Communications)
+                .or_insert(SubscriptionState {
+                    sid: sub.sid,
+                    markets: std::collections::HashSet::new(),
+                });
+        }
+
+        Ok(())
+    }
+
     /// Raw subscribe without local state management.
     async fn subscribe_raw(
         &self,
         channels: &[Channel],
         markets: &[&str],
+    ) -> Result<SubscribeResult> {
+        self.subscribe_raw_with_sharding(channels, markets, None)
+            .await
+    }
+
+    /// Raw subscribe with optional sharding support.
+    async fn subscribe_raw_with_sharding(
+        &self,
+        channels: &[Channel],
+        markets: &[&str],
+        sharding: Option<super::command::CommunicationsSharding>,
     ) -> Result<SubscribeResult> {
         let (tx, rx) = oneshot::channel();
 
@@ -629,6 +695,7 @@ impl KalshiStreamHandle {
         let cmd = StreamCommand::Subscribe {
             channels: channel_strings,
             market_tickers: market_strings,
+            sharding,
             response: tx,
         };
 
@@ -786,6 +853,7 @@ mod tests {
                     channels,
                     market_tickers,
                     response,
+                    ..
                 }) => {
                     // Verify correct command type and params
                     assert_eq!(channels, vec!["ticker"]);