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, 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 target
76/// with a summary, 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    for (t, summary) in w.targets.iter().zip(&last.summaries) {
80        if let (Some(summary), Some(venue)) = (summary, &t.venue) {
81            md.push_str(&crate::cli::market::market_summary_to_markdown(
82                summary, venue, &t.id,
83            ));
84            md.push('\n');
85        }
86    }
87    md.push_str(
88        "## Last values\n\n| Target | Check | Value | Status | Detail |\n|---|---|---|---|---|\n",
89    );
90    for t in &last.snapshot.targets {
91        for c in &t.cells {
92            let value = c.value.map(|v| v.to_string()).unwrap_or_else(|| "-".into());
93            let detail = c.status.message().unwrap_or("").replace('|', "\\|");
94            md.push_str(&format!(
95                "| {} | {} | {} | {} | {} |\n",
96                t.target,
97                c.label.replace('|', "\\|"),
98                value,
99                c.status.name(),
100                detail
101            ));
102        }
103    }
104    md
105}
106
107/// Writes the report of `last` to `path`.
108pub fn write_report(path: &Path, w: &Watchlist, last: &Tick) -> Result<()> {
109    std::fs::write(path, report_text(w, last))
110        .map_err(|e| ScopeError::io_context(e, format!("cannot write {}", path.display())))
111}
112
113/// The run's schedule; a run with no duration goes until cancelled.
114pub fn schedule(w: &Watchlist) -> Result<Schedule> {
115    match w.duration {
116        Some(d) => Schedule::new(w.interval, d),
117        None => Schedule::until_cancelled(w.interval),
118    }
119}
120
121/// Runs `ready` on its schedule. After each tick, `on_tick` gets the run
122/// number, the tick and the events since the tick before. The CSV gets one
123/// row per cell per tick; at the end of the duration, the report is
124/// written. A CSV or report write error ends the run and is returned.
125///
126/// Returns the last tick.
127pub async fn run(
128    ready: &Ready,
129    src: Sources<'_>,
130    out: &Output,
131    mut on_tick: impl FnMut(u64, &Tick, &[Event]),
132) -> Result<Tick> {
133    let w = &ready.list;
134    if let Some(path) = &w.outputs.csv {
135        std::fs::write(path, csv_header())
136            .map_err(|e| ScopeError::io_context(e, format!("cannot write {}", path.display())))?;
137    }
138    // The step closure is called once per tick and its future outlives
139    // the call, so the state it changes is shared through cells.
140    let prev = std::cell::RefCell::new(Snapshot::default());
141    let on_tick = std::cell::RefCell::new(&mut on_tick);
142    let (prev, on_tick) = (&prev, &on_tick);
143    let last = repeat::run(&schedule(w)?, &[], None, out, |n| async move {
144        let t = tick(ready, src, out).await;
145        let events = diff(&prev.borrow(), &t.snapshot);
146        if let Some(path) = &w.outputs.csv {
147            let at = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();
148            append(path, &csv_rows(&at, &t.snapshot))?;
149        }
150        (on_tick.borrow_mut())(n, &t, &events);
151        *prev.borrow_mut() = t.snapshot.clone();
152        let failed: Vec<String> = events
153            .iter()
154            .filter(|e| matches!(e, Event::BreachStarted { .. }))
155            .map(Event::banner)
156            .collect();
157        Ok(RunOutput {
158            sample: Sample {
159                values: Vec::new(),
160                healthy: t.snapshot.breaches() == 0,
161                failed_checks: failed,
162            },
163            snapshot: String::new(),
164            view: None,
165            value: t,
166        })
167    })
168    .await?
169    .unwrap_or_default();
170    if let Some(path) = &w.outputs.report {
171        write_report(path, w, &last)?;
172        out.note_saved("Report", path.display());
173    }
174    Ok(last)
175}
176
177fn append(path: &Path, rows: &str) -> Result<()> {
178    let context = || format!("cannot append to {}", path.display());
179    let mut file = std::fs::OpenOptions::new()
180        .append(true)
181        .open(path)
182        .map_err(|e| ScopeError::io_context(e, context()))?;
183    file.write_all(rows.as_bytes())
184        .map_err(|e| ScopeError::io_context(e, context()))
185}
186
187#[cfg(test)]
188mod tests {
189    use super::*;
190    use scope::chains::mocks::MockClientFactory;
191    use scope::watch::{Cell, Rule, RuleStatus, Target, TargetKind, TargetSnapshot};
192    use std::time::Duration;
193
194    #[test]
195    fn csv_rows_hold_every_cell_and_quote_commas() {
196        let snapshot = Snapshot {
197            targets: vec![TargetSnapshot::new(
198                "USDC@binance",
199                vec![
200                    Cell {
201                        label: "Spread".into(),
202                        rule: None,
203                        value: Some(3.4),
204                        status: RuleStatus::Fail("Spread: 3.4%, max 3%".into()),
205                        detail: Some("Spread: 3.4%, max 3%".into()),
206                    },
207                    // A passing check keeps its value and text (the 0.8.0
208                    // smoke test found them empty).
209                    Cell {
210                        label: "Bid depth".into(),
211                        rule: None,
212                        value: Some(4200.0),
213                        status: RuleStatus::Pass,
214                        detail: Some("Bid depth: 4200 USDT (min 3000)".into()),
215                    },
216                    Cell {
217                        label: "holders >= 10".into(),
218                        rule: Some(Rule::parse("holders >= 10").unwrap()),
219                        value: None,
220                        status: RuleStatus::Error("timeout, retry".into()),
221                        detail: None,
222                    },
223                ],
224            )],
225        };
226        let rows = csv_rows("2026-10-16T12:00:00Z", &snapshot);
227        let lines: Vec<&str> = rows.lines().collect();
228        assert_eq!(lines.len(), 3);
229        assert_eq!(
230            lines[0],
231            "2026-10-16T12:00:00Z,USDC@binance,Spread,3.4,,,fail,preset,,\"Spread: 3.4%, max 3%\""
232        );
233        assert_eq!(
234            lines[1],
235            "2026-10-16T12:00:00Z,USDC@binance,Bid depth,4200,,,pass,preset,,Bid depth: 4200 USDT (min 3000)"
236        );
237        assert_eq!(
238            lines[2],
239            "2026-10-16T12:00:00Z,USDC@binance,holders,,>=,10,error,token,\"timeout, retry\","
240        );
241        assert_eq!(csv_header().split(',').count(), 10);
242    }
243
244    #[tokio::test]
245    async fn a_run_ticks_writes_the_csv_and_the_report() {
246        let dir = tempfile::tempdir().unwrap();
247        let csv = dir.path().join("w.csv");
248        let report = dir.path().join("w.md");
249        let mut t = Target::new(
250            TargetKind::Token,
251            "0x1234567890123456789012345678901234567890",
252        );
253        t.rules = vec![Rule::parse("price_usd > 0").unwrap()];
254        let mut w = Watchlist::new("test");
255        // Room for several ticks even when a tick is slow (a coverage build
256        // runs instrumented): 15 ms left room for one, and the test needs
257        // two.
258        w.interval = Duration::from_millis(50);
259        w.duration = Some(Duration::from_millis(500));
260        w.outputs.csv = Some(csv.clone());
261        w.outputs.report = Some(report.clone());
262        w.targets = vec![t];
263        let ready = Ready {
264            list: w,
265            contracts: vec![None],
266        };
267        let factory = MockClientFactory::new();
268        let config = scope::config::Config::default();
269        let src = Sources {
270            config: &config,
271            clients: &factory,
272        };
273        let mut ticks = Vec::new();
274        let (out, _cap) = Output::capture();
275        let last = run(&ready, src, &out, |n, _, events| {
276            ticks.push((n, events.len()))
277        })
278        .await
279        .unwrap();
280        // The tick count depends on timing; the runs are numbered from 1.
281        let runs: Vec<u64> = ticks.iter().map(|t| t.0).collect();
282        assert!(runs.len() >= 2, "{runs:?}");
283        assert_eq!(runs, (1..=runs.len() as u64).collect::<Vec<_>>());
284        assert!(ticks.iter().all(|t| t.1 == 0), "a pass starts no event");
285        assert_eq!(last.snapshot.targets.len(), 1);
286        let text = std::fs::read_to_string(&csv).unwrap();
287        assert_eq!(
288            text.lines().count(),
289            1 + runs.len(),
290            "header and one row per tick"
291        );
292        let md = std::fs::read_to_string(&report).unwrap();
293        assert!(md.contains("# Watchlist Report: test"), "{md}");
294        assert!(md.contains("| price_usd > 0 |"), "{md}");
295    }
296
297    #[test]
298    fn the_report_has_a_market_section_per_target_with_a_summary_and_a_venue() {
299        use scope::market::{BookSource, HealthThresholds, MarketSummary, OrderBook};
300        let level = |price, quantity| scope::market::OrderBookLevel { price, quantity };
301        let book = OrderBook {
302            pair: "USDC/USDT".into(),
303            source: BookSource::Exchange,
304            bids: vec![level(0.999, 5_000.0)],
305            asks: vec![level(1.001, 5_000.0)],
306        };
307        let summary = MarketSummary::from_order_book(&book, &HealthThresholds::default(), None);
308        let mut on_venue = Target::new(TargetKind::Pair, "USDC");
309        on_venue.venue = Some("binance".into());
310        // A summary with no venue to name has no section.
311        let no_venue = Target::new(TargetKind::Pair, "DAI");
312        let mut w = Watchlist::new("pegs");
313        w.targets = vec![on_venue, no_venue];
314        let last = Tick {
315            snapshot: Snapshot {
316                targets: vec![TargetSnapshot::new(
317                    "USDC@binance",
318                    vec![Cell {
319                        label: "a|b".into(),
320                        rule: None,
321                        value: None,
322                        status: RuleStatus::Error("x|y".into()),
323                        detail: None,
324                    }],
325                )],
326            },
327            summaries: vec![Some(summary.clone()), Some(summary)],
328        };
329        let md = report_text(&w, &last);
330        assert!(md.starts_with("# Watchlist Report: pegs\n\n"), "{md}");
331        assert_eq!(md.matches("# Market Health Report:").count(), 1, "{md}");
332        let section = md.find("# Market Health Report: USDC").unwrap();
333        assert!(section < md.find("## Last values").unwrap(), "{md}");
334        assert!(!md.contains("Market Health Report: DAI"), "{md}");
335        // A pipe in a label or an error would split the table cell.
336        assert!(
337            md.contains("| USDC@binance | a\\|b | - | error | x\\|y |"),
338            "{md}"
339        );
340    }
341
342    #[tokio::test]
343    async fn a_report_that_cannot_be_written_is_an_error() {
344        let dir = tempfile::tempdir().unwrap();
345        let mut w = Watchlist::new("test");
346        w.interval = Duration::from_millis(5);
347        w.duration = Some(Duration::from_millis(1));
348        // A folder cannot be written as a file.
349        w.outputs.report = Some(dir.path().to_path_buf());
350        let ready = Ready {
351            list: w,
352            contracts: vec![],
353        };
354        let factory = MockClientFactory::new();
355        let config = scope::config::Config::default();
356        let src = Sources {
357            config: &config,
358            clients: &factory,
359        };
360        let (out, _cap) = Output::capture();
361        let err = run(&ready, src, &out, |_, _, _| {}).await.unwrap_err();
362        assert!(err.to_string().contains("cannot write"), "{err}");
363    }
364}
scope_cli/cli/watch/run.rs

scope_cli/cli/watch/
run.rs
