diff --git a/CHANGELOG.md b/CHANGELOG.md index 26a6114..06e2a6d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,22 @@ All notable changes to this project will be documented in this file. +## [0.2.0] - 2026-03-21 + +### Changed + +- kept the repository as a single Cargo package with internal modules +- replaced the `download` command with the `data` command group +- expanded SQLite from candle storage into candle plus bot journal storage + +### Added + +- incremental and follow-mode candle synchronization +- live-trading runtime with dry-run and live execution modes +- manual trade close command +- monitor command group for balances, open orders, recent trades, cancel, and manual close +- signed Binance Spot client support for account and order endpoints + ## [0.1.0] - 2026-03-18 ### Added diff --git a/Cargo.lock b/Cargo.lock index 712f146..49ee67f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -158,6 +158,15 @@ dependencies = [ "wyz", ] +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + [[package]] name = "borsh" version = "1.6.1" @@ -295,6 +304,25 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + [[package]] name = "deranged" version = "0.5.8" @@ -310,6 +338,17 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6184e33543162437515c2e2b48714794e37845ec9851711914eec9d308f6ebe8" +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", + "subtle", +] + [[package]] name = "displaydoc" version = "0.2.5" @@ -424,6 +463,16 @@ dependencies = [ "slab", ] +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + [[package]] name = "getrandom" version = "0.2.17" @@ -512,6 +561,21 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "hex" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" + +[[package]] +name = "hmac" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" +dependencies = [ + "digest", +] + [[package]] name = "http" version = "1.4.0" @@ -1019,18 +1083,21 @@ dependencies = [ [[package]] name = "quantforge" -version = "0.1.0" +version = "0.2.0" dependencies = [ "anyhow", "assert_cmd", "async-trait", "clap", + "hex", + "hmac", "predicates", "reqwest", "rusqlite", "rust_decimal", "serde", "serde_json", + "sha2", "tempfile", "thiserror 1.0.69", "time", @@ -1038,6 +1105,7 @@ dependencies = [ "tracing", "tracing-subscriber", "url", + "uuid", ] [[package]] @@ -1463,6 +1531,17 @@ dependencies = [ "serde", ] +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -1478,6 +1557,16 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + [[package]] name = "simdutf8" version = "0.1.5" @@ -1706,6 +1795,7 @@ dependencies = [ "libc", "mio", "pin-project-lite", + "signal-hook-registry", "socket2", "tokio-macros", "windows-sys 0.61.2", @@ -1874,6 +1964,12 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "typenum" +version = "1.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "562d481066bde0658276a35467c4af00bdc6ee726305698a55b86e61d7ad82bb" + [[package]] name = "unicode-ident" version = "1.0.24" @@ -1922,7 +2018,9 @@ version = "1.22.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a68d3c8f01c0cfa54a75291d83601161799e4a89a39e0929f4b0354d88757a37" dependencies = [ + "getrandom 0.4.2", "js-sys", + "serde_core", "wasm-bindgen", ] diff --git a/Cargo.toml b/Cargo.toml index 79e55c6..813a8c6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,9 +1,9 @@ [package] name = "quantforge" -version = "0.1.0" +version = "0.2.0" edition = "2024" rust-version = "1.85" -description = "Deterministic CLI-first market data ingestion and backtesting framework in Rust." +description = "Deterministic CLI-first market data ingestion, backtesting, and controlled live trading framework in Rust." license = "MIT" repository = "https://github.com/formaldehid/quantforge" homepage = "https://github.com/formaldehid/quantforge" @@ -27,17 +27,21 @@ path = "src/main.rs" anyhow = "1" async-trait = "0.1" clap = { version = "4", features = ["derive", "env"] } +hex = "0.4" +hmac = "0.12" reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } rusqlite = { version = "0.32", features = ["bundled"] } rust_decimal = { version = "1", features = ["serde", "std"] } serde = { version = "1", features = ["derive"] } serde_json = "1" +sha2 = "0.10" thiserror = "1" time = { version = "0.3", features = ["formatting", "macros", "parsing"] } -tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] } +tokio = { version = "1", features = ["rt-multi-thread", "macros", "time", "signal"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } url = "2" +uuid = { version = "1", features = ["v4", "serde"] } [dev-dependencies] assert_cmd = "2" diff --git a/README.md b/README.md index c5a6626..e2cf0d5 100644 --- a/README.md +++ b/README.md @@ -1,81 +1,94 @@ # QuantForge -QuantForge is a CLI-first, deterministic trading research toolkit written in Rust. -It focuses on market data ingestion, strict normalization, reproducible backtesting, -and a strategy SDK for building TradingView/PineScript-like workflows in pure Rust. +QuantForge is a CLI-first trading systems framework in Rust for deterministic research, +SQLite-backed market data, and controlled strategy execution. -This initial release is intentionally small and opinionated: - -- Binance Spot OHLCV download over public REST endpoints -- SQLite candle storage with idempotent upserts -- validation for gaps, duplicates, ordering, and OHLC sanity -- deterministic event-driven backtesting -- built-in SMA crossover example strategy +Version `0.2.0` keeps the repository as a **single Cargo package**, following the +`0.1.0` contributor experience, while adding the first live-trading and monitoring +path for **Binance Spot**. There is **no UI**. The CLI is the product. -## Why a single crate? +## What changed in 0.2.0 + +- single-crate repository layout with internal Rust modules +- `data` command group for historical and incremental candle ingestion into SQLite +- `trade run` command for a polling strategy bot that reads closed bars from SQLite, + evaluates a strategy, and can place or close spot market orders +- `trade close` command for operator-driven position exits +- `monitor` command group for balances, open orders, recent trades, manual cancel, + and manual close +- bot run state and order or trade journal persisted in SQLite + +QuantForge is engineering software. It is **not** investment advice, does not +recommend trades, and does not promise profitability. + +## Why keep a single crate? -This repository is intentionally structured as a single Cargo package for the first -public release so versioning, publishing, installation, and contributor onboarding -stay simple. +This repository stays as a single Cargo package in `0.2.0` so versioning, +publishing, installation, and contributor onboarding remain simple. Internal boundaries still exist as Rust modules: - `model` - normalized market types and validation -- `exchange` - exchange client trait and Binance implementation -- `storage` - candle store trait and SQLite implementation -- `sdk` - strategy trait, context, indicators, example strategies +- `ports` - exchange and storage traits plus request types +- `exchange` - Binance Spot REST client +- `storage` - SQLite candle store and bot journal +- `sdk` - strategy trait, indicators, built-in strategies - `backtest` - deterministic bar-by-bar engine +- `data_sync` - historical and incremental ingestion loop +- `live` - polling live-trade engine -## Determinism contract +## Scope and guardrails -Backtests in QuantForge follow these rules: +- exchange in 0.2.0: **Binance Spot** +- storage in 0.2.0: **SQLite** +- strategy execution model: closed-bar, event-driven +- built-in strategy in 0.2.0: SMA crossover example +- default execution mode: **dry-run** +- live order placement requires explicit `--mode live` and Binance credentials from + environment variables -- timestamps are stored as UTC epoch milliseconds -- prices and volumes use decimal arithmetic, not floating-point math -- strategies observe a bar at close and submit intent -- the engine executes that intent at the **next bar open** -- data validation is explicit, not best-effort +## Credentials -## Install - -If the `quantforge` package name is available on crates.io: +QuantForge keeps secrets out of the repository. For live and monitor commands, use: ```bash -cargo install quantforge --locked +export QF_BINANCE_API_KEY="..." +export QF_BINANCE_API_SECRET="..." +export QF_BINANCE_BASE_URL="https://testnet.binance.vision/" ``` -If you are working from source: - -```bash -cargo install --path . --locked -``` +Use Binance Spot testnet first. ## Quickstart -Download candles into SQLite: +Sync candles into SQLite: ```bash -quantforge --db data/market.sqlite download \ +cargo run -- \ + --db data/market.sqlite \ + data sync \ --symbol BTCUSDT \ - --interval 1m \ - --start 2024-01-01T00:00:00Z \ - --end 2024-01-02T00:00:00Z + --interval 1m ``` Validate the stored series: ```bash -quantforge --db data/market.sqlite validate \ +cargo run -- \ + --db data/market.sqlite \ + data validate \ --symbol BTCUSDT \ --interval 1m ``` -Run the built-in SMA crossover backtest: +Backtest the SMA crossover example: ```bash -quantforge --db data/market.sqlite backtest \ +cargo run -- \ + --db data/market.sqlite \ + backtest \ --symbol BTCUSDT \ --interval 1m \ --fast 20 \ @@ -84,41 +97,97 @@ quantforge --db data/market.sqlite backtest \ --fee-bps 10 ``` -## Local development +Run the strategy against live-updating database candles without sending real orders: -Run the full quality gate before opening a PR: +```bash +cargo run -- \ + --db data/market.sqlite \ + trade run \ + --symbol BTCUSDT \ + --interval 1m \ + --fast 20 \ + --slow 50 \ + --quote-order-qty 100 \ + --mode dry-run \ + --poll-secs 5 +``` + +Run against Binance Spot testnet with live order placement: ```bash -cargo fmt --all -- --check -cargo clippy --all-targets --all-features -- -D warnings -cargo test --all-features --locked -./scripts/release-check.sh +export QF_BINANCE_BASE_URL="https://testnet.binance.vision/" + +cargo run -- \ + --db data/market.sqlite \ + trade run \ + --symbol BTCUSDT \ + --interval 1m \ + --fast 20 \ + --slow 50 \ + --quote-order-qty 100 \ + --mode live \ + --bootstrap-enter \ + --poll-secs 5 ``` -## Docker +Inspect balances, open orders, and recent trades: -Build the CLI image: +```bash +cargo run -- \ + --db data/market.sqlite \ + monitor status \ + --symbol BTCUSDT +``` + +Cancel an order manually: ```bash -docker build -t quantforge:0.1.0 . +cargo run -- \ + --db data/market.sqlite \ + monitor cancel-order \ + --symbol BTCUSDT \ + --order-id 123456789 \ + --yes ``` -Run the CLI inside Docker: +Close the bot-managed position manually: ```bash -docker run --rm -v "$PWD/data:/data" quantforge:0.1.0 \ - --db /data/market.sqlite \ - validate --symbol BTCUSDT --interval 1m +cargo run -- \ + --db data/market.sqlite \ + trade close \ + --symbol BTCUSDT \ + --yes +``` + +## Determinism contract + +- timestamps are stored as UTC epoch milliseconds +- prices and volumes use decimal arithmetic, not floating-point math +- strategies evaluate only closed bars +- backtests execute intent on the **next bar open** +- live trading polls closed bars and records run state into SQLite + +## Local development + +Run the full quality gate before opening a PR: + +```bash +cargo generate-lockfile +cargo fmt --all -- --check +cargo clippy --all-targets --all-features -- -D warnings +cargo test --all-features +./scripts/release-check.sh ``` ## Roadmap -- incremental download and resume support - additional exchanges such as Bybit -- more indicators and a richer strategy SDK -- alternative storage backends such as Parquet or Postgres +- restart reconciliation between local bot state and exchange state +- richer indicators and more built-in strategies +- alternative storage backends such as Postgres - research-oriented parameter sweeps and snapshot reports ## License -Licensed under MIT license. +Licensed under MIT. diff --git a/src/backtest.rs b/src/backtest.rs index 03ce232..6c1971b 100644 --- a/src/backtest.rs +++ b/src/backtest.rs @@ -1,20 +1,10 @@ -use rust_decimal::Decimal; -use thiserror::Error; -use tracing::info; - use crate::{ - model::{Candle, MarketId, TimestampMs}, - sdk::{Strategy, StrategyContext, StrategyError, TargetPosition}, + Candle, ClosedTrade, MarketId, Strategy, StrategyContext, TargetPosition, TimestampMs, }; +use rust_decimal::Decimal; +use tracing::info; -#[derive(Error, Debug)] -pub enum BacktestError { - #[error("no candles provided")] - NoCandles, - - #[error("strategy error: {0}")] - Strategy(#[from] StrategyError), -} +use crate::EngineError; #[derive(Clone, Debug)] pub struct BacktestConfig { @@ -33,16 +23,6 @@ impl Default for BacktestConfig { } } -#[derive(Clone, Debug, PartialEq)] -pub struct Trade { - pub entry_time_ms: TimestampMs, - pub entry_price: Decimal, - pub exit_time_ms: TimestampMs, - pub exit_price: Decimal, - pub qty: Decimal, - pub pnl: Decimal, -} - #[derive(Clone, Debug, PartialEq)] pub struct BacktestResult { pub initial_cash: Decimal, @@ -50,7 +30,7 @@ pub struct BacktestResult { pub total_return_pct: Decimal, pub trade_count: usize, pub max_drawdown_pct: Decimal, - pub trades: Vec, + pub trades: Vec, } #[derive(Clone, Debug)] @@ -68,9 +48,9 @@ impl BacktestEngine { market: &MarketId, candles: &[Candle], strategy: &mut dyn Strategy, - ) -> Result { + ) -> Result { if candles.is_empty() { - return Err(BacktestError::NoCandles); + return Err(EngineError::NoCandles); } let fee_rate = self.cfg.fee_bps / Decimal::from(10_000); @@ -87,7 +67,7 @@ impl BacktestEngine { now_ms: candles[0].open_time_ms, cash, position_qty: qty, - desired_next: None, + desired_target: None, }; strategy.on_start(&mut ctx)?; @@ -98,6 +78,7 @@ impl BacktestEngine { if index > 0 { if let Some(target) = pending_target.take() { execute_target( + market, target, candle.open, candle.open_time_ms, @@ -112,7 +93,7 @@ impl BacktestEngine { ctx.cash = cash; ctx.position_qty = qty; - ctx.desired_next = None; + ctx.desired_target = None; let equity = cash + qty * candle.close; if equity > peak_equity { @@ -126,21 +107,23 @@ impl BacktestEngine { } strategy.on_bar(&mut ctx, candle)?; - pending_target = ctx.desired_next; + pending_target = ctx.desired_target; } if self.cfg.close_out_at_end && qty > Decimal::ZERO { - let last = candles.last().expect("non-empty checked above"); - execute_target( - TargetPosition::Flat, - last.close, - last.close_time_ms, - fee_rate, - &mut cash, - &mut qty, - &mut open_trade, - &mut trades, - ); + if let Some(last) = candles.last() { + execute_target( + market, + TargetPosition::Flat, + last.close, + last.close_time_ms, + fee_rate, + &mut cash, + &mut qty, + &mut open_trade, + &mut trades, + ); + } } ctx.cash = cash; @@ -184,6 +167,7 @@ struct OpenTrade { #[allow(clippy::too_many_arguments)] fn execute_target( + market: &MarketId, target: TargetPosition, price: Decimal, timestamp_ms: TimestampMs, @@ -191,7 +175,7 @@ fn execute_target( cash: &mut Decimal, qty: &mut Decimal, open_trade: &mut Option, - trades: &mut Vec, + trades: &mut Vec, ) { match target { TargetPosition::Flat => { @@ -204,13 +188,16 @@ fn execute_target( let cash_after = *cash + notional - fee; if let Some(open_trade) = open_trade.take() { - trades.push(Trade { + trades.push(ClosedTrade { + symbol: market.symbol.clone(), entry_time_ms: open_trade.entry_time_ms, - entry_price: open_trade.entry_price, exit_time_ms: timestamp_ms, + entry_price: open_trade.entry_price, exit_price: price, qty: open_trade.qty, - pnl: cash_after - open_trade.cash_before, + gross_quote_pnl: cash_after - open_trade.cash_before, + entry_order_id: None, + exit_order_id: None, }); } @@ -250,7 +237,7 @@ struct EngineContext { now_ms: TimestampMs, cash: Decimal, position_qty: Decimal, - desired_next: Option, + desired_target: Option, } impl StrategyContext for EngineContext { @@ -271,70 +258,87 @@ impl StrategyContext for EngineContext { } fn set_target_position(&mut self, target: TargetPosition) { - self.desired_next = Some(target); + self.desired_target = Some(target); } } #[cfg(test)] mod tests { use super::*; - use crate::{ - model::{ExchangeId, Interval, Symbol}, - sdk::strategies::SmaCrossStrategy, - }; + use crate::{ExchangeId, Interval, Symbol}; use std::str::FromStr; - fn candle(open_time_ms: i64, open: &str, close: &str) -> Candle { - let open = Decimal::from_str(open).expect("decimal"); - let close = Decimal::from_str(close).expect("decimal"); - let high = open.max(close) + Decimal::ONE; - let low = open.min(close) - Decimal::ONE; + #[derive(Debug)] + struct ScriptedStrategy { + targets: Vec<(TimestampMs, TargetPosition)>, + } + + impl Strategy for ScriptedStrategy { + fn name(&self) -> &'static str { + "scripted" + } + + fn on_bar( + &mut self, + ctx: &mut dyn StrategyContext, + bar: &Candle, + ) -> Result<(), crate::StrategyError> { + for (ts, target) in &self.targets { + if *ts == bar.open_time_ms { + ctx.set_target_position(*target); + } + } + Ok(()) + } + } + + fn market() -> MarketId { + MarketId::new( + ExchangeId::BinanceSpot, + Symbol::new("BTCUSDT").expect("symbol"), + Interval::M1, + ) + } + fn candle(open_time_ms: i64, open: &str, close: &str) -> Candle { Candle { open_time_ms, close_time_ms: open_time_ms + 59_999, - open, - high, - low, - close, - volume: Decimal::from(100), + open: Decimal::from_str(open).expect("decimal"), + high: Decimal::from_str(close).expect("decimal"), + low: Decimal::from_str(open).expect("decimal"), + close: Decimal::from_str(close).expect("decimal"), + volume: Decimal::ONE, trades: Some(1), } } #[test] - fn backtest_is_deterministic() { - let market = MarketId::new( - ExchangeId::BinanceSpot, - Symbol::new("BTCUSDT").expect("symbol"), - Interval::M1, - ); + fn backtest_executes_on_next_bar_open_and_records_trade() { let candles = vec![ candle(0, "100", "100"), - candle(60_000, "100", "101"), - candle(120_000, "101", "102"), - candle(180_000, "102", "105"), - candle(240_000, "105", "103"), - candle(300_000, "103", "99"), - candle(360_000, "99", "98"), - candle(420_000, "98", "101"), + candle(60_000, "110", "110"), + candle(120_000, "120", "120"), ]; + let mut strategy = ScriptedStrategy { + targets: vec![ + (0, TargetPosition::LongAllIn), + (60_000, TargetPosition::Flat), + ], + }; - let mut strategy = SmaCrossStrategy::new(2, 3).expect("strategy"); - let engine = BacktestEngine::new(BacktestConfig { - initial_cash: Decimal::from(10_000), - fee_bps: Decimal::from(10), + let result = BacktestEngine::new(BacktestConfig { + initial_cash: Decimal::from(1_000), + fee_bps: Decimal::ZERO, close_out_at_end: true, - }); - - let result_a = engine - .run(&market, &candles, &mut strategy) - .expect("backtest"); - let mut strategy_b = SmaCrossStrategy::new(2, 3).expect("strategy"); - let result_b = engine - .run(&market, &candles, &mut strategy_b) - .expect("backtest"); - - assert_eq!(result_a, result_b); + }) + .run(&market(), &candles, &mut strategy) + .expect("backtest"); + + assert_eq!(result.trade_count, 1); + assert_eq!(result.trades[0].entry_price, Decimal::from(110)); + assert_eq!(result.trades[0].exit_price, Decimal::from(120)); + assert!(result.final_equity > Decimal::from(1_000)); + assert!(result.trades[0].gross_quote_pnl > Decimal::ZERO); } } diff --git a/src/data_sync.rs b/src/data_sync.rs new file mode 100644 index 0000000..16b67bc --- /dev/null +++ b/src/data_sync.rs @@ -0,0 +1,237 @@ +use crate::EngineError; +use crate::{CandleStore, KlineRequest, MarketDataSource, MarketId, now_utc_ms}; +use std::time::Duration; +use tracing::{info, warn}; + +#[derive(Clone, Debug)] +pub struct DataSyncConfig { + pub market: MarketId, + pub start_time_ms: Option, + pub end_time_ms: Option, + pub batch_limit: u16, + pub follow: bool, + pub poll_interval: Duration, + pub max_loops: Option, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct DataSyncSummary { + pub iterations: usize, + pub written: usize, + pub last_open_time_ms: Option, +} + +pub struct DataSyncEngine<'a> { + source: &'a dyn MarketDataSource, + store: &'a dyn CandleStore, +} + +impl<'a> std::fmt::Debug for DataSyncEngine<'a> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("DataSyncEngine").finish_non_exhaustive() + } +} + +impl<'a> DataSyncEngine<'a> { + pub fn new(source: &'a dyn MarketDataSource, store: &'a dyn CandleStore) -> Self { + Self { source, store } + } + + pub async fn run(&self, cfg: &DataSyncConfig) -> Result { + let mut summary = DataSyncSummary::default(); + let mut loops = 0usize; + let step_ms = cfg.market.interval.step_ms(); + let mut next_start_time_ms = initial_start_time_ms(cfg, now_utc_ms()); + + loop { + let now_ms = now_utc_ms(); + let end_time_ms = effective_end_time_ms(cfg, now_ms); + + if next_start_time_ms <= end_time_ms { + let written = sync_market_range( + self.source, + self.store, + &cfg.market, + next_start_time_ms, + end_time_ms, + cfg.batch_limit, + ) + .await?; + summary.written += written; + summary.last_open_time_ms = self.store.max_open_time_ms(&cfg.market)?; + + if let Some(max_open_time_ms) = summary.last_open_time_ms { + next_start_time_ms = next_start_time_ms.max(max_open_time_ms + step_ms); + } + + info!( + written, + total_written = summary.written, + next_start_time_ms, + last_open_time_ms = ?summary.last_open_time_ms, + "data sync iteration finished" + ); + } else { + info!( + next_start_time_ms, + end_time_ms, "data sync is waiting for the requested time window" + ); + } + + loops += 1; + summary.iterations = loops; + + if !should_continue(cfg, next_start_time_ms, loops) { + break; + } + if sleep_or_shutdown(cfg.poll_interval).await { + break; + } + } + + Ok(summary) + } +} + +fn initial_start_time_ms(cfg: &DataSyncConfig, now_ms: i64) -> i64 { + cfg.start_time_ms.unwrap_or(now_ms) +} + +fn effective_end_time_ms(cfg: &DataSyncConfig, now_ms: i64) -> i64 { + cfg.end_time_ms + .map(|end_time_ms| end_time_ms.min(now_ms)) + .unwrap_or(now_ms) +} + +fn should_continue(cfg: &DataSyncConfig, next_start_time_ms: i64, loops: usize) -> bool { + if cfg.max_loops.map(|max| loops >= max).unwrap_or(false) { + return false; + } + + match cfg.end_time_ms { + None => true, + Some(end_time_ms) => cfg.follow && next_start_time_ms <= end_time_ms, + } +} + +pub(crate) async fn sync_market_range( + source: &dyn MarketDataSource, + store: &dyn CandleStore, + market: &MarketId, + start_ms: i64, + end_ms: i64, + batch_limit: u16, +) -> Result { + if end_ms < start_ms { + return Ok(0); + } + + let step_ms = market.interval.step_ms(); + let mut cursor = start_ms; + let mut total = 0usize; + + while cursor <= end_ms { + let batch = source + .fetch_klines(&KlineRequest { + symbol: market.symbol.clone(), + interval: market.interval, + start_time_ms: Some(cursor), + end_time_ms: Some(end_ms), + limit: Some(batch_limit.min(1000)), + }) + .await?; + + if batch.is_empty() { + break; + } + + total += store.upsert_candles(market, &batch)?; + + let last_open_time_ms = batch + .last() + .map(|candle| candle.open_time_ms) + .ok_or_else(|| EngineError::InvalidState("expected non-empty batch".to_string()))?; + let next_cursor = last_open_time_ms + step_ms; + if next_cursor <= cursor { + warn!(cursor, next_cursor, "data sync cursor did not advance"); + break; + } + cursor = next_cursor; + } + + Ok(total) +} + +pub(crate) async fn sleep_or_shutdown(duration: Duration) -> bool { + tokio::select! { + _ = tokio::time::sleep(duration) => false, + _ = tokio::signal::ctrl_c() => true, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ExchangeId, Interval, Symbol}; + + fn market() -> MarketId { + MarketId::new( + ExchangeId::BinanceSpot, + Symbol::new("BTCUSDT").expect("symbol"), + Interval::M1, + ) + } + + fn config( + start_time_ms: Option, + end_time_ms: Option, + follow: bool, + ) -> DataSyncConfig { + DataSyncConfig { + market: market(), + start_time_ms, + end_time_ms, + batch_limit: 1000, + follow, + poll_interval: Duration::from_secs(1), + max_loops: None, + } + } + + #[test] + fn omitted_start_defaults_to_now() { + let cfg = config(None, Some(5_000), false); + assert_eq!(initial_start_time_ms(&cfg, 12_345), 12_345); + } + + #[test] + fn omitted_end_uses_current_time_for_each_iteration() { + let cfg = config(Some(1_000), None, false); + assert_eq!(effective_end_time_ms(&cfg, 9_999), 9_999); + } + + #[test] + fn explicit_end_is_capped_at_current_time() { + let cfg = config(Some(1_000), Some(20_000), false); + assert_eq!(effective_end_time_ms(&cfg, 9_999), 9_999); + } + + #[test] + fn omitted_end_keeps_sync_running_without_follow_flag() { + let cfg = config(Some(1_000), None, false); + assert!(should_continue(&cfg, 2_000, 1)); + } + + #[test] + fn bounded_range_without_follow_stops_after_first_iteration() { + let cfg = config(Some(1_000), Some(5_000), false); + assert!(!should_continue(&cfg, 2_000, 1)); + } + + #[test] + fn bounded_range_with_follow_continues_until_end_is_reached() { + let cfg = config(Some(1_000), Some(5_000), true); + assert!(should_continue(&cfg, 5_000, 1)); + assert!(!should_continue(&cfg, 5_001, 1)); + } +} diff --git a/src/engine.rs b/src/engine.rs new file mode 100644 index 0000000..f114807 --- /dev/null +++ b/src/engine.rs @@ -0,0 +1,24 @@ +use thiserror::Error; + +use crate::{ExchangeError, StorageError, StrategyError}; + +#[derive(Error, Debug)] +pub enum EngineError { + #[error("no candles provided")] + NoCandles, + + #[error("invalid configuration: {0}")] + InvalidConfig(String), + + #[error("invalid state: {0}")] + InvalidState(String), + + #[error(transparent)] + Strategy(#[from] StrategyError), + + #[error(transparent)] + Exchange(#[from] ExchangeError), + + #[error(transparent)] + Storage(#[from] StorageError), +} diff --git a/src/exchange.rs b/src/exchange.rs index 5c603fa..efd44be 100644 --- a/src/exchange.rs +++ b/src/exchange.rs @@ -1,53 +1,47 @@ +use crate::{ + AccountTrade, AssetBalance, CancelOrderRequest, Candle, ExchangeError, ExchangeId, + ExchangeOrder, Fill, KlineRequest, MarketDataSource, MarketOrderRequest, OrderQueryRequest, + OrderStatus, Side, Symbol, SymbolRules, TradingVenue, now_utc_ms, +}; use async_trait::async_trait; +use hmac::{Hmac, Mac}; +use reqwest::{Method, StatusCode}; use rust_decimal::Decimal; use serde::{Deserialize, Serialize}; -use std::error::Error as StdError; -use thiserror::Error; -use tracing::debug; -use url::Url; +use serde_json::Value; +use sha2::Sha256; +use std::env; +use url::{Url, form_urlencoded}; -use crate::model::{Candle, ExchangeId, Interval, Symbol, TimestampMs}; +type HmacSha256 = Hmac; #[derive(Clone, Debug)] -pub struct KlineRequest { - pub symbol: Symbol, - pub interval: Interval, - pub start_time_ms: Option, - pub end_time_ms: Option, - pub limit: Option, +pub struct BinanceCredentials { + pub api_key: String, + pub secret: String, } -#[derive(Error, Debug)] -pub enum ExchangeError { - #[error("transport error")] - Transport(#[source] Box), - - #[error("api error: code={code:?} message={message}")] - Api { code: Option, message: String }, - - #[error("invalid response: {message}")] - InvalidResponse { message: String }, -} - -impl ExchangeError { - pub fn transport(err: E) -> Self - where - E: StdError + Send + Sync + 'static, - { - Self::Transport(Box::new(err)) +impl BinanceCredentials { + pub fn from_env() -> Option { + let api_key = env::var("QF_BINANCE_API_KEY").ok()?; + let secret = env::var("QF_BINANCE_API_SECRET").ok()?; + Some(Self { api_key, secret }) } -} -#[async_trait] -pub trait MarketDataSource: Send + Sync { - fn exchange_id(&self) -> ExchangeId; - async fn fetch_klines(&self, req: &KlineRequest) -> Result, ExchangeError>; + pub fn from_required_env() -> Result { + match Self::from_env() { + Some(value) => Ok(value), + None => Err(ExchangeError::MissingCredentials), + } + } } #[derive(Clone, Debug)] pub struct BinanceSpotClient { base_url: Url, http: reqwest::Client, + credentials: Option, + recv_window_ms: u64, } impl BinanceSpotClient { @@ -58,36 +52,617 @@ impl BinanceSpotClient { Self { base_url, http: reqwest::Client::new(), + credentials: None, + recv_window_ms: 5_000, } } - pub fn with_http(mut base_url: Url, http: reqwest::Client) -> Self { - if !base_url.as_str().ends_with('/') { - base_url.set_path(&format!("{}/", base_url.path())); + pub fn with_credentials(mut self, credentials: BinanceCredentials) -> Self { + self.credentials = Some(credentials); + self + } + + pub fn with_http(mut self, http: reqwest::Client) -> Self { + self.http = http; + self + } + + pub fn with_recv_window_ms(mut self, recv_window_ms: u64) -> Self { + self.recv_window_ms = recv_window_ms; + self + } + + fn join(&self, path: &str) -> Result { + self.base_url + .join(path) + .map_err(|err| ExchangeError::InvalidResponse { + message: format!("failed to join base URL and path `{path}`: {err}"), + }) + } + + fn signed_query(&self, params: Vec<(&str, String)>) -> Result { + let credentials = self + .credentials + .as_ref() + .ok_or(ExchangeError::MissingCredentials)?; + + let mut serializer = form_urlencoded::Serializer::new(String::new()); + for (key, value) in params { + serializer.append_pair(key, &value); + } + serializer.append_pair("recvWindow", &self.recv_window_ms.to_string()); + serializer.append_pair("timestamp", &now_utc_ms().to_string()); + let query = serializer.finish(); + + let mut mac = HmacSha256::new_from_slice(credentials.secret.as_bytes()).map_err(|err| { + ExchangeError::InvalidRequest { + message: format!("invalid HMAC secret: {err}"), + } + })?; + mac.update(query.as_bytes()); + let signature = hex::encode(mac.finalize().into_bytes()); + + Ok(format!("{query}&signature={signature}")) + } + + async fn send_public( + &self, + method: Method, + path: &str, + params: Vec<(&str, String)>, + ) -> Result { + let mut url = self.join(path)?; + let query = encode_query(params); + if !query.is_empty() { + url.set_query(Some(&query)); } - Self { base_url, http } + + let response = self + .http + .request(method, url) + .send() + .await + .map_err(ExchangeError::transport)?; + + decode_json(response).await + } + + async fn send_signed( + &self, + method: Method, + path: &str, + params: Vec<(&str, String)>, + ) -> Result { + let credentials = self + .credentials + .as_ref() + .ok_or(ExchangeError::MissingCredentials)?; + + let mut url = self.join(path)?; + let query = self.signed_query(params)?; + url.set_query(Some(&query)); + + let response = self + .http + .request(method, url) + .header("X-MBX-APIKEY", &credentials.api_key) + .send() + .await + .map_err(ExchangeError::transport)?; + + decode_json(response).await } } -#[derive(Serialize)] -struct KlinesParams<'a> { - symbol: &'a str, - interval: &'a str, - #[serde(rename = "startTime", skip_serializing_if = "Option::is_none")] - start_time: Option, - #[serde(rename = "endTime", skip_serializing_if = "Option::is_none")] - end_time: Option, - #[serde(skip_serializing_if = "Option::is_none")] - limit: Option, +#[async_trait] +impl MarketDataSource for BinanceSpotClient { + fn exchange_id(&self) -> ExchangeId { + ExchangeId::BinanceSpot + } + + async fn fetch_klines(&self, request: &KlineRequest) -> Result, ExchangeError> { + let mut params = vec![ + ("symbol", request.symbol.as_str().to_string()), + ("interval", request.interval.as_str().to_string()), + ]; + if let Some(start_time_ms) = request.start_time_ms { + params.push(("startTime", start_time_ms.to_string())); + } + if let Some(end_time_ms) = request.end_time_ms { + params.push(("endTime", end_time_ms.to_string())); + } + if let Some(limit) = request.limit { + params.push(("limit", limit.min(1000).to_string())); + } + + let raw = self + .send_public(Method::GET, "api/v3/klines", params) + .await?; + let rows: Vec = + serde_json::from_value(raw).map_err(|err| ExchangeError::InvalidResponse { + message: format!("failed to decode klines response: {err}"), + })?; + + let mut candles = Vec::with_capacity(rows.len()); + for row in rows { + candles.push(Candle { + open_time_ms: row.0, + open: parse_decimal(&row.1, "open")?, + high: parse_decimal(&row.2, "high")?, + low: parse_decimal(&row.3, "low")?, + close: parse_decimal(&row.4, "close")?, + volume: parse_decimal(&row.5, "volume")?, + close_time_ms: row.6, + trades: Some(row.8), + }); + } + + Ok(candles) + } + + async fn fetch_symbol_rules(&self, symbol: &Symbol) -> Result { + let raw = self + .send_public( + Method::GET, + "api/v3/exchangeInfo", + vec![("symbol", symbol.as_str().to_string())], + ) + .await?; + + let response: BinanceExchangeInfoResponse = + serde_json::from_value(raw).map_err(|err| ExchangeError::InvalidResponse { + message: format!("failed to decode exchangeInfo response: {err}"), + })?; + + let info = + response + .symbols + .into_iter() + .next() + .ok_or_else(|| ExchangeError::InvalidResponse { + message: format!( + "exchangeInfo returned no symbol rules for {}", + symbol.as_str() + ), + })?; + + Ok(parse_symbol_rules(info)?) + } } -#[derive(Deserialize, Debug)] +#[async_trait] +impl TradingVenue for BinanceSpotClient { + fn exchange_id(&self) -> ExchangeId { + ExchangeId::BinanceSpot + } + + async fn account_balances(&self) -> Result, ExchangeError> { + let raw = self + .send_signed( + Method::GET, + "api/v3/account", + vec![("omitZeroBalances", "true".to_string())], + ) + .await?; + + let response: BinanceAccountResponse = + serde_json::from_value(raw).map_err(|err| ExchangeError::InvalidResponse { + message: format!("failed to decode account response: {err}"), + })?; + + let mut balances = Vec::with_capacity(response.balances.len()); + for balance in response.balances { + balances.push(AssetBalance { + asset: balance.asset, + free: parse_decimal(&balance.free, "free")?, + locked: parse_decimal(&balance.locked, "locked")?, + }); + } + Ok(balances) + } + + async fn open_orders( + &self, + symbol: Option<&Symbol>, + ) -> Result, ExchangeError> { + let mut params = Vec::new(); + if let Some(symbol) = symbol { + params.push(("symbol", symbol.as_str().to_string())); + } + + let raw = self + .send_signed(Method::GET, "api/v3/openOrders", params) + .await?; + let raw_items: Vec = + serde_json::from_value(raw).map_err(|err| ExchangeError::InvalidResponse { + message: format!("failed to decode openOrders response: {err}"), + })?; + + raw_items.into_iter().map(parse_order).collect() + } + + async fn recent_trades( + &self, + symbol: &Symbol, + limit: usize, + ) -> Result, ExchangeError> { + let raw = self + .send_signed( + Method::GET, + "api/v3/myTrades", + vec![ + ("symbol", symbol.as_str().to_string()), + ("limit", limit.min(1000).to_string()), + ], + ) + .await?; + + let rows: Vec = + serde_json::from_value(raw).map_err(|err| ExchangeError::InvalidResponse { + message: format!("failed to decode myTrades response: {err}"), + })?; + + let mut trades = Vec::with_capacity(rows.len()); + for row in rows { + trades.push(AccountTrade { + symbol: Symbol::new(row.symbol)?, + trade_id: row.id, + order_id: row.order_id, + side: if row.is_buyer { Side::Buy } else { Side::Sell }, + price: parse_decimal(&row.price, "trade.price")?, + qty: parse_decimal(&row.qty, "trade.qty")?, + quote_qty: parse_decimal(&row.quote_qty, "trade.quoteQty")?, + commission: parse_decimal(&row.commission, "trade.commission")?, + commission_asset: Some(row.commission_asset), + time_ms: row.time, + }); + } + + Ok(trades) + } + + async fn submit_market_order( + &self, + request: &MarketOrderRequest, + ) -> Result { + if request.quantity.is_none() && request.quote_order_qty.is_none() { + return Err(ExchangeError::InvalidRequest { + message: "submit_market_order requires quantity or quote_order_qty".to_string(), + }); + } + + let mut params = vec![ + ("symbol", request.symbol.as_str().to_string()), + ("side", request.side.as_str().to_string()), + ("type", "MARKET".to_string()), + ("newOrderRespType", "FULL".to_string()), + ]; + + if let Some(quantity) = request.quantity { + params.push(("quantity", quantity.to_string())); + } + if let Some(quote_order_qty) = request.quote_order_qty { + params.push(("quoteOrderQty", quote_order_qty.to_string())); + } + if let Some(client_order_id) = &request.new_client_order_id { + params.push(("newClientOrderId", client_order_id.clone())); + } + + let raw = self + .send_signed(Method::POST, "api/v3/order", params) + .await?; + parse_order(raw) + } + + async fn cancel_order( + &self, + request: &CancelOrderRequest, + ) -> Result { + let mut params = vec![("symbol", request.symbol.as_str().to_string())]; + if let Some(order_id) = request.order_id { + params.push(("orderId", order_id.to_string())); + } + if let Some(client_order_id) = &request.client_order_id { + params.push(("origClientOrderId", client_order_id.clone())); + } + if request.order_id.is_none() && request.client_order_id.is_none() { + return Err(ExchangeError::InvalidRequest { + message: "cancel_order requires order_id or client_order_id".to_string(), + }); + } + + let raw = self + .send_signed(Method::DELETE, "api/v3/order", params) + .await?; + parse_order(raw) + } + + async fn query_order( + &self, + request: &OrderQueryRequest, + ) -> Result { + let mut params = vec![("symbol", request.symbol.as_str().to_string())]; + if let Some(order_id) = request.order_id { + params.push(("orderId", order_id.to_string())); + } + if let Some(client_order_id) = &request.client_order_id { + params.push(("origClientOrderId", client_order_id.clone())); + } + if request.order_id.is_none() && request.client_order_id.is_none() { + return Err(ExchangeError::InvalidRequest { + message: "query_order requires order_id or client_order_id".to_string(), + }); + } + + let raw = self + .send_signed(Method::GET, "api/v3/order", params) + .await?; + parse_order(raw) + } +} + +async fn decode_json(response: reqwest::Response) -> Result { + let status = response.status(); + let body = response.text().await.map_err(ExchangeError::transport)?; + + if !status.is_success() { + if let Ok(api_error) = serde_json::from_str::(&body) { + return Err(ExchangeError::Api { + code: Some(api_error.code), + message: api_error.msg, + }); + } + + return Err(ExchangeError::Api { + code: status_to_code(status), + message: format!("http {status}: {body}"), + }); + } + + serde_json::from_str::(&body).map_err(|err| ExchangeError::InvalidResponse { + message: format!("failed to decode JSON body: {err}; body={body}"), + }) +} + +fn status_to_code(status: StatusCode) -> Option { + Some(i64::from(status.as_u16())) +} + +fn encode_query(params: Vec<(&str, String)>) -> String { + let mut serializer = form_urlencoded::Serializer::new(String::new()); + for (key, value) in params { + serializer.append_pair(key, &value); + } + serializer.finish() +} + +fn parse_symbol_rules(info: BinanceSymbolInfo) -> Result { + let mut rules = SymbolRules { + symbol: Symbol::new(info.symbol)?, + base_asset: info.base_asset, + quote_asset: info.quote_asset, + min_qty: None, + max_qty: None, + step_size: None, + market_min_qty: None, + market_max_qty: None, + market_step_size: None, + min_notional: None, + tick_size: None, + }; + + for filter in info.filters { + let filter_type = filter + .get("filterType") + .and_then(Value::as_str) + .unwrap_or_default(); + + match filter_type { + "LOT_SIZE" => { + rules.min_qty = parse_optional_filter_decimal(&filter, "minQty")?; + rules.max_qty = parse_optional_filter_decimal(&filter, "maxQty")?; + rules.step_size = parse_optional_filter_decimal(&filter, "stepSize")?; + } + "MARKET_LOT_SIZE" => { + rules.market_min_qty = parse_optional_filter_decimal(&filter, "minQty")?; + rules.market_max_qty = parse_optional_filter_decimal(&filter, "maxQty")?; + rules.market_step_size = parse_optional_filter_decimal(&filter, "stepSize")?; + } + "MIN_NOTIONAL" => { + rules.min_notional = parse_optional_filter_decimal(&filter, "minNotional")?; + } + "NOTIONAL" => { + if rules.min_notional.is_none() { + rules.min_notional = parse_optional_filter_decimal(&filter, "minNotional")?; + } + } + "PRICE_FILTER" => { + rules.tick_size = parse_optional_filter_decimal(&filter, "tickSize")?; + } + _ => {} + } + } + + Ok(rules) +} + +fn parse_optional_filter_decimal( + filter: &Value, + field: &str, +) -> Result, ExchangeError> { + match filter.get(field).and_then(Value::as_str) { + Some(raw) => Ok(Some(parse_decimal(raw, field)?)), + None => Ok(None), + } +} + +fn parse_order(raw: Value) -> Result { + let response: BinanceOrderResponse = + serde_json::from_value(raw.clone()).map_err(|err| ExchangeError::InvalidResponse { + message: format!("failed to decode order response: {err}; raw={raw}"), + })?; + + let symbol = Symbol::new(response.symbol)?; + let side = response.side.parse::()?; + let fills = response + .fills + .into_iter() + .map(|fill| { + Ok(Fill { + price: parse_decimal(&fill.price, "fill.price")?, + qty: parse_decimal(&fill.qty, "fill.qty")?, + commission: parse_decimal(&fill.commission, "fill.commission")?, + commission_asset: fill.commission_asset, + trade_id: fill.trade_id, + }) + }) + .collect::, ExchangeError>>()?; + + let requested_qty = match response.orig_qty { + Some(value) => Some(parse_decimal(&value, "origQty")?), + None => None, + }; + let requested_quote_qty = match response.orig_quote_order_qty { + Some(value) => { + let parsed = parse_decimal(&value, "origQuoteOrderQty")?; + if parsed > Decimal::ZERO { + Some(parsed) + } else { + None + } + } + None => None, + }; + + let executed_qty = parse_decimal(&response.executed_qty, "executedQty")?; + let cumulative_quote_qty = parse_decimal(&response.cumulative_quote_qty, "cumulativeQuoteQty")?; + let avg_price = if executed_qty > Decimal::ZERO { + Some(cumulative_quote_qty / executed_qty) + } else { + None + }; + + Ok(ExchangeOrder { + symbol, + side, + order_type: response.order_type, + status: OrderStatus::from_exchange(response.status.as_deref().unwrap_or("UNKNOWN")), + order_id: response.order_id, + client_order_id: response.client_order_id, + requested_qty, + requested_quote_qty, + executed_qty, + cumulative_quote_qty, + avg_price, + transact_time_ms: response.transact_time, + fills, + raw, + }) +} + +fn parse_decimal(raw: &str, field: &str) -> Result { + raw.parse::() + .map_err(|err| ExchangeError::InvalidResponse { + message: format!("failed to parse decimal field `{field}`: {err}"), + }) +} + +#[derive(Debug, Deserialize)] struct BinanceApiError { code: i64, msg: String, } -#[derive(Deserialize, Debug)] +#[derive(Debug, Deserialize)] +struct BinanceExchangeInfoResponse { + symbols: Vec, +} + +#[derive(Debug, Deserialize)] +struct BinanceSymbolInfo { + symbol: String, + #[serde(rename = "baseAsset")] + base_asset: String, + #[serde(rename = "quoteAsset")] + quote_asset: String, + filters: Vec, +} + +#[derive(Debug, Deserialize)] +struct BinanceAccountResponse { + balances: Vec, +} + +#[derive(Debug, Deserialize)] +struct BinanceBalance { + asset: String, + free: String, + locked: String, +} + +#[derive(Debug, Deserialize)] +struct BinanceAccountTrade { + symbol: String, + id: i64, + #[serde(rename = "orderId")] + order_id: i64, + price: String, + qty: String, + #[serde(rename = "quoteQty")] + quote_qty: String, + commission: String, + #[serde(rename = "commissionAsset")] + commission_asset: String, + time: i64, + #[serde(rename = "isBuyer")] + is_buyer: bool, +} + +#[derive(Debug, Deserialize)] +struct BinanceFill { + price: String, + qty: String, + commission: String, + #[serde(rename = "commissionAsset")] + commission_asset: Option, + #[serde(rename = "tradeId")] + trade_id: Option, +} + +fn default_zero_string() -> String { + "0".to_string() +} + +#[derive(Debug, Deserialize)] +struct BinanceOrderResponse { + symbol: String, + #[serde(rename = "orderId")] + order_id: Option, + #[serde(rename = "clientOrderId")] + client_order_id: Option, + side: String, + #[serde(rename = "type")] + order_type: String, + status: Option, + #[serde(rename = "origQty")] + orig_qty: Option, + #[serde(rename = "origQuoteOrderQty")] + orig_quote_order_qty: Option, + #[serde(rename = "executedQty", default = "default_zero_string")] + executed_qty: String, + #[serde( + rename = "cummulativeQuoteQty", + alias = "cumulativeQuoteQty", + default = "default_zero_string" + )] + cumulative_quote_qty: String, + #[serde(rename = "transactTime")] + transact_time: Option, + #[serde(default)] + fills: Vec, +} + +#[derive(Debug, Deserialize, Serialize)] struct BinanceKlineRow( i64, String, @@ -103,118 +678,40 @@ struct BinanceKlineRow( String, ); -#[async_trait] -impl MarketDataSource for BinanceSpotClient { - fn exchange_id(&self) -> ExchangeId { - ExchangeId::BinanceSpot - } +#[cfg(test)] +mod tests { + use super::*; - async fn fetch_klines(&self, req: &KlineRequest) -> Result, ExchangeError> { - let url = - self.base_url - .join("api/v3/klines") - .map_err(|e| ExchangeError::InvalidResponse { - message: format!("invalid base url join: {e}"), - })?; - - let params = KlinesParams { - symbol: req.symbol.as_str(), - interval: req.interval.as_str(), - start_time: req.start_time_ms, - end_time: req.end_time_ms, - limit: req.limit.map(|value| value.min(1000)), - }; - - debug!( - %url, - symbol = params.symbol, - interval = params.interval, - start_time_ms = params.start_time, - end_time_ms = params.end_time, - limit = params.limit, - "fetching klines" + #[test] + fn encode_query_works() { + assert_eq!( + encode_query(vec![ + ("symbol", "BTCUSDT".to_string()), + ("limit", "10".to_string()) + ]), + "symbol=BTCUSDT&limit=10" ); + } - let response = self - .http - .get(url) - .query(¶ms) - .send() - .await - .map_err(ExchangeError::transport)?; - - let status = response.status(); - if !status.is_success() { - let body = response.text().await.map_err(ExchangeError::transport)?; - if let Ok(api_error) = serde_json::from_str::(&body) { - return Err(ExchangeError::Api { - code: Some(api_error.code), - message: api_error.msg, - }); - } - return Err(ExchangeError::Api { - code: None, - message: format!("http {status}: {body}"), - }); - } - - let rows: Vec = response.json().await.map_err(ExchangeError::transport)?; - let mut candles = Vec::with_capacity(rows.len()); - - for BinanceKlineRow( - open_time, - open_s, - high_s, - low_s, - close_s, - volume_s, - close_time, - _quote_asset_volume, - trades, - _taker_buy_base_volume, - _taker_buy_quote_volume, - _ignore, - ) in rows - { - let open = open_s - .parse::() - .map_err(|e| ExchangeError::InvalidResponse { - message: format!("failed to parse open decimal: {e}"), - })?; - let high = high_s - .parse::() - .map_err(|e| ExchangeError::InvalidResponse { - message: format!("failed to parse high decimal: {e}"), - })?; - let low = low_s - .parse::() - .map_err(|e| ExchangeError::InvalidResponse { - message: format!("failed to parse low decimal: {e}"), - })?; - let close = close_s - .parse::() - .map_err(|e| ExchangeError::InvalidResponse { - message: format!("failed to parse close decimal: {e}"), - })?; - let volume = - volume_s - .parse::() - .map_err(|e| ExchangeError::InvalidResponse { - message: format!("failed to parse volume decimal: {e}"), - })?; - - candles.push(Candle { - open_time_ms: open_time, - close_time_ms: close_time, - open, - high, - low, - close, - volume, - trades: Some(trades), - }); - } + #[test] + fn parse_order_maps_side_and_status() { + let raw = serde_json::json!({ + "symbol": "BTCUSDT", + "orderId": 7, + "clientOrderId": "abc", + "side": "BUY", + "type": "MARKET", + "status": "FILLED", + "origQty": "0.01000000", + "executedQty": "0.01000000", + "cummulativeQuoteQty": "100.00000000", + "transactTime": 1, + "fills": [] + }); - Ok(candles) + let order = parse_order(raw).expect("order"); + assert_eq!(order.side, Side::Buy); + assert_eq!(order.status, OrderStatus::Filled); + assert_eq!(order.order_id, Some(7)); } } diff --git a/src/lib.rs b/src/lib.rs index 17cb994..5664555 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,16 +3,32 @@ #![doc = include_str!("../README.md")] pub mod backtest; +pub mod data_sync; +pub mod engine; pub mod exchange; +pub mod live; pub mod model; +pub mod ports; pub mod sdk; pub mod storage; -pub use backtest::{BacktestConfig, BacktestEngine, BacktestError, BacktestResult, Trade}; -pub use exchange::{BinanceSpotClient, ExchangeError, KlineRequest, MarketDataSource}; +pub use backtest::{BacktestConfig, BacktestEngine, BacktestResult}; +pub use data_sync::{DataSyncConfig, DataSyncEngine, DataSyncSummary}; +pub use engine::EngineError; +pub use exchange::{BinanceCredentials, BinanceSpotClient}; +pub use live::{LiveTradeConfig, LiveTradeEngine, LiveTradeSummary}; pub use model::{ - Candle, ExchangeId, Interval, MarketId, ModelError, Symbol, TimestampMs, ValidationIssue, - ValidationReport, ms_to_rfc3339, parse_rfc3339_to_ms, validate_candles, + AccountTrade, AssetBalance, BotRunState, Candle, ClosedTrade, ExchangeId, ExchangeOrder, + ExecutionMode, Fill, Interval, MarketId, ModelError, OrderStatus, PositionState, RunStatus, + Side, Symbol, SymbolRules, TargetPosition, TimestampMs, ValidationIssue, ValidationReport, + ms_to_rfc3339, now_utc_ms, parse_rfc3339_to_ms, round_down_to_step, validate_candles, }; -pub use sdk::{Sma, Strategy, StrategyContext, StrategyError, TargetPosition}; -pub use storage::{CandleQuery, CandleStore, SqliteCandleStore, StorageError}; +pub use ports::{ + CancelOrderRequest, CandleQuery, CandleStore, ExchangeError, KlineRequest, MarketDataSource, + MarketOrderRequest, OrderQueryRequest, RunJournalStore, StorageError, TradingVenue, +}; +pub use sdk::{ + BuiltInStrategyConfig, Indicator, Sma, Strategy, StrategyContext, StrategyError, + strategies::SmaCrossStrategy, +}; +pub use storage::SqliteStore; diff --git a/src/live.rs b/src/live.rs new file mode 100644 index 0000000..0907aeb --- /dev/null +++ b/src/live.rs @@ -0,0 +1,685 @@ +use crate::{ + BotRunState, BuiltInStrategyConfig, Candle, CandleQuery, CandleStore, ClosedTrade, + ExchangeOrder, ExecutionMode, MarketDataSource, MarketId, MarketOrderRequest, PositionState, + RunJournalStore, RunStatus, Side, Strategy, StrategyContext, SymbolRules, TargetPosition, + TimestampMs, TradingVenue, now_utc_ms, round_down_to_step, +}; +use crate::{ + EngineError, + data_sync::{sleep_or_shutdown, sync_market_range}, +}; +use rust_decimal::Decimal; +use std::time::Duration; +use tracing::{info, warn}; +use uuid::Uuid; + +#[derive(Clone, Debug)] +pub struct LiveTradeConfig { + pub market: MarketId, + pub strategy: BuiltInStrategyConfig, + pub execution_mode: ExecutionMode, + pub quote_order_qty: Decimal, + pub poll_interval: Duration, + pub bootstrap_bars: usize, + pub bootstrap_enter: bool, + pub batch_limit: u16, + pub run_id: Option, + pub max_loops: Option, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct LiveTradeSummary { + pub run_id: String, + pub processed_bars: usize, + pub submitted_orders: usize, + pub closed_trades: usize, + pub last_processed_open_time_ms: Option, +} + +pub struct LiveTradeEngine<'a> { + market_data: &'a dyn MarketDataSource, + candle_store: &'a dyn CandleStore, + journal_store: &'a dyn RunJournalStore, + trading_venue: Option<&'a dyn TradingVenue>, +} + +impl<'a> std::fmt::Debug for LiveTradeEngine<'a> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("LiveTradeEngine").finish_non_exhaustive() + } +} + +impl<'a> LiveTradeEngine<'a> { + pub fn new( + market_data: &'a dyn MarketDataSource, + candle_store: &'a dyn CandleStore, + journal_store: &'a dyn RunJournalStore, + trading_venue: Option<&'a dyn TradingVenue>, + ) -> Self { + Self { + market_data, + candle_store, + journal_store, + trading_venue, + } + } + + pub async fn run(&self, cfg: &LiveTradeConfig) -> Result { + let mut summary = LiveTradeSummary::default(); + let mut run_state = self.load_or_create_run_state(cfg)?; + let rules = self + .market_data + .fetch_symbol_rules(&cfg.market.symbol) + .await?; + let mut strategy = cfg.strategy.build()?; + + let result = self + .run_inner(cfg, &rules, &mut run_state, strategy.as_mut(), &mut summary) + .await; + + match result { + Ok(()) => { + run_state.status = RunStatus::Stopped; + run_state.updated_at_ms = now_utc_ms(); + run_state.stopped_at_ms = Some(run_state.updated_at_ms); + self.journal_store.save_run_state(&run_state)?; + summary.run_id = run_state.run_id.clone(); + summary.last_processed_open_time_ms = run_state.last_processed_open_time_ms; + Ok(summary) + } + Err(err) => { + run_state.status = RunStatus::Failed; + run_state.updated_at_ms = now_utc_ms(); + run_state.last_error = Some(err.to_string()); + run_state.stopped_at_ms = Some(run_state.updated_at_ms); + self.journal_store.save_run_state(&run_state)?; + Err(err) + } + } + } + + fn load_or_create_run_state(&self, cfg: &LiveTradeConfig) -> Result { + if let Some(run_id) = &cfg.run_id { + if let Some(existing) = self.journal_store.load_run_state(run_id)? { + return Ok(existing); + } + } + + let now_ms = now_utc_ms(); + Ok(BotRunState { + run_id: cfg + .run_id + .clone() + .unwrap_or_else(|| format!("qf-{}", Uuid::new_v4().simple())), + market: cfg.market.clone(), + strategy_name: cfg.strategy.strategy_name().to_string(), + strategy_config: serde_json::to_value(&cfg.strategy).map_err(|err| { + EngineError::InvalidConfig(format!("failed to serialize strategy config: {err}")) + })?, + status: RunStatus::Starting, + last_processed_open_time_ms: None, + started_at_ms: now_ms, + updated_at_ms: now_ms, + stopped_at_ms: None, + last_error: None, + position: PositionState::flat(), + }) + } + + async fn run_inner( + &self, + cfg: &LiveTradeConfig, + rules: &SymbolRules, + run_state: &mut BotRunState, + strategy: &mut dyn Strategy, + summary: &mut LiveTradeSummary, + ) -> Result<(), EngineError> { + self.journal_store.save_run_state(run_state)?; + + let now = now_utc_ms(); + let bootstrap_start = run_state + .last_processed_open_time_ms + .map(|value| value + cfg.market.interval.step_ms()) + .or_else(|| Some(now - (cfg.market.interval.step_ms() * cfg.bootstrap_bars as i64))); + + if let Some(start_ms) = bootstrap_start { + sync_market_range( + self.market_data, + self.candle_store, + &cfg.market, + start_ms, + now, + cfg.batch_limit, + ) + .await?; + } + + let mut ctx = LiveStrategyContext::new(cfg.market.clone(), run_state.position.qty); + strategy.on_start(&mut ctx)?; + + let bootstrap_candles = self + .candle_store + .load_recent_candles(&cfg.market, cfg.bootstrap_bars)?; + let closed_bootstrap = filter_closed_candles(bootstrap_candles); + + let mut last_bootstrap_target = current_target(&run_state.position); + for candle in &closed_bootstrap { + ctx.now_ms = candle.close_time_ms; + ctx.position_qty = run_state.position.qty; + ctx.desired_target = None; + strategy.on_bar(&mut ctx, candle)?; + if let Some(target) = ctx.desired_target { + last_bootstrap_target = target; + } + } + + if run_state.last_processed_open_time_ms.is_none() { + run_state.last_processed_open_time_ms = closed_bootstrap.last().map(|c| c.open_time_ms); + run_state.status = RunStatus::Running; + run_state.updated_at_ms = now_utc_ms(); + + if cfg.bootstrap_enter && last_bootstrap_target != current_target(&run_state.position) { + if let Some(reference_bar) = closed_bootstrap.last() { + self.execute_target( + cfg, + rules, + run_state, + last_bootstrap_target, + reference_bar, + summary, + ) + .await?; + } + } + self.journal_store.save_run_state(run_state)?; + } + + let mut loops = 0usize; + loop { + let end_ms = now_utc_ms(); + let start_ms = run_state + .last_processed_open_time_ms + .map(|value| value + cfg.market.interval.step_ms()) + .unwrap_or_else(|| end_ms - cfg.market.interval.step_ms()); + + if start_ms <= end_ms { + sync_market_range( + self.market_data, + self.candle_store, + &cfg.market, + start_ms, + end_ms, + cfg.batch_limit, + ) + .await?; + } + + let new_candles = self.candle_store.load_candles( + &cfg.market, + CandleQuery { + start_time_ms: run_state + .last_processed_open_time_ms + .map(|value| value + cfg.market.interval.step_ms()), + end_time_ms: None, + limit: None, + }, + )?; + let closed_new_candles = filter_closed_candles(new_candles); + + for candle in closed_new_candles { + if run_state + .last_processed_open_time_ms + .map(|value| candle.open_time_ms <= value) + .unwrap_or(false) + { + continue; + } + + ctx.now_ms = candle.close_time_ms; + ctx.position_qty = run_state.position.qty; + ctx.desired_target = None; + + strategy.on_bar(&mut ctx, &candle)?; + let desired = ctx + .desired_target + .unwrap_or_else(|| current_target(&run_state.position)); + + if desired != current_target(&run_state.position) { + self.execute_target(cfg, rules, run_state, desired, &candle, summary) + .await?; + } + + run_state.last_processed_open_time_ms = Some(candle.open_time_ms); + run_state.status = RunStatus::Running; + run_state.updated_at_ms = now_utc_ms(); + self.journal_store.save_run_state(run_state)?; + summary.processed_bars += 1; + summary.last_processed_open_time_ms = run_state.last_processed_open_time_ms; + } + + loops += 1; + if cfg.max_loops.map(|max| loops >= max).unwrap_or(false) { + break; + } + if sleep_or_shutdown(cfg.poll_interval).await { + break; + } + } + + strategy.on_finish(&mut ctx)?; + Ok(()) + } + + async fn execute_target( + &self, + cfg: &LiveTradeConfig, + rules: &SymbolRules, + run_state: &mut BotRunState, + target: TargetPosition, + reference_bar: &Candle, + summary: &mut LiveTradeSummary, + ) -> Result<(), EngineError> { + let order = match cfg.execution_mode { + ExecutionMode::DryRun => synthetic_market_order( + rules, + run_state, + target, + cfg.quote_order_qty, + reference_bar, + )?, + ExecutionMode::Live => { + let venue = self.trading_venue.ok_or_else(|| { + EngineError::InvalidConfig("live mode requires a trading venue".to_string()) + })?; + match target { + TargetPosition::LongAllIn => { + if let Some(min_notional) = rules.min_notional { + if cfg.quote_order_qty < min_notional { + return Err(EngineError::InvalidConfig(format!( + "quote_order_qty {} is below exchange min_notional {}", + cfg.quote_order_qty, min_notional + ))); + } + } + + venue + .submit_market_order(&MarketOrderRequest { + symbol: cfg.market.symbol.clone(), + side: Side::Buy, + quantity: None, + quote_order_qty: Some(cfg.quote_order_qty), + new_client_order_id: Some(new_client_order_id( + "entry", + &run_state.run_id, + )), + }) + .await? + } + TargetPosition::Flat => { + let balances = venue.account_balances().await?; + let free_base_qty = balances + .into_iter() + .find(|balance| balance.asset.eq_ignore_ascii_case(&rules.base_asset)) + .map(|balance| balance.free) + .unwrap_or(Decimal::ZERO); + + let requested_qty = free_base_qty.min(run_state.position.qty); + let requested_qty = maybe_round_qty(requested_qty, rules); + + if requested_qty <= Decimal::ZERO { + warn!( + requested_qty = %requested_qty, + run_position_qty = %run_state.position.qty, + "ignoring flat target because no sellable quantity remained" + ); + return Ok(()); + } + + venue + .submit_market_order(&MarketOrderRequest { + symbol: cfg.market.symbol.clone(), + side: Side::Sell, + quantity: Some(requested_qty), + quote_order_qty: None, + new_client_order_id: Some(new_client_order_id( + "exit", + &run_state.run_id, + )), + }) + .await? + } + } + } + }; + + info!( + run_id = %run_state.run_id, + side = %order.side, + status = %order.status.as_str(), + executed_qty = %order.executed_qty, + avg_price = ?order.average_price(), + "order submitted" + ); + + self.journal_store + .append_order_event(&run_state.run_id, &order)?; + summary.submitted_orders += 1; + + match target { + TargetPosition::LongAllIn => { + let qty = order.net_base_qty_after_base_fees(&rules.base_asset); + if qty <= Decimal::ZERO { + warn!("entry order had zero executed quantity"); + return Ok(()); + } + + run_state.position = PositionState { + qty, + entry_price: order.average_price(), + entry_time_ms: order.transact_time_ms.or(Some(reference_bar.close_time_ms)), + entry_order_id: order.order_id, + }; + } + TargetPosition::Flat => { + let closed_qty = order.executed_qty.min(run_state.position.qty); + if closed_qty <= Decimal::ZERO { + warn!("exit order had zero executed quantity"); + return Ok(()); + } + + let entry_price = run_state + .position + .entry_price + .unwrap_or(reference_bar.close); + let exit_price = order.average_price().unwrap_or(reference_bar.close); + + let closed_trade = ClosedTrade { + symbol: cfg.market.symbol.clone(), + entry_time_ms: run_state + .position + .entry_time_ms + .unwrap_or(reference_bar.open_time_ms), + exit_time_ms: order + .transact_time_ms + .unwrap_or(reference_bar.close_time_ms), + entry_price, + exit_price, + qty: closed_qty, + gross_quote_pnl: (exit_price - entry_price) * closed_qty, + entry_order_id: run_state.position.entry_order_id, + exit_order_id: order.order_id, + }; + self.journal_store + .append_closed_trade(&run_state.run_id, &closed_trade)?; + summary.closed_trades += 1; + + let remaining_qty = (run_state.position.qty - closed_qty).max(Decimal::ZERO); + if remaining_qty > Decimal::ZERO { + run_state.position.qty = remaining_qty; + } else { + run_state.position = PositionState::flat(); + } + } + } + + run_state.updated_at_ms = now_utc_ms(); + run_state.status = RunStatus::Running; + run_state.last_error = None; + self.journal_store.save_run_state(run_state)?; + Ok(()) + } +} + +fn current_target(position: &PositionState) -> TargetPosition { + if position.is_open() { + TargetPosition::LongAllIn + } else { + TargetPosition::Flat + } +} + +fn filter_closed_candles(candles: Vec) -> Vec { + let now_ms = now_utc_ms(); + candles + .into_iter() + .filter(|candle| candle.close_time_ms <= now_ms) + .collect() +} + +fn maybe_round_qty(qty: Decimal, rules: &SymbolRules) -> Decimal { + if let Some(step_size) = rules.effective_market_step_size() { + round_down_to_step(qty, step_size) + } else { + qty + } +} + +fn synthetic_market_order( + rules: &SymbolRules, + run_state: &BotRunState, + target: TargetPosition, + quote_order_qty: Decimal, + reference_bar: &Candle, +) -> Result { + let side = match target { + TargetPosition::LongAllIn => Side::Buy, + TargetPosition::Flat => Side::Sell, + }; + + let (requested_qty, requested_quote_qty, executed_qty, cumulative_quote_qty) = match target { + TargetPosition::LongAllIn => { + if reference_bar.close <= Decimal::ZERO { + return Err(EngineError::InvalidState( + "cannot simulate market buy with non-positive reference price".to_string(), + )); + } + let raw_qty = quote_order_qty / reference_bar.close; + let qty = maybe_round_qty(raw_qty, rules); + (None, Some(quote_order_qty), qty, qty * reference_bar.close) + } + TargetPosition::Flat => { + let qty = maybe_round_qty(run_state.position.qty, rules); + (Some(qty), None, qty, qty * reference_bar.close) + } + }; + + Ok(ExchangeOrder { + symbol: run_state.market.symbol.clone(), + side, + order_type: "MARKET".to_string(), + status: crate::OrderStatus::Filled, + order_id: None, + client_order_id: Some(new_client_order_id("dry", &run_state.run_id)), + requested_qty, + requested_quote_qty, + executed_qty, + cumulative_quote_qty, + avg_price: Some(reference_bar.close), + transact_time_ms: Some(reference_bar.close_time_ms), + fills: Vec::new(), + raw: serde_json::json!({ + "execution_mode": "dry_run", + "reference_open_time_ms": reference_bar.open_time_ms, + "reference_close_time_ms": reference_bar.close_time_ms + }), + }) +} + +fn sanitize_client_order_id_fragment(input: &str, max_len: usize) -> String { + let out: String = input + .chars() + .filter(|c| c.is_ascii_alphanumeric() || *c == '-' || *c == '_') + .take(max_len) + .collect(); + + if out.is_empty() { + "run".to_string() + } else { + out + } +} + +fn new_client_order_id(tag: &str, run_id: &str) -> String { + let tag = sanitize_client_order_id_fragment(tag, 5); + let prefix = sanitize_client_order_id_fragment(run_id, 8); + + let nonce = Uuid::new_v4().simple().to_string(); + let nonce = &nonce[..8]; + + // keep timestamp short so total length stays <= 36 + let ts = (now_utc_ms() % 100_000_000).to_string(); + + // qf---- + let id = format!("qf-{tag}-{prefix}-{ts}-{nonce}"); + debug_assert!(id.len() <= 36); + id +} + +#[derive(Debug)] +struct LiveStrategyContext { + market: MarketId, + now_ms: TimestampMs, + cash: Decimal, + position_qty: Decimal, + desired_target: Option, +} + +impl LiveStrategyContext { + fn new(market: MarketId, position_qty: Decimal) -> Self { + Self { + market, + now_ms: now_utc_ms(), + cash: Decimal::ZERO, + position_qty, + desired_target: None, + } + } +} + +impl StrategyContext for LiveStrategyContext { + fn market(&self) -> &MarketId { + &self.market + } + + fn now_ms(&self) -> TimestampMs { + self.now_ms + } + + fn cash(&self) -> Decimal { + self.cash + } + + fn position_qty(&self) -> Decimal { + self.position_qty + } + + fn set_target_position(&mut self, target: TargetPosition) { + self.desired_target = Some(target); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ExchangeId, Interval, Symbol}; + use std::str::FromStr; + + fn market() -> MarketId { + MarketId::new( + ExchangeId::BinanceSpot, + Symbol::new("BTCUSDT").expect("symbol"), + Interval::M1, + ) + } + + fn rules() -> SymbolRules { + SymbolRules { + symbol: Symbol::new("BTCUSDT").expect("symbol"), + base_asset: "BTC".to_string(), + quote_asset: "USDT".to_string(), + min_qty: Some(Decimal::from_str("0.001").expect("decimal")), + max_qty: None, + step_size: Some(Decimal::from_str("0.001").expect("decimal")), + market_min_qty: Some(Decimal::from_str("0.001").expect("decimal")), + market_max_qty: None, + market_step_size: Some(Decimal::from_str("0.001").expect("decimal")), + min_notional: Some(Decimal::from(10)), + tick_size: Some(Decimal::from_str("0.01").expect("decimal")), + } + } + + fn reference_bar() -> Candle { + Candle { + open_time_ms: 0, + close_time_ms: 59_999, + open: Decimal::from(10_000), + high: Decimal::from(10_000), + low: Decimal::from(10_000), + close: Decimal::from(10_000), + volume: Decimal::ONE, + trades: Some(1), + } + } + + fn run_state() -> BotRunState { + BotRunState { + run_id: "run-1".to_string(), + market: market(), + strategy_name: "sma_cross".to_string(), + strategy_config: serde_json::json!({"kind":"sma_cross","fast":20,"slow":50}), + status: RunStatus::Running, + last_processed_open_time_ms: None, + started_at_ms: 0, + updated_at_ms: 0, + stopped_at_ms: None, + last_error: None, + position: PositionState { + qty: Decimal::from_str("0.0254").expect("decimal"), + entry_price: Some(Decimal::from(9_900)), + entry_time_ms: Some(0), + entry_order_id: Some(7), + }, + } + } + + #[test] + fn synthetic_buy_uses_quote_order_qty_and_rounds_down() { + let order = synthetic_market_order( + &rules(), + &run_state(), + TargetPosition::LongAllIn, + Decimal::from(123), + &reference_bar(), + ) + .expect("order"); + + assert_eq!(order.side, Side::Buy); + assert_eq!(order.requested_quote_qty, Some(Decimal::from(123))); + assert_eq!( + order.executed_qty, + Decimal::from_str("0.012").expect("decimal") + ); + } + + #[test] + fn synthetic_sell_uses_position_qty_and_rounds_down() { + let order = synthetic_market_order( + &rules(), + &run_state(), + TargetPosition::Flat, + Decimal::from(123), + &reference_bar(), + ) + .expect("order"); + + assert_eq!(order.side, Side::Sell); + assert_eq!( + order.requested_qty, + Some(Decimal::from_str("0.025").expect("decimal")) + ); + assert_eq!( + order.executed_qty, + Decimal::from_str("0.025").expect("decimal") + ); + } +} diff --git a/src/main.rs b/src/main.rs index 454c4ba..bfd344f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,23 +1,27 @@ -use anyhow::{Context, Result, bail}; -use clap::{Parser, Subcommand}; +use anyhow::{Context, Result, anyhow, bail}; +use clap::{Args, Parser, Subcommand, ValueEnum}; +use quantforge::SqliteStore; +use quantforge::{ + BacktestConfig, BacktestEngine, DataSyncConfig, DataSyncEngine, LiveTradeConfig, + LiveTradeEngine, +}; +use quantforge::{BinanceCredentials, BinanceSpotClient}; +use quantforge::{ + BuiltInStrategyConfig, CandleQuery, CandleStore, ClosedTrade, ExchangeId, ExecutionMode, + Interval, MarketDataSource, MarketId, PositionState, RunJournalStore, Side, Symbol, + TradingVenue, ms_to_rfc3339, now_utc_ms, parse_rfc3339_to_ms, round_down_to_step, + validate_candles, +}; use rust_decimal::Decimal; -use std::path::PathBuf; -use tracing::{info, warn}; +use std::{path::PathBuf, time::Duration}; use tracing_subscriber::EnvFilter; use url::Url; -use quantforge::sdk::strategies::SmaCrossStrategy; -use quantforge::{ - BacktestConfig, BacktestEngine, BinanceSpotClient, CandleQuery, CandleStore, ExchangeId, - Interval, KlineRequest, MarketDataSource, MarketId, SqliteCandleStore, Strategy, Symbol, - parse_rfc3339_to_ms, validate_candles, -}; - #[derive(Parser, Debug)] #[command( name = "quantforge", version, - about = "Deterministic CLI-first market data ingestion and backtesting framework." + about = "CLI-first market data ingestion, research, and controlled live trading framework." )] struct Cli { /// SQLite database path. @@ -29,7 +33,16 @@ struct Cli { )] db: PathBuf, - /// Log level (trace, debug, info, warn, error). + /// Binance API base URL. Use https://testnet.binance.vision/ for Spot testnet. + #[arg( + long, + global = true, + env = "QF_BINANCE_BASE_URL", + default_value = "https://api.binance.com/" + )] + binance_base_url: String, + + /// Log filter (trace, debug, info, warn, error or a full tracing filter expression). #[arg(long, global = true, env = "QF_LOG", default_value = "info")] log_level: String, @@ -39,55 +52,243 @@ struct Cli { #[derive(Subcommand, Debug)] enum Command { - /// Download Binance Spot OHLCV candles into SQLite. - Download { - #[arg(long)] - symbol: String, - #[arg(long, default_value = "1m")] - interval: String, - #[arg(long)] - start: String, - #[arg(long)] - end: String, - #[arg(long, default_value = "https://api.binance.com")] - base_url: String, - #[arg(long, default_value_t = 1000)] - limit: u16, + /// Historical and incremental data ingestion into SQLite. + Data { + #[command(subcommand)] + command: DataCommand, }, - /// Validate stored candles for duplicates, gaps, ordering, and OHLC sanity. - Validate { - #[arg(long)] - symbol: String, - #[arg(long, default_value = "1m")] - interval: String, - #[arg(long)] - start: Option, - #[arg(long)] - end: Option, + /// Deterministic backtest using locally stored candles. + Backtest(BacktestArgs), + + /// Run or manually close the strategy bot. + Trade { + #[command(subcommand)] + command: TradeCommand, }, - /// Backtest the built-in SMA crossover example strategy. - Backtest { - #[arg(long)] - symbol: String, - #[arg(long, default_value = "1m")] - interval: String, - #[arg(long)] - start: Option, - #[arg(long)] - end: Option, - #[arg(long, default_value_t = 20)] - fast: usize, - #[arg(long, default_value_t = 50)] - slow: usize, - #[arg(long, default_value = "10000")] - cash: String, - #[arg(long, default_value = "10")] - fee_bps: String, + /// Observe Binance state and manage orders manually when needed. + Monitor { + #[command(subcommand)] + command: MonitorCommand, }, } +#[derive(Subcommand, Debug)] +enum DataCommand { + /// Sync candles from Binance into SQLite. + Sync(DataSyncArgs), + + /// Validate stored candles for duplicates, gaps, ordering, and OHLC sanity. + Validate(DataValidateArgs), +} + +#[derive(Args, Debug)] +struct DataSyncArgs { + #[arg(long)] + symbol: String, + #[arg(long, default_value = "1m")] + interval: String, + /// RFC3339 start time. Omit to begin syncing from the current time. + #[arg(long)] + start: Option, + /// RFC3339 end time. Omit to keep syncing indefinitely. + #[arg(long)] + end: Option, + #[arg(long, default_value_t = 1000)] + limit: u16, + /// When --end is set, keep polling until that end boundary is reached. + #[arg(long, default_value_t = false)] + follow: bool, + #[arg(long, default_value_t = 5)] + poll_secs: u64, + #[arg(long)] + max_loops: Option, +} + +#[derive(Args, Debug)] +struct DataValidateArgs { + #[arg(long)] + symbol: String, + #[arg(long, default_value = "1m")] + interval: String, + #[arg(long)] + start: Option, + #[arg(long)] + end: Option, +} + +#[derive(Args, Debug)] +struct BacktestArgs { + #[arg(long)] + symbol: String, + #[arg(long, default_value = "1m")] + interval: String, + #[arg(long)] + start: Option, + #[arg(long)] + end: Option, + #[arg(long, default_value_t = 20)] + fast: usize, + #[arg(long, default_value_t = 50)] + slow: usize, + #[arg(long, default_value = "10000")] + cash: String, + #[arg(long, default_value = "10")] + fee_bps: String, +} + +#[derive(Subcommand, Debug)] +enum TradeCommand { + /// Run the polling strategy bot against SQLite-backed live candles. + Run(TradeRunArgs), + + /// Close the bot-managed position with a market sell. + Close(TradeCloseArgs), +} + +#[derive(Args, Debug)] +struct TradeRunArgs { + #[arg(long)] + symbol: String, + #[arg(long, default_value = "1m")] + interval: String, + #[arg(long, default_value_t = 20)] + fast: usize, + #[arg(long, default_value_t = 50)] + slow: usize, + #[arg(long, default_value = "100")] + quote_order_qty: String, + #[arg(long, value_enum, default_value_t = CliExecutionMode::DryRun)] + mode: CliExecutionMode, + #[arg(long, default_value_t = 5)] + poll_secs: u64, + #[arg(long, default_value_t = 300)] + bootstrap_bars: usize, + #[arg(long, default_value_t = false)] + bootstrap_enter: bool, + #[arg(long, default_value_t = 1000)] + limit: u16, + #[arg(long)] + run_id: Option, + #[arg(long)] + max_loops: Option, +} + +#[derive(Args, Debug)] +struct TradeCloseArgs { + #[arg(long)] + symbol: String, + #[arg(long, default_value = "1m")] + interval: String, + #[arg(long, default_value = "sma_cross")] + strategy_name: String, + #[arg(long)] + run_id: Option, + #[arg(long, default_value_t = false)] + yes: bool, +} + +#[derive(Subcommand, Debug)] +enum MonitorCommand { + /// Print balances, latest run state, open orders, and recent trades. + Status(MonitorStatusArgs), + + /// Poll status repeatedly. + Watch(MonitorWatchArgs), + + /// List current open orders on the symbol. + Orders(MonitorOrdersArgs), + + /// List recent Binance trades on the symbol. + Trades(MonitorTradesArgs), + + /// Cancel a specific order manually. + CancelOrder(MonitorCancelOrderArgs), + + /// Close the current free base-asset balance for the symbol. + ClosePosition(MonitorClosePositionArgs), +} + +#[derive(Args, Debug)] +struct MonitorStatusArgs { + #[arg(long)] + symbol: String, + #[arg(long, default_value = "1m")] + interval: String, + #[arg(long, default_value = "sma_cross")] + strategy_name: String, + #[arg(long, default_value_t = 10)] + recent_trades: usize, +} + +#[derive(Args, Debug)] +struct MonitorWatchArgs { + #[arg(long)] + symbol: String, + #[arg(long, default_value = "1m")] + interval: String, + #[arg(long, default_value = "sma_cross")] + strategy_name: String, + #[arg(long, default_value_t = 10)] + recent_trades: usize, + #[arg(long, default_value_t = 5)] + poll_secs: u64, + #[arg(long)] + max_loops: Option, +} + +#[derive(Args, Debug)] +struct MonitorOrdersArgs { + #[arg(long)] + symbol: String, +} + +#[derive(Args, Debug)] +struct MonitorTradesArgs { + #[arg(long)] + symbol: String, + #[arg(long, default_value_t = 20)] + limit: usize, +} + +#[derive(Args, Debug)] +struct MonitorCancelOrderArgs { + #[arg(long)] + symbol: String, + #[arg(long)] + order_id: Option, + #[arg(long)] + client_order_id: Option, + #[arg(long, default_value_t = false)] + yes: bool, +} + +#[derive(Args, Debug)] +struct MonitorClosePositionArgs { + #[arg(long)] + symbol: String, + #[arg(long, default_value_t = false)] + yes: bool, +} + +#[derive(Copy, Clone, Debug, Eq, PartialEq, ValueEnum)] +enum CliExecutionMode { + #[value(name = "dry-run")] + DryRun, + #[value(name = "live")] + Live, +} + +impl From for ExecutionMode { + fn from(value: CliExecutionMode) -> Self { + match value { + CliExecutionMode::DryRun => ExecutionMode::DryRun, + CliExecutionMode::Live => ExecutionMode::Live, + } + } +} + #[tokio::main] async fn main() -> Result<()> { let cli = Cli::parse(); @@ -98,126 +299,51 @@ async fn main() -> Result<()> { .with_context(|| format!("failed to create parent directory {}", parent.display()))?; } - let store = SqliteCandleStore::new(&cli.db); - store.init().context("failed to initialize sqlite store")?; + let store = SqliteStore::new(&cli.db); + CandleStore::init(&store).context("failed to initialize sqlite store")?; - match cli.command { - Command::Download { - symbol, - interval, - start, - end, - base_url, - limit, - } => { - let market = parse_market(symbol, interval)?; - let start_ms = parse_rfc3339_to_ms(&start).context("failed to parse --start")?; - let end_ms = parse_rfc3339_to_ms(&end).context("failed to parse --end")?; - if end_ms < start_ms { - bail!("--end must be greater than or equal to --start"); - } + let base_url = + Url::parse(&cli.binance_base_url).context("failed to parse --binance-base-url")?; + let public_client = BinanceSpotClient::new(base_url.clone()); + let private_client = BinanceCredentials::from_env() + .map(|credentials| BinanceSpotClient::new(base_url.clone()).with_credentials(credentials)); - let base_url = Url::parse(&base_url).context("failed to parse --base-url")?; - let client = BinanceSpotClient::new(base_url); - download_klines(&client, &store, &market, start_ms, end_ms, limit).await?; - } - Command::Validate { - symbol, - interval, - start, - end, - } => { - let market = parse_market(symbol, interval)?; - let candles = store.load_candles( - &market, - CandleQuery { - start_time_ms: start - .as_deref() - .map(parse_rfc3339_to_ms) - .transpose() - .context("failed to parse --start")?, - end_time_ms: end - .as_deref() - .map(parse_rfc3339_to_ms) - .transpose() - .context("failed to parse --end")?, - limit: None, - }, - )?; - - let report = validate_candles(&market, &candles); - println!( - "market: {} {} {}", - report.market.exchange, report.market.symbol, report.market.interval - ); - println!("candles: {}", report.candle_count); - println!("issues: {}", report.issues.len()); - for (index, issue) in report.issues.iter().take(20).enumerate() { - println!(" {:02}: {:?}", index, issue); + match cli.command { + Command::Data { command } => match command { + DataCommand::Sync(args) => handle_data_sync(&store, &public_client, args).await?, + DataCommand::Validate(args) => handle_data_validate(&store, args)?, + }, + Command::Backtest(args) => handle_backtest(&store, args)?, + Command::Trade { command } => match command { + TradeCommand::Run(args) => { + handle_trade_run(&store, &public_client, private_client.as_ref(), args).await? } - if report.issues.len() > 20 { - println!(" ... ({} more)", report.issues.len() - 20); + TradeCommand::Close(args) => { + let private_client = private_client.as_ref().ok_or_else(|| { + anyhow!("trade close requires QF_BINANCE_API_KEY and QF_BINANCE_API_SECRET") + })?; + handle_trade_close(&store, private_client, args).await? } - - if report.is_ok() { - info!("validation passed without issues"); - } else { - warn!(issues = report.issues.len(), "validation reported issues"); - } - } - Command::Backtest { - symbol, - interval, - start, - end, - fast, - slow, - cash, - fee_bps, - } => { - let market = parse_market(symbol, interval)?; - let candles = store.load_candles( - &market, - CandleQuery { - start_time_ms: start - .as_deref() - .map(parse_rfc3339_to_ms) - .transpose() - .context("failed to parse --start")?, - end_time_ms: end - .as_deref() - .map(parse_rfc3339_to_ms) - .transpose() - .context("failed to parse --end")?, - limit: None, - }, - )?; - - let mut strategy = SmaCrossStrategy::new(fast, slow)?; - let engine = BacktestEngine::new(BacktestConfig { - initial_cash: cash.parse::().context("failed to parse --cash")?, - fee_bps: fee_bps - .parse::() - .context("failed to parse --fee-bps")?, - close_out_at_end: true, - }); - let result = engine.run(&market, &candles, &mut strategy)?; - - println!("strategy: {}", strategy.name()); - println!("final_equity: {}", result.final_equity); - println!("total_return_pct: {}", result.total_return_pct); - println!("max_drawdown_pct: {}", result.max_drawdown_pct); - println!("trade_count: {}", result.trade_count); - for trade in result.trades.iter().rev().take(5).rev() { - println!( - "trade: entry={} @ {} exit={} @ {} qty={} pnl={}", - trade.entry_time_ms, - trade.entry_price, - trade.exit_time_ms, - trade.exit_price, - trade.qty, - trade.pnl - ); + }, + Command::Monitor { command } => { + let private_client = private_client.as_ref().ok_or_else(|| { + anyhow!("monitor commands require QF_BINANCE_API_KEY and QF_BINANCE_API_SECRET") + })?; + match command { + MonitorCommand::Status(args) => { + handle_monitor_status(&store, private_client, args).await? + } + MonitorCommand::Watch(args) => { + handle_monitor_watch(&store, private_client, args).await? + } + MonitorCommand::Orders(args) => handle_monitor_orders(private_client, args).await?, + MonitorCommand::Trades(args) => handle_monitor_trades(private_client, args).await?, + MonitorCommand::CancelOrder(args) => { + handle_monitor_cancel_order(private_client, args).await? + } + MonitorCommand::ClosePosition(args) => { + handle_monitor_close_position(private_client, args).await? + } } } } @@ -243,56 +369,523 @@ fn parse_market(symbol: String, interval: String) -> Result { Ok(MarketId::new(ExchangeId::BinanceSpot, symbol, interval)) } -async fn download_klines( - source: &dyn MarketDataSource, - store: &dyn CandleStore, - market: &MarketId, - start_ms: i64, - end_ms: i64, - limit: u16, +fn strategy_config(fast: usize, slow: usize) -> BuiltInStrategyConfig { + BuiltInStrategyConfig::SmaCross { fast, slow } +} + +async fn handle_data_sync( + store: &SqliteStore, + client: &BinanceSpotClient, + args: DataSyncArgs, ) -> Result<()> { - let step_ms = market.interval.step_ms(); - let mut cursor = start_ms; - let mut total = 0usize; - - while cursor <= end_ms { - let batch = source - .fetch_klines(&KlineRequest { - symbol: market.symbol.clone(), - interval: market.interval, - start_time_ms: Some(cursor), - end_time_ms: Some(end_ms), - limit: Some(limit), - }) - .await - .context("failed to fetch klines")?; - - if batch.is_empty() { - info!(cursor, "no more candles returned; stopping"); - break; + let market = parse_market(args.symbol, args.interval)?; + let engine = DataSyncEngine::new(client, store); + let summary = engine + .run(&DataSyncConfig { + market, + start_time_ms: args + .start + .as_deref() + .map(parse_rfc3339_to_ms) + .transpose() + .context("failed to parse --start")?, + end_time_ms: args + .end + .as_deref() + .map(parse_rfc3339_to_ms) + .transpose() + .context("failed to parse --end")?, + batch_limit: args.limit, + follow: args.follow, + poll_interval: Duration::from_secs(args.poll_secs), + max_loops: args.max_loops, + }) + .await?; + + println!("iterations: {}", summary.iterations); + println!("written: {}", summary.written); + println!( + "last_open_time: {}", + summary + .last_open_time_ms + .map(ms_to_rfc3339) + .unwrap_or_else(|| "none".to_string()) + ); + Ok(()) +} + +fn handle_data_validate(store: &SqliteStore, args: DataValidateArgs) -> Result<()> { + let market = parse_market(args.symbol, args.interval)?; + let candles = store.load_candles( + &market, + CandleQuery { + start_time_ms: args + .start + .as_deref() + .map(parse_rfc3339_to_ms) + .transpose() + .context("failed to parse --start")?, + end_time_ms: args + .end + .as_deref() + .map(parse_rfc3339_to_ms) + .transpose() + .context("failed to parse --end")?, + limit: None, + }, + )?; + + let report = validate_candles(&market, &candles); + println!( + "market: {} {} {}", + report.market.exchange, report.market.symbol, report.market.interval + ); + println!("candles: {}", report.candle_count); + println!("issues: {}", report.issues.len()); + for (index, issue) in report.issues.iter().take(20).enumerate() { + println!(" {:02}: {:?}", index, issue); + } + if report.issues.len() > 20 { + println!(" ... ({} more)", report.issues.len() - 20); + } + Ok(()) +} + +fn handle_backtest(store: &SqliteStore, args: BacktestArgs) -> Result<()> { + let market = parse_market(args.symbol, args.interval)?; + let candles = store.load_candles( + &market, + CandleQuery { + start_time_ms: args + .start + .as_deref() + .map(parse_rfc3339_to_ms) + .transpose() + .context("failed to parse --start")?, + end_time_ms: args + .end + .as_deref() + .map(parse_rfc3339_to_ms) + .transpose() + .context("failed to parse --end")?, + limit: None, + }, + )?; + + let mut strategy = strategy_config(args.fast, args.slow) + .build() + .context("failed to build strategy")?; + let engine = BacktestEngine::new(BacktestConfig { + initial_cash: args + .cash + .parse::() + .context("failed to parse --cash")?, + fee_bps: args + .fee_bps + .parse::() + .context("failed to parse --fee-bps")?, + close_out_at_end: true, + }); + + let result = engine.run(&market, &candles, strategy.as_mut())?; + println!("strategy: {}", strategy.name()); + println!("final_equity: {}", result.final_equity); + println!("total_return_pct: {}", result.total_return_pct); + println!("max_drawdown_pct: {}", result.max_drawdown_pct); + println!("trade_count: {}", result.trade_count); + for trade in result.trades.iter().rev().take(5).rev() { + println!( + "trade: entry={} @ {} exit={} @ {} qty={} gross_pnl={}", + trade.entry_time_ms, + trade.entry_price, + trade.exit_time_ms, + trade.exit_price, + trade.qty, + trade.gross_quote_pnl + ); + } + Ok(()) +} + +async fn handle_trade_run( + store: &SqliteStore, + public_client: &BinanceSpotClient, + private_client: Option<&BinanceSpotClient>, + args: TradeRunArgs, +) -> Result<()> { + let market = parse_market(args.symbol, args.interval)?; + let engine = LiveTradeEngine::new( + public_client, + store, + store, + if matches!(args.mode, CliExecutionMode::Live) { + Some( + private_client + .ok_or_else(|| anyhow!("trade run --mode live requires Binance credentials"))? + as &dyn TradingVenue, + ) + } else { + None + }, + ); + + let summary = engine + .run(&LiveTradeConfig { + market, + strategy: strategy_config(args.fast, args.slow), + execution_mode: args.mode.into(), + quote_order_qty: args + .quote_order_qty + .parse::() + .context("failed to parse --quote-order-qty")?, + poll_interval: Duration::from_secs(args.poll_secs), + bootstrap_bars: args.bootstrap_bars, + bootstrap_enter: args.bootstrap_enter, + batch_limit: args.limit, + run_id: args.run_id, + max_loops: args.max_loops, + }) + .await?; + + println!("run_id: {}", summary.run_id); + println!("processed_bars: {}", summary.processed_bars); + println!("submitted_orders: {}", summary.submitted_orders); + println!("closed_trades: {}", summary.closed_trades); + println!( + "last_processed_open_time: {}", + summary + .last_processed_open_time_ms + .map(ms_to_rfc3339) + .unwrap_or_else(|| "none".to_string()) + ); + Ok(()) +} + +async fn handle_trade_close( + store: &SqliteStore, + private_client: &BinanceSpotClient, + args: TradeCloseArgs, +) -> Result<()> { + let market = parse_market(args.symbol, args.interval)?; + let rules = private_client.fetch_symbol_rules(&market.symbol).await?; + let mut run_state = if let Some(run_id) = args.run_id { + store + .load_run_state(&run_id)? + .ok_or_else(|| anyhow!("no run found for run_id={run_id}"))? + } else { + store + .latest_run_for_market(&market, &args.strategy_name)? + .ok_or_else(|| { + anyhow!( + "no run found for market={} strategy={}", + market.symbol, + args.strategy_name + ) + })? + }; + + let balances = private_client.account_balances().await?; + let free_base_qty = balances + .iter() + .find(|balance| balance.asset.eq_ignore_ascii_case(&rules.base_asset)) + .map(|balance| balance.free) + .unwrap_or(Decimal::ZERO); + + let qty = round_quantity_for_rules(free_base_qty.min(run_state.position.qty), &rules); + println!("run_id: {}", run_state.run_id); + println!("sell_qty: {}", qty); + if !args.yes { + println!("No order sent. Re-run with --yes to execute the market sell."); + return Ok(()); + } + if qty <= Decimal::ZERO { + bail!("no sellable quantity available for {}", rules.base_asset); + } + + let order = private_client + .submit_market_order(&quantforge::MarketOrderRequest { + symbol: market.symbol.clone(), + side: Side::Sell, + quantity: Some(qty), + quote_order_qty: None, + new_client_order_id: Some(format!("manual-close-{}", now_utc_ms())), + }) + .await?; + + store.append_order_event(&run_state.run_id, &order)?; + if run_state.position.is_open() && order.executed_qty > Decimal::ZERO { + let entry_price = run_state.position.entry_price.unwrap_or(Decimal::ZERO); + let exit_price = order.average_price().unwrap_or(entry_price); + let closed_qty = order.executed_qty.min(run_state.position.qty); + let trade = ClosedTrade { + symbol: market.symbol.clone(), + entry_time_ms: run_state + .position + .entry_time_ms + .unwrap_or(order.transact_time_ms.unwrap_or(now_utc_ms())), + exit_time_ms: order.transact_time_ms.unwrap_or(now_utc_ms()), + entry_price, + exit_price, + qty: closed_qty, + gross_quote_pnl: (exit_price - entry_price) * closed_qty, + entry_order_id: run_state.position.entry_order_id, + exit_order_id: order.order_id, + }; + store.append_closed_trade(&run_state.run_id, &trade)?; + + let remaining_qty = (run_state.position.qty - closed_qty).max(Decimal::ZERO); + run_state.updated_at_ms = now_utc_ms(); + run_state.last_error = None; + + if remaining_qty > Decimal::ZERO { + run_state.position.qty = remaining_qty; + run_state.status = quantforge::RunStatus::Running; + run_state.stopped_at_ms = None; + } else { + run_state.position = PositionState::flat(); + run_state.status = quantforge::RunStatus::Stopped; + run_state.stopped_at_ms = Some(run_state.updated_at_ms); } - total += store - .upsert_candles(market, &batch) - .context("failed to upsert candles")?; - - let last_open_time_ms = batch - .last() - .map(|candle| candle.open_time_ms) - .context("empty batch after non-empty check")?; - let next_cursor = last_open_time_ms + step_ms; - if next_cursor <= cursor { - warn!( - cursor, - next_cursor, "cursor did not advance; stopping to avoid infinite loop" - ); + store.save_run_state(&run_state)?; + } + + print_order(&order); + Ok(()) +} + +async fn handle_monitor_status( + store: &SqliteStore, + private_client: &BinanceSpotClient, + args: MonitorStatusArgs, +) -> Result<()> { + let market = parse_market(args.symbol, args.interval)?; + let rules = private_client.fetch_symbol_rules(&market.symbol).await?; + let balances = private_client.account_balances().await?; + let open_orders = private_client.open_orders(Some(&market.symbol)).await?; + let trades = private_client + .recent_trades(&market.symbol, args.recent_trades) + .await?; + let run = store.latest_run_for_market(&market, &args.strategy_name)?; + + println!("symbol: {}", market.symbol); + println!("base_asset: {}", rules.base_asset); + println!("quote_asset: {}", rules.quote_asset); + println!("time_utc_ms: {}", now_utc_ms()); + + if let Some(run) = run { + println!("latest_run_id: {}", run.run_id); + println!("run_status: {}", run.status.as_str()); + println!( + "last_processed_open_time: {}", + run.last_processed_open_time_ms + .map(ms_to_rfc3339) + .unwrap_or_else(|| "none".to_string()) + ); + println!("position_qty: {}", run.position.qty); + println!( + "entry_price: {}", + run.position + .entry_price + .map(|value| value.to_string()) + .unwrap_or_else(|| "none".to_string()) + ); + } else { + println!("latest_run_id: none"); + } + + println!("balances:"); + for balance in balances.iter().filter(|balance| { + balance.asset.eq_ignore_ascii_case(&rules.base_asset) + || balance.asset.eq_ignore_ascii_case(&rules.quote_asset) + }) { + println!( + " {} free={} locked={}", + balance.asset, balance.free, balance.locked + ); + } + + println!("open_orders: {}", open_orders.len()); + for order in open_orders.iter().take(10) { + println!( + " id={:?} side={} status={} qty={} avg_price={}", + order.order_id, + order.side, + order.status.as_str(), + order.executed_qty, + order + .average_price() + .map(|value| value.to_string()) + .unwrap_or_else(|| "n/a".to_string()) + ); + } + + println!("recent_trades: {}", trades.len()); + for trade in trades.iter().take(10) { + println!( + " id={} order_id={} side={} qty={} price={} commission={} {} time={}", + trade.trade_id, + trade.order_id, + trade.side, + trade.qty, + trade.price, + trade.commission, + trade.commission_asset.as_deref().unwrap_or("?"), + ms_to_rfc3339(trade.time_ms), + ); + } + + Ok(()) +} + +async fn handle_monitor_watch( + store: &SqliteStore, + private_client: &BinanceSpotClient, + args: MonitorWatchArgs, +) -> Result<()> { + let mut loops = 0usize; + loop { + println!("---"); + handle_monitor_status( + store, + private_client, + MonitorStatusArgs { + symbol: args.symbol.clone(), + interval: args.interval.clone(), + strategy_name: args.strategy_name.clone(), + recent_trades: args.recent_trades, + }, + ) + .await?; + + loops += 1; + if args.max_loops.map(|max| loops >= max).unwrap_or(false) { break; } - cursor = next_cursor; - info!(written = batch.len(), total, cursor, "download progress"); + tokio::select! { + _ = tokio::time::sleep(Duration::from_secs(args.poll_secs)) => {} + _ = tokio::signal::ctrl_c() => break, + } + } + + Ok(()) +} + +async fn handle_monitor_orders( + private_client: &BinanceSpotClient, + args: MonitorOrdersArgs, +) -> Result<()> { + let symbol = Symbol::new(args.symbol)?; + let orders = private_client.open_orders(Some(&symbol)).await?; + println!("open_orders: {}", orders.len()); + for order in orders { + print_order(&order); } + Ok(()) +} - info!(total, "download completed"); +async fn handle_monitor_trades( + private_client: &BinanceSpotClient, + args: MonitorTradesArgs, +) -> Result<()> { + let symbol = Symbol::new(args.symbol)?; + let trades = private_client.recent_trades(&symbol, args.limit).await?; + println!("recent_trades: {}", trades.len()); + for trade in trades { + println!( + "trade_id={} order_id={} side={} qty={} price={} quote_qty={} commission={} {} time={}", + trade.trade_id, + trade.order_id, + trade.side, + trade.qty, + trade.price, + trade.quote_qty, + trade.commission, + trade.commission_asset.as_deref().unwrap_or("?"), + ms_to_rfc3339(trade.time_ms), + ); + } Ok(()) } + +async fn handle_monitor_cancel_order( + private_client: &BinanceSpotClient, + args: MonitorCancelOrderArgs, +) -> Result<()> { + let symbol = Symbol::new(args.symbol)?; + if !args.yes { + println!("No order canceled. Re-run with --yes to execute the cancel."); + return Ok(()); + } + let order = private_client + .cancel_order(&quantforge::CancelOrderRequest { + symbol, + order_id: args.order_id, + client_order_id: args.client_order_id, + }) + .await?; + print_order(&order); + Ok(()) +} + +async fn handle_monitor_close_position( + private_client: &BinanceSpotClient, + args: MonitorClosePositionArgs, +) -> Result<()> { + let symbol = Symbol::new(args.symbol)?; + let rules = private_client.fetch_symbol_rules(&symbol).await?; + let balances = private_client.account_balances().await?; + let free_base_qty = balances + .iter() + .find(|balance| balance.asset.eq_ignore_ascii_case(&rules.base_asset)) + .map(|balance| balance.free) + .unwrap_or(Decimal::ZERO); + let qty = round_quantity_for_rules(free_base_qty, &rules); + + println!("base_asset: {}", rules.base_asset); + println!("sell_qty: {}", qty); + if !args.yes { + println!("No order sent. Re-run with --yes to execute the market sell."); + return Ok(()); + } + if qty <= Decimal::ZERO { + bail!("no sellable quantity available for {}", rules.base_asset); + } + + let order = private_client + .submit_market_order(&quantforge::MarketOrderRequest { + symbol, + side: Side::Sell, + quantity: Some(qty), + quote_order_qty: None, + new_client_order_id: Some(format!("manual-close-{}", now_utc_ms())), + }) + .await?; + print_order(&order); + Ok(()) +} + +fn round_quantity_for_rules(qty: Decimal, rules: &quantforge::SymbolRules) -> Decimal { + match rules.effective_market_step_size() { + Some(step_size) => round_down_to_step(qty, step_size), + None => qty, + } +} + +fn print_order(order: &quantforge::ExchangeOrder) { + println!( + "order: id={:?} client_id={:?} symbol={} side={} status={} executed_qty={} cumulative_quote_qty={} avg_price={}", + order.order_id, + order.client_order_id, + order.symbol, + order.side, + order.status.as_str(), + order.executed_qty, + order.cumulative_quote_qty, + order + .average_price() + .map(|value| value.to_string()) + .unwrap_or_else(|| "n/a".to_string()) + ); +} diff --git a/src/model.rs b/src/model.rs index a9cbf25..3660e8d 100644 --- a/src/model.rs +++ b/src/model.rs @@ -2,6 +2,7 @@ use rust_decimal::Decimal; use serde::{Deserialize, Serialize}; use std::{collections::HashSet, fmt, str::FromStr}; use thiserror::Error; +use time::OffsetDateTime; pub type TimestampMs = i64; @@ -9,12 +10,12 @@ pub type TimestampMs = i64; pub struct Symbol(String); impl Symbol { - pub fn new(s: impl Into) -> Result { - let s = s.into().trim().to_string(); - if s.is_empty() { + pub fn new(value: impl Into) -> Result { + let value = value.into().trim().to_string(); + if value.is_empty() { return Err(ModelError::InvalidSymbol("empty".to_string())); } - Ok(Self(s.to_ascii_uppercase())) + Ok(Self(value.to_ascii_uppercase())) } pub fn as_str(&self) -> &str { @@ -24,7 +25,7 @@ impl Symbol { impl fmt::Display for Symbol { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.write_str(&self.0) + f.write_str(self.as_str()) } } @@ -57,28 +58,61 @@ impl fmt::Display for ExchangeId { #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] pub enum Interval { + S1, M1, + M3, M5, + M15, + M30, H1, + H2, + H4, + H6, + H8, + H12, D1, + D3, + W1, } impl Interval { pub fn as_str(self) -> &'static str { match self { + Self::S1 => "1s", Self::M1 => "1m", + Self::M3 => "3m", Self::M5 => "5m", + Self::M15 => "15m", + Self::M30 => "30m", Self::H1 => "1h", + Self::H2 => "2h", + Self::H4 => "4h", + Self::H6 => "6h", + Self::H8 => "1m", + Self::H12 => "12h", Self::D1 => "1d", + Self::D3 => "3d", + Self::W1 => "1w", } } pub fn step_ms(self) -> i64 { match self { - Self::M1 => 60_000, - Self::M5 => 300_000, - Self::H1 => 3_600_000, - Self::D1 => 86_400_000, + Self::S1 => 1_000, + Self::M1 => 60 * 1_000, + Self::M3 => 3 * 60 * 1_000, + Self::M5 => 5 * 60 * 1_000, + Self::M15 => 15 * 60 * 1_000, + Self::M30 => 30 * 60 * 1_000, + Self::H1 => 60 * 60 * 1_000, + Self::H2 => 2 * 60 * 60 * 1_000, + Self::H4 => 4 * 60 * 60 * 1_000, + Self::H6 => 6 * 60 * 60 * 1_000, + Self::H8 => 8 * 60 * 60 * 1_000, + Self::H12 => 12 * 60 * 60 * 1_000, + Self::D1 => 24 * 60 * 60 * 1_000, + Self::D3 => 3 * 24 * 60 * 60 * 1_000, + Self::W1 => 7 * 24 * 60 * 60 * 1_000, } } } @@ -94,10 +128,21 @@ impl FromStr for Interval { fn from_str(s: &str) -> Result { match s.trim() { + "1s" => Ok(Self::S1), "1m" => Ok(Self::M1), + "3m" => Ok(Self::M3), "5m" => Ok(Self::M5), + "15m" => Ok(Self::M15), + "30m" => Ok(Self::M30), "1h" => Ok(Self::H1), + "2h" => Ok(Self::H2), + "4h" => Ok(Self::H4), + "6h" => Ok(Self::H6), + "8h" => Ok(Self::H8), + "12h" => Ok(Self::H12), "1d" => Ok(Self::D1), + "3d" => Ok(Self::D3), + "1w" => Ok(Self::W1), other => Err(ModelError::InvalidInterval(other.to_string())), } } @@ -132,6 +177,294 @@ pub struct Candle { pub trades: Option, } +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum Side { + Buy, + Sell, +} + +impl Side { + pub fn as_str(self) -> &'static str { + match self { + Self::Buy => "BUY", + Self::Sell => "SELL", + } + } + + pub fn opposite(self) -> Self { + match self { + Self::Buy => Self::Sell, + Self::Sell => Self::Buy, + } + } +} + +impl fmt::Display for Side { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +impl FromStr for Side { + type Err = ModelError; + + fn from_str(value: &str) -> Result { + match value.trim().to_ascii_uppercase().as_str() { + "BUY" => Ok(Self::Buy), + "SELL" => Ok(Self::Sell), + other => Err(ModelError::InvalidSide(other.to_string())), + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum TargetPosition { + Flat, + LongAllIn, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum ExecutionMode { + DryRun, + Live, +} + +impl ExecutionMode { + pub fn as_str(self) -> &'static str { + match self { + Self::DryRun => "dry_run", + Self::Live => "live", + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum RunStatus { + Starting, + Running, + Stopped, + Failed, +} + +impl RunStatus { + pub fn as_str(self) -> &'static str { + match self { + Self::Starting => "starting", + Self::Running => "running", + Self::Stopped => "stopped", + Self::Failed => "failed", + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum OrderStatus { + New, + PartiallyFilled, + Filled, + Canceled, + Rejected, + Expired, + PendingNew, + Unknown, +} + +impl OrderStatus { + pub fn from_exchange(value: &str) -> Self { + match value.trim().to_ascii_uppercase().as_str() { + "NEW" => Self::New, + "PARTIALLY_FILLED" => Self::PartiallyFilled, + "FILLED" => Self::Filled, + "CANCELED" => Self::Canceled, + "REJECTED" => Self::Rejected, + "EXPIRED" => Self::Expired, + "PENDING_NEW" => Self::PendingNew, + _ => Self::Unknown, + } + } + + pub fn as_str(self) -> &'static str { + match self { + Self::New => "NEW", + Self::PartiallyFilled => "PARTIALLY_FILLED", + Self::Filled => "FILLED", + Self::Canceled => "CANCELED", + Self::Rejected => "REJECTED", + Self::Expired => "EXPIRED", + Self::PendingNew => "PENDING_NEW", + Self::Unknown => "UNKNOWN", + } + } + + pub fn is_terminal(self) -> bool { + matches!( + self, + Self::Filled | Self::Canceled | Self::Rejected | Self::Expired + ) + } +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct Fill { + pub price: Decimal, + pub qty: Decimal, + pub commission: Decimal, + pub commission_asset: Option, + pub trade_id: Option, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct ExchangeOrder { + pub symbol: Symbol, + pub side: Side, + pub order_type: String, + pub status: OrderStatus, + pub order_id: Option, + pub client_order_id: Option, + pub requested_qty: Option, + pub requested_quote_qty: Option, + pub executed_qty: Decimal, + pub cumulative_quote_qty: Decimal, + pub avg_price: Option, + pub transact_time_ms: Option, + pub fills: Vec, + pub raw: serde_json::Value, +} + +impl ExchangeOrder { + pub fn average_price(&self) -> Option { + if let Some(price) = self.avg_price { + return Some(price); + } + if self.executed_qty > Decimal::ZERO { + return Some(self.cumulative_quote_qty / self.executed_qty); + } + None + } + + pub fn net_base_qty_after_base_fees(&self, base_asset: &str) -> Decimal { + let mut qty = self.executed_qty; + for fill in &self.fills { + if fill + .commission_asset + .as_deref() + .map(|asset| asset.eq_ignore_ascii_case(base_asset)) + .unwrap_or(false) + { + qty -= fill.commission; + } + } + qty.max(Decimal::ZERO) + } +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct AccountTrade { + pub symbol: Symbol, + pub trade_id: i64, + pub order_id: i64, + pub side: Side, + pub price: Decimal, + pub qty: Decimal, + pub quote_qty: Decimal, + pub commission: Decimal, + pub commission_asset: Option, + pub time_ms: TimestampMs, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct ClosedTrade { + pub symbol: Symbol, + pub entry_time_ms: TimestampMs, + pub exit_time_ms: TimestampMs, + pub entry_price: Decimal, + pub exit_price: Decimal, + pub qty: Decimal, + pub gross_quote_pnl: Decimal, + pub entry_order_id: Option, + pub exit_order_id: Option, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct AssetBalance { + pub asset: String, + pub free: Decimal, + pub locked: Decimal, +} + +impl AssetBalance { + pub fn total(&self) -> Decimal { + self.free + self.locked + } +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct SymbolRules { + pub symbol: Symbol, + pub base_asset: String, + pub quote_asset: String, + pub min_qty: Option, + pub max_qty: Option, + pub step_size: Option, + pub market_min_qty: Option, + pub market_max_qty: Option, + pub market_step_size: Option, + pub min_notional: Option, + pub tick_size: Option, +} + +impl SymbolRules { + pub fn effective_market_step_size(&self) -> Option { + self.market_step_size.or(self.step_size) + } + + pub fn effective_market_min_qty(&self) -> Option { + self.market_min_qty.or(self.min_qty) + } + + pub fn effective_market_max_qty(&self) -> Option { + self.market_max_qty.or(self.max_qty) + } +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct PositionState { + pub qty: Decimal, + pub entry_price: Option, + pub entry_time_ms: Option, + pub entry_order_id: Option, +} + +impl PositionState { + pub fn flat() -> Self { + Self { + qty: Decimal::ZERO, + entry_price: None, + entry_time_ms: None, + entry_order_id: None, + } + } + + pub fn is_open(&self) -> bool { + self.qty > Decimal::ZERO + } +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct BotRunState { + pub run_id: String, + pub market: MarketId, + pub strategy_name: String, + pub strategy_config: serde_json::Value, + pub status: RunStatus, + pub last_processed_open_time_ms: Option, + pub started_at_ms: TimestampMs, + pub updated_at_ms: TimestampMs, + pub stopped_at_ms: Option, + pub last_error: Option, + pub position: PositionState, +} + #[derive(Error, Debug)] pub enum ModelError { #[error("invalid symbol: {0}")] @@ -140,6 +473,9 @@ pub enum ModelError { #[error("invalid interval: {0}")] InvalidInterval(String), + #[error("invalid side: {0}")] + InvalidSide(String), + #[error("invalid rfc3339 timestamp: {0}")] InvalidTimestamp(String), @@ -152,8 +488,7 @@ pub fn parse_rfc3339_to_ms(input: &str) -> Result { if input.is_empty() { return Err(ModelError::InvalidTimestamp("empty".to_string())); } - - let dt = time::OffsetDateTime::parse(input, &time::format_description::well_known::Rfc3339)?; + let dt = OffsetDateTime::parse(input, &time::format_description::well_known::Rfc3339)?; Ok(dt.unix_timestamp() * 1000 + i64::from(dt.millisecond())) } @@ -161,20 +496,32 @@ pub fn ms_to_rfc3339(ms: TimestampMs) -> String { let seconds = ms.div_euclid(1000); let millis = ms.rem_euclid(1000) as u16; - let dt = match time::OffsetDateTime::from_unix_timestamp(seconds) { + let dt = match OffsetDateTime::from_unix_timestamp(seconds) { Ok(dt) => match dt.replace_millisecond(millis) { Ok(adjusted) => adjusted, - Err(_) => time::OffsetDateTime::UNIX_EPOCH, + Err(_) => OffsetDateTime::UNIX_EPOCH, }, - Err(_) => time::OffsetDateTime::UNIX_EPOCH, + Err(_) => OffsetDateTime::UNIX_EPOCH, }; match dt.format(&time::format_description::well_known::Rfc3339) { - Ok(s) => s, + Ok(value) => value, Err(_) => "1970-01-01T00:00:00Z".to_string(), } } +pub fn now_utc_ms() -> TimestampMs { + let now = OffsetDateTime::now_utc(); + now.unix_timestamp() * 1000 + i64::from(now.millisecond()) +} + +pub fn round_down_to_step(value: Decimal, step: Decimal) -> Decimal { + if step <= Decimal::ZERO { + return value; + } + (value / step).trunc() * step +} + #[derive(Clone, Debug, PartialEq, Eq)] pub enum ValidationIssue { OutOfOrder { @@ -277,57 +624,77 @@ mod tests { use super::*; use std::str::FromStr; + fn market() -> MarketId { + MarketId::new( + ExchangeId::BinanceSpot, + Symbol::new("BTCUSDT").expect("symbol"), + Interval::M1, + ) + } + #[test] - fn interval_parse_and_step() { - assert_eq!(Interval::from_str("1m").expect("interval"), Interval::M1); - assert_eq!(Interval::M5.step_ms(), 300_000); + fn round_down_to_step_behaves() { + assert_eq!( + round_down_to_step( + Decimal::from_str("1.234").expect("decimal"), + Decimal::from_str("0.01").expect("decimal") + ), + Decimal::from_str("1.23").expect("decimal") + ); } #[test] - fn parse_timestamp() { - let ms = parse_rfc3339_to_ms("2024-01-01T00:00:00Z").expect("timestamp"); - assert_eq!(ms, 1_704_067_200_000); + fn validation_detects_gap() { + let candles = vec![ + Candle { + open_time_ms: 0, + close_time_ms: 59_999, + open: Decimal::ONE, + high: Decimal::ONE, + low: Decimal::ONE, + close: Decimal::ONE, + volume: Decimal::ONE, + trades: Some(1), + }, + Candle { + open_time_ms: 120_000, + close_time_ms: 179_999, + open: Decimal::ONE, + high: Decimal::ONE, + low: Decimal::ONE, + close: Decimal::ONE, + volume: Decimal::ONE, + trades: Some(1), + }, + ]; + + let report = validate_candles(&market(), &candles); + assert_eq!(report.issues.len(), 1); + assert!(matches!(report.issues[0], ValidationIssue::Gap { .. })); } #[test] - fn detect_gap_and_duplicate() { - let market = MarketId::new( - ExchangeId::BinanceSpot, - Symbol::new("BTCUSDT").expect("symbol"), - Interval::M1, - ); - let c0 = Candle { - open_time_ms: 0, - close_time_ms: 59_999, - open: Decimal::from(1), - high: Decimal::from(2), - low: Decimal::from(1), - close: Decimal::from(2), - volume: Decimal::from(10), - trades: Some(1), - }; - let c1 = Candle { - open_time_ms: 120_000, - ..c0.clone() - }; - let c2 = Candle { - open_time_ms: 120_000, - ..c0.clone() + fn order_average_price_falls_back_to_ratio() { + let order = ExchangeOrder { + symbol: Symbol::new("BTCUSDT").expect("symbol"), + side: Side::Buy, + order_type: "MARKET".to_string(), + status: OrderStatus::Filled, + order_id: Some(1), + client_order_id: Some("abc".to_string()), + requested_qty: None, + requested_quote_qty: Some(Decimal::from_str("100").expect("decimal")), + executed_qty: Decimal::from_str("0.01").expect("decimal"), + cumulative_quote_qty: Decimal::from_str("100").expect("decimal"), + avg_price: None, + transact_time_ms: Some(1), + fills: Vec::new(), + raw: serde_json::json!({}), }; - let report = validate_candles(&market, &[c0, c1, c2]); - assert!(!report.is_ok()); - assert!( - report - .issues - .iter() - .any(|issue| matches!(issue, ValidationIssue::Gap { .. })) - ); - assert!( - report - .issues - .iter() - .any(|issue| matches!(issue, ValidationIssue::DuplicateOpenTime { .. })) + assert_eq!( + order.average_price(), + Some(Decimal::from_str("10000").expect("decimal")) ); } } diff --git a/src/ports.rs b/src/ports.rs new file mode 100644 index 0000000..0056270 --- /dev/null +++ b/src/ports.rs @@ -0,0 +1,174 @@ +use async_trait::async_trait; +use rust_decimal::Decimal; +use std::error::Error as StdError; +use thiserror::Error; + +use crate::model::{ + AccountTrade, AssetBalance, BotRunState, Candle, ClosedTrade, ExchangeId, ExchangeOrder, + Interval, MarketId, ModelError, Side, Symbol, SymbolRules, TimestampMs, +}; + +#[derive(Clone, Debug, Default)] +pub struct CandleQuery { + pub start_time_ms: Option, + pub end_time_ms: Option, + pub limit: Option, +} + +#[derive(Clone, Debug)] +pub struct KlineRequest { + pub symbol: Symbol, + pub interval: Interval, + pub start_time_ms: Option, + pub end_time_ms: Option, + pub limit: Option, +} + +#[derive(Clone, Debug)] +pub struct MarketOrderRequest { + pub symbol: Symbol, + pub side: Side, + pub quantity: Option, + pub quote_order_qty: Option, + pub new_client_order_id: Option, +} + +#[derive(Clone, Debug)] +pub struct CancelOrderRequest { + pub symbol: Symbol, + pub order_id: Option, + pub client_order_id: Option, +} + +#[derive(Clone, Debug)] +pub struct OrderQueryRequest { + pub symbol: Symbol, + pub order_id: Option, + pub client_order_id: Option, +} + +#[derive(Error, Debug)] +pub enum ExchangeError { + #[error("transport error")] + Transport(#[source] Box), + + #[error("api error: code={code:?} message={message}")] + Api { code: Option, message: String }, + + #[error("missing credentials")] + MissingCredentials, + + #[error("invalid response: {message}")] + InvalidResponse { message: String }, + + #[error("invalid request: {message}")] + InvalidRequest { message: String }, +} + +impl ExchangeError { + pub fn transport(err: E) -> Self + where + E: StdError + Send + Sync + 'static, + { + Self::Transport(Box::new(err)) + } +} + +impl From for ExchangeError { + fn from(err: ModelError) -> Self { + Self::InvalidResponse { + message: err.to_string(), + } + } +} + +#[derive(Error, Debug)] +pub enum StorageError { + #[error("storage error")] + Other(#[source] Box), + + #[error("invalid argument: {0}")] + InvalidArgument(String), +} + +impl StorageError { + pub fn other(err: E) -> Self + where + E: StdError + Send + Sync + 'static, + { + Self::Other(Box::new(err)) + } +} + +#[async_trait] +pub trait MarketDataSource: Send + Sync { + fn exchange_id(&self) -> ExchangeId; + async fn fetch_klines(&self, request: &KlineRequest) -> Result, ExchangeError>; + async fn fetch_symbol_rules(&self, symbol: &Symbol) -> Result; +} + +#[async_trait] +pub trait TradingVenue: Send + Sync { + fn exchange_id(&self) -> ExchangeId; + async fn account_balances(&self) -> Result, ExchangeError>; + async fn open_orders( + &self, + symbol: Option<&Symbol>, + ) -> Result, ExchangeError>; + async fn recent_trades( + &self, + symbol: &Symbol, + limit: usize, + ) -> Result, ExchangeError>; + async fn submit_market_order( + &self, + request: &MarketOrderRequest, + ) -> Result; + async fn cancel_order( + &self, + request: &CancelOrderRequest, + ) -> Result; + async fn query_order( + &self, + request: &OrderQueryRequest, + ) -> Result; +} + +pub trait CandleStore: Send + Sync { + fn init(&self) -> Result<(), StorageError>; + fn upsert_candles(&self, market: &MarketId, candles: &[Candle]) -> Result; + fn load_candles( + &self, + market: &MarketId, + query: CandleQuery, + ) -> Result, StorageError>; + fn load_recent_candles( + &self, + market: &MarketId, + limit: usize, + ) -> Result, StorageError>; + fn max_open_time_ms(&self, market: &MarketId) -> Result, StorageError>; +} + +pub trait RunJournalStore: Send + Sync { + fn init(&self) -> Result<(), StorageError>; + fn save_run_state(&self, state: &BotRunState) -> Result<(), StorageError>; + fn load_run_state(&self, run_id: &str) -> Result, StorageError>; + fn latest_run_for_market( + &self, + market: &MarketId, + strategy_name: &str, + ) -> Result, StorageError>; + fn append_order_event(&self, run_id: &str, order: &ExchangeOrder) -> Result<(), StorageError>; + fn append_closed_trade(&self, run_id: &str, trade: &ClosedTrade) -> Result<(), StorageError>; + fn list_order_events( + &self, + run_id: &str, + limit: usize, + ) -> Result, StorageError>; + fn list_closed_trades( + &self, + run_id: &str, + limit: usize, + ) -> Result, StorageError>; +} diff --git a/src/sdk.rs b/src/sdk.rs index 3472806..e30933b 100644 --- a/src/sdk.rs +++ b/src/sdk.rs @@ -1,8 +1,9 @@ use rust_decimal::Decimal; +use serde::{Deserialize, Serialize}; use std::collections::VecDeque; use thiserror::Error; -use crate::model::{Candle, MarketId, TimestampMs}; +use crate::model::{Candle, MarketId, TargetPosition, TimestampMs}; #[derive(Error, Debug)] pub enum StrategyError { @@ -16,12 +17,6 @@ impl StrategyError { } } -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub enum TargetPosition { - Flat, - LongAllIn, -} - pub trait StrategyContext { fn market(&self) -> &MarketId; fn now_ms(&self) -> TimestampMs; @@ -30,7 +25,7 @@ pub trait StrategyContext { fn set_target_position(&mut self, target: TargetPosition); } -pub trait Strategy { +pub trait Strategy: Send { fn name(&self) -> &'static str; fn on_start(&mut self, _ctx: &mut dyn StrategyContext) -> Result<(), StrategyError> { @@ -44,6 +39,14 @@ pub trait Strategy { } } +pub trait Indicator { + type Input; + type Output; + + fn reset(&mut self); + fn update(&mut self, input: Self::Input) -> Option; +} + #[derive(Clone, Debug)] pub struct Sma { window: usize, @@ -56,17 +59,26 @@ impl Sma { if window == 0 { return Err(StrategyError::msg("SMA window must be greater than zero")); } - Ok(Self { window, sum: Decimal::ZERO, values: VecDeque::with_capacity(window), }) } +} + +impl Indicator for Sma { + type Input = Decimal; + type Output = Decimal; - pub fn update(&mut self, value: Decimal) -> Option { - self.values.push_back(value); - self.sum += value; + fn reset(&mut self) { + self.sum = Decimal::ZERO; + self.values.clear(); + } + + fn update(&mut self, input: Self::Input) -> Option { + self.values.push_back(input); + self.sum += input; if self.values.len() > self.window { if let Some(removed) = self.values.pop_front() { @@ -82,6 +94,28 @@ impl Sma { } } +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum BuiltInStrategyConfig { + SmaCross { fast: usize, slow: usize }, +} + +impl BuiltInStrategyConfig { + pub fn strategy_name(&self) -> &'static str { + match self { + Self::SmaCross { .. } => "sma_cross", + } + } + + pub fn build(&self) -> Result, StrategyError> { + match self { + Self::SmaCross { fast, slow } => { + Ok(Box::new(strategies::SmaCrossStrategy::new(*fast, *slow)?)) + } + } + } +} + pub mod strategies { use super::*; @@ -129,13 +163,10 @@ pub mod strategies { let slow_now = self.slow.update(bar.close); if let (Some(fast_now), Some(slow_now)) = (fast_now, slow_now) { - if let (Some(prev_fast), Some(prev_slow)) = (self.prev_fast, self.prev_slow) { - if prev_fast <= prev_slow && fast_now > slow_now { - ctx.set_target_position(TargetPosition::LongAllIn); - } - if prev_fast >= prev_slow && fast_now < slow_now { - ctx.set_target_position(TargetPosition::Flat); - } + if fast_now > slow_now { + ctx.set_target_position(TargetPosition::LongAllIn); + } else if fast_now < slow_now { + ctx.set_target_position(TargetPosition::Flat); } self.prev_fast = Some(fast_now); @@ -161,9 +192,5 @@ mod tests { sma.update(Decimal::from_str("3").expect("decimal")), Some(Decimal::from_str("2").expect("decimal")) ); - assert_eq!( - sma.update(Decimal::from_str("6").expect("decimal")), - Some(Decimal::from_str("3.6666666666666666666666666667").expect("decimal")) - ); } } diff --git a/src/storage.rs b/src/storage.rs index 9e00c24..62653a7 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -1,58 +1,24 @@ +use crate::{ + BotRunState, Candle, CandleQuery, CandleStore, ClosedTrade, ExchangeOrder, MarketId, + RunJournalStore, StorageError, TimestampMs, +}; use rusqlite::{Connection, OptionalExtension, ToSql, params, params_from_iter}; use rust_decimal::Decimal; use std::{ - error::Error as StdError, path::{Path, PathBuf}, + str::FromStr, + time::Duration, }; -use thiserror::Error; use tracing::info; -use crate::model::{Candle, MarketId, TimestampMs}; - -const SCHEMA_VERSION: &str = "1"; - -#[derive(Clone, Debug, Default)] -pub struct CandleQuery { - pub start_time_ms: Option, - pub end_time_ms: Option, - pub limit: Option, -} - -#[derive(Error, Debug)] -pub enum StorageError { - #[error("storage error")] - Other(#[source] Box), - - #[error("invalid argument: {0}")] - InvalidArgument(String), -} - -impl StorageError { - pub fn other(err: E) -> Self - where - E: StdError + Send + Sync + 'static, - { - Self::Other(Box::new(err)) - } -} - -pub trait CandleStore: Send + Sync { - fn init(&self) -> Result<(), StorageError>; - fn upsert_candles(&self, market: &MarketId, candles: &[Candle]) -> Result; - fn load_candles( - &self, - market: &MarketId, - query: CandleQuery, - ) -> Result, StorageError>; - fn max_open_time_ms(&self, market: &MarketId) -> Result, StorageError>; -} +const SCHEMA_VERSION: &str = "2"; #[derive(Clone, Debug)] -pub struct SqliteCandleStore { +pub struct SqliteStore { path: PathBuf, } -impl SqliteCandleStore { +impl SqliteStore { pub fn new(path: impl Into) -> Self { Self { path: path.into() } } @@ -69,12 +35,13 @@ impl SqliteCandleStore { connection .pragma_update(None, "synchronous", "NORMAL") .map_err(StorageError::other)?; + connection + .busy_timeout(Duration::from_secs(5)) + .map_err(StorageError::other)?; Ok(connection) } -} -impl CandleStore for SqliteCandleStore { - fn init(&self) -> Result<(), StorageError> { + fn initialize_schema(&self) -> Result<(), StorageError> { let connection = self.open()?; connection .execute_batch( @@ -101,31 +68,80 @@ impl CandleStore for SqliteCandleStore { CREATE INDEX IF NOT EXISTS idx_candles_market_time ON candles(exchange, symbol, interval, open_time_ms); + + CREATE TABLE IF NOT EXISTS bot_runs ( + run_id TEXT PRIMARY KEY, + exchange TEXT NOT NULL, + symbol TEXT NOT NULL, + interval TEXT NOT NULL, + strategy_name TEXT NOT NULL, + status TEXT NOT NULL, + state_json TEXT NOT NULL, + started_at_ms INTEGER NOT NULL, + updated_at_ms INTEGER NOT NULL, + stopped_at_ms INTEGER, + last_error TEXT + ); + + CREATE INDEX IF NOT EXISTS idx_bot_runs_market + ON bot_runs(exchange, symbol, interval, strategy_name, updated_at_ms DESC); + + CREATE TABLE IF NOT EXISTS order_events ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + run_id TEXT NOT NULL, + symbol TEXT NOT NULL, + side TEXT NOT NULL, + order_type TEXT NOT NULL, + status TEXT NOT NULL, + order_id INTEGER, + client_order_id TEXT, + transact_time_ms INTEGER, + raw_json TEXT NOT NULL + ); + + CREATE INDEX IF NOT EXISTS idx_order_events_run + ON order_events(run_id, seq DESC); + + CREATE TABLE IF NOT EXISTS closed_trades ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + run_id TEXT NOT NULL, + symbol TEXT NOT NULL, + entry_time_ms INTEGER NOT NULL, + exit_time_ms INTEGER NOT NULL, + entry_price TEXT NOT NULL, + exit_price TEXT NOT NULL, + qty TEXT NOT NULL, + gross_quote_pnl TEXT NOT NULL, + entry_order_id INTEGER, + exit_order_id INTEGER + ); + + CREATE INDEX IF NOT EXISTS idx_closed_trades_run + ON closed_trades(run_id, seq DESC); "#, ) .map_err(StorageError::other)?; - let version = connection - .query_row( - "SELECT value FROM meta WHERE key = 'schema_version'", - [], - |row| row.get::<_, String>(0), + connection + .execute( + r#" + INSERT INTO meta(key, value) + VALUES ('schema_version', ?1) + ON CONFLICT(key) DO UPDATE SET value = excluded.value + "#, + [SCHEMA_VERSION], ) - .optional() .map_err(StorageError::other)?; - if version.is_none() { - connection - .execute( - "INSERT INTO meta(key, value) VALUES ('schema_version', ?1)", - [SCHEMA_VERSION], - ) - .map_err(StorageError::other)?; - } - - info!(db_path = %self.path.display(), "sqlite candle store initialized"); + info!(db_path = %self.path.display(), "sqlite store initialized"); Ok(()) } +} + +impl CandleStore for SqliteStore { + fn init(&self) -> Result<(), StorageError> { + self.initialize_schema() + } fn upsert_candles(&self, market: &MarketId, candles: &[Candle]) -> Result { if candles.is_empty() { @@ -201,16 +217,15 @@ impl CandleStore for SqliteCandleStore { let start_time_ms = query.start_time_ms; let end_time_ms = query.end_time_ms; - let mut params: Vec<&dyn ToSql> = vec![&exchange, &symbol, &interval]; + let mut sql_params: Vec<&dyn ToSql> = vec![&exchange, &symbol, &interval]; - if let Some(ref start_time_ms_value) = start_time_ms { - sql.push_str(&format!(" AND open_time_ms >= ?{}", params.len() + 1)); - params.push(start_time_ms_value); + if let Some(ref start_value) = start_time_ms { + sql.push_str(&format!(" AND open_time_ms >= ?{}", sql_params.len() + 1)); + sql_params.push(start_value); } - - if let Some(ref end_time_ms_value) = end_time_ms { - sql.push_str(&format!(" AND open_time_ms <= ?{}", params.len() + 1)); - params.push(end_time_ms_value); + if let Some(ref end_value) = end_time_ms { + sql.push_str(&format!(" AND open_time_ms <= ?{}", sql_params.len() + 1)); + sql_params.push(end_value); } sql.push_str(" ORDER BY open_time_ms ASC"); @@ -220,21 +235,15 @@ impl CandleStore for SqliteCandleStore { let mut stmt = connection.prepare(&sql).map_err(StorageError::other)?; let rows = stmt - .query_map(params_from_iter(params), |row| { - let open = row.get::<_, String>(2)?; - let high = row.get::<_, String>(3)?; - let low = row.get::<_, String>(4)?; - let close = row.get::<_, String>(5)?; - let volume = row.get::<_, String>(6)?; - - parse_row( + .query_map(params_from_iter(sql_params), |row| { + parse_candle_row( row.get(0)?, row.get(1)?, - &open, - &high, - &low, - &close, - &volume, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + row.get::<_, String>(4)?, + row.get::<_, String>(5)?, + row.get::<_, String>(6)?, row.get::<_, Option>(7)?, ) }) @@ -244,7 +253,55 @@ impl CandleStore for SqliteCandleStore { for row in rows { candles.push(row.map_err(StorageError::other)?); } + Ok(candles) + } + fn load_recent_candles( + &self, + market: &MarketId, + limit: usize, + ) -> Result, StorageError> { + let connection = self.open()?; + let mut stmt = connection + .prepare( + r#" + SELECT open_time_ms, close_time_ms, open, high, low, close, volume, trades + FROM candles + WHERE exchange = ?1 AND symbol = ?2 AND interval = ?3 + ORDER BY open_time_ms DESC + LIMIT ?4 + "#, + ) + .map_err(StorageError::other)?; + + let rows = stmt + .query_map( + params![ + market.exchange.as_str(), + market.symbol.as_str(), + market.interval.as_str(), + limit as i64 + ], + |row| { + parse_candle_row( + row.get(0)?, + row.get(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + row.get::<_, String>(4)?, + row.get::<_, String>(5)?, + row.get::<_, String>(6)?, + row.get::<_, Option>(7)?, + ) + }, + ) + .map_err(StorageError::other)?; + + let mut candles = Vec::new(); + for row in rows { + candles.push(row.map_err(StorageError::other)?); + } + candles.reverse(); Ok(candles) } @@ -269,61 +326,295 @@ impl CandleStore for SqliteCandleStore { } } -fn parse_decimal(input: &str, field: &str) -> Result { - input.parse::().map_err(|e| { - rusqlite::Error::FromSqlConversionFailure( - 0, - rusqlite::types::Type::Text, - Box::new(std::io::Error::new( - std::io::ErrorKind::InvalidData, - format!("failed to parse {field}: {e}"), - )), - ) - }) +impl RunJournalStore for SqliteStore { + fn init(&self) -> Result<(), StorageError> { + self.initialize_schema() + } + + fn save_run_state(&self, state: &BotRunState) -> Result<(), StorageError> { + let connection = self.open()?; + let state_json = serde_json::to_string(state).map_err(StorageError::other)?; + connection + .execute( + r#" + INSERT INTO bot_runs( + run_id, exchange, symbol, interval, strategy_name, status, + state_json, started_at_ms, updated_at_ms, stopped_at_ms, last_error + ) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) + ON CONFLICT(run_id) DO UPDATE SET + status = excluded.status, + state_json = excluded.state_json, + updated_at_ms = excluded.updated_at_ms, + stopped_at_ms = excluded.stopped_at_ms, + last_error = excluded.last_error + "#, + params![ + &state.run_id, + state.market.exchange.as_str(), + state.market.symbol.as_str(), + state.market.interval.as_str(), + &state.strategy_name, + state.status.as_str(), + state_json, + state.started_at_ms, + state.updated_at_ms, + state.stopped_at_ms, + state.last_error.as_deref(), + ], + ) + .map_err(StorageError::other)?; + Ok(()) + } + + fn load_run_state(&self, run_id: &str) -> Result, StorageError> { + let connection = self.open()?; + let state_json = connection + .query_row( + "SELECT state_json FROM bot_runs WHERE run_id = ?1", + [run_id], + |row| row.get::<_, String>(0), + ) + .optional() + .map_err(StorageError::other)?; + + match state_json { + Some(json) => { + let state = + serde_json::from_str::(&json).map_err(StorageError::other)?; + Ok(Some(state)) + } + None => Ok(None), + } + } + + fn latest_run_for_market( + &self, + market: &MarketId, + strategy_name: &str, + ) -> Result, StorageError> { + let connection = self.open()?; + let state_json = connection + .query_row( + r#" + SELECT state_json + FROM bot_runs + WHERE exchange = ?1 + AND symbol = ?2 + AND interval = ?3 + AND strategy_name = ?4 + ORDER BY updated_at_ms DESC + LIMIT 1 + "#, + params![ + market.exchange.as_str(), + market.symbol.as_str(), + market.interval.as_str(), + strategy_name + ], + |row| row.get::<_, String>(0), + ) + .optional() + .map_err(StorageError::other)?; + + match state_json { + Some(json) => { + let state = + serde_json::from_str::(&json).map_err(StorageError::other)?; + Ok(Some(state)) + } + None => Ok(None), + } + } + + fn append_order_event(&self, run_id: &str, order: &ExchangeOrder) -> Result<(), StorageError> { + let connection = self.open()?; + let raw_json = serde_json::to_string(order).map_err(StorageError::other)?; + connection + .execute( + r#" + INSERT INTO order_events( + run_id, symbol, side, order_type, status, order_id, client_order_id, + transact_time_ms, raw_json + ) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) + "#, + params![ + run_id, + order.symbol.as_str(), + order.side.as_str(), + &order.order_type, + order.status.as_str(), + order.order_id, + order.client_order_id.as_deref(), + order.transact_time_ms, + raw_json + ], + ) + .map_err(StorageError::other)?; + Ok(()) + } + + fn append_closed_trade(&self, run_id: &str, trade: &ClosedTrade) -> Result<(), StorageError> { + let connection = self.open()?; + connection + .execute( + r#" + INSERT INTO closed_trades( + run_id, symbol, entry_time_ms, exit_time_ms, entry_price, exit_price, + qty, gross_quote_pnl, entry_order_id, exit_order_id + ) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10) + "#, + params![ + run_id, + trade.symbol.as_str(), + trade.entry_time_ms, + trade.exit_time_ms, + trade.entry_price.to_string(), + trade.exit_price.to_string(), + trade.qty.to_string(), + trade.gross_quote_pnl.to_string(), + trade.entry_order_id, + trade.exit_order_id + ], + ) + .map_err(StorageError::other)?; + Ok(()) + } + + fn list_order_events( + &self, + run_id: &str, + limit: usize, + ) -> Result, StorageError> { + let connection = self.open()?; + let mut stmt = connection + .prepare( + r#" + SELECT raw_json + FROM order_events + WHERE run_id = ?1 + ORDER BY seq DESC + LIMIT ?2 + "#, + ) + .map_err(StorageError::other)?; + + let rows = stmt + .query_map(params![run_id, limit as i64], |row| row.get::<_, String>(0)) + .map_err(StorageError::other)?; + + let mut items = Vec::new(); + for row in rows { + let raw_json = row.map_err(StorageError::other)?; + let order = + serde_json::from_str::(&raw_json).map_err(StorageError::other)?; + items.push(order); + } + Ok(items) + } + + fn list_closed_trades( + &self, + run_id: &str, + limit: usize, + ) -> Result, StorageError> { + let connection = self.open()?; + let mut stmt = connection + .prepare( + r#" + SELECT symbol, entry_time_ms, exit_time_ms, entry_price, exit_price, + qty, gross_quote_pnl, entry_order_id, exit_order_id + FROM closed_trades + WHERE run_id = ?1 + ORDER BY seq DESC + LIMIT ?2 + "#, + ) + .map_err(StorageError::other)?; + + let rows = stmt + .query_map(params![run_id, limit as i64], |row| { + Ok(ClosedTrade { + symbol: row + .get::<_, String>(0)? + .parse() + .map_err(to_from_sql_error)?, + entry_time_ms: row.get(1)?, + exit_time_ms: row.get(2)?, + entry_price: parse_decimal_str(&row.get::<_, String>(3)?)?, + exit_price: parse_decimal_str(&row.get::<_, String>(4)?)?, + qty: parse_decimal_str(&row.get::<_, String>(5)?)?, + gross_quote_pnl: parse_decimal_str(&row.get::<_, String>(6)?)?, + entry_order_id: row.get(7)?, + exit_order_id: row.get(8)?, + }) + }) + .map_err(StorageError::other)?; + + let mut trades = Vec::new(); + for row in rows { + trades.push(row.map_err(StorageError::other)?); + } + Ok(trades) + } } #[allow(clippy::too_many_arguments)] -fn parse_row( +fn parse_candle_row( open_time_ms: i64, close_time_ms: i64, - open: &str, - high: &str, - low: &str, - close: &str, - volume: &str, + open: String, + high: String, + low: String, + close: String, + volume: String, trades: Option, ) -> Result { Ok(Candle { open_time_ms, close_time_ms, - open: parse_decimal(open, "open")?, - high: parse_decimal(high, "high")?, - low: parse_decimal(low, "low")?, - close: parse_decimal(close, "close")?, - volume: parse_decimal(volume, "volume")?, + open: parse_decimal_str(&open)?, + high: parse_decimal_str(&high)?, + low: parse_decimal_str(&low)?, + close: parse_decimal_str(&close)?, + volume: parse_decimal_str(&volume)?, trades: trades.map(|value| value as u64), }) } +fn parse_decimal_str(raw: &str) -> Result { + Decimal::from_str(raw).map_err(to_from_sql_error) +} + +fn to_from_sql_error(err: E) -> rusqlite::Error +where + E: std::error::Error + Send + Sync + 'static, +{ + rusqlite::Error::FromSqlConversionFailure(0, rusqlite::types::Type::Text, Box::new(err)) +} + #[cfg(test)] mod tests { use super::*; - use crate::model::{ExchangeId, Interval, Symbol}; - use std::str::FromStr; + use crate::{ExchangeId, Interval, PositionState, RunStatus, Side, Symbol, now_utc_ms}; use tempfile::tempdir; - #[test] - fn sqlite_roundtrip() { - let tempdir = tempdir().expect("tempdir"); - let db_path = tempdir.path().join("market.sqlite"); - let store = SqliteCandleStore::new(&db_path); - store.init().expect("init"); - - let market = MarketId::new( + fn market() -> MarketId { + MarketId::new( ExchangeId::BinanceSpot, Symbol::new("BTCUSDT").expect("symbol"), Interval::M1, - ); + ) + } + + #[test] + fn sqlite_roundtrip_for_candles_and_runs() { + let tempdir = tempdir().expect("tempdir"); + let db_path = tempdir.path().join("market.sqlite"); + let store = SqliteStore::new(&db_path); + CandleStore::init(&store).expect("init"); let first = Candle { open_time_ms: 1_700_000_000_000, @@ -335,37 +626,83 @@ mod tests { volume: Decimal::from_str("12.34").expect("decimal"), trades: Some(42), }; - let second = Candle { - open_time_ms: 1_700_000_060_000, - close_time_ms: 1_700_000_119_999, - open: Decimal::from_str("105").expect("decimal"), - high: Decimal::from_str("120").expect("decimal"), - low: Decimal::from_str("100").expect("decimal"), - close: Decimal::from_str("115").expect("decimal"), - volume: Decimal::from_str("56.78").expect("decimal"), - trades: Some(43), - }; - let count = store - .upsert_candles(&market, &[first.clone(), second.clone()]) + store + .upsert_candles(&market(), std::slice::from_ref(&first)) .expect("upsert"); - assert_eq!(count, 2); - let loaded = store - .load_candles( - &market, - CandleQuery { - start_time_ms: Some(first.open_time_ms), - end_time_ms: Some(second.open_time_ms), - limit: None, - }, - ) - .expect("load"); + .load_recent_candles(&market(), 1) + .expect("load recent"); + assert_eq!(loaded, vec![first]); + + let run_state = BotRunState { + run_id: "run-1".to_string(), + market: market(), + strategy_name: "sma_cross".to_string(), + strategy_config: serde_json::json!({"kind":"sma_cross","fast":20,"slow":50}), + status: RunStatus::Running, + last_processed_open_time_ms: Some(1_700_000_000_000), + started_at_ms: now_utc_ms(), + updated_at_ms: now_utc_ms(), + stopped_at_ms: None, + last_error: None, + position: PositionState::flat(), + }; + + RunJournalStore::save_run_state(&store, &run_state).expect("save run"); + let loaded_run = store + .load_run_state("run-1") + .expect("load run") + .expect("run state"); + assert_eq!(loaded_run.run_id, "run-1"); + + let order = ExchangeOrder { + symbol: Symbol::new("BTCUSDT").expect("symbol"), + side: Side::Buy, + order_type: "MARKET".to_string(), + status: crate::OrderStatus::Filled, + order_id: Some(7), + client_order_id: Some("abc".to_string()), + requested_qty: None, + requested_quote_qty: Some(Decimal::from_str("100").expect("decimal")), + executed_qty: Decimal::from_str("0.01").expect("decimal"), + cumulative_quote_qty: Decimal::from_str("100").expect("decimal"), + avg_price: Some(Decimal::from_str("10000").expect("decimal")), + transact_time_ms: Some(1), + fills: Vec::new(), + raw: serde_json::json!({}), + }; + store + .append_order_event("run-1", &order) + .expect("append order"); + assert_eq!( + store + .list_order_events("run-1", 10) + .expect("list order") + .len(), + 1 + ); - assert_eq!(loaded, vec![first, second]); + let trade = ClosedTrade { + symbol: Symbol::new("BTCUSDT").expect("symbol"), + entry_time_ms: 1, + exit_time_ms: 2, + entry_price: Decimal::from_str("10000").expect("decimal"), + exit_price: Decimal::from_str("10100").expect("decimal"), + qty: Decimal::from_str("0.01").expect("decimal"), + gross_quote_pnl: Decimal::from_str("1").expect("decimal"), + entry_order_id: Some(7), + exit_order_id: Some(8), + }; + store + .append_closed_trade("run-1", &trade) + .expect("append trade"); assert_eq!( - store.max_open_time_ms(&market).expect("max"), - Some(1_700_000_060_000) + store + .list_closed_trades("run-1", 10) + .expect("list trade") + .len(), + 1 ); } } diff --git a/tests/cli.rs b/tests/cli.rs index 3f30c3c..c3ee9a2 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -2,12 +2,13 @@ use assert_cmd::Command; use predicates::prelude::*; #[test] -fn help_lists_core_commands() { +fn help_lists_v020_command_groups() { let mut cmd = Command::cargo_bin("quantforge").expect("binary"); cmd.arg("--help"); cmd.assert() .success() - .stdout(predicate::str::contains("download")) - .stdout(predicate::str::contains("validate")) - .stdout(predicate::str::contains("backtest")); + .stdout(predicate::str::contains("data")) + .stdout(predicate::str::contains("backtest")) + .stdout(predicate::str::contains("trade")) + .stdout(predicate::str::contains("monitor")); }