1use super::{Ready, Sources};
9use crate::cli::address::{AddressArgs, analyze_address};
10use crate::cli::crawl::{self, Period};
11use crate::cli::market::CrawlDexSource;
12use crate::cli::output::Output;
13use scope::compliance::datasource::BlockchainDataClient;
14use scope::compliance::risk::RiskEngine;
15use scope::market::{
16 HealthThresholds, MarketSummary, MarketVenue, Pair, Trade, fetch_summary_with_trades,
17};
18use scope::metrics::{MetricGroup, MetricId, describe, sample};
19use scope::watch::{
20 Cell, History, Preset, Reading, Readings, RuleStatus, Snapshot, Target, TargetSnapshot, preset,
21};
22
23pub const PRESET_ERROR_LABEL: &str = "market health";
25
26#[derive(Debug, Clone, Default)]
28pub struct Tick {
29 pub snapshot: Snapshot,
30 pub summaries: Vec<Option<MarketSummary>>,
33}
34
35#[derive(Debug, Clone)]
37pub struct TargetReading {
38 pub row: TargetSnapshot,
39 pub summary: Option<MarketSummary>,
40}
41
42#[derive(Debug, Clone, Copy)]
45pub struct Clock {
46 pub now_ms: u64,
47 pub window_ms: u64,
48}
49
50impl Clock {
51 pub fn now(ready: &Ready) -> Self {
53 let now = chrono::Utc::now().timestamp_millis();
54 Self {
55 now_ms: u64::try_from(now).unwrap_or(0),
56 window_ms: u64::try_from(ready.list.interval.as_millis()).unwrap_or(u64::MAX),
57 }
58 }
59}
60
61pub async fn tick(ready: &Ready, src: Sources<'_>, out: &Output, history: &mut History) -> Tick {
64 let mut result = Tick::default();
65 let clock = Clock::now(ready);
66 for (t, contract) in ready.list.targets.iter().zip(&ready.contracts) {
67 for reading in read_target(t, contract.as_deref(), src, out, history, clock).await {
68 result.snapshot.targets.push(reading.row);
69 result.summaries.push(reading.summary);
70 }
71 }
72 result
73}
74
75type SourceResult = Result<Vec<(MetricId, Reading)>, String>;
77
78fn readings(values: Vec<(MetricId, Option<f64>)>) -> Vec<(MetricId, Reading)> {
80 values.into_iter().map(|(id, v)| (id, Ok(v))).collect()
81}
82
83pub async fn read_target(
87 t: &Target,
88 contract: Option<&str>,
89 src: Sources<'_>,
90 out: &Output,
91 history: &mut History,
92 clock: Clock,
93) -> Vec<TargetReading> {
94 let thresholds = preset::thresholds(t, &src.config.market.health);
95 let token = if t.needs(MetricGroup::Token) {
96 Some(read_token(t, contract, src, out).await)
97 } else {
98 None
99 };
100 let address = if t.needs(MetricGroup::Address) {
101 Some(read_address(t, src, out).await)
102 } else {
103 None
104 };
105 let rows = t.rows();
106 let one_row = rows.len() == 1;
107 let mut result = Vec::with_capacity(rows.len());
108 for venue in rows {
109 let label = t.row_label(venue);
110 let mut cells = Vec::new();
111 let mut summary = None;
112 let mut market: Option<SourceResult> = None;
113 if let Some(venue) = venue
114 && t.needs_market()
115 {
116 match read_market(t, venue, contract, &thresholds, src, out).await {
117 Ok((s, trades)) => {
118 if t.preset == Some(Preset::MarketHealth) {
119 cells.extend(preset::cells(&s));
120 }
121 let since = t.rules.iter().any(|r| r.metric.since_last_tick()).then(|| {
122 let tape = trades.as_deref().map_err(String::as_str);
123 history.read(&label, tape, s.volume_24h, clock.now_ms, clock.window_ms)
124 });
125 market = Some(Ok(market_values(t, &s, &thresholds, since.as_ref())));
126 summary = Some(s);
127 }
128 Err(e) => {
129 if t.preset == Some(Preset::MarketHealth) {
130 cells.push(Cell {
131 label: PRESET_ERROR_LABEL.to_string(),
132 rule: None,
133 value: None,
134 status: RuleStatus::Error(e.clone()),
135 detail: None,
136 });
137 }
138 market = Some(Err(e));
139 }
140 }
141 }
142 let rules = t
145 .rules
146 .iter()
147 .filter(|r| one_row || (r.metric.group() == MetricGroup::Market) == venue.is_some());
148 for rule in rules {
149 let source = match rule.metric.group() {
150 MetricGroup::Market => market.as_ref(),
151 MetricGroup::Token => token.as_ref(),
152 MetricGroup::Address => address.as_ref(),
153 };
154 let (value, status) = match source {
155 Some(Ok(values)) => {
156 let reading = values
157 .iter()
158 .find(|(id, _)| *id == rule.metric)
159 .map_or(Ok(None), |(_, r)| r.clone());
160 match reading {
161 Ok(value) => (value, rule.eval(value)),
162 Err(status) => (None, status),
163 }
164 }
165 Some(Err(e)) => (None, RuleStatus::Error(e.clone())),
166 None if rule.metric.group() == MetricGroup::Market => {
167 (None, RuleStatus::Error("no venue".to_string()))
168 }
169 None => (None, RuleStatus::Error("no source".to_string())),
170 };
171 cells.push(Cell {
172 label: rule.to_string(),
173 rule: Some(*rule),
174 value,
175 status,
176 detail: None,
177 });
178 }
179 result.push(TargetReading {
180 row: TargetSnapshot::new(label, cells).on_venue(venue),
181 summary,
182 });
183 }
184 result
185}
186
187async fn read_market(
191 t: &Target,
192 venue_name: &str,
193 contract: Option<&str>,
194 thresholds: &HealthThresholds,
195 src: Sources<'_>,
196 out: &Output,
197) -> Result<(MarketSummary, Result<Vec<Trade>, String>), String> {
198 let venue = MarketVenue::from_name_with_http(venue_name, src.clients.http())
199 .map_err(|e| e.to_string())?;
200 let dex = CrawlDexSource {
201 factory: src.clients,
202 out,
203 token: contract.map(str::to_string),
204 };
205 fetch_summary_with_trades(&venue, &Pair::parse(&t.id), thresholds, &dex)
206 .await
207 .map_err(|e| e.to_string())
208}
209
210fn market_values(
214 t: &Target,
215 summary: &MarketSummary,
216 thresholds: &HealthThresholds,
217 since: Option<&Readings>,
218) -> Vec<(MetricId, Reading)> {
219 let table: Vec<_> = t
220 .rules
221 .iter()
222 .filter(|r| r.metric.group() == MetricGroup::Market)
223 .map(|r| describe(r.metric, thresholds))
224 .collect();
225 let s = sample(summary, &table);
226 table
227 .iter()
228 .zip(s.values)
229 .map(|(m, v)| {
230 let reading = since.and_then(|r| r.get(m.id)).cloned().unwrap_or(Ok(v));
231 (m.id, reading)
232 })
233 .collect()
234}
235
236async fn read_token(
238 t: &Target,
239 contract: Option<&str>,
240 src: Sources<'_>,
241 out: &Output,
242) -> SourceResult {
243 let chain = t.chain.as_deref().unwrap_or(scope::chains::DEFAULT_CHAIN);
244 let input = contract.unwrap_or(&t.id);
245 let a =
246 crawl::fetch_analytics_for_input(input, chain, Period::Hour24, 10, src.clients, None, out)
247 .await
248 .map_err(|e| e.to_string())?;
249 Ok(readings(vec![
250 (MetricId::PriceUsd, Some(a.price_usd)),
251 (MetricId::PriceChange24hPct, Some(a.price_change_24h)),
252 (MetricId::Volume24h, Some(a.volume_24h)),
253 (MetricId::LiquidityUsd, Some(a.liquidity_usd)),
254 (MetricId::Holders, Some(a.total_holders as f64)),
255 ]))
256}
257
258async fn read_address(t: &Target, src: Sources<'_>, out: &Output) -> SourceResult {
261 let chain = t
262 .chain
263 .clone()
264 .unwrap_or_else(|| scope::chains::DEFAULT_CHAIN.to_string());
265 let needs = |m: MetricId| t.rules.iter().any(|r| r.metric == m);
266 let mut values = Vec::new();
267 if needs(MetricId::NativeBalance) {
268 let args = AddressArgs {
269 address: t.id.clone(),
270 chain: chain.clone(),
271 format: None,
272 include_txs: false,
273 include_tokens: false,
274 limit: 1,
275 report: None,
276 dossier: false,
277 };
278 let client = src
279 .clients
280 .create_chain_client(&chain)
281 .map_err(|e| e.to_string())?;
282 let report = analyze_address(&args, client.as_ref(), out)
283 .await
284 .map_err(|e| e.to_string())?;
285 values.push((
286 MetricId::NativeBalance,
287 leading_number(&report.balance.formatted),
288 ));
289 }
290 if needs(MetricId::RiskScore) {
291 let engine = RiskEngine::with_optional_client(
292 BlockchainDataClient::from_env_opt_with_http(src.clients.http()),
293 );
294 let risk = engine
295 .assess_address(&t.id, &chain)
296 .await
297 .map_err(|e| e.to_string())?;
298 values.push((MetricId::RiskScore, Some(f64::from(risk.overall_score))));
299 }
300 Ok(readings(values))
301}
302
303fn leading_number(text: &str) -> Option<f64> {
305 text.split_whitespace()
306 .next()?
307 .replace(',', "")
308 .parse()
309 .ok()
310}
311
312#[cfg(test)]
313mod tests {
314 use super::*;
315 use scope::chains::mocks::MockClientFactory;
316 use scope::watch::{Rule, TargetKind, TargetStatus, Watchlist};
317
318 fn ready(targets: Vec<Target>) -> Ready {
319 let contracts = vec![None; targets.len()];
320 Ready {
321 list: Watchlist {
322 targets,
323 ..Watchlist::new("t")
324 },
325 contracts,
326 }
327 }
328
329 fn quiet() -> Output {
330 Output::capture().0
331 }
332
333 #[test]
334 fn leading_numbers() {
335 assert_eq!(leading_number("1,234.5 ETH"), Some(1234.5));
336 assert_eq!(leading_number("0.1"), Some(0.1));
337 assert_eq!(leading_number("n/a"), None);
338 }
339
340 #[tokio::test]
341 async fn token_rules_read_the_analytics_once_and_judge_each_rule() {
342 let factory = MockClientFactory::new();
343 let config = scope::config::Config::default();
344 let src = Sources {
345 config: &config,
346 clients: &factory,
347 };
348 let mut t = Target::new(
349 TargetKind::Token,
350 "0x1234567890123456789012345678901234567890",
351 );
352 t.chain = Some("ethereum".into());
353 t.rules = vec![
354 Rule::parse("price_usd > 0").unwrap(),
355 Rule::parse("holders >= 100000000000").unwrap(),
356 ];
357 let tick = tick(&ready(vec![t]), src, &quiet(), &mut History::default()).await;
358 let row = &tick.snapshot.targets[0];
359 assert_eq!(row.cells.len(), 2);
360 assert_eq!(row.cells[0].label, "price_usd > 0");
361 assert!(row.cells[0].value.is_some(), "{row:?}");
362 assert_eq!(row.cells[0].status, RuleStatus::Pass, "{row:?}");
363 assert!(
364 matches!(row.cells[1].status, RuleStatus::Fail(_)),
365 "{row:?}"
366 );
367 assert_eq!(row.status, TargetStatus::Breached);
368 assert!(tick.summaries[0].is_none());
369 }
370
371 async fn summary_checks(
374 venue: &str,
375 flags: &[&str],
376 factory: &MockClientFactory,
377 ) -> (Vec<(String, String)>, bool) {
378 let mut argv = vec!["scope", "market", "summary", "USDC", "--venue", venue];
379 argv.extend(flags);
380 argv.extend(["--format", "json"]);
381 let cli = <crate::cli::Cli as clap::Parser>::try_parse_from(argv).unwrap();
382 let (o, cap) = Output::capture();
383 crate::cli::dispatch::dispatch_with(
384 cli.command,
385 &scope::config::Config::default(),
386 factory,
387 &o,
388 )
389 .await
390 .unwrap();
391 let v: serde_json::Value = serde_json::from_str(&cap.out()).unwrap();
392 let checks = v["checks"]
393 .as_array()
394 .unwrap()
395 .iter()
396 .map(|c| {
397 (
398 c["status"].as_str().unwrap().to_string(),
399 c["message"].as_str().unwrap().to_string(),
400 )
401 })
402 .collect();
403 (checks, v["healthy"].as_bool().unwrap())
404 }
405
406 #[tokio::test]
410 async fn the_market_health_preset_gives_the_checks_of_market_summary() {
411 let factory = usdc_on_eth();
412 let config = scope::config::Config::default();
413 let src = Sources {
414 config: &config,
415 clients: &factory,
416 };
417 for (flags, overrides) in [
418 (vec![], scope::market::HealthOverrides::default()),
419 (
420 vec!["--max-spread-pct", "0.000001", "--min-depth", "1e15"],
421 scope::market::HealthOverrides {
422 max_spread_pct: Some(0.000001),
423 min_depth: Some(1e15),
424 ..Default::default()
425 },
426 ),
427 ] {
428 let (checks, healthy) = summary_checks("eth", &flags, &factory).await;
429 assert_eq!(healthy, flags.is_empty(), "flags {flags:?}: {checks:?}");
431 let mut t = Target::new(TargetKind::Pair, "USDC");
432 t.venues = vec!["eth".into()];
433 t.preset = Some(Preset::MarketHealth);
434 t.health = overrides;
435 let tick = tick(&ready(vec![t]), src, &quiet(), &mut History::default()).await;
436 let row = &tick.snapshot.targets[0];
437 let cells: Vec<(String, String)> = row
438 .cells
439 .iter()
440 .map(|c| (c.status.name().to_string(), c.label.clone()))
441 .collect();
442 let expected: Vec<(String, String)> = checks
443 .iter()
444 .map(|(status, message)| {
445 let name = message.split_once(':').map_or(message.as_str(), |(n, _)| n);
446 (status.clone(), name.trim().to_string())
447 })
448 .collect();
449 assert!(!expected.is_empty());
450 assert_eq!(cells, expected, "flags {flags:?}");
451 for (cell, (_, message)) in row.cells.iter().zip(&checks) {
452 if let Some(m) = cell.status.message() {
453 assert_eq!(m, message, "the same text");
454 }
455 }
456 assert_eq!(
457 row.status == TargetStatus::Breached,
458 !healthy,
459 "flags {flags:?}"
460 );
461 }
462 }
463
464 #[tokio::test]
467 async fn since_last_tick_rules_read_the_history_of_the_run() {
468 let factory = usdc_on_eth();
469 let config = scope::config::Config::default();
470 let src = Sources {
471 config: &config,
472 clients: &factory,
473 };
474 let mut t = Target::new(TargetKind::Pair, "USDC");
475 t.venues = vec!["eth".into()];
476 t.rules = vec![
477 Rule::parse("largest_trade < 50000").unwrap(),
478 Rule::parse("volume_change_pct <= 100").unwrap(),
479 Rule::parse("mid_price > 0").unwrap(),
480 ];
481 let ready = ready(vec![t]);
482 let mut history = History::default();
483 let first = tick(&ready, src, &quiet(), &mut history).await;
484 let row = &first.snapshot.targets[0];
485 assert!(
486 matches!(&row.cells[0].status, RuleStatus::Error(m) if m == "largest_trade: a DEX pool has no trade tape here"),
487 "{row:?}"
488 );
489 assert!(
490 matches!(&row.cells[1].status, RuleStatus::NotApplicable(m) if m.contains("second reading")),
491 "{row:?}"
492 );
493 assert_eq!(row.cells[2].status, RuleStatus::Pass, "{row:?}");
494 let second = tick(&ready, src, &quiet(), &mut history).await;
496 let cell = &second.snapshot.targets[0].cells[1];
497 assert_eq!(cell.value, Some(0.0), "{cell:?}");
498 assert_eq!(cell.status, RuleStatus::Pass);
499 }
500
501 fn usdc_on_eth() -> MockClientFactory {
504 let mut factory = MockClientFactory::new();
505 factory.mock_dex.search_results = vec![scope::chains::TokenSearchResult {
506 address: "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48".to_string(),
507 symbol: "USDC".to_string(),
508 name: "USD Coin".to_string(),
509 chain: "ethereum".to_string(),
510 price_usd: Some(1.0),
511 volume_24h: 1_000_000.0,
512 liquidity_usd: 5_000_000.0,
513 market_cap: None,
514 }];
515 if let Some(td) = factory.mock_dex.token_data.as_mut() {
516 td.address = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48".to_string();
517 td.symbol = "USDC".to_string();
518 td.pairs = vec![scope::chains::DexPair {
519 dex_name: "uniswap".into(),
520 pair_address: "0xpool".into(),
521 base_token: "USDC".into(),
522 quote_token: "USDT".into(),
523 price_usd: 1.0,
524 volume_24h: 1_000_000.0,
525 liquidity_usd: 50_000_000.0,
526 liquidity_base: Some(25_000_000.0),
527 price_change_24h: 0.0,
528 buys_24h: 0,
529 sells_24h: 0,
530 buys_6h: 0,
531 sells_6h: 0,
532 buys_1h: 0,
533 sells_1h: 0,
534 pair_created_at: None,
535 url: None,
536 }];
537 }
538 factory
539 }
540
541 #[tokio::test]
542 async fn address_rules_read_the_native_balance_and_the_risk_score() {
543 let factory = crate::cli::dispatch::tests::OfflineFactory::new();
546 let config = scope::config::Config::default();
547 let src = Sources {
548 config: &config,
549 clients: &factory,
550 };
551 let mut t = Target::new(
552 TargetKind::Address,
553 "0x742d35Cc6634C0532925a3b844Bc9e7595f1b3c2",
554 );
555 t.rules = vec![
556 Rule::parse("native_balance > 0.5").unwrap(),
558 Rule::parse("native_balance < 0.5").unwrap(),
559 Rule::parse("risk_score <= 10").unwrap(),
560 ];
561 let tick = tick(&ready(vec![t]), src, &quiet(), &mut History::default()).await;
562 let row = &tick.snapshot.targets[0];
563 assert_eq!(row.cells.len(), 3, "{row:?}");
564 assert_eq!(row.cells[0].value, Some(1.0), "{row:?}");
565 assert_eq!(row.cells[0].status, RuleStatus::Pass, "{row:?}");
566 assert!(
567 matches!(row.cells[1].status, RuleStatus::Fail(_)),
568 "{row:?}"
569 );
570 let risk = row.cells[2].value.expect("a risk score");
571 assert!((0.0..=10.0).contains(&risk), "{risk}");
572 assert_eq!(row.cells[2].status, RuleStatus::Pass, "{row:?}");
573 assert_eq!(row.status, TargetStatus::Breached);
574 assert!(tick.summaries[0].is_none(), "no market source was read");
575 }
576
577 #[tokio::test]
578 async fn an_address_on_its_own_chain_reads_its_balance() {
579 let factory = crate::cli::dispatch::tests::OfflineFactory::new();
580 let config = scope::config::Config::default();
581 let src = Sources {
582 config: &config,
583 clients: &factory,
584 };
585 let mut t = Target::new(
586 TargetKind::Address,
587 "0x742d35Cc6634C0532925a3b844Bc9e7595f1b3c2",
588 );
589 t.chain = Some("polygon".into());
590 t.rules = vec![Rule::parse("native_balance >= 1").unwrap()];
591 let clock = Clock {
592 now_ms: 0,
593 window_ms: 60_000,
594 };
595 let readings = read_target(&t, None, src, &quiet(), &mut History::default(), clock).await;
596 let [reading] = readings.as_slice() else {
597 panic!("one row: {readings:?}");
598 };
599 assert_eq!(reading.row.cells.len(), 1);
600 assert_eq!(reading.row.cells[0].value, Some(1.0));
601 assert_eq!(reading.row.status, TargetStatus::Passed);
602 }
603
604 #[tokio::test]
605 async fn a_failed_source_errors_its_cells_and_the_next_target_still_runs() {
606 let factory = MockClientFactory::new();
607 let config = scope::config::Config::default();
608 let src = Sources {
609 config: &config,
610 clients: &factory,
611 };
612 let mut bad = Target::new(TargetKind::Pair, "USDC");
613 bad.venues = vec!["no-such-venue".into()];
614 bad.preset = Some(Preset::MarketHealth);
615 bad.rules = vec![Rule::parse("spread_pct < 1").unwrap()];
616 let mut good = Target::new(
617 TargetKind::Token,
618 "0x1234567890123456789012345678901234567890",
619 );
620 good.rules = vec![Rule::parse("price_usd > 0").unwrap()];
621 let tick = tick(
622 &ready(vec![bad, good]),
623 src,
624 &quiet(),
625 &mut History::default(),
626 )
627 .await;
628 let row = &tick.snapshot.targets[0];
629 assert_eq!(row.cells[0].label, PRESET_ERROR_LABEL);
630 let RuleStatus::Error(text) = &row.cells[0].status else {
631 panic!("{row:?}")
632 };
633 assert!(
634 text.contains("no-such-venue"),
635 "the exact error text: {text}"
636 );
637 assert!(matches!(row.cells[1].status, RuleStatus::Error(_)));
638 assert_eq!(row.status, TargetStatus::Unknown, "never passed");
639 assert_eq!(tick.snapshot.targets[1].status, TargetStatus::Passed);
640 }
641
642 #[tokio::test]
647 async fn a_target_on_two_venues_gives_a_row_per_venue_and_an_id_row() {
648 let factory = usdc_on_eth();
649 let config = scope::config::Config::default();
650 let src = Sources {
651 config: &config,
652 clients: &factory,
653 };
654 let mut t = Target::new(TargetKind::Token, "USDC");
655 t.chain = Some("ethereum".into());
656 t.venues = vec!["eth".into(), "no-such-venue".into()];
657 t.rules = vec![
658 Rule::parse("mid_price > 0").unwrap(),
659 Rule::parse("volume_change_pct <= 100").unwrap(),
660 Rule::parse("price_usd > 0").unwrap(),
661 ];
662 let mut ready = ready(vec![t]);
663 ready.contracts = vec![Some("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48".into())];
664 let mut history = History::default();
665 let first = tick(&ready, src, &quiet(), &mut history).await;
666 let labels: Vec<(&str, Option<&str>)> = first
667 .snapshot
668 .targets
669 .iter()
670 .map(|r| (r.target.as_str(), r.venue.as_deref()))
671 .collect();
672 assert_eq!(
673 labels,
674 vec![
675 ("USDC@eth", Some("eth")),
676 ("USDC@no-such-venue", Some("no-such-venue")),
677 ("USDC", None),
678 ]
679 );
680 let [eth, bad, id] = first.snapshot.targets.as_slice() else {
681 unreachable!()
682 };
683 let names =
684 |r: &TargetSnapshot| r.cells.iter().map(|c| c.label.clone()).collect::<Vec<_>>();
685 assert_eq!(names(eth), ["mid_price > 0", "volume_change_pct <= 100"]);
686 assert_eq!(eth.cells[0].status, RuleStatus::Pass, "{eth:?}");
687 assert!(
688 bad.cells
689 .iter()
690 .all(|c| matches!(&c.status, RuleStatus::Error(m) if m.contains("no-such-venue"))),
691 "{bad:?}"
692 );
693 assert_eq!(names(id), ["price_usd > 0"]);
694 assert_eq!(id.status, TargetStatus::Passed, "{id:?}");
695 let read: Vec<bool> = first.summaries.iter().map(Option::is_some).collect();
697 assert_eq!(read, [true, false, false]);
698 let second = tick(&ready, src, &quiet(), &mut history).await;
700 assert_eq!(second.snapshot.targets[0].cells[1].value, Some(0.0));
701 assert!(matches!(
702 second.snapshot.targets[1].cells[1].status,
703 RuleStatus::Error(_)
704 ));
705 }
706}