feat(tradein/backtest): full-spine prediction via _price_from_inputs + range-coverage/calibration/segment metrics (#1966)

This commit is contained in:
bot-backend 2026-06-27 14:06:28 +03:00
parent 050dd73c93
commit 28e36c18e2

View file

@ -217,9 +217,8 @@ def _rows_for_bucket(
def test_derive_ratios_per_bucket_exact() -> None: def test_derive_ratios_per_bucket_exact() -> None:
# Each bucket ≥ MIN_BUCKET deals so every bucket gets its OWN ratio. # Each bucket ≥ MIN_BUCKET deals so every bucket gets its OWN ratio.
# bucket 1: sold/ask = 80k/100k = 0.80 ; bucket 2: 150k/200k = 0.75. # bucket 1: sold/ask = 80k/100k = 0.80 ; bucket 2: 150k/200k = 0.75.
rows = ( rows = _rows_for_bucket(1, n=bt.MIN_BUCKET, ask=100_000.0, sold=80_000.0) + _rows_for_bucket(
_rows_for_bucket(1, n=bt.MIN_BUCKET, ask=100_000.0, sold=80_000.0) 2, n=bt.MIN_BUCKET, ask=200_000.0, sold=150_000.0
+ _rows_for_bucket(2, n=bt.MIN_BUCKET, ask=200_000.0, sold=150_000.0)
) )
ratios, meta = bt._derive_room_ratios(rows) ratios, meta = bt._derive_room_ratios(rows)
assert ratios[1] == pytest.approx(0.80) assert ratios[1] == pytest.approx(0.80)
@ -328,9 +327,7 @@ def test_corrected_metrics_cancel_plus_30_pct_bias_to_zero() -> None:
# in-sample and re-applying it MUST collapse the corrected bias to ~0. # in-sample and re-applying it MUST collapse the corrected bias to ~0.
rows: list[tuple[float, float, int]] = [] rows: list[tuple[float, float, int]] = []
for bucket, sold in ((0, 80_000.0), (1, 100_000.0), (2, 150_000.0)): for bucket, sold in ((0, 80_000.0), (1, 100_000.0), (2, 150_000.0)):
rows += _rows_for_bucket( rows += _rows_for_bucket(bucket, n=bt.MIN_BUCKET, ask=1.30 * sold, sold=sold)
bucket, n=bt.MIN_BUCKET, ask=1.30 * sold, sold=sold
)
# sanity: the ASKING block really is +30%. # sanity: the ASKING block really is +30%.
asking = bt._compute_metrics(rows) asking = bt._compute_metrics(rows)
@ -346,9 +343,7 @@ def test_corrected_metrics_cancel_plus_30_pct_bias_to_zero() -> None:
assert corrected["overall"]["median_bias_pct"] == pytest.approx(0.0, abs=1e-6) assert corrected["overall"]["median_bias_pct"] == pytest.approx(0.0, abs=1e-6)
assert corrected["overall"]["mape_pct"] == pytest.approx(0.0, abs=1e-6) assert corrected["overall"]["mape_pct"] == pytest.approx(0.0, abs=1e-6)
for bucket in (0, 1, 2): for bucket in (0, 1, 2):
assert corrected["per_rooms"][bucket]["median_bias_pct"] == pytest.approx( assert corrected["per_rooms"][bucket]["median_bias_pct"] == pytest.approx(0.0, abs=1e-6)
0.0, abs=1e-6
)
def test_corrected_metrics_global_fallback_cancels_uniform_bias() -> None: def test_corrected_metrics_global_fallback_cancels_uniform_bias() -> None:
@ -469,8 +464,18 @@ def test_argparse_defaults() -> None:
def test_argparse_overrides() -> None: def test_argparse_overrides() -> None:
ns = bt._parse_args( ns = bt._parse_args(
["--sample", "50", "--since", "2024-01-01", "--radius", "2000", [
"--rooms-tolerance", "1", "--holdout-split", "--json"] "--sample",
"50",
"--since",
"2024-01-01",
"--radius",
"2000",
"--rooms-tolerance",
"1",
"--holdout-split",
"--json",
]
) )
assert ns.sample == 50 assert ns.sample == 50
assert ns.since == "2024-01-01" assert ns.since == "2024-01-01"
@ -478,3 +483,328 @@ def test_argparse_overrides() -> None:
assert ns.rooms_tolerance == 1 assert ns.rooms_tolerance == 1
assert ns.holdout_split is True assert ns.holdout_split is True
assert ns.json is True assert ns.json is True
# --------------------------------------------------------------------------- #
# #1966 full spine — --engine flag.
# --------------------------------------------------------------------------- #
def test_argparse_engine_defaults_to_full() -> None:
assert bt._parse_args([]).engine == "full"
def test_argparse_engine_asking_core_override() -> None:
assert bt._parse_args(["--engine", "asking-core"]).engine == "asking-core"
assert bt._parse_args(["--engine", "full"]).engine == "full"
def test_argparse_engine_rejects_unknown() -> None:
with pytest.raises(SystemExit):
bt._parse_args(["--engine", "bogus"])
# --------------------------------------------------------------------------- #
# #1966 _bucketize_confidence / _segment_label — pure bucketing.
# --------------------------------------------------------------------------- #
def test_bucketize_confidence_canonical_passthrough() -> None:
assert bt._bucketize_confidence("high") == "high"
assert bt._bucketize_confidence("medium") == "medium"
assert bt._bucketize_confidence("low") == "low"
def test_bucketize_confidence_unknown_maps_to_other() -> None:
assert bt._bucketize_confidence("weird") == "other"
assert bt._bucketize_confidence("") == "other"
def test_segment_label_bands_and_boundaries() -> None:
# Boundaries are upper-exclusive: 120k falls into комфорт, not эконом.
assert bt._segment_label(100_000) == "эконом"
assert bt._segment_label(119_999) == "эконом"
assert bt._segment_label(120_000) == "комфорт"
assert bt._segment_label(150_000) == "комфорт"
assert bt._segment_label(160_000) == "бизнес"
assert bt._segment_label(219_999) == "бизнес"
assert bt._segment_label(220_000) == "элит"
assert bt._segment_label(300_000) == "премиум"
assert bt._segment_label(2_000_000) == "премиум" # +inf tail catches the top
# --------------------------------------------------------------------------- #
# #1966 _segment_metrics — per-price-segment signed error (band by SOLD ppm²).
# --------------------------------------------------------------------------- #
def test_segment_metrics_buckets_by_sold_price() -> None:
rows = [
(110_000.0, 100_000.0), # sold эконом, +10
(132_000.0, 110_000.0), # sold эконом, +20 → median эконом bias +15
(165_000.0, 150_000.0), # sold комфорт, +10
(330_000.0, 300_000.0), # sold премиум, +10
]
seg = bt._segment_metrics(rows)
assert set(seg.keys()) == {label for label, _ in bt.PRICE_SEGMENTS_PPM2}
assert seg["эконом"]["n"] == 2
assert seg["эконом"]["median_bias_pct"] == pytest.approx(15.0)
assert seg["комфорт"]["n"] == 1
assert seg["комфорт"]["median_bias_pct"] == pytest.approx(10.0)
assert seg["премиум"]["n"] == 1
# bands with no rows are still present with n=0.
assert seg["бизнес"]["n"] == 0
assert seg["элит"]["n"] == 0
def test_segment_metrics_drops_nonpositive_sold() -> None:
seg = bt._segment_metrics([(100_000.0, 0.0), (100_000.0, -1.0)])
assert all(seg[label]["n"] == 0 for label, _ in bt.PRICE_SEGMENTS_PPM2)
# --------------------------------------------------------------------------- #
# #1966 _range_coverage — inside / outside / boundary inclusive.
# --------------------------------------------------------------------------- #
def test_range_coverage_inside_outside_and_boundary_inclusive() -> None:
rows = [
(100.0, 90.0, 110.0), # inside
(80.0, 90.0, 110.0), # below low → outside
(120.0, 90.0, 110.0), # above high → outside
(90.0, 90.0, 110.0), # exactly on low → covered (inclusive)
(110.0, 90.0, 110.0), # exactly on high → covered (inclusive)
]
cov = bt._range_coverage(rows)
assert cov["n"] == 5
assert cov["n_covered"] == 3
assert cov["coverage_pct"] == pytest.approx(60.0)
def test_range_coverage_empty_returns_none_pct() -> None:
cov = bt._range_coverage([])
assert cov["n"] == 0
assert cov["n_covered"] == 0
assert cov["coverage_pct"] is None
def test_range_coverage_full_and_zero() -> None:
assert bt._range_coverage([(100.0, 50.0, 150.0)])["coverage_pct"] == pytest.approx(100.0)
assert bt._range_coverage([(10.0, 50.0, 150.0)])["coverage_pct"] == pytest.approx(0.0)
# --------------------------------------------------------------------------- #
# #1966 _sharpness — median relative range width (guards coverage gaming).
# --------------------------------------------------------------------------- #
def test_sharpness_median_relative_width() -> None:
rows = [
(100.0, 90.0, 110.0), # width 20 / point 100 = 0.20
(200.0, 150.0, 250.0), # width 100 / point 200 = 0.50
]
sh = bt._sharpness(rows)
assert sh["n"] == 2
assert sh["median_rel_width"] == pytest.approx(0.35) # median(0.20, 0.50)
def test_sharpness_drops_nonpositive_point_and_empty() -> None:
assert bt._sharpness([(0.0, 1.0, 2.0)])["median_rel_width"] is None
assert bt._sharpness([(-5.0, 1.0, 2.0)])["n"] == 0
assert bt._sharpness([])["median_rel_width"] is None
# --------------------------------------------------------------------------- #
# #1966 _calibration_metrics — per-confidence n / coverage% / MAPE%.
# --------------------------------------------------------------------------- #
def test_calibration_metrics_per_confidence_n_coverage_mape() -> None:
rows = [
("high", 5.0, True),
("high", 15.0, True), # high: n=2, covered 2/2=100%, mape median(5,15)=10
("low", 40.0, False),
("low", 60.0, True), # low: n=2, covered 1/2=50%, mape median(40,60)=50
]
cal = bt._calibration_metrics(rows)
# canonical buckets always present.
assert set(("high", "medium", "low")).issubset(cal.keys())
assert cal["high"]["n"] == 2
assert cal["high"]["coverage_pct"] == pytest.approx(100.0)
assert cal["high"]["mape_pct"] == pytest.approx(10.0)
assert cal["low"]["n"] == 2
assert cal["low"]["coverage_pct"] == pytest.approx(50.0)
assert cal["low"]["mape_pct"] == pytest.approx(50.0)
# empty canonical bucket renders with n=0 / None metrics, not missing.
assert cal["medium"]["n"] == 0
assert cal["medium"]["coverage_pct"] is None
assert cal["medium"]["mape_pct"] is None
def test_calibration_metrics_handles_none_signed_and_covered() -> None:
# A prediction with no expected_sold (signed None) and no range (covered None)
# still counts toward n but not toward coverage/MAPE.
rows: list[tuple[str, float | None, bool | None]] = [
("high", None, None),
("high", 10.0, True),
]
cal = bt._calibration_metrics(rows)
assert cal["high"]["n"] == 2
assert cal["high"]["coverage_pct"] == pytest.approx(100.0) # only the 1 with covered
assert cal["high"]["mape_pct"] == pytest.approx(10.0) # only the 1 with signed
def test_calibration_metrics_appends_other_bucket() -> None:
cal = bt._calibration_metrics([("exotic", 5.0, True)])
assert "other" in cal
assert cal["other"]["n"] == 1
# canonical three still present even though empty.
assert cal["high"]["n"] == 0
# --------------------------------------------------------------------------- #
# #1966 Prediction + _compute_full_metrics — integration of the new blocks.
# --------------------------------------------------------------------------- #
def _pred(
*,
rooms: int = 2,
area: float = 50.0,
sold_ppm2: float = 100_000.0,
median_ppm2: float = 120_000.0,
confidence: str = "high",
es_ppm2: float | None = 100_000.0,
es_price: float | None = 5_000_000.0,
range_low: float | None = 4_500_000.0,
range_high: float | None = 5_500_000.0,
anchor_tier: str | None = None,
deal_id: int = 1,
) -> bt.Prediction:
return bt.Prediction(
deal_id=deal_id,
rooms=rooms,
area_m2=area,
sold_ppm2=sold_ppm2,
median_ppm2=median_ppm2,
confidence=confidence,
anchor_tier=anchor_tier,
expected_sold_ppm2=es_ppm2,
expected_sold_price=es_price,
range_low=range_low,
range_high=range_high,
)
def test_prediction_sold_total_property() -> None:
p = _pred(sold_ppm2=100_000.0, area=50.0)
assert p.sold_total == pytest.approx(5_000_000.0)
def test_compute_full_metrics_structure_and_blocks() -> None:
preds = [
# sold_total = 100k*50 = 5.0M, range [4.5M, 5.5M] → covered; es +0%
_pred(deal_id=1, confidence="high", es_ppm2=100_000.0, sold_ppm2=100_000.0),
# sold_total = 200k*50 = 10.0M, range [4.5M,5.5M] → NOT covered; es -50%
_pred(
deal_id=3,
confidence="low",
es_ppm2=100_000.0,
sold_ppm2=200_000.0,
median_ppm2=120_000.0,
),
]
m = bt._compute_full_metrics(preds, n_no_prediction=4, per_rooms_no_prediction={2: 4})
# expected_sold block: overall + per_rooms + per_segment, carries no-pred count.
assert m["expected_sold"]["overall"]["n"] == 2
assert m["expected_sold"]["overall"]["n_no_analogs"] == 4 # repurposed = no_pred
assert "per_segment" in m["expected_sold"]
assert set(m["expected_sold"]["per_segment"].keys()) == {
label for label, _ in bt.PRICE_SEGMENTS_PPM2
}
# range coverage: 1 of 2 inside → 50% overall.
assert m["range_coverage"]["overall"]["n"] == 2
assert m["range_coverage"]["overall"]["n_covered"] == 1
assert m["range_coverage"]["overall"]["coverage_pct"] == pytest.approx(50.0)
# per-confidence: high covered 100%, low covered 0%.
assert m["range_coverage"]["per_confidence"]["high"]["coverage_pct"] == pytest.approx(100.0)
assert m["range_coverage"]["per_confidence"]["low"]["coverage_pct"] == pytest.approx(0.0)
# calibration: high tighter/accurate, low not.
assert m["calibration"]["high"]["n"] == 1
assert m["calibration"]["high"]["coverage_pct"] == pytest.approx(100.0)
assert m["calibration"]["high"]["mape_pct"] == pytest.approx(0.0)
assert m["calibration"]["low"]["coverage_pct"] == pytest.approx(0.0)
assert m["calibration"]["low"]["mape_pct"] == pytest.approx(50.0)
# sharpness present.
assert m["sharpness"]["n"] == 2
assert m["sharpness"]["median_rel_width"] is not None
# confidence order: canonical three first.
assert m["confidence_order"][:3] == ["high", "medium", "low"]
def test_compute_full_metrics_empty_is_safe() -> None:
m = bt._compute_full_metrics([])
assert m["expected_sold"]["overall"]["n"] == 0
assert m["range_coverage"]["overall"]["coverage_pct"] is None
assert m["calibration"]["high"]["n"] == 0
assert m["sharpness"]["median_rel_width"] is None
def test_compute_full_metrics_no_expected_sold_counts_in_calibration_only() -> None:
# A priced deal with no expected_sold (ratio unresolved) and no range:
# counts in calibration n but contributes nothing to expected_sold / coverage.
preds = [
_pred(
deal_id=1,
confidence="medium",
es_ppm2=None,
es_price=None,
range_low=None,
range_high=None,
)
]
m = bt._compute_full_metrics(preds)
assert m["expected_sold"]["overall"]["n"] == 0 # no es row
assert m["range_coverage"]["overall"]["n"] == 0 # no range row
assert m["calibration"]["medium"]["n"] == 1 # still counted
assert m["calibration"]["medium"]["coverage_pct"] is None
# --------------------------------------------------------------------------- #
# #1966 _render_full_table — smoke (must not crash, renders all blocks).
# --------------------------------------------------------------------------- #
def test_render_full_table_runs_on_real_metrics() -> None:
preds = [
_pred(deal_id=1, confidence="high", es_ppm2=100_000.0, sold_ppm2=100_000.0),
_pred(deal_id=3, confidence="low", es_ppm2=100_000.0, sold_ppm2=200_000.0),
]
m = bt._compute_full_metrics(preds, n_no_prediction=1, per_rooms_no_prediction={2: 1})
m["headline"] = {
"deal_median_ppm2": 100_000.0,
"ask_median_ppm2": 120_000.0,
"spread_pct": 20.0,
}
out = bt._render_full_table(m)
assert "full spine" in out
assert "EXPECTED_SOLD" in out
assert "RANGE COVERAGE" in out
assert "CONFIDENCE CALIBRATION" in out
assert "SHARPNESS" in out
assert "per price-segment" in out
assert "эконом" in out # segment band rendered
assert "regression baseline" in out # caveat present
def test_render_full_table_handles_empty_sample() -> None:
m = bt._compute_full_metrics([])
m["headline"] = {"deal_median_ppm2": None, "ask_median_ppm2": None, "spread_pct": None}
out = bt._render_full_table(m)
assert "n/a" in out # None metrics render as n/a, no crash
assert "BACKTEST" in out