1# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca2"""`RunSpec.data_dir` (the prospective bundle mechanism): the three parquet loads and the geo cache are redirected to3the given directory; the default layout (data/processed | data/synthetic) is untouched. Synthetic data, mock provider."""4from __future__ import annotations56import json7from pathlib import Path89import polars as pl10import pytest1112from agentic_valuation.config import PROCESSED, SYNTHETIC13from agentic_valuation.experiments.run_arm import RunSpec, data_base, feature_backend, load_analytic, run_arm, select_subsample141516def _bundle(tmp_path: Path, syn_analytic: pl.DataFrame, syn_splits: pl.DataFrame, syn_subsamples: pl.DataFrame) -> tuple[Path, list[str]]:17 """A bundle whose 'prospective' rows are the 6 latest synthetic sales relabelled (the history stays as it is)."""18 d = tmp_path / "bundle"19 d.mkdir()20 pros = syn_analytic.sort(["date", "tx_id"]).tail(6)["tx_id"].to_list()21 splits = syn_splits.with_columns(pl.when(pl.col("tx_id").is_in(pros)).then(pl.lit("prospective")).otherwise(pl.col("split")).alias("split"))22 subs = pl.concat([23 syn_subsamples.filter(~pl.col("tx_id").is_in(pros)),24 pl.DataFrame({"tx_id": sorted(pros), "split": ["prospective"] * 6, "stratum": ["s"] * 6, "draw_seed": [1] * 6}, schema=syn_subsamples.schema),25 ])26 syn_analytic.write_parquet(d / "analytic.parquet")27 splits.write_parquet(d / "splits.parquet")28 subs.write_parquet(d / "llm_subsamples.parquet")29 pl.DataFrame({"tx_id": pros, "flood_mapped": [True] * 6, "in_flood_zone": [False] * 6, "flood_zone_kind": [None] * 6, "transit_stops_within_1km": [3] * 6,30 "dauid": ["x"] * 6, "da_population": [500] * 6, "da_population_density": [1000.0] * 6, "da_median_age": [40.0] * 6, "da_median_income": [50000.0] * 6,31 "da_pct_renters": [30.0] * 6, "cbd_name": ["Montréal"] * 6, "distance_to_cbd_km": [5.0] * 6},32 schema_overrides={"flood_zone_kind": pl.String, "transit_stops_within_1km": pl.Int32}).write_parquet(d / "geo_features.parquet")33 return d, sorted(pros)343536def test_defaults_untouched():37 assert RunSpec().data_dir is None38 assert data_base(False) == PROCESSED and data_base(True) == SYNTHETIC39 assert data_base(False, None) == PROCESSED and data_base(True, None) == SYNTHETIC40 assert data_base(True, Path("/x/y")) == Path("/x/y") and data_base(False, "/x/y") == Path("/x/y")414243def test_data_dir_redirects_loads(tmp_path: Path, syn_analytic, syn_splits, syn_subsamples):44 d, pros = _bundle(tmp_path, syn_analytic, syn_splits, syn_subsamples)45 df = load_analytic(False, d) # synthetic=False: no fallback to the generator, the directory is authoritative46 assert df.height == syn_analytic.height and set(df.filter(pl.col("split") == "prospective")["tx_id"]) == set(pros)47 sample = select_subsample(df, "prospective", 4, 1, False, d)48 assert sample["tx_id"].to_list() == pros[:4] and set(sample["split"]) == {"prospective"}49 fb = feature_backend(False, d)50 assert fb.geo_cache is not None and fb.geo_cache.height == 651 with pytest.raises(FileNotFoundError):52 load_analytic(False, tmp_path / "missing")535455def test_run_arm_with_data_dir(tmp_path: Path, monkeypatch, syn_analytic, syn_splits, syn_subsamples):56 monkeypatch.setenv("AV_PROVIDER_MODE", "mock")57 d, pros = _bundle(tmp_path, syn_analytic, syn_splits, syn_subsamples)58 spec = RunSpec(arm="E_single_llm", committee="mixed_random", split="prospective", n=3, synthetic=False, out_dir=tmp_path / "runs", run_id="t_datadir", data_dir=d)59 run_dir = run_arm(spec)60 preds = pl.read_parquet(run_dir / "predictions.parquet")61 assert sorted(preds["tx_id"].unique().to_list()) == pros[:3] and set(preds["split"]) == {"prospective"}62 m = json.loads((run_dir / "manifest.json").read_text())63 assert m["data_dir"] == str(d) and m["data_source"] == str(d / "analytic.parquet") and m["geo_cache"] == str(d / "geo_features.parquet")64 assert m["split"] == "prospective" and m["n_sampled"] == 3 and m["n_failed_this_session"] == 0 and m["spec"]["data_dir"] == str(d)65 pk = json.loads((run_dir / "packets" / f"{pros[0]}.json").read_text())66 assert "price" not in pk["subject"]676869def test_run_arm_default_manifest_unchanged(tmp_path: Path, monkeypatch):70 monkeypatch.setenv("AV_PROVIDER_MODE", "mock")71 spec = RunSpec(arm="E_single_llm", committee="all_anthropic", split="validation", n=2, synthetic=True, out_dir=tmp_path, run_id="t_default")72 m = json.loads((run_arm(spec) / "manifest.json").read_text())73 assert m["data_dir"] is None and m["data_source"].startswith("data/synthetic/") and m["spec"]["data_dir"] is None74