OpenBB/openbb_platform/providers/cftc/tests/test_fx.py

from datetime import date

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

from openbb_cftc.utils.fx import (
    base_notional,
    build_forward_points,
    extract_fx_observations,
    pair_rate,
)

TRADE_DATE = date(2026, 7, 15)


@pytest.mark.parametrize(
    ("record", "expected"),
    [
        (
            {
                "Notional amount-Leg 1": "871,620",
                "Notional currency-Leg 1": "EUR",
                "Notional amount-Leg 2": "1,000,000",
                "Notional currency-Leg 2": "USD",
            },
            1.147285,
        ),
        (
            {
                "Notional amount-Leg 1": "1,000,000",
                "Notional currency-Leg 1": "USD",
                "Notional amount-Leg 2": "871,620",
                "Notional currency-Leg 2": "EUR",
            },
            1.147285,
        ),
    ],
)
def test_pair_rate_from_either_leg_order(record, expected):
    assert pair_rate(record, "EUR", "USD") == pytest.approx(expected, abs=1e-5)


@pytest.mark.parametrize(
    "record",
    [
        {"Notional amount-Leg 1": "", "Notional currency-Leg 1": "EUR"},
        {
            "Notional amount-Leg 1": "100",
            "Notional currency-Leg 1": "GBP",
            "Notional amount-Leg 2": "100",
            "Notional currency-Leg 2": "JPY",
        },
    ],
)
def test_pair_rate_returns_none_when_unusable(record):
    assert pair_rate(record, "EUR", "USD") is None


def test_base_notional_finds_either_leg():
    first = {
        "Notional amount-Leg 1": "5,000,000",
        "Notional currency-Leg 1": "EUR",
        "Notional amount-Leg 2": "1",
        "Notional currency-Leg 2": "USD",
    }
    second = {
        "Notional amount-Leg 1": "1",
        "Notional currency-Leg 1": "USD",
        "Notional amount-Leg 2": "5,000,000",
        "Notional currency-Leg 2": "EUR",
    }

    assert base_notional(first, "EUR") == 5_000_000.0
    assert base_notional(second, "EUR") == 5_000_000.0
    assert base_notional({"Notional currency-Leg 1": "JPY"}, "EUR") is None


def test_extract_rejects_an_unknown_pair(fx_records):
    with pytest.raises(OpenBBError, match="Invalid pair"):
        extract_fx_observations(fx_records, pair="EURNOK", trade_date=TRADE_DATE)


def test_extract_applies_the_notional_floor(fx_records):
    everything = extract_fx_observations(
        fx_records, pair="EURUSD", trade_date=TRADE_DATE, min_notional=0
    )
    institutional = extract_fx_observations(
        fx_records, pair="EURUSD", trade_date=TRADE_DATE, min_notional=1_000_000
    )

    assert len(institutional) < len(everything)
    assert all(o["notional"] >= 1_000_000 for o in institutional)


def test_extract_inverts_quotes_for_usd_base_pairs(fx_records):
    observations = extract_fx_observations(
        fx_records, pair="USDJPY", trade_date=TRADE_DATE
    )

    assert observations
    assert all(100 < o["rate"] < 300 for o in observations)


def test_extract_skips_unusable_rows():
    base = {
        "UPI FISN": "NA/Fwd EUR USD",
        "Action type": "NEWT",
        "Event type": "TRAD",
        "Expiration Date": "2026-08-15",
        "Notional amount-Leg 1": "5,000,000",
        "Notional currency-Leg 1": "EUR",
        "Notional amount-Leg 2": "5,700,000",
        "Notional currency-Leg 2": "USD",
    }
    rows = [
        {**base, "UPI FISN": "NA/Fwd NDF KRW USD"},
        {**base, "Action type": "CORR"},
        {**base, "Event type": "COMP"},
        {**base, "Expiration Date": ""},
        {**base, "Expiration Date": "2026-07-14"},
        {**base, "Notional amount-Leg 1": "100", "Notional amount-Leg 2": "114"},
        {**base, "Notional currency-Leg 2": "JPY"},
    ]

    assert extract_fx_observations(rows, pair="EURUSD", trade_date=TRADE_DATE) == []
    assert (
        len(extract_fx_observations([base], pair="EURUSD", trade_date=TRADE_DATE)) == 1
    )


@pytest.mark.parametrize(
    ("pair", "descending"),
    [("USDJPY", True), ("USDCAD", True), ("EURUSD", False)],
)
def test_build_forward_points_has_the_right_sign(fx_records, pair, descending):
    points = build_forward_points(fx_records, pair=pair, trade_date=TRADE_DATE)
    longest = points[-1]["forward_points"]

    assert points[0]["tenor"] == "SPOT"
    assert points[0]["forward_points"] == pytest.approx(0.0, abs=1e-9)
    assert (longest < 0) is descending


def test_build_forward_points_usdjpy_is_monotonic(fx_records):
    points = build_forward_points(fx_records, pair="USDJPY", trade_date=TRADE_DATE)
    series = [p["forward_points"] for p in points]

    assert len(series) > 3
    assert all(a >= b for a, b in zip(series, series[1:]))


def test_build_forward_points_notional_floor_fixes_contamination(fx_records):
    clean = build_forward_points(fx_records, pair="EURUSD", trade_date=TRADE_DATE)
    dirty = build_forward_points(
        fx_records, pair="EURUSD", trade_date=TRADE_DATE, min_notional=0
    )
    clean_3m = next(p["forward_points"] for p in clean if p["tenor"] == "3M")
    dirty_3m = next(p["forward_points"] for p in dirty if p["tenor"] == "3M")

    assert clean_3m < 100
    assert dirty_3m > 200


def test_build_forward_points_fields(fx_records):
    points = build_forward_points(fx_records, pair="EURUSD", trade_date=TRADE_DATE)
    row = points[1]

    assert row["date"] == TRADE_DATE
    assert row["pair"] == "EURUSD"
    assert row["num_trades"] >= 2
    assert row["min_rate"] <= row["forward_rate"] <= row["max_rate"]
    assert row["total_notional"] > 0
    assert row["spot_rate"] == points[0]["spot_rate"]


def test_build_forward_points_min_trades(fx_records):
    shallow = build_forward_points(fx_records, pair="EURUSD", trade_date=TRADE_DATE)
    deep = build_forward_points(
        fx_records, pair="EURUSD", trade_date=TRADE_DATE, min_trades=10
    )

    assert len(deep) < len(shallow)
    assert all(p["num_trades"] >= 10 for p in deep)


def test_build_forward_points_raises_without_trades(fx_records):
    with pytest.raises(OpenBBError, match="notional"):
        build_forward_points(
            fx_records, pair="EURUSD", trade_date=TRADE_DATE, min_notional=1e15
        )


def test_build_forward_points_raises_without_spot(fx_records):
    with pytest.raises(OpenBBError, match="spot cannot be established"):
        build_forward_points(fx_records, pair="EURUSD", trade_date=date(2020, 1, 1))


def test_build_forward_points_raises_when_min_trades_empties_it(fx_records):
    with pytest.raises(OpenBBError, match="at least"):
        build_forward_points(
            fx_records, pair="EURUSD", trade_date=TRADE_DATE, min_trades=10_000
        )