import warnings
from functools import reduce
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from IPython.display import clear_output, Markdown
from inflection import titleize
from tqdm import tqdm
from tulip.data.bloomberg import BloombergClient as bb
from tulip.plots import plot_lines
warnings.filterwarnings("ignore")
def prettify_legend(str):
replacements = {
"Us": "US",
"Eu": "EU",
"Jp": "JP",
# add more: 'old': 'new',
}
out = titleize(str)
replace_all = lambda s: reduce(
lambda acc, kv: acc.replace(*kv), replacements.items(), s
)
return replace_all(out)
def corr_styler(df, caption="Correlation Matrix"):
"""
Returns a pandas Styler for the correlation matrix of df,
with a coolwarm background gradient and 2-decimal formatting.
"""
corr = df.corr()
return (
corr.style.background_gradient(cmap="coolwarm") # or 'RdBu', 'viridis', etc.
.format("{:.2f}") # two decimal places
.set_caption(caption)
)
def style_correlation_lower_triangle(
df, font_size="90%", width="50px", index_width="100px", truncate_n=None
):
"""
Styles the lower triangle of a correlation matrix DataFrame.
Parameters:
- df: pandas DataFrame of correlation values.
- font_size: CSS font-size for all cells.
- width: CSS width for column header and data cells.
- index_width: CSS width for the index (row header) cells.
- truncate_n: Optional int. If provided, column labels longer than truncate_n
will be truncated to truncate_n-3 characters with '...'.
"""
# mask out upper triangle
mask = np.triu(np.ones(df.shape), k=1).astype(bool)
df_masked = df.mask(mask).copy()
# apply first-word reduction
df_masked.index.name = None
df_masked.columns.name = None
# optional truncation of column labels
if truncate_n is not None:
def trunc_label(s):
return s if len(s) <= truncate_n else s[: truncate_n - 3] + "..."
df_masked.columns = df_masked.columns.map(trunc_label)
# styling
styled = (
df_masked.style.background_gradient(cmap="RdYlGn_r", vmin=-1, vmax=1, axis=None)
.map(lambda val: "background-color: white" if pd.isna(val) else "")
.format("{:.1%}", na_rep="")
.set_table_styles(
[
# data cells
{
"selector": "td",
"props": [
("font-size", font_size),
("text-align", "center"),
("width", width),
("height", width),
("padding", "0px"),
],
},
# column headers
{
"selector": "th.col_heading",
"props": [
("font-size", font_size),
("text-align", "center"),
("width", width),
("height", width),
("padding", "0px"),
],
},
# row headers (index)
{
"selector": "th.row_heading",
"props": [
("font-size", font_size),
("text-align", "center"),
("width", index_width),
("height", width),
("padding", "0px"),
],
},
# table properties
{"selector": "table", "props": [("border-collapse", "collapse")]},
]
)
.set_properties(**{"text-align": "center", "padding": "0px"})
)
return styled
clear_output()ASSET_CLASSES = dict(
gold_london="GOLDLNPM Index",
gold_us="XAU Curncy",
US_Equities="SPX Index",
JP_Equities="TPX Index",
dollar="DXY Index",
WLD_ex_US_equities="TGPVAN17 Index", # FTSE Developed All Cap ex US Net Tax (US RIC) Index
US_Treasuries_Agg="LUATTRUU Index",
US_Govt_2YR_bond="TU1 Comdty",
US_Govt_5YR_bond="FV1 Comdty",
US_Govt_10YR_bond="TY1 Comdty",
EU_Govt_2YR_bond="DU1 Comdty",
EU_Govt_5YR_bond="OE1 Comdty",
EU_Govt_10YR_bond="RX1 Comdty",
)
FIGSIZE = (1000, 400)
X_RANGE = ("2015-01-01", "2025-12-31")
Y_RANGE = (1, -1)list(ASSET_CLASSES.values())from polars_bloomberg import BQuery
with BQuery() as bq:
MARKET_DATA = bq.bdh(
list(ASSET_CLASSES.values()),
["PX_LAST", "PX_CLOSE_1D", "YLD_YTM_MID"],
start_date=pd.Timestamp("2010-01-01"),
end_date=pd.Timestamp.today(),
)import polars as pl
returns = (
MARKET_DATA.with_columns(
pl.col("PX_LAST").truediv(pl.col("PX_CLOSE_1D")).sub(1).alias("return")
)
.select(["security", "date", "return"])
.filter(pl.col("return").is_not_null())
)
returns = (
returns.pivot(
values="return",
index="date",
columns="security",
)
.to_pandas()
.set_index("date")
)returns = returns.rename(columns={v: k for k, v in ASSET_CLASSES.items()})# Concatenate all computed returns along columns, sort the index, and drop rows with all NaNs
reference_asset = "US_Equities"
corr_30_days = (
returns.dropna()
.rolling(30, min_periods=20)
.corr()
.xs(reference_asset, level=1)
.drop(reference_asset, axis=1)
)
corr_60_days = (
returns.dropna()
.rolling(60, min_periods=20)
.corr()
.xs(reference_asset, level=1)
.drop(reference_asset, axis=1)
)
corr_90_days = (
returns.dropna()
.rolling(90, min_periods=20)
.corr()
.xs(reference_asset, level=1)
.drop(reference_asset, axis=1)
)
clear_output()Correlation Monitor¶
days = 90
all_corr_days = returns.dropna().rolling(days, min_periods=20).corr()
last_date = all_corr_days.index.get_level_values("date").max()
back_n = all_corr_days.index.get_level_values("date")[-days]
last_corr = all_corr_days.xs(last_date)
back_n_corr = all_corr_days.xs(back_n)
last_corr.index = last_corr.columns = last_corr.index.map(prettify_legend)
back_n_corr.index = back_n_corr.columns = back_n_corr.index.map(prettify_legend)
style_correlation_lower_triangle(
last_corr, truncate_n=15, index_width="120px"
).set_caption(caption=f"Last {days} day Correlation Matrix")style_correlation_lower_triangle(
back_n_corr, truncate_n=12, index_width="120px"
).set_caption(caption=f"{days} day Correlation Matrix {days} days ago")Ex-US Equities (Against US Equities)¶
Y_RANGE = (1, -1)
for asset in ["JP_Equities", "WLD_ex_US_equities"]:
plot_lines(
[
corr_30_days[asset].rename("30 trading days"),
corr_60_days[asset].rename("60 trading days"),
corr_90_days[asset].rename("90 trading days"),
],
tick_format="0.0%",
default_y_range=Y_RANGE,
default_x_range=X_RANGE,
show_0=True,
title=f"<b>{prettify_legend(asset)} correlation to {prettify_legend(reference_asset)}</b>",
figsize=FIGSIZE,
source="Kate Capital",
).show()United States Govt. Bond (Against US Equities)¶
for asset in [
"US_Treasuries_Agg",
"US_Govt_2YR_bond",
"US_Govt_5YR_bond",
"US_Govt_10YR_bond",
]:
plot_lines(
[
corr_30_days[asset].rename("30 trading days"),
corr_60_days[asset].rename("60 trading days"),
corr_90_days[asset].rename("90 trading days"),
],
tick_format="0.0%",
default_y_range=Y_RANGE,
default_x_range=X_RANGE,
show_0=True,
title=f"<b>{prettify_legend(asset)} correlation to {prettify_legend(reference_asset)}</b>",
figsize=FIGSIZE,
source="Kate Capital",
).show()Euro Bond Govt. (Against US Equities)¶
for asset in ["EU_Govt_2YR_bond", "EU_Govt_5YR_bond", "EU_Govt_10YR_bond"]:
plot_lines(
[
corr_30_days[asset].rename("30 trading days"),
corr_60_days[asset].rename("60 trading days"),
corr_90_days[asset].rename("90 trading days"),
],
tick_format="0.0%",
default_y_range=Y_RANGE,
default_x_range=X_RANGE,
show_0=True,
title=f"<b>{prettify_legend(asset)} correlation to {prettify_legend(reference_asset)}</b>",
figsize=FIGSIZE,
source="Kate Capital",
).show()Gold (Against US Equities)¶
for asset in ["gold_us"]:
plot_lines(
[
corr_30_days[asset].rename("30 trading days"),
corr_60_days[asset].rename("60 trading days"),
corr_90_days[asset].rename("90 trading days"),
],
tick_format="0.0%",
default_y_range=Y_RANGE,
default_x_range=X_RANGE,
show_0=True,
title=f"<b>{prettify_legend(asset).replace('US', '(US)')} correlation to {prettify_legend(reference_asset)}</b>",
figsize=FIGSIZE,
source="Kate Capital",
).show()Dollar (Against US Equities)¶
for asset in ["dollar"]:
plot_lines(
[
corr_30_days[asset].rename("30 trading days"),
corr_60_days[asset].rename("60 tradinbg days"),
corr_90_days[asset].rename("90 trading days"),
],
tick_format="0.0%",
default_y_range=Y_RANGE,
default_x_range=X_RANGE,
show_0=True,
title=f"<b>US {prettify_legend(asset)} correlation to {prettify_legend(reference_asset)}</b>",
figsize=FIGSIZE,
source="Kate Capital",
).show()Correlation of Yields Against US 10Yr¶
bb.bdh("RX1 Comdty", fields=["YLD_YTM_MID"])# todo Fix bb.bdh("RX1 Comdty", fields=["YLD_YTM_MID"], redo=True)# SOFR_SWAP_SPREAD = {
# "SOFR-2Y Swap Spread": "USSFCT02 BGN Curncy", # USOSFR2 Curncy vs CT2 Govt
# "SOFR-3Y Swap Spread": "USSFCT03 BGN Curncy", # USOSFR3 Curncy vs CT3 Govt
# "SOFR-5Y Swap Spread": "USSFCT05 BGN Curncy", # USOSFR5 Curncy vs CT5 Govt
# }
# reference_yield = "US 10Yr"
# YIELDS = dict(
# US_Govt_2YR_bond="US 2Yr",
# US_Govt_5YR_bond="US 5Yr",
# US_Govt_10YR_bond="US 10Yr",
# # EU_Govt_2YR_bond="EU 2Yr",
# # EU_Govt_5YR_bond="EU 5Yr",
# # EU_Govt_10YR_bond="EU 10Yr",
# )
# yields = {}
# for k, d in YIELDS.items():
# yields[d] = MARKET_DATA[k].xs("YLD_YTM_MID", axis=1, level=1)
# yields = pd.concat(yields, axis=1).sort_index().dropna()
# yields_30d_corr = (
# yields.rolling(30)
# .corr()
# .xs(reference_yield, level=1)
# .drop([reference_yield], axis=1)
# )
# yields_60d_corr = (
# yields.rolling(60)
# .corr()
# .xs(reference_yield, level=1)
# .drop([reference_yield], axis=1)
# )
# yields_90d_corr = (
# yields.rolling(90)
# .corr()
# .xs(reference_yield, level=1)
# .drop([reference_yield], axis=1)
# )
# for asset in ["US 2Yr", "US 5Yr"]:
# plot_lines(
# [
# yields_30d_corr[asset].rename("30 trading days"),
# yields_60d_corr[asset].rename("60 trading days"),
# yields_90d_corr[asset].rename("90 trading days"),
# ],
# tick_format="0.0%",
# default_y_range=Y_RANGE,
# default_x_range=X_RANGE,
# show_0=True,
# title=f"<b>Yields: {prettify_legend(asset)} correlation to {prettify_legend(reference_yield)}</b>",
# figsize=FIGSIZE,
# source="Kate Capital",
# ).show()
# for asset in ["EU 2Yr", "EU 5Yr", "EU 10Yr"]:
# plot_lines(
# [
# yields_30d_corr[asset].rename("30 trading days"),
# yields_60d_corr[asset].rename("60 trading days"),
# yields_90d_corr[asset].rename("90 trading days"),
# ],
# tick_format="0.0%",
# default_y_range=Y_RANGE,
# default_x_range=X_RANGE,
# show_0=True,
# title=f"<b>Yields: {prettify_legend(asset)} correlation to {prettify_legend(reference_yield)}</b>",
# figsize=FIGSIZE,
# source="Kate Capital",
# ).show()from tulip_mania.notebook_related import notebook_updated
notebook_updated()