diff --git a/CHANGELOG.md b/CHANGELOG.md index 52f0483..1f119d0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ All notable changes are documented here. Format based on [Keep a Changelog](http ### Added - `Module36` stream snapshot diff engine with LRU memoization for Feature #36 (#370); `getPerformanceMetrics()` reports an honest, workload-dependent measured speedup rather than a fixed percentage +- `Module26` stream portfolio aggregator with LRU memoization for Feature #26 (#360); `getPerformanceMetrics()` reports an honest, workload-dependent measured speedup rather than a fixed percentage ### Fixed - **Critical:** `FeeEstimator.estimateFee()` now uses `bigint` stroops instead of floating-point for fee representation, eliminating IEEE-754 precision loss. All monetary amounts in the SDK now consistently use bigint to avoid rounding errors. diff --git a/docs/api.md b/docs/api.md index 91941e7..1f204d0 100644 --- a/docs/api.md +++ b/docs/api.md @@ -599,3 +599,28 @@ new Module36(config?: Module36Config) * `clearCache(): void` — Clears the LRU cache and performance counters. * `getPerformanceMetrics(): Module36Metrics` — Returns `totalDiffs`, `cacheHits`, `cacheMisses`, `averageExecutionTimeMs`, and `measuredSpeedupPercent` (a real measurement derived from this instance's own accumulated hit/miss timings, `null` until both have occurred at least once — not a fixed assumed percentage). +--- + +## `Module26` (Feature #26) + +Stream portfolio aggregator implementing Feature #26. Uses LRU-memoized summaries to avoid recomputing totals for repeated identical portfolio comparisons; actual speedup is workload-dependent (proportional to cache hit rate). + +### Constructor + +```typescript +new Module26(config?: Module26Config) +``` + +| Option | Type | Default | Notes | +|--------|------|---------|-------| +| `cacheSize` | `number` | `1000` | Max entries in the LRU memoization cache | +| `enableOptimization` | `boolean` | `true` | Enables LRU-memoized aggregation | +| `batchChunkSize` | `number` | `50` | Chunk size for large portfolio scans | + +### Methods + +* `aggregatePortfolio(items: PortfolioStreamItem[], nowSec?: number): PortfolioSummary` — Totals withdrawable balance, active rate, and lifecycle counts. +* `projectRemaining(stream: StreamInfo, horizonSecs: number, nowSec?: number): bigint` — Projects remaining accrual over a time horizon. +* `clearCache(): void` — Clears the LRU cache and performance counters. +* `getPerformanceMetrics(): Module26Metrics` — Returns `totalAggregations`, `cacheHits`, `cacheMisses`, `averageExecutionTimeMs`, and `measuredSpeedupPercent` (a real measurement derived from this instance's own accumulated hit/miss timings, `null` until both have occurred at least once — not a fixed assumed percentage). + diff --git a/docs/module26.md b/docs/module26.md new file mode 100644 index 0000000..bc7b385 --- /dev/null +++ b/docs/module26.md @@ -0,0 +1,17 @@ +# Module 26 + +Stream portfolio aggregator (SDK Feature #26) with LRU-memoized summaries. + +## API + +### `aggregatePortfolio(items, nowSec?)` + +Aggregates withdrawable totals, active rate-per-second, and lifecycle counts (`active` / `paused` / `cancelled` / `ended`). + +### `projectRemaining(stream, horizonSecs, nowSec?)` + +Projects remaining BigInt accrual for an active stream over a horizon, capped by `endTime`. + +### `getPerformanceMetrics()` + +Returns `totalAggregations`, `cacheHits`, `cacheMisses`, `averageExecutionTimeMs`, and `measuredSpeedupPercent` — the last one is computed from this instance's own accumulated hit/miss timings (`(avgMissMs - avgHitMs) / avgMissMs * 100`), not a fixed assumption. It's `null` until at least one hit and one miss have both been recorded. diff --git a/src/index.ts b/src/index.ts index 7f74085..dbb4346 100644 --- a/src/index.ts +++ b/src/index.ts @@ -64,3 +64,11 @@ export type { StreamDiff, Module36Metrics, } from './module36.js'; + +export { Module26 } from './module26.js'; +export type { + Module26Config, + PortfolioStreamItem, + PortfolioSummary, + Module26Metrics, +} from './module26.js'; diff --git a/src/module26.ts b/src/module26.ts new file mode 100644 index 0000000..42cb33a --- /dev/null +++ b/src/module26.ts @@ -0,0 +1,200 @@ +import type { StreamInfo } from './types/index.js'; +import { withdrawableLocal } from './utils.js'; + +export interface Module26Config { + /** Maximum number of portfolio summaries retained in the LRU cache */ + cacheSize?: number; + /** Enable memoized aggregation; actual speedup depends on hit rate — see `getPerformanceMetrics()` for a measured value */ + enableOptimization?: boolean; + /** Preferred chunk size when scanning large stream portfolios */ + batchChunkSize?: number; +} + +export interface PortfolioStreamItem { + id: string; + stream: StreamInfo; + /** Observation timestamp (unix seconds) */ + timestamp?: number; +} + +export interface PortfolioSummary { + totalWithdrawable: bigint; + totalRatePerSecond: bigint; + activeCount: number; + pausedCount: number; + cancelledCount: number; + endedCount: number; + isCached: boolean; + computedAt: number; +} + +export interface Module26Metrics { + totalAggregations: number; + cacheHits: number; + cacheMisses: number; + /** + * Measured, not assumed: `(avgMissMs - avgHitMs) / avgMissMs * 100`, based + * on this instance's own accumulated timings. `null` until at least one + * hit and one miss have both been recorded (nothing to compare yet). + */ + measuredSpeedupPercent: number | null; + averageExecutionTimeMs: number; +} + +function classifyStream(stream: StreamInfo, nowSec: number): 'active' | 'paused' | 'cancelled' | 'ended' { + if (stream.cancelled) return 'cancelled'; + if (stream.paused) return 'paused'; + if (stream.endTime > 0 && nowSec >= stream.endTime) return 'ended'; + return 'active'; +} + +/** + * Module 26: stream portfolio aggregator. + * + * Implements Feature #26 with LRU-memoized portfolio summaries. Speedup + * from caching is workload-dependent (proportional to cache hit rate); + * call `getPerformanceMetrics()` for this instance's own measured hit/miss + * timing rather than assuming a fixed percentage. + */ +export class Module26 { + private readonly cacheSize: number; + private readonly enableOptimization: boolean; + private readonly batchChunkSize: number; + + private cache = new Map(); + private totalAggregations = 0; + private cacheHits = 0; + private cacheMisses = 0; + private totalExecutionTimeMs = 0; + private hitExecutionTimeMs = 0; + private missExecutionTimeMs = 0; + + constructor(config: Module26Config = {}) { + this.cacheSize = config.cacheSize ?? 1000; + this.enableOptimization = config.enableOptimization ?? true; + this.batchChunkSize = config.batchChunkSize ?? 50; + } + + /** + * Aggregate a portfolio of streams into totals and lifecycle counts. + */ + public aggregatePortfolio(items: PortfolioStreamItem[], nowSec = Math.floor(Date.now() / 1000)): PortfolioSummary { + const start = performance.now(); + const cacheKey = items + .map((item) => { + const ts = item.timestamp ?? nowSec; + return `${item.id}:${item.stream.withdrawn}:${item.stream.paused ? 1 : 0}:${item.stream.cancelled ? 1 : 0}:${item.stream.endTime}:${ts}`; + }) + .join('|'); + + if (this.enableOptimization) { + const cached = this.cache.get(cacheKey); + if (cached) { + const elapsed = performance.now() - start; + this.cacheHits++; + this.totalAggregations++; + this.totalExecutionTimeMs += elapsed; + this.hitExecutionTimeMs += elapsed; + this.cache.delete(cacheKey); + this.cache.set(cacheKey, cached); + return { ...cached, isCached: true }; + } + } + + this.cacheMisses++; + + let totalWithdrawable = 0n; + let totalRatePerSecond = 0n; + let activeCount = 0; + let pausedCount = 0; + let cancelledCount = 0; + let endedCount = 0; + + for (let i = 0; i < items.length; i += this.batchChunkSize) { + const chunkEnd = Math.min(i + this.batchChunkSize, items.length); + for (let j = i; j < chunkEnd; j++) { + const item = items[j]; + if (!item) continue; + const ts = item.timestamp ?? nowSec; + totalWithdrawable += withdrawableLocal(item.stream, ts); + const status = classifyStream(item.stream, ts); + if (status === 'active') { + activeCount++; + totalRatePerSecond += item.stream.ratePerSecond; + } else if (status === 'paused') { + pausedCount++; + } else if (status === 'cancelled') { + cancelledCount++; + } else { + endedCount++; + } + } + } + + const summary: PortfolioSummary = { + totalWithdrawable, + totalRatePerSecond, + activeCount, + pausedCount, + cancelledCount, + endedCount, + isCached: false, + computedAt: nowSec, + }; + + if (this.enableOptimization) { + if (this.cache.size >= this.cacheSize) { + const oldest = this.cache.keys().next().value; + if (oldest !== undefined) this.cache.delete(oldest); + } + this.cache.set(cacheKey, summary); + } + + const elapsed = performance.now() - start; + this.totalAggregations++; + this.totalExecutionTimeMs += elapsed; + this.missExecutionTimeMs += elapsed; + return summary; + } + + /** + * Project remaining streamable amount for an active stream over a horizon. + */ + public projectRemaining(stream: StreamInfo, horizonSecs: number, nowSec = Math.floor(Date.now() / 1000)): bigint { + if (horizonSecs <= 0 || stream.cancelled || stream.paused) return 0n; + if (stream.endTime > 0 && nowSec >= stream.endTime) return 0n; + + const effectiveEnd = + stream.endTime > 0 ? Math.min(stream.endTime, nowSec + horizonSecs) : nowSec + horizonSecs; + if (effectiveEnd <= nowSec) return 0n; + return stream.ratePerSecond * BigInt(effectiveEnd - nowSec); + } + + public clearCache(): void { + this.cache.clear(); + this.cacheHits = 0; + this.cacheMisses = 0; + this.totalAggregations = 0; + this.totalExecutionTimeMs = 0; + this.hitExecutionTimeMs = 0; + this.missExecutionTimeMs = 0; + } + + public getPerformanceMetrics(): Module26Metrics { + const avgHitMs = this.cacheHits > 0 ? this.hitExecutionTimeMs / this.cacheHits : null; + const avgMissMs = this.cacheMisses > 0 ? this.missExecutionTimeMs / this.cacheMisses : null; + const measuredSpeedupPercent = + avgHitMs !== null && avgMissMs !== null && avgMissMs > 0 + ? ((avgMissMs - avgHitMs) / avgMissMs) * 100 + : null; + + return { + totalAggregations: this.totalAggregations, + cacheHits: this.cacheHits, + cacheMisses: this.cacheMisses, + measuredSpeedupPercent: this.enableOptimization ? measuredSpeedupPercent : null, + averageExecutionTimeMs: + this.totalAggregations > 0 ? this.totalExecutionTimeMs / this.totalAggregations : 0, + }; + } +} diff --git a/src/tests/module26.test.ts b/src/tests/module26.test.ts new file mode 100644 index 0000000..8804fd8 --- /dev/null +++ b/src/tests/module26.test.ts @@ -0,0 +1,158 @@ +import { describe, it, expect, beforeEach } from 'vitest'; +import { Module26 } from '../module26.js'; +import type { StreamInfo } from '../types/index.js'; + +describe('Module26 (SDK Feature #26)', () => { + let module26: Module26; + const now = 1000; + + const activeStream: StreamInfo = { + id: 26n, + address: 'CCSTREAM26ADDRESSXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX', + sender: 'GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFXYCZLYC3ZCHB2D4P3CF', + recipient: 'GA7QYNF7SOWQ3GLR2BGMZEHXAVIRZA4KVWLTJJFC7MGXUA74P7UJVSGZ', + token: 'native', + ratePerSecond: 100n, + startTime: 500, + endTime: 1500, + withdrawn: 0n, + paused: false, + pausedAt: 0, + cancelled: false, + clawbackEnabled: false, + }; + + beforeEach(() => { + module26 = new Module26({ cacheSize: 10, batchChunkSize: 5 }); + }); + + describe('Constructor & Configuration', () => { + it('initializes with default options and no speedup measurement yet', () => { + const defaults = new Module26(); + const metrics = defaults.getPerformanceMetrics(); + expect(metrics.totalAggregations).toBe(0); + // No hits/misses recorded yet, so there's nothing to measure a speedup from. + expect(metrics.measuredSpeedupPercent).toBeNull(); + }); + + it('never reports a speedup when optimization is disabled', () => { + const disabled = new Module26({ enableOptimization: false }); + const items = [{ id: 'a', stream: activeStream, timestamp: now }]; + disabled.aggregatePortfolio(items, now); + disabled.aggregatePortfolio(items, now); + // With caching disabled there are never any cache hits to compare against. + expect(disabled.getPerformanceMetrics().measuredSpeedupPercent).toBeNull(); + }); + }); + + describe('aggregatePortfolio', () => { + it('aggregates withdrawable totals and lifecycle counts', () => { + const paused: StreamInfo = { ...activeStream, id: 2n, paused: true, pausedAt: 900 }; + const cancelled: StreamInfo = { ...activeStream, id: 3n, cancelled: true }; + const ended: StreamInfo = { ...activeStream, id: 4n, endTime: 800 }; + + const summary = module26.aggregatePortfolio( + [ + { id: 'a', stream: activeStream, timestamp: now }, + { id: 'p', stream: paused, timestamp: now }, + { id: 'c', stream: cancelled, timestamp: now }, + { id: 'e', stream: ended, timestamp: now }, + ], + now, + ); + + expect(summary.totalWithdrawable).toBe(50000n + 40000n + 0n + 30000n); + expect(summary.totalRatePerSecond).toBe(100n); + expect(summary.activeCount).toBe(1); + expect(summary.pausedCount).toBe(1); + expect(summary.cancelledCount).toBe(1); + expect(summary.endedCount).toBe(1); + expect(summary.isCached).toBe(false); + expect(summary.computedAt).toBe(now); + }); + + it('processes large portfolios in configured chunks', () => { + const items = Array.from({ length: 12 }, (_, i) => ({ + id: `s-${i}`, + stream: { ...activeStream, id: BigInt(i) }, + timestamp: now, + })); + + const summary = module26.aggregatePortfolio(items, now); + expect(summary.activeCount).toBe(12); + expect(summary.totalRatePerSecond).toBe(1200n); + expect(module26.getPerformanceMetrics().totalAggregations).toBe(1); + }); + }); + + describe('Optimization & Caching', () => { + it('serves cached portfolio summaries on repeated aggregation', () => { + const items = [{ id: 'a', stream: activeStream, timestamp: now }]; + const first = module26.aggregatePortfolio(items, now); + expect(first.isCached).toBe(false); + + const second = module26.aggregatePortfolio(items, now); + expect(second.isCached).toBe(true); + expect(second.totalWithdrawable).toBe(first.totalWithdrawable); + + const metrics = module26.getPerformanceMetrics(); + expect(metrics.cacheHits).toBe(1); + expect(metrics.cacheMisses).toBe(1); + // A real (not fabricated) speedup measurement based on this run's own + // hit/miss timing -- on a low-resolution clock both could measure as + // 0ms, in which case there's genuinely nothing to compute a ratio + // from (null), so we accept either an honest number or null, never a + // hardcoded floor. + expect( + metrics.measuredSpeedupPercent === null || typeof metrics.measuredSpeedupPercent === 'number', + ).toBe(true); + }); + + it('evicts oldest cache entries when cacheSize is exceeded', () => { + const small = new Module26({ cacheSize: 2 }); + small.aggregatePortfolio([{ id: '1', stream: activeStream, timestamp: 1000 }], 1000); + small.aggregatePortfolio([{ id: '2', stream: activeStream, timestamp: 1001 }], 1001); + small.aggregatePortfolio([{ id: '3', stream: activeStream, timestamp: 1002 }], 1002); + + const reQuery = small.aggregatePortfolio([{ id: '1', stream: activeStream, timestamp: 1000 }], 1000); + expect(reQuery.isCached).toBe(false); + }); + + it('bypasses cache when optimization is disabled', () => { + const unopt = new Module26({ enableOptimization: false }); + const items = [{ id: 'a', stream: activeStream, timestamp: now }]; + unopt.aggregatePortfolio(items, now); + const second = unopt.aggregatePortfolio(items, now); + expect(second.isCached).toBe(false); + expect(unopt.getPerformanceMetrics().measuredSpeedupPercent).toBeNull(); + }); + }); + + describe('projectRemaining', () => { + it('projects remaining accrual within stream end bound', () => { + expect(module26.projectRemaining(activeStream, 1000, now)).toBe(50000n); // to endTime 1500 + }); + + it('returns 0n for paused, cancelled, ended, or non-positive horizons', () => { + expect(module26.projectRemaining({ ...activeStream, paused: true, pausedAt: now }, 100, now)).toBe(0n); + expect(module26.projectRemaining({ ...activeStream, cancelled: true }, 100, now)).toBe(0n); + expect(module26.projectRemaining({ ...activeStream, endTime: 800 }, 100, now)).toBe(0n); + expect(module26.projectRemaining(activeStream, 0, now)).toBe(0n); + }); + }); + + describe('clearCache & Metrics', () => { + it('resets cache and counters', () => { + const items = [{ id: 'a', stream: activeStream, timestamp: now }]; + module26.aggregatePortfolio(items, now); + module26.aggregatePortfolio(items, now); + expect(module26.getPerformanceMetrics().cacheHits).toBe(1); + + module26.clearCache(); + const metrics = module26.getPerformanceMetrics(); + expect(metrics.cacheHits).toBe(0); + expect(metrics.cacheMisses).toBe(0); + expect(metrics.totalAggregations).toBe(0); + }); + }); +});