hypercall/runtime/tasks/

historical_pnl.rs

use std::time::Duration;

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

use crate::pnl_attribution::{encode_attribution, Attribution, SymbolAttribution};
use crate::read_cache::portfolio::PortfolioCache;
use hypercall_db::AnalyticsWriter;
use hypercall_types::{
    WalletAddress, HISTORICAL_PNL_INTERVAL_1D_MS, HISTORICAL_PNL_INTERVAL_1H_MS,
    HISTORICAL_PNL_INTERVAL_5M_MS,
};
use rust_decimal_macros::dec;
use std::collections::HashMap;
use std::sync::Arc;

pub const INTERVAL_5M_MS: i64 = HISTORICAL_PNL_INTERVAL_5M_MS;
pub const INTERVAL_1H_MS: i64 = HISTORICAL_PNL_INTERVAL_1H_MS;
pub const INTERVAL_1D_MS: i64 = HISTORICAL_PNL_INTERVAL_1D_MS;

/// Configuration for the historical equity snapshot background task.
#[derive(Debug, Clone)]
pub struct HistoricalPnlTaskConfig {
    /// Retention cap per `(wallet_address, 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 HistoricalPnlTaskConfig {
    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 HistoricalPnlTaskConfig {
    pub fn from_config(config: &crate::backend_config::HistoricalPnlRuntimeConfig) -> Self {
        Self {
            max_periods: config.max_periods,
            capture_every_5m_ms: config.capture_every_5m_ms,
            capture_every_1h_ms: config.capture_every_1h_ms,
            capture_every_1d_ms: config.capture_every_1d_ms,
        }
    }
}

#[derive(Clone)]
/// Background task that periodically snapshots wallet equity into the database.
pub struct HistoricalPnlSnapshotTask {
    portfolio_cache: Arc<PortfolioCache>,
    db: Arc<dyn AnalyticsWriter>,
    config: HistoricalPnlTaskConfig,
}

impl HistoricalPnlSnapshotTask {
    pub fn new(
        portfolio_cache: Arc<PortfolioCache>,
        db: Arc<dyn AnalyticsWriter>,
        config: HistoricalPnlTaskConfig,
    ) -> Self {
        Self {
            portfolio_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 portfolios = self.portfolio_cache.get_service().all_portfolios().await;
        let wallets: Vec<WalletAddress> = portfolios.keys().copied().collect();

        if wallets.is_empty() {
            debug!("Historical pnl snapshot skipped for {}: no wallets", label);
            return;
        }

        let mut snapshots: Vec<(WalletAddress, Decimal, Option<Vec<u8>>)> =
            Vec::with_capacity(wallets.len());

        for wallet in &wallets {
            match self
                .portfolio_cache
                .compute_wallet_margin_snapshot(wallet)
                .await
            {
                Ok(snapshot) => {
                    // Build attribution from portfolio positions + settlement PnL.
                    //
                    // Unrealized is (re)computed from the wallet's position + theo mark at
                    // bucket_timestamp_ms rather than read from pos.unrealized_pnl, because
                    // the live portfolio cache falls back to intrinsic value when UPNL=0
                    // (see commit 5afaa73af) which makes the tail of the chart go flat and
                    // disagree with the backfill's theo-based historical values.
                    //
                    // The final Attribution is then rolled up to collapse closed positions
                    // into per-underlying realized buckets, bounding payload growth as
                    // settled options accumulate forever.
                    let attr_bytes = {
                        let mut by_symbol = HashMap::new();
                        let mut total_pnl = dec!(0);

                        // Collect open-position symbols to look marks up for in one query.
                        let open_symbols: Vec<String> = portfolios
                            .get(wallet)
                            .map(|b| {
                                b.positions
                                    .iter()
                                    .filter(|(_, p)| p.amount != dec!(0))
                                    .map(|(sym, _)| sym.clone())
                                    .collect()
                            })
                            .unwrap_or_default();

                        let marks = if open_symbols.is_empty() {
                            HashMap::new()
                        } else {
                            match self
                                .db
                                .get_theo_marks_at_timestamp(&open_symbols, bucket_timestamp_ms)
                                .await
                            {
                                Ok(m) => m,
                                Err(e) => {
                                    error!(
                                        "attribution theo-mark lookup failed wallet={} ts={}: {}",
                                        wallet, bucket_timestamp_ms, e
                                    );
                                    HashMap::new()
                                }
                            }
                        };

                        // Open positions from portfolio, unrealized from theo marks.
                        if let Some(balance) = portfolios.get(wallet) {
                            for (symbol, pos) in &balance.positions {
                                if pos.amount == dec!(0) && pos.realized_pnl == dec!(0) {
                                    continue;
                                }
                                let unrealized = if pos.amount == dec!(0) {
                                    dec!(0)
                                } else {
                                    let mark = marks.get(symbol).copied().unwrap_or(dec!(0));
                                    (mark - pos.entry_price) * pos.amount
                                };
                                let sa = SymbolAttribution {
                                    position: pos.amount,
                                    entry_price: pos.entry_price,
                                    realized_pnl: pos.realized_pnl,
                                    unrealized_pnl: unrealized,
                                    total_pnl: pos.realized_pnl + unrealized,
                                };
                                total_pnl += sa.total_pnl;
                                by_symbol.insert(symbol.clone(), sa);
                            }
                        }

                        let settlement_result: anyhow::Result<Vec<(String, Decimal)>> = self
                            .db
                            .get_settled_pnl_by_symbol(wallet, bucket_timestamp_ms)
                            .await;
                        if let Ok(settlements) = settlement_result {
                            for (reference_symbol, pnl) in settlements {
                                if by_symbol.contains_key(&reference_symbol) {
                                    continue;
                                }
                                if pnl == dec!(0) {
                                    continue;
                                }
                                let sa = SymbolAttribution {
                                    position: dec!(0),
                                    entry_price: dec!(0),
                                    realized_pnl: pnl,
                                    unrealized_pnl: dec!(0),
                                    total_pnl: pnl,
                                };
                                total_pnl += sa.total_pnl;
                                by_symbol.insert(reference_symbol, sa);
                            }
                        }

                        // Emit individual option symbols (no per-underlying
                        // rollup): the frontend wants option names like
                        // "BTC-20260414-55000-C" and does its own top-N
                        // consolidation if a wallet has too many settled
                        // rows to render.
                        let attr = Attribution {
                            by_symbol,
                            total_pnl,
                        };

                        if attr.by_symbol.is_empty() {
                            None
                        } else {
                            Some(encode_attribution(&attr))
                        }
                    };

                    snapshots.push((*wallet, snapshot.margin_summary.equity, attr_bytes));
                }
                Err(e) => {
                    error!(
                        "Historical pnl snapshot failed wallet={} interval={} bucket_ts={}: {}",
                        wallet, label, bucket_timestamp_ms, e
                    );
                    counter!(
                        "ht_historical_pnl_snapshot_total",
                        "interval" => label,
                        "status" => "error"
                    )
                    .increment(1);
                    continue;
                }
            }
        }

        if snapshots.is_empty() {
            debug!(
                "Historical pnl snapshot skipped for {}: no wallets with computed equity",
                label
            );
            return;
        }

        match self
            .db
            .upsert_historical_pnl_batch(
                interval_ms,
                bucket_timestamp_ms,
                &snapshots,
                self.config.max_periods,
            )
            .await
        {
            Ok(affected) => {
                info!(
                    "Historical pnl snapshot captured interval={} wallets={} affected_rows={} bucket_ts={}",
                    label,
                    snapshots.len(),
                    affected,
                    bucket_timestamp_ms
                );
                counter!("ht_historical_pnl_snapshot_total", "interval" => label, "status" => "success")
                    .increment(1);
            }
            Err(e) => {
                error!(
                    "Failed to persist historical pnl snapshot interval={} bucket_ts={}: {}",
                    label, bucket_timestamp_ms, e
                );
                counter!("ht_historical_pnl_snapshot_total", "interval" => label, "status" => "error")
                    .increment(1);
            }
        }
    }
}

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
}

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

    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 pnl 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_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 exact_5m = 1_700_000_100_000i64;
        assert_eq!(
            next_bucket_start_ms(exact_5m, INTERVAL_5M_MS),
            exact_5m + INTERVAL_5M_MS
        );

        let between = exact_5m + 1234;
        assert_eq!(
            next_bucket_start_ms(between, INTERVAL_5M_MS),
            exact_5m + INTERVAL_5M_MS
        );
    }
}