|
|
import traceback |
|
|
from datetime import datetime, date |
|
|
from io import BytesIO |
|
|
from typing import TypedDict, Literal, List, Optional, Annotated |
|
|
from zipfile import ZipFile |
|
|
from xml.etree import ElementTree |
|
|
import requests |
|
|
import os |
|
|
|
|
|
from langchain_core.tools import tool |
|
|
|
|
|
supported_currencies = ['USD', 'JPY', 'EUR', 'GBP', 'CHF', 'AUD'] |
|
|
|
|
|
class CurvePoint(TypedDict): |
|
|
""" |
|
|
Interest rate for a tenor |
|
|
""" |
|
|
tenor: Annotated[str, 'tenor of the interest rate. (examples: tenor 1M is One month interest rate, tenor 1Y is One year interest rate)'] |
|
|
maturity_date: Annotated[date, 'date that the tenor expires (matured)'] |
|
|
rate: Annotated[float, 'interest rate of the tenor'] |
|
|
|
|
|
class IRCurve(TypedDict): |
|
|
""" |
|
|
Interest rate curve holds interest rates on the effective date for the currency |
|
|
""" |
|
|
effective_date: Annotated[date, 'the date that the interest rates were observed'] |
|
|
currency: Annotated[Literal['USD', 'JPY', 'EUR', 'GBP', 'CHF', 'AUD'], 'the currency of the interest rate'] |
|
|
curve_points: Annotated[List[CurvePoint], 'interest rates for the tenors (examples: tenor 1M is One month interest rate, tenor 1Y is One year interest rate)'] |
|
|
|
|
|
def _build_ir(xml: str, currency: Literal['USD', 'JPY', 'EUR', 'GBP', 'CHF', 'AUD']) -> IRCurve: |
|
|
root = ElementTree.fromstring(xml) |
|
|
effective_date_str = root.find('interestRateCurve').find('ois').find('snaptime').text |
|
|
effective_date = datetime.strptime(effective_date_str, '%Y-%m-%dT%H:%M:%S').date() |
|
|
curve_points = [] |
|
|
for curve_point in root.find('interestRateCurve').find('ois').findall('curvepoint'): |
|
|
tenor = curve_point.find('tenor').text |
|
|
maturity_date_str = curve_point.find('maturitydate').text |
|
|
maturity_date = datetime.strptime(maturity_date_str, '%Y-%m-%d').date() |
|
|
rate = float(curve_point.find('parrate').text) |
|
|
curve_points.append(CurvePoint(tenor=tenor, maturity_date=maturity_date, rate=rate)) |
|
|
return IRCurve(effective_date=effective_date, currency=currency, curve_points=curve_points) |
|
|
|
|
|
@tool |
|
|
def load_interest_rate_curve(date_obj: date, currency: Literal['USD', 'JPY', 'EUR', 'GBP', 'CHF', 'AUD']) \ |
|
|
-> Annotated[Optional[IRCurve], 'Interest rate object in a Python dictionary']: |
|
|
""" |
|
|
Loads interest rate curve for the given date and given currency |
|
|
Returns Interest rate object (IRCurve, which is a Python dictionary) if the interest rates can be found |
|
|
Returns None if the interest rates cannot be found |
|
|
""" |
|
|
try: |
|
|
url = f'https://rfr.ihsmarkit.com/InterestRates_{currency}_{date_obj.strftime('%Y%m%d')}.zip?email={os.environ['ir_curve_email']}' |
|
|
req = requests.get(url) |
|
|
zip = ZipFile(BytesIO(req.content)) |
|
|
file = f'InterestRates_{currency}_{date_obj.strftime('%Y%m%d')}.xml' |
|
|
lines = zip.open(file).readlines() |
|
|
xml = "".join([line.decode('utf-8') for line in lines]) |
|
|
return _build_ir(xml, currency) |
|
|
except Exception as e: |
|
|
print(traceback.format_exc()) |
|
|
return None |
|
|
|
|
|
|
|
|
|
|
|
|
