orcaset creates financial models as code. Models are defined by a collection of formulas, similar to traditional models in a spreadsheet. However, unlike spreadsheets, formulas in orcaset are not constrained to an eager, fixed cell grid. Instead, values are only materialized when they are requested. This lazy approach separates the model logic from model execution, which enables observability and performance improvements among other benefits.
The core evaluation engine is built around effect handlers implemented in Python using generators. Yielding a dependency is an effect (a task asks for a value without computing it) and the handler is the scheduler that intercepts that demand, evaluates it, and resumes the parent task. This article walks through the core ideas behind that system.
Resolving Dependencies
A model is composed of many (dependency) -> value evaluation tasks where the output from one task feeds the input for the following task. orcaset discovers and resolves those tasks using generators.
Concretely, a Task is a zero-argument function that returns either:
a value of type T, or
a generator (Step[T]) that i) yields its dependencies as tasks, ii) receives their resolved values back via send, and iii) finally returns a value of type T.
Tasks are driven by a scheduler that invokes the task function, receives any dependency requests it yields, evaluates those dependencies, and sends their values back. Those yields are the effects. The scheduler’s control loop is the handler.
The snippet below builds a basic task type and scheduler loop.
from collections.abc import Callable, Generator
from typing import Any
type Task[T] = Callable[[], T | Step[T]]
type Step[T] = Generator[Task[Any], Any, T]
def scheduler[T](task: Task[T]) -> T:
"""Drive a task to completion, recursively resolving yielded dependencies."""
result = task()
if not isinstance(result, Generator):
return result
try:
dep = next(result)
while True:
dep = result.send(scheduler(dep))
except StopIteration as e:
return e.valuefrom collections.abc import Callable, Generator
from typing import Any
type Task[T] = Callable[[], T | Step[T]]
type Step[T] = Generator[Task[Any], Any, T]
def scheduler[T](task: Task[T]) -> T:
"""Drive a task to completion, recursively resolving yielded dependencies."""
result = task()
if not isinstance(result, Generator):
return result
try:
dep = next(result)
while True:
dep = result.send(scheduler(dep))
except StopIteration as e:
return e.valuefrom collections.abc import Callable, Generator
from typing import Any
type Task[T] = Callable[[], T | Step[T]]
type Step[T] = Generator[Task[Any], Any, T]
def scheduler[T](task: Task[T]) -> T:
"""Drive a task to completion, recursively resolving yielded dependencies."""
result = task()
if not isinstance(result, Generator):
return result
try:
dep = next(result)
while True:
dep = result.send(scheduler(dep))
except StopIteration as e:
return e.valueWe can model a simple two-period revenue model using two functions. jan_revenue is a leaf that returns the initial value of 100. feb_revenue equals January revenue increased by 10%.
def jan_revenue() -> float:
return 100.0
def feb_revenue() -> Step[float]:
jan_rev = yield jan_revenue
return jan_rev * 1.1
print(scheduler(feb_revenue)) def jan_revenue() -> float:
return 100.0
def feb_revenue() -> Step[float]:
jan_rev = yield jan_revenue
return jan_rev * 1.1
print(scheduler(feb_revenue)) def jan_revenue() -> float:
return 100.0
def feb_revenue() -> Step[float]:
jan_rev = yield jan_revenue
return jan_rev * 1.1
print(scheduler(feb_revenue)) February revenue isn’t known until the scheduler materializes it. The execution flow looks like this:
Evaluate feb_revenue, which immediately hits the dependency on jan_revenue
Pause feb_revenue and hand control back to the scheduler by yielding the jan_revenue task
The scheduler calls the jan_revenue task, which returns 100.0
The scheduler sends 100.0 back to feb_revenue
feb_revenue resumes execution and returns 100.0 * 1.1
Keyed Dependencies
The prior revenue model works, but it is awkward. Each monthly period is a separate function. Ideally we’d define the entire revenue line as one function parameterized by period. Conceptually something like:
def revenue(period: int):
"""Revenue for periods i >= 0"""
if period == 0:
return 100
else:
return revenue(period - 1) * 1.1def revenue(period: int):
"""Revenue for periods i >= 0"""
if period == 0:
return 100
else:
return revenue(period - 1) * 1.1def revenue(period: int):
"""Revenue for periods i >= 0"""
if period == 0:
return 100
else:
return revenue(period - 1) * 1.1This sketch is ordinary Python recursion, not yet wired through the scheduler. In addition to being more concise, this form clearly highlights value-level dependencies: revenue for period i depends on revenue at period i - 1.
The current definition of Task only takes zero-argument callables as dependencies. There’s no way to pass the period argument through the scheduler. In order to pass it through, we need a new task definition that pairs a single-argument function with an argument value.
type KeyedTask[K, T] = tuple[Callable[[K], Step[T] | T], K]
type Step[T] = Generator[Task[Any] | KeyedTask[Any, Any], Any, T]
type KeyedTask[K, T] = tuple[Callable[[K], Step[T] | T], K]
type Step[T] = Generator[Task[Any] | KeyedTask[Any, Any], Any, T]
type KeyedTask[K, T] = tuple[Callable[[K], Step[T] | T], K]
type Step[T] = Generator[Task[Any] | KeyedTask[Any, Any], Any, T]
The scheduler also needs to accept keyed tasks and pass the key as the function argument.
def scheduler[K, T](task: Task[T] | KeyedTask[K, T]) -> T:
"""Drive a task to completion, recursively resolving yielded dependencies."""
if isinstance(task, tuple):
fn, key = task
result = fn(key)
else:
result = task()
if not isinstance(result, Generator):
return result
try:
dep = next(result)
while True:
dep = result.send(scheduler(dep))
except StopIteration as e:
return e.valuedef scheduler[K, T](task: Task[T] | KeyedTask[K, T]) -> T:
"""Drive a task to completion, recursively resolving yielded dependencies."""
if isinstance(task, tuple):
fn, key = task
result = fn(key)
else:
result = task()
if not isinstance(result, Generator):
return result
try:
dep = next(result)
while True:
dep = result.send(scheduler(dep))
except StopIteration as e:
return e.valuedef scheduler[K, T](task: Task[T] | KeyedTask[K, T]) -> T:
"""Drive a task to completion, recursively resolving yielded dependencies."""
if isinstance(task, tuple):
fn, key = task
result = fn(key)
else:
result = task()
if not isinstance(result, Generator):
return result
try:
dep = next(result)
while True:
dep = result.send(scheduler(dep))
except StopIteration as e:
return e.valueNow we can parameterize tasks by period and define the entire revenue line in a single function.
def revenue(period: int):
if period == 0:
return 100.0
else:
prior_revenue = yield (revenue, period - 1)
return prior_revenue * 1.1
for i in range(3):
print(f"Revenue for period {i}: {scheduler((revenue, i))}")
def revenue(period: int):
if period == 0:
return 100.0
else:
prior_revenue = yield (revenue, period - 1)
return prior_revenue * 1.1
for i in range(3):
print(f"Revenue for period {i}: {scheduler((revenue, i))}")
def revenue(period: int):
if period == 0:
return 100.0
else:
prior_revenue = yield (revenue, period - 1)
return prior_revenue * 1.1
for i in range(3):
print(f"Revenue for period {i}: {scheduler((revenue, i))}")
In orcaset, users don’t yield (task, key) pairs directly. They go through thin wrappers, mainly just for typing purposes in the full library. Conceptually they are just:
def get[T](task: Task[T]) -> Step[T]:
value = yield task
return value
def get_at[K: Hashable, T](task: Callable[[K], Step[T] | T], key: K) -> Step[T]:
value = yield (task, key)
return valuedef get[T](task: Task[T]) -> Step[T]:
value = yield task
return value
def get_at[K: Hashable, T](task: Callable[[K], Step[T] | T], key: K) -> Step[T]:
value = yield (task, key)
return valuedef get[T](task: Task[T]) -> Step[T]:
value = yield task
return value
def get_at[K: Hashable, T](task: Callable[[K], Step[T] | T], key: K) -> Step[T]:
value = yield (task, key)
return valueSo the demand in revenue would be written prior_revenue = yield from get_at(revenue, period - 1). The rest of this article keeps yielding tuples so the scheduler mechanics stay visible.
Tracking Dependencies
Effect handlers separate model logic (tasks that demand values) from model execution (the scheduler that satisfies those demands). We can extend the handler without changing task code. For example, to track dependencies as demands are resolved.
We can trace dependencies by adding the currently executing task to a stack and saving an edge from the current task to any new tasks that are discovered during execution. Since we need to hold state during a resolution run, we’ll wrap the scheduler function into a class Context with instance variables for the stack and the dependency edges.
class Context:
"""Evaluate keyed tasks while recording which cells demand which dependencies."""
def __init__(self) -> None:
self._stack: list[KeyedTask[Any, Any]] = []
self.deps: dict[KeyedTask[Any, Any], set[KeyedTask[Any, Any]]] = {}
def get_at[K, T](self, task: KeyedTask[K, T]) -> T:
if self._stack:
self.deps.setdefault(self._stack[-1], set()).add(task)
fn, key = task
self._stack.append(task)
try:
result = fn(key)
if not isinstance(result, Generator):
return result
try:
dep = next(result)
while True:
assert isinstance(dep, tuple)
dep = result.send(self.get_at(dep))
except StopIteration as e:
return e.value
finally:
self._stack.pop()class Context:
"""Evaluate keyed tasks while recording which cells demand which dependencies."""
def __init__(self) -> None:
self._stack: list[KeyedTask[Any, Any]] = []
self.deps: dict[KeyedTask[Any, Any], set[KeyedTask[Any, Any]]] = {}
def get_at[K, T](self, task: KeyedTask[K, T]) -> T:
if self._stack:
self.deps.setdefault(self._stack[-1], set()).add(task)
fn, key = task
self._stack.append(task)
try:
result = fn(key)
if not isinstance(result, Generator):
return result
try:
dep = next(result)
while True:
assert isinstance(dep, tuple)
dep = result.send(self.get_at(dep))
except StopIteration as e:
return e.value
finally:
self._stack.pop()class Context:
"""Evaluate keyed tasks while recording which cells demand which dependencies."""
def __init__(self) -> None:
self._stack: list[KeyedTask[Any, Any]] = []
self.deps: dict[KeyedTask[Any, Any], set[KeyedTask[Any, Any]]] = {}
def get_at[K, T](self, task: KeyedTask[K, T]) -> T:
if self._stack:
self.deps.setdefault(self._stack[-1], set()).add(task)
fn, key = task
self._stack.append(task)
try:
result = fn(key)
if not isinstance(result, Generator):
return result
try:
dep = next(result)
while True:
assert isinstance(dep, tuple)
dep = result.send(self.get_at(dep))
except StopIteration as e:
return e.value
finally:
self._stack.pop()A Context records dependency edges for all tasks it evaluates. The example below prints dependencies for the first three revenue periods. The edge labels are not pretty, but they clearly show the dependency chain from revenue at period 2 to revenue at period 1 to revenue at period 0.
ctx = Context()
for i in range(3):
print(f"Revenue for period {i}: {ctx.get_at((revenue, i))}")
for cell, children in ctx.deps.items():
print(f"{cell} -> {children}")
ctx = Context()
for i in range(3):
print(f"Revenue for period {i}: {ctx.get_at((revenue, i))}")
for cell, children in ctx.deps.items():
print(f"{cell} -> {children}")
ctx = Context()
for i in range(3):
print(f"Revenue for period {i}: {ctx.get_at((revenue, i))}")
for cell, children in ctx.deps.items():
print(f"{cell} -> {children}")
The orcaset library uses the same principles to track dependencies across calculations, with better task identities, human-readable labels, and a more ergonomic API.
Effect Handlers in Orcaset
The main purpose of effect handlers in orcaset is to separate model logic from execution. It gives users granular control over the flow of programs to perform side effects that make models faster, open, and inspectable. Specifically, it allows orcaset users to:
Track dependencies: Dependency graphs for debugging model logic and verifying results
Memoize values: Financial models are highly recursive; the work explodes without memoization
Mock values: Useful for testing or scenario analysis
Step through calculations: Inspect model drivers and execution flow
Replay results: Re-emit prior resolutions for auditability and reproducibility
While the ideas here match orcaset‘s effect handler system, the library implements the scheduler differently for scalability. For example, the recursive handler in this article (dep = result.send(scheduler(dep))) adds a Python stack frame per task. Moderately deep models would hit CPython’s default 1,000-frame limit and raise a RecursionError. orcaset uses an iterative scheduler so dependency chains can be arbitrarily deep.
This post might vaguely remind you of a build system. There are many (dependency) -> value tasks that need to be discovered and resolved in order. The design of orcaset is partly inspired by Build Systems à la Carte: Theory and Practice (Andrey Mokhov, Neil Mitchell and Simon Peyton Jones). Calculation tasks are ordered using a suspending scheduler via generators (rather than a topological sort or restarting scheduler). Values in orcaset are currently static so the “rebuilder” dimension from the paper is not applicable, but future library development may add reactivity which would fit the library into the authors’ scheduler-rebuilder matrix nicely.
Star the repository at https://github.com/orcaset/orcaset-py to follow future development or examine the full codebase!
