Wickra 0.1.0: streaming-first technical indicators
A multi-language technical analysis library: 25 indicators across trend,
momentum, volatility, and volume families, every one a state machine with
O(1) per-tick updates. Batch evaluation is provided by a blanket extension
trait over the streaming primitive, so live trading bots and historical
backtests run the same code path.
What ships in this initial drop:
crates/wickra-core - 25 indicators, Indicator/BatchExt/Chain traits,
OHLCV types with validation; 171 unit tests,
property tests, Wilder/Bollinger textbook tests.
crates/wickra - top-level facade + criterion benches for every
indicator at 1K/10K/100K series sizes.
crates/wickra-data - streaming CSV reader, tick-to-candle aggregator,
multi-timeframe resampler, Binance Spot kline
WebSocket adapter behind feature live-binance;
11 unit + 1 doctest.
bindings/python - PyO3 + maturin, NumPy I/O, type stubs (.pyi),
56 pytest tests including streaming==batch
equivalence, Wilder reference values, lifecycle.
bindings/node - napi-rs native module, TypeScript .d.ts
auto-generated, 7 node --test cases.
bindings/wasm - wasm-bindgen ES module for browser/bundler/Node;
interactive HTML demo at examples/index.html.
examples/ - Python and Rust scripts: backtest, live trading,
parallel multi-asset, multi-timeframe, Binance.
benchmarks/ - cross-library comparison against TA-Lib,
pandas-ta, finta, talipp; Wickra wins every
category by 11-1030x (batch) and 17x+ streaming.
.github/workflows/ - CI matrix (Rust + Python + Node + WASM on
Linux/macOS/Windows), release pipeline for
PyPI wheels and npm.
Indicators (25):
Trend SMA EMA WMA DEMA TEMA HMA KAMA
Momentum RSI MACD Stochastic CCI ROC WilliamsR ADX MFI TRIX
AwesomeOscillator Aroon
Volatility BollingerBands ATR Keltner Donchian PSAR
Volume OBV VWAP (cumulative + rolling)
cargo clippy --workspace --all-targets -D warnings is clean. License: Apache-2.0.
This commit is contained in:
kingchenc
@@ -0,0 +1,226 @@
|
||||
//! Roll trade ticks up into candles of an arbitrary timeframe.
|
||||
|
||||
use crate::error::{Error, Result};
|
||||
use wickra_core::{Candle, Tick};
|
||||
|
||||
/// A candle bucket size measured in the same unit as the tick timestamps.
|
||||
///
|
||||
/// Wickra is unit-agnostic about timestamps: choose whichever makes sense for
|
||||
/// your source (milliseconds for Binance trade events, microseconds for IB,
|
||||
/// seconds for daily bars).
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub struct Timeframe {
|
||||
bucket: i64,
|
||||
}
|
||||
|
||||
impl Timeframe {
|
||||
/// Construct a timeframe with the given bucket size in the chosen unit.
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns [`Error::InvalidTimeframe`] if `bucket <= 0`.
|
||||
pub fn new(bucket: i64) -> Result<Self> {
|
||||
if bucket <= 0 {
|
||||
return Err(Error::InvalidTimeframe(format!(
|
||||
"bucket size must be positive, got {bucket}"
|
||||
)));
|
||||
}
|
||||
Ok(Self { bucket })
|
||||
}
|
||||
|
||||
/// Convenience: build a millisecond timeframe.
|
||||
pub fn millis(ms: i64) -> Result<Self> {
|
||||
Self::new(ms)
|
||||
}
|
||||
|
||||
/// Convenience: build a seconds-resolution timeframe.
|
||||
pub fn seconds(s: i64) -> Result<Self> {
|
||||
Self::new(s)
|
||||
}
|
||||
|
||||
/// One-minute timeframe in milliseconds (`60_000`).
|
||||
pub fn one_minute_ms() -> Self {
|
||||
Self::new(60_000).expect("60_000 > 0")
|
||||
}
|
||||
|
||||
/// Bucket size.
|
||||
pub const fn bucket(self) -> i64 {
|
||||
self.bucket
|
||||
}
|
||||
|
||||
/// Floor a raw timestamp to this timeframe's bucket boundary.
|
||||
pub fn floor(self, ts: i64) -> i64 {
|
||||
ts - ts.rem_euclid(self.bucket)
|
||||
}
|
||||
}
|
||||
|
||||
/// Incrementally builds candles out of arriving ticks.
|
||||
///
|
||||
/// Each call to [`TickAggregator::push`] returns `Some(Candle)` if a previously
|
||||
/// open bar just closed (i.e. the new tick belongs to a new bucket). Use
|
||||
/// [`TickAggregator::flush`] at the end of a stream to capture the final open
|
||||
/// bar.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct TickAggregator {
|
||||
timeframe: Timeframe,
|
||||
open_bar: Option<OpenBar>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
struct OpenBar {
|
||||
bucket_start: i64,
|
||||
open: f64,
|
||||
high: f64,
|
||||
low: f64,
|
||||
close: f64,
|
||||
volume: f64,
|
||||
}
|
||||
|
||||
impl OpenBar {
|
||||
fn from_tick(t: Tick, bucket_start: i64) -> Self {
|
||||
Self {
|
||||
bucket_start,
|
||||
open: t.price,
|
||||
high: t.price,
|
||||
low: t.price,
|
||||
close: t.price,
|
||||
volume: t.volume,
|
||||
}
|
||||
}
|
||||
|
||||
fn absorb(&mut self, t: Tick) {
|
||||
if t.price > self.high {
|
||||
self.high = t.price;
|
||||
}
|
||||
if t.price < self.low {
|
||||
self.low = t.price;
|
||||
}
|
||||
self.close = t.price;
|
||||
self.volume += t.volume;
|
||||
}
|
||||
|
||||
fn into_candle(self) -> Candle {
|
||||
Candle::new_unchecked(
|
||||
self.open,
|
||||
self.high,
|
||||
self.low,
|
||||
self.close,
|
||||
self.volume,
|
||||
self.bucket_start,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl TickAggregator {
|
||||
/// Construct a new aggregator for the given timeframe.
|
||||
pub fn new(timeframe: Timeframe) -> Self {
|
||||
Self {
|
||||
timeframe,
|
||||
open_bar: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Push a tick. Returns `Some(Candle)` if a bar boundary was crossed and a
|
||||
/// previously open bar just closed.
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns an error if `tick.timestamp` is strictly less than the start of
|
||||
/// the currently open bar (out-of-order ticks are not supported).
|
||||
pub fn push(&mut self, tick: Tick) -> Result<Option<Candle>> {
|
||||
let bucket = self.timeframe.floor(tick.timestamp);
|
||||
if let Some(mut bar) = self.open_bar {
|
||||
if bucket < bar.bucket_start {
|
||||
return Err(Error::Malformed(format!(
|
||||
"tick timestamp {} is older than the open bar start {}",
|
||||
tick.timestamp, bar.bucket_start
|
||||
)));
|
||||
}
|
||||
if bucket > bar.bucket_start {
|
||||
// Close the previous bar and start a new one with this tick.
|
||||
self.open_bar = Some(OpenBar::from_tick(tick, bucket));
|
||||
return Ok(Some(bar.into_candle()));
|
||||
}
|
||||
bar.absorb(tick);
|
||||
self.open_bar = Some(bar);
|
||||
return Ok(None);
|
||||
}
|
||||
self.open_bar = Some(OpenBar::from_tick(tick, bucket));
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
/// Drain the currently open bar (if any) and return it. Useful at the end of
|
||||
/// a backtest or when shutting down a live aggregator.
|
||||
pub fn flush(&mut self) -> Option<Candle> {
|
||||
self.open_bar.take().map(OpenBar::into_candle)
|
||||
}
|
||||
|
||||
/// Configured timeframe.
|
||||
pub const fn timeframe(&self) -> Timeframe {
|
||||
self.timeframe
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn t(price: f64, ts: i64) -> Tick {
|
||||
Tick::new(price, 1.0, ts).unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn timeframe_rejects_non_positive() {
|
||||
assert!(Timeframe::new(0).is_err());
|
||||
assert!(Timeframe::new(-1).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn floors_to_bucket_boundary() {
|
||||
let tf = Timeframe::new(100).unwrap();
|
||||
assert_eq!(tf.floor(0), 0);
|
||||
assert_eq!(tf.floor(99), 0);
|
||||
assert_eq!(tf.floor(100), 100);
|
||||
assert_eq!(tf.floor(150), 100);
|
||||
assert_eq!(tf.floor(250), 200);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn aggregates_ticks_into_one_candle_within_bucket() {
|
||||
let mut agg = TickAggregator::new(Timeframe::new(60).unwrap());
|
||||
assert_eq!(agg.push(t(10.0, 0)).unwrap(), None);
|
||||
assert_eq!(agg.push(t(12.0, 15)).unwrap(), None);
|
||||
assert_eq!(agg.push(t(8.0, 30)).unwrap(), None);
|
||||
assert_eq!(agg.push(t(11.0, 50)).unwrap(), None);
|
||||
let bar = agg.flush().expect("open bar");
|
||||
assert_eq!(bar.open, 10.0);
|
||||
assert_eq!(bar.high, 12.0);
|
||||
assert_eq!(bar.low, 8.0);
|
||||
assert_eq!(bar.close, 11.0);
|
||||
assert!((bar.volume - 4.0).abs() < 1e-12);
|
||||
assert_eq!(bar.timestamp, 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn emits_candle_on_bucket_crossing() {
|
||||
let mut agg = TickAggregator::new(Timeframe::new(60).unwrap());
|
||||
agg.push(t(10.0, 0)).unwrap();
|
||||
agg.push(t(12.0, 30)).unwrap();
|
||||
let closed = agg.push(t(15.0, 60)).unwrap().expect("emits");
|
||||
assert_eq!(closed.open, 10.0);
|
||||
assert_eq!(closed.high, 12.0);
|
||||
assert_eq!(closed.low, 10.0);
|
||||
assert_eq!(closed.close, 12.0);
|
||||
|
||||
// The new tick at ts=60 opens the next bar.
|
||||
let still_open = agg.flush().unwrap();
|
||||
assert_eq!(still_open.open, 15.0);
|
||||
assert_eq!(still_open.timestamp, 60);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_out_of_order_ticks() {
|
||||
let mut agg = TickAggregator::new(Timeframe::new(60).unwrap());
|
||||
agg.push(t(10.0, 100)).unwrap();
|
||||
let err = agg.push(t(11.0, 30)).unwrap_err();
|
||||
assert!(matches!(err, Error::Malformed(_)));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,129 @@
|
||||
//! Stream OHLCV candles out of a CSV file.
|
||||
//!
|
||||
//! The reader is generic over the column layout, but ships with a sensible
|
||||
//! default ("timestamp,open,high,low,close,volume") that matches the standard
|
||||
//! Binance / Yahoo Finance / kaggle dataset format.
|
||||
|
||||
use std::path::Path;
|
||||
|
||||
use serde::Deserialize;
|
||||
|
||||
use crate::error::{Error, Result};
|
||||
use wickra_core::Candle;
|
||||
|
||||
/// Default OHLCV CSV row layout.
|
||||
///
|
||||
/// The timestamp is parsed as an `i64`; if your file ships an RFC3339 / ISO8601
|
||||
/// string instead, use [`CandleReader::with_timestamp_parser`].
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct DefaultRow {
|
||||
pub timestamp: i64,
|
||||
pub open: f64,
|
||||
pub high: f64,
|
||||
pub low: f64,
|
||||
pub close: f64,
|
||||
pub volume: f64,
|
||||
}
|
||||
|
||||
impl DefaultRow {
|
||||
fn into_candle(self) -> Result<Candle> {
|
||||
Candle::new(
|
||||
self.open,
|
||||
self.high,
|
||||
self.low,
|
||||
self.close,
|
||||
self.volume,
|
||||
self.timestamp,
|
||||
)
|
||||
.map_err(Error::from)
|
||||
}
|
||||
}
|
||||
|
||||
/// Streaming OHLCV CSV reader.
|
||||
#[derive(Debug)]
|
||||
pub struct CandleReader<R: std::io::Read> {
|
||||
reader: csv::Reader<R>,
|
||||
}
|
||||
|
||||
impl CandleReader<std::fs::File> {
|
||||
/// Open a CSV file at `path`. The first line is treated as a header by default.
|
||||
pub fn open<P: AsRef<Path>>(path: P) -> Result<Self> {
|
||||
let reader = csv::ReaderBuilder::new()
|
||||
.has_headers(true)
|
||||
.from_path(path)?;
|
||||
Ok(Self { reader })
|
||||
}
|
||||
}
|
||||
|
||||
impl<R: std::io::Read> CandleReader<R> {
|
||||
/// Build a reader from any [`std::io::Read`] source.
|
||||
pub fn from_reader(inner: R) -> Self {
|
||||
Self {
|
||||
reader: csv::ReaderBuilder::new()
|
||||
.has_headers(true)
|
||||
.from_reader(inner),
|
||||
}
|
||||
}
|
||||
|
||||
/// Replace the underlying reader; useful for testing.
|
||||
pub fn from_csv_reader(reader: csv::Reader<R>) -> Self {
|
||||
Self { reader }
|
||||
}
|
||||
|
||||
/// Iterator over decoded candles.
|
||||
pub fn candles(&mut self) -> impl Iterator<Item = Result<Candle>> + '_ {
|
||||
self.reader.deserialize::<DefaultRow>().map(|row_res| {
|
||||
let row = row_res?;
|
||||
row.into_candle()
|
||||
})
|
||||
}
|
||||
|
||||
/// Read the entire stream into a `Vec<Candle>`. Convenient for backtests.
|
||||
pub fn read_all(&mut self) -> Result<Vec<Candle>> {
|
||||
self.candles().collect()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::io::Write;
|
||||
|
||||
#[test]
|
||||
fn reads_well_formed_csv() {
|
||||
let mut tmp = tempfile::NamedTempFile::new().unwrap();
|
||||
writeln!(tmp, "timestamp,open,high,low,close,volume").unwrap();
|
||||
writeln!(tmp, "1,10.0,11.0,9.0,10.5,100").unwrap();
|
||||
writeln!(tmp, "2,10.5,11.5,10.0,11.0,150").unwrap();
|
||||
writeln!(tmp, "3,11.0,12.0,10.5,11.5,200").unwrap();
|
||||
tmp.flush().unwrap();
|
||||
|
||||
let mut r = CandleReader::open(tmp.path()).unwrap();
|
||||
let candles = r.read_all().unwrap();
|
||||
assert_eq!(candles.len(), 3);
|
||||
assert_eq!(candles[0].open, 10.0);
|
||||
assert_eq!(candles[2].close, 11.5);
|
||||
assert_eq!(candles[1].timestamp, 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_invalid_ohlc() {
|
||||
let mut tmp = tempfile::NamedTempFile::new().unwrap();
|
||||
writeln!(tmp, "timestamp,open,high,low,close,volume").unwrap();
|
||||
// high < low → core validation rejects it.
|
||||
writeln!(tmp, "1,10.0,8.0,9.0,9.5,100").unwrap();
|
||||
tmp.flush().unwrap();
|
||||
|
||||
let mut r = CandleReader::open(tmp.path()).unwrap();
|
||||
let candles: Result<Vec<Candle>> = r.candles().collect();
|
||||
assert!(candles.is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn from_reader_works_on_in_memory_data() {
|
||||
let data = "timestamp,open,high,low,close,volume\n1,1,2,0,1,10\n2,1,2,0,1,10\n";
|
||||
let mut r = CandleReader::from_reader(data.as_bytes());
|
||||
let v = r.read_all().unwrap();
|
||||
assert_eq!(v.len(), 2);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
//! Error types specific to the data sources.
|
||||
|
||||
use thiserror::Error;
|
||||
|
||||
/// Errors produced by the data layer.
|
||||
#[derive(Debug, Error)]
|
||||
pub enum Error {
|
||||
#[error("I/O error: {0}")]
|
||||
Io(#[from] std::io::Error),
|
||||
|
||||
#[error("CSV error: {0}")]
|
||||
Csv(#[from] csv::Error),
|
||||
|
||||
#[error("invalid timeframe: {0}")]
|
||||
InvalidTimeframe(String),
|
||||
|
||||
#[error("indicator-core error: {0}")]
|
||||
Core(#[from] wickra_core::Error),
|
||||
|
||||
#[error("malformed payload: {0}")]
|
||||
Malformed(String),
|
||||
|
||||
#[cfg(feature = "live-binance")]
|
||||
#[error("websocket error: {0}")]
|
||||
WebSocket(#[from] tokio_tungstenite::tungstenite::Error),
|
||||
|
||||
#[cfg(feature = "live-binance")]
|
||||
#[error("JSON decode error: {0}")]
|
||||
Json(#[from] serde_json::Error),
|
||||
}
|
||||
|
||||
/// Convenience alias for `Result<T, wickra_data::Error>`.
|
||||
pub type Result<T> = core::result::Result<T, Error>;
|
||||
@@ -0,0 +1,24 @@
|
||||
//! `wickra-data`: offline and online data sources for the Wickra indicator engine.
|
||||
//!
|
||||
//! - [`csv`]: stream OHLCV bars out of CSV files without buffering the whole
|
||||
//! history in memory.
|
||||
//! - [`aggregator`]: roll trade ticks up into candles of arbitrary timeframes.
|
||||
//! - [`resample`]: convert a stream of candles from one timeframe to a coarser one.
|
||||
//! - [`live`] (feature `live-binance`): connect to exchange websockets and yield
|
||||
//! typed events compatible with the rest of the crate.
|
||||
|
||||
#![cfg_attr(docsrs, feature(doc_auto_cfg))]
|
||||
// `tokio_tungstenite::Error` is large by itself (~200 B). Boxing every Err
|
||||
// variant per clippy::result_large_err just shifts allocation pressure into
|
||||
// the hot path. We accept the size because errors are rare in this crate.
|
||||
#![allow(clippy::result_large_err)]
|
||||
|
||||
pub mod aggregator;
|
||||
pub mod csv;
|
||||
pub mod error;
|
||||
pub mod resample;
|
||||
|
||||
#[cfg(feature = "live-binance")]
|
||||
pub mod live;
|
||||
|
||||
pub use error::{Error, Result};
|
||||
@@ -0,0 +1,4 @@
|
||||
//! Live exchange feeds. Each adapter is feature-gated; the Binance adapter
|
||||
//! lives behind the `live-binance` feature.
|
||||
|
||||
pub mod binance;
|
||||
@@ -0,0 +1,293 @@
|
||||
//! Binance spot WebSocket kline feed.
|
||||
//!
|
||||
//! Subscribes to Binance's `<symbol>@kline_<interval>` stream and emits a
|
||||
//! [`KlineEvent`] every time the server pushes a new tick. The event tells you
|
||||
//! whether the current candle is still open or has just closed.
|
||||
//!
|
||||
//! Example (requires the `live-binance` feature):
|
||||
//!
|
||||
//! ```no_run
|
||||
//! use wickra_data::live::binance::{BinanceKlineStream, Interval};
|
||||
//! # async fn run() -> wickra_data::Result<()> {
|
||||
//! let mut stream = BinanceKlineStream::connect(&["BTCUSDT".to_string()], Interval::OneMinute).await?;
|
||||
//! while let Some(event) = stream.next_event().await? {
|
||||
//! if event.is_closed {
|
||||
//! println!("closed {} @ {}", event.symbol, event.candle.close);
|
||||
//! }
|
||||
//! }
|
||||
//! # Ok(()) }
|
||||
//! ```
|
||||
|
||||
use futures_util::SinkExt;
|
||||
use futures_util::StreamExt;
|
||||
use serde::Deserialize;
|
||||
use tokio::net::TcpStream;
|
||||
use tokio_tungstenite::tungstenite::Message;
|
||||
use tokio_tungstenite::MaybeTlsStream;
|
||||
use tokio_tungstenite::WebSocketStream;
|
||||
|
||||
use crate::error::{Error, Result};
|
||||
use wickra_core::Candle;
|
||||
|
||||
/// Supported Binance kline intervals. The `as_str` value matches Binance's
|
||||
/// wire-format strings (`"1m"`, `"5m"`, `"1h"`, etc.).
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum Interval {
|
||||
OneSecond,
|
||||
OneMinute,
|
||||
ThreeMinutes,
|
||||
FiveMinutes,
|
||||
FifteenMinutes,
|
||||
ThirtyMinutes,
|
||||
OneHour,
|
||||
TwoHours,
|
||||
FourHours,
|
||||
SixHours,
|
||||
EightHours,
|
||||
TwelveHours,
|
||||
OneDay,
|
||||
OneWeek,
|
||||
}
|
||||
|
||||
impl Interval {
|
||||
/// Wire-format string used in the stream name.
|
||||
pub fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
Self::OneSecond => "1s",
|
||||
Self::OneMinute => "1m",
|
||||
Self::ThreeMinutes => "3m",
|
||||
Self::FiveMinutes => "5m",
|
||||
Self::FifteenMinutes => "15m",
|
||||
Self::ThirtyMinutes => "30m",
|
||||
Self::OneHour => "1h",
|
||||
Self::TwoHours => "2h",
|
||||
Self::FourHours => "4h",
|
||||
Self::SixHours => "6h",
|
||||
Self::EightHours => "8h",
|
||||
Self::TwelveHours => "12h",
|
||||
Self::OneDay => "1d",
|
||||
Self::OneWeek => "1w",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// One push from the Binance kline stream.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct KlineEvent {
|
||||
/// Symbol in lowercase form as sent by Binance (e.g. `"btcusdt"`).
|
||||
pub symbol: String,
|
||||
/// Interval the candle belongs to.
|
||||
pub interval: Interval,
|
||||
/// Candle in its current state (may still be open).
|
||||
pub candle: Candle,
|
||||
/// Whether the candle has been closed by the server. Closed events are the
|
||||
/// only ones safe to use for bar-completion logic.
|
||||
pub is_closed: bool,
|
||||
}
|
||||
|
||||
/// A live Binance kline stream.
|
||||
#[derive(Debug)]
|
||||
pub struct BinanceKlineStream {
|
||||
socket: WebSocketStream<MaybeTlsStream<TcpStream>>,
|
||||
/// Interval requested at connect time. Used to tag every event.
|
||||
interval: Interval,
|
||||
}
|
||||
|
||||
/// Wire-format representation of an incoming Binance kline tick. Public so callers
|
||||
/// can deserialize it themselves if they prefer.
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct RawWsEnvelope {
|
||||
/// Stream name, e.g. `"btcusdt@kline_1m"`.
|
||||
pub stream: String,
|
||||
pub data: RawKlinePayload,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct RawKlinePayload {
|
||||
#[serde(rename = "e")]
|
||||
pub event_type: String,
|
||||
#[serde(rename = "E")]
|
||||
pub event_time: i64,
|
||||
#[serde(rename = "s")]
|
||||
pub symbol: String,
|
||||
#[serde(rename = "k")]
|
||||
pub kline: RawKline,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct RawKline {
|
||||
#[serde(rename = "t")]
|
||||
pub open_time: i64,
|
||||
#[serde(rename = "T")]
|
||||
pub close_time: i64,
|
||||
#[serde(rename = "s")]
|
||||
pub symbol: String,
|
||||
#[serde(rename = "i")]
|
||||
pub interval: String,
|
||||
#[serde(rename = "o")]
|
||||
pub open: String,
|
||||
#[serde(rename = "c")]
|
||||
pub close: String,
|
||||
#[serde(rename = "h")]
|
||||
pub high: String,
|
||||
#[serde(rename = "l")]
|
||||
pub low: String,
|
||||
#[serde(rename = "v")]
|
||||
pub volume: String,
|
||||
#[serde(rename = "x")]
|
||||
pub is_closed: bool,
|
||||
}
|
||||
|
||||
impl BinanceKlineStream {
|
||||
/// Connect to Binance's combined-stream endpoint for one or more symbols.
|
||||
///
|
||||
/// Symbols may be passed in either case; they are lowercased to match
|
||||
/// Binance's stream-name conventions.
|
||||
pub async fn connect(symbols: &[String], interval: Interval) -> Result<Self> {
|
||||
if symbols.is_empty() {
|
||||
return Err(Error::Malformed(
|
||||
"BinanceKlineStream requires at least one symbol".into(),
|
||||
));
|
||||
}
|
||||
let streams: Vec<String> = symbols
|
||||
.iter()
|
||||
.map(|s| format!("{}@kline_{}", s.to_lowercase(), interval.as_str()))
|
||||
.collect();
|
||||
let url = format!(
|
||||
"wss://stream.binance.com:9443/stream?streams={}",
|
||||
streams.join("/")
|
||||
);
|
||||
let url = url::Url::parse(&url).map_err(|e| Error::Malformed(e.to_string()))?;
|
||||
let (socket, _) = tokio_tungstenite::connect_async(url.as_str()).await?;
|
||||
Ok(Self { socket, interval })
|
||||
}
|
||||
|
||||
/// Receive the next kline event. Yields `Ok(None)` when the server closes
|
||||
/// the connection cleanly.
|
||||
pub async fn next_event(&mut self) -> Result<Option<KlineEvent>> {
|
||||
loop {
|
||||
let msg = match self.socket.next().await {
|
||||
Some(Ok(m)) => m,
|
||||
Some(Err(e)) => return Err(Error::from(e)),
|
||||
None => return Ok(None),
|
||||
};
|
||||
match msg {
|
||||
Message::Text(text) => {
|
||||
let envelope: RawWsEnvelope = serde_json::from_str(&text)?;
|
||||
return Ok(Some(envelope.into_event(self.interval)?));
|
||||
}
|
||||
Message::Binary(bytes) => {
|
||||
let envelope: RawWsEnvelope = serde_json::from_slice(&bytes)?;
|
||||
return Ok(Some(envelope.into_event(self.interval)?));
|
||||
}
|
||||
Message::Ping(payload) => {
|
||||
self.socket.send(Message::Pong(payload)).await?;
|
||||
}
|
||||
Message::Pong(_) | Message::Frame(_) => {}
|
||||
Message::Close(_) => return Ok(None),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Close the underlying socket cleanly.
|
||||
pub async fn close(mut self) -> Result<()> {
|
||||
self.socket.close(None).await?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl RawWsEnvelope {
|
||||
fn into_event(self, interval: Interval) -> Result<KlineEvent> {
|
||||
let k = self.data.kline;
|
||||
let open: f64 = k
|
||||
.open
|
||||
.parse()
|
||||
.map_err(|_| Error::Malformed(format!("bad open '{}'", k.open)))?;
|
||||
let high: f64 = k
|
||||
.high
|
||||
.parse()
|
||||
.map_err(|_| Error::Malformed(format!("bad high '{}'", k.high)))?;
|
||||
let low: f64 = k
|
||||
.low
|
||||
.parse()
|
||||
.map_err(|_| Error::Malformed(format!("bad low '{}'", k.low)))?;
|
||||
let close: f64 = k
|
||||
.close
|
||||
.parse()
|
||||
.map_err(|_| Error::Malformed(format!("bad close '{}'", k.close)))?;
|
||||
let volume: f64 = k
|
||||
.volume
|
||||
.parse()
|
||||
.map_err(|_| Error::Malformed(format!("bad volume '{}'", k.volume)))?;
|
||||
let candle = Candle::new(open, high, low, close, volume, k.open_time)?;
|
||||
Ok(KlineEvent {
|
||||
symbol: self.data.symbol.to_lowercase(),
|
||||
interval,
|
||||
candle,
|
||||
is_closed: k.is_closed,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn parses_real_binance_payload() {
|
||||
// Sample event format from Binance's public docs (truncated).
|
||||
let json = r#"{
|
||||
"stream": "btcusdt@kline_1m",
|
||||
"data": {
|
||||
"e": "kline",
|
||||
"E": 1700000000000,
|
||||
"s": "BTCUSDT",
|
||||
"k": {
|
||||
"t": 1700000000000,
|
||||
"T": 1700000059999,
|
||||
"s": "BTCUSDT",
|
||||
"i": "1m",
|
||||
"f": 1,
|
||||
"L": 100,
|
||||
"o": "30000.0",
|
||||
"c": "30050.0",
|
||||
"h": "30100.0",
|
||||
"l": "29950.0",
|
||||
"v": "12.5",
|
||||
"n": 50,
|
||||
"x": false,
|
||||
"q": "375000.0",
|
||||
"V": "6.25",
|
||||
"Q": "187500.0",
|
||||
"B": "0"
|
||||
}
|
||||
}
|
||||
}"#;
|
||||
let env: RawWsEnvelope = serde_json::from_str(json).unwrap();
|
||||
let evt = env.into_event(Interval::OneMinute).unwrap();
|
||||
assert_eq!(evt.symbol, "btcusdt");
|
||||
assert_eq!(evt.candle.open, 30_000.0);
|
||||
assert_eq!(evt.candle.close, 30_050.0);
|
||||
assert!(!evt.is_closed);
|
||||
assert_eq!(evt.interval, Interval::OneMinute);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_non_parsable_numbers() {
|
||||
let json = r#"{
|
||||
"stream": "btcusdt@kline_1m",
|
||||
"data": {
|
||||
"e": "kline", "E": 0, "s": "BTCUSDT",
|
||||
"k": {
|
||||
"t": 0, "T": 0, "s": "BTCUSDT", "i": "1m",
|
||||
"f": 0, "L": 0,
|
||||
"o": "not-a-number", "c": "0", "h": "0", "l": "0",
|
||||
"v": "0", "n": 0, "x": false, "q": "0", "V": "0", "Q": "0", "B": "0"
|
||||
}
|
||||
}
|
||||
}"#;
|
||||
let env: RawWsEnvelope = serde_json::from_str(json).unwrap();
|
||||
let err = env.into_event(Interval::OneMinute).unwrap_err();
|
||||
assert!(matches!(err, Error::Malformed(_)));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
//! Resample an existing candle stream from a finer timeframe to a coarser one.
|
||||
|
||||
use crate::aggregator::Timeframe;
|
||||
use crate::error::Result;
|
||||
use wickra_core::Candle;
|
||||
|
||||
/// Roll a stream of candles up to a coarser timeframe.
|
||||
///
|
||||
/// Used to derive 5m bars from a 1m feed, or 1h bars from 5m bars, without
|
||||
/// touching the original tick stream. The output timeframe's bucket must be a
|
||||
/// strict multiple of the input timeframe's bucket, but this is not enforced
|
||||
/// — callers are responsible for picking sensible aggregations.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Resampler {
|
||||
timeframe: Timeframe,
|
||||
open: Option<RolledBar>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
struct RolledBar {
|
||||
bucket_start: i64,
|
||||
open: f64,
|
||||
high: f64,
|
||||
low: f64,
|
||||
close: f64,
|
||||
volume: f64,
|
||||
}
|
||||
|
||||
impl RolledBar {
|
||||
fn from_candle(c: Candle, bucket_start: i64) -> Self {
|
||||
Self {
|
||||
bucket_start,
|
||||
open: c.open,
|
||||
high: c.high,
|
||||
low: c.low,
|
||||
close: c.close,
|
||||
volume: c.volume,
|
||||
}
|
||||
}
|
||||
|
||||
fn absorb(&mut self, c: Candle) {
|
||||
if c.high > self.high {
|
||||
self.high = c.high;
|
||||
}
|
||||
if c.low < self.low {
|
||||
self.low = c.low;
|
||||
}
|
||||
self.close = c.close;
|
||||
self.volume += c.volume;
|
||||
}
|
||||
|
||||
fn into_candle(self) -> Candle {
|
||||
Candle::new_unchecked(
|
||||
self.open,
|
||||
self.high,
|
||||
self.low,
|
||||
self.close,
|
||||
self.volume,
|
||||
self.bucket_start,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl Resampler {
|
||||
/// Build a resampler targeting the given output timeframe.
|
||||
pub fn new(timeframe: Timeframe) -> Self {
|
||||
Self {
|
||||
timeframe,
|
||||
open: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Push a finer-grained candle. Returns the coarser candle that just closed,
|
||||
/// if any.
|
||||
pub fn push(&mut self, candle: Candle) -> Option<Candle> {
|
||||
let bucket = self.timeframe.floor(candle.timestamp);
|
||||
match self.open {
|
||||
Some(mut bar) if bucket == bar.bucket_start => {
|
||||
bar.absorb(candle);
|
||||
self.open = Some(bar);
|
||||
None
|
||||
}
|
||||
Some(bar) => {
|
||||
let closed = bar.into_candle();
|
||||
self.open = Some(RolledBar::from_candle(candle, bucket));
|
||||
Some(closed)
|
||||
}
|
||||
None => {
|
||||
self.open = Some(RolledBar::from_candle(candle, bucket));
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Flush the currently open coarser bar, if any.
|
||||
pub fn flush(&mut self) -> Option<Candle> {
|
||||
self.open.take().map(RolledBar::into_candle)
|
||||
}
|
||||
}
|
||||
|
||||
/// Roll an entire iterator of candles into a `Vec` of coarser candles. The final
|
||||
/// open bar (if any) is appended via [`Resampler::flush`].
|
||||
pub fn resample_all<I>(timeframe: Timeframe, iter: I) -> Result<Vec<Candle>>
|
||||
where
|
||||
I: IntoIterator<Item = Result<Candle>>,
|
||||
{
|
||||
let mut r = Resampler::new(timeframe);
|
||||
let mut out = Vec::new();
|
||||
for c in iter {
|
||||
let c = c?;
|
||||
if let Some(closed) = r.push(c) {
|
||||
out.push(closed);
|
||||
}
|
||||
}
|
||||
if let Some(last) = r.flush() {
|
||||
out.push(last);
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn c(ts: i64, o: f64, h: f64, l: f64, cl: f64, v: f64) -> Candle {
|
||||
Candle::new(o, h, l, cl, v, ts).unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resamples_1m_to_5m() {
|
||||
let tf = Timeframe::new(5).unwrap();
|
||||
let one_m = vec![
|
||||
c(0, 10.0, 11.0, 9.0, 10.5, 10.0),
|
||||
c(1, 10.5, 12.0, 10.0, 11.5, 12.0),
|
||||
c(2, 11.5, 13.0, 11.0, 12.5, 15.0),
|
||||
c(3, 12.5, 12.8, 11.5, 12.0, 8.0),
|
||||
c(4, 12.0, 12.2, 11.0, 11.5, 6.0),
|
||||
c(5, 11.5, 11.9, 11.0, 11.5, 4.0),
|
||||
];
|
||||
let rolled = resample_all(tf, one_m.into_iter().map(Ok)).unwrap();
|
||||
// First 5 candles share bucket 0 -> aggregate. Last candle opens bucket 5.
|
||||
assert_eq!(rolled.len(), 2);
|
||||
let a = rolled[0];
|
||||
assert_eq!(a.open, 10.0);
|
||||
assert_eq!(a.close, 11.5);
|
||||
assert_eq!(a.high, 13.0);
|
||||
assert_eq!(a.low, 9.0);
|
||||
assert!((a.volume - 51.0).abs() < 1e-12);
|
||||
let b = rolled[1];
|
||||
assert_eq!(b.open, 11.5);
|
||||
assert_eq!(b.timestamp, 5);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user