hypercall/runtime/tasks/

historical_theo.rs

use std::sync::Arc;
use std::time::Duration;

use metrics::counter;
use rust_decimal::Decimal;
use tokio::time::{interval_at, Instant, MissedTickBehavior};
use tracing::{debug, error, info};

use super::historical_pnl::{INTERVAL_1D_MS, INTERVAL_1H_MS, INTERVAL_5M_MS};
use crate::read_cache::greeks::GreeksCache;
use crate::read_cache::instruments_registry::InstrumentsCache;
use hypercall_db::AnalyticsWriter;
use std::collections::BTreeSet;

/// Configuration for the historical theo snapshot background task.
#[derive(Debug, Clone)]
pub struct HistoricalTheoTaskConfig {
    /// Retention cap per `(symbol, interval)` in persisted snapshots.
    pub max_periods: i64,
    /// Capture cadence for the 5m bucket writer.
    pub capture_every_5m_ms: i64,
    /// Capture cadence for the 1h bucket writer.
    pub capture_every_1h_ms: i64,
    /// Capture cadence for the 1d bucket writer.
    pub capture_every_1d_ms: i64,
}

impl Default for HistoricalTheoTaskConfig {
    fn default() -> Self {
        Self {
            max_periods: 100,
            capture_every_5m_ms: INTERVAL_5M_MS,
            capture_every_1h_ms: INTERVAL_1H_MS,
            capture_every_1d_ms: INTERVAL_1D_MS,
        }
    }
}

impl HistoricalTheoTaskConfig {
    pub fn from_env() -> Self {
        let default = Self::default();
        Self {
            max_periods: read_positive_i64("HISTORICAL_THEO_MAX_PERIODS", default.max_periods),
            capture_every_5m_ms: read_positive_i64(
                "HISTORICAL_THEO_CAPTURE_EVERY_5M_MS",
                default.capture_every_5m_ms,
            ),
            capture_every_1h_ms: read_positive_i64(
                "HISTORICAL_THEO_CAPTURE_EVERY_1H_MS",
                default.capture_every_1h_ms,
            ),
            capture_every_1d_ms: read_positive_i64(
                "HISTORICAL_THEO_CAPTURE_EVERY_1D_MS",
                default.capture_every_1d_ms,
            ),
        }
    }
}

#[derive(Clone)]
pub struct HistoricalTheoSnapshotTask {
    instruments_cache: Arc<InstrumentsCache>,
    greeks_cache: Arc<GreeksCache>,
    db: Arc<dyn AnalyticsWriter>,
    config: HistoricalTheoTaskConfig,
}

impl HistoricalTheoSnapshotTask {
    pub fn new(
        instruments_cache: Arc<InstrumentsCache>,
        greeks_cache: Arc<GreeksCache>,
        db: Arc<dyn AnalyticsWriter>,
        config: HistoricalTheoTaskConfig,
    ) -> Self {
        Self {
            instruments_cache,
            greeks_cache,
            db,
            config,
        }
    }

    async fn capture_interval(&self, interval_ms: i64, label: &'static str) {
        let bucket_timestamp_ms = align_timestamp_ms(now_timestamp_ms(), interval_ms);
        let active_instrument_symbols: Vec<String> = self
            .instruments_cache
            .get_all()
            .await
            .into_iter()
            .filter(|instrument| instrument.status.is_active())
            .map(|instrument| instrument.id)
            .collect();
        let configured_underlyings = self.greeks_cache.configured_underlyings();
        let snapshot_symbols =
            build_snapshot_symbols(active_instrument_symbols, configured_underlyings);

        if snapshot_symbols.is_empty() {
            debug!(
                "Historical theo snapshot skipped for {}: no active instruments or configured underlyings",
                label
            );
            return;
        }

        let mut snapshots: Vec<(String, Decimal)> = Vec::with_capacity(snapshot_symbols.len());
        for symbol in snapshot_symbols {
            let theoretical_mark = match self.greeks_cache.get_theoretical_mark(&symbol).await {
                Ok(price) => price,
                Err(error) => {
                    error!(
                        symbol,
                        interval = label,
                        bucket_timestamp_ms,
                        error = %error,
                        "Historical theo snapshot failed to compute theoretical mark"
                    );
                    counter!(
                        "ht_historical_theo_snapshot_total",
                        "interval" => label,
                        "status" => "error"
                    )
                    .increment(1);
                    continue;
                }
            };

            let Some(theoretical_mark) = Decimal::from_f64_retain(theoretical_mark) else {
                error!(
                    symbol,
                    interval = label,
                    bucket_timestamp_ms,
                    theoretical_mark,
                    "Historical theo snapshot failed to convert theoretical mark"
                );
                counter!(
                    "ht_historical_theo_snapshot_total",
                    "interval" => label,
                    "status" => "error"
                )
                .increment(1);
                continue;
            };

            snapshots.push((symbol, theoretical_mark));
        }

        if snapshots.is_empty() {
            debug!(
                "Historical theo snapshot skipped for {}: no symbols with computed theos",
                label
            );
            return;
        }

        match self
            .db
            .upsert_historical_theo_batch(
                interval_ms,
                bucket_timestamp_ms,
                &snapshots,
                self.config.max_periods,
            )
            .await
        {
            Ok(affected) => {
                info!(
                    "Historical theo snapshot captured interval={} symbols={} affected_rows={} bucket_ts={}",
                    label,
                    snapshots.len(),
                    affected,
                    bucket_timestamp_ms
                );
                counter!("ht_historical_theo_snapshot_total", "interval" => label, "status" => "success")
                    .increment(snapshots.len() as u64);
            }
            Err(error) => {
                error!(
                    interval = label,
                    bucket_timestamp_ms,
                    error = %error,
                    "Failed to persist historical theo snapshot"
                );
                counter!("ht_historical_theo_snapshot_total", "interval" => label, "status" => "error")
                    .increment(snapshots.len() as u64);
            }
        }
    }
}

fn build_snapshot_symbols(
    active_instrument_symbols: Vec<String>,
    configured_underlyings: Vec<String>,
) -> Vec<String> {
    let configured_perp_symbols = configured_underlyings
        .iter()
        .map(|underlying| format!("{}-PERP", underlying))
        .collect::<Vec<_>>();

    active_instrument_symbols
        .into_iter()
        .chain(configured_underlyings.iter().cloned())
        .chain(configured_perp_symbols)
        .collect::<BTreeSet<_>>()
        .into_iter()
        .collect()
}

fn aligned_interval(interval_ms: i64) -> tokio::time::Interval {
    let now_ms = now_timestamp_ms();
    let next_ms = next_bucket_start_ms(now_ms, interval_ms);
    let delay_ms = (next_ms - now_ms).max(0) as u64;

    let start_at = Instant::now() + Duration::from_millis(delay_ms);
    let mut ticker = interval_at(start_at, Duration::from_millis(interval_ms as u64));
    ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
    ticker
}

fn now_timestamp_ms() -> i64 {
    chrono::Utc::now().timestamp_millis()
}

fn align_timestamp_ms(timestamp_ms: i64, interval_ms: i64) -> i64 {
    timestamp_ms - (timestamp_ms % interval_ms)
}

fn next_bucket_start_ms(timestamp_ms: i64, interval_ms: i64) -> i64 {
    align_timestamp_ms(timestamp_ms, interval_ms) + interval_ms
}

fn read_positive_i64(key: &str, default: i64) -> i64 {
    match std::env::var(key) {
        Ok(value) => match value.parse::<i64>() {
            Ok(parsed) if parsed > 0 => parsed,
            _ => {
                error!(
                    "Invalid {}='{}' (must be positive integer), using default {}",
                    key, value, default
                );
                default
            }
        },
        Err(_) => default,
    }
}

#[async_trait::async_trait]
impl crate::shared::service::Service for HistoricalTheoSnapshotTask {
    fn name(&self) -> &'static str {
        "HistoricalTheoSnapshotTask"
    }

    fn owner(&self) -> crate::shared::service::ServiceOwner {
        crate::shared::service::ServiceOwner::Api
    }

    async fn run(
        self: std::sync::Arc<Self>,
        mut shutdown: crate::shared::ShutdownRx,
    ) -> anyhow::Result<()> {
        let mut ticker_5m = aligned_interval(self.config.capture_every_5m_ms);
        let mut ticker_1h = aligned_interval(self.config.capture_every_1h_ms);
        let mut ticker_1d = aligned_interval(self.config.capture_every_1d_ms);

        loop {
            tokio::select! {
                _ = shutdown.recv() => {
                    info!("Historical theo snapshot task received shutdown signal");
                    break;
                }
                _ = ticker_5m.tick() => {
                    self.capture_interval(INTERVAL_5M_MS, "5m").await;
                }
                _ = ticker_1h.tick() => {
                    self.capture_interval(INTERVAL_1H_MS, "1h").await;
                }
                _ = ticker_1d.tick() => {
                    self.capture_interval(INTERVAL_1D_MS, "1d").await;
                }
            }
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_build_snapshot_symbols_includes_active_instruments_and_underlyings() {
        let symbols = build_snapshot_symbols(
            vec![
                "BTC-CALL-50000-20261231".to_string(),
                "BTC-PERP".to_string(),
                "BTC-PERP".to_string(),
            ],
            vec!["BTC".to_string(), "ETH".to_string(), "BTC".to_string()],
        );

        assert_eq!(
            symbols,
            vec![
                "BTC".to_string(),
                "BTC-CALL-50000-20261231".to_string(),
                "BTC-PERP".to_string(),
                "ETH".to_string(),
                "ETH-PERP".to_string(),
            ]
        );
    }

    #[test]
    fn test_align_timestamp_ms() {
        let ts = 1_700_000_123_456i64;
        let aligned_5m = align_timestamp_ms(ts, INTERVAL_5M_MS);
        assert_eq!(aligned_5m % INTERVAL_5M_MS, 0);
        assert!(aligned_5m <= ts);
        assert!((ts - aligned_5m) < INTERVAL_5M_MS);

        let aligned_1h = align_timestamp_ms(ts, INTERVAL_1H_MS);
        assert_eq!(aligned_1h % INTERVAL_1H_MS, 0);
        assert!(aligned_1h <= ts);
        assert!((ts - aligned_1h) < INTERVAL_1H_MS);
    }

    #[test]
    fn test_next_bucket_start_ms() {
        let ts = 1_700_000_123_456i64;
        let next_5m = next_bucket_start_ms(ts, INTERVAL_5M_MS);
        assert_eq!(next_5m % INTERVAL_5M_MS, 0);
        assert!(next_5m > ts);
        assert_eq!(
            next_5m - align_timestamp_ms(ts, INTERVAL_5M_MS),
            INTERVAL_5M_MS
        );
    }
}