Jaspers Charts: Interactive price chart view with lines or candles and volume, backed by Yahoo, Massive, or Databento bar data.
| 1 | import { useData } from '@jaspers-ai/sdk' |
| 2 | import { useEffect, useRef, useState } from 'react' |
| 3 | import type { FeedResult } from './Feeds' |
| 4 | import { mergeBars, type Bar, type Interval } from './plan' |
| 5 | import { databentoArgs, databentoDaily, databentoRetryEnd, isRateLimit, massiveArgs, massiveBars, weeklyBars } from './sources' |
| 6 | |
| 7 | /** Bars kept across source and interval switches: "massive:<interval>:<symbol>" and "databento:<symbol>" (daily). */ |
| 8 | export type BarCache = Map<string, Bar[]> |
| 9 | |
| 10 | interface Job { |
| 11 | id: number |
| 12 | symbol: string |
| 13 | kind: 'full' | 'recent' |
| 14 | attempt: number |
| 15 | args: Record<string, unknown> |
| 16 | /** Databento: the earlier end a refused call is retried at, and how many times it was moved. */ |
| 17 | end?: string |
| 18 | endTries: number |
| 19 | } |
| 20 | |
| 21 | // Massive's limit is per minute: the first wait is half a window, then whole windows. |
| 22 | const BACKOFF_S = [30, 65, 65, 90, 120, 120] |
| 23 | |
| 24 | /** |
| 25 | * Loads Massive or Databento bars one symbol at a time, through the installed plugins' sources: a full |
| 26 | * history on first sight of a symbol, a short recent window merged in on each refresh tick, and a |
| 27 | * rate-limited call waited out and tried again. Draws nothing; reports each symbol under |
| 28 | * "<source>:<interval>:<symbol>". |
| 29 | */ |
| 30 | export function QueuedFeeds(props: { |
| 31 | source: 'massive' | 'databento' |
| 32 | symbols: string[] |
| 33 | interval: Interval |
| 34 | tick: number |
| 35 | cache: BarCache |
| 36 | onResult: (key: string, result: FeedResult) => void |
| 37 | }) { |
| 38 | const { source, symbols, interval, tick, cache, onResult } = props |
| 39 | const [queue, setQueue] = useState<Job[]>([]) |
| 40 | const [waiting, setWaiting] = useState(false) |
| 41 | const known = useRef(new Set<string>()) |
| 42 | const ids = useRef(1) |
| 43 | const timer = useRef<ReturnType<typeof setTimeout> | null>(null) |
| 44 | const firstTick = useRef(tick) |
| 45 | useEffect(() => () => void (timer.current && clearTimeout(timer.current)), []) |
| 46 | |
| 47 | const cacheKey = (s: string) => (source === 'databento' ? `databento:${s}` : `massive:${interval}:${s}`) |
| 48 | const shown = (s: string) => { |
| 49 | const bars = cache.get(cacheKey(s)) |
| 50 | return bars && source === 'databento' && interval === '1wk' ? weeklyBars(bars) : bars |
| 51 | } |
| 52 | const report = (s: string, r: Partial<FeedResult>) => |
| 53 | onResult(`${source}:${interval}:${s}`, { bars: shown(s) ?? [], error: null, loading: false, ...r }) |
| 54 | const job = (symbol: string, kind: 'full' | 'recent', attempt = 0, end?: string, endTries = 0): Job => { |
| 55 | const now = new Date() |
| 56 | const id = ids.current++ |
| 57 | const args = |
| 58 | source === 'massive' ? massiveArgs(symbol, interval, kind, now) : databentoArgs(symbol, kind, now, id + attempt, end) |
| 59 | return { id, symbol, kind, attempt, args, end, endTries } |
| 60 | } |
| 61 | const local = (s: string) => (source === 'databento' && s.startsWith('^') ? 'no index data on Databento' : null) |
| 62 | |
| 63 | // New symbols: cached bars at once, else a full load queued. Removed symbols leave the queue. |
| 64 | useEffect(() => { |
| 65 | const add: Job[] = [] |
| 66 | for (const s of symbols) { |
| 67 | if (known.current.has(s)) continue |
| 68 | known.current.add(s) |
| 69 | const refused = local(s) |
| 70 | if (refused) report(s, { error: refused, note: 'failed' }) |
| 71 | else if (cache.has(cacheKey(s))) report(s, {}) |
| 72 | else { |
| 73 | report(s, { loading: true, note: 'queued' }) |
| 74 | add.push(job(s, 'full')) |
| 75 | } |
| 76 | } |
| 77 | for (const s of [...known.current]) if (!symbols.includes(s)) known.current.delete(s) |
| 78 | setQueue((q) => [...q.filter((j) => symbols.includes(j.symbol)), ...add]) |
| 79 | }, [symbols.join(',')]) |
| 80 | |
| 81 | // Each refresh: a recent window for every cached symbol, the full load again for one that failed. |
| 82 | useEffect(() => { |
| 83 | if (tick === firstTick.current) return |
| 84 | setQueue((q) => { |
| 85 | const add: Job[] = [] |
| 86 | for (const s of symbols) { |
| 87 | if (local(s) || q.some((j) => j.symbol === s)) continue |
| 88 | add.push(job(s, cache.has(cacheKey(s)) ? 'recent' : 'full')) |
| 89 | } |
| 90 | return [...q, ...add] |
| 91 | }) |
| 92 | }, [tick]) |
| 93 | |
| 94 | const done = (j: Job, result: { rows?: unknown[]; error?: string }) => { |
| 95 | const key = cacheKey(j.symbol) |
| 96 | const had = cache.has(key) |
| 97 | if (result.error !== undefined) { |
| 98 | const message = result.error.split('\n')[0].slice(0, 200) |
| 99 | // Databento has not published up to the end asked: try once more at the end it names, quietly. |
| 100 | const earlier = source === 'databento' && j.endTries < 1 ? databentoRetryEnd(result.error, String(j.args.end)) : null |
| 101 | if (earlier) { |
| 102 | const next = job(j.symbol, j.kind, j.attempt, earlier, j.endTries + 1) |
| 103 | setQueue((q) => q.map((x) => (x.id === j.id ? next : x))) |
| 104 | return |
| 105 | } |
| 106 | if (isRateLimit(message) && j.attempt < BACKOFF_S.length) { |
| 107 | const wait = BACKOFF_S[j.attempt] |
| 108 | report(j.symbol, { loading: !had, note: `rate-limited, retrying in ${wait}s`, error: null, detail: message }) |
| 109 | setWaiting(true) |
| 110 | timer.current = setTimeout(() => { |
| 111 | setQueue((q) => (q[0]?.id === j.id ? [job(j.symbol, j.kind, j.attempt + 1, j.end, j.endTries), ...q.slice(1)] : q)) |
| 112 | setWaiting(false) |
| 113 | }, wait * 1000) |
| 114 | return |
| 115 | } |
| 116 | // An index the Massive plan does not cover says so, rather than the raw 403. |
| 117 | const shownError = |
| 118 | source === 'massive' && String(j.args.ticker).startsWith('I:') && /\b403\b|not_authorized|not entitled|does not include/i.test(message) |
| 119 | ? 'no index data on your plan' |
| 120 | : message |
| 121 | // A refresh that fails keeps the bars already drawn. |
| 122 | report(j.symbol, had ? { note: 'refresh failed', detail: message } : { error: shownError, note: 'failed', detail: message }) |
| 123 | } else { |
| 124 | const rows = result.rows ?? [] |
| 125 | const bars = source === 'massive' ? massiveBars(rows, interval, j.kind) : databentoDaily(rows) |
| 126 | if (j.kind === 'full') { |
| 127 | if (bars.length > 0) cache.set(key, bars) |
| 128 | } else if (bars.length > 0) cache.set(key, mergeBars([cache.get(key) ?? [], bars])) |
| 129 | if (cache.has(key)) report(j.symbol, {}) |
| 130 | else report(j.symbol, { error: 'No data', note: 'failed' }) |
| 131 | // Databento's end is exclusive: the full load ends yesterday, so ask at once for the last published day. |
| 132 | if (source === 'databento' && j.kind === 'full' && bars.length > 0) { |
| 133 | const next = job(j.symbol, 'recent') |
| 134 | setQueue((q) => [...q.filter((x) => x.id !== j.id), next]) |
| 135 | return |
| 136 | } |
| 137 | } |
| 138 | setQueue((q) => q.filter((x) => x.id !== j.id)) |
| 139 | } |
| 140 | |
| 141 | const head = waiting ? undefined : queue[0] |
| 142 | // The symbol now loading shows as loading, not queued; a refresh keeps its chip as it is. |
| 143 | useEffect(() => { |
| 144 | if (head && head.kind === 'full') report(head.symbol, { loading: true, note: head.attempt > 0 ? 'retrying' : undefined }) |
| 145 | }, [head?.id]) |
| 146 | if (!head) return null |
| 147 | return ( |
| 148 | <Request |
| 149 | key={head.id} |
| 150 | sourceId={source === 'massive' ? 'massive/aggregates' : 'databento/underlying-ohlcv'} |
| 151 | args={head.args} |
| 152 | onDone={(r) => done(head, r)} |
| 153 | /> |
| 154 | ) |
| 155 | } |
| 156 | |
| 157 | /** One call to another plugin's source; reports once when it answers. */ |
| 158 | function Request(props: { sourceId: string; args: Record<string, unknown>; onDone: (r: { rows?: unknown[]; error?: string }) => void }) { |
| 159 | const { data, text, error, loading } = useData(props.sourceId, props.args) |
| 160 | const fired = useRef(false) |
| 161 | const onDone = useRef(props.onDone) |
| 162 | onDone.current = props.onDone |
| 163 | useEffect(() => { |
| 164 | if (fired.current) return |
| 165 | if (error) { |
| 166 | fired.current = true |
| 167 | onDone.current({ error: String(error) }) |
| 168 | } else if (!loading && Array.isArray(data)) { |
| 169 | fired.current = true |
| 170 | onDone.current({ rows: data as unknown[] }) |
| 171 | } else if (!loading && data === undefined && text) { |
| 172 | fired.current = true |
| 173 | onDone.current({ error: text }) |
| 174 | } |
| 175 | }, [data, text, error, loading]) |
| 176 | return null |
| 177 | } |