use std::collections::HashMap;
use futures::future::try_join_all;
use crate::{
Action, Candle, HistoryMeta, HistoryResponse, Interval, Range, YfClient, YfError,
core::client::{CacheMode, RetryConfig},
history::HistoryBuilder,
};
use paft::prelude::Money;
use rust_decimal::prelude::{FromPrimitive, ToPrimitive};
#[derive(Debug, Clone)]
pub struct DownloadResult {
pub series: HashMap<String, Vec<Candle>>,
pub meta: HashMap<String, Option<HistoryMeta>>,
pub actions: HashMap<String, Vec<Action>>,
pub adjusted: bool,
}
#[allow(clippy::struct_excessive_bools)]
pub struct DownloadBuilder {
client: YfClient,
symbols: Vec<String>,
range: Option<Range>,
period: Option<(i64, i64)>,
interval: Interval,
auto_adjust: bool,
back_adjust: bool,
include_prepost: bool,
include_actions: bool,
keepna: bool,
rounding: bool,
repair: bool,
cache_mode: CacheMode,
retry_override: Option<RetryConfig>,
}
impl DownloadBuilder {
#[must_use]
pub fn new(client: &YfClient) -> Self {
Self {
client: client.clone(),
symbols: Vec::new(),
range: Some(Range::M6),
period: None,
interval: Interval::D1,
auto_adjust: true,
back_adjust: false,
include_prepost: false,
include_actions: true,
keepna: false,
rounding: false,
repair: false,
cache_mode: CacheMode::Use,
retry_override: None,
}
}
#[must_use]
pub const fn cache_mode(mut self, mode: CacheMode) -> Self {
self.cache_mode = mode;
self
}
#[must_use]
pub fn retry_policy(mut self, cfg: Option<RetryConfig>) -> Self {
self.retry_override = cfg;
self
}
#[must_use]
pub fn symbols<I, S>(mut self, syms: I) -> Self
where
I: IntoIterator<Item = S>,
S: Into<String>,
{
self.symbols = syms.into_iter().map(std::convert::Into::into).collect();
self
}
#[must_use]
pub fn add_symbol(mut self, sym: impl Into<String>) -> Self {
self.symbols.push(sym.into());
self
}
#[must_use]
pub const fn range(mut self, range: Range) -> Self {
self.period = None;
self.range = Some(range);
self
}
#[must_use]
pub const fn between(
mut self,
start: chrono::DateTime<chrono::Utc>,
end: chrono::DateTime<chrono::Utc>,
) -> Self {
self.range = None;
self.period = Some((start.timestamp(), end.timestamp()));
self
}
#[must_use]
pub const fn interval(mut self, interval: Interval) -> Self {
self.interval = interval;
self
}
#[must_use]
pub const fn auto_adjust(mut self, yes: bool) -> Self {
self.auto_adjust = yes;
self
}
#[must_use]
pub const fn back_adjust(mut self, yes: bool) -> Self {
self.back_adjust = yes;
self
}
#[must_use]
pub const fn prepost(mut self, yes: bool) -> Self {
self.include_prepost = yes;
self
}
#[must_use]
pub const fn actions(mut self, yes: bool) -> Self {
self.include_actions = yes;
self
}
#[must_use]
pub const fn keepna(mut self, yes: bool) -> Self {
self.keepna = yes;
self
}
#[must_use]
pub const fn rounding(mut self, yes: bool) -> Self {
self.rounding = yes;
self
}
#[must_use]
pub const fn repair(mut self, yes: bool) -> Self {
self.repair = yes;
self
}
pub async fn run(self) -> Result<DownloadResult, YfError> {
if self.symbols.is_empty() {
return Err(YfError::InvalidParams("no symbols specified".into()));
}
let need_adjust_in_fetch = self.auto_adjust || self.back_adjust;
let period_dt = if let Some((p1, p2)) = self.period {
use chrono::{TimeZone, Utc};
let start = Utc
.timestamp_opt(p1, 0)
.single()
.ok_or_else(|| YfError::InvalidParams("invalid period1".into()))?;
let end = Utc
.timestamp_opt(p2, 0)
.single()
.ok_or_else(|| YfError::InvalidParams("invalid period2".into()))?;
Some((start, end))
} else {
None
};
let futures = self.symbols.iter().map(|sym| {
let sym = sym.clone();
let mut hb: HistoryBuilder = HistoryBuilder::new(&self.client, sym.clone())
.interval(self.interval)
.auto_adjust(need_adjust_in_fetch)
.prepost(self.include_prepost)
.actions(self.include_actions)
.keepna(self.keepna)
.cache_mode(self.cache_mode)
.retry_policy(self.retry_override.clone());
if let Some((start, end)) = period_dt {
hb = hb.between(start, end);
} else if let Some(r) = self.range {
hb = hb.range(r);
} else {
hb = hb.range(Range::M6);
}
async move {
let full: HistoryResponse = hb.fetch_full().await?;
Ok::<(String, HistoryResponse), YfError>((sym, full))
}
});
let joined: Vec<(String, HistoryResponse)> = try_join_all(futures).await?;
let mut series: std::collections::HashMap<String, Vec<Candle>> =
std::collections::HashMap::new();
let mut meta: std::collections::HashMap<String, Option<HistoryMeta>> =
std::collections::HashMap::new();
let mut actions: std::collections::HashMap<String, Vec<Action>> =
std::collections::HashMap::new();
for (sym, mut resp) in joined {
let mut v = resp.candles;
if self.back_adjust
&& let Some(raw) = resp.unadjusted_close.take()
{
for (i, c) in v.iter_mut().enumerate() {
if let Some(rc) = raw.get(i)
&& rc.amount().to_f64().is_some_and(|v| v.is_finite())
{
c.close = rc.clone();
}
}
}
if self.repair {
repair_scale_outliers(&mut v);
}
if self.rounding {
for c in &mut v {
if c.open.amount().to_f64().is_some_and(|v| v.is_finite()) {
c.open = Money::new(
rust_decimal::Decimal::from_f64(round2(
c.open.amount().to_f64().unwrap_or(0.0),
))
.unwrap_or_default(),
c.open.currency().clone(),
);
}
if c.high.amount().to_f64().is_some_and(|v| v.is_finite()) {
c.high = Money::new(
rust_decimal::Decimal::from_f64(round2(
c.high.amount().to_f64().unwrap_or(0.0),
))
.unwrap_or_default(),
c.high.currency().clone(),
);
}
if c.low.amount().to_f64().is_some_and(|v| v.is_finite()) {
c.low = Money::new(
rust_decimal::Decimal::from_f64(round2(
c.low.amount().to_f64().unwrap_or(0.0),
))
.unwrap_or_default(),
c.low.currency().clone(),
);
}
if c.close.amount().to_f64().is_some_and(|v| v.is_finite()) {
c.close = Money::new(
rust_decimal::Decimal::from_f64(round2(
c.close.amount().to_f64().unwrap_or(0.0),
))
.unwrap_or_default(),
c.close.currency().clone(),
);
}
}
}
if self.include_actions {
actions.insert(sym.clone(), resp.actions);
}
meta.insert(sym.clone(), resp.meta);
series.insert(sym, v);
}
Ok(DownloadResult {
series,
meta,
actions,
adjusted: need_adjust_in_fetch,
})
}
}
fn round2(x: f64) -> f64 {
(x * 100.0).round() / 100.0
}
fn repair_scale_outliers(rows: &mut [Candle]) {
if rows.len() < 3 {
return;
}
for i in 1..rows.len() - 1 {
let (left, right) = rows.split_at_mut(i);
let prev = &left[i - 1];
let (cur, rem) = right.split_first_mut().expect("right has at least 1");
let next = &rem[0];
let p = &prev.close;
let n = &next.close;
let c = &cur.close;
if !(p.amount().to_f64().is_some_and(|v| v.is_finite())
&& n.amount().to_f64().is_some_and(|v| v.is_finite())
&& c.amount().to_f64().is_some_and(|v| v.is_finite()))
{
continue;
}
let p_val = p.amount().to_f64().unwrap_or(0.0);
let n_val = n.amount().to_f64().unwrap_or(0.0);
let c_val = c.amount().to_f64().unwrap_or(0.0);
let baseline = f64::midpoint(p_val, n_val);
if baseline <= 0.0 {
continue;
}
let ratio = c_val / baseline;
if ratio > 50.0 && ratio < 200.0 {
let scale = if (80.0..125.0).contains(&ratio) {
0.01
} else {
1.0 / ratio
};
scale_row_prices(cur, scale);
continue;
}
if ratio > 0.0 && ratio < 0.02 {
let scale = if (0.008..0.0125).contains(&ratio) {
100.0
} else {
1.0 / ratio
};
scale_row_prices(cur, scale);
}
}
}
fn scale_row_prices(c: &mut Candle, scale: f64) {
if c.open.amount().to_f64().is_some_and(|v| v.is_finite()) {
c.open = c
.open
.mul(rust_decimal::Decimal::from_f64_retain(scale).unwrap_or_default());
}
if c.high.amount().to_f64().is_some_and(|v| v.is_finite()) {
c.high = c
.high
.mul(rust_decimal::Decimal::from_f64_retain(scale).unwrap_or_default());
}
if c.low.amount().to_f64().is_some_and(|v| v.is_finite()) {
c.low = c
.low
.mul(rust_decimal::Decimal::from_f64_retain(scale).unwrap_or_default());
}
if c.close.amount().to_f64().is_some_and(|v| v.is_finite()) {
c.close = c
.close
.mul(rust_decimal::Decimal::from_f64_retain(scale).unwrap_or_default());
}
}