VaR: the shape of the computation comes before the optimisation

A market risk engine looks nothing like an ingestion pipeline. The inputs are modest — a few million positions, a few hundred scenarios. What costs is what the computation produces from them. Until you see that, you optimise in the wrong place.

What a historical VaR actually computes

The principle is simple to state. Take today's portfolio, apply the market moves observed over a historical window — one or two years of daily scenarios — and you get a distribution of profit and loss. VaR at 99% is a quantile of that distribution; CVaR — the expected shortfall — is the mean of the tail beyond it.

In practice you do not fully revalue each position under each scenario: too expensive. You use a sensitivity approximation — the portfolio's sensitivities times the scenario's shocks, plus possibly a second-order term.

pnl(position, scenario) = Σ_factor  sensi(position, factor) × shock(scenario, factor)

This formula is the key to everything that follows. It says the computation is a matrix product between a sensitivity matrix and a shock matrix — and a matrix product has an optimal execution shape, which is not that of an ETL chain.

The cross product is the real volume

Put orders of magnitude on it. One million positions, two hundred and fifty scenarios, fifty risk factors. The inputs fit in a few gigabytes. The position × scenario cross product is two hundred and fifty million rows before aggregation.

The volume is not what you read, it is what you generate. Any read-side optimisation addresses a few gigabytes; the real cost is in an object a hundred times larger that exists only in memory.

This is why the first instinct — compress better, read fewer columns, add input partitions — gives disappointing gains. It addresses the smaller half of the problem.

Three possible shapes, three very different costs

The same formula can be expressed three ways in Spark, and the cost gap between them is an order of magnitude.

Naive shape: join on the risk factor.

val pnl = sensitivities                       // (position, factor, sensi)
  .join(shocks, Seq("factor"))                // (scenario, factor, shock)
  .withColumn("contrib", col("sensi") * col("shock"))
  .groupBy("position", "scenario")
  .agg(sum("contrib").as("pnl"))

It is correct and it is the worst case. The join on factor has low cardinality — a few dozen distinct values — so the shuffle concentrates the data onto very few partitions. You get massive skew, and a groupBy behind it that redistributes everything a second time.

Improved shape: broadcast the shocks. The shock matrix is small — a few hundred scenarios by a few dozen factors, a few megabytes. It is meant to be broadcast, not joined.

val shocksBc = spark.sparkContext.broadcast(
  shocks.collect().groupBy(_.getAs[String]("factor"))
)

val pnl = sensitivities.mapPartitions { it =>
  val table = shocksBc.value                   // read once per partition
  it.flatMap { row =>
    table(row.factor).map { s =>
      (row.position, s.scenario, row.sensi * s.shock)
    }
  }
}.toDF("position", "scenario", "contrib")

No shuffle to cross the data any more: each sensitivity partition produces its contributions locally. The aggregation remains, but it runs on a high-cardinality key, so it distributes properly.

Dense shape: one vector per position. When the number of scenarios is fixed and moderate, the most efficient representation accumulates contributions into an array of known size, in a single pass, never materialising the cross product.

val n = shocksBc.value.scenarioCount

val pnlByPosition = sensitivities.groupByKey(_.position).mapGroups { (pos, rows) =>
  val vector = new Array[Double](n)           // one array per position
  rows.foreach { r =>
    val factorShocks = shocksBc.value.forFactor(r.factor)
    var s = 0
    while (s < n) { vector(s) += r.sensi * factorShocks(s); s += 1 }
  }
  (pos, vector)
}

This is the shape that changes the scale of the computation: the cross product is never materialised, memory is bounded and predictable, and the final sort for the quantile runs on one array per position instead of two hundred and fifty million rows.

Compute the sensitivities once

The other major gain lies elsewhere and is more mundane: do not recompute what does not depend on the scenario. Sensitivities are a property of the portfolio and the valuation date, not of the scenario. An engine that recomputes them inside the scenario loop does the same work two hundred and fifty times.

It is a common mistake and hard to see, because it is structural: the code is organised as a loop « for each scenario, value the portfolio », which is the natural business phrasing. The efficient phrasing inverts the order: « value once, then apply all scenarios ».

// the heavy computation leaves the loop and is cached
val sensitivities = value(portfolio, valuationDate).persist(MEMORY_AND_DISK)
sensitivities.count()                       // explicit materialisation

// P&L, VaR and CVaR share the same base
val pnl    = applyShocks(sensitivities, shocksBc)
val var99  = quantile(pnl, 0.99)
val cvar99 = tailMean(pnl, 0.99)

The count() after the persist is not decorative: without an action the cache is never filled, and the three uses that follow each recompute the base.

Stable partitioning end to end

Last point, and the one with the most durable gains. If sensitivities, contributions and aggregates are partitioned on the same key, Spark does not need to redistribute between stages.

val sensiPart = sensitivities.repartition(400, col("position"))
                             .sortWithinPartitions("position")
                             .persist()

The choice of key deserves thought. Partitioning by position distributes evenly but breaks portfolio-level aggregates. Partitioning by portfolio keeps aggregates local but skews — portfolios are not the same size. On the engines I optimised, the answer was partitioning by position with a two-step aggregation: partial and local, then final on an already reduced volume.

The general method, if it needs summarising: write the shape of the computation first, measure the volume generated, and only then tune the engine. A well-shaped risk engine on a medium cluster beats a badly shaped one on a large cluster, and it costs less to run every day.