(row: dict)
| 112 | ) |
| 113 | |
| 114 | def predict_2(row: dict) -> int: |
| 115 | vote_a = sig_a.predict(row) |
| 116 | vote_c = sig_c.predict(row) |
| 117 | |
| 118 | activate_b = _should_activate_signal_b(row) |
| 119 | promoted_b = shadow_tracker is not None and shadow_tracker.promoted |
| 120 | if (activate_b or promoted_b) and sig_b is not None and ensemble_3sig is not None: |
| 121 | vote_b = sig_b.predict(row) |
| 122 | result = ensemble_3sig.decide([vote_a, vote_b, vote_c]) |
| 123 | return 1 if result.tier_id is None else result.tier_id |
| 124 | |
| 125 | result = ensemble_2sig.decide([vote_a, vote_c]) |
| 126 | tier_id = 1 if result.tier_id is None else result.tier_id |
| 127 | |
| 128 | # Shadow: run Signal B on rows where it stays inactive so we can |
| 129 | # continue counterfactual tracking. |
| 130 | if sig_b is not None and shadow_tracker is not None and not activate_b: |
| 131 | vote_b = sig_b.predict(row) |
| 132 | gold = row.get("target_tier_id") # available in eval, None in production |
| 133 | shadow_tracker.record( |
| 134 | signal_a_pred=vote_a.tier_id, signal_a_conf=vote_a.confidence, |
| 135 | signal_b_pred=vote_b.tier_id, signal_b_conf=vote_b.confidence, |
| 136 | signal_c_pred=vote_c.tier_id, signal_c_conf=vote_c.confidence, |
| 137 | ensemble_2sig_tier=tier_id, |
| 138 | gold_tier=gold, |
| 139 | ) |
| 140 | |
| 141 | return tier_id |
| 142 | |
| 143 | return predict_2, shadow_tracker |
| 144 |
nothing calls this directly
no test coverage detected