scope_cli/cli/watch/
run.rs

1//! A watchlist run: ticks on the watchlist's schedule, through the one
2//! repeat runner, with the CSV trail and the final report.
3
4use super::fetch::Tick;
5use super::{Ready, Sources, tick};
6use crate::cli::output::Output;
7use crate::cli::repeat::{self, RunOutput, Schedule};
8use scope::error::{Result, ScopeError};
9use scope::metrics::Sample;
10use scope::watch::{Event, History, Snapshot, Watchlist, diff};
11use std::io::Write as _;
12use std::path::Path;
13
14/// The CSV header of a watchlist run (one row per cell per tick).
15pub fn csv_header() -> &'static str {
16    "ts,target,metric,value,op,threshold,status,source,error,detail\n"
17}
18
19/// The CSV rows of one tick: every cell of every target. `source` is
20/// `preset` for a preset check and the metric's group for a rule.
21pub fn csv_rows(at: &str, snapshot: &Snapshot) -> String {
22    let mut rows = String::new();
23    for t in &snapshot.targets {
24        for c in &t.cells {
25            let (metric, op, threshold, source) = match &c.rule {
26                Some(r) => (
27                    r.metric.name().to_string(),
28                    r.op.symbol().to_string(),
29                    r.value.to_string(),
30                    format!("{:?}", r.metric.group()).to_lowercase(),
31                ),
32                None => (
33                    c.label.clone(),
34                    String::new(),
35                    String::new(),
36                    "preset".into(),
37                ),
38            };
39            let value = c.value.map(|v| v.to_string()).unwrap_or_default();
40            let error = match &c.status {
41                scope::watch::RuleStatus::Error(e) => e.as_str(),
42                _ => "",
43            };
44            let fields = [
45                at,
46                &t.target,
47                &metric,
48                &value,
49                &op,
50                &threshold,
51                c.status.name(),
52                &source,
53                error,
54                // A preset check's text: it keeps the numbers of a check
55                // that passed, too.
56                c.detail.as_deref().unwrap_or(""),
57            ];
58            let cells: Vec<String> = fields.iter().map(|f| csv_field(f)).collect();
59            rows.push_str(&cells.join(","));
60            rows.push('\n');
61        }
62    }
63    rows
64}
65
66/// A CSV field, quoted when it holds a comma, a quote or a line break.
67fn csv_field(s: &str) -> String {
68    if s.contains([',', '"', '\n', '\r']) {
69        format!("\"{}\"", s.replace('"', "\"\""))
70    } else {
71        s.to_string()
72    }
73}
74
75/// The report of a run: one `market summary --report` section per row
76/// with a summary (one per venue), then every cell's last value.
77pub fn report_text(w: &Watchlist, last: &Tick) -> String {
78    let mut md = format!("# Watchlist Report: {}\n\n", w.name);
79    // The tick's rows, and so its summaries, follow each target's rows.
80    let rows = w
81        .targets
82        .iter()
83        .flat_map(|t| t.rows().into_iter().map(move |venue| (t, venue)));
84    for ((t, venue), summary) in rows.zip(&last.summaries) {
85        if let (Some(summary), Some(venue)) = (summary, venue) {
86            md.push_str(&crate::cli::market::market_summary_to_markdown(
87                summary, venue, &t.id,
88            ));
89            md.push('\n');
90        }
91    }
92    md.push_str(
93        "## Last values\n\n| Target | Check | Value | Status | Detail |\n|---|---|---|---|---|\n",
94    );
95    for t in &last.snapshot.targets {
96        for c in &t.cells {
97            let value = c.value.map(|v| v.to_string()).unwrap_or_else(|| "-".into());
98            let detail = c.status.message().unwrap_or("").replace('|', "\\|");
99            md.push_str(&format!(
100                "| {} | {} | {} | {} | {} |\n",
101                t.target,
102                c.label.replace('|', "\\|"),
103                value,
104                c.status.name(),
105                detail
106            ));
107        }
108    }
109    md
110}
111
112/// Writes the report of `last` to `path`.
113pub fn write_report(path: &Path, w: &Watchlist, last: &Tick) -> Result<()> {
114    std::fs::write(path, report_text(w, last))
115        .map_err(|e| ScopeError::io_context(e, format!("cannot write {}", path.display())))
116}
117
118/// The run's schedule; a run with no duration goes until cancelled.
119pub fn schedule(w: &Watchlist) -> Result<Schedule> {
120    match w.duration {
121        Some(d) => Schedule::new(w.interval, d),
122        None => Schedule::until_cancelled(w.interval),
123    }
124}
125
126/// Runs `ready` on its schedule. After each tick, `on_tick` gets the run
127/// number, the tick and the events since the tick before. The CSV gets one
128/// row per cell per tick; at the end of the duration, the report is
129/// written. A CSV or report write error ends the run and is returned.
130///
131/// Returns the last tick.
132pub async fn run(
133    ready: &Ready,
134    src: Sources<'_>,
135    out: &Output,
136    mut on_tick: impl FnMut(u64, &Tick, &[Event]),
137) -> Result<Tick> {
138    let w = &ready.list;
139    if let Some(path) = &w.outputs.csv {
140        std::fs::write(path, csv_header())
141            .map_err(|e| ScopeError::io_context(e, format!("cannot write {}", path.display())))?;
142    }
143    // The step closure is called once per tick and its future outlives
144    // the call, so the state it changes is shared through cells.
145    let prev = std::cell::RefCell::new(Snapshot::default());
146    let history = std::cell::RefCell::new(History::default());
147    let on_tick = std::cell::RefCell::new(&mut on_tick);
148    let (prev, history, on_tick) = (&prev, &history, &on_tick);
149    let last = repeat::run(&schedule(w)?, &[], None, out, |n| async move {
150        // Taken out for the tick, so no borrow is held across its awaits.
151        let mut memory = history.take();
152        let t = tick(ready, src, out, &mut memory).await;
153        history.replace(memory);
154        let events = diff(&prev.borrow(), &t.snapshot);
155        if let Some(path) = &w.outputs.csv {
156            let at = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();
157            append(path, &csv_rows(&at, &t.snapshot))?;
158        }
159        (on_tick.borrow_mut())(n, &t, &events);
160        *prev.borrow_mut() = t.snapshot.clone();
161        let failed: Vec<String> = events
162            .iter()
163            .filter(|e| matches!(e, Event::BreachStarted { .. }))
164            .map(Event::banner)
165            .collect();
166        Ok(RunOutput {
167            sample: Sample {
168                values: Vec::new(),
169                healthy: t.snapshot.breaches() == 0,
170                failed_checks: failed,
171            },
172            snapshot: String::new(),
173            view: None,
174            value: t,
175        })
176    })
177    .await?
178    .unwrap_or_default();
179    if let Some(path) = &w.outputs.report {
180        write_report(path, w, &last)?;
181        out.note_saved("Report", path.display());
182    }
183    Ok(last)
184}
185
186fn append(path: &Path, rows: &str) -> Result<()> {
187    let context = || format!("cannot append to {}", path.display());
188    let mut file = std::fs::OpenOptions::new()
189        .append(true)
190        .open(path)
191        .map_err(|e| ScopeError::io_context(e, context()))?;
192    file.write_all(rows.as_bytes())
193        .map_err(|e| ScopeError::io_context(e, context()))
194}
195
196#[cfg(test)]
197mod tests {
198    use super::*;
199    use scope::chains::mocks::MockClientFactory;
200    use scope::watch::{Cell, Rule, RuleStatus, Target, TargetKind, TargetSnapshot};
201    use std::time::Duration;
202
203    #[test]
204    fn csv_rows_hold_every_cell_and_quote_commas() {
205        let snapshot = Snapshot {
206            targets: vec![TargetSnapshot::new(
207                "USDC@binance",
208                vec![
209                    Cell {
210                        label: "Spread".into(),
211                        rule: None,
212                        value: Some(3.4),
213                        status: RuleStatus::Fail("Spread: 3.4%, max 3%".into()),
214                        detail: Some("Spread: 3.4%, max 3%".into()),
215                    },
216                    // A passing check keeps its value and text (the 0.8.0
217                    // smoke test found them empty).
218                    Cell {
219                        label: "Bid depth".into(),
220                        rule: None,
221                        value: Some(4200.0),
222                        status: RuleStatus::Pass,
223                        detail: Some("Bid depth: 4200 USDT (min 3000)".into()),
224                    },
225                    Cell {
226                        label: "holders >= 10".into(),
227                        rule: Some(Rule::parse("holders >= 10").unwrap()),
228                        value: None,
229                        status: RuleStatus::Error("timeout, retry".into()),
230                        detail: None,
231                    },
232                ],
233            )],
234        };
235        let rows = csv_rows("2026-10-16T12:00:00Z", &snapshot);
236        let lines: Vec<&str> = rows.lines().collect();
237        assert_eq!(lines.len(), 3);
238        assert_eq!(
239            lines[0],
240            "2026-10-16T12:00:00Z,USDC@binance,Spread,3.4,,,fail,preset,,\"Spread: 3.4%, max 3%\""
241        );
242        assert_eq!(
243            lines[1],
244            "2026-10-16T12:00:00Z,USDC@binance,Bid depth,4200,,,pass,preset,,Bid depth: 4200 USDT (min 3000)"
245        );
246        assert_eq!(
247            lines[2],
248            "2026-10-16T12:00:00Z,USDC@binance,holders,,>=,10,error,token,\"timeout, retry\","
249        );
250        assert_eq!(csv_header().split(',').count(), 10);
251    }
252
253    #[tokio::test]
254    async fn a_run_ticks_writes_the_csv_and_the_report() {
255        let dir = tempfile::tempdir().unwrap();
256        let csv = dir.path().join("w.csv");
257        let report = dir.path().join("w.md");
258        let mut t = Target::new(
259            TargetKind::Token,
260            "0x1234567890123456789012345678901234567890",
261        );
262        t.rules = vec![Rule::parse("price_usd > 0").unwrap()];
263        let mut w = Watchlist::new("test");
264        // Room for several ticks even when a tick is slow (a coverage build
265        // runs instrumented): 15 ms left room for one, and the test needs
266        // two.
267        w.interval = Duration::from_millis(50);
268        w.duration = Some(Duration::from_millis(500));
269        w.outputs.csv = Some(csv.clone());
270        w.outputs.report = Some(report.clone());
271        w.targets = vec![t];
272        let ready = Ready {
273            list: w,
274            contracts: vec![None],
275        };
276        let factory = MockClientFactory::new();
277        let config = scope::config::Config::default();
278        let src = Sources {
279            config: &config,
280            clients: &factory,
281        };
282        let mut ticks = Vec::new();
283        let (out, _cap) = Output::capture();
284        let last = run(&ready, src, &out, |n, _, events| {
285            ticks.push((n, events.len()))
286        })
287        .await
288        .unwrap();
289        // The tick count depends on timing; the runs are numbered from 1.
290        let runs: Vec<u64> = ticks.iter().map(|t| t.0).collect();
291        assert!(runs.len() >= 2, "{runs:?}");
292        assert_eq!(runs, (1..=runs.len() as u64).collect::<Vec<_>>());
293        assert!(ticks.iter().all(|t| t.1 == 0), "a pass starts no event");
294        assert_eq!(last.snapshot.targets.len(), 1);
295        let text = std::fs::read_to_string(&csv).unwrap();
296        assert_eq!(
297            text.lines().count(),
298            1 + runs.len(),
299            "header and one row per tick"
300        );
301        let md = std::fs::read_to_string(&report).unwrap();
302        assert!(md.contains("# Watchlist Report: test"), "{md}");
303        assert!(md.contains("| price_usd > 0 |"), "{md}");
304    }
305
306    #[test]
307    fn the_report_has_a_market_section_per_target_with_a_summary_and_a_venue() {
308        use scope::market::{BookSource, HealthThresholds, MarketSummary, OrderBook};
309        let level = |price, quantity| scope::market::OrderBookLevel { price, quantity };
310        let book = OrderBook {
311            pair: "USDC/USDT".into(),
312            source: BookSource::Exchange,
313            bids: vec![level(0.999, 5_000.0)],
314            asks: vec![level(1.001, 5_000.0)],
315        };
316        let summary = MarketSummary::from_order_book(&book, &HealthThresholds::default(), None);
317        let mut on_venue = Target::new(TargetKind::Pair, "USDC");
318        on_venue.venues = vec!["binance".into()];
319        // A summary with no venue to name has no section.
320        let no_venue = Target::new(TargetKind::Pair, "DAI");
321        let mut w = Watchlist::new("pegs");
322        w.targets = vec![on_venue, no_venue];
323        let last = Tick {
324            snapshot: Snapshot {
325                targets: vec![TargetSnapshot::new(
326                    "USDC@binance",
327                    vec![Cell {
328                        label: "a|b".into(),
329                        rule: None,
330                        value: None,
331                        status: RuleStatus::Error("x|y".into()),
332                        detail: None,
333                    }],
334                )],
335            },
336            summaries: vec![Some(summary.clone()), Some(summary)],
337        };
338        let md = report_text(&w, &last);
339        assert!(md.starts_with("# Watchlist Report: pegs\n\n"), "{md}");
340        assert_eq!(md.matches("# Market Health Report:").count(), 1, "{md}");
341        let section = md.find("# Market Health Report: USDC").unwrap();
342        assert!(section < md.find("## Last values").unwrap(), "{md}");
343        assert!(!md.contains("Market Health Report: DAI"), "{md}");
344        // A pipe in a label or an error would split the table cell.
345        assert!(
346            md.contains("| USDC@binance | a\\|b | - | error | x\\|y |"),
347            "{md}"
348        );
349    }
350
351    // Design 2026-11-04, section 2.6: a target on two venues gives a
352    // market section and a set of CSV rows per venue.
353    #[test]
354    fn two_venues_give_two_sections_and_two_labelled_row_sets() {
355        use scope::market::{BookSource, HealthThresholds, MarketSummary, OrderBook};
356        let level = |price, quantity| scope::market::OrderBookLevel { price, quantity };
357        let book = OrderBook {
358            pair: "USDN/USDT".into(),
359            source: BookSource::Exchange,
360            bids: vec![level(0.999, 5_000.0)],
361            asks: vec![level(1.001, 5_000.0)],
362        };
363        let summary = MarketSummary::from_order_book(&book, &HealthThresholds::default(), None);
364        let mut t = Target::new(TargetKind::Pair, "USDN");
365        t.venues = vec!["weex".into(), "binance".into()];
366        t.rules = vec![scope::watch::Rule::parse("spread_pct < 1").unwrap()];
367        let mut w = Watchlist::new("pegs");
368        w.targets = vec![t];
369        let row = |venue: &str| {
370            TargetSnapshot::new(
371                format!("USDN@{venue}"),
372                vec![Cell {
373                    label: "spread_pct < 1".into(),
374                    rule: scope::watch::Rule::parse("spread_pct < 1").ok(),
375                    value: Some(0.2),
376                    status: RuleStatus::Pass,
377                    detail: None,
378                }],
379            )
380            .on_venue(Some(venue))
381        };
382        let last = Tick {
383            snapshot: Snapshot {
384                targets: vec![row("weex"), row("binance")],
385            },
386            summaries: vec![Some(summary.clone()), Some(summary)],
387        };
388        let md = report_text(&w, &last);
389        assert_eq!(
390            md.matches("# Market Health Report: USDN").count(),
391            2,
392            "{md}"
393        );
394        assert!(
395            md.contains("| USDN@weex |") && md.contains("| USDN@binance |"),
396            "{md}"
397        );
398        let csv = csv_rows("t0", &last.snapshot);
399        let targets: Vec<&str> = csv.lines().map(|l| l.split(',').nth(1).unwrap()).collect();
400        assert_eq!(targets, ["USDN@weex", "USDN@binance"]);
401    }
402
403    #[tokio::test]
404    async fn a_report_that_cannot_be_written_is_an_error() {
405        let dir = tempfile::tempdir().unwrap();
406        let mut w = Watchlist::new("test");
407        w.interval = Duration::from_millis(5);
408        w.duration = Some(Duration::from_millis(1));
409        // A folder cannot be written as a file.
410        w.outputs.report = Some(dir.path().to_path_buf());
411        let ready = Ready {
412            list: w,
413            contracts: vec![],
414        };
415        let factory = MockClientFactory::new();
416        let config = scope::config::Config::default();
417        let src = Sources {
418            config: &config,
419            clients: &factory,
420        };
421        let (out, _cap) = Output::capture();
422        let err = run(&ready, src, &out, |_, _, _| {}).await.unwrap_err();
423        assert!(err.to_string().contains("cannot write"), "{err}");
424    }
425}
scope_cli/cli/watch/run.rs

scope_cli/cli/watch/
run.rs
