feat(data): native live Binance kline feed in 9 languages (+ 3d/1M intervals) (#313)
* feat(data): C-ABI Binance feed + Go binding (F4 wip)
C ABI exposes the existing async BinanceKlineStream (tokio + TLS, auto-reconnect,
mock-server-tested in wickra-data) through a blocking poll: wickra_binance_connect
/ _next(out, timeout_ms) -> {1 event, 0 timeout, -1 closed} / _close / _free over
an opaque BinanceStream that owns a current-thread runtime. WickraKlineEvent
carries OHLCV + open_time + is_closed + a 16-byte symbol buffer. `live-binance`
is now a default feature of wickra-c (the published DLL ships the feed; the wasm
build drops it via --no-default-features).
Go: NewBinanceFeed(symbols, interval, baseURL) + Next(timeout) + Close, with a
deterministic error-path smoke (the connect->event pipeline is covered by the
Rust mock-WS-server tests).
* feat(data): Binance feed C# + Java bindings (F4 wip)
C#: BinanceFeed(symbols, interval, baseUrl?) + Next(timeout) -> KlineEvent? +
Dispose; bespoke WickraKlineEvent native struct (fixed symbol buffer + byte
is_closed) since the scalar struct parser can't model it. `char` maps to `byte`
for the const char* params.
Java: BinanceFeed + KlineEvent record + BinanceInterval enum over Panama FFM;
the event is read at hand-computed offsets (symbol@0, doubles@16..48,
open_time@56, is_closed@64; 72-byte struct). Both with deterministic error-path
smokes (pipeline covered by the Rust mock-WS-server tests).
* feat(data): Binance feed R binding (F4 wip)
R: BinanceFeed(symbols, interval, base_url) + binance_next(feed, timeout_ms) ->
named list | NULL + binance_close, via bespoke .Call glue (wk_binance_*). The
glue + its registration entries are gated out of the Emscripten/wasm build
(#ifndef __EMSCRIPTEN__) since r-universe/webR has no raw sockets. NAMESPACE
exports added by hand (roxygen2 not installed locally). Deterministic error-path
smoke; pipeline covered by the Rust mock-WS-server tests.
* feat(data): native Binance feed for Node + Python; CHANGELOG (F4 complete)
Node (napi) BinanceFeed: new(symbols, interval, baseUrl?) + next(timeoutMs) ->
KlineEvent | null + close. Python (pyo3) BinanceFeed: same, with next releasing
the GIL (py.detach) while it waits. Both drive the mock-server-tested async
BinanceKlineStream on a single-thread tokio runtime (blocking poll); wickra-data
gains the live-binance feature + tokio in each binding.
Completes F4: the live Binance kline feed is now native in all 9 languages
(WASM excluded), with no third-party WebSocket client in any of them.
This commit is contained in:
kingchenc
@@ -14,6 +14,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
and map to the correct wire-format strings.
|
||||
|
||||
### Added
|
||||
- **Native live Binance kline feed in 9 languages (data layer).** `BinanceFeed`
|
||||
streams live OHLCV candles straight from Binance's public WebSocket — no
|
||||
third-party WebSocket client (`ws`, `websockets`, `gorilla/websocket`, …) in any
|
||||
binding. Construct it with comma-separated symbols + an interval, then poll
|
||||
`next(timeout)` for the next event (or `null`/`None` on timeout); the connection
|
||||
reconnects transparently. Exposed natively (Node.js / Python — a blocking poll
|
||||
that drives the tested async stream on a tokio runtime) and over the C ABI as Go
|
||||
`Next()`, C# `Next()`, Java `next()`, and the R `binance_next()`; C / C++ call
|
||||
`wickra_binance_connect` / `_next` / `_close` / `_free` directly. The connect →
|
||||
read → reconnect pipeline is covered by the existing mock-WS-server tests. WASM
|
||||
is excluded (a browser has no raw sockets; use the host `WebSocket`). The C ABI
|
||||
ships the feed by default (`live-binance` feature); the wasm build drops it.
|
||||
- **CSV candle reading in all 10 languages (data layer).** The `CandleReader`
|
||||
parses a `timestamp,open,high,low,close,volume` CSV buffer (a leading UTF-8 BOM
|
||||
and field whitespace are tolerated) into candles: construct it from a CSV string
|
||||
|
||||
Generated
+3
@@ -1968,6 +1968,7 @@ dependencies = [
|
||||
name = "wickra-c"
|
||||
version = "0.9.2"
|
||||
dependencies = [
|
||||
"tokio",
|
||||
"wickra-core",
|
||||
"wickra-data",
|
||||
]
|
||||
@@ -2016,6 +2017,7 @@ dependencies = [
|
||||
"napi",
|
||||
"napi-build",
|
||||
"napi-derive",
|
||||
"tokio",
|
||||
"wickra-core",
|
||||
"wickra-data",
|
||||
]
|
||||
@@ -2026,6 +2028,7 @@ version = "0.9.2"
|
||||
dependencies = [
|
||||
"numpy",
|
||||
"pyo3",
|
||||
"tokio",
|
||||
"wickra-core",
|
||||
"wickra-data",
|
||||
]
|
||||
|
||||
+13
-1
@@ -48,8 +48,16 @@ float_cmp = "allow"
|
||||
# package's wasm build (r-universe / webR) compiles this crate with
|
||||
# --no-default-features to drop rayon; wickra-core falls back to its serial
|
||||
# batch path, which is cfg-gated behind the same feature.
|
||||
default = ["parallel"]
|
||||
#
|
||||
# `live-binance` is on by default too so the published C ABI ships the native
|
||||
# Binance kline feed (tokio + TLS WebSocket) — the six C-ABI languages with no
|
||||
# native WebSocket (C, C++, Go, R, plus Node/Python via their own bindings) get
|
||||
# it without a third-party client. The wasm build drops it via
|
||||
# --no-default-features (a browser has no raw TCP/TLS sockets; WASM users use the
|
||||
# host `WebSocket`), exactly like rayon.
|
||||
default = ["parallel", "live-binance"]
|
||||
parallel = ["wickra-core/parallel"]
|
||||
live-binance = ["wickra-data/live-binance", "dep:tokio"]
|
||||
|
||||
[dependencies]
|
||||
# Direct path dep rather than `workspace = true`: a member-level
|
||||
@@ -58,3 +66,7 @@ parallel = ["wickra-core/parallel"]
|
||||
# `publish = false`, so no version pin is needed (and none to keep in sync).
|
||||
wickra-core = { path = "../../crates/wickra-core", default-features = false }
|
||||
wickra-data = { path = "../../crates/wickra-data" }
|
||||
# Only pulled in by `live-binance`: a single-thread runtime drives the async
|
||||
# Binance feed behind the blocking C-ABI poll. The feed's own I/O features come
|
||||
# transitively from wickra-data; this just needs the runtime + timeout.
|
||||
tokio = { version = "1", features = ["rt", "net", "time"], optional = true }
|
||||
|
||||
@@ -102,6 +102,8 @@ typedef struct BetaNeutralSpread BetaNeutralSpread;
|
||||
|
||||
typedef struct BetterVolume BetterVolume;
|
||||
|
||||
typedef struct BinanceStream BinanceStream;
|
||||
|
||||
typedef struct BipowerVariation BipowerVariation;
|
||||
|
||||
typedef struct BodySizePct BodySizePct;
|
||||
@@ -1690,6 +1692,17 @@ typedef struct WickraCandle {
|
||||
int64_t timestamp;
|
||||
} WickraCandle;
|
||||
|
||||
typedef struct WickraKlineEvent {
|
||||
uint8_t symbol[16];
|
||||
double open;
|
||||
double high;
|
||||
double low;
|
||||
double close;
|
||||
double volume;
|
||||
int64_t open_time;
|
||||
bool is_closed;
|
||||
} WickraKlineEvent;
|
||||
|
||||
#ifdef __cplusplus
|
||||
extern "C" {
|
||||
#endif // __cplusplus
|
||||
@@ -13748,6 +13761,18 @@ uintptr_t wickra_candle_reader_read(struct CandleReader *handle,
|
||||
|
||||
void wickra_candle_reader_free(struct CandleReader *handle);
|
||||
|
||||
struct BinanceStream *wickra_binance_connect(const char *symbols,
|
||||
uint8_t interval,
|
||||
const char *base_url);
|
||||
|
||||
int32_t wickra_binance_next(struct BinanceStream *handle,
|
||||
struct WickraKlineEvent *out,
|
||||
int64_t timeout_ms);
|
||||
|
||||
void wickra_binance_close(struct BinanceStream *handle);
|
||||
|
||||
void wickra_binance_free(struct BinanceStream *handle);
|
||||
|
||||
#ifdef __cplusplus
|
||||
} // extern "C"
|
||||
#endif // __cplusplus
|
||||
|
||||
@@ -68379,6 +68379,199 @@ pub unsafe extern "C" fn wickra_candle_reader_free(handle: *mut CandleReader) {
|
||||
}
|
||||
}
|
||||
|
||||
// ===== Live Binance kline feed (feature `live-binance`; blocking poll) =====
|
||||
|
||||
/// C-ABI view of one kline event from the live Binance feed.
|
||||
#[cfg(feature = "live-binance")]
|
||||
#[repr(C)]
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
pub struct WickraKlineEvent {
|
||||
/// NUL-terminated lowercase symbol (e.g. `b"btcusdt\0"`), truncated to 15 bytes.
|
||||
pub symbol: [u8; 16],
|
||||
pub open: f64,
|
||||
pub high: f64,
|
||||
pub low: f64,
|
||||
pub close: f64,
|
||||
pub volume: f64,
|
||||
/// Candle open time as Binance sends it (milliseconds since the Unix epoch).
|
||||
pub open_time: i64,
|
||||
/// Whether the candle is closed; only closed candles are final.
|
||||
pub is_closed: bool,
|
||||
}
|
||||
|
||||
/// Opaque live Binance kline stream: the async stream plus a single-thread tokio
|
||||
/// runtime that drives it. Each poll runs one async step to completion.
|
||||
#[cfg(feature = "live-binance")]
|
||||
#[derive(Debug)]
|
||||
pub struct BinanceStream {
|
||||
runtime: tokio::runtime::Runtime,
|
||||
inner: wickra_data::live::binance::BinanceKlineStream,
|
||||
}
|
||||
|
||||
/// Map a `u8` interval code (the `Interval` declaration order, `0..=15`) to the
|
||||
/// feed's interval. Returns `None` for an unknown code.
|
||||
#[cfg(feature = "live-binance")]
|
||||
fn binance_interval(code: u8) -> Option<wickra_data::live::binance::Interval> {
|
||||
use wickra_data::live::binance::Interval;
|
||||
Some(match code {
|
||||
0 => Interval::OneSecond,
|
||||
1 => Interval::OneMinute,
|
||||
2 => Interval::ThreeMinutes,
|
||||
3 => Interval::FiveMinutes,
|
||||
4 => Interval::FifteenMinutes,
|
||||
5 => Interval::ThirtyMinutes,
|
||||
6 => Interval::OneHour,
|
||||
7 => Interval::TwoHours,
|
||||
8 => Interval::FourHours,
|
||||
9 => Interval::SixHours,
|
||||
10 => Interval::EightHours,
|
||||
11 => Interval::TwelveHours,
|
||||
12 => Interval::OneDay,
|
||||
13 => Interval::ThreeDays,
|
||||
14 => Interval::OneWeek,
|
||||
15 => Interval::OneMonth,
|
||||
_ => return None,
|
||||
})
|
||||
}
|
||||
|
||||
/// Connect to Binance's live kline stream for one or more comma-separated
|
||||
/// `symbols` (case-insensitive, e.g. `"BTCUSDT,ETHUSDT"`) at the given `interval`
|
||||
/// code (`0..=15`, the `Interval` declaration order). `base_url` overrides the
|
||||
/// endpoint (`NULL` = production `wss://stream.binance.com:9443`); pass a `ws://…`
|
||||
/// URL to target a test server. Returns `NULL` on a null/empty/invalid symbol
|
||||
/// list, an unknown interval, a bad URL, or a failed initial connect. Release
|
||||
/// with `wickra_binance_free`.
|
||||
///
|
||||
/// # Safety
|
||||
/// `symbols` must be a valid NUL-terminated UTF-8 C string; `base_url` must be
|
||||
/// `NULL` or a valid NUL-terminated UTF-8 C string.
|
||||
#[cfg(feature = "live-binance")]
|
||||
#[no_mangle]
|
||||
pub unsafe extern "C" fn wickra_binance_connect(
|
||||
symbols: *const c_char,
|
||||
interval: u8,
|
||||
base_url: *const c_char,
|
||||
) -> *mut BinanceStream {
|
||||
if symbols.is_null() {
|
||||
return ptr::null_mut();
|
||||
}
|
||||
let Ok(symbols_str) = core::ffi::CStr::from_ptr(symbols).to_str() else {
|
||||
return ptr::null_mut();
|
||||
};
|
||||
let symbol_list: Vec<String> = symbols_str
|
||||
.split(',')
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty())
|
||||
.map(str::to_owned)
|
||||
.collect();
|
||||
if symbol_list.is_empty() {
|
||||
return ptr::null_mut();
|
||||
}
|
||||
let Some(interval) = binance_interval(interval) else {
|
||||
return ptr::null_mut();
|
||||
};
|
||||
let mut config = wickra_data::live::binance::BinanceConfig::default();
|
||||
if !base_url.is_null() {
|
||||
let Ok(url) = core::ffi::CStr::from_ptr(base_url).to_str() else {
|
||||
return ptr::null_mut();
|
||||
};
|
||||
url.clone_into(&mut config.base_url);
|
||||
}
|
||||
let Ok(runtime) = tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
else {
|
||||
return ptr::null_mut();
|
||||
};
|
||||
match runtime.block_on(
|
||||
wickra_data::live::binance::BinanceKlineStream::connect_with_config(
|
||||
&symbol_list,
|
||||
interval,
|
||||
config,
|
||||
),
|
||||
) {
|
||||
Ok(inner) => Box::into_raw(Box::new(BinanceStream { runtime, inner })),
|
||||
Err(_) => ptr::null_mut(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Poll for the next kline event, waiting up to `timeout_ms` milliseconds.
|
||||
/// Returns `1` and writes the event to `*out` when one arrives, `0` on timeout
|
||||
/// (no event yet — call again), or `-1` if the stream is closed/errored or
|
||||
/// `handle`/`out` is `NULL`. A dropped connection is reconnected transparently
|
||||
/// within a single call (which may consume part of the timeout).
|
||||
///
|
||||
/// # Safety
|
||||
/// `handle` (from `wickra_binance_connect`, not freed) and `out` must be valid or `NULL`.
|
||||
#[cfg(feature = "live-binance")]
|
||||
#[no_mangle]
|
||||
pub unsafe extern "C" fn wickra_binance_next(
|
||||
handle: *mut BinanceStream,
|
||||
out: *mut WickraKlineEvent,
|
||||
timeout_ms: i64,
|
||||
) -> i32 {
|
||||
let Some(stream) = handle.as_mut() else {
|
||||
return -1;
|
||||
};
|
||||
if out.is_null() {
|
||||
return -1;
|
||||
}
|
||||
let timeout = std::time::Duration::from_millis(timeout_ms.max(0) as u64);
|
||||
let runtime = &stream.runtime;
|
||||
let inner = &mut stream.inner;
|
||||
let result =
|
||||
runtime.block_on(async { tokio::time::timeout(timeout, inner.next_event()).await });
|
||||
match result {
|
||||
Ok(Ok(Some(event))) => {
|
||||
let mut symbol = [0_u8; 16];
|
||||
for (slot, byte) in symbol.iter_mut().zip(event.symbol.bytes()).take(15) {
|
||||
*slot = byte;
|
||||
}
|
||||
*out = WickraKlineEvent {
|
||||
symbol,
|
||||
open: event.candle.open,
|
||||
high: event.candle.high,
|
||||
low: event.candle.low,
|
||||
close: event.candle.close,
|
||||
volume: event.candle.volume,
|
||||
open_time: event.candle.timestamp,
|
||||
is_closed: event.is_closed,
|
||||
};
|
||||
1
|
||||
}
|
||||
// Closed stream or any transport/parse error: both terminal for the poll.
|
||||
Ok(Ok(None) | Err(_)) => -1,
|
||||
Err(_) => 0,
|
||||
}
|
||||
}
|
||||
|
||||
/// Close the stream's connection; afterwards every `wickra_binance_next` returns
|
||||
/// `-1`. No-op on a `NULL` handle.
|
||||
///
|
||||
/// # Safety
|
||||
/// `handle` must be valid (from `wickra_binance_connect`, not freed), or `NULL`.
|
||||
#[cfg(feature = "live-binance")]
|
||||
#[no_mangle]
|
||||
pub unsafe extern "C" fn wickra_binance_close(handle: *mut BinanceStream) {
|
||||
if let Some(stream) = handle.as_mut() {
|
||||
let runtime = &stream.runtime;
|
||||
let inner = &mut stream.inner;
|
||||
let _ = runtime.block_on(inner.close());
|
||||
}
|
||||
}
|
||||
|
||||
/// Destroy a stream created by `wickra_binance_connect`. No-op if `handle` is `NULL`.
|
||||
///
|
||||
/// # Safety
|
||||
/// `handle` must have been returned by `wickra_binance_connect` and not previously freed, or `NULL`.
|
||||
#[cfg(feature = "live-binance")]
|
||||
#[no_mangle]
|
||||
pub unsafe extern "C" fn wickra_binance_free(handle: *mut BinanceStream) {
|
||||
if !handle.is_null() {
|
||||
drop(Box::from_raw(handle));
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
using System;
|
||||
using Wickra;
|
||||
using Xunit;
|
||||
|
||||
// The live Binance feed's connect → read → reconnect pipeline is covered
|
||||
// deterministically by the Rust mock-WS-server tests in wickra-data. Here we
|
||||
// only assert the binding's error paths, which need no network.
|
||||
public class BinanceFeedTests
|
||||
{
|
||||
[Fact]
|
||||
public void RejectsUnknownInterval()
|
||||
{
|
||||
Assert.Throws<ArgumentException>(() =>
|
||||
new BinanceFeed("BTCUSDT", (BinanceInterval)99));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void RejectsEmptySymbols()
|
||||
{
|
||||
Assert.Throws<ArgumentException>(() =>
|
||||
new BinanceFeed("", BinanceInterval.OneMinute));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void RejectsUnreachableEndpoint()
|
||||
{
|
||||
Assert.Throws<ArgumentException>(() =>
|
||||
new BinanceFeed("BTCUSDT", BinanceInterval.OneMinute, "ws://127.0.0.1:1"));
|
||||
}
|
||||
}
|
||||
@@ -41508,3 +41508,95 @@ public sealed class Zlema : IDisposable
|
||||
public void Dispose() => _handle.Dispose();
|
||||
}
|
||||
|
||||
/// <summary>Kline interval for the live Binance feed.</summary>
|
||||
public enum BinanceInterval : byte
|
||||
{
|
||||
OneSecond,
|
||||
OneMinute,
|
||||
ThreeMinutes,
|
||||
FiveMinutes,
|
||||
FifteenMinutes,
|
||||
ThirtyMinutes,
|
||||
OneHour,
|
||||
TwoHours,
|
||||
FourHours,
|
||||
SixHours,
|
||||
EightHours,
|
||||
TwelveHours,
|
||||
OneDay,
|
||||
ThreeDays,
|
||||
OneWeek,
|
||||
OneMonth,
|
||||
}
|
||||
|
||||
/// <summary>One event from the live Binance feed.</summary>
|
||||
public readonly record struct KlineEvent(string Symbol, double Open, double High, double Low, double Close, double Volume, long OpenTime, bool IsClosed);
|
||||
|
||||
/// <summary>A live Binance kline stream over the Wickra C ABI.</summary>
|
||||
public sealed class BinanceFeed : IDisposable
|
||||
{
|
||||
private readonly WickraHandle _handle;
|
||||
|
||||
/// <summary>Connect to Binance's live kline stream for the given comma-separated symbols
|
||||
/// (case-insensitive) at <paramref name="interval"/>. <paramref name="baseUrl"/> overrides the
|
||||
/// endpoint (null = production; pass a ws:// URL to target a test server).</summary>
|
||||
public BinanceFeed(string symbols, BinanceInterval interval, string? baseUrl = null)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(symbols);
|
||||
var symBytes = System.Text.Encoding.UTF8.GetBytes(symbols + '\0');
|
||||
var urlBytes = baseUrl is null ? null : System.Text.Encoding.UTF8.GetBytes(baseUrl + '\0');
|
||||
nint ptr;
|
||||
unsafe
|
||||
{
|
||||
fixed (byte* sp = symBytes)
|
||||
fixed (byte* up = urlBytes)
|
||||
{
|
||||
ptr = NativeMethods.wickra_binance_connect(sp, (byte)interval, up);
|
||||
}
|
||||
}
|
||||
if (ptr == nint.Zero)
|
||||
{
|
||||
throw new ArgumentException("invalid BinanceFeed parameters");
|
||||
}
|
||||
_handle = new WickraHandle(ptr, NativeMethods.wickra_binance_free);
|
||||
}
|
||||
|
||||
/// <summary>Poll for the next kline event, waiting up to <paramref name="timeout"/>.
|
||||
/// Returns the event, or null on timeout (call again). Throws once the stream is
|
||||
/// closed or has errored out.</summary>
|
||||
public KlineEvent? Next(TimeSpan timeout)
|
||||
{
|
||||
WickraKlineEvent ev;
|
||||
int code;
|
||||
unsafe
|
||||
{
|
||||
code = NativeMethods.wickra_binance_next(_handle.DangerousGetHandle(), &ev, (long)timeout.TotalMilliseconds);
|
||||
}
|
||||
GC.KeepAlive(_handle);
|
||||
if (code == 0)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
if (code != 1)
|
||||
{
|
||||
throw new InvalidOperationException("binance feed closed");
|
||||
}
|
||||
string symbol;
|
||||
unsafe
|
||||
{
|
||||
symbol = Marshal.PtrToStringUTF8((nint)ev.symbol) ?? string.Empty;
|
||||
}
|
||||
return new KlineEvent(symbol, ev.open, ev.high, ev.low, ev.close, ev.volume, ev.open_time, ev.is_closed != 0);
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
unsafe
|
||||
{
|
||||
NativeMethods.wickra_binance_close(_handle.DangerousGetHandle());
|
||||
}
|
||||
GC.KeepAlive(_handle);
|
||||
_handle.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -12442,6 +12442,18 @@ internal static partial class NativeMethods
|
||||
[LibraryImport(WickraNative.LibraryName)]
|
||||
internal static partial void wickra_candle_reader_free(nint handle);
|
||||
|
||||
[LibraryImport(WickraNative.LibraryName)]
|
||||
internal static unsafe partial nint wickra_binance_connect(byte* symbols, byte interval, byte* baseUrl);
|
||||
|
||||
[LibraryImport(WickraNative.LibraryName)]
|
||||
internal static unsafe partial int wickra_binance_next(nint handle, WickraKlineEvent* @out, long timeoutMs);
|
||||
|
||||
[LibraryImport(WickraNative.LibraryName)]
|
||||
internal static partial void wickra_binance_close(nint handle);
|
||||
|
||||
[LibraryImport(WickraNative.LibraryName)]
|
||||
internal static partial void wickra_binance_free(nint handle);
|
||||
|
||||
}
|
||||
|
||||
[StructLayout(LayoutKind.Sequential)]
|
||||
@@ -13298,3 +13310,16 @@ internal static partial class NativeMethods
|
||||
public double direction;
|
||||
}
|
||||
|
||||
[StructLayout(LayoutKind.Sequential)]
|
||||
internal unsafe struct WickraKlineEvent
|
||||
{
|
||||
public fixed byte symbol[16];
|
||||
public double open;
|
||||
public double high;
|
||||
public double low;
|
||||
public double close;
|
||||
public double volume;
|
||||
public long open_time;
|
||||
public byte is_closed;
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
package wickra
|
||||
|
||||
import "testing"
|
||||
|
||||
// The live Binance feed's connect → read → reconnect pipeline is covered
|
||||
// deterministically by the Rust mock-WS-server tests in wickra-data. Here we
|
||||
// only assert the binding's error paths, which need no network: a bad interval,
|
||||
// an empty symbol list, and an unreachable endpoint must all surface as an
|
||||
// error rather than a usable handle.
|
||||
func TestBinanceFeedRejectsBadParams(t *testing.T) {
|
||||
if _, err := NewBinanceFeed("BTCUSDT", BinanceInterval(99), ""); err == nil {
|
||||
t.Fatal("expected an error for an unknown interval code")
|
||||
}
|
||||
if _, err := NewBinanceFeed("", OneMinute, ""); err == nil {
|
||||
t.Fatal("expected an error for an empty symbol list")
|
||||
}
|
||||
if _, err := NewBinanceFeed("BTCUSDT", OneMinute, "ws://127.0.0.1:1"); err == nil {
|
||||
t.Fatal("expected an error connecting to an unreachable endpoint")
|
||||
}
|
||||
}
|
||||
@@ -102,6 +102,8 @@ typedef struct BetaNeutralSpread BetaNeutralSpread;
|
||||
|
||||
typedef struct BetterVolume BetterVolume;
|
||||
|
||||
typedef struct BinanceStream BinanceStream;
|
||||
|
||||
typedef struct BipowerVariation BipowerVariation;
|
||||
|
||||
typedef struct BodySizePct BodySizePct;
|
||||
@@ -1690,6 +1692,17 @@ typedef struct WickraCandle {
|
||||
int64_t timestamp;
|
||||
} WickraCandle;
|
||||
|
||||
typedef struct WickraKlineEvent {
|
||||
uint8_t symbol[16];
|
||||
double open;
|
||||
double high;
|
||||
double low;
|
||||
double close;
|
||||
double volume;
|
||||
int64_t open_time;
|
||||
bool is_closed;
|
||||
} WickraKlineEvent;
|
||||
|
||||
#ifdef __cplusplus
|
||||
extern "C" {
|
||||
#endif // __cplusplus
|
||||
@@ -13748,6 +13761,18 @@ uintptr_t wickra_candle_reader_read(struct CandleReader *handle,
|
||||
|
||||
void wickra_candle_reader_free(struct CandleReader *handle);
|
||||
|
||||
struct BinanceStream *wickra_binance_connect(const char *symbols,
|
||||
uint8_t interval,
|
||||
const char *base_url);
|
||||
|
||||
int32_t wickra_binance_next(struct BinanceStream *handle,
|
||||
struct WickraKlineEvent *out,
|
||||
int64_t timeout_ms);
|
||||
|
||||
void wickra_binance_close(struct BinanceStream *handle);
|
||||
|
||||
void wickra_binance_free(struct BinanceStream *handle);
|
||||
|
||||
#ifdef __cplusplus
|
||||
} // extern "C"
|
||||
#endif // __cplusplus
|
||||
|
||||
@@ -3,12 +3,14 @@
|
||||
package wickra
|
||||
|
||||
/*
|
||||
#include <stdlib.h>
|
||||
#include "wickra.h"
|
||||
*/
|
||||
import "C"
|
||||
|
||||
import (
|
||||
"runtime"
|
||||
"time"
|
||||
"unsafe"
|
||||
)
|
||||
|
||||
@@ -43319,3 +43321,113 @@ func (ind *Zlema) Close() {
|
||||
runtime.SetFinalizer(ind, nil)
|
||||
}
|
||||
}
|
||||
|
||||
// ===== Live Binance kline feed (feature `live-binance`) =====
|
||||
|
||||
// BinanceInterval selects a kline interval (the Interval declaration order).
|
||||
type BinanceInterval uint8
|
||||
|
||||
// Kline intervals supported by the live Binance feed.
|
||||
const (
|
||||
OneSecond BinanceInterval = iota
|
||||
OneMinute
|
||||
ThreeMinutes
|
||||
FiveMinutes
|
||||
FifteenMinutes
|
||||
ThirtyMinutes
|
||||
OneHour
|
||||
TwoHours
|
||||
FourHours
|
||||
SixHours
|
||||
EightHours
|
||||
TwelveHours
|
||||
OneDay
|
||||
ThreeDays
|
||||
OneWeek
|
||||
OneMonth
|
||||
)
|
||||
|
||||
// KlineEvent is one event from the live Binance feed.
|
||||
type KlineEvent struct {
|
||||
Symbol string
|
||||
Open float64
|
||||
High float64
|
||||
Low float64
|
||||
Close float64
|
||||
Volume float64
|
||||
OpenTime int64
|
||||
IsClosed bool
|
||||
}
|
||||
|
||||
// BinanceFeed is a live Binance kline stream over the Wickra C ABI.
|
||||
type BinanceFeed struct {
|
||||
handle *C.struct_BinanceStream
|
||||
}
|
||||
|
||||
// NewBinanceFeed connects to Binance's live kline stream for the given
|
||||
// comma-separated symbols (case-insensitive) at interval. baseURL overrides
|
||||
// the endpoint ("" = production wss://stream.binance.com:9443; pass a ws://
|
||||
// URL to target a test server). It returns ErrInvalidParams on a bad symbol
|
||||
// list, an unknown interval, a bad URL, or a failed initial connect.
|
||||
func NewBinanceFeed(symbols string, interval BinanceInterval, baseURL string) (*BinanceFeed, error) {
|
||||
csym := C.CString(symbols)
|
||||
defer C.free(unsafe.Pointer(csym))
|
||||
var curl *C.char
|
||||
if baseURL != "" {
|
||||
curl = C.CString(baseURL)
|
||||
defer C.free(unsafe.Pointer(curl))
|
||||
}
|
||||
ptr := C.wickra_binance_connect(csym, C.uint8_t(interval), curl)
|
||||
if ptr == nil {
|
||||
return nil, ErrInvalidParams
|
||||
}
|
||||
obj := &BinanceFeed{handle: ptr}
|
||||
runtime.SetFinalizer(obj, (*BinanceFeed).Close)
|
||||
return obj, nil
|
||||
}
|
||||
|
||||
// Next polls for the next kline event, waiting up to timeout. It returns the
|
||||
// event and true when one arrives, the zero value and false on timeout, or
|
||||
// ErrFeedClosed once the stream is closed or has errored out.
|
||||
func (f *BinanceFeed) Next(timeout time.Duration) (KlineEvent, bool, error) {
|
||||
var ev C.struct_WickraKlineEvent
|
||||
code := int(C.wickra_binance_next(f.handle, &ev, C.int64_t(timeout.Milliseconds())))
|
||||
runtime.KeepAlive(f)
|
||||
switch code {
|
||||
case 1:
|
||||
return klineFromC(&ev), true, nil
|
||||
case 0:
|
||||
return KlineEvent{}, false, nil
|
||||
default:
|
||||
return KlineEvent{}, false, ErrFeedClosed
|
||||
}
|
||||
}
|
||||
|
||||
func klineFromC(ev *C.struct_WickraKlineEvent) KlineEvent {
|
||||
n := 0
|
||||
for n < len(ev.symbol) && ev.symbol[n] != 0 {
|
||||
n++
|
||||
}
|
||||
sym := C.GoBytes(unsafe.Pointer(&ev.symbol[0]), C.int(n))
|
||||
return KlineEvent{
|
||||
Symbol: string(sym),
|
||||
Open: float64(ev.open),
|
||||
High: float64(ev.high),
|
||||
Low: float64(ev.low),
|
||||
Close: float64(ev.close),
|
||||
Volume: float64(ev.volume),
|
||||
OpenTime: int64(ev.open_time),
|
||||
IsClosed: bool(ev.is_closed),
|
||||
}
|
||||
}
|
||||
|
||||
// Close ends the stream and frees the native handle. Idempotent and safe to
|
||||
// call alongside the finalizer.
|
||||
func (f *BinanceFeed) Close() {
|
||||
if f.handle != nil {
|
||||
C.wickra_binance_close(f.handle)
|
||||
C.wickra_binance_free(f.handle)
|
||||
f.handle = nil
|
||||
runtime.SetFinalizer(f, nil)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -28,3 +28,7 @@ import "errors"
|
||||
// ErrInvalidParams is returned by a New<Indicator> constructor when the native
|
||||
// constructor rejects the supplied parameters (for example a zero period).
|
||||
var ErrInvalidParams = errors.New("wickra: invalid indicator parameters")
|
||||
|
||||
// ErrFeedClosed is returned by BinanceFeed.Next once the stream has been closed
|
||||
// or has errored out and exhausted its reconnect attempts.
|
||||
var ErrFeedClosed = errors.New("wickra: binance feed closed")
|
||||
|
||||
@@ -0,0 +1,92 @@
|
||||
// Generated from bindings/c/include/wickra.h. Do not edit by hand.
|
||||
package org.wickra;
|
||||
|
||||
import org.wickra.internal.NativeMethods;
|
||||
import org.wickra.internal.WickraNative;
|
||||
import java.lang.foreign.Arena;
|
||||
import java.lang.foreign.MemorySegment;
|
||||
import java.lang.ref.Cleaner;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import static java.lang.foreign.ValueLayout.*;
|
||||
|
||||
/** A live Binance kline stream over the Wickra C ABI. Not thread-safe; close when done. */
|
||||
public final class BinanceFeed implements AutoCloseable {
|
||||
private final MemorySegment handle;
|
||||
private final Cleaner.Cleanable cleanable;
|
||||
|
||||
/** Connect to Binance's live kline stream for the given comma-separated
|
||||
* symbols (case-insensitive) at interval. baseUrl overrides the endpoint
|
||||
* (null = production; pass a ws:// URL to target a test server). */
|
||||
public BinanceFeed(String symbols, BinanceInterval interval, String baseUrl) {
|
||||
if (symbols == null) {
|
||||
throw new NullPointerException("symbols");
|
||||
}
|
||||
MemorySegment h;
|
||||
try (Arena a = Arena.ofConfined()) {
|
||||
byte[] sb = symbols.getBytes(StandardCharsets.UTF_8);
|
||||
MemorySegment sym = a.allocate(sb.length + 1L);
|
||||
MemorySegment.copy(sb, 0, sym, JAVA_BYTE, 0L, sb.length);
|
||||
MemorySegment url = MemorySegment.NULL;
|
||||
if (baseUrl != null) {
|
||||
byte[] ub = baseUrl.getBytes(StandardCharsets.UTF_8);
|
||||
url = a.allocate(ub.length + 1L);
|
||||
MemorySegment.copy(ub, 0, url, JAVA_BYTE, 0L, ub.length);
|
||||
}
|
||||
h = (MemorySegment) NativeMethods.WICKRA_BINANCE_CONNECT.invokeExact(
|
||||
sym, (byte) interval.ordinal(), url);
|
||||
} catch (Throwable t) {
|
||||
throw WickraNative.rethrow(t);
|
||||
}
|
||||
if (h.address() == 0L) {
|
||||
throw new IllegalArgumentException("invalid BinanceFeed parameters");
|
||||
}
|
||||
this.handle = h;
|
||||
this.cleanable = WickraNative.register(this, h, NativeMethods.WICKRA_BINANCE_FREE);
|
||||
}
|
||||
|
||||
/** Connect with the production endpoint. */
|
||||
public BinanceFeed(String symbols, BinanceInterval interval) {
|
||||
this(symbols, interval, null);
|
||||
}
|
||||
|
||||
/** Poll for the next kline event, waiting up to timeoutMillis. Returns the
|
||||
* event, or null on timeout (call again). Throws once the stream is closed. */
|
||||
public KlineEvent next(long timeoutMillis) {
|
||||
try (Arena a = Arena.ofConfined()) {
|
||||
MemorySegment out = a.allocate(72L);
|
||||
int code = (int) NativeMethods.WICKRA_BINANCE_NEXT.invokeExact(handle, out, timeoutMillis);
|
||||
if (code == 0) {
|
||||
return null;
|
||||
}
|
||||
if (code != 1) {
|
||||
throw new IllegalStateException("binance feed closed");
|
||||
}
|
||||
byte[] symBytes = new byte[16];
|
||||
MemorySegment.copy(out, JAVA_BYTE, 0L, symBytes, 0, 16);
|
||||
int n = 0;
|
||||
while (n < 16 && symBytes[n] != 0) {
|
||||
n++;
|
||||
}
|
||||
return new KlineEvent(
|
||||
new String(symBytes, 0, n, StandardCharsets.UTF_8),
|
||||
out.get(JAVA_DOUBLE, 16L),
|
||||
out.get(JAVA_DOUBLE, 24L),
|
||||
out.get(JAVA_DOUBLE, 32L),
|
||||
out.get(JAVA_DOUBLE, 40L),
|
||||
out.get(JAVA_DOUBLE, 48L),
|
||||
out.get(JAVA_LONG, 56L),
|
||||
out.get(JAVA_BYTE, 64L) != 0);
|
||||
} catch (Throwable t) {
|
||||
throw WickraNative.rethrow(t);
|
||||
}
|
||||
}
|
||||
|
||||
@Override public void close() {
|
||||
try {
|
||||
NativeMethods.WICKRA_BINANCE_CLOSE.invokeExact(handle);
|
||||
} catch (Throwable t) {
|
||||
throw WickraNative.rethrow(t);
|
||||
}
|
||||
cleanable.clean();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
// Generated from bindings/c/include/wickra.h. Do not edit by hand.
|
||||
package org.wickra;
|
||||
|
||||
/** Kline interval for the live Binance feed (ordinal = the C ABI code). */
|
||||
public enum BinanceInterval {
|
||||
ONE_SECOND,
|
||||
ONE_MINUTE,
|
||||
THREE_MINUTES,
|
||||
FIVE_MINUTES,
|
||||
FIFTEEN_MINUTES,
|
||||
THIRTY_MINUTES,
|
||||
ONE_HOUR,
|
||||
TWO_HOURS,
|
||||
FOUR_HOURS,
|
||||
SIX_HOURS,
|
||||
EIGHT_HOURS,
|
||||
TWELVE_HOURS,
|
||||
ONE_DAY,
|
||||
THREE_DAYS,
|
||||
ONE_WEEK,
|
||||
ONE_MONTH
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
// Generated from bindings/c/include/wickra.h. Do not edit by hand.
|
||||
package org.wickra;
|
||||
|
||||
/** One event from the live Binance feed. */
|
||||
public record KlineEvent(String symbol, double open, double high, double low,
|
||||
double close, double volume, long openTime, boolean isClosed) {}
|
||||
@@ -3959,6 +3959,10 @@ public final class NativeMethods {
|
||||
public static MethodHandle WICKRA_CANDLE_READER_COUNT;
|
||||
public static MethodHandle WICKRA_CANDLE_READER_READ;
|
||||
public static MethodHandle WICKRA_CANDLE_READER_FREE;
|
||||
public static MethodHandle WICKRA_BINANCE_CONNECT;
|
||||
public static MethodHandle WICKRA_BINANCE_NEXT;
|
||||
public static MethodHandle WICKRA_BINANCE_CLOSE;
|
||||
public static MethodHandle WICKRA_BINANCE_FREE;
|
||||
|
||||
static {
|
||||
init0();
|
||||
@@ -8039,6 +8043,10 @@ public final class NativeMethods {
|
||||
WICKRA_CANDLE_READER_COUNT = h("wickra_candle_reader_count", FunctionDescriptor.of(JAVA_LONG, ADDRESS));
|
||||
WICKRA_CANDLE_READER_READ = h("wickra_candle_reader_read", FunctionDescriptor.of(JAVA_LONG, ADDRESS, ADDRESS, JAVA_LONG));
|
||||
WICKRA_CANDLE_READER_FREE = h("wickra_candle_reader_free", FunctionDescriptor.ofVoid(ADDRESS));
|
||||
WICKRA_BINANCE_CONNECT = h("wickra_binance_connect", FunctionDescriptor.of(ADDRESS, ADDRESS, JAVA_BYTE, ADDRESS));
|
||||
WICKRA_BINANCE_NEXT = h("wickra_binance_next", FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS, JAVA_LONG));
|
||||
WICKRA_BINANCE_CLOSE = h("wickra_binance_close", FunctionDescriptor.ofVoid(ADDRESS));
|
||||
WICKRA_BINANCE_FREE = h("wickra_binance_free", FunctionDescriptor.ofVoid(ADDRESS));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
package org.wickra;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
|
||||
/**
|
||||
* The live Binance feed's connect → read → reconnect pipeline is covered
|
||||
* deterministically by the Rust mock-WS-server tests in wickra-data. Here we
|
||||
* only assert the binding's error paths, which need no network.
|
||||
*/
|
||||
class BinanceFeedTest {
|
||||
@Test
|
||||
void rejectsEmptySymbols() {
|
||||
assertThrows(IllegalArgumentException.class,
|
||||
() -> new BinanceFeed("", BinanceInterval.ONE_MINUTE));
|
||||
}
|
||||
|
||||
@Test
|
||||
void rejectsUnreachableEndpoint() {
|
||||
assertThrows(IllegalArgumentException.class,
|
||||
() -> new BinanceFeed("BTCUSDT", BinanceInterval.ONE_MINUTE, "ws://127.0.0.1:1"));
|
||||
}
|
||||
}
|
||||
@@ -25,9 +25,11 @@ workspace = true
|
||||
|
||||
[dependencies]
|
||||
wickra-core = { workspace = true }
|
||||
wickra-data = { workspace = true }
|
||||
wickra-data = { workspace = true, features = ["live-binance"] }
|
||||
napi = { version = "2.16", features = ["napi8"] }
|
||||
napi-derive = "2.16"
|
||||
# Drives the async Binance feed behind the blocking poll.
|
||||
tokio = { version = "1", features = ["rt", "net", "time"] }
|
||||
|
||||
[build-dependencies]
|
||||
napi-build = "2"
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
// The live Binance feed's connect → read → reconnect pipeline is covered
|
||||
// deterministically by the Rust mock-WS-server tests in wickra-data. Here we only
|
||||
// assert the binding's error paths, which need no network.
|
||||
|
||||
const test = require('node:test');
|
||||
const assert = require('node:assert/strict');
|
||||
const { BinanceFeed } = require('..');
|
||||
|
||||
test('binance feed rejects an unknown interval', () => {
|
||||
assert.throws(() => new BinanceFeed('BTCUSDT', 99));
|
||||
});
|
||||
|
||||
test('binance feed rejects an empty symbol list', () => {
|
||||
assert.throws(() => new BinanceFeed('', 1));
|
||||
});
|
||||
|
||||
test('binance feed rejects an unreachable endpoint', () => {
|
||||
assert.throws(() => new BinanceFeed('BTCUSDT', 1, 'ws://127.0.0.1:1'));
|
||||
});
|
||||
Vendored
+35
@@ -603,6 +603,17 @@ export interface CandleValue {
|
||||
volume: number
|
||||
timestamp: number
|
||||
}
|
||||
/** One event from the live Binance feed. */
|
||||
export interface KlineEvent {
|
||||
symbol: string
|
||||
open: number
|
||||
high: number
|
||||
low: number
|
||||
close: number
|
||||
volume: number
|
||||
openTime: number
|
||||
isClosed: boolean
|
||||
}
|
||||
export type SmaNode = SMA
|
||||
export declare class SMA {
|
||||
constructor(period: number)
|
||||
@@ -5939,6 +5950,30 @@ export declare class VolumeWeightedMacd {
|
||||
isReady(): boolean
|
||||
warmupPeriod(): number
|
||||
}
|
||||
export type BinanceFeedNode = BinanceFeed
|
||||
/**
|
||||
* A live Binance kline feed. `next` is a blocking poll (it waits up to the
|
||||
* timeout on the calling thread) driving the mock-server-tested wickra-data
|
||||
* `BinanceKlineStream` on a single-thread tokio runtime — use a short timeout in
|
||||
* a loop, or run it on a Worker thread, to keep the event loop responsive.
|
||||
*/
|
||||
export declare class BinanceFeed {
|
||||
/**
|
||||
* Connect to Binance's live kline stream for the given comma-separated
|
||||
* `symbols` (case-insensitive) at the given `interval` code (`0..=15`).
|
||||
* `baseUrl` overrides the endpoint (omit for production; pass a `ws://` URL
|
||||
* to target a test server).
|
||||
*/
|
||||
constructor(symbols: string, interval: number, baseUrl?: string | undefined | null)
|
||||
/**
|
||||
* Poll for the next kline event, waiting up to `timeoutMs` milliseconds.
|
||||
* Returns the event, or `null` on timeout (call again). Throws once the
|
||||
* stream is closed.
|
||||
*/
|
||||
next(timeoutMs: number): KlineEvent | null
|
||||
/** Close the stream; subsequent `next` calls throw. */
|
||||
close(): void
|
||||
}
|
||||
export type TickAggregatorNode = TickAggregator
|
||||
/** Roll trade ticks up into fixed-timeframe OHLCV candles. */
|
||||
export declare class TickAggregator {
|
||||
|
||||
File diff suppressed because one or more lines are too long
@@ -21884,6 +21884,138 @@ pub struct CandleValue {
|
||||
pub timestamp: f64,
|
||||
}
|
||||
|
||||
/// One event from the live Binance feed.
|
||||
#[napi(object)]
|
||||
pub struct KlineEvent {
|
||||
pub symbol: String,
|
||||
pub open: f64,
|
||||
pub high: f64,
|
||||
pub low: f64,
|
||||
pub close: f64,
|
||||
pub volume: f64,
|
||||
pub open_time: f64,
|
||||
pub is_closed: bool,
|
||||
}
|
||||
|
||||
/// Map a `u8` interval code (`0..=15`, the `Interval` declaration order) to the
|
||||
/// feed interval.
|
||||
fn binance_interval(code: u8) -> Option<wickra_data::live::binance::Interval> {
|
||||
use wickra_data::live::binance::Interval;
|
||||
Some(match code {
|
||||
0 => Interval::OneSecond,
|
||||
1 => Interval::OneMinute,
|
||||
2 => Interval::ThreeMinutes,
|
||||
3 => Interval::FiveMinutes,
|
||||
4 => Interval::FifteenMinutes,
|
||||
5 => Interval::ThirtyMinutes,
|
||||
6 => Interval::OneHour,
|
||||
7 => Interval::TwoHours,
|
||||
8 => Interval::FourHours,
|
||||
9 => Interval::SixHours,
|
||||
10 => Interval::EightHours,
|
||||
11 => Interval::TwelveHours,
|
||||
12 => Interval::OneDay,
|
||||
13 => Interval::ThreeDays,
|
||||
14 => Interval::OneWeek,
|
||||
15 => Interval::OneMonth,
|
||||
_ => return None,
|
||||
})
|
||||
}
|
||||
|
||||
/// A live Binance kline feed. `next` is a blocking poll (it waits up to the
|
||||
/// timeout on the calling thread) driving the mock-server-tested wickra-data
|
||||
/// `BinanceKlineStream` on a single-thread tokio runtime — use a short timeout in
|
||||
/// a loop, or run it on a Worker thread, to keep the event loop responsive.
|
||||
#[napi(js_name = "BinanceFeed")]
|
||||
pub struct BinanceFeedNode {
|
||||
runtime: tokio::runtime::Runtime,
|
||||
inner: wickra_data::live::binance::BinanceKlineStream,
|
||||
}
|
||||
|
||||
#[napi]
|
||||
impl BinanceFeedNode {
|
||||
/// Connect to Binance's live kline stream for the given comma-separated
|
||||
/// `symbols` (case-insensitive) at the given `interval` code (`0..=15`).
|
||||
/// `baseUrl` overrides the endpoint (omit for production; pass a `ws://` URL
|
||||
/// to target a test server).
|
||||
#[napi(constructor)]
|
||||
pub fn new(symbols: String, interval: u8, base_url: Option<String>) -> napi::Result<Self> {
|
||||
let iv = binance_interval(interval).ok_or_else(|| {
|
||||
NapiError::new(
|
||||
Status::InvalidArg,
|
||||
"unknown interval code (expected 0..=15)",
|
||||
)
|
||||
})?;
|
||||
let symbol_list: Vec<String> = symbols
|
||||
.split(',')
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty())
|
||||
.map(str::to_owned)
|
||||
.collect();
|
||||
if symbol_list.is_empty() {
|
||||
return Err(NapiError::new(
|
||||
Status::InvalidArg,
|
||||
"at least one symbol is required",
|
||||
));
|
||||
}
|
||||
let mut config = wickra_data::live::binance::BinanceConfig::default();
|
||||
if let Some(url) = base_url {
|
||||
config.base_url = url;
|
||||
}
|
||||
let runtime = tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
.map_err(|e| NapiError::new(Status::GenericFailure, e.to_string()))?;
|
||||
let inner = runtime
|
||||
.block_on(
|
||||
wickra_data::live::binance::BinanceKlineStream::connect_with_config(
|
||||
&symbol_list,
|
||||
iv,
|
||||
config,
|
||||
),
|
||||
)
|
||||
.map_err(map_data_err)?;
|
||||
Ok(Self { runtime, inner })
|
||||
}
|
||||
|
||||
/// Poll for the next kline event, waiting up to `timeoutMs` milliseconds.
|
||||
/// Returns the event, or `null` on timeout (call again). Throws once the
|
||||
/// stream is closed.
|
||||
#[napi]
|
||||
pub fn next(&mut self, timeout_ms: f64) -> napi::Result<Option<KlineEvent>> {
|
||||
let dur = std::time::Duration::from_millis(timeout_ms.max(0.0) as u64);
|
||||
let runtime = &self.runtime;
|
||||
let inner = &mut self.inner;
|
||||
let result = runtime.block_on(tokio::time::timeout(dur, inner.next_event()));
|
||||
match result {
|
||||
Ok(Ok(Some(ev))) => Ok(Some(KlineEvent {
|
||||
symbol: ev.symbol,
|
||||
open: ev.candle.open,
|
||||
high: ev.candle.high,
|
||||
low: ev.candle.low,
|
||||
close: ev.candle.close,
|
||||
volume: ev.candle.volume,
|
||||
#[allow(clippy::cast_precision_loss)]
|
||||
open_time: ev.candle.timestamp as f64,
|
||||
is_closed: ev.is_closed,
|
||||
})),
|
||||
Ok(Ok(None) | Err(_)) => Err(NapiError::new(
|
||||
Status::GenericFailure,
|
||||
"binance feed closed",
|
||||
)),
|
||||
Err(_) => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
/// Close the stream; subsequent `next` calls throw.
|
||||
#[napi]
|
||||
pub fn close(&mut self) {
|
||||
let runtime = &self.runtime;
|
||||
let inner = &mut self.inner;
|
||||
let _ = runtime.block_on(inner.close());
|
||||
}
|
||||
}
|
||||
|
||||
/// Roll trade ticks up into fixed-timeframe OHLCV candles.
|
||||
#[napi(js_name = "TickAggregator")]
|
||||
pub struct TickAggregatorNode {
|
||||
|
||||
@@ -22,6 +22,8 @@ workspace = true
|
||||
|
||||
[dependencies]
|
||||
wickra-core = { workspace = true }
|
||||
wickra-data = { workspace = true }
|
||||
wickra-data = { workspace = true, features = ["live-binance"] }
|
||||
pyo3 = { workspace = true }
|
||||
numpy = { workspace = true }
|
||||
# Drives the async Binance feed behind the blocking, GIL-releasing poll.
|
||||
tokio = { version = "1", features = ["rt", "net", "time"] }
|
||||
|
||||
@@ -362,6 +362,7 @@ from ._wickra import (
|
||||
TickAggregator,
|
||||
Resampler,
|
||||
CandleReader,
|
||||
BinanceFeed,
|
||||
# Market Profile
|
||||
CompositeProfile,
|
||||
HighLowVolumeNodes,
|
||||
@@ -910,6 +911,7 @@ __all__ = [
|
||||
"TickAggregator",
|
||||
"Resampler",
|
||||
"CandleReader",
|
||||
"BinanceFeed",
|
||||
# Market Profile
|
||||
"CompositeProfile",
|
||||
"HighLowVolumeNodes",
|
||||
|
||||
+108
-1
@@ -13,7 +13,7 @@
|
||||
#![allow(clippy::many_single_char_names)]
|
||||
|
||||
use numpy::{IntoPyArray, PyArray1, PyArray2, PyReadonlyArray1};
|
||||
use pyo3::exceptions::{PyTypeError, PyValueError};
|
||||
use pyo3::exceptions::{PyRuntimeError, PyTypeError, PyValueError};
|
||||
use pyo3::prelude::*;
|
||||
use pyo3::types::PyDict;
|
||||
use wickra_core as wc;
|
||||
@@ -28289,6 +28289,7 @@ fn _wickra(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
|
||||
m.add_class::<PyTickAggregator>()?;
|
||||
m.add_class::<PyResampler>()?;
|
||||
m.add_class::<PyCandleReader>()?;
|
||||
m.add_class::<PyBinanceFeed>()?;
|
||||
// Candlestick patterns.
|
||||
m.add_class::<PyDoji>()?;
|
||||
m.add_class::<PyHammer>()?;
|
||||
@@ -28572,6 +28573,112 @@ fn map_data_err(e: wickra_data::Error) -> PyErr {
|
||||
PyValueError::new_err(e.to_string())
|
||||
}
|
||||
|
||||
/// Map a `u8` interval code (`0..=15`, the `Interval` declaration order) to the
|
||||
/// feed interval.
|
||||
fn binance_interval(code: u8) -> Option<wickra_data::live::binance::Interval> {
|
||||
use wickra_data::live::binance::Interval;
|
||||
Some(match code {
|
||||
0 => Interval::OneSecond,
|
||||
1 => Interval::OneMinute,
|
||||
2 => Interval::ThreeMinutes,
|
||||
3 => Interval::FiveMinutes,
|
||||
4 => Interval::FifteenMinutes,
|
||||
5 => Interval::ThirtyMinutes,
|
||||
6 => Interval::OneHour,
|
||||
7 => Interval::TwoHours,
|
||||
8 => Interval::FourHours,
|
||||
9 => Interval::SixHours,
|
||||
10 => Interval::EightHours,
|
||||
11 => Interval::TwelveHours,
|
||||
12 => Interval::OneDay,
|
||||
13 => Interval::ThreeDays,
|
||||
14 => Interval::OneWeek,
|
||||
15 => Interval::OneMonth,
|
||||
_ => return None,
|
||||
})
|
||||
}
|
||||
|
||||
/// `(symbol, open, high, low, close, volume, open_time, is_closed)`.
|
||||
type KlineTuple = (String, f64, f64, f64, f64, f64, i64, bool);
|
||||
|
||||
/// A live Binance kline feed (blocking poll). The connect / read / reconnect
|
||||
/// pipeline is the mock-server-tested wickra-data `BinanceKlineStream`, driven on
|
||||
/// a single-thread tokio runtime; `next` releases the GIL while it waits.
|
||||
#[pyclass(name = "BinanceFeed", module = "wickra._wickra", skip_from_py_object)]
|
||||
struct PyBinanceFeed {
|
||||
runtime: tokio::runtime::Runtime,
|
||||
inner: wickra_data::live::binance::BinanceKlineStream,
|
||||
}
|
||||
|
||||
#[pymethods]
|
||||
impl PyBinanceFeed {
|
||||
#[new]
|
||||
#[pyo3(signature = (symbols, interval, base_url = None))]
|
||||
fn new(symbols: &str, interval: u8, base_url: Option<&str>) -> PyResult<Self> {
|
||||
let iv = binance_interval(interval)
|
||||
.ok_or_else(|| PyValueError::new_err("unknown interval code (expected 0..=15)"))?;
|
||||
let symbol_list: Vec<String> = symbols
|
||||
.split(',')
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty())
|
||||
.map(str::to_owned)
|
||||
.collect();
|
||||
if symbol_list.is_empty() {
|
||||
return Err(PyValueError::new_err("at least one symbol is required"));
|
||||
}
|
||||
let mut config = wickra_data::live::binance::BinanceConfig::default();
|
||||
if let Some(url) = base_url {
|
||||
url.clone_into(&mut config.base_url);
|
||||
}
|
||||
let runtime = tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
.map_err(|e| PyRuntimeError::new_err(e.to_string()))?;
|
||||
let inner = runtime
|
||||
.block_on(
|
||||
wickra_data::live::binance::BinanceKlineStream::connect_with_config(
|
||||
&symbol_list,
|
||||
iv,
|
||||
config,
|
||||
),
|
||||
)
|
||||
.map_err(map_data_err)?;
|
||||
Ok(Self { runtime, inner })
|
||||
}
|
||||
|
||||
/// Poll for the next kline event, waiting up to `timeout_ms`. Returns a
|
||||
/// `(symbol, open, high, low, close, volume, open_time, is_closed)` tuple, or
|
||||
/// `None` on timeout (call again). Raises once the stream is closed.
|
||||
#[pyo3(signature = (timeout_ms = 1000.0))]
|
||||
fn next(&mut self, py: Python<'_>, timeout_ms: f64) -> PyResult<Option<KlineTuple>> {
|
||||
let dur = std::time::Duration::from_millis(timeout_ms.max(0.0) as u64);
|
||||
let runtime = &self.runtime;
|
||||
let inner = &mut self.inner;
|
||||
let result = py.detach(|| runtime.block_on(tokio::time::timeout(dur, inner.next_event())));
|
||||
match result {
|
||||
Ok(Ok(Some(ev))) => Ok(Some((
|
||||
ev.symbol,
|
||||
ev.candle.open,
|
||||
ev.candle.high,
|
||||
ev.candle.low,
|
||||
ev.candle.close,
|
||||
ev.candle.volume,
|
||||
ev.candle.timestamp,
|
||||
ev.is_closed,
|
||||
))),
|
||||
Ok(Ok(None) | Err(_)) => Err(PyRuntimeError::new_err("binance feed closed")),
|
||||
Err(_) => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
/// Close the stream; subsequent `next` calls raise.
|
||||
fn close(&mut self) {
|
||||
let runtime = &self.runtime;
|
||||
let inner = &mut self.inner;
|
||||
let _ = runtime.block_on(inner.close());
|
||||
}
|
||||
}
|
||||
|
||||
/// Roll trade ticks up into fixed-timeframe OHLCV candles.
|
||||
#[pyclass(
|
||||
name = "TickAggregator",
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
"""The live Binance feed's connect -> read -> reconnect pipeline is covered
|
||||
deterministically by the Rust mock-WS-server tests in wickra-data. Here we only
|
||||
assert the binding's error paths, which need no network.
|
||||
"""
|
||||
import pytest
|
||||
import wickra as ta
|
||||
|
||||
|
||||
def test_binance_feed_rejects_bad_params():
|
||||
with pytest.raises(ValueError):
|
||||
ta.BinanceFeed("BTCUSDT", 99) # unknown interval code
|
||||
with pytest.raises(ValueError):
|
||||
ta.BinanceFeed("", 1) # empty symbol list
|
||||
with pytest.raises(ValueError):
|
||||
ta.BinanceFeed("BTCUSDT", 1, "ws://127.0.0.1:1") # unreachable endpoint
|
||||
@@ -55,6 +55,7 @@ export(BeltHold)
|
||||
export(Beta)
|
||||
export(BetaNeutralSpread)
|
||||
export(BetterVolume)
|
||||
export(BinanceFeed)
|
||||
export(BipowerVariation)
|
||||
export(BodySizePct)
|
||||
export(BollingerBands)
|
||||
@@ -527,6 +528,8 @@ export(ZeroLagMacd)
|
||||
export(ZigZag)
|
||||
export(Zlema)
|
||||
export(batch)
|
||||
export(binance_close)
|
||||
export(binance_next)
|
||||
export(is_ready)
|
||||
export(name)
|
||||
export(push)
|
||||
|
||||
@@ -4143,3 +4143,40 @@ Zlema <- function(period) {
|
||||
.wk_obj("zlema", ptr, "Zlema")
|
||||
}
|
||||
|
||||
#' Connect to a live Binance kline feed
|
||||
#'
|
||||
#' Opens a live Binance kline stream for one or more comma-separated `symbols`
|
||||
#' (case-insensitive) at the given `interval` code (an integer `0:15`, the same
|
||||
#' order as the other bindings: 0 = 1s, 1 = 1m, ... 12 = 1d, 13 = 3d, 14 = 1w,
|
||||
#' 15 = 1M). `base_url` overrides the endpoint (`NULL` = production). Not
|
||||
#' available in the wasm (r-universe/webR) build, which has no raw sockets.
|
||||
#'
|
||||
#' @keywords internal
|
||||
#' @export
|
||||
BinanceFeed <- function(symbols, interval, base_url = NULL) {
|
||||
ptr <- .Call("wk_binance_connect", symbols, as.integer(interval), base_url,
|
||||
PACKAGE = "wickra")
|
||||
structure(list(ptr = ptr), class = "wickra_binance_feed")
|
||||
}
|
||||
|
||||
#' Poll a Binance feed for the next kline event
|
||||
#'
|
||||
#' Waits up to `timeout_ms` milliseconds. Returns a named list (`symbol`, `open`,
|
||||
#' `high`, `low`, `close`, `volume`, `open_time`, `is_closed`) when an event
|
||||
#' arrives, or `NULL` on timeout. Errors once the stream is closed.
|
||||
#'
|
||||
#' @keywords internal
|
||||
#' @export
|
||||
binance_next <- function(feed, timeout_ms = 1000) {
|
||||
.Call("wk_binance_next", feed$ptr, as.numeric(timeout_ms), PACKAGE = "wickra")
|
||||
}
|
||||
|
||||
#' Close a Binance feed
|
||||
#'
|
||||
#' @keywords internal
|
||||
#' @export
|
||||
binance_close <- function(feed) {
|
||||
.Call("wk_binance_close", feed$ptr, PACKAGE = "wickra")
|
||||
invisible(NULL)
|
||||
}
|
||||
|
||||
|
||||
@@ -22660,6 +22660,55 @@ SEXP wk_zlema_reset(SEXP e) {
|
||||
return R_NilValue;
|
||||
}
|
||||
|
||||
/* Live Binance kline feed (feature `live-binance`). Gated out of the
|
||||
* Emscripten/wasm build (r-universe/webR), which has no raw TCP/TLS sockets. */
|
||||
#ifndef __EMSCRIPTEN__
|
||||
static void binance_fin(SEXP e) {
|
||||
struct BinanceStream *h = (struct BinanceStream *)R_ExternalPtrAddr(e);
|
||||
if (h) wickra_binance_free(h);
|
||||
R_ClearExternalPtr(e);
|
||||
}
|
||||
SEXP wk_binance_connect(SEXP symbols, SEXP interval, SEXP base_url) {
|
||||
const char *url = (base_url == R_NilValue || Rf_xlength(base_url) == 0)
|
||||
? NULL : CHAR(STRING_ELT(base_url, 0));
|
||||
struct BinanceStream *h = wickra_binance_connect(
|
||||
CHAR(STRING_ELT(symbols, 0)), (uint8_t)Rf_asInteger(interval), url);
|
||||
if (!h) Rf_error("invalid BinanceFeed parameters");
|
||||
SEXP e = PROTECT(R_MakeExternalPtr(h, R_NilValue, R_NilValue));
|
||||
R_RegisterCFinalizerEx(e, binance_fin, TRUE);
|
||||
UNPROTECT(1);
|
||||
return e;
|
||||
}
|
||||
SEXP wk_binance_next(SEXP e, SEXP timeout_ms) {
|
||||
struct BinanceStream *h = (struct BinanceStream *)R_ExternalPtrAddr(e);
|
||||
struct WickraKlineEvent ev;
|
||||
int code = wickra_binance_next(h, &ev, (int64_t)Rf_asReal(timeout_ms));
|
||||
if (code == 0) return R_NilValue;
|
||||
if (code != 1) Rf_error("binance feed closed");
|
||||
const char *names[] = {"symbol", "open", "high", "low", "close", "volume",
|
||||
"open_time", "is_closed"};
|
||||
SEXP out = PROTECT(Rf_allocVector(VECSXP, 8));
|
||||
SET_VECTOR_ELT(out, 0, Rf_mkString((const char *)ev.symbol));
|
||||
SET_VECTOR_ELT(out, 1, Rf_ScalarReal(ev.open));
|
||||
SET_VECTOR_ELT(out, 2, Rf_ScalarReal(ev.high));
|
||||
SET_VECTOR_ELT(out, 3, Rf_ScalarReal(ev.low));
|
||||
SET_VECTOR_ELT(out, 4, Rf_ScalarReal(ev.close));
|
||||
SET_VECTOR_ELT(out, 5, Rf_ScalarReal(ev.volume));
|
||||
SET_VECTOR_ELT(out, 6, Rf_ScalarReal((double)ev.open_time));
|
||||
SET_VECTOR_ELT(out, 7, Rf_ScalarLogical(ev.is_closed));
|
||||
SEXP nm = PROTECT(Rf_allocVector(STRSXP, 8));
|
||||
for (int i = 0; i < 8; i++) SET_STRING_ELT(nm, i, Rf_mkChar(names[i]));
|
||||
Rf_setAttrib(out, R_NamesSymbol, nm);
|
||||
UNPROTECT(2);
|
||||
return out;
|
||||
}
|
||||
SEXP wk_binance_close(SEXP e) {
|
||||
struct BinanceStream *h = (struct BinanceStream *)R_ExternalPtrAddr(e);
|
||||
wickra_binance_close(h);
|
||||
return R_NilValue;
|
||||
}
|
||||
#endif
|
||||
|
||||
static const R_CallMethodDef CallEntries[] = {
|
||||
{"wk_abandoned_baby_new", (DL_FUNC)&wk_abandoned_baby_new, 0},
|
||||
{"wk_abandoned_baby_update", (DL_FUNC)&wk_abandoned_baby_update, 7},
|
||||
@@ -26088,6 +26137,11 @@ static const R_CallMethodDef CallEntries[] = {
|
||||
{"wk_zlema_is_ready", (DL_FUNC)&wk_zlema_is_ready, 1},
|
||||
{"wk_zlema_name", (DL_FUNC)&wk_zlema_name, 1},
|
||||
{"wk_zlema_reset", (DL_FUNC)&wk_zlema_reset, 1},
|
||||
#ifndef __EMSCRIPTEN__
|
||||
{"wk_binance_connect", (DL_FUNC)&wk_binance_connect, 3},
|
||||
{"wk_binance_next", (DL_FUNC)&wk_binance_next, 2},
|
||||
{"wk_binance_close", (DL_FUNC)&wk_binance_close, 1},
|
||||
#endif
|
||||
{NULL, NULL, 0}
|
||||
};
|
||||
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
# The live Binance feed's connect -> read -> reconnect pipeline is covered
|
||||
# deterministically by the Rust mock-WS-server tests in wickra-data. Here we only
|
||||
# assert the binding's error paths, which need no network.
|
||||
|
||||
test_that("binance feed rejects bad parameters", {
|
||||
expect_error(BinanceFeed("", 1L), "BinanceFeed")
|
||||
expect_error(BinanceFeed("BTCUSDT", 1L, "ws://127.0.0.1:1"), "BinanceFeed")
|
||||
})
|
||||
Reference in New Issue
Block a user