global: MASSIVE snapshot

crates/brk_computer/src/distribution/state/block.rs

@@ -0,0 +1,34 @@
+use std::ops::{Add, AddAssign, SubAssign};
+
+use brk_types::{Dollars, SupplyState, Timestamp};
+use serde::Serialize;
+
+#[derive(Debug, Clone, Serialize)]
+pub struct BlockState {
+    #[serde(flatten)]
+    pub supply: SupplyState,
+    #[serde(skip)]
+    pub price: Option<Dollars>,
+    #[serde(skip)]
+    pub timestamp: Timestamp,
+}
+
+impl Add<BlockState> for BlockState {
+    type Output = Self;
+    fn add(mut self, rhs: BlockState) -> Self::Output {
+        self.supply += &rhs.supply;
+        self
+    }
+}
+
+impl AddAssign<&BlockState> for BlockState {
+    fn add_assign(&mut self, rhs: &Self) {
+        self.supply += &rhs.supply;
+    }
+}
+
+impl SubAssign<&BlockState> for BlockState {
+    fn sub_assign(&mut self, rhs: &Self) {
+        self.supply -= &rhs.supply;
+    }
+}

crates/brk_computer/src/distribution/state/cohort/address.rs

@@ -0,0 +1,184 @@
+use std::path::Path;
+
+use brk_error::Result;
+use brk_types::{Dollars, Height, LoadedAddressData, Sats, SupplyState};
+use vecdb::unlikely;
+
+use super::{
+    super::cost_basis::RealizedState,
+    base::CohortState,
+};
+
+#[derive(Clone)]
+pub struct AddressCohortState {
+    pub addr_count: u64,
+    pub inner: CohortState,
+}
+
+impl AddressCohortState {
+    pub fn new(path: &Path, name: &str, compute_dollars: bool) -> Self {
+        Self {
+            addr_count: 0,
+            inner: CohortState::new(path, name, compute_dollars),
+        }
+    }
+
+    /// Reset state for fresh start.
+    pub fn reset(&mut self) {
+        self.addr_count = 0;
+        self.inner.supply = SupplyState::default();
+        self.inner.sent = Sats::ZERO;
+        self.inner.satblocks_destroyed = Sats::ZERO;
+        self.inner.satdays_destroyed = Sats::ZERO;
+        if let Some(realized) = self.inner.realized.as_mut() {
+            *realized = RealizedState::NAN;
+        }
+    }
+
+    pub fn reset_price_to_amount_if_needed(&mut self) -> Result<()> {
+        self.inner.reset_price_to_amount_if_needed()
+    }
+
+    pub fn reset_single_iteration_values(&mut self) {
+        self.inner.reset_single_iteration_values();
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    pub fn send(
+        &mut self,
+        addressdata: &mut LoadedAddressData,
+        value: Sats,
+        current_price: Option<Dollars>,
+        prev_price: Option<Dollars>,
+        blocks_old: usize,
+        days_old: f64,
+        older_than_hour: bool,
+    ) -> Result<()> {
+        let compute_price = current_price.is_some();
+
+        let prev_realized_price = compute_price.then(|| addressdata.realized_price());
+        let prev_supply_state = SupplyState {
+            utxo_count: addressdata.utxo_count() as u64,
+            value: addressdata.balance(),
+        };
+
+        addressdata.send(value, prev_price)?;
+
+        let supply_state = SupplyState {
+            utxo_count: addressdata.utxo_count() as u64,
+            value: addressdata.balance(),
+        };
+
+        self.inner.send_(
+            &SupplyState {
+                utxo_count: 1,
+                value,
+            },
+            current_price,
+            prev_price,
+            blocks_old,
+            days_old,
+            older_than_hour,
+            compute_price.then(|| (addressdata.realized_price(), &supply_state)),
+            prev_realized_price.map(|prev_price| (prev_price, &prev_supply_state)),
+        );
+
+        Ok(())
+    }
+
+    pub fn receive(
+        &mut self,
+        address_data: &mut LoadedAddressData,
+        value: Sats,
+        price: Option<Dollars>,
+    ) {
+        self.receive_outputs(address_data, value, price, 1);
+    }
+
+    pub fn receive_outputs(
+        &mut self,
+        address_data: &mut LoadedAddressData,
+        value: Sats,
+        price: Option<Dollars>,
+        output_count: u32,
+    ) {
+        let compute_price = price.is_some();
+
+        let prev_realized_price = compute_price.then(|| address_data.realized_price());
+        let prev_supply_state = SupplyState {
+            utxo_count: address_data.utxo_count() as u64,
+            value: address_data.balance(),
+        };
+
+        address_data.receive_outputs(value, price, output_count);
+
+        let supply_state = SupplyState {
+            utxo_count: address_data.utxo_count() as u64,
+            value: address_data.balance(),
+        };
+
+        self.inner.receive_(
+            &SupplyState {
+                utxo_count: output_count as u64,
+                value,
+            },
+            price,
+            compute_price.then(|| (address_data.realized_price(), &supply_state)),
+            prev_realized_price.map(|prev_price| (prev_price, &prev_supply_state)),
+        );
+    }
+
+    pub fn add(&mut self, addressdata: &LoadedAddressData) {
+        self.addr_count += 1;
+        self.inner.increment_(
+            &addressdata.into(),
+            addressdata.realized_cap,
+            addressdata.realized_price(),
+        );
+    }
+
+    pub fn subtract(&mut self, addressdata: &LoadedAddressData) {
+        let addr_supply: SupplyState = addressdata.into();
+        let realized_price = addressdata.realized_price();
+
+        // Check for potential underflow before it happens
+        if unlikely(self.inner.supply.utxo_count < addr_supply.utxo_count) {
+            panic!(
+                "AddressCohortState::subtract underflow!\n\
+                Cohort state: addr_count={}, supply={}\n\
+                Address being subtracted: {}\n\
+                Address supply: {}\n\
+                Realized price: {}\n\
+                This means the address is not properly tracked in this cohort.",
+                self.addr_count, self.inner.supply, addressdata, addr_supply, realized_price
+            );
+        }
+        if unlikely(self.inner.supply.value < addr_supply.value) {
+            panic!(
+                "AddressCohortState::subtract value underflow!\n\
+                Cohort state: addr_count={}, supply={}\n\
+                Address being subtracted: {}\n\
+                Address supply: {}\n\
+                Realized price: {}\n\
+                This means the address is not properly tracked in this cohort.",
+                self.addr_count, self.inner.supply, addressdata, addr_supply, realized_price
+            );
+        }
+
+        self.addr_count = self.addr_count.checked_sub(1).unwrap_or_else(|| {
+            panic!(
+                "AddressCohortState::subtract addr_count underflow! addr_count=0\n\
+                Address being subtracted: {}\n\
+                Realized price: {}",
+                addressdata, realized_price
+            )
+        });
+
+        self.inner
+            .decrement_(&addr_supply, addressdata.realized_cap, realized_price);
+    }
+
+    pub fn write(&mut self, height: Height, cleanup: bool) -> Result<()> {
+        self.inner.write(height, cleanup)
+    }
+}

crates/brk_computer/src/distribution/state/cohort/base.rs

@@ -0,0 +1,398 @@
+use std::path::Path;
+
+use brk_error::Result;
+use brk_types::{Dollars, Height, Sats, SupplyState};
+
+use crate::internal::PERCENTILES_LEN;
+
+use super::super::cost_basis::{
+    CachedUnrealizedState, PriceToAmount, RealizedState, UnrealizedState,
+};
+
+/// State tracked for each cohort during computation.
+#[derive(Clone)]
+pub struct CohortState {
+    /// Current supply in this cohort
+    pub supply: SupplyState,
+
+    /// Realized cap and profit/loss (requires price data)
+    pub realized: Option<RealizedState>,
+
+    /// Amount sent in current block
+    pub sent: Sats,
+
+    /// Satoshi-blocks destroyed (supply * blocks_old when spent)
+    pub satblocks_destroyed: Sats,
+
+    /// Satoshi-days destroyed (supply * days_old when spent)
+    pub satdays_destroyed: Sats,
+
+    /// Price distribution for percentile calculations (requires price data)
+    price_to_amount: Option<PriceToAmount>,
+
+    /// Cached unrealized state for O(k) incremental updates.
+    cached_unrealized: Option<CachedUnrealizedState>,
+}
+
+impl CohortState {
+    /// Create new cohort state.
+    pub fn new(path: &Path, name: &str, compute_dollars: bool) -> Self {
+        Self {
+            supply: SupplyState::default(),
+            realized: compute_dollars.then_some(RealizedState::NAN),
+            sent: Sats::ZERO,
+            satblocks_destroyed: Sats::ZERO,
+            satdays_destroyed: Sats::ZERO,
+            price_to_amount: compute_dollars.then_some(PriceToAmount::create(path, name)),
+            cached_unrealized: None,
+        }
+    }
+
+    /// Import state from checkpoint.
+    pub fn import_at_or_before(&mut self, height: Height) -> Result<Height> {
+        // Invalidate cache when importing new data
+        self.cached_unrealized = None;
+
+        match self.price_to_amount.as_mut() {
+            Some(p) => p.import_at_or_before(height),
+            None => Ok(height),
+        }
+    }
+
+    /// Reset price_to_amount if needed (for starting fresh).
+    pub fn reset_price_to_amount_if_needed(&mut self) -> Result<()> {
+        if let Some(p) = self.price_to_amount.as_mut() {
+            p.clean()?;
+            p.init();
+        }
+        // Invalidate cache when data is reset
+        self.cached_unrealized = None;
+        Ok(())
+    }
+
+    /// Apply pending price_to_amount updates. Must be called before reads.
+    pub fn apply_pending(&mut self) {
+        if let Some(p) = self.price_to_amount.as_mut() {
+            p.apply_pending();
+        }
+    }
+
+    /// Get first (lowest) price entry in distribution.
+    pub fn price_to_amount_first_key_value(&self) -> Option<(Dollars, &Sats)> {
+        self.price_to_amount.as_ref()?.first_key_value()
+    }
+
+    /// Get last (highest) price entry in distribution.
+    pub fn price_to_amount_last_key_value(&self) -> Option<(Dollars, &Sats)> {
+        self.price_to_amount.as_ref()?.last_key_value()
+    }
+
+    /// Reset per-block values before processing next block.
+    pub fn reset_single_iteration_values(&mut self) {
+        self.sent = Sats::ZERO;
+        self.satdays_destroyed = Sats::ZERO;
+        self.satblocks_destroyed = Sats::ZERO;
+        if let Some(realized) = self.realized.as_mut() {
+            realized.reset_single_iteration_values();
+        }
+    }
+
+    /// Add supply to this cohort (e.g., when UTXO ages into cohort).
+    pub fn increment(&mut self, supply: &SupplyState, price: Option<Dollars>) {
+        self.supply += supply;
+
+        if supply.value > Sats::ZERO
+            && let Some(realized) = self.realized.as_mut()
+        {
+            let price = price.unwrap();
+            realized.increment(supply, price);
+            self.price_to_amount
+                .as_mut()
+                .unwrap()
+                .increment(price, supply);
+
+            // Update cache for added supply
+            if let Some(cache) = self.cached_unrealized.as_mut() {
+                cache.on_receive(price, supply.value);
+            }
+        }
+    }
+
+    /// Add supply with pre-computed realized cap (for address cohorts).
+    pub fn increment_(
+        &mut self,
+        supply: &SupplyState,
+        realized_cap: Dollars,
+        realized_price: Dollars,
+    ) {
+        self.supply += supply;
+
+        if supply.value > Sats::ZERO
+            && let Some(realized) = self.realized.as_mut()
+        {
+            realized.increment_(realized_cap);
+            self.price_to_amount
+                .as_mut()
+                .unwrap()
+                .increment(realized_price, supply);
+
+            // Update cache for added supply
+            if let Some(cache) = self.cached_unrealized.as_mut() {
+                cache.on_receive(realized_price, supply.value);
+            }
+        }
+    }
+
+    /// Remove supply from this cohort (e.g., when UTXO ages out of cohort).
+    pub fn decrement(&mut self, supply: &SupplyState, price: Option<Dollars>) {
+        self.supply -= supply;
+
+        if supply.value > Sats::ZERO
+            && let Some(realized) = self.realized.as_mut()
+        {
+            let price = price.unwrap();
+            realized.decrement(supply, price);
+            self.price_to_amount
+                .as_mut()
+                .unwrap()
+                .decrement(price, supply);
+
+            // Update cache for removed supply
+            if let Some(cache) = self.cached_unrealized.as_mut() {
+                cache.on_send(price, supply.value);
+            }
+        }
+    }
+
+    /// Remove supply with pre-computed realized cap (for address cohorts).
+    pub fn decrement_(
+        &mut self,
+        supply: &SupplyState,
+        realized_cap: Dollars,
+        realized_price: Dollars,
+    ) {
+        self.supply -= supply;
+
+        if supply.value > Sats::ZERO
+            && let Some(realized) = self.realized.as_mut()
+        {
+            realized.decrement_(realized_cap);
+            self.price_to_amount
+                .as_mut()
+                .unwrap()
+                .decrement(realized_price, supply);
+
+            // Update cache for removed supply
+            if let Some(cache) = self.cached_unrealized.as_mut() {
+                cache.on_send(realized_price, supply.value);
+            }
+        }
+    }
+
+    /// Process received output (new UTXO in cohort).
+    pub fn receive(&mut self, supply: &SupplyState, price: Option<Dollars>) {
+        self.receive_(supply, price, price.map(|price| (price, supply)), None);
+    }
+
+    /// Process received output with custom price_to_amount updates (for address cohorts).
+    pub fn receive_(
+        &mut self,
+        supply: &SupplyState,
+        price: Option<Dollars>,
+        price_to_amount_increment: Option<(Dollars, &SupplyState)>,
+        price_to_amount_decrement: Option<(Dollars, &SupplyState)>,
+    ) {
+        self.supply += supply;
+
+        if supply.value > Sats::ZERO
+            && let Some(realized) = self.realized.as_mut()
+        {
+            let price = price.unwrap();
+            realized.receive(supply, price);
+
+            if let Some((price, supply)) = price_to_amount_increment
+                && supply.value.is_not_zero()
+            {
+                self.price_to_amount
+                    .as_mut()
+                    .unwrap()
+                    .increment(price, supply);
+
+                // Update cache for added supply
+                if let Some(cache) = self.cached_unrealized.as_mut() {
+                    cache.on_receive(price, supply.value);
+                }
+            }
+
+            if let Some((price, supply)) = price_to_amount_decrement
+                && supply.value.is_not_zero()
+            {
+                self.price_to_amount
+                    .as_mut()
+                    .unwrap()
+                    .decrement(price, supply);
+
+                // Update cache for removed supply
+                if let Some(cache) = self.cached_unrealized.as_mut() {
+                    cache.on_send(price, supply.value);
+                }
+            }
+        }
+    }
+
+    /// Process spent input (UTXO leaving cohort).
+    pub fn send(
+        &mut self,
+        supply: &SupplyState,
+        current_price: Option<Dollars>,
+        prev_price: Option<Dollars>,
+        blocks_old: usize,
+        days_old: f64,
+        older_than_hour: bool,
+    ) {
+        self.send_(
+            supply,
+            current_price,
+            prev_price,
+            blocks_old,
+            days_old,
+            older_than_hour,
+            None,
+            prev_price.map(|prev_price| (prev_price, supply)),
+        );
+    }
+
+    /// Process spent input with custom price_to_amount updates (for address cohorts).
+    #[allow(clippy::too_many_arguments)]
+    pub fn send_(
+        &mut self,
+        supply: &SupplyState,
+        current_price: Option<Dollars>,
+        prev_price: Option<Dollars>,
+        blocks_old: usize,
+        days_old: f64,
+        older_than_hour: bool,
+        price_to_amount_increment: Option<(Dollars, &SupplyState)>,
+        price_to_amount_decrement: Option<(Dollars, &SupplyState)>,
+    ) {
+        if supply.utxo_count == 0 {
+            return;
+        }
+
+        self.supply -= supply;
+
+        if supply.value > Sats::ZERO {
+            self.sent += supply.value;
+            self.satblocks_destroyed += supply.value * blocks_old;
+            self.satdays_destroyed +=
+                Sats::from((u64::from(supply.value) as f64 * days_old).floor() as u64);
+
+            if let Some(realized) = self.realized.as_mut() {
+                let current_price = current_price.unwrap();
+                let prev_price = prev_price.unwrap();
+                realized.send(supply, current_price, prev_price, older_than_hour);
+
+                if let Some((price, supply)) = price_to_amount_increment
+                    && supply.value.is_not_zero()
+                {
+                    self.price_to_amount
+                        .as_mut()
+                        .unwrap()
+                        .increment(price, supply);
+
+                    // Update cache for added supply
+                    if let Some(cache) = self.cached_unrealized.as_mut() {
+                        cache.on_receive(price, supply.value);
+                    }
+                }
+
+                if let Some((price, supply)) = price_to_amount_decrement
+                    && supply.value.is_not_zero()
+                {
+                    self.price_to_amount
+                        .as_mut()
+                        .unwrap()
+                        .decrement(price, supply);
+
+                    // Update cache for removed supply
+                    if let Some(cache) = self.cached_unrealized.as_mut() {
+                        cache.on_send(price, supply.value);
+                    }
+                }
+            }
+        }
+    }
+
+    /// Compute prices at percentile thresholds.
+    pub fn compute_percentile_prices(&self) -> [Dollars; PERCENTILES_LEN] {
+        match self.price_to_amount.as_ref() {
+            Some(p) if !p.is_empty() => p.compute_percentiles(),
+            _ => [Dollars::NAN; PERCENTILES_LEN],
+        }
+    }
+
+    /// Compute unrealized profit/loss at current price.
+    /// Uses O(k) incremental updates for height_price where k = flip range size.
+    pub fn compute_unrealized_states(
+        &mut self,
+        height_price: Dollars,
+        date_price: Option<Dollars>,
+    ) -> (UnrealizedState, Option<UnrealizedState>) {
+        let price_to_amount = match self.price_to_amount.as_ref() {
+            Some(p) if !p.is_empty() => p,
+            _ => {
+                return (
+                    UnrealizedState::NAN,
+                    date_price.map(|_| UnrealizedState::NAN),
+                );
+            }
+        };
+
+        // Date unrealized: compute from scratch (only at date boundaries, ~144x less frequent)
+        let date_state = date_price.map(|date_price| {
+            CachedUnrealizedState::compute_full_standalone(date_price, price_to_amount)
+        });
+
+        // Height unrealized: use incremental cache (O(k) where k = flip range)
+        let height_state = if let Some(cache) = self.cached_unrealized.as_mut() {
+            cache.get_at_price(height_price, price_to_amount).clone()
+        } else {
+            let cache = CachedUnrealizedState::compute_fresh(height_price, price_to_amount);
+            let state = cache.state.clone();
+            self.cached_unrealized = Some(cache);
+            state
+        };
+
+        (height_state, date_state)
+    }
+
+    /// Flush state to disk at checkpoint.
+    pub fn write(&mut self, height: Height, cleanup: bool) -> Result<()> {
+        if let Some(p) = self.price_to_amount.as_mut() {
+            p.write(height, cleanup)?;
+        }
+        Ok(())
+    }
+
+    /// Get first (lowest) price in distribution.
+    pub fn min_price(&self) -> Option<Dollars> {
+        self.price_to_amount
+            .as_ref()?
+            .first_key_value()
+            .map(|(k, _)| k)
+    }
+
+    /// Get last (highest) price in distribution.
+    pub fn max_price(&self) -> Option<Dollars> {
+        self.price_to_amount
+            .as_ref()?
+            .last_key_value()
+            .map(|(k, _)| k)
+    }
+
+    /// Get iterator over price_to_amount for merged percentile computation.
+    /// Returns None if price data is not tracked for this cohort.
+    pub fn price_to_amount_iter(&self) -> Option<impl Iterator<Item = (Dollars, &Sats)>> {
+        self.price_to_amount.as_ref().map(|p| p.iter())
+    }
+}

crates/brk_computer/src/distribution/state/cohort/mod.rs

@@ -0,0 +1,7 @@
+mod address;
+mod base;
+mod utxo;
+
+pub use address::*;
+pub use base::*;
+pub use utxo::*;

crates/brk_computer/src/distribution/state/cohort/utxo.rs

@@ -0,0 +1,34 @@
+use std::path::Path;
+
+use brk_error::Result;
+use brk_types::{Sats, SupplyState};
+use derive_deref::{Deref, DerefMut};
+
+use super::{
+    super::cost_basis::RealizedState,
+    base::CohortState,
+};
+
+#[derive(Clone, Deref, DerefMut)]
+pub struct UTXOCohortState(CohortState);
+
+impl UTXOCohortState {
+    pub fn new(path: &Path, name: &str, compute_dollars: bool) -> Self {
+        Self(CohortState::new(path, name, compute_dollars))
+    }
+
+    pub fn reset_price_to_amount_if_needed(&mut self) -> Result<()> {
+        self.0.reset_price_to_amount_if_needed()
+    }
+
+    /// Reset state for fresh start.
+    pub fn reset(&mut self) {
+        self.0.supply = SupplyState::default();
+        self.0.sent = Sats::ZERO;
+        self.0.satblocks_destroyed = Sats::ZERO;
+        self.0.satdays_destroyed = Sats::ZERO;
+        if let Some(realized) = self.0.realized.as_mut() {
+            *realized = RealizedState::NAN;
+        }
+    }
+}

crates/brk_computer/src/distribution/state/cost_basis/mod.rs

@@ -0,0 +1,7 @@
+mod price_to_amount;
+mod realized;
+mod unrealized;
+
+pub use price_to_amount::*;
+pub use realized::*;
+pub use unrealized::*;

crates/brk_computer/src/distribution/state/cost_basis/price_to_amount.rs

@@ -0,0 +1,273 @@
+use std::{
+    collections::BTreeMap,
+    fs,
+    ops::Bound,
+    path::{Path, PathBuf},
+};
+
+use brk_error::{Error, Result};
+use brk_types::{CentsCompact, Dollars, Height, Sats, SupplyState};
+use derive_deref::{Deref, DerefMut};
+use pco::standalone::{simple_decompress, simpler_compress};
+use rustc_hash::FxHashMap;
+use serde::{Deserialize, Serialize};
+use vecdb::Bytes;
+
+use crate::{
+    internal::{PERCENTILES, PERCENTILES_LEN},
+    utils::OptionExt,
+};
+
+#[derive(Clone, Debug)]
+pub struct PriceToAmount {
+    pathbuf: PathBuf,
+    state: Option<State>,
+    /// Pending deltas: (total_increment, total_decrement) per price.
+    /// Flushed to BTreeMap before reads and at end of block.
+    pending: FxHashMap<CentsCompact, (Sats, Sats)>,
+}
+
+const STATE_AT_: &str = "state_at_";
+const STATE_TO_KEEP: usize = 10;
+
+impl PriceToAmount {
+    pub fn create(path: &Path, name: &str) -> Self {
+        Self {
+            pathbuf: path.join(format!("{name}_price_to_amount")),
+            state: None,
+            pending: FxHashMap::default(),
+        }
+    }
+
+    pub fn import_at_or_before(&mut self, height: Height) -> Result<Height> {
+        let files = self.read_dir(None)?;
+        let (&height, path) = files.range(..=height).next_back().ok_or(Error::NotFound(
+            "No price state found at or before height".into(),
+        ))?;
+        self.state = Some(State::deserialize(&fs::read(path)?)?);
+        self.pending.clear();
+        Ok(height)
+    }
+
+    fn assert_pending_empty(&self) {
+        assert!(
+            self.pending.is_empty(),
+            "PriceToAmount: pending not empty, call apply_pending first"
+        );
+    }
+
+    pub fn iter(&self) -> impl Iterator<Item = (Dollars, &Sats)> {
+        self.assert_pending_empty();
+        self.state.u().iter().map(|(k, v)| (k.to_dollars(), v))
+    }
+
+    /// Iterate over entries in a price range with explicit bounds.
+    pub fn range(
+        &self,
+        bounds: (Bound<Dollars>, Bound<Dollars>),
+    ) -> impl Iterator<Item = (Dollars, &Sats)> {
+        self.assert_pending_empty();
+
+        let start = match bounds.0 {
+            Bound::Included(d) => Bound::Included(CentsCompact::from(d)),
+            Bound::Excluded(d) => Bound::Excluded(CentsCompact::from(d)),
+            Bound::Unbounded => Bound::Unbounded,
+        };
+
+        let end = match bounds.1 {
+            Bound::Included(d) => Bound::Included(CentsCompact::from(d)),
+            Bound::Excluded(d) => Bound::Excluded(CentsCompact::from(d)),
+            Bound::Unbounded => Bound::Unbounded,
+        };
+
+        self.state
+            .u()
+            .range((start, end))
+            .map(|(k, v)| (k.to_dollars(), v))
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.pending.is_empty() && self.state.u().is_empty()
+    }
+
+    pub fn first_key_value(&self) -> Option<(Dollars, &Sats)> {
+        self.assert_pending_empty();
+        self.state
+            .u()
+            .first_key_value()
+            .map(|(k, v)| (k.to_dollars(), v))
+    }
+
+    pub fn last_key_value(&self) -> Option<(Dollars, &Sats)> {
+        self.assert_pending_empty();
+        self.state
+            .u()
+            .last_key_value()
+            .map(|(k, v)| (k.to_dollars(), v))
+    }
+
+    /// Accumulate increment in pending batch. O(1).
+    pub fn increment(&mut self, price: Dollars, supply_state: &SupplyState) {
+        self.pending.entry(CentsCompact::from(price)).or_default().0 += supply_state.value;
+    }
+
+    /// Accumulate decrement in pending batch. O(1).
+    pub fn decrement(&mut self, price: Dollars, supply_state: &SupplyState) {
+        self.pending.entry(CentsCompact::from(price)).or_default().1 += supply_state.value;
+    }
+
+    /// Apply pending deltas to BTreeMap. O(k log n) where k = unique prices in pending.
+    /// Must be called before any read operations.
+    pub fn apply_pending(&mut self) {
+        for (cents, (inc, dec)) in self.pending.drain() {
+            let entry = self.state.um().entry(cents).or_default();
+            *entry += inc;
+            if *entry < dec {
+                panic!(
+                    "PriceToAmount::apply_pending underflow!\n\
+                    Path: {:?}\n\
+                    Price: {}\n\
+                    Current + increments: {}\n\
+                    Trying to decrement by: {}",
+                    self.pathbuf,
+                    cents.to_dollars(),
+                    entry,
+                    dec
+                );
+            }
+            *entry -= dec;
+            if *entry == Sats::ZERO {
+                self.state.um().remove(&cents);
+            }
+        }
+    }
+
+    pub fn init(&mut self) {
+        self.state.replace(State::default());
+        self.pending.clear();
+    }
+
+    /// Compute percentile prices by iterating the BTreeMap directly.
+    /// O(n) where n = number of unique prices.
+    pub fn compute_percentiles(&self) -> [Dollars; PERCENTILES_LEN] {
+        self.assert_pending_empty();
+
+        let state = match self.state.as_ref() {
+            Some(s) if !s.is_empty() => s,
+            _ => return [Dollars::NAN; PERCENTILES_LEN],
+        };
+
+        let total: u64 = state.values().map(|&s| u64::from(s)).sum();
+        if total == 0 {
+            return [Dollars::NAN; PERCENTILES_LEN];
+        }
+
+        let mut result = [Dollars::NAN; PERCENTILES_LEN];
+        let mut cumsum = 0u64;
+        let mut idx = 0;
+
+        for (&cents, &amount) in state.iter() {
+            cumsum += u64::from(amount);
+            while idx < PERCENTILES_LEN && cumsum >= total * u64::from(PERCENTILES[idx]) / 100 {
+                result[idx] = cents.to_dollars();
+                idx += 1;
+            }
+        }
+
+        result
+    }
+
+    pub fn clean(&mut self) -> Result<()> {
+        let _ = fs::remove_dir_all(&self.pathbuf);
+        fs::create_dir_all(&self.pathbuf)?;
+        Ok(())
+    }
+
+    fn read_dir(&self, keep_only_before: Option<Height>) -> Result<BTreeMap<Height, PathBuf>> {
+        Ok(fs::read_dir(&self.pathbuf)?
+            .filter_map(|entry| {
+                let path = entry.ok()?.path();
+                let name = path.file_name()?.to_str()?;
+                let height_str = name.strip_prefix(STATE_AT_).unwrap_or(name);
+                if let Ok(h) = height_str.parse::<u32>().map(Height::from) {
+                    if keep_only_before.is_none_or(|height| h < height) {
+                        Some((h, path))
+                    } else {
+                        let _ = fs::remove_file(path);
+                        None
+                    }
+                } else {
+                    None
+                }
+            })
+            .collect::<BTreeMap<Height, PathBuf>>())
+    }
+
+    /// Flush state to disk, optionally cleaning up old state files.
+    pub fn write(&mut self, height: Height, cleanup: bool) -> Result<()> {
+        self.apply_pending();
+
+        if cleanup {
+            let files = self.read_dir(Some(height))?;
+
+            for (_, path) in files
+                .iter()
+                .take(files.len().saturating_sub(STATE_TO_KEEP - 1))
+            {
+                fs::remove_file(path)?;
+            }
+        }
+
+        fs::write(self.path_state(height), self.state.u().serialize()?)?;
+
+        Ok(())
+    }
+
+    fn path_state(&self, height: Height) -> PathBuf {
+        Self::path_state_(&self.pathbuf, height)
+    }
+    fn path_state_(path: &Path, height: Height) -> PathBuf {
+        path.join(u32::from(height).to_string())
+    }
+}
+
+#[derive(Clone, Default, Debug, Deref, DerefMut, Serialize, Deserialize)]
+struct State(BTreeMap<CentsCompact, Sats>);
+
+const COMPRESSION_LEVEL: usize = 4;
+
+impl State {
+    fn serialize(&self) -> vecdb::Result<Vec<u8>> {
+        let keys: Vec<i32> = self.keys().map(|k| i32::from(*k)).collect();
+        let values: Vec<u64> = self.values().map(|v| u64::from(*v)).collect();
+
+        let compressed_keys = simpler_compress(&keys, COMPRESSION_LEVEL)?;
+        let compressed_values = simpler_compress(&values, COMPRESSION_LEVEL)?;
+
+        let mut buffer = Vec::new();
+        buffer.extend(keys.len().to_bytes());
+        buffer.extend(compressed_keys.len().to_bytes());
+        buffer.extend(compressed_keys);
+        buffer.extend(compressed_values);
+
+        Ok(buffer)
+    }
+
+    fn deserialize(data: &[u8]) -> vecdb::Result<Self> {
+        let entry_count = usize::from_bytes(&data[0..8])?;
+        let keys_len = usize::from_bytes(&data[8..16])?;
+
+        let keys: Vec<i32> = simple_decompress(&data[16..16 + keys_len])?;
+        let values: Vec<u64> = simple_decompress(&data[16 + keys_len..])?;
+
+        let map: BTreeMap<CentsCompact, Sats> = keys
+            .into_iter()
+            .zip(values)
+            .map(|(k, v)| (CentsCompact::from(k), Sats::from(v)))
+            .collect();
+
+        assert_eq!(map.len(), entry_count);
+
+        Ok(Self(map))
+    }
+}

crates/brk_computer/src/distribution/state/cost_basis/realized.rs

@@ -0,0 +1,102 @@
+use std::cmp::Ordering;
+
+use brk_types::{CheckedSub, Dollars, SupplyState};
+
+#[derive(Debug, Default, Clone)]
+pub struct RealizedState {
+    pub cap: Dollars,
+    pub profit: Dollars,
+    pub loss: Dollars,
+    pub value_created: Dollars,
+    pub adj_value_created: Dollars,
+    pub value_destroyed: Dollars,
+    pub adj_value_destroyed: Dollars,
+}
+
+impl RealizedState {
+    pub const NAN: Self = Self {
+        cap: Dollars::NAN,
+        profit: Dollars::NAN,
+        loss: Dollars::NAN,
+        value_created: Dollars::NAN,
+        adj_value_created: Dollars::NAN,
+        value_destroyed: Dollars::NAN,
+        adj_value_destroyed: Dollars::NAN,
+    };
+
+    pub fn reset_single_iteration_values(&mut self) {
+        if self.cap != Dollars::NAN {
+            self.profit = Dollars::ZERO;
+            self.loss = Dollars::ZERO;
+            self.value_created = Dollars::ZERO;
+            self.adj_value_created = Dollars::ZERO;
+            self.value_destroyed = Dollars::ZERO;
+            self.adj_value_destroyed = Dollars::ZERO;
+        }
+    }
+
+    pub fn increment(&mut self, supply_state: &SupplyState, price: Dollars) {
+        if supply_state.value.is_zero() {
+            return;
+        }
+
+        self.increment_(price * supply_state.value)
+    }
+
+    pub fn increment_(&mut self, realized_cap: Dollars) {
+        if self.cap == Dollars::NAN {
+            self.cap = Dollars::ZERO;
+            self.profit = Dollars::ZERO;
+            self.loss = Dollars::ZERO;
+            self.value_created = Dollars::ZERO;
+            self.adj_value_created = Dollars::ZERO;
+            self.value_destroyed = Dollars::ZERO;
+            self.adj_value_destroyed = Dollars::ZERO;
+        }
+
+        self.cap += realized_cap;
+    }
+
+    pub fn decrement(&mut self, supply_state: &SupplyState, price: Dollars) {
+        self.decrement_(price * supply_state.value);
+    }
+
+    pub fn decrement_(&mut self, realized_cap: Dollars) {
+        self.cap = self.cap.checked_sub(realized_cap).unwrap();
+    }
+
+    pub fn receive(&mut self, supply_state: &SupplyState, current_price: Dollars) {
+        self.increment(supply_state, current_price);
+    }
+
+    pub fn send(
+        &mut self,
+        supply_state: &SupplyState,
+        current_price: Dollars,
+        prev_price: Dollars,
+        older_than_hour: bool,
+    ) {
+        let current_value = current_price * supply_state.value;
+        let prev_value = prev_price * supply_state.value;
+
+        self.value_created += current_value;
+        self.value_destroyed += prev_value;
+
+        if older_than_hour {
+            self.adj_value_created += current_value;
+            self.adj_value_destroyed += prev_value;
+        }
+
+        match current_price.cmp(&prev_price) {
+            Ordering::Greater => {
+                self.profit += current_value.checked_sub(prev_value).unwrap();
+            }
+            Ordering::Less => {
+                self.loss += prev_value.checked_sub(current_value).unwrap();
+            }
+            Ordering::Equal => {}
+        }
+
+        self.decrement(supply_state, prev_price);
+    }
+}

crates/brk_computer/src/distribution/state/cost_basis/unrealized.rs

@@ -0,0 +1,228 @@
+use std::ops::Bound;
+
+use brk_types::{Dollars, Sats};
+use vecdb::CheckedSub;
+
+use super::price_to_amount::PriceToAmount;
+
+#[derive(Debug, Default, Clone)]
+pub struct UnrealizedState {
+    pub supply_in_profit: Sats,
+    pub supply_in_loss: Sats,
+    pub unrealized_profit: Dollars,
+    pub unrealized_loss: Dollars,
+}
+
+impl UnrealizedState {
+    pub const NAN: Self = Self {
+        supply_in_profit: Sats::ZERO,
+        supply_in_loss: Sats::ZERO,
+        unrealized_profit: Dollars::NAN,
+        unrealized_loss: Dollars::NAN,
+    };
+
+    pub const ZERO: Self = Self {
+        supply_in_profit: Sats::ZERO,
+        supply_in_loss: Sats::ZERO,
+        unrealized_profit: Dollars::ZERO,
+        unrealized_loss: Dollars::ZERO,
+    };
+}
+
+/// Cached unrealized state for O(k) incremental updates.
+/// k = number of entries in price flip range (typically tiny).
+#[derive(Debug, Clone)]
+pub struct CachedUnrealizedState {
+    pub state: UnrealizedState,
+    at_price: Dollars,
+}
+
+impl CachedUnrealizedState {
+    /// Create new cache by computing from scratch. O(n).
+    pub fn compute_fresh(price: Dollars, price_to_amount: &PriceToAmount) -> Self {
+        let state = Self::compute_full_standalone(price, price_to_amount);
+        Self {
+            state,
+            at_price: price,
+        }
+    }
+
+    /// Get unrealized state at new_price. O(k) where k = flip range size.
+    pub fn get_at_price(
+        &mut self,
+        new_price: Dollars,
+        price_to_amount: &PriceToAmount,
+    ) -> &UnrealizedState {
+        if new_price != self.at_price {
+            self.update_for_price_change(new_price, price_to_amount);
+        }
+        &self.state
+    }
+
+    /// Update cached state when a receive happens.
+    /// Determines profit/loss classification relative to cached price.
+    pub fn on_receive(&mut self, purchase_price: Dollars, sats: Sats) {
+        if purchase_price <= self.at_price {
+            self.state.supply_in_profit += sats;
+            if purchase_price < self.at_price {
+                let diff = self.at_price.checked_sub(purchase_price).unwrap();
+                self.state.unrealized_profit += diff * sats;
+            }
+        } else {
+            self.state.supply_in_loss += sats;
+            let diff = purchase_price.checked_sub(self.at_price).unwrap();
+            self.state.unrealized_loss += diff * sats;
+        }
+    }
+
+    /// Update cached state when a send happens from historical price.
+    pub fn on_send(&mut self, historical_price: Dollars, sats: Sats) {
+        if historical_price <= self.at_price {
+            // Was in profit
+            self.state.supply_in_profit -= sats;
+            if historical_price < self.at_price {
+                let diff = self.at_price.checked_sub(historical_price).unwrap();
+                let profit_removed = diff * sats;
+                self.state.unrealized_profit = self
+                    .state
+                    .unrealized_profit
+                    .checked_sub(profit_removed)
+                    .unwrap_or(Dollars::ZERO);
+            }
+        } else {
+            // Was in loss
+            self.state.supply_in_loss -= sats;
+            let diff = historical_price.checked_sub(self.at_price).unwrap();
+            let loss_removed = diff * sats;
+            self.state.unrealized_loss = self
+                .state
+                .unrealized_loss
+                .checked_sub(loss_removed)
+                .unwrap_or(Dollars::ZERO);
+        }
+    }
+
+    /// Incremental update for price change. O(k) where k = entries in flip range.
+    fn update_for_price_change(&mut self, new_price: Dollars, price_to_amount: &PriceToAmount) {
+        let old_price = self.at_price;
+        let delta_f64 = f64::from(new_price) - f64::from(old_price);
+
+        // Update profit/loss for entries that DON'T flip
+        // Profit changes by delta * supply_in_profit
+        // Loss changes by -delta * supply_in_loss
+        if delta_f64 > 0.0 {
+            // Price went up: profits increase, losses decrease
+            self.state.unrealized_profit += Dollars::from(delta_f64) * self.state.supply_in_profit;
+            let loss_decrease = Dollars::from(delta_f64) * self.state.supply_in_loss;
+            self.state.unrealized_loss = self
+                .state
+                .unrealized_loss
+                .checked_sub(loss_decrease)
+                .unwrap_or(Dollars::ZERO);
+        } else if delta_f64 < 0.0 {
+            // Price went down: profits decrease, losses increase
+            let profit_decrease = Dollars::from(-delta_f64) * self.state.supply_in_profit;
+            self.state.unrealized_profit = self
+                .state
+                .unrealized_profit
+                .checked_sub(profit_decrease)
+                .unwrap_or(Dollars::ZERO);
+            self.state.unrealized_loss += Dollars::from(-delta_f64) * self.state.supply_in_loss;
+        }
+
+        // Handle flipped entries (only iterate the small range between prices)
+        if new_price > old_price {
+            // Price went up: entries where old < price <= new flip from loss to profit
+            for (price, &sats) in
+                price_to_amount.range((Bound::Excluded(old_price), Bound::Included(new_price)))
+            {
+                // Move from loss to profit
+                self.state.supply_in_loss -= sats;
+                self.state.supply_in_profit += sats;
+
+                // Undo the loss adjustment applied above for this entry
+                // We decreased loss by delta * sats, but this entry should be removed entirely
+                // Original loss: (price - old_price) * sats
+                // After global adjustment: original - delta * sats (negative, wrong)
+                // Correct: 0 (removed from loss)
+                // Correction: add back delta * sats, then add original loss
+                let delta_adj = Dollars::from(delta_f64) * sats;
+                self.state.unrealized_loss += delta_adj;
+                if price > old_price {
+                    let original_loss = price.checked_sub(old_price).unwrap() * sats;
+                    self.state.unrealized_loss += original_loss;
+                }
+
+                // Undo the profit adjustment applied above for this entry
+                // We increased profit by delta * sats, but this entry was not in profit before
+                // Correct profit: (new_price - price) * sats
+                // Correction: subtract delta * sats, add correct profit
+                let profit_adj = Dollars::from(delta_f64) * sats;
+                self.state.unrealized_profit = self
+                    .state
+                    .unrealized_profit
+                    .checked_sub(profit_adj)
+                    .unwrap_or(Dollars::ZERO);
+                if new_price > price {
+                    let correct_profit = new_price.checked_sub(price).unwrap() * sats;
+                    self.state.unrealized_profit += correct_profit;
+                }
+            }
+        } else if new_price < old_price {
+            // Price went down: entries where new < price <= old flip from profit to loss
+            for (price, &sats) in
+                price_to_amount.range((Bound::Excluded(new_price), Bound::Included(old_price)))
+            {
+                // Move from profit to loss
+                self.state.supply_in_profit -= sats;
+                self.state.supply_in_loss += sats;
+
+                // Undo the profit adjustment applied above for this entry
+                let delta_adj = Dollars::from(-delta_f64) * sats;
+                self.state.unrealized_profit += delta_adj;
+                if old_price > price {
+                    let original_profit = old_price.checked_sub(price).unwrap() * sats;
+                    self.state.unrealized_profit += original_profit;
+                }
+
+                // Undo the loss adjustment applied above for this entry
+                let loss_adj = Dollars::from(-delta_f64) * sats;
+                self.state.unrealized_loss = self
+                    .state
+                    .unrealized_loss
+                    .checked_sub(loss_adj)
+                    .unwrap_or(Dollars::ZERO);
+                if price > new_price {
+                    let correct_loss = price.checked_sub(new_price).unwrap() * sats;
+                    self.state.unrealized_loss += correct_loss;
+                }
+            }
+        }
+
+        self.at_price = new_price;
+    }
+
+    /// Full computation from scratch (no cache). O(n).
+    pub fn compute_full_standalone(
+        current_price: Dollars,
+        price_to_amount: &PriceToAmount,
+    ) -> UnrealizedState {
+        let mut state = UnrealizedState::ZERO;
+
+        for (price, &sats) in price_to_amount.iter() {
+            if price <= current_price {
+                state.supply_in_profit += sats;
+                if price < current_price {
+                    let diff = current_price.checked_sub(price).unwrap();
+                    state.unrealized_profit += diff * sats;
+                }
+            } else {
+                state.supply_in_loss += sats;
+                let diff = price.checked_sub(current_price).unwrap();
+                state.unrealized_loss += diff * sats;
+            }
+        }
+
+        state
+    }
+}

crates/brk_computer/src/distribution/state/mod.rs

@@ -0,0 +1,9 @@
+mod block;
+mod cohort;
+mod cost_basis;
+mod transacted;
+
+pub use block::*;
+pub use cohort::*;
+pub use cost_basis::*;
+pub use transacted::*;

crates/brk_computer/src/distribution/state/transacted.rs

@@ -0,0 +1,50 @@
+use std::ops::{Add, AddAssign};
+
+use brk_cohort::{ByAmountRange, GroupedByType};
+use brk_types::{OutputType, Sats, SupplyState};
+
+#[derive(Default, Debug)]
+pub struct Transacted {
+    pub spendable_supply: SupplyState,
+    pub by_type: GroupedByType<SupplyState>,
+    pub by_size_group: ByAmountRange<SupplyState>,
+}
+
+impl Transacted {
+    #[allow(clippy::inconsistent_digit_grouping)]
+    pub fn iterate(&mut self, value: Sats, _type: OutputType) {
+        let supply = SupplyState {
+            utxo_count: 1,
+            value,
+        };
+
+        *self.by_type.get_mut(_type) += &supply;
+
+        if _type.is_unspendable() {
+            return;
+        }
+
+        self.spendable_supply += &supply;
+
+        *self.by_size_group.get_mut(value) += &supply;
+    }
+}
+
+impl Add for Transacted {
+    type Output = Self;
+    fn add(self, rhs: Self) -> Self::Output {
+        Self {
+            spendable_supply: self.spendable_supply + rhs.spendable_supply,
+            by_type: self.by_type + rhs.by_type,
+            by_size_group: self.by_size_group + rhs.by_size_group,
+        }
+    }
+}
+
+impl AddAssign for Transacted {
+    fn add_assign(&mut self, rhs: Self) {
+        self.by_size_group += rhs.by_size_group;
+        self.spendable_supply += &rhs.spendable_supply;
+        self.by_type += rhs.by_type;
+    }
+}