portfolio_analyzer/

fetcher.rs

/// Yahoo Finance OHLCV data fetcher with retry logic
use crate::indicators::OHLCV;
use anyhow::{anyhow, Result};
use reqwest::blocking::Client;
use serde_json::Value;
use std::fs;
use std::path::{Path, PathBuf};
use std::time::Duration;
use tracing::{debug, warn};

// Supported: 1h (used for 4h aggregation), 1d, 1wk, 1mo, 3mo
pub const TIMEFRAMES: [(&str, &str, &str); 5] = [
    ("4h", "1h", "1y"),
    ("1d", "1d", "2y"),
    ("1w", "1wk", "5y"),
    ("1mo", "1mo", "10y"),
    ("3mo", "3mo", "max"),
];

pub const TF_WEIGHTS: [f64; 5] = [0.03, 0.20, 0.27, 0.30, 0.20];

// Some brokerages strip the dot/dash from Yahoo Finance symbols
fn yahoo_sym(symbol: &str) -> &str {
    match symbol {
        "BRKB" => "BRK-B",
        "BRKA" => "BRK-A",
        "BFAM" => "BFAM",
        _ => symbol,
    }
}

fn yahoo_url(symbol: &str, interval: &str, range: &str) -> String {
    format!(
        "https://query1.finance.yahoo.com/v8/finance/chart/{}?interval={}&range={}&includeAdjustedClose=true",
        yahoo_sym(symbol), interval, range
    )
}

fn safe_component(value: &str) -> String {
    value
        .chars()
        .map(|character| {
            if character.is_ascii_alphanumeric() || matches!(character, '-' | '_' | '.') {
                character
            } else {
                '_'
            }
        })
        .collect()
}

fn cache_dir() -> Option<PathBuf> {
    std::env::var_os("SD_OHLCV_CACHE_DIR").map(PathBuf::from)
}

fn cache_path(symbol: &str, interval: &str, range: &str) -> Option<PathBuf> {
    Some(cache_dir()?.join(format!(
        "{}__{}__{}.csv",
        safe_component(yahoo_sym(symbol)),
        safe_component(interval),
        safe_component(range)
    )))
}

fn offline_requested() -> bool {
    std::env::var("SD_OFFLINE")
        .is_ok_and(|value| matches!(value.to_ascii_lowercase().as_str(), "1" | "true" | "yes"))
}

/// Read OHLCV from a local CSV. Required columns are open, high, low, close,
/// and volume; names are matched case-insensitively and extra columns are
/// ignored.
pub fn load_ohlcv_csv(path: impl AsRef<Path>) -> Result<Vec<OHLCV>> {
    let path = path.as_ref();
    let mut reader =
        csv::Reader::from_path(path).map_err(|error| anyhow!("{}: {error}", path.display()))?;
    let headers = reader.headers()?.clone();
    let find = |name: &str| {
        headers
            .iter()
            .position(|header| header.trim().eq_ignore_ascii_case(name))
            .ok_or_else(|| anyhow!("{}: missing CSV column {name:?}", path.display()))
    };
    let open = find("open")?;
    let high = find("high")?;
    let low = find("low")?;
    let close = find("close")?;
    let volume = find("volume")?;
    let mut bars = Vec::new();
    for (row_number, row) in reader.records().enumerate() {
        let row = row?;
        let parse = |index: usize, name: &str| -> Result<f64> {
            row.get(index)
                .ok_or_else(|| anyhow!("row {} missing {name}", row_number + 2))?
                .trim()
                .parse::<f64>()
                .map_err(|error| anyhow!("row {} invalid {name}: {error}", row_number + 2))
        };
        let bar = OHLCV {
            open: parse(open, "open")?,
            high: parse(high, "high")?,
            low: parse(low, "low")?,
            close: parse(close, "close")?,
            volume: parse(volume, "volume")?,
        };
        if [bar.open, bar.high, bar.low, bar.close, bar.volume]
            .into_iter()
            .all(f64::is_finite)
        {
            bars.push(bar);
        }
    }
    Ok(bars)
}

fn write_ohlcv_csv(path: &Path, bars: &[OHLCV]) -> Result<()> {
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent)?;
    }
    let mut writer = csv::Writer::from_path(path)?;
    writer.write_record(["open", "high", "low", "close", "volume"])?;
    for bar in bars {
        writer.write_record([
            bar.open.to_string(),
            bar.high.to_string(),
            bar.low.to_string(),
            bar.close.to_string(),
            bar.volume.to_string(),
        ])?;
    }
    writer.flush()?;
    Ok(())
}

fn cached_daily_price(symbol: &str) -> Option<f64> {
    let directory = cache_dir()?;
    let prefix = format!("{}__1d__", safe_component(yahoo_sym(symbol)));
    let mut candidates = fs::read_dir(directory)
        .ok()?
        .filter_map(|entry| entry.ok().map(|entry| entry.path()))
        .filter(|path| {
            path.file_name()
                .and_then(|name| name.to_str())
                .is_some_and(|name| name.starts_with(&prefix) && name.ends_with(".csv"))
        })
        .collect::<Vec<_>>();
    candidates.sort();
    candidates
        .into_iter()
        .filter_map(|path| load_ohlcv_csv(path).ok())
        .filter_map(|bars| bars.last().map(|bar| bar.close))
        .next_back()
}

fn parse_chart(body: &Value) -> Result<Vec<OHLCV>> {
    let result = body
        .get("chart")
        .ok_or_else(|| anyhow!("no chart key"))?
        .get("result")
        .ok_or_else(|| anyhow!("no result key"))?
        .get(0)
        .ok_or_else(|| anyhow!("empty result"))?;

    let timestamps = result
        .get("timestamp")
        .ok_or_else(|| anyhow!("no timestamps"))?
        .as_array()
        .ok_or_else(|| anyhow!("timestamps not array"))?;

    let ind = result
        .get("indicators")
        .ok_or_else(|| anyhow!("no indicators"))?;
    let quote = ind
        .get("quote")
        .ok_or_else(|| anyhow!("no quote"))?
        .get(0)
        .ok_or_else(|| anyhow!("empty quote"))?;

    let opens = quote
        .get("open")
        .and_then(|v| v.as_array())
        .ok_or_else(|| anyhow!("no open"))?;
    let highs = quote
        .get("high")
        .and_then(|v| v.as_array())
        .ok_or_else(|| anyhow!("no high"))?;
    let lows = quote
        .get("low")
        .and_then(|v| v.as_array())
        .ok_or_else(|| anyhow!("no low"))?;
    let closes = quote
        .get("close")
        .and_then(|v| v.as_array())
        .ok_or_else(|| anyhow!("no close"))?;
    let volumes = quote
        .get("volume")
        .and_then(|v| v.as_array())
        .ok_or_else(|| anyhow!("no volume"))?;

    let n = timestamps.len();
    let mut bars = Vec::with_capacity(n);
    for i in 0..n {
        let o = opens[i].as_f64();
        let h = highs[i].as_f64();
        let l = lows[i].as_f64();
        let c = closes[i].as_f64();
        let v = volumes[i].as_f64().unwrap_or(0.0);
        if let (Some(o), Some(h), Some(l), Some(c)) = (o, h, l, c) {
            bars.push(OHLCV {
                open: o,
                high: h,
                low: l,
                close: c,
                volume: v,
            });
        }
    }
    Ok(bars)
}

/// Aggregate 1h bars into 4h bars (every 4 consecutive 1h bars → 1 4h bar)
fn aggregate_4h(bars: Vec<OHLCV>) -> Vec<OHLCV> {
    bars.chunks(4)
        .map(|chunk| {
            let o = chunk[0].open;
            let h = chunk
                .iter()
                .map(|b| b.high)
                .fold(f64::NEG_INFINITY, f64::max);
            let l = chunk.iter().map(|b| b.low).fold(f64::INFINITY, f64::min);
            let c = chunk.last().unwrap().close;
            let v = chunk.iter().map(|b| b.volume).sum();
            OHLCV {
                open: o,
                high: h,
                low: l,
                close: c,
                volume: v,
            }
        })
        .collect()
}

/// Fetch the most recent closing price (last valid close in a 5-day 1d window)
pub fn fetch_current_price(client: &Client, symbol: &str) -> Result<f64> {
    if let Some(price) = cached_daily_price(symbol) {
        debug!("{symbol} current price from offline OHLCV cache: {price:.4}");
        return Ok(price);
    }
    if offline_requested() {
        return Err(anyhow!(
            "offline cache miss for {symbol}; populate SD_OHLCV_CACHE_DIR before disconnecting"
        ));
    }
    let url = yahoo_url(symbol, "1d", "5d");
    let mut last_err = anyhow!("no attempts");

    for attempt in 0..3u32 {
        if attempt > 0 {
            std::thread::sleep(Duration::from_millis(500 * 2u64.pow(attempt)));
        }
        match client
            .get(&url)
            .header("User-Agent", "Mozilla/5.0")
            .timeout(Duration::from_secs(20))
            .send()
        {
            Ok(resp) => {
                if resp.status().as_u16() == 429 {
                    warn!("{symbol} price rate-limited, backing off");
                    std::thread::sleep(Duration::from_secs(5));
                    last_err = anyhow!("rate limited");
                    continue;
                }
                match resp.json::<Value>() {
                    Ok(body) => {
                        let closes = body
                            .get("chart")
                            .and_then(|c| c.get("result"))
                            .and_then(|r| r.get(0))
                            .and_then(|r| r.get("indicators"))
                            .and_then(|i| i.get("quote"))
                            .and_then(|q| q.get(0))
                            .and_then(|q| q.get("close"))
                            .and_then(|c| c.as_array());
                        match closes {
                            Some(arr) => {
                                if let Some(price) = arr
                                    .iter()
                                    .rev()
                                    .find_map(|v| v.as_f64().filter(|&p| p > 0.0))
                                {
                                    debug!("{symbol} current price: {price:.4}");
                                    return Ok(price);
                                }
                                last_err = anyhow!("no valid close in response");
                            }
                            None => {
                                last_err = anyhow!("no close array in response");
                            }
                        }
                    }
                    Err(e) => {
                        last_err = anyhow!("json parse: {}", e);
                    }
                }
            }
            Err(e) => {
                last_err = anyhow!("http: {}", e);
            }
        }
    }
    Err(last_err)
}

pub fn fetch_ohlcv(
    client: &Client,
    symbol: &str,
    interval: &str,
    range: &str,
    is_4h: bool,
) -> Result<Vec<OHLCV>> {
    let cached = cache_path(symbol, interval, range);
    if let Some(path) = cached.as_ref().filter(|path| path.exists()) {
        let mut bars = load_ohlcv_csv(path)?;
        if is_4h {
            bars = aggregate_4h(bars);
        }
        if bars.len() < 30 {
            return Err(anyhow!(
                "{} contains insufficient bars after aggregation: {}",
                path.display(),
                bars.len()
            ));
        }
        debug!(
            "{symbol} [{interval}] loaded {} bars from {}",
            bars.len(),
            path.display()
        );
        return Ok(bars);
    }
    if offline_requested() {
        return Err(anyhow!(
            "offline cache miss for {symbol} interval={interval} range={range}; expected {}",
            cached
                .as_deref()
                .map(|path| path.display().to_string())
                .unwrap_or_else(|| "SD_OHLCV_CACHE_DIR is not set".to_owned())
        ));
    }
    let url = yahoo_url(symbol, interval, range);
    let mut last_err = anyhow!("no attempts");

    for attempt in 0..3u32 {
        if attempt > 0 {
            std::thread::sleep(Duration::from_millis(500 * 2u64.pow(attempt)));
        }
        match client
            .get(&url)
            .header("User-Agent", "Mozilla/5.0")
            .timeout(Duration::from_secs(20))
            .send()
        {
            Ok(resp) => {
                if resp.status().as_u16() == 429 {
                    warn!("{symbol} rate-limited, backing off");
                    std::thread::sleep(Duration::from_secs(5));
                    last_err = anyhow!("rate limited");
                    continue;
                }
                match resp.json::<Value>() {
                    Ok(body) => match parse_chart(&body) {
                        Ok(mut bars) => {
                            if bars.len() < 30 {
                                return Err(anyhow!("insufficient bars: {}", bars.len()));
                            }
                            if let Some(path) = &cached {
                                if let Err(error) = write_ohlcv_csv(path, &bars) {
                                    warn!(
                                        "{symbol} [{interval}] could not write cache {}: {error}",
                                        path.display()
                                    );
                                }
                            }
                            if is_4h {
                                bars = aggregate_4h(bars);
                            }
                            if bars.len() < 30 {
                                return Err(anyhow!("insufficient 4h bars: {}", bars.len()));
                            }
                            debug!("{symbol} [{interval}] fetched {} bars", bars.len());
                            return Ok(bars);
                        }
                        Err(e) => {
                            last_err = e;
                        }
                    },
                    Err(e) => {
                        last_err = anyhow!("json parse: {}", e);
                    }
                }
            }
            Err(e) => {
                last_err = anyhow!("http: {}", e);
            }
        }
    }
    Err(last_err)
}