AdrianoDev
116 lines
3.8 KiB
Python
116 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
from mcp_macro.cot import classify_extreme, compute_percentile
|
|
|
|
|
|
def test_compute_percentile_basic():
|
|
history = [10, 20, 30, 40, 50, 60, 70, 80, 90, 100]
|
|
assert compute_percentile(50, history) == 50.0
|
|
assert compute_percentile(10, history) == 10.0
|
|
assert compute_percentile(100, history) == 100.0
|
|
|
|
|
|
def test_compute_percentile_value_below_min():
|
|
history = [10, 20, 30]
|
|
assert compute_percentile(5, history) == 0.0
|
|
|
|
|
|
def test_compute_percentile_value_above_max():
|
|
history = [10, 20, 30]
|
|
assert compute_percentile(40, history) == 100.0
|
|
|
|
|
|
def test_compute_percentile_empty_history():
|
|
assert compute_percentile(50, []) is None
|
|
|
|
|
|
def test_classify_extreme_below_threshold():
|
|
assert classify_extreme(3.0) == "extreme_short"
|
|
assert classify_extreme(5.0) == "extreme_short" # boundary inclusive
|
|
|
|
|
|
def test_classify_extreme_above_threshold():
|
|
assert classify_extreme(96.0) == "extreme_long"
|
|
assert classify_extreme(95.0) == "extreme_long" # boundary inclusive
|
|
|
|
|
|
def test_classify_extreme_neutral():
|
|
assert classify_extreme(50.0) == "neutral"
|
|
assert classify_extreme(94.99) == "neutral"
|
|
assert classify_extreme(5.01) == "neutral"
|
|
|
|
|
|
def test_classify_extreme_none_input():
|
|
assert classify_extreme(None) == "neutral"
|
|
|
|
|
|
from mcp_macro.cot import parse_disagg_row, parse_tff_row
|
|
|
|
|
|
# Payload Socrata reale (subset campi rilevanti, valori arbitrari per test)
|
|
TFF_SOCRATA_ROW = {
|
|
"report_date_as_yyyy_mm_dd": "2026-04-22T00:00:00.000",
|
|
"dealer_positions_long_all": "12345",
|
|
"dealer_positions_short_all": "23456",
|
|
"asset_mgr_positions_long": "654321",
|
|
"asset_mgr_positions_short": "200000",
|
|
"lev_money_positions_long": "100000",
|
|
"lev_money_positions_short": "350000",
|
|
"other_rept_positions_long": "50000",
|
|
"other_rept_positions_short": "50000",
|
|
"open_interest_all": "2500000",
|
|
}
|
|
|
|
DISAGG_SOCRATA_ROW = {
|
|
"report_date_as_yyyy_mm_dd": "2026-04-22T00:00:00.000",
|
|
"prod_merc_positions_long_all": "100000",
|
|
"prod_merc_positions_short_all": "300000",
|
|
"swap_positions_long_all": "50000",
|
|
"swap_positions_short_all": "60000",
|
|
"m_money_positions_long_all": "200000",
|
|
"m_money_positions_short_all": "80000",
|
|
"other_rept_positions_long_all": "10000",
|
|
"other_rept_positions_short_all": "10000",
|
|
"open_interest_all": "1500000",
|
|
}
|
|
|
|
|
|
def test_parse_tff_row_extracts_all_fields():
|
|
row = parse_tff_row(TFF_SOCRATA_ROW)
|
|
assert row["report_date"] == "2026-04-22"
|
|
assert row["dealer_long"] == 12345
|
|
assert row["dealer_short"] == 23456
|
|
assert row["dealer_net"] == 12345 - 23456
|
|
assert row["asset_mgr_long"] == 654321
|
|
assert row["asset_mgr_net"] == 654321 - 200000
|
|
assert row["lev_funds_long"] == 100000
|
|
assert row["lev_funds_short"] == 350000
|
|
assert row["lev_funds_net"] == 100000 - 350000
|
|
assert row["other_long"] == 50000
|
|
assert row["other_net"] == 0
|
|
assert row["open_interest"] == 2500000
|
|
|
|
|
|
def test_parse_tff_row_handles_missing_field():
|
|
payload = {"report_date_as_yyyy_mm_dd": "2026-04-22T00:00:00.000"}
|
|
row = parse_tff_row(payload)
|
|
assert row["report_date"] == "2026-04-22"
|
|
assert row["dealer_long"] == 0
|
|
assert row["dealer_net"] == 0
|
|
|
|
|
|
def test_parse_disagg_row_extracts_all_fields():
|
|
row = parse_disagg_row(DISAGG_SOCRATA_ROW)
|
|
assert row["report_date"] == "2026-04-22"
|
|
assert row["producer_long"] == 100000
|
|
assert row["producer_short"] == 300000
|
|
assert row["producer_net"] == -200000
|
|
assert row["swap_long"] == 50000
|
|
assert row["swap_net"] == -10000
|
|
assert row["managed_money_long"] == 200000
|
|
assert row["managed_money_short"] == 80000
|
|
assert row["managed_money_net"] == 120000
|
|
assert row["other_long"] == 10000
|
|
assert row["other_net"] == 0
|
|
assert row["open_interest"] == 1500000
|