File size: 3,120 Bytes
15e8f83 ec50e84 15e8f83 591f5c9 15e8f83 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 | 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
# if __name__ == '__main__':
# print(load_interest_rate_curve.invoke(date(2025, 1, 24), 'USD')))
|