Conviction campaign 2026-09-27 (FAILED, research knob kept) + plain-words alert wording
Owner question: can score and confidence be combined into one number that is easier to understand and for the system to decide on? Research (pre-registered, nothing ships to the signal path): - signals.grading_mode: "pair" (default, the shipped two-floor rule, byte- identical: all five reference MD5s reproduce on both scripts), "min" (min(|score|, 0.6*conf)), "product" (|score|*conf/100); thresholds fixed from the control's corner points, no tuning. Helpers _final_signal_and_grade / _transitioning_boost / conviction / CONVICTION_THRESHOLDS in signal_aggregator; mirrored in K2 (GRADING_MODE_CODES, PF_conv_* packed from the live module); twin refuses other values; M15 OOS MD5 equal on both scripts for both modes. - Verdict: FAILED on M5 and M15. K-MIN moves 43-54 trades for -1.2R / -2.9R (clauses a,b,e / a,b,d); K-PROD -29R on M15 (PF 1.29, W3 DD 28.9R), neutral on M5. H1/H4 product +11.9R / +17.0R on W3 from extra low-score trades (H1 PF 1.98 -> 1.75, DD doubles): hypothesis only. - Code review (high) before the runs: 6 findings fixed, 1 recorded (engine_config_fingerprint omits the switch under "pair", so live rows keep their fingerprint). Delivery (books untouched): alerts lead with the grade and print the two raw numbers in words (main.plain_words: "Direction: strong buy · Reliability: good · timeframes agree", cut-offs = the grade table's own) on Telegram, /status, the CLI and the dashboard card (plain_words in the snapshot); "MTF" label replaced by "Other timeframes"; guides updated. Suite 566 passed. Evidence: backtest/research/conviction_campaign_2026-09-27.md Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Harwinder Sekhon
+252
-1
@@ -1310,6 +1310,158 @@ def _():
|
||||
assert report.action_recommendation != ""
|
||||
|
||||
|
||||
# ── grading_mode (research switch, campaign 2026-09-27) ──────────────
|
||||
def _grading_report():
|
||||
config = PlatformConfig()
|
||||
agg = SignalAggregator(config)
|
||||
dfs = {
|
||||
Timeframe.H1: make_ohlcv(300, seed=42),
|
||||
Timeframe.M15: make_ohlcv(300, seed=43),
|
||||
Timeframe.H4: make_ohlcv(300, seed=44),
|
||||
}
|
||||
return config, agg, agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||||
|
||||
|
||||
@test("SignalConfig.grading_mode defaults to 'pair'; _signal_pair/_quality_pair reproduce the CLAUDE.md §2 table")
|
||||
def _():
|
||||
from config import SignalConfig as _SC
|
||||
from signal_aggregator import _signal_pair, _quality_pair, _transitioning_boost
|
||||
assert _SC().grading_mode == "pair"
|
||||
|
||||
def table_signal(s, c):
|
||||
if s >= 35 and c >= 55: return SignalStrength.STRONG_BUY
|
||||
if s >= 25 and c >= 50: return SignalStrength.BUY
|
||||
if s <= -35 and c >= 55: return SignalStrength.STRONG_SELL
|
||||
if s <= -25 and c >= 50: return SignalStrength.SELL
|
||||
return SignalStrength.NEUTRAL
|
||||
|
||||
def table_grade(a, c, m):
|
||||
if a >= 40 and c >= 65 and m: return "A+"
|
||||
if a >= 33 and c >= 55 and m: return "A"
|
||||
if a >= 28 and c >= 50: return "B"
|
||||
if a >= 25 and c >= 45: return "C"
|
||||
if a >= 15: return "D"
|
||||
return "F"
|
||||
|
||||
edges = []
|
||||
for b in (15, 25, 28, 30, 33, 35, 40):
|
||||
edges += [b - 0.01, b, b + 0.01]
|
||||
scores = sorted({0.0} | {e for e in edges} | {-e for e in edges} | {-100.0, 100.0, 60.0, -60.0})
|
||||
confs = [0.0, 44.99, 45.0, 49.99, 50.0, 54.99, 55.0, 64.99, 65.0, 95.0]
|
||||
n = 0
|
||||
for s in scores:
|
||||
for c in confs:
|
||||
assert _signal_pair(s, c) == table_signal(s, c), (s, c)
|
||||
for m in (True, False):
|
||||
assert _quality_pair(abs(s), c, m) == table_grade(abs(s), c, m), (s, c, m)
|
||||
want = "A" if (abs(s) >= 30 and c >= 50 and m) else "B" if (abs(s) >= 25 and c >= 45) else None
|
||||
assert _transitioning_boost(abs(s), c, m, "pair") == want, (s, c, m)
|
||||
n += 1
|
||||
assert n > 800
|
||||
# the real report's grade is the helper of its own final values
|
||||
config, agg, report = _grading_report()
|
||||
assert report.signal_quality == _quality_pair(abs(report.final_score), report.final_confidence, report.mtf_agreement)
|
||||
assert report.final_signal == _signal_pair(report.final_score, report.final_confidence)
|
||||
|
||||
|
||||
@test("grading_mode='min': conviction = min(|score|, 0.6*conf); grades and signal at the boundaries")
|
||||
def _():
|
||||
from signal_aggregator import conviction, _signal_conv, _quality_conv, _transitioning_boost, CONVICTION_THRESHOLDS
|
||||
conv = lambda s, c: conviction(abs(s), c, "min")
|
||||
grade = lambda s, c, m=True: _quality_conv(abs(s), conv(s, c), m, "min")
|
||||
sig = lambda s, c: _signal_conv(s, conv(s, c), "min")
|
||||
assert conv(40, 66.7) >= 40 and conv(40, 66.6) < 40
|
||||
assert conv(95, 95) == 0.6 * 95 and conv(10, 95) == 10 and conv(50, 95) == 50 # min picks the smaller axis
|
||||
assert grade(40, 66.7) == "A+" and grade(40, 66.6) == "A" and grade(40, 66.7, False) == "B"
|
||||
assert grade(28, 46.7) == "B" and grade(28, 46.6) == "C"
|
||||
assert grade(25, 41.7) == "C" and grade(25, 41.6) == "D"
|
||||
assert grade(14.9, 95) == "F" and grade(15, 0) == "D"
|
||||
assert sig(35, 58.4) == SignalStrength.STRONG_BUY and sig(35, 58.3) == SignalStrength.BUY
|
||||
assert sig(-35, 58.4) == SignalStrength.STRONG_SELL and sig(-25, 41.7) == SignalStrength.SELL
|
||||
assert sig(0, 95) == SignalStrength.NEUTRAL and sig(60, 0) == SignalStrength.NEUTRAL
|
||||
assert _transitioning_boost(26, 60, False, "min") == "B"
|
||||
assert _transitioning_boost(30, 50, True, "min") == "A" and _transitioning_boost(30, 49.9, True, "min") == "B"
|
||||
assert _transitioning_boost(24, 95, True, "min") is None
|
||||
assert CONVICTION_THRESHOLDS["min"] == (40.0, 33.0, 28.0, 25.0, 35.0, 25.0, 30.0, 25.0)
|
||||
# the real report under the mode: grade == helper of the recomputed final values
|
||||
config, agg, report = _grading_report()
|
||||
config.signals.grading_mode = "min"
|
||||
agg._compute_final_signal(report)
|
||||
assert report.signal_quality == _quality_conv(abs(report.final_score), conv(report.final_score, report.final_confidence),
|
||||
report.mtf_agreement, "min")
|
||||
|
||||
|
||||
@test("grading_mode='product': conviction = |score|*conf/100; the control corners reproduce; high confidence can lift a mid score")
|
||||
def _():
|
||||
from signal_aggregator import conviction, _signal_conv, _quality_conv, _transitioning_boost, CONVICTION_THRESHOLDS
|
||||
conv = lambda s, c: conviction(abs(s), c, "product")
|
||||
grade = lambda s, c, m=True: _quality_conv(abs(s), conv(s, c), m, "product")
|
||||
sig = lambda s, c: _signal_conv(s, conv(s, c), "product")
|
||||
assert conv(40, 65) == 26.0 and conv(33, 55) == 18.15 and conv(28, 50) == 14.0 and conv(25, 45) == 11.25
|
||||
assert grade(40, 65) == "A+" and grade(33, 55) == "A" and grade(28, 50) == "B" and grade(25, 45) == "C"
|
||||
assert grade(40, 64.9) == "A" and grade(33, 54.9) == "B"
|
||||
assert sig(35, 55) == SignalStrength.STRONG_BUY and sig(35, 54.9) == SignalStrength.BUY
|
||||
assert sig(25, 50) == SignalStrength.BUY and sig(-25, 50) == SignalStrength.SELL and sig(25, 49.9) == SignalStrength.NEUTRAL
|
||||
assert grade(30, 87) == "A+" # |score| < 40 lifted by confidence
|
||||
assert grade(24, 95) == "A" and grade(24, 95, False) == "B"
|
||||
assert grade(20, 60) == "C" # |score| < 25: the M5 gate-1 interaction case
|
||||
assert grade(20, 56) == "D" and grade(60, 0) == "D" # 11.2 < 11.25; conf 0 -> conviction 0
|
||||
assert grade(14.9, 95) == "B" and grade(10, 95) == "F" # 14.155 >= 14 (the trade-off); 9.5 < 11.25 and |score| < 15
|
||||
assert _transitioning_boost(30, 50, True, "product") == "A" and _transitioning_boost(25, 45, False, "product") == "B"
|
||||
assert _transitioning_boost(25, 44.9, False, "product") is None
|
||||
assert CONVICTION_THRESHOLDS["product"] == (26.0, 18.15, 14.0, 11.25, 19.25, 12.5, 15.0, 11.25)
|
||||
config, agg, report = _grading_report()
|
||||
config.signals.grading_mode = "product"
|
||||
agg._compute_final_signal(report)
|
||||
assert report.signal_quality == _quality_conv(abs(report.final_score), conv(report.final_score, report.final_confidence),
|
||||
report.mtf_agreement, "product")
|
||||
|
||||
|
||||
@test("grading_mode: gate 6c boost reads the same conviction in every mode")
|
||||
def _():
|
||||
from signal_aggregator import _transitioning_boost
|
||||
for mode in ("pair", "min", "product"):
|
||||
assert _transitioning_boost(30, 50, True, mode) == "A", mode
|
||||
assert _transitioning_boost(30, 50, False, mode) == "B", mode
|
||||
assert _transitioning_boost(26, 60, False, mode) == "B", mode
|
||||
assert _transitioning_boost(10, 95, True, mode) is None, mode
|
||||
assert _transitioning_boost(26, 44, False, "pair") is None
|
||||
assert _transitioning_boost(26, 44, False, "min") == "B" # 0.6*44 = 26.4 >= 25
|
||||
assert _transitioning_boost(26, 44, False, "product") == "B" # 11.44 >= 11.25
|
||||
|
||||
|
||||
@test("grading_mode unknown raises at construction and at recompute; the fingerprint carries the field")
|
||||
def _():
|
||||
from signal_aggregator import grading_mode_of
|
||||
config = PlatformConfig()
|
||||
config.signals.grading_mode = "geomean"
|
||||
try:
|
||||
SignalAggregator(config)
|
||||
raise AssertionError("an unknown grading_mode must fail at construction")
|
||||
except ValueError as e:
|
||||
assert "grading_mode" in str(e) and "geomean" in str(e), e
|
||||
config, agg, report = _grading_report()
|
||||
fp_pair = agg.engine_config_fingerprint()
|
||||
config.signals.grading_mode = "geomean"
|
||||
try:
|
||||
agg._compute_final_signal(report)
|
||||
raise AssertionError("an unknown grading_mode must fail at recompute")
|
||||
except ValueError:
|
||||
pass
|
||||
config.signals.grading_mode = "product"
|
||||
assert agg.engine_config_fingerprint() != fp_pair, "the mode is part of the scoring config"
|
||||
config.signals.grading_mode = "pair"
|
||||
assert agg.engine_config_fingerprint() == fp_pair
|
||||
assert grading_mode_of(config.signals) == "pair"
|
||||
from signal_aggregator import conviction as _conv
|
||||
for bad in ("pair", "geomean", ""):
|
||||
try:
|
||||
_conv(30.0, 60.0, bad)
|
||||
raise AssertionError(f"conviction() must refuse {bad!r}")
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
|
||||
@test("Event-risk signal gating obeys news.event_risk_caps_signals")
|
||||
def _():
|
||||
config = PlatformConfig()
|
||||
@@ -1728,11 +1880,50 @@ def _():
|
||||
assert isinstance(msg, str)
|
||||
assert len(msg) > 0
|
||||
assert "XAU/USD" in msg
|
||||
assert "Score" in msg
|
||||
assert "Score *" not in msg and "Conf *" not in msg and "MTF ✅" not in msg and "MTF ❌" not in msg, \
|
||||
"raw numbers replaced by words (2026-09-27)"
|
||||
assert "📈 MTF:" not in msg, "the per-timeframe detail line is labelled in words"
|
||||
assert "📐 Direction: *" in msg and "Reliability: *" in msg and "timeframes " in msg
|
||||
import telegram_bot as _tb
|
||||
assert msg.splitlines()[0].startswith(f"{_tb._SIGNAL_EMOJI[report.final_signal]} *Grade {report.signal_quality} ·"), "grade first"
|
||||
assert "Regime" in msg
|
||||
assert "Not financial advice" in msg
|
||||
|
||||
|
||||
@test("plain words for score / confidence sit on the grade table's floors and reach every surface")
|
||||
def _():
|
||||
from main import direction_words, reliability_words, plain_words, report_to_dict
|
||||
assert direction_words(40) == "very strong buy" and direction_words(39.99) == "strong buy"
|
||||
assert direction_words(-33) == "strong sell" and direction_words(-32.99) == "clear sell"
|
||||
assert direction_words(25) == "clear buy" and direction_words(24.99) == "weak buy"
|
||||
assert direction_words(15) == "weak buy" and direction_words(14.99) == "no clear direction"
|
||||
assert direction_words(0) == "no clear direction"
|
||||
assert reliability_words(65) == "high" and reliability_words(64.99) == "good"
|
||||
assert reliability_words(55) == "good" and reliability_words(54.99) == "fair"
|
||||
assert reliability_words(45) == "fair" and reliability_words(44.99) == "low" and reliability_words(0) == "low"
|
||||
report = SignalReport(symbol="XAU/USD", display_name="XAU/USD", timestamp="now")
|
||||
report.final_score, report.final_confidence, report.mtf_agreement = 38.2, 62.0, True
|
||||
assert plain_words(report) == "Direction: strong buy · Reliability: good · timeframes agree"
|
||||
report.mtf_agreement = False
|
||||
assert plain_words(report).endswith("timeframes disagree")
|
||||
assert report_to_dict(report)["plain_words"] == plain_words(report)
|
||||
# the words are consistent with the grade the same numbers would earn
|
||||
from signal_aggregator import _quality_pair
|
||||
for score, conf in ((40, 65), (33, 55), (28, 50), (25, 45), (15, 10), (5, 90)):
|
||||
grade = _quality_pair(score, conf, True)
|
||||
words = direction_words(score)
|
||||
if grade == "A+":
|
||||
assert words.startswith("very strong")
|
||||
elif grade == "A":
|
||||
assert words.startswith("strong")
|
||||
elif grade in ("B", "C"):
|
||||
assert words.startswith("clear"), (score, conf, words)
|
||||
elif grade == "D":
|
||||
assert words.startswith("weak")
|
||||
else:
|
||||
assert words == "no clear direction"
|
||||
|
||||
|
||||
@test("format_signal_message makes chart pattern actions Markdown-safe")
|
||||
def _():
|
||||
report = SignalReport(symbol="XAU/USD", display_name="XAU/USD", timestamp="now")
|
||||
@@ -14446,6 +14637,30 @@ def _():
|
||||
assert dict(zip(_ck37.PARAMS_F, _ck37.pack_params(_mk37(Timeframe.M5), _m37_bt(_mk37(Timeframe.M5), dfs))[0].tolist()))["fb2_ema_lo"] == 0.9
|
||||
with patch.dict(os.environ, {"FB2_EMA_LO": "wide"}):
|
||||
assert _ck37.fb2_ema_lo_from_env() == 0.5
|
||||
# grading_mode: the factories pack the shipped pair rule (code 0, no thresholds);
|
||||
# the research modes pack 1 / 2 and the live module's threshold tuple verbatim
|
||||
from signal_aggregator import CONVICTION_THRESHOLDS as _CT
|
||||
conv_names = ("conv_aplus", "conv_a", "conv_b", "conv_c", "conv_strong", "conv_buy", "conv_boost_a", "conv_boost_b")
|
||||
from signal_aggregator import GRADING_MODES as _GM
|
||||
assert _ck37.GRADING_MODE_CODES == {"pair": 0, "min": 1, "product": 2} and set(_GM) == set(_ck37.GRADING_MODE_CODES)
|
||||
for tf in (Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4):
|
||||
pf0, pi0, _ = _ck37.pack_params(_mk37(tf), _m37_bt(_mk37(tf), dfs))
|
||||
f0, i0 = dict(zip(_ck37.PARAMS_F, pf0.tolist())), dict(zip(_ck37.PARAMS_I, pi0.tolist()))
|
||||
assert i0["grading_mode"] == 0 and all(f0[n] == 0.0 for n in conv_names), tf
|
||||
for code, mode in ((1, "min"), (2, "product")):
|
||||
cfg = _mk37(Timeframe.M15)
|
||||
cfg.signals.grading_mode = mode
|
||||
pfm, pim, _ = _ck37.pack_params(cfg, _m37_bt(cfg, dfs))
|
||||
fm, im = dict(zip(_ck37.PARAMS_F, pfm.tolist())), dict(zip(_ck37.PARAMS_I, pim.tolist()))
|
||||
assert im["grading_mode"] == code, mode
|
||||
assert tuple(fm[n] for n in conv_names) == _CT[mode], mode
|
||||
bad = _mk37(Timeframe.M15)
|
||||
bad.signals.grading_mode = "geomean"
|
||||
try:
|
||||
_ck37.pack_params(bad, _m37_bt(bad, dfs))
|
||||
raise AssertionError("the packer must refuse an unknown grading_mode")
|
||||
except ValueError as e:
|
||||
assert "grading_mode" in str(e), e
|
||||
# the kernel #defines are generated from the same name lists
|
||||
src = _ck37._defines()
|
||||
for k, name in enumerate(_ck37.PARAMS_F):
|
||||
@@ -14784,6 +14999,41 @@ def _():
|
||||
print(f" ({t['n']} bars, {t['cands']} candidates, directions {sorted(t['direction'])})")
|
||||
|
||||
|
||||
def _tweak_min37(cfg):
|
||||
cfg.signals.grading_mode = "min"
|
||||
|
||||
|
||||
def _tweak_product37(cfg):
|
||||
cfg.signals.grading_mode = "product"
|
||||
|
||||
|
||||
@test("CUDA stage: K2 == analyze_from_dataframes — M15 factory, grading_mode='min' (research conviction rule, all grades)")
|
||||
def _():
|
||||
if not _GPU37:
|
||||
print(" (no CUDA: kernel not exercised)")
|
||||
return
|
||||
t = _k2_case37("M15 grading_mode=min", Timeframe.M15, _tweak_min37,
|
||||
{"quality_filter": ["A+", "A", "B", "C", "D"]}, "main", 1, None)
|
||||
assert {"A+", "A", "B", "C", "D", "F"} <= t["quality"], t["quality"]
|
||||
assert t["cands"] >= 20, t["cands"]
|
||||
print(f" ({t['n']} bars, {t['cands']} candidates, grades {sorted(t['quality'])})")
|
||||
|
||||
|
||||
@test("CUDA stage: K2 == analyze_from_dataframes — M5 factory, grading_mode='product' (a low-score C reaches the M5 minimum-score gate)")
|
||||
def _():
|
||||
if not _GPU37:
|
||||
print(" (no CUDA: kernel not exercised)")
|
||||
return
|
||||
t = _k2_case37("M5 grading_mode=product", Timeframe.M5, _tweak_product37,
|
||||
{"quality_filter": ["A+", "A", "B", "C", "D"]}, "main", 1, None)
|
||||
# (no product-C survives on this fixture: every low-score C is demoted by gate 1)
|
||||
assert {"A+", "A", "B", "D", "F"} <= t["quality"], t["quality"]
|
||||
# under the product rule a C/B can carry |score| < 25, so gate 1 fires
|
||||
# (unreachable under the shipped pair rule — see the M5 factory case)
|
||||
assert "m5_score" in t["warn"], sorted(t["warn"])
|
||||
print(f" ({t['n']} bars, {t['cands']} candidates, m5_score demotions reachable)")
|
||||
|
||||
|
||||
@test("CUDA stage: K2 == analyze_from_dataframes — H1 and H4 factories (H1 shorts blocked; D1 confirmation shorter than 30 bars)")
|
||||
def _():
|
||||
if not _GPU37:
|
||||
@@ -15031,6 +15281,7 @@ def _():
|
||||
(lambda c: setattr(c.technical, "swing_sr_lookback", _ck37.MAX_SR_LOOKBACK), "swing_sr_lookback"),
|
||||
(lambda c: setattr(c.xau_strategy, "cp_lookback", _ck37.MAX_CP_LOOKBACK), "cp_lookback"),
|
||||
(lambda c: setattr(c.technical, "ema_mode", "3ema"), "ema_mode"),
|
||||
(lambda c: setattr(c.signals, "grading_mode", "geomean"), "grading_mode"),
|
||||
(lambda c: setattr(c, "confirmation_timeframes",
|
||||
[Timeframe.M5, Timeframe.H1, Timeframe.H4, Timeframe.D1, Timeframe.M30]), "confirmation timeframes"),
|
||||
(lambda c: (setattr(c.xau_strategy, "big_candle_atr_multiple", 3.0),
|
||||
|
||||
Reference in New Issue
Block a user