61 lines
3 KiB
Python
61 lines
3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
import cohort001_feature_failure_mining_v1 as runner
|
|
|
|
|
|
def test_entry_bar_uses_exported_column_then_raw_ledger_and_rejects_timestamps():
|
|
assert runner.entry_bar({"entry_bar": 4}) == (4, "column.entry_bar")
|
|
assert runner.entry_bar({"raw_ledger_json": '{"entry_index":3}'}) == (3, "raw_ledger_json.entry_index")
|
|
assert runner.entry_bar({"entry_at": "2026-01-01T00:00:00Z"}) == (None, None)
|
|
|
|
|
|
def test_json_compatible_normalizes_numpy_scalars_and_nonfinite_floats():
|
|
payload = runner.json_compatible({
|
|
"integer": np.int64(7),
|
|
"float": np.float32(1.25),
|
|
"boolean": np.bool_(True),
|
|
"nested": [np.float64(np.nan), np.float64(np.inf), float("-inf")],
|
|
})
|
|
|
|
assert payload == {
|
|
"integer": 7,
|
|
"float": pytest.approx(1.25),
|
|
"boolean": True,
|
|
"nested": [None, None, None],
|
|
}
|
|
assert json.loads(json.dumps(payload, allow_nan=False))["nested"] == [None, None, None]
|
|
|
|
|
|
def test_runner_mines_entry_values_and_skips_unmappable_rows(tmp_path, monkeypatch):
|
|
pa = pytest.importorskip("pyarrow")
|
|
pq = pytest.importorskip("pyarrow.parquet")
|
|
engineering = tmp_path / "engineering.json"
|
|
engineering.write_text(json.dumps({"primitives": [{"indicator_id": 1, "period": 10, "p1": 0}]}))
|
|
semantic = tmp_path / "semantic.parquet"
|
|
pq.write_table(pa.Table.from_pylist([{"feature_key": "1:10:0", "domain": "trend", "output_type": "continuous"}]), semantic)
|
|
checkpoints = tmp_path / "checkpoints"
|
|
checkpoints.mkdir()
|
|
np.save(checkpoints / "hs22_1_10_0.npy", np.array([1.0, 2.0, 8.0, 9.0]))
|
|
context = tmp_path / "context.parquet"
|
|
pq.write_table(pa.Table.from_pylist([
|
|
{"failure_label": "GOOD_ENTRY", "entry_bar": 0, "fold": "F1", "feature_triples_json": '[{"indicator_id":1,"period":10,"p1":0}]'},
|
|
{"failure_label": "GOOD_ENTRY", "entry_bar": 1, "fold": "F2", "feature_triples_json": '[{"indicator_id":1,"period":10,"p1":0}]'},
|
|
{"failure_label": "WRONG_DIRECTION", "entry_bar": 2, "fold": "F1", "feature_triples_json": '[{"indicator_id":1,"period":10,"p1":0}]'},
|
|
{"failure_label": "WRONG_DIRECTION", "entry_bar": 3, "fold": "F2", "feature_triples_json": '[{"indicator_id":1,"period":10,"p1":0}]'},
|
|
{"failure_label": "WRONG_DIRECTION", "entry_at": "not a bar", "feature_triples_json": '[]'},
|
|
]), context)
|
|
output = tmp_path / "output"
|
|
monkeypatch.setattr(sys, "argv", ["runner", "--failure-context", str(context), "--semantic-map", str(semantic), "--oracle-checkpoint-dir", str(checkpoints), "--engineering-map", str(engineering), "--output-dir", str(output)])
|
|
runner.main()
|
|
result = json.loads((output / "cohort001_feature_failure_mining_v1.json").read_text())
|
|
item = result["feature_failure_comparisons"][0]
|
|
assert item["median_difference_failure_minus_good"] == pytest.approx(7.0)
|
|
assert item["fold_consistency"] == 1.0
|
|
assert result["skipped"]["unmappable_entry_bar"] == 1
|
|
assert item["attribution"] == "ASSOCIATIVE_NOT_CAUSAL"
|