Line data Source code
1 : import type { HyperDBSourceConnection } from '../../platform/hyperdb/hyperdb_incremental_extract.js';
2 : import { sqlLiteral } from '../shared/sql_template.js';
3 :
4 : /// A trade of a symbol with the best bid and ask at its time, a row of the table `ticks`
5 : export interface Tick {
6 : /// The symbol, e.g. AAPL or BTC-USD
7 : symbol: string;
8 : /// The time of the trade in milliseconds since the epoch
9 : ts: number;
10 : /// The price of the trade
11 : price: number;
12 : /// The traded quantity
13 : size: number;
14 : /// The best bid, null if unknown
15 : bid: number | null;
16 : /// The best ask, null if unknown
17 : ask: number | null;
18 : }
19 :
20 : /// The state of a feed, shown next to its name
21 : export interface TickFeedStatus {
22 : /// Is the feed delivering ticks?
23 : state: 'connecting' | 'live' | 'paused' | 'failed';
24 : /// The reason of a failure or a reconnect
25 : detail?: string;
26 : }
27 :
28 : /// A source of ticks, e.g. a generator or a market data stream
29 : export interface TickFeed {
30 : /// Start delivering ticks, the feed calls emit with the ticks since its previous call
31 : start(emit: (ticks: Tick[]) => void, onStatus: (status: TickFeedStatus) => void): void;
32 : /// Stop delivering ticks
33 : stop(): void;
34 : }
35 :
36 : /// The statement that inserts ticks into the table `ticks`
37 1 : export function insertTicksSql(ticks: readonly Tick[]): string {
38 1 : const rows = ticks.map(tick => `(${[
39 1 : sqlLiteral(tick.symbol),
40 1 : `${sqlLiteral(new Date(tick.ts).toISOString())}::timestamptz`,
41 1 : sqlLiteral(tick.price),
42 1 : sqlLiteral(tick.size),
43 1 : sqlLiteral(tick.bid),
44 1 : sqlLiteral(tick.ask),
45 1 : ].join(', ')})`);
46 1 : return `insert into ticks (symbol, ts, price, size, bid, ask) values\n${rows.join(',\n')}`;
47 1 : }
48 :
49 : /// The statement that deletes the ticks older than the retention
50 1 : export function deleteTicksBeforeSql(retentionMs: number): string {
51 1 : return `delete from ticks where ts < now() - interval '${Math.round(retentionMs / 1000)} seconds'`;
52 1 : }
53 :
54 : /// The options of a tick writer
55 : export interface TickWriterOptions {
56 : /// The time between two inserts
57 : flushIntervalMs?: number;
58 : /// The maximum number of ticks of an insert, the rest waits for the next one
59 : maxBatchSize?: number;
60 : /// The age of the oldest ticks that the table keeps
61 : retentionMs?: number;
62 : /// The time between two deletions of old ticks
63 : retentionIntervalMs?: number;
64 : /// Receives the failures of statements, the writer keeps going
65 : onError?: (error: unknown) => void;
66 : }
67 :
68 : /// Write the ticks of a feed into the table `ticks` in batches.
69 : /// All statements run one after another on the connection of the writer, a HyperDB WASM connection runs one at a time.
70 1 : export class TickWriter {
71 : /// The ticks waiting for the next insert
72 1 : private pending: Tick[] = [];
73 : /// The timer of the next flush
74 1 : private timer: ReturnType<typeof setTimeout> | null = null;
75 : /// The running flush
76 1 : private flushing: Promise<void> | null = null;
77 : /// The time of the last deletion of old ticks, the first one is due an interval after the start
78 1 : private lastRetention = Date.now();
79 : /// Is the writer stopped?
80 1 : private stopped = false;
81 : /// The number of inserted ticks
82 1 : written = 0;
83 : /// The number of ticks that were dropped because the inserts did not keep up
84 1 : dropped = 0;
85 :
86 1 : private readonly flushIntervalMs: number;
87 1 : private readonly maxBatchSize: number;
88 1 : private readonly retentionMs: number;
89 1 : private readonly retentionIntervalMs: number;
90 :
91 1 : constructor(private readonly connection: HyperDBSourceConnection, private readonly options: TickWriterOptions = {}) {
92 1 : this.flushIntervalMs = options.flushIntervalMs ?? 200;
93 1 : this.maxBatchSize = options.maxBatchSize ?? 10000;
94 1 : this.retentionMs = options.retentionMs ?? 30 * 60 * 1000;
95 1 : this.retentionIntervalMs = options.retentionIntervalMs ?? 10 * 1000;
96 1 : }
97 :
98 : /// Queue ticks for the next insert. Beyond 10 batches the oldest ticks are dropped, the table stays live.
99 1 : push(ticks: readonly Tick[]): void {
100 1 : if (this.stopped) return;
101 : // A spread of a large backfill would exceed the argument limit of push
102 1 : for (const tick of ticks) this.pending.push(tick);
103 1 : const overflow = this.pending.length - 10 * this.maxBatchSize;
104 1 : if (overflow > 0) {
105 1 : this.pending.splice(0, overflow);
106 1 : this.dropped += overflow;
107 1 : }
108 1 : this.timer ??= setTimeout(() => {
109 1 : this.timer = null;
110 1 : this.flushing = this.flush();
111 1 : }, this.flushIntervalMs);
112 1 : }
113 :
114 : /// Insert the pending ticks and delete old ones when it is time to
115 1 : async flush(): Promise<void> {
116 1 : await this.flushing;
117 1 : if (this.stopped) return;
118 1 : try {
119 1 : while (this.pending.length > 0) {
120 1 : const batch = this.pending.splice(0, this.maxBatchSize);
121 1 : await this.connection.execute(insertTicksSql(batch));
122 1 : this.written += batch.length;
123 1 : }
124 1 : const now = Date.now();
125 1 : if (now - this.lastRetention >= this.retentionIntervalMs) {
126 1 : this.lastRetention = now;
127 1 : await this.connection.execute(deleteTicksBeforeSql(this.retentionMs));
128 1 : }
129 1 : } catch (e) {
130 1 : this.options.onError?.(e);
131 1 : }
132 1 : }
133 :
134 : /// Stop writing, pending ticks are dropped
135 1 : async stop(): Promise<void> {
136 1 : this.stopped = true;
137 1 : if (this.timer != null) clearTimeout(this.timer);
138 1 : this.timer = null;
139 1 : this.pending = [];
140 1 : await this.flushing?.catch(() => { });
141 1 : }
142 1 : }
|
