Curva SOFR

import qcfinancial as qcf
import pandas as pd
import aux_functions as aux

data = pd.read_excel("./input/20240621_sofr_data.xlsx")

data.style.format({'rate':'{:.4%}'})

# Debe coincidir con la fecha de los datos
trade_date = qcf.QCDate(21, 6, 2024)

# Convención de las tasas de las patas fijas
yf = qcf.QCAct360()
wf = qcf.QCLinearWf()

# Los parámetros se organizan en un dict.
common_params = {
    "rec_pay": qcf.RecPay.RECEIVE,
    "start_date": qcf.QCDate(25, 6, 2024),
    "bus_adj_rule": qcf.BusyAdjRules.MODFOLLOW,
    "settlement_stub_period": qcf.StubPeriod.SHORTFRONT,
    "settlement_calendar": qcf.BusinessCalendar(trade_date, 50),
    "settlement_lag": 0,  # Se impone = 0
    "initial_notional": 1_000_000,
    "amort_is_cashflow": True,
    "notional_currency": qcf.QCUSD(),
    "is_bond": False,
    "sett_lag_behaviour": qcf.SettLagBehaviour.DONT_MOVE,
}

for p in qcf.LegFactory.build_bullet_fixed_rate_leg.__doc__.split(','):
    print(p)

# Aquí se almacenarán los resultados
fixed_rate_legs = []

# Se recorre el DataFrame con la data
for t in data.itertuples():
    # Madurez del contrato
    tenor = qcf.Tenor(t.tenor)
    
    # Se calcula el número de meses de la madurez
    months = tenor.get_months() + 12 * tenor.get_years()

    # Se calcula la fecha final del swap sin aplicar todavía ajustes de calendario
    if (days:=tenor.get_days()) > 0:
        end_date = common_params["start_date"].add_days(days)
    else:
        end_date = common_params["start_date"].add_months(months)

    # Se define un dict con los parámetros propios de cada contrato
    other_params = {
        "end_date": end_date,
        "settlement_periodicity": qcf.Tenor(t.pay_freq),
        "interest_rate": qcf.QCInterestRate(t.rate, yf, wf),
    }

    # Se construye y almacena la pata fija correspondiente
    fixed_rate_legs.append(
        qcf.LegFactory.build_bullet_fixed_rate_leg(
            **(common_params | other_params),
        )
    )

aux.leg_as_dataframe(fixed_rate_legs[0]).style.format(aux.format_dict)

aux.leg_as_dataframe(fixed_rate_legs[14]).style.format(aux.format_dict)

# Se define los vectores de plazos y tasas
plazos = qcf.long_vec()
tasas = qcf.double_vec()

# Para rellenarlos se utiliza la información contenida 
# en las patas fijas.
for leg in fixed_rate_legs:
    # Número de cupones de la pata
    num_cup = leg.size()

    # Último cashflow de la pata
    cashflow = leg.get_cashflow_at(num_cup - 1)

    # Se calcula el número de días desde start_date 
    # hasta la última settlement_date
    plazo = common_params["start_date"].day_diff(cashflow.get_settlement_date())
    plazos.append(plazo)

    # Se obtiene el valor de la tasa fija
    tasa = cashflow.get_rate().get_value()
    tasas.append(tasa)

# Con la información anterior, se termina de construir la curva
curva = qcf.QCCurve(plazos, tasas)
interpolator = qcf.QCLinearInterpolator(curva)
initial_zcc = qcf.ZeroCouponCurve(
    interpolator, 
    rate:=(qcf.QCInterestRate(
        0.0, 
        qcf.QCAct365(), 
        qcf.QCContinousWf()
    ))
)

pv = qcf.PresentValue()

# Se resuelve la ecuación:
# VP(pata_fija(i), z1,...,z(i),...,zN) - nocional = 0, para todo i
for i, leg in enumerate(fixed_rate_legs):
    
    # Se define la función objetivo
    def obj(zcc):
        # VP - nocional
        return pv.pv(common_params["start_date"], leg, zcc) - common_params["initial_notional"]
    
    # Aquí comienza la resolución
    error = 1_000
    epsilon = .00001
    x = initial_zcc.get_rate_at(i)  # Valor inicial para Newton-Raphson
    new_zcc = initial_zcc  # En new_zcc se almacena el resultado
    
    # Se aplica Newton-Raphson
    while error > epsilon:
        x = x - obj(new_zcc) / pv.get_derivatives()[i]  # La derivada del VP se calcula al momento de valorizar 
        tasas[i] = x
        # Se reconstruye la curva con el valor de la iteración
        curva = qcf.QCCurve(plazos, tasas)
        interpolator = qcf.QCLinearInterpolator(curva)
        new_zcc = qcf.ZeroCouponCurve(
            interpolator, 
            rate,
        )
        # Se calcula el nuevo error
        error = abs(obj(new_zcc))

check = []
for i, leg in enumerate(fixed_rate_legs):
    check.append({
        "leg_number": i, 
        "present_value": pv.pv(common_params['start_date'], leg, new_zcc),
    })
df_check = pd.DataFrame(check)
df_check.style.format({"present_value": "{:,.4f}"})

df_curva = pd.concat([pd.DataFrame(plazos), pd.DataFrame(tasas)], axis=1)
df_curva.columns = ['plazo', 'tasa']
df_curva.style.format({'tasa':'{:.4%}'})