Fixed income: versioned curve marks, restatements, carry & rolldown

curves = cu.make_yield_curves(days=250)
clean_df = curves.to_pandas()
print(f"{len(clean_df):,} rows x {clean_df.shape[1]} columns, "
      f"{clean_df['ts'].nunique()} mark dates")
clean_df.head()

mark_dates = clean_df["ts"].unique()
feed_df = clean_df.copy()
bad_date = pd.Timestamp(mark_dates[120])
bad_mask = (feed_df["ts"] == bad_date) & (feed_df["tenor_years"] == 10.0)
feed_df.loc[bad_mask, "yield_pct"] += 0.25
print(f"vendor feed: {len(feed_df):,} rows, bad 10y mark on {bad_date.date()}")

schema = pa.schema(
    [
        pa.field("ts", pa.timestamp("us", tz="UTC"), nullable=False),
        pa.field("tenor_years", pa.float64()),
        pa.field("yield_pct", pa.float64()),
    ]
)
db.create_table("curves", schema, time_column="ts", sort_key=["ts", "tenor_years"])

for day_ts, day_rows in feed_df.groupby("ts"):
    db.append("curves", pa.Table.from_pandas(day_rows, schema=schema, preserve_index=False),
              note=f"EOD marks {day_ts.date()}")

[
    {k: v[k] for k in ("sequence", "op", "rows", "note") if k in v}
    for v in db.versions("curves")[-3:]
]

# Long-to-wide by conditional aggregation. As a dict of tenors it is one
# definition instead of five near-identical CASE arms - add a tenor by adding
# an entry.
KEY_TENORS = {"y3m": 0.25, "y2": 2, "y5": 5, "y10": 10, "y30": 30}

key_rates = (
    db.table("curves")
    .group_by("ts")
    .agg(**{
        name: when(col("tenor_years") == t).then(col("yield_pct")).max()
        for name, t in KEY_TENORS.items()
    })
    .sort("ts")
    .to_pandas()
)
key_rates["s2s10"] = key_rates["y10"] - key_rates["y2"]
key_rates["fly_2_5_10"] = 2 * key_rates["y5"] - key_rates["y2"] - key_rates["y10"]
key_rates["slope_proxy"] = key_rates["y10"] - key_rates["y3m"]
num_cols = key_rates.columns.drop("ts")
key_rates.tail(3).round(dict.fromkeys(num_cols, 4))

fig, ax = plt.subplots(figsize=(10, 4))
ax.plot(key_rates["ts"], 100 * key_rates["s2s10"], lw=1.0, label="2s10s slope")
ax.plot(key_rates["ts"], 100 * key_rates["fly_2_5_10"], lw=1.0, label="2s5s10s fly")
ax.axvline(bad_date, color="0.6", ls=":", lw=1)
ax.annotate("suspicious print", xy=(bad_date, 100 * key_rates.loc[key_rates["ts"] == bad_date, "s2s10"].iloc[0]),
            xytext=(10, 15), textcoords="offset points", fontsize=8)
ax.set_title("Curve shape time series - note the one-day 2s10s spike")
ax.set_xlabel("mark date")
ax.set_ylabel("bp")
ax.legend(fontsize=8)
fig.tight_layout()

PREV_Y = sql_expr("lag(yield_pct)").over(partition_by="tenor_years", order_by="ts")

(
    db.table("curves")
    .with_columns(d1=col("yield_pct") - PREV_Y)
    .filter(col("d1").is_not_null())
    .select(
        "ts", "tenor_years",
        yield_pct=col("yield_pct").round(4),
        chg_bp=(100 * col("d1")).round(1),
        abs_move=col("d1").abs(),
    )
    .sort("abs_move", descending=True)
    .limit(4)
    .select("ts", "tenor_years", "yield_pct", "chg_bp")
    .to_pandas()
)

day_us = int(bad_date.value // 1000)
corrected = clean_df[clean_df["ts"] == bad_date]  # the true marks for that day

plan = db.plan_replace_range(
    "curves", day_us, day_us + 1,
    data=pa.Table.from_pandas(corrected, schema=schema, preserve_index=False),
    note=f"restate {bad_date.date()}: 10y stale contributor, -25bp",
)
print("summary:", plan.summary)
print("\nbefore (10y row):")
print(plan.before_sample.to_pandas().query("tenor_years == 10"))
print("\nafter (10y row):")
print(plan.after_sample.to_pandas().query("tenor_years == 10"))

commit = plan.apply()
print(f"applied as v{commit['sequence']} ({commit['op']})")
[
    {k: v[k] for k in ("sequence", "op", "rows", "note") if k in v}
    for v in db.versions("curves")[-2:]
]

v_pre_restate = db.versions("curves")[-2]["sequence"]  # head just before the restatement
was_y, now_y = col("yield_pct", relation="l"), col("yield_pct", relation="r")

(
    db.table("curves", version=v_pre_restate)
    .join(db.table("curves"), on=["ts", "tenor_years"])
    .filter(
        col("ts", relation="l") == pd.Timestamp(day_us, unit="us", tz="UTC").isoformat(),
        col("tenor_years", relation="l").is_in([2, 5, 10, 30]),
    )
    .select(
        tenor_years=col("tenor_years", relation="l"),
        as_marked_that_night=was_y.round(4),
        restated=now_y.round(4),
        diff_bp=(100 * (now_y - was_y)).round(1),
    )
    .sort("tenor_years")
    .to_pandas()
)

# No verb for a scalar subquery: resolve the latest mark date, then filter.
latest_ts = db.table("curves").select(m=col("ts").max()).to_pandas()["m"][0]

latest = (
    db.table("curves")
    .filter(col("ts") == latest_ts.isoformat())
    .select("tenor_years", "yield_pct")
    .sort("tenor_years")
    .to_pandas()
)

tenors, ylds = latest["tenor_years"].values, latest["yield_pct"].values
y = lambda t: float(np.interp(t, tenors, ylds))

rolldown_bp = 100 * (y(5) - y(4))
carry_bp = 100 * (y(5) - y(0.25))
dur_4y = 3.7  # rough modified duration of the aged position
print(f"5y yield {y(5):.3f}%, 4y yield {y(4):.3f}%, 3m funding {y(0.25):.3f}%")
print(f"rolldown (1y horizon): {rolldown_bp:+.1f} bp  (~{rolldown_bp * dur_4y / 100:+.2f}% price)")
print(f"carry over funding:    {carry_bp:+.1f} bp")
print(f"carry + roll:          {carry_bp + rolldown_bp:+.1f} bp per year, curve unchanged")

fig, ax = plt.subplots(figsize=(9, 4))
show_dates = mark_dates[:: len(mark_dates) // 5][:6]
colors = plt.cm.viridis(np.linspace(0.15, 0.85, len(show_dates)))
for d, c in zip(show_dates, colors):
    g = clean_df[clean_df["ts"] == d]
    ax.plot(g["tenor_years"], g["yield_pct"], marker="o", ms=3, lw=1.1,
            color=c, label=str(pd.Timestamp(d).date()))
ax.set_xscale("log")
ax.set_xticks([0.25, 1, 2, 5, 10, 30])
ax.set_xticklabels(["3m", "1y", "2y", "5y", "10y", "30y"])
ax.set_title("Curve snapshots across the year")
ax.set_xlabel("tenor")
ax.set_ylabel("yield (%)")
ax.legend(fontsize=8)
fig.tight_layout()