Documentation
¶
Overview ¶
Package candles derives live OHLC candle series from the two sources the Korbit API offers — the REST candles endpoint and the public trade stream — because the API has no candle WebSocket channel. It is the single home of the bucket math shared by its two consumers:
- the monitor command's synthesized `candle` channel (Synth, synth.go), which needs exact decimal-string output and stream-grade guarantees;
- the TUI's candle chart (a Series per open chart, converted to chart floats at the edge — chartCandles in internal/tui/chart.go).
Model — REST seeds are authoritative, trades extend the live edge ¶
A Series holds one symbol+interval's candles, ascending by bucket start; the last bucket is the live (still-open) one. Seed/Backfill merge authoritative REST rows in by bucket timestamp: an incoming row overwrites a same-Time bucket (authoritative wins, so a re-seed heals any folded approximation) and non-overlapping rows union in (so a recent re-seed never drops older loaded history). FoldTrade extends the live bucket as trades arrive (close = last price, high/low = running extremes, volume accumulates) and rolls over to a fresh bucket when a trade falls past it; RollTo does the same on a clock tick, so a bucket closes on time even when no trade crosses the boundary.
Bucket alignment always derives from the seeded grid (newStart = liveStart + n*period), never from the unix epoch: the server anchors day/week buckets to its own calendar, and only its own rows know that grid. This is why a Series folds nothing before its first Seed.
A bucket with no trades has no traded price, and the two consumers want opposite things from it, so it is a per-Series MODE (Series.Reset vs Series.ResetGapless): the monitor's stream must stay gapless on its grid (a consumer waits on bar-closes and keeps indicator arrays aligned), so in gapless mode a jumped-over bucket is synthesized flat at the previous close with volume "0" — volume "0" being the unambiguous "no trades, prices are carried-forward filler" marker; the TUI's chart must not draw a price at which nothing traded, so in plain mode the bucket is skipped and the slot-based chart compresses the time axis across it.
Money stays decimal strings ¶
Prices and volumes are decimal strings end to end: comparison and volume accumulation run through shopspring/decimal, and a value no arithmetic ever touched passes through verbatim. Each Candle also carries float64 mirrors (computed once, when the value is set) so a chart consumer can render without re-parsing; the floats are display copies, never inputs to the math. A row whose OHLC does not parse as a finite value is dropped, so a NaN/Inf can never enter a series.
Index ¶
- Constants
- Variables
- func IntervalMs(interval string) int64
- func ValidInterval(interval string) bool
- type Bar
- type Candle
- type Config
- type SeedResult
- type Series
- func (s *Series) At(i int) Candle
- func (s *Series) Backfill(bars []Bar) (added int)
- func (s *Series) Candles() []Candle
- func (s *Series) Empty() bool
- func (s *Series) FoldTrade(id int64, price, qty string, ts int64) (finalized []Candle, updated bool)
- func (s *Series) Interval() string
- func (s *Series) LastTradeID() int64
- func (s *Series) Len() int
- func (s *Series) Live() (Candle, bool)
- func (s *Series) OldestTime() int64
- func (s *Series) PeriodMs() int64
- func (s *Series) Reset(interval string)
- func (s *Series) ResetGapless(interval string)
- func (s *Series) RollTo(nowMs int64) (finalized []Candle)
- func (s *Series) Seed(bars []Bar, asOfTradeID int64)
- func (s *Series) SeedRebase(bars []Bar, asOfTradeID int64)
- func (s *Series) TrimOldest(keep int)
- type Synth
Constants ¶
const Channel = "candle"
Channel is the synthesized candle channel's name in the monitor's output. It is CLI-derived — deliberately NOT one of the stream package's subscribable channel names, because no such WebSocket channel exists on the server.
Variables ¶
var Intervals = []string{"1", "5", "15", "30", "60", "240", "1D", "1W"}
Intervals is the canonical candle-interval enum, exactly the REST candles endpoint's values (minutes, then 1D/1W). It is the single source for flag validation and error text; no aliases are accepted.
Functions ¶
func IntervalMs ¶
IntervalMs maps a canonical interval to its duration in milliseconds, or 0 if unrecognized.
func ValidInterval ¶
ValidInterval reports whether interval is one of the canonical enum values.
Types ¶
type Bar ¶
type Bar struct {
Timestamp int64 `json:"timestamp"`
Open string `json:"open"`
High string `json:"high"`
Low string `json:"low"`
Close string `json:"close"`
Volume string `json:"volume"`
}
Bar is one OHLC row in the REST candles response shape: a bucket-start timestamp (unix ms) and decimal-string prices. It is the input to Series.Seed/Series.Backfill (unmarshaled straight from the REST rows) and the price fields of the monitor's emitted candle payload.
type Candle ¶
type Candle struct {
Time int64 // bucket start, unix ms
Open, High, Low, Close, Volume string
OpenF, HighF, LowF, CloseF, VolumeF float64
}
Candle is one stored bucket: the decimal strings that are the source of truth, plus float64 mirrors for chart consumers (display copies computed when the string was set — never inputs to the bucket math).
type Config ¶
type Config struct {
Symbols []string
Intervals []string
// History is the number of closed candles to emit per scope before live
// streaming (the --candle-history value; 0 = live only).
History int
// Fetch returns up to limit REST candle rows for symbol+interval ending at
// endMs (0 = up to the current still-open bucket), ascending. It must be
// safe for concurrent use.
Fetch func(ctx context.Context, symbol, interval string, limit int, endMs int64) ([]Bar, error)
// ServerNow is the server-clock estimate in unix ms (quiet-market
// finalization compares bucket ends against server-issued timestamps).
ServerNow func() int64
// Now is the local clock, stamped on the notices this layer raises.
Now func() int64
// Log is an optional operational logger (nil = silent).
Log *slog.Logger
// Tunables (0 = default), exposed so tests run the real machinery fast.
FinalizeGraceMs int64 // quiet-market finalization lag; default 2000
RetryMinMs int64 // seed-retry backoff floor; default 1000
RetryMaxMs int64 // seed-retry backoff cap; default 30000
ReplayBufferCap int // per-scope in-flight trade buffer cap; default 8192
}
Config configures a Synth. Symbols and Intervals must be normalized and validated by the caller (canonical interval values, deduped).
type SeedResult ¶
type SeedResult struct {
// contains filtered or unexported fields
}
SeedResult is one finished seed fetch, delivered on Results. Its fields are internal; the monitor loop only shuttles it from Results into Apply.
type Series ¶
type Series struct {
// contains filtered or unexported fields
}
Series accumulates the candle series for one symbol+interval. The zero value is not ready; call Series.Reset with an interval first. It is not safe for concurrent use; both consumers drive it from a single goroutine.
func (*Series) At ¶
At returns the candle at index i (0 = oldest), copying one bucket. A renderer walks the series with Len/At to convert it without the whole-slice allocation Candles makes — the copy that matters on a hot redraw path. i must be in [0, Len); an out-of-range index panics like a slice access.
func (*Series) Backfill ¶
Backfill merges an older page of authoritative REST rows and reports how many buckets were newly added. Zero means the page held nothing the series lacked — the caller can treat that as the start of available history and stop paging (use OldestTime minus one as the next page's end bound). Backfill never touches the trade high-water mark.
func (*Series) FoldTrade ¶
func (s *Series) FoldTrade(id int64, price, qty string, ts int64) (finalized []Candle, updated bool)
FoldTrade extends the live bucket with one public trade, or rolls over when the trade falls past it. It returns the buckets the roll finalized (empty intermediates synthesized flat at the previous close — capped, see maxSynthOnJump) and whether the trade changed the series at all. Trades are deduped by id (ids at or below the high-water mark are ignored), so feeding an overlapping batch repeatedly is safe. A trade older than the live bucket, an unseeded series, or an unknown interval is a no-op.
func (*Series) LastTradeID ¶
LastTradeID is the fold dedup high-water: the highest trade id already folded, or the asOfTradeID the last Seed carried. A consumer pulls only trades past it so an already-folded batch costs nothing.
func (*Series) OldestTime ¶
OldestTime is the bucket start (unix ms) of the oldest loaded candle, or 0 if the series is empty. It is the end bound for the next Backfill page.
func (*Series) Reset ¶
Reset points the series at a new interval and clears any accumulated state. An unrecognized interval leaves the series seed-able but unable to fold or roll (periodMs 0), so it degrades to whatever the last Seed delivered. A jump over empty buckets leaves a hole (see Series.fillEmpty); use ResetGapless for the synthesizing mode.
func (*Series) ResetGapless ¶
ResetGapless is Reset in the gapless mode: jumps synthesize the skipped empty buckets (flat at the previous close, volume "0") so the live-derived series never has grid holes. This is the monitor synthesizer's mode; RollTo is designed for this mode too (its whole product is on-time flat closes).
func (*Series) RollTo ¶
RollTo closes the live bucket (repeatedly) when nowMs has passed its end, synthesizing each successor flat at the previous close with volume "0", and returns the finalized buckets. The bucket containing nowMs stays live. Call it on a clock tick so a bucket closes on time in a quiet market; pass a grace-adjusted now so a slightly-late trade still lands before its bucket is sealed. A jump past maxSynthOnJump buckets skips the intermediates (they are left as a hole) and re-opens at the bucket containing nowMs. RollTo is gapless-mode-only (its whole product is synthesized flat buckets, exactly what plain mode exists to avoid) and is a no-op after a plain Reset.
func (*Series) Seed ¶
Seed merges authoritative REST rows into the series, overwriting any folded approximation on the buckets they cover while preserving older loaded history (see the package doc's merge rule). asOfTradeID is the newest public trade id the caller had seen when it kicked off the fetch; the series will not fold a trade at or below it, so trades the fetched rows already counted are not double-counted. Pass 0 if unknown (the next strictly-newer trade still folds). Rows with an unparseable price are dropped.
func (*Series) SeedRebase ¶
SeedRebase is Seed for a consumer that will REPLAY its own record of the trades delivered after asOfTradeID: unlike Seed, which only ever RAISES the fold high-water mark, it RESETS the mark to asOfTradeID so trades whose folds the authoritative merge just overwrote can be re-applied on top of the fetched rows. Only safe when the caller's replay set is exactly the post-asOf deliveries (the monitor synthesizer's in-flight buffer) and the series received NO folds while the fetch was in flight — anything else double-counts.
func (*Series) TrimOldest ¶
TrimOldest drops all but the newest keep buckets, releasing the backing memory. A long-running consumer bounds growth with it: the monitor keeps a small emit window, the TUI chart a large scroll-back cap so an open chart's series cannot grow without limit as buckets roll over.
type Synth ¶
type Synth struct {
// contains filtered or unexported fields
}
Synth derives the monitor's synthesized `candle` channel: it consumes the session's public trade events and reliability notices and produces candle Data events (Origin stream.OriginDerived, channel Channel) plus its own BACKFILL_* notices, one Series per symbol×interval.
Guarantees, and where they come from ¶
The candle stream inherits the trade pipeline's guarantees instead of re-implementing them: while the public connection is up, the session's trade stream is deduped and gap-patched, so folded buckets are exact over the DELIVERED trades — there is NO periodic REST re-sync. (Exact over delivered, not over executed: the public endpoint may drop a frame mid-connection without a reconnect, which is undetectable — see the stream package doc — so a derived bar can differ slightly from the server's own aggregation. The REST candles endpoint stays authoritative where exact values matter; this channel is for real-time signals.) REST is touched only to SEED a scope (the current partial bucket plus the requested history depth; a subscription that starts mid-bucket cannot learn the bucket's true open/high/low/volume from trades alone) and to RE-SEED it after an event that may have cost trades a fold: a public reconnect, a trade DATA_GAP, or a completed gap patch (whose rows can land behind an already-rolled live edge, where folding must not touch history). A failed seed self-heals on an exponential backoff, each attempt a fresh BACKFILL_START (reason "retry"), mirroring the stream layer's private recovery. Like that layer, every BACKFILL_START is PAIRED with a BACKFILL_DONE — on failure the DONE (failures:1) follows the BACKFILL_FAILED alarm, and a result discarded because its scope died still closes its START — because consumers balance on the pairing (the --stateful store's backfill counter would otherwise wedge Health.Backfilling true forever).
Seed ordering: the snapshot kicks the fetch; in-flight trades replay ¶
The initial and reconnect seeds are kicked by the trade subscription's OWN snapshot frame, not eagerly, and every kick captures the trade-id watermark at that moment: everything delivered before the kick was executed before the request went out, so it is inside the fetched rows and must not re-fold. Trades delivered while the fetch is in flight are BUFFERED (not folded) and replayed on top of the authoritative rows at Apply (Series.SeedRebase resets the fold mark to the kick watermark), so a trade the fetch missed can never be lost — without the replay, a trade executed after the response was built but delivered before it applied would vanish, and a finalizing bucket could keep a stale close/extreme forever. The residual error runs the other, self-healing way: a buffered trade executed BEFORE the response was built is counted by both the fetched row and its replay, a volume overcount bounded by one round trip's trades on the applied bucket, gone at the next rollover; prices are exact (replay re-asserts real extremes and ends at the true latest close). Gap-patch triggers (DATA_GAP, gap BACKFILL_DONE) re-seed immediately — their rows were delivered before the notice by construction.
Consumers order per key: the key is {interval, timestamp} and the NEWEST line for a key supersedes prior lines — a re-seed re-emits corrected bars (final:true for closed buckets) under exactly that rule. A bucket with no trades still closes on time (see Tick) as a flat zero-volume bar, and closed buckets a REST seed skipped are filled the same way on emission, so the emitted series is gapless on its grid.
Finalization clock ¶
Trade-driven rollover finalizes immediately — a strictly-newer trade past the boundary proves the bucket complete. In a quiet market Tick finalizes instead, comparing the bucket end against the SERVER-clock estimate (this is a compare-against-server-timestamps use, the trade/bucket timestamps being server-issued) minus a grace so a slightly-late frame still lands. While the public connection is down finalization is suspended — the synthesizer cannot know the market went quiet — and the post-reconnect re-seed catches history up.
Threading ¶
All methods run on the monitor's single event-loop goroutine. The only concurrency is the seed fetches: goroutines that call Fetch and post to the Results channel; the loop selects on Results and feeds each value back into Apply. Start wires the fetch context and must be called exactly once, before any event is fed.
func (*Synth) Apply ¶
func (s *Synth) Apply(res SeedResult) []stream.Event
Apply folds one seed result into its scope: on success the authoritative rows overwrite/extend the series and the candle lines are emitted (closed bars final, the live bucket not); on failure a BACKFILL_FAILED notice is raised and a retry is scheduled with exponential backoff.
func (*Synth) OnData ¶
OnData consumes one session data event. Only public trade events for a configured symbol do anything. A snapshot frame kicks any stale scope's seed (initial or post-reconnect; see the package doc); then each row folds into every seeded interval series of the symbol, and the resulting candle lines are emitted (finalized buckets first, then the updated live bucket).
func (*Synth) OnNotice ¶
OnNotice consumes one session notice. It tracks the public connection state (finalization gates on it), marks scopes stale on a public reconnect (the resubscribe snapshot then kicks the re-seed), and re-seeds immediately on a trade DATA_GAP or a completed trade gap patch — events whose rows can land behind an already-rolled live edge, where folding must not touch history. The notice itself is not re-emitted — the monitor already delivers every session notice.
func (*Synth) Results ¶
func (s *Synth) Results() <-chan SeedResult
Results delivers finished seed fetches; the monitor loop selects on it and passes each value to Apply.
func (*Synth) Start ¶
Start wires the context that bounds every fetch this synthesizer ever runs. Seeds are not kicked here — the trade subscription's snapshot kicks them (see the package doc's seed-ordering rationale).
Source Files