File size: 4,412 Bytes
e610a2f
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
import json

import pytest

from scripts.close_loop_validate import (
    GateConfig,
    ValidationMetrics,
    ValidationRegistry,
    evaluate_promotion_gate,
    is_feature_version_promoted,
)


def _metrics(
    *,
    accuracy: float = 50.0,
    direction_accuracy: float = 48.0,
    signal_count: int = 100,
    buy_signal_win_rate: float = 55.0,
    severe_data_warnings: int = 0,
) -> ValidationMetrics:
    return ValidationMetrics(
        accuracy=accuracy,
        direction_accuracy=direction_accuracy,
        signal_count=signal_count,
        buy_signal_win_rate=buy_signal_win_rate,
        severe_data_warnings=severe_data_warnings,
    )


def test_promotion_gate_passes_when_all_thresholds_clear():
    baseline = _metrics()
    candidate = _metrics(
        accuracy=55.0,
        direction_accuracy=46.0,
        signal_count=70,
        buy_signal_win_rate=55.0,
        severe_data_warnings=0,
    )

    decision = evaluate_promotion_gate(
        baseline,
        candidate,
        GateConfig(
            min_accuracy_delta_pp=5.0,
            min_direction_delta_pp=-2.0,
            min_signal_count_ratio=0.70,
        ),
    )

    assert decision.passed is True
    assert decision.failed_checks == []
    assert decision.deltas["accuracy_delta_pp"] == pytest.approx(5.0)
    assert decision.deltas["direction_accuracy_delta_pp"] == pytest.approx(-2.0)
    assert decision.deltas["buy_signal_win_rate_delta_pp"] == pytest.approx(0.0)


@pytest.mark.parametrize(
    ("candidate_overrides", "failed_check"),
    [
        ({"accuracy": 54.9}, "accuracy_delta_pp"),
        ({"direction_accuracy": 45.9}, "direction_accuracy_delta_pp"),
        ({"signal_count": 69}, "signal_count_ratio"),
        ({"buy_signal_win_rate": 54.9}, "buy_signal_win_rate_delta_pp"),
        ({"severe_data_warnings": 1}, "severe_data_warnings"),
    ],
)
def test_promotion_gate_fails_when_any_required_condition_fails(
    candidate_overrides,
    failed_check,
):
    baseline = _metrics()
    candidate_kwargs = {
        "accuracy": 55.0,
        "direction_accuracy": 46.0,
        "signal_count": 70,
        "buy_signal_win_rate": 55.0,
        "severe_data_warnings": 0,
    }
    candidate_kwargs.update(candidate_overrides)

    decision = evaluate_promotion_gate(
        baseline,
        _metrics(**candidate_kwargs),
        GateConfig(
            min_accuracy_delta_pp=5.0,
            min_direction_delta_pp=-2.0,
            min_signal_count_ratio=0.70,
        ),
    )

    assert decision.passed is False
    assert failed_check in decision.failed_checks


def test_validation_registry_reports_promoted_feature_version(tmp_path):
    registry_path = tmp_path / "validation_registry.json"
    registry_path.write_text(
        json.dumps({"promoted_feature_version": "features_c21_oldwang"}),
        encoding="utf-8",
    )

    registry = ValidationRegistry(registry_path)

    assert registry.get_promoted_feature_version() == "features_c21_oldwang"


def test_validation_registry_rejects_unpromoted_feature_versions(tmp_path):
    registry_path = tmp_path / "validation_registry.json"
    registry_path.write_text(
        json.dumps({"promoted_feature_version": "features_c21_oldwang"}),
        encoding="utf-8",
    )

    registry = ValidationRegistry(registry_path)

    with pytest.raises(ValueError, match="not promoted"):
        registry.assert_promoted_feature_version("features_c22_unvalidated")


def test_missing_registry_fails_closed_for_publishability(tmp_path):
    missing = tmp_path / "missing_registry.json"

    assert not is_feature_version_promoted("institutional-macd-oldwang-v2", missing)


def test_corrupt_registry_fails_closed_for_publishability(tmp_path):
    registry_path = tmp_path / "validation_registry.json"
    registry_path.write_text("{not-json", encoding="utf-8")

    assert not is_feature_version_promoted("institutional-macd-oldwang-v2", registry_path)


def test_registry_requires_promoted_status_pass(tmp_path):
    registry_path = tmp_path / "validation_registry.json"
    registry_path.write_text(
        json.dumps(
            {
                "promoted": {
                    "feature_version": "features_c22_unvalidated",
                    "status": "fail",
                }
            }
        ),
        encoding="utf-8",
    )

    assert not is_feature_version_promoted("features_c22_unvalidated", registry_path)