tesser_markets/
lib.rs

1//! Thread-safe registry providing authoritative instrument metadata.
2
3use std::collections::HashMap;
4use std::fs;
5use std::path::{Path, PathBuf};
6use std::sync::{Arc, RwLock};
7
8use rust_decimal::Decimal;
9use serde::Deserialize;
10use tesser_broker::{BrokerError, ExecutionClient};
11use tesser_core::{AssetId, ExchangeId, Instrument, InstrumentKind, Price, Quantity, Symbol};
12use thiserror::Error;
13
14/// Shared registry storing immutable instrument definitions.
15#[derive(Clone, Default)]
16pub struct MarketRegistry {
17    inner: Arc<RwLock<HashMap<Symbol, Instrument>>>,
18}
19
20impl MarketRegistry {
21    /// Construct a registry from the provided instruments.
22    pub fn from_instruments(instruments: Vec<Instrument>) -> Result<Self, MarketRegistryError> {
23        if instruments.is_empty() {
24            return Err(MarketRegistryError::Empty);
25        }
26        let mut map = HashMap::new();
27        for instrument in instruments {
28            map.insert(instrument.symbol, instrument);
29        }
30        Ok(Self {
31            inner: Arc::new(RwLock::new(map)),
32        })
33    }
34
35    /// Load instrument metadata from a static TOML file.
36    pub fn load_from_file(path: impl AsRef<Path>) -> Result<Self, MarketRegistryError> {
37        let mut catalog = InstrumentCatalog::new();
38        catalog.add_file(path)?;
39        catalog.build()
40    }
41
42    /// Fetch market metadata from the execution client.
43    pub async fn load_from_exchange(
44        client: &(dyn ExecutionClient + Send + Sync),
45        category: &str,
46    ) -> Result<Self, MarketRegistryError> {
47        let instruments = client
48            .list_instruments(category)
49            .await
50            .map_err(MarketRegistryError::Broker)?;
51        Self::from_instruments(instruments)
52    }
53
54    /// Retrieve instrument metadata for a symbol.
55    pub fn get(&self, symbol: impl Into<Symbol>) -> Option<Instrument> {
56        let symbol = symbol.into();
57        self.inner
58            .read()
59            .ok()
60            .and_then(|map| map.get(&symbol).cloned())
61    }
62
63    /// Validate that each supplied symbol exists.
64    pub fn validate_symbols<I, S>(&self, symbols: I) -> Result<(), MarketRegistryError>
65    where
66        I: IntoIterator<Item = S>,
67        S: Into<Symbol>,
68    {
69        let map = self
70            .inner
71            .read()
72            .map_err(|_| MarketRegistryError::Poisoned)?;
73        for symbol in symbols {
74            let symbol = symbol.into();
75            if !map.contains_key(&symbol) {
76                return Err(MarketRegistryError::UnknownSymbol(symbol));
77            }
78        }
79        Ok(())
80    }
81
82    /// Return a snapshot of all registered instruments.
83    pub fn instruments(&self) -> Vec<Instrument> {
84        self.inner
85            .read()
86            .map(|map| map.values().cloned().collect())
87            .unwrap_or_default()
88    }
89
90    /// Round a quantity down to the coarsest lot size shared by two venues.
91    pub fn normalize_pair_quantity(
92        &self,
93        first: Symbol,
94        second: Symbol,
95        quantity: Quantity,
96    ) -> Option<Quantity> {
97        let first_instr = self.get(first)?;
98        let second_instr = self.get(second)?;
99        let step = first_instr.lot_size.max(second_instr.lot_size);
100        if step <= Decimal::ZERO {
101            return Some(quantity);
102        }
103        let normalized = (quantity / step).floor() * step;
104        Some(normalized)
105    }
106}
107
108#[derive(Debug, Error)]
109pub enum MarketRegistryError {
110    #[error("no instruments were loaded into the registry")]
111    Empty,
112    #[error("unknown symbol '{0}'")]
113    UnknownSymbol(Symbol),
114    #[error("instrument '{symbol}' missing required field '{field}'")]
115    MissingField { symbol: Symbol, field: &'static str },
116    #[error("instrument '{symbol}' has conflicting value for '{field}'")]
117    ConflictingField { symbol: Symbol, field: &'static str },
118    #[error("markets file is invalid: {0}")]
119    InvalidFormat(toml::de::Error),
120    #[error("failed to read markets file at {path}: {source}")]
121    Io {
122        path: PathBuf,
123        #[source]
124        source: std::io::Error,
125    },
126    #[error("market registry poisoned by concurrent access")]
127    Poisoned,
128    #[error("broker error: {0}")]
129    Broker(BrokerError),
130}
131
132/// Accumulates instrument metadata from multiple sources before producing a registry.
133#[derive(Default)]
134pub struct InstrumentCatalog {
135    entries: HashMap<Symbol, InstrumentInfo>,
136}
137
138impl InstrumentCatalog {
139    #[must_use]
140    pub fn new() -> Self {
141        Self {
142            entries: HashMap::new(),
143        }
144    }
145
146    pub fn add_file(&mut self, path: impl AsRef<Path>) -> Result<(), MarketRegistryError> {
147        let path = path.as_ref();
148        let contents = fs::read_to_string(path).map_err(|err| MarketRegistryError::Io {
149            path: path.to_path_buf(),
150            source: err,
151        })?;
152        let file: MarketFile =
153            toml::from_str(&contents).map_err(MarketRegistryError::InvalidFormat)?;
154        for raw in file.markets {
155            self.insert(raw.into_typed())?;
156        }
157        Ok(())
158    }
159
160    pub fn add_instruments(
161        &mut self,
162        instruments: Vec<Instrument>,
163    ) -> Result<(), MarketRegistryError> {
164        for instrument in instruments {
165            self.insert(InstrumentInfo::from(instrument))?;
166        }
167        Ok(())
168    }
169
170    #[must_use]
171    pub fn is_empty(&self) -> bool {
172        self.entries.is_empty()
173    }
174
175    pub fn build(self) -> Result<MarketRegistry, MarketRegistryError> {
176        let instruments = self
177            .entries
178            .into_values()
179            .map(|info| info.into_instrument())
180            .collect::<Result<Vec<_>, _>>()?;
181        MarketRegistry::from_instruments(instruments)
182    }
183
184    fn insert(&mut self, info: InstrumentInfo) -> Result<(), MarketRegistryError> {
185        let key = info.key();
186        if let Some(existing) = self.entries.remove(&key) {
187            let merged = existing.merge(info)?;
188            self.entries.insert(key, merged);
189        } else {
190            self.entries.insert(key, info);
191        }
192        Ok(())
193    }
194}
195
196#[derive(Deserialize)]
197struct MarketFile {
198    #[serde(default)]
199    markets: Vec<RawInstrumentInfo>,
200}
201
202#[derive(Clone, Debug, Deserialize)]
203struct RawInstrumentInfo {
204    exchange: ExchangeId,
205    symbol: String,
206    #[serde(default)]
207    base: Option<String>,
208    #[serde(default)]
209    quote: Option<String>,
210    #[serde(default)]
211    settlement_currency: Option<String>,
212    #[serde(default)]
213    kind: Option<InstrumentKind>,
214    #[serde(default)]
215    tick_size: Option<Price>,
216    #[serde(default)]
217    lot_size: Option<Quantity>,
218}
219
220impl RawInstrumentInfo {
221    fn into_typed(self) -> InstrumentInfo {
222        let exchange = self.exchange;
223        let symbol = Symbol::from_code(exchange, self.symbol);
224        InstrumentInfo {
225            symbol,
226            base: self.base.map(|code| AssetId::from_code(exchange, code)),
227            quote: self.quote.map(|code| AssetId::from_code(exchange, code)),
228            settlement_currency: self
229                .settlement_currency
230                .map(|code| AssetId::from_code(exchange, code)),
231            kind: self.kind,
232            tick_size: self.tick_size,
233            lot_size: self.lot_size,
234        }
235    }
236}
237
238#[derive(Clone, Debug)]
239struct InstrumentInfo {
240    symbol: Symbol,
241    base: Option<AssetId>,
242    quote: Option<AssetId>,
243    settlement_currency: Option<AssetId>,
244    kind: Option<InstrumentKind>,
245    tick_size: Option<Price>,
246    lot_size: Option<Quantity>,
247}
248
249impl InstrumentInfo {
250    fn key(&self) -> Symbol {
251        self.symbol
252    }
253
254    fn merge(self, other: InstrumentInfo) -> Result<InstrumentInfo, MarketRegistryError> {
255        let symbol = self.key();
256        if symbol != other.key() {
257            return Err(MarketRegistryError::ConflictingField {
258                symbol,
259                field: "symbol",
260            });
261        }
262        Ok(InstrumentInfo {
263            symbol,
264            base: merge_field(symbol, "base", self.base, other.base)?,
265            quote: merge_field(symbol, "quote", self.quote, other.quote)?,
266            settlement_currency: merge_field(
267                symbol,
268                "settlement_currency",
269                self.settlement_currency,
270                other.settlement_currency,
271            )?,
272            kind: merge_field(symbol, "kind", self.kind, other.kind)?,
273            tick_size: merge_field(symbol, "tick_size", self.tick_size, other.tick_size)?,
274            lot_size: merge_field(symbol, "lot_size", self.lot_size, other.lot_size)?,
275        })
276    }
277
278    fn into_instrument(self) -> Result<Instrument, MarketRegistryError> {
279        let symbol = self.symbol;
280        let base = require_field(symbol, "base", self.base)?;
281        let quote = require_field(symbol, "quote", self.quote)?;
282        let settlement = require_field(symbol, "settlement_currency", self.settlement_currency)?;
283        let kind = require_field(symbol, "kind", self.kind)?;
284        let tick_size = require_field(symbol, "tick_size", self.tick_size)?;
285        let lot_size = require_field(symbol, "lot_size", self.lot_size)?;
286
287        Ok(Instrument {
288            symbol,
289            base,
290            quote,
291            kind,
292            settlement_currency: settlement,
293            tick_size,
294            lot_size,
295        })
296    }
297}
298
299fn merge_field<T: Clone + PartialEq>(
300    symbol: Symbol,
301    field: &'static str,
302    primary: Option<T>,
303    secondary: Option<T>,
304) -> Result<Option<T>, MarketRegistryError> {
305    match (primary, secondary) {
306        (Some(existing), Some(replacement)) => {
307            if existing != replacement {
308                Err(MarketRegistryError::ConflictingField { symbol, field })
309            } else {
310                Ok(Some(existing))
311            }
312        }
313        (Some(existing), None) => Ok(Some(existing)),
314        (None, Some(value)) => Ok(Some(value)),
315        (None, None) => Ok(None),
316    }
317}
318
319fn require_field<T>(
320    symbol: Symbol,
321    field: &'static str,
322    value: Option<T>,
323) -> Result<T, MarketRegistryError> {
324    value.ok_or(MarketRegistryError::MissingField { symbol, field })
325}
326
327impl From<Instrument> for InstrumentInfo {
328    fn from(value: Instrument) -> Self {
329        Self {
330            symbol: value.symbol,
331            base: Some(value.base),
332            quote: Some(value.quote),
333            settlement_currency: Some(value.settlement_currency),
334            kind: Some(value.kind),
335            tick_size: Some(value.tick_size),
336            lot_size: Some(value.lot_size),
337        }
338    }
339}
tesser_markets/lib.rs

tesser_markets/
lib.rs
