Upload files.

This commit is contained in:
0xcathiefish
2026-02-12 08:11:37 +00:00
parent 47fa783913
commit b33ef907d4
2029 changed files with 53320 additions and 0 deletions
+14
View File
@@ -0,0 +1,14 @@
RUST_LOG="INFO,binance_sdk::common::utils=off,binance_sdk::common::websocket=off"
DATABASE_URL="postgres://xxx:xxxxxxx"
TRACKED_SYMBOL=[BTCUSDT,XRPUSDT,BNBUSDT,SOLUSDT,ETHUSDT]
# Proxy settings (optional — leave empty or remove for direct connection)
# Required for fast multi-proxy parallel downloads from Binance Data Archive
PROXY_HOST=
PROXY_USERNAME=
PROXY_PASSWORD=
PROXY_PROTOCOL=https
PROXY_PORT_START=10000
PROXY_PORT_END=10099
+2
View File
@@ -0,0 +1,2 @@
/target
+3981
View File
File diff suppressed because it is too large Load Diff
+28
View File
@@ -0,0 +1,28 @@
[package]
name = "backend"
version = "0.1.0"
edition = "2021"
[dependencies]
binance-sdk = { version = "35.0.0", features = ["derivatives_trading_usds_futures", "spot"] }
rand = "0.9"
sqlx = { version = "0.8", features = ["runtime-tokio", "postgres"] }
axum = { version = "0.8", features = ["ws"] }
tower-http = { version = "0.6", features = ["cors"] }
dotenv = "0.15.0"
json = "0.12.4"
log = "0.4.29"
serde = { version = "1.0.228", features = ["derive"] }
serde_json = "1.0.145"
thiserror = "2.0.17"
tokio = {version = "1.48.0", features = ["full"] }
env_logger = "0.11"
reqwest = { version = "0.12", features = ["stream"] }
zip = "2.2"
chrono = "0.4"
csv = "1.3"
futures = "0.3"
tokio-tungstenite = { version = "0.26", features = ["native-tls"] }
+14
View File
@@ -0,0 +1,14 @@
# Build stage
FROM rust:1.89-slim-bookworm AS builder
WORKDIR /app
RUN apt-get update && apt-get install -y pkg-config libssl-dev && rm -rf /var/lib/apt/lists/*
COPY Cargo.toml Cargo.lock ./
COPY src ./src
RUN cargo build --release
# Runtime stage
FROM debian:bookworm-slim
RUN apt-get update && apt-get install -y ca-certificates libssl3 && rm -rf /var/lib/apt/lists/*
COPY --from=builder /app/target/release/backend /usr/local/bin/backend
EXPOSE 3000
CMD ["backend"]
+296
View File
@@ -0,0 +1,296 @@
# 📊 Binance USDS Futures — Data Collection & API Backend
A high-performance Rust backend that collects, stores, and serves Binance USDS-M Futures K-line data in real-time. Designed as the data engine for custom charting frontends — supports both [KlineChart](https://klinecharts.com/) and [TradingView](https://www.tradingview.com/charting-library-docs/) out of the box.
## ✨ Features
- **Real-time WebSocket streaming** — Subscribe to 1-minute K-line updates via Binance combined streams, with auto-reconnect and backfill on disconnect
- **Multi-strategy historical sync** — Monthly ZIP → Daily ZIP → REST API fallback for fastest possible backfill
- **TimescaleDB-powered storage** — Hypertable-optimized with `time_bucket` aggregation for 8 timeframes (1m, 5m, 15m, 1h, 4h, 1D, 1W, 1M)
- **Dual API interface** — KlineChart REST API + TradingView UDF-compatible datafeed
- **Live WebSocket broadcast** — Push real-time candle updates to connected frontend clients
- **Canvas persistence** — Save/load chart drawings per symbol to local filesystem
- **Net Volume & Taker Buy Volume** — Custom indicators included in every response
- **Proxy pool support** — Up to 100 proxy clients (port 1000010099) for high-throughput parallel downloads
- **Docker ready** — Multi-stage build with minimal runtime image
## 🏗️ Architecture
```
┌──────────────────────────────────────────────────────────────────┐
│ Axum HTTP Server (:3000) │
│ │
│ ┌─────────────────────┐ ┌──────────────────────────┐ │
│ │ KlineChart API │ │ TradingView UDF API │ │
│ │ /api/klines │ │ /config /history /ws │ │
│ │ /api/symbols │ │ /symbols /search │ │
│ │ /api/status │ │ /canvas/* │ │
│ └────────┬────────────┘ └────────────┬─────────────┘ │
│ │ │ │
│ └──────────┬──────────────────────┘ │
│ ▼ │
│ ┌────────────────┐ │
│ │ Scheduler │ │
│ │ (Command Bus) │ │
│ └───┬────────┬───┘ │
│ │ │ │
│ ┌──────────▼──┐ ┌──▼──────────────────────┐ │
│ │ Binance │ │ Historical Downloader │ │
│ │ Collector │ │ (ZIP + REST backfill) │ │
│ │ (WebSocket)│ └─────────────────────────┘ │
│ └─────────────┘ │
│ │
│ ┌─────────────────────────┐ │
│ │ DatabaseHandler │ │
│ │ (TimescaleDB + batch) │ │
│ └─────────────────────────┘ │
└──────────────────────────────────────────────────────────────────┘
```
## 📁 Project Structure
```
src/
├── main.rs # Axum server bootstrap & route composition
├── binance_collector.rs # WebSocket real-time collection + REST sync
├── historical_downloader.rs # Binance data archive (ZIP) downloader
├── database.rs # TimescaleDB operations, batch insert, aggregation
├── scheduler.rs # Task coordination & collector lifecycle
├── klinechart.rs # KlineChart REST API handlers
├── tradingview.rs # TradingView UDF API + WebSocket + Canvas
├── structs.rs # Data types (CandleData, Interval, WsMessage…)
├── error.rs # Custom error types (CollectorError, SchedulerError)
└── lib.rs # Public module exports
tests/
├── connection_test.rs # Database connection tests
├── database_test.rs # CRUD & query tests
├── scheduler_test.rs # Scheduler command & lifecycle tests
├── sync_test.rs # Single symbol sync tests
└── sync_full_history_test.rs # Full historical backfill tests
examples/
├── sync_all.rs # Sync all symbols (standard)
├── sync_all_fast.rs # Sync all symbols (parallel with proxy pool)
└── sql.txt # Reference SQL for TimescaleDB setup
```
## 🚀 Quick Start
### Prerequisites
- **Rust** 1.70+ (edition 2021)
- **PostgreSQL** with [TimescaleDB](https://docs.timescale.com/) extension
- (Recommended) Third-party rotating proxy with multi-port support for parallel downloads
### Environment Setup
```bash
cp .env.example .env
# Edit .env with your values
```
`.env.example`:
```env
RUST_LOG="INFO,binance_sdk::common::utils=off,binance_sdk::common::websocket=off"
DATABASE_URL="postgres://user:password@host:5432/crypto_database"
TRACKED_SYMBOL=[BTCUSDT,XRPUSDT,BNBUSDT,SOLUSDT,ETHUSDT]
# Proxy settings (optional — leave empty for direct connection)
PROXY_HOST=dc.your-proxy-provider.com
PROXY_USERNAME=your_username
PROXY_PASSWORD=your_password
PROXY_PROTOCOL=https
PROXY_PORT_START=10000
PROXY_PORT_END=10099
```
> [!WARNING]
> **Proxy is strongly recommended.** The backend downloads historical data from [Binance Data Archive](https://data.binance.vision/) for all tracked symbols. With a multi-port proxy pool (e.g. 100 concurrent connections), a full sync completes in minutes. **Without a proxy, syncing may take several days** due to single-connection rate limits. If `PROXY_HOST` is left empty, the backend falls back to a single direct connection.
### Run
```bash
# Development
cargo run
# Release build
cargo build --release
./target/release/backend
# Run tests
cargo test
```
### Docker
```bash
# Build image
docker build -t backend .
# Using docker-compose (connects to existing `cycle` network)
docker compose up -d
```
## 📡 API Reference
### KlineChart API
| Method | Endpoint | Description |
|--------|----------|-------------|
| **GET** | `/api/klines/{symbol}` | Query K-line data |
| **GET** | `/api/symbols` | List all tracked symbols |
| **POST** | `/api/symbols` | Add symbol to tracking (triggers backfill) |
| **DELETE** | `/api/symbols/{symbol}` | Remove symbol from tracking |
| **GET** | `/api/status` | Get scheduler status |
**Query Parameters** for `/api/klines/{symbol}`:
| Param | Type | Default | Description |
|-------|------|---------|-------------|
| `limit` | `i64` | `800` | Number of candles to return |
| `interval` | `string` | `1m` | Timeframe: `1m`, `5m`, `15m`, `1h`, `4h`, `1d`, `1w`, `1M` |
| `end_time` | `i64` | *now* | Unix timestamp (ms) upper bound |
<details>
<summary>📄 Response Example</summary>
```json
{
"success": true,
"data": [
{
"symbol": "BTCUSDT",
"timestamp": 1704067200000,
"open": 42000.0,
"high": 42100.0,
"low": 41900.0,
"close": 42050.0,
"volume": 1000.5,
"taker_buy_volume": 600.3,
"net_volume": 200.1,
"is_closed": true
}
]
}
```
</details>
---
### TradingView UDF API
Fully compatible with the [TradingView UDF Datafeed API](https://www.tradingview.com/charting-library-docs/latest/connecting_data/UDF/).
| Method | Endpoint | Description |
|--------|----------|-------------|
| **GET** | `/config` | Datafeed configuration |
| **GET** | `/time` | Server time (seconds) |
| **GET** | `/symbols` | Resolve symbol info |
| **GET** | `/search` | Search symbols |
| **GET** | `/history` | Historical OHLCV data (includes `nv` and `tbv`) |
| **GET** | `/tracked-symbols` | List configured symbols |
| **GET** | `/daily-opens` | Daily open prices for all symbols |
| **WS** | `/ws` | Real-time K-line push via WebSocket |
<details>
<summary>📄 History Response Example</summary>
```json
{
"s": "ok",
"t": [1704067200, 1704153600],
"o": [42000.0, 42050.0],
"h": [42100.0, 42200.0],
"l": [41900.0, 41950.0],
"c": [42050.0, 42150.0],
"v": [1000.5, 1200.3],
"nv": [200.1, -150.5],
"tbv": [600.3, 525.4]
}
```
</details>
<details>
<summary>📄 WebSocket Protocol</summary>
**Subscribe:**
```json
{ "type": "subscribe", "data": { "symbols": ["BTCUSDT", "ETHUSDT"] } }
```
**Kline Update (server → client):**
```json
{ "type": "kline", "data": { "symbol": "BTCUSDT", "timestamp": 1704067200000, "open": 42000.0, "high": 42100.0, "low": 41900.0, "close": 42050.0, "volume": 1000.5, "taker_buy_volume": 600.3, "net_volume": 200.1, "is_closed": false } }
```
**Keepalive:** `ping` / `pong`
</details>
---
### Canvas API (Drawing Persistence)
Save and load chart drawings per symbol to the local filesystem.
| Method | Endpoint | Description |
|--------|----------|-------------|
| **GET** | `/canvas/list` | List saved canvases for a symbol |
| **GET** | `/canvas/load` | Load canvas drawings |
| **POST** | `/canvas/save` | Save canvas drawings |
| **DELETE** | `/canvas/delete` | Delete a canvas |
## ⚙️ Core Components
### BinanceCollector
- Connects to Binance WebSocket combined streams for real-time 1m K-line data
- Supports up to **50 symbols per connection** (Binance limit); auto-splits into multiple connections
- **Auto-reconnect** with gap detection — backfills missed data on disconnect
- REST API sync with rate limiting (150ms interval, ~480 req/min)
### HistoricalDownloader
- **3-tier download strategy**: Monthly ZIP → Daily ZIP → REST API (fastest to slowest)
- Downloads from [Binance Data Archive](https://data.binance.vision/) for bulk historical data
- Concurrent downloads across proxy pool for maximum throughput
- CSV parsing from ZIP archives
### DatabaseHandler
- **TimescaleDB** hypertable for time-series optimization
- **Batch insert**: 100 candles or 5-second flush timeout
- `time_bucket` aggregation for multi-timeframe queries (1m → 1M)
- Gap detection and data integrity checks
- Data cutoff: only syncs data from **2024-01-01 UTC** onwards
### Scheduler
- Command-based control via `mpsc` channels:
- `AddSymbol` — backfill + restart collector
- `RemoveSymbol` — deactivate + restart collector
- `RestartCollector` / `GetStatus` / `Shutdown`
- Manages full lifecycle: symbol tracking → historical backfill → real-time streaming
## 📦 Dependencies
| Crate | Purpose |
|-------|---------|
| `axum` | Web framework with WebSocket support |
| `sqlx` | Async PostgreSQL / TimescaleDB driver |
| `tokio` | Async runtime |
| `binance-sdk` | Official Binance connector (USDS futures + spot) |
| `tokio-tungstenite` | WebSocket client for Binance streams |
| `reqwest` | HTTP client for REST API & archive downloads |
| `tower-http` | CORS middleware |
| `serde` / `serde_json` | Serialization |
| `chrono` | Date/time handling |
| `csv` / `zip` | Historical data archive parsing |
| `thiserror` | Custom error types |
| `env_logger` | Logging |
## 📜 License
MIT
Generated By Claude Opus 4.6
+17
View File
@@ -0,0 +1,17 @@
services:
backend:
image: ghcr.io/exchanges-lab/statistic/backend:v1.2
container_name: backend
networks:
- cycle
#ports:
# - "3000:3000"
env_file:
- .env
volumes:
- ./storage:/storage
restart: unless-stopped
networks:
cycle:
external: true
+47
View File
@@ -0,0 +1,47 @@
-- Create klines table for 1-minute candlestick data
CREATE TABLE IF NOT EXISTS klines_1m (
symbol VARCHAR(20) NOT NULL,
timestamp BIGINT NOT NULL,
open DOUBLE PRECISION NOT NULL,
high DOUBLE PRECISION NOT NULL,
low DOUBLE PRECISION NOT NULL,
close DOUBLE PRECISION NOT NULL,
volume DOUBLE PRECISION NOT NULL,
taker_buy_volume DOUBLE PRECISION NOT NULL,
net_volume DOUBLE PRECISION NOT NULL,
PRIMARY KEY (symbol, timestamp)
);
-- Convert to TimescaleDB hypertable for better time-series performance
-- chunk_time_interval = 7 days (in milliseconds: 7 * 24 * 60 * 60 * 1000 = 604800000)
SELECT create_hypertable('klines_1m', 'timestamp',
chunk_time_interval => 604800000,
if_not_exists => TRUE
);
-- Create index for faster queries by symbol
CREATE INDEX IF NOT EXISTS idx_klines_1m_symbol ON klines_1m (symbol, timestamp DESC);
-- Table for tracking active symbols
CREATE TABLE IF NOT EXISTS tracked_symbols (
symbol VARCHAR(20) PRIMARY KEY,
added_at TIMESTAMP DEFAULT NOW(),
is_active BOOLEAN DEFAULT TRUE
);
-- Insert some default symbols
INSERT INTO tracked_symbols (symbol) VALUES
('btcusdt'),
('ethusdt')
ON CONFLICT (symbol) DO NOTHING;
-- Useful queries:
-- Get latest timestamp for a symbol
-- SELECT MAX(timestamp) FROM klines_1m WHERE symbol = 'BTCUSDT';
-- Get latest N candles for a symbol
-- SELECT * FROM klines_1m WHERE symbol = 'BTCUSDT' ORDER BY timestamp DESC LIMIT 100;
-- Get all active symbols
-- SELECT symbol FROM tracked_symbols WHERE is_active = TRUE;
+137
View File
@@ -0,0 +1,137 @@
use backend::{BinanceCollector, CandleData, DatabaseHandler};
use log::{info, error};
use dotenv::dotenv;
use tokio::sync::mpsc;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
#[tokio::main]
async fn main() {
dotenv().ok();
env_logger::init();
// 1. Connect to database
let database_url = std::env::var("DATABASE_URL")
.expect("DATABASE_URL must be set");
info!("Connecting to database...");
let db = Arc::new(
DatabaseHandler::new(&database_url)
.await
.expect("Failed to connect to database")
);
// 2. Get all symbols (filter USDT pairs only)
info!("Fetching all symbols from Binance...");
let all_symbols = BinanceCollector::get_symbol().await.unwrap();
let symbols: Vec<_> = all_symbols
.into_iter()
.filter(|s| s.symbol.ends_with("USDT"))
.collect();
info!("Got {} USDT symbols to sync", symbols.len());
// 3. Build clients
info!("Building clients...");
let clients = BinanceCollector::build_clients().await;
info!("Built {} working clients", clients.len());
if clients.is_empty() {
error!("No working clients, exiting");
return;
}
// 4. Sync each symbol one by one
let total_symbols = symbols.len();
for (i, symbol) in symbols.into_iter().enumerate() {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as i64;
let start_time = symbol.start_timestamp;
let total_range_ms = now - start_time;
// Estimate total candles (1 candle per minute)
let estimated_total = (total_range_ms / 60000) as u64;
info!("[{}/{}] Starting sync for {} (estimated {} candles)",
i + 1, total_symbols, symbol.symbol, estimated_total);
// Create channels
let (tx, rx) = mpsc::channel::<CandleData>(100000);
let (db_tx, db_rx) = mpsc::channel::<CandleData>(100000);
// Progress tracking
let candle_count = Arc::new(AtomicU64::new(0));
let last_progress = Arc::new(AtomicU64::new(0));
// Progress monitor and forward task
let symbol_name = symbol.symbol.clone();
let candle_count_clone = candle_count.clone();
let last_progress_clone = last_progress.clone();
let idx = i + 1;
let progress_handle = tokio::spawn(async move {
let mut rx = rx;
while let Some(candle) = rx.recv().await {
let count = candle_count_clone.fetch_add(1, Ordering::Relaxed) + 1;
// Calculate progress based on received candles vs estimated total
let progress = if estimated_total > 0 {
((count as f64 / estimated_total as f64) * 100.0).min(100.0) as u64
} else {
100
};
// Report every 20%
let last = last_progress_clone.load(Ordering::Relaxed);
let milestone = (progress / 20) * 20;
if milestone > last && milestone <= 100 {
if last_progress_clone.compare_exchange(
last, milestone, Ordering::Relaxed, Ordering::Relaxed
).is_ok() {
info!("[{}/{}] {} progress: {}% ({}/{} candles)",
idx, total_symbols, symbol_name, milestone, count, estimated_total);
}
}
// Forward to database
let _ = db_tx.send(candle).await;
}
});
// Database consumer
let db_clone = db.clone();
let consumer_handle = tokio::spawn(async move {
db_clone.start_consumer(db_rx).await;
});
// Sync this symbol
let result = BinanceCollector::sync_from_scratch(
symbol.symbol.clone(),
symbol.start_timestamp,
clients.clone(),
tx,
).await;
// Wait for tasks to finish
let _ = progress_handle.await;
let _ = consumer_handle.await;
let final_count = candle_count.load(Ordering::Relaxed);
match result {
Ok(count) => {
info!("[{}/{}] {} completed: {} candles",
i + 1, total_symbols, symbol.symbol, count);
}
Err(e) => {
error!("[{}/{}] {} failed: {} (received: {} candles)",
i + 1, total_symbols, symbol.symbol, e, final_count);
}
}
}
info!("All symbols synced!");
}
+162
View File
@@ -0,0 +1,162 @@
use backend::{BinanceCollector, CandleData, DatabaseHandler, HistoricalDownloader};
use log::{info, error};
use dotenv::dotenv;
use tokio::sync::mpsc;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
#[tokio::main]
async fn main() {
dotenv().ok();
env_logger::init();
// 1. Connect to database
let database_url = std::env::var("DATABASE_URL")
.expect("DATABASE_URL must be set");
info!("Connecting to database...");
let db = Arc::new(
DatabaseHandler::new(&database_url)
.await
.expect("Failed to connect to database")
);
// 2. Get all USDT symbols
info!("Fetching all symbols from Binance...");
let all_symbols = BinanceCollector::get_symbol().await.unwrap();
let symbols: Vec<_> = all_symbols
.into_iter()
.filter(|s| s.symbol.ends_with("USDT"))
.collect();
info!("Got {} USDT symbols to sync", symbols.len());
// 3. Build clients
info!("Building download clients (with proxies)...");
let download_clients = HistoricalDownloader::build_clients().await;
info!("Building API clients (for recent data)...");
let api_clients = BinanceCollector::build_clients().await;
info!("Built {} download clients, {} API clients", download_clients.len(), api_clients.len());
// Get last complete month timestamp
let archive_end = HistoricalDownloader::last_complete_month_end();
info!("Archive data available until: {}", archive_end);
// Use the global cutoff timestamp
info!("Data cutoff: 2024-01-01 00:00:00 UTC (timestamp: {})", backend::DATA_CUTOFF_TIMESTAMP);
// 4. Sync each symbol
let total_symbols = symbols.len();
for (i, symbol) in symbols.into_iter().enumerate() {
info!("========================================");
info!("[{}/{}] {} - Starting fast sync", i + 1, total_symbols, symbol.symbol);
// Check if we have data already
let latest_ts = db.get_latest_timestamp(&symbol.symbol).await.ok().flatten();
let start_time = match latest_ts {
Some(ts) => {
info!("[{}/{}] {} - Has data until {}, continuing from there",
i + 1, total_symbols, symbol.symbol, ts);
ts + 60000
}
None => {
// No existing data, use the later of symbol start or cutoff
let actual_start = symbol.start_timestamp.max(backend::DATA_CUTOFF_TIMESTAMP);
if symbol.start_timestamp < backend::DATA_CUTOFF_TIMESTAMP {
info!("[{}/{}] {} - Original start is before 2024-01-01, starting from cutoff instead",
i + 1, total_symbols, symbol.symbol);
}
actual_start
}
};
// Create channels
let (tx, rx) = mpsc::channel::<CandleData>(500000);
let (db_tx, db_rx) = mpsc::channel::<CandleData>(500000);
// Progress tracking
let candle_count = Arc::new(AtomicU64::new(0));
let candle_count_clone = candle_count.clone();
let symbol_name = symbol.symbol.clone();
let idx = i + 1;
// Progress forwarder
let progress_handle = tokio::spawn(async move {
let mut rx = rx;
let mut last_report = 0u64;
while let Some(candle) = rx.recv().await {
let count = candle_count_clone.fetch_add(1, Ordering::Relaxed) + 1;
if count - last_report >= 500000 {
info!("[{}/{}] {} - Progress: {} candles", idx, total_symbols, symbol_name, count);
last_report = count;
}
let _ = db_tx.send(candle).await;
}
});
// DB consumer (batch insert)
let db_clone = db.clone();
let consumer_handle = tokio::spawn(async move {
db_clone.start_consumer(db_rx).await;
});
// STEP 1: Download historical data from archive (with proxies)
if start_time < archive_end && !download_clients.is_empty() {
info!("[{}/{}] {} - Downloading from archive with {} proxies...",
i + 1, total_symbols, symbol.symbol, download_clients.len());
match HistoricalDownloader::download_symbol_with_clients(
&symbol.symbol,
start_time,
&download_clients,
tx.clone(),
).await {
Ok(count) => info!("[{}/{}] {} - Archive download: {} candles",
i + 1, total_symbols, symbol.symbol, count),
Err(e) => error!("[{}/{}] {} - Archive download failed: {}",
i + 1, total_symbols, symbol.symbol, e),
}
}
// STEP 2: Sync recent data via API
let api_start = archive_end.max(start_time);
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as i64;
if now - api_start > 60000 && !api_clients.is_empty() {
info!("[{}/{}] {} - Syncing recent data via API...", i + 1, total_symbols, symbol.symbol);
match BinanceCollector::sync_from_scratch(
symbol.symbol.clone(),
api_start,
api_clients.clone(),
tx.clone(),
).await {
Ok(count) => info!("[{}/{}] {} - API sync: {} candles",
i + 1, total_symbols, symbol.symbol, count),
Err(e) => error!("[{}/{}] {} - API sync failed: {}",
i + 1, total_symbols, symbol.symbol, e),
}
}
// Close channel and wait
drop(tx);
let _ = progress_handle.await;
let _ = consumer_handle.await;
let final_count = candle_count.load(Ordering::Relaxed);
info!("[{}/{}] {} - Complete: {} total candles",
i + 1, total_symbols, symbol.symbol, final_count);
}
info!("========================================");
info!("All symbols synced!");
}
+709
View File
@@ -0,0 +1,709 @@
use crate::error::*;
use crate::structs::*;
use crate::historical_downloader::HistoricalDownloader;
use binance_sdk::{
config::{ConfigurationRestApi, ProxyConfig, ProxyAuth},
derivatives_trading_usds_futures::{
DerivativesTradingUsdsFuturesRestApi,
rest_api::{
RestApi,
KlineCandlestickDataIntervalEnum, KlineCandlestickDataParams, KlineCandlestickDataResponseItemInner,
},
websocket_streams::KlineCandlestickStreamsResponseK,
},
};
use futures::StreamExt;
use log::{debug, error, info, warn};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{mpsc, RwLock};
use reqwest::Client;
use tokio_tungstenite::{connect_async, tungstenite::Message};
use rand::Rng;
use serde::Deserialize;
/// Read proxy configuration from environment variables.
/// Returns None if PROXY_HOST is not set (direct connection).
fn proxy_config_from_env(port: u16) -> Option<ProxyConfig> {
let host = std::env::var("PROXY_HOST").ok()?;
if host.is_empty() {
return None;
}
let username = std::env::var("PROXY_USERNAME").unwrap_or_default();
let password = std::env::var("PROXY_PASSWORD").unwrap_or_default();
let protocol = std::env::var("PROXY_PROTOCOL").unwrap_or_else(|_| "https".to_string());
let auth = if username.is_empty() {
None
} else {
Some(ProxyAuth {
username,
password,
})
};
Some(ProxyConfig {
host,
port,
protocol: Some(protocol),
auth,
})
}
// Combined stream response wrapper
#[derive(Debug, Deserialize)]
struct CombinedStreamWrapper {
#[allow(dead_code)]
stream: String,
data: CombinedStreamData,
}
#[derive(Debug, Deserialize)]
struct CombinedStreamData {
s: Option<String>,
k: Option<KlineCandlestickStreamsResponseK>,
}
const BATCH_SIZE: i64 = 1000;
const ONE_MINUTE_MS: i64 = 60_000;
const MAX_SYMBOLS_PER_WS: usize = 50; // Binance limit per WebSocket connection
// Rate limit: 2400 weight/min, limit=1000 costs 5 weight
// 2400/5 = 480 requests/min = 8 req/sec = 125ms interval
// Use 150ms for safety margin
const REQUEST_INTERVAL_MS: u64 = 150;
pub struct BinanceCollector {
pub symbols: Vec<String>,
rest_client: RestApi,
last_closed_timestamps: Arc<RwLock<HashMap<String, i64>>>,
}
impl BinanceCollector {
pub fn new(symbols: Vec<String>) -> Self {
assert!(!symbols.is_empty(), "symbols cannot be empty");
let symbols: Vec<String> = symbols.into_iter().map(|s| s.to_lowercase()).collect();
let port: u16 = rand::rng().random_range(10036..=10066);
let mut config_builder = ConfigurationRestApi::builder()
.timeout(10000);
if let Some(proxy) = proxy_config_from_env(port) {
config_builder = config_builder.proxy(proxy);
}
let rest_client_config = config_builder
.build()
.expect("Failed to initialize the rest api client");
let rest_client = DerivativesTradingUsdsFuturesRestApi::production(rest_client_config);
Self {
symbols,
rest_client,
last_closed_timestamps: Arc::new(RwLock::new(HashMap::new())),
}
}
/// Backfill K-line data for a specific symbol from start_time to end_time.
/// Uses limit=1000 per request (weight=5, ~480 requests/min allowed).
/// Returns total number of candles fetched.
pub async fn backfill(
&self,
symbol: &str,
start_time: i64,
end_time: i64,
candle_tx: &mpsc::Sender<CandleData>,
) -> Result<u64, CollectorError> {
let symbol_upper = symbol.to_uppercase();
let symbol_lower = symbol.to_lowercase();
let mut current_start = start_time;
let mut total_count: u64 = 0;
info!(
"Starting backfill for {} from {} to {}",
symbol_upper, start_time, end_time
);
while current_start < end_time {
let params = KlineCandlestickDataParams::builder(
symbol_upper.clone(),
KlineCandlestickDataIntervalEnum::Interval1m,
)
.start_time(Some(current_start))
.end_time(Some(end_time))
.limit(Some(BATCH_SIZE))
.build()
.map_err(|e| CollectorError::RestApiError(e.to_string()))?;
let response = self.rest_client
.kline_candlestick_data(params)
.await
.map_err(|e| CollectorError::RestApiError(e.to_string()))?;
let klines = response.data().await
.map_err(|e| CollectorError::RestApiError(e.to_string()))?;
let batch_count = klines.len();
if batch_count == 0 {
break;
}
let mut last_timestamp = current_start;
for kline in &klines {
if let Some(candle) = Self::parse_rest_kline(&symbol_upper, kline) {
last_timestamp = candle.timestamp;
if let Err(e) = candle_tx.send(candle).await {
warn!("Failed to send candle: {}", e);
}
total_count += 1;
}
}
// Update last_closed_timestamp for this symbol
{
let mut timestamps = self.last_closed_timestamps.write().await;
timestamps.insert(symbol_lower.clone(), last_timestamp);
}
debug!(
"Fetched {} klines for {}, total: {}, last_ts: {}",
batch_count, symbol_upper, total_count, last_timestamp
);
current_start = last_timestamp + ONE_MINUTE_MS;
if batch_count < BATCH_SIZE as usize {
break;
}
tokio::time::sleep(tokio::time::Duration::from_millis(REQUEST_INTERVAL_MS)).await;
}
info!(
"Backfill completed for {}: {} candles",
symbol_upper, total_count
);
Ok(total_count)
}
/// Backfill all symbols in the collector.
pub async fn backfill_all(
&self,
start_time: i64,
end_time: i64,
candle_tx: &mpsc::Sender<CandleData>,
) -> Result<u64, CollectorError> {
let mut total_count: u64 = 0;
for symbol in &self.symbols.clone() {
let count = self.backfill(symbol, start_time, end_time, candle_tx).await?;
total_count += count;
}
Ok(total_count)
}
/// Start WebSocket streams for real-time K-line data on all symbols.
/// Creates multiple connections if symbols > MAX_SYMBOLS_PER_WS.
/// Automatically reconnects and backfills missed data on disconnect.
pub async fn start_stream(
&self,
candle_tx: mpsc::Sender<CandleData>,
) -> Result<(), CollectorError> {
let total_symbols = self.symbols.len();
let num_connections = (total_symbols + MAX_SYMBOLS_PER_WS - 1) / MAX_SYMBOLS_PER_WS;
info!(
"Starting {} WebSocket connections for {} symbols",
num_connections, total_symbols
);
// Split symbols into batches
let batches: Vec<Vec<String>> = self.symbols
.chunks(MAX_SYMBOLS_PER_WS)
.map(|chunk| chunk.to_vec())
.collect();
loop {
// Start all WebSocket connections in parallel
let mut handles = Vec::new();
for (i, batch) in batches.iter().enumerate() {
let batch_symbols = batch.clone();
let tx = candle_tx.clone();
let last_ts = self.last_closed_timestamps.clone();
let handle = tokio::spawn(async move {
Self::run_single_ws_connection(i, batch_symbols, tx, last_ts).await
});
handles.push(handle);
}
// Wait for any connection to fail
let mut all_ok = true;
for (i, handle) in handles.into_iter().enumerate() {
match handle.await {
Ok(Ok(())) => info!("WS connection {} ended normally", i),
Ok(Err(e)) => {
error!("WS connection {} error: {}", i, e);
all_ok = false;
}
Err(e) => {
error!("WS connection {} join error: {}", i, e);
all_ok = false;
}
}
}
if all_ok {
break;
}
// Reconnect after error
error!("WebSocket error. Reconnecting in 5 seconds...");
tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
// Backfill missed data
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as i64;
let timestamps = self.last_closed_timestamps.read().await.clone();
for symbol in &self.symbols {
if let Some(&last_ts) = timestamps.get(symbol) {
if now - last_ts > ONE_MINUTE_MS {
info!("Backfilling missed data for {} from {}", symbol, last_ts);
if let Err(e) = self.backfill(symbol, last_ts, now, &candle_tx).await {
warn!("Backfill failed for {}: {}", symbol, e);
}
}
}
}
}
Ok(())
}
/// Run a single WebSocket connection for a batch of symbols (combined streams)
async fn run_single_ws_connection(
connection_id: usize,
symbols: Vec<String>,
candle_tx: mpsc::Sender<CandleData>,
last_closed_timestamps: Arc<RwLock<HashMap<String, i64>>>,
) -> Result<(), CollectorError> {
// Build combined stream URL: wss://fstream.binance.com/stream?streams=symbol1@kline_1m/symbol2@kline_1m
let streams: Vec<String> = symbols.iter()
.map(|s| format!("{}@kline_1m", s.to_lowercase()))
.collect();
let url = format!("wss://fstream.binance.com/stream?streams={}", streams.join("/"));
let (ws_stream, _) = connect_async(&url)
.await
.map_err(|e| CollectorError::ConnectionFailed(e.to_string()))?;
info!("WS {} connected with {} symbols (combined stream)", connection_id, symbols.len());
let (_, mut read) = ws_stream.split();
// Process incoming messages
while let Some(msg_result) = read.next().await {
match msg_result {
Ok(Message::Text(text)) => {
// Combined stream format: {"stream":"btcusdt@kline_1m","data":{...}}
if let Ok(wrapper) = serde_json::from_str::<CombinedStreamWrapper>(&text) {
if let Some(k) = wrapper.data.k {
let symbol = wrapper.data.s.unwrap_or_default();
let candle = Self::parse_ws_kline(&symbol, &k);
if let Err(e) = candle_tx.send(candle.clone()).await {
warn!("Failed to send candle: {}", e);
}
if candle.is_closed {
let mut timestamps = last_closed_timestamps.write().await;
timestamps.insert(candle.symbol.to_lowercase(), candle.timestamp);
debug!(
"Kline closed: {} ts={} c={:.2} net_vol={:.4}",
candle.symbol, candle.timestamp, candle.close, candle.net_volume
);
}
}
}
}
Ok(Message::Ping(data)) => {
debug!("WS {} received ping", connection_id);
// tungstenite auto-responds to pings
let _ = data;
}
Ok(Message::Close(_)) => {
info!("WS {} received close", connection_id);
break;
}
Ok(_) => {}
Err(e) => {
error!("WS {} error: {}", connection_id, e);
break;
}
}
}
Err(CollectorError::ConnectionFailed("Disconnected".to_string()))
}
fn parse_rest_kline(symbol: &str, kline: &Vec<KlineCandlestickDataResponseItemInner>) -> Option<CandleData> {
if kline.len() < 12 {
return None;
}
let timestamp = match &kline[0] {
KlineCandlestickDataResponseItemInner::Integer(v) => *v,
_ => return None,
};
let open = Self::parse_string_field(&kline[1])?;
let high = Self::parse_string_field(&kline[2])?;
let low = Self::parse_string_field(&kline[3])?;
let close = Self::parse_string_field(&kline[4])?;
let volume = Self::parse_string_field(&kline[5])?;
let taker_buy_volume = Self::parse_string_field(&kline[9])?;
let net_volume = CandleData::calculate_net_volume(volume, taker_buy_volume);
Some(CandleData {
symbol: symbol.to_uppercase(),
timestamp,
open,
high,
low,
close,
volume,
taker_buy_volume,
net_volume,
is_closed: true,
})
}
fn parse_string_field(item: &KlineCandlestickDataResponseItemInner) -> Option<f64> {
match item {
KlineCandlestickDataResponseItemInner::String(s) => s.parse().ok(),
_ => None,
}
}
fn parse_ws_kline(symbol: &str, k: &KlineCandlestickStreamsResponseK) -> CandleData {
let timestamp = k.t.unwrap_or(0);
let open = k.o.as_ref().and_then(|s| s.parse().ok()).unwrap_or(0.0);
let high = k.h.as_ref().and_then(|s| s.parse().ok()).unwrap_or(0.0);
let low = k.l.as_ref().and_then(|s| s.parse().ok()).unwrap_or(0.0);
let close = k.c.as_ref().and_then(|s| s.parse().ok()).unwrap_or(0.0);
let volume = k.v.as_ref().and_then(|s| s.parse().ok()).unwrap_or(0.0);
let taker_buy_volume = k.v_uppercase.as_ref().and_then(|s| s.parse().ok()).unwrap_or(0.0);
let is_closed = k.x.unwrap_or(false);
let net_volume = CandleData::calculate_net_volume(volume, taker_buy_volume);
CandleData {
symbol: symbol.to_uppercase(),
timestamp,
open,
high,
low,
close,
volume,
taker_buy_volume,
net_volume,
is_closed,
}
}
/// Get all tradable symbol in current market
pub async fn get_symbol() -> Result<Vec<Symbol>, CollectorError> {
let port: u16 = rand::rng().random_range(10036..=10066);
let mut config_builder = ConfigurationRestApi::builder()
.timeout(10000);
if let Some(proxy) = proxy_config_from_env(port) {
config_builder = config_builder.proxy(proxy);
}
let rest_client_config = config_builder
.build()
.expect("Failed to initialize the rest api client");
let rest_client = DerivativesTradingUsdsFuturesRestApi::production(rest_client_config);
let mut symbol_list = Vec::new();
let response = rest_client
.exchange_information()
.await
.map_err(|e| CollectorError::GetSymbolError(e.to_string()))?;
let data = response.data().await.unwrap();
let symbol_vec = data.symbols.unwrap();
for item in symbol_vec {
if item.contract_type.unwrap() == "PERPETUAL" {
let symbol = Symbol {
symbol: item.symbol.unwrap(),
start_timestamp: item.onboard_date.unwrap(),
};
symbol_list.push(symbol);
}
}
Ok(symbol_list)
}
/// Build proxy list (port 10000-10099) and create clients
pub async fn build_clients() -> Vec<Arc<RestApi>> {
let mut handles = Vec::new();
let port_start: u16 = std::env::var("PROXY_PORT_START")
.ok().and_then(|v| v.parse().ok()).unwrap_or(10000);
let port_end: u16 = std::env::var("PROXY_PORT_END")
.ok().and_then(|v| v.parse().ok()).unwrap_or(10099);
for port in port_start..=port_end {
let handle = tokio::spawn(async move {
let mut config_builder = ConfigurationRestApi::builder()
.timeout(5000);
if let Some(proxy) = proxy_config_from_env(port) {
config_builder = config_builder.proxy(proxy);
}
let config = config_builder
.build()
.ok()?;
let client = DerivativesTradingUsdsFuturesRestApi::production(config);
// Test connection
if client.check_server_time().await.ok()?.data().await.is_ok() {
Some(Arc::new(client))
} else {
None
}
});
handles.push(handle);
}
let mut clients = Vec::new();
for handle in handles {
if let Ok(Some(client)) = handle.await {
clients.push(client);
}
}
info!("Built {} working clients", clients.len());
clients
}
/// Comprehensive sync: Monthly ZIP → Daily ZIP → API (fastest to slowest)
pub async fn sync_comprehensive(
symbol: String,
start_time: i64,
api_clients: Vec<Arc<RestApi>>,
http_clients: &[Arc<Client>],
candle_tx: mpsc::Sender<CandleData>,
) -> Result<u64, CollectorError> {
let mut total = 0u64;
let mut current_start = start_time;
// Phase 1: Monthly ZIP download (complete months only)
let month_end = HistoricalDownloader::last_complete_month_end();
if current_start < month_end && !http_clients.is_empty() {
info!("{}: Phase 1 - Monthly ZIP download", symbol);
match HistoricalDownloader::download_symbol_with_clients(
&symbol,
current_start,
http_clients,
candle_tx.clone(),
).await {
Ok(count) => {
total += count;
if count > 0 {
current_start = month_end;
info!("{}: Monthly ZIP done, {} candles", symbol, count);
}
}
Err(e) => warn!("{}: Monthly ZIP failed: {}", symbol, e),
}
}
// Phase 2: Daily ZIP download (current month's completed days)
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as i64;
let today_start = (now / (24 * 60 * 60 * 1000)) * (24 * 60 * 60 * 1000);
if current_start < today_start && !http_clients.is_empty() {
info!("{}: Phase 2 - Daily ZIP download", symbol);
match HistoricalDownloader::download_days_with_clients(
&symbol,
current_start,
http_clients,
candle_tx.clone(),
).await {
Ok(count) => {
total += count;
if count > 0 {
current_start = today_start;
info!("{}: Daily ZIP done, {} candles", symbol, count);
}
}
Err(e) => warn!("{}: Daily ZIP failed: {}", symbol, e),
}
}
// Phase 3: API for remaining (today's data)
if current_start < now && !api_clients.is_empty() {
info!("{}: Phase 3 - API sync for today", symbol);
match Self::sync_from_scratch(
symbol.clone(),
current_start,
api_clients,
candle_tx,
).await {
Ok(count) => {
total += count;
info!("{}: API sync done, {} candles", symbol, count);
}
Err(e) => warn!("{}: API sync failed: {}", symbol, e),
}
}
info!("{}: Comprehensive sync complete, total {} candles", symbol, total);
Ok(total)
}
/// Sync single symbol using ALL clients in parallel (each client handles a time segment)
pub async fn sync_from_scratch(
symbol: String,
start_time: i64,
clients: Vec<Arc<RestApi>>,
candle_tx: mpsc::Sender<CandleData>,
) -> Result<u64, CollectorError> {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as i64;
let num_clients = clients.len();
if num_clients == 0 {
return Err(CollectorError::RestApiError("No clients".to_string()));
}
let total_duration = now - start_time;
let segment_size = total_duration / num_clients as i64;
info!(
"Syncing {} with {} clients, {} -> {} ({} ms per segment)",
symbol, num_clients, start_time, now, segment_size
);
// Each client gets a time segment
let mut handles = Vec::new();
for (i, client) in clients.into_iter().enumerate() {
let seg_start = start_time + (i as i64 * segment_size);
let seg_end = if i == num_clients - 1 {
now // Last segment goes to now
} else {
start_time + ((i + 1) as i64 * segment_size)
};
let sym = symbol.clone();
let tx = candle_tx.clone();
handles.push(tokio::spawn(async move {
Self::sync_segment(sym, seg_start, seg_end, client, tx).await
}));
}
// Wait and sum results
let mut total = 0u64;
for (i, h) in handles.into_iter().enumerate() {
match h.await {
Ok(Ok(count)) => {
total += count;
debug!("Client {} done: {} candles", i, count);
}
Ok(Err(e)) => error!("Client {} error: {}", i, e),
Err(e) => error!("Client {} join error: {}", i, e),
}
}
info!("{} sync complete: {} candles", symbol, total);
Ok(total)
}
/// Sync a specific time segment
async fn sync_segment(
symbol: String,
start_time: i64,
end_time: i64,
client: Arc<RestApi>,
candle_tx: mpsc::Sender<CandleData>,
) -> Result<u64, CollectorError> {
let mut current = start_time;
let mut count = 0u64;
while current < end_time {
let params = KlineCandlestickDataParams::builder(
symbol.clone(),
KlineCandlestickDataIntervalEnum::Interval1m,
)
.start_time(Some(current))
.end_time(Some(end_time))
.limit(Some(BATCH_SIZE))
.build()
.map_err(|e| CollectorError::RestApiError(e.to_string()))?;
let res = client
.kline_candlestick_data(params)
.await
.map_err(|e| CollectorError::RestApiError(e.to_string()))?;
let klines = res.data().await
.map_err(|e| CollectorError::RestApiError(e.to_string()))?;
if klines.is_empty() { break; }
let batch_len = klines.len();
for kline in &klines {
if let Some(candle) = Self::parse_rest_kline(&symbol, kline) {
current = candle.timestamp + ONE_MINUTE_MS;
let _ = candle_tx.send(candle).await;
count += 1;
}
}
if batch_len < BATCH_SIZE as usize { break; }
tokio::time::sleep(tokio::time::Duration::from_millis(REQUEST_INTERVAL_MS)).await;
}
Ok(count)
}
}
+479
View File
@@ -0,0 +1,479 @@
use crate::structs::*;
use log::{debug, error, info};
use sqlx::{PgPool, postgres::PgPoolOptions};
use tokio::sync::mpsc;
use tokio::time::{Duration, interval};
const BUFFER_SIZE: usize = 100;
const FLUSH_INTERVAL_SECS: u64 = 5;
pub struct DatabaseHandler {
pool: PgPool,
}
impl DatabaseHandler {
pub async fn new(database_url: &str) -> Result<Self, sqlx::Error> {
let pool = PgPoolOptions::new()
.max_connections(20)
.min_connections(5)
.connect(database_url)
.await?;
info!("Database connected (pool: 5-20 connections)");
Ok(Self { pool })
}
/// Start consuming from rx and batch insert into database
pub async fn start_consumer(&self, mut rx: mpsc::Receiver<CandleData>) {
let mut buffer: Vec<CandleData> = Vec::with_capacity(BUFFER_SIZE);
let mut flush_timer = interval(Duration::from_secs(FLUSH_INTERVAL_SECS));
info!("Database consumer started");
loop {
tokio::select! {
// Receive candle data
candle = rx.recv() => {
match candle {
Some(c) => {
buffer.push(c);
if buffer.len() >= BUFFER_SIZE {
self.flush_buffer(&mut buffer).await;
}
}
None => {
// Channel closed, flush remaining and exit
if !buffer.is_empty() {
self.flush_buffer(&mut buffer).await;
}
info!("Database consumer stopped (channel closed)");
break;
}
}
}
// Flush on timer
_ = flush_timer.tick() => {
if !buffer.is_empty() {
self.flush_buffer(&mut buffer).await;
}
}
}
}
}
async fn flush_buffer(&self, buffer: &mut Vec<CandleData>) {
if buffer.is_empty() {
return;
}
let count = buffer.len();
match self.batch_insert(buffer).await {
Ok(_) => {
debug!("Inserted {} candles", count);
}
Err(e) => {
error!("Failed to insert {} candles: {}", count, e);
}
}
buffer.clear();
}
async fn batch_insert(&self, candles: &[CandleData]) -> Result<(), sqlx::Error> {
if candles.is_empty() {
return Ok(());
}
// Deduplicate by (symbol, timestamp) - keep last occurrence
use std::collections::HashMap;
let mut dedup_map: HashMap<(String, i64), &CandleData> = HashMap::new();
for c in candles {
dedup_map.insert((c.symbol.clone(), c.timestamp), c);
}
let unique_candles: Vec<&CandleData> = dedup_map.into_values().collect();
if unique_candles.is_empty() {
return Ok(());
}
// Build batch insert query
let mut query = String::from(
"INSERT INTO klines_1m (symbol, timestamp, open, high, low, close, volume, taker_buy_volume, net_volume) VALUES "
);
let mut values: Vec<String> = Vec::with_capacity(unique_candles.len());
for (i, _) in unique_candles.iter().enumerate() {
let idx = i * 9;
values.push(format!(
"(${}, ${}, ${}, ${}, ${}, ${}, ${}, ${}, ${})",
idx + 1, idx + 2, idx + 3, idx + 4, idx + 5, idx + 6, idx + 7, idx + 8, idx + 9
));
}
query.push_str(&values.join(", "));
query.push_str(" ON CONFLICT (symbol, timestamp) DO UPDATE SET open = EXCLUDED.open, high = EXCLUDED.high, low = EXCLUDED.low, close = EXCLUDED.close, volume = EXCLUDED.volume, taker_buy_volume = EXCLUDED.taker_buy_volume, net_volume = EXCLUDED.net_volume");
let mut query_builder = sqlx::query(&query);
for c in unique_candles {
query_builder = query_builder
.bind(&c.symbol)
.bind(c.timestamp)
.bind(c.open)
.bind(c.high)
.bind(c.low)
.bind(c.close)
.bind(c.volume)
.bind(c.taker_buy_volume)
.bind(c.net_volume);
}
query_builder.execute(&self.pool).await?;
Ok(())
}
/// Get latest timestamp for a symbol
pub async fn get_latest_timestamp(&self, symbol: &str) -> Result<Option<i64>, sqlx::Error> {
let row: Option<(Option<i64>,)> = sqlx::query_as(
"SELECT MAX(timestamp) FROM klines_1m WHERE symbol = $1"
)
.bind(symbol.to_uppercase())
.fetch_optional(&self.pool)
.await?;
Ok(row.and_then(|r| r.0))
}
/// Get today's UTC 00:00 open price for all tracked symbols
pub async fn get_daily_opens(&self) -> Result<std::collections::HashMap<String, f64>, sqlx::Error> {
use chrono::Utc;
let now = Utc::now();
let today_start = now.date_naive().and_hms_opt(0, 0, 0).unwrap().and_utc().timestamp_millis();
let rows: Vec<(String, f64)> = sqlx::query_as(
"SELECT symbol, open FROM klines_1m WHERE timestamp = $1"
)
.bind(today_start)
.fetch_all(&self.pool)
.await?;
Ok(rows.into_iter().collect())
}
/// Get all active symbols from tracked_symbols table
pub async fn get_active_symbols(&self) -> Result<Vec<String>, sqlx::Error> {
let rows: Vec<(String,)> = sqlx::query_as(
"SELECT symbol FROM tracked_symbols WHERE is_active = TRUE"
)
.fetch_all(&self.pool)
.await?;
Ok(rows.into_iter().map(|r| r.0).collect())
}
/// Get tracked symbols from TRACKED_SYMBOL env variable
/// Format: TRACKED_SYMBOL=[BTCUSDT,ETHUSDT,BNBUSDT]
pub fn get_symbols_from_env() -> Vec<String> {
std::env::var("TRACKED_SYMBOL")
.unwrap_or_default()
.trim_start_matches('[')
.trim_end_matches(']')
.split(',')
.map(|s| s.trim().to_uppercase())
.filter(|s| !s.is_empty())
.collect()
}
/// Insert candle directly (for single inserts)
pub async fn insert_candle(&self, candle: &CandleData) -> Result<(), sqlx::Error> {
sqlx::query(
"INSERT INTO klines_1m (symbol, timestamp, open, high, low, close, volume, taker_buy_volume, net_volume)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (symbol, timestamp) DO UPDATE SET
open = EXCLUDED.open, high = EXCLUDED.high, low = EXCLUDED.low,
close = EXCLUDED.close, volume = EXCLUDED.volume,
taker_buy_volume = EXCLUDED.taker_buy_volume, net_volume = EXCLUDED.net_volume"
)
.bind(&candle.symbol)
.bind(candle.timestamp)
.bind(candle.open)
.bind(candle.high)
.bind(candle.low)
.bind(candle.close)
.bind(candle.volume)
.bind(candle.taker_buy_volume)
.bind(candle.net_volume)
.execute(&self.pool)
.await?;
Ok(())
}
/// Add a new symbol to tracked_symbols
pub async fn add_symbol(&self, symbol: &str) -> Result<(), sqlx::Error> {
let symbol_lower = symbol.to_lowercase();
sqlx::query(
"INSERT INTO tracked_symbols (symbol, is_active)
VALUES ($1, TRUE)
ON CONFLICT (symbol) DO UPDATE SET is_active = TRUE"
)
.bind(&symbol_lower)
.execute(&self.pool)
.await?;
info!("Added symbol to tracking: {}", symbol_lower);
Ok(())
}
/// Remove (deactivate) a symbol from tracking
pub async fn remove_symbol(&self, symbol: &str) -> Result<(), sqlx::Error> {
let symbol_lower = symbol.to_lowercase();
sqlx::query(
"UPDATE tracked_symbols SET is_active = FALSE WHERE symbol = $1"
)
.bind(&symbol_lower)
.execute(&self.pool)
.await?;
info!("Removed symbol from tracking: {}", symbol_lower);
Ok(())
}
/// Check if a symbol is currently being tracked
pub async fn is_symbol_tracked(&self, symbol: &str) -> Result<bool, sqlx::Error> {
let symbol_lower = symbol.to_lowercase();
let row: Option<(bool,)> = sqlx::query_as(
"SELECT is_active FROM tracked_symbols WHERE symbol = $1"
)
.bind(&symbol_lower)
.fetch_optional(&self.pool)
.await?;
Ok(row.map(|r| r.0).unwrap_or(false))
}
/// Delete all kline data for a symbol (use with caution)
pub async fn delete_symbol_data(&self, symbol: &str) -> Result<u64, sqlx::Error> {
let symbol_upper = symbol.to_uppercase();
let result = sqlx::query(
"DELETE FROM klines_1m WHERE symbol = $1"
)
.bind(&symbol_upper)
.execute(&self.pool)
.await?;
let deleted = result.rows_affected();
info!("Deleted {} candles for symbol: {}", deleted, symbol_upper);
Ok(deleted)
}
/// Get klines for a symbol with limit and optional end_time
pub async fn get_klines(&self, symbol: &str, limit: i64, end_time: Option<i64>) -> Result<Vec<CandleData>, sqlx::Error> {
let symbol_upper = symbol.to_uppercase();
let rows: Vec<(String, i64, f64, f64, f64, f64, f64, f64, f64)> = if let Some(et) = end_time {
sqlx::query_as(
"SELECT symbol, timestamp, open, high, low, close, volume, taker_buy_volume, net_volume
FROM klines_1m
WHERE symbol = $1 AND timestamp <= $3
ORDER BY timestamp DESC
LIMIT $2"
)
.bind(&symbol_upper)
.bind(limit)
.bind(et)
.fetch_all(&self.pool)
.await?
} else {
sqlx::query_as(
"SELECT symbol, timestamp, open, high, low, close, volume, taker_buy_volume, net_volume
FROM klines_1m
WHERE symbol = $1
ORDER BY timestamp DESC
LIMIT $2"
)
.bind(&symbol_upper)
.bind(limit)
.fetch_all(&self.pool)
.await?
};
let mut candles: Vec<CandleData> = rows.into_iter().map(|r| CandleData {
symbol: r.0,
timestamp: r.1,
open: r.2,
high: r.3,
low: r.4,
close: r.5,
volume: r.6,
taker_buy_volume: r.7,
net_volume: r.8,
is_closed: true,
}).collect();
// Reverse to chronological order (oldest first)
candles.reverse();
Ok(candles)
}
/// Get klines aggregated to a specific interval using TimescaleDB time_bucket
pub async fn get_klines_aggregated(
&self,
symbol: &str,
interval: crate::Interval,
limit: i64,
end_time: Option<i64>
) -> Result<Vec<CandleData>, sqlx::Error> {
use crate::Interval;
// For 1m interval, just return raw data
if interval == Interval::Min1 {
return self.get_klines(symbol, limit, end_time).await;
}
let symbol_upper = symbol.to_uppercase();
// Calculate interval in milliseconds for time range filtering
let interval_ms: i64 = match interval {
Interval::Min1 => 60_000,
Interval::Min5 => 5 * 60_000,
Interval::Min15 => 15 * 60_000,
Interval::Hour1 => 60 * 60_000,
Interval::Hour4 => 4 * 60 * 60_000,
Interval::Day1 => 24 * 60 * 60_000,
Interval::Week1 => 7 * 24 * 60 * 60_000,
Interval::Month1 => 30 * 24 * 60 * 60_000,
};
// Convert interval to PostgreSQL interval string
let interval_str = match interval {
Interval::Min1 => "1 minute",
Interval::Min5 => "5 minutes",
Interval::Min15 => "15 minutes",
Interval::Hour1 => "1 hour",
Interval::Hour4 => "4 hours",
Interval::Day1 => "1 day",
Interval::Week1 => "1 week",
Interval::Month1 => "1 month",
};
// Calculate time range to limit scan (add 10% buffer)
let et = end_time.unwrap_or_else(|| chrono::Utc::now().timestamp_millis());
let time_range_needed = interval_ms * limit * 11 / 10;
let start_time = et - time_range_needed;
// Use TimescaleDB time_bucket for efficient aggregation with time range filter
let query = format!(
r#"
SELECT
$1 as symbol,
(EXTRACT(EPOCH FROM time_bucket('{interval}', to_timestamp(timestamp/1000.0))) * 1000)::bigint as bucket_ts,
(array_agg(open ORDER BY timestamp ASC))[1] as open,
max(high) as high,
min(low) as low,
(array_agg(close ORDER BY timestamp DESC))[1] as close,
sum(volume) as volume,
sum(taker_buy_volume) as taker_buy_volume,
sum(net_volume) as net_volume
FROM klines_1m
WHERE symbol = $1 AND timestamp >= $3 AND timestamp <= $4
GROUP BY bucket_ts
ORDER BY bucket_ts DESC
LIMIT $2
"#,
interval = interval_str,
);
let rows: Vec<(String, i64, f64, f64, f64, f64, f64, f64, f64)> = sqlx::query_as(&query)
.bind(&symbol_upper)
.bind(limit)
.bind(start_time)
.bind(et)
.fetch_all(&self.pool)
.await?;
let mut candles: Vec<CandleData> = rows.into_iter().map(|r| CandleData {
symbol: r.0,
timestamp: r.1,
open: r.2,
high: r.3,
low: r.4,
close: r.5,
volume: r.6,
taker_buy_volume: r.7,
net_volume: r.8,
is_closed: true,
}).collect();
// Reverse to chronological order (oldest first)
candles.reverse();
Ok(candles)
}
/// Find gaps in 1-minute kline data for a symbol.
/// Returns a list of (start_timestamp, end_timestamp) pairs representing gaps.
/// Each gap represents missing data from start_timestamp to end_timestamp (exclusive).
/// Only considers data after DATA_CUTOFF_TIMESTAMP (2024-01-01).
pub async fn find_gaps(&self, symbol: &str, cutoff_timestamp: i64) -> Result<Vec<(i64, i64)>, sqlx::Error> {
let symbol_upper = symbol.to_uppercase();
// Query to find gaps using window function
// We look for cases where the next timestamp is more than 1 minute away
// Only consider data after cutoff_timestamp
let rows: Vec<(i64, i64)> = sqlx::query_as(
r#"
WITH ordered_klines AS (
SELECT timestamp,
LEAD(timestamp) OVER (ORDER BY timestamp) as next_timestamp
FROM klines_1m
WHERE symbol = $1 AND timestamp >= $2
)
SELECT timestamp + 60000 as gap_start, next_timestamp as gap_end
FROM ordered_klines
WHERE next_timestamp - timestamp > 60000
ORDER BY timestamp
"#
)
.bind(&symbol_upper)
.bind(cutoff_timestamp)
.fetch_all(&self.pool)
.await?;
Ok(rows)
}
/// Get the earliest timestamp for a symbol
pub async fn get_earliest_timestamp(&self, symbol: &str) -> Result<Option<i64>, sqlx::Error> {
let row: Option<(Option<i64>,)> = sqlx::query_as(
"SELECT MIN(timestamp) FROM klines_1m WHERE symbol = $1"
)
.bind(symbol.to_uppercase())
.fetch_optional(&self.pool)
.await?;
Ok(row.and_then(|r| r.0))
}
/// Get count of klines for a symbol
pub async fn get_kline_count(&self, symbol: &str) -> Result<i64, sqlx::Error> {
let row: (i64,) = sqlx::query_as(
"SELECT COUNT(*) FROM klines_1m WHERE symbol = $1"
)
.bind(symbol.to_uppercase())
.fetch_one(&self.pool)
.await?;
Ok(row.0)
}
}
+47
View File
@@ -0,0 +1,47 @@
use thiserror::Error;
#[derive(Error, Debug)]
pub enum CollectorError {
#[error("WebSocket connection failed: {0}")]
ConnectionFailed(String),
#[error("REST API request failed: {0}")]
RestApiError(String),
#[error("Data parsing error: {0}")]
ParseError(String),
#[error("Invalid kline data: {0}")]
InvalidKlineData(String),
#[error("Failed to get opened symbol, with error {0}")]
GetSymbolError(String),
}
#[derive(Error, Debug)]
pub enum SchedulerError {
#[error("Database error: {0}")]
DatabaseError(String),
#[error("Collector error: {0}")]
CollectorError(String),
#[error("Backfill error: {0}")]
BackfillError(String),
#[error("No active symbols")]
NoActiveSymbols,
}
impl From<sqlx::Error> for SchedulerError {
fn from(err: sqlx::Error) -> Self {
SchedulerError::DatabaseError(err.to_string())
}
}
impl From<CollectorError> for SchedulerError {
fn from(err: CollectorError) -> Self {
SchedulerError::CollectorError(err.to_string())
}
}
+522
View File
@@ -0,0 +1,522 @@
use crate::structs::CandleData;
use crate::error::CollectorError;
use chrono::{Datelike, NaiveDate, Utc};
use log::{info, warn};
use std::io::{Cursor, Read};
use std::sync::Arc;
use tokio::sync::mpsc;
use futures::future::join_all;
use reqwest::Client;
const BASE_URL: &str = "https://data.binance.vision/data/futures/um/monthly/klines";
const DAILY_BASE_URL: &str = "https://data.binance.vision/data/futures/um/daily/klines";
/// Download historical klines from Binance data archive
pub struct HistoricalDownloader;
impl HistoricalDownloader {
/// Build proxy clients from env vars (PROXY_HOST, PROXY_USERNAME, PROXY_PASSWORD, PROXY_PROTOCOL)
/// Port range defaults to PROXY_PORT_START..PROXY_PORT_END (default 10000..10099)
/// Returns direct (no-proxy) clients if PROXY_HOST is not set.
pub async fn build_clients() -> Vec<Arc<Client>> {
let mut handles = Vec::new();
let proxy_host = std::env::var("PROXY_HOST").unwrap_or_default();
let proxy_username = std::env::var("PROXY_USERNAME").unwrap_or_default();
let proxy_password = std::env::var("PROXY_PASSWORD").unwrap_or_default();
let proxy_protocol = std::env::var("PROXY_PROTOCOL").unwrap_or_else(|_| "https".to_string());
let port_start: u16 = std::env::var("PROXY_PORT_START")
.ok().and_then(|v| v.parse().ok()).unwrap_or(10000);
let port_end: u16 = std::env::var("PROXY_PORT_END")
.ok().and_then(|v| v.parse().ok()).unwrap_or(10099);
if proxy_host.is_empty() {
// No proxy configured — return a single direct client
info!("No proxy configured (PROXY_HOST not set), using direct connection");
if let Ok(client) = Client::builder()
.timeout(std::time::Duration::from_secs(60))
.build()
{
return vec![Arc::new(client)];
}
return vec![];
}
for port in port_start..=port_end {
let host = proxy_host.clone();
let user = proxy_username.clone();
let pass = proxy_password.clone();
let proto = proxy_protocol.clone();
handles.push(tokio::spawn(async move {
let proxy_url = if user.is_empty() {
format!("{}://{}:{}", proto, host, port)
} else {
format!("{}://{}:{}@{}:{}", proto, user, pass, host, port)
};
match reqwest::Proxy::all(&proxy_url) {
Ok(proxy) => {
match Client::builder()
.proxy(proxy)
.timeout(std::time::Duration::from_secs(60))
.build()
{
Ok(client) => Some(Arc::new(client)),
Err(_) => None,
}
}
Err(_) => None,
}
}));
}
let mut clients = Vec::new();
for handle in handles {
if let Ok(Some(client)) = handle.await {
clients.push(client);
}
}
info!("Built {} download clients with proxies", clients.len());
clients
}
/// Download all monthly klines for a symbol using proxy clients
pub async fn download_symbol_with_clients(
symbol: &str,
start_timestamp: i64,
clients: &[Arc<Client>],
candle_tx: mpsc::Sender<CandleData>,
) -> Result<u64, CollectorError> {
let symbol_upper = symbol.to_uppercase();
if clients.is_empty() {
return Err(CollectorError::RestApiError("No clients available".to_string()));
}
// Calculate start and end months
let start_date = Self::timestamp_to_date(start_timestamp);
let now = Utc::now().naive_utc().date();
// Generate list of months to download (exclude current month - incomplete)
let months = Self::generate_months(start_date, now);
if months.is_empty() {
info!("{}: No complete months to download", symbol_upper);
return Ok(0);
}
info!(
"{}: Downloading {} months ({}-{:02} to {}-{:02}) with {} clients",
symbol_upper,
months.len(),
months.first().unwrap().0, months.first().unwrap().1,
months.last().unwrap().0, months.last().unwrap().1,
clients.len()
);
// Download all months in parallel using all clients
let mut handles = Vec::new();
for (i, &(year, month)) in months.iter().enumerate() {
let client = clients[i % clients.len()].clone();
let sym = symbol_upper.clone();
let tx = candle_tx.clone();
handles.push(tokio::spawn(async move {
Self::download_month_with_client(&sym, year, month, &client, tx).await
}));
}
// Wait for all downloads
let results = join_all(handles).await;
let mut total_candles = 0u64;
for result in results {
match result {
Ok(Ok(count)) => total_candles += count,
Ok(Err(e)) => warn!("Download error: {}", e),
Err(e) => warn!("Join error: {}", e),
}
}
info!("{}: Downloaded {} candles from archive", symbol_upper, total_candles);
Ok(total_candles)
}
/// Download a single month's klines using a specific client
async fn download_month_with_client(
symbol: &str,
year: i32,
month: u32,
client: &Client,
candle_tx: mpsc::Sender<CandleData>,
) -> Result<u64, CollectorError> {
let url = format!(
"{}/{}/1m/{}-1m-{}-{:02}.zip",
BASE_URL, symbol, symbol, year, month
);
// Download ZIP file with proxy client
let response = client.get(&url)
.send()
.await
.map_err(|e| CollectorError::RestApiError(format!("Download failed: {}", e)))?;
if !response.status().is_success() {
return Err(CollectorError::RestApiError(
format!("HTTP {}: {}", response.status(), url)
));
}
let bytes = response.bytes()
.await
.map_err(|e| CollectorError::RestApiError(format!("Read bytes failed: {}", e)))?;
// Extract CSV from ZIP
let cursor = Cursor::new(bytes);
let mut archive = zip::ZipArchive::new(cursor)
.map_err(|e| CollectorError::RestApiError(format!("ZIP error: {}", e)))?;
let mut csv_content = String::new();
{
let mut file = archive.by_index(0)
.map_err(|e| CollectorError::RestApiError(format!("ZIP file error: {}", e)))?;
file.read_to_string(&mut csv_content)
.map_err(|e| CollectorError::RestApiError(format!("Read CSV error: {}", e)))?;
}
// Parse CSV and send candles
let count = Self::parse_csv(symbol, &csv_content, &candle_tx).await?;
info!("{} {}-{:02}: {} candles", symbol, year, month, count);
Ok(count)
}
/// Download all monthly klines (without proxy - original method)
pub async fn download_symbol(
symbol: &str,
start_timestamp: i64,
candle_tx: mpsc::Sender<CandleData>,
max_parallel: usize,
) -> Result<u64, CollectorError> {
let symbol_upper = symbol.to_uppercase();
let start_date = Self::timestamp_to_date(start_timestamp);
let now = Utc::now().naive_utc().date();
let months = Self::generate_months(start_date, now);
if months.is_empty() {
info!("{}: No complete months to download", symbol_upper);
return Ok(0);
}
info!(
"{}: Downloading {} months from {}-{:02} to {}-{:02}",
symbol_upper,
months.len(),
months.first().unwrap().0, months.first().unwrap().1,
months.last().unwrap().0, months.last().unwrap().1
);
let mut total_candles = 0u64;
for chunk in months.chunks(max_parallel) {
let mut handles = Vec::new();
for &(year, month) in chunk {
let sym = symbol_upper.clone();
let tx = candle_tx.clone();
handles.push(tokio::spawn(async move {
Self::download_month(&sym, year, month, tx).await
}));
}
let results = join_all(handles).await;
for result in results {
match result {
Ok(Ok(count)) => total_candles += count,
Ok(Err(e)) => warn!("Download error: {}", e),
Err(e) => warn!("Join error: {}", e),
}
}
}
info!("{}: Downloaded {} candles from archive", symbol_upper, total_candles);
Ok(total_candles)
}
/// Download a single month (without proxy)
async fn download_month(
symbol: &str,
year: i32,
month: u32,
candle_tx: mpsc::Sender<CandleData>,
) -> Result<u64, CollectorError> {
let url = format!(
"{}/{}/1m/{}-1m-{}-{:02}.zip",
BASE_URL, symbol, symbol, year, month
);
let response = reqwest::get(&url)
.await
.map_err(|e| CollectorError::RestApiError(format!("Download failed: {}", e)))?;
if !response.status().is_success() {
return Err(CollectorError::RestApiError(
format!("HTTP {}: {}", response.status(), url)
));
}
let bytes = response.bytes()
.await
.map_err(|e| CollectorError::RestApiError(format!("Read bytes failed: {}", e)))?;
let cursor = Cursor::new(bytes);
let mut archive = zip::ZipArchive::new(cursor)
.map_err(|e| CollectorError::RestApiError(format!("ZIP error: {}", e)))?;
let mut csv_content = String::new();
{
let mut file = archive.by_index(0)
.map_err(|e| CollectorError::RestApiError(format!("ZIP file error: {}", e)))?;
file.read_to_string(&mut csv_content)
.map_err(|e| CollectorError::RestApiError(format!("Read CSV error: {}", e)))?;
}
let count = Self::parse_csv(symbol, &csv_content, &candle_tx).await?;
info!("{} {}-{:02}: {} candles", symbol, year, month, count);
Ok(count)
}
/// Parse CSV content and send candles through channel
async fn parse_csv(
symbol: &str,
csv_content: &str,
candle_tx: &mpsc::Sender<CandleData>,
) -> Result<u64, CollectorError> {
let mut reader = csv::ReaderBuilder::new()
.has_headers(true)
.from_reader(csv_content.as_bytes());
let mut count = 0u64;
for result in reader.records() {
let record = match result {
Ok(r) => r,
Err(_) => continue,
};
if record.len() < 11 {
continue;
}
let timestamp: i64 = record[0].parse().unwrap_or(0);
let open: f64 = record[1].parse().unwrap_or(0.0);
let high: f64 = record[2].parse().unwrap_or(0.0);
let low: f64 = record[3].parse().unwrap_or(0.0);
let close: f64 = record[4].parse().unwrap_or(0.0);
let volume: f64 = record[5].parse().unwrap_or(0.0);
let taker_buy_volume: f64 = record[9].parse().unwrap_or(0.0);
let net_volume = taker_buy_volume * 2.0 - volume;
let candle = CandleData {
symbol: symbol.to_string(),
timestamp,
open,
high,
low,
close,
volume,
taker_buy_volume,
net_volume,
is_closed: true,
};
if candle_tx.send(candle).await.is_err() {
break;
}
count += 1;
}
Ok(count)
}
fn timestamp_to_date(ts: i64) -> NaiveDate {
let secs = ts / 1000;
chrono::DateTime::from_timestamp(secs, 0)
.unwrap_or_else(|| Utc::now())
.naive_utc()
.date()
}
fn generate_months(start: NaiveDate, end: NaiveDate) -> Vec<(i32, u32)> {
let mut months = Vec::new();
let mut year = start.year();
let mut month = start.month();
let end_year = if end.month() == 1 { end.year() - 1 } else { end.year() };
let end_month = if end.month() == 1 { 12 } else { end.month() - 1 };
loop {
if year > end_year || (year == end_year && month > end_month) {
break;
}
months.push((year, month));
month += 1;
if month > 12 {
month = 1;
year += 1;
}
}
months
}
pub fn last_complete_month_end() -> i64 {
let now = Utc::now().naive_utc();
let first_of_this_month = NaiveDate::from_ymd_opt(now.year(), now.month(), 1)
.unwrap()
.and_hms_opt(0, 0, 0)
.unwrap();
first_of_this_month.and_utc().timestamp_millis()
}
/// Download daily klines for a symbol (for current incomplete month)
pub async fn download_days_with_clients(
symbol: &str,
start_timestamp: i64,
clients: &[Arc<Client>],
candle_tx: mpsc::Sender<CandleData>,
) -> Result<u64, CollectorError> {
let symbol_upper = symbol.to_uppercase();
if clients.is_empty() {
return Err(CollectorError::RestApiError("No clients available".to_string()));
}
let start_date = Self::timestamp_to_date(start_timestamp);
let yesterday = Utc::now().naive_utc().date().pred_opt().unwrap_or(start_date);
let days = Self::generate_days(start_date, yesterday);
if days.is_empty() {
info!("{}: No days to download", symbol_upper);
return Ok(0);
}
info!(
"{}: Downloading {} days ({} to {}) with {} clients",
symbol_upper, days.len(),
days.first().unwrap(), days.last().unwrap(),
clients.len()
);
let mut handles = Vec::new();
for (i, date) in days.iter().enumerate() {
let client = clients[i % clients.len()].clone();
let sym = symbol_upper.clone();
let tx = candle_tx.clone();
let d = *date;
handles.push(tokio::spawn(async move {
Self::download_day_with_client(&sym, d, &client, tx).await
}));
}
let results = join_all(handles).await;
let mut total_candles = 0u64;
for result in results {
match result {
Ok(Ok(count)) => total_candles += count,
Ok(Err(e)) => warn!("Daily download error: {}", e),
Err(e) => warn!("Join error: {}", e),
}
}
info!("{}: Downloaded {} candles from daily archive", symbol_upper, total_candles);
Ok(total_candles)
}
/// Download a single day's klines
async fn download_day_with_client(
symbol: &str,
date: NaiveDate,
client: &Client,
candle_tx: mpsc::Sender<CandleData>,
) -> Result<u64, CollectorError> {
let url = format!(
"{}/{}/1m/{}-1m-{}.zip",
DAILY_BASE_URL, symbol, symbol, date.format("%Y-%m-%d")
);
let response = client.get(&url)
.send()
.await
.map_err(|e| CollectorError::RestApiError(format!("Download failed: {}", e)))?;
if !response.status().is_success() {
return Err(CollectorError::RestApiError(
format!("HTTP {}: {}", response.status(), url)
));
}
let bytes = response.bytes()
.await
.map_err(|e| CollectorError::RestApiError(format!("Read bytes failed: {}", e)))?;
let cursor = Cursor::new(bytes);
let mut archive = zip::ZipArchive::new(cursor)
.map_err(|e| CollectorError::RestApiError(format!("ZIP error: {}", e)))?;
let mut csv_content = String::new();
{
let mut file = archive.by_index(0)
.map_err(|e| CollectorError::RestApiError(format!("ZIP file error: {}", e)))?;
file.read_to_string(&mut csv_content)
.map_err(|e| CollectorError::RestApiError(format!("Read CSV error: {}", e)))?;
}
let count = Self::parse_csv(symbol, &csv_content, &candle_tx).await?;
info!("{} {}: {} candles", symbol, date, count);
Ok(count)
}
/// Generate list of days between start and end (inclusive)
fn generate_days(start: NaiveDate, end: NaiveDate) -> Vec<NaiveDate> {
let mut days = Vec::new();
let mut current = start;
while current <= end {
days.push(current);
current = current.succ_opt().unwrap_or(end);
if current == end && days.last() == Some(&end) {
break;
}
}
days
}
/// Get timestamp for start of current month (for daily download start point)
pub fn current_month_start() -> i64 {
let now = Utc::now().naive_utc();
let first_of_this_month = NaiveDate::from_ymd_opt(now.year(), now.month(), 1)
.unwrap()
.and_hms_opt(0, 0, 0)
.unwrap();
first_of_this_month.and_utc().timestamp_millis()
}
}
+103
View File
@@ -0,0 +1,103 @@
use crate::{
DatabaseHandler, SchedulerCommand,
CandleData, KlineQuery, ApiResponse, AddSymbolRequest, SchedulerStatus,
};
use axum::{
Router,
routing::{get, post, delete},
extract::{Path, Query, State},
response::Json,
http::StatusCode,
};
use std::sync::Arc;
use tokio::sync::{mpsc, oneshot};
pub struct KlineChartState {
pub db: Arc<DatabaseHandler>,
pub command_tx: mpsc::Sender<SchedulerCommand>,
}
pub fn klinechart_routes() -> Router<Arc<KlineChartState>> {
Router::new()
.route("/api/klines/{symbol}", get(get_klines))
.route("/api/symbols", get(get_symbols))
.route("/api/symbols", post(add_symbol))
.route("/api/symbols/{symbol}", delete(remove_symbol))
.route("/api/status", get(get_status))
}
// GET /api/klines/{symbol}?limit=800&interval=1m&end_time=...
async fn get_klines(
State(state): State<Arc<KlineChartState>>,
Path(symbol): Path<String>,
Query(query): Query<KlineQuery>,
) -> Json<ApiResponse<Vec<CandleData>>> {
let limit = query.limit.unwrap_or(800);
let interval = query.interval.unwrap_or_default();
let end_time = query.end_time;
match state.db.get_klines_aggregated(&symbol, interval, limit, end_time).await {
Ok(candles) => Json(ApiResponse::ok(candles)),
Err(e) => Json(ApiResponse::err(&e.to_string())),
}
}
// GET /api/symbols
async fn get_symbols(
State(state): State<Arc<KlineChartState>>,
) -> Json<ApiResponse<Vec<String>>> {
match state.db.get_active_symbols().await {
Ok(symbols) => Json(ApiResponse::ok(symbols)),
Err(e) => Json(ApiResponse::err(&e.to_string())),
}
}
// POST /api/symbols
async fn add_symbol(
State(state): State<Arc<KlineChartState>>,
Json(req): Json<AddSymbolRequest>,
) -> (StatusCode, Json<ApiResponse<String>>) {
let result = state.command_tx.send(SchedulerCommand::AddSymbol {
symbol: req.symbol.clone(),
backfill_from: req.backfill_from,
}).await;
match result {
Ok(_) => (StatusCode::OK, Json(ApiResponse::ok(format!("Adding symbol: {}", req.symbol)))),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::err(&e.to_string()))),
}
}
// DELETE /api/symbols/{symbol}
async fn remove_symbol(
State(state): State<Arc<KlineChartState>>,
Path(symbol): Path<String>,
) -> (StatusCode, Json<ApiResponse<String>>) {
let result = state.command_tx.send(SchedulerCommand::RemoveSymbol {
symbol: symbol.clone(),
}).await;
match result {
Ok(_) => (StatusCode::OK, Json(ApiResponse::ok(format!("Removing symbol: {}", symbol)))),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::err(&e.to_string()))),
}
}
// GET /api/status
async fn get_status(
State(state): State<Arc<KlineChartState>>,
) -> Json<ApiResponse<SchedulerStatus>> {
let (reply_tx, reply_rx) = oneshot::channel();
let send_result = state.command_tx.send(SchedulerCommand::GetStatus { reply: reply_tx }).await;
if send_result.is_err() {
return Json(ApiResponse::err("Failed to send status request"));
}
match reply_rx.await {
Ok(status) => Json(ApiResponse::ok(status)),
Err(_) => Json(ApiResponse::err("Failed to receive status")),
}
}
+23
View File
@@ -0,0 +1,23 @@
pub mod structs;
pub use structs::*;
pub mod error;
pub use error::*;
pub mod binance_collector;
pub use binance_collector::*;
pub mod database;
pub use database::*;
pub mod scheduler;
pub use scheduler::*;
pub mod klinechart;
pub use klinechart::*;
pub mod tradingview;
pub use tradingview::*;
pub mod historical_downloader;
pub use historical_downloader::*;
+83
View File
@@ -0,0 +1,83 @@
use backend::{
DatabaseHandler, Scheduler, create_command_channel,
KlineChartState, klinechart_routes,
TradingViewState, tradingview_routes,
CandleData,
};
use axum::Router;
use tower_http::cors::{CorsLayer, Any};
use axum::http::Method;
use std::sync::Arc;
use log::info;
use dotenv::*;
use tokio::sync::mpsc;
#[tokio::main]
async fn main() {
dotenv().ok();
env_logger::init();
info!("Current Version 1.4");
let database_url = std::env::var("DATABASE_URL")
.unwrap_or_else(|_| "postgres://quant:2Nr!Ya&oVvY5pp@172.18.0.2:5432/crypto_database".to_string());
info!("Connecting to database...");
let db = Arc::new(
DatabaseHandler::new(&database_url)
.await
.expect("Failed to connect to database")
);
let (command_tx, command_rx) = create_command_channel();
// Create TradingView state with broadcast channel
let (tradingview_state, _) = TradingViewState::new(db.clone());
let tradingview_state = Arc::new(tradingview_state);
// Create channel for WebSocket broadcasts
let (ws_broadcast_tx, mut ws_broadcast_rx) = mpsc::channel::<CandleData>(10000);
// Forward candles to TradingView broadcast
let tv_state_clone = tradingview_state.clone();
tokio::spawn(async move {
while let Some(candle) = ws_broadcast_rx.recv().await {
tv_state_clone.broadcast_candle(candle);
}
});
// Start scheduler in background with ws_broadcast_tx
let scheduler_db = db.clone();
tokio::spawn(async move {
let mut scheduler = Scheduler::new(scheduler_db, command_rx, Some(ws_broadcast_tx));
scheduler.run().await;
});
// KlineChart state (for klinechart frontend)
let klinechart_state = Arc::new(KlineChartState {
db: db.clone(),
command_tx,
});
// CORS for frontend
let cors = CorsLayer::new()
.allow_origin(Any)
.allow_methods([Method::GET, Method::POST, Method::DELETE, Method::OPTIONS])
.allow_headers(Any);
// Merge routes from both modules
let app = Router::new()
.merge(klinechart_routes().with_state(klinechart_state))
.merge(tradingview_routes().with_state(tradingview_state))
.layer(cors);
let addr = "0.0.0.0:3000";
info!("Starting API server on {}", addr);
info!("KlineChart API: /api/klines, /api/symbols, /api/status");
info!("TradingView UDF: /config, /symbols, /search, /history, /time");
let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
axum::serve(listener, app).await.unwrap();
}
+320
View File
@@ -0,0 +1,320 @@
use crate::database::DatabaseHandler;
use crate::binance_collector::BinanceCollector;
use crate::historical_downloader::HistoricalDownloader;
use crate::error::SchedulerError;
use crate::structs::*;
use crate::DATA_CUTOFF_TIMESTAMP;
use log::{info, error, warn};
use std::sync::Arc;
use tokio::sync::{mpsc, RwLock};
use tokio::task::JoinHandle;
const ONE_MINUTE_MS: i64 = 60_000;
pub struct Scheduler {
db: Arc<DatabaseHandler>,
command_rx: mpsc::Receiver<SchedulerCommand>,
collector_handle: Option<JoinHandle<()>>,
is_running: Arc<RwLock<bool>>,
active_symbols: Arc<RwLock<Vec<String>>>,
ws_broadcast_tx: Option<mpsc::Sender<CandleData>>,
}
impl Scheduler {
pub fn new(
db: Arc<DatabaseHandler>,
command_rx: mpsc::Receiver<SchedulerCommand>,
ws_broadcast_tx: Option<mpsc::Sender<CandleData>>,
) -> Self {
Self {
db,
command_rx,
collector_handle: None,
is_running: Arc::new(RwLock::new(false)),
active_symbols: Arc::new(RwLock::new(Vec::new())),
ws_broadcast_tx,
}
}
pub async fn run(&mut self) {
info!("Scheduler started");
// Initial startup
if let Err(e) = self.start_collector().await {
error!("Failed to start collector on init: {}", e);
}
// Command processing loop
while let Some(cmd) = self.command_rx.recv().await {
match cmd {
SchedulerCommand::AddSymbol { symbol, backfill_from: _ } => {
info!("AddSymbol command ignored - symbols managed by get_symbol()");
let _ = self.db.add_symbol(&symbol).await;
}
SchedulerCommand::RemoveSymbol { symbol } => {
info!("RemoveSymbol: {}", symbol);
let _ = self.db.remove_symbol(&symbol).await;
}
SchedulerCommand::RestartCollector => {
self.handle_restart_collector().await;
}
SchedulerCommand::GetStatus { reply } => {
self.handle_get_status(reply).await;
}
SchedulerCommand::Shutdown => {
info!("Shutdown command received");
self.stop_collector().await;
break;
}
}
}
info!("Scheduler stopped");
}
async fn handle_restart_collector(&mut self) {
info!("Restarting collector...");
self.stop_collector().await;
if let Err(e) = self.start_collector().await {
error!("Failed to restart collector: {}", e);
}
}
async fn handle_get_status(&self, reply: tokio::sync::oneshot::Sender<SchedulerStatus>) {
let is_running = *self.is_running.read().await;
let active_symbols = self.active_symbols.read().await.clone();
let collector_connected = self.collector_handle.is_some() && is_running;
let _ = reply.send(SchedulerStatus {
is_running,
active_symbols,
collector_connected,
});
}
async fn start_collector(&mut self) -> Result<(), SchedulerError> {
// STEP 1: Get symbols from TRACKED_SYMBOLS env variable
let symbol_names = DatabaseHandler::get_symbols_from_env();
if symbol_names.is_empty() {
warn!("No symbols in TRACKED_SYMBOLS env variable. Set TRACKED_SYMBOLS=BTCUSDT,ETHUSDT,...");
return Ok(());
}
info!("Tracking {} symbols from env: {:?}", symbol_names.len(), symbol_names);
// Update active symbols
{
let mut active = self.active_symbols.write().await;
*active = symbol_names.clone();
}
// STEP 2: Start WebSocket FIRST to capture real-time data immediately
info!("Starting WebSocket collector FIRST (priority: real-time data)...");
let collector = BinanceCollector::new(symbol_names.clone());
let (candle_tx, mut candle_rx) = mpsc::channel::<CandleData>(10000);
// Consumer that both saves to DB and broadcasts to WebSocket clients
let db = self.db.clone();
let ws_tx = self.ws_broadcast_tx.clone();
tokio::spawn(async move {
while let Some(candle) = candle_rx.recv().await {
// Broadcast to WebSocket clients immediately (all updates for real-time charts)
if let Some(ref tx) = ws_tx {
let _ = tx.send(candle.clone()).await;
}
// Only save closed candles to database
if candle.is_closed {
if let Err(e) = db.insert_candle(&candle).await {
error!("Failed to insert realtime candle: {}", e);
}
}
}
});
let is_running = self.is_running.clone();
let handle = tokio::spawn(async move {
{
let mut running = is_running.write().await;
*running = true;
}
if let Err(e) = collector.start_stream(candle_tx).await {
error!("Collector error: {}", e);
}
{
let mut running = is_running.write().await;
*running = false;
}
});
self.collector_handle = Some(handle);
info!("WebSocket collector started! Real-time data is now being captured.");
// STEP 3: Background sync - runs in parallel with WebSocket
// This fills in historical data without blocking real-time updates
let db_for_sync = self.db.clone();
let symbols_for_sync = symbol_names.clone();
tokio::spawn(async move {
info!("Starting background historical sync...");
// Build API clients for parallel sync
let api_clients = BinanceCollector::build_clients().await;
if api_clients.is_empty() {
error!("No working API clients available for background sync");
return;
}
info!("Built {} API clients for background sync", api_clients.len());
// Build HTTP clients for ZIP downloads
let http_clients = HistoricalDownloader::build_clients().await;
info!("Built {} HTTP clients for ZIP downloads", http_clients.len());
// Create channel for background sync (lower priority, uses batching)
let (sync_tx, sync_rx) = mpsc::channel::<CandleData>(100000);
let db_consumer = db_for_sync.clone();
let consumer_handle = tokio::spawn(async move {
db_consumer.start_consumer(sync_rx).await;
});
// Sync symbols that need updates
let mut synced_count = 0;
let total_symbols = symbols_for_sync.len();
for symbol in &symbols_for_sync {
let latest_ts = db_for_sync.get_latest_timestamp(symbol).await.ok().flatten();
let start_time = match latest_ts {
Some(ts) => ts + ONE_MINUTE_MS,
None => DATA_CUTOFF_TIMESTAMP,
};
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as i64;
// Skip if less than 2 minutes behind (WebSocket will catch up)
if now - start_time < ONE_MINUTE_MS * 2 {
synced_count += 1;
continue;
}
let behind_mins = (now - start_time) / ONE_MINUTE_MS;
info!("[{}/{}] Syncing {} ({} minutes behind)...",
synced_count + 1, total_symbols, symbol, behind_mins);
// Use comprehensive sync: Monthly ZIP → Daily ZIP → API
match BinanceCollector::sync_comprehensive(
symbol.clone(),
start_time,
api_clients.clone(),
&http_clients,
sync_tx.clone(),
).await {
Ok(count) => {
synced_count += 1;
if count > 0 {
info!("[{}/{}] {} synced: {} candles", synced_count, total_symbols, symbol, count);
}
}
Err(e) => {
synced_count += 1;
error!("{} sync failed: {}", symbol, e);
}
}
}
// Close sync channel
drop(sync_tx);
let _ = consumer_handle.await;
info!("Background historical sync complete! Synced {} symbols", synced_count);
// STEP 4: Gap detection and repair (also in background)
info!("Starting background gap detection and repair...");
let (gap_tx, gap_rx) = mpsc::channel::<CandleData>(100000);
let db_gap = db_for_sync.clone();
let gap_consumer_handle = tokio::spawn(async move {
db_gap.start_consumer(gap_rx).await;
});
// Rebuild clients for gap repair
let clients = BinanceCollector::build_clients().await;
let mut total_gaps_repaired = 0u64;
let mut symbols_with_gaps = 0;
for symbol in &symbols_for_sync {
match db_for_sync.find_gaps(symbol, DATA_CUTOFF_TIMESTAMP).await {
Ok(gaps) => {
if gaps.is_empty() {
continue;
}
symbols_with_gaps += 1;
info!("{} - Found {} gaps to repair", symbol, gaps.len());
for (gap_start, gap_end) in gaps {
let gap_duration_mins = (gap_end - gap_start) / ONE_MINUTE_MS;
// Skip very small gaps (less than 2 minutes)
if gap_duration_mins < 2 {
continue;
}
info!(
"{} - Repairing gap: {} minutes",
symbol, gap_duration_mins
);
match BinanceCollector::sync_from_scratch(
symbol.clone(),
gap_start,
clients.clone(),
gap_tx.clone(),
).await {
Ok(count) => {
total_gaps_repaired += count;
}
Err(e) => {
error!("{} - Gap repair failed: {}", symbol, e);
}
}
}
}
Err(e) => {
warn!("{} - Failed to check gaps: {}", symbol, e);
}
}
}
// Close gap repair channel
drop(gap_tx);
let _ = gap_consumer_handle.await;
info!("Gap detection complete! {} symbols had gaps, repaired {} candles total",
symbols_with_gaps, total_gaps_repaired);
});
Ok(())
}
async fn stop_collector(&mut self) {
if let Some(handle) = self.collector_handle.take() {
info!("Stopping collector...");
handle.abort();
let _ = handle.await;
let mut running = self.is_running.write().await;
*running = false;
}
}
}
pub fn create_command_channel() -> (mpsc::Sender<SchedulerCommand>, mpsc::Receiver<SchedulerCommand>) {
mpsc::channel(100)
}
+159
View File
@@ -0,0 +1,159 @@
use serde::{Serialize, Deserialize};
use tokio::sync::oneshot;
/// Data cutoff timestamp: 2024-01-01 00:00:00 UTC
/// Data before this timestamp will not be synced
pub const DATA_CUTOFF_TIMESTAMP: i64 = 1704067200000;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CandleData {
pub symbol: String,
pub timestamp: i64,
pub open: f64,
pub high: f64,
pub low: f64,
pub close: f64,
pub volume: f64,
pub taker_buy_volume: f64,
pub net_volume: f64,
pub is_closed: bool,
}
impl CandleData {
pub fn calculate_net_volume(total_volume: f64, taker_buy_volume: f64) -> f64 {
2.0 * taker_buy_volume - total_volume
}
}
/// Commands for controlling the Scheduler
pub enum SchedulerCommand {
/// Add a new symbol to track. Triggers backfill then collector restart.
AddSymbol {
symbol: String,
backfill_from: Option<i64>, // None = use EARLIEST_TIME
},
/// Remove a symbol from tracking. Triggers collector restart.
RemoveSymbol { symbol: String },
/// Restart the collector with current active symbols from database
RestartCollector,
/// Get current scheduler status
GetStatus { reply: oneshot::Sender<SchedulerStatus> },
/// Graceful shutdown
Shutdown,
}
/// Scheduler status for API responses
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchedulerStatus {
pub is_running: bool,
pub active_symbols: Vec<String>,
pub collector_connected: bool,
}
// API Request/Response structs
#[derive(Debug, Clone, Copy, Deserialize, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum Interval {
#[serde(rename = "1m")]
Min1,
#[serde(rename = "5m")]
Min5,
#[serde(rename = "15m")]
Min15,
#[serde(rename = "1h")]
Hour1,
#[serde(rename = "4h")]
Hour4,
#[serde(rename = "1d")]
Day1,
#[serde(rename = "1w")]
Week1,
#[serde(rename = "1M")]
Month1,
}
impl Default for Interval {
fn default() -> Self {
Interval::Min1
}
}
#[derive(Deserialize)]
pub struct KlineQuery {
pub limit: Option<i64>,
pub interval: Option<Interval>,
pub end_time: Option<i64>,
}
#[derive(Serialize)]
pub struct ApiResponse<T> {
pub success: bool,
pub data: Option<T>,
pub error: Option<String>,
}
impl<T> ApiResponse<T> {
pub fn ok(data: T) -> Self {
Self { success: true, data: Some(data), error: None }
}
pub fn err(msg: &str) -> Self {
Self { success: false, data: None, error: Some(msg.to_string()) }
}
}
#[derive(Deserialize)]
pub struct AddSymbolRequest {
pub symbol: String,
pub backfill_from: Option<i64>,
}
#[derive(Serialize,Deserialize,Debug,Clone)]
pub struct Symbol{
pub symbol: String,
pub start_timestamp: i64,
}
// WebSocket message types
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", content = "data")]
pub enum WsMessage {
/// Kline update for a symbol
#[serde(rename = "kline")]
Kline(CandleData),
/// Ticker update with price info for watchlist
#[serde(rename = "ticker")]
Ticker(TickerUpdate),
/// Subscribe to symbols
#[serde(rename = "subscribe")]
Subscribe { symbols: Vec<String> },
/// Unsubscribe from symbols
#[serde(rename = "unsubscribe")]
Unsubscribe { symbols: Vec<String> },
/// Ping/Pong for keepalive
#[serde(rename = "ping")]
Ping,
#[serde(rename = "pong")]
Pong,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TickerUpdate {
pub symbol: String,
pub price: f64,
pub change_24h: f64,
pub change_percent_24h: f64,
pub volume_24h: f64,
pub timestamp: i64,
}
+559
View File
@@ -0,0 +1,559 @@
use crate::{DatabaseHandler, Interval, CandleData, WsMessage};
use axum::{
Router,
routing::{get, post, delete},
extract::{Query, State, WebSocketUpgrade, ws::{Message, WebSocket}},
response::{Json, IntoResponse},
http::StatusCode,
};
use futures::{StreamExt, SinkExt};
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};
use std::path::PathBuf;
use tokio::sync::{broadcast, RwLock};
use tokio::fs;
use log::{info, warn, debug, error};
pub struct TradingViewState {
pub db: Arc<DatabaseHandler>,
pub candle_tx: broadcast::Sender<CandleData>,
}
impl TradingViewState {
pub fn new(db: Arc<DatabaseHandler>) -> (Self, broadcast::Receiver<CandleData>) {
let (candle_tx, candle_rx) = broadcast::channel(10000);
(Self { db, candle_tx }, candle_rx)
}
/// Send a candle update to all connected WebSocket clients
pub fn broadcast_candle(&self, candle: CandleData) {
// Ignore errors (no subscribers)
let _ = self.candle_tx.send(candle);
}
}
pub fn tradingview_routes() -> Router<Arc<TradingViewState>> {
Router::new()
.route("/config", get(get_config))
.route("/time", get(get_time))
.route("/symbols", get(get_symbol_info))
.route("/search", get(search_symbols))
.route("/tracked-symbols", get(get_tracked_symbols))
.route("/daily-opens", get(get_daily_opens))
.route("/history", get(get_history))
.route("/ws", get(ws_handler))
// Canvas API
.route("/canvas/list", get(canvas_list))
.route("/canvas/load", get(canvas_load))
.route("/canvas/save", post(canvas_save))
.route("/canvas/delete", delete(canvas_delete))
}
// ============ UDF Response Types ============
#[derive(Serialize)]
struct UdfConfig {
supported_resolutions: Vec<&'static str>,
supports_group_request: bool,
supports_marks: bool,
supports_search: bool,
supports_timescale_marks: bool,
}
#[derive(Serialize)]
struct UdfSymbolInfo {
symbol: String,
ticker: String,
name: String,
full_name: String,
description: String,
exchange: String,
listed_exchange: String,
#[serde(rename = "type")]
symbol_type: String,
currency_code: String,
session: String,
timezone: String,
minmovement: i32,
minmov: i32,
minmovement2: i32,
minmov2: i32,
pricescale: i64,
supported_resolutions: Vec<&'static str>,
has_intraday: bool,
has_daily: bool,
has_weekly_and_monthly: bool,
data_status: String,
}
#[derive(Serialize)]
struct UdfSearchResult {
symbol: String,
full_name: String,
description: String,
exchange: String,
ticker: String,
#[serde(rename = "type")]
symbol_type: String,
}
#[derive(Serialize)]
#[serde(untagged)]
enum UdfHistoryResponse {
Ok {
s: String, // "ok"
t: Vec<i64>, // timestamps (seconds)
o: Vec<f64>, // open
h: Vec<f64>, // high
l: Vec<f64>, // low
c: Vec<f64>, // close
v: Vec<f64>, // volume
nv: Vec<f64>, // net volume (custom)
tbv: Vec<f64>, // taker buy volume (custom)
},
NoData {
s: String, // "no_data"
#[serde(skip_serializing_if = "Option::is_none")]
#[serde(rename = "nextTime")]
next_time: Option<i64>,
},
Error {
s: String, // "error"
errmsg: String,
},
}
// ============ Query Parameters ============
#[derive(Deserialize)]
struct SymbolQuery {
symbol: String,
}
#[allow(dead_code)]
#[derive(Deserialize)]
struct SearchQuery {
query: Option<String>,
#[serde(rename = "type")]
symbol_type: Option<String>,
exchange: Option<String>,
limit: Option<i32>,
}
#[derive(Deserialize)]
struct HistoryQuery {
symbol: String,
resolution: String,
from: i64, // unix timestamp (seconds)
to: i64, // unix timestamp (seconds)
countback: Option<i64>,
}
// ============ Handlers ============
// GET /config
async fn get_config() -> Json<UdfConfig> {
Json(UdfConfig {
supported_resolutions: vec!["1", "5", "15", "60", "240", "1D", "1W", "1M"],
supports_group_request: false,
supports_marks: false,
supports_search: true,
supports_timescale_marks: false,
})
}
// GET /time
async fn get_time() -> String {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs();
now.to_string()
}
// GET /symbols?symbol=BTCUSDT
async fn get_symbol_info(
Query(query): Query<SymbolQuery>,
) -> Json<UdfSymbolInfo> {
let symbol = query.symbol.to_uppercase();
Json(UdfSymbolInfo {
symbol: symbol.clone(),
ticker: symbol.clone(),
name: symbol.clone(),
full_name: format!("BINANCE:{}", symbol),
description: symbol.clone(),
exchange: "BINANCE".to_string(),
listed_exchange: "BINANCE".to_string(),
symbol_type: "crypto".to_string(),
currency_code: "USDT".to_string(),
session: "24x7".to_string(),
timezone: "Etc/UTC".to_string(),
minmovement: 1,
minmov: 1,
minmovement2: 0,
minmov2: 0,
pricescale: 100000000, // 8 decimal places for crypto
supported_resolutions: vec!["1", "5", "15", "60", "240", "1D", "1W", "1M"],
has_intraday: true,
has_daily: true,
has_weekly_and_monthly: true,
data_status: "streaming".to_string(),
})
}
// GET /tracked-symbols - 返回 TRACKED_SYMBOL 环境变量中配置的 symbols
async fn get_tracked_symbols() -> Json<Vec<String>> {
Json(crate::DatabaseHandler::get_symbols_from_env())
}
// GET /daily-opens - 返回所有 symbol 当天 UTC 00:00 的开盘价
async fn get_daily_opens(
State(state): State<Arc<TradingViewState>>,
) -> Json<std::collections::HashMap<String, f64>> {
match state.db.get_daily_opens().await {
Ok(opens) => Json(opens),
Err(_) => Json(std::collections::HashMap::new()),
}
}
// GET /search?query=BTC&limit=10
async fn search_symbols(
State(state): State<Arc<TradingViewState>>,
Query(query): Query<SearchQuery>,
) -> Json<Vec<UdfSearchResult>> {
let search_term = query.query.unwrap_or_default().to_uppercase();
let limit = query.limit.unwrap_or(30) as usize;
let symbols = match state.db.get_active_symbols().await {
Ok(s) => s,
Err(_) => return Json(vec![]),
};
let results: Vec<UdfSearchResult> = symbols
.into_iter()
.filter(|s| search_term.is_empty() || s.to_uppercase().contains(&search_term))
.take(limit)
.map(|s| {
let upper = s.to_uppercase();
UdfSearchResult {
symbol: upper.clone(),
full_name: format!("BINANCE:{}", upper),
description: upper.clone(),
exchange: "BINANCE".to_string(),
ticker: upper,
symbol_type: "crypto".to_string(),
}
})
.collect();
Json(results)
}
// GET /history?symbol=BTCUSDT&resolution=1&from=...&to=...
async fn get_history(
State(state): State<Arc<TradingViewState>>,
Query(query): Query<HistoryQuery>,
) -> Json<UdfHistoryResponse> {
let symbol = query.symbol.to_uppercase();
// Convert resolution string to Interval enum
let interval = match query.resolution.as_str() {
"1" => Interval::Min1,
"5" => Interval::Min5,
"15" => Interval::Min15,
"60" => Interval::Hour1,
"240" => Interval::Hour4,
"D" | "1D" => Interval::Day1,
"W" | "1W" => Interval::Week1,
"M" | "1M" => Interval::Month1,
_ => Interval::Min1,
};
// Convert from/to (seconds) to milliseconds for database query
let from_ms = query.from * 1000;
let to_ms = query.to * 1000;
// Calculate limit based on countback or time range
let limit = query.countback.unwrap_or(1000);
match state.db.get_klines_aggregated(&symbol, interval, limit, Some(to_ms)).await {
Ok(candles) => {
// Filter by from_ms and convert to UDF format
let filtered: Vec<_> = candles
.into_iter()
.filter(|c| c.timestamp >= from_ms && c.timestamp <= to_ms)
.collect();
if filtered.is_empty() {
return Json(UdfHistoryResponse::NoData {
s: "no_data".to_string(),
next_time: None,
});
}
let t: Vec<i64> = filtered.iter().map(|c| c.timestamp / 1000).collect();
let o: Vec<f64> = filtered.iter().map(|c| c.open).collect();
let h: Vec<f64> = filtered.iter().map(|c| c.high).collect();
let l: Vec<f64> = filtered.iter().map(|c| c.low).collect();
let c: Vec<f64> = filtered.iter().map(|c| c.close).collect();
let v: Vec<f64> = filtered.iter().map(|c| c.volume).collect();
let nv: Vec<f64> = filtered.iter().map(|c| c.net_volume).collect();
let tbv: Vec<f64> = filtered.iter().map(|c| c.taker_buy_volume).collect();
Json(UdfHistoryResponse::Ok {
s: "ok".to_string(),
t, o, h, l, c, v, nv, tbv,
})
}
Err(e) => Json(UdfHistoryResponse::Error {
s: "error".to_string(),
errmsg: e.to_string(),
}),
}
}
// ============ WebSocket Handler ============
async fn ws_handler(
ws: WebSocketUpgrade,
State(state): State<Arc<TradingViewState>>,
) -> impl IntoResponse {
ws.on_upgrade(move |socket| handle_ws_connection(socket, state))
}
async fn handle_ws_connection(socket: WebSocket, state: Arc<TradingViewState>) {
let (mut sender, mut receiver) = socket.split();
// Subscribe to broadcast channel
let mut candle_rx = state.candle_tx.subscribe();
// Subscribed symbols for this client
let subscribed_symbols: Arc<RwLock<HashSet<String>>> = Arc::new(RwLock::new(HashSet::new()));
let subscribed_symbols_clone = subscribed_symbols.clone();
info!("WebSocket client connected");
// Task to receive messages from client
let recv_task = tokio::spawn(async move {
while let Some(msg) = receiver.next().await {
match msg {
Ok(Message::Text(text)) => {
// Parse incoming message
if let Ok(ws_msg) = serde_json::from_str::<WsMessage>(&text) {
match ws_msg {
WsMessage::Subscribe { symbols } => {
let mut subs = subscribed_symbols_clone.write().await;
for s in symbols {
subs.insert(s.to_uppercase());
}
debug!("Client subscribed to {} symbols", subs.len());
}
WsMessage::Unsubscribe { symbols } => {
let mut subs = subscribed_symbols_clone.write().await;
for s in symbols {
subs.remove(&s.to_uppercase());
}
}
WsMessage::Ping => {
// Pong is handled by the send task
}
_ => {}
}
}
}
Ok(Message::Close(_)) => {
info!("WebSocket client disconnected");
break;
}
Err(e) => {
warn!("WebSocket receive error: {}", e);
break;
}
_ => {}
}
}
});
// Task to send messages to client
let send_task = tokio::spawn(async move {
loop {
tokio::select! {
// Forward candle updates to client
result = candle_rx.recv() => {
match result {
Ok(candle) => {
let subs = subscribed_symbols.read().await;
// Send to client if subscribed or if subscribed to all (empty set means all)
if subs.is_empty() || subs.contains(&candle.symbol.to_uppercase()) {
let msg = WsMessage::Kline(candle);
if let Ok(json) = serde_json::to_string(&msg) {
if sender.send(Message::Text(json.into())).await.is_err() {
break;
}
}
}
}
Err(broadcast::error::RecvError::Lagged(n)) => {
warn!("WebSocket client lagged {} messages", n);
}
Err(broadcast::error::RecvError::Closed) => {
break;
}
}
}
}
}
});
// Wait for either task to finish
tokio::select! {
_ = recv_task => {}
_ = send_task => {}
}
info!("WebSocket connection closed");
}
// ============ Canvas API ============
const STORAGE_DIR: &str = "storage";
const DEFAULT_USER: &str = "default";
#[derive(Deserialize)]
struct CanvasListQuery {
symbol: String,
}
#[derive(Deserialize)]
struct CanvasLoadQuery {
symbol: String,
name: String,
}
#[derive(Deserialize)]
struct CanvasSaveBody {
symbol: String,
name: String,
data: serde_json::Value,
}
#[derive(Deserialize)]
struct CanvasDeleteQuery {
symbol: String,
name: String,
}
#[derive(Serialize)]
struct CanvasListResponse {
canvases: Vec<String>,
}
fn get_user_id(headers: &axum::http::HeaderMap) -> String {
headers.get("X-User-Id")
.and_then(|v| v.to_str().ok())
.filter(|s| !s.is_empty())
.unwrap_or(DEFAULT_USER)
.to_string()
}
fn get_canvas_dir(user_id: &str, symbol: &str) -> PathBuf {
PathBuf::from(STORAGE_DIR)
.join(user_id)
.join(symbol.to_uppercase())
}
fn get_canvas_path(user_id: &str, symbol: &str, name: &str) -> PathBuf {
get_canvas_dir(user_id, symbol).join(format!("{}.json", name))
}
// GET /canvas/list?symbol=BTCUSDT
async fn canvas_list(headers: axum::http::HeaderMap, Query(query): Query<CanvasListQuery>) -> Json<CanvasListResponse> {
let user_id = get_user_id(&headers);
let dir = get_canvas_dir(&user_id, &query.symbol);
let mut canvases = Vec::new();
if let Ok(mut entries) = fs::read_dir(&dir).await {
while let Ok(Some(entry)) = entries.next_entry().await {
if let Some(name) = entry.file_name().to_str() {
if name.ends_with(".json") {
canvases.push(name.trim_end_matches(".json").to_string());
}
}
}
}
canvases.sort();
Json(CanvasListResponse { canvases })
}
// GET /canvas/load?symbol=BTCUSDT&name=default
async fn canvas_load(
headers: axum::http::HeaderMap,
Query(query): Query<CanvasLoadQuery>,
) -> Result<Json<serde_json::Value>, StatusCode> {
let user_id = get_user_id(&headers);
let path = get_canvas_path(&user_id, &query.symbol, &query.name);
match fs::read_to_string(&path).await {
Ok(content) => {
match serde_json::from_str(&content) {
Ok(data) => Ok(Json(data)),
Err(e) => {
error!("Failed to parse canvas {}: {}", path.display(), e);
Err(StatusCode::INTERNAL_SERVER_ERROR)
}
}
}
Err(_) => Err(StatusCode::NOT_FOUND),
}
}
// POST /canvas/save
async fn canvas_save(
headers: axum::http::HeaderMap,
Json(body): Json<CanvasSaveBody>,
) -> Result<StatusCode, StatusCode> {
let user_id = get_user_id(&headers);
let dir = get_canvas_dir(&user_id, &body.symbol);
let path = get_canvas_path(&user_id, &body.symbol, &body.name);
// Create directory if not exists
if let Err(e) = fs::create_dir_all(&dir).await {
error!("Failed to create dir {}: {}", dir.display(), e);
return Err(StatusCode::INTERNAL_SERVER_ERROR);
}
// Write canvas data
let content = serde_json::to_string_pretty(&body.data)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
fs::write(&path, content).await
.map_err(|e| {
error!("Failed to write canvas {}: {}", path.display(), e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
info!("Saved canvas: {}", path.display());
Ok(StatusCode::OK)
}
// DELETE /canvas/delete?symbol=BTCUSDT&name=default
async fn canvas_delete(
headers: axum::http::HeaderMap,
Query(query): Query<CanvasDeleteQuery>,
) -> StatusCode {
let user_id = get_user_id(&headers);
let path = get_canvas_path(&user_id, &query.symbol, &query.name);
match fs::remove_file(&path).await {
Ok(_) => {
info!("Deleted canvas: {}", path.display());
StatusCode::OK
}
Err(_) => StatusCode::NOT_FOUND,
}
}
+86
View File
@@ -0,0 +1,86 @@
use backend::{BinanceCollector, CandleData};
use tokio::sync::mpsc;
use tokio::time::{timeout, Duration};
use dotenv::dotenv;
use log::{info,error,debug};
#[tokio::test]
async fn test_backfill_and_stream() {
dotenv().ok();
env_logger::init();
let collector = BinanceCollector::new(vec!["btcusdt".to_string(), "ethusdt".to_string()]);
let (tx, mut rx) = mpsc::channel::<CandleData>(1000);
// Calculate time range: last 10 minutes
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as i64;
let start_time = now - (10 * 60 * 1000); // 10 minutes ago
// First, backfill recent data for btcusdt
info!("Starting backfill...");
let backfill_count = collector.backfill("btcusdt", start_time, now, &tx).await.unwrap();
info!("Backfill completed: {} candles", backfill_count);
// Then start WebSocket stream
let collector_handle = tokio::spawn(async move {
collector.start_stream(tx).await
});
let mut received_count = 0;
// Drain backfill data first
while let Ok(Some(candle)) = timeout(Duration::from_millis(100), rx.recv()).await {
info!(
"[BACKFILL] {} ts={} c={:.2} net_vol={:.4}",
candle.symbol, candle.timestamp, candle.close, candle.net_volume
);
received_count += 1;
}
info!("Received {} backfill candles from channel", received_count);
// Wait for WebSocket data (up to 90 seconds for at least 1 closed candle)
let mut ws_count = 0;
let ws_result = timeout(Duration::from_secs(90), async {
while let Some(candle) = rx.recv().await {
info!(
"[WEBSOCKET] {} ts={} c={:.2} net_vol={:.4}",
candle.symbol, candle.timestamp, candle.close, candle.net_volume
);
ws_count += 1;
if ws_count >= 1 {
return true;
}
}
false
}).await;
collector_handle.abort();
info!("\nFinal: Backfill={}, WebSocket={}", backfill_count, ws_count);
assert!(backfill_count >= 5, "Expected at least 5 backfill candles, got {}", backfill_count);
match ws_result {
Ok(true) => info!("Test completed successfully!"),
Ok(false) => info!("WebSocket closed before receiving data"),
Err(_) => info!("WebSocket timed out (normal if test runs mid-minute)"),
}
}
#[tokio::test]
async fn test_get_symbol() {
dotenv().ok();
env_logger::init();
let collector = BinanceCollector::new(vec!["btcusdt".to_string(), "ethusdt".to_string()]);
let result = BinanceCollector::get_symbol().await.unwrap();
info!("Result = {:?}",result);
}
+128
View File
@@ -0,0 +1,128 @@
use backend::{DatabaseHandler, CandleData};
use tokio::sync::mpsc;
use dotenv::dotenv;
use log::info;
use std::env;
fn get_database_url() -> String {
dotenv().ok();
env::var("DATABASE_URL").unwrap_or_else(|_| {
"postgres://quant:2Nr!Ya&oVvY5pp@172.18.0.10:5432/crypto_database".to_string()
})
}
#[tokio::test]
async fn test_database_connection() {
dotenv().ok();
env_logger::init();
let db_url = get_database_url();
info!("Connecting to database...");
let db = DatabaseHandler::new(&db_url).await;
assert!(db.is_ok(), "Failed to connect to database: {:?}", db.err());
info!("Database connection successful!");
}
#[tokio::test]
async fn test_get_active_symbols() {
dotenv().ok();
env_logger::init();
let db_url = get_database_url();
let db = DatabaseHandler::new(&db_url).await.expect("Failed to connect");
let symbols = db.get_active_symbols().await;
info!("Active symbols: {:?}", symbols);
assert!(symbols.is_ok(), "Failed to get active symbols: {:?}", symbols.err());
}
#[tokio::test]
async fn test_insert_and_query() {
dotenv().ok();
env_logger::init();
let db_url = get_database_url();
let db = DatabaseHandler::new(&db_url).await.expect("Failed to connect");
// Create test candle
let test_candle = CandleData {
symbol: "TESTUSDT".to_string(),
timestamp: 1700000000000, // Fixed test timestamp
open: 100.0,
high: 105.0,
low: 99.0,
close: 102.0,
volume: 1000.0,
taker_buy_volume: 600.0,
net_volume: 200.0,
is_closed: true,
};
// Insert
let result = db.insert_candle(&test_candle).await;
assert!(result.is_ok(), "Failed to insert candle: {:?}", result.err());
info!("Inserted test candle");
// Query latest timestamp
let latest = db.get_latest_timestamp("TESTUSDT").await;
assert!(latest.is_ok(), "Failed to get latest timestamp: {:?}", latest.err());
let ts = latest.unwrap();
assert!(ts.is_some(), "No timestamp found for TESTUSDT");
assert_eq!(ts.unwrap(), 1700000000000, "Timestamp mismatch");
info!("Latest timestamp for TESTUSDT: {:?}", ts);
}
#[tokio::test]
async fn test_batch_consumer() {
dotenv().ok();
env_logger::init();
let db_url = get_database_url();
let db = DatabaseHandler::new(&db_url).await.expect("Failed to connect");
let (tx, rx) = mpsc::channel::<CandleData>(100);
// Spawn consumer
let db_handle = tokio::spawn(async move {
db.start_consumer(rx).await;
});
// Send test candles
let base_ts = 1700000100000i64;
for i in 0..10 {
let candle = CandleData {
symbol: "BATCHTEST".to_string(),
timestamp: base_ts + (i * 60000),
open: 100.0 + i as f64,
high: 105.0 + i as f64,
low: 99.0 + i as f64,
close: 102.0 + i as f64,
volume: 1000.0,
taker_buy_volume: 600.0,
net_volume: 200.0,
is_closed: true,
};
tx.send(candle).await.unwrap();
}
// Close channel to trigger flush
drop(tx);
// Wait for consumer to finish
let _ = tokio::time::timeout(
tokio::time::Duration::from_secs(10),
db_handle
).await;
info!("Batch insert test completed");
}
+322
View File
@@ -0,0 +1,322 @@
use backend::{
Scheduler, DatabaseHandler, SchedulerCommand,
create_command_channel,
};
use std::sync::Arc;
use tokio::sync::oneshot;
use tokio::time::{timeout, Duration};
use dotenv::dotenv;
use log::info;
use std::env;
fn get_database_url() -> String {
dotenv().ok();
env::var("DATABASE_URL").unwrap_or_else(|_| {
"postgres://quant:2Nr!Ya&oVvY5pp@172.18.0.2:5432/crypto_database".to_string()
})
}
#[tokio::test]
async fn test_scheduler_startup_and_shutdown() {
dotenv().ok();
env_logger::try_init().ok();
let db_url = get_database_url();
let db = Arc::new(DatabaseHandler::new(&db_url).await.expect("Failed to connect to database"));
let (command_tx, command_rx) = create_command_channel();
let mut scheduler = Scheduler::new(db.clone(), command_rx, None);
// Start scheduler in background
let scheduler_handle = tokio::spawn(async move {
scheduler.run().await;
});
// Give it time to start
tokio::time::sleep(Duration::from_secs(2)).await;
// Send shutdown command
command_tx.send(SchedulerCommand::Shutdown).await.expect("Failed to send shutdown");
// Wait for scheduler to stop
let result = timeout(Duration::from_secs(10), scheduler_handle).await;
assert!(result.is_ok(), "Scheduler did not shutdown in time");
info!("Scheduler startup and shutdown test passed");
}
#[tokio::test]
async fn test_get_status() {
dotenv().ok();
env_logger::try_init().ok();
let db_url = get_database_url();
let db = Arc::new(DatabaseHandler::new(&db_url).await.expect("Failed to connect to database"));
let (command_tx, command_rx) = create_command_channel();
let mut scheduler = Scheduler::new(db.clone(), command_rx, None);
// Start scheduler
let scheduler_handle = tokio::spawn(async move {
scheduler.run().await;
});
// Give it time to start
tokio::time::sleep(Duration::from_secs(2)).await;
// Request status
let (reply_tx, reply_rx) = oneshot::channel();
command_tx.send(SchedulerCommand::GetStatus { reply: reply_tx }).await.expect("Failed to send GetStatus");
let status = timeout(Duration::from_secs(5), reply_rx).await
.expect("Timeout waiting for status")
.expect("Failed to receive status");
info!("Scheduler status: {:?}", status);
assert!(status.is_running || status.active_symbols.is_empty(), "Scheduler should be running or have no symbols");
// Shutdown
command_tx.send(SchedulerCommand::Shutdown).await.expect("Failed to send shutdown");
let _ = timeout(Duration::from_secs(10), scheduler_handle).await;
info!("GetStatus test passed");
}
#[tokio::test]
async fn test_add_and_remove_symbol() {
dotenv().ok();
env_logger::try_init().ok();
let db_url = get_database_url();
let db = Arc::new(DatabaseHandler::new(&db_url).await.expect("Failed to connect to database"));
let test_symbol = "testscheduler";
// Clean up first - remove test symbol if exists
let _ = db.remove_symbol(test_symbol).await;
let (command_tx, command_rx) = create_command_channel();
let mut scheduler = Scheduler::new(db.clone(), command_rx, None);
// Start scheduler
let scheduler_handle = tokio::spawn(async move {
scheduler.run().await;
});
// Give it time to start
tokio::time::sleep(Duration::from_secs(2)).await;
// Add symbol (with recent backfill_from to avoid long backfill)
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as i64;
let recent_time = now - (5 * 60 * 1000); // 5 minutes ago
info!("Adding test symbol: {}", test_symbol);
command_tx.send(SchedulerCommand::AddSymbol {
symbol: test_symbol.to_string(),
backfill_from: Some(recent_time),
}).await.expect("Failed to send AddSymbol");
// Wait for processing
tokio::time::sleep(Duration::from_secs(5)).await;
// Check status
let (reply_tx, reply_rx) = oneshot::channel();
command_tx.send(SchedulerCommand::GetStatus { reply: reply_tx }).await.expect("Failed to send GetStatus");
let status = timeout(Duration::from_secs(5), reply_rx).await
.expect("Timeout")
.expect("Failed to receive");
info!("Status after add: {:?}", status);
// Verify symbol was added
let is_tracked = db.is_symbol_tracked(test_symbol).await.expect("Failed to check tracking");
assert!(is_tracked, "Symbol should be tracked after AddSymbol");
// Remove symbol
info!("Removing test symbol: {}", test_symbol);
command_tx.send(SchedulerCommand::RemoveSymbol {
symbol: test_symbol.to_string(),
}).await.expect("Failed to send RemoveSymbol");
// Wait for processing
tokio::time::sleep(Duration::from_secs(3)).await;
// Verify symbol was removed
let is_tracked = db.is_symbol_tracked(test_symbol).await.expect("Failed to check tracking");
assert!(!is_tracked, "Symbol should not be tracked after RemoveSymbol");
// Shutdown
command_tx.send(SchedulerCommand::Shutdown).await.expect("Failed to send shutdown");
let _ = timeout(Duration::from_secs(10), scheduler_handle).await;
info!("Add and remove symbol test passed");
}
#[tokio::test]
async fn test_restart_collector() {
dotenv().ok();
env_logger::try_init().ok();
let db_url = get_database_url();
let db = Arc::new(DatabaseHandler::new(&db_url).await.expect("Failed to connect to database"));
let (command_tx, command_rx) = create_command_channel();
let mut scheduler = Scheduler::new(db.clone(), command_rx, None);
// Start scheduler
let scheduler_handle = tokio::spawn(async move {
scheduler.run().await;
});
// Give it time to start
tokio::time::sleep(Duration::from_secs(2)).await;
// Get initial status
let (reply_tx, reply_rx) = oneshot::channel();
command_tx.send(SchedulerCommand::GetStatus { reply: reply_tx }).await.unwrap();
let status_before = reply_rx.await.unwrap();
info!("Status before restart: {:?}", status_before);
// Send restart command
info!("Sending RestartCollector command");
command_tx.send(SchedulerCommand::RestartCollector).await.expect("Failed to send RestartCollector");
// Wait for restart
tokio::time::sleep(Duration::from_secs(5)).await;
// Get status after restart
let (reply_tx, reply_rx) = oneshot::channel();
command_tx.send(SchedulerCommand::GetStatus { reply: reply_tx }).await.unwrap();
let status_after = reply_rx.await.unwrap();
info!("Status after restart: {:?}", status_after);
// Shutdown
command_tx.send(SchedulerCommand::Shutdown).await.expect("Failed to send shutdown");
let _ = timeout(Duration::from_secs(10), scheduler_handle).await;
info!("Restart collector test passed");
}
#[tokio::test]
async fn test_multiple_commands() {
dotenv().ok();
env_logger::try_init().ok();
let db_url = get_database_url();
let db = Arc::new(DatabaseHandler::new(&db_url).await.expect("Failed to connect to database"));
let (command_tx, command_rx) = create_command_channel();
let mut scheduler = Scheduler::new(db.clone(), command_rx, None);
// Start scheduler
let scheduler_handle = tokio::spawn(async move {
scheduler.run().await;
});
tokio::time::sleep(Duration::from_secs(2)).await;
// Send multiple status requests rapidly
for i in 0..5 {
let (reply_tx, reply_rx) = oneshot::channel();
command_tx.send(SchedulerCommand::GetStatus { reply: reply_tx }).await.unwrap();
let status = reply_rx.await.unwrap();
info!("Status request {}: {:?}", i, status);
}
// Shutdown
command_tx.send(SchedulerCommand::Shutdown).await.expect("Failed to send shutdown");
let _ = timeout(Duration::from_secs(10), scheduler_handle).await;
info!("Multiple commands test passed");
}
#[tokio::test]
async fn test_real_btcusdt_collection() {
dotenv().ok();
env_logger::try_init().ok();
let db_url = get_database_url();
let db = Arc::new(DatabaseHandler::new(&db_url).await.expect("Failed to connect to database"));
let symbol = "btcusdt";
// Clean up - remove from tracking first
let _ = db.remove_symbol(symbol).await;
let (command_tx, command_rx) = create_command_channel();
let mut scheduler = Scheduler::new(db.clone(), command_rx, None);
// Start scheduler
let scheduler_handle = tokio::spawn(async move {
scheduler.run().await;
});
tokio::time::sleep(Duration::from_secs(2)).await;
// Calculate start time: 5 minutes ago
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as i64;
let five_min_ago = now - (5 * 60 * 1000);
info!("Adding BTCUSDT, backfill from {} (5 min ago)", five_min_ago);
// Add BTCUSDT with backfill from 5 minutes ago
command_tx.send(SchedulerCommand::AddSymbol {
symbol: symbol.to_string(),
backfill_from: Some(five_min_ago),
}).await.expect("Failed to send AddSymbol");
// Wait for backfill to complete
info!("Waiting for backfill...");
tokio::time::sleep(Duration::from_secs(10)).await;
// Check latest timestamp in database
let latest_ts = db.get_latest_timestamp(symbol).await.expect("Failed to get timestamp");
info!("Latest BTCUSDT timestamp in DB: {:?}", latest_ts);
assert!(latest_ts.is_some(), "Should have BTCUSDT data in database");
// Wait for 2 more candles (about 2 minutes + buffer)
info!("Waiting for 2 live candles (~2.5 minutes)...");
tokio::time::sleep(Duration::from_secs(150)).await;
// Check new latest timestamp
let new_latest_ts = db.get_latest_timestamp(symbol).await.expect("Failed to get timestamp");
info!("New latest BTCUSDT timestamp: {:?}", new_latest_ts);
assert!(new_latest_ts.unwrap() > latest_ts.unwrap(), "Should have received new candles");
// Get status
let (reply_tx, reply_rx) = oneshot::channel();
command_tx.send(SchedulerCommand::GetStatus { reply: reply_tx }).await.unwrap();
let status = reply_rx.await.unwrap();
info!("Final status: {:?}", status);
assert!(status.active_symbols.contains(&symbol.to_string()), "BTCUSDT should be in active symbols");
// Clean up - remove symbol
command_tx.send(SchedulerCommand::RemoveSymbol {
symbol: symbol.to_string(),
}).await.expect("Failed to remove symbol");
tokio::time::sleep(Duration::from_secs(2)).await;
// Shutdown
command_tx.send(SchedulerCommand::Shutdown).await.expect("Failed to send shutdown");
let _ = timeout(Duration::from_secs(10), scheduler_handle).await;
info!("Real BTCUSDT collection test passed!");
info!("Check database: SELECT * FROM klines_1m WHERE symbol = 'BTCUSDT' ORDER BY timestamp DESC LIMIT 10;");
}
+65
View File
@@ -0,0 +1,65 @@
use backend::{BinanceCollector, CandleData};
use tokio::sync::mpsc;
use dotenv::dotenv;
use log::info;
#[tokio::test]
async fn test_sync_full_history_single_symbol() {
dotenv().ok();
env_logger::init();
// 1. Get all symbols
let symbol_vec = BinanceCollector::get_symbol().await.unwrap();
info!("Got {} symbols", symbol_vec.len());
// 2. Find ETHUSDT start time
let test_symbol = "ETHUSDT";
let test_start_time = symbol_vec
.iter()
.find(|item| item.symbol == test_symbol)
.map(|item| item.start_timestamp)
.expect("ETHUSDT not found");
info!("{} start_timestamp: {}", test_symbol, test_start_time);
// 3. Build clients (all clients work together on this symbol)
let clients = BinanceCollector::build_clients().await;
info!("Built {} clients", clients.len());
assert!(!clients.is_empty(), "Need at least 1 client");
// 4. Create channel
let (tx, mut rx) = mpsc::channel::<CandleData>(100000);
// 5. Sync from scratch (all clients work in parallel on different time segments)
info!("Starting full history sync for {} with {} clients...", test_symbol, clients.len());
let sync_handle = tokio::spawn(async move {
BinanceCollector::sync_from_scratch(
test_symbol.to_string(),
test_start_time,
clients, // Pass all clients
tx,
).await
});
// 6. Consume and count
let mut count = 0u64;
let mut last_ts = 0i64;
while let Some(candle) = rx.recv().await {
count += 1;
last_ts = candle.timestamp;
if count % 100000 == 0 {
info!("Progress: {} candles, last_ts: {}", count, last_ts);
}
}
let result = sync_handle.await.unwrap();
info!("Sync result: {:?}", result);
info!("Total received: {} candles", count);
info!("Last timestamp: {}", last_ts);
assert!(count > 0, "Should have synced some candles");
}
+68
View File
@@ -0,0 +1,68 @@
use backend::{BinanceCollector, CandleData};
use tokio::sync::mpsc;
use dotenv::dotenv;
use log::info;
#[tokio::test]
async fn test_build_clients() {
dotenv().ok();
env_logger::init();
info!("Testing build_clients...");
let clients = BinanceCollector::build_clients().await;
info!("Built {} working clients", clients.len());
assert!(clients.len() > 0, "Should have at least 1 working client");
}
#[tokio::test]
async fn test_sync_single_symbol() {
dotenv().ok();
let _ = env_logger::try_init();
info!("Testing sync_from_scratch for single symbol...");
// Build clients (all work together)
let clients = BinanceCollector::build_clients().await;
assert!(!clients.is_empty(), "Need at least 1 client");
info!("Built {} clients", clients.len());
// Create channel
let (tx, mut rx) = mpsc::channel::<CandleData>(10000);
// Sync last 5 minutes of BTCUSDT
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as i64;
let start_time = now - (5 * 60 * 1000); // 5 minutes ago
info!("Syncing BTCUSDT from {} to {}", start_time, now);
let count = BinanceCollector::sync_from_scratch(
"BTCUSDT".to_string(),
start_time,
clients, // All clients work together
tx,
).await.unwrap();
info!("Sync returned {} candles", count);
// Drain channel
let mut received = 0;
while let Ok(candle) = rx.try_recv() {
info!(
"Candle: {} ts={} o={:.2} h={:.2} l={:.2} c={:.2} nv={:.4}",
candle.symbol, candle.timestamp,
candle.open, candle.high, candle.low, candle.close,
candle.net_volume
);
received += 1;
}
info!("Received {} candles from channel", received);
assert!(count >= 3, "Expected at least 3 candles for 5 min, got {}", count);
}