import pandas as pd
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from IPython.display import Markdown
from tulip.data.bloomberg import BloombergClient as bb
from xbbg import blp
from tqdm import tqdm
from tulip.data.bloomberg.collections import (
POLICY_RATES,
WIRP,
CURVES,
COUNTRY_CURRENCY,
)# Charting parameters
years_limit = 10
# Matplotlib style settings
plt.rcParams["figure.figsize"] = (12, 5)
plt.rcParams["axes.grid"] = True
plt.rcParams["grid.alpha"] = 0.3
# Default colors for key tenors
TENOR_COLORS = {
"2Y": "#1f77b4",
"5Y": "#ff7f0e",
"10Y": "#2ca02c",
"30Y": "#d62728",
"YC (10Y-2Y)": "black",
}
def plot_yields_matplotlib(df, title, years_limit=None, highlight_tenors=None):
"""
Plot yields using matplotlib.
Parameters:
-----------
df : DataFrame
DataFrame containing yield data with dates as index
title : str
Chart title
years_limit : int, optional
Number of years to limit the x-axis to
highlight_tenors : list, optional
List of tenors to highlight (others will be dimmed)
"""
if highlight_tenors is None:
highlight_tenors = ["2Y", "5Y", "10Y", "30Y", "YC (10Y-2Y)"]
fig, ax = plt.subplots()
# Filter data by years if specified
plot_df = df.copy()
if years_limit:
cutoff_date = pd.Timestamp.now() - pd.DateOffset(years=years_limit)
plot_df = plot_df[plot_df.index >= cutoff_date]
# Plot each column
for col in plot_df.columns:
is_highlighted = col in highlight_tenors
color = TENOR_COLORS.get(col, None)
alpha = 1.0 if is_highlighted else 0.3
linewidth = 2 if col == "YC (10Y-2Y)" else 1.5 if is_highlighted else 1
(line,) = ax.plot(
plot_df.index,
plot_df[col],
label=col,
color=color,
alpha=alpha,
linewidth=linewidth,
)
# Fill for yield curve
if col == "YC (10Y-2Y)":
ax.fill_between(plot_df.index, 0, plot_df[col], alpha=0.15, color="gray")
ax.axhline(y=0, color="black", linewidth=0.5, linestyle="-")
ax.set_title(title, fontweight="bold", fontsize=12)
ax.set_ylabel("Yield (%)")
ax.yaxis.set_major_formatter(plt.FuncFormatter(lambda x, p: f"{x:.1f}%"))
ax.xaxis.set_major_formatter(mdates.DateFormatter("%Y"))
ax.xaxis.set_major_locator(mdates.YearLocator())
ax.legend(loc="upper left", fontsize=8, ncol=3)
plt.tight_layout()
plt.show()
def plot_country_yields(country, nominal_df, real_df, bei_df, years_limit=None):
"""
Plot nominal yields, real yields, and breakeven inflation for a given country.
"""
plot_yields_matplotlib(nominal_df, f"{country} Nominal Yields", years_limit)
plot_yields_matplotlib(real_df, f"{country} Real Yields", years_limit)
plot_yields_matplotlib(bei_df, f"{country} Breakeven Inflation (BEI)", years_limit)
def analyze_yield_metrics(df, column_name=None, rounding_precision=2):
"""
Analyze yield metrics for a specific column or all columns in the dataframe.
Args:
df: DataFrame containing yield data with dates as index
column_name: Optional specific column to analyze. If None, analyzes all columns.
rounding_precision: Number of decimal places to round the results to
Returns:
DataFrame with analysis results
"""
import pandas as pd
import numpy as np
columns_to_analyze = [column_name] if column_name else df.columns
results = []
for col in columns_to_analyze:
if col not in df.columns:
continue
series = df[col].dropna()
if len(series) == 0:
continue
latest_value = series.iloc[-1]
latest_date = series.index[-1]
# Calculate averages for different time periods
full_avg = series.mean()
since_2009_avg = series[series.index >= "2009-01-01"].mean()
since_2022_avg = series[series.index >= "2022-01-01"].mean()
# Calculate percentiles
full_percentile = percentile_rank(series, latest_value)
since_2009_percentile = percentile_rank(
series[series.index >= "2009-01-01"], latest_value
)
since_2022_percentile = percentile_rank(
series[series.index >= "2022-01-01"], latest_value
)
results.append(
{
"Metric": col,
"Latest Value": round(latest_value, rounding_precision),
"Latest Date": latest_date,
"Full History Avg": round(full_avg, rounding_precision),
"Since 2009 Avg": round(since_2009_avg, rounding_precision),
"Since 2022 Avg": round(since_2022_avg, rounding_precision),
"Full History Percentile": full_percentile,
"Since 2009 Percentile": since_2009_percentile,
"Since 2022 Percentile": since_2022_percentile,
"vs Full Avg": round(latest_value - full_avg, rounding_precision),
"vs 2009 Avg": round(latest_value - since_2009_avg, rounding_precision),
"vs 2022 Avg": round(latest_value - since_2022_avg, rounding_precision),
}
)
return pd.DataFrame(results)
def percentile_rank(series, value):
"""Calculate the percentile rank of a value within a series"""
if len(series) == 0:
return None
return round(100 * (series < value).mean())
def analyze_country_yields(nominal_df, real_df, bei_df):
# Analyze each yield type
nominal_analysis = analyze_yield_metrics(nominal_df)
nominal_analysis["Type"] = "Nominal"
real_analysis = analyze_yield_metrics(real_df)
real_analysis["Type"] = "Real"
bei_analysis = analyze_yield_metrics(bei_df)
bei_analysis["Type"] = "BEI (Inflation)"
# Concatenate all analyses
combined_analysis = pd.concat([nominal_analysis, real_analysis, bei_analysis])
# Create MultiIndex with Type and Metric
combined_analysis = combined_analysis.set_index(["Type", "Metric"])
return combined_analysisYields Through Time¶
# todo: Expressed in local currency for now
countries = [
"USD", # United States
"EUR", # Euro
"GBP", # United Kingdom
"JPY", # Japan
"SEK", # Sweden
"AUD", # Australia
"CAD", # Canada
# "KRW", # South Korea
# "CNY", # China
# "INR", # India
"BRL", # Brazil
"MXN", # Mexico
"ZAR", # South Africa
# "RUB", # Russia
# "TRY", # Turkey
# "INR", # India
]
today_str = pd.Timestamp("today").strftime("%Y%m%d")
week_ago_str = (pd.Timestamp("today") - pd.offsets.BusinessDay(n=5)).strftime("%Y%m%d")
wirp_period = (pd.Timestamp("today") + pd.offsets.YearEnd()).strftime("%b%Y").upper()
# todo: we must save this to silver with some automatism. Because it takes a long time to run
yield_data = {}
for country in tqdm(countries, leave=False):
yield_data[country] = {}
yield_data[country]["nominal"] = bb.get_govt_yields(
currency=country, yield_type="nominal"
)
yield_data[country]["real"] = bb.get_govt_yields(
currency=country, yield_type="real"
)
yield_data[country]["bei"] = bb.get_govt_yields(currency=country, yield_type="bei")
from IPython.display import clear_output
clear_output(wait=True)
United States¶
country = "USD"
nominal_df = yield_data[country]["nominal"].droplevel(
["currency", "yield_type"], axis=1
)
nominal_df["YC (10Y-2Y)"] = nominal_df["10Y"].sub(nominal_df["2Y"])
real_df = yield_data[country]["real"].droplevel(["currency", "yield_type"], axis=1)
bei_df = yield_data[country]["bei"].droplevel(["currency", "yield_type"], axis=1)
plot_country_yields(country, nominal_df, real_df, bei_df, years_limit=years_limit)
combined_analysis = analyze_country_yields(nominal_df, real_df, bei_df)
display(combined_analysis)Eurozone¶
country = "EUR"
nominal_df = yield_data[country]["nominal"].droplevel(
["currency", "yield_type"], axis=1
)
nominal_df["YC (10Y-2Y)"] = nominal_df["10Y"].sub(nominal_df["2Y"])
real_df = yield_data[country]["real"].droplevel(["currency", "yield_type"], axis=1)
bei_df = yield_data[country]["bei"].droplevel(["currency", "yield_type"], axis=1)
plot_country_yields(country, nominal_df, real_df, bei_df, years_limit=years_limit)
combined_analysis = analyze_country_yields(nominal_df, real_df, bei_df)
display(combined_analysis)
Japan¶
country = "JPY"
nominal_df = yield_data[country]["nominal"].droplevel(
["currency", "yield_type"], axis=1
)
nominal_df["YC (10Y-2Y)"] = nominal_df["10Y"].sub(nominal_df["2Y"])
real_df = yield_data[country]["real"].droplevel(["currency", "yield_type"], axis=1)
bei_df = yield_data[country]["bei"].droplevel(["currency", "yield_type"], axis=1)
plot_country_yields(country, nominal_df, real_df, bei_df, years_limit=years_limit)
combined_analysis = analyze_country_yields(nominal_df, real_df, bei_df)
display(combined_analysis)United Kingdom¶
country = "GBP"
nominal_df = yield_data[country]["nominal"].droplevel(
["currency", "yield_type"], axis=1
)
nominal_df["YC (10Y-2Y)"] = nominal_df["10Y"].sub(nominal_df["2Y"])
real_df = yield_data[country]["real"].droplevel(["currency", "yield_type"], axis=1)
bei_df = yield_data[country]["bei"].droplevel(["currency", "yield_type"], axis=1)
plot_country_yields(country, nominal_df, real_df, bei_df, years_limit=years_limit)
combined_analysis = analyze_country_yields(nominal_df, real_df, bei_df)
display(combined_analysis)Canada¶
country = "CAD"
nominal_df = yield_data[country]["nominal"].droplevel(
["currency", "yield_type"], axis=1
)
nominal_df["YC (10Y-2Y)"] = nominal_df["10Y"].sub(nominal_df["2Y"])
real_df = yield_data[country]["real"].droplevel(["currency", "yield_type"], axis=1)
bei_df = yield_data[country]["bei"].droplevel(["currency", "yield_type"], axis=1)
plot_country_yields(country, nominal_df, real_df, bei_df, years_limit=years_limit)
combined_analysis = analyze_country_yields(nominal_df, real_df, bei_df)
display(combined_analysis)Australia¶
country = "AUD"
nominal_df = yield_data[country]["nominal"].droplevel(
["currency", "yield_type"], axis=1
)
nominal_df["YC (10Y-2Y)"] = nominal_df["10Y"].sub(nominal_df["2Y"])
real_df = yield_data[country]["real"].droplevel(["currency", "yield_type"], axis=1)
bei_df = yield_data[country]["bei"].droplevel(["currency", "yield_type"], axis=1)
plot_country_yields(country, nominal_df, real_df, bei_df, years_limit=years_limit)
combined_analysis = analyze_country_yields(nominal_df, real_df, bei_df)
display(combined_analysis)Mexico¶
country = "MXN"
nominal_df = yield_data[country]["nominal"].droplevel(
["currency", "yield_type"], axis=1
)
nominal_df["YC (10Y-2Y)"] = nominal_df["10Y"].sub(nominal_df["2Y"])
real_df = yield_data[country]["real"].droplevel(["currency", "yield_type"], axis=1)
bei_df = yield_data[country]["bei"].droplevel(["currency", "yield_type"], axis=1)
plot_country_yields(country, nominal_df, real_df, bei_df, years_limit=years_limit)
combined_analysis = analyze_country_yields(nominal_df, real_df, bei_df)
display(combined_analysis)South Africa¶
country = "ZAR"
nominal_df = yield_data[country]["nominal"].droplevel(
["currency", "yield_type"], axis=1
)
nominal_df["YC (10Y-2Y)"] = nominal_df["10Y"].sub(nominal_df["2Y"])
real_df = yield_data[country]["real"].droplevel(["currency", "yield_type"], axis=1)
bei_df = yield_data[country]["bei"].droplevel(["currency", "yield_type"], axis=1)
plot_country_yields(country, nominal_df, real_df, bei_df, years_limit=years_limit)
combined_analysis = analyze_country_yields(nominal_df, real_df, bei_df)
display(combined_analysis)from tulip_mania.notebook_related import notebook_updated
notebook_updated()