Harwinder Sekhon
272 lines
12 KiB
Python
272 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
panel_payload_drytest_2026-08-15.py — live DeepSeek validation of the
|
|
enriched agent-panel payloads + hardened prompts (2026-08-15).
|
|
|
|
Spends ~6 REAL DeepSeek requests (against the shared 500/day cap) with
|
|
FABRICATED payloads — made-up prices/indicators, no real market data —
|
|
to verify the new technical-persona contract end to end:
|
|
|
|
P1 coherent bullish evidence -> technical persona SUPPORTS
|
|
P2 bearish evidence vs BUY brief -> technical persona OPPOSES
|
|
P3 conflicting evidence -> NEUTRAL or low confidence
|
|
P4 empty evidence -> NEUTRAL with confidence < 30
|
|
(the anti-hallucination contract)
|
|
P5 hallucination trap -> note invents no price levels
|
|
P6 output discipline -> <=20-word notes, exact JSON keys
|
|
P7 synthesizer over 4 supports -> AGREE verdict
|
|
|
|
Usage: venv/bin/python backtest/research/panel_payload_drytest_2026-08-15.py
|
|
Exit code 0 = all probes passed. Raw replies land in
|
|
backtest/research/panel_drytest_2026-08-15.jsonl
|
|
"""
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pandas as pd
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
import agent_panel as ap # noqa: E402
|
|
from app_config import deepseek_api_key, get_app_config # noqa: E402
|
|
from llm_client import get_llm_client # noqa: E402
|
|
|
|
ARTIFACT = Path(__file__).with_suffix(".jsonl")
|
|
|
|
|
|
# ── Fabrication helpers (all numbers are invented) ───────────────────
|
|
|
|
def _staircase(start: float, step: float, n: int) -> pd.DataFrame:
|
|
"""n synthetic bars drifting `step` per bar; last close = start+n*step."""
|
|
rows, price = [], start
|
|
for _ in range(n):
|
|
o, c = price, price + step
|
|
rows.append({"Open": o, "Close": c,
|
|
"High": max(o, c) + abs(step) * 0.4,
|
|
"Low": min(o, c) - abs(step) * 0.4})
|
|
price = c
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
def _choppy(center: float, n: int) -> pd.DataFrame:
|
|
rows, price, sign = [], center, 1
|
|
for _ in range(n):
|
|
step = sign * 1.5
|
|
o, c = price, price + step
|
|
rows.append({"Open": o, "Close": c,
|
|
"High": max(o, c) + 0.6, "Low": min(o, c) - 0.6})
|
|
price, sign = c, -sign
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
_FAKE_PRICE = 2000.00 # deliberately made-up
|
|
_FAKE_ATR = 5.0
|
|
|
|
_BRIEF_BUY = {"pair": "XAU/USD", "timeframe": "15min", "direction": "BUY",
|
|
"quality": "A", "score": 38.5, "confidence": 62,
|
|
"price": _FAKE_PRICE, "market_regime": "trending_up",
|
|
"volatility_regime": "normal", "mtf_agreement": True}
|
|
|
|
_BULL_INDICATORS = {
|
|
"ema_trend": {"signal": "STRONG_BUY", "confidence": 90, "value": 1.0},
|
|
"rsi": {"signal": "BUY", "confidence": 65, "value": 58.0},
|
|
"macd": {"signal": "STRONG_BUY", "confidence": 80, "value": 1.25},
|
|
"bollinger": {"signal": "BUY", "confidence": 60, "value": 0.7},
|
|
"adx": {"signal": "BUY", "confidence": 70, "value": 32.0},
|
|
"roc": {"signal": "BUY", "confidence": 55, "value": 0.35},
|
|
"keltner": {"signal": "BUY", "confidence": 65, "value": 3.1},
|
|
"candlestick": {"signal": "BUY", "confidence": 60, "value": 2.0},
|
|
}
|
|
|
|
_BEAR_INDICATORS = {
|
|
"ema_trend": {"signal": "STRONG_SELL", "confidence": 85, "value": 0.1},
|
|
"rsi": {"signal": "SELL", "confidence": 70, "value": 74.0},
|
|
"macd": {"signal": "STRONG_SELL", "confidence": 80, "value": -1.4},
|
|
"bollinger": {"signal": "SELL", "confidence": 60, "value": 0.95},
|
|
"adx": {"signal": "SELL", "confidence": 70, "value": 35.0},
|
|
"roc": {"signal": "SELL", "confidence": 60, "value": -0.5},
|
|
"keltner": {"signal": "SELL", "confidence": 65, "value": -2.8},
|
|
"candlestick": {"signal": "SELL", "confidence": 55, "value": -2.0},
|
|
}
|
|
|
|
_MIXED_INDICATORS = {
|
|
"ema_trend": {"signal": "BUY", "confidence": 55, "value": 0.6},
|
|
"rsi": {"signal": "NEUTRAL", "confidence": 40, "value": 51.0},
|
|
"macd": {"signal": "SELL", "confidence": 55, "value": -0.3},
|
|
"bollinger": {"signal": "NEUTRAL", "confidence": 35, "value": 0.5},
|
|
"adx": {"signal": "NEUTRAL", "confidence": 40, "value": 18.0},
|
|
"roc": {"signal": "SELL", "confidence": 50, "value": -0.1},
|
|
"keltner": {"signal": "BUY", "confidence": 50, "value": 0.8},
|
|
"candlestick": {"signal": "NEUTRAL", "confidence": 30, "value": 0.0},
|
|
}
|
|
|
|
|
|
def _tech_payload(indicators, mtf, candles_df=None, key_levels=None,
|
|
brief=None, exhaustion=0):
|
|
payload = {"brief": brief or dict(_BRIEF_BUY),
|
|
"indicators": indicators, "mtf": mtf,
|
|
"key_levels": key_levels or {}, "exhaustion": exhaustion}
|
|
if candles_df is not None:
|
|
# the REAL production normalizer, not a hand-rolled copy
|
|
payload["candles"] = {
|
|
"15min": ap._normalize_candles(candles_df, _FAKE_ATR,
|
|
len(candles_df))}
|
|
return payload
|
|
|
|
|
|
# ── Probe runner ─────────────────────────────────────────────────────
|
|
|
|
_RESULTS = []
|
|
|
|
|
|
def _record(probe, payload, reply, take, passed, detail):
|
|
_RESULTS.append({"probe": probe, "pass": bool(passed),
|
|
"detail": detail,
|
|
"stance": getattr(take, "stance", None),
|
|
"confidence": getattr(take, "confidence", None),
|
|
"note": getattr(take, "note", None)})
|
|
with ARTIFACT.open("a") as fh:
|
|
fh.write(json.dumps({"probe": probe, "payload": payload,
|
|
"raw_reply": reply}, default=str) + "\n")
|
|
mark = "PASS" if passed else "FAIL"
|
|
print(f"[{mark}] {probe}: {detail}")
|
|
if take is not None:
|
|
print(f" -> {take.stance} @ {take.confidence:.0f} "
|
|
f"| {take.note}")
|
|
|
|
|
|
def _ask(payload):
|
|
"""Real API call through the production client + production
|
|
validation (the same path run_panel uses)."""
|
|
reply = get_llm_client().chat_json(ap._TECH_SYSTEM, payload)
|
|
return reply, ap._validate_take("technical", reply)
|
|
|
|
|
|
def _want(probe: str) -> bool:
|
|
"""--only P3 runs just that probe (saves requests on re-runs)."""
|
|
if "--only" not in sys.argv:
|
|
return True
|
|
return sys.argv[sys.argv.index("--only") + 1] == probe
|
|
|
|
|
|
def main() -> int:
|
|
app = get_app_config()
|
|
if not deepseek_api_key(app):
|
|
print("ABORT: no DeepSeek API key (env DEEPSEEK_API_KEY / "
|
|
"config.json deepseek.api_key).")
|
|
return 2
|
|
if app.deepseek_settings.get("enabled", True) is False:
|
|
print("ABORT: deepseek.enabled is false in config.json.")
|
|
return 2
|
|
if ARTIFACT.exists() and "--only" not in sys.argv:
|
|
ARTIFACT.unlink() # full run = fresh wire log; --only appends
|
|
|
|
mtf_agree = {"5min": {"signal": "BUY", "score": 30.0},
|
|
"1h": {"signal": "STRONG_BUY", "score": 45.0}}
|
|
levels = {"resistance": 2012.0, "support": 1985.0}
|
|
|
|
# P1 — coherent bullish, with rising ATR staircase (last close 2000)
|
|
if _want("P1"):
|
|
p1 = _tech_payload(_BULL_INDICATORS, mtf_agree,
|
|
candles_df=_staircase(_FAKE_PRICE - 20 * 0.6,
|
|
0.6, 20),
|
|
key_levels=levels)
|
|
reply, take = _ask(p1)
|
|
_record("P1 coherent-bullish", p1, reply, take,
|
|
take is not None and take.stance == "support"
|
|
and take.confidence >= 50,
|
|
"expect support @ conf>=50")
|
|
|
|
# P2 — bearish everything against a BUY brief, falling staircase
|
|
if _want("P2"):
|
|
brief2 = dict(_BRIEF_BUY, market_regime="trending_down",
|
|
mtf_agreement=False)
|
|
mtf_sell = {"5min": {"signal": "SELL", "score": -28.0},
|
|
"1h": {"signal": "STRONG_SELL", "score": -41.0}}
|
|
p2 = _tech_payload(_BEAR_INDICATORS, mtf_sell, brief=brief2,
|
|
candles_df=_staircase(_FAKE_PRICE + 20 * 0.6,
|
|
-0.6, 20),
|
|
key_levels=levels, exhaustion=3)
|
|
reply, take = _ask(p2)
|
|
_record("P2 bearish-vs-BUY-brief", p2, reply, take,
|
|
take is not None and take.stance == "oppose",
|
|
"expect oppose (BUY brief fights all evidence)")
|
|
|
|
# P3 — genuinely mixed evidence: MTF neutral both ways, ranging
|
|
# regime, choppy candles. (An H1 SELL against a BUY is a *listed
|
|
# conflict* in the prompt — oppose is correct there, not mixed.)
|
|
if _want("P3"):
|
|
brief3 = dict(_BRIEF_BUY, market_regime="ranging", score=12.0,
|
|
confidence=35, mtf_agreement=False)
|
|
mtf_mixed = {"5min": {"signal": "NEUTRAL", "score": 4.0},
|
|
"1h": {"signal": "NEUTRAL", "score": -3.0}}
|
|
p3 = _tech_payload(_MIXED_INDICATORS, mtf_mixed, brief=brief3,
|
|
candles_df=_choppy(_FAKE_PRICE, 20))
|
|
reply, take = _ask(p3)
|
|
_record("P3 conflicting", p3, reply, take,
|
|
take is not None and (take.stance == "neutral"
|
|
or take.confidence < 60),
|
|
"expect neutral or conf<60 on genuinely mixed evidence")
|
|
|
|
# P4 — empty evidence: contract demands neutral @ conf<30
|
|
if _want("P4"):
|
|
p4 = _tech_payload({}, {}, key_levels=None)
|
|
reply, take = _ask(p4)
|
|
_record("P4 empty-evidence", p4, reply, take,
|
|
take is not None and take.stance == "neutral"
|
|
and take.confidence < 30,
|
|
"contract: thin/missing evidence -> neutral, conf<30")
|
|
|
|
# P5 — hallucination trap: no key_levels, no candles; the note must
|
|
# not cite any 3+-digit number absent from the payload
|
|
if _want("P5"):
|
|
p5 = _tech_payload(_BULL_INDICATORS, mtf_agree, key_levels=None)
|
|
reply, take = _ask(p5)
|
|
invented = []
|
|
if take is not None:
|
|
payload_blob = json.dumps(p5)
|
|
for num in re.findall(r"\d{3,}", take.note):
|
|
if num not in payload_blob:
|
|
invented.append(num)
|
|
_record("P5 hallucination-trap", p5, reply, take,
|
|
take is not None and not invented,
|
|
f"invented numbers in note: {invented or 'none'}")
|
|
|
|
# P6 — output discipline across the persona replies gathered above
|
|
notes = [r for r in _RESULTS if r["note"]]
|
|
wordy = [r["probe"] for r in notes if len(r["note"].split()) > 20]
|
|
_record("P6 output-discipline", {}, None, None,
|
|
notes and not wordy,
|
|
f"notes >20 words: {wordy or 'none'} "
|
|
f"(cap is 15; 20 = fail threshold)")
|
|
|
|
# P7 — synthesizer over four fabricated SUPPORT takes
|
|
if _want("P7"):
|
|
takes = [{"persona": p, "stance": "support", "confidence": c,
|
|
"note": "evidence aligned"}
|
|
for p, c in (("macro", 70), ("technical", 80),
|
|
("risk", 65), ("contrarian", 60))]
|
|
synth_payload = {"direction": "BUY", "takes": takes}
|
|
reply = get_llm_client().chat_json(ap._SYNTH_SYSTEM, synth_payload)
|
|
verdict = reply.get("verdict") if isinstance(reply, dict) else None
|
|
rationale = (reply or {}).get("rationale", "")
|
|
_record("P7 synthesizer", synth_payload, reply, None,
|
|
verdict == "AGREE" and len(str(rationale).split()) <= 40,
|
|
f"expect AGREE + <=40-word rationale; got {verdict!r} "
|
|
f"({len(str(rationale).split())} words)")
|
|
|
|
n_pass = sum(1 for r in _RESULTS if r["pass"])
|
|
n_api = sum(1 for r in _RESULTS if r["probe"] != "P6 output-discipline")
|
|
print(f"\n{n_pass}/{len(_RESULTS)} probes passed "
|
|
f"({n_api} API requests spent; "
|
|
f"raw wire log: {ARTIFACT.name})")
|
|
return 0 if n_pass == len(_RESULTS) else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|