Version 0.1.0 · Apache-2.0 · Python
Holoscan Finance
Holoscan Finance builds a streaming financial-services application as a Holoscan graph. Replay a validated sequence of market ticks, keep a rolling window of statistics for each symbol, and mark a portfolio against its limits. Prices and positions use exact decimals, and the reference application runs offline on checked-in synthetic data.
Technical reference
This module is a technical reference. It is not investment, trading, accounting, regulatory, or risk-management advice, and it is not a production risk engine.
At a glance
| Package | holoscan-finance |
|---|---|
| Version | 0.1.0 |
| Python namespace | holoscan.finance |
| Holoscan SDK | 4.3.0 or later, tested with 4.3.0 |
| Platforms | x86_64, aarch64 |
| Language | Python 3.10 through 3.13 |
| License | Apache-2.0 |
| Maintainer | Bruce Hashemian, NVIDIA |
| Debian package | holoscan-finance |
| PyPI package | holoscan-finance |
Operators
| Operator | What it does |
|---|---|
MarketDataReplayOp |
Emits a finite sequence of validated market ticks, in order, without waiting on wall-clock time |
RollingStatisticsOp |
Keeps an independent window per symbol and emits returns, a moving average, and return deviation |
PortfolioRiskOp |
Marks configured long and short positions with the latest prices and evaluates portfolio limits |
load_market_ticks_csv reads the module’s CSV format: a timezone-aware timestamp, a strictly increasing sequence, a symbol, a positive price, and a three-letter currency.
RollingStatisticsOp emits one features message for every tick. The moving average and return deviation stay empty until the window has enough observations. Return deviation is a descriptive window statistic. It is not annualized volatility or Value at Risk.
PortfolioRiskOp emits a snapshot after every features message, including a partial snapshot while some positions are still unpriced. Limits are checked only when every position has a price. A breach is data on the snapshot, not an error that stops the graph.
Data models
| Model | What it carries |
|---|---|
MarketTick |
One observation: timestamp, sequence, symbol, price, and currency |
MarketFeatures |
That tick plus the symbol’s current rolling window |
Position |
A signed quantity and average entry price. Positive is long, negative is short. |
RiskLimits |
Optional caps on gross exposure, absolute net exposure, and unrealized loss |
PortfolioSnapshot |
Marks, exposures, unrealized P&L, unpriced symbols, and any limit breaches |
Prices, quantities, and P&L use decimal.Decimal. Construct them from strings.
Sample application
intraday_risk_pipeline replays synthetic ticks, computes a short rolling window, marks a small long and short portfolio, and logs each snapshot and limit breach. It needs no network access and no market-data service.
./holohub run intraday_risk_pipeline
Use it in an application
from decimal import Decimal
from holoscan.core import Application
from holoscan.finance import (
MarketDataReplayOp,
PortfolioRiskOp,
Position,
RiskLimits,
RollingStatisticsOp,
load_market_ticks_csv,
)
class RiskApplication(Application):
def compose(self) -> None:
ticks = load_market_ticks_csv("market_ticks.csv")
replay = MarketDataReplayOp(self, ticks=ticks, name="market_data_replay")
statistics = RollingStatisticsOp(self, window_size=20, min_periods=5, name="rolling_statistics")
risk = PortfolioRiskOp(
self,
positions=(Position(symbol="NVDA", quantity=Decimal("100"), average_price=Decimal("150.00")),),
limits=RiskLimits(max_gross_exposure=Decimal("25000"), max_abs_net_exposure=Decimal("25000"), max_unrealized_loss=Decimal("1000")),
name="portfolio_risk",
)
self.add_flow(replay, statistics, {("tick", "tick")})
self.add_flow(statistics, risk, {("features", "features")})
RiskApplication().run()
Install the Holoscan SDK first, then pip install holoscan-finance.
Source is in the Holoscan Finance repository.