834 lines
32 KiB
Python
834 lines
32 KiB
Python
"""Unit tests for the read-only backtest harness (issue #648).
|
||
|
||
Covers the PURE aggregation / metric helpers, factored out of the DB code so
|
||
they're testable without a live database:
|
||
- _compute_metrics — signed/abs error %, median bias, MAPE, per-rooms split
|
||
- _errors_summary — bias / MAPE / p25 / p75 of a signed-error list
|
||
- _bucketize_rooms / _rooms_label — 4+ collapse, студия labelling
|
||
- _derive_room_ratios — per-rooms asking→sold ratio, global fallback, guards
|
||
- _apply_ratios + corrected metrics — ratios that cancel a known bias → ~0
|
||
|
||
No DB / network / mocks: these operate on plain lists/tuples.
|
||
|
||
NOTE: importing scripts.backtest_estimator pulls app.services.estimator →
|
||
app.core.config.Settings, which REQUIRES DATABASE_URL. Set a dummy value
|
||
BEFORE importing app modules (same pattern as tests/test_estimator_pure_units.py
|
||
and tests/test_audit_address_mismatch.py).
|
||
"""
|
||
|
||
import os
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
import math
|
||
|
||
import pytest
|
||
|
||
from scripts import backtest_estimator as bt
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# _bucketize_rooms / _rooms_label
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def test_bucketize_studio_and_negative_clamp_to_zero() -> None:
|
||
assert bt._bucketize_rooms(0) == 0
|
||
assert bt._bucketize_rooms(-3) == 0
|
||
|
||
|
||
def test_bucketize_four_plus_collapses() -> None:
|
||
assert bt._bucketize_rooms(4) == 4
|
||
assert bt._bucketize_rooms(5) == 4
|
||
assert bt._bucketize_rooms(9) == 4
|
||
|
||
|
||
def test_bucketize_passthrough_for_one_to_three() -> None:
|
||
assert bt._bucketize_rooms(1) == 1
|
||
assert bt._bucketize_rooms(2) == 2
|
||
assert bt._bucketize_rooms(3) == 3
|
||
|
||
|
||
def test_rooms_label() -> None:
|
||
assert bt._rooms_label(0) == "студия"
|
||
assert bt._rooms_label(1) == "1к"
|
||
assert bt._rooms_label(3) == "3к"
|
||
assert bt._rooms_label(4) == "4+"
|
||
assert bt._rooms_label(7) == "4+"
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# _errors_summary
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def test_errors_summary_empty_returns_all_none() -> None:
|
||
s = bt._errors_summary([])
|
||
assert s["n"] == 0
|
||
assert s["median_bias_pct"] is None
|
||
assert s["mape_pct"] is None
|
||
assert s["p25_pct"] is None
|
||
assert s["p75_pct"] is None
|
||
|
||
|
||
def test_errors_summary_uses_median_abs_for_mape_not_mean() -> None:
|
||
# signed errors with an asymmetric outlier: median |err| (=10) differs
|
||
# sharply from the MEAN |err| (=40). The brief defines MAPE as the MEDIAN
|
||
# absolute error, so we assert the robust median is used.
|
||
signed = [10.0, 10.0, 10.0, 130.0]
|
||
s = bt._errors_summary(signed)
|
||
assert s["mape_pct"] == 10.0 # median(|10,10,10,130|) = 10, not mean 40
|
||
assert s["median_bias_pct"] == 10.0 # median([10,10,10,130]) = 10
|
||
|
||
|
||
def test_errors_summary_signed_bias_can_be_negative() -> None:
|
||
# Under-prediction → negative bias.
|
||
s = bt._errors_summary([-20.0, -10.0, -30.0])
|
||
assert s["median_bias_pct"] == -20.0
|
||
assert s["mape_pct"] == 20.0 # median of |[-20,-10,-30]| = median[10,20,30]
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# _compute_metrics — signed/abs error %, bias, MAPE, per-rooms
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def test_compute_metrics_empty_overall_is_none_per_rooms_all_present() -> None:
|
||
m = bt._compute_metrics([])
|
||
assert m["overall"]["n"] == 0
|
||
assert m["overall"]["median_bias_pct"] is None
|
||
assert m["overall"]["mape_pct"] is None
|
||
assert m["overall"]["n_no_analogs"] == 0
|
||
# Every room bucket must still appear (with n=0) so the report renders.
|
||
assert set(m["per_rooms"].keys()) == set(bt.ROOM_BUCKETS)
|
||
for bucket in bt.ROOM_BUCKETS:
|
||
assert m["per_rooms"][bucket]["n"] == 0
|
||
assert m["per_rooms"][bucket]["median_bias_pct"] is None
|
||
assert m["per_rooms"][bucket]["label"] == bt._rooms_label(bucket)
|
||
|
||
|
||
def test_compute_metrics_known_plus_22_pct_overprediction() -> None:
|
||
# The headline finding: asking median over-predicts SOLD by ~+22%.
|
||
# pred = 1.22 * sold for every row → signed error must be exactly +22%,
|
||
# MAPE +22%, p25 == p75 == +22% (no spread).
|
||
rows = [
|
||
(122_000.0, 100_000.0, 1),
|
||
(244_000.0, 200_000.0, 2),
|
||
(366_000.0, 300_000.0, 3),
|
||
]
|
||
m = bt._compute_metrics(rows)
|
||
assert m["overall"]["n"] == 3
|
||
assert m["overall"]["median_bias_pct"] == pytest.approx(22.0)
|
||
assert m["overall"]["mape_pct"] == pytest.approx(22.0)
|
||
assert m["overall"]["p25_pct"] == pytest.approx(22.0)
|
||
assert m["overall"]["p75_pct"] == pytest.approx(22.0)
|
||
|
||
|
||
def test_compute_metrics_signed_error_formula() -> None:
|
||
# Single row, hand-computed: 100*(150k-120k)/120k = +25.0%.
|
||
m = bt._compute_metrics([(150_000.0, 120_000.0, 2)])
|
||
assert m["overall"]["median_bias_pct"] == pytest.approx(25.0)
|
||
assert m["overall"]["mape_pct"] == pytest.approx(25.0)
|
||
|
||
|
||
def test_compute_metrics_abs_error_distinct_from_signed() -> None:
|
||
# Mixed over/under: signed bias near 0 but MAPE (median |err|) is positive.
|
||
# rows: +50%, -50%, +50%, -50% → median signed in {-50,+50} band,
|
||
# median |err| = 50.
|
||
rows = [
|
||
(150_000.0, 100_000.0, 1), # +50
|
||
(50_000.0, 100_000.0, 1), # -50
|
||
(150_000.0, 100_000.0, 1), # +50
|
||
(50_000.0, 100_000.0, 1), # -50
|
||
]
|
||
m = bt._compute_metrics(rows)
|
||
assert m["overall"]["mape_pct"] == pytest.approx(50.0)
|
||
# signed median of [-50,-50,50,50] = 0.0 (mean of two middles)
|
||
assert m["overall"]["median_bias_pct"] == pytest.approx(0.0)
|
||
|
||
|
||
def test_compute_metrics_per_rooms_split_and_four_plus_collapse() -> None:
|
||
rows = [
|
||
(110_000.0, 100_000.0, 0), # студия: +10
|
||
(130_000.0, 100_000.0, 0), # студия: +30 → median bucket 0 = +20
|
||
(90_000.0, 100_000.0, 2), # 2к: -10
|
||
(200_000.0, 100_000.0, 5), # 4+ (5 collapses): +100
|
||
(300_000.0, 100_000.0, 4), # 4+ : +200 → median bucket 4 = +150
|
||
]
|
||
m = bt._compute_metrics(rows)
|
||
|
||
assert m["per_rooms"][0]["n"] == 2
|
||
assert m["per_rooms"][0]["median_bias_pct"] == pytest.approx(20.0)
|
||
assert m["per_rooms"][0]["label"] == "студия"
|
||
|
||
assert m["per_rooms"][2]["n"] == 1
|
||
assert m["per_rooms"][2]["median_bias_pct"] == pytest.approx(-10.0)
|
||
|
||
# rooms=5 and rooms=4 both land in bucket 4.
|
||
assert m["per_rooms"][4]["n"] == 2
|
||
assert m["per_rooms"][4]["median_bias_pct"] == pytest.approx(150.0)
|
||
assert m["per_rooms"][4]["label"] == "4+"
|
||
|
||
# buckets 1 and 3 had no rows.
|
||
assert m["per_rooms"][1]["n"] == 0
|
||
assert m["per_rooms"][3]["n"] == 0
|
||
|
||
# overall n counts every matched row.
|
||
assert m["overall"]["n"] == 5
|
||
|
||
|
||
def test_compute_metrics_drops_nonpositive_sold() -> None:
|
||
# sold_ppm2 <= 0 cannot be divided → row dropped, not counted, no crash.
|
||
rows = [
|
||
(120_000.0, 0.0, 1), # dropped
|
||
(120_000.0, -5.0, 2), # dropped
|
||
(122_000.0, 100_000.0, 1), # kept → +22
|
||
]
|
||
m = bt._compute_metrics(rows)
|
||
assert m["overall"]["n"] == 1
|
||
assert m["overall"]["median_bias_pct"] == pytest.approx(22.0)
|
||
|
||
|
||
def test_compute_metrics_carries_no_analog_counts() -> None:
|
||
rows = [(122_000.0, 100_000.0, 1)]
|
||
m = bt._compute_metrics(
|
||
rows,
|
||
n_no_analogs=7,
|
||
per_rooms_no_analogs={1: 4, 2: 3},
|
||
)
|
||
assert m["overall"]["n_no_analogs"] == 7
|
||
assert m["per_rooms"][1]["n_no_analogs"] == 4
|
||
assert m["per_rooms"][2]["n_no_analogs"] == 3
|
||
# bucket with no skipped deals defaults to 0.
|
||
assert m["per_rooms"][0]["n_no_analogs"] == 0
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# _derive_room_ratios — per-rooms asking→sold ratio, global fallback, guards.
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def _rows_for_bucket(
|
||
bucket: int, *, n: int, ask: float, sold: float
|
||
) -> list[tuple[float, float, int]]:
|
||
"""n identical (ask, sold, bucket) rows — keeps per-bucket median == ask/sold."""
|
||
return [(ask, sold, bucket) for _ in range(n)]
|
||
|
||
|
||
def test_derive_ratios_per_bucket_exact() -> None:
|
||
# 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.
|
||
rows = _rows_for_bucket(1, n=bt.MIN_BUCKET, ask=100_000.0, sold=80_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)
|
||
assert ratios[1] == pytest.approx(0.80)
|
||
assert ratios[2] == pytest.approx(0.75)
|
||
assert meta["fallback_buckets"] == [] # both buckets were big enough
|
||
assert meta["bucket_n"][1] == bt.MIN_BUCKET
|
||
assert meta["bucket_n"][2] == bt.MIN_BUCKET
|
||
|
||
|
||
def test_derive_ratios_median_not_mean() -> None:
|
||
# A bucket whose ask/sold pairs vary: ratio must use the MEDIAN of each
|
||
# series, not a per-row mean. asks median = 100k, solds median = 90k → 0.9.
|
||
rows = [
|
||
(80_000.0, 60_000.0, 1),
|
||
(100_000.0, 90_000.0, 1), # median row
|
||
(300_000.0, 200_000.0, 1),
|
||
*_rows_for_bucket(1, n=bt.MIN_BUCKET, ask=100_000.0, sold=90_000.0),
|
||
]
|
||
ratios, _ = bt._derive_room_ratios(rows)
|
||
# median ask and median sold are both pinned to 100k/90k by the padding.
|
||
assert ratios[1] == pytest.approx(0.90)
|
||
|
||
|
||
def test_derive_ratios_thin_bucket_uses_global_fallback() -> None:
|
||
# bucket 1 has plenty (own ratio 0.80); bucket 2 has only 1 deal (< MIN) →
|
||
# must inherit the GLOBAL ratio, and be flagged as a fallback bucket.
|
||
rows = [
|
||
*_rows_for_bucket(1, n=bt.MIN_BUCKET, ask=100_000.0, sold=80_000.0),
|
||
(200_000.0, 120_000.0, 2), # lone bucket-2 deal
|
||
]
|
||
ratios, meta = bt._derive_room_ratios(rows)
|
||
assert ratios[1] == pytest.approx(0.80)
|
||
assert 2 in meta["fallback_buckets"]
|
||
# global = median(all sold) / median(all ask). With MIN_BUCKET copies of
|
||
# (100k/80k) plus one (200k/120k), both medians stay at the dense point.
|
||
assert ratios[2] == pytest.approx(meta["global_ratio"])
|
||
assert meta["global_ratio"] is not None
|
||
|
||
|
||
def test_derive_ratios_respects_custom_min_bucket() -> None:
|
||
# With min_bucket=2, a 1-deal bucket falls back; a 2-deal bucket keeps own.
|
||
rows = [
|
||
(100_000.0, 50_000.0, 1), # lone bucket-1 deal → fallback
|
||
(100_000.0, 90_000.0, 2),
|
||
(100_000.0, 90_000.0, 2), # 2 deals → own ratio 0.9
|
||
]
|
||
ratios, meta = bt._derive_room_ratios(rows, min_bucket=2)
|
||
assert 1 in meta["fallback_buckets"]
|
||
assert 2 not in meta["fallback_buckets"]
|
||
assert ratios[2] == pytest.approx(0.90)
|
||
|
||
|
||
def test_derive_ratios_empty_returns_empty_and_safe_meta() -> None:
|
||
ratios, meta = bt._derive_room_ratios([])
|
||
assert ratios == {}
|
||
assert meta["global_ratio"] is None
|
||
assert meta["fallback_buckets"] == []
|
||
assert all(n == 0 for n in meta["bucket_n"].values())
|
||
|
||
|
||
def test_derive_ratios_skips_nonpositive_and_guards_div_by_zero() -> None:
|
||
# pred<=0 or sold<=0 rows are dropped; a bucket left with only bad rows
|
||
# gets neither its own ratio NOR a (here non-existent) global one → omitted,
|
||
# and the function does not raise ZeroDivisionError.
|
||
rows = [
|
||
(0.0, 100_000.0, 1), # pred 0 → dropped (would div-by-zero)
|
||
(100_000.0, 0.0, 1), # sold 0 → dropped
|
||
(-5.0, 100_000.0, 2), # pred <0 → dropped
|
||
]
|
||
ratios, meta = bt._derive_room_ratios(rows)
|
||
assert ratios == {} # nothing valid survived
|
||
assert meta["global_ratio"] is None # no valid pred → no global ratio
|
||
|
||
|
||
def test_derive_ratios_bucket_zero_global_pred_no_ratio() -> None:
|
||
# If a bucket's own pred median is 0 (all-zero preds) it can't form a ratio;
|
||
# with no global fallback either, it must be omitted, not crash.
|
||
rows = [(0.0, 100_000.0, 1) for _ in range(bt.MIN_BUCKET)]
|
||
ratios, meta = bt._derive_room_ratios(rows)
|
||
assert 1 not in ratios
|
||
assert meta["global_ratio"] is None
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# _apply_ratios + corrected metrics — ratios that cancel a known bias → ~0.
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def test_apply_ratios_multiplies_pred_by_bucket_ratio() -> None:
|
||
rows = [(100_000.0, 90_000.0, 1), (200_000.0, 150_000.0, 2)]
|
||
out = bt._apply_ratios(rows, {1: 0.9, 2: 0.75})
|
||
assert out[0] == (pytest.approx(90_000.0), 90_000.0, 1)
|
||
assert out[1] == (pytest.approx(150_000.0), 150_000.0, 2)
|
||
|
||
|
||
def test_apply_ratios_missing_bucket_leaves_pred_unchanged() -> None:
|
||
# bucket 3 absent from the ratios map → identity multiplier (×1.0).
|
||
rows = [(123_456.0, 100_000.0, 3)]
|
||
out = bt._apply_ratios(rows, {1: 0.9})
|
||
assert out[0][0] == pytest.approx(123_456.0)
|
||
|
||
|
||
def test_corrected_metrics_cancel_plus_30_pct_bias_to_zero() -> None:
|
||
# Construct a uniform +30% asking bias (pred = 1.30 * sold) across buckets
|
||
# with enough deals that each bucket forms its OWN ratio. Deriving the ratio
|
||
# in-sample and re-applying it MUST collapse the corrected bias to ~0.
|
||
rows: list[tuple[float, float, int]] = []
|
||
for bucket, sold in ((0, 80_000.0), (1, 100_000.0), (2, 150_000.0)):
|
||
rows += _rows_for_bucket(bucket, n=bt.MIN_BUCKET, ask=1.30 * sold, sold=sold)
|
||
|
||
# sanity: the ASKING block really is +30%.
|
||
asking = bt._compute_metrics(rows)
|
||
assert asking["overall"]["median_bias_pct"] == pytest.approx(30.0)
|
||
|
||
ratios, meta = bt._derive_room_ratios(rows)
|
||
# every per-bucket ratio == 1/1.30 ≈ 0.7692.
|
||
for bucket in (0, 1, 2):
|
||
assert ratios[bucket] == pytest.approx(1.0 / 1.30, rel=1e-6)
|
||
assert meta["fallback_buckets"] == []
|
||
|
||
corrected = bt._compute_metrics(bt._apply_ratios(rows, ratios))
|
||
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)
|
||
for bucket in (0, 1, 2):
|
||
assert corrected["per_rooms"][bucket]["median_bias_pct"] == pytest.approx(0.0, abs=1e-6)
|
||
|
||
|
||
def test_corrected_metrics_global_fallback_cancels_uniform_bias() -> None:
|
||
# Even when buckets are too thin for their OWN ratio, the GLOBAL fallback
|
||
# (uniform +30% here) still cancels the systematic bias to ~0.
|
||
rows = [
|
||
(130_000.0, 100_000.0, 1), # +30
|
||
(260_000.0, 200_000.0, 2), # +30
|
||
(104_000.0, 80_000.0, 0), # +30
|
||
]
|
||
ratios, meta = bt._derive_room_ratios(rows) # all buckets < MIN_BUCKET
|
||
assert set(meta["fallback_buckets"]) == {0, 1, 2}
|
||
# meta["global_ratio"] is rounded to 4dp for the report; the APPLIED ratios
|
||
# in `ratios` keep full precision, so the corrected bias still cancels to 0.
|
||
assert meta["global_ratio"] == pytest.approx(0.7692, abs=5e-5)
|
||
assert ratios[1] == pytest.approx(1.0 / 1.30, rel=1e-9) # full precision applied
|
||
corrected = bt._compute_metrics(bt._apply_ratios(rows, ratios))
|
||
assert corrected["overall"]["median_bias_pct"] == pytest.approx(0.0, abs=1e-6)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Rendering smoke tests — table + empty render must not crash.
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def test_render_table_runs_on_real_metrics() -> None:
|
||
m = bt._compute_metrics([(122_000.0, 100_000.0, 1)], n_no_analogs=2)
|
||
headline = {
|
||
"deal_median_ppm2": 100_000.0,
|
||
"ask_median_ppm2": 122_000.0,
|
||
"spread_pct": 22.0,
|
||
}
|
||
out = bt._render_table(m, headline)
|
||
assert "BACKTEST" in out
|
||
assert "OVERALL" in out
|
||
assert "+22.0" in out # bias rendered with sign
|
||
assert "100 000" in out # ppm2 formatted with space thousands separator
|
||
|
||
|
||
def test_render_table_handles_empty_sample() -> None:
|
||
m = bt._compute_metrics([])
|
||
headline = {"deal_median_ppm2": None, "ask_median_ppm2": None, "spread_pct": None}
|
||
out = bt._render_table(m, headline)
|
||
assert "n/a" in out # None metrics render as n/a, no crash
|
||
|
||
|
||
def test_render_table_includes_corrected_block_and_in_sample_warning() -> None:
|
||
# Build a metrics dict the way run_backtest does, with a corrected block.
|
||
rows = [(130_000.0, 100_000.0, 1)] * bt.MIN_BUCKET
|
||
m = bt._compute_metrics(rows)
|
||
ratios, meta = bt._derive_room_ratios(rows)
|
||
meta["holdout_split"] = False
|
||
m["ratios"] = ratios
|
||
m["ratios_meta"] = meta
|
||
m["corrected"] = bt._compute_metrics(bt._apply_ratios(rows, ratios))
|
||
headline = {
|
||
"deal_median_ppm2": 100_000.0,
|
||
"ask_median_ppm2": 130_000.0,
|
||
"spread_pct": 30.0,
|
||
}
|
||
out = bt._render_table(m, headline)
|
||
assert "ASKING" in out
|
||
assert "CORRECTED" in out
|
||
assert "ratio=" in out # derived ratio line rendered
|
||
assert "IN-SAMPLE" in out # honesty caveat shown when not holdout
|
||
|
||
|
||
def test_render_table_corrected_block_holdout_message() -> None:
|
||
rows = [(130_000.0, 100_000.0, 1)] * bt.MIN_BUCKET
|
||
m = bt._compute_metrics(rows)
|
||
ratios, meta = bt._derive_room_ratios(rows)
|
||
meta["holdout_split"] = True
|
||
m["ratios"] = ratios
|
||
m["ratios_meta"] = meta
|
||
m["corrected"] = bt._compute_metrics(bt._apply_ratios(rows, ratios))
|
||
headline = {"deal_median_ppm2": None, "ask_median_ppm2": None, "spread_pct": None}
|
||
out = bt._render_table(m, headline)
|
||
assert "OUT-OF-SAMPLE" in out
|
||
assert "IN-SAMPLE" not in out # holdout path swaps the caveat
|
||
|
||
|
||
def test_render_table_no_corrected_block_when_absent() -> None:
|
||
# Backward-compat: a metrics dict without "corrected" still renders (ASKING
|
||
# only) and does not raise.
|
||
m = bt._compute_metrics([(122_000.0, 100_000.0, 1)])
|
||
headline = {
|
||
"deal_median_ppm2": 100_000.0,
|
||
"ask_median_ppm2": 122_000.0,
|
||
"spread_pct": 22.0,
|
||
}
|
||
out = bt._render_table(m, headline)
|
||
assert "ASKING" in out
|
||
assert "CORRECTED" not in out
|
||
|
||
|
||
def test_fmt_helpers_handle_none_and_nan_safely() -> None:
|
||
assert bt._fmt_pct(None) == " n/a"
|
||
assert bt._fmt_ppm2(None) == "n/a"
|
||
# sanity: finite values format
|
||
assert "+22" in bt._fmt_pct(22.0)
|
||
assert not math.isnan(22.0)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# argparse — defaults match the brief.
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def test_argparse_defaults() -> None:
|
||
ns = bt._parse_args([])
|
||
assert ns.sample == 300
|
||
assert ns.since == "2025-06-01"
|
||
assert ns.radius == 1000
|
||
assert ns.rooms_tolerance == 0
|
||
assert ns.json is False
|
||
assert ns.holdout_split is False
|
||
|
||
|
||
def test_argparse_overrides() -> None:
|
||
ns = bt._parse_args(
|
||
[
|
||
"--sample",
|
||
"50",
|
||
"--since",
|
||
"2024-01-01",
|
||
"--radius",
|
||
"2000",
|
||
"--rooms-tolerance",
|
||
"1",
|
||
"--holdout-split",
|
||
"--json",
|
||
]
|
||
)
|
||
assert ns.sample == 50
|
||
assert ns.since == "2024-01-01"
|
||
assert ns.radius == 2000
|
||
assert ns.rooms_tolerance == 1
|
||
assert ns.holdout_split 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"])
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# #2002 --resolve-house-id flag + its coverage accumulator.
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def test_argparse_resolve_house_id_defaults_off() -> None:
|
||
# Default OFF keeps the harness byte-identical so the regression gate stands.
|
||
assert bt._parse_args([]).resolve_house_id is False
|
||
|
||
|
||
def test_argparse_resolve_house_id_override() -> None:
|
||
assert bt._parse_args(["--resolve-house-id"]).resolve_house_id is True
|
||
|
||
|
||
def test_house_id_resolution_as_json_shape() -> None:
|
||
res = bt._HouseIdResolution(total=10, resolved=4, imv_reachable=2)
|
||
assert res.as_json() == {"resolved": 4, "total": 10, "imv_reachable": 2}
|
||
|
||
|
||
def test_house_id_resolution_defaults_zero() -> None:
|
||
res = bt._HouseIdResolution()
|
||
assert res.as_json() == {"resolved": 0, "total": 0, "imv_reachable": 0}
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# #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
|