OpenBB/openbb_platform/providers/cftc/openbb_cftc/utils/cfets.py

"""CFETS/NIFC daily bulletin client for CNY interest rate swaps."""

from datetime import date as dateType

CFETS_IRS_BULLETIN_URL = "https://www.chinamoney.com.cn/ags/ms/cm-u-dlrp/IrsDlyBltn"
MAX_BULLETIN_LOOKBACK_DAYS = 10

IRS_LEG_CONVENTIONS: dict[str, dict] = {
    "FR007": {"day_count": "ACT/365F", "basis": 365.0, "payment_period_days": 91},
    "SHIBOR3M": {"day_count": "ACT/360", "basis": 360.0, "payment_period_days": 91},
    "SHIBORO/N": {"day_count": "ACT/360", "basis": 360.0, "payment_period_days": 91},
    "LPR1Y": {"day_count": "ACT/365F", "basis": 365.0, "payment_period_days": 91},
    "LPR5Y": {"day_count": "ACT/365F", "basis": 365.0, "payment_period_days": 91},
}


def mentions_cny(record: dict) -> bool:
    """Whether a print names CNY in its FISN or underlier."""
    tokens = (record.get("UPI Underlier Name") or "").split() + (
        record.get("UPI FISN") or ""
    ).split()

    return "CNY" in tokens


def normalize_reference(name: str) -> str:
    """Collapse a reference-rate name to its case- and space-insensitive key."""
    return (name or "").replace(" ", "").upper()


def _today_cct() -> dateType:
    """Return the current trading date in China Coordinated Time (UTC+8)."""
    from datetime import datetime, timedelta, timezone

    return (datetime.now(timezone.utc) + timedelta(hours=8)).date()


async def _download_bulletin(search_date: str) -> dict:
    """POST the daily bulletin service for one search date."""
    import json
    import ssl

    import aiohttp
    import certifi
    from openbb_core.app.model.abstract.error import OpenBBError

    context = ssl.create_default_context(cafile=certifi.where())
    context.verify_flags &= ~ssl.VERIFY_X509_STRICT
    headers = {
        "User-Agent": (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
            " (KHTML, like Gecko) Chrome/126.0 Safari/537.36"
        ),
        "Referer": "https://www.chinamoney.com.cn/english/mdtrptdbl/",
        "Content-Type": "application/x-www-form-urlencoded",
    }

    try:
        async with (
            aiohttp.ClientSession(
                headers=headers, connector=aiohttp.TCPConnector(ssl=context)
            ) as session,
            session.post(
                CFETS_IRS_BULLETIN_URL, data=f"lang=en&searchDate={search_date}"
            ) as response,
        ):
            response.raise_for_status()
            payload = json.loads(await response.text())
    except OpenBBError:
        raise
    except Exception as exc:
        raise OpenBBError(f"Failed to fetch the IRS bulletin -> {exc}") from exc

    if (payload.get("head") or {}).get("rep_code") != "200" or not payload.get(
        "records"
    ):
        raise OpenBBError(
            f"Unexpected IRS bulletin response from CFETS for {search_date}."
        )

    return payload


async def fetch_irs_bulletin(
    search_date: dateType | None, use_cache: bool = True
) -> dict:
    """Return one day's IRS bulletin, served from the document store when closed."""
    from datetime import timedelta

    from openbb_core.app.model.abstract.error import OpenBBError

    from openbb_cftc.utils import store

    today = _today_cct()
    target = search_date or today
    key = f"cfets-irs-bulletin:{target.isoformat()}"

    if use_cache and search_date is not None:
        cached = store.get_document(key)

        if cached is not None:
            return cached

    payload: dict = {}

    for offset in range(MAX_BULLETIN_LOOKBACK_DAYS):
        day = target - timedelta(days=offset)

        try:
            payload = await _download_bulletin(day.isoformat())
            break
        except OpenBBError:
            if offset == MAX_BULLETIN_LOOKBACK_DAYS - 1:
                raise
    last_date = (payload.get("data") or {}).get("lastDate") or ""

    if use_cache and last_date and last_date < today.isoformat():
        store.put_document(f"cfets-irs-bulletin:{last_date}", payload)

        if search_date is not None and target.isoformat() != last_date:
            store.put_document(key, payload)

    return payload


async def cny_curve_records(day: dateType, use_cache: bool = True) -> list[dict]:
    """Return the FR007 par curve as tape-shaped CNY IRS records, for curve building."""
    from datetime import timedelta

    from openbb_core.app.model.abstract.error import OpenBBError
    from openbb_core.provider.utils.errors import EmptyDataError

    try:
        payload = await fetch_irs_bulletin(day, use_cache=use_cache)
        nodes = bulletin_nodes(payload, "FR007")
    except (OpenBBError, EmptyDataError):
        return []

    records: list[dict] = []

    for node in nodes:
        volume = node["volume"]
        notional = volume * 1_000_000.0 if volume else 1_000_000.0
        records.append(
            {
                "UPI FISN": "NA/Swap Fxd Flt CNY",
                "UPI Underlier Name": "CNY-CNREPOFIX=CFXS-Reuters",
                "Action type": "NEWT",
                "Event type": "TRAD",
                "Notional currency-Leg 1": "CNY",
                "Notional currency-Leg 2": "CNY",
                "Fixed rate-Leg 1": f"{node['par_rate']:.8f}",
                "Effective Date": day.isoformat(),
                "Expiration Date": (
                    day + timedelta(days=node["tenor_days"])
                ).isoformat(),
                "Notional amount-Leg 1": f"{notional:,.0f}",
                "Fixed rate day count convention-leg 1": "A005",
                "Fixed rate payment frequency period-Leg 1": "MNTH",
                "Fixed rate payment frequency period multiplier-Leg 1": "3",
                "Cleared": "Y",
                "Dissemination Timestamp": f"{day.isoformat()}T00:00:00Z",
            }
        )

    return records


def bulletin_reference_names(payload: dict) -> list[str]:
    """Return the reference rates the bulletin published, in its own order."""
    return [
        record.get("refIntrstRateNmEn") or record.get("refIntrstRateNm") or ""
        for record in payload.get("records") or []
    ]


def bulletin_nodes(payload: dict, reference_rate: str) -> list[dict]:
    """Return one reference rate's traded par nodes, sorted by tenor."""
    from openbb_core.provider.utils.errors import EmptyDataError

    wanted = normalize_reference(reference_rate)
    conventions = IRS_LEG_CONVENTIONS.get(wanted, {})

    for record in payload.get("records") or []:
        name = record.get("refIntrstRateNmEn") or record.get("refIntrstRateNm") or ""

        if normalize_reference(name) != wanted:
            continue

        nodes: list[dict] = []

        for row in record.get("rateList") or []:
            try:
                rate = float(row.get("wghtdAvgFxngRate"))
                days = int(row.get("sortNo"))
            except (TypeError, ValueError):
                continue

            try:
                volume = float(row.get("trdVol"))
            except (TypeError, ValueError):
                volume = None

            nodes.append(
                {
                    "tenor": (row.get("prd") or "").strip(),
                    "tenor_days": days,
                    "tenor_years": round(days / 365.0, 6),
                    "par_rate": rate / 100.0,
                    "volume": volume,
                    "is_fixing": False,
                    "day_count": conventions.get("day_count"),
                    "day_count_basis": conventions.get("basis"),
                    "payment_period_days": conventions.get("payment_period_days"),
                }
            )

        nodes.sort(key=lambda node: node["tenor_days"])

        return nodes

    available = ", ".join(name for name in bulletin_reference_names(payload) if name)

    raise EmptyDataError(
        f"'{reference_rate}' was not published in the bulletin for"
        f" {(payload.get('data') or {}).get('lastDate')}."
        + (f" Published reference rates: {available}." if available else "")
    )