diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..4f1caa6 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,107 @@ +name: CI + +on: + push: + pull_request: + +env: + CARGO_TERM_COLOR: always + +permissions: + contents: read + +jobs: + build: + name: Build + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: step-security/harden-runner@v2 + with: + egress-policy: audit + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@stable + - uses: Swatinem/rust-cache@v2 + with: + cache-on-failure: true + - name: Install just + uses: taiki-e/install-action@just + - name: Build + run: just build + + test: + name: Test + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: step-security/harden-runner@v2 + with: + egress-policy: audit + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@stable + - uses: Swatinem/rust-cache@v2 + with: + cache-on-failure: true + - name: Install cargo-nextest + uses: taiki-e/install-action@cargo-nextest + - name: Install just + uses: taiki-e/install-action@just + - name: Test + run: just test + + format: + name: Format + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: step-security/harden-runner@v2 + with: + egress-policy: audit + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@nightly + with: + components: rustfmt + - name: Install just + uses: taiki-e/install-action@just + - name: Check formatting + run: just check-format + + clippy: + name: Clippy + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: step-security/harden-runner@v2 + with: + egress-policy: audit + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@stable + with: + components: clippy + - uses: Swatinem/rust-cache@v2 + with: + cache-on-failure: true + - name: Install just + uses: taiki-e/install-action@just + - name: Clippy + run: just check-clippy + + udeps: + name: Unused Dependencies + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: step-security/harden-runner@v2 + with: + egress-policy: audit + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@nightly + - uses: Swatinem/rust-cache@v2 + with: + cache-on-failure: true + - name: Install cargo-udeps + uses: taiki-e/install-action@cargo-udeps + - name: Install just + uses: taiki-e/install-action@just + - name: Check unused dependencies + run: just check-udeps diff --git a/.github/workflows/lychee.yml b/.github/workflows/lychee.yml new file mode 100644 index 0000000..ee664e4 --- /dev/null +++ b/.github/workflows/lychee.yml @@ -0,0 +1,24 @@ +name: Lychee Link Check + +on: + push: + pull_request: + workflow_dispatch: + +permissions: + contents: read + +jobs: + lychee-checks: + name: Check Links + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: step-security/harden-runner@v2 + with: + egress-policy: audit + - uses: actions/checkout@v4 + - uses: lycheeverse/lychee-action@v2 + with: + args: --config ./lychee.toml . + fail: true diff --git a/.gitignore b/.gitignore index ad67955..6686961 100644 --- a/.gitignore +++ b/.gitignore @@ -1,21 +1,7 @@ -# Generated by Cargo -# will have compiled files and executables -debug -target - -# These are backup files generated by rustfmt -**/*.rs.bk - -# MSVC Windows builds of rustc generate these, which store debugging information -*.pdb - -# Generated by cargo mutants -# Contains mutation testing data -**/mutants.out*/ - -# RustRover -# JetBrains specific template is maintained in a separate JetBrains.gitignore that can -# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore -# and can be added to the global gitignore or merged into this file. For a more nuclear -# option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ +/target +Cargo.lock +*.swp +*.swo +.DS_Store +.idea/ +.vscode/ diff --git a/.lycheecache b/.lycheecache new file mode 100644 index 0000000..e69de29 diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..6fd8eee --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,94 @@ +[workspace.package] +version = "0.1.0" +edition = "2024" +rust-version = "1.88" +license = "MIT" +homepage = "https://github.com/factordynamics/paracas" +repository = "https://github.com/factordynamics/paracas" +documentation = "https://docs.rs/paracas" +authors = ["Factor Dynamics"] +keywords = ["dukascopy", "forex", "tick-data", "trading", "historical-data"] +categories = ["finance", "command-line-utilities", "encoding"] + +[workspace] +resolver = "2" +members = ["crates/*", "bin"] +default-members = ["bin"] + +[workspace.lints.rust] +missing-debug-implementations = "warn" +missing-docs = "warn" +unreachable-pub = "warn" +unused-must-use = "deny" +rust-2018-idioms = "deny" +unnameable-types = "warn" + +[workspace.lints.rustdoc] +all = "warn" + +[workspace.lints.clippy] +all = { level = "warn", priority = -1 } +missing-const-for-fn = "warn" +use-self = "warn" +option-if-let-else = "warn" +redundant-clone = "warn" + +[profile.dev] +debug = "line-tables-only" + +[profile.release] +opt-level = 3 +lto = "thin" +debug = "none" +strip = "symbols" +codegen-units = 16 + +[workspace.dependencies] +# Internal crates +paracas-lib = { path = "crates/paracas-lib", version = "0.1.0" } +paracas-types = { path = "crates/paracas-types", version = "0.1.0" } +paracas-instruments = { path = "crates/paracas-instruments", version = "0.1.0" } +paracas-fetch = { path = "crates/paracas-fetch", version = "0.1.0" } +paracas-aggregate = { path = "crates/paracas-aggregate", version = "0.1.0" } +paracas-format = { path = "crates/paracas-format", version = "0.1.0" } + +# Async runtime +tokio = { version = "1.42", features = ["full"] } +futures = "0.3" +async-trait = "0.1" + +# HTTP client +reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "gzip", "stream"] } +bytes = "1.9" + +# Compression +lzma-rs = "0.3" + +# Serialization +serde = { version = "1.0", features = ["derive"] } +serde_json = "1.0" +csv-async = { version = "1.3", features = ["tokio"] } + +# Arrow/Parquet +arrow = { version = "54", features = ["chrono-tz"] } +parquet = { version = "54", features = ["async", "arrow"] } + +# Date/Time +chrono = { version = "0.4", features = ["serde"] } + +# CLI +clap = { version = "4.5", features = ["derive"] } +indicatif = "0.17" + +# Error handling +thiserror = "2.0" +anyhow = "1.0" + +# Binary parsing +byteorder = "1.5" + +# Utilities +derive_more = { version = "1.0", default-features = false, features = ["display", "from", "into"] } + +# Testing +approx = "0.5" diff --git a/Justfile b/Justfile new file mode 100644 index 0000000..abe17c3 --- /dev/null +++ b/Justfile @@ -0,0 +1,140 @@ +set positional-arguments + +alias t := test +alias f := fix +alias b := build +alias c := clean + +default: + @just --list + +# ============================================================ +# CI +# ============================================================ + +# Run the full CI suite +ci: fix check lychee check-udeps + +# Run all checks +check: check-format check-clippy test + +# Check links in documentation +lychee: + @command -v lychee >/dev/null 2>&1 || cargo install lychee + lychee --config ./lychee.toml . + +# ============================================================ +# Formatting +# ============================================================ + +# Check code formatting +check-format: + cargo +nightly fmt --all -- --check + +# Format code +format-fix: + cargo fix --allow-dirty --allow-staged + cargo +nightly fmt --all + +# ============================================================ +# Linting +# ============================================================ + +# Run clippy +check-clippy: + cargo clippy --workspace --all-targets -- -D warnings + +# Auto-fix clippy issues +clippy-fix: + cargo clippy --workspace --all-targets --fix --allow-dirty --allow-staged + +# ============================================================ +# Testing +# ============================================================ + +# Run tests +test: + @command -v cargo-nextest >/dev/null 2>&1 || cargo install cargo-nextest + RUSTFLAGS="-D warnings" cargo nextest run --workspace --all-features + +# Watch tests +watch-test: + cargo watch -x "nextest run --workspace --all-features" + +# ============================================================ +# Building +# ============================================================ + +# Build release +build: + cargo build --workspace --release + +# Clean build artifacts +clean: + cargo clean + +# ============================================================ +# Dependencies +# ============================================================ + +# Check for unused dependencies +check-udeps: + @command -v cargo-udeps >/dev/null 2>&1 || cargo install cargo-udeps + cargo +nightly udeps --workspace --all-features --all-targets + +# ============================================================ +# Documentation +# ============================================================ + +# Generate and open documentation +doc: + cargo doc --workspace --no-deps --open + +# Watch for changes and run checks +watch-check: + cargo watch -x "clippy --workspace --all-targets" + +# ============================================================ +# Install +# ============================================================ + +# Install paracas from crates.io +install: + cargo install paracas + +# Install paracas locally from source +install-local: + cargo install --path bin + +# ============================================================ +# Download Commands +# ============================================================ + +# Download tick data (all available history by default) +download instrument start="" end="" output="" format="csv": + cargo run --package paracas --release -- download {{instrument}} \ + {{ if start != "" { "-s " + start } else { "" } }} \ + {{ if end != "" { "-e " + end } else { "" } }} \ + {{ if output != "" { "-o " + output } else { "" } }} \ + -f {{format}} + +# Download with OHLCV aggregation +download-ohlcv instrument start="" end="" output="" format="csv" timeframe="h1": + cargo run --package paracas --release -- download {{instrument}} \ + {{ if start != "" { "-s " + start } else { "" } }} \ + {{ if end != "" { "-e " + end } else { "" } }} \ + {{ if output != "" { "-o " + output } else { "" } }} \ + -f {{format}} -t {{timeframe}} + +# List available instruments +list-instruments: + cargo run --package paracas --release -- list + +# Show instrument info +info instrument: + cargo run --package paracas --release -- info {{instrument}} + +# Compound command for development +fix: + cargo +nightly fmt --all + cargo clippy --workspace --all-targets --fix --allow-dirty --allow-staged diff --git a/README.md b/README.md index 116ce96..f9b921b 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,81 @@ # paracas -A Parallelized, Performant CLI Tool to Download Historical Intraday Tick Data. Built in Rust. + +High-performance Rust CLI for downloading historical tick data from Dukascopy. + +## Features + +- **Fast**: Concurrent downloads with connection pooling +- **Flexible**: CSV, JSON, and Parquet output formats +- **Complete**: All 1000+ Dukascopy instruments supported +- **Aggregation**: Built-in OHLCV aggregation (1m, 5m, 15m, 30m, 1h, 4h, 1d) + +## Installation + +```bash +cargo install paracas +``` + +## Usage + +### Download Tick Data + +```bash +# Download all available EUR/USD tick data (outputs to eurusd.csv by default) +paracas download eurusd + +# Download with specific date range +paracas download eurusd -s 2024-01-01 -e 2024-01-31 + +# Download BTC/USD as Parquet +paracas download btcusd -f parquet + +# Download with 1-hour OHLCV aggregation +paracas download eurusd -s 2024-01-01 -e 2024-01-31 -t h1 + +# Specify custom output file +paracas download eurusd -o my_data.csv +``` + +### List Instruments + +```bash +# List all instruments +paracas list + +# Filter by category +paracas list --category forex + +# Search instruments +paracas list --search btc +``` + +### Instrument Info + +```bash +paracas info eurusd +``` + +## Output Formats + +| Format | Extension | Description | +|--------|-----------|-------------| +| CSV | `.csv` | Comma-separated values | +| JSON | `.json` | JSON array | +| Parquet | `.parquet` | Apache Parquet columnar format | + +## Timeframes + +| Timeframe | Flag | Description | +|-----------|------|-------------| +| Tick | (default) | Raw tick data | +| 1 minute | `-t m1` | 1-minute OHLCV bars | +| 5 minutes | `-t m5` | 5-minute OHLCV bars | +| 15 minutes | `-t m15` | 15-minute OHLCV bars | +| 30 minutes | `-t m30` | 30-minute OHLCV bars | +| 1 hour | `-t h1` | 1-hour OHLCV bars | +| 4 hours | `-t h4` | 4-hour OHLCV bars | +| 1 day | `-t d1` | Daily OHLCV bars | + +## License + +MIT License - see [LICENSE](LICENSE) for details. diff --git a/bin/Cargo.toml b/bin/Cargo.toml new file mode 100644 index 0000000..7bd7a56 --- /dev/null +++ b/bin/Cargo.toml @@ -0,0 +1,33 @@ +[package] +name = "paracas" +description = "CLI for downloading Dukascopy tick data" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +homepage.workspace = true +repository.workspace = true +documentation.workspace = true +authors.workspace = true +keywords.workspace = true +categories.workspace = true + +[[bin]] +name = "paracas" +path = "src/main.rs" + +[lints] +workspace = true + +[features] +default = ["parquet"] +parquet = ["paracas-lib/parquet"] + +[dependencies] +paracas-lib = { workspace = true } +tokio = { workspace = true } +futures = { workspace = true } +clap = { workspace = true } +indicatif = { workspace = true } +anyhow = { workspace = true } +chrono = { workspace = true } diff --git a/bin/README.md b/bin/README.md new file mode 100644 index 0000000..a59b0ae --- /dev/null +++ b/bin/README.md @@ -0,0 +1,50 @@ +# paracas-cli + +Command-line interface for downloading Dukascopy tick data. + +## Installation + +```bash +cargo install paracas-cli +``` + +## Commands + +### Download + +Download tick data for an instrument: + +```bash +# Download EUR/USD ticks as CSV +paracas download -i eurusd -s 2024-01-01 -e 2024-01-31 -o data.csv + +# Download as Parquet with 1-hour aggregation +paracas download -i btcusd -s 2024-01-01 -e 2024-12-31 -o data.parquet -f parquet -t h1 +``` + +### List + +List available instruments: + +```bash +# List all instruments +paracas list + +# Filter by category +paracas list --category forex + +# Search +paracas list --search btc +``` + +### Info + +Show instrument details: + +```bash +paracas info eurusd +``` + +## License + +MIT License - see [LICENSE](../LICENSE) for details. diff --git a/bin/src/main.rs b/bin/src/main.rs new file mode 100644 index 0000000..137a3a5 --- /dev/null +++ b/bin/src/main.rs @@ -0,0 +1,390 @@ +//! paracas CLI - High-performance Dukascopy tick data downloader. + +use anyhow::{Context, Result, bail}; +use chrono::NaiveDate; +use clap::{Parser, Subcommand, ValueEnum}; +use futures::StreamExt; +use indicatif::{ProgressBar, ProgressStyle}; +use paracas_lib::prelude::*; +use std::fs::File; +use std::io::BufWriter; +use std::path::PathBuf; + +#[derive(Parser)] +#[command(name = "paracas")] +#[command(about = "High-performance Dukascopy tick data downloader", long_about = None)] +#[command(version)] +struct Cli { + #[command(subcommand)] + command: Commands, + + /// Verbosity level (-v, -vv, -vvv) + #[arg(short, long, action = clap::ArgAction::Count, global = true)] + verbose: u8, + + /// Quiet mode (suppress progress output) + #[arg(short, long, global = true)] + quiet: bool, +} + +#[derive(Subcommand)] +enum Commands { + /// Download tick data + Download { + /// Instrument identifier (e.g., eurusd, btcusd) + instrument: String, + + /// Start date (YYYY-MM-DD). Defaults to instrument's earliest available data. + #[arg(short, long)] + start: Option, + + /// End date (YYYY-MM-DD). Defaults to today. + #[arg(short, long)] + end: Option, + + /// Output file path. Defaults to . + #[arg(short, long)] + output: Option, + + /// Output format + #[arg(short, long, value_enum, default_value = "csv")] + format: Format, + + /// OHLCV aggregation timeframe (omit for raw ticks) + #[arg(short, long)] + timeframe: Option, + + /// Maximum concurrent downloads + #[arg(long, default_value = "32")] + concurrency: usize, + }, + + /// List available instruments + List { + /// Filter by category (forex, crypto, index, stock, commodity, etf, bond) + #[arg(short, long)] + category: Option, + + /// Search pattern + #[arg(short, long)] + search: Option, + }, + + /// Show instrument details + Info { + /// Instrument identifier + instrument: String, + }, +} + +#[derive(Clone, Copy, ValueEnum)] +enum Format { + Csv, + Json, + Ndjson, + Parquet, +} + +#[tokio::main] +async fn main() -> Result<()> { + let cli = Cli::parse(); + + match cli.command { + Commands::Download { + instrument, + start, + end, + output, + format, + timeframe, + concurrency, + } => { + download( + &instrument, + start.as_deref(), + end.as_deref(), + output, + format, + timeframe.as_deref(), + concurrency, + cli.quiet, + ) + .await + } + Commands::List { category, search } => { + list_instruments(category.as_deref(), search.as_deref()) + } + Commands::Info { instrument } => show_info(&instrument), + } +} + +#[allow(clippy::too_many_arguments)] +async fn download( + instrument_id: &str, + start_str: Option<&str>, + end_str: Option<&str>, + output: Option, + format: Format, + timeframe_str: Option<&str>, + concurrency: usize, + quiet: bool, +) -> Result<()> { + // Lookup instrument + let registry = InstrumentRegistry::global(); + let instrument = registry + .get(instrument_id) + .with_context(|| format!("Unknown instrument: {instrument_id}"))?; + + // Parse start date (default to instrument's earliest available data) + let start = match start_str { + Some(s) => NaiveDate::parse_from_str(s, "%Y-%m-%d") + .with_context(|| format!("Invalid start date: {s}"))?, + None => instrument + .start_tick_date() + .map(|dt| dt.date_naive()) + .unwrap_or_else(|| NaiveDate::from_ymd_opt(2003, 5, 5).expect("valid date")), + }; + + // Parse end date (default to today) + let end = match end_str { + Some(s) => NaiveDate::parse_from_str(s, "%Y-%m-%d") + .with_context(|| format!("Invalid end date: {s}"))?, + None => chrono::Utc::now().date_naive(), + }; + + let range = DateRange::new(start, end)?; + + // Determine output path (default to .) + let output = output.unwrap_or_else(|| { + let ext = match format { + Format::Csv => "csv", + Format::Json => "json", + Format::Ndjson => "ndjson", + Format::Parquet => "parquet", + }; + PathBuf::from(format!("{}.{}", instrument_id, ext)) + }); + + // Parse timeframe + let timeframe = match timeframe_str { + Some(tf) => tf + .parse::() + .map_err(|e| anyhow::anyhow!("{e}"))?, + None => Timeframe::Tick, + }; + + // Create client + let config = ClientConfig { + concurrency, + ..Default::default() + }; + let client = DownloadClient::new(config)?; + + // Setup progress bar + let total_hours = range.total_hours() as u64; + let progress = if quiet { + ProgressBar::hidden() + } else { + let pb = ProgressBar::new(total_hours); + pb.set_style( + ProgressStyle::default_bar() + .template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {pos}/{len} hours ({percent}%) {msg}") + .expect("Invalid progress template") + .progress_chars("=>-"), + ); + pb.set_message(format!("{} {} -> {}", instrument.id(), start, end)); + pb + }; + + // Download and collect ticks using the resilient stream + // This will retry on transient errors and skip hours that fail after retries + let mut all_ticks: Vec = Vec::new(); + let mut skipped_hours = 0u64; + let mut stream = paracas_lib::tick_stream_resilient(&client, instrument, range); + + while let Some(batch) = stream.next().await { + if batch.had_error() { + skipped_hours += 1; + } + all_ticks.extend(batch.ticks); + progress.inc(1); + } + + let finish_msg = if skipped_hours > 0 { + format!( + "Downloaded {} ticks ({} hours skipped due to errors)", + all_ticks.len(), + skipped_hours + ) + } else { + format!("Downloaded {} ticks", all_ticks.len()) + }; + progress.finish_with_message(finish_msg); + + // Aggregate if needed + if timeframe.is_tick() { + // Write raw ticks + write_ticks(&all_ticks, &output, format)?; + } else { + // Aggregate to OHLCV + let bars = aggregate_ticks(&all_ticks, timeframe); + write_ohlcv(&bars, &output, format)?; + } + + if !quiet { + println!("Output written to: {}", output.display()); + } + + Ok(()) +} + +fn aggregate_ticks(ticks: &[Tick], timeframe: Timeframe) -> Vec { + let mut aggregator = TickAggregator::new(timeframe); + let mut bars = Vec::new(); + + for tick in ticks { + if let Some(bar) = aggregator.process(*tick) { + bars.push(bar); + } + } + + if let Some(bar) = aggregator.finish() { + bars.push(bar); + } + + bars +} + +fn write_ticks(ticks: &[Tick], output: &PathBuf, format: Format) -> Result<()> { + let file = File::create(output)?; + let writer = BufWriter::new(file); + + match format { + Format::Csv => { + let formatter = CsvFormatter::new(); + formatter.write_ticks(ticks, writer)?; + } + Format::Json => { + let formatter = JsonFormatter::new(); + formatter.write_ticks(ticks, writer)?; + } + Format::Ndjson => { + let formatter = JsonFormatter::ndjson(); + formatter.write_ticks(ticks, writer)?; + } + Format::Parquet => { + #[cfg(feature = "parquet")] + { + let formatter = ParquetFormatter::new(); + formatter.write_ticks(ticks, writer)?; + } + #[cfg(not(feature = "parquet"))] + { + bail!("Parquet support not compiled in"); + } + } + } + + Ok(()) +} + +fn write_ohlcv(bars: &[Ohlcv], output: &PathBuf, format: Format) -> Result<()> { + let file = File::create(output)?; + let writer = BufWriter::new(file); + + match format { + Format::Csv => { + let formatter = CsvFormatter::new(); + formatter.write_ohlcv(bars, writer)?; + } + Format::Json => { + let formatter = JsonFormatter::new(); + formatter.write_ohlcv(bars, writer)?; + } + Format::Ndjson => { + let formatter = JsonFormatter::ndjson(); + formatter.write_ohlcv(bars, writer)?; + } + Format::Parquet => { + #[cfg(feature = "parquet")] + { + let formatter = ParquetFormatter::new(); + formatter.write_ohlcv(bars, writer)?; + } + #[cfg(not(feature = "parquet"))] + { + bail!("Parquet support not compiled in"); + } + } + } + + Ok(()) +} + +fn list_instruments(category: Option<&str>, search: Option<&str>) -> Result<()> { + let registry = InstrumentRegistry::global(); + + let instruments: Vec<_> = match (category, search) { + (Some(cat), _) => { + let category = parse_category(cat)?; + registry.by_category(category).collect() + } + (_, Some(pattern)) => registry.search(pattern), + (None, None) => registry.all().collect(), + }; + + if instruments.is_empty() { + println!("No instruments found."); + return Ok(()); + } + + println!("{:<15} {:<20} {:<10}", "ID", "NAME", "CATEGORY"); + println!("{}", "-".repeat(50)); + + for instrument in &instruments { + println!( + "{:<15} {:<20} {:<10}", + instrument.id(), + instrument.name(), + instrument.category() + ); + } + + println!("\nTotal: {} instruments", instruments.len()); + Ok(()) +} + +fn show_info(instrument_id: &str) -> Result<()> { + let registry = InstrumentRegistry::global(); + let instrument = registry + .get(instrument_id) + .with_context(|| format!("Unknown instrument: {instrument_id}"))?; + + println!("Instrument: {}", instrument.name()); + println!("ID: {}", instrument.id()); + println!("Category: {}", instrument.category()); + println!("Description: {}", instrument.description()); + println!("Decimal Factor: {}", instrument.decimal_factor()); + + if let Some(start) = instrument.start_tick_date() { + println!("Data Available From: {}", start.format("%Y-%m-%d")); + } + + Ok(()) +} + +fn parse_category(s: &str) -> Result { + match s.to_lowercase().as_str() { + "forex" => Ok(Category::Forex), + "crypto" => Ok(Category::Crypto), + "index" => Ok(Category::Index), + "stock" => Ok(Category::Stock), + "commodity" => Ok(Category::Commodity), + "etf" => Ok(Category::Etf), + "bond" => Ok(Category::Bond), + _ => bail!( + "Unknown category: {}. Valid options: forex, crypto, index, stock, commodity, etf, bond", + s + ), + } +} diff --git a/crates/paracas-aggregate/Cargo.toml b/crates/paracas-aggregate/Cargo.toml new file mode 100644 index 0000000..470832b --- /dev/null +++ b/crates/paracas-aggregate/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "paracas-aggregate" +description = "OHLCV aggregation for paracas tick data downloader" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +homepage.workspace = true +repository.workspace = true +documentation.workspace = true +authors.workspace = true +keywords.workspace = true +categories.workspace = true + +[lints] +workspace = true + +[package.metadata.docs.rs] +all-features = true +rustdoc-args = ["--cfg", "docsrs"] + +[dependencies] +paracas-types = { workspace = true } +chrono = { workspace = true } +serde = { workspace = true } diff --git a/crates/paracas-aggregate/README.md b/crates/paracas-aggregate/README.md new file mode 100644 index 0000000..375ff4f --- /dev/null +++ b/crates/paracas-aggregate/README.md @@ -0,0 +1,35 @@ +# paracas-aggregate + +OHLCV aggregation for the paracas tick data downloader. + +## Features + +- Tick-to-OHLCV aggregation +- Multiple timeframes (1s, 1m, 5m, 15m, 30m, 1h, 4h, 1d) +- Streaming aggregation for memory efficiency + +## Usage + +```rust,ignore +use paracas_aggregate::{Ohlcv, TickAggregator}; +use paracas_types::Timeframe; + +// Create an aggregator for 1-hour bars +let mut aggregator = TickAggregator::new(Timeframe::Hour1); + +// Process ticks +for tick in ticks { + if let Some(bar) = aggregator.process(tick) { + println!("Completed bar: {:?}", bar); + } +} + +// Get any remaining partial bar +if let Some(bar) = aggregator.finish() { + println!("Final bar: {:?}", bar); +} +``` + +## License + +MIT License - see [LICENSE](../../LICENSE) for details. diff --git a/crates/paracas-aggregate/src/aggregator.rs b/crates/paracas-aggregate/src/aggregator.rs new file mode 100644 index 0000000..dad1e14 --- /dev/null +++ b/crates/paracas-aggregate/src/aggregator.rs @@ -0,0 +1,240 @@ +//! Streaming tick-to-OHLCV aggregation. + +use chrono::{DateTime, Datelike, TimeZone, Timelike, Utc}; +use paracas_types::{Tick, Timeframe}; + +use crate::Ohlcv; + +/// Streaming tick aggregator. +/// +/// Aggregates ticks into OHLCV bars based on the configured timeframe. +#[derive(Debug)] +pub struct TickAggregator { + timeframe: Timeframe, + current_bar: Option, +} + +impl TickAggregator { + /// Creates a new aggregator for the given timeframe. + #[must_use] + pub const fn new(timeframe: Timeframe) -> Self { + Self { + timeframe, + current_bar: None, + } + } + + /// Returns the timeframe being aggregated to. + #[must_use] + pub const fn timeframe(&self) -> Timeframe { + self.timeframe + } + + /// Processes a tick, potentially emitting a completed bar. + /// + /// Returns `Some(bar)` when a bar is completed by this tick, + /// `None` otherwise. + pub fn process(&mut self, tick: Tick) -> Option { + let bar_start = self.bar_start_for(tick.timestamp); + + match self.current_bar.take() { + Some(mut builder) if builder.timestamp == bar_start => { + // Same bar, update it + builder.update(&tick); + self.current_bar = Some(builder); + None + } + Some(builder) => { + // New bar started, finish the old one + let completed = builder.finish(); + self.current_bar = Some(OhlcvBuilder::new(bar_start, &tick)); + Some(completed) + } + None => { + // First tick + self.current_bar = Some(OhlcvBuilder::new(bar_start, &tick)); + None + } + } + } + + /// Finishes aggregation, returning any remaining partial bar. + #[must_use] + pub fn finish(self) -> Option { + self.current_bar.map(|b| b.finish()) + } + + /// Calculates the bar start time for a given timestamp. + fn bar_start_for(&self, timestamp: DateTime) -> DateTime { + match self.timeframe { + Timeframe::Tick => timestamp, + Timeframe::Second1 => truncate_to_seconds(timestamp, 1), + Timeframe::Minute1 => truncate_to_minutes(timestamp, 1), + Timeframe::Minute5 => truncate_to_minutes(timestamp, 5), + Timeframe::Minute15 => truncate_to_minutes(timestamp, 15), + Timeframe::Minute30 => truncate_to_minutes(timestamp, 30), + Timeframe::Hour1 => truncate_to_hours(timestamp, 1), + Timeframe::Hour4 => truncate_to_hours(timestamp, 4), + Timeframe::Day1 => truncate_to_day(timestamp), + } + } +} + +/// Builder for OHLCV bars. +#[derive(Debug)] +struct OhlcvBuilder { + timestamp: DateTime, + open: f64, + high: f64, + low: f64, + close: f64, + volume: f64, + tick_count: u32, +} + +impl OhlcvBuilder { + /// Creates a new builder from the first tick. + fn new(timestamp: DateTime, tick: &Tick) -> Self { + let mid = tick.mid(); + let volume = f64::from(tick.total_volume()); + Self { + timestamp, + open: mid, + high: mid, + low: mid, + close: mid, + volume, + tick_count: 1, + } + } + + /// Updates the builder with a new tick. + fn update(&mut self, tick: &Tick) { + let mid = tick.mid(); + self.high = self.high.max(mid); + self.low = self.low.min(mid); + self.close = mid; + self.volume += f64::from(tick.total_volume()); + self.tick_count += 1; + } + + /// Finishes building and returns the OHLCV bar. + const fn finish(self) -> Ohlcv { + Ohlcv::new( + self.timestamp, + self.open, + self.high, + self.low, + self.close, + self.volume, + self.tick_count, + ) + } +} + +/// Truncates a timestamp to the start of a second boundary. +fn truncate_to_seconds(dt: DateTime, interval: u32) -> DateTime { + let second = dt.second() / interval * interval; + Utc.with_ymd_and_hms( + dt.year(), + dt.month(), + dt.day(), + dt.hour(), + dt.minute(), + second, + ) + .unwrap() +} + +/// Truncates a timestamp to the start of a minute boundary. +fn truncate_to_minutes(dt: DateTime, interval: u32) -> DateTime { + let minute = dt.minute() / interval * interval; + Utc.with_ymd_and_hms(dt.year(), dt.month(), dt.day(), dt.hour(), minute, 0) + .unwrap() +} + +/// Truncates a timestamp to the start of an hour boundary. +fn truncate_to_hours(dt: DateTime, interval: u32) -> DateTime { + let hour = dt.hour() / interval * interval; + Utc.with_ymd_and_hms(dt.year(), dt.month(), dt.day(), hour, 0, 0) + .unwrap() +} + +/// Truncates a timestamp to the start of the day. +fn truncate_to_day(dt: DateTime) -> DateTime { + Utc.with_ymd_and_hms(dt.year(), dt.month(), dt.day(), 0, 0, 0) + .unwrap() +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::TimeDelta; + + fn make_tick(hour: u32, minute: u32, second: u32, millis: u32, ask: f64, bid: f64) -> Tick { + let timestamp = Utc + .with_ymd_and_hms(2024, 1, 1, hour, minute, second) + .unwrap() + + TimeDelta::milliseconds(i64::from(millis)); + Tick::new(timestamp, ask, bid, 100.0, 100.0) + } + + #[test] + fn test_minute_aggregation() { + let mut agg = TickAggregator::new(Timeframe::Minute1); + + // First tick at 12:00:00 + let tick1 = make_tick(12, 0, 0, 0, 1.1001, 1.1000); + assert!(agg.process(tick1).is_none()); + + // Second tick at 12:00:30 (same minute) + let tick2 = make_tick(12, 0, 30, 0, 1.1010, 1.1005); + assert!(agg.process(tick2).is_none()); + + // Third tick at 12:01:00 (new minute, completes first bar) + let tick3 = make_tick(12, 1, 0, 0, 1.0990, 1.0985); + let bar = agg.process(tick3).unwrap(); + + assert_eq!(bar.tick_count, 2); + assert!((bar.open - 1.10005).abs() < 1e-10); // mid of first tick + assert!((bar.close - 1.10075).abs() < 1e-10); // mid of second tick + } + + #[test] + fn test_hour_aggregation() { + let mut agg = TickAggregator::new(Timeframe::Hour1); + + let tick1 = make_tick(12, 0, 0, 0, 1.1001, 1.1000); + assert!(agg.process(tick1).is_none()); + + let tick2 = make_tick(12, 30, 0, 0, 1.1050, 1.1045); + assert!(agg.process(tick2).is_none()); + + let tick3 = make_tick(13, 0, 0, 0, 1.0990, 1.0985); + let bar = agg.process(tick3).unwrap(); + + assert_eq!(bar.tick_count, 2); + assert_eq!(bar.timestamp.hour(), 12); + } + + #[test] + fn test_finish() { + let mut agg = TickAggregator::new(Timeframe::Hour1); + + let tick1 = make_tick(12, 0, 0, 0, 1.1001, 1.1000); + agg.process(tick1); + + let bar = agg.finish().unwrap(); + assert_eq!(bar.tick_count, 1); + } + + #[test] + fn test_truncate_functions() { + let dt = Utc.with_ymd_and_hms(2024, 1, 15, 14, 37, 45).unwrap(); + + assert_eq!(truncate_to_minutes(dt, 5).minute(), 35); + assert_eq!(truncate_to_minutes(dt, 15).minute(), 30); + assert_eq!(truncate_to_hours(dt, 4).hour(), 12); + assert_eq!(truncate_to_day(dt).hour(), 0); + } +} diff --git a/crates/paracas-aggregate/src/lib.rs b/crates/paracas-aggregate/src/lib.rs new file mode 100644 index 0000000..4a9ba3d --- /dev/null +++ b/crates/paracas-aggregate/src/lib.rs @@ -0,0 +1,18 @@ +//! OHLCV aggregation for paracas tick data downloader. +//! +//! This crate provides tick-to-OHLCV (candlestick) aggregation: +//! +//! - [`Ohlcv`] - OHLCV bar data structure +//! - [`TickAggregator`] - Streaming tick aggregator + +#![doc = include_str!("../README.md")] +#![doc(issue_tracker_base_url = "https://github.com/factordynamics/paracas/issues/")] +#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))] +#![warn(missing_docs)] +#![forbid(unsafe_code)] + +mod aggregator; +mod ohlcv; + +pub use aggregator::TickAggregator; +pub use ohlcv::Ohlcv; diff --git a/crates/paracas-aggregate/src/ohlcv.rs b/crates/paracas-aggregate/src/ohlcv.rs new file mode 100644 index 0000000..8157807 --- /dev/null +++ b/crates/paracas-aggregate/src/ohlcv.rs @@ -0,0 +1,128 @@ +//! OHLCV (candlestick) data structure. + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +/// OHLCV bar (candlestick) data. +#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)] +pub struct Ohlcv { + /// Bar open time (start of the period). + pub timestamp: DateTime, + /// Opening price (first tick's mid price). + pub open: f64, + /// Highest price during the period. + pub high: f64, + /// Lowest price during the period. + pub low: f64, + /// Closing price (last tick's mid price). + pub close: f64, + /// Total volume (sum of ask + bid volumes). + pub volume: f64, + /// Number of ticks in the bar. + pub tick_count: u32, +} + +impl Ohlcv { + /// Creates a new OHLCV bar. + #[must_use] + pub const fn new( + timestamp: DateTime, + open: f64, + high: f64, + low: f64, + close: f64, + volume: f64, + tick_count: u32, + ) -> Self { + Self { + timestamp, + open, + high, + low, + close, + volume, + tick_count, + } + } + + /// Returns the price range (high - low). + #[must_use] + pub fn range(&self) -> f64 { + self.high - self.low + } + + /// Returns the body size (|close - open|). + #[must_use] + pub fn body(&self) -> f64 { + (self.close - self.open).abs() + } + + /// Returns true if this is a bullish (green) bar. + #[must_use] + pub fn is_bullish(&self) -> bool { + self.close > self.open + } + + /// Returns true if this is a bearish (red) bar. + #[must_use] + pub fn is_bearish(&self) -> bool { + self.close < self.open + } + + /// Returns the typical price ((high + low + close) / 3). + #[must_use] + pub fn typical_price(&self) -> f64 { + (self.high + self.low + self.close) / 3.0 + } + + /// Returns the weighted close ((high + low + 2*close) / 4). + #[must_use] + pub fn weighted_close(&self) -> f64 { + (self.high + self.low + 2.0 * self.close) / 4.0 + } +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::TimeZone; + + fn create_test_bar() -> Ohlcv { + let timestamp = Utc.with_ymd_and_hms(2024, 1, 1, 12, 0, 0).unwrap(); + Ohlcv::new(timestamp, 1.1000, 1.1050, 1.0980, 1.1020, 1000.0, 500) + } + + #[test] + fn test_range() { + let bar = create_test_bar(); + assert!((bar.range() - 0.0070).abs() < 1e-10); + } + + #[test] + fn test_body() { + let bar = create_test_bar(); + assert!((bar.body() - 0.0020).abs() < 1e-10); + } + + #[test] + fn test_bullish() { + let bar = create_test_bar(); + assert!(bar.is_bullish()); + assert!(!bar.is_bearish()); + } + + #[test] + fn test_bearish() { + let timestamp = Utc.with_ymd_and_hms(2024, 1, 1, 12, 0, 0).unwrap(); + let bar = Ohlcv::new(timestamp, 1.1020, 1.1050, 1.0980, 1.1000, 1000.0, 500); + assert!(!bar.is_bullish()); + assert!(bar.is_bearish()); + } + + #[test] + fn test_typical_price() { + let bar = create_test_bar(); + let expected = (1.1050 + 1.0980 + 1.1020) / 3.0; + assert!((bar.typical_price() - expected).abs() < 1e-10); + } +} diff --git a/crates/paracas-fetch/Cargo.toml b/crates/paracas-fetch/Cargo.toml new file mode 100644 index 0000000..675cf80 --- /dev/null +++ b/crates/paracas-fetch/Cargo.toml @@ -0,0 +1,34 @@ +[package] +name = "paracas-fetch" +description = "HTTP client and data fetching for paracas tick data downloader" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +homepage.workspace = true +repository.workspace = true +documentation.workspace = true +authors.workspace = true +keywords.workspace = true +categories.workspace = true + +[lints] +workspace = true + +[package.metadata.docs.rs] +all-features = true +rustdoc-args = ["--cfg", "docsrs"] + +[dependencies] +paracas-types = { workspace = true } +tokio = { workspace = true } +futures = { workspace = true } +reqwest = { workspace = true } +bytes = { workspace = true } +lzma-rs = { workspace = true } +byteorder = { workspace = true } +chrono = { workspace = true } +thiserror = { workspace = true } + +[dev-dependencies] +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/crates/paracas-fetch/README.md b/crates/paracas-fetch/README.md new file mode 100644 index 0000000..58b0fdf --- /dev/null +++ b/crates/paracas-fetch/README.md @@ -0,0 +1,52 @@ +# paracas-fetch + +HTTP client and data fetching for the paracas tick data downloader. + +## Features + +- Concurrent HTTP downloads with connection pooling +- LZMA decompression for bi5 files +- Binary tick data parsing +- Streaming async API with backpressure + +## Architecture + +The fetch pipeline consists of: + +1. **URL Builder** - Constructs Dukascopy data URLs +2. **HTTP Client** - Downloads bi5 files with retries +3. **Decompressor** - LZMA decompression +4. **Parser** - Binary tick data parsing + +## Usage + +```rust,ignore +use paracas_fetch::{DownloadClient, ClientConfig, tick_stream}; +use paracas_instruments::InstrumentRegistry; +use paracas_types::DateRange; +use futures::StreamExt; + +#[tokio::main] +async fn main() { + let client = DownloadClient::new(ClientConfig::default()).unwrap(); + let registry = InstrumentRegistry::global(); + let instrument = registry.get("eurusd").unwrap(); + + let range = DateRange::new( + chrono::NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(), + chrono::NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(), + ).unwrap(); + + let mut stream = tick_stream(&client, instrument, range); + while let Some(result) = stream.next().await { + match result { + Ok(tick) => println!("{:?}", tick), + Err(e) => eprintln!("Error: {}", e), + } + } +} +``` + +## License + +MIT License - see [LICENSE](../../LICENSE) for details. diff --git a/crates/paracas-fetch/src/client.rs b/crates/paracas-fetch/src/client.rs new file mode 100644 index 0000000..1238682 --- /dev/null +++ b/crates/paracas-fetch/src/client.rs @@ -0,0 +1,214 @@ +//! HTTP client for downloading bi5 files. + +use bytes::Bytes; +use reqwest::Client; +use std::time::Duration; +use thiserror::Error; + +/// Configuration for the download client. +#[derive(Debug, Clone)] +pub struct ClientConfig { + /// Maximum concurrent downloads. + pub concurrency: usize, + /// Request timeout. + pub timeout: Duration, + /// Maximum retry attempts for failed requests. + pub max_retries: u32, + /// Base delay for exponential backoff (in milliseconds). + pub base_delay_ms: u64, + /// Maximum delay between retries (in milliseconds). + pub max_delay_ms: u64, + /// User agent string. + pub user_agent: String, +} + +impl Default for ClientConfig { + fn default() -> Self { + Self { + concurrency: 10, // Lower concurrency to avoid overwhelming the server + timeout: Duration::from_secs(60), + max_retries: 10, // More retries for transient failures + base_delay_ms: 500, // Start with 500ms delay + max_delay_ms: 30_000, // Max 30 seconds between retries + user_agent: format!("paracas/{}", env!("CARGO_PKG_VERSION")), + } + } +} + +/// Errors that can occur during downloads. +#[derive(Error, Debug)] +pub enum DownloadError { + /// HTTP request failed. + #[error("HTTP error: {0}")] + Http(#[from] reqwest::Error), + + /// Request timed out. + #[error("Request timed out after {0} attempts")] + Timeout(u32), + + /// Server returned an error status. + #[error("Server error: {status}")] + ServerError { + /// HTTP status code. + status: u16, + }, +} + +/// HTTP client with connection pooling and retry logic. +#[derive(Debug, Clone)] +pub struct DownloadClient { + client: Client, + config: ClientConfig, +} + +impl DownloadClient { + /// Creates a new download client with the given configuration. + /// + /// # Errors + /// + /// Returns an error if the HTTP client cannot be created. + pub fn new(config: ClientConfig) -> Result { + let client = Client::builder() + .pool_max_idle_per_host(config.concurrency) + .timeout(config.timeout) + .user_agent(&config.user_agent) + .gzip(true) + .build()?; + Ok(Self { client, config }) + } + + /// Creates a client with default configuration. + /// + /// # Errors + /// + /// Returns an error if the HTTP client cannot be created. + pub fn with_defaults() -> Result { + Self::new(ClientConfig::default()) + } + + /// Returns the client configuration. + #[must_use] + pub const fn config(&self) -> &ClientConfig { + &self.config + } + + /// Downloads a single bi5 file, returning the compressed bytes. + /// + /// Returns `Ok(None)` if the file does not exist (404). + /// + /// # Errors + /// + /// Returns an error if the download fails after all retries. + pub async fn download(&self, url: &str) -> Result, DownloadError> { + let mut attempts = 0; + + loop { + match self.client.get(url).send().await { + Ok(response) => { + if response.status() == reqwest::StatusCode::NOT_FOUND { + return Ok(None); // No data for this hour + } + + // Retry on server errors (5xx) and rate limiting (429) + if response.status().is_server_error() + || response.status() == reqwest::StatusCode::TOO_MANY_REQUESTS + { + if attempts < self.config.max_retries { + attempts += 1; + let delay = self.calculate_backoff_delay(attempts); + tokio::time::sleep(delay).await; + continue; + } + return Err(DownloadError::ServerError { + status: response.status().as_u16(), + }); + } + + response.error_for_status_ref()?; + return Ok(Some(response.bytes().await?)); + } + Err(e) if self.is_retryable_error(&e) && attempts < self.config.max_retries => { + attempts += 1; + let delay = self.calculate_backoff_delay(attempts); + tokio::time::sleep(delay).await; + } + Err(e) => return Err(e.into()), + } + } + } + + /// Calculates the backoff delay with exponential backoff and jitter. + fn calculate_backoff_delay(&self, attempt: u32) -> Duration { + // Exponential backoff: base_delay * 2^attempt + let exp_delay = self + .config + .base_delay_ms + .saturating_mul(1u64 << attempt.min(10)); + + // Cap at max delay + let capped_delay = exp_delay.min(self.config.max_delay_ms); + + // Add jitter (±25%) + let jitter_range = capped_delay / 4; + let jitter = if jitter_range > 0 { + // Simple deterministic jitter based on attempt number + // This avoids needing a random number generator + let jitter_offset = (attempt as u64 * 17) % (jitter_range * 2); + jitter_offset.saturating_sub(jitter_range) + } else { + 0 + }; + + let final_delay = (capped_delay as i64 + jitter as i64).max(100) as u64; + Duration::from_millis(final_delay) + } + + /// Determines if an error is retryable. + fn is_retryable_error(&self, error: &reqwest::Error) -> bool { + // Don't retry builder errors (configuration issues) + if error.is_builder() { + return false; + } + + // Retry on timeouts, connection errors, and request errors + error.is_timeout() || error.is_connect() || error.is_request() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_client_config_default() { + let config = ClientConfig::default(); + assert_eq!(config.concurrency, 10); + assert_eq!(config.max_retries, 10); + assert_eq!(config.timeout, Duration::from_secs(60)); + assert_eq!(config.base_delay_ms, 500); + assert_eq!(config.max_delay_ms, 30_000); + } + + #[tokio::test] + async fn test_client_creation() { + let client = DownloadClient::with_defaults(); + assert!(client.is_ok()); + } + + #[test] + fn test_backoff_delay_calculation() { + let client = DownloadClient::with_defaults().unwrap(); + + // First attempt: base_delay * 2 = 1000ms (plus jitter) + let delay1 = client.calculate_backoff_delay(1); + assert!(delay1.as_millis() >= 750 && delay1.as_millis() <= 1250); + + // Second attempt: base_delay * 4 = 2000ms (plus jitter) + let delay2 = client.calculate_backoff_delay(2); + assert!(delay2.as_millis() >= 1500 && delay2.as_millis() <= 2500); + + // High attempt should be capped at max_delay + let delay_high = client.calculate_backoff_delay(20); + assert!(delay_high.as_millis() <= 37500); // max_delay + 25% jitter + } +} diff --git a/crates/paracas-fetch/src/decompress.rs b/crates/paracas-fetch/src/decompress.rs new file mode 100644 index 0000000..0171fe3 --- /dev/null +++ b/crates/paracas-fetch/src/decompress.rs @@ -0,0 +1,64 @@ +//! LZMA decompression for bi5 files. + +use lzma_rs::lzma_decompress; +use std::io::{BufReader, Cursor}; +use thiserror::Error; + +/// Errors that can occur during decompression. +#[derive(Error, Debug)] +pub enum DecompressError { + /// LZMA decompression failed. + #[error("LZMA decompression failed: {0}")] + LzmaError(String), + + /// Empty input data. + #[error("Empty input data")] + EmptyInput, +} + +/// Decompresses LZMA-compressed bi5 data. +/// +/// Dukascopy bi5 files are LZMA-compressed binary data containing tick records. +/// +/// # Errors +/// +/// Returns an error if decompression fails. +/// +/// # Example +/// +/// ```ignore +/// use paracas_fetch::decompress_bi5; +/// +/// let compressed = /* bi5 data from HTTP */; +/// let decompressed = decompress_bi5(&compressed)?; +/// ``` +pub fn decompress_bi5(compressed: &[u8]) -> Result, DecompressError> { + if compressed.is_empty() { + return Err(DecompressError::EmptyInput); + } + + let mut decompressed = Vec::new(); + let mut reader = BufReader::new(Cursor::new(compressed)); + + lzma_decompress(&mut reader, &mut decompressed) + .map_err(|e| DecompressError::LzmaError(e.to_string()))?; + + Ok(decompressed) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_empty_input() { + let result = decompress_bi5(&[]); + assert!(matches!(result, Err(DecompressError::EmptyInput))); + } + + #[test] + fn test_invalid_lzma() { + let result = decompress_bi5(&[0x00, 0x01, 0x02, 0x03]); + assert!(matches!(result, Err(DecompressError::LzmaError(_)))); + } +} diff --git a/crates/paracas-fetch/src/lib.rs b/crates/paracas-fetch/src/lib.rs new file mode 100644 index 0000000..a22a33e --- /dev/null +++ b/crates/paracas-fetch/src/lib.rs @@ -0,0 +1,26 @@ +//! HTTP client and data fetching for paracas tick data downloader. +//! +//! This crate provides the data download pipeline: +//! +//! - [`url::tick_url`] - Constructs Dukascopy data URLs +//! - [`DownloadClient`] - HTTP client with connection pooling and retries +//! - [`decompress::decompress_bi5`] - LZMA decompression +//! - [`parse::parse_ticks`] - Binary tick data parsing +//! - [`tick_stream`] - Async streaming tick download + +#![doc = include_str!("../README.md")] +#![doc(issue_tracker_base_url = "https://github.com/factordynamics/paracas/issues/")] +#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))] +#![warn(missing_docs)] +#![forbid(unsafe_code)] + +mod client; +mod decompress; +mod parse; +mod stream; +pub mod url; + +pub use client::{ClientConfig, DownloadClient, DownloadError}; +pub use decompress::{DecompressError, decompress_bi5}; +pub use parse::{ParseError, parse_ticks, tick_count}; +pub use stream::{TickBatch, flatten_ticks, tick_stream, tick_stream_resilient}; diff --git a/crates/paracas-fetch/src/parse.rs b/crates/paracas-fetch/src/parse.rs new file mode 100644 index 0000000..dbddd6c --- /dev/null +++ b/crates/paracas-fetch/src/parse.rs @@ -0,0 +1,122 @@ +//! Binary tick parsing from bi5 format. + +use byteorder::{BigEndian, ByteOrder}; +use paracas_types::RawTick; +use thiserror::Error; + +/// Errors that can occur during tick parsing. +#[derive(Error, Debug, Clone, PartialEq, Eq)] +pub enum ParseError { + /// Invalid data length (not a multiple of tick size). + #[error("Invalid data length: {0} bytes (expected multiple of {1})")] + InvalidLength(usize, usize), + + /// Incomplete tick record. + #[error("Incomplete tick record at offset {0}")] + IncompleteRecord(usize), +} + +/// Parses raw ticks from decompressed bi5 data. +/// +/// The bi5 format stores ticks as 20 bytes in big-endian order: +/// - `u32`: milliseconds offset from hour start (bytes 0-3) +/// - `u32`: ask price raw (bytes 4-7) +/// - `u32`: bid price raw (bytes 8-11) +/// - `f32`: ask volume (bytes 12-15) +/// - `f32`: bid volume (bytes 16-19) +/// +/// # Arguments +/// +/// * `data` - Decompressed bi5 data +/// +/// # Returns +/// +/// An iterator over parsed raw ticks. +/// +/// # Errors +/// +/// Returns an error if the data length is invalid. +pub fn parse_ticks(data: &[u8]) -> Result + '_, ParseError> { + if !data.len().is_multiple_of(RawTick::SIZE) { + return Err(ParseError::InvalidLength(data.len(), RawTick::SIZE)); + } + + Ok(data.chunks_exact(RawTick::SIZE).map(parse_single_tick)) +} + +/// Parses a single tick from a 20-byte chunk. +#[inline] +fn parse_single_tick(data: &[u8]) -> RawTick { + RawTick::new( + BigEndian::read_u32(&data[0..4]), + BigEndian::read_u32(&data[4..8]), + BigEndian::read_u32(&data[8..12]), + BigEndian::read_f32(&data[12..16]), + BigEndian::read_f32(&data[16..20]), + ) +} + +/// Returns the number of ticks in the given data. +#[must_use] +pub const fn tick_count(data_len: usize) -> usize { + data_len / RawTick::SIZE +} + +#[cfg(test)] +mod tests { + use super::*; + + fn create_test_tick_bytes(ms: u32, ask: u32, bid: u32, ask_vol: f32, bid_vol: f32) -> Vec { + let mut bytes = vec![0u8; 20]; + BigEndian::write_u32(&mut bytes[0..4], ms); + BigEndian::write_u32(&mut bytes[4..8], ask); + BigEndian::write_u32(&mut bytes[8..12], bid); + BigEndian::write_f32(&mut bytes[12..16], ask_vol); + BigEndian::write_f32(&mut bytes[16..20], bid_vol); + bytes + } + + #[test] + fn test_parse_single_tick() { + let bytes = create_test_tick_bytes(1000, 112345, 112340, 100.0, 200.0); + let tick = parse_single_tick(&bytes); + + assert_eq!(tick.ms_offset, 1000); + assert_eq!(tick.ask_raw, 112345); + assert_eq!(tick.bid_raw, 112340); + assert!((tick.ask_volume - 100.0).abs() < 0.001); + assert!((tick.bid_volume - 200.0).abs() < 0.001); + } + + #[test] + fn test_parse_multiple_ticks() { + let mut data = create_test_tick_bytes(0, 100, 99, 10.0, 20.0); + data.extend(create_test_tick_bytes(1000, 101, 100, 15.0, 25.0)); + + let ticks: Vec<_> = parse_ticks(&data).unwrap().collect(); + assert_eq!(ticks.len(), 2); + assert_eq!(ticks[0].ms_offset, 0); + assert_eq!(ticks[1].ms_offset, 1000); + } + + #[test] + fn test_invalid_length() { + let data = vec![0u8; 25]; // Not a multiple of 20 + let result = parse_ticks(&data); + assert!(matches!(result, Err(ParseError::InvalidLength(25, 20)))); + } + + #[test] + fn test_empty_data() { + let data: Vec = vec![]; + let ticks: Vec<_> = parse_ticks(&data).unwrap().collect(); + assert!(ticks.is_empty()); + } + + #[test] + fn test_tick_count() { + assert_eq!(tick_count(0), 0); + assert_eq!(tick_count(20), 1); + assert_eq!(tick_count(200), 10); + } +} diff --git a/crates/paracas-fetch/src/stream.rs b/crates/paracas-fetch/src/stream.rs new file mode 100644 index 0000000..3199791 --- /dev/null +++ b/crates/paracas-fetch/src/stream.rs @@ -0,0 +1,225 @@ +//! Streaming tick download pipeline. + +use chrono::{DateTime, Utc}; +use futures::stream::{self, Stream, StreamExt}; +use paracas_types::{DateRange, Instrument, ParacasError, Tick}; + +use crate::{DownloadClient, decompress_bi5, parse_ticks, url::tick_url}; + +/// A batch of ticks from a single hour. +#[derive(Debug, Clone)] +pub struct TickBatch { + /// The hour start timestamp. + pub hour: DateTime, + /// The ticks in this batch. + pub ticks: Vec, + /// Whether this batch had an error that was skipped. + pub had_error: bool, +} + +impl TickBatch { + /// Creates a new tick batch. + #[must_use] + pub const fn new(hour: DateTime, ticks: Vec) -> Self { + Self { + hour, + ticks, + had_error: false, + } + } + + /// Creates a new tick batch that represents a skipped error. + #[must_use] + pub const fn skipped_error(hour: DateTime) -> Self { + Self { + hour, + ticks: Vec::new(), + had_error: true, + } + } + + /// Returns true if the batch is empty. + #[must_use] + pub const fn is_empty(&self) -> bool { + self.ticks.is_empty() + } + + /// Returns the number of ticks in the batch. + #[must_use] + pub const fn len(&self) -> usize { + self.ticks.len() + } + + /// Returns true if this batch had an error that was skipped. + #[must_use] + pub const fn had_error(&self) -> bool { + self.had_error + } +} + +/// Creates an async stream of tick batches for the given instrument and date range. +/// +/// This function downloads, decompresses, and parses tick data concurrently +/// using the configured number of parallel connections. +/// +/// # Arguments +/// +/// * `client` - The HTTP client to use for downloads +/// * `instrument` - The instrument to download data for +/// * `range` - The date range to download +/// +/// # Returns +/// +/// An async stream of tick batches, one per hour. +pub fn tick_stream<'a>( + client: &'a DownloadClient, + instrument: &'a Instrument, + range: DateRange, +) -> impl Stream> + 'a { + let decimal_factor = instrument.decimal_factor_f64(); + let instrument_id = instrument.id().to_string(); + let concurrency = client.config().concurrency; + + stream::iter(range.hours()) + .map(move |hour| { + let url = tick_url(&instrument_id, hour); + let client = client.clone(); + async move { (hour, client.download(&url).await) } + }) + .buffer_unordered(concurrency) + .map(move |(hour, result)| process_download_result(hour, result, decimal_factor)) +} + +/// Processes a download result into a tick batch. +fn process_download_result( + hour: DateTime, + result: Result, crate::DownloadError>, + decimal_factor: f64, +) -> Result { + match result { + Ok(Some(compressed)) => { + let decompressed = + decompress_bi5(&compressed).map_err(|e| ParacasError::Decompress(e.to_string()))?; + + let ticks: Vec = parse_ticks(&decompressed) + .map_err(|e| ParacasError::Parse(e.to_string()))? + .map(|raw| raw.normalize(hour, decimal_factor)) + .collect(); + + Ok(TickBatch::new(hour, ticks)) + } + Ok(None) => { + // No data for this hour + Ok(TickBatch::new(hour, Vec::new())) + } + Err(e) => Err(ParacasError::Http(e.to_string())), + } +} + +/// Creates a resilient async stream that skips failed hours instead of failing entirely. +/// +/// This is useful for long-running downloads where occasional server errors +/// should not abort the entire operation. +/// +/// # Arguments +/// +/// * `client` - The HTTP client to use for downloads +/// * `instrument` - The instrument to download data for +/// * `range` - The date range to download +/// +/// # Returns +/// +/// An async stream of tick batches. Failed hours are returned as empty batches +/// with `had_error` set to true. +pub fn tick_stream_resilient<'a>( + client: &'a DownloadClient, + instrument: &'a Instrument, + range: DateRange, +) -> impl Stream + 'a { + let decimal_factor = instrument.decimal_factor_f64(); + let instrument_id = instrument.id().to_string(); + let concurrency = client.config().concurrency; + + stream::iter(range.hours()) + .map(move |hour| { + let url = tick_url(&instrument_id, hour); + let client = client.clone(); + async move { (hour, client.download(&url).await) } + }) + .buffer_unordered(concurrency) + .map(move |(hour, result)| process_download_result_resilient(hour, result, decimal_factor)) +} + +/// Processes a download result into a tick batch, skipping errors. +#[allow(clippy::option_if_let_else)] // Nested matches are more readable here +fn process_download_result_resilient( + hour: DateTime, + result: Result, crate::DownloadError>, + decimal_factor: f64, +) -> TickBatch { + match result { + Ok(Some(compressed)) => { + match decompress_bi5(&compressed) { + Ok(decompressed) => match parse_ticks(&decompressed) { + Ok(raw_ticks) => { + let ticks: Vec = raw_ticks + .map(|raw| raw.normalize(hour, decimal_factor)) + .collect(); + TickBatch::new(hour, ticks) + } + Err(_) => { + // Parse error - return empty batch with error flag + TickBatch::skipped_error(hour) + } + }, + Err(_) => { + // Decompression error - return empty batch with error flag + TickBatch::skipped_error(hour) + } + } + } + Ok(None) => { + // No data for this hour + TickBatch::new(hour, Vec::new()) + } + Err(_) => { + // HTTP error - return empty batch with error flag + TickBatch::skipped_error(hour) + } + } +} + +/// Flattens a tick batch stream into individual ticks. +/// +/// This is useful when you want to process ticks one at a time rather than +/// in batches. +pub fn flatten_ticks( + batch_stream: impl Stream>, +) -> impl Stream> { + batch_stream.flat_map(|result| match result { + Ok(batch) => stream::iter(batch.ticks.into_iter().map(Ok)).left_stream(), + Err(e) => stream::once(async move { Err(e) }).right_stream(), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_tick_batch_new() { + let hour = Utc::now(); + let batch = TickBatch::new(hour, vec![]); + assert!(batch.is_empty()); + assert_eq!(batch.len(), 0); + assert!(!batch.had_error()); + } + + #[test] + fn test_tick_batch_skipped_error() { + let hour = Utc::now(); + let batch = TickBatch::skipped_error(hour); + assert!(batch.is_empty()); + assert!(batch.had_error()); + } +} diff --git a/crates/paracas-fetch/src/url.rs b/crates/paracas-fetch/src/url.rs new file mode 100644 index 0000000..1a88f08 --- /dev/null +++ b/crates/paracas-fetch/src/url.rs @@ -0,0 +1,68 @@ +//! Dukascopy URL construction. + +use chrono::{DateTime, Datelike, Timelike, Utc}; + +/// Base URL for Dukascopy data feed. +pub const BASE_URL: &str = "https://datafeed.dukascopy.com/datafeed"; + +/// Builds the URL for a specific hour's tick data. +/// +/// URL format: `{BASE_URL}/{INSTRUMENT}/{YEAR}/{MONTH}/{DAY}/{HOUR}h_ticks.bi5` +/// +/// Note: Dukascopy uses 0-indexed months (January = 00). +/// +/// # Example +/// +/// ``` +/// use paracas_fetch::url::tick_url; +/// use chrono::{TimeZone, Utc}; +/// +/// let hour = Utc.with_ymd_and_hms(2024, 1, 15, 12, 0, 0).unwrap(); +/// let url = tick_url("eurusd", hour); +/// assert_eq!(url, "https://datafeed.dukascopy.com/datafeed/EURUSD/2024/00/15/12h_ticks.bi5"); +/// ``` +#[must_use] +pub fn tick_url(instrument: &str, hour: DateTime) -> String { + format!( + "{}/{}/{}/{:02}/{:02}/{:02}h_ticks.bi5", + BASE_URL, + instrument.to_uppercase(), + hour.year(), + hour.month() - 1, // Dukascopy uses 0-indexed months + hour.day(), + hour.hour() + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::TimeZone; + + #[test] + fn test_tick_url_january() { + let hour = Utc.with_ymd_and_hms(2024, 1, 15, 12, 0, 0).unwrap(); + let url = tick_url("eurusd", hour); + assert_eq!( + url, + "https://datafeed.dukascopy.com/datafeed/EURUSD/2024/00/15/12h_ticks.bi5" + ); + } + + #[test] + fn test_tick_url_december() { + let hour = Utc.with_ymd_and_hms(2024, 12, 31, 23, 0, 0).unwrap(); + let url = tick_url("btcusd", hour); + assert_eq!( + url, + "https://datafeed.dukascopy.com/datafeed/BTCUSD/2024/11/31/23h_ticks.bi5" + ); + } + + #[test] + fn test_tick_url_uppercase() { + let hour = Utc.with_ymd_and_hms(2024, 6, 1, 0, 0, 0).unwrap(); + let url = tick_url("GBPJPY", hour); + assert!(url.contains("GBPJPY")); + } +} diff --git a/crates/paracas-format/Cargo.toml b/crates/paracas-format/Cargo.toml new file mode 100644 index 0000000..5c7004e --- /dev/null +++ b/crates/paracas-format/Cargo.toml @@ -0,0 +1,39 @@ +[package] +name = "paracas-format" +description = "Output formatters for paracas tick data downloader" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +homepage.workspace = true +repository.workspace = true +documentation.workspace = true +authors.workspace = true +keywords.workspace = true +categories.workspace = true + +[lints] +workspace = true + +[package.metadata.docs.rs] +all-features = true +rustdoc-args = ["--cfg", "docsrs"] + +[features] +default = ["csv", "json", "parquet"] +csv = [] +json = [] +parquet = ["dep:arrow", "dep:parquet"] + +[dependencies] +paracas-types = { workspace = true } +paracas-aggregate = { workspace = true } +chrono = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } +arrow = { workspace = true, optional = true } +parquet = { workspace = true, optional = true } + +[dev-dependencies] +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/crates/paracas-format/README.md b/crates/paracas-format/README.md new file mode 100644 index 0000000..09bb875 --- /dev/null +++ b/crates/paracas-format/README.md @@ -0,0 +1,36 @@ +# paracas-format + +Output formatters for the paracas tick data downloader. + +## Supported Formats + +- **CSV** - Comma-separated values +- **JSON** - JSON array or newline-delimited JSON (NDJSON) +- **Parquet** - Apache Parquet columnar format (requires `parquet` feature) + +## Usage + +```rust,no_run +use paracas_format::{CsvFormatter, Formatter, OutputFormat}; +use paracas_types::Tick; +use std::io::Cursor; + +fn main() -> Result<(), Box> { + let ticks: Vec = vec![]; + let mut output = Cursor::new(Vec::new()); + + let formatter = CsvFormatter::new(); + formatter.write_ticks(&ticks, &mut output)?; + Ok(()) +} +``` + +## Features + +- `csv` - CSV format support (default) +- `json` - JSON format support (default) +- `parquet` - Parquet format support (default) + +## License + +MIT License - see [LICENSE](../../LICENSE) for details. diff --git a/crates/paracas-format/src/csv.rs b/crates/paracas-format/src/csv.rs new file mode 100644 index 0000000..5c57be6 --- /dev/null +++ b/crates/paracas-format/src/csv.rs @@ -0,0 +1,163 @@ +//! CSV output format. + +use paracas_aggregate::Ohlcv; +use paracas_types::Tick; +use std::io::Write; + +use crate::{FormatError, Formatter}; + +/// CSV formatter. +#[derive(Debug, Clone, Default)] +pub struct CsvFormatter { + /// Field delimiter (default: comma). + delimiter: char, + /// Whether to include header row. + include_header: bool, +} + +impl CsvFormatter { + /// Creates a new CSV formatter with default settings. + #[must_use] + pub const fn new() -> Self { + Self { + delimiter: ',', + include_header: true, + } + } + + /// Sets the field delimiter. + #[must_use] + pub const fn with_delimiter(mut self, delimiter: char) -> Self { + self.delimiter = delimiter; + self + } + + /// Sets whether to include a header row. + #[must_use] + pub const fn with_header(mut self, include: bool) -> Self { + self.include_header = include; + self + } + + /// Creates a tab-separated values (TSV) formatter. + #[must_use] + pub const fn tsv() -> Self { + Self { + delimiter: '\t', + include_header: true, + } + } +} + +impl Formatter for CsvFormatter { + fn write_ticks( + &self, + ticks: &[Tick], + mut writer: W, + ) -> Result<(), FormatError> { + let d = self.delimiter; + + if self.include_header { + writeln!(writer, "timestamp{d}ask{d}bid{d}ask_volume{d}bid_volume")?; + } + + for tick in ticks { + writeln!( + writer, + "{}{d}{}{d}{}{d}{}{d}{}", + tick.timestamp.format("%Y-%m-%dT%H:%M:%S%.3fZ"), + tick.ask, + tick.bid, + tick.ask_volume, + tick.bid_volume + )?; + } + + Ok(()) + } + + fn write_ohlcv( + &self, + bars: &[Ohlcv], + mut writer: W, + ) -> Result<(), FormatError> { + let d = self.delimiter; + + if self.include_header { + writeln!( + writer, + "timestamp{d}open{d}high{d}low{d}close{d}volume{d}tick_count" + )?; + } + + for bar in bars { + writeln!( + writer, + "{}{d}{}{d}{}{d}{}{d}{}{d}{}{d}{}", + bar.timestamp.format("%Y-%m-%dT%H:%M:%SZ"), + bar.open, + bar.high, + bar.low, + bar.close, + bar.volume, + bar.tick_count + )?; + } + + Ok(()) + } + + fn extension(&self) -> &str { + "csv" + } +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::{TimeZone, Utc}; + use std::io::Cursor; + + fn create_test_tick() -> Tick { + let timestamp = Utc.with_ymd_and_hms(2024, 1, 15, 12, 30, 45).unwrap(); + Tick::new(timestamp, 1.1001, 1.1000, 100.0, 200.0) + } + + #[test] + fn test_csv_ticks() { + let formatter = CsvFormatter::new(); + let ticks = vec![create_test_tick()]; + let mut output = Cursor::new(Vec::new()); + + formatter.write_ticks(&ticks, &mut output).unwrap(); + + let result = String::from_utf8(output.into_inner()).unwrap(); + assert!(result.contains("timestamp,ask,bid,ask_volume,bid_volume")); + assert!(result.contains("2024-01-15T12:30:45.000Z")); + assert!(result.contains("1.1001")); + } + + #[test] + fn test_csv_no_header() { + let formatter = CsvFormatter::new().with_header(false); + let ticks = vec![create_test_tick()]; + let mut output = Cursor::new(Vec::new()); + + formatter.write_ticks(&ticks, &mut output).unwrap(); + + let result = String::from_utf8(output.into_inner()).unwrap(); + assert!(!result.contains("timestamp,ask")); + } + + #[test] + fn test_tsv() { + let formatter = CsvFormatter::tsv(); + let ticks = vec![create_test_tick()]; + let mut output = Cursor::new(Vec::new()); + + formatter.write_ticks(&ticks, &mut output).unwrap(); + + let result = String::from_utf8(output.into_inner()).unwrap(); + assert!(result.contains("timestamp\task\tbid")); + } +} diff --git a/crates/paracas-format/src/formatter.rs b/crates/paracas-format/src/formatter.rs new file mode 100644 index 0000000..a966a8d --- /dev/null +++ b/crates/paracas-format/src/formatter.rs @@ -0,0 +1,99 @@ +//! Output format abstraction. + +use paracas_aggregate::Ohlcv; +use paracas_types::Tick; +use std::io::Write; +use thiserror::Error; + +/// Output format identifier. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)] +pub enum OutputFormat { + /// CSV format. + #[default] + Csv, + /// JSON array format. + Json, + /// Newline-delimited JSON format. + Ndjson, + /// Apache Parquet format. + Parquet, +} + +impl OutputFormat { + /// Returns the file extension for this format. + #[must_use] + pub const fn extension(&self) -> &'static str { + match self { + Self::Csv => "csv", + Self::Json => "json", + Self::Ndjson => "ndjson", + Self::Parquet => "parquet", + } + } + + /// Returns all available formats. + #[must_use] + pub const fn all() -> &'static [Self] { + &[Self::Csv, Self::Json, Self::Ndjson, Self::Parquet] + } +} + +impl std::fmt::Display for OutputFormat { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.extension()) + } +} + +impl std::str::FromStr for OutputFormat { + type Err = FormatError; + + fn from_str(s: &str) -> Result { + match s.to_lowercase().as_str() { + "csv" => Ok(Self::Csv), + "json" => Ok(Self::Json), + "ndjson" | "jsonl" => Ok(Self::Ndjson), + "parquet" | "pq" => Ok(Self::Parquet), + _ => Err(FormatError::UnknownFormat(s.to_string())), + } + } +} + +/// Errors that can occur during formatting. +#[derive(Error, Debug)] +pub enum FormatError { + /// Unknown output format. + #[error("Unknown format: {0}")] + UnknownFormat(String), + + /// I/O error. + #[error("I/O error: {0}")] + Io(#[from] std::io::Error), + + /// JSON serialization error. + #[error("JSON error: {0}")] + Json(#[from] serde_json::Error), + + /// Arrow/Parquet error. + #[error("Parquet error: {0}")] + Parquet(String), +} + +/// Trait for output formatters. +pub trait Formatter: Send + Sync { + /// Writes tick data to the output. + /// + /// # Errors + /// + /// Returns an error if writing fails. + fn write_ticks(&self, ticks: &[Tick], writer: W) -> Result<(), FormatError>; + + /// Writes OHLCV data to the output. + /// + /// # Errors + /// + /// Returns an error if writing fails. + fn write_ohlcv(&self, bars: &[Ohlcv], writer: W) -> Result<(), FormatError>; + + /// Returns the file extension for this format. + fn extension(&self) -> &str; +} diff --git a/crates/paracas-format/src/json.rs b/crates/paracas-format/src/json.rs new file mode 100644 index 0000000..de63f9d --- /dev/null +++ b/crates/paracas-format/src/json.rs @@ -0,0 +1,169 @@ +//! JSON output format. + +use paracas_aggregate::Ohlcv; +use paracas_types::Tick; +use std::io::Write; + +use crate::{FormatError, Formatter}; + +/// JSON output style. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum JsonStyle { + /// JSON array (standard JSON). + #[default] + Array, + /// Newline-delimited JSON (NDJSON/JSONL). + Ndjson, +} + +/// JSON formatter. +#[derive(Debug, Clone, Default)] +pub struct JsonFormatter { + /// Output style. + style: JsonStyle, + /// Whether to pretty-print (only for array style). + pretty: bool, +} + +impl JsonFormatter { + /// Creates a new JSON formatter with default settings (array style). + #[must_use] + pub const fn new() -> Self { + Self { + style: JsonStyle::Array, + pretty: false, + } + } + + /// Creates a new NDJSON formatter. + #[must_use] + pub const fn ndjson() -> Self { + Self { + style: JsonStyle::Ndjson, + pretty: false, + } + } + + /// Sets whether to pretty-print output (array style only). + #[must_use] + pub const fn with_pretty(mut self, pretty: bool) -> Self { + self.pretty = pretty; + self + } + + /// Sets the output style. + #[must_use] + pub const fn with_style(mut self, style: JsonStyle) -> Self { + self.style = style; + self + } +} + +impl Formatter for JsonFormatter { + fn write_ticks( + &self, + ticks: &[Tick], + mut writer: W, + ) -> Result<(), FormatError> { + match self.style { + JsonStyle::Array => { + if self.pretty { + serde_json::to_writer_pretty(&mut writer, ticks)?; + } else { + serde_json::to_writer(&mut writer, ticks)?; + } + writeln!(writer)?; + } + JsonStyle::Ndjson => { + for tick in ticks { + serde_json::to_writer(&mut writer, tick)?; + writeln!(writer)?; + } + } + } + Ok(()) + } + + fn write_ohlcv( + &self, + bars: &[Ohlcv], + mut writer: W, + ) -> Result<(), FormatError> { + match self.style { + JsonStyle::Array => { + if self.pretty { + serde_json::to_writer_pretty(&mut writer, bars)?; + } else { + serde_json::to_writer(&mut writer, bars)?; + } + writeln!(writer)?; + } + JsonStyle::Ndjson => { + for bar in bars { + serde_json::to_writer(&mut writer, bar)?; + writeln!(writer)?; + } + } + } + Ok(()) + } + + fn extension(&self) -> &str { + match self.style { + JsonStyle::Array => "json", + JsonStyle::Ndjson => "ndjson", + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::{TimeZone, Utc}; + use std::io::Cursor; + + fn create_test_tick() -> Tick { + let timestamp = Utc.with_ymd_and_hms(2024, 1, 15, 12, 30, 45).unwrap(); + Tick::new(timestamp, 1.1001, 1.1000, 100.0, 200.0) + } + + #[test] + fn test_json_array() { + let formatter = JsonFormatter::new(); + let ticks = vec![create_test_tick()]; + let mut output = Cursor::new(Vec::new()); + + formatter.write_ticks(&ticks, &mut output).unwrap(); + + let result = String::from_utf8(output.into_inner()).unwrap(); + assert!(result.starts_with('[')); + assert!(result.contains("\"ask\":1.1001")); + } + + #[test] + fn test_ndjson() { + let formatter = JsonFormatter::ndjson(); + let ticks = vec![create_test_tick(), create_test_tick()]; + let mut output = Cursor::new(Vec::new()); + + formatter.write_ticks(&ticks, &mut output).unwrap(); + + let result = String::from_utf8(output.into_inner()).unwrap(); + let lines: Vec<_> = result.lines().collect(); + assert_eq!(lines.len(), 2); + assert!(lines[0].starts_with('{')); + } + + #[test] + fn test_pretty_json() { + let formatter = JsonFormatter::new().with_pretty(true); + let ticks = vec![create_test_tick()]; + let mut output = Cursor::new(Vec::new()); + + formatter.write_ticks(&ticks, &mut output).unwrap(); + + let result = String::from_utf8(output.into_inner()).unwrap(); + assert!(result.contains('\n')); + assert!(result.contains(" ")); // Indentation + } +} diff --git a/crates/paracas-format/src/lib.rs b/crates/paracas-format/src/lib.rs new file mode 100644 index 0000000..17cd34f --- /dev/null +++ b/crates/paracas-format/src/lib.rs @@ -0,0 +1,28 @@ +//! Output formatters for paracas tick data downloader. +//! +//! This crate provides formatters for writing tick and OHLCV data +//! to various output formats: +//! +//! - [`CsvFormatter`] - CSV format +//! - [`JsonFormatter`] - JSON array or NDJSON format +//! - [`ParquetFormatter`] - Apache Parquet columnar format + +#![doc = include_str!("../README.md")] +#![doc(issue_tracker_base_url = "https://github.com/factordynamics/paracas/issues/")] +#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))] +#![warn(missing_docs)] +#![forbid(unsafe_code)] + +mod csv; +mod formatter; +mod json; + +#[cfg(feature = "parquet")] +mod parquet; + +pub use crate::csv::CsvFormatter; +pub use formatter::{FormatError, Formatter, OutputFormat}; +pub use json::{JsonFormatter, JsonStyle}; + +#[cfg(feature = "parquet")] +pub use crate::parquet::ParquetFormatter; diff --git a/crates/paracas-format/src/parquet.rs b/crates/paracas-format/src/parquet.rs new file mode 100644 index 0000000..2ef40c5 --- /dev/null +++ b/crates/paracas-format/src/parquet.rs @@ -0,0 +1,236 @@ +//! Apache Parquet output format. + +use arrow::array::{Float32Array, Float64Array, TimestampMicrosecondArray, UInt32Array}; +use arrow::datatypes::{DataType, Field, Schema, TimeUnit}; +use arrow::record_batch::RecordBatch; +use paracas_aggregate::Ohlcv; +use paracas_types::Tick; +use parquet::arrow::ArrowWriter; +use parquet::basic::Compression; +use parquet::file::properties::WriterProperties; +use std::io::Write; +use std::sync::Arc; + +use crate::{FormatError, Formatter}; + +/// Parquet formatter. +#[derive(Debug, Clone)] +pub struct ParquetFormatter { + /// Row group size (number of rows per group). + row_group_size: usize, + /// Compression codec. + compression: Compression, +} + +impl Default for ParquetFormatter { + fn default() -> Self { + Self { + row_group_size: 100_000, + compression: Compression::SNAPPY, + } + } +} + +impl ParquetFormatter { + /// Creates a new Parquet formatter with default settings. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Sets the row group size. + #[must_use] + pub const fn with_row_group_size(mut self, size: usize) -> Self { + self.row_group_size = size; + self + } + + /// Sets the compression codec. + #[must_use] + pub const fn with_compression(mut self, compression: Compression) -> Self { + self.compression = compression; + self + } + + /// Creates the Arrow schema for tick data. + fn tick_schema() -> Schema { + Schema::new(vec![ + Field::new( + "timestamp", + DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())), + false, + ), + Field::new("ask", DataType::Float64, false), + Field::new("bid", DataType::Float64, false), + Field::new("ask_volume", DataType::Float32, false), + Field::new("bid_volume", DataType::Float32, false), + ]) + } + + /// Creates the Arrow schema for OHLCV data. + fn ohlcv_schema() -> Schema { + Schema::new(vec![ + Field::new( + "timestamp", + DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())), + false, + ), + Field::new("open", DataType::Float64, false), + Field::new("high", DataType::Float64, false), + Field::new("low", DataType::Float64, false), + Field::new("close", DataType::Float64, false), + Field::new("volume", DataType::Float64, false), + Field::new("tick_count", DataType::UInt32, false), + ]) + } + + /// Converts ticks to Arrow RecordBatch. + fn ticks_to_batch(ticks: &[Tick]) -> Result { + let timestamps: Vec<_> = ticks + .iter() + .map(|t| t.timestamp.timestamp_micros()) + .collect(); + let asks: Vec<_> = ticks.iter().map(|t| t.ask).collect(); + let bids: Vec<_> = ticks.iter().map(|t| t.bid).collect(); + let ask_vols: Vec<_> = ticks.iter().map(|t| t.ask_volume).collect(); + let bid_vols: Vec<_> = ticks.iter().map(|t| t.bid_volume).collect(); + + RecordBatch::try_new( + Arc::new(Self::tick_schema()), + vec![ + Arc::new(TimestampMicrosecondArray::from(timestamps).with_timezone("UTC")), + Arc::new(Float64Array::from(asks)), + Arc::new(Float64Array::from(bids)), + Arc::new(Float32Array::from(ask_vols)), + Arc::new(Float32Array::from(bid_vols)), + ], + ) + .map_err(|e| FormatError::Parquet(e.to_string())) + } + + /// Converts OHLCV bars to Arrow RecordBatch. + fn ohlcv_to_batch(bars: &[Ohlcv]) -> Result { + let timestamps: Vec<_> = bars + .iter() + .map(|b| b.timestamp.timestamp_micros()) + .collect(); + let opens: Vec<_> = bars.iter().map(|b| b.open).collect(); + let highs: Vec<_> = bars.iter().map(|b| b.high).collect(); + let lows: Vec<_> = bars.iter().map(|b| b.low).collect(); + let closes: Vec<_> = bars.iter().map(|b| b.close).collect(); + let volumes: Vec<_> = bars.iter().map(|b| b.volume).collect(); + let tick_counts: Vec<_> = bars.iter().map(|b| b.tick_count).collect(); + + RecordBatch::try_new( + Arc::new(Self::ohlcv_schema()), + vec![ + Arc::new(TimestampMicrosecondArray::from(timestamps).with_timezone("UTC")), + Arc::new(Float64Array::from(opens)), + Arc::new(Float64Array::from(highs)), + Arc::new(Float64Array::from(lows)), + Arc::new(Float64Array::from(closes)), + Arc::new(Float64Array::from(volumes)), + Arc::new(UInt32Array::from(tick_counts)), + ], + ) + .map_err(|e| FormatError::Parquet(e.to_string())) + } +} + +impl Formatter for ParquetFormatter { + fn write_ticks(&self, ticks: &[Tick], writer: W) -> Result<(), FormatError> { + let schema = Arc::new(Self::tick_schema()); + let props = WriterProperties::builder() + .set_compression(self.compression) + .set_max_row_group_size(self.row_group_size) + .build(); + + let mut arrow_writer = ArrowWriter::try_new(writer, schema, Some(props)) + .map_err(|e| FormatError::Parquet(e.to_string()))?; + + // Write in batches + for chunk in ticks.chunks(self.row_group_size) { + let batch = Self::ticks_to_batch(chunk)?; + arrow_writer + .write(&batch) + .map_err(|e| FormatError::Parquet(e.to_string()))?; + } + + arrow_writer + .close() + .map_err(|e| FormatError::Parquet(e.to_string()))?; + + Ok(()) + } + + fn write_ohlcv(&self, bars: &[Ohlcv], writer: W) -> Result<(), FormatError> { + let schema = Arc::new(Self::ohlcv_schema()); + let props = WriterProperties::builder() + .set_compression(self.compression) + .set_max_row_group_size(self.row_group_size) + .build(); + + let mut arrow_writer = ArrowWriter::try_new(writer, schema, Some(props)) + .map_err(|e| FormatError::Parquet(e.to_string()))?; + + // Write in batches + for chunk in bars.chunks(self.row_group_size) { + let batch = Self::ohlcv_to_batch(chunk)?; + arrow_writer + .write(&batch) + .map_err(|e| FormatError::Parquet(e.to_string()))?; + } + + arrow_writer + .close() + .map_err(|e| FormatError::Parquet(e.to_string()))?; + + Ok(()) + } + + fn extension(&self) -> &str { + "parquet" + } +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::{TimeZone, Utc}; + use std::io::Cursor; + + fn create_test_tick() -> Tick { + let timestamp = Utc.with_ymd_and_hms(2024, 1, 15, 12, 30, 45).unwrap(); + Tick::new(timestamp, 1.1001, 1.1000, 100.0, 200.0) + } + + #[test] + fn test_parquet_ticks() { + let formatter = ParquetFormatter::new(); + let ticks = vec![create_test_tick()]; + let mut output = Cursor::new(Vec::new()); + + formatter.write_ticks(&ticks, &mut output).unwrap(); + + // Parquet files start with "PAR1" magic bytes + let data = output.into_inner(); + assert!(data.len() > 4); + assert_eq!(&data[0..4], b"PAR1"); + } + + #[test] + fn test_tick_schema() { + let schema = ParquetFormatter::tick_schema(); + assert_eq!(schema.fields().len(), 5); + assert!(schema.field_with_name("timestamp").is_ok()); + assert!(schema.field_with_name("ask").is_ok()); + } + + #[test] + fn test_ohlcv_schema() { + let schema = ParquetFormatter::ohlcv_schema(); + assert_eq!(schema.fields().len(), 7); + assert!(schema.field_with_name("open").is_ok()); + assert!(schema.field_with_name("close").is_ok()); + } +} diff --git a/crates/paracas-instruments/Cargo.toml b/crates/paracas-instruments/Cargo.toml new file mode 100644 index 0000000..efacd60 --- /dev/null +++ b/crates/paracas-instruments/Cargo.toml @@ -0,0 +1,26 @@ +[package] +name = "paracas-instruments" +description = "Instrument registry for paracas tick data downloader" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +homepage.workspace = true +repository.workspace = true +documentation.workspace = true +authors.workspace = true +keywords.workspace = true +categories.workspace = true + +[lints] +workspace = true + +[package.metadata.docs.rs] +all-features = true +rustdoc-args = ["--cfg", "docsrs"] + +[dependencies] +paracas-types = { workspace = true } +chrono = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } diff --git a/crates/paracas-instruments/README.md b/crates/paracas-instruments/README.md new file mode 100644 index 0000000..7846e29 --- /dev/null +++ b/crates/paracas-instruments/README.md @@ -0,0 +1,37 @@ +# paracas-instruments + +Instrument registry for the paracas tick data downloader. + +## Features + +- Registry of 1000+ Dukascopy instruments +- Lookup by ID (case-insensitive) +- Filter by category (forex, crypto, stocks, etc.) +- Search by name pattern + +## Usage + +```rust +use paracas_instruments::InstrumentRegistry; + +let registry = InstrumentRegistry::global(); + +// Lookup by ID +if let Some(instrument) = registry.get("eurusd") { + println!("{}: decimal_factor = {}", instrument.name(), instrument.decimal_factor()); +} + +// Filter by category +for instrument in registry.forex() { + println!("{}", instrument.id()); +} + +// Search +for instrument in registry.search("btc") { + println!("{}", instrument.name()); +} +``` + +## License + +MIT License - see [LICENSE](../../LICENSE) for details. diff --git a/crates/paracas-instruments/data/instruments.json b/crates/paracas-instruments/data/instruments.json new file mode 100644 index 0000000..d304c61 --- /dev/null +++ b/crates/paracas-instruments/data/instruments.json @@ -0,0 +1,354 @@ +{ + "eurusd": { + "id": "eurusd", + "name": "EUR/USD", + "description": "Euro vs US Dollar", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2003-05-05T00:00:00Z" + }, + "gbpusd": { + "id": "gbpusd", + "name": "GBP/USD", + "description": "British Pound vs US Dollar", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2003-05-05T00:00:00Z" + }, + "usdjpy": { + "id": "usdjpy", + "name": "USD/JPY", + "description": "US Dollar vs Japanese Yen", + "category": "forex", + "decimal_factor": 1000, + "start_tick_date": "2003-05-05T00:00:00Z" + }, + "usdchf": { + "id": "usdchf", + "name": "USD/CHF", + "description": "US Dollar vs Swiss Franc", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2003-05-05T00:00:00Z" + }, + "audusd": { + "id": "audusd", + "name": "AUD/USD", + "description": "Australian Dollar vs US Dollar", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2003-08-03T00:00:00Z" + }, + "usdcad": { + "id": "usdcad", + "name": "USD/CAD", + "description": "US Dollar vs Canadian Dollar", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2003-08-03T00:00:00Z" + }, + "nzdusd": { + "id": "nzdusd", + "name": "NZD/USD", + "description": "New Zealand Dollar vs US Dollar", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2003-08-03T00:00:00Z" + }, + "eurgbp": { + "id": "eurgbp", + "name": "EUR/GBP", + "description": "Euro vs British Pound", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2003-08-03T00:00:00Z" + }, + "eurjpy": { + "id": "eurjpy", + "name": "EUR/JPY", + "description": "Euro vs Japanese Yen", + "category": "forex", + "decimal_factor": 1000, + "start_tick_date": "2003-08-03T00:00:00Z" + }, + "eurchf": { + "id": "eurchf", + "name": "EUR/CHF", + "description": "Euro vs Swiss Franc", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2003-08-03T00:00:00Z" + }, + "gbpjpy": { + "id": "gbpjpy", + "name": "GBP/JPY", + "description": "British Pound vs Japanese Yen", + "category": "forex", + "decimal_factor": 1000, + "start_tick_date": "2003-08-03T00:00:00Z" + }, + "gbpchf": { + "id": "gbpchf", + "name": "GBP/CHF", + "description": "British Pound vs Swiss Franc", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2003-08-03T00:00:00Z" + }, + "chfjpy": { + "id": "chfjpy", + "name": "CHF/JPY", + "description": "Swiss Franc vs Japanese Yen", + "category": "forex", + "decimal_factor": 1000, + "start_tick_date": "2003-08-03T00:00:00Z" + }, + "euraud": { + "id": "euraud", + "name": "EUR/AUD", + "description": "Euro vs Australian Dollar", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2005-10-30T00:00:00Z" + }, + "eurcad": { + "id": "eurcad", + "name": "EUR/CAD", + "description": "Euro vs Canadian Dollar", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2005-10-30T00:00:00Z" + }, + "eurnzd": { + "id": "eurnzd", + "name": "EUR/NZD", + "description": "Euro vs New Zealand Dollar", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2005-10-30T00:00:00Z" + }, + "gbpaud": { + "id": "gbpaud", + "name": "GBP/AUD", + "description": "British Pound vs Australian Dollar", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2005-10-30T00:00:00Z" + }, + "gbpcad": { + "id": "gbpcad", + "name": "GBP/CAD", + "description": "British Pound vs Canadian Dollar", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2005-10-30T00:00:00Z" + }, + "gbpnzd": { + "id": "gbpnzd", + "name": "GBP/NZD", + "description": "British Pound vs New Zealand Dollar", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2005-10-30T00:00:00Z" + }, + "audcad": { + "id": "audcad", + "name": "AUD/CAD", + "description": "Australian Dollar vs Canadian Dollar", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2005-10-30T00:00:00Z" + }, + "audchf": { + "id": "audchf", + "name": "AUD/CHF", + "description": "Australian Dollar vs Swiss Franc", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2005-10-30T00:00:00Z" + }, + "audjpy": { + "id": "audjpy", + "name": "AUD/JPY", + "description": "Australian Dollar vs Japanese Yen", + "category": "forex", + "decimal_factor": 1000, + "start_tick_date": "2005-10-30T00:00:00Z" + }, + "audnzd": { + "id": "audnzd", + "name": "AUD/NZD", + "description": "Australian Dollar vs New Zealand Dollar", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2005-10-30T00:00:00Z" + }, + "cadjpy": { + "id": "cadjpy", + "name": "CAD/JPY", + "description": "Canadian Dollar vs Japanese Yen", + "category": "forex", + "decimal_factor": 1000, + "start_tick_date": "2005-10-30T00:00:00Z" + }, + "cadchf": { + "id": "cadchf", + "name": "CAD/CHF", + "description": "Canadian Dollar vs Swiss Franc", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2005-10-30T00:00:00Z" + }, + "nzdjpy": { + "id": "nzdjpy", + "name": "NZD/JPY", + "description": "New Zealand Dollar vs Japanese Yen", + "category": "forex", + "decimal_factor": 1000, + "start_tick_date": "2005-10-30T00:00:00Z" + }, + "nzdcad": { + "id": "nzdcad", + "name": "NZD/CAD", + "description": "New Zealand Dollar vs Canadian Dollar", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2005-10-30T00:00:00Z" + }, + "nzdchf": { + "id": "nzdchf", + "name": "NZD/CHF", + "description": "New Zealand Dollar vs Swiss Franc", + "category": "forex", + "decimal_factor": 100000, + "start_tick_date": "2005-10-30T00:00:00Z" + }, + "xauusd": { + "id": "xauusd", + "name": "XAU/USD", + "description": "Gold vs US Dollar", + "category": "commodity", + "decimal_factor": 1000, + "start_tick_date": "2004-05-05T00:00:00Z" + }, + "xagusd": { + "id": "xagusd", + "name": "XAG/USD", + "description": "Silver vs US Dollar", + "category": "commodity", + "decimal_factor": 1000, + "start_tick_date": "2004-05-05T00:00:00Z" + }, + "btcusd": { + "id": "btcusd", + "name": "BTC/USD", + "description": "Bitcoin vs US Dollar", + "category": "crypto", + "decimal_factor": 100, + "start_tick_date": "2017-01-01T00:00:00Z" + }, + "ethusd": { + "id": "ethusd", + "name": "ETH/USD", + "description": "Ethereum vs US Dollar", + "category": "crypto", + "decimal_factor": 100, + "start_tick_date": "2017-06-01T00:00:00Z" + }, + "ltcusd": { + "id": "ltcusd", + "name": "LTC/USD", + "description": "Litecoin vs US Dollar", + "category": "crypto", + "decimal_factor": 100, + "start_tick_date": "2017-07-01T00:00:00Z" + }, + "xrpusd": { + "id": "xrpusd", + "name": "XRP/USD", + "description": "Ripple vs US Dollar", + "category": "crypto", + "decimal_factor": 100000, + "start_tick_date": "2017-09-01T00:00:00Z" + }, + "usa500idxusd": { + "id": "usa500idxusd", + "name": "US 500", + "description": "S&P 500 Index", + "category": "index", + "decimal_factor": 10, + "start_tick_date": "2013-05-05T00:00:00Z" + }, + "usa30idxusd": { + "id": "usa30idxusd", + "name": "US 30", + "description": "Dow Jones Industrial Average", + "category": "index", + "decimal_factor": 10, + "start_tick_date": "2013-05-05T00:00:00Z" + }, + "usaborrowusd": { + "id": "usatechidxusd", + "name": "US Tech 100", + "description": "NASDAQ 100 Index", + "category": "index", + "decimal_factor": 10, + "start_tick_date": "2013-05-05T00:00:00Z" + }, + "deuidxeur": { + "id": "deuidxeur", + "name": "Germany 40", + "description": "DAX 40 Index", + "category": "index", + "decimal_factor": 10, + "start_tick_date": "2013-05-05T00:00:00Z" + }, + "gbridxgbp": { + "id": "gbridxgbp", + "name": "UK 100", + "description": "FTSE 100 Index", + "category": "index", + "decimal_factor": 10, + "start_tick_date": "2013-05-05T00:00:00Z" + }, + "jpnidxjpy": { + "id": "jpnidxjpy", + "name": "Japan 225", + "description": "Nikkei 225 Index", + "category": "index", + "decimal_factor": 1, + "start_tick_date": "2013-05-05T00:00:00Z" + }, + "uscrude": { + "id": "uscrude", + "name": "WTI Crude Oil", + "description": "West Texas Intermediate Crude Oil", + "category": "commodity", + "decimal_factor": 1000, + "start_tick_date": "2013-05-05T00:00:00Z" + }, + "ukcrude": { + "id": "ukcrude", + "name": "Brent Crude Oil", + "description": "Brent Crude Oil", + "category": "commodity", + "decimal_factor": 1000, + "start_tick_date": "2013-05-05T00:00:00Z" + }, + "natgas": { + "id": "natgas", + "name": "Natural Gas", + "description": "Natural Gas", + "category": "commodity", + "decimal_factor": 1000, + "start_tick_date": "2013-05-05T00:00:00Z" + }, + "copper": { + "id": "copper", + "name": "Copper", + "description": "Copper", + "category": "commodity", + "decimal_factor": 10000, + "start_tick_date": "2013-05-05T00:00:00Z" + } +} diff --git a/crates/paracas-instruments/src/lib.rs b/crates/paracas-instruments/src/lib.rs new file mode 100644 index 0000000..c5654ad --- /dev/null +++ b/crates/paracas-instruments/src/lib.rs @@ -0,0 +1,173 @@ +//! Instrument registry for paracas tick data downloader. +//! +//! This crate provides access to the full list of Dukascopy instruments +//! with their metadata including decimal factors for price normalization. +//! +//! # Example +//! +//! ``` +//! use paracas_instruments::InstrumentRegistry; +//! +//! let registry = InstrumentRegistry::global(); +//! +//! // Lookup by ID +//! if let Some(instrument) = registry.get("eurusd") { +//! println!("{}: {}", instrument.name(), instrument.decimal_factor()); +//! } +//! ``` + +#![doc = include_str!("../README.md")] +#![doc(issue_tracker_base_url = "https://github.com/factordynamics/paracas/issues/")] +#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))] +#![warn(missing_docs)] +#![forbid(unsafe_code)] + +use std::collections::HashMap; +use std::sync::OnceLock; + +use paracas_types::{Category, Instrument}; + +/// The instrument metadata JSON embedded at compile time. +const INSTRUMENTS_JSON: &str = include_str!("../data/instruments.json"); + +/// Global instrument registry instance. +static REGISTRY: OnceLock = OnceLock::new(); + +/// Registry of all supported Dukascopy instruments. +#[derive(Debug)] +pub struct InstrumentRegistry { + instruments: HashMap, +} + +impl InstrumentRegistry { + /// Returns the global instrument registry. + /// + /// The registry is initialized lazily on first access. + #[must_use] + pub fn global() -> &'static Self { + REGISTRY.get_or_init(Self::load) + } + + /// Loads instruments from the embedded JSON data. + fn load() -> Self { + let instruments: HashMap = + serde_json::from_str(INSTRUMENTS_JSON).expect("Invalid instruments.json"); + Self { instruments } + } + + /// Looks up an instrument by ID (case-insensitive). + #[must_use] + pub fn get(&self, id: &str) -> Option<&Instrument> { + self.instruments.get(&id.to_lowercase()) + } + + /// Returns all instruments as an iterator. + pub fn all(&self) -> impl Iterator { + self.instruments.values() + } + + /// Returns the total number of instruments. + #[must_use] + pub fn len(&self) -> usize { + self.instruments.len() + } + + /// Returns true if the registry is empty. + #[must_use] + pub fn is_empty(&self) -> bool { + self.instruments.is_empty() + } + + /// Returns all forex instruments. + pub fn forex(&self) -> impl Iterator { + self.instruments.values().filter(|i| i.is_forex()) + } + + /// Returns all cryptocurrency instruments. + pub fn crypto(&self) -> impl Iterator { + self.instruments.values().filter(|i| i.is_crypto()) + } + + /// Returns all index instruments. + pub fn indices(&self) -> impl Iterator { + self.instruments.values().filter(|i| i.is_index()) + } + + /// Returns all stock instruments. + pub fn stocks(&self) -> impl Iterator { + self.instruments.values().filter(|i| i.is_stock()) + } + + /// Returns all commodity instruments. + pub fn commodities(&self) -> impl Iterator { + self.instruments.values().filter(|i| i.is_commodity()) + } + + /// Returns instruments matching the given category. + pub fn by_category(&self, category: Category) -> impl Iterator { + self.instruments + .values() + .filter(move |i| i.category() == category) + } + + /// Searches instruments by name or ID pattern (case-insensitive). + pub fn search(&self, pattern: &str) -> Vec<&Instrument> { + let pattern = pattern.to_lowercase(); + self.instruments + .values() + .filter(|i| { + i.id().to_lowercase().contains(&pattern) + || i.name().to_lowercase().contains(&pattern) + }) + .collect() + } + + /// Returns all instrument IDs sorted alphabetically. + pub fn ids(&self) -> Vec<&str> { + let mut ids: Vec<&str> = self.instruments.keys().map(String::as_str).collect(); + ids.sort(); + ids + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_registry_loads() { + let registry = InstrumentRegistry::global(); + assert!(!registry.is_empty()); + } + + #[test] + fn test_get_eurusd() { + let registry = InstrumentRegistry::global(); + let eurusd = registry.get("eurusd").expect("EURUSD should exist"); + assert_eq!(eurusd.id(), "eurusd"); + assert_eq!(eurusd.decimal_factor(), 100_000); + } + + #[test] + fn test_get_case_insensitive() { + let registry = InstrumentRegistry::global(); + assert!(registry.get("EURUSD").is_some()); + assert!(registry.get("EurUsd").is_some()); + assert!(registry.get("eurusd").is_some()); + } + + #[test] + fn test_forex_filter() { + let registry = InstrumentRegistry::global(); + let forex: Vec<_> = registry.forex().collect(); + assert!(!forex.is_empty()); + assert!(forex.iter().all(|i| i.is_forex())); + } + + #[test] + fn test_search() { + let registry = InstrumentRegistry::global(); + let results = registry.search("eur"); + assert!(!results.is_empty()); + } +} diff --git a/crates/paracas-lib/Cargo.toml b/crates/paracas-lib/Cargo.toml new file mode 100644 index 0000000..94214f4 --- /dev/null +++ b/crates/paracas-lib/Cargo.toml @@ -0,0 +1,35 @@ +[package] +name = "paracas-lib" +description = "High-performance Rust library for downloading Dukascopy tick data" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +homepage.workspace = true +repository.workspace = true +documentation.workspace = true +authors.workspace = true +keywords.workspace = true +categories.workspace = true + +[lints] +workspace = true + +[package.metadata.docs.rs] +all-features = true +rustdoc-args = ["--cfg", "docsrs"] + +[features] +default = ["full"] +full = ["fetch", "aggregate", "format", "parquet"] +fetch = ["dep:paracas-fetch"] +aggregate = ["dep:paracas-aggregate"] +format = ["dep:paracas-format"] +parquet = ["format", "paracas-format/parquet"] + +[dependencies] +paracas-types = { workspace = true } +paracas-instruments = { workspace = true } +paracas-fetch = { workspace = true, optional = true } +paracas-aggregate = { workspace = true, optional = true } +paracas-format = { workspace = true, optional = true } diff --git a/crates/paracas-lib/README.md b/crates/paracas-lib/README.md new file mode 100644 index 0000000..d0e495f --- /dev/null +++ b/crates/paracas-lib/README.md @@ -0,0 +1,56 @@ +# paracas-lib + +High-performance Rust library for downloading historical tick data from Dukascopy. + +## Features + +- **Fast**: Concurrent downloads with connection pooling +- **Flexible**: CSV, JSON, and Parquet output formats +- **Complete**: All 1000+ Dukascopy instruments supported +- **Aggregation**: Built-in OHLCV aggregation + +## Quick Start + +```rust,ignore +use paracas_lib::prelude::*; +use futures::StreamExt; + +#[tokio::main] +async fn main() -> Result<(), Box> { + // Get instrument + let registry = InstrumentRegistry::global(); + let instrument = registry.get("eurusd").unwrap(); + + // Create client + let client = DownloadClient::with_defaults()?; + + // Define date range + let range = DateRange::new( + chrono::NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(), + chrono::NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(), + )?; + + // Stream ticks + let mut stream = tick_stream(&client, instrument, range); + while let Some(batch) = stream.next().await { + let batch = batch?; + println!("Downloaded {} ticks for {:?}", batch.len(), batch.hour); + } + + Ok(()) +} +``` + +## Crates + +This is a facade crate that re-exports functionality from: + +- `paracas-types` - Core types (Tick, Instrument, DateRange) +- `paracas-instruments` - Instrument registry +- `paracas-fetch` - HTTP client and data fetching +- `paracas-aggregate` - OHLCV aggregation +- `paracas-format` - Output formatters + +## License + +MIT License - see [LICENSE](../../LICENSE) for details. diff --git a/crates/paracas-lib/src/lib.rs b/crates/paracas-lib/src/lib.rs new file mode 100644 index 0000000..31ce087 --- /dev/null +++ b/crates/paracas-lib/src/lib.rs @@ -0,0 +1,88 @@ +//! High-performance Rust library for downloading Dukascopy tick data. +//! +//! This is a facade crate that re-exports functionality from the paracas +//! workspace crates for convenient access. +//! +//! # Quick Start +//! +//! ```ignore +//! use paracas_lib::prelude::*; +//! use futures::StreamExt; +//! +//! #[tokio::main] +//! async fn main() -> Result<(), Box> { +//! let registry = InstrumentRegistry::global(); +//! let instrument = registry.get("eurusd").unwrap(); +//! let client = DownloadClient::with_defaults()?; +//! +//! let range = DateRange::new( +//! chrono::NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(), +//! chrono::NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(), +//! )?; +//! +//! let mut stream = tick_stream(&client, instrument, range); +//! while let Some(batch) = stream.next().await { +//! println!("Downloaded {} ticks", batch?.len()); +//! } +//! +//! Ok(()) +//! } +//! ``` + +#![doc = include_str!("../README.md")] +#![doc(issue_tracker_base_url = "https://github.com/factordynamics/paracas/issues/")] +#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))] +#![warn(missing_docs)] +#![forbid(unsafe_code)] + +// Re-export core types +pub use paracas_types::*; + +// Re-export instrument registry +pub use paracas_instruments::InstrumentRegistry; + +// Re-export fetch functionality +#[cfg(feature = "fetch")] +pub use paracas_fetch::{ + ClientConfig, DecompressError, DownloadClient, DownloadError, ParseError, TickBatch, + tick_stream, tick_stream_resilient, +}; + +// Re-export aggregation +#[cfg(feature = "aggregate")] +pub use paracas_aggregate::{Ohlcv, TickAggregator}; + +// Re-export formatters +#[cfg(feature = "format")] +pub use paracas_format::{CsvFormatter, FormatError, Formatter, JsonFormatter, OutputFormat}; + +#[cfg(all(feature = "format", feature = "parquet"))] +pub use paracas_format::ParquetFormatter; + +/// Prelude module for convenient imports. +/// +/// ``` +/// use paracas_lib::prelude::*; +/// ``` +pub mod prelude { + pub use paracas_types::{ + Category, DateRange, DateRangeError, Instrument, ParacasError, RawTick, Result, Tick, + Timeframe, + }; + + pub use paracas_instruments::InstrumentRegistry; + + #[cfg(feature = "fetch")] + pub use paracas_fetch::{ + ClientConfig, DownloadClient, TickBatch, tick_stream, tick_stream_resilient, + }; + + #[cfg(feature = "aggregate")] + pub use paracas_aggregate::{Ohlcv, TickAggregator}; + + #[cfg(feature = "format")] + pub use paracas_format::{CsvFormatter, Formatter, JsonFormatter, OutputFormat}; + + #[cfg(all(feature = "format", feature = "parquet"))] + pub use paracas_format::ParquetFormatter; +} diff --git a/crates/paracas-types/Cargo.toml b/crates/paracas-types/Cargo.toml new file mode 100644 index 0000000..7dc4aae --- /dev/null +++ b/crates/paracas-types/Cargo.toml @@ -0,0 +1,26 @@ +[package] +name = "paracas-types" +description = "Core types for paracas tick data downloader" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +homepage.workspace = true +repository.workspace = true +documentation.workspace = true +authors.workspace = true +keywords.workspace = true +categories.workspace = true + +[lints] +workspace = true + +[package.metadata.docs.rs] +all-features = true +rustdoc-args = ["--cfg", "docsrs"] + +[dependencies] +chrono = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } diff --git a/crates/paracas-types/README.md b/crates/paracas-types/README.md new file mode 100644 index 0000000..e6bf3d6 --- /dev/null +++ b/crates/paracas-types/README.md @@ -0,0 +1,16 @@ +# paracas-types + +Core types for the paracas tick data downloader. + +## Types + +- `Tick` - A single tick with timestamp, ask, bid, and volumes +- `RawTick` - Raw tick from bi5 binary format (before normalization) +- `Instrument` - Financial instrument with metadata +- `Timeframe` - OHLCV aggregation timeframe +- `DateRange` - Date range for data retrieval +- `ParacasError` - Error types + +## License + +MIT License - see [LICENSE](../../LICENSE) for details. diff --git a/crates/paracas-types/src/date_range.rs b/crates/paracas-types/src/date_range.rs new file mode 100644 index 0000000..837c7c7 --- /dev/null +++ b/crates/paracas-types/src/date_range.rs @@ -0,0 +1,191 @@ +//! Date range and hour iteration. + +use chrono::{DateTime, NaiveDate, NaiveTime, TimeZone, Utc}; + +use crate::DateRangeError; + +/// A range of dates for data retrieval. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct DateRange { + /// Start date (inclusive). + pub start: NaiveDate, + /// End date (inclusive). + pub end: NaiveDate, +} + +impl DateRange { + /// Creates a new date range, validating that start <= end. + /// + /// # Errors + /// + /// Returns an error if start > end. + pub fn new(start: NaiveDate, end: NaiveDate) -> Result { + if start > end { + return Err(DateRangeError::InvalidRange { start, end }); + } + Ok(Self { start, end }) + } + + /// Creates a date range for a single day. + #[must_use] + pub const fn single_day(date: NaiveDate) -> Self { + Self { + start: date, + end: date, + } + } + + /// Returns an iterator over all hours in the date range. + pub fn hours(&self) -> HourIterator { + HourIterator::new(self.start, self.end) + } + + /// Returns the total number of hours in the range. + #[must_use] + pub fn total_hours(&self) -> usize { + let days = (self.end - self.start).num_days() + 1; + (days * 24) as usize + } + + /// Returns the total number of days in the range. + #[must_use] + pub fn total_days(&self) -> usize { + ((self.end - self.start).num_days() + 1) as usize + } + + /// Returns true if the range contains the given date. + #[must_use] + pub fn contains(&self, date: NaiveDate) -> bool { + date >= self.start && date <= self.end + } +} + +impl std::fmt::Display for DateRange { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{} to {}", self.start, self.end) + } +} + +/// Iterator over all hours in a date range. +#[derive(Debug, Clone)] +pub struct HourIterator { + current: DateTime, + end: DateTime, +} + +impl HourIterator { + /// Creates a new hour iterator for the given date range. + fn new(start: NaiveDate, end: NaiveDate) -> Self { + let start_dt = + Utc.from_utc_datetime(&start.and_time(NaiveTime::from_hms_opt(0, 0, 0).unwrap())); + // End at 23:00 of the end date (last hour of the day) + let end_dt = + Utc.from_utc_datetime(&end.and_time(NaiveTime::from_hms_opt(23, 0, 0).unwrap())); + + Self { + current: start_dt, + end: end_dt, + } + } +} + +impl Iterator for HourIterator { + type Item = DateTime; + + fn next(&mut self) -> Option { + if self.current > self.end { + return None; + } + + let result = self.current; + self.current += chrono::TimeDelta::hours(1); + Some(result) + } + + fn size_hint(&self) -> (usize, Option) { + if self.current > self.end { + return (0, Some(0)); + } + let hours = (self.end - self.current).num_hours() as usize + 1; + (hours, Some(hours)) + } +} + +impl ExactSizeIterator for HourIterator {} + +/// Extracts the hour start timestamp from a Dukascopy URL. +/// +/// URL format: `https://datafeed.dukascopy.com/datafeed/{INSTRUMENT}/{YEAR}/{MONTH}/{DAY}/{HOUR}h_ticks.bi5` +/// Note: Month in URL is 0-indexed (January = 00). +#[must_use] +pub fn hour_from_url(url: &str) -> Option> { + let parts: Vec<&str> = url.split('/').collect(); + if parts.len() < 5 { + return None; + } + + // Parse from the end: .../{YEAR}/{MONTH}/{DAY}/{HOUR}h_ticks.bi5 + let hour_part = parts.last()?; + let hour: u32 = hour_part.strip_suffix("h_ticks.bi5")?.parse().ok()?; + let day: u32 = parts.get(parts.len() - 2)?.parse().ok()?; + let month: u32 = parts.get(parts.len() - 3)?.parse().ok()?; + let year: i32 = parts.get(parts.len() - 4)?.parse().ok()?; + + // Month in URL is 0-indexed + Utc.with_ymd_and_hms(year, month + 1, day, hour, 0, 0) + .single() +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::{Datelike, Timelike}; + + #[test] + fn test_date_range_new() { + let start = NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(); + let end = NaiveDate::from_ymd_opt(2024, 1, 31).unwrap(); + let range = DateRange::new(start, end).unwrap(); + + assert_eq!(range.start, start); + assert_eq!(range.end, end); + } + + #[test] + fn test_date_range_invalid() { + let start = NaiveDate::from_ymd_opt(2024, 1, 31).unwrap(); + let end = NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(); + assert!(DateRange::new(start, end).is_err()); + } + + #[test] + fn test_total_hours() { + let start = NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(); + let end = NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(); + let range = DateRange::new(start, end).unwrap(); + + assert_eq!(range.total_hours(), 24); + } + + #[test] + fn test_hour_iterator() { + let start = NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(); + let range = DateRange::single_day(start); + let hours: Vec<_> = range.hours().collect(); + + assert_eq!(hours.len(), 24); + assert_eq!(hours[0].hour(), 0); + assert_eq!(hours[23].hour(), 23); + } + + #[test] + fn test_hour_from_url() { + let url = "https://datafeed.dukascopy.com/datafeed/EURUSD/2024/00/15/12h_ticks.bi5"; + let hour = hour_from_url(url).unwrap(); + + assert_eq!(hour.year(), 2024); + assert_eq!(hour.month(), 1); // 00 -> January + assert_eq!(hour.day(), 15); + assert_eq!(hour.hour(), 12); + } +} diff --git a/crates/paracas-types/src/error.rs b/crates/paracas-types/src/error.rs new file mode 100644 index 0000000..a5ccee7 --- /dev/null +++ b/crates/paracas-types/src/error.rs @@ -0,0 +1,63 @@ +//! Error types for paracas. + +use chrono::NaiveDate; +use thiserror::Error; + +/// Result type alias for paracas operations. +pub type Result = std::result::Result; + +/// Errors that can occur during data download and processing. +#[derive(Error, Debug)] +pub enum ParacasError { + /// HTTP request failed. + #[error("HTTP error: {0}")] + Http(String), + + /// LZMA decompression failed. + #[error("Decompression error: {0}")] + Decompress(String), + + /// Invalid data format. + #[error("Parse error: {0}")] + Parse(String), + + /// Instrument not found. + #[error("Unknown instrument: {0}")] + UnknownInstrument(String), + + /// Invalid date range. + #[error(transparent)] + DateRange(#[from] DateRangeError), + + /// No data available for the requested period. + #[error("No data available for {instrument} in requested range")] + NoDataAvailable { + /// The instrument that had no data. + instrument: String, + }, + + /// I/O error. + #[error("I/O error: {0}")] + Io(#[from] std::io::Error), + + /// Output format error. + #[error("Format error: {0}")] + Format(String), + + /// JSON serialization error. + #[error("JSON error: {0}")] + Json(#[from] serde_json::Error), +} + +/// Error for invalid date ranges. +#[derive(Error, Debug, Clone, PartialEq, Eq)] +pub enum DateRangeError { + /// Start date is after end date. + #[error("Invalid date range: {start} > {end}")] + InvalidRange { + /// The start date. + start: NaiveDate, + /// The end date. + end: NaiveDate, + }, +} diff --git a/crates/paracas-types/src/instrument.rs b/crates/paracas-types/src/instrument.rs new file mode 100644 index 0000000..8949357 --- /dev/null +++ b/crates/paracas-types/src/instrument.rs @@ -0,0 +1,213 @@ +//! Financial instrument definitions. + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +/// Instrument category. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum Category { + /// Foreign exchange currency pairs. + Forex, + /// Cryptocurrencies. + Crypto, + /// Stock indices. + Index, + /// Individual stocks. + Stock, + /// Commodities (metals, energy, agriculture). + Commodity, + /// Exchange-traded funds. + Etf, + /// Government bonds. + Bond, +} + +impl Category { + /// Returns the category as a string slice. + #[must_use] + pub const fn as_str(&self) -> &'static str { + match self { + Self::Forex => "forex", + Self::Crypto => "crypto", + Self::Index => "index", + Self::Stock => "stock", + Self::Commodity => "commodity", + Self::Etf => "etf", + Self::Bond => "bond", + } + } +} + +impl std::fmt::Display for Category { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.as_str()) + } +} + +/// Represents a tradable financial instrument. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct Instrument { + /// Unique identifier (e.g., "eurusd", "btcusd"). + id: String, + /// Human-readable name (e.g., "EUR/USD"). + name: String, + /// Description of the instrument. + description: String, + /// Instrument category. + category: Category, + /// Decimal factor for price normalization. + decimal_factor: u32, + /// Earliest available tick data timestamp. + start_tick_date: Option>, +} + +impl Instrument { + /// Creates a new instrument. + #[must_use] + pub fn new( + id: impl Into, + name: impl Into, + description: impl Into, + category: Category, + decimal_factor: u32, + start_tick_date: Option>, + ) -> Self { + Self { + id: id.into(), + name: name.into(), + description: description.into(), + category, + decimal_factor, + start_tick_date, + } + } + + /// Returns the instrument identifier. + #[must_use] + pub fn id(&self) -> &str { + &self.id + } + + /// Returns the human-readable name. + #[must_use] + pub fn name(&self) -> &str { + &self.name + } + + /// Returns the description. + #[must_use] + pub fn description(&self) -> &str { + &self.description + } + + /// Returns the instrument category. + #[must_use] + pub const fn category(&self) -> Category { + self.category + } + + /// Returns the decimal factor for price normalization. + #[must_use] + pub const fn decimal_factor(&self) -> u32 { + self.decimal_factor + } + + /// Returns the decimal factor as f64 for price calculations. + #[must_use] + pub fn decimal_factor_f64(&self) -> f64 { + f64::from(self.decimal_factor) + } + + /// Returns the earliest available tick data timestamp. + #[must_use] + pub const fn start_tick_date(&self) -> Option> { + self.start_tick_date + } + + /// Returns true if tick data is available for the given date. + #[must_use] + pub fn has_data_for(&self, date: DateTime) -> bool { + self.start_tick_date.is_some_and(|start| date >= start) + } + + /// Returns true if this is a forex instrument. + #[must_use] + pub const fn is_forex(&self) -> bool { + matches!(self.category, Category::Forex) + } + + /// Returns true if this is a cryptocurrency instrument. + #[must_use] + pub const fn is_crypto(&self) -> bool { + matches!(self.category, Category::Crypto) + } + + /// Returns true if this is an index instrument. + #[must_use] + pub const fn is_index(&self) -> bool { + matches!(self.category, Category::Index) + } + + /// Returns true if this is a stock instrument. + #[must_use] + pub const fn is_stock(&self) -> bool { + matches!(self.category, Category::Stock) + } + + /// Returns true if this is a commodity instrument. + #[must_use] + pub const fn is_commodity(&self) -> bool { + matches!(self.category, Category::Commodity) + } +} + +impl std::fmt::Display for Instrument { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{} ({})", self.name, self.id) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::TimeZone; + + #[test] + fn test_instrument_creation() { + let start = Utc.with_ymd_and_hms(2003, 5, 5, 0, 0, 0).unwrap(); + let instrument = Instrument::new( + "eurusd", + "EUR/USD", + "Euro vs US Dollar", + Category::Forex, + 100_000, + Some(start), + ); + + assert_eq!(instrument.id(), "eurusd"); + assert_eq!(instrument.name(), "EUR/USD"); + assert_eq!(instrument.decimal_factor(), 100_000); + assert!(instrument.is_forex()); + assert!(!instrument.is_crypto()); + } + + #[test] + fn test_has_data_for() { + let start = Utc.with_ymd_and_hms(2003, 5, 5, 0, 0, 0).unwrap(); + let instrument = Instrument::new( + "eurusd", + "EUR/USD", + "Euro vs US Dollar", + Category::Forex, + 100_000, + Some(start), + ); + + let before = Utc.with_ymd_and_hms(2003, 1, 1, 0, 0, 0).unwrap(); + let after = Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap(); + + assert!(!instrument.has_data_for(before)); + assert!(instrument.has_data_for(after)); + } +} diff --git a/crates/paracas-types/src/lib.rs b/crates/paracas-types/src/lib.rs new file mode 100644 index 0000000..77e154d --- /dev/null +++ b/crates/paracas-types/src/lib.rs @@ -0,0 +1,27 @@ +//! Core types for paracas tick data downloader. +//! +//! This crate provides the fundamental data structures used throughout paracas: +//! +//! - [`Tick`] - A single price tick with timestamp, ask, bid, and volumes +//! - [`RawTick`] - Raw tick from bi5 binary format before price normalization +//! - [`Instrument`] - Financial instrument with metadata +//! - [`Timeframe`] - OHLCV aggregation timeframe +//! - [`DateRange`] - Date range for data retrieval + +#![doc = include_str!("../README.md")] +#![doc(issue_tracker_base_url = "https://github.com/factordynamics/paracas/issues/")] +#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))] +#![warn(missing_docs)] +#![forbid(unsafe_code)] + +mod date_range; +mod error; +mod instrument; +mod tick; +mod timeframe; + +pub use date_range::{DateRange, HourIterator, hour_from_url}; +pub use error::{DateRangeError, ParacasError, Result}; +pub use instrument::{Category, Instrument}; +pub use tick::{RawTick, Tick}; +pub use timeframe::{Timeframe, TimeframeParseError}; diff --git a/crates/paracas-types/src/tick.rs b/crates/paracas-types/src/tick.rs new file mode 100644 index 0000000..8adeebc --- /dev/null +++ b/crates/paracas-types/src/tick.rs @@ -0,0 +1,150 @@ +//! Tick data representation. + +use chrono::{DateTime, TimeDelta, Utc}; +use serde::{Deserialize, Serialize}; + +/// A single tick representing a price update. +#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)] +pub struct Tick { + /// Timestamp of the tick (UTC). + pub timestamp: DateTime, + /// Ask (offer) price. + pub ask: f64, + /// Bid price. + pub bid: f64, + /// Volume available at the ask price. + pub ask_volume: f32, + /// Volume available at the bid price. + pub bid_volume: f32, +} + +impl Tick { + /// Creates a new tick. + #[must_use] + pub const fn new( + timestamp: DateTime, + ask: f64, + bid: f64, + ask_volume: f32, + bid_volume: f32, + ) -> Self { + Self { + timestamp, + ask, + bid, + ask_volume, + bid_volume, + } + } + + /// Returns the mid price (average of ask and bid). + #[must_use] + pub fn mid(&self) -> f64 { + (self.ask + self.bid) / 2.0 + } + + /// Returns the spread (ask - bid). + #[must_use] + pub fn spread(&self) -> f64 { + self.ask - self.bid + } + + /// Returns the total volume (ask + bid volume). + #[must_use] + pub fn total_volume(&self) -> f32 { + self.ask_volume + self.bid_volume + } +} + +/// Raw tick as read from bi5 file (before price normalization). +/// +/// The bi5 format stores ticks as 20 bytes in big-endian order: +/// - `u32`: milliseconds offset from hour start +/// - `u32`: ask price (raw, needs division by decimal factor) +/// - `u32`: bid price (raw, needs division by decimal factor) +/// - `f32`: ask volume +/// - `f32`: bid volume +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct RawTick { + /// Milliseconds offset from the hour start. + pub ms_offset: u32, + /// Raw ask price (needs division by decimal factor). + pub ask_raw: u32, + /// Raw bid price (needs division by decimal factor). + pub bid_raw: u32, + /// Ask volume. + pub ask_volume: f32, + /// Bid volume. + pub bid_volume: f32, +} + +impl RawTick { + /// Size in bytes of a raw tick record. + pub const SIZE: usize = 20; + + /// Creates a new raw tick. + #[must_use] + pub const fn new( + ms_offset: u32, + ask_raw: u32, + bid_raw: u32, + ask_volume: f32, + bid_volume: f32, + ) -> Self { + Self { + ms_offset, + ask_raw, + bid_raw, + ask_volume, + bid_volume, + } + } + + /// Normalizes the raw tick using the instrument's decimal factor. + /// + /// The decimal factor converts the raw integer prices to floating-point + /// prices. For example, EUR/USD has a decimal factor of 100,000, so a + /// raw price of 112345 becomes 1.12345. + #[must_use] + pub fn normalize(self, hour_start: DateTime, decimal_factor: f64) -> Tick { + let timestamp = hour_start + TimeDelta::milliseconds(i64::from(self.ms_offset)); + Tick { + timestamp, + ask: f64::from(self.ask_raw) / decimal_factor, + bid: f64::from(self.bid_raw) / decimal_factor, + ask_volume: self.ask_volume, + bid_volume: self.bid_volume, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::TimeZone; + + #[test] + fn test_tick_mid_price() { + let tick = Tick::new(Utc::now(), 1.1001, 1.1000, 100.0, 200.0); + assert!((tick.mid() - 1.10005).abs() < 1e-10); + } + + #[test] + fn test_tick_spread() { + let tick = Tick::new(Utc::now(), 1.1001, 1.1000, 100.0, 200.0); + assert!((tick.spread() - 0.0001).abs() < 1e-10); + } + + #[test] + fn test_raw_tick_normalize() { + let hour_start = Utc.with_ymd_and_hms(2024, 1, 1, 12, 0, 0).unwrap(); + let raw = RawTick::new(1000, 110010, 110000, 100.0, 200.0); + let tick = raw.normalize(hour_start, 100_000.0); + + assert_eq!(tick.timestamp, hour_start + TimeDelta::milliseconds(1000)); + assert!((tick.ask - 1.1001).abs() < 1e-10); + assert!((tick.bid - 1.1000).abs() < 1e-10); + assert!((tick.ask_volume - 100.0).abs() < 1e-10); + assert!((tick.bid_volume - 200.0).abs() < 1e-10); + } +} diff --git a/crates/paracas-types/src/timeframe.rs b/crates/paracas-types/src/timeframe.rs new file mode 100644 index 0000000..2f108f4 --- /dev/null +++ b/crates/paracas-types/src/timeframe.rs @@ -0,0 +1,164 @@ +//! OHLCV aggregation timeframe definitions. + +use serde::{Deserialize, Serialize}; +use std::str::FromStr; + +/// OHLCV aggregation timeframe. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, Default)] +#[serde(rename_all = "lowercase")] +pub enum Timeframe { + /// Tick-by-tick (no aggregation). + #[default] + Tick, + /// 1-second bars. + #[serde(rename = "s1")] + Second1, + /// 1-minute bars. + #[serde(rename = "m1")] + Minute1, + /// 5-minute bars. + #[serde(rename = "m5")] + Minute5, + /// 15-minute bars. + #[serde(rename = "m15")] + Minute15, + /// 30-minute bars. + #[serde(rename = "m30")] + Minute30, + /// 1-hour bars. + #[serde(rename = "h1")] + Hour1, + /// 4-hour bars. + #[serde(rename = "h4")] + Hour4, + /// Daily bars. + #[serde(rename = "d1")] + Day1, +} + +impl Timeframe { + /// Returns the duration in seconds, or None for tick data. + #[must_use] + pub const fn seconds(&self) -> Option { + match self { + Self::Tick => None, + Self::Second1 => Some(1), + Self::Minute1 => Some(60), + Self::Minute5 => Some(300), + Self::Minute15 => Some(900), + Self::Minute30 => Some(1800), + Self::Hour1 => Some(3600), + Self::Hour4 => Some(14400), + Self::Day1 => Some(86400), + } + } + + /// Returns the duration in milliseconds, or None for tick data. + #[must_use] + pub const fn milliseconds(&self) -> Option { + match self.seconds() { + Some(s) => Some(s * 1000), + None => None, + } + } + + /// Returns true if this is tick data (no aggregation). + #[must_use] + pub const fn is_tick(&self) -> bool { + matches!(self, Self::Tick) + } + + /// Returns the timeframe as a string identifier. + #[must_use] + pub const fn as_str(&self) -> &'static str { + match self { + Self::Tick => "tick", + Self::Second1 => "s1", + Self::Minute1 => "m1", + Self::Minute5 => "m5", + Self::Minute15 => "m15", + Self::Minute30 => "m30", + Self::Hour1 => "h1", + Self::Hour4 => "h4", + Self::Day1 => "d1", + } + } + + /// Returns all available timeframes. + #[must_use] + pub const fn all() -> &'static [Self] { + &[ + Self::Tick, + Self::Second1, + Self::Minute1, + Self::Minute5, + Self::Minute15, + Self::Minute30, + Self::Hour1, + Self::Hour4, + Self::Day1, + ] + } +} + +impl std::fmt::Display for Timeframe { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.as_str()) + } +} + +impl FromStr for Timeframe { + type Err = TimeframeParseError; + + fn from_str(s: &str) -> Result { + match s.to_lowercase().as_str() { + "tick" => Ok(Self::Tick), + "s1" | "1s" | "second" | "second1" => Ok(Self::Second1), + "m1" | "1m" | "minute" | "minute1" => Ok(Self::Minute1), + "m5" | "5m" | "minute5" => Ok(Self::Minute5), + "m15" | "15m" | "minute15" => Ok(Self::Minute15), + "m30" | "30m" | "minute30" => Ok(Self::Minute30), + "h1" | "1h" | "hour" | "hour1" => Ok(Self::Hour1), + "h4" | "4h" | "hour4" => Ok(Self::Hour4), + "d1" | "1d" | "day" | "day1" | "daily" => Ok(Self::Day1), + _ => Err(TimeframeParseError(s.to_string())), + } + } +} + +/// Error returned when parsing an invalid timeframe string. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TimeframeParseError(String); + +impl std::fmt::Display for TimeframeParseError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "invalid timeframe '{}', expected one of: tick, s1, m1, m5, m15, m30, h1, h4, d1", + self.0 + ) + } +} + +impl std::error::Error for TimeframeParseError {} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_timeframe_seconds() { + assert_eq!(Timeframe::Tick.seconds(), None); + assert_eq!(Timeframe::Minute1.seconds(), Some(60)); + assert_eq!(Timeframe::Hour1.seconds(), Some(3600)); + assert_eq!(Timeframe::Day1.seconds(), Some(86400)); + } + + #[test] + fn test_timeframe_parse() { + assert_eq!("m1".parse::().unwrap(), Timeframe::Minute1); + assert_eq!("1h".parse::().unwrap(), Timeframe::Hour1); + assert_eq!("H4".parse::().unwrap(), Timeframe::Hour4); + assert!("invalid".parse::().is_err()); + } +} diff --git a/lychee.toml b/lychee.toml new file mode 100644 index 0000000..c1db962 --- /dev/null +++ b/lychee.toml @@ -0,0 +1,5 @@ +include = ["README.md", "crates/*/README.md", "bin/README.md"] +accept = [200, 204, 301, 302, 303, 307, 308] +timeout = 30 +max_concurrency = 16 +cache = true