1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

import databento as db
import pandas as pd

class PriceMovementScanner:
    """Scanner for detecting large price movements in all US equities."""

    def __init__(
        self,
        pct_threshold: float = 0.03,  # Default threshold for alert
    ) -> None:
        """Initialize scanner with a configurable threshold."""
        self.pct_threshold = pct_threshold
        self.start = pd.Timestamp.now(tz="US/Eastern").normalize()
        self.premarket_start = self.start.replace(hour=4)  # Start of today's pre-market session

        self.symbol_directory: dict[int, str] = {}
        self.last_day_lookup: dict[str, float] = self.get_last_day_lookup()
        self.is_signal_lit: dict[str, bool] = {symbol: False for symbol in self.last_day_lookup}

    def get_last_day_lookup(self) -> dict[str, float]:
        """Get yesterday's closing prices for all symbols."""
        client = db.Historical()

        # Get OHLCV-1d data from the previous session
        data = client.timeseries.get_range(
            dataset="EQUS.SUMMARY",
            schema="ohlcv-1d",
            symbols="ALL_SYMBOLS",
            start=(self.start - pd.offsets.BusinessDay(1)).date(),
        )

        # Request symbology: This is required for ALL_SYMBOLS requests
        # which don't automatically map instrument ID to raw ticker symbol
        symbology_json = data.request_symbology(client)
        data.insert_symbology_json(symbology_json)

        df = data.to_df()

        return df.set_index("symbol")["close"].to_dict()

    def scan(self, event: db.DBNRecord) -> None:
        """
        Scan for large price movements in market data events.
        """
        if isinstance(event, db.SymbolMappingMsg):
            self.symbol_directory[event.instrument_id] = event.stype_out_symbol
            return

        if not isinstance(event, db.MBP1Msg):
            return

        symbol = self.symbol_directory[event.instrument_id]

        # Skip if alert already triggered for the symbol
        if self.is_signal_lit[symbol]:
            return

        bid = event.levels[0].pretty_bid_px
        ask = event.levels[0].pretty_ask_px

        # Skip if one side of the book is empty
        if pd.isna(bid) or pd.isna(ask):
            return

        # Calculate change since the previous close
        mid = (bid + ask) / 2

        previous_close = self.last_day_lookup.get(symbol)
        # New listing, no data from previous session
        if previous_close is None:
            return

        pct_change = (mid - previous_close) / previous_close

        # Trigger alert if the threshold is exceeded and no previous alert
        if abs(pct_change) > self.pct_threshold:
            ts = event.pretty_ts_event.tz_convert("US/Eastern")
            print(
                f"[{ts.isoformat()}] {symbol} moved by {pct_change:.2%} "
                f"(current: {mid:.4f}, previous: {previous_close:.4f})",
            )
            self.is_signal_lit[symbol] = True


if __name__ == "__main__":
    # Instantiate scanner class
    scanner = PriceMovementScanner()

    # Create a live client
    live = db.Live(key="$YOUR_API_KEY")

    # Subscribe MBP-1 schema for ALL_SYMBOLS
    # Start subscription at pre-market open
    live.subscribe(
        dataset="EQUS.MINI",
        schema="mbp-1",
        symbols="ALL_SYMBOLS",
        start=scanner.premarket_start,
    )

    # Add callback and start
    live.add_callback(scanner.scan)
    live.start()

    # Run indefinitely
    live.block_for_close()

1
2
3
4
5
6
7
8
9
10
11

[2025-05-06T04:00:00.023079908-04:00] CORN moved by 4.19% (current: 19.1400, previous: 18.3700)
[2025-05-06T04:00:00.046518714-04:00] CPER moved by 4.04% (current: 30.4200, previous: 29.2400)
[2025-05-06T04:00:00.047119403-04:00] FETH moved by -8.56% (current: 16.6050, previous: 18.1600)
[2025-05-06T04:00:00.069275113-04:00] XXRP moved by -3.80% (current: 29.3300, previous: 30.4900)
[2025-05-06T04:00:00.118536539-04:00] ETHT moved by 24.40% (current: 5.7100, previous: 4.5900)
...
[2025-05-06T09:33:43.338614253-04:00] XMVM moved by 19.04% (current: 62.5450, previous: 52.5400)
[2025-05-06T09:33:46.814411294-04:00] BELFB moved by -5.09% (current: 65.4500, previous: 68.9600)
[2025-05-06T09:33:53.530064671-04:00] CATO moved by -3.25% (current: 2.2350, previous: 2.3100)
[2025-05-06T09:33:53.711869173-04:00] WRB moved by -11.42% (current: 64.2150, previous: 72.4900)
[2025-05-06T09:33:55.223052174-04:00] QUS moved by -3.72% (current: 149.8250, previous: 155.6100)