flowchart LR
A[OTC Products] --> B[Portfolio]
B --> C{Analysis Run}
D[Benchmarks] --> E[Analysis Definition]
E --> C
C --> F(Results)
Working with risk managment systems
As a quantitative finance professional you’ll often find yourself with risk management systems (RMS). RMS’s are extensive frameworks that let you properly define a book (portfolio) of your financial transactions and run varia of pricing and risk analysis on it. For big financial players, like investment banks, the RMS will be internal proprietary codbase that is run in-house. For smaller enterprises or second-line reporting it’s not feasable to tackle creating such vast infrastructure. Hence, where there’s a need, someone will try to make money on it. This leads us to third-party (or vendor) RMS, of which there are plenty (ex. Murex, Acadia).
Working with vendor RMS, especially one that covers computations for you, entails juggling multiple resources to obtain your risk metrics. Defining OTC products, benchmarks, portfolios, and running risk analysis can involve numerous API calls, each requiring proper setup and cleanup. This can lead to messy code and potential errors or performance bottlenecks if resources aren’t handled correctly.
Thankfully, Python provides a powerful concept called context managers (CM) that streamline resource managment. True to the language’s ‘batteries included’ philosophy, there’s also a contextlib library that contains variety of tools for easing up your work with CMs. Today we’ll look at a (mock-up) usage of ExitStack class in real-life scenario of running risk analysis on RMS. If you need a refresher on CMs, check out this tutorial by RealPython.
Setting the stage
To run an analysis, the RMS first needs to know what our positions are. In case of tradable assets it’s simple — we provide a market identifier and how much of the instrument we are holding. What do we do if we have some bespoke agreement with specific counterparty (an over-the-counter transaction)? We will need to define it from scratch in the RMS using data from the term sheet (assuming this kind of agreement is covered).
Next, we need to specify the risk metrics we want to calculate — define the analysis scope. Let’s say we hold some equity options and we are intertested in their deltas and beta exposures. The betas are defined with respect to some benchmark — ex. portfolio holding 1 stock in US500 ETF. So we define the benchmark and link it to our analysis.
Finally — once portfolio and analysis are defined in RMS — we call the API to start the calculation and respond with results. This is the control flow we execute to get to this point:
If we know we’re never going to use all of the resources, we should clean up the server artifacts after receving the results. So for each resource we should have a CM.
Mock functions
The setup described above comes from a real-life situation I worked through. I can’t show you the actual API usage or data (or even the name of RMS itself), so we need to define some mocker functions. Mocks like this are actually not an uncommon thing — such approach is prevalent in testing API client code. In our case it would look like this:
from enum import StrEnum
from uuid import uuid4
class MockObject(StrEnum):
"""Types of mock objects."""
ANALYSIS = "analysis"
BENCHMARK = "benchmark"
OTC_PRODUCTS = "otc_products"
PORTFOLIO = "portfolio"
def mock_object(object_type: MockObject) -> str:
"""Mock a UUID for a given object type.
Args:
object_type: Type of object.
"""
return f"{object_type}_{uuid4()}"
def mock_preparation(object_type: MockObject, **kwargs) -> None:
"""Mock preparation of an object.
Args:
object_type: Type of object.
"""
print(f"Preparing {object_type}" + (f" using {kwargs}" if kwargs else "."))
def mock_clean_up(object_uuid: str) -> None:
"""Mock clean up of an object.
Args:
object_uuid: Uuid of the object.
"""
print(f"Cleaning up after {object_uuid}.")For each of the four types of resources we mock the preparation, object (ex. API response, some id of definition on server) and the clean up process.
Context managers
Easiest way to define a CM is through contextlib.contextmanager decorator. To use it, you need a function that returns a generator. Code executed on enter should come before yield statement and the one for the exit afterwards. The generator yields the result of the CM (ex. handle to an opened file), the y in with x(*args) as y:.
from contextlib import contextmanager
from typing import Generator
@contextmanager
def analysis(
*,
benchmark_uuid: str,
) -> Generator[str, None, None]:
"""Mock definition of an analysis.
Example: equity delta and correlation with benchmark.
Args:
benchmark_uuid: Uuid of the benchmark.
"""
mock_preparation(
MockObject.ANALYSIS,
benchmark_name=benchmark_uuid,
)
analysis_uuid = mock_object(MockObject.ANALYSIS)
yield analysis_uuid
mock_clean_up(analysis_uuid)Modern approach to Python development leans heavily towards type annotations. Dynamical typing is powerful but can lead to unwieldy code. To properly annotate the analysis function we need to import Generator from typing module. Remember, the @contextmanager decorator takes the function and turns it into CM — a class with __enter__ and __exit__ methods. The Generator needs three inputs but in our case only the first one is important — YieldType, here str (see for more).
With this done implementing the 3 remaining CMs is easy, just remember our flow chart.
@contextmanager
def benchmark() -> Generator[str, None, None]:
"""Mock definition of a benchmark.
Args:
otc_products_uuid: Uuid of the otc products.
"""
mock_preparation(
MockObject.BENCHMARK,
)
benchmark_uuid = mock_object(MockObject.BENCHMARK)
yield benchmark_uuid
mock_clean_up(benchmark_uuid)
@contextmanager
def otc_products() -> Generator[str, None, None]:
"""Mock definition of an otc products.
Args:
otc_products_uuid: Uuid of the otc products.
"""
mock_preparation(MockObject.OTC_PRODUCTS)
otcs_uuid = mock_object(MockObject.OTC_PRODUCTS)
yield otcs_uuid
mock_clean_up(otcs_uuid)
@contextmanager
def portfolio(
*,
portfolio_name: str,
otc_products_uuid: str,
) -> Generator[str, None, None]:
"""Mock definition of a portfolio.
Args:
otc_products_uuid: Uuid of the otc products.
"""
mock_preparation(
MockObject.PORTFOLIO,
portfolio_name=portfolio_name,
otc_products_uuid=otc_products_uuid,
)
portfolio_uuid = mock_object(MockObject.PORTFOLIO)
yield portfolio_uuid
mock_clean_up(portfolio_uuid)Analysis results
No stress or complexity here, to run the analysis we need to specify which analysis to run on which portfolio.
import pandas as pd
def analysis_results(
*,
analysis_uuid: str,
portfolio_uuid: str,
) -> pd.DataFrame:
"""Mock running the analysis on a given portfolio.
Returns empty dataframe.
Args:
analysis_uuid: Uuid of the analysis.
portfolio_uuid: Uuid of the portfolio.
"""
print(f"Running analysis {analysis_uuid} on portfolio {portfolio_uuid}.")
return pd.DataFrame()Finally, we can run some (mock) risk analysis!
Using contexts directly
First, we use the managers directly through with clause, remembering the dependencies from our flow chart.
PORTFOLIO = "portfolio_1"
def run_analysis() -> pd.DataFrame:
"""Mock running the analysis using with clauses."""
with otc_products() as otc_uuid:
with benchmark() as benchmark_uuid:
with portfolio(
portfolio_name=PORTFOLIO,
otc_products_uuid=otc_uuid,
) as portfolio_uuid:
with analysis(
benchmark_uuid=benchmark_uuid,
) as analysis_uuid:
results = analysis_results(
analysis_uuid=analysis_uuid,
portfolio_uuid=portfolio_uuid,
)
return resultsThis is terrible! I am already getting lost, needed few tries to get it right. We ended up with 6 levels of indentation, the code is confusing, the flow is obtuse. Let’s run it either way, to see if at least works.
def print_title(title: str) -> None:
"""Print a title padded, surrounded by dashes and empty lines."""
print("\n" + title.center(60, "-") + "\n")
print_title("Running analysis.")
run_analysis()
---------------------Running analysis.----------------------
Preparing otc_products.
Preparing benchmark.
Preparing portfolio using {'portfolio_name': 'portfolio_1', 'otc_products_uuid': 'otc_products_69dafefc-792f-4b1f-8008-bbf25a008a61'}
Preparing analysis using {'benchmark_name': 'benchmark_db1187ee-c8d2-4fc2-9cb9-9979f79c6550'}
Running analysis analysis_d2044a97-296b-4dac-91ae-f922e1b9de5d on portfolio portfolio_7e9878f3-c551-4a66-8783-89dd38a1fff0.
Cleaning up after analysis_d2044a97-296b-4dac-91ae-f922e1b9de5d.
Cleaning up after portfolio_7e9878f3-c551-4a66-8783-89dd38a1fff0.
Cleaning up after benchmark_db1187ee-c8d2-4fc2-9cb9-9979f79c6550.
Cleaning up after otc_products_69dafefc-792f-4b1f-8008-bbf25a008a61.Great, the behaviour is as expected, everything is cleaned after nicely. We achieved the goal but the code is unmaintainable. Looks like a subject of the joke “good code makes your job safe for a day, but terrible code in production makes it safe for a lifetime”. Being reckless and with no regard to job security as we are, we’ll fix it.
I can clearly recall the most unamanagable and unreadable code I’ve seen in my career and the culprit was fired in the end. Different reasons, long time later, but still. So the joke is just a joke, don’t rely on a bad code as your job insurance.
The ExitStack
Here comes in the MVP — ExitStack from contextlib, made for streamlining complex context managment situationships. Conceptually it’s just a First-In-Last-Out (FILO) stack. You put CMs on top, one by one. When CM is pushed to stack, its __enter__ method is called and you can intercept the result. ExitStack is a CM itself, it’s __exit__ method is just calling the exits of CMs in reverse order.
flowchart LR
A(Enter CM A) ---> B(Enter CM B)
B ---> C(Enter CM C)
C ---> D[Do stuff]
D ---> E(Exit CM C)
E ---> F(Exit CM B)
F ---> G(Exit CM A)
A -.- G
B -.- F
C -.- E
So the flow is exactly the same as in our first attempt. Let’s try it!
from contextlib import ExitStack
def run_analysis_with_exit_stack() -> None:
"""Mock running the analysis using exit stack."""
with ExitStack() as stack:
otc_uuid = stack.enter_context(otc_products())
benchmark_uuid = stack.enter_context(benchmark())
portfolio_uuid = stack.enter_context(
portfolio(
portfolio_name=PORTFOLIO,
otc_products_uuid=otc_uuid,
)
)
analysis_uuid = stack.enter_context(
analysis(
benchmark_uuid=benchmark_uuid,
)
)
results = analysis_results(
analysis_uuid=analysis_uuid,
portfolio_uuid=portfolio_uuid,
)That’s amazing (if the approach works)! In our code we end up with only single with clause and the outputs of CMs are defined just like the regular variables. We just need to wrap the CM calls in stack.enter_context method that pushes each CM to the stack.
-------------Running analysis with exit stack.--------------
Preparing otc_products.
Preparing benchmark.
Preparing portfolio using {'portfolio_name': 'portfolio_1', 'otc_products_uuid': 'otc_products_02dde7b6-af71-43fe-8b8f-d7fa8550a9d1'}
Preparing analysis using {'benchmark_name': 'benchmark_a47cb0b1-567f-4e24-8781-0a209e1c8a41'}
Running analysis analysis_fc4a8ec1-35b5-4c74-bfb5-a00519c67f79 on portfolio portfolio_313e1e55-bfed-4521-81de-0139f8f0166a.
Cleaning up after analysis_fc4a8ec1-35b5-4c74-bfb5-a00519c67f79.
Cleaning up after portfolio_313e1e55-bfed-4521-81de-0139f8f0166a.
Cleaning up after benchmark_a47cb0b1-567f-4e24-8781-0a209e1c8a41.
Cleaning up after otc_products_02dde7b6-af71-43fe-8b8f-d7fa8550a9d1.It works as well! We also get a package of benefits for free.
Disabling the clean up
Working with API is tricky and debugging could be a painful experience. If we notice something iffy with the results we are reciving, it could be due to a bug at any of the stages. In such case disabling the artifact clean up and examining them is a good way to investigate. How do we do that? Comment out the exit code in our resource CMs? Nope, now we know better. With exit stack approach we just need to clean up the stack before exiting its context.
def run_analysis_with_exit_stack(
clean_up: bool = True,
) -> None:
"""Mock running the analysis using exit stack.
Args:
clean_up: Whether to clean up after the objects.
"""
with ExitStack() as stack:
otc_uuid = stack.enter_context(otc_products())
benchmark_uuid = stack.enter_context(benchmark())
portfolio_uuid = stack.enter_context(
portfolio(
portfolio_name=PORTFOLIO,
otc_products_uuid=otc_uuid,
)
)
analysis_uuid = stack.enter_context(
analysis(
benchmark_uuid=benchmark_uuid,
)
)
results = analysis_results(
analysis_uuid=analysis_uuid,
portfolio_uuid=portfolio_uuid,
)
if not clean_up:
_ = stack.pop_all()
return resultsThe _ = some_function() is a Pythonic way of disregarding outputs of some_function. Method pop_all actually moves the stack contents to a new stack, but we don’t care about that. We just want to get rid of them from our current one.
print_title("Running analysis with exit stack and no clean up.")
run_analysis_with_exit_stack(clean_up=False)
-----Running analysis with exit stack and no clean up.------
Preparing otc_products.
Preparing benchmark.
Preparing portfolio using {'portfolio_name': 'portfolio_1', 'otc_products_uuid': 'otc_products_26a705b0-4131-4832-8a0c-428ae121a2b9'}
Preparing analysis using {'benchmark_name': 'benchmark_0ace28ef-02d7-4123-879c-ceaa15ebfbb8'}
Running analysis analysis_cfa43ae5-9e5b-4dd3-9bd9-345881dcb00e on portfolio portfolio_da057b91-7c24-4725-bcb4-c76033558a96.Multiple portfolios
Benefit #2 — what do we do if we have multiple managers and many portfolios to re-run for? Or — outside of the example scope — we want to held multiple files open at the same time? Easy, we just push to the stack in a loop or a list comprehension.
PORTFOLIOS = ["portfolio_1", "portfolio_2", "portfolio_3"]
def run_analysis_with_exit_stack(clean_up: bool = True):
"""Mock running the analysis for multiple portfolios using exit stack.
Args:
clean_up: Whether to clean up after the objects.
"""
with ExitStack() as stack:
otc_uuid = stack.enter_context(otc_products())
benchmark_uuid = stack.enter_context(benchmark())
portfolio_uuids = [
stack.enter_context(
portfolio(
portfolio_name=portfolio_name,
otc_products_uuid=otc_uuid,
)
)
for portfolio_name in PORTFOLIOS
]
analysis_uuid = stack.enter_context(
analysis(
benchmark_uuid=benchmark_uuid,
)
)
result_parts = [
analysis_results(
analysis_uuid=analysis_uuid,
portfolio_uuid=portfolio_uuid,
)
for portfolio_uuid in portfolio_uuids
]
results = pd.concat(result_parts)
if not clean_up:
_ = stack.pop_all()
return results
print_title("Running analysis with exit stack on multiple portfolios.")
run_analysis_with_exit_stack(clean_up=True)
--Running analysis with exit stack on multiple portfolios.--
Preparing otc_products.
Preparing benchmark.
Preparing portfolio using {'portfolio_name': 'portfolio_1', 'otc_products_uuid': 'otc_products_af35cbd8-cf23-496e-8c70-a243cb830476'}
Preparing portfolio using {'portfolio_name': 'portfolio_2', 'otc_products_uuid': 'otc_products_af35cbd8-cf23-496e-8c70-a243cb830476'}
Preparing portfolio using {'portfolio_name': 'portfolio_3', 'otc_products_uuid': 'otc_products_af35cbd8-cf23-496e-8c70-a243cb830476'}
Preparing analysis using {'benchmark_name': 'benchmark_efbaeaec-5786-45b0-bee9-29dfc1e91cd0'}
Running analysis analysis_bbed7e96-4379-4acb-84f3-290e55488a65 on portfolio portfolio_d6f1c231-78c3-4913-88aa-7e561aa73603.
Running analysis analysis_bbed7e96-4379-4acb-84f3-290e55488a65 on portfolio portfolio_75a4b4ef-b3d0-435e-aabc-20bf4ba2f60b.
Running analysis analysis_bbed7e96-4379-4acb-84f3-290e55488a65 on portfolio portfolio_90644add-d71e-49dd-b96a-96ae991aae2c.
Cleaning up after analysis_bbed7e96-4379-4acb-84f3-290e55488a65.
Cleaning up after portfolio_90644add-d71e-49dd-b96a-96ae991aae2c.
Cleaning up after portfolio_75a4b4ef-b3d0-435e-aabc-20bf4ba2f60b.
Cleaning up after portfolio_d6f1c231-78c3-4913-88aa-7e561aa73603.
Cleaning up after benchmark_efbaeaec-5786-45b0-bee9-29dfc1e91cd0.
Cleaning up after otc_products_af35cbd8-cf23-496e-8c70-a243cb830476.Conclusion
Today we’ve learnt a new Python tool and seen an example of how quantitative developer might set up risk reporting job on vendor RMS. Sound like a very niche and unlikely situation for you? Maybe. But the moral here is to go and explore the Python standard library. Without using any additional packages we improved readability and flexibility of our initial attempt. Python really has ‘batteries included’, see for yourself!
Note
Download the whole code here.