gavindiaz/paracas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: background downloading ⚡

Browse Source
This commit is contained in:
Andreas Bigger
2025-12-29 17:05:07 -05:00
parent 7129747d5e
commit 835bddd2ee
23 changed files with 3683 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bin/src/commands/daemon_run.rs
+159
View File
@@ -0,0 +1,159 @@
//! Hidden daemon entry point for background downloads.
//!
//! This module provides the entry point for daemon processes spawned
//! with `--daemon-run <job_id>`. It loads the job from disk and executes
//! the download tasks.
use crate::display::{Format, aggregate_ticks, write_ohlcv, write_ticks};
use anyhow::{Context, Result, bail};
use futures::StreamExt;
use paracas_daemon::{DaemonProgress, JobId, JobStatus, StateManager};
use paracas_lib::prelude::*;
use std::path::PathBuf;
/// Execute a background download job.
///
/// This is called when paracas is spawned with `--daemon-run <job_id>`.
/// The function loads the job from disk, executes all pending tasks,
/// and saves progress periodically.
pub(crate) async fn daemon_run(job_id_str: &str) -> Result<()> {
let job_id: JobId = job_id_str.parse().context("Invalid job ID")?;
let state_manager =
StateManager::with_default_path().context("Failed to initialize state manager")?;
let job = state_manager.load_job(job_id).context("Job not found")?;
if !matches!(job.status, JobStatus::Pending | JobStatus::Running) {
bail!("Job is not in a runnable state: {:?}", job.status);
}
let progress = DaemonProgress::new(state_manager.clone(), job);
// Mark job as running
{
let mut job = progress.job().await;
job.mark_started(std::process::id());
state_manager.save_job(&job)?;
}
// Process each task
let job = progress.job().await;
for (task_idx, task) in job.tasks.iter().enumerate() {
if matches!(task.status, JobStatus::Completed) {
continue; // Skip already completed tasks
}
if let Err(e) = execute_task(&progress, task_idx).await {
progress.mark_task_failed(task_idx, &e.to_string()).await;
}
progress.save_checkpoint().await?;
}
// Mark job as completed or failed based on task results
if progress.all_tasks_finished().await {
if progress.failed_tasks().await == 0 {
progress.mark_job_completed().await;
} else {
let failed_count = progress.failed_tasks().await;
let msg = format!("{} tasks failed", failed_count);
progress.mark_job_failed(&msg).await;
}
}
progress.save_checkpoint().await?;
Ok(())
}
/// Execute a single download task.
async fn execute_task(progress: &DaemonProgress, task_idx: usize) -> Result<()> {
progress.mark_task_running(task_idx).await;
let job = progress.job().await;
let task = &job.tasks[task_idx];
// Get instrument
let registry = InstrumentRegistry::global();
let instrument = registry
.get(&task.instrument_id)
.context("Unknown instrument")?;
// Parse date range
let start = chrono::NaiveDate::parse_from_str(&task.start_date, "%Y-%m-%d")?;
let end = chrono::NaiveDate::parse_from_str(&task.end_date, "%Y-%m-%d")?;
let range = DateRange::new(start, end)?;
// Create client
let config = ClientConfig {
concurrency: job.concurrency,
..Default::default()
};
let client = DownloadClient::new(config)?;
// Download ticks
let mut all_ticks: Vec<Tick> = Vec::new();
let mut stream = paracas_lib::tick_stream_resilient(&client, instrument, range);
let mut hours_completed = 0u64;
while let Some(batch) = stream.next().await {
all_ticks.extend(batch.ticks);
hours_completed += 1;
// Update progress periodically (every 10 hours)
if hours_completed.is_multiple_of(10) {
progress
.update_task_progress(task_idx, hours_completed, all_ticks.len() as u64)
.await;
}
}
// Parse timeframe and aggregate if needed
let timeframe = task
.timeframe
.parse::<Timeframe>()
.map_err(|e| anyhow::anyhow!("{e}"))?;
// Parse format
let format = parse_format(&task.format)?;
// Write output
let output_path = task.output_path.clone();
write_output(&all_ticks, &output_path, format, timeframe)?;
let bytes_written = std::fs::metadata(&output_path)
.map(|m| m.len())
.unwrap_or(0);
progress.mark_task_completed(task_idx, bytes_written).await;
Ok(())
}
/// Parse a format string into a Format enum.
fn parse_format(format: &str) -> Result<Format> {
match format.to_lowercase().as_str() {
"csv" => Ok(Format::Csv),
"json" => Ok(Format::Json),
"ndjson" => Ok(Format::Ndjson),
"parquet" => Ok(Format::Parquet),
_ => bail!("Unknown format: {}", format),
}
}
/// Write ticks or OHLCV data to the output file.
fn write_output(
ticks: &[Tick],
output: &PathBuf,
format: Format,
timeframe: Timeframe,
) -> Result<()> {
if timeframe.is_tick() {
write_ticks(ticks, output, format)?;
} else {
let bars = aggregate_ticks(ticks, timeframe);
write_ohlcv(&bars, output, format)?;
}
Ok(())
}
bin/src/commands/download.rs
+221
View File
@@ -0,0 +1,221 @@
//! Download command implementation.
//!
//! This module handles downloading tick data from Dukascopy and writing it to various output formats.
use crate::display::{Format, aggregate_ticks, write_ohlcv, write_ticks};
use anyhow::{Context, Result};
use chrono::NaiveDate;
use futures::StreamExt;
use indicatif::{ProgressBar, ProgressStyle};
use paracas_daemon::{DaemonSpawner, DownloadJob, InstrumentTask, StateManager};
use paracas_lib::prelude::*;
use std::path::PathBuf;
/// Download tick data for an instrument.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn download(
instrument_id: &str,
start_str: Option<&str>,
end_str: Option<&str>,
output: Option<PathBuf>,
format: Format,
timeframe_str: Option<&str>,
concurrency: usize,
background: bool,
_yes: bool,
quiet: bool,
) -> Result<()> {
// Handle background mode
if background {
return spawn_background_download(
instrument_id,
start_str,
end_str,
output,
format,
timeframe_str,
concurrency,
);
}
// 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 <instrument>.<format>)
let output = output
.unwrap_or_else(|| PathBuf::from(format!("{}.{}", instrument_id, format.extension())));
// Parse timeframe
let timeframe = match timeframe_str {
Some(tf) => tf
.parse::<Timeframe>()
.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<Tick> = 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(())
}
/// Spawn a background download job for a single instrument.
#[allow(clippy::too_many_arguments)]
fn spawn_background_download(
instrument_id: &str,
start_str: Option<&str>,
end_str: Option<&str>,
output: Option<PathBuf>,
format: Format,
timeframe_str: Option<&str>,
concurrency: usize,
) -> Result<()> {
let registry = InstrumentRegistry::global();
let instrument = registry
.get(instrument_id)
.with_context(|| format!("Unknown instrument: {instrument_id}"))?;
// Determine start date
let start = start_str
.map(|s| s.to_string())
.or_else(|| {
instrument
.start_tick_date()
.map(|d| d.format("%Y-%m-%d").to_string())
})
.unwrap_or_else(|| "2003-05-05".to_string());
// Determine end date
let end = end_str
.map(|s| s.to_string())
.unwrap_or_else(|| chrono::Utc::now().format("%Y-%m-%d").to_string());
// Determine output path
let output_path = output
.unwrap_or_else(|| PathBuf::from(format!("{}.{}", instrument_id, format.extension())));
// Make output path absolute
let output_path = if output_path.is_absolute() {
output_path
} else {
std::env::current_dir()
.unwrap_or_else(|_| PathBuf::from("."))
.join(output_path)
};
// Calculate total hours for progress tracking
let start_date = NaiveDate::parse_from_str(&start, "%Y-%m-%d")?;
let end_date = NaiveDate::parse_from_str(&end, "%Y-%m-%d")?;
let range = DateRange::new(start_date, end_date)?;
// Determine timeframe string (default to "tick")
let timeframe = timeframe_str
.map(|s| s.to_string())
.unwrap_or_else(|| "tick".to_string());
let task = InstrumentTask::new(
instrument_id.to_string(),
start,
end,
output_path,
format.to_string(),
timeframe,
range.total_hours() as u32,
);
let mut job = DownloadJob::new(vec![task], concurrency);
let state_manager =
StateManager::with_default_path().context("Failed to initialize state manager")?;
let spawner = DaemonSpawner::new(state_manager).context("Failed to create daemon spawner")?;
let job_id = spawner
.spawn(&mut job)
.context("Failed to spawn background job")?;
println!("Background download started.");
println!("Job ID: {}", job_id);
println!("Check status with: paracas status {}", job_id);
Ok(())
}
bin/src/commands/download_all.rs
+351
View File
@@ -0,0 +1,351 @@
//! Download all instruments command.
//!
//! This module handles batch downloading of multiple instruments, with support for
//! category filtering, parallel downloads, and download estimation.
use crate::display::{Format, aggregate_ticks, parse_category, write_ohlcv, write_ticks};
use anyhow::{Context, Result};
use chrono::NaiveDate;
use futures::stream::{self, StreamExt};
use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
use paracas_daemon::{DaemonSpawner, DownloadJob, InstrumentTask, StateManager};
use paracas_estimate::Estimator;
use paracas_lib::prelude::*;
use std::io::Write as _;
use std::path::PathBuf;
/// Execute the download-all command.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn download_all(
category: Option<&str>,
start_str: Option<&str>,
end_str: Option<&str>,
output_dir: PathBuf,
format: Format,
timeframe_str: Option<&str>,
parallel_instruments: usize,
concurrency: usize,
background: bool,
yes: bool,
quiet: bool,
) -> Result<()> {
// 1. Get instruments based on category filter (or all)
let registry = InstrumentRegistry::global();
let instruments: Vec<_> = match category {
Some(cat) => {
let category = parse_category(cat)?;
registry.by_category(category).collect()
}
None => registry.all().collect(),
};
if instruments.is_empty() {
anyhow::bail!("No instruments found matching criteria");
}
// Parse end date (default to today)
let today = chrono::Utc::now().date_naive();
let end = match end_str {
Some(s) => NaiveDate::parse_from_str(s, "%Y-%m-%d")
.with_context(|| format!("Invalid end date: {s}"))?,
None => today,
};
// Parse start date or use earliest instrument date
let start = match start_str {
Some(s) => NaiveDate::parse_from_str(s, "%Y-%m-%d")
.with_context(|| format!("Invalid start date: {s}"))?,
None => {
// Use the earliest start date among all selected instruments
instruments
.iter()
.filter_map(|i| i.start_tick_date())
.map(|dt| dt.date_naive())
.min()
.unwrap_or_else(|| NaiveDate::from_ymd_opt(2003, 5, 5).expect("valid date"))
}
};
let range = DateRange::new(start, end)?;
// 2. Show estimate and get confirmation
let estimator = Estimator::global();
let estimate = estimator.estimate_batch(&instruments, &range);
if !yes && !quiet {
println!("Download plan:");
println!(" Instruments: {}", instruments.len());
println!(" Date range: {} to {}", start, end);
println!(
" Estimated download size: {}",
Estimator::format_bytes(estimate.estimated_compressed_bytes)
);
println!(
" Estimated output size: {}",
Estimator::format_bytes(estimate.estimated_output_bytes)
);
println!(
" Estimated time: {}",
Estimator::format_duration(estimate.estimated_duration)
);
println!();
// Simple y/n confirmation
print!("Proceed with download? [y/N] ");
std::io::stdout().flush()?;
let mut input = String::new();
std::io::stdin().read_line(&mut input)?;
if !input.trim().eq_ignore_ascii_case("y") {
println!("Cancelled.");
return Ok(());
}
}
// 3. If background mode, spawn daemon
if background {
return spawn_background_download_all(
&instruments,
start,
end,
&output_dir,
format,
timeframe_str,
concurrency,
);
}
// 4. Create output directory if needed
std::fs::create_dir_all(&output_dir)?;
// 5. Parse timeframe
let timeframe = match timeframe_str {
Some(tf) => tf
.parse::<Timeframe>()
.map_err(|e| anyhow::anyhow!("{e}"))?,
None => Timeframe::Tick,
};
// 6. Download instruments in parallel
let multi_progress = MultiProgress::new();
let results: Vec<_> = stream::iter(instruments.into_iter())
.map(|instrument| {
let pb = multi_progress.add(ProgressBar::new(100));
pb.set_style(
ProgressStyle::default_bar()
.template("{prefix:.bold} [{bar:30.cyan/blue}] {percent}% {msg}")
.unwrap()
.progress_chars("=>-"),
);
pb.set_prefix(format!("{:>12}", instrument.id()));
download_single_instrument(
instrument,
start,
end,
output_dir.clone(),
format,
timeframe,
concurrency,
pb,
quiet,
)
})
.buffer_unordered(parallel_instruments)
.collect()
.await;
// 7. Report summary
let (successes, failures): (Vec<_>, Vec<_>) = results.iter().partition(|r| r.is_ok());
if !quiet {
println!("\nDownload complete:");
println!(" Successful: {}", successes.len());
if !failures.is_empty() {
println!(" Failed: {}", failures.len());
for (i, err) in failures.iter().enumerate() {
if let Err(e) = err {
println!(" {}: {}", i + 1, e);
}
}
}
}
// Return error if any downloads failed
if !failures.is_empty() {
anyhow::bail!(
"{} out of {} downloads failed",
failures.len(),
successes.len() + failures.len()
);
}
Ok(())
}
/// Download a single instrument with progress tracking.
#[allow(clippy::too_many_arguments)]
async fn download_single_instrument(
instrument: &Instrument,
start: NaiveDate,
end: NaiveDate,
output_dir: PathBuf,
format: Format,
timeframe: Timeframe,
concurrency: usize,
progress: ProgressBar,
quiet: bool,
) -> Result<()> {
// Adjust start date based on instrument's available data
let effective_start = instrument
.start_tick_date()
.map_or(start, |instrument_start| {
let instrument_start_date = instrument_start.date_naive();
if start < instrument_start_date {
instrument_start_date
} else {
start
}
});
// Skip if the instrument has no data in the requested range
if effective_start > end {
progress.finish_with_message("skipped (no data)");
return Ok(());
}
let range = DateRange::new(effective_start, end)?;
let total_hours = range.total_hours() as u64;
progress.set_length(total_hours);
// Create client
let config = ClientConfig {
concurrency,
..Default::default()
};
let client = DownloadClient::new(config)?;
// Download and collect ticks
let mut all_ticks: Vec<Tick> = 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 tick_count = all_ticks.len();
let finish_msg = if skipped_hours > 0 {
format!("{} ticks ({} hrs skipped)", tick_count, skipped_hours)
} else {
format!("{} ticks", tick_count)
};
progress.finish_with_message(finish_msg);
// Determine output path
let output_path = output_dir.join(format!("{}.{}", instrument.id(), format.extension()));
// Aggregate if needed
if timeframe.is_tick() {
write_ticks(&all_ticks, &output_path, format)?;
} else {
let bars = aggregate_ticks(&all_ticks, timeframe);
write_ohlcv(&bars, &output_path, format)?;
}
if !quiet {
progress.println(format!(" Written: {}", output_path.display()));
}
Ok(())
}
/// Spawn a background download job for multiple instruments.
#[allow(clippy::too_many_arguments)]
fn spawn_background_download_all(
instruments: &[&Instrument],
start: NaiveDate,
end: NaiveDate,
output_dir: &PathBuf,
format: Format,
timeframe_str: Option<&str>,
concurrency: usize,
) -> Result<()> {
// Make output directory absolute
let output_dir = if output_dir.is_absolute() {
output_dir.clone()
} else {
std::env::current_dir()
.unwrap_or_else(|_| PathBuf::from("."))
.join(output_dir)
};
// Create output directory if needed
std::fs::create_dir_all(&output_dir)?;
// Determine timeframe string (default to "tick")
let timeframe = timeframe_str
.map(|s| s.to_string())
.unwrap_or_else(|| "tick".to_string());
// Create tasks for each instrument
let mut tasks = Vec::with_capacity(instruments.len());
for instrument in instruments {
// Adjust start date based on instrument's available data
let effective_start = instrument
.start_tick_date()
.map_or(start, |instrument_start| {
let instrument_start_date = instrument_start.date_naive();
if start < instrument_start_date {
instrument_start_date
} else {
start
}
});
// Skip if the instrument has no data in the requested range
if effective_start > end {
continue;
}
let range = DateRange::new(effective_start, end)?;
let output_path = output_dir.join(format!("{}.{}", instrument.id(), format.extension()));
let task = InstrumentTask::new(
instrument.id().to_string(),
effective_start.format("%Y-%m-%d").to_string(),
end.format("%Y-%m-%d").to_string(),
output_path,
format.to_string(),
timeframe.clone(),
range.total_hours() as u32,
);
tasks.push(task);
}
if tasks.is_empty() {
anyhow::bail!("No instruments with data in the specified date range");
}
let mut job = DownloadJob::new(tasks, concurrency);
let state_manager =
StateManager::with_default_path().context("Failed to initialize state manager")?;
let spawner = DaemonSpawner::new(state_manager).context("Failed to create daemon spawner")?;
let job_id = spawner
.spawn(&mut job)
.context("Failed to spawn background job")?;
println!("Background download started.");
println!("Job ID: {}", job_id);
println!("Instruments: {}", job.tasks.len());
println!("Check status with: paracas status {}", job_id);
Ok(())
}
bin/src/commands/info.rs
+104
View File
@@ -0,0 +1,104 @@
//! Info command implementation.
//!
//! This module handles displaying detailed information about a specific instrument,
//! including size estimates for different time periods.
use anyhow::{Context, Result};
use paracas_estimate::Estimator;
use paracas_lib::prelude::*;
/// Show detailed information about an instrument, including size estimates.
pub(crate) fn show_info(instrument_id: &str) -> Result<()> {
let registry = InstrumentRegistry::global();
let instrument = registry
.get(instrument_id)
.with_context(|| format!("Unknown instrument: {instrument_id}"))?;
// Basic info
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"));
// Calculate estimates for different time periods
let today = chrono::Utc::now().date_naive();
let estimator = Estimator::global();
println!("\nDownload Estimates:");
println!(
"{:<20} {:>12} {:>12} {:>12}",
"PERIOD", "DOWNLOAD", "OUTPUT (CSV)", "EST. TIME"
);
println!("{}", "-".repeat(60));
// Last 1 day
if let Ok(range) = DateRange::new(today - chrono::Duration::days(1), today) {
let est = estimator.estimate_single(instrument, &range);
println!(
"{:<20} {:>12} {:>12} {:>12}",
"Last 1 day",
Estimator::format_bytes(est.estimated_compressed_bytes),
Estimator::format_bytes(est.estimated_output_bytes),
Estimator::format_duration(est.estimated_duration),
);
}
// Last 1 week
if let Ok(range) = DateRange::new(today - chrono::Duration::days(7), today) {
let est = estimator.estimate_single(instrument, &range);
println!(
"{:<20} {:>12} {:>12} {:>12}",
"Last 1 week",
Estimator::format_bytes(est.estimated_compressed_bytes),
Estimator::format_bytes(est.estimated_output_bytes),
Estimator::format_duration(est.estimated_duration),
);
}
// Last 1 month
if let Ok(range) = DateRange::new(today - chrono::Duration::days(30), today) {
let est = estimator.estimate_single(instrument, &range);
println!(
"{:<20} {:>12} {:>12} {:>12}",
"Last 1 month",
Estimator::format_bytes(est.estimated_compressed_bytes),
Estimator::format_bytes(est.estimated_output_bytes),
Estimator::format_duration(est.estimated_duration),
);
}
// Last 1 year
if let Ok(range) = DateRange::new(today - chrono::Duration::days(365), today) {
let est = estimator.estimate_single(instrument, &range);
println!(
"{:<20} {:>12} {:>12} {:>12}",
"Last 1 year",
Estimator::format_bytes(est.estimated_compressed_bytes),
Estimator::format_bytes(est.estimated_output_bytes),
Estimator::format_duration(est.estimated_duration),
);
}
// Full history (from start to today)
let start_date = start.date_naive();
if let Ok(range) = DateRange::new(start_date, today) {
let est = estimator.estimate_single(instrument, &range);
let years = (today - start_date).num_days() as f64 / 365.25;
println!(
"{:<20} {:>12} {:>12} {:>12}",
format!("Full history ({:.1}y)", years),
Estimator::format_bytes(est.estimated_compressed_bytes),
Estimator::format_bytes(est.estimated_output_bytes),
Estimator::format_duration(est.estimated_duration),
);
}
println!("\nNote: Estimates are based on historical averages and may vary.");
}
Ok(())
}
bin/src/commands/list.rs
+41
View File
@@ -0,0 +1,41 @@
//! List command implementation.
//!
//! This module handles listing available instruments with optional filtering.
use crate::display::parse_category;
use anyhow::Result;
use paracas_lib::prelude::*;
/// List available instruments with optional category filter or search pattern.
pub(crate) 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(())
}
bin/src/commands/mod.rs
+8
View File
@@ -0,0 +1,8 @@
//! CLI command implementations.
pub(crate) mod daemon_run;
pub(crate) mod download;
pub(crate) mod download_all;
pub(crate) mod info;
pub(crate) mod list;
pub(crate) mod status;
bin/src/commands/status.rs
+192
View File
@@ -0,0 +1,192 @@
//! Background job status command.
use anyhow::{Context, Result};
use paracas_daemon::{DownloadJob, JobStatus, StateManager};
/// Execute the status command.
pub(crate) fn status(
job_id: Option<&str>,
running_only: bool,
show_all: bool,
follow: Option<u64>,
cancel_id: Option<&str>,
) -> Result<()> {
let state_manager =
StateManager::with_default_path().context("Failed to initialize state manager")?;
// Handle cancellation request
if let Some(id) = cancel_id {
return cancel_job(&state_manager, id);
}
// Handle follow/watch mode
if let Some(interval) = follow {
return watch_jobs(&state_manager, job_id, interval);
}
// Show specific job or list jobs
#[allow(clippy::option_if_let_else)]
match job_id {
Some(id) => show_job_detail(&state_manager, id),
None => list_jobs(&state_manager, running_only, show_all),
}
}
fn show_job_detail(state: &StateManager, job_id: &str) -> Result<()> {
let id = job_id.parse().context("Invalid job ID format")?;
let job = state.load_job(id).context("Job not found")?;
println!("Job: {}", job.id);
println!("Status: {:?}", job.status);
println!("Created: {}", job.created_at.format("%Y-%m-%d %H:%M:%S"));
if let Some(started) = job.started_at {
println!("Started: {}", started.format("%Y-%m-%d %H:%M:%S"));
}
if let Some(completed) = job.completed_at {
println!("Completed: {}", completed.format("%Y-%m-%d %H:%M:%S"));
}
println!("Progress: {:.1}%", job.progress_percent());
println!(
"PID: {}",
job.pid
.map(|p| p.to_string())
.unwrap_or_else(|| "N/A".into())
);
println!(
"Log: {}",
job.log_file
.as_ref()
.map(|p| p.display().to_string())
.unwrap_or_else(|| "N/A".into())
);
println!("\nTasks:");
for (i, task) in job.tasks.iter().enumerate() {
let progress = if task.hours_total > 0 {
(task.hours_completed as f64 / task.hours_total as f64) * 100.0
} else {
0.0
};
println!(
" {}. {} [{:?}] {:.1}% ({}/{} hours)",
i + 1,
task.instrument_id,
task.status,
progress,
task.hours_completed,
task.hours_total,
);
if let Some(ref err) = task.error_message {
println!(" Error: {}", err);
}
}
Ok(())
}
fn list_jobs(state: &StateManager, running_only: bool, show_all: bool) -> Result<()> {
let jobs = state.list_jobs()?;
let filtered: Vec<_> = jobs
.into_iter()
.filter(|job| {
if running_only {
matches!(job.status, JobStatus::Running | JobStatus::Pending)
} else if show_all {
true
} else {
// Default: show recent (last 24h) or active
let is_recent = job.created_at > chrono::Utc::now() - chrono::Duration::hours(24);
is_recent || matches!(job.status, JobStatus::Running | JobStatus::Pending)
}
})
.collect();
if filtered.is_empty() {
println!("No jobs found.");
if !show_all {
println!("Use --all to show all historical jobs.");
}
return Ok(());
}
println!(
"{:<36} {:<12} {:<10} {:<20}",
"JOB ID", "STATUS", "PROGRESS", "CREATED"
);
println!("{}", "-".repeat(80));
for job in &filtered {
println!(
"{:<36} {:<12} {:>8.1}% {:<20}",
job.id,
format!("{:?}", job.status),
job.progress_percent(),
job.created_at.format("%Y-%m-%d %H:%M"),
);
}
println!("\nTotal: {} jobs", filtered.len());
Ok(())
}
fn cancel_job(state: &StateManager, job_id: &str) -> Result<()> {
let id = job_id.parse().context("Invalid job ID format")?;
let mut job: DownloadJob = state.load_job(id).context("Job not found")?;
if !matches!(job.status, JobStatus::Running | JobStatus::Pending) {
anyhow::bail!("Job is not running (status: {:?})", job.status);
}
// Send SIGTERM to the process if running
if let Some(pid) = job.pid {
#[cfg(unix)]
{
use std::process::Command;
let _ = Command::new("kill")
.args(["-TERM", &pid.to_string()])
.status();
}
#[cfg(windows)]
{
use std::process::Command;
let _ = Command::new("taskkill")
.args(["/PID", &pid.to_string()])
.status();
}
}
job.status = JobStatus::Cancelled;
state.save_job(&job)?;
println!("Job {} cancelled.", id);
Ok(())
}
fn watch_jobs(state: &StateManager, job_id: Option<&str>, interval_secs: u64) -> Result<()> {
use std::io::Write;
let interval = std::time::Duration::from_secs(interval_secs);
loop {
// Clear screen
print!("\x1B[2J\x1B[1;1H");
std::io::stdout().flush()?;
println!(
"Watching jobs (refresh every {}s, Ctrl+C to exit)\n",
interval_secs
);
match job_id {
Some(id) => show_job_detail(state, id)?,
None => list_jobs(state, true, false)?,
}
std::thread::sleep(interval);
}
}
bin/src/display.rs
+138
View File
@@ -0,0 +1,138 @@
//! Display utilities and output formatting for the paracas CLI.
use anyhow::{Result, bail};
use clap::ValueEnum;
use paracas_lib::prelude::*;
use std::fs::File;
use std::io::BufWriter;
use std::path::PathBuf;
/// Output format for downloaded data.
#[derive(Clone, Copy, ValueEnum)]
pub(crate) enum Format {
Csv,
Json,
Ndjson,
Parquet,
}
impl Format {
/// Returns the file extension for this format.
pub(crate) const fn extension(&self) -> &'static str {
match self {
Self::Csv => "csv",
Self::Json => "json",
Self::Ndjson => "ndjson",
Self::Parquet => "parquet",
}
}
}
impl std::fmt::Display for Format {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.extension())
}
}
/// Aggregate ticks into OHLCV bars using the given timeframe.
pub(crate) fn aggregate_ticks(ticks: &[Tick], timeframe: Timeframe) -> Vec<Ohlcv> {
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
}
/// Write ticks to a file in the specified format.
pub(crate) 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(())
}
/// Write OHLCV bars to a file in the specified format.
pub(crate) 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(())
}
/// Parse a category string into a Category enum.
pub(crate) fn parse_category(s: &str) -> Result<Category> {
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
),
}
}
bin/src/main.rs
+135 -290
View File
@@ -1,22 +1,21 @@
//! 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 anyhow::{Context, Result};
use clap::{Parser, Subcommand};
use std::path::PathBuf;
mod commands;
mod display;
use display::Format;
#[derive(Parser)]
#[command(name = "paracas")]
#[command(about = "High-performance Dukascopy tick data downloader", long_about = None)]
#[command(version)]
struct Cli {
#[command(subcommand)]
command: Commands,
command: Option<Commands>,
/// Verbosity level (-v, -vv, -vvv)
#[arg(short, long, action = clap::ArgAction::Count, global = true)]
@@ -25,6 +24,10 @@ struct Cli {
/// Quiet mode (suppress progress output)
#[arg(short, long, global = true)]
quiet: bool,
/// Hidden: Run as daemon with job ID (internal use only)
#[arg(long, hide = true)]
daemon_run: Option<String>,
}
#[derive(Subcommand)]
@@ -57,6 +60,14 @@ enum Commands {
/// Maximum concurrent downloads
#[arg(long, default_value = "32")]
concurrency: usize,
/// Run in background as daemon
#[arg(long)]
background: bool,
/// Skip confirmation prompt (for background mode)
#[arg(long)]
yes: bool,
},
/// List available instruments
@@ -75,21 +86,88 @@ enum Commands {
/// Instrument identifier
instrument: String,
},
}
#[derive(Clone, Copy, ValueEnum)]
enum Format {
Csv,
Json,
Ndjson,
Parquet,
/// Check background job status
Status {
/// Specific job ID to check
job_id: Option<String>,
/// Show only running jobs
#[arg(long)]
running: bool,
/// Show all jobs (including completed)
#[arg(long)]
all: bool,
/// Follow/watch mode (refresh every N seconds)
#[arg(short, long)]
follow: Option<u64>,
/// Cancel a running job
#[arg(long)]
cancel: Option<String>,
},
/// Download all instruments (or filter by category)
DownloadAll {
/// Filter by category (forex, crypto, index, commodity)
#[arg(short, long)]
category: Option<String>,
/// Start date (YYYY-MM-DD). Defaults to each instrument's earliest data.
#[arg(short, long)]
start: Option<String>,
/// End date (YYYY-MM-DD). Defaults to today.
#[arg(short, long)]
end: Option<String>,
/// Output directory. Files named <instrument>.<format>
#[arg(short, long, default_value = ".")]
output_dir: PathBuf,
/// Output format
#[arg(short, long, value_enum, default_value = "csv")]
format: Format,
/// OHLCV aggregation timeframe (omit for raw ticks)
#[arg(short, long)]
timeframe: Option<String>,
/// Maximum concurrent instruments to download
#[arg(long, default_value = "4")]
parallel_instruments: usize,
/// Maximum concurrent HTTP requests per instrument
#[arg(long, default_value = "32")]
concurrency: usize,
/// Run in background as daemon
#[arg(long)]
background: bool,
/// Skip confirmation prompt
#[arg(long)]
yes: bool,
},
}
#[tokio::main]
async fn main() -> Result<()> {
let cli = Cli::parse();
match cli.command {
// Check for daemon mode first (internal use)
if let Some(job_id) = cli.daemon_run {
return commands::daemon_run::daemon_run(&job_id).await;
}
// Require a command otherwise
let command = cli
.command
.context("No command provided. Use --help for usage.")?;
match command {
Commands::Download {
instrument,
start,
@@ -98,8 +176,10 @@ async fn main() -> Result<()> {
format,
timeframe,
concurrency,
background,
yes,
} => {
download(
commands::download::download(
&instrument,
start.as_deref(),
end.as_deref(),
@@ -107,284 +187,49 @@ async fn main() -> Result<()> {
format,
timeframe.as_deref(),
concurrency,
background,
yes,
cli.quiet,
)
.await
}
Commands::List { category, search } => {
list_instruments(category.as_deref(), search.as_deref())
commands::list::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<PathBuf>,
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 <instrument>.<format>)
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::<Timeframe>()
.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<Tick> = 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;
Commands::Info { instrument } => commands::info::show_info(&instrument),
Commands::Status {
job_id,
running,
all,
follow,
cancel,
} => commands::status::status(job_id.as_deref(), running, all, follow, cancel.as_deref()),
Commands::DownloadAll {
category,
start,
end,
output_dir,
format,
timeframe,
parallel_instruments,
concurrency,
background,
yes,
} => {
commands::download_all::download_all(
category.as_deref(),
start.as_deref(),
end.as_deref(),
output_dir,
format,
timeframe.as_deref(),
parallel_instruments,
concurrency,
background,
yes,
cli.quiet,
)
.await
}
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<Ohlcv> {
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<Category> {
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
),
}
}