scope/market/
catalog.rs

1//! The venue catalog: the venue descriptors kept in the repo
2//! (`crates/scope-core/venues/`), synced into the scope home by
3//! `scope venues update`. Nothing is compiled in, so a new or fixed venue
4//! reaches users without a release.
5//!
6//! The catalog is an `index.json` (schema, and each venue's file and
7//! SHA-256) and one YAML file per venue. A sync fetches the index, then
8//! every file; it checks each against its checksum and its id, writes them
9//! to a new folder and swaps it in, so a failed sync leaves the old catalog
10//! as it was. Checksums are over the text with `\n` line ends, so a Windows
11//! checkout (`\r\n`) and the served files agree.
12
13use crate::error::{Result, ScopeError};
14use crate::http::{HttpClient, Request};
15use crate::market::descriptor::VenueDescriptor;
16use serde::{Deserialize, Serialize};
17use sha2::{Digest, Sha256};
18use std::path::{Path, PathBuf};
19
20/// Where `scope venues update` fetches the catalog from, unless the config
21/// says otherwise (`venues.catalog_url`).
22pub const DEFAULT_CATALOG_URL: &str = "https://raw.githubusercontent.com/robot-accomplice/scope-blockchain-analysis/main/crates/scope-core/venues";
23
24/// The index file of a catalog.
25pub const INDEX_FILE: &str = "index.json";
26
27/// The newest index schema this scope reads.
28pub const SCHEMA: u32 = 1;
29
30/// The file the user copies to describe an exchange of their own.
31pub const EXAMPLE_FILE: &str = "example.yaml.sample";
32
33/// The catalog's table of contents.
34#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
35pub struct Index {
36    pub schema: u32,
37    pub venues: Vec<Entry>,
38}
39
40/// One venue of the catalog.
41#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
42pub struct Entry {
43    pub id: String,
44    pub file: String,
45    pub sha256: String,
46}
47
48/// What a sync changed.
49#[derive(Debug, Clone, Default, PartialEq, Eq)]
50pub struct SyncReport {
51    pub added: Vec<String>,
52    pub updated: Vec<String>,
53    pub removed: Vec<String>,
54    pub unchanged: usize,
55}
56
57impl SyncReport {
58    /// `12 venues: 1 added (weex), 0 updated, 0 removed`.
59    pub fn summary(&self) -> String {
60        let total = self.added.len() + self.updated.len() + self.unchanged;
61        let list = |ids: &[String]| {
62            if ids.is_empty() {
63                String::new()
64            } else {
65                format!(" ({})", ids.join(", "))
66            }
67        };
68        format!(
69            "{total} venues: {} added{}, {} updated{}, {} removed{}",
70            self.added.len(),
71            list(&self.added),
72            self.updated.len(),
73            list(&self.updated),
74            self.removed.len(),
75            list(&self.removed)
76        )
77    }
78}
79
80/// The SHA-256 of `text` with `\n` line ends, in lowercase hex.
81pub fn checksum(text: &str) -> String {
82    hex::encode(Sha256::digest(text.replace("\r\n", "\n").as_bytes()))
83}
84
85/// The index of the descriptors in `dir` (`*.yaml`), sorted by file name.
86/// Used to write the repo's `index.json` and to check it.
87pub fn build_index(dir: &Path) -> Result<Index> {
88    let mut files: Vec<PathBuf> = std::fs::read_dir(dir)
89        .map_err(|e| ScopeError::io_context(e, format!("Cannot read {}", dir.display())))?
90        .flatten()
91        .map(|e| e.path())
92        .filter(|p| p.extension().is_some_and(|ext| ext == "yaml"))
93        .collect();
94    files.sort();
95    let mut venues = Vec::new();
96    for path in files {
97        let text = std::fs::read_to_string(&path)
98            .map_err(|e| ScopeError::io_context(e, format!("Cannot read {}", path.display())))?;
99        let desc = parse(&text, &path.display().to_string())?;
100        venues.push(Entry {
101            id: desc.id,
102            file: path.file_name().unwrap().to_string_lossy().into_owned(),
103            sha256: checksum(&text),
104        });
105    }
106    Ok(Index {
107        schema: SCHEMA,
108        venues,
109    })
110}
111
112/// `text` as a descriptor, or an error naming `what`.
113fn parse(text: &str, what: &str) -> Result<VenueDescriptor> {
114    serde_saphyr::from_str(text)
115        .map_err(|e| ScopeError::InvalidInput(format!("Venue file {what} does not parse: {e}")))
116}
117
118/// A catalog file name: a plain `*.yaml` name, nothing that leaves the
119/// catalog folder.
120fn safe_file_name(file: &str) -> bool {
121    file.ends_with(".yaml")
122        && !file.starts_with('.')
123        && file
124            .chars()
125            .all(|c| c.is_ascii_alphanumeric() || matches!(c, '_' | '-' | '.'))
126}
127
128/// Fetches `url` and returns its body, or an error that names it.
129async fn fetch(http: &dyn HttpClient, url: &str) -> Result<String> {
130    let response = http.send(Request::get(url)).await?;
131    if !response.is_success() {
132        return Err(ScopeError::Network(format!(
133            "{url} answered HTTP {}",
134            response.status_code
135        )));
136    }
137    Ok(response.body)
138}
139
140/// Syncs the catalog at `base_url` into `dest`: fetches the index and every
141/// venue file, checks each, and swaps the new catalog in. On any error the
142/// old catalog stays as it was.
143pub async fn sync(http: &dyn HttpClient, base_url: &str, dest: &Path) -> Result<SyncReport> {
144    let base = base_url.trim_end_matches('/');
145    let index: Index = serde_json::from_str(&fetch(http, &format!("{base}/{INDEX_FILE}")).await?)
146        .map_err(|e| {
147        ScopeError::InvalidInput(format!("The venue catalog index does not parse: {e}"))
148    })?;
149    if index.schema > SCHEMA {
150        return Err(ScopeError::InvalidInput(format!(
151            "The venue catalog needs a newer scope (catalog schema {}, this scope reads {SCHEMA}). Update scope, then run `scope venues update` again.",
152            index.schema
153        )));
154    }
155
156    let mut files = Vec::with_capacity(index.venues.len());
157    for entry in &index.venues {
158        if !safe_file_name(&entry.file) {
159            return Err(ScopeError::InvalidInput(format!(
160                "The venue catalog names an unsafe file: {:?}",
161                entry.file
162            )));
163        }
164        let text = fetch(http, &format!("{base}/{}", entry.file)).await?;
165        let got = checksum(&text);
166        if got != entry.sha256 {
167            return Err(ScopeError::InvalidInput(format!(
168                "Venue file {} does not match the catalog index (sha256 {got}, expected {}); the catalog was not changed.",
169                entry.file, entry.sha256
170            )));
171        }
172        let desc = parse(&text, &entry.file)?;
173        if desc.id != entry.id {
174            return Err(ScopeError::InvalidInput(format!(
175                "Venue file {} has id {:?}; the catalog index says {:?}.",
176                entry.file, desc.id, entry.id
177            )));
178        }
179        files.push((entry, text));
180    }
181
182    let old = installed(dest);
183    let parent = dest.parent().unwrap_or(dest);
184    std::fs::create_dir_all(parent)
185        .map_err(|e| ScopeError::io_context(e, format!("Cannot create {}", parent.display())))?;
186    let staging = parent.join("catalog.new");
187    let _ = std::fs::remove_dir_all(&staging);
188    std::fs::create_dir_all(&staging)
189        .map_err(|e| ScopeError::io_context(e, format!("Cannot create {}", staging.display())))?;
190    for (entry, text) in &files {
191        let path = staging.join(&entry.file);
192        std::fs::write(&path, text)
193            .map_err(|e| ScopeError::io_context(e, format!("Cannot write {}", path.display())))?;
194    }
195    let index_json = serde_json::to_string_pretty(&index).unwrap_or_default();
196    std::fs::write(staging.join(INDEX_FILE), index_json + "\n")
197        .map_err(|e| ScopeError::io_context(e, "Cannot write the catalog index".to_string()))?;
198    let retired = parent.join("catalog.old");
199    let _ = std::fs::remove_dir_all(&retired);
200    if dest.exists() {
201        std::fs::rename(dest, &retired)
202            .map_err(|e| ScopeError::io_context(e, format!("Cannot replace {}", dest.display())))?;
203    }
204    if let Err(e) = std::fs::rename(&staging, dest) {
205        // Put the old catalog back before reporting.
206        let _ = std::fs::rename(&retired, dest);
207        return Err(ScopeError::io_context(
208            e,
209            format!("Cannot install the catalog in {}", dest.display()),
210        ));
211    }
212    let _ = std::fs::remove_dir_all(&retired);
213
214    let mut report = SyncReport::default();
215    for entry in &index.venues {
216        match old.iter().find(|o| o.id == entry.id) {
217            None => report.added.push(entry.id.clone()),
218            Some(o) if o.sha256 != entry.sha256 => report.updated.push(entry.id.clone()),
219            Some(_) => report.unchanged += 1,
220        }
221    }
222    report.removed = old
223        .iter()
224        .filter(|o| !index.venues.iter().any(|e| e.id == o.id))
225        .map(|o| o.id.clone())
226        .collect();
227    Ok(report)
228}
229
230/// The entries of the catalog installed in `dir` (from its index), or none.
231pub fn installed(dir: &Path) -> Vec<Entry> {
232    std::fs::read_to_string(dir.join(INDEX_FILE))
233        .ok()
234        .and_then(|text| serde_json::from_str::<Index>(&text).ok())
235        .map(|index| index.venues)
236        .unwrap_or_default()
237}
238
239/// Where the example venue file goes: the user's venues folder.
240pub fn example_path() -> PathBuf {
241    crate::paths::venues_dir().join(EXAMPLE_FILE)
242}
243
244/// Writes [`EXAMPLE`] to `path` unless a file is there already.
245pub fn write_example(path: &Path) -> std::io::Result<()> {
246    if path.exists() {
247        return Ok(());
248    }
249    if let Some(dir) = path.parent() {
250        std::fs::create_dir_all(dir)?;
251    }
252    std::fs::write(path, EXAMPLE)
253}
254
255/// A commented venue descriptor: how to describe an exchange by hand. It is
256/// not a venue (scope reads `*.yaml` and `*.yml`, not `*.sample`).
257pub const EXAMPLE: &str = r#"# How to add an exchange to scope by hand
258# =========================================
259#
260# Copy this file to a name ending in .yaml in this folder (for example
261# myexchange.yaml), fill it in, and check it with:
262#
263#     scope venues validate ~/.scope-bca/venues/myexchange.yaml
264#
265# Files in this folder override a catalog venue with the same `id`. The
266# catalog itself lives in the `catalog/` folder and is refreshed by
267# `scope venues update`; do not edit it, your changes there are replaced.
268#
269# Every path below is the exchange's public REST API: no keys are needed.
270# `{pair}` is the venue's symbol for the pair (see `symbol`), `{limit}` the
271# number of items asked for, `{interval}` a candle interval such as 1h.
272
273# A short, lowercase name: what `--venue` takes.
274id: myexchange
275# The name shown in listings.
276name: My Exchange Spot
277# The API's root URL, without a trailing slash.
278base_url: https://api.myexchange.example
279timeout_secs: 15
280rate_limit_per_sec: 10
281
282# How the pair is written in the exchange's URLs: BTC/USDT becomes BTCUSDT
283# with "{base}{quote}", BTC_USDT with "{base}_{quote}", BTC-USDT with
284# "{base}-{quote}". `default_quote` is used when only the base is given.
285symbol:
286  template: "{base}{quote}"
287  default_quote: USDT
288
289capabilities:
290  # The order book. `level_format: positional` means each level is
291  # ["price", "quantity"]; use `object` with `level_price` and
292  # `level_size` when levels are {"price": ..., "size": ...}.
293  order_book:
294    path: /api/v3/depth
295    params:
296      symbol: "{pair}"
297      limit: "100"
298    response:
299      asks_key: asks
300      bids_key: bids
301      level_format: positional
302
303  # The 24-hour ticker: map each field to the key the exchange uses.
304  ticker:
305    path: /api/v3/ticker/24hr
306    params:
307      symbol: "{pair}"
308    response:
309      last_price: lastPrice
310      high_24h: highPrice
311      low_24h: lowPrice
312      volume_24h: volume
313      quote_volume_24h: quoteVolume
314      best_bid: bidPrice
315      best_ask: askPrice
316
317  # Recent trades, for the bot-pattern advisory. `side` maps the
318  # exchange's value to buy or sell.
319  trades:
320    path: /api/v3/trades
321    params:
322      symbol: "{pair}"
323      limit: "{limit}"
324    response:
325      price: price
326      quantity: qty
327      quote_quantity: quoteQty
328      timestamp_ms: time
329      id: id
330      side:
331        field: isBuyerMaker
332        mapping:
333          "true": sell
334          "false": buy
335
336  # Candles, for `scope market ohlc`. `array_of_arrays` reads rows such as
337  # [open_time, open, high, low, close, volume, close_time].
338  ohlc:
339    path: /api/v3/klines
340    params:
341      symbol: "{pair}"
342      interval: "{interval}"
343      limit: "{limit}"
344    response:
345      ohlc_format: array_of_arrays
346      ohlc_fields: [open_time, open, high, low, close, volume, close_time]
347
348# The venues in catalog/ are complete, working examples of other API shapes
349# (nested data keys, object levels, other symbol formats): copy the one
350# closest to your exchange.
351"#;
352
353#[cfg(test)]
354mod tests {
355    use super::*;
356    use crate::http::Response;
357    use std::sync::Mutex;
358
359    /// The repo catalog folder.
360    fn repo() -> PathBuf {
361        Path::new(env!("CARGO_MANIFEST_DIR")).join("venues")
362    }
363
364    #[test]
365    fn test_the_repo_index_matches_the_venue_files() {
366        // A venue file changed without `index.json`? Regenerate it:
367        //   SCOPE_UPDATE_VENUE_INDEX=1 cargo test -p scope-bca-core repo_index
368        let built = build_index(&repo()).unwrap();
369        let path = repo().join(INDEX_FILE);
370        if std::env::var_os("SCOPE_UPDATE_VENUE_INDEX").is_some() {
371            let json = serde_json::to_string_pretty(&built).unwrap() + "\n";
372            std::fs::write(&path, json).unwrap();
373        }
374        let on_disk: Index =
375            serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
376        assert_eq!(
377            on_disk, built,
378            "crates/scope-core/venues/index.json is stale; run with SCOPE_UPDATE_VENUE_INDEX=1 to rewrite it"
379        );
380        assert!(built.venues.len() >= 11);
381    }
382
383    #[test]
384    fn test_checksums_ignore_windows_line_ends() {
385        assert_eq!(
386            checksum("id: a\nname: b\n"),
387            checksum("id: a\r\nname: b\r\n")
388        );
389        assert_ne!(checksum("id: a\n"), checksum("id: b\n"));
390    }
391
392    #[test]
393    fn test_the_example_is_a_valid_descriptor_that_is_not_loaded() {
394        let desc: VenueDescriptor = serde_saphyr::from_str(EXAMPLE).unwrap();
395        assert_eq!(desc.id, "myexchange");
396        assert_eq!(desc.capability_names().len(), 4);
397        let dir = tempfile::tempdir().unwrap();
398        let path = dir.path().join(EXAMPLE_FILE);
399        write_example(&path).unwrap();
400        // Not overwritten once the user has it.
401        std::fs::write(&path, "mine").unwrap();
402        write_example(&path).unwrap();
403        assert_eq!(std::fs::read_to_string(&path).unwrap(), "mine");
404        let registry =
405            crate::market::VenueRegistry::load_dirs(&dir.path().join("none"), dir.path()).unwrap();
406        assert!(registry.is_empty(), "a .sample file is not a venue");
407    }
408
409    #[test]
410    fn test_file_names_stay_in_the_catalog() {
411        assert!(safe_file_name("weex.yaml"));
412        assert!(safe_file_name("crypto_com.yaml"));
413        for bad in [
414            "../x.yaml",
415            "a/b.yaml",
416            "..\\x.yaml",
417            ".hidden.yaml",
418            "x.yml",
419            "x.sh",
420        ] {
421            assert!(!safe_file_name(bad), "{bad}");
422        }
423    }
424
425    /// A catalog server in memory: path → (status, body).
426    struct Served(Mutex<Vec<(String, u16, String)>>);
427
428    #[async_trait::async_trait]
429    impl HttpClient for Served {
430        async fn send(&self, request: Request) -> std::result::Result<Response, ScopeError> {
431            let routes = self.0.lock().unwrap();
432            let (_, status, body) = routes
433                .iter()
434                .find(|(path, _, _)| request.url.ends_with(path.as_str()))
435                .cloned()
436                .unwrap_or((String::new(), 404, String::new()));
437            Ok(Response {
438                status_code: status,
439                headers: Default::default(),
440                body,
441            })
442        }
443    }
444
445    fn venue(id: &str, name: &str) -> String {
446        format!(
447            "id: {id}\nname: {name}\nbase_url: https://{id}.example\nsymbol:\n  template: \"{{base}}{{quote}}\"\n  default_quote: USDT\ncapabilities:\n  ticker:\n    path: /t\n    response:\n      last_price: p\n"
448        )
449    }
450
451    fn served(venues: &[(&str, &str, &str)]) -> Served {
452        let index = Index {
453            schema: SCHEMA,
454            venues: venues
455                .iter()
456                .map(|(id, file, body)| Entry {
457                    id: id.to_string(),
458                    file: file.to_string(),
459                    sha256: checksum(body),
460                })
461                .collect(),
462        };
463        let mut routes = vec![(
464            format!("/{INDEX_FILE}"),
465            200,
466            serde_json::to_string(&index).unwrap(),
467        )];
468        routes.extend(
469            venues
470                .iter()
471                .map(|(_, file, body)| (format!("/{file}"), 200, body.to_string())),
472        );
473        Served(Mutex::new(routes))
474    }
475
476    #[tokio::test]
477    async fn test_a_sync_installs_reports_and_replaces_the_catalog() {
478        let home = tempfile::tempdir().unwrap();
479        let dest = home.path().join("venues").join("catalog");
480        let a = venue("alpha", "Alpha");
481        let b = venue("beta", "Beta");
482        let report = sync(
483            &served(&[("alpha", "alpha.yaml", &a), ("beta", "beta.yaml", &b)]),
484            "https://x/c",
485            &dest,
486        )
487        .await
488        .unwrap();
489        assert_eq!(report.added, ["alpha", "beta"]);
490        assert_eq!(
491            report.summary(),
492            "2 venues: 2 added (alpha, beta), 0 updated, 0 removed"
493        );
494        let registry =
495            crate::market::VenueRegistry::load_dirs(&dest, &home.path().join("u")).unwrap();
496        assert_eq!(registry.list(), ["alpha", "beta"]);
497
498        // Beta changes, alpha goes, gamma comes.
499        let b2 = venue("beta", "Beta 2");
500        let c = venue("gamma", "Gamma");
501        let report = sync(
502            &served(&[("beta", "beta.yaml", &b2), ("gamma", "gamma.yaml", &c)]),
503            "https://x/c/",
504            &dest,
505        )
506        .await
507        .unwrap();
508        assert_eq!(report.added, ["gamma"]);
509        assert_eq!(report.updated, ["beta"]);
510        assert_eq!(report.removed, ["alpha"]);
511        assert!(!dest.join("alpha.yaml").exists());
512        assert!(!home.path().join("venues").join("catalog.new").exists());
513        assert!(!home.path().join("venues").join("catalog.old").exists());
514    }
515
516    #[tokio::test]
517    async fn test_a_bad_file_leaves_the_old_catalog() {
518        let home = tempfile::tempdir().unwrap();
519        let dest = home.path().join("catalog");
520        let a = venue("alpha", "Alpha");
521        sync(&served(&[("alpha", "alpha.yaml", &a)]), "https://x", &dest)
522            .await
523            .unwrap();
524
525        // The file served is not the one the index hashed.
526        let tampered = served(&[("alpha", "alpha.yaml", &a)]);
527        tampered.0.lock().unwrap()[1].2 = venue("alpha", "Evil");
528        let err = sync(&tampered, "https://x", &dest)
529            .await
530            .unwrap_err()
531            .to_string();
532        assert!(err.contains("does not match the catalog index"), "{err}");
533        // An id that is not the index's.
534        let err = sync(&served(&[("beta", "alpha.yaml", &a)]), "https://x", &dest)
535            .await
536            .unwrap_err()
537            .to_string();
538        assert!(err.contains("the catalog index says"), "{err}");
539        // A path out of the folder.
540        let err = sync(
541            &served(&[("alpha", "../alpha.yaml", &a)]),
542            "https://x",
543            &dest,
544        )
545        .await
546        .unwrap_err()
547        .to_string();
548        assert!(err.contains("unsafe file"), "{err}");
549        // A newer schema.
550        let newer = served(&[]);
551        newer.0.lock().unwrap()[0].2 = r#"{"schema": 99, "venues": []}"#.to_string();
552        let err = sync(&newer, "https://x", &dest)
553            .await
554            .unwrap_err()
555            .to_string();
556        assert!(err.contains("needs a newer scope"), "{err}");
557        // Offline: the index cannot be fetched.
558        let err = sync(&Served(Mutex::new(vec![])), "https://x", &dest)
559            .await
560            .unwrap_err()
561            .to_string();
562        assert!(err.contains("HTTP 404"), "{err}");
563
564        // Through all of it, the first catalog stays.
565        assert_eq!(installed(&dest).len(), 1);
566        assert_eq!(std::fs::read_to_string(dest.join("alpha.yaml")).unwrap(), a);
567    }
568}
scope/market/catalog.rs

scope/market/
catalog.rs
