From 9b9a14d8a4bf2d9595e724eb462835a169c6f076 Mon Sep 17 00:00:00 2001 From: artcashin <64659740+artcashin@users.noreply.github.com> Date: Fri, 9 Oct 2026 15:52:19 -0400 Subject: [PATCH 1/2] perf(cftc): price FX prints from their peers without re-extracting the pair fx_market prices each FX print from the pair's other prints (leave-one-out) and re-extracted every peer record, twice, for each print; the swap_trades forex priceability gate does that for every print of the day, so a forex swap_trades call was quadratic in a pair's prints: over an hour of CPU on a day of ~43k FX prints, holding the event loop the whole time. option_vol likewise re-extracted and re-implied every vanilla option of the pair for each option print. fx_market now prepares each pair once per call. The extraction's only cross-record input is its median anchor, and leaving out one print leaves one of a few possible anchors, so each anchor's oriented, carry-checked observations are built once and a print only re-takes the medians its own observation touches. Markets carry a key (pair, anchor, left-out prints, min_trades); equal keys mean equal markets, and option_vol prices a family once per key. Results are unchanged. A new test checks every print's spot and forward against extract_fx_observations + ndf_forward_curve over its peers; on a real 2026-10-08 DTCC file, 889 sampled prints (506 forwards, 383 options) gave field-for-field the same value_fx_trade output, and all 57,536 records valued in 5.1 s. Tests that stubbed the extraction helpers now stub _peer_curve, the new seam, with unchanged assertions. Co-Authored-By: Claude Opus 5.5 --- .../cftc/openbb_cftc/utils/fx_valuation.py | 254 ++++++++++++++---- .../providers/cftc/tests/test_fx_valuation.py | 167 ++++++++---- 2 files changed, 317 insertions(+), 104 deletions(-) diff --git a/openbb_platform/providers/cftc/openbb_cftc/utils/fx_valuation.py b/openbb_platform/providers/cftc/openbb_cftc/utils/fx_valuation.py index 8785206da4b8..fc0d427f6f56 100644 --- a/openbb_platform/providers/cftc/openbb_cftc/utils/fx_valuation.py +++ b/openbb_platform/providers/cftc/openbb_cftc/utils/fx_valuation.py @@ -281,34 +281,163 @@ def _discount_for( return memo[key] +def _median_without(ordered: list[float], removed: list[float]) -> float | None: + """statistics.median of a sorted list after removing values that are in it, or None.""" + from bisect import bisect_left + + if not removed: + if not ordered: + return None + n = len(ordered) + mid = n // 2 + return ordered[mid] if n % 2 else (ordered[mid - 1] + ordered[mid]) / 2 + + gone = sorted(bisect_left(ordered, value) for value in removed) + + for i in range(1, len(gone)): # equal values: take successive slots + if gone[i] <= gone[i - 1]: + gone[i] = gone[i - 1] + 1 + + n = len(ordered) - len(gone) + + if n <= 0: + return None + + def at(rank: int) -> float: + index = rank + for slot in gone: + if slot <= index: + index += 1 + return ordered[index] + + mid = n // 2 + return at(mid) if n % 2 else (at(mid - 1) + at(mid)) / 2 + + +def _pair_view(fx_records: Iterable[dict], trade: dict, curve_date: dateType, memo: dict) -> dict: + """A pair's curve observations, prepared once per day for leave-one-out pricing. + + fx_market prices each print from its peers (the pair's other prints). + Re-extracting every peer per print is quadratic in a busy pair's day. The + extraction's only cross-record input is its median anchor, and leaving out + one print leaves one of a few possible anchors, so each anchor's oriented, + carry-checked observations are built once and each print only adjusts the + medians its own observation touches. + """ + from openbb_cftc.utils.fx import _fx_observation, resolve_pair + + key = f"fx-view:{trade['pair']}" + + if key not in memo: + _, spec = resolve_pair(trade["pair"]) + candidates = [] + by_id: dict[str, list[int]] = {} + + for record in _pair_subset(fx_records, trade, memo): + if is_package_leg(record): + continue + observation = _fx_observation(record, curve_date, spec, CURVE_MIN_NOTIONAL) + if observation is not None: + ident = (record.get("Dissemination Identifier") or "").strip() + by_id.setdefault(ident, []).append(len(candidates)) + candidates.append((record, observation)) + + memo[key] = { + "candidates": candidates, + "by_id": by_id, + "rates": sorted(observation["rate"] for _, observation in candidates), + "anchors": {}, + } + + return memo[key] + + +def _anchored(view: dict, anchor: float) -> dict: + """Every candidate oriented and carry-checked against one anchor, grouped by day.""" + from math import log + + from openbb_cftc.utils.fx import _stated_rate, carry_bound + + if anchor not in view["anchors"]: + kept: list[tuple[int, float] | None] = [] + by_day: dict[int, list[float]] = {} + + for record, observation in view["candidates"]: + stated = _stated_rate(record, anchor) + rate = stated if stated is not None else observation["rate"] + if abs(log(rate / anchor)) <= carry_bound(observation["days"]): + kept.append((observation["days"], rate)) + by_day.setdefault(observation["days"], []).append(rate) + else: + kept.append(None) + + for rates in by_day.values(): + rates.sort() + + view["anchors"][anchor] = {"kept": kept, "by_day": by_day, "medians": {}, "near": {}} + + return view["anchors"][anchor] + + +def _peer_curve(fx_records: Iterable[dict], trade: dict, curve_date: dateType, memo: dict) -> dict: + """The peer curve fx_market needs: extract_fx_observations over the pair minus the trade's own print.""" + view = _pair_view(fx_records, trade, curve_date, memo) + own = view["by_id"].get(trade["dissemination_identifier"], []) + anchor = _median_without(view["rates"], [view["candidates"][i][1]["rate"] for i in own]) + + key = (trade["pair"], anchor, tuple(own)) + + if anchor is None: + return {"nodes": [], "spot": None, "cip_near": None, "key": key} + + anchored = _anchored(view, anchor) + removed: dict[int, list[float]] = {} + + for i in own: + if anchored["kept"][i] is not None: + days, rate = anchored["kept"][i] + removed.setdefault(days, []).append(rate) + + nodes = [] + for days in sorted(anchored["by_day"]): + if days <= 2: + continue + out = removed.get(days, []) + if not out and days in anchored["medians"]: + median = anchored["medians"][days] + else: + median = _median_without(anchored["by_day"][days], out) + if not out: + anchored["medians"][days] = median + if median is not None: + nodes.append((days, median)) + + def near(low: int) -> float | None: + if low not in anchored["near"]: + anchored["near"][low] = sorted( + r for d, rs in anchored["by_day"].items() if low <= d <= 9 for r in rs + ) + out = [r for d, rs in removed.items() if low <= d <= 9 for r in rs] + return _median_without(anchored["near"][low], out) + + spot = near(3) if any(3 <= d <= 9 for d, _ in nodes) else (nodes[0][1] if nodes else None) + return {"nodes": nodes, "spot": spot, "cip_near": near(0), "key": key} + + def _cip_market( - peers: list[dict], + near_spot: float | None, rates_records: Iterable[dict], trade: dict, curve_date: dateType, min_trades: int, memo: dict, + key: tuple | None = None, ) -> dict | None: """Both legs' curves and a covered-interest-parity forward, when prints are sparse.""" - from statistics import median - - from openbb_cftc.utils.fx import extract_fx_observations - - near = [ - observation["rate"] - for observation in extract_fx_observations( - peers, - pair=trade["pair"], - trade_date=curve_date, - min_notional=CURVE_MIN_NOTIONAL, - ) - if observation["days"] <= 9 - ] - - if not near: + if near_spot is None: return None - spot = median(near) + spot = near_spot if spot <= 0.0: return None @@ -324,7 +453,7 @@ def _cip_market( def forward(years: float, _s=spot, _b=df_base, _q=df_quote) -> float: return _s * _b(years) / _q(years) - return {"spot": spot, "forward": forward, "df_base": df_base, "df_quote": df_quote} + return {"spot": spot, "forward": forward, "df_base": df_base, "df_quote": df_quote, "key": key} def fx_market( @@ -336,37 +465,39 @@ def fx_market( memo: dict | None = None, ) -> dict | None: """Spot, forward interpolator and both legs' discount curves for a pair, or None.""" - from openbb_cftc.utils.fx import extract_fx_observations - from openbb_cftc.utils.fx_vol import implied_df, ndf_forward_curve + from bisect import bisect_right + + from openbb_cftc.utils.fx_vol import implied_df memo = {} if memo is None else memo - peers = [ - record - for record in _pair_subset(fx_records, trade, memo) - if (record.get("Dissemination Identifier") or "").strip() - != trade["dissemination_identifier"] - and not is_package_leg(record) - ] - nodes = { - observation["days"] - for observation in extract_fx_observations( - peers, - pair=trade["pair"], - trade_date=curve_date, - min_notional=CURVE_MIN_NOTIONAL, - ) - if observation["days"] > 2 - } + curve = _peer_curve(fx_records, trade, curve_date, memo) + nodes = curve["nodes"] + key = (*curve["key"], min_trades) if len(nodes) < MIN_FORWARD_NODES: - return _cip_market(peers, rates_records, trade, curve_date, min_trades, memo) + return _cip_market(curve["cip_near"], rates_records, trade, curve_date, min_trades, memo, key) - spot, forward = ndf_forward_curve( - peers, trade["pair"], curve_date, min_notional=CURVE_MIN_NOTIONAL - ) + spot = curve["spot"] + days = [node[0] for node in nodes] + + def forward(years: float, _nodes=nodes, _days=days) -> float: + target = years * 365.0 - if spot is None or forward is None or spot <= 0.0: - return _cip_market(peers, rates_records, trade, curve_date, min_trades, memo) + if target <= _days[0]: + return _nodes[0][1] + + if target >= _days[-1]: + return _nodes[-1][1] + + upper = bisect_right(_days, target) + day_low, rate_low = _nodes[upper - 1] + day_high, rate_high = _nodes[upper] + weight = (target - day_low) / (day_high - day_low) + + return rate_low + weight * (rate_high - rate_low) + + if spot is None or spot <= 0.0: + return _cip_market(curve["cip_near"], rates_records, trade, curve_date, min_trades, memo, key) df_base = _discount_for(rates_records, trade["base"], curve_date, min_trades, memo) @@ -385,7 +516,8 @@ def _base_df(years: float, _quote=df_quote) -> float: df_base = _base_df - return {"spot": spot, "forward": forward, "df_base": df_base, "df_quote": df_quote} + # key: equal keys mean equal markets (same pair, anchor and left-out prints). + return {"spot": spot, "forward": forward, "df_base": df_base, "df_quote": df_quote, "key": key} def value_fx_forward(trade: dict, market: dict) -> dict: @@ -422,17 +554,27 @@ def option_vol( return None memo = {} if memo is None else memo - options = extract_vanilla_options( - _pair_subset(fx_records, trade, memo), - trade["base"], - trade["quote"], - curve_date, - market["spot"], - family=family, - ) - priced = price_options( - options, market["spot"], market["df_quote"], market["df_base"] - ) + # The priced options depend only on the market and the family, and option + # prints never enter the forward curve, so every option print of a pair + # shares one market. Price the family once per market, not once per print. + cache = ("fx-optvol", market["key"], family) if market.get("key") else None + + if cache is not None and cache in memo: + priced = memo[cache] + else: + options = extract_vanilla_options( + _pair_subset(fx_records, trade, memo), + trade["base"], + trade["quote"], + curve_date, + market["spot"], + family=family, + ) + priced = price_options( + options, market["spot"], market["df_quote"], market["df_base"] + ) + if cache is not None: + memo[cache] = priced if not priced: return None diff --git a/openbb_platform/providers/cftc/tests/test_fx_valuation.py b/openbb_platform/providers/cftc/tests/test_fx_valuation.py index e36834978369..6ffbffa5d801 100644 --- a/openbb_platform/providers/cftc/tests/test_fx_valuation.py +++ b/openbb_platform/providers/cftc/tests/test_fx_valuation.py @@ -304,17 +304,26 @@ def _observations(days_list): return [{"days": d, "rate": 1460.0} for d in days_list] +def _curve(observations): + """The peer curve fx_market reads, built from observations by the same rules.""" + from statistics import median + + by_day: dict = {} + for observation in observations: + by_day.setdefault(observation["days"], []).append(observation["rate"]) + nodes = sorted((days, median(rates)) for days, rates in by_day.items() if days > 2) + near = [r for days, rates in by_day.items() if 3 <= days <= 9 for r in rates] + cip = [r for days, rates in by_day.items() if days <= 9 for r in rates] + spot = median(near) if near else (nodes[0][1] if nodes else None) + + return {"nodes": nodes, "spot": spot, "cip_near": median(cip) if cip else None, "key": ("test",)} + + def _patch_market(monkeypatch, nodes, spot=1460.0, base_ok=True, quote_ok=True): from openbb_core.provider.utils.errors import EmptyDataError - monkeypatch.setattr( - "openbb_cftc.utils.fx.extract_fx_observations", - lambda *a, **k: _observations(nodes), - ) - monkeypatch.setattr( - "openbb_cftc.utils.fx_vol.ndf_forward_curve", - lambda *a, **k: (spot, (lambda years: spot * (1.0 + 0.01 * years))), - ) + curve = _curve([{"days": d, "rate": spot} for d in nodes]) + monkeypatch.setattr(fv, "_peer_curve", lambda *a, **k: curve) def _rate_df(records, currency, day, min_trades): if currency == "USD" and not base_ok: @@ -365,9 +374,8 @@ def test_fx_market_requires_enough_nodes(monkeypatch): def test_fx_market_recovers_a_missing_spot_through_cip(monkeypatch): _patch_market(monkeypatch, [7, 38, 90]) - monkeypatch.setattr( - "openbb_cftc.utils.fx_vol.ndf_forward_curve", lambda *a, **k: (None, None) - ) + curve = {**_curve(_observations([7, 38, 90])), "spot": None} + monkeypatch.setattr(fv, "_peer_curve", lambda *a, **k: curve) trade = fv.extract_fx_trade(_forward(), DAY) market = fv.fx_market([], [], trade, DAY) @@ -376,10 +384,7 @@ def test_fx_market_recovers_a_missing_spot_through_cip(monkeypatch): def test_fx_market_cip_fallback_prices_off_both_curves(monkeypatch): - monkeypatch.setattr( - "openbb_cftc.utils.fx.extract_fx_observations", - lambda *a, **k: [{"days": 2, "rate": 1460.0}, {"days": 5, "rate": 1461.0}], - ) + monkeypatch.setattr(fv, "_peer_curve", lambda *a, **k: _curve([{"days": 2, "rate": 1460.0}, {"days": 5, "rate": 1461.0}])) def _rate_df(records, currency, day, min_trades): if currency == "USD": @@ -401,10 +406,7 @@ def _rate_df(records, currency, day, min_trades): def test_fx_market_cip_fallback_needs_both_curves(monkeypatch): from openbb_core.provider.utils.errors import EmptyDataError - monkeypatch.setattr( - "openbb_cftc.utils.fx.extract_fx_observations", - lambda *a, **k: [{"days": 2, "rate": 1460.0}], - ) + monkeypatch.setattr(fv, "_peer_curve", lambda *a, **k: _curve([{"days": 2, "rate": 1460.0}])) def _usd_only(records, currency, day, min_trades): if currency == "USD": @@ -419,54 +421,53 @@ def _usd_only(records, currency, day, min_trades): def test_fx_market_cip_fallback_needs_a_near_print(monkeypatch): - monkeypatch.setattr( - "openbb_cftc.utils.fx.extract_fx_observations", - lambda *a, **k: [{"days": 38, "rate": 1460.0}], - ) + monkeypatch.setattr(fv, "_peer_curve", lambda *a, **k: _curve([{"days": 38, "rate": 1460.0}])) trade = fv.extract_fx_trade(_forward(), DAY) assert fv.fx_market([], [], trade, DAY) is None def test_fx_market_cip_fallback_rejects_a_nonpositive_spot(monkeypatch): - monkeypatch.setattr( - "openbb_cftc.utils.fx.extract_fx_observations", - lambda *a, **k: [{"days": 2, "rate": 0.0}], - ) + monkeypatch.setattr(fv, "_peer_curve", lambda *a, **k: _curve([{"days": 2, "rate": 0.0}])) trade = fv.extract_fx_trade(_forward(), DAY) assert fv.fx_market([], [], trade, DAY) is None -def test_fx_market_excludes_the_trades_own_print(monkeypatch): - seen = {} - - def _observe(records, **kwargs): - seen["ids"] = [r.get("Dissemination Identifier") for r in records] - return _observations([7, 38, 90]) +def _print(identifier: str, rate: float, days: int) -> dict: + """A KRW/USD NDF print at a rate, settling a number of days after DAY.""" + from datetime import timedelta - monkeypatch.setattr("openbb_cftc.utils.fx.extract_fx_observations", _observe) - monkeypatch.setattr( - "openbb_cftc.utils.fx_vol.ndf_forward_curve", - lambda *a, **k: (1460.0, lambda years: 1460.0), + return _forward( + **{ + "Dissemination Identifier": identifier, + "Exchange rate": f"{rate:,.2f}", + "Expiration Date": (DAY + timedelta(days=days)).isoformat(), + } ) + + +def test_fx_market_excludes_the_trades_own_print(monkeypatch): monkeypatch.setattr( "openbb_cftc.utils.fx_vol.rate_discount_factor", lambda *a, **k: lambda years: 0.98**years, ) - monkeypatch.setattr( - "openbb_cftc.utils.fx_vol.implied_df", lambda *a, **k: lambda years: 0.97 - ) - own = _forward() - peer = _forward(**{"Dissemination Identifier": "4402190022000000201"}) - trade = fv.extract_fx_trade(own, DAY) - fv.fx_market([own, peer], [], trade, DAY) + own = _print("4402190022000000101", 1500.0, 5) + peers = [ + _print("4402190022000000201", 1460.0, 5), + _print("4402190022000000202", 1460.0, 38), + _print("4402190022000000203", 1460.0, 90), + ] + records = [own, *peers] + + market = fv.fx_market(records, [], fv.extract_fx_trade(own, DAY), DAY) + assert market["spot"] == pytest.approx(1460.0) - assert seen["ids"] == ["4402190022000000201"] + market = fv.fx_market(records, [], fv.extract_fx_trade(peers[0], DAY), DAY) + assert market["spot"] == pytest.approx(1500.0) def test_fx_market_memo_builds_the_pair_subset_and_curves_once(monkeypatch): - _patch_market(monkeypatch, [7, 38, 90]) subset_calls: list = [] df_calls: list = [] real_subset = fv.pair_records @@ -481,10 +482,11 @@ def _counted_df(records, currency, day, min_trades): monkeypatch.setattr(fv, "pair_records", _counted_subset) monkeypatch.setattr("openbb_cftc.utils.fx_vol.rate_discount_factor", _counted_df) + records = [_print(f"44021900220000002{n}", 1460.0, d) for n, d in enumerate((5, 38, 90))] trade = fv.extract_fx_trade(_forward(), DAY) memo: dict = {} - first = fv.fx_market([], [], trade, DAY, memo=memo) - second = fv.fx_market([], [], trade, DAY, memo=memo) + first = fv.fx_market(records, [], trade, DAY, memo=memo) + second = fv.fx_market(records, [], trade, DAY, memo=memo) assert first["spot"] == second["spot"] assert len(subset_calls) == 1 @@ -568,3 +570,72 @@ def test_value_fx_trade_gates_a_rate_beyond_the_carry_bound(monkeypatch): record = _forward(**{"Exchange rate": "2,900.00", "Exchange rate basis": "USD/KRW"}) assert fv.value_fx_trade(record, [], [], DAY) is None + + +def test_fx_market_matches_the_leave_one_out_curve(monkeypatch): + """Every print's market equals one built from its peers by extraction.""" + import random + + from openbb_cftc.utils.fx import extract_fx_observations + from openbb_cftc.utils.fx_vol import ndf_forward_curve + + monkeypatch.setattr( + "openbb_cftc.utils.fx_vol.rate_discount_factor", + lambda *a, **k: lambda years: 0.98**years, + ) + rng = random.Random(3) + records = [ + _print(f"44021900220000{n:05d}", rng.choice([1455.0, 1458.5, 1460.0, 1462.0, 1468.0, 1520.0, 1390.0]), rng.choice([1, 2, 5, 7, 7, 30, 38, 90, 180])) + for n in range(60) + ] + records.append(_print("4402190022000000007", 1461.0, 38)) # a second print sharing an identifier + memo: dict = {} + compared = 0 + + for record in records: + trade = fv.extract_fx_trade(record, DAY) + if trade is None: + continue + market = fv.fx_market(records, [], trade, DAY, memo=memo) + peers = [ + r + for r in records + if r["Dissemination Identifier"] != trade["dissemination_identifier"] + and not fv.is_package_leg(r) + ] + nodes = { + o["days"] + for o in extract_fx_observations(peers, pair=trade["pair"], trade_date=DAY, min_notional=fv.CURVE_MIN_NOTIONAL) + if o["days"] > 2 + } + if len(nodes) < fv.MIN_FORWARD_NODES: + continue + spot, forward = ndf_forward_curve(peers, trade["pair"], DAY, min_notional=fv.CURVE_MIN_NOTIONAL) + assert market["spot"] == spot + for years in (0.005, 0.02, 0.1, 0.25, 0.5, 1.0): + assert market["forward"](years) == forward(years) + compared += 1 + + assert compared >= 40 + + +def test_option_vol_prices_a_family_once_per_market(monkeypatch): + calls: list = [] + + def _priced(*a, **k): + calls.append(1) + return [{"tenor_days": 38, "implied_vol": 0.10}] + + monkeypatch.setattr("openbb_cftc.utils.fx_vol.extract_vanilla_options", lambda *a, **k: [{}]) + monkeypatch.setattr("openbb_cftc.utils.fx_vol.price_options", _priced) + first = fv.extract_fx_trade(_option(), DAY) + second = fv.extract_fx_trade(_option(**{"Dissemination Identifier": "4402190022000000999"}), DAY) + market = {**_market(1.14), "key": ("EURUSD", 1.14, (), 1)} + memo: dict = {} + + assert fv.option_vol([], first, market, DAY, memo) == pytest.approx(0.10) + assert fv.option_vol([], second, market, DAY, memo) == pytest.approx(0.10) + assert len(calls) == 1 + + fv.option_vol([], first, _market(1.14), DAY, memo) + assert len(calls) == 2 From 70c3257a44ce9f7de7d3597e8da1ef1a2c0165fb Mon Sep 17 00:00:00 2001 From: artcashin <64659740+artcashin@users.noreply.github.com> Date: Mon, 12 Oct 2026 01:31:34 +0000 Subject: [PATCH 2/2] refactor(cftc): build each pair's curve once with the shared NDF utility Address review: drop the leave-one-out reimplementation (_pair_view, _anchored, _median_without and the inline forward interpolator). fx_market now calls extract_fx_observations and ndf_forward_curve once per pair and curve date over all of the pair's prints, package legs included, so valuation and the vol surface use the same NDF curve. The curve, market and option-pricing memo keys include the curve date. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01SEna2ixXYhEGEAQdJM2D3G --- .../cftc/openbb_cftc/utils/fx_valuation.py | 266 ++++++------------ .../providers/cftc/tests/test_fx_valuation.py | 224 +++++++-------- 2 files changed, 186 insertions(+), 304 deletions(-) diff --git a/openbb_platform/providers/cftc/openbb_cftc/utils/fx_valuation.py b/openbb_platform/providers/cftc/openbb_cftc/utils/fx_valuation.py index fc0d427f6f56..dfb9bfbe45e9 100644 --- a/openbb_platform/providers/cftc/openbb_cftc/utils/fx_valuation.py +++ b/openbb_platform/providers/cftc/openbb_cftc/utils/fx_valuation.py @@ -281,163 +281,51 @@ def _discount_for( return memo[key] -def _median_without(ordered: list[float], removed: list[float]) -> float | None: - """statistics.median of a sorted list after removing values that are in it, or None.""" - from bisect import bisect_left +def _pair_curve( + fx_records: Iterable[dict], trade: dict, curve_date: dateType, memo: dict +) -> dict: + """Return the pair's forward curve for the day, built once and shared by its prints.""" + from openbb_cftc.utils.fx import extract_fx_observations + from openbb_cftc.utils.fx_vol import ndf_forward_curve - if not removed: - if not ordered: - return None - n = len(ordered) - mid = n // 2 - return ordered[mid] if n % 2 else (ordered[mid - 1] + ordered[mid]) / 2 - - gone = sorted(bisect_left(ordered, value) for value in removed) - - for i in range(1, len(gone)): # equal values: take successive slots - if gone[i] <= gone[i - 1]: - gone[i] = gone[i - 1] + 1 - - n = len(ordered) - len(gone) - - if n <= 0: - return None - - def at(rank: int) -> float: - index = rank - for slot in gone: - if slot <= index: - index += 1 - return ordered[index] - - mid = n // 2 - return at(mid) if n % 2 else (at(mid - 1) + at(mid)) / 2 - - -def _pair_view(fx_records: Iterable[dict], trade: dict, curve_date: dateType, memo: dict) -> dict: - """A pair's curve observations, prepared once per day for leave-one-out pricing. - - fx_market prices each print from its peers (the pair's other prints). - Re-extracting every peer per print is quadratic in a busy pair's day. The - extraction's only cross-record input is its median anchor, and leaving out - one print leaves one of a few possible anchors, so each anchor's oriented, - carry-checked observations are built once and each print only adjusts the - medians its own observation touches. - """ - from openbb_cftc.utils.fx import _fx_observation, resolve_pair - - key = f"fx-view:{trade['pair']}" + key = f"fx-curve:{trade['pair']}:{curve_date.isoformat()}" if key not in memo: - _, spec = resolve_pair(trade["pair"]) - candidates = [] - by_id: dict[str, list[int]] = {} - - for record in _pair_subset(fx_records, trade, memo): - if is_package_leg(record): - continue - observation = _fx_observation(record, curve_date, spec, CURVE_MIN_NOTIONAL) - if observation is not None: - ident = (record.get("Dissemination Identifier") or "").strip() - by_id.setdefault(ident, []).append(len(candidates)) - candidates.append((record, observation)) - + records = _pair_subset(fx_records, trade, memo) + observations = extract_fx_observations( + records, + pair=trade["pair"], + trade_date=curve_date, + min_notional=CURVE_MIN_NOTIONAL, + ) + spot, forward = ndf_forward_curve( + records, trade["pair"], curve_date, min_notional=CURVE_MIN_NOTIONAL + ) memo[key] = { - "candidates": candidates, - "by_id": by_id, - "rates": sorted(observation["rate"] for _, observation in candidates), - "anchors": {}, + "nodes": len({o["days"] for o in observations if o["days"] > 2}), + "near": [o["rate"] for o in observations if o["days"] <= 9], + "spot": spot, + "forward": forward, } return memo[key] -def _anchored(view: dict, anchor: float) -> dict: - """Every candidate oriented and carry-checked against one anchor, grouped by day.""" - from math import log - - from openbb_cftc.utils.fx import _stated_rate, carry_bound - - if anchor not in view["anchors"]: - kept: list[tuple[int, float] | None] = [] - by_day: dict[int, list[float]] = {} - - for record, observation in view["candidates"]: - stated = _stated_rate(record, anchor) - rate = stated if stated is not None else observation["rate"] - if abs(log(rate / anchor)) <= carry_bound(observation["days"]): - kept.append((observation["days"], rate)) - by_day.setdefault(observation["days"], []).append(rate) - else: - kept.append(None) - - for rates in by_day.values(): - rates.sort() - - view["anchors"][anchor] = {"kept": kept, "by_day": by_day, "medians": {}, "near": {}} - - return view["anchors"][anchor] - - -def _peer_curve(fx_records: Iterable[dict], trade: dict, curve_date: dateType, memo: dict) -> dict: - """The peer curve fx_market needs: extract_fx_observations over the pair minus the trade's own print.""" - view = _pair_view(fx_records, trade, curve_date, memo) - own = view["by_id"].get(trade["dissemination_identifier"], []) - anchor = _median_without(view["rates"], [view["candidates"][i][1]["rate"] for i in own]) - - key = (trade["pair"], anchor, tuple(own)) - - if anchor is None: - return {"nodes": [], "spot": None, "cip_near": None, "key": key} - - anchored = _anchored(view, anchor) - removed: dict[int, list[float]] = {} - - for i in own: - if anchored["kept"][i] is not None: - days, rate = anchored["kept"][i] - removed.setdefault(days, []).append(rate) - - nodes = [] - for days in sorted(anchored["by_day"]): - if days <= 2: - continue - out = removed.get(days, []) - if not out and days in anchored["medians"]: - median = anchored["medians"][days] - else: - median = _median_without(anchored["by_day"][days], out) - if not out: - anchored["medians"][days] = median - if median is not None: - nodes.append((days, median)) - - def near(low: int) -> float | None: - if low not in anchored["near"]: - anchored["near"][low] = sorted( - r for d, rs in anchored["by_day"].items() if low <= d <= 9 for r in rs - ) - out = [r for d, rs in removed.items() if low <= d <= 9 for r in rs] - return _median_without(anchored["near"][low], out) - - spot = near(3) if any(3 <= d <= 9 for d, _ in nodes) else (nodes[0][1] if nodes else None) - return {"nodes": nodes, "spot": spot, "cip_near": near(0), "key": key} - - def _cip_market( - near_spot: float | None, + near: list[float], rates_records: Iterable[dict], trade: dict, curve_date: dateType, min_trades: int, memo: dict, - key: tuple | None = None, ) -> dict | None: """Both legs' curves and a covered-interest-parity forward, when prints are sparse.""" - if near_spot is None: + from statistics import median + + if not near: return None - spot = near_spot + spot = median(near) if spot <= 0.0: return None @@ -453,52 +341,21 @@ def _cip_market( def forward(years: float, _s=spot, _b=df_base, _q=df_quote) -> float: return _s * _b(years) / _q(years) - return {"spot": spot, "forward": forward, "df_base": df_base, "df_quote": df_quote, "key": key} + return {"spot": spot, "forward": forward, "df_base": df_base, "df_quote": df_quote} -def fx_market( - fx_records: Iterable[dict], +def _curve_market( + spot: float, + forward: Callable[[float], float], rates_records: Iterable[dict], trade: dict, curve_date: dateType, - min_trades: int = 1, - memo: dict | None = None, + min_trades: int, + memo: dict, ) -> dict | None: - """Spot, forward interpolator and both legs' discount curves for a pair, or None.""" - from bisect import bisect_right - + """Both legs' discount curves around an observed forward curve.""" from openbb_cftc.utils.fx_vol import implied_df - memo = {} if memo is None else memo - curve = _peer_curve(fx_records, trade, curve_date, memo) - nodes = curve["nodes"] - key = (*curve["key"], min_trades) - - if len(nodes) < MIN_FORWARD_NODES: - return _cip_market(curve["cip_near"], rates_records, trade, curve_date, min_trades, memo, key) - - spot = curve["spot"] - days = [node[0] for node in nodes] - - def forward(years: float, _nodes=nodes, _days=days) -> float: - target = years * 365.0 - - if target <= _days[0]: - return _nodes[0][1] - - if target >= _days[-1]: - return _nodes[-1][1] - - upper = bisect_right(_days, target) - day_low, rate_low = _nodes[upper - 1] - day_high, rate_high = _nodes[upper] - weight = (target - day_low) / (day_high - day_low) - - return rate_low + weight * (rate_high - rate_low) - - if spot is None or spot <= 0.0: - return _cip_market(curve["cip_near"], rates_records, trade, curve_date, min_trades, memo, key) - df_base = _discount_for(rates_records, trade["base"], curve_date, min_trades, memo) if df_base is not None: @@ -516,8 +373,47 @@ def _base_df(years: float, _quote=df_quote) -> float: df_base = _base_df - # key: equal keys mean equal markets (same pair, anchor and left-out prints). - return {"spot": spot, "forward": forward, "df_base": df_base, "df_quote": df_quote, "key": key} + return {"spot": spot, "forward": forward, "df_base": df_base, "df_quote": df_quote} + + +def fx_market( + fx_records: Iterable[dict], + rates_records: Iterable[dict], + trade: dict, + curve_date: dateType, + min_trades: int = 1, + memo: dict | None = None, +) -> dict | None: + """Spot, forward interpolator and both legs' discount curves for a pair, or None.""" + memo = {} if memo is None else memo + key = f"fx-market:{trade['pair']}:{curve_date.isoformat()}:{min_trades}" + + if key in memo: + return memo[key] + + curve = _pair_curve(fx_records, trade, curve_date, memo) + spot, forward = curve["spot"], curve["forward"] + + if ( + curve["nodes"] < MIN_FORWARD_NODES + or spot is None + or forward is None + or spot <= 0.0 + ): + market = _cip_market( + curve["near"], rates_records, trade, curve_date, min_trades, memo + ) + else: + market = _curve_market( + spot, forward, rates_records, trade, curve_date, min_trades, memo + ) + + if market is not None: + market["key"] = key + + memo[key] = market + + return market def value_fx_forward(trade: dict, market: dict) -> dict: @@ -554,13 +450,10 @@ def option_vol( return None memo = {} if memo is None else memo - # The priced options depend only on the market and the family, and option - # prints never enter the forward curve, so every option print of a pair - # shares one market. Price the family once per market, not once per print. - cache = ("fx-optvol", market["key"], family) if market.get("key") else None + key = f"fx-options:{market['key']}:{family}" if market.get("key") else None - if cache is not None and cache in memo: - priced = memo[cache] + if key is not None and key in memo: + priced = memo[key] else: options = extract_vanilla_options( _pair_subset(fx_records, trade, memo), @@ -573,8 +466,9 @@ def option_vol( priced = price_options( options, market["spot"], market["df_quote"], market["df_base"] ) - if cache is not None: - memo[cache] = priced + + if key is not None: + memo[key] = priced if not priced: return None diff --git a/openbb_platform/providers/cftc/tests/test_fx_valuation.py b/openbb_platform/providers/cftc/tests/test_fx_valuation.py index 6ffbffa5d801..8455264189bd 100644 --- a/openbb_platform/providers/cftc/tests/test_fx_valuation.py +++ b/openbb_platform/providers/cftc/tests/test_fx_valuation.py @@ -1,4 +1,4 @@ -from datetime import date +from datetime import date, timedelta import pytest @@ -304,26 +304,17 @@ def _observations(days_list): return [{"days": d, "rate": 1460.0} for d in days_list] -def _curve(observations): - """The peer curve fx_market reads, built from observations by the same rules.""" - from statistics import median - - by_day: dict = {} - for observation in observations: - by_day.setdefault(observation["days"], []).append(observation["rate"]) - nodes = sorted((days, median(rates)) for days, rates in by_day.items() if days > 2) - near = [r for days, rates in by_day.items() if 3 <= days <= 9 for r in rates] - cip = [r for days, rates in by_day.items() if days <= 9 for r in rates] - spot = median(near) if near else (nodes[0][1] if nodes else None) - - return {"nodes": nodes, "spot": spot, "cip_near": median(cip) if cip else None, "key": ("test",)} - - def _patch_market(monkeypatch, nodes, spot=1460.0, base_ok=True, quote_ok=True): from openbb_core.provider.utils.errors import EmptyDataError - curve = _curve([{"days": d, "rate": spot} for d in nodes]) - monkeypatch.setattr(fv, "_peer_curve", lambda *a, **k: curve) + monkeypatch.setattr( + "openbb_cftc.utils.fx.extract_fx_observations", + lambda *a, **k: _observations(nodes), + ) + monkeypatch.setattr( + "openbb_cftc.utils.fx_vol.ndf_forward_curve", + lambda *a, **k: (spot, (lambda years: spot * (1.0 + 0.01 * years))), + ) def _rate_df(records, currency, day, min_trades): if currency == "USD" and not base_ok: @@ -374,8 +365,9 @@ def test_fx_market_requires_enough_nodes(monkeypatch): def test_fx_market_recovers_a_missing_spot_through_cip(monkeypatch): _patch_market(monkeypatch, [7, 38, 90]) - curve = {**_curve(_observations([7, 38, 90])), "spot": None} - monkeypatch.setattr(fv, "_peer_curve", lambda *a, **k: curve) + monkeypatch.setattr( + "openbb_cftc.utils.fx_vol.ndf_forward_curve", lambda *a, **k: (None, None) + ) trade = fv.extract_fx_trade(_forward(), DAY) market = fv.fx_market([], [], trade, DAY) @@ -384,7 +376,10 @@ def test_fx_market_recovers_a_missing_spot_through_cip(monkeypatch): def test_fx_market_cip_fallback_prices_off_both_curves(monkeypatch): - monkeypatch.setattr(fv, "_peer_curve", lambda *a, **k: _curve([{"days": 2, "rate": 1460.0}, {"days": 5, "rate": 1461.0}])) + monkeypatch.setattr( + "openbb_cftc.utils.fx.extract_fx_observations", + lambda *a, **k: [{"days": 2, "rate": 1460.0}, {"days": 5, "rate": 1461.0}], + ) def _rate_df(records, currency, day, min_trades): if currency == "USD": @@ -406,7 +401,10 @@ def _rate_df(records, currency, day, min_trades): def test_fx_market_cip_fallback_needs_both_curves(monkeypatch): from openbb_core.provider.utils.errors import EmptyDataError - monkeypatch.setattr(fv, "_peer_curve", lambda *a, **k: _curve([{"days": 2, "rate": 1460.0}])) + monkeypatch.setattr( + "openbb_cftc.utils.fx.extract_fx_observations", + lambda *a, **k: [{"days": 2, "rate": 1460.0}], + ) def _usd_only(records, currency, day, min_trades): if currency == "USD": @@ -421,53 +419,92 @@ def _usd_only(records, currency, day, min_trades): def test_fx_market_cip_fallback_needs_a_near_print(monkeypatch): - monkeypatch.setattr(fv, "_peer_curve", lambda *a, **k: _curve([{"days": 38, "rate": 1460.0}])) + monkeypatch.setattr( + "openbb_cftc.utils.fx.extract_fx_observations", + lambda *a, **k: [{"days": 38, "rate": 1460.0}], + ) trade = fv.extract_fx_trade(_forward(), DAY) assert fv.fx_market([], [], trade, DAY) is None def test_fx_market_cip_fallback_rejects_a_nonpositive_spot(monkeypatch): - monkeypatch.setattr(fv, "_peer_curve", lambda *a, **k: _curve([{"days": 2, "rate": 0.0}])) + monkeypatch.setattr( + "openbb_cftc.utils.fx.extract_fx_observations", + lambda *a, **k: [{"days": 2, "rate": 0.0}], + ) trade = fv.extract_fx_trade(_forward(), DAY) assert fv.fx_market([], [], trade, DAY) is None -def _print(identifier: str, rate: float, days: int) -> dict: - """A KRW/USD NDF print at a rate, settling a number of days after DAY.""" - from datetime import timedelta - - return _forward( - **{ - "Dissemination Identifier": identifier, - "Exchange rate": f"{rate:,.2f}", - "Expiration Date": (DAY + timedelta(days=days)).isoformat(), - } - ) +def test_fx_market_uses_the_same_curve_as_the_vol_surface(monkeypatch): + seen = {} + def _curve(records, pair, as_of, **kwargs): + seen["ids"] = [r.get("Dissemination Identifier") for r in records] + return 1460.0, lambda years: 1460.0 -def test_fx_market_excludes_the_trades_own_print(monkeypatch): + monkeypatch.setattr( + "openbb_cftc.utils.fx.extract_fx_observations", + lambda *a, **k: _observations([7, 38, 90]), + ) + monkeypatch.setattr("openbb_cftc.utils.fx_vol.ndf_forward_curve", _curve) monkeypatch.setattr( "openbb_cftc.utils.fx_vol.rate_discount_factor", lambda *a, **k: lambda years: 0.98**years, ) - own = _print("4402190022000000101", 1500.0, 5) - peers = [ - _print("4402190022000000201", 1460.0, 5), - _print("4402190022000000202", 1460.0, 38), - _print("4402190022000000203", 1460.0, 90), + monkeypatch.setattr( + "openbb_cftc.utils.fx_vol.implied_df", lambda *a, **k: lambda years: 0.97 + ) + own = _forward() + peer = _forward(**{"Dissemination Identifier": "4402190022000000201"}) + leg = _forward( + **{ + "Dissemination Identifier": "4402190022000000202", + "Package indicator": "TRUE", + } + ) + trade = fv.extract_fx_trade(own, DAY) + fv.fx_market([own, peer, leg], [], trade, DAY) + + assert seen["ids"] == [ + own["Dissemination Identifier"], + "4402190022000000201", + "4402190022000000202", ] - records = [own, *peers] - market = fv.fx_market(records, [], fv.extract_fx_trade(own, DAY), DAY) - assert market["spot"] == pytest.approx(1460.0) - market = fv.fx_market(records, [], fv.extract_fx_trade(peers[0], DAY), DAY) - assert market["spot"] == pytest.approx(1500.0) +def test_fx_market_builds_the_curve_once_per_pair_and_day(monkeypatch): + _patch_market(monkeypatch, [7, 38, 90]) + calls: list = [] + + def _curve(records, pair, as_of, **kwargs): + calls.append(as_of) + return 1460.0, lambda years: 1460.0 + + monkeypatch.setattr("openbb_cftc.utils.fx_vol.ndf_forward_curve", _curve) + first = _forward() + second = _forward(**{"Dissemination Identifier": "4402190022000000201"}) + memo: dict = {} + + for record in (first, second): + fv.fx_market( + [first, second], [], fv.extract_fx_trade(record, DAY), DAY, memo=memo + ) + + assert calls == [DAY] + + later = DAY + timedelta(days=1) + fv.fx_market( + [first, second], [], fv.extract_fx_trade(first, later), later, memo=memo + ) + + assert calls == [DAY, later] def test_fx_market_memo_builds_the_pair_subset_and_curves_once(monkeypatch): + _patch_market(monkeypatch, [7, 38, 90]) subset_calls: list = [] df_calls: list = [] real_subset = fv.pair_records @@ -482,11 +519,10 @@ def _counted_df(records, currency, day, min_trades): monkeypatch.setattr(fv, "pair_records", _counted_subset) monkeypatch.setattr("openbb_cftc.utils.fx_vol.rate_discount_factor", _counted_df) - records = [_print(f"44021900220000002{n}", 1460.0, d) for n, d in enumerate((5, 38, 90))] trade = fv.extract_fx_trade(_forward(), DAY) memo: dict = {} - first = fv.fx_market(records, [], trade, DAY, memo=memo) - second = fv.fx_market(records, [], trade, DAY, memo=memo) + first = fv.fx_market([], [], trade, DAY, memo=memo) + second = fv.fx_market([], [], trade, DAY, memo=memo) assert first["spot"] == second["spot"] assert len(subset_calls) == 1 @@ -530,6 +566,27 @@ def test_option_vol_returns_none_without_a_family(): assert fv.option_vol([], trade, _market(1.14), DAY) is None +def test_option_vol_prices_a_family_once_per_market(monkeypatch): + calls: list = [] + + def _price(*args, **kwargs): + calls.append(args) + return [{"tenor_days": 38, "implied_vol": 0.10}] + + monkeypatch.setattr( + "openbb_cftc.utils.fx_vol.extract_vanilla_options", lambda *a, **k: [{}] + ) + monkeypatch.setattr("openbb_cftc.utils.fx_vol.price_options", _price) + trade = fv.extract_fx_trade(_option(), DAY) + market = {**_market(1.14), "key": "fx-market:EURUSD:2026-07-24:1"} + memo: dict = {} + + for _ in range(3): + assert fv.option_vol([], trade, market, DAY, memo) == pytest.approx(0.10) + + assert len(calls) == 1 + + def test_value_fx_trade_prices_a_forward(monkeypatch): _patch_market(monkeypatch, [7, 38, 90]) valued = fv.value_fx_trade(_forward(), [], [], DAY) @@ -570,72 +627,3 @@ def test_value_fx_trade_gates_a_rate_beyond_the_carry_bound(monkeypatch): record = _forward(**{"Exchange rate": "2,900.00", "Exchange rate basis": "USD/KRW"}) assert fv.value_fx_trade(record, [], [], DAY) is None - - -def test_fx_market_matches_the_leave_one_out_curve(monkeypatch): - """Every print's market equals one built from its peers by extraction.""" - import random - - from openbb_cftc.utils.fx import extract_fx_observations - from openbb_cftc.utils.fx_vol import ndf_forward_curve - - monkeypatch.setattr( - "openbb_cftc.utils.fx_vol.rate_discount_factor", - lambda *a, **k: lambda years: 0.98**years, - ) - rng = random.Random(3) - records = [ - _print(f"44021900220000{n:05d}", rng.choice([1455.0, 1458.5, 1460.0, 1462.0, 1468.0, 1520.0, 1390.0]), rng.choice([1, 2, 5, 7, 7, 30, 38, 90, 180])) - for n in range(60) - ] - records.append(_print("4402190022000000007", 1461.0, 38)) # a second print sharing an identifier - memo: dict = {} - compared = 0 - - for record in records: - trade = fv.extract_fx_trade(record, DAY) - if trade is None: - continue - market = fv.fx_market(records, [], trade, DAY, memo=memo) - peers = [ - r - for r in records - if r["Dissemination Identifier"] != trade["dissemination_identifier"] - and not fv.is_package_leg(r) - ] - nodes = { - o["days"] - for o in extract_fx_observations(peers, pair=trade["pair"], trade_date=DAY, min_notional=fv.CURVE_MIN_NOTIONAL) - if o["days"] > 2 - } - if len(nodes) < fv.MIN_FORWARD_NODES: - continue - spot, forward = ndf_forward_curve(peers, trade["pair"], DAY, min_notional=fv.CURVE_MIN_NOTIONAL) - assert market["spot"] == spot - for years in (0.005, 0.02, 0.1, 0.25, 0.5, 1.0): - assert market["forward"](years) == forward(years) - compared += 1 - - assert compared >= 40 - - -def test_option_vol_prices_a_family_once_per_market(monkeypatch): - calls: list = [] - - def _priced(*a, **k): - calls.append(1) - return [{"tenor_days": 38, "implied_vol": 0.10}] - - monkeypatch.setattr("openbb_cftc.utils.fx_vol.extract_vanilla_options", lambda *a, **k: [{}]) - monkeypatch.setattr("openbb_cftc.utils.fx_vol.price_options", _priced) - first = fv.extract_fx_trade(_option(), DAY) - second = fv.extract_fx_trade(_option(**{"Dissemination Identifier": "4402190022000000999"}), DAY) - market = {**_market(1.14), "key": ("EURUSD", 1.14, (), 1)} - memo: dict = {} - - assert fv.option_vol([], first, market, DAY, memo) == pytest.approx(0.10) - assert fv.option_vol([], second, market, DAY, memo) == pytest.approx(0.10) - assert len(calls) == 1 - - fv.option_vol([], first, _market(1.14), DAY, memo) - assert len(calls) == 2